Restructure omni services and add Chatwoot research snapshot

omni_inbound/src/index.ts

@@ -0,0 +1,38 @@
+import { closeInboundQueue } from "./queue";
+import { startServer } from "./server";
+
+const server = startServer();
+
+async function shutdown(signal: string) {
+  console.log(`[omni_inbound] shutting down by ${signal}`);
+
+  try {
+    await new Promise<void>((resolve, reject) => {
+      server.close((error) => {
+        if (error) {
+          reject(error);
+          return;
+        }
+        resolve();
+      });
+    });
+  } catch {
+    // ignore shutdown errors
+  }
+
+  try {
+    await closeInboundQueue();
+  } catch {
+    // ignore shutdown errors
+  }
+
+  process.exit(0);
+}
+
+process.on("SIGINT", () => {
+  void shutdown("SIGINT");
+});
+
+process.on("SIGTERM", () => {
+  void shutdown("SIGTERM");
+});

omni_inbound/src/queue.ts

@@ -0,0 +1,67 @@
+import { createHash } from "node:crypto";
+import { Queue, type ConnectionOptions } from "bullmq";
+import type { OmniInboundEnvelopeV1 } from "./types";
+
+export const RECEIVER_FLOW_QUEUE_NAME = (
+  process.env.RECEIVER_FLOW_QUEUE_NAME ||
+  process.env.INBOUND_QUEUE_NAME ||
+  "receiver.flow"
+).trim();
+
+let queueInstance: Queue<OmniInboundEnvelopeV1, unknown, "ingest"> | null = null;
+
+function redisConnectionFromEnv(): ConnectionOptions {
+  const raw = (process.env.REDIS_URL || "redis://localhost:6379").trim();
+  const parsed = new URL(raw);
+
+  return {
+    host: parsed.hostname,
+    port: parsed.port ? Number(parsed.port) : 6379,
+    username: parsed.username ? decodeURIComponent(parsed.username) : undefined,
+    password: parsed.password ? decodeURIComponent(parsed.password) : undefined,
+    db: parsed.pathname && parsed.pathname !== "/" ? Number(parsed.pathname.slice(1)) : undefined,
+    maxRetriesPerRequest: null,
+  };
+}
+
+function toJobId(idempotencyKey: string) {
+  const hash = createHash("sha256").update(idempotencyKey).digest("hex");
+  return `inbound:${hash}`;
+}
+
+export function inboundQueue() {
+  if (queueInstance) return queueInstance;
+
+  queueInstance = new Queue<OmniInboundEnvelopeV1, unknown, "ingest">(RECEIVER_FLOW_QUEUE_NAME, {
+    connection: redisConnectionFromEnv(),
+    defaultJobOptions: {
+      removeOnComplete: { count: 10000 },
+      removeOnFail: { count: 20000 },
+      attempts: 8,
+      backoff: { type: "exponential", delay: 1000 },
+    },
+  });
+
+  return queueInstance;
+}
+
+export async function enqueueInboundEvent(envelope: OmniInboundEnvelopeV1) {
+  const q = inboundQueue();
+  const jobId = toJobId(envelope.idempotencyKey);
+
+  return q.add("ingest", envelope, {
+    jobId,
+  });
+}
+
+export function isDuplicateJobError(error: unknown) {
+  if (!error || typeof error !== "object") return false;
+  const message = String((error as { message?: string }).message || "").toLowerCase();
+  return message.includes("job") && message.includes("exists");
+}
+
+export async function closeInboundQueue() {
+  if (!queueInstance) return;
+  await queueInstance.close();
+  queueInstance = null;
+}

omni_inbound/src/server.ts

