feat: Publish committed records to Kafka document topic by justincorrigible · Pull Request #208 · overture-stack/lyric

justincorrigible · 2026-06-26T19:18:28Z

Summary

Add Kafka publishing to Lyric: after each commit, affected records are pushed as individual JSON messages to a configurable Kafka topic. A published_at timestamp on the submissions table now tracks successful publication without touching submission status or the existing state machine.

Issues

No ticket. needed for overture-dev setup.

Description of Changes

This PR adds opt-in Kafka publishing. Setting KAFKA_BROKERS enables it; omitting it leaves Lyric behaviour unchanged.

NOTE: onFinishCommit's type signature changed, and may require adjustment when integrating.

`packages/data-model`

Added nullable published_at timestamp column to the submissions table
Migration 0013: ALTER TABLE submissions ADD COLUMN published_at timestamp; backfills the Unix epoch (1970-01-01) for all existing COMMITTED rows as a "pre-Kafka" sentinel (distinguishes "committed before Kafka existed" from "publish failed")

`packages/data-provider`

Breaking change: onFinishCommit callback signature changed from () => void to (resultOnCommit: ResultOnCommit) => Promise<void> - coordinated with PCGL/iMicroSeq (@leoraba)
New src/external/kafkaPublisher.ts: createKafkaPublisher factory implementing the document topic pattern - one message per affected record (inserts, updates, deletes), batched in a single producer.send call; deleted records have isValid forced to false
New src/external/kafkaPublishTracker.ts: createPublishTracker writes published_at timestamp on successful send; publish and tracking errors are isolated in separate try/catch blocks
provider() accepts an optional db parameter so the server can share one DB pool between the provider and the publish tracker
New exports: createKafkaPublisher, createPublishTracker, KafkaProducer, KafkaPublisherConfig, connect, getLogger, Logger
17 new mocha/chai unit tests for createKafkaPublisher in test/unit/external/kafkaPublisher.spec.ts
New docs/kafka-publishing.md: integration guide for library consumers (message format, wiring pattern, config reference, error handling, publish tracking semantics)

`apps/server`

Added kafkajs dependency
New src/config/kafka.ts: setupKafka reads env vars, connects producer, checks/creates the topic via Admin client on startup, wires createKafkaPublisher + createPublishTracker; kafkajs retry config (5 retries, 300ms-30s exponential backoff); graceful disconnect
src/config/app.ts: refactored from a static appConfig export to buildAppConfig(overrides) + getDbConfig() for composable setup that accepts the Kafka callback
src/server.ts: creates a shared logger and DB pool before the provider; passes both to setupKafka and provider; Kafka disconnect included in graceful shutdown sequence before worker pool drain
apps/server/.env.schema: added KAFKA_BROKERS, KAFKA_TOPIC, KAFKA_CLIENT_ID (all commented - opt-in)

Special Instructions

Run pnpm install to install kafkajs
Run migration 0013 (pnpm run migrate-dev)
Kafka is opt-in: set KAFKA_BROKERS to enable; omit it to run unchanged
New env vars (all optional when Kafka is not used): KAFKA_BROKERS, KAFKA_TOPIC, KAFKA_CLIENT_ID - see DEVELOPMENT.md

Readiness Checklist

Self Review
- I have performed a self review of code
- I have run the application locally and manually tested the feature
- I have checked all updates to correct typos and misspellings
Formatting
- Code follows the project style guide
- Automated code formatters (ie. Prettier) have been run
Local Testing
- Successfully built all packages locally
- Successfully ran all test suites, all unit and integration tests pass
Updated Tests
- 17 new unit tests added for createKafkaPublisher covering the publish path, onSuccess callback, onError callback, and empty result short-circuit
Documentation
- New environment variables added to apps/server/.env.schema and documented in DEVELOPMENT.md
- All changes to server HTTP endpoints have open-api documentation (no new endpoints in this PR)
- All new functions exported from their module have TSDoc comment documentation

justincorrigible · 2026-06-27T02:45:27Z

pushed an update because I realized the package I had used wasn't compatible with the latest versions of Kafka we're deploying in overture-dev. the new one supports v3.x (used in other envs) and v4.x (the current one).

note: had some issues with the package in the build pipeline because of the way we're blocking the post install scripts. the dockerfile acrobatics are an isolated solution for it without compromising the rest of the image

leoraba · 2026-07-15T22:10:30Z

  # all packages in subdirs of packages/
  - 'packages/*'
 allowBuilds:
+  '@confluentinc/kafka-javascript': true


is it really needed to be true?

Yes, because of the way our build pipeline works: @confluentinc/kafka-JavaScript bundles a native add-on built on librdkafka (a C library). It uses node-pre-gyp in its install lifecycle script to download a pre-built platform binary (the .node file). Without that binary, the package fails at runtime e.g. once deployed, with "Could not locate the bindings file" as a critical error.

The cause is that pnpm workspaces block all package lifecycle scripts by default, even on a plain pnpm install with no flags (which is a security feature we want). Now, to state the obvious, to benefit those who don't know, allowBuilds is the workspace-level opt-in list for which packages are permitted to run those scripts.... this should be used sparingly, and it works as an exception to that security feature, which is why @leoraba is asking to confirm.
Without '@confluentinc/kafka-javascript': true, pnpm silently skips its install script and the binary never lands, breaking both local dev and even the CI build stage (which does a full install before compilation).

The Dockerfile prod-deps stage is a separate concern: it uses --no-scripts explicitly as an OWASP requirement, which overrides allowBuilds entirely; so there we have to invoke node-pre-gyp manually after the fact.
i.e. allowBuilds only matters for installs that don't pass --no-scripts.

Hope that clarifies why this line is here.

leoraba

Approved, just left a comment regarding the message format, which might need an improvement later

joneubank · 2026-07-16T19:08:03Z

+	topic: string;
+};
+
+const toMessage = (record: SubmittedDataResponse, action: 'delete' | 'insert' | 'update') => ({


Surprised we don't have an existing type that enumerates these options. Would prefer these be defined as constants with a corresponding type. This is reflective of the type options that already exist, so it might exist somewhere

Adding a new type for this, because the only ones that already exist are:

SUBMISSION_ACTION_TYPE: ['INSERTS', 'UPDATES', 'DELETES'] which are plural, uppercase, and represent submission categories, not individual record events.
AUDIT_ACTION: ['UPDATE', 'DELETE', 'MIGRATION'] with no insert, includes MIGRATION (?), and "audit" seems like it would diverge over time.

joneubank · 2026-07-16T19:10:45Z


 				if (configData.onFinishCommit && resultCommit) {
-					configData.onFinishCommit(resultCommit);
+					await configData.onFinishCommit(resultCommit);


@leoraba is this a case where we intended the onFinishCommit to occur asynchronously and not be waited for? or is this a missed await? if the former, needs a comment to clarify why its not being waited for.

submissionService.ts:99 calls workerPool.commitSubmission(commitData) with no await: as the worker runs fully async and the HTTP 200 goes back immediately with PROCESSING status... so the await here is meant to just control the sequencing within the worker's own flow.

may be worth just adding a comment to make the intent obvious 🤷 (adding one now, for good measure)

I've reviewed the context - onFinishCommit likely does not need to be awaited, but this is happening in a worker already so we're not blocking responses to a user so its safer to await so that our error handling can handle anything unexpected in this function call.

tldr - await is good ✅

joneubank · 2026-07-16T19:46:51Z

+		logger.error('[kafka] Failed to connect to broker(s) after retries, shutting down', err);
+		throw err;
+	}
+	logger.info(`[kafka] Connected (clientId: ${clientId}). Publishing commits to topic: ${topic}`);


👍 was confirming that we log the clientId, since its important that the logs say it if we end up using the default.

justincorrigible · 2026-07-16T20:34:10Z

 		allowedOrigins: process.env.ALLOWED_ORIGINS?.split(',') || [],
 		corsEnabled: getBoolean(process.env.CORS_ENABLED, false),
+		logLevel: process.env.LOG_LEVEL || 'info',
+		port: process.env.PORT || 3030,


alphabetics are bae 👍

justincorrigible force-pushed the kafka branch 2 times, most recently from 08eac38 to 97e86bf Compare June 26, 2026 19:36

justincorrigible requested review from joneubank and leoraba June 26, 2026 19:42

leoraba reviewed Jun 26, 2026

View reviewed changes

Comment thread packages/data-model/src/models/submissions.ts Outdated

justincorrigible force-pushed the kafka branch from 97e86bf to 9640669 Compare June 27, 2026 02:28

justincorrigible force-pushed the kafka branch 6 times, most recently from 6780df7 to 09c8ddb Compare June 27, 2026 03:35

add kafka publishing capabilities

dc65ccc

justincorrigible force-pushed the kafka branch from 09c8ddb to dc65ccc Compare July 15, 2026 19:09

leoraba reviewed Jul 15, 2026

View reviewed changes

Comment thread packages/data-model/migrations/meta/0014_snapshot.json Outdated

leoraba reviewed Jul 15, 2026

View reviewed changes

Comment thread apps/server/src/config/kafka.ts Outdated

leoraba reviewed Jul 15, 2026

View reviewed changes

update snapshot, handle connection failure loops

3c794a3

leoraba reviewed Jul 16, 2026

View reviewed changes

Comment thread packages/data-provider/docs/kafka-publishing.md

leoraba approved these changes Jul 16, 2026

View reviewed changes

add action, remove isValid coercion

20b2dcc

joneubank reviewed Jul 16, 2026

View reviewed changes

Comment thread apps/server/src/server.ts

joneubank reviewed Jul 16, 2026

View reviewed changes

Comment thread apps/server/src/config/kafka.ts Outdated

joneubank reviewed Jul 16, 2026

View reviewed changes

Comment thread apps/server/src/config/kafka.ts

joneubank reviewed Jul 16, 2026

View reviewed changes

Comment thread apps/server/src/config/kafka.ts Outdated

address further PR feedback

2743c4f

and further feedback

45ef91c

joneubank approved these changes Jul 16, 2026

View reviewed changes

justincorrigible added 2 commits July 16, 2026 16:20

add logger level config from ENV

19b257b

add roadmap item for Kafka message format

75f8868

justincorrigible commented Jul 16, 2026

View reviewed changes

justincorrigible merged commit c4990bc into main Jul 16, 2026

justincorrigible deleted the kafka branch July 16, 2026 20:38

leoraba mentioned this pull request Jul 17, 2026

Release 0.18.0 #211

Open

Uh oh!

Conversation

justincorrigible commented Jun 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Issues

Description of Changes

packages/data-model

packages/data-provider

apps/server

Special Instructions

Readiness Checklist

Uh oh!

Uh oh!

justincorrigible commented Jun 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

leoraba left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

justincorrigible Jul 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

justincorrigible commented Jun 26, 2026 •

edited

Loading

`packages/data-model`

`packages/data-provider`

`apps/server`

justincorrigible commented Jun 27, 2026 •

edited

Loading

justincorrigible Jul 16, 2026 •

edited

Loading