import db from '@adonisjs/lucid/services/db'
import logger from '@adonisjs/core/services/logger'
import type { QueryClientContract } from '@adonisjs/lucid/types/database'
import type { Knex } from 'knex'
import QueueService from '#services/QueueService'
import NameService from '#services/NameService'
import { parseXrefGroups } from '#lib/nntp/xref'
import { fileKey, parseSubject, weakFileKey } from '#lib/subject/index'
import { releaseIdentity, type ReleaseIdentity } from '#lib/release/index'
import { inferTotalParts, type YencMeta } from '#lib/yenc/index'

/** One segment, with the identity of the file it belongs to already resolved. */
export interface IngestSegment {
  fileKey: string
  poster: string
  /** Subject without segment counter. */
  subject: string
  name: string
  totalParts: number
  number: number
  /** Without angle brackets. */
  messageId: string
  bytes: number
  articleNumber?: bigint | number
  postedAt?: Date
  groups: string[]
  /**
   * The identity comes from the yEnc name only (obfuscated posts). Such names
   * can be shared by several equal-sized files of one post, so a second
   * article for the same part number marks the binary as conflicted.
   */
  weakIdentity?: boolean
  /** Release (download) this file belongs to, when it can be determined. */
  release?: ReleaseIdentity | null
  /** Exact file size from =ybegin (body-derived segments only). */
  size?: number
  /** yEnc CRCs (body-derived segments only), to verify split volumes. */
  pcrc32?: string
  partSize?: number
  fileCrc32?: string
  /** Part 1 only: MD5 of the first 16 KiB of the file, as par2 records it. */
  md5_16k?: string
}

export interface CompletedBinary {
  id: number
  fileKey: string
}

const BINARY_CHUNK = 1000
/** last_seen_at is only rewritten when it is older than this (it drives GC at 48 h). */
const TOUCH_MS = 10 * 60 * 1000

type Merge = 'least' | 'coalesce' | 'touch'

/**
 * Inserts rows keyed by `key` and returns their ids. Existing rows are not
 * locked unless something in them changes (`least`: an earlier value,
 * `coalesce`: a value where there was none, `touch`: a timestamp more than
 * TOUCH_MS newer). Parallel fetch jobs see mostly the same rows, and
 * `INSERT … ON CONFLICT DO UPDATE` made them wait on each other for every one.
 */
async function upsertKeyed(
  knex: Knex,
  table: 'binaries' | 'releases',
  key: string,
  rows: Record<string, unknown>[],
  merge: Record<string, [Merge, string]>
): Promise<Map<string, number>> {
  const ids = new Map<string, number>()
  if (rows.length === 0) return ids
  const inserted: Record<string, any>[] = await knex(table).insert(rows).onConflict(key).ignore().returning(['id', key])
  for (const row of inserted) ids.set(row[key], Number(row.id))

  const incoming = new Map(rows.filter((r) => !ids.has(r[key] as string)).map((r) => [r[key] as string, r]))
  if (incoming.size === 0) return ids
  const columns = Object.keys(merge)
  const existing: Record<string, any>[] = await knex(table).whereIn(key, [...incoming.keys()]).select('id', key, ...columns)

  const changes: Record<string, unknown>[] = []
  for (const row of existing) {
    ids.set(row[key], Number(row.id))
    const next = incoming.get(row[key])!
    const changed = columns.some((column) => {
      const [kind] = merge[column]
      const current = row[column]
      const value = next[column]
      if (value === null || value === undefined) return false
      if (kind === 'coalesce') return current === null
      if (kind === 'least') return current === null || value < current
      return current === null || +(value as Date) - +current > TOUCH_MS
    })
    if (changed) changes.push({ id: Number(row.id), ...Object.fromEntries(columns.map((c) => [c, next[c] ?? null])) })
  }
  if (changes.length === 0) return ids

  const set = columns
    .map((c) => {
      const [kind] = merge[c]
      const fn = kind === 'least' ? 'least' : kind === 'touch' ? 'greatest' : 'coalesce'
      return `${c} = ${fn}(t.${c}, v.${c})`
    })
    .join(', ')
  const types = columns.map((c) => `${c} ${merge[c][1]}`).join(', ')
  changes.sort((a, b) => (a.id as number) - (b.id as number))
  for (let attempt = 1; ; attempt++) {
    try {
      await knex.raw(
        `UPDATE ${table} t SET ${set}
         FROM jsonb_to_recordset(?::jsonb) AS v(id bigint, ${types})
         WHERE t.id = v.id`,
        [JSON.stringify(changes)]
      )
      return ids
    } catch (error: any) {
      // Rare now, but two jobs can still update overlapping rows in a different order.
      if (error.code !== '40P01' || attempt >= 3) throw error
    }
  }
}
/** Size-only matching below this is too ambiguous (nfo, sfv, small volumes). */
export const PAR2_SIZE_ONLY_MIN = 1024 * 1024
/** A par2 set and its files are posted together; don't match across days. */
export const PAR2_MATCH_WINDOW_SECONDS = 12 * 60 * 60
const PART_CHUNK = 2000

