From 6932d79ee07afb8f8870fe205c2a2edcb0ae6934 Mon Sep 17 00:00:00 2001 From: priyamkarn Date: Thu, 18 Jun 2026 04:05:21 +0530 Subject: [PATCH 1/2] fix: address kafka plugin review feedback --- dtk/cli/add.ts | 1 + dtk/package-lock.json | 41 ++---- dtk/package.json | 1 + dtk/templates/plugins/kafka/README.md | 120 ++++++++++++++++++ dtk/templates/plugins/kafka/env.txt | 2 + dtk/templates/plugins/kafka/example.ts | 36 ++++++ dtk/templates/plugins/kafka/plugin.json | 37 ++++++ dtk/templates/plugins/kafka/service.test.ts | 133 ++++++++++++++++++++ dtk/templates/plugins/kafka/service.ts | 64 ++++++++++ dtk/templates/plugins/kafka/types.ts | 20 +++ example/.env.template | 2 + example/GUIDE.md | 65 ++++++++++ example/package-lock.json | 40 ++---- example/package.json | 6 +- example/src/runbooks/kafka.ts | 36 ++++++ example/src/services/kafka.test.ts | 131 +++++++++++++++++++ example/src/services/kafka.ts | 64 ++++++++++ example/src/suite.ts | 5 + example/src/types/kafka.ts | 20 +++ example/src/types/suite.ts | 2 + tools/README.md | 18 +++ tools/kafka/docker-compose.yml | 33 +++++ 22 files changed, 815 insertions(+), 62 deletions(-) create mode 100644 dtk/templates/plugins/kafka/README.md create mode 100644 dtk/templates/plugins/kafka/env.txt create mode 100644 dtk/templates/plugins/kafka/example.ts create mode 100644 dtk/templates/plugins/kafka/plugin.json create mode 100644 dtk/templates/plugins/kafka/service.test.ts create mode 100644 dtk/templates/plugins/kafka/service.ts create mode 100644 dtk/templates/plugins/kafka/types.ts create mode 100644 example/src/runbooks/kafka.ts create mode 100644 example/src/services/kafka.test.ts create mode 100644 example/src/services/kafka.ts create mode 100644 example/src/types/kafka.ts create mode 100644 tools/kafka/docker-compose.yml diff --git a/dtk/cli/add.ts b/dtk/cli/add.ts index abe4bac..1dc2f37 100644 --- a/dtk/cli/add.ts +++ b/dtk/cli/add.ts @@ -16,6 +16,7 @@ const PLUGIN_MAP: Record = { 'open-ai': 'open-ai', 'redis': 'redis', 'sql': 'sql', + 'kafka': 'kafka', }; interface PluginTransform { diff --git a/dtk/package-lock.json b/dtk/package-lock.json index 1f3a8cc..cc83628 100644 --- a/dtk/package-lock.json +++ b/dtk/package-lock.json @@ -26,6 +26,7 @@ "axios": "^1.17.0", "dotenv": "^17.0.0", "jest": "^30.4.2", + "kafkajs": "^2.2.4", "knex": "^3.1.0", "openai": "^6.42.0", "redis": "^6.0.0", @@ -2677,9 +2678,6 @@ "arm64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2694,9 +2692,6 @@ "arm64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -2711,9 +2706,6 @@ "loong64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2728,9 +2720,6 @@ "loong64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -2745,9 +2734,6 @@ "ppc64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2762,9 +2748,6 @@ "riscv64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2779,9 +2762,6 @@ "riscv64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -2796,9 +2776,6 @@ "s390x" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2813,9 +2790,6 @@ "x64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2830,9 +2804,6 @@ "x64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -5122,6 +5093,16 @@ "node": ">=6" } }, + "node_modules/kafkajs": { + "version": "2.2.4", + "resolved": "https://registry.npmjs.org/kafkajs/-/kafkajs-2.2.4.tgz", + "integrity": "sha512-j/YeapB1vfPT2iOIUn/vxdyKEuhuY2PxMBvf5JWux6iSaukAccrMtXEY/Lb7OvavDhOWME589bpLrEdnVHjfjA==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">=14.0.0" + } + }, "node_modules/knex": { "version": "3.2.10", "resolved": "https://registry.npmjs.org/knex/-/knex-3.2.10.tgz", diff --git a/dtk/package.json b/dtk/package.json index c8a0d75..1d8f8d7 100644 --- a/dtk/package.json +++ b/dtk/package.json @@ -45,6 +45,7 @@ "axios": "^1.17.0", "dotenv": "^17.0.0", "jest": "^30.4.2", + "kafkajs": "^2.2.4", "knex": "^3.1.0", "openai": "^6.42.0", "redis": "^6.0.0", diff --git a/dtk/templates/plugins/kafka/README.md b/dtk/templates/plugins/kafka/README.md new file mode 100644 index 0000000..0468db0 --- /dev/null +++ b/dtk/templates/plugins/kafka/README.md @@ -0,0 +1,120 @@ +# kafka plugin + +Adds Kafka support to your dtk suite via [KafkaJS](https://kafka.js.org/). + +Provides three methods on `ctx.services.kafka`: **produce**, **consume**, and **disconnect**. + +--- + +## Installation + +```bash +dtk add kafka +``` + +This copies `src/services/kafka.ts` and `src/types/kafka.ts` into your project, patches `suite.ts` and `types/suite.ts`, and installs `kafkajs`. + +--- + +## Configuration + +Add the following to your `.env` (the template is added automatically): + +``` +KAFKA_BROKERS=localhost:9092 +KAFKA_CLIENT_ID=dtk-client +``` + +For multiple brokers, comma-separate them: + +``` +KAFKA_BROKERS=broker1:9092,broker2:9092,broker3:9092 +``` + +--- + +## Local development + +A `docker-compose.yml` using [Redpanda](https://redpanda.com/) is included for local dev — Redpanda is fully Kafka-API-compatible and starts in seconds. + +```bash +# from dtk/templates/plugins/kafka/ +docker compose up -d + +# connect on localhost:19092 +KAFKA_BROKERS=localhost:19092 +``` + +Redpanda Console (UI) is available at http://localhost:8080. + +--- + +## Usage + +```ts +import "../load-env.js"; +import { suite } from "../suite.js"; + +await suite() + .kafka({ + brokers: process.env.KAFKA_BROKERS!.split(",").map((b) => b.trim()), + clientId: process.env.KAFKA_CLIENT_ID ?? "dtk-client", + }) + .step("produce-message", async (ctx) => { + await ctx.services.kafka.produce({ + topic: "my-topic", + messages: [{ value: "hello" }], + }); + }) + .step("consume-message", async (ctx) => { + await ctx.services.kafka.consume({ + topic: "my-topic", + groupId: "my-group", + fromBeginning: true, + handler: async ({ message }) => { + console.log(message.value?.toString()); + }, + }); + }) + .step("disconnect", async (ctx) => { + await ctx.services.kafka.disconnect(); + }) + .run("stopOnError"); +``` + +--- + +## API + +### `.kafka(config)` + +| Field | Type | Required | Description | +|---|---|---|---| +| `brokers` | `string[]` | ✅ | List of broker addresses | +| `clientId` | `string` | | Kafka client identifier | +| `ssl` | `boolean` | | Enable TLS | +| `sasl` | `object` | | SASL credentials (`mechanism`, `username`, `password`) | + +### `ctx.services.kafka.produce(options)` + +Sends messages to a topic. Connects the producer on first call and reuses it for subsequent calls. + +| Field | Type | Description | +|---|---|---| +| `topic` | `string` | Target topic | +| `messages` | `Message[]` | KafkaJS `Message` array | + +### `ctx.services.kafka.consume(options)` + +Subscribes to a topic and runs the consumer. Calling `consume()` a second time without calling `disconnect()` first will throw — KafkaJS does not allow re-subscribing a running consumer. + +| Field | Type | Description | +|---|---|---| +| `topic` | `string` | Topic to subscribe to | +| `groupId` | `string` | Consumer group ID | +| `fromBeginning` | `boolean` | Read from the earliest offset (default: `false`) | +| `handler` | `(payload: EachMessagePayload) => Promise` | Called for each message | + +### `ctx.services.kafka.disconnect()` + +Disconnects both producer and consumer and resets internal state. Safe to call even if neither was connected. diff --git a/dtk/templates/plugins/kafka/env.txt b/dtk/templates/plugins/kafka/env.txt new file mode 100644 index 0000000..5d48bcb --- /dev/null +++ b/dtk/templates/plugins/kafka/env.txt @@ -0,0 +1,2 @@ +KAFKA_BROKERS=localhost:9092 +KAFKA_CLIENT_ID=dtk-client diff --git a/dtk/templates/plugins/kafka/example.ts b/dtk/templates/plugins/kafka/example.ts new file mode 100644 index 0000000..8e9029e --- /dev/null +++ b/dtk/templates/plugins/kafka/example.ts @@ -0,0 +1,36 @@ +import "../load-env.js"; +import { suite } from "../suite.js"; + +await suite() + .kafka({ + brokers: process.env.KAFKA_BROKERS!.split(",").map((b) => b.trim()).filter(Boolean), + clientId: process.env.KAFKA_CLIENT_ID ?? "dtk-client", + }) + .step("produce-message", async (ctx) => { + await ctx.services.kafka.produce({ + topic: "example-topic", + messages: [{ value: "hello from dtk" }], + }); + console.log("message produced to example-topic"); + }) + .step("consume-message", async (ctx) => { + await ctx.services.kafka.consume({ + topic: "example-topic", + groupId: "dtk-group", + fromBeginning: true, + handler: async (payload) => { + console.log( + "received:", + payload.message.value?.toString(), + "partition:", + payload.partition, + "offset:", + payload.message.offset + ); + }, + }); + }) + .step("disconnect", async (ctx) => { + await ctx.services.kafka.disconnect(); + }) + .run("stopOnError"); diff --git a/dtk/templates/plugins/kafka/plugin.json b/dtk/templates/plugins/kafka/plugin.json new file mode 100644 index 0000000..131174c --- /dev/null +++ b/dtk/templates/plugins/kafka/plugin.json @@ -0,0 +1,37 @@ +{ + "name": "kafka", + "description": "Kafka -- produce, consume, disconnect via KafkaJS", + "dependencies": { + "kafkajs": "^2.2.4" + }, + "files": [ + { "src": "service.ts", "dest": "src/services/kafka.ts" }, + { "src": "types.ts", "dest": "src/types/kafka.ts" }, + { "src": "service.test.ts", "dest": "src/services/kafka.test.ts" } + ], + "env": "env.txt", + "example": "example.ts", + "transforms": { + "service.ts": [ + { "from": "./types.js", "to": "../types/kafka.js" } + ], + "service.test.ts": [ + { "from": "./service.js", "to": "./kafka.js" } + ] + }, + "patches": { + "src/suite.ts": { + "imports": [ + "import { createKafkaService } from \"./services/kafka.js\";", + "import type { KafkaConfig } from \"./types/kafka.js\";" + ], + "configs": " private kafkaConfig?: KafkaConfig;", + "methods": " kafka(config: KafkaConfig): this { this.kafkaConfig = config; return this; }", + "services": " kafka: createKafkaService(this.kafkaConfig)," + }, + "src/types/suite.ts": { + "type-imports": "import type { KafkaConsumeOptions } from \"./kafka.js\";", + "service-types": " kafka: { produce(options: { topic: string; messages: import('kafkajs').Message[] }): Promise; consume(options: KafkaConsumeOptions): Promise; disconnect(): Promise; };" + } + } +} diff --git a/dtk/templates/plugins/kafka/service.test.ts b/dtk/templates/plugins/kafka/service.test.ts new file mode 100644 index 0000000..6ecaac4 --- /dev/null +++ b/dtk/templates/plugins/kafka/service.test.ts @@ -0,0 +1,133 @@ +import { createKafkaService } from './service.js'; + +jest.mock('kafkajs'); +import { Kafka } from 'kafkajs'; + +// Plain jest.fn() at the top — return values are set in beforeEach after clearAllMocks() +const mockProducer = { + connect: jest.fn(), + send: jest.fn(), + disconnect: jest.fn(), +}; + +const mockConsumer = { + connect: jest.fn(), + subscribe: jest.fn(), + run: jest.fn(), + disconnect: jest.fn(), +}; + +beforeEach(() => { + jest.clearAllMocks(); + mockProducer.connect.mockResolvedValue(undefined); + mockProducer.send.mockResolvedValue(undefined); + mockProducer.disconnect.mockResolvedValue(undefined); + mockConsumer.connect.mockResolvedValue(undefined); + mockConsumer.subscribe.mockResolvedValue(undefined); + mockConsumer.run.mockResolvedValue(undefined); + mockConsumer.disconnect.mockResolvedValue(undefined); + (Kafka as jest.Mock).mockImplementation(() => ({ + producer: () => mockProducer, + consumer: () => mockConsumer, + })); +}); + +describe('createKafkaService', () => { + const config = { + brokers: ['localhost:9092'], + clientId: 'test-client', + }; + + describe('produce', () => { + it('connects the producer and sends a message', async () => { + const kafka = createKafkaService(config); + await kafka.produce({ topic: 'test-topic', messages: [{ value: 'hello' }] }); + expect(mockProducer.connect).toHaveBeenCalledTimes(1); + expect(mockProducer.send).toHaveBeenCalledWith({ + topic: 'test-topic', + messages: [{ value: 'hello' }], + }); + }); + + it('reuses the producer on subsequent produce calls', async () => { + const kafka = createKafkaService(config); + await kafka.produce({ topic: 'test-topic', messages: [{ value: 'first' }] }); + await kafka.produce({ topic: 'test-topic', messages: [{ value: 'second' }] }); + expect(mockProducer.connect).toHaveBeenCalledTimes(1); + expect(mockProducer.send).toHaveBeenCalledTimes(2); + }); + + it('throws when called without config', async () => { + const kafka = createKafkaService(); + await expect(kafka.produce({ topic: 'test-topic', messages: [] })).rejects.toThrow( + 'kafka service is not configured' + ); + }); + }); + + describe('consume', () => { + it('connects the consumer, subscribes, and runs with the provided handler', async () => { + const handler = jest.fn(); + const kafka = createKafkaService(config); + await kafka.consume({ topic: 'test-topic', groupId: 'test-group', handler }); + expect(mockConsumer.connect).toHaveBeenCalledTimes(1); + expect(mockConsumer.subscribe).toHaveBeenCalledWith({ + topic: 'test-topic', + fromBeginning: false, + }); + expect(mockConsumer.run).toHaveBeenCalledWith({ eachMessage: handler }); + }); + + it('subscribes fromBeginning when the option is true', async () => { + const kafka = createKafkaService(config); + await kafka.consume({ topic: 'test-topic', groupId: 'test-group', fromBeginning: true, handler: jest.fn() }); + expect(mockConsumer.subscribe).toHaveBeenCalledWith({ + topic: 'test-topic', + fromBeginning: true, + }); + }); + + it('throws if consume is called a second time while the consumer is running', async () => { + const kafka = createKafkaService(config); + await kafka.consume({ topic: 'test-topic', groupId: 'test-group', handler: jest.fn() }); + await expect( + kafka.consume({ topic: 'other-topic', groupId: 'test-group', handler: jest.fn() }) + ).rejects.toThrow('kafka consumer is already running'); + }); + + it('throws when called without config', async () => { + const kafka = createKafkaService(); + await expect( + kafka.consume({ topic: 'test-topic', groupId: 'test-group', handler: jest.fn() }) + ).rejects.toThrow('kafka service is not configured'); + }); + }); + + describe('disconnect', () => { + it('disconnects both producer and consumer', async () => { + const kafka = createKafkaService(config); + await kafka.produce({ topic: 'test-topic', messages: [{ value: 'hi' }] }); + await kafka.consume({ topic: 'test-topic', groupId: 'test-group', handler: jest.fn() }); + await kafka.disconnect(); + expect(mockProducer.disconnect).toHaveBeenCalledTimes(1); + expect(mockConsumer.disconnect).toHaveBeenCalledTimes(1); + }); + + it('is a no-op when neither producer nor consumer was created', async () => { + const kafka = createKafkaService(config); + await expect(kafka.disconnect()).resolves.toBeUndefined(); + expect(mockProducer.disconnect).not.toHaveBeenCalled(); + expect(mockConsumer.disconnect).not.toHaveBeenCalled(); + }); + + it('resets state so consume can be called again after disconnect', async () => { + const kafka = createKafkaService(config); + await kafka.consume({ topic: 'test-topic', groupId: 'test-group', handler: jest.fn() }); + await kafka.disconnect(); + // Should not throw after disconnect resets consumerStarted + await expect( + kafka.consume({ topic: 'test-topic', groupId: 'test-group', handler: jest.fn() }) + ).resolves.toBeUndefined(); + }); + }); +}); diff --git a/dtk/templates/plugins/kafka/service.ts b/dtk/templates/plugins/kafka/service.ts new file mode 100644 index 0000000..6fbfaad --- /dev/null +++ b/dtk/templates/plugins/kafka/service.ts @@ -0,0 +1,64 @@ +import { Kafka } from 'kafkajs'; +import type { Producer, Consumer } from 'kafkajs'; +import type { KafkaConfig, KafkaProduceOptions, KafkaConsumeOptions } from './types.js'; + +export function createKafkaService(config?: KafkaConfig) { + const ensureConfig = () => { + if (!config) throw new Error('kafka service is not configured -- call .kafka(config) on the suite'); + }; + + let producer: Producer | null = null; + let consumer: Consumer | null = null; + let consumerStarted = false; + + const getProducer = async (): Promise => { + ensureConfig(); + if (!producer) { + const kafka = new Kafka({ brokers: config!.brokers, clientId: config!.clientId, ssl: config!.ssl, sasl: config!.sasl }); + producer = kafka.producer(); + await producer.connect(); + } + return producer; + }; + + const getConsumer = async (groupId: string): Promise => { + ensureConfig(); + if (!consumer) { + const kafka = new Kafka({ brokers: config!.brokers, clientId: config!.clientId, ssl: config!.ssl, sasl: config!.sasl }); + consumer = kafka.consumer({ groupId }); + await consumer.connect(); + } + return consumer; + }; + + return { + produce: async (options: KafkaProduceOptions): Promise => { + const p = await getProducer(); + await p.send({ topic: options.topic, messages: options.messages }); + }, + + consume: async (options: KafkaConsumeOptions): Promise => { + if (consumerStarted) { + throw new Error( + 'kafka consumer is already running -- call disconnect() before consuming again' + ); + } + const c = await getConsumer(options.groupId); + await c.subscribe({ topic: options.topic, fromBeginning: options.fromBeginning ?? false }); + await c.run({ eachMessage: options.handler }); + consumerStarted = true; + }, + + disconnect: async (): Promise => { + if (producer) { + await producer.disconnect(); + producer = null; + } + if (consumer) { + await consumer.disconnect(); + consumer = null; + } + consumerStarted = false; + }, + }; +} diff --git a/dtk/templates/plugins/kafka/types.ts b/dtk/templates/plugins/kafka/types.ts new file mode 100644 index 0000000..344b9a6 --- /dev/null +++ b/dtk/templates/plugins/kafka/types.ts @@ -0,0 +1,20 @@ +import type { Message, EachMessagePayload, KafkaConfig as KafkaJSConfig } from 'kafkajs'; + +export interface KafkaConfig { + brokers: string[]; + clientId?: string; + ssl?: KafkaJSConfig['ssl']; + sasl?: KafkaJSConfig['sasl']; +} + +export interface KafkaProduceOptions { + topic: string; + messages: Message[]; +} + +export interface KafkaConsumeOptions { + topic: string; + groupId: string; + fromBeginning?: boolean; + handler: (payload: EachMessagePayload) => Promise; +} diff --git a/example/.env.template b/example/.env.template index 082c75d..d5291ba 100644 --- a/example/.env.template +++ b/example/.env.template @@ -8,3 +8,5 @@ S3_BUCKET_NAME= OPENAI_API_KEY= REDIS_URL= SQL_CONNECTION_STRING= +KAFKA_BROKERS=localhost:19092 +KAFKA_CLIENT_ID=dtk-client diff --git a/example/GUIDE.md b/example/GUIDE.md index f7f54f0..43165ec 100644 --- a/example/GUIDE.md +++ b/example/GUIDE.md @@ -531,6 +531,71 @@ Supported clients: `pg` (PostgreSQL), `mysql2` (MySQL / MariaDB), `mssql` (SQL S --- +### kafka + +```bash +dtk add kafka +``` + +Requires env vars: + +``` +KAFKA_BROKERS=localhost:19092 +KAFKA_CLIENT_ID=dtk-client +``` + +For local dev, start Redpanda from `tools/kafka/`: + +```bash +cd tools/kafka && docker compose up -d +``` + +Example usage: + +```ts +await suite() + .kafka({ + brokers: process.env.KAFKA_BROKERS!.split(",").map((b) => b.trim()), + clientId: process.env.KAFKA_CLIENT_ID ?? "dtk-client", + }) + .step("produce-message", async (ctx) => { + await ctx.services.kafka.produce({ + topic: "example-topic", + messages: [{ value: "hello from dtk" }], + }); + }) + .step("consume-message", async (ctx) => { + await ctx.services.kafka.consume({ + topic: "example-topic", + groupId: "dtk-group", + fromBeginning: true, + handler: async ({ message }) => { + console.log(message.value?.toString()); + }, + }); + }) + .step("disconnect", async (ctx) => { + await ctx.services.kafka.disconnect(); + }) + .run("stopOnError"); +``` + +Available methods on `ctx.services.kafka`: + +| Method | Description | +|---|---| +| `produce({ topic, messages })` | Sends messages to a topic. Connects the producer on first call and reuses it. | +| `consume({ topic, groupId, fromBeginning?, handler })` | Subscribes and runs the consumer. Calling this a second time without first calling `disconnect()` throws. | +| `disconnect()` | Disconnects producer and consumer and resets state. Safe to call even if neither was connected. | + +Run the example runbook: + +```bash +npm run runbook:kafka +``` + +--- + ## Writing a custom service If there is no plugin for the service you need, wire one in manually. Four files are involved. diff --git a/example/package-lock.json b/example/package-lock.json index 121bfd4..a678fd7 100644 --- a/example/package-lock.json +++ b/example/package-lock.json @@ -16,6 +16,7 @@ "@aws-sdk/util-dynamodb": "^3.996.4", "axios": "^1.17.0", "dotenv": "^17.0.0", + "kafkajs": "^2.2.4", "knex": "^3.2.10", "mssql": "^12.5.5", "mysql2": "^3.22.5", @@ -2893,9 +2894,6 @@ "arm64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2910,9 +2908,6 @@ "arm64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -2927,9 +2922,6 @@ "loong64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2944,9 +2936,6 @@ "loong64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -2961,9 +2950,6 @@ "ppc64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2978,9 +2964,6 @@ "riscv64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2995,9 +2978,6 @@ "riscv64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -3012,9 +2992,6 @@ "s390x" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -3029,9 +3006,6 @@ "x64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -3046,9 +3020,6 @@ "x64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -5708,6 +5679,15 @@ "safe-buffer": "^5.0.1" } }, + "node_modules/kafkajs": { + "version": "2.2.4", + "resolved": "https://registry.npmjs.org/kafkajs/-/kafkajs-2.2.4.tgz", + "integrity": "sha512-j/YeapB1vfPT2iOIUn/vxdyKEuhuY2PxMBvf5JWux6iSaukAccrMtXEY/Lb7OvavDhOWME589bpLrEdnVHjfjA==", + "license": "MIT", + "engines": { + "node": ">=14.0.0" + } + }, "node_modules/knex": { "version": "3.2.10", "resolved": "https://registry.npmjs.org/knex/-/knex-3.2.10.tgz", diff --git a/example/package.json b/example/package.json index 2d3b253..571384e 100644 --- a/example/package.json +++ b/example/package.json @@ -11,7 +11,8 @@ "runbook:aws-s3": "tsx src/runbooks/aws-s3.ts", "runbook:open-ai": "tsx src/runbooks/open-ai.ts", "runbook:redis": "tsx src/runbooks/redis.ts", - "runbook:sql": "tsx src/runbooks/sql.ts" + "runbook:sql": "tsx src/runbooks/sql.ts", + "runbook:kafka": "tsx src/runbooks/kafka.ts" }, "dependencies": { "@aws-sdk/client-dynamodb": "^3.1068.0", @@ -27,7 +28,8 @@ "mysql2": "^3.22.5", "pg": "^8.21.0", "openai": "^6.0.0", - "redis": "^6.0.0" + "redis": "^6.0.0", + "kafkajs": "^2.2.4" }, "devDependencies": { "@types/jest": "^30.0.0", diff --git a/example/src/runbooks/kafka.ts b/example/src/runbooks/kafka.ts new file mode 100644 index 0000000..8e9029e --- /dev/null +++ b/example/src/runbooks/kafka.ts @@ -0,0 +1,36 @@ +import "../load-env.js"; +import { suite } from "../suite.js"; + +await suite() + .kafka({ + brokers: process.env.KAFKA_BROKERS!.split(",").map((b) => b.trim()).filter(Boolean), + clientId: process.env.KAFKA_CLIENT_ID ?? "dtk-client", + }) + .step("produce-message", async (ctx) => { + await ctx.services.kafka.produce({ + topic: "example-topic", + messages: [{ value: "hello from dtk" }], + }); + console.log("message produced to example-topic"); + }) + .step("consume-message", async (ctx) => { + await ctx.services.kafka.consume({ + topic: "example-topic", + groupId: "dtk-group", + fromBeginning: true, + handler: async (payload) => { + console.log( + "received:", + payload.message.value?.toString(), + "partition:", + payload.partition, + "offset:", + payload.message.offset + ); + }, + }); + }) + .step("disconnect", async (ctx) => { + await ctx.services.kafka.disconnect(); + }) + .run("stopOnError"); diff --git a/example/src/services/kafka.test.ts b/example/src/services/kafka.test.ts new file mode 100644 index 0000000..9b8e36f --- /dev/null +++ b/example/src/services/kafka.test.ts @@ -0,0 +1,131 @@ +import { createKafkaService } from './kafka.js'; + +jest.mock('kafkajs'); +import { Kafka } from 'kafkajs'; + +const mockProducer = { + connect: jest.fn(), + send: jest.fn(), + disconnect: jest.fn(), +}; + +const mockConsumer = { + connect: jest.fn(), + subscribe: jest.fn(), + run: jest.fn(), + disconnect: jest.fn(), +}; + +beforeEach(() => { + jest.clearAllMocks(); + mockProducer.connect.mockResolvedValue(undefined); + mockProducer.send.mockResolvedValue(undefined); + mockProducer.disconnect.mockResolvedValue(undefined); + mockConsumer.connect.mockResolvedValue(undefined); + mockConsumer.subscribe.mockResolvedValue(undefined); + mockConsumer.run.mockResolvedValue(undefined); + mockConsumer.disconnect.mockResolvedValue(undefined); + (Kafka as jest.Mock).mockImplementation(() => ({ + producer: () => mockProducer, + consumer: () => mockConsumer, + })); +}); + +describe('createKafkaService', () => { + const config = { + brokers: ['localhost:9092'], + clientId: 'test-client', + }; + + describe('produce', () => { + it('connects the producer and sends a message', async () => { + const kafka = createKafkaService(config); + await kafka.produce({ topic: 'test-topic', messages: [{ value: 'hello' }] }); + expect(mockProducer.connect).toHaveBeenCalledTimes(1); + expect(mockProducer.send).toHaveBeenCalledWith({ + topic: 'test-topic', + messages: [{ value: 'hello' }], + }); + }); + + it('reuses the producer on subsequent produce calls', async () => { + const kafka = createKafkaService(config); + await kafka.produce({ topic: 'test-topic', messages: [{ value: 'first' }] }); + await kafka.produce({ topic: 'test-topic', messages: [{ value: 'second' }] }); + expect(mockProducer.connect).toHaveBeenCalledTimes(1); + expect(mockProducer.send).toHaveBeenCalledTimes(2); + }); + + it('throws when called without config', async () => { + const kafka = createKafkaService(); + await expect(kafka.produce({ topic: 'test-topic', messages: [] })).rejects.toThrow( + 'kafka service is not configured' + ); + }); + }); + + describe('consume', () => { + it('connects the consumer, subscribes, and runs with the provided handler', async () => { + const handler = jest.fn(); + const kafka = createKafkaService(config); + await kafka.consume({ topic: 'test-topic', groupId: 'test-group', handler }); + expect(mockConsumer.connect).toHaveBeenCalledTimes(1); + expect(mockConsumer.subscribe).toHaveBeenCalledWith({ + topic: 'test-topic', + fromBeginning: false, + }); + expect(mockConsumer.run).toHaveBeenCalledWith({ eachMessage: handler }); + }); + + it('subscribes fromBeginning when the option is true', async () => { + const kafka = createKafkaService(config); + await kafka.consume({ topic: 'test-topic', groupId: 'test-group', fromBeginning: true, handler: jest.fn() }); + expect(mockConsumer.subscribe).toHaveBeenCalledWith({ + topic: 'test-topic', + fromBeginning: true, + }); + }); + + it('throws if consume is called a second time while the consumer is running', async () => { + const kafka = createKafkaService(config); + await kafka.consume({ topic: 'test-topic', groupId: 'test-group', handler: jest.fn() }); + await expect( + kafka.consume({ topic: 'other-topic', groupId: 'test-group', handler: jest.fn() }) + ).rejects.toThrow('kafka consumer is already running'); + }); + + it('throws when called without config', async () => { + const kafka = createKafkaService(); + await expect( + kafka.consume({ topic: 'test-topic', groupId: 'test-group', handler: jest.fn() }) + ).rejects.toThrow('kafka service is not configured'); + }); + }); + + describe('disconnect', () => { + it('disconnects both producer and consumer', async () => { + const kafka = createKafkaService(config); + await kafka.produce({ topic: 'test-topic', messages: [{ value: 'hi' }] }); + await kafka.consume({ topic: 'test-topic', groupId: 'test-group', handler: jest.fn() }); + await kafka.disconnect(); + expect(mockProducer.disconnect).toHaveBeenCalledTimes(1); + expect(mockConsumer.disconnect).toHaveBeenCalledTimes(1); + }); + + it('is a no-op when neither producer nor consumer was created', async () => { + const kafka = createKafkaService(config); + await expect(kafka.disconnect()).resolves.toBeUndefined(); + expect(mockProducer.disconnect).not.toHaveBeenCalled(); + expect(mockConsumer.disconnect).not.toHaveBeenCalled(); + }); + + it('resets state so consume can be called again after disconnect', async () => { + const kafka = createKafkaService(config); + await kafka.consume({ topic: 'test-topic', groupId: 'test-group', handler: jest.fn() }); + await kafka.disconnect(); + await expect( + kafka.consume({ topic: 'test-topic', groupId: 'test-group', handler: jest.fn() }) + ).resolves.toBeUndefined(); + }); + }); +}); diff --git a/example/src/services/kafka.ts b/example/src/services/kafka.ts new file mode 100644 index 0000000..80e2046 --- /dev/null +++ b/example/src/services/kafka.ts @@ -0,0 +1,64 @@ +import { Kafka } from 'kafkajs'; +import type { Producer, Consumer } from 'kafkajs'; +import type { KafkaConfig, KafkaProduceOptions, KafkaConsumeOptions } from '../types/kafka.js'; + +export function createKafkaService(config?: KafkaConfig) { + const ensureConfig = () => { + if (!config) throw new Error('kafka service is not configured -- call .kafka(config) on the suite'); + }; + + let producer: Producer | null = null; + let consumer: Consumer | null = null; + let consumerStarted = false; + + const getProducer = async (): Promise => { + ensureConfig(); + if (!producer) { + const kafka = new Kafka({ brokers: config!.brokers, clientId: config!.clientId, ssl: config!.ssl, sasl: config!.sasl }); + producer = kafka.producer(); + await producer.connect(); + } + return producer; + }; + + const getConsumer = async (groupId: string): Promise => { + ensureConfig(); + if (!consumer) { + const kafka = new Kafka({ brokers: config!.brokers, clientId: config!.clientId, ssl: config!.ssl, sasl: config!.sasl }); + consumer = kafka.consumer({ groupId }); + await consumer.connect(); + } + return consumer; + }; + + return { + produce: async (options: KafkaProduceOptions): Promise => { + const p = await getProducer(); + await p.send({ topic: options.topic, messages: options.messages }); + }, + + consume: async (options: KafkaConsumeOptions): Promise => { + if (consumerStarted) { + throw new Error( + 'kafka consumer is already running -- call disconnect() before consuming again' + ); + } + const c = await getConsumer(options.groupId); + await c.subscribe({ topic: options.topic, fromBeginning: options.fromBeginning ?? false }); + await c.run({ eachMessage: options.handler }); + consumerStarted = true; + }, + + disconnect: async (): Promise => { + if (producer) { + await producer.disconnect(); + producer = null; + } + if (consumer) { + await consumer.disconnect(); + consumer = null; + } + consumerStarted = false; + }, + }; +} diff --git a/example/src/suite.ts b/example/src/suite.ts index 5b8d259..62a37a2 100644 --- a/example/src/suite.ts +++ b/example/src/suite.ts @@ -18,6 +18,8 @@ import { createRedisService } from "./services/redis.js"; import type { RedisConfig } from "./types/redis.js"; import { createSqlService } from "./services/sql.js"; import type { SqlConfig } from "./types/sql.js"; +import { createKafkaService } from "./services/kafka.js"; +import type { KafkaConfig } from "./types/kafka.js"; // dtk:imports import type { OAuthConfig, BasicAuthConfig, BearerTokenConfig, StepContext, StepFn, Step, SuiteRunOption } from "./types/suite.js"; @@ -36,6 +38,7 @@ class Suite { private openAiConfig?: OpenAiConfig; private redisConfig?: RedisConfig; private sqlConfig?: SqlConfig; + private kafkaConfig?: KafkaConfig; // dtk:configs oauth(config: OAuthConfig): this { this.oauthConfig = config; return this; } @@ -48,6 +51,7 @@ class Suite { openAi(config: OpenAiConfig): this { this.openAiConfig = config; return this; } redis(config: RedisConfig): this { this.redisConfig = config; return this; } sql(config: SqlConfig): this { this.sqlConfig = config; return this; } + kafka(config: KafkaConfig): this { this.kafkaConfig = config; return this; } // dtk:methods step(name: string, fn: StepFn): this { @@ -92,6 +96,7 @@ class Suite { openAi: createOpenAIService(this.openAiConfig), redis: createRedisService(this.redisConfig), sql: createSqlService(this.sqlConfig), + kafka: createKafkaService(this.kafkaConfig), // dtk:services }, }; diff --git a/example/src/types/kafka.ts b/example/src/types/kafka.ts new file mode 100644 index 0000000..344b9a6 --- /dev/null +++ b/example/src/types/kafka.ts @@ -0,0 +1,20 @@ +import type { Message, EachMessagePayload, KafkaConfig as KafkaJSConfig } from 'kafkajs'; + +export interface KafkaConfig { + brokers: string[]; + clientId?: string; + ssl?: KafkaJSConfig['ssl']; + sasl?: KafkaJSConfig['sasl']; +} + +export interface KafkaProduceOptions { + topic: string; + messages: Message[]; +} + +export interface KafkaConsumeOptions { + topic: string; + groupId: string; + fromBeginning?: boolean; + handler: (payload: EachMessagePayload) => Promise; +} diff --git a/example/src/types/suite.ts b/example/src/types/suite.ts index 557a7f5..8bb6277 100644 --- a/example/src/types/suite.ts +++ b/example/src/types/suite.ts @@ -8,6 +8,7 @@ import type { OpenAiConfig } from "./open-ai.js"; import type { Model } from "openai"; import type { Response as OpenAiApiResponse } from "openai/resources/responses/responses"; import type { SqlOps } from "./sql.js"; +import type { KafkaConsumeOptions } from "./kafka.js"; // dtk:type-imports export type { HttpOptions }; @@ -51,6 +52,7 @@ export interface StepContext { openAi: { listModels(): Promise<{ data: Model[] }>; response(model: string, format: string, message: string): Promise; }; redis: { get(key: string): Promise; set(key: string, value: string, ttlSeconds?: number): Promise; del(key: string): Promise; exists(key: string): Promise; expire(key: string, ttlSeconds: number): Promise; hset(key: string, field: string, value: string): Promise; hget(key: string, field: string): Promise; keys(pattern: string): Promise; quit(): Promise; }; sql: { query>(sql: string, params?: unknown[]): Promise; execute(sql: string, params?: unknown[]): Promise; callProc>(name: string, params?: unknown[]): Promise; transaction(fn: (ops: SqlOps) => Promise): Promise; disconnect(): Promise; }; + kafka: { produce(options: { topic: string; messages: import('kafkajs').Message[] }): Promise; consume(options: KafkaConsumeOptions): Promise; disconnect(): Promise; }; // dtk:service-types }; } diff --git a/tools/README.md b/tools/README.md index 6e6b928..b2c2338 100644 --- a/tools/README.md +++ b/tools/README.md @@ -50,6 +50,24 @@ docker compose up -d ``` +### Kafka (`kafka/`) + +Runs [Redpanda](https://redpanda.com/) — a Kafka-API-compatible broker — on port `19092`, plus Redpanda Console (UI) on port `8080`. + +```bash +cd tools/kafka +docker compose up -d +``` + +Set the following in your project's `.env`: + +``` +KAFKA_BROKERS=localhost:19092 +KAFKA_CLIENT_ID=dtk-client +``` + +Redpanda Console is available at http://localhost:8080 and lets you inspect topics, consumer groups, and messages. + ### AWS (`aws/`) No docker files. These include helper files to demo and test the runbooks. diff --git a/tools/kafka/docker-compose.yml b/tools/kafka/docker-compose.yml new file mode 100644 index 0000000..9a1bb39 --- /dev/null +++ b/tools/kafka/docker-compose.yml @@ -0,0 +1,33 @@ +services: + redpanda: + image: redpandadata/redpanda:latest + container_name: dtk-redpanda + command: + - redpanda + - start + - --kafka-addr internal://0.0.0.0:9092,external://0.0.0.0:19092 + - --advertise-kafka-addr internal://redpanda:9092,external://localhost:19092 + - --pandaproxy-addr internal://0.0.0.0:8082,external://0.0.0.0:18082 + - --advertise-pandaproxy-addr internal://redpanda:8082,external://localhost:18082 + - --schema-registry-addr internal://0.0.0.0:8081,external://0.0.0.0:18081 + - --rpc-addr redpanda:33145 + - --advertise-rpc-addr redpanda:33145 + - --smp 1 + - --memory 1G + - --mode dev-container + - --default-log-level=warn + ports: + - "19092:19092" + - "18082:18082" + - "18081:18081" + - "9644:9644" + + redpanda-console: + image: redpandadata/console:latest + container_name: dtk-redpanda-console + depends_on: + - redpanda + ports: + - "8080:8080" + environment: + - KAFKA_BROKERS=redpanda:9092 From 22e7b041737b7ffb8725a7282ee2073e3145af53 Mon Sep 17 00:00:00 2001 From: priyamkarn Date: Fri, 19 Jun 2026 04:36:21 +0530 Subject: [PATCH 2/2] fix: restore mongodb, move kafka docs to README/GUIDE/tools --- dtk/README.md | 63 ++++++++++++++ dtk/templates/init/GUIDE.md | 65 ++++++++++++++ dtk/templates/plugins/kafka/README.md | 120 -------------------------- example/.env.template | 2 + example/package.json | 3 +- example/src/suite.ts | 5 ++ example/src/types/suite.ts | 2 + tools/README.md | 31 +++++++ 8 files changed, 170 insertions(+), 121 deletions(-) delete mode 100644 dtk/templates/plugins/kafka/README.md diff --git a/dtk/README.md b/dtk/README.md index d62dd87..e10b10b 100644 --- a/dtk/README.md +++ b/dtk/README.md @@ -512,6 +512,69 @@ Available methods on the mongodb service: --- +### kafka + +Produce and consume messages on a Kafka topic via [KafkaJS](https://kafka.js.org/). + +```bash +dtk add kafka +``` + +Env vars appended to `.env.template`: + +``` +KAFKA_BROKERS=localhost:9092 +KAFKA_CLIENT_ID=dtk-client +``` + +For local dev, start [Redpanda](https://redpanda.com/) from `tools/kafka/`: + +```bash +cd tools/kafka && docker compose up -d +# then set KAFKA_BROKERS=localhost:19092 +``` + +```ts +import "../load-env.js"; +import { suite } from "../suite.js"; + +await suite() + .kafka({ + brokers: process.env.KAFKA_BROKERS!.split(",").map((b) => b.trim()).filter(Boolean), + clientId: process.env.KAFKA_CLIENT_ID ?? "dtk-client", + }) + .step("produce-message", async (ctx) => { + await ctx.services.kafka.produce({ + topic: "example-topic", + messages: [{ value: "hello from dtk" }], + }); + }) + .step("consume-message", async (ctx) => { + await ctx.services.kafka.consume({ + topic: "example-topic", + groupId: "dtk-group", + fromBeginning: true, + handler: async ({ message }) => { + console.log(message.value?.toString()); + }, + }); + }) + .step("disconnect", async (ctx) => { + await ctx.services.kafka.disconnect(); + }) + .run("stopOnError"); +``` + +Available methods on `ctx.services.kafka`: + +| Method | Description | +|---|---| +| `produce({ topic, messages })` | Sends messages to a topic. Connects the producer on first call and reuses it. | +| `consume({ topic, groupId, fromBeginning?, handler })` | Subscribes and runs the consumer. Throws if called a second time without `disconnect()`. | +| `disconnect()` | Disconnects producer and consumer and resets state. Safe to call even if neither was connected. | + +--- + ## Writing runbooks A runbook is a TypeScript file that uses the `suite()` builder to chain steps and run them in sequence. diff --git a/dtk/templates/init/GUIDE.md b/dtk/templates/init/GUIDE.md index f7f54f0..4fb4ff2 100644 --- a/dtk/templates/init/GUIDE.md +++ b/dtk/templates/init/GUIDE.md @@ -531,6 +531,71 @@ Supported clients: `pg` (PostgreSQL), `mysql2` (MySQL / MariaDB), `mssql` (SQL S --- +### kafka + +```bash +dtk add kafka +``` + +Requires env vars: + +``` +KAFKA_BROKERS=localhost:19092 +KAFKA_CLIENT_ID=dtk-client +``` + +For local dev, start Redpanda from `tools/kafka/`: + +```bash +cd tools/kafka && docker compose up -d +``` + +Example usage: + +```ts +await suite() + .kafka({ + brokers: process.env.KAFKA_BROKERS!.split(",").map((b) => b.trim()), + clientId: process.env.KAFKA_CLIENT_ID ?? "dtk-client", + }) + .step("produce-message", async (ctx) => { + await ctx.services.kafka.produce({ + topic: "example-topic", + messages: [{ value: "hello from dtk" }], + }); + }) + .step("consume-message", async (ctx) => { + await ctx.services.kafka.consume({ + topic: "example-topic", + groupId: "dtk-group", + fromBeginning: true, + handler: async ({ message }) => { + console.log(message.value?.toString()); + }, + }); + }) + .step("disconnect", async (ctx) => { + await ctx.services.kafka.disconnect(); + }) + .run("stopOnError"); +``` + +Available methods on `ctx.services.kafka`: + +| Method | Description | +|---|---| +| `produce({ topic, messages })` | Sends messages to a topic. Connects the producer on first call and reuses it. | +| `consume({ topic, groupId, fromBeginning?, handler })` | Subscribes and runs the consumer. Calling this a second time without `disconnect()` throws. | +| `disconnect()` | Disconnects producer and consumer and resets state. Safe to call even if neither was connected. | + +Run the example runbook: + +```bash +npm run runbook:kafka +``` + +--- + ## Writing a custom service If there is no plugin for the service you need, wire one in manually. Four files are involved. diff --git a/dtk/templates/plugins/kafka/README.md b/dtk/templates/plugins/kafka/README.md deleted file mode 100644 index 0468db0..0000000 --- a/dtk/templates/plugins/kafka/README.md +++ /dev/null @@ -1,120 +0,0 @@ -# kafka plugin - -Adds Kafka support to your dtk suite via [KafkaJS](https://kafka.js.org/). - -Provides three methods on `ctx.services.kafka`: **produce**, **consume**, and **disconnect**. - ---- - -## Installation - -```bash -dtk add kafka -``` - -This copies `src/services/kafka.ts` and `src/types/kafka.ts` into your project, patches `suite.ts` and `types/suite.ts`, and installs `kafkajs`. - ---- - -## Configuration - -Add the following to your `.env` (the template is added automatically): - -``` -KAFKA_BROKERS=localhost:9092 -KAFKA_CLIENT_ID=dtk-client -``` - -For multiple brokers, comma-separate them: - -``` -KAFKA_BROKERS=broker1:9092,broker2:9092,broker3:9092 -``` - ---- - -## Local development - -A `docker-compose.yml` using [Redpanda](https://redpanda.com/) is included for local dev — Redpanda is fully Kafka-API-compatible and starts in seconds. - -```bash -# from dtk/templates/plugins/kafka/ -docker compose up -d - -# connect on localhost:19092 -KAFKA_BROKERS=localhost:19092 -``` - -Redpanda Console (UI) is available at http://localhost:8080. - ---- - -## Usage - -```ts -import "../load-env.js"; -import { suite } from "../suite.js"; - -await suite() - .kafka({ - brokers: process.env.KAFKA_BROKERS!.split(",").map((b) => b.trim()), - clientId: process.env.KAFKA_CLIENT_ID ?? "dtk-client", - }) - .step("produce-message", async (ctx) => { - await ctx.services.kafka.produce({ - topic: "my-topic", - messages: [{ value: "hello" }], - }); - }) - .step("consume-message", async (ctx) => { - await ctx.services.kafka.consume({ - topic: "my-topic", - groupId: "my-group", - fromBeginning: true, - handler: async ({ message }) => { - console.log(message.value?.toString()); - }, - }); - }) - .step("disconnect", async (ctx) => { - await ctx.services.kafka.disconnect(); - }) - .run("stopOnError"); -``` - ---- - -## API - -### `.kafka(config)` - -| Field | Type | Required | Description | -|---|---|---|---| -| `brokers` | `string[]` | ✅ | List of broker addresses | -| `clientId` | `string` | | Kafka client identifier | -| `ssl` | `boolean` | | Enable TLS | -| `sasl` | `object` | | SASL credentials (`mechanism`, `username`, `password`) | - -### `ctx.services.kafka.produce(options)` - -Sends messages to a topic. Connects the producer on first call and reuses it for subsequent calls. - -| Field | Type | Description | -|---|---|---| -| `topic` | `string` | Target topic | -| `messages` | `Message[]` | KafkaJS `Message` array | - -### `ctx.services.kafka.consume(options)` - -Subscribes to a topic and runs the consumer. Calling `consume()` a second time without calling `disconnect()` first will throw — KafkaJS does not allow re-subscribing a running consumer. - -| Field | Type | Description | -|---|---|---| -| `topic` | `string` | Topic to subscribe to | -| `groupId` | `string` | Consumer group ID | -| `fromBeginning` | `boolean` | Read from the earliest offset (default: `false`) | -| `handler` | `(payload: EachMessagePayload) => Promise` | Called for each message | - -### `ctx.services.kafka.disconnect()` - -Disconnects both producer and consumer and resets internal state. Safe to call even if neither was connected. diff --git a/example/.env.template b/example/.env.template index d5291ba..d833c19 100644 --- a/example/.env.template +++ b/example/.env.template @@ -8,5 +8,7 @@ S3_BUCKET_NAME= OPENAI_API_KEY= REDIS_URL= SQL_CONNECTION_STRING= +MONGODB_URI= +MONGODB_DATABASE= KAFKA_BROKERS=localhost:19092 KAFKA_CLIENT_ID=dtk-client diff --git a/example/package.json b/example/package.json index b73ddf1..1244f59 100644 --- a/example/package.json +++ b/example/package.json @@ -12,7 +12,8 @@ "runbook:open-ai": "tsx src/runbooks/open-ai.ts", "runbook:redis": "tsx src/runbooks/redis.ts", "runbook:sql": "tsx src/runbooks/sql.ts", - "runbook:kafka": "tsx src/runbooks/kafka.ts" + "runbook:kafka": "tsx src/runbooks/kafka.ts", + "runbook:mongodb": "tsx src/runbooks/mongodb.ts" }, "dependencies": { "@aws-sdk/client-dynamodb": "^3.1068.0", diff --git a/example/src/suite.ts b/example/src/suite.ts index 62a37a2..292b087 100644 --- a/example/src/suite.ts +++ b/example/src/suite.ts @@ -18,6 +18,8 @@ import { createRedisService } from "./services/redis.js"; import type { RedisConfig } from "./types/redis.js"; import { createSqlService } from "./services/sql.js"; import type { SqlConfig } from "./types/sql.js"; +import { createMongoService } from "./services/mongodb.js"; +import type { MongoConfig } from "./types/mongodb.js"; import { createKafkaService } from "./services/kafka.js"; import type { KafkaConfig } from "./types/kafka.js"; // dtk:imports @@ -38,6 +40,7 @@ class Suite { private openAiConfig?: OpenAiConfig; private redisConfig?: RedisConfig; private sqlConfig?: SqlConfig; + private mongodbConfig?: MongoConfig; private kafkaConfig?: KafkaConfig; // dtk:configs @@ -51,6 +54,7 @@ class Suite { openAi(config: OpenAiConfig): this { this.openAiConfig = config; return this; } redis(config: RedisConfig): this { this.redisConfig = config; return this; } sql(config: SqlConfig): this { this.sqlConfig = config; return this; } + mongodb(config: MongoConfig): this { this.mongodbConfig = config; return this; } kafka(config: KafkaConfig): this { this.kafkaConfig = config; return this; } // dtk:methods @@ -96,6 +100,7 @@ class Suite { openAi: createOpenAIService(this.openAiConfig), redis: createRedisService(this.redisConfig), sql: createSqlService(this.sqlConfig), + mongodb: createMongoService(this.mongodbConfig), kafka: createKafkaService(this.kafkaConfig), // dtk:services }, diff --git a/example/src/types/suite.ts b/example/src/types/suite.ts index 8bb6277..314095a 100644 --- a/example/src/types/suite.ts +++ b/example/src/types/suite.ts @@ -8,6 +8,7 @@ import type { OpenAiConfig } from "./open-ai.js"; import type { Model } from "openai"; import type { Response as OpenAiApiResponse } from "openai/resources/responses/responses"; import type { SqlOps } from "./sql.js"; +import type { MongoDocument, MongoFilter, MongoUpdate } from "./mongodb.js"; import type { KafkaConsumeOptions } from "./kafka.js"; // dtk:type-imports @@ -52,6 +53,7 @@ export interface StepContext { openAi: { listModels(): Promise<{ data: Model[] }>; response(model: string, format: string, message: string): Promise; }; redis: { get(key: string): Promise; set(key: string, value: string, ttlSeconds?: number): Promise; del(key: string): Promise; exists(key: string): Promise; expire(key: string, ttlSeconds: number): Promise; hset(key: string, field: string, value: string): Promise; hget(key: string, field: string): Promise; keys(pattern: string): Promise; quit(): Promise; }; sql: { query>(sql: string, params?: unknown[]): Promise; execute(sql: string, params?: unknown[]): Promise; callProc>(name: string, params?: unknown[]): Promise; transaction(fn: (ops: SqlOps) => Promise): Promise; disconnect(): Promise; }; + mongodb: { insertOne(collection: string, doc: MongoDocument): Promise<{ insertedId: string }>; insertMany(collection: string, docs: MongoDocument[]): Promise<{ insertedCount: number; insertedIds: string[] }>; findOne(collection: string, filter: MongoFilter): Promise; find(collection: string, filter?: MongoFilter): Promise; updateOne(collection: string, filter: MongoFilter, update: MongoUpdate): Promise<{ matchedCount: number; modifiedCount: number }>; updateMany(collection: string, filter: MongoFilter, update: MongoUpdate): Promise<{ matchedCount: number; modifiedCount: number }>; deleteOne(collection: string, filter: MongoFilter): Promise<{ deletedCount: number }>; deleteMany(collection: string, filter: MongoFilter): Promise<{ deletedCount: number }>; disconnect(): Promise; }; kafka: { produce(options: { topic: string; messages: import('kafkajs').Message[] }): Promise; consume(options: KafkaConsumeOptions): Promise; disconnect(): Promise; }; // dtk:service-types }; diff --git a/tools/README.md b/tools/README.md index b2c2338..1ac558a 100644 --- a/tools/README.md +++ b/tools/README.md @@ -50,6 +50,37 @@ docker compose up -d ``` +### MongoDB (`mongodb/`) + +Runs a MongoDB 6.0 (LTS) instance on the default port `27017`. + +```bash +cd tools/mongodb +docker compose up -d +``` + +Set the following in your project's `.env`: + +``` +MONGODB_URI=mongodb://dtk:dtk@localhost:27017/dtk?authSource=admin +MONGODB_DATABASE=dtk +``` + +Credentials: user `dtk`, password `dtk`, database `dtk`. + +On first start, the init script in `mongodb/init/` runs automatically and seeds: + +- `users` collection with three sample documents +- `accounts` collection with two seeded documents + +These scripts only run against a fresh data volume. To reset and re-run them: + +```bash +docker compose down -v +docker compose up -d +``` + + ### Kafka (`kafka/`) Runs [Redpanda](https://redpanda.com/) — a Kafka-API-compatible broker — on port `19092`, plus Redpanda Console (UI) on port `8080`.