let5see/clawdbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

gateway: force ws-only clients

Browse Source
This commit is contained in:
Peter Steinberger
2025-12-10 16:27:54 +00:00
parent c2adda1cfe
commit 55772eec5a
9 changed files with 80 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
src/cli/program.ts
View File

@@ -14,9 +14,9 @@ import { defaultRuntime } from "../runtime.js";
import { VERSION } from "../version.js";
import { startWebChatServer } from "../webchat/server.js";
import { createDefaultDeps } from "./deps.js";
import { forceFreePort, listPortListeners, parseLsofOutput } from "./ports.js";
import { forceFreePort } from "./ports.js";
export { forceFreePort, listPortListeners, parseLsofOutput };
export { forceFreePort };
export function buildProgram() {
const program = new Command();
@@ -136,11 +136,6 @@ export function buildProgram() {
"--provider <provider>",
"Delivery provider: whatsapp|telegram (default: whatsapp)",
)
.option(
"--spawn-gateway",
"Start a local gateway on 127.0.0.1:18789 if none is running",
false,
)
.option("--dry-run", "Print payload and skip sending", false)
.option("--json", "Output result as JSON", false)
.option("--verbose", "Verbose logging", false)
@@ -280,50 +275,23 @@ Examples:
.option("--url <url>", "Gateway WebSocket URL", "ws://127.0.0.1:18789")
.option("--token <token>", "Gateway token (if required)")
.option("--timeout <ms>", "Timeout in ms", "10000")
.option("--expect-final", "Wait for final response (agent)", false)
.option(
"--spawn-gateway",
"Start a local gateway if none is listening on --url",
false,
);
.option("--expect-final", "Wait for final response (agent)", false);
const callWithSpawn = async (
const callGatewayCli = async (
method: string,
opts: {
url?: string;
token?: string;
timeout?: string;
expectFinal?: boolean;
spawnGateway?: boolean;
},
opts: { url?: string; token?: string; timeout?: string; expectFinal?: boolean },
params?: unknown,
) => {
const timeoutMs = Number(opts.timeout ?? 10_000);
const attempt = async () =>
callGateway({
url: opts.url,
token: opts.token,
method,
params,
expectFinal: Boolean(opts.expectFinal),
timeoutMs,
clientName: "cli",
mode: "cli",
});
try {
return await attempt();
} catch (err) {
if (!opts.spawnGateway) throw err;
// Only spawn if there is clearly no listener.
const url = new URL(opts.url ?? "ws://127.0.0.1:18789");
const port = Number(url.port || 18789);
const listeners = listPortListeners(port);
if (listeners.length > 0) throw err;
await startGatewayServer(port);
return await attempt();
}
};
) =>
callGateway({
url: opts.url,
token: opts.token,
method,
params,
expectFinal: Boolean(opts.expectFinal),
timeoutMs: Number(opts.timeout ?? 10_000),
clientName: "cli",
mode: "cli",
});
gatewayCallOpts(
gateway
@@ -337,7 +305,7 @@ Examples:
.action(async (method, opts) => {
try {
const params = JSON.parse(String(opts.params ?? "{}"));
const result = await callWithSpawn(method, opts, params);
const result = await callGatewayCli(method, opts, params);
defaultRuntime.log(JSON.stringify(result, null, 2));
} catch (err) {
defaultRuntime.error(`Gateway call failed: ${String(err)}`);
@@ -352,7 +320,7 @@ Examples:
.description("Fetch Gateway health")
.action(async (opts) => {
try {
const result = await callWithSpawn("health", opts);
const result = await callGatewayCli("health", opts);
defaultRuntime.log(JSON.stringify(result, null, 2));
} catch (err) {
defaultRuntime.error(String(err));
@@ -367,7 +335,7 @@ Examples:
.description("Fetch Gateway status")
.action(async (opts) => {
try {
const result = await callWithSpawn("status", opts);
const result = await callGatewayCli("status", opts);
defaultRuntime.log(JSON.stringify(result, null, 2));
} catch (err) {
defaultRuntime.error(String(err));
@@ -387,7 +355,7 @@ Examples:
.action(async (opts) => {
try {
const idempotencyKey = opts.idempotencyKey ?? randomIdempotencyKey();
const result = await callWithSpawn("send", opts, {
const result = await callGatewayCli("send", opts, {
to: opts.to,
message: opts.message,
mediaUrl: opts.mediaUrl,
@@ -415,7 +383,7 @@ Examples:
.action(async (opts) => {
try {
const idempotencyKey = opts.idempotencyKey ?? randomIdempotencyKey();
const result = await callWithSpawn(
const result = await callGatewayCli(
"agent",
{ ...opts, expectFinal: true },
{
@@ -482,17 +450,10 @@ Examples:
program
.command("health")
.description(
"Probe WhatsApp Web health (creds + Baileys connect) and session store",
)
.description("Fetch health from the running gateway")
.option("--json", "Output JSON instead of text", false)
.option("--timeout <ms>", "Connection timeout in milliseconds", "10000")
.option("--verbose", "Verbose logging", false)
.option(
"--probe",
"Also attempt a live Baileys connect (can conflict if gateway is already connected)",
true,
)
.action(async (opts) => {
setVerbose(Boolean(opts.verbose));
const timeout = opts.timeout
@@ -510,7 +471,6 @@ Examples:
{
json: Boolean(opts.json),
timeoutMs: timeout,
probe: opts.probe ?? true,
},
defaultRuntime,
);

10
src/commands/health.ts
View File

@@ -12,6 +12,7 @@ import {
waitForWaConnection,
webAuthExists,
} from "../web/session.js";
import { callGateway } from "../gateway/call.js";
type HealthConnect = {
ok: boolean;
@@ -236,12 +237,13 @@ export async function getHealthSnapshot(
}
export async function healthCommand(
opts: { json?: boolean; timeoutMs?: number; probe?: boolean },
opts: { json?: boolean; timeoutMs?: number },
runtime: RuntimeEnv,
) {
const probe = opts.probe ?? true;
const summary = await getHealthSnapshot(opts.timeoutMs, {
probe,
// Always query the running gateway; do not open a direct Baileys socket here.
const summary = await callGateway<HealthSummary>({
method: "health",
timeoutMs: opts.timeoutMs,
});
const fatal =
!summary.web.linked ||

19
src/commands/send.ts
View File

@@ -1,7 +1,5 @@
import type { CliDeps } from "../cli/deps.js";
import { listPortListeners } from "../cli/ports.js";
import { callGateway, randomIdempotencyKey } from "../gateway/call.js";
import { startGatewayServer } from "../gateway/server.js";
import { success } from "../globals.js";
import type { RuntimeEnv } from "../runtime.js";
@@ -13,7 +11,6 @@ export async function sendCommand(
json?: boolean;
dryRun?: boolean;
media?: string;
spawnGateway?: boolean;
},
deps: CliDeps,
runtime: RuntimeEnv,
@@ -74,21 +71,7 @@ export async function sendCommand(
mode: "cli",
});
let result: { messageId: string } | undefined;
try {
result = await sendViaGateway();
} catch (err) {
if (!opts.spawnGateway) throw err;
// Only spawn when nothing is listening.
try {
const listeners = listPortListeners(18789);
if (listeners.length > 0) throw err;
await startGatewayServer(18789);
result = await sendViaGateway();
} catch {
throw err;
}
}
const result = await sendViaGateway();
runtime.log(
success(

8
src/commands/status.ts
View File

@@ -10,13 +10,14 @@ import { info } from "../globals.js";
import { buildProviderSummary } from "../infra/provider-summary.js";
import { peekSystemEvents } from "../infra/system-events.js";
import type { RuntimeEnv } from "../runtime.js";
import { callGateway } from "../gateway/call.js";
import { resolveHeartbeatSeconds } from "../web/reconnect.js";
import {
getWebAuthAgeMs,
logWebSelfId,
webAuthExists,
} from "../web/session.js";
import { getHealthSnapshot, type HealthSummary } from "./health.js";
import type { HealthSummary } from "./health.js";
export type SessionStatus = {
key: string;
@@ -193,7 +194,10 @@ export async function statusCommand(
) {
const summary = await getStatusSummary();
const health: HealthSummary | undefined = opts.deep
? await getHealthSnapshot(opts.timeoutMs)
? await callGateway<HealthSummary>({
method: "health",
timeoutMs: opts.timeoutMs,
})
: undefined;
if (opts.json) {

2
src/infra/gateway-lock.ts
View File

@@ -3,6 +3,7 @@ import os from "node:os";
import path from "node:path";
import { flockSync } from "fs-ext";
import { getLogger } from "../logging.js";
const defaultLockPath = () =>
process.env.CLAWDIS_GATEWAY_LOCK_PATH ??
@@ -43,6 +44,7 @@ export async function acquireGatewayLock(
fs.ftruncateSync(fd, 0);
fs.writeSync(fd, `${process.pid}\n`, 0, "utf8");
fs.fsyncSync(fd);
getLogger().info({ pid: process.pid, lockPath }, "gateway lock acquired");
let released = false;
const release = async (): Promise<void> => {

112
src/web/outbound.ts
View File

@@ -1,26 +1,24 @@
import { randomUUID } from "node:crypto";
import type { AnyMessageContent } from "@whiskeysockets/baileys";
import { logVerbose } from "../globals.js";
import { logInfo } from "../logger.js";
import { getChildLogger } from "../logging.js";
import { toWhatsappJid } from "../utils.js";
import { getActiveWebListener } from "./active-listener.js";
import { loadWebMedia } from "./media.js";
import { createWaSocket, waitForWaConnection } from "./session.js";
export async function sendMessageWhatsApp(
to: string,
body: string,
options: { verbose: boolean; mediaUrl?: string },
): Promise<{ messageId: string; toJid: string }> {
let text = body;
const correlationId = randomUUID();
const active = getActiveWebListener();
const usingActive = Boolean(active);
const sock = usingActive
? null
: await createWaSocket(false, options.verbose);
if (!active) {
throw new Error(
"No active gateway listener. Start the gateway before sending WhatsApp messages.",
);
}
const logger = getChildLogger({
module: "web-outbound",
correlationId,
@@ -28,54 +26,25 @@ export async function sendMessageWhatsApp(
});
try {
const jid = toWhatsappJid(to);
if (!usingActive) {
logInfo("🔌 Connecting to WhatsApp Web…");
logger.info("connecting to whatsapp web");
if (!sock) {
throw new Error("WhatsApp socket unavailable");
}
await waitForWaConnection(sock);
try {
await sock.sendPresenceUpdate("composing", jid);
} catch (err) {
logVerbose(`Presence update skipped: ${String(err)}`);
}
}
let payload: AnyMessageContent = { text: body };
let mediaBuffer: Buffer | undefined;
let mediaType: string | undefined;
if (options.mediaUrl) {
const media = await loadWebMedia(options.mediaUrl);
const caption = body || undefined;
const caption = text || undefined;
mediaBuffer = media.buffer;
mediaType = media.contentType;
if (media.kind === "audio") {
// WhatsApp expects explicit opus codec for PTT voice notes.
const mimetype =
mediaType =
media.contentType === "audio/ogg"
? "audio/ogg; codecs=opus"
: (media.contentType ?? "application/octet-stream");
payload = { audio: media.buffer, ptt: true, mimetype };
: media.contentType ?? "application/octet-stream";
} else if (media.kind === "video") {
const mimetype = media.contentType ?? "application/octet-stream";
payload = {
video: media.buffer,
caption,
mimetype,
};
text = caption ?? "";
} else if (media.kind === "image") {
const mimetype = media.contentType ?? "application/octet-stream";
payload = {
image: media.buffer,
caption,
mimetype,
};
text = caption ?? "";
} else {
// Fallback to document for anything else (pdf, etc.).
const fileName = media.fileName ?? "file";
const mimetype = media.contentType ?? "application/octet-stream";
payload = {
document: media.buffer,
fileName,
caption,
mimetype,
};
text = caption ?? "";
}
}
logInfo(
@@ -85,39 +54,30 @@ export async function sendMessageWhatsApp(
{ jid, hasMedia: Boolean(options.mediaUrl) },
"sending message",
);
const result = usingActive
? await (async () => {
if (!active) throw new Error("Active web listener missing");
let mediaBuffer: Buffer | undefined;
let mediaType: string | undefined;
if (options.mediaUrl) {
const media = await loadWebMedia(options.mediaUrl);
mediaBuffer = media.buffer;
mediaType = media.contentType;
}
await active.sendComposingTo(to);
return active.sendMessage(to, body, mediaBuffer, mediaType);
})()
: await (async () => {
if (!sock) throw new Error("WhatsApp socket unavailable");
return sock.sendMessage(jid, payload);
})();
const messageId = usingActive
? ((result as { messageId?: string })?.messageId ?? "unknown")
: ((result as { key?: { id?: string } } | undefined)?.key?.id ??
"unknown");
const result = await (async () => {
if (!active) throw new Error("Active web listener missing");
let mediaBuffer: Buffer | undefined;
let mediaType: string | undefined;
if (options.mediaUrl) {
const media = await loadWebMedia(options.mediaUrl);
mediaBuffer = media.buffer;
mediaType = media.contentType;
}
await active.sendComposingTo(to);
return active.sendMessage(to, text, mediaBuffer, mediaType);
})();
const messageId =
(result as { messageId?: string })?.messageId ?? "unknown";
logInfo(
`✅ Sent via web session. Message ID: ${messageId} -> ${jid}${options.mediaUrl ? " (media)" : ""}`,
);
logger.info({ jid, messageId }, "sent message");
return { messageId, toJid: jid };
} finally {
if (!usingActive) {
try {
sock?.ws?.close();
} catch (err) {
logVerbose(`Socket close failed: ${String(err)}`);
}
}
} catch (err) {
logger.error(
{ err: String(err), to, hasMedia: Boolean(options.mediaUrl) },
"failed to send via web session",
);
throw err;
}
}