feat: auto committed

services/content/src/shared/outbox/events.test.ts

@@ -0,0 +1,110 @@
+import { describe, it, expect } from "vitest";
+import {
+  CONTENT_TOPICS,
+  AGGREGATE_TYPES,
+  EVENT_TYPES,
+  getTopicForEvent,
+  buildEventPayload,
+} from "./events.js";
+
+describe("events", () => {
+  describe("CONTENT_TOPICS", () => {
+    it("should have 4 aggregate topics", () => {
+      expect(Object.keys(CONTENT_TOPICS)).toHaveLength(4);
+      expect(CONTENT_TOPICS.TEXTBOOK).toBe("edu.content.textbook.events");
+      expect(CONTENT_TOPICS.CHAPTER).toBe("edu.content.chapter.events");
+      expect(CONTENT_TOPICS.KNOWLEDGE_POINT).toBe(
+        "edu.content.knowledge_point.events",
+      );
+      expect(CONTENT_TOPICS.QUESTION).toBe("edu.content.question.events");
+    });
+  });
+
+  describe("AGGREGATE_TYPES", () => {
+    it("should have 4 aggregate types", () => {
+      expect(AGGREGATE_TYPES.TEXTBOOK).toBe("Textbook");
+      expect(AGGREGATE_TYPES.CHAPTER).toBe("Chapter");
+      expect(AGGREGATE_TYPES.KNOWLEDGE_POINT).toBe("KnowledgePoint");
+      expect(AGGREGATE_TYPES.QUESTION).toBe("Question");
+    });
+  });
+
+  describe("EVENT_TYPES", () => {
+    it("should have 15 event types", () => {
+      expect(Object.keys(EVENT_TYPES)).toHaveLength(15);
+    });
+  });
+
+  describe("getTopicForEvent", () => {
+    it("should route textbook events to textbook topic", () => {
+      expect(getTopicForEvent(EVENT_TYPES.TEXTBOOK_CREATED)).toBe(
+        CONTENT_TOPICS.TEXTBOOK,
+      );
+      expect(getTopicForEvent(EVENT_TYPES.TEXTBOOK_UPDATED)).toBe(
+        CONTENT_TOPICS.TEXTBOOK,
+      );
+      expect(getTopicForEvent(EVENT_TYPES.TEXTBOOK_PUBLISHED)).toBe(
+        CONTENT_TOPICS.TEXTBOOK,
+      );
+      expect(getTopicForEvent(EVENT_TYPES.TEXTBOOK_ARCHIVED)).toBe(
+        CONTENT_TOPICS.TEXTBOOK,
+      );
+    });
+
+    it("should route chapter events to chapter topic", () => {
+      expect(getTopicForEvent(EVENT_TYPES.CHAPTER_CREATED)).toBe(
+        CONTENT_TOPICS.CHAPTER,
+      );
+      expect(getTopicForEvent(EVENT_TYPES.CHAPTER_DELETED)).toBe(
+        CONTENT_TOPICS.CHAPTER,
+      );
+    });
+
+    it("should route KP events to knowledge_point topic", () => {
+      expect(getTopicForEvent(EVENT_TYPES.KP_CREATED)).toBe(
+        CONTENT_TOPICS.KNOWLEDGE_POINT,
+      );
+      expect(getTopicForEvent(EVENT_TYPES.KP_PREREQUISITE_ADDED)).toBe(
+        CONTENT_TOPICS.KNOWLEDGE_POINT,
+      );
+    });
+
+    it("should route question events to question topic", () => {
+      expect(getTopicForEvent(EVENT_TYPES.QUESTION_CREATED)).toBe(
+        CONTENT_TOPICS.QUESTION,
+      );
+      expect(getTopicForEvent(EVENT_TYPES.QUESTION_PUBLISHED)).toBe(
+        CONTENT_TOPICS.QUESTION,
+      );
+    });
+
+    it("should return fallback topic for unknown events", () => {
+      expect(getTopicForEvent("unknown.event")).toBe("edu.content.fallback");
+    });
+  });
+
+  describe("buildEventPayload", () => {
+    it("should build payload with event_id, aggregate_id, event_type, occurred_at, action", () => {
+      const payload = buildEventPayload("textbook.created", "tb-1", {
+        title: "Test",
+      });
+
+      expect(payload.event_id).toBeDefined();
+      expect(payload.aggregate_id).toBe("tb-1");
+      expect(payload.event_type).toBe("edu.content.textbook.created");
+      expect(payload.occurred_at).toBeGreaterThan(0);
+      expect(payload.action).toBe("created");
+      expect(payload.title).toBe("Test");
+    });
+
+    it("should extract action from event type with multiple dots", () => {
+      const payload = buildEventPayload(
+        "knowledge_point.prerequisite_added",
+        "kp-1",
+        {},
+      );
+      expect(payload.action).toBe("prerequisite_added");
+      expect(payload.event_type).toBe("edu.content.knowledge_point.prerequisite_added");
+    });
+  });
+});

