flusso

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 — running in about five commands
• What it does — the two files you write
• How the pipeline works — the bit that does the work
• The CLI — build, check, run
• just commands — the shortcuts you'll actually use
• Requirements — what Postgres and OpenSearch need first
• Deploying it — Docker image + Helm chart
• Docs — everything else, linked
• Project layout — where the code lives
• Testing & development

Quickstart

The 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 installed (cargo install just --locked):

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:

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.

No just? Every recipe is just a thin wrapper; the raw cargo run -- … and docker compose … commands are right there in the 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.

[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.

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 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:

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.