Files
openclaw/extensions/googlechat/src/monitor-webhook.ts
Eden d572c5cb2d fix(line): run post-ack webhook processing on its own admitted work root (#107734)
* fix(line): run post-ack webhook processing on its own admitted work root

LINE acks the webhook and dispatches event processing fire-and-forget on
the same async chain. The HTTP request admission that chain inherited is
released as soon as the route handler returns, and a released admission
refuses subordinate queue work - so every LINE inbound agent turn fails
with "GatewayDrainingError: Gateway is draining; new tasks are not
accepted" even though the gateway is healthy. DMs, group mentions, and
postbacks are all affected; the user-visible symptom is the bot replying
"Sorry, I encountered an error processing your message." to everything.

Add runDetachedWebhookWork to the plugin-sdk webhook-request-guards
surface (a thin wrapper over the gateway independent-root continuation,
the same shape core uses in gateway/server/hooks.ts) and route all three
LINE ack-first dispatch sites through it: the gateway monitor handler
(the live path), and the createLineNodeWebhookHandler / Express
middleware handlers (public webhook building blocks an embedder can
register under the gateway). #65375 unified these three into one ack-first
pattern; keeping the detach consistent avoids re-introducing the same
latent defect in the two that are not on the live gateway path today.

The continuation is reserved synchronously while the request is still
admitted, so the detached processing stays accepted and a real restart
drain can wait for it instead of stranding it mid-turn.

Tests pin every layer: the guards suite proves detached post-ack work is
admitted after the request admission is released (and that the inherited
chain without the helper is refused); the monitor lifecycle suite and the
webhook-node suite assert each dispatch site goes through the detached
root. Red/green verified: reverting any dispatch fails its test.

* fix(plugin-sdk): account for runDetachedWebhookWork in public surface budget

* fix(channels): track detached webhook processing

Co-authored-by: 許元豪 <146086744+edenfunf@users.noreply.github.com>

* docs: refresh generated docs map

* chore(plugin-sdk): refresh API baseline

* fix(webhooks): preserve post-ack ordering

* test(plugin-sdk): satisfy detached work lint

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>
2026-07-16 03:39:02 -07:00

369 lines
12 KiB
TypeScript

// Googlechat plugin module implements monitor webhook behavior.
import type { IncomingMessage, ServerResponse } from "node:http";
import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coerce-runtime";
import {
normalizeWebhookPath,
resolveRequestClientIp,
type FixedWindowRateLimiter,
} from "openclaw/plugin-sdk/webhook-ingress";
import {
readJsonWebhookBodyOrReject,
runDetachedWebhookWork,
type WebhookInFlightLimiter,
} from "openclaw/plugin-sdk/webhook-request-guards";
import {
resolveWebhookTargetWithAuthOrReject,
withResolvedWebhookRequestPipeline,
} from "openclaw/plugin-sdk/webhook-targets";
import { verifyGoogleChatRequest } from "./auth.js";
import type { WebhookTarget } from "./monitor-types.js";
import type {
GoogleChatAction,
GoogleChatActionParameter,
GoogleChatEvent,
GoogleChatMessage,
GoogleChatSpace,
GoogleChatUser,
} from "./types.js";
function extractBearerToken(header: unknown): string {
const authHeader = Array.isArray(header)
? typeof header[0] === "string"
? header[0]
: ""
: typeof header === "string"
? header
: "";
return normalizeLowercaseStringOrEmpty(authHeader).startsWith("bearer ")
? authHeader.slice("bearer ".length).trim()
: "";
}
const ADD_ON_PREAUTH_MAX_BYTES = 16 * 1024;
const ADD_ON_PREAUTH_TIMEOUT_MS = 3_000;
type ParsedGoogleChatInboundPayload =
| { ok: true; event: GoogleChatEvent; addOnBearerToken: string }
| { ok: false };
type ParsedGoogleChatInboundSuccess = Extract<ParsedGoogleChatInboundPayload, { ok: true }>;
function recordParamsToActionParameters(
params?: Record<string, string>,
): GoogleChatActionParameter[] | undefined {
if (!params) {
return undefined;
}
const entries = Object.entries(params)
.filter((entry): entry is [string, string] => typeof entry[1] === "string")
.map(([key, value]) => ({ key, value }));
return entries.length > 0 ? entries : undefined;
}
function parseGoogleChatInboundPayload(
raw: unknown,
res: ServerResponse,
): ParsedGoogleChatInboundPayload {
if (!raw || typeof raw !== "object" || Array.isArray(raw)) {
res.statusCode = 400;
res.end("invalid payload");
return { ok: false };
}
let eventPayload = raw;
let addOnBearerToken = "";
// Transform Google Workspace Add-on format to standard Chat API format.
const rawObj = raw as {
commonEventObject?: {
hostApp?: string;
invokedFunction?: string;
parameters?: Record<string, string>;
};
chat?: {
messagePayload?: { space?: GoogleChatSpace; message?: GoogleChatMessage };
buttonClickedPayload?: {
space?: GoogleChatSpace;
message?: GoogleChatMessage;
user?: GoogleChatUser;
action?: GoogleChatAction;
};
user?: GoogleChatUser;
eventTime?: string;
};
authorizationEventObject?: { systemIdToken?: string };
};
if (rawObj.commonEventObject?.hostApp === "CHAT") {
addOnBearerToken =
typeof rawObj.authorizationEventObject?.systemIdToken === "string"
? rawObj.authorizationEventObject.systemIdToken.trim()
: "";
}
if (rawObj.commonEventObject?.hostApp === "CHAT" && rawObj.chat?.messagePayload) {
const chat = rawObj.chat;
const messagePayload = chat.messagePayload;
eventPayload = {
type: "MESSAGE",
space: messagePayload?.space,
message: messagePayload?.message,
user: chat.user,
eventTime: chat.eventTime,
};
} else if (rawObj.commonEventObject?.hostApp === "CHAT") {
const chat = rawObj.chat;
const buttonClickedPayload = chat?.buttonClickedPayload;
if (buttonClickedPayload) {
const invokedFunction = rawObj.commonEventObject.invokedFunction;
const actionParameters = recordParamsToActionParameters(rawObj.commonEventObject.parameters);
eventPayload = {
type: "CARD_CLICKED",
space: buttonClickedPayload.space,
message: buttonClickedPayload.message,
user: buttonClickedPayload.user ?? chat.user,
eventTime: chat.eventTime,
action:
buttonClickedPayload.action ??
({
...(typeof invokedFunction === "string" ? { actionMethodName: invokedFunction } : {}),
...(actionParameters ? { parameters: actionParameters } : {}),
} satisfies GoogleChatAction),
commonEventObject: {
...(typeof invokedFunction === "string" ? { invokedFunction } : {}),
parameters: rawObj.commonEventObject.parameters,
},
};
}
}
const event = eventPayload as GoogleChatEvent;
const eventType = event.type ?? (eventPayload as { eventType?: string }).eventType;
if (typeof eventType !== "string") {
res.statusCode = 400;
res.end("invalid payload");
return { ok: false };
}
if (!event.space || typeof event.space !== "object" || Array.isArray(event.space)) {
res.statusCode = 400;
res.end("invalid payload");
return { ok: false };
}
if (eventType === "MESSAGE") {
if (!event.message || typeof event.message !== "object" || Array.isArray(event.message)) {
res.statusCode = 400;
res.end("invalid payload");
return { ok: false };
}
} else if (eventType === "CARD_CLICKED") {
if (!event.user || typeof event.user !== "object" || Array.isArray(event.user)) {
res.statusCode = 400;
res.end("invalid payload");
return { ok: false };
}
}
return { ok: true, event, addOnBearerToken };
}
type GoogleChatWebhookAuthRejection = {
target: WebhookTarget;
reason: string;
};
async function verifyGoogleChatTargetAuth(
target: WebhookTarget,
bearer: string,
): Promise<{ ok: true } | { ok: false; reason: string }> {
const verification = await verifyGoogleChatRequest({
bearer,
audienceType: target.audienceType,
audience: target.audience,
expectedAddOnPrincipal: target.account.config.appPrincipal,
});
return verification.ok ? { ok: true } : { ok: false, reason: verification.reason ?? "unknown" };
}
function logGoogleChatWebhookAuthRejections(rejections: GoogleChatWebhookAuthRejection[]): void {
for (const rejection of rejections) {
rejection.target.runtime.log?.(
`[${rejection.target.account.accountId}] Google Chat webhook auth rejected: ${rejection.reason}`,
);
}
}
function logGoogleChatWebhookAuthRejectedForTargets(
targets: readonly WebhookTarget[],
reason: string,
): void {
logGoogleChatWebhookAuthRejections(targets.map((target) => ({ target, reason })));
}
async function resolveGoogleChatWebhookTargetWithAuthOrReject(params: {
targets: readonly WebhookTarget[];
res: ServerResponse;
bearer: string;
}): Promise<WebhookTarget | null> {
const rejections: GoogleChatWebhookAuthRejection[] = [];
let verifiedTargetCount = 0;
const selectedTarget = await resolveWebhookTargetWithAuthOrReject({
targets: params.targets,
res: params.res,
isMatch: async (target) => {
const verification = await verifyGoogleChatTargetAuth(target, params.bearer);
if (verification.ok) {
verifiedTargetCount += 1;
return true;
}
rejections.push({ target, reason: verification.reason });
return false;
},
});
if (!selectedTarget && verifiedTargetCount === 0) {
logGoogleChatWebhookAuthRejections(rejections);
}
return selectedTarget;
}
export function warnAppPrincipalMisconfiguration(params: {
accountId: string;
audienceType?: string;
appPrincipal?: string | null;
log?: (message: string) => void;
}): void {
if (params.audienceType !== "app-url") {
return;
}
const principal = params.appPrincipal?.trim();
if (!principal) {
params.log?.(
`[${params.accountId}] appPrincipal is missing for audienceType "app-url"; add-on token verification will fail. Set appPrincipal to the numeric OAuth 2.0 client ID (uniqueId, 21 digits), not an email.`,
);
} else if (principal.includes("@")) {
params.log?.(
`[${params.accountId}] appPrincipal "${principal}" looks like an email address. Set appPrincipal to the numeric OAuth 2.0 client ID (uniqueId, 21 digits), not an email.`,
);
}
}
export function createGoogleChatWebhookRequestHandler(params: {
webhookTargets: Map<string, WebhookTarget[]>;
webhookRateLimiter: FixedWindowRateLimiter;
webhookInFlightLimiter: WebhookInFlightLimiter;
processEvent: (event: GoogleChatEvent, target: WebhookTarget) => Promise<void>;
}): (req: IncomingMessage, res: ServerResponse) => Promise<boolean> {
return async (req: IncomingMessage, res: ServerResponse): Promise<boolean> => {
const path = normalizeWebhookPath(new URL(req.url ?? "/", "http://localhost").pathname);
// Shared-path registrations use the same gateway proxy settings in normal runtime setup.
const config = params.webhookTargets.get(path)?.[0]?.config;
const clientIp =
resolveRequestClientIp(
req,
config?.gateway?.trustedProxies,
config?.gateway?.allowRealIpFallback === true,
) ?? "unknown";
return await withResolvedWebhookRequestPipeline({
req,
res,
targetsByPath: params.webhookTargets,
allowMethods: ["POST"],
requireJsonContentType: true,
rateLimiter: params.webhookRateLimiter,
rateLimitKey: `${path}:${clientIp}`,
inFlightLimiter: params.webhookInFlightLimiter,
handle: async ({ targets }) => {
const headerBearer = extractBearerToken(req.headers.authorization);
let selectedTarget: WebhookTarget | null;
let parsedEvent: GoogleChatEvent | null;
const readAndParseEvent = async (
profile: "pre-auth" | "post-auth",
): Promise<ParsedGoogleChatInboundSuccess | null> => {
const body = await readJsonWebhookBodyOrReject({
req,
res,
profile,
...(profile === "pre-auth"
? {
maxBytes: ADD_ON_PREAUTH_MAX_BYTES,
timeoutMs: ADD_ON_PREAUTH_TIMEOUT_MS,
}
: {}),
emptyObjectOnEmpty: false,
invalidJsonMessage: "invalid payload",
});
if (!body.ok) {
return null;
}
const parsed = parseGoogleChatInboundPayload(body.value, res);
return parsed.ok ? parsed : null;
};
if (headerBearer) {
selectedTarget = await resolveGoogleChatWebhookTargetWithAuthOrReject({
targets,
res,
bearer: headerBearer,
});
if (!selectedTarget) {
return true;
}
const parsed = await readAndParseEvent("post-auth");
if (!parsed) {
return true;
}
parsedEvent = parsed.event;
} else {
const parsed = await readAndParseEvent("pre-auth");
if (!parsed) {
return true;
}
parsedEvent = parsed.event;
if (!parsed.addOnBearerToken) {
logGoogleChatWebhookAuthRejectedForTargets(targets, "missing token");
res.statusCode = 401;
res.end("unauthorized");
return true;
}
selectedTarget = await resolveGoogleChatWebhookTargetWithAuthOrReject({
targets,
res,
bearer: parsed.addOnBearerToken,
});
if (!selectedTarget) {
return true;
}
}
if (!selectedTarget || !parsedEvent) {
res.statusCode = 401;
res.end("unauthorized");
return true;
}
const dispatchTarget = selectedTarget;
dispatchTarget.statusSink?.({ lastInboundAt: Date.now() });
// Reserve the detached task before the HTTP admission is released;
// otherwise later queue work inherits a released admission root.
void runDetachedWebhookWork(() => params.processEvent(parsedEvent, dispatchTarget)).catch(
(err: unknown) => {
dispatchTarget.runtime.error?.(
`[${dispatchTarget.account.accountId}] Google Chat webhook failed: ${String(err)}`,
);
},
);
res.statusCode = 200;
res.setHeader("Content-Type", "application/json");
res.end("{}");
return true;
},
});
};
}