buzz-sheet/tests/security-redis.integration.test.ts

import { randomUUID } from "node:crypto";

import Redis from "ioredis";
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";

vi.mock("server-only", () => ({}));

const { closeRedisEventStreams, createRedisEventResponse } = await import("@/lib/events/redis-stream");
const { getLatestDeploymentStatus, publishDeploymentStatus } = await import("@/lib/deployment/repository");
const { updateVisitorPresence } = await import("@/lib/presence");
const { closeRedisClient, redisEventChannel } = await import("@/lib/redis/client");
const { consumeRateLimit } = await import("@/lib/security/rate-limit");

const integrationUrl = process.env.REDIS_INTEGRATION_URL;
const describeWithRedis = integrationUrl ? describe : describe.skip;

describeWithRedis("shared Redis security paths", () => {
  const runId = randomUUID();
  const original = {
    url: process.env.REDIS_URL,
    event: process.env.REDIS_EVENT_PREFIX,
    presence: process.env.REDIS_PRESENCE_PREFIX,
    security: process.env.REDIS_SECURITY_PREFIX,
  };
  let publisher: Redis;

  beforeAll(async () => {
    process.env.REDIS_URL = integrationUrl;
    process.env.REDIS_EVENT_PREFIX = `buzz:test:${runId}:events`;
    process.env.REDIS_PRESENCE_PREFIX = `buzz:test:${runId}:presence`;
    process.env.REDIS_SECURITY_PREFIX = `buzz:test:${runId}:security`;
    publisher = new Redis(integrationUrl!);
    await publisher.ping();
  });

  afterAll(async () => {
    closeRedisEventStreams();
    await closeRedisClient();
    if (publisher) {
      const keys = await publisher.keys(`buzz:test:${runId}:*`);
      if (keys.length) await publisher.del(...keys);
      await publisher.quit();
    }
    for (const [key, value] of Object.entries(original)) {
      const envName = { url: "REDIS_URL", event: "REDIS_EVENT_PREFIX", presence: "REDIS_PRESENCE_PREFIX", security: "REDIS_SECURITY_PREFIX" }[key]!;
      if (value === undefined) delete process.env[envName];
      else process.env[envName] = value;
    }
  });

  it("counts requests atomically across callers", async () => {
    const results = await Promise.all(Array.from({ length: 8 }, () => consumeRateLimit(`test:${runId}`, { window: 60, max: 5 })));
    expect(results.filter((result) => result.allowed)).toHaveLength(5);
    expect(results.filter((result) => !result.allowed)).toHaveLength(3);
  });

  it("tracks visitor counts without a separate Redis round trip per visitor", async () => {
    const first = await updateVisitorPresence({ tabId: `${runId}:1`, browserId: `${runId}:browser-1`, guideId: "guide-a", active: true });
    expect(first.global).toBe(1);
    expect(first.guide).toBe(1);
    const second = await updateVisitorPresence({ tabId: `${runId}:2`, browserId: `${runId}:browser-2`, guideId: "guide-a", active: true });
    expect(second.global).toBe(2);
    expect(second.guide).toBe(2);
    const count = await publisher.zcard(`${process.env.REDIS_PRESENCE_PREFIX}:visitors`);
    expect(count).toBe(2);
  });

  it("counts a browser once across tabs without losing either guide", async () => {
    const browserId = `${runId}:browser-3`;
    const firstTab = `${runId}:3`;
    const secondTab = `${runId}:4`;
    const firstGuide = await updateVisitorPresence({ tabId: firstTab, browserId, guideId: "guide-c", active: true });
    const secondGuide = await updateVisitorPresence({ tabId: secondTab, browserId, guideId: "guide-d", active: true });
    expect(secondGuide.global).toBe(firstGuide.global);
    expect(secondGuide.guide).toBe(1);
    expect(secondGuide.byGuide["guide-c"]).toBe(1);
    const sameGuide = await updateVisitorPresence({ tabId: secondTab, browserId, guideId: "guide-c", active: true });
    expect(sameGuide.guide).toBe(1);
    expect(sameGuide.byGuide["guide-d"]).toBeUndefined();
    const afterLeave = await updateVisitorPresence({ tabId: secondTab, browserId, guideId: "guide-d", active: false });
    expect(afterLeave.byGuide["guide-c"]).toBe(1);
    expect(afterLeave.byGuide["guide-d"]).toBeUndefined();
  });

  it("fans one subscribed topic out to two SSE clients", async () => {
    const topic = `page:${runId}`;
    const first = new AbortController();
    const second = new AbortController();
    const [responseA, responseB] = await Promise.all([
      createRedisEventResponse(topic, first.signal),
      createRedisEventResponse(topic, second.signal),
    ]);
    const readerA = responseA.body!.getReader();
    const readerB = responseB.body!.getReader();
    await Promise.all([readerA.read(), readerB.read()]);
    const payload = JSON.stringify({ type: "page.updated", id: runId, version: 1 });
    await publisher.publish(redisEventChannel(topic), payload);
    const [a, b] = await Promise.all([readerA.read(), readerB.read()]);
    const decoder = new TextDecoder();
    expect(decoder.decode(a.value)).toContain(payload);
    expect(decoder.decode(b.value)).toContain(payload);
    first.abort();
    second.abort();
    await Promise.all([readerA.cancel(), readerB.cancel()]);
  });

  it("does not let stale deployment results replace a newer announcement", async () => {
    const firstId = "a".repeat(40);
    const secondId = "b".repeat(40);
    expect((await publishDeploymentStatus(firstId, "deploying")).accepted).toBe(true);
    expect((await publishDeploymentStatus(secondId, "deploying")).accepted).toBe(true);
    expect((await publishDeploymentStatus(firstId, "ready")).accepted).toBe(false);
    expect(await getLatestDeploymentStatus()).toMatchObject({
      deploymentId: secondId,
      status: "deploying",
    });
    expect((await publishDeploymentStatus(secondId, "failed")).accepted).toBe(true);
  });
});