import type { CryptoProvider, ShadeEnvelope } from '@shade/core';
import { encodeEnvelope } from '@shade/proto';
import { InboxClient, type FetchedBlob } from './client.js';
import {
  MemoryOutgoingQueueStore,
  type OutgoingEntry,
  type OutgoingQueueStore,
} from './queue-store.js';
import { MemoryCursorStore, type CursorStore } from './cursor-store.js';
import { computeMsgId } from './msg-id.js';
import { InboxClientEvents, type InboxClientListener } from './events.js';

/**
 * Caller-supplied incoming-blob handler. Receives raw wire bytes; the app
 * is expected to call `decodeEnvelope` + `Shade.receive` (or whatever
 * decrypt path it owns) and either return a sender-hint for telemetry
 * (the address the SDK extracted, or `null`) or throw to keep the blob
 * on the server for a later retry.
 *
 * V4.8: `raw.from` is the relay-captured sender fingerprint (8-byte hex
 * of SHA-256 over the sender's signing key). Empty when the relay is
 * pre-4.8 or didn't track the sender. Use it as the `fp:<hex>` bootstrap
 * label for X3DH first-contact handshakes.
 */
export type DecryptHandler = (
  raw: {
    msgId: string;
    ciphertext: Uint8Array;
    receivedAt: number;
    expiresAt: number;
    from?: string;
  },
) => Promise<string | null | undefined> | string | null | undefined;

export interface InboxOptions {
  /** Inbox-server base URL (e.g. "https://inbox.example.com"). */
  baseUrl: string;
  /** Recipient address — the address that owns this queue. */
  ownAddress: string;
  /** Crypto provider — used for signing requests. */
  crypto: CryptoProvider;
  /** Identity signing private key. */
  signingPrivateKey: Uint8Array;
  /** Identity signing public key. */
  signingPublicKey: Uint8Array;
  /**
   * Default TTL for outgoing PUTs (seconds). The server clamps to its own
   * min/max. Defaults to 7 days as mandated by the V3.6 spec.
   */
  defaultTtlSeconds?: number;
  /** Polling interval (ms). Default: 30s. Set to 0 to disable auto-poll. */
  pollIntervalMs?: number;
  /** Outgoing queue persistence. Defaults to in-memory. */
  queueStore?: OutgoingQueueStore;
  /** Cursor persistence. Defaults to in-memory. */
  cursorStore?: CursorStore;
  /** Override the underlying fetch (tests). */
  fetch?: typeof fetch;
  /** Maximum delivery attempts before dropping a queued entry. Default: 10. */
  maxAttempts?: number;
}

const DEFAULT_TTL_SECONDS = 7 * 24 * 60 * 60;
const DEFAULT_POLL_INTERVAL_MS = 30_000;
const DEFAULT_MAX_ATTEMPTS = 10;
/**
 * Cap for the cross-channel msgId dedup LRU. Each entry is a 64-char hex
 * string; 4096 entries ≈ 256 KiB of overhead, plenty of headroom for
 * bursty traffic (the LRU only needs to span the window between a bridge
 * push and the next inbox-poll catching up — typically 30 s × the
 * recipient's throughput).
 */
const DEFAULT_DEDUP_LRU_CAP = 4096;

/**
 * High-level inbox orchestrator.
 *
 * Responsibilities:
 *   - Outgoing: serialize ciphertext → queue → flush to server with retry.
 *   - Incoming: poll → decrypt-handler → ack on success.
 *   - Push hook: `onMessageQueued(handler)` lets a transport vendor (FCM,
 *     APNs) wake the recipient out-of-band when a blob is enqueued for them
 *     here. The hook is called with the recipient address — the actual
 *     push-delivery wire is left to the integrator.
 *
 * The class never *encrypts* — that's `Shade.send`'s job. Apps wire it up
 * like:
 *
 *   const envelope = await shade.send(bob, "hi");
 *   await inbox.send(bob, envelope);
 *
 * On Bob's device:
 *
 *   inbox.onIncoming(async (env) => {
 *     await shade.receive(senderAddress, env);
 *   });
 *   inbox.start();
 */
export class Inbox {
  private readonly client: InboxClient;
  private readonly queueStore: OutgoingQueueStore;
  private readonly cursorStore: CursorStore;
  private readonly events = new InboxClientEvents();
  private readonly defaultTtlSeconds: number;
  private readonly pollIntervalMs: number;
  private readonly maxAttempts: number;

