ComfyUI_frontend/src/stores/queueStore.ts

import _ from 'es-toolkit/compat'
import { defineStore } from 'pinia'
import { computed, ref, shallowRef, toRaw, toValue } from 'vue'

import { isCloud } from '@/platform/distribution/types'
import { reconcileHistory } from '@/platform/remote/comfyui/history/reconciliation'
import { getWorkflowFromHistory } from '@/platform/workflow/cloud'
import type {
  ComfyWorkflowJSON,
  NodeId
} from '@/platform/workflow/validation/schemas/workflowSchema'
import type {
  HistoryTaskItem,
  ResultItem,
  StatusWsMessageStatus,
  TaskItem,
  TaskOutput,
  TaskPrompt,
  TaskStatus,
  TaskType
} from '@/schemas/apiSchema'
import { api } from '@/scripts/api'
import type { ComfyApp } from '@/scripts/app'
import { useExtensionService } from '@/services/extensionService'
import { useNodeOutputStore } from '@/stores/imagePreviewStore'
import { useExecutionStore } from '@/stores/executionStore'
import { getMediaTypeFromFilename } from '@/utils/formatUtil'

// Task type used in the API.
type APITaskType = 'queue' | 'history'

enum TaskItemDisplayStatus {
  Running = 'Running',
  Pending = 'Pending',
  Completed = 'Completed',
  Failed = 'Failed',
  Cancelled = 'Cancelled'
}

export class ResultItemImpl {
  filename: string
  subfolder: string
  type: string

  nodeId: NodeId
  // 'audio' | 'images' | ...
  mediaType: string

  // VHS output specific fields
  format?: string
  frame_rate?: number

  constructor(obj: Record<string, any>) {
    this.filename = obj.filename ?? ''
    this.subfolder = obj.subfolder ?? ''
    this.type = obj.type ?? ''

    this.nodeId = obj.nodeId
    this.mediaType = obj.mediaType

    this.format = obj.format
    this.frame_rate = obj.frame_rate
  }

  get urlParams(): URLSearchParams {
    const params = new URLSearchParams()
    params.set('filename', this.filename)
    params.set('type', this.type)
    params.set('subfolder', this.subfolder)

    if (this.format) {
      params.set('format', this.format)
    }
    if (this.frame_rate) {
      params.set('frame_rate', this.frame_rate.toString())
    }
    return params
  }

  /**
   * VHS advanced preview URL. `/viewvideo` endpoint is provided by VHS node.
   *
   * `/viewvideo` always returns a webm file.
   */
  get vhsAdvancedPreviewUrl(): string {
    return api.apiURL('/viewvideo?' + this.urlParams)
  }

  get url(): string {
    return api.apiURL('/view?' + this.urlParams)
  }

  get urlWithTimestamp(): string {
    return `${this.url}&t=${+new Date()}`
  }

  get isVhsFormat(): boolean {
    return !!this.format && !!this.frame_rate
  }

  get htmlVideoType(): string | undefined {
    if (this.isWebm) {
      return 'video/webm'
    }
    if (this.isMp4) {
      return 'video/mp4'
    }

    if (this.isVhsFormat) {
      if (this.format?.endsWith('webm')) {
        return 'video/webm'
      }
      if (this.format?.endsWith('mp4')) {
        return 'video/mp4'
      }
    }
    return undefined
  }

  get htmlAudioType(): string | undefined {
    if (this.isMp3) {
      return 'audio/mpeg'
    }
    if (this.isWav) {
      return 'audio/wav'
    }
    if (this.isOgg) {
      return 'audio/ogg'
    }
    if (this.isFlac) {
      return 'audio/flac'
    }
    return undefined
  }

  get isGif(): boolean {
    return this.filename.endsWith('.gif')
  }

  get isWebp(): boolean {
    return this.filename.endsWith('.webp')
  }

  get isWebm(): boolean {
    return this.filename.endsWith('.webm')
  }

  get isMp4(): boolean {
    return this.filename.endsWith('.mp4')
  }

  get isVideoBySuffix(): boolean {
    return this.isWebm || this.isMp4
  }

