usenet-indexer/app/services/SpotService.ts

import { createHash } from 'node:crypto'
import db from '@adonisjs/lucid/services/db'
import logger from '@adonisjs/core/services/logger'
import indexerConfig from '#config/indexer'
import NntpService from '#services/NntpService'
import NzbImportService from '#services/NzbImportService'
import IngestService, { type OverviewHeader } from '#services/IngestService'
import { parseNzb } from '#lib/nzb/index'
import {
  moderationTarget,
  passwordFromSpot,
  spotterId,
  decodeSpotBinary,
  decodeSpotText,
  parseSpotArticleHeaders,
  parseSpotHeader,
  parseSpotXml,
  verifySpotHeader,
  type SpotHeader,
  type SpotXml,
} from '#lib/spotnet/index'

/** Spots whose NZB is larger than this are skipped (Spotweb skips >50 KB of XML too). */
const MAX_NZB_SEGMENTS = 20
/** A personal dispose only counts within this long after the spot (Spotweb: 5 days). */
const DISPOSE_WINDOW_SECONDS = 5 * 24 * 60 * 60
/** Rows per INSERT: 14 columns each, under Postgres' 65,535 bind parameters. */
const INSERT_CHUNK = 2000
/** Spot images larger than this many articles aren't fetched. */
const MAX_IMAGE_SEGMENTS = 10
/** A claimed spot whose worker died is taken again after this long. */
const RECLAIM_AFTER = '1 hour'

/**
 * A spot's title: from its XML, which is decoded properly (the Subject of a
 * raw header is read as ISO-8859-1), else from the header.
 */
function spotTitle(head: { header: SpotHeader; xml: SpotXml | null }): string {
  return head.xml?.title || head.header.title
}

/**
 * A placeholder NZB (e.g. Dummy.rar + Dummy.nzb, 2.6 KB, shared by many
 * spots): less than 1% of the size the spot announces (the real NZB is
 * handed out elsewhere), or under 16 KB in all (text-only spots: requests,
 * announcements). Its files are not imported, and the spot is hidden.
 */
export function isDummyNzb(nzb: { files: { segments: { bytes: number }[] }[] }, spotSize: number): boolean {
  const bytes = nzb.files.reduce((sum, f) => sum + f.segments.reduce((s, x) => s + x.bytes, 0), 0)
  return bytes < 16 * 1024 || (spotSize > 0 && bytes < spotSize / 100)
}

export function isSpotGroup(name: string): boolean {
  return indexerConfig.spotGroups.includes(name)
}

/**
 * Spotnet (CONCEPT.md D29). Spots are signed posts in free.pt; their full
 * header holds XML with the message ids of a compressed NZB, which is
 * imported like a posted NZB (D26). Spot NZBs name many obfuscated posts
 * (nyuu, ngPost, the skipped multi-group uploader).
 */
class SpotService {
  /** Queues the spots among a batch of one spot group's headers. Returns how many. */
  async recordHeaders(groupName: string, overviews: [number, OverviewHeader][]): Promise<number> {
    const now = new Date()
    const rows = []
    for (const [articleNumber, header] of overviews) {
      if (!header?.from || !header['message-id']) continue
      const spot = parseSpotHeader(header.from, header.subject ?? '', header['message-id'], header.date)
      if (!spot?.title) continue
      rows.push({
        message_id: spot.messageId,
        group_name: groupName,
        article_number: Number(articleNumber),
        title: spot.title,
        tag: spot.tag || null,
        poster: spot.poster,
        category: spot.category,
        subcats: JSON.stringify(spot.subcats),
        size: spot.size || null,
        posted_at: spot.stamp ? new Date(spot.stamp * 1000) : null,
        key_id: spot.keyId,
        spotter_id: spotterId(spot.selfSignedPubKey) || null,
        // From the overview; titles with accents may fail here (UTF-8 decoding)
        // and are verified again on their raw header by worker:spots.
        verified: verifySpotHeader(spot),
        created_at: now,
      })
    }
    for (let i = 0; i < rows.length; i += INSERT_CHUNK) {
      await db.knexQuery().table('spots').insert(rows.slice(i, i + INSERT_CHUNK)).onConflict('message_id').ignore()
    }
    return rows.length
  }

