'use node';

/**
 * Video-link ingest orchestrator. Single owner of the videoLinkJobs row
 * lifecycle: queued → fetching_metadata → (fetching_captions → indexing →
 * completed) OR (extracting_audio → transcribing_handoff → completed-via-
 * reactive-join).
 *
 * Design (after 30 + 15 sub-agent reviews):
 *   - Top-level `try { ... } finally { rm jobDir }` guarantees /tmp/vlink-
 *     <uuid>/ cleanup even on action timeout / OOM / kill -9.
 *   - Phase-boundary status re-check: cancel mutations flip `status` to
 *     `'skipped'`, this action's `assertNotCancelled()` peeks the row
 *     between expensive operations and bails out gracefully (mirrors
 *     transcribe_audio.ts:317-337).
 *   - 4h duration cap rejects videos AFTER cheap metadata fetch but
 *     BEFORE any audio download (R14: pre-spend abuse guard).
 *   - Caption language precedence: fidelity-first hybrid (R5):
 *       1. Manual subs in original language
 *       2. Manual subs in UI locale (non-translated)
 *       3. Manual subs in en (LLM's strongest translation pivot)
 *       4. Manual subs in ANY language
 *       5. Auto-gen ASR in original language (if non-translated)
 *       6. Auto-gen translation into UI locale (last caption resort)
 *       7. Fall through to Whisper-audio path
 *   - Error classification: stderr from ytdlp.ts is already classified;
 *     retry [30s, 60s, 120s] aligned with TRANSCRIBE_RETRY_DELAYS_MS;
 *     botDetection/rateLimited bypass short-retry (long-jitter or fail).
 */

import { promises as fs } from 'node:fs';

import { v } from 'convex/values';

import { CHAT_AUDIO_MAX_DURATION_SEC } from '../../lib/shared/file-types';
import { internal } from '../_generated/api';
import type { Id } from '../_generated/dataModel';
import type { ActionCtx } from '../_generated/server';
import { internalAction } from '../_generated/server';
import {
  joinSegmentsWithParagraphs,
  CAPTION_PROFILE,
  formatHms,
} from '../file_metadata/paragraphize';
import { sanitizeUntrustedField } from '../lib/untrusted_content';
import {
  captionsToParagraphSegments,
  parseVtt,
  rollingWindowDedup,
} from './captions_parser';
import type { VideoLinkErrorReason } from './internal_mutations';
import { assertSafeUrl, UrlSafetyError } from './url_safety';
import {
  YtDlpError,
  createJobDir,
  ytdlpExtractAudio,
  ytdlpJson,
  ytdlpWriteSubs,
  type YtDlpMetadata,
} from './ytdlp';

const INGEST_RETRY_DELAYS_MS = [30_000, 60_000, 120_000];
const MAX_ATTEMPTS = INGEST_RETRY_DELAYS_MS.length + 1;

const NEVER_RETRY: ReadonlySet<string> = new Set([
  'privateOrAgeGated',
  'geoblocked',
  'unsupported',
  'unavailable',
  'liveStream',
  'premiere',
  'memberOnly',
  'jsRuntimeMissing',
  'binaryNotInstalled',
  'outputValidationFailed',
  // botDetection + rateLimited would benefit from a long-jitter retry
  // but short retries make YouTube's rate-limit worse. Treat as
  // never-retry for v1; user can manually retry after waiting.
  'botDetection',
  'rateLimited',
]);

interface CaptionSelection {
  /** yt-dlp `--sub-langs` value (e.g. `en` or `zh-Hans`). */
  lang: string;
  /** Whether yt-dlp's `--write-auto-subs` flag must be added. */
  includeAutoGenerated: boolean;
  /** Reported back into the job row for debugging + chip tooltip. */
  trackKind: 'manual' | 'asr' | 'auto-translated';
}

