import db from '@adonisjs/lucid/services/db'
import { createReadStream, createWriteStream } from 'node:fs'
import { mkdir, rename, stat } from 'node:fs/promises'
import { join } from 'node:path'
import type { Readable } from 'node:stream'
import { pipeline } from 'node:stream/promises'
import coordinatorConfig from '#config/coordinator'
import RuleSets from '#services/rule_sets'
import CatalogSpace from '#services/catalog_space'
import { LeaseLost } from '#services/leases'
import type { GroupLease, Lane, NextRange } from '#lib/edge/index'

export const LANES: Lane[] = ['history', 'tail']

/**
 * Group leases (REBUILD.md, "Voyager 3: window indexing", 1 and 5): a worker thread holds a group and asks for its
 * next range each time it finished one, so a group's ranges are indexed in article order through one carried
 * window. The coordinator keeps one row a group and lane (group_leases): the holder, its heartbeat, the range it
 * works and the snapshot its window stands at.
 *
 * - Two lanes (2026-10-27): the `history` lane works the ranges the bounds hand out (LEASE_MIN/MAX_PRIORITY,
 *   LEASE_CAPTURE_ONLY: the windows, the history), the `tail` lane the tail planner's ranges (priority at or below
 *   the tail's). Each lane has its own holder, window and snapshot, so a thread on the tail never wades through a
 *   group's history first (a group's ranges come in article order, and a held group's history is days of ranges), and
 *   the tail of a group whose history is held isn't stuck behind it. The one seam a group is where the history lane
 *   reaches the tail lane's first range (its ranges are done: the history lane is caught up).
 * - lease: the free group with the most urgent planned range of the lane (the ranges' priority), and among those a
 *   mix: the group with the most planned articles and the longest unheld one (never held first) take turns, so a
 *   big group snapshotted at a stop doesn't wait behind hundreds of small never-held ones, and a quiet group's tail
 *   is visited all the same. A holder quiet for LEASE_TTL_SECONDS is taken over. Taking a group resets what the
 *   lane's snapshot doesn't cover: the lane's ranges from the snapshot's article on that the window did after it
 *   (leased, done since the snapshot, or read past it) are planned again (the batch store's keys make the redo
 *   idempotent); without a snapshot the window is lost and the group continues at its resume point (one seam).
 * - next: the group's lowest planned range of the lane (or the one this node already has), leased to the node; none
 *   when the lane is caught up: the worker snapshots and gives the group back, and gets it again when the tail
 *   planned more.
 * - snapshot: the serialized window, kept as a file on bulk (SNAPSHOT_DIR; `<group>.gz`, the tail's
 *   `<group>-tail.gz`), given to the next holder of the lane.
 */
class GroupLeases {
  get dir(): string {
    return coordinatorConfig.snapshotDir
  }

  /**
   * The lane's ranges, as a condition on `r`: the tail lane's are at or below the tail's priority, the history lane's
   * above it — within the lease bounds when `bounds` (what is handed out), or all of them (what a lane's window did).
   */
  private laneFilter(lane: Lane, bounds: boolean): { sql: string; bindings: number[] } {
    const tail = coordinatorConfig.tail.priority
    if (lane === 'tail') return { sql: 'r.priority <= ?', bindings: [tail] }
    if (!bounds) return { sql: 'r.priority > ?', bindings: [tail] }
    const { leasePriorityMin: min, leasePriorityMax: max, leaseCaptureOnly } = coordinatorConfig
    return {
      sql: `r.priority > ? AND r.priority BETWEEN ? AND ?${leaseCaptureOnly ? ' AND r.capture' : ''}`,
      bindings: [tail, min, max],
    }
  }

