refactor: separate the application service from the Bubble Tea TUI

internal/app/events.go

@@ -0,0 +1,166 @@
+package app
+
+import (
+	"sync"
+	"sync/atomic"
+
+	"github.com/evg4b/uncors/internal/config"
+	"github.com/evg4b/uncors/internal/server"
+)
+
+const eventsBufferSize = 1000
+
+// Event is something the service reports to whoever is presenting it. The set
+// is deliberately small: it covers what the application already communicated,
+// and nothing more.
+type Event interface {
+	isEvent()
+}
+
+// LifecycleState is the state of the server as the service understands it.
+type LifecycleState int
+
+const (
+	StateStarting LifecycleState = iota
+	StateStarted
+	StateStartFailed
+	StateReloading
+	StateReloaded
+	StateReloadFailed
+	StateStopping
+	StateStopped
+)
+
+// LifecycleEvent reports a change of server state. Mappings carries the
+// configuration of the generation the state refers to, so a presenter never
+// has to reach back into the service to describe what is serving.
+type LifecycleEvent struct {
+	State    LifecycleState
+	Mappings config.Mappings
+	Err      error
+
+	// Interrupted marks a stop caused by SIGINT, where the terminal has already
+	// echoed "^C" and the presenter may need to move past it.
+	Interrupted bool
+}
+
+func (LifecycleEvent) isEvent() {}
+
+// Level classifies a LogEvent for presentation. It carries no rendering.
+type Level int
+
+const (
+	LevelInfo Level = iota
+	LevelWarn
+	LevelError
+)
+
+// LogEvent is a message from the service addressed to the user.
+type LogEvent struct {
+	Level   Level
+	Prefix  string
+	Message string
+}
+
+func (LogEvent) isEvent() {}
+
+// RequestEvent forwards one server request lifecycle event to the presenter.
+// The service is the single consumer of the request tracker, because it keeps
+// the authoritative set of in-flight requests; clients render what it passes
+// on rather than reading the tracker themselves.
+type RequestEvent struct {
+	Event server.RequestEvent
+}
+
+func (RequestEvent) isEvent() {}
+
+// Status is the latest lifecycle state, always readable regardless of whether
+// the notification for it was delivered.
+type Status struct {
+	State    LifecycleState
+	Mappings config.Mappings
+	Err      error
+}
+
+// emitter fans service events out to the single presenting client.
+//
+// Log events are dropped when the client cannot keep up, and counted, exactly
+// as request activity is: presentation must never be able to stall the
+// service. Lifecycle notifications are dropped under the same pressure, which
+// is safe only because the latest state is also recorded in status - a client
+// that misses a notification can still read the truth.
+type emitter struct {
+	events chan Event
+
+	mu     sync.RWMutex
+	status Status
+
+	// sendMu guards closed together with the send itself, so Close can never
+	// race a send onto an already-closed channel.
+	sendMu  sync.RWMutex
+	closed  bool
+	dropped atomic.Uint64
+}
+
+func newEmitter() *emitter {
+	return &emitter{events: make(chan Event, eventsBufferSize)}
+}
+
+func (e *emitter) Events() <-chan Event {
+	return e.events
+}
+
+func (e *emitter) Status() Status {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
+
+	return e.status
+}
+
+func (e *emitter) Dropped() uint64 {
+	return e.dropped.Load()
+}
+
+func (e *emitter) EmitLifecycle(event LifecycleEvent) {
+	e.mu.Lock()
+	e.status = Status{State: event.State, Mappings: event.Mappings, Err: event.Err}
+	e.mu.Unlock()
+
+	e.send(event)
+}
+
+func (e *emitter) EmitLog(level Level, message string) {
+	e.send(LogEvent{Level: level, Message: message})
+}
+
+// Close stops delivery. It is safe to call concurrently with an emit and safe
+// to call more than once.
+func (e *emitter) Close() {
+	e.sendMu.Lock()
+	defer e.sendMu.Unlock()
+
+	if e.closed {
+		return
+	}
+
+	e.closed = true
+
+	close(e.events)
+}
+
+func (e *emitter) send(event Event) {
+	e.sendMu.RLock()
+	defer e.sendMu.RUnlock()
+
+	if e.closed {
+		e.dropped.Add(1)
+
+		return
+	}
+
+	select {
+	case e.events <- event:
+	default:
+		e.dropped.Add(1)
+	}
+}

internal/app/events_test.go

