descrybe/apps/api/internal/processing/pipeline.go

package processing

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"strings"
	"time"

	"github.com/descrybe/descrybe-v2/apps/api/internal/aiprompts"
	"github.com/descrybe/descrybe-v2/apps/api/internal/billing"
	"github.com/descrybe/descrybe-v2/apps/api/internal/catalog"
	"github.com/descrybe/descrybe-v2/apps/api/internal/company"
	"github.com/descrybe/descrybe-v2/apps/api/internal/security"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgxpool"
)

type Worker interface {
	ProcessJob(ctx context.Context, jobID uuid.UUID) error
}

type Pipeline struct {
	Pool      *pgxpool.Pool
	Billing   *billing.Service
	Engine    *Engine
	BatchSize int
	// ProgressEvery controls how often job counters/step_progress are written during a run.
	// <=0 uses defaultProgressEvery. Always flushed at end of each claim batch.
	ProgressEvery int
	Limiter       *StartLimiter
	// AI resolves per-company BYOK completers (prefer company key, else platform).
	AI CompanyCompleterResolver
	// Prompts resolves per-company editable AI prompt templates.
	Prompts *aiprompts.Service
}

// Defaults for large jobs: claim enough rows per round-trip without unbounded memory.
const (
	defaultBatchSize     = 100
	maxBatchSize         = 500
	defaultProgressEvery = 25
)

func resolveBatchSize(n int) int {
	if n <= 0 {
		return defaultBatchSize
	}
	if n > maxBatchSize {
		return maxBatchSize
	}
	return n
}

func resolveProgressEvery(n int) int {
	if n <= 0 {
		return defaultProgressEvery
	}
	return n
}

// shouldFlushJobProgress decides when to persist mid-job counters/step_progress.
// batchDone forces a flush of any pending successes (end of claim batch / cancel / credits stop).
func shouldFlushJobProgress(successesSinceFlush, progressEvery int, batchDone bool) bool {
	if successesSinceFlush <= 0 {
		return false
	}
	if batchDone {
		return true
	}
	return successesSinceFlush >= resolveProgressEvery(progressEvery)
}

// shouldDebitProductProcessing reports whether processOne should ConsumeCredits.
// BYOK skips managed burn; enhance input-hash reuse (SkipCreditDebit) skips the
// flat product_processing debit when there was no LLM and no meaningful rework.
func shouldDebitProductProcessing(usingBYOK bool, result StepResult) bool {
	return !usingBYOK && !result.SkipCreditDebit
}

// AI roles for admin-configured completers (platform / company bindings).
// Keep in sync with platformsettings.AIRole*. Product pipeline uses AIRoleProcessing.
// AIRoleSupport is a FUTURE ticket-assist slot only — do not auto-reply tickets
// from the pipeline; see support.TryAutoReplyLLM. Docs Ask stays no-LLM.
const (
	AIRoleProcessing    = "processing"
	AIRoleVectorization = "vectorization"
	AIRoleDocsAPI       = "docs_api"
	AIRoleSupport       = "support"
)

// CompanyCompleterResolver picks an LLM client for a tenant job.
// Implemented by aiprovider.Service; kept as an interface to avoid import cycles.
type CompanyCompleterResolver interface {
	ResolveCompleter(ctx context.Context, companyID uuid.UUID) (c Completer, modeLabel string, usingBYOK bool, err error)
	// ResolveCompleterForRole prefers an admin role binding when set; otherwise
	// falls back to ResolveCompleter (company BYOK → platform OpenAI → env).
	ResolveCompleterForRole(ctx context.Context, companyID uuid.UUID, role string) (c Completer, modeLabel string, usingBYOK bool, err error)
}

func NewPipeline(pool *pgxpool.Pool) *Pipeline {
	return &Pipeline{
		Pool:          pool,
		Billing:       &billing.Service{Pool: pool},
		Engine:        &Engine{Vector: NoopVectorCategorizer{}, EPREL: nil},
		BatchSize:     defaultBatchSize,
		ProgressEvery: defaultProgressEvery,
		Limiter:       NewStartLimiter(20, time.Minute),
	}
}

type Job struct {
	ID                uuid.UUID      `json:"id"`
	CompanyID         uuid.UUID      `json:"company_id"`
	Status            string         `json:"status"`
	TotalProducts     int            `json:"total_products"`
	ProcessedProducts int            `json:"processed_products"`
	ProcessingType    string         `json:"processing_type"`
	CurrentStep       string         `json:"current_step"`
	StepProgress      []StepProgress `json:"step_progress"`
	Error             *string        `json:"error,omitempty"`
	StartedAt         *time.Time     `json:"started_at,omitempty"`
	CompletedAt       *time.Time     `json:"completed_at,omitempty"`
	CreatedAt         time.Time      `json:"created_at"`
}

// StartJob creates one or more pending processing jobs for the given raw products.
// Ownership, credits/plan gates, and start rate-limiting apply once to the full set.
// When len(owned) exceeds maxJobProducts, products are auto-split into multiple jobs
// of ≤maxJobProducts so ClaimNext SKIP LOCKED workers stay parallelizable.
// Absolute request cap is MaxStartProducts (100k–1M scale).
func (p *Pipeline) StartJob(ctx context.Context, companyID, userID uuid.UUID, rawIDs []uuid.UUID, processingType string) ([]Job, error) {
	if processingType == "" {
		processingType = "full"
	}
	if len(rawIDs) == 0 {
		return nil, ErrRawIDsRequired
	}
	if len(rawIDs) > startProductCap() {
		return nil, fmt.Errorf("%w (max %d)", ErrTooManyProducts, startProductCap())
	}
	if p.Limiter != nil && !p.Limiter.Allow(companyID) {
		return nil, ErrRateLimited
	}
	owned, err := p.filterOwnedRawIDs(ctx, companyID, rawIDs)
	if err != nil {
		return nil, err
	}
	if len(owned) == 0 {
		return nil, ErrNoMatchingProducts
	}
	ownedSet := make(map[uuid.UUID]struct{}, len(owned))
	for _, id := range owned {
		ownedSet[id] = struct{}{}
	}
	for _, id := range rawIDs {
		if _, ok := ownedSet[id]; !ok {
			return nil, ErrRawProductsNotFound
		}
	}
	if p.Billing != nil {
		if err := p.assertProcessingGates(ctx, companyID, processingType, len(owned)); err != nil {
			return nil, err
		}
	}

	progress := InitialStepProgress(processingType)
	progressJSON, err := marshalStepProgress(progress)
	if err != nil {
		return nil, err
	}
	firstStep := ""
	if len(progress) > 0 {
		firstStep = progress[0].Step
	}

	chunks := chunkUUIDs(owned, perJobProductCap())
	tx, err := p.Pool.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)

	jobIDs := make([]uuid.UUID, 0, len(chunks))
	for _, chunk := range chunks {
		var jobID uuid.UUID
		err = tx.QueryRow(ctx, `
			INSERT INTO processing_jobs (
				company_id, user_id, status, total_products, processing_type, current_step, step_progress
			) VALUES ($1, $2, 'pending', $3, $4, $5, $6::jsonb) RETURNING id`,
			companyID, userID, len(chunk), processingType, firstStep, progressJSON).Scan(&jobID)
		if err != nil {
			return nil, err
		}
		if err := insertJobProducts(ctx, tx, jobID, chunk); err != nil {
			return nil, err
		}
		jobIDs = append(jobIDs, jobID)
	}
	if err := tx.Commit(ctx); err != nil {
		return nil, err
	}

	jobs := make([]Job, 0, len(jobIDs))
	for _, jobID := range jobIDs {
		job, err := p.GetJob(ctx, companyID, jobID)
		if err != nil {
			return nil, err
		}
		jobs = append(jobs, job)
	}
	log.Printf("processing: started jobs=%d company=%s products=%d type=%s", len(jobs), companyID, len(owned), processingType)
	return jobs, nil
}

