feat: implement brand asset cleanup event handling and reconciliation logic

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 {

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

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

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
+		SELECT id, tenant_id, brand_id, attempts
-		FROM brands
+		FROM brand_asset_cleanup_events
-		WHERE deleted_at IS NOT NULL
+		WHERE (
-		  AND status = 'deleting'
+		        status IN ('pending', 'failed')
-		ORDER BY deleted_at ASC, id ASC
+		        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, intervalParam(staleAfter))
-	`, limit)
 	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
+		var item brandCleanupEvent
-		if err := rows.Scan(&item.TenantID, &item.BrandID); err != nil {
+		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

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

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

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

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;

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;

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';

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