# flusso

> [!IMPORTANT]
> ## 🤖 Generative AI disclosure
>
> **Generative AI was used in this project to produce boilerplate and
> documentation.** Every single line of code has been manually reviewed and
> revised by a human software developer who can be blamed accordingly.

**Keep OpenSearch in sync with Postgres, driven by declarative config.**

You write a bit of YAML describing what a search document should look like.
flusso builds the index, seeds it from your existing rows, then tails Postgres'
logical replication stream so the index stays current — no cron job, no nightly
reindex, no hand-rolled `for row in rows: es.index(...)` script you'll regret at
2am.

In short: you describe the *what*, flusso handles the *keep-it-in-sync*.

## Contents

- [Quickstart](#quickstart) — running in about five commands
- [What it does](#what-it-does) — the two files you write
- [How the pipeline works](#how-the-pipeline-works) — the bit that does the work
- [The CLI](#the-cli) — `build`, `check`, `run`
- [`just` commands](#just-commands) — the shortcuts you'll actually use
- [Requirements](#requirements) — what Postgres and OpenSearch need first
- [Deploying it](#deploying-it) — Docker image + Helm chart
- [Docs](#docs) — everything else, linked
- [Project layout](#project-layout) — where the code lives
- [Testing & development](#testing--development)

## Quickstart

The [`dev/`](dev/) directory is a complete, runnable example — a docker-compose
stack (Postgres wired for logical replication, OpenSearch, Dashboards,
Prometheus, Grafana), seeded data, and a matching config. With
[`just`](https://just.systems) installed (`cargo install just --locked`):

```sh
just up        # bring the whole stack up and wait for it to be healthy
just check     # validate the config + schemas against the database
just run       # backfill OpenSearch, then follow live changes (serves /status + /metrics)
```

Then, in another terminal, poke it and watch changes stream through:

```sh
just psql                                            # make some changes
curl -s localhost:9200/users/_search?pretty          # see them land in OpenSearch
just status                                          # live pipeline status
```

That's it. Run `just` on its own to see every recipe. The full walk-through —
resetting state, inspecting the slot, OpenSearch Dashboards on `:5601` — lives in
[`dev/README.md`](dev/README.md).

> No `just`? Every recipe is just a thin wrapper; the raw `cargo run -- …` and
> `docker compose …` commands are right there in the [`justfile`](justfile).

## What it does

A deployment is two kinds of files. That's the whole mental model.

**`flusso.toml`** — one per deployment. Where the data comes from, where it goes,
and which indexes to build.

```toml
[source]
type = "postgres"
connection_url = { env = "DATABASE_URL" }   # secrets stay as env refs, never baked in

[sinks.primary]
type = "opensearch"
url = "https://localhost:9200"
password = { env = "OS_PASSWORD" }

[sinks.audit]          # sinks fan out — write the same docs to more than one place
type = "stdout"
pretty = true

[[index]]
name = "users"
schema = "users.schema.yml"
```

**`*.schema.yml`** — one per index. What a single search document looks like:
its table, its fields, and the related tables that fold in.

```yaml
table: users
primary_key: id

soft_delete:
  column: deleted          # users.deleted = true → tombstone instead of upsert

fields:
  - keyword: email
    transforms: [lowercase, trim]

  # Fold each user's recent orders in as a nested array.
  - has_many: orders
    table: orders
    foreign_key: user_id
    primary_key: id
    order_by: [{ column: created_at, direction: desc }]
    limit: 5
    fields:
      - double: total
      - keyword: status

  # ...or just count them. A count is always a long.
  - count: orderCount
    table: orders
    foreign_key: user_id
```

Every field declares its **type** from a fixed set ([`SCHEMA.md`](SCHEMA.md) lists
them all) that bridges a Postgres column and an OpenSearch mapping. So a schema is
self-describing: flusso derives the full index mapping — and validates your config
— without ever touching a database.

The neat part: change a user *or one of their orders* and flusso rebuilds the
whole `users` document and re-emits it. It figures out which documents a changed
row affects, reassembles each, and writes it by a deterministic id. You don't tell
it what to update; it works it out.

## How the pipeline works

The engine wires pluggable edges together and runs the loop:

```text
ChangeCapture ─▶ queue ─▶ resolve ─▶ build ─▶ Sink ─▶ flush ─▶ ack
```

A **capture** task drains the source's change stream into a bounded queue (full
queue → capture blocks; back-pressure for free). A **worker** pulls changes,
works out which document ids they touch, builds each document, and writes it.
Writes are **batched** — N changes, or whatever shows up in a short window, flush
together as one bulk round-trip.

Delivery is **at-least-once** — not exactly-once, because exactly-once is mostly a
story distributed systems tell at conferences. A change's ack is confirmed only
*after* the flush that made its document durable, so the replication slot advances
exactly when the data has landed. Crash before the flush? The batch is redelivered
on restart and re-applied idempotently — same id, same result, no duplicates.

Before going live, the engine runs an optional **backfill**: it asks each sink
whether an index is already seeded and snapshots the tables for the ones that
aren't. Whether a backfill is needed is the *destination's* call, not the source's.

Queue, source, sink, and document builder are all trait objects — so WAL-vs-polling,
stdout-vs-OpenSearch, in-process-channel-vs-durable-broker are all swappable
without touching the engine loop. (Today the menu is short: Postgres in,
OpenSearch out. The seams are there for when it isn't.)

## The CLI

Three subcommands, and **every flag also reads a `FLUSSO_*` env var** (the flag
wins when both are set) — handy for containers.

- **`flusso build`** — compile a config and its schemas into one portable
  `flusso.lock`. No database, no secrets baked in (`{ env = "VAR" }` refs are
  carried through and resolved wherever it runs). Ship one file instead of a tree
  of YAML.
- **`flusso run`** — stream changes through the engine. Like `cargo run`, it
  compiles first: when a `flusso.toml` is present (the default, or `--config`) it
  recompiles and **rewrites `flusso.lock`**, then runs — so the committed lock
  stays current for free. With no config it loads the existing `flusso.lock`, and
  `--locked` runs the lock as-is without recompiling. Credentials are resolved
  here, in the running environment.
- **`flusso check`** — validate the config and print the fully-typed mapping, with
  no database. Drop `--offline` and it also confirms the declared types match the
  live database and grumbles about any that don't.

```sh
flusso --help
flusso build  --config flusso.toml -o flusso.lock   # build the portable artifact
flusso check  --config flusso.toml                  # validate (+ check vs database)
flusso check  --config flusso.toml --offline        # validate without a database
flusso run                                          # run the compiled flusso.lock
flusso run    --config flusso.toml                  # compile from source and run
flusso run    --skip-backfill                       # resume live capture only
```

Logging honors `RUST_LOG` (default `info`); `FLUSSO_LOG_FORMAT=json` for structured
logs. Set the standard `OTEL_EXPORTER_OTLP_ENDPOINT` and traces export there too.
Every environment variable flusso reads — secrets, the `FLUSSO_*` flags, telemetry
— is collected in one place: [`CONFIG.md`](CONFIG.md).

## `just` commands

Common workflows are wrapped in the [`justfile`](justfile). Run `just` for the
full menu; the greatest hits:

| Recipe | Does |
| --- | --- |
| `just up` / `just down` / `just reset` | Start / stop / wipe-and-restart the dev stack |
| `just check` / `just check-offline` | Validate config + schemas (with / without a DB) |
| `just run` / `just run-live` | Backfill + follow / resume live only |
| `just build-lock` | Compile a portable `flusso.lock` |
| `just demo` | Run flusso *inside* the cluster — no host toolchain |
| `just status` / `just metrics` / `just eta` | Live status / raw metrics / backlog drain ETA |
| `just psql` / `just grafana` | psql shell / open the Grafana dashboard |
| `just test` / `just test-all` / `just doc` | Fast tests / + Postgres e2e / doctests |
| `just lint` / `just fmt` / `just ci` | Lint / format / the full local CI gate |

## Requirements

flusso doesn't own Postgres or OpenSearch — it's a guest in both. A few things
have to be true *before* it can run. The [`dev/`](dev/) stack sets all of this up
for you; here's what you'd replicate against your own infrastructure. Full
per-source/per-sink options are in [`SOURCES_AND_SINKS.md`](SOURCES_AND_SINKS.md).

**Postgres (the source):**

- **PG 14+** with **`wal_level = logical`** (a restart-required setting), and
  `max_wal_senders` / `max_replication_slots` high enough for flusso plus any
  other consumers.
- **A publication** covering *every* table any index reads — root tables *and*
  every table a join or aggregate pulls from. flusso derives that set from your
  schema and **manages the publication for you** when the source role is
  privileged enough: it creates it on first connect and extends it as you add
  tables, exactly as it does the slot. The catch is privilege — creating or
  extending a publication needs table ownership plus `CREATE` on the database (or
  superuser), a stronger grant than the read-only role below. If the role can't,
  flusso doesn't fail: it logs the exact `CREATE PUBLICATION` / `ALTER PUBLICATION
  … ADD TABLE` to run, and `flusso check` prints the same. Set `[source]
  manage_publication = false` (or `--no` via `FLUSSO_MANAGE_PUBLICATION=false`) to
  turn management off and manage the publication yourself.
- **A replication slot** — this one flusso always creates on first connect (it
  needs only the `REPLICATION` attribute). Heads-up: Postgres hoards WAL until
  flusso confirms it, so a flusso that's down for a long time means WAL piling up
  on the server. Drop the slot when you retire a deployment, unless you're a fan
  of disk-full pages.
- **Row identity on every replicated table** — a primary key (usual case) or an
  explicit `REPLICA IDENTITY`. Keyless tables can't be addressed, so flusso skips
  them in backfill and errors on a live change it can't key.
- **A role with `REPLICATION` + `SELECT`** on the published tables, via the usual
  `postgres://user:pass@host:port/db` URL. That's enough to stream and to create
  the slot; for flusso to also *manage the publication* (above) the role must own
  those tables and hold `CREATE` on the database — otherwise flusso just prints
  the SQL for you to run with a privileged role.

**OpenSearch (the sink):**

- **OpenSearch 2.x** (also speaks Elasticsearch 7.x on the query side via
  [`flusso-query`](CLIENT.md)).
- **A reachable HTTP(S) endpoint** as the sink `url`. Optional HTTP basic auth
  (`username` / `password`); `tls_verify` defaults to `true` — flip it off only
  for self-signed dev clusters.
- **A user that can manage and write flusso's indexes** plus the small hidden
  `flusso_meta` index where it records "this one's seeded". flusso owns the index
  lifecycle: it derives a strict typed mapping per schema and names each index from
  a hash of that schema (`users_<hash>`), so a structural change rolls onto a fresh
  index and re-seeds instead of fighting a mismatched one. The plain logical name
  (`users`) is kept as an alias on the current index, so you can always query it
  without knowing the hash.

## Deploying it

- **Container image** — the [`Dockerfile`](Dockerfile) builds a registry-ready,
  config-less image (you mount a config or bake your own `flusso.lock`). It also
  has a `demo` target with the dev config baked in, which is what `just demo`
  runs. [`DEPLOY.md`](DEPLOY.md) has the recipes for shipping the smallest
  possible image — bake your own lock, or compile one in-Docker even when your
  schemas are scattered across a monorepo.
- **Kubernetes** — the [Helm chart](deploy/helm/flusso/) deploys flusso as a
  single instance (it consumes one replication slot, so it's *firmly* a party of
  one) with config via ConfigMap, secrets via env, a Service, and an optional
  Prometheus `ServiceMonitor`. See [its README](deploy/helm/flusso/README.md).

## Docs

| Doc | What's in it |
| --- | --- |
| [`SCHEMA.md`](SCHEMA.md) | Every config + schema key: field types, joins, aggregates, filters, validation rules |
| [`SOURCES_AND_SINKS.md`](SOURCES_AND_SINKS.md) | Every source and sink option — batch sizes, retries, analysis modes |
| [`CONFIG.md`](CONFIG.md) | Every environment variable in one place — secrets, `FLUSSO_*` flags, logging & telemetry |
| [`CLIENT.md`](CLIENT.md) | `flusso-query`, the typed query-side client and its `#[derive(FlussoDocument)]` |
| [`DEPLOY.md`](DEPLOY.md) | Docker recipes — smallest image, baking/compiling a `flusso.lock`, scoped `.dockerignore` |
| [`dev/README.md`](dev/README.md) | The dev stack walk-through |
| [`deploy/helm/flusso/README.md`](deploy/helm/flusso/README.md) | The Helm chart |
| [`CLAUDE.md`](CLAUDE.md) | Architecture + contributor notes (also where the AI takes its instructions) |

## Project layout

Crates live under `libs/` and `apps/`. The **numeric prefix is the dependency
layer** — a crate only depends on lower-numbered ones.

| Crate | Path | Role |
| --- | --- | --- |
| `schema-core` | `libs/0-core` | The validated domain model every other crate produces and consumes — the sole layer-0 crate. |
| `schema-config-toml` | `libs/2-schema/1-config-toml` | Parses `flusso.toml` → neutral entities. |
| `schema-index-yaml` | `libs/2-schema/1-index-yaml` | Parses `*.schema.yml` → core types. |
| `schema` | `libs/2-schema` | Config entry point. `load()` reads a config + its schemas into one validated `Config`. |
| `queue-core` | `libs/1-queue/0-core` | The work-queue abstraction, generic over the payload. |
| `queue-channel` | `libs/1-queue/1-channel` | In-process queue over a bounded `tokio` mpsc channel. |
| `sources-core` | `libs/1-sources/0-core` | Source abstractions: `cdc` (*what changed?*) and `document` (*what to build?*). |
| `sources-postgres` | `libs/1-sources/1-postgres` | Postgres source: WAL capture, backfill, document building. |
| `sinks-core` | `libs/1-sinks/0-core` | The `Sink` trait, JSON rendering, and a fan-out sink. |
| `sinks-stdout` | `libs/1-sinks/1-stdout` | Writes each operation to stdout (NDJSON or pretty). |
| `sinks-opensearch` | `libs/1-sinks/2-opensearch` | OpenSearch sink: bulk API, typed mappings, hashed index names + latest-alias, seeding markers. |
| `engine` | `libs/2-engine` | The sync engine — capture → queue → build → sink, batched, at-least-once. |
| `daemon` | `libs/3-daemon` | Wires a `Config` into a running pipeline and exposes live `Status`. |
| `flusso-cli` | `apps/cli` | The `flusso` binary: transport, telemetry, signals. |

Each config-format crate works in two stages — **parse** (serde → permissive
entity types, unknown fields rejected) then **convert** (lift into `schema-core`,
apply the rules the format can't express). Secrets are *not* resolved here, so a
compiled config never carries a secret it wasn't literally given.

## Testing & development

Tests run with [`cargo-nextest`](https://nexte.st)
(`cargo install cargo-nextest --locked`):

```sh
just test        # fast: unit + parse/convert, no external deps
just test-all    # + Postgres e2e (needs Docker; spins up containers via testcontainers)
just doc         # doctests — nextest doesn't run these
just ci          # the full local gate: lint → e2e → doctests
```

The e2e tests are `#[ignore]`d by default and legitimately slow/flaky, so
`.config/nextest.toml` caps their concurrency and retries them. A few crates also
carry [Criterion](https://bheisler.github.io/criterion.rs/) benchmarks
(`cargo bench`) — the engine, the OpenSearch sink, and the Postgres source.

Editor support: point your editor at
`libs/2-schema/1-config-toml/schemas/config.schema.json` and
`libs/2-schema/1-index-yaml/schemas/index.schema.yml` for completion and inline
validation — the bundled example schemas already reference them via a
`yaml-language-server` modeline.

Want to hack on it? [`CLAUDE.md`](CLAUDE.md) has the architecture tour and the
house rules (the lints are strict — no `unwrap`, no `println!`, and yes, they fail
the build).

## License

Licensed under the [Apache License, Version 2.0](LICENSE). Unless you explicitly
state otherwise, any contribution intentionally submitted for inclusion in this
work shall be licensed as above, without any additional terms or conditions.