function selectCaptionLanguage(
  meta: YtDlpMetadata,
  userLocale: string | undefined,
): CaptionSelection | null {
  const uiBase = userLocale?.split('-')[0]?.toLowerCase();
  const original = meta.language?.toLowerCase();
  const manualLangs = Object.keys(meta.subtitles ?? {}).map((l) =>
    l.toLowerCase(),
  );
  const autoLangs = Object.keys(meta.automatic_captions ?? {}).map((l) =>
    l.toLowerCase(),
  );

  // 1. Manual subs in original language
  if (original) {
    const hit = manualLangs.find(
      (l) => l === original || l.split('-')[0] === original,
    );
    if (hit)
      return { lang: hit, includeAutoGenerated: false, trackKind: 'manual' };
  }
  // 2. Manual subs in UI locale (non-translated)
  if (uiBase) {
    const hit = manualLangs.find(
      (l) => l === userLocale?.toLowerCase() || l.split('-')[0] === uiBase,
    );
    if (hit)
      return { lang: hit, includeAutoGenerated: false, trackKind: 'manual' };
  }
  // 3. Manual English (LLM's strongest pivot)
  {
    const hit = manualLangs.find((l) => l === 'en' || l.startsWith('en-'));
    if (hit)
      return { lang: hit, includeAutoGenerated: false, trackKind: 'manual' };
  }
  // 4. ANY manual subtitle
  if (manualLangs.length > 0) {
    return {
      lang: manualLangs[0],
      includeAutoGenerated: false,
      trackKind: 'manual',
    };
  }
  // 5. Auto-gen ASR in original language
  if (original) {
    const hit = autoLangs.find(
      (l) => l === original || l.split('-')[0] === original,
    );
    if (hit) return { lang: hit, includeAutoGenerated: true, trackKind: 'asr' };
  }
  // 6. Auto-gen in UI locale (machine-translated by YouTube — last resort)
  if (uiBase) {
    const hit = autoLangs.find((l) => l.split('-')[0] === uiBase);
    if (hit)
      return {
        lang: hit,
        includeAutoGenerated: true,
        trackKind: 'auto-translated',
      };
  }
  // 7. Auto-gen en
  {
    const hit = autoLangs.find((l) => l === 'en' || l.startsWith('en-'));
    if (hit) return { lang: hit, includeAutoGenerated: true, trackKind: 'asr' };
  }

  return null;
}

