274 lines
11 KiB
TypeScript
274 lines
11 KiB
TypeScript
import PQueue from 'p-queue';
|
|
import { logger } from './logger.js';
|
|
import type { Repository, SpaceWebhookRecord, WebhookEventType } from './db/repository.js';
|
|
import { decrypt, loadKeyFromEnv } from './mcp/crypto.js';
|
|
import { ssrfSafeFetch } from './engine/tools/shared/ssrf.js';
|
|
import type { NotificationEvent, WebhookAdapter } from './webhook-adapters/types.js';
|
|
import { buildDiscordPayload } from './webhook-adapters/discord.js';
|
|
import { buildSlackPayload } from './webhook-adapters/slack.js';
|
|
import { buildTeamsPayload } from './webhook-adapters/teams.js';
|
|
|
|
/**
|
|
* Workspace webhook delivery service (issue #797, PR1 — Discord; PR2 — Slack;
|
|
* PR3 — Teams. All three known providers now have adapters registered below.)
|
|
*
|
|
* Modeled closely on src/push-service.ts: worker hooks call `enqueue()`
|
|
* fire-and-forget, a bounded PQueue absorbs concurrency, and delivery
|
|
* failures never throw back into the worker's job-completion path.
|
|
*
|
|
* SECURITY (do not regress):
|
|
* - The webhook URL is decrypted here (crypto.ts, same AES-256-GCM scheme
|
|
* as mcp_servers.static_token_enc) ONLY for the outbound fetch. It is
|
|
* never logged, never included in a thrown Error message, and never
|
|
* returned to a caller.
|
|
* - All outbound sends go through `ssrfSafeFetch` — never raw `fetch`.
|
|
* - https:// is enforced here in addition to the API-layer validation,
|
|
* since ssrfSafeFetch itself does not gate on scheme.
|
|
*/
|
|
|
|
export interface WebhookNotifyInput {
|
|
event: WebhookEventType;
|
|
spaceId: string;
|
|
taskId: number;
|
|
taskTitle: string;
|
|
pieceName: string;
|
|
spaceName: string;
|
|
}
|
|
|
|
export interface WebhookServiceOptions {
|
|
queueConcurrency?: number;
|
|
perSendTimeoutMs?: number;
|
|
payloadMaxBytes?: number;
|
|
maxRedirects?: number;
|
|
maxRetryAttempts?: number;
|
|
backoffMs?: number;
|
|
/** Absolute base URL for task-detail links; relative path when unset. */
|
|
publicBaseUrl?: string;
|
|
}
|
|
|
|
const DEFAULT_OPTIONS: Required<WebhookServiceOptions> = {
|
|
queueConcurrency: 8,
|
|
perSendTimeoutMs: 10_000,
|
|
payloadMaxBytes: 8_192,
|
|
maxRedirects: 5,
|
|
maxRetryAttempts: 2, // initial + 2 retries = up to 3 sends
|
|
backoffMs: 500,
|
|
publicBaseUrl: '',
|
|
};
|
|
|
|
const ADAPTERS: Partial<Record<SpaceWebhookRecord['provider'], WebhookAdapter>> = {
|
|
discord: buildDiscordPayload as WebhookAdapter,
|
|
slack: buildSlackPayload as WebhookAdapter,
|
|
teams: buildTeamsPayload as WebhookAdapter,
|
|
};
|
|
|
|
export type SendResult = { ok: true } | { ok: false; error: string };
|
|
|
|
function sleep(ms: number): Promise<void> {
|
|
return new Promise(resolve => setTimeout(resolve, ms));
|
|
}
|
|
|
|
function withTimeout<T>(promise: Promise<T>, ms: number): Promise<T> {
|
|
return new Promise((resolve, reject) => {
|
|
const timer = setTimeout(() => reject(new Error(`webhook send timed out after ${ms}ms`)), ms);
|
|
promise.then(
|
|
value => { clearTimeout(timer); resolve(value); },
|
|
err => { clearTimeout(timer); reject(err); },
|
|
);
|
|
});
|
|
}
|
|
|
|
/** Build the task-detail deep link used in the payload. Relative when no publicBaseUrl configured. */
|
|
export function buildTaskUrl(spaceId: string, taskId: number, publicBaseUrl: string): string {
|
|
const path = `/?page=spaces&space=${encodeURIComponent(spaceId)}&chat=${taskId}`;
|
|
return publicBaseUrl ? `${publicBaseUrl.replace(/\/+$/, '')}${path}` : path;
|
|
}
|
|
|
|
/**
|
|
* Serialize a provider payload with a byte-size cap. When the detailed
|
|
* payload exceeds `maxBytes`, falls back to a generic (includeDetails=false)
|
|
* payload built from the same adapter — mirrors push-service.ts's
|
|
* buildPushPayload trim-then-fallback strategy.
|
|
*/
|
|
export function encodeWebhookPayload(
|
|
adapter: WebhookAdapter,
|
|
event: NotificationEvent,
|
|
maxBytes: number,
|
|
): string {
|
|
const full = JSON.stringify(adapter(event));
|
|
if (Buffer.byteLength(full, 'utf8') <= maxBytes) return full;
|
|
if (!event.includeDetails) return full; // already generic; nothing left to trim
|
|
const generic = JSON.stringify(adapter({ ...event, includeDetails: false }));
|
|
return generic;
|
|
}
|
|
|
|
export class WebhookDeliveryService {
|
|
private readonly queue: PQueue;
|
|
private readonly options: Required<WebhookServiceOptions>;
|
|
|
|
constructor(
|
|
private readonly repo: Repository,
|
|
options: WebhookServiceOptions = {},
|
|
) {
|
|
this.options = { ...DEFAULT_OPTIONS, ...options };
|
|
this.queue = new PQueue({ concurrency: this.options.queueConcurrency });
|
|
}
|
|
|
|
/**
|
|
* Fire-and-forget — worker hooks call this and never await the promise.
|
|
* Silently drops webhooks whose `events` list does not include this event
|
|
* (no error, no log spam — this is expected filtering, not a failure).
|
|
*/
|
|
enqueue(input: WebhookNotifyInput): void {
|
|
let webhooks: SpaceWebhookRecord[];
|
|
try {
|
|
webhooks = this.repo.listSpaceWebhooks(input.spaceId);
|
|
} catch (err) {
|
|
logger.error(`[webhook] listSpaceWebhooks failed space=${input.spaceId} err=${String(err)}`);
|
|
return;
|
|
}
|
|
for (const webhook of webhooks) {
|
|
if (!webhook.enabled) continue;
|
|
if (!webhook.events.includes(input.event)) continue; // silent drop by design
|
|
const notification: NotificationEvent = {
|
|
event: input.event,
|
|
taskId: input.taskId,
|
|
taskTitle: input.taskTitle,
|
|
pieceName: input.pieceName,
|
|
spaceName: input.spaceName,
|
|
taskUrl: buildTaskUrl(input.spaceId, input.taskId, this.options.publicBaseUrl),
|
|
includeDetails: webhook.includeDetails,
|
|
};
|
|
void this.queue
|
|
.add(() => this.sendWithRetry(webhook, notification))
|
|
.catch(err => {
|
|
logger.error(`[webhook] queue error webhook=${webhook.id} event=${input.event} err=${String(err)}`);
|
|
});
|
|
}
|
|
}
|
|
|
|
/** Returns when the queue is empty. Mainly useful in tests / shutdown. */
|
|
async waitIdle(): Promise<void> {
|
|
await this.queue.onIdle();
|
|
}
|
|
|
|
/**
|
|
* Manual "send test" action from the settings UI. Awaited directly (not
|
|
* fire-and-forget) so the API route can report success/failure inline.
|
|
* Runs through the same queue so a burst of test-sends still respects the
|
|
* concurrency cap.
|
|
*/
|
|
async sendTest(webhookId: string): Promise<SendResult> {
|
|
const webhook = this.repo.getSpaceWebhookById(webhookId);
|
|
if (!webhook) return { ok: false, error: 'webhook not found' };
|
|
let spaceName = '';
|
|
try {
|
|
const space = await this.repo.getSpace(webhook.spaceId);
|
|
spaceName = space?.title ?? '';
|
|
} catch {
|
|
// best-effort; test payload still sends with an empty workspace name
|
|
}
|
|
const notification: NotificationEvent = {
|
|
event: 'succeeded',
|
|
taskId: 0,
|
|
taskTitle: 'テスト通知',
|
|
pieceName: 'Webhook 動作確認',
|
|
spaceName,
|
|
taskUrl: buildTaskUrl(webhook.spaceId, 0, this.options.publicBaseUrl),
|
|
includeDetails: webhook.includeDetails,
|
|
};
|
|
return this.queue.add(() => this.sendWithRetry(webhook, notification)) as Promise<SendResult>;
|
|
}
|
|
|
|
private async sendWithRetry(webhook: SpaceWebhookRecord, event: NotificationEvent): Promise<SendResult> {
|
|
const adapter = ADAPTERS[webhook.provider];
|
|
if (!adapter) {
|
|
// Defensive: every known WebhookProvider has an adapter registered
|
|
// above as of PR3. This only fires for a future provider value that
|
|
// slipped past the API layer's IMPLEMENTED_PROVIDERS gate. Not a
|
|
// delivery failure — skip silently rather than incrementing failure_count.
|
|
logger.warn(`[webhook] no adapter for provider=${webhook.provider} webhook=${webhook.id} — skipping`);
|
|
return { ok: false, error: `provider not implemented: ${webhook.provider}` };
|
|
}
|
|
|
|
let url: string;
|
|
try {
|
|
url = decrypt(webhook.urlEnc, loadKeyFromEnv());
|
|
} catch (err) {
|
|
// Never include the decrypt error's context (could echo ciphertext bytes)
|
|
// and never log the URL itself.
|
|
logger.error(`[webhook] decrypt failed webhook=${webhook.id} err=${(err as Error).message}`);
|
|
this.repo.markSpaceWebhookFailure(webhook.id);
|
|
return { ok: false, error: 'internal error' };
|
|
}
|
|
if (!url.startsWith('https://')) {
|
|
logger.error(`[webhook] non-https URL rejected webhook=${webhook.id}`);
|
|
this.repo.markSpaceWebhookFailure(webhook.id);
|
|
return { ok: false, error: 'webhook URL must be https' };
|
|
}
|
|
|
|
const body = encodeWebhookPayload(adapter, event, this.options.payloadMaxBytes);
|
|
|
|
let attempt = 0;
|
|
while (true) {
|
|
let res: Response;
|
|
try {
|
|
res = await withTimeout(
|
|
ssrfSafeFetch(
|
|
url,
|
|
[],
|
|
{
|
|
method: 'POST',
|
|
headers: { 'content-type': 'application/json' },
|
|
body,
|
|
},
|
|
this.options.maxRedirects,
|
|
),
|
|
this.options.perSendTimeoutMs,
|
|
);
|
|
} catch (err) {
|
|
const msg = (err as Error).message ?? String(err);
|
|
// SSRF-blocked (or otherwise fundamentally disallowed) destinations
|
|
// will never succeed on retry — treat as permanent.
|
|
if (msg.includes('SSRF blocked')) {
|
|
logger.warn(`[webhook] blocked destination webhook=${webhook.id}: ${msg}`);
|
|
this.repo.markSpaceWebhookFailure(webhook.id);
|
|
return { ok: false, error: 'destination not allowed' };
|
|
}
|
|
// Network error / timeout / DNS failure: transient, retry.
|
|
if (attempt < this.options.maxRetryAttempts) {
|
|
attempt += 1;
|
|
await sleep(this.options.backoffMs * Math.pow(2, attempt) + Math.random() * 300);
|
|
continue;
|
|
}
|
|
logger.warn(`[webhook] send failed (network) webhook=${webhook.id} attempt=${attempt} msg=${msg}`);
|
|
this.repo.markSpaceWebhookFailure(webhook.id);
|
|
return { ok: false, error: 'network error' };
|
|
}
|
|
|
|
if (res.ok) {
|
|
this.repo.markSpaceWebhookSuccess(webhook.id);
|
|
return { ok: true };
|
|
}
|
|
|
|
const code = res.status;
|
|
// 401 / 403 / 404 / 410: permanent — the destination is gone or rejects us.
|
|
if (code === 401 || code === 403 || code === 404 || code === 410) {
|
|
this.repo.markSpaceWebhookFailure(webhook.id);
|
|
logger.info(`[webhook] permanent failure (${code}) webhook=${webhook.id}`);
|
|
return { ok: false, error: `webhook rejected (${code})` };
|
|
}
|
|
// 429 / 5xx: transient — backoff + jitter, retry up to maxRetryAttempts.
|
|
const isTransient = code === 429 || (code >= 500 && code < 600);
|
|
if (isTransient && attempt < this.options.maxRetryAttempts) {
|
|
attempt += 1;
|
|
await sleep(this.options.backoffMs * Math.pow(2, attempt) + Math.random() * 300);
|
|
continue;
|
|
}
|
|
this.repo.markSpaceWebhookFailure(webhook.id);
|
|
logger.warn(`[webhook] send failed webhook=${webhook.id} status=${code} attempt=${attempt}`);
|
|
return { ok: false, error: `webhook send failed (${code})` };
|
|
}
|
|
}
|
|
}
|