From 509f9e0c712f83e2e66209dfb2de82b8622a5c3a Mon Sep 17 00:00:00 2001
From: liangxu <liangxu0920@qq.com>
Date: Wed, 20 May 2026 21:59:36 +0800
Subject: [PATCH] Add AI point usage cycle cleanup

---
 .../src/components/ArticleDetailDrawer.vue    |  12 -
 apps/admin-web/src/views/AIPointUsageView.vue |   3 +-
 server/cmd/scheduler/main.go                  |   7 +
 .../ai_point_usage_cleanup_worker.go          | 246 ++++++++++++++++++
 .../ai_point_usage_cleanup_worker_test.go     |  49 ++++
 ...point_usage_cleanup_scheduler_job.down.sql |   8 +
 ...i_point_usage_cleanup_scheduler_job.up.sql |  11 +
 7 files changed, 322 insertions(+), 14 deletions(-)
 create mode 100644 server/internal/scheduler/ai_point_usage_cleanup_worker.go
 create mode 100644 server/internal/scheduler/ai_point_usage_cleanup_worker_test.go
 create mode 100644 server/migrations_ops/20260520150000_add_ai_point_usage_cleanup_scheduler_job.down.sql
 create mode 100644 server/migrations_ops/20260520150000_add_ai_point_usage_cleanup_scheduler_job.up.sql