export const ingestVideoLink = internalAction({
  args: {
    jobId: v.id('videoLinkJobs'),
    userLocale: v.optional(v.string()),
  },
  async handler(ctx, args) {
    const job = await ctx.runQuery(
      internal.video_links.internal_queries.getJobById,
      {
        jobId: args.jobId,
      },
    );
    if (!job) return;
    // Bail on any non-`queued` initial state — `skipped`/`failed` are the
    // explicit cancel/terminal-error cases, and `completed`/`indexing`/
    // `transcribing_handoff`/`extracting_audio`/`fetching_*` indicate a
    // duplicate-schedule or already-running prior instance. Re-running
    // the orchestrator on those states burns yt-dlp metadata fetches and
    // (without CAS, before this fix) could overwrite terminal artifacts.
    if (job.status !== 'queued') {
      console.warn(
        JSON.stringify({
          event: 'video_link.duplicate_schedule',
          jobId: args.jobId,
          observedStatus: job.status,
        }),
      );
      return;
    }

    /** Re-read the job and bail if it's been cancelled, watchdog-failed,
     * or cascade-deleted. Treats `failed` as terminal in addition to
     * `skipped` so a watchdog flip mid-flight (e.g. extracting_audio
     * exceeds its 20-min window) is caught at the next checkpoint and
     * the worker stops doing expensive work / writing rows. Without
     * this, a watchdog-induced `failed` could be silently overwritten
     * by the worker's next CAS-less patch. */
    const assertNotCancelled = async (): Promise<boolean> => {
      const fresh = await ctx.runQuery(
        internal.video_links.internal_queries.getJobById,
        {
          jobId: args.jobId,
        },
      );
      // Row missing (cascade-deleted by thread cleanup or subject erasure
      // mid-action) is treated as cancellation — the original `!fresh ||`
      // short-circuit returned true here, letting subsequent mutations
      // crash on the gone row (R2 review).
      if (!fresh) return false;
      return fresh.status !== 'skipped' && fresh.status !== 'failed';
    };

    /** Best-effort orphan-blob cleanup when we bail between
     * `ctx.storage.store` and the cross-table write that would record
     * the storageId on the job row. Suppresses delete errors (the blob
     * may have already been reaped by cascade) but logs them. */
    const deleteOrphanBlob = async (
      storageId: Id<'_storage'>,
    ): Promise<void> => {
      try {
        await ctx.storage.delete(storageId);
      } catch (err) {
        console.warn(
          JSON.stringify({
            event: 'video_link.orphan_blob_delete_failed',
            jobId: args.jobId,
            storageId,
            error: err instanceof Error ? err.message : String(err),
          }),
        );
      }
    };

    const { jobDir, cleanup } = await createJobDir();

    const fail = async (
      reasonCode: VideoLinkErrorReason,
      message: string,
    ): Promise<void> => {
      console.error(
        JSON.stringify({
          event: 'video_link.failed',
          jobId: args.jobId,
          orgId: job.organizationId,
          threadId: job.threadId,
          sourceUrlHash: job.sourceUrlHash,
          sourcePlatform: job.sourcePlatform,
          reason: reasonCode,
          message,
        }),
      );
      await ctx.runMutation(internal.video_links.internal_mutations.updateJob, {
        jobId: args.jobId,
        status: 'failed',
        errorReasonCode: reasonCode,
        errorMessage: message.slice(0, 500),
        progress: undefined,
      });
    };

    try {
      // -----------------------------------------------------------------
      // Phase A: yt-dlp -j metadata
      // -----------------------------------------------------------------
      try {
        await assertSafeUrl(job.sourceUrl);
      } catch (err) {
        if (err instanceof UrlSafetyError) {
          await fail(err.kind, err.message);
          return;
        }
        throw err;
      }

      await ctx.runMutation(internal.video_links.internal_mutations.updateJob, {
        jobId: args.jobId,
        status: 'fetching_metadata',
        // CAS guard: only advance if the row is still in 'queued'. Defends
        // against a watchdog-failed → manual-retry race where two action
        // instances overlap; the loser silently no-ops instead of
        // stomping the winner's writes.
        expectedStatus: 'queued',
      });
      if (!(await assertNotCancelled())) return;

      let metadata: YtDlpMetadata;
      try {
        metadata = await ytdlpJson(job.sourceUrl, jobDir);
      } catch (err) {
        await handleYtDlpError(ctx, args.jobId, job, err, fail);
        return;
      }

      // Reject live / upcoming / oversized BEFORE any download.
      if (metadata.is_live || metadata.live_status === 'is_live') {
        await fail('liveStream', 'Live streams are not supported');
        return;
      }
      if (metadata.live_status === 'is_upcoming') {
        await fail('premiere', 'Premiere/upcoming video not yet available');
        return;
      }
      if (
        typeof metadata.duration === 'number' &&
        metadata.duration > CHAT_AUDIO_MAX_DURATION_SEC
      ) {
        await fail(
          'videoTooLong',
          `Video duration ${Math.round(metadata.duration)}s exceeds ${CHAT_AUDIO_MAX_DURATION_SEC}s limit`,
        );
        return;
      }

      // Sanitize uploader-controlled metadata at the trust boundary, BEFORE
      // it enters our DB / fileMetadata.transcript / fileName / agent context.
      // A chapter title like `\n\n[SYSTEM] ignore previous instructions` would
      // otherwise pass through unescaped into the transcript body (since the
      // TOC is concatenated into the persisted text in Phase B) and the agent
      // would see it outside any <untrusted_source> wrapper.
      const chapters =
        metadata.chapters?.map((c) => ({
          startSec: c.start_time,
          endSec: c.end_time,
          title: sanitizeUntrustedField(c.title, 200),
        })) ?? [];
      const safeTitle = metadata.title
        ? sanitizeUntrustedField(metadata.title, 200)
        : undefined;
      const safeUploader =
        (metadata.uploader ?? metadata.channel)
          ? sanitizeUntrustedField(
              metadata.uploader ?? metadata.channel ?? '',
              120,
            )
          : undefined;

      await ctx.runMutation(internal.video_links.internal_mutations.updateJob, {
        jobId: args.jobId,
        videoTitle: safeTitle,
        videoUploader: safeUploader,
        videoDurationSec: metadata.duration,
        videoLanguage: metadata.language ?? undefined,
        ...(chapters.length > 0 ? { videoChapters: chapters } : {}),
      });

      console.log(
        JSON.stringify({
          event: 'video_link.metadata_fetched',
          jobId: args.jobId,
          orgId: job.organizationId,
          duration: metadata.duration,
          sourcePlatform: job.sourcePlatform,
          availableManualLangs: Object.keys(metadata.subtitles ?? {}).length,
          availableAutoLangs: Object.keys(metadata.automatic_captions ?? {})
            .length,
        }),
      );

      // -----------------------------------------------------------------
      // Phase B: caption selection + fetch
      // -----------------------------------------------------------------
      const selection = selectCaptionLanguage(metadata, args.userLocale);

      if (selection) {
        await ctx.runMutation(
          internal.video_links.internal_mutations.updateJob,
          {
            jobId: args.jobId,
            status: 'fetching_captions',
            expectedStatus: 'fetching_metadata',
            captionLang: selection.lang,
            captionTrackKind: selection.trackKind,
          },
        );
        if (!(await assertNotCancelled())) return;

        // Re-validate URL against fresh DNS before spawning yt-dlp again.
        // Defends against TTL=0 rebind between Phase A and Phase B.
        try {
          await assertSafeUrl(job.sourceUrl);
        } catch (err) {
          if (err instanceof UrlSafetyError) {
            await fail(err.kind, err.message);
            return;
          }
          throw err;
        }

        let vttPath: string | null = null;
        try {
          vttPath = await ytdlpWriteSubs(
            job.sourceUrl,
            selection.lang,
            jobDir,
            {
              includeAutoGenerated: selection.includeAutoGenerated,
            },
          );
        } catch (err) {
          // Caption fetch failure ≠ feature failure — fall through to
          // Whisper. Log + continue.
          console.warn(
            JSON.stringify({
              event: 'video_link.captions_miss',
              jobId: args.jobId,
              reason: err instanceof YtDlpError ? err.reason : 'parse_failed',
            }),
          );
          vttPath = null;
        }

        if (vttPath) {
          const buf = await fs.readFile(vttPath);
          const segments = rollingWindowDedup(parseVtt(buf));
          if (segments.length > 0) {
            const transcriptBody = joinSegmentsWithParagraphs(
              captionsToParagraphSegments(segments),
              '',
              { profile: CAPTION_PROFILE, addTimestamps: true },
            );

            if (transcriptBody.length > 0) {
              await ctx.runMutation(
                internal.video_links.internal_mutations.updateJob,
                {
                  jobId: args.jobId,
                  status: 'indexing',
                  expectedStatus: 'fetching_captions',
                },
              );
              if (!(await assertNotCancelled())) return;

              // Prepend chapter TOC when chapters are present — agent
              // can cite "Chapter 3: …" in summaries.
              // `chapters[i].title` was already passed through
              // `sanitizeUntrustedField` at the metadata-fetch trust
              // boundary above, so this concatenation is safe — chapter
              // titles can no longer carry literal `\n[SYSTEM]` payloads
              // or `</untrusted_source>` close-tag attempts.
              const chapterToc =
                chapters.length > 0
                  ? 'Chapters:\n' +
                    chapters
                      .map((c) => `  ${formatHms(c.startSec)} ${c.title}`)
                      .join('\n') +
                    '\n\n'
                  : '';
              const fullTranscript = chapterToc + transcriptBody;

              const blob = new Blob([fullTranscript], {
                type: 'text/plain; charset=utf-8',
              });
              const storageId = await ctx.storage.store(blob);

              // GDPR / cascade race window: between the storage.store
              // above and the cross-table writes below, an erasure or
              // thread-cleanup pass can delete the videoLinkJobs row.
              // Without this re-check, `insertSyntheticFileMetadata`
              // would create a fresh fileMetadata row + schedule RAG
              // ingestion that survives the erasure entirely. The
              // mutation itself has a defense-in-depth `db.get(jobId)`
              // guard, but the orchestrator-level bail keeps the failure
              // surface tight (no scheduled job stays queued).
              if (!(await assertNotCancelled())) {
                await deleteOrphanBlob(storageId);
                return;
              }

              // Record `storageId` on the job row BEFORE the synthetic
              // fileMetadata insert. CAS-gated against `indexing` — if a
              // watchdog flip raced us during storage.store, the CAS
              // misses and we drop the orphan blob without further
              // writes. Audio branch (Phase C, below) follows the same
              // pattern.
              const storagePatch = await ctx.runMutation(
                internal.video_links.internal_mutations.updateJob,
                {
                  jobId: args.jobId,
                  storageId,
                  expectedStatus: 'indexing',
                },
              );
              if (storagePatch !== 'ok') {
                await deleteOrphanBlob(storageId);
                return;
              }

              await ctx.runMutation(
                internal.video_links.internal_mutations
                  .insertSyntheticFileMetadata,
                {
                  jobId: args.jobId,
                  storageId,
                  transcript: fullTranscript,
                  fileSize: blob.size,
                  videoTitle: safeTitle ?? 'Untitled video',
                  videoUploader: safeUploader,
                  videoDurationSec: metadata.duration ?? 0,
                  sourceUrl: job.sourceUrl,
                  sourcePlatform: job.sourcePlatform,
                  transcriptSource:
                    selection.trackKind === 'manual'
                      ? 'captions_human'
                      : 'captions_auto',
                  captionLang: selection.lang,
                  threadId: job.threadId,
                  organizationId: job.organizationId,
                  uploadedBy: job.uploadedBy,
                },
              );

              console.log(
                JSON.stringify({
                  event: 'video_link.completed',
                  jobId: args.jobId,
                  orgId: job.organizationId,
                  source: 'captions',
                  trackKind: selection.trackKind,
                  lang: selection.lang,
                }),
              );
              return;
            }
          }
        }
      }

      // -----------------------------------------------------------------
      // Phase C: captions missed → Whisper fallback via audio extract
      // -----------------------------------------------------------------
      // Phase C entry can be reached from two pre-states: `fetching_metadata`
      // (no caption track selected) OR `fetching_captions` (selection
      // made, but yt-dlp failed to fetch / parsed to empty). Read the
      // current status BEFORE the patch to choose the right CAS predicate
      // — without this, a watchdog flip during the long Phase B yt-dlp
      // call would be silently undone by an unguarded patch here. And
      // run `assertNotCancelled` BEFORE the patch (the previous order
      // had it after — which always observed the worker's own write and
      // so could never detect a watchdog/cancel that landed during the
      // captions phase).
      if (!(await assertNotCancelled())) return;
      const phaseCPre = await ctx.runQuery(
        internal.video_links.internal_queries.getJobById,
        { jobId: args.jobId },
      );
      if (
        !phaseCPre ||
        (phaseCPre.status !== 'fetching_metadata' &&
          phaseCPre.status !== 'fetching_captions')
      ) {
        // Either deleted, already in a terminal state, or somehow past
        // Phase C entry — nothing safe to do.
        return;
      }
      const phaseCEntry = await ctx.runMutation(
        internal.video_links.internal_mutations.updateJob,
        {
          jobId: args.jobId,
          status: 'extracting_audio',
          expectedStatus: phaseCPre.status,
        },
      );
      if (phaseCEntry !== 'ok') return;

      // Re-validate URL against fresh DNS before the longest yt-dlp call
      // (15-min wall-clock). Defends against rebind across Phase B → C.
      try {
        await assertSafeUrl(job.sourceUrl);
      } catch (err) {
        if (err instanceof UrlSafetyError) {
          await fail(err.kind, err.message);
          return;
        }
        throw err;
      }

      let audioPath: string;
      try {
        audioPath = await ytdlpExtractAudio(job.sourceUrl, jobDir);
      } catch (err) {
        await handleYtDlpError(ctx, args.jobId, job, err, fail);
        return;
      }

      if (!(await assertNotCancelled())) return;

      const audioBuf = await fs.readFile(audioPath);
      const audioBlob = new Blob([audioBuf], { type: 'audio/ogg' });
      const audioStorageId = await ctx.storage.store(audioBlob);

      // GDPR / cascade race window: between storage.store and the
      // saveFileMetadata cross-table write below, erasure can delete
      // the job row. Without this check, saveFileMetadata would land
      // an orphan fileMetadata row + schedule transcribe_audio that
      // both survive the erasure pass.
      if (!(await assertNotCancelled())) {
        await deleteOrphanBlob(audioStorageId);
        return;
      }

      // CRITICAL: record `storageId` on the job row BEFORE inserting the
      // fileMetadata row. If `saveFileMetadata` fails (validator, OOM,
      // network blip) the cancel cleanup path needs `job.storageId` to
      // find the blob and delete it — without this pre-write the blob
      // orphans in `_storage` indefinitely (R2 cluster E). CAS-gated
      // against `extracting_audio` so a watchdog flip during the long
      // yt-dlp call gets caught here.
      const audioStoragePatch = await ctx.runMutation(
        internal.video_links.internal_mutations.updateJob,
        {
          jobId: args.jobId,
          storageId: audioStorageId,
          expectedStatus: 'extracting_audio',
        },
      );
      if (audioStoragePatch !== 'ok') {
        await deleteOrphanBlob(audioStorageId);
        return;
      }

      // Internal variant (not the public `api.file_metadata.mutations.*`).
      // Scheduled actions run with no user context — the public mutation
      // calls `authComponent.getAuthUser(ctx)` which throws Unauthenticated.
      // The internal variant accepts uploadedBy + threadId as explicit args
      // and handles the same insert + transcribe_audio scheduling.
      // Capture the returned fileMetadataId — without it, the chip's
      // reactive join in queries.ts:projectJob can't read through to
      // transcriptionStatus/transcriptionProgress, so the chip would
      // stay stuck at the literal "Transcribing…" i18n string with no
      // progress and no failure surfacing.
      const fileMetadataId = await ctx.runMutation(
        internal.file_metadata.internal_mutations.saveFileMetadata,
        {
          organizationId: job.organizationId,
          storageId: audioStorageId,
          fileName: `${safeTitle ?? 'video'}.ogg`,
          contentType: 'audio/ogg',
          size: audioBlob.size,
          // 'video_link' instead of 'user' so prompt-injection-aware
          // downstream code can differentiate third-party content.
          source: 'video_link',
          uploadedBy: job.uploadedBy,
          ...(job.threadId !== undefined && { threadId: job.threadId }),
          // sourceUrl/sourcePlatform/videoTitle/uploader/duration NOT
          // copied here — `start_agent_chat.ts` JOINs videoLinkJobs by
          // storageId. Avoids two writers of the same data; the legacy
          // fileMetadata fields stay as @deprecated fallback.
        },
      );

      await ctx.runMutation(internal.video_links.internal_mutations.updateJob, {
        jobId: args.jobId,
        status: 'transcribing_handoff',
        expectedStatus: 'extracting_audio',
        storageId: audioStorageId,
        fileMetadataId,
        transcriptSource: 'whisper',
      });

      console.log(
        JSON.stringify({
          event: 'video_link.transcribing_handoff',
          jobId: args.jobId,
          orgId: job.organizationId,
          audioStorageId,
          audioBytes: audioBlob.size,
        }),
      );

      // From here on, the chip's reactive query joins fileMetadata's
      // transcriptionStatus and projects it as displayStatus. We don't
      // wait for Whisper to complete — when transcribe_audio.ts finishes,
      // the chip flips to 'completed' purely through Convex reactivity.
    } finally {
      await cleanup();
    }
  },
});

