pipekit/README.md

# Pipekit

A database sync tool with a web UI. A **module** defines one job — a source
query against one database, merged into a destination table on another. Pipekit
resolves watermarks, runs the extract, stages the rows, merges them, and records
what happened. Modules can be grouped, ordered, and scheduled with cron.

It is an orchestration layer over [`jrunner`](#requirements), an external Java
CLI that does all the actual JDBC work. Pipekit never talks to a remote database
directly — it shells out. What it adds is the state, history, and UI that a
directory of shell scripts can't provide: run logs with timings and row counts,
live output while a job runs, a wizard for authoring new modules, and a
scheduler.

---

## Install

### Requirements

- **Python 3.10+**
- **jrunner** on `PATH`, with `JAVA_HOME` set — the Java CLI that performs every
  JDBC read and write. Install `/opt/jrunner` first; `deploy.sh` refuses to run
  without it and injects the detected `JAVA_HOME` into the systemd unit.
- **sudo** — the deploy creates a system user, writes to `/etc/pipekit`, and
  installs a systemd unit.

### Deploy

```bash
./deploy.sh
```

Idempotent — re-run it after any code update. It creates the `pipekit` system
user and group, adds you to that group, builds the venv at `.venv`, installs the
`/usr/local/bin/pipekit` launcher, creates `/etc/pipekit/secrets.env` (mode
0640), initialises the SQLite schema, registers a JDBC driver row for every jar
shipped with jrunner, and installs the systemd unit. It does **not** start the
service.

Then add your connection passwords and start it:

```bash
sudo pipekit secrets set MYDB_PW          # prompts if the value is omitted
sudo systemctl start pipekit
```

The web UI is on `api_port` from `config.yaml` — **8200** by default.

### Service management

```bash
sudo systemctl status pipekit
sudo systemctl restart pipekit      # after any code change
sudo journalctl -u pipekit -f       # follow logs
```

### Running outside the service

```bash
pipekit serve                                        # host/port from config.yaml
pipekit serve --host 0.0.0.0 --port 8080 --reload    # dev, auto-reload
```

### CLI

```bash
pipekit init                    # create/upgrade the SQLite schema
pipekit doctor                  # health check: config, jrunner, DB
pipekit run <module_name>       # run one module synchronously
pipekit set-password <user>     # HTTP Basic Auth credentials for /api/*
pipekit secrets set KEY [VALUE] # add/update a secret
pipekit secrets list            # show stored keys, never values
pipekit drivers list            # registered driver kinds
```

Configuration also round-trips to plain text, so modules can be diffed and
version-controlled rather than living only in the database:

```bash
pipekit export                  # dump drivers/connections/modules/groups to <repo>/config
pipekit apply --dry-run         # print the change plan without touching the DB
pipekit apply                   # rehydrate from those files, matching by name
pipekit apply --prune           # also delete rows absent from the files
```

### Configuration

`config.yaml` in the repo root holds the database path, jrunner path, driver
directory, API host/port, and an optional SMTP block. Two environment variables
override the defaults: `PIPEKIT_CONFIG` for the config file and
`PIPEKIT_SECRETS` for the secrets file.

---

## About the project

### Tech stack

- **Python 3.10+** — no ORM, no migrations framework, no task queue
- **FastAPI + Uvicorn** for both the JSON API and the HTML pages
- **Jinja2** templates; **HTMX** (from CDN) plus vanilla JS for interactivity.
  There is no build step and no frontend toolchain. Alpine.js is named in some
  older docs but is not loaded.
- **SQLite** (stdlib `sqlite3`) — a single `pipekit.db` file holds all state
- **PyYAML** for config, **croniter** for schedule evaluation
- **python-multipart** — required for HTML form POSTs; it is not a transitive
  dependency of the FastAPI base wheel, and form posts return 500 without it
- **jrunner** — a separate Java tool, the only thing that speaks JDBC

Deployed as the `pipekit` system user, venv at `/opt/pipekit/.venv`, secrets at
`/etc/pipekit/secrets.env`.

### Architecture

Five layers, bottom to top:

1. **SQLite** — one file, all state
2. **`repo.py`** — every CRUD operation for every table; the only module that
   touches the database
3. **`engine/`** — runs a module: acquires the lock, resolves watermarks, calls
   jrunner, builds and runs the merge, fires hooks, writes the run log, releases
   the lock
4. **`api/`** — JSON REST under `/api/*`, HTTP Basic Auth (except `/health`)
5. **`web/`** — HTML pages at `/`, currently unauthenticated

### Data model

- **driver** — a registered JDBC driver (kind, jar, class, URL template)
- **connection** — a named database (JDBC URL, username, password)
- **module** — one sync job: source and dest connection, source query, merge
  strategy and key, column map, enabled flag, running lock
- **watermark** — a named placeholder with its own resolver SQL. The first cell
  of the first row is substituted for `{watermark_name}` in the source query.
- **hook** — post-merge SQL, ordered, fired on success / failure / always
- **run_log** — immutable history: resolved SQL, merge SQL, watermark values,
  stdout/stderr, timings, and the live log streamed during the run
- **grp** / **group_member** / **group_run** — named, ordered sets of modules
  and their executions
- **schedule** — a cron expression bound to a group, evaluated in **local server
  time**

### How a run works

```
run_module(module_id)
  → atomic UPDATE module SET running=1 WHERE running=0   # fails if already locked
  → resolve each watermark via jrunner                   # always live, even on dry runs
  → substitute {name} tokens into the source query
  → build the merge SQL
  → if dry_run: write the run_log and stop — no data moves
  → DROP + CREATE pipekit_staging.{module_name}          # self-heals schema drift
  → jrunner migrate: source → staging                    # stdout streamed to live_log
  → run the merge, then the hooks in order
  → write run_log; UPDATE module SET running=0 (in finally)
```

A group run creates a `group_run` row and calls `run_module` for each enabled
member in order. It continues past individual failures — every member runs — and
the group's final status reflects the worst outcome.

Run statuses: `running`, `success`, `dry_run`, `error`, `cancelled`.

### Merge strategies

- **full** — `TRUNCATE` the dest, then insert everything
- **incremental** — delete the rows matching `merge_key`, then insert. The key is
  stored comma-separated and expands to a multi-column predicate.
- **append** — insert only

The staging table is recreated on every run and exists only for the duration of
that run.

### Credentials

Connection passwords are stored as `$VAR_NAME` references, never as values. They
resolve at run time from `/etc/pipekit/secrets.env`. Secrets never reach the
database, and never appear on a command line.

### Performance note

When the destination is SQL Server or Postgres, jrunner is invoked with `-b` so
the load uses the native bulk path — `SQLServerBulkCopy` over TDS, or
`COPY … FROM STDIN`. This is automatic, per-destination, with nothing to
configure, and the difference is large: one 1.27M-row SQL Server load went from
~111 minutes to ~4. DB2 destinations keep the ordinary INSERT path.

### Reconciliation

`reconcile.py` compares a module's live source against its synced destination
using column-wise aggregates — counts, sums, min/max, summed lengths. These rely
only on arithmetic and ordering, which every backend computes identically, so no
shared hash function or byte-identical row serialisation is needed.

```bash
PIPEKIT_SECRETS=/etc/pipekit/secrets.env .venv/bin/python reconcile.py <module> [--quick]
```

Three depths: the default compares every column; `--quick` drops min/max and
length sums; `--super-quick` compares only row counts, which takes seconds even
on millions of rows and still catches missing or duplicated rows. Exit status is
non-zero when anything diverges. Note that it compares whole tables against a
live source, so an incremental module will always show drift proportional to the
time since its last run.

---

## Further reading

- **`SPEC.md`** — the authoritative design document, and the rationale behind
  every architectural decision here
- **`CLAUDE.md`** — orientation for AI coding agents working in this repo