function chunks<T>(items: T[], size: number): T[][] {
  const result: T[][] = []
  for (let i = 0; i < items.length; i += size) result.push(items.slice(i, i + size))
  return result
}

function validDate(date?: Date) {
  return date && !Number.isNaN(date.getTime()) ? date : null
}

/** An XOVER overview entry as returned by nntp-js. */
export type OverviewHeader = Record<string, string>

function common(header: OverviewHeader, groupName: string, articleNumber?: bigint | number) {
  const xrefGroups = parseXrefGroups(header.xref)
  return {
    poster: header.from ?? '',
    messageId: (header['message-id'] ?? '').replace(/^<|>$/g, ''),
    bytes: Number.parseInt(header[':bytes'] ?? '0', 10) || 0,
    articleNumber,
    postedAt: header.date ? new Date(header.date) : undefined,
    groups: xrefGroups.length > 0 ? xrefGroups : [groupName],
  }
}

/** Builds a segment from the subject line; null when the subject is obfuscated. */
export function segmentFromHeader(
  header: OverviewHeader,
  groupName: string,
  articleNumber?: bigint | number
): IngestSegment | null {
  const parsed = header.subject ? parseSubject(header.subject) : null
  if (!parsed || !header['message-id']) return null
  const base = common(header, groupName, articleNumber)
  return {
    ...base,
    fileKey: fileKey(base.poster, parsed.base, parsed.totalSegments),
    subject: parsed.base,
    name: parsed.name,
    totalParts: parsed.totalSegments,
    number: parsed.segment,
    release: releaseIdentity({ poster: base.poster, subject: parsed.base, name: parsed.name, weak: false }),
  }
}

/**
 * Builds a segment for an obfuscated post from the yEnc header in its body.
 * Subject and poster can be random per article, so the identity comes from
 * the yEnc header only (see weakFileKey).
 */
export function segmentFromYenc(
  header: OverviewHeader,
  groupName: string,
  meta: Pick<YencMeta, 'begin' | 'part'> & Partial<Pick<YencMeta, 'end'>>,
  articleNumber?: bigint | number
): IngestSegment | null {
  const base = common(header, groupName, articleNumber)
  const totalParts = inferTotalParts(meta)
  const number = meta.begin.part ?? 1
  if (!base.messageId || number < 1 || number > totalParts) return null
  const subject = `"${meta.begin.name}" yEnc`
  return {
    ...base,
    fileKey: weakFileKey(meta.begin.name, meta.begin.size, totalParts),
    subject,
    name: meta.begin.name,
    totalParts,
    number,
    weakIdentity: true,
    release: releaseIdentity({ poster: base.poster, subject, name: meta.begin.name, weak: true }),
    size: meta.begin.size,
    // Single-part posts only carry crc32, which is then also the part's CRC.
    pcrc32: meta.end?.pcrc32 ?? (meta.part ? undefined : meta.end?.crc32),
    partSize: meta.part ? meta.part.end - meta.part.begin + 1 : meta.begin.size,
    fileCrc32: meta.end?.crc32,
  }
}

/**
 * Stores segments in `binaries`/`parts` and queues every binary that became
 * complete exactly once. Idempotent: ingesting the same segments again is a
 * no-op. See CONCEPT.md (D12).
 */
