chore: update app generation workflow

server/internal/tenant/app/article_imitation_service.go

@@ -409,6 +409,30 @@ func (s *ArticleImitationService) executeGeneration(ctx context.Context, job art
 	quotaRepo := repository.NewQuotaRepository(tx)
 	auditTx := repository.NewAuditRepository(tx)
 
+	if err := EnsureGenerationArticleWritable(ctx, tx, job.ArticleID, job.TenantID); err != nil {
+		if IsGenerationArticleUnavailable(err) {
+			rollbackGenerationTx(tx)
+			if cleanupErr := HandleGenerationTaskArticleUnavailable(ctx, s.pool, s.cache, ClaimedGenerationTask{
+				ID:                 job.TaskID,
+				TenantID:           job.TenantID,
+				OperatorID:         job.OperatorID,
+				ArticleID:          job.ArticleID,
+				QuotaReservationID: job.ReservationID,
+				TaskType:           articleImitationTaskType,
+				InputParams:        cloneInputParams(job.InputParams),
+			}, articleImitationTaskType, false); cleanupErr != nil && s.logger != nil {
+				s.logger.Warn("imitation generation article unavailable cleanup failed",
+					zap.Error(cleanupErr),
+					zap.Int64("task_id", job.TaskID),
+					zap.Int64("article_id", job.ArticleID),
+				)
+			}
+			return
+		}
+		failWithRollback("persist_article_status", err)
+		return
+	}
+
 	versionID, err := articleRepo.CreateArticleVersion(ctx, repository.CreateArticleVersionInput{
 		ArticleID:       job.ArticleID,
 		Title:           title,

server/internal/tenant/app/generation_task_runtime.go

@@ -10,6 +10,7 @@ import (
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 
+	sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache"
 	"github.com/geo-platform/tenant-api/internal/shared/llm"
 	"github.com/geo-platform/tenant-api/internal/tenant/repository"
 )
@@ -26,6 +27,12 @@ type ClaimedGenerationTask struct {
 
 const generationCleanupTimeout = 30 * time.Second
 
+var errGenerationArticleUnavailable = errors.New("generation article is unavailable")
+
+type generationDBTX interface {
+	QueryRow(context.Context, string, ...interface{}) pgx.Row
+}
+
 type promptRuleTaskContext struct {
 	PromptRuleID int64
 	BrandID      int64
@@ -46,6 +53,124 @@ func rollbackGenerationTx(tx pgx.Tx) {
 	_ = tx.Rollback(ctx)
 }
 
+func generationStringPtr(value string) *string {
+	return &value
+}
+
+func EnsureGenerationArticleWritable(ctx context.Context, db generationDBTX, articleID, tenantID int64) error {
+	if articleID <= 0 || tenantID <= 0 {
+		return errGenerationArticleUnavailable
+	}
+
+	var lockedArticleID int64
+	if err := db.QueryRow(ctx, `
+		SELECT id
+		FROM articles
+		WHERE id = $1
+		  AND tenant_id = $2
+		  AND deleted_at IS NULL
+		FOR UPDATE
+	`, articleID, tenantID).Scan(&lockedArticleID); err != nil {
+		return err
+	}
+	return nil
+}
+
+func IsGenerationArticleUnavailable(err error) bool {
+	return errors.Is(err, errGenerationArticleUnavailable) || errors.Is(err, pgx.ErrNoRows)
+}
+
+func HandleGenerationTaskArticleUnavailable(
+	ctx context.Context,
+	pool *pgxpool.Pool,
+	cache sharedcache.Cache,
+	task ClaimedGenerationTask,
+	taskType string,
+	updateKolTaskRecord bool,
+) error {
+	cleanupCtx, cancel := newGenerationCleanupContext()
+	defer cancel()
+
+	const errMsg = "文章已删除，生成结果已丢弃"
+	now := time.Now()
+	auditRepo := repository.NewAuditRepository(pool)
+	quotaRepo := repository.NewQuotaRepository(pool)
+	usageRepo := repository.NewKolUsageRepository(pool)
+
+	var cleanupErr error
+	if task.ID > 0 {
+		if err := auditRepo.UpdateGenerationTaskStatus(cleanupCtx, repository.GenerationTaskStatusInput{
+			ID:           task.ID,
+			TenantID:     task.TenantID,
+			Status:       "cancelled",
+			ErrorMessage: generationStringPtr(errMsg),
+			CompletedAt:  &now,
+		}); err != nil {
+			cleanupErr = errors.Join(cleanupErr, fmt.Errorf("cancel generation task: %w", err))
+		}
+		if updateKolTaskRecord {
+			_ = auditRepo.UpdateTaskRecordStatus(cleanupCtx, repository.TaskRecordInput{
+				TenantID:     task.TenantID,
+				TaskType:     kolGenerationTaskRecordType,
+				ResourceType: kolGenerationTaskRecordScope,
+				ResourceID:   task.ID,
+				Status:       "cancelled",
+				ErrorMessage: generationStringPtr(errMsg),
+			})
+			_ = usageRepo.MarkFailedByTask(cleanupCtx, task.TenantID, task.ID, errMsg)
+		}
+	}
+
+	if task.QuotaReservationID > 0 {
+		var balance int
+		balanceErr := error(nil)
+		if task.OperatorID > 0 {
+			balance, balanceErr = quotaRepo.GetCurrentBalance(cleanupCtx, task.TenantID, "article_generation")
+		}
+		refunded, err := quotaRepo.RefundPendingReservation(cleanupCtx, task.QuotaReservationID, task.TenantID)
+		if err != nil {
+			cleanupErr = errors.Join(cleanupErr, fmt.Errorf("refund generation reservation: %w", err))
+		} else if refunded {
+			if balanceErr != nil {
+				cleanupErr = errors.Join(cleanupErr, fmt.Errorf("load quota balance for refund: %w", balanceErr))
+			} else if task.OperatorID > 0 {
+				reason := "generation_cancel_refund"
+				referenceType := "generation_task"
+				taskReferenceID := task.ID
+				if _, err := quotaRepo.InsertQuotaLedger(cleanupCtx, repository.QuotaLedgerInput{
+					TenantID:      task.TenantID,
+					OperatorID:    task.OperatorID,
+					QuotaType:     "article_generation",
+					Delta:         1,
+					BalanceAfter:  balance + 1,
+					Reason:        &reason,
+					ReferenceType: &referenceType,
+					ReferenceID:   &taskReferenceID,
+				}); err != nil {
+					cleanupErr = errors.Join(cleanupErr, fmt.Errorf("insert generation cancel refund ledger: %w", err))
+				}
+			}
+		}
+	}
+
+	if task.ArticleID > 0 {
+		invalidateArticleCaches(cleanupCtx, cache, task.TenantID, &task.ArticleID)
+	}
+
+	RecordGenerationTaskEvent(GenerationTaskEventTransition, GenerationTaskLogContext{
+		TaskID:        task.ID,
+		TenantID:      task.TenantID,
+		OperatorID:    task.OperatorID,
+		ArticleID:     task.ArticleID,
+		ReservationID: task.QuotaReservationID,
+		TaskType:      taskType,
+		Stage:         "article_deleted",
+		StatusAfter:   "cancelled",
+		Result:        GenerationTaskResultSkipped,
+	})
+	return cleanupErr
+}
+
 func HandleClaimedGenerationTaskFailure(
 	ctx context.Context,
 	pool *pgxpool.Pool,

server/internal/tenant/app/prompt_generate_service.go

@@ -593,6 +593,30 @@ func (s *PromptRuleGenerationService) executeGeneration(ctx context.Context, job
 	quotaRepo := repository.NewQuotaRepository(tx)
 	auditTx := repository.NewAuditRepository(tx)
 
+	if err := EnsureGenerationArticleWritable(ctx, tx, job.ArticleID, job.TenantID); err != nil {
+		if IsGenerationArticleUnavailable(err) {
+			rollbackGenerationTx(tx)
+			if cleanupErr := HandleGenerationTaskArticleUnavailable(ctx, s.pool, s.cache, ClaimedGenerationTask{
+				ID:                 job.TaskID,
+				TenantID:           job.TenantID,
+				OperatorID:         job.OperatorID,
+				ArticleID:          job.ArticleID,
+				QuotaReservationID: job.ReservationID,
+				TaskType:           "custom_generation",
+				InputParams:        cloneInputParams(job.InputParams),
+			}, "custom_generation", false); cleanupErr != nil && s.logger != nil {
+				s.logger.Warn("prompt rule generation article unavailable cleanup failed",
+					zap.Error(cleanupErr),
+					zap.Int64("task_id", job.TaskID),
+					zap.Int64("article_id", job.ArticleID),
+				)
+			}
+			return
+		}
+		failWithRollback("persist_article_status", err)
+		return
+	}
+
 	versionID, err := articleRepo.CreateArticleVersion(ctx, repository.CreateArticleVersionInput{
 		ArticleID:       job.ArticleID,
 		Title:           title,

server/internal/tenant/app/template_service.go

@@ -545,6 +545,30 @@ func (s *TemplateService) executeGeneration(ctx context.Context, job generationJ
 	quotaRepo := repository.NewQuotaRepository(tx)
 	auditTx := repository.NewAuditRepository(tx)
 
+	if err := EnsureGenerationArticleWritable(ctx, tx, job.ArticleID, job.TenantID); err != nil {
+		if IsGenerationArticleUnavailable(err) {
+			rollbackGenerationTx(tx)
+			if cleanupErr := HandleGenerationTaskArticleUnavailable(ctx, s.pool, s.cache, ClaimedGenerationTask{
+				ID:                 job.TaskID,
+				TenantID:           job.TenantID,
+				OperatorID:         job.OperatorID,
+				ArticleID:          job.ArticleID,
+				QuotaReservationID: job.ReservationID,
+				TaskType:           "template",
+				InputParams:        cloneInputParams(job.Params),
+			}, "template", false); cleanupErr != nil && s.logger != nil {
+				s.logger.Warn("template generation article unavailable cleanup failed",
+					zap.Error(cleanupErr),
+					zap.Int64("task_id", job.TaskID),
+					zap.Int64("article_id", job.ArticleID),
+				)
+			}
+			return
+		}
+		failWithRollback("persist_article_status", err)
+		return
+	}
+
 	versionID, err := articleRepo.CreateArticleVersion(ctx, repository.CreateArticleVersionInput{
 		ArticleID:       job.ArticleID,
 		Title:           title,

server/internal/tenant/repository/generated/audit.sql.go

@@ -129,9 +129,9 @@ const updateGenerationTaskStatus = `-- name: UpdateGenerationTaskStatus :exec
 UPDATE generation_tasks SET status = $1::text, error_message = $2,
        started_at = COALESCE($3, started_at),
        completed_at = COALESCE($4, completed_at),
-       lease_token = CASE WHEN $1::text IN ('completed', 'failed') THEN NULL ELSE lease_token END,
-       lease_owner = CASE WHEN $1::text IN ('completed', 'failed') THEN NULL ELSE lease_owner END,
-       lease_expires_at = CASE WHEN $1::text IN ('completed', 'failed') THEN NULL ELSE lease_expires_at END,
+       lease_token = CASE WHEN $1::text IN ('completed', 'failed', 'cancelled', 'canceled', 'aborted') THEN NULL ELSE lease_token END,
+       lease_owner = CASE WHEN $1::text IN ('completed', 'failed', 'cancelled', 'canceled', 'aborted') THEN NULL ELSE lease_owner END,
+       lease_expires_at = CASE WHEN $1::text IN ('completed', 'failed', 'cancelled', 'canceled', 'aborted') THEN NULL ELSE lease_expires_at END,
        updated_at = NOW()
 WHERE id = $5 AND tenant_id = $6
 `

server/internal/tenant/repository/queries/audit.sql

@@ -42,8 +42,8 @@ WHERE tenant_id = sqlc.arg(tenant_id)
 UPDATE generation_tasks SET status = sqlc.arg(status)::text, error_message = sqlc.arg(error_message),
        started_at = COALESCE(sqlc.arg(started_at), started_at),
        completed_at = COALESCE(sqlc.arg(completed_at), completed_at),
-       lease_token = CASE WHEN sqlc.arg(status)::text IN ('completed', 'failed') THEN NULL ELSE lease_token END,
-       lease_owner = CASE WHEN sqlc.arg(status)::text IN ('completed', 'failed') THEN NULL ELSE lease_owner END,
-       lease_expires_at = CASE WHEN sqlc.arg(status)::text IN ('completed', 'failed') THEN NULL ELSE lease_expires_at END,
+       lease_token = CASE WHEN sqlc.arg(status)::text IN ('completed', 'failed', 'cancelled', 'canceled', 'aborted') THEN NULL ELSE lease_token END,
+       lease_owner = CASE WHEN sqlc.arg(status)::text IN ('completed', 'failed', 'cancelled', 'canceled', 'aborted') THEN NULL ELSE lease_owner END,
+       lease_expires_at = CASE WHEN sqlc.arg(status)::text IN ('completed', 'failed', 'cancelled', 'canceled', 'aborted') THEN NULL ELSE lease_expires_at END,
        updated_at = NOW()
 WHERE id = sqlc.arg(id) AND tenant_id = sqlc.arg(tenant_id);

server/internal/worker/generate/kol_generation_worker.go

@@ -179,6 +179,15 @@ func (w *KolGenerationWorker) Process(ctx context.Context, task tenantapp.Claime
 	auditTx := repository.NewAuditRepository(tx)
 	usageRepo := repository.NewKolUsageRepository(tx)
 
+	if err := tenantapp.EnsureGenerationArticleWritable(ctx, tx, task.ArticleID, task.TenantID); err != nil {
+		if tenantapp.IsGenerationArticleUnavailable(err) {
+			w.rollbackAndCancelUnavailableArticle(ctx, tx, task)
+			return
+		}
+		w.rollbackAndFail(ctx, tx, task, "persist_article_status", err)
+		return
+	}
+
 	versionID, err := articleRepo.CreateArticleVersion(ctx, repository.CreateArticleVersionInput{
 		ArticleID:       task.ArticleID,
 		Title:           title,
@@ -270,6 +279,19 @@ func (w *KolGenerationWorker) rollbackAndFail(ctx context.Context, tx pgx.Tx, ta
 	w.HandleTaskFailure(ctx, task, stage, taskErr)
 }
 
+func (w *KolGenerationWorker) rollbackAndCancelUnavailableArticle(ctx context.Context, tx pgx.Tx, task tenantapp.ClaimedGenerationTask) {
+	cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	_ = tx.Rollback(cleanupCtx)
+	if err := tenantapp.HandleGenerationTaskArticleUnavailable(ctx, w.pool, w.cache, task, kolGenerationTaskType, true); err != nil && w.logger != nil {
+		w.logger.Warn("kol generation article unavailable cleanup failed",
+			zap.Error(err),
+			zap.Int64("task_id", task.ID),
+			zap.Int64("article_id", task.ArticleID),
+		)
+	}
+}
+
 var (
 	errMissingRenderedPrompt = errors.New("kol generation task missing rendered prompt")
 	errEmptyGeneratedContent = errors.New("kol generation returned empty content")