✨ Launch the ingestion feature

.changeset/config.json

@@ -12,6 +12,7 @@
       "@chkit/plugin-codegen",
       "@chkit/plugin-pull",
       "@chkit/plugin-backfill",
+      "@chkit/plugin-ingest",
       "@chkit/plugin-obsessiondb"
     ]
   ],

.changeset/ingestion-runtime.md

@@ -0,0 +1,20 @@
+---
+"@chkit/plugin-ingest": patch
+"@chkit/clickhouse": patch
+"@chkit/core": patch
+"chkit": patch
+---
+
+Add `@chkit/plugin-ingest`, the first cut of scheduled pull ingestion into ClickHouse. Streams are ordinary TypeScript: a `read` async generator fetches, maps, and yields destination-shaped rows, and `definePipeline` returns a tagged, non-durable group of streams; only pipelines exported from the project entry participate, with no global registry. `chkit ingest run` executes the selected streams (`--tag` is repeatable with exact AND semantics; an explicit empty selection fails), `chkit ingest list` shows the loaded graph, and `chkit ingest status` prints committed checkpoints.
+
+Progress follows one rule: rows are saved before the bookmark advances. Every batch is written with a stable `insert_deduplication_token`, and only after the ClickHouse acknowledgement does the executor append a `batch_committed` fact to the append-only ingestion journal. Checkpoints are a projection of that journal, so a crashed or lost-acknowledgement run replays from the last durable boundary with the same batch identity instead of skipping rows. Bundled strategies are `timestampWindow({ start, overlapMs })` (with a custom `from` callback alternative), `cursorState` for provider-owned state, and the full-sync fallback; `--backfill <id>` runs an explicit range in an isolated checkpoint namespace.
+
+`rawTable` and `rawRows` land provider objects untouched in a native `JSON` column, so typed shapes are derived inside ClickHouse with ordinary views or materialized views instead of being mapped in pipeline code.
+
+`FetchContext` exposes source request and cancellation capabilities independently of checkpoint types; `ReadContext` extends it.
+
+Source operations run through `context.attempt`, which owns fetch permits, p-retry-shaped retry policy, `Retry-After`, cancellation, and failure classification (`HttpError.fromResponse` is the canonical boundary for fetch-based readers). Pipelines carry separate `maxStreams`, `maxFetches`, and `maxLoads` ceilings, executions have a duration budget, and the executor emits OpenTelemetry spans.
+
+`@chkit/core` gains a singular `entry` config field, mutually exclusive with `schema` globs: the module is imported once, its exported schema definitions are collected, and exported plugin-domain definitions are collected by their plugins. `@chkit/clickhouse` `insert()` accepts per-insert `settings`.
+
+Successful syncs rotate batch identity using the existing journal, while failed runs retain their replay identity. Execution cancellation also bounds journal I/O, stalled writes cannot report success, and loader construction failures release their permits. Ingestion requires a direct ClickHouse connection; incompatible host executors fail before any writes. Project `entry` and `schema` settings replace the inherited source mode when layering configuration.

CLAUDE.md

@@ -21,6 +21,7 @@ This is a monorepo managed with Bun workspaces and Turborepo.
 | `packages/plugin-codegen` | `@chkit/plugin-codegen` | Plugin: TypeScript type + Zod schema generation |
 | `packages/plugin-pull` | `@chkit/plugin-pull` | Plugin: introspect live ClickHouse into schema files |
 | `packages/plugin-backfill` | `@chkit/plugin-backfill` | Plugin: time-windowed data backfill with checkpoints |
+| `packages/plugin-ingest` | `@chkit/plugin-ingest` | Plugin: scheduled pull ingestion with journaled checkpoints |
 | `packages/plugin-obsessiondb` | `@chkit/plugin-obsessiondb` | Plugin: ObsessionDB integration; auto-rewrites `Shared` engines for non-ObsessionDB targets |
 
 ### Documentation

README.md

