Skip to content

feat(database): add SQL live updates via triggers and change-log #1602

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
JohnChantz wants to merge 4 commits into from
Open

feat(database): add SQL live updates via triggers and change-log #1602

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
584bb74
feat(database): add SQL live updates via triggers and change-log
JohnChantz Sep 11, 2026
3031479
refactor(database): drop unused SQL realtime aliases
JohnChantz Sep 11, 2026
b79d4b8
fix(database): SQL live-update queue, PK, and topology interventions …
kkopanidis Sep 13, 2026
50b76e4
Merge pull request #1609 from ConduitPlatform/cursor/sql-wal-cdc-687f
kkopanidis Sep 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 36 additions & 28 deletions modules/database/README.mdx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,20 +34,20 @@ since the latter need to go through parsers that are otherwise unnecessary for M

## Environment Variables 📃 <a name="env-vars"></a>

| Variable | Description | Required | Example | Default |
|:--------------------:|:-----------------------------------------------------| :------: | :----------------: | :------: |
| `CONDUIT_SERVER` | Conduit Core's address and port | True | `0.0.0.0:55152` | - |
| `SERVICE_URL` | This should be where this service listens on. If behind a LB it should point to the LB's IP/DNS | False | `0.0.0.0:55190` |
| `GRPC_PORT` | The port number the gRPC server will listen to | False | `55190` |
| `GRPC_KEY` | Specifying a secret enables gRPC signed request protection (**use across modules**) | False | `someRandomSecret` | - |
| `DB_CONN_URI` | DB Connection URI | False | `postgres://conduit:pass@localhost:5432/conduit` | `mongodb://localhost:27017` |
| `DB_TYPE` | DB Engine Type | False | `postgres` | `mongodb` |
| Variable | Description | Required | Example | Default |
| :--------------: | :---------------------------------------------------------------------------------------------- | :------: | :----------------------------------------------: | :-------------------------: |
| `CONDUIT_SERVER` | Conduit Core's address and port | True | `0.0.0.0:55152` | - |
| `SERVICE_URL` | This should be where this service listens on. If behind a LB it should point to the LB's IP/DNS | False | `0.0.0.0:55190` |
| `GRPC_PORT` | The port number the gRPC server will listen to | False | `55190` |
| `GRPC_KEY` | Specifying a secret enables gRPC signed request protection (**use across modules**) | False | `someRandomSecret` | - |
| `DB_CONN_URI` | DB Connection URI | False | `postgres://conduit:pass@localhost:5432/conduit` | `mongodb://localhost:27017` |
| `DB_TYPE` | DB Engine Type | False | `postgres` | `mongodb` |

## Replica Set Configuration

When using MongoDB with a replica set (e.g., MongoDB Atlas), the database module supports configuring read preference, write concern, and read concern through the admin panel at `PATCH /config/database`.

Live document updates also require a replica set or sharded cluster. Local Compose files initialize a single-node replica set so change streams can be exercised.
Live document updates on MongoDB also require a replica set or sharded cluster. Local Compose files initialize a single-node replica set so change streams can be exercised. PostgreSQL live updates require logical replication (`wal_level=logical`, a `pgoutput` publication, and a replication slot) — the same class of topology tax as a Mongo replica set. MySQL, MariaDB, and SQLite live updates are out of v1.

### Live updates

Expand All @@ -60,22 +60,29 @@ unsubscribe({ schema: 'Order', documentId?: string })

Events arrive as `change` with `{ version, operation, schema, documentId, occurredAt, resumeToken }` and contain no document fields. Consumers should refetch through their authorized REST or custom-endpoint path.

MongoDB uses native change streams. PostgreSQL uses in-process WAL CDC (`pgoutput` publication + a **temporary** logical slot). That is WAL CDC, Postgres-only — not a changelog table, not triggers, not Debezium. There is **no catch-up**: re-subscribe, leader restart, or slot drop does not replay missed events; clients fetch current data. `LISTEN`/`NOTIFY` is not the capture path.

The database role needs permission to `CREATE PUBLICATION`, `ALTER PUBLICATION`, and to create a logical replication slot (`REPLICATION` / managed-Postgres logical-replication grants). A transaction-mode pooler cannot speak the replication protocol; use a direct/session URI. Tables without a primary key get `REPLICA IDENTITY FULL` so UPDATE/DELETE can be published. `TRUNCATE` is not emitted as document events. Custom PKs use the schema’s physical primary key (`idField`), not a virtual `_id`.

MySQL, MariaDB, and SQLite are unsupported for live updates. Do not enable `realtime` on those engines.

Client subscribers must authenticate. Schemas with document-level authorization reject schema-wide subscriptions and require a document ID plus a `read` check. Admin consumers use `POST /realtime/ticket` for a 30-second handshake token; session JWTs and masterkeys must not be sent from browser code.

Admin sockets must be enabled (`admin.transports.sockets`) and the Admin socket port (`ADMIN_SOCKET_PORT`, default 3031) reachable from the UI.

### Configuration Options

