.publish(message, opts)
  │
  ├─ 1. Validate message (topic.validate())
  ├─ 2. Check idempotency key (skip if duplicate)
  ├─ 3. Assign partition:
  │    ├─ partitionKey provided? → hash(key) % partitions
  │    └─ No key? → round-robin assignment
  ├─ 4. Assign monotonic offset within partition
  ├─ 5. Persist message record to Storage
  └─ 6. Notify Broker (wake subscriber poll loops)


.subscribe({ group, handler })
  │
  │  ┌──── Poll Loop (per group, per process) ──────────┐
  │  │                                                   │
  ├──┤ 1. Query Storage for messages after group offset  │
  │  │    (respects partition assignment, batchSize)     │
  │  │ 2. Apply subscription filter (if set)             │
  │  │ 3. Create delivery record for each message        │
  │  │ 4. Execute handler(ctx) per delivery              │
  │  │                                                   │
  │  │ On handler completion:                            │
  │  │  ├─ auto ack mode: ack on resolve, nack on reject │
  │  │  └─ manual mode: wait for ctx.ack() / ctx.nack()  │
  │  │                                                   │
  │  │ 5. Commit offset after successful ack             │
  │  │ 6. On nack: apply retry backoff                   │
  │  │    All retries exhausted? → Dead letter            │
  │  └───────────────────────────────────────────────────┘

Topic: "order-events" (16 partitions)
  │
  ├─ Group: "billing-service"    offset=142  (processing message 143)
  ├─ Group: "analytics-service"  offset=89   (behind, catching up)
  └─ Group: "shipping-service"   offset=142  (up to date)

import { OqronKit, pubsubModule } from "oqronkit";

await OqronKit.init({
  config: {
    project: "billing-platform",
    environment: "production",
    modules: [
      pubsubModule({
        concurrency: 10,
        ackTimeoutMs: 30_000,
        retries: {
          max: 5,
          strategy: "exponential",
          baseDelay: 1_000,
          maxDelay: 60_000,
        },
      }),
    ],
  },
});

import { topic } from "oqronkit";

type OrderEvent =
  | { type: "order.created"; orderId: string; accountId: string; total: number }
  | { type: "order.paid"; orderId: string; accountId: string; paymentId: string };

export const orderEvents = topic<OrderEvent>({
  name: "order-events",
  tags: ["orders", "billing"],

  retention: {
    maxAgeMs: 7 * 24 * 60 * 60 * 1000,
    maxCount: 100_000,
  },

  distribution: {
    partitions: 16,
    partitionKey: (message) => message.accountId,
  },

  validate: (message) =>
    message.orderId ? true : "orderId is required",
});

const messageId = await orderEvents.publish(
  {
    type: "order.created",
    orderId: "ord_123",
    accountId: "acct_1",
    total: 99.99,
  },
  {
    idempotencyKey: "order-created:ord_123",
    partitionKey: "acct_1",
    correlationId: "req_abc",
    headers: {
      source: "checkout",
    },
  },
);

await orderEvents.publishBatch([
  {
    message: {
      type: "order.created",
      orderId: "ord_124",
      accountId: "acct_1",
      total: 149.5,
    },
    options: { idempotencyKey: "order-created:ord_124" },
  },
  {
    message: {
      type: "order.paid",
      orderId: "ord_124",
      accountId: "acct_1",
      paymentId: "pay_456",
    },
    options: { idempotencyKey: "order-paid:ord_124" },
  },
]);

import { orderEvents } from "./order-events";

await orderEvents.subscribe({
  group: "billing-service",
  concurrency: 8,
  batchSize: 50,
  maxInFlight: 200,
  startFrom: "earliest",

  filter: (event) => event.type === "order.paid",

  handler: async (ctx) => {
    await createInvoice(ctx.message.orderId);
    ctx.log.info(`invoice created for ${ctx.message.orderId}`);
  },
});

import { orderEvents } from "./order-events";

await orderEvents.subscribe({
  group: "analytics-service",
  concurrency: 20,

  handler: async (ctx) => {
    await trackEvent(ctx.message);
  },
});

await orderEvents.subscribe({
  group: "warehouse-service",
  ackMode: "manual",
  ackTimeoutMs: 60_000,

  handler: async (ctx) => {
    try {
      await reserveInventory(ctx.message);
      await ctx.ack();
    } catch (err) {
      await ctx.nack(err instanceof Error ? err.message : "reservation failed");
    }
  },
});

const accountEvents = topic<AccountEvent>({
  name: "account-events",
  distribution: {
    partitions: 32,
    partitionKey: (event) => event.accountId,
  },
});

await accountEvents.publish(event, {
  partitionKey: event.accountId,
});

await orderEvents.subscribe({
  group: "shipping-service",
  retries: {
    max: 3,
    strategy: "exponential",
    baseDelay: 2_000,
    maxDelay: 30_000,
  },
  deadLetter: {
    enabled: true,
    onDead: async (messageId, message, error) => {
      await alertOps({ messageId, message, error });
    },
  },
  handler: async (ctx) => {
    await scheduleShipment(ctx.message);
  },
});

const dead = await orderEvents.deadLetters("shipping-service", { limit: 50 });

await orderEvents.retryDeadLetter("shipping-service", dead[0].messageId);
await orderEvents.retryAllDeadLetters("shipping-service");

// Reprocess from the beginning.
await orderEvents.seek("analytics-service", { position: "earliest" });

// Start at the newest published message.
await orderEvents.seek("analytics-service", { position: "latest" });

// Rewind to an offset, timestamp, or message.
await orderEvents.seek("analytics-service", { offset: 42 });
await orderEvents.seek("analytics-service", { timestamp: new Date("2026-01-01") });
await orderEvents.seek("analytics-service", { messageId: "msg_123" });

await orderEvents.replay({
  group: "analytics-service",
  from: new Date("2026-01-01"),
});

const lag = await orderEvents.lag("billing-service");
const stats = await orderEvents.stats();

await orderEvents.pause("billing-service");
await orderEvents.resume("billing-service");

// Pause or resume every group on the topic.
await orderEvents.pause();
await orderEvents.resume();

// Remove all persisted PubSub state for this topic.
const removed = await orderEvents.purge();