diff --git a/apps/admin-web/src/components/ArticleDetailDrawer.vue b/apps/admin-web/src/components/ArticleDetailDrawer.vue
index 32f38e5..b0fd3d1 100644
--- a/apps/admin-web/src/components/ArticleDetailDrawer.vue
+++ b/apps/admin-web/src/components/ArticleDetailDrawer.vue
@@ -79,12 +79,6 @@ const versionColumns = computed<TableColumnsType<ArticleVersion>>(() => [
     dataIndex: 'title',
     key: 'title',
   },
-  {
-    title: t('common.source'),
-    dataIndex: 'source_label',
-    key: 'source_label',
-    width: 120,
-  },
   {
     title: t('common.wordCount'),
     dataIndex: 'word_count',
@@ -304,9 +298,6 @@ async function copyPublishLink(record: PublishRecord): Promise<void> {
           <a-descriptions-item :label="t('article.meta.currentVersion')">
             {{ detail.version_no ?? '--' }}
           </a-descriptions-item>
-          <a-descriptions-item :label="t('article.meta.source')">
-            {{ detail.source_label || '--' }}
-          </a-descriptions-item>
           <a-descriptions-item :label="t('article.meta.wordCount')">
             {{ detail.word_count || '--' }}
           </a-descriptions-item>
@@ -373,9 +364,6 @@ async function copyPublishLink(record: PublishRecord): Promise<void> {
                   <span>{{ formatDateTime(record.created_at) }}</span>
                 </div>
               </template>
-              <template v-else-if="column.key === 'source_label'">
-                {{ record.source_label || '--' }}
-              </template>
               <template v-else-if="column.key === 'created_at'">
                 {{ formatDateTime(record.created_at) }}
               </template>
diff --git a/apps/admin-web/src/views/AIPointUsageView.vue b/apps/admin-web/src/views/AIPointUsageView.vue
index fae00f6..0a87d68 100644
--- a/apps/admin-web/src/views/AIPointUsageView.vue
+++ b/apps/admin-web/src/views/AIPointUsageView.vue
@@ -240,8 +240,7 @@ function getAIPointAmountMeta(item: AIPointUsageLog): {
           <template v-if="column.key === 'usage_type'">
             <div class="usage-type-cell">
               <strong>{{ getAIPointUsageTypeLabel(record.usage_type) }}</strong>
-              <span v-if="record.model">{{ record.model }}</span>
-              <span v-else-if="record.error_message" class="usage-error">
+              <span v-if="record.error_message" class="usage-error">
                 {{ record.error_message }}
               </span>
             </div>
diff --git a/server/cmd/scheduler/main.go b/server/cmd/scheduler/main.go
index 47de276..2db4981 100644
--- a/server/cmd/scheduler/main.go
+++ b/server/cmd/scheduler/main.go
@@ -95,6 +95,7 @@ func main() {
 		WithConfigProvider(app.ConfigStore).
 		WithCache(app.Cache)
 	monitoringRetentionWorker := internalscheduler.NewMonitoringRetentionWorker(app.MonitoringDB, app.Logger)
+	aiPointUsageCleanupWorker := internalscheduler.NewAIPointUsageCleanupWorker(app.DB, app.Logger)
 
 	controlPlane := internalscheduler.NewControlPlane(app.DB, app.Logger, []internalscheduler.ControlledJob{
 		{
@@ -145,6 +146,12 @@ func main() {
 				return monitoringRetentionWorker.RunOnce(runCtx, jobCtx)
 			},
 		},
+		{
+			Key: "ai_point_usage_cleanup",
+			Run: func(runCtx context.Context, jobCtx internalscheduler.JobRunContext) (map[string]any, error) {
+				return aiPointUsageCleanupWorker.RunOnce(runCtx, jobCtx)
+			},
+		},
 	})
 	startSchedulerWorker(ctx, &workerWG, "control_plane", app.Logger.Sugar(), controlPlane.Run)
 
diff --git a/server/internal/scheduler/ai_point_usage_cleanup_worker.go b/server/internal/scheduler/ai_point_usage_cleanup_worker.go
new file mode 100644
index 0000000..45c8de8
--- /dev/null
+++ b/server/internal/scheduler/ai_point_usage_cleanup_worker.go
@@ -0,0 +1,246 @@
+package scheduler
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+	"go.uber.org/zap"
+
+	tenantrepo "github.com/geo-platform/tenant-api/internal/tenant/repository"
+)
+
+const (
+	defaultAIPointUsageCleanupBatch   = 5000
+	defaultAIPointUsageCleanupBatches = 100
+)
+
+type AIPointUsageCleanupWorker struct {
+	pool   *pgxpool.Pool
+	logger *zap.Logger
+}
+
+func NewAIPointUsageCleanupWorker(pool *pgxpool.Pool, logger *zap.Logger) *AIPointUsageCleanupWorker {
+	return &AIPointUsageCleanupWorker{pool: pool, logger: logger}
+}
+
+func (w *AIPointUsageCleanupWorker) RunOnce(ctx context.Context, run JobRunContext) (map[string]any, error) {
+	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", defaultAIPointUsageCleanupBatch)
+	if run.Job != nil && run.Job.BatchSize != nil {
+		batchSize = *run.Job.BatchSize
+	}
+	if batchSize <= 0 {
+		batchSize = defaultAIPointUsageCleanupBatch
+	}
+	maxBatches := AsInt(run.Config, "max_batches_per_run", defaultAIPointUsageCleanupBatches)
+	if maxBatches <= 0 {
+		maxBatches = defaultAIPointUsageCleanupBatches
+	}
+	statementTimeout := AsDuration(run.Config, "statement_timeout", 5*time.Second)
+	dryRun := run.DryRun
+	now := time.Now().UTC()
+
+	stats := map[string]any{
+		"batch_size":          batchSize,
+		"max_batches_per_run": maxBatches,
+		"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 statement timeout: %w", err)
+		}
+		defer conn.Exec(context.Background(), `RESET statement_timeout`)
+	}
+
+	tenantIDs, err := w.loadTenantIDs(ctx, conn)
+	if err != nil {
+		return stats, fmt.Errorf("load ai point usage tenants: %w", err)
+	}
+
+	planRepo := tenantrepo.NewTenantPlanRepository(conn)
+	totalDeleted := int64(0)
+	totalCandidateProbe := int64(0)
+	tenantCount := 0
+	cleanedTenantCount := 0
+	skippedNoPlan := 0
+	batches := 0
+	batchLimitReached := false
+
+	for _, tenantID := range tenantIDs {
+		if ctx.Err() != nil {
+			return stats, ErrorFromContext(ctx)
+		}
+		if !dryRun && batches >= maxBatches {
+			batchLimitReached = true
+			break
+		}
+
+		tenantCount++
+		access, err := planRepo.GetTenantPlanAccess(ctx, tenantID, now)
+		if err != nil {
+			return stats, fmt.Errorf("load tenant %d plan access: %w", tenantID, err)
+		}
+		cutoff, ok := aiPointUsageCleanupCutoff(access, now)
+		if !ok {
+			skippedNoPlan++
+			continue
+		}
+
+		if dryRun {
+			candidates, err := w.probeTenant(ctx, conn, tenantID, cutoff, batchSize)
+			if err != nil {
+				return stats, fmt.Errorf("probe tenant %d ai point usage: %w", tenantID, err)
+			}
+			if candidates > 0 {
+				cleanedTenantCount++
+				totalCandidateProbe += candidates
+			}
+			continue
+		}
+
+		deleted, tenantBatches, err := w.deleteTenant(ctx, conn, tenantID, cutoff, batchSize, maxBatches-batches)
+		if err != nil {
+			return stats, fmt.Errorf("delete tenant %d ai point usage: %w", tenantID, err)
+		}
+		if deleted > 0 {
+			cleanedTenantCount++
+			totalDeleted += deleted
+			batches += tenantBatches
+		}
+	}
+
+	stats["tenant_count"] = tenantCount
+	stats["cleaned_tenant_count"] = cleanedTenantCount
+	stats["skipped_no_plan_count"] = skippedNoPlan
+	stats["deleted_count"] = totalDeleted
+	stats["batch_count"] = batches
+	stats["batch_limit_reached"] = batchLimitReached
+	if dryRun {
+		stats["candidate_probe_count"] = totalCandidateProbe
+	}
+	stats["duration_ms"] = time.Since(startedAt).Milliseconds()
+	return stats, nil
+}
+
+func (w *AIPointUsageCleanupWorker) loadTenantIDs(ctx context.Context, conn *pgxpool.Conn) ([]int64, error) {
+	rows, err := conn.Query(ctx, `
+SELECT DISTINCT tenant_id
+FROM ai_point_usage_logs
+ORDER BY tenant_id
+`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var tenantIDs []int64
+	for rows.Next() {
+		var tenantID int64
+		if err := rows.Scan(&tenantID); err != nil {
+			return nil, err
+		}
+		tenantIDs = append(tenantIDs, tenantID)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return tenantIDs, nil
+}
+
+func (w *AIPointUsageCleanupWorker) probeTenant(ctx context.Context, conn *pgxpool.Conn, tenantID int64, cutoff time.Time, limit int) (int64, error) {
+	if limit <= 0 {
+		limit = defaultAIPointUsageCleanupBatch
+	}
+	var count int64
+	if err := conn.QueryRow(ctx, aiPointUsageCleanupProbeSQL(), tenantID, cutoff.UTC(), limit).Scan(&count); err != nil {
+		return 0, err
+	}
+	return count, nil
+}
+
+func (w *AIPointUsageCleanupWorker) deleteTenant(ctx context.Context, conn *pgxpool.Conn, tenantID int64, cutoff time.Time, batchSize, maxBatches int) (int64, int, error) {
+	total := int64(0)
+	batches := 0
+	for batches < maxBatches {
+		tag, err := conn.Exec(ctx, aiPointUsageCleanupDeleteSQL(), tenantID, cutoff.UTC(), batchSize)
+		if err != nil {
+			return total, batches, err
+		}
+		affected := tag.RowsAffected()
+		if affected == 0 {
+			break
+		}
+		total += affected
+		batches++
+		if affected < int64(batchSize) {
+			break
+		}
+	}
+	return total, batches, nil
+}
+
+func aiPointUsageCleanupCutoff(access *tenantrepo.TenantPlanAccess, now time.Time) (time.Time, bool) {
+	if access == nil {
+		return time.Time{}, false
+	}
+
+	referenceTime := now.UTC()
+	if !access.EndAt.IsZero() && !access.EndAt.After(referenceTime) && access.EndAt.After(access.StartAt) {
+		referenceTime = access.EndAt.Add(-time.Nanosecond)
+	}
+	if referenceTime.Before(access.StartAt) {
+		referenceTime = access.StartAt
+	}
+
+	windowStart, _, _ := access.AIPointsWindow(referenceTime)
+	if windowStart.IsZero() {
+		return time.Time{}, false
+	}
+	return windowStart.UTC(), true
+}
+
+func aiPointUsageCleanupProbeSQL() string {
+	return `
+SELECT COUNT(*)::bigint
+FROM (
+	SELECT id
+	FROM ai_point_usage_logs
+	WHERE tenant_id = $1
+	  AND created_at < $2
+	  AND status IN ('completed', 'refunded', 'failed')
+	ORDER BY created_at ASC, id ASC
+	LIMIT $3
+) candidates
+`
+}
+
+func aiPointUsageCleanupDeleteSQL() string {
+	return `
+WITH doomed AS (
+	SELECT id
+	FROM ai_point_usage_logs
+	WHERE tenant_id = $1
+	  AND created_at < $2
+	  AND status IN ('completed', 'refunded', 'failed')
+	ORDER BY created_at ASC, id ASC
+	LIMIT $3
+	FOR UPDATE SKIP LOCKED
+)
+DELETE FROM ai_point_usage_logs l
+USING doomed
+WHERE l.id = doomed.id
+`
+}
diff --git a/server/internal/scheduler/ai_point_usage_cleanup_worker_test.go b/server/internal/scheduler/ai_point_usage_cleanup_worker_test.go
new file mode 100644
index 0000000..32e3ea1
--- /dev/null
+++ b/server/internal/scheduler/ai_point_usage_cleanup_worker_test.go
@@ -0,0 +1,49 @@
+package scheduler
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	sharedconfig "github.com/geo-platform/tenant-api/internal/shared/config"
+	tenantrepo "github.com/geo-platform/tenant-api/internal/tenant/repository"
+)
+
+func TestAIPointUsageCleanupCutoffUsesCurrentCycleStart(t *testing.T) {
+	access := &tenantrepo.TenantPlanAccess{
+		StartAt: time.Date(2026, 5, 12, 21, 0, 0, 0, time.UTC),
+		EndAt:   time.Date(2026, 8, 12, 21, 0, 0, 0, time.UTC),
+		Policy: sharedconfig.PlanQuotaPolicy{
+			AIPointsMonthly: 1500,
+		},
+	}
+
+	got, ok := aiPointUsageCleanupCutoff(access, time.Date(2026, 6, 20, 10, 0, 0, 0, time.UTC))
+	if !ok {
+		t.Fatal("expected cutoff")
+	}
+	want := time.Date(2026, 6, 12, 21, 0, 0, 0, time.UTC)
+	if !got.Equal(want) {
+		t.Fatalf("cutoff = %s, want %s", got.Format(time.RFC3339), want.Format(time.RFC3339))
+	}
+}
+
+func TestAIPointUsageCleanupSQLIsBounded(t *testing.T) {
+	deleteSQL := aiPointUsageCleanupDeleteSQL()
+	for _, want := range []string{
+		"created_at < $2",
+		"status IN ('completed', 'refunded', 'failed')",
+		"LIMIT $3",
+		"FOR UPDATE SKIP LOCKED",
+		"DELETE FROM ai_point_usage_logs l",
+	} {
+		if !strings.Contains(deleteSQL, want) {
+			t.Fatalf("cleanup delete SQL missing %q in SQL:\n%s", want, deleteSQL)
+		}
+	}
+
+	probeSQL := aiPointUsageCleanupProbeSQL()
+	if !strings.Contains(probeSQL, "LIMIT $3") {
+		t.Fatalf("cleanup probe SQL should cap dry-run probe:\n%s", probeSQL)
+	}
+}
diff --git a/server/migrations_ops/20260520150000_add_ai_point_usage_cleanup_scheduler_job.down.sql b/server/migrations_ops/20260520150000_add_ai_point_usage_cleanup_scheduler_job.down.sql
new file mode 100644
index 0000000..d6398a8
--- /dev/null
+++ b/server/migrations_ops/20260520150000_add_ai_point_usage_cleanup_scheduler_job.down.sql
@@ -0,0 +1,8 @@
+DELETE FROM ops.scheduler_job_triggers
+WHERE job_key = 'ai_point_usage_cleanup';
+
+DELETE FROM ops.scheduler_job_runs
+WHERE job_key = 'ai_point_usage_cleanup';
+
+DELETE FROM ops.scheduler_jobs
+WHERE job_key = 'ai_point_usage_cleanup';
diff --git a/server/migrations_ops/20260520150000_add_ai_point_usage_cleanup_scheduler_job.up.sql b/server/migrations_ops/20260520150000_add_ai_point_usage_cleanup_scheduler_job.up.sql
new file mode 100644
index 0000000..d9da400
--- /dev/null
+++ b/server/migrations_ops/20260520150000_add_ai_point_usage_cleanup_scheduler_job.up.sql
@@ -0,0 +1,11 @@
+INSERT INTO ops.scheduler_jobs
+    (job_key, display_name, category, description, enabled, schedule_type, interval_seconds, timezone, timeout_seconds, batch_size, config)
+VALUES
+    ('ai_point_usage_cleanup', 'AI 点数流水周期清理', 'quota', '按当前会员周期保留 AI 点数流水，定期删除历史周期记录。', true, 'interval', 86400, 'Asia/Shanghai', 300, 5000,
+     '{"batch_size":5000,"max_batches_per_run":100,"dry_run":false,"statement_timeout":"5s","run_at_local":"03:00"}'::jsonb)
+ON CONFLICT (job_key) DO UPDATE
+SET display_name = EXCLUDED.display_name,
+    category = EXCLUDED.category,
+    description = EXCLUDED.description,
+    config = ops.scheduler_jobs.config || EXCLUDED.config,
+    updated_at = NOW();