services/content/src/shared/outbox/events.ts

@@ -0,0 +1,80 @@
+import { createId } from "@paralleldrive/cuid2";
+
+export const CONTENT_TOPICS = {
+  TEXTBOOK: "edu.content.textbook.events",
+  CHAPTER: "edu.content.chapter.events",
+  KNOWLEDGE_POINT: "edu.content.knowledge_point.events",
+  QUESTION: "edu.content.question.events",
+} as const;
+
+export const AGGREGATE_TYPES = {
+  TEXTBOOK: "Textbook",
+  CHAPTER: "Chapter",
+  KNOWLEDGE_POINT: "KnowledgePoint",
+  QUESTION: "Question",
+} as const;
+
+export const EVENT_TYPES = {
+  TEXTBOOK_CREATED: "textbook.created",
+  TEXTBOOK_UPDATED: "textbook.updated",
+  TEXTBOOK_PUBLISHED: "textbook.published",
+  TEXTBOOK_ARCHIVED: "textbook.archived",
+  CHAPTER_CREATED: "chapter.created",
+  CHAPTER_UPDATED: "chapter.updated",
+  CHAPTER_DELETED: "chapter.deleted",
+  KP_CREATED: "knowledge_point.created",
+  KP_UPDATED: "knowledge_point.updated",
+  KP_PREREQUISITE_ADDED: "knowledge_point.prerequisite_added",
+  KP_PREREQUISITE_REMOVED: "knowledge_point.prerequisite_removed",
+  QUESTION_CREATED: "question.created",
+  QUESTION_UPDATED: "question.updated",
+  QUESTION_PUBLISHED: "question.published",
+  QUESTION_DELETED: "question.deleted",
+} as const;
+
+const TOPIC_MAP: Record<string, string> = {
+  [EVENT_TYPES.TEXTBOOK_CREATED]: CONTENT_TOPICS.TEXTBOOK,
+  [EVENT_TYPES.TEXTBOOK_UPDATED]: CONTENT_TOPICS.TEXTBOOK,
+  [EVENT_TYPES.TEXTBOOK_PUBLISHED]: CONTENT_TOPICS.TEXTBOOK,
+  [EVENT_TYPES.TEXTBOOK_ARCHIVED]: CONTENT_TOPICS.TEXTBOOK,
+  [EVENT_TYPES.CHAPTER_CREATED]: CONTENT_TOPICS.CHAPTER,
+  [EVENT_TYPES.CHAPTER_UPDATED]: CONTENT_TOPICS.CHAPTER,
+  [EVENT_TYPES.CHAPTER_DELETED]: CONTENT_TOPICS.CHAPTER,
+  [EVENT_TYPES.KP_CREATED]: CONTENT_TOPICS.KNOWLEDGE_POINT,
+  [EVENT_TYPES.KP_UPDATED]: CONTENT_TOPICS.KNOWLEDGE_POINT,
+  [EVENT_TYPES.KP_PREREQUISITE_ADDED]: CONTENT_TOPICS.KNOWLEDGE_POINT,
+  [EVENT_TYPES.KP_PREREQUISITE_REMOVED]: CONTENT_TOPICS.KNOWLEDGE_POINT,
+  [EVENT_TYPES.QUESTION_CREATED]: CONTENT_TOPICS.QUESTION,
+  [EVENT_TYPES.QUESTION_UPDATED]: CONTENT_TOPICS.QUESTION,
+  [EVENT_TYPES.QUESTION_PUBLISHED]: CONTENT_TOPICS.QUESTION,
+  [EVENT_TYPES.QUESTION_DELETED]: CONTENT_TOPICS.QUESTION,
+};
+
+export function getTopicForEvent(eventType: string): string {
+  return TOPIC_MAP[eventType] ?? "edu.content.fallback";
+}
+
+export interface EventPayload {
+  event_id: string;
+  aggregate_id: string;
+  event_type: string;
+  occurred_at: number;
+  action: string;
+  [key: string]: unknown;
+}
+
+export function buildEventPayload(
+  eventType: string,
+  aggregateId: string,
+  data: Record<string, unknown>,
+): EventPayload {
+  const action = eventType.split(".").pop() ?? eventType;
+  return {
+    event_id: createId(),
+    aggregate_id: aggregateId,
+    event_type: `edu.content.${eventType}`,
+    occurred_at: Date.now(),
+    action,
+    ...data,
+  };
+}

