usenet-indexer/commands/IndexScheduler.ts

import { BaseCommand } from '@adonisjs/core/ace'
import { CommandOptions } from '@adonisjs/core/types/ace'
import Group from '#models/group'
import NntpService from '#services/NntpService'
import QueueService from '#services/QueueService'
import indexerConfig from '#config/indexer'
import Shutdown from '#services/Shutdown'
import InspectionQueue from '#services/InspectionQueue'
import FetchRangeService from '#services/FetchRangeService'
import SplitService from '#services/SplitService'
import BackfillService from '#services/BackfillService'

export default class IndexScheduler extends BaseCommand {
  public static commandName = 'index:scheduler'
  public static description = 'Periodically checks for new articles and schedules them for fetching.'

  public static options: CommandOptions = {
    startApp: true,
  }

  private pool = NntpService

  public async run() {
    this.logger.info('Scheduler started. Awaiting tasks...')

    const schedule = async () => {
      this.logger.info('Checking for new headers...')
      const groups = await Group.query().where('active', true)

      if (groups.length === 0) {
        this.logger.info('No active groups to index. Add some via `node ace db:seed` or manually.')
        return
      }

      let conn
      let failed = false
      try {
        conn = await this.pool.acquire()

        for (const group of groups) {
          try {
            const { first: firstArticle, last: lastArticle } = await this.pool.selectGroup(
              conn,
              group.name
            )

            await FetchRangeService.requeue(group)

            // Postgres returns bigint columns as strings.
            const lastQueued = BigInt(group.lastQueuedId ?? 0)
            const { batchSize, maxBatchesPerTick, initialBackfill } = indexerConfig
            let startId =
              lastQueued === 0n ? lastArticle - initialBackfill + 1n : lastQueued + 1n
            if (startId < firstArticle) startId = firstArticle

            if (startId > lastArticle) {
              await FetchRangeService.advanceWatermark(group)
              this.logger.info(`No new headers for group ${group.name}.`)
              continue
            }

            let batches = 0
            let endId = startId - 1n
            while (endId < lastArticle && batches < maxBatchesPerTick) {
              const batchStart = endId + 1n
              endId = batchStart + batchSize - 1n < lastArticle ? batchStart + batchSize - 1n : lastArticle
              await FetchRangeService.queue(group, batchStart, endId)
              batches++
            }

            this.logger.info(
              `Queued ${batches} fetch job(s) for ${group.name}: articles ${startId} to ${endId}` +
                (endId < lastArticle ? ` (${lastArticle - endId} left for next tick)` : '')
            )

            group.lastQueuedId = endId
            await group.save()
            const watermark = await FetchRangeService.advanceWatermark(group)
            if (endId - watermark > batchSize * BigInt(maxBatchesPerTick) * 3n) {
              this.logger.warning(
                `${group.name}: indexed up to ${watermark}, ${endId - watermark} articles behind the queue.`
              )
            }
          } catch (err: any) {
            failed = true
            this.logger.error(`Error processing group ${group.name}: ${err.message}`)
          }
        }
      } catch (err: any) {
        failed = true
        this.logger.error(`Error in scheduler main loop: ${err.message}`)
      } finally {
        if (conn) {
          this.pool.release(conn, failed)
        }
      }
    }

    const split = async () => {
      try {
        const result = await SplitService.splitPending()
        if (result.parents > 0) {
          this.logger.info(
            `Split ${result.parents} conflicted binaries: ${result.volumes} verified volumes, ${result.completed} completed.`
          )
        }
      } catch (err: any) {
        this.logger.error(`Error splitting conflicted binaries: ${err.message}`)
      }
    }

    // The inspection backlog lives in Postgres; BullMQ gets a working set (LOG.md 2026-10-07 (8)).
    const feedInspection = async () => {
      const moved = await InspectionQueue.feed()
      if (moved > 0) this.logger.info(`Inspection: ${moved} files moved from the backlog to the queue.`)
    }

    const backfill = async () => {
      try {
        const queued = await BackfillService.tick()
        if (queued > 0) this.logger.info(`Queued ${queued} backfill range(s).`)
      } catch (err: any) {
        this.logger.error(`Error queuing backfill: ${err.message}`)
      }
    }

    // Run immediately, then wait a full interval after each run so runs never overlap. A failed
    // round (e.g. Postgres restarting) is logged and the next one tries again: an error that escaped
    // here once ended the loop while the process stayed up, and live indexing stopped for ~3 hours
    // (LOG.md 2026-10-07 (3)).
    while (!Shutdown.requested) {
      for (const step of [schedule, backfill, split, feedInspection]) {
        try {
          await step()
        } catch (err: any) {
          this.logger.error(`Scheduler round failed, retrying next round: ${err.message}`)
        }
      }
      await Shutdown.sleep(indexerConfig.intervalMs)
    }
    await this.pool.shutdown()
    await QueueService.closeAll()
  }
}