@@ -0,0 +1,108 @@
+import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
+import { RECEIVER_FLOW_QUEUE_NAME, enqueueInboundEvent, isDuplicateJobError } from "./queue";
+import { parseTelegramBusinessUpdate } from "./telegram";
+
+const MAX_BODY_SIZE_BYTES = Number(process.env.MAX_BODY_SIZE_BYTES || 1024 * 1024);
+
+function writeJson(res: ServerResponse, statusCode: number, body: unknown) {
+  const payload = JSON.stringify(body);
+  res.statusCode = statusCode;
+  res.setHeader("content-type", "application/json; charset=utf-8");
+  res.end(payload);
+}
+
+async function readJsonBody(req: IncomingMessage): Promise<unknown> {
+  const chunks: Buffer[] = [];
+  let total = 0;
+
+  for await (const chunk of req) {
+    const buf = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
+    total += buf.length;
+    if (total > MAX_BODY_SIZE_BYTES) {
+      throw new Error(`payload_too_large:${MAX_BODY_SIZE_BYTES}`);
+    }
+    chunks.push(buf);
+  }
+
+  const raw = Buffer.concat(chunks).toString("utf8");
+  if (!raw) return {};
+  return JSON.parse(raw);
+}
+
+function validateTelegramSecret(req: IncomingMessage): boolean {
+  const expected = (process.env.TELEGRAM_WEBHOOK_SECRET || "").trim();
+  if (!expected) return true;
+
+  const incoming = String(req.headers["x-telegram-bot-api-secret-token"] || "").trim();
+  return incoming !== "" && incoming === expected;
+}
+
+export function startServer() {
+  const port = Number(process.env.PORT || 8080);
+
+  const server = createServer(async (req, res) => {
+    if (!req.url || !req.method) {
+      writeJson(res, 404, { ok: false, error: "not_found" });
+      return;
+    }
+
+    if (req.url === "/health" && req.method === "GET") {
+      writeJson(res, 200, {
+        ok: true,
+        service: "omni_inbound",
+        queue: RECEIVER_FLOW_QUEUE_NAME,
+        now: new Date().toISOString(),
+      });
+      return;
+    }
+
+    if (req.url === "/webhooks/telegram/business" && req.method === "POST") {
+      if (!validateTelegramSecret(req)) {
+        writeJson(res, 401, { ok: false, error: "invalid_webhook_secret" });
+        return;
+      }
+
+      try {
+        const body = await readJsonBody(req);
+        const envelope = parseTelegramBusinessUpdate(body);
+
+        await enqueueInboundEvent(envelope);
+
+        writeJson(res, 200, {
+          ok: true,
+          queued: true,
+          duplicate: false,
+          providerEventId: envelope.providerEventId,
+          idempotencyKey: envelope.idempotencyKey,
+        });
+      } catch (error) {
+        if (isDuplicateJobError(error)) {
+          writeJson(res, 200, {
+            ok: true,
+            queued: false,
+            duplicate: true,
+          });
+          return;
+        }
+
+        const message = error instanceof Error ? error.message : String(error);
+        const statusCode = message.startsWith("payload_too_large:") ? 413 : 503;
+        writeJson(res, statusCode, {
+          ok: false,
+          error: "receiver_enqueue_failed",
+          message,
+        });
+      }
+
+      return;
+    }
+
+    writeJson(res, 404, { ok: false, error: "not_found" });
+  });
+
+  server.listen(port, "0.0.0.0", () => {
+    console.log(`[omni_inbound] listening on :${port}`);
+  });
+
+  return server;
+}

omni_inbound/src/telegram.ts

