AntflyAntfly

Stream PostgreSQL into Antfly

An Antfly table that mirrors a PostgreSQL table on its own, every insert, update, and delete included, with PostgreSQL still the system of record

The Result#

Every insert, update, and delete on a PostgreSQL table lands in an Antfly table without a sync job or a daemon in between. Antfly subscribes to PostgreSQL logical replication, backfills the existing rows, then follows the stream. Nothing changes in the application that writes to PostgreSQL. The table is created with a replication source, and rows can be read back from Antfly as soon as they land:

curl -X POST http://127.0.0.1:8080/db/v1/tables/users \
  -u admin:admin \
  -H "Content-Type: application/json" \
  -d '{
    "replication_sources": [
      {
        "type": "postgres",
        "dsn": "${secret:pg_dsn}",
        "postgres_table": "users",
        "key_template": "id"
      }
    ]
  }'

curl -s http://127.0.0.1:8080/db/v1/tables/users/documents/user-1 -u admin:admin

{ "id": "user-1", "name": "Alice", "email": "alice@example.com", "score": 100 }

Before You Start#

Antfly running, PostgreSQL 14 or later with wal_level = logical (15 or later only if you use publication_filter), and a database user with the REPLICATION attribute:

SHOW wal_level;  -- logical

Build It#

1. Turn On Logical Replication#

On self-hosted PostgreSQL, set the WAL level and give the stream room for one slot and one sender per replication source:

ALTER SYSTEM SET wal_level = logical;
ALTER SYSTEM SET max_replication_slots = 10;
ALTER SYSTEM SET max_wal_senders = 10;

Restart PostgreSQL, then confirm:

SHOW wal_level;  -- logical

wal_level is server-wide, not per database. Managed services set it through a flag or a console switch instead; Managed PostgreSQL Setup has each one.

Grant the replication attribute to the user Antfly will connect as:

ALTER USER antfly WITH REPLICATION;

2. Store the Connection String as a Secret#

The DSN carries a password, so it goes in the secret store and the table config references it by name:

curl -X PUT http://127.0.0.1:8080/db/v1/secrets/pg_dsn \
  -u admin:admin \
  -H "Content-Type: application/json" \
  -d '{"value": "postgres://antfly:secret@pg-host:5432/mydb?sslmode=require"}'

Use the direct connection, never a pooler. PgBouncer and Supavisor do not speak the replication protocol, and a DSN that points at one reconnects forever without an obvious error. The secrets API writes to the local store, so it answers in standalone mode; on a multi-node cluster it returns 503, and the DSN goes in the secret store file each node is started with instead.

3. Create the Replicated Table#

The source table in PostgreSQL:

CREATE TABLE users (
    id TEXT PRIMARY KEY,
    name TEXT,
    email TEXT,
    score INT
);

The Antfly table, with a replication_sources block naming it:

curl -X POST http://127.0.0.1:8080/db/v1/tables/users \
  -u admin:admin \
  -H "Content-Type: application/json" \
  -d '{
    "replication_sources": [
      {
        "type": "postgres",
        "dsn": "${secret:pg_dsn}",
        "postgres_table": "users",
        "key_template": "id"
      }
    ]
  }'

Antfly creates a replication slot and a publication, snapshots the rows already in the table through the normal batch path, then polls the slot for changes. The source's status.phase in the table detail moves from snapshot to streaming:

curl -s http://127.0.0.1:8080/db/v1/tables/users -u admin:admin

Three failures show up here. status.last_error names the Antfly error, such as ForeignQueryFailed; PostgreSQL's own message is in the Antfly log. permission denied to create replication slot means the user lacks REPLICATION (or the managed-service role); pre-create the slot and pass slot_name. publication does not exist means the user cannot create publications; have the table owner create one and pass publication_name. wal_level is not logical means step 1 did not take effect, usually because the restart is still pending.

4. Watch a Row Arrive#

Insert in PostgreSQL:

INSERT INTO users (id, name, email, score) VALUES ('user-1', 'Alice', 'alice@example.com', 100);

Read it back from Antfly:

curl -s http://127.0.0.1:8080/db/v1/tables/users/documents/user-1 -u admin:admin

{ "id": "user-1", "name": "Alice", "email": "alice@example.com", "score": 100 }

Every column became a field. Updates flow the same way. Replicated rows go through the normal write path, so any embeddings index on the table embeds them as they land, exactly as a direct write would.

If nothing arrives, the DSN is the first suspect: check that PostgreSQL is reachable from the Antfly metadata node, that sslmode=require is set for cloud providers, and that the host is the direct endpoint rather than a pooler.

5. Shape the Document#

Three fields on the source control how a row becomes a document.

key_template derives the document key. When omitted, Antfly uses the _id column if present and falls back to id; set it explicitly when the source uses another primary key. A column name uses that column; a template composes several:

"key_template": "{{tenant_id}}:{{user_id}}"

publication_filter is a PostgreSQL 15+ row filter, applied only when Antfly creates the publication. Changing it later does not alter an existing publication.

on_update replaces passthrough with explicit transforms for inserts and updates. Column references resolve to the row; literals are set every time:

