chore(content): merge content full implementation into main

Merge feat/content-ai09 with complete content service

services/content/src/shared/errors/application-error.ts

@@ -1,11 +1,12 @@
 export type ErrorType =
-  | 'validation'
-  | 'not_found'
-  | 'permission_denied'
-  | 'conflict'
-  | 'business'
-  | 'database'
-  | 'internal';
+  | "validation"
+  | "not_found"
+  | "permission_denied"
+  | "conflict"
+  | "business"
+  | "database"
+  | "internal"
+  | "neo4j_unavailable";
 
 export interface ErrorDetails {
   [key: string]: unknown;
@@ -40,57 +41,70 @@ export abstract class ApplicationError extends Error {
 }
 
 export class ValidationError extends ApplicationError {
-  readonly type = 'validation' as const;
+  readonly type = "validation" as const;
   readonly statusCode = 400;
   constructor(message: string, details?: ErrorDetails) {
-    super(message, 'CONTENT_VALIDATION_ERROR', details);
+    super(message, "CONTENT_VALIDATION_ERROR", details);
   }
 }
 
 export class NotFoundError extends ApplicationError {
-  readonly type = 'not_found' as const;
+  readonly type = "not_found" as const;
   readonly statusCode = 404;
   constructor(resource: string, id: string) {
-    super(`${resource} not found: ${id}`, 'CONTENT_NOT_FOUND', { resource, id });
+    super(`${resource} not found: ${id}`, "CONTENT_NOT_FOUND", {
+      resource,
+      id,
+    });
   }
 }
 
 export class PermissionDeniedError extends ApplicationError {
-  readonly type = 'permission_denied' as const;
+  readonly type = "permission_denied" as const;
   readonly statusCode = 403;
   constructor(permission: string) {
-    super(`Permission denied: ${permission}`, 'CONTENT_PERMISSION_DENIED', { permission });
+    super(`Permission denied: ${permission}`, "CONTENT_PERMISSION_DENIED", {
+      permission,
+    });
   }
 }
 
 export class ConflictError extends ApplicationError {
-  readonly type = 'conflict' as const;
+  readonly type = "conflict" as const;
   readonly statusCode = 409;
   constructor(message: string, details?: ErrorDetails) {
-    super(message, 'CONTENT_CONFLICT', details);
+    super(message, "CONTENT_CONFLICT", details);
   }
 }
 
 export class BusinessError extends ApplicationError {
-  readonly type = 'business' as const;
+  readonly type = "business" as const;
   readonly statusCode = 422;
   constructor(message: string, details?: ErrorDetails) {
-    super(message, 'CONTENT_BUSINESS_ERROR', details);
+    super(message, "CONTENT_BUSINESS_ERROR", details);
   }
 }
 
 export class DatabaseError extends ApplicationError {
-  readonly type = 'database' as const;
+  readonly type = "database" as const;
   readonly statusCode = 500;
   constructor(message: string, details?: ErrorDetails) {
-    super(message, 'CONTENT_DATABASE_ERROR', details);
+    super(message, "CONTENT_DATABASE_ERROR", details);
   }
 }
 
 export class InternalError extends ApplicationError {
-  readonly type = 'internal' as const;
+  readonly type = "internal" as const;
   readonly statusCode = 500;
   constructor(message: string, details?: ErrorDetails) {
-    super(message, 'CONTENT_INTERNAL_ERROR', details);
+    super(message, "CONTENT_INTERNAL_ERROR", details);
+  }
+}
+
+export class Neo4jUnavailableError extends ApplicationError {
+  readonly type = "neo4j_unavailable" as const;
+  readonly statusCode = 503;
+  constructor(message: string, details?: ErrorDetails) {
+    super(message, "CONTENT_NEO4J_UNAVAILABLE", details);
   }
 }

services/content/src/shared/errors/global-error.filter.test.ts

@@ -0,0 +1,211 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import type { ArgumentsHost } from "@nestjs/common";
+import { HttpException } from "@nestjs/common";
+import type { Request, Response } from "express";
+import { z, ZodError } from "zod";
+import { GlobalErrorFilter } from "./global-error.filter.js";
+import {
+  NotFoundError,
+  ValidationError,
+  InternalError,
+} from "./application-error.js";
+
+function createMockHost(
+  req: Partial<Request>,
+  res: Partial<Response>,
+): ArgumentsHost {
+  return {
+    switchToHttp: () => ({
+      getResponse: () => res,
+      getRequest: () => req,
+    }),
+  } as unknown as ArgumentsHost;
+}
+
+function createMockResponse(): {
+  response: Response;
+  statusMock: ReturnType<typeof vi.fn>;
+  jsonMock: ReturnType<typeof vi.fn>;
+} {
+  const statusMock = vi.fn().mockReturnThis();
+  const jsonMock = vi.fn();
+  return {
+    response: { status: statusMock, json: jsonMock } as unknown as Response,
+    statusMock,
+    jsonMock,
+  };
+}
+
+function createMockRequest(headers: Record<string, unknown> = {}): Request {
+  return { headers } as unknown as Request;
+}
+
+describe("GlobalErrorFilter", () => {
+  let filter: GlobalErrorFilter;
+
+  beforeEach(() => {
+    filter = new GlobalErrorFilter();
+  });
+
+  describe("ApplicationError branch", () => {
+    it("should respond with ApplicationError statusCode and toJSON body", () => {
+      const { response, statusMock, jsonMock } = createMockResponse();
+      const host = createMockHost(
+        createMockRequest({ "x-request-id": "trace-abc" }),
+        response,
+      );
+      const error = new NotFoundError("Chapter", "ch-1");
+
+      filter.catch(error, host);
+
+      expect(statusMock).toHaveBeenCalledWith(404);
+      expect(jsonMock).toHaveBeenCalledWith(
+        expect.objectContaining({
+          success: false,
+          error: expect.objectContaining({
+            code: "CONTENT_NOT_FOUND",
+            traceId: "trace-abc",
+          }),
+        }),
+      );
+      expect(error.traceId).toBe("trace-abc");
+    });
+
+    it("should handle ValidationError with statusCode 400", () => {
+      const { response, statusMock, jsonMock } = createMockResponse();
+      const host = createMockHost(createMockRequest(), response);
+      const error = new ValidationError("Invalid input");
+
+      filter.catch(error, host);
+
+      expect(statusMock).toHaveBeenCalledWith(400);
+      expect(jsonMock).toHaveBeenCalledWith(
+        expect.objectContaining({
+          error: expect.objectContaining({ code: "CONTENT_VALIDATION_ERROR" }),
+        }),
+      );
+    });
+  });
+
+  describe("ZodError branch", () => {
+    it("should respond with statusCode 400 and CONTENT_VALIDATION_ERROR", () => {
+      const { response, statusMock, jsonMock } = createMockResponse();
+      const host = createMockHost(
+        createMockRequest({ "x-request-id": "trace-zod" }),
+        response,
+      );
+      const parseResult = z.object({ x: z.string() }).safeParse({ x: 123 });
+      const zodError = parseResult.error as ZodError;
+
+      filter.catch(zodError, host);
+
+      expect(statusMock).toHaveBeenCalledWith(400);
+      expect(jsonMock).toHaveBeenCalledWith(
+        expect.objectContaining({
+          success: false,
+          error: expect.objectContaining({
+            code: "CONTENT_VALIDATION_ERROR",
+            message: "Validation failed",
+            traceId: "trace-zod",
+            details: expect.any(Object),
+          }),
+        }),
+      );
+    });
+  });
+
+  describe("HttpException branch", () => {
+    it("should handle HttpException with string response", () => {
+      const { response, statusMock, jsonMock } = createMockResponse();
+      const host = createMockHost(createMockRequest(), response);
+      const error = new HttpException("Forbidden resource", 403);
+
+      filter.catch(error, host);
+
+      expect(statusMock).toHaveBeenCalledWith(403);
+      expect(jsonMock).toHaveBeenCalledWith(
+        expect.objectContaining({
+          error: expect.objectContaining({
+            code: "HTTP_ERROR",
+            message: "Forbidden resource",
+          }),
+        }),
+      );
+    });
+
+    it("should handle HttpException with object response containing message string", () => {
+      const { response, statusMock, jsonMock } = createMockResponse();
+      const host = createMockHost(createMockRequest(), response);
+      const error = new HttpException({ message: "Custom error message" }, 422);
+
+      filter.catch(error, host);
+
+      expect(statusMock).toHaveBeenCalledWith(422);
+      expect(jsonMock).toHaveBeenCalledWith(
+        expect.objectContaining({
+          error: expect.objectContaining({
+            message: "Custom error message",
+          }),
+        }),
+      );
+    });
+
+    it("should fall back to exception.message for object response without message", () => {
+      const { response, statusMock, jsonMock } = createMockResponse();
+      const host = createMockHost(createMockRequest(), response);
+      const error = new HttpException({ data: "value" }, 400);
+
+      filter.catch(error, host);
+
+      expect(statusMock).toHaveBeenCalledWith(400);
+      expect(jsonMock).toHaveBeenCalledWith(
+        expect.objectContaining({
+          error: expect.objectContaining({
+            code: "HTTP_ERROR",
+            message: error.message,
+          }),
+        }),
+      );
+    });
+  });
+
+  describe("unknown error branch", () => {
+    it("should respond with statusCode 500 and INTERNAL_ERROR", () => {
+      const { response, statusMock, jsonMock } = createMockResponse();
+      const host = createMockHost(
+        createMockRequest({ "x-request-id": "trace-unknown" }),
+        response,
+      );
+      const error = new Error("something exploded");
+
+      filter.catch(error, host);
+
+      expect(statusMock).toHaveBeenCalledWith(500);
+      expect(jsonMock).toHaveBeenCalledWith(
+        expect.objectContaining({
+          success: false,
+          error: expect.objectContaining({
+            code: "INTERNAL_ERROR",
+            message: "An unexpected error occurred",
+            traceId: "trace-unknown",
+          }),
+        }),
+      );
+    });
+
+    it("should default traceId to unknown when x-request-id header is missing", () => {
+      const { response, statusMock, jsonMock } = createMockResponse();
+      const host = createMockHost(createMockRequest(), response);
+      const error = new InternalError("boom");
+
+      filter.catch(error, host);
+
+      expect(jsonMock).toHaveBeenCalledWith(
+        expect.objectContaining({
+          error: expect.objectContaining({ traceId: "unknown" }),
+        }),
+      );
+      expect(statusMock).toHaveBeenCalledWith(500);
+    });
+  });
+});

services/content/src/shared/health/health.controller.test.ts

@@ -0,0 +1,83 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { HttpException } from "@nestjs/common";
+
+const mockGetDb = vi.fn();
+const mockGetNeo4jSession = vi.fn();
+const mockIsKafkaConnected = vi.fn();
+
+vi.mock("../../config/database.js", () => ({
+  getDb: () => mockGetDb(),
+}));
+
+vi.mock("../../config/neo4j.js", () => ({
+  getNeo4jSession: () => mockGetNeo4jSession(),
+}));
+
+vi.mock("../../config/kafka.js", () => ({
+  isKafkaProducerConnected: () => mockIsKafkaConnected(),
+}));
+
+vi.mock("../../config/env.js", () => ({
+  env: { NEO4J_URL: undefined, NEO4J_PASSWORD: undefined },
+}));
+
+import { HealthController } from "./health.controller.js";
+
+describe("HealthController", () => {
+  let controller: HealthController;
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    controller = new HealthController();
+  });
+
+  describe("liveness", () => {
+    it("should return ok status with service name", () => {
+      const result = controller.liveness();
+      expect(result.status).toBe("ok");
+      expect(result.service).toBe("content");
+      expect(result.timestamp).toBeDefined();
+    });
+  });
+
+  describe("readiness", () => {
+    it("should return ok when all dependencies healthy", async () => {
+      mockGetDb.mockReturnValue({
+        execute: vi.fn().mockResolvedValue(undefined),
+      });
+      mockIsKafkaConnected.mockReturnValue(true);
+      const result = await controller.readiness();
+      expect(result.status).toBe("ok");
+      expect(result.dependencies).toHaveLength(3);
+      const names = result.dependencies.map((d) => d.name);
+      expect(names).toEqual(["mysql", "neo4j", "kafka"]);
+      expect(result.dependencies.every((d) => d.status === "ok")).toBe(true);
+    });
+
+    it("should report neo4j as ok when not configured", async () => {
+      mockGetDb.mockReturnValue({
+        execute: vi.fn().mockResolvedValue(undefined),
+      });
+      mockIsKafkaConnected.mockReturnValue(true);
+      const result = await controller.readiness();
+      const neo4jDep = result.dependencies.find((d) => d.name === "neo4j");
+      expect(neo4jDep?.status).toBe("ok");
+    });
+
+    it("should report mysql error when db check fails", async () => {
+      mockGetDb.mockReturnValue({
+        execute: vi.fn().mockRejectedValue(new Error("db down")),
+      });
+      mockIsKafkaConnected.mockReturnValue(true);
+      await expect(controller.readiness()).rejects.toThrow(HttpException);
+    });
+
+    it("should report kafka error when producer disconnected", async () => {
+      mockGetDb.mockReturnValue({
+        execute: vi.fn().mockResolvedValue(undefined),
+      });
+      mockIsKafkaConnected.mockReturnValue(false);
+      await expect(controller.readiness()).rejects.toThrow(HttpException);
+    });
+  });
+});