class IngestService {
  async ingest(segments: IngestSegment[]): Promise<CompletedBinary[]> {
    if (segments.length === 0) return []

    // Short statements, no enclosing transaction: every step is idempotent, and
    // row locks held for a whole batch made parallel fetch jobs wait on each other.
    const binaryIds = await (async () => {
      const knex = db.connection().getWriteClient()
      const now = new Date()
      const releaseIds = await this.upsertReleases(knex, segments, now)

      // One row per file_key (ON CONFLICT cannot touch a row twice per statement),
      // sorted so concurrent workers lock rows in the same order and cannot deadlock.
      const binaries = new Map<string, Record<string, unknown>>()
      for (const segment of segments) {
        const postedAt = validDate(segment.postedAt)
        const existing = binaries.get(segment.fileKey)
        if (existing) {
          if (postedAt && (!existing.posted_at || postedAt < (existing.posted_at as Date))) {
            existing.posted_at = postedAt
          }
          continue
        }
        binaries.set(segment.fileKey, {
          file_key: segment.fileKey,
          poster: segment.poster,
          subject: segment.subject,
          name: segment.name,
          total_parts: segment.totalParts,
          groups: JSON.stringify(segment.groups),
          posted_at: postedAt,
          first_seen_at: now,
          last_seen_at: now,
          release_id: segment.release ? releaseIds.get(segment.release.key) : null,
          size: segment.size ?? null,
          md5_16k: segment.md5_16k ?? null,
        })
      }
      const binaryRows = [...binaries.values()].sort((a, b) =>
        (a.file_key as string).localeCompare(b.file_key as string)
      )

      const idsByKey = new Map<string, number>()
      for (const chunk of chunks(binaryRows, BINARY_CHUNK)) {
        const ids = await upsertKeyed(knex, 'binaries', 'file_key', chunk, {
          last_seen_at: ['touch', 'timestamptz'],
          posted_at: ['least', 'timestamptz'],
          release_id: ['coalesce', 'bigint'],
          md5_16k: ['coalesce', 'varchar'],
        })
        for (const [key, id] of ids) idsByKey.set(key, id)
      }

      const partRows = segments
        .map((segment) => ({
          binary_id: idsByKey.get(segment.fileKey)!,
          number: segment.number,
          message_id: segment.messageId,
          bytes: segment.bytes,
          article_number: segment.articleNumber?.toString() ?? null,
          pcrc32: segment.pcrc32 ?? null,
          part_size: segment.partSize ?? null,
          file_crc32: segment.fileCrc32 ?? null,
          md5_16k: segment.md5_16k ?? null,
        }))
      const weakRows = partRows.filter((_, i) => segments[i].weakIdentity)

      partRows.sort((a, b) => a.binary_id - b.binary_id || a.number - b.number)
      for (const chunk of chunks(partRows, PART_CHUNK)) {
        await knex('parts').insert(chunk).onConflict(['binary_id', 'number']).ignore()
      }

      if (weakRows.length > 0) await this.flagConflicts(db.connection(), weakRows)

      return [...idsByKey.values()]
    })()

    // par2 naming needs a size, which only body-derived segments have; for
    // header batches these queries were most of Postgres' work.
    const sized = [...new Set(segments.filter((s) => s.size != null).map((s) => s.fileKey))]
    if (sized.length > 0) {
      const sizedIds = (await db.from('binaries').whereIn('file_key', sized).select('id')).map((r) => Number(r.id))
      await this.matchPar2Releases(sizedIds)
      await NameService.resolve(sizedIds)
    }

    // After commit, so the last writer of a binary always sees all of its parts.
    const completed = await this.claimCompleted(binaryIds)
    await this.enqueue(completed)
    return completed
  }

  /**
   * Creates or touches the releases referenced by `segments` and returns
   * their ids by release key. Also used by `release:rebuild`.
   */
  async upsertReleases(
    knex: Knex,
    segments: Pick<IngestSegment, 'release' | 'poster' | 'groups' | 'postedAt' | 'weakIdentity'>[],
    now = new Date()
  ): Promise<Map<string, number>> {
    const releases = new Map<string, Record<string, unknown>>()
    for (const segment of segments) {
      const release = segment.release
      if (!release) continue
      const postedAt = validDate(segment.postedAt)
      const existing = releases.get(release.key)
      if (existing) {
        if (postedAt && (!existing.posted_at || postedAt < (existing.posted_at as Date))) {
          existing.posted_at = postedAt
        }
        // Not every file carries the [i/M] counter (e.g. a par2 index): take it from any that does.
        if (existing.file_total === null && release.fileTotal) existing.file_total = release.fileTotal
        continue
      }
      releases.set(release.key, {
        release_key: release.key,
        name: release.name,
        // Obfuscating posters randomise the poster, so it says nothing about the release.
        poster: segment.weakIdentity ? null : segment.poster,
        file_total: release.fileTotal ?? null,
        source: release.source,
        groups: JSON.stringify(segment.groups),
        posted_at: postedAt,
        first_seen_at: now,
        last_seen_at: now,
      })
    }

    const ids = new Map<string, number>()
    const rows = [...releases.values()].sort((a, b) =>
      (a.release_key as string).localeCompare(b.release_key as string)
    )
    for (const chunk of chunks(rows, BINARY_CHUNK)) {
      const result = await upsertKeyed(knex, 'releases', 'release_key', chunk, {
        last_seen_at: ['touch', 'timestamptz'],
        posted_at: ['least', 'timestamptz'],
        file_total: ['coalesce', 'integer'],
        source: ['coalesce', 'varchar'],
      })
      for (const [key, id] of result) ids.set(key, id)
    }
    return ids
  }

