usenet-indexer/app/controllers/node_controller.ts

import type { HttpContext } from '@adonisjs/core/http'
import { Readable } from 'node:stream'
import { setTimeout as sleep } from 'node:timers/promises'
import db from '@adonisjs/lucid/services/db'
import Leases, { LeaseLost } from '#services/leases'
import GroupLeases from '#services/group_leases'
import BatchStore, { StaleBatch } from '#services/batch_store'
import Spool from '#services/spool'
import Corpus from '#services/corpus'
import { KEEPALIVE, encodeFrame, fromWire, type WireBatch } from '#lib/edge/index'

/**
 * How long the stream waits for a chunk the prefetcher hasn't spooled yet. Long: a chunk that keeps
 * coming in isn't given up (only a stalled one is), and some groups' chunks take 20+ minutes
 * (a.b.0day.stuffz, 17 MB in 22 min; LOG.md 2026-10-10 (3)).
 */
const CHUNK_WAIT_MS = 60 * 60 * 1000
/** While waiting, an empty frame this often: an HTTP client gives up on a quiet body (Node's after 5 minutes). */
const KEEPALIVE_MS = 20_000

/**
 * The edge workers' API (REBUILD.md, parts 1–2): a lease, the range's
 * chunks as one stream, results per batch, and the end of the range.
 */
export default class NodeController {
  private node(ctx: HttpContext) {
    return Number(ctx.auth.use('node').getUserOrFail().id)
  }

  async lease(ctx: HttpContext) {
    const lease = await Leases.lease(this.node(ctx))
    if (!lease) return ctx.response.noContent()
    return lease
  }

  // Group leases (REBUILD.md, "Voyager 3: window indexing", 1 and 5): a worker thread holds a group and takes its
  // ranges in article order; the window's snapshot goes through here too, so a worker needs no disk of its own.

  async leaseGroup(ctx: HttpContext) {
    const lease = await GroupLeases.lease(this.node(ctx))
    if (!lease) return ctx.response.noContent()
    return lease
  }

  async nextRange(ctx: HttpContext) {
    try {
      const next = await GroupLeases.next(Number(ctx.params.id), this.node(ctx))
      if (!next) return ctx.response.noContent()
      return next
    } catch (error) {
      if (error instanceof LeaseLost) return ctx.response.conflict({ error: error.message })
      throw error
    }
  }

  async groupHeartbeat(ctx: HttpContext) {
    return this.lifecycle(ctx, (groupId, nodeId) => GroupLeases.heartbeat(groupId, nodeId))
  }

  async releaseGroup(ctx: HttpContext) {
    return this.lifecycle(ctx, (groupId, nodeId) => GroupLeases.release(groupId, nodeId))
  }

  /**
   * PUT /groups/:id/snapshot?next=N: the window's serialized state standing at article N, streamed to disk as sent
   * (gzip; the body parser leaves an octet-stream alone, so the request is still readable here).
   */
  async putSnapshot(ctx: HttpContext) {
    const next = Number(ctx.request.qs().next)
    if (!Number.isFinite(next) || next <= 0) return ctx.response.badRequest({ error: 'next (the snapshot\'s article) is needed' })
    const raw = ctx.request.raw()
    const body = raw !== null ? Readable.from([Buffer.from(raw, 'binary')]) : ctx.request.request
    return this.lifecycle(ctx, (groupId, nodeId) => GroupLeases.putSnapshot(groupId, nodeId, next, body))
  }

  async getSnapshot(ctx: HttpContext) {
    try {
      const snapshot = await GroupLeases.getSnapshot(Number(ctx.params.id), this.node(ctx))
      if (!snapshot) return ctx.response.notFound({ error: 'no snapshot' })
      ctx.response.header('Content-Type', 'application/octet-stream')
      ctx.response.header('Content-Length', String(snapshot.size))
      ctx.response.header('X-Snapshot-Next', String(snapshot.next))
      return ctx.response.stream(snapshot.stream)
    } catch (error) {
      if (error instanceof LeaseLost) return ctx.response.conflict({ error: error.message })
      throw error
    }
  }

