omni_chat: consume receiver.flow and persist inbound telegram
This commit is contained in:
Ruslan Bakiev
283
omni_chat/src/worker.ts
Normal file
283
omni_chat/src/worker.ts
Normal file
@@ -0,0 +1,283 @@
|
||||
import { Queue, Worker, type ConnectionOptions, type Job } from "bullmq";
|
||||
import { Prisma } from "@prisma/client";
|
||||
import { prisma } from "./utils/prisma";
|
||||
|
||||
type JsonObject = Record<string, unknown>;
|
||||
|
||||
type OmniInboundEnvelopeV1 = {
|
||||
version: 1;
|
||||
idempotencyKey: string;
|
||||
provider: string;
|
||||
channel: "TELEGRAM" | "WHATSAPP" | "INSTAGRAM" | "PHONE" | "EMAIL" | "INTERNAL";
|
||||
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;
|
||||
updateId?: string | null;
|
||||
[key: string]: unknown;
|
||||
};
|
||||
};
|
||||
|
||||
export const RECEIVER_FLOW_QUEUE_NAME = (process.env.RECEIVER_FLOW_QUEUE_NAME || "receiver.flow").trim();
|
||||
|
||||
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 normalizeText(input: unknown) {
|
||||
const t = String(input ?? "").trim();
|
||||
return t || "[no text]";
|
||||
}
|
||||
|
||||
function parseOccurredAt(input: string | null | undefined) {
|
||||
const d = new Date(String(input ?? ""));
|
||||
if (Number.isNaN(d.getTime())) return new Date();
|
||||
return d;
|
||||
}
|
||||
|
||||
async function resolveTeamId(env: OmniInboundEnvelopeV1) {
|
||||
const n = env.payloadNormalized ?? ({} as OmniInboundEnvelopeV1["payloadNormalized"]);
|
||||
const bcId = String(n.businessConnectionId ?? "").trim();
|
||||
if (bcId) {
|
||||
const linked = await prisma.telegramBusinessConnection.findFirst({
|
||||
where: { businessConnectionId: bcId },
|
||||
orderBy: { updatedAt: "desc" },
|
||||
select: { teamId: true },
|
||||
});
|
||||
if (linked?.teamId) return linked.teamId;
|
||||
}
|
||||
|
||||
const externalContactId = String(n.contactExternalId ?? n.threadExternalId ?? "").trim();
|
||||
if (externalContactId) {
|
||||
const pseudo = `link:${externalContactId}`;
|
||||
const linked = await prisma.telegramBusinessConnection.findFirst({
|
||||
where: { businessConnectionId: pseudo },
|
||||
orderBy: { updatedAt: "desc" },
|
||||
select: { teamId: true },
|
||||
});
|
||||
if (linked?.teamId) return linked.teamId;
|
||||
}
|
||||
|
||||
const fallbackTeamId = String(process.env.DEFAULT_TEAM_ID || "").trim();
|
||||
if (fallbackTeamId) return fallbackTeamId;
|
||||
|
||||
const demo = await prisma.team.findFirst({
|
||||
where: { id: "demo-team" },
|
||||
select: { id: true },
|
||||
});
|
||||
return demo?.id ?? null;
|
||||
}
|
||||
|
||||
async function resolveContact(input: { teamId: string; externalContactId: string }) {
|
||||
const existingIdentity = await prisma.omniContactIdentity.findFirst({
|
||||
where: {
|
||||
teamId: input.teamId,
|
||||
channel: "TELEGRAM",
|
||||
externalId: input.externalContactId,
|
||||
},
|
||||
select: { contactId: true },
|
||||
});
|
||||
if (existingIdentity?.contactId) {
|
||||
return existingIdentity.contactId;
|
||||
}
|
||||
|
||||
const contact = await prisma.contact.create({
|
||||
data: {
|
||||
teamId: input.teamId,
|
||||
name: `Telegram ${input.externalContactId}`,
|
||||
company: "",
|
||||
country: "",
|
||||
location: "",
|
||||
},
|
||||
select: { id: true },
|
||||
});
|
||||
|
||||
await prisma.omniContactIdentity.create({
|
||||
data: {
|
||||
teamId: input.teamId,
|
||||
contactId: contact.id,
|
||||
channel: "TELEGRAM",
|
||||
externalId: input.externalContactId,
|
||||
},
|
||||
});
|
||||
|
||||
return contact.id;
|
||||
}
|
||||
|
||||
async function upsertThread(input: {
|
||||
teamId: string;
|
||||
contactId: string;
|
||||
externalChatId: string;
|
||||
businessConnectionId: string | null;
|
||||
}) {
|
||||
const existing = await prisma.omniThread.findFirst({
|
||||
where: {
|
||||
teamId: input.teamId,
|
||||
channel: "TELEGRAM",
|
||||
externalChatId: input.externalChatId,
|
||||
businessConnectionId: input.businessConnectionId,
|
||||
},
|
||||
select: { id: true },
|
||||
});
|
||||
|
||||
if (existing) {
|
||||
await prisma.omniThread.update({
|
||||
where: { id: existing.id },
|
||||
data: { contactId: input.contactId },
|
||||
});
|
||||
return existing;
|
||||
}
|
||||
|
||||
return prisma.omniThread.create({
|
||||
data: {
|
||||
teamId: input.teamId,
|
||||
contactId: input.contactId,
|
||||
channel: "TELEGRAM",
|
||||
externalChatId: input.externalChatId,
|
||||
businessConnectionId: input.businessConnectionId,
|
||||
title: null,
|
||||
},
|
||||
select: { id: true },
|
||||
});
|
||||
}
|
||||
|
||||
async function ingestInbound(env: OmniInboundEnvelopeV1) {
|
||||
if (env.channel !== "TELEGRAM" || env.direction !== "IN") return;
|
||||
|
||||
const teamId = await resolveTeamId(env);
|
||||
if (!teamId) {
|
||||
console.warn("[omni_chat] skip inbound: team not resolved", env.providerEventId);
|
||||
return;
|
||||
}
|
||||
|
||||
const n = env.payloadNormalized ?? ({} as OmniInboundEnvelopeV1["payloadNormalized"]);
|
||||
const externalContactId = String(n.contactExternalId ?? n.threadExternalId ?? "").trim();
|
||||
const externalChatId = String(n.threadExternalId ?? n.contactExternalId ?? "").trim();
|
||||
|
||||
if (!externalContactId || !externalChatId) {
|
||||
console.warn("[omni_chat] skip inbound: missing contact/chat ids", env.providerEventId);
|
||||
return;
|
||||
}
|
||||
|
||||
const businessConnectionId = String(n.businessConnectionId ?? "").trim() || null;
|
||||
const text = normalizeText(n.text);
|
||||
const occurredAt = parseOccurredAt(env.occurredAt);
|
||||
|
||||
const contactId = await resolveContact({ teamId, externalContactId });
|
||||
const thread = await upsertThread({
|
||||
teamId,
|
||||
contactId,
|
||||
externalChatId,
|
||||
businessConnectionId,
|
||||
});
|
||||
|
||||
if (env.providerMessageId) {
|
||||
await prisma.omniMessage.upsert({
|
||||
where: {
|
||||
threadId_providerMessageId: {
|
||||
threadId: thread.id,
|
||||
providerMessageId: env.providerMessageId,
|
||||
},
|
||||
},
|
||||
create: {
|
||||
teamId,
|
||||
contactId,
|
||||
threadId: thread.id,
|
||||
direction: "IN",
|
||||
channel: "TELEGRAM",
|
||||
status: "DELIVERED",
|
||||
text,
|
||||
providerMessageId: env.providerMessageId,
|
||||
providerUpdateId: String((n.updateId as string | null | undefined) ?? env.providerEventId),
|
||||
rawJson: (env.payloadRaw ?? null) as Prisma.InputJsonValue,
|
||||
occurredAt,
|
||||
},
|
||||
update: {
|
||||
text,
|
||||
providerUpdateId: String((n.updateId as string | null | undefined) ?? env.providerEventId),
|
||||
rawJson: (env.payloadRaw ?? null) as Prisma.InputJsonValue,
|
||||
occurredAt,
|
||||
},
|
||||
});
|
||||
} else {
|
||||
await prisma.omniMessage.create({
|
||||
data: {
|
||||
teamId,
|
||||
contactId,
|
||||
threadId: thread.id,
|
||||
direction: "IN",
|
||||
channel: "TELEGRAM",
|
||||
status: "DELIVERED",
|
||||
text,
|
||||
providerMessageId: null,
|
||||
providerUpdateId: String((n.updateId as string | null | undefined) ?? env.providerEventId),
|
||||
rawJson: (env.payloadRaw ?? null) as Prisma.InputJsonValue,
|
||||
occurredAt,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
await prisma.contactMessage.create({
|
||||
data: {
|
||||
contactId,
|
||||
kind: "MESSAGE",
|
||||
direction: "IN",
|
||||
channel: "TELEGRAM",
|
||||
content: text,
|
||||
occurredAt,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
let workerInstance: Worker<OmniInboundEnvelopeV1, unknown, "ingest"> | null = null;
|
||||
|
||||
export function startReceiverWorker() {
|
||||
if (workerInstance) return workerInstance;
|
||||
|
||||
const worker = new Worker<OmniInboundEnvelopeV1, unknown, "ingest">(
|
||||
RECEIVER_FLOW_QUEUE_NAME,
|
||||
async (job) => {
|
||||
await ingestInbound(job.data);
|
||||
},
|
||||
{
|
||||
connection: redisConnectionFromEnv(),
|
||||
concurrency: Number(process.env.OMNI_CHAT_WORKER_CONCURRENCY || 4),
|
||||
},
|
||||
);
|
||||
|
||||
worker.on("failed", (job: Job<OmniInboundEnvelopeV1, unknown, "ingest"> | undefined, err: Error) => {
|
||||
console.error(`[omni_chat] receiver job failed id=${job?.id || "unknown"}: ${err?.message || err}`);
|
||||
});
|
||||
|
||||
workerInstance = worker;
|
||||
return worker;
|
||||
}
|
||||
|
||||
export async function closeReceiverWorker() {
|
||||
if (!workerInstance) return;
|
||||
await workerInstance.close();
|
||||
workerInstance = null;
|
||||
}
|
||||
|
||||
export function receiverQueue() {
|
||||
return new Queue<OmniInboundEnvelopeV1, unknown, "ingest">(RECEIVER_FLOW_QUEUE_NAME, {
|
||||
connection: redisConnectionFromEnv(),
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user