import { test } from 'node:test';
import assert from 'node:assert/strict';
import type { IncomingMessage } from '../types/openwa';
import type { SttResult, SttProvider } from './openai-stt.client.ts';
import type { TranscriptionPayload } from './webhook.delivery.ts';
import {
  decodedBase64Size,
  TranscriptionCoordinator,
  KvStore,
  TranscriptionConfig,
  ChatSink,
  ChatDeliveryMode,
} from './transcription.coordinator.ts';

function makeStore(seed: Record<string, unknown> = {}): KvStore & { keys(): string[] } {
  const m = new Map<string, unknown>(Object.entries(seed));
  return {
    get: async <T>(k: string) => (m.has(k) ? (m.get(k) as T) : null),
    set: async (k, v) => void m.set(k, v),
    delete: async (k) => void m.delete(k),
    // Mirrors the host: list(prefix) filters for you, and a bare list() returns everything.
    list: async (prefix?: string) => [...m.keys()].filter((k) => !prefix || k.startsWith(prefix)),
    keys: () => [...m.keys()],
  };
}

function voiceMsg(over: Partial<IncomingMessage> = {}): IncomingMessage {
  return {
    id: 'm1',
    from: 'x@s.whatsapp.net',
    to: 'y@s.whatsapp.net',
    chatId: 'c1@s.whatsapp.net',
    body: '',
    type: 'voice',
    timestamp: 0,
    fromMe: false,
    isGroup: false,
    media: { mimetype: 'audio/ogg; codecs=opus', data: Buffer.from('AUDIO').toString('base64'), sizeBytes: 5 },
    ...over,
  };
}

interface FakeProvider extends SttProvider {
  calls: Array<{ audio: Uint8Array; mimetype: string }>;
}
type ChatSend =
  | { kind: 'sendText'; sessionId: string; chatId: string; text: string }
  | { kind: 'reply'; sessionId: string; chatId: string; quoted: string; text: string };

function setup(opts: {
  result?: SttResult;
  providerThrows?: boolean;
  deliveryThrows?: boolean;
  noDelivery?: boolean;
  chatDelivery?: ChatDeliveryMode;
  config?: Partial<TranscriptionConfig>;
  store?: KvStore;
  now?: () => number;
}) {
  const provider: FakeProvider = {
    calls: [],
    async transcribe(audio, mimetype) {
      this.calls.push({ audio, mimetype });
      if (opts.providerThrows) throw new Error('stt down');
      return opts.result ?? { text: 'hola', language: 'es' };
    },
  };
  const deliveries: TranscriptionPayload[] = [];
  const delivery = opts.noDelivery
    ? undefined
    : {
        async deliver(e: TranscriptionPayload) {
          deliveries.push(e);
          if (opts.deliveryThrows) throw new Error('hook 502');
        },
      };
  const chatSends: ChatSend[] = [];
  const chat: ChatSink = {
    sendText: async (sessionId, chatId, text) => void chatSends.push({ kind: 'sendText', sessionId, chatId, text }),
    reply: async (sessionId, chatId, quoted, text) =>
      void chatSends.push({ kind: 'reply', sessionId, chatId, quoted, text }),
  };
  const warns: string[] = [];
  const co = new TranscriptionCoordinator({
    provider,
    delivery,
    chat,
    chatDelivery: opts.chatDelivery ?? 'off',
    store: opts.store ?? makeStore(),
    config: { enabledMessageTypes: ['voice'], maxSizeBytes: 1000, maxPerHour: 100, ...opts.config },
    providerLabel: 'faster-whisper',
    model: 'small',
    logger: { warn: m => void warns.push(m) },
    now: opts.now,
  });
  return { co, provider, deliveries, chatSends, warns };
}

