import type { IncomingMessage } from '../types/openwa';
import type { SttProvider, SttResult } from './openai-stt.client.ts';
import type { TranscriptDelivery, TranscriptionPayload } from './webhook.delivery.ts';

/** Minimal KV surface the coordinator needs (adapted from `ctx.storage` by the plugin). */
export interface KvStore {
  get<T = unknown>(key: string): Promise<T | null>;
  set(key: string, value: unknown): Promise<void>;
  delete(key: string): Promise<void>;
  list(prefix?: string): Promise<string[]>;
}

// How many recent message ids the per-session dedup list keeps. Before host 0.26.0 the host re-stats every one of a
// plugin's storage keys on EVERY write, so the pre-1.1.0 scheme — one key per transcribed voice note,
// never deleted — made each write more expensive than the last, forever. One bounded list per session
// costs the same single read + write it always did, and the key count is now O(sessions).
// Every note that reaches the claim consumes a slot, including ones later SKIPPED (rate-limited, too
// large, unreadable media) — the claim is deliberately first so each outcome fires at most once. So 500
// ids is 500 inbound notes, not 500 transcriptions: it is roughly eight hours at the default 60/hour cap
// only when traffic stays near that cap, and proportionally less on a busy line.
const SEEN_HISTORY = 500;

// Legacy `seen:<sessionId>:<messageId>` keys are swept a few at a time (see sweepLegacySeen) rather than
// all at once: an upgraded install may hold thousands, and onEnable now runs unattended at host boot,
// before sessions connect, where a long delete loop would burn the lifecycle budget for nothing.
const LEGACY_SWEEP_PER_RUN = 25;

// Per-session tail for the dedup claim. MODULE scope, not instance scope: the plugin rebuilds the
// coordinator whenever the resolved config signature changes, and with per-session config that can be
// every message — an instance field would hand each rebuilt coordinator a fresh, empty chain and silently
// undo the serialization. One worker runs one plugin, so a module-level map is exactly the right scope.
// Only the claim is serialized; the STT call and delivery stay concurrent.
const claimTails = new Map<string, Promise<unknown>>();

// From OpenWA 0.23.6 a Baileys session delivers, after it reconnects, what WhatsApp queued while it was
// disconnected, each message with its original send time. Five minutes is far above clock skew between
// WhatsApp and the gateway and matches the host's own auto-reply age limit.
const LATE_AFTER_MS = 5 * 60_000;

export interface CoordinatorLogger {
  warn(message: string, meta?: Record<string, unknown>): void;
}

/** Send the transcript into a WhatsApp chat (adapted from `ctx.messages`). */
export interface ChatSink {
  sendText(sessionId: string, chatId: string, text: string): Promise<unknown>;
  reply(sessionId: string, chatId: string, quotedMessageId: string, text: string): Promise<unknown>;
}

/** off = no chat message; self = note to the bot's own number; reply = quote-reply to the sender. */
export type ChatDeliveryMode = 'off' | 'self' | 'reply';

/**
 * Decoded byte length of a base64 string, without decoding it. Exact for well-formed base64; a value
 * containing whitespace over-counts, which fails safe (the audio is skipped as too large rather than
 * being decoded into the worker heap).
 */
export function decodedBase64Size(b64: string): number {
  const pad = b64.endsWith('==') ? 2 : b64.endsWith('=') ? 1 : 0;
  // (length - padding) * 3/4 is exact for BOTH padded and unpadded input. The obvious
  // floor(length/4)*3 - pad form silently UNDER-counts unpadded base64 (a 3-char tail decodes to 2
  // bytes but rounds to 0), which would let an oversized note through the guard rather than fail safe.
  return Math.floor(((b64.length - pad) * 3) / 4);
}

export interface TranscriptionConfig {
  /** Message types to transcribe, e.g. ['voice'] (PTT). */
  enabledMessageTypes: string[];
  /** Skip audio larger than this (decoded bytes) — the exact cost guard. */
  maxSizeBytes: number;
  /** Best-effort per-session hourly cap on transcriptions. */
  maxPerHour: number;
}

export interface CoordinatorDeps {
  provider: SttProvider;
  /** Webhook sink. Optional: omit for chat-only operation. */
  delivery?: TranscriptDelivery;
  /** Chat sink (used when chatDelivery !== 'off'). */
  chat?: ChatSink;
  chatDelivery: ChatDeliveryMode;
  store: KvStore;
  config: TranscriptionConfig;
  /** Label recorded in the delivered event, e.g. 'faster-whisper'. */
  providerLabel: string;
  model: string;
  logger: CoordinatorLogger;
  /** Injectable clock for the hourly rate-limit bucket and the late check (defaults to Date.now). */
  now?: () => number;
}

