import type { AppConfig } from './config.js'; import { decryptText, encryptText, sha256Text } from './crypto.js'; import { getPool, withTransaction } from './db.js'; import { diagnosticError } from './diagnostics.js'; import { ExternalWebhookClient, EXTERNAL_WEBHOOK_CONFIG_ID, type ExternalWebhookSendResult } from './external-webhook-client.js'; import { buildLeaderTaskSummary, isLeaderTaskSummaryStatus } from './leadership-task-summary.js'; export interface LeaderNotificationLogger { info(metadata: Record, message?: string): void; warn(metadata: Record, message?: string): void; error(metadata: Record, message?: string): void; } export interface LeaderSummaryWebhookSender { send(content: string): Promise; } export interface PublicLeaderSummaryWebhookStatus { delivery_mode: 'external_webhook'; webhook_config_id: typeof EXTERNAL_WEBHOOK_CONFIG_ID; configured: boolean; enabled: boolean; state: 'not_configured' | 'invalid_configuration' | 'initializing' | 'active' | 'configuration_changed'; configuration_error: string | null; status_message: string; endpoint_fingerprint: string; starts_at: string | null; revision: number | null; pending_count: number; failed_count: number; uncertain_count: number; accepted_count: number; last_accepted_at: string | null; updated_at: string | null; } interface ClaimedWebhookDelivery { id: string; taskId: string; content: string; payloadFingerprint: string; attemptCount: number; } const PROJECTABLE_STATUSES = [ 'completed', 'dry_run', 'failed', 'parse_failed', 'parse_blocked', 'agent_parse_blocked', 'blocked', 'operation_blocked', 'cancelled', 'reconciliation_pending', 'saved_unverified', 'execution_uncertain', 'uncertain' ] as const; const NEEDS_REVIEW_STATUSES = [ 'reconciliation_pending', 'saved_unverified', 'execution_uncertain', 'uncertain' ] as const; const noopLogger: LeaderNotificationLogger = { info: () => undefined, warn: () => undefined, error: () => undefined }; function text(value: unknown): string { return value == null ? '' : String(value).trim(); } function booleanValue(value: unknown): boolean { return value === true || text(value) === 'true'; } function iso(value: unknown): string | null { if (!value) return null; const date = new Date(String(value)); return Number.isNaN(date.getTime()) ? null : date.toISOString(); } function jsonObject(value: unknown): Record { return value && typeof value === 'object' && !Array.isArray(value) ? value as Record : {}; } export class LeaderNotificationService { private projectorTimer: NodeJS.Timeout | null = null; private projectorInFlight: Promise | null = null; private projectorOrganizationId = ''; private readonly sender: LeaderSummaryWebhookSender | null; private readonly leaseOwner = `webhook:${process.pid}`; constructor( private readonly config: AppConfig, private readonly logger: LeaderNotificationLogger = noopLogger, sender?: LeaderSummaryWebhookSender ) { this.sender = sender || ( config.leaderSummaryWebhookEnabled ? new ExternalWebhookClient({ url: config.WEBHOOK_SEND_URL as string, token: config.WEBHOOK_EXTERNAL_TOKEN as string }) : null ); } private log( level: 'info' | 'warn' | 'error', event: string, metadata: Record, message: string ): void { try { this.logger[level]({ diagnostic_event: `leader_summary.${event}`, notification_event: event, ...metadata }, message); } catch { // Notification persistence and task state never depend on logging. } } private configurationFingerprint(): string | null { if (!this.config.leaderSummaryWebhookEnabled) return null; return sha256Text([ 'leader-summary-webhook-v1', this.config.WEBHOOK_SEND_URL, this.config.WEBHOOK_EXTERNAL_TOKEN, EXTERNAL_WEBHOOK_CONFIG_ID ].join('\u0000')); } private endpointFingerprint(): string { return this.config.leaderSummaryWebhookEnabled && this.config.WEBHOOK_SEND_URL ? sha256Text(this.config.WEBHOOK_SEND_URL).slice(0, 12) : ''; } startProjector(organizationId: string): void { if (this.projectorTimer) return; this.projectorOrganizationId = organizationId; this.projectorTimer = setInterval(() => void this.projectTick(), 2_000); void this.projectTick(); } async stopProjector(): Promise { if (this.projectorTimer) clearInterval(this.projectorTimer); this.projectorTimer = null; if (this.projectorInFlight) await this.projectorInFlight.catch(() => undefined); this.projectorOrganizationId = ''; } private async projectTick(): Promise { if (!this.projectorOrganizationId) return; if (this.projectorInFlight) return this.projectorInFlight; this.projectorInFlight = this.runTick(this.projectorOrganizationId) .catch((error) => { this.log('warn', 'webhook_tick_failed', { organization_id: this.projectorOrganizationId, ...diagnosticError(error, 'leader_summary_webhook_tick_failed') }, 'Leader task summary webhook tick failed'); }) .finally(() => { this.projectorInFlight = null; }); return this.projectorInFlight; } private async runTick(organizationId: string): Promise { const changed = await this.reconcileWebhookState(organizationId); if (changed) { this.log('info', 'webhook_configuration_reconciled', { organization_id: organizationId, configured: this.config.leaderSummaryWebhookEnabled, endpoint_fingerprint: this.endpointFingerprint() }, 'Leader task summary webhook configuration reconciled'); } if (!this.config.leaderSummaryWebhookEnabled || !this.sender) return; const projected = await this.projectPending(organizationId, 100); if (projected > 0) { this.log('info', 'webhook_summaries_projected', { organization_id: organizationId, projected_count: projected }, 'Leader task summaries projected for external webhook delivery'); } await this.dispatchPending(organizationId); } private async auditConfigurationChange( client: import('pg').PoolClient, organizationId: string, eventType: string, metadata: Record ): Promise { await client.query( `INSERT INTO audit_events (organization_id, actor_user_id, event_type, entity_type, entity_id, request_id, metadata) VALUES ($1::uuid, NULL, $2, 'leader_task_summary_webhook', $1::uuid::text, NULL, $3)`, [organizationId, eventType, metadata] ); } async reconcileWebhookState(organizationId: string): Promise { const configurationFingerprint = this.configurationFingerprint(); return withTransaction(this.config, async (client) => { await client.query( `SELECT pg_advisory_xact_lock( hashtextextended($1::text || ':leader-summary-webhook-config', 0) )`, [organizationId] ); const existingResult = await client.query( `SELECT * FROM leader_task_summary_webhook_state WHERE organization_id = $1 FOR UPDATE`, [organizationId] ); const existing = existingResult.rows[0] as Record | undefined; if (!configurationFingerprint) { if (!existing || !booleanValue(existing.enabled)) return false; const previousRevision = Number(existing.revision || 0); const nextRevision = previousRevision + 1; await client.query( `UPDATE leader_task_summary_webhook_state SET enabled = false, starts_at = now(), revision = $2, updated_at = now() WHERE organization_id = $1`, [organizationId, nextRevision] ); await client.query( `UPDATE leader_task_summary_webhook_deliveries SET delivery_status = CASE WHEN delivery_status = 'sending' THEN 'uncertain' ELSE 'cancelled' END, last_error = CASE WHEN delivery_status = 'sending' THEN 'Webhook 配置停用时请求状态未知,已停止自动重试。' ELSE 'Webhook 配置已停用,待发摘要已取消。' END, updated_at = now() WHERE organization_id = $1 AND webhook_revision = $2 AND delivery_status IN ('pending', 'sending')`, [organizationId, previousRevision] ); await this.auditConfigurationChange( client, organizationId, 'leader_summary_webhook.automatic_disabled', { revision: nextRevision, historical_backfill: false } ); return true; } if (!existing) { await client.query( `INSERT INTO leader_task_summary_webhook_state (organization_id, enabled, starts_at, revision, configuration_fingerprint) VALUES ($1, true, now(), 0, $2)`, [organizationId, configurationFingerprint] ); await this.auditConfigurationChange( client, organizationId, 'leader_summary_webhook.automatic_enabled', { revision: 0, endpoint_fingerprint: this.endpointFingerprint(), webhook_config_id: EXTERNAL_WEBHOOK_CONFIG_ID, historical_backfill: false } ); return true; } const sameConfiguration = text(existing.configuration_fingerprint) === configurationFingerprint; if (booleanValue(existing.enabled) && sameConfiguration) return false; const previousRevision = Number(existing.revision || 0); const nextRevision = previousRevision + 1; await client.query( `UPDATE leader_task_summary_webhook_state SET enabled = true, starts_at = now(), revision = $2, configuration_fingerprint = $3, updated_at = now() WHERE organization_id = $1`, [organizationId, nextRevision, configurationFingerprint] ); await client.query( `UPDATE leader_task_summary_webhook_deliveries SET delivery_status = CASE WHEN delivery_status = 'sending' THEN 'uncertain' ELSE 'cancelled' END, last_error = CASE WHEN delivery_status = 'sending' THEN 'Webhook 配置变化时请求状态未知,已停止自动重试。' ELSE 'Webhook 配置已变化,旧配置待发摘要已取消。' END, updated_at = now() WHERE organization_id = $1 AND webhook_revision = $2 AND delivery_status IN ('pending', 'sending')`, [organizationId, previousRevision] ); await this.auditConfigurationChange( client, organizationId, sameConfiguration ? 'leader_summary_webhook.automatic_reenabled' : 'leader_summary_webhook.configuration_changed', { revision: nextRevision, endpoint_fingerprint: this.endpointFingerprint(), webhook_config_id: EXTERNAL_WEBHOOK_CONFIG_ID, historical_backfill: false } ); return true; }); } async getWebhookStatus(organizationId: string): Promise { const result = await getPool(this.config).query( `SELECT state.*, COALESCE(delivery.pending_count, 0)::int AS pending_count, COALESCE(delivery.failed_count, 0)::int AS failed_count, COALESCE(delivery.uncertain_count, 0)::int AS uncertain_count, COALESCE(delivery.accepted_count, 0)::int AS accepted_count, delivery.last_accepted_at FROM leader_task_summary_webhook_state state LEFT JOIN LATERAL ( SELECT count(*) FILTER ( WHERE d.delivery_status IN ('pending', 'sending') ) AS pending_count, count(*) FILTER (WHERE d.delivery_status = 'failed') AS failed_count, count(*) FILTER (WHERE d.delivery_status = 'uncertain') AS uncertain_count, count(*) FILTER (WHERE d.delivery_status = 'accepted') AS accepted_count, max(d.accepted_at) AS last_accepted_at FROM leader_task_summary_webhook_deliveries d WHERE d.organization_id = state.organization_id AND d.webhook_revision = state.revision ) delivery ON true WHERE state.organization_id = $1`, [organizationId] ); const row = result.rows[0] as Record | undefined; const configurationFingerprint = this.configurationFingerprint(); const configured = Boolean(configurationFingerprint); const configurationMatches = Boolean( row && configurationFingerprint && text(row.configuration_fingerprint) === configurationFingerprint ); const enabled = configured && configurationMatches && booleanValue(row?.enabled); let state: PublicLeaderSummaryWebhookStatus['state']; let statusMessage: string; if (this.config.leaderSummaryWebhookConfigurationError) { state = 'invalid_configuration'; statusMessage = '外部 Webhook 配置无效,组长摘要已单独停用;普通任务与 AgentBus 服务不受影响。'; } else if (!configured) { state = 'not_configured'; statusMessage = '运行环境尚未同时配置 WEBHOOK_SEND_URL 与 WEBHOOK_EXTERNAL_TOKEN,摘要不会外发。'; } else if (!row) { state = 'initializing'; statusMessage = '外部 Webhook 配置正在初始化,暂时不会发送。'; } else if (!enabled) { state = 'configuration_changed'; statusMessage = '外部 Webhook 配置正在切换生效起点,暂时不会发送。'; } else { state = 'active'; statusMessage = '组长摘要将通过固定外部 Webhook 推送;成功仅表示接口已受理,不代表微信已经送达。'; } return { delivery_mode: 'external_webhook', webhook_config_id: EXTERNAL_WEBHOOK_CONFIG_ID, configured, enabled, state, configuration_error: this.config.leaderSummaryWebhookConfigurationError, status_message: statusMessage, endpoint_fingerprint: this.endpointFingerprint(), starts_at: iso(row?.starts_at), revision: row ? Number(row.revision || 0) : null, pending_count: Number(row?.pending_count || 0), failed_count: Number(row?.failed_count || 0), uncertain_count: Number(row?.uncertain_count || 0), accepted_count: Number(row?.accepted_count || 0), last_accepted_at: iso(row?.last_accepted_at), updated_at: iso(row?.updated_at) }; } async projectPending(organizationId: string, limit = 100): Promise { const configurationFingerprint = this.configurationFingerprint(); if (!configurationFingerprint) return 0; const boundedLimit = Math.max(1, Math.min(500, Math.trunc(limit))); return withTransaction(this.config, async (client) => { const lock = await client.query( `SELECT pg_try_advisory_xact_lock( hashtextextended($1::text || ':leader-summary-webhook-projector', 0) ) AS acquired`, [organizationId] ); if (!booleanValue(lock.rows[0]?.acquired)) return 0; const candidates = await client.query( `SELECT state.revision AS webhook_revision, task.id AS task_row_id, task.task_id, task.status, task.business_route_id, task.success_receipt, task.created_at, assignee.username AS assignee_username, source_event.id AS source_outbox_event_id, COALESCE(delivery_state.has_exposed_needs_review, false) AS has_exposed_needs_review FROM leader_task_summary_webhook_state state JOIN tasks task ON task.organization_id = state.organization_id AND task.assigned_user_id IS NOT NULL AND task.source IN ('manual', 'agentbus') JOIN users assignee ON assignee.id = task.assigned_user_id AND assignee.organization_id = task.organization_id AND assignee.role <> 'admin' JOIN LATERAL ( SELECT event.id FROM outbox_events event WHERE event.organization_id = state.organization_id AND event.topic = 'task.updated' AND event.aggregate_type = 'task' AND event.aggregate_id = task.task_id AND event.created_at >= state.starts_at AND event.payload ->> 'status' = ANY($3::text[]) AND (event.payload ->> 'archived') IS DISTINCT FROM 'true' AND (event.payload ->> 'restored') IS DISTINCT FROM 'true' ORDER BY event.id DESC LIMIT 1 ) source_event ON true LEFT JOIN LATERAL ( SELECT bool_or( delivery.milestone = 'needs_review' AND delivery.delivery_status <> 'cancelled' ) AS has_needs_review, bool_or( delivery.milestone = 'needs_review' AND delivery.delivery_status IN ('sending', 'accepted', 'uncertain') ) AS has_exposed_needs_review, bool_or( delivery.milestone = 'final' AND delivery.delivery_status <> 'cancelled' ) AS has_final, bool_or( delivery.milestone = 'resolved' AND delivery.delivery_status <> 'cancelled' ) AS has_resolved FROM leader_task_summary_webhook_deliveries delivery WHERE delivery.organization_id = state.organization_id AND delivery.webhook_revision = state.revision AND delivery.task_id = task.id ) delivery_state ON true WHERE state.organization_id = $1 AND state.enabled = true AND state.configuration_fingerprint = $2 AND task.status = ANY($3::text[]) AND ( (task.status = ANY($4::text[]) AND NOT COALESCE(delivery_state.has_needs_review, false)) OR (NOT (task.status = ANY($4::text[])) AND ( (COALESCE(delivery_state.has_exposed_needs_review, false) AND NOT COALESCE(delivery_state.has_resolved, false)) OR (NOT COALESCE(delivery_state.has_exposed_needs_review, false) AND NOT COALESCE(delivery_state.has_final, false)) )) ) ORDER BY source_event.id ASC, task.id LIMIT $5`, [ organizationId, configurationFingerprint, [...PROJECTABLE_STATUSES], [...NEEDS_REVIEW_STATUSES], boundedLimit ] ); let projected = 0; for (const row of candidates.rows as Record[]) { if (!isLeaderTaskSummaryStatus(row.status)) continue; const currentNeedsReview = (NEEDS_REVIEW_STATUSES as readonly string[]).includes(text(row.status)); let hadExposedNeedsReview = booleanValue(row.has_exposed_needs_review); if (!currentNeedsReview) { await client.query( `UPDATE leader_task_summary_webhook_deliveries SET delivery_status = 'cancelled', last_error = '任务已形成确定结果,未发送的旧核验提醒已取消。', updated_at = now() WHERE organization_id = $1 AND webhook_revision = $2 AND task_id = $3 AND milestone = 'needs_review' AND delivery_status = 'pending'`, [organizationId, row.webhook_revision, row.task_row_id] ); const exposure = await client.query( `SELECT EXISTS ( SELECT 1 FROM leader_task_summary_webhook_deliveries WHERE organization_id = $1 AND webhook_revision = $2 AND task_id = $3 AND milestone = 'needs_review' AND delivery_status IN ('sending', 'accepted', 'uncertain') ) AS exposed`, [organizationId, row.webhook_revision, row.task_row_id] ); hadExposedNeedsReview = booleanValue(exposure.rows[0]?.exposed); } const projection = buildLeaderTaskSummary({ taskId: text(row.task_id), status: text(row.status), businessRouteId: text(row.business_route_id) || null, assigneeUsername: text(row.assignee_username), createdAt: String(row.created_at), successReceipt: jsonObject(row.success_receipt), hadNeedsReview: hadExposedNeedsReview }); if (!projection) continue; const payloadText = projection.messageText; const inserted = await client.query( `INSERT INTO leader_task_summary_webhook_deliveries (organization_id, webhook_revision, task_id, source_outbox_event_id, milestone, payload_ciphertext, payload_fingerprint) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (organization_id, webhook_revision, task_id, milestone) DO NOTHING RETURNING id`, [ organizationId, row.webhook_revision, row.task_row_id, row.source_outbox_event_id, projection.milestone, encryptText(this.config, payloadText), sha256Text(payloadText) ] ); projected += Number(inserted.rowCount || 0); } return projected; }); } private async claimPendingDeliveries( organizationId: string, limit = 1 ): Promise { const configurationFingerprint = this.configurationFingerprint(); if (!configurationFingerprint) return []; const boundedLimit = Math.max(1, Math.min(5, Math.trunc(limit))); return withTransaction(this.config, async (client) => { await client.query( `SELECT pg_advisory_xact_lock( hashtextextended($1::text || ':leader-summary-webhook-dispatcher', 0) )`, [organizationId] ); await client.query( `UPDATE leader_task_summary_webhook_deliveries SET delivery_status = 'uncertain', last_error = '发送租约过期,接口是否已受理未知,已停止自动重试。', updated_at = now() WHERE organization_id = $1 AND delivery_status = 'sending' AND updated_at < now() - interval '1 minute'`, [organizationId] ); const pending = await client.query( `SELECT delivery.*, task.task_id AS public_task_id FROM leader_task_summary_webhook_deliveries delivery JOIN leader_task_summary_webhook_state state ON state.organization_id = delivery.organization_id AND state.enabled = true AND state.revision = delivery.webhook_revision AND state.configuration_fingerprint = $2 JOIN tasks task ON task.id = delivery.task_id AND task.organization_id = delivery.organization_id WHERE delivery.organization_id = $1 AND delivery.delivery_status = 'pending' AND delivery.attempt_count = 0 ORDER BY delivery.created_at ASC, delivery.id ASC FOR UPDATE OF delivery SKIP LOCKED LIMIT $3`, [organizationId, configurationFingerprint, boundedLimit] ); const deliveries: ClaimedWebhookDelivery[] = []; for (const row of pending.rows as Record[]) { try { const content = decryptText(this.config, text(row.payload_ciphertext)).trim(); if (!content || sha256Text(content) !== text(row.payload_fingerprint)) { throw new Error('invalid leader summary webhook payload'); } const updated = await client.query( `UPDATE leader_task_summary_webhook_deliveries SET delivery_status = 'sending', attempt_count = attempt_count + 1, last_error = $2, updated_at = now() WHERE id = $1 AND delivery_status = 'pending' AND attempt_count = 0 RETURNING attempt_count`, [row.id, `sending:${this.leaseOwner}`] ); if (!updated.rowCount) continue; deliveries.push({ id: text(row.id), taskId: text(row.public_task_id), content, payloadFingerprint: text(row.payload_fingerprint).slice(0, 12), attemptCount: Number(updated.rows[0]?.attempt_count || 1) }); } catch (error) { await client.query( `UPDATE leader_task_summary_webhook_deliveries SET delivery_status = 'cancelled', last_error = '摘要投递密文或结构无效,已停止发送。', updated_at = now() WHERE id = $1`, [row.id] ); this.log('error', 'webhook_delivery_invalid', { delivery_id: text(row.id), task_id: text(row.public_task_id), ...diagnosticError(error, 'leader_summary_webhook_delivery_invalid') }, 'Leader task summary webhook delivery is invalid'); } } return deliveries; }); } private async markDeliveryOutcome( deliveryId: string, result: ExternalWebhookSendResult ): Promise { const deliveryStatus = result.outcome === 'accepted' ? 'accepted' : result.outcome === 'rejected' ? 'failed' : 'uncertain'; await getPool(this.config).query( `UPDATE leader_task_summary_webhook_deliveries SET delivery_status = $2, response_status = $3, last_error = $4, accepted_at = CASE WHEN $2 = 'accepted' THEN now() ELSE NULL END, updated_at = now() WHERE id = $1 AND delivery_status = 'sending'`, [deliveryId, deliveryStatus, result.httpStatus, result.errorCode] ); } private async dispatchPending(organizationId: string): Promise { if (!this.sender) return; const deliveries = await this.claimPendingDeliveries(organizationId, 1); for (const delivery of deliveries) { this.log('info', 'webhook_delivery_sending', { organization_id: organizationId, delivery_id: delivery.id, task_id: delivery.taskId, payload_fingerprint: delivery.payloadFingerprint, attempt_count: delivery.attemptCount }, 'Leader task summary sending to external webhook'); let result: ExternalWebhookSendResult; try { result = await this.sender.send(delivery.content); } catch { result = { outcome: 'uncertain', httpStatus: null, errorCode: 'webhook_sender_failed' }; } await this.markDeliveryOutcome(delivery.id, result); const metadata = { organization_id: organizationId, delivery_id: delivery.id, task_id: delivery.taskId, payload_fingerprint: delivery.payloadFingerprint, http_status: result.httpStatus, outcome: result.outcome, error_code: result.errorCode }; if (result.outcome === 'accepted') { this.log('info', 'webhook_delivery_accepted', metadata, 'Leader task summary accepted by external webhook'); } else if (result.outcome === 'rejected') { this.log('warn', 'webhook_delivery_rejected', metadata, 'Leader task summary rejected by external webhook'); } else { this.log('warn', 'webhook_delivery_uncertain', metadata, 'Leader task summary webhook result is uncertain; automatic retry disabled'); } } } }