test('transcribes a voice note and delivers a completed event with decoded audio + payload', async () => {
  const { co, provider, deliveries, chatSends } = setup({});
  await co.handle('s1', voiceMsg());
  assert.equal(provider.calls.length, 1);
  assert.deepEqual([...provider.calls[0].audio], [...Buffer.from('AUDIO')]);
  assert.equal(provider.calls[0].mimetype, 'audio/ogg; codecs=opus');
  assert.deepEqual(deliveries[0], {
    event: 'message.transcription',
    sessionId: 's1',
    messageId: 'm1',
    chatId: 'c1@s.whatsapp.net',
    status: 'completed',
    source: 'speech-to-text',
    untrusted: true,
    transcription: { text: 'hola', language: 'es', provider: 'faster-whisper', model: 'small' },
  });
  assert.equal(chatSends.length, 0); // chatDelivery off by default
});

test('skips a non-enabled message type — no STT, no event', async () => {
  const { co, provider, deliveries } = setup({});
  await co.handle('s1', voiceMsg({ type: 'image' }));
  assert.equal(provider.calls.length, 0);
  assert.equal(deliveries.length, 0);
});

test('skips self-sent, status broadcast, newsletter, and broadcast voice notes', async () => {
  const { co, provider, deliveries } = setup({});
  await co.handle('s1', voiceMsg({ id: 'm-self', fromMe: true }));
  await co.handle('s1', voiceMsg({ id: 'm-status', isStatusBroadcast: true }));
  await co.handle('s1', voiceMsg({ id: 'm-news', chatId: '1203630@newsletter' }));
  await co.handle('s1', voiceMsg({ id: 'm-bcast', chatId: 'status@broadcast' }));
  assert.equal(provider.calls.length, 0);
  assert.equal(deliveries.length, 0);
});

test('skips when media is absent — no event', async () => {
  const { co, provider, deliveries } = setup({});
  await co.handle('s1', voiceMsg({ media: undefined }));
  assert.equal(provider.calls.length, 0);
  assert.equal(deliveries.length, 0);
});

test('emits a skipped event when media was omitted over the inbound cap', async () => {
  const { co, provider, deliveries } = setup({});
  await co.handle('s1', voiceMsg({ media: { mimetype: 'audio/ogg', omitted: true, sizeBytes: 99999999 } }));
  assert.equal(provider.calls.length, 0);
  assert.equal(deliveries[0].status, 'skipped');
  assert.equal(deliveries[0].reason, 'media_unavailable');
});

test('emits a skipped event when the audio exceeds maxSizeBytes', async () => {
  const { co, provider, deliveries } = setup({ config: { maxSizeBytes: 3 } }); // 'AUDIO' decodes to 5 bytes
  await co.handle('s1', voiceMsg());
  assert.equal(provider.calls.length, 0);
  assert.equal(deliveries[0].status, 'skipped');
  assert.equal(deliveries[0].reason, 'too_large');
});

test('idempotency: the same messageId is processed once (one STT call, one event)', async () => {
  const { co, provider, deliveries } = setup({});
  await co.handle('s1', voiceMsg());
  await co.handle('s1', voiceMsg()); // re-fire
  assert.equal(provider.calls.length, 1);
  assert.equal(deliveries.length, 1);
});

test('rate limit: emits a skipped event once maxPerHour is reached, with no more STT', async () => {
  const { co, provider, deliveries } = setup({ config: { maxPerHour: 2 }, now: () => 0 });
  for (const id of ['a', 'b', 'c']) await co.handle('s1', voiceMsg({ id }));
  assert.equal(provider.calls.length, 2);
  assert.equal(deliveries[deliveries.length - 1].status, 'skipped');
  assert.equal(deliveries[deliveries.length - 1].reason, 'rate_limited');
});

test('emits a skipped event when the transcript is empty/whitespace', async () => {
  const { co, deliveries } = setup({ result: { text: '   ' } });
  await co.handle('s1', voiceMsg());
  assert.equal(deliveries[0].status, 'skipped');
  assert.equal(deliveries[0].reason, 'empty');
});

test('emits a failed event (and never throws) when STT errors', async () => {
  const { co, deliveries, warns } = setup({ providerThrows: true });
  await assert.doesNotReject(co.handle('s1', voiceMsg()));
  assert.equal(deliveries[0].status, 'failed');
  assert.ok(warns.length >= 1);
});

