import {
  CONSULTATION_DB_TTL_MS,
  DurableAudioSessionManager,
} from "../src/services/transcription/recovery/DurableAudioSessionManager";
import {
  dropConsultationDb,
  getConsultationDb,
} from "../src/services/transcription/recovery/db/ConsultationDexieDb";
import { flushMicrotasks } from "./helpers/testUtils";

const createAudioPayload = (byteLength = 2048): ArrayBuffer =>
  new Uint8Array(byteLength).fill(1).buffer;

const clearConsultationDb = async () => {
  await dropConsultationDb();
};

const blockDurableTransactions = () => {
  const db = getConsultationDb();
  let releaseWrites!: () => void;
  const writesReleased = new Promise<void>((resolve) => {
    releaseWrites = resolve;
  });

  const originalTransaction = (db.transaction as any).bind(db);
  let blockedTransactionCount = 0;
  const activeBlockedTransactions = new Set<number>();
  let nextBlockedTransactionId = 1;

  jest.spyOn(db as any, "transaction").mockImplementation(
    async (mode: any, ...args: any[]) => {
      const tables = args.slice(0, -1);
      const touchesDurableStores =
        mode === "rw" &&
        (tables.includes(db.manifests) ||
          tables.includes(db.segments) ||
          tables.includes(db.payloads));

      if (!touchesDurableStores) {
        return originalTransaction.apply(db, [mode, ...args]);
      }

      const blockedId = nextBlockedTransactionId;
      nextBlockedTransactionId += 1;
      blockedTransactionCount += 1;
      activeBlockedTransactions.add(blockedId);

      await writesReleased;

      try {
        return await originalTransaction.apply(db, [mode, ...args]);
      } finally {
        activeBlockedTransactions.delete(blockedId);
      }
    }
  );

  return {
    db,
    releaseWrites,
    getBlockedTransactionCount: () => blockedTransactionCount,
    getActiveBlockedTransactions: () => activeBlockedTransactions.size,
  };
};