@@ -0,0 +1,150 @@
+package app_test
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/evg4b/uncors/internal/app"
+	"github.com/evg4b/uncors/internal/config"
+	"github.com/evg4b/uncors/internal/di"
+	"github.com/evg4b/uncors/testing/testutils"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func drain(t *testing.T, service *app.Service) []app.Event {
+	t.Helper()
+
+	events := make([]app.Event, 0)
+
+	for {
+		select {
+		case event := <-service.Events():
+			events = append(events, event)
+		default:
+			return events
+		}
+	}
+}
+
+func statesOf(events []app.Event) []app.LifecycleState {
+	states := make([]app.LifecycleState, 0, len(events))
+
+	for _, event := range events {
+		if lifecycle, ok := event.(app.LifecycleEvent); ok {
+			states = append(states, lifecycle.State)
+		}
+	}
+
+	return states
+}
+
+func TestServiceEmitsLifecycle(t *testing.T) {
+	t.Run("a successful start reports starting then started with its mappings", func(t *testing.T) {
+		port := testutils.GetFreePort(t)
+		cfg := configFor(port)
+
+		service := newService(t, cfg, "", func() (*config.UncorsConfig, error) { return cfg, nil })
+
+		require.NoError(t, service.Start(t.Context()))
+
+		events := drain(t, service)
+		assert.Equal(t, []app.LifecycleState{app.StateStarting, app.StateStarted}, statesOf(events))
+
+		started, ok := events[1].(app.LifecycleEvent)
+		require.True(t, ok)
+		assert.Equal(t, cfg.Mappings, started.Mappings,
+			"the event must describe the generation without the client asking the service")
+	})
+
+	t.Run("a failed start reports the error", func(t *testing.T) {
+		port := testutils.GetFreePort(t)
+
+		occupy(t, port)
+
+		cfg := configFor(port)
+		service := newService(t, cfg, "", func() (*config.UncorsConfig, error) { return cfg, nil })
+
+		require.Error(t, service.Start(t.Context()))
+		assert.Equal(t, app.StateStartFailed, service.Status().State)
+		require.Error(t, service.Status().Err)
+	})
+
+	t.Run("a failed reload reports the error and keeps the old mappings", func(t *testing.T) {
+		port := testutils.GetFreePort(t)
+		cfg := configFor(port)
+
+		service := newService(t, cfg, "", func() (*config.UncorsConfig, error) { return nil, errLoadFailed })
+
+		require.NoError(t, service.Start(t.Context()))
+
+		service.Reload()
+
+		assert.Equal(t, app.StateReloadFailed, service.Status().State)
+		require.ErrorIs(t, service.Status().Err, errLoadFailed)
+		assert.Same(t, cfg, service.Config())
+	})
+}
+
+// Status must stay truthful even when nobody is draining the stream, because
+// that is the whole reason lifecycle is state rather than only a notification.
+func TestStatusSurvivesAnUndrainedStream(t *testing.T) {
+	port := testutils.GetFreePort(t)
+	cfg := configFor(port)
+
+	service := newService(t, cfg, "", func() (*config.UncorsConfig, error) { return cfg, nil })
+
+	require.NoError(t, service.Start(t.Context()))
+
+	for range 3000 {
+		service.Reload()
+	}
+
+	assert.Positive(t, service.DroppedEvents(), "an undrained stream must drop and count, not block")
+	assert.Equal(t, app.StateReloaded, service.Status().State)
+}
+
+// Emitting must never block the caller, whatever the consumer is doing.
+func TestEmittingNeverBlocks(t *testing.T) {
+	port := testutils.GetFreePort(t)
+	cfg := configFor(port)
+
+	service := newService(t, cfg, "", func() (*config.UncorsConfig, error) { return cfg, nil })
+
+	require.NoError(t, service.Start(t.Context()))
+
+	var waitGroup sync.WaitGroup
+
+	waitGroup.Go(func() {
+		for range 500 {
+			service.Reload()
+		}
+	})
+
+	done := make(chan struct{})
+
+	go func() {
+		waitGroup.Wait()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-t.Context().Done():
+		t.Fatal("emitting blocked")
+	}
+}
+
+func TestClosedServiceStopsDelivery(t *testing.T) {
+	cfg := configFor(testutils.GetFreePort(t))
+
+	container := di.NewContainer()
+	service := app.New(container, cfg, "", func() (*config.UncorsConfig, error) { return cfg, nil })
+
+	require.NoError(t, service.Close())
+
+	_, open := <-service.Events()
+	assert.False(t, open, "Close must close the event stream so consumers terminate")
+
+	require.NoError(t, container.Close())
+}