Ingest pipeline

Bronze: what lands first#

Bronze keeps the source bytes, so a replay re-derives everything downstream without calling a vendor again.

• raw_events holds the three canonical streams and nothing else: orders, creators and inventory. Each row stores the exact payload and a sha256 of its key-sorted serialization. A unique index over organization, connector type, stream, external id and payload hash, with an insert that ignores conflicts, makes re-ingesting the same payload a no-op.
• import_staging_rows holds every other stream: ad spend, payouts, returns, search-console rows, CRM leads, dataset records and the rest. A staging row carries the mapping template and version that produced it, its dedupe key, and a status of queued, promoted, rejected or superseded.

Bronze is append-only. A correction is a later row, not an edit.

Silver: typed facts#

A per-stream promoter turns Bronze and staging rows into typed facts. Canonical marketplace streams upsert from the sync processor; every other stream goes through a named promoter dispatched by stream. Idempotency is structural:

1. Duplicates inside one batch collapse last-wins before the write, because Postgres refuses to update the same conflict key twice in one statement.
2. A chunked multi-row upsert applies ON CONFLICT DO UPDATE on a declared natural key, for example an order on its organization, platform, shop and platform order id. Where a key column is nullable, the constraint is NULLS NOT DISTINCT, because under the Postgres default a null-keyed row never conflicts with itself and duplicates on every sync.

ON CONFLICT DO NOTHING is deliberately absent from these paths: a re-pull of a closed day must be able to correct a late-settling number.

Gold: what dashboards read#

The agg_* rollups are the read surface: organization by day, by SKU and day, by creator and day, shopper by month, campaign, the content-intelligence rollups and the organization-dataset period rollup. They are refreshed nightly and on demand by the materialization functions in @yng/metrics.

The brand is a member of every rollup's unique key. A key that omits the brand produces a row that is the sum of all brands, which the organization-wide leg of the brand policy then shows under every brand. Where one rollup consumes another's output, order is fixed: API content is promoted into creator-daily facts before the creator rollup runs, and every output a sync writes must be consumed by a promoter or rollup, not only by a display router.

Watermarks and re-pulls#

Each connector stream keeps one watermark row with a cursor, a high-water mark, the last success and backfill times and a running row count.

• An incremental run resumes from the cursor, or from the high-water mark where the vendor offers a modified-since filter instead of a cursor.
• A backfill runs a bounded window and advances the high-water mark without persisting a cursor, because a bounded historical page token would corrupt the incremental frontier.
• A replay rebuilds Silver from Bronze and calls no vendor at all.

Vendors restate closed days, so several streams re-pull a short trailing window on purpose: the daily video finalize re-pulls the last closed days in Vietnam time, and each marketing source re-pulls its own look-back window. A rolling-window third-party pull prunes its own stale predecessors after each write, because supersede-by-natural-key does not apply to a window that shifts.

The durable handoff#

The web app never publishes to the queue directly. A mutation writes its business row and a job_outbox row in the same tenant-pinned transaction, so the state change and the intent to do asynchronous work commit or roll back together.

A dispatcher in the worker claims committed rows under a short lease with FOR UPDATE SKIP LOCKED, revalidates each against its Zod schema, publishes to the queue outside any transaction, then marks the row published in a second short transaction. The job id is stable and unique per queue, so a crash between "the queue accepted it" and "the row is marked published" is replay-safe. An expired lease is reclaimed by the claim predicate itself; there is no separate reaper.

Failure classification and the dead-letter table#

Failures are classified, never blanket-retried. These are permanent: an unprocessable payload, a connector authentication error, an envelope decrypt failure, a mapping-template error, a shop-to-brand mismatch, and deterministic Postgres not-null and foreign-key violations. Everything else is transient and retried with backoff.

A permanent failure writes one dead-letter row carrying the full payload, error code, message and attempt count, and is rethrown as unrecoverable so the queue stops retrying. Rows are deduplicated per connector, job type and error code among unresolved rows; a repeat bumps the count instead of adding a row. A later successful sync resolves that connector's open rows, so the table drains itself. A vendor rate limit is not a dead letter: it parks the connector with a cooldown and a degraded health status.

