suaveplan/packages/api-stack/api/api-plugin-event-bus/tests/integration/tenant-isolation.test.ts

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<string, unknown>();
    services.set("app", app);
    const plugin = createEventBusPlugin({
        mountRoutes: false,
        requireAuthForRoutes: false,
    });
    const ctx = createPluginContext<EventBusPluginConfig>({
        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<string, unknown>();
        services.set("app", app);
        const plugin = createEventBusPlugin({
            mountRoutes: false,
            requireAuthForRoutes: false,
        });
        const ctx = createPluginContext<EventBusPluginConfig>({
            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 }]);
    });
});