feat: add SSE endpoint and server mode system

Add server-sent events at GET /api/events for live mutation
notifications. Add server mode (normal/read-only/degraded) exposed
via /health, /ready, and SSE server.state events.

New files:
- broker.go: SSE event hub with client management, non-blocking
  fan-out, ring buffer history for Last-Event-ID replay, heartbeat
- servermode.go: mode state machine with periodic MinIO health
  check and SIGUSR1 read-only toggle
- sse_handler.go: HTTP handler using http.Flusher and
  ResponseController to disable WriteTimeout for long-lived SSE
- broker_test.go, servermode_test.go: 13 unit tests

Modified:
- handlers.go: Server struct gains broker/serverState fields,
  Health/Ready include mode and sse_clients, write handlers
  emit item.created/updated/deleted and revision.created events
- routes.go: register GET /api/events, add RequireWritable
  middleware to all 8 editor-gated route groups
- middleware.go: RequireWritable returns 503 in read-only mode
- csv.go, ods.go: emit bulk item.created events after import
- storage.go: add Ping() method for health checks
- config.go: add ReadOnly field to ServerConfig
- main.go: create broker/state, start background goroutines,
  SIGUSR1 handler, graceful shutdown sequence

Closes #38, closes #39
This commit is contained in:
Forbes
2026-02-08 15:59:23 -06:00
parent 21227b7586
commit 3d7302f383
15 changed files with 747 additions and 14 deletions

View File

@@ -178,9 +178,19 @@ func main() {
}
}
// Create SSE broker and server state
broker := api.NewBroker(logger)
serverState := api.NewServerState(logger, store, broker)
if cfg.Server.ReadOnly {
serverState.SetReadOnly(true)
logger.Warn().Msg("server started in read-only mode")
}
broker.StartHeartbeat()
serverState.StartStorageHealthCheck()
// Create API server
server := api.NewServer(logger, database, schemas, cfg.Schemas.Directory, store,
authService, sessionManager, oidcBackend, &cfg.Auth)
authService, sessionManager, oidcBackend, &cfg.Auth, broker, serverState)
router := api.NewRouter(server, logger)
// Create HTTP server
@@ -201,6 +211,16 @@ func main() {
}
}()
// SIGUSR1: toggle read-only mode
usr1 := make(chan os.Signal, 1)
signal.Notify(usr1, syscall.SIGUSR1)
go func() {
for range usr1 {
serverState.ToggleReadOnly()
logger.Info().Str("mode", string(serverState.Mode())).Msg("read-only mode toggled via SIGUSR1")
}
}()
// Wait for interrupt signal
quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
@@ -208,7 +228,10 @@ func main() {
logger.Info().Msg("shutting down server")
// Graceful shutdown with timeout
// Graceful shutdown: close SSE connections first, then HTTP server
broker.Shutdown()
serverState.Shutdown()
shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()