Database change data capture and continuous sync

A database row describes current state. Analytics usually needs the facts that produced that state: an order was placed, a subscription changed, or an account was closed.

GraphJSON does not provide a managed database connector. Use a change-data capture tool, transactional outbox, or incremental poller that you operate, then deliver a deliberately modeled analytical event through the logging API.

application database
      ↓
CDC stream, outbox, or incremental reader
      ↓
validate and map
      ↓
durable delivery queue
      ↓
GraphJSON collection

Keep the database authoritative. GraphJSON is the analytical destination, not the place where application state is reconstructed for serving requests.

Choose the source pattern#

Pattern	Best when	Primary risk
Transactional outbox	You own application writes and can add an outbox row in the same transaction	Outbox cleanup and consumer operations
Log-based CDC	You need changes from existing tables without changing every writer	Low-level row changes can lack business meaning
Incremental polling	The source exposes a reliable cursor such as monotonic version plus primary key	Missed or repeated rows around cursor boundaries
Periodic snapshot	Only current state matters and the dataset is bounded	Deletes and changes between snapshots are easy to miss

Prefer an outbox for business events you control. It records intent such as order_fulfilled, not merely “column status changed.”

Use log-based CDC when the database is the only dependable change source. Debezium’s architecture is a useful reference for reading database logs and resuming from offsets.

Polling is acceptable when all of these are true:

the ordering cursor cannot move backward
ties have a deterministic secondary key
deletions are represented explicitly
a reader can repeat an overlap safely
source retention exceeds the longest possible outage

An updated_at timestamp alone is usually not a safe cursor. Several rows can share it, clocks can move, and updates can occur after a page was read.

Prefer business events over row-shaped copies#

This row update:

{
  "table": "subscriptions",
  "operation": "update",
  "before": {"status": "trialing"},
  "after": {"status": "active"}
}

becomes a smaller analytical fact:

{
  "event": "subscription_status_changed",
  "event_id": "cdc:billing:000184:91",
  "account_id": "acct_7",
  "subscription_id": "sub_42",
  "previous_status": "trialing",
  "status": "active",
  "effective_at": 1785081600,
  "source_version": 314,
  "source": "billing-db",
  "schema_version": 2
}

Send effective_at as the GraphJSON request timestamp when it represents the business-effective time. Keep the source log position outside the event unless it is useful for reconciliation. Never forward a complete before and after row by default; it increases payload size, schema churn, and privacy exposure.

Maintain a mapping registry:

Source change	Analytical event	Collection
Order first enters `paid`	`order_paid`	`commerce_events`
Subscription plan changes	`subscription_plan_changed`	`billing_events`
Account is soft-deleted	`account_deleted`	`account_state_events`
Internal cache timestamp changes	None	None

Not every database mutation deserves an analytics event.

Use a canonical change envelope#

Normalize source-specific records before mapping them:

type SourceChange = {
  source: string;
  sourcePosition: string;
  entity: string;
  entityId: string;
  operation: "insert" | "update" | "delete";
  occurredAt: number;
  observedAt: number;
  sourceVersion: number | string;
  before?: Record<string, unknown>;
  after?: Record<string, unknown>;
};

This boundary lets source readers evolve independently from analytical event contracts. Validate the envelope, then allowlist fields into the destination event.

Treat source positions as opaque values. A database log sequence number, Kafka partition and offset, or (version, primary_key) cursor has meaning only to its source reader.

Represent inserts, updates, and deletes deliberately#

An insert can be:

a new business fact
the first observation of existing state
a bootstrap record with no historical transition

Mark bootstrap observations when they should not count as real-time creation:

{
  "event": "account_state_observed",
  "account_id": "acct_7",
  "state": "active",
  "observation_kind": "bootstrap",
  "source_version": 314
}

For updates, emit only when an approved field crosses a meaningful boundary. Repeatedly copying an unchanged row creates noise.

For deletes, never infer that a missing row was deleted. Capture a tombstone or explicit lifecycle event:

{
  "event": "account_deleted",
  "event_id": "cdc:accounts:000201:12",
  "account_id": "acct_7",
  "deletion_kind": "soft_delete",
  "source_version": 315
}