services/content/src/shared/outbox/outbox.module.ts

@@ -0,0 +1,8 @@
+import { Module } from "@nestjs/common";
+import { OutboxService } from "./outbox.service.js";
+
+@Module({
+  providers: [OutboxService],
+  exports: [OutboxService],
+})
+export class OutboxModule {}

services/content/src/shared/outbox/outbox.publisher.test.ts

@@ -0,0 +1,165 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+
+const mockProducer = vi.hoisted(() => ({ send: vi.fn() }));
+
+vi.mock("../../config/kafka.js", () => ({
+  producer: mockProducer,
+}));
+
+vi.mock("../observability/logger.js", () => ({
+  logger: {
+    info: vi.fn(),
+    error: vi.fn(),
+    warn: vi.fn(),
+    debug: vi.fn(),
+  },
+}));
+
+vi.mock("./outbox.repository.js", () => ({
+  outboxRepository: {
+    findPending: vi.fn(),
+    markPublished: vi.fn(),
+    incrementRetry: vi.fn(),
+    markFailed: vi.fn(),
+    create: vi.fn(),
+  },
+}));
+
+import { OutboxPublisher } from "./outbox.publisher.js";
+import { outboxRepository } from "./outbox.repository.js";
+import type { OutboxMessage } from "./outbox.schema.js";
+
+function createMessage(
+  overrides: Partial<OutboxMessage> = {},
+): OutboxMessage {
+  return {
+    id: "msg-1",
+    aggregateType: "Chapter",
+    aggregateId: "ch-1",
+    eventType: "chapter.created",
+    topic: "edu.content.chapter.events",
+    payload: '{"event_id":"e1"}',
+    status: "pending",
+    retryCount: 0,
+    createdAt: new Date(),
+    publishedAt: null,
+    nextRetryAt: null,
+    lastError: null,
+    ...overrides,
+  };
+}
+
+describe("OutboxPublisher", () => {
+  let publisher: OutboxPublisher;
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    publisher = new OutboxPublisher();
+  });
+
+  describe("dispatch (via poll)", () => {
+    it("should send message to kafka and mark as published", async () => {
+      const message = createMessage();
+      vi.mocked(outboxRepository.findPending).mockResolvedValue([message]);
+      mockProducer.send.mockResolvedValue(undefined);
+
+      // Access private poll via casting
+      await (publisher as unknown as { poll: () => Promise<void> }).poll();
+
+      expect(mockProducer.send).toHaveBeenCalledWith({
+        topic: message.topic,
+        messages: [
+          {
+            key: message.aggregateId,
+            value: message.payload,
+            headers: {
+              eventId: message.id,
+              eventType: message.eventType,
+              aggregateType: message.aggregateType,
+            },
+          },
+        ],
+      });
+      expect(outboxRepository.markPublished).toHaveBeenCalledWith(message.id);
+    });
+
+    it("should increment retry when send fails and below max retries", async () => {
+      const message = createMessage({ id: "msg-2", retryCount: 0 });
+      vi.mocked(outboxRepository.findPending).mockResolvedValue([message]);
+      mockProducer.send.mockRejectedValue(new Error("kafka down"));
+
+      await (publisher as unknown as { poll: () => Promise<void> }).poll();
+
+      expect(outboxRepository.incrementRetry).toHaveBeenCalledWith(
+        "msg-2",
+        "kafka down",
+      );
+      expect(outboxRepository.markFailed).not.toHaveBeenCalled();
+    });
+
+    it("should mark failed when retry count reaches max", async () => {
+      const message = createMessage({ id: "msg-3", retryCount: 4 });
+      vi.mocked(outboxRepository.findPending).mockResolvedValue([message]);
+      mockProducer.send.mockRejectedValue(new Error("kafka down"));
+
+      await (publisher as unknown as { poll: () => Promise<void> }).poll();
+
+      expect(outboxRepository.markFailed).toHaveBeenCalledWith(
+        "msg-3",
+        "kafka down",
+      );
+      expect(outboxRepository.incrementRetry).not.toHaveBeenCalled();
+    });
+
+    it("should handle non-Error rejection in dispatch", async () => {
+      const message = createMessage({ id: "msg-4", retryCount: 0 });
+      vi.mocked(outboxRepository.findPending).mockResolvedValue([message]);
+      mockProducer.send.mockRejectedValue("string error");
+
+      await (publisher as unknown as { poll: () => Promise<void> }).poll();
+
+      expect(outboxRepository.incrementRetry).toHaveBeenCalledWith(
+        "msg-4",
+        "string error",
+      );
+    });
+  });
+
+  describe("poll error handling", () => {
+    it("should swallow errors from findPending", async () => {
+      vi.mocked(outboxRepository.findPending).mockRejectedValue(
+        new Error("db error"),
+      );
+      await expect(
+        (publisher as unknown as { poll: () => Promise<void> }).poll(),
+      ).resolves.not.toThrow();
+    });
+
+    it("should process multiple messages in a batch", async () => {
+      const messages = [
+        createMessage({ id: "m1" }),
+        createMessage({ id: "m2" }),
+      ];
+      vi.mocked(outboxRepository.findPending).mockResolvedValue(messages);
+      mockProducer.send.mockResolvedValue(undefined);
+
+      await (publisher as unknown as { poll: () => Promise<void> }).poll();
+
+      expect(mockProducer.send).toHaveBeenCalledTimes(2);
+      expect(outboxRepository.markPublished).toHaveBeenCalledWith("m1");
+      expect(outboxRepository.markPublished).toHaveBeenCalledWith("m2");
+    });
+  });
+
+  describe("start/stop", () => {
+    it("should start and stop without error", async () => {
+      await publisher.start();
+      await publisher.stop();
+      // Should not throw
+    });
+
+    it("should stop without error when not started", async () => {
+      await publisher.stop();
+    });
+  });
+});

