fix: users can retriger transcription

apps/jobs/src/workers/maintenance.ts

@@ -1,12 +1,13 @@
 import './../bootstrap';
 
-import { Worker } from 'bullmq';
+import { Queue, Worker } from 'bullmq';
 
-import { pruneDeadLetterJobsOlderThan } from '@cio/db/queries';
+import { pruneDeadLetterJobsOlderThan, reapStuckMediaJobs } from '@cio/db/queries';
 import {
   JOB_NAMES,
   QUEUE_NAMES,
   ZDeadLetterCleanupPayload,
+  ZMediaJobReapPayload,
   ZRetentionCompactPayload,
   createRedisConnection
 } from '@cio/jobs';
@@ -36,11 +37,56 @@ const worker = new Worker(
       return { compacted: 0 };
     }
 
+    if (job.name === JOB_NAMES.maintenance.mediaJobReap) {
+      const data = ZMediaJobReapPayload.parse(job.data ?? {});
+      const cutoffIso = new Date(Date.now() - data.staleAfterMinutes * 60 * 1_000).toISOString();
+      const reaped = await reapStuckMediaJobs(cutoffIso);
+
+      if (reaped.length > 0) {
+        log.warn('media-job-reap-done', {
+          reaped: reaped.length,
+          cutoffIso,
+          ids: reaped.map((row) => row.id)
+        });
+      }
+
+      return { reaped: reaped.length };
+    }
+
     throw new Error(`Unknown maintenance job: ${job.name}`);
   },
   { connection, concurrency: 1 }
 );
 