/**
 * The framework-agnostic core: decide whether to transcribe an inbound message, run the guards (type,
 * size, idempotency, rate limit), call STT, and report the outcome — a `completed` event with the
 * transcript, or a `failed`/`skipped` event explaining why. Fail-open throughout: never throws to the
 * caller, so a transcription failure can never disrupt message delivery.
 */
export class TranscriptionCoordinator {
  private readonly now: () => number;

  private serialize<T>(key: string, fn: () => Promise<T>): Promise<T> {
    const prev = claimTails.get(key) ?? Promise.resolve();
    const next = prev.catch(() => undefined).then(fn);
    claimTails.set(key, next);
    void next.catch(() => undefined).finally(() => {
      if (claimTails.get(key) === next) claimTails.delete(key);
    });
    return next;
  }

  constructor(private readonly deps: CoordinatorDeps) {
    this.now = deps.now ?? (() => Date.now());
  }

  async handle(sessionId: string, msg: IncomingMessage): Promise<void> {
    const { config, store, logger } = this.deps;
    try {
      // Not our concern → no event at all.
      const lowerChatId = (msg.chatId ?? '').toLowerCase();
      if (
        msg.fromMe ||
        msg.isStatusBroadcast ||
        lowerChatId.endsWith('@newsletter') ||
        lowerChatId.endsWith('@broadcast')
      ) {
        return;
      }
      if (!config.enabledMessageTypes.includes(msg.type)) return;
      if (!msg.media) return;

      // Judged once, on arrival, so a slow STT call cannot tip a live note over the bound. A missing or
      // unusable send time, or one ahead of the gateway clock, reads as live.
      const sent = new Date((msg.timestamp ?? 0) * 1000);
      const late = sent.getTime() > 0 && this.now() - sent.getTime() > LATE_AFTER_MS;

      // Idempotency first, so every outcome (including skips) fires at most once per message id. Kept in
      // durable storage rather than memory: with chatDelivery 'reply' a duplicate transcript is a second
      // quote-reply the CONTACT sees, so this has to survive a worker restart, which is exactly when
      // WhatsApp is most likely to redeliver. Best-effort: read-modify-write is not atomic across a truly
      // simultaneous #466 re-fire (documented).
      // Serialized per session: the check and the write are a read-modify-write over ONE list, every
      // await in between is an IPC round-trip to the host, and `handle()` is deliberately floated off the
      // hook — so a burst (the engine materializes several voice notes at once) 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 read-modify-write and so no such window; this restores that guarantee.
      const claimed = await this.serialize(sessionId, async () => {
        const seenKey = `seen:${sessionId}`;
        const seen = (await store.get<string[]>(seenKey)) ?? [];
        if (seen.includes(msg.id)) return false;
        await store.set(seenKey, [msg.id, ...seen].slice(0, SEEN_HISTORY));
        return true;
      });
      if (!claimed) return;
      await this.sweepLegacySeen(sessionId);

      // A late note is never quote-replied (see emit), so when that reply is the only sink nobody would
      // receive the transcript: skip before spending STT money or a slot of the hourly cap on it.
      if (late && this.deps.chatDelivery === 'reply' && !this.deps.delivery) {
        await this.emit(sessionId, msg, { status: 'skipped', reason: 'late' });
        return;
      }

      const media = msg.media;
      if (media.omitted || !media.data) {
        await this.emit(sessionId, msg, { status: 'skipped', reason: 'media_unavailable' });
        return;
      }
      if (!media.mimetype || !media.mimetype.startsWith('audio/')) return; // defensive: not audio

      // Size is checked BEFORE decoding. Decoding first meant an oversized note — the very thing this
      // guard exists to reject — was materialized as a Buffer anyway, on top of the base64 string already
      // in memory, against a 256 MB worker heap that the host does not respawn if it blows.
      if (decodedBase64Size(media.data) > config.maxSizeBytes) {
        await this.emit(sessionId, msg, { status: 'skipped', reason: 'too_large' });
        return;
      }
      const audio = Buffer.from(media.data, 'base64');

      // One key per session holding {bucket, count}, not one key per session per HOUR — the hourly keys
      // were never deleted, so they accumulated for the life of the install.
      //
      // Serialized on the same per-session tail as the claim, because collapsing the hours into ONE key
      // turned this into a read-modify-write that the old scheme did not have: two notes straddling an
      // hour boundary wrote different keys before, but now the pre-boundary write can land last and
      // restore the previous bucket, restarting the new hour's count at zero and doubling a cap whose
      // stated job is bounding paid-API spend.
      const withinCap = await this.serialize(sessionId, async () => {
        const rateKey = `rate:${sessionId}`;
        const bucket = Math.floor(this.now() / 3_600_000);
        const rate = await store.get<{ bucket: number; count: number }>(rateKey);
        const count = rate?.bucket === bucket ? rate.count : 0;
        if (count >= config.maxPerHour) return false;
        await store.set(rateKey, { bucket, count: count + 1 });
        return true;
      });
      if (!withinCap) {
        await this.emit(sessionId, msg, { status: 'skipped', reason: 'rate_limited' });
        return;
      }

      let result: SttResult;
      try {
        result = await this.deps.provider.transcribe(audio, media.mimetype);
      } catch (err) {
        await this.emit(sessionId, msg, { status: 'failed', reason: err instanceof Error ? err.message : String(err) });
        return;
      }
      if (!result.text.trim()) {
        await this.emit(sessionId, msg, { status: 'skipped', reason: 'empty' });
        return;
      }

      await this.emit(sessionId, msg, {
        status: 'completed',
        late,
        text: result.text,
        transcription: {
          text: result.text,
          language: result.language,
          provider: this.deps.providerLabel,
          model: this.deps.model,
        },
      });
    } catch (err) {
      // Fail-open: a transcription/delivery failure must never disrupt message delivery.
      logger.warn('Transcription failed (skipped)', {
        messageId: msg.id,
        error: err instanceof Error ? err.message : String(err),
      });
    }
  }

