geo/server/internal/tenant/app/publish_job_service.go

package app

import (
	"context"
	"encoding/json"
	"errors"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgxpool"
	goredis "github.com/redis/go-redis/v9"
	"go.uber.org/zap"

	"github.com/geo-platform/tenant-api/internal/shared/auth"
	sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache"
	sharedcompliance "github.com/geo-platform/tenant-api/internal/shared/compliance"
	"github.com/geo-platform/tenant-api/internal/shared/config"
	"github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq"
	"github.com/geo-platform/tenant-api/internal/shared/response"
	"github.com/geo-platform/tenant-api/internal/shared/stream"
	tenantcompliance "github.com/geo-platform/tenant-api/internal/tenant/compliance"
	"github.com/geo-platform/tenant-api/internal/tenant/repository"
)

type PublishJobService struct {
	pool       *pgxpool.Pool
	messaging  *rabbitmq.Client
	redis      *goredis.Client
	logger     *zap.Logger
	cache      sharedcache.Cache
	compliance *tenantcompliance.Service
}

func NewPublishJobService(
	pool *pgxpool.Pool,
	messaging *rabbitmq.Client,
	redis *goredis.Client,
	logger *zap.Logger,
) *PublishJobService {
	return NewPublishJobServiceWithConfig(pool, messaging, redis, logger, config.NewStaticProvider(&config.Config{
		Compliance: config.ComplianceConfig{Enabled: true},
	}))
}

func NewPublishJobServiceWithConfig(
	pool *pgxpool.Pool,
	messaging *rabbitmq.Client,
	redis *goredis.Client,
	logger *zap.Logger,
	cfg config.Provider,
) *PublishJobService {
	return &PublishJobService{
		pool:       pool,
		messaging:  messaging,
		redis:      redis,
		logger:     logger,
		compliance: tenantcompliance.NewService(pool, cfg, logger).WithRabbitMQ(messaging),
	}
}

func (s *PublishJobService) WithCache(c sharedcache.Cache) *PublishJobService {
	s.cache = c
	return s
}

type CreatePublishJobAccountRequest struct {
	AccountID string `json:"account_id" binding:"required"`
}

type CreatePublishJobRequest struct {
	Title            string                           `json:"title" binding:"required"`
	ContentRef       map[string]any                   `json:"content_ref" binding:"required"`
	Accounts         []CreatePublishJobAccountRequest `json:"accounts" binding:"required,min=1"`
	ScheduledAt      *time.Time                       `json:"scheduled_at"`
	ArticleVersionID *int64                           `json:"article_version_id,omitempty"`
	AckRecordID      *int64                           `json:"ack_record_id,omitempty"`
}

type CreatePublishJobResponse struct {
	// JobID is set only when this request created a new publish job. It is empty
	// when every requested target was already covered by an existing publish task.
	JobID string `json:"job_id"`
	// TaskIDs is the target-ordered union of created and already-existing tasks.
	TaskIDs         []string `json:"task_ids"`
	CreatedTaskIDs  []string `json:"created_task_ids"`
	ExistingTaskIDs []string `json:"existing_task_ids"`
	RequeuedTaskIDs []string `json:"requeued_task_ids,omitempty"`
}

type publishTarget struct {
	account        *repository.DesktopAccount
	accountSeed    *platformAccountSeed
	targetClientID uuid.UUID
	dedupKey       string
}

type existingPublishRecordTarget struct {
	TaskID            uuid.UUID
	PlatformAccountID int64
}

const desktopPublishDedupActiveConstraint = "uq_desktop_tasks_publish_dedup_active"

