chore: merge feat/delivery-module into main

Author: exec-unit

src/modules/delivery/delivery-worker.module.ts

@@ -0,0 +1,14 @@
+import { Module } from '@nestjs/common'
+import { TasksRepository } from '../tasks/tasks.repository.js'
+import { WebhookProcessor } from './workers/webhook.processor.js'
+
+/**
+ * Worker-side module for the Delivery domain.
+ * Contains only the webhook BullMQ processor.
+ * Intentionally excludes HTTP controllers and SseService to keep the
+ * worker process lean and free of any HTTP-serving infrastructure.
+ */
+@Module({
+  providers: [WebhookProcessor, TasksRepository],
+})
+export class DeliveryWorkerModule {}

src/modules/delivery/delivery.module.ts

@@ -0,0 +1,17 @@
+import { Module } from '@nestjs/common'
+import { TasksRepository } from '../tasks/tasks.repository.js'
+import { EventsRepository } from './events.repository.js'
+import { SseService } from './sse.service.js'
+import { StreamController } from './stream.controller.js'
+
+/**
+ * API-side module for the Delivery domain.
+ * Registers the SSE stream endpoint and its dependencies.
+ * Intentionally excludes WebhookProcessor — that belongs to the Worker process.
+ */
+@Module({
+  controllers: [StreamController],
+  providers: [SseService, TasksRepository, EventsRepository],
+  exports: [EventsRepository],
+})
+export class DeliveryModule {}

src/modules/delivery/sse.service.ts

@@ -0,0 +1,159 @@
+import { Injectable, Logger, NotFoundException } from '@nestjs/common'
+import { Observable } from 'rxjs'
+import type { MessageEvent } from '@nestjs/common'
+import type { Redis } from 'ioredis'
+import { RedisService } from '#libs/redis/index.js'
+import { TasksRepository } from '../tasks/tasks.repository.js'
+import { EventsRepository } from './events.repository.js'
+import type { TaskEventSelect, TaskEventType } from './schemas/events.sql.js'
+
+const TERMINAL_EVENT_TYPES = new Set<TaskEventType>(['completed', 'failed', 'cancelled'])
+
+/**
+ * 25 s keeps the connection alive through reverse proxies (nginx proxy_read_timeout,
+ * ALB idle timeout) that would otherwise silently drop quiet SSE connections.
+ */
+const HEARTBEAT_INTERVAL_MS = 25_000
+
+const HEARTBEAT_EVENT: MessageEvent = { data: '', type: 'ping' }
+
+/**
+ * Maps a DB event row to the SSE wire format.
+ * `seq` becomes Last-Event-ID so clients can resume after disconnect.
+ * DB-internal fields (id, taskId, createdAt) are stripped from the payload.
+ */
+function toMessageEvent(event: TaskEventSelect): MessageEvent {
+  const { id: _id, taskId: _taskId, seq, createdAt: _createdAt, ...eventPayload } = event
+  return {
+    id: String(seq),
+    type: eventPayload.eventType,
+    data: JSON.stringify(eventPayload),
+  }
+}
+
+/**
+ * Delivers task lifecycle events over SSE via event sourcing + Redis Pub/Sub.
+ *
+ * Each connection gets a dedicated ioredis subscriber client because ioredis
+ * enters subscriber-only mode on `.subscribe()`, blocking all other commands
+ * on that connection.
+ */
+@Injectable()
+export class SseService {
+  private readonly logger = new Logger(SseService.name)
+
+  constructor(
+    private readonly redis: RedisService,
+    private readonly tasksRepo: TasksRepository,
+    private readonly eventsRepo: EventsRepository,
+  ) {}
+
+  /**
+   * Opens an SSE stream for a task. Replays history from `afterSeq`,
+   * then streams live Pub/Sub events until a terminal state is reached.
+   * Includes parallel heartbeat pings to prevent proxy idle timeouts.
+   */
+  async stream(taskId: string, afterSeq: number): Promise<Observable<MessageEvent>> {
+    const task = await this.tasksRepo.findById(taskId)
+    if (!task) throw new NotFoundException(`Task ${taskId} not found`)
+
+    return new Observable<MessageEvent>((subscriber) => {
+      let isDone = false
+      let historyLoaded = false
+      let highestHistorySeq = afterSeq
+      const liveBuffer: TaskEventSelect[] = []
+
+      // Dedicated subscriber client — isolated from the shared business connection.
+      const subClient = (this.redis.client as Redis).duplicate()
+      const channel = `task:${taskId}`
+
+      void subClient.subscribe(channel, (err) => {
+        if (err) {
+          this.logger.error(`Failed to subscribe to channel ${channel}: ${err.message}`)
+          subscriber.error(err)
+        }
+      })
+
+      subClient.on('message', (_chan: string, rawMessage: string) => {
+        if (isDone) return
+        try {
+          const event = JSON.parse(rawMessage) as TaskEventSelect
+
+          if (!historyLoaded) {
+            liveBuffer.push(event)
+            return
+          }
+
+          if (event.seq > highestHistorySeq) {
+            highestHistorySeq = event.seq
+            subscriber.next(toMessageEvent(event))
+            if (TERMINAL_EVENT_TYPES.has(event.eventType)) {
+              isDone = true
+              subscriber.complete()
+            }
+          }
+        } catch (err) {
+          this.logger.error(
+            `Failed to parse Pub/Sub message on ${channel}: ${String(err)}`,
+          )
+        }
+      })
+
+      subClient.on('error', (err: Error) => {
+        this.logger.error(`Redis subscriber error on ${channel}: ${err.message}`)
+        subscriber.error(err)
+      })
+
+      const heartbeatTimer = setInterval(() => {
+        if (!isDone) subscriber.next(HEARTBEAT_EVENT)
+      }, HEARTBEAT_INTERVAL_MS)
+
+      // Fetch history AFTER subscribing to Redis to guarantee no missed events.
+      this.eventsRepo
+        .findEventsSince(taskId, afterSeq)
+        .then((events) => {
+          if (isDone) return
+
+          for (const event of events) {
+            highestHistorySeq = Math.max(highestHistorySeq, event.seq)
+            subscriber.next(toMessageEvent(event))
+            if (TERMINAL_EVENT_TYPES.has(event.eventType)) {
+              isDone = true
+              subscriber.complete()
+              return
+            }
+          }
+
+          historyLoaded = true
+          for (const event of liveBuffer) {
+            if (event.seq > highestHistorySeq) {
+              highestHistorySeq = event.seq
+              subscriber.next(toMessageEvent(event))
+              if (TERMINAL_EVENT_TYPES.has(event.eventType)) {
+                isDone = true
+                subscriber.complete()
+                return
+              }
+            }
+          }
+        })
+        .catch((err: unknown) => {
+          if (!isDone) subscriber.error(err)
+        })
+
+      return () => {
+        isDone = true
+        clearInterval(heartbeatTimer)
+        subClient
+          .unsubscribe(channel)
+          .finally(() => {
+            void subClient.quit()
+          })
+          .catch((err: unknown) => {
+            this.logger.error(`Failed to unsubscribe SSE client: ${String(err)}`)
+          })
+        this.logger.debug(`SSE stream closed for taskId=${taskId}`)
+      }
+    })
+  }
+}

