Blame for src/rendering/queue/processors/video.processor.ts

/**
 * Video Processor
 * 任务处理器 - 主编排器
 */

import { performance } from 'node:perf_hooks'
import { videoQueue } from '../bull-config'
import { finalizeJobTimings, storeJobResult } from '../../../infrastructure/redis/job/redis-job-result-store'
import { deleteJobStage } from '../../../infrastructure/redis/job/redis-job-state-store'
import { touchJobTracking } from '../../../infrastructure/redis/job/redis-job-tracking-store'
import { clearJobCancelled } from '../../job/job-cancel-store'
import { JobCancelledError } from '../../../shared/errors/errors'
import { createLogger } from '../../../shared/logging/logger'
import type { VideoJobData } from '../../../types'
import { runEditFlow, runGenerationFlow, runPreGeneratedFlow } from './video-processor-flows'
import { getRetryMeta, shouldDisableQueueRetry, storeProcessingStage } from './video-processor-utils'
import { getCurrentJobLogSummary, getCurrentJobTokenUsage, runWithJobLogContext } from '../../job/job-log-context'
import { recordTokenUsage, type TokenUsageTotals } from '../../../access'
import { normalizeTimings } from '../../../shared/time/timings'

const logger = createLogger('VideoProcessor')
const activeExecutions = new Map<string, Promise<unknown>>()
const JOB_HEARTBEAT_MS = 15_000

const EMPTY_TOKEN_TOTALS: TokenUsageTotals = {
  promptTokens: 0,
  completionTokens: 0,
  totalTokens: 0,
  measuredCalls: 0,
  unmeasuredCalls: 0
}

/**
 * Writes this Job's cumulative usage to the CatKey ledger.
 *
 * Called only on a terminal outcome. Intermediate failed attempts are skipped on purpose: the
 * upsert is keyed on the Job id, so the attempt that finally succeeds would overwrite them anyway,
 * and a Job that never succeeds should not leave a half-counted row behind.
 *
 * The recorder itself decides whether anything is metered — an absent `catKeyId` means BYOK — so
 * there is no mode check here.
 */
async function recordJobTokenUsage(data: VideoJobData): Promise<void> {
  await recordTokenUsage({
    catKeyId: data.catKeyId,
    model: data.customApiConfig?.model ?? '',
    sourceKind: 'workflow_job',
    sourceId: data.jobId,
    totals: getCurrentJobTokenUsage()?.totals ?? EMPTY_TOKEN_TOTALS
  })
}

function emitJobSummary(args: {
  jobId: string
  taskType: 'pre-generated' | 'ai-edit' | 'generation'
  result: 'completed' | 'failed'
  outputMode: string
  timings?: Record<string, number>
  renderPeakMemoryMB?: number
  error?: string
  attempt?: number
  maxAttempts?: number
}): void {
  const tokenSummary = getCurrentJobLogSummary()
  logger.info('job_summary', {
    _logType: 'job_summary',
    jobId: args.jobId,
    taskType: args.taskType,
    result: args.result,
    outputMode: args.outputMode,
    attempt: args.attempt,
    maxAttempts: args.maxAttempts,
    timings: args.timings,
    renderPeakMemoryMB: args.renderPeakMemoryMB,
    error: args.error,
    tokens: tokenSummary
      ? {
          totals: tokenSummary.totals,
          calls: tokenSummary.calls
        }
      : {
          totals: {
            promptTokens: 0,
            completionTokens: 0,
            totalTokens: 0,
            measuredCalls: 0,
            unmeasuredCalls: 0
          },
          calls: []
        }
  })
}

