import { afterEach, describe, expect, test } from "bun:test";
import { KubeConfig } from "@kubernetes/client-node";
import http from "node:http";
import https from "node:https";
import { KubernetesLogs } from "../../server/kubernetes-logs";
import {
  createLogService,
  encodeLogEvent,
  KubernetesLogError,
  MANAGED_BY_SELECTOR,
  type ContainerLogRequest,
  type KubernetesDeployment,
  type KubernetesLogsBackend,
  type KubernetesPod,
  type KubernetesService,
  type LogEvent,
} from "../../server/log-service";

class FakeBackend implements KubernetesLogsBackend {
  deployments: KubernetesDeployment[] = [];
  service: KubernetesService | undefined;
  pods: KubernetesPod[] = [];
  reads = new Map<string, string | Error>();
  streams = new Map<
    string,
    (signal: AbortSignal) => AsyncIterable<string | Uint8Array>
  >();
  deploymentCalls: Array<Readonly<Record<string, string>>> = [];
  podSelectors: Array<Readonly<Record<string, string>>> = [];
  requests: ContainerLogRequest[] = [];

  async listDeployments(
    _namespace: string,
    labels: Readonly<Record<string, string>>,
  ) {
    this.deploymentCalls.push(labels);
    return this.deployments;
  }

  async getService(_namespace: string, _name: string) {
    return this.service;
  }

  async listPods(
    _namespace: string,
    selector: Readonly<Record<string, string>>,
  ) {
    this.podSelectors.push(selector);
    return this.pods;
  }

  async readContainerLogs(request: ContainerLogRequest) {
    this.requests.push(request);
    const value = this.reads.get(`${request.pod}/${request.container}`) ?? "";
    if (value instanceof Error) throw value;
    return value;
  }

  streamContainerLogs(request: ContainerLogRequest, signal: AbortSignal) {
    this.requests.push(request);
    const stream = this.streams.get(`${request.pod}/${request.container}`);
    if (stream) return stream(signal);
    return openStream(signal);
  }
}

async function* openStream(signal: AbortSignal): AsyncGenerator<string> {
  await untilAbort(signal);
  yield* [] as string[];
}

function untilAbort(signal: AbortSignal): Promise<void> {
  if (signal.aborted) return Promise.resolve();
  return new Promise((resolve) =>
    signal.addEventListener("abort", () => resolve(), { once: true }),
  );
}

async function* chunks(...values: Array<string | Uint8Array>) {
  for (const value of values) yield value;
}

async function nextOfType<T extends LogEvent["type"]>(
  iterator: AsyncIterator<LogEvent>,
  type: T,
  limit = 30,
): Promise<Extract<LogEvent, { type: T }>> {
  for (let index = 0; index < limit; index += 1) {
    const result = await iterator.next();
    if (result.done) throw new Error(`Stream ended before ${type}`);
    if (result.value.type === type)
      return result.value as Extract<LogEvent, { type: T }>;
  }
  throw new Error(`No ${type} event received`);
}

function setup() {
  const backend = new FakeBackend();
  backend.deployments = [{ name: "web", selector: { app: "web" } }];
  backend.pods = [
    {
      name: "web-abc",
      uid: "uid-1",
      phase: "Running",
      containers: ["web", "sidecar"],
    },
  ];
  const service = createLogService(
    backend,
    () => new Date("2026-09-02T12:00:00.000Z"),
  );
  return { backend, service };
}

