import db from '@adonisjs/lucid/services/db'
import logger from '@adonisjs/core/services/logger'
import { setTimeout as sleep } from 'node:timers/promises'
import env from '#start/env'
import coordinatorConfig from '#config/coordinator'
import { XoverClient } from '#lib/nntp/index'
import { YencError } from '#lib/yenc/index'
import { inspect, innerName, PROBE_VERSION, type Inspection } from '#lib/probe/index'
import { NntpShare } from '#services/nntp_shares'
import { DiskSpace } from '#services/disk_space'
import { decidePar2Completeness } from '#services/par2_completeness'
import { twinKey, twinNames } from '#services/hash_twins'
import PendingLadder, { type PendingProbe } from '#services/pending_ladder'
import { PendingNamer } from '#services/pending_namer'

interface Candidate {
  id: number
  releaseId: number
  kind: string
  messageId: string
  /** A pending release's target (the pending ladder), not a catalog candidate: id and releaseId are 0. */
  pending?: PendingProbe
}

type Outcome = { candidate: Candidate; status: 'done' | 'missing' | 'error' | 'waiting'; inspection?: Inspection; cached?: boolean }

/**
 * The order candidates are probed in: what the par2 merge needs first (a random-named file's own start, then the
 * par2 sets and archives of releases without a readable name), then the rest (names and passwords of readable ones),
 * then 7z listings (the end of a split 7z's last volume; queued from 2026-10-13, probed from 2026-10-23).
 */
const TIERS: { kind: string; unreadable: boolean }[] = [
  { kind: 'lone', unreadable: false },
  { kind: 'par2', unreadable: true },
  { kind: 'archive', unreadable: true },
  { kind: 'par2', unreadable: false },
  { kind: 'archive', unreadable: false },
  { kind: '7z-tail', unreadable: true },
  { kind: '7z-tail', unreadable: false },
]
const FLUSH_MS = 2000
/** Between sweeps that discard the waiting releases of dropped signatures. */
const SWEEP_MS = 5 * 60_000
/** A flush that deadlocks (against the batch store's pending upsert) keeps its outcomes for the next, this often. */
const DEADLOCK_RETRIES = 5

/**
 * JSON for a jsonb parameter, without NUL characters: jsonb can't hold \u0000 (a name read from an article had one),
 * and the whole flush failed with it (LOG.md 2026-10-18 (78)).
 */
function jsonb(value: unknown): string {
  return JSON.stringify(value, (_key, v) => (typeof v === 'string' ? v.replaceAll('\0', '') : v))
}

/** Rounds of every lane's pipeline still queued when the next claim starts. */
const LOW_ROUNDS = 8

/**
 * The candidate queue's levels: the next claim starts below `low`, LOW_ROUNDS rounds of every lane's pipeline (at
 * ~0.2 s an article and a pipeline of 8, ~13 s of work), and fills it to `target`, twice that. A claim takes 1.5–4 s
 * and more (random slices of an 11 M-row table); refilling at half of four rounds left 3–6 s, and at 20–24 lanes the
 * lanes waited for candidates up to 19% of their time (LOG.md 2026-10-28 (120)). At least `minBatch` either way.
 */
export function queueLevels(lanes: number, pipeline: number, minBatch: number): { low: number; target: number } {
  const low = Math.max(minBatch, lanes * pipeline * LOW_ROUNDS)
  return { low, target: 2 * low }
}

/** Slices a claim takes from random points of the queue. */
const SLICES = 20
const REPORT_MS = 60_000

/**
 * Phase 2's first step (REBUILD.md, part 6): reads the probe queue's
 * articles (settled releases first, by kind) over PROBE_CONNECTIONS
 * connections of its own, without compression, and stores what they show:
 * per release and kind an Inspection, and on the release whether it's
 * passworded and its inner name. One process: fetching (~0.1 s an article)
 * dwarfs parsing (a few ms), so there's no queue between them.
 */
