mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-26 13:11:12 +00:00
fix(line): preserve webhook events through dispatch failures (#109655)
* test(line): remove obsolete replay-cache coverage * fix(line): durably spool webhook events Co-authored-by: NianJiuZst <180004567+NianJiuZst@users.noreply.github.com> * test(line): tighten webhook spool lifecycle proof * fix(line): use canonical turn adoption lifecycle Co-authored-by: NianJiuZst <180004567+NianJiuZst@users.noreply.github.com> * refactor(line): keep spool types private Co-authored-by: NianJiuZst <180004567+NianJiuZst@users.noreply.github.com> --------- Co-authored-by: NianJiuZst <180004567+NianJiuZst@users.noreply.github.com>
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
cc0b8479d9
commit
5199bfafea
@@ -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.<id>.webhookPath` and update the URL accordingly.
|
||||
|
||||
|
||||
@@ -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<typeof import("./bot-handlers.js").handleLineWebhookEvents>[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<string, HistoryEntry[]>;
|
||||
replayCache?: ReturnType<typeof createLineWebhookReplayCache>;
|
||||
accessGroups?: Record<string, { type: "message.senders"; members: Record<string, string[]> }>;
|
||||
}): Parameters<typeof handleLineWebhookEvents>[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<typeof createLineWebhookReplayCache>,
|
||||
): Parameters<typeof handleLineWebhookEvents>[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<void>((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<void>((_, 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<typeof handleLineWebhookEvents>[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<typeof handleLineWebhookEvents>[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. */
|
||||
|
||||
@@ -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<void>;
|
||||
replayCache?: LineWebhookReplayCache;
|
||||
processMessage: (
|
||||
ctx: LineInboundContext,
|
||||
control: { abortSignal?: AbortSignal; onTurnAdopted?: () => Promise<void> },
|
||||
) => Promise<void>;
|
||||
abortSignal?: AbortSignal;
|
||||
onTurnAdopted?: () => Promise<void>;
|
||||
groupHistories?: Map<string, HistoryEntry[]>;
|
||||
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<LineReplayEvent>({
|
||||
dedupe: {
|
||||
ttlMs: LINE_WEBHOOK_REPLAY_WINDOW_MS,
|
||||
memoryMaxSize: LINE_WEBHOOK_REPLAY_MAX_ENTRIES,
|
||||
},
|
||||
buildReplayKey: ({ event, accountId }) => buildLineWebhookReplayKey(event, accountId)?.key,
|
||||
});
|
||||
}
|
||||
|
||||
type LineWebhookReplayCache = ReturnType<typeof createLineWebhookReplayCache>;
|
||||
|
||||
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;
|
||||
|
||||
@@ -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<void>;
|
||||
onMessage?: (
|
||||
ctx: LineInboundContext,
|
||||
control: { abortSignal?: AbortSignal; onTurnAdopted?: () => Promise<void> },
|
||||
) => Promise<void>;
|
||||
}
|
||||
|
||||
interface LineBot {
|
||||
handleWebhook: (body: webhook.CallbackRequest) => Promise<void>;
|
||||
account: ResolvedLineAccount;
|
||||
stop: () => Promise<void>;
|
||||
}
|
||||
|
||||
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<string, HistoryEntry[]>();
|
||||
|
||||
const handleWebhook = async (body: webhook.CallbackRequest): Promise<void> => {
|
||||
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,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -21,6 +21,7 @@ const {
|
||||
createLineBotMock: vi.fn(() => ({
|
||||
account: { accountId: "default" },
|
||||
handleWebhook: vi.fn<LineHandleWebhook>(),
|
||||
stop: vi.fn(),
|
||||
})),
|
||||
createLineNodeWebhookHandlerMock: vi.fn<() => LineNodeWebhookHandler>(() =>
|
||||
vi.fn<LineNodeWebhookHandler>(async () => {}),
|
||||
@@ -176,6 +177,7 @@ describe("monitorLineProvider lifecycle", () => {
|
||||
createLineBotMock.mockImplementation(() => ({
|
||||
account: { accountId: "default" },
|
||||
handleWebhook: vi.fn<LineHandleWebhook>(),
|
||||
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<typeof vi.fn>;
|
||||
};
|
||||
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();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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<void>;
|
||||
stop: () => void;
|
||||
stop: () => Promise<void>;
|
||||
}
|
||||
|
||||
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<void> | undefined;
|
||||
const stopHandler = (): Promise<void> => {
|
||||
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);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
|
||||
|
||||
@@ -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<string>;
|
||||
|
||||
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<void> };
|
||||
@@ -67,7 +58,6 @@ export function createLineNodeWebhookHandler(params: {
|
||||
return;
|
||||
}
|
||||
|
||||
let receiveContext: MessageReceiveContext<webhook.CallbackRequest> | 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");
|
||||
|
||||
656
extensions/line/src/webhook-spool.test.ts
Normal file
656
extensions/line/src/webhook-spool.test.ts
Normal file
@@ -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<Parameters<typeof createLineWebhookSpool>[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<T>(
|
||||
fn: (queue: ReturnType<typeof createChannelIngressQueue<SpoolPayload>>) => Promise<T>,
|
||||
): Promise<T> {
|
||||
const stateDir = path.join(os.tmpdir(), `openclaw-line-spool-${crypto.randomUUID()}`);
|
||||
await fs.mkdir(stateDir);
|
||||
const queue = createChannelIngressQueue<SpoolPayload>({
|
||||
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<LineWebhookDeliveryOutcome> {
|
||||
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<void>((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<void>((_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<void>((resolve) => {
|
||||
finishRelease = resolve;
|
||||
});
|
||||
const release = vi.fn(async (...args: Parameters<typeof queue.release>) => {
|
||||
await releaseAllowed;
|
||||
return await queue.release(...args);
|
||||
});
|
||||
const deliver = vi.fn(
|
||||
async (
|
||||
_event: webhook.Event,
|
||||
_destination: string,
|
||||
control: { abortSignal: AbortSignal },
|
||||
) =>
|
||||
await new Promise<void>((_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<void>((_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<void>((_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<void>((resolve) => {
|
||||
reportClaimed = resolve;
|
||||
});
|
||||
let releaseClaimResult: (() => void) | undefined;
|
||||
const returnClaim = new Promise<void>((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();
|
||||
});
|
||||
});
|
||||
});
|
||||
540
extensions/line/src/webhook-spool.ts
Normal file
540
extensions/line/src/webhook-spool.ts
Normal file
@@ -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<void> },
|
||||
) => Promise<void>;
|
||||
queue?: ChannelIngressQueue<LineWebhookSpoolPayload>;
|
||||
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<void>;
|
||||
start: () => void;
|
||||
stop: () => Promise<void>;
|
||||
};
|
||||
|
||||
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<LineWebhookSpoolPayload>,
|
||||
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<LineWebhookSpoolPayload>({
|
||||
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<typeof setTimeout> | undefined;
|
||||
let drainDeadline: number | undefined;
|
||||
const drainTasks = new Set<Promise<void>>();
|
||||
const activeDeliveries = new Map<string, Promise<void>>();
|
||||
const activeClaims = new Map<
|
||||
string,
|
||||
{ abortController: AbortController; stopRefresh: () => void }
|
||||
>();
|
||||
const backoffBlockedLanes = new Map<string, number>();
|
||||
|
||||
const prune = async (): Promise<void> => {
|
||||
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<LineWebhookSpoolPayload>;
|
||||
label: string;
|
||||
transition: () => Promise<boolean>;
|
||||
abortSignal?: AbortSignal;
|
||||
}): Promise<void> => {
|
||||
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<LineWebhookSpoolPayload>,
|
||||
): Promise<LineWebhookDeliveryOutcome> => {
|
||||
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<void> => {
|
||||
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()]);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -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<void> {
|
||||
const { channelSecret, onEvents, runtime } = options;
|
||||
|
||||
return async (req: Request, res: Response, _next: NextFunction): Promise<void> => {
|
||||
let receiveContext: MessageReceiveContext<webhook.CallbackRequest> | 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" });
|
||||
|
||||
Reference in New Issue
Block a user