const servers: Array<http.Server | https.Server> = [];
const tlsFixture = {
  key: Buffer.from(`-----BEGIN PRIVATE KEY-----
MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCUiNil5x0/YDDX
VWBT9hQQOldUzaRdCooc+Nx3CaZaKtjIlWDJQ24GXz+bJPi9kDhNTpoGAKy7EiMj
C+8pXS3ggnMEGIHF89eYzgsMlKUZeZ6x+q9blw04tdKRHRpA96Hps5MZ0FQFZ6p5
WzPqlyymvP8lAbeYhugekQjzqWfpmecNE3in7WpJH4XkM9w30u9mvjzV+5iyyIuD
HI27Hqujbqm2zkEN07EU1RQMTqTXGzlz7oX5bmfhG04w/47Jz1rLJ80qISWCmVD9
8hbRsg2j67tHWMOoBhOa7qh1Tme67LCNEFdbFuc5UD1mP+22K9D/zCC2RtJQu5Sq
dEJW3iovAgMBAAECggEAD2K3ckPk0y4/EOcOkdPhEyc/6ZBdkKepU8PxbkEpIpji
mLBkdKSP7ogKOiNTwqsAMf3M1YdXXQ9NZXF0hgfZWzKYAFobgyo1cGYTXeu9yExB
RHVPmcClRXUMCS0HDai49FC+EYPzWBX7YhOw5oFfRiw4j5hEcL+0ponmb/rhwSAg
AGGIM5nGBT3wpXcbOubdxe8/Ex7vg+refUrDGs4T7sbYijoWm5utz3bdc/JxLDf0
KwMesf7RZ+eeO2C6qOB4gKNjZr5JxApcJZ2WoG9MdWkaYnU8Kyquf2vhJdDZmqtx
kV0y8KSWiPTbVqXZOYcg5AKcymdF7P04zVe4rkEuoQKBgQDRDz9jem5r9U25J6VT
Oi7oWZmgsDcchfeerRBhQmEPiASkOFu2VgiM3CeUeGnNmrRlRsKUEoS00Bph8AMk
z0SfnBTFjIIiB9gERnqrH1vSnCbLhvnhkd7XVtoIG3JjXnJgALoIkmp9DM82VTFu
kBaPpOx1i5vQyI68xpAUNhjVjwKBgQC14p5qxrwn5y0VQg/9MkgLUfbsROqRX8SP
O1RRhR8teA8lL+jnTzq7HauPMnP8kEjfMHXjV124eDLKUHLrOoNwTg7S0MmgYcRU
DWaOCA5u5MpHxCfz2h+OGDVp0FP/QCyLwu6vjkvK+dMhvzK92QHzTqLsia2Tpspx
3HFLEm1RYQKBgG44E7tmuQDB+5A6jrcqXcCyPISzYtru5nYJ2DDuxi1iENBjxjaD
dU6OY2+rbFyxy5n5jGx0tvJ9JOutlnq5q/xaVbkxMwquB/15CwNdLRQEr49uQh/i
wBHYAGt1zQEGslZbC7mpN+tl7Xk/wSgBX2OsF96BFE0m79om9Z8yRjWRAoGAAeBI
iglqv26fBG0eBRqTq6o4xc8gLEe0m1WdVQnufGWUommQGXKzxGJV9rAqihxi5Ap3
7NRl3xU+UN/rj4mW+X2UoZANxF29zLAmsqhancI2Y+8eCmHhmXGee2zusN9Ulkx4
cc8h8QIKr3ptZ4/peT0CaTYyWCeMRwhjEscp4YECgYBlQyBkjrhBLswIK+Avul0C
heCCZkDZstdkYTlAh6+tnf0ycMEKXPF9T1O67fSRmzQnTuvb6VWgtMh3zvOMy+Ce
UnVMOhKLS0wPPTZc75kKju1C2hHZ2eSgOFbNFrQvWJMXGNjEvdscXjiCoCBbhBt+
Pc3eI1dwQpPwp363CGPLrg==
-----END PRIVATE KEY-----\n`),
  cert: Buffer.from(`-----BEGIN CERTIFICATE-----
MIIC3DCCAcSgAwIBAgIIMGH60RwCdvswDQYJKoZIhvcNAQEMBQAwFDESMBAGA1UE
AxMJMTI3LjAuMC4xMB4XDTI2MDkyNzEwMjUyN1oXDTM2MDkyNDEwMjUyN1owFDES
MBAGA1UEAxMJMTI3LjAuMC4xMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKC
AQEAlIjYpecdP2Aw11VgU/YUEDpXVM2kXQqKHPjcdwmmWirYyJVgyUNuBl8/myT4
vZA4TU6aBgCsuxIjIwvvKV0t4IJzBBiBxfPXmM4LDJSlGXmesfqvW5cNOLXSkR0a
QPeh6bOTGdBUBWeqeVsz6pcsprz/JQG3mIboHpEI86ln6ZnnDRN4p+1qSR+F5DPc
N9LvZr481fuYssiLgxyNux6ro26pts5BDdOxFNUUDE6k1xs5c+6F+W5n4RtOMP+O
yc9ayyfNKiElgplQ/fIW0bINo+u7R1jDqAYTmu6odU5nuuywjRBXWxbnOVA9Zj/t
tivQ/8wgtkbSULuUqnRCVt4qLwIDAQABozIwMDAdBgNVHQ4EFgQUeDmjpd05dSFB
yENQKuBjSFVItWQwDwYDVR0RBAgwBocEfwAAATANBgkqhkiG9w0BAQwFAAOCAQEA
SZLDBqpXGPrmKfz3qvPdda3wJUp5MAvy1ZEA2IAMHMutUGemIcEXoU+4qd8aNhzz
EVS2gKsaFZ59MXqC0xhRnNreqP3lP5lPVUV+EAOXvAF+vrg07KBRrxYuE2hzgKxl
FP6swGVsMGyWF266PFe4p3QhcxSUvLBtCuYyJTEvL5FDhByE5M+F7zh5RFiiHvx5
txljZfJcP0wDZVUalNGbJR1nXJ7GRVPO92ec38ly6PIvlR/BAvNCoLdIF69C12qC
c+5Sp92YbZ6+Lq1p75M6mY44A8FSXE40x3HXM8HcpeatXOxZuM3YAkrO337AgEx0
9CZrJPz0ekLRl4UW5LG8mA==
-----END CERTIFICATE-----\n`),
};
const unrelatedCa = Buffer.from(`-----BEGIN CERTIFICATE-----
MIICwzCCAaugAwIBAgIISfrhnhbLiWQwDQYJKoZIhvcNAQEMBQAwEDEOMAwGA1UE
AxMFd3JvbmcwHhcNMjYwOTI3MTAyNzAyWhcNMzYwOTI0MTAyNzAyWjAQMQ4wDAYD
VQQDEwV3cm9uZzCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAK4MZhty
U3AIPjb92r7F7tsFX3UUo9fdZUeLjli2a97DZTiO/hs7a/oce27MlXq0BcpP6Rm5
d0kCIb6IU7SgP3sH8dovaLnC1Xv9KdyFnkRp0eXD6xExmZ9Y6ftDHRkEI+la41ai
GdRp/ebLnTaRLxRAQUpd2xCvaa/weenGDxKFNNunVncF9WzAWYhUZqFR1wXBtGo0
0AAuzuxYoenyFLHoozeoBJ7Cr23aR/MP0zWQK3blz6TZ7fTcZtDkgsThga+6F8O+
OJGM7di6FRx7O37DRK6yVphgI1nClbEmw/sAtqc3MYQAh6CzoRFbceGQ6QEdGjq3
zrv7tJhjQ2x/CNECAwEAAaMhMB8wHQYDVR0OBBYEFKIQJVhaqJhy9cwoA1lCUPff
O//lMA0GCSqGSIb3DQEBDAUAA4IBAQCQHcUDcgLie6wFWpCUFgcT0UMBcZ7FL3me
3vmvOv7T/ydOJGbgaKqz+c3NdrApsL9Jk0wmh0YelYK7TuhZnsNJ0pbehe509vJ9
10B8bKNmZqgVgdTppsaw9UDFcEQn1UMhkhtYs51+acVntNMXz8/qEhrto8mGdPKO
39ydfekUA6fAw9Q8F/sRtvG7bB6Cmj/8Tieq/dJjO7ZXgqY6HaYrX9qQxlapRu2l
LgdidPbFADDoUsSXBVqX3F4Il+qRQrrGmB/TPPbGCsW/A0TA2jOPHGoh3mczjEm7
UDgBJ/wg7flQRRYcz278Hm+71JrSOEdpC4F1ssbttKE5Jhg+91uU
-----END CERTIFICATE-----\n`);