func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req CreatePublishJobRequest) (*CreatePublishJobResponse, error) {
	if actor.TenantID == 0 || actor.PrimaryWorkspaceID == 0 || actor.UserID == 0 {
		return nil, response.ErrUnauthorized(40132, "workspace_scope_missing", "workspace context is required")
	}
	brandID, err := requireCurrentBrandID(ctx)
	if err != nil {
		return nil, err
	}

	if s.pool == nil {
		return nil, response.ErrServiceUnavailable(50342, "desktop_publish_job_store_unavailable", "desktop publish job store is unavailable")
	}

	contentRefJSON, err := marshalOptionalJSON(req.ContentRef)
	if err != nil {
		return nil, response.ErrBadRequest(40091, "invalid_content_ref", "content_ref must be serializable")
	}
	articleID, err := resolvePublishArticleID(req.ContentRef)
	if err != nil {
		return nil, err
	}

	requestedAccountIDs, err := parseUniquePublishAccountIDs(req.Accounts)
	if err != nil {
		return nil, err
	}
	runtimeHealth := loadDesktopAccountRuntimeHealth(ctx, s.redis, actor.PrimaryWorkspaceID, requestedAccountIDs)

	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return nil, response.ErrInternal(50096, "desktop_publish_job_begin_failed", "failed to start desktop publish job transaction")
	}
	defer tx.Rollback(ctx)

	accountRepo := repository.NewDesktopAccountRepository(tx)
	taskRepo := repository.NewDesktopTaskRepository(tx)
	articleMeta, err := loadPublishableArticleMeta(ctx, tx, actor.TenantID, brandID, articleID)
	if err != nil {
		return nil, err
	}

	targets := make([]publishTarget, 0, len(req.Accounts))

	for _, accountID := range requestedAccountIDs {
		account, lookupErr := accountRepo.GetByDesktopID(ctx, accountID, actor.PrimaryWorkspaceID)
		if lookupErr != nil {
			if errors.Is(lookupErr, pgx.ErrNoRows) {
				return nil, response.ErrBadRequest(40093, "desktop_account_not_found", "one or more selected desktop accounts do not exist")
			}
			return nil, response.ErrInternal(50098, "desktop_account_lookup_failed", "failed to load selected desktop account")
		}
		if account.DeletedAt != nil || account.DeleteRequestedAt != nil {
			return nil, response.ErrBadRequest(40093, "desktop_account_not_found", "one or more selected desktop accounts are being unbound or no longer exist")
		}
		accountHealth := account.Health
		if report, ok := runtimeHealth[account.DesktopID]; ok && strings.TrimSpace(report.Health) != "" {
			accountHealth = effectiveDesktopAccountRuntimeHealth(report)
		}
		if accountHealth != "live" {
			return nil, response.ErrConflict(40993, "desktop_account_not_publishable", "one or more selected desktop accounts require re-authorization before publishing")
		}
		targetClientID := s.resolveTargetClientID(ctx, account)
		if targetClientID == nil {
			return nil, response.ErrConflict(40992, "desktop_account_client_missing", "one or more selected desktop accounts are not currently available on any desktop client")
		}

		accountSeed, seedErr := loadPlatformAccountSeedByDesktopID(ctx, tx, actor.TenantID, actor.PrimaryWorkspaceID, account.DesktopID)
		if seedErr != nil {
			return nil, seedErr
		}

		targets = append(targets, publishTarget{
			account:        account,
			accountSeed:    accountSeed,
			targetClientID: *targetClientID,
			dedupKey:       desktopPublishDedupKey(actor.TenantID, actor.PrimaryWorkspaceID, articleID, accountSeed.PlatformID, account.DesktopID),
		})
	}

	effectiveVersionID, err := s.compliance.ResolvePublishableVersion(ctx, actor.TenantID, articleID, req.ArticleVersionID)
	if err != nil {
		return nil, err
	}
	targetPlatforms := publishTargetPlatforms(targets)
	gate, err := s.compliance.GateForPublish(ctx, tenantcompliance.GateForPublishRequest{
		TenantID:         actor.TenantID,
		ArticleID:        articleID,
		ArticleVersionID: effectiveVersionID,
		ActorID:          actor.UserID,
		TargetPlatforms:  targetPlatforms,
		AckRecordID:      req.AckRecordID,
		TriggerSource:    "publish_gate",
	})
	if err != nil {
		return nil, compliancePublishError(err, gate)
	}
	if gate != nil {
		switch gate.Decision {
		case sharedcompliance.GateDecisionBlock:
			return nil, compliancePublishError(response.ErrConflict(41001, "compliance_blocked", "content compliance blocked this publish request"), gate)
		case sharedcompliance.GateDecisionNeedsAck:
			return nil, compliancePublishError(response.ErrConflict(41002, "compliance_needs_ack", "content compliance requires acknowledgement before publishing"), gate)
		}
	}

	dedupKeys := desktopPublishDedupKeysFromTargets(targets)
	if len(targets) > 0 && len(dedupKeys) == 0 {
		return nil, response.ErrInternal(50130, "desktop_publish_dedup_key_missing", "failed to build desktop publish deduplication key")
	}
	if err := lockDesktopPublishDedupKeys(ctx, tx, dedupKeys); err != nil {
		return nil, err
	}
	existingTasks, err := loadExistingPublishRecordTargets(ctx, tx, actor.TenantID, actor.PrimaryWorkspaceID, articleID, targets)
	if err != nil {
		return nil, err
	}
	newTargets := make([]publishTarget, 0, len(targets))
	taskIDsByAccount := make(map[uuid.UUID]string, len(targets))
	existingTaskIDs := make([]string, 0, len(targets))
	responseJobID := ""
	for _, target := range targets {
		if existing, ok := existingTasks[target.accountSeed.ID]; ok {
			taskIDsByAccount[target.account.DesktopID] = existing.TaskID.String()
			existingTaskIDs = append(existingTaskIDs, existing.TaskID.String())
			continue
		}
		newTargets = append(newTargets, target)
	}
	if len(newTargets) == 0 {
		// No new task or job was created, so leave compliance acknowledgements
		// untouched and let the deferred rollback release xact locks.
		taskIDs := publishTaskIDsInTargetOrder(targets, taskIDsByAccount)
		return &CreatePublishJobResponse{
			JobID:           responseJobID,
			TaskIDs:         taskIDs,
			CreatedTaskIDs:  []string{},
			ExistingTaskIDs: existingTaskIDs,
		}, nil
	}

	accountSeeds := make([]platformAccountSeed, 0, len(newTargets))
	for _, target := range newTargets {
		accountSeeds = append(accountSeeds, *target.accountSeed)
	}

	if publishBatchRequiresCover(accountSeeds) && strings.TrimSpace(pointerStringValue(articleMeta.CoverAssetURL)) == "" {
		return nil, response.ErrBadRequest(40044, "publish_cover_required", "selected publishing platform requires a cover image")
	}

	jobID := uuid.New()
	job, err := taskRepo.CreatePublishJob(ctx, repository.CreateDesktopPublishJobParams{
		DesktopID:        jobID,
		TenantID:         actor.TenantID,
		WorkspaceID:      actor.PrimaryWorkspaceID,
		CreatedByUserID:  actor.UserID,
		Title:            req.Title,
		ContentRef:       contentRefJSON,
		ScheduledAt:      req.ScheduledAt,
		ArticleID:        &articleID,
		ArticleVersionID: &effectiveVersionID,
	})
	if err != nil {
		return nil, response.ErrInternal(50097, "desktop_publish_job_create_failed", "failed to create desktop publish job")
	}
	responseJobID = job.DesktopID.String()

	publishBatchID, publishRecordIDs, err := createPublishBatchRecords(
		ctx,
		tx,
		actor.TenantID,
		actor.UserID,
		articleID,
		"publish",
		articleMeta.CoverAssetURL,
		accountSeeds,
	)
	if err != nil {
		return nil, err
	}

	createdTasks := make([]*repository.DesktopTask, 0, len(newTargets))
	createdTaskIDs := make([]string, 0, len(newTargets))
	for _, target := range newTargets {
		publishRecordID := publishRecordIDs[target.accountSeed.ID]
		payloadJSON, payloadErr := marshalOptionalJSON(map[string]any{
			"title":               req.Title,
			"content_ref":         req.ContentRef,
			"account_id":          target.account.DesktopID.String(),
			"platform":            target.account.Platform,
			"article_id":          articleID,
			"article_version_id":  effectiveVersionID,
			"publish_batch_id":    publishBatchID,
			"publish_record_id":   publishRecordID,
			"platform_account_id": target.accountSeed.ID,
		})
		if payloadErr != nil {
			return nil, response.ErrBadRequest(40094, "invalid_desktop_task_payload", "publish job payload must be serializable")
		}

		task, createErr := taskRepo.CreateTask(ctx, repository.CreateDesktopTaskParams{
			DesktopID:       uuid.New(),
			JobID:           job.DesktopID,
			TenantID:        actor.TenantID,
			WorkspaceID:     actor.PrimaryWorkspaceID,
			TargetAccountID: target.account.DesktopID,
			TargetClientID:  target.targetClientID,
			Platform:        target.account.Platform,
			Kind:            "publish",
			Payload:         payloadJSON,
			Status:          "queued",
			DedupKey:        &target.dedupKey,
		})
		if createErr != nil {
			if isPostgresUniqueViolation(createErr, desktopPublishDedupActiveConstraint) {
				return nil, response.ErrConflict(40995, "desktop_publish_duplicate", "article has already been submitted for this platform account")
			}
			return nil, response.ErrInternal(50099, "desktop_task_create_failed", "failed to create desktop publish task")
		}

		createdTasks = append(createdTasks, task)
		taskIDsByAccount[target.account.DesktopID] = task.DesktopID.String()
		createdTaskIDs = append(createdTaskIDs, task.DesktopID.String())
	}

	if err := s.consumePublishAckTx(ctx, tx, req.AckRecordID, gate, actor.TenantID, articleID, effectiveVersionID, &jobID); err != nil {
		return nil, err
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, response.ErrInternal(50100, "desktop_publish_job_commit_failed", "failed to commit desktop publish job")
	}
	invalidateArticleCaches(ctx, s.cache, actor.TenantID, &articleID)

	for _, task := range createdTasks {
		dispatchEvent := DesktopDispatchEventFromTask(task, "task_available")
		if publishErr := publishDesktopTaskEvent(context.Background(), s.messaging, DesktopTaskEvent{
			Type:              "task_available",
			TaskID:            task.DesktopID.String(),
			JobID:             task.JobID.String(),
			WorkspaceID:       task.WorkspaceID,
			TargetAccountID:   task.TargetAccountID.String(),
			TargetClientID:    task.TargetClientID.String(),
			Platform:          task.Platform,
			Title:             dispatchEvent.Title,
			BusinessDate:      dispatchEvent.BusinessDate,
			SchedulerGroupKey: dispatchEvent.SchedulerGroupKey,
			QuestionText:      dispatchEvent.QuestionText,
			Status:            task.Status,
			Kind:              task.Kind,
			UpdatedAt:         task.UpdatedAt,
		}); publishErr != nil {
			s.logWarn("desktop task available event publish failed", publishErr, zap.String("task_id", task.DesktopID.String()))
		}

		s.publishDispatchAvailable(task, dispatchEvent)
	}

	return &CreatePublishJobResponse{
		JobID:           responseJobID,
		TaskIDs:         publishTaskIDsInTargetOrder(targets, taskIDsByAccount),
		CreatedTaskIDs:  createdTaskIDs,
		ExistingTaskIDs: existingTaskIDs,
	}, nil
}

