feat: add monitoring marked articles and retention updates

server/internal/scheduler/monitoring_retention_worker.go

@@ -22,9 +22,10 @@ type MonitoringRetentionWorker struct {
 }
 
 type retentionTarget struct {
-	Name      string
-	DeleteSQL string
-	ProbeSQL  string
+	Name         string
+	DeleteSQL    string
+	ProbeSQL     string
+	PreDeleteSQL string
 }
 
 func NewMonitoringRetentionWorker(monitoringPool *pgxpool.Pool, logger *zap.Logger) *MonitoringRetentionWorker {
@@ -136,11 +137,10 @@ func (w *MonitoringRetentionWorker) deleteTarget(ctx context.Context, conn *pgxp
 	total := int64(0)
 	batches := 0
 	for batches < maxBatches {
-		tag, err := conn.Exec(ctx, target.DeleteSQL, cutoffDate, batchSize)
+		affected, err := w.deleteTargetBatch(ctx, conn, target, cutoffDate, batchSize)
 		if err != nil {
 			return total, batches, err
 		}
-		affected := tag.RowsAffected()
 		if affected == 0 {
 			break
 		}
@@ -153,6 +153,40 @@ func (w *MonitoringRetentionWorker) deleteTarget(ctx context.Context, conn *pgxp
 	return total, batches, nil
 }
 
+func (w *MonitoringRetentionWorker) deleteTargetBatch(ctx context.Context, conn *pgxpool.Conn, target retentionTarget, cutoffDate string, batchSize int) (int64, error) {
+	if strings.TrimSpace(target.PreDeleteSQL) == "" {
+		tag, err := conn.Exec(ctx, target.DeleteSQL, cutoffDate, batchSize)
+		if err != nil {
+			return 0, err
+		}
+		return tag.RowsAffected(), nil
+	}
+
+	tx, err := conn.Begin(ctx)
+	if err != nil {
+		return 0, fmt.Errorf("begin delete batch: %w", err)
+	}
+	committed := false
+	defer func() {
+		if !committed {
+			_ = tx.Rollback(context.Background())
+		}
+	}()
+
+	if _, err := tx.Exec(ctx, target.PreDeleteSQL, cutoffDate, batchSize); err != nil {
+		return 0, fmt.Errorf("pre-delete batch: %w", err)
+	}
+	tag, err := tx.Exec(ctx, target.DeleteSQL, cutoffDate, batchSize)
+	if err != nil {
+		return 0, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return 0, fmt.Errorf("commit delete batch: %w", err)
+	}
+	committed = true
+	return tag.RowsAffected(), nil
+}
+
 func retentionTargets() []retentionTarget {
 	return []retentionTarget{
 		{
@@ -190,6 +224,12 @@ func retentionTargets() []retentionTarget {
 			ProbeSQL:  batchProbeSQL("question_monitor_runs", "id", "business_date < $1::date", "business_date ASC, id ASC"),
 			DeleteSQL: batchDeleteSQL("question_monitor_runs", "id", "business_date < $1::date", "business_date ASC, id ASC"),
 		},
+		{
+			Name:         "monitoring_user_marked_articles",
+			ProbeSQL:     batchProbeSQL("monitoring_user_marked_articles", "id", "$1::date IS NOT NULL AND expires_at < NOW()", "expires_at ASC, id ASC"),
+			PreDeleteSQL: markedArticleAttributionCleanupSQL(),
+			DeleteSQL:    batchDeleteSQL("monitoring_user_marked_articles", "id", "$1::date IS NOT NULL AND expires_at < NOW()", "expires_at ASC, id ASC"),
+		},
 	}
 }
 
@@ -223,3 +263,24 @@ func batchDeleteSQL(table, idColumn, predicate, orderBy string) string {
 	}
 	return strings.Join(parts, "\n")
 }
+
+func markedArticleAttributionCleanupSQL() string {
+	parts := []string{
+		"WITH doomed AS (",
+		"SELECT tenant_id, id",
+		"FROM monitoring_user_marked_articles",
+		"WHERE $1::date IS NOT NULL AND expires_at < NOW()",
+		"ORDER BY expires_at ASC, id ASC",
+		"LIMIT $2",
+		"FOR UPDATE SKIP LOCKED",
+		")",
+		"UPDATE monitoring_citation_facts cf",
+		"SET content_source_type = NULL,",
+		"    content_source_id = NULL",
+		"FROM doomed",
+		"WHERE cf.tenant_id = doomed.tenant_id",
+		"  AND cf.content_source_type = 'manual_mark'",
+		"  AND cf.content_source_id = doomed.id",
+	}
+	return strings.Join(parts, "\n")
+}

server/internal/scheduler/monitoring_retention_worker_test.go

@@ -34,3 +34,34 @@ func TestBatchProbeSQLCapsDryRunProbe(t *testing.T) {
 		t.Fatalf("batchProbeSQL should count only the limited candidate subquery:\n%s", sql)
 	}
 }
+
+func TestMarkedArticleRetentionClearsManualAttributionBeforeDelete(t *testing.T) {
+	var target retentionTarget
+	for _, item := range retentionTargets() {
+		if item.Name == "monitoring_user_marked_articles" {
+			target = item
+			break
+		}
+	}
+	if target.Name == "" {
+		t.Fatal("monitoring_user_marked_articles retention target missing")
+	}
+	if target.PreDeleteSQL == "" {
+		t.Fatal("monitoring_user_marked_articles retention target must clear citation attribution before delete")
+	}
+
+	for _, want := range []string{
+		"SELECT tenant_id, id",
+		"UPDATE monitoring_citation_facts cf",
+		"cf.tenant_id = doomed.tenant_id",
+		"content_source_type = 'manual_mark'",
+		"content_source_id = doomed.id",
+		"SET content_source_type = NULL",
+		"LIMIT $2",
+		"FOR UPDATE SKIP LOCKED",
+	} {
+		if !strings.Contains(target.PreDeleteSQL, want) {
+			t.Fatalf("marked article attribution cleanup SQL missing %q in SQL:\n%s", want, target.PreDeleteSQL)
+		}
+	}
+}