  private incomingHandler: DecryptHandler | null = null;
  private flushTimer: ReturnType<typeof setTimeout> | null = null;
  private pollTimer: ReturnType<typeof setTimeout> | null = null;
  private flushing = false;
  private polling = false;
  private started = false;
  private registered = false;

  /**
   * Bounded msgId dedup window. Used by both the inbox-poll path
   * (`pollOnce` → `handleBlob`) and the bridge-push path
   * (`acceptBridgeFrame`). The relay stores blobs durably and pushes
   * them to every active delivery channel; without a shared dedup gate
   * here the recipient processes the same envelope twice — once from
   * the bridge, again from the next inbox-poll. The duplicate receive
   * trips on consumed one-time prekeys ("OPK not found") and pollutes
   * logs even when the canonical first delivery succeeded. See V4.8.3
   * Prism FR `cross-channel-duplicate-fanout-v4.8.2.md`.
   *
   * Insertion order is FIFO; the oldest msgId is evicted once the LRU
   * exceeds `DEFAULT_DEDUP_LRU_CAP`.
   */
  private readonly deliveredIds = new Set<string>();
  private readonly deliveredOrder: string[] = [];

  constructor(private readonly options: InboxOptions) {
    const clientOptions: ConstructorParameters<typeof InboxClient>[0] = {
      baseUrl: options.baseUrl,
      crypto: options.crypto,
      signingPrivateKey: options.signingPrivateKey,
    };
    if (options.fetch !== undefined) clientOptions.fetch = options.fetch;
    this.client = new InboxClient(clientOptions);
    this.queueStore = options.queueStore ?? new MemoryOutgoingQueueStore();
    this.cursorStore = options.cursorStore ?? new MemoryCursorStore();
    this.defaultTtlSeconds = options.defaultTtlSeconds ?? DEFAULT_TTL_SECONDS;
    this.pollIntervalMs = options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS;
    this.maxAttempts = options.maxAttempts ?? DEFAULT_MAX_ATTEMPTS;
  }

  /** Subscribe to client events (queued / delivered / received / failed). */
  on(listener: InboxClientListener): () => void {
    return this.events.on(listener);
  }

  /**
   * Push-hook: invoked once per new blob enqueued for delivery. Vendors
   * implement FCM/APNs/email wake-ups inside the handler. The handler
   * fires *after* the blob is durably in the local queue but before the
   * server has confirmed PUT.
   */
  onMessageQueued(handler: (recipientAddress: string, msgId: string) => void | Promise<void>): () => void {
    return this.events.on((e) => {
      if (e.name === 'inbox.message_queued') {
        Promise.resolve(handler(e.data.recipientAddress, e.data.msgId)).catch((err) => {
          console.error('[Shade] onMessageQueued handler threw:', err);
        });
      }
    });
  }

  /** Register a handler that processes incoming blobs. Replaces any prior. */
  onIncoming(handler: DecryptHandler): () => void {
    this.incomingHandler = handler;
    return () => {
      if (this.incomingHandler === handler) this.incomingHandler = null;
    };
  }

  /**
   * Idempotent TOFU registration with the server. Called automatically by
   * `start()`; can also be invoked directly e.g. during onboarding.
   */
  async register(): Promise<void> {
    if (this.registered) return;
    await this.client.register({
      address: this.options.ownAddress,
      signingKey: this.options.signingPublicKey,
    });
    this.registered = true;
    // V4.8: gate the first poll on register success. `start()` calls
    // `register()` fire-and-forget; without this kick, the very first
    // `pollOnce()` (scheduled synchronously alongside register) would
    // race the register HTTP RTT and return SHADE_NOT_FOUND. The
    // pollOnce() guard skips polls until `registered === true`; this
    // immediate schedule ensures we don't wait the full pollIntervalMs
    // for the next attempt once register lands.
    if (this.started) this.schedulePoll(0);
  }

  /** Drop the address from the server. Local queue/cursor are preserved. */
  async unregister(): Promise<void> {
    await this.client.unregister(this.options.ownAddress);
    this.registered = false;
  }