func compliancePublishError(err error, gate *tenantcompliance.GateOutcome) error {
	appErr := response.Normalize(err)
	if gate == nil || gate.Result == nil {
		return appErr
	}
	detail, marshalErr := json.Marshal(gate.Result)
	if marshalErr != nil {
		return appErr
	}
	copyErr := *appErr
	copyErr.Detail = string(detail)
	return &copyErr
}

func (s *PublishJobService) consumePublishAckTx(
	ctx context.Context,
	tx pgx.Tx,
	ackRecordID *int64,
	gate *tenantcompliance.GateOutcome,
	tenantID int64,
	articleID int64,
	articleVersionID int64,
	consumedByJobID *uuid.UUID,
) error {
	if ackRecordID == nil {
		return nil
	}
	if gate == nil || gate.Result == nil {
		return response.ErrBadRequest(41003, "compliance_ack_mismatch", "ack record cannot be consumed without a matching compliance result")
	}
	consumed, consumeErr := s.compliance.ConsumeAckTx(ctx, tx, *ackRecordID, tenantcompliance.ConsumeAckQuery{
		TenantID:          tenantID,
		ArticleID:         articleID,
		ArticleVersionID:  articleVersionID,
		CheckRecordID:     gate.Result.RecordID,
		ContentHash:       gate.Result.ContentHash,
		PolicyFingerprint: gate.Result.PolicyFingerprint,
		ConsumedByJobID:   consumedByJobID,
	})
	if consumeErr != nil {
		return response.ErrInternal(51004, "compliance_ack_lookup_failed", "failed to consume compliance ack")
	}
	if !consumed {
		return response.ErrBadRequest(41003, "compliance_ack_mismatch", "ack record is expired, consumed, or no longer matches this publish request")
	}
	return nil
}

