feat(api,domain): add batch commands endpoint and seed file tooling

packages/api/src/routes/sync.test.ts

@@ -7,6 +7,7 @@ import {
   productTypeHandler,
   stockItemHandler,
   tenantHandler,
+  type SeedFile,
 } from "@heim/domain";
 import { LocalKeyManagementService } from "../crypto/kms.ts";
 import { encryptPayload } from "../crypto/payload-encryption.ts";
@@ -467,3 +468,186 @@ describe("POST /api/sync/commands", () => {
     );
   });
 });
+
+describe("POST /api/sync/commands/batch", () => {
+  it("returns 401 when no session", async () => {
+    const client = makeClient();
+    const pool = makePool(client);
+    const app = makeApp(null, pool);
+
+    const body: SeedFile = {
+      version: 1,
+      commands: [
+        {
+          streamId: "pt-1",
+          streamType: "ProductType",
+          type: "CreateProductType",
+          payload: { name: "Oil" },
+        },
+      ],
+    };
+
+    const res = await supertest(app).post("/api/sync/commands/batch").send(body);
+
+    expect(res.status).toBe(401);
+    expect(res.body).toMatchObject({ error: "not_authenticated" });
+  });
+
+  it("returns 400 for empty commands array", async () => {
+    const client = makeClient();
+    const pool = makePool(client);
+    const app = makeApp(SESSION, pool);
+
+    const res = await supertest(app)
+      .post("/api/sync/commands/batch")
+      .send({ version: 1, commands: [] });
+
+    expect(res.status).toBe(400);
+    expect(res.body).toMatchObject({ error: "empty_commands" });
+  });
+
+  it("returns 400 for unsupported version", async () => {
+    const client = makeClient();
+    const pool = makePool(client);
+    const app = makeApp(SESSION, pool);
+
+    const res = await supertest(app)
+      .post("/api/sync/commands/batch")
+      .send({
+        version: 99,
+        commands: [
+          {
+            streamId: "pt-1",
+            streamType: "ProductType",
+            type: "CreateProductType",
+            payload: { name: "Oil" },
+          },
+        ],
+      });
+
+    expect(res.status).toBe(400);
+    expect(res.body).toMatchObject({ error: "unsupported_version" });
+  });
+
+  it("returns 400 for unknown stream type", async () => {
+    const client = makeClient();
+    const pool = makePool(client);
+    const app = makeApp(SESSION, pool);
+
+    const res = await supertest(app)
+      .post("/api/sync/commands/batch")
+      .send({
+        version: 1,
+        commands: [{ streamId: "x-1", streamType: "UnknownType", type: "Create", payload: {} }],
+      });
+
+    expect(res.status).toBe(400);
+    expect(res.body).toMatchObject({
+      error: "unknown_stream_type",
+      index: 0,
+      streamType: "UnknownType",
+    });
+  });
+
+  it("processes batch successfully with multiple commands", async () => {
+    const client = makeClient();
+    const pool = makePool(client);
+    const query = vi.mocked(client.query);
+
+    const app = makeApp(SESSION, pool);
+
+    const res = await supertest(app)
+      .post("/api/sync/commands/batch")
+      .send({
+        version: 1,
+        commands: [
+          {
+            streamId: "pt-1",
+            streamType: "ProductType",
+            type: "CreateProductType",
+            payload: { name: "Olive Oil", category: "pantry" },
+          },
+          {
+            streamId: "pt-2",
+            streamType: "ProductType",
+            type: "CreateProductType",
+            payload: { name: "Butter", category: "dairy" },
+          },
+        ],
+      });
+
+    expect(res.status).toBe(200);
+    expect(res.body).toMatchObject({ ok: true, imported: 2 });
+
+    // Verify transaction boundaries
+    const calls = query.mock.calls.map(([sql]) => {
+      const s = (sql as string).trim();
+      if (s === "BEGIN" || s === "COMMIT" || s === "ROLLBACK") return s;
+      if (s.startsWith("SELECT")) return "SELECT";
+      if (s.startsWith("INSERT")) return "INSERT";
+      return s;
+    });
+    expect(calls[0]).toBe("BEGIN");
+    expect(calls[calls.length - 1]).toBe("COMMIT");
+  });
+
+  it("rolls back entire batch when a command is rejected", async () => {
+    const client = makeClient();
+    const pool = makePool(client);
+    const query = vi.mocked(client.query);
+
+    // BEGIN
+    query.mockResolvedValueOnce({ rows: [] } as never);
+    // loadStreamEvents for first command — already has a ProductType
+    query.mockResolvedValueOnce({
+      rows: [
+        {
+          id: "evt-existing",
+          tenant_id: "tenant-1",
+          stream_id: "pt-1",
+          stream_type: "ProductType",
+          stream_position: 1,
+          event_type: "ProductTypeCreated",
+          correlation_id: "corr-0",
+          causation_id: "command:corr-0",
+          acting_principal_id: "principal-1",
+          effective_principal_id: null,
+          payload: { name: "Existing", category: null },
+          metadata: {},
+          actual_time: new Date(),
+        },
+      ],
+    } as never);
+    // ROLLBACK
+    query.mockResolvedValueOnce({ rows: [] } as never);
+
+    const app = makeApp(SESSION, pool);
+
+    // CreateProductType on existing stream → rejected
+    const res = await supertest(app)
+      .post("/api/sync/commands/batch")
+      .send({
+        version: 1,
+        commands: [
+          {
+            streamId: "pt-1",
+            streamType: "ProductType",
+            type: "CreateProductType",
+            payload: { name: "Duplicate" },
+          },
+        ],
+      });
+
+    expect(res.status).toBe(422);
+    expect(res.body).toMatchObject({
+      error: "command_rejected",
+      index: 0,
+      reason: "Product type already exists",
+    });
+
+    // Verify ROLLBACK was called, not COMMIT
+    const calls = query.mock.calls.map(([sql]) => (sql as string).trim());
+    expect(calls).toContain("ROLLBACK");
+    expect(calls).not.toContain("COMMIT");
+  });
+});