test('fail-open: a delivery error is swallowed and warned', async () => {
  const { co, warns } = setup({ deliveryThrows: true });
  await assert.doesNotReject(co.handle('s1', voiceMsg()));
  assert.ok(warns.length >= 1);
});

test('chatDelivery=self sends the transcript to the bot own number (msg.to)', async () => {
  const { co, chatSends } = setup({ chatDelivery: 'self' });
  await co.handle('s1', voiceMsg());
  assert.deepEqual(chatSends[0], { kind: 'sendText', sessionId: 's1', chatId: 'y@s.whatsapp.net', text: 'hola' });
});

test('chatDelivery=reply quote-replies in the original chat', async () => {
  const { co, chatSends } = setup({ chatDelivery: 'reply' });
  await co.handle('s1', voiceMsg());
  assert.deepEqual(chatSends[0], {
    kind: 'reply',
    sessionId: 's1',
    chatId: 'c1@s.whatsapp.net',
    quoted: 'm1',
    text: 'hola',
  });
});

test('chat delivery does not fire for a skipped/failed outcome', async () => {
  const { co, chatSends } = setup({ chatDelivery: 'self', config: { maxSizeBytes: 3 } }); // oversize → skipped
  await co.handle('s1', voiceMsg());
  assert.equal(chatSends.length, 0);
});

test('works chat-only when no webhook delivery is configured', async () => {
  const { co, chatSends } = setup({ noDelivery: true, chatDelivery: 'self' });
  await assert.doesNotReject(co.handle('s1', voiceMsg()));
  assert.equal(chatSends.length, 1);
});

test('a webhook delivery failure does not suppress the in-chat transcript delivery', async () => {
  // The two sinks are independent: a 502 from the delivery webhook must not swallow the chat send.
  const { co, chatSends, warns } = setup({ deliveryThrows: true, chatDelivery: 'self' });
  await assert.doesNotReject(co.handle('s1', voiceMsg()));
  assert.equal(chatSends.length, 1);   // chat sink still fired despite the webhook failure
  assert.ok(warns.length >= 1);        // the webhook failure is still warned
});

// ── Storage growth (1.1.0) ──────────────────────────────────────────────────────────────────────────
// Before host 0.26.0 the host re-stats every one of a plugin's storage keys on EVERY write. Pre-1.1.0 this plugin wrote one
// never-deleted key per transcribed voice note plus one per session per hour, so each write cost more
// than the last, forever.

test('storage keys stay bounded at one dedup list + one rate counter per session', async () => {
  const store = makeStore();
  const { co } = setup({ store, config: { maxPerHour: 10_000 } });
  for (let i = 0; i < 200; i++) await co.handle('s1', voiceMsg({ id: `m${i}` }));
  assert.deepEqual(store.keys().sort(), ['rate:s1', 'seen:s1']);
});

test('the dedup list is capped, and still dedups within its window', async () => {
  const store = makeStore();
  const { co, provider } = setup({ store, config: { maxPerHour: 10_000 } });
  for (let i = 0; i < 600; i++) await co.handle('s1', voiceMsg({ id: `m${i}` }));
  const seen = (await store.get<string[]>('seen:s1'))!;
  assert.ok(seen.length <= 500, `dedup list grew to ${seen.length}`);
  const before = provider.calls.length;
  await co.handle('s1', voiceMsg({ id: 'm599' })); // the most recent id is still remembered
  assert.equal(provider.calls.length, before, 'a recently-seen message must not be transcribed again');
});

test('the hourly counter reuses one key across hours instead of leaving one behind per hour', async () => {
  const store = makeStore();
  let hour = 0;
  const { co, provider } = setup({ store, config: { maxPerHour: 1 }, now: () => hour * 3_600_000 });
  await co.handle('s1', voiceMsg({ id: 'a' }));
  await co.handle('s1', voiceMsg({ id: 'b' })); // over the cap in the same hour
  assert.equal(provider.calls.length, 1);
  hour = 1; // next hour: the budget resets, and no second rate key appears
  await co.handle('s1', voiceMsg({ id: 'c' }));
  assert.equal(provider.calls.length, 2);
  assert.equal(store.keys().filter(k => k.startsWith('rate:')).length, 1);
});

