HH-441: harden durable storage and recovery (#92)

* HH-441: harden durable storage and recovery

* HH-441: clear recovery review blockers

* HH-441: enforce offsite backup failure domain

---------

Co-authored-by: Rogee <rogee@ipao.vip>

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 <command> [args]")
 	fmt.Println("")

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))
+}

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:

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.

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))
+}

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)
+		})
+	}
+}

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
-}
+}

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) ---

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)
 }

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
+$$;

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;

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.
+-- 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;

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"

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))"