src/modules/delivery/stream.controller.ts

@@ -0,0 +1,66 @@
+import {
+  Controller,
+  Get,
+  Header,
+  Headers,
+  HttpStatus,
+  Param,
+  ParseUUIDPipe,
+  Sse,
+} from '@nestjs/common'
+import { ApiTags, ApiOperation, ApiResponse, ApiHeader } from '@nestjs/swagger'
+import type { MessageEvent } from '@nestjs/common'
+import type { Observable } from 'rxjs'
+import { SseService } from './sse.service.js'
+
+/**
+ * Parses the `Last-Event-ID` header into a non-negative integer sequence number.
+ * Falls back to 0 (replay all) on missing or malformed values — this is
+ * intentionally forgiving: a wrong seq causes extra data, not data loss.
+ */
+function parseLastEventId(raw: string | undefined): number {
+  if (!raw) return 0
+  const parsed = Number(raw)
+  return Number.isFinite(parsed) && parsed >= 0 ? Math.floor(parsed) : 0
+}
+
+/**
+ * Exposes the Server-Sent Events endpoint for real-time task lifecycle notifications.
+ * Uses event sourcing: clients reconnecting with `Last-Event-ID` receive all missed events.
+ */
+@ApiTags('tasks')
+@Controller('tasks')
+export class StreamController {
+  constructor(private readonly sseService: SseService) {}
+
+  /**
+   * Opens a persistent SSE connection. Replays missed events from `Last-Event-ID`,
+   * then delivers live updates until the task reaches a terminal state.
+   */
+  @Get(':id/stream')
+  @Sse()
+  // Prevent reverse proxies and CDNs from buffering the SSE response.
+  @Header('X-Accel-Buffering', 'no')
+  @Header('Cache-Control', 'no-cache')
+  @ApiOperation({ summary: 'Open SSE stream for real-time task events' })
+  @ApiHeader({
+    name: 'Last-Event-ID',
+    description: 'Resume replay from this event sequence number (inclusive)',
+    required: false,
+  })
+  @ApiResponse({
+    status: HttpStatus.OK,
+    description: 'SSE stream opened; events follow task lifecycle',
+  })
+  @ApiResponse({
+    status: HttpStatus.NOT_FOUND,
+    description: 'Task not found',
+  })
+  async stream(
+    @Param('id', ParseUUIDPipe) id: string,
+    @Headers('last-event-id') lastEventId: string | undefined,
+  ): Promise<Observable<MessageEvent>> {
+    const afterSeq = parseLastEventId(lastEventId)
+    return this.sseService.stream(id, afterSeq)
+  }
+}