+/**
+ * BullMQ repeatable scheduler for the media-job reaper. `upsertJobScheduler`
+ * is idempotent on the scheduler id, so it's safe to call on every worker
+ * boot. Without this, stuck `media_job` rows would never get cleaned up.
+ */
+const MEDIA_JOB_REAP_INTERVAL_MS = 5 * 60 * 1_000;
+// BullMQ requires a dedicated connection per Worker; give the Queue its own
+// so the scheduler upsert doesn't piggyback on the worker's blocking client.
+const schedulerConnection = createRedisConnection();
+const maintenanceQueue = new Queue(QUEUE_NAMES.maintenance, { connection: schedulerConnection });
+
+async function registerSchedulers(): Promise<void> {
+  try {
+    await maintenanceQueue.upsertJobScheduler(
+      'media-job-reap-scheduler',
+      { every: MEDIA_JOB_REAP_INTERVAL_MS },
+      { name: JOB_NAMES.maintenance.mediaJobReap, data: {} }
+    );
+    log.info('maintenance-scheduler-registered', {
+      name: JOB_NAMES.maintenance.mediaJobReap,
+      everyMs: MEDIA_JOB_REAP_INTERVAL_MS
+    });
+  } catch (err) {
+    log.error('maintenance-scheduler-register-failed', { error: errorMessage(err) });
+  }
+}
+
+void registerSchedulers();
+
 worker.on('ready', () => log.info('maintenance-worker-ready'));
 worker.on('failed', (job, err) =>
   log.error('maintenance-job-failed', { jobName: job?.name, error: errorMessage(err) })
@@ -50,6 +96,8 @@ worker.on('error', (err) => log.error('maintenance-worker-error', { error: error
 const shutdown = async (signal: string) => {
   log.info('maintenance-worker-shutdown', { signal });
   await worker.close();
+  await maintenanceQueue.close();
+  await schedulerConnection.quit();
   await connection.quit();
   process.exit(0);
 };

packages/db/src/queries/jobs/media-job.ts

@@ -1,6 +1,6 @@
 import * as schema from '@db/schema';
 
-import { and, desc, eq, inArray, sql } from '@db/drizzle';
+import { and, desc, eq, inArray, lt, sql } from '@db/drizzle';
 import { db, type DbOrTxClient } from '@db/drizzle';
 import type { TJobStatus, TMediaJob, TNewMediaJob } from '@db/types';
 
@@ -109,6 +109,39 @@ export async function updateMediaJob(
   }
 }
 
+/**
+ * Flip any `queued` / `running` media_job rows untouched since `staleBeforeIso`
+ * to `failed`. These belong to crashed workers, Redis-less runs, or BullMQ
+ * jobs that vanished — without this sweep the active-job guard would keep
+ * blocking new runs forever.
+ *
+ * Returns the rows that were updated so callers can log or dead-letter them.
+ */
+export async function reapStuckMediaJobs(
+  staleBeforeIso: string,
+  dbClient: DbOrTxClient = db
+): Promise<Pick<TMediaJob, 'id' | 'organizationId' | 'assetId'>[]> {
+  try {
+    return await dbClient
+      .update(schema.mediaJob)
+      .set({
+        status: 'failed',
+        stage: 'failed',
+        error: { code: 'STUCK_AFTER_RESTART', message: 'No worker progress within the stale window' },
+        updatedAt: new Date().toISOString()
+      })
+      .where(and(inArray(schema.mediaJob.status, ['queued', 'running']), lt(schema.mediaJob.updatedAt, staleBeforeIso)))
+      .returning({
+        id: schema.mediaJob.id,
+        organizationId: schema.mediaJob.organizationId,
+        assetId: schema.mediaJob.assetId
+      });
+  } catch (error) {
+    console.error('reapStuckMediaJobs error:', error);
+    throw new Error('Failed to reap stuck media jobs');
+  }
+}
+
 export async function requestMediaJobCancel(
   jobId: string,
   orgId: string,

packages/db/src/queries/media-transcript/media-transcript.ts

@@ -1,6 +1,6 @@
 import * as schema from '@db/schema';
 
-import { and, eq, inArray } from '@db/drizzle';
+import { and, eq, gte, inArray } from '@db/drizzle';
 import { db, type DbOrTxClient } from '@db/drizzle';
 import type { TMediaTranscript, TNewMediaTranscript } from '@db/types';
 
@@ -99,14 +99,28 @@ export async function listMediaTranscriptsByAssetIds(
 }
 
 /**
- * Returns true if any media_job for this asset is still queued or running.
+ * Default freshness window for "active" media_job rows. Rows older than this
+ * are assumed to belong to a crashed worker / a Redis-less run and are
+ * ignored by `hasActiveMediaJobForAsset` so users can retry. A separate
+ * sweeper (`reapStuckMediaJobs`) flips those stale rows to `failed`.
+ */
+export const MEDIA_JOB_STALE_AFTER_MINUTES = 30;
+
+/**
+ * Returns true if any media_job for this asset is still queued or running and
+ * has been touched within `staleAfterMinutes` (default 30). Stale rows are
+ * treated as dead: they shouldn't block new runs because the worker that
+ * owned them is gone (server restart, Redis flush, lost BullMQ job, etc.).
  */
 export async function hasActiveMediaJobForAsset(
   assetId: string,
   organizationId: string,
-  dbClient: DbOrTxClient = db
+  options: { staleAfterMinutes?: number; dbClient?: DbOrTxClient } = {}
 ): Promise<boolean> {
+  const { staleAfterMinutes = MEDIA_JOB_STALE_AFTER_MINUTES, dbClient = db } = options;
+
   try {
+    const freshSinceIso = new Date(Date.now() - staleAfterMinutes * 60 * 1_000).toISOString();
     const [row] = await dbClient
       .select({ id: schema.mediaJob.id })
       .from(schema.mediaJob)
@@ -114,7 +128,8 @@ export async function hasActiveMediaJobForAsset(
         and(
           eq(schema.mediaJob.assetId, assetId),
           eq(schema.mediaJob.organizationId, organizationId),
-          inArray(schema.mediaJob.status, ['queued', 'running'])
+          inArray(schema.mediaJob.status, ['queued', 'running']),
+          gte(schema.mediaJob.updatedAt, freshSinceIso)
         )
       )
       .limit(1);

packages/jobs/src/payloads/maintenance.ts

@@ -16,3 +16,14 @@ export const ZDeadLetterCleanupPayload = z.object({
   olderThanDays: z.number().int().positive().default(180)
 });
 export type TDeadLetterCleanupPayload = z.infer<typeof ZDeadLetterCleanupPayload>;
+
+/**
+ * Sweep `media_job` rows that are still `queued` / `running` but haven't been
+ * touched in `staleAfterMinutes`. Marks them `failed` so the active-job guard
+ * stops blocking new runs after a worker crash, server restart, or lost
+ * BullMQ job.
+ */
+export const ZMediaJobReapPayload = z.object({
+  staleAfterMinutes: z.number().int().positive().default(30)
+});
+export type TMediaJobReapPayload = z.infer<typeof ZMediaJobReapPayload>;

packages/jobs/src/queues/names.ts

@@ -37,6 +37,7 @@ export const JOB_NAMES = {
   },
   maintenance: {
     retentionCompact: 'retention-compact',
-    deadLetterCleanup: 'dead-letter-cleanup'
+    deadLetterCleanup: 'dead-letter-cleanup',
+    mediaJobReap: 'media-job-reap'
   }
 } as const;