From 536786331be160008d67f4a66aa137e51af1b996 Mon Sep 17 00:00:00 2001
From: Makito <i@maki.to>
Date: Sat, 30 Aug 2025 18:40:58 +0900
Subject: [PATCH] refactor(stage-ui): make queue framework-agnostic (#526)

* refactor(stage-ui): make queue framework-agnostic

* fix(stage-ui): update usages

* fix(stage-ui): trust the shift

* fix(stage-ui): typecheck

* fix(stage-ui): keep caught error unknown
---
 apps/stage-web/src/pages/queue.vue            |  14 +-
 .../src/pages/test/queues/delays.vue          |   4 +-
 .../src/pages/test/queues/emotions.vue        |   8 +-
 .../src/pages/test/queues/messages.vue        |   8 +-
 .../Providers/SpeechStreamingPlayground.vue   |  10 +-
 .../stage-ui/src/components/Scenes/Stage.vue  |  25 ++--
 packages/stage-ui/src/composables/index.ts    |   1 -
 packages/stage-ui/src/composables/queue.ts    | 123 ------------------
 packages/stage-ui/src/composables/queues.ts   |  15 +--
 packages/stage-ui/src/stores/chat.ts          |   8 +-
 packages/stage-ui/src/utils/queue.ts          | 102 +++++++++++++++
 packages/stage-ui/src/utils/tts.ts            |   4 +-
 12 files changed, 147 insertions(+), 175 deletions(-)
 delete mode 100644 packages/stage-ui/src/composables/queue.ts
 create mode 100644 packages/stage-ui/src/utils/queue.ts

diff --git a/apps/stage-web/src/pages/queue.vue b/apps/stage-web/src/pages/queue.vue
index 8480fb4f7..bfbd16803 100644
--- a/apps/stage-web/src/pages/queue.vue
+++ b/apps/stage-web/src/pages/queue.vue
@@ -1,11 +1,11 @@
 <script setup lang="ts">
 import { sleep } from '@moeru/std'
-import { useQueue } from '@proj-airi/stage-ui/composables/queue'
+import { createQueue } from '@proj-airi/stage-ui/utils/queue'
 import { onMounted, ref } from 'vue'
 
 const temp = ref<string>('')
 
-const audioQueue = useQueue<string>({
+const audioQueue = createQueue<string>({
   handlers: [
     async (text) => {
       // eslint-disable-next-line no-console
@@ -13,16 +13,16 @@ const audioQueue = useQueue<string>({
     },
   ],
 })
-const ttsQueue = useQueue<string>({
+const ttsQueue = createQueue<string>({
   handlers: [
     async (ctx) => {
       // eslint-disable-next-line no-console
       console.log('ready to stream speech audio for', ctx)
-      audioQueue.add(ctx.data)
+      audioQueue.enqueue(ctx.data)
     },
   ],
 })
-const textQueue = useQueue<string>({
+const textQueue = createQueue<string>({
   handlers: [
     async (ctx) => {
       const endMarker = ['.', '?', '!']
@@ -40,7 +40,7 @@ const textQueue = useQueue<string>({
         const afterPeriod = ctx.data.slice(periodIndex + 1)
 
         temp.value += beforePeriod
-        ttsQueue.add(temp.value.trim())
+        ttsQueue.enqueue(temp.value.trim())
         temp.value = afterPeriod
 
         newEndPartDiscovered = true
@@ -129,7 +129,7 @@ const textParts = [
 async function mockTextPartsStreamHandler() {
   for (const part of textParts) {
     await sleep(100)
-    textQueue.add(part)
+    textQueue.enqueue(part)
   }
 }
 
diff --git a/apps/stage-web/src/pages/test/queues/delays.vue b/apps/stage-web/src/pages/test/queues/delays.vue
index d2da56faf..038a5ac31 100644
--- a/apps/stage-web/src/pages/test/queues/delays.vue
+++ b/apps/stage-web/src/pages/test/queues/delays.vue
@@ -26,9 +26,9 @@ function onSendMessage() {
   processing.value = true
   const tokens = messageInput.value.split('')
   for (const token of tokens)
-    delaysQueue.add(token)
+    delaysQueue.enqueue(token)
 
-  delaysQueue.add(llmInferenceEndToken)
+  delaysQueue.enqueue(llmInferenceEndToken)
   messageInput.value = ''
   processing.value = false
 }
diff --git a/apps/stage-web/src/pages/test/queues/emotions.vue b/apps/stage-web/src/pages/test/queues/emotions.vue
index d9cda7518..da694fb8c 100644
--- a/apps/stage-web/src/pages/test/queues/emotions.vue
+++ b/apps/stage-web/src/pages/test/queues/emotions.vue
@@ -1,9 +1,9 @@
 <script setup lang="ts">
 import type { Emotion } from '@proj-airi/stage-ui/constants/emotions'
 
-import { useQueue } from '@proj-airi/stage-ui/composables/queue'
 import { useEmotionsMessageQueue } from '@proj-airi/stage-ui/composables/queues'
 import { llmInferenceEndToken } from '@proj-airi/stage-ui/constants'
+import { createQueue } from '@proj-airi/stage-ui/utils/queue'
 import { Textarea } from '@proj-airi/ui'
 import { ref } from 'vue'
 
@@ -12,7 +12,7 @@ const messagesProcessed = ref<string[]>([])
 const emotionsProcessed = ref<string[]>([])
 const processing = ref<boolean>(false)
 
-const emotionsQueue = useQueue<Emotion>({
+const emotionsQueue = createQueue<Emotion>({
   handlers: [
     async (ctx) => {
       emotionsProcessed.value.push(ctx.data)
@@ -26,9 +26,9 @@ function onSendMessage() {
   processing.value = true
   const tokens = messageInput.value.split('')
   for (const token of tokens)
-    emotionMessageContentQueue.add(token)
+    emotionMessageContentQueue.enqueue(token)
 
-  emotionMessageContentQueue.add(llmInferenceEndToken)
+  emotionMessageContentQueue.enqueue(llmInferenceEndToken)
   messageInput.value = ''
   processing.value = false
 }
diff --git a/apps/stage-web/src/pages/test/queues/messages.vue b/apps/stage-web/src/pages/test/queues/messages.vue
index 1d81db3d5..0fb34a882 100644
--- a/apps/stage-web/src/pages/test/queues/messages.vue
+++ b/apps/stage-web/src/pages/test/queues/messages.vue
@@ -1,7 +1,7 @@
 <script setup lang="ts">
-import { useQueue } from '@proj-airi/stage-ui/composables/queue'
 import { useMessageContentQueue } from '@proj-airi/stage-ui/composables/queues'
 import { llmInferenceEndToken } from '@proj-airi/stage-ui/constants'
+import { createQueue } from '@proj-airi/stage-ui/utils/queue'
 import { Textarea } from '@proj-airi/ui'
 import { ref } from 'vue'
 
@@ -9,7 +9,7 @@ const messageInput = ref<string>('')
 const ttsProcessed = ref<string[]>([])
 const processing = ref<boolean>(false)
 
-const ttsQueue = useQueue<string>({
+const ttsQueue = createQueue<string>({
   handlers: [
     async (ctx) => {
       ttsProcessed.value.push(ctx.data)
@@ -26,9 +26,9 @@ async function onSendMessage() {
   // await sleep(100)
   // messageContentQueue.add(token)
   // }
-  messageContentQueue.add(messageInput.value)
+  messageContentQueue.enqueue(messageInput.value)
 
-  messageContentQueue.add(llmInferenceEndToken)
+  messageContentQueue.enqueue(llmInferenceEndToken)
   messageInput.value = ''
   processing.value = false
 }
diff --git a/packages/stage-ui/src/components/Scenarios/Providers/SpeechStreamingPlayground.vue b/packages/stage-ui/src/components/Scenarios/Providers/SpeechStreamingPlayground.vue
index 4fffa6b67..fe46827df 100644
--- a/packages/stage-ui/src/components/Scenarios/Providers/SpeechStreamingPlayground.vue
+++ b/packages/stage-ui/src/components/Scenarios/Providers/SpeechStreamingPlayground.vue
@@ -4,9 +4,9 @@ import type { TTSInputChunk } from '../../../utils/tts'
 import { animate } from 'animejs'
 import { ref } from 'vue'
 
-import { useQueue } from '../../../composables/queue'
 import { useMessageContentQueue } from '../../../composables/queues'
 import { useAudioContext } from '../../../stores/audio'
+import { createQueue } from '../../../utils/queue'
 import { chunkTTSInput } from '../../../utils/tts'
 
 const props = defineProps<{
@@ -21,7 +21,7 @@ const nowSpeaking = ref(false)
 const ttsInputChunks = ref<TTSInputChunk[]>([])
 const speechGenerationIndex = ref(-1)
 
-const audioQueue = useQueue<{ audioBuffer: AudioBuffer, text: string }>({
+const audioQueue = createQueue<{ audioBuffer: AudioBuffer, text: string }>({
   handlers: [
     (ctx) => {
       return new Promise((resolve) => {
@@ -54,19 +54,19 @@ async function handleSpeechGeneration(ctx: { data: string }) {
 
     // Decode the ArrayBuffer into an AudioBuffer
     const audioBuffer = await audioContext.decodeAudioData(res)
-    await audioQueue.add({ audioBuffer, text: ctx.data })
+    audioQueue.enqueue({ audioBuffer, text: ctx.data })
   }
   catch (error) {
     console.error('Speech generation failed:', error)
   }
 }
 
-const ttsQueue = useQueue<string>({ handlers: [handleSpeechGeneration] })
+const ttsQueue = createQueue<string>({ handlers: [handleSpeechGeneration] })
 
 const messageContentQueue = useMessageContentQueue(ttsQueue)
 
 async function testStreaming() {
-  await messageContentQueue.add(props.text)
+  messageContentQueue.enqueue(props.text)
 }
 
 async function testChunking() {
diff --git a/packages/stage-ui/src/components/Scenes/Stage.vue b/packages/stage-ui/src/components/Scenes/Stage.vue
index 8095494a0..514e3cf1a 100644
--- a/packages/stage-ui/src/components/Scenes/Stage.vue
+++ b/packages/stage-ui/src/components/Scenes/Stage.vue
@@ -17,7 +17,6 @@ import { onMounted, onUnmounted, ref } from 'vue'
 import Live2DScene from './Live2D.vue'
 import VRMScene from './VRM.vue'
 
-import { useQueue } from '../../composables/queue'
 import { useDelayMessageQueue, useEmotionsMessageQueue, useMessageContentQueue } from '../../composables/queues'
 import { llmInferenceEndToken } from '../../constants'
 import { EMOTION_EmotionMotionName_value, EMOTION_VRMExpressionName_value, EmotionThinkMotionName } from '../../constants/emotions'
@@ -28,6 +27,7 @@ import { useSpeechStore } from '../../stores/modules/speech'
 import { useProvidersStore } from '../../stores/providers'
 import { useSettings } from '../../stores/settings'
 import { useVRM } from '../../stores/vrm'
+import { createQueue } from '../../utils/queue'
 
 withDefaults(defineProps<{
   paused?: boolean
@@ -76,7 +76,7 @@ const nowSpeaking = ref(false)
 const lipSyncStarted = ref(false)
 let currentAudioSource: AudioBufferSourceNode | null = null
 
-const audioQueue = useQueue<{ audioBuffer: AudioBuffer, text: string }>({
+const audioQueue = createQueue<{ audioBuffer: AudioBuffer, text: string }>({
   handlers: [
     (ctx) => {
       return new Promise((resolve) => {
@@ -150,29 +150,24 @@ async function handleSpeechGeneration(ctx: { data: string }) {
 
     // Decode the ArrayBuffer into an AudioBuffer
     const audioBuffer = await audioContext.decodeAudioData(res)
-    await audioQueue.add({ audioBuffer, text: ctx.data })
+    audioQueue.enqueue({ audioBuffer, text: ctx.data })
   }
   catch (error) {
     console.error('Speech generation failed:', error)
   }
 }
 
-const ttsQueue = useQueue<string>({
+const ttsQueue = createQueue<string>({
   handlers: [
     handleSpeechGeneration,
   ],
 })
 
-ttsQueue.on('add', (content) => {
-  // eslint-disable-next-line no-console
-  console.debug('ttsQueue added', content)
-})
-
 const messageContentQueue = useMessageContentQueue(ttsQueue)
 
 const { currentMotion } = storeToRefs(useLive2d())
 
-const emotionsQueue = useQueue<Emotion>({
+const emotionsQueue = createQueue<Emotion>({
   handlers: [
     async (ctx) => {
       if (stageModelRenderer.value === 'vrm') {
@@ -232,7 +227,7 @@ onBeforeMessageComposed(async () => {
     catch {}
     currentAudioSource = null
   }
-  audioQueue.queue.value = []
+  audioQueue.clear()
   setupAnalyser()
   setupLipSync()
 })
@@ -242,16 +237,16 @@ onBeforeSend(async () => {
 })
 
 onTokenLiteral(async (literal) => {
-  await messageContentQueue.add(literal)
+  messageContentQueue.enqueue(literal)
 })
 
 onTokenSpecial(async (special) => {
-  await delaysQueue.add(special)
-  await emotionMessageContentQueue.add(special)
+  delaysQueue.enqueue(special)
+  emotionMessageContentQueue.enqueue(special)
 })
 
 onStreamEnd(async () => {
-  await delaysQueue.add(llmInferenceEndToken)
+  delaysQueue.enqueue(llmInferenceEndToken)
 })
 
 onAssistantResponseEnd(async (_message) => {
diff --git a/packages/stage-ui/src/composables/index.ts b/packages/stage-ui/src/composables/index.ts
index 5199bf051..c69895225 100644
--- a/packages/stage-ui/src/composables/index.ts
+++ b/packages/stage-ui/src/composables/index.ts
@@ -2,7 +2,6 @@ export * from './audio'
 export * from './llmmarkerParser'
 export * from './markdown'
 export * from './micvad'
-export * from './queue'
 export * from './queues'
 export * from './vrm'
 export * from './whisper'
diff --git a/packages/stage-ui/src/composables/queue.ts b/packages/stage-ui/src/composables/queue.ts
deleted file mode 100644
index d7de3896c..000000000
--- a/packages/stage-ui/src/composables/queue.ts
+++ /dev/null
@@ -1,123 +0,0 @@
-import type { Ref } from 'vue'
-
-import { ref } from 'vue'
-
-export interface HandlerContext<T> {
-  data: T
-  itemsToBeProcessed: () => number
-  emit: (eventName: string, ...params: any[]) => void
-}
-
-export interface Events<T> {
-  add: Array<(payload: T) => void>
-  pick: Array<(payload: T) => void>
-  processing: Array<(payload: T, handler: (param: HandlerContext<T>) => Promise<any>) => void>
-  error: Array<(payload: T, error: Error, handler: (param: HandlerContext<T>) => Promise<any>) => void>
-  processed: Array<<R>(payload: T, result: R, handler: (param: HandlerContext<T>) => Promise<any>) => void>
-  done: Array<(payload: T) => void>
-}
-
-export function useQueue<T>(options: {
-  handlers: Array<(ctx: HandlerContext<T>) => Promise<void>>
-}) {
-  const queue = ref<T[]>([]) as Ref<T[]>
-  const isProcessing = ref(false)
-  const internalEventListeners: Events<T> = {
-    add: [],
-    pick: [],
-    processing: [],
-    error: [],
-    processed: [],
-    done: [],
-  }
-  const internalHandlerEventListeners: Record<string, Array<(...params: any[]) => void>> = {}
-
-  function on<E extends keyof Events<T>>(eventName: E, Listener: Events<T>[E][number]) {
-    internalEventListeners[eventName].push(Listener as any)
-  }
-
-  function emit<E extends keyof Events<T>>(eventName: E, ...params: Parameters<Events<T>[E][number]>) {
-    const handlers = internalEventListeners[eventName] as Events<T>[E]
-    handlers.forEach((handler) => {
-      (handler as any)(...params)
-    })
-  }
-
-  function onHandlerEvent(eventName: string, handler: (...params: any[]) => void) {
-    internalHandlerEventListeners[eventName] = internalHandlerEventListeners[eventName] || []
-    internalHandlerEventListeners[eventName].push(handler)
-  }
-
-  function emitHandlerEvent(eventName: string, ...params: any[]) {
-    const handlers = internalHandlerEventListeners[eventName] || []
-    handlers.forEach((handler) => {
-      handler(...params)
-    })
-  }
-
-  async function add(payload: T) {
-    queue.value.push(payload)
-    emit('add', payload)
-  }
-
-  function pick() {
-    const payload = queue.value.shift()
-    if (!payload)
-      return
-
-    emit('pick', payload)
-    return payload
-  }
-
-  // Listener for item add / enqueue event
-  async function addItemListener() {
-    if (isProcessing.value)
-      return
-
-    const payload = pick()
-    if (!payload)
-      return
-
-    isProcessing.value = true
-
-    for (const handler of options.handlers) {
-      // If there is a need to register customised listener for processing event, then this line of code should be rewritten
-      // handlers as the input parameter is only designed for the add event
-      emit('processing', payload, handler)
-      try {
-        // Use handler to deal with the newly enqueued item
-        const result = await handler({ data: payload, itemsToBeProcessed: () => queue.value.length, emit: emitHandlerEvent })
-        // If there is a need to register customised listener for processing event, then this line of code should be rewritten
-        // handlers as the input parameter is only designed for the add event
-        emit('processed', payload, result, handler)
-      }
-      catch (err) {
-        // If there is a need to register customised listener for processing event, then this line of code should be rewritten
-        // handlers as the input parameter is only designed for the add event
-        emit('error', payload, err as Error, handler)
-        continue
-      }
-    }
-
-    isProcessing.value = false
-    emit('done', payload)
-
-    // Process next item if any
-    if (queue.value.length > 0)
-      addItemListener()
-  }
-
-  on('add', addItemListener)
-  // Lilia: I'm not sure why do we need to register handleItem to 'done', if queue is not empty, addItemListner will continue to call addItemListner. Calling this function again when 'done' event is triggered will only lead to payload = undefined (since queue is already empty) and return
-  // Maybe leave 'done' event to be registered other customised listeners would be more reasonable
-  // on('done', addItemListener)
-
-  return {
-    add,
-    on,
-    onHandlerEvent,
-    queue,
-  }
-}
-
-export type UseQueueReturn<T> = ReturnType<typeof useQueue<T>>
diff --git a/packages/stage-ui/src/composables/queues.ts b/packages/stage-ui/src/composables/queues.ts
index a44967566..cb0633146 100644
--- a/packages/stage-ui/src/composables/queues.ts
+++ b/packages/stage-ui/src/composables/queues.ts
@@ -1,12 +1,12 @@
 import type { Emotion } from '../constants/emotions'
-import type { UseQueueReturn } from './queue'
+import type { UseQueueReturn } from '../utils/queue'
 
 import { sleep } from '@moeru/std'
 
 import { EMOTION_VALUES } from '../constants/emotions'
+import { createQueue } from '../utils/queue'
 import { createControllableStream } from '../utils/stream'
 import { chunkToTTSQueue } from '../utils/tts'
-import { useQueue } from './queue'
 
 export function useEmotionsMessageQueue(emotionsQueue: UseQueueReturn<Emotion>) {
   function splitEmotion(content: string) {
@@ -27,14 +27,13 @@ export function useEmotionsMessageQueue(emotionsQueue: UseQueueReturn<Emotion>)
     }
   }
 
-  return useQueue<string>({
+  return createQueue<string>({
     handlers: [
       async (ctx) => {
         // if the message is an emotion, push the last content to the message queue
         if (EMOTION_VALUES.includes(ctx.data as Emotion)) {
           ctx.emit('emotion', ctx.data as Emotion)
-          await emotionsQueue.add(ctx.data as Emotion)
-
+          emotionsQueue.enqueue(ctx.data as Emotion)
           return
         }
 
@@ -44,7 +43,7 @@ export function useEmotionsMessageQueue(emotionsQueue: UseQueueReturn<Emotion>)
           const { ok, emotion } = splitEmotion(ctx.data)
           if (ok) {
             ctx.emit('emotion', emotion)
-            await emotionsQueue.add(emotion)
+            emotionsQueue.enqueue(emotion)
           }
         }
       },
@@ -87,7 +86,7 @@ export function useDelayMessageQueue() {
     }
   }
 
-  return useQueue<string>({
+  return createQueue<string>({
     handlers: [
       async (ctx) => {
         // iterate through the message to find the emotions
@@ -107,7 +106,7 @@ export function useMessageContentQueue(ttsQueue: UseQueueReturn<string>) {
 
   chunkToTTSQueue(stream.getReader(), ttsQueue)
 
-  return useQueue<string>({
+  return createQueue<string>({
     handlers: [
       async (ctx) => {
         controller.enqueue(encoder.encode(ctx.data))
diff --git a/packages/stage-ui/src/stores/chat.ts b/packages/stage-ui/src/stores/chat.ts
index 9905f6356..ad01fd21b 100644
--- a/packages/stage-ui/src/stores/chat.ts
+++ b/packages/stage-ui/src/stores/chat.ts
@@ -7,9 +7,9 @@ import type { ChatAssistantMessage, ChatMessage, ChatSlices } from '../types/cha
 import { defineStore, storeToRefs } from 'pinia'
 import { ref, toRaw } from 'vue'
 
-import { useQueue } from '../composables'
 import { useLlmmarkerParser } from '../composables/llmmarkerParser'
 import { useLLM } from '../stores/llm'
+import { createQueue } from '../utils/queue'
 import { TTS_FLUSH_INSTRUCTION } from '../utils/tts'
 import { useAiriCardStore } from './modules'
 
@@ -144,7 +144,7 @@ export const useChatStore = defineStore('chat', () => {
         minLiteralEmitLength: 24, // Avoid emitting literals too fast. This is a magic number and can be changed later.
       })
 
-      const toolCallQueue = useQueue<ChatSlices>({
+      const toolCallQueue = createQueue<ChatSlices>({
         handlers: [
           async (ctx) => {
             if (ctx.data.type === 'tool-call') {
@@ -184,13 +184,13 @@ export const useChatStore = defineStore('chat', () => {
         headers,
         async onStreamEvent(event: StreamEvent) {
           if (event.type === 'tool-call') {
-            toolCallQueue.add({
+            toolCallQueue.enqueue({
               type: 'tool-call',
               toolCall: event,
             })
           }
           else if (event.type === 'tool-result') {
-            toolCallQueue.add({
+            toolCallQueue.enqueue({
               type: 'tool-call-result',
               id: event.toolCallId,
               result: event.result,
diff --git a/packages/stage-ui/src/utils/queue.ts b/packages/stage-ui/src/utils/queue.ts
new file mode 100644
index 000000000..618885a0e
--- /dev/null
+++ b/packages/stage-ui/src/utils/queue.ts
@@ -0,0 +1,102 @@
+export interface HandlerContext<T> {
+  data: T
+  emit: (eventName: string, ...params: any[]) => void
+}
+
+export interface Events<T> {
+  enqueue: Array<(payload: T, queueLength: number) => void>
+  dequeue: Array<(payload: T, queueLength: number) => void>
+  process: Array<(payload: T, handler: (param: HandlerContext<T>) => Promise<any>) => void>
+  error: Array<(payload: T, error: unknown, handler: (param: HandlerContext<T>) => Promise<any>) => void>
+  result: Array<<R>(payload: T, result: R, handler: (param: HandlerContext<T>) => Promise<any>) => void>
+  drain: Array<() => void>
+}
+
+export function createQueue<T>(options: {
+  handlers: Array<(ctx: HandlerContext<T>) => Promise<void>>
+}) {
+  const queue: T[] = []
+  let drainTask: Promise<any> | undefined
+
+  const internalEventListeners: Events<T> = {
+    enqueue: [],
+    dequeue: [],
+    process: [],
+    error: [],
+    result: [],
+    drain: [],
+  }
+  const internalHandlerEventListeners: Record<string, Array<(...params: any[]) => void>> = {}
+
+  function on<E extends keyof Events<T>>(eventName: E, listener: Events<T>[E][number]) {
+    internalEventListeners[eventName].push(listener as any)
+  }
+
+  function emit<E extends keyof Events<T>>(eventName: E, ...params: Parameters<Events<T>[E][number]>) {
+    const listeners = internalEventListeners[eventName] as Events<T>[E]
+    listeners.forEach(listener => (listener as any)(...params))
+  }
+
+  function onHandlerEvent(eventName: string, listener: (...params: any[]) => void) {
+    internalHandlerEventListeners[eventName] = internalHandlerEventListeners[eventName] || []
+    internalHandlerEventListeners[eventName].push(listener)
+  }
+
+  function emitHandlerEvent(eventName: string, ...params: any[]) {
+    const listeners = internalHandlerEventListeners[eventName] || []
+    listeners.forEach(listener => listener(...params))
+  }
+
+  function enqueue(payload: T) {
+    queue.push(payload)
+    emit('enqueue', payload, queue.length)
+    if (!drainTask) {
+      drainTask = drain()
+    }
+  }
+
+  function clear() {
+    queue.length = 0
+  }
+
+  // Internal
+  // Drain the queue and call the handlers with each dequeued item
+  async function drain() {
+    while (queue.length > 0) {
+      // The async func should never yield at here, and shift() should
+      // always return an item from the queue.
+      // We cannot check for `undefined` here, because there could be
+      // someone using the queue to enqueue and dequeue `undefined`.
+      const payload = queue.shift() as T
+      emit('dequeue', payload, queue.length)
+      for (const handler of options.handlers) {
+        emit('process', payload, handler)
+        try {
+          const result = await handler({ data: payload, emit: emitHandlerEvent })
+          emit('result', payload, result, handler)
+        }
+        catch (err) {
+          // Keep `unknown` and let the event listener handle the error type
+          emit('error', payload, err, handler)
+          continue
+        }
+      }
+    }
+    emit('drain')
+    drainTask = undefined
+  }
+
+  function length() {
+    return queue.length
+  }
+
+  return {
+    enqueue,
+    clear,
+    length,
+    on,
+    onHandlerEvent,
+  }
+}
+
+export type UseQueueReturn<T> = ReturnType<typeof createQueue<T>>
diff --git a/packages/stage-ui/src/utils/tts.ts b/packages/stage-ui/src/utils/tts.ts
index f22c787ca..92f955220 100644
--- a/packages/stage-ui/src/utils/tts.ts
+++ b/packages/stage-ui/src/utils/tts.ts
@@ -1,6 +1,6 @@
 import type { ReaderLike } from 'clustr'
 
-import type { UseQueueReturn } from '../composables/queue'
+import type { UseQueueReturn } from './queue'
 
 import { readGraphemeClusters } from 'clustr'
 
@@ -207,7 +207,7 @@ export async function chunkToTTSQueue(reader: ReaderLike, queue: UseQueueReturn<
       // TODO: remove later
       // eslint-disable-next-line no-console
       console.debug('chunk to be pushed: ', chunk)
-      await queue.add(chunk.text)
+      queue.enqueue(chunk.text)
     }
   }
   catch (e) {