From ce698d7782e45af626d61228899b637083454a6e Mon Sep 17 00:00:00 2001 From: Dave Rolle Date: Wed, 5 Aug 2026 15:41:20 +0200 Subject: [PATCH] Carousels: deliver every video, and stop shouting about photos MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An Instagram photo carousel has no video, so yt-dlp fails once per photo item. In a group that produced a šŸ’„ report quoting yt-dlp's raw plumbing: "Dbnd91uAyMW: No video formats found!; please report this issue on https://github.com/yt-dlp/yt-dlp/issues?q= ...". The same went for any link whose target is not media at all: an archive.ph page holding a PDF was reported as a download failure. Worse, a post that mixes photos and videos delivered only its first video. The rest were dropped silently, because a post resolved to a single info. A post is now scraped as a list. One job per message URL delivers every video it holds, in order, one reply per video, capped at ten (the scrape looks at fifty, because a carousel spends a playlist index on each of its photos too). The bookkeeping that used to track "the video" is now per-entry, keyed on yt-dlp's identity minus the format so a re-scrape that drifts still matches: which videos are settled, which have been announced, whether anything reached the chat at all. A run whose only casualties are photo items is salvaged: the videos that did resolve are kept and cached, and the photos are not treated as a failure. A post whose items are ALL photos, and a link to a non-media file, are both "not a video": one attempt, no retry, and a group hears nothing, matching the existing policy for photo posts. A lone item that reports no formats is still a real failure the chat is told about, since a carousel is what emits one such line per item (verified against the real yt-dlp: a nine-photo post gives nine). User-facing errors no longer carry yt-dlp's "please report this issue" boilerplate, and a post's one report is now elected: a retryable failure outranks a permanent one, and a photo item never speaks for a sibling that failed for a real reason. - A bare playlist or channel URL now delivers its first ten videos with a notice, where it used to take the first only. - A confirmation parked for one video of a post no longer un-records the whole message, so cancelling it cannot re-send what already landed. - Migration 6 drops the cached video_info rows: the payload went from one - yt-dlp introduces its issue-tracker paragraph with a clause of its own in some errors; the whole invitation is stripped, not just its tail. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01K7QTJrxfmVdrARFcmmKtXN --- e2e.sh | 4 +- src/blob-store.ts | 18 +- src/db.ts | 3 + src/download-video.ts | 168 +++++-- src/handlers.ts | 265 ++++++---- src/job-queue.ts | 33 +- test/bin/yt-dlp | 4 +- test/blob-store.test.ts | 34 ++ test/db.test.ts | 21 +- test/download-video.test.ts | 359 +++++++++++++- test/e2e.test.ts | 51 +- test/handlers.test.ts | 944 +++++++++++++++++++++++++++++++++++- test/job-queue.test.ts | 22 + test/test-utils.ts | 6 +- 14 files changed, 1738 insertions(+), 194 deletions(-) diff --git a/e2e.sh b/e2e.sh index 57cfe77..810935a 100755 --- a/e2e.sh +++ b/e2e.sh @@ -1,8 +1,8 @@ #!/bin/sh # Usage: ./e2e.sh [full] [-u] # full also test rate-limit-prone sites (youtube rejects more than a few -# hits per hour). Used by the deploy gate (prod.sh); the pre-push -# hook runs the reduced set. +# hits per hour) and the slow multi-video cases. Used by the deploy +# gate (prod.sh); the pre-push hook runs the reduced set. # -u refresh snapshots. Implies full: `bun test -u` deletes snapshots of # tests it didn't run, so a reduced -u run would silently prune the # full-mode snapshots. diff --git a/src/blob-store.ts b/src/blob-store.ts index 7c1bb0b..63e09ac 100644 --- a/src/blob-store.ts +++ b/src/blob-store.ts @@ -12,6 +12,11 @@ const envDir = Bun.env.BLOB_DIR || '/storage/blobs/'; const BLOB_DIR = envDir.endsWith('/') ? envDir : `${envDir}/`; await mkdir(BLOB_DIR, { recursive: true }); +// the generic extractor's id is just the URL basename: two hosts' /video.mp4 +// both yield id 'video' +const hasVideoIdentity = (info: VideoInfo): boolean => + !!(info.extractor && info.id && info.extractor !== 'generic'); + // A blob's key is yt-dlp's stable per-video identity, extractor:id:format // (known from --dump-json before the download), so a blob we already have is found // without re-downloading, and two URLs for one video resolve to the same key. @@ -21,16 +26,21 @@ await mkdir(BLOB_DIR, { recursive: true }); // with colliding titles would otherwise share a key (and, worse, the first one's // cached file_id). This is the raw DB key; blobName turns it into the on-disk // filename. -// the generic extractor's id is just the URL basename (verified against the -// real yt-dlp: two hosts' /video.mp4 both yield id 'video'), so it is NOT an -// identity; those go through the URL-salted fallback like identity-less infos export const blobKey = (info: VideoInfo): string => - info.extractor && info.id && info.extractor !== 'generic' + hasVideoIdentity(info) ? `${info.extractor}:${info.id}:${info.format_id ?? ''}` : info.webpage_url ? `${info.filename}:${info.webpage_url}` : info.filename; +// The fallback cannot reuse blobKey's filename, which carries the format id (see +// yt-dlp.conf's --output); yt-dlp suffixes " (N)" to the titles of one page's +// entries, so title still separates them. +export const videoKey = (info: VideoInfo): string => + hasVideoIdentity(info) + ? `${info.extractor}:${info.id}` + : `${info.title}:${info.id ?? ''}:${info.webpage_url ?? ''}`; + const extOf = (info: VideoInfo) => { const e = info.ext || diff --git a/src/db.ts b/src/db.ts index 450a3ba..19bfd6e 100644 --- a/src/db.ts +++ b/src/db.ts @@ -82,6 +82,9 @@ export const MIGRATIONS: string[] = [ UPDATE video_info SET webpage_url = json_extract(info, '$.webpage_url'); CREATE INDEX video_info_webpage_col ON video_info (webpage_url); DROP INDEX video_info_webpage;`, + // `info` holds the JSON array of a post's video entries; rows predating that + // hold one bare object. A six-hour cache, so drop them instead of converting. + `DELETE FROM video_info;`, ]; // Exported so the migration tests replay THIS loop against a scratch DB (a diff --git a/src/download-video.ts b/src/download-video.ts index c718451..2204f82 100644 --- a/src/download-video.ts +++ b/src/download-video.ts @@ -38,33 +38,44 @@ const getErrorMessage = (proc: Bun.ReadableSubprocess) => ? `yt-dlp was killed with signal ${proc.signalCode}` : `yt-dlp exited with code ${proc.exitCode}`; +const errorLines = (stderr: string) => + stderr.split('\n').filter((line) => line.startsWith('ERROR:')); + // carries the failing yt-dlp's stderr so callers can classify it (see // isPermanentError) export class YtdlpError extends Error { + declare readonly stdout: string; constructor( message: string, readonly stderr: string, // killed by a signal (timeout/OOM) rather than exiting with a code readonly signalled = false, + stdout = '', ) { - // yt-dlp's own ERROR: line (the last one is the fatal one) says WHY it - // failed; the exit code alone helps nobody, so it's only the fallback. + // yt-dlp's own ERROR: line says WHY it failed; the exit code alone helps + // nobody, so it's only the fallback. // A signal kill keeps its message: the timeout is the real story there, // and any ERROR line in a killed process's output is a stale partial. - // De-noise for the report: drop the [extractor] tag and the "(caused - // by ...)" suffix that repeats the main clause; the raw line still - // streams to the chat verbatim, and classification reads raw stderr. - const errLine = signalled - ? undefined - : stderr - .split('\n') - .findLast((line) => line.startsWith('ERROR:')) - ?.slice('ERROR:'.length) - .replace(/^\s*\[[^\]]+\]\s*/, '') - .replace(/\s*\(caused by .*\)\s*$/, '') - .trim(); + const lines = signalled ? [] : errorLines(stderr); + const errLine = (lines.findLast((l) => !PHOTO_ITEM.test(l)) ?? lines.at(-1)) + ?.slice('ERROR:'.length) + // De-noising is lossless: the raw line still streams to the chat + // verbatim, and classification reads raw stderr. + .replace(/^\s*\[[^\]]+\]\s*/, '') + // yt-dlp appends this after any "(caused by ...)", so it has to be + // stripped first + .replace( + /[;,]?\s*(?:if you believe this is an error,\s*)?please report this issue on .*/i, + '', + ) + .replace(/\s*\(caused by .*\)\s*$/, '') + .trim(); super(errLine || message); this.name = 'YtdlpError'; + // hidden from console.error, which prints an Error's own properties whole: + // a playlist's dump-json runs to megabytes (see STDERR_TAIL for the cap on + // the other half) + Object.defineProperty(this, 'stdout', { value: stdout, enumerable: false }); } } @@ -72,11 +83,14 @@ export type FailureKind = 'not-a-video' | 'unavailable' | 'transient'; const NOT_A_VIDEO_PATTERNS = [ /unsupported url/i, - // the extractor tag is required by each pattern: yt-dlp's f4m downloader emits an - // untagged "No media found" for a transient condition that must stay retryable /\[reddit\] .+: no media found/i, /\[instagram\] .+: there is no video in this post/i, + /extension \(.+\) is unusual and will be skipped/i, ]; +// The extractor raises the friendly message above only for a single post; a +// carousel is a playlist, whose photo items each fall through to yt-dlp's +// generic no-formats error, a verdict on the item rather than the post. +const PHOTO_ITEM = /\[instagram\] .+: no video formats found/i; const PERMANENT_PATTERNS = [ /unable to extract/i, /no video formats found/i, @@ -95,12 +109,15 @@ const PERMANENT_PATTERNS = [ export const classifyFailure = (e: unknown): FailureKind => { if (!(e instanceof YtdlpError) || e.signalled) return 'transient'; - const errorLines = e.stderr - .split('\n') - .filter((line) => line.startsWith('ERROR:')); - if (errorLines.some((l) => NOT_A_VIDEO_PATTERNS.some((re) => re.test(l)))) + const all = errorLines(e.stderr); + if (!all.length) return 'transient'; + // one line is the whole post reporting no formats, a failure the chat hears + // about + const rest = all.length > 1 ? all.filter((l) => !PHOTO_ITEM.test(l)) : all; + if (!rest.length) return 'not-a-video'; + if (rest.some((l) => NOT_A_VIDEO_PATTERNS.some((re) => re.test(l)))) return 'not-a-video'; - if (errorLines.some((l) => PERMANENT_PATTERNS.some((re) => re.test(l)))) + if (rest.some((l) => PERMANENT_PATTERNS.some((re) => re.test(l)))) return 'unavailable'; return 'transient'; }; @@ -412,17 +429,20 @@ const execYtdlp = limit( // our own shutdown kill, not a failure: the queue leaves the job for the // next boot (a timeout kill sets no flag and stays a retryable YtdlpError) if (shuttingDown && proc.signalCode != null) throw new ShutdownAbort(); + + // stdout is read last, after stderr is fully drained: Bun buffers a piped + // child's stdout, so draining stderr to EOF first can't deadlock on a full + // stdout pipe (it would if stdout were unbuffered and left unread) + const stdout = await Bun.readableStreamToText(proc.stdout); if (proc.exitCode !== 0) throw new YtdlpError( getErrorMessage(proc), stderr, proc.signalCode != null, + stdout, ); - // stdout is read last, after stderr is fully drained: Bun buffers a piped - // child's stdout, so draining stderr to EOF first can't deadlock on a full - // stdout pipe (it would if stdout were unbuffered and left unread) - return await Bun.readableStreamToText(proc.stdout); + return stdout; }, ); @@ -451,16 +471,36 @@ const sweepInfoStmt = db.query( export const sweepStaleInfo = () => sweepInfoStmt.run(Date.now() - INFO_TTL_MS); +// a playlist or channel URL resolves to the same multi-entry shape as a post, +// but runs to thousands of entries +export const MAX_POST_VIDEOS = 10; +// yt-dlp spends a playlist index on a carousel's photos too, so a bound at the +// delivery cap would hide videos behind them +const MAX_SCRAPE_ITEMS = 50; + +const onlyPhotoItemsFailed = (e: YtdlpError) => { + const lines = errorLines(e.stderr); + // no ERROR line at all means the run failed for a reason this stderr no + // longer holds: the tail trim drops the oldest lines + return !!lines.length && lines.every((line) => PHOTO_ITEM.test(line)); +}; + +const parseEntries = (out: string): VideoInfo[] => + out + .split('\n') + .filter((line) => line.startsWith('{')) + .map((line) => JSON.parse(line) as VideoInfo); + // Coalesced, not memoized: the DB row above serves repeat lookups, so keeping // settled results in memory would only pin megabytes of dump-json per URL and // grow stale (see INFO_TTL_MS); the in-flight entry alone stops two concurrent // jobs from both scraping. -export const getInfo = coalesce( +export const getInfos = coalesce( async ( log: LogMessage, url: string, verbose: boolean = false, - ): Promise => { + ): Promise => { // a verbose request bypasses the cache (like the coalesce key below) so its // yt-dlp output is actually streamed to the chat for debugging const cached = selectInfoStmt.get(url, Date.now() - INFO_TTL_MS); @@ -468,41 +508,79 @@ export const getInfo = coalesce( log.append(`šŸ§ Scraping ${url}...`); - const infoStr = await execYtdlp(log, url, verbose, '--dump-json'); - const info = JSON.parse(infoStr) as VideoInfo; - const hadCanonical = !!info.webpage_url; - info.webpage_url ||= url; - // narrowed to a truthy string by the ||= above; a local const carries that - // through the tx closure (TS drops property narrowing across the boundary) - const canonical = info.webpage_url; - // also key a row by the canonical webpage_url so a later alias request - // for the same video hits the cache instead of re-scraping. Reuse - // yt-dlp's own string when nothing changed: re-serializing a multi-MB - // payload just to store identical content is wasted event-loop time. - const str = hadCanonical ? infoStr : JSON.stringify(info); + let out: string; + try { + out = await execYtdlp( + log, + url, + verbose, + '--dump-json', + // A link copied off a playing YouTube page carries the playlist it was + // playing in, and yt-dlp would take the playlist over the video. + // Instagram carousels are unaffected: they come back as a playlist + // whether or not this flag is set. + '--no-playlist', + '--playlist-end', + String(MAX_SCRAPE_ITEMS), + ); + } catch (e) { + // A carousel mixing photos with videos fails on every photo item, so the + // run exits non-zero even though it resolved the videos we came for. A + // signal kill (a timeout) is different: its output stops mid-post, and + // possibly mid-line, so there is nothing trustworthy to salvage. + if ( + !(e instanceof YtdlpError) || + e.signalled || + !e.stdout.trim() || + !onlyPhotoItemsFailed(e) + ) { + throw e; + } + out = e.stdout; + } + const infos = parseEntries(out); + if (!infos.length) throw new Error('yt-dlp found no video at that URL'); + + for (const info of infos) info.webpage_url ||= url; + const canonical = infos[0]!.webpage_url!; + const namesAll = infos.every((i) => i.webpage_url === canonical); + const str = JSON.stringify(infos); const now = Date.now(); + // A salvaged list that really did stop short is dropped by removeCachedUrl + // when a job cannot find its entry in it. tx(() => { insertInfoStmt.run(url, str, canonical, now); - if (canonical !== url) { + if (namesAll && canonical !== url) { insertInfoStmt.run(canonical, str, canonical, now); } }); - return info; + return infos; }, (_log, url, verbose) => !verbose && url, ); +export const getInfo = async ( + log: LogMessage, + url: string, + verbose: boolean = false, +): Promise => (await getInfos(log, url, verbose))[0]!; + // Evict a video's cached info (the url row and its canonical alias share one // webpage_url) after a download failure: the likely cause is the expired signed // URLs above, so on the retry (or the next request) getInfo must re-scrape // rather than replay the same doomed row. -const deleteInfoStmt = db.query( - `DELETE FROM video_info WHERE webpage_url = ?`, +const deleteInfoStmt = db.query( + `DELETE FROM video_info WHERE webpage_url = ? OR url = ?`, ); -export const removeCachedInfo = (info: VideoInfo) => { - if (info.webpage_url) deleteInfoStmt.run(info.webpage_url); +// `url` is the request that produced the row: a playlist's entries carry their +// own webpage_url, so the entry alone cannot name the row it came from. +export const removeCachedInfo = (info: VideoInfo, url?: string) => { + const key = info.webpage_url || url; + if (key) deleteInfoStmt.run(key, url ?? key); }; +export const removeCachedUrl = (url: string) => deleteInfoStmt.run(url, url); + const logFormats = ({ formats }: any) => // log all formats for debugging purposes formats && diff --git a/src/handlers.ts b/src/handlers.ts index 14b869d..3790f4c 100644 --- a/src/handlers.ts +++ b/src/handlers.ts @@ -5,6 +5,7 @@ import { releaseAbandoned, releaseBlob, setBlobDuration, + videoKey, withBlobLock, } from './blob-store'; import { db } from './db'; @@ -13,10 +14,13 @@ import { classifyFailure, downloadVideo, getInfo, + getInfos, isDownloaded, isPermanentError, + MAX_POST_VIDEOS, probeDuration, removeCachedInfo, + removeCachedUrl, sendInfo, sendVideo, tooLargeMessage, @@ -211,6 +215,19 @@ const isAlwaysRespondHost = (url: string): boolean => { const isTerminal = (e: unknown, attempt: number, kind?: FailureKind) => isPermanentError(e, kind) || attempt >= MAX_ATTEMPTS; +// A retryable failure outranks a permanent one, or the post's other videos +// never retry; a not-a-video item draws no group reply at all (see tellGroup), +// so it must never answer for a sibling that failed for a real reason. +const speaksFor = (e: unknown) => + !isPermanentError(e) ? 2 : classifyFailure(e) === 'not-a-video' ? 0 : 1; + +const markSettled = (job: UrlJob, info: VideoInfo) => { + (job.settledIds ??= []).push(videoKey(info)); +}; +const markAnnounced = (job: UrlJob, info: VideoInfo) => { + (job.announcedIds ??= []).push(videoKey(info)); +}; + const processUrlJob = async ( telegram: Telegram, job: UrlJob, @@ -219,92 +236,146 @@ const processUrlJob = async ( const { url, chatId, chatType, messageId, verbose } = job; // progress logs go to private chats only: see logFor const log = logFor(telegram, chatType, logDestFor(job)); + const isGroupChat = chatType !== 'private'; + // A private thread whose sends all failed has no messageId and carried + // nothing, so skipping the video on the retry would lose the verdict in + // silence. A group is silent by policy and misses nothing. + const verdictReachedChat = () => isGroupChat || log.messageId != null; + let reopenWanted = false; + const flushReopen = () => { + if (reopenWanted && !job.answered) reopenEditRetry(job, url); + }; let info: VideoInfo | undefined; + // whether any ENTRY failed in yt-dlp, which the elected report may not say: + // a sibling's send failure can outrank it and still leave stale info cached + let scrapeStale = false; + const failedEntries: VideoInfo[] = []; try { - info = await getInfo(log, url, verbose); - // Print the info block once per DELIVERED thread: the flag says a prior - // attempt appended it, and logMessageId says that attempt's sends - // actually reached the chat (all-failed sends stash undefined, and the - // retry posts a fresh thread that needs the info again). - if (!(job.infoShown && job.logMessageId != null)) { - await sendInfo(log, info, verbose); - } - job.infoShown = true; - // a long video is often also too big to send; reject from the scraped - // estimate before downloading (or offering to download) something we can - // never deliver. sendVideo still gates on the real on-disk size for an - // estimate that was missing or wrong. (A group's NoLog stays silent here, - // matching the group-silence policy above.) - const tooLarge = tooLargeToSend(info); - if (tooLarge) { - log.append(`\n${tooLargeMessage(tooLarge)}`); - await log.flush(); - // estimates are unreliable and formats change, so an edit must be able - // to retry this verdict too - reopenEditRetry(job, url); - return; - } - // scraped metadata can lack duration; the blob row keeps the ffprobe'd - // real one from a previous download only if some past probe SUCCEEDED. A - // probe-failed, later-disposed video (only file_id remains) stays unknown - // and falls through. - // `||`, not `??`: a scraped duration of 0 means "unknown" (the same reason - // the post-download backstop re-checks 0), so it too falls through to the - // blob row's probed duration - const duration = calcDuration(info) || getBlob(info)?.duration; - const isGroupChat = chatType !== 'private'; - if (isGroupChat && duration && duration > LONG_VIDEO_THRESHOLD_SECS) { - await requestConfirmation(telegram, job, info, duration); - return; + const all = await getInfos(log, url, verbose); + const infos = all.slice(0, MAX_POST_VIDEOS); + const many = infos.length > 1 || undefined; + if (all.length > infos.length && !(job.capShown && job.logMessageId != null)) { + log.append( + `\nšŸ“š More than ${infos.length} videos here; sending the first ${infos.length}.`, + ); + job.capShown = true; } - // set inside the lock when the post-download gate parks a confirmation, so - // the too-large un-record below skips that (non-terminal) return path - let confirmed = false; - // serialize every byte-touching step for this video (download, probe, send) - // so a concurrent job for the same blob takes turns with us: it reuses our - // result or re-downloads cleanly, instead of racing us on the bytes - const sent = await withBlobLock(info, async () => { - console.debug(await downloadVideo(log, info!, verbose)); - if (isGroupChat) { - // The real duration, probed and stored during the download just above - // (or during the first download, when this one was a cache hit). A - // null row value with bytes present (a crash landed between recording - // the blob and storing the duration, or that probe failed once) is - // re-probed here while the bytes are still on disk. - const blob = getBlob(info!); - let actualDuration = blob?.duration; - if (!actualDuration && blob && !blob.file_id) { - actualDuration = await probeDuration(blob.path); - if (actualDuration) setBlobDuration(info!, actualDuration); + if (job.logMessageId == null) job.announcedIds = undefined; + let failure: unknown; + // undefined is a value a rejection can carry + let anyFailed = false; + // The queue's bookkeeping, the edit-retry gesture and the one-reply-per-link + // policy are all keyed to the URL the message carries, so one job must + // deliver every video of the post. + for (const entry of infos) { + const key = videoKey(entry); + if (job.settledIds?.includes(key)) continue; + info = entry; + try { + if (!job.announcedIds?.includes(key)) { + await sendInfo(log, info, verbose); + markAnnounced(job, entry); } - if (actualDuration && actualDuration > LONG_VIDEO_THRESHOLD_SECS) { - // Enrich the parked payload too, so the confirmed job sends the - // video with its real duration metadata. The probed duration is - // already net of removed sponsor segments, so the chapters must go - // or calcDuration would subtract them a second time. - const infoWithDuration = { - ...info!, - duration: actualDuration, - sponsorblock_chapters: undefined, - }; - await requestConfirmation( - telegram, - job, - infoWithDuration, - actualDuration, - true, - ); - confirmed = true; - return; + // a long video is often also too big to send; reject from the scraped + // estimate before downloading (or offering to download) something we can + // never deliver. sendVideo still gates on the real on-disk size for an + // estimate that was missing or wrong. (A group's NoLog stays silent here, + // matching the group-silence policy above.) + const tooLarge = tooLargeToSend(info); + if (tooLarge) { + log.append(`\n${tooLargeMessage(tooLarge)}`); + await log.flush(); + // estimates are unreliable and formats change, so an edit must be able + // to retry this verdict too + reopenWanted = true; + if (verdictReachedChat()) markSettled(job, info); + continue; + } + // scraped metadata can lack duration; the blob row keeps the ffprobe'd + // real one from a previous download only if some past probe SUCCEEDED. A + // probe-failed, later-disposed video (only file_id remains) stays unknown + // and falls through. + // `||`, not `??`: a scraped duration of 0 means "unknown" (the same reason + // the post-download backstop re-checks 0), so it too falls through to the + // blob row's probed duration + const duration = calcDuration(info) || getBlob(info)?.duration; + if (isGroupChat && duration && duration > LONG_VIDEO_THRESHOLD_SECS) { + await requestConfirmation(telegram, job, info, duration, false, many); + job.answered = true; + markSettled(job, info); + continue; + } + // set inside the lock when the post-download gate parks a confirmation, so + // the too-large un-record below skips that (non-terminal) path + let confirmed = false; + const current = info; + // serialize every byte-touching step for this video (download, probe, send) + // so a concurrent job for the same blob takes turns with us: it reuses our + // result or re-downloads cleanly, instead of racing us on the bytes + const sent = await withBlobLock(current, async () => { + console.debug(await downloadVideo(log, current, verbose)); + if (isGroupChat) { + // The real duration, probed and stored during the download just above + // (or during the first download, when this one was a cache hit). A + // null row value with bytes present (a crash landed between recording + // the blob and storing the duration, or that probe failed once) is + // re-probed here while the bytes are still on disk. + const blob = getBlob(current); + let actualDuration = blob?.duration; + if (!actualDuration && blob && !blob.file_id) { + actualDuration = await probeDuration(blob.path); + if (actualDuration) setBlobDuration(current, actualDuration); + } + if (actualDuration && actualDuration > LONG_VIDEO_THRESHOLD_SECS) { + // Enrich the parked payload too, so the confirmed job sends the + // video with its real duration metadata. The probed duration is + // already net of removed sponsor segments, so the chapters must go + // or calcDuration would subtract them a second time. + const infoWithDuration = { + ...current, + duration: actualDuration, + sponsorblock_chapters: undefined, + }; + await requestConfirmation( + telegram, + job, + infoWithDuration, + actualDuration, + true, + many, + ); + confirmed = true; + return; + } + } + return sendVideo(telegram, log, current, chatId, messageId); + }); + // sendVideo returns undefined when the real on-disk bytes exceeded the + // limit (a missing/under estimate slipped past tooLargeToSend above); it + // already discarded them. Ask for the gesture like a terminal verdict. + // The confirmation path (confirmed) is not a too-large one. + if (!sent && !confirmed) { + // flush so verdictReachedChat below sees the landed thread's id + await log.flush(); + reopenWanted = true; + } + if (sent || confirmed) job.answered = true; + if (sent || confirmed || verdictReachedChat()) { + markSettled(job, current); + } + } catch (e) { + // shutdown is not this post's failure: it must reach the queue whole + if (e instanceof ShutdownAbort) throw e; + failedEntries.push(entry); + scrapeStale ||= e instanceof YtdlpError; + if (!anyFailed || speaksFor(e) > speaksFor(failure)) { + failure = e; + anyFailed = true; } } - return sendVideo(telegram, log, info!, chatId, messageId); - }); - // sendVideo returns undefined when the real on-disk bytes exceeded the - // limit (a missing/under estimate slipped past tooLargeToSend above); it - // already discarded them. Un-record like a terminal verdict so an edit can - // retry. The confirmation return path (confirmed) is not a too-large one. - if (!sent && !confirmed) reopenEditRetry(job, url); + } + if (anyFailed) throw failure; + flushReopen(); } catch (e: any) { // not a failure: no report, no eviction, no release; stash the log // pointer for the re-run (see ShutdownAbort). Flush first: a debounced @@ -317,9 +388,11 @@ const processUrlJob = async ( } // a failed download often means the cached info's signed media URLs have // expired: evict so the retry (or the next request) re-scrapes. Scoped to - // yt-dlp failures: after a send failure the info is fine, and keeping it - // guarantees the retry maps to the same blob key and reuses the bytes. - if (info && e instanceof YtdlpError) removeCachedInfo(info); + // yt-dlp failures: a post whose videos only failed to SEND still has good + // info, and keeping it maps the retry to the same blob key and reuses the + // bytes. + if (info && (scrapeStale || e instanceof YtdlpError)) + removeCachedInfo(info, url); const kind = classifyFailure(e); const terminal = isTerminal(e, attempt, kind); // product policy: a not-a-video post (photo/article) never draws a group @@ -343,8 +416,9 @@ const processUrlJob = async ( // reached only on a terminal failure (reportJobFailure rethrows retryable // ones, whose retry reuses the blob; a parked confirmation returned above). // Release the bytes this dead job downloaded. - if (info) await releaseAbandoned(info); - reopenEditRetry(job, url); + for (const f of failedEntries) await releaseAbandoned(f); + reopenWanted = true; + flushReopen(); } }; @@ -367,14 +441,21 @@ const processConfirmedJob = async ( // The payload pins the info snapshot the user confirmed, but its embedded // signed media URLs expire in hours; a confirm clicked later than that // would replay them into guaranteed 403s for every attempt. When there is - // no blob yet (nothing downloaded to reuse), re-resolve through getInfo: + // no blob yet (nothing downloaded to reuse), re-resolve through getInfos: // fresh within its TTL is a cheap DB hit, stale re-scrapes live URLs. // (Re-checked per attempt; a doomed replay evicts its row below, so the - // NEXT attempt's getInfo re-scrapes. No unconditional retry refresh: a + // NEXT attempt's getInfos re-scrapes. No unconditional retry refresh: a // retry whose blob survived, the common transient-send case, must reuse // its cached file_id rather than gamble on a fresh scrape.) if (info.webpage_url && !(await isDownloaded(info))) { - info = await getInfo(log, info.webpage_url, verbose); + const fresh = await getInfos(log, info.webpage_url, verbose); + const want = videoKey(info); + const resolved = fresh.find((i) => videoKey(i) === want); + if (!resolved) { + removeCachedUrl(info.webpage_url); + throw new Error('that video is no longer in the post'); + } + info = resolved; } // the re-resolve can drift the key (e.g. a different format_id), stranding // the parked identity's (fileless) row; released here once so every outcome @@ -399,14 +480,14 @@ const processConfirmedJob = async ( const r = report(); r.append(tooLargeMessage()); await r.flush(); - reopenEditRetry(job, job.url); + if (!job.partOfPost) reopenEditRetry(job, job.url); } } catch (e: any) { // a shutdown abort is not a failure; see processUrlJob's twin guard if (e instanceof ShutdownAbort) throw e; // evict likely-expired cached info so the NEXT request re-scrapes; this // job's own retries can't benefit (the payload pins its info snapshot) - if (e instanceof YtdlpError) removeCachedInfo(info); + if (e instanceof YtdlpError) removeCachedInfo(info, job.url); const terminal = isTerminal(e, attempt); await reportJobFailure(job, report(), e, attempt, terminal); // terminal failure (retryable ones rethrew above and will reuse the blob): @@ -415,7 +496,7 @@ const processConfirmedJob = async ( // un-record the originating message's URL so editing it retries, exactly // like a terminal url job (the payload carries the url the record used; // info.webpage_url may be a different alias) - reopenEditRetry(job, job.url); + if (!job.partOfPost) reopenEditRetry(job, job.url); } }; @@ -475,6 +556,7 @@ const requestConfirmation = async ( info: VideoInfo, duration: number, postDownload: boolean = false, + partOfPost: true | undefined = undefined, ) => { const id = await addPending({ info, @@ -485,6 +567,7 @@ const requestConfirmation = async ( chatType: job.chatType, userId: job.fromId, postDownload, + partOfPost, }); try { diff --git a/src/job-queue.ts b/src/job-queue.ts index f497144..ffea952 100644 --- a/src/job-queue.ts +++ b/src/job-queue.ts @@ -21,10 +21,21 @@ export type UrlJob = JobBase & { kind: 'url'; url: string; fromId: number; - // whether the info block already reached the chat, so a retry's continued - // thread doesn't print it twice (its text may sit in an earlier chunk than - // the one logText carries, so string-matching logText can't answer this) - infoShown?: boolean; + // the videos of a multi-video post this job is done with (sent, or ruled out + // by a gate), by videoKey, so a retry picks up where it stopped + settledIds?: string[]; + // and the ones whose info block already reached the chat, so a retry does + // not print it again for a video it announced but could not deliver + announcedIds?: string[]; + // whether the post has answered the message at all: a video sent, or a + // confirmation prompt parked. Either is a promise the edit-retry gesture + // must not undo by re-running the whole post. + answered?: true; + // whether the "more than N videos here" notice already reached the chat, so + // a retry's continued thread doesn't print it twice (its text may sit in an + // earlier chunk than the one logText carries, so string-matching logText + // can't answer this) + capShown?: boolean; }; export type ConfirmedJob = JobBase & { @@ -36,6 +47,10 @@ export type ConfirmedJob = JobBase & { // un-record it and re-open the edit-retry gesture. Optional: rows parked // before this field existed lack it and just skip the un-record. url?: string; + // this video is one of several in its post, so the message's record belongs + // to the job delivering them all: un-recording it here would re-deliver the + // ones that already landed + partOfPost?: true; }; export type Job = UrlJob | ConfirmedJob; @@ -270,7 +285,10 @@ const run = async (id: number) => { const before = { logMessageId: job.logMessageId, logText: job.logText, - infoShown: (job as UrlJob).infoShown, + capShown: (job as UrlJob).capShown, + settled: (job as UrlJob).settledIds?.length, + announced: (job as UrlJob).announcedIds?.length, + answered: (job as UrlJob).answered, }; try { await processor!(job, attempt); @@ -278,7 +296,10 @@ const run = async (id: number) => { const dirty = before.logMessageId !== job.logMessageId || before.logText !== job.logText || - before.infoShown !== (job as UrlJob).infoShown; + before.capShown !== (job as UrlJob).capShown || + before.settled !== (job as UrlJob).settledIds?.length || + before.announced !== (job as UrlJob).announcedIds?.length || + before.answered !== (job as UrlJob).answered; const persistJob = (attempts: number) => dirty ? bumpAttemptsStmt.run(attempts, JSON.stringify(job), id) diff --git a/test/bin/yt-dlp b/test/bin/yt-dlp index 7a0258f..c688c89 100755 --- a/test/bin/yt-dlp +++ b/test/bin/yt-dlp @@ -8,8 +8,10 @@ d=/tmp/stub echo "$0 $*" >> "$d/args" while [ -f "$d/block" ]; do sleep 0.05; done [ -f "$d/stderr" ] && cat "$d/stderr" >&2 -[ -f "$d/signal" ] && kill "-$(cat "$d/signal")" $$ +# the real binary streams entries as it resolves them, so a kill lands on a +# run that has already emitted some [ -f "$d/stdout" ] && cat "$d/stdout" +[ -f "$d/signal" ] && kill "-$(cat "$d/signal")" $$ # simulate producing the downloaded file at the path named in the outfile # control, honoring a `--paths home:` arg the way the real binary does # (the output lands under that dir) diff --git a/test/blob-store.test.ts b/test/blob-store.test.ts index 43ca9e6..da061ae 100644 --- a/test/blob-store.test.ts +++ b/test/blob-store.test.ts @@ -12,6 +12,7 @@ import { BLOB_TTL_MS, blobKey, blobPath, + videoKey, getBlob, recordBlob, releaseBlob, @@ -49,6 +50,39 @@ describe('blobKey / blobPath', () => { expect(blobKey(info())).not.toBe(blobKey(info({ format_id: '22' }))); }); + it('keys a video without its format, so a re-scrape that drifts still matches', () => { + expect(videoKey(info())).toBe(videoKey(info({ format_id: '22' }))); + expect(blobKey(info())).not.toBe(blobKey(info({ format_id: '22' }))); + }); + + it('separates two identity-less videos of one page by title', () => { + const page = (title: string, format_id: string) => + info({ + extractor: 'generic', + id: 'master', + title, + format_id, + filename: `/x/${title}.${format_id}.mp4`, + webpage_url: 'https://p', + }); + expect(videoKey(page('Clip (1)', 'a'))).not.toBe( + videoKey(page('Clip (2)', 'a')), + ); + expect(videoKey(page('Clip (1)', 'a'))).toBe(videoKey(page('Clip (1)', 'b'))); + }); + + it('separates same-titled identity-less videos by their ids', () => { + const clip = (id: string) => + info({ extractor: 'generic', id, title: 'Clip', webpage_url: 'https://p' }); + expect(videoKey(clip('v1'))).not.toBe(videoKey(clip('v2'))); + }); + + it('separates same-titled identity-less videos of different pages', () => { + const clip = (webpage_url: string) => + info({ extractor: 'generic', id: 'master', title: 'Clip', webpage_url }); + expect(videoKey(clip('https://p1'))).not.toBe(videoKey(clip('https://p2'))); + }); + it('falls back to the filename when the identity is missing', () => { const i = { filename: '/storage/x/foo.mp4', title: 'T' } as any; expect(blobKey(i)).toBe('/storage/x/foo.mp4'); diff --git a/test/db.test.ts b/test/db.test.ts index 14a0142..6012a55 100644 --- a/test/db.test.ts +++ b/test/db.test.ts @@ -47,11 +47,30 @@ describe('migrations', () => { Date.now(), ); - migrate(db); + migrate(db, 5); // stop on the backfill: the next migration empties the table const row = db .query('SELECT webpage_url FROM video_info WHERE url = ?') .get('https://alias.example') as { webpage_url: string }; expect(row.webpage_url).toBe('https://canonical.example'); }); + + it('drops cached info rows written before the payload became an array', () => { + const db = new Database(':memory:'); + migrate(db, 5); + db.query( + 'INSERT INTO video_info (url, info, webpage_url, created_at) VALUES (?, ?, ?, ?)', + ).run( + 'https://alias.example', + JSON.stringify({ webpage_url: 'https://alias.example', title: 'T' }), + 'https://alias.example', + Date.now(), + ); + + migrate(db); + + expect(db.query('SELECT COUNT(*) c FROM video_info').get()).toEqual({ + c: 0, + }); + }); }); diff --git a/test/download-video.test.ts b/test/download-video.test.ts index f02af38..71ae67f 100644 --- a/test/download-video.test.ts +++ b/test/download-video.test.ts @@ -45,10 +45,12 @@ import { classifyFailure, downloadVideo, getInfo, + getInfos, isPermanentError, liveYtdlpSize, probeDuration, removeCachedInfo, + removeCachedUrl, resetShutdown, sendInfo, sendVideo, @@ -79,7 +81,7 @@ afterAll(async () => { beforeEach(async () => { jest.clearAllMocks(); resetDb(); - getInfo.cache.clear(); + getInfos.cache.clear(); downloadVideo.cache.clear(); await rm(STUB_DIR, { recursive: true, force: true }); await mkdir(STUB_DIR, { recursive: true }); @@ -374,7 +376,7 @@ describe('getInfo', () => { const info = await getInfo(log as any, url, true); // verbose expect(info).toEqual(urlInfo); // freshly scraped, not the cached row - expect(await stubArgs()).toEndWith(`yt-dlp ${url} --verbose --dump-json`); + expect(await stubArgs()).toEndWith(`yt-dlp ${url} --verbose --dump-json --no-playlist --playlist-end 50`); }); it('scrapes and caches when not in the DB', async () => { @@ -383,9 +385,9 @@ describe('getInfo', () => { expect(info).toEqual(urlInfo); expect(appendedText()).toBe(`\u{1f9d0} Scraping ${url}...`); expect(await stubArgs()).toEndWith( - `yt-dlp ${url} --no-warnings --dump-json`, + `yt-dlp ${url} --no-warnings --dump-json --no-playlist --playlist-end 50`, ); - expect(JSON.parse(infoRow(url)!.info)).toEqual(urlInfo); + expect(JSON.parse(infoRow(url)!.info)).toEqual([urlInfo]); }); it('caches the canonical url too, so an alias request skips the scrape', async () => { @@ -398,7 +400,7 @@ describe('getInfo', () => { expect(infoCount()).toBe(2); // alias + canonical, no duplicate // a later request for the canonical hits the DB, not the scraper - getInfo.cache.clear(); // drop the in-memory memo to force a DB read + getInfos.cache.clear(); // drop the in-memory memo to force a DB read await stub({ stdout: 'not valid json: must not be scraped' }); const again = await getInfo(log as any, canon); expect(again.webpage_url).toBe(canon); @@ -413,14 +415,286 @@ describe('getInfo', () => { const info = await getInfo(log as any, url); expect(info.filename).toBe(VideoInfo.filename); // the fresh scrape - expect(infoRow(url)!.info).toBe(JSON.stringify(urlInfo)); // row refreshed + expect(infoRow(url)!.info).toBe(JSON.stringify([urlInfo])); // row refreshed + }); + + it('returns every entry of a multi-video post', async () => { + const entries = [ + { ...VideoInfo, id: 'a', webpage_url: url }, + { ...VideoInfo, id: 'b', webpage_url: url }, + { ...VideoInfo, id: 'c', webpage_url: url }, + ]; + await stub({ + stdout: entries.map((e) => `${JSON.stringify(e)}\n`).join(''), + }); + + expect(await getInfos(log as any, url)).toEqual(entries); + expect(infoCount()).toBe(1); + expect(JSON.parse(infoRow(url)!.info)).toEqual(entries); + }); + + it('serves every entry back out of the DB cache', async () => { + const entries = ['a', 'b', 'c'].map((id) => ({ + ...VideoInfo, + id, + webpage_url: url, + })); + await stub({ stdout: entries.map((e) => JSON.stringify(e)).join('\n') }); + await getInfos(log as any, url); + + getInfos.cache.clear(); + await stub({ stdout: 'not valid json: must not be scraped' }); + + expect(await getInfos(log as any, url)).toEqual(entries); + }); + + it('returns the FIRST entry from getInfo', async () => { + await stub({ + stdout: [ + JSON.stringify({ ...VideoInfo, id: 'first', webpage_url: url }), + JSON.stringify({ ...VideoInfo, id: 'second', webpage_url: url }), + ].join('\n'), + }); + + expect((await getInfo(log as any, url)).id).toBe('first'); + }); + + it('aliases a multi-entry post to the URL all its entries share', async () => { + const canon = 'https://test.invalid/post'; + const entries = ['a', 'b'].map((id) => ({ + ...VideoInfo, + id, + webpage_url: canon, + })); + await stub({ stdout: entries.map((e) => JSON.stringify(e)).join('\n') }); + + await getInfos(log as any, url); + + expect(JSON.parse(infoRow(canon)!.info)).toEqual(entries); + }); + + it('caches a salvaged scrape like any other', async () => { + const video = { ...VideoInfo, id: 'v1', webpage_url: url }; + await stub({ + exit: '1', + stdout: JSON.stringify(video), + stderr: 'ERROR: [Instagram] photo1: No video formats found!\n', + }); + await getInfos(log as any, url); + + getInfos.cache.clear(); + await stub({ exit: '0', stdout: 'not valid json: must not be scraped' }); + + expect(await getInfos(log as any, url)).toEqual([video]); + }); + + it('removeCachedUrl drops a row an entry could not name itself', async () => { + await stub({ + stdout: [ + JSON.stringify({ ...VideoInfo, id: '1', webpage_url: 'https://a' }), + JSON.stringify({ ...VideoInfo, id: '2', webpage_url: 'https://b' }), + ].join('\n'), + }); + await getInfos(log as any, url); + expect(infoCount()).toBe(1); + + removeCachedUrl(url); + + expect(infoCount()).toBe(0); + }); + + it('removeCachedInfo evicts a playlist row by the requested url', async () => { + const entry = { ...VideoInfo, id: '2', webpage_url: 'https://b' }; + await stub({ + stdout: [ + JSON.stringify({ ...VideoInfo, id: '1', webpage_url: 'https://a' }), + JSON.stringify(entry), + ].join('\n'), + }); + await getInfos(log as any, url); + + removeCachedInfo(entry, url); + + expect(infoCount()).toBe(0); + }); + + it('does not salvage a run whose items failed for any other reason', async () => { + await stub({ + exit: '1', + stdout: JSON.stringify({ ...VideoInfo, id: 'v1', webpage_url: url }), + stderr: + 'ERROR: [Instagram] photo1: No video formats found!\n' + + 'ERROR: [Instagram] v2: Requested content is not available, rate-limit reached or login required\n', + }); + + await expect(getInfos(log as any, url)).rejects.toBeInstanceOf(YtdlpError); + expect(infoCount()).toBe(0); + }); + + it('does not salvage a failure whose reason is not in the retained stderr', async () => { + await stub({ + exit: '1', + stdout: JSON.stringify({ ...VideoInfo, id: 'v1', webpage_url: url }), + stderr: 'a progress line that crowded the errors out\n', + }); + + await expect(getInfos(log as any, url)).rejects.toBeInstanceOf(YtdlpError); + expect(infoCount()).toBe(0); + }); + + it('keeps the salvage payload out of the logged error', async () => { + const e = new YtdlpError('failed', 'stderr', false, 'x'.repeat(1000)); + expect(Object.keys(e)).not.toContain('stdout'); + expect(e.stdout).toHaveLength(1000); + }); + + it('does not salvage the output of a timed-out run', async () => { + await stub({ + signal: 'TERM', + stdout: JSON.stringify({ ...VideoInfo, id: 'v1', webpage_url: url }), + // photo-item errors too, so only the signal itself can refuse the salvage + stderr: 'ERROR: [Instagram] photo1: No video formats found!\n', + }); + + const err = await getInfos(log as any, url).catch((e) => e); + expect(err).toBeInstanceOf(YtdlpError); + // else this passes on the empty-stdout guard instead of the signalled one + expect(err.stdout).not.toBe(''); + }); + + it('fills in the requested URL for an entry that carries none', async () => { + await stub({ + stdout: ['a', 'b'] + .map((id) => JSON.stringify({ ...VideoInfo, id, webpage_url: undefined })) + .join('\n'), + }); + + const entries = await getInfos(log as any, url); + + // one left unfilled makes namesAll false, so the alias row is never + // written and its blobKey loses the URL salt + expect(entries.map((e) => e.webpage_url)).toEqual([url, url]); + // a row keyed by undefined would be written per scrape: SQLite treats + // NULLs as distinct, so the cache would never hit + expect(infoCount()).toBe(1); + }); + + it('lets a shutdown abort through the salvage', async () => { + await stub({ block: '1', stdout: infoStr }); + const scrape = getInfos(log as any, url).catch((e) => e); + await waitUntil(async () => (await stubArgs()) !== ''); + + abortDownloads(); + try { + // salvaging it would burn an attempt on a job the next boot re-runs + expect((await scrape).name).toBe('ShutdownAbort'); + } finally { + resetShutdown(); + await rm(`${STUB_DIR}/block`, { force: true }); + } + }); + + it('does not alias a playlist to its first video', async () => { + const first = 'https://test.invalid/watch?v=1'; + await stub({ + stdout: [ + JSON.stringify({ ...VideoInfo, id: '1', webpage_url: first }), + JSON.stringify({ + ...VideoInfo, + id: '2', + webpage_url: 'https://test.invalid/watch?v=2', + }), + ].join('\n'), + }); + + await getInfos(log as any, url); + + expect(infoRow(first)).toBeNull(); + expect(infoCount()).toBe(1); + }); + + it('keeps the videos a partly failing scrape did resolve', async () => { + const video = { ...VideoInfo, id: 'vid', webpage_url: url }; + await stub({ + exit: '1', + stdout: JSON.stringify(video), + stderr: + 'ERROR: [Instagram] photo1: No video formats found!; please report this issue on https://github.com/yt-dlp/yt-dlp/issues?q=\n', + }); + + expect(await getInfos(log as any, url)).toEqual([video]); + }); + + it('rethrows when a failing scrape resolved nothing', async () => { + await stub({ + exit: '1', + stdout: '', + stderr: + 'ERROR: [Instagram] photo1: No video formats found!\n' + + 'ERROR: [Instagram] photo2: No video formats found!\n', + }); + + const err = await getInfos(log as any, url).catch((e) => e); + expect(err).toBeInstanceOf(YtdlpError); + expect(classifyFailure(err)).toBe('not-a-video'); + }); + + it('judges another extractor\'s no-formats run on its own lines', async () => { + await stub({ + exit: '1', + stdout: '', + stderr: + 'ERROR: [Reddit] a: No video formats found!\n' + + 'ERROR: [Reddit] b: No video formats found!\n', + }); + + const err = await getInfos(log as any, url).catch((e) => e); + expect(classifyFailure(err)).toBe('unavailable'); + }); + + it('still reports a lone post that came back with no formats', async () => { + await stub({ + exit: '1', + stdout: '', + stderr: 'ERROR: [Instagram] reel1: No video formats found!\n', + }); + + const err = await getInfos(log as any, url).catch((e) => e); + expect(classifyFailure(err)).toBe('unavailable'); + }); + + it('evicts the canonical row too when the request came in under an alias', async () => { + const canonical = 'https://host/canonical'; + const entry = { ...VideoInfo, webpage_url: canonical }; + seedInfoRow(url, entry); + seedInfoRow(canonical, entry); + + removeCachedInfo(entry as any, url); + + // the alias row alone would leave the canonical one replaying the same + // expired signed URLs for the rest of its TTL + expect(infoCount()).toBe(0); + }); + + it('throws a readable error when a clean scrape yields no video at all', async () => { + await stub({ stdout: '' }); + + await expect(getInfos(log as any, url)).rejects.toThrow( + 'yt-dlp found no video at that URL', + ); }); it('drops the proc from liveYtdlp even when the scrape throws', async () => { // the execYtdlp finally must clear the Set on every exit path; a leaked // dead proc would grow the Set unbounded and let abortDownloads kill a // stale handle. A nonzero exit throws AFTER the finally ran. - await stub({ exit: '1', stderr: 'boom' }); + await stub({ + exit: '1', + stderr: 'boom', + // an empty stdout is what makes this throw: a failing run that still + // dumped an entry gets salvaged + stdout: '', + }); await expect(getInfo(log as any, url)).rejects.toBeInstanceOf(YtdlpError); expect(liveYtdlpSize()).toBe(0); }); @@ -603,6 +877,31 @@ describe('downloadVideo', () => { 'ERROR: [generic] Unable to download webpage: HTTP Error 502: BAD GATEWAY (caused by )\n', message: 'Unable to download webpage: HTTP Error 502: BAD GATEWAY', }, + { + exit: '1', + stderr: + 'ERROR: [Instagram] Dbnd91uAyMW: No video formats found!; please report this issue on https://github.com/yt-dlp/yt-dlp/issues?q= , filling out the appropriate issue template. Confirm you are on the latest version using yt-dlp -U\n', + message: 'Dbnd91uAyMW: No video formats found!', + }, + { + exit: '1', + stderr: + "ERROR: [generic] An extractor error has occurred. (caused by KeyError('media')); please report this issue on https://github.com/yt-dlp/yt-dlp/issues?q= , filling out the appropriate issue template. Confirm you are on the latest version using yt-dlp -U\n", + message: 'An extractor error has occurred.', + }, + { + exit: '1', + stderr: + 'ERROR: [youtube] Failed to extract the player response. Please report this issue on https://github.com/yt-dlp/yt-dlp/issues?q= , filling out the appropriate issue template. Confirm you are on the latest version using yt-dlp -U\n', + message: 'Failed to extract the player response.', + }, + { + exit: '1', + stderr: + 'ERROR: [Instagram] Da4FbMds5BU: Instagram sent an empty media response. Check if this post is accessible in your browser without being logged-in. Otherwise, if the post is accessible in browser without being logged-in, please report this issue on https://github.com/yt-dlp/yt-dlp/issues?q= , filling out the appropriate issue template. Confirm you are on the latest version using yt-dlp -U\n', + message: + 'Da4FbMds5BU: Instagram sent an empty media response. Check if this post is accessible in your browser without being logged-in. Otherwise, if the post is accessible in browser without being logged-in', + }, ])( 'error messages for failures: %j', async ({ signal, exit, stderr, message }) => { @@ -777,6 +1076,39 @@ describe('isPermanentError', () => { expect(isPermanentError(new YtdlpError('failed', stderr))).toBe(false); }); + it('reports the post\'s own failure, not a photo item that follows it', () => { + const stderr = + 'ERROR: [Instagram] vid1: Requested content is not available, rate-limit reached\n' + + 'ERROR: [Instagram] photo2: No video formats found!\n' + + 'ERROR: [Instagram] photo3: No video formats found!'; + // the trailing photo lines would tell the chat the post holds no video, + // for a post that holds one the rate limit kept us from + expect(new YtdlpError('failed', stderr).message).toBe( + 'vid1: Requested content is not available, rate-limit reached', + ); + }); + + it('falls back to the photo line when photos are all a post has', () => { + const stderr = + 'ERROR: [Instagram] photo1: No video formats found!\n' + + 'ERROR: [Instagram] photo2: No video formats found!'; + expect(new YtdlpError('failed', stderr).message).toBe( + 'photo2: No video formats found!', + ); + }); + + it('drops the whole bug-report invitation, not a fragment of it', () => { + const stderr = + "ERROR: The extracted extension ('pdf') is unusual and will be skipped " + + 'for safety reasons. If you believe this is an error, please report this ' + + 'issue on https://github.com/yt-dlp/yt-dlp/issues?q= , filling out the ' + + 'appropriate issue template.'; + expect(new YtdlpError('failed', stderr).message).toBe( + "The extracted extension ('pdf') is unusual and will be skipped for " + + 'safety reasons.', + ); + }); + it('treats a signal-killed failure as retryable even if stderr looks permanent', () => { const e = new YtdlpError('Timed out', 'ERROR: Unsupported URL: x', true); expect(isPermanentError(e)).toBe(false); @@ -858,12 +1190,24 @@ describe('classifyFailure', () => { 'ERROR: Unsupported URL: https://example.com/article', 'ERROR: [Reddit] 92dd8: No media found', 'ERROR: [Instagram] DbHhjdBJT9O: There is no video in this post', + // an archive.ph page holding a PDF + "ERROR: The extracted extension ('pdf') is unusual and will be skipped for safety reasons. If you believe this is an error, please report this issue on https://github.com/yt-dlp/yt-dlp/issues?q= , filling out the appropriate issue template. Confirm you are on the latest version using yt-dlp -U", + // a photo carousel + 'ERROR: [Instagram] Dbnd83GgnNK: No video formats found!; please report this issue on https://github.com/yt-dlp/yt-dlp/issues?q= , filling out the appropriate issue template. Confirm you are on the latest version using yt-dlp -U\n' + + 'ERROR: [Instagram] Dbnd91uAyMW: No video formats found!; please report this issue on https://github.com/yt-dlp/yt-dlp/issues?q= , filling out the appropriate issue template. Confirm you are on the latest version using yt-dlp -U', ])('classifies %j as not-a-video', (stderr) => { expect(classifyFailure(new YtdlpError('failed', stderr))).toBe( 'not-a-video', ); }); + it('judges a carousel on its non-photo items, so one bad item stays retryable', () => { + const stderr = + 'ERROR: [Instagram] photo1: No video formats found!\n' + + 'ERROR: [Instagram] vid2: Requested content is not available, rate-limit reached or login required'; + expect(classifyFailure(new YtdlpError('failed', stderr))).toBe('transient'); + }); + it('keeps the f4m downloader\'s untagged "No media found" retryable', () => { expect( classifyFailure(new YtdlpError('failed', 'ERROR: No media found')), @@ -873,6 +1217,7 @@ describe('classifyFailure', () => { it.each([ 'ERROR: Private video. Sign in if you have access', 'ERROR: Unable to download webpage: HTTP Error 410: Gone', + 'ERROR: [youtube] dQw4w9WgXcQ: No video formats found!; please report this issue', ])('classifies %j as unavailable', (stderr) => { expect(classifyFailure(new YtdlpError('failed', stderr))).toBe('unavailable'); }); diff --git a/test/e2e.test.ts b/test/e2e.test.ts index fc2544e..4c57e4b 100644 --- a/test/e2e.test.ts +++ b/test/e2e.test.ts @@ -10,7 +10,7 @@ import { } from 'bun:test'; import { blobPath, recordBlob } from '../src/blob-store'; import { resetDb } from '../src/db'; -import { downloadVideo, getInfo } from '../src/download-video'; +import { downloadVideo, getInfos } from '../src/download-video'; import { jobsIdle, seedJob, setRetryBaseMs } from '../src/job-queue'; import { FORMAT_ID_RE, MOCK_USER_ID, withBotApi } from './simulate-bot-api'; import { rowCount, spyMock, waitUntil } from './test-utils'; @@ -80,7 +80,7 @@ const scrub = (messages: unknown) => ); const clearInMemoryCache = () => { - getInfo.cache.clear(); + getInfos.cache.clear(); downloadVideo.cache.clear(); }; @@ -131,15 +131,17 @@ describe.if(!!Bun.env.TEST_E2E)('message handler', async () => { const groupChat = { id: -1000000000001, title: 'Test Group', type: 'supergroup' }; - it( - 'stays silent for a not-a-video link in a group', - () => + it.each([ + // a lone photo + 'https://www.instagram.com/p/DbHhjdBJT9O/', + // a photo carousel (see PHOTO_ITEM) + 'https://www.instagram.com/p/Dbnd-yeAP9S/', + ])( + 'stays silent for a not-a-video link in a group: %s', + (url) => withBotApi(async (api) => { clearInMemoryCache(); - api.sendTextMessageToBot( - urlMessage('https://www.instagram.com/p/DbHhjdBJT9O/'), - groupChat, - ); + api.sendTextMessageToBot(urlMessage(url), groupChat); // gate on the job actually starting, else a never-started job passes this // silence test vacuously expect(await waitUntil(() => !jobsIdle(), 15_000)).toBe(true); @@ -151,8 +153,8 @@ describe.if(!!Bun.env.TEST_E2E)('message handler', async () => { const reports = api.sentMessages.filter(isFailureReport); expect(reports).toHaveLength(1); // pin the tolerated report to rate-limit/login wording: a broken - // not-a-video gate would instead report "There is no video in this - // post" and must still fail this test + // not-a-video gate would instead report a no-video wording and must + // still fail this test expect(reports[0]!.text).toMatch(/rate.?limit|login required|empty media response/i); } else { expect(api.sentMessages).toEqual([]); @@ -161,6 +163,33 @@ describe.if(!!Bun.env.TEST_E2E)('message handler', async () => { 45_000, ); + it.if(!!Bun.env.TEST_E2E_FULL)( + 'delivers every video of a post that mixes photos and videos', + () => + withBotApi(async (api) => { + clearInMemoryCache(); + const videos = () => api.sentMessages.filter((m: any) => m.video); + api.sendTextMessageToBot( + // a post holding three videos and a photo + urlMessage('https://www.instagram.com/p/DIqghhpok2K/'), + ); + await waitUntil( + () => videos().length >= 3 || api.sentMessages.some(isFailureReport), + 120_000, + ); + const reports = api.sentMessages.filter(isFailureReport); + // tolerate a rate-limited scrape the way the group-silence test does + if (reports.length) { + expect(reports[0]!.text).toMatch( + /rate.?limit|login required|empty media response/i, + ); + } else { + expect(videos()).toHaveLength(3); + } + }), + 150_000, + ); + it( 'reports one šŸ’„ for a whitelisted failing link in a group', () => diff --git a/test/handlers.test.ts b/test/handlers.test.ts index d54b428..a6386bf 100644 --- a/test/handlers.test.ts +++ b/test/handlers.test.ts @@ -28,6 +28,7 @@ import { createMockMessageCtx, memoize, rowCount, + seedInfoRow, spyMock, telegramError, } from './test-utils.ts'; @@ -44,12 +45,35 @@ spyMock(console, 'debug'); // suppress debug logs // guard: nothing here should hit the real filesystem unlink spyOn(fsPromises, 'unlink').mockResolvedValue(undefined); +// a real flush posts the thread, which is what gives it a message id +const landThread = async () => { + mockLog.messageId = 4242; +}; const mockLog = { append: mock(), - flush: mock(), - messageId: 4242, + flush: mock(landThread), + messageId: 4242 as number | undefined, text: 'prior log content', }; +const freshThread = async (fn: () => Promise) => { + mockLog.messageId = undefined; + try { + return await fn(); + } finally { + mockLog.messageId = 4242; + } +}; +// a thread whose every send failed: flushing it leaves it without an id +const deadThread = async (fn: () => Promise) => { + mockLog.messageId = undefined; + mockLog.flush.mockImplementation(async () => {}); + try { + return await fn(); + } finally { + mockLog.flush.mockImplementation(landThread); + mockLog.messageId = 4242; + } +}; spyOn(logMessage, 'LogMessage').mockReturnValue(mockLog as never); // logFor constructs LogMessage through log-message's module-internal binding, // which the constructor spy above can't reach: mock it directly, mirroring @@ -144,6 +168,10 @@ const mockGetInfo = spyOn(downloadVideo, 'getInfo').mockImplementation( ), ); +const mockGetInfos = spyOn(downloadVideo, 'getInfos').mockImplementation( + async (log, url, verbose) => [await downloadVideo.getInfo(log, url, verbose)], +); + const mockSendInfo = spyMock(downloadVideo, 'sendInfo'); const mockDownloadVideo = spyOn( downloadVideo, @@ -178,8 +206,10 @@ describe.each([false, true])('textMessageHandler, edit: %p', (isEdit) => { verbose: false, // the mock ran the job inline, and processUrlJob mutates its job // (the mutation is what persists across retries); the recorded call - // arg is that same object, so the flag shows here - infoShown: true, + // arg is that same object, so these show here + announcedIds: ['test:id'], + answered: true, + settledIds: ['test:id'], }, expect.any(Function), // the handled-urls record, run inside the tx ); @@ -500,7 +530,7 @@ describe('inlineQueryHandler', () => { it('handles errors gracefully and shows error to user', async () => { const ctx = createMockInlineQueryCtx(); - mockGetInfo.mockRejectedValue(new Error('fail!')); + mockGetInfo.mockRejectedValueOnce(new Error('fail!')); const mockError = spyOn(console, 'error').mockImplementationOnce(() => {}); await inlineQueryHandler(ctx as any); @@ -515,6 +545,20 @@ describe('inlineQueryHandler', () => { expect(mockError).toHaveBeenCalledTimes(1); }); + it('still answers when the download fails after the scrape resolved', async () => { + const ctx = createMockInlineQueryCtx(); + spyOn(console, 'error').mockImplementationOnce(() => {}); + mockDownloadVideo.mockRejectedValueOnce( + new downloadVideo.YtdlpError('failed', 'ERROR: HTTP Error 403'), + ); + + await inlineQueryHandler(ctx as any); + + expect(ctx.answerInlineQuery).toHaveBeenCalledWith([ + expect.objectContaining({ title: 'Failed to process video' }), + ]); + }); + it('shows a sensible message when the inline failure is not an Error', async () => { const ctx = createMockInlineQueryCtx(); mockGetInfo.mockRejectedValueOnce('boom'); @@ -1164,7 +1208,7 @@ describe('post-download duration check', () => { const confirmedJob = (overrides: Record = {}) => ({ kind: 'confirmed', - info: { filename: 'v.mp4', title: 'T', webpage_url: 'u' }, + info: { filename: 'v.mp4', title: 'T', webpage_url: 'u', extractor: 'test', id: 'id' }, verbose: false, messageId: 7, chatId: 7, @@ -1183,6 +1227,8 @@ describe('confirmed job stale-info refresh', () => { filename: 'v.mp4', title: 'Old Snapshot', webpage_url: 'https://example.com', + extractor: 'test', + id: 'id', }, }); await processJob({} as any, job, 1); @@ -1212,11 +1258,27 @@ describe('confirmed job stale-info refresh', () => { }); describe('confirmed job oversize report', () => { + it('un-records a lone confirmed video that turns out too large', async () => { + const job = confirmedJob({ url: 'https://example.com/lone' }); + db.query( + 'INSERT INTO handled_urls (chat_id, message_id, url, created_at) VALUES (?, ?, ?, ?)', + ).run(job.chatId, job.messageId, job.url, Date.now()); + mockSendVideo.mockResolvedValueOnce(undefined as any); + + await processJob({} as any, job, 1); + + expect(rowCount('handled_urls')).toBe(0); + }); + it('reports too-large (not silently) when real bytes overshoot the estimate', async () => { // the real bytes overshoot a missing/under estimate, so sendVideo returns // undefined; verify the report routes around the silent progress NoLog mockSendVideo.mockResolvedValueOnce(undefined as any); - const job = confirmedJob({ chatId: -100 }); + // a format the refresh moves off, so the drift guard actually fires + const job = confirmedJob({ + chatId: -100, + info: { ...confirmedJob().info, format_id: 'parked' }, + }); await expect(processJob({} as any, job, 1)).resolves.toBeUndefined(); @@ -1224,14 +1286,11 @@ describe('confirmed job oversize report', () => { expect(mockLog.append).toHaveBeenCalledWith( expect.stringContaining('Video too large'), ); - // sendVideo already discarded the drifted info's oversize bytes; the only - // release is the drift guard freeing the parked (pre-refresh) job.info - // identity, whose key the getInfo refresh drifted away from. It has no - // recorded blob here, so the release is a harmless no-op, but never the - // drifted `info` sendVideo already handled (no double-release of that). - for (const [released] of mockReleaseBlob.mock.calls) { - expect((released as any).webpage_url).toBe(job.info.webpage_url); - } + // sendVideo already discarded the drifted info's oversize bytes; this + // release is the drift guard freeing the parked (pre-refresh) identity + expect(mockReleaseBlob.mock.calls.map(([i]: any) => i.format_id)).toEqual([ + 'parked', + ]); }); }); @@ -1254,24 +1313,24 @@ describe('job retry classification', () => { }); it('re-prints the info block only when the delivered thread lacks it', async () => { - // died during the scrape: infoShown never set ā†’ info must print, and the - // flag is set for persistence (the retry bump re-serializes the job) - const j1 = { ...urlJob, logText: 'šŸ§ Scraping x...' }; + // died during the scrape: nothing announced ā†’ info must print, and the + // key is recorded for persistence (the retry bump re-serializes the job) + const j1 = { ...urlJob, url: 'https://example.com/i1', logText: 'šŸ§ Scraping x...' }; await processJob({} as any, j1 as any, 2); expect(mockSendInfo).toHaveBeenCalledTimes(1); - expect((j1 as any).infoShown).toBe(true); + expect((j1 as any).announcedIds).toEqual(['test:id']); jest.clearAllMocks(); - // shown AND delivered (a thread exists): skip, even though the info text - // may sit in an earlier chunk than the stashed last one - const j2 = { ...urlJob, logMessageId: 4242, logText: 'x', infoShown: true }; + // announced AND delivered (a thread exists): skip, even though the info + // text may sit in an earlier chunk than the stashed last one + const j2 = { ...urlJob, url: 'https://example.com/i2', logMessageId: 4242, logText: 'x', announcedIds: ['test:id'] }; await processJob({} as any, j2 as any, 2); expect(mockSendInfo).not.toHaveBeenCalled(); jest.clearAllMocks(); // appended but NEVER delivered (every send failed, so no thread was // stashed): the retry posts a fresh thread, which needs the info again - const j3 = { ...urlJob, logMessageId: undefined, infoShown: true }; + const j3 = { ...urlJob, url: 'https://example.com/i3', logMessageId: undefined, announcedIds: ['test:id'] }; await processJob({} as any, j3 as any, 2); expect(mockSendInfo).toHaveBeenCalledTimes(1); @@ -1637,3 +1696,842 @@ describe('group terminal-failure feedback (issues #14/#17)', () => { expectGroupSilent(ctx); }); }); + +describe('multi-video posts (carousels)', () => { + // clearAllMocks does not reset implementations, and this override would + // otherwise follow every test appended after this block + afterAll(() => + mockGetInfos.mockImplementation(async (log, url, verbose) => [ + await downloadVideo.getInfo(log, url, verbose), + ]), + ); + const post = 'https://www.instagram.com/p/carousel'; + const entries = ['a', 'b', 'c'].map((id) => ({ + webpage_url: post, + title: `Video ${id}`, + extractor: 'Instagram', + id, + filename: `${id}.mp4`, + })); + + const carousel = () => + mockGetInfos.mockImplementation(async () => entries as any); + + const sentIds = () => + mockSendVideo.mock.calls.map(([, , info]: any) => info.id); + + it('delivers every video of the post, in order', async () => { + carousel(); + await processJob({} as any, { ...urlJob, url: post } as any, 1); + expect(sentIds()).toEqual(['a', 'b', 'c']); + }); + + it('remembers what it sent, so a retry resumes instead of re-sending', async () => { + spyMock(console, 'error'); + carousel(); + mockSendVideo.mockImplementationOnce(async () => ({ + video: { file_id: 'id' }, + })).mockImplementationOnce(async () => { + throw new Error('transient'); + }); + const job = { ...urlJob, url: post } as any; + + await processJob({} as any, job, 1).catch(() => {}); + await processJob({} as any, job, 2); + + expect(sentIds().filter((id: string) => id === 'a')).toEqual(['a']); + expect(sentIds().filter((id: string) => id === 'c')).toEqual(['c']); + // marking an entry before its outcome is known would skip it here instead + expect(sentIds().filter((id: string) => id === 'b')).toEqual(['b', 'b']); + }); + + it('parks a long video and carries on with the rest of the post', async () => { + mockGetInfos.mockImplementation(async () => [ + { ...entries[0], duration: 30 * 60 }, + entries[1], + entries[2], + ] as any); + const groupJob = { ...urlJob, url: post, chatId: -100, chatType: 'group' }; + const tg = { sendMessage: mock(async () => ({ message_id: 1 })) } as any; + + await processJob(tg, groupJob as any, 1); + await processJob(tg, groupJob as any, 2); + + // re-parking a settled video would prompt twice + expect(tg.sendMessage).toHaveBeenCalledTimes(1); + expect(sentIds()).toEqual(['b', 'c']); + }); + + it('counts a parked confirmation as part of the post having landed', async () => { + spyMock(console, 'error'); + mockGetInfos.mockImplementation(async () => [ + { ...entries[0], duration: 30 * 60 }, + entries[1], + ] as any); + db.query( + 'INSERT INTO handled_urls (chat_id, message_id, url, created_at) VALUES (?, ?, ?, ?)', + ).run(-100, urlJob.messageId, post, Date.now()); + mockDownloadVideo.mockRejectedValue( + new downloadVideo.YtdlpError('failed', 'ERROR: Video unavailable'), + ); + const tg = { sendMessage: mock(async () => ({ message_id: 1 })) } as any; + try { + await processJob( + tg, + { ...urlJob, url: post, chatId: -100, chatType: 'group' } as any, + jobQueue.MAX_ATTEMPTS, + ); + + // un-recording would let an edit re-park the confirmation + expect(rowCount('handled_urls')).toBe(1); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('announces the cap once, not again on every attempt', async () => { + spyMock(console, 'error'); + const many = Array.from({ length: 25 }, (_, i) => ({ + ...entries[0], + id: `v${i}`, + })); + mockGetInfos.mockImplementation(async () => many as any); + mockSendVideo.mockImplementationOnce(async () => { + throw new Error('transient'); + }); + const job = { ...urlJob, url: post, logMessageId: 7 } as any; + + await processJob({} as any, job, 1).catch(() => {}); + await processJob({} as any, job, 2); + + expect( + mockLog.append.mock.calls.filter(([s]: any) => + String(s).includes('videos here'), + ), + ).toHaveLength(1); + }); + + it('sends a confirmed video with no identity after its format drifted', async () => { + const parked = { + webpage_url: post, + title: 'Clip', + extractor: 'generic', + id: 'master', + format_id: 'hls-1080', + filename: 'Clip.hls-1080.mp4', + }; + mockGetInfos.mockImplementation(async () => [ + { ...parked, format_id: 'hls-1200', filename: 'Clip.hls-1200.mp4' }, + ] as any); + + await processJob({} as any, confirmedJob({ info: parked }), 1); + + expect( + mockSendVideo.mock.calls.map(([, , i]: any) => i.filename), + ).toEqual(['Clip.hls-1200.mp4']); + }); + + it('does not re-send an identity-less video whose format drifted on the retry', async () => { + spyMock(console, 'error'); + const page = (fmt: string) => + ['Clip (1)', 'Clip (2)'].map((title) => ({ + webpage_url: post, + title, + extractor: 'generic', + id: 'master', + format_id: fmt, + filename: `${title}.${fmt}.mp4`, + })); + mockGetInfos.mockImplementation(async () => page('hls-1080') as any); + mockDownloadVideo.mockImplementationOnce(async () => 'downloaded'); + mockDownloadVideo.mockImplementationOnce(async () => { + throw new downloadVideo.YtdlpError('failed', 'ERROR: HTTP Error 503'); + }); + const job = { ...urlJob, url: post } as any; + try { + await processJob({} as any, job, 1).catch(() => {}); + mockGetInfos.mockImplementation(async () => page('hls-1200') as any); + await processJob({} as any, job, 2); + + expect( + mockSendVideo.mock.calls.map(([, , i]: any) => i.title), + ).toEqual(['Clip (1)', 'Clip (2)']); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('tells two videos apart when the extractor gives them the same id', async () => { + mockGetInfos.mockImplementation( + async () => + [ + { + webpage_url: post, + title: 'Clip (1)', + extractor: 'generic', + id: 'master', + filename: 'one.mp4', + }, + { + webpage_url: post, + title: 'Clip (2)', + extractor: 'generic', + id: 'master', + filename: 'two.mp4', + }, + ] as any, + ); + + await processJob({} as any, { ...urlJob, url: post } as any, 1); + + expect(mockSendVideo.mock.calls.map(([, , i]: any) => i.filename)).toEqual([ + 'one.mp4', + 'two.mp4', + ]); + }); + + it('delivers the rest of the post when one video fails', async () => { + spyMock(console, 'error'); + carousel(); + mockDownloadVideo.mockImplementation(async (_log: any, info: any) => { + if (info.id === 'b') { + throw new downloadVideo.YtdlpError( + 'failed', + 'ERROR: Unsupported URL: https://x', + ); + } + return 'downloaded'; + }); + try { + await processJob( + {} as any, + { ...urlJob, url: post } as any, + jobQueue.MAX_ATTEMPTS, + ); + expect(sentIds()).toEqual(['a', 'c']); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('releases the blob of the video that failed, not whichever came last', async () => { + spyMock(console, 'error'); + carousel(); + mockDownloadVideo.mockImplementation(async (_log: any, info: any) => { + if (info.id === 'a') { + throw new downloadVideo.YtdlpError('failed', 'ERROR: Video unavailable'); + } + return 'downloaded'; + }); + try { + await processJob( + {} as any, + { ...urlJob, url: post } as any, + jobQueue.MAX_ATTEMPTS, + ); + expect(mockReleaseBlob.mock.calls.map(([i]: any) => i.id)).toEqual(['a']); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('reports a rejection that carries no value at all', async () => { + spyMock(console, 'error'); + carousel(); + mockDownloadVideo.mockRejectedValueOnce(undefined); + try { + await expect( + processJob({} as any, { ...urlJob, url: post } as any, 1), + ).rejects.toBeUndefined(); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('comes back for a video that could still succeed, whatever failed last', async () => { + spyMock(console, 'error'); + carousel(); + mockDownloadVideo.mockImplementation(async (_log: any, info: any) => { + if (info.id === 'a') { + throw new downloadVideo.YtdlpError( + 'failed', + 'ERROR: Unable to download webpage: HTTP Error 503', + ); + } + if (info.id === 'b') { + throw new downloadVideo.YtdlpError('failed', 'ERROR: Video unavailable'); + } + return 'downloaded'; + }); + try { + // keeping the permanent failure would drop 'a' with no retry + await expect( + processJob({} as any, { ...urlJob, url: post } as any, 1), + ).rejects.toBeDefined(); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('counts a post-download prompt as part of the post having landed', async () => { + spyMock(console, 'error'); + mockGetInfos.mockImplementation(async () => [entries[0], entries[1]] as any); + db.query( + 'INSERT INTO handled_urls (chat_id, message_id, url, created_at) VALUES (?, ?, ?, ?)', + ).run(-100, urlJob.messageId, post, Date.now()); + mockDownloadVideo.mockImplementation(async (_log: any, info: any) => { + if (info.id === 'b') { + throw new downloadVideo.YtdlpError('failed', 'ERROR: Video unavailable'); + } + // no scraped duration: only the probe after this download crosses the + // gate, so it is the post-download prompt that fires + blobStore.recordBlob(info); + blobStore.setBlobDuration(info, 30 * 60); + return 'downloaded'; + }); + const tg = { sendMessage: mock(async () => ({ message_id: 1 })) } as any; + try { + await processJob( + tg, + { ...urlJob, url: post, chatId: -100, chatType: 'group' } as any, + jobQueue.MAX_ATTEMPTS, + ); + + expect(tg.sendMessage).toHaveBeenCalled(); + expect(rowCount('handled_urls')).toBe(1); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('comes back for a confirmed entry once the post lists it again', async () => { + spyMock(console, 'error'); + mockGetInfos.mockImplementation(async () => [entries[0]] as any); + const job = confirmedJob({ info: entries[2], url: post, partOfPost: true }); + + // attempt 1, not the last: the miss has to be retryable, or the eviction + // above it buys nothing + await expect(processJob({} as any, job, 1)).rejects.toBeDefined(); + + mockGetInfos.mockImplementation(async () => entries as any); + await processJob({} as any, job, 2); + + expect(sentIds()).toEqual(['c']); + }); + + it('evicts a post row a confirmed entry cannot name itself', async () => { + spyMock(console, 'error'); + const parked = { ...entries[2], webpage_url: post }; + mockGetInfos.mockImplementation(async () => [ + { ...parked, webpage_url: 'https://c' }, + ] as any); + db.query( + 'INSERT INTO video_info (url, info, webpage_url, created_at) VALUES (?, ?, ?, ?)', + ).run(post, '[]', 'https://elsewhere', Date.now()); + mockDownloadVideo.mockRejectedValue( + new downloadVideo.YtdlpError('failed', 'ERROR: Video unavailable'), + ); + try { + await processJob( + {} as any, + confirmedJob({ info: parked, url: post }), + jobQueue.MAX_ATTEMPTS, + ); + expect(rowCount('video_info')).toBe(0); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('keeps the message recorded when a confirmed entry of a post fails', async () => { + spyMock(console, 'error'); + const job = confirmedJob({ info: entries[2], url: post, partOfPost: true }); + db.query( + 'INSERT INTO handled_urls (chat_id, message_id, url, created_at) VALUES (?, ?, ?, ?)', + ).run(job.chatId, job.messageId, post, Date.now()); + mockGetInfos.mockImplementation(async () => entries as any); + mockDownloadVideo.mockRejectedValue( + new downloadVideo.YtdlpError('failed', 'ERROR: Video unavailable'), + ); + try { + await processJob({} as any, job, jobQueue.MAX_ATTEMPTS); + + expect(rowCount('handled_urls')).toBe(1); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('keeps the message recorded when a confirmed entry of a post is too large', async () => { + mockGetInfos.mockImplementation(async () => entries as any); + const job = confirmedJob({ info: entries[2], url: post, partOfPost: true }); + db.query( + 'INSERT INTO handled_urls (chat_id, message_id, url, created_at) VALUES (?, ?, ?, ?)', + ).run(job.chatId, job.messageId, post, Date.now()); + mockSendVideo.mockResolvedValueOnce(undefined as any); + + await processJob({} as any, job, 1); + + expect(rowCount('handled_urls')).toBe(1); + }); + + it('releases every video that downloaded bytes and then failed', async () => { + spyMock(console, 'error'); + carousel(); + mockSendVideo.mockRejectedValue(telegramError(403, 'Forbidden')); + try { + await processJob({} as any, { ...urlJob, url: post } as any, 1); + expect(mockReleaseBlob.mock.calls.map(([i]: any) => i.id)).toEqual([ + 'a', + 'b', + 'c', + ]); + } finally { + mockSendVideo.mockResolvedValue({ video: { file_id: 'id' } } as any); + } + }); + + it('comes back for a video that could still succeed, whatever failed first', async () => { + spyMock(console, 'error'); + carousel(); + mockDownloadVideo.mockImplementation(async (_log: any, info: any) => { + if (info.id === 'a') { + throw new downloadVideo.YtdlpError('failed', 'ERROR: Video unavailable'); + } + if (info.id === 'b') { + throw new downloadVideo.YtdlpError( + 'failed', + 'ERROR: Unable to download webpage: HTTP Error 503', + ); + } + return 'downloaded'; + }); + try { + await expect( + processJob({} as any, { ...urlJob, url: post } as any, 1), + ).rejects.toBeDefined(); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('keeps the message recorded across attempts once a video has landed', async () => { + spyMock(console, 'error'); + carousel(); + db.query( + 'INSERT INTO handled_urls (chat_id, message_id, url, created_at) VALUES (?, ?, ?, ?)', + ).run(urlJob.chatId, urlJob.messageId, post, Date.now()); + mockDownloadVideo.mockImplementation(async (_log: any, info: any) => { + if (info.id !== 'a') { + throw new downloadVideo.YtdlpError('failed', 'ERROR: Video unavailable'); + } + return 'downloaded'; + }); + const job = { ...urlJob, url: post } as any; + try { + await processJob({} as any, job, 1).catch(() => {}); + await processJob({} as any, job, jobQueue.MAX_ATTEMPTS); + + expect(rowCount('handled_urls')).toBe(1); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('leaves the message recorded once part of the post has landed', async () => { + spyMock(console, 'error'); + carousel(); + db.query( + 'INSERT INTO handled_urls (chat_id, message_id, url, created_at) VALUES (?, ?, ?, ?)', + ).run(urlJob.chatId, urlJob.messageId, post, Date.now()); + mockDownloadVideo.mockImplementation(async (_log: any, info: any) => { + if (info.id === 'c') { + throw new downloadVideo.YtdlpError( + 'failed', + 'ERROR: Unsupported URL: https://x', + ); + } + return 'downloaded'; + }); + try { + await processJob( + {} as any, + { ...urlJob, url: post } as any, + jobQueue.MAX_ATTEMPTS, + ); + + // un-recording here would re-send 'a' and 'b' on the next edit + expect(rowCount('handled_urls')).toBe(1); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('keeps a too-large verdict owed when the thread carrying it never landed', async () => { + spyMock(console, 'error'); + mockGetInfos.mockImplementation(async () => [ + { ...entries[0], filesize: 3_000_000_000 }, + entries[1], + ] as any); + await deadThread(async () => { + mockSendVideo.mockImplementationOnce(async () => { + throw new Error('transient'); + }); + const job = { ...urlJob, url: post } as any; + + await processJob({} as any, job, 1).catch(() => {}); + + // marking it would make the retry skip the video without ever having + // told the user why + expect(job.settledIds ?? []).not.toContain('Instagram:a'); + }); + }); + + it('says a video is too large once, not again on every attempt', async () => { + spyMock(console, 'error'); + mockGetInfos.mockImplementation(async () => [ + { ...entries[0], filesize: 3_000_000_000 }, + entries[1], + ] as any); + await freshThread(async () => { + mockSendVideo.mockImplementationOnce(async () => { + throw new Error('transient'); + }); + const job = { ...urlJob, url: post } as any; + + await processJob({} as any, job, 1).catch(() => {}); + await processJob({} as any, job, 2); + + expect( + mockLog.append.mock.calls.filter(([s]: any) => + String(s).includes('too large'), + ), + ).toHaveLength(1); + }); + }); + + it('owes nothing more for a video the real bytes made too large', async () => { + mockGetInfos.mockImplementation(async () => [entries[0], entries[1]] as any); + await freshThread(async () => { + mockSendVideo.mockResolvedValueOnce(undefined as any); + const job = { ...urlJob, url: post } as any; + + await processJob({} as any, job, 1); + + // leaving it unmarked would re-download the same oversized video, and + // repeat the verdict, on every later attempt + expect(job.settledIds).toContain('Instagram:a'); + }); + }); + + it('owes nothing more for an oversized video a group never hears about', async () => { + mockGetInfos.mockImplementation(async () => [entries[0], entries[1]] as any); + mockSendVideo.mockResolvedValueOnce(undefined as any); + const job = { ...urlJob, url: post, chatId: -100, chatType: 'group' } as any; + + await processJob({} as any, job, 1); + + expect(job.settledIds).toContain('Instagram:a'); + }); + + it('keeps an oversized video owed when the thread carrying its verdict never landed', async () => { + mockGetInfos.mockImplementation(async () => [entries[0], entries[1]] as any); + await deadThread(async () => { + mockSendVideo.mockResolvedValueOnce(undefined as any); + const job = { ...urlJob, url: post } as any; + + await processJob({} as any, job, 1); + + expect(job.settledIds ?? []).not.toContain('Instagram:a'); + }); + }); + + it('tags a post-download prompt as part of the post as well', async () => { + spyMock(console, 'error'); + mockGetInfos.mockImplementation(async () => [entries[0], entries[1]] as any); + mockDownloadVideo.mockImplementation(async (_log: any, info: any) => { + blobStore.recordBlob(info); + blobStore.setBlobDuration(info, 30 * 60); + return 'downloaded'; + }); + const tg = { sendMessage: mock(async () => ({ message_id: 1 })) } as any; + try { + await processJob( + tg, + { ...urlJob, url: post, chatId: -100, chatType: 'group' } as any, + 1, + ); + + const data: string = + tg.sendMessage.mock.calls[0][2].reply_markup.inline_keyboard[0][0] + .callback_data; + const parked = await pendingDownloads.getPending( + data.slice('dl:'.length), + ); + expect(parked!.partOfPost).toBe(true); + + db.query('DELETE FROM pending').run(); + // an entry the run above never downloaded, so its duration is again + // knowable only after the probe + mockGetInfos.mockImplementation(async () => [entries[2]] as any); + await processJob( + tg, + { ...urlJob, url: post, chatId: -100, chatType: 'group', messageId: 42 } as any, + 1, + ); + + const lone: string = + tg.sendMessage.mock.calls.at(-1)![2].reply_markup.inline_keyboard[0][0] + .callback_data; + expect( + (await pendingDownloads.getPending(lone.slice('dl:'.length)))! + .partOfPost, + ).toBeUndefined(); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('remembers a video it sent even when the thread never landed', async () => { + spyMock(console, 'error'); + mockGetInfos.mockImplementation(async () => [entries[0], entries[1]] as any); + await deadThread(async () => { + mockSendVideo.mockImplementationOnce(async () => { + throw new Error('transient'); + }); + const job = { ...urlJob, url: post } as any; + + await processJob({} as any, job, 1).catch(() => {}); + + // the video went out on its own Telegram call, not through the log thread + expect(job.settledIds).toContain('Instagram:b'); + }); + }); + + it('parks a confirmation tagged with whether it is part of a post', async () => { + const long = { ...entries[0], duration: 30 * 60 }; + const groupJob = () => + ({ ...urlJob, url: post, chatId: -100, chatType: 'group' }) as any; + const tg = { sendMessage: mock(async () => ({ message_id: 1 })) } as any; + const parked = async (call: number) => { + const data: string = + tg.sendMessage.mock.calls[call][2].reply_markup.inline_keyboard[0][0] + .callback_data; + return await pendingDownloads.getPending(data.slice('dl:'.length)); + }; + + mockGetInfos.mockImplementation(async () => [long, entries[1]] as any); + await processJob(tg, groupJob(), 1); + expect((await parked(0))!.partOfPost).toBe(true); + + mockGetInfos.mockImplementation(async () => [long] as any); + await processJob(tg, { ...groupJob(), messageId: 42 }, 1); + expect((await parked(1))!.partOfPost).toBeUndefined(); + }); + + it('announces the videos a retry reaches for the first time', async () => { + spyMock(console, 'error'); + carousel(); + // a shutdown mid-post: 'b' is announced and aborted, 'c' is never reached + mockDownloadVideo.mockImplementationOnce(async () => 'downloaded'); + mockDownloadVideo.mockImplementationOnce(async () => { + throw new jobQueue.ShutdownAbort(); + }); + const job = { ...urlJob, url: post, logMessageId: 9 } as any; + try { + await processJob({} as any, job, 1).catch(() => {}); + jest.clearAllMocks(); + await processJob({} as any, job, 2); + + // only 'c' is new to the chat + expect(mockSendInfo).toHaveBeenCalledTimes(1); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('evicts the cached post when any entry failed in yt-dlp', async () => { + spyMock(console, 'error'); + mockGetInfos.mockImplementation(async () => [entries[0], entries[1]] as any); + seedInfoRow(post, entries[0]); + // the send failure outranks the download one, so only the loop knows the + // scrape went stale + mockSendVideo.mockImplementationOnce(async () => { + throw new Error('transient'); + }); + mockDownloadVideo.mockImplementation(async (_log: any, info: any) => { + if (info.id === 'b') { + throw new downloadVideo.YtdlpError( + 'failed', + 'ERROR: unable to download video data: HTTP Error 403: Forbidden', + ); + } + return 'downloaded'; + }); + try { + await processJob({} as any, { ...urlJob, url: post } as any, 1).catch( + () => {}, + ); + + // keeping it makes every later attempt replay the same expired URLs + expect(rowCount('video_info')).toBe(0); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('reports the sibling that really failed, not the item with no video', async () => { + spyMock(console, 'error'); + mockGetInfos.mockImplementation(async () => [entries[0], entries[1]] as any); + mockDownloadVideo.mockImplementation(async (_log: any, info: any) => { + throw new downloadVideo.YtdlpError( + 'failed', + info.id === 'a' + ? 'ERROR: [Instagram] x: There is no video in this post' + : 'ERROR: Video unavailable', + ); + }); + try { + await processJob( + {} as any, + { ...urlJob, url: post, chatId: -100, chatType: 'group' } as any, + jobQueue.MAX_ATTEMPTS, + ); + + // letting 'a' speak for the post would leave the group with silence for + // a video that failed for a reason worth hearing + expect( + (logMessage.LogMessage as any).mock.calls.filter( + ([, dest]: any) => dest?.replyTo === urlJob.messageId, + ), + ).toHaveLength(1); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('answers a group with one report however many videos fail', async () => { + spyMock(console, 'error'); + carousel(); + mockDownloadVideo.mockRejectedValue( + new downloadVideo.YtdlpError( + 'failed', + 'ERROR: Unable to download webpage: HTTP Error 403: Forbidden', + ), + ); + try { + await processJob( + {} as any, + { ...urlJob, url: post, chatId: -100, chatType: 'group' } as any, + jobQueue.MAX_ATTEMPTS, + ); + + expect( + (logMessage.LogMessage as any).mock.calls.filter( + ([, dest]: any) => dest?.replyTo === urlJob.messageId, + ), + ).toHaveLength(1); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); + + it('prints an info block for each video it sends', async () => { + carousel(); + await processJob({} as any, { ...urlJob, url: post } as any, 1); + expect(mockSendInfo).toHaveBeenCalledTimes(3); + }); + + it('caps a playlist-sized post at ten videos and says how many it skipped', async () => { + const many = Array.from({ length: 25 }, (_, i) => ({ + ...entries[0], + id: `v${i}`, + })); + mockGetInfos.mockImplementation(async () => many as any); + + await processJob({} as any, { ...urlJob, url: post } as any, 1); + + expect(sentIds()).toHaveLength(10); + expect(mockLog.append).toHaveBeenCalledWith( + '\nšŸ“š More than 10 videos here; sending the first 10.', + ); + }); + + it('re-announces the cap on a fresh thread the failed attempt never posted', async () => { + const many = Array.from({ length: 25 }, (_, i) => ({ + ...entries[0], + id: `v${i}`, + })); + mockGetInfos.mockImplementation(async () => many as any); + + await processJob( + {} as any, + { ...urlJob, url: post, capShown: true, logMessageId: undefined } as any, + 2, + ); + + expect(mockLog.append).toHaveBeenCalledWith( + expect.stringContaining('videos here'), + ); + }); + + it('does not announce a cap for a post that was not truncated', async () => { + carousel(); + await processJob({} as any, { ...urlJob, url: post } as any, 1); + expect(mockLog.append).not.toHaveBeenCalledWith( + expect.stringContaining('videos here'), + ); + }); + + it('re-resolves a confirmed entry to its own video, not the first', async () => { + carousel(); + await processJob( + {} as any, + confirmedJob({ info: { ...entries[2], filename: 'c.mp4' } }), + 1, + ); + expect(sentIds()).toEqual(['c']); + }); + + it('sends the confirmed video, not entry 0, when the re-scrape lost it', async () => { + spyMock(console, 'error'); + mockGetInfos.mockImplementation(async () => [entries[0]] as any); + seedInfoRow(post, entries[0]); + + await processJob( + {} as any, + confirmedJob({ info: { ...entries[2], filename: 'c.mp4' } }), + jobQueue.MAX_ATTEMPTS, + ); + + expect(mockSendVideo).not.toHaveBeenCalled(); + // without the eviction every attempt re-reads the same short list, and so + // does every later request for the post until the row's six hours are up + expect(rowCount('video_info')).toBe(0); + }); + + it('evicts a post row an entry cannot name itself when its download fails', async () => { + spyMock(console, 'error'); + // the playlist shape: entries carry their OWN webpage_url, so neither the + // row's url nor its webpage_url column is one the entry can name + mockGetInfos.mockImplementation(async () => [ + { ...entries[0], webpage_url: 'https://a' }, + ] as any); + db.query( + 'INSERT INTO video_info (url, info, webpage_url, created_at) VALUES (?, ?, ?, ?)', + ).run(post, '[]', 'https://elsewhere', Date.now()); + mockDownloadVideo.mockRejectedValue( + new downloadVideo.YtdlpError('failed', 'ERROR: Video unavailable'), + ); + try { + await processJob( + {} as any, + { ...urlJob, url: post } as any, + jobQueue.MAX_ATTEMPTS, + ); + expect(rowCount('video_info')).toBe(0); + } finally { + mockDownloadVideo.mockResolvedValue('downloaded' as any); + } + }); +}); diff --git a/test/job-queue.test.ts b/test/job-queue.test.ts index d017686..6ecf8ee 100644 --- a/test/job-queue.test.ts +++ b/test/job-queue.test.ts @@ -14,6 +14,7 @@ import { startJobQueue, stopJobQueue, type Job, + type UrlJob, } from '../src/job-queue'; import { addPending, getPending } from '../src/pending-downloads'; import { rowCount, spyMock, waitUntil, withFailingWrite } from './test-utils'; @@ -373,6 +374,27 @@ it('carries a processor mutation forward to the retry', async () => { expect(seen).toEqual([undefined, 99]); // the retry saw the persisted mutation }); +it('carries a delivered-videos mutation forward to the retry', async () => { + // a multi-video post records what it sent before anything else on the job + // moves, so a retry that lost it would send those videos a second time + spyMock(console, 'error'); + const seen: (string[] | undefined)[] = []; + let n = 0; + const processor = mock(async (j: Job) => { + seen.push((j as UrlJob).settledIds); + if (n++ === 0) { + (j as UrlJob).settledIds = ['sent']; + throw new Error('transient'); + } + }); + await startJobQueue(processor); + + await enqueueJob(job()); + + await waitUntil(jobsIdle); + expect(seen).toEqual([undefined, ['sent']]); +}); + it('clears a pending retry backoff on stop (the row recovers next boot)', async () => { spyMock(console, 'error'); setRetryBaseMs(500); diff --git a/test/test-utils.ts b/test/test-utils.ts index a281de3..15069fc 100644 --- a/test/test-utils.ts +++ b/test/test-utils.ts @@ -60,8 +60,8 @@ export const telegramError = (code: number, description: string) => response: { error_code: code, description }, }); -// seed a video_info row the way getInfo stores one (webpage_url denormalized -// into its own column, mirroring insertInfoStmt) +// seed a video_info row the way getInfos stores a single-video post +// (webpage_url denormalized into its own column, mirroring insertInfoStmt) export const seedInfoRow = ( url: string, info: unknown, @@ -73,7 +73,7 @@ export const seedInfoRow = ( ) .run( url, - JSON.stringify(info), + JSON.stringify([info]), (info as any)?.webpage_url ?? null, createdAt, );