services/content/src/shared/outbox/outbox.publisher.ts

@@ -0,0 +1,81 @@
+import { logger } from "../observability/logger.js";
+import { producer } from "../../config/kafka.js";
+import { outboxRepository } from "./outbox.repository.js";
+import type { OutboxMessage } from "./outbox.schema.js";
+
+const POLL_INTERVAL_MS = 5000;
+const BATCH_SIZE = 100;
+const MAX_RETRY = 5;
+
+export class OutboxPublisher {
+  private intervalId: ReturnType<typeof setInterval> | null = null;
+  private isPolling = false;
+
+  async start(): Promise<void> {
+    logger.info("OutboxPublisher started");
+    this.intervalId = setInterval(() => {
+      void this.poll();
+    }, POLL_INTERVAL_MS);
+  }
+
+  async stop(): Promise<void> {
+    if (this.intervalId) {
+      clearInterval(this.intervalId);
+      this.intervalId = null;
+    }
+    logger.info("OutboxPublisher stopped");
+  }
+
+  private async poll(): Promise<void> {
+    if (this.isPolling) return;
+    this.isPolling = true;
+    try {
+      const messages = await outboxRepository.findPending(BATCH_SIZE);
+      for (const message of messages) {
+        await this.dispatch(message);
+      }
+    } catch (error) {
+      logger.error({ error }, "Outbox poll failed");
+    } finally {
+      this.isPolling = false;
+    }
+  }
+
+  private async dispatch(message: OutboxMessage): Promise<void> {
+    try {
+      await producer.send({
+        topic: message.topic,
+        messages: [
+          {
+            key: message.aggregateId,
+            value: message.payload,
+            headers: {
+              eventId: message.id,
+              eventType: message.eventType,
+              aggregateType: message.aggregateType,
+            },
+          },
+        ],
+      });
+      await outboxRepository.markPublished(message.id);
+      logger.info(
+        { id: message.id, eventType: message.eventType, topic: message.topic },
+        "Outbox message published",
+      );
+    } catch (error) {
+      const errorMessage =
+        error instanceof Error ? error.message : String(error);
+      logger.error(
+        { error, id: message.id, eventType: message.eventType },
+        "Outbox publish failed",
+      );
+      if (message.retryCount + 1 >= MAX_RETRY) {
+        await outboxRepository.markFailed(message.id, errorMessage);
+      } else {
+        await outboxRepository.incrementRetry(message.id, errorMessage);
+      }
+    }
+  }
+}
+
+export const outboxPublisher = new OutboxPublisher();