An analytical deletion event does not itself remove older personal data from GraphJSON. Follow Data lifecycle and privacy for the required deletion process.

Preserve order without assuming global order#

Most sources guarantee order only within a partition, table, entity, or log. Carry a monotonic source_version when later state must supersede earlier state.

For current state, use both event time and version:

SELECT
  JSONExtractString(json, 'account_id') AS account_id,
  argMax(
    JSONExtractString(json, 'status'),
    tuple(
      timestamp,
      JSONExtractInt(json, 'source_version')
    )
  ) AS current_status
FROM account_state_events
WHERE JSONExtractString(json, 'event') = 'account_status_changed'
GROUP BY account_id

Do not use GraphJSON insertion order as business order. Network retries and parallel workers can change arrival order.

Make delivery replay-safe#

Derive a stable event ID from immutable source identity:

event_id = source + partition + source_position + mapping_version

Store the mapping version when the same source record might be replayed through a corrected transform. Do not generate a fresh random ID on every retry.

The GraphJSON ingestion API does not provide a delivery idempotency key. At-least-once pipelines can therefore produce repeated rows after an unknown network outcome. Preserve event_id and deduplicate exact metrics in SQL as described in Deduplicate events.

Checkpoint only after a batch has been accepted. Keep:

last acknowledged source position per partition
mapping and schema versions
attempt count and last error
oldest pending record age
permanent-rejection dead letter

Send at most 50 events per /api/bulk-log request, keep one collection per request, and use bounded concurrency. Follow High-volume ingestion and backfills for retry and capacity controls.

Bootstrap without losing concurrent changes#

A continuous sync often needs historical state plus a live tail:

1. record source log position P
2. take a consistent snapshot
3. import the snapshot as bootstrap observations
4. begin consuming changes after P
5. reconcile the overlap
6. declare a completeness watermark

The exact procedure depends on the database and CDC system. Some tools provide a coordinated snapshot; others require an explicit overlap. Never start a multi-hour table export and then consume only changes after the export finishes. Updates that occurred during the export would be lost.

Keep bootstrap and live events distinguishable. Creation, conversion, and velocity metrics should not treat an imported current row as if it was created today.

Handle schema changes at the mapping boundary#

CDC exposes physical schema changes quickly. Protect GraphJSON with:

an allowlist of source tables and columns
type validation before delivery
explicit null behavior
source-to-event contract tests
a quarantine path for unknown required values
metrics for unseen source schema versions

Do not automatically forward a newly added database column. It may contain personal data, secrets, or a high-cardinality value unsuitable for analytics.

Use Executable event contracts to test the destination event independently of the source record.

Reconcile continuously#

For settled partitions, compare:

Check	Detects
Source changes mapped vs events accepted	Missing delivery or unexpected filtering
Distinct stable event IDs	Retry duplicates
Min/max source position	Truncated range
Entity and operation distribution	Mapping or routing mistakes
Key amount sums	Lost or malformed values
Oldest unprocessed source time	Falling behind

Do not compare a live source head with an analytical window that is still receiving late changes. Publish a data_complete_through watermark and use the same settled boundary on both sides.

The complete operating pattern is in Reconcile analytics with source systems.

Recovery procedure#

When delivery falls behind:

stop advancing the affected checkpoint
preserve the source log or snapshot needed for recovery
identify the last independently verified position
fix and version the mapping
replay a small canary interval
compare IDs, counts, types, and key sums
resume with bounded concurrency
reconcile through a settled watermark

Do not skip a poison record silently to make lag return to zero. Quarantine it with a safe reason and keep the gap visible until an owner resolves it.

Launch checklist#

The source database remains authoritative.
The selected cursor survives ties, restarts, and deletions.
Business events are mapped from approved changes.
Occurrence time and observation time are distinct.
Every event has a stable replay identity.
Checkpoints advance only after accepted delivery.
Unknown outcomes are safe to retry and deduplicate.
Bootstrap records cannot inflate creation metrics.
New source columns are denied by default.
Live lag, rejection, and reconciliation signals have owners.
Source-log retention exceeds the recovery objective.