test('pre-1.1.0 per-message dedup keys are swept away as transcriptions run', async () => {
  const legacy: Record<string, unknown> = {};
  for (let i = 0; i < 60; i++) legacy[`seen:s1:old${i}`] = 1;
  const store = makeStore(legacy);
  const { co } = setup({ store, config: { maxPerHour: 10_000 } });
  for (let i = 0; i < 3; i++) await co.handle('s1', voiceMsg({ id: `m${i}` }));
  assert.equal(store.keys().filter(k => k.startsWith('seen:s1:')).length, 0, 'legacy keys should be gone');
  assert.ok(store.keys().includes('seen:s1'), 'the new dedup list survives the sweep');
});

// ── Size guard (1.1.0) ──────────────────────────────────────────────────────────────────────────────
// Checked before decoding: decoding first materialized an oversized note as a Buffer on top of the
// base64 string already in memory, against a 256 MB worker heap the host does not respawn.

test('oversized audio is skipped without ever being decoded', async () => {
  const { co, provider, deliveries } = setup({ config: { maxSizeBytes: 10 } });
  const big = { media: { mimetype: 'audio/ogg', data: Buffer.alloc(5000).toString('base64') } };
  await co.handle('s1', voiceMsg(big as Partial<IncomingMessage>));
  assert.equal(provider.calls.length, 0);
  assert.equal(deliveries[0].status, 'skipped');
  assert.equal(deliveries[0].reason, 'too_large');
});

// Padding matters: the naive floor(len/4)*3 - pad form under-counts UNPADDED base64, which would let an
// oversized note through the guard instead of failing safe.
test('decodedBase64Size is exact for padded and unpadded input', () => {
  for (const raw of ['A', 'AB', 'ABC', 'ABCD', 'ABCDE', '', 'hello world!!']) {
    const b64 = Buffer.from(raw).toString('base64');
    assert.equal(decodedBase64Size(b64), Buffer.byteLength(raw), `padded: ${JSON.stringify(raw)}`);
    const unpadded = b64.replace(/=+$/, '');
    assert.equal(decodedBase64Size(unpadded), Buffer.byteLength(raw), `unpadded: ${JSON.stringify(raw)}`);
  }
});

test('audio exactly at the limit is still transcribed', async () => {
  const raw = Buffer.alloc(300, 7);
  const { co, provider } = setup({ config: { maxSizeBytes: raw.byteLength } });
  await co.handle('s1', voiceMsg({ media: { mimetype: 'audio/ogg', data: raw.toString('base64') } }));
  assert.equal(provider.calls.length, 1, 'a note exactly at maxSizeBytes must not be rejected');
});

// Regression: the dedup claim is a read-modify-write over ONE list per session, and handle() is floated
// off the hook — so a burst of voice notes interleaved and the last writer overwrote the others' ids.
// Each lost id is a second paid STT call and, with chatDelivery 'reply', a duplicate transcript the
// CONTACT sees. The pre-1.1.0 one-key-per-note scheme had no such window.
test('a concurrent burst of voice notes claims every id exactly once', async () => {
  const inner = makeStore();
  const tick = () => new Promise(r => setTimeout(r, 1));
  const slow: KvStore & { keys(): string[] } = {
    get: async <T>(k: string) => { await tick(); return inner.get<T>(k); },
    set: async (k, v) => { await tick(); return inner.set(k, v); },
    delete: k => inner.delete(k),
    list: p => inner.list(p),
    keys: () => inner.keys(),
  };
  const { co, provider } = setup({ store: slow, config: { maxPerHour: 10_000 } });
  const ids = ['a', 'b', 'c', 'd'];
  await Promise.all(ids.map(id => co.handle('s1', voiceMsg({ id }))));

  assert.equal(provider.calls.length, 4, 'each note transcribed once');
  const seen = (await slow.get<string[]>('seen:s1')) ?? [];
  assert.deepEqual([...seen].sort(), ids, 'every id retained — a lost one re-transcribes on redelivery');

  // Replaying the burst must now be a complete no-op.
  await Promise.all(ids.map(id => co.handle('s1', voiceMsg({ id }))));
  assert.equal(provider.calls.length, 4, 'no duplicate STT call on redelivery');
});

