diff --git a/server/cmd/scheduler/main.go b/server/cmd/scheduler/main.go index e7ed5bc..b67b8d8 100644 --- a/server/cmd/scheduler/main.go +++ b/server/cmd/scheduler/main.go @@ -124,6 +124,8 @@ func main() { WithCache(app.Cache) monitoringRetentionWorker := internalscheduler.NewMonitoringRetentionWorker(app.MonitoringDB, app.Logger) brandAssetCleanupWorker := internalscheduler.NewBrandAssetCleanupWorker(app.DB, app.MonitoringDB, app.Cache, app.Logger) + brandAssetCleanupEventWorker := internalscheduler.NewBrandAssetCleanupEventWorker(brandAssetCleanupWorker, app.DB, app.Logger) + brandAssetCleanupReconcileWorker := internalscheduler.NewBrandAssetCleanupReconcileWorker(app.DB, app.Logger) aiPointUsageCleanupWorker := internalscheduler.NewAIPointUsageCleanupWorker(app.DB, app.Logger) schedulerRunRetentionWorker := internalscheduler.NewSchedulerRunRetentionWorker(app.DB, app.Logger) @@ -188,6 +190,12 @@ func main() { return brandAssetCleanupWorker.RunOnce(runCtx, jobCtx) }, }, + { + Key: "brand_asset_cleanup_reconcile", + Run: func(runCtx context.Context, jobCtx internalscheduler.JobRunContext) (map[string]any, error) { + return brandAssetCleanupReconcileWorker.RunOnce(runCtx, jobCtx) + }, + }, { Key: "ai_point_usage_cleanup", Run: func(runCtx context.Context, jobCtx internalscheduler.JobRunContext) (map[string]any, error) { @@ -215,6 +223,7 @@ func main() { }, }) startSchedulerWorker(ctx, &workerWG, "control_plane", app.Logger.Sugar(), controlPlane.Run) + startSchedulerWorker(ctx, &workerWG, "brand_asset_cleanup_events", app.Logger.Sugar(), brandAssetCleanupEventWorker.Run) var metricsServer *http.Server if app.Config().Scheduler.HTTPPort > 0 { diff --git a/server/internal/scheduler/brand_asset_cleanup_event_worker.go b/server/internal/scheduler/brand_asset_cleanup_event_worker.go new file mode 100644 index 0000000..825d168 --- /dev/null +++ b/server/internal/scheduler/brand_asset_cleanup_event_worker.go @@ -0,0 +1,231 @@ +package scheduler + +import ( + "context" + "database/sql" + "errors" + "fmt" + "os" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "go.uber.org/zap" + + tenantapp "github.com/geo-platform/tenant-api/internal/tenant/app" +) + +const ( + defaultBrandAssetCleanupEventWorkerBatchSize = defaultBrandAssetCleanupBatch + defaultBrandAssetCleanupEventWorkerMaxWait = time.Hour + defaultBrandAssetCleanupEventWorkerListenRetry = 5 * time.Second + defaultBrandAssetCleanupEventWorkerProcessTimeout = 2 * time.Minute +) + +type BrandAssetCleanupEventWorker struct { + worker *BrandAssetCleanupWorker + pool *pgxpool.Pool + logger *zap.Logger + workerID string + batchSize int + maxWait time.Duration + runTimeout time.Duration + staleAfter time.Duration + jobRunConfig map[string]any +} + +func NewBrandAssetCleanupEventWorker(worker *BrandAssetCleanupWorker, pool *pgxpool.Pool, logger *zap.Logger) *BrandAssetCleanupEventWorker { + hostname, _ := os.Hostname() + if hostname == "" { + hostname = "unknown" + } + return &BrandAssetCleanupEventWorker{ + worker: worker, + pool: pool, + logger: logger, + workerID: fmt.Sprintf("%s:%d:%d", hostname, os.Getpid(), time.Now().UnixNano()), + batchSize: defaultBrandAssetCleanupEventWorkerBatchSize, + maxWait: defaultBrandAssetCleanupEventWorkerMaxWait, + runTimeout: defaultBrandAssetCleanupEventWorkerProcessTimeout, + staleAfter: defaultBrandAssetCleanupEventStaleAfter, + jobRunConfig: map[string]any{ + "batch_size": defaultBrandAssetCleanupEventWorkerBatchSize, + "statement_timeout": "8s", + "event_stale_after": defaultBrandAssetCleanupEventStaleAfter.String(), + "worker_id": fmt.Sprintf("brand_asset_cleanup:%s:%d", hostname, os.Getpid()), + }, + } +} + +func (w *BrandAssetCleanupEventWorker) Run(ctx context.Context) { + if w == nil || w.worker == nil || w.pool == nil { + return + } + + wake := make(chan struct{}, 1) + go w.listen(ctx, wake) + signalBrandAssetCleanup(wake) + + for { + processed, err := w.processDue(ctx) + if err != nil && ctx.Err() == nil && w.logger != nil { + w.logger.Warn("brand asset cleanup event worker failed", zap.Error(err)) + } + if processed >= w.batchSize && processed > 0 { + continue + } + + waitFor := w.nextWait(ctx) + timer := time.NewTimer(waitFor) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-wake: + timer.Stop() + case <-timer.C: + } + } +} + +func (w *BrandAssetCleanupEventWorker) listen(ctx context.Context, wake chan<- struct{}) { + for { + if err := w.listenOnce(ctx, wake); err != nil && ctx.Err() == nil && w.logger != nil { + w.logger.Warn("brand asset cleanup listener stopped", zap.Error(err)) + } + signalBrandAssetCleanup(wake) + + select { + case <-ctx.Done(): + return + case <-time.After(defaultBrandAssetCleanupEventWorkerListenRetry): + signalBrandAssetCleanup(wake) + } + } +} + +func (w *BrandAssetCleanupEventWorker) listenOnce(ctx context.Context, wake chan<- struct{}) error { + conn, err := w.pool.Acquire(ctx) + if err != nil { + return err + } + defer conn.Release() + + if _, err := conn.Exec(ctx, fmt.Sprintf("LISTEN %s", tenantapp.BrandAssetCleanupNotifyChannel)); err != nil { + return err + } + defer func() { + unlistenCtx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _, _ = conn.Exec(unlistenCtx, fmt.Sprintf("UNLISTEN %s", tenantapp.BrandAssetCleanupNotifyChannel)) + }() + + for { + if _, err := conn.Conn().WaitForNotification(ctx); err != nil { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return nil + } + return err + } + signalBrandAssetCleanup(wake) + } +} + +func (w *BrandAssetCleanupEventWorker) processDue(parent context.Context) (int, error) { + timeout := w.runTimeout + if timeout <= 0 { + timeout = defaultBrandAssetCleanupEventWorkerProcessTimeout + } + ctx, cancel := context.WithTimeout(parent, timeout) + defer cancel() + + stats, err := w.worker.RunOnce(ctx, JobRunContext{ + Trigger: "event", + Config: w.jobRunConfig, + }) + if err != nil { + return 0, err + } + return intFromStats(stats, "event_count"), nil +} + +func (w *BrandAssetCleanupEventWorker) nextWait(ctx context.Context) time.Duration { + waitFor := w.maxWait + if waitFor <= 0 { + waitFor = defaultBrandAssetCleanupEventWorkerMaxWait + } + nextAt, err := nextBrandAssetCleanupDueAt(ctx, w.pool, w.staleAfter) + if err != nil { + if w.logger != nil { + w.logger.Warn("brand asset cleanup next due query failed", zap.Error(err)) + } + return minBrandCleanupDuration(waitFor, time.Minute) + } + if nextAt == nil { + return waitFor + } + until := time.Until(*nextAt) + if until <= 0 { + return time.Millisecond + } + return minBrandCleanupDuration(until, waitFor) +} + +func nextBrandAssetCleanupDueAt(ctx context.Context, pool *pgxpool.Pool, staleAfter time.Duration) (*time.Time, error) { + if pool == nil { + return nil, nil + } + if staleAfter <= 0 { + staleAfter = defaultBrandAssetCleanupEventStaleAfter + } + var due sql.NullTime + err := pool.QueryRow(ctx, ` + SELECT MIN(due_at) + FROM ( + SELECT next_attempt_at AS due_at + FROM brand_asset_cleanup_events + WHERE status IN ('pending', 'failed') + UNION ALL + SELECT locked_at + $1::interval AS due_at + FROM brand_asset_cleanup_events + WHERE status = 'processing' + AND locked_at IS NOT NULL + ) due_events + `, intervalParam(staleAfter)).Scan(&due) + if err != nil { + return nil, err + } + if !due.Valid { + return nil, nil + } + return &due.Time, nil +} + +func intFromStats(stats map[string]any, key string) int { + if stats == nil { + return 0 + } + switch value := stats[key].(type) { + case int: + return value + case int64: + return int(value) + case float64: + return int(value) + default: + return 0 + } +} + +func signalBrandAssetCleanup(ch chan<- struct{}) { + select { + case ch <- struct{}{}: + default: + } +} + +func minBrandCleanupDuration(a, b time.Duration) time.Duration { + if a < b { + return a + } + return b +} diff --git a/server/internal/scheduler/brand_asset_cleanup_reconcile_worker.go b/server/internal/scheduler/brand_asset_cleanup_reconcile_worker.go new file mode 100644 index 0000000..b6fc4ea --- /dev/null +++ b/server/internal/scheduler/brand_asset_cleanup_reconcile_worker.go @@ -0,0 +1,152 @@ +package scheduler + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "go.uber.org/zap" + + tenantapp "github.com/geo-platform/tenant-api/internal/tenant/app" +) + +const defaultBrandAssetCleanupReconcileBatch = 100 + +type BrandAssetCleanupReconcileWorker struct { + pool *pgxpool.Pool + logger *zap.Logger +} + +func NewBrandAssetCleanupReconcileWorker(pool *pgxpool.Pool, logger *zap.Logger) *BrandAssetCleanupReconcileWorker { + return &BrandAssetCleanupReconcileWorker{pool: pool, logger: logger} +} + +func (w *BrandAssetCleanupReconcileWorker) RunOnce(ctx context.Context, run JobRunContext) (map[string]any, error) { + stats := map[string]any{} + if w == nil || w.pool == nil { + return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil + } + + startedAt := time.Now() + batchSize := AsInt(run.Config, "batch_size", defaultBrandAssetCleanupReconcileBatch) + if run.Job != nil && run.Job.BatchSize != nil { + batchSize = *run.Job.BatchSize + } + if batchSize <= 0 { + batchSize = defaultBrandAssetCleanupReconcileBatch + } + statementTimeout := AsDuration(run.Config, "statement_timeout", 5*time.Second) + eventStaleAfter := AsDuration(run.Config, "event_stale_after", defaultBrandAssetCleanupEventStaleAfter) + dryRun := run.DryRun + stats["batch_size"] = batchSize + stats["dry_run"] = dryRun + + conn, err := w.pool.Acquire(ctx) + if err != nil { + return stats, fmt.Errorf("acquire tenant connection: %w", err) + } + defer conn.Release() + + if statementTimeout > 0 { + if _, err := conn.Exec(ctx, `SELECT set_config('statement_timeout', $1, false)`, fmt.Sprintf("%dms", statementTimeout.Milliseconds())); err != nil { + return stats, fmt.Errorf("set tenant statement timeout: %w", err) + } + defer conn.Exec(context.Background(), `RESET statement_timeout`) + } + + candidates, err := w.loadCandidates(ctx, conn, batchSize) + if err != nil { + return stats, fmt.Errorf("load brand cleanup reconcile candidates: %w", err) + } + runnableEventCount, err := w.countRunnableEvents(ctx, conn, eventStaleAfter) + if err != nil { + return stats, fmt.Errorf("count runnable brand cleanup events: %w", err) + } + + enqueued := int64(0) + if !dryRun && (len(candidates) > 0 || runnableEventCount > 0) { + tx, err := conn.Begin(ctx) + if err != nil { + return stats, fmt.Errorf("begin brand cleanup reconcile tx: %w", err) + } + committed := false + defer func() { + if !committed { + _ = tx.Rollback(context.Background()) + } + }() + + for _, candidate := range candidates { + if ctx.Err() != nil { + return stats, ErrorFromContext(ctx) + } + if _, err := tenantapp.EnqueueBrandAssetCleanupEvent(ctx, tx, candidate.TenantID, candidate.BrandID); err != nil { + return stats, fmt.Errorf("enqueue brand cleanup event tenant=%d brand=%d: %w", candidate.TenantID, candidate.BrandID, err) + } + enqueued++ + } + if err := tenantapp.NotifyBrandAssetCleanup(ctx, tx); err != nil { + return stats, fmt.Errorf("notify brand cleanup: %w", err) + } + if err := tx.Commit(ctx); err != nil { + return stats, fmt.Errorf("commit brand cleanup reconcile tx: %w", err) + } + committed = true + } + + stats["candidate_count"] = len(candidates) + stats["runnable_event_count"] = runnableEventCount + stats["enqueued_event_count"] = enqueued + stats["duration_ms"] = time.Since(startedAt).Milliseconds() + return stats, nil +} + +func (w *BrandAssetCleanupReconcileWorker) loadCandidates(ctx context.Context, conn *pgxpool.Conn, limit int) ([]brandCleanupCandidate, error) { + rows, err := conn.Query(ctx, ` + SELECT tenant_id, id + FROM brands + WHERE deleted_at IS NOT NULL + AND status = 'deleting' + AND NOT EXISTS ( + SELECT 1 + FROM brand_asset_cleanup_events e + WHERE e.tenant_id = brands.tenant_id + AND e.brand_id = brands.id + AND e.status IN ('pending', 'processing', 'failed') + ) + ORDER BY deleted_at ASC, id ASC + LIMIT $1 + `, limit) + if err != nil { + return nil, err + } + defer rows.Close() + + items := make([]brandCleanupCandidate, 0, limit) + for rows.Next() { + var item brandCleanupCandidate + if err := rows.Scan(&item.TenantID, &item.BrandID); err != nil { + return nil, err + } + items = append(items, item) + } + return items, rows.Err() +} + +func (w *BrandAssetCleanupReconcileWorker) countRunnableEvents(ctx context.Context, conn *pgxpool.Conn, staleAfter time.Duration) (int64, error) { + var count int64 + err := conn.QueryRow(ctx, ` + SELECT COUNT(*)::bigint + FROM brand_asset_cleanup_events + WHERE ( + status IN ('pending', 'failed') + AND next_attempt_at <= NOW() + ) + OR ( + status = 'processing' + AND locked_at < NOW() - $1::interval + ) + `, intervalParam(staleAfter)).Scan(&count) + return count, err +} diff --git a/server/internal/scheduler/brand_asset_cleanup_worker.go b/server/internal/scheduler/brand_asset_cleanup_worker.go index 17c7f3a..483bbb1 100644 --- a/server/internal/scheduler/brand_asset_cleanup_worker.go +++ b/server/internal/scheduler/brand_asset_cleanup_worker.go @@ -14,6 +14,7 @@ import ( ) const defaultBrandAssetCleanupBatch = 20 +const defaultBrandAssetCleanupEventStaleAfter = 10 * time.Minute type BrandAssetCleanupWorker struct { pool *pgxpool.Pool @@ -27,6 +28,13 @@ type brandCleanupCandidate struct { BrandID int64 } +type brandCleanupEvent struct { + EventID int64 + TenantID int64 + BrandID int64 + Attempts int +} + type brandCleanupStats struct { BrandCount int64 ArticleCount int64 @@ -96,6 +104,7 @@ func (w *BrandAssetCleanupWorker) RunOnce(ctx context.Context, run JobRunContext batchSize = defaultBrandAssetCleanupBatch } statementTimeout := AsDuration(run.Config, "statement_timeout", 5*time.Second) + eventStaleAfter := AsDuration(run.Config, "event_stale_after", defaultBrandAssetCleanupEventStaleAfter) dryRun := run.DryRun stats["batch_size"] = batchSize @@ -114,13 +123,17 @@ func (w *BrandAssetCleanupWorker) RunOnce(ctx context.Context, run JobRunContext defer conn.Exec(context.Background(), `RESET statement_timeout`) } - candidates, err := w.claimCandidates(ctx, conn, batchSize) + events, err := w.loadCleanupEvents(ctx, conn, run, batchSize, eventStaleAfter, dryRun) if err != nil { - return stats, fmt.Errorf("claim brand cleanup candidates: %w", err) + return stats, fmt.Errorf("claim brand cleanup events: %w", err) } total := brandCleanupStats{} failures := int64(0) - for _, candidate := range candidates { + for _, event := range events { + candidate := brandCleanupCandidate{ + TenantID: event.TenantID, + BrandID: event.BrandID, + } if dryRun { brandStats, err := w.probeBrand(ctx, conn, candidate) if err != nil { @@ -136,38 +149,64 @@ func (w *BrandAssetCleanupWorker) RunOnce(ctx context.Context, run JobRunContext if err != nil { failures++ w.warn("brand asset cleanup failed", candidate, err) + if markErr := w.markCleanupEventFailed(context.Background(), conn, event, err); markErr != nil && w.logger != nil { + w.logger.Warn("brand asset cleanup event failure mark failed", + zap.Int64("event_id", event.EventID), + zap.Int64("tenant_id", candidate.TenantID), + zap.Int64("brand_id", candidate.BrandID), + zap.Error(markErr), + ) + } + continue + } + if err := w.markCleanupEventSuccess(context.Background(), conn, event.EventID); err != nil { + failures++ + w.warn("brand asset cleanup event success mark failed", candidate, err) continue } total.add(brandStats) tenantapp.InvalidateBrandDeletionCaches(context.Background(), w.cache, candidate.TenantID, candidate.BrandID) } - stats["candidate_count"] = len(candidates) + stats["event_count"] = len(events) + stats["candidate_count"] = len(events) stats["failure_count"] = failures total.writeTo(stats) stats["duration_ms"] = time.Since(startedAt).Milliseconds() return stats, nil } -func (w *BrandAssetCleanupWorker) claimCandidates(ctx context.Context, conn *pgxpool.Conn, limit int) ([]brandCleanupCandidate, error) { +func (w *BrandAssetCleanupWorker) loadCleanupEvents(ctx context.Context, conn *pgxpool.Conn, run JobRunContext, limit int, staleAfter time.Duration, dryRun bool) ([]brandCleanupEvent, error) { + if dryRun { + return w.probeCleanupEvents(ctx, conn, limit, staleAfter) + } + return w.claimCleanupEvents(ctx, conn, run, limit, staleAfter) +} + +func (w *BrandAssetCleanupWorker) probeCleanupEvents(ctx context.Context, conn *pgxpool.Conn, limit int, staleAfter time.Duration) ([]brandCleanupEvent, error) { rows, err := conn.Query(ctx, ` - SELECT tenant_id, id - FROM brands - WHERE deleted_at IS NOT NULL - AND status = 'deleting' - ORDER BY deleted_at ASC, id ASC + SELECT id, tenant_id, brand_id, attempts + FROM brand_asset_cleanup_events + WHERE ( + status IN ('pending', 'failed') + AND next_attempt_at <= NOW() + ) + OR ( + status = 'processing' + AND locked_at < NOW() - $2::interval + ) + ORDER BY next_attempt_at ASC, id ASC LIMIT $1 - FOR UPDATE SKIP LOCKED - `, limit) + `, limit, intervalParam(staleAfter)) if err != nil { return nil, err } defer rows.Close() - items := make([]brandCleanupCandidate, 0, limit) + items := make([]brandCleanupEvent, 0, limit) for rows.Next() { - var item brandCleanupCandidate - if err := rows.Scan(&item.TenantID, &item.BrandID); err != nil { + var item brandCleanupEvent + if err := rows.Scan(&item.EventID, &item.TenantID, &item.BrandID, &item.Attempts); err != nil { return nil, err } items = append(items, item) @@ -175,6 +214,88 @@ func (w *BrandAssetCleanupWorker) claimCandidates(ctx context.Context, conn *pgx return items, rows.Err() } +func (w *BrandAssetCleanupWorker) claimCleanupEvents(ctx context.Context, conn *pgxpool.Conn, run JobRunContext, limit int, staleAfter time.Duration) ([]brandCleanupEvent, error) { + lockedBy := "scheduler" + if value, ok := run.Config["worker_id"].(string); ok && value != "" { + lockedBy = value + } + if run.TriggerID != nil { + lockedBy = fmt.Sprintf("scheduler:%s:%d", run.Trigger, *run.TriggerID) + } + rows, err := conn.Query(ctx, ` + WITH picked AS ( + SELECT id + FROM brand_asset_cleanup_events + WHERE ( + status IN ('pending', 'failed') + AND next_attempt_at <= NOW() + ) + OR ( + status = 'processing' + AND locked_at < NOW() - $2::interval + ) + ORDER BY next_attempt_at ASC, id ASC + LIMIT $1 + FOR UPDATE SKIP LOCKED + ) + UPDATE brand_asset_cleanup_events e + SET status = 'processing', + attempts = e.attempts + 1, + locked_at = NOW(), + locked_by = $3, + updated_at = NOW() + FROM picked + WHERE e.id = picked.id + RETURNING e.id, e.tenant_id, e.brand_id, e.attempts + `, limit, intervalParam(staleAfter), lockedBy) + if err != nil { + return nil, err + } + defer rows.Close() + + items := make([]brandCleanupEvent, 0, limit) + for rows.Next() { + var item brandCleanupEvent + if err := rows.Scan(&item.EventID, &item.TenantID, &item.BrandID, &item.Attempts); err != nil { + return nil, err + } + items = append(items, item) + } + return items, rows.Err() +} + +func (w *BrandAssetCleanupWorker) markCleanupEventSuccess(ctx context.Context, conn *pgxpool.Conn, eventID int64) error { + _, err := conn.Exec(ctx, ` + UPDATE brand_asset_cleanup_events + SET status = 'success', + last_error = NULL, + locked_at = NULL, + locked_by = NULL, + processed_at = NOW(), + updated_at = NOW() + WHERE id = $1 + `, eventID) + return err +} + +func (w *BrandAssetCleanupWorker) markCleanupEventFailed(ctx context.Context, conn *pgxpool.Conn, event brandCleanupEvent, cause error) error { + message := "" + if cause != nil { + message = cause.Error() + } + _, err := conn.Exec(ctx, ` + UPDATE brand_asset_cleanup_events + SET status = 'failed', + last_error = $2, + next_attempt_at = $3, + locked_at = NULL, + locked_by = NULL, + updated_at = NOW() + WHERE id = $1 + `, event.EventID, message, time.Now().UTC().Add(brandCleanupRetryDelay(event.Attempts))) + return err +} + func (w *BrandAssetCleanupWorker) probeBrand(ctx context.Context, conn *pgxpool.Conn, c brandCleanupCandidate) (brandCleanupStats, error) { stats := brandCleanupStats{BrandCount: 1} if err := conn.QueryRow(ctx, `SELECT COUNT(*)::bigint FROM articles WHERE tenant_id = $1 AND brand_id = $2 AND deleted_at IS NULL`, c.TenantID, c.BrandID).Scan(&stats.ArticleCount); err != nil { @@ -853,6 +974,23 @@ func nullableInt64Array(values []int64) any { return values } +func intervalParam(value time.Duration) string { + if value <= 0 { + value = defaultBrandAssetCleanupEventStaleAfter + } + return fmt.Sprintf("%d milliseconds", value.Milliseconds()) +} + +func brandCleanupRetryDelay(attempts int) time.Duration { + if attempts <= 1 { + return time.Minute + } + if attempts > 6 { + attempts = 6 + } + return time.Duration(1<<(attempts-1)) * time.Minute +} + func (s *brandCleanupStats) add(other brandCleanupStats) { s.BrandCount += other.BrandCount s.ArticleCount += other.ArticleCount diff --git a/server/internal/scheduler/brand_asset_cleanup_worker_test.go b/server/internal/scheduler/brand_asset_cleanup_worker_test.go index c32ed85..def39dc 100644 --- a/server/internal/scheduler/brand_asset_cleanup_worker_test.go +++ b/server/internal/scheduler/brand_asset_cleanup_worker_test.go @@ -3,6 +3,7 @@ package scheduler import ( "context" "testing" + "time" ) func TestBrandAssetCleanupRunOnceSkipsWhenStoreMissing(t *testing.T) { @@ -41,3 +42,57 @@ func TestBrandCleanupStatsWriteToIncludesTerminalMetrics(t *testing.T) { } } } + +func TestBrandCleanupRetryDelayBacksOffAndCaps(t *testing.T) { + cases := []struct { + attempts int + want time.Duration + }{ + {attempts: 0, want: time.Minute}, + {attempts: 1, want: time.Minute}, + {attempts: 2, want: 2 * time.Minute}, + {attempts: 6, want: 32 * time.Minute}, + {attempts: 12, want: 32 * time.Minute}, + } + + for _, tc := range cases { + if got := brandCleanupRetryDelay(tc.attempts); got != tc.want { + t.Fatalf("brandCleanupRetryDelay(%d) = %s, want %s", tc.attempts, got, tc.want) + } + } +} + +func TestBrandAssetCleanupReconcileRunOnceSkipsWhenStoreMissing(t *testing.T) { + stats, err := (*BrandAssetCleanupReconcileWorker)(nil).RunOnce(context.Background(), JobRunContext{}) + if err != nil { + t.Fatalf("RunOnce returned error: %v", err) + } + if stats["skipped"] != true { + t.Fatalf("stats[skipped] = %v, want true", stats["skipped"]) + } + if stats["reason"] != "worker_not_ready" { + t.Fatalf("stats[reason] = %v, want worker_not_ready", stats["reason"]) + } +} + +func TestBrandAssetCleanupEventWorkerRunSkipsWhenMissingDependencies(t *testing.T) { + (*BrandAssetCleanupEventWorker)(nil).Run(context.Background()) + (&BrandAssetCleanupEventWorker{}).Run(context.Background()) +} + +func TestIntFromStatsReadsNumericValues(t *testing.T) { + for _, tc := range []struct { + value any + want int + }{ + {value: 3, want: 3}, + {value: int64(4), want: 4}, + {value: float64(5), want: 5}, + {value: "6", want: 0}, + } { + got := intFromStats(map[string]any{"event_count": tc.value}, "event_count") + if got != tc.want { + t.Fatalf("intFromStats(%T(%v)) = %d, want %d", tc.value, tc.value, got, tc.want) + } + } +} diff --git a/server/internal/tenant/app/brand_asset_cleanup_event.go b/server/internal/tenant/app/brand_asset_cleanup_event.go new file mode 100644 index 0000000..56f1b06 --- /dev/null +++ b/server/internal/tenant/app/brand_asset_cleanup_event.go @@ -0,0 +1,50 @@ +package app + +import ( + "context" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" +) + +const BrandAssetCleanupNotifyChannel = "brand_asset_cleanup_events" + +type brandAssetCleanupEventQuerier interface { + Exec(context.Context, string, ...any) (pgconn.CommandTag, error) + QueryRow(context.Context, string, ...any) pgx.Row +} + +func EnqueueBrandAssetCleanupEvent(ctx context.Context, q brandAssetCleanupEventQuerier, tenantID, brandID int64) (int64, error) { + var eventID int64 + err := q.QueryRow(ctx, ` + INSERT INTO brand_asset_cleanup_events (tenant_id, brand_id, status, next_attempt_at, last_error, processed_at) + VALUES ($1, $2, 'pending', NOW(), NULL, NULL) + ON CONFLICT (tenant_id, brand_id) WHERE status IN ('pending', 'processing', 'failed') DO UPDATE + SET status = CASE + WHEN brand_asset_cleanup_events.status = 'processing' THEN brand_asset_cleanup_events.status + ELSE 'pending' + END, + next_attempt_at = CASE + WHEN brand_asset_cleanup_events.status = 'processing' THEN brand_asset_cleanup_events.next_attempt_at + ELSE NOW() + END, + last_error = CASE + WHEN brand_asset_cleanup_events.status = 'processing' THEN brand_asset_cleanup_events.last_error + ELSE NULL + END, + processed_at = NULL, + updated_at = NOW() + RETURNING id + `, tenantID, brandID).Scan(&eventID) + if err != nil { + return 0, err + } + return eventID, nil +} + +func NotifyBrandAssetCleanup(ctx context.Context, q brandAssetCleanupEventQuerier) error { + _, err := q.Exec(ctx, ` + SELECT pg_notify($1, '') + `, BrandAssetCleanupNotifyChannel) + return err +} diff --git a/server/internal/tenant/app/brand_service.go b/server/internal/tenant/app/brand_service.go index bb88d9a..fd8131f 100644 --- a/server/internal/tenant/app/brand_service.go +++ b/server/internal/tenant/app/brand_service.go @@ -309,6 +309,13 @@ func (s *BrandService) Delete(ctx context.Context, id int64) error { _, _ = tx.Exec(ctx, `UPDATE brand_questions SET deleted_at = NOW(), updated_at = NOW() WHERE brand_id = $1 AND tenant_id = $2 AND deleted_at IS NULL`, id, actor.TenantID) _, _ = tx.Exec(ctx, `UPDATE competitors SET deleted_at = NOW(), updated_at = NOW() WHERE brand_id = $1 AND tenant_id = $2 AND deleted_at IS NULL`, id, actor.TenantID) + if _, err := EnqueueBrandAssetCleanupEvent(ctx, tx, actor.TenantID, id); err != nil { + return response.ErrInternal(50010, "delete_failed", "failed to enqueue brand cleanup event") + } + if err := NotifyBrandAssetCleanup(ctx, tx); err != nil { + return response.ErrInternal(50010, "delete_failed", "failed to notify brand cleanup") + } + afterJSON, _ := json.Marshal(map[string]interface{}{"id": id, "status": "deleting", "cleanup": "background"}) if err := tx.Commit(ctx); err != nil { return err diff --git a/server/migrations/20260620110000_create_brand_asset_cleanup_events.down.sql b/server/migrations/20260620110000_create_brand_asset_cleanup_events.down.sql new file mode 100644 index 0000000..dfa0145 --- /dev/null +++ b/server/migrations/20260620110000_create_brand_asset_cleanup_events.down.sql @@ -0,0 +1,4 @@ +DROP INDEX IF EXISTS idx_brands_deleting_cleanup_candidate; +DROP INDEX IF EXISTS idx_brand_asset_cleanup_events_processing; + +DROP TABLE IF EXISTS brand_asset_cleanup_events; diff --git a/server/migrations/20260620110000_create_brand_asset_cleanup_events.up.sql b/server/migrations/20260620110000_create_brand_asset_cleanup_events.up.sql new file mode 100644 index 0000000..75861e5 --- /dev/null +++ b/server/migrations/20260620110000_create_brand_asset_cleanup_events.up.sql @@ -0,0 +1,36 @@ +CREATE TABLE IF NOT EXISTS brand_asset_cleanup_events ( + id BIGSERIAL PRIMARY KEY, + tenant_id BIGINT NOT NULL, + brand_id BIGINT NOT NULL, + status VARCHAR(20) NOT NULL DEFAULT 'pending', + attempts INT NOT NULL DEFAULT 0, + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + last_error TEXT, + locked_at TIMESTAMPTZ, + locked_by TEXT, + processed_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CHECK (status IN ('pending', 'processing', 'success', 'failed')) +); + +CREATE UNIQUE INDEX IF NOT EXISTS uk_brand_asset_cleanup_events_active + ON brand_asset_cleanup_events (tenant_id, brand_id) + WHERE status IN ('pending', 'processing', 'failed'); + +CREATE INDEX IF NOT EXISTS idx_brand_asset_cleanup_events_pending + ON brand_asset_cleanup_events (next_attempt_at, id) + INCLUDE (tenant_id, brand_id) + WHERE status IN ('pending', 'failed'); + +CREATE INDEX IF NOT EXISTS idx_brand_asset_cleanup_events_processing + ON brand_asset_cleanup_events (locked_at, id) + INCLUDE (tenant_id, brand_id) + WHERE status = 'processing' + AND locked_at IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_brands_deleting_cleanup_candidate + ON brands (deleted_at, id) + INCLUDE (tenant_id) + WHERE status = 'deleting' + AND deleted_at IS NOT NULL; diff --git a/server/migrations_ops/20260620110000_event_drive_brand_asset_cleanup.down.sql b/server/migrations_ops/20260620110000_event_drive_brand_asset_cleanup.down.sql new file mode 100644 index 0000000..a4f8c06 --- /dev/null +++ b/server/migrations_ops/20260620110000_event_drive_brand_asset_cleanup.down.sql @@ -0,0 +1,12 @@ +DELETE FROM ops.scheduler_job_triggers +WHERE job_key = 'brand_asset_cleanup_reconcile'; + +DELETE FROM ops.scheduler_jobs +WHERE job_key = 'brand_asset_cleanup_reconcile'; + +UPDATE ops.scheduler_jobs +SET schedule_type = 'interval', + interval_seconds = 60, + config = config - 'suppress_empty_success_runs' - 'empty_success_metric_keys', + updated_at = NOW() +WHERE job_key = 'brand_asset_cleanup'; diff --git a/server/migrations_ops/20260620110000_event_drive_brand_asset_cleanup.up.sql b/server/migrations_ops/20260620110000_event_drive_brand_asset_cleanup.up.sql new file mode 100644 index 0000000..420743c --- /dev/null +++ b/server/migrations_ops/20260620110000_event_drive_brand_asset_cleanup.up.sql @@ -0,0 +1,35 @@ +UPDATE ops.scheduler_jobs +SET schedule_type = 'manual', + interval_seconds = 3600, + config = config + || jsonb_build_object( + 'batch_size', 20, + 'statement_timeout', '8s', + 'dry_run', false, + 'suppress_empty_success_runs', true, + 'empty_success_metric_keys', jsonb_build_array('event_count', 'candidate_count') + ), + updated_at = NOW() +WHERE job_key = 'brand_asset_cleanup'; + +DELETE FROM ops.scheduler_job_triggers +WHERE job_key = 'brand_asset_cleanup' + AND reason IN ('brand deleted', 'brand cleanup continuation', 'brand cleanup reconcile'); + +INSERT INTO ops.scheduler_jobs + (job_key, display_name, category, description, enabled, schedule_type, interval_seconds, timezone, timeout_seconds, batch_size, config) +VALUES + ('brand_asset_cleanup_reconcile', '品牌资产清理兜底巡检', 'tenant', '低频巡检仍处于 deleting 的品牌并补发清理事件；正常路径由业务事件 worker 通过 LISTEN/NOTIFY 驱动。', true, 'interval', 3600, 'Asia/Shanghai', 60, 100, + '{"batch_size":100,"statement_timeout":"8s","dry_run":false,"suppress_empty_success_runs":true,"empty_success_metric_keys":["candidate_count","runnable_event_count","enqueued_event_count"]}'::jsonb) +ON CONFLICT (job_key) DO UPDATE +SET display_name = EXCLUDED.display_name, + category = EXCLUDED.category, + description = EXCLUDED.description, + enabled = EXCLUDED.enabled, + schedule_type = EXCLUDED.schedule_type, + interval_seconds = EXCLUDED.interval_seconds, + timezone = EXCLUDED.timezone, + timeout_seconds = EXCLUDED.timeout_seconds, + batch_size = EXCLUDED.batch_size, + config = ops.scheduler_jobs.config || EXCLUDED.config, + updated_at = NOW();