@@ -0,0 +1,127 @@
+import { createHash } from "node:crypto";
+import type { JsonObject, OmniInboundEnvelopeV1 } from "./types";
+
+const MAX_TEXT_LENGTH = 4096;
+
+function asObject(value: unknown): JsonObject {
+  return value && typeof value === "object" && !Array.isArray(value) ? (value as JsonObject) : {};
+}
+
+function pickMessage(update: JsonObject): JsonObject {
+  const candidates = [
+    update.message,
+    update.edited_message,
+    update.business_message,
+    update.edited_business_message,
+    update.channel_post,
+    update.edited_channel_post,
+  ];
+
+  for (const candidate of candidates) {
+    const obj = asObject(candidate);
+    if (Object.keys(obj).length > 0) return obj;
+  }
+
+  return {};
+}
+
+function pickEventType(update: JsonObject): string {
+  if (update.business_message) return "business_message";
+  if (update.edited_business_message) return "edited_business_message";
+  if (update.business_connection) return "business_connection";
+  if (update.deleted_business_messages) return "deleted_business_messages";
+  if (update.message) return "message";
+  if (update.edited_message) return "edited_message";
+  return "unknown";
+}
+
+function isoFromUnix(value: unknown) {
+  if (typeof value !== "number" || !Number.isFinite(value)) return null;
+  return new Date(value * 1000).toISOString();
+}
+
+function cropText(value: unknown) {
+  if (typeof value !== "string") return null;
+  return value.slice(0, MAX_TEXT_LENGTH);
+}
+
+function requireString(value: unknown, fallback: string) {
+  const v = String(value ?? "").trim();
+  return v || fallback;
+}
+
+function makeFallbackEventId(raw: unknown) {
+  return createHash("sha256").update(JSON.stringify(raw ?? null)).digest("hex");
+}
+
+export function parseTelegramBusinessUpdate(raw: unknown): OmniInboundEnvelopeV1 {
+  const update = asObject(raw);
+  const message = pickMessage(update);
+  const receivedAt = new Date().toISOString();
+
+  const updateId = update.update_id;
+  const messageId = message.message_id;
+  const businessConnection = asObject(update.business_connection);
+
+  const providerEventId =
+    (updateId != null && requireString(updateId, "")) ||
+    (messageId != null && requireString(messageId, "")) ||
+    makeFallbackEventId(raw);
+
+  const providerMessageId = messageId != null ? String(messageId) : null;
+
+  const chat = asObject(message.chat);
+  const from = asObject(message.from);
+
+  const threadExternalId =
+    chat.id != null
+      ? String(chat.id)
+      : businessConnection.user_chat_id != null
+        ? String(businessConnection.user_chat_id)
+        : null;
+
+  const contactExternalId = from.id != null ? String(from.id) : null;
+
+  const text = cropText(message.text) ?? cropText(message.caption);
+
+  const businessConnectionId =
+    message.business_connection_id != null
+      ? String(message.business_connection_id)
+      : businessConnection.id != null
+        ? String(businessConnection.id)
+        : null;
+
+  const occurredAt =
+    isoFromUnix(message.date) ??
+    isoFromUnix(businessConnection.date) ??
+    receivedAt;
+
+  const eventType = pickEventType(update);
+
+  const idempotencyKey = ["telegram_business", providerEventId, businessConnectionId || "no-bc"].join(":");
+
+  return {
+    version: 1,
+    idempotencyKey,
+    provider: "telegram_business",
+    channel: "TELEGRAM",
+    direction: "IN",
+    providerEventId,
+    providerMessageId,
+    eventType,
+    occurredAt,
+    receivedAt,
+    payloadRaw: raw,
+    payloadNormalized: {
+      threadExternalId,
+      contactExternalId,
+      text,
+      businessConnectionId,
+      updateId: updateId != null ? String(updateId) : null,
+      chatTitle: typeof chat.title === "string" ? chat.title : null,
+      fromUsername: typeof from.username === "string" ? from.username : null,
+      fromFirstName: typeof from.first_name === "string" ? from.first_name : null,
+      fromLastName: typeof from.last_name === "string" ? from.last_name : null,
+    },
+  };
+}

omni_inbound/src/types.ts

@@ -0,0 +1,24 @@
+export type OmniInboundChannel = "TELEGRAM" | "WHATSAPP" | "INSTAGRAM" | "PHONE" | "EMAIL" | "INTERNAL";
+
+export type OmniInboundEnvelopeV1 = {
+  version: 1;
+  idempotencyKey: string;
+  provider: string;
+  channel: OmniInboundChannel;
+  direction: "IN";
+  providerEventId: string;
+  providerMessageId: string | null;
+  eventType: string;
+  occurredAt: string;
+  receivedAt: string;
+  payloadRaw: unknown;
+  payloadNormalized: {
+    threadExternalId: string | null;
+    contactExternalId: string | null;
+    text: string | null;
+    businessConnectionId: string | null;
+    [key: string]: unknown;
+  };
+};
+
+export type JsonObject = Record<string, unknown>;