diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ce658012..222851e2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -178,6 +178,9 @@ jobs: MEILI_MASTER_KEY: ci-meili-secret-16 GOCHAT_JWT_SECRET: ci-smoke-jwt-secret-at-least-32-characters GOCHAT_JWT_PREVIOUS_SECRETS: ci-previous-jwt-secret-at-least-32-characters + GOCHAT_BACKUP_OFFSITE_DIR: /mnt/gochat-offsite + GOCHAT_BACKUP_OFFSITE_SOURCE: backup.example.test:/gochat + GOCHAT_BACKUP_OFFSITE_FSTYPE: nfs4 steps: - uses: actions/checkout@v4 - name: Build production images from repository root @@ -200,16 +203,9 @@ jobs: test -s /app/frontend/dist/index.html' - name: Start production Compose and smoke core pages run: | - if MEILI_MASTER_KEY=too-short deploy/docker/preflight.sh; then - echo "preflight accepted a short Meilisearch key" >&2 - exit 1 - fi - if POSTGRES_IMAGE_REF=pgvector/pgvector:pg16 deploy/docker/preflight.sh; then - echo "preflight accepted a mutable PostgreSQL image" >&2 - exit 1 - fi - deploy/docker/preflight.sh + deploy/docker/preflight_test.sh 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/app" | grep -q '/assets/' diff --git a/backend/cmd/migrate/main.go b/backend/cmd/migrate/main.go index 24a1d75d..4e702407 100644 --- a/backend/cmd/migrate/main.go +++ b/backend/cmd/migrate/main.go @@ -43,6 +43,10 @@ func main() { fmt.Println("Migrations applied successfully (up)") case "down": + if err := rejectDestructiveProductionMigration(env, -1); err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } if err := database.RollbackMigrations(dbURL, migrationsPath); err != nil { fmt.Fprintf(os.Stderr, "Error running migrations down: %v\n", err) os.Exit(1) @@ -89,6 +93,10 @@ func main() { fmt.Fprintf(os.Stderr, "Error: invalid step count '%s': %v\n", os.Args[2], err) os.Exit(1) } + if err := rejectDestructiveProductionMigration(env, steps); err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } if err := database.MigrateSteps(dbURL, migrationsPath, steps); err != nil { fmt.Fprintf(os.Stderr, "Error running %d migration steps: %v\n", steps, err) os.Exit(1) @@ -102,6 +110,13 @@ func main() { } } +func rejectDestructiveProductionMigration(env string, steps int) error { + if env == "production" && steps < 0 { + return fmt.Errorf("destructive schema down is disabled in production; roll back the application image or restore a backup") + } + return nil +} + func printUsage() { fmt.Println("Usage: migrate [args]") fmt.Println("") diff --git a/backend/cmd/migrate/migrate_test.go b/backend/cmd/migrate/migrate_test.go index beed199f..b8710d87 100644 --- a/backend/cmd/migrate/migrate_test.go +++ b/backend/cmd/migrate/migrate_test.go @@ -218,3 +218,8 @@ func TestCurrentVersion(t *testing.T) { assert.Equal(t, uint(3), version) assert.False(t, dirty) } +func TestRejectDestructiveProductionMigration(t *testing.T) { + require.Error(t, rejectDestructiveProductionMigration("production", -1)) + require.NoError(t, rejectDestructiveProductionMigration("production", 1)) + require.NoError(t, rejectDestructiveProductionMigration("development", -1)) +} diff --git a/backend/configs/config.production.yaml b/backend/configs/config.production.yaml index a696f061..a96b1591 100644 --- a/backend/configs/config.production.yaml +++ b/backend/configs/config.production.yaml @@ -7,7 +7,7 @@ server: database: dsn: "postgres://gochat:CHANGE_ME@postgres:5432/gochat_production?sslmode=disable" - run_migrations: true + run_migrations: false migrations_path: "/app/migrations" redis: diff --git a/backend/internal/app/bootstrap.go b/backend/internal/app/bootstrap.go index f8b0a503..807742c2 100644 --- a/backend/internal/app/bootstrap.go +++ b/backend/internal/app/bootstrap.go @@ -73,6 +73,9 @@ func Bootstrap(env string) (*App, error) { if err != nil { return nil, fmt.Errorf("config load failed: %w", err) } + if err := validateStartupMigrations(env, cfg); err != nil { + return nil, err + } // Step 2: Validate configuration if err := config.Validate(cfg); err != nil { @@ -788,6 +791,10 @@ func Bootstrap(env string) (*App, error) { WithWidgetAuth(inboxRepo, contactInboxRepo). WithConversationRepo(conversationRepo). WithAccessDB(db) + service.RegisterUploadCleanupJobs(workerPool, uploadService) + if _, err := service.EnqueueUploadCleanup(context.Background(), workerPool, time.Now()); err != nil { + applogger.L().Warnf("failed to enqueue upload cleanup: %v", err) + } uploadHandler := v1.NewUploadHandler(uploadService).WithAccessAuth(jwtService) // Step 9: Wire handlers (HTTP presentation layer) @@ -1000,6 +1007,13 @@ func Bootstrap(env string) (*App, error) { }, nil } +func validateStartupMigrations(env string, cfg *config.Config) error { + if env == "production" && cfg.Database.RunMigrations { + return fmt.Errorf("database.run_migrations must be false in production; run the one-shot migration job before web/worker startup") + } + return nil +} + // hubTypingAdapter implements service.TypingIndicator by broadcasting // typing_on/typing_off events directly through the WS hub's SendToAccount method. // This avoids the Redis dependency required by the full ws.TypingTracker. diff --git a/backend/internal/app/startup_migrations_test.go b/backend/internal/app/startup_migrations_test.go new file mode 100644 index 00000000..4305949e --- /dev/null +++ b/backend/internal/app/startup_migrations_test.go @@ -0,0 +1,14 @@ +package app + +import ( + "testing" + + "github.com/gochat/gochat/internal/config" + "github.com/stretchr/testify/require" +) + +func TestValidateStartupMigrations(t *testing.T) { + cfg := &config.Config{Database: config.DatabaseConfig{RunMigrations: true}} + require.ErrorContains(t, validateStartupMigrations("production", cfg), "one-shot migration job") + require.NoError(t, validateStartupMigrations("development", cfg)) +} diff --git a/backend/internal/database/high_risk_migrations_test.go b/backend/internal/database/high_risk_migrations_test.go new file mode 100644 index 00000000..b1799198 --- /dev/null +++ b/backend/internal/database/high_risk_migrations_test.go @@ -0,0 +1,134 @@ +package database + +import ( + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gorm.io/driver/postgres" + "gorm.io/gorm" + "gorm.io/gorm/logger" +) + +func TestHighRiskMigrationsPreserveRollbackData(t *testing.T) { + if os.Getenv("GOCHAT_TEST_DB") == "sqlite" { + t.Skip("requires PostgreSQL migration semantics") + } + dsn := os.Getenv("GOCHAT_TEST_DB_URL") + if dsn == "" { + dsn = "host=localhost port=5432 user=postgres password=postgres dbname=gochat_test sslmode=disable" + } + db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)}) + require.NoError(t, err) + sqlDB, err := db.DB() + require.NoError(t, err) + sqlDB.SetMaxOpenConns(1) + t.Cleanup(func() { _ = sqlDB.Close() }) + + schema := "high_risk_migrations_" + time.Now().Format("20060102150405000000000") + require.NoError(t, db.Exec("CREATE SCHEMA "+schema).Error) + t.Cleanup(func() { _ = db.Exec("DROP SCHEMA " + schema + " CASCADE").Error }) + require.NoError(t, db.Exec("SET search_path TO "+schema).Error) + require.NoError(t, db.Exec(` + CREATE TABLE reporting_events_rollups ( + id SERIAL PRIMARY KEY, account_id INTEGER NOT NULL, dimension VARCHAR(50) NOT NULL, + dimension_value VARCHAR(255) NOT NULL, metric_name VARCHAR(50) NOT NULL, + value DOUBLE PRECISION NOT NULL, value_in_business_hours DOUBLE PRECISION, + period VARCHAR(50) NOT NULL, created_at TIMESTAMPTZ DEFAULT NOW(), + updated_at TIMESTAMPTZ DEFAULT NOW(), deleted_at TIMESTAMPTZ, + UNIQUE(account_id, dimension, dimension_value, metric_name, period) + ); + INSERT INTO reporting_events_rollups + (account_id, dimension, dimension_value, metric_name, value, value_in_business_hours, period) + VALUES (1, 'inbox', '42', 'conversations_count', 7, 3, '2026-08-21'); + CREATE TABLE custom_attribute_definitions (attribute_model TEXT, attribute_name TEXT); + CREATE TABLE conversations (id BIGINT PRIMARY KEY, custom_attributes JSONB, updated_at TIMESTAMPTZ); + INSERT INTO custom_attribute_definitions VALUES ('conversation_attribute', 'swt_source_url'); + INSERT INTO conversations VALUES (1, '{"swt_source_url":"https://example.test"}', NOW()); + `).Error) + + up48, err := os.ReadFile(filepath.Join("..", "..", "migrations", "000048_fix_reporting_events_rollups_schema.up.sql")) + require.NoError(t, err) + require.NoError(t, db.Exec(string(up48)).Error) + var rollup struct { + DimensionID int64 + Count int64 + SumValue float64 + } + require.NoError(t, db.Table("reporting_events_rollups").Select("dimension_id, count, sum_value").Scan(&rollup).Error) + assert.Equal(t, int64(42), rollup.DimensionID) + assert.Equal(t, int64(7), rollup.Count) + assert.Equal(t, float64(7), rollup.SumValue) + + up76, err := os.ReadFile(filepath.Join("..", "..", "migrations", "000076_replace_shangwutong_source_attributes_with_messages.up.sql")) + require.NoError(t, err) + require.NoError(t, db.Exec(string(up76)).Error) + var definitions, conversations int64 + require.NoError(t, db.Table("custom_attribute_definitions").Count(&definitions).Error) + require.NoError(t, db.Table("conversations").Where("custom_attributes->>'swt_source_url' IS NOT NULL").Count(&conversations).Error) + assert.Equal(t, int64(1), definitions) + assert.Equal(t, int64(1), conversations) + + down48, err := os.ReadFile(filepath.Join("..", "..", "migrations", "000048_fix_reporting_events_rollups_schema.down.sql")) + require.NoError(t, err) + require.ErrorContains(t, db.Exec(string(down48)).Error, "irreversible") +} + +func TestMigration48RejectsUnsafeLegacyData(t *testing.T) { + if os.Getenv("GOCHAT_TEST_DB") == "sqlite" { + t.Skip("requires PostgreSQL migration semantics") + } + dsn := os.Getenv("GOCHAT_TEST_DB_URL") + if dsn == "" { + dsn = "host=localhost port=5432 user=postgres password=postgres dbname=gochat_test sslmode=disable" + } + db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)}) + require.NoError(t, err) + sqlDB, err := db.DB() + require.NoError(t, err) + sqlDB.SetMaxOpenConns(1) + t.Cleanup(func() { _ = sqlDB.Close() }) + + up48, err := os.ReadFile(filepath.Join("..", "..", "migrations", "000048_fix_reporting_events_rollups_schema.up.sql")) + require.NoError(t, err) + tests := []struct { + name, values, want string + rows int64 + }{ + {"invalid_period", "(1, 'inbox', '42', 'conversations_count', 1, 'not-a-date')", "cannot convert to date", 1}, + {"invalid_dimension", "(1, 'inbox', 'vip', 'conversations_count', 1, '2026-08-21')", "cannot convert to bigint", 1}, + {"converted_key_collision", "(1, 'inbox', '042', 'conversations_count', 1, '2026-08-21'), (1, 'inbox', '42', 'conversations_count', 2, '2026-08-21')", "unique-key collision", 2}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + schema := "migration_48_rejection_" + tt.name + t.Cleanup(func() { _ = db.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE").Error }) + require.NoError(t, db.Exec("DROP SCHEMA IF EXISTS "+schema+" CASCADE").Error) + require.NoError(t, db.Exec("CREATE SCHEMA "+schema).Error) + require.NoError(t, db.Exec("SET search_path TO "+schema).Error) + require.NoError(t, db.Exec(` + CREATE TABLE reporting_events_rollups ( + id SERIAL PRIMARY KEY, account_id INTEGER NOT NULL, dimension VARCHAR(50) NOT NULL, + dimension_value VARCHAR(255) NOT NULL, metric_name VARCHAR(50) NOT NULL, + value DOUBLE PRECISION NOT NULL, value_in_business_hours DOUBLE PRECISION, + period VARCHAR(50) NOT NULL, created_at TIMESTAMPTZ DEFAULT NOW(), + updated_at TIMESTAMPTZ DEFAULT NOW(), deleted_at TIMESTAMPTZ, + UNIQUE(account_id, dimension, dimension_value, metric_name, period) + ); + INSERT INTO reporting_events_rollups + (account_id, dimension, dimension_value, metric_name, value, period) + VALUES `+tt.values).Error) + + require.ErrorContains(t, db.Exec(string(up48)).Error, tt.want) + require.NoError(t, db.Exec("ROLLBACK").Error) + var rows int64 + require.NoError(t, db.Table(schema+".reporting_events_rollups").Count(&rows).Error) + assert.Equal(t, tt.rows, rows) + require.NoError(t, db.Exec("SET search_path TO public").Error) + require.NoError(t, db.Exec("DROP SCHEMA "+schema+" CASCADE").Error) + }) + } +} diff --git a/backend/internal/repository/direct_upload_repo.go b/backend/internal/repository/direct_upload_repo.go index be8002c8..f6b3d8f0 100644 --- a/backend/internal/repository/direct_upload_repo.go +++ b/backend/internal/repository/direct_upload_repo.go @@ -64,6 +64,11 @@ func (r *DirectUploadRepo) UpdateStatus(ctx context.Context, id uint, status mod return r.db.WithContext(ctx).Model(&model.DirectUpload{}).Where("id = ?", id).Update("status", status).Error } +// Delete permanently removes a staged upload after its object is gone. +func (r *DirectUploadRepo) Delete(ctx context.Context, id uint) error { + return r.db.WithContext(ctx).Unscoped().Delete(&model.DirectUpload{}, id).Error +} + // FindExpired retrieves all direct uploads that have passed their expiry timestamp // and are still in pending status. func (r *DirectUploadRepo) FindExpired(ctx context.Context, before time.Time) ([]model.DirectUpload, error) { @@ -83,4 +88,4 @@ func (r *DirectUploadRepo) BatchDeleteExpired(ctx context.Context, before time.T Where("status = ? AND expires_at < ?", model.DirectUploadStatusPending, before). Delete(&model.DirectUpload{}) return result.RowsAffected, result.Error -} \ No newline at end of file +} diff --git a/backend/internal/service/upload_service.go b/backend/internal/service/upload_service.go index 668a211e..1ab54770 100644 --- a/backend/internal/service/upload_service.go +++ b/backend/internal/service/upload_service.go @@ -26,9 +26,15 @@ import ( "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/repository" "github.com/gochat/gochat/internal/security" + "github.com/gochat/gochat/internal/worker" applogger "github.com/gochat/gochat/pkg/logger" ) +const ( + TaskTypeCleanupExpiredUploads = "storage:cleanup_expired_uploads" + uploadCleanupInterval = 24 * time.Hour +) + // UploadService handles file uploads for both account-level and widget direct uploads. type UploadService struct { directUploadRepo *repository.DirectUploadRepo @@ -846,14 +852,139 @@ func randomStorageKey() string { return hex.EncodeToString(buf) } -// CleanupExpiredUploads removes expired direct upload records and their files. +// CleanupExpiredUploads removes each object before its record. Failed object +// deletions leave the row intact so the durable job can retry safely. func (s *UploadService) CleanupExpiredUploads(ctx context.Context) (int64, error) { - count, err := s.directUploadRepo.BatchDeleteExpired(ctx, time.Now()) + now := time.Now() + uploads, err := s.directUploadRepo.FindExpired(ctx, now) if err != nil { - return 0, fmt.Errorf("failed to cleanup expired uploads: %w", err) + return 0, fmt.Errorf("find expired uploads: %w", err) } - applogger.L().Infof("Cleaned up %d expired direct uploads", count) - return count, nil + var count int64 + var cleanupErr error + for i := range uploads { + if err := s.removeUploadFile(uploads[i].FileURL); err != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("remove upload %s: %w", uploads[i].UploadUUID, err)) + continue + } + if err := s.directUploadRepo.Delete(ctx, uploads[i].ID); err != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("delete upload %s: %w", uploads[i].UploadUUID, err)) + continue + } + count++ + } + orphans, err := s.cleanupOrphanedUploadFiles(ctx, now.Add(-uploadCleanupInterval)) + count += orphans + cleanupErr = errors.Join(cleanupErr, err) + applogger.L().Infof("Cleaned up %d expired or orphaned uploads", count) + return count, cleanupErr +} + +// cleanupOrphanedUploadFiles reconciles only roots owned by UploadService. +// The grace period keeps an in-flight file-to-row write from racing cleanup. +func (s *UploadService) cleanupOrphanedUploadFiles(ctx context.Context, cutoff time.Time) (int64, error) { + if s.accessDB == nil { + return 0, nil + } + references := map[string]struct{}{} + var urls []string + for _, column := range []string{"file_url", "thumb_url"} { + urls = nil + if err := s.accessDB.WithContext(ctx).Model(&model.DirectUpload{}). + Where(column+" LIKE ?", "/uploads/%").Pluck(column, &urls).Error; err != nil { + return 0, fmt.Errorf("load direct upload references: %w", err) + } + for _, url := range urls { + references[url] = struct{}{} + } + } + if s.accessDB.Migrator().HasTable(&model.User{}) { + urls = nil + if err := s.accessDB.WithContext(ctx).Model(&model.User{}). + Where("avatar_url LIKE ?", "/uploads/account/%").Pluck("avatar_url", &urls).Error; err != nil { + return 0, fmt.Errorf("load avatar references: %w", err) + } + for _, url := range urls { + references[url] = struct{}{} + } + } + + localPath := "./uploads" + if s.cfg != nil && s.cfg.Storage.LocalPath != "" { + localPath = s.cfg.Storage.LocalPath + } + var removed int64 + var reconcileErr error + for _, subdir := range []string{"account", "widget_direct"} { + root := filepath.Join(localPath, subdir) + err := filepath.Walk(root, func(path string, info os.FileInfo, walkErr error) error { + if walkErr != nil { + if errors.Is(walkErr, os.ErrNotExist) { + return nil + } + return walkErr + } + if !info.Mode().IsRegular() || info.ModTime().After(cutoff) { + return nil + } + relative, err := filepath.Rel(localPath, path) + if err != nil { + return err + } + fileURL := "/uploads/" + filepath.ToSlash(relative) + if _, ok := references[fileURL]; ok { + return nil + } + if err := os.Remove(path); err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil + } + return err + } + removed++ + return nil + }) + if err != nil && !errors.Is(err, os.ErrNotExist) { + reconcileErr = errors.Join(reconcileErr, fmt.Errorf("reconcile %s uploads: %w", subdir, err)) + } + } + return removed, reconcileErr +} + +func (s *UploadService) removeUploadFile(fileURL string) error { + relative := strings.TrimPrefix(fileURL, "/uploads/") + if relative == fileURL || relative == "" || filepath.IsAbs(relative) || strings.HasPrefix(filepath.Clean(relative), "..") { + return fmt.Errorf("invalid upload path %q", fileURL) + } + localPath := s.cfg.Storage.LocalPath + if localPath == "" { + localPath = "./uploads" + } + err := os.Remove(filepath.Join(localPath, relative)) + if errors.Is(err, os.ErrNotExist) { + return nil + } + return err +} + +// RegisterUploadCleanupJobs runs orphan cleanup daily through the existing +// durable, idempotent worker queue. +func RegisterUploadCleanupJobs(wp *worker.WorkerPool, svc *UploadService) { + wp.Register(TaskTypeCleanupExpiredUploads, func(ctx context.Context, _ *model.BackgroundJob) error { + _, cleanupErr := svc.CleanupExpiredUploads(ctx) + _, enqueueErr := EnqueueUploadCleanup(ctx, wp, time.Now().Add(uploadCleanupInterval)) + return errors.Join(cleanupErr, enqueueErr) + }) +} + +func EnqueueUploadCleanup(ctx context.Context, wp *worker.WorkerPool, at time.Time) (*model.BackgroundJob, error) { + bucket := at.UTC().Truncate(uploadCleanupInterval).Unix() + return wp.Enqueue(ctx, TaskTypeCleanupExpiredUploads, nil, + worker.WithQueue("low"), + worker.WithScheduledAt(at), + worker.WithMaxAttempts(10), + worker.WithIdempotencyKey(fmt.Sprintf("storage:cleanup_expired_uploads:%d", bucket)), + ) } // --- MIME detection helpers (reuse patterns from widget_theme_service.go) --- diff --git a/backend/internal/service/upload_service_test.go b/backend/internal/service/upload_service_test.go index 12293601..d1a88931 100644 --- a/backend/internal/service/upload_service_test.go +++ b/backend/internal/service/upload_service_test.go @@ -5,12 +5,14 @@ import ( "encoding/json" "mime/multipart" "net/http" + "net/http/httptest" "os" "path/filepath" "strings" "testing" "time" + "github.com/gin-gonic/gin" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "gorm.io/driver/sqlite" @@ -145,7 +147,8 @@ func setupUploadServiceTest(t *testing.T) (*gorm.DB, *UploadService, string) { directUploadRepo := repository.NewDirectUploadRepo(db) uploadService := NewUploadService(directUploadRepo, cfg). - WithWidgetAuth(repository.NewInboxRepo(db), repository.NewContactInboxRepo(db)) + WithWidgetAuth(repository.NewInboxRepo(db), repository.NewContactInboxRepo(db)). + WithAccessDB(db) return db, uploadService, tmpDir } @@ -195,6 +198,25 @@ func TestUploadService_AccountUpload_Success(t *testing.T) { assert.GreaterOrEqual(t, len(files), 1, "file should be saved on disk") } +func TestUploadSharedStorageSurvivesReplicaRecreation(t *testing.T) { + _, svc, storagePath := setupUploadServiceTest(t) + content := []byte("shared attachment") + upload, err := svc.AccountUpload(context.Background(), 1, AccountUploadRequest{ + FileHeader: createTestFileHeader(t, "shared.txt", content), + }) + require.NoError(t, err) + + for range 2 { + replica := gin.New() + replica.StaticFS("/uploads", gin.Dir(storagePath, false)) + response := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodGet, upload.FileURL, nil) + replica.ServeHTTP(response, request) + assert.Equal(t, http.StatusOK, response.Code) + assert.Equal(t, content, response.Body.Bytes()) + } +} + func TestUploadService_AccountUpload_NoAccountID(t *testing.T) { _, svc, _ := setupUploadServiceTest(t) @@ -459,7 +481,7 @@ func TestUploadService_PrivateAttachmentAccessIsTenantAndSessionScoped(t *testin } func TestUploadService_CleanupExpiredUploads(t *testing.T) { - db, svc, _ := setupUploadServiceTest(t) + db, svc, tmpDir := setupUploadServiceTest(t) // Insert a pending-but-expired upload record (pending status with past expires_at) // BatchDeleteExpired deletes pending uploads where expires_at < now @@ -476,6 +498,9 @@ func TestUploadService_CleanupExpiredUploads(t *testing.T) { ExpiresAt: time.Now().Add(-1 * time.Hour), // already past expiry } require.NoError(t, db.Create(expiredUpload).Error) + expiredPath := filepath.Join(tmpDir, "account", "1", "old_file.png") + require.NoError(t, os.MkdirAll(filepath.Dir(expiredPath), 0o755)) + require.NoError(t, os.WriteFile(expiredPath, []byte("expired"), 0o600)) // Insert a non-expired upload (pending status with future expires_at) activeUpload := &model.DirectUpload{ @@ -491,13 +516,55 @@ func TestUploadService_CleanupExpiredUploads(t *testing.T) { ExpiresAt: time.Now().Add(24 * time.Hour), } require.NoError(t, db.Create(activeUpload).Error) + failedUpload := &model.DirectUpload{ + UploadUUID: "failed-uuid-789", AccountID: 1, Status: model.DirectUploadStatusPending, + Source: model.DirectUploadSourceAccount, OriginalName: "invalid.png", FileType: "image", + MimeType: "image/png", FileSize: 100, FileURL: "https://example.test/invalid.png", + ExpiresAt: time.Now().Add(-time.Hour), + } + require.NoError(t, db.Create(failedUpload).Error) count, err := svc.CleanupExpiredUploads(context.Background()) - require.NoError(t, err) + require.ErrorContains(t, err, "invalid upload path") assert.Equal(t, int64(1), count, "should clean up 1 expired upload") + assert.NoFileExists(t, expiredPath) // Verify active upload still exists var remaining model.DirectUpload require.NoError(t, db.Where("upload_uuid = ?", "active-uuid-456").First(&remaining).Error) assert.Equal(t, model.DirectUploadStatusPending, remaining.Status) + remaining = model.DirectUpload{} + require.NoError(t, db.Where("upload_uuid = ?", "failed-uuid-789").First(&remaining).Error) +} + +func TestUploadService_ReconcilesCrashWindowOrphans(t *testing.T) { + db, svc, storagePath := setupUploadServiceTest(t) + dir := filepath.Join(storagePath, "account", "1") + require.NoError(t, os.MkdirAll(dir, 0o755)) + orphan := filepath.Join(dir, "crash-window.png") + referenced := filepath.Join(dir, "referenced.png") + fresh := filepath.Join(dir, "in-flight.png") + for _, path := range []string{orphan, referenced, fresh} { + require.NoError(t, os.WriteFile(path, testPNG, 0o600)) + } + old := time.Now().Add(-uploadCleanupInterval - time.Hour) + require.NoError(t, os.Chtimes(orphan, old, old)) + require.NoError(t, os.Chtimes(referenced, old, old)) + require.NoError(t, db.Create(&model.DirectUpload{ + UploadUUID: "referenced", AccountID: 1, Status: model.DirectUploadStatusPending, + Source: model.DirectUploadSourceAccount, OriginalName: "referenced.png", FileType: "image", + MimeType: "image/png", FileSize: int64(len(testPNG)), FileURL: "/uploads/account/1/referenced.png", + ExpiresAt: time.Now().Add(time.Hour), + }).Error) + + count, err := svc.CleanupExpiredUploads(context.Background()) + require.NoError(t, err) + assert.Equal(t, int64(1), count) + assert.NoFileExists(t, orphan) + assert.FileExists(t, referenced) + assert.FileExists(t, fresh) + + count, err = svc.CleanupExpiredUploads(context.Background()) + require.NoError(t, err) + assert.Zero(t, count) } diff --git a/backend/migrations/000048_fix_reporting_events_rollups_schema.down.sql b/backend/migrations/000048_fix_reporting_events_rollups_schema.down.sql index c8d7dd4a..36d19873 100644 --- a/backend/migrations/000048_fix_reporting_events_rollups_schema.down.sql +++ b/backend/migrations/000048_fix_reporting_events_rollups_schema.down.sql @@ -1,22 +1,5 @@ --- Revert reporting_events_rollups to the original (incorrect) init schema structure. --- WARNING: this restores the broken column names (dimension, dimension_value, metric_name, --- value, value_in_business_hours, period) that do not match the GORM model. - -DROP TABLE IF EXISTS reporting_events_rollups; - -CREATE TABLE IF NOT EXISTS reporting_events_rollups ( - id SERIAL PRIMARY KEY, - account_id INTEGER NOT NULL, - dimension VARCHAR(50) NOT NULL, - dimension_value VARCHAR(255) NOT NULL, - metric_name VARCHAR(50) NOT NULL, - value DOUBLE PRECISION NOT NULL, - value_in_business_hours DOUBLE PRECISION DEFAULT 0, - period VARCHAR(50) NOT NULL, - created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), - updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), - deleted_at TIMESTAMP WITH TIME ZONE, - UNIQUE(account_id, dimension, dimension_value, metric_name, period) -); - -CREATE INDEX idx_reporting_events_rollups_deleted_at ON reporting_events_rollups(deleted_at); +DO $$ +BEGIN + RAISE EXCEPTION 'migration 000048 is irreversible; restore the pre-upgrade backup instead of destroying rollup data'; +END +$$; diff --git a/backend/migrations/000048_fix_reporting_events_rollups_schema.up.sql b/backend/migrations/000048_fix_reporting_events_rollups_schema.up.sql index 4df18496..fc8967d4 100644 --- a/backend/migrations/000048_fix_reporting_events_rollups_schema.up.sql +++ b/backend/migrations/000048_fix_reporting_events_rollups_schema.up.sql @@ -1,18 +1,67 @@ --- Fix reporting_events_rollups table structure to match GORM model --- The init schema (000001) created columns: dimension, dimension_value, metric_name, --- value, value_in_business_hours, period — but the GORM model and all Go code use: --- date, dimension_type, dimension_id, metric, count, sum_value, sum_value_business_hours. --- This migration drops and recreates the table with the correct schema. --- The table is empty in practice (rollups are computed on-demand), so data loss is negligible. +-- Convert the legacy rollup schema without dropping its data. The explicit +-- transaction makes lock/statement timeout failures leave the old table intact. +BEGIN; +LOCK TABLE reporting_events_rollups IN ACCESS EXCLUSIVE MODE; -DROP TABLE IF EXISTS reporting_events_rollups; +DO $$ +DECLARE + invalid_periods BIGINT; + invalid_dimensions BIGINT; + invalid_counts BIGINT; + oversized_dimensions BIGINT; + collision_groups BIGINT; +BEGIN + SELECT count(*) INTO invalid_periods + FROM reporting_events_rollups + WHERE NOT pg_input_is_valid(period, 'date'); + IF invalid_periods > 0 THEN + RAISE EXCEPTION 'migration 000048: % rollup period value(s) cannot convert to date', invalid_periods + USING HINT = 'Repair or remove the reported legacy rows explicitly, then retry the migration.'; + END IF; -CREATE TABLE reporting_events_rollups ( + SELECT count(*) INTO invalid_dimensions + FROM reporting_events_rollups + WHERE NOT pg_input_is_valid(dimension_value, 'bigint'); + IF invalid_dimensions > 0 THEN + RAISE EXCEPTION 'migration 000048: % rollup dimension_value(s) cannot convert to bigint', invalid_dimensions + USING HINT = 'Map non-numeric legacy dimensions explicitly, then retry the migration.'; + END IF; + + SELECT count(*) INTO invalid_counts + FROM reporting_events_rollups + WHERE metric_name LIKE '%count%' AND NOT pg_input_is_valid(value::text, 'bigint'); + IF invalid_counts > 0 THEN + RAISE EXCEPTION 'migration 000048: % count value(s) cannot convert losslessly to bigint', invalid_counts + USING HINT = 'Repair fractional, non-finite, or out-of-range count values, then retry the migration.'; + END IF; + + SELECT count(*) INTO oversized_dimensions + FROM reporting_events_rollups + WHERE length(dimension) > 20; + IF oversized_dimensions > 0 THEN + RAISE EXCEPTION 'migration 000048: % dimension value(s) exceed the new 20-character limit', oversized_dimensions + USING HINT = 'Shorten or explicitly map oversized dimensions, then retry the migration.'; + END IF; + + SELECT count(*) INTO collision_groups + FROM ( + SELECT 1 + FROM reporting_events_rollups + GROUP BY account_id, period::date, dimension, dimension_value::BIGINT, metric_name + HAVING count(*) > 1 + ) collisions; + IF collision_groups > 0 THEN + RAISE EXCEPTION 'migration 000048: % unique-key collision group(s) appear after type conversion', collision_groups + USING HINT = 'Merge or choose one legacy row in each collision group explicitly, then retry the migration.'; + END IF; +END $$; + +CREATE TABLE reporting_events_rollups_v2 ( id SERIAL PRIMARY KEY, account_id INTEGER NOT NULL, date DATE NOT NULL, dimension_type VARCHAR(20) NOT NULL, - dimension_id INTEGER NOT NULL, + dimension_id BIGINT NOT NULL, metric VARCHAR(50) NOT NULL, count BIGINT NOT NULL DEFAULT 0, sum_value DOUBLE PRECISION NOT NULL DEFAULT 0, @@ -23,7 +72,36 @@ CREATE TABLE reporting_events_rollups ( UNIQUE(account_id, date, dimension_type, dimension_id, metric) ); +INSERT INTO reporting_events_rollups_v2 ( + id, account_id, date, dimension_type, dimension_id, metric, count, + sum_value, sum_value_business_hours, created_at, updated_at, deleted_at +) +SELECT + id, + account_id, + period::date, + dimension, + dimension_value::BIGINT, + metric_name, + CASE WHEN metric_name LIKE '%count%' THEN value::BIGINT ELSE 1 END, + value, + COALESCE(value_in_business_hours, 0), + created_at, + updated_at, + deleted_at +FROM reporting_events_rollups; + +SELECT setval( + pg_get_serial_sequence('reporting_events_rollups_v2', 'id'), + COALESCE((SELECT MAX(id) FROM reporting_events_rollups_v2), 1), + EXISTS (SELECT 1 FROM reporting_events_rollups_v2) +); + +DROP TABLE reporting_events_rollups; +ALTER TABLE reporting_events_rollups_v2 RENAME TO reporting_events_rollups; +ALTER INDEX reporting_events_rollups_v2_pkey RENAME TO reporting_events_rollups_pkey; CREATE INDEX idx_reporting_events_rollups_deleted_at ON reporting_events_rollups(deleted_at); CREATE INDEX idx_reporting_events_rollups_account_id ON reporting_events_rollups(account_id) WHERE deleted_at IS NULL; CREATE INDEX idx_reporting_events_rollups_date ON reporting_events_rollups(date) WHERE deleted_at IS NULL; CREATE INDEX idx_reporting_events_rollups_dimension ON reporting_events_rollups(dimension_type, dimension_id) WHERE deleted_at IS NULL; +COMMIT; diff --git a/backend/migrations/000076_replace_shangwutong_source_attributes_with_messages.up.sql b/backend/migrations/000076_replace_shangwutong_source_attributes_with_messages.up.sql index ecdd3720..11582e94 100644 --- a/backend/migrations/000076_replace_shangwutong_source_attributes_with_messages.up.sql +++ b/backend/migrations/000076_replace_shangwutong_source_attributes_with_messages.up.sql @@ -1,52 +1,4 @@ -DELETE FROM custom_attribute_definitions -WHERE attribute_model = 'conversation_attribute' - AND attribute_name IN ( - 'swt_source_url', - 'swt_source_search_term', - 'swt_source_purchase_term', - 'swt_source_keyword_id', - 'swt_source_channel', - 'swt_source_realtime_location', - 'swt_source_region', - 'swt_source_ad_account_id', - 'swt_source_wakeable', - 'swt_baidu_conversation_type', - 'swt_baidu_agent_name', - 'swt_baidu_ssid' - ); - -UPDATE conversations -SET - custom_attributes = COALESCE(custom_attributes, '{}'::jsonb) - - 'swt_source_url' - - 'swt_source_description' - - 'swt_source_search_term' - - 'swt_source_purchase_term' - - 'swt_source_keyword_id' - - 'swt_source_channel' - - 'swt_source_realtime_location' - - 'swt_source_region' - - 'swt_source_ad_account_id' - - 'swt_source_wakeable' - - 'swt_baidu_conversation_type' - - 'swt_baidu_agent_name' - - 'swt_baidu_ssid', - updated_at = CURRENT_TIMESTAMP -WHERE COALESCE(custom_attributes, '{}'::jsonb) ?| ARRAY[ - 'swt_source_url', - 'swt_source_description', - 'swt_source_search_term', - 'swt_source_purchase_term', - 'swt_source_keyword_id', - 'swt_source_channel', - 'swt_source_realtime_location', - 'swt_source_region', - 'swt_source_ad_account_id', - 'swt_source_wakeable', - 'swt_baidu_conversation_type', - 'swt_baidu_agent_name', - 'swt_baidu_ssid' -]; - --- Historical kind=8 system messages are backfilled from the Connector SQLite --- store because PostgreSQL intentionally does not retain the raw encoded payload. \ No newline at end of file +-- Keep the legacy source attributes during the expand/contract window. New +-- connector versions emit source information as system messages, while an old +-- application digest still needs these definitions and values after rollback. +SELECT 1; diff --git a/backend/scripts/db_backup.sh b/backend/scripts/db_backup.sh index 39d62948..ed360ee8 100755 --- a/backend/scripts/db_backup.sh +++ b/backend/scripts/db_backup.sh @@ -1,54 +1,62 @@ -#!/bin/bash -# GoChat Database Backup Script -# Reference: Chatwoot uses pg_dump for backup in production deployments -# Usage: ./scripts/db_backup.sh [env] +#!/usr/bin/env bash +# Create one encrypted, off-site copy containing PostgreSQL, attachments, and +# the Connector's online SQLite backup. set -euo pipefail +umask 077 -ENV="${1:-production}" -TIMESTAMP=$(date +%Y%m%d_%H%M%S) -BACKUP_DIR="${GOCHAT_BACKUP_DIR:-/var/backups/gochat}" -DSN="${GOCHAT_DATABASE_DSN:-}" -RETENTION_DAYS="${GOCHAT_BACKUP_RETENTION_DAYS:-30}" +dsn=${GOCHAT_DATABASE_DSN:?GOCHAT_DATABASE_DSN is required} +storage=${GOCHAT_STORAGE_PATH:?GOCHAT_STORAGE_PATH is required} +connector=${GOCHAT_CONNECTOR_BACKUP_FILE:?GOCHAT_CONNECTOR_BACKUP_FILE is required} +backup_dir=${GOCHAT_BACKUP_DIR:-/var/backups/gochat} +offsite_dir=${GOCHAT_BACKUP_OFFSITE_DIR:?GOCHAT_BACKUP_OFFSITE_DIR is required} +passphrase_file=${GOCHAT_BACKUP_PASSPHRASE_FILE:?GOCHAT_BACKUP_PASSPHRASE_FILE is required} +retention_days=${GOCHAT_BACKUP_RETENTION_DAYS:-30} +version=${GOCHAT_VERSION:-unknown} +timestamp=$(date -u +%Y%m%dT%H%M%SZ) -if [[ -z "$DSN" ]]; then - echo "[$(date)] ERROR: GOCHAT_DATABASE_DSN is not set" - exit 1 -fi - -# Parse DSN: postgres://user:password@host:port/dbname?sslmode=... -DB_USER=$(echo "$DSN" | sed 's|.*://||; s|:.*||') -DB_PASS=$(echo "$DSN" | sed 's|.*://[^:]*:||; s|@.*||') -DB_HOST=$(echo "$DSN" | sed 's|.*@||; s|:.*||') -DB_PORT=$(echo "$DSN" | sed 's|.*@.*:||; s|/.*||') -DB_NAME=$(echo "$DSN" | sed 's|.*/||; s|\?.*||') - -mkdir -p "${BACKUP_DIR}" - -BACKUP_FILE="${BACKUP_DIR}/${DB_NAME}_${TIMESTAMP}.sql.gz" - -echo "[$(date)] Starting backup of ${DB_NAME} on ${DB_HOST}:${DB_PORT}" - -# pg_dump with compression — mirrors Chatwoot backup approach -PGPASSWORD="${DB_PASS}" pg_dump \ - -h "${DB_HOST}" \ - -p "${DB_PORT}" \ - -U "${DB_USER}" \ - -d "${DB_NAME}" \ - --format=custom \ - --compress=9 \ - | gzip > "${BACKUP_FILE}" - -BACKUP_SIZE=$(du -h "${BACKUP_FILE}" | cut -f1) -echo "[$(date)] Backup complete: ${BACKUP_FILE} (${BACKUP_SIZE})" - -# Prune old backups beyond retention period -find "${BACKUP_DIR}" -name "*.sql.gz" -mtime +"${RETENTION_DAYS}" -delete -echo "[$(date)] Pruned backups older than ${RETENTION_DAYS} days" - -# Verify backup integrity -gunzip -t "${BACKUP_FILE}" && echo "[$(date)] Backup integrity verified" || { - echo "[$(date)] ERROR: Backup integrity check failed!" - rm -f "${BACKUP_FILE}" - exit 1 +[[ -d "$storage" ]] || { echo "storage path does not exist: $storage" >&2; exit 1; } +[[ -z "$(find "$storage" -type l -print -quit)" ]] || { echo "storage must not contain symbolic links" >&2; exit 1; } +[[ -f "$connector" ]] || { echo "connector backup does not exist: $connector" >&2; exit 1; } +[[ -r "$passphrase_file" ]] || { echo "backup passphrase file is not readable" >&2; exit 1; } +server_major=$(($(psql "$dsn" -Atqc "SHOW server_version_num") / 10000)) +client_major=$(pg_dump --version | awk '{print $NF}' | cut -d. -f1) +[[ "$client_major" == "$server_major" ]] || { + echo "pg_dump major $client_major must match PostgreSQL major $server_major" >&2 + exit 1 } + +install -d -m 0700 "$backup_dir" "$offsite_dir" +[[ "$(realpath "$backup_dir")" != "$(realpath "$offsite_dir")" ]] || { + echo "off-site directory must use a different path/failure domain" >&2 + exit 1 +} + +work_dir=$(mktemp -d "$backup_dir/.backup.XXXXXX") +trap 'rm -rf "$work_dir"' EXIT +stage=$work_dir/snapshot +mkdir "$stage" + +pg_dump --dbname "$dsn" --format=custom --compress=9 --file "$stage/postgres.dump" +pg_restore --list "$stage/postgres.dump" >/dev/null +tar -C "$storage" -cf "$stage/attachments.tar" . +cp "$connector" "$stage/connector.db" + +created_at_epoch=$(date -u +%s) +{ + echo "created_at=$timestamp" + echo "created_at_epoch=$created_at_epoch" + echo "version=$version" + echo "rpo_target_seconds=86400" +} >"$stage/manifest" +(cd "$stage" && sha256sum postgres.dump attachments.tar connector.db manifest >SHA256SUMS) + +bundle=$backup_dir/gochat-$timestamp.tar.enc +tar -C "$stage" -cf - . | openssl enc -aes-256-cbc -pbkdf2 -salt -pass "file:$passphrase_file" -out "$bundle" +(cd "$backup_dir" && sha256sum "$(basename "$bundle")" >"$(basename "$bundle").sha256") +openssl enc -d -aes-256-cbc -pbkdf2 -pass "file:$passphrase_file" -in "$bundle" | tar -tf - >/dev/null + +cp "$bundle" "$bundle.sha256" "$offsite_dir/" +find "$backup_dir" "$offsite_dir" -maxdepth 1 -type f -name 'gochat-*.tar.enc*' -mtime "+$retention_days" -delete + +echo "backup=$bundle offsite=$offsite_dir/$(basename "$bundle") version=$version created_at=$timestamp" diff --git a/backend/scripts/db_restore.sh b/backend/scripts/db_restore.sh new file mode 100755 index 00000000..d4f742be --- /dev/null +++ b/backend/scripts/db_restore.sh @@ -0,0 +1,70 @@ +#!/usr/bin/env bash +# Restore an encrypted GoChat backup into an empty database/storage target and +# print auditable RPO/RTO and business-integrity evidence. + +set -euo pipefail +umask 077 + +bundle=${1:?usage: db_restore.sh /path/to/gochat-*.tar.enc} +dsn=${GOCHAT_DATABASE_DSN:?GOCHAT_DATABASE_DSN is required} +storage=${GOCHAT_STORAGE_PATH:?GOCHAT_STORAGE_PATH is required} +connector=${GOCHAT_CONNECTOR_DB_PATH:?GOCHAT_CONNECTOR_DB_PATH is required} +passphrase_file=${GOCHAT_BACKUP_PASSPHRASE_FILE:?GOCHAT_BACKUP_PASSPHRASE_FILE is required} +started_at=$(date -u +%s) + +[[ -f "$bundle" ]] || { echo "backup bundle does not exist: $bundle" >&2; exit 1; } +[[ -f "$bundle.sha256" ]] || { echo "backup checksum does not exist: $bundle.sha256" >&2; exit 1; } +(cd "$(dirname "$bundle")" && sha256sum -c "$(basename "$bundle").sha256") + +server_major=$(($(psql "$dsn" -Atqc "SHOW server_version_num") / 10000)) +client_major=$(pg_restore --version | awk '{print $NF}' | cut -d. -f1) +[[ "$client_major" == "$server_major" ]] || { + echo "pg_restore major $client_major must match PostgreSQL major $server_major" >&2 + exit 1 +} + +table_count=$(psql "$dsn" -Atqc "SELECT count(*) FROM pg_tables WHERE schemaname = 'public'") +[[ "$table_count" == "0" ]] || { echo "target database is not empty" >&2; exit 1; } +install -d -m 0700 "$storage" "$(dirname "$connector")" +[[ -z "$(find "$storage" -mindepth 1 -maxdepth 1 -print -quit)" ]] || { echo "target storage is not empty" >&2; exit 1; } +[[ ! -e "$connector" ]] || { echo "target connector database already exists" >&2; exit 1; } + +work_dir=$(mktemp -d "$(dirname "$bundle")/.restore.XXXXXX") +trap 'rm -rf "$work_dir"' EXIT +archive=$work_dir/snapshot.tar +openssl enc -d -aes-256-cbc -pbkdf2 -pass "file:$passphrase_file" -in "$bundle" -out "$archive" +while IFS= read -r path; do + case "$path" in + /*|../*|*/../*|*/..) echo "unsafe archive path: $path" >&2; exit 1 ;; + esac +done < <(tar -tf "$archive") +if tar -tvf "$archive" | awk '{type = substr($1, 1, 1); if (type != "-" && type != "d") found = 1} END {exit(found ? 0 : 1)}'; then + echo "backup archive contains unsupported entry types" >&2 + exit 1 +fi +tar -C "$work_dir" -xf "$archive" +rm "$archive" +(cd "$work_dir" && sha256sum -c SHA256SUMS) +pg_restore --list "$work_dir/postgres.dump" >/dev/null + +pg_restore --exit-on-error --no-owner --no-privileges --dbname "$dsn" "$work_dir/postgres.dump" +while IFS= read -r path; do + case "$path" in + /*|../*|*/../*|*/..) echo "unsafe attachment path: $path" >&2; exit 1 ;; + esac +done < <(tar -tf "$work_dir/attachments.tar") +if tar -tvf "$work_dir/attachments.tar" | awk '{type = substr($1, 1, 1); if (type != "-" && type != "d") found = 1} END {exit(found ? 0 : 1)}'; then + echo "attachment archive contains unsupported entry types" >&2 + exit 1 +fi +tar -C "$storage" -xf "$work_dir/attachments.tar" +install -m 0600 "$work_dir/connector.db" "$connector" + +migration_version=$(psql "$dsn" -Atqc "SELECT COALESCE(MAX(version), 0) FROM schema_migrations WHERE NOT dirty") +accounts=$(psql "$dsn" -Atqc "SELECT count(*) FROM accounts WHERE deleted_at IS NULL") +attachments=$(psql "$dsn" -Atqc "SELECT count(*) FROM attachments WHERE deleted_at IS NULL") +backup_epoch=$(awk -F= '$1 == "created_at_epoch" {print $2}' "$work_dir/manifest") +version=$(awk -F= '$1 == "version" {print substr($0, index($0, "=") + 1)}' "$work_dir/manifest") +finished_at=$(date -u +%s) + +echo "restore=ok version=$version migration=$migration_version accounts=$accounts attachments=$attachments rpo_seconds=$((started_at-backup_epoch)) rto_seconds=$((finished_at-started_at))" diff --git a/deploy/docker/Dockerfile b/deploy/docker/Dockerfile index 4642100f..43ed645c 100644 --- a/deploy/docker/Dockerfile +++ b/deploy/docker/Dockerfile @@ -50,6 +50,9 @@ RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 \ -X github.com/gochat/gochat/internal/config.BuildDate=${BUILD_DATE}" \ -o /gochat-worker ./cmd/gochat/ +RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 \ + go build -ldflags="-s -w" -o /migrate ./cmd/migrate/ + # ========== Production Stage ========== FROM alpine:3.21 @@ -60,17 +63,20 @@ LABEL org.opencontainers.image.version="${VERSION}" \ org.opencontainers.image.revision="${COMMIT_SHA}" \ org.opencontainers.image.created="${BUILD_DATE}" -# Install runtime dependencies -RUN apk --no-cache add ca-certificates tzdata curl && addgroup -S gochat && adduser -S gochat -G gochat +# Install runtime and recovery dependencies. +RUN apk --no-cache add bash ca-certificates tzdata curl openssl postgresql16-client && \ + addgroup -S gochat && adduser -S gochat -G gochat WORKDIR /app # Copy binary and configs from builder COPY --from=builder /gochat /app/gochat COPY --from=builder /gochat-worker /app/gochat-worker +COPY --from=builder /migrate /app/migrate COPY --chown=gochat:gochat --from=frontend-builder /app/frontend/dist /app/frontend/dist COPY backend/configs/ /app/configs/ COPY backend/migrations/ /app/migrations/ +COPY backend/scripts/db_backup.sh backend/scripts/db_restore.sh /app/scripts/ ENV GOCHAT_FRONTEND_DIST=/app/frontend/dist diff --git a/deploy/docker/docker-compose.prod.yml b/deploy/docker/docker-compose.prod.yml index 2094f25f..fd7b56b5 100644 --- a/deploy/docker/docker-compose.prod.yml +++ b/deploy/docker/docker-compose.prod.yml @@ -8,7 +8,7 @@ x-gochat-environment: &gochat-environment GOCHAT_SERVER_MODE: release GOCHAT_SERVER_CORS_ALLOWED_ORIGINS: ${GOCHAT_SERVER_CORS_ALLOWED_ORIGINS:?set production CORS origins} GOCHAT_DATABASE_DSN: ${GOCHAT_DATABASE_DSN:-postgres://${POSTGRES_USER:-gochat}:${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD}@postgres:5432/${POSTGRES_DB:-gochat_production}?sslmode=disable} - GOCHAT_DATABASE_RUN_MIGRATIONS: "true" + GOCHAT_DATABASE_RUN_MIGRATIONS: "false" GOCHAT_DATABASE_MIGRATIONS_PATH: /app/migrations GOCHAT_REDIS_DSN: redis://:${REDIS_PASSWORD:?set REDIS_PASSWORD}@redis:6379 GOCHAT_SEARCH_ENGINE: meilisearch @@ -126,6 +126,75 @@ services: memory: 512M cpus: "1.0" + migrate: + image: *gochat-image + profiles: ["ops"] + depends_on: + postgres: + condition: service_healthy + entrypoint: ["/app/migrate"] + command: ["up"] + restart: "no" + environment: + <<: *gochat-environment + GOCHAT_DATABASE_DSN: postgres://${POSTGRES_USER:-gochat}:${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD}@postgres:5432/${POSTGRES_DB:-gochat_production}?sslmode=disable&lock_timeout=5000&statement_timeout=900000 + GOCHAT_DATABASE_RUN_MIGRATIONS: "false" + + backup: + image: *gochat-image + user: "0:0" + profiles: ["ops"] + depends_on: + postgres: + condition: service_healthy + entrypoint: ["/app/scripts/db_backup.sh"] + restart: "no" + environment: + GOCHAT_DATABASE_DSN: postgres://${POSTGRES_USER:-gochat}:${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD}@postgres:5432/${POSTGRES_DB:-gochat_production}?sslmode=disable + GOCHAT_STORAGE_PATH: /source/storage/uploads + GOCHAT_CONNECTOR_BACKUP_FILE: /source/connector/${GOCHAT_CONNECTOR_BACKUP_NAME:-latest.db} + GOCHAT_BACKUP_DIR: /backup/local + GOCHAT_BACKUP_OFFSITE_DIR: /backup/offsite + GOCHAT_BACKUP_PASSPHRASE_FILE: /run/secrets/backup-passphrase + GOCHAT_BACKUP_RETENTION_DAYS: ${GOCHAT_BACKUP_RETENTION_DAYS:-30} + GOCHAT_VERSION: ${GOCHAT_IMAGE_REF} + volumes: + - gochat_storage:/source/storage:ro + - shangwutong_backups:/source/connector:ro + - ${GOCHAT_BACKUP_DIR:-./backups/local}:/backup/local + - type: bind + source: ${GOCHAT_BACKUP_OFFSITE_DIR:?set an existing external off-site mount point} + target: /backup/offsite + bind: + create_host_path: false + - ${GOCHAT_BACKUP_PASSPHRASE_FILE:-./.secrets/backup-passphrase}:/run/secrets/backup-passphrase:ro + + restore: + image: *gochat-image + user: "0:0" + profiles: ["ops"] + depends_on: + postgres: + condition: service_healthy + entrypoint: ["/app/scripts/db_restore.sh"] + command: ["/backup/offsite/${GOCHAT_RESTORE_BUNDLE:-missing.tar.enc}"] + restart: "no" + environment: + GOCHAT_DATABASE_DSN: postgres://${POSTGRES_USER:-gochat}:${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD}@postgres:5432/${POSTGRES_DB:-gochat_production}?sslmode=disable + GOCHAT_STORAGE_PATH: /restore/storage/uploads + GOCHAT_CONNECTOR_DB_PATH: /restore/connector/connector.db + GOCHAT_BACKUP_PASSPHRASE_FILE: /run/secrets/backup-passphrase + volumes: + - gochat_storage:/restore/storage + - shangwutong_data:/restore/connector + - type: bind + source: ${GOCHAT_BACKUP_OFFSITE_DIR:?set an existing external off-site mount point} + target: /backup/offsite + read_only: true + bind: + create_host_path: false + - ${GOCHAT_BACKUP_PASSPHRASE_FILE:-./.secrets/backup-passphrase}:/run/secrets/backup-passphrase:ro + shangwutong: image: ${SHANGWUTONG_IMAGE_REF:?set SHANGWUTONG_IMAGE_REF to an immutable image digest} restart: always diff --git a/deploy/docker/preflight.sh b/deploy/docker/preflight.sh index 11fceb74..0c278cee 100755 --- a/deploy/docker/preflight.sh +++ b/deploy/docker/preflight.sh @@ -12,7 +12,7 @@ if (($#)); then compose_args=(--env-file "$env_file" "${compose_args[@]}") fi -required=(GOCHAT_IMAGE_REF SHANGWUTONG_IMAGE_REF GOCHAT_SERVER_CORS_ALLOWED_ORIGINS POSTGRES_PASSWORD REDIS_PASSWORD MEILI_MASTER_KEY GOCHAT_JWT_SECRET) +required=(GOCHAT_IMAGE_REF SHANGWUTONG_IMAGE_REF GOCHAT_SERVER_CORS_ALLOWED_ORIGINS POSTGRES_PASSWORD REDIS_PASSWORD MEILI_MASTER_KEY GOCHAT_JWT_SECRET GOCHAT_BACKUP_OFFSITE_DIR GOCHAT_BACKUP_OFFSITE_SOURCE GOCHAT_BACKUP_OFFSITE_FSTYPE) for name in "${required[@]}"; do value=${!name:-} if [[ -z $value || ${value^^} == *CHANGE_ME* ]]; then @@ -21,6 +21,44 @@ for name in "${required[@]}"; do fi done +if [[ $GOCHAT_BACKUP_OFFSITE_DIR != /* ]] || [[ ! -d $GOCHAT_BACKUP_OFFSITE_DIR ]]; then + echo "GOCHAT_BACKUP_OFFSITE_DIR must be an existing external mount point" >&2 + exit 1 +fi +offsite_dir=$(realpath -e -- "$GOCHAT_BACKUP_OFFSITE_DIR") +if [[ $offsite_dir == / ]]; then + echo "GOCHAT_BACKUP_OFFSITE_DIR must not be /" >&2 + exit 1 +fi +if ! offsite_info=$(findmnt -M "$offsite_dir" -n -o SOURCE,FSTYPE,MAJ:MIN); then + echo "GOCHAT_BACKUP_OFFSITE_DIR must be an existing external mount point" >&2 + exit 1 +fi +read -r offsite_source offsite_type offsite_device <<< "$offsite_info" +if [[ $offsite_source != "$GOCHAT_BACKUP_OFFSITE_SOURCE" || $offsite_type != "$GOCHAT_BACKUP_OFFSITE_FSTYPE" ]]; then + echo "GOCHAT_BACKUP_OFFSITE_DIR mount source/type does not match the approved values" >&2 + exit 1 +fi +case $offsite_type in + tmpfs | devtmpfs | ramfs) + echo "GOCHAT_BACKUP_OFFSITE_DIR must not use an in-memory filesystem" >&2 + exit 1 + ;; +esac + +local_probe=${GOCHAT_BACKUP_DIR:-./backups/local} +if [[ $local_probe != /* ]]; then + local_probe=$script_dir/$local_probe +fi +while [[ ! -e $local_probe && $local_probe != / ]]; do + local_probe=$(dirname -- "$local_probe") +done +read -r local_source local_device < <(findmnt -T "$local_probe" -n -o SOURCE,MAJ:MIN) +if [[ $offsite_source == "$local_source" || $offsite_device == "$local_device" ]]; then + echo "GOCHAT_BACKUP_OFFSITE_DIR must use a different source/device than GOCHAT_BACKUP_DIR" >&2 + exit 1 +fi + if ((${#GOCHAT_JWT_SECRET} < 32)); then echo "GOCHAT_JWT_SECRET must be at least 32 characters" >&2 exit 1 diff --git a/deploy/docker/preflight_test.sh b/deploy/docker/preflight_test.sh new file mode 100755 index 00000000..812bd93c --- /dev/null +++ b/deploy/docker/preflight_test.sh @@ -0,0 +1,96 @@ +#!/usr/bin/env bash +set -euo pipefail + +root=$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd) +tmp=$(mktemp -d) +trap 'rm -rf "$tmp"' EXIT +mkdir -p "$tmp/bin" "$tmp/local" "$tmp/offsite" + +cat > "$tmp/bin/findmnt" <<'EOF' +#!/usr/bin/env bash +if [[ $1 == -M && $2 == "$GOCHAT_BACKUP_OFFSITE_DIR" ]]; then + printf '%s %s %s\n' "$TEST_OFFSITE_SOURCE" "$TEST_OFFSITE_FSTYPE" "$TEST_OFFSITE_DEVICE" +elif [[ $1 == -T ]]; then + printf '%s %s\n' "$TEST_LOCAL_SOURCE" "$TEST_LOCAL_DEVICE" +else + exit 1 +fi +EOF +cat > "$tmp/bin/docker" <<'EOF' +#!/usr/bin/env bash +if [[ -n ${TEST_MUTABLE_IMAGE:-} ]]; then + echo 'pgvector/pgvector:pg16' +else + printf '%s\n' \ + 'gochat@example.invalid/gochat@sha256:0000000000000000000000000000000000000000000000000000000000000000' \ + 'gochat@example.invalid/connector@sha256:1111111111111111111111111111111111111111111111111111111111111111' +fi +EOF +chmod +x "$tmp/bin/findmnt" "$tmp/bin/docker" + +export PATH="$tmp/bin:$PATH" +export GOCHAT_IMAGE_REF='gochat@example.invalid/gochat@sha256:0000000000000000000000000000000000000000000000000000000000000000' +export SHANGWUTONG_IMAGE_REF='gochat@example.invalid/connector@sha256:1111111111111111111111111111111111111111111111111111111111111111' +export GOCHAT_SERVER_CORS_ALLOWED_ORIGINS=https://chat.example.test +export POSTGRES_PASSWORD=ci-postgres-secret +export REDIS_PASSWORD=ci-redis-secret +export MEILI_MASTER_KEY=ci-meili-secret-16 +export GOCHAT_JWT_SECRET=ci-smoke-jwt-secret-at-least-32-characters +export GOCHAT_BACKUP_DIR="$tmp/local" +export GOCHAT_BACKUP_OFFSITE_DIR="$tmp/offsite" +export GOCHAT_BACKUP_OFFSITE_SOURCE='backup.example.test:/gochat' +export GOCHAT_BACKUP_OFFSITE_FSTYPE=nfs4 +export TEST_OFFSITE_SOURCE=$GOCHAT_BACKUP_OFFSITE_SOURCE +export TEST_OFFSITE_FSTYPE=$GOCHAT_BACKUP_OFFSITE_FSTYPE +export TEST_OFFSITE_DEVICE=0:42 +export TEST_LOCAL_SOURCE=/dev/sda1 +export TEST_LOCAL_DEVICE=8:1 + +run_preflight() { + "$root/deploy/docker/preflight.sh" > "$tmp/output" 2>&1 +} + +expect_failure() { + local name=$1 expected=$2 + if run_preflight; then + echo "preflight accepted $name" >&2 + exit 1 + fi + grep -F "$expected" "$tmp/output" >/dev/null +} + +GOCHAT_BACKUP_OFFSITE_DIR=/ +expect_failure 'the root filesystem' 'must not be /' + +GOCHAT_BACKUP_OFFSITE_DIR=/tmp +TEST_OFFSITE_SOURCE=tmpfs +TEST_OFFSITE_FSTYPE=tmpfs +GOCHAT_BACKUP_OFFSITE_SOURCE=tmpfs +GOCHAT_BACKUP_OFFSITE_FSTYPE=tmpfs +expect_failure 'tmpfs' 'must not use an in-memory filesystem' + +GOCHAT_BACKUP_OFFSITE_DIR=$tmp/offsite +TEST_OFFSITE_SOURCE='backup.example.test:/gochat' +TEST_OFFSITE_FSTYPE=nfs4 +GOCHAT_BACKUP_OFFSITE_SOURCE=$TEST_OFFSITE_SOURCE +GOCHAT_BACKUP_OFFSITE_FSTYPE=$TEST_OFFSITE_FSTYPE +TEST_OFFSITE_DEVICE=$TEST_LOCAL_DEVICE +expect_failure 'the local backup device' 'must use a different source/device' + +TEST_OFFSITE_DEVICE=0:42 +GOCHAT_BACKUP_OFFSITE_SOURCE='other.example.test:/gochat' +expect_failure 'an unapproved mount source' 'does not match the approved values' + +GOCHAT_BACKUP_OFFSITE_SOURCE=$TEST_OFFSITE_SOURCE +MEILI_MASTER_KEY=too-short +expect_failure 'a short Meilisearch key' 'must be at least 16 bytes' +MEILI_MASTER_KEY=ci-meili-secret-16 + +TEST_MUTABLE_IMAGE=1 +export TEST_MUTABLE_IMAGE +expect_failure 'a mutable production image' 'must be pinned to a sha256 digest' +unset TEST_MUTABLE_IMAGE + +run_preflight +grep -F 'production preflight passed' "$tmp/output" >/dev/null +echo 'preflight tests passed' diff --git a/docs/ops/01-rolling-upgrade.md b/docs/ops/01-rolling-upgrade.md index 757b6347..18ef5a11 100644 --- a/docs/ops/01-rolling-upgrade.md +++ b/docs/ops/01-rolling-upgrade.md @@ -1,65 +1,148 @@ -# GoChat Rolling Upgrade Strategy -# Reference: Chatwoot deployment uses zero-downtime upgrade pattern +# Production backup, restore, and upgrade runbook -## Overview +Production web and worker processes never migrate on startup. The Compose file +also requires `GOCHAT_IMAGE_REF` to be an immutable `image@sha256:digest`; tags such +as `latest` are not an acceptable rollback record. -GoChat follows a blue-green deployment strategy for production upgrades, -ensuring zero downtime during version transitions. +The bundled `gochat_storage` volume is shared and durable for replicas on one +Docker host. Multi-host replicas must provision that volume with a shared +volume driver/filesystem; never use separate node-local volumes. -## Upgrade Process +## Recovery objectives -### Step 1: Pre-flight Checks -1. Verify new Docker image is built and pushed: `docker pull gochat/gochat:${NEW_VERSION}` -2. Run database migrations on a staging environment first -3. Verify backward compatibility of migrations (new code must work with old schema) -4. Check feature flags — new features should be disabled by default +- RPO: 24 hours. Run the encrypted backup at least daily and alert if the latest + off-site bundle is older than 24 hours. +- RTO: 4 hours. Rehearse a clean-environment restore quarterly and record the + script's `rpo_seconds`, `rto_seconds`, image digest, migration version, account + count, and attachment count. +- Local and off-site backup directories must be different mounts/failure + domains. The bundle is AES-256 encrypted; keep the passphrase file in the + secret manager, never beside either backup copy. + +## Daily encrypted backup + +Set these operator-owned paths before any Compose command: -### Step 2: Database Migration ```bash -# Run migrations BEFORE deploying new code -# Migrations must be backward-compatible -docker compose -f docker-compose.prod.yml exec gochat /app/gochat migrate up +export GOCHAT_IMAGE_REF='ghcr.io/gochat/gochat@sha256:' +export GOCHAT_BACKUP_DIR='/mnt/backup-local/gochat' +export GOCHAT_BACKUP_OFFSITE_DIR='/mnt/gochat-offsite' +export GOCHAT_BACKUP_OFFSITE_SOURCE='backup.example.com:/gochat' +export GOCHAT_BACKUP_OFFSITE_FSTYPE='nfs4' +export GOCHAT_BACKUP_PASSPHRASE_FILE='/run/secrets/gochat-backup-passphrase' +export GOCHAT_CONNECTOR_BACKUP_NAME="connector-$(date -u +%Y%m%dT%H%M%SZ).db" ``` -### Step 3: Blue-Green Deployment (Docker Compose) +Backup/restore are audited one-shot containers and run as root only to read or +rebuild Docker volumes; web and worker remain non-root. +`GOCHAT_BACKUP_OFFSITE_DIR` must be an existing external mount point provisioned +outside this Compose project. Set its approved source and filesystem type to the +exact values reported by `findmnt -M "$GOCHAT_BACKUP_OFFSITE_DIR"`; the preflight +rejects `/`, in-memory filesystems, unapproved mount metadata, and the local +backup device. + +Create a consistent online Connector backup, then create and verify the single +encrypted bundle containing PostgreSQL, attachments, and that Connector copy: + ```bash -# 1. Deploy new version as "green" alongside "blue" (current) -docker compose -f docker-compose.prod.yml up -d --no-deps gochat-green - -# 2. Wait for health check to pass -curl -f http://gochat-green:3000/health - -# 3. Switch traffic (update nginx/upstream config) -# nginx: switch upstream from blue to green - -# 4. Drain old connections on blue -# Wait 30s for in-flight requests to complete - -# 5. Stop blue -docker compose -f docker-compose.prod.yml stop gochat +docker compose -f deploy/docker/docker-compose.prod.yml exec shangwutong \ + shangwutong backup --output "/backup/$GOCHAT_CONNECTOR_BACKUP_NAME" +docker compose -f deploy/docker/docker-compose.prod.yml --profile ops run --rm backup ``` -### Step 4: Verification -1. Smoke test: hit /health endpoint -2. Check logs for errors: `docker compose logs gochat --since 5m` -3. Verify metrics: Prometheus dashboard should show normal traffic -4. Monitor for 15 minutes before finalizing +Schedule those two commands daily. Preserve their final `backup=... offsite=...` +line as audit evidence. A failed dump, archive checksum, decrypt/list check, or +off-site copy exits non-zero and must alert. + +## Clean-environment restore rehearsal + +Use an isolated host/project with empty volumes and an empty PostgreSQL database. +Do not point this procedure at the live project. -### Step 5: Rollback (if needed) ```bash -# Docker Compose rollback -docker compose -f docker-compose.prod.yml exec gochat /app/gochat migrate down ${N} -docker compose -f docker-compose.prod.yml up -d --no-deps gochat-${OLD_VERSION} +export COMPOSE_PROJECT_NAME=gochat-restore-$(date +%Y%m%d) +export GOCHAT_RESTORE_BUNDLE='gochat-.tar.enc' +docker compose -f deploy/docker/docker-compose.prod.yml --profile ops run --rm restore ``` -## Migration Compatibility Rules +The restore refuses a non-empty database, attachment directory, or Connector DB. +It verifies the encrypted bundle and internal checksums before restoring, then +prints the RPO/RTO, image version, migration version, accounts, and attachments. +Afterward start web/worker with the recorded digest and verify `/health`, one +attachment URL, and Connector `readyz`. -- Migrations MUST be additive only in production (add columns, never remove) -- Column removals require a 2-phase migration: soft-remove then hard-remove -- New columns should have defaults or be nullable -- Renames require a 3-phase migration: add new → copy data → remove old +## Upgrade preflight -## Worker Upgrade +1. Record the running digest as `OLD_IMAGE`; pull and record `NEW_IMAGE` by + digest. Never derive rollback state from a mutable tag. +2. Complete the backup above and restore it in the isolated environment. +3. Stop writes or enter the maintenance window. Migration 000079 takes + `SHARE ROW EXCLUSIVE` locks; migration 000048 takes `ACCESS EXCLUSIVE` on the + rollup table. The migration job uses a 5-second lock timeout and 15-minute + statement timeout, so contention fails instead of waiting indefinitely. +4. Capture these pre-migration checks: -Workers drain naturally: set a shutdown deadline, let in-flight jobs finish, -then stop. New workers pick up queued jobs from Redis. +```sql +SELECT count(*) AS rollups FROM reporting_events_rollups; +SELECT count(*) AS swt_source_rows FROM conversations + WHERE COALESCE(custom_attributes, '{}'::jsonb) ?| ARRAY['swt_source_url','swt_source_search_term']; +SELECT account_id, (config->>'assistant_id')::bigint, count(*) + FROM agent_bots + WHERE bot_type = 'captain' AND config->>'assistant_id' ~ '^[0-9]+$' + GROUP BY 1, 2 HAVING count(*) > 1; +``` + +While migrating, observe lock waits with: + +```sql +SELECT pid, wait_event_type, wait_event, clock_timestamp() - query_start AS elapsed, query +FROM pg_stat_activity +WHERE datname = current_database() AND state <> 'idle'; +``` + +## One-shot migration and application rollout + +Run exactly one named migration service before changing web/worker. Its output is +the audit log; `golang-migrate` is idempotent and PostgreSQL serializes competing +migration runners, but the deployment pipeline must contain only this one step. + +```bash +export GOCHAT_IMAGE_REF="$NEW_IMAGE" +time docker compose -f deploy/docker/docker-compose.prod.yml --profile ops \ + up --abort-on-container-exit --exit-code-from migrate migrate +docker compose -f deploy/docker/docker-compose.prod.yml logs migrate +docker compose -f deploy/docker/docker-compose.prod.yml up -d --no-deps gochat worker +curl -fsS http://127.0.0.1:3000/health +``` + +Post-migration checks: + +```sql +SELECT version, dirty FROM schema_migrations; +SELECT count(*) AS rollups FROM reporting_events_rollups; +SELECT count(*) AS duplicate_bot_bindings FROM ( + SELECT account_id, captain_assistant_id FROM agent_bots + WHERE captain_assistant_id IS NOT NULL GROUP BY 1, 2 HAVING count(*) > 1 +) duplicates; +SELECT count(*) AS orphaned_inbox_bindings FROM agent_bot_inboxes b +LEFT JOIN agent_bots a ON a.id = b.agent_bot_id WHERE a.id IS NULL; +``` + +Migration 000048 preserves every legacy rollup row, 000076 retains legacy +Connector attributes for old-image compatibility, and 000079 repairs references +before deduplication. Any count mismatch or dirty version stops the rollout. + +## Rollback + +Rollback the application only, using the recorded digest: + +```bash +export GOCHAT_IMAGE_REF="$OLD_IMAGE" +docker compose -f deploy/docker/docker-compose.prod.yml up -d --no-deps gochat worker +curl -fsS http://127.0.0.1:3000/health +``` + +Production `migrate down` and negative `migrate steps` are blocked. Never run a +destructive schema down during an application rollback. If the new schema itself +is unusable, stop web/worker and restore the verified pre-upgrade bundle into a +clean database and volumes.