services/content/src/shared/outbox/outbox.repository.ts

@@ -0,0 +1,59 @@
+import { eq, sql, and, or, isNull, lte } from "drizzle-orm";
+import type { MySql2Database } from "drizzle-orm/mysql2";
+import { getDb } from "../../config/database.js";
+import {
+  outbox,
+  type OutboxMessage,
+  type NewOutboxMessage,
+} from "./outbox.schema.js";
+
+type DbClient = MySql2Database;
+
+export class OutboxRepository {
+  async create(message: NewOutboxMessage, tx?: DbClient): Promise<void> {
+    const client = tx ?? getDb();
+    await client.insert(outbox).values(message);
+  }
+
+  async findPending(limit: number = 100): Promise<OutboxMessage[]> {
+    const now = new Date();
+    return getDb()
+      .select()
+      .from(outbox)
+      .where(
+        and(
+          eq(outbox.status, "pending"),
+          or(isNull(outbox.nextRetryAt), lte(outbox.nextRetryAt, now)),
+        ),
+      )
+      .limit(limit);
+  }
+
+  async markPublished(id: string): Promise<void> {
+    await getDb()
+      .update(outbox)
+      .set({ status: "published", publishedAt: new Date(), lastError: null })
+      .where(eq(outbox.id, id));
+  }
+
+  async incrementRetry(id: string, errorMessage: string): Promise<void> {
+    const backoffMs = 5000 * 2 ** 1;
+    await getDb()
+      .update(outbox)
+      .set({
+        retryCount: sql`${outbox.retryCount} + 1`,
+        nextRetryAt: new Date(Date.now() + backoffMs),
+        lastError: errorMessage,
+      })
+      .where(eq(outbox.id, id));
+  }
+
+  async markFailed(id: string, errorMessage: string): Promise<void> {
+    await getDb()
+      .update(outbox)
+      .set({ status: "failed", lastError: errorMessage })
+      .where(eq(outbox.id, id));
+  }
+}
+
+export const outboxRepository = new OutboxRepository();

services/content/src/shared/outbox/outbox.schema.ts

@@ -0,0 +1,33 @@
+import {
+  mysqlTable,
+  varchar,
+  text,
+  timestamp,
+  int,
+  index,
+} from "drizzle-orm/mysql-core";
+
+export const outbox = mysqlTable(
+  "content_outbox_events",
+  {
+    id: varchar("id", { length: 32 }).notNull().primaryKey(),
+    aggregateType: varchar("aggregate_type", { length: 64 }).notNull(),
+    aggregateId: varchar("aggregate_id", { length: 32 }).notNull(),
+    eventType: varchar("event_type", { length: 100 }).notNull(),
+    topic: varchar("topic", { length: 128 }).notNull(),
+    payload: text("payload").notNull(),
+    status: varchar("status", { length: 20 }).notNull().default("pending"),
+    retryCount: int("retry_count").notNull().default(0),
+    createdAt: timestamp("created_at").notNull().defaultNow(),
+    publishedAt: timestamp("published_at"),
+    nextRetryAt: timestamp("next_retry_at"),
+    lastError: text("last_error"),
+  },
+  (table) => ({
+    statusRetryIdx: index("idx_outbox_status_retry").on(table.status, table.nextRetryAt),
+    aggregateIdx: index("idx_outbox_aggregate").on(table.aggregateType, table.aggregateId),
+  }),
+);
+
+export type OutboxMessage = typeof outbox.$inferSelect;
+export type NewOutboxMessage = typeof outbox.$inferInsert;

services/content/src/shared/outbox/outbox.service.test.ts

