data quality and reconciliation

why

external and historical data arrives wrong, duplicated, and incomplete; a sync that guesses identities converts those flaws into silent corruption that compounds on every rerun. explicit identity plus measured completeness makes disagreement visible and repairable — the only footing a migration or backfill can safely move reads onto:

duplicates cannot exist
invalid rows are quarantined with a cause
every run records expected vs accepted, so silent partial success is impossible

rule

ingested and historical data is untrusted until its identity, schema version, ownership, and invariants are validated. reconciliation compares named authorities and records the discrepancy; it never overwrites one side by guessing.

ingestion contract

question	required answer
what identifies the source fact?	tenant/provider-connection-qualified external id and source version/cursor.
which system is authoritative?	one named owner for each field and transition.
how are duplicates handled?	unique external identity plus idempotent write.
how are invalid rows handled?	typed rejection/quarantine with causal diagnostic, never partial success.
how is completeness measured?	source window, expected count/checksum, and recorded watermark.

retain an opaque source payload only for audit/replay
extract every product-behavior-driving field into typed columns, relations, or subtype tables

repair without corrupting history

backfill in bounded, restartable batches
record input range, schema/version, attempt, result count, rejected count, reconciliation result
move a read only after target count, identities, and sampled invariants agree with the declared source
correct a bad fact through its normal transition or a named repair action; never hand-edit applied history

good and bad

bad

// sync provider workflows by matching display name
async function syncProviderWorkflows(connectionId: string) {
  const sourceWorkflows = await provider.listWorkflows();

  for (const source of sourceWorkflows) {
    const existing = await db.select().from(workflow).where(eq(workflow.display_name, source.name));

    if (existing.length > 0) {
      // update existing row
      await db
        .update(workflow)
        .set({
          display_name: source.name,
          status: source.status,
        })
        .where(eq(workflow.id, existing[0].id));
    } else {
      // create new row
      await db.insert(workflow).values({
        id: createWorkflowId(),
        display_name: source.name,
        status: source.status,
      });
    }
  }
}

guessing by display name silently corrupts data:

provider renames "User Onboarding" to "Onboarding" — the upsert creates a duplicate row (display name no longer matches)
a workflow with invalid JSON in an optional field is silently skipped; the sync "succeeds" but the dashboard shows stale data with no evidence of incompleteness
the next run sees the same problem rows and creates more duplicates
nobody knows the sync was incomplete until users report missing workflows

good

// schema: establish external identity as the source of truth
create table workflow (
  id text primary key, -- internal record id
  workspace_id text not null references workspace(id),
  provider_connection_id text not null references provider_connection(id),
  external_id text not null, -- provider's workflow id (stable across renames)
  display_name text not null,
  status text not null,
  synced_at text not null,
  unique (workspace_id, provider_connection_id, external_id) -- source identity
);

// sync service: measure and report
async function syncProviderWorkflows(connectionId: string) {
  const watermark = { source: 'provider', cursor: new Date(), version: '2' };
  const expected: number[] = [];
  const accepted: string[] = [];
  const quarantined: { externalId: string; error: string }[] = [];

  try {
    const sourceWorkflows = await provider.listWorkflows();
    expected.push(sourceWorkflows.length);

    for (const source of sourceWorkflows) {
      try {
        // parse and validate source fact
        const parsed = WorkflowSourceSchema.parse(source);

        // upsert by external identity (not display name)
        await db.insert(workflow).values({
          id: createWorkflowId(),
          workspace_id: connectionId,
          provider_connection_id: connectionId,
          external_id: parsed.id, // stable provider id
          display_name: parsed.name,
          status: parsed.status,
          synced_at: new Date().toISOString(),
        }).onConflictDoUpdate({
          target: [workflow.workspace_id, workflow.provider_connection_id, workflow.external_id],
          set: { display_name: parsed.name, status: parsed.status, synced_at: new Date().toISOString() },
        });

        accepted.push(parsed.id);
      } catch (error) {
        quarantined.push({ externalId: source.id, error: error.message });
      }
    }
  } catch (error) {
    throw new SyncFailedError(`sync failed: ${error.message}`, { watermark });
  }

  // record the reconciliation result
  await db.insert(syncRecord).values({
    connection_id: connectionId,
    source_window: watermark,
    expected_count: expected[0],
    accepted_count: accepted.length,
    quarantined_count: quarantined.length,
    quarantined: JSON.stringify(quarantined),
    status: quarantined.length === 0 ? 'complete' : 'incomplete',
  });

  return {
    expected: expected[0],
    accepted: accepted.length,
    quarantined: quarantined.length,
    diagnostics: quarantined.map(q => `${q.externalId}: ${q.error}`),
  };
}

idempotent reruns: unique constraint on external id prevents duplicates across retries
measured: every sync records expected vs accepted vs quarantined
repairable: quarantined rows are listed with their parse error for investigation and reprocessing
renames no longer duplicate: identity is external id, not display name
silent incompleteness impossible: every sync records complete vs incomplete