async function kubernetesHttpsServer(
  tls: typeof tlsFixture,
  options: { ca?: Buffer; skipTLSVerify?: boolean },
  handler: http.RequestListener,
): Promise<KubernetesLogs> {
  const server = https.createServer(tls, handler);
  servers.push(server);
  await new Promise<void>((resolve, reject) => {
    server.once("error", reject);
    server.listen(0, "127.0.0.1", resolve);
  });
  const address = server.address();
  if (!address || typeof address === "string") throw new Error("Missing port");
  const config = new KubeConfig();
  config.loadFromOptions({
    clusters: [{
      name: "test",
      server: `https://127.0.0.1:${address.port}`,
      ...(options.ca && { caData: options.ca.toString("base64") }),
      ...(options.skipTLSVerify && { skipTLSVerify: true }),
    }],
    users: [{ name: "test" }],
    contexts: [{ name: "test", cluster: "test", user: "test" }],
    currentContext: "test",
  });
  return new KubernetesLogs(config);
}

async function kubernetesServer(
  handler: http.RequestListener,
): Promise<KubernetesLogs> {
  const server = http.createServer(handler);
  servers.push(server);
  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
  const address = server.address();
  if (!address || typeof address === "string") throw new Error("Missing port");
  const config = new KubeConfig();
  config.loadFromOptions({
    clusters: [
      {
        name: "test",
        server: `http://127.0.0.1:${address.port}`,
        skipTLSVerify: true,
      },
    ],
    users: [{ name: "test" }],
    contexts: [{ name: "test", cluster: "test", user: "test" }],
    currentContext: "test",
  });
  return new KubernetesLogs(config);
}