// The pre-1.1.0 rate counter was `rate:<sid>:<hour>` — one key per hour, also never deleted. Sweeping
// only the `seen:` family would leave a year-old install with ~8760 counter keys: exactly the per-write
// stat cost the sweep exists to remove.
test('the legacy sweep clears the old hourly rate keys too', async () => {
  const legacy: Record<string, unknown> = {};
  for (let i = 0; i < 30; i++) legacy[`seen:s1:old${i}`] = 1;
  for (let i = 0; i < 30; i++) legacy[`rate:s1:${400000 + i}`] = 5;
  const store = makeStore(legacy);
  const { co } = setup({ store, config: { maxPerHour: 10_000 } });
  for (let i = 0; i < 4; i++) await co.handle('s1', voiceMsg({ id: `m${i}` }));
  assert.equal(store.keys().filter(k => k.startsWith('seen:s1:')).length, 0, 'legacy dedup keys gone');
  assert.equal(store.keys().filter(k => k.startsWith('rate:s1:')).length, 0, 'legacy rate keys gone');
  assert.ok(store.keys().includes('seen:s1'), 'the 1.1.0 dedup list survives');
  assert.ok(store.keys().includes('rate:s1'), 'the 1.1.0 rate key survives');
});

// The host's list() swallows its own errors and resolves []. Retiring on a single empty listing would
// therefore strand the legacy keys forever on one transient failure — the very thing being fixed.
test('the sweep does not retire on a single empty listing', async () => {
  const store = makeStore();
  let listCalls = 0;
  const flaky: KvStore & { keys(): string[] } = {
    ...store,
    list: async (p?: string) => { listCalls++; return listCalls <= 2 ? [] : store.list(p); },
  };
  const { co } = setup({ store: flaky, config: { maxPerHour: 10_000 } });
  await co.handle('s1', voiceMsg({ id: 'a' }));
  const afterFirst = listCalls;
  await co.handle('s1', voiceMsg({ id: 'b' }));
  assert.ok(listCalls > afterFirst, 'a second pass still runs after one clean-looking listing');
});

// Regression: the per-session claim chain must survive a coordinator rebuild. The plugin rebuilds the
// coordinator whenever the resolved config signature changes, which with per-session config can be every
// message — an instance-scoped chain handed each rebuild a fresh empty map and silently undid the
// serialization the previous release added.
test('the dedup claim stays serialized across coordinator rebuilds', async () => {
  const inner = makeStore();
  const tick = () => new Promise(r => setTimeout(r, 1));
  const slow: KvStore & { keys(): string[] } = {
    get: async <T>(k: string) => { await tick(); return inner.get<T>(k); },
    set: async (k, v) => { await tick(); return inner.set(k, v); },
    delete: k => inner.delete(k), list: p => inner.list(p), keys: () => inner.keys(),
  };
  const ids = ['a', 'b', 'c', 'd'];
  // Every note handled by its OWN freshly built coordinator, as a per-session config override causes.
  await Promise.all(ids.map(id => setup({ store: slow, config: { maxPerHour: 10_000 } }).co.handle('s1', voiceMsg({ id }))));
  const seen = (await slow.get<string[]>('seen:s1')) ?? [];
  assert.deepEqual([...seen].sort(), ids, 'a rebuild must not reset the claim chain');
});