export class Prober {
  private queue: Candidate[] = []
  private outcomes: Outcome[] = []
  private refilling: Promise<void> | null = null
  private stopped = false
  private stats = Prober.freshStats()
  /** Flushes deadlocked in a row, their outcomes kept. */
  private deadlocks = 0
  /** Lanes running (each one connection, once it has work); at most the share's grant. Counted apart from the set: a lane leaving takes itself off at once. */
  private lanes = new Set<Promise<void>>()
  private running = 0
  /** The last lane to look found nothing to claim: the prober wants its minimum only. */
  private dry = false
  /** What it stores (inspections, par2 file lists, the WAL) fills the NVMe: no claims below PROBE_MIN_FREE_GB. */
  private space = new DiskSpace('probing', '/', coordinatorConfig.probe.minFreeGb)
  private share = new NntpShare('prober', coordinatorConfig.nntpShares.proberPriority, coordinatorConfig.probe.minConnections, coordinatorConfig.probe.connections)

  /** Per minute: what was read, and where the lanes' time went (ms summed over lanes; flushes ms on their own). */
  private static freshStats() {
    return { articles: 0, bytes: 0, done: 0, missing: 0, error: 0, cached: 0, pending: 0, discarded: 0, waitMs: 0, idleMs: 0, fetchMs: 0, inspectMs: 0, flushMs: 0, claimMs: 0, since: Date.now() }
  }
  /** The pending ladder's namer: named pending releases into the catalog, alongside the lanes. */
  private namer = new PendingNamer()

  async run() {
    // A previous run's claims: one prober at a time.
    const { rowCount } = await db.rawQuery(`UPDATE probe_candidates SET status = 'waiting' WHERE status = 'taken'`)
    if (rowCount) logger.info(`Probing: ${rowCount} candidates from a previous run back in the queue.`)
    const released = await PendingLadder.release()
    if (released) logger.info(`Probing: ${released} pending releases from a previous run back on the ladder.`)
    const namer = this.namer.run(() => this.stopped).catch((error) => logger.error(`Naming failed: ${error.message}`))
    // The ladder's discard sweep beside the lanes: a dropped signature's rows take minutes on the HDD (LOG.md (133)).
    const sweeper = (async () => {
      while (!this.stopped) {
        try {
          this.stats.discarded += await PendingLadder.discard()
        } catch (error: any) {
          logger.error(`Probing: the discard sweep failed: ${error.message}`)
        }
        for (let waited = 0; waited < SWEEP_MS && !this.stopped; waited += 1000) await sleep(1000)
      }
    })()
    logger.info(`Probing with ${this.share.min}–${this.share.max} connections (its share of the account's).`)
    const flusher = (async () => {
      while (!this.stopped || this.outcomes.length > 0) {
        await sleep(FLUSH_MS)
        const flushed = Date.now()
        await this.flush().catch((error) => logger.error(`Probing: storing results failed: ${error.message}`))
        this.stats.flushMs += Date.now() - flushed
        this.report()
      }
    })()
    // Lanes as the share allows: the coordinator's history comes first (app/services/nntp_shares.ts).
    while (!this.stopped) {
      try {
        const was = this.share.granted
        const granted = await this.share.sync(this.dry ? 0 : this.share.max, this.running)
        if (granted !== was) logger.info(`Probing: ${granted} connections (was ${was}).`)
        while (this.running < granted) {
          this.running++
          const lane: Promise<void> = this.lane()
            .catch((error) => logger.error(`Probing: a lane failed: ${error.message}`))
            .finally(() => this.lanes.delete(lane))
          this.lanes.add(lane)
        }
      } catch (error: any) {
        logger.error(`Probing: NNTP share failed: ${error.message}`)
      }
      for (let waited = 0; waited < coordinatorConfig.nntpShares.everyMs && !this.stopped; waited += 1000) await sleep(1000)
    }
    await Promise.all(this.lanes)
    await namer
    await sweeper
    await this.share.leave().catch(() => {})
    // What was claimed and not probed goes back.
    for (const candidate of this.queue.splice(0)) this.outcomes.push({ candidate, status: 'waiting' })
    this.stopped = true
    await flusher
  }

  stop() {
    this.stopped = true
  }