async function handleYtDlpError(
  ctx: ActionCtx,
  jobId: Id<'videoLinkJobs'>,
  job: { attempts?: number },
  err: unknown,
  fail: (reasonCode: VideoLinkErrorReason, message: string) => Promise<void>,
): Promise<void> {
  if (err instanceof YtDlpError) {
    const attempts = (job.attempts ?? 0) + 1;
    if (NEVER_RETRY.has(err.reason) || attempts >= MAX_ATTEMPTS) {
      await fail(err.reason, err.sanitizedStderr || err.message);
      return;
    }
    // Schedule retry with backoff. PERSIST `attempts` — without writing
    // it back, every retried run reads `job.attempts ?? 0` and recomputes
    // 1, so MAX_ATTEMPTS is never reached and transient failures retry
    // forever bounded only by NEVER_RETRY (round-2 V7 / HIGH #12).
    const delayMs =
      INGEST_RETRY_DELAYS_MS[
        Math.min(attempts - 1, INGEST_RETRY_DELAYS_MS.length - 1)
      ];
    // Clear any storageId/fileMetadataId from the prior attempt before
    // re-running the orchestrator. Without this, a retry that previously
    // patched `storageId` (Phase C, Whisper path) would carry the stale
    // id into the fresh action, double-storing the audio blob on success
    // (the second `ctx.storage.store` overwrites the field; the original
    // blob orphans). `cleanupCancelledVideoLink` already handles the
    // delete bookkeeping safely.
    await ctx.scheduler.runAfter(
      0,
      internal.video_links.internal_mutations.cleanupCancelledVideoLink,
      { jobId },
    );
    await ctx.runMutation(internal.video_links.internal_mutations.updateJob, {
      jobId,
      status: 'queued',
      attempts,
      errorReasonCode: err.reason,
      errorMessage: err.sanitizedStderr.slice(0, 500),
    });
    await ctx.scheduler.runAfter(
      delayMs,
      internal.video_links.ingest_video_link.ingestVideoLink,
      { jobId, userLocale: undefined },
    );
    console.warn(
      JSON.stringify({
        event: 'video_link.retrying',
        jobId,
        attempts,
        delayMs,
        reason: err.reason,
        // Truncated sanitized stderr so operators can diagnose
        // `transient`-classified failures without inspecting the DB row
        // (the classifier returns 'transient' for any pattern it doesn't
        // recognize — typically because yt-dlp added a new error message
        // shape upstream, and silent-retry is the worst-case for debug).
        stderr: err.sanitizedStderr.slice(0, 400),
        message: err.message,
      }),
    );
    return;
  }
  const message = err instanceof Error ? err.message : String(err);
  await fail('transient', message);
}