  get isImageBySuffix(): boolean {
    return this.isGif || this.isWebp
  }

  get isMp3(): boolean {
    return this.filename.endsWith('.mp3')
  }

  get isWav(): boolean {
    return this.filename.endsWith('.wav')
  }

  get isOgg(): boolean {
    return this.filename.endsWith('.ogg')
  }

  get isFlac(): boolean {
    return this.filename.endsWith('.flac')
  }

  get isAudioBySuffix(): boolean {
    return this.isMp3 || this.isWav || this.isOgg || this.isFlac
  }

  get isVideo(): boolean {
    const isVideoByType =
      this.mediaType === 'video' || !!this.format?.startsWith('video/')
    return (
      this.isVideoBySuffix ||
      (isVideoByType && !this.isImageBySuffix && !this.isAudioBySuffix)
    )
  }

  get isImage(): boolean {
    return (
      this.isImageBySuffix ||
      (this.mediaType === 'images' &&
        !this.isVideoBySuffix &&
        !this.isAudioBySuffix)
    )
  }

  get isAudio(): boolean {
    const isAudioByType =
      this.mediaType === 'audio' || !!this.format?.startsWith('audio/')
    return (
      this.isAudioBySuffix ||
      (isAudioByType && !this.isImageBySuffix && !this.isVideoBySuffix)
    )
  }

  get is3D(): boolean {
    return getMediaTypeFromFilename(this.filename) === '3D'
  }

  get supportsPreview(): boolean {
    return this.isImage || this.isVideo || this.isAudio || this.is3D
  }
}

export class TaskItemImpl {
  readonly taskType: TaskType
  readonly prompt: TaskPrompt
  readonly status?: TaskStatus
  readonly outputs: TaskOutput
  readonly flatOutputs: ReadonlyArray<ResultItemImpl>

  constructor(
    taskType: TaskType,
    prompt: TaskPrompt,
    status?: TaskStatus,
    outputs?: TaskOutput,
    flatOutputs?: ReadonlyArray<ResultItemImpl>
  ) {
    this.taskType = taskType
    this.prompt = prompt
    this.status = status
    // Remove animated outputs from the outputs object
    // outputs.animated is an array of boolean values that indicates if the images
    // array in the result are animated or not.
    // The queueStore does not use this information.
    // It is part of the legacy API response. We should redesign the backend API.
    // https://github.com/Comfy-Org/ComfyUI_frontend/issues/2739
    this.outputs = _.mapValues(outputs ?? {}, (nodeOutputs) =>
      _.omit(nodeOutputs, 'animated')
    )
    this.flatOutputs = flatOutputs ?? this.calculateFlatOutputs()
  }

  calculateFlatOutputs(): ReadonlyArray<ResultItemImpl> {
    if (!this.outputs) {
      return []
    }
    return Object.entries(this.outputs).flatMap(([nodeId, nodeOutputs]) =>
      Object.entries(nodeOutputs).flatMap(([mediaType, items]) =>
        (items as ResultItem[]).map(
          (item: ResultItem) =>
            new ResultItemImpl({
              ...item,
              nodeId,
              mediaType
            })
        )
      )
    )
  }

  get previewOutput(): ResultItemImpl | undefined {
    return (
      this.flatOutputs.find(
        // Prefer saved media files over the temp previews
        (output) => output.type === 'output' && output.supportsPreview
      ) ?? this.flatOutputs.find((output) => output.supportsPreview)
    )
  }

  get apiTaskType(): APITaskType {
    switch (this.taskType) {
      case 'Running':
      case 'Pending':
        return 'queue'
      case 'History':
        return 'history'
    }
  }

  get key() {
    return this.promptId + this.displayStatus
  }

  get queueIndex() {
    return this.prompt[0]
  }

  get promptId() {
    return this.prompt[1]
  }

  get promptInputs() {
    return this.prompt[2]
  }

  get extraData() {
    return this.prompt[3]
  }

  get outputsToExecute() {
    return this.prompt[4]
  }

  get extraPngInfo() {
    return this.extraData.extra_pnginfo
  }

