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/cli/add.ts b/dtk/cli/add.ts index 6b7d3d4..1dc2f37 100644 --- a/dtk/cli/add.ts +++ b/dtk/cli/add.ts @@ -16,7 +16,7 @@ const PLUGIN_MAP: Record = { 'open-ai': 'open-ai', 'redis': 'redis', 'sql': 'sql', - 'mongodb': 'mongodb', + 'kafka': 'kafka', }; interface PluginTransform { diff --git a/dtk/package-lock.json b/dtk/package-lock.json index b61abcb..963ff84 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", "mongodb": "^6.17.0", "openai": "^6.42.0", @@ -2705,9 +2706,6 @@ "arm64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2722,9 +2720,6 @@ "arm64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -2739,9 +2734,6 @@ "loong64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2756,9 +2748,6 @@ "loong64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -2773,9 +2762,6 @@ "ppc64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2790,9 +2776,6 @@ "riscv64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2807,9 +2790,6 @@ "riscv64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -2824,9 +2804,6 @@ "s390x" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2841,9 +2818,6 @@ "x64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2858,9 +2832,6 @@ "x64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -5160,6 +5131,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 4ca38e9..2aa2cfc 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", "mongodb": "^6.17.0", "openai": "^6.42.0", 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/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 b914e45..d833c19 100644 --- a/example/.env.template +++ b/example/.env.template @@ -10,3 +10,5 @@ REDIS_URL= SQL_CONNECTION_STRING= MONGODB_URI= MONGODB_DATABASE= +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 56945c2..6aa4085 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", "mongodb": "^6.17.0", "mssql": "^12.5.5", @@ -2918,9 +2919,6 @@ "arm64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2935,9 +2933,6 @@ "arm64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -2952,9 +2947,6 @@ "loong64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -2969,9 +2961,6 @@ "loong64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -2986,9 +2975,6 @@ "ppc64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -3003,9 +2989,6 @@ "riscv64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -3020,9 +3003,6 @@ "riscv64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -3037,9 +3017,6 @@ "s390x" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -3054,9 +3031,6 @@ "x64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -3071,9 +3045,6 @@ "x64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -5742,6 +5713,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 88fe93e..1244f59 100644 --- a/example/package.json +++ b/example/package.json @@ -12,6 +12,7 @@ "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:mongodb": "tsx src/runbooks/mongodb.ts" }, "dependencies": { @@ -28,8 +29,8 @@ "mssql": "^12.5.5", "mysql2": "^3.22.5", "openai": "^6.0.0", - "pg": "^8.21.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 0acdbd0..292b087 100644 --- a/example/src/suite.ts +++ b/example/src/suite.ts @@ -20,6 +20,8 @@ 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 import type { OAuthConfig, BasicAuthConfig, BearerTokenConfig, StepContext, StepFn, Step, SuiteRunOption } from "./types/suite.js"; @@ -39,6 +41,7 @@ class Suite { private redisConfig?: RedisConfig; private sqlConfig?: SqlConfig; private mongodbConfig?: MongoConfig; + private kafkaConfig?: KafkaConfig; // dtk:configs oauth(config: OAuthConfig): this { this.oauthConfig = config; return this; } @@ -52,6 +55,7 @@ class Suite { 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 step(name: string, fn: StepFn): this { @@ -97,6 +101,7 @@ class Suite { redis: createRedisService(this.redisConfig), sql: createSqlService(this.sqlConfig), mongodb: createMongoService(this.mongodbConfig), + 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 b605a17..314095a 100644 --- a/example/src/types/suite.ts +++ b/example/src/types/suite.ts @@ -9,6 +9,7 @@ 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 export type { HttpOptions }; @@ -53,6 +54,7 @@ export interface StepContext { 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 c5c938f..1ac558a 100644 --- a/tools/README.md +++ b/tools/README.md @@ -80,6 +80,25 @@ 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`. + +```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