func parseUniquePublishAccountIDs(accounts []CreatePublishJobAccountRequest) ([]uuid.UUID, error) {
	if len(accounts) == 0 {
		return nil, response.ErrBadRequest(40092, "invalid_desktop_account_id", "at least one account is required")
	}
	seen := make(map[uuid.UUID]struct{}, len(accounts))
	ids := make([]uuid.UUID, 0, len(accounts))
	for _, accountReq := range accounts {
		accountID, parseErr := uuid.Parse(strings.TrimSpace(accountReq.AccountID))
		if parseErr != nil {
			return nil, response.ErrBadRequest(40092, "invalid_desktop_account_id", "account_id must be a uuid")
		}
		if _, ok := seen[accountID]; ok {
			continue
		}
		seen[accountID] = struct{}{}
		ids = append(ids, accountID)
	}
	if len(ids) == 0 {
		return nil, response.ErrBadRequest(40092, "invalid_desktop_account_id", "at least one account is required")
	}
	return ids, nil
}

func desktopPublishDedupKey(tenantID, workspaceID, articleID int64, platformID string, accountID uuid.UUID) string {
	parts := []string{
		"publish",
		strconv.FormatInt(tenantID, 10),
		strconv.FormatInt(workspaceID, 10),
		strconv.FormatInt(articleID, 10),
		strings.TrimSpace(platformID),
		accountID.String(),
	}
	encoded, err := json.Marshal(parts)
	if err != nil {
		return ""
	}
	return string(encoded)
}