  /**
   * Takes up to `limit` pending spots, marked `processing`: recent ones
   * (to be imported) first, then the history in posting order, so live spots
   * don't wait behind a backfill and moderation follows its targets.
   */
  async claim(limit: number): Promise<any[]> {
    const claim = async (filter: string, bindings: unknown[], n: number) =>
      (
        await db.rawQuery(
          `UPDATE spots SET status = 'processing', processed_at = now()
           WHERE message_id IN (
             SELECT message_id FROM spots WHERE ${filter}
             ORDER BY article_number LIMIT ? FOR UPDATE SKIP LOCKED)
           RETURNING *`,
          [...bindings, n]
        )
      ).rows as any[]
    const rows = await claim(`status = 'pending' AND posted_at >= ?`, [this.importCutoff()], limit)
    if (rows.length < limit) rows.push(...(await claim(`status = 'pending'`, [], limit - rows.length)))
    if (rows.length < limit) {
      rows.push(...(await claim(`status = 'processing' AND processed_at < now() - interval '${RECLAIM_AFTER}'`, [], limit - rows.length)))
    }
    return rows
  }

  /** Spots posted before this are listed, not imported: their NZB is read on first download. */
  importCutoff(): Date {
    return new Date(Date.now() - indexerConfig.spotImportDays * 24 * 60 * 60 * 1000)
  }

  /** Old, verified spots that aren't moderation: listed in bulk, without reading anything more. */
  isListable(spot: any): boolean {
    return (
      spot.verified === true &&
      spot.posted_at !== null &&
      new Date(spot.posted_at) < this.importCutoff() &&
      !moderationTarget({ keyId: spot.key_id, title: spot.title })
    )
  }

  /**
   * Creates the listing releases (source `spot`, no files) for these spots.
   * The release key is the one an import of the spot's NZB uses, so the
   * import on first download fills this release.
   */
  async list(spots: any[]): Promise<number> {
    if (spots.length === 0) return 0
    const keyOf = (spot: any) => createHash('sha1').update(`nzb\n${spot.message_id}`).digest('hex')
    const ids = await IngestService.upsertReleases(
      db.connection().getWriteClient(),
      spots.map((spot) => ({
        release: { key: keyOf(spot), name: spot.title, source: 'spot' as const },
        poster: spot.poster,
        groups: [spot.group_name],
        postedAt: spot.posted_at ? new Date(spot.posted_at) : undefined,
      }))
    )
    const rows = spots.map((spot) => ({ message_id: spot.message_id, release_id: ids.get(keyOf(spot)), size: spot.size }))
    await db.rawQuery(
      `UPDATE releases r SET size = v.size FROM jsonb_to_recordset(?::jsonb) AS v(release_id bigint, size bigint)
       WHERE r.id = v.release_id AND r.file_count = 0 AND v.size IS NOT NULL`,
      [JSON.stringify(rows)]
    )
    await db.rawQuery(
      `UPDATE spots s SET status = 'listed', release_id = v.release_id, processed_at = now()
       FROM jsonb_to_recordset(?::jsonb) AS v(message_id text, release_id bigint)
       WHERE s.message_id = v.message_id`,
      [JSON.stringify(rows)]
    )
    return rows.length
  }