@@ -0,0 +1,109 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+
+vi.mock("@paralleldrive/cuid2", () => ({
+  createId: vi.fn().mockReturnValue("test-event-id"),
+}));
+
+vi.mock("./outbox.repository.js", () => ({
+  outboxRepository: {
+    create: vi.fn(),
+  },
+}));
+
+import { OutboxService } from "./outbox.service.js";
+import { outboxRepository } from "./outbox.repository.js";
+import { getTopicForEvent } from "./events.js";
+
+describe("OutboxService", () => {
+  let service: OutboxService;
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    service = new OutboxService();
+  });
+
+  describe("publish", () => {
+    it("should call outboxRepository.create with correct payload structure", async () => {
+      const data = { title: "Test Chapter" };
+      const eventId = await service.publish(
+        "chapter.created",
+        "Chapter",
+        "ch-1",
+        data,
+      );
+
+      expect(outboxRepository.create).toHaveBeenCalledTimes(1);
+      const [message] = vi.mocked(outboxRepository.create).mock.calls[0]!;
+      expect(message.id).toBe("test-event-id");
+      expect(message.aggregateType).toBe("Chapter");
+      expect(message.aggregateId).toBe("ch-1");
+      expect(message.eventType).toBe("chapter.created");
+      expect(message.status).toBe("pending");
+      expect(typeof message.payload).toBe("string");
+      expect(eventId).toBe("test-event-id");
+    });
+
+    it("should set status to pending", async () => {
+      await service.publish("chapter.updated", "Chapter", "ch-1", {});
+      const [message] = vi.mocked(outboxRepository.create).mock.calls[0]!;
+      expect(message.status).toBe("pending");
+    });
+
+    it("should set topic based on event type via getTopicForEvent", async () => {
+      await service.publish("chapter.created", "Chapter", "ch-1", {});
+      const [message] = vi.mocked(outboxRepository.create).mock.calls[0]!;
+      expect(message.topic).toBe(getTopicForEvent("chapter.created"));
+    });
+
+    it("should serialize payload as JSON string containing event data", async () => {
+      const data = { title: "Test Chapter", order: 2 };
+      await service.publish("chapter.created", "Chapter", "ch-1", data);
+      const [message] = vi.mocked(outboxRepository.create).mock.calls[0]!;
+
+      const parsed = JSON.parse(message.payload);
+      expect(parsed.aggregate_id).toBe("ch-1");
+      expect(parsed.event_type).toBe("edu.content.chapter.created");
+      expect(parsed.action).toBe("created");
+      expect(parsed.title).toBe("Test Chapter");
+      expect(parsed.order).toBe(2);
+    });
+
+    it("should return the generated eventId", async () => {
+      const eventId = await service.publish(
+        "chapter.deleted",
+        "Chapter",
+        "ch-1",
+        {},
+      );
+      expect(eventId).toBe("test-event-id");
+    });
+
+    it("should pass tx to repository when provided", async () => {
+      const tx = {} as never;
+      await service.publish(
+        "chapter.created",
+        "Chapter",
+        "ch-1",
+        {},
+        tx,
+      );
+      const [, passedTx] = vi.mocked(outboxRepository.create).mock.calls[0]!;
+      expect(passedTx).toBe(tx);
+    });
+
+    it("should pass undefined tx when not provided", async () => {
+      await service.publish("chapter.created", "Chapter", "ch-1", {});
+      const [, passedTx] = vi.mocked(outboxRepository.create).mock.calls[0]!;
+      expect(passedTx).toBeUndefined();
+    });
+
+    it("should include aggregate data in payload", async () => {
+      const data = { name: "Chapter A", status: "draft" };
+      await service.publish("chapter.created", "Chapter", "ch-9", data);
+      const [message] = vi.mocked(outboxRepository.create).mock.calls[0]!;
+      const parsed = JSON.parse(message.payload);
+      expect(parsed.name).toBe("Chapter A");
+      expect(parsed.status).toBe("draft");
+    });
+  });
+});

services/content/src/shared/outbox/outbox.service.ts

@@ -0,0 +1,35 @@
+import { Injectable } from "@nestjs/common";
+import { createId } from "@paralleldrive/cuid2";
+import type { MySql2Database } from "drizzle-orm/mysql2";
+import { outboxRepository } from "./outbox.repository.js";
+import { getTopicForEvent, buildEventPayload } from "./events.js";
+
+@Injectable()
+export class OutboxService {
+  async publish(
+    eventType: string,
+    aggregateType: string,
+    aggregateId: string,
+    data: Record<string, unknown>,
+    tx?: MySql2Database,
+  ): Promise<string> {
+    const eventId = createId();
+    const topic = getTopicForEvent(eventType);
+    const payload = buildEventPayload(eventType, aggregateId, data);
+
+    await outboxRepository.create(
+      {
+        id: eventId,
+        aggregateType,
+        aggregateId,
+        eventType,
+        topic,
+        payload: JSON.stringify(payload),
+        status: "pending",
+      },
+      tx,
+    );
+
+    return eventId;
+  }
+}