describe("DurableAudioSessionManager", () => {
  beforeEach(async () => {
    await clearConsultationDb();
  });

  afterEach(async () => {
    jest.restoreAllMocks();
    jest.useRealTimers();
    await clearConsultationDb();
  });

  it("clears all consultation buckets when a new session starts", async () => {
    const db = getConsultationDb();
    const now = 1_000_000;
    const staleUpdatedAt = now - 10_000;
    const freshUpdatedAt = now - 1_000;

    await db.manifests.bulkPut([
      {
        sessionId: "stale-session",
        streamEpoch: "stale",
        startedAt: staleUpdatedAt,
        sampleRate: 16000,
        channels: 1,
        encoding: "pcm_s16le",
        totalSegments: 1,
        totalBytes: 32,
        totalDurationMs: 50,
        lastAssignedSeq: 1,
        lastDurableSeq: 1,
        lastSentSeq: 1,
        hardCommittedMs: 0,
        softCommittedMs: 0,
        replayInFlight: false,
        reconnectCount: 0,
        archiveState: "capturing",
        createdAt: staleUpdatedAt,
        updatedAt: staleUpdatedAt,
      },
      {
        sessionId: "fresh-session",
        streamEpoch: "fresh",
        startedAt: freshUpdatedAt,
        sampleRate: 16000,
        channels: 1,
        encoding: "pcm_s16le",
        totalSegments: 1,
        totalBytes: 32,
        totalDurationMs: 50,
        lastAssignedSeq: 1,
        lastDurableSeq: 1,
        lastSentSeq: 1,
        hardCommittedMs: 0,
        softCommittedMs: 0,
        replayInFlight: false,
        reconnectCount: 0,
        archiveState: "capturing",
        createdAt: freshUpdatedAt,
        updatedAt: freshUpdatedAt,
      },
    ]);

    await db.segments.bulkPut([
      {
        sessionId: "stale-session",
        segmentId: 1,
        startSeq: 1,
        endSeq: 1,
        startMs: 0,
        endMs: 50,
        durationMs: 50,
        byteLength: 32,
        chunkCount: 1,
        payloadKey: "stale-session:1",
        storageKind: "idb",
        keepForReplay: true,
        keepForArchive: true,
        createdAt: staleUpdatedAt,
      },
      {
        sessionId: "fresh-session",
        segmentId: 1,
        startSeq: 1,
        endSeq: 1,
        startMs: 0,
        endMs: 50,
        durationMs: 50,
        byteLength: 32,
        chunkCount: 1,
        payloadKey: "fresh-session:1",
        storageKind: "idb",
        keepForReplay: true,
        keepForArchive: true,
        createdAt: freshUpdatedAt,
      },
    ]);

    await db.payloads.bulkPut([
      {
        payloadKey: "stale-session:1",
        sessionId: "stale-session",
        segmentId: 1,
        byteLength: 32,
        payload: new Blob([createAudioPayload(32)]),
        createdAt: staleUpdatedAt,
      },
      {
        payloadKey: "fresh-session:1",
        sessionId: "fresh-session",
        segmentId: 1,
        byteLength: 32,
        payload: new Blob([createAudioPayload(32)]),
        createdAt: freshUpdatedAt,
      },
    ]);

    await db.recoveryCheckpoints.bulkPut([
      {
        sessionId: "stale-session",
        hardCommittedMs: 0,
        softCommittedMs: 0,
        lastFinalCoverageMs: 0,
        lastStablePartialCoverageMs: 0,
      },
      {
        sessionId: "fresh-session",
        hardCommittedMs: 0,
        softCommittedMs: 0,
        lastFinalCoverageMs: 0,
        lastStablePartialCoverageMs: 0,
      },
    ]);

    await db.chunkIndex.bulkPut([
      {
        sessionId: "stale-session",
        seq: 1,
        chunkId: "stale-session:1",
        segmentId: 1,
        startMs: 0,
        endMs: 50,
        durationMs: 50,
        offsetBytes: 0,
        byteLength: 32,
      },
      {
        sessionId: "fresh-session",
        seq: 1,
        chunkId: "fresh-session:1",
        segmentId: 1,
        startMs: 0,
        endMs: 50,
        durationMs: 50,
        offsetBytes: 0,
        byteLength: 32,
      },
    ]);

    await db.diagnostics.bulkPut([
      {
        sessionId: "stale-session",
        ts: staleUpdatedAt,
        code: "stale",
      },
      {
        sessionId: "fresh-session",
        ts: freshUpdatedAt,
        code: "fresh",
      },
    ]);

    const manager = new DurableAudioSessionManager();
    await manager.startSession(now);

    const activeDb = getConsultationDb();

    expect(await activeDb.manifests.get("stale-session")).toBeUndefined();
    expect(await activeDb.manifests.get("fresh-session")).toBeUndefined();
    expect(await activeDb.payloads.get("stale-session:1")).toBeUndefined();
    expect(await activeDb.payloads.get("fresh-session:1")).toBeUndefined();
    expect(await activeDb.recoveryCheckpoints.get("stale-session")).toBeUndefined();
    expect(await activeDb.recoveryCheckpoints.get("fresh-session")).toBeUndefined();

    const remainingSegments = await activeDb.segments.toArray();
    const remainingChunkIndex = await activeDb.chunkIndex.toArray();
    const remainingDiagnostics = await activeDb.diagnostics.toArray();

    expect(remainingSegments).toHaveLength(0);
    expect(remainingChunkIndex).toHaveLength(0);
    expect(remainingDiagnostics).toHaveLength(0);
  });

  it("does not accumulate persisted audio across successive sessions", async () => {
    const firstManager = new DurableAudioSessionManager({
      config: {
        segmentFlushBytes: 256,
        segmentFlushDurationMs: 60_000,
        segmentFlushIntervalMs: 60_000,
      },
    });

    await firstManager.startSession(0);
    await firstManager.appendCaptureChunk(createAudioPayload(256), 16, 1);
    await firstManager.appendCaptureChunk(createAudioPayload(256), 16, 2);
    await firstManager.flush("stop", 3);

    let activeDb = getConsultationDb();
    expect(await activeDb.manifests.count()).toBe(1);
    expect(await activeDb.recoveryCheckpoints.count()).toBe(1);
    expect(await activeDb.segments.count()).toBe(2);
    expect(await activeDb.payloads.count()).toBe(2);
    expect(await activeDb.diagnostics.count()).toBeGreaterThanOrEqual(2);

    const secondManager = new DurableAudioSessionManager({
      config: {
        segmentFlushBytes: 256,
        segmentFlushDurationMs: 60_000,
        segmentFlushIntervalMs: 60_000,
      },
    });

    await secondManager.startSession(10);

    activeDb = getConsultationDb();
    expect(await activeDb.manifests.count()).toBe(1);
    expect(await activeDb.recoveryCheckpoints.count()).toBe(1);
    expect(await activeDb.segments.count()).toBe(0);
    expect(await activeDb.payloads.count()).toBe(0);
    expect(await activeDb.diagnostics.count()).toBe(0);

    await secondManager.appendCaptureChunk(createAudioPayload(256), 16, 11);
    await secondManager.flush("stop", 12);

    expect(await activeDb.manifests.count()).toBe(1);
    expect(await activeDb.recoveryCheckpoints.count()).toBe(1);
    expect(await activeDb.segments.count()).toBe(1);
    expect(await activeDb.payloads.count()).toBe(1);
  });

  it("drops the consultation database after stop-triggered ttl", async () => {
    jest.useFakeTimers();

    const manager = new DurableAudioSessionManager();
    await manager.startSession(0);
    await manager.appendCaptureChunk(createAudioPayload(256), 16, 1);
    await manager.flush("stop", 2);
    manager.scheduleStorageDropAfterStop(CONSULTATION_DB_TTL_MS);

    let activeDb = getConsultationDb();
    expect(await activeDb.manifests.count()).toBeGreaterThan(0);

    jest.advanceTimersByTime(CONSULTATION_DB_TTL_MS);
    await flushMicrotasks();
    await flushMicrotasks();

    activeDb = getConsultationDb();
    expect(await activeDb.manifests.count()).toBe(0);
    expect(await activeDb.segments.count()).toBe(0);
    expect(await activeDb.payloads.count()).toBe(0);
  });

  it("reopens the consultation database after a DatabaseClosedError and keeps writing", async () => {
    const manager = new DurableAudioSessionManager({
      config: {
        segmentFlushBytes: 256,
        segmentFlushDurationMs: 60_000,
        segmentFlushIntervalMs: 60_000,
      },
    });

    await manager.startSession(0);
    await dropConsultationDb();

    await manager.appendCaptureChunk(createAudioPayload(256), 16, 1);

    const activeDb = getConsultationDb();
    expect(await activeDb.manifests.count()).toBeGreaterThan(0);
    expect(await activeDb.segments.count()).toBeGreaterThan(0);
    expect(await activeDb.payloads.count()).toBeGreaterThan(0);
  });

  it("queues subsequent appends behind a blocked durable flush", async () => {
    const manager = new DurableAudioSessionManager({
      config: {
        segmentFlushBytes: 2048,
        segmentFlushDurationMs: 60_000,
        segmentFlushIntervalMs: 60_000,
        maxPendingBytes: 1024,
      },
    });

    await manager.startSession(0);
    const { db, releaseWrites, getBlockedTransactionCount, getActiveBlockedTransactions } =
      blockDurableTransactions();

    let blockedFlushSettled = false;
    const blockedFlushPromise = manager
      .appendCaptureChunk(createAudioPayload(2048), 125, 1)
      .finally(() => {
        blockedFlushSettled = true;
      });

    await flushMicrotasks();
    await flushMicrotasks();

    expect(getBlockedTransactionCount()).toBe(1);
    expect(getActiveBlockedTransactions()).toBe(1);
    expect(blockedFlushSettled).toBe(false);
    expect(await db.segments.count()).toBe(0);
    expect(await db.payloads.count()).toBe(0);

    const queuedAppends = Array.from({ length: 3 }, (_, index) =>
      manager.appendCaptureChunk(createAudioPayload(256), 16, 10 + index)
    );
    await flushMicrotasks();
    await flushMicrotasks();

    const pendingBytes = (manager as any).segmentBuilder.getPendingBytes();
    const earlyResults = await Promise.all(
      queuedAppends.map((promise) =>
        Promise.race([
          promise.then(() => "fulfilled"),
          new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 0)),
        ])
      )
    );

    expect(earlyResults.every((result) => result === "pending")).toBe(true);
    expect(pendingBytes).toBe(0);

    const stalledStatus = await manager.getConsultationStorageStatus();
    expect(stalledStatus).toEqual(
      expect.objectContaining({
        durableSegments: 0,
        durableBytes: 0,
      })
    );

    releaseWrites();
    await blockedFlushPromise;
    await Promise.all(queuedAppends);
    await manager.flush("stop", 999);

    expect(getActiveBlockedTransactions()).toBe(0);
    expect(await db.segments.count()).toBe(2);
    expect(await db.payloads.count()).toBe(2);
  });

  it("caps in-memory backlog when IndexedDB stops draining", async () => {
    const manager = new DurableAudioSessionManager({
      config: {
        segmentFlushBytes: 2048,
        segmentFlushDurationMs: 60_000,
        segmentFlushIntervalMs: 60_000,
        maxPendingBytes: 1024,
      },
    });

    await manager.startSession(0);
    const { releaseWrites } = blockDurableTransactions();

    const blockedFlushPromise = manager.appendCaptureChunk(
      createAudioPayload(2048),
      125,
      1
    );

    await flushMicrotasks();
    await flushMicrotasks();

    const sustainedBacklog = Array.from({ length: 6 }, (_, index) =>
      manager.appendCaptureChunk(createAudioPayload(256), 16, 10 + index)
    );

    try {
      await flushMicrotasks();
      await flushMicrotasks();

      const pendingBytes = (manager as any).segmentBuilder.getPendingBytes();

      expect(pendingBytes).toBeLessThanOrEqual(1024);
    } finally {
      releaseWrites();
      await Promise.all(sustainedBacklog);
      await blockedFlushPromise;
      await manager.flush("stop", 999);
    }
  });

  it("applies IDB backpressure when pending memory backlog reaches the limit", async () => {
    const manager = new DurableAudioSessionManager({
      config: {
        segmentFlushBytes: 2048,
        segmentFlushDurationMs: 60_000,
        segmentFlushIntervalMs: 60_000,
        maxPendingBytes: 1024,
      },
    });

    await manager.startSession(0);
    const { db, releaseWrites } = blockDurableTransactions();

    const blockedFlushPromise = manager.appendCaptureChunk(
      createAudioPayload(2048),
      125,
      1
    );

    await flushMicrotasks();
    await flushMicrotasks();

    const appendPromises = Array.from({ length: 6 }, (_, index) =>
      manager.appendCaptureChunk(createAudioPayload(256), 16, 10 + index)
    );
    await flushMicrotasks();
    await flushMicrotasks();

    const pendingBytes = (manager as any).segmentBuilder.getPendingBytes();
    const stalledStatus = await manager.getConsultationStorageStatus();

    try {
      const earlyResults = await Promise.allSettled(
        appendPromises.map((promise) =>
          Promise.race([
            promise.then(() => "fulfilled"),
            new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 0)),
          ])
        )
      );

      const fulfilledCount = earlyResults.filter(
        (result) => result.status === "fulfilled" && result.value === "fulfilled"
      ).length;

      expect(fulfilledCount).toBeLessThan(6);
      expect(await db.segments.count()).toBe(0);
      expect(await db.payloads.count()).toBe(0);
      expect(stalledStatus).toEqual(
        expect.objectContaining({
          durableSegments: 0,
          durableBytes: 0,
        })
      );
      expect(pendingBytes).toBeLessThanOrEqual(1024);
    } finally {
      releaseWrites();
      await Promise.all(appendPromises);
      await blockedFlushPromise;
      await manager.flush("stop", 999);
    }
  });
});