npm package: @tamasha/kafka-connection-v2
Description: Minimal KafkaJS connection helper with clean lifecycle management
Author: Tamasha Team
Latest version: 1.0.48

# @tamasha/kafka-connection-v2

Minimal KafkaJS helper that focuses on the essentials: connect quickly, enforce per-action schemas, and subscribe with clean teardown semantics. Inspired by the original `@tamasha/kafka-connection`, this version keeps only the primitives needed for simple services.

## Highlights

- Lightweight wrapper around `kafkajs` with lazy producer and consumer creation.
- Built-in topic schema registry keyed by action (one topic per microservice, many actions per topic).
- Generated TypeScript payload unions for every topic/action combination.
- Explicit lifecycle methods (`connect`, `disconnect`) to simplify graceful shutdown.
- No external dependencies beyond `kafkajs`.

## Installation

```bash
npm install @tamasha/kafka-connection-v2
```

## Quick start

```ts
import {
  KafkaConnectionV2,
  USER_EVENTS_TOPIC,
  UserEventMessage,
} from "@tamasha/kafka-connection-v2";

const connection = new KafkaConnectionV2({
  clientId: "example-service",
  brokers: ["localhost:9092"],
});

await connection.connect();

const loginEvent: UserEventMessage = {
  payload: {
    action: "login",
    userId: "u-1",
    timestamp: Date.now(),
    service: "auth-service",
    sessionId: "session-xyz",
  },
};

await connection.produce({
  topic: USER_EVENTS_TOPIC,
  messages: [
    {
      value: JSON.stringify(loginEvent),
      // key will automatically be set to loginEvent.payload.userId if omitted
    },
  ],
});

await connection.subscribe(
  { groupId: "example-service-consumer" },
  USER_EVENTS_TOPIC,
  async ({ message }) => {
    const payload = JSON.parse(message.value?.toString("utf-8") ?? "{}") as UserEventMessage;
    console.log("Received:", {
      key: message.key?.toString(),
      payload,
    });
  }
);

// Later, during shutdown
await connection.disconnect();
```

## API

### `new KafkaConnectionV2(config)`

Creates a new connection manager. `config` is forwarded to the underlying `Kafka` constructor, with `brokers` required. Built-in schemas are loaded automatically; use `getSchemaRegistry()` if you need introspection.

### `connect()`

Creates the internal Kafka clients (producer and consumer) lazily and verifies the brokers can be reached. Calling `produce` or `subscribe` will auto-connect if needed, but `connect` is useful for startup health checks.

### `produce(record)`

Sends a single `ProducerRecord`. Ensures the producer is connected before sending and validates each message payload against the configured topic schema.

### `subscribe(consumerConfig, topics, handler, options?)`

Initialises a consumer (one per connection) and starts consuming the provided topics. The handler receives the standard `EachMessagePayload`. Set `options.useBatch = true` to receive `EachBatchPayload` instead. Incoming messages are validated prior to invoking the handler.

### `disconnect()`

Stops and disconnects the managed consumer and producer.

## Built-in schemas & types

- `DEFAULT_TOPIC_SCHEMAS` exposes the runtime registry used for validation.
- `UserEventMessage`, `OrderEventMessage`, and `KafkaEventMessage` describe the allowed JSON payloads per topic/action.
- `connection.getSchemaRegistry()` returns the `TopicSchemaRegistry` instance, useful for debugging or listing available topics/actions.
- Partition keys derive automatically from each topic's schema (`payload.userId`, `payload.orderId`, etc.).

## License

MIT © Tamasha Team