services/content/src/shared/health/health.controller.ts

@@ -1,9 +1,25 @@
 import { Controller, Get, HttpException, HttpStatus } from "@nestjs/common";
 import { sql } from "drizzle-orm";
-import { db } from "../../config/database.js";
+import { getDb } from "../../config/database.js";
+import { getNeo4jSession } from "../../config/neo4j.js";
+import { isKafkaProducerConnected } from "../../config/kafka.js";
+import { env } from "../../config/env.js";
 
 const SERVICE_NAME = "content";
 
+interface DependencyCheck {
+  name: string;
+  status: "ok" | "error";
+  error?: string;
+}
+
+interface ReadinessResponse {
+  status: "ok" | "error";
+  service: string;
+  timestamp: string;
+  dependencies: DependencyCheck[];
+}
+
 @Controller()
 export class HealthController {
   @Get("healthz")
@@ -16,29 +32,83 @@ export class HealthController {
   }
 
   @Get("readyz")
-  async readiness(): Promise<{
-    status: string;
-    service: string;
-    timestamp: string;
-  }> {
+  async readiness(): Promise<ReadinessResponse> {
+    const dependencies: DependencyCheck[] = [];
+
+    // 1. DB check
     try {
-      await db.execute(sql`SELECT 1`);
-      return {
+      await getDb().execute(sql`SELECT 1`);
+      dependencies.push({ name: "mysql", status: "ok" });
+    } catch (err) {
+      dependencies.push({
+        name: "mysql",
+        status: "error",
+        error: err instanceof Error ? err.message : String(err),
+      });
+    }
+
+    // 2. Neo4j check (optional — only if configured)
+    if (env.NEO4J_URL && env.NEO4J_PASSWORD) {
+      const session = getNeo4jSession();
+      if (!session) {
+        dependencies.push({
+          name: "neo4j",
+          status: "error",
+          error: "driver not initialized",
+        });
+      } else {
+        try {
+          await session.executeRead((tx) => tx.run("RETURN 1"));
+          dependencies.push({ name: "neo4j", status: "ok" });
+        } catch (err) {
+          dependencies.push({
+            name: "neo4j",
+            status: "error",
+            error: err instanceof Error ? err.message : String(err),
+          });
+        } finally {
+          await session.close();
+        }
+      }
+    } else {
+      dependencies.push({
+        name: "neo4j",
         status: "ok",
-        service: SERVICE_NAME,
-        timestamp: new Date().toISOString(),
-      };
-    } catch (error) {
+        error: "not configured (optional)",
+      });
+    }
+
+    // 3. Kafka check (non-blocking — producer may be disconnected if broker unavailable)
+    if (isKafkaProducerConnected()) {
+      dependencies.push({ name: "kafka", status: "ok" });
+    } else {
+      dependencies.push({
+        name: "kafka",
+        status: "error",
+        error: "producer not connected",
+      });
+    }
+
+    const allOk = dependencies.every((d) => d.status === "ok");
+    const status = allOk ? "ok" : "error";
+
+    if (!allOk) {
       throw new HttpException(
         {
-          status: "error",
+          status,
           service: SERVICE_NAME,
           timestamp: new Date().toISOString(),
-          error:
-            error instanceof Error ? error.message : "database unreachable",
-        },
+          dependencies,
+        } satisfies ReadinessResponse,
         HttpStatus.SERVICE_UNAVAILABLE,
       );
     }
+
+    return {
+      status,
+      service: SERVICE_NAME,
+      timestamp: new Date().toISOString(),
+      dependencies,
+    } satisfies ReadinessResponse;
   }
 }

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