  /**
   * Enqueue an envelope for delivery. The actual PUT happens
   * asynchronously — the call returns once the entry is durably in the
   * outgoing queue. Returns the deterministic msgId.
   *
   * `envelope` accepts a `ShadeEnvelope` (the type returned from
   * `Shade.send`) or pre-encoded wire bytes (`Uint8Array`).
   */
  async send(args: {
    recipientAddress: string;
    envelope: ShadeEnvelope | Uint8Array;
    ttlSeconds?: number;
  }): Promise<string> {
    const ciphertext =
      args.envelope instanceof Uint8Array ? args.envelope : encodeEnvelope(args.envelope);
    const msgId = await computeMsgId(ciphertext);
    const ttlSeconds = args.ttlSeconds ?? this.defaultTtlSeconds;

    const entry: OutgoingEntry = {
      recipientAddress: args.recipientAddress,
      msgId,
      ciphertext,
      ttlSeconds,
      queuedAt: Date.now(),
      attempts: 0,
    };
    await this.queueStore.enqueue(entry);

    this.events.emit('inbox.message_queued', {
      recipientAddress: args.recipientAddress,
      msgId,
      bytes: ciphertext.length,
      ttlSeconds,
    });

    // Kick the flush loop. Don't await — caller doesn't need to block on
    // network for a "queued" return.
    this.scheduleFlush();
    return msgId;
  }

  /**
   * Force a flush + poll cycle now (useful right after registering or
   * after a push-trigger arrives). Does not throw on transient errors.
   */
  async tick(): Promise<{ flushed: number; received: number }> {
    const flushResult = await this.flushOnce();
    // `null` means another flush was concurrent; report 0 newly-flushed
    // for this caller (the other flush counted them).
    const flushed = flushResult?.delivered ?? 0;
    const received = await this.pollOnce();
    return { flushed, received };
  }

  /** Start background flush + poll timers. Idempotent. */
  start(): void {
    if (this.started) return;
    this.started = true;
    this.register().catch((err) => {
      console.warn('[Shade] Inbox register failed (will retry):', (err as Error).message);
      this.scheduleRegisterRetry();
    });
    this.scheduleFlush();
    // V4.8: do NOT schedule the first poll synchronously. `register()`
    // success kicks `schedulePoll(0)` so the first poll fires after the
    // server has acknowledged the address. Pre-fix this raced register
    // and burned a 404 on every fresh-address `start()`.
    if (this.registered) this.schedulePoll(0);
  }

  /** Stop background timers. Pending entries remain in the queue. */
  stop(): void {
    this.started = false;
    if (this.flushTimer) {
      clearTimeout(this.flushTimer);
      this.flushTimer = null;
    }
    if (this.pollTimer) {
      clearTimeout(this.pollTimer);
      this.pollTimer = null;
    }
  }

  /** Number of entries currently waiting to be flushed. */
  async pendingCount(): Promise<number> {
    return this.queueStore.size();
  }

  // ─── internals ──────────────────────────────────────────────

  private scheduleRegisterRetry(): void {
    if (!this.started) return;
    setTimeout(() => {
      if (!this.started) return;
      this.register().catch(() => this.scheduleRegisterRetry());
    }, 5000);
  }

  private scheduleFlush(delayMs = 0): void {
    if (this.flushTimer) return;
    this.flushTimer = setTimeout(() => {
      this.flushTimer = null;
      this.flushOnce()
        .then((result) => {
          // `result === null` means another flush was already in flight
          // and this call early-returned via the `flushing` guard. The
          // already-running flush will reschedule itself when it
          // finishes; do not double-schedule from here.
          if (result === null) return;
          if (result.remaining === 0) return;
          // V4.8.5 — distinguish healthy-drain-but-more-queued from
          // all-attempts-failed. Pre-fix, both cases used a 15 s
          // backoff. Under sustained traffic (Prism's typing-into-a-
          // chatty-shell pattern), bursts of envelopes enqueued
          // *during* a flush would sit ~10–15 s behind the backoff
          // timer before the next drain — visible to the receiver as a
          // "10 s silence then 25-frame burst" wave. Healthy drain
          // (delivered > 0) means the network is fine and we should
          // immediately drain whatever piled up; reserve the 15 s
          // retry for the case where every attempt this round failed.
          if (this.started) {
            const delay = result.delivered > 0 ? 0 : 15_000;
            this.scheduleFlush(delay);
          }
        })
        .catch(() => {
          if (this.started) this.scheduleFlush(15_000);
        });
    }, delayMs);
  }