@@ -114,6 +114,7 @@ See the [configuration docs](https://chkit.obsessiondb.com/configuration/overvie
 | [`@chkit/plugin-pull`](packages/plugin-pull) | Pull live schema into local files |
 | [`@chkit/plugin-codegen`](packages/plugin-codegen) | Codegen plugin for the CLI |
 | [`@chkit/plugin-backfill`](packages/plugin-backfill) | Backfill plugin for data migrations |
+| [`@chkit/plugin-ingest`](packages/plugin-ingest) | Ingestion plugin: scheduled API pulls with journaled checkpoints |
 | [`@chkit/plugin-obsessiondb`](packages/plugin-obsessiondb) | ObsessionDB integration: auto-rewrite `Shared` engines for ClickHouse targets |
 
 ## Python

apps/docs/src/content/docs/configuration/overview.mdx

@@ -5,11 +5,12 @@ description: "clickhouse.config.ts / clickhouse.config.py structure and defaults
 
 import { Tabs, TabItem } from '@astrojs/starlight/components';
 
-`chkit` is configured through `clickhouse.config.ts` (TypeScript) or `clickhouse.config.py` (Python). The option keys and defaults are identical.
+`chkit` is configured through `clickhouse.config.ts` (TypeScript) or `clickhouse.config.py` (Python). Shared configuration options use the same keys and defaults in both languages.
 
 ## Core Fields
 
 - `schema`: glob path to [schema files](/schema/dsl-reference/)
+- `entry`: optional project entry module in place of `schema` globs (TypeScript only)
 - `outDir`: root folder for generated artifacts
 - `migrationsDir`: SQL migration file folder
 - `metaDir`: state folder (`snapshot.json`)
@@ -65,6 +66,18 @@ Migration state (the journal of applied migrations) is not stored in `metaDir`.
   </TabItem>
 </Tabs>
 
+## Project entry (`entry`, TypeScript only)
+
+Instead of `schema` globs you can point chkit at one entry module:
+
+```ts
+export default defineConfig({
+  entry: './src/chkit.ts',
+})
+```
+
+chkit imports the module once. Schema definitions it exports (directly or re-exported from other files) are collected exactly like glob-matched schema files, and exported plugin-domain definitions such as [ingestion pipelines](/plugins/ingest/) are collected by their plugins. `entry` and `schema` are mutually exclusive.
+
 ## Cluster mode (`ON CLUSTER`)
 
 For self-managed multi-node ClickHouse clusters, set `clickhouse.cluster` to the cluster name from your server's `remote_servers` config:

apps/docs/src/content/docs/plugins/ingest.md

@@ -0,0 +1,187 @@
+---
+title: Ingest Plugin
+description: Scheduled pull ingestion from application APIs into ClickHouse with journaled checkpoints.
+sidebar:
+  order: 5
+---
+
+This document covers practical usage of the optional TypeScript `ingest` plugin.
+
+## What it does
+
+- Runs finite, scheduled pulls from application APIs into chkit-managed tables.
+- Keeps progress in an append-only ingestion journal inside the target database. Checkpoints are a projection of that journal.
+- Saves rows before advancing the bookmark. A crash may cause rereading; it never causes unsaved rows to be skipped.
+- Retries source requests with backoff, `Retry-After`, and failure classification.
+- Selects streams by exact tags so any external scheduler (cron, CI, Kubernetes) can drive it.
+
+The plugin never creates or changes destination tables. Your chkit schema stays the only DDL authority.
+
+## Plugin setup
+
+Ingestion uses the singular `entry` config field instead of `schema` globs. The entry module is imported once: exported tables are collected as schema, and exported pipelines form the ingestion graph. Importing a pipeline without exporting it does not activate it; remove its export to deactivate it.
+
+Configure a direct `clickhouse` connection for ingestion. A host-provided executor, including the ObsessionDB workbench executor, cannot currently guarantee JSON row encoding and per-insert deduplication settings.
+
+```ts
+// clickhouse.config.ts
+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 ?? '' },
+})
+```
+
+## Writing a stream
+
+A stream is a destination table plus a `read` generator. Fetching and mapping are ordinary code inside `read`; each yielded chunk carries rows already shaped for the table.
+
+```ts
+// src/chkit.ts
+import { table } from '@chkit/core'
+import { HttpError, definePipeline, defineStream, ingestionColumns, paginate, timestampWindow } from '@chkit/plugin-ingest'
+
+export const tickets = table({
+  database: 'crm',
+  name: 'tickets',
+  columns: [
+    { name: 'id', type: 'String' },
+    { name: 'updated_at', type: "DateTime64(3, 'UTC')" },
+    { name: 'raw', type: 'String' },
+    ...ingestionColumns,
+  ],
+  engine: 'ReplacingMergeTree(updated_at)',
+  primaryKey: ['id'],
+  orderBy: ['id'],
+})
+
+const ticketStream = defineStream({
+  id: 'helpdesk.tickets',
+  destination: tickets,
+  tags: ['schedule:1h'],
+  incremental: timestampWindow({
+    // Re-read one hour of overlap; ReplacingMergeTree reconciles repeats.
+    start: new Date(0),
+    overlapMs: 3_600_000,
+  }),
+  async *read(context) {
+    const pages = paginate({
+      context,
+      label: 'GET /tickets',
+      fetchPage: async (cursor: string | undefined, signal) => {
+        const url = new URL('https://api.example.com/tickets')
+        url.searchParams.set('updated_since', context.selection.from.toISOString())
+        if (cursor) url.searchParams.set('cursor', cursor)
+        const response = await fetch(url, { signal, headers: { Authorization: `Bearer ${process.env.HELPDESK_TOKEN}` } })
+        if (!response.ok) throw await HttpError.fromResponse(response)
+        const body = await response.json()
+        return { items: body.data, next: body.next_cursor ?? undefined }
+      },
+    })
+    for await (const items of pages) {
+      yield { rows: items.map((item) => ({ id: item.id, updated_at: item.updated_at, raw: JSON.stringify(item) })) }
+    }
+  },
+})
+
+export const helpdesk = definePipeline({ id: 'helpdesk', streams: [ticketStream], maxFetches: 4 })
+```
+
+Spread `ingestionColumns` into every destination table. The loader fills `_chkit_batch_id` and `_chkit_run_id`; `_chkit_ingested_at` is set by ClickHouse at the physical insert.
+
+Batch identity decides whether a retry is deduplicated. By default it includes a content hash of the rows, which prefers a possible duplicate over suppressing rows that changed between attempts; a field like `synced_at: new Date()` therefore defeats retry deduplication. When a chunk covers a stable source interval, declare it with `id` (for example `yield { rows, id: \`page:${cursor}\` }`): the chunk then becomes its own write unit and its identity ignores row content.
+
+## Landing raw objects
+
+Mapping fields in the reader is optional, and usually the wrong place for it. `rawTable` defines a landing table that stores each provider object untouched in a native `JSON` column next to a stable `id`; `rawRows` shapes a page for it. Typed tables are then ordinary chkit views (or materialized views) over the raw layer:
+
+```ts
+import { view } from '@chkit/core'
+import { defineStream, rawRows, rawTable } from '@chkit/plugin-ingest'
+
+export const rawTickets = rawTable({ database: 'crm', name: 'tickets_raw' })
+
+export const tickets = view({
+  database: 'crm',
+  name: 'tickets',
+  as: `SELECT
+  id AS ticket_id,
+  raw.subject::String AS subject,
+  raw.requester.email::String AS requester_email,
+  arrayMap(t -> t.name::String, raw.tags[]) AS tags,
+  parseDateTime64BestEffortOrNull(raw.updated_at::String, 3, 'UTC') AS updated_at
+FROM crm.tickets_raw FINAL`,
+})
+
+const ticketStream = defineStream({
+  id: 'helpdesk.tickets',
+  destination: rawTickets,
+  async *read(context) {
+    for await (const page of listTickets(context)) yield { rows: rawRows(page, (ticket) => ticket.id) }
+  },
+})
+```
+
+The raw table is a `ReplacingMergeTree`, so overlapping windows and replays collapse to the latest version of each `id`. Because the transform lives in ClickHouse, changing it never requires re-fetching the source: a view picks the change up immediately, and a materialized view can be rebuilt from the raw table. A `_raw` suffix next to the typed view of the same name keeps the pair easy to find.
+
+## Progress and checkpoints
+
+| Strategy | Use when | Bookmark advances |
+|---|---|---|
+| none (full sync) | The source is small or has no change filter | Never; every run reads everything |
+| `timestampWindow({ start, overlapMs })` | The API filters by an updated-since timestamp | To the run cutoff, after the whole window loaded |
+| `cursorState({ id, version, parse })` | The provider owns the state: compound cursor, change token, page position | Whenever a yielded chunk carries `state` and its rows have been saved |
+
+`timestampWindow` starts its first sync at `start`. Later runs begin at the committed watermark minus `overlapMs` (zero by default). Explicit backfill bounds take precedence. For custom lower bounds, use `timestampWindow({ from: ({ watermark, cutoff }) => ... })` instead. Both forms retain the same checkpoint format and only advance after the entire window is saved.
+
+Provider clients can accept the exported `FetchContext` type, containing `attempt` and `signal`. `ReadContext` extends it with the stream selection and checkpoint, so readers can pass their context directly without coupling clients to checkpoint generics.
+
+With `cursorState`, `state` on a chunk must be the complete state that is safe to resume from once every row up to that chunk is saved. Omit it when you cannot make that claim; the run then restarts from the previous checkpoint after a failure.
+
+A checkpoint records its strategy id and version. Changing either makes the next run fail rather than reinterpret old state.
+
+## Retry policy
+
+Pipeline `retry` settings are defaults; a stream can override individual settings. Source attempts and reader restarts use [`p-retry`](https://github.com/sindresorhus/p-retry) for exponential backoff, jitter, retry counts, and `maxRetryTime`. The supported options are `retries`, `factor`, `minTimeout`, `maxTimeout`, `randomize`, `maxRetryTime`, `shouldRetry`, and `shouldConsumeRetry`.
+
+The policy callbacks receive p-retry's `attemptNumber`, `retriesLeft`, and `retriesConsumed`, plus a `FetchFailure` preserving the original `cause` and normalized `classification`. Returning `false` from `shouldConsumeRetry` follows p-retry's behavior: it skips consuming a retry and skips its backoff. A provider's `Retry-After` still applies, including for those unconsumed retries. Retry waits release fetch and load capacity for other streams.
+
+Once a source attempt exhausts its policy, it fails the stream; reader recovery does not multiply that retry budget. Opaque reader failures restart from the latest committed checkpoint.
+
+## Commands
+
+```sh
+chkit ingest list                       # streams in the loaded graph
+chkit ingest run                        # run every stream
+chkit ingest run --tag schedule:1h      # exact tag match; repeat --tag for AND
+chkit ingest run --tag stream:helpdesk.tickets
+chkit ingest status                     # committed checkpoint per stream
+chkit ingest run --backfill jan --from 2026-01-01 --to 2026-02-01
+```
+
+Every stream also carries the derived tags `pipeline:<id>` and `stream:<id>`. `schedule:<cadence>` is a convention only: chkit never interprets it. A `--tag` filter that matches nothing fails before any work runs.
+
+A backfill uses its own checkpoint namespace, so it never moves the scheduled bookmark. Rerunning the same `--backfill` id resumes it.
+
+`chkit check` verifies that every stream destination carries the ingestion metadata columns.
+
+## Delivery guarantee
+
+Ingestion is at-least-once. Each batch is inserted with a stable `insert_deduplication_token`, so a retry after a lost acknowledgement is suppressed while the table's deduplication window covers it. Pick a destination engine that reconciles repeats for your data, for example `ReplacingMergeTree` keyed by the provider id.
+
+Successful syncs start a new batch identity cycle, recorded by the existing journal. Failed or interrupted syncs retain their cycle for replay. This also applies to full syncs, which have no incremental bookmark.
+
+The duration budget bounds journal operations as well as readers. Shutdown gives unfinished readers or writes up to five seconds to settle; each terminal journal append has a separate five-second limit. Interrupted writes never count as successful ingestion.
+
+Run at most one ingestion process per project and target at a time. Use your scheduler's concurrency control (for example a GitHub Actions concurrency group) to enforce it.
+
+## Options
+
+| Option | Default | Description |
+|---|---|---|
+| `journalTable` | `_chkit_ingestion_journal` | Journal table name in the configured database |
+| `maxDurationSeconds` | `3600` | Execution budget. Exhausting it ends the run as incomplete and keeps committed progress |
+| `prefetchBatches` | `1` | Mapped batches buffered between fetching and loading |

apps/docs/src/content/docs/plugins/overview.mdx

@@ -62,3 +62,4 @@ If you deploy to [ObsessionDB](https://obsessiondb.com), start at the dedicated
 - [`@chkit/plugin-codegen`](/plugins/codegen/) — TypeScript row types and optional Zod schemas (Python: Pydantic models), generated from your schema files.
 - [`@chkit/plugin-pull`](/plugins/pull/) — introspect a live ClickHouse database into local schema files. Useful for adopting chkit on an existing database. Built into the Python CLI as `chkit pull`.
 - [`@chkit/plugin-backfill`](/plugins/backfill/) — time-windowed data backfill with checkpoints, for materialized views and historical data loads.
+- [`@chkit/plugin-ingest`](/plugins/ingest/) — scheduled pull ingestion from application APIs with journaled checkpoints (TypeScript only).