  async lease(nodeId: number, lane: Lane = 'history'): Promise<GroupLease | null> {
    // The catalog at its cap on the NVMe: the tail only (CatalogSpace), as the range leases did.
    if (lane === 'history' && (await CatalogSpace.full())) return null
    const rules = await RuleSets.current()
    const ttl = coordinatorConfig.leaseTtlSeconds
    const todo = this.laneFilter(lane, true)
    const all = this.laneFilter(lane, false)
    return db.transaction(async (trx) => {
      // One lease at a time: the choice reads several tables.
      await trx.rawQuery(`SELECT pg_advisory_xact_lock(hashtextextended('group-lease', 0))`)
      // The free groups with planned work in the lane (a range leased by a holder gone quiet counts: a group whose
      // last range that holder took would never be taken over), by the most urgent range's priority; within a priority the
      // most planned articles and the longest unheld alternate: one lease starts from the biggest, the next from
      // the longest unheld (the sequence counts the leases; ranked afresh each time, "biggest first" would always
      // win).
      const { rows: turns } = await trx.rawQuery(`SELECT nextval('group_lease_turns') % 2 AS odd`)
      const [first, second] = Number(turns[0].odd) === 1 ? [1, 0] : [0, 1]
      const { rows } = await trx.rawQuery(
        `WITH todo AS (
           SELECT r.group_id, min(r.priority) AS priority, sum(r.end_id - r.next_id + 1) AS articles
           FROM ranges r
           WHERE (r.status = 'planned' OR (r.status = 'leased' AND r.heartbeat_at < now() - make_interval(secs => ?)))
             AND r.next_id <= r.end_id AND ${todo.sql}
           GROUP BY r.group_id
         ), free AS (
           SELECT g.id, g.name, g.priority AS group_priority, l.snapshot_next, l.snapshot_at, l.heartbeat_at,
                  todo.priority, todo.articles
           FROM todo JOIN groups g ON g.id = todo.group_id AND g.active
           LEFT JOIN group_leases l ON l.group_id = g.id AND l.lane = ?
           WHERE l.node_id IS NULL OR l.heartbeat_at < now() - make_interval(secs => ?)
         ), turns AS (
           SELECT free.*,
                  least(2 * row_number() OVER (PARTITION BY priority ORDER BY articles DESC, group_priority, id) - ?,
                        2 * row_number() OVER (PARTITION BY priority ORDER BY heartbeat_at NULLS FIRST, group_priority, id) - ?) AS turn
           FROM free
         )
         SELECT * FROM turns ORDER BY priority, turn, group_priority, id LIMIT 1`,
        [ttl, ...todo.bindings, lane, ttl, first, second]
      )
      const row = rows[0]
      if (!row) return null
      const groupId = Number(row.id)
      // What the snapshot doesn't cover is read again: the lane's ranges from its article on that the window did
      // after it (a range done before the snapshot was taken — by the old range leases, say — stays done).
      if (row.snapshot_next !== null) {
        await trx.rawQuery(
          `UPDATE ranges r SET status = 'planned', node_id = NULL, next_id = greatest(r.start_id, ?::bigint)
           WHERE r.group_id = ? AND r.end_id >= ?::bigint AND ${all.sql}
             AND (r.status = 'leased' OR (r.status = 'done' AND (r.done_at IS NULL OR r.done_at >= ?)) OR (r.status = 'planned' AND r.next_id > ?::bigint))`,
          [
            row.snapshot_next,
            groupId,
            row.snapshot_next,
            ...all.bindings,
            row.snapshot_at ?? new Date(0),
            row.snapshot_next,
          ]
        )
      } else {
        await trx.rawQuery(
          `UPDATE ranges r SET status = 'planned', node_id = NULL WHERE r.group_id = ? AND r.status = 'leased' AND ${all.sql}`,
          [groupId, ...all.bindings]
        )
      }
      await trx.rawQuery(
        `INSERT INTO group_leases (group_id, lane, node_id, leased_at, heartbeat_at, range_id)
         VALUES (?, ?, ?, now(), now(), NULL)
         ON CONFLICT (group_id, lane) DO UPDATE SET node_id = excluded.node_id, leased_at = now(), heartbeat_at = now(), range_id = NULL`,
        [groupId, lane, nodeId]
      )
      await trx.from('nodes').where('id', nodeId).update({ last_seen_at: new Date() })
      return {
        group: { id: groupId, name: row.name, priority: Number(row.group_priority) },
        lane,
        rules,
        snapshot: row.snapshot_next !== null ? { next: Number(row.snapshot_next) } : null,
      }
    })
  }

  /** The group's lease row of the lane, when this node holds it. */
  async held(groupId: number, nodeId: number, lane: Lane = 'history') {
    const lease = await db
      .from('group_leases')
      .where('group_id', groupId)
      .where('lane', lane)
      .first()
    if (!lease || Number(lease.node_id) !== nodeId)
      throw new LeaseLost(`Group ${groupId} (${lane}) is not held by this node`)
    return lease
  }