func desktopPublishDedupKeysFromTargets(targets []publishTarget) []string {
	seen := make(map[string]struct{}, len(targets))
	keys := make([]string, 0, len(targets))
	for _, target := range targets {
		key := strings.TrimSpace(target.dedupKey)
		if key == "" {
			continue
		}
		if _, ok := seen[key]; ok {
			continue
		}
		seen[key] = struct{}{}
		keys = append(keys, key)
	}
	sort.Strings(keys)
	return keys
}

func publishTargetPlatforms(targets []publishTarget) []string {
	seen := make(map[string]struct{}, len(targets))
	platforms := make([]string, 0, len(targets))
	for _, target := range targets {
		if target.accountSeed == nil {
			continue
		}
		platformID := strings.TrimSpace(target.accountSeed.PlatformID)
		if platformID == "" {
			continue
		}
		if _, ok := seen[platformID]; ok {
			continue
		}
		seen[platformID] = struct{}{}
		platforms = append(platforms, platformID)
	}
	sort.Strings(platforms)
	return platforms
}

func lockDesktopPublishDedupKeys(ctx context.Context, tx pgx.Tx, keys []string) error {
	if len(keys) == 0 {
		return nil
	}
	if _, err := tx.Exec(ctx, `
		SELECT pg_advisory_xact_lock(hashtextextended(dedup_key, 0))
		FROM (
			SELECT dedup_key
			FROM unnest($1::text[]) AS dedup_keys(dedup_key)
			ORDER BY dedup_key
		) AS ordered_dedup_keys
	`, keys); err != nil {
		return response.ErrInternal(50128, "desktop_publish_dedup_lock_failed", "failed to lock desktop publish deduplication key")
	}
	return nil
}

func loadExistingPublishRecordTargets(
	ctx context.Context,
	tx pgx.Tx,
	tenantID int64,
	workspaceID int64,
	articleID int64,
	targets []publishTarget,
) (map[int64]existingPublishRecordTarget, error) {
	if len(targets) == 0 {
		return map[int64]existingPublishRecordTarget{}, nil
	}

	platformAccountIDs := make([]int64, 0, len(targets))
	platformAccountIDSet := make(map[int64]struct{}, len(targets))
	for _, target := range targets {
		if target.accountSeed == nil || target.accountSeed.ID <= 0 {
			continue
		}
		if _, ok := platformAccountIDSet[target.accountSeed.ID]; ok {
			continue
		}
		platformAccountIDSet[target.accountSeed.ID] = struct{}{}
		platformAccountIDs = append(platformAccountIDs, target.accountSeed.ID)
	}
	if len(platformAccountIDs) == 0 {
		return map[int64]existingPublishRecordTarget{}, nil
	}

	rows, err := tx.Query(ctx, `
		SELECT DISTINCT ON (pr.platform_account_id)
			dt.desktop_id,
			pr.platform_account_id
		FROM publish_records pr
		JOIN platform_accounts pa
		  ON pa.id = pr.platform_account_id
		 AND pa.tenant_id = pr.tenant_id
		JOIN desktop_tasks dt
		  ON dt.tenant_id = pr.tenant_id
		 AND dt.tenant_id = $1
		 AND dt.workspace_id = $2
		 AND dt.kind = 'publish'
		 AND dt.payload ? 'publish_record_id'
		 -- Keep this expression aligned with idx_desktop_tasks_publish_record_payload.
		 AND CASE
		    WHEN (dt.payload->>'publish_record_id') ~ '^[0-9]+$'
		      THEN (dt.payload->>'publish_record_id')::bigint
		    ELSE NULL
		  END = pr.id
		WHERE pr.tenant_id = $1
		  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,
			CASE
				WHEN pr.status = 'success' OR dt.status = 'succeeded' THEN 0
				WHEN pr.status IN ('queued', 'publishing') OR dt.status IN ('queued', 'in_progress') THEN 1
				ELSE 2
			  END,
			GREATEST(
				COALESCE(dt.updated_at, '-infinity'::timestamptz),
				COALESCE(pr.updated_at, '-infinity'::timestamptz)
			) DESC,
			GREATEST(
				COALESCE(dt.created_at, '-infinity'::timestamptz),
				COALESCE(pr.created_at, '-infinity'::timestamptz)
			) DESC,
			dt.created_at DESC
		`, tenantID, workspaceID, articleID, platformAccountIDs)
	if err != nil {
		return nil, response.ErrInternal(50129, "desktop_publish_dedup_lookup_failed", "failed to inspect existing publish records")
	}
	defer rows.Close()

	existing := make(map[int64]existingPublishRecordTarget, len(platformAccountIDs))
	for rows.Next() {
		var item existingPublishRecordTarget
		if scanErr := rows.Scan(
			&item.TaskID,
			&item.PlatformAccountID,
		); scanErr != nil {
			return nil, response.ErrInternal(50129, "desktop_publish_dedup_lookup_failed", "failed to parse existing publish record")
		}
		existing[item.PlatformAccountID] = item
	}
	if err := rows.Err(); err != nil {
		return nil, response.ErrInternal(50129, "desktop_publish_dedup_lookup_failed", "failed to iterate existing publish records")
	}
	return existing, nil
}

