Update chat events/transcription flow and container startup fixes

Frontend/server/queues/outboundDelivery.ts

@@ -0,0 +1,200 @@
+import { Queue, Worker, type JobsOptions } from "bullmq";
+import { prisma } from "../utils/prisma";
+import { getRedis } from "../utils/redis";
+
+export const OUTBOUND_DELIVERY_QUEUE_NAME = "omni-outbound";
+
+export type OutboundDeliveryJob = {
+  omniMessageId: string;
+  endpoint: string;
+  method?: "POST" | "PUT" | "PATCH";
+  headers?: Record<string, string>;
+  payload: unknown;
+  timeoutMs?: number;
+  channel?: string;
+  provider?: string;
+};
+
+function ensureHttpUrl(value: string) {
+  const raw = (value ?? "").trim();
+  if (!raw) throw new Error("endpoint is required");
+  const parsed = new URL(raw);
+  if (parsed.protocol !== "http:" && parsed.protocol !== "https:") {
+    throw new Error(`Unsupported endpoint protocol: ${parsed.protocol}`);
+  }
+  return parsed.toString();
+}
+
+function compactError(error: unknown) {
+  if (!error) return "unknown_error";
+  if (typeof error === "string") return error;
+  const anyErr = error as any;
+  return String(anyErr?.message ?? anyErr);
+}
+
+function extractProviderMessageId(body: unknown): string | null {
+  const obj = body as any;
+  if (!obj || typeof obj !== "object") return null;
+  const candidate =
+    obj?.message_id ??
+    obj?.messageId ??
+    obj?.id ??
+    obj?.result?.message_id ??
+    obj?.result?.id ??
+    null;
+  if (candidate == null) return null;
+  return String(candidate);
+}
+
+export function outboundDeliveryQueue() {
+  return new Queue<OutboundDeliveryJob>(OUTBOUND_DELIVERY_QUEUE_NAME, {
+    connection: getRedis(),
+    defaultJobOptions: {
+      removeOnComplete: { count: 1000 },
+      removeOnFail: { count: 5000 },
+    },
+  });
+}
+
+export async function enqueueOutboundDelivery(input: OutboundDeliveryJob, opts?: JobsOptions) {
+  const endpoint = ensureHttpUrl(input.endpoint);
+  const q = outboundDeliveryQueue();
+
+  // Keep source message in pending before actual send starts.
+  await prisma.omniMessage.update({
+    where: { id: input.omniMessageId },
+    data: {
+      status: "PENDING",
+      rawJson: {
+        queue: {
+          queueName: OUTBOUND_DELIVERY_QUEUE_NAME,
+          enqueuedAt: new Date().toISOString(),
+        },
+        deliveryRequest: {
+          endpoint,
+          method: input.method ?? "POST",
+          channel: input.channel ?? null,
+          provider: input.provider ?? null,
+          payload: input.payload,
+        },
+      },
+    },
+  });
+
+  return q.add("deliver", { ...input, endpoint }, {
+    jobId: `omni:${input.omniMessageId}`,
+    attempts: 12,
+    backoff: { type: "exponential", delay: 1000 },
+    ...opts,
+  });
+}
+
+export function startOutboundDeliveryWorker() {
+  return new Worker<OutboundDeliveryJob>(
+    OUTBOUND_DELIVERY_QUEUE_NAME,
+    async (job) => {
+      const msg = await prisma.omniMessage.findUnique({
+        where: { id: job.data.omniMessageId },
+        include: { thread: true },
+      });
+      if (!msg) return;
+
+      // Idempotency: if already sent/delivered, do not resend.
+      if ((msg.status === "SENT" || msg.status === "DELIVERED" || msg.status === "READ") && msg.providerMessageId) {
+        return;
+      }
+
+      const endpoint = ensureHttpUrl(job.data.endpoint);
+      const timeoutMs = Math.max(1000, Math.min(job.data.timeoutMs ?? 20000, 120000));
+      const method = job.data.method ?? "POST";
+      const headers: Record<string, string> = {
+        "content-type": "application/json",
+        ...(job.data.headers ?? {}),
+      };
+
+      const requestStartedAt = new Date().toISOString();
+      try {
+        const response = await fetch(endpoint, {
+          method,
+          headers,
+          body: JSON.stringify(job.data.payload ?? {}),
+          signal: AbortSignal.timeout(timeoutMs),
+        });
+
+        const text = await response.text();
+        const responseBody = (() => {
+          try {
+            return JSON.parse(text);
+          } catch {
+            return text;
+          }
+        })();
+
+        if (!response.ok) {
+          throw new Error(`HTTP ${response.status}: ${typeof responseBody === "string" ? responseBody : JSON.stringify(responseBody)}`);
+        }
+
+        const providerMessageId = extractProviderMessageId(responseBody);
+        await prisma.omniMessage.update({
+          where: { id: msg.id },
+          data: {
+            status: "SENT",
+            providerMessageId,
+            rawJson: {
+              queue: {
+                queueName: OUTBOUND_DELIVERY_QUEUE_NAME,
+                completedAt: new Date().toISOString(),
+                attemptsMade: job.attemptsMade + 1,
+              },
+              deliveryRequest: {
+                endpoint,
+                method,
+                channel: job.data.channel ?? null,
+                provider: job.data.provider ?? null,
+                startedAt: requestStartedAt,
+                payload: job.data.payload ?? null,
+              },
+              deliveryResponse: {
+                status: response.status,
+                body: responseBody,
+              },
+            },
+          },
+        });
+      } catch (error) {
+        const isLastAttempt =
+          typeof job.opts.attempts === "number" && job.attemptsMade + 1 >= job.opts.attempts;
+
+        if (isLastAttempt) {
+          await prisma.omniMessage.update({
+            where: { id: msg.id },
+            data: {
+              status: "FAILED",
+              rawJson: {
+                queue: {
+                  queueName: OUTBOUND_DELIVERY_QUEUE_NAME,
+                  failedAt: new Date().toISOString(),
+                  attemptsMade: job.attemptsMade + 1,
+                },
+                deliveryRequest: {
+                  endpoint,
+                  method,
+                  channel: job.data.channel ?? null,
+                  provider: job.data.provider ?? null,
+                  startedAt: requestStartedAt,
+                  payload: job.data.payload ?? null,
+                },
+                deliveryError: {
+                  message: compactError(error),
+                },
+              },
+            },
+          });
+        }
+
+        throw error;
+      }
+    },
+    { connection: getRedis() },
+  );
+}