From b19373e2a4e9f255d2c77f176f2482a073904067 Mon Sep 17 00:00:00 2001 From: GreenEclipse Date: Sun, 16 Aug 2026 17:38:15 +0200 Subject: [PATCH] fix --- apps/api/cmd/api/main.go | 6 + apps/api/cmd/worker/main.go | 15 ++ apps/api/internal/config/config.go | 66 +++++++-- apps/api/internal/config/config_test.go | 64 +++++++++ apps/api/internal/processing/category_test.go | 13 ++ apps/api/internal/processing/pipeline.go | 23 ++- apps/api/internal/processing/stuck_cleanup.go | 50 +++++++ .../stuck_cleanup_integration_test.go | 132 ++++++++++++++++++ apps/api/internal/processing/v1_legacy.go | 11 ++ 9 files changed, 365 insertions(+), 15 deletions(-) diff --git a/apps/api/cmd/api/main.go b/apps/api/cmd/api/main.go index 497ea7c..e00c7fb 100644 --- a/apps/api/cmd/api/main.go +++ b/apps/api/cmd/api/main.go @@ -26,6 +26,12 @@ func main() { if err != nil { log.Fatalf("config: %v", err) } + if cfg.InsecureLocalProductionActive() { + slog.Warn("ALLOW_INSECURE_LOCAL_PRODUCTION=1 with loopback WEB_ORIGIN — not for public deploy", + "web_origin", cfg.WebOrigin, + "http_addr", cfg.HTTPAddr, + ) + } if cfg.ShouldWarnRateLimits() { slog.Warn(cfg.RateLimitWarningMessage(), "rate_limit_replicas", cfg.RateLimitReplicas, diff --git a/apps/api/cmd/worker/main.go b/apps/api/cmd/worker/main.go index 53877fc..cd22585 100644 --- a/apps/api/cmd/worker/main.go +++ b/apps/api/cmd/worker/main.go @@ -157,6 +157,21 @@ func main() { jobSlots := processing.NewJobSlots(processing.DefaultProcessingWorkers) syncSlots := jobs.NewSyncSlots(jobs.DefaultSyncWorkers) log.Printf("worker started - processing workers=%d sync workers=%d poll=%s LISTEN=%s,%s (ClaimNext SKIP LOCKED) + feed sync claim + support AI auto + woo/shopify claim + scheduled enqueue + billing cycles", jobSlots.Workers, syncSlots.Workers, cfg.ProcessingPollInterval, jobs.ChannelProcessingJobs, jobs.ChannelFeedSyncJobs) + if cfg.InsecureLocalProductionActive() { + log.Printf("worker WARNING: ALLOW_INSECURE_LOCAL_PRODUCTION=1 with loopback WEB_ORIGIN=%s — not for public deploy", cfg.WebOrigin) + } + + // Before claiming a fresh heartbeat, reclaim running orphans left by a prior + // crash/SIGTERM (ClaimNext only selects pending). Skip when another worker is live. + probe := jobs.ProbeWorkerReadiness(ctx, pool, jobs.DefaultHeartbeatStaleAfter) + if probe.WorkerCheck == "missing" || probe.WorkerCheck == "stale" { + if res, err := processing.ReclaimOrphanedRunning(ctx, pool); err != nil { + log.Printf("worker orphan reclaim: %v", err) + } else if res.JobsRequeued > 0 || res.ProductsReset > 0 || res.SyncJobsRequeued > 0 { + log.Printf("worker orphan reclaim jobs_requeued=%d products_reset=%d sync_requeued=%d (prior worker %s)", + res.JobsRequeued, res.ProductsReset, res.SyncJobsRequeued, probe.WorkerCheck) + } + } if err := jobs.TouchHeartbeat(ctx, pool, jobs.ProcessingWorkerID); err != nil { log.Printf("worker heartbeat bootstrap: %v", err) diff --git a/apps/api/internal/config/config.go b/apps/api/internal/config/config.go index e429d29..06cf2e9 100644 --- a/apps/api/internal/config/config.go +++ b/apps/api/internal/config/config.go @@ -109,6 +109,12 @@ type Config struct { // Non-production always allows scrapes. Production without this flag: loopback only. MetricsPublic bool + // AllowInsecureLocalProduction permits loopback WEB_ORIGIN (http or https) when + // APP_ENV=production|prod. Env: ALLOW_INSECURE_LOCAL_PRODUCTION=1. + // Public (non-loopback) origins still require https. Use only for local systemd / + // npm run dev mislabeled as production — never for a public deploy. + AllowInsecureLocalProduction bool + // Postgres pgx pool (api + worker). Defaults preserve historical NewPool hardcodes // and add idle recycle + statement_timeout for multi-tenant churn. // See docs/ops-runtime.md § Postgres pgx pool and db.PoolOptions comments. @@ -183,8 +189,9 @@ func Load() (Config, error) { StripeWebhookSecret: os.Getenv("STRIPE_WEBHOOK_SECRET"), StripeMock: getenvBool("STRIPE_MOCK", false), StripePriceIDs: loadStripePriceIDs(), - MetricsPublic: getenvBool("METRICS_PUBLIC", false), - DBMaxConns: getenvInt("DB_MAX_CONNS", 20), + MetricsPublic: getenvBool("METRICS_PUBLIC", false), + AllowInsecureLocalProduction: getenvBool("ALLOW_INSECURE_LOCAL_PRODUCTION", false), + DBMaxConns: getenvInt("DB_MAX_CONNS", 20), DBMinConns: getenvInt("DB_MIN_CONNS", 2), DBMaxConnLifetime: getenvDuration("DB_MAX_CONN_LIFETIME", time.Hour), DBMaxConnLifetimeJitter: getenvDurationAllowZero("DB_MAX_CONN_LIFETIME_JITTER", 6*time.Minute), @@ -224,10 +231,21 @@ func (c Config) IsProduction() bool { // CookieSecure is true when session/CSRF cookies must carry the Secure flag. // Prefer SessionSecure; also force Secure when APP_ENV is production (defense in depth). +// Exception: insecure local production escape with http loopback WEB_ORIGIN follows SessionSecure +// so cookies work on http://localhost during mislabeled local deploys. func (c Config) CookieSecure() bool { + if c.InsecureLocalProductionActive() && strings.HasPrefix(strings.ToLower(strings.TrimSpace(c.WebOrigin)), "http://") { + return c.SessionSecure + } return c.SessionSecure || c.IsProduction() } +// InsecureLocalProductionActive reports APP_ENV=production|prod with the explicit +// loopback escape (ALLOW_INSECURE_LOCAL_PRODUCTION) and a loopback WEB_ORIGIN. +func (c Config) InsecureLocalProductionActive() bool { + return c.IsProduction() && c.AllowInsecureLocalProduction && isLoopbackWebOriginHost(c.WebOrigin) +} + // ShouldWarnRateLimits reports whether operators opted into multi-replica rate-limit // awareness or requested an unsupported shared backend. func (c Config) ShouldWarnRateLimits() bool { @@ -273,14 +291,8 @@ func (c Config) validate() error { if !c.IsProduction() { return nil } - if !c.SessionSecure { - return fmt.Errorf("SESSION_SECURE=true is required when APP_ENV=production") - } - if !strings.HasPrefix(strings.ToLower(strings.TrimSpace(c.WebOrigin)), "https://") { - return fmt.Errorf("WEB_ORIGIN must be https in production") - } - if isLoopbackWebOriginHost(c.WebOrigin) { - return fmt.Errorf("WEB_ORIGIN must not be localhost/loopback in production") + if err := c.validateProductionWebOrigin(); err != nil { + return err } if strings.TrimSpace(c.AppEncryptionKey) == "" { return fmt.Errorf("APP_ENCRYPTION_KEY is required in production") @@ -299,6 +311,40 @@ func (c Config) validate() error { return nil } +// validateProductionWebOrigin fails closed for public production: https + non-loopback +// WEB_ORIGIN and SESSION_SECURE. Loopback http(s) is allowed only with the explicit +// ALLOW_INSECURE_LOCAL_PRODUCTION escape (local systemd / npm run dev mislabeled as prod). +func (c Config) validateProductionWebOrigin() error { + origin := strings.TrimSpace(c.WebOrigin) + https := strings.HasPrefix(strings.ToLower(origin), "https://") + loopback := isLoopbackWebOriginHost(origin) + localEscape := c.AllowInsecureLocalProduction && loopback + + if localEscape { + // SESSION_SECURE is optional for http://localhost escape so cookies work locally. + if https && !c.SessionSecure { + return fmt.Errorf("SESSION_SECURE=true is required when APP_ENV=production and WEB_ORIGIN is https") + } + return nil + } + + // Prefer WEB_ORIGIN diagnostics for the common local-mislabeled-production crash loop + // (http://localhost + APP_ENV=production) before SESSION_SECURE. + if !https { + if loopback { + return fmt.Errorf("WEB_ORIGIN must be https in production (got %q). Public deploy: set WEB_ORIGIN=https://your.domain. Local systemd/npm run dev: set APP_ENV=development, or ALLOW_INSECURE_LOCAL_PRODUCTION=1 with loopback WEB_ORIGIN only — otherwise api/worker exit and leave processing_jobs status=running forever", origin) + } + return fmt.Errorf("WEB_ORIGIN must be https in production (got %q); set WEB_ORIGIN=https://your.public.domain", origin) + } + if loopback { + return fmt.Errorf("WEB_ORIGIN must not be localhost/loopback in production (got %q). Public deploy: use your https domain. Local only: APP_ENV=development or ALLOW_INSECURE_LOCAL_PRODUCTION=1", origin) + } + if !c.SessionSecure { + return fmt.Errorf("SESSION_SECURE=true is required when APP_ENV=production") + } + return nil +} + // validateSMTPConfig no longer fails closed at boot: SMTP credentials live in // admin platform settings (with optional env fallback). Incomplete SMTP_ENABLED // env is ignored until admin configures delivery. diff --git a/apps/api/internal/config/config_test.go b/apps/api/internal/config/config_test.go index 15e5301..a390722 100644 --- a/apps/api/internal/config/config_test.go +++ b/apps/api/internal/config/config_test.go @@ -217,6 +217,61 @@ func TestProductionValidateFailsClosed(t *testing.T) { } } +func TestProductionLoopbackWebOriginEscape(t *testing.T) { + base := Config{ + AppEnv: "production", + WebOrigin: "http://localhost:28472", + SessionSecure: false, + AppEncryptionKey: "enc", + TokenSigningSecret: "tok", + ProcessingPollInterval: 250 * time.Millisecond, + DBMaxConns: 20, + DBMinConns: 2, + DBMaxConnLifetime: time.Hour, + DBMaxConnIdleTime: 5 * time.Minute, + DBHealthCheckPeriod: time.Minute, + DBStatementTimeout: 30 * time.Second, + } + + denied := base + if err := denied.validate(); err == nil { + t.Fatal("expected http loopback WEB_ORIGIN rejected without escape") + } else if !strings.Contains(err.Error(), "ALLOW_INSECURE_LOCAL_PRODUCTION") { + t.Fatalf("error should mention escape hatch: %v", err) + } + + allowed := base + allowed.AllowInsecureLocalProduction = true + if err := allowed.validate(); err != nil { + t.Fatal(err) + } + if !allowed.InsecureLocalProductionActive() { + t.Fatal("expected insecure local production active") + } + if allowed.CookieSecure() { + t.Fatal("http loopback escape should not force CookieSecure") + } + + httpsLocal := allowed + httpsLocal.WebOrigin = "https://localhost:28472" + if err := httpsLocal.validate(); err == nil { + t.Fatal("expected SESSION_SECURE required for https loopback escape") + } + httpsLocal.SessionSecure = true + if err := httpsLocal.validate(); err != nil { + t.Fatal(err) + } + + // Escape must not weaken public (non-loopback) http. + publicHTTP := base + publicHTTP.AllowInsecureLocalProduction = true + publicHTTP.WebOrigin = "http://app.example.com" + publicHTTP.SessionSecure = true + if err := publicHTTP.validate(); err == nil { + t.Fatal("expected public http WEB_ORIGIN rejected even with escape flag") + } +} + func TestValidateSMTPEnabledDoesNotRequireEnvHost(t *testing.T) { // SMTP credentials live in admin platform settings; boot must succeed with SMTP_ENABLED=true and empty host. cfg := Config{ @@ -298,6 +353,15 @@ func TestCookieSecure(t *testing.T) { if (Config{AppEnv: "development", SessionSecure: false}).CookieSecure() { t.Fatal("development without SessionSecure should not enable CookieSecure") } + insecureLocal := Config{ + AppEnv: "production", + WebOrigin: "http://localhost:28472", + AllowInsecureLocalProduction: true, + SessionSecure: false, + } + if insecureLocal.CookieSecure() { + t.Fatal("insecure local production http escape should not force CookieSecure") + } } func TestLoadSessionSecureDefaultsWithAppEnv(t *testing.T) { diff --git a/apps/api/internal/processing/category_test.go b/apps/api/internal/processing/category_test.go index cb2f0e5..f7dadf6 100644 --- a/apps/api/internal/processing/category_test.go +++ b/apps/api/internal/processing/category_test.go @@ -181,6 +181,19 @@ func TestCoerceCategoryToCompanyUniqueID_nameToUID(t *testing.T) { } } +func TestResolveCompanyCategoryUniqueID_usedByV1Projection(t *testing.T) { + t.Parallel() + names := map[string]string{"50": "Štedilniki"} + valid := map[string]struct{}{"50": {}} + // Poll/projection path: mapped name-only token must resolve like processOne. + if got := resolveCompanyCategoryUniqueID("Štedilniki", names, valid); got != "50" { + t.Fatalf("got %q want 50", got) + } + if got := resolveCompanyCategoryUniqueID("50", names, valid); got != "50" { + t.Fatalf("uid passthrough got %q", got) + } +} + func TestCategoryUniqueIDFromAny_nestedNameOnly(t *testing.T) { t.Parallel() got := categoryUniqueIDFromAny(map[string]any{"name": "Štedilniki"}) diff --git a/apps/api/internal/processing/pipeline.go b/apps/api/internal/processing/pipeline.go index b1354bc..cbf79f9 100644 --- a/apps/api/internal/processing/pipeline.go +++ b/apps/api/internal/processing/pipeline.go @@ -548,11 +548,13 @@ func (p *Pipeline) RetryJob(ctx context.Context, companyID, id uuid.UUID) (Job, firstStep = progress[0].Step } - itemStatuses := []string{"failed", "cancelled"} + // Include "processing": ProcessJob can fail after claim (hydrate/etc.) and leave + // rows stuck in processing — RetryJob must reclaim them or the job never drains. + itemStatuses := []string{"failed", "cancelled", "processing"} resetProcessed := false if job.Status == "completed" { // Completed retries should rerun the whole job, not immediately no-op. - itemStatuses = []string{"processed", "failed", "cancelled"} + itemStatuses = []string{"processed", "failed", "cancelled", "processing"} resetProcessed = true } @@ -677,6 +679,7 @@ func (p *Pipeline) ProcessJob(ctx context.Context, jobID uuid.UUID) error { items, err := p.loadPendingItems(ctx, jobID, batch) if err != nil { flushProgress(true) + _ = p.reclaimOrphanedProcessingItems(ctx, jobID) return err } if len(items) == 0 { @@ -687,24 +690,34 @@ func (p *Pipeline) ProcessJob(ctx context.Context, jobID uuid.UUID) error { if err := p.Pool.QueryRow(ctx, ` SELECT COUNT(*) FROM processing_job_products WHERE job_id = $1 AND processed_product_id IS NULL - AND status IN ('pending', 'processing')`, jobID).Scan(&open); err != nil { + AND status IN ('pending', 'processing')`, jobID).Scan(&open); err != nil { flushProgress(true) + _ = p.reclaimOrphanedProcessingItems(ctx, jobID) return fmt.Errorf("processing: count open items job=%s: %w", jobID, err) } if open > 0 { - if _, err := p.Pool.Exec(ctx, ` + ct, err := p.Pool.Exec(ctx, ` UPDATE processing_job_products SET status = 'pending', error = NULL, updated_at = now() - WHERE job_id = $1 AND status = 'processing' AND processed_product_id IS NULL`, jobID); err != nil { + WHERE job_id = $1 AND status = 'processing' AND processed_product_id IS NULL`, jobID) + if err != nil { flushProgress(true) + _ = p.reclaimOrphanedProcessingItems(ctx, jobID) return fmt.Errorf("processing: reclaim fresh processing job=%s: %w", jobID, err) } + // Fail closed: pending-only open that claim skipped must not spin forever. + if ct.RowsAffected() == 0 { + flushProgress(true) + _ = p.reclaimOrphanedProcessingItems(ctx, jobID) + return fmt.Errorf("processing: open items unclaimable job=%s open=%d", jobID, open) + } continue } break } if err := p.hydrateJobItems(ctx, companyID, items); err != nil { flushProgress(true) + _ = p.reclaimOrphanedProcessingItems(ctx, jobID) return fmt.Errorf("processing: hydrate batch job=%s: %w", jobID, err) } creditsStop := false diff --git a/apps/api/internal/processing/stuck_cleanup.go b/apps/api/internal/processing/stuck_cleanup.go index cd9da02..483aaf9 100644 --- a/apps/api/internal/processing/stuck_cleanup.go +++ b/apps/api/internal/processing/stuck_cleanup.go @@ -18,6 +18,56 @@ type StuckCleanupResult struct { SyncJobsMarkedFailed int64 } +// OrphanReclaimResult counts rows touched by ReclaimOrphanedRunning. +type OrphanReclaimResult struct { + JobsRequeued int64 + ProductsReset int64 + SyncJobsRequeued int64 +} + +// ReclaimOrphanedRunning resets in-flight work left by a dead worker process so +// ClaimNext / feed sync claim can pick it up again. Call only when the processing +// worker heartbeat is missing or stale (single-worker architecture) — never while +// another live worker may own the rows. +// +// Unlike CleanupStuck (age-gated fail), this requeues immediately: running → pending +// for jobs/sync jobs, and processing → pending for job products without a result. +func ReclaimOrphanedRunning(ctx context.Context, pool *pgxpool.Pool) (OrphanReclaimResult, error) { + var out OrphanReclaimResult + if pool == nil { + return out, fmt.Errorf("reclaim orphaned running: nil pool") + } + + ctJobs, err := pool.Exec(ctx, ` + UPDATE processing_jobs + SET status = 'pending', started_at = NULL, error = NULL, updated_at = now() + WHERE status = 'running'`) + if err != nil { + return out, fmt.Errorf("reclaim orphaned jobs: %w", err) + } + out.JobsRequeued = ctJobs.RowsAffected() + + ctProd, err := pool.Exec(ctx, ` + UPDATE processing_job_products + SET status = 'pending', error = NULL, updated_at = now() + WHERE status = 'processing' AND processed_product_id IS NULL`) + if err != nil { + return out, fmt.Errorf("reclaim orphaned products: %w", err) + } + out.ProductsReset = ctProd.RowsAffected() + + ctSync, err := pool.Exec(ctx, ` + UPDATE feed_sync_jobs + SET status = 'pending', started_at = NULL, error = NULL, completed_at = NULL, updated_at = now() + WHERE status = 'running'`) + if err != nil { + return out, fmt.Errorf("reclaim orphaned sync jobs: %w", err) + } + out.SyncJobsRequeued = ctSync.RowsAffected() + + return out, nil +} + // CleanupStuck aligns worker and admin stuck-cleanup semantics: // mark long-running jobs failed, reset stranded processing_job_products // from 'processing' back to 'pending', and fail aged running feed_sync_jobs diff --git a/apps/api/internal/processing/stuck_cleanup_integration_test.go b/apps/api/internal/processing/stuck_cleanup_integration_test.go index abea8e8..d40881f 100644 --- a/apps/api/internal/processing/stuck_cleanup_integration_test.go +++ b/apps/api/internal/processing/stuck_cleanup_integration_test.go @@ -200,6 +200,138 @@ func TestCleanupStuckResetsJobsAndProducts(t *testing.T) { } } +func TestReclaimOrphanedRunningRequeuesImmediately(t *testing.T) { + dsn := os.Getenv("DATABASE_URL") + if dsn == "" { + t.Skip("DATABASE_URL not set") + } + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + + pg, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatal(err) + } + defer pg.Close() + + var companyID uuid.UUID + err = pg.QueryRow(ctx, ` + SELECT company_id + FROM raw_products + WHERE company_id IS NOT NULL + ORDER BY updated_at DESC + LIMIT 1`).Scan(&companyID) + if errorsIsNoRows(err) { + t.Skip("no raw_products rows available") + } + if err != nil { + t.Fatal(err) + } + + var userID uuid.UUID + err = pg.QueryRow(ctx, `SELECT id FROM users ORDER BY created_at DESC LIMIT 1`).Scan(&userID) + if errorsIsNoRows(err) { + t.Skip("no users rows available") + } + if err != nil { + t.Fatal(err) + } + + var rawID uuid.UUID + err = pg.QueryRow(ctx, ` + SELECT id + FROM raw_products + WHERE company_id = $1 + ORDER BY updated_at DESC + LIMIT 1`, companyID).Scan(&rawID) + if errorsIsNoRows(err) { + t.Skip("no raw_products for selected company") + } + if err != nil { + t.Fatal(err) + } + + var jobID uuid.UUID + err = pg.QueryRow(ctx, ` + INSERT INTO processing_jobs ( + company_id, user_id, status, total_products, processed_products, + processing_type, started_at, updated_at + ) VALUES ($1, $2, 'running', 1, 0, 'full', now(), now()) + RETURNING id`, + companyID, userID, + ).Scan(&jobID) + if err != nil { + t.Fatal(err) + } + defer func() { + _, _ = pg.Exec(context.Background(), `DELETE FROM processing_job_products WHERE job_id = $1`, jobID) + _, _ = pg.Exec(context.Background(), `DELETE FROM processing_jobs WHERE id = $1`, jobID) + }() + + if _, err := pg.Exec(ctx, ` + INSERT INTO processing_job_products (job_id, raw_product_id, status, updated_at) + VALUES ($1, $2, 'processing', now())`, jobID, rawID); err != nil { + t.Fatal(err) + } + + var syncFeedID uuid.UUID + err = pg.QueryRow(ctx, ` + INSERT INTO input_feeds (company_id, name, url, feed_type, status, sync_interval_minutes, options) + VALUES ($1, 'orphan-reclaim-feed', 'https://example.com/orphan.csv', 'csv', 'active', 60, '{}'::jsonb) + RETURNING id`, companyID).Scan(&syncFeedID) + if err != nil { + t.Fatal(err) + } + defer func() { + _, _ = pg.Exec(context.Background(), `DELETE FROM feed_sync_jobs WHERE feed_id = $1`, syncFeedID) + _, _ = pg.Exec(context.Background(), `DELETE FROM input_feeds WHERE id = $1`, syncFeedID) + }() + + var syncID uuid.UUID + err = pg.QueryRow(ctx, ` + INSERT INTO feed_sync_jobs (feed_id, company_id, status, started_at, updated_at) + VALUES ($1, $2, 'running', now(), now()) + RETURNING id`, syncFeedID, companyID).Scan(&syncID) + if err != nil { + t.Fatal(err) + } + + res, err := ReclaimOrphanedRunning(ctx, pg) + if err != nil { + t.Fatal(err) + } + if res.JobsRequeued < 1 { + t.Fatalf("jobs_requeued=%d want >= 1", res.JobsRequeued) + } + if res.ProductsReset < 1 { + t.Fatalf("products_reset=%d want >= 1", res.ProductsReset) + } + if res.SyncJobsRequeued < 1 { + t.Fatalf("sync_requeued=%d want >= 1", res.SyncJobsRequeued) + } + + var jobStatus, productStatus, syncStatus string + if err := pg.QueryRow(ctx, `SELECT status FROM processing_jobs WHERE id = $1`, jobID).Scan(&jobStatus); err != nil { + t.Fatal(err) + } + if jobStatus != "pending" { + t.Fatalf("job status=%q want pending", jobStatus) + } + if err := pg.QueryRow(ctx, `SELECT status FROM processing_job_products WHERE job_id = $1`, jobID).Scan(&productStatus); err != nil { + t.Fatal(err) + } + if productStatus != "pending" { + t.Fatalf("product status=%q want pending", productStatus) + } + if err := pg.QueryRow(ctx, `SELECT status FROM feed_sync_jobs WHERE id = $1`, syncID).Scan(&syncStatus); err != nil { + t.Fatal(err) + } + if syncStatus != "pending" { + t.Fatalf("sync status=%q want pending", syncStatus) + } +} + func TestLoadPendingItemsReclaimsAgedProcessing(t *testing.T) { dsn := os.Getenv("DATABASE_URL") if dsn == "" { diff --git a/apps/api/internal/processing/v1_legacy.go b/apps/api/internal/processing/v1_legacy.go index 719ef00..75f5e8c 100644 --- a/apps/api/internal/processing/v1_legacy.go +++ b/apps/api/internal/processing/v1_legacy.go @@ -21,6 +21,7 @@ var ( v1BreakTagRe = regexp.MustCompile(`(?i)`) v1BlockEndRe = regexp.MustCompile(`(?i)`) ) + // ParseV1ProcessingType mirrors legacy parseV1ProcessingTypeFromBody. // Accepts string ("full" / step), JSON array of steps, or nil (defaults to full). func ParseV1ProcessingType(raw any) (storageValue string, responseValue any, err error) { @@ -311,6 +312,16 @@ func (p *Pipeline) LoadV1ProcessJobItems(ctx context.Context, companyID, jobID u if catStr == "" { catStr = categoryUniqueIDFromMaps(mapped, rawData) } + // Coerce display names → taxonomy unique_id (same as processOne / d161b28). + if catStr != "" && len(categoryNames) > 0 { + valid := make(map[string]struct{}, len(categoryNames)) + for uid := range categoryNames { + valid[uid] = struct{}{} + } + if resolved := resolveCompanyCategoryUniqueID(catStr, categoryNames, valid); resolved != "" { + catStr = resolved + } + } if catNameStr == "" && catStr != "" { if name, ok := categoryNames[catStr]; ok && name != "" { catNameStr = name