feat(logger): enhance logging functionality with tag support and message relaying

- Updated the ConsolaCompatibleLogger to include a tag parameter for better log categorization.
- Introduced a relayLogMessage function to facilitate message relaying with associated tags and log levels.
- Refactored the createBuilderLoggerAdapter function to utilize the new tagging system, improving log context and traceability.

Signed-off-by: Innei <tukon479@gmail.com>

Author: Innei

be/apps/core/src/modules/content/photo/builder/photo-builder-logger.adapter.ts

@@ -1,42 +1,58 @@
 import type { Logger as BuilderLogger } from '@afilmory/builder/logger/index.js'
+import { relayLogMessage } from '@afilmory/builder/logger/index.js'
 import type { PrettyLogger } from '@afilmory/framework'
 import type { ConsolaInstance } from 'consola'
 
 class ConsolaCompatibleLogger {
-  constructor(private readonly logger: PrettyLogger) {}
+  constructor(
+    private readonly logger: PrettyLogger,
+    private readonly tag: string,
+  ) {}
+
+  private forward(level: string, args: unknown[]): void {
+    relayLogMessage(this.tag, level, args)
+  }
 
   info(...args: unknown[]): void {
+    this.forward('info', args)
     this.logger.info(...args)
   }
 
   success(...args: unknown[]): void {
+    this.forward('success', args)
     this.logger.info(...args)
   }
 
   warn(...args: unknown[]): void {
+    this.forward('warn', args)
     this.logger.warn(...args)
   }
 
   error(...args: unknown[]): void {
+    this.forward('error', args)
     this.logger.error(...args)
   }
 
   log(...args: unknown[]): void {
+    this.forward('log', args)
     this.logger.log(...args)
   }
 
   debug(...args: unknown[]): void {
+    this.forward('debug', args)
     this.logger.debug(...args)
   }
 
   withTag(tag: string): ConsolaCompatibleLogger {
-    return new ConsolaCompatibleLogger(this.logger.extend(tag))
+    const modifier = String(tag ?? '')
+    const nextTag = modifier ? combineTags(this.tag, modifier) : this.tag
+    return new ConsolaCompatibleLogger(this.logger.extend(tag), nextTag)
   }
 }
 
 export function createBuilderLoggerAdapter(baseLogger: PrettyLogger): BuilderLogger {
   const createTaggedLogger = (tag: string): ConsolaCompatibleLogger =>
-    new ConsolaCompatibleLogger(baseLogger.extend(tag))
+    new ConsolaCompatibleLogger(baseLogger.extend(tag), tag)
 
   return {
     main: createTaggedLogger('PhotoBuilder:Main') as unknown as ConsolaInstance,
@@ -49,3 +65,13 @@ export function createBuilderLoggerAdapter(baseLogger: PrettyLogger): BuilderLog
     worker: (id: number) => createTaggedLogger(`PhotoBuilder:Worker-${id}`) as unknown as ConsolaInstance,
   }
 }
+
+function combineTags(parentTag: string, childTag: string): string {
+  if (!parentTag) {
+    return childTag
+  }
+  if (!childTag) {
+    return parentTag
+  }
+  return `${parentTag}/${childTag}`
+}

be/apps/core/src/modules/shared/http/sse.ts

@@ -1,4 +1,5 @@
 import type { Context } from 'hono'
