@chkit/plugin-ingest

@chkit/plugin-ingest implements chkit Sync (beta), the optional TypeScript layer for syncing business application data into ClickHouse. Define and migrate destination tables with chkit, then add readers to load them. Use an external scheduler for repeated runs.

The executor records progress after writes succeed. Run-scoped journal identities keep overlapping executions independent. A retry resumes from a validated checkpoint and may reread rows.

• simpleLoader() is the cheap, direct, at-least-once default; native retry tokens have finite retention.
• stagedLoader() opts into immutable-key deduplication across overlaps/replays: first/last whole-row winners within a batch, existing destination keys always win. It adds DDL, scratch storage and potentially full destination-key scans—not upserts or snapshot transactions.
• Staging requires plain MergeTree/Atomic on one pinned server and cooperating writers sharing an executor/package/process; no Shared/replicated targets or cross-process uniqueness. Read the setup and safety contract before use, especially ambiguous publication and restart recovery.

paginate() yields complete { items, next, metadata? } pages, including empty terminal pages. Read rows from page.items; use next for request continuation and explicitly map checkpoint metadata into a chunk's state with cursorState. The executor commits that state after the rows are saved. This replaces the previous item-array return type; existing readers must use page.items.

import { defineConfig } from '@chkit/core'
import { ingest } from '@chkit/plugin-ingest'

export default defineConfig({
  entry: './src/chkit.ts',
  plugins: [ingest()],
  clickhouse: { url: process.env.CLICKHOUSE_URL ?? '' },
})

chkit ingest run --tag schedule:1h

The append-only ClickHouse journal records each run's starting checkpoint and local event sequence. Two processes may start from the same checkpoint without reusing event identities. Complete snapshot validation excludes malformed run tails while retaining acknowledged progress. A stale but valid read can cause replay; destination writes remain at-least-once.

Diagnose the journal or preview evidence-preserving recovery:

chkit ingest doctor --tag stream:helpdesk.tickets
chkit ingest repair --tag stream:helpdesk.tickets

Overlapping valid runs need no repair. For damaged evidence, stop all ingestion writers and review the plan. chkit ingest repair --tag stream:helpdesk.tickets --apply <fingerprint> materializes a verified archive and replacement journal while preserving the original. Keep writers stopped, set ingest({ journalTable: '<replacementTable>' }) to the returned table, then restart them. The command leaves activation to this explicit configuration change. Other streams' evidence is copied unchanged.

Documentation: chkit Sync (beta)

For raw storage, typed rows, and where to map fields, see Destinations and transformations.

Install the source-authoring skill for a coding agent:

npx skills add obsessiondb/chkit --skill chkit-ingestion