  /**
   * Delete a few pre-1.1.0 `seen:<sessionId>:<messageId>` keys per transcription. Without this an
   * upgraded install keeps paying the old cost forever: those keys are never read again, but a host before
   * 0.26.0 still stats every one of them on every write. Awaited rather than floated — the whole coordinator
   * already runs off the message-delivery path, and floating it made the cleanup non-deterministic —
   * but fully best-effort: a failure here must never affect a transcription.
   */
  private async sweepLegacySeen(sessionId: string): Promise<void> {
    // BOTH pre-1.1.0 families. The dedup markers were `seen:<sid>:<msgId>` and the rate counter was
    // `rate:<sid>:<hour>` — one key per hour, also never deleted. Sweeping only the first would leave a
    // year-old install with ~8760 counter keys, i.e. the exact per-write stat cost this is here to remove.
    // The trailing ':' is what keeps these from matching the 1.1.0 keys `seen:<sid>` and `rate:<sid>`.
    //
    // Deliberately NO "this session is clean, stop looking" flag. Two attempts at one were both wrong: the
    // host's list() swallows its own errors and resolves [], so an empty result cannot be told apart from a
    // failed read, and any retirement rule built on that can strand the very keys this removes — a burst of
    // notes during one transient failure was enough. What the flag saved is a readdir of a directory this
    // release already made small, on a path that is rate-limited to a few dozen notes an hour. Not a trade.
    const prefixes = [`seen:${sessionId}:`, `rate:${sessionId}:`];
    try {
      for (const prefix of prefixes) {
        const stale = (await this.deps.store.list(prefix)).filter(k => k.startsWith(prefix));
        for (const key of stale.slice(0, LEGACY_SWEEP_PER_RUN)) await this.deps.store.delete(key);
      }
    } catch {
      /* best-effort cleanup — never surfaces, and retries on the next transcription */
    }
  }


  private async emit(
    sessionId: string,
    msg: IncomingMessage,
    o: {
      status: TranscriptionPayload['status'];
      reason?: string;
      /** Delivered long after it was sent: goes to the webhook and `self`, never quote-replied. */
      late?: boolean;
      text?: string;
      transcription?: NonNullable<TranscriptionPayload['transcription']>;
    },
  ): Promise<void> {
    if (o.status !== 'completed') {
      this.deps.logger.warn(`Transcription ${o.status}: ${o.reason}`, { messageId: msg.id });
    }
    const sent = new Date((msg.timestamp ?? 0) * 1000);
    const payload: TranscriptionPayload = {
      event: 'message.transcription',
      sessionId,
      messageId: msg.id,
      chatId: msg.chatId,
      status: o.status,
      source: 'speech-to-text',
      untrusted: true,
      ...(sent.getTime() > 0 ? { timestamp: msg.timestamp } : {}),
      ...(o.reason ? { reason: o.reason } : {}),
      ...(o.transcription ? { transcription: o.transcription } : {}),
    };
    // The webhook and in-chat deliveries are independent sinks — isolate the webhook so a failing
    // delivery endpoint can't swallow the in-chat transcript (and vice versa is unaffected: chat is last).
    if (this.deps.delivery) {
      try {
        await this.deps.delivery.deliver(payload);
      } catch (err) {
        this.deps.logger.warn('Transcript webhook delivery failed', {
          messageId: msg.id,
          error: err instanceof Error ? err.message : String(err),
        });
      }
    }
    if (o.status === 'completed' && o.text && this.deps.chat && this.deps.chatDelivery !== 'off') {
      if (this.deps.chatDelivery === 'self') await this.deps.chat.sendText(sessionId, msg.to, o.text);
      else if (!o.late) await this.deps.chat.reply(sessionId, msg.chatId, msg.id, o.text);
    }
  }
}