  /**
   * The group's next range of the lane in article order (or the one this node has), leased to the node; null when
   * the lane is caught up.
   */
  async next(groupId: number, nodeId: number, lane: Lane = 'history'): Promise<NextRange | null> {
    await this.held(groupId, nodeId, lane)
    const rules = await RuleSets.current()
    const todo = this.laneFilter(lane, true)
    const all = this.laneFilter(lane, false)
    const mine = (
      await db.rawQuery(
        `SELECT r.id, r.start_id, r.end_id, r.next_id FROM ranges r
         WHERE r.group_id = ? AND r.status = 'leased' AND r.node_id = ? AND ${all.sql} ORDER BY r.start_id LIMIT 1`,
        [groupId, nodeId, ...all.bindings]
      )
    ).rows[0]
    const row =
      mine ??
      (
        await db.rawQuery(
          `UPDATE ranges r SET status = 'leased', node_id = ?, leased_at = now(), heartbeat_at = now(), attempts = attempts + 1, rule_version = ?
           WHERE r.id = (
             SELECT r.id FROM ranges r WHERE r.group_id = ? AND r.status = 'planned' AND r.next_id <= r.end_id AND ${todo.sql}
             ORDER BY r.start_id LIMIT 1 FOR UPDATE SKIP LOCKED)
           RETURNING r.id, r.start_id, r.end_id, r.next_id`,
          [nodeId, rules.version, groupId, ...todo.bindings]
        )
      ).rows[0]
    await db
      .from('group_leases')
      .where('group_id', groupId)
      .where('lane', lane)
      .update({ range_id: row ? Number(row.id) : null, heartbeat_at: new Date() })
    if (!row) return null
    return {
      range: {
        id: Number(row.id),
        start: Number(row.start_id),
        end: Number(row.end_id),
        next: Number(row.next_id),
      },
    }
  }

  async heartbeat(groupId: number, nodeId: number, lane: Lane = 'history') {
    await this.held(groupId, nodeId, lane)
    await db
      .from('group_leases')
      .where('group_id', groupId)
      .where('lane', lane)
      .update({ heartbeat_at: new Date() })
    await db.from('nodes').where('id', nodeId).update({ last_seen_at: new Date() })
  }

  /** The lane a range belongs to, by its priority. */
  laneOf(priority: number): Lane {
    return priority <= coordinatorConfig.tail.priority ? 'tail' : 'history'
  }

  /** A range of the group finished by its holder: counted on the lane's lease (Leases.finish does the range). */
  async rangeDone(groupId: number, headers: number, lane: Lane = 'history') {
    await db.rawQuery(
      `UPDATE group_leases SET ranges_done = ranges_done + 1, headers = headers + ?, range_id = NULL WHERE group_id = ? AND lane = ?`,
      [headers, groupId, lane]
    )
  }

  /** The worker gives the group back (after its snapshot): its range stays at its resume point for the next holder. */
  async release(groupId: number, nodeId: number, lane: Lane = 'history') {
    await this.held(groupId, nodeId, lane)
    const all = this.laneFilter(lane, false)
    await db.rawQuery(
      `UPDATE ranges r SET status = 'planned', node_id = NULL
       WHERE r.group_id = ? AND r.status = 'leased' AND r.node_id = ? AND ${all.sql}`,
      [groupId, nodeId, ...all.bindings]
    )
    await db
      .from('group_leases')
      .where('group_id', groupId)
      .where('lane', lane)
      .update({ node_id: null, range_id: null })
  }

  snapshotPath(groupId: number, lane: Lane = 'history') {
    return join(this.dir, lane === 'tail' ? `${groupId}-tail.gz` : `${groupId}.gz`)
  }

  /**
   * The window's snapshot standing at `next` (the next article to index), streamed to disk (a busy group's is
   * gigabytes) and written whole before it replaces the old one.
   */
  async putSnapshot(
    groupId: number,
    nodeId: number,
    next: number,
    body: Readable,
    lane: Lane = 'history'
  ) {
    await this.held(groupId, nodeId, lane)
    await mkdir(this.dir, { recursive: true })
    const path = this.snapshotPath(groupId, lane)
    await pipeline(body, createWriteStream(`${path}.part`))
    await rename(`${path}.part`, path)
    await db
      .from('group_leases')
      .where('group_id', groupId)
      .where('lane', lane)
      .update({ snapshot_next: next, snapshot_at: new Date(), heartbeat_at: new Date() })
  }

  /** The snapshot's stream and size, or null when there is none. */
  async getSnapshot(
    groupId: number,
    nodeId: number,
    lane: Lane = 'history'
  ): Promise<{ stream: Readable; size: number; next: number } | null> {
    const lease = await this.held(groupId, nodeId, lane)
    if (lease.snapshot_next === null) return null
    const path = this.snapshotPath(groupId, lane)
    try {
      const { size } = await stat(path)
      return { stream: createReadStream(path), size, next: Number(lease.snapshot_next) }
    } catch {
      return null
    }
  }
}

export default new GroupLeases()