  /**
   * One connection's loop: PROBE_PIPELINE articles at a time (pipelined), reconnecting after a failure; done when
   * there are more lanes than the share (its connection closed, the coordinator takes it on its next sync).
   */
  private async lane() {
    let client: XoverClient | null = null
    let usedAt = Date.now()
    try {
      while (!this.stopped) {
        if (this.running > this.share.granted) break
        const batch: Candidate[] = []
        const waited = Date.now()
        while (batch.length < coordinatorConfig.probe.pipeline) {
          const candidate = await this.next()
          if (!candidate) break
          batch.push(candidate)
        }
        this.stats.waitMs += Date.now() - waited
        this.dry = batch.length === 0
        if (batch.length === 0) {
          // Nothing to probe for a while: the connection closes (nntpIdleCloseMs), and opens again with work.
          const idleMs = coordinatorConfig.nntpIdleCloseMs
          if (client && idleMs > 0 && Date.now() - usedAt >= idleMs) {
            client.close()
            client = null
          }
          this.stats.idleMs += 30_000
          await sleep(30_000)
          continue
        }
        try {
          client ??= await XoverClient.connect({
            host: env.get('NNTP_HOST')!,
            port: env.get('NNTP_PORT', 563),
            user: env.get('NNTP_USER'),
            password: env.get('NNTP_PASSWORD'),
            secure: env.get('NNTP_SECURE', true),
            compress: false,
            timeoutMs: coordinatorConfig.nntpTimeoutMs,
          })
          const fetched = Date.now()
          const bodies = await client.bodies(batch.map((c) => c.messageId))
          this.stats.fetchMs += Date.now() - fetched
          const inspected = Date.now()
          for (const [i, candidate] of batch.entries()) {
            const body = bodies[i]
            this.stats.articles++
            if (body === null) {
              // Taken down or expired: final.
              this.outcomes.push({ candidate, status: 'missing' })
              continue
            }
            this.stats.bytes += body.length
            try {
              this.outcomes.push({ candidate, status: 'done', inspection: inspect(body, candidate.kind) })
            } catch (error: any) {
              // Not yEnc, a broken one, or data a parser chokes on: a final answer, not a connection problem.
              if (!(error instanceof YencError)) logger.warn(`Probing ${candidate.messageId}: ${error.message}`)
              this.outcomes.push({ candidate, status: 'error' })
            }
          }
          this.stats.inspectMs += Date.now() - inspected
          usedAt = Date.now()
        } catch (error: any) {
          logger.warn(`Probing ${batch[0].messageId} and ${batch.length - 1} more: ${error.message}`)
          for (const candidate of batch) this.outcomes.push({ candidate, status: 'waiting' })
          client?.close()
          client = null
          await sleep(5000)
        }
      }
    } finally {
      client?.close()
      this.running--
    }
  }

  /** The queue's low-water mark and the level a claim fills it to (queueLevels). */
  private get levels() {
    return queueLevels(this.running, coordinatorConfig.probe.pipeline, coordinatorConfig.probe.batch)
  }

  /** Claims to the queue's target; the claim after starts when it's below its low-water mark (queueLevels). */
  private get batch() {
    return this.levels.target
  }

  private async next(): Promise<Candidate | undefined> {
    if (this.queue.length < this.levels.low) {
      this.refilling ??= this.claim().finally(() => (this.refilling = null))
      if (this.queue.length === 0) await this.refilling
    }
    return this.queue.shift()
  }

  /**
   * Waiting candidates of settled releases, by kind, claimed so another round
   * doesn't take them: SLICES slices from random points of the queue. In id
   * order it went group by group, and some groups' articles take 0.6–3 s
   * where most take 20–60 ms (a.b.zogg 2016: 7 articles/s on 6 connections).
   */
  private async claim() {
    const started = Date.now()
    try {
      await this.claimTiers()
    } finally {
      this.stats.claimMs += Date.now() - started
    }
  }

  private async claimTiers() {
    // The queue empties, the lanes go dry and the share shrinks to its minimum until there's room again.
    if (await this.space.short()) return
    await this.claimCatalog()
    await this.claimPending()
  }