packages/api/src/routes/sync.ts

@@ -1,10 +1,12 @@
+import { randomUUID } from "node:crypto";
 import { Router } from "express";
 import type { Pool } from "pg";
 import {
   AGGREGATE_REGISTRY,
   buildAggregate,
   type Command,
   type CommandHandlerRegistry,
+  type SeedFile,
 } from "@heim/domain";
 import type { KeyManagementService } from "../crypto/kms.ts";
 import { appendEvents } from "../event-store/append-events.ts";
@@ -151,5 +153,92 @@ export function createSyncRouter(
     }
   });
 
+  router.post("/commands/batch", async (req, res) => {
+    if (!req.session) {
+      res.status(401).json({ error: "not_authenticated" });
+      return;
+    }
+
+    const body = req.body as SeedFile;
+
+    if (body.version !== 1) {
+      res.status(400).json({ error: "unsupported_version" });
+      return;
+    }
+
+    if (!Array.isArray(body.commands) || body.commands.length === 0) {
+      res.status(400).json({ error: "empty_commands" });
+      return;
+    }
+
+    const { tenantId, principalId } = req.session;
+    const correlationId = randomUUID();
+
+    const client = await pool.connect();
+    try {
+      await client.query("BEGIN");
+
+      let imported = 0;
+
+      for (let i = 0; i < body.commands.length; i++) {
+        const seedCmd = body.commands[i]!;
+
+        const config = AGGREGATE_REGISTRY[seedCmd.streamType];
+        if (!config) {
+          await client.query("ROLLBACK");
+          res.status(400).json({
+            error: "unknown_stream_type",
+            index: i,
+            streamType: seedCmd.streamType,
+          });
+          return;
+        }
+
+        const streamEvents = await loadStreamEvents(client, tenantId, seedCmd.streamId);
+        const aggregate = buildAggregate(config.initial, streamEvents, config.apply);
+
+        const commandId = randomUUID();
+        const command: Command = {
+          commandId,
+          correlationId,
+          causationId: `batch:${correlationId}`,
+          streamId: seedCmd.streamId,
+          streamType: seedCmd.streamType,
+          type: seedCmd.type,
+          payload: seedCmd.payload,
+          expectedVersion: aggregate.version,
+          actualTime: seedCmd.actualTime ? new Date(seedCmd.actualTime) : new Date(),
+          tenantId,
+          actingPrincipalId: principalId,
+          effectivePrincipalId: null,
+        };
+
+        const result = commandRegistry.handle(aggregate.state, command, config);
+
+        if (!result.ok) {
+          await client.query("ROLLBACK");
+          res.status(422).json({
+            error: "command_rejected",
+            index: i,
+            reason: result.reason,
+          });
+          return;
+        }
+
+        await appendEvents(client, [...result.events]);
+        await projectorRegistry.apply(client, result.events);
+        imported++;
+      }
+
+      await client.query("COMMIT");
+      res.json({ ok: true, imported });
+    } catch {
+      await client.query("ROLLBACK").catch(() => {});
+      res.status(500).json({ error: "internal_error" });
+    } finally {
+      client.release();
+    }
+  });
+
   return router;
 }

packages/domain/src/index.ts

@@ -72,4 +72,11 @@ export {
   type TenantMember,
   type TenantState,
 } from "./tenant/index.ts";
+export {
+  registerSeedConverter,
+  snapshotsToSeedFile,
+  type SeedCommand,
+  type SeedFile,
+  type SnapshotToSeedOptions,
+} from "./seed.ts";
 export { applyUserEvent, INITIAL_USER_STATE, type UserState } from "./user/index.ts";