usenet-indexer/commands/ranges_dates.ts

import { BaseCommand, flags } from '@adonisjs/core/ace'
import type { CommandOptions } from '@adonisjs/core/types/ace'
import db from '@adonisjs/lucid/services/db'
import coordinatorConfig from '#config/coordinator'
import NntpPool from '#services/nntp_pool'
import { NntpShare } from '#services/nntp_shares'
import { estimateDate, sampleGroup, type DateSample } from '#services/group_dates'

/**
 * When each active group's articles were posted (group_dates, sampled from the news server) and from that every
 * range's estimated posting date (ranges.est_date), so ranges can be chosen by era: ranges:capture marks a
 * posting-time window of every group for the corpus (REBUILD.md, corpus). A group gets a sample every --every
 * articles (16 … 1000): boneless (161B articles) ~320, three weeks apart; ~15k XOVER commands of 32 articles in all.
 */
export default class RangesDates extends BaseCommand {
  static commandName = 'ranges:dates'
  static description =
    "Samples when each active group's articles were posted and estimates every range's date"
  static options: CommandOptions = { startApp: true }

  @flags.string({ description: 'Only groups whose name matches this SQL LIKE pattern' })
  declare group?: string

  @flags.number({ description: 'Articles between two samples of a group (default 500M)' })
  declare every?: number

  @flags.number({
    description:
      'Connections, a share of the account beside the coordinator and the prober (default 6)',
  })
  declare connections?: number

  @flags.boolean({ description: 'Only groups without samples yet' })
  declare missing: boolean

  @flags.boolean({
    description: 'No sampling: estimate the ranges from the samples stored already',
  })
  declare estimateOnly: boolean

  async run() {
    try {
      if (!this.estimateOnly) await this.sample()
      await this.estimate()
      await this.report()
    } finally {
      await NntpPool.close()
    }
  }

  private groups(missing = this.missing) {
    const q = db.from('groups').where('active', true).select('id', 'name').orderBy('priority')
    if (this.group) q.where('name', 'like', this.group)
    if (missing)
      q.whereNotExists((sub) =>
        sub.from('group_dates').whereRaw('group_dates.group_id = groups.id')
      )
    return q
  }

  private async sample() {
    const every = this.every ?? 500_000_000
    const connections = this.connections ?? 6
    const groups = await this.groups()
    const share = new NntpShare(
      'dates',
      coordinatorConfig.nntpShares.coordinatorPriority + 1,
      1,
      connections
    )
    const first = await NntpPool.share(share)
    const sharing = setInterval(
      () =>
        NntpPool.share(share).catch((error) => this.logger.warning(`NNTP share: ${error.message}`)),
      coordinatorConfig.nntpShares.everyMs
    )
    this.logger.info(
      `Sampling ${groups.length} groups over ${first.granted} connections, a sample every ${every / 1e6}M articles.`
    )
    const startedAt = Date.now()
    let done = 0
    let commands = 0
    let samples = 0
    const queue = [...groups]
    const worker = async () => {
      for (let g = queue.shift(); g; g = queue.shift()) {
        try {
          const result = await sampleGroup((work) => NntpPool.use(work), g.name, {
            every,
            size: 32,
          })
          for (const f of result.failed) this.logger.warning(`${g.name} at ${f}`)
          await db.transaction(async (trx) => {
            await trx.from('group_dates').where('group_id', g.id).delete()
            if (result.samples.length > 0)
              await trx.table('group_dates').multiInsert(
                result.samples.map((s) => ({
                  group_id: g.id,
                  article: s.article,
                  posted_at: s.postedAt,
                }))
              )
          })
          commands += result.commands
          samples += result.samples.length
          if (result.samples.length < 2)
            this.logger.warning(
              `${g.name}: ${result.samples.length} samples over ${result.low}–${result.high}.`
            )
        } catch (error: any) {
          this.logger.warning(`${g.name}: ${error.message}`)
        }
        done++
        if (done % 50 === 0)
          this.logger.info(
            `${done} of ${groups.length} groups: ${samples} samples, ${commands} commands, ${((Date.now() - startedAt) / 1000).toFixed(0)} s.`
          )
      }
    }
    try {
      await Promise.all(
        Array.from({ length: Math.max(1, Math.min(connections, groups.length)) }, worker)
      )
    } finally {
      clearInterval(sharing)
    }
    this.logger.info(
      `Sampled ${groups.length} groups: ${samples} samples, ${commands} commands, ${((Date.now() - startedAt) / 1000).toFixed(0)} s.`
    )
  }

  /** Every range of the groups sampled gets est_date from the group's curve (1000 a statement). */
  private async estimate() {
    const groups = await this.groups(false).whereExists((sub) =>
      sub.from('group_dates').whereRaw('group_dates.group_id = groups.id')
    )
    let updated = 0
    for (const g of groups) {
      const rows = await db
        .from('group_dates')
        .where('group_id', g.id)
        .orderBy('article')
        .select('article', 'posted_at')
      const samples: DateSample[] = rows.map((r) => ({
        article: Number(r.article),
        postedAt: new Date(r.posted_at),
      }))
      const ranges = await db
        .from('ranges')
        .where('group_id', g.id)
        .orderBy('start_id')
        .select('id', 'start_id')
      for (let i = 0; i < ranges.length; i += 1000) {
        const batch = ranges.slice(i, i + 1000)
        const values = batch.map(() => '(?::bigint, ?::timestamptz)').join(', ')
        const bindings = batch.flatMap((r) => [
          Number(r.id),
          estimateDate(samples, Number(r.start_id))!.toISOString(),
        ])
        await db.rawQuery(
          `UPDATE ranges AS r SET est_date = v.d FROM (VALUES ${values}) AS v(id, d) WHERE r.id = v.id`,
          bindings
        )
        updated += batch.length
      }
    }
    this.logger.info(`${updated} ranges of ${groups.length} groups dated.`)
  }

  /** The planned history by posting year: what a window would take. */
  private async report() {
    const { rows } = await db.rawQuery(`
      SELECT extract(year FROM est_date)::int AS year, count(*) AS ranges, count(DISTINCT group_id) AS groups,
        round(sum(end_id - start_id + 1) / 1e9, 1) AS articles,
        round(sum(end_id - start_id + 1) FILTER (WHERE status = 'done') / 1e9, 1) AS done,
        round(sum(end_id - start_id + 1) FILTER (WHERE capture) / 1e9, 1) AS capture
      FROM ranges WHERE est_date IS NOT NULL GROUP BY 1 ORDER BY 1`)
    const undatedRows = await db.rawQuery(
      `SELECT count(*) AS undated FROM ranges WHERE est_date IS NULL`
    )
    const undated = undatedRows.rows[0].undated
    this.logger.info(
      `By estimated posting year (B articles; done: indexed, capture: marked for the corpus); ${undated} ranges undated:`
    )
    for (const r of rows)
      this.logger.info(
        `  ${r.year}: ${String(r.ranges).padStart(7)} ranges of ${String(r.groups).padStart(3)} groups, ${String(r.articles).padStart(6)}B articles, ${r.done}B done, ${r.capture}B capture`
      )
  }
}