@@ -0,0 +1,54 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+
+const mockCloseDb = vi.fn();
+const mockCloseNeo4j = vi.fn();
+
+vi.mock("../../config/database.js", () => ({
+  closeDb: () => mockCloseDb(),
+}));
+
+vi.mock("../../config/neo4j.js", () => ({
+  closeNeo4j: () => mockCloseNeo4j(),
+}));
+
+import { LifecycleService } from "./lifecycle.service.js";
+
+describe("LifecycleService", () => {
+  let service: LifecycleService;
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    service = new LifecycleService();
+  });
+
+  describe("onModuleInit", () => {
+    it("should not throw on init", () => {
+      expect(() => service.onModuleInit()).not.toThrow();
+    });
+  });
+
+  describe("onApplicationShutdown", () => {
+    it("should close neo4j and db on shutdown", async () => {
+      mockCloseNeo4j.mockResolvedValue(undefined);
+      mockCloseDb.mockResolvedValue(undefined);
+      await service.onApplicationShutdown("SIGTERM");
+      expect(mockCloseNeo4j).toHaveBeenCalled();
+      expect(mockCloseDb).toHaveBeenCalled();
+    });
+
+    it("should handle signal undefined as unknown", async () => {
+      mockCloseNeo4j.mockResolvedValue(undefined);
+      mockCloseDb.mockResolvedValue(undefined);
+      await service.onApplicationShutdown();
+      expect(mockCloseDb).toHaveBeenCalled();
+    });
+
+    it("should swallow errors during shutdown", async () => {
+      mockCloseNeo4j.mockRejectedValue(new Error("neo4j close failed"));
+      mockCloseDb.mockResolvedValue(undefined);
+      await expect(
+        service.onApplicationShutdown("SIGTERM"),
+      ).resolves.not.toThrow();
+    });
+  });
+});

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