  /**
   * Verifies the spot on its raw header (the signature covers ISO-8859-1
   * bytes, which the overview, read as UTF-8, may have mangled), then
   * applies moderation, lists an old spot or imports a recent one's NZB.
   * Returns the new status.
   */
  async process(spot: any): Promise<string> {
    const update: Record<string, unknown> = { processed_at: new Date(), error: null }
    try {
      const head = await this.readHead(spot.message_id)
      if (!head) {
        update.status = 'unverified'
      } else if (moderationTarget(head.header)) {
        update.status = 'moderation'
        await this.moderate(head.header, moderationTarget(head.header)!)
      } else {
        update.title = spotTitle(head)
        update.poster = head.header.poster
        update.spotter_id = spotterId(head.header.selfSignedPubKey) || null
        Object.assign(update, this.details(head.xml))
        if (new Date(head.header.stamp * 1000) < this.importCutoff()) {
          await this.list([{ ...spot, title: spotTitle(head), poster: head.header.poster }])
          update.status = 'listed'
        } else {
          Object.assign(update, await this.importNzb(spot.message_id, head))
        }
      }
    } catch (error: any) {
      update.status = 'failed'
      update.error = String(error.message).slice(0, 500)
      logger.warn(`Spot ${spot.message_id} "${spot.title}": ${error.message}`)
    }
    await db.from('spots').where('message_id', spot.message_id).update(update)
    if (update.status === 'imported' || update.status === 'listed') await this.applyPassword(spot.message_id)
    return update.status as string
  }

  /**
   * The NZB of a listing, on first download: reads and imports it. Returns
   * the release that now holds its files (the listing's own, unless the NZB
   * joined files already known elsewhere), or null.
   */
  async importListing(releaseId: number): Promise<number | null> {
    const spot = await db.from('spots').where('release_id', releaseId).where('status', 'listed').first()
    if (!spot) return null
    return this.reimport(spot)
  }

  /**
   * Reads a spot's NZB again and imports it, so its files end up in the
   * spot's own release under the spot's title. Returns that release, or null.
   */
  async reimport(spot: any): Promise<number | null> {
    const head = await this.readHead(spot.message_id)
    if (!head) return null
    const result = await this.importNzb(spot.message_id, head)
    await db
      .from('spots')
      .where('message_id', spot.message_id)
      .update({ ...result, ...this.details(head.xml), processed_at: new Date() })
    await this.applyPassword(spot.message_id)
    return (result.release_id as number | undefined) ?? null
  }

  /**
   * A password the spotter wrote in the spot's title or description (not in
   * the NZB) goes on its release, unless the release has one already.
   */
  async applyPassword(messageId: string): Promise<string | null> {
    const spot = await db.from('spots').where('message_id', messageId).whereNotNull('release_id').first()
    const password = spot ? passwordFromSpot(spot.title, spot.description) : null
    if (!password) return null
    await db
      .from('releases')
      .where('id', spot.release_id)
      .whereNull('password')
      .update({ password, password_source: 'spot' })
    return password
  }

  /** Columns for what the spot's XML adds to its overview: description, image, website. */
  private details(xml: SpotXml | null): Record<string, unknown> {
    return {
      description: xml?.description || null,
      image: xml?.image ? JSON.stringify(xml.image) : null,
      ...(xml?.website ? { website: xml.website } : {}),
      details_at: new Date(),
    }
  }

  /**
   * A spot row with its description and image, reading its full header
   * first when that hasn't happened yet (old spots are listed from their
   * overview only). Returns the row as it is when Usenet can't be reached.
   */
  async withDetails(spot: any): Promise<any> {
    if (spot.details_at) return spot
    try {
      const head = await this.readHead(spot.message_id)
      const update = head ? this.details(head.xml) : { details_at: new Date() }
      await db.from('spots').where('message_id', spot.message_id).update(update)
      await this.applyPassword(spot.message_id)
      return { ...spot, ...update, image: head?.xml?.image ?? null }
    } catch (error: any) {
      logger.warn(`Spot ${spot.message_id} details: ${error.message}`)
      // The article is gone (430): nothing to read later either.
      if (/^4\d\d/.test(String(error.message))) {
        await db.from('spots').where('message_id', spot.message_id).update({ details_at: new Date() })
      }
      return spot
    }
  }

  /**
   * Reads the details of up to `limit` imported spots that don't have them
   * (imported before details were kept), newest first. Returns how many.
   * Listed spots are left to their first view: there are millions.
   * Returns how many were read (0 when Usenet can't be reached).
   */
  async fillDetails(limit: number): Promise<number> {
    const spots = await db
      .from('spots')
      .where('status', 'imported')
      .whereNull('details_at')
      .orderBy('posted_at', 'desc')
      .limit(limit)
    let read = 0
    for (const spot of spots) if ((await this.withDetails(spot)).details_at) read++
    return read
  }