// Regression: the sweep must never retire itself. The host's list() resolves [] on a read error, so an
// empty listing cannot be told apart from a failed one — any retirement rule built on it strands the very
// keys the sweep exists to remove.
test('a transient list() failure does not stop the legacy sweep for good', async () => {
  const store = makeStore({ 'seen:s1:legacy-1': 1 });
  let broken = true;
  const flaky: KvStore & { keys(): string[] } = {
    ...store,
    list: async (p?: string) => (broken ? [] : store.list(p)),
  };
  const { co } = setup({ store: flaky, config: { maxPerHour: 10_000 } });
  await co.handle('s1', voiceMsg({ id: 'm1' }));      // listing broken → nothing swept
  assert.ok(store.keys().includes('seen:s1:legacy-1'), 'still there while storage is broken');
  broken = false;
  await co.handle('s1', voiceMsg({ id: 'm2' }));      // storage recovers → sweep must still run
  assert.ok(!store.keys().includes('seen:s1:legacy-1'), 'the sweep resumed after the transient failure');
});

// ── Late delivery (1.3.0) ───────────────────────────────────────────────────────────────────────────
// From OpenWA 0.23.6 a Baileys session delivers, after it reconnects, the voice notes WhatsApp queued while
// it was disconnected, each with its original send time. A quote-reply to one of those lands in the chat
// long after the conversation moved on.

const NOW = 1_758_000_000_000;

test('a late voice note is transcribed for the webhook but not quote-replied', async () => {
  const { co, provider, deliveries, chatSends } = setup({ chatDelivery: 'reply', now: () => NOW });
  await co.handle('s1', voiceMsg({ timestamp: NOW / 1000 - 600 }));
  assert.equal(provider.calls.length, 1);
  assert.equal(deliveries[0].status, 'completed');
  assert.equal(deliveries[0].timestamp, NOW / 1000 - 600);
  assert.deepEqual(chatSends, []);
});

test('with a quote-reply as the only sink, a late note is not sent to speech-to-text', async () => {
  const { co, provider, chatSends, warns } = setup({ chatDelivery: 'reply', noDelivery: true, now: () => NOW });
  await co.handle('s1', voiceMsg({ timestamp: NOW / 1000 - 600 }));
  assert.equal(provider.calls.length, 0);
  assert.deepEqual(chatSends, []);
  assert.ok(warns.includes('Transcription skipped: late'), warns.join(' | '));
});

test("the webhook event carries the note's send time", async () => {
  const live = setup({ now: () => NOW });
  await live.co.handle('s1', voiceMsg({ timestamp: NOW / 1000 }));
  assert.equal(live.deliveries[0].timestamp, NOW / 1000);
  const big = setup({ config: { maxSizeBytes: 3 }, now: () => NOW }); // 'AUDIO' decodes to 5 bytes
  await big.co.handle('s1', voiceMsg({ timestamp: NOW / 1000 }));
  assert.equal(big.deliveries[0].reason, 'too_large');
  assert.equal(big.deliveries[0].timestamp, NOW / 1000);
});

test('a late note still reaches the self note, and a fresh one is still quote-replied', async () => {
  // With no webhook, the self note is the only copy, so a late note must still be transcribed for it.
  const self = setup({ chatDelivery: 'self', noDelivery: true, now: () => NOW });
  await self.co.handle('s1', voiceMsg({ timestamp: NOW / 1000 - 600 }));
  assert.deepEqual(self.chatSends, [{ kind: 'sendText', sessionId: 's1', chatId: 'y@s.whatsapp.net', text: 'hola' }]);
  // Exactly five minutes old, and a send time well ahead of the gateway clock, both still count as fresh.
  for (const timestamp of [NOW / 1000 - 300, NOW / 1000 + 600]) {
    const fresh = setup({ chatDelivery: 'reply', noDelivery: true, now: () => NOW });
    await fresh.co.handle('s1', voiceMsg({ timestamp }));
    assert.equal(fresh.chatSends.length, 1, `timestamp ${timestamp}`);
    assert.equal(fresh.chatSends[0].kind, 'reply');
  }
});