Data Flow

This page traces the major data flows through the platform from store upload to downstream consumption.

Ingestion pipeline (hot landing path)

The core pipeline processes POST /v1/sync/batch requests from the Flutter POS app.

Request shape

{
  "storeId": "<uuid>",
  "deviceId": "device-abc123",
  "batchId": "<uuid>",
  "schemaVersion": 1,
  "rows": [
    {
      "seq": 1001,
      "op": "upsert",
      "table": "appointments",
      "pk": "apt-001",
      "version": 1,
      "changedAt": 1718150400000,
      "payload": { ... }
    }
  ]
}

Pipeline steps (`BatchApplyService.ApplyAsync()`)

Request arrives
       │
1. Set tenant      ── set_config('app.store_id', storeId) for RLS
       │
2. Resolve store   ── Check active flag + token validation + writer-lease CAS
       │              Unauthorized → 401, Forbidden → 403, DeviceConflict → 409
       │
3. Duplicate batch ── If batchId already seen → return OK immediately (idempotent)
       │
4. Validate/dedupe ── C# batch-internal: reject bad rows, LWW-dedupe within batch
       │              Must use SAME ordering as the SQL guard
       │
5. Clock-skew clamp── ChangedAtGuard.Clamp() bounds changed_at to server_now + skew
       │
6. LWW upsert      ── unnest() INSERT ... ON CONFLICT (store_id,source_table,pk)
       │              DO UPDATE WHERE EXCLUDED.changed_at > existing
       │                                OR (equal AND EXCLUDED.seq > existing)
       │              Returns landed keys via RETURNING
       │
7. Suppression      ── accepted - landed = suppressed → WARN log
   detection           (prevents silent data loss)
       │
8. Record audit    ── ingest_batches (stats) + ingest_rejections (per-row)
       │
9. COMMIT          ── Single READ COMMITTED transaction

Database: `ingestion` schema

Table	Purpose
`ingest_rows`	Core landing table: `(store_id, source_table, pk)` composite PK, JSONB `payload`, `op` (`upsert`/`delete`)
`ingest_batches`	Batch audit ledger (row_count, rejected_count, suppressed_count)
`ingest_rejections`	Per-rejected-row audit (seq, table, pk, reason)
`stores`	Per-store control row (active, token_hash, expected_device_id, writer_lease_epoch)
`capability_cursor`	Single-row table tracking last-seen `capability_seq` from Accounts

Key guardrails