"on_update": [
  { "op": "$set", "path": "name", "value": "{{user_name}}" },
  { "op": "$set", "path": "email", "value": "{{user_email}}" },
  { "op": "$set", "path": "active", "value": true },
  { "op": "$merge", "value": "{{metadata}}" }
]

$merge flattens a JSONB column into top-level fields: {"role": "admin", "team": "eng"} becomes role and team on the document. {{metadata.role}} reaches into decoded JSONB for a single key.

on_delete decides what a PostgreSQL DELETE does. Remove the document:

"on_delete": [
  { "op": "$delete_document" }
]

Or keep it and mark it:

"on_delete": [
  { "op": "$set", "path": "active", "value": false }
]

Omit on_delete and Antfly derives $unset operations from the on_update paths, removing only the fields this source set. With neither set, a delete unsets every replicated column except the key fields.

For deletes to arrive at all, the source table needs a replica identity. A primary key provides one. A table without one needs:

ALTER TABLE users REPLICA IDENTITY FULL;

Several PostgreSQL tables can feed one Antfly table. Give each source its own on_update and the auto-derived deletes stay scoped: deleting a row from scores unsets score and level and leaves name and email from the users source in place:

"replication_sources": [
  {
    "type": "postgres",
    "dsn": "${secret:pg_dsn}",
    "postgres_table": "users",
    "key_template": "id",
    "on_update": [
      { "op": "$set", "path": "name", "value": "{{name}}" },
      { "op": "$set", "path": "email", "value": "{{email}}" }
    ]
  },
  {
    "type": "postgres",
    "dsn": "${secret:pg_dsn}",
    "postgres_table": "scores",
    "key_template": "user_id",
    "on_update": [
      { "op": "$set", "path": "score", "value": "{{score}}" },
      { "op": "$set", "path": "level", "value": "{{level}}" }
    ]
  }
]

Managed PostgreSQL Setup#

Every provider needs the same two things, wal_level = logical and a user allowed to replicate, and each exposes them differently. Pre-create the publication as the table owner whenever the replication user does not own the table, then pass publication_name. Antfly still checks for the named publication on every poll and creates it if it is missing, so the name has to exist before the replication user connects:

CREATE PUBLICATION antfly_pub_users FOR TABLE users;

Google Cloud SQL. Set the instance flag cloudsql.logical_decoding = on (console: Instance > Configuration > Flags, or gcloud sql instances patch INSTANCE_NAME --database-flags=cloudsql.logical_decoding=on); the instance restarts. Grant with ALTER USER myuser WITH REPLICATION;, or the cloudsql.replication role if the user cannot be altered. Connect through the Cloud SQL Auth Proxy or direct IP with SSL.

Google AlloyDB. Set alloydb.logical_decoding = on (gcloud alloydb instances update INSTANCE_NAME --cluster=CLUSTER_NAME --region=REGION --database-flags=alloydb.logical_decoding=on); the instance restarts. Replication grants and publications work as on Cloud SQL, with the alloydb.replication role.

Supabase. Logical replication is on by default. Create the publication in the SQL editor with the public. schema prefix, then use the direct connection string from Settings > Database > Connection string > URI, on port 5432. Port 6543 is the pooler and cannot replicate. Pass both publication_name and a slot_name.

Neon. Enable logical replication under Project > Settings > Logical Replication; the compute endpoint restarts. Create the publication in the SQL editor. Use the direct (unpooled) connection string with sslmode=require. Slots are supported on the primary branch; Neon's logical replication guide lists current limits.

Amazon RDS and Aurora PostgreSQL. Set rds.logical_replication = 1 in the DB or cluster parameter group (aws rds modify-db-parameter-group --db-parameter-group-name my-param-group --parameters "ParameterName=rds.logical_replication,ParameterValue=1,ApplyMethod=pending-reboot") and reboot. Grant with GRANT rds_replication TO myuser;. On Aurora, point the DSN at the writer endpoint; readers cannot serve as replication sources.

Tradeoffs#

Two things are yours to decide: what happens to the replication slot when Antfly is down, and what a PostgreSQL delete means for the document.

Replication runs one way. Antfly never writes back, so your application keeps its transactions, constraints, and backups exactly where they are, and Antfly holds a searchable copy that is as current as the stream.

The cost of that arrangement is the replication slot. A slot makes PostgreSQL retain WAL until the subscriber has consumed it. While Antfly is up, that is a few seconds of WAL. While Antfly is down for a long stretch, PostgreSQL keeps every change since the last checkpoint, and disk usage grows until the stream resumes. Monitor slot lag the way you would for any logical subscriber, and if you decommission an Antfly table, drop its slot in PostgreSQL rather than leaving it to hold WAL forever.

That one-way flow is also why the delete decision is yours. PostgreSQL knows a row is gone; only you know whether the document should vanish from search or stay, marked inactive, so that history remains retrievable. $delete_document is right for data that should not be found once removed. A soft delete is right when the document is evidence, such as a resolved ticket that still teaches a classifier.

Use Agent Skills#

Everything above is also encoded in the Antfly skill, so a coding agent can execute this guide for you:

npx skills add antflydb/antfly-skills

Then prompt it with the outcome, for example "Stream my Postgres orders table into Antfly with logical replication and soft-delete on DELETE", and use this page to judge the result.

Next Steps#