  /**
   * Takes up to `limit` listed spots whose article and NZB haven't been read,
   * newest first, marking them so another reader skips them.
   */
  async claimListings(limit: number): Promise<any[]> {
    const { rows } = await db.rawQuery(
      `UPDATE spots SET nzb_read_at = now()
       WHERE message_id IN (
         SELECT message_id FROM spots WHERE status = 'listed' AND nzb_read_at IS NULL
         ORDER BY posted_at DESC LIMIT ? FOR UPDATE SKIP LOCKED)
       RETURNING *`,
      [limit]
    )
    return rows
  }

  /**
   * Reads a listed spot's article and NZB without importing its files
   * (LOG.md 2026-10-08 (10)): the details (description, image), the NZB's
   * own password (`password_source = 'nzb'`, over one from the spot's text),
   * else one in the spot's text; a placeholder NZB hides the spot (`dummy`).
   * Returns what it found. A connection failure gives the spot back for later.
   */
  async readListing(spot: any): Promise<string> {
    try {
      const head = await this.readHead(spot.message_id)
      if (!head) {
        await db.from('spots').where('message_id', spot.message_id).update({ details_at: new Date() })
        return 'unverified'
      }
      const update: Record<string, unknown> = this.details(head.xml)
      let outcome = 'nothing'
      let nzbXml: string | null = null
      try {
        nzbXml = await this.readNzb(head)
      } catch (error: any) {
        if (!/^4\d\d/.test(String(error.message))) throw error
        // The NZB's articles are gone: the listing can't be downloaded any more.
        update.error = 'NZB articles gone'
        outcome = 'nzb-gone'
      }
      const nzb = nzbXml ? parseNzb(nzbXml) : null
      if (nzb && nzb.files.length > 0 && isDummyNzb(nzb, head.header.size)) {
        update.status = 'dummy'
        outcome = 'dummy'
      } else if (nzb?.meta.password && spot.release_id) {
        await db
          .from('releases')
          .where('id', spot.release_id)
          .where((q) => q.whereNull('password').orWhere('password_source', 'spot'))
          .update({ password: nzb.meta.password, password_source: 'nzb' })
        outcome = 'nzb-password'
      }
      await db.from('spots').where('message_id', spot.message_id).update(update)
      if (outcome !== 'dummy' && (await this.applyPassword(spot.message_id)) && outcome !== 'nzb-password') outcome = 'spot-password'
      return outcome
    } catch (error: any) {
      // The spot's own article is gone (430): nothing to read later either.
      if (/^4\d\d/.test(String(error.message))) {
        await db.from('spots').where('message_id', spot.message_id).update({ details_at: new Date() })
        return 'gone'
      }
      await db.from('spots').where('message_id', spot.message_id).update({ nzb_read_at: null })
      logger.warn(`Spot ${spot.message_id} listing: ${error.message}`)
      return 'retry'
    }
  }

  /** Reads up to `limit` unread listings, `concurrency` at a time. Returns the outcomes counted. */
  async readListings(limit: number, concurrency: number): Promise<Record<string, number>> {
    const spots = await this.claimListings(limit)
    const counts: Record<string, number> = {}
    let next = 0
    await Promise.all(
      Array.from({ length: Math.max(1, concurrency) }, async () => {
        while (next < spots.length) {
          const outcome = await this.readListing(spots[next++])
          counts[outcome] = (counts[outcome] ?? 0) + 1
        }
      })
    )
    return counts
  }

  /** The spot's image as posted on Usenet (Spotnet posts them uncompressed), or null. */
  async image(spot: any): Promise<Buffer | null> {
    const segments: string[] = spot.image?.segments ?? []
    if (segments.length === 0 || segments.length > MAX_IMAGE_SEGMENTS) return null
    const bodies = []
    for (const id of segments) bodies.push(await this.fetch(`BODY <${id.replace(/^<|>$/g, '')}>`))
    return decodeSpotBinary(bodies, false)
  }

