Running the enrich pipeline durably

The enrichment pipeline has one job: turn a raw row into typed attributes and indexing surfaces, then return an explicit success, quarantine, or failure result. Durability belongs around that function. Your workflow platform schedules steps and retries them; your EnrichStore persists the row state.

Three primitives, one pipeline

enrich handles typed extraction and quality gates. resolve uses a CandidateProvider to score and cluster related rows. search consumes a Retriever; search and facets are the same backend capability, not separate product systems.

The package graph makes those boundaries visible:

Layer	Contract	I/O
Pure transform	`enrich` / `enrichRow`	Only the injected `GenerateFn` and optional image/embed closures
Capability factory	`createEnricher`	`EnrichStore`, `CandidateProvider`, and caller-owned model closures
Backend bundle	`@samesake/postgres` or another adapter	Backend-specific persistence, blocking, retrieval, and facets

Wire a capability factory

import { createEnricher } from "@samesake/enrich";
import { catalog } from "./catalog.ts";
import { generate, embed } from "./models.ts";
import { store } from "./store.ts";

const enricher = createEnricher({
  collection: catalog,
  generate,
  embed,
  store,
  concurrency: 8,
});

await enricher.upsert(rows);
const enriched = await enricher.enrich({ limit: 500 });
const groups = await enricher.resolve({ limit: 500 });

products is a neutral starter preset; a custom definePreset bundle or a domain preset can supply the fields, stages, surfaces, and evaluation targets. The collection still supplies the model and dimension declarations for its own embedding spaces.

createEnricher exposes upsert, enrich, resolve, retryFailed, and evaluate. resolve requires both a dedup configuration and a store with enriched-row loading and candidate generation. The scorer is pure; the candidate provider is where a backend’s blocking strategy lives.

Keep durability outside the core

The store owns the row state machine. A production store normally supports:

upsert and loadDirty for content-hash-based incremental work;
writeEnriched for ready or quarantined outputs;
recordFailure, loadRetryable, and markDead for the retry policy;
loadEnriched and candidates when resolution is enabled.

The workflow platform owns scheduling, retries, concurrency ceilings, and dead-letter handling. Keep the stages separate so a transient model failure does not repeat a successful upstream transform.

catalog event or schedule
          │
          ▼
        upsert ──► loadDirty ──► enrich ──► writeEnriched
                                      │
                                      ├── quarantine (successful, not indexable)
                                      └── failure → retry / dead letter
          │
          └──────────────► resolve (optional candidate blocking + pure scoring)

Backend choices are peers

Backend	Enrichment state	Candidate blocking	Search + facets
`@samesake/postgres`	SQL-backed `EnrichStore`	SQL candidate provider	SQL retrieval with exact facets
Cloudflare D1 + a vector index	D1-backed `EnrichStore`	Native index or consumer query	Consumer `Retriever` and declared facet behavior
Turbopuffer or another hosted index	Consumer-owned store adapter	Backend-native shortlist	Consumer `Retriever` and facet behavior

Postgres is the reference backend, not a requirement of the pure packages. See enrichment without the reference backend for a complete port-oriented shape and search without the reference backend for the retrieval side.