diff --git a/docs/channels/line.md b/docs/channels/line.md index d5c371f1e366..3d5b6a68278f 100644 --- a/docs/channels/line.md +++ b/docs/channels/line.md @@ -42,9 +42,12 @@ openclaw plugins install ./path/to/local/line-plugin https://gateway-host/line/webhook ``` -The Gateway answers LINE's webhook verification (GET) and acknowledges signed -inbound events (POST) immediately after signature and payload validation; agent -processing continues asynchronously. +The Gateway answers LINE's webhook verification (GET). For signed inbound events +(POST), it writes each event to the durable ingress queue before returning `200`; +agent processing continues asynchronously. Failed delivery is retried from the +queue, including after a Gateway restart, and poison events become failed queue +records after bounded retries. If durable persistence fails, the request returns +`500` instead of acknowledging an event that could be lost. If you need a custom path, set `channels.line.webhookPath` or `channels.line.accounts..webhookPath` and update the URL accordingly. diff --git a/extensions/line/src/bot-handlers.test.ts b/extensions/line/src/bot-handlers.test.ts index 6aba9db969f3..83fbb00bc106 100644 --- a/extensions/line/src/bot-handlers.test.ts +++ b/extensions/line/src/bot-handlers.test.ts @@ -5,7 +5,6 @@ import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vites import type { LineAccountConfig } from "./types.js"; type MessageEvent = webhook.MessageEvent; -type PostbackEvent = webhook.PostbackEvent; // Avoid pulling in globals/pairing/media dependencies; this suite only asserts // allowlist/groupPolicy gating and message-context wiring. @@ -176,7 +175,6 @@ vi.mock("./bot-message-context.js", () => ({ })); let handleLineWebhookEvents: typeof import("./bot-handlers.js").handleLineWebhookEvents; -let createLineWebhookReplayCache: typeof import("./bot-handlers.js").createLineWebhookReplayCache; type LineWebhookContext = Parameters[1]; const createRuntime = () => ({ log: vi.fn(), error: vi.fn(), exit: vi.fn() }); @@ -228,7 +226,6 @@ function createLineWebhookTestContext(params: { groupAllowFrom?: LineAccountConfig["groupAllowFrom"]; requireMention?: boolean; groupHistories?: Map; - replayCache?: ReturnType; accessGroups?: Record }>; }): Parameters[1] { const allowFrom = params.allowFrom ?? (params.dmPolicy === "open" ? ["*"] : undefined); @@ -260,22 +257,9 @@ function createLineWebhookTestContext(params: { mediaMaxBytes: 1, processMessage: params.processMessage, ...(params.groupHistories ? { groupHistories: params.groupHistories } : {}), - ...(params.replayCache ? { replayCache: params.replayCache } : {}), }; } -function createOpenGroupReplayContext( - processMessage: LineWebhookContext["processMessage"], - replayCache: ReturnType, -): Parameters[1] { - return createLineWebhookTestContext({ - processMessage, - groupPolicy: "open", - requireMention: false, - replayCache, - }); -} - async function expectGroupMessageBlocked(params: { processMessage: LineWebhookContext["processMessage"]; event: MessageEvent; @@ -300,23 +284,9 @@ async function expectRequireMentionGroupMessageProcessed(event: MessageEvent) { expect(processMessage).toHaveBeenCalledTimes(1); } -async function startInflightReplayDuplicate(params: { - event: MessageEvent; - processMessage: LineWebhookContext["processMessage"]; -}) { - const context = createOpenGroupReplayContext( - params.processMessage, - createLineWebhookReplayCache(), - ); - const firstRun = handleLineWebhookEvents([params.event], context); - await Promise.resolve(); - const secondRun = handleLineWebhookEvents([params.event], context); - return { firstRun, secondRun }; -} - describe("handleLineWebhookEvents", () => { beforeAll(async () => { - ({ handleLineWebhookEvents, createLineWebhookReplayCache } = await import("./bot-handlers.js")); + ({ handleLineWebhookEvents } = await import("./bot-handlers.js")); }); afterAll(() => { @@ -803,173 +773,6 @@ describe("handleLineWebhookEvents", () => { expect(pairingRequest?.accountId).toBe("work"); }); - it("deduplicates replayed webhook events by webhookEventId before processing", async () => { - const processMessage = vi.fn(); - const event = createReplayMessageEvent({ - messageId: "m-replay", - groupId: "group-replay", - userId: "user-replay", - webhookEventId: "evt-replay-1", - isRedelivery: true, - }); - const context = createOpenGroupReplayContext(processMessage, createLineWebhookReplayCache()); - - await handleLineWebhookEvents([event], context); - await handleLineWebhookEvents([event], context); - - expect(buildLineMessageContextMock).toHaveBeenCalledTimes(1); - expect(processMessage).toHaveBeenCalledTimes(1); - }); - - it("skips concurrent redeliveries while the first event is still processing", async () => { - let resolveFirst: (() => void) | undefined; - const firstDone = new Promise((resolve) => { - resolveFirst = resolve; - }); - const processMessage = vi.fn(async () => { - await firstDone; - }); - const event = createReplayMessageEvent({ - messageId: "m-inflight", - groupId: "group-inflight", - userId: "user-inflight", - webhookEventId: "evt-inflight-1", - isRedelivery: true, - }); - const { firstRun, secondRun } = await startInflightReplayDuplicate({ event, processMessage }); - resolveFirst?.(); - await Promise.all([firstRun, secondRun]); - - expect(buildLineMessageContextMock).toHaveBeenCalledTimes(1); - expect(processMessage).toHaveBeenCalledTimes(1); - }); - - it("commits in-flight failures so concurrent duplicates do not retry", async () => { - let rejectFirst: ((err: Error) => void) | undefined; - const firstDone = new Promise((_, reject) => { - rejectFirst = reject; - }); - const processMessage = vi.fn(async () => { - await firstDone; - }); - const event = createReplayMessageEvent({ - messageId: "m-inflight-fail", - groupId: "group-inflight", - userId: "user-inflight", - webhookEventId: "evt-inflight-fail-1", - isRedelivery: true, - }); - const { firstRun, secondRun } = await startInflightReplayDuplicate({ event, processMessage }); - const firstFailure = expect(firstRun).rejects.toThrow("transient inflight failure"); - rejectFirst?.(new Error("transient inflight failure")); - - await firstFailure; - await expect(secondRun).resolves.toBeUndefined(); - expect(processMessage).toHaveBeenCalledTimes(1); - }); - - it("deduplicates redeliveries by LINE message id when webhookEventId changes", async () => { - const processMessage = vi.fn(); - const event = { - type: "message", - message: { id: "m-dup-1", type: "text", text: "hello" }, - replyToken: "reply-token", - timestamp: Date.now(), - source: { type: "group", groupId: "group-dup", userId: "user-dup" }, - mode: "active", - webhookEventId: "evt-dup-1", - deliveryContext: { isRedelivery: false }, - } as MessageEvent; - - const context: Parameters[1] = { - cfg: { - channels: { line: { groupPolicy: "allowlist", groupAllowFrom: ["user-dup"] } }, - }, - account: { - accountId: "default", - enabled: true, - channelAccessToken: "token", - channelSecret: "secret", - tokenSource: "config", - config: { - groupPolicy: "allowlist", - groupAllowFrom: ["user-dup"], - groups: { "*": { requireMention: false } }, - }, - }, - runtime: createRuntime(), - mediaMaxBytes: 1, - processMessage, - replayCache: createLineWebhookReplayCache(), - }; - - await handleLineWebhookEvents([event], context); - await handleLineWebhookEvents( - [ - { - ...event, - webhookEventId: "evt-dup-redelivery", - deliveryContext: { isRedelivery: true }, - } as MessageEvent, - ], - context, - ); - - expect(buildLineMessageContextMock).toHaveBeenCalledTimes(1); - expect(processMessage).toHaveBeenCalledTimes(1); - }); - - it("deduplicates postback redeliveries by webhookEventId when replyToken changes", async () => { - const processMessage = vi.fn(); - buildLinePostbackContextMock.mockResolvedValue({ - ctxPayload: { From: "line:user:user-postback" }, - route: { agentId: "default" }, - isGroup: false, - accountId: "default", - }); - const event = { - type: "postback", - postback: { data: "action=confirm" }, - replyToken: "reply-token-1", - timestamp: Date.now(), - source: { type: "user", userId: "user-postback" }, - mode: "active", - webhookEventId: "evt-postback-1", - deliveryContext: { isRedelivery: false }, - } as PostbackEvent; - - const context: Parameters[1] = { - cfg: { channels: { line: { dmPolicy: "open", allowFrom: ["*"] } } }, - account: { - accountId: "default", - enabled: true, - channelAccessToken: "token", - channelSecret: "secret", - tokenSource: "config", - config: { dmPolicy: "open", allowFrom: ["*"] }, - }, - runtime: createRuntime(), - mediaMaxBytes: 1, - processMessage, - replayCache: createLineWebhookReplayCache(), - }; - - await handleLineWebhookEvents([event], context); - await handleLineWebhookEvents( - [ - { - ...event, - replyToken: "reply-token-2", - deliveryContext: { isRedelivery: true }, - } as PostbackEvent, - ], - context, - ); - - expect(buildLinePostbackContextMock).toHaveBeenCalledTimes(1); - expect(processMessage).toHaveBeenCalledTimes(1); - }); - it("skips group messages by default when requireMention is not configured", async () => { const processMessage = vi.fn(); const event = createTestMessageEvent({ @@ -1508,29 +1311,5 @@ describe("handleLineWebhookEvents", () => { // Should be skipped because there is a non-bot mention and the bot was not mentioned. expect(processMessage).not.toHaveBeenCalled(); }); - - it("keeps replay cache committed after a non-retryable event failure", async () => { - const processMessage = vi - .fn() - .mockRejectedValueOnce(new Error("transient failure")) - .mockResolvedValueOnce(undefined); - const event = createReplayMessageEvent({ - messageId: "m-fail-then-retry", - groupId: "group-retry", - userId: "user-retry", - webhookEventId: "evt-fail-then-retry", - isRedelivery: false, - }); - const context = createOpenGroupReplayContext(processMessage, createLineWebhookReplayCache()); - - await expect(handleLineWebhookEvents([event], context)).rejects.toThrow("transient failure"); - await handleLineWebhookEvents([event], context); - - expect(buildLineMessageContextMock).toHaveBeenCalledTimes(1); - expect(processMessage).toHaveBeenCalledTimes(1); - expect(context.runtime.error).toHaveBeenCalledWith( - "line: event handler failed: Error: transient failure", - ); - }); }); /* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/extensions/line/src/bot-handlers.ts b/extensions/line/src/bot-handlers.ts index 172f6bb38cae..af12e630a7ec 100644 --- a/extensions/line/src/bot-handlers.ts +++ b/extensions/line/src/bot-handlers.ts @@ -10,7 +10,6 @@ import { resolvePairingIdLabel, upsertChannelPairingRequest, } from "openclaw/plugin-sdk/conversation-runtime"; -import { createChannelReplayGuard } from "openclaw/plugin-sdk/persistent-dedupe"; import { DEFAULT_GROUP_HISTORY_LIMIT, createChannelHistoryWindow, @@ -72,65 +71,20 @@ interface LineHandlerContext { account: ResolvedLineAccount; runtime: RuntimeEnv; mediaMaxBytes: number; - processMessage: (ctx: LineInboundContext) => Promise; - replayCache?: LineWebhookReplayCache; + processMessage: ( + ctx: LineInboundContext, + control: { abortSignal?: AbortSignal; onTurnAdopted?: () => Promise }, + ) => Promise; + abortSignal?: AbortSignal; + onTurnAdopted?: () => Promise; groupHistories?: Map; historyLimit?: number; } -const LINE_WEBHOOK_REPLAY_WINDOW_MS = 10 * 60 * 1000; -const LINE_WEBHOOK_REPLAY_MAX_ENTRIES = 4096; - function normalizeLineIngressEntry(value: string): string | null { return normalizeLineAllowEntry(value) || null; } -type LineReplayEvent = { event: WebhookEvent; accountId: string }; - -export function createLineWebhookReplayCache() { - return createChannelReplayGuard({ - dedupe: { - ttlMs: LINE_WEBHOOK_REPLAY_WINDOW_MS, - memoryMaxSize: LINE_WEBHOOK_REPLAY_MAX_ENTRIES, - }, - buildReplayKey: ({ event, accountId }) => buildLineWebhookReplayKey(event, accountId)?.key, - }); -} - -type LineWebhookReplayCache = ReturnType; - -function buildLineWebhookReplayKey( - event: WebhookEvent, - accountId: string, -): { key: string; eventId: string } | null { - if (event.type === "message") { - const messageId = event.message?.id?.trim(); - if (messageId) { - return { - key: `${accountId}|message:${messageId}`, - eventId: `message:${messageId}`, - }; - } - } - const eventId = (event as { webhookEventId?: string }).webhookEventId?.trim(); - if (!eventId) { - return null; - } - - const source = ( - event as { - source?: { type?: string; userId?: string; groupId?: string; roomId?: string }; - } - ).source; - const sourceId = - source?.type === "group" - ? `group:${source.groupId ?? ""}` - : source?.type === "room" - ? `room:${source.roomId ?? ""}` - : `user:${source?.userId ?? ""}`; - return { key: `${accountId}|${event.type}|${sourceId}|${eventId}`, eventId: `event:${eventId}` }; -} - function resolveLineGroupConfig(params: { config: ResolvedLineAccount["config"]; groupId?: string; @@ -485,7 +439,10 @@ async function handleMessageEvent(event: MessageEvent, context: LineHandlerConte return; } - await processMessage(messageContext); + await processMessage(messageContext, { + ...(context.abortSignal ? { abortSignal: context.abortSignal } : {}), + ...(context.onTurnAdopted ? { onTurnAdopted: context.onTurnAdopted } : {}), + }); historyReservation.commit(); } finally { historyReservation.release(); @@ -537,7 +494,10 @@ async function handlePostbackEvent( return; } - await context.processMessage(postbackContext); + await context.processMessage(postbackContext, { + ...(context.abortSignal ? { abortSignal: context.abortSignal } : {}), + ...(context.onTurnAdopted ? { onTurnAdopted: context.onTurnAdopted } : {}), + }); } export async function handleLineWebhookEvents( @@ -547,28 +507,7 @@ export async function handleLineWebhookEvents( let firstError: unknown; for (const event of events) { try { - if (!context.replayCache) { - await handleLineWebhookEvent(event, context); - continue; - } - const replayEvent = { event, accountId: context.account.accountId }; - const result = await context.replayCache.processGuarded( - replayEvent, - async () => await handleLineWebhookEvent(event, context), - { onError: "commit" }, - ); - const replayId = buildLineWebhookReplayKey(event, context.account.accountId)?.eventId; - if (result.kind === "inflight") { - logVerbose(`line: skipped in-flight replayed webhook event ${replayId ?? "unknown"}`); - try { - await result.pending; - } catch (err) { - context.runtime.error?.(danger(`line: replayed in-flight event failed: ${String(err)}`)); - firstError ??= err; - } - } else if (result.kind === "duplicate") { - logVerbose(`line: skipped replayed webhook event ${replayId ?? "unknown"}`); - } + await handleLineWebhookEvent(event, context); } catch (err) { context.runtime.error?.(danger(`line: event handler failed: ${String(err)}`)); firstError ??= err; diff --git a/extensions/line/src/bot.ts b/extensions/line/src/bot.ts index 18d2e2c4178d..842d618f5a7e 100644 --- a/extensions/line/src/bot.ts +++ b/extensions/line/src/bot.ts @@ -9,9 +9,10 @@ import { type RuntimeEnv, } from "openclaw/plugin-sdk/runtime-env"; import { resolveLineAccount } from "./accounts.js"; -import { createLineWebhookReplayCache, handleLineWebhookEvents } from "./bot-handlers.js"; +import { handleLineWebhookEvents } from "./bot-handlers.js"; import type { LineInboundContext } from "./bot-message-context.js"; import type { ResolvedLineAccount } from "./types.js"; +import { createLineWebhookSpool } from "./webhook-spool.js"; interface LineBotOptions { channelAccessToken: string; @@ -20,12 +21,16 @@ interface LineBotOptions { runtime?: RuntimeEnv; config?: OpenClawConfig; mediaMaxMb?: number; - onMessage?: (ctx: LineInboundContext) => Promise; + onMessage?: ( + ctx: LineInboundContext, + control: { abortSignal?: AbortSignal; onTurnAdopted?: () => Promise }, + ) => Promise; } interface LineBot { handleWebhook: (body: webhook.CallbackRequest) => Promise; account: ResolvedLineAccount; + stop: () => Promise; } export function createLineBot(opts: LineBotOptions): LineBot { @@ -44,28 +49,28 @@ export function createLineBot(opts: LineBotOptions): LineBot { (async () => { logVerbose("line: no message handler configured"); }); - const replayCache = createLineWebhookReplayCache(); const groupHistories = new Map(); - - const handleWebhook = async (body: webhook.CallbackRequest): Promise => { - if (!body.events || body.events.length === 0) { - return; - } - - await handleLineWebhookEvents(body.events, { - cfg, - account, - runtime, - mediaMaxBytes, - processMessage, - replayCache, - groupHistories, - historyLimit: cfg.messages?.groupChat?.historyLimit ?? DEFAULT_GROUP_HISTORY_LIMIT, - }); - }; + const spool = createLineWebhookSpool({ + accountId: account.accountId, + runtime, + deliver: async (event, _destination, control) => + await handleLineWebhookEvents([event], { + cfg, + account, + runtime, + mediaMaxBytes, + processMessage, + ...(control.abortSignal ? { abortSignal: control.abortSignal } : {}), + ...(control.onTurnAdopted ? { onTurnAdopted: control.onTurnAdopted } : {}), + groupHistories, + historyLimit: cfg.messages?.groupChat?.historyLimit ?? DEFAULT_GROUP_HISTORY_LIMIT, + }), + }); + spool.start(); return { - handleWebhook, + handleWebhook: spool.accept, account, + stop: spool.stop, }; } diff --git a/extensions/line/src/monitor.lifecycle.test.ts b/extensions/line/src/monitor.lifecycle.test.ts index 9f4a56770272..6c455b230154 100644 --- a/extensions/line/src/monitor.lifecycle.test.ts +++ b/extensions/line/src/monitor.lifecycle.test.ts @@ -21,6 +21,7 @@ const { createLineBotMock: vi.fn(() => ({ account: { accountId: "default" }, handleWebhook: vi.fn(), + stop: vi.fn(), })), createLineNodeWebhookHandlerMock: vi.fn<() => LineNodeWebhookHandler>(() => vi.fn(async () => {}), @@ -176,6 +177,7 @@ describe("monitorLineProvider lifecycle", () => { createLineBotMock.mockImplementation(() => ({ account: { accountId: "default" }, handleWebhook: vi.fn(), + stop: vi.fn(), })); // Clear call history only; the implementation was wired to the actual // helper once in the module mock factory. @@ -266,7 +268,7 @@ describe("monitorLineProvider lifecycle", () => { expect(registration.route.pluginId).toBe("line"); expect(registration.route).not.toHaveProperty("path"); expect(registration.route).not.toHaveProperty("replaceExisting"); - monitor.stop(); + await monitor.stop(); }); it("stops immediately when signal is already aborted", async () => { @@ -293,8 +295,8 @@ describe("monitorLineProvider lifecycle", () => { }); expect(unregisterHttpMock).not.toHaveBeenCalled(); - monitor.stop(); - monitor.stop(); + await monitor.stop(); + await monitor.stop(); expect(unregisterHttpMock).toHaveBeenCalledTimes(1); }); @@ -322,7 +324,7 @@ describe("monitorLineProvider lifecycle", () => { expect(registration.target.accountId).toBe("work"); expect(registration.route.accountId).toBe("work"); - monitor.stop(); + await monitor.stop(); }); it("does not register a webhook when bot startup fails", async () => { @@ -380,8 +382,8 @@ describe("monitorLineProvider lifecycle", () => { expect(firstBot.handleWebhook).not.toHaveBeenCalled(); expect(secondBot.handleWebhook).toHaveBeenCalledTimes(1); - firstMonitor.stop(); - secondMonitor.stop(); + await firstMonitor.stop(); + await secondMonitor.stop(); }); it("dispatches a signed POST to a configured trailing-slash webhook path", async () => { @@ -414,10 +416,10 @@ describe("monitorLineProvider lifecycle", () => { expect(res.statusCode).toBe(200); expect(bot.handleWebhook).toHaveBeenCalledTimes(1); - monitor.stop(); + await monitor.stop(); }); - it("runs matched event processing on a detached admitted work root", async () => { + it("durably admits matched events before acknowledging", async () => { const monitor = await monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret @@ -441,16 +443,13 @@ describe("monitorLineProvider lifecycle", () => { handleWebhook: ReturnType; }; expect(res.statusCode).toBe(200); - // The request admission is released once the route handler returns, so - // event processing dispatched on the inherited chain would be refused as - // draining; the dispatch must reserve its own root. - expect(runDetachedWebhookWorkMock).toHaveBeenCalledTimes(1); + expect(runDetachedWebhookWorkMock).not.toHaveBeenCalled(); expect(bot.handleWebhook).toHaveBeenCalledTimes(1); - monitor.stop(); + await monitor.stop(); }); - it("acknowledges shared-path POST requests before matched event processing completes", async () => { + it("waits for shared-path durable admission before acknowledging", async () => { const monitor = await monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret @@ -479,16 +478,20 @@ describe("monitorLineProvider lifecycle", () => { }) as unknown as IncomingMessage; const res = createRouteResponse(); - await route.handler(req, res); - - expect(res.statusCode).toBe(200); - expect(res.headersSent).toBe(true); + const request = route.handler(req, res); + await vi.waitFor(() => { + expect(bot.handleWebhook).toHaveBeenCalledTimes(1); + }); + expect(res.headersSent).toBe(false); expect(bot.handleWebhook).toHaveBeenCalledTimes(1); if (!releaseWebhook) { throw new Error("expected pending LINE webhook handler"); } releaseWebhook(); - monitor.stop(); + await request; + expect(res.statusCode).toBe(200); + expect(res.headersSent).toBe(true); + await monitor.stop(); }); it("rejects ambiguous shared-path webhook signatures", async () => { @@ -530,8 +533,8 @@ describe("monitorLineProvider lifecycle", () => { expect(firstBot.handleWebhook).not.toHaveBeenCalled(); expect(secondBot.handleWebhook).not.toHaveBeenCalled(); - firstMonitor.stop(); - secondMonitor.stop(); + await firstMonitor.stop(); + await secondMonitor.stop(); }); it("rejects webhook requests above the shared in-flight limit before body handling", async () => { @@ -589,6 +592,6 @@ describe("monitorLineProvider lifecycle", () => { heldRequests.splice(0).forEach((req) => req.destroy()); await Promise.allSettled(firstRequests); - monitor.stop(); + await monitor.stop(); }); }); diff --git a/extensions/line/src/monitor.ts b/extensions/line/src/monitor.ts index 93f98a823f60..b4532b85f345 100644 --- a/extensions/line/src/monitor.ts +++ b/extensions/line/src/monitor.ts @@ -20,7 +20,6 @@ import { import { beginWebhookRequestPipelineOrReject, createWebhookInFlightLimiter, - runDetachedWebhookWork, } from "openclaw/plugin-sdk/webhook-request-guards"; import { resolveDefaultLineAccountId } from "./accounts.js"; import { deliverLineAutoReply } from "./auto-reply-delivery.js"; @@ -45,6 +44,7 @@ import { import { buildTemplateMessageFromPayload } from "./template-messages.js"; import type { LineChannelData, ResolvedLineAccount } from "./types.js"; import { createLineNodeWebhookHandler, readLineWebhookRequestBody } from "./webhook-node.js"; +import { LineWebhookTerminalDeliveryError } from "./webhook-spool.js"; import { parseLineWebhookBody, validateLineSignature } from "./webhook-utils.js"; interface MonitorLineProviderOptions { @@ -61,7 +61,7 @@ interface MonitorLineProviderOptions { interface LineProviderMonitor { account: ResolvedLineAccount; handleWebhook: (body: webhook.CallbackRequest) => Promise; - stop: () => void; + stop: () => Promise; } const lineWebhookInFlightLimiter = createWebhookInFlightLimiter(); @@ -141,7 +141,7 @@ export async function monitorLineProvider( accountId, runtime, config, - onMessage: async (ctx) => { + onMessage: async (ctx, deliveryControl) => { if (!ctx) { return; } @@ -164,15 +164,24 @@ export async function monitorLineProvider( const displayName = await displayNamePromise; logVerbose(`line: received message from ${displayName} (${ctxPayload.From})`); + let replyTokenUsed = false; + let turnAdopted = false; try { const textLimit = 5000; - let replyTokenUsed = false; const core = getLineRuntime(); const turnResult = await core.channel.inbound.run({ channel: "line", accountId: route.accountId, raw: ctx, + turnAdoptionLifecycle: { + admission: "exclusive", + onAdopted: async () => { + await deliveryControl.onTurnAdopted?.(); + turnAdopted = true; + }, + ...(deliveryControl.abortSignal ? { abortSignal: deliveryControl.abortSignal } : {}), + }, adapter: { ingest: () => ({ id: ctxPayload.MessageSid ?? `${ctxPayload.From}:${Date.now()}`, @@ -191,6 +200,9 @@ export async function monitorLineProvider( core.channel.reply.dispatchReplyWithBufferedBlockDispatcher, record: ctx.turn.record, replyPipeline: {}, + ...(deliveryControl.abortSignal + ? { replyOptions: { abortSignal: deliveryControl.abortSignal } } + : {}), delivery: { durable: (payload, info) => resolveLineDurableReplyOptions({ @@ -265,18 +277,13 @@ export async function monitorLineProvider( } } catch (err) { runtime.error?.(danger(`line: auto-reply failed: ${String(err)}`)); - - if (replyToken) { - try { - await replyMessageLine( - replyToken, - [{ type: "text", text: "Sorry, I encountered an error processing your message." }], - { cfg: config, accountId: ctx.accountId }, - ); - } catch (replyErr) { - runtime.error?.(danger(`line: error reply failed: ${String(replyErr)}`)); - } + if (turnAdopted || replyTokenUsed) { + throw new LineWebhookTerminalDeliveryError( + "LINE delivery failed after consuming the event reply token.", + { cause: err }, + ); } + throw err; } finally { stopLoading?.(); } @@ -377,23 +384,13 @@ export async function monitorLineProvider( return; } - requestLifecycle.release(); + if (body.events && body.events.length > 0) { + logVerbose(`line: received ${body.events.length} webhook events`); + await match.target.bot.handleWebhook(body); + } res.statusCode = 200; res.setHeader("Content-Type", "application/json"); res.end(JSON.stringify({ status: "ok" })); - - if (body.events && body.events.length > 0) { - logVerbose(`line: received ${body.events.length} webhook events`); - // Detach event processing from the request admission before the ack - // releases it; an inherited released admission refuses queue work. - void runDetachedWebhookWork(() => match.target.bot.handleWebhook(body)).catch( - (err: unknown) => { - match.target.runtime.error?.( - danger(`line webhook dispatch failed: ${String(err)}`), - ); - }, - ); - } } catch (err) { if (isRequestBodyLimitError(err, "PAYLOAD_TOO_LARGE")) { res.statusCode = 413; @@ -423,28 +420,36 @@ export async function monitorLineProvider( logVerbose(`line: registered webhook handler at ${normalizedPath}`); let stopped = false; - const stopHandler = () => { + let stopPromise: Promise | undefined; + const stopHandler = (): Promise => { + if (stopPromise) { + return stopPromise; + } if (stopped) { - return; + return Promise.resolve(); } stopped = true; logVerbose(`line: stopping provider for account ${resolvedAccountId}`); unregisterHttp(); + stopPromise = bot.stop(); + return stopPromise; }; + const stopOnAbort = () => void stopHandler(); if (abortSignal?.aborted) { - stopHandler(); + await stopHandler(); } else if (abortSignal) { - abortSignal.addEventListener("abort", stopHandler, { once: true }); + abortSignal.addEventListener("abort", stopOnAbort, { once: true }); await waitForAbortSignal(abortSignal); + await stopHandler(); } return { account: bot.account, handleWebhook: bot.handleWebhook, - stop: () => { - stopHandler(); - abortSignal?.removeEventListener("abort", stopHandler); + stop: async () => { + await stopHandler(); + abortSignal?.removeEventListener("abort", stopOnAbort); }, }; } diff --git a/extensions/line/src/webhook-node.test.ts b/extensions/line/src/webhook-node.test.ts index ffc9e5b24916..24832007ed71 100644 --- a/extensions/line/src/webhook-node.test.ts +++ b/extensions/line/src/webhook-node.test.ts @@ -329,21 +329,29 @@ describe("LINE webhook shared POST contract", () => { }, ); + it.each(sharedWebhookPostContractCases)("$name dispatches signed events", async ({ invoke }) => { + const result = await invoke({ + rawBody: JSON.stringify({ events: [{ type: "message" }] }), + signed: true, + }); + + expect(result.status).toBe(200); + expect(result.body).toEqual({ status: "ok" }); + expect(result.dispatched).toHaveBeenCalledTimes(1); + }); + it.each(sharedWebhookPostContractCases)( - "$name acknowledges signed events before failed background processing is logged", + "$name returns 500 when durable admission fails", async ({ invoke }) => { const result = await invoke({ - failWith: new Error("transient failure"), + failWith: new Error("persist failed"), rawBody: JSON.stringify({ events: [{ type: "message" }] }), signed: true, }); - expect(result.status).toBe(200); - expect(result.body).toEqual({ status: "ok" }); - expect(result.dispatched).toHaveBeenCalledTimes(1); - await vi.waitFor(() => { - expect(result.runtimeError).toHaveBeenCalledTimes(1); - }); + expect(result.status).toBe(500); + expect(result.body).toEqual({ error: "Internal server error" }); + expect(result.runtimeError).toHaveBeenCalledTimes(1); }, ); }); @@ -413,7 +421,7 @@ describe("createLineNodeWebhookHandler", () => { expect(bot.handleWebhook).not.toHaveBeenCalled(); }); - it("dispatches signed POST event processing through the detached admitted work root", async () => { + it("durably admits signed POST events before acknowledging", async () => { runDetachedWebhookWorkSpy.mockClear(); const rawBody = JSON.stringify({ events: [{ type: "message" }] }); const { bot, handler, secret } = createPostWebhookTestHarness(rawBody); @@ -422,9 +430,7 @@ describe("createLineNodeWebhookHandler", () => { await runSignedPost({ handler, rawBody, secret, res }); expect(res.statusCode).toBe(200); - // The request admission is released once the handler returns; the inherited - // chain would be refused as draining, so dispatch must reserve its own root. - expect(runDetachedWebhookWorkSpy).toHaveBeenCalledTimes(1); + expect(runDetachedWebhookWorkSpy).not.toHaveBeenCalled(); expect(bot.handleWebhook).toHaveBeenCalledTimes(1); }); @@ -480,7 +486,7 @@ describe("createLineNodeWebhookHandler", () => { expect(payload.events).toEqual([{ type: "message" }]); }); - it("acknowledges signed event requests before event processing completes", async () => { + it("waits for durable admission before acknowledging signed event requests", async () => { const rawBody = JSON.stringify({ events: [{ type: "message" }] }); let releaseAuthenticated: (() => void) | undefined; const bot = { @@ -509,14 +515,14 @@ describe("createLineNodeWebhookHandler", () => { expect(bot.handleWebhook).toHaveBeenCalledTimes(1); }); - await request; - - expect(res.statusCode).toBe(200); - expect(res.headersSent).toBe(true); + expect(res.headersSent).toBe(false); if (!releaseAuthenticated) { throw new Error("Expected LINE authenticated request release callback to be initialized"); } releaseAuthenticated(); + await request; + expect(res.statusCode).toBe(200); + expect(res.headersSent).toBe(true); }); it("returns 400 for invalid JSON payload even when signature is valid", async () => { @@ -560,13 +566,13 @@ describe("createLineWebhookMiddleware", () => { expect(payload.events).toEqual(expectedEvents); }); - it("dispatches middleware event processing through the detached admitted work root", async () => { + it("waits for middleware event admission before acknowledging", async () => { runDetachedWebhookWorkSpy.mockClear(); const { res, onEvents } = await invokeWebhook({ body: JSON.stringify({ events: [{ type: "message" }] }), }); expect(res.status).toHaveBeenCalledWith(200); - expect(runDetachedWebhookWorkSpy).toHaveBeenCalledTimes(1); + expect(runDetachedWebhookWorkSpy).not.toHaveBeenCalled(); expect(onEvents).toHaveBeenCalledTimes(1); }); diff --git a/extensions/line/src/webhook-node.ts b/extensions/line/src/webhook-node.ts index ef45887729a1..35881291371d 100644 --- a/extensions/line/src/webhook-node.ts +++ b/extensions/line/src/webhook-node.ts @@ -1,16 +1,11 @@ // Line plugin module implements webhook node behavior. import type { IncomingMessage, ServerResponse } from "node:http"; import type { webhook } from "@line/bot-sdk"; -import { - createMessageReceiveContext, - type MessageReceiveContext, -} from "openclaw/plugin-sdk/channel-outbound"; import { danger, logVerbose, type RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; import { isRequestBodyLimitError, readRequestBodyWithLimit, requestBodyErrorToText, - runDetachedWebhookWork, } from "openclaw/plugin-sdk/webhook-request-guards"; import { parseLineWebhookBody, validateLineSignature } from "./webhook-utils.js"; @@ -31,10 +26,6 @@ export async function readLineWebhookRequestBody( type ReadBodyFn = (req: IncomingMessage, maxBytes: number, timeoutMs?: number) => Promise; -function logLineWebhookDispatchError(runtime: RuntimeEnv | undefined, err: unknown): void { - runtime?.error?.(danger(`line webhook dispatch failed: ${String(err)}`)); -} - export function createLineNodeWebhookHandler(params: { channelSecret: string; bot: { handleWebhook: (body: webhook.CallbackRequest) => Promise }; @@ -67,7 +58,6 @@ export function createLineNodeWebhookHandler(params: { return; } - let receiveContext: MessageReceiveContext | undefined; try { const signatureHeader = req.headers["x-line-signature"]; const signature = @@ -109,33 +99,14 @@ export function createLineNodeWebhookHandler(params: { } params.onRequestAuthenticated?.(); - - receiveContext = createMessageReceiveContext({ - id: `${Date.now()}:line:webhook`, - channel: "line", - message: body, - ackPolicy: "after_receive_record", - onAck: () => { - res.statusCode = 200; - res.setHeader("Content-Type", "application/json"); - res.end(JSON.stringify({ status: "ok" })); - }, - }); - - if (receiveContext.shouldAckAfter("receive_record")) { - await receiveContext.ack(); - } - if (body.events && body.events.length > 0) { logVerbose(`line: received ${body.events.length} webhook events`); - // Detach event processing from the request admission before the ack - // releases it; an inherited released admission refuses queue work. - void runDetachedWebhookWork(() => params.bot.handleWebhook(body)).catch((err: unknown) => - logLineWebhookDispatchError(params.runtime, err), - ); + await params.bot.handleWebhook(body); } + res.statusCode = 200; + res.setHeader("Content-Type", "application/json"); + res.end(JSON.stringify({ status: "ok" })); } catch (err) { - await receiveContext?.nack(err); if (isRequestBodyLimitError(err, "PAYLOAD_TOO_LARGE")) { res.statusCode = 413; res.setHeader("Content-Type", "application/json"); diff --git a/extensions/line/src/webhook-spool.test.ts b/extensions/line/src/webhook-spool.test.ts new file mode 100644 index 000000000000..140c7573e488 --- /dev/null +++ b/extensions/line/src/webhook-spool.test.ts @@ -0,0 +1,656 @@ +// Line tests cover durable webhook admission, replay, and dead-lettering. +import crypto from "node:crypto"; +import fs from "node:fs/promises"; +import type { IncomingMessage, ServerResponse } from "node:http"; +import os from "node:os"; +import path from "node:path"; +import type { webhook } from "@line/bot-sdk"; +import { + closeOpenClawStateDatabaseForTest, + createChannelIngressQueueForTests as createChannelIngressQueue, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createLineNodeWebhookHandler } from "./webhook-node.js"; +import { createLineWebhookSpool, LineWebhookTerminalDeliveryError } from "./webhook-spool.js"; + +type LineWebhookDeliveryOutcome = Parameters< + NonNullable[0]["onOutcome"]> +>[0]; + +type SpoolPayload = { + version: number; + destination: string; + event: webhook.Event; +}; + +const runtime = (): RuntimeEnv => ({ error: vi.fn(), exit: vi.fn(), log: vi.fn() }); + +function createEvent(eventId: string, userId = "user-1"): webhook.Event { + return { + type: "message", + message: { id: `message-${eventId}`, type: "text", text: "hello" }, + replyToken: "test-auth-token", + timestamp: Date.now(), + source: { type: "user", userId }, + mode: "active", + webhookEventId: eventId, + deliveryContext: { isRedelivery: false }, + } as webhook.MessageEvent; +} + +function callback(event: webhook.Event): webhook.CallbackRequest { + return { destination: "destination-1", events: [event] }; +} + +async function withQueue( + fn: (queue: ReturnType>) => Promise, +): Promise { + const stateDir = path.join(os.tmpdir(), `openclaw-line-spool-${crypto.randomUUID()}`); + await fs.mkdir(stateDir); + const queue = createChannelIngressQueue({ + channelId: "line", + accountId: "default", + stateDir, + }); + try { + return await fn(queue); + } finally { + closeOpenClawStateDatabaseForTest(); + await fs.rm(stateDir, { recursive: true, force: true }); + } +} + +async function waitForOutcome( + outcomes: LineWebhookDeliveryOutcome[], + kind: LineWebhookDeliveryOutcome["kind"], +): Promise { + await vi.waitFor(() => { + expect(outcomes.some((outcome) => outcome.kind === kind)).toBe(true); + }); + const outcome = outcomes.find((candidate) => candidate.kind === kind); + if (!outcome) { + throw new Error(`Expected LINE webhook spool outcome ${kind}`); + } + return outcome; +} + +function createResponse(): ServerResponse & { body?: string } { + const response = { + statusCode: 0, + headersSent: false, + setHeader: vi.fn(), + end: vi.fn((body?: string) => { + response.headersSent = true; + response.body = body; + }), + body: undefined as string | undefined, + }; + return response as unknown as ServerResponse & { body?: string }; +} + +describe("LINE webhook spool", () => { + afterEach(() => { + closeOpenClawStateDatabaseForTest(); + }); + + it("acknowledges only after durable persistence even when delivery fails", async () => { + await withQueue(async (queue) => { + const outcomes: LineWebhookDeliveryOutcome[] = []; + const spool = createLineWebhookSpool({ + accountId: "default", + runtime: runtime(), + queue, + maxAttempts: 1, + retryBaseMs: 0, + deliver: async () => { + throw new Error("dispatch failed"); + }, + onOutcome: (outcome) => outcomes.push(outcome), + }); + spool.start(); + const body = JSON.stringify(callback(createEvent("event-ack"))); + const channelSecret = "test-auth-token"; + const handler = createLineNodeWebhookHandler({ + channelSecret, + bot: { handleWebhook: spool.accept }, + runtime: runtime(), + readBody: async () => body, + }); + const response = createResponse(); + + await handler( + { + method: "POST", + headers: { + "x-line-signature": crypto + .createHmac("SHA256", channelSecret) + .update(body) + .digest("base64"), + }, + } as unknown as IncomingMessage, + response, + ); + + expect(response.statusCode).toBe(200); + expect(response.body).toBe(JSON.stringify({ status: "ok" })); + const outcome = await waitForOutcome(outcomes, "dead-lettered"); + expect(outcome).toMatchObject({ + eventId: "event-ack", + attempt: 1, + reason: "retry-limit-exceeded", + }); + expect((await queue.enqueue("event-ack", {} as SpoolPayload)).kind).toBe("failed"); + await spool.stop(); + }); + }); + + it("recovers an in-flight event after restart and delivers it", async () => { + await withQueue(async (queue) => { + const event = createEvent("event-restart"); + await queue.enqueue( + "event-restart", + { version: 1, destination: "destination-1", event }, + { laneKey: "user:user-1" }, + ); + expect(await queue.claim("event-restart", { ownerId: "dead-gateway" })).not.toBeNull(); + const delivered = vi.fn(async () => {}); + const outcomes: LineWebhookDeliveryOutcome[] = []; + const restartedSpool = createLineWebhookSpool({ + accountId: "default", + runtime: runtime(), + queue, + claimStaleMs: 25, + retryBaseMs: 0, + deliver: delivered, + onOutcome: (outcome) => outcomes.push(outcome), + }); + restartedSpool.start(); + + await waitForOutcome(outcomes, "completed"); + expect(delivered).toHaveBeenCalledTimes(1); + expect((await queue.enqueue("event-restart", {} as SpoolPayload)).kind).toBe("completed"); + await restartedSpool.stop(); + }); + }); + + it("does not reclaim a fresh claim held by another live spool", async () => { + await withQueue(async (queue) => { + const event = createEvent("event-live-owner"); + await queue.enqueue( + "event-live-owner", + { version: 1, destination: "destination-1", event }, + { laneKey: "user:user-1" }, + ); + expect(await queue.claim("event-live-owner", { ownerId: "live-gateway" })).not.toBeNull(); + const deliver = vi.fn(async () => {}); + const spool = createLineWebhookSpool({ + accountId: "default", + runtime: runtime(), + queue, + claimStaleMs: 60_000, + deliver, + }); + spool.start(); + + await vi.waitFor(async () => { + expect(await queue.listClaims()).toHaveLength(1); + }); + expect(deliver).not.toHaveBeenCalled(); + await spool.stop(); + }); + }); + + it("bounds poison-event retries and records a typed dead letter", async () => { + await withQueue(async (queue) => { + const deliver = vi.fn(async () => { + throw new Error("poison"); + }); + const outcomes: LineWebhookDeliveryOutcome[] = []; + const spool = createLineWebhookSpool({ + accountId: "default", + runtime: runtime(), + queue, + maxAttempts: 3, + retryBaseMs: 0, + retryMaxMs: 0, + deliver, + onOutcome: (outcome) => outcomes.push(outcome), + }); + spool.start(); + await spool.accept(callback(createEvent("event-poison"))); + + const outcome = await waitForOutcome(outcomes, "dead-lettered"); + expect(outcome).toEqual({ + kind: "dead-lettered", + eventId: "event-poison", + attempt: 3, + reason: "retry-limit-exceeded", + error: "poison", + }); + expect(deliver).toHaveBeenCalledTimes(3); + const duplicate = await queue.enqueue("event-poison", {} as SpoolPayload); + expect(duplicate.kind).toBe("failed"); + if (duplicate.kind === "failed") { + expect(duplicate.record.reason).toBe("retry-limit-exceeded"); + } + await spool.stop(); + }); + }); + + it("dead-letters without retry after delivery side effects commit", async () => { + await withQueue(async (queue) => { + const deliver = vi.fn(async () => { + throw new LineWebhookTerminalDeliveryError("reply token consumed"); + }); + const outcomes: LineWebhookDeliveryOutcome[] = []; + const spool = createLineWebhookSpool({ + accountId: "default", + runtime: runtime(), + queue, + deliver, + onOutcome: (outcome) => outcomes.push(outcome), + }); + spool.start(); + await spool.accept(callback(createEvent("event-terminal"))); + + const outcome = await waitForOutcome(outcomes, "dead-lettered"); + expect(outcome).toMatchObject({ + eventId: "event-terminal", + attempt: 1, + reason: "delivery-side-effects-committed", + }); + expect(deliver).toHaveBeenCalledTimes(1); + await spool.stop(); + }); + }); + + it("retries completion persistence without redelivering", async () => { + await withQueue(async (queue) => { + const complete = vi + .fn(queue.complete.bind(queue)) + .mockRejectedValueOnce(new Error("database busy")); + const deliver = vi.fn(async () => {}); + const outcomes: LineWebhookDeliveryOutcome[] = []; + const spool = createLineWebhookSpool({ + accountId: "default", + runtime: runtime(), + queue: { ...queue, complete }, + claimRefreshMs: 1, + deliver, + onOutcome: (outcome) => outcomes.push(outcome), + }); + spool.start(); + await spool.accept(callback(createEvent("event-completion-retry"))); + + await waitForOutcome(outcomes, "completed"); + expect(complete).toHaveBeenCalledTimes(2); + expect(deliver).toHaveBeenCalledTimes(1); + await spool.stop(); + }); + }); + + it("finishes completion persistence when shutdown follows successful delivery", async () => { + await withQueue(async (queue) => { + let reportFirstFailure: (() => void) | undefined; + const firstFailure = new Promise((resolve) => { + reportFirstFailure = resolve; + }); + const complete = vi.fn(queue.complete.bind(queue)).mockImplementationOnce(async () => { + reportFirstFailure?.(); + throw new Error("database busy"); + }); + const deliver = vi.fn(async () => {}); + const spool = createLineWebhookSpool({ + accountId: "default", + runtime: runtime(), + queue: { ...queue, complete }, + deliver, + }); + spool.start(); + await spool.accept(callback(createEvent("event-completion-stop"))); + await firstFailure; + + await spool.stop(); + + expect(complete).toHaveBeenCalledTimes(2); + expect(deliver).toHaveBeenCalledTimes(1); + expect((await queue.enqueue("event-completion-stop", {} as SpoolPayload)).kind).toBe( + "completed", + ); + }); + }); + + it("completes at durable turn adoption without replaying later failures", async () => { + await withQueue(async (queue) => { + const deliver = vi.fn(async (_event, _destination, control) => { + await control.onTurnAdopted(); + throw new Error("settle failed after adoption"); + }); + const outcomes: LineWebhookDeliveryOutcome[] = []; + const spool = createLineWebhookSpool({ + accountId: "default", + runtime: runtime(), + queue, + deliver, + onOutcome: (outcome) => outcomes.push(outcome), + }); + spool.start(); + await spool.accept(callback(createEvent("event-adopted"))); + + await waitForOutcome(outcomes, "completed"); + expect(deliver).toHaveBeenCalledTimes(1); + expect((await queue.enqueue("event-adopted", {} as SpoolPayload)).kind).toBe("completed"); + await spool.stop(); + }); + }); + + it("keeps an adopted delivery cancellable until the turn settles", async () => { + await withQueue(async (queue) => { + const adopted = vi.fn(); + const deliver = vi.fn(async (_event, _destination, control) => { + await control.onTurnAdopted(); + adopted(); + await new Promise((_resolve, reject) => { + control.abortSignal.addEventListener("abort", () => reject(new Error("aborted")), { + once: true, + }); + }); + }); + const outcomes: LineWebhookDeliveryOutcome[] = []; + const spool = createLineWebhookSpool({ + accountId: "default", + runtime: runtime(), + queue, + deliver, + onOutcome: (outcome) => outcomes.push(outcome), + }); + spool.start(); + await spool.accept(callback(createEvent("event-adopted-stop"))); + await vi.waitFor(() => { + expect(adopted).toHaveBeenCalledTimes(1); + }); + + await spool.stop(); + + expect(outcomes).toContainEqual({ kind: "completed", eventId: "event-adopted-stop" }); + expect((await queue.enqueue("event-adopted-stop", {} as SpoolPayload)).kind).toBe( + "completed", + ); + }); + }); + + it("retries terminal-state persistence without redelivering", async () => { + await withQueue(async (queue) => { + const fail = vi.fn(queue.fail.bind(queue)).mockRejectedValueOnce(new Error("database busy")); + const deliver = vi.fn(async () => { + throw new LineWebhookTerminalDeliveryError("reply token consumed"); + }); + const outcomes: LineWebhookDeliveryOutcome[] = []; + const spool = createLineWebhookSpool({ + accountId: "default", + runtime: runtime(), + queue: { ...queue, fail }, + claimRefreshMs: 1, + deliver, + onOutcome: (outcome) => outcomes.push(outcome), + }); + spool.start(); + await spool.accept(callback(createEvent("event-terminal-retry"))); + + await waitForOutcome(outcomes, "dead-lettered"); + expect(fail).toHaveBeenCalledTimes(2); + expect(deliver).toHaveBeenCalledTimes(1); + await spool.stop(); + }); + }); + + it("drains a persisted batch prefix when a later enqueue fails", async () => { + await withQueue(async (queue) => { + let enqueueCount = 0; + const enqueue: typeof queue.enqueue = async (...args) => { + enqueueCount += 1; + if (enqueueCount === 2) { + throw new Error("database busy"); + } + return await queue.enqueue(...args); + }; + const deliver = vi.fn(async () => {}); + const outcomes: LineWebhookDeliveryOutcome[] = []; + const spool = createLineWebhookSpool({ + accountId: "default", + runtime: runtime(), + queue: { ...queue, enqueue }, + deliver, + onOutcome: (outcome) => outcomes.push(outcome), + }); + spool.start(); + + await expect( + spool.accept({ + destination: "destination-1", + events: [createEvent("event-prefix-1"), createEvent("event-prefix-2")], + }), + ).rejects.toThrow("database busy"); + await waitForOutcome(outcomes, "completed"); + expect(deliver).toHaveBeenCalledTimes(1); + expect(deliver).toHaveBeenCalledWith( + expect.objectContaining({ webhookEventId: "event-prefix-1" }), + "destination-1", + expect.objectContaining({ abortSignal: expect.any(AbortSignal) }), + ); + await spool.stop(); + }); + }); + + it("stops refreshing and releases a pre-adoption claim on shutdown", async () => { + await withQueue(async (queue) => { + let finishRelease: (() => void) | undefined; + const releaseAllowed = new Promise((resolve) => { + finishRelease = resolve; + }); + const release = vi.fn(async (...args: Parameters) => { + await releaseAllowed; + return await queue.release(...args); + }); + const deliver = vi.fn( + async ( + _event: webhook.Event, + _destination: string, + control: { abortSignal: AbortSignal }, + ) => + await new Promise((_resolve, reject) => { + control.abortSignal.addEventListener("abort", () => reject(new Error("aborted")), { + once: true, + }); + }), + ); + const spool = createLineWebhookSpool({ + accountId: "default", + runtime: runtime(), + queue: { ...queue, release }, + deliver, + }); + spool.start(); + await spool.accept(callback(createEvent("event-stop"))); + await vi.waitFor(async () => { + expect(await queue.listClaims()).toHaveLength(1); + }); + + let stopped = false; + const stop = spool.stop().then(() => { + stopped = true; + }); + await vi.waitFor(() => { + expect(release).toHaveBeenCalledTimes(1); + }); + expect(stopped).toBe(false); + finishRelease?.(); + await stop; + + expect(await queue.listClaims()).toHaveLength(0); + expect(await queue.listPending()).toHaveLength(1); + expect(deliver).toHaveBeenCalledTimes(1); + }); + }); + + it("dead-letters terminal delivery errors even when shutdown aborts the attempt", async () => { + await withQueue(async (queue) => { + const deliver = vi.fn( + async ( + _event: webhook.Event, + _destination: string, + control: { abortSignal: AbortSignal }, + ) => + await new Promise((_resolve, reject) => { + control.abortSignal.addEventListener( + "abort", + () => reject(new LineWebhookTerminalDeliveryError("reply token consumed")), + { once: true }, + ); + }), + ); + const spool = createLineWebhookSpool({ + accountId: "default", + runtime: runtime(), + queue, + deliver, + }); + spool.start(); + await spool.accept(callback(createEvent("event-terminal-stop"))); + await vi.waitFor(async () => { + expect(await queue.listClaims()).toHaveLength(1); + }); + + await spool.stop(); + + expect(await queue.listPending()).toHaveLength(0); + const duplicate = await queue.enqueue("event-terminal-stop", {} as SpoolPayload); + expect(duplicate.kind).toBe("failed"); + if (duplicate.kind === "failed") { + expect(duplicate.record.reason).toBe("delivery-side-effects-committed"); + } + }); + }); + + it("aborts delivery and releases the row when claim refresh loses ownership", async () => { + await withQueue(async (queue) => { + const refreshClaim = vi.fn(async () => false); + const deliver = vi.fn( + async ( + _event: webhook.Event, + _destination: string, + control: { abortSignal: AbortSignal }, + ) => + await new Promise((_resolve, reject) => { + control.abortSignal.addEventListener("abort", () => reject(new Error("aborted")), { + once: true, + }); + }), + ); + const spool = createLineWebhookSpool({ + accountId: "default", + runtime: runtime(), + queue: { ...queue, refreshClaim }, + claimRefreshMs: 1, + deliver, + }); + spool.start(); + await spool.accept(callback(createEvent("event-refresh-loss"))); + + await vi.waitFor(async () => { + expect(refreshClaim).toHaveBeenCalled(); + expect(await queue.listClaims()).toHaveLength(0); + expect(await queue.listPending()).toHaveLength(1); + }); + expect(deliver).toHaveBeenCalledTimes(1); + await spool.stop(); + }); + }); + + it("releases a claim won concurrently with shutdown before delivery starts", async () => { + await withQueue(async (queue) => { + let reportClaimed: (() => void) | undefined; + const claimed = new Promise((resolve) => { + reportClaimed = resolve; + }); + let releaseClaimResult: (() => void) | undefined; + const returnClaim = new Promise((resolve) => { + releaseClaimResult = resolve; + }); + const claimNext: typeof queue.claimNext = async (...args) => { + const claim = await queue.claimNext(...args); + reportClaimed?.(); + await returnClaim; + return claim; + }; + const deliver = vi.fn(async () => {}); + const spool = createLineWebhookSpool({ + accountId: "default", + runtime: runtime(), + queue: { ...queue, claimNext }, + deliver, + }); + spool.start(); + await spool.accept(callback(createEvent("event-stop-race"))); + await claimed; + + const stop = spool.stop(); + releaseClaimResult?.(); + await stop; + + await vi.waitFor(async () => { + expect(await queue.listClaims()).toHaveLength(0); + expect(await queue.listPending()).toHaveLength(1); + }); + expect(deliver).not.toHaveBeenCalled(); + }); + }); + + it("delivers ready lanes beyond a full page of delayed retries", async () => { + await withQueue(async (queue) => { + for (let index = 0; index < 100; index += 1) { + const eventId = `a-delayed-${String(index).padStart(3, "0")}`; + const event = createEvent(eventId, `user-delayed-${index}`); + await queue.enqueue( + eventId, + { version: 1, destination: "destination-1", event }, + { + laneKey: `user:user-delayed-${index}`, + }, + ); + const claim = await queue.claim(eventId); + if (!claim) { + throw new Error(`Expected delayed LINE claim ${eventId}`); + } + await queue.release(claim, { lastError: "delayed" }); + } + const readyEvent = createEvent("z-ready", "user-ready"); + await queue.enqueue( + "z-ready", + { version: 1, destination: "destination-1", event: readyEvent }, + { laneKey: "user:user-ready" }, + ); + const outcomes: LineWebhookDeliveryOutcome[] = []; + const deliver = vi.fn(async () => {}); + const spool = createLineWebhookSpool({ + accountId: "default", + runtime: runtime(), + queue, + retryBaseMs: 60_000, + retryMaxMs: 60_000, + deliver, + onOutcome: (outcome) => outcomes.push(outcome), + }); + spool.start(); + + await waitForOutcome(outcomes, "completed"); + expect(deliver).toHaveBeenCalledWith( + readyEvent, + "destination-1", + expect.objectContaining({ abortSignal: expect.any(AbortSignal) }), + ); + await spool.stop(); + }); + }); +}); diff --git a/extensions/line/src/webhook-spool.ts b/extensions/line/src/webhook-spool.ts new file mode 100644 index 000000000000..3a13652e47ad --- /dev/null +++ b/extensions/line/src/webhook-spool.ts @@ -0,0 +1,540 @@ +// Line plugin module owns durable webhook admission and replay. +import { createHash, randomUUID } from "node:crypto"; +import type { webhook } from "@line/bot-sdk"; +import type { + ChannelIngressQueue, + ChannelIngressQueueClaim, + ChannelIngressQueueRecord, +} from "openclaw/plugin-sdk/channel-outbound"; +import { danger, sleepWithAbort, type RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; +import { runDetachedWebhookWork } from "openclaw/plugin-sdk/webhook-request-guards"; +import { getLineRuntime } from "./runtime.js"; + +const LINE_WEBHOOK_SPOOL_VERSION = 1; +const LINE_WEBHOOK_MAX_ATTEMPTS = 8; +const LINE_WEBHOOK_RETRY_BASE_MS = 1_000; +const LINE_WEBHOOK_RETRY_MAX_MS = 3 * 60_000; +const LINE_WEBHOOK_CLAIM_STALE_MS = 30_000; +const LINE_WEBHOOK_CLAIM_REFRESH_MS = 10_000; +const LINE_WEBHOOK_MAX_CONCURRENT_DELIVERIES = 8; +const LINE_WEBHOOK_DRAIN_SCAN_LIMIT = 100; +const LINE_WEBHOOK_TOMBSTONE_TTL_MS = 30 * 24 * 60 * 60_000; + +type LineWebhookDeadLetterReason = + | "delivery-side-effects-committed" + | "invalid-event" + | "retry-limit-exceeded"; + +type LineWebhookDeliveryOutcome = + | { kind: "completed"; eventId: string } + | { kind: "retry-scheduled"; eventId: string; attempt: number; error: string } + | { + kind: "dead-lettered"; + eventId: string; + attempt: number; + reason: LineWebhookDeadLetterReason; + error: string; + }; + +type LineWebhookSpoolPayload = { + version: number; + destination: string; + event: webhook.Event; +}; + +type LineWebhookSpoolOptions = { + accountId: string; + runtime: RuntimeEnv; + deliver: ( + event: webhook.Event, + destination: string, + control: { abortSignal: AbortSignal; onTurnAdopted: () => Promise }, + ) => Promise; + queue?: ChannelIngressQueue; + maxAttempts?: number; + retryBaseMs?: number; + retryMaxMs?: number; + claimStaleMs?: number; + claimRefreshMs?: number; + onOutcome?: (outcome: LineWebhookDeliveryOutcome) => void; +}; + +class LineWebhookClaimOwnershipError extends Error { + constructor(eventId: string) { + super(`LINE webhook spool event ${eventId} lost claim ownership.`); + this.name = "LineWebhookClaimOwnershipError"; + } +} + +class LineWebhookClaimRefreshError extends Error { + constructor(eventId: string, options?: { cause?: unknown }) { + super(`LINE webhook spool event ${eventId} claim refresh failed.`, options); + this.name = "LineWebhookClaimRefreshError"; + } +} + +export class LineWebhookTerminalDeliveryError extends Error { + readonly reason = "delivery-side-effects-committed" as const; + + constructor(message: string, options?: { cause?: unknown }) { + super(message, options); + this.name = "LineWebhookTerminalDeliveryError"; + } +} + +type LineWebhookSpool = { + accept: (body: webhook.CallbackRequest) => Promise; + start: () => void; + stop: () => Promise; +}; + +function eventIdFor(event: webhook.Event): string { + const eventId = (event as { webhookEventId?: unknown }).webhookEventId; + if (typeof eventId === "string" && eventId.trim()) { + return eventId.trim(); + } + return `invalid:${createHash("sha256").update(JSON.stringify(event)).digest("hex")}`; +} + +function laneKeyFor(event: webhook.Event): string { + const source = (event as { source?: webhook.Event["source"] }).source; + if (source?.type === "group") { + return `group:${source.groupId}`; + } + if (source?.type === "room") { + return `room:${source.roomId}`; + } + if (source?.type === "user") { + return `user:${source.userId}`; + } + return `event:${eventIdFor(event)}`; +} + +function isValidPayload(payload: LineWebhookSpoolPayload): boolean { + return ( + payload.version === LINE_WEBHOOK_SPOOL_VERSION && + typeof payload.destination === "string" && + typeof payload.event === "object" && + payload.event !== null && + typeof (payload.event as { webhookEventId?: unknown }).webhookEventId === "string" && + Boolean((payload.event as { webhookEventId: string }).webhookEventId.trim()) + ); +} + +function errorText(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +function retryDelayMs( + record: ChannelIngressQueueRecord, + now: number, + baseMs: number, + maxMs: number, +): number { + if (record.attempts < 1 || record.lastAttemptAt === undefined) { + return 0; + } + const delay = Math.min(maxMs, baseMs * 2 ** Math.min(record.attempts - 1, 8)); + return Math.max(0, record.lastAttemptAt + delay - now); +} + +export function createLineWebhookSpool(options: LineWebhookSpoolOptions): LineWebhookSpool { + const queue = + options.queue ?? + getLineRuntime().state.openChannelIngressQueue({ + accountId: options.accountId, + }); + const ownerId = `${process.pid}:${randomUUID()}`; + const maxAttempts = Math.max(1, options.maxAttempts ?? LINE_WEBHOOK_MAX_ATTEMPTS); + const retryBaseMs = Math.max(0, options.retryBaseMs ?? LINE_WEBHOOK_RETRY_BASE_MS); + const retryMaxMs = Math.max(retryBaseMs, options.retryMaxMs ?? LINE_WEBHOOK_RETRY_MAX_MS); + const claimStaleMs = Math.max(0, options.claimStaleMs ?? LINE_WEBHOOK_CLAIM_STALE_MS); + const claimRefreshMs = Math.max(1, options.claimRefreshMs ?? LINE_WEBHOOK_CLAIM_REFRESH_MS); + let running = false; + let draining = false; + let drainRequested = false; + let drainTimer: ReturnType | undefined; + let drainDeadline: number | undefined; + const drainTasks = new Set>(); + const activeDeliveries = new Map>(); + const activeClaims = new Map< + string, + { abortController: AbortController; stopRefresh: () => void } + >(); + const backoffBlockedLanes = new Map(); + + const prune = async (): Promise => { + await queue.prune({ + completedTtlMs: LINE_WEBHOOK_TOMBSTONE_TTL_MS, + failedTtlMs: LINE_WEBHOOK_TOMBSTONE_TTL_MS, + }); + }; + + const scheduleDrain = (delayMs: number): void => { + if (!running) { + return; + } + const deadline = Date.now() + Math.max(0, delayMs); + if (drainTimer && drainDeadline !== undefined && drainDeadline <= deadline) { + return; + } + if (drainTimer) { + clearTimeout(drainTimer); + } + drainDeadline = deadline; + drainTimer = setTimeout( + () => { + drainTimer = undefined; + drainDeadline = undefined; + // Timers preserve the HTTP request's admission context. Give every drain + // its own tracked root so durable work can continue after the request ACK. + const task = runDetachedWebhookWork(drain).catch((error: unknown) => { + options.runtime.error?.( + danger(`line: webhook spool admission failed: ${errorText(error)}`), + ); + scheduleDrain(retryBaseMs || 1_000); + }); + drainTasks.add(task); + void task.finally(() => drainTasks.delete(task)); + }, + Math.max(0, deadline - Date.now()), + ); + drainTimer.unref?.(); + }; + + const persistClaimTransition = async (params: { + claim: ChannelIngressQueueClaim; + label: string; + transition: () => Promise; + abortSignal?: AbortSignal; + }): Promise => { + let persistenceAttempt = 0; + while (true) { + try { + if (!(await params.transition())) { + throw new LineWebhookClaimOwnershipError(params.claim.id); + } + return; + } catch (error) { + if (error instanceof LineWebhookClaimOwnershipError) { + throw error; + } + persistenceAttempt += 1; + const delayMs = Math.min(5_000, 250 * 2 ** Math.min(persistenceAttempt - 1, 5)); + options.runtime.error?.( + danger( + `line: webhook event ${params.claim.id} ${params.label} persist failed; retrying: ${errorText(error)}`, + ), + ); + await sleepWithAbort(delayMs, params.abortSignal); + } + } + }; + + const finishClaim = async ( + claim: ChannelIngressQueueClaim, + ): Promise => { + const attempt = claim.attempts + 1; + const abortController = new AbortController(); + let adopted = false; + let refreshing = true; + const claimRefreshTimer = setInterval(() => { + if (!refreshing) { + return; + } + void (async () => { + try { + const refreshed = await queue.refreshClaim?.(claim); + if (refreshed !== true && !adopted && refreshing) { + throw new LineWebhookClaimOwnershipError(claim.id); + } + } catch (error) { + if (adopted || !refreshing) { + return; + } + options.runtime.error?.( + danger(`line: webhook spool claim refresh failed: ${errorText(error)}`), + ); + stopRefresh(); + abortController.abort(new LineWebhookClaimRefreshError(claim.id, { cause: error })); + } + })(); + }, claimRefreshMs); + claimRefreshTimer.unref?.(); + const stopRefresh = () => { + if (!refreshing) { + return; + } + refreshing = false; + clearInterval(claimRefreshTimer); + }; + activeClaims.set(claim.id, { abortController, stopRefresh }); + try { + if (!isValidPayload(claim.payload)) { + const error = "LINE webhook spool payload was invalid."; + await persistClaimTransition({ + claim, + label: "invalid dead-letter", + transition: async () => + await queue.fail(claim, { reason: "invalid-event", message: error }), + }); + return { + kind: "dead-lettered", + eventId: claim.id, + attempt, + reason: "invalid-event", + error, + }; + } + try { + await options.deliver(claim.payload.event, claim.payload.destination, { + abortSignal: abortController.signal, + onTurnAdopted: async () => { + await persistClaimTransition({ + claim, + label: "adoption completion", + transition: async () => await queue.complete(claim), + }); + adopted = true; + stopRefresh(); + }, + }); + } catch (error) { + if (adopted) { + return { kind: "completed", eventId: claim.id }; + } + const message = errorText(error); + if (error instanceof LineWebhookTerminalDeliveryError) { + await persistClaimTransition({ + claim, + label: "terminal dead-letter", + transition: async () => await queue.fail(claim, { reason: error.reason, message }), + }); + return { + kind: "dead-lettered", + eventId: claim.id, + attempt, + reason: error.reason, + error: message, + }; + } + if (abortController.signal.aborted) { + const refreshFailed = + abortController.signal.reason instanceof LineWebhookClaimRefreshError; + await persistClaimTransition({ + claim, + label: refreshFailed ? "refresh-failure release" : "shutdown release", + transition: async () => + await queue.release( + claim, + refreshFailed + ? { lastError: errorText(abortController.signal.reason) } + : { recordAttempt: false }, + ), + }); + throw error; + } + if (attempt >= maxAttempts) { + await persistClaimTransition({ + claim, + label: "retry-limit dead-letter", + transition: async () => + await queue.fail(claim, { reason: "retry-limit-exceeded", message }), + }); + return { + kind: "dead-lettered", + eventId: claim.id, + attempt, + reason: "retry-limit-exceeded", + error: message, + }; + } + await persistClaimTransition({ + claim, + label: "retry release", + transition: async () => await queue.release(claim, { lastError: message }), + }); + return { kind: "retry-scheduled", eventId: claim.id, attempt, error: message }; + } + + if (adopted) { + return { kind: "completed", eventId: claim.id }; + } + + await persistClaimTransition({ + claim, + label: "completion", + transition: async () => await queue.complete(claim), + }); + return { kind: "completed", eventId: claim.id }; + } finally { + activeClaims.delete(claim.id); + stopRefresh(); + } + }; + + const drain = async (): Promise => { + if (!running) { + return; + } + if (draining) { + drainRequested = true; + return; + } + draining = true; + drainRequested = false; + try { + // The short lease recovers a crashed predecessor before LINE's reply-token + // use window, while refreshes protect a still-live rolling-restart owner. + await queue.recoverStaleClaims({ staleMs: claimStaleMs }); + const claims = await queue.listClaims(); + const now = Date.now(); + let nextDelay = Number.POSITIVE_INFINITY; + for (const claim of claims) { + nextDelay = Math.min(nextDelay, Math.max(0, claim.claim.claimedAt + claimStaleMs - now)); + } + const blockedLaneKeys = new Set(activeDeliveries.keys()); + for (const [laneKey, deadline] of backoffBlockedLanes) { + if (deadline <= now) { + backoffBlockedLanes.delete(laneKey); + } else { + blockedLaneKeys.add(laneKey); + nextDelay = Math.min(nextDelay, deadline - now); + } + } + for (const claim of claims) { + if (claim.laneKey) { + blockedLaneKeys.add(claim.laneKey); + } + } + let scanned = 0; + while ( + activeDeliveries.size < LINE_WEBHOOK_MAX_CONCURRENT_DELIVERIES && + scanned < LINE_WEBHOOK_DRAIN_SCAN_LIMIT + ) { + if (!running) { + break; + } + const claim = await queue.claimNext({ + ownerId, + blockedLaneKeys, + orderBy: "id", + scanLimit: LINE_WEBHOOK_DRAIN_SCAN_LIMIT, + }); + if (!claim) { + break; + } + scanned += 1; + if (!running) { + await persistClaimTransition({ + claim, + label: "shutdown claim release", + transition: async () => await queue.release(claim, { recordAttempt: false }), + }); + break; + } + const laneKey = claim.laneKey ?? `event:${claim.id}`; + blockedLaneKeys.add(laneKey); + const delay = retryDelayMs(claim, now, retryBaseMs, retryMaxMs); + if (delay > 0) { + backoffBlockedLanes.set(laneKey, now + delay); + nextDelay = Math.min(nextDelay, delay); + await persistClaimTransition({ + claim, + label: "backoff release", + transition: async () => await queue.release(claim, { recordAttempt: false }), + }); + continue; + } + // The scan root may finish immediately after launching work. Reserve a + // separate admitted continuation for the full delivery lifecycle. + const delivery = runDetachedWebhookWork(() => finishClaim(claim)) + .then((outcome) => { + options.onOutcome?.(outcome); + if (outcome.kind === "retry-scheduled") { + options.runtime.error?.( + danger( + `line: webhook event ${outcome.eventId} delivery failed on attempt ${outcome.attempt}: ${outcome.error}`, + ), + ); + } else if (outcome.kind === "dead-lettered") { + options.runtime.error?.( + danger( + `line: webhook event ${outcome.eventId} dead-lettered (${outcome.reason}): ${outcome.error}`, + ), + ); + } + }) + .catch((error: unknown) => { + options.runtime.error?.( + danger(`line: webhook event ${claim.id} worker failed: ${errorText(error)}`), + ); + }) + .finally(() => { + if (activeDeliveries.get(laneKey) === delivery) { + activeDeliveries.delete(laneKey); + } + scheduleDrain(0); + }); + activeDeliveries.set(laneKey, delivery); + } + if (scanned >= LINE_WEBHOOK_DRAIN_SCAN_LIMIT) { + scheduleDrain(0); + } + if (Number.isFinite(nextDelay)) { + scheduleDrain(nextDelay); + } + } catch (error) { + options.runtime.error?.(danger(`line: webhook spool drain failed: ${errorText(error)}`)); + scheduleDrain(retryBaseMs || 1_000); + } finally { + draining = false; + if (drainRequested) { + scheduleDrain(0); + } + } + }; + + return { + accept: async (body) => { + const events = body.events ?? []; + if (events.length === 0) { + return; + } + await prune(); + const receivedAt = Date.now(); + for (const event of events) { + await queue.enqueue( + eventIdFor(event), + { + version: LINE_WEBHOOK_SPOOL_VERSION, + destination: body.destination ?? "", + event, + }, + { receivedAt, laneKey: laneKeyFor(event) }, + ); + scheduleDrain(0); + } + }, + start: () => { + if (running) { + return; + } + running = true; + scheduleDrain(0); + }, + stop: async () => { + running = false; + if (drainTimer) { + clearTimeout(drainTimer); + drainTimer = undefined; + drainDeadline = undefined; + } + for (const active of activeClaims.values()) { + active.stopRefresh(); + active.abortController.abort(); + } + await Promise.allSettled([...drainTasks, ...activeDeliveries.values()]); + }, + }; +} diff --git a/extensions/line/src/webhook.ts b/extensions/line/src/webhook.ts index 65a7924f1c34..cc78348d1365 100644 --- a/extensions/line/src/webhook.ts +++ b/extensions/line/src/webhook.ts @@ -1,12 +1,7 @@ // Line plugin module implements webhook behavior. import type { webhook } from "@line/bot-sdk"; import type { NextFunction, Request, Response } from "express"; -import { - createMessageReceiveContext, - type MessageReceiveContext, -} from "openclaw/plugin-sdk/channel-outbound"; import { danger, logVerbose, type RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; -import { runDetachedWebhookWork } from "openclaw/plugin-sdk/webhook-request-guards"; import { parseLineWebhookBody, validateLineSignature } from "./webhook-utils.js"; const LINE_WEBHOOK_MAX_RAW_BODY_BYTES = 64 * 1024; @@ -34,17 +29,12 @@ function parseWebhookBody(rawBody?: string | null): webhook.CallbackRequest | nu return parseLineWebhookBody(rawBody); } -function logLineWebhookDispatchError(runtime: RuntimeEnv | undefined, err: unknown): void { - runtime?.error?.(danger(`line webhook dispatch failed: ${String(err)}`)); -} - export function createLineWebhookMiddleware( options: LineWebhookOptions, ): (req: Request, res: Response, _next: NextFunction) => Promise { const { channelSecret, onEvents, runtime } = options; return async (req: Request, res: Response, _next: NextFunction): Promise => { - let receiveContext: MessageReceiveContext | undefined; try { const signature = req.headers["x-line-signature"]; @@ -77,30 +67,12 @@ export function createLineWebhookMiddleware( return; } - receiveContext = createMessageReceiveContext({ - id: `${Date.now()}:line:webhook`, - channel: "line", - message: body, - ackPolicy: "after_receive_record", - onAck: () => { - res.status(200).json({ status: "ok" }); - }, - }); - - if (receiveContext.shouldAckAfter("receive_record")) { - await receiveContext.ack(); - } - if (body.events && body.events.length > 0) { logVerbose(`line: received ${body.events.length} webhook events`); - // Detach event processing from the request admission before the ack - // releases it; an inherited released admission refuses queue work. - void runDetachedWebhookWork(() => onEvents(body)).catch((err: unknown) => - logLineWebhookDispatchError(runtime, err), - ); + await onEvents(body); } + res.status(200).json({ status: "ok" }); } catch (err) { - await receiveContext?.nack(err); runtime?.error?.(danger(`line webhook error: ${String(err)}`)); if (!res.headersSent) { res.status(500).json({ error: "Internal server error" });