  /**
   * Assigns body-derived binaries without a release and without a
   * fingerprint to a known par2 set by size: only when that size is listed
   * once in the set, the file is large, and exactly one set nearby lists it.
   * Binaries with a fingerprint are matched exactly by NameService instead.
   */
  async matchPar2Releases(binaryIds: number[]) {
    if (binaryIds.length === 0) return
    await db.rawQuery(
      `UPDATE binaries b SET release_id = m.release_id
       FROM (
         SELECT b2.id, min(rf.release_id) AS release_id
         FROM binaries b2
         JOIN release_files rf ON rf.size = b2.size
         JOIN releases r ON r.id = rf.release_id
         WHERE b2.id = ANY(?) AND b2.release_id IS NULL AND b2.md5_16k IS NULL AND b2.size >= ?
           AND abs(extract(epoch FROM (r.posted_at - b2.posted_at))) < ?
           AND (SELECT count(*) FROM release_files x WHERE x.release_id = rf.release_id AND x.size = rf.size) = 1
         GROUP BY b2.id
         HAVING count(DISTINCT rf.release_id) = 1
       ) m
       WHERE b.id = m.id`,
      [binaryIds, PAR2_SIZE_ONLY_MIN, PAR2_MATCH_WINDOW_SECONDS]
    )
  }

  /**
   * Flags binaries where a stored part has a different message-id and a
   * different part CRC than the incoming one: two files were merged under
   * one identity. Identical reposts (same CRC) are ignored.
   */
  private async flagConflicts(
    trx: QueryClientContract,
    rows: {
      binary_id: number
      number: number
      message_id: string
      bytes: number
      article_number: string | null
      pcrc32: string | null
      part_size: number | null
      file_crc32: string | null
      md5_16k: string | null
    }[]
  ) {
    const bindings = [
      rows.map((r) => r.binary_id),
      rows.map((r) => r.number),
      rows.map((r) => r.message_id),
      rows.map((r) => r.bytes),
      rows.map((r) => r.article_number),
      rows.map((r) => r.pcrc32),
      rows.map((r) => r.part_size),
      rows.map((r) => r.file_crc32),
      rows.map((r) => r.md5_16k),
    ]
    // Keep the extra articles: SplitService separates the volumes from them.
    const conflicts = await trx.rawQuery(
      `INSERT INTO part_conflicts (binary_id, number, message_id, bytes, article_number, pcrc32, part_size, file_crc32, md5_16k)
       SELECT v.binary_id, v.number, v.message_id, v.bytes, v.article_number, v.pcrc32, v.part_size, v.file_crc32, v.md5_16k
       FROM unnest(?::bigint[], ?::int[], ?::text[], ?::int[], ?::bigint[], ?::text[], ?::int[], ?::text[], ?::text[])
         AS v(binary_id, number, message_id, bytes, article_number, pcrc32, part_size, file_crc32, md5_16k)
       JOIN parts p ON p.binary_id = v.binary_id AND p.number = v.number
       WHERE p.message_id <> v.message_id
         -- Same part CRC: the same data posted again (a repost), not another file.
         AND (p.pcrc32 IS NULL OR v.pcrc32 IS NULL OR p.pcrc32 <> v.pcrc32)
       ON CONFLICT DO NOTHING
       RETURNING binary_id`,
      bindings as any
    )
    const binaryIds: number[] = [
      ...new Set<number>(conflicts.rows.map((r: { binary_id: string }) => Number(r.binary_id))),
    ]
    if (binaryIds.length === 0) return
    const result = await trx.rawQuery(
      `UPDATE binaries SET conflicted_at = now()
       WHERE conflicted_at IS NULL AND id = ANY(?) RETURNING name`,
      [binaryIds]
    )
    for (const row of result.rows) {
      logger.warn(`Binary "${row.name}" has conflicting parts; it will be split into volumes.`)
    }
  }

  /**
   * Marks complete binaries among `binaryIds` as claimed. The single UPDATE
   * locks each row, so concurrent callers never claim the same binary twice.
   */
  async claimCompleted(binaryIds: number[]): Promise<CompletedBinary[]> {
    if (binaryIds.length === 0) return []
    const result = await db.rawQuery(
      `UPDATE binaries b SET completed_at = now()
       WHERE b.id = ANY(?) AND b.completed_at IS NULL AND b.conflicted_at IS NULL
         AND (SELECT count(*) FROM parts p WHERE p.binary_id = b.id) >= b.total_parts
       RETURNING b.id, b.file_key`,
      [binaryIds]
    )
    return result.rows.map((row: { id: string; file_key: string }) => ({
      id: Number(row.id),
      fileKey: row.file_key,
    }))
  }

  async enqueue(completed: CompletedBinary[]) {
    if (completed.length === 0) return
    // jobId = file_key: BullMQ ignores an add() for an id that already exists.
    await QueueService.fileQueue.addBulk(
      completed.map(({ id, fileKey }) => ({
        name: 'process-file',
        data: { binaryId: id },
        opts: { jobId: fileKey },
      }))
    )
  }
}

export default new IngestService()