MSO Cloud · Documentation

Ingest pipeline

Source: docs/architecture/reference/ingest-pipeline.md Updated 2026-09-21
On this page

How a record travels from a vendor API, a spreadsheet tab or an uploaded file to a number on a dashboard, and the four mechanisms that make a re-run land on the same answer. This is C4 Level 3 for ingest; the containers around it are in System Architecture.

flowchart TB
  subgraph IN["Entry points"]
    direction LR
    API["Vendor API<br/>scheduled pull"]
    UPL["CSV / XLSX upload"]
    SHT["Google Sheets<br/>tab binding"]
  end
  BR["Bronze<br/>raw_events · import_staging_rows"]
  SI["Silver<br/>orders · creator_performance_daily ·<br/>inventory_snapshots · promoted streams"]
  GO["Gold<br/>agg_org_day · agg_org_sku_day ·<br/>agg_org_creator_day · agg_org_campaign"]
  RD["Dashboards · reports · Cube ·<br/>data assistant · decision engine"]
  MI["market_intel store<br/>labelled estimate"]
  DLQ["sync_dead_letters"]
  IN --> BR
  BR -->|"promoters re-derive"| SI
  SI --> GO
  GO --> RD
  IN -.->|"third-party estimates skip the warehouse"| MI
  MI -.->|"never a first-party KPI"| RD
  BR -.->|"permanent failure, once"| DLQ

Bronze — what lands first#

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

  • raw_events holds the three canonical streams and nothing else: orders, creators, inventory. Each row stores the exact payload plus a sha256 of its key-sorted serialization. The unique index raw_events_dedupe_uq covers (org_id, connector_type, stream_name, external_id, payload_hash), and writeBronze inserts with ON CONFLICT DO NOTHING, so re-ingesting the same payload is 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. The canonical marketplace streams upsert directly from the sync processor; every other stream goes through a named promoter (promoteAdSpend, promoteBudgetPlan, promoteCrmLead, promoteContentMention, promoteDatasetRecord, promoteSocialMention and their siblings, dispatched by promoteStagedStream).

Idempotency is structural, in two steps:

  1. Intra-batch duplicates collapse last-wins before the write (dedupeLastWins), because Postgres refuses to update the same conflict key twice in one statement.
  2. A chunked multi-row upsert then applies ON CONFLICT DO UPDATE on a declared natural key — orders on (org_id, platform, platform_shop_id, platform_order_id), video_product_performance on (org_id, platform, video, product, sku, metric_date), and so on. Where a key column is nullable the constraint is declared NULLS NOT DISTINCT, because under the Postgres default a null-keyed row never conflicts with itself and silently duplicates on every sync.

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

Gold — what dashboards read#

The agg_* rollups are the read surface: agg_org_day, agg_org_sku_day, agg_org_creator_day, agg_org_shopper_month, agg_org_campaign, the content-intelligence rollups agg_content_dim_day, agg_content_dim_total and agg_content_cross_day, and the tenant-dataset rollup agg_org_dataset_period. They are refreshed by refreshOrgMaterializations and its per-table functions in the metrics package, nightly and on demand.

brand_id is a member of every one of their unique keys. A rollup key that omits the brand produces a row that is the sum of all brands, and the permissive null-brand leg of the row-level-security policy then makes that row visible under every brand.

Refresh order matters where one rollup consumes another's output: promoteApiContentToCreatorDaily runs before refreshAggOrgCreatorDay.

Watermarks and re-pulls#

connector_watermarks holds one row per (connector_id, stream_name) with last_cursor, high_water_mark, last_success_at, last_backfill_at 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 few closed days in Vietnam time and self-heals late-settling analytics through the upsert. Each marketing core re-pulls its own lookback window for the same reason. A rolling-window market-intelligence pull prunes its own stale predecessors after each persist, because supersede-by-natural-key does not apply to a window that shifts.

The durable handoff#

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

The dispatcher in the worker claims committed rows under a short lease, publishes to the queue outside any transaction, then marks the row published in a second short transaction. The job id is stable and colon-free, and (queue_name, job_id) is unique, 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, so no separate reaper exists.

Failure classification and the dead-letter queue#

Failures are classified, never blanket-retried. isPermanentSyncError treats these as permanent: an unprocessable payload, a connector authentication error, an envelope decrypt failure, a template specification error, a shop-to-brand mismatch, and the deterministic Postgres constraint violations 23502 (not null) and 23503 (foreign key).

A permanent failure writes one sync_dead_letters row through writeDeadLetter and is rethrown as unrecoverable so the queue stops retrying it. The row carries the full payload for replay, the error code and message, and the attempt count. Deduplication is on (connector_id, job_type, error_code) among unresolved rows, or (queue, job_type, error_code) for a failure with no connector — a repeat bumps the attempt count instead of adding a row.

The queue drains itself: a later successful sync resolves that connector's open rows. 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#

Two upload paths exist, and they land in different places.

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 anywhere in the parser. A template names the source kind, the target stream, and one transform per column from a closed set, and its dedupe_key must be a superset of that stream's natural key — validateDedupeContract refuses to save one that is not. A template is immutable once active: an edit writes version N+1 and archives version N in the same transaction, enforced by a NULLS NOT DISTINCT unique index on (org_id, brand_id, source_kind, version).

Two file-format rules earn their place. A marketplace XLSX export often prepends a pivot sheet, so the decoder picks the largest sheet rather than the first one. An upload is capped at 8 MiB.

A Google Sheets tab is the same contract with a different transport: one tab per dataset, one mapping template per tab, and a tab may only feed a dataset whose records are its rows.

Brand binding and shop ownership#

Every ingest write resolves its brand through resolveIngestBrand, and the answer is never a silent default:

  • A single-brand organization pins automatically.
  • A multi-brand organization with no pinned brand is refused with ingest_requires_brand.
  • A client-supplied brand that disagrees with the server-resolved brand is refused with ingest_brand_conflict.
  • A brand the caller cannot view is refused as forbidden.

Marketplace writes additionally assert shop ownership. The platform_shops registry maps (org_id, platform, canonical_shop_id) to exactly one brand; an unregistered shop is claimed by the first writer, and a shop owned by another brand raises a mismatch, which the classifier above treats as permanent.

Queues and schedules#

Ingest work is spread over named queues — a shared sync queue plus per-platform queues for TikTok Shop, Shopee, Lazada and file work, a sync-marketing queue, a dataset queue, webhook queues, and a manual-upload queue. Sync and manual upload retry five times with exponential backoff from 30 seconds; the AI queue retries three times from 10 seconds; the shorter operational queues retry twice.

Recurring work runs as declarative Job Schedulers upserted by name at boot, with any undeclared scheduler swept away, so the schedule set in code is the schedule set that runs. Every pattern is pinned to Vietnam time. The ingest-shaped entries cover incremental order polling every few minutes, hourly inventory and live polling, a daily video finalize, daily market-intelligence and product syncs, a cadence sweep for marketing and dataset connectors, a nightly seven-day order backfill, credential refresh, and staging purges.