diff --git a/AGENTS.md b/AGENTS.md index fb19bb175119..5ae9e51752b6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -74,6 +74,7 @@ Skills own workflows; root owns hard policy and routing. - Core runtime consumes only current canonical shapes/config/data. Legacy or retired shapes normalize only in doctor/migration code before runtime; no runtime shims, aliases, or fallback readers. - State/storage migrations are database-first. Runtime reads/writes the canonical store only. Old file stores, sidecars, aliases, and fallback readers belong in `openclaw doctor --fix` migration code only, never steady-state runtime. - Storage default: SQLite only. Do not add JSON/JSONL/TXT/sidecar files for OpenClaw-owned runtime state, caches, queues, registries, indexes, cursors, checkpoints, or plugin scratch data. +- SQLite runtime access uses Kysely helpers, not raw SQL statement strings, except schema DDL, migrations, low-level DB bootstrap, or narrowly justified SQLite primitives. - Use the shared state DB (`state/openclaw.sqlite`) for global runtime state and plugin KV data. Use the per-agent DB (`agents//agent/openclaw-agent.sqlite`) for agent-scoped state/cache. Use a dedicated SQLite DB only when schema, volume, or lifecycle clearly does not fit those stores. - Legacy state/cache files are migration debt. When touching code that reads/writes them, prefer moving the data into SQLite or calling out the refactor follow-up; do not add parallel file paths. - File storage must be a named product artifact: import/export, user attachment, log, backup, or external tool contract. If it is app state or cache, it belongs in SQLite. diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index 775048cbc65e..5ab5c1a8071d 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -1,2 +1,2 @@ -eadfc9b897a05664735f8e2abcb70cb3f33c19427c20802bf8b035520b7a2ea1 plugin-sdk-api-baseline.json -8e10e093068d73b9ac50d3f265bf7d892652b0392c677be4e332248499cf7ed0 plugin-sdk-api-baseline.jsonl +19bdf1196ec771a00777a16fd1e9c3662b8fd788a81034e705c41a74ee79c7ec plugin-sdk-api-baseline.json +43feff80c90adad0f821d1f1e184a9bff1e93d81e6d53a26a26fd9e2972be759 plugin-sdk-api-baseline.jsonl diff --git a/extensions/telegram/src/polling-session.test.ts b/extensions/telegram/src/polling-session.test.ts index e5edd4dbd571..9b0ed3ed38e9 100644 --- a/extensions/telegram/src/polling-session.test.ts +++ b/extensions/telegram/src/polling-session.test.ts @@ -3,7 +3,16 @@ import os from "node:os"; import path from "node:path"; import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contract"; import { MAX_TIMER_TIMEOUT_MS } from "openclaw/plugin-sdk/number-runtime"; -import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { createChannelIngressQueue } from "../../../src/channels/message/ingress-queue.js"; +import { executeSqliteQuerySync, getNodeSqliteKysely } from "../../../src/infra/kysely-sync.js"; +import type { DB as OpenClawStateKyselyDatabase } from "../../../src/state/openclaw-state-db.generated.js"; +import { + closeOpenClawStateDatabaseForTest, + openOpenClawStateDatabase, +} from "../../../src/state/openclaw-state-db.js"; +import { clearTelegramRuntime, setTelegramRuntime } from "./runtime.js"; +import type { TelegramRuntime } from "./runtime.types.js"; import type { TelegramIngressWorkerMessage } from "./telegram-ingress-worker.js"; const runMock = vi.hoisted(() => vi.fn()); @@ -96,9 +105,21 @@ type WorkerPollErrorListener = (message: { type WorkerMessageListener = (message: TelegramIngressWorkerMessage) => void; type AsyncVoidFn = () => Promise; type MockCallSource = { mock: { calls: Array> } }; +type TelegramPollingTestDatabase = Pick; const POLLING_TEST_WATCHDOG_INTERVAL_MS = 30_000; +function installTelegramIngressQueueRuntime(resolveStateDir: () => string): void { + setTelegramRuntime({ + state: { + resolveStateDir, + openChannelIngressQueue: ( + options?: Omit[0], "channelId">, + ) => createChannelIngressQueue({ ...(options ?? {}), channelId: "telegram" }), + }, + } as TelegramRuntime); +} + function mockObjectArg( source: MockCallSource, label: string, @@ -411,17 +432,59 @@ async function pendingUpdateIds(spoolDir: string, limit: number | "all" = 100): return (await listTelegramSpooledUpdates({ spoolDir, limit })).map((update) => update.updateId); } -async function failedUpdateIds(spoolDir: string): Promise { - const entries = await fs.readdir(spoolDir).catch((err) => { - if ((err as { code?: string }).code === "ENOENT") { - return []; - } - throw err; +function normalizeTelegramTestAccountId(spoolDir: string): string { + const trimmed = path.basename(spoolDir).trim(); + return trimmed ? trimmed.replace(/[^a-z0-9._-]+/gi, "_") : "default"; +} + +function telegramTestQueueName(spoolDir: string): string { + return JSON.stringify(["telegram", normalizeTelegramTestAccountId(spoolDir)]); +} + +function openTelegramSpoolTestKysely(spoolDir: string) { + const database = openOpenClawStateDatabase({ + env: { ...process.env, OPENCLAW_STATE_DIR: spoolDir }, }); - return entries - .filter((entry) => entry.endsWith(".json.failed")) - .map((entry) => Number(entry.slice(0, 16))) - .toSorted((a, b) => a - b); + return { + database, + kysely: getNodeSqliteKysely(database.db), + }; +} + +async function failedUpdateIds(spoolDir: string): Promise { + const { database, kysely } = openTelegramSpoolTestKysely(spoolDir); + const rows = executeSqliteQuerySync( + database.db, + kysely + .selectFrom("channel_ingress_events") + .select("event_id") + .where("queue_name", "=", telegramTestQueueName(spoolDir)) + .where("status", "=", "failed") + .orderBy("event_id", "asc"), + ).rows; + return rows.map((row) => Number(row.event_id)); +} + +async function adoptClaimOwner(params: { + spoolDir: string; + updateId: number; + ownerId: string; + claimedAt: number; +}): Promise { + const { database, kysely } = openTelegramSpoolTestKysely(params.spoolDir); + executeSqliteQuerySync( + database.db, + kysely + .updateTable("channel_ingress_events") + .set({ + claim_owner: params.ownerId, + claimed_at: params.claimedAt, + updated_at: params.claimedAt, + }) + .where("queue_name", "=", telegramTestQueueName(params.spoolDir)) + .where("event_id", "=", String(params.updateId).padStart(16, "0")) + .where("status", "=", "claimed"), + ); } async function withTempSpool(fn: (spoolDir: string) => Promise): Promise { @@ -429,6 +492,7 @@ async function withTempSpool(fn: (spoolDir: string) => Promise): Promise { sleepWithAbortMock.mockReset().mockResolvedValue(undefined); drainPendingDeliveriesMock.mockReset().mockResolvedValue(undefined); resetTelegramReplyFenceForTests(); + installTelegramIngressQueueRuntime(() => + path.join(os.tmpdir(), "openclaw-telegram-test-state"), + ); + }); + + afterEach(() => { + clearTelegramRuntime(); + closeOpenClawStateDatabaseForTest(); }); it("uses backoff helpers for recoverable polling retries", async () => { @@ -667,7 +739,14 @@ describe("TelegramPollingSession", () => { const runPromise = session.runUntilAbort(); await vi.waitFor(() => expect(handleUpdate).toHaveBeenCalledTimes(1)); - await vi.waitFor(async () => expect(await fs.readdir(tempDir)).toEqual([])); + await vi.waitFor(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([])); + await vi.waitFor(async () => + expect( + await listTelegramSpooledUpdateClaims({ + spoolDir: tempDir, + }), + ).toEqual([]), + ); abort.abort(); await runPromise; @@ -686,6 +765,76 @@ describe("TelegramPollingSession", () => { expect(init).toHaveBeenCalledBefore(handleUpdate); expect(handleUpdate).toHaveBeenCalledWith({ update_id: 42, message: { text: "hello" } }); } finally { + abort.abort(); + await fs.rm(tempDir, { recursive: true, force: true }); + } + }); + + it("writes isolated worker updates through the main runtime queue", async () => { + const abort = new AbortController(); + const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-")); + const handleUpdate = vi.fn(async () => undefined); + const bot = { + api: { + deleteWebhook: vi.fn(async () => true), + config: { use: vi.fn() }, + }, + init: vi.fn(async () => undefined), + handleUpdate, + stop: vi.fn(async () => undefined), + }; + createTelegramBotMock.mockReturnValueOnce(bot); + let onMessage: WorkerMessageListener | undefined; + let stopWorker: (() => void) | undefined; + const workerDone = new Promise((resolve) => { + stopWorker = resolve; + }); + const ackSpooledUpdate = vi.fn(); + const createWorker = vi.fn(() => ({ + onMessage: vi.fn((listener: WorkerMessageListener) => { + onMessage = listener; + return () => undefined; + }), + ackSpooledUpdate, + stop: vi.fn(async () => { + stopWorker?.(); + }), + task: vi.fn(async () => { + await workerDone; + }), + })); + + try { + const session = createPollingSession({ + abortSignal: abort.signal, + isolatedIngress: { + enabled: true, + spoolDir: tempDir, + createWorker, + drainIntervalMs: 10, + }, + }); + + const runPromise = session.runUntilAbort(); + await vi.waitFor(() => expect(onMessage).toBeDefined()); + onMessage?.({ + type: "update", + requestId: "write-1", + update: { update_id: 42, message: { text: "hello" } }, + queued: 1, + }); + + await vi.waitFor(() => + expect(ackSpooledUpdate).toHaveBeenCalledWith("write-1", { ok: true, updateId: 42 }), + ); + await vi.waitFor(() => + expect(handleUpdate).toHaveBeenCalledWith({ update_id: 42, message: { text: "hello" } }), + ); + await vi.waitFor(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([])); + abort.abort(); + await runPromise; + } finally { + abort.abort(); await fs.rm(tempDir, { recursive: true, force: true }); } }); @@ -735,7 +884,14 @@ describe("TelegramPollingSession", () => { const runPromise = session.runUntilAbort(); await vi.waitFor(() => expect(handleUpdate).toHaveBeenCalledTimes(1)); - await vi.waitFor(async () => expect(await fs.readdir(tempDir)).toEqual([])); + await vi.waitFor(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([])); + await vi.waitFor(async () => + expect( + await listTelegramSpooledUpdateClaims({ + spoolDir: tempDir, + }), + ).toEqual([]), + ); abort.abort(); await runPromise; @@ -1128,7 +1284,7 @@ describe("TelegramPollingSession", () => { await runPromise; expect(events).toEqual(["handled:42", "handled:44"]); expect(await pendingUpdateIds(tempDir)).toEqual([43]); - expect((await fs.readdir(tempDir)).toSorted()).toEqual(["0000000000000043.json"]); + expect(await listTelegramSpooledUpdateClaims({ spoolDir: tempDir })).toEqual([]); stopWorker(); }); }); @@ -1189,21 +1345,12 @@ describe("TelegramPollingSession", () => { if (!claimed) { throw new Error("Expected claimed update"); } - await fs.writeFile( - claimed.path, - `${JSON.stringify({ - version: 1, - updateId: 42, - receivedAt: interrupted.receivedAt, - update: interruptedUpdate, - claim: { - processId: "other-process", - processPid: process.pid, - claimedAt: Date.now(), - }, - })}\n`, - { mode: 0o600 }, - ); + await adoptClaimOwner({ + spoolDir: tempDir, + updateId: 42, + ownerId: `${process.pid}:other-process`, + claimedAt: Date.now(), + }); const recovered = await recoverStaleTelegramSpooledUpdateClaims({ spoolDir: tempDir, @@ -1213,10 +1360,11 @@ describe("TelegramPollingSession", () => { expect(recovered).toBe(0); expect(await pendingUpdateIds(tempDir)).toEqual([43]); - expect((await fs.readdir(tempDir)).toSorted()).toEqual([ - "0000000000000042.json.processing", - "0000000000000043.json", - ]); + expect( + (await listTelegramSpooledUpdateClaims({ spoolDir: tempDir })).map( + (claim) => claim.updateId, + ), + ).toEqual([42]); }); }); @@ -2360,7 +2508,7 @@ describe("TelegramPollingSession", () => { } }); - it("keeps a timed-out lane guarded when its failed tombstone cannot be written", async () => { + it("keeps a timed-out lane guarded when its failed state cannot be written", async () => { vi.useFakeTimers({ shouldAdvanceTime: true }); const abort = new AbortController(); const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-")); @@ -2371,15 +2519,10 @@ describe("TelegramPollingSession", () => { const regularTurnDone = new Promise((resolve) => { releaseRegularTurn = resolve; }); - const originalWriteFile = fs.writeFile.bind(fs); - const writeFileSpy = vi - .spyOn(fs, "writeFile") - .mockImplementation(async (...args: Parameters) => { - if (typeof args[0] === "string" && args[0].includes(".json.failed.")) { - throw new Error("disk full"); - } - return await originalWriteFile(...args); - }); + const spoolModule = await import("./telegram-ingress-spool.js"); + const failSpy = vi + .spyOn(spoolModule, "failTelegramSpooledUpdateClaim") + .mockRejectedValueOnce(new Error("disk full")); createTelegramBotMock.mockReturnValueOnce({ api: { deleteWebhook: vi.fn(async () => true), @@ -2460,7 +2603,7 @@ describe("TelegramPollingSession", () => { await vi.advanceTimersByTimeAsync(20_000); await runPromise; } finally { - writeFileSpy.mockRestore(); + failSpy.mockRestore(); releaseRegularTurn?.(); abort.abort(); stopWorker?.(); diff --git a/extensions/telegram/src/polling-session.ts b/extensions/telegram/src/polling-session.ts index 9c2f071fd60e..68073641f1af 100644 --- a/extensions/telegram/src/polling-session.ts +++ b/extensions/telegram/src/polling-session.ts @@ -36,6 +36,7 @@ import { recoverStaleTelegramSpooledUpdateClaims, releaseTelegramSpooledUpdateClaim, resolveTelegramIngressSpoolDir, + writeTelegramSpooledUpdate, type ClaimedTelegramSpooledUpdate, type TelegramSpooledUpdate, } from "./telegram-ingress-spool.js"; @@ -767,6 +768,23 @@ export class TelegramPollingSession { }); const stalledBacklogKeys = new Set(); const unsubscribe = worker.onMessage((message) => { + const ackSpooledUpdate = ( + requestId: string, + result: + | { ok: true; updateId: number } + | { + ok: false; + message: string; + }, + ): void => { + try { + worker.ackSpooledUpdate?.(requestId, result); + } catch (err) { + this.opts.log( + `[telegram][diag] isolated polling worker ack failed: ${formatErrorMessage(err)}`, + ); + } + }; if (message.type === "poll-start") { liveness.noteGetUpdatesStarted({ offset: message.offset }, message.startedAt); pollState.startedAt = message.startedAt; @@ -792,6 +810,23 @@ export class TelegramPollingSession { pollState.error = message.message; return; } + if (message.type === "update") { + void writeTelegramSpooledUpdate({ + spoolDir, + update: message.update, + }).then( + (updateId) => { + ackSpooledUpdate(message.requestId, { ok: true, updateId }); + }, + (err: unknown) => { + ackSpooledUpdate(message.requestId, { + ok: false, + message: formatErrorMessage(err), + }); + }, + ); + return; + } if (message.type === "spooled") { liveness.noteGetUpdatesActivity(); } diff --git a/extensions/telegram/src/telegram-ingress-spool.test.ts b/extensions/telegram/src/telegram-ingress-spool.test.ts index b61857e4c434..a8f8db16048b 100644 --- a/extensions/telegram/src/telegram-ingress-spool.test.ts +++ b/extensions/telegram/src/telegram-ingress-spool.test.ts @@ -1,7 +1,11 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; -import { describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it } from "vitest"; +import { createChannelIngressQueue } from "../../../src/channels/message/ingress-queue.js"; +import { closeOpenClawStateDatabaseForTest } from "../../../src/state/openclaw-state-db.js"; +import { clearTelegramRuntime, setTelegramRuntime } from "./runtime.js"; +import type { TelegramRuntime } from "./runtime.types.js"; import { claimTelegramSpooledUpdate, deleteTelegramSpooledUpdate, @@ -15,16 +19,37 @@ import { writeTelegramSpooledUpdate, } from "./telegram-ingress-spool.js"; +function installTelegramIngressQueueRuntime(resolveStateDir: () => string): void { + setTelegramRuntime({ + state: { + resolveStateDir, + openChannelIngressQueue: ( + options?: Omit[0], "channelId">, + ) => createChannelIngressQueue({ ...(options ?? {}), channelId: "telegram" }), + }, + } as TelegramRuntime); +} + async function withTempSpool(fn: (spoolDir: string) => Promise): Promise { - const spoolDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-")); + const stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-")); + const spoolDir = path.join(stateDir, "telegram", "ingress-spool-test"); + await fs.mkdir(spoolDir, { recursive: true }); + installTelegramIngressQueueRuntime(() => stateDir); try { return await fn(spoolDir); } finally { - await fs.rm(spoolDir, { recursive: true, force: true }); + clearTelegramRuntime(); + closeOpenClawStateDatabaseForTest(); + await fs.rm(stateDir, { recursive: true, force: true }); } } describe("Telegram ingress spool", () => { + afterEach(() => { + clearTelegramRuntime(); + closeOpenClawStateDatabaseForTest(); + }); + it("persists updates durably in update_id order and deletes handled entries", async () => { await withTempSpool(async (spoolDir) => { await writeTelegramSpooledUpdate({ @@ -84,9 +109,8 @@ describe("Telegram ingress spool", () => { if (!claimed) { throw new Error("Expected a claimed update"); } - await fs.writeFile(claimed.pendingPath, "duplicate pending race", { mode: 0o600 }); await deleteTelegramSpooledUpdate(claimed); - expect(await fs.readdir(spoolDir)).toEqual([]); + expect(await listTelegramSpooledUpdateClaims({ spoolDir })).toEqual([]); }); }); @@ -139,48 +163,17 @@ describe("Telegram ingress spool", () => { expect(await listTelegramSpooledUpdates({ spoolDir })).toEqual([]); expect(await listTelegramSpooledUpdateClaims({ spoolDir })).toEqual([]); - const entries = await fs.readdir(spoolDir); - expect(entries).toEqual(["0000000000000032.json.failed"]); - const failed = JSON.parse( - await fs.readFile(path.join(spoolDir, "0000000000000032.json.failed"), "utf8"), - ) as { - update?: unknown; - claim?: unknown; - failure?: { reason?: string; message?: string; failedAt?: number }; - }; - expect(failed.update).toBeUndefined(); - expect(failed.claim).toBeUndefined(); - expect(failed.failure).toEqual({ - reason: "handler-timeout", - message: "timed out", - failedAt: 123, - }); await writeTelegramSpooledUpdate({ spoolDir, update: { update_id: 32, message: { text: "redelivered poison" } }, + now: 124, }); expect(await listTelegramSpooledUpdates({ spoolDir })).toEqual([]); - expect(await fs.readdir(spoolDir)).toEqual(["0000000000000032.json.failed"]); - - const leakedProcessingPath = path.join(spoolDir, "0000000000000032.json.processing"); - await fs.writeFile( - leakedProcessingPath, - `${JSON.stringify({ - version: 1, - updateId: 32, - receivedAt: 100, - update: { update_id: 32, message: { text: "crashed poison claim" } }, - })}\n`, - { mode: 0o600 }, - ); - const staleTime = new Date(Date.now() - TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS - 1); - await fs.utimes(leakedProcessingPath, staleTime, staleTime); await expect(recoverStaleTelegramSpooledUpdateClaims({ spoolDir })).resolves.toBe(0); expect(await listTelegramSpooledUpdates({ spoolDir })).toEqual([]); expect(await listTelegramSpooledUpdateClaims({ spoolDir })).toEqual([]); - expect(await fs.readdir(spoolDir)).toEqual(["0000000000000032.json.failed"]); }); }); @@ -197,56 +190,44 @@ describe("Telegram ingress spool", () => { await deleteTelegramSpooledUpdate(update); await expect(claimTelegramSpooledUpdate(update)).resolves.toBeNull(); - expect(await fs.readdir(spoolDir)).toEqual([]); + expect(await listTelegramSpooledUpdates({ spoolDir })).toEqual([]); }); }); - it("recovers stale processing claims without replaying fresh claims", async () => { + it("recovers stale processing claims selected by the caller", async () => { await withTempSpool(async (spoolDir) => { - await writeTelegramSpooledUpdate({ - spoolDir, - update: { update_id: 40, message: { text: "fresh" } }, - }); await writeTelegramSpooledUpdate({ spoolDir, update: { update_id: 41, message: { text: "stale" } }, }); const updates = await listTelegramSpooledUpdates({ spoolDir }); - const fresh = updates.find((update) => update.updateId === 40); const stale = updates.find((update) => update.updateId === 41); - if (!fresh || !stale) { + if (!stale) { throw new Error("Expected spooled updates"); } - const claimedFresh = await claimTelegramSpooledUpdate(fresh); const claimedStale = await claimTelegramSpooledUpdate(stale); - if (!claimedFresh || !claimedStale) { + if (!claimedStale) { throw new Error("Expected claimed updates"); } const now = Date.now(); - const oldClaimTime = new Date(now - TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS - 1); - await fs.utimes(claimedStale.path, oldClaimTime, oldClaimTime); const recovered = await recoverStaleTelegramSpooledUpdateClaims({ spoolDir, - now, + now: now + TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS + 1, }); expect(recovered).toBe(1); expect( (await listTelegramSpooledUpdates({ spoolDir })).map((update) => update.updateId), ).toEqual([41]); - expect((await fs.readdir(spoolDir)).toSorted()).toEqual([ - "0000000000000040.json.processing", - "0000000000000041.json", - ]); }); }); - it("handles ENOENT race when processing file is removed before recovery rename", async () => { + it("lets recovery callers keep a claim in processing", async () => { await withTempSpool(async (spoolDir) => { await writeTelegramSpooledUpdate({ spoolDir, - update: { update_id: 45, message: { text: "vanishes" } }, + update: { update_id: 45, message: { text: "busy" } }, }); const update = (await listTelegramSpooledUpdates({ spoolDir }))[0]; if (!update) { @@ -260,16 +241,17 @@ describe("Telegram ingress spool", () => { const recovered = await recoverStaleTelegramSpooledUpdateClaims({ spoolDir, staleMs: 0, - shouldRecover: async () => { + shouldRecover: () => { shouldRecoverCalls += 1; - await fs.unlink(claimed.path); - return true; + return false; }, }); expect(recovered).toBe(0); expect(shouldRecoverCalls).toBe(1); - expect(await fs.readdir(spoolDir)).toEqual([]); + expect( + (await listTelegramSpooledUpdateClaims({ spoolDir })).map((claim) => claim.updateId), + ).toEqual([45]); }); }); diff --git a/extensions/telegram/src/telegram-ingress-spool.ts b/extensions/telegram/src/telegram-ingress-spool.ts index c006a7b83ec8..0002881b4b4f 100644 --- a/extensions/telegram/src/telegram-ingress-spool.ts +++ b/extensions/telegram/src/telegram-ingress-spool.ts @@ -1,18 +1,27 @@ import { randomUUID } from "node:crypto"; -import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; -import { readJsonFileWithFallback } from "openclaw/plugin-sdk/json-store"; +import { + type ChannelIngressQueue, + type ChannelIngressQueueClaim, + type ChannelIngressQueueClaimRef, + type ChannelIngressQueueRecord, +} from "openclaw/plugin-sdk/channel-outbound"; import { resolveStateDir } from "openclaw/plugin-sdk/state-paths"; +import { getTelegramRuntime } from "./runtime.js"; const SPOOL_VERSION = 1; +const TELEGRAM_INGRESS_SPOOL_PREFIX = "ingress-spool-"; export const TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS = 6 * 60 * 60 * 1000; +const TELEGRAM_SPOOLED_UPDATE_FAILED_TTL_MS = 30 * 24 * 60 * 60 * 1000; +const TELEGRAM_SPOOLED_UPDATE_FAILED_MAX_ENTRIES = 1000; const TELEGRAM_SPOOLED_UPDATE_PROCESS_ID = `${process.pid}:${randomUUID()}`; type TelegramSpooledUpdateClaimOwner = { processId: string; processPid: number; claimedAt: number; + claimToken?: string; }; type TelegramSpooledUpdatePayload = { @@ -20,16 +29,6 @@ type TelegramSpooledUpdatePayload = { updateId: number; receivedAt: number; update: unknown; - claim?: TelegramSpooledUpdateClaimOwner; - failure?: { - reason: string; - message: string; - failedAt: number; - }; -}; - -type TelegramFailedSpooledUpdatePayload = Omit & { - failure: NonNullable; }; export type TelegramSpooledUpdate = { @@ -61,7 +60,11 @@ export function resolveTelegramIngressSpoolDir(params: { env?: NodeJS.ProcessEnv; }): string { const stateDir = resolveStateDir(params.env, os.homedir); - return path.join(stateDir, "telegram", `ingress-spool-${normalizeAccountId(params.accountId)}`); + return path.join( + stateDir, + "telegram", + `${TELEGRAM_INGRESS_SPOOL_PREFIX}${normalizeAccountId(params.accountId)}`, + ); } export function resolveTelegramUpdateId(update: unknown): number | null { @@ -80,94 +83,75 @@ function processingFileName(updateId: number): string { return `${spoolFileName(updateId)}.processing`; } -function failedFileName(updateId: number): string { - return `${spoolFileName(updateId)}.failed`; +function queueEventId(updateId: number): string { + return String(updateId).padStart(16, "0"); } -function isProcessingFileName(fileName: string): boolean { - return fileName.endsWith(".json.processing"); -} - -function pendingFileNameFromProcessing(fileName: string): string { - return fileName.slice(0, -".processing".length); +function pendingPath(spoolDir: string, updateId: number): string { + return path.join(spoolDir, spoolFileName(updateId)); } function processingPath(spoolDir: string, updateId: number): string { return path.join(spoolDir, processingFileName(updateId)); } -function failedPath(spoolDir: string, updateId: number): string { - return path.join(spoolDir, failedFileName(updateId)); -} - -async function pathExists(filePath: string): Promise { - try { - await fs.access(filePath); - return true; - } catch (err) { - if ((err as { code?: string }).code === "ENOENT") { - return false; - } - throw err; - } -} - -async function unlinkIfPresent(filePath: string): Promise { - try { - await fs.unlink(filePath); - } catch (err) { - if ((err as { code?: string }).code === "ENOENT") { - return; - } - throw err; - } -} - -function parseSpooledUpdate(value: unknown, filePath: string): TelegramSpooledUpdate | null { - if (!value || typeof value !== "object") { - return null; - } - const payload = value as Partial; - if (payload.version !== SPOOL_VERSION || !isValidUpdateId(payload.updateId)) { - return null; - } - const update: TelegramSpooledUpdate = { - updateId: payload.updateId, - path: filePath, - update: payload.update, - receivedAt: typeof payload.receivedAt === "number" ? payload.receivedAt : 0, - }; - if ( - payload.claim && - typeof payload.claim.processId === "string" && - isValidUpdateId(payload.claim.processPid) && - typeof payload.claim.claimedAt === "number" - ) { - update.claim = payload.claim; - } - return update; -} - -function buildClaimedPayload(update: TelegramSpooledUpdate): TelegramSpooledUpdatePayload { +function resolveQueueParts(spoolDir: string): { + accountId: string; + stateDir: string; +} { + const basename = path.basename(spoolDir); + const accountId = normalizeAccountId( + basename.startsWith(TELEGRAM_INGRESS_SPOOL_PREFIX) + ? basename.slice(TELEGRAM_INGRESS_SPOOL_PREFIX.length) + : basename, + ); + const stateDir = + basename.startsWith(TELEGRAM_INGRESS_SPOOL_PREFIX) && + path.basename(path.dirname(spoolDir)) === "telegram" + ? path.dirname(path.dirname(spoolDir)) + : spoolDir; return { - version: SPOOL_VERSION, - updateId: update.updateId, - receivedAt: update.receivedAt, - update: update.update, - claim: { - processId: TELEGRAM_SPOOLED_UPDATE_PROCESS_ID, - processPid: process.pid, - claimedAt: Date.now(), - }, + accountId, + stateDir, }; } +function createTelegramIngressQueue( + spoolDir: string, +): ChannelIngressQueue { + const parts = resolveQueueParts(spoolDir); + return getTelegramRuntime().state.openChannelIngressQueue({ + accountId: parts.accountId, + stateDir: parts.stateDir, + }); +} + +async function pruneTelegramIngressQueue( + queue: ChannelIngressQueue, + now?: number, +): Promise { + await queue.prune({ + failedTtlMs: TELEGRAM_SPOOLED_UPDATE_FAILED_TTL_MS, + failedMaxEntries: TELEGRAM_SPOOLED_UPDATE_FAILED_MAX_ENTRIES, + ...(now === undefined ? {} : { now }), + }); +} + +function processPidFromOwnerId(ownerId: string): number { + const pid = Number.parseInt(ownerId.split(":", 1)[0] ?? "", 10); + return Number.isSafeInteger(pid) && pid > 0 ? pid : -1; +} + function processExists(pid: number): boolean { + if (!Number.isSafeInteger(pid) || pid <= 0) { + return false; + } try { process.kill(pid, 0); return true; } catch (err) { - return (err as { code?: string }).code !== "ESRCH"; + const code = (err as { code?: string }).code; + return code !== "ESRCH" && code !== "EINVAL"; } } @@ -175,6 +159,52 @@ function isFreshClaimOwner(claim: TelegramSpooledUpdateClaimOwner): boolean { return Date.now() - claim.claimedAt < TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS; } +function parseQueueRecord( + spoolDir: string, + record: ChannelIngressQueueRecord, +): TelegramSpooledUpdate | null { + const payload = record.payload; + if (payload.version !== SPOOL_VERSION || !isValidUpdateId(payload.updateId)) { + return null; + } + return { + updateId: payload.updateId, + path: pendingPath(spoolDir, payload.updateId), + update: payload.update, + receivedAt: payload.receivedAt, + }; +} + +function parseQueueClaim( + spoolDir: string, + record: ChannelIngressQueueClaim, +): ClaimedTelegramSpooledUpdate | null { + const update = parseQueueRecord(spoolDir, record); + if (!update) { + return null; + } + return { + ...update, + path: processingPath(spoolDir, update.updateId), + pendingPath: pendingPath(spoolDir, update.updateId), + claim: { + processId: record.claim.ownerId, + processPid: processPidFromOwnerId(record.claim.ownerId), + claimedAt: record.claim.claimedAt, + claimToken: record.claim.token, + }, + }; +} + +function sortTelegramUpdates(updates: T[]): T[] { + return updates.toSorted((a, b) => a.updateId - b.updateId); +} + +function queueMutationTarget(update: TelegramSpooledUpdate): string | ChannelIngressQueueClaimRef { + const id = queueEventId(update.updateId); + return update.claim?.claimToken ? { id, claim: { token: update.claim.claimToken } } : id; +} + export function isTelegramSpooledUpdateClaimOwnedByOtherLiveProcess( claim: ClaimedTelegramSpooledUpdate, ): boolean { @@ -195,26 +225,19 @@ export async function writeTelegramSpooledUpdate(params: { if (updateId === null) { throw new Error("Telegram update missing numeric update_id."); } - await fs.mkdir(params.spoolDir, { recursive: true }); - const targetPath = path.join(params.spoolDir, spoolFileName(updateId)); - const claimedPath = processingPath(params.spoolDir, updateId); - const tombstonePath = failedPath(params.spoolDir, updateId); - if ((await pathExists(claimedPath)) || (await pathExists(tombstonePath))) { - return updateId; - } - const tempPath = path.join(params.spoolDir, `${spoolFileName(updateId)}.${randomUUID()}.tmp`); - const payload: TelegramSpooledUpdatePayload = { - version: SPOOL_VERSION, - updateId, - receivedAt: params.now ?? Date.now(), - update: params.update, - }; - await fs.writeFile(tempPath, `${JSON.stringify(payload)}\n`, { mode: 0o600 }); - if ((await pathExists(claimedPath)) || (await pathExists(tombstonePath))) { - await unlinkIfPresent(tempPath); - return updateId; - } - await fs.rename(tempPath, targetPath); + const receivedAt = params.now ?? Date.now(); + const queue = createTelegramIngressQueue(params.spoolDir); + await pruneTelegramIngressQueue(queue, params.now); + await queue.enqueue( + queueEventId(updateId), + { + version: SPOOL_VERSION, + updateId, + receivedAt, + update: params.update, + }, + { receivedAt }, + ); return updateId; } @@ -222,95 +245,38 @@ export async function listTelegramSpooledUpdates(params: { spoolDir: string; limit?: number | "all"; }): Promise { - let entries: string[]; - try { - entries = await fs.readdir(params.spoolDir); - } catch (err) { - if ((err as { code?: string }).code === "ENOENT") { - return []; - } - throw err; - } - const entrySet = new Set(entries); - const files = entries - .filter((entry) => entry.endsWith(".json") && !entrySet.has(`${entry}.failed`)) - .toSorted(); - const limitedFiles = - params.limit === "all" ? files : files.slice(0, Math.max(1, params.limit ?? 100)); - const updates: TelegramSpooledUpdate[] = []; - for (const file of limitedFiles) { - const filePath = path.join(params.spoolDir, file); - const { value } = await readJsonFileWithFallback(filePath, null); - const parsed = parseSpooledUpdate(value, filePath); - if (parsed) { - updates.push(parsed); - } - } - return updates; + const records = await createTelegramIngressQueue(params.spoolDir).listPending({ + limit: params.limit ?? 100, + orderBy: "id", + }); + return sortTelegramUpdates( + records.flatMap((record) => { + const update = parseQueueRecord(params.spoolDir, record); + return update ? [update] : []; + }), + ); } export async function deleteTelegramSpooledUpdate(update: TelegramSpooledUpdate): Promise { - await unlinkIfPresent(update.path); - if ("pendingPath" in update && typeof update.pendingPath === "string") { - await unlinkIfPresent(update.pendingPath); - } + await createTelegramIngressQueue(path.dirname(update.path)).delete(queueMutationTarget(update)); } export async function claimTelegramSpooledUpdate( update: TelegramSpooledUpdate, ): Promise { - const claimedPath = processingPath(path.dirname(update.path), update.updateId); - const holdPath = path.join( - path.dirname(update.path), - `${spoolFileName(update.updateId)}.${randomUUID()}.claim`, - ); - const tempPath = path.join( - path.dirname(update.path), - `${processingFileName(update.updateId)}.${randomUUID()}.tmp`, - ); - try { - const claimedAt = new Date(); - await fs.writeFile(tempPath, `${JSON.stringify(buildClaimedPayload(update))}\n`, { - mode: 0o600, - }); - await fs.link(update.path, holdPath); - await fs.link(tempPath, claimedPath); - await unlinkIfPresent(tempPath); - await unlinkIfPresent(holdPath); - await fs.utimes(claimedPath, claimedAt, claimedAt); - await unlinkIfPresent(update.path); - } catch (err) { - const code = (err as { code?: string }).code; - await unlinkIfPresent(tempPath); - await unlinkIfPresent(holdPath); - if (code === "ENOENT" || code === "EEXIST") { - return null; - } - throw err; - } - return { - ...update, - path: claimedPath, - pendingPath: update.path, - }; + const spoolDir = path.dirname(update.path); + const claimed = await createTelegramIngressQueue(spoolDir).claim(queueEventId(update.updateId), { + ownerId: TELEGRAM_SPOOLED_UPDATE_PROCESS_ID, + }); + return claimed ? parseQueueClaim(spoolDir, claimed) : null; } export async function releaseTelegramSpooledUpdateClaim( update: ClaimedTelegramSpooledUpdate, ): Promise { - try { - await fs.rename(update.path, update.pendingPath); - } catch (err) { - const code = (err as { code?: string }).code; - if (code === "ENOENT") { - return; - } - if (code === "EEXIST") { - await unlinkIfPresent(update.path); - return; - } - throw err; - } + await createTelegramIngressQueue(path.dirname(update.pendingPath)).release( + queueMutationTarget(update), + ); } export async function failTelegramSpooledUpdateClaim(params: { @@ -319,70 +285,26 @@ export async function failTelegramSpooledUpdateClaim(params: { message: string; now?: number; }): Promise { - const tombstonePath = failedPath(path.dirname(params.update.path), params.update.updateId); - const tempPath = path.join( - path.dirname(params.update.path), - `${failedFileName(params.update.updateId)}.${randomUUID()}.tmp`, - ); - try { - const { value } = await readJsonFileWithFallback(params.update.path, null); - const parsed = parseSpooledUpdate(value, params.update.path); - if (!parsed) { - return false; - } - const payload: TelegramFailedSpooledUpdatePayload = { - version: SPOOL_VERSION, - updateId: parsed.updateId, - receivedAt: parsed.receivedAt, - failure: { - reason: params.reason, - message: params.message, - failedAt: params.now ?? Date.now(), - }, - }; - await fs.writeFile(tempPath, `${JSON.stringify(payload)}\n`, { mode: 0o600 }); - await fs.rename(tempPath, tombstonePath); - await unlinkIfPresent(params.update.path); - await unlinkIfPresent(params.update.pendingPath); - return true; - } catch (err) { - await unlinkIfPresent(tempPath); - if ((err as { code?: string }).code === "ENOENT") { - return false; - } - throw err; - } + const queue = createTelegramIngressQueue(path.dirname(params.update.pendingPath)); + const failed = await queue.fail(queueMutationTarget(params.update), { + reason: params.reason, + message: params.message, + ...(params.now === undefined ? {} : { failedAt: params.now }), + }); + await pruneTelegramIngressQueue(queue, params.now); + return failed; } export async function listTelegramSpooledUpdateClaims(params: { spoolDir: string; }): Promise { - let entries: string[]; - try { - entries = await fs.readdir(params.spoolDir); - } catch (err) { - if ((err as { code?: string }).code === "ENOENT") { - return []; - } - throw err; - } - const claims: ClaimedTelegramSpooledUpdate[] = []; - const entrySet = new Set(entries); - for (const file of entries.filter(isProcessingFileName).toSorted()) { - if (entrySet.has(`${pendingFileNameFromProcessing(file)}.failed`)) { - continue; - } - const filePath = path.join(params.spoolDir, file); - const { value } = await readJsonFileWithFallback(filePath, null); - const parsed = parseSpooledUpdate(value, filePath); - if (parsed) { - claims.push({ - ...parsed, - pendingPath: path.join(params.spoolDir, pendingFileNameFromProcessing(file)), - }); - } - } - return claims; + const claims = await createTelegramIngressQueue(params.spoolDir).listClaims(); + return sortTelegramUpdates( + claims.flatMap((claim) => { + const update = parseQueueClaim(params.spoolDir, claim); + return update ? [update] : []; + }), + ); } export async function recoverStaleTelegramSpooledUpdateClaims(params: { @@ -391,73 +313,17 @@ export async function recoverStaleTelegramSpooledUpdateClaims(params: { now?: number; shouldRecover?: (claim: ClaimedTelegramSpooledUpdate) => boolean | Promise; }): Promise { - let entries: string[]; - try { - entries = await fs.readdir(params.spoolDir); - } catch (err) { - if ((err as { code?: string }).code === "ENOENT") { - return 0; - } - throw err; - } - const staleMs = Math.max( - 0, - Math.floor(params.staleMs ?? TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS), - ); - const now = params.now ?? Date.now(); - let recovered = 0; - const entrySet = new Set(entries); - for (const entry of entries.filter(isProcessingFileName).toSorted()) { - const claimedPath = path.join(params.spoolDir, entry); - const pendingPath = path.join(params.spoolDir, pendingFileNameFromProcessing(entry)); - if (entrySet.has(`${pendingFileNameFromProcessing(entry)}.failed`)) { - await unlinkIfPresent(claimedPath); - await unlinkIfPresent(pendingPath); - continue; - } - let stat; - try { - stat = await fs.stat(claimedPath); - } catch (err) { - if ((err as { code?: string }).code === "ENOENT") { - continue; - } - throw err; - } - if (now - stat.mtimeMs < staleMs) { - continue; - } - if (params.shouldRecover) { - const { value } = await readJsonFileWithFallback(claimedPath, null); - const parsed = parseSpooledUpdate(value, claimedPath); - if ( - parsed && - !(await params.shouldRecover({ - ...parsed, - pendingPath, - })) - ) { - continue; - } - } - if (await pathExists(pendingPath)) { - await unlinkIfPresent(claimedPath); - } else { - try { - await fs.rename(claimedPath, pendingPath); - } catch (err) { - const code = (err as { code?: string }).code; - if (code === "ENOENT") { - continue; + const shouldRecover = params.shouldRecover; + return await createTelegramIngressQueue(params.spoolDir).recoverStaleClaims({ + staleMs: params.staleMs ?? TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS, + ...(params.now === undefined ? {} : { now: params.now }), + ...(shouldRecover + ? { + shouldRecover: async (claim: ChannelIngressQueueClaim) => { + const update = parseQueueClaim(params.spoolDir, claim); + return update ? await shouldRecover(update) : false; + }, } - if (code === "EEXIST") { - await unlinkIfPresent(claimedPath); - } else { - throw err; - } - } - } - recovered += 1; - } - return recovered; + : {}), + }); } diff --git a/extensions/telegram/src/telegram-ingress-worker.runtime.ts b/extensions/telegram/src/telegram-ingress-worker.runtime.ts index 9ce534d774f5..3f67366720a8 100644 --- a/extensions/telegram/src/telegram-ingress-worker.runtime.ts +++ b/extensions/telegram/src/telegram-ingress-worker.runtime.ts @@ -8,8 +8,8 @@ import { TELEGRAM_GET_UPDATES_REQUEST_TIMEOUT_MS, resolveTelegramLongPollTimeoutSeconds, } from "./request-timeouts.js"; -import { writeTelegramSpooledUpdate } from "./telegram-ingress-spool.js"; import type { + TelegramIngressWorkerCommand, TelegramIngressWorkerMessage, TelegramIngressWorkerOptions, } from "./telegram-ingress-worker.js"; @@ -20,6 +20,14 @@ const retryInitialMs = 1000; const retryMaxMs = 30_000; let stopped = false; let activeController: AbortController | undefined; +let nextSpoolRequestId = 0; +const pendingSpoolRequests = new Map< + string, + { + resolve(updateId: number): void; + reject(err: Error): void; + } +>(); function post(message: TelegramIngressWorkerMessage): void { if (parentPort) { @@ -44,14 +52,53 @@ function resolveBackoff(attempt: number): number { return Math.min(retryMaxMs, retryInitialMs * 2 ** Math.max(0, attempt - 1)); } -parentPort?.on("message", (message: { type?: string }) => { - if (message?.type !== "stop") { +function rejectPendingSpoolRequests(err: Error): void { + for (const pending of pendingSpoolRequests.values()) { + pending.reject(err); + } + pendingSpoolRequests.clear(); +} + +parentPort?.on("message", (message: TelegramIngressWorkerCommand) => { + if (message?.type === "stop") { + stopped = true; + const err = new Error("telegram ingress worker stopped"); + activeController?.abort(err); + rejectPendingSpoolRequests(err); return; } - stopped = true; - activeController?.abort(new Error("telegram ingress worker stopped")); + if (message?.type !== "spool-ack") { + return; + } + const pending = pendingSpoolRequests.get(message.requestId); + if (!pending) { + return; + } + pendingSpoolRequests.delete(message.requestId); + if (message.result.ok) { + pending.resolve(message.result.updateId); + return; + } + pending.reject(new Error(message.result.message)); }); +async function requestSpoolUpdate(params: { update: unknown; queued: number }): Promise { + if (!parentPort) { + throw new Error("Telegram ingress worker missing parent port."); + } + const requestId = String(++nextSpoolRequestId); + const updateId = await new Promise((resolve, reject) => { + pendingSpoolRequests.set(requestId, { resolve, reject }); + post({ + type: "update", + requestId, + update: params.update, + queued: params.queued, + }); + }); + return updateId; +} + async function fetchJson(params: { fetch: typeof fetch; url: string; @@ -127,10 +174,7 @@ async function main(): Promise { if (stopped) { break; } - const updateId = await writeTelegramSpooledUpdate({ - spoolDir: options.spoolDir, - update, - }); + const updateId = await requestSpoolUpdate({ update, queued: result.length }); if (lastUpdateId === null || updateId > lastUpdateId) { lastUpdateId = updateId; } diff --git a/extensions/telegram/src/telegram-ingress-worker.ts b/extensions/telegram/src/telegram-ingress-worker.ts index a443b49873f0..f37d00c20420 100644 --- a/extensions/telegram/src/telegram-ingress-worker.ts +++ b/extensions/telegram/src/telegram-ingress-worker.ts @@ -22,6 +22,30 @@ export type TelegramIngressWorkerMessage = type: "spooled"; updateId: number; queued: number; + } + | { + type: "update"; + requestId: string; + update: unknown; + queued: number; + }; + +export type TelegramIngressWorkerCommand = + | { + type: "stop"; + } + | { + type: "spool-ack"; + requestId: string; + result: + | { + ok: true; + updateId: number; + } + | { + ok: false; + message: string; + }; }; export type TelegramIngressWorkerOptions = { @@ -37,6 +61,18 @@ export type TelegramIngressWorkerOptions = { export type TelegramIngressWorkerHandle = { onMessage(listener: (message: TelegramIngressWorkerMessage) => void): () => void; + ackSpooledUpdate?( + requestId: string, + result: + | { + ok: true; + updateId: number; + } + | { + ok: false; + message: string; + }, + ): void; stop(): Promise; task(): Promise; }; @@ -73,9 +109,18 @@ export const createTelegramIngressWorker: TelegramIngressWorkerFactory = (option listeners.delete(listener); }; }, + ackSpooledUpdate(requestId, result) { + try { + Reflect.apply(Reflect.get(worker, "postMessage") as (value: unknown) => void, worker, [ + { type: "spool-ack", requestId, result } satisfies TelegramIngressWorkerCommand, + ]); + } catch { + // Worker may have exited after the parent committed the queue write. + } + }, async stop() { Reflect.apply(Reflect.get(worker, "postMessage") as (value: unknown) => void, worker, [ - { type: "stop" }, + { type: "stop" } satisfies TelegramIngressWorkerCommand, ]); const timeout = setTimeout(() => { void worker.terminate(); diff --git a/extensions/whatsapp/src/inbound.media.test.ts b/extensions/whatsapp/src/inbound.media.test.ts index 5e9c2b592015..e9bc5882fd31 100644 --- a/extensions/whatsapp/src/inbound.media.test.ts +++ b/extensions/whatsapp/src/inbound.media.test.ts @@ -140,15 +140,25 @@ vi.mock("openclaw/plugin-sdk/media-store", async () => { }; }); -vi.mock("./runtime.js", () => ({ - getOptionalWhatsAppRuntime: () => undefined, - getWhatsAppRuntime: () => ({ - state: { - openKeyedStore: () => createInMemoryKeyedStore(), - }, - }), - setWhatsAppRuntime: vi.fn(), -})); +vi.mock("./runtime.js", async () => { + const { createChannelIngressQueue } = await vi.importActual< + typeof import("../../../src/channels/message/ingress-queue.js") + >("../../../src/channels/message/ingress-queue.js"); + const stateDir = `/tmp/openclaw-whatsapp-inbound-media-${Date.now()}-${Math.random()}`; + return { + getOptionalWhatsAppRuntime: () => undefined, + getWhatsAppRuntime: () => ({ + state: { + resolveStateDir: () => stateDir, + openKeyedStore: () => createInMemoryKeyedStore(), + openChannelIngressQueue: ( + options?: Omit[0], "channelId">, + ) => createChannelIngressQueue({ ...(options ?? {}), channelId: "whatsapp" }), + }, + }), + setWhatsAppRuntime: vi.fn(), + }; +}); const HOME = path.join(os.tmpdir(), `openclaw-inbound-media-${crypto.randomUUID()}`); const ORIGINAL_HOME = process.env.HOME; diff --git a/extensions/whatsapp/src/inbound/durable-receive.ts b/extensions/whatsapp/src/inbound/durable-receive.ts index 4b1b7ddac69d..67f53c177199 100644 --- a/extensions/whatsapp/src/inbound/durable-receive.ts +++ b/extensions/whatsapp/src/inbound/durable-receive.ts @@ -1,6 +1,6 @@ import { createHash } from "node:crypto"; import type { WAMessage } from "baileys"; -import { createDurableInboundReceiveJournal } from "openclaw/plugin-sdk/channel-outbound"; +import { createDurableInboundReceiveJournalFromQueue } from "openclaw/plugin-sdk/channel-outbound"; import type { PluginJsonValue } from "openclaw/plugin-sdk/plugin-entry"; import { getWhatsAppRuntime } from "../runtime.js"; import { BufferJSON } from "../session.runtime.js"; @@ -56,24 +56,25 @@ export function deserializeWhatsAppDurableInboundMessage( } export function createWhatsAppDurableInboundReceiveJournal(accountId: string) { - const runtime = getWhatsAppRuntime(); const accountPart = hashNamespacePart(accountId); - return createDurableInboundReceiveJournal< + const runtime = getWhatsAppRuntime(); + const queue = runtime.state.openChannelIngressQueue< WhatsAppDurableInboundPayload, WhatsAppDurableInboundMetadata, WhatsAppDurableInboundCompletedMetadata >({ - pendingStore: runtime.state.openKeyedStore({ - namespace: `inbound.v1.pending.${accountPart}`, - maxEntries: WHATSAPP_DURABLE_INBOUND_PENDING_MAX_ENTRIES, - defaultTtlMs: WHATSAPP_DURABLE_INBOUND_PENDING_TTL_MS, - }), - completedStore: runtime.state.openKeyedStore({ - namespace: `inbound.v1.completed.${accountPart}`, - maxEntries: WHATSAPP_DURABLE_INBOUND_COMPLETED_MAX_ENTRIES, - defaultTtlMs: WHATSAPP_DURABLE_INBOUND_COMPLETED_TTL_MS, - }), - pendingTtlMs: WHATSAPP_DURABLE_INBOUND_PENDING_TTL_MS, - completedTtlMs: WHATSAPP_DURABLE_INBOUND_COMPLETED_TTL_MS, + accountId: accountPart, + stateDir: runtime.state.resolveStateDir(), + }); + return createDurableInboundReceiveJournalFromQueue({ + queue, + retention: { + pendingTtlMs: WHATSAPP_DURABLE_INBOUND_PENDING_TTL_MS, + completedTtlMs: WHATSAPP_DURABLE_INBOUND_COMPLETED_TTL_MS, + failedTtlMs: WHATSAPP_DURABLE_INBOUND_PENDING_TTL_MS, + pendingMaxEntries: WHATSAPP_DURABLE_INBOUND_PENDING_MAX_ENTRIES, + completedMaxEntries: WHATSAPP_DURABLE_INBOUND_COMPLETED_MAX_ENTRIES, + failedMaxEntries: WHATSAPP_DURABLE_INBOUND_PENDING_MAX_ENTRIES, + }, }); } diff --git a/extensions/whatsapp/src/monitor-inbox.test-harness.ts b/extensions/whatsapp/src/monitor-inbox.test-harness.ts index 707655f60108..d0186d96fcd3 100644 --- a/extensions/whatsapp/src/monitor-inbox.test-harness.ts +++ b/extensions/whatsapp/src/monitor-inbox.test-harness.ts @@ -63,6 +63,7 @@ const pluginRuntimeMocks = vi.hoisted(() => { type StoreEntry = { key: string; value: unknown; createdAt: number }; const stores = new Map>(); let nextRegisterIfAbsentError: Error | undefined; + let stateDir = `/tmp/openclaw-whatsapp-ingress-${Date.now()}-${Math.random()}`; const openKeyedStore = vi.fn((options: { namespace: string }) => { let store = stores.get(options.namespace); @@ -102,6 +103,7 @@ const pluginRuntimeMocks = vi.hoisted(() => { return { openKeyedStore, + stateDir: () => stateDir, failNextRegisterIfAbsent: (error: Error) => { nextRegisterIfAbsentError = error; }, @@ -109,6 +111,7 @@ const pluginRuntimeMocks = vi.hoisted(() => { stores.clear(); nextRegisterIfAbsentError = undefined; openKeyedStore.mockClear(); + stateDir = `/tmp/openclaw-whatsapp-ingress-${Date.now()}-${Math.random()}`; }, }; }); @@ -132,14 +135,23 @@ vi.mock("openclaw/plugin-sdk/channel-activity-runtime", async () => { }; }); -vi.mock("./runtime.js", () => ({ - getWhatsAppRuntime: () => ({ - state: { - openKeyedStore: pluginRuntimeMocks.openKeyedStore, - }, - }), - setWhatsAppRuntime: vi.fn(), -})); +vi.mock("./runtime.js", async () => { + const { createChannelIngressQueue } = await vi.importActual< + typeof import("../../../src/channels/message/ingress-queue.js") + >("../../../src/channels/message/ingress-queue.js"); + return { + getWhatsAppRuntime: () => ({ + state: { + resolveStateDir: pluginRuntimeMocks.stateDir, + openKeyedStore: pluginRuntimeMocks.openKeyedStore, + openChannelIngressQueue: ( + options?: Omit[0], "channelId">, + ) => createChannelIngressQueue({ ...(options ?? {}), channelId: "whatsapp" }), + }, + }), + setWhatsAppRuntime: vi.fn(), + }; +}); const inboundRuntimeMocks = vi.hoisted(() => { const wrapperKeys = [ diff --git a/src/channels/message/durable-receive.test.ts b/src/channels/message/durable-receive.test.ts index 82a06da38709..32151adb7e8f 100644 --- a/src/channels/message/durable-receive.test.ts +++ b/src/channels/message/durable-receive.test.ts @@ -1,9 +1,15 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; import { describe, expect, it } from "vitest"; import type { PluginStateEntry, PluginStateKeyedStore, } from "../../plugin-state/plugin-state-store.types.js"; +import { closeOpenClawStateDatabaseForTest } from "../../state/openclaw-state-db.js"; +import { createDurableInboundReceiveJournalFromQueue } from "./durable-receive.js"; import { createDurableInboundReceiveJournal } from "./durable-receive.js"; +import { createChannelIngressQueue } from "./ingress-queue.js"; type TestPayload = { body: string }; type TestMetadata = { source: string }; @@ -56,6 +62,16 @@ function createMemoryStore(): PluginStateKeyedStore { }; } +async function withTempState(fn: (stateDir: string) => Promise): Promise { + const stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-durable-receive-")); + try { + return await fn(stateDir); + } finally { + closeOpenClawStateDatabaseForTest(); + await fs.rm(stateDir, { recursive: true, force: true }); + } +} + describe("createDurableInboundReceiveJournal", () => { it("accepts pending records once and reports duplicate pending deliveries", async () => { const journal = createDurableInboundReceiveJournal< @@ -298,4 +314,70 @@ describe("createDurableInboundReceiveJournal", () => { }, ]); }); + + it("can use the shared channel ingress queue as durable storage", async () => { + await withTempState(async (stateDir) => { + const queue = createChannelIngressQueue({ + channelId: "test", + accountId: "account", + stateDir, + now: () => 10, + }); + const journal = createDurableInboundReceiveJournalFromQueue({ + queue, + retention: { completedMaxEntries: 1 }, + }); + + await expect( + journal.accept("message-1", { body: "hello" }, { metadata: { source: "live" } }), + ).resolves.toMatchObject({ + kind: "accepted", + duplicate: false, + record: { + id: "message-1", + payload: { body: "hello" }, + metadata: { source: "live" }, + receivedAt: 10, + }, + }); + + await expect(journal.pending()).resolves.toMatchObject([ + { + id: "message-1", + payload: { body: "hello" }, + metadata: { source: "live" }, + }, + ]); + + await expect(journal.release("message-1", { lastError: "retry" })).resolves.toBe(true); + await expect(journal.pending()).resolves.toMatchObject([ + { + id: "message-1", + attempts: 1, + lastError: "retry", + }, + ]); + + await journal.complete("message-1", { + metadata: { delivered: true }, + completedAt: 20, + }); + await expect(journal.accept("message-1", { body: "again" })).resolves.toMatchObject({ + kind: "completed", + duplicate: true, + record: { + id: "message-1", + completedAt: 20, + metadata: { delivered: true }, + }, + }); + + await journal.accept("message-2", { body: "new" }); + await journal.complete("message-2", { completedAt: 21 }); + await expect(journal.accept("message-1", { body: "past retention" })).resolves.toMatchObject({ + kind: "accepted", + duplicate: false, + }); + }); + }); }); diff --git a/src/channels/message/durable-receive.ts b/src/channels/message/durable-receive.ts index d6b1bdbdc50b..fd43b772ecc3 100644 --- a/src/channels/message/durable-receive.ts +++ b/src/channels/message/durable-receive.ts @@ -1,4 +1,5 @@ import type { PluginStateKeyedStore } from "../../plugin-state/plugin-state-store.types.js"; +import type { ChannelIngressQueue, ChannelIngressQueuePruneOptions } from "./ingress-queue.js"; export type DurableInboundReceivePendingRecord = { id: string; @@ -72,6 +73,11 @@ export type DurableInboundReceiveJournal; }; +export type DurableInboundReceiveQueueJournalOptions = { + queue: ChannelIngressQueue; + retention?: ChannelIngressQueuePruneOptions; +}; + function normalizeDurableInboundReceiveId(id: string): string { const normalized = id.trim(); if (!normalized) { @@ -222,3 +228,80 @@ export function createDurableInboundReceiveJournal< deletePending: (id) => options.pendingStore.delete(normalizeDurableInboundReceiveId(id)), }; } + +export function createDurableInboundReceiveJournalFromQueue< + TPayload, + TMetadata = unknown, + TCompletedMetadata = unknown, +>( + options: DurableInboundReceiveQueueJournalOptions, +): DurableInboundReceiveJournal { + const prune = async (protectId?: string) => { + if (options.retention) { + await options.queue.prune({ + ...options.retention, + ...(protectId === undefined ? {} : { protectIds: [protectId] }), + }); + } + }; + return { + accept: async (id, payload, acceptOptions) => { + await prune(); + const result = await options.queue.enqueue(normalizeDurableInboundReceiveId(id), payload, { + ...(acceptOptions?.metadata === undefined ? {} : { metadata: acceptOptions.metadata }), + ...(acceptOptions?.receivedAt === undefined + ? {} + : { receivedAt: acceptOptions.receivedAt }), + }); + await prune(normalizeDurableInboundReceiveId(id)); + if (result.kind === "accepted") { + return { kind: "accepted", duplicate: false, record: result.record }; + } + if (result.kind === "completed") { + return { kind: "completed", duplicate: true, record: result.record }; + } + if (result.kind === "pending" || result.kind === "claimed") { + return { kind: "pending", duplicate: true, record: result.record }; + } + return { + kind: "pending", + duplicate: true, + record: { + id: result.record.id, + payload, + receivedAt: result.record.failedAt, + updatedAt: result.record.failedAt, + attempts: 0, + }, + }; + }, + pending: async () => { + await prune(); + return await options.queue.listPending({ limit: "all" }); + }, + complete: async (id, completeOptions) => { + await options.queue.complete(normalizeDurableInboundReceiveId(id), { + ...(completeOptions?.metadata === undefined ? {} : { metadata: completeOptions.metadata }), + ...(completeOptions?.completedAt === undefined + ? {} + : { completedAt: completeOptions.completedAt }), + }); + await prune(normalizeDurableInboundReceiveId(id)); + }, + release: async (id, releaseOptions) => { + const released = await options.queue.release(normalizeDurableInboundReceiveId(id), { + ...(releaseOptions?.lastError === undefined ? {} : { lastError: releaseOptions.lastError }), + ...(releaseOptions?.releasedAt === undefined + ? {} + : { releasedAt: releaseOptions.releasedAt }), + }); + await prune(normalizeDurableInboundReceiveId(id)); + return released; + }, + deletePending: async (id) => { + const deleted = await options.queue.delete(normalizeDurableInboundReceiveId(id)); + await prune(); + return deleted; + }, + }; +} diff --git a/src/channels/message/index.ts b/src/channels/message/index.ts index f1de2a7921be..e5e87e7f960a 100644 --- a/src/channels/message/index.ts +++ b/src/channels/message/index.ts @@ -1,7 +1,11 @@ export { deriveDurableFinalDeliveryRequirements } from "./capabilities.js"; export { defineChannelMessageAdapter } from "./adapter.js"; export { createChannelMessageAdapterFromOutbound } from "./outbound-bridge.js"; -export { createDurableInboundReceiveJournal } from "./durable-receive.js"; +export { + createDurableInboundReceiveJournal, + createDurableInboundReceiveJournalFromQueue, +} from "./durable-receive.js"; +export { createChannelIngressQueue } from "./ingress-queue.js"; export { listDeclaredChannelMessageLiveCapabilities, listDeclaredDurableFinalCapabilities, @@ -48,8 +52,20 @@ export type { DurableInboundReceiveJournal, DurableInboundReceiveJournalOptions, DurableInboundReceivePendingRecord, + DurableInboundReceiveQueueJournalOptions, DurableInboundReceiveReleaseOptions, } from "./durable-receive.js"; +export type { + ChannelIngressQueue, + ChannelIngressQueueClaim, + ChannelIngressQueueClaimRef, + ChannelIngressQueueCompletedRecord, + ChannelIngressQueueEnqueueResult, + ChannelIngressQueueFailedRecord, + ChannelIngressQueuePruneOptions, + ChannelIngressQueueRecord, + CreateChannelIngressQueueOptions, +} from "./ingress-queue.js"; export type { ChannelMessageOutboundBridgeAdapter, ChannelMessageOutboundBridgeResult, diff --git a/src/channels/message/ingress-queue.test.ts b/src/channels/message/ingress-queue.test.ts new file mode 100644 index 000000000000..1fea6aff0505 --- /dev/null +++ b/src/channels/message/ingress-queue.test.ts @@ -0,0 +1,323 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { executeSqliteQuerySync, getNodeSqliteKysely } from "../../infra/kysely-sync.js"; +import type { DB as OpenClawStateKyselyDatabase } from "../../state/openclaw-state-db.generated.js"; +import { + closeOpenClawStateDatabaseForTest, + openOpenClawStateDatabase, +} from "../../state/openclaw-state-db.js"; +import { createChannelIngressQueue } from "./ingress-queue.js"; + +type ChannelIngressTestDatabase = Pick; + +async function withTempState(fn: (stateDir: string) => Promise): Promise { + const stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-ingress-queue-")); + try { + return await fn(stateDir); + } finally { + closeOpenClawStateDatabaseForTest(); + await fs.rm(stateDir, { recursive: true, force: true }); + } +} + +describe("channel ingress queue", () => { + afterEach(() => { + closeOpenClawStateDatabaseForTest(); + }); + + it("deduplicates pending and completed ingress events", async () => { + await withTempState(async (stateDir) => { + const queue = createChannelIngressQueue< + { text: string }, + { source: string }, + { handledBy: string } + >({ + channelId: "test", + accountId: "account", + stateDir, + now: () => 100, + }); + + const accepted = await queue.enqueue( + "event-1", + { text: "first" }, + { metadata: { source: "fixture" }, receivedAt: 50 }, + ); + const pending = await queue.enqueue("event-1", { text: "duplicate" }); + await queue.complete("event-1", { metadata: { handledBy: "worker" }, completedAt: 150 }); + const completed = await queue.enqueue("event-1", { text: "late duplicate" }); + + expect(accepted.kind).toBe("accepted"); + expect(pending.kind).toBe("pending"); + expect(pending.record.payload).toEqual({ text: "first" }); + expect(completed).toEqual({ + kind: "completed", + duplicate: true, + record: { + id: "event-1", + channelId: "test", + accountId: "account", + queueName: JSON.stringify(["test", "account"]), + completedAt: 150, + metadata: { handledBy: "worker" }, + }, + }); + expect(await queue.listPending()).toEqual([]); + + expect( + await queue.complete("missing-event", { + metadata: { handledBy: "late-worker" }, + completedAt: 200, + }), + ).toBe(true); + expect(await queue.enqueue("missing-event", { text: "late duplicate" })).toMatchObject({ + kind: "completed", + duplicate: true, + record: { + id: "missing-event", + completedAt: 200, + metadata: { handledBy: "late-worker" }, + }, + }); + + await queue.enqueue(" spaced-event ", { text: "spaced" }); + expect(await queue.complete(" spaced-event ", { completedAt: 250 })).toBe(true); + expect(await queue.enqueue("spaced-event", { text: "duplicate" })).toMatchObject({ + kind: "completed", + duplicate: true, + record: { id: "spaced-event", completedAt: 250 }, + }); + }); + }); + + it("keeps channel and account queue identities unambiguous", async () => { + await withTempState(async (stateDir) => { + const first = createChannelIngressQueue<{ text: string }>({ + channelId: "a", + accountId: "b:c", + stateDir, + }); + const second = createChannelIngressQueue<{ text: string }>({ + channelId: "a:b", + accountId: "c", + stateDir, + }); + + expect(await first.enqueue("same-id", { text: "first" })).toMatchObject({ + kind: "accepted", + }); + expect(await second.enqueue("same-id", { text: "second" })).toMatchObject({ + kind: "accepted", + }); + + await first.complete("same-id"); + + expect(await first.enqueue("same-id", { text: "first duplicate" })).toMatchObject({ + kind: "completed", + }); + expect(await second.enqueue("same-id", { text: "second duplicate" })).toMatchObject({ + kind: "pending", + record: { payload: { text: "second" } }, + }); + }); + }); + + it("can bound pending scans and prune stale pending rows", async () => { + await withTempState(async (stateDir) => { + let clock = 1; + const queue = createChannelIngressQueue<{ index: number }>({ + channelId: "test", + accountId: "account", + stateDir, + now: () => clock++, + }); + + await queue.enqueue("0002", { index: 2 }); + await queue.enqueue("0001", { index: 1 }); + await queue.enqueue("0003", { index: 3 }); + + expect( + (await queue.listPending({ limit: 2, orderBy: "id" })).map((record) => record.id), + ).toEqual(["0001", "0002"]); + expect(await queue.prune({ pendingTtlMs: 3, pendingMaxEntries: 1, now: 7 })).toBe(2); + expect((await queue.listPending({ limit: "all" })).map((record) => record.id)).toEqual([ + "0003", + ]); + }); + }); + + it("does not prune protected rows while enforcing max-entry limits", async () => { + await withTempState(async (stateDir) => { + const queue = createChannelIngressQueue<{ index: number }>({ + channelId: "test", + accountId: "account", + stateDir, + now: () => 10, + }); + + await queue.enqueue("z", { index: 1 }); + await queue.enqueue("a", { index: 2 }); + + expect(await queue.prune({ pendingMaxEntries: 1, protectIds: ["a"] })).toBe(0); + expect( + (await queue.listPending({ limit: "all", orderBy: "id" })).map((row) => row.id), + ).toEqual(["a", "z"]); + }); + }); + + it("prunes max-entry overflow across bounded batches", async () => { + await withTempState(async (stateDir) => { + let clock = 1; + const queue = createChannelIngressQueue<{ index: number }>({ + channelId: "test", + accountId: "account", + stateDir, + now: () => clock++, + }); + + for (let index = 0; index < 520; index += 1) { + await queue.enqueue(String(index).padStart(4, "0"), { index }); + } + + expect(await queue.prune({ pendingMaxEntries: 2 })).toBe(518); + expect((await queue.listPending({ limit: "all" })).map((row) => row.id)).toEqual([ + "0518", + "0519", + ]); + }); + }); + + it("claims, releases, and skips blocked lanes", async () => { + await withTempState(async (stateDir) => { + let clock = 1; + const queue = createChannelIngressQueue<{ text: string }>({ + channelId: "test", + accountId: "account", + stateDir, + now: () => clock++, + }); + + await queue.enqueue("a", { text: "blocked" }, { laneKey: "chat-1", receivedAt: 1 }); + await queue.enqueue("b", { text: "open" }, { laneKey: "chat-2", receivedAt: 2 }); + + const claimed = await queue.claimNext({ + ownerId: "worker", + blockedLaneKeys: ["chat-1"], + }); + + expect(claimed?.id).toBe("b"); + if (!claimed) { + throw new Error("Expected a claimed ingress event"); + } + expect(await queue.release(claimed, { lastError: "retry", releasedAt: 20 })).toBe(true); + expect((await queue.listPending()).find((record) => record.id === "b")).toMatchObject({ + attempts: 1, + lastAttemptAt: 20, + lastError: "retry", + }); + }); + }); + + it("requires claim tokens before mutating claimed rows", async () => { + await withTempState(async (stateDir) => { + const queue = createChannelIngressQueue<{ text: string }>({ + channelId: "test", + accountId: "account", + stateDir, + now: () => 10, + }); + + await queue.enqueue("event-1", { text: "claimed" }); + const claimed = await queue.claim("event-1", { ownerId: "worker" }); + if (!claimed) { + throw new Error("Expected a claimed ingress event"); + } + + expect(await queue.complete("event-1")).toBe(false); + expect(await queue.release("event-1")).toBe(false); + expect(await queue.fail("event-1", { reason: "stale-handler" })).toBe(false); + expect(await queue.delete("event-1")).toBe(false); + + expect(await queue.complete(claimed, { completedAt: 20 })).toBe(true); + const duplicate = await queue.enqueue("event-1", { text: "duplicate" }); + expect(duplicate.kind).toBe("completed"); + }); + }); + + it("recovers stale claims and prunes completed or failed rows", async () => { + await withTempState(async (stateDir) => { + const queue = createChannelIngressQueue<{ text: string }>({ + channelId: "test", + accountId: "account", + stateDir, + now: () => 10, + }); + + await queue.enqueue("old", { text: "old" }); + await queue.enqueue("keep", { text: "keep" }); + const old = await queue.claim("old", { ownerId: "worker" }); + const keep = await queue.claim("keep", { ownerId: "worker" }); + if (!keep) { + throw new Error("Expected a claimed ingress event"); + } + + expect( + await queue.recoverStaleClaims({ + staleMs: 5, + now: 20, + shouldRecover: (claim) => claim.id === old?.id, + }), + ).toBe(1); + expect((await queue.listPending()).map((record) => record.id)).toEqual(["old"]); + expect((await queue.listClaims()).map((record) => record.id)).toEqual(["keep"]); + + await queue.complete("old", { completedAt: 25 }); + await queue.fail(keep, { reason: "poison", message: "bad", failedAt: 25 }); + await queue.enqueue("retry", { text: "retry" }); + await queue.release("retry", { lastError: "stale retry text", releasedAt: 26 }); + await queue.complete("retry", { completedAt: 27 }); + + const database = openOpenClawStateDatabase({ + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + }); + const kysely = getNodeSqliteKysely(database.db); + const rows = executeSqliteQuerySync( + database.db, + kysely + .selectFrom("channel_ingress_events") + .select(["event_id", "payload_json", "metadata_json", "last_attempt_at", "last_error"]) + .where("event_id", "in", ["old", "keep", "retry"]) + .orderBy("event_id", "asc"), + ).rows; + expect(rows).toEqual([ + { + event_id: "keep", + last_attempt_at: null, + last_error: "bad", + metadata_json: null, + payload_json: "null", + }, + { + event_id: "old", + last_attempt_at: null, + last_error: null, + metadata_json: null, + payload_json: "null", + }, + { + event_id: "retry", + last_attempt_at: null, + last_error: null, + metadata_json: null, + payload_json: "null", + }, + ]); + + expect(await queue.prune({ completedTtlMs: 10, failedTtlMs: 10, now: 40 })).toBe(3); + expect(await queue.listPending()).toEqual([]); + expect(await queue.listClaims()).toEqual([]); + }); + }); +}); diff --git a/src/channels/message/ingress-queue.ts b/src/channels/message/ingress-queue.ts new file mode 100644 index 000000000000..27de960d15ca --- /dev/null +++ b/src/channels/message/ingress-queue.ts @@ -0,0 +1,829 @@ +import { randomUUID } from "node:crypto"; +import type { DatabaseSync } from "node:sqlite"; +import type { Selectable } from "kysely"; +import { + executeSqliteQuerySync, + executeSqliteQueryTakeFirstSync, + getNodeSqliteKysely, +} from "../../infra/kysely-sync.js"; +import type { + ChannelIngressEvents, + DB as OpenClawStateKyselyDatabase, +} from "../../state/openclaw-state-db.generated.js"; +import { + openOpenClawStateDatabase, + runOpenClawStateWriteTransaction, +} from "../../state/openclaw-state-db.js"; + +export type ChannelIngressQueueRecord = { + id: string; + channelId: string; + accountId: string; + queueName: string; + payload: TPayload; + metadata?: TMetadata; + receivedAt: number; + updatedAt: number; + laneKey?: string; + attempts: number; + lastAttemptAt?: number; + lastError?: string; +}; + +export type ChannelIngressQueueClaim = ChannelIngressQueueRecord< + TPayload, + TMetadata +> & { + claim: { + token: string; + ownerId: string; + claimedAt: number; + }; +}; + +export type ChannelIngressQueueClaimRef = { + id: string; + claim: { + token: string; + }; +}; + +export type ChannelIngressQueueCompletedRecord = { + id: string; + channelId: string; + accountId: string; + queueName: string; + completedAt: number; + metadata?: TCompletedMetadata; +}; + +export type ChannelIngressQueueFailedRecord = { + id: string; + channelId: string; + accountId: string; + queueName: string; + failedAt: number; + reason: string; + message?: string; +}; + +export type ChannelIngressQueuePruneOptions = { + pendingTtlMs?: number; + completedTtlMs?: number; + failedTtlMs?: number; + pendingMaxEntries?: number; + completedMaxEntries?: number; + failedMaxEntries?: number; + protectIds?: Iterable; + now?: number; +}; + +export type ChannelIngressQueueEnqueueResult = + | { + kind: "accepted"; + duplicate: false; + record: ChannelIngressQueueRecord; + } + | { + kind: "pending"; + duplicate: true; + record: ChannelIngressQueueRecord; + } + | { + kind: "claimed"; + duplicate: true; + record: ChannelIngressQueueClaim; + } + | { + kind: "completed"; + duplicate: true; + record: ChannelIngressQueueCompletedRecord; + } + | { + kind: "failed"; + duplicate: true; + record: ChannelIngressQueueFailedRecord; + }; + +export type ChannelIngressQueue = { + enqueue( + id: string, + payload: TPayload, + options?: { + metadata?: TMetadata; + receivedAt?: number; + laneKey?: string; + }, + ): Promise>; + listPending(options?: { + limit?: number | "all"; + orderBy?: "received" | "id"; + }): Promise>>; + listClaims(): Promise>>; + claimNext(options?: { + ownerId?: string; + blockedLaneKeys?: Iterable; + staleMs?: number; + }): Promise | null>; + claim( + id: string, + options?: { ownerId?: string }, + ): Promise | null>; + complete( + idOrClaim: string | ChannelIngressQueueClaimRef, + options?: { metadata?: TCompletedMetadata; completedAt?: number }, + ): Promise; + release( + idOrClaim: string | ChannelIngressQueueClaimRef, + options?: { lastError?: string; releasedAt?: number }, + ): Promise; + fail( + idOrClaim: string | ChannelIngressQueueClaimRef, + options: { reason: string; message?: string; failedAt?: number }, + ): Promise; + delete( + idOrClaim: + | string + | ChannelIngressQueueRecord + | ChannelIngressQueueClaimRef, + ): Promise; + recoverStaleClaims(options?: { + staleMs?: number; + now?: number; + shouldRecover?: ( + claim: ChannelIngressQueueClaim, + ) => boolean | Promise; + }): Promise; + prune(options?: ChannelIngressQueuePruneOptions): Promise; +}; + +export type CreateChannelIngressQueueOptions = { + channelId: string; + accountId?: string; + stateDir?: string; + now?: () => number; +}; + +type ChannelIngressDatabase = Pick; +type ChannelIngressRow = Selectable; + +function normalizePart(value: string | undefined, fallback: string): string { + const normalized = value?.trim(); + return normalized ? normalized : fallback; +} + +function openStateDatabase(stateDir?: string) { + return openOpenClawStateDatabase({ + env: stateDir ? { ...process.env, OPENCLAW_STATE_DIR: stateDir } : process.env, + }); +} + +function getChannelIngressKysely(db: DatabaseSync) { + return getNodeSqliteKysely(db); +} + +function affectedRows(result: { numAffectedRows?: bigint }): number { + return Number(result.numAffectedRows ?? 0n); +} + +function parseJson(value: string): T { + return JSON.parse(value) as T; +} + +function baseRecord( + row: ChannelIngressRow, +): ChannelIngressQueueRecord { + return { + id: row.event_id, + channelId: row.channel_id, + accountId: row.account_id, + queueName: row.queue_name, + payload: parseJson(row.payload_json), + ...(row.metadata_json === null ? {} : { metadata: parseJson(row.metadata_json) }), + receivedAt: Number(row.received_at), + updatedAt: Number(row.updated_at), + ...(row.lane_key === null ? {} : { laneKey: row.lane_key }), + attempts: Number(row.attempts), + ...(row.last_attempt_at === null ? {} : { lastAttemptAt: Number(row.last_attempt_at) }), + ...(row.last_error === null ? {} : { lastError: row.last_error }), + }; +} + +function claimedRecord( + row: ChannelIngressRow, +): ChannelIngressQueueClaim { + return { + ...baseRecord(row), + claim: { + token: row.claim_token ?? "", + ownerId: row.claim_owner ?? "", + claimedAt: Number(row.claimed_at ?? 0), + }, + }; +} + +function completedRecord( + row: ChannelIngressRow, +): ChannelIngressQueueCompletedRecord { + return { + id: row.event_id, + channelId: row.channel_id, + accountId: row.account_id, + queueName: row.queue_name, + completedAt: Number(row.completed_at ?? row.updated_at), + ...(row.completed_metadata_json === null + ? {} + : { metadata: parseJson(row.completed_metadata_json) }), + }; +} + +function failedRecord(row: ChannelIngressRow): ChannelIngressQueueFailedRecord { + return { + id: row.event_id, + channelId: row.channel_id, + accountId: row.account_id, + queueName: row.queue_name, + failedAt: Number(row.failed_at ?? row.updated_at), + reason: row.failed_reason ?? "failed", + ...(row.last_error === null ? {} : { message: row.last_error }), + }; +} + +function selectRow(db: DatabaseSync, queueName: string, id: string) { + const kysely = getChannelIngressKysely(db); + return executeSqliteQueryTakeFirstSync( + db, + kysely + .selectFrom("channel_ingress_events") + .selectAll() + .where("queue_name", "=", queueName) + .where("event_id", "=", id), + ); +} + +function idFrom(idOrRecord: string | { id: string }): string { + const id = normalizePart(typeof idOrRecord === "string" ? idOrRecord : idOrRecord.id, ""); + if (!id) { + throw new Error("Channel ingress event id cannot be empty"); + } + return id; +} + +function claimTokenFrom( + idOrClaim: string | { id: string; claim?: { token: string } }, +): string | null { + return typeof idOrClaim === "string" ? null : (idOrClaim.claim?.token ?? null); +} + +function rowToEnqueueResult( + row: ChannelIngressRow, +): ChannelIngressQueueEnqueueResult { + if (row.status === "completed") { + return { kind: "completed", duplicate: true, record: completedRecord(row) }; + } + if (row.status === "failed") { + return { kind: "failed", duplicate: true, record: failedRecord(row) }; + } + if (row.status === "claimed") { + return { kind: "claimed", duplicate: true, record: claimedRecord(row) }; + } + return { kind: "pending", duplicate: true, record: baseRecord(row) }; +} + +function normalizeLimit(limit: number | "all" | undefined): number { + return limit === "all" ? Number.MAX_SAFE_INTEGER : Math.max(1, Math.floor(limit ?? 100)); +} + +function normalizeMaxEntries(value: number | undefined): number | null { + return value === undefined ? null : Math.max(0, Math.floor(value)); +} + +function normalizedProtectedIds(ids: Iterable | undefined): string[] { + return [...(ids ?? [])].map((id) => id.trim()).filter(Boolean); +} + +function queueNameForParts(channelId: string, accountId: string): string { + return JSON.stringify([channelId, accountId]); +} + +export function createChannelIngressQueue< + TPayload, + TMetadata = unknown, + TCompletedMetadata = unknown, +>( + options: CreateChannelIngressQueueOptions, +): ChannelIngressQueue { + const channelId = normalizePart(options.channelId, "unknown"); + const accountId = normalizePart(options.accountId, "default"); + const queueName = queueNameForParts(channelId, accountId); + const now = options.now ?? Date.now; + + const enqueue: ChannelIngressQueue["enqueue"] = async ( + id, + payload, + enqueueOptions, + ) => { + const eventId = normalizePart(id, ""); + if (!eventId) { + throw new Error("Channel ingress event id cannot be empty"); + } + const receivedAt = enqueueOptions?.receivedAt ?? now(); + const updatedAt = now(); + const database = openStateDatabase(options.stateDir); + return runOpenClawStateWriteTransaction( + (tx) => { + const kysely = getChannelIngressKysely(tx.db); + const insert = executeSqliteQuerySync( + tx.db, + kysely + .insertInto("channel_ingress_events") + .values({ + queue_name: queueName, + event_id: eventId, + channel_id: channelId, + account_id: accountId, + status: "pending", + lane_key: enqueueOptions?.laneKey ?? null, + payload_json: JSON.stringify(payload), + metadata_json: + enqueueOptions?.metadata === undefined + ? null + : JSON.stringify(enqueueOptions.metadata), + received_at: receivedAt, + updated_at: updatedAt, + attempts: 0, + }) + .onConflict((conflict) => conflict.columns(["queue_name", "event_id"]).doNothing()), + ); + const row = selectRow(tx.db, queueName, eventId); + if (!row) { + throw new Error(`Failed to read channel ingress event ${queueName}/${eventId}`); + } + if (affectedRows(insert) > 0) { + return { + kind: "accepted", + duplicate: false, + record: baseRecord(row), + }; + } + return rowToEnqueueResult(row); + }, + { path: database.path }, + ); + }; + + const listPending: ChannelIngressQueue< + TPayload, + TMetadata, + TCompletedMetadata + >["listPending"] = async (listOptions) => { + const { db } = openStateDatabase(options.stateDir); + const kysely = getChannelIngressKysely(db); + const baseQuery = kysely + .selectFrom("channel_ingress_events") + .selectAll() + .where("queue_name", "=", queueName) + .where("status", "=", "pending") + .limit(normalizeLimit(listOptions?.limit)); + const query = + listOptions?.orderBy === "id" + ? baseQuery.orderBy("event_id", "asc") + : baseQuery.orderBy("received_at", "asc").orderBy("event_id", "asc"); + const rows = executeSqliteQuerySync(db, query).rows; + return rows.map((row) => baseRecord(row)); + }; + + const listClaims: ChannelIngressQueue< + TPayload, + TMetadata, + TCompletedMetadata + >["listClaims"] = async () => { + const { db } = openStateDatabase(options.stateDir); + const kysely = getChannelIngressKysely(db); + const rows = executeSqliteQuerySync( + db, + kysely + .selectFrom("channel_ingress_events") + .selectAll() + .where("queue_name", "=", queueName) + .where("status", "=", "claimed") + .orderBy("claimed_at", "asc") + .orderBy("received_at", "asc") + .orderBy("event_id", "asc"), + ).rows; + return rows.map((row) => claimedRecord(row)); + }; + + const recoverStaleClaims: ChannelIngressQueue< + TPayload, + TMetadata, + TCompletedMetadata + >["recoverStaleClaims"] = async (recoverOptions) => { + const staleMs = Math.max(0, Math.floor(recoverOptions?.staleMs ?? 0)); + const cutoff = (recoverOptions?.now ?? now()) - staleMs; + const claims = (await listClaims()).filter((claim) => claim.claim.claimedAt <= cutoff); + let recovered = 0; + for (const claim of claims) { + if (recoverOptions?.shouldRecover && !(await recoverOptions.shouldRecover(claim))) { + continue; + } + if (await release(claim, { releasedAt: recoverOptions?.now ?? now() })) { + recovered += 1; + } + } + return recovered; + }; + + const claimNext: ChannelIngressQueue< + TPayload, + TMetadata, + TCompletedMetadata + >["claimNext"] = async (claimOptions) => { + if (claimOptions?.staleMs !== undefined) { + await recoverStaleClaims({ staleMs: claimOptions.staleMs }); + } + const blocked = new Set( + [...(claimOptions?.blockedLaneKeys ?? [])].map((key) => key.trim()).filter(Boolean), + ); + const database = openStateDatabase(options.stateDir); + return runOpenClawStateWriteTransaction( + (tx) => { + const kysely = getChannelIngressKysely(tx.db); + const baseSelect = kysely + .selectFrom("channel_ingress_events") + .select(["event_id", "lane_key"]) + .where("queue_name", "=", queueName) + .where("status", "=", "pending"); + const select = + blocked.size === 0 + ? baseSelect + : baseSelect.where((eb) => + eb.or([eb("lane_key", "is", null), eb("lane_key", "not in", [...blocked])]), + ); + const selected = executeSqliteQueryTakeFirstSync( + tx.db, + select.orderBy("received_at", "asc").orderBy("event_id", "asc").limit(1), + ); + if (!selected) { + return null; + } + const token = randomUUID(); + const claimedAt = now(); + const ownerId = normalizePart(claimOptions?.ownerId, `${process.pid}`); + const result = executeSqliteQuerySync( + tx.db, + kysely + .updateTable("channel_ingress_events") + .set({ + status: "claimed", + claim_token: token, + claim_owner: ownerId, + claimed_at: claimedAt, + updated_at: claimedAt, + }) + .where("queue_name", "=", queueName) + .where("event_id", "=", selected.event_id) + .where("status", "=", "pending"), + ); + if (affectedRows(result) === 0) { + return null; + } + const row = selectRow(tx.db, queueName, selected.event_id); + return row ? claimedRecord(row) : null; + }, + { path: database.path }, + ); + }; + + const claim: ChannelIngressQueue["claim"] = async ( + id, + claimOptions, + ) => { + const eventId = normalizePart(id, ""); + if (!eventId) { + throw new Error("Channel ingress event id cannot be empty"); + } + const database = openStateDatabase(options.stateDir); + return runOpenClawStateWriteTransaction( + (tx) => { + const kysely = getChannelIngressKysely(tx.db); + const token = randomUUID(); + const claimedAt = now(); + const ownerId = normalizePart(claimOptions?.ownerId, `${process.pid}`); + const result = executeSqliteQuerySync( + tx.db, + kysely + .updateTable("channel_ingress_events") + .set({ + status: "claimed", + claim_token: token, + claim_owner: ownerId, + claimed_at: claimedAt, + updated_at: claimedAt, + }) + .where("queue_name", "=", queueName) + .where("event_id", "=", eventId) + .where("status", "=", "pending"), + ); + if (affectedRows(result) === 0) { + return null; + } + const row = selectRow(tx.db, queueName, eventId); + return row ? claimedRecord(row) : null; + }, + { path: database.path }, + ); + }; + + const complete: ChannelIngressQueue["complete"] = async ( + idOrClaim, + completeOptions, + ) => { + const eventId = idFrom(idOrClaim); + const token = claimTokenFrom(idOrClaim); + const completedAt = completeOptions?.completedAt ?? now(); + const database = openStateDatabase(options.stateDir); + return runOpenClawStateWriteTransaction( + (tx) => { + const kysely = getChannelIngressKysely(tx.db); + const baseUpdate = kysely + .updateTable("channel_ingress_events") + .set({ + status: "completed", + completed_at: completedAt, + completed_metadata_json: + completeOptions?.metadata === undefined + ? null + : JSON.stringify(completeOptions.metadata), + payload_json: "null", + metadata_json: null, + claim_token: null, + claim_owner: null, + claimed_at: null, + last_attempt_at: null, + last_error: null, + updated_at: completedAt, + }) + .where("queue_name", "=", queueName) + .where("event_id", "=", eventId); + const update = + token === null + ? baseUpdate.where("status", "=", "pending") + : baseUpdate.where("status", "=", "claimed").where("claim_token", "=", token); + const result = executeSqliteQuerySync(tx.db, update); + if (affectedRows(result) > 0) { + return true; + } + if (token !== null) { + return false; + } + const insert = executeSqliteQuerySync( + tx.db, + kysely + .insertInto("channel_ingress_events") + .values({ + queue_name: queueName, + event_id: eventId, + channel_id: channelId, + account_id: accountId, + status: "completed", + lane_key: null, + payload_json: "null", + metadata_json: null, + received_at: completedAt, + updated_at: completedAt, + attempts: 0, + completed_at: completedAt, + completed_metadata_json: + completeOptions?.metadata === undefined + ? null + : JSON.stringify(completeOptions.metadata), + }) + .onConflict((conflict) => conflict.columns(["queue_name", "event_id"]).doNothing()), + ); + return affectedRows(insert) > 0; + }, + { path: database.path }, + ); + }; + + const release: ChannelIngressQueue["release"] = async ( + idOrClaim, + releaseOptions, + ) => { + const eventId = idFrom(idOrClaim); + const token = claimTokenFrom(idOrClaim); + const releasedAt = releaseOptions?.releasedAt ?? now(); + const database = openStateDatabase(options.stateDir); + return runOpenClawStateWriteTransaction( + (tx) => { + const kysely = getChannelIngressKysely(tx.db); + const baseUpdate = kysely + .updateTable("channel_ingress_events") + .set((eb) => ({ + status: "pending", + claim_token: null, + claim_owner: null, + claimed_at: null, + attempts: eb("attempts", "+", 1), + last_attempt_at: releasedAt, + ...(releaseOptions?.lastError === undefined + ? {} + : { last_error: releaseOptions.lastError }), + updated_at: releasedAt, + })) + .where("queue_name", "=", queueName) + .where("event_id", "=", eventId); + const update = + token === null + ? baseUpdate.where("status", "=", "pending") + : baseUpdate.where("status", "=", "claimed").where("claim_token", "=", token); + return affectedRows(executeSqliteQuerySync(tx.db, update)) > 0; + }, + { path: database.path }, + ); + }; + + const fail: ChannelIngressQueue["fail"] = async ( + idOrClaim, + failOptions, + ) => { + const eventId = idFrom(idOrClaim); + const token = claimTokenFrom(idOrClaim); + const failedAt = failOptions.failedAt ?? now(); + const database = openStateDatabase(options.stateDir); + return runOpenClawStateWriteTransaction( + (tx) => { + const kysely = getChannelIngressKysely(tx.db); + const baseUpdate = kysely + .updateTable("channel_ingress_events") + .set({ + status: "failed", + failed_at: failedAt, + failed_reason: failOptions.reason, + last_error: failOptions.message ?? null, + payload_json: "null", + metadata_json: null, + claim_token: null, + claim_owner: null, + claimed_at: null, + updated_at: failedAt, + }) + .where("queue_name", "=", queueName) + .where("event_id", "=", eventId); + const update = + token === null + ? baseUpdate.where("status", "=", "pending") + : baseUpdate.where("status", "=", "claimed").where("claim_token", "=", token); + return affectedRows(executeSqliteQuerySync(tx.db, update)) > 0; + }, + { path: database.path }, + ); + }; + + const deleteEntry: ChannelIngressQueue< + TPayload, + TMetadata, + TCompletedMetadata + >["delete"] = async (idOrRecord) => { + const eventId = idFrom(idOrRecord); + const token = claimTokenFrom(idOrRecord); + const database = openStateDatabase(options.stateDir); + return runOpenClawStateWriteTransaction( + (tx) => { + const kysely = getChannelIngressKysely(tx.db); + const baseDelete = kysely + .deleteFrom("channel_ingress_events") + .where("queue_name", "=", queueName) + .where("event_id", "=", eventId); + const deleteQuery = + token === null + ? baseDelete.where("status", "=", "pending") + : baseDelete.where("status", "=", "claimed").where("claim_token", "=", token); + return affectedRows(executeSqliteQuerySync(tx.db, deleteQuery)) > 0; + }, + { path: database.path }, + ); + }; + + const prune: ChannelIngressQueue["prune"] = async ( + pruneOptions, + ) => { + const current = pruneOptions?.now ?? now(); + const pendingCutoff = + pruneOptions?.pendingTtlMs === undefined ? null : current - pruneOptions.pendingTtlMs; + const completedCutoff = + pruneOptions?.completedTtlMs === undefined ? null : current - pruneOptions.completedTtlMs; + const failedCutoff = + pruneOptions?.failedTtlMs === undefined ? null : current - pruneOptions.failedTtlMs; + const pendingMaxEntries = normalizeMaxEntries(pruneOptions?.pendingMaxEntries); + const completedMaxEntries = normalizeMaxEntries(pruneOptions?.completedMaxEntries); + const failedMaxEntries = normalizeMaxEntries(pruneOptions?.failedMaxEntries); + const protectIds = normalizedProtectedIds(pruneOptions?.protectIds); + if ( + pendingCutoff === null && + completedCutoff === null && + failedCutoff === null && + pendingMaxEntries === null && + completedMaxEntries === null && + failedMaxEntries === null + ) { + return 0; + } + const database = openStateDatabase(options.stateDir); + return runOpenClawStateWriteTransaction( + (tx) => { + const kysely = getChannelIngressKysely(tx.db); + let deleted = 0; + if (pendingCutoff !== null) { + let deleteQuery = kysely + .deleteFrom("channel_ingress_events") + .where("queue_name", "=", queueName) + .where("status", "=", "pending") + .where("updated_at", "<", pendingCutoff); + if (protectIds.length > 0) { + deleteQuery = deleteQuery.where("event_id", "not in", protectIds); + } + deleted += affectedRows(executeSqliteQuerySync(tx.db, deleteQuery)); + } + if (completedCutoff !== null) { + let deleteQuery = kysely + .deleteFrom("channel_ingress_events") + .where("queue_name", "=", queueName) + .where("status", "=", "completed") + .where("completed_at", "<", completedCutoff); + if (protectIds.length > 0) { + deleteQuery = deleteQuery.where("event_id", "not in", protectIds); + } + deleted += affectedRows(executeSqliteQuerySync(tx.db, deleteQuery)); + } + if (failedCutoff !== null) { + let deleteQuery = kysely + .deleteFrom("channel_ingress_events") + .where("queue_name", "=", queueName) + .where("status", "=", "failed") + .where("failed_at", "<", failedCutoff); + if (protectIds.length > 0) { + deleteQuery = deleteQuery.where("event_id", "not in", protectIds); + } + deleted += affectedRows(executeSqliteQuerySync(tx.db, deleteQuery)); + } + const pruneMaxEntries = (status: string, maxEntries: number | null) => { + if (maxEntries === null) { + return; + } + const batchSize = 500; + const protectedSet = new Set(protectIds); + while (true) { + const rowsToDelete = executeSqliteQuerySync( + tx.db, + kysely + .selectFrom("channel_ingress_events") + .select("event_id") + .where("queue_name", "=", queueName) + .where("status", "=", status) + .orderBy("updated_at", "desc") + .orderBy("event_id", "desc") + .limit(maxEntries + batchSize), + ).rows.slice(maxEntries); + const ids = rowsToDelete + .map((row) => row.event_id) + .filter((id) => !protectedSet.has(id)); + if (ids.length === 0) { + return; + } + deleted += affectedRows( + executeSqliteQuerySync( + tx.db, + kysely + .deleteFrom("channel_ingress_events") + .where("queue_name", "=", queueName) + .where("status", "=", status) + .where("event_id", "in", ids), + ), + ); + } + }; + pruneMaxEntries("pending", pendingMaxEntries); + pruneMaxEntries("completed", completedMaxEntries); + pruneMaxEntries("failed", failedMaxEntries); + return deleted; + }, + { path: database.path }, + ); + }; + + return { + enqueue, + listPending, + listClaims, + claimNext, + claim, + complete, + release, + fail, + delete: deleteEntry, + recoverStaleClaims, + prune, + }; +} diff --git a/src/plugin-sdk/channel-outbound.ts b/src/plugin-sdk/channel-outbound.ts index 2df0971d6bf0..2f955f6c6caa 100644 --- a/src/plugin-sdk/channel-outbound.ts +++ b/src/plugin-sdk/channel-outbound.ts @@ -78,6 +78,7 @@ export { classifyDurableSendRecoveryState, createChannelMessageAdapterFromOutbound, createDurableInboundReceiveJournal, + createDurableInboundReceiveJournalFromQueue, createMessageReceiptFromOutboundResults, listMessageReceiptPlatformIds, createMessageReceiveContext, @@ -134,6 +135,14 @@ export type { ChannelMessageUnknownSendReconciliationResult, CreateChannelReplyPipelineParams, CreateChannelMessageAdapterFromOutboundParams, + ChannelIngressQueue, + ChannelIngressQueueClaim, + ChannelIngressQueueClaimRef, + ChannelIngressQueueCompletedRecord, + ChannelIngressQueueEnqueueResult, + ChannelIngressQueueFailedRecord, + ChannelIngressQueuePruneOptions, + ChannelIngressQueueRecord, DeriveDurableFinalDeliveryRequirementsParams, ChannelMessageLiveCapabilityProof, ChannelMessageLiveCapabilityProofMap, @@ -155,6 +164,7 @@ export type { DurableInboundReceiveJournal, DurableInboundReceiveJournalOptions, DurableInboundReceivePendingRecord, + DurableInboundReceiveQueueJournalOptions, DurableInboundReceiveReleaseOptions, DurableMessageSendIntent, DurableMessageSendState, diff --git a/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts b/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts index 5b622ce8f88e..6e8ce15f750f 100644 --- a/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts +++ b/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts @@ -767,6 +767,9 @@ export function createPluginRuntimeMock(overrides: DeepPartial = openSyncKeyedStore: vi.fn(() => { throw new Error("openSyncKeyedStore mock is not configured"); }) as unknown as PluginRuntime["state"]["openSyncKeyedStore"], + openChannelIngressQueue: vi.fn(() => { + throw new Error("openChannelIngressQueue mock is not configured"); + }) as unknown as PluginRuntime["state"]["openChannelIngressQueue"], }, tasks: { runs: { diff --git a/src/plugins/registry.ts b/src/plugins/registry.ts index 8ac1c83f40ad..8501d116ca9f 100644 --- a/src/plugins/registry.ts +++ b/src/plugins/registry.ts @@ -15,6 +15,7 @@ import { } from "../agents/harness/registry.js"; import type { AgentHarness } from "../agents/harness/types.js"; import type { AnyAgentTool } from "../agents/tools/common.js"; +import { createChannelIngressQueue } from "../channels/message/ingress-queue.js"; import type { ChannelPlugin } from "../channels/plugins/types.plugin.js"; import { normalizeCommandDescriptorName, @@ -2629,6 +2630,17 @@ export function createPluginRegistry(registryParams: PluginRegistryParams) { assertPluginStateAllowed(); return createPluginStateSyncKeyedStore(pluginId, options); }, + openChannelIngressQueue: ( + options?: Omit[0], "channelId">, + ) => { + assertPluginStateAllowed(); + const stateDir = options?.stateDir ?? baseState.resolveStateDir(); + return createChannelIngressQueue({ + ...(options ?? {}), + channelId: pluginId, + stateDir, + }); + }, } satisfies PluginRuntime["state"]; } if (prop === "config") { diff --git a/src/plugins/runtime/index.ts b/src/plugins/runtime/index.ts index ea9632e66c03..0fbf6578867a 100644 --- a/src/plugins/runtime/index.ts +++ b/src/plugins/runtime/index.ts @@ -256,6 +256,11 @@ export function createPluginRuntime(_options: CreatePluginRuntimeOptions = {}): openSyncKeyedStore: () => { throw new Error("openSyncKeyedStore is only available through the plugin runtime proxy."); }, + openChannelIngressQueue: () => { + throw new Error( + "openChannelIngressQueue is only available through the plugin runtime proxy.", + ); + }, }, tasks, taskFlow, diff --git a/src/plugins/runtime/types-core.ts b/src/plugins/runtime/types-core.ts index b6cf629030b4..8b1a0a0cd3d9 100644 --- a/src/plugins/runtime/types-core.ts +++ b/src/plugins/runtime/types-core.ts @@ -314,6 +314,16 @@ export type PluginRuntimeCore = { openSyncKeyedStore: ( options: import("../../plugin-state/plugin-state-store.types.js").OpenKeyedStoreOptions, ) => import("../../plugin-state/plugin-state-store.types.js").PluginStateSyncKeyedStore; + openChannelIngressQueue: ( + options?: Omit< + import("../../channels/message/ingress-queue.js").CreateChannelIngressQueueOptions, + "channelId" + >, + ) => import("../../channels/message/ingress-queue.js").ChannelIngressQueue< + TPayload, + TMetadata, + TCompletedMetadata + >; }; tasks: { runs: PluginRuntimeTaskRuns; diff --git a/src/state/openclaw-state-db.generated.d.ts b/src/state/openclaw-state-db.generated.d.ts index 0e74c85b5128..16fe7d08c63b 100644 --- a/src/state/openclaw-state-db.generated.d.ts +++ b/src/state/openclaw-state-db.generated.d.ts @@ -144,6 +144,29 @@ export interface CaptureSessions { started_at: number; } +export interface ChannelIngressEvents { + account_id: string; + attempts: Generated; + channel_id: string; + claim_owner: string | null; + claim_token: string | null; + claimed_at: number | null; + completed_at: number | null; + completed_metadata_json: string | null; + event_id: string; + failed_at: number | null; + failed_reason: string | null; + lane_key: string | null; + last_attempt_at: number | null; + last_error: string | null; + metadata_json: string | null; + payload_json: string; + queue_name: string; + received_at: number; + status: string; + updated_at: number; +} + export interface ChannelPairingAllowEntries { account_id: string; channel_key: string; @@ -931,6 +954,7 @@ export interface DB { capture_blobs: CaptureBlobs; capture_events: CaptureEvents; capture_sessions: CaptureSessions; + channel_ingress_events: ChannelIngressEvents; channel_pairing_allow_entries: ChannelPairingAllowEntries; channel_pairing_requests: ChannelPairingRequests; command_log_entries: CommandLogEntries; diff --git a/src/state/openclaw-state-schema.generated.ts b/src/state/openclaw-state-schema.generated.ts index b99f2b34fda0..a70e130fd111 100644 --- a/src/state/openclaw-state-schema.generated.ts +++ b/src/state/openclaw-state-schema.generated.ts @@ -607,6 +607,39 @@ CREATE INDEX IF NOT EXISTS idx_plugin_state_expiry CREATE INDEX IF NOT EXISTS idx_plugin_state_listing ON plugin_state_entries(plugin_id, namespace, created_at, entry_key); +CREATE TABLE IF NOT EXISTS channel_ingress_events ( + queue_name TEXT NOT NULL, + event_id TEXT NOT NULL, + channel_id TEXT NOT NULL, + account_id TEXT NOT NULL, + status TEXT NOT NULL, + lane_key TEXT, + payload_json TEXT NOT NULL, + metadata_json TEXT, + received_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + claim_token TEXT, + claim_owner TEXT, + claimed_at INTEGER, + attempts INTEGER NOT NULL DEFAULT 0, + last_attempt_at INTEGER, + last_error TEXT, + failed_reason TEXT, + failed_at INTEGER, + completed_at INTEGER, + completed_metadata_json TEXT, + PRIMARY KEY (queue_name, event_id) +); + +CREATE INDEX IF NOT EXISTS idx_channel_ingress_pending + ON channel_ingress_events(queue_name, status, received_at, event_id); + +CREATE INDEX IF NOT EXISTS idx_channel_ingress_claims + ON channel_ingress_events(queue_name, status, claimed_at); + +CREATE INDEX IF NOT EXISTS idx_channel_ingress_lane + ON channel_ingress_events(queue_name, status, lane_key); + CREATE TABLE IF NOT EXISTS plugin_blob_entries ( plugin_id TEXT NOT NULL, namespace TEXT NOT NULL, diff --git a/src/state/openclaw-state-schema.sql b/src/state/openclaw-state-schema.sql index 9f9ecbd3206f..adf75eb9e834 100644 --- a/src/state/openclaw-state-schema.sql +++ b/src/state/openclaw-state-schema.sql @@ -602,6 +602,39 @@ CREATE INDEX IF NOT EXISTS idx_plugin_state_expiry CREATE INDEX IF NOT EXISTS idx_plugin_state_listing ON plugin_state_entries(plugin_id, namespace, created_at, entry_key); +CREATE TABLE IF NOT EXISTS channel_ingress_events ( + queue_name TEXT NOT NULL, + event_id TEXT NOT NULL, + channel_id TEXT NOT NULL, + account_id TEXT NOT NULL, + status TEXT NOT NULL, + lane_key TEXT, + payload_json TEXT NOT NULL, + metadata_json TEXT, + received_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + claim_token TEXT, + claim_owner TEXT, + claimed_at INTEGER, + attempts INTEGER NOT NULL DEFAULT 0, + last_attempt_at INTEGER, + last_error TEXT, + failed_reason TEXT, + failed_at INTEGER, + completed_at INTEGER, + completed_metadata_json TEXT, + PRIMARY KEY (queue_name, event_id) +); + +CREATE INDEX IF NOT EXISTS idx_channel_ingress_pending + ON channel_ingress_events(queue_name, status, received_at, event_id); + +CREATE INDEX IF NOT EXISTS idx_channel_ingress_claims + ON channel_ingress_events(queue_name, status, claimed_at); + +CREATE INDEX IF NOT EXISTS idx_channel_ingress_lane + ON channel_ingress_events(queue_name, status, lane_key); + CREATE TABLE IF NOT EXISTS plugin_blob_entries ( plugin_id TEXT NOT NULL, namespace TEXT NOT NULL,