| Setting | Values | Default | Description |
| :---------------: | :------------------------------------------------------------------------ | :-------: | :------------------------------------------------------- |
| `readPreference` | `primary`, `primaryPreferred`, `secondary`, `secondaryPreferred`, `nearest` | `primary` | Controls which replica set members receive read queries |
| `writeConcern` | `1`, `majority` | `1` | How many members must acknowledge a write |
| `readConcern` | `local`, `available`, `majority`, `linearizable`, `snapshot` | `local` | Consistency level for read operations |
| `realtime.enabled` | `true`, `false` | `false` | Enable MongoDB change-stream live updates for opted-in schemas |
| Setting | Values | Default | Description |
| :----------------: | :-------------------------------------------------------------------------- | :-------: | :------------------------------------------------------ |
| `readPreference` | `primary`, `primaryPreferred`, `secondary`, `secondaryPreferred`, `nearest` | `primary` | Controls which replica set members receive read queries |
| `writeConcern` | `1`, `majority` | `1` | How many members must acknowledge a write |
| `readConcern` | `local`, `available`, `majority`, `linearizable`, `snapshot` | `local` | Consistency level for read operations |
| `realtime.enabled` | `true`, `false` | `false` | Enable live updates for opted-in schemas |

### Recommended Production Settings

For MongoDB Atlas deployments with read replicas:

- **readPreference**: `secondaryPreferred` — distributes reads across replicas, falls back to primary
- **writeConcern**: `majority` — ensures writes survive replica set elections
- **readConcern**: `local` — suitable for most workloads
Expand All @@ -87,7 +94,7 @@ Modules can override the configured readPreference on individual queries when th
```typescript
const doc = await MySchema.getInstance().findOne(
{ _id: someId },
{ readPreference: 'primary' }
{ readPreference: 'primary' },
);
```

Expand All @@ -100,19 +107,19 @@ Standalone deployments (without replicas) work identically regardless of these s

| Operator | Description |
| :--------: | :----------------------------------------------------------------------------------------------------------------------- |
| `in` | Matches any of the values specified in an array |
| `in` | Matches any of the values specified in an array |
| `contains` | Checks if a value is contained in an array or not. |
| `nin` | Selects the documents where the value of a field is not equal any value in the specified array. |
| `eq` | Matches documents where the value of a field equals the specified value. |
| `ne` | Selects the documents where the value of the field is not equal to the specified value |
| `lt` | Selects the documents where the value of the field is less than (i.e. <) the specified value. |
| `gt` | Selects the documents where the value of the field is greater than (i.e. >) the specified value. |
| `lte` | Selects the documents where the value of the field is less or equal than (i.e. <=) the specified value. |
| `gte` | Selects the documents where the value of the field is greater or equal than (i.e. >=) the specified value. |
| `or` | Performs OR operation on an array of two or more expressions and selects the documents that satisfy at least one |
| `and` | Performs AND operation on an array of two or more expressions and selects the documents that satisfy at least one |
| `not` | Performs NOT operation on an array of two or more expressions and selects the documents that do not match the expression |
| `regex` | Select the documents where the value of the field matches the regex. |
| `nin` | Selects the documents where the value of a field is not equal any value in the specified array. |
| `eq` | Matches documents where the value of a field equals the specified value. |
| `ne` | Selects the documents where the value of the field is not equal to the specified value |
| `lt` | Selects the documents where the value of the field is less than (i.e. <) the specified value. |
| `gt` | Selects the documents where the value of the field is greater than (i.e. >) the specified value. |
| `lte` | Selects the documents where the value of the field is less or equal than (i.e. <=) the specified value. |
| `gte` | Selects the documents where the value of the field is greater or equal than (i.e. >=) the specified value. |
| `or` | Performs OR operation on an array of two or more expressions and selects the documents that satisfy at least one |
| `and` | Performs AND operation on an array of two or more expressions and selects the documents that satisfy at least one |
| `not` | Performs NOT operation on an array of two or more expressions and selects the documents that do not match the expression |
| `regex` | Select the documents where the value of the field matches the regex. |

These operators have been tested thoroughly in varying levels of complexity.

Expand All @@ -137,6 +144,7 @@ add its id in place of the data.
## Sequelize Caveats

Currently, the Sequelize implementation has the following limitations:

- No index creation
- Update queries only update provided fields without the option to replace entire rows, unless all columns are provided.
- Like operations do not work
3 changes: 2 additions & 1 deletion modules/database/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,9 +72,10 @@
"@jest/globals": "^30.5.1",
"@types/convict": "^6.1.6",
"@types/jest": "^30.0.0",
"@types/object-hash": "^3.0.6",
"@types/lodash-es": "^4.17.12",
"@types/node": "24.13.3",
"@types/object-hash": "^3.0.6",
"@types/pg": "^8.23.1",
"copyfiles": "^2.4.1",
"jest": "^30.5.1",
"rimraf": "^6.1.3",
Expand Down
1 change: 1 addition & 0 deletions modules/database/src/adapters/sequelize-adapter/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,7 @@ export abstract class SequelizeAdapter extends DatabaseAdapter<SequelizeSchema>
this.models['_DeclaredSchema'].originalSchema.collectionName;
for (const table of tableNames) {
if (table === declaredSchemaTableName) continue;
if (table.startsWith('_cnd_')) continue;
const tableInDeclaredSchemas = declaredSchemas.some(
(declaredSchema: ConduitSchema) => {
if (declaredSchema.collectionName && declaredSchema.collectionName !== '') {
Expand Down
2 changes: 1 addition & 1 deletion modules/database/src/config/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ const AppConfigSchema = {
},
realtime: {
enabled: {
doc: 'Enable MongoDB change-stream live updates for opted-in schemas',
doc: 'Enable live updates for opted-in schemas',
format: 'Boolean',
default: false,
},
Expand Down
Loading