265 lines
11 KiB
TypeScript
265 lines
11 KiB
TypeScript
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()
|