diff --git a/server/internal/tenant/app/article_service.go b/server/internal/tenant/app/article_service.go index 5ac1669..1f2ddbd 100644 --- a/server/internal/tenant/app/article_service.go +++ b/server/internal/tenant/app/article_service.go @@ -874,6 +874,9 @@ func (s *ArticleService) Delete(ctx context.Context, id int64) error { if err != nil || tag.RowsAffected() == 0 { return response.ErrNotFound(40411, "article_not_found", "article not found") } + if err := cancelQueuedPublishTasksForUnavailableArticle(ctx, tx, actor.TenantID, id); err != nil { + return err + } if err := repository.NewImageReferenceRepository(tx).DeleteByArticle(ctx, actor.TenantID, id); err != nil { return response.ErrInternal(50010, "delete_failed", "failed to delete article image references") } diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index 79ceb0c..fc7970e 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/server/internal/tenant/app/desktop_task_service.go @@ -287,18 +287,20 @@ func (s *DesktopTaskService) recheckQueuedPublishTasksForClient(ctx context.Cont j.created_by_user_id, j.article_id, j.article_version_id, - COALESCE(array_agg(DISTINCT dt.platform_id) FILTER (WHERE dt.platform_id <> ''), '{}') AS platforms + COALESCE(array_agg(DISTINCT dt.platform_id) FILTER (WHERE dt.platform_id <> ''), '{}') AS platforms, + BOOL_OR(a.deleted_at IS NOT NULL) AS article_deleted FROM desktop_tasks dt JOIN desktop_publish_jobs j ON j.desktop_id = dt.job_id + LEFT JOIN articles a ON a.id = j.article_id AND a.tenant_id = j.tenant_id WHERE dt.target_client_id = $1 AND dt.kind = 'publish' AND dt.status = 'queued' AND j.status = 'queued' AND j.article_id IS NOT NULL AND j.article_version_id IS NOT NULL - GROUP BY j.desktop_id, j.tenant_id, j.workspace_id, j.created_by_user_id, j.article_id, j.article_version_id - ORDER BY MIN(dt.created_at) - LIMIT 5 + GROUP BY j.desktop_id, j.tenant_id, j.workspace_id, j.created_by_user_id, j.article_id, j.article_version_id + ORDER BY MIN(dt.created_at) + LIMIT 5 `, client.ID) if err != nil { s.logWarn("desktop publish compliance recheck query failed", err, zap.String("client_id", client.ID.String())) @@ -314,11 +316,12 @@ func (s *DesktopTaskService) recheckQueuedPublishTasksForClient(ctx context.Cont ArticleID int64 ArticleVersionID int64 Platforms []string + ArticleDeleted bool } candidates := make([]candidate, 0) for rows.Next() { var item candidate - if scanErr := rows.Scan(&item.JobID, &item.TenantID, &item.WorkspaceID, &item.CreatedByUserID, &item.ArticleID, &item.ArticleVersionID, &item.Platforms); scanErr != nil { + if scanErr := rows.Scan(&item.JobID, &item.TenantID, &item.WorkspaceID, &item.CreatedByUserID, &item.ArticleID, &item.ArticleVersionID, &item.Platforms, &item.ArticleDeleted); scanErr != nil { s.logWarn("desktop publish compliance recheck scan failed", scanErr, zap.String("client_id", client.ID.String())) return response.ErrInternal(50118, "desktop_publish_compliance_recheck_failed", "failed to scan publish job for compliance recheck") } @@ -332,6 +335,12 @@ func (s *DesktopTaskService) recheckQueuedPublishTasksForClient(ctx context.Cont } for _, item := range candidates { + if item.ArticleDeleted { + if cancelErr := s.cancelUnavailableArticleQueuedPublishTasks(ctx, item.TenantID, item.ArticleID); cancelErr != nil { + return cancelErr + } + continue + } gate, gateErr := s.compliance.GateForPublish(ctx, tenantcompliance.GateForPublishRequest{ TenantID: item.TenantID, ArticleID: item.ArticleID, @@ -347,6 +356,12 @@ func (s *DesktopTaskService) recheckQueuedPublishTasksForClient(ctx context.Cont continue } if gateErr != nil { + if isComplianceInvalidArticleVersionError(gateErr) { + if cancelErr := s.cancelUnavailableArticleQueuedPublishTasks(ctx, item.TenantID, item.ArticleID); cancelErr != nil { + return cancelErr + } + continue + } return response.ErrServiceUnavailable(41004, "compliance_engine_unavailable", "scheduled publish compliance recheck failed") } if gate != nil && gate.Decision == sharedcompliance.GateDecisionBlock { @@ -358,6 +373,82 @@ func (s *DesktopTaskService) recheckQueuedPublishTasksForClient(ctx context.Cont return nil } +func (s *DesktopTaskService) cancelUnavailableArticleQueuedPublishTasks(ctx context.Context, tenantID, articleID int64) error { + if s == nil || s.pool == nil || tenantID <= 0 || articleID <= 0 { + return nil + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return response.ErrInternal(50122, "publish_task_cancel_unavailable_article_begin_failed", "failed to start unavailable article publish task cleanup") + } + defer tx.Rollback(ctx) + + taskRefs, err := queuedPublishTaskRefsForArticle(ctx, tx, tenantID, articleID) + if err != nil { + return err + } + if err := cancelQueuedPublishTasksForUnavailableArticle(ctx, tx, tenantID, articleID); err != nil { + return err + } + if err := tx.Commit(ctx); err != nil { + return response.ErrInternal(50123, "publish_task_cancel_unavailable_article_commit_failed", "failed to commit unavailable article publish task cleanup") + } + + for _, ref := range taskRefs { + task, getErr := s.repo.GetByDesktopID(ctx, ref.TaskID, ref.WorkspaceID) + if getErr == nil && task != nil { + s.publishTaskEvent(ctx, task, "task_canceled") + } + } + return nil +} + +type desktopTaskRef struct { + TaskID uuid.UUID + WorkspaceID int64 +} + +func queuedPublishTaskRefsForArticle(ctx context.Context, tx pgx.Tx, tenantID, articleID int64) ([]desktopTaskRef, error) { + if tx == nil || tenantID <= 0 || articleID <= 0 { + return nil, nil + } + rows, err := tx.Query(ctx, ` + SELECT dt.desktop_id, dt.workspace_id + FROM desktop_tasks dt + JOIN desktop_publish_jobs j + ON j.desktop_id = dt.job_id + AND j.tenant_id = dt.tenant_id + WHERE dt.tenant_id = $1 + AND dt.kind = 'publish' + AND dt.status = 'queued' + AND j.article_id = $2 + AND j.status = 'queued' + ORDER BY dt.created_at ASC, dt.desktop_id ASC + `, tenantID, articleID) + if err != nil { + return nil, response.ErrInternal(50124, "publish_task_cancel_unavailable_article_lookup_failed", "failed to inspect unavailable article publish tasks") + } + defer rows.Close() + + taskRefs := make([]desktopTaskRef, 0) + for rows.Next() { + var ref desktopTaskRef + if scanErr := rows.Scan(&ref.TaskID, &ref.WorkspaceID); scanErr != nil { + return nil, response.ErrInternal(50124, "publish_task_cancel_unavailable_article_lookup_failed", "failed to scan unavailable article publish tasks") + } + taskRefs = append(taskRefs, ref) + } + if err := rows.Err(); err != nil { + return nil, response.ErrInternal(50124, "publish_task_cancel_unavailable_article_lookup_failed", "failed to iterate unavailable article publish tasks") + } + return taskRefs, nil +} + +func isComplianceInvalidArticleVersionError(err error) bool { + appErr, ok := err.(*response.AppError) + return ok && appErr.Code == 41005 && appErr.Message == "invalid_article_version" +} + const desktopTaskRepositoryReturningColumns = ` t.desktop_id, t.job_id, diff --git a/server/internal/tenant/app/desktop_task_service_test.go b/server/internal/tenant/app/desktop_task_service_test.go index 4cbe358..5c52edf 100644 --- a/server/internal/tenant/app/desktop_task_service_test.go +++ b/server/internal/tenant/app/desktop_task_service_test.go @@ -4,6 +4,8 @@ import ( "regexp" "strings" "testing" + + "github.com/geo-platform/tenant-api/internal/shared/response" ) func TestDesktopTaskRecoverySelectQueryMatchesRecoveredLeaseScanOrder(t *testing.T) { @@ -136,6 +138,17 @@ func TestResolvePublishRecoveryOutcome(t *testing.T) { } } +func TestIsComplianceInvalidArticleVersionError(t *testing.T) { + t.Parallel() + + if !isComplianceInvalidArticleVersionError(response.ErrBadRequest(41005, "invalid_article_version", "article version does not belong to this article")) { + t.Fatal("invalid article version compliance errors should be treated as stale publish tasks") + } + if isComplianceInvalidArticleVersionError(response.ErrServiceUnavailable(41004, "compliance_engine_unavailable", "compliance gate failed closed")) { + t.Fatal("compliance engine failures should not be treated as stale publish tasks") + } +} + func recoverDesktopTaskSelectColumns(query string) []string { re := regexp.MustCompile(`(?is)select\s+(.*?)\s+from\s+desktop_tasks`) match := re.FindStringSubmatch(query) diff --git a/server/internal/tenant/app/publish_record_support.go b/server/internal/tenant/app/publish_record_support.go index 7c39999..67e5090 100644 --- a/server/internal/tenant/app/publish_record_support.go +++ b/server/internal/tenant/app/publish_record_support.go @@ -111,6 +111,156 @@ func createPublishBatchRecords( return publishBatchID, recordIDs, nil } +func cancelQueuedPublishTasksForUnavailableArticle(ctx context.Context, tx pgx.Tx, tenantID, articleID int64) error { + if tx == nil || tenantID <= 0 || articleID <= 0 { + return nil + } + errorJSON, err := marshalOptionalJSON(map[string]any{ + "code": "article_unavailable", + "message": "article is unavailable before publishing", + "source": "article_unavailable_cleanup", + }) + if err != nil { + return response.ErrInternal(50120, "publish_task_cancel_payload_failed", "failed to encode publish task cancellation payload") + } + + if _, err := tx.Exec(ctx, ` + WITH affected_records AS ( + UPDATE publish_records pr + SET status = 'cancelled', + error_message = COALESCE(NULLIF(error_message, ''), 'article_unavailable'), + updated_at = NOW() + FROM desktop_publish_jobs j + JOIN desktop_tasks dt + ON dt.job_id = j.desktop_id + AND dt.tenant_id = j.tenant_id + WHERE pr.id::text = dt.payload->>'publish_record_id' + AND pr.tenant_id = $1 + AND pr.article_id = $2 + AND pr.status IN ('queued', 'publishing', 'pending', 'running') + AND j.tenant_id = $1 + AND j.article_id = $2 + AND j.status = 'queued' + AND dt.kind = 'publish' + AND dt.status = 'queued' + RETURNING pr.publish_batch_id + ), + affected_batches AS ( + SELECT DISTINCT publish_batch_id + FROM affected_records + WHERE publish_batch_id IS NOT NULL + ), + updated_tasks AS ( + UPDATE desktop_tasks dt + SET status = 'aborted', + error = $3, + 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 dt.kind = 'publish' + AND dt.status = 'queued' + AND j.article_id = $2 + AND j.status = 'queued' + 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 = $2 + AND j.status = 'queued' + AND EXISTS ( + SELECT 1 + FROM updated_tasks ut + WHERE ut.job_id = j.desktop_id + ) + RETURNING j.desktop_id + ), + recalculated_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.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.status IN ('success', 'published', 'publish_success') + ) AND EXISTS ( + SELECT 1 + FROM publish_records pr + WHERE pr.publish_batch_id = pb.id + 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.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 + RETURNING pb.id + ), + recalculated_article AS ( + UPDATE articles a + SET publish_status = CASE + WHEN EXISTS ( + SELECT 1 + FROM publish_records pr + WHERE pr.article_id = a.id + AND pr.tenant_id = a.tenant_id + AND pr.status IN ('queued', 'publishing', 'pending', 'running') + ) THEN 'publishing' + WHEN EXISTS ( + SELECT 1 + FROM publish_records pr + WHERE pr.article_id = a.id + AND pr.tenant_id = a.tenant_id + 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.status NOT IN ('success', 'published', 'publish_success') + ) THEN 'partial_success' + WHEN EXISTS ( + SELECT 1 + FROM publish_records pr + WHERE pr.article_id = a.id + AND pr.tenant_id = a.tenant_id + AND pr.status IN ('success', 'published', 'publish_success') + ) THEN 'success' + ELSE 'failed' + END, + updated_at = NOW() + WHERE a.tenant_id = $1 + AND a.id = $2 + AND a.deleted_at IS NULL + AND EXISTS (SELECT 1 FROM updated_jobs) + RETURNING a.id + ) + SELECT COUNT(1) FROM updated_jobs + `, tenantID, articleID, errorJSON); err != nil { + return response.ErrInternal(50121, "publish_task_cancel_deleted_article_failed", "failed to cancel queued publish tasks for deleted article") + } + return nil +} + func syncDesktopPublishTaskState( ctx context.Context, tx pgx.Tx,