// assertProcessingGates enforces plan features and credit/SKU caps before starting or retrying work.
func (p *Pipeline) assertProcessingGates(ctx context.Context, companyID uuid.UUID, processingType string, batchSize int) error {
	if p == nil || p.Billing == nil {
		return nil
	}
	if err := p.Billing.AssertProcessingFeatures(ctx, companyID, processingType); err != nil {
		return err
	}
	opts := billing.ProcessingGateOpts{
		RequiresAI:    billing.ProcessingTypeRequiresAI(processingType) || billing.ProcessingTypeIsEmailCampaignAI(processingType),
		RequiresEPREL: billing.ProcessingTypeRequiresEPREL(processingType),
	}
	return p.Billing.AssertCanStartProcessing(ctx, companyID, batchSize, opts)
}

// FormatStartJobsResponse keeps top-level job fields (id, …) for single-job clients.
// When StartJob auto-splits, sibling ids and the full jobs list are additive.
func FormatStartJobsResponse(jobs []Job) any {
	if len(jobs) == 0 {
		return map[string]any{}
	}
	if len(jobs) == 1 {
		return jobs[0]
	}
	siblings := make([]uuid.UUID, 0, len(jobs)-1)
	total := 0
	for i, j := range jobs {
		total += j.TotalProducts
		if i > 0 {
			siblings = append(siblings, j.ID)
		}
	}
	return struct {
		Job
		Jobs                []Job       `json:"jobs"`
		SiblingJobIDs       []uuid.UUID `json:"sibling_job_ids"`
		JobCount            int         `json:"job_count"`
		TotalProductsQueued int         `json:"total_products_queued"`
	}{
		Job:                 jobs[0],
		Jobs:                jobs,
		SiblingJobIDs:       siblings,
		JobCount:            len(jobs),
		TotalProductsQueued: total,
	}
}

// FormatListJobsResponse annotates auto-split sibling batches for list clients.
// Jobs that share processing_type and the same created_at Unix second get
// sibling_job_ids, job_count, and total_products_queued; lone jobs are unchanged.
func FormatListJobsResponse(jobs []Job) []any {
	if len(jobs) == 0 {
		return []any{}
	}
	type batchKey struct {
		ptype string
		sec   int64
	}
	groups := make(map[batchKey][]int, len(jobs))
	for i, j := range jobs {
		k := batchKey{ptype: j.ProcessingType, sec: j.CreatedAt.Unix()}
		groups[k] = append(groups[k], i)
	}
	out := make([]any, len(jobs))
	for i, j := range jobs {
		k := batchKey{ptype: j.ProcessingType, sec: j.CreatedAt.Unix()}
		idxs := groups[k]
		if len(idxs) <= 1 {
			out[i] = j
			continue
		}
		siblings := make([]uuid.UUID, 0, len(idxs)-1)
		total := 0
		for _, idx := range idxs {
			total += jobs[idx].TotalProducts
			if idx != i {
				siblings = append(siblings, jobs[idx].ID)
			}
		}
		out[i] = struct {
			Job
			SiblingJobIDs       []uuid.UUID `json:"sibling_job_ids"`
			JobCount            int         `json:"job_count"`
			TotalProductsQueued int         `json:"total_products_queued"`
		}{
			Job:                 j,
			SiblingJobIDs:       siblings,
			JobCount:            len(idxs),
			TotalProductsQueued: total,
		}
	}
	return out
}

// maxJobProducts is the per-job product cap (SKIP LOCKED claim unit).
// MaxStartProducts is the absolute API StartJob / v1 process request cap (auto-split above maxJobProducts).
const (
	maxJobProducts   = 5000
	MaxStartProducts = 1_000_000
	ownedFilterChunk = 5000
)

// testMaxStartProducts overrides startProductCap when > 0 (tests only).
var testMaxStartProducts int

func startProductCap() int {
	if testMaxStartProducts > 0 {
		return testMaxStartProducts
	}
	return MaxStartProducts
}

// StartProductCap is the effective StartJob / v1 process product cap (honors test overrides).
func StartProductCap() int {
	return startProductCap()
}

// SetTestStartProductCap overrides StartProductCap for tests; pass 0 to restore MaxStartProducts.
func SetTestStartProductCap(n int) {
	testMaxStartProducts = n
}

// testMaxJobProducts overrides perJobProductCap when > 0 (tests only).
var testMaxJobProducts int

func perJobProductCap() int {
	if testMaxJobProducts > 0 {
		return testMaxJobProducts
	}
	return maxJobProducts
}

func chunkUUIDs(ids []uuid.UUID, size int) [][]uuid.UUID {
	if len(ids) == 0 {
		return nil
	}
	if size <= 0 {
		size = len(ids)
	}
	out := make([][]uuid.UUID, 0, (len(ids)+size-1)/size)
	for i := 0; i < len(ids); i += size {
		end := i + size
		if end > len(ids) {
			end = len(ids)
		}
		out = append(out, ids[i:end])
	}
	return out
}

func insertJobProducts(ctx context.Context, tx pgx.Tx, jobID uuid.UUID, rawIDs []uuid.UUID) error {
	if len(rawIDs) == 0 {
		return nil
	}
	rows := make([][]any, len(rawIDs))
	for i, rid := range rawIDs {
		rows[i] = []any{jobID, rid, "pending"}
	}
	_, err := tx.CopyFrom(ctx,
		pgx.Identifier{"processing_job_products"},
		[]string{"job_id", "raw_product_id", "status"},
		pgx.CopyFromRows(rows),
	)
	return err
}