  /** The pending ladder's share of a claim: at least pendingShare of the batch, and whatever the catalog's tiers left. */
  private async claimPending() {
    const want = Math.max(Math.floor(this.batch * coordinatorConfig.probe.pendingShare), this.batch - this.queue.length)
    const probes = await PendingLadder.claim(want)
    if (probes.length === 0) return
    const { rows: cached } = await db.rawQuery(`SELECT message_id, status, result FROM probe_cache WHERE message_id = ANY(?) AND version = ?`, [
      probes.map((p) => p.messageId),
      PROBE_VERSION,
    ])
    const known = new Map(cached.map((c: any) => [c.message_id, c]))
    for (const probe of probes) {
      const candidate: Candidate = { id: 0, releaseId: 0, kind: probe.kind, messageId: probe.messageId, pending: probe }
      const hit: any = known.get(probe.messageId)
      if (hit) {
        this.stats.cached++
        this.outcomes.push({ candidate, status: hit.status, inspection: hit.result ?? undefined, cached: true })
      } else {
        this.queue.push(candidate)
      }
    }
  }

  private async claimCatalog() {
    for (const { kind, unreadable } of TIERS) {
      const want = this.batch - this.queue.length
      if (want <= 0) return
      // The starting points chosen here: in the query Postgres drew random() once for all slices.
      // Over the claimable ones of this tier only: the rest of the queue isn't settled yet.
      // Per value of unreadable, each an end of the index: over `unreadable >= false` Postgres walked every waiting
      // entry of the kind, 2.5 minutes a claim at 28M archive candidates (LOG.md 2026-10-18 (74)).
      const values = unreadable ? [true] : [false, true]
      const ends = values
        .map(() => `SELECT min(id) AS lo, max(id) AS hi FROM probe_candidates WHERE status = 'waiting' AND ready AND kind = ? AND unreadable = ?`)
        .join(' UNION ALL ')
      const [{ lo, hi }] = (
        await db.rawQuery(
          `SELECT min(lo) AS lo, max(hi) AS hi FROM (${ends}) e`,
          values.flatMap((v) => [kind, v])
        )
      ).rows
      if (lo === null) continue
      const starts = Array.from({ length: SLICES }, () => Number(lo) + Math.floor(Math.random() * (Number(hi) - Number(lo) + 1)))
      const { rows } = await db.rawQuery(
        `UPDATE probe_candidates SET status = 'taken' WHERE id IN (
           SELECT pc.id FROM unnest(?::bigint[]) AS o(start)
           CROSS JOIN LATERAL (
             SELECT pc.id FROM probe_candidates pc
             WHERE pc.status = 'waiting' AND pc.ready AND pc.kind = ? AND pc.unreadable >= ? AND pc.id >= o.start
             ORDER BY pc.kind, pc.unreadable, pc.id LIMIT ? FOR UPDATE SKIP LOCKED) pc)
         RETURNING id, release_id, kind, message_id`,
        [starts, kind, unreadable, Math.ceil(want / SLICES)]
      )
      const claimed = rows.map((r: any) => ({ id: Number(r.id), releaseId: Number(r.release_id), kind: r.kind, messageId: r.message_id }))
      // Probed before (an earlier catalog's candidate of the same article): answered from probe_cache.
      const { rows: cached } = await db.rawQuery(`SELECT message_id, status, result FROM probe_cache WHERE message_id = ANY(?) AND version = ?`, [
        claimed.map((c: Candidate) => c.messageId),
        PROBE_VERSION,
      ])
      const known = new Map(cached.map((c: any) => [c.message_id, c]))
      for (const candidate of claimed) {
        const hit: any = known.get(candidate.messageId)
        if (hit) {
          this.stats.cached++
          this.outcomes.push({ candidate, status: hit.status, inspection: hit.result ?? undefined, cached: true })
        } else {
          this.queue.push(candidate)
        }
      }
    }
  }

