fix: address kafka plugin review feedback

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.

dtk/cli/add.ts

@@ -16,7 +16,7 @@ const PLUGIN_MAP: Record<string, string> = {
   'open-ai': 'open-ai',
   'redis': 'redis',
   'sql': 'sql',
-  'mongodb': 'mongodb',
+  'kafka': 'kafka',
 };
 
 interface PluginTransform {

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",

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.

dtk/templates/plugins/kafka/env.txt

@@ -0,0 +1,2 @@
+KAFKA_BROKERS=localhost:9092
+KAFKA_CLIENT_ID=dtk-client

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");

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<void>; consume(options: KafkaConsumeOptions): Promise<void>; disconnect(): Promise<void>; };"
+    }
+  }
+}