import { createHash } from 'node:crypto'
import db from '@adonisjs/lucid/services/db'
import indexerConfig from '#config/indexer'
import NntpService from '#services/NntpService'
import QueueService from '#services/QueueService'
import { type OverviewHeader } from '#services/IngestService'
import { decodeSegment, inferTotalParts, parseYencMeta } from '#lib/yenc/index'
import { sniffContentType } from '#lib/content/index'
import { parseRarHeaders } from '#lib/rar/index'

/** Expanding re-reads the file's article range with XOVER; beyond this it is too costly. */
const EXPAND_MAX_RANGE = 200_000
/** A claimed row whose worker died is taken again after this long. */
const RECLAIM_AFTER = '1 hour'

export interface DeferredArticle {
  articleNumber: number
  header: OverviewHeader
}

/** File name from a deferred uploader's subject: `[i/M] "name" yEnc`, no part counter. */
export function deferredName(subject: string | undefined): string | null {
  return /"([^"]+)"/.exec(subject ?? '')?.[1]?.trim() || null
}

export function isDeferred(header: OverviewHeader): boolean {
  const id = header['message-id'] ?? ''
  return indexerConfig.deferBodyMessageIds.some((re) => re.test(id))
}

/** A recorded range never spans more articles than this (one XOVER to expand it). */
const RANGE_MAX_SPAN = 50_000

export function bodyJob(header: OverviewHeader, group: string, articleNumber: number | bigint) {
  return {
    name: 'process-body',
    data: { header, group, articleNumber: Number(articleNumber) },
    // By message-id: an article crossposted to several indexed groups is downloaded once.
    opts: { jobId: `body-${createHash('sha1').update(header['message-id'] ?? '').digest('hex')}` },
  }
}

/**
 * Deferred uploads (CONCEPT.md D28). Their articles skip the body queue and
 * are recorded here per file, from the headers alone. `worker:deferred`
 * later downloads one body per file for its size, part count and content
 * type, and sends small files (NZBs in RARs, par2 index files) to the body
 * queue completely.
 */
class DeferredService {
  /**
   * Records one group's deferred articles: per file when the subject names
   * the file, otherwise only as article ranges. Returns the number of files.
   */
  async record(groupName: string, articles: DeferredArticle[]): Promise<number> {
    const named = articles.filter((a) => deferredName(a.header.subject) !== null)
    await this.recordRanges(groupName, articles.filter((a) => deferredName(a.header.subject) === null))
    return this.recordFiles(groupName, named)
  }

  private async recordRanges(groupName: string, articles: DeferredArticle[]) {
    if (articles.length === 0) return
    const sorted = [...articles].sort((a, b) => Number(a.articleNumber) - Number(b.articleNumber))
    const now = new Date()
    const rows: Record<string, unknown>[] = []
    for (const article of sorted) {
      const n = Number(article.articleNumber)
      const bytes = Number.parseInt(article.header[':bytes'] ?? '0', 10) || 0
      const last = rows.at(-1)
      if (last && n - (last.first_article as number) <= RANGE_MAX_SPAN) {
        last.last_article = n
        ;(last.article_count as number)++
        last.bytes = (last.bytes as number) + bytes
      } else {
        rows.push({ group_name: groupName, first_article: n, last_article: n, article_count: 1, bytes, created_at: now })
      }
    }
    await db.table('deferred_ranges').multiInsert(rows)
  }

  private async recordFiles(groupName: string, articles: DeferredArticle[]): Promise<number> {
    const files = new Map<string, {
      name: string; subject: string; poster: string; first: number; firstId: string; last: number
      count: number; bytes: number; postedAt: Date | null
    }>()
    for (const { articleNumber, header } of articles) {
      const name = deferredName(header.subject)!
      const n = Number(articleNumber)
      const bytes = Number.parseInt(header[':bytes'] ?? '0', 10) || 0
      const date = header.date ? new Date(header.date) : null
      const postedAt = date && !Number.isNaN(date.getTime()) ? date : null
      const file = files.get(name)
      if (!file) {
        files.set(name, {
          name, subject: header.subject ?? '', poster: header.from ?? '', first: n,
          firstId: (header['message-id'] ?? '').replace(/^<|>$/g, ''), last: n, count: 1, bytes, postedAt,
        })
        continue
      }
      if (n < file.first) {
        file.first = n
        file.firstId = (header['message-id'] ?? '').replace(/^<|>$/g, '')
      }
      file.last = Math.max(file.last, n)
      file.count++
      file.bytes += bytes
      if (postedAt && (!file.postedAt || postedAt < file.postedAt)) file.postedAt = postedAt
    }
    if (files.size === 0) return 0

    const now = new Date()
    const rows = [...files.values()].map((f) => ({
      file_key: createHash('sha1').update(`deferred\n${f.name}`).digest('hex'),
      name: f.name,
      subject: f.subject,
      poster: f.poster,
      group_name: groupName,
      first_article: f.first,
      first_message_id: f.firstId,
      last_article: f.last,
      article_count: f.count,
      bytes: f.bytes,
      posted_at: f.postedAt,
      first_seen_at: now,
      last_seen_at: now,
    }))
    // Counts and the article range only add up within the group the file was first seen in.
    await db.rawQuery(
      `INSERT INTO deferred_files AS d (${Object.keys(rows[0]).join(', ')})
       SELECT * FROM jsonb_to_recordset(?::jsonb) AS r(
         file_key varchar, name text, subject text, poster text, group_name text, first_article bigint,
         first_message_id text, last_article bigint, article_count int, bytes bigint, posted_at timestamptz,
         first_seen_at timestamptz, last_seen_at timestamptz)
       ON CONFLICT (file_key) DO UPDATE SET
         last_seen_at = EXCLUDED.last_seen_at,
         first_message_id = CASE WHEN d.group_name = EXCLUDED.group_name AND EXCLUDED.first_article < d.first_article
           THEN EXCLUDED.first_message_id ELSE d.first_message_id END,
         first_article = CASE WHEN d.group_name = EXCLUDED.group_name THEN LEAST(d.first_article, EXCLUDED.first_article) ELSE d.first_article END,
         last_article = CASE WHEN d.group_name = EXCLUDED.group_name THEN GREATEST(d.last_article, EXCLUDED.last_article) ELSE d.last_article END,
         article_count = d.article_count + CASE WHEN d.group_name = EXCLUDED.group_name THEN EXCLUDED.article_count ELSE 0 END,
         bytes = d.bytes + CASE WHEN d.group_name = EXCLUDED.group_name THEN EXCLUDED.bytes ELSE 0 END,
         posted_at = LEAST(d.posted_at, EXCLUDED.posted_at)`,
      [JSON.stringify(rows)]
    )
    return files.size
  }