func publishTaskIDsInTargetOrder(targets []publishTarget, taskIDsByAccount map[uuid.UUID]string) []string {
	taskIDs := make([]string, 0, len(targets))
	seen := make(map[string]struct{}, len(targets))
	for _, target := range targets {
		taskID := strings.TrimSpace(taskIDsByAccount[target.account.DesktopID])
		if taskID == "" {
			continue
		}
		if _, ok := seen[taskID]; ok {
			continue
		}
		seen[taskID] = struct{}{}
		taskIDs = append(taskIDs, taskID)
	}
	return taskIDs
}

func isPostgresUniqueViolation(err error, constraintName string) bool {
	var pgErr *pgconn.PgError
	return errors.As(err, &pgErr) && pgErr.Code == "23505" && pgErr.ConstraintName == constraintName
}

func (s *PublishJobService) RetryByDesktopTask(
	ctx context.Context,
	client *repository.DesktopClient,
	taskID uuid.UUID,
) (*CreatePublishJobResponse, error) {
	if client == nil {
		return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
	}
	if s.pool == nil {
		return nil, response.ErrServiceUnavailable(50342, "desktop_publish_job_store_unavailable", "desktop publish job store is unavailable")
	}

	taskRepo := repository.NewDesktopTaskRepository(s.pool)
	task, err := taskRepo.GetByDesktopID(ctx, taskID, client.WorkspaceID)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, response.ErrNotFound(40482, "desktop_task_not_found", "desktop task not found")
		}
		return nil, response.ErrInternal(50095, "desktop_task_lookup_failed", "failed to inspect desktop task state")
	}
	if task.Kind != "publish" {
		return nil, response.ErrBadRequest(40095, "desktop_task_not_publish", "only publish tasks can be retried")
	}
	if err := s.authorizePublishTaskForClientUser(ctx, client, task); err != nil {
		return nil, err
	}
	if task.Status == "queued" || task.Status == "in_progress" {
		return nil, response.ErrConflict(40994, "desktop_task_retry_conflict", "desktop task is still pending or running")
	}
	if shouldRequeuePublishTaskInsteadOfCreatingNew(task) {
		return s.requeueUnknownPublishTask(ctx, task)
	}

	payload := unmarshalJSONObject(task.Payload)
	articleID, err := s.resolveRetryArticleID(ctx, client.TenantID, payload)
	if err != nil {
		return nil, err
	}
	brandID, err := s.resolveRetryArticleBrandID(ctx, client.TenantID, articleID)
	if err != nil {
		return nil, err
	}

	title := strings.TrimSpace(stringPointerValue(extractStringPointer(payload, "title")))
	if title == "" {
		title = "文章发布"
	}

	publishCtx := auth.WithCurrentBrandID(ctx, brandID)
	return s.Create(publishCtx, auth.Actor{
		TenantID:           client.TenantID,
		UserID:             client.UserID,
		PrimaryWorkspaceID: client.WorkspaceID,
	}, CreatePublishJobRequest{
		Title: title,
		ContentRef: map[string]any{
			"article_id": articleID,
		},
		Accounts: []CreatePublishJobAccountRequest{
			{AccountID: task.TargetAccountID.String()},
		},
	})
}

func (s *PublishJobService) RetryTenantDesktopTask(
	ctx context.Context,
	actor auth.Actor,
	taskID uuid.UUID,
) (*CreatePublishJobResponse, error) {
	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")
	}
	if s.pool == nil {
		return nil, response.ErrServiceUnavailable(50342, "desktop_publish_job_store_unavailable", "desktop publish job store is unavailable")
	}

	taskRepo := repository.NewDesktopTaskRepository(s.pool)
	task, err := taskRepo.GetByDesktopID(ctx, taskID, workspaceID)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, response.ErrNotFound(40482, "desktop_task_not_found", "desktop task not found")
		}
		return nil, response.ErrInternal(50095, "desktop_task_lookup_failed", "failed to inspect desktop task state")
	}
	if task.Kind != "publish" {
		return nil, response.ErrBadRequest(40095, "desktop_task_not_publish", "only publish tasks can be retried")
	}
	if err := s.authorizePublishTaskForActor(ctx, actor, workspaceID, task); err != nil {
		return nil, err
	}
	if task.Status == "queued" || task.Status == "in_progress" {
		return nil, response.ErrConflict(40994, "desktop_task_retry_conflict", "desktop task is still pending or running")
	}
	if shouldRequeuePublishTaskInsteadOfCreatingNew(task) {
		return s.requeueUnknownPublishTask(ctx, task)
	}

	payload := unmarshalJSONObject(task.Payload)
	articleID, err := s.resolveRetryArticleID(ctx, actor.TenantID, payload)
	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 = "文章发布"
	}

	publishCtx := auth.WithCurrentBrandID(ctx, brandID)
	return s.Create(publishCtx, auth.Actor{
		TenantID:           actor.TenantID,
		UserID:             actor.UserID,
		PrimaryWorkspaceID: workspaceID,
		Role:               actor.Role,
	}, CreatePublishJobRequest{
		Title: title,
		ContentRef: map[string]any{
			"article_id": articleID,
		},
		Accounts: []CreatePublishJobAccountRequest{
			{AccountID: task.TargetAccountID.String()},
		},
	})
}