func (p *Pipeline) filterOwnedRawIDs(ctx context.Context, companyID uuid.UUID, rawIDs []uuid.UUID) ([]uuid.UUID, error) {
	found := make(map[uuid.UUID]struct{}, len(rawIDs))
	for _, batch := range chunkUUIDs(rawIDs, ownedFilterChunk) {
		rows, err := p.Pool.Query(ctx, `
			SELECT id FROM raw_products WHERE company_id = $1 AND id = ANY($2::uuid[])`, companyID, batch)
		if err != nil {
			return nil, err
		}
		for rows.Next() {
			var id uuid.UUID
			if err := rows.Scan(&id); err != nil {
				rows.Close()
				return nil, err
			}
			found[id] = struct{}{}
		}
		err = rows.Err()
		rows.Close()
		if err != nil {
			return nil, err
		}
	}
	out := make([]uuid.UUID, 0, len(found))
	seen := make(map[uuid.UUID]struct{}, len(found))
	for _, id := range rawIDs {
		if _, ok := found[id]; !ok {
			continue
		}
		if _, dup := seen[id]; dup {
			continue
		}
		seen[id] = struct{}{}
		out = append(out, id)
	}
	return out, nil
}

func (p *Pipeline) scanJob(row pgx.Row) (Job, error) {
	var j Job
	var progressBytes []byte
	err := row.Scan(
		&j.ID, &j.CompanyID, &j.Status, &j.TotalProducts, &j.ProcessedProducts, &j.ProcessingType,
		&j.CurrentStep, &progressBytes, &j.Error, &j.StartedAt, &j.CompletedAt, &j.CreatedAt,
	)
	if err != nil {
		return Job{}, err
	}
	if len(progressBytes) > 0 {
		_ = json.Unmarshal(progressBytes, &j.StepProgress)
	}
	if j.StepProgress == nil {
		j.StepProgress = []StepProgress{}
	}
	return j, nil
}

func (p *Pipeline) GetJob(ctx context.Context, companyID, id uuid.UUID) (Job, error) {
	return p.scanJob(p.Pool.QueryRow(ctx, `
		SELECT id, company_id, status, total_products, processed_products, processing_type,
			COALESCE(current_step, ''), COALESCE(step_progress, '[]'::jsonb), error, started_at, completed_at, created_at
		FROM processing_jobs WHERE id = $1 AND company_id = $2`, id, companyID))
}