Uploads and mapping templates#

• Canonical streams (orders, creators, inventory, creator costs, targets, products) parse straight into Bronze, idempotent on the Bronze dedupe key.
• Every other stream goes through a mapping template. The template is data, not code: no per-source branch exists in the parser. It names the source kind, the target stream and one transform per column from a closed set, and its dedupe key must cover that stream's natural key or it cannot be saved. A template is immutable once active: an edit writes version N+1 and archives version N in the same transaction. The field list is in Import mapping template.
• A marketplace XLSX export often prepends a pivot sheet, so the decoder picks the largest sheet rather than the first. An upload is capped at 8 MiB.
• A Google Sheets tab is the same contract over a different transport: one tab per dataset and one template per tab. Independent sheet ranges are read concurrently under a hard cap, outside any database transaction, then merged in configured order.

Brand binding and shop ownership#

Every ingest write resolves its brand through one function, and the answer is never a silent default. A single-brand organization pins automatically; a multi-brand organization with no pinned brand is refused; a client-supplied brand that disagrees with the server-resolved one is refused; a brand the caller cannot view is refused. Marketplace writes also assert shop ownership: a shop maps to exactly one brand, an unregistered shop is claimed by its first writer, and a shop owned by another brand raises a mismatch that is classified as permanent.

Queues and schedules#

Ingest work is spread over named queues: a shared sync queue, per-marketplace queues, a marketing sync queue, a dataset queue, webhook queues and a manual-upload queue. Sync and manual upload retry five times with exponential backoff from 30 seconds; shorter operational queues retry two or three times. Recurring work runs as declarative job schedulers upserted by name at boot, and any undeclared scheduler is removed, so the schedule set in code is the schedule set that runs. Every pattern is pinned to Vietnam time.

Limits#

• Third-party estimates bypass Bronze, Silver and Gold and are never reconciled.
• Uploads above the size cap are refused; onboarding a new corpus shape may need operator assistance.
• A late restatement outside a stream's trailing re-pull window is picked up only by a backfill.

Design intent#

A transactional outbox between web and worker

• Decision. Durable asynchronous work is an outbox row written in the same transaction as the business row, published to the queue by the worker afterwards.
• Why. Publishing directly from a request either announces work for a row that then rolls back, or commits a row whose work is never queued when the process dies between the two steps.
• What it costs. Queue latency includes one dispatcher cycle, and every payload needs a named Zod schema that the dispatcher revalidates.
• How it is enforced. A static audit in the canonical gate refuses a direct queue publish or an after-commit callback for durable work in the web app; the lease and stable job id make publication replay-safe by construction.

Permanent and transient failures are handled differently

• Decision. Each failure is classified; permanent failures dead-letter once and stop retrying, transient ones retry with backoff, and rate limits park the connector.
• Why. Blanket retries turn one bad credential into thousands of identical errors and hide the real cause; blanket dead-lettering turns a network blip into manual work.
• What it costs. A new error type must be classified deliberately; an unclassified error is treated as transient.
• How it is enforced. One classifier in the connector core decides permanence; the dead-letter table's dedupe key prevents duplicates; a successful sync resolves open rows, and any refused dead-letter write is counted as a metric that must stay at zero.

Re-running ingest converges on the same numbers

• Decision. Bronze deduplicates by content hash, Silver upserts on natural keys with NULLS NOT DISTINCT, and watermarks advance only on a successful run.
• Why. Vendors restate recent days and jobs crash mid-run; a pipeline that is not idempotent inflates totals on every retry.
• What it costs. Every new stream declares its natural key, and a template's dedupe key must cover it before the template can be saved.
• How it is enforced. An audit refuses a nullable column in an upsert target without NULLS NOT DISTINCT; the dedupe contract is validated when a template is saved; CI runs the real promoters over adversarial fixtures on a fresh database before reconciling.

Related#

• Data planes and the warehouse
• System architecture
• Import mapping template
• Connectors and data planes
• Upload CSV or Excel files