src/modules/delivery/workers/webhook.processor.ts

@@ -0,0 +1,116 @@
+import {
+  Injectable,
+  Logger,
+  type OnModuleDestroy,
+  type OnModuleInit,
+} from '@nestjs/common'
+import { Worker } from 'bullmq'
+import { RedisService } from '#libs/redis/index.js'
+import {
+  HttpRetryService,
+  DispatchFailedError,
+  NonRetryableError,
+} from '#libs/http-retry/index.js'
+import { TasksRepository } from '../../tasks/tasks.repository.js'
+
+interface WebhookJobData {
+  taskId: string
+}
+
+/**
+ * Consumes the `webhook` queue.
+ * Delivers the final task result to the client-configured webhookUrl (best-effort).
+ *
+ * The task is already in a terminal state by the time this runs — delivery
+ * failure does NOT change task status. Uses exponential backoff for transient
+ * failures; 4xx responses are treated as non-retryable client misconfiguration.
+ */
+@Injectable()
+export class WebhookProcessor implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(WebhookProcessor.name)
+  private worker!: Worker
+
+  constructor(
+    private readonly redis: RedisService,
+    private readonly tasksRepo: TasksRepository,
+    private readonly httpRetry: HttpRetryService,
+  ) {}
+
+  onModuleInit(): void {
+    this.worker = new Worker(
+      'webhook',
+      (job) => this.process(job.data as WebhookJobData),
+      {
+        // Workers must not share the business Redis connection.
+        connection: (this.redis.client as import('ioredis').Redis).duplicate(),
+        concurrency: 20,
+      },
+    )
+
+    this.worker.on('failed', (job, err) => {
+      this.logger.error(
+        `Webhook job ${job?.id ?? 'unknown'} failed permanently: ${String(err)}`,
+      )
+    })
+
+    this.logger.log('Webhook worker started')
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    await this.worker.close()
+    this.logger.log('Webhook worker closed')
+  }
+
+  private async process({ taskId }: WebhookJobData): Promise<void> {
+    // Always re-fetch from DB: job data is enqueued at result-submission time
+    // but the DB state is the authoritative source of truth.
+    const task = await this.tasksRepo.findById(taskId)
+
+    if (!task) {
+      // Task was deleted between enqueue and now (e.g. retention sweep).
+      this.logger.warn(`Webhook skipped: task ${taskId} no longer exists`)
+      return
+    }
+
+    if (!task.webhookUrl) {
+      this.logger.warn(`Webhook skipped: task ${taskId} has no webhookUrl`)
+      return
+    }
+
+    const payload: Record<string, unknown> = {
+      taskId: task.id,
+      status: task.status,
+      ...(task.result !== null ? { result: task.result } : {}),
+      ...(task.error !== null ? { error: task.error } : {}),
+    }
+
+    try {
+      await this.httpRetry.post(
+        task.webhookUrl,
+        payload,
+        {},
+        { attempts: 5, baseDelayMs: 1_000, maxDelayMs: 60_000 },
+      )
+      this.logger.log(`Webhook delivered for taskId=${taskId} to ${task.webhookUrl}`)
+    } catch (err) {
+      if (err instanceof NonRetryableError) {
+        // 4xx from the webhookUrl endpoint — client-side misconfiguration, pointless to retry.
+        this.logger.error(
+          `Webhook delivery rejected (non-retryable) for taskId=${taskId}: ${err.message}`,
+        )
+        return
+      }
+
+      if (err instanceof DispatchFailedError) {
+        // All retry attempts exhausted — webhook_delivery_failed event for downstream alerting.
+        this.logger.error(
+          `webhook_delivery_failed taskId=${taskId} url=${task.webhookUrl}: ${err.message}`,
+        )
+        return
+      }
+
+      // Unexpected error (e.g. DB outage on re-fetch) — re-throw to retry the job.
+      throw err
+    }
+  }
+}