  get clientId() {
    return this.extraData.client_id
  }

  get workflow(): ComfyWorkflowJSON | undefined {
    return this.extraPngInfo?.workflow
  }

  get messages() {
    return this.status?.messages || []
  }

  /**
   * Server-provided creation time in milliseconds, when available.
   *
   * Sources:
   * - Queue: 5th tuple element may be a metadata object with { create_time }.
   * - History (Cloud V2): Adapter injects create_time into prompt[3].extra_data.
   */
  get createTime(): number | undefined {
    const extra = (this.extraData as any) || {}
    const fromExtra =
      typeof extra.create_time === 'number' ? extra.create_time : undefined
    if (typeof fromExtra === 'number') return fromExtra

    return undefined
  }

  get interrupted() {
    return _.some(
      this.messages,
      (message) => message[0] === 'execution_interrupted'
    )
  }

  get isHistory() {
    return this.taskType === 'History'
  }

  get isRunning() {
    return this.taskType === 'Running'
  }

  get displayStatus(): TaskItemDisplayStatus {
    switch (this.taskType) {
      case 'Running':
        return TaskItemDisplayStatus.Running
      case 'Pending':
        return TaskItemDisplayStatus.Pending
      case 'History':
        if (this.interrupted) return TaskItemDisplayStatus.Cancelled

        switch (this.status!.status_str) {
          case 'success':
            return TaskItemDisplayStatus.Completed
          case 'error':
            return TaskItemDisplayStatus.Failed
        }
    }
  }

  get executionStartTimestamp() {
    const message = this.messages.find(
      (message) => message[0] === 'execution_start'
    )
    return message ? message[1].timestamp : undefined
  }

  get executionEndTimestamp() {
    const messages = this.messages.filter((message) =>
      [
        'execution_success',
        'execution_interrupted',
        'execution_error'
      ].includes(message[0])
    )
    if (!messages.length) {
      return undefined
    }
    return _.max(messages.map((message) => message[1].timestamp))
  }

  get executionTime() {
    if (!this.executionStartTimestamp || !this.executionEndTimestamp) {
      return undefined
    }
    return this.executionEndTimestamp - this.executionStartTimestamp
  }

  get executionTimeInSeconds() {
    return this.executionTime !== undefined
      ? this.executionTime / 1000
      : undefined
  }

  public async loadWorkflow(app: ComfyApp) {
    let workflowData = this.workflow

    if (isCloud && !workflowData && this.isHistory) {
      workflowData = await getWorkflowFromHistory(
        (url) => app.api.fetchApi(url),
        this.promptId
      )
    }

    if (!workflowData) {
      return
    }

    await app.loadGraphData(toRaw(workflowData))

    if (!this.outputs) {
      return
    }

    const nodeOutputsStore = useNodeOutputStore()
    const rawOutputs = toRaw(this.outputs)
    for (const nodeExecutionId in rawOutputs) {
      nodeOutputsStore.setNodeOutputsByExecutionId(
        nodeExecutionId,
        rawOutputs[nodeExecutionId]
      )
    }
    useExtensionService().invokeExtensions(
      'onNodeOutputsUpdated',
      app.nodeOutputs
    )
  }

  public flatten(): TaskItemImpl[] {
    if (this.displayStatus !== TaskItemDisplayStatus.Completed) {
      return [this]
    }

    return this.flatOutputs.map(
      (output: ResultItemImpl, i: number) =>
        new TaskItemImpl(
          this.taskType,
          [
            this.queueIndex,
            `${this.promptId}-${i}`,
            this.promptInputs,
            this.extraData,
            this.outputsToExecute
          ],
          this.status,
          {
            [output.nodeId]: {
              [output.mediaType]: [output]
            }
          },
          [output]
        )
    )
  }

  public toTaskItem(): TaskItem {
    const item: HistoryTaskItem = {
      taskType: 'History',
      prompt: this.prompt,
      status: this.status!,
      outputs: this.outputs
    }
    return item
  }
}

const sortNewestFirst = (a: TaskItemImpl, b: TaskItemImpl) =>
  b.queueIndex - a.queueIndex