+import { streamSSE } from 'hono/streaming'
 
 export interface CreateProgressSseResponseOptions<TEvent> {
   context: Context
@@ -22,108 +23,67 @@ export function createProgressSseResponse<TEvent>({
   heartbeatIntervalMs = DEFAULT_HEARTBEAT_MS,
   handler,
 }: CreateProgressSseResponseOptions<TEvent>): Response {
-  const encoder = new TextEncoder()
-  let cleanup: (() => void) | undefined
+  context.header('Cache-Control', 'no-cache, no-transform')
+  context.header('X-Accel-Buffering', 'no')
 
-  const stream = new ReadableStream<Uint8Array>({
-    start: (controller) => {
-      let closed = false
-      const rawRequest = context.req.raw
-      const abortSignal = rawRequest.signal
+  return streamSSE(context, async (stream) => {
+    const controller = new AbortController()
+    const abortSignal = controller.signal
 
-      let heartbeat: ReturnType<typeof setInterval> | undefined
-      let abortHandler: (() => void) | undefined
+    let heartbeat: ReturnType<typeof setInterval> | undefined
 
-      const cleanupInternal = () => {
-        if (heartbeat) {
-          clearInterval(heartbeat)
-          heartbeat = undefined
-        }
-
-        if (abortHandler) {
-          abortSignal.removeEventListener('abort', abortHandler)
-          abortHandler = undefined
-        }
-
-        if (!closed) {
-          closed = true
-          try {
-            controller.close()
-          } catch {
-            /* ignore */
-          }
-        }
-      }
-
-      const sendChunk = (chunk: string) => {
-        if (closed) {
-          return
-        }
-
-        try {
-          controller.enqueue(encoder.encode(chunk))
-        } catch {
-          cleanupInternal()
-          cleanup = undefined
-        }
+    const stopHeartbeat = () => {
+      if (heartbeat) {
+        clearInterval(heartbeat)
+        heartbeat = undefined
       }
+    }
 
-      const sendEvent = (event: TEvent) => {
-        sendChunk(`event: ${eventName}\ndata: ${JSON.stringify(event)}\n\n`)
+    stream.onAbort(() => {
+      if (!abortSignal.aborted) {
+        controller.abort()
       }
+      stopHeartbeat()
+    })
 
-      heartbeat = setInterval(() => {
-        sendChunk(`: keep-alive ${new Date().toISOString()}\n\n`)
-      }, heartbeatIntervalMs)
+    const sendEvent = (event: TEvent) => {
+      return stream.writeSSE({
+        event: eventName,
+        data: JSON.stringify(event),
+      })
+    }
 
-      abortHandler = () => {
-        const currentCleanup = cleanup
-        cleanup = undefined
-        currentCleanup?.()
-      }
+    const sendChunk = (chunk: string) => {
+      void stream.write(chunk)
+    }
 
-      abortSignal.addEventListener('abort', abortHandler)
+    await stream.write(': connected\n\n')
 
-      cleanup = () => {
-        cleanupInternal()
-        cleanup = undefined
+    heartbeat = setInterval(() => {
+      if (abortSignal.aborted) {
+        stopHeartbeat()
+        return
       }
 
-      sendChunk(': connected\n\n')
-      ;(async () => {
-        try {
-          await handler({
-            sendEvent,
-            sendChunk,
-            abortSignal,
-          })
-        } catch (error) {
-          console.error('SSE handler failed', error)
-        } finally {
-          const currentCleanup = cleanup
-          cleanup = undefined
-          currentCleanup?.()
-        }
-      })().catch((error) => {
-        console.error('Unhandled SSE handler error', error)
-        const currentCleanup = cleanup
-        cleanup = undefined
-        currentCleanup?.()
+      stream.write(`: keep-alive ${new Date().toISOString()}\n\n`).catch((error) => {
+        console.error('SSE heartbeat failed', error)
+        stopHeartbeat()
       })
-    },
-    cancel() {
-      const currentCleanup = cleanup
-      cleanup = undefined
-      currentCleanup?.()
-    },
-  })
-
-  return new Response(stream, {
-    headers: {
-      'Content-Type': 'text/event-stream',
-      'Cache-Control': 'no-cache, no-transform',
-      Connection: 'keep-alive',
-      'X-Accel-Buffering': 'no',
-    },
+    }, heartbeatIntervalMs)
+
+    try {
+      await handler({
+        sendEvent: (event) => {
+          void sendEvent(event)
+        },
+        sendChunk,
+        abortSignal,
+      })
+    } catch (error) {
+      console.error('SSE handler failed', error)
+    } finally {
+      stopHeartbeat()
+      stream.close()
+    }
   })
 }

packages/builder/src/logger/index.ts

@@ -1,4 +1,4 @@
-import type {ConsolaInstance} from 'consola';
+import type { ConsolaInstance } from 'consola'
 import consola from 'consola'
 
 export type LogLevel = 'log' | 'info' | 'success' | 'warn' | 'error' | 'debug' | 'trace' | 'start' | 'fatal'
@@ -26,6 +26,10 @@ export function setConsoleForwarding(enabled: boolean): void {
   forwardToConsole = enabled
 }
 
+export function relayLogMessage(tag: string, level: LogLevel | string, args: unknown[]): void {
+  notifyListener(tag, level, args)
+}
+
 function notifyListener(tag: string, level: LogLevel | string, args: unknown[]): void {
   listener?.({
     tag,