  /**
   * The range's chunks from its resume point: a stored range's from the corpus (a worker needs no mount; REBUILD.md
   * design 1), else from the spool, waiting for those still being fetched.
   */
  async stream(ctx: HttpContext) {
    const nodeId = this.node(ctx)
    const rangeId = Number(ctx.params.id)
    const range = await this.leased(ctx, rangeId, nodeId)
    if (!range) return
    const spans = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }, Number(range.next_id))
    const group = range.capture ? await db.from('groups').where('id', range.group_id).select('name').first() : null
    const corpusRange = group ? { group: group.name as string, start: Number(range.start_id), end: Number(range.end_id) } : null
    async function* frames() {
      for (const span of spans) {
        if (corpusRange && (await Corpus.has(corpusRange, span))) {
          const data = await Corpus.read(corpusRange, span)
          if (data) {
            yield encodeFrame({ ...span, data })
            await Leases.heartbeat(rangeId, nodeId)
            continue
          }
        }
        const waitUntil = Date.now() + CHUNK_WAIT_MS
        let quietSince = Date.now()
        while (!(await Spool.has(rangeId, span))) {
          if (Date.now() > waitUntil) throw new Error(`Chunk ${span.from}–${span.to} of range ${rangeId} wasn't spooled in time`)
          if (Date.now() - quietSince >= KEEPALIVE_MS) {
            quietSince = Date.now()
            yield encodeFrame(KEEPALIVE)
            await Leases.heartbeat(rangeId, nodeId)
          }
          await sleep(500)
        }
        yield encodeFrame({ ...span, data: await Spool.read(rangeId, span) })
        // Streaming keeps the lease alive.
        await Leases.heartbeat(rangeId, nodeId)
      }
    }
    ctx.response.header('Content-Type', 'application/octet-stream')
    ctx.response.stream(Readable.from(frames()))
  }

  async batch(ctx: HttpContext) {
    const nodeId = this.node(ctx)
    const rangeId = Number(ctx.params.id)
    try {
      return await BatchStore.store(rangeId, nodeId, fromWire(ctx.request.body() as WireBatch))
    } catch (error) {
      if (error instanceof StaleBatch) return ctx.response.conflict({ error: error.message })
      throw error
    }
  }

  async heartbeat(ctx: HttpContext) {
    return this.lifecycle(ctx, (rangeId, nodeId) => Leases.heartbeat(rangeId, nodeId))
  }

  async finish(ctx: HttpContext) {
    return this.lifecycle(ctx, async (rangeId, nodeId) => {
      const range = await Leases.leased(rangeId, nodeId)
      await Leases.finish(rangeId, nodeId)
      await GroupLeases.rangeDone(Number(range.group_id), Number(range.headers))
    })
  }

  /** POST /ranges/:id/corrupt { from }: a chunk that doesn't inflate; the range goes back. */
  async corrupt(ctx: HttpContext) {
    const from = Number(ctx.request.input('from'))
    return this.lifecycle(ctx, (rangeId, nodeId) => Leases.corrupt(rangeId, nodeId, from))
  }

  async release(ctx: HttpContext) {
    return this.lifecycle(ctx, (rangeId, nodeId) => Leases.release(rangeId, nodeId))
  }

  private async lifecycle(ctx: HttpContext, action: (rangeId: number, nodeId: number) => Promise<void>) {
    try {
      await action(Number(ctx.params.id), this.node(ctx))
      return { ok: true }
    } catch (error) {
      if (error instanceof LeaseLost) return ctx.response.conflict({ error: error.message })
      throw error
    }
  }

  private async leased(ctx: HttpContext, rangeId: number, nodeId: number) {
    try {
      return await Leases.leased(rangeId, nodeId)
    } catch (error) {
      if (error instanceof LeaseLost) {
        ctx.response.conflict({ error: error.message })
        return null
      }
      throw error
    }
  }
}