const toTaskItemImpls = (tasks: TaskItem[]): TaskItemImpl[] =>
  tasks.map(
    (task) =>
      new TaskItemImpl(
        task.taskType,
        task.prompt,
        'status' in task ? task.status : undefined,
        'outputs' in task ? task.outputs : undefined
      )
  )

export const useQueueStore = defineStore('queue', () => {
  // Use shallowRef because TaskItemImpl instances are immutable and arrays are
  // replaced entirely (not mutated), so deep reactivity would waste performance
  const runningTasks = shallowRef<TaskItemImpl[]>([])
  const pendingTasks = shallowRef<TaskItemImpl[]>([])
  const historyTasks = shallowRef<TaskItemImpl[]>([])
  const maxHistoryItems = ref(64)
  const isLoading = ref(false)

  const tasks = computed<TaskItemImpl[]>(
    () =>
      [
        ...pendingTasks.value,
        ...runningTasks.value,
        ...historyTasks.value
      ] as TaskItemImpl[]
  )

  const flatTasks = computed<TaskItemImpl[]>(() =>
    tasks.value.flatMap((task: TaskItemImpl) => task.flatten())
  )

  const lastHistoryQueueIndex = computed<number>(() =>
    historyTasks.value.length ? historyTasks.value[0].queueIndex : -1
  )

  const hasPendingTasks = computed<boolean>(() => pendingTasks.value.length > 0)

  const update = async () => {
    isLoading.value = true
    try {
      const [queue, history] = await Promise.all([
        api.getQueue(),
        api.getHistory(maxHistoryItems.value)
      ])

      runningTasks.value = toTaskItemImpls(queue.Running).sort(sortNewestFirst)
      pendingTasks.value = toTaskItemImpls(queue.Pending).sort(sortNewestFirst)

      const currentHistory = toValue(historyTasks)

      const appearedTasks = [...pendingTasks.value, ...runningTasks.value]
      const executionStore = useExecutionStore()
      appearedTasks.forEach((task) => {
        const promptIdString = String(task.promptId)
        const workflowId = task.workflow?.id
        if (workflowId && promptIdString) {
          executionStore.registerPromptWorkflowIdMapping(
            promptIdString,
            workflowId
          )
        }
      })

      const items = reconcileHistory(
        history.History,
        currentHistory.map((impl) => impl.toTaskItem()),
        toValue(maxHistoryItems),
        toValue(lastHistoryQueueIndex)
      )

      // Reuse existing TaskItemImpl instances or create new
      const existingByPromptId = new Map(
        currentHistory.map((impl) => [impl.promptId, impl])
      )

      historyTasks.value = items.map(
        (item) =>
          existingByPromptId.get(item.prompt[1]) ?? toTaskItemImpls([item])[0]
      )
    } finally {
      isLoading.value = false
    }
  }

  const clear = async (
    targets: ('queue' | 'history')[] = ['queue', 'history']
  ) => {
    if (targets.length === 0) {
      return
    }
    await Promise.all(targets.map((type) => api.clearItems(type)))
    await update()
  }

  const deleteTask = async (task: TaskItemImpl) => {
    await api.deleteItem(task.apiTaskType, task.promptId)
    await update()
  }

  return {
    runningTasks,
    pendingTasks,
    historyTasks,
    maxHistoryItems,
    isLoading,

    tasks,
    flatTasks,
    lastHistoryQueueIndex,
    hasPendingTasks,

    update,
    clear,
    delete: deleteTask
  }
})

export const useQueuePendingTaskCountStore = defineStore(
  'queuePendingTaskCount',
  {
    state: () => ({
      count: 0
    }),
    actions: {
      update(e: CustomEvent<StatusWsMessageStatus>) {
        this.count = e.detail?.exec_info?.queue_remaining || 0
      }
    }
  }
)

export type AutoQueueMode = 'disabled' | 'instant' | 'change'

export const useQueueSettingsStore = defineStore('queueSettingsStore', {
  state: () => ({
    mode: 'disabled' as AutoQueueMode,
    batchCount: 1
  })
})