delete writes a tombstone (op='delete', payload=NULL), never a hard delete
NUL-byte rejection: BatchValidator rejects JSON payloads with \u0000 (PostgreSQL jsonb can't hold U+0000) — a single bad row would otherwise reject the entire batch
ingest_app role (NOLOGIN): The request path does SET ROLE ingest_app for RLS. This role is never used as a connection username

DataHub read substrate

DataHub (src/SalonCloud.DataHub.Infrastructure) is a migration-only project — it has no runtime service. It creates PostgreSQL views in the data_hub schema, all with security_invoker = true.

How it works

Downstream services read DataHub views — never ingestion.ingest_rows directly. The views are live queries, always current, with no replication lag or materialization.

View categories

Category	Views	Access
Raw passthrough	`data_hub.rows`	`datahub_app` only (PII vector)
Typed projections	`customers`, `tickets`, `payments`, `appointments`, `staff`, `services`, `staff_schedules`, `staff_time_off`, `appointment_services`, `store_settings`, `categories`	Both roles
PII superset	`customers_pii` (name, phone, email, notes, birthday)	`datahub_app` only
Derived/operational	`staff_service_categories`, `store_sync_liveness`, `store_media`	Both roles
Meta-observability	`table_catalog` (typed flag), `field_drift` (untyped payload keys)	`datahub_maint` only

Fail-safe casts

A critical invariant: a cast raising in a view's target list takes down the query for every store. DataHub provides fail-safe functions that return NULL instead of raising:

Function	Returns
`try_bigint(text)`	`bigint` or NULL
`try_int(text)`	`int` or NULL
`try_numeric(text)`	`numeric` or NULL
`try_timestamp(text)`	`timestamptz` or NULL
`try_jsonb(text)`	`jsonb` or NULL

All are IMMUTABLE STRICT PARALLEL SAFE. Every typed view uses these for payload casts. Consuming code must IsDBNull-guard nullable columns.

PII roles

Role	RLS Scope	Can Read
`datahub_app`	Per-store (`app.store_id` must be set)	All typed views + `rows` + `customers_pii`
`datahub_maint`	Cross-tenant (`USING(true)`)	Typed views + `table_catalog` + `field_drift`. NOT `rows` or `customers_pii`

Current consumers:

Booking — reads via datahub_maint (cross-tenant; adds its own WHERE store_id = ...)
Accounts — reads store_settings via datahub_maint
No per-store RLS consumer exists yet; future store-facing query APIs should use datahub_app

Capability mirror

Ingestion and Booking maintain local mirrors of store state from Accounts via a pull-based pattern.

Accounts                        Ingestion / Booking
────────                        ───────────────────
capability_changes table        capability_cursor table
  (monotonic capability_seq)      (last-seen seq)

Every 15s:
  GET /internal/store-capabilities?since={cursor}&limit={n}
  ─────────────────────────────────────────────────────►
  ◄── [{storeId, active, tokenHash, timezone, ...}, ...]
                                                         │
                                       Apply via ICapabilitySink
                                       (seq-guarded, idempotent)
                                       Advance cursor

Key property: pull, not push. A downed consumer catches up naturally on restart by reading from its cursor. The cursor is a keyset over a monotonic sequence, so store state changes mid-scan are still seen.

Accounts can nudge consumers for lower latency via POST /internal/capabilities/nudge.

Booking flow

Public consumer endpoints (`/v1/book/`)

Consumer           Booking Service
────────           ───────────────
  │
  │ GET /{slug}/profile          → store name, timezone, currency
  │ GET /{slug}/services         → active bookable services
  │ GET /{slug}/availability     → slot start-times (rate-limited)
  │ POST /pow                    → proof-of-work challenge
  │ POST /{slug}/otp/request     → request OTP (always 202)
  │ POST /{slug}/otp/verify      → verify OTP → consumer token
  │ POST /{slug}/holds           → pre-identity slot hold (PoW-gated)
  │ DELETE /{slug}/holds/{id}    → release hold
  │ POST /{slug}/bookings        → create booking (needs consumer token)
  │
  │ Self-service management:
  │ GET /manage/{token}          → manage view
  │ POST /manage/{token}/cancel  → cancel
  │ POST /manage/{token}/reschedule → reschedule

Booking reconciliation

Booking runs a BookingReconcileHostedService (every 30s) that reads data_hub.appointments for appointments with cloud_booking_id set, reconciles them against local booking state, and sends notifications for changes.

Write-back to store

When Booking creates/modifies an appointment, it writes an outbox record. The store app polls GET /v1/booking/outbox and acks via POST /v1/booking/outbox/ack.

Store lifecycle

1. Store app self-onboards       POST /onboard (Accounts)
       │
2. Admin activates store         POST /v1/ops/stores/{id}/activate (Admin → Accounts)
       │                          → sets active=true, creates writer lease, provisions device token
       │
3. Store app receives token      Via capability mirror →
       │                          device registers with token
       │
4. Store app starts syncing      POST /v1/sync/batch (Ingestion)
       │
5. Admin manages entitlements    PUT /v1/ops/stores/{id}/entitlements/{feature}
   and devices                   POST /v1/ops/stores/{id}/devices/{deviceId}/revoke

Push and messaging

Booking / Accounts / Observability
       │
       │ POST /internal/push/nudge or /internal/sms/send
       ▼
   ┌───────┐     ┌───────────┐
   │ Push  │     │ Messaging │
   └───┬───┘     └─────┬─────┘
       │               │
       ▼               ▼
  FCM/APNs         Twilio / Gmail SMTP
  (store devices)  (consumers, operators)

Push delivers in-app nudges to store devices (new booking alert, etc.)
Messaging handles email (OTP codes, booking confirmations) and SMS (OTP, status webhooks). In development, both use logging transports instead of real providers

More on deployment and operations in Operations.

#Data Flow

#Ingestion pipeline (hot landing path)

#Request shape

#Pipeline steps (BatchApplyService.ApplyAsync())

#Database: ingestion schema

#Key guardrails

#DataHub read substrate

#How it works

#View categories

#Fail-safe casts

#PII roles

#Capability mirror

#Booking flow

#Public consumer endpoints (/v1/book/)

#Booking reconciliation

#Write-back to store

#Store lifecycle

#Push and messaging