  /** The spot's raw header, verified; null when it doesn't verify. */
  private async readHead(messageId: string): Promise<{ header: SpotHeader; xml: SpotXml | null } | null> {
    const raw = await this.fetch(`HEAD <${messageId}>`)
    const { headers, xml } = parseSpotArticleHeaders(raw.toString('latin1'))
    const header = parseSpotHeader(headers.from ?? '', headers.subject ?? '', messageId, headers.date)
    if (!header || !verifySpotHeader(header)) return null
    return { header, xml: parseSpotXml(decodeSpotText(xml)) }
  }

  /** Downloads and decodes the spot's compressed NZB; null when it has none (or a too large one). */
  private async readNzb(head: { xml: SpotXml | null }): Promise<string | null> {
    const segments = head.xml?.nzbSegments ?? []
    if (segments.length === 0 || segments.length > MAX_NZB_SEGMENTS) return null
    const bodies = []
    for (const id of segments) bodies.push(await this.fetch(`BODY <${id.replace(/^<|>$/g, '')}>`))
    return decodeSpotBinary(bodies).toString('utf8')
  }

  /** Downloads the spot's compressed NZB and imports it as a release (D26). */
  private async importNzb(messageId: string, head: { header: SpotHeader; xml: SpotXml | null }): Promise<Record<string, unknown>> {
    const nzbXml = await this.readNzb(head)
    if (!nzbXml) return { status: 'no-nzb' }
    const nzb = parseNzb(nzbXml)
    if (!nzb || nzb.files.length === 0) return { status: 'no-nzb' }
    const { header } = head
    if (isDummyNzb(nzb, header.size)) return { status: 'dummy' }
    const title = spotTitle(head)
    const result = await NzbImportService.import(
      {
        id: null,
        filename: `${title}.nzb`,
        poster: header.poster,
        groups: [...new Set(nzb.files.flatMap((f) => f.groups))],
        date: header.stamp * 1000,
        first_message_id: messageId,
      },
      nzbXml,
      { name: title, source: 'spot', moveContainer: false, trusted: true }
    )
    return result ? { status: 'imported', release_id: result.releaseId } : { status: 'no-nzb' }
  }

  /**
   * Applies a delete/dispose/remove: the target spot is marked `removed`
   * (its release is then hidden from search). As in Spotweb, a personal
   * dispose must come from the same spotter within 5 days; a moderator's
   * (signed with Spotnet's key, no personal key) always counts.
   */
  private async moderate(header: SpotHeader, targetId: string) {
    const target = await db.from('spots').where('message_id', targetId).first()
    if (!target) return
    const personal = header.selfSignedPubKey.length > 50 ? spotterId(header.selfSignedPubKey) : ''
    if (personal) {
      const age = header.stamp - (target.posted_at ? new Date(target.posted_at).getTime() / 1000 : 0)
      if (target.spotter_id !== personal || age > DISPOSE_WINDOW_SECONDS) return
    }
    await db.from('spots').where('message_id', targetId).update({ status: 'removed' })
    logger.info(`Spot "${target.title}" removed by ${personal ? 'its poster' : 'a moderator'}.`)
    // Spots that joined its release (a correction, a repost) take the files over now.
    if (target.release_id) {
      const sharing = await db
        .from('spots')
        .where('release_id', target.release_id)
        .where('status', 'imported')
        .whereNot('message_id', targetId)
      for (const spot of sharing) await this.reimport(spot)
    }
  }

  /** One HEAD or BODY, as raw bytes. */
  private async fetch(command: string): Promise<Buffer> {
    const conn = await NntpService.acquire()
    let failed = false
    try {
      const { data } = await NntpService.timed<any>(conn._artcmd(command), command)
      return Buffer.from(data.buffer, data.byteOffset, data.byteLength)
    } catch (error: any) {
      // 430 (no such article) is an answer, not a broken connection.
      failed = !/^4\d\d/.test(String(error.message))
      throw error
    } finally {
      NntpService.release(conn, failed)
    }
  }
}

export default new SpotService()