From 798ea43c2f086011cc1c522b5aeaa05e6664f48c Mon Sep 17 00:00:00 2001 From: Rogee Date: Sat, 22 Aug 2026 02:38:15 +0800 Subject: [PATCH] HH-442: isolate runtime processes and harden shutdown (#90) * HH-442: isolate runtime processes and harden shutdown * HH-442: harden worker shutdown races * HH-442: gate dependency shutdown on active handlers --------- Co-authored-by: Rogee --- .github/workflows/ci.yml | 4 +- backend/cmd/gochat/coverage_test.go | 14 ++ backend/cmd/gochat/main.go | 29 ++- backend/configs/config.yaml | 6 + backend/configs/prometheus_alerts.yml | 2 +- backend/configs/prometheus_alerts_test.yml | 27 ++ backend/internal/app/app.go | 138 ++++++++-- backend/internal/app/bootstrap.go | 21 +- backend/internal/app/readiness.go | 57 +++++ backend/internal/app/readiness_test.go | 50 ++++ backend/internal/app/runtime_test.go | 237 ++++++++++++++++++ backend/internal/app/shutdown.go | 61 +++-- backend/internal/config/config.go | 94 ++++--- backend/internal/database/migrate.go | 52 ++++ backend/internal/database/migrate_test.go | 24 ++ backend/internal/handler/handler_test.go | 62 +++++ backend/internal/handler/health_handler.go | 128 ++++++---- backend/internal/handler/metrics_handler.go | 186 +++++++++----- backend/internal/lifecycle/handler_group.go | 63 +++++ backend/internal/middleware/coverage7_test.go | 5 + backend/internal/middleware/rate_limit.go | 2 +- backend/internal/router/coverage_test.go | 16 -- backend/internal/router/router.go | 20 -- .../notification_delivery_lifecycle_test.go | 93 +++++++ .../service/notification_delivery_service.go | 85 ++++--- backend/internal/worker/worker.go | 158 ++++++++++-- backend/internal/worker/worker_test.go | 122 +++++++++ backend/scripts/health_check.sh | 228 ++++++----------- backend/scripts/health_check_test.go | 76 ++++++ deploy/docker/Dockerfile | 2 +- deploy/docker/Dockerfile.dev | 3 +- deploy/docker/docker-compose.dev.yml | 4 +- deploy/docker/docker-compose.prod.yml | 8 +- deploy/docker/docker-compose.yml | 22 ++ deploy/quickstart/compose.yaml | 25 +- 35 files changed, 1673 insertions(+), 451 deletions(-) create mode 100644 backend/configs/prometheus_alerts_test.yml create mode 100644 backend/internal/app/readiness.go create mode 100644 backend/internal/app/readiness_test.go create mode 100644 backend/internal/app/runtime_test.go create mode 100644 backend/internal/lifecycle/handler_group.go delete mode 100644 backend/internal/router/coverage_test.go create mode 100644 backend/internal/service/notification_delivery_lifecycle_test.go mode change 100644 => 100755 backend/scripts/health_check.sh create mode 100644 backend/scripts/health_check_test.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 222851e2..25ef9220 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -31,6 +31,8 @@ jobs: id: chrome - name: Install browser harness run: python -m pip install browser-harness==0.1.9 + - name: Test Prometheus alert rules + run: docker run --rm --entrypoint promtool -v "$PWD/backend/configs:/configs:ro" prom/prometheus:v3.5.0@sha256:63805ebb8d2b3920190daf1cb14a60871b16fd38bed42b857a3182bc621f4996 test rules /configs/prometheus_alerts_test.yml - name: Test working-directory: backend env: @@ -207,7 +209,7 @@ jobs: docker compose -f deploy/docker/docker-compose.prod.yml config --format json | python3 -c 'import json, os, sys; config = json.load(sys.stdin); assert all(config["services"][service]["environment"]["GOCHAT_JWT_PREVIOUS_SECRETS"] == os.environ["GOCHAT_JWT_PREVIOUS_SECRETS"] for service in ("gochat", "worker"))' docker compose -f deploy/docker/docker-compose.prod.yml --profile ops run --rm migrate docker compose -f deploy/docker/docker-compose.prod.yml up -d --wait gochat - curl -fsS "http://127.0.0.1:$GOCHAT_PORT/health" | grep -q '"status":"ok"' + curl -fsS "http://127.0.0.1:$GOCHAT_PORT/health" | grep -q '"status":"healthy"' curl -fsS "http://127.0.0.1:$GOCHAT_PORT/app" | grep -q '/assets/' - name: Stop production Compose if: always() diff --git a/backend/cmd/gochat/coverage_test.go b/backend/cmd/gochat/coverage_test.go index 46bdc033..f8a3a237 100644 --- a/backend/cmd/gochat/coverage_test.go +++ b/backend/cmd/gochat/coverage_test.go @@ -4,12 +4,26 @@ import ( "testing" "github.com/gochat/gochat/internal/config" + "github.com/stretchr/testify/require" ) func TestPrintUsage_Cov1(t *testing.T) { printUsage() } +func TestParseServeArgs(t *testing.T) { + workerOnly, err := parseServeArgs(nil) + require.NoError(t, err) + require.False(t, workerOnly) + + workerOnly, err = parseServeArgs([]string{"--worker-only"}) + require.NoError(t, err) + require.True(t, workerOnly) + + _, err = parseServeArgs([]string{"--unknown"}) + require.EqualError(t, err, `unknown serve option "--unknown"`) +} + func TestShouldRunSeedMigrations_True_Cov1(t *testing.T) { cfg := &config.Config{Database: config.DatabaseConfig{RunMigrations: true}} result := shouldRunSeedMigrations(cfg) diff --git a/backend/cmd/gochat/main.go b/backend/cmd/gochat/main.go index f18809b1..ec195348 100644 --- a/backend/cmd/gochat/main.go +++ b/backend/cmd/gochat/main.go @@ -35,7 +35,9 @@ func main() { var err error switch cmd { case "serve", "server", "run": - err = serve() + err = serve(os.Args[2:]) + case "worker": + err = serve([]string{"--worker-only"}) case "seed": err = seed() case "init": @@ -53,13 +55,18 @@ func main() { } func printUsage() { - fmt.Println("Usage: gochat [serve|seed|init]") - fmt.Println(" serve Start the GoChat HTTP server") + fmt.Println("Usage: gochat [serve [--worker-only]|worker|seed|init]") + fmt.Println(" serve Start the GoChat HTTP server") + fmt.Println(" worker Start background workers without HTTP") fmt.Println(" seed Create deterministic development/smoke data") fmt.Println(" init Initialize super admin account (interactive or via flags)") } -func serve() error { +func serve(args []string) error { + workerOnly, err := parseServeArgs(args) + if err != nil { + return err + } env := os.Getenv("GOCHAT_ENV") if env == "" { env = "development" @@ -68,9 +75,23 @@ func serve() error { if err != nil { return err } + if workerOnly { + return application.RunWorker() + } return application.Run() } +func parseServeArgs(args []string) (bool, error) { + workerOnly := false + for _, arg := range args { + if arg != "--worker-only" { + return false, fmt.Errorf("unknown serve option %q", arg) + } + workerOnly = true + } + return workerOnly, nil +} + func seed() error { env := os.Getenv("GOCHAT_ENV") if env == "" { diff --git a/backend/configs/config.yaml b/backend/configs/config.yaml index 9d49d58b..f94e2912 100644 --- a/backend/configs/config.yaml +++ b/backend/configs/config.yaml @@ -2,6 +2,12 @@ server: host: "0.0.0.0" port: 3000 mode: "debug" # debug, release, test + read_header_timeout_seconds: 5 + read_timeout_seconds: 30 + write_timeout_seconds: 30 + idle_timeout_seconds: 120 + shutdown_timeout_seconds: 30 + max_header_bytes: 1048576 cors: allowed_origins: [] # empty = Allow-Origin:* in debug mode; production must list exact origins # Examples: diff --git a/backend/configs/prometheus_alerts.yml b/backend/configs/prometheus_alerts.yml index 5ca3fe6b..fef7c9ed 100644 --- a/backend/configs/prometheus_alerts.yml +++ b/backend/configs/prometheus_alerts.yml @@ -17,7 +17,7 @@ groups: # High error rate - alert: GoChatHighErrorRate - expr: rate(http_requests_total{job="gochat", status=~"5.."}[5m]) / rate(http_requests_total{job="gochat"}[5m]) > 0.05 + expr: sum by (job, instance) (rate(http_requests_total{job="gochat", status=~"5.."}[5m])) / sum by (job, instance) (rate(http_requests_total{job="gochat"}[5m])) > 0.05 for: 5m labels: severity: warning diff --git a/backend/configs/prometheus_alerts_test.yml b/backend/configs/prometheus_alerts_test.yml new file mode 100644 index 00000000..420c9561 --- /dev/null +++ b/backend/configs/prometheus_alerts_test.yml @@ -0,0 +1,27 @@ +rule_files: + - prometheus_alerts.yml + +evaluation_interval: 1m + +tests: + - interval: 1m + input_series: + - series: 'http_requests_total{job="gochat",instance="below-threshold",method="GET",route="/ok",status="200"}' + values: '0+96x12' + - series: 'http_requests_total{job="gochat",instance="below-threshold",method="GET",route="/error",status="500"}' + values: '0+4x12' + - series: 'http_requests_total{job="gochat",instance="above-threshold",method="GET",route="/ok",status="200"}' + values: '0+94x12' + - series: 'http_requests_total{job="gochat",instance="above-threshold",method="GET",route="/error",status="500"}' + values: '0+6x12' + alert_rule_test: + - eval_time: 10m + alertname: GoChatHighErrorRate + exp_alerts: + - exp_labels: + instance: above-threshold + job: gochat + severity: warning + exp_annotations: + summary: GoChat error rate above 5% + description: Error rate is 6% over the last 5 minutes. diff --git a/backend/internal/app/app.go b/backend/internal/app/app.go index 6dc3ac82..b6d431ee 100644 --- a/backend/internal/app/app.go +++ b/backend/internal/app/app.go @@ -2,9 +2,14 @@ package app import ( "context" + "errors" "fmt" "log" "net/http" + "os/signal" + "sync" + "sync/atomic" + "syscall" "time" "github.com/gin-gonic/gin" @@ -12,6 +17,7 @@ import ( "github.com/gochat/gochat/internal/canned" "github.com/gochat/gochat/internal/config" ws "github.com/gochat/gochat/internal/handler/ws" + "github.com/gochat/gochat/internal/lifecycle" "github.com/gochat/gochat/internal/model" channelmodel "github.com/gochat/gochat/internal/model/channel" "github.com/gochat/gochat/internal/pubsub" @@ -35,6 +41,10 @@ type App struct { wsRelay *wspkg.BroadcastRelay notificationDeliverySvc *service.NotificationDeliveryService workerPool *worker.WorkerPool + ready *atomic.Bool + notificationRunning atomic.Bool + handlerGroupOnce sync.Once + handlerGroup *lifecycle.HandlerGroup } // New creates and initializes the application. @@ -82,38 +92,115 @@ func New(cfg *config.Config) (*App, error) { return application, nil } -// Run starts the notification delivery pipeline and the HTTP server. +// Run starts only the web process. Background jobs are consumed by RunWorker. func (a *App) Run() error { - ctx, cancel := context.WithCancel(context.Background()) + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + return a.runWeb(ctx) +} + +func (a *App) runWeb(ctx context.Context) error { + lifecycle, cancel := context.WithCancel(context.Background()) defer cancel() + if a.wsHub != nil { + go a.wsHub.Run(lifecycle) + } if a.wsRelay != nil { - if err := a.wsRelay.Start(ctx); err != nil { + if err := a.wsRelay.Start(lifecycle); err != nil { return fmt.Errorf("failed to start WebSocket relay: %w", err) } - defer a.wsRelay.Stop() } + + server := a.HTTPServer() + serveErr := make(chan error, 1) + go func() { + log.Printf("Starting gochat web server on %s", server.Addr) + serveErr <- server.ListenAndServe() + }() + + select { + case err := <-serveErr: + cancel() + _ = a.Shutdown(a.shutdownTimeout()) + if errors.Is(err, http.ErrServerClosed) { + return nil + } + return err + case <-ctx.Done(): + } + + if a.ready != nil { + a.ready.Store(false) + } + timeout := a.shutdownTimeout() + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), timeout) + defer shutdownCancel() + httpErr := server.Shutdown(shutdownCtx) + cancel() + return errors.Join(httpErr, a.shutdown(shutdownCtx)) +} + +// RunWorker consumes background jobs and notifications without opening an HTTP listener. +func (a *App) RunWorker() error { + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + return a.runWorker(ctx) +} + +func (a *App) runWorker(ctx context.Context) error { if a.workerPool != nil { if err := a.workerPool.Start(); err != nil { return fmt.Errorf("failed to start background worker: %w", err) } applogger.L().Info("Background worker started") } - - // Start notification delivery service (Watermill router) in background + notificationErr := make(chan error, 1) if a.notificationDeliverySvc != nil { - go func() { - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) - defer cancel() - if err := a.notificationDeliverySvc.Start(ctx); err != nil { - applogger.L().Errorf("Notification delivery service start error: %v", err) - } - }() + a.notificationRunning.Store(true) + go func() { notificationErr <- a.notificationDeliverySvc.Start(a.handlers().Context()) }() applogger.L().Info("Notification delivery pipeline started") } - addr := fmt.Sprintf("%s:%d", a.config.Server.Host, a.config.Server.Port) - log.Printf("Starting gochat server on %s", addr) - return a.engine.Run(addr) + select { + case <-ctx.Done(): + case err := <-notificationErr: + if err != nil { + return errors.Join(err, a.Shutdown(a.shutdownTimeout())) + } + } + return a.Shutdown(a.shutdownTimeout()) +} + +// HTTPServer applies the production request-boundary limits from configuration. +func (a *App) HTTPServer() *http.Server { + cfg := a.config.Server + return &http.Server{ + Addr: a.Address(), + Handler: a.Handler(), + ReadHeaderTimeout: positiveDuration(cfg.ReadHeaderTimeoutS, 5*time.Second), + ReadTimeout: positiveDuration(cfg.ReadTimeoutS, 30*time.Second), + WriteTimeout: positiveDuration(cfg.WriteTimeoutS, 30*time.Second), + IdleTimeout: positiveDuration(cfg.IdleTimeoutS, 120*time.Second), + MaxHeaderBytes: positiveInt(cfg.MaxHeaderBytes, 1<<20), + } +} + +func (a *App) shutdownTimeout() time.Duration { + return positiveDuration(a.config.Server.ShutdownTimeoutS, 30*time.Second) +} + +func positiveDuration(seconds int, fallback time.Duration) time.Duration { + if seconds <= 0 { + return fallback + } + return time.Duration(seconds) * time.Second +} + +func positiveInt(value, fallback int) int { + if value <= 0 { + return fallback + } + return value } // Address returns the server listen address. @@ -123,7 +210,24 @@ func (a *App) Address() string { // Handler returns the gin engine as http.Handler for use with http.Server. func (a *App) Handler() http.Handler { - return a.engine + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + handlers := a.handlers() + if !handlers.Begin() { + http.Error(w, http.StatusText(http.StatusServiceUnavailable), http.StatusServiceUnavailable) + return + } + defer handlers.Done() + a.engine.ServeHTTP(w, r) + }) +} + +func (a *App) handlers() *lifecycle.HandlerGroup { + a.handlerGroupOnce.Do(func() { + if a.handlerGroup == nil { + a.handlerGroup = lifecycle.NewHandlerGroup() + } + }) + return a.handlerGroup } // Config returns the application config. diff --git a/backend/internal/app/bootstrap.go b/backend/internal/app/bootstrap.go index 807742c2..f090223d 100644 --- a/backend/internal/app/bootstrap.go +++ b/backend/internal/app/bootstrap.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "strings" + "sync/atomic" "time" "github.com/gin-gonic/gin" @@ -28,6 +29,7 @@ import ( whatsappchannel "github.com/gochat/gochat/internal/channel/whatsapp" "github.com/gochat/gochat/internal/config" "github.com/gochat/gochat/internal/database" + basehandler "github.com/gochat/gochat/internal/handler" v1 "github.com/gochat/gochat/internal/handler/api/v1" webhook "github.com/gochat/gochat/internal/handler/webhook" widget "github.com/gochat/gochat/internal/handler/widget" @@ -944,10 +946,16 @@ func Bootstrap(env string) (*App, error) { // (ref: Chatwoot Rails middleware stack in config/application.rb) gin.SetMode(cfg.Server.Mode) engine := gin.New() + startedAt := time.Now() + httpMetrics := basehandler.NewHTTPMetrics(startedAt) + ready := &atomic.Bool{} + ready.Store(true) + readinessChecks := newReadinessChecks(cfg, db, rdb, ready) // Global middleware — applies to ALL routes corsMiddleware := middleware.CORS(middleware.CORSConfigFromAppConfig(cfg)) - engine.Use(middleware.Recovery()) // panic recovery + engine.Use(middleware.Recovery()) // panic recovery + engine.Use(httpMetrics.Middleware()) engine.Use(middleware.RequestLogger()) // structured request logging engine.Use(middleware.RateLimit(rdb)) // rate limiting (ref: Chatwoot rack-attack) engine.Use(corsMiddleware) // CORS with configurable whitelist @@ -967,6 +975,10 @@ func Bootstrap(env string) (*App, error) { sessionMwCfg := middleware.SessionMiddlewareConfigFromAppConfig(cfg) engine.Use(middleware.SessionMiddleware(sessionStore, sessionMwCfg)) } + engine.GET("/health", basehandler.HealthHandler(db, startedAt, config.Version, readinessChecks...)) + engine.GET("/ready", basehandler.ReadyHandler(db, readinessChecks...)) + engine.GET("/live", basehandler.LiveHandler()) + engine.GET("/metrics", httpMetrics.Handler()) // Step 11: WebSocket hub already created above (before handler wiring) // so handlers can reference it via the hubTypingAdapter. @@ -994,7 +1006,7 @@ func Bootstrap(env string) (*App, error) { applogger.L().Info("All dependencies wired successfully") - return &App{ + application := &App{ config: cfg, reloader: reloader, db: db, @@ -1004,7 +1016,10 @@ func Bootstrap(env string) (*App, error) { wsRelay: wsRelay, notificationDeliverySvc: notificationDeliverySvc, workerPool: workerPool, - }, nil + ready: ready, + } + notificationDeliverySvc.SetHandlerGroup(application.handlers()) + return application, nil } func validateStartupMigrations(env string, cfg *config.Config) error { diff --git a/backend/internal/app/readiness.go b/backend/internal/app/readiness.go new file mode 100644 index 00000000..b43dce9a --- /dev/null +++ b/backend/internal/app/readiness.go @@ -0,0 +1,57 @@ +package app + +import ( + "context" + "fmt" + "io" + "net/http" + "strings" + "sync/atomic" + "time" + + "github.com/gochat/gochat/internal/config" + "github.com/gochat/gochat/internal/database" + basehandler "github.com/gochat/gochat/internal/handler" + "github.com/gochat/gochat/internal/search" + "github.com/redis/go-redis/v9" + "gorm.io/gorm" +) + +func newReadinessChecks(cfg *config.Config, db *gorm.DB, rdb redis.UniversalClient, ready *atomic.Bool) []basehandler.DependencyCheck { + latestMigration, latestMigrationErr := database.LatestVersion(cfg.Database.GetMigrationsPath()) + checks := []basehandler.DependencyCheck{ + {Name: "redis", Check: func(ctx context.Context) error { return rdb.Ping(ctx).Err() }}, + {Name: "migrations", Check: func(ctx context.Context) error { + if latestMigrationErr != nil { + return latestMigrationErr + } + return database.CheckVersion(ctx, db, latestMigration) + }}, + {Name: "draining", Check: func(context.Context) error { + if !ready.Load() { + return fmt.Errorf("shutdown in progress") + } + return nil + }}, + } + if !strings.EqualFold(cfg.Search.Engine, search.EngineMeilisearch) { + return checks + } + searchClient := &http.Client{Timeout: time.Duration(cfg.Search.TimeoutSeconds) * time.Second} + return append(checks, basehandler.DependencyCheck{Name: "search", Check: func(ctx context.Context) error { + request, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimRight(cfg.Search.Host, "/")+"/health", nil) + if err != nil { + return err + } + response, err := searchClient.Do(request) + if err != nil { + return err + } + defer response.Body.Close() + _, _ = io.Copy(io.Discard, response.Body) + if response.StatusCode != http.StatusOK { + return fmt.Errorf("search health returned %s", response.Status) + } + return nil + }}) +} diff --git a/backend/internal/app/readiness_test.go b/backend/internal/app/readiness_test.go new file mode 100644 index 00000000..4327977a --- /dev/null +++ b/backend/internal/app/readiness_test.go @@ -0,0 +1,50 @@ +package app + +import ( + "context" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync/atomic" + "testing" + + "github.com/alicebob/miniredis/v2" + "github.com/gochat/gochat/internal/config" + "github.com/redis/go-redis/v9" + "github.com/stretchr/testify/require" + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +func TestReadinessChecksRequiredDependenciesAndDrainState(t *testing.T) { + migrations := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(migrations, "000001_initial.up.sql"), []byte("SELECT 1;"), 0o600)) + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + require.NoError(t, err) + require.NoError(t, db.Exec("CREATE TABLE schema_migrations (version INTEGER NOT NULL, dirty BOOLEAN NOT NULL)").Error) + require.NoError(t, db.Exec("INSERT INTO schema_migrations(version, dirty) VALUES (1, false)").Error) + mini := miniredis.RunT(t) + rdb := redis.NewClient(&redis.Options{Addr: mini.Addr()}) + t.Cleanup(func() { _ = rdb.Close() }) + search := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })) + defer search.Close() + ready := &atomic.Bool{} + ready.Store(true) + cfg := &config.Config{ + Database: config.DatabaseConfig{MigrationsPath: migrations}, + Search: config.SearchConfig{Engine: "meilisearch", Host: search.URL, TimeoutSeconds: 1}, + } + + checks := newReadinessChecks(cfg, db, rdb, ready) + require.Len(t, checks, 4) + byName := map[string]func(context.Context) error{} + for _, check := range checks { + byName[check.Name] = check.Check + require.NoError(t, check.Check(context.Background()), check.Name) + } + ready.Store(false) + require.ErrorContains(t, byName["draining"](context.Background()), "shutdown") + mini.Close() + require.Error(t, byName["redis"](context.Background())) +} diff --git a/backend/internal/app/runtime_test.go b/backend/internal/app/runtime_test.go new file mode 100644 index 00000000..8c4c8c44 --- /dev/null +++ b/backend/internal/app/runtime_test.go @@ -0,0 +1,237 @@ +package app + +import ( + "bufio" + "context" + "errors" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/gin-gonic/gin" + "github.com/gochat/gochat/internal/config" + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/pubsub" + "github.com/gochat/gochat/internal/worker" + "github.com/redis/go-redis/v9" + "github.com/stretchr/testify/require" + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +type shutdownOrderPubSub struct { + db *gorm.DB + jobID uint + status string + err error +} + +type shutdownRedisPubSub struct{ client *redis.Client } + +func (p *shutdownRedisPubSub) Publish(ctx context.Context, topic string, _ pubsub.Event) error { + return p.client.Set(ctx, topic, "available", 0).Err() +} +func (*shutdownRedisPubSub) Subscribe(context.Context, string, pubsub.EventHandler) error { + return nil +} +func (*shutdownRedisPubSub) Unsubscribe(context.Context, string) error { return nil } +func (p *shutdownRedisPubSub) Close() error { return p.client.Close() } + +func (*shutdownOrderPubSub) Publish(context.Context, string, pubsub.Event) error { return nil } +func (*shutdownOrderPubSub) Subscribe(context.Context, string, pubsub.EventHandler) error { + return nil +} +func (*shutdownOrderPubSub) Unsubscribe(context.Context, string) error { return nil } +func (p *shutdownOrderPubSub) Close() error { + var job model.BackgroundJob + p.err = p.db.First(&job, p.jobID).Error + p.status = job.Status + return nil +} + +func TestWebProcessDoesNotConsumeJobs(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + require.NoError(t, err) + require.NoError(t, db.AutoMigrate(&model.BackgroundJob{})) + pool := worker.NewWorkerPool(db) + var handled atomic.Int32 + pool.Register("web_must_not_consume", func(context.Context, *model.BackgroundJob) error { + handled.Add(1) + return nil + }) + _, err = pool.Enqueue(context.Background(), "web_must_not_consume", nil) + require.NoError(t, err) + + engine := gin.New() + engine.GET("/live", func(c *gin.Context) { c.Status(http.StatusOK) }) + application := &App{ + config: &config.Config{Server: config.ServerConfig{ + Host: "127.0.0.1", Port: 0, ReadHeaderTimeoutS: 1, ReadTimeoutS: 1, + WriteTimeoutS: 1, IdleTimeoutS: 1, ShutdownTimeoutS: 1, MaxHeaderBytes: 1024, + }}, + db: db, engine: engine, workerPool: pool, + } + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(50 * time.Millisecond) + cancel() + }() + require.NoError(t, application.runWeb(ctx)) + require.Zero(t, handled.Load()) +} + +func TestWorkerProcessConsumesJobsWithoutHTTP(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + require.NoError(t, err) + require.NoError(t, db.AutoMigrate(&model.BackgroundJob{})) + pool := worker.NewWorkerPoolWithOptions(db, worker.WithPollInterval(5*time.Millisecond)) + ctx, cancel := context.WithCancel(context.Background()) + pool.Register("worker_only", func(context.Context, *model.BackgroundJob) error { + cancel() + return nil + }) + _, err = pool.Enqueue(context.Background(), "worker_only", nil) + require.NoError(t, err) + + occupied, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer occupied.Close() + port := occupied.Addr().(*net.TCPAddr).Port + application := &App{ + config: &config.Config{Server: config.ServerConfig{Host: "127.0.0.1", Port: port, ShutdownTimeoutS: 1}}, + db: db, engine: gin.New(), workerPool: pool, + } + require.NoError(t, application.runWorker(ctx)) +} + +func TestShutdownWaitsForCancelledJobPersistenceBeforeClosingDependencies(t *testing.T) { + db, err := gorm.Open(sqlite.Open("file:shutdown-order?mode=memory&cache=shared"), &gorm.Config{}) + require.NoError(t, err) + require.NoError(t, db.AutoMigrate(&model.BackgroundJob{})) + pool := worker.NewWorkerPoolWithOptions(db, + worker.WithPollInterval(5*time.Millisecond), + worker.WithBackoff(func(int) time.Duration { return 0 }), + ) + started := make(chan struct{}) + pool.Register("shutdown_order", func(ctx context.Context, _ *model.BackgroundJob) error { + close(started) + <-ctx.Done() + return ctx.Err() + }) + job, err := pool.Enqueue(context.Background(), "shutdown_order", nil) + require.NoError(t, err) + require.NoError(t, pool.Start()) + <-started + + ps := &shutdownOrderPubSub{db: db, jobID: job.ID} + application := &App{db: db, pubsub: ps, workerPool: pool} + require.ErrorIs(t, application.Shutdown(20*time.Millisecond), context.DeadlineExceeded) + require.NoError(t, ps.err) + require.Equal(t, model.BackgroundJobStatusRetrying, ps.status) +} + +func TestHTTPShutdownWaitsForSlowHandlerBeforeClosingDependencies(t *testing.T) { + db, err := gorm.Open(sqlite.Open("file:http-shutdown-order?mode=memory&cache=shared"), &gorm.Config{}) + require.NoError(t, err) + redisServer := miniredis.RunT(t) + redisClient := redis.NewClient(&redis.Options{Addr: redisServer.Addr()}) + ps := &shutdownRedisPubSub{client: redisClient} + + started := make(chan struct{}) + release := make(chan struct{}) + handlerErr := make(chan error, 1) + engine := gin.New() + application := &App{db: db, pubsub: ps, engine: engine} + engine.GET("/slow", func(c *gin.Context) { + close(started) + <-release + var one int + dbErr := db.Raw("SELECT 1").Scan(&one).Error + redisErr := ps.Publish(c.Request.Context(), "shutdown-order", pubsub.Event{Type: "still-open"}) + handlerErr <- errors.Join(dbErr, redisErr) + c.Status(http.StatusNoContent) + }) + + server := httptest.NewServer(application.Handler()) + defer server.Close() + requestDone := make(chan error, 1) + go func() { + response, requestErr := http.Get(server.URL + "/slow") + if requestErr == nil { + _ = response.Body.Close() + } + requestDone <- requestErr + }() + <-started + + shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + require.ErrorIs(t, server.Config.Shutdown(shutdownCtx), context.DeadlineExceeded) + shutdownDone := make(chan error, 1) + go func() { shutdownDone <- application.shutdown(shutdownCtx) }() + select { + case err := <-shutdownDone: + t.Fatalf("dependencies closed before slow handler completed: %v", err) + case <-time.After(20 * time.Millisecond): + } + + close(release) + require.NoError(t, <-handlerErr) + require.NoError(t, <-requestDone) + require.NoError(t, <-shutdownDone) + + sqlDB, err := db.DB() + require.NoError(t, err) + require.Error(t, sqlDB.Ping()) + require.Error(t, ps.Publish(context.Background(), "shutdown-order", pubsub.Event{Type: "closed"})) +} + +func TestHTTPServerLimits(t *testing.T) { + engine := gin.New() + engine.GET("/", func(c *gin.Context) { c.Status(http.StatusOK) }) + application := &App{config: &config.Config{Server: config.ServerConfig{ + Host: "127.0.0.1", Port: 0, ReadHeaderTimeoutS: 2, ReadTimeoutS: 3, + WriteTimeoutS: 4, IdleTimeoutS: 5, MaxHeaderBytes: 1024, + }}, engine: engine} + server := application.HTTPServer() + require.Equal(t, 2*time.Second, server.ReadHeaderTimeout) + require.Equal(t, 3*time.Second, server.ReadTimeout) + require.Equal(t, 4*time.Second, server.WriteTimeout) + require.Equal(t, 5*time.Second, server.IdleTimeout) + require.Equal(t, 1024, server.MaxHeaderBytes) + + server.ReadHeaderTimeout = 50 * time.Millisecond + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + serveDone := make(chan error, 1) + go func() { serveDone <- server.Serve(listener) }() + + slow, err := net.Dial("tcp", listener.Addr().String()) + require.NoError(t, err) + _, err = slow.Write([]byte("GET / HTTP/1.1\r\nHost: localhost\r\nX-Slow:")) + require.NoError(t, err) + require.NoError(t, slow.SetReadDeadline(time.Now().Add(500*time.Millisecond))) + started := time.Now() + line, readErr := bufio.NewReader(slow).ReadString('\n') + require.Less(t, time.Since(started), 300*time.Millisecond) + require.True(t, readErr != nil || strings.Contains(line, "400"), "expected timeout rejection, got %q (%v)", line, readErr) + require.NoError(t, slow.Close()) + + large, err := net.Dial("tcp", listener.Addr().String()) + require.NoError(t, err) + _, err = large.Write([]byte("GET / HTTP/1.1\r\nHost: localhost\r\nX-Large: " + strings.Repeat("a", 10_000) + "\r\n\r\n")) + require.NoError(t, err) + require.NoError(t, large.SetReadDeadline(time.Now().Add(time.Second))) + status, err := bufio.NewReader(large).ReadString('\n') + require.NoError(t, err) + require.Contains(t, status, "431") + require.NoError(t, large.Close()) + + require.NoError(t, server.Shutdown(context.Background())) + require.ErrorIs(t, <-serveDone, http.ErrServerClosed) +} diff --git a/backend/internal/app/shutdown.go b/backend/internal/app/shutdown.go index ec16382d..e59e93c4 100644 --- a/backend/internal/app/shutdown.go +++ b/backend/internal/app/shutdown.go @@ -2,9 +2,8 @@ package app import ( "context" - "os" - "os/signal" - "syscall" + "errors" + "fmt" "time" applogger "github.com/gochat/gochat/pkg/logger" @@ -14,15 +13,27 @@ import ( // Pattern follows Chatwoot's Puma graceful shutdown (config/puma.rb) // and Rails signal handling (SIGTERM -> graceful stop). // Order: 1. Stop accepting new connections (done by http.Server.Shutdown in main) -// 2. Stop config hot-reloader -// 3. Close notification delivery service (Watermill router + subscriber) -// 4. Close WebSocket Hub (disconnect all clients) -// 5. Close PubSub (stop message publishing/consuming) -// 6. Close database connection pool -// 7. Flush logger buffers +// 2. Stop config hot-reloader +// 3. Close notification delivery service (Watermill router + subscriber) +// 4. Close WebSocket Hub (disconnect all clients) +// 5. Close PubSub (stop message publishing/consuming) +// 6. Close database connection pool +// 7. Flush logger buffers func (a *App) Shutdown(timeout time.Duration) error { + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + return a.shutdown(ctx) +} + +func (a *App) shutdown(ctx context.Context) error { start := time.Now() applogger.L().Info("Shutting down GoChat application...") + if a.ready != nil { + a.ready.Store(false) + } + var shutdownErrs []error + handlers := a.handlers() + handlers.Stop() // Step 0: Stop config hot-reloader if a.reloader != nil { @@ -30,29 +41,44 @@ func (a *App) Shutdown(timeout time.Duration) error { applogger.L().Info("Config hot-reloader stopped") } - ctx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() + // Stop claiming new jobs first and let active handlers drain. + if a.workerPool != nil { + if err := a.workerPool.Shutdown(ctx); err != nil { + shutdownErrs = append(shutdownErrs, fmt.Errorf("worker shutdown: %w", err)) + } + } // Step 0.5: Close notification delivery service (Watermill router + subscriber) - if a.notificationDeliverySvc != nil { + if a.notificationDeliverySvc != nil && a.notificationRunning.Swap(false) { if closeErr := a.notificationDeliverySvc.Close(); closeErr != nil { applogger.L().Errorf("Notification delivery service close error: %v", closeErr) + shutdownErrs = append(shutdownErrs, closeErr) } else { applogger.L().Info("Notification delivery service closed") } } + // Router.Close may time out while HTTP or notification handlers still use + // shared dependencies. Keep those dependencies alive until every handler exits. + handlers.Wait() + // Step 1: Close WebSocket Hub — disconnect all connected clients if a.wsHub != nil { a.wsHub.Shutdown(ctx) applogger.L().Info("WebSocket hub closed") } + if a.wsRelay != nil { + if err := a.wsRelay.Stop(); err != nil { + shutdownErrs = append(shutdownErrs, err) + } + } // Step 2: Close PubSub — stop event publishing and consuming if a.pubsub != nil { if closer, ok := a.pubsub.(interface{ Close() error }); ok { if closeErr := closer.Close(); closeErr != nil { applogger.L().Errorf("PubSub close error: %v", closeErr) + shutdownErrs = append(shutdownErrs, closeErr) } else { applogger.L().Info("PubSub closed") } @@ -65,6 +91,7 @@ func (a *App) Shutdown(timeout time.Duration) error { if err == nil { if closeErr := sqlDB.Close(); closeErr != nil { applogger.L().Errorf("Database close error: %v", closeErr) + shutdownErrs = append(shutdownErrs, closeErr) } else { applogger.L().Info("Database connection pool closed") } @@ -77,13 +104,5 @@ func (a *App) Shutdown(timeout time.Duration) error { duration := time.Since(start) applogger.L().Infof("GoChat shutdown complete (took %v)", duration) - return nil + return errors.Join(shutdownErrs...) } - -// WaitForShutdownSignal blocks until a termination signal is received. -func WaitForShutdownSignal() { - quit := make(chan os.Signal, 1) - signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) - <-quit - applogger.L().Info("Received shutdown signal") -} \ No newline at end of file diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index 5b3315bf..121bfd68 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -84,10 +84,16 @@ type OAuthConfig struct { } type ServerConfig struct { - Host string `mapstructure:"host"` - Port int `mapstructure:"port"` - Mode string `mapstructure:"mode"` // debug, release, test - CORS CORSConfig `mapstructure:"cors"` + Host string `mapstructure:"host"` + Port int `mapstructure:"port"` + Mode string `mapstructure:"mode"` // debug, release, test + ReadHeaderTimeoutS int `mapstructure:"read_header_timeout_seconds"` + ReadTimeoutS int `mapstructure:"read_timeout_seconds"` + WriteTimeoutS int `mapstructure:"write_timeout_seconds"` + IdleTimeoutS int `mapstructure:"idle_timeout_seconds"` + ShutdownTimeoutS int `mapstructure:"shutdown_timeout_seconds"` + MaxHeaderBytes int `mapstructure:"max_header_bytes"` + CORS CORSConfig `mapstructure:"cors"` } // CORSConfig holds CORS middleware configuration. @@ -474,40 +480,46 @@ func LoadWithEnv(env string) (*Config, error) { // Bind specific env keys that viper can't auto-infer for nested structs // These are common overrides that users set via environment variables envBindings := map[string]string{ - "GOCHAT_SERVER_HOST": "server.host", - "GOCHAT_SERVER_PORT": "server.port", - "GOCHAT_SERVER_MODE": "server.mode", - "GOCHAT_SERVER_CORS_ALLOWED_ORIGINS": "server.cors.allowed_origins", - "GOCHAT_DATABASE_DSN": "database.dsn", - "GOCHAT_DATABASE_MAX_IDLE_CONNS": "database.max_idle_conns", - "GOCHAT_DATABASE_MAX_OPEN_CONNS": "database.max_open_conns", - "GOCHAT_DATABASE_CONN_MAX_LIFETIME": "database.conn_max_lifetime", - "GOCHAT_DATABASE_RUN_MIGRATIONS": "database.run_migrations", - "GOCHAT_DATABASE_MIGRATIONS_PATH": "database.migrations_path", - "GOCHAT_REDIS_DSN": "redis.dsn", - "GOCHAT_REDIS_POOL_SIZE": "redis.pool_size", - "GOCHAT_JWT_SECRET": "jwt.secret", - "JWT_SECRET": "jwt.secret", // Alias for compatibility (no prefix) - "GOCHAT_JWT_PREVIOUS_SECRETS": "jwt.previous_secrets", - "GOCHAT_JWT_ALLOW_INSECURE_HEADER_AUTH": "jwt.allow_insecure_header_auth", - "GOCHAT_JWT_EXPIRY_HOURS": "jwt.expiry_hours", - "GOCHAT_JWT_ACCESS_EXPIRY_MINUTES": "jwt.access_expiry_minutes", - "GOCHAT_JWT_REFRESH_EXPIRY_HOURS": "jwt.refresh_expiry_hours", - "GOCHAT_LOG_LEVEL": "log.level", - "GOCHAT_LOG_FORMAT": "log.format", - "GOCHAT_WORKER_CONCURRENCY": "worker.concurrency", - "GOCHAT_WORKER_REDIS_STREAM_PREFIX": "worker.redis_stream_prefix", - "GOCHAT_WORKER_REDIS_CONSUMER_GROUP": "worker.redis_consumer_group", - "GOCHAT_WORKER_REDIS_BLOCK_TIMEOUT_S": "worker.redis_block_timeout_s", - "GOCHAT_WORKER_REDIS_SWEEP_INTERVAL_S": "worker.redis_sweep_interval_s", - "GOCHAT_SEARCH_ENGINE": "search.engine", - "GOCHAT_SEARCH_HOST": "search.host", - "GOCHAT_SEARCH_API_KEY": "search.api_key", - "GOCHAT_SEARCH_INDEX_PREFIX": "search.index_prefix", - "GOCHAT_SEARCH_TIMEOUT_SECONDS": "search.timeout_seconds", - "GOCHAT_STORAGE_PROVIDER": "storage.provider", - "GOCHAT_STORAGE_LOCAL_PATH": "storage.local_path", - "GOCHAT_STORAGE_MAX_FILE_SIZE": "storage.max_file_size", + "GOCHAT_SERVER_HOST": "server.host", + "GOCHAT_SERVER_PORT": "server.port", + "GOCHAT_SERVER_MODE": "server.mode", + "GOCHAT_SERVER_READ_HEADER_TIMEOUT_SECONDS": "server.read_header_timeout_seconds", + "GOCHAT_SERVER_READ_TIMEOUT_SECONDS": "server.read_timeout_seconds", + "GOCHAT_SERVER_WRITE_TIMEOUT_SECONDS": "server.write_timeout_seconds", + "GOCHAT_SERVER_IDLE_TIMEOUT_SECONDS": "server.idle_timeout_seconds", + "GOCHAT_SERVER_SHUTDOWN_TIMEOUT_SECONDS": "server.shutdown_timeout_seconds", + "GOCHAT_SERVER_MAX_HEADER_BYTES": "server.max_header_bytes", + "GOCHAT_SERVER_CORS_ALLOWED_ORIGINS": "server.cors.allowed_origins", + "GOCHAT_DATABASE_DSN": "database.dsn", + "GOCHAT_DATABASE_MAX_IDLE_CONNS": "database.max_idle_conns", + "GOCHAT_DATABASE_MAX_OPEN_CONNS": "database.max_open_conns", + "GOCHAT_DATABASE_CONN_MAX_LIFETIME": "database.conn_max_lifetime", + "GOCHAT_DATABASE_RUN_MIGRATIONS": "database.run_migrations", + "GOCHAT_DATABASE_MIGRATIONS_PATH": "database.migrations_path", + "GOCHAT_REDIS_DSN": "redis.dsn", + "GOCHAT_REDIS_POOL_SIZE": "redis.pool_size", + "GOCHAT_JWT_SECRET": "jwt.secret", + "JWT_SECRET": "jwt.secret", // Alias for compatibility (no prefix) + "GOCHAT_JWT_PREVIOUS_SECRETS": "jwt.previous_secrets", + "GOCHAT_JWT_ALLOW_INSECURE_HEADER_AUTH": "jwt.allow_insecure_header_auth", + "GOCHAT_JWT_EXPIRY_HOURS": "jwt.expiry_hours", + "GOCHAT_JWT_ACCESS_EXPIRY_MINUTES": "jwt.access_expiry_minutes", + "GOCHAT_JWT_REFRESH_EXPIRY_HOURS": "jwt.refresh_expiry_hours", + "GOCHAT_LOG_LEVEL": "log.level", + "GOCHAT_LOG_FORMAT": "log.format", + "GOCHAT_WORKER_CONCURRENCY": "worker.concurrency", + "GOCHAT_WORKER_REDIS_STREAM_PREFIX": "worker.redis_stream_prefix", + "GOCHAT_WORKER_REDIS_CONSUMER_GROUP": "worker.redis_consumer_group", + "GOCHAT_WORKER_REDIS_BLOCK_TIMEOUT_S": "worker.redis_block_timeout_s", + "GOCHAT_WORKER_REDIS_SWEEP_INTERVAL_S": "worker.redis_sweep_interval_s", + "GOCHAT_SEARCH_ENGINE": "search.engine", + "GOCHAT_SEARCH_HOST": "search.host", + "GOCHAT_SEARCH_API_KEY": "search.api_key", + "GOCHAT_SEARCH_INDEX_PREFIX": "search.index_prefix", + "GOCHAT_SEARCH_TIMEOUT_SECONDS": "search.timeout_seconds", + "GOCHAT_STORAGE_PROVIDER": "storage.provider", + "GOCHAT_STORAGE_LOCAL_PATH": "storage.local_path", + "GOCHAT_STORAGE_MAX_FILE_SIZE": "storage.max_file_size", // G10: OAuth config for new channel integrations (Twitter, Microsoft, Google) "GOCHAT_OAUTH_TWITTER_CLIENT_ID": "oauth.twitter.client_id", "GOCHAT_OAUTH_TWITTER_CLIENT_SECRET": "oauth.twitter.client_secret", @@ -698,6 +710,12 @@ func setDefaults(v *viper.Viper) { v.SetDefault("server.host", "0.0.0.0") v.SetDefault("server.port", 3000) v.SetDefault("server.mode", "debug") + v.SetDefault("server.read_header_timeout_seconds", 5) + v.SetDefault("server.read_timeout_seconds", 30) + v.SetDefault("server.write_timeout_seconds", 30) + v.SetDefault("server.idle_timeout_seconds", 120) + v.SetDefault("server.shutdown_timeout_seconds", 30) + v.SetDefault("server.max_header_bytes", 1<<20) v.SetDefault("database.max_idle_conns", 10) v.SetDefault("database.max_open_conns", 100) diff --git a/backend/internal/database/migrate.go b/backend/internal/database/migrate.go index 7c912dd2..ae995786 100644 --- a/backend/internal/database/migrate.go +++ b/backend/internal/database/migrate.go @@ -1,15 +1,19 @@ package database import ( + "context" "errors" "fmt" "os" + "path/filepath" + "strconv" "strings" "github.com/golang-migrate/migrate/v4" _ "github.com/golang-migrate/migrate/v4/database/postgres" _ "github.com/golang-migrate/migrate/v4/database/sqlite3" _ "github.com/golang-migrate/migrate/v4/source/file" + "gorm.io/gorm" ) var unsupportedPQEnvironmentKeys = []string{"PGSERVICE", "PGSERVICEFILE", "PGREALM"} @@ -164,6 +168,54 @@ func CurrentVersion(dbURL string, migrationsPath string) (uint, bool, error) { return version, dirty, nil } +// LatestVersion returns the highest numbered up migration on disk. +func LatestVersion(migrationsPath string) (uint, error) { + entries, err := os.ReadDir(migrationsPath) + if err != nil { + return 0, fmt.Errorf("read migrations: %w", err) + } + var latest uint64 + for _, entry := range entries { + name := entry.Name() + if entry.IsDir() || !strings.HasSuffix(name, ".up.sql") { + continue + } + prefix := strings.SplitN(filepath.Base(name), "_", 2)[0] + version, parseErr := strconv.ParseUint(prefix, 10, 64) + if parseErr != nil { + continue + } + if version > latest { + latest = version + } + } + if latest == 0 { + return 0, errors.New("no numbered up migrations found") + } + return uint(latest), nil +} + +// CheckVersion verifies that golang-migrate reached the expected clean version. +func CheckVersion(ctx context.Context, db *gorm.DB, expected uint) error { + if db == nil { + return errors.New("database is not configured") + } + var state struct { + Version uint + Dirty bool + } + if err := db.WithContext(ctx).Table("schema_migrations").Select("version, dirty").Take(&state).Error; err != nil { + return fmt.Errorf("read schema migration state: %w", err) + } + if state.Dirty { + return fmt.Errorf("schema migration %d is dirty", state.Version) + } + if state.Version != expected { + return fmt.Errorf("schema migration version %d, expected %d", state.Version, expected) + } + return nil +} + func newMigrate(dbURL, migrationsPath string) (*migrate.Migrate, error) { if strings.TrimSpace(migrationsPath) == "" { return nil, errors.New("migration path must not be empty") diff --git a/backend/internal/database/migrate_test.go b/backend/internal/database/migrate_test.go index e4d214b2..1c40a78c 100644 --- a/backend/internal/database/migrate_test.go +++ b/backend/internal/database/migrate_test.go @@ -1,10 +1,14 @@ package database import ( + "context" "os" + "path/filepath" "testing" "github.com/stretchr/testify/require" + "gorm.io/driver/sqlite" + "gorm.io/gorm" ) func TestMigrationCommandsRejectBlankPath(t *testing.T) { @@ -56,3 +60,23 @@ func TestSanitizePostgresEnvironment(t *testing.T) { require.False(t, exists, "%s should be removed at application startup", key) } } + +func TestLatestVersionAndCheckVersion(t *testing.T) { + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, "000001_first.up.sql"), []byte("SELECT 1;"), 0o600)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "000012_latest.up.sql"), []byte("SELECT 1;"), 0o600)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "README"), nil, 0o600)) + + latest, err := LatestVersion(dir) + require.NoError(t, err) + require.Equal(t, uint(12), latest) + + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + require.NoError(t, err) + require.NoError(t, db.Exec("CREATE TABLE schema_migrations (version INTEGER NOT NULL, dirty BOOLEAN NOT NULL)").Error) + require.NoError(t, db.Exec("INSERT INTO schema_migrations(version, dirty) VALUES (12, false)").Error) + require.NoError(t, CheckVersion(context.Background(), db, latest)) + require.Error(t, CheckVersion(context.Background(), db, 13)) + require.NoError(t, db.Exec("UPDATE schema_migrations SET dirty = true").Error) + require.ErrorContains(t, CheckVersion(context.Background(), db, latest), "dirty") +} diff --git a/backend/internal/handler/handler_test.go b/backend/internal/handler/handler_test.go index a43a43cb..2d80c381 100644 --- a/backend/internal/handler/handler_test.go +++ b/backend/internal/handler/handler_test.go @@ -1,7 +1,9 @@ package handler import ( + "context" "encoding/json" + "errors" "net/http" "net/http/httptest" "testing" @@ -93,6 +95,49 @@ func TestReadyHandler_Ready(t *testing.T) { assert.True(t, resp["ready"].(bool)) } +func TestReadyHandler_RequiredDependencyFailure(t *testing.T) { + db := newTestDB(t) + router := gin.New() + router.GET("/ready", ReadyHandler(db, DependencyCheck{Name: "redis", Check: func(context.Context) error { + return errors.New("connection refused") + }})) + + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/ready", nil)) + assert.Equal(t, http.StatusServiceUnavailable, w.Code) + assert.JSONEq(t, `{"ready":false,"checks":{"database":"healthy","redis":"unhealthy: connection refused"}}`, w.Body.String()) +} + +func TestDatabaseOutageFailsReadinessButNotLiveness(t *testing.T) { + db := newTestDB(t) + sqlDB, err := db.DB() + require.NoError(t, err) + require.NoError(t, sqlDB.Close()) + router := gin.New() + router.GET("/ready", ReadyHandler(db)) + router.GET("/live", LiveHandler()) + + ready := httptest.NewRecorder() + router.ServeHTTP(ready, httptest.NewRequest(http.MethodGet, "/ready", nil)) + assert.Equal(t, http.StatusServiceUnavailable, ready.Code) + live := httptest.NewRecorder() + router.ServeHTTP(live, httptest.NewRequest(http.MethodGet, "/live", nil)) + assert.Equal(t, http.StatusOK, live.Code) +} + +func TestHealthHandler_SelectedDependency(t *testing.T) { + db := newTestDB(t) + router := gin.New() + router.GET("/health", HealthHandler(db, time.Now(), "test", DependencyCheck{Name: "redis", Check: func(context.Context) error { return nil }})) + + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/health?check=redis", nil)) + assert.Equal(t, http.StatusOK, w.Code) + var response HealthResponse + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &response)) + assert.Equal(t, map[string]string{"redis": "healthy"}, response.Checks) +} + func TestReadyHandler_NilDB_NotReady(t *testing.T) { // ReadyHandler does not handle nil DB (panics on db.DB()). t.Skip("ReadyHandler panics on nil DB; nil DB is not a valid runtime state") @@ -144,6 +189,23 @@ func TestPrometheusHandler(t *testing.T) { assert.Contains(t, body, "# TYPE") } +func TestHTTPMetricsRecordsRequestsErrorsAndLatency(t *testing.T) { + metrics := NewHTTPMetrics(time.Now()) + router := gin.New() + router.Use(metrics.Middleware()) + router.GET("/items/:id", func(c *gin.Context) { c.Status(http.StatusInternalServerError) }) + router.GET("/metrics", metrics.Handler()) + + router.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/items/42", nil)) + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + + body := w.Body.String() + assert.Contains(t, body, `http_requests_total{method="GET",route="/items/:id",status="500"} 1`) + assert.Contains(t, body, `http_request_errors_total{method="GET",route="/items/:id",status="500"} 1`) + assert.Contains(t, body, `http_request_duration_seconds_count{method="GET",route="/items/:id",status="500"} 1`) +} + func TestFormatGauge(t *testing.T) { result := formatGauge("test_metric", 42) assert.Equal(t, "test_metric 42", result) diff --git a/backend/internal/handler/health_handler.go b/backend/internal/handler/health_handler.go index d38deea8..bf9d9a89 100644 --- a/backend/internal/handler/health_handler.go +++ b/backend/internal/handler/health_handler.go @@ -1,6 +1,8 @@ package handler import ( + "context" + "errors" "net/http" "runtime" "strconv" @@ -10,6 +12,11 @@ import ( "gorm.io/gorm" ) +type DependencyCheck struct { + Name string + Check func(context.Context) error +} + // HealthResponse is the structured health check response. type HealthResponse struct { Status string `json:"status"` @@ -19,52 +26,26 @@ type HealthResponse struct { Checks map[string]string `json:"checks"` } -// HealthHandler returns application health status. -// Reference: Chatwoot uses /health for monitoring in docker-compose.production.yaml -func HealthHandler(db *gorm.DB, startTime time.Time, version string) gin.HandlerFunc { +// HealthHandler reports dependency state and optional runtime diagnostics. +func HealthHandler(db *gorm.DB, startTime time.Time, version string, dependencies ...DependencyCheck) gin.HandlerFunc { return func(c *gin.Context) { - checks := make(map[string]string) - overall := "healthy" - - // Database check - sqlDB, err := db.DB() - if err != nil { - checks["database"] = "unhealthy: " + err.Error() - overall = "unhealthy" - } else if err := sqlDB.Ping(); err != nil { - checks["database"] = "unhealthy: " + err.Error() - overall = "unhealthy" - } else { - checks["database"] = "healthy" - } - - // Memory check - var m runtime.MemStats - runtime.ReadMemStats(&m) - memMB := m.Alloc / 1024 / 1024 - checks["memory_alloc_mb"] = strconv.FormatUint(memMB, 10) - if memMB > 500 { - checks["memory_warning"] = "high memory usage" - } - - // Goroutine check - goroutines := runtime.NumGoroutine() - checks["goroutines"] = strconv.Itoa(goroutines) - if goroutines > 1000 { - checks["goroutine_warning"] = "high goroutine count" - } - - // Uptime + selected := c.Query("check") + checks, healthy := runDependencyChecks(c.Request.Context(), db, selected, dependencies) uptime := time.Since(startTime) - checks["uptime_seconds"] = strconv.FormatUint(uint64(uptime.Seconds()), 10) - - statusCode := http.StatusOK - if overall == "unhealthy" { - statusCode = http.StatusServiceUnavailable + if selected == "" { + var memory runtime.MemStats + runtime.ReadMemStats(&memory) + checks["memory_alloc_mb"] = strconv.FormatUint(memory.Alloc/1024/1024, 10) + checks["goroutines"] = strconv.Itoa(runtime.NumGoroutine()) + checks["uptime_seconds"] = strconv.FormatUint(uint64(uptime.Seconds()), 10) } + status, statusCode := "healthy", http.StatusOK + if !healthy { + status, statusCode = "unhealthy", http.StatusServiceUnavailable + } c.JSON(statusCode, HealthResponse{ - Status: overall, + Status: status, Timestamp: time.Now().UTC().Format(time.RFC3339), Version: version, Uptime: uptime.String(), @@ -73,27 +54,64 @@ func HealthHandler(db *gorm.DB, startTime time.Time, version string) gin.Handler } } -// ReadyHandler returns whether the app is ready to accept traffic. -// Used by K8s readiness probes — returns 503 if not ready. -func ReadyHandler(db *gorm.DB) gin.HandlerFunc { +// ReadyHandler returns 503 while draining or when a required dependency fails. +func ReadyHandler(db *gorm.DB, dependencies ...DependencyCheck) gin.HandlerFunc { return func(c *gin.Context) { - sqlDB, err := db.DB() - if err != nil { - c.JSON(http.StatusServiceUnavailable, gin.H{"ready": false, "reason": "db error"}) - return + checks, ready := runDependencyChecks(c.Request.Context(), db, "", dependencies) + statusCode := http.StatusOK + if !ready { + statusCode = http.StatusServiceUnavailable } - if err := sqlDB.Ping(); err != nil { - c.JSON(http.StatusServiceUnavailable, gin.H{"ready": false, "reason": "db unreachable"}) - return - } - c.JSON(http.StatusOK, gin.H{"ready": true}) + c.JSON(statusCode, gin.H{"ready": ready, "checks": checks}) } } -// LiveHandler returns whether the app process is alive. -// Used by K8s liveness probes — simplest possible check. +// LiveHandler only proves that the process can serve HTTP; dependency failures +// belong to readiness so an outage does not trigger a restart loop. func LiveHandler() gin.HandlerFunc { return func(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"alive": true}) } } + +func runDependencyChecks(ctx context.Context, db *gorm.DB, selected string, dependencies []DependencyCheck) (map[string]string, bool) { + checks := make(map[string]string, len(dependencies)+1) + healthy := true + all := append([]DependencyCheck{{Name: "database", Check: databasePing(db)}}, dependencies...) + found := selected == "" + for _, dependency := range all { + if selected != "" && dependency.Name != selected { + continue + } + found = true + if dependency.Check == nil { + checks[dependency.Name] = "unhealthy: check is not configured" + healthy = false + continue + } + if err := dependency.Check(ctx); err != nil { + checks[dependency.Name] = "unhealthy: " + err.Error() + healthy = false + } else { + checks[dependency.Name] = "healthy" + } + } + if !found { + checks[selected] = "unhealthy: unknown check" + healthy = false + } + return checks, healthy +} + +func databasePing(db *gorm.DB) func(context.Context) error { + return func(ctx context.Context) error { + if db == nil { + return errors.New("database is not configured") + } + sqlDB, err := db.DB() + if err != nil { + return err + } + return sqlDB.PingContext(ctx) + } +} diff --git a/backend/internal/handler/metrics_handler.go b/backend/internal/handler/metrics_handler.go index 4db6bdde..13b0aa4d 100644 --- a/backend/internal/handler/metrics_handler.go +++ b/backend/internal/handler/metrics_handler.go @@ -4,79 +4,147 @@ import ( "fmt" "net/http" "runtime" + "sort" + "strconv" + "strings" + "sync" "time" "github.com/gin-gonic/gin" ) -// MetricsHandler exposes Prometheus-compatible metrics in text exposition format. -// Reference: Chatwoot uses Prometheus exporter for Sidekiq, Rails metrics -// This provides Go runtime + application metrics on a dedicated port. +var durationBuckets = [...]float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5} -// PrometheusHandler returns metrics in Prometheus text format. -func PrometheusHandler(startTime time.Time) gin.HandlerFunc { +type requestMetricKey struct { + Method string + Route string + Status int +} + +type requestMetric struct { + Count uint64 + Errors uint64 + Sum float64 + Buckets [len(durationBuckets)]uint64 +} + +// HTTPMetrics records bounded route-template labels without a Prometheus client dependency. +type HTTPMetrics struct { + startTime time.Time + mu sync.RWMutex + requests map[requestMetricKey]requestMetric +} + +func NewHTTPMetrics(startTime time.Time) *HTTPMetrics { + return &HTTPMetrics{startTime: startTime, requests: make(map[requestMetricKey]requestMetric)} +} + +func (m *HTTPMetrics) Middleware() gin.HandlerFunc { return func(c *gin.Context) { - var m runtime.MemStats - runtime.ReadMemStats(&m) - - uptime := time.Since(startTime).Seconds() - - metrics := []string{ - // Go runtime metrics - formatGauge("gochat_go_goroutines", uint64(runtime.NumGoroutine())), - formatGauge("gochat_go_memory_alloc_bytes", m.Alloc), - formatGauge("gochat_go_memory_sys_bytes", m.Sys), - formatGauge("gochat_go_memory_total_alloc_bytes", m.TotalAlloc), - formatGauge("gochat_go_gc_pause_total_ns", m.PauseTotalNs), - formatCounter("gochat_go_gc_count", uint64(m.NumGC)), - // Application metrics - formatGauge("gochat_uptime_seconds", uint64(uptime)), - formatGauge("gochat_threads_count", uint64(runtime.NumCPU())), + started := time.Now() + c.Next() + route := c.FullPath() + if route == "" { + route = "unmatched" } - - // HELP and TYPE annotations - help := []string{ - "# HELP gochat_go_goroutines Number of goroutines currently running", - "# TYPE gochat_go_goroutines gauge", - "# HELP gochat_go_memory_alloc_bytes Bytes of allocated heap objects", - "# TYPE gochat_go_memory_alloc_bytes gauge", - "# HELP gochat_go_memory_sys_bytes Bytes obtained from system", - "# TYPE gochat_go_memory_sys_bytes gauge", - "# HELP gochat_uptime_seconds Application uptime in seconds", - "# TYPE gochat_uptime_seconds gauge", - } - - output := "" - for _, h := range help { - output += h + "\n" - } - for _, m := range metrics { - output += m + "\n" - } - - c.Header("Content-Type", "text/plain; version=0.0.4; charset=utf-8") - c.String(http.StatusOK, output) + m.observe(requestMetricKey{Method: c.Request.Method, Route: route, Status: c.Writer.Status()}, time.Since(started).Seconds()) } } -func formatGauge(name string, value uint64) string { - return name + " " + formatValue(value) +func (m *HTTPMetrics) observe(key requestMetricKey, seconds float64) { + m.mu.Lock() + metric := m.requests[key] + metric.Count++ + metric.Sum += seconds + if key.Status >= http.StatusInternalServerError { + metric.Errors++ + } + for i, boundary := range durationBuckets { + if seconds <= boundary { + metric.Buckets[i]++ + } + } + m.requests[key] = metric + m.mu.Unlock() } -func formatCounter(name string, value uint64) string { - return name + " " + formatValue(value) +func (m *HTTPMetrics) Handler() gin.HandlerFunc { + return func(c *gin.Context) { + var memory runtime.MemStats + runtime.ReadMemStats(&memory) + keys, snapshot := m.snapshot() + + var output strings.Builder + output.WriteString("# HELP http_requests_total Total HTTP requests\n# TYPE http_requests_total counter\n") + output.WriteString("# HELP http_request_errors_total Total HTTP 5xx responses\n# TYPE http_request_errors_total counter\n") + output.WriteString("# HELP http_request_duration_seconds HTTP request duration\n# TYPE http_request_duration_seconds histogram\n") + for _, key := range keys { + metric := snapshot[key] + labels := requestLabels(key) + fmt.Fprintf(&output, "http_requests_total{%s} %d\n", labels, metric.Count) + fmt.Fprintf(&output, "http_request_errors_total{%s} %d\n", labels, metric.Errors) + for i, boundary := range durationBuckets { + fmt.Fprintf(&output, "http_request_duration_seconds_bucket{%s,le=%q} %d\n", labels, strconv.FormatFloat(boundary, 'g', -1, 64), metric.Buckets[i]) + } + fmt.Fprintf(&output, "http_request_duration_seconds_bucket{%s,le=\"+Inf\"} %d\n", labels, metric.Count) + fmt.Fprintf(&output, "http_request_duration_seconds_sum{%s} %s\n", labels, strconv.FormatFloat(metric.Sum, 'g', -1, 64)) + fmt.Fprintf(&output, "http_request_duration_seconds_count{%s} %d\n", labels, metric.Count) + } + + output.WriteString("# HELP gochat_go_goroutines Number of goroutines currently running\n# TYPE gochat_go_goroutines gauge\n") + output.WriteString("# HELP gochat_go_memory_alloc_bytes Bytes of allocated heap objects\n# TYPE gochat_go_memory_alloc_bytes gauge\n") + output.WriteString("# HELP gochat_go_memory_sys_bytes Bytes obtained from system\n# TYPE gochat_go_memory_sys_bytes gauge\n") + output.WriteString("# HELP gochat_uptime_seconds Application uptime in seconds\n# TYPE gochat_uptime_seconds gauge\n") + fmt.Fprintf(&output, "gochat_go_goroutines %d\n", runtime.NumGoroutine()) + fmt.Fprintf(&output, "gochat_go_memory_alloc_bytes %d\n", memory.Alloc) + fmt.Fprintf(&output, "gochat_go_memory_sys_bytes %d\n", memory.Sys) + fmt.Fprintf(&output, "gochat_go_memory_total_alloc_bytes %d\n", memory.TotalAlloc) + fmt.Fprintf(&output, "gochat_go_gc_pause_total_ns %d\n", memory.PauseTotalNs) + fmt.Fprintf(&output, "gochat_go_gc_count %d\n", memory.NumGC) + fmt.Fprintf(&output, "gochat_uptime_seconds %s\n", strconv.FormatFloat(time.Since(m.startTime).Seconds(), 'f', 3, 64)) + fmt.Fprintf(&output, "gochat_threads_count %d\n", runtime.NumCPU()) + + c.Data(http.StatusOK, "text/plain; version=0.0.4; charset=utf-8", []byte(output.String())) + } } -func formatValue(v uint64) string { - // Simple uint64 formatting without strconv dependency - if v == 0 { - return "0" +func (m *HTTPMetrics) snapshot() ([]requestMetricKey, map[requestMetricKey]requestMetric) { + m.mu.RLock() + snapshot := make(map[requestMetricKey]requestMetric, len(m.requests)) + keys := make([]requestMetricKey, 0, len(m.requests)) + for key, metric := range m.requests { + keys = append(keys, key) + snapshot[key] = metric } - result := "" - for v > 0 { - digit := v % 10 - result = fmt.Sprintf("%d%s", digit, result) - v /= 10 - } - return result + m.mu.RUnlock() + sort.Slice(keys, func(i, j int) bool { + left, right := keys[i], keys[j] + if left.Route != right.Route { + return left.Route < right.Route + } + if left.Method != right.Method { + return left.Method < right.Method + } + return left.Status < right.Status + }) + return keys, snapshot } + +func requestLabels(key requestMetricKey) string { + return fmt.Sprintf(`method="%s",route="%s",status="%s"`, escapeLabel(key.Method), escapeLabel(key.Route), strconv.Itoa(key.Status)) +} + +func escapeLabel(value string) string { + value = strings.ReplaceAll(value, `\`, `\\`) + value = strings.ReplaceAll(value, "\n", `\n`) + return strings.ReplaceAll(value, `"`, `\"`) +} + +// PrometheusHandler is retained for callers that only need runtime metrics. +func PrometheusHandler(startTime time.Time) gin.HandlerFunc { + return NewHTTPMetrics(startTime).Handler() +} + +func formatGauge(name string, value uint64) string { return name + " " + formatValue(value) } +func formatCounter(name string, value uint64) string { return name + " " + formatValue(value) } +func formatValue(value uint64) string { return strconv.FormatUint(value, 10) } diff --git a/backend/internal/lifecycle/handler_group.go b/backend/internal/lifecycle/handler_group.go new file mode 100644 index 00000000..7168d8c0 --- /dev/null +++ b/backend/internal/lifecycle/handler_group.go @@ -0,0 +1,63 @@ +package lifecycle + +import ( + "context" + "sync" +) + +// HandlerGroup stops new handlers, cancels their shared lifecycle context, +// and waits for handlers already using application dependencies. +type HandlerGroup struct { + ctx context.Context + cancel context.CancelFunc + + mu sync.Mutex + done *sync.Cond + active int + stopped bool +} + +func NewHandlerGroup() *HandlerGroup { + ctx, cancel := context.WithCancel(context.Background()) + g := &HandlerGroup{ctx: ctx, cancel: cancel} + g.done = sync.NewCond(&g.mu) + return g +} + +func (g *HandlerGroup) Context() context.Context { return g.ctx } + +func (g *HandlerGroup) Begin() bool { + g.mu.Lock() + defer g.mu.Unlock() + if g.stopped { + return false + } + g.active++ + return true +} + +func (g *HandlerGroup) Done() { + g.mu.Lock() + g.active-- + if g.active == 0 { + g.done.Broadcast() + } + g.mu.Unlock() +} + +func (g *HandlerGroup) Stop() { + g.mu.Lock() + if !g.stopped { + g.stopped = true + g.cancel() + } + g.mu.Unlock() +} + +func (g *HandlerGroup) Wait() { + g.mu.Lock() + for g.active != 0 { + g.done.Wait() + } + g.mu.Unlock() +} diff --git a/backend/internal/middleware/coverage7_test.go b/backend/internal/middleware/coverage7_test.go index 480df1dc..cd323f0b 100644 --- a/backend/internal/middleware/coverage7_test.go +++ b/backend/internal/middleware/coverage7_test.go @@ -1859,6 +1859,11 @@ func TestIsRateLimitExemptPath_Metrics_Cov7(t *testing.T) { assert.True(t, isRateLimitExemptPath("/metrics")) } +func TestIsRateLimitExemptPath_Probes_Cov7(t *testing.T) { + assert.True(t, isRateLimitExemptPath("/ready")) + assert.True(t, isRateLimitExemptPath("/live")) +} + func TestIsRateLimitExemptPath_Other_Cov7(t *testing.T) { assert.False(t, isRateLimitExemptPath("/api/v1/conversations")) } diff --git a/backend/internal/middleware/rate_limit.go b/backend/internal/middleware/rate_limit.go index 8be11487..430896fc 100644 --- a/backend/internal/middleware/rate_limit.go +++ b/backend/internal/middleware/rate_limit.go @@ -305,7 +305,7 @@ func RateLimit(rdb *redis.Client) gin.HandlerFunc { func isRateLimitExemptPath(path string) bool { switch path { - case "/health", "/metrics": + case "/health", "/ready", "/live", "/metrics": return true default: return false diff --git a/backend/internal/router/coverage_test.go b/backend/internal/router/coverage_test.go deleted file mode 100644 index 61e5fd74..00000000 --- a/backend/internal/router/coverage_test.go +++ /dev/null @@ -1,16 +0,0 @@ -package router - -import ( - "net/http/httptest" - "testing" - - "github.com/gin-gonic/gin" -) - -func TestHealthCheck_Cov1(t *testing.T) { - gin.SetMode(gin.TestMode) - w := httptest.NewRecorder() - c, _ := gin.CreateTestContext(w) - c.Request = httptest.NewRequest("GET", "/health", nil) - healthCheck(c) -} diff --git a/backend/internal/router/router.go b/backend/internal/router/router.go index 84c1da63..37fc9d2c 100644 --- a/backend/internal/router/router.go +++ b/backend/internal/router/router.go @@ -11,7 +11,6 @@ import ( "path/filepath" "strconv" "strings" - "time" "github.com/gin-gonic/gin" swaggerFiles "github.com/swaggo/files" @@ -33,9 +32,6 @@ import ( "gorm.io/gorm" ) -// startTime records when the application process launched, used for uptime in /health. -var startTime = time.Now() - // Handlers holds all instantiated handler structs for route registration. // Passed from bootstrap to avoid global state and keep dependency wiring explicit. type Handlers struct { @@ -201,9 +197,6 @@ func RegisterRoutes( corsCfg middleware.CORSConfig, db *gorm.DB, ) { - // Health check endpoint (ref: Chatwoot health_check route) - engine.GET("/health", healthCheck) - // Swagger UI — interactive API documentation engine.GET("/swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler)) @@ -2238,19 +2231,6 @@ func webhookProviderUnavailable(c *gin.Context) { }) } -func healthCheck(c *gin.Context) { - uptime := time.Since(startTime) - c.JSON(200, gin.H{ - "status": "ok", - "service": "gochat", - "version": config.Version, - "commit": config.CommitSHA, - "buildDate": config.BuildDate, - "uptime": uptime.String(), - "uptimeSeconds": uint64(uptime.Seconds()), - }) -} - func dashboardIndex(c *gin.Context) { if dashboardWantsJSON(c) { c.JSON(http.StatusNotAcceptable, gin.H{"error": "Please use API routes instead of dashboard routes for JSON requests"}) diff --git a/backend/internal/service/notification_delivery_lifecycle_test.go b/backend/internal/service/notification_delivery_lifecycle_test.go new file mode 100644 index 00000000..861c52c3 --- /dev/null +++ b/backend/internal/service/notification_delivery_lifecycle_test.go @@ -0,0 +1,93 @@ +package service + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/ThreeDotsLabs/watermill" + "github.com/ThreeDotsLabs/watermill/message" + "github.com/gochat/gochat/internal/lifecycle" + "github.com/stretchr/testify/require" + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +type notificationLifecycleSubscriber struct { + ready chan struct{} + messages chan *message.Message + once sync.Once + ctx context.Context +} + +func (s *notificationLifecycleSubscriber) Subscribe(ctx context.Context, _ string) (<-chan *message.Message, error) { + s.ctx = ctx + s.once.Do(func() { close(s.ready) }) + return s.messages, nil +} + +func (*notificationLifecycleSubscriber) Close() error { return nil } + +func (s *notificationLifecycleSubscriber) Send(msg *message.Message) { + msg.SetContext(s.ctx) + s.messages <- msg +} + +func TestNotificationCloseTimeoutWaitsForHandlerBeforeDependenciesClose(t *testing.T) { + db, err := gorm.Open(sqlite.Open("file:notification-shutdown-order?mode=memory&cache=shared"), &gorm.Config{}) + require.NoError(t, err) + router, err := message.NewRouter(message.RouterConfig{CloseTimeout: 10 * time.Millisecond}, watermill.NopLogger{}) + require.NoError(t, err) + subscriber := ¬ificationLifecycleSubscriber{ + ready: make(chan struct{}), + messages: make(chan *message.Message), + } + handlers := lifecycle.NewHandlerGroup() + service := &NotificationDeliveryService{router: router, subscriber: subscriber, handlers: handlers} + + started := make(chan struct{}) + cancelled := make(chan struct{}) + probeDependency := make(chan struct{}) + handlerDBErr := make(chan error, 1) + router.AddNoPublisherHandler("blocked", "blocked", subscriber, service.track(func(msg *message.Message) error { + close(started) + <-msg.Context().Done() + close(cancelled) + <-probeDependency + var one int + handlerDBErr <- db.Raw("SELECT 1").Scan(&one).Error + return nil + })) + + runDone := make(chan error, 1) + go func() { runDone <- service.Start(handlers.Context()) }() + <-subscriber.ready + subscriber.Send(message.NewMessage("blocked", nil)) + <-started + + handlers.Stop() + <-cancelled + require.ErrorContains(t, service.Close(), "router close timeout") + + waitDone := make(chan struct{}) + go func() { + handlers.Wait() + close(waitDone) + }() + select { + case <-waitDone: + t.Fatal("handler gate opened before the blocked handler completed") + case <-time.After(20 * time.Millisecond): + } + require.NoError(t, db.Exec("SELECT 1").Error) + + close(probeDependency) + require.NoError(t, <-handlerDBErr) + <-waitDone + sqlDB, err := db.DB() + require.NoError(t, err) + require.NoError(t, sqlDB.Close()) + require.Error(t, sqlDB.Ping()) + require.NoError(t, <-runDone) +} diff --git a/backend/internal/service/notification_delivery_service.go b/backend/internal/service/notification_delivery_service.go index c7e03f53..d1611cba 100644 --- a/backend/internal/service/notification_delivery_service.go +++ b/backend/internal/service/notification_delivery_service.go @@ -3,6 +3,7 @@ package service import ( "context" "encoding/json" + "errors" "fmt" "github.com/ThreeDotsLabs/watermill" @@ -12,6 +13,7 @@ import ( applogger "github.com/gochat/gochat/pkg/logger" + "github.com/gochat/gochat/internal/lifecycle" "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/pubsub" "github.com/gochat/gochat/internal/repository" @@ -22,24 +24,26 @@ import ( // creation and multi-channel delivery (push, email, webhook). // // Architecture mapping: -// Chatwoot Wisper → Sidekiq workers → NotificationDeliveryService -// Each event (message_created, conversation_assigned, etc.) triggers: -// 1. Create a Notification record in the DB -// 2. Check NotificationPreference for push/email enabled -// 3. Deliver push via PushDeliveryService -// 4. Deliver webhook via WebhookDeliveryService (HTTP POST with HMAC signing) +// +// Chatwoot Wisper → Sidekiq workers → NotificationDeliveryService +// Each event (message_created, conversation_assigned, etc.) triggers: +// 1. Create a Notification record in the DB +// 2. Check NotificationPreference for push/email enabled +// 3. Deliver push via PushDeliveryService +// 4. Deliver webhook via WebhookDeliveryService (HTTP POST with HMAC signing) // // Reference: Chatwoot notification_service.rb + P2B M8 spec type NotificationDeliveryService struct { - notificationService *NotificationService - pushDeliveryService *PushDeliveryService - webhookDeliverySvc *WebhookDeliveryService - webhookSignatureSvc *security.WebhookSignatureService - notifPrefRepo *repository.NotificationPreferenceRepo - pushTokenRepo *repository.PushTokenRepo - webhookSubRepo *repository.WebhookSubscriptionRepo - router *message.Router - subscriber *redisstream.Subscriber + notificationService *NotificationService + pushDeliveryService *PushDeliveryService + webhookDeliverySvc *WebhookDeliveryService + webhookSignatureSvc *security.WebhookSignatureService + notifPrefRepo *repository.NotificationPreferenceRepo + pushTokenRepo *repository.PushTokenRepo + webhookSubRepo *repository.WebhookSubscriptionRepo + router *message.Router + subscriber message.Subscriber + handlers *lifecycle.HandlerGroup } // NewNotificationDeliveryService creates a delivery service and registers Watermill handlers. @@ -72,15 +76,15 @@ func NewNotificationDeliveryService( } s := &NotificationDeliveryService{ - notificationService: notificationService, - pushDeliveryService: pushDeliveryService, - webhookDeliverySvc: webhookDeliverySvc, - webhookSignatureSvc: webhookSignatureSvc, - notifPrefRepo: notifPrefRepo, - pushTokenRepo: pushTokenRepo, - webhookSubRepo: webhookSubRepo, - router: router, - subscriber: subscriber, + notificationService: notificationService, + pushDeliveryService: pushDeliveryService, + webhookDeliverySvc: webhookDeliverySvc, + webhookSignatureSvc: webhookSignatureSvc, + notifPrefRepo: notifPrefRepo, + pushTokenRepo: pushTokenRepo, + webhookSubRepo: webhookSubRepo, + router: router, + subscriber: subscriber, } s.registerHandlers() @@ -116,7 +120,7 @@ func (s *NotificationDeliveryService) registerHandlers() { handlerName, topic, s.subscriber, - s.handleNotificationEvent(notifType), + s.track(s.handleNotificationEvent(notifType)), ) } @@ -125,10 +129,26 @@ func (s *NotificationDeliveryService) registerHandlers() { "notif-delivery-system-notification-handler", pubsub.TopicSystemNotification, s.subscriber, - s.handleSystemNotification(), + s.track(s.handleSystemNotification()), ) } +func (s *NotificationDeliveryService) SetHandlerGroup(handlers *lifecycle.HandlerGroup) { + s.handlers = handlers +} + +func (s *NotificationDeliveryService) track(handler message.NoPublishHandlerFunc) message.NoPublishHandlerFunc { + return func(msg *message.Message) error { + if s.handlers != nil { + if !s.handlers.Begin() { + return context.Canceled + } + defer s.handlers.Done() + } + return handler(msg) + } +} + // --- Event payload structure --- // The Watermill message payload is expected to be JSON with these fields. @@ -159,7 +179,7 @@ func (s *NotificationDeliveryService) handleNotificationEvent(notifType string) return nil } - ctx := context.Background() + ctx := msg.Context() // Step 1: Create notification record notif := &model.Notification{ @@ -209,7 +229,7 @@ func (s *NotificationDeliveryService) handleSystemNotification() func(msg *messa return nil } - ctx := context.Background() + ctx := msg.Context() notif := &model.Notification{ AccountID: nilIfZero(payload.AccountID), @@ -313,10 +333,11 @@ func (s *NotificationDeliveryService) Start(ctx context.Context) error { // Close shuts down the delivery router and subscriber. func (s *NotificationDeliveryService) Close() error { - if err := s.router.Close(); err != nil { - applogger.L().Errorf("notif-delivery: failed to close router: %v", err) + routerErr := s.router.Close() + if routerErr != nil { + applogger.L().Errorf("notif-delivery: failed to close router: %v", routerErr) } - return s.subscriber.Close() + return errors.Join(routerErr, s.subscriber.Close()) } // --- Watermill logger adapter for delivery service --- @@ -340,4 +361,4 @@ func (a *deliveryWatermillAdapter) Trace(msg string, fields watermill.LogFields) func (a *deliveryWatermillAdapter) With(fields watermill.LogFields) watermill.LoggerAdapter { return a -} \ No newline at end of file +} diff --git a/backend/internal/worker/worker.go b/backend/internal/worker/worker.go index 73751e3a..778cf1d7 100644 --- a/backend/internal/worker/worker.go +++ b/backend/internal/worker/worker.go @@ -58,10 +58,13 @@ type WorkerPool struct { blockTimeout time.Duration // XREADGROUP block duration, default 5s sweepInterval time.Duration // compensation sweep interval, default 30s - mu sync.RWMutex - ctx context.Context - cancel context.CancelFunc - wg sync.WaitGroup + mu sync.RWMutex + claimMu sync.RWMutex // serializes lifecycle cancellation with claim transitions + ctx context.Context + cancel context.CancelFunc + jobCtx context.Context + jobCancel context.CancelFunc + wg sync.WaitGroup } type Option func(*WorkerPool) @@ -343,11 +346,21 @@ func (wp *WorkerPool) Start() error { return nil } wp.ctx, wp.cancel = context.WithCancel(context.Background()) - ctx := wp.ctx + wp.jobCtx, wp.jobCancel = context.WithCancel(context.Background()) + ctx, jobCtx := wp.ctx, wp.jobCtx + cancel, jobCancel := wp.cancel, wp.jobCancel workerCount := wp.workerCount wp.mu.Unlock() if _, err := wp.RequeueStaleJobs(ctx); err != nil { + wp.mu.Lock() + wp.cancel = nil + wp.ctx = nil + wp.jobCancel = nil + wp.jobCtx = nil + wp.mu.Unlock() + cancel() + jobCancel() return err } @@ -358,7 +371,7 @@ func (wp *WorkerPool) Start() error { for i := 0; i < workerCount; i++ { wp.wg.Add(1) - go wp.run(ctx, i) + go wp.run(ctx, jobCtx, i) } // Start the sweep goroutine for delayed-job delivery and stale-job recovery. @@ -373,31 +386,98 @@ func (wp *WorkerPool) Start() error { func (wp *WorkerPool) Stop() error { wp.mu.Lock() cancel := wp.cancel + jobCancel := wp.jobCancel wp.cancel = nil wp.ctx = nil + wp.jobCancel = nil + wp.jobCtx = nil wp.mu.Unlock() if cancel != nil { cancel() } + if jobCancel != nil { + jobCancel() + } + wp.claimMu.Lock() + wp.claimMu.Unlock() wp.wg.Wait() return nil } +// Shutdown stops claiming new jobs, lets active handlers finish, and only +// cancels them if the drain deadline expires. +func (wp *WorkerPool) Shutdown(ctx context.Context) error { + wp.mu.Lock() + cancel := wp.cancel + jobCancel := wp.jobCancel + wp.cancel = nil + wp.ctx = nil + wp.jobCancel = nil + wp.jobCtx = nil + wp.mu.Unlock() + if cancel == nil { + return nil + } + cancel() + claimsDone := make(chan struct{}) + go func() { + wp.claimMu.Lock() + wp.claimMu.Unlock() + close(claimsDone) + }() + timedOut := false + select { + case <-claimsDone: + case <-ctx.Done(): + timedOut = true + if jobCancel != nil { + jobCancel() + } + <-claimsDone + } + done := make(chan struct{}) + go func() { + wp.wg.Wait() + close(done) + }() + if timedOut { + <-done + return ctx.Err() + } + select { + case <-done: + if jobCancel != nil { + jobCancel() + } + return nil + case <-ctx.Done(): + if jobCancel != nil { + jobCancel() + } + <-done + return ctx.Err() + } +} + func (wp *WorkerPool) ProcessOne(ctx context.Context) (bool, error) { + return wp.processOne(nil, ctx) +} + +func (wp *WorkerPool) processOne(lifecycleCtx, jobCtx context.Context) (bool, error) { if wp.db == nil { return false, ErrWorkerDatabaseRequired } - if err := ctx.Err(); err != nil { + if err := jobCtx.Err(); err != nil { return false, err } - job, err := wp.claimNext(ctx) + job, err := wp.claimNext(lifecycleCtx, jobCtx) if err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return false, nil } return false, err } - return true, wp.perform(ctx, job) + return true, wp.perform(jobCtx, job) } func (wp *WorkerPool) RequeueStaleJobs(ctx context.Context) (int64, error) { @@ -418,7 +498,7 @@ func (wp *WorkerPool) RequeueStaleJobs(ctx context.Context) (int64, error) { // run is the per-goroutine consume loop. When Redis is configured it uses // XREADGROUP BLOCK; otherwise it falls back to DB polling. -func (wp *WorkerPool) run(ctx context.Context, index int) { +func (wp *WorkerPool) run(ctx, jobCtx context.Context, index int) { defer wp.wg.Done() // Each goroutine gets a unique Redis consumer name so XINFO CONSUMERS @@ -426,7 +506,7 @@ func (wp *WorkerPool) run(ctx context.Context, index int) { consumerID := fmt.Sprintf("%s-%d", wp.workerID, index) if wp.rdb == nil { - wp.runDBPollLoop(ctx) + wp.runDBPollLoop(ctx, jobCtx) return } @@ -460,7 +540,10 @@ func (wp *WorkerPool) run(ctx context.Context, index int) { for _, xstream := range results { for _, msg := range xstream.Messages { - wp.processRedisMessage(ctx, xstream.Stream, msg) + if ctx.Err() != nil { + return + } + wp.processRedisMessage(ctx, jobCtx, xstream.Stream, msg) } } } @@ -468,11 +551,16 @@ func (wp *WorkerPool) run(ctx context.Context, index int) { // runDBPollLoop is the legacy DB-polling loop, used when Redis is unavailable // (e.g. SQLite test mode) or not configured. -func (wp *WorkerPool) runDBPollLoop(ctx context.Context) { +func (wp *WorkerPool) runDBPollLoop(ctx, jobCtx context.Context) { ticker := time.NewTicker(wp.pollInterval) defer ticker.Stop() for { - processed, _ := wp.ProcessOne(ctx) + select { + case <-ctx.Done(): + return + default: + } + processed, _ := wp.processOne(ctx, jobCtx) if processed { continue } @@ -488,24 +576,27 @@ func (wp *WorkerPool) runDBPollLoop(ctx context.Context) { // the handler, and acknowledges the Redis message. The DB claim check // guarantees at-most-once execution even if the same job is XADD'd multiple // times (e.g. by both Enqueue and sweep). -func (wp *WorkerPool) processRedisMessage(ctx context.Context, stream string, msg redis.XMessage) { +func (wp *WorkerPool) processRedisMessage(lifecycleCtx, jobCtx context.Context, stream string, msg redis.XMessage) { + if lifecycleCtx.Err() != nil { + return + } jobIDStr, ok := msg.Values["job_id"] if !ok { - wp.ackRedis(ctx, stream, msg.ID) + wp.ackRedis(jobCtx, stream, msg.ID) return } jobID, err := strconv.ParseUint(fmt.Sprintf("%v", jobIDStr), 10, 64) if err != nil { - wp.ackRedis(ctx, stream, msg.ID) + wp.ackRedis(jobCtx, stream, msg.ID) return } var job model.BackgroundJob - if err := wp.db.WithContext(ctx).First(&job, jobID).Error; err != nil { + if err := wp.db.WithContext(jobCtx).First(&job, jobID).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { // Job was deleted; ACK and drop. - wp.ackRedis(ctx, stream, msg.ID) + wp.ackRedis(jobCtx, stream, msg.ID) return } // DB error — do not ACK so Redis can redeliver to another consumer. @@ -516,8 +607,13 @@ func (wp *WorkerPool) processRedisMessage(ctx context.Context, stream string, ms // Claim check: atomically transition queued/retrying → running. // RowsAffected == 0 means another consumer already claimed it or the // job is not yet due; ACK to avoid redelivery loops. + wp.claimMu.RLock() + if lifecycleCtx.Err() != nil { + wp.claimMu.RUnlock() + return + } now := wp.now() - result := wp.db.WithContext(ctx).Model(&model.BackgroundJob{}). + result := wp.db.WithContext(jobCtx).Model(&model.BackgroundJob{}). Where("id = ? AND status IN ? AND scheduled_at <= ?", jobID, []string{model.BackgroundJobStatusQueued, model.BackgroundJobStatusRetrying}, @@ -529,25 +625,26 @@ func (wp *WorkerPool) processRedisMessage(ctx context.Context, stream string, ms "locked_by": wp.workerID, "attempts": gorm.Expr("attempts + 1"), }) + wp.claimMu.RUnlock() if result.Error != nil { applogger.L().Errorf("claim job %d failed: %v", jobID, result.Error) return } if result.RowsAffected == 0 { - wp.ackRedis(ctx, stream, msg.ID) + wp.ackRedis(jobCtx, stream, msg.ID) return } // Reload the job with updated attempts/locked_by fields. - wp.db.WithContext(ctx).First(&job, jobID) + wp.db.WithContext(jobCtx).First(&job, jobID) - if err := wp.perform(ctx, &job); err != nil { - if failErr := wp.fail(ctx, &job, err); failErr != nil { + if err := wp.perform(jobCtx, &job); err != nil { + if failErr := wp.fail(jobCtx, &job, err); failErr != nil { applogger.L().Errorf("record job %d failure: %v", job.ID, failErr) } } - wp.ackRedis(ctx, stream, msg.ID) + wp.ackRedis(jobCtx, stream, msg.ID) } func (wp *WorkerPool) ackRedis(ctx context.Context, stream, msgID string) { @@ -660,9 +757,16 @@ func (wp *WorkerPool) sweepDueJobs(ctx context.Context) error { return nil } -func (wp *WorkerPool) claimNext(ctx context.Context) (*model.BackgroundJob, error) { +func (wp *WorkerPool) claimNext(lifecycleCtx, jobCtx context.Context) (*model.BackgroundJob, error) { + wp.claimMu.RLock() + defer wp.claimMu.RUnlock() + if lifecycleCtx != nil { + if err := lifecycleCtx.Err(); err != nil { + return nil, err + } + } var job model.BackgroundJob - err := wp.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + err := wp.db.WithContext(jobCtx).Transaction(func(tx *gorm.DB) error { query := tx.Where("status IN ? AND scheduled_at <= ?", []string{model.BackgroundJobStatusQueued, model.BackgroundJobStatusRetrying}, wp.now()) if len(wp.queues) > 0 { query = query.Where("queue IN ?", wp.queues) diff --git a/backend/internal/worker/worker_test.go b/backend/internal/worker/worker_test.go index 1fb4439e..7c1a045f 100644 --- a/backend/internal/worker/worker_test.go +++ b/backend/internal/worker/worker_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "path/filepath" + "sync" "sync/atomic" "testing" "time" @@ -277,6 +278,80 @@ func TestWorkerPoolStopPersistsCancelledJobRetry(t *testing.T) { } } +func TestWorkerPoolShutdownDrainsActiveJob(t *testing.T) { + db := newWorkerTestDB(t) + wp := NewWorkerPoolWithOptions(db, WithPollInterval(5*time.Millisecond)) + started := make(chan struct{}) + release := make(chan struct{}) + wp.Register("drain", func(ctx context.Context, job *model.BackgroundJob) error { + close(started) + select { + case <-release: + return nil + case <-ctx.Done(): + return ctx.Err() + } + }) + job, err := wp.Enqueue(context.Background(), "drain", nil) + require.NoError(t, err) + require.NoError(t, wp.Start()) + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("worker did not start job") + } + + done := make(chan error, 1) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + go func() { done <- wp.Shutdown(ctx) }() + select { + case err := <-done: + t.Fatalf("shutdown returned before active job drained: %v", err) + case <-time.After(30 * time.Millisecond): + } + close(release) + require.NoError(t, <-done) + require.Equal(t, model.BackgroundJobStatusCompleted, loadJob(t, db, job.ID).Status) +} + +func TestWorkerPoolShutdownDeadlineCancelsActiveJob(t *testing.T) { + db := newWorkerTestDB(t) + wp := NewWorkerPoolWithOptions(db, WithPollInterval(5*time.Millisecond), WithBackoff(func(int) time.Duration { return 0 })) + started := make(chan struct{}) + cancelled := make(chan struct{}) + release := make(chan struct{}) + wp.Register("deadline", func(ctx context.Context, job *model.BackgroundJob) error { + close(started) + <-ctx.Done() + close(cancelled) + <-release + return ctx.Err() + }) + job, err := wp.Enqueue(context.Background(), "deadline", nil) + require.NoError(t, err) + require.NoError(t, wp.Start()) + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("worker did not start job") + } + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + done := make(chan error, 1) + go func() { done <- wp.Shutdown(ctx) }() + <-cancelled + select { + case err := <-done: + t.Fatalf("shutdown returned before cancelled handler released: %v", err) + default: + } + close(release) + require.ErrorIs(t, <-done, context.DeadlineExceeded) + require.Equal(t, model.BackgroundJobStatusRetrying, loadJob(t, db, job.ID).Status) +} + // --- Redis Stream path tests --- func newMiniRedis(t *testing.T) (*miniredis.Miniredis, *redis.Client) { @@ -300,6 +375,53 @@ func newRedisWorkerPool(t *testing.T, db *gorm.DB, rdb redis.UniversalClient, op return wp } +func TestRedisMessageRejectedWhenShutdownFollowsBatchCheck(t *testing.T) { + db := newWorkerTestDB(t) + _, rdb := newMiniRedis(t) + wp := newRedisWorkerPool(t, db, rdb) + var handled atomic.Int32 + wp.Register("shutdown_race", func(context.Context, *model.BackgroundJob) error { + handled.Add(1) + return nil + }) + job, err := wp.Enqueue(context.Background(), "shutdown_race", nil) + require.NoError(t, err) + + lifecycleCtx, cancel := context.WithCancel(context.Background()) + require.NoError(t, lifecycleCtx.Err()) + cancel() // SIGTERM after the batch-level lifecycle check. + wp.processRedisMessage(lifecycleCtx, context.Background(), wp.streamKeyFor(job.Queue), redis.XMessage{ + ID: "1-0", Values: map[string]any{"job_id": job.ID}, + }) + + require.Zero(t, handled.Load()) + require.Equal(t, model.BackgroundJobStatusQueued, loadJob(t, db, job.ID).Status) +} + +func TestRedisClaimCanceledWhenShutdownRacesClaim(t *testing.T) { + db := newWorkerTestDB(t) + _, rdb := newMiniRedis(t) + wp := newRedisWorkerPool(t, db, rdb, WithBlockTimeout(10*time.Millisecond)) + require.NoError(t, wp.Start()) + + claimStarted := make(chan struct{}) + var once sync.Once + require.NoError(t, db.Callback().Update().Before("gorm:update").Register("test:block_claim", func(tx *gorm.DB) { + once.Do(func() { close(claimStarted) }) + <-tx.Statement.Context.Done() + })) + t.Cleanup(func() { db.Callback().Update().Remove("test:block_claim") }) + job, err := wp.Enqueue(context.Background(), "shutdown_claim_race", nil) + require.NoError(t, err) + <-claimStarted + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + require.ErrorIs(t, wp.Shutdown(ctx), context.DeadlineExceeded) + + require.Equal(t, model.BackgroundJobStatusQueued, loadJob(t, db, job.ID).Status) +} + // TestRedisEnqueueAndProcessEndToEnd verifies the full Redis path: Enqueue // XADDs to the stream, XREADGROUP picks it up, DB claim succeeds, handler // runs, and the job reaches "completed" status. diff --git a/backend/scripts/health_check.sh b/backend/scripts/health_check.sh old mode 100644 new mode 100755 index c8c8e216..4f1ad54a --- a/backend/scripts/health_check.sh +++ b/backend/scripts/health_check.sh @@ -1,175 +1,103 @@ #!/bin/bash -# GoChat Health Check Script -# Reference: Chatwoot health_check endpoint pattern -# Checks: HTTP server, database, Redis, worker queues, WebSocket hub -# -# Usage: -# ./scripts/health_check.sh [--full] [--json] [--timeout SECONDS] -# -# Exit codes: -# 0 = healthy -# 1 = degraded (some checks failed but core is OK) -# 2 = unhealthy (critical checks failed) +# Exit codes: 0 healthy, 1 degraded, 2 unhealthy/invalid invocation. -set -euo pipefail +set -uo pipefail -# Configuration GOCHAT_HOST="${GOCHAT_HOST:-localhost}" GOCHAT_PORT="${GOCHAT_PORT:-3000}" -GOCHAT_METRICS_PORT="${GOCHAT_METRICS_PORT:-9090}" +GOCHAT_METRICS_PORT="${GOCHAT_METRICS_PORT:-${GOCHAT_PORT}}" TIMEOUT="${TIMEOUT:-5}" FULL_CHECK=false JSON_OUTPUT=false +RESULTS=() +CRITICAL_FAIL=0 +DEGRADED_FAIL=0 + +for arg in "$@"; do + [[ "${arg}" == "--json" ]] && JSON_OUTPUT=true +done + +usage_error() { + if [[ "${JSON_OUTPUT}" == "true" ]]; then + printf '{"error":"%s","checks":[],"summary":{"critical_failures":1,"degraded_failures":0}}\n' "$1" + else + printf 'health_check: %s\n' "$1" >&2 + fi + exit 2 +} -# Parse arguments while [[ $# -gt 0 ]]; do - case "$1" in - --full) FULL_CHECK=true; shift ;; - --json) JSON_OUTPUT=true; shift ;; - --timeout) TIMEOUT="$2"; shift 2 ;; - --help) echo "Usage: $0 [--full] [--json] [--timeout SECONDS]"; exit 0 ;; - *) echo "Unknown option: $1"; exit 1 ;; - esac + case "$1" in + --full) FULL_CHECK=true; shift ;; + --json) JSON_OUTPUT=true; shift ;; + --timeout) + [[ $# -ge 2 && "$2" =~ ^[1-9][0-9]*$ ]] || usage_error "--timeout requires a positive integer" + TIMEOUT="$2" + shift 2 + ;; + --help) + if [[ "${JSON_OUTPUT}" == "true" ]]; then + printf '{"usage":"health_check.sh [--full] [--json] [--timeout SECONDS]"}\n' + else + printf 'Usage: %s [--full] [--json] [--timeout SECONDS]\n' "$0" + fi + exit 0 + ;; + *) usage_error "unknown option" ;; + esac done BASE_URL="http://${GOCHAT_HOST}:${GOCHAT_PORT}" METRICS_URL="http://${GOCHAT_HOST}:${GOCHAT_METRICS_PORT}" -# Result tracking -RESULTS=() -CRITICAL_FAIL=0 -DEGRADED_FAIL=0 - check_result() { - local name="$1" status="$2" detail="$3" - RESULTS+=("${name}|${status}|${detail}") - if [[ "$status" == "CRITICAL_FAIL" ]]; then - CRITICAL_FAIL=$((CRITICAL_FAIL + 1)) - elif [[ "$status" == "DEGRADED" ]]; then - DEGRADED_FAIL=$((DEGRADED_FAIL + 1)) - fi + local name="$1" status="$2" detail="$3" + RESULTS+=("${name}|${status}|${detail}") + [[ "${status}" == "CRITICAL_FAIL" ]] && CRITICAL_FAIL=$((CRITICAL_FAIL + 1)) + [[ "${status}" == "DEGRADED" ]] && DEGRADED_FAIL=$((DEGRADED_FAIL + 1)) } -# ---- Check 1: HTTP liveness endpoint ---- -check_liveness() { - local response - response=$(curl -sf --max-time "${TIMEOUT}" "${BASE_URL}/live" 2>&1) && { - check_result "liveness" "OK" "${response}" - } || { - check_result "liveness" "CRITICAL_FAIL" "HTTP /live endpoint unreachable" - } +check_url() { + local name="$1" severity="$2" url="$3" ok_detail="$4" fail_detail="$5" + if curl -fsS --max-time "${TIMEOUT}" -o /dev/null "${url}"; then + check_result "${name}" "OK" "${ok_detail}" + else + check_result "${name}" "${severity}" "${fail_detail}" + fi } -# ---- Check 2: HTTP readiness endpoint ---- -check_readiness() { - local response - response=$(curl -sf --max-time "${TIMEOUT}" "${BASE_URL}/ready" 2>&1) && { - check_result "readiness" "OK" "${response}" - } || { - check_result "readiness" "CRITICAL_FAIL" "HTTP /ready endpoint unreachable" - } -} +check_url "liveness" "CRITICAL_FAIL" "${BASE_URL}/live" "process is alive" "liveness endpoint unreachable" +check_url "readiness" "CRITICAL_FAIL" "${BASE_URL}/ready" "dependencies are ready" "readiness endpoint failed" -# ---- Check 3: Database connectivity ---- -check_database() { - if [[ "${FULL_CHECK}" == "true" ]]; then - local response - response=$(curl -sf --max-time "${TIMEOUT}" "${BASE_URL}/health?check=database" 2>&1) && { - check_result "database" "OK" "${response}" - } || { - check_result "database" "CRITICAL_FAIL" "Database connectivity failed" - } - else - check_result "database" "SKIPPED" "Not in full mode" - fi -} - -# ---- Check 4: Redis connectivity ---- -check_redis() { - if [[ "${FULL_CHECK}" == "true" ]]; then - local response - response=$(curl -sf --max-time "${TIMEOUT}" "${BASE_URL}/health?check=redis" 2>&1) && { - check_result "redis" "OK" "${response}" - } || { - check_result "redis" "DEGRADED" "Redis connectivity failed (non-critical)" - } - else - check_result "redis" "SKIPPED" "Not in full mode" - fi -} - -# ---- Check 5: Metrics endpoint ---- -check_metrics() { - local response - response=$(curl -sf --max-time "${TIMEOUT}" "${METRICS_URL}/metrics" 2>&1) && { - check_result "metrics" "OK" "Prometheus metrics endpoint responding" - } || { - check_result "metrics" "DEGRADED" "Metrics endpoint unreachable (non-critical)" - } -} - -# ---- Check 6: WebSocket hub (full mode) ---- -check_websocket() { - if [[ "${FULL_CHECK}" == "true" ]]; then - # WebSocket health check via HTTP status endpoint - local response - response=$(curl -sf --max-time "${TIMEOUT}" "${BASE_URL}/health?check=websocket" 2>&1) && { - check_result "websocket" "OK" "${response}" - } || { - check_result "websocket" "DEGRADED" "WebSocket hub unreachable" - } - else - check_result "websocket" "SKIPPED" "Not in full mode" - fi -} - -# ---- Run all checks ---- -check_liveness -check_readiness -check_database -check_redis -check_metrics -check_websocket - -# ---- Output results ---- -if [[ "${JSON_OUTPUT}" == "true" ]]; then - # JSON output for programmatic consumption - echo '{' - echo ' "checks": [' - for i in "${!RESULTS[@]}"; do - IFS='|' read -r name status detail <<< "${RESULTS[$i]}" - comma="" - [[ $i -gt 0 ]] && comma="," - echo " ${comma}{\"name\": \"${name}\", \"status\": \"${status}\", \"detail\": \"${detail}\"}" - done - echo ' ],' - echo ' "summary": {' - echo ' "critical_failures": ${CRITICAL_FAIL},' - echo ' "degraded_failures": ${DEGRADED_FAIL}' - echo ' }' - echo '}' +if [[ "${FULL_CHECK}" == "true" ]]; then + check_url "database" "CRITICAL_FAIL" "${BASE_URL}/health?check=database" "database is healthy" "database check failed" + check_url "redis" "CRITICAL_FAIL" "${BASE_URL}/health?check=redis" "redis is healthy" "redis check failed" else - # Human-readable output - echo "=== GoChat Health Check Report ===" - for result in "${RESULTS[@]}"; do - IFS='|' read -r name status detail <<< "${result}" - case "$status" in - OK) icon="✅" ;; - CRITICAL_FAIL) icon="❌" ;; - DEGRADED) icon="⚠️" ;; - SKIPPED) icon="⏭️" ;; - esac - echo " ${icon} ${name}: ${status} — ${detail}" - done - echo "" - echo "Critical failures: ${CRITICAL_FAIL} | Degraded: ${DEGRADED_FAIL}" + check_result "database" "SKIPPED" "not in full mode" + check_result "redis" "SKIPPED" "not in full mode" fi -# ---- Determine exit code ---- -if [[ ${CRITICAL_FAIL} -gt 0 ]]; then - exit 2 -elif [[ ${DEGRADED_FAIL} -gt 0 ]]; then - exit 1 +check_url "metrics" "DEGRADED" "${METRICS_URL}/metrics" "metrics endpoint is responding" "metrics endpoint unreachable" + +if [[ "${JSON_OUTPUT}" == "true" ]]; then + printf '{"checks":[' + for i in "${!RESULTS[@]}"; do + IFS='|' read -r name status detail <<< "${RESULTS[$i]}" + [[ $i -gt 0 ]] && printf ',' + printf '{"name":"%s","status":"%s","detail":"%s"}' "${name}" "${status}" "${detail}" + done + printf '],"summary":{"critical_failures":%d,"degraded_failures":%d}}\n' "${CRITICAL_FAIL}" "${DEGRADED_FAIL}" else - exit 0 -fi \ No newline at end of file + printf '=== GoChat Health Check Report ===\n' + for result in "${RESULTS[@]}"; do + IFS='|' read -r name status detail <<< "${result}" + printf ' %s: %s — %s\n' "${name}" "${status}" "${detail}" + done + printf 'Critical failures: %d | Degraded: %d\n' "${CRITICAL_FAIL}" "${DEGRADED_FAIL}" +fi + +if [[ ${CRITICAL_FAIL} -gt 0 ]]; then + exit 2 +elif [[ ${DEGRADED_FAIL} -gt 0 ]]; then + exit 1 +fi diff --git a/backend/scripts/health_check_test.go b/backend/scripts/health_check_test.go new file mode 100644 index 00000000..ddf70bc6 --- /dev/null +++ b/backend/scripts/health_check_test.go @@ -0,0 +1,76 @@ +package scripts + +import ( + "encoding/json" + "errors" + "net" + "net/http" + "net/http/httptest" + "net/url" + "os" + "os/exec" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestHealthCheckJSONExitCodes(t *testing.T) { + tests := []struct { + name string + status func(string) int + exitCode int + }{ + {name: "healthy", status: func(string) int { return http.StatusOK }, exitCode: 0}, + {name: "degraded", status: func(path string) int { + if path == "/metrics" { + return http.StatusServiceUnavailable + } + return http.StatusOK + }, exitCode: 1}, + {name: "unhealthy", status: func(path string) int { + if path == "/live" || path == "/ready" { + return http.StatusServiceUnavailable + } + return http.StatusOK + }, exitCode: 2}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(test.status(r.URL.Path)) + })) + defer server.Close() + output, exitCode := runHealthScript(t, server.URL, "--json", "--full") + require.Equal(t, test.exitCode, exitCode) + var payload map[string]any + require.NoError(t, json.Unmarshal(output, &payload), string(output)) + }) + } +} + +func TestHealthCheckJSONInvalidInvocation(t *testing.T) { + command := exec.Command("./health_check.sh", "--json", "--timeout") + output, err := command.Output() + var exitErr *exec.ExitError + require.True(t, errors.As(err, &exitErr)) + require.Equal(t, 2, exitErr.ExitCode()) + require.JSONEq(t, `{"error":"--timeout requires a positive integer","checks":[],"summary":{"critical_failures":1,"degraded_failures":0}}`, string(output)) +} + +func runHealthScript(t *testing.T, rawURL string, args ...string) ([]byte, int) { + t.Helper() + parsed, err := url.Parse(rawURL) + require.NoError(t, err) + host, port, err := net.SplitHostPort(parsed.Host) + require.NoError(t, err) + command := exec.Command("./health_check.sh", args...) + command.Env = append(os.Environ(), "GOCHAT_HOST="+host, "GOCHAT_PORT="+port, "GOCHAT_METRICS_PORT="+port) + output, err := command.Output() + if err == nil { + return output, 0 + } + var exitErr *exec.ExitError + require.True(t, errors.As(err, &exitErr), "health script failed to execute: %v", err) + return output, exitErr.ExitCode() +} diff --git a/deploy/docker/Dockerfile b/deploy/docker/Dockerfile index 43ed645c..d801a7d0 100644 --- a/deploy/docker/Dockerfile +++ b/deploy/docker/Dockerfile @@ -87,7 +87,7 @@ RUN mkdir -p /app/storage/uploads && chown -R gochat:gochat /app/storage USER gochat # Health check — references the /health endpoint we will add -HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 CMD curl -f http://localhost:3000/health || exit 1 +HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 CMD curl -f http://localhost:3000/ready || exit 1 EXPOSE 3000 diff --git a/deploy/docker/Dockerfile.dev b/deploy/docker/Dockerfile.dev index 2b9f536b..0f69777f 100644 --- a/deploy/docker/Dockerfile.dev +++ b/deploy/docker/Dockerfile.dev @@ -4,12 +4,13 @@ FROM golang:1.25.13-alpine # Install air (hot reload tool) and development tools -RUN go install github.com/air-verse/air@latest && apk add --no-cache git +RUN go install github.com/air-verse/air@v1.66.0 && apk add --no-cache git WORKDIR /app # Pre-download dependencies for faster rebuilds COPY backend/go.mod backend/go.sum ./ +COPY backend/internal/pgvector_stub/ ./internal/pgvector_stub/ RUN go mod download # Full source mount via volume in compose — no COPY needed here diff --git a/deploy/docker/docker-compose.dev.yml b/deploy/docker/docker-compose.dev.yml index 5d0ea065..7753e30f 100644 --- a/deploy/docker/docker-compose.dev.yml +++ b/deploy/docker/docker-compose.dev.yml @@ -48,6 +48,7 @@ services: context: ../.. dockerfile: deploy/docker/Dockerfile.dev container_name: gochat-app + stop_grace_period: 30s depends_on: postgres: condition: service_healthy @@ -70,12 +71,13 @@ services: context: ../.. dockerfile: deploy/docker/Dockerfile.dev container_name: gochat-worker + stop_grace_period: 30s depends_on: postgres: condition: service_healthy redis: condition: service_healthy - command: ["serve", "--worker-only"] + command: ["go", "run", "./cmd/gochat", "worker"] volumes: - ../../backend:/app env_file: ../../.env diff --git a/deploy/docker/docker-compose.prod.yml b/deploy/docker/docker-compose.prod.yml index fd7b56b5..972cb6ef 100644 --- a/deploy/docker/docker-compose.prod.yml +++ b/deploy/docker/docker-compose.prod.yml @@ -77,6 +77,7 @@ services: gochat: image: *gochat-image restart: always + stop_grace_period: ${GOCHAT_STOP_GRACE_PERIOD:-35s} depends_on: postgres: condition: service_healthy @@ -90,7 +91,7 @@ services: volumes: - gochat_storage:/app/storage healthcheck: - test: ["CMD", "curl", "-fsS", "http://127.0.0.1:3000/health"] + test: ["CMD", "curl", "-fsS", "http://127.0.0.1:3000/ready"] interval: 10s timeout: 5s start_period: 15s @@ -107,14 +108,17 @@ services: worker: image: *gochat-image restart: always + stop_grace_period: ${GOCHAT_STOP_GRACE_PERIOD:-35s} depends_on: + gochat: + condition: service_healthy postgres: condition: service_healthy redis: condition: service_healthy meilisearch: condition: service_healthy - command: ["serve", "--worker-only"] + command: ["worker"] environment: <<: *gochat-environment GOCHAT_DATABASE_RUN_MIGRATIONS: "false" diff --git a/deploy/docker/docker-compose.yml b/deploy/docker/docker-compose.yml index 487381f9..51e12f8e 100644 --- a/deploy/docker/docker-compose.yml +++ b/deploy/docker/docker-compose.yml @@ -48,6 +48,7 @@ services: context: ../.. dockerfile: deploy/docker/Dockerfile container_name: gochat-app + stop_grace_period: 30s depends_on: postgres: condition: service_healthy @@ -63,6 +64,27 @@ services: volumes: - ../../backend:/app:delegated + worker: + build: + context: ../.. + dockerfile: deploy/docker/Dockerfile + container_name: gochat-worker + stop_grace_period: 30s + depends_on: + postgres: + condition: service_healthy + redis: + condition: service_healthy + env_file: ../../.env + environment: + - GOCHAT_ENV=development + - GOCHAT_DATABASE_DSN=postgres://postgres:postgres@postgres:5432/gochat_dev?sslmode=disable + - GOCHAT_DATABASE_RUN_MIGRATIONS=false + - GOCHAT_REDIS_DSN=redis://redis:6379 + volumes: + - ../../backend:/app:delegated + command: ["worker"] + volumes: postgres_data: redis_data: diff --git a/deploy/quickstart/compose.yaml b/deploy/quickstart/compose.yaml index 6e0e00a1..9f9fd920 100644 --- a/deploy/quickstart/compose.yaml +++ b/deploy/quickstart/compose.yaml @@ -68,6 +68,7 @@ services: context: ../.. dockerfile: deploy/docker/Dockerfile restart: unless-stopped + stop_grace_period: ${GOCHAT_STOP_GRACE_PERIOD:-35s} depends_on: postgres: condition: service_healthy @@ -103,11 +104,33 @@ services: volumes: - gochat-storage:/app/storage healthcheck: - test: ["CMD", "curl", "-fsS", "http://127.0.0.1:3000/health"] + test: ["CMD", "curl", "-fsS", "http://127.0.0.1:3000/ready"] interval: 10s timeout: 5s retries: 30 + worker: + build: + context: ../.. + dockerfile: deploy/docker/Dockerfile + restart: unless-stopped + stop_grace_period: ${GOCHAT_STOP_GRACE_PERIOD:-35s} + depends_on: + gochat: + condition: service_healthy + postgres: + condition: service_healthy + redis: + condition: service_healthy + meilisearch: + condition: service_healthy + environment: + <<: *gochat-env + GOCHAT_DATABASE_RUN_MIGRATIONS: "false" + volumes: + - gochat-storage:/app/storage + command: ["worker"] + seed: build: context: ../..