  private schedulePoll(delayMs: number): void {
    if (!this.started || this.pollIntervalMs === 0) return;
    if (this.pollTimer) clearTimeout(this.pollTimer);
    this.pollTimer = setTimeout(() => {
      this.pollTimer = null;
      this.pollOnce()
        .catch(() => {})
        .finally(() => this.schedulePoll(this.pollIntervalMs));
    }, delayMs);
  }

  /**
   * Drain the outgoing queue. Returns `null` if another flush is already
   * in flight (the running flush owns the rescheduling); otherwise
   * returns the count of newly-delivered envelopes and the queue size
   * after the drain so the caller can decide whether to immediately
   * re-flush (more piled up during the drain — healthy network) or
   * back off (everything failed).
   *
   * V4.8.5: drain is parallel-per-recipient. Each `recipientAddress`
   * gets its own sequential worker (so per-peer order is preserved),
   * but distinct recipients run concurrently. Pre-fix, a single slow
   * POST head-of-line-blocked the entire queue — including small
   * frames bound for unrelated peers. See Prism FR
   * `per-recipient-flush-concurrency-v4.8.md`.
   */
  private async flushOnce(): Promise<{ delivered: number; remaining: number } | null> {
    if (this.flushing) return null;
    this.flushing = true;
    let delivered = 0;
    try {
      const entries = await this.queueStore.list();
      // Group by recipient. Within a bucket we drain sequentially so
      // per-peer message order matches enqueue order (the relay
      // assigns `receivedAt` on PUT arrival; concurrent POSTs to the
      // same peer would let the second arrive first and the recipient
      // would observe out-of-order envelopes). Across buckets, no
      // ordering guarantee exists in Shade's wire model anyway, so
      // parallel drain is safe.
      const buckets = new Map<string, OutgoingEntry[]>();
      for (const entry of entries) {
        let bucket = buckets.get(entry.recipientAddress);
        if (!bucket) {
          bucket = [];
          buckets.set(entry.recipientAddress, bucket);
        }
        bucket.push(entry);
      }

      const drainBucket = async (bucket: OutgoingEntry[]): Promise<number> => {
        let count = 0;
        for (const entry of bucket) {
          try {
            const result = await this.client.put({
              recipientAddress: entry.recipientAddress,
              senderSigningKey: this.options.signingPublicKey,
              envelope: entry.ciphertext,
              ttlSeconds: entry.ttlSeconds,
            });
            await this.queueStore.remove(entry.recipientAddress, entry.msgId);
            count++;
            this.events.emit('inbox.message_delivered', {
              recipientAddress: entry.recipientAddress,
              msgId: result.msgId,
              idempotent: result.idempotent,
            });
          } catch (err) {
            await this.queueStore.bumpAttempts(entry.recipientAddress, entry.msgId);
            const attempts = entry.attempts + 1;
            this.events.emit('inbox.message_failed', {
              recipientAddress: entry.recipientAddress,
              msgId: entry.msgId,
              attempts,
              error: (err as Error).message,
            });
            if (attempts >= this.maxAttempts) {
              await this.queueStore.remove(entry.recipientAddress, entry.msgId);
            }
          }
        }
        return count;
      };

      const counts = await Promise.all(
        Array.from(buckets.values(), drainBucket),
      );
      delivered = counts.reduce((a, b) => a + b, 0);
    } finally {
      this.flushing = false;
    }
    const remaining = await this.queueStore.size();
    return { delivered, remaining };
  }

  private async pollOnce(): Promise<number> {
    if (this.polling) return 0;
    if (!this.incomingHandler) return 0;
    this.polling = true;
    let total = 0;
    try {
      let cursor = await this.cursorStore.load(this.options.ownAddress);
      // Pull all pages.
      // eslint-disable-next-line no-constant-condition
      while (true) {
        let page;
        try {
          page = await this.client.fetch({
            address: this.options.ownAddress,
            sinceCursor: cursor,
          });
        } catch (err) {
          // Surface but don't crash the loop.
          console.warn('[Shade] Inbox fetch failed:', (err as Error).message);
          break;
        }
        for (const blob of page.blobs) {
          const handled = await this.handleBlob(blob);
          if (handled) total++;
          if (blob.receivedAt > cursor) cursor = blob.receivedAt;
        }
        await this.cursorStore.save(this.options.ownAddress, cursor);
        this.events.emit('inbox.poll_completed', {
          ownAddress: this.options.ownAddress,
          count: page.blobs.length,
          cursor,
        });
        if (!page.hasMore) break;
      }
    } finally {
      this.polling = false;
    }
    return total;
  }