func shouldRequeuePublishTaskInsteadOfCreatingNew(task *repository.DesktopTask) bool {
	return task != nil && task.Kind == "publish" && task.Status == "unknown"
}

func (s *PublishJobService) requeueUnknownPublishTask(
	ctx context.Context,
	task *repository.DesktopTask,
) (*CreatePublishJobResponse, error) {
	if task == nil {
		return nil, response.ErrNotFound(40482, "desktop_task_not_found", "desktop task not found")
	}
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return nil, response.ErrInternal(50105, "desktop_task_reconcile_begin_failed", "failed to start desktop task reconcile")
	}
	defer tx.Rollback(ctx)

	taskRepo := repository.NewDesktopTaskRepository(tx)
	requeued, err := taskRepo.Reconcile(ctx, task.DesktopID, task.WorkspaceID, "retry", nil, nil)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, response.ErrConflict(40985, "desktop_task_reconcile_conflict", "desktop task is not waiting for reconcile")
		}
		return nil, response.ErrInternal(50094, "desktop_task_reconcile_failed", "failed to reconcile desktop task")
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, response.ErrInternal(50106, "desktop_task_reconcile_commit_failed", "failed to commit desktop task reconcile")
	}

	publishDesktopTaskLifecycleEvent(ctx, s.messaging, s.logger, requeued, "task_available")
	return createPublishJobResponseForRequeuedTask(requeued), nil
}

func createPublishJobResponseForRequeuedTask(task *repository.DesktopTask) *CreatePublishJobResponse {
	if task == nil {
		return &CreatePublishJobResponse{
			TaskIDs:         []string{},
			CreatedTaskIDs:  []string{},
			ExistingTaskIDs: []string{},
			RequeuedTaskIDs: []string{},
		}
	}
	taskID := task.DesktopID.String()
	return &CreatePublishJobResponse{
		JobID:           "",
		TaskIDs:         []string{taskID},
		CreatedTaskIDs:  []string{},
		ExistingTaskIDs: []string{},
		RequeuedTaskIDs: []string{taskID},
	}
}

func (s *PublishJobService) authorizePublishTaskForClientUser(
	ctx context.Context,
	client *repository.DesktopClient,
	task *repository.DesktopTask,
) error {
	if client == nil || task == nil {
		return response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
	}

	var ownsTask bool
	err := s.pool.QueryRow(ctx, `
		SELECT EXISTS (
			SELECT 1
			FROM desktop_publish_jobs
			WHERE desktop_id = $1
			  AND workspace_id = $2
			  AND created_by_user_id = $3
		)
	`, task.JobID, client.WorkspaceID, client.UserID).Scan(&ownsTask)
	if err != nil {
		return response.ErrInternal(50115, "desktop_publish_job_lookup_failed", "failed to validate desktop publish task ownership")
	}
	if !ownsTask {
		return response.ErrForbidden(40384, "desktop_publish_task_not_owned", "desktop publish task belongs to another user")
	}
	return nil
}

func (s *PublishJobService) authorizePublishTaskForActor(
	ctx context.Context,
	actor auth.Actor,
	workspaceID int64,
	task *repository.DesktopTask,
) error {
	if task == nil || actor.TenantID == 0 || actor.UserID == 0 || workspaceID == 0 {
		return response.ErrUnauthorized(40132, "workspace_scope_missing", "workspace context is required")
	}

	var ownsTask bool
	err := s.pool.QueryRow(ctx, `
		SELECT EXISTS (
			SELECT 1
			FROM desktop_publish_jobs
			WHERE tenant_id = $1
			  AND desktop_id = $2
			  AND workspace_id = $3
			  AND created_by_user_id = $4
		)
	`, actor.TenantID, task.JobID, workspaceID, actor.UserID).Scan(&ownsTask)
	if err != nil {
		return response.ErrInternal(50115, "desktop_publish_job_lookup_failed", "failed to validate desktop publish task ownership")
	}
	if !ownsTask {
		return response.ErrForbidden(40384, "desktop_publish_task_not_owned", "desktop publish task belongs to another user")
	}
	return nil
}