afterEach(async () => {
  await Promise.all(
    servers.splice(0).map(
      (server) =>
        new Promise<void>((resolve, reject) =>
          server.listening
            ? server.close((error) => (error ? reject(error) : resolve()))
            : resolve(),
        ),
    ),
  );
});

describe("LogService", () => {
  test("enumerates every container in managed deployment pods", async () => {
    const { backend, service } = setup();
    expect(
      await service.listContainers({
        namespace: "demo",
        target: { kind: "managed-deployments" },
      }),
    ).toEqual([
      {
        namespace: "demo",
        targetKind: "deployment",
        targetName: "web",
        pod: "web-abc",
        podUid: "uid-1",
        container: "web",
      },
      {
        namespace: "demo",
        targetKind: "deployment",
        targetName: "web",
        pod: "web-abc",
        podUid: "uid-1",
        container: "sidecar",
      },
    ]);
    expect(backend.deploymentCalls).toEqual([MANAGED_BY_SELECTOR]);
    expect(backend.podSelectors).toEqual([{ app: "web" }]);
  });

  test("uses a named service selector without listing deployments", async () => {
    const { backend, service } = setup();
    backend.service = { name: "frontend", selector: { role: "frontend" } };
    const result = await service.listContainers({
      namespace: "demo",
      target: { kind: "service", name: "frontend" },
    });
    expect(result[0]).toMatchObject({
      targetKind: "service",
      targetName: "frontend",
    });
    expect(backend.deploymentCalls).toHaveLength(0);
    expect(backend.podSelectors).toEqual([{ role: "frontend" }]);
  });

  test("rejects missing and selectorless services", async () => {
    const { backend, service } = setup();
    const options = {
      namespace: "demo",
      target: { kind: "service" as const, name: "missing" },
    };
    await expect(service.listContainers(options)).rejects.toMatchObject({
      retryable: false,
    });
    backend.service = { name: "missing", selector: {} };
    await expect(service.listContainers(options)).rejects.toThrow(
      "no pod selector",
    );
  });

  test("collects timestamped structured lines and passes log options", async () => {
    const { backend, service } = setup();
    backend.reads.set(
      "web-abc/web",
      "2026-09-02T11:59:00.123456Z first\nsecond\n",
    );
    backend.reads.set("web-abc/sidecar", "side\n");
    const events = await service.collect({
      namespace: "demo",
      target: { kind: "managed-deployments" },
      tailLines: 25,
      sinceTime: new Date("2026-09-02T11:00:00Z"),
      timestamps: true,
    });
    expect(events).toHaveLength(3);
    expect(events[0]).toMatchObject({
      type: "log",
      timestamp: "2026-09-02T12:00:00.000Z",
      logTimestamp: "2026-09-02T11:59:00.123456Z",
      message: "first",
      pod: "web-abc",
      container: "web",
    });
    expect(events[1]).toMatchObject({ message: "second" });
    expect(backend.requests[0]).toEqual({
      namespace: "demo",
      pod: "web-abc",
      container: "web",
      tailLines: 25,
      sinceSeconds: undefined,
      sinceTime: "2026-09-02T11:00:00.000Z",
      timestamps: true,
    });
    expect(JSON.parse(encodeLogEvent(events[0]!))).toEqual(events[0]);
  });

  test("passes a line count through a followed Kubernetes log request", async () => {
    let received: URL | undefined;
    const kubernetes = await kubernetesServer((request, response) => {
      received = new URL(request.url!, "http://localhost");
      response.writeHead(200);
      response.end("latest\n");
    });
    const controller = new AbortController();
    const iterator = kubernetes.streamContainerLogs(
      {
        namespace: "demo",
        pod: "web-abc",
        container: "web",
        tailLines: 12,
        timestamps: false,
      },
      controller.signal,
    );
    const chunks: Uint8Array[] = [];
    for await (const chunk of iterator) chunks.push(chunk);
    expect(new TextDecoder().decode(Buffer.concat(chunks))).toBe("latest\n");
    expect(received?.searchParams.get("tailLines")).toBe("12");
    expect(received?.searchParams.get("follow")).toBe("true");
  });

  test("streams logs from HTTPS trusted by the configured cluster CA", async () => {
    const tls = tlsFixture;
    const kubernetes = await kubernetesHttpsServer(
      tls,
      { ca: tls.cert },
      (_request, response) => response.end("trusted\n"),
    );
    const controller = new AbortController();
    const chunks: Uint8Array[] = [];
    for await (const chunk of kubernetes.streamContainerLogs({
      namespace: "demo", pod: "web-abc", container: "web", timestamps: false,
    }, controller.signal)) chunks.push(chunk);
    expect(Buffer.concat(chunks).toString()).toBe("trusted\n");
  });

  test("rejects HTTPS logs with an untrusted cluster CA", async () => {
    const tls = tlsFixture;
    const kubernetes = await kubernetesHttpsServer(
      tls,
      { ca: unrelatedCa },
      (_request, response) => response.end("should not be reached\n"),
    );
    const controller = new AbortController();
    await expect(async () => {
      for await (const _chunk of kubernetes.streamContainerLogs({
        namespace: "demo", pod: "web-abc", container: "web", timestamps: false,
      }, controller.signal)) { /* consume until TLS rejects */ }
    }).toThrow();
  });

  test("allows an untrusted HTTPS log certificate only with explicit opt-in", async () => {
    const tls = tlsFixture;
    const kubernetes = await kubernetesHttpsServer(
      tls,
      { skipTLSVerify: true },
      (_request, response) => response.end("opted in\n"),
    );
    const controller = new AbortController();
    const chunks: Uint8Array[] = [];
    for await (const chunk of kubernetes.streamContainerLogs({
      namespace: "demo", pod: "web-abc", container: "web", timestamps: false,
    }, controller.signal)) chunks.push(chunk);
    expect(Buffer.concat(chunks).toString()).toBe("opted in\n");
  });

  test("returns per-container safe errors and continues collection", async () => {
    const { backend, service } = setup();
    backend.reads.set(
      "web-abc/web",
      new KubernetesLogError("container unavailable", false),
    );
    backend.reads.set("web-abc/sidecar", "healthy");
    const events = await service.collect({
      namespace: "demo",
      target: { kind: "managed-deployments" },
    });
    expect(events[0]).toEqual({
      type: "error",
      timestamp: "2026-09-02T12:00:00.000Z",
      namespace: "demo",
      pod: "web-abc",
      container: "web",
      message: "container unavailable",
      retryable: false,
    });
    expect(events[1]).toMatchObject({ type: "log", message: "healthy" });
    expect(JSON.stringify(events)).not.toContain("env");
    expect(JSON.stringify(events)).not.toContain("Secret");
  });

  test("validates tail and since options before Kubernetes calls", async () => {
    const { backend, service } = setup();
    await expect(
      service.collect({
        namespace: "demo",
        target: { kind: "managed-deployments" },
        tailLines: -1,
      }),
    ).rejects.toThrow("tailLines");
    await expect(
      service.collect({
        namespace: "demo",
        target: { kind: "managed-deployments" },
        sinceSeconds: 5,
        sinceTime: "2026-09-02T00:00:00Z",
      }),
    ).rejects.toThrow("mutually exclusive");
    expect(backend.deploymentCalls).toHaveLength(0);
  });

  test("follows chunked lines and stops active streams on abort", async () => {
    const { backend, service } = setup();
    let cancelled = false;
    backend.streams.set("web-abc/web", (signal) =>
      (async function* () {
        try {
          yield new TextEncoder().encode("hel");
          yield "lo\npartial";
          await untilAbort(signal);
        } finally {
          cancelled = true;
        }
      })(),
    );
    const controller = new AbortController();
    const iterator = service.follow({
      namespace: "demo",
      target: { kind: "managed-deployments" },
      signal: controller.signal,
      discoveryIntervalMs: 5,
      heartbeatIntervalMs: 1_000,
    });
    expect(await nextOfType(iterator, "log")).toMatchObject({
      message: "hello",
    });
    controller.abort();
    expect((await iterator.next()).done).toBe(true);
    expect(cancelled).toBe(true);
  });

  test("discards queued lines immediately on external abort", async () => {
    const { backend, service } = setup();
    backend.pods[0] = { ...backend.pods[0]!, containers: ["web"] };
    backend.streams.set("web-abc/web", () => chunks("one\ntwo\nthree\n"));
    const controller = new AbortController();
    const iterator = service.follow({
      namespace: "demo",
      target: { kind: "managed-deployments" },
      signal: controller.signal,
      queueCapacity: 5,
      discoveryIntervalMs: 1_000,
      heartbeatIntervalMs: 1_000,
    });
    expect(await nextOfType(iterator, "log")).toMatchObject({ message: "one" });
    await Bun.sleep(1);
    controller.abort();
    expect((await iterator.next()).done).toBe(true);
  });

  test("discovers replacement pod UIDs while following", async () => {
    const { backend, service } = setup();
    backend.pods[0] = { ...backend.pods[0]!, containers: ["web"] };
    backend.streams.set("web-abc/web", () => chunks("old\n"));
    const controller = new AbortController();
    const iterator = service.follow({
      namespace: "demo",
      target: { kind: "managed-deployments" },
      signal: controller.signal,
      discoveryIntervalMs: 5,
      heartbeatIntervalMs: 1_000,
    });
    expect(await nextOfType(iterator, "log")).toMatchObject({
      message: "old",
      podUid: "uid-1",
    });
    backend.pods = [
      { name: "web-new", uid: "uid-2", containers: ["web"], phase: "Running" },
    ];
    backend.streams.set("web-new/web", () => chunks("new\n"));
    expect(await nextOfType(iterator, "log")).toMatchObject({
      message: "new",
      pod: "web-new",
      podUid: "uid-2",
    });
    controller.abort();
    await iterator.next();
  });

  test("emits heartbeats and retryable stream errors before retrying", async () => {
    const { backend, service } = setup();
    backend.pods[0] = { ...backend.pods[0]!, containers: ["web"] };
    let attempts = 0;
    backend.streams.set("web-abc/web", () =>
      (async function* () {
        attempts += 1;
        if (attempts === 1) throw new Error("temporary disconnect");
        yield "recovered\n";
      })(),
    );
    const controller = new AbortController();
    const iterator = service.follow({
      namespace: "demo",
      target: { kind: "managed-deployments" },
      signal: controller.signal,
      discoveryIntervalMs: 5,
      heartbeatIntervalMs: 2,
      retryIntervalMs: 7,
    });
    expect(await nextOfType(iterator, "error")).toMatchObject({
      message: "temporary disconnect",
      retryable: true,
      retryAfterMs: 7,
      pod: "web-abc",
    });
    expect(await nextOfType(iterator, "heartbeat")).toMatchObject({
      timestamp: "2026-09-02T12:00:00.000Z",
    });
    expect(await nextOfType(iterator, "log")).toMatchObject({
      message: "recovered",
    });
    controller.abort();
    await iterator.next();
    expect(attempts).toBe(2);
  });

  test("emits a coded TLS failure once without retrying", async () => {
    const { backend, service } = setup();
    backend.pods[0] = { ...backend.pods[0]!, containers: ["web"] };
    let attempts = 0;
    backend.streams.set("web-abc/web", () =>
      (async function* () {
        attempts += 1;
        yield* [] as string[];
        const error = new Error("self signed certificate in certificate chain") as Error & {
          code: string;
        };
        error.code = "SELF_SIGNED_CERT_IN_CHAIN";
        throw error;
      })(),
    );
    const controller = new AbortController();
    const iterator = service.follow({
      namespace: "demo",
      target: { kind: "managed-deployments" },
      signal: controller.signal,
      discoveryIntervalMs: 5,
      heartbeatIntervalMs: 1_000,
    });
    expect(await nextOfType(iterator, "error")).toMatchObject({
      message: "self signed certificate in certificate chain",
      retryable: false,
    });
    expect(await nextOfType(iterator, "heartbeat")).toMatchObject({
      type: "heartbeat",
    });
    expect(attempts).toBe(1);
    controller.abort();
    expect((await iterator.next()).done).toBe(true);
    expect(attempts).toBe(1);
  });

  test("reports a broken upstream log connection per container without losing other streams", async () => {
    const urls: URL[] = [];
    const kubernetes = await kubernetesServer((request, response) => {
      const url = new URL(request.url!, "http://localhost");
      urls.push(url);
      if (url.searchParams.get("container") === "web") {
        request.socket.destroy();
      } else {
        response.writeHead(200);
        response.write("sidecar healthy\n");
      }
    });
    const { backend, service } = setup();
    backend.streams.set("web-abc/web", (signal) =>
      kubernetes.streamContainerLogs(
        { namespace: "demo", pod: "web-abc", container: "web", timestamps: false },
        signal,
      ),
    );
    backend.streams.set("web-abc/sidecar", (signal) =>
      kubernetes.streamContainerLogs(
        { namespace: "demo", pod: "web-abc", container: "sidecar", timestamps: false },
        signal,
      ),
    );
    const controller = new AbortController();
    const iterator = service.follow({
      namespace: "demo",
      target: { kind: "managed-deployments" },
      signal: controller.signal,
      discoveryIntervalMs: 1_000,
      heartbeatIntervalMs: 1_000,
      retryIntervalMs: 1_000,
    });
    try {
      const events = await Promise.all([iterator.next(), iterator.next()]);
      expect(events.map(({ value }) => value?.type).sort()).toEqual([
        "error",
        "log",
      ]);
      const error = events.find(({ value }) => value?.type === "error")!.value;
      expect(error).toMatchObject({
        pod: "web-abc",
        container: "web",
        retryable: true,
      });
      expect(error?.message).toMatch(
        /socket.*connection.*closed|socket hang up|fetch failed/i,
      );
      expect(error?.message).not.toContain("Error occurred in log request");
      expect(
        events.find(({ value }) => value?.type === "log")!.value,
      ).toMatchObject({ container: "sidecar", message: "sidecar healthy" });
      expect(urls).toHaveLength(2);
      expect(urls[0]!.searchParams.get("follow")).toBe("true");
    } finally {
      controller.abort();
      await iterator.return(undefined);
    }
  });

  test.each([
    [
      403,
      JSON.stringify({ kind: "Status", message: "logs forbidden" }),
      "logs forbidden",
      false,
    ],
    [500, "proxy unavailable", "proxy unavailable", true],
  ])(
    "keeps upstream HTTP %i status and body for follow errors",
    async (status, body, detail, retryable) => {
      const kubernetes = await kubernetesServer((_request, response) => {
        response.writeHead(status);
        response.end(body);
      });
      const read = async () => {
        for await (const _chunk of kubernetes.streamContainerLogs(
          {
            namespace: "demo",
            pod: "web-abc",
            container: "web",
            timestamps: false,
          },
          new AbortController().signal,
        )) {
          // The upstream response must fail before yielding log data.
        }
      };
      await expect(read()).rejects.toMatchObject({
        message: `Kubernetes log request failed (HTTP ${status}): ${detail}`,
        retryable,
      });
    },
  );

  test("aborting a follow request closes the upstream stream", async () => {
    let disconnected!: () => void;
    const closed = new Promise<void>((resolve) => (disconnected = resolve));
    const kubernetes = await kubernetesServer((_request, response) => {
      response.on("close", disconnected);
      response.writeHead(200);
      response.write("first\n");
    });
    const controller = new AbortController();
    const iterator = kubernetes.streamContainerLogs(
      { namespace: "demo", pod: "web-abc", container: "web", timestamps: false },
      controller.signal,
    )[Symbol.asyncIterator]();
    expect(new TextDecoder().decode((await iterator.next()).value)).toBe(
      "first\n",
    );
    controller.abort();
    expect((await iterator.next()).done).toBe(true);
    await closed;
  });

  test("emits one terminal target error and closes follow mode", async () => {
    const { service } = setup();
    const iterator = service.follow({
      namespace: "demo",
      target: { kind: "service", name: "missing" },
      signal: new AbortController().signal,
      discoveryIntervalMs: 5,
      heartbeatIntervalMs: 5,
    });
    expect(await nextOfType(iterator, "error")).toMatchObject({
      message: "Service missing was not found",
      retryable: false,
    });
    expect((await iterator.next()).done).toBe(true);
  });

  test("applies bounded-queue backpressure to fast producers", async () => {
    const { backend, service } = setup();
    backend.pods[0] = { ...backend.pods[0]!, containers: ["web"] };
    let produced = 0;
    backend.streams.set("web-abc/web", () =>
      (async function* () {
        for (let index = 0; index < 20; index += 1) {
          produced += 1;
          yield `${index}\n`;
        }
      })(),
    );
    const controller = new AbortController();
    const iterator = service.follow({
      namespace: "demo",
      target: { kind: "managed-deployments" },
      signal: controller.signal,
      queueCapacity: 1,
      discoveryIntervalMs: 1_000,
      heartbeatIntervalMs: 1_000,
    });
    expect((await iterator.next()).value).toMatchObject({ message: "0" });
    await Bun.sleep(10);
    expect(produced).toBeLessThanOrEqual(3);
    controller.abort();
    await iterator.return?.(undefined);
  });
});