@@ -0,0 +1,112 @@
+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,163 @@
+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,39 @@
+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,103 @@
+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;
+  }
+}

services/content/src/shared/sync/neo4j-sync.worker.ts

@@ -0,0 +1,187 @@
+import { logger } from "../observability/logger.js";
+import { neo4jSyncConsumer } from "../../config/kafka.js";
+import { getNeo4jSession } from "../../config/neo4j.js";
+import { CONTENT_TOPICS, EVENT_TYPES } from "../outbox/events.js";
+
+interface EventEnvelope {
+  event_id: string;
+  aggregate_id: string;
+  event_type: string;
+  occurred_at: number;
+  action: string;
+  kp_id?: string;
+  chapter_id?: string;
+  title?: string;
+  difficulty?: number;
+  prerequisite_id?: string;
+  [key: string]: unknown;
+}
+
+const processedEvents = new Set<string>();
+const DEDUP_MAX_SIZE = 10000;
+
+export class Neo4jSyncWorker {
+  private running = false;
+
+  async start(): Promise<void> {
+    try {
+      await neo4jSyncConsumer.connect();
+      await neo4jSyncConsumer.subscribe({
+        topic: CONTENT_TOPICS.KNOWLEDGE_POINT,
+        fromBeginning: false,
+      });
+
+      this.running = true;
+      await neo4jSyncConsumer.run({
+        eachMessage: async ({ message }) => {
+          await this.handleMessage(message);
+        },
+      });
+      logger.info("Neo4jSyncWorker started");
+    } catch (err) {
+      logger.warn(
+        {
+          err: err instanceof Error ? err.message : String(err),
+        },
+        "Neo4jSyncWorker failed to start (Neo4j/Kafka may be unavailable)",
+      );
+    }
+  }
+
+  async stop(): Promise<void> {
+    this.running = false;
+    try {
+      await neo4jSyncConsumer.stop();
+    } catch {
+      // ignore
+    }
+    logger.info("Neo4jSyncWorker stopped");
+  }
+
+  isRunning(): boolean {
+    return this.running;
+  }
+
+  private async handleMessage(message: {
+    key: Buffer | null;
+    value: Buffer | null;
+  }): Promise<void> {
+    const value = message.value?.toString();
+    if (!value) return;
+
+    let event: EventEnvelope;
+    try {
+      event = JSON.parse(value) as EventEnvelope;
+    } catch {
+      logger.error({ value }, "Failed to parse event payload");
+      return;
+    }
+
+    if (processedEvents.has(event.event_id)) {
+      return;
+    }
+    processedEvents.add(event.event_id);
+    if (processedEvents.size > DEDUP_MAX_SIZE) {
+      const first = processedEvents.values().next().value;
+      if (first) processedEvents.delete(first);
+    }
+
+    const session = getNeo4jSession();
+    if (!session) {
+      logger.debug(
+        { eventType: event.event_type },
+        "Neo4j not available, skipping sync",
+      );
+      return;
+    }
+
+    try {
+      await this.syncEvent(session, event);
+    } catch (err) {
+      logger.error(
+        {
+          err: err instanceof Error ? err.message : String(err),
+          eventId: event.event_id,
+          eventType: event.event_type,
+        },
+        "Neo4j sync failed",
+      );
+    } finally {
+      await session.close();
+    }
+  }
+
+  private async syncEvent(
+    session: NonNullable<ReturnType<typeof getNeo4jSession>>,
+    event: EventEnvelope,
+  ): Promise<void> {
+    switch (event.event_type) {
+      case `edu.content.${EVENT_TYPES.KP_CREATED}`:
+      case `edu.content.${EVENT_TYPES.KP_UPDATED}`:
+        await this.upsertNode(session, event);
+        break;
+      case `edu.content.${EVENT_TYPES.KP_PREREQUISITE_ADDED}`:
+        await this.addPrerequisite(session, event);
+        break;
+      case `edu.content.${EVENT_TYPES.KP_PREREQUISITE_REMOVED}`:
+        await this.removePrerequisite(session, event);
+        break;
+      default:
+        logger.debug(
+          { eventType: event.event_type },
+          "Unhandled event type for Neo4j sync",
+        );
+    }
+  }
+
+  private async upsertNode(
+    session: NonNullable<ReturnType<typeof getNeo4jSession>>,
+    event: EventEnvelope,
+  ): Promise<void> {
+    await session.executeWrite((tx) =>
+      tx.run(
+        `MERGE (kp:KnowledgePoint {id: $id})
+         SET kp.title = $title, kp.difficulty = $difficulty, kp.updatedAt = datetime()`,
+        {
+          id: event.kp_id ?? event.aggregate_id,
+          title: event.title ?? "",
+          difficulty: event.difficulty ?? 3,
+        },
+      ),
+    );
+  }
+
+  private async addPrerequisite(
+    session: NonNullable<ReturnType<typeof getNeo4jSession>>,
+    event: EventEnvelope,
+  ): Promise<void> {
+    await session.executeWrite((tx) =>
+      tx.run(
+        `MATCH (kp:KnowledgePoint {id: $kpId}), (prereq:KnowledgePoint {id: $prereqId})
+         MERGE (prereq)-[:PREREQUISITE_OF]->(kp)`,
+        {
+          kpId: event.aggregate_id,
+          prereqId: event.prerequisite_id ?? "",
+        },
+      ),
+    );
+  }
+
+  private async removePrerequisite(
+    session: NonNullable<ReturnType<typeof getNeo4jSession>>,
+    event: EventEnvelope,
+  ): Promise<void> {
+    await session.executeWrite((tx) =>
+      tx.run(
+        `MATCH (prereq:KnowledgePoint {id: $prereqId})-[r:PREREQUISITE_OF]->(kp:KnowledgePoint {id: $kpId})
+         DELETE r`,
+        {
+          kpId: event.aggregate_id,
+          prereqId: event.prerequisite_id ?? "",
+        },
+      ),
+    );
+  }
+}
+
+export const neo4jSyncWorker = new Neo4jSyncWorker();