  /**
   * Feed a blob delivered by a bridge transport (WS / SSE / long-poll
   * push) into the same dispatch + ack pipeline that `pollOnce` uses.
   *
   * Wire-up pattern:
   * ```ts
   * const bridge = new FallbackBridgeTransport([...]);
   * await bridge.connect({
   *   onMessage: async (msg) => {
   *     await inbox.acceptBridgeFrame({
   *       msgId: msg.msgId!,           // present on v4.8+ relays
   *       ciphertext: msg.bytes,
   *       receivedAt: msg.receivedAt,
   *       expiresAt: msg.expiresAt ?? Date.now() + 7 * 24 * 3600 * 1000,
   *       ...(msg.from !== undefined ? { from: msg.from } : {}),
   *     });
   *   },
   * });
   * ```
   *
   * The Inbox's bounded msgId LRU is shared between this path and
   * `pollOnce`, so whichever channel delivers first wins; the
   * other channel acks-and-skips when the same msgId comes back
   * around. Both paths also DELETE the blob from the relay on success
   * so subsequent polls don't see it either.
   *
   * Returns `true` if the blob was newly dispatched, `false` if it
   * was a duplicate or rejected by the handler (handler still gets a
   * chance to retry on the next poll if it threw).
   */
  async acceptBridgeFrame(blob: FetchedBlob): Promise<boolean> {
    return this.handleBlob(blob);
  }

  private async handleBlob(blob: FetchedBlob): Promise<boolean> {
    if (!this.incomingHandler) return false;

    // Cross-channel msgId dedup. If the bridge already delivered this
    // blob, the inbox-poll copy must not re-dispatch (would re-trigger
    // X3DH / consume an OPK we no longer have). We still ack so the
    // relay drops the now-redundant copy.
    if (this.deliveredIds.has(blob.msgId)) {
      await this.ackQuietly(blob.msgId);
      return false;
    }

    // Defense-in-depth: verify msgId ↔ ciphertext at the client too. A
    // server bug or malicious operator can't sneak a different blob past
    // the client's hash check.
    const recomputed = await computeMsgId(blob.ciphertext);
    if (recomputed !== blob.msgId) {
      this.events.emit('inbox.message_decrypt_failed', {
        msgId: blob.msgId,
        error: 'msgId/ciphertext mismatch',
      });
      // Don't ack — let the operator notice the divergence.
      return false;
    }

    let senderHint: string | null = null;
    try {
      const raw: {
        msgId: string;
        ciphertext: Uint8Array;
        receivedAt: number;
        expiresAt: number;
        from?: string;
      } = {
        msgId: blob.msgId,
        ciphertext: blob.ciphertext,
        receivedAt: blob.receivedAt,
        expiresAt: blob.expiresAt,
      };
      if (blob.from !== undefined) raw.from = blob.from;
      const result = await this.incomingHandler(raw);
      senderHint = result ?? null;
    } catch (err) {
      this.events.emit('inbox.message_decrypt_failed', {
        msgId: blob.msgId,
        error: (err as Error).message,
      });
      // Don't ack — caller can retry on next poll.
      return false;
    }

    // Mark before the ack so a slow-network ack doesn't leave a window
    // where a parallel pollOnce sees the same msgId and re-dispatches.
    this.recordDelivered(blob.msgId);
    await this.ackQuietly(blob.msgId);
    this.events.emit('inbox.message_received', {
      senderHint,
      msgId: blob.msgId,
    });
    return true;
  }

  private async ackQuietly(msgId: string): Promise<void> {
    try {
      await this.client.ack({ address: this.options.ownAddress, msgId });
    } catch (err) {
      // Dispatch (or skip) succeeded; the ack just failed. Next poll
      // will see the blob again and the dedup gate above will skip it.
      console.warn('[Shade] Inbox ack failed (will retry):', (err as Error).message);
    }
  }

  private recordDelivered(msgId: string): void {
    if (this.deliveredIds.has(msgId)) return;
    this.deliveredIds.add(msgId);
    this.deliveredOrder.push(msgId);
    if (this.deliveredOrder.length > DEFAULT_DEDUP_LRU_CAP) {
      const evicted = this.deliveredOrder.shift()!;
      this.deliveredIds.delete(evicted);
    }
  }
}