func (p *Pipeline) ListJobs(ctx context.Context, companyID uuid.UUID, limit int) ([]Job, error) {
	if limit <= 0 || limit > 200 {
		limit = 50
	}
	rows, err := p.Pool.Query(ctx, `
		SELECT id, company_id, status, total_products, processed_products, processing_type,
			COALESCE(current_step, ''), COALESCE(step_progress, '[]'::jsonb), error, started_at, completed_at, created_at
		FROM processing_jobs WHERE company_id = $1
		ORDER BY created_at DESC LIMIT $2`, companyID, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := make([]Job, 0)
	for rows.Next() {
		j, err := p.scanJob(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, j)
	}
	return out, rows.Err()
}

func (p *Pipeline) CancelJob(ctx context.Context, companyID, id uuid.UUID) (Job, error) {
	job, err := p.GetJob(ctx, companyID, id)
	if err != nil {
		return Job{}, err
	}
	switch job.Status {
	case "pending", "running":
		// ok
	default:
		return Job{}, ErrJobNotCancellable
	}
	for i := range job.StepProgress {
		switch strings.ToLower(job.StepProgress[i].Status) {
		case "pending", "running", "processing":
			job.StepProgress[i].Status = "cancelled"
		}
	}
	progressJSON, err := marshalStepProgress(job.StepProgress)
	if err != nil {
		return Job{}, err
	}
	tx, err := p.Pool.Begin(ctx)
	if err != nil {
		return Job{}, err
	}
	defer tx.Rollback(ctx)

	ct, err := tx.Exec(ctx, `
		UPDATE processing_jobs SET status = 'cancelled', completed_at = now(), updated_at = now(),
			current_step = 'cancelled', step_progress = $3::jsonb
		WHERE id = $1 AND company_id = $2 AND status IN ('pending', 'running')`, id, companyID, progressJSON)
	if err != nil {
		return Job{}, err
	}
	if ct.RowsAffected() == 0 {
		return Job{}, ErrJobNotCancellable
	}
	if err := cancelPendingJobProducts(ctx, tx.Exec, id); err != nil {
		return Job{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return Job{}, err
	}
	log.Printf("processing: cancelled job=%s company=%s", id, companyID)
	return p.GetJob(ctx, companyID, id)
}

// RetryJob requeues failed/cancelled items (or whole failed job) back to pending.
func (p *Pipeline) RetryJob(ctx context.Context, companyID, id uuid.UUID) (Job, error) {
	job, err := p.GetJob(ctx, companyID, id)
	if err != nil {
		return Job{}, err
	}
	switch job.Status {
	case "failed", "cancelled", "completed":
		// ok
	case "pending", "running":
		return Job{}, ErrJobStillActive
	default:
		return Job{}, ErrJobNotRetryable
	}
	if p.Limiter != nil && !p.Limiter.Allow(companyID) {
		return Job{}, ErrRateLimited
	}
	if err := p.assertProcessingGates(ctx, companyID, job.ProcessingType, job.TotalProducts); err != nil {
		return Job{}, err
	}

	progress := InitialStepProgress(job.ProcessingType)
	progressJSON, err := marshalStepProgress(progress)
	if err != nil {
		return Job{}, err
	}
	firstStep := ""
	if len(progress) > 0 {
		firstStep = progress[0].Step
	}

	itemStatuses := []string{"failed", "cancelled"}
	resetProcessed := false
	if job.Status == "completed" {
		// Completed retries should rerun the whole job, not immediately no-op.
		itemStatuses = []string{"processed", "failed", "cancelled"}
		resetProcessed = true
	}

	_, err = p.Pool.Exec(ctx, `
		UPDATE processing_job_products SET status = 'pending', error = NULL, updated_at = now()
		WHERE job_id = $1 AND status = ANY($2::text[])`, id, itemStatuses)
	if err != nil {
		return Job{}, err
	}

	processedProducts := job.ProcessedProducts
	if resetProcessed {
		processedProducts = 0
	}

	_, err = p.Pool.Exec(ctx, `
		UPDATE processing_jobs SET
			status = 'pending', error = NULL, started_at = NULL, completed_at = NULL,
			processed_products = $2,
			current_step = $3, step_progress = $4::jsonb, updated_at = now()
		WHERE id = $1 AND company_id = $5`, id, processedProducts, firstStep, progressJSON, companyID)
	if err != nil {
		return Job{}, err
	}
	log.Printf("processing: retry job=%s company=%s", id, companyID)
	return p.GetJob(ctx, companyID, id)
}

// ProcessJob runs the multi-step pipeline for pending job products.
// Idempotent: pending items only; existing processed_products rows are updated in place.
// Terminal job statuses (completed/cancelled/failed) are no-ops — RetryJob requeues work.
func (p *Pipeline) ProcessJob(ctx context.Context, jobID uuid.UUID) error {
	var companyID uuid.UUID
	var status, processingType string
	var alreadyProcessed int
	err := p.Pool.QueryRow(ctx, `
		SELECT company_id, status, processing_type, processed_products
		FROM processing_jobs WHERE id = $1`, jobID).
		Scan(&companyID, &status, &processingType, &alreadyProcessed)
	if err != nil {
		return err
	}
	if !IsProcessableJobStatus(status) {
		return nil
	}

	modeLabel := AIProviderInternal
	usingBYOK := false
	jobEngine := p.Engine
	if jobEngine == nil {
		jobEngine = &Engine{Vector: NoopVectorCategorizer{}}
	}
	if p.AI != nil {
		c, label, byok, rerr := p.AI.ResolveCompleterForRole(ctx, companyID, AIRoleProcessing)
		if rerr != nil {
			log.Printf("processing: ai resolve job=%s role=%s err=%s", jobID, AIRoleProcessing, TruncateError(rerr))
		} else {
			if label != "" {
				modeLabel = normalizeProviderMode(label)
			}
			usingBYOK = byok
			cloned := *jobEngine
			cloned.Completer = c
			cloned.ProviderMode = modeLabel
			jobEngine = &cloned
		}
	} else if label := jobEngine.EngineProviderMode(); label != "" {
		modeLabel = label
	}

	progress := InitialStepProgress(processingType)
	if len(progress) > 0 {
		progress[0].Status = "running"
	}
	progressJSON, err := marshalStepProgress(progress)
	if err != nil {
		return fmt.Errorf("processing: initial step_progress job=%s: %w", jobID, err)
	}
	current := ""
	if len(progress) > 0 {
		current = progress[0].Step
	}
	_, err = p.Pool.Exec(ctx, `
		UPDATE processing_jobs SET status = 'running', started_at = COALESCE(started_at, now()), updated_at = now(),
			current_step = $2, step_progress = $3::jsonb, ai_provider_mode = $4
		WHERE id = $1 AND status IN ('pending', 'running')`, jobID, current, progressJSON, modeLabel)
	if err != nil {
		return err
	}

	batch := resolveBatchSize(p.BatchSize)
	progressEvery := resolveProgressEvery(p.ProgressEvery)
	jobCache := p.loadJobScopedCache(ctx, companyID, jobID)
	processed := alreadyProcessed
	failed := 0
	tokenTotal := 0
	var lastResult *StepResult
	sinceFlush := 0
	tokensSinceFlush := 0

	flushProgress := func(batchDone bool) {
		if !shouldFlushJobProgress(sinceFlush, progressEvery, batchDone) {
			return
		}
		mode := modeLabel
		if lastResult != nil && lastResult.AIProviderMode != "" {
			mode = lastResult.AIProviderMode
		}
		if err := p.flushJobCountersAndProgress(ctx, jobID, processingType, lastResult, processed, tokensSinceFlush, mode); err != nil {
			log.Printf("processing: flush progress job=%s err=%s", jobID, TruncateError(err))
		}
		sinceFlush = 0
		tokensSinceFlush = 0
	}

	for {
		if err := stopOnCancel(p.jobCancelled(ctx, jobID)); err != nil {
			flushProgress(true)
			if errors.Is(err, errJobCancelled) {
				return nil
			}
			return fmt.Errorf("processing: check cancel job=%s: %w", jobID, err)
		}
		items, err := p.loadPendingItems(ctx, jobID, batch)
		if err != nil {
			flushProgress(true)
			return err
		}
		if len(items) == 0 {
			break
		}
		if err := p.hydrateJobItems(ctx, companyID, items); err != nil {
			flushProgress(true)
			return fmt.Errorf("processing: hydrate batch job=%s: %w", jobID, err)
		}
		creditsStop := false
		for i := range items {
			// Throttle cancel polls: once per claim batch plus every progressEvery items.
			if i > 0 && i%progressEvery == 0 {
				if err := stopOnCancel(p.jobCancelled(ctx, jobID)); err != nil {
					flushProgress(true)
					if errors.Is(err, errJobCancelled) {
						return nil
					}
					return fmt.Errorf("processing: check cancel job=%s: %w", jobID, err)
				}
			}
			ok, tokens, result, itemErr := p.processOne(ctx, companyID, jobID, &items[i], processingType, jobEngine, modeLabel, usingBYOK, &jobCache)
			if itemErr != nil {
				failed++
				if _, err := p.Pool.Exec(ctx, `
					UPDATE processing_job_products SET status = 'failed', error = $2, updated_at = now() WHERE id = $1`,
					items[i].ID, TruncateError(itemErr)); err != nil {
					flushProgress(true)
					return fmt.Errorf("processing: mark item failed job=%s item=%s: %w", jobID, items[i].ID, err)
				}
				if _, err := p.Pool.Exec(ctx, `
					UPDATE raw_products SET processing_status = 'failed', updated_at = now()
					WHERE id = $1 AND company_id = $2`, items[i].RawID, companyID); err != nil {
					log.Printf("processing: mark raw failed job=%s raw=%s err=%s", jobID, items[i].RawID, TruncateError(err))
				}
				log.Printf("processing: item failed job=%s raw=%s err=%s", jobID, items[i].RawID, TruncateError(itemErr))
				// Stop the job: further items would burn provider cost with no wallet left.
				if errors.Is(itemErr, billing.ErrInsufficientCredits) {
					ct, ferr := p.Pool.Exec(ctx, `
						UPDATE processing_job_products
						SET status = 'failed', error = $2, updated_at = now()
						WHERE job_id = $1 AND status IN ('pending', 'processing')`,
						jobID, TruncateError(billing.ErrInsufficientCredits))
					if ferr != nil {
						log.Printf("processing: fail pending on credits job=%s err=%s", jobID, TruncateError(ferr))
					} else {
						failed += int(ct.RowsAffected())
					}
					creditsStop = true
					break
				}
				continue
			}
			if ok {
				processed++
				tokenTotal += tokens
				sinceFlush++
				tokensSinceFlush += tokens
				lastResult = &result
				flushProgress(false)
			}
		}
		flushProgress(true)
		if creditsStop {
			break
		}
	}

	finalStatus := "completed"
	var errMsg *string
	if failed > 0 && processed == alreadyProcessed {
		finalStatus = "failed"
		msg := FormatJobUserError(JobErrAllFailedKey, failed)
		errMsg = &msg
	} else if failed > 0 {
		msg := FormatJobUserError(JobErrPartialFailedKey, failed)
		errMsg = &msg
	}
	finalProgress := finalizeStepProgress(processingType, lastResult, finalStatus == "failed")
	finalJSON, err := marshalStepProgress(finalProgress)
	if err != nil {
		return fmt.Errorf("processing: final step_progress job=%s: %w", jobID, err)
	}
	finalStep := "done"
	if finalStatus == "failed" {
		finalStep = "failed"
	}
	finalMode := modeLabel
	if lastResult != nil && lastResult.AIProviderMode != "" {
		finalMode = lastResult.AIProviderMode
	}
	_, err = p.Pool.Exec(ctx, `
		UPDATE processing_jobs
		SET status = $2, processed_products = $3, error = $4, completed_at = now(), updated_at = now(),
			estimated_tokens = GREATEST(estimated_tokens, $5),
			current_step = $6, step_progress = $7::jsonb,
			ai_provider_mode = $8
		WHERE id = $1 AND status = 'running'`, jobID, finalStatus, processed, errMsg, tokenTotal, finalStep, finalJSON, finalMode)
	log.Printf("processing: finished job=%s status=%s processed=%d failed=%d mode=%s", jobID, finalStatus, processed, failed, finalMode)
	return err
}

// flushJobCountersAndProgress writes processed_products, token delta, and step_progress in one round-trip.
func (p *Pipeline) flushJobCountersAndProgress(ctx context.Context, jobID uuid.UUID, processingType string, result *StepResult, processed, tokenDelta int, mode string) error {
	prog := progressFromResult(processingType, result)
	b, err := marshalStepProgress(prog)
	if err != nil {
		return err
	}
	current := ""
	for i := len(prog) - 1; i >= 0; i-- {
		if prog[i].Status == "done" || prog[i].Status == "skipped" {
			current = prog[i].Step
			break
		}
	}
	if current == "" && len(prog) > 0 {
		current = prog[0].Step
	}
	_, err = p.Pool.Exec(ctx, `
		UPDATE processing_jobs
		SET processed_products = $2,
			estimated_tokens = estimated_tokens + $3,
			current_step = $4,
			step_progress = $5::jsonb,
			ai_provider_mode = $6,
			updated_at = now()
		WHERE id = $1`, jobID, processed, tokenDelta, current, b, mode)
	return err
}

// jobScopedCache holds per-job lookups reused across processOne calls.
type jobScopedCache struct {
	stdDefs               []StandardFieldDef
	brandPrompt           string
	language              string
	enhanceSystemTemplate string
	enhanceUserTemplate   string
	enhanceByLang         map[string]PromptTemplates
	// categoryPromptsByLang maps lower(trim(name)) → lang → sanitized prompt.
	categoryPromptsByLang map[string]company.LangPromptMap
	contentLanguages      []string
	// Entitlements snapshot — avoids EntitlementsForCompany N+1 per product.
	billingEnabled   bool
	canUseAI         bool
	allowEPREL       bool
	remainingCredits int
}

func (c *jobScopedCache) stepPolicy() StepPolicy {
	if c == nil || !c.billingEnabled {
		// No billing service (tests): allow gated steps so unit tests stay self-contained.
		return StepPolicy{AllowAI: true, AllowEPREL: true}
	}
	return StepPolicy{
		AllowAI:    c.canUseAI && c.remainingCredits > 0,
		AllowEPREL: c.allowEPREL,
	}
}

func (c *jobScopedCache) noteCreditDebit(debit int) {
	if c == nil || !c.billingEnabled || debit < 1 {
		return
	}
	c.remainingCredits -= debit
	if c.remainingCredits < 0 {
		c.remainingCredits = 0
	}
}

func (p *Pipeline) loadJobScopedCache(ctx context.Context, companyID, jobID uuid.UUID) jobScopedCache {
	var cache jobScopedCache
	cache.language = company.LoadLanguage(ctx, p.Pool, companyID)
	cache.contentLanguages = company.LoadContentLanguages(ctx, p.Pool, companyID)
	stdDefs, err := p.loadEnabledStandardFields(ctx, companyID)
	if err != nil {
		log.Printf("processing: load standard fields job=%s company=%s err=%s", jobID, companyID, TruncateError(err))
	} else {
		cache.stdDefs = stdDefs
	}
	if p.Billing != nil {
		cache.billingEnabled = true
		if ent, err := p.Billing.EntitlementsForCompany(ctx, companyID); err != nil {
			log.Printf("processing: load entitlements job=%s company=%s err=%s", jobID, companyID, TruncateError(err))
		} else {
			cache.canUseAI = ent.CanUseAI
			cache.allowEPREL = ent.CanUseEPREL
			cache.remainingCredits = ent.RemainingCredits
		}
		if p.Billing.AIBrandApplyAllowed(ctx, companyID) {
			if brand, err := company.LoadBrand(ctx, p.Pool, companyID); err == nil {
				cache.brandPrompt = brand.PromptBlock()
			}
		}
	}
	if p.Prompts != nil {
		cache.enhanceByLang = make(map[string]PromptTemplates, len(cache.contentLanguages)+1)
		langs := cache.contentLanguages
		if len(langs) == 0 {
			langs = []string{cache.language}
		}
		for _, lang := range langs {
			resolved, err := p.Prompts.Resolve(ctx, companyID, aiprompts.KeyProductEnhance, lang)
			if err != nil {
				log.Printf("processing: load prompts job=%s company=%s lang=%s err=%s", jobID, companyID, lang, TruncateError(err))
				continue
			}
			cache.enhanceByLang[lang] = PromptTemplates{System: resolved.SystemTemplate, User: resolved.UserTemplate}
			if lang == cache.language {
				cache.enhanceSystemTemplate = resolved.SystemTemplate
				cache.enhanceUserTemplate = resolved.UserTemplate
			}
		}
		if cache.enhanceSystemTemplate == "" {
			if resolved, err := p.Prompts.Resolve(ctx, companyID, aiprompts.KeyProductEnhance, cache.language); err != nil {
				log.Printf("processing: load prompts job=%s company=%s err=%s", jobID, companyID, TruncateError(err))
			} else {
				cache.enhanceSystemTemplate = resolved.SystemTemplate
				cache.enhanceUserTemplate = resolved.UserTemplate
				cache.enhanceByLang[cache.language] = PromptTemplates{System: resolved.SystemTemplate, User: resolved.UserTemplate}
			}
		}
	}
	cache.categoryPromptsByLang = p.loadCategoryEnhancePrompts(ctx, companyID, jobID)
	return cache
}

func (p *Pipeline) loadCategoryEnhancePrompts(ctx context.Context, companyID, jobID uuid.UUID) map[string]company.LangPromptMap {
	rows, err := p.Pool.Query(ctx, `
		SELECT name, COALESCE(prompt, '{}'::jsonb)
		FROM categories
		WHERE company_id = $1 AND prompt <> '{}'::jsonb`, companyID)
	if err != nil {
		log.Printf("processing: load category prompts job=%s company=%s err=%s", jobID, companyID, TruncateError(err))
		return nil
	}
	defer rows.Close()
	out := make(map[string]company.LangPromptMap)
	for rows.Next() {
		var name string
		var raw []byte
		if err := rows.Scan(&name, &raw); err != nil {
			log.Printf("processing: scan category prompt job=%s err=%s", jobID, TruncateError(err))
			continue
		}
		key := strings.ToLower(strings.TrimSpace(name))
		if key == "" {
			continue
		}
		m, err := company.DecodeLangPromptMap(raw)
		if err != nil || !company.HasAnyPrompt(m) {
			continue
		}
		// Re-sanitize with category rune cap.
		cleaned := company.LangPromptMap{}
		for lang, prompt := range m {
			p := strings.TrimSpace(security.SanitizePrompt(prompt, catalog.MaxCategoryPromptRunes))
			if p == "" {
				continue
			}
			cleaned[lang] = p
		}
		if company.HasAnyPrompt(cleaned) {
			out[key] = cleaned
		}
	}
	if err := rows.Err(); err != nil {
		log.Printf("processing: category prompts rows job=%s err=%s", jobID, TruncateError(err))
	}
	return out
}

func categoryEnhancePromptFor(prompts map[string]company.LangPromptMap, category, language string) string {
	if len(prompts) == 0 {
		return ""
	}
	key := strings.ToLower(strings.TrimSpace(category))
	if key == "" {
		return ""
	}
	return company.PromptForLanguage(prompts[key], language)
}

func progressFromResult(processingType string, result *StepResult) []StepProgress {
	base := InitialStepProgress(processingType)
	if result == nil {
		return base
	}
	seen := map[string]map[string]any{}
	if steps, ok := result.GPTResponse["steps"].([]any); ok {
		for _, s := range steps {
			m, _ := s.(map[string]any)
			if m == nil {
				continue
			}
			name, _ := m["step"].(string)
			seen[name] = m
		}
	}
	for i := range base {
		raw := seen[base[i].Step]
		if raw == nil {
			base[i].Status = "pending"
			continue
		}
		status := "done"
		note := ""
		if r, ok := raw["raw"].(map[string]any); ok {
			if st, ok := r["status"].(string); ok && st != "" {
				switch st {
				case "skipped", "unchanged":
					status = "skipped"
				case "failed", "parse_failed":
					// AI enhance kept prior copy; surface as failed so UI is not "done" with a cryptic note.
					status = "failed"
				}
			}
			if reason, ok := r["reason"].(string); ok {
				note = reason
			}
			if errStr, ok := r["error"].(string); ok && errStr != "" {
				note = errStr
			}
			if status == "failed" && note != "" && strings.Contains(strings.ToLower(note), "unmarshal") {
				note = "AI returned invalid JSON; kept original title/description"
			}
		}
		base[i].Status = status
		base[i].Note = note
	}
	return base
}

func finalizeStepProgress(processingType string, result *StepResult, failed bool) []StepProgress {
	prog := progressFromResult(processingType, result)
	for i := range prog {
		if prog[i].Status == "pending" || prog[i].Status == "running" {
			if failed {
				prog[i].Status = "failed"
			} else {
				prog[i].Status = "done"
			}
		}
	}
	return prog
}

type jobItem struct {
	ID, RawID uuid.UUID
	// Preloaded by hydrateJobItems (one query per claim batch).
	hydrated                                       bool
	gtin                                           string
	mappedBytes, rawBytes                          []byte
	priorName, priorDesc, priorHash, priorCategory string
	priorLocalized                                 company.LocalizedContent
	hasPrior                                       bool
}

// loadPendingItems claims the next pending job products atomically so concurrent
// ProcessJob workers (or overlapping invocations) cannot process the same row.
// Aged 'processing' rows (StuckAgeInterval, same as CleanupStuck) are also
// reclaimed — crash mid-item must not leave products unclaimable until the ops ticker.
func (p *Pipeline) loadPendingItems(ctx context.Context, jobID uuid.UUID, limit int) ([]jobItem, error) {
	rows, err := p.Pool.Query(ctx, `
		UPDATE processing_job_products
		SET status = 'processing', updated_at = now()
		WHERE id IN (
			SELECT id FROM processing_job_products
			WHERE job_id = $1 AND (
				status = 'pending'
				OR (status = 'processing' AND updated_at < now() - interval '`+StuckAgeInterval+`')
			)
			ORDER BY created_at
			LIMIT $2
			FOR UPDATE SKIP LOCKED
		)
		RETURNING id, raw_product_id`, jobID, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items := make([]jobItem, 0, limit)
	for rows.Next() {
		var it jobItem
		if err := rows.Scan(&it.ID, &it.RawID); err != nil {
			return nil, err
		}
		items = append(items, it)
	}
	return items, rows.Err()
}

// hydrateJobItems batch-loads raw product payloads + prior enhance hash for a claim batch.
// Avoids 2 QueryRow round-trips per processOne on the hot path.
func (p *Pipeline) hydrateJobItems(ctx context.Context, companyID uuid.UUID, items []jobItem) error {
	if len(items) == 0 {
		return nil
	}
	rawIDs := make([]uuid.UUID, len(items))
	byRaw := make(map[uuid.UUID][]int, len(items))
	for i := range items {
		rawIDs[i] = items[i].RawID
		byRaw[items[i].RawID] = append(byRaw[items[i].RawID], i)
	}
	rows, err := p.Pool.Query(ctx, `
		SELECT rp.id,
			rp.gtin,
			COALESCE(rp.mapped_data, '{}'::jsonb),
			COALESCE(rp.raw_data, '{}'::jsonb),
			COALESCE(pp.processed_name, ''),
			COALESCE(pp.processed_description, ''),
			COALESCE(pp.field_sources->>'enhance_input_hash', ''),
			COALESCE(pp.category, ''),
			COALESCE(pp.localized_content, '{}'::jsonb),
			(pp.id IS NOT NULL) AS has_prior
		FROM raw_products rp
		LEFT JOIN processed_products pp
			ON pp.company_id = rp.company_id AND pp.raw_product_id = rp.id
		WHERE rp.company_id = $1 AND rp.id = ANY($2::uuid[])`, companyID, rawIDs)
	if err != nil {
		return err
	}
	defer rows.Close()
	found := 0
	for rows.Next() {
		var rawID uuid.UUID
		var gtin string
		var mappedBytes, rawBytes, localizedBytes []byte
		var priorName, priorDesc, priorHash, priorCategory string
		var hasPrior bool
		if err := rows.Scan(&rawID, &gtin, &mappedBytes, &rawBytes, &priorName, &priorDesc, &priorHash, &priorCategory, &localizedBytes, &hasPrior); err != nil {
			return err
		}
		idxs := byRaw[rawID]
		if len(idxs) == 0 {
			continue
		}
		found++
		priorLocalized, _ := company.DecodeLocalizedContent(localizedBytes)
		for _, i := range idxs {
			items[i].hydrated = true
			items[i].gtin = gtin
			items[i].mappedBytes = mappedBytes
			items[i].rawBytes = rawBytes
			items[i].priorName = priorName
			items[i].priorDesc = priorDesc
			items[i].priorHash = priorHash
			items[i].priorCategory = priorCategory
			items[i].priorLocalized = priorLocalized
			items[i].hasPrior = hasPrior
		}
	}
	if err := rows.Err(); err != nil {
		return err
	}
	if found != len(byRaw) {
		return fmt.Errorf("hydrate: missing raw_products for claimed items (found %d of %d)", found, len(byRaw))
	}
	return nil
}

var errJobCancelled = errors.New("processing job cancelled")

// stopOnCancel interprets jobCancelled results for ProcessJob.
// nil means continue; errJobCancelled means clean stop; any other error fails closed.
func stopOnCancel(cancelled bool, err error) error {
	if err != nil {
		return err
	}
	if cancelled {
		return errJobCancelled
	}
	return nil
}

func marshalStepProgress(progress []StepProgress) ([]byte, error) {
	b, err := json.Marshal(progress)
	if err != nil {
		return nil, fmt.Errorf("marshal step_progress: %w", err)
	}
	return b, nil
}

// cancelPendingJobProducts marks pending/processing job items cancelled.
// Fail closed: callers must not report cancel success when this Exec fails.
func cancelPendingJobProducts(ctx context.Context, exec func(context.Context, string, ...any) (pgconn.CommandTag, error), jobID uuid.UUID) error {
	_, err := exec(ctx, `
		UPDATE processing_job_products SET status = 'cancelled', updated_at = now()
		WHERE job_id = $1 AND status IN ('pending', 'processing')`, jobID)
	if err != nil {
		return fmt.Errorf("cancel job products job=%s: %w", jobID, err)
	}
	return nil
}

func (p *Pipeline) jobCancelled(ctx context.Context, jobID uuid.UUID) (bool, error) {
	var status string
	err := p.Pool.QueryRow(ctx, `SELECT status FROM processing_jobs WHERE id = $1`, jobID).Scan(&status)
	if err != nil {
		return false, err
	}
	return status == "cancelled", nil
}

func (p *Pipeline) processOne(ctx context.Context, companyID, jobID uuid.UUID, it *jobItem, processingType string, engine *Engine, modeLabel string, usingBYOK bool, cache *jobScopedCache) (bool, int, StepResult, error) {
	if it == nil {
		return false, 0, StepResult{}, fmt.Errorf("processing: nil job item")
	}
	gtin := it.gtin
	mappedBytes := it.mappedBytes
	rawBytes := it.rawBytes
	if !it.hydrated {
		// Fail closed: claim batches must be hydrated before processOne.
		err := p.Pool.QueryRow(ctx, `
			SELECT gtin, COALESCE(mapped_data, '{}'::jsonb), COALESCE(raw_data, '{}'::jsonb)
			FROM raw_products WHERE id = $1 AND company_id = $2`, it.RawID, companyID).
			Scan(&gtin, &mappedBytes, &rawBytes)
		if err != nil {
			return false, 0, StepResult{}, err
		}
	}
	mapped := map[string]any{}
	raw := map[string]any{}
	if err := json.Unmarshal(mappedBytes, &mapped); err != nil {
		log.Printf("processing: unmarshal mapped job=%s raw=%s err=%s", jobID, it.RawID, TruncateError(err))
		mapped = map[string]any{}
	}
	if err := json.Unmarshal(rawBytes, &raw); err != nil {
		log.Printf("processing: unmarshal raw job=%s raw=%s err=%s", jobID, it.RawID, TruncateError(err))
		raw = map[string]any{}
	}

	// Keep raw_products.mapped_data as synced; enrich a process-time copy only.
	var stdDefs []StandardFieldDef
	var brandPrompt, language, enhanceSystemTemplate, enhanceUserTemplate string
	var categoryPromptsByLang map[string]company.LangPromptMap
	var contentLanguages []string
	var enhanceByLang map[string]PromptTemplates
	if cache != nil {
		stdDefs = cache.stdDefs
		brandPrompt = cache.brandPrompt
		language = cache.language
		enhanceSystemTemplate = cache.enhanceSystemTemplate
		enhanceUserTemplate = cache.enhanceUserTemplate
		categoryPromptsByLang = cache.categoryPromptsByLang
		contentLanguages = cache.contentLanguages
		enhanceByLang = cache.enhanceByLang
	}
	enriched := EnrichMapped(mapped)
	if len(stdDefs) > 0 {
		enriched = FillMissingStandardFields(enriched, raw, stdDefs)
	}
	if gtin == "" {
		gtin = stringFromMap(enriched, "gtin", "ean", "EAN", "upc")
	}

	in := ProductInput{
		GTIN:                  SanitizeText(gtin),
		Name:                  stringFromMap(enriched, "name", "title", "product_name"),
		Description:           stringFromMap(enriched, "description", "desc", "body"),
		Mapped:                enriched,
		Raw:                   raw,
		StandardFields:        stdDefs,
		BrandPrompt:           brandPrompt,
		Language:              language,
		ContentLanguages:      contentLanguages,
		EnhanceByLang:         enhanceByLang,
		EnhanceSystemTemplate: enhanceSystemTemplate,
		EnhanceUserTemplate:   enhanceUserTemplate,
		CategoryPromptsByLang: categoryPromptsByLang,
	}
	if it.hydrated {
		if it.hasPrior {
			in.PriorProcessedName = it.priorName
			in.PriorProcessedDescription = it.priorDesc
			in.PriorEnhanceHash = it.priorHash
			in.PriorCategory = it.priorCategory
			in.PriorLocalized = it.priorLocalized
		}
	} else {
		// Fallback path when hydrate was skipped (should not happen in ProcessJob).
		var priorName, priorDesc, priorHash, priorCategory string
		var localizedBytes []byte
		errPrior := p.Pool.QueryRow(ctx, `
			SELECT COALESCE(processed_name, ''), COALESCE(processed_description, ''),
					COALESCE(field_sources->>'enhance_input_hash', ''),
					COALESCE(category, ''),
					COALESCE(localized_content, '{}'::jsonb)
			FROM processed_products
			WHERE company_id = $1 AND raw_product_id = $2`, companyID, it.RawID).
			Scan(&priorName, &priorDesc, &priorHash, &priorCategory, &localizedBytes)
		if errPrior != nil && !errors.Is(errPrior, pgx.ErrNoRows) {
			log.Printf("processing: load prior enhance hash job=%s raw=%s err=%s", jobID, it.RawID, TruncateError(errPrior))
		} else if errPrior == nil {
			in.PriorProcessedName = priorName
			in.PriorProcessedDescription = priorDesc
			in.PriorEnhanceHash = priorHash
			in.PriorCategory = priorCategory
			in.PriorLocalized, _ = company.DecodeLocalizedContent(localizedBytes)
		}
	}

	if engine == nil {
		engine = p.Engine
	}
	if engine == nil {
		engine = &Engine{Vector: NoopVectorCategorizer{}}
	}
	policy := cache.stepPolicy()
	result, err := engine.RunSteps(ctx, companyID.String(), in, processingType, nil, policy)
	if err != nil {
		return false, 0, StepResult{}, err
	}

	attrsJSON, procAttrsJSON, gptJSON, sourcesJSON, err := marshalProcessOnePayload(result)
	if err != nil {
		return false, 0, result, err
	}
	providerMode := result.AIProviderMode
	if providerMode == "" {
		if modeLabel != "" {
			providerMode = modeLabel
		} else if result.TotalTokens > 0 {
			providerMode = AIProviderInternal
		} else {
			providerMode = AIProviderUnknown
		}
	}
	result.AIProviderMode = providerMode

	// Debit + persist atomically: insufficient credits cannot leave a catalog row,
	// and a failed upsert/mark rolls back the debit.
	tx, err := p.Pool.Begin(ctx)
	if err != nil {
		return false, 0, result, err
	}
	defer tx.Rollback(ctx)

	// Debit before persisting the deliverable so insufficient credits cannot yield free AI output.
	// BYOK (company key): skip managed credit burn for inference.
	// Hash-skip (ai_enhance_unchanged): skip flat product_processing debit — no LLM/rework.
	if p.Billing != nil && shouldDebitProductProcessing(usingBYOK, result) {
		if err := p.Billing.ConsumeCreditsTx(ctx, tx, companyID, result.TotalTokens, "product_processing"); err != nil {
			return false, 0, result, err
		}
		cache.noteCreditDebit(p.Billing.EstimateDebit(ctx, "product_processing", result.TotalTokens))
	}

	processedID, err := p.upsertProcessedProduct(ctx, tx, companyID, it.RawID, gtin, result, attrsJSON, procAttrsJSON, gptJSON, sourcesJSON, providerMode)
	if err != nil {
		return false, 0, result, err
	}

	_, err = tx.Exec(ctx, `
		UPDATE processing_job_products
		SET status = 'processed', processed_product_id = $2, error = NULL, updated_at = now()
		WHERE id = $1`, it.ID, processedID)
	if err != nil {
		return false, 0, result, err
	}
	if _, err := tx.Exec(ctx, `
		UPDATE raw_products SET is_processed = true, processing_status = 'processed', updated_at = now()
		WHERE id = $1 AND company_id = $2`, it.RawID, companyID); err != nil {
		return false, 0, result, err
	}
	if err := tx.Commit(ctx); err != nil {
		return false, 0, result, err
	}

	return true, result.TotalTokens, result, nil
}

// marshalProcessOnePayload serializes deliverable JSON before credit debit / persist.
// Fail closed: nil/partial payloads must not be written after a successful RunSteps.
func marshalProcessOnePayload(result StepResult) (attrs, procAttrs, gpt, sources []byte, err error) {
	if attrs, err = json.Marshal(result.Attributes); err != nil {
		return nil, nil, nil, nil, fmt.Errorf("marshal attributes: %w", err)
	}
	if procAttrs, err = json.Marshal(result.ProcessedAttributes); err != nil {
		return nil, nil, nil, nil, fmt.Errorf("marshal processed_attributes: %w", err)
	}
	if gpt, err = json.Marshal(result.GPTResponse); err != nil {
		return nil, nil, nil, nil, fmt.Errorf("marshal gpt_response: %w", err)
	}
	if sources, err = json.Marshal(result.FieldSources); err != nil {
		return nil, nil, nil, nil, fmt.Errorf("marshal field_sources: %w", err)
	}
	return attrs, procAttrs, gpt, sources, nil
}

// upsertProcessedProductSQL is the race-safe persist for one raw product.
// Requires unique index processed_products_company_raw_uidx (019 migration).
const upsertProcessedProductSQL = `
	INSERT INTO processed_products (
		company_id, raw_product_id, product_id, name, category, description,
		processed_name, processed_description, status, attributes, processed_attributes,
		gpt_response, total_tokens, field_sources, ai_provider_mode, localized_content
	) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,'needs_review',$9,$10,$11,$12,$13,$14,$15::jsonb)
	ON CONFLICT (company_id, raw_product_id) DO UPDATE SET
		product_id = EXCLUDED.product_id,
		name = EXCLUDED.name,
		category = COALESCE(NULLIF(BTRIM(EXCLUDED.category), ''), processed_products.category),
		description = EXCLUDED.description,
		processed_name = EXCLUDED.processed_name,
		processed_description = EXCLUDED.processed_description,
		status = 'needs_review',
		attributes = EXCLUDED.attributes,
		processed_attributes = EXCLUDED.processed_attributes,
		gpt_response = EXCLUDED.gpt_response,
		total_tokens = COALESCE(processed_products.total_tokens, 0) + EXCLUDED.total_tokens,
		field_sources = EXCLUDED.field_sources,
		ai_provider_mode = EXCLUDED.ai_provider_mode,
		localized_content = EXCLUDED.localized_content,
		updated_at = now()
	RETURNING id`

func (p *Pipeline) upsertProcessedProduct(
	ctx context.Context,
	tx pgx.Tx,
	companyID, rawID uuid.UUID,
	gtin string,
	result StepResult,
	attrsJSON, procAttrsJSON, gptJSON, sourcesJSON []byte,
	providerMode string,
) (uuid.UUID, error) {
	localized := result.LocalizedContent
	if localized == nil {
		localized = company.LocalizedContent{}
	}
	if len(localized) == 0 && (strings.TrimSpace(result.ProcessedName) != "" || strings.TrimSpace(result.ProcessedDescription) != "") {
		localized[company.DefaultLanguage] = company.LocalizedFields{
			ProcessedName:        result.ProcessedName,
			ProcessedDescription: result.ProcessedDescription,
		}
	}
	locJSON, err := company.EncodeLocalizedContent(localized)
	if err != nil {
		return uuid.Nil, err
	}
	queryRow := p.Pool.QueryRow
	if tx != nil {
		queryRow = tx.QueryRow
	}
	var processedID uuid.UUID
	err = queryRow(ctx, upsertProcessedProductSQL,
		companyID, rawID, gtin, result.Name, result.Category, result.Description,
		result.ProcessedName, result.ProcessedDescription, attrsJSON, procAttrsJSON, gptJSON, result.TotalTokens, sourcesJSON, providerMode, string(locJSON),
	).Scan(&processedID)
	return processedID, err
}

func (p *Pipeline) ClaimNext(ctx context.Context) (uuid.UUID, error) {
	var id uuid.UUID
	err := p.Pool.QueryRow(ctx, `
		UPDATE processing_jobs
		SET status = 'running', started_at = now(), updated_at = now()
		WHERE id = (
			SELECT id FROM processing_jobs
			WHERE status = 'pending'
			ORDER BY priority DESC, created_at
			LIMIT 1
			FOR UPDATE SKIP LOCKED
		)
		RETURNING id`).Scan(&id)
	if errors.Is(err, pgx.ErrNoRows) {
		return uuid.Nil, pgx.ErrNoRows
	}
	return id, err
}