  /**
   * The outcomes so far: candidates' status, inspections, and the releases' passworded and inner name. A deadlock
   * (the batch store's pending upsert locks the same rows; LOG.md (134)) keeps the outcomes for the next flush
   * instead of losing the probes.
   */
  private async flush() {
    const taken = this.outcomes.splice(0)
    if (taken.length === 0) return
    try {
      await this.store(taken)
    } catch (error: any) {
      if (error?.code === '40P01' && this.deadlocks < DEADLOCK_RETRIES) {
        this.deadlocks++
        this.outcomes.unshift(...taken)
        logger.warn(`Probing: storing results deadlocked, kept for the next flush (${this.deadlocks}).`)
        return
      }
      throw error
    }
    this.deadlocks = 0
    for (const o of taken) if (o.status !== 'waiting') this.stats[o.status]++
    this.stats.pending += taken.filter((o) => o.candidate.pending && o.status !== 'waiting').length
  }

  private async store(taken: Outcome[]) {
    // The ladder's outcomes apart from the catalog's: the next target, a name for the namer, or the end of the climb.
    const pending = taken.filter((o) => o.candidate.pending)
    const outcomes = taken.filter((o) => !o.candidate.pending)
    await db.transaction(async (trx) => {
      if (outcomes.length > 0) {
        await trx.rawQuery(
          `UPDATE probe_candidates pc SET status = v.status FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, status text) WHERE pc.id = v.id`,
          [jsonb(outcomes.map((o) => ({ id: o.candidate.id, status: o.status })))]
        )
      }
      await PendingLadder.record(pending.map((o) => ({ probe: o.candidate.pending!, status: o.status, inspection: o.inspection })), trx)
      // Final answers read from the news server, kept by message-id for the next catalog (a reset).
      // One per message-id: the same article is a candidate of more than one release, and ON CONFLICT can't take a
      // row twice in one statement.
      const fresh = [...new Map(taken.filter((o) => !o.cached && o.status !== 'waiting').map((o) => [o.candidate.messageId, o])).values()]
      if (fresh.length > 0) {
        await trx.rawQuery(
          `INSERT INTO probe_cache (message_id, status, result, version)
           SELECT v.message_id, v.status, v.result, ? FROM jsonb_to_recordset(?::jsonb) AS v(message_id text, status text, result jsonb)
           ON CONFLICT (message_id) DO UPDATE SET status = excluded.status, result = excluded.result, version = excluded.version, created_at = now()`,
          [PROBE_VERSION, jsonb(fresh.map((o) => ({ message_id: o.candidate.messageId, status: o.status, result: o.inspection ?? null })))]
        )
      }
      const done = outcomes.filter((o) => o.inspection)
      if (done.length === 0) return
      await trx.rawQuery(
        `INSERT INTO release_inspections (release_id, candidate_id, kind, result)
         SELECT v.release_id, v.candidate_id, v.kind, v.result FROM jsonb_to_recordset(?::jsonb) AS v(release_id bigint, candidate_id bigint, kind text, result jsonb)
         ON CONFLICT (release_id, candidate_id) DO UPDATE SET result = excluded.result, created_at = now()`,
        // A par2 set's file list goes to par2_files, not into every inspection.
        [jsonb(done.map((o) => ({ release_id: o.candidate.releaseId, candidate_id: o.candidate.id, kind: o.candidate.kind, result: { ...o.inspection, par2: undefined } })))]
      )
      await this.storeForMerge(trx, done)
      // A par2 index read: the release's completeness by its set, when it has no counter.
      await decidePar2Completeness(trx, [...new Set(done.filter((o) => o.inspection!.par2).map((o) => o.candidate.releaseId))])
      // Each touched release from all its inspections (earlier rounds may have stored others).
      const ids = [...new Set(done.map((o) => o.candidate.releaseId))]
      const { rows } = await trx.rawQuery(
        `SELECT i.release_id, i.kind, i.result, r.name FROM release_inspections i JOIN releases r ON r.id = i.release_id WHERE i.release_id = ANY(?)`,
        [ids]
      )
      const byRelease = new Map<number, { name: string; all: { kind: string; result: Inspection }[] }>()
      for (const r of rows) {
        const id = Number(r.release_id)
        const entry = byRelease.get(id) ?? { name: r.name as string, all: [] as { kind: string; result: Inspection }[] }
        entry.all.push({ kind: r.kind, result: r.result })
        byRelease.set(id, entry)
      }
      // A twin by content names a release its own inspections didn't (hash_twins.ts; LOG.md (130)).
      const twins = await twinNames(
        done.filter((o) => o.inspection!.hash16k && o.inspection!.size).map((o) => ({ hash16k: o.inspection!.hash16k!, size: o.inspection!.size })),
        trx
      )
      const twinOf = new Map<number, string>()
      for (const o of done) {
        const twin = o.inspection!.hash16k ? twins.get(twinKey(o.inspection!.hash16k, o.inspection!.size)) : undefined
        if (twin && twin !== byRelease.get(o.candidate.releaseId)?.name) twinOf.set(o.candidate.releaseId, twin)
      }
      const updates = [...byRelease].map(([id, { name, all }]) => ({
        id,
        passworded: all.some((i) => i.result.passworded === true),
        inner_name: innerName(all, name) ?? twinOf.get(id) ?? null,
      }))
      await trx.rawQuery(
        `UPDATE releases r SET passworded = v.passworded, inner_name = v.inner_name, inspected_at = now()
         FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, passworded boolean, inner_name text) WHERE r.id = v.id`,
        [jsonb(updates)]
      )
    })
  }