func (s *PublishJobService) resolveRetryArticleBrandID(ctx context.Context, tenantID int64, articleID int64) (int64, error) {
	var brandID int64
	err := s.pool.QueryRow(ctx, `
		SELECT brand_id
		FROM articles
		WHERE id = $1
		  AND tenant_id = $2
		  AND deleted_at IS NULL
	`, articleID, tenantID).Scan(&brandID)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return 0, response.ErrNotFound(40411, "article_not_found", "article not found")
		}
		return 0, response.ErrInternal(50037, "article_check_failed", "failed to validate article")
	}
	if brandID <= 0 {
		return 0, response.ErrInternal(50037, "article_check_failed", "failed to validate article")
	}
	return brandID, nil
}

func (s *PublishJobService) resolveRetryArticleID(ctx context.Context, tenantID int64, payload map[string]any) (int64, error) {
	if articleID, ok := resolveRetryArticleIDFromPayload(payload); ok {
		return articleID, nil
	}

	if publishRecordID, ok := extractInt64FromMap(payload, "publish_record_id"); ok && publishRecordID > 0 {
		var articleID int64
		err := s.pool.QueryRow(ctx, `
			SELECT article_id
			FROM publish_records
			WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL
		`, publishRecordID, tenantID).Scan(&articleID)
		if err == nil && articleID > 0 {
			return articleID, nil
		}
		if err != nil && !errors.Is(err, pgx.ErrNoRows) {
			return 0, response.ErrInternal(50113, "desktop_publish_record_lookup_failed", "failed to lookup desktop publish record")
		}
	}

	if publishBatchID, ok := extractInt64FromMap(payload, "publish_batch_id"); ok && publishBatchID > 0 {
		var articleID int64
		err := s.pool.QueryRow(ctx, `
			SELECT article_id
			FROM publish_batches
			WHERE id = $1 AND tenant_id = $2
		`, publishBatchID, tenantID).Scan(&articleID)
		if err == nil && articleID > 0 {
			return articleID, nil
		}
		if err != nil && !errors.Is(err, pgx.ErrNoRows) {
			return 0, response.ErrInternal(50114, "desktop_publish_batch_lookup_failed", "failed to lookup desktop publish batch")
		}
	}

	return 0, response.ErrBadRequest(40096, "desktop_task_article_missing", "desktop publish task is missing article_id")
}

func resolveRetryArticleIDFromPayload(payload map[string]any) (int64, bool) {
	if articleID, ok := extractInt64FromMap(payload, "article_id"); ok && articleID > 0 {
		return articleID, true
	}
	if len(payload) == 0 {
		return 0, false
	}
	nestedContentRef, ok := payload["content_ref"].(map[string]any)
	if !ok {
		return 0, false
	}
	if articleID, ok := extractInt64FromMap(nestedContentRef, "article_id", "id"); ok && articleID > 0 {
		return articleID, true
	}
	return 0, false
}

// publishDispatchAvailable pushes a per-client dispatch frame onto the topic
// exchange. Every tenant-api instance consumes the exchange and forwards the
// frame to the WebSocket owning the target client id. This is the primary
// signal the desktop client reacts to; the fanout event above is kept only so
// that other UIs (admin-web) see the same task lifecycle.
func (s *PublishJobService) publishDispatchAvailable(task *repository.DesktopTask, event stream.DesktopDispatchEvent) {
	if s == nil || s.messaging == nil || task == nil {
		return
	}
	if err := publishDesktopDispatchEvent(context.Background(), s.messaging, s.logger, event); err != nil {
		s.logWarn("desktop dispatch publish failed", err,
			zap.String("task_id", task.DesktopID.String()),
		)
	}
}

func (s *PublishJobService) logWarn(message string, err error, fields ...zap.Field) {
	if s.logger == nil || err == nil {
		return
	}
	fields = append(fields, zap.Error(err))
	s.logger.Warn(message, fields...)
}

func (s *PublishJobService) resolveTargetClientID(ctx context.Context, account *repository.DesktopAccount) *uuid.UUID {
	if account == nil {
		return nil
	}

	presence := loadDesktopAccountPresence(ctx, s.redis, []uuid.UUID{account.DesktopID})
	if clientID, ok := presence[account.DesktopID]; ok {
		return &clientID
	}

	if account.ClientID == nil {
		return nil
	}

	clientID := *account.ClientID
	return &clientID
}

func stringPointerValue(value *string) string {
	if value == nil {
		return ""
	}
	return strings.TrimSpace(*value)
}