import { OpenAPIHono } from "@hono/zod-openapi"; import { createPluginContext } from "@suaveplan/api-plugin"; import { EventBus } from "@suaveplan/events"; import { describe, expect, it } from "@suaveplan/testing/runner"; import { createEventBusPlugin } from "../../src/plugin/plugin.js"; import type { EventBusService } from "../../src/services/event-bus-service.types.js"; import type { EventBusPluginConfig } from "../../src/types/types.js"; const noopLogger = { debug: () => undefined, info: () => undefined, warn: () => undefined, error: () => undefined, }; async function bootPlugin() { const app = new OpenAPIHono(); const services = new Map(); services.set("app", app); const plugin = createEventBusPlugin({ mountRoutes: false, requireAuthForRoutes: false, }); const ctx = createPluginContext({ logger: noopLogger, eventBus: new EventBus(), signal: new AbortController().signal, config: plugin.defaults as EventBusPluginConfig, services, }); await plugin.onInit?.(ctx); return { app, service: services.get("eventBusService") as EventBusService, }; } describe("tenant isolation — end-to-end against the real plugin lifecycle", () => { it("tenant A publish does not reach tenant B subscriber", { timeout: 30_000, }, async () => { const { service } = await bootPlugin(); const received: string[] = []; service.subscribe( "order.created", () => { received.push("B"); }, { tenantId: "tenant-B" } ); service.subscribe( "order.created", () => { received.push("A"); }, { tenantId: "tenant-A" } ); const r = await service.publish( "order.created", { id: "o1" }, { tenantId: "tenant-A" } ); expect(received).toEqual(["A"]); expect(r.delivered).toBe(1); expect(r.dropped).toBe(false); }); it("tenant B cannot unsubscribe tenant A's subscription", { timeout: 30_000, }, async () => { const { service } = await bootPlugin(); const handleA = service.subscribe("order.created", () => undefined, { tenantId: "tenant-A", }); // tenant B attempt — throws TENANT_MISMATCH. expect(() => service.unsubscribe(handleA.id, "tenant-B")).toThrow(); // tenant A's subscription still active. const received: number[] = []; service.subscribe( "order.created", () => { received.push(1); }, { tenantId: "tenant-A" } ); await service.publish("order.created", {}, { tenantId: "tenant-A" }); expect(received.length).toBe(1); }); it("listSubscriptions filters by calling tenant", { timeout: 30_000, }, async () => { const { service } = await bootPlugin(); service.subscribe("a", () => undefined, { tenantId: "tenant-A" }); service.subscribe("a", () => undefined, { tenantId: "tenant-A" }); service.subscribe("a", () => undefined, { tenantId: "tenant-B" }); const aList = service.listSubscriptions({ tenantId: "tenant-A" }); expect(aList.length).toBe(2); expect(aList.every((s) => s.tenantId === "tenant-A")).toBe(true); }); it("idempotent subscription fires handler once for duplicate publish", { timeout: 30_000, }, async () => { const { service } = await bootPlugin(); const seen: number[] = []; service.subscribe( "order.completed", () => { seen.push(1); }, { tenantId: "tenant-A", idempotencyKey: (p) => String((p as { id: string }).id), } ); await service.publish( "order.completed", { id: "o1" }, { tenantId: "tenant-A" } ); await service.publish( "order.completed", { id: "o1" }, { tenantId: "tenant-A" } ); await service.publish( "order.completed", { id: "o2" }, { tenantId: "tenant-A" } ); expect(seen.length).toBe(2); }); it("c.set('eventBus') handle round-trips publishes between handlers", { timeout: 30_000, }, async () => { // Register the tenant middleware BEFORE the plugin's // requestBus middleware so the latter observes the tenant // when it runs (Hono dispatches middleware in registration // order, so a tenant setter registered after the plugin // middleware would be invisible to it). const app = new OpenAPIHono(); app.use("*", async (c, next) => { c.set("tenant" as never, { id: "tenant-A" } as never); await next(); }); const services = new Map(); services.set("app", app); const plugin = createEventBusPlugin({ mountRoutes: false, requireAuthForRoutes: false, }); const ctx = createPluginContext({ logger: noopLogger, eventBus: new EventBus(), signal: new AbortController().signal, config: plugin.defaults as EventBusPluginConfig, services, }); await plugin.onInit?.(ctx); const service = services.get("eventBusService") as EventBusService; const received: unknown[] = []; service.subscribe( "ping", (p) => { received.push(p); }, { tenantId: "tenant-A" } ); app.get("/publish", async (c) => { const handle = c.get("eventBus"); await handle.publish("ping", { hello: 1 }); return c.text("ok"); }); const res = await app.request("/publish"); expect(res.status).toBe(200); expect(received).toEqual([{ hello: 1 }]); }); });