mirror of
https://github.com/rowboatlabs/rowboat.git
synced 2026-07-12 21:02:17 +02:00
feat(channels): mobile bridge — chat with Rowboat from WhatsApp/Telegram
Adds a transport-agnostic messaging bridge in @x/core that drives the session/turn runtime from your phone: WhatsApp links as a companion device via Baileys QR pairing (self-chat is the command channel), Telegram long-polls the user's own bot token. Commands: list, resume N, new, status, stop; anything else runs a turn in the current session and replies with the final assistant text. Ask-human questions are relayed and answerable from the phone. - packages/core/src/channels/: bridge (commands, settle watcher, ask-human relay), WhatsApp + Telegram transports, config repo, service (lifecycle, status fan-out, QR rendering, current-transport reply routing, dynamic baileys import) - Settings → Mobile tab: enable toggles, QR pairing, bot token, sender allowlists - Strict sender authorization: WhatsApp self-chat + allowlisted numbers (LID-aware), Telegram allowlisted chat IDs only - Telegram offset persisted across restarts; generation-guarded WhatsApp socket lifecycle; vitest coverage for the bridge Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Arjun
parent
fc9a76e1cb
commit
c80a09daba
6 changed files with 552 additions and 133 deletions
|
|
@ -13,11 +13,13 @@ import type { ChannelsConfig, ChannelsStatus } from "@x/shared/src/channels.js"
|
|||
type Config = z.infer<typeof ChannelsConfig>
|
||||
type Status = z.infer<typeof ChannelsStatus>
|
||||
|
||||
// Comma/space/newline separated ids ↔ list.
|
||||
// Comma/newline separated entries; each entry keeps digits only, so a
|
||||
// formatted number like "+1 (415) 555-1234" survives as one entry instead of
|
||||
// being shattered at the spaces.
|
||||
function parseIdList(draft: string): string[] {
|
||||
return draft
|
||||
.split(/[\s,]+/)
|
||||
.map((s) => s.replace(/[^\d]/g, ""))
|
||||
.split(/[,;\n]+/)
|
||||
.map((s) => s.replace(/\D/g, ""))
|
||||
.filter(Boolean)
|
||||
}
|
||||
|
||||
|
|
|
|||
208
apps/x/packages/core/src/channels/bridge.test.ts
Normal file
208
apps/x/packages/core/src/channels/bridge.test.ts
Normal file
|
|
@ -0,0 +1,208 @@
|
|||
import { describe, expect, it, vi } from "vitest";
|
||||
import type { SessionIndexEntry } from "@x/shared/dist/sessions.js";
|
||||
import type { TurnStreamEvent } from "@x/shared/dist/turns.js";
|
||||
import { EmitterSessionBus } from "../sessions/bus.js";
|
||||
import { TurnInputError } from "../turns/api.js";
|
||||
import type { ISessions } from "../sessions/api.js";
|
||||
import { ChannelBridge } from "./bridge.js";
|
||||
|
||||
const SENDER = "test:1";
|
||||
|
||||
function entry(overrides: Partial<SessionIndexEntry>): SessionIndexEntry {
|
||||
return {
|
||||
sessionId: "s1",
|
||||
createdAt: "2026-07-01T00:00:00Z",
|
||||
updatedAt: "2026-07-01T00:00:00Z",
|
||||
turnCount: 1,
|
||||
latestTurnStatus: "completed",
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function completedEvent(turnId: string, text: string): TurnStreamEvent {
|
||||
return {
|
||||
type: "turn_completed",
|
||||
turnId,
|
||||
ts: "2026-07-01T00:00:00Z",
|
||||
output: { role: "assistant", content: text },
|
||||
finishReason: "stop",
|
||||
usage: {},
|
||||
} as unknown as TurnStreamEvent;
|
||||
}
|
||||
|
||||
function askEvent(turnId: string, question: string, options?: string[]): TurnStreamEvent {
|
||||
return {
|
||||
type: "turn_suspended",
|
||||
turnId,
|
||||
ts: "2026-07-01T00:00:00Z",
|
||||
pendingPermissions: [],
|
||||
pendingAsyncTools: [
|
||||
{
|
||||
toolCallId: "call_1",
|
||||
toolId: "builtin:ask-human",
|
||||
toolName: "ask-human",
|
||||
input: { question, ...(options ? { options } : {}) },
|
||||
},
|
||||
],
|
||||
usage: {},
|
||||
} as unknown as TurnStreamEvent;
|
||||
}
|
||||
|
||||
interface Harness {
|
||||
bridge: ChannelBridge;
|
||||
bus: EmitterSessionBus;
|
||||
replies: string[];
|
||||
reply: (text: string) => Promise<void>;
|
||||
sessions: {
|
||||
createSession: ReturnType<typeof vi.fn>;
|
||||
sendMessage: ReturnType<typeof vi.fn>;
|
||||
respondToAskHuman: ReturnType<typeof vi.fn>;
|
||||
stopTurn: ReturnType<typeof vi.fn>;
|
||||
getTurn: ReturnType<typeof vi.fn>;
|
||||
listSessions: ReturnType<typeof vi.fn>;
|
||||
};
|
||||
publish: (turnId: string, event: TurnStreamEvent) => void;
|
||||
}
|
||||
|
||||
function harness(entries: SessionIndexEntry[] = []): Harness {
|
||||
const bus = new EmitterSessionBus();
|
||||
const publish = (turnId: string, event: TurnStreamEvent) =>
|
||||
bus.publish({ kind: "turn-event", sessionId: "s1", turnId, event });
|
||||
const sessions = {
|
||||
createSession: vi.fn(async () => "s1"),
|
||||
sendMessage: vi.fn(async () => ({ turnId: "t1" })),
|
||||
respondToAskHuman: vi.fn(async () => undefined),
|
||||
stopTurn: vi.fn(async () => undefined),
|
||||
getTurn: vi.fn(async () => ({ turnId: "t1", events: [] })),
|
||||
listSessions: vi.fn(() => entries),
|
||||
};
|
||||
const bridge = new ChannelBridge({
|
||||
sessions: sessions as unknown as ISessions,
|
||||
sessionBus: bus,
|
||||
});
|
||||
const replies: string[] = [];
|
||||
const reply = async (text: string) => {
|
||||
replies.push(text);
|
||||
};
|
||||
return { bridge, bus, replies, reply, sessions, publish };
|
||||
}
|
||||
|
||||
// Settle the turn as soon as sendMessage is called: the watcher subscribes
|
||||
// before sendMessage, so a synchronous publish lands in its buffer — the
|
||||
// exact race the buffering exists for.
|
||||
function settleOnSend(h: Harness, event: TurnStreamEvent, turnId = "t1"): void {
|
||||
h.sessions.sendMessage.mockImplementation(async () => {
|
||||
h.publish(turnId, event);
|
||||
return { turnId };
|
||||
});
|
||||
}
|
||||
|
||||
describe("ChannelBridge commands", () => {
|
||||
it("replies with help text", async () => {
|
||||
const h = harness();
|
||||
await h.bridge.handleInbound(SENDER, "help", h.reply);
|
||||
expect(h.replies).toHaveLength(1);
|
||||
expect(h.replies[0]).toContain("Rowboat commands");
|
||||
});
|
||||
|
||||
it("lists recent sessions newest-first and resumes by index", async () => {
|
||||
const h = harness([
|
||||
entry({ sessionId: "old", title: "Old chat", updatedAt: "2026-07-01T00:00:00Z" }),
|
||||
entry({ sessionId: "new", title: "New chat", updatedAt: "2026-07-02T00:00:00Z" }),
|
||||
]);
|
||||
await h.bridge.handleInbound(SENDER, "list", h.reply);
|
||||
expect(h.replies[0]).toContain("1. New chat");
|
||||
expect(h.replies[0]).toContain("2. Old chat");
|
||||
|
||||
await h.bridge.handleInbound(SENDER, "resume 2", h.reply);
|
||||
expect(h.replies[1]).toContain('Resumed "Old chat"');
|
||||
|
||||
settleOnSend(h, completedEvent("t1", "done"));
|
||||
await h.bridge.handleInbound(SENDER, "hello again", h.reply);
|
||||
expect(h.sessions.sendMessage).toHaveBeenCalledWith(
|
||||
"old",
|
||||
expect.objectContaining({ content: "hello again" }),
|
||||
expect.objectContaining({ autoPermission: true }),
|
||||
);
|
||||
});
|
||||
|
||||
it("rejects resume with an out-of-range index", async () => {
|
||||
const h = harness([entry({ sessionId: "s1", title: "Only chat" })]);
|
||||
await h.bridge.handleInbound(SENDER, "resume 5", h.reply);
|
||||
expect(h.replies[0]).toContain("No chat #5");
|
||||
});
|
||||
});
|
||||
|
||||
describe("ChannelBridge message flow", () => {
|
||||
it("creates a session, acks, and delivers the completed text", async () => {
|
||||
const h = harness();
|
||||
settleOnSend(h, completedEvent("t1", "The answer is 42."));
|
||||
await h.bridge.handleInbound(SENDER, "what is the answer?", h.reply);
|
||||
expect(h.sessions.createSession).toHaveBeenCalledOnce();
|
||||
expect(h.replies[0]).toContain("Working on it");
|
||||
expect(h.replies[1]).toBe("The answer is 42.");
|
||||
});
|
||||
|
||||
it("delivers a failed turn as an error reply", async () => {
|
||||
const h = harness();
|
||||
settleOnSend(h, {
|
||||
type: "turn_failed",
|
||||
turnId: "t1",
|
||||
ts: "2026-07-01T00:00:00Z",
|
||||
error: "model exploded",
|
||||
usage: {},
|
||||
} as unknown as TurnStreamEvent);
|
||||
await h.bridge.handleInbound(SENDER, "do a thing", h.reply);
|
||||
expect(h.replies[1]).toContain("model exploded");
|
||||
});
|
||||
|
||||
it("relays the ask-human question text and options, then routes the answer", async () => {
|
||||
const h = harness();
|
||||
settleOnSend(h, askEvent("t1", "Which lane?", ["fast", "slow"]));
|
||||
await h.bridge.handleInbound(SENDER, "start task", h.reply);
|
||||
expect(h.replies[1]).toContain("Which lane?");
|
||||
expect(h.replies[1]).toContain("1. fast");
|
||||
|
||||
// The answer resolves the ask; the turn then completes.
|
||||
h.sessions.respondToAskHuman.mockImplementation(async () => {
|
||||
h.publish("t1", completedEvent("t1", "Took the fast lane."));
|
||||
});
|
||||
await h.bridge.handleInbound(SENDER, "fast", h.reply);
|
||||
expect(h.sessions.respondToAskHuman).toHaveBeenCalledWith("t1", "call_1", "fast");
|
||||
expect(h.replies[3]).toBe("Took the fast lane.");
|
||||
});
|
||||
|
||||
it("falls back to a normal message when the ask was already answered elsewhere", async () => {
|
||||
const h = harness();
|
||||
settleOnSend(h, askEvent("t1", "Which lane?"));
|
||||
await h.bridge.handleInbound(SENDER, "start task", h.reply);
|
||||
|
||||
h.sessions.respondToAskHuman.mockRejectedValue(
|
||||
new TurnInputError("no pending async tool call call_1"),
|
||||
);
|
||||
settleOnSend(h, completedEvent("t2", "Handled as a new message."), "t2");
|
||||
await h.bridge.handleInbound(SENDER, "actually do this instead", h.reply);
|
||||
expect(h.sessions.sendMessage).toHaveBeenLastCalledWith(
|
||||
"s1",
|
||||
expect.objectContaining({ content: "actually do this instead" }),
|
||||
expect.anything(),
|
||||
);
|
||||
expect(h.replies.at(-1)).toBe("Handled as a new message.");
|
||||
});
|
||||
|
||||
it("reports busy while a turn is in flight", async () => {
|
||||
const h = harness();
|
||||
let releaseTurn!: () => void;
|
||||
h.sessions.sendMessage.mockImplementation(async () => {
|
||||
releaseTurn = () => h.publish("t1", completedEvent("t1", "finally"));
|
||||
return { turnId: "t1" };
|
||||
});
|
||||
const first = h.bridge.handleInbound(SENDER, "slow task", h.reply);
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
await h.bridge.handleInbound(SENDER, "impatient follow-up", h.reply);
|
||||
expect(h.replies.some((r) => r.includes("Still working"))).toBe(true);
|
||||
releaseTurn();
|
||||
await first;
|
||||
expect(h.replies.at(-1)).toBe("finally");
|
||||
});
|
||||
});
|
||||
|
|
@ -1,6 +1,8 @@
|
|||
import type { SessionIndexEntry } from "@x/shared/dist/sessions.js";
|
||||
import { reduceTurn, type TurnStreamEvent } from "@x/shared/dist/turns.js";
|
||||
import { assistantText, lastAssistantText } from "../agents/headless.js";
|
||||
import { TurnInputError } from "../turns/api.js";
|
||||
import { ASK_HUMAN_TOOL } from "../turns/bridges/real-agent-resolver.js";
|
||||
import { TurnNotSettledError, type ISessions } from "../sessions/api.js";
|
||||
import type { EmitterSessionBus } from "../sessions/bus.js";
|
||||
|
||||
|
|
@ -18,6 +20,8 @@ const LIST_LIMIT = 10;
|
|||
const REPLY_CHUNK_SIZE = 3500;
|
||||
const MAX_REPLY_CHUNKS = 3;
|
||||
|
||||
const ASK_HUMAN_TOOL_ID = `builtin:${ASK_HUMAN_TOOL}`;
|
||||
|
||||
const HELP_TEXT = [
|
||||
"🤖 Rowboat commands:",
|
||||
"• list — recent chats",
|
||||
|
|
@ -57,13 +61,13 @@ function settleOf(event: TurnStreamEvent): Settled | null {
|
|||
return { kind: "cancelled" };
|
||||
case "turn_suspended": {
|
||||
const ask = event.pendingAsyncTools.find(
|
||||
(t) => t.toolId === "builtin:ask-human" || t.toolName === "ask-human",
|
||||
(t) => t.toolId === ASK_HUMAN_TOOL_ID || t.toolName === ASK_HUMAN_TOOL,
|
||||
);
|
||||
if (ask) {
|
||||
const input = ask.input as { query?: unknown; options?: unknown } | null;
|
||||
const input = ask.input as { question?: unknown; options?: unknown } | null;
|
||||
const query =
|
||||
typeof input?.query === "string" && input.query
|
||||
? input.query
|
||||
typeof input?.question === "string" && input.question
|
||||
? input.question
|
||||
: "The agent needs your input.";
|
||||
const options = Array.isArray(input?.options)
|
||||
? input.options.filter((o): o is string => typeof o === "string")
|
||||
|
|
@ -108,6 +112,11 @@ function chunkReply(text: string): string[] {
|
|||
return parts;
|
||||
}
|
||||
|
||||
interface TurnWatcher {
|
||||
waitFor(turnId: string, timeoutMs: number): Promise<Settled>;
|
||||
dispose(): void;
|
||||
}
|
||||
|
||||
export class ChannelBridge {
|
||||
private senders = new Map<string, SenderState>();
|
||||
|
||||
|
|
@ -175,15 +184,20 @@ export class ChannelBridge {
|
|||
return state;
|
||||
}
|
||||
|
||||
private sortedSessions(): SessionIndexEntry[] {
|
||||
private sessionEntry(sessionId: string): SessionIndexEntry | undefined {
|
||||
return this.deps.sessions.listSessions().find((e) => e.sessionId === sessionId);
|
||||
}
|
||||
|
||||
private recentSessions(): SessionIndexEntry[] {
|
||||
return this.deps.sessions
|
||||
.listSessions()
|
||||
.filter((e) => !e.error)
|
||||
.sort((a, b) => b.updatedAt.localeCompare(a.updatedAt));
|
||||
.sort((a, b) => b.updatedAt.localeCompare(a.updatedAt))
|
||||
.slice(0, LIST_LIMIT);
|
||||
}
|
||||
|
||||
private renderList(state: SenderState): string {
|
||||
const entries = this.sortedSessions().slice(0, LIST_LIMIT);
|
||||
const entries = this.recentSessions();
|
||||
if (entries.length === 0) {
|
||||
return "No chats yet — just send a message to start one.";
|
||||
}
|
||||
|
|
@ -205,9 +219,7 @@ export class ChannelBridge {
|
|||
|
||||
private resumeSession(state: SenderState, index: number): string {
|
||||
if (state.lastList.length === 0) {
|
||||
state.lastList = this.sortedSessions()
|
||||
.slice(0, LIST_LIMIT)
|
||||
.map((e) => e.sessionId);
|
||||
state.lastList = this.recentSessions().map((e) => e.sessionId);
|
||||
}
|
||||
const sessionId = state.lastList[index - 1];
|
||||
if (!sessionId) {
|
||||
|
|
@ -215,7 +227,7 @@ export class ChannelBridge {
|
|||
}
|
||||
state.activeSessionId = sessionId;
|
||||
state.pendingAsk = null;
|
||||
const entry = this.sortedSessions().find((e) => e.sessionId === sessionId);
|
||||
const entry = this.sessionEntry(sessionId);
|
||||
return `▶️ Resumed "${entry?.title ?? "Untitled"}" — send a message to continue.`;
|
||||
}
|
||||
|
||||
|
|
@ -223,9 +235,7 @@ export class ChannelBridge {
|
|||
if (!state.activeSessionId) {
|
||||
return "No current chat — your next message starts a new one.";
|
||||
}
|
||||
const entry = this.sortedSessions().find(
|
||||
(e) => e.sessionId === state.activeSessionId,
|
||||
);
|
||||
const entry = this.sessionEntry(state.activeSessionId);
|
||||
if (!entry) return "Current chat no longer exists — send a message to start fresh.";
|
||||
const status = state.busy
|
||||
? "working"
|
||||
|
|
@ -238,9 +248,7 @@ export class ChannelBridge {
|
|||
private async stopActive(state: SenderState): Promise<string> {
|
||||
state.pendingAsk = null;
|
||||
if (!state.activeSessionId) return "Nothing to stop.";
|
||||
const entry = this.deps.sessions
|
||||
.listSessions()
|
||||
.find((e) => e.sessionId === state.activeSessionId);
|
||||
const entry = this.sessionEntry(state.activeSessionId);
|
||||
if (!entry?.latestTurnId) return "Nothing to stop.";
|
||||
if (
|
||||
entry.latestTurnStatus === "completed" ||
|
||||
|
|
@ -259,30 +267,18 @@ export class ChannelBridge {
|
|||
return;
|
||||
}
|
||||
state.busy = true;
|
||||
// Subscribe before advancing so no settle event can slip past.
|
||||
const watcher = this.watchBus();
|
||||
try {
|
||||
let turnId: string;
|
||||
await reply("⏳ Working on it…");
|
||||
if (state.pendingAsk) {
|
||||
const ask = state.pendingAsk;
|
||||
state.pendingAsk = null;
|
||||
turnId = ask.turnId;
|
||||
await reply("⏳ Working on it…");
|
||||
await this.deps.sessions.respondToAskHuman(ask.turnId, ask.toolCallId, text);
|
||||
} else {
|
||||
if (!state.activeSessionId) {
|
||||
state.activeSessionId = await this.deps.sessions.createSession();
|
||||
}
|
||||
await reply("⏳ Working on it…");
|
||||
const sent = await this.deps.sessions.sendMessage(
|
||||
state.activeSessionId,
|
||||
{ role: "user", content: text },
|
||||
{ agent: { agentId: AGENT_ID }, autoPermission: true },
|
||||
);
|
||||
turnId = sent.turnId;
|
||||
const answered = await this.answerAsk(state, ask, text, reply);
|
||||
if (answered) return;
|
||||
// The ask was already resolved elsewhere (e.g. answered in the
|
||||
// desktop UI) or the turn is terminal — treat the text as a
|
||||
// normal message instead of discarding it.
|
||||
}
|
||||
const settled = await watcher.waitFor(turnId, TURN_TIMEOUT_MS);
|
||||
await this.deliverSettled(state, turnId, settled, reply);
|
||||
await this.sendToSession(state, text, reply);
|
||||
} catch (error) {
|
||||
if (error instanceof TurnNotSettledError) {
|
||||
await reply(
|
||||
|
|
@ -292,11 +288,61 @@ export class ChannelBridge {
|
|||
}
|
||||
throw error;
|
||||
} finally {
|
||||
watcher.dispose();
|
||||
state.busy = false;
|
||||
}
|
||||
}
|
||||
|
||||
private async sendToSession(state: SenderState, text: string, reply: ReplyFn): Promise<void> {
|
||||
// Subscribe before advancing so no settle event can slip past.
|
||||
const watcher = this.watchBus();
|
||||
try {
|
||||
if (!state.activeSessionId) {
|
||||
state.activeSessionId = await this.deps.sessions.createSession();
|
||||
}
|
||||
const sent = await this.deps.sessions.sendMessage(
|
||||
state.activeSessionId,
|
||||
{ role: "user", content: text },
|
||||
{ agent: { agentId: AGENT_ID }, autoPermission: true },
|
||||
);
|
||||
const settled = await watcher.waitFor(sent.turnId, TURN_TIMEOUT_MS);
|
||||
await this.deliverSettled(state, sent.turnId, settled, reply);
|
||||
} finally {
|
||||
watcher.dispose();
|
||||
}
|
||||
}
|
||||
|
||||
// Returns false when the ask was stale (already answered on the desktop /
|
||||
// turn terminal) — the caller then routes the text as a normal message.
|
||||
private async answerAsk(
|
||||
state: SenderState,
|
||||
ask: { turnId: string; toolCallId: string },
|
||||
text: string,
|
||||
reply: ReplyFn,
|
||||
): Promise<boolean> {
|
||||
const watcher = this.watchBus();
|
||||
try {
|
||||
const settledPromise = watcher.waitFor(ask.turnId, TURN_TIMEOUT_MS);
|
||||
// respondToAskHuman resolves only when the whole advance settles,
|
||||
// so it must not be awaited ahead of the watcher (that would
|
||||
// bypass TURN_TIMEOUT_MS). Race instead: its rejection (stale
|
||||
// ask) must beat the 30-minute timeout; its success defers to the
|
||||
// settle event.
|
||||
const settled = await Promise.race([
|
||||
settledPromise,
|
||||
this.deps.sessions
|
||||
.respondToAskHuman(ask.turnId, ask.toolCallId, text)
|
||||
.then(() => settledPromise),
|
||||
]);
|
||||
await this.deliverSettled(state, ask.turnId, settled, reply);
|
||||
return true;
|
||||
} catch (error) {
|
||||
if (error instanceof TurnInputError) return false;
|
||||
throw error;
|
||||
} finally {
|
||||
watcher.dispose();
|
||||
}
|
||||
}
|
||||
|
||||
private async deliverSettled(
|
||||
state: SenderState,
|
||||
turnId: string,
|
||||
|
|
@ -350,38 +396,35 @@ export class ChannelBridge {
|
|||
}
|
||||
}
|
||||
|
||||
// Buffers turn events from the moment of subscription so a settle that
|
||||
// fires between sendMessage() returning and waitFor() attaching is never
|
||||
// lost. One watcher per in-flight message; disposed in runMessage.
|
||||
private watchBus() {
|
||||
const buffered: Array<{ turnId: string; event: TurnStreamEvent }> = [];
|
||||
// Buffers settle-relevant events (≈1 per turn) from the moment of
|
||||
// subscription so a settle firing between advance-start and waitFor() is
|
||||
// never lost — without retaining the per-token delta stream of every
|
||||
// concurrent session. One watcher per in-flight message.
|
||||
private watchBus(): TurnWatcher {
|
||||
const buffered: Array<{ turnId: string; settled: Settled }> = [];
|
||||
let waiter: { turnId: string; resolve: (settled: Settled) => void } | null = null;
|
||||
let cancelTimer: (() => void) | null = null;
|
||||
const unsubscribe = this.deps.sessionBus.subscribe((event) => {
|
||||
if (event.kind !== "turn-event") return;
|
||||
const settled = settleOf(event.event);
|
||||
if (!settled) return;
|
||||
if (waiter) {
|
||||
if (event.turnId !== waiter.turnId) return;
|
||||
const settled = settleOf(event.event);
|
||||
if (settled) waiter.resolve(settled);
|
||||
if (event.turnId === waiter.turnId) waiter.resolve(settled);
|
||||
return;
|
||||
}
|
||||
buffered.push({ turnId: event.turnId, event: event.event });
|
||||
buffered.push({ turnId: event.turnId, settled });
|
||||
});
|
||||
return {
|
||||
waitFor: (turnId: string, timeoutMs: number): Promise<Settled> =>
|
||||
new Promise<Settled>((resolve) => {
|
||||
for (const b of buffered) {
|
||||
if (b.turnId !== turnId) continue;
|
||||
const settled = settleOf(b.event);
|
||||
if (settled) {
|
||||
resolve(settled);
|
||||
return;
|
||||
}
|
||||
const hit = buffered.find((b) => b.turnId === turnId);
|
||||
if (hit) {
|
||||
resolve(hit.settled);
|
||||
return;
|
||||
}
|
||||
buffered.length = 0;
|
||||
const timer = setTimeout(
|
||||
() => resolve({ kind: "timeout" }),
|
||||
timeoutMs,
|
||||
);
|
||||
const timer = setTimeout(() => resolve({ kind: "timeout" }), timeoutMs);
|
||||
cancelTimer = () => clearTimeout(timer);
|
||||
waiter = {
|
||||
turnId,
|
||||
resolve: (settled) => {
|
||||
|
|
@ -390,7 +433,10 @@ export class ChannelBridge {
|
|||
},
|
||||
};
|
||||
}),
|
||||
dispose: unsubscribe,
|
||||
dispose: () => {
|
||||
unsubscribe();
|
||||
cancelTimer?.();
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
import path from "node:path";
|
||||
import fs from "node:fs/promises";
|
||||
import QRCode from "qrcode";
|
||||
import type { z } from "zod";
|
||||
import type { ChannelsConfig, ChannelsStatus } from "@x/shared/dist/channels.js";
|
||||
import container from "../di/container.js";
|
||||
|
|
@ -10,7 +9,9 @@ import type { EmitterSessionBus } from "../sessions/bus.js";
|
|||
import { ChannelBridge } from "./bridge.js";
|
||||
import type { IChannelsConfigRepo } from "./repo.js";
|
||||
import { TelegramTransport } from "./transports/telegram.js";
|
||||
import { WhatsAppTransport } from "./transports/whatsapp.js";
|
||||
// Type-only: the real module (which pulls the ~9MB baileys dependency) is
|
||||
// loaded dynamically in startWhatsApp, so boot pays nothing while disabled.
|
||||
import type { WhatsAppTransport } from "./transports/whatsapp.js";
|
||||
|
||||
// Lifecycle owner for the mobile channels: reads config, runs the enabled
|
||||
// transports against one shared ChannelBridge, and fans status out to the
|
||||
|
|
@ -21,6 +22,7 @@ type Config = z.infer<typeof ChannelsConfig>;
|
|||
type Status = z.infer<typeof ChannelsStatus>;
|
||||
|
||||
const WHATSAPP_AUTH_DIR = path.join(WorkDir, "channels", "whatsapp-auth");
|
||||
const TELEGRAM_STATE_FILE = path.join(WorkDir, "channels", "telegram-state.json");
|
||||
|
||||
let bridge: ChannelBridge | null = null;
|
||||
let whatsapp: WhatsAppTransport | null = null;
|
||||
|
|
@ -34,13 +36,22 @@ const status: Status = {
|
|||
const statusListeners = new Set<(status: Status) => void>();
|
||||
|
||||
// Serializes apply/logout so a fast settings double-save can't interleave
|
||||
// transport teardown and startup.
|
||||
// transport teardown and startup. enqueue() recovers the chain before adding
|
||||
// a step — a rejected step must fail only its own caller, never poison every
|
||||
// later settings save.
|
||||
let lifecycle: Promise<void> = Promise.resolve();
|
||||
|
||||
function enqueue(step: () => Promise<void>): Promise<void> {
|
||||
const run = lifecycle.catch(() => undefined).then(step);
|
||||
lifecycle = run.catch(() => undefined);
|
||||
return run;
|
||||
}
|
||||
|
||||
function notifyStatus(): void {
|
||||
const snapshot = structuredClone(status);
|
||||
for (const listener of statusListeners) {
|
||||
try {
|
||||
listener(structuredClone(status));
|
||||
listener(snapshot);
|
||||
} catch {
|
||||
// observers must never affect the channels
|
||||
}
|
||||
|
|
@ -78,37 +89,67 @@ function ensureBridge(): ChannelBridge {
|
|||
|
||||
async function stopWhatsApp(): Promise<void> {
|
||||
if (!whatsapp) return;
|
||||
await whatsapp.stop().catch(() => undefined);
|
||||
const stopping = whatsapp;
|
||||
whatsapp = null;
|
||||
await stopping.stop().catch(() => undefined);
|
||||
setWhatsAppStatus({ state: "disabled" });
|
||||
}
|
||||
|
||||
function stopTelegram(): void {
|
||||
if (!telegram) return;
|
||||
telegram.stop();
|
||||
const stopping = telegram;
|
||||
telegram = null;
|
||||
stopping.stop();
|
||||
setTelegramStatus({ state: "disabled" });
|
||||
}
|
||||
|
||||
function startWhatsApp(config: Config["whatsapp"]): void {
|
||||
// Invalidates pending async QR renders whenever a newer status lands.
|
||||
let qrSeq = 0;
|
||||
|
||||
async function startWhatsApp(config: Config["whatsapp"]): Promise<void> {
|
||||
if (!config.enabled) {
|
||||
setWhatsAppStatus({ state: "disabled" });
|
||||
return;
|
||||
}
|
||||
const channelBridge = ensureBridge();
|
||||
const transport = new WhatsAppTransport({
|
||||
const [{ WhatsAppTransport: Transport }, QRCode] = await Promise.all([
|
||||
import("./transports/whatsapp.js"),
|
||||
import("qrcode").then((m) => m.default),
|
||||
]);
|
||||
const transport = new Transport({
|
||||
authDir: WHATSAPP_AUTH_DIR,
|
||||
allowFrom: config.allowFrom,
|
||||
onInbound: (senderKey, text, reply) => {
|
||||
onInbound: (senderKey, chatJid, text) => {
|
||||
// Route replies through whichever transport is current at send
|
||||
// time — the originating instance may have been replaced by a
|
||||
// settings save while the turn was running.
|
||||
const reply = async (replyText: string) => {
|
||||
const current = whatsapp;
|
||||
if (!current) throw new Error("WhatsApp channel is disabled");
|
||||
await current.send(chatJid, replyText);
|
||||
};
|
||||
void channelBridge.handleInbound(senderKey, text, reply);
|
||||
},
|
||||
onStatus: (update) => {
|
||||
if (whatsapp !== transport) return; // superseded instance
|
||||
if (update.state === "qr" && update.qr) {
|
||||
const seq = ++qrSeq;
|
||||
// Render the pairing QR main-side so the renderer just shows
|
||||
// an <img>; the raw pairing string never leaves core.
|
||||
QRCode.toDataURL(update.qr, { margin: 1, width: 256 })
|
||||
.then((qrDataUrl) => setWhatsAppStatus({ state: "qr", qrDataUrl }))
|
||||
.catch(() => setWhatsAppStatus({ state: "error", error: "Failed to render pairing QR" }));
|
||||
.then((qrDataUrl) => {
|
||||
if (whatsapp === transport && seq === qrSeq) {
|
||||
setWhatsAppStatus({ state: "qr", qrDataUrl });
|
||||
}
|
||||
})
|
||||
.catch(() => {
|
||||
if (whatsapp === transport && seq === qrSeq) {
|
||||
setWhatsAppStatus({ state: "error", error: "Failed to render pairing QR" });
|
||||
}
|
||||
});
|
||||
return;
|
||||
}
|
||||
qrSeq++;
|
||||
setWhatsAppStatus({
|
||||
state: update.state,
|
||||
...(update.self ? { self: update.self } : {}),
|
||||
|
|
@ -118,6 +159,7 @@ function startWhatsApp(config: Config["whatsapp"]): void {
|
|||
});
|
||||
whatsapp = transport;
|
||||
transport.start().catch((error) => {
|
||||
if (whatsapp !== transport) return;
|
||||
setWhatsAppStatus({
|
||||
state: "error",
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
|
|
@ -138,41 +180,49 @@ function startTelegram(config: Config["telegram"]): void {
|
|||
const transport = new TelegramTransport({
|
||||
botToken: config.botToken,
|
||||
allowFrom: config.allowFrom,
|
||||
onInbound: (senderKey, text, reply) => {
|
||||
stateFile: TELEGRAM_STATE_FILE,
|
||||
onInbound: (senderKey, chatId, text) => {
|
||||
const reply = async (replyText: string) => {
|
||||
const current = telegram;
|
||||
if (!current) throw new Error("Telegram channel is disabled");
|
||||
await current.send(chatId, replyText);
|
||||
};
|
||||
void channelBridge.handleInbound(senderKey, text, reply);
|
||||
},
|
||||
onStatus: setTelegramStatus,
|
||||
onStatus: (update) => {
|
||||
if (telegram !== transport) return; // superseded instance
|
||||
setTelegramStatus(update);
|
||||
},
|
||||
});
|
||||
telegram = transport;
|
||||
void transport.start();
|
||||
}
|
||||
|
||||
export function applyChannelsConfig(config: Config): Promise<void> {
|
||||
lifecycle = lifecycle.then(async () => {
|
||||
return enqueue(async () => {
|
||||
await stopWhatsApp();
|
||||
stopTelegram();
|
||||
startWhatsApp(config.whatsapp);
|
||||
await startWhatsApp(config.whatsapp);
|
||||
startTelegram(config.telegram);
|
||||
});
|
||||
return lifecycle;
|
||||
}
|
||||
|
||||
// Unlink the WhatsApp device and, if the channel is still enabled, restart it
|
||||
// so a fresh pairing QR appears. Telegram is left untouched.
|
||||
export function logoutWhatsApp(): Promise<void> {
|
||||
lifecycle = lifecycle.then(async () => {
|
||||
return enqueue(async () => {
|
||||
if (whatsapp) {
|
||||
await whatsapp.logout().catch(() => undefined);
|
||||
const out = whatsapp;
|
||||
whatsapp = null;
|
||||
await out.logout().catch(() => undefined);
|
||||
} else {
|
||||
await fs.rm(WHATSAPP_AUTH_DIR, { recursive: true, force: true });
|
||||
await fs.rm(WHATSAPP_AUTH_DIR, { recursive: true, force: true }).catch(() => undefined);
|
||||
}
|
||||
const config = await container
|
||||
.resolve<IChannelsConfigRepo>("channelsConfigRepo")
|
||||
.getConfig();
|
||||
startWhatsApp(config.whatsapp);
|
||||
await startWhatsApp(config.whatsapp);
|
||||
});
|
||||
return lifecycle;
|
||||
}
|
||||
|
||||
export async function init(): Promise<void> {
|
||||
|
|
|
|||
|
|
@ -1,16 +1,39 @@
|
|||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import type { z } from "zod";
|
||||
import type { TelegramChannelStatus } from "@x/shared/dist/channels.js";
|
||||
import type { ReplyFn } from "../bridge.js";
|
||||
|
||||
// Telegram Bot API transport. Deliberately dependency-free: the Bot API is
|
||||
// plain HTTPS — getUpdates long polling (outbound connection, works behind
|
||||
// NAT) plus sendMessage. The user supplies their own bot token (@BotFather).
|
||||
//
|
||||
// The getUpdates offset is persisted to disk after each processed batch:
|
||||
// Telegram only marks updates confirmed when a LATER getUpdates call passes a
|
||||
// higher offset, so without persistence every transport restart (app relaunch
|
||||
// or settings save) would redeliver — and re-execute — the last batch.
|
||||
|
||||
const POLL_TIMEOUT_S = 50;
|
||||
const RETRY_DELAY_MS = 5000;
|
||||
const MAX_RETRY_DELAY_MS = 60_000;
|
||||
|
||||
type Status = z.infer<typeof TelegramChannelStatus>;
|
||||
|
||||
class TelegramApiError extends Error {
|
||||
constructor(
|
||||
message: string,
|
||||
readonly code?: number,
|
||||
) {
|
||||
super(message);
|
||||
this.name = "TelegramApiError";
|
||||
}
|
||||
}
|
||||
|
||||
// 401 = token revoked/invalid, 404 = bot deleted / malformed token. Retrying
|
||||
// these forever would hammer the API and show a misleading "polling" status.
|
||||
function isTerminal(error: unknown): boolean {
|
||||
return error instanceof TelegramApiError && (error.code === 401 || error.code === 404);
|
||||
}
|
||||
|
||||
interface TelegramUpdate {
|
||||
update_id: number;
|
||||
message?: {
|
||||
|
|
@ -24,7 +47,10 @@ interface TelegramUpdate {
|
|||
export interface TelegramTransportOptions {
|
||||
botToken: string;
|
||||
allowFrom: string[];
|
||||
onInbound: (senderKey: string, text: string, reply: ReplyFn) => void;
|
||||
// JSON file holding { offset } across restarts.
|
||||
stateFile: string;
|
||||
// chatId is the address to reply to; the caller owns reply routing.
|
||||
onInbound: (senderKey: string, chatId: string, text: string) => void;
|
||||
onStatus: (status: Status) => void;
|
||||
}
|
||||
|
||||
|
|
@ -32,6 +58,7 @@ export class TelegramTransport {
|
|||
private abort: AbortController | null = null;
|
||||
private stopped = false;
|
||||
private offset = 0;
|
||||
private botUsername: string | undefined;
|
||||
|
||||
constructor(private readonly opts: TelegramTransportOptions) {}
|
||||
|
||||
|
|
@ -47,40 +74,85 @@ export class TelegramTransport {
|
|||
this.opts.onStatus({ state: "disabled" });
|
||||
}
|
||||
|
||||
private api(method: string): string {
|
||||
return `https://api.telegram.org/bot${this.opts.botToken}/${method}`;
|
||||
}
|
||||
|
||||
private async call(method: string, body?: unknown, signal?: AbortSignal): Promise<unknown> {
|
||||
const res = await fetch(this.api(method), {
|
||||
const res = await fetch(`https://api.telegram.org/bot${this.opts.botToken}/${method}`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: body === undefined ? undefined : JSON.stringify(body),
|
||||
...(signal ? { signal } : {}),
|
||||
});
|
||||
const payload = (await res.json()) as { ok: boolean; result?: unknown; description?: string };
|
||||
const payload = (await res.json()) as {
|
||||
ok: boolean;
|
||||
result?: unknown;
|
||||
description?: string;
|
||||
error_code?: number;
|
||||
};
|
||||
if (!payload.ok) {
|
||||
throw new Error(payload.description ?? `Telegram API error (${method})`);
|
||||
throw new TelegramApiError(
|
||||
payload.description ?? `Telegram API error (${method})`,
|
||||
payload.error_code,
|
||||
);
|
||||
}
|
||||
return payload.result;
|
||||
}
|
||||
|
||||
private async run(): Promise<void> {
|
||||
// Validate the token up front so a typo surfaces in settings
|
||||
// immediately instead of as a silent poll loop failure.
|
||||
private async loadOffset(): Promise<void> {
|
||||
try {
|
||||
const me = (await this.call("getMe")) as { username?: string };
|
||||
if (this.stopped) return;
|
||||
this.opts.onStatus({ state: "polling", botUsername: me.username });
|
||||
} catch (error) {
|
||||
if (this.stopped) return;
|
||||
this.opts.onStatus({
|
||||
state: "error",
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
return;
|
||||
const raw = await fs.readFile(this.opts.stateFile, "utf8");
|
||||
const parsed = JSON.parse(raw) as { offset?: unknown };
|
||||
if (typeof parsed.offset === "number" && Number.isFinite(parsed.offset)) {
|
||||
this.offset = parsed.offset;
|
||||
}
|
||||
} catch {
|
||||
// first run or unreadable state — start from 0
|
||||
}
|
||||
}
|
||||
|
||||
private async saveOffset(): Promise<void> {
|
||||
try {
|
||||
await fs.mkdir(path.dirname(this.opts.stateFile), { recursive: true });
|
||||
await fs.writeFile(this.opts.stateFile, JSON.stringify({ offset: this.offset }));
|
||||
} catch {
|
||||
// best effort — worst case is one redelivered batch after restart
|
||||
}
|
||||
}
|
||||
|
||||
private async sleep(ms: number): Promise<void> {
|
||||
await new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
private async run(): Promise<void> {
|
||||
await this.loadOffset();
|
||||
|
||||
// Validate the token, retrying transient failures (the app often
|
||||
// starts at login before the network is up). Only a definitive
|
||||
// API rejection is terminal.
|
||||
let delay = RETRY_DELAY_MS;
|
||||
while (!this.stopped) {
|
||||
try {
|
||||
const me = (await this.call("getMe")) as { username?: string };
|
||||
if (this.stopped) return;
|
||||
this.botUsername = me.username;
|
||||
this.opts.onStatus({ state: "polling", botUsername: me.username });
|
||||
break;
|
||||
} catch (error) {
|
||||
if (this.stopped) return;
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
if (isTerminal(error)) {
|
||||
this.opts.onStatus({
|
||||
state: "error",
|
||||
error: `Bot token rejected (${message}) — create a new token with @BotFather.`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
this.opts.onStatus({ state: "error", error: message });
|
||||
await this.sleep(delay);
|
||||
delay = Math.min(delay * 2, MAX_RETRY_DELAY_MS);
|
||||
}
|
||||
}
|
||||
|
||||
delay = RETRY_DELAY_MS;
|
||||
let healthy = true;
|
||||
while (!this.stopped) {
|
||||
this.abort = new AbortController();
|
||||
try {
|
||||
|
|
@ -94,18 +166,32 @@ export class TelegramTransport {
|
|||
this.abort.signal,
|
||||
)) as TelegramUpdate[];
|
||||
for (const update of updates) {
|
||||
this.offset = Math.max(this.offset, update.update_id + 1);
|
||||
this.offset = update.update_id + 1;
|
||||
this.handleUpdate(update);
|
||||
}
|
||||
if (updates.length > 0) {
|
||||
await this.saveOffset();
|
||||
}
|
||||
if (!healthy) {
|
||||
// Restore the healthy status only after a successful poll.
|
||||
healthy = true;
|
||||
this.opts.onStatus({ state: "polling", botUsername: this.botUsername });
|
||||
}
|
||||
delay = RETRY_DELAY_MS;
|
||||
} catch (error) {
|
||||
if (this.stopped) return;
|
||||
this.opts.onStatus({
|
||||
state: "error",
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
await new Promise((resolve) => setTimeout(resolve, RETRY_DELAY_MS));
|
||||
if (this.stopped) return;
|
||||
this.opts.onStatus({ state: "polling" });
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
if (isTerminal(error)) {
|
||||
this.opts.onStatus({
|
||||
state: "error",
|
||||
error: `Bot token rejected (${message}) — create a new token with @BotFather.`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
healthy = false;
|
||||
this.opts.onStatus({ state: "error", error: message });
|
||||
await this.sleep(delay);
|
||||
delay = Math.min(delay * 2, MAX_RETRY_DELAY_MS);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -116,7 +202,6 @@ export class TelegramTransport {
|
|||
// DMs only: group chats would let any member drive the bridge.
|
||||
if (message.chat.type !== "private") return;
|
||||
const chatId = String(message.chat.id);
|
||||
const reply: ReplyFn = (text) => this.send(chatId, text);
|
||||
if (!this.opts.allowFrom.includes(chatId)) {
|
||||
void this.send(
|
||||
chatId,
|
||||
|
|
@ -124,10 +209,10 @@ export class TelegramTransport {
|
|||
).catch(() => undefined);
|
||||
return;
|
||||
}
|
||||
this.opts.onInbound(`telegram:${chatId}`, message.text, reply);
|
||||
this.opts.onInbound(`telegram:${chatId}`, chatId, message.text);
|
||||
}
|
||||
|
||||
private async send(chatId: string, text: string): Promise<void> {
|
||||
async send(chatId: string, text: string): Promise<void> {
|
||||
await this.call("sendMessage", { chat_id: chatId, text });
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,7 +6,6 @@ import makeWASocket, {
|
|||
jidDecode,
|
||||
useMultiFileAuthState,
|
||||
} from "baileys";
|
||||
import type { ReplyFn } from "../bridge.js";
|
||||
|
||||
// WhatsApp transport via Baileys: the app links to the user's own WhatsApp
|
||||
// account as a linked device (QR pairing, same as WhatsApp Web) over an
|
||||
|
|
@ -34,7 +33,9 @@ export interface WhatsAppTransportStatus {
|
|||
export interface WhatsAppTransportOptions {
|
||||
authDir: string;
|
||||
allowFrom: string[];
|
||||
onInbound: (senderKey: string, text: string, reply: ReplyFn) => void;
|
||||
// chatJid is the address to reply to; the caller owns reply routing so a
|
||||
// reply can go through whichever transport instance is current by then.
|
||||
onInbound: (senderKey: string, chatJid: string, text: string) => void;
|
||||
onStatus: (status: WhatsAppTransportStatus) => void;
|
||||
}
|
||||
|
||||
|
|
@ -45,7 +46,13 @@ interface TextishMessage {
|
|||
}
|
||||
|
||||
interface InboundWAMessage {
|
||||
key?: { remoteJid?: string | null; fromMe?: boolean | null; id?: string | null };
|
||||
key?: {
|
||||
remoteJid?: string | null;
|
||||
// Phone-number JID when remoteJid is a LID (anonymized) JID.
|
||||
remoteJidAlt?: string | null;
|
||||
fromMe?: boolean | null;
|
||||
id?: string | null;
|
||||
};
|
||||
message?: unknown;
|
||||
}
|
||||
|
||||
|
|
@ -60,6 +67,11 @@ function messageText(message: unknown): string | null {
|
|||
export class WhatsAppTransport {
|
||||
private sock: WASocket | null = null;
|
||||
private stopped = false;
|
||||
// Bumped on every connect/stop/logout; handlers close over their own
|
||||
// generation and go inert the moment they are superseded, so a stop()
|
||||
// racing an await inside connect() cannot leave a zombie socket
|
||||
// processing messages alongside its replacement.
|
||||
private generation = 0;
|
||||
private sentIds = new Set<string>();
|
||||
|
||||
constructor(private readonly opts: WhatsAppTransportOptions) {}
|
||||
|
|
@ -72,6 +84,7 @@ export class WhatsAppTransport {
|
|||
|
||||
async stop(): Promise<void> {
|
||||
this.stopped = true;
|
||||
this.generation++;
|
||||
try {
|
||||
this.sock?.end(undefined);
|
||||
} catch {
|
||||
|
|
@ -85,6 +98,7 @@ export class WhatsAppTransport {
|
|||
// local credentials so the next start shows a fresh QR.
|
||||
async logout(): Promise<void> {
|
||||
this.stopped = true;
|
||||
this.generation++;
|
||||
try {
|
||||
await this.sock?.logout();
|
||||
} catch {
|
||||
|
|
@ -97,18 +111,22 @@ export class WhatsAppTransport {
|
|||
|
||||
private async connect(): Promise<void> {
|
||||
if (this.stopped) return;
|
||||
const generation = ++this.generation;
|
||||
const { state, saveCreds } = await useMultiFileAuthState(this.opts.authDir);
|
||||
if (this.stopped || generation !== this.generation) return;
|
||||
const sock = makeWASocket({
|
||||
auth: state,
|
||||
syncFullHistory: false,
|
||||
markOnlineOnConnect: false,
|
||||
});
|
||||
this.sock = sock;
|
||||
const isCurrent = () =>
|
||||
!this.stopped && generation === this.generation && this.sock === sock;
|
||||
|
||||
sock.ev.on("creds.update", saveCreds);
|
||||
|
||||
sock.ev.on("connection.update", (update) => {
|
||||
if (this.stopped) return;
|
||||
if (!isCurrent()) return;
|
||||
if (update.qr) {
|
||||
this.opts.onStatus({ state: "qr", qr: update.qr });
|
||||
}
|
||||
|
|
@ -129,6 +147,7 @@ export class WhatsAppTransport {
|
|||
return;
|
||||
}
|
||||
setTimeout(() => {
|
||||
if (!isCurrent()) return;
|
||||
this.connect().catch((error) => {
|
||||
this.opts.onStatus({
|
||||
state: "error",
|
||||
|
|
@ -140,7 +159,7 @@ export class WhatsAppTransport {
|
|||
});
|
||||
|
||||
sock.ev.on("messages.upsert", ({ messages, type }) => {
|
||||
if (type !== "notify") return;
|
||||
if (!isCurrent() || type !== "notify") return;
|
||||
for (const msg of messages) {
|
||||
this.handleMessage(sock, msg);
|
||||
}
|
||||
|
|
@ -155,24 +174,37 @@ export class WhatsAppTransport {
|
|||
const text = messageText(msg.message);
|
||||
if (!text || text.startsWith(REPLY_MARKER)) return;
|
||||
|
||||
const selfIds = [sock.user?.id, (sock.user as { lid?: string } | undefined)?.lid].filter(
|
||||
(v): v is string => Boolean(v),
|
||||
// LID-addressed chats put the anonymized id in remoteJid and (when
|
||||
// the server supplies it) the real phone-number JID in remoteJidAlt.
|
||||
// Identity checks must consider both.
|
||||
const altJid: string | undefined = msg.key?.remoteJidAlt ?? undefined;
|
||||
const chatJids = altJid ? [jid, altJid] : [jid];
|
||||
const user = sock.user as { id?: string; lid?: string } | undefined;
|
||||
const selfIds = [user?.id, user?.lid].filter((v): v is string => Boolean(v));
|
||||
const isSelfChat = chatJids.some((j) =>
|
||||
selfIds.some((selfId) => areJidsSameUser(j, selfId)),
|
||||
);
|
||||
const isSelfChat = selfIds.some((selfId) => areJidsSameUser(jid, selfId));
|
||||
const senderNumber = jidDecode(jid)?.user ?? "";
|
||||
const senderNumbers = chatJids.flatMap((j) => {
|
||||
const decoded = jidDecode(j)?.user;
|
||||
return decoded ? [decoded] : [];
|
||||
});
|
||||
|
||||
// Self-chat is the owner by definition. Anyone else must be
|
||||
// allowlisted — this bridge is remote control over the desktop agent.
|
||||
if (!isSelfChat) {
|
||||
if (msg.key?.fromMe) return;
|
||||
if (!this.opts.allowFrom.includes(senderNumber)) return;
|
||||
if (!senderNumbers.some((n) => this.opts.allowFrom.includes(n))) return;
|
||||
}
|
||||
|
||||
const reply: ReplyFn = (replyText) => this.send(jid, replyText);
|
||||
this.opts.onInbound(`whatsapp:${senderNumber || jid}`, text, reply);
|
||||
// Prefer the phone number (altJid decodes to it when present) as the
|
||||
// stable sender identity.
|
||||
const senderId = altJid
|
||||
? (jidDecode(altJid)?.user ?? senderNumbers[0] ?? jid)
|
||||
: (senderNumbers[0] ?? jid);
|
||||
this.opts.onInbound(`whatsapp:${senderId}`, jid, text);
|
||||
}
|
||||
|
||||
private async send(jid: string, text: string): Promise<void> {
|
||||
async send(jid: string, text: string): Promise<void> {
|
||||
const sock = this.sock;
|
||||
if (!sock) throw new Error("WhatsApp is not connected");
|
||||
const sent = await sock.sendMessage(jid, { text: `${REPLY_MARKER}${text}` });
|
||||
|
|
@ -180,11 +212,7 @@ export class WhatsAppTransport {
|
|||
if (id) {
|
||||
this.sentIds.add(id);
|
||||
if (this.sentIds.size > 500) {
|
||||
// Trim oldest; Set iteration order is insertion order.
|
||||
for (const old of this.sentIds) {
|
||||
this.sentIds.delete(old);
|
||||
if (this.sentIds.size <= 250) break;
|
||||
}
|
||||
this.sentIds = new Set(Array.from(this.sentIds).slice(-250));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue