// Quarantine, and the want-list that makes evicting from it safe.
//
// An envelope whose signing key this observer has never seen is not wrong — a `device` record is
// published to a public repo, and that repo may simply not have been polled yet. So it waits. But
// waiting is unbounded memory unless it is bounded, and bounding it means eviction, and eviction on
// the private path is not the harmless thing it is on the public one: **no PDS holds a re-fetchable
// copy**. Dropping an envelope silently is dropping a record from this replica's history for good,
// and the fork would be invisible — version summaries would agree, because neither side knows the
// record was ever there.
//
// Hence the rule this file exists to enforce: *eviction is never silent loss*. Every eviction, and
// every refusal for capacity, writes a durable want entry naming exactly
// `(did, collection, rkey, recordCid)`, and catch-up re-requests wants EXPLICITLY rather than
// inferring completeness from summaries. The cost of a burst is a re-fetch. It is never a permanent
// divergence.
//
// What is NOT durable, stated so nobody has to infer it from the absence of a table: the held set
// itself. A restart drops whatever was still waiting on a key, and writes no want for it. That is
// covered, but by a different mechanism — a quarantined envelope was never admitted, so it is absent
// from `summarize()`, and the next `catchUp` asks for it again as a matter of course. The want list
// exists for the case that argument does NOT cover: an envelope this replica evicted while its peers
// believe it landed, which no summary diff would ever surface. `PrivateSpaceIngestor`'s gossip inbox
// has the same property for the same reason.

import type { PrivateEnvelope } from './envelope.js'
import { encode } from './cbor.js'
import { envelopeKey } from './envelope.js'
import { wantFor, wantKey, type WantEntry } from './admission.js'

/**
 * Records this observer knows it is missing. Durable — a want that only lived in memory would be
 * lost by the restart that most often follows the burst that caused it.
 */
export interface WantList {
  add(want: WantEntry): void
  /** Called when the record finally lands, from catch-up or from any peer's gossip. */
  clear(want: WantEntry): void
  all(): WantEntry[]
}

export class MemoryWantList implements WantList {
  readonly #wants = new Map<string, WantEntry>()

  add(want: WantEntry): void {
    this.#wants.set(wantKey(want), { ...want })
  }

  clear(want: WantEntry): void {
    this.#wants.delete(wantKey(want))
  }

  all(): WantEntry[] {
    return [...this.#wants.values()].sort((left, right) =>
      wantKey(left) < wantKey(right) ? -1 : wantKey(left) > wantKey(right) ? 1 : 0,
    )
  }
}

export interface QuarantineLimits {
  maxEnvelopes: number
  maxBytes: number
}

export const DEFAULT_QUARANTINE_LIMITS: QuarantineLimits = {
  maxEnvelopes: 1000,
  maxBytes: 8 * 1024 * 1024,
}

interface Held {
  envelope: PrivateEnvelope
  bytes: number
  /** Insertion order, so eviction is oldest-first and deterministic. */
  sequence: number
}

/**
 * A bounded holding area for envelopes waiting on a device key.
 *
 * Oldest-first eviction, by both count and bytes. Deterministic on purpose: two replicas fed the
 * same burst evict the same envelopes and write the same wants, so a test can assert the recovery
 * rather than hope for it.
 */
export class Quarantine {
  readonly #held = new Map<string, Held>()
  readonly #limits: QuarantineLimits
  readonly #wants: WantList
  #sequence = 0
  #bytes = 0

  constructor(wants: WantList, limits: QuarantineLimits = DEFAULT_QUARANTINE_LIMITS) {
    this.#wants = wants
    this.#limits = limits
  }

  get size(): number {
    return this.#held.size
  }

  get bytes(): number {
    return this.#bytes
  }

  /**
   * Hold an envelope, evicting older ones if it does not fit. Returns what was evicted — every
   * eviction has already been written to the want list, so a caller that ignores the return value
   * still cannot lose the record.
   */
  hold(envelope: PrivateEnvelope): WantEntry[] {
    const key = envelopeKey(envelope)
    if (this.#held.has(key)) return []
    // The whole ENVELOPE, not just its record: what is retained here is the envelope, and measuring
    // the record alone let a peer pin gigabytes behind a 207-byte accounting by padding the fields
    // around it. `admit()` caps those fields too, so both halves of the bound are real.
    const bytes = encode(envelope).length
    // An envelope that could never fit is refused outright rather than emptying the queue for a
    // record it would then evict anyway — but it still leaves a want, because it is still a record
    // this replica does not have.
    if (bytes > this.#limits.maxBytes) {
      const want = wantFor(envelope)
      this.#wants.add(want)
      return [want]
    }
    this.#sequence += 1
    this.#held.set(key, { envelope, bytes, sequence: this.#sequence })
    this.#bytes += bytes
    const evicted: WantEntry[] = []
    while (
      this.#held.size > this.#limits.maxEnvelopes ||
      this.#bytes > this.#limits.maxBytes
    ) {
      const oldest = [...this.#held.entries()].sort(
        ([, left], [, right]) => left.sequence - right.sequence,
      )[0]
      if (!oldest) break
      const [oldestKey, held] = oldest
      this.#held.delete(oldestKey)
      this.#bytes -= held.bytes
      const want = wantFor(held.envelope)
      this.#wants.add(want)
      evicted.push(want)
    }
    return evicted
  }

  /**
   * Take out everything signed by a device that has just become known, so the caller can re-run
   * admission over it. Draining is by `(did, deviceKeyId)` because that is exactly what changed.
   */
  drain(did: string, deviceKeyId: string): PrivateEnvelope[] {
    const drained: PrivateEnvelope[] = []
    for (const [key, held] of [...this.#held.entries()].sort(
      ([, left], [, right]) => left.sequence - right.sequence,
    )) {
      if (held.envelope.did !== did || held.envelope.deviceKeyId !== deviceKeyId) continue
      this.#held.delete(key)
      this.#bytes -= held.bytes
      drained.push(held.envelope)
    }
    return drained
  }

  /** Everything still waiting, oldest first — for diagnostics, so residual state is visible. */
  waiting(): PrivateEnvelope[] {
    return [...this.#held.values()]
      .sort((left, right) => left.sequence - right.sequence)
      .map((held) => held.envelope)
  }
}