WIP.

internal/api/api.go

@@ -61,5 +61,6 @@ func (manager *ApiManagerCtx) Mount(r chi.Router) {
 		r.Post("/recreate", manager.roomRecreate)
 	})
 
+	r.Get("/events", manager.events)
 	r.Get("/docker-compose.yaml", manager.dockerCompose)
 }

internal/api/rooms.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"io"
 	"net/http"
 	"strconv"
@@ -250,6 +251,38 @@ func (manager *ApiManagerCtx) roomGenericAction(action func(ctx context.Context,
 	}
 }
 
+func (manager *ApiManagerCtx) events(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+
+	flusher, ok := w.(http.Flusher)
+	if !ok {
+		http.Error(w, "Connection does not support streaming", http.StatusBadRequest)
+		return
+	}
+
+	events, errs := manager.rooms.Events(r.Context())
+	for {
+		select {
+		case _, ok := <-errs:
+			if !ok {
+				manager.logger.Debug().Msg("sse channel closed")
+			}
+			return
+		case e := <-events:
+			jsonData, err := json.Marshal(e)
+			if err != nil {
+				http.Error(w, err.Error(), 500)
+				return
+			}
+
+			fmt.Fprintf(w, "data: %s\n\n", jsonData)
+			flusher.Flush()
+		}
+	}
+}
+
 func (manager *ApiManagerCtx) dockerCompose(w http.ResponseWriter, r *http.Request) {
 	response, err := manager.rooms.ExportAsDockerCompose(r.Context())
 	if err != nil {

internal/room/events.go

@@ -0,0 +1,112 @@
+package room
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/m1k1o/neko-rooms/internal/types"
+
+	dockerTypes "github.com/docker/docker/api/types"
+	"github.com/docker/docker/api/types/filters"
+)
+
+func (manager *RoomManagerCtx) EventsLoopStart() {
+	ctx, cancel := context.WithCancel(context.Background())
+	manager.eventsLoopCtx = ctx
+	manager.eventsLoopCancel = cancel
+
+	msgs, errs := manager.client.Events(ctx, dockerTypes.EventsOptions{
+		Filters: filters.NewArgs(
+			filters.Arg("type", "container"),
+			filters.Arg("label", fmt.Sprintf("m1k1o.neko_rooms.instance=%s", manager.config.InstanceName)),
+			filters.Arg("event", "create"),
+			filters.Arg("event", "start"),
+			//filters.Arg("event", "health_status"),
+			filters.Arg("event", "stop"),
+			filters.Arg("event", "destroy"),
+		),
+	})
+
+	manager.eventsWg.Add(1)
+	go func() {
+		defer manager.eventsWg.Done()
+
+		for {
+			select {
+			case err, ok := <-errs:
+				if !ok {
+					manager.logger.Fatal().Msg("docker event error channel closed")
+					return
+				}
+
+				manager.logger.Err(err).Msg("got docker event error")
+				return
+			case msg, ok := <-msgs:
+				if !ok {
+					manager.logger.Fatal().Msg("docker event message channel closed")
+					return
+				}
+
+				e := types.RoomEvent{
+					ID:     msg.Actor.ID[:12],
+					Action: msg.Action,
+					Labels: msg.Actor.Attributes,
+				}
+
+				manager.logger.Info().
+					Str("id", e.ID).
+					Str("action", e.Action).
+					Msg("got docker event")
+
+				// broadcast event
+				manager.eventsMu.Lock()
+				for _, listener := range manager.eventsListeners {
+					listener <- e
+				}
+				manager.eventsMu.Unlock()
+			}
+		}
+	}()
+}
+
+func (manager *RoomManagerCtx) EventsLoopStop() error {
+	manager.eventsLoopCancel()
+	manager.eventsWg.Wait()
+	return nil
+}
+
+func (manager *RoomManagerCtx) Events(ctx context.Context) (<-chan types.RoomEvent, <-chan error) {
+	messages := make(chan types.RoomEvent)
+	errs := make(chan error, 1)
+
+	// add listener
+	manager.eventsMu.Lock()
+	manager.eventsListeners = append(manager.eventsListeners, messages)
+	manager.eventsMu.Unlock()
+
+	manager.eventsWg.Add(1)
+	go func() {
+		defer manager.eventsWg.Done()
+		defer close(errs)
+
+		select {
+		case <-ctx.Done():
+			errs <- ctx.Err()
+		case <-manager.eventsLoopCtx.Done():
+			errs <- fmt.Errorf("room manager shutdown")
+			return
+		}
+
+		// remove listener
+		manager.eventsMu.Lock()
+		for i, listener := range manager.eventsListeners {
+			if listener == messages {
+				manager.eventsListeners = append(manager.eventsListeners[:i], manager.eventsListeners[i+1:]...)
+				break
+			}
+		}
+		manager.eventsMu.Unlock()
+	}()
+
+	return messages, errs
+}

internal/room/manager.go

@@ -12,6 +12,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/docker/cli/opts"
@@ -55,6 +56,12 @@ type RoomManagerCtx struct {
 	logger zerolog.Logger
 	config *config.Room
 	client *dockerClient.Client
+
+	eventsMu         sync.Mutex
+	eventsWg         sync.WaitGroup
+	eventsListeners  []chan types.RoomEvent
+	eventsLoopCtx    context.Context
+	eventsLoopCancel context.CancelFunc
 }
 
 func (manager *RoomManagerCtx) Config() types.RoomsConfig {

internal/types/room.go

@@ -138,6 +138,12 @@ type RoomMember struct {
 	Muted bool   `json:"muted"`
 }
 
+type RoomEvent struct {
+	ID     string            `json:"id"`
+	Action string            `json:"action"`
+	Labels map[string]string `json:"-"`
+}
+
 var ErrRoomNotFound = fmt.Errorf("room not found")
 
 type RoomManager interface {
@@ -155,4 +161,8 @@ type RoomManager interface {
 	Start(ctx context.Context, id string) error
 	Stop(ctx context.Context, id string) error
 	Restart(ctx context.Context, id string) error
+
+	EventsLoopStart()
+	EventsLoopStop() error
+	Events(ctx context.Context) (<-chan RoomEvent, <-chan error)
 }

neko.go

@@ -131,6 +131,7 @@ func (main *MainCtx) Start() {
 		client,
 		main.Configs.Room,
 	)
+	main.roomManager.EventsLoopStart()
 
 	main.pullManager = pull.New(
 		client,
@@ -169,6 +170,9 @@ func (main *MainCtx) Shutdown() {
 
 	err = main.pullManager.Shutdown()
 	main.logger.Err(err).Msg("pull manager shutdown")
+
+	err = main.roomManager.EventsLoopStop()
+	main.logger.Err(err).Msg("room events loop shutdown")
 }
 
 func (main *MainCtx) ServeCommand(cmd *cobra.Command, args []string) {