  /** What the par2 merge matches (app/services/par2_merger.ts): files' first-16-KiB hashes, par2 sets' file lists. */
  private async storeForMerge(trx: any, done: Outcome[]) {
    const hashed = done.filter((o) => o.inspection!.hash16k)
    if (hashed.length > 0) {
      await trx.rawQuery(
        `INSERT INTO file_hashes (candidate_id, release_id, hash16k, size)
         SELECT v.candidate_id, v.release_id, v.hash16k, v.size FROM jsonb_to_recordset(?::jsonb) AS v(candidate_id bigint, release_id bigint, hash16k text, size bigint)
         ON CONFLICT (candidate_id) DO NOTHING`,
        [jsonb(hashed.map((o) => ({ candidate_id: o.candidate.id, release_id: o.candidate.releaseId, hash16k: o.inspection!.hash16k, size: o.inspection!.size })))]
      )
    }
    for (const o of done) {
      const set = o.inspection!.par2
      if (!set) continue
      await trx.rawQuery(
        `INSERT INTO par2_files (set_id, file_id, hash16k, length, name)
         SELECT ?, v.file_id, v.hash16k, v.length, v.name FROM jsonb_to_recordset(?::jsonb) AS v(file_id text, hash16k text, length bigint, name text)
         ON CONFLICT DO NOTHING`,
        [set.setId, jsonb(set.files.map((f) => ({ file_id: f.fileId, hash16k: f.hash16k, length: f.length, name: f.name })))]
      )
      await trx.rawQuery(`INSERT INTO par2_sets (set_id, release_id, files) VALUES (?, ?, ?) ON CONFLICT DO NOTHING`, [set.setId, o.candidate.releaseId, set.files.length])
    }
  }

  private report() {
    const seconds = (Date.now() - this.stats.since) / 1000
    if (seconds < REPORT_MS / 1000) return
    const { articles, bytes, done, missing, error } = this.stats
    logger.info(
      `Probed ${articles} articles in ${seconds.toFixed(0)} s: ${(articles / seconds).toFixed(1)}/s, ${((bytes * 8) / seconds / 1e6).toFixed(0)} Mbit/s; ` +
        `${done} inspected, ${missing} missing, ${error} not yEnc; ${this.stats.cached} from the probe cache; ` +
        `the pending ladder ${this.stats.pending} probed, ${this.stats.discarded} discarded by signature. ` +
        `Lanes: fetching ${(this.stats.fetchMs / 1000).toFixed(0)} s, waiting for candidates ${(this.stats.waitMs / 1000).toFixed(0)} s, ` +
        `idle ${(this.stats.idleMs / 1000).toFixed(0)} s, inspecting ${(this.stats.inspectMs / 1000).toFixed(1)} s; ` +
        `claims ${(this.stats.claimMs / 1000).toFixed(1)} s, flushes ${(this.stats.flushMs / 1000).toFixed(1)} s.`
    )
    this.stats = Prober.freshStats()
  }
}