async function executeVideoJob(job: any): Promise<unknown> {
  const executionStartMono = performance.now()
  const data = job.data as VideoJobData
  const contextAttempt = typeof job.attemptsMade === 'number' ? job.attemptsMade + 1 : 1

  return runWithJobLogContext(
    {
      jobId: data.jobId,
      outputMode: data.outputMode || 'video',
      attempts: contextAttempt
    },
    async () => {
  const {
    jobId,
    concept,
    quality,
    outputMode = 'video',
    preGeneratedCode,
    editCode,
    editInstructions,
    promptOverrides,
    referenceImages
  } = data

  logger.info('Processing video job', {
    jobId,
    concept,
    outputMode,
    quality,
    hasPreGeneratedCode: !!preGeneratedCode,
    hasEditRequest: !!editInstructions,
    referenceImageCount: referenceImages?.length || 0
  })

  const timings: Record<string, number> = {}
  const submittedAtMs = Date.parse(data.timestamp)
  if (Number.isFinite(submittedAtMs)) {
    const startedAtMs = typeof job.processedOn === 'number' ? job.processedOn : Date.now()
    timings.queue = Math.max(0, startedAtMs - submittedAtMs)
  }
  const retryMeta = getRetryMeta(job)
  const initialStage = preGeneratedCode
    ? 'rendering'
    : editCode && editInstructions
      ? 'generating'
      : 'analyzing'

  try {
    await storeProcessingStage(jobId, initialStage, {
      attempt: retryMeta.currentAttempt,
      status: 'processing',
    })

    let taskType: 'pre-generated' | 'ai-edit' | 'generation'
    let result
    if (preGeneratedCode) {
      result = await runPreGeneratedFlow({ job, data, promptOverrides, timings })
      taskType = 'pre-generated'
      logger.info('Job completed (pre-generated code)', { jobId, timings })
    } else if (editCode && editInstructions) {
      result = await runEditFlow({ job, data, promptOverrides, timings })
      taskType = 'ai-edit'
      logger.info('Job completed', { jobId, source: 'ai-edit', timings })
    } else {
      result = await runGenerationFlow({ job, data, promptOverrides, timings })
      taskType = 'generation'
      logger.info('Job completed', { jobId, source: 'generation', timings })
    }

    // Persist the spans that can only be known after the result write: store itself and the
    // monotonic, end-to-end total. `storeOutcome` is checked so a duplicate execution that
    // preserved an earlier completed result cannot rewrite that result's timing.
    timings.total = Math.max(0, (timings.queue || 0) + (performance.now() - executionStartMono))
    const finalizedAt = Date.now()
    if (result.storeOutcome === 'stored') {
      const finalized = await finalizeJobTimings(
        jobId,
        'completed',
        normalizeTimings(timings) ?? {},
        finalizedAt
      )
      if (!finalized) {
        logger.warn('Failed to persist final job timings', { jobId, status: 'completed' })
      }
    }

    // Terminal success: the authoritative `job:result:*` is now in place, so the intermediate
    // `job:stage:*` / `job:tracking:*` markers are no longer needed. Removing them keeps Redis
    // from showing a stale "rendering"/"generating" state after completion and prevents any future
    // stage-based consumer (e.g. an SSE progress feed) from reporting a finished job as in-flight.
    await deleteJobStage(jobId)

    emitJobSummary({
      jobId,
      taskType,
      result: 'completed',
      outputMode,
      timings,
      renderPeakMemoryMB: result.renderPeakMemoryMB,
      attempt: retryMeta.currentAttempt,
      maxAttempts: retryMeta.maxAttempts
    })
    await recordJobTokenUsage(data)
    return result
  } catch (error) {
    const errorMessage = error instanceof Error ? error.message : String(error)
    const cancelReason = error instanceof JobCancelledError ? error.details : undefined
    const currentRetryMeta = getRetryMeta(job)
    const disableQueueRetry = shouldDisableQueueRetry(errorMessage)
    const willQueueRetry = !disableQueueRetry && currentRetryMeta.hasRemainingAttempts

    if (disableQueueRetry) {
      try {
        job.discard()
        logger.warn('Queue retry disabled for exhausted code retry', {
          jobId,
          error: errorMessage,
          currentAttempt: currentRetryMeta.currentAttempt,
          maxAttempts: currentRetryMeta.maxAttempts
        })
      } catch (discardError) {
        logger.warn('Failed to discard job retry', { jobId, error: discardError })
      }
    }

    if (willQueueRetry) {
      logger.warn('Job attempt failed, Bull will retry', {
        jobId,
        error: errorMessage,
        currentAttempt: currentRetryMeta.currentAttempt,
        maxAttempts: currentRetryMeta.maxAttempts
      })
      throw error
    }

    logger.error('Job failed', {
      jobId,
      error: errorMessage,
      timings,
      currentAttempt: currentRetryMeta.currentAttempt,
      maxAttempts: currentRetryMeta.maxAttempts
    })

    const failureStoreStart = performance.now()
    const storeOutcome = await storeJobResult(jobId, {
      status: 'failed',
      data: {
        error: errorMessage,
        cancelReason,
        outputMode,
        timings: normalizeTimings(timings),
        tokenUsage: getCurrentJobTokenUsage()
      }
    })
    timings.store = performance.now() - failureStoreStart
    timings.total = Math.max(0, (timings.queue || 0) + (performance.now() - executionStartMono))
    if (storeOutcome === 'stored') {
      const finalized = await finalizeJobTimings(
        jobId,
        'failed',
        normalizeTimings(timings) ?? {},
        Date.now()
      )
      if (!finalized) {
        logger.warn('Failed to persist final job timings', { jobId, status: 'failed' })
      }
    }
    if (!cancelReason) {
      await clearJobCancelled(jobId)
    }

    // Terminal failure (no remaining Bull retries): same cleanup as success. The `willQueueRetry`
    // branch above threw before reaching here, so this only runs when the job is truly done.
    await deleteJobStage(jobId)

    // `preserved-completed` is the success-path guard; `preserved-failed` is the failure-path
    // guard used when an earlier execution already stored a completed result.
    if (storeOutcome === 'preserved-completed' || storeOutcome === 'preserved-failed') {
      logger.warn('Duplicate execution failed after a completed result; preserving success', {
        jobId,
        error: errorMessage,
        currentAttempt: currentRetryMeta.currentAttempt,
      })
      return {
        success: true,
        source: 'preserved-completed-result',
      }
    }

    emitJobSummary({
      jobId,
      taskType: editCode && editInstructions ? 'ai-edit' : preGeneratedCode ? 'pre-generated' : 'generation',
      result: 'failed',
      outputMode,
      timings,
      error: errorMessage,
      attempt: currentRetryMeta.currentAttempt,
      maxAttempts: currentRetryMeta.maxAttempts
    })

    // Terminal failure: the attempt consumed tokens it will never be credited for by a retry, so
    // the row is written here. `preserved-completed` returned above and must not reach this point.
    await recordJobTokenUsage(data)

    throw error
  }
    }
  )
}

videoQueue.process(async (job) => {
  const data = job.data as VideoJobData
  const jobId = data.jobId || String(job.id)
  const activeExecution = activeExecutions.get(jobId)

  if (activeExecution) {
    logger.warn('Duplicate Bull execution joined the existing in-process job', {
      jobId,
      attemptsMade: job.attemptsMade,
    })
    return activeExecution
  }

  const execution = executeVideoJob(job)
  activeExecutions.set(jobId, execution)
  const heartbeat = setInterval(() => {
    void touchJobTracking(jobId, (job.attemptsMade ?? 0) + 1)
  }, JOB_HEARTBEAT_MS)
  heartbeat.unref()

  try {
    return await execution
  } finally {
    clearInterval(heartbeat)
    if (activeExecutions.get(jobId) === execution) {
      activeExecutions.delete(jobId)
    }
  }
})