  /** Takes up to `limit` files that stopped growing; they are marked `processing`. */
  async claim(limit: number): Promise<any[]> {
    const settled = new Date(Date.now() - indexerConfig.deferredSettleMs)
    const { rows } = await db.rawQuery(
      `UPDATE deferred_files SET status = 'processing', processed_at = now()
       WHERE id IN (
         SELECT id FROM deferred_files
         WHERE (status = 'pending' AND last_seen_at < ?)
            OR (status = 'processing' AND processed_at < now() - interval '${RECLAIM_AFTER}')
         ORDER BY id LIMIT ? FOR UPDATE SKIP LOCKED)
       RETURNING *`,
      [settled, limit]
    )
    return rows
  }

  /**
   * Queues the bodies of up to `limit` recorded ranges, oldest first: the
   * articles are found again with XOVER. Returns ranges and articles queued.
   */
  async expandRanges(limit: number): Promise<{ ranges: number; articles: number }> {
    const ranges = await db.from('deferred_ranges').whereNull('expanded_at').orderBy('id').limit(limit)
    let articles = 0
    for (const range of ranges) {
      const conn = await NntpService.acquire()
      let failed = false
      let overviews: [number, OverviewHeader][]
      try {
        await conn.group(range.group_name)
        overviews = (await conn.xover(String(range.first_article), String(range.last_article))).overviews
      } catch (error) {
        failed = true
        throw error
      } finally {
        NntpService.release(conn, failed)
      }
      const jobs = overviews
        .filter(([, header]) => header && isDeferred(header) && deferredName(header.subject) === null)
        .map(([articleNumber, header]) => bodyJob(header, range.group_name, articleNumber))
      for (let i = 0; i < jobs.length; i += 5000) await QueueService.bodyQueue.addBulk(jobs.slice(i, i + 5000))
      await db.from('deferred_ranges').where('id', range.id).update({ expanded_at: new Date() })
      articles += jobs.length
    }
    return { ranges: ranges.length, articles }
  }

  /** Reads one body of the file; expands small files into the body queue. Returns the new status. */
  async process(file: any): Promise<string> {
    const update: Record<string, unknown> = { processed_at: new Date(), error: null }
    try {
      const conn = await NntpService.acquire()
      let failed = false
      let body: Buffer
      try {
        body = await NntpService.fetchBody(conn, file.first_message_id)
      } catch (error) {
        failed = true
        throw error
      } finally {
        NntpService.release(conn, failed)
      }

      const meta = parseYencMeta(body)
      const totalParts = inferTotalParts(meta)
      Object.assign(update, {
        yenc_name: meta.begin.name,
        size: meta.begin.size,
        total_parts: totalParts,
        sampled_part: meta.begin.part ?? 1,
      })
      if ((meta.begin.part ?? 1) === 1) {
        const data = decodeSegment(body, { lenientSize: true, verifyCrc: false }).data
        update.content_type = sniffContentType(data)
        const rar = parseRarHeaders(data)
        if (rar) update.encrypted = rar.headersEncrypted || rar.entries.some((e) => e.encrypted)
      }

      update.status = 'sampled'
      if (totalParts <= indexerConfig.deferredExpandMaxParts && update.encrypted !== true) {
        const queued = await this.expand(file)
        if (queued > 0) update.status = 'expanded'
      }
    } catch (error: any) {
      update.status = 'failed'
      update.error = String(error.message).slice(0, 500)
    }
    await db.from('deferred_files').where('id', file.id).update(update)
    return update.status as string
  }

  /** Finds the file's articles again (XOVER over its range) and queues their bodies. */
  private async expand(file: any): Promise<number> {
    const first = Number(file.first_article)
    const last = Number(file.last_article)
    if (last - first > EXPAND_MAX_RANGE) return 0

    const conn = await NntpService.acquire()
    let failed = false
    let overviews: [number, OverviewHeader][]
    try {
      await conn.group(file.group_name)
      overviews = (await conn.xover(String(first), String(last))).overviews
    } catch (error) {
      failed = true
      throw error
    } finally {
      NntpService.release(conn, failed)
    }
    const jobs = overviews
      .filter(([, header]) => header && deferredName(header.subject) === file.name)
      .map(([articleNumber, header]) => bodyJob(header, file.group_name, articleNumber))
    if (jobs.length > 0) await QueueService.bodyQueue.addBulk(jobs)
    return jobs.length
  }
}

export default new DeferredService()