From 6e0519a232ad1db3af532ef02ec4192cb86892d6 Mon Sep 17 00:00:00 2001 From: liangxu Date: Fri, 19 Jun 2026 11:45:13 +0800 Subject: [PATCH] feat: add brand asset cleanup worker and related functionality - Implemented a new BrandAssetCleanupWorker to handle the cleanup of brand-related assets after a brand is deleted. - Added SQL queries for cleaning up articles, keywords, questions, competitors, and monitoring data associated with a brand. - Introduced a new API endpoint to delete publish records. - Updated the router to include the new delete publish record endpoint. - Added tests for the BrandAssetCleanupWorker to ensure proper functionality. - Created migration scripts to support soft deletion of publish records and to add the brand asset cleanup scheduler job. --- .../src/components/GenerateTaskDrawer.vue | 14 +- apps/admin-web/src/lib/api.ts | 14 +- .../src/views/PublishManagementView.vue | 128 ++- packages/shared-types/src/index.ts | 7 + server/cmd/scheduler/main.go | 7 + .../scheduler/brand_asset_cleanup_worker.go | 940 ++++++++++++++++++ .../brand_asset_cleanup_worker_test.go | 43 + .../internal/shared/swagger/descriptions.go | 5 +- server/internal/tenant/app/article_service.go | 1 + server/internal/tenant/app/brand_service.go | 35 +- server/internal/tenant/app/cache_support.go | 7 + .../tenant/app/desktop_task_service.go | 27 +- .../tenant/app/enterprise_site_service.go | 3 +- server/internal/tenant/app/media_service.go | 169 +++- .../tenant/app/publish_job_service.go | 10 +- .../tenant/app/publish_record_support.go | 23 +- .../tenant/app/publish_record_support_test.go | 18 + .../tenant/repository/desktop_task_repo.go | 21 +- .../repository/generated/desktop_task.sql.go | 92 +- .../generated/desktop_task_sql_test.go | 21 + .../tenant/repository/generated/querier.go | 1 + .../repository/queries/desktop_task.sql | 29 + .../tenant/transport/media_handler.go | 14 + server/internal/tenant/transport/router.go | 5 +- .../internal/tenant/transport/router_test.go | 1 + ...00_add_publish_record_soft_delete.down.sql | 9 + ...0000_add_publish_record_soft_delete.up.sql | 12 + ...brand_asset_cleanup_scheduler_job.down.sql | 2 + ...d_brand_asset_cleanup_scheduler_job.up.sql | 17 + 29 files changed, 1621 insertions(+), 54 deletions(-) create mode 100644 server/internal/scheduler/brand_asset_cleanup_worker.go create mode 100644 server/internal/scheduler/brand_asset_cleanup_worker_test.go create mode 100644 server/migrations/20260618190000_add_publish_record_soft_delete.down.sql create mode 100644 server/migrations/20260618190000_add_publish_record_soft_delete.up.sql create mode 100644 server/migrations_ops/20260619100000_add_brand_asset_cleanup_scheduler_job.down.sql create mode 100644 server/migrations_ops/20260619100000_add_brand_asset_cleanup_scheduler_job.up.sql diff --git a/apps/admin-web/src/components/GenerateTaskDrawer.vue b/apps/admin-web/src/components/GenerateTaskDrawer.vue index 3bbe49c..ced7400 100644 --- a/apps/admin-web/src/components/GenerateTaskDrawer.vue +++ b/apps/admin-web/src/components/GenerateTaskDrawer.vue @@ -592,13 +592,13 @@ function handlePromptSaved(ruleId: number): void { } function toggleSchedulePublishAccount(account: PublishAccountCard): void { - if (!account.selectable) { - return - } if (form.publishAccountIds.includes(account.id)) { form.publishAccountIds = form.publishAccountIds.filter((id) => id !== account.id) return } + if (!account.selectable) { + return + } form.publishAccountIds = [...form.publishAccountIds, account.id] } @@ -606,6 +606,10 @@ function isSchedulePublishAccountSelected(accountId: string): boolean { return form.publishAccountIds.includes(accountId) } +function canToggleSchedulePublishAccount(account: PublishAccountCard): boolean { + return account.selectable || isSchedulePublishAccountSelected(account.id) +} + function buildScheduleEnterpriseSiteTargets(): ScheduleEnterpriseSiteTarget[] { return form.publishEnterpriseSiteIds .map((siteId) => ({ @@ -1325,8 +1329,10 @@ function normalizeJsonRecord(value: Record): Record + + {{ record.brandName }} + + 公司已删除 + + @@ -567,11 +615,50 @@ async function copyErrorMessage(record: PublishRecordItem): Promise { - + + + + + + + + + + + + + + + @@ -746,6 +833,30 @@ async function copyErrorMessage(record: PublishRecordItem): Promise { font-weight: 500; } +.publish-brand-line { + display: inline-flex; + min-width: 0; + max-width: 100%; + align-items: center; + gap: 6px; + margin-top: 3px; + color: #64748b; + font-size: 12px; + line-height: 1.5; +} + +.publish-brand-name { + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + +.publish-brand-tag { + flex: 0 0 auto; + margin-inline-end: 0; + font-size: 11px; +} + .publish-account-cell { display: flex; min-width: 0; @@ -843,6 +954,11 @@ async function copyErrorMessage(record: PublishRecordItem): Promise { color: #1677ff; } +.publish-action-btn--danger:hover:not(:disabled) { + background: #fff1f0; + color: #cf1322; +} + .publish-action-tooltip-wrap { display: inline-flex; } diff --git a/packages/shared-types/src/index.ts b/packages/shared-types/src/index.ts index 5201d4d..643f0ae 100644 --- a/packages/shared-types/src/index.ts +++ b/packages/shared-types/src/index.ts @@ -1262,6 +1262,9 @@ export interface PublishRecord { id: number publish_batch_id: number article_id: number + brand_id: number + brand_name?: string | null + brand_deleted?: boolean article_title?: string | null platform_account_id?: number | null desktop_account_id: string | null @@ -1296,6 +1299,10 @@ export interface PublishRecordListResponse { history_total: number } +export interface DeletePublishRecordResponse { + deleted: boolean +} + export type EnterpriseSiteCmsType = 'pbootcms' | 'wordpress' export type EnterpriseSiteStatus = 'draft' | 'connected' | 'error' | 'disabled' diff --git a/server/cmd/scheduler/main.go b/server/cmd/scheduler/main.go index 86b84dc..e7ed5bc 100644 --- a/server/cmd/scheduler/main.go +++ b/server/cmd/scheduler/main.go @@ -123,6 +123,7 @@ func main() { WithConfigProvider(app.ConfigStore). WithCache(app.Cache) monitoringRetentionWorker := internalscheduler.NewMonitoringRetentionWorker(app.MonitoringDB, app.Logger) + brandAssetCleanupWorker := internalscheduler.NewBrandAssetCleanupWorker(app.DB, app.MonitoringDB, app.Cache, app.Logger) aiPointUsageCleanupWorker := internalscheduler.NewAIPointUsageCleanupWorker(app.DB, app.Logger) schedulerRunRetentionWorker := internalscheduler.NewSchedulerRunRetentionWorker(app.DB, app.Logger) @@ -181,6 +182,12 @@ func main() { return monitoringRetentionWorker.RunOnce(runCtx, jobCtx) }, }, + { + Key: "brand_asset_cleanup", + Run: func(runCtx context.Context, jobCtx internalscheduler.JobRunContext) (map[string]any, error) { + return brandAssetCleanupWorker.RunOnce(runCtx, jobCtx) + }, + }, { Key: "ai_point_usage_cleanup", Run: func(runCtx context.Context, jobCtx internalscheduler.JobRunContext) (map[string]any, error) { diff --git a/server/internal/scheduler/brand_asset_cleanup_worker.go b/server/internal/scheduler/brand_asset_cleanup_worker.go new file mode 100644 index 0000000..17c7f3a --- /dev/null +++ b/server/internal/scheduler/brand_asset_cleanup_worker.go @@ -0,0 +1,940 @@ +package scheduler + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "go.uber.org/zap" + + sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache" + tenantapp "github.com/geo-platform/tenant-api/internal/tenant/app" +) + +const defaultBrandAssetCleanupBatch = 20 + +type BrandAssetCleanupWorker struct { + pool *pgxpool.Pool + monitoringPool *pgxpool.Pool + cache sharedcache.Cache + logger *zap.Logger +} + +type brandCleanupCandidate struct { + TenantID int64 + BrandID int64 +} + +type brandCleanupStats struct { + BrandCount int64 + ArticleCount int64 + GenerationTaskCount int64 + GenerationTaskRecordCount int64 + KolUsageCount int64 + QuotaReservationRefundCount int64 + CancelledPublishRecordCount int64 + PublishBatchCount int64 + DesktopPublishJobCount int64 + DesktopPublishTaskCount int64 + DesktopMonitorTaskCount int64 + PromptRuleCount int64 + PromptRuleGroupCount int64 + ScheduleTaskCount int64 + KeywordCount int64 + QuestionCount int64 + CompetitorCount int64 + ImageReferenceCount int64 + MediaSupplyOrderCount int64 + MediaSupplyOrderItemCount int64 + ComplianceReviewJobCount int64 + ComplianceManualReviewCount int64 + MonitoringCollectRequestCount int64 + MonitoringCollectOutboxCount int64 + MonitoringCollectTaskCount int64 + MonitoringSnapshotCount int64 + MonitoringRunCount int64 + MonitoringParseResultCount int64 + MonitoringCitationFactCount int64 + MonitoringBrandDailyCount int64 + MonitoringBrandPlatformCount int64 + MonitoringArticleAliasCount int64 + MonitoringMarkedArticleCount int64 + MarkedDeletedBrandCount int64 + CompletedBrandCount int64 +} + +type monitoringCleanupResult struct { + Stats brandCleanupStats + MonitorTaskIDs []int64 +} + +func NewBrandAssetCleanupWorker(pool, monitoringPool *pgxpool.Pool, cache sharedcache.Cache, logger *zap.Logger) *BrandAssetCleanupWorker { + return &BrandAssetCleanupWorker{ + pool: pool, + monitoringPool: monitoringPool, + cache: cache, + logger: logger, + } +} + +func (w *BrandAssetCleanupWorker) RunOnce(ctx context.Context, run JobRunContext) (map[string]any, error) { + stats := map[string]any{} + if w == nil || w.pool == nil { + stats["skipped"] = true + stats["reason"] = "worker_not_ready" + return stats, nil + } + + startedAt := time.Now() + batchSize := AsInt(run.Config, "batch_size", defaultBrandAssetCleanupBatch) + if run.Job != nil && run.Job.BatchSize != nil { + batchSize = *run.Job.BatchSize + } + if batchSize <= 0 { + batchSize = defaultBrandAssetCleanupBatch + } + statementTimeout := AsDuration(run.Config, "statement_timeout", 5*time.Second) + 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.claimCandidates(ctx, conn, batchSize) + if err != nil { + return stats, fmt.Errorf("claim brand cleanup candidates: %w", err) + } + total := brandCleanupStats{} + failures := int64(0) + for _, candidate := range candidates { + if dryRun { + brandStats, err := w.probeBrand(ctx, conn, candidate) + if err != nil { + failures++ + w.warn("brand asset cleanup probe failed", candidate, err) + continue + } + total.add(brandStats) + continue + } + + brandStats, err := w.cleanupBrand(ctx, conn, candidate) + if err != nil { + failures++ + w.warn("brand asset cleanup failed", candidate, err) + continue + } + total.add(brandStats) + tenantapp.InvalidateBrandDeletionCaches(context.Background(), w.cache, candidate.TenantID, candidate.BrandID) + } + + stats["candidate_count"] = len(candidates) + 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) { + 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 + LIMIT $1 + FOR UPDATE SKIP LOCKED + `, 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 *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 { + return stats, err + } + if err := conn.QueryRow(ctx, `SELECT COUNT(*)::bigint FROM brand_keywords WHERE tenant_id = $1 AND brand_id = $2 AND deleted_at IS NULL`, c.TenantID, c.BrandID).Scan(&stats.KeywordCount); err != nil { + return stats, err + } + if err := conn.QueryRow(ctx, `SELECT COUNT(*)::bigint FROM brand_questions WHERE tenant_id = $1 AND brand_id = $2 AND deleted_at IS NULL`, c.TenantID, c.BrandID).Scan(&stats.QuestionCount); err != nil { + return stats, err + } + if err := conn.QueryRow(ctx, `SELECT COUNT(*)::bigint FROM competitors WHERE tenant_id = $1 AND brand_id = $2 AND deleted_at IS NULL`, c.TenantID, c.BrandID).Scan(&stats.CompetitorCount); err != nil { + return stats, err + } + return stats, nil +} + +func (w *BrandAssetCleanupWorker) cleanupBrand(ctx context.Context, conn *pgxpool.Conn, c brandCleanupCandidate) (brandCleanupStats, error) { + stats := brandCleanupStats{BrandCount: 1} + monitorTaskIDs, err := w.loadMonitoringTaskIDsForBrand(ctx, c) + if err != nil { + return stats, err + } + + tx, err := conn.Begin(ctx) + if err != nil { + return stats, err + } + committed := false + defer func() { + if !committed { + _ = tx.Rollback(context.Background()) + } + }() + + articleIDs, err := w.loadBrandArticleIDs(ctx, tx, c) + if err != nil { + return stats, err + } + articleIDArg := nullableInt64Array(articleIDs) + + generationStats, err := w.cleanupGenerationAssets(ctx, tx, c, articleIDArg) + if err != nil { + return stats, err + } + stats.GenerationTaskCount += generationStats.GenerationTaskCount + stats.GenerationTaskRecordCount += generationStats.GenerationTaskRecordCount + stats.KolUsageCount += generationStats.KolUsageCount + stats.QuotaReservationRefundCount += generationStats.QuotaReservationRefundCount + + count, err := execRows(ctx, tx, ` + UPDATE articles + SET deleted_at = COALESCE(deleted_at, NOW()), + generate_status = CASE WHEN generate_status IN ('pending', 'generating', 'running') THEN 'failed' ELSE generate_status END, + updated_at = NOW() + WHERE tenant_id = $1 + AND brand_id = $2 + AND deleted_at IS NULL + `, c.TenantID, c.BrandID) + if err != nil { + return stats, err + } + stats.ArticleCount += count + + count, err = execRows(ctx, tx, ` + UPDATE image_asset_references + SET updated_at = NOW() + WHERE tenant_id = $1 + AND article_id = ANY($2::bigint[]) + `, c.TenantID, articleIDArg) + if err != nil { + return stats, err + } + stats.ImageReferenceCount += count + if _, err := tx.Exec(ctx, ` + DELETE FROM image_asset_references + WHERE tenant_id = $1 + AND article_id = ANY($2::bigint[]) + `, c.TenantID, articleIDArg); err != nil { + return stats, err + } + + desktopStats, err := w.cancelDesktopPublishAssets(ctx, tx, c, articleIDArg) + if err != nil { + return stats, err + } + stats.DesktopPublishJobCount += desktopStats.DesktopPublishJobCount + stats.DesktopPublishTaskCount += desktopStats.DesktopPublishTaskCount + stats.CancelledPublishRecordCount += desktopStats.CancelledPublishRecordCount + stats.PublishBatchCount += desktopStats.PublishBatchCount + + count, err = execRows(ctx, tx, ` + UPDATE desktop_tasks + SET status = 'aborted', + error = jsonb_build_object( + 'code', 'brand_deleted', + 'message', 'brand deleted before monitor task completion', + 'source', 'brand_asset_cleanup' + ), + active_attempt_id = NULL, + lease_token_hash = NULL, + lease_expires_at = NULL, + updated_at = NOW() + WHERE tenant_id = $1 + AND kind = 'monitor' + AND status IN ('queued', 'in_progress', 'unknown') + AND monitor_task_id = ANY($2::bigint[]) + `, c.TenantID, nullableInt64Array(monitorTaskIDs)) + if err != nil { + return stats, err + } + stats.DesktopMonitorTaskCount += count + + count, err = execRows(ctx, tx, ` + UPDATE prompt_rules + SET deleted_at = COALESCE(deleted_at, NOW()), + updated_at = NOW() + WHERE tenant_id = $1 + AND brand_id = $2 + AND deleted_at IS NULL + `, c.TenantID, c.BrandID) + if err != nil { + return stats, err + } + stats.PromptRuleCount += count + + count, err = execRows(ctx, tx, ` + UPDATE prompt_rule_groups + SET deleted_at = COALESCE(deleted_at, NOW()), + updated_at = NOW() + WHERE tenant_id = $1 + AND brand_id = $2 + AND deleted_at IS NULL + `, c.TenantID, c.BrandID) + if err != nil { + return stats, err + } + stats.PromptRuleGroupCount += count + + count, err = execRows(ctx, tx, ` + UPDATE schedule_tasks + SET deleted_at = COALESCE(deleted_at, NOW()), + status = CASE WHEN status = 'enabled' THEN 'disabled' ELSE status END, + updated_at = NOW() + WHERE tenant_id = $1 + AND brand_id = $2 + AND deleted_at IS NULL + `, c.TenantID, c.BrandID) + if err != nil { + return stats, err + } + stats.ScheduleTaskCount += count + + count, err = execRows(ctx, tx, ` + UPDATE brand_keywords + SET deleted_at = COALESCE(deleted_at, NOW()), + updated_at = NOW() + WHERE tenant_id = $1 + AND brand_id = $2 + AND deleted_at IS NULL + `, c.TenantID, c.BrandID) + if err != nil { + return stats, err + } + stats.KeywordCount += count + + count, err = execRows(ctx, tx, ` + UPDATE brand_questions + SET deleted_at = COALESCE(deleted_at, NOW()), + updated_at = NOW() + WHERE tenant_id = $1 + AND brand_id = $2 + AND deleted_at IS NULL + `, c.TenantID, c.BrandID) + if err != nil { + return stats, err + } + stats.QuestionCount += count + + count, err = execRows(ctx, tx, ` + UPDATE competitors + SET deleted_at = COALESCE(deleted_at, NOW()), + updated_at = NOW() + WHERE tenant_id = $1 + AND brand_id = $2 + AND deleted_at IS NULL + `, c.TenantID, c.BrandID) + if err != nil { + return stats, err + } + stats.CompetitorCount += count + + mediaStats, err := w.cleanupMediaSupply(ctx, tx, c) + if err != nil { + return stats, err + } + stats.MediaSupplyOrderCount += mediaStats.MediaSupplyOrderCount + stats.MediaSupplyOrderItemCount += mediaStats.MediaSupplyOrderItemCount + + complianceStats, err := w.cancelComplianceAssets(ctx, tx, c, articleIDArg) + if err != nil { + return stats, err + } + stats.ComplianceReviewJobCount += complianceStats.ComplianceReviewJobCount + stats.ComplianceManualReviewCount += complianceStats.ComplianceManualReviewCount + + if err := tx.Commit(ctx); err != nil { + return stats, err + } + committed = true + + monitoringCleanup, err := w.cleanupMonitoringBrand(ctx, c, articleIDs) + if err != nil { + return stats, err + } + stats.add(monitoringCleanup.Stats) + + tag, err := conn.Exec(ctx, ` + UPDATE brands + SET status = 'deleted', + updated_at = NOW() + WHERE tenant_id = $1 + AND id = $2 + AND deleted_at IS NOT NULL + AND status = 'deleting' + `, c.TenantID, c.BrandID) + if err != nil { + return stats, err + } + stats.CompletedBrandCount += tag.RowsAffected() + return stats, nil +} + +func (w *BrandAssetCleanupWorker) loadBrandArticleIDs(ctx context.Context, tx pgx.Tx, c brandCleanupCandidate) ([]int64, error) { + rows, err := tx.Query(ctx, ` + SELECT id + FROM articles + WHERE tenant_id = $1 + AND brand_id = $2 + ORDER BY id ASC + `, c.TenantID, c.BrandID) + if err != nil { + return nil, err + } + defer rows.Close() + + ids := make([]int64, 0) + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} + +func (w *BrandAssetCleanupWorker) loadMonitoringTaskIDsForBrand(ctx context.Context, c brandCleanupCandidate) ([]int64, error) { + if w == nil || w.monitoringPool == nil { + return nil, nil + } + conn, err := w.monitoringPool.Acquire(ctx) + if err != nil { + return nil, fmt.Errorf("acquire monitoring connection: %w", err) + } + defer conn.Release() + return w.loadMonitoringTaskIDs(ctx, conn, c) +} + +func (w *BrandAssetCleanupWorker) cleanupGenerationAssets(ctx context.Context, tx pgx.Tx, c brandCleanupCandidate, articleIDs any) (brandCleanupStats, error) { + stats := brandCleanupStats{} + if err := tx.QueryRow(ctx, ` + WITH affected_tasks AS ( + UPDATE generation_tasks + SET status = 'failed', + error_message = COALESCE(NULLIF(error_message, ''), 'brand_deleted'), + completed_at = COALESCE(completed_at, NOW()), + lease_token = NULL, + lease_owner = NULL, + lease_expires_at = NULL, + updated_at = NOW() + WHERE tenant_id = $1 + AND article_id = ANY($2::bigint[]) + AND status IN ('queued', 'running') + RETURNING id, quota_reservation_id + ), + updated_task_records AS ( + UPDATE task_records tr + SET status = 'failed', + error_message = COALESCE(NULLIF(error_message, ''), 'brand_deleted'), + updated_at = NOW() + FROM affected_tasks t + WHERE tr.tenant_id = $1 + AND tr.task_type = 'article_generation' + AND tr.resource_type = 'generation_task' + AND tr.resource_id = t.id + AND tr.status IN ('pending', 'queued', 'running') + RETURNING tr.id + ), + updated_kol_usage AS ( + UPDATE kol_usage_logs u + SET status = 'failed', + error_message = COALESCE(NULLIF(error_message, ''), 'brand_deleted'), + finished_at = COALESCE(finished_at, NOW()) + FROM affected_tasks t + WHERE u.tenant_id = $1 + AND u.generation_task_id = t.id + AND u.status IN ('pending', 'running') + RETURNING u.id + ), + refunded_reservations AS ( + UPDATE quota_reservations qr + SET status = 'refunded', + refunded_amount = reserved_amount, + updated_at = NOW() + FROM affected_tasks t + WHERE qr.id = t.quota_reservation_id + AND qr.tenant_id = $1 + AND qr.status = 'pending' + RETURNING qr.id + ) + SELECT + (SELECT COUNT(*)::bigint FROM affected_tasks), + (SELECT COUNT(*)::bigint FROM updated_task_records), + (SELECT COUNT(*)::bigint FROM updated_kol_usage), + (SELECT COUNT(*)::bigint FROM refunded_reservations) + `, c.TenantID, articleIDs).Scan( + &stats.GenerationTaskCount, + &stats.GenerationTaskRecordCount, + &stats.KolUsageCount, + &stats.QuotaReservationRefundCount, + ); err != nil { + return stats, err + } + return stats, nil +} + +func (w *BrandAssetCleanupWorker) cancelDesktopPublishAssets(ctx context.Context, tx pgx.Tx, c brandCleanupCandidate, articleIDs any) (brandCleanupStats, error) { + stats := brandCleanupStats{} + if err := tx.QueryRow(ctx, ` + WITH affected_records AS ( + UPDATE publish_records pr + SET status = 'cancelled', + error_message = COALESCE(NULLIF(pr.error_message, ''), 'brand_deleted'), + updated_at = NOW() + WHERE pr.tenant_id = $1 + AND pr.article_id = ANY($2::bigint[]) + AND pr.deleted_at IS NULL + AND pr.status IN ('queued', 'publishing', 'pending', 'running') + RETURNING pr.publish_batch_id + ), + affected_batches AS ( + SELECT DISTINCT publish_batch_id + FROM affected_records + WHERE publish_batch_id IS NOT NULL + ), + updated_batches AS ( + UPDATE publish_batches pb + SET status = CASE + WHEN EXISTS ( + SELECT 1 + FROM publish_records pr + WHERE pr.publish_batch_id = pb.id + AND pr.deleted_at IS NULL + AND pr.status IN ('queued', 'publishing', 'pending', 'running') + ) THEN 'publishing' + WHEN EXISTS ( + SELECT 1 + FROM publish_records pr + WHERE pr.publish_batch_id = pb.id + AND pr.deleted_at IS NULL + AND pr.status IN ('success', 'published', 'publish_success') + ) AND EXISTS ( + SELECT 1 + FROM publish_records pr + WHERE pr.publish_batch_id = pb.id + AND pr.deleted_at IS NULL + AND pr.status NOT IN ('success', 'published', 'publish_success') + ) THEN 'partial_success' + WHEN EXISTS ( + SELECT 1 + FROM publish_records pr + WHERE pr.publish_batch_id = pb.id + AND pr.deleted_at IS NULL + AND pr.status IN ('success', 'published', 'publish_success') + ) THEN 'success' + ELSE 'failed' + END, + updated_at = NOW() + FROM affected_batches ab + WHERE pb.id = ab.publish_batch_id + AND pb.tenant_id = $1 + RETURNING pb.id + ) + SELECT + (SELECT COUNT(*)::bigint FROM affected_records), + (SELECT COUNT(*)::bigint FROM updated_batches) + `, c.TenantID, articleIDs).Scan(&stats.CancelledPublishRecordCount, &stats.PublishBatchCount); err != nil { + return stats, err + } + + if err := tx.QueryRow(ctx, ` + WITH updated_tasks AS ( + UPDATE desktop_tasks dt + SET status = 'aborted', + error = jsonb_build_object( + 'code', 'brand_deleted', + 'message', 'brand deleted before publish completion', + 'source', 'brand_asset_cleanup' + ), + active_attempt_id = NULL, + lease_token_hash = NULL, + lease_expires_at = NULL, + updated_at = NOW() + FROM desktop_publish_jobs j + WHERE dt.job_id = j.desktop_id + AND dt.tenant_id = j.tenant_id + AND dt.tenant_id = $1 + AND j.article_id = ANY($2::bigint[]) + AND dt.kind = 'publish' + AND dt.status IN ('queued', 'in_progress', 'unknown') + RETURNING dt.job_id + ), + updated_jobs AS ( + UPDATE desktop_publish_jobs j + SET status = 'cancelled', + updated_at = NOW() + WHERE j.tenant_id = $1 + AND j.article_id = ANY($2::bigint[]) + AND j.status IN ('queued', 'blocked_by_compliance') + RETURNING j.desktop_id + ) + SELECT + (SELECT COUNT(*)::bigint FROM updated_tasks), + (SELECT COUNT(*)::bigint FROM updated_jobs) + `, c.TenantID, articleIDs).Scan(&stats.DesktopPublishTaskCount, &stats.DesktopPublishJobCount); err != nil { + return stats, err + } + + return stats, nil +} + +func (w *BrandAssetCleanupWorker) cleanupMediaSupply(ctx context.Context, tx pgx.Tx, c brandCleanupCandidate) (brandCleanupStats, error) { + stats := brandCleanupStats{} + if err := tx.QueryRow(ctx, ` + WITH affected_orders AS ( + UPDATE media_supply_orders + SET status = CASE WHEN status IN ('queued', 'submitting') THEN 'failed' ELSE status END, + error_message = CASE + WHEN status IN ('queued', 'submitting') THEN COALESCE(NULLIF(error_message, ''), 'brand_deleted') + ELSE error_message + END, + deleted_at = COALESCE(deleted_at, NOW()), + completed_at = CASE WHEN status IN ('queued', 'submitting') THEN COALESCE(completed_at, NOW()) ELSE completed_at END, + updated_at = NOW() + WHERE tenant_id = $1 + AND brand_id = $2 + AND deleted_at IS NULL + RETURNING id + ), + affected_items AS ( + UPDATE media_supply_order_items i + SET status = CASE WHEN i.status IN ('queued', 'submitting') THEN 'failed' ELSE i.status END, + updated_at = NOW() + FROM affected_orders o + WHERE i.order_id = o.id + RETURNING i.id + ) + SELECT + (SELECT COUNT(*)::bigint FROM affected_orders), + (SELECT COUNT(*)::bigint FROM affected_items) + `, c.TenantID, c.BrandID).Scan(&stats.MediaSupplyOrderCount, &stats.MediaSupplyOrderItemCount); err != nil { + return stats, err + } + return stats, nil +} + +func (w *BrandAssetCleanupWorker) cancelComplianceAssets(ctx context.Context, tx pgx.Tx, c brandCleanupCandidate, articleIDs any) (brandCleanupStats, error) { + stats := brandCleanupStats{} + count, err := execRows(ctx, tx, ` + UPDATE compliance_review_jobs + SET status = 'failed', + last_error = COALESCE(NULLIF(last_error, ''), 'brand_deleted'), + updated_at = NOW() + WHERE tenant_id = $1 + AND article_id = ANY($2::bigint[]) + AND status IN ('pending', 'running') + `, c.TenantID, articleIDs) + if err != nil { + return stats, err + } + stats.ComplianceReviewJobCount += count + + if err := tx.QueryRow(ctx, ` + WITH affected_reviews AS ( + UPDATE compliance_manual_reviews + SET status = 'cancelled', + cancelled_by = COALESCE(cancelled_by, requested_by), + cancelled_at = COALESCE(cancelled_at, NOW()), + cancel_reason = COALESCE(NULLIF(cancel_reason, ''), 'brand_deleted') + WHERE tenant_id = $1 + AND article_id = ANY($2::bigint[]) + AND status = 'pending' + RETURNING article_version_id + ), + updated_versions AS ( + UPDATE article_versions av + SET manual_review_status = 'none', + manual_review_id = NULL, + manual_review_updated_at = NOW() + FROM affected_reviews r + WHERE av.id = r.article_version_id + RETURNING av.id + ) + SELECT COUNT(*)::bigint FROM affected_reviews + `, c.TenantID, articleIDs).Scan(&stats.ComplianceManualReviewCount); err != nil { + return stats, err + } + + return stats, nil +} + +func (w *BrandAssetCleanupWorker) cleanupMonitoringBrand(ctx context.Context, c brandCleanupCandidate, articleIDs []int64) (monitoringCleanupResult, error) { + result := monitoringCleanupResult{} + if w == nil || w.monitoringPool == nil { + return result, nil + } + + conn, err := w.monitoringPool.Acquire(ctx) + if err != nil { + return result, fmt.Errorf("acquire monitoring connection: %w", err) + } + defer conn.Release() + + monitorTaskIDs, err := w.loadMonitoringTaskIDs(ctx, conn, c) + if err != nil { + return result, err + } + result.MonitorTaskIDs = monitorTaskIDs + + tx, err := conn.Begin(ctx) + if err != nil { + return result, err + } + committed := false + defer func() { + if !committed { + _ = tx.Rollback(context.Background()) + } + }() + + count, err := execRows(ctx, tx, ` + DELETE FROM monitoring_collect_dispatch_outbox + WHERE task_id IN ( + SELECT id + FROM monitoring_collect_tasks + WHERE tenant_id = $1 AND brand_id = $2 + ) + OR request_id IN ( + SELECT request_id + FROM monitoring_collect_requests + WHERE tenant_id = $1 AND brand_id = $2 + ) + `, c.TenantID, c.BrandID) + if err != nil { + return result, err + } + result.Stats.MonitoringCollectOutboxCount += count + + count, err = execRows(ctx, tx, `DELETE FROM monitoring_collect_requests WHERE tenant_id = $1 AND brand_id = $2`, c.TenantID, c.BrandID) + if err != nil { + return result, err + } + result.Stats.MonitoringCollectRequestCount += count + + count, err = execRows(ctx, tx, `DELETE FROM monitoring_citation_facts WHERE tenant_id = $1 AND brand_id = $2`, c.TenantID, c.BrandID) + if err != nil { + return result, err + } + result.Stats.MonitoringCitationFactCount += count + + count, err = execRows(ctx, tx, `DELETE FROM question_monitor_parse_results WHERE tenant_id = $1 AND brand_id = $2`, c.TenantID, c.BrandID) + if err != nil { + return result, err + } + result.Stats.MonitoringParseResultCount += count + + count, err = execRows(ctx, tx, `DELETE FROM question_monitor_runs WHERE tenant_id = $1 AND brand_id = $2`, c.TenantID, c.BrandID) + if err != nil { + return result, err + } + result.Stats.MonitoringRunCount += count + + count, err = execRows(ctx, tx, `DELETE FROM monitoring_collect_tasks WHERE tenant_id = $1 AND brand_id = $2`, c.TenantID, c.BrandID) + if err != nil { + return result, err + } + result.Stats.MonitoringCollectTaskCount += count + + count, err = execRows(ctx, tx, `DELETE FROM monitoring_brand_platform_daily WHERE tenant_id = $1 AND brand_id = $2`, c.TenantID, c.BrandID) + if err != nil { + return result, err + } + result.Stats.MonitoringBrandPlatformCount += count + + count, err = execRows(ctx, tx, `DELETE FROM monitoring_brand_daily WHERE tenant_id = $1 AND brand_id = $2`, c.TenantID, c.BrandID) + if err != nil { + return result, err + } + result.Stats.MonitoringBrandDailyCount += count + + count, err = execRows(ctx, tx, `DELETE FROM monitoring_question_config_snapshots WHERE tenant_id = $1 AND brand_id = $2`, c.TenantID, c.BrandID) + if err != nil { + return result, err + } + result.Stats.MonitoringSnapshotCount += count + + count, err = execRows(ctx, tx, ` + DELETE FROM monitoring_article_url_aliases + WHERE tenant_id = $1 + AND article_id = ANY($2::bigint[]) + `, c.TenantID, nullableInt64Array(articleIDs)) + if err != nil { + return result, err + } + result.Stats.MonitoringArticleAliasCount += count + + count, err = execRows(ctx, tx, `DELETE FROM monitoring_user_marked_articles WHERE tenant_id = $1 AND brand_id = $2`, c.TenantID, c.BrandID) + if err != nil { + return result, err + } + result.Stats.MonitoringMarkedArticleCount += count + + if err := tx.Commit(ctx); err != nil { + return result, err + } + committed = true + return result, nil +} + +func (w *BrandAssetCleanupWorker) loadMonitoringTaskIDs(ctx context.Context, conn *pgxpool.Conn, c brandCleanupCandidate) ([]int64, error) { + rows, err := conn.Query(ctx, ` + SELECT id + FROM monitoring_collect_tasks + WHERE tenant_id = $1 + AND brand_id = $2 + `, c.TenantID, c.BrandID) + if err != nil { + return nil, err + } + defer rows.Close() + + ids := make([]int64, 0) + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} + +func execRows(ctx context.Context, tx pgx.Tx, sql string, args ...any) (int64, error) { + tag, err := tx.Exec(ctx, sql, args...) + if err != nil { + return 0, err + } + return tag.RowsAffected(), nil +} + +func nullableInt64Array(values []int64) any { + if len(values) == 0 { + return []int64{-1} + } + return values +} + +func (s *brandCleanupStats) add(other brandCleanupStats) { + s.BrandCount += other.BrandCount + s.ArticleCount += other.ArticleCount + s.GenerationTaskCount += other.GenerationTaskCount + s.GenerationTaskRecordCount += other.GenerationTaskRecordCount + s.KolUsageCount += other.KolUsageCount + s.QuotaReservationRefundCount += other.QuotaReservationRefundCount + s.CancelledPublishRecordCount += other.CancelledPublishRecordCount + s.PublishBatchCount += other.PublishBatchCount + s.DesktopPublishJobCount += other.DesktopPublishJobCount + s.DesktopPublishTaskCount += other.DesktopPublishTaskCount + s.DesktopMonitorTaskCount += other.DesktopMonitorTaskCount + s.PromptRuleCount += other.PromptRuleCount + s.PromptRuleGroupCount += other.PromptRuleGroupCount + s.ScheduleTaskCount += other.ScheduleTaskCount + s.KeywordCount += other.KeywordCount + s.QuestionCount += other.QuestionCount + s.CompetitorCount += other.CompetitorCount + s.ImageReferenceCount += other.ImageReferenceCount + s.MediaSupplyOrderCount += other.MediaSupplyOrderCount + s.MediaSupplyOrderItemCount += other.MediaSupplyOrderItemCount + s.ComplianceReviewJobCount += other.ComplianceReviewJobCount + s.ComplianceManualReviewCount += other.ComplianceManualReviewCount + s.MonitoringCollectRequestCount += other.MonitoringCollectRequestCount + s.MonitoringCollectOutboxCount += other.MonitoringCollectOutboxCount + s.MonitoringCollectTaskCount += other.MonitoringCollectTaskCount + s.MonitoringSnapshotCount += other.MonitoringSnapshotCount + s.MonitoringRunCount += other.MonitoringRunCount + s.MonitoringParseResultCount += other.MonitoringParseResultCount + s.MonitoringCitationFactCount += other.MonitoringCitationFactCount + s.MonitoringBrandDailyCount += other.MonitoringBrandDailyCount + s.MonitoringBrandPlatformCount += other.MonitoringBrandPlatformCount + s.MonitoringArticleAliasCount += other.MonitoringArticleAliasCount + s.MonitoringMarkedArticleCount += other.MonitoringMarkedArticleCount + s.MarkedDeletedBrandCount += other.MarkedDeletedBrandCount + s.CompletedBrandCount += other.CompletedBrandCount +} + +func (s brandCleanupStats) writeTo(out map[string]any) { + out["brand_count"] = s.BrandCount + out["article_count"] = s.ArticleCount + out["generation_task_count"] = s.GenerationTaskCount + out["generation_task_record_count"] = s.GenerationTaskRecordCount + out["kol_usage_count"] = s.KolUsageCount + out["quota_reservation_refund_count"] = s.QuotaReservationRefundCount + out["cancelled_publish_record_count"] = s.CancelledPublishRecordCount + out["publish_batch_count"] = s.PublishBatchCount + out["desktop_publish_job_count"] = s.DesktopPublishJobCount + out["desktop_publish_task_count"] = s.DesktopPublishTaskCount + out["desktop_monitor_task_count"] = s.DesktopMonitorTaskCount + out["prompt_rule_count"] = s.PromptRuleCount + out["prompt_rule_group_count"] = s.PromptRuleGroupCount + out["schedule_task_count"] = s.ScheduleTaskCount + out["keyword_count"] = s.KeywordCount + out["question_count"] = s.QuestionCount + out["competitor_count"] = s.CompetitorCount + out["image_reference_count"] = s.ImageReferenceCount + out["media_supply_order_count"] = s.MediaSupplyOrderCount + out["media_supply_order_item_count"] = s.MediaSupplyOrderItemCount + out["compliance_review_job_count"] = s.ComplianceReviewJobCount + out["compliance_manual_review_count"] = s.ComplianceManualReviewCount + out["monitoring_collect_request_count"] = s.MonitoringCollectRequestCount + out["monitoring_collect_outbox_count"] = s.MonitoringCollectOutboxCount + out["monitoring_collect_task_count"] = s.MonitoringCollectTaskCount + out["monitoring_snapshot_count"] = s.MonitoringSnapshotCount + out["monitoring_run_count"] = s.MonitoringRunCount + out["monitoring_parse_result_count"] = s.MonitoringParseResultCount + out["monitoring_citation_fact_count"] = s.MonitoringCitationFactCount + out["monitoring_brand_daily_count"] = s.MonitoringBrandDailyCount + out["monitoring_brand_platform_count"] = s.MonitoringBrandPlatformCount + out["monitoring_article_alias_count"] = s.MonitoringArticleAliasCount + out["monitoring_marked_article_count"] = s.MonitoringMarkedArticleCount + out["completed_brand_count"] = s.CompletedBrandCount +} + +func (w *BrandAssetCleanupWorker) warn(message string, c brandCleanupCandidate, err error) { + if w == nil || w.logger == nil { + return + } + w.logger.Warn(message, + zap.Int64("tenant_id", c.TenantID), + zap.Int64("brand_id", c.BrandID), + zap.Error(err), + ) +} diff --git a/server/internal/scheduler/brand_asset_cleanup_worker_test.go b/server/internal/scheduler/brand_asset_cleanup_worker_test.go new file mode 100644 index 0000000..c32ed85 --- /dev/null +++ b/server/internal/scheduler/brand_asset_cleanup_worker_test.go @@ -0,0 +1,43 @@ +package scheduler + +import ( + "context" + "testing" +) + +func TestBrandAssetCleanupRunOnceSkipsWhenStoreMissing(t *testing.T) { + stats, err := (*BrandAssetCleanupWorker)(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 TestBrandCleanupStatsWriteToIncludesTerminalMetrics(t *testing.T) { + stats := brandCleanupStats{ + BrandCount: 1, + ArticleCount: 2, + MonitoringCollectTaskCount: 3, + MonitoringCitationFactCount: 4, + CompletedBrandCount: 1, + } + out := map[string]any{} + stats.writeTo(out) + + for key, want := range map[string]any{ + "brand_count": int64(1), + "article_count": int64(2), + "monitoring_collect_task_count": int64(3), + "monitoring_citation_fact_count": int64(4), + "completed_brand_count": int64(1), + } { + if got := out[key]; got != want { + t.Fatalf("out[%s] = %v, want %v", key, got, want) + } + } +} diff --git a/server/internal/shared/swagger/descriptions.go b/server/internal/shared/swagger/descriptions.go index 71e432b..89318a4 100644 --- a/server/internal/shared/swagger/descriptions.go +++ b/server/internal/shared/swagger/descriptions.go @@ -197,7 +197,8 @@ var routeDocs = map[string]routeDoc{ "POST /api/tenant/tasks/:id/reconcile": {"对账发布任务", "管理后台对发布任务做对账：状态校正、结果回填。"}, "POST /api/tenant/tasks/:id/cancel": {"取消发布任务（后台）", "管理员在后台取消还未派发或还在执行的发布任务。"}, "GET /api/tenant/publish-tasks": {"发布任务列表（后台）", "管理后台分页查询本 Workspace 下的发布队列与历史发送结果。"}, - "GET /api/tenant/publish-records": {"发布记录列表（后台）", "管理后台分页查询当前品牌的发布中记录和历史发布记录。"}, + "GET /api/tenant/publish-records": {"发布记录列表（后台）", "管理后台分页查询当前用户的发布中记录和历史发布记录；已删除品牌会返回删除标记。"}, + "DELETE /api/tenant/publish-records/:id": {"删除发布记录（后台）", "管理后台软删除当前用户发起的发布记录，并取消关联的等待发布任务。"}, "POST /api/tenant/publish-tasks/:id/retry": {"重试发布任务（后台）", "管理后台对失败的发布任务重新创建一次发布任务。"}, "POST /api/tenant/publish-jobs": {"批量创建发布任务", "为一批文章/账号批量生成发布任务，进入派发队列。"}, @@ -327,7 +328,7 @@ var routeDocs = map[string]routeDoc{ "POST /api/tenant/brands": {"新建品牌", "创建一个新品牌。"}, "GET /api/tenant/brands/:id": {"品牌详情", "返回品牌基础信息及统计。"}, "PUT /api/tenant/brands/:id": {"更新品牌", "更新品牌名称、行业、官网等。"}, - "DELETE /api/tenant/brands/:id": {"删除品牌", "删除品牌及其下属关键词、问题、竞品。"}, + "DELETE /api/tenant/brands/:id": {"删除品牌", "标记品牌删除并由后台任务静默清理其文章、词库、竞品和采集数据；发布记录保留在用户发文总账。"}, "GET /api/tenant/brands/:id/keywords": {"品牌关键词列表", "返回某品牌下的关键词。"}, "POST /api/tenant/brands/:id/keywords": {"新增关键词", "为某品牌添加关键词。"}, diff --git a/server/internal/tenant/app/article_service.go b/server/internal/tenant/app/article_service.go index 1f2ddbd..4fa6a8c 100644 --- a/server/internal/tenant/app/article_service.go +++ b/server/internal/tenant/app/article_service.go @@ -122,6 +122,7 @@ const articlePublishStatusAggregateJoin = ` FROM publish_records pr WHERE pr.tenant_id = a.tenant_id AND pr.article_id = a.id + AND pr.deleted_at IS NULL ORDER BY COALESCE(pr.target_type, 'platform_account'), COALESCE(pr.platform_account_id, pr.target_connection_id), diff --git a/server/internal/tenant/app/brand_service.go b/server/internal/tenant/app/brand_service.go index 84b4b57..bb88d9a 100644 --- a/server/internal/tenant/app/brand_service.go +++ b/server/internal/tenant/app/brand_service.go @@ -278,19 +278,38 @@ func (s *BrandService) Delete(ctx context.Context, id int64) error { _ = tx.Rollback(ctx) }() - tag, err := tx.Exec(ctx, `UPDATE brands SET deleted_at = NOW(), updated_at = NOW() WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL`, id, actor.TenantID) - if err != nil || tag.RowsAffected() == 0 { - return response.ErrNotFound(40420, "brand_not_found", "brand not found") + tag, err := tx.Exec(ctx, ` + UPDATE brands + SET status = 'deleting', + deleted_at = COALESCE(deleted_at, NOW()), + updated_at = NOW() + WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL + `, id, actor.TenantID) + if err != nil { + return response.ErrInternal(50010, "delete_failed", "failed to mark brand for deletion") + } + if tag.RowsAffected() == 0 { + var alreadyDeleted bool + if lookupErr := tx.QueryRow(ctx, ` + SELECT deleted_at IS NOT NULL + FROM brands + WHERE id = $1 AND tenant_id = $2 + `, id, actor.TenantID).Scan(&alreadyDeleted); lookupErr != nil { + if errors.Is(lookupErr, pgx.ErrNoRows) { + return response.ErrNotFound(40420, "brand_not_found", "brand not found") + } + return response.ErrInternal(50010, "delete_failed", "failed to inspect brand deletion state") + } + if !alreadyDeleted { + return response.ErrNotFound(40420, "brand_not_found", "brand not found") + } + return nil } _, _ = tx.Exec(ctx, `UPDATE brand_keywords 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 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 := s.cleanupMonitoringAfterBrandDelete(ctx, actor.TenantID, id); err != nil { - return err - } - - afterJSON, _ := json.Marshal(map[string]interface{}{"id": id, "action": "cascade_soft_delete"}) + 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/internal/tenant/app/cache_support.go b/server/internal/tenant/app/cache_support.go index cf7802c..0dda130 100644 --- a/server/internal/tenant/app/cache_support.go +++ b/server/internal/tenant/app/cache_support.go @@ -158,6 +158,13 @@ func invalidateBrandCaches(ctx context.Context, c sharedcache.Cache, tenantID, b invalidateWorkspaceCaches(ctx, c, tenantID) } +func InvalidateBrandDeletionCaches(ctx context.Context, c sharedcache.Cache, tenantID, brandID int64) { + invalidateBrandCaches(ctx, c, tenantID, brandID) + invalidateArticleCaches(ctx, c, tenantID, nil) + invalidatePromptRuleCaches(ctx, c, tenantID, brandID, nil) + invalidateScheduleTaskCaches(ctx, c, tenantID, nil) +} + func promptRuleGroupsCacheKey(tenantID, brandID int64) string { return fmt.Sprintf("prompt_rule:groups:%d:%d", tenantID, brandID) } diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index 1691fca..b2cc652 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/server/internal/tenant/app/desktop_task_service.go @@ -973,7 +973,10 @@ func (s *DesktopTaskService) Cancel(ctx context.Context, client *repository.Desk if req.LeaseToken != nil && strings.TrimSpace(*req.LeaseToken) != "" { task, err = taskRepo.CancelByLease(ctx, desktopID, HashDesktopClientToken(strings.TrimSpace(*req.LeaseToken)), errorJSON) } else { - task, err = taskRepo.CancelByClient(ctx, desktopID, client.ID, errorJSON) + task, err = taskRepo.CancelQueuedPublishByOwner(ctx, desktopID, client.TenantID, client.WorkspaceID, client.UserID, errorJSON) + if errors.Is(err, pgx.ErrNoRows) { + task, err = taskRepo.CancelByClient(ctx, desktopID, client.ID, errorJSON) + } } if err != nil { if errors.Is(err, pgx.ErrNoRows) { @@ -1015,7 +1018,8 @@ func (s *DesktopTaskService) Cancel(ctx context.Context, client *repository.Desk } func (s *DesktopTaskService) TenantCancel(ctx context.Context, actor auth.Actor, desktopID uuid.UUID, req CancelDesktopTaskRequest) (*DesktopTaskView, error) { - if actor.PrimaryWorkspaceID == 0 { + workspaceID := auth.CurrentWorkspaceID(ctx) + if actor.TenantID == 0 || actor.UserID == 0 || workspaceID == 0 { return nil, response.ErrUnauthorized(40132, "workspace_scope_missing", "workspace context is required") } @@ -1027,7 +1031,14 @@ func (s *DesktopTaskService) TenantCancel(ctx context.Context, actor auth.Actor, return nil, response.ErrBadRequest(40088, "invalid_desktop_task_cancel", "cancel payload must be serializable") } - task, err := s.repo.TenantCancel(ctx, desktopID, actor.PrimaryWorkspaceID, errorJSON) + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, response.ErrInternal(50107, "desktop_task_cancel_begin_failed", "failed to start desktop task cancel") + } + defer tx.Rollback(ctx) + + taskRepo := repository.NewDesktopTaskRepository(tx) + task, err := taskRepo.TenantCancel(ctx, desktopID, actor.TenantID, workspaceID, actor.UserID, errorJSON) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, response.ErrConflict(40984, "desktop_task_cancel_conflict", "desktop task cannot be canceled in its current state") @@ -1035,7 +1046,17 @@ func (s *DesktopTaskService) TenantCancel(ctx context.Context, actor auth.Actor, return nil, response.ErrInternal(50093, "desktop_task_tenant_cancel_failed", "failed to cancel desktop task") } + publishOutcome, err := syncDesktopPublishTaskState(ctx, tx, task) + if err != nil { + return nil, err + } + + if err := tx.Commit(ctx); err != nil { + return nil, response.ErrInternal(50108, "desktop_task_cancel_commit_failed", "failed to commit desktop task cancel") + } + s.publishTaskEvent(ctx, task, "task_canceled") + s.afterRecoveredPublishOutcomes(ctx, []*desktopPublishSyncOutcome{publishOutcome}) view := buildDesktopTaskView(task) return &view, nil diff --git a/server/internal/tenant/app/enterprise_site_service.go b/server/internal/tenant/app/enterprise_site_service.go index 92ec306..70a1b06 100644 --- a/server/internal/tenant/app/enterprise_site_service.go +++ b/server/internal/tenant/app/enterprise_site_service.go @@ -269,6 +269,7 @@ func (s *EnterpriseSiteService) List(ctx context.Context) ([]EnterpriseSiteConne WHERE pr.tenant_id = $1 AND pr.target_type = 'enterprise_site' AND pr.target_connection_id IS NOT NULL + AND pr.deleted_at IS NULL ORDER BY pr.target_connection_id, pr.created_at DESC, pr.id DESC ) SELECT @@ -1022,7 +1023,7 @@ func (s *EnterpriseSiteService) finishEnterprisePublishRecord( published_at = $5, error_message = $6, updated_at = NOW() - WHERE id = $7 AND tenant_id = $8 + WHERE id = $7 AND tenant_id = $8 AND deleted_at IS NULL `, status, externalID, externalURL, nullableJSON(responsePayload), publishedAt, errorMessage, publishRecordID, tenantID); err != nil { return response.ErrInternal(50228, "enterprise_site_publish_record_update_failed", "企业站点发布结果更新失败") } diff --git a/server/internal/tenant/app/media_service.go b/server/internal/tenant/app/media_service.go index e8ddb79..de9d213 100644 --- a/server/internal/tenant/app/media_service.go +++ b/server/internal/tenant/app/media_service.go @@ -4,9 +4,11 @@ import ( "context" "crypto/rand" "encoding/hex" + "errors" "strings" "time" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "github.com/geo-platform/tenant-api/internal/shared/auth" @@ -38,6 +40,9 @@ type PublishRecordResponse struct { ID int64 `json:"id"` PublishBatchID int64 `json:"publish_batch_id"` ArticleID int64 `json:"article_id"` + BrandID int64 `json:"brand_id"` + BrandName *string `json:"brand_name"` + BrandDeleted bool `json:"brand_deleted"` ArticleTitle *string `json:"article_title,omitempty"` PlatformAccountID *int64 `json:"platform_account_id"` DesktopAccountID *string `json:"desktop_account_id"` @@ -72,6 +77,10 @@ type PublishRecordListResponse struct { HistoryTotal int `json:"history_total"` } +type DeletePublishRecordResponse struct { + Deleted bool `json:"deleted"` +} + func (s *MediaService) ListPlatforms(ctx context.Context) ([]MediaPlatformResponse, error) { rows, err := s.pool.Query(ctx, ` SELECT id, platform_id, name, category, short_name, accent_color, login_url, logo_url, status, sort_order @@ -108,9 +117,8 @@ func (s *MediaService) ListPlatforms(ctx context.Context) ([]MediaPlatformRespon func (s *MediaService) ListPublishRecords(ctx context.Context, req ListPublishRecordsRequest) (*PublishRecordListResponse, error) { actor := auth.MustActor(ctx) - brandID, err := requireCurrentBrandID(ctx) - if err != nil { - return nil, err + if actor.TenantID == 0 || actor.UserID == 0 { + return nil, response.ErrUnauthorized(40132, "workspace_scope_missing", "workspace context is required") } page := req.Page @@ -125,11 +133,11 @@ func (s *MediaService) ListPublishRecords(ctx context.Context, req ListPublishRe pendingStatuses := []string{"queued", "publishing"} historyStatuses := []string{"success", "failed", "cancelled", "canceled"} - pendingItems, err := s.listPublishRecordsByStatuses(ctx, actor.TenantID, brandID, pendingStatuses, title, 0, 0, "pr.created_at ASC, pr.id ASC") + pendingItems, err := s.listPublishRecordsByStatuses(ctx, actor.TenantID, actor.UserID, pendingStatuses, title, 0, 0, "pr.created_at ASC, pr.id ASC") if err != nil { return nil, err } - historyTotal, err := s.countPublishRecordsByStatuses(ctx, actor.TenantID, brandID, historyStatuses, title) + historyTotal, err := s.countPublishRecordsByStatuses(ctx, actor.TenantID, actor.UserID, historyStatuses, title) if err != nil { return nil, err } @@ -152,7 +160,7 @@ func (s *MediaService) ListPublishRecords(ctx context.Context, req ListPublishRe historyOffset = 0 } if remaining > 0 && historyOffset < historyTotal { - historyItems, listErr := s.listPublishRecordsByStatuses(ctx, actor.TenantID, brandID, historyStatuses, title, remaining, historyOffset, "pr.updated_at DESC, pr.id DESC") + historyItems, listErr := s.listPublishRecordsByStatuses(ctx, actor.TenantID, actor.UserID, historyStatuses, title, remaining, historyOffset, "pr.updated_at DESC, pr.id DESC") if listErr != nil { return nil, listErr } @@ -172,7 +180,7 @@ func (s *MediaService) ListPublishRecords(ctx context.Context, req ListPublishRe func (s *MediaService) listPublishRecordsByStatuses( ctx context.Context, tenantID int64, - brandID int64, + userID int64, statuses []string, title string, limit int, @@ -181,6 +189,7 @@ func (s *MediaService) listPublishRecordsByStatuses( ) ([]PublishRecordResponse, error) { query := ` SELECT pr.id, pr.publish_batch_id, pr.article_id, + a.brand_id, b.name AS brand_name, (b.id IS NULL OR b.deleted_at IS NOT NULL OR b.status IN ('deleting', 'deleted')) AS brand_deleted, COALESCE(av.title, NULLIF(a.wizard_state_json ->> 'title', '')) AS article_title, pr.platform_account_id, pa.desktop_id::text, dt.desktop_id::text, COALESCE(pr.target_type, 'platform_account'), pr.target_connection_id, pr.platform_id, @@ -194,6 +203,7 @@ func (s *MediaService) listPublishRecordsByStatuses( pr.external_manage_url, pr.published_at, pr.error_message, pr.created_at, pr.updated_at FROM publish_records pr JOIN articles a ON a.id = pr.article_id AND a.tenant_id = pr.tenant_id + LEFT JOIN brands b ON b.id = a.brand_id AND b.tenant_id = a.tenant_id LEFT JOIN article_versions av ON av.id = a.current_version_id AND av.article_id = a.id LEFT JOIN media_platforms mp ON mp.platform_id = pr.platform_id LEFT JOIN platform_accounts pa ON pa.id = pr.platform_account_id @@ -210,15 +220,21 @@ func (s *MediaService) listPublishRecordsByStatuses( LIMIT 1 ) dt ON TRUE WHERE pr.tenant_id = $1 - AND a.brand_id = $2 - AND a.deleted_at IS NULL + AND EXISTS ( + SELECT 1 + FROM publish_batches pb + WHERE pb.id = pr.publish_batch_id + AND pb.tenant_id = pr.tenant_id + AND pb.initiator_user_id = $2 + ) + AND pr.deleted_at IS NULL AND pr.status = ANY($3) AND ( $4 = '' OR COALESCE(av.title, NULLIF(a.wizard_state_json ->> 'title', ''), '') ILIKE '%' || $4 || '%' ) ` - args := []any{tenantID, brandID, statuses, title} + args := []any{tenantID, userID, statuses, title} if strings.TrimSpace(orderBy) != "" { query += "\nORDER BY " + orderBy @@ -243,6 +259,9 @@ func (s *MediaService) listPublishRecordsByStatuses( &item.ID, &item.PublishBatchID, &item.ArticleID, + &item.BrandID, + &item.BrandName, + &item.BrandDeleted, &item.ArticleTitle, &item.PlatformAccountID, &desktopAccountID, @@ -281,7 +300,7 @@ func (s *MediaService) listPublishRecordsByStatuses( func (s *MediaService) countPublishRecordsByStatuses( ctx context.Context, tenantID int64, - brandID int64, + userID int64, statuses []string, title string, ) (int, error) { @@ -292,14 +311,20 @@ func (s *MediaService) countPublishRecordsByStatuses( JOIN articles a ON a.id = pr.article_id AND a.tenant_id = pr.tenant_id LEFT JOIN article_versions av ON av.id = a.current_version_id AND av.article_id = a.id WHERE pr.tenant_id = $1 - AND a.brand_id = $2 - AND a.deleted_at IS NULL + AND EXISTS ( + SELECT 1 + FROM publish_batches pb + WHERE pb.id = pr.publish_batch_id + AND pb.tenant_id = pr.tenant_id + AND pb.initiator_user_id = $2 + ) + AND pr.deleted_at IS NULL AND pr.status = ANY($3) AND ( $4 = '' OR COALESCE(av.title, NULLIF(a.wizard_state_json ->> 'title', ''), '') ILIKE '%' || $4 || '%' ) - `, tenantID, brandID, statuses, title).Scan(&count) + `, tenantID, userID, statuses, title).Scan(&count) if err != nil { return 0, response.ErrInternal(50043, "publish_record_query_failed", "failed to list publish records") } @@ -317,7 +342,9 @@ func (s *MediaService) ListArticlePublishRecords(ctx context.Context, articleID func (s *MediaService) listArticlePublishRecords(ctx context.Context, tenantID, brandID, articleID int64) ([]PublishRecordResponse, error) { rows, err := s.pool.Query(ctx, ` - SELECT pr.id, pr.publish_batch_id, pr.article_id, pr.platform_account_id, pa.desktop_id::text, + SELECT pr.id, pr.publish_batch_id, pr.article_id, + a.brand_id, b.name AS brand_name, (b.id IS NULL OR b.deleted_at IS NOT NULL OR b.status IN ('deleting', 'deleted')) AS brand_deleted, + pr.platform_account_id, pa.desktop_id::text, COALESCE(pr.target_type, 'platform_account'), pr.target_connection_id, pr.platform_id, COALESCE(mp.name, esc.site_name, esc.name, pr.platform_id) AS platform_name, CASE @@ -334,7 +361,8 @@ func (s *MediaService) listArticlePublishRecords(ctx context.Context, tenantID, ON esc.id = pr.target_connection_id AND esc.tenant_id = pr.tenant_id JOIN articles a ON a.id = pr.article_id AND a.tenant_id = pr.tenant_id - WHERE pr.tenant_id = $1 AND pr.article_id = $2 AND a.brand_id = $3 AND a.deleted_at IS NULL + LEFT JOIN brands b ON b.id = a.brand_id AND b.tenant_id = a.tenant_id + WHERE pr.tenant_id = $1 AND pr.article_id = $2 AND a.brand_id = $3 AND a.deleted_at IS NULL AND pr.deleted_at IS NULL ORDER BY pr.created_at DESC, pr.id DESC `, tenantID, articleID, brandID) if err != nil { @@ -350,6 +378,9 @@ func (s *MediaService) listArticlePublishRecords(ctx context.Context, tenantID, &item.ID, &item.PublishBatchID, &item.ArticleID, + &item.BrandID, + &item.BrandName, + &item.BrandDeleted, &item.PlatformAccountID, &desktopAccountID, &item.TargetType, @@ -377,6 +408,112 @@ func (s *MediaService) listArticlePublishRecords(ctx context.Context, tenantID, return items, nil } +func (s *MediaService) DeletePublishRecord(ctx context.Context, recordID int64) (*DeletePublishRecordResponse, error) { + if recordID <= 0 { + return nil, response.ErrBadRequest(40001, "invalid_publish_record_id", "publish record id must be a positive number") + } + actor := auth.MustActor(ctx) + if actor.TenantID == 0 || actor.UserID == 0 { + return nil, response.ErrUnauthorized(40132, "workspace_scope_missing", "workspace context is required") + } + + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, response.ErrInternal(50044, "publish_record_delete_begin_failed", "failed to delete publish record") + } + defer tx.Rollback(ctx) + + var publishBatchID int64 + var articleID int64 + var cancelErrorJSON []byte + cancelErrorJSON, err = marshalOptionalJSON(map[string]any{ + "reason": "publish_record_deleted", + "source": "tenant_web", + }) + if err != nil { + return nil, response.ErrInternal(50044, "publish_record_delete_payload_failed", "failed to delete publish record") + } + err = tx.QueryRow(ctx, ` + SELECT pr.publish_batch_id, pr.article_id + FROM publish_records pr + JOIN articles a + ON a.id = pr.article_id + AND a.tenant_id = pr.tenant_id + JOIN publish_batches pb + ON pb.id = pr.publish_batch_id + AND pb.tenant_id = pr.tenant_id + WHERE pr.id = $1 + AND pr.tenant_id = $2 + AND pb.initiator_user_id = $3 + AND pr.deleted_at IS NULL + FOR UPDATE OF pr + `, recordID, actor.TenantID, actor.UserID).Scan(&publishBatchID, &articleID) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, response.ErrNotFound(40435, "publish_record_not_found", "publish record not found") + } + return nil, response.ErrInternal(50044, "publish_record_delete_lookup_failed", "failed to delete publish record") + } + + if tag, err := tx.Exec(ctx, ` + UPDATE publish_records + SET deleted_at = NOW(), + updated_at = NOW() + WHERE id = $1 + AND tenant_id = $2 + AND deleted_at IS NULL + `, recordID, actor.TenantID); err != nil { + return nil, response.ErrInternal(50044, "publish_record_delete_failed", "failed to delete publish record") + } else if tag.RowsAffected() == 0 { + return nil, response.ErrNotFound(40435, "publish_record_not_found", "publish record not found") + } + + if _, err := tx.Exec(ctx, ` + UPDATE desktop_tasks + SET status = 'aborted', + error = $3, + active_attempt_id = NULL, + lease_token_hash = NULL, + lease_expires_at = NULL, + updated_at = NOW() + WHERE tenant_id = $1 + AND kind = 'publish' + AND status = 'queued' + AND payload->>'publish_record_id' = $2 + `, actor.TenantID, recordID, cancelErrorJSON); err != nil { + return nil, response.ErrInternal(50044, "publish_record_delete_cancel_task_failed", "failed to cancel queued publish task before deleting record") + } + + batchStatus, err := recalculatePublishBatchStatus(ctx, tx, publishBatchID) + if err != nil { + return nil, err + } + if _, err := tx.Exec(ctx, ` + UPDATE publish_batches + SET status = $1, updated_at = NOW() + WHERE id = $2 AND tenant_id = $3 + `, batchStatus, publishBatchID, actor.TenantID); err != nil { + return nil, response.ErrInternal(50054, "publish_batch_update_failed", "failed to update publish batch") + } + + articleStatus, err := recalculateArticlePublishStatus(ctx, tx, actor.TenantID, articleID) + if err != nil { + return nil, err + } + if _, err := tx.Exec(ctx, ` + UPDATE articles + SET publish_status = $1, updated_at = NOW() + WHERE id = $2 AND tenant_id = $3 AND deleted_at IS NULL + `, articleStatus, articleID, actor.TenantID); err != nil { + return nil, response.ErrInternal(50055, "article_publish_update_failed", "failed to update article publish status") + } + + if err := tx.Commit(ctx); err != nil { + return nil, response.ErrInternal(50044, "publish_record_delete_commit_failed", "failed to delete publish record") + } + return &DeletePublishRecordResponse{Deleted: true}, nil +} + func publishBatchRequiresCover(accounts []platformAccountSeed) bool { for _, account := range accounts { if publishPlatformRequiresCover(account.PlatformID) { diff --git a/server/internal/tenant/app/publish_job_service.go b/server/internal/tenant/app/publish_job_service.go index cf6e708..08087ab 100644 --- a/server/internal/tenant/app/publish_job_service.go +++ b/server/internal/tenant/app/publish_job_service.go @@ -566,6 +566,7 @@ func loadExistingPublishRecordTargets( AND pa.workspace_id = $2 AND pr.article_id = $3 AND pr.platform_account_id = ANY($4::bigint[]) + AND pr.deleted_at IS NULL AND pr.status IN ('queued', 'publishing', 'success') AND dt.status IN ('queued', 'in_progress', 'succeeded', 'unknown') ORDER BY pr.platform_account_id, @@ -731,13 +732,18 @@ func (s *PublishJobService) RetryTenantDesktopTask( if err != nil { return nil, err } + brandID, err := s.resolveRetryArticleBrandID(ctx, actor.TenantID, articleID) + if err != nil { + return nil, err + } title := strings.TrimSpace(stringPointerValue(extractStringPointer(payload, "title"))) if title == "" { title = "文章发布" } - return s.Create(ctx, auth.Actor{ + publishCtx := auth.WithCurrentBrandID(ctx, brandID) + return s.Create(publishCtx, auth.Actor{ TenantID: actor.TenantID, UserID: actor.UserID, PrimaryWorkspaceID: workspaceID, @@ -895,7 +901,7 @@ func (s *PublishJobService) resolveRetryArticleID(ctx context.Context, tenantID err := s.pool.QueryRow(ctx, ` SELECT article_id FROM publish_records - WHERE id = $1 AND tenant_id = $2 + WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL `, publishRecordID, tenantID).Scan(&articleID) if err == nil && articleID > 0 { return articleID, nil diff --git a/server/internal/tenant/app/publish_record_support.go b/server/internal/tenant/app/publish_record_support.go index 126cad3..c84db2c 100644 --- a/server/internal/tenant/app/publish_record_support.go +++ b/server/internal/tenant/app/publish_record_support.go @@ -138,6 +138,7 @@ func cancelQueuedPublishTasksForUnavailableArticle(ctx context.Context, tx pgx.T AND pr.tenant_id = $1 AND pr.article_id = $2 AND pr.status IN ('queued', 'publishing', 'pending', 'running') + AND pr.deleted_at IS NULL AND j.tenant_id = $1 AND j.article_id = $2 AND j.status = 'queued' @@ -189,23 +190,27 @@ func cancelQueuedPublishTasksForUnavailableArticle(ctx context.Context, tx pgx.T SELECT 1 FROM publish_records pr WHERE pr.publish_batch_id = pb.id + AND pr.deleted_at IS NULL AND pr.status IN ('queued', 'publishing', 'pending', 'running') ) THEN 'publishing' WHEN EXISTS ( SELECT 1 FROM publish_records pr WHERE pr.publish_batch_id = pb.id + AND pr.deleted_at IS NULL AND pr.status IN ('success', 'published', 'publish_success') ) AND EXISTS ( SELECT 1 FROM publish_records pr WHERE pr.publish_batch_id = pb.id + AND pr.deleted_at IS NULL AND pr.status NOT IN ('success', 'published', 'publish_success') ) THEN 'partial_success' WHEN EXISTS ( SELECT 1 FROM publish_records pr WHERE pr.publish_batch_id = pb.id + AND pr.deleted_at IS NULL AND pr.status IN ('success', 'published', 'publish_success') ) THEN 'success' ELSE 'failed' @@ -223,6 +228,7 @@ func cancelQueuedPublishTasksForUnavailableArticle(ctx context.Context, tx pgx.T FROM publish_records pr WHERE pr.article_id = a.id AND pr.tenant_id = a.tenant_id + AND pr.deleted_at IS NULL AND pr.status IN ('queued', 'publishing', 'pending', 'running') ) THEN 'publishing' WHEN EXISTS ( @@ -230,12 +236,14 @@ func cancelQueuedPublishTasksForUnavailableArticle(ctx context.Context, tx pgx.T FROM publish_records pr WHERE pr.article_id = a.id AND pr.tenant_id = a.tenant_id + AND pr.deleted_at IS NULL AND pr.status IN ('success', 'published', 'publish_success') ) AND EXISTS ( SELECT 1 FROM publish_records pr WHERE pr.article_id = a.id AND pr.tenant_id = a.tenant_id + AND pr.deleted_at IS NULL AND pr.status NOT IN ('success', 'published', 'publish_success') ) THEN 'partial_success' WHEN EXISTS ( @@ -243,6 +251,7 @@ func cancelQueuedPublishTasksForUnavailableArticle(ctx context.Context, tx pgx.T FROM publish_records pr WHERE pr.article_id = a.id AND pr.tenant_id = a.tenant_id + AND pr.deleted_at IS NULL AND pr.status IN ('success', 'published', 'publish_success') ) THEN 'success' ELSE 'failed' @@ -296,6 +305,7 @@ func cancelQueuedPublishTasksForDesktopAccountUnbind( AND dt.kind = 'publish' AND dt.status = 'queued' AND pr.status IN ('queued', 'publishing', 'pending', 'running') + AND pr.deleted_at IS NULL `, tenantID, workspaceID, accountID) if err != nil { return 0, response.ErrInternal(50121, "publish_task_cancel_account_unbind_failed", "failed to inspect queued publish tasks for unbound account") @@ -335,6 +345,7 @@ func cancelQueuedPublishTasksForDesktopAccountUnbind( WHERE tenant_id = $1 AND id = ANY($2::bigint[]) AND status IN ('queued', 'publishing', 'pending', 'running') + AND deleted_at IS NULL `, tenantID, recordIDs); err != nil { return 0, response.ErrInternal(50121, "publish_task_cancel_account_unbind_failed", "failed to cancel publish records for unbound account") } @@ -412,7 +423,7 @@ func syncDesktopPublishTaskState( if err := tx.QueryRow(ctx, ` SELECT publish_batch_id, article_id FROM publish_records - WHERE id = $1 AND tenant_id = $2 + WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL `, publishRecordID, task.TenantID).Scan(&publishBatchID, &articleID); err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, response.ErrNotFound(40434, "publish_record_not_found", "publish record not found") @@ -451,7 +462,7 @@ func syncDesktopPublishTaskState( response_payload_json = $7, error_message = $8, updated_at = NOW() - WHERE id = $9 AND tenant_id = $10 + WHERE id = $9 AND tenant_id = $10 AND deleted_at IS NULL `, publishStatus, externalArticleID, externalArticleURL, @@ -567,7 +578,9 @@ func desktopTaskToPublishStatus(taskStatus string) string { return "failed" case "succeeded": return "success" - case "failed", "aborted": + case "aborted": + return "cancelled" + case "failed": return "failed" default: return "failed" @@ -602,6 +615,7 @@ func recalculatePublishBatchStatus(ctx context.Context, tx pgx.Tx, publishBatchI SELECT status FROM publish_records WHERE publish_batch_id = $1 + AND deleted_at IS NULL `, publishBatchID) if err != nil { return "", response.ErrInternal(50054, "publish_batch_status_query_failed", "failed to inspect publish batch status") @@ -621,7 +635,7 @@ func recalculatePublishBatchStatus(ctx context.Context, tx pgx.Tx, publishBatchI } if len(statuses) == 0 { - return "publishing", nil + return "failed", nil } counts := map[string]int{ @@ -650,6 +664,7 @@ func recalculateArticlePublishStatus(ctx context.Context, tx pgx.Tx, tenantID, a SELECT status FROM publish_records WHERE tenant_id = $1 AND article_id = $2 + AND deleted_at IS NULL `, tenantID, articleID) if err != nil { return "", response.ErrInternal(50055, "article_publish_status_query_failed", "failed to inspect article publish status") diff --git a/server/internal/tenant/app/publish_record_support_test.go b/server/internal/tenant/app/publish_record_support_test.go index 862b84a..128cc26 100644 --- a/server/internal/tenant/app/publish_record_support_test.go +++ b/server/internal/tenant/app/publish_record_support_test.go @@ -175,6 +175,24 @@ func TestPublishBatchStatusToArticleStatusKeepsPartialSuccess(t *testing.T) { } } +func TestRecalculatePublishBatchStatusDefaultsToFailed(t *testing.T) { + tx := &capturePublishDedupLookupTx{rows: emptyPublishDedupRows{}} + + got, err := recalculatePublishBatchStatus(context.Background(), tx, 1) + if err != nil { + t.Fatalf("recalculatePublishBatchStatus() error = %v", err) + } + if got != "failed" { + t.Fatalf("recalculatePublishBatchStatus() = %q, want failed", got) + } +} + +func TestDesktopTaskToPublishStatusMapsAbortToCancelled(t *testing.T) { + if got := desktopTaskToPublishStatus("aborted"); got != "cancelled" { + t.Fatalf("desktopTaskToPublishStatus(aborted) = %q, want cancelled", got) + } +} + func TestRecalculateArticlePublishStatusDefaultsToUnpublished(t *testing.T) { tx := &capturePublishDedupLookupTx{rows: emptyPublishDedupRows{}} diff --git a/server/internal/tenant/repository/desktop_task_repo.go b/server/internal/tenant/repository/desktop_task_repo.go index 1073d90..a9021c2 100644 --- a/server/internal/tenant/repository/desktop_task_repo.go +++ b/server/internal/tenant/repository/desktop_task_repo.go @@ -124,7 +124,8 @@ type DesktopTaskRepository interface { Complete(ctx context.Context, desktopID uuid.UUID, leaseTokenHash []byte, status string, resultJSON, errorJSON []byte) (*DesktopTask, error) CancelByLease(ctx context.Context, desktopID uuid.UUID, leaseTokenHash []byte, errorJSON []byte) (*DesktopTask, error) CancelByClient(ctx context.Context, desktopID, clientID uuid.UUID, errorJSON []byte) (*DesktopTask, error) - TenantCancel(ctx context.Context, desktopID uuid.UUID, workspaceID int64, errorJSON []byte) (*DesktopTask, error) + CancelQueuedPublishByOwner(ctx context.Context, desktopID uuid.UUID, tenantID, workspaceID, userID int64, errorJSON []byte) (*DesktopTask, error) + TenantCancel(ctx context.Context, desktopID uuid.UUID, tenantID, workspaceID, userID int64, errorJSON []byte) (*DesktopTask, error) Reconcile(ctx context.Context, desktopID uuid.UUID, workspaceID int64, status string, resultJSON, errorJSON []byte) (*DesktopTask, error) MarkUnknownByClient(ctx context.Context, clientID uuid.UUID, workspaceID int64, errorJSON []byte) (int64, error) CreateAttempt(ctx context.Context, params CreateDesktopTaskAttemptParams) (*DesktopTaskAttempt, error) @@ -263,11 +264,27 @@ func (r *desktopTaskRepository) CancelByClient(ctx context.Context, desktopID, c return desktopTaskFromGenerated(row), nil } -func (r *desktopTaskRepository) TenantCancel(ctx context.Context, desktopID uuid.UUID, workspaceID int64, errorJSON []byte) (*DesktopTask, error) { +func (r *desktopTaskRepository) CancelQueuedPublishByOwner(ctx context.Context, desktopID uuid.UUID, tenantID, workspaceID, userID int64, errorJSON []byte) (*DesktopTask, error) { + row, err := r.q.CancelQueuedPublishDesktopTaskByOwner(ctx, generated.CancelQueuedPublishDesktopTaskByOwnerParams{ + Error: errorJSON, + DesktopID: pgUUID(desktopID), + TenantID: tenantID, + WorkspaceID: workspaceID, + UserID: userID, + }) + if err != nil { + return nil, err + } + return desktopTaskFromGenerated(row), nil +} + +func (r *desktopTaskRepository) TenantCancel(ctx context.Context, desktopID uuid.UUID, tenantID, workspaceID, userID int64, errorJSON []byte) (*DesktopTask, error) { row, err := r.q.TenantCancelDesktopTask(ctx, generated.TenantCancelDesktopTaskParams{ Error: errorJSON, DesktopID: pgUUID(desktopID), + TenantID: tenantID, WorkspaceID: workspaceID, + UserID: userID, }) if err != nil { return nil, err diff --git a/server/internal/tenant/repository/generated/desktop_task.sql.go b/server/internal/tenant/repository/generated/desktop_task.sql.go index 66fb113..31478a2 100644 --- a/server/internal/tenant/repository/generated/desktop_task.sql.go +++ b/server/internal/tenant/repository/generated/desktop_task.sql.go @@ -135,6 +135,83 @@ func (q *Queries) CancelDesktopTaskByLease(ctx context.Context, arg CancelDeskto return i, err } +const cancelQueuedPublishDesktopTaskByOwner = `-- name: CancelQueuedPublishDesktopTaskByOwner :one +UPDATE desktop_tasks AS t +SET status = 'aborted', + error = $1, + active_attempt_id = NULL, + lease_token_hash = NULL, + lease_expires_at = NULL, + updated_at = now() +FROM desktop_publish_jobs AS j +WHERE t.desktop_id = $2 + AND t.tenant_id = $3 + AND t.workspace_id = $4 + AND t.kind = 'publish' + AND t.status = 'queued' + AND j.desktop_id = t.job_id + AND j.tenant_id = t.tenant_id + AND j.workspace_id = t.workspace_id + AND j.created_by_user_id = $5 +RETURNING t.id, t.desktop_id, t.job_id, t.tenant_id, t.workspace_id, t.target_account_id, t.target_client_id, t.platform_id, t.kind, t.payload, t.status, t.dedup_key, t.active_attempt_id, t.lease_token_hash, t.lease_expires_at, t.attempts, t.result, t.error, t.created_at, t.updated_at, t.priority, t.lane, t.lane_weight, t.source, t.scheduler_group_key, t.monitor_task_id, t.supersedes_task_id, t.control_flags, t.interrupt_generation, t.started_at, t.interrupted_at, t.interrupt_reason, t.enqueued_at, t.publish_submit_started_at +` + +type CancelQueuedPublishDesktopTaskByOwnerParams struct { + Error []byte `json:"error"` + DesktopID pgtype.UUID `json:"desktop_id"` + TenantID int64 `json:"tenant_id"` + WorkspaceID int64 `json:"workspace_id"` + UserID int64 `json:"user_id"` +} + +func (q *Queries) CancelQueuedPublishDesktopTaskByOwner(ctx context.Context, arg CancelQueuedPublishDesktopTaskByOwnerParams) (DesktopTask, error) { + row := q.db.QueryRow(ctx, cancelQueuedPublishDesktopTaskByOwner, + arg.Error, + arg.DesktopID, + arg.TenantID, + arg.WorkspaceID, + arg.UserID, + ) + var i DesktopTask + err := row.Scan( + &i.ID, + &i.DesktopID, + &i.JobID, + &i.TenantID, + &i.WorkspaceID, + &i.TargetAccountID, + &i.TargetClientID, + &i.PlatformID, + &i.Kind, + &i.Payload, + &i.Status, + &i.DedupKey, + &i.ActiveAttemptID, + &i.LeaseTokenHash, + &i.LeaseExpiresAt, + &i.Attempts, + &i.Result, + &i.Error, + &i.CreatedAt, + &i.UpdatedAt, + &i.Priority, + &i.Lane, + &i.LaneWeight, + &i.Source, + &i.SchedulerGroupKey, + &i.MonitorTaskID, + &i.SupersedesTaskID, + &i.ControlFlags, + &i.InterruptGeneration, + &i.StartedAt, + &i.InterruptedAt, + &i.InterruptReason, + &i.EnqueuedAt, + &i.PublishSubmitStartedAt, + ) + return i, err +} + const completeDesktopTask = `-- name: CompleteDesktopTask :one UPDATE desktop_tasks SET status = $1, @@ -845,19 +922,30 @@ SET status = 'aborted', lease_expires_at = NULL, updated_at = now() WHERE desktop_id = $2 - AND workspace_id = $3 + AND tenant_id = $3 + AND workspace_id = $4 AND status = 'queued' + AND EXISTS ( + SELECT 1 + FROM desktop_publish_jobs AS j + WHERE j.desktop_id = desktop_tasks.job_id + AND j.tenant_id = desktop_tasks.tenant_id + AND j.workspace_id = desktop_tasks.workspace_id + AND j.created_by_user_id = $5 + ) RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, result, error, created_at, updated_at, priority, lane, lane_weight, source, scheduler_group_key, monitor_task_id, supersedes_task_id, control_flags, interrupt_generation, started_at, interrupted_at, interrupt_reason, enqueued_at, publish_submit_started_at ` type TenantCancelDesktopTaskParams struct { Error []byte `json:"error"` DesktopID pgtype.UUID `json:"desktop_id"` + TenantID int64 `json:"tenant_id"` WorkspaceID int64 `json:"workspace_id"` + UserID int64 `json:"user_id"` } func (q *Queries) TenantCancelDesktopTask(ctx context.Context, arg TenantCancelDesktopTaskParams) (DesktopTask, error) { - row := q.db.QueryRow(ctx, tenantCancelDesktopTask, arg.Error, arg.DesktopID, arg.WorkspaceID) + row := q.db.QueryRow(ctx, tenantCancelDesktopTask, arg.Error, arg.DesktopID, arg.TenantID, arg.WorkspaceID, arg.UserID) var i DesktopTask err := row.Scan( &i.ID, diff --git a/server/internal/tenant/repository/generated/desktop_task_sql_test.go b/server/internal/tenant/repository/generated/desktop_task_sql_test.go index da9b130..7398b37 100644 --- a/server/internal/tenant/repository/generated/desktop_task_sql_test.go +++ b/server/internal/tenant/repository/generated/desktop_task_sql_test.go @@ -43,3 +43,24 @@ func TestReconcileDesktopTaskRetryResetsAttempts(t *testing.T) { t.Fatalf("retry reconcile must not increment attempts; query:\n%s", query) } } + +func TestCancelQueuedPublishDesktopTaskByOwnerIsScopedToCreator(t *testing.T) { + query := cancelQueuedPublishDesktopTaskByOwner + + for _, fragment := range []string{ + "FROM desktop_publish_jobs AS j", + "t.tenant_id = $3", + "t.workspace_id = $4", + "t.kind = 'publish'", + "t.status = 'queued'", + "j.desktop_id = t.job_id", + "j.created_by_user_id = $5", + } { + if !strings.Contains(query, fragment) { + t.Fatalf("owner cancel query missing %q; query:\n%s", fragment, query) + } + } + if strings.Contains(query, "target_client_id =") { + t.Fatalf("owner cancel query must not require the task target client; query:\n%s", query) + } +} diff --git a/server/internal/tenant/repository/generated/querier.go b/server/internal/tenant/repository/generated/querier.go index d35f034..3e36542 100644 --- a/server/internal/tenant/repository/generated/querier.go +++ b/server/internal/tenant/repository/generated/querier.go @@ -14,6 +14,7 @@ type Querier interface { ApproveKolSubscription(ctx context.Context, arg ApproveKolSubscriptionParams) error CancelDesktopTaskByClient(ctx context.Context, arg CancelDesktopTaskByClientParams) (DesktopTask, error) CancelDesktopTaskByLease(ctx context.Context, arg CancelDesktopTaskByLeaseParams) (DesktopTask, error) + CancelQueuedPublishDesktopTaskByOwner(ctx context.Context, arg CancelQueuedPublishDesktopTaskByOwnerParams) (DesktopTask, error) ClearDesktopAccountDeleteRequested(ctx context.Context, arg ClearDesktopAccountDeleteRequestedParams) (ClearDesktopAccountDeleteRequestedRow, error) CompleteDesktopTask(ctx context.Context, arg CompleteDesktopTaskParams) (DesktopTask, error) ConfirmReservation(ctx context.Context, arg ConfirmReservationParams) error diff --git a/server/internal/tenant/repository/queries/desktop_task.sql b/server/internal/tenant/repository/queries/desktop_task.sql index cc124c1..6f0cfca 100644 --- a/server/internal/tenant/repository/queries/desktop_task.sql +++ b/server/internal/tenant/repository/queries/desktop_task.sql @@ -169,6 +169,26 @@ WHERE desktop_id = sqlc.arg(desktop_id) AND status = 'queued' RETURNING *; +-- name: CancelQueuedPublishDesktopTaskByOwner :one +UPDATE desktop_tasks AS t +SET status = 'aborted', + error = sqlc.narg(error), + active_attempt_id = NULL, + lease_token_hash = NULL, + lease_expires_at = NULL, + updated_at = now() +FROM desktop_publish_jobs AS j +WHERE t.desktop_id = sqlc.arg(desktop_id) + AND t.tenant_id = sqlc.arg(tenant_id) + AND t.workspace_id = sqlc.arg(workspace_id) + AND t.kind = 'publish' + AND t.status = 'queued' + AND j.desktop_id = t.job_id + AND j.tenant_id = t.tenant_id + AND j.workspace_id = t.workspace_id + AND j.created_by_user_id = sqlc.arg(user_id) +RETURNING t.*; + -- name: TenantCancelDesktopTask :one UPDATE desktop_tasks SET status = 'aborted', @@ -178,8 +198,17 @@ SET status = 'aborted', lease_expires_at = NULL, updated_at = now() WHERE desktop_id = sqlc.arg(desktop_id) + AND tenant_id = sqlc.arg(tenant_id) AND workspace_id = sqlc.arg(workspace_id) AND status = 'queued' + AND EXISTS ( + SELECT 1 + FROM desktop_publish_jobs AS j + WHERE j.desktop_id = desktop_tasks.job_id + AND j.tenant_id = desktop_tasks.tenant_id + AND j.workspace_id = desktop_tasks.workspace_id + AND j.created_by_user_id = sqlc.arg(user_id) + ) RETURNING *; -- name: ReconcileDesktopTask :one diff --git a/server/internal/tenant/transport/media_handler.go b/server/internal/tenant/transport/media_handler.go index b285f40..aa8e700 100644 --- a/server/internal/tenant/transport/media_handler.go +++ b/server/internal/tenant/transport/media_handler.go @@ -80,3 +80,17 @@ func (h *MediaHandler) ListPublishRecords(c *gin.Context) { } response.Success(c, data) } + +func (h *MediaHandler) DeletePublishRecord(c *gin.Context) { + recordID, err := strconv.ParseInt(c.Param("id"), 10, 64) + if err != nil || recordID <= 0 { + response.Error(c, response.ErrBadRequest(40001, "invalid_publish_record_id", "publish record id must be a positive number")) + return + } + data, err := h.svc.DeletePublishRecord(c.Request.Context(), recordID) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, data) +} diff --git a/server/internal/tenant/transport/router.go b/server/internal/tenant/transport/router.go index b30f429..8d8adfd 100644 --- a/server/internal/tenant/transport/router.go +++ b/server/internal/tenant/transport/router.go @@ -93,8 +93,9 @@ func RegisterRoutes(app *bootstrap.App) { tenantProtected.POST("/tasks/:id/reconcile", desktopTaskHandler.Reconcile) tenantProtected.POST("/tasks/:id/cancel", desktopTaskHandler.TenantCancel) tenantProtected.GET("/publish-tasks", desktopTaskHandler.TenantListPublish) - tenantProtected.POST("/publish-tasks/:id/retry", middleware.RequireCurrentBrand(), desktopTaskHandler.TenantRetryPublish) - tenantProtected.GET("/publish-records", middleware.RequireCurrentBrand(), mediaHandler.ListAllPublishRecords) + tenantProtected.POST("/publish-tasks/:id/retry", desktopTaskHandler.TenantRetryPublish) + tenantProtected.GET("/publish-records", mediaHandler.ListAllPublishRecords) + tenantProtected.DELETE("/publish-records/:id", mediaHandler.DeletePublishRecord) tenantProtected.POST("/publish-jobs", middleware.RequireCurrentBrand(), publishJobHandler.Create) complianceHandler := NewComplianceHandler(app) diff --git a/server/internal/tenant/transport/router_test.go b/server/internal/tenant/transport/router_test.go index 4cf0a68..2be6d1e 100644 --- a/server/internal/tenant/transport/router_test.go +++ b/server/internal/tenant/transport/router_test.go @@ -35,6 +35,7 @@ func TestProtectedRoutes_RequireAuth(t *testing.T) { {http.MethodGet, "/api/tenant/publish-tasks"}, {http.MethodPost, "/api/tenant/publish-tasks/:id/retry"}, {http.MethodGet, "/api/tenant/publish-records"}, + {http.MethodDelete, "/api/tenant/publish-records/:id"}, {http.MethodPost, "/api/tenant/publish-jobs"}, {http.MethodGet, "/api/tenant/media/platforms"}, {http.MethodGet, "/api/tenant/workspace/overview"}, diff --git a/server/migrations/20260618190000_add_publish_record_soft_delete.down.sql b/server/migrations/20260618190000_add_publish_record_soft_delete.down.sql new file mode 100644 index 0000000..59ec2f2 --- /dev/null +++ b/server/migrations/20260618190000_add_publish_record_soft_delete.down.sql @@ -0,0 +1,9 @@ +DROP INDEX CONCURRENTLY IF EXISTS idx_publish_records_visible_status; + +DROP INDEX CONCURRENTLY IF EXISTS idx_publish_records_article_account_active; +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_publish_records_article_account_active + ON publish_records (tenant_id, article_id, platform_account_id, updated_at DESC, id DESC) + WHERE status IN ('queued', 'publishing', 'success'); + +ALTER TABLE publish_records + DROP COLUMN IF EXISTS deleted_at; diff --git a/server/migrations/20260618190000_add_publish_record_soft_delete.up.sql b/server/migrations/20260618190000_add_publish_record_soft_delete.up.sql new file mode 100644 index 0000000..25ba896 --- /dev/null +++ b/server/migrations/20260618190000_add_publish_record_soft_delete.up.sql @@ -0,0 +1,12 @@ +ALTER TABLE publish_records + ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ; + +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_publish_records_visible_status + ON publish_records (tenant_id, status, created_at DESC, id DESC) + WHERE deleted_at IS NULL; + +DROP INDEX CONCURRENTLY IF EXISTS idx_publish_records_article_account_active; +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_publish_records_article_account_active + ON publish_records (tenant_id, article_id, platform_account_id, updated_at DESC, id DESC) + WHERE status IN ('queued', 'publishing', 'success') + AND deleted_at IS NULL; diff --git a/server/migrations_ops/20260619100000_add_brand_asset_cleanup_scheduler_job.down.sql b/server/migrations_ops/20260619100000_add_brand_asset_cleanup_scheduler_job.down.sql new file mode 100644 index 0000000..b0451cd --- /dev/null +++ b/server/migrations_ops/20260619100000_add_brand_asset_cleanup_scheduler_job.down.sql @@ -0,0 +1,2 @@ +DELETE FROM ops.scheduler_jobs +WHERE job_key = 'brand_asset_cleanup'; diff --git a/server/migrations_ops/20260619100000_add_brand_asset_cleanup_scheduler_job.up.sql b/server/migrations_ops/20260619100000_add_brand_asset_cleanup_scheduler_job.up.sql new file mode 100644 index 0000000..8dbc2a0 --- /dev/null +++ b/server/migrations_ops/20260619100000_add_brand_asset_cleanup_scheduler_job.up.sql @@ -0,0 +1,17 @@ +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', '品牌资产后台清理', 'tenant', '删除品牌后静默清理该品牌下的文章、词库、竞品和监控采集数据；发布记录作为用户发文总账保留。', true, 'interval', 60, 'Asia/Shanghai', 120, 20, + '{"batch_size":20,"statement_timeout":"8s","dry_run":false}'::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();