forked from wangziqi/gongxue-base
feat: add platform audit alert notifications
This commit is contained in:
Codex
@@ -14,6 +14,7 @@
|
||||
"platform-billing:once": "tsx src/index.ts --once --job platform-billing",
|
||||
"platform-dunning:once": "tsx src/index.ts --once --job platform-dunning",
|
||||
"platform-audit-alerts:once": "tsx src/index.ts --once --job platform-audit-alerts",
|
||||
"platform-audit-notifications:once": "tsx src/index.ts --once --job platform-audit-notifications",
|
||||
"assets:once": "tsx src/index.ts --once --job assets",
|
||||
"imports:once": "tsx src/index.ts --once --job imports",
|
||||
"public-banks:once": "tsx src/index.ts --once --job public-banks",
|
||||
|
||||
@@ -27,6 +27,11 @@ export interface WorkerConfig {
|
||||
platformAuditAlertBatchSize: number;
|
||||
platformAuditAlertWorkerId: string;
|
||||
platformAuditAlertLookbackDays: number;
|
||||
platformAuditNotificationBatchSize: number;
|
||||
platformAuditNotificationMaxAttempts: number;
|
||||
platformAuditNotificationBackoffSeconds: number[];
|
||||
platformAuditNotificationRequestTimeoutMs: number;
|
||||
platformAuditNotificationAllowInsecureLocalhost: boolean;
|
||||
assetBatchSize: number;
|
||||
assetMinAgeSeconds: number;
|
||||
assetRecheckIntervalSeconds: number;
|
||||
@@ -111,6 +116,9 @@ function validateProductionConfig(nextConfig: WorkerConfig) {
|
||||
if (nextConfig.assetSecurityScanFailOpen) {
|
||||
failures.push('WORKER_ASSET_SECURITY_SCAN_FAIL_OPEN=true is not allowed in production workers');
|
||||
}
|
||||
if (nextConfig.platformAuditNotificationAllowInsecureLocalhost) {
|
||||
failures.push('WORKER_PLATFORM_AUDIT_NOTIFICATION_ALLOW_INSECURE_LOCALHOST=true is not allowed in production workers');
|
||||
}
|
||||
const scannerModes = nextConfig.assetSecurityScanner
|
||||
.split(',')
|
||||
.map(item => item.trim().toLowerCase())
|
||||
@@ -184,6 +192,13 @@ const loadedConfig: WorkerConfig = {
|
||||
platformAuditAlertBatchSize: envNumber('WORKER_PLATFORM_AUDIT_ALERT_BATCH_SIZE', 200),
|
||||
platformAuditAlertWorkerId: envString('WORKER_PLATFORM_AUDIT_ALERT_ID', `platform-audit-alerts-${process.pid}`),
|
||||
platformAuditAlertLookbackDays: envNumber('WORKER_PLATFORM_AUDIT_ALERT_LOOKBACK_DAYS', 14),
|
||||
platformAuditNotificationBatchSize: envNumber('WORKER_PLATFORM_AUDIT_NOTIFICATION_BATCH_SIZE', 50),
|
||||
platformAuditNotificationMaxAttempts: envNumber('WORKER_PLATFORM_AUDIT_NOTIFICATION_MAX_ATTEMPTS', 5),
|
||||
platformAuditNotificationBackoffSeconds: envList('WORKER_PLATFORM_AUDIT_NOTIFICATION_BACKOFF_SECONDS', '10,60,300,900,1800')
|
||||
.map((value: string) => Number(value))
|
||||
.filter((value: number) => Number.isFinite(value) && value > 0),
|
||||
platformAuditNotificationRequestTimeoutMs: envNumber('WORKER_PLATFORM_AUDIT_NOTIFICATION_REQUEST_TIMEOUT_MS', 10_000),
|
||||
platformAuditNotificationAllowInsecureLocalhost: envBoolean('WORKER_PLATFORM_AUDIT_NOTIFICATION_ALLOW_INSECURE_LOCALHOST', false),
|
||||
assetBatchSize: envNumber('WORKER_ASSET_BATCH_SIZE', 50),
|
||||
assetMinAgeSeconds: envNumber('WORKER_ASSET_MIN_AGE_SECONDS', 300),
|
||||
assetRecheckIntervalSeconds: envNumber('WORKER_ASSET_RECHECK_INTERVAL_SECONDS', 60 * 60 * 24),
|
||||
|
||||
@@ -70,6 +70,16 @@ async function runOnce() {
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-audit-notifications') {
|
||||
const { processPlatformAuditNotificationBatch } = await import('./jobs/platform-audit-notifications.js');
|
||||
const result = await processPlatformAuditNotificationBatch();
|
||||
console.log(
|
||||
`[worker] platform-audit-notifications batch enqueued=${result.enqueued}`
|
||||
+ ` processed=${result.processed} sent=${result.sent} failed=${result.failed}`
|
||||
+ ` retrying=${result.retrying} discarded=${result.discarded}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'assets') {
|
||||
const result = await processAssetBatch();
|
||||
console.log(
|
||||
|
||||
612
apps/worker/src/jobs/platform-audit-notifications.ts
Normal file
612
apps/worker/src/jobs/platform-audit-notifications.ts
Normal file
@@ -0,0 +1,612 @@
|
||||
import crypto from 'node:crypto';
|
||||
import type pg from 'pg';
|
||||
import { pool } from '../db.js';
|
||||
import { config } from '../config.js';
|
||||
|
||||
const PROVIDERS = ['generic', 'dingtalk', 'feishu', 'wecom'] as const;
|
||||
type NotificationProvider = typeof PROVIDERS[number];
|
||||
|
||||
interface NotificationChannelRow {
|
||||
id: string;
|
||||
channelCode: string;
|
||||
name: string;
|
||||
provider: NotificationProvider;
|
||||
webhookUrl: string;
|
||||
secretRef: string | null;
|
||||
minSeverity: string;
|
||||
statusFilter: string[];
|
||||
actionPatterns: string[];
|
||||
tenantIds: string[];
|
||||
timeoutSec: number | null;
|
||||
}
|
||||
|
||||
interface NotificationEventRow {
|
||||
id: string;
|
||||
channelId: string;
|
||||
channelCode: string;
|
||||
channelName: string;
|
||||
alertId: string;
|
||||
auditLogId: string | null;
|
||||
provider: NotificationProvider;
|
||||
attempts: number;
|
||||
webhookUrl: string;
|
||||
secretRef: string | null;
|
||||
timeoutSec: number | null;
|
||||
tenantId: string | null;
|
||||
tenantSlug: string | null;
|
||||
tenantName: string | null;
|
||||
severity: string;
|
||||
alertStatus: string;
|
||||
action: string;
|
||||
targetType: string | null;
|
||||
targetId: string | null;
|
||||
title: string;
|
||||
summary: string | null;
|
||||
details: Record<string, unknown> | null;
|
||||
alertCreatedAt: string;
|
||||
}
|
||||
|
||||
interface SecretRow {
|
||||
secretValue: string | null;
|
||||
secretJson: Record<string, unknown> | null;
|
||||
}
|
||||
|
||||
interface PreparedRequest {
|
||||
provider: NotificationProvider;
|
||||
url: string;
|
||||
body: Record<string, unknown>;
|
||||
headers: Record<string, string>;
|
||||
}
|
||||
|
||||
interface SendResult {
|
||||
ok: boolean;
|
||||
httpCode: number;
|
||||
responseSummary: string;
|
||||
errorMessage: string;
|
||||
}
|
||||
|
||||
interface ProcessResult {
|
||||
enqueued: number;
|
||||
processed: number;
|
||||
sent: number;
|
||||
failed: number;
|
||||
retrying: number;
|
||||
discarded: number;
|
||||
}
|
||||
|
||||
function objectValue(value: unknown): Record<string, unknown> {
|
||||
return value && typeof value === 'object' && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
}
|
||||
|
||||
function stringValue(value: unknown, fallback = '') {
|
||||
return typeof value === 'string' && value.trim() ? value.trim() : fallback;
|
||||
}
|
||||
|
||||
function truncate(value: unknown, max = 1900) {
|
||||
return String(value ?? '').slice(0, max);
|
||||
}
|
||||
|
||||
function severityRank(value: string) {
|
||||
if (value === 'critical') return 4;
|
||||
if (value === 'high') return 3;
|
||||
if (value === 'medium') return 2;
|
||||
return 1;
|
||||
}
|
||||
|
||||
function redactAuditNotificationValue(value: unknown, parentKey = '', depth = 0): unknown {
|
||||
if (value === null || value === undefined) return value;
|
||||
if (depth > 8) return '[REDACTED_DEPTH_LIMIT]';
|
||||
if (
|
||||
/(?:password|passwd|secret|token|credential|private[_-]?key|api[_-]?key|app[_-]?secret|authorization|cookie|session|cert|signature|nonce)$/i
|
||||
.test(parentKey)
|
||||
) {
|
||||
return '[REDACTED]';
|
||||
}
|
||||
if (Array.isArray(value)) return value.map(item => redactAuditNotificationValue(item, parentKey, depth + 1));
|
||||
if (typeof value === 'object') {
|
||||
const output: Record<string, unknown> = {};
|
||||
for (const [key, item] of Object.entries(value as Record<string, unknown>)) {
|
||||
output[key] = redactAuditNotificationValue(item, key, depth + 1);
|
||||
}
|
||||
return output;
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function matchesPattern(value: string, patterns: string[]) {
|
||||
if (!patterns.length) return true;
|
||||
return patterns.some(pattern => (
|
||||
value === pattern
|
||||
|| (pattern.endsWith('*') && value.startsWith(pattern.slice(0, -1)))
|
||||
));
|
||||
}
|
||||
|
||||
function parseSecretRef(ref: string | null) {
|
||||
if (!ref) return null;
|
||||
const parts = ref.split(':');
|
||||
if (parts.length !== 3 || parts[0] !== 'app_private.platform_secrets') return null;
|
||||
return { scope: parts[1], key: parts[2] };
|
||||
}
|
||||
|
||||
function secretText(secret: SecretRow | null, keys: string[]) {
|
||||
if (!secret) return '';
|
||||
if (secret.secretValue?.trim()) return secret.secretValue.trim();
|
||||
const json = objectValue(secret.secretJson);
|
||||
for (const key of keys) {
|
||||
const value = json[key];
|
||||
if (typeof value === 'string' && value.trim()) return value.trim();
|
||||
}
|
||||
return '';
|
||||
}
|
||||
|
||||
function validateWebhookUrl(rawUrl: string) {
|
||||
let url: URL;
|
||||
try {
|
||||
url = new URL(rawUrl);
|
||||
} catch {
|
||||
throw new Error('Platform audit notification webhook URL is invalid');
|
||||
}
|
||||
const isLocalhost = ['127.0.0.1', 'localhost', '::1'].includes(url.hostname);
|
||||
if (url.protocol !== 'https:' && !(config.platformAuditNotificationAllowInsecureLocalhost && isLocalhost)) {
|
||||
throw new Error('Platform audit notification webhook URL must use HTTPS outside local development');
|
||||
}
|
||||
url.username = '';
|
||||
url.password = '';
|
||||
return url;
|
||||
}
|
||||
|
||||
function dingtalkSign(secret: string): Record<string, string> {
|
||||
if (!secret) return {};
|
||||
const timestamp = String(Date.now());
|
||||
const sign = crypto
|
||||
.createHmac('sha256', secret)
|
||||
.update(`${timestamp}\n${secret}`)
|
||||
.digest('base64');
|
||||
return { timestamp, sign };
|
||||
}
|
||||
|
||||
function feishuSign(secret: string): Record<string, string> {
|
||||
if (!secret) return {};
|
||||
const timestamp = String(Math.floor(Date.now() / 1000));
|
||||
const sign = crypto
|
||||
.createHmac('sha256', `${timestamp}\n${secret}`)
|
||||
.update('')
|
||||
.digest('base64');
|
||||
return { timestamp, sign };
|
||||
}
|
||||
|
||||
function appendQuery(url: URL, params: Record<string, string>) {
|
||||
for (const [key, value] of Object.entries(params)) {
|
||||
if (value) url.searchParams.set(key, value);
|
||||
}
|
||||
return url.toString();
|
||||
}
|
||||
|
||||
function alertPayload(task: NotificationEventRow) {
|
||||
return {
|
||||
event: 'platform.audit.alert',
|
||||
alert: {
|
||||
id: task.alertId,
|
||||
severity: task.severity,
|
||||
status: task.alertStatus,
|
||||
title: task.title,
|
||||
summary: task.summary,
|
||||
action: task.action,
|
||||
targetType: task.targetType,
|
||||
targetId: task.targetId,
|
||||
createdAt: task.alertCreatedAt,
|
||||
details: redactAuditNotificationValue(task.details || {}),
|
||||
},
|
||||
tenant: task.tenantId
|
||||
? {
|
||||
id: task.tenantId,
|
||||
slug: task.tenantSlug,
|
||||
name: task.tenantName,
|
||||
}
|
||||
: null,
|
||||
source: {
|
||||
channelId: task.channelId,
|
||||
channelCode: task.channelCode,
|
||||
auditLogId: task.auditLogId,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function alertMarkdown(task: NotificationEventRow) {
|
||||
const payload = alertPayload(task);
|
||||
const tenant = payload.tenant;
|
||||
return [
|
||||
`### 平台审计告警：${task.title}`,
|
||||
`- 级别：${task.severity}`,
|
||||
`- 状态：${task.alertStatus}`,
|
||||
`- 租户：${tenant?.name || tenant?.slug || '平台'}`,
|
||||
`- 动作：${task.action}`,
|
||||
`- 目标：${[task.targetType, task.targetId].filter(Boolean).join(':') || '无'}`,
|
||||
`- 摘要：${task.summary || '无'}`,
|
||||
`- 告警ID：${task.alertId}`,
|
||||
`- 时间：${String(task.alertCreatedAt).slice(0, 19).replace('T', ' ')}`,
|
||||
].join('\n');
|
||||
}
|
||||
|
||||
function prepareRequest(task: NotificationEventRow, secret: SecretRow | null): PreparedRequest {
|
||||
const target = validateWebhookUrl(task.webhookUrl);
|
||||
const provider = task.provider;
|
||||
const secretValue = secretText(secret, ['secret', 'signSecret', 'webhookSecret']);
|
||||
const headers = { 'content-type': 'application/json' };
|
||||
const markdown = alertMarkdown(task);
|
||||
|
||||
if (provider === 'dingtalk') {
|
||||
return {
|
||||
provider,
|
||||
url: appendQuery(target, dingtalkSign(secretValue)),
|
||||
headers,
|
||||
body: {
|
||||
msgtype: 'markdown',
|
||||
markdown: {
|
||||
title: '平台审计告警',
|
||||
text: markdown,
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
if (provider === 'feishu') {
|
||||
return {
|
||||
provider,
|
||||
url: target.toString(),
|
||||
headers,
|
||||
body: {
|
||||
msg_type: 'interactive',
|
||||
...feishuSign(secretValue),
|
||||
card: {
|
||||
config: { wide_screen_mode: true },
|
||||
header: { title: { tag: 'plain_text', content: '平台审计告警' }, template: 'red' },
|
||||
elements: [{ tag: 'markdown', content: markdown }],
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
if (provider === 'wecom') {
|
||||
return {
|
||||
provider,
|
||||
url: target.toString(),
|
||||
headers,
|
||||
body: {
|
||||
msgtype: 'markdown',
|
||||
markdown: { content: markdown },
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
provider,
|
||||
url: target.toString(),
|
||||
headers,
|
||||
body: alertPayload(task),
|
||||
};
|
||||
}
|
||||
|
||||
async function loadEnabledChannels(client: pg.PoolClient) {
|
||||
const result = await client.query<NotificationChannelRow>(
|
||||
`
|
||||
select id, channel_code as "channelCode", name, provider,
|
||||
webhook_url as "webhookUrl", secret_ref as "secretRef",
|
||||
min_severity as "minSeverity", status_filter as "statusFilter",
|
||||
action_patterns as "actionPatterns", tenant_ids as "tenantIds",
|
||||
timeout_sec as "timeoutSec"
|
||||
from public.platform_audit_notification_channels
|
||||
where enabled = true
|
||||
order by created_at asc
|
||||
`,
|
||||
);
|
||||
return result.rows;
|
||||
}
|
||||
|
||||
async function loadOpenAlerts(client: pg.PoolClient, limit: number) {
|
||||
const result = await client.query<{
|
||||
id: string;
|
||||
auditLogId: string;
|
||||
tenantId: string | null;
|
||||
severity: string;
|
||||
status: string;
|
||||
action: string;
|
||||
}>(
|
||||
`
|
||||
select id, audit_log_id as "auditLogId", tenant_id as "tenantId",
|
||||
severity, status, action
|
||||
from public.platform_audit_alerts
|
||||
where status in ('open', 'acknowledged')
|
||||
order by
|
||||
case severity when 'critical' then 1 when 'high' then 2 when 'medium' then 3 else 4 end,
|
||||
created_at asc
|
||||
limit $1
|
||||
`,
|
||||
[limit],
|
||||
);
|
||||
return result.rows;
|
||||
}
|
||||
|
||||
async function enqueueNotificationEvents(client: pg.PoolClient, limit: number) {
|
||||
const channels = await loadEnabledChannels(client);
|
||||
if (!channels.length) return 0;
|
||||
const alerts = await loadOpenAlerts(client, Math.max(limit * 4, 50));
|
||||
let enqueued = 0;
|
||||
|
||||
for (const alert of alerts) {
|
||||
for (const channel of channels) {
|
||||
if (enqueued >= limit) return enqueued;
|
||||
if (!channel.statusFilter.includes(alert.status)) continue;
|
||||
if (severityRank(alert.severity) < severityRank(channel.minSeverity)) continue;
|
||||
if (!matchesPattern(alert.action, channel.actionPatterns)) continue;
|
||||
if (channel.tenantIds.length && (!alert.tenantId || !channel.tenantIds.includes(alert.tenantId))) continue;
|
||||
|
||||
const result = await client.query(
|
||||
`
|
||||
insert into public.platform_audit_notification_events (
|
||||
channel_id, alert_id, audit_log_id, provider, status,
|
||||
request_payload, metadata
|
||||
)
|
||||
values ($1, $2, $3, $4, 'pending', $5::jsonb, $6::jsonb)
|
||||
on conflict (channel_id, alert_id) do nothing
|
||||
returning id
|
||||
`,
|
||||
[
|
||||
channel.id,
|
||||
alert.id,
|
||||
alert.auditLogId,
|
||||
channel.provider,
|
||||
JSON.stringify({
|
||||
channelCode: channel.channelCode,
|
||||
alertId: alert.id,
|
||||
severity: alert.severity,
|
||||
action: alert.action,
|
||||
}),
|
||||
JSON.stringify({ enqueuedBy: 'platform-audit-notifications-worker' }),
|
||||
],
|
||||
);
|
||||
if (result.rowCount) enqueued += 1;
|
||||
}
|
||||
}
|
||||
return enqueued;
|
||||
}
|
||||
|
||||
async function claimDueEvents(client: pg.PoolClient, limit: number) {
|
||||
const result = await client.query<NotificationEventRow>(
|
||||
`
|
||||
with due as (
|
||||
select e.id
|
||||
from public.platform_audit_notification_events e
|
||||
where e.status in ('pending', 'retrying')
|
||||
and coalesce(e.next_attempt_at, e.scheduled_at, e.created_at) <= now()
|
||||
order by coalesce(e.next_attempt_at, e.scheduled_at, e.created_at) asc, e.created_at asc
|
||||
limit $1
|
||||
for update skip locked
|
||||
)
|
||||
update public.platform_audit_notification_events e
|
||||
set status = 'processing',
|
||||
last_attempt_at = now(),
|
||||
updated_at = now()
|
||||
from due
|
||||
join public.platform_audit_notification_channels c on true
|
||||
join public.platform_audit_alerts a on true
|
||||
left join public.tenants t on t.id = a.tenant_id
|
||||
where e.id = due.id
|
||||
and c.id = e.channel_id
|
||||
and a.id = e.alert_id
|
||||
returning e.id, e.channel_id as "channelId", c.channel_code as "channelCode",
|
||||
c.name as "channelName", e.alert_id as "alertId",
|
||||
e.audit_log_id as "auditLogId", e.provider, e.attempts,
|
||||
c.webhook_url as "webhookUrl", c.secret_ref as "secretRef",
|
||||
c.timeout_sec as "timeoutSec", a.tenant_id as "tenantId",
|
||||
t.slug::text as "tenantSlug", t.name as "tenantName",
|
||||
a.severity, a.status as "alertStatus", a.action,
|
||||
a.target_type as "targetType", a.target_id as "targetId",
|
||||
a.title, a.summary, a.details,
|
||||
a.created_at as "alertCreatedAt"
|
||||
`,
|
||||
[limit],
|
||||
);
|
||||
return result.rows;
|
||||
}
|
||||
|
||||
async function recoverStaleProcessingEvents(client: pg.PoolClient) {
|
||||
const staleMs = Math.max(config.platformAuditNotificationRequestTimeoutMs * 3, 60_000);
|
||||
await client.query(
|
||||
`
|
||||
update public.platform_audit_notification_events
|
||||
set status = 'retrying',
|
||||
next_attempt_at = now(),
|
||||
last_error = coalesce(last_error, 'Recovered stale processing platform audit notification'),
|
||||
updated_at = now()
|
||||
where status = 'processing'
|
||||
and coalesce(last_attempt_at, updated_at, created_at) < now() - ($1::int * interval '1 millisecond')
|
||||
`,
|
||||
[staleMs],
|
||||
);
|
||||
}
|
||||
|
||||
async function loadSecret(client: pg.PoolClient, secretRef: string | null) {
|
||||
const parsed = parseSecretRef(secretRef);
|
||||
if (!parsed) return null;
|
||||
const result = await client.query<SecretRow>(
|
||||
`
|
||||
select secret_value as "secretValue", secret_json as "secretJson"
|
||||
from app_private.platform_secrets
|
||||
where secret_scope = $1
|
||||
and secret_key = $2
|
||||
limit 1
|
||||
`,
|
||||
[parsed.scope, parsed.key],
|
||||
);
|
||||
return result.rows[0] || null;
|
||||
}
|
||||
|
||||
async function sendWebhook(request: PreparedRequest, timeoutMs: number): Promise<SendResult> {
|
||||
const controller = new AbortController();
|
||||
const timeout = setTimeout(() => controller.abort(), timeoutMs);
|
||||
try {
|
||||
const response = await fetch(request.url, {
|
||||
method: 'POST',
|
||||
headers: request.headers,
|
||||
body: JSON.stringify(request.body),
|
||||
signal: controller.signal,
|
||||
});
|
||||
const text = await response.text().catch(() => '');
|
||||
return {
|
||||
ok: response.ok,
|
||||
httpCode: response.status,
|
||||
responseSummary: truncate(text, 1900),
|
||||
errorMessage: response.ok ? '' : `Platform audit notification webhook returned HTTP ${response.status}`,
|
||||
};
|
||||
} catch (error) {
|
||||
return {
|
||||
ok: false,
|
||||
httpCode: 0,
|
||||
responseSummary: '',
|
||||
errorMessage: error instanceof Error ? error.message : String(error),
|
||||
};
|
||||
} finally {
|
||||
clearTimeout(timeout);
|
||||
}
|
||||
}
|
||||
|
||||
async function markEventResult(
|
||||
client: pg.PoolClient,
|
||||
task: NotificationEventRow,
|
||||
request: PreparedRequest | null,
|
||||
result: SendResult,
|
||||
attempt: number,
|
||||
) {
|
||||
const backoff = config.platformAuditNotificationBackoffSeconds[Math.min(attempt - 1, config.platformAuditNotificationBackoffSeconds.length - 1)] || 60;
|
||||
const requestPayload = {
|
||||
provider: request?.provider || task.provider,
|
||||
urlHost: request ? new URL(request.url).hostname : null,
|
||||
body: request?.body ? redactAuditNotificationValue(request.body) : null,
|
||||
};
|
||||
|
||||
if (result.ok) {
|
||||
await client.query(
|
||||
`
|
||||
update public.platform_audit_notification_events
|
||||
set status = 'sent',
|
||||
attempts = $2,
|
||||
next_attempt_at = null,
|
||||
last_error = null,
|
||||
last_http_code = $3,
|
||||
last_response_summary = $4,
|
||||
request_payload = $5::jsonb,
|
||||
sent_at = now(),
|
||||
updated_at = now()
|
||||
where id = $1
|
||||
`,
|
||||
[task.id, attempt, result.httpCode, truncate(result.responseSummary), JSON.stringify(requestPayload)],
|
||||
);
|
||||
return 'sent';
|
||||
}
|
||||
|
||||
const terminal = attempt >= config.platformAuditNotificationMaxAttempts;
|
||||
await client.query(
|
||||
`
|
||||
update public.platform_audit_notification_events
|
||||
set status = $2,
|
||||
attempts = $3,
|
||||
next_attempt_at = case when $2 = 'retrying' then now() + ($4::int * interval '1 second') else null end,
|
||||
last_error = $5,
|
||||
last_http_code = $6,
|
||||
last_response_summary = $7,
|
||||
request_payload = $8::jsonb,
|
||||
updated_at = now()
|
||||
where id = $1
|
||||
`,
|
||||
[
|
||||
task.id,
|
||||
terminal ? 'failed' : 'retrying',
|
||||
attempt,
|
||||
backoff,
|
||||
truncate(result.errorMessage),
|
||||
result.httpCode,
|
||||
truncate(result.responseSummary),
|
||||
JSON.stringify(requestPayload),
|
||||
],
|
||||
);
|
||||
return terminal ? 'failed' : 'retrying';
|
||||
}
|
||||
|
||||
async function discardEvent(client: pg.PoolClient, task: NotificationEventRow, message: string) {
|
||||
await client.query(
|
||||
`
|
||||
update public.platform_audit_notification_events
|
||||
set status = 'discarded',
|
||||
attempts = attempts + 1,
|
||||
last_attempt_at = now(),
|
||||
last_error = $2,
|
||||
request_payload = $3::jsonb,
|
||||
updated_at = now()
|
||||
where id = $1
|
||||
`,
|
||||
[
|
||||
task.id,
|
||||
truncate(message),
|
||||
JSON.stringify({
|
||||
provider: task.provider,
|
||||
urlHost: '',
|
||||
body: null,
|
||||
}),
|
||||
],
|
||||
);
|
||||
return 'discarded';
|
||||
}
|
||||
|
||||
async function processEvent(task: NotificationEventRow) {
|
||||
const client = await pool.connect();
|
||||
try {
|
||||
if (!PROVIDERS.includes(task.provider)) {
|
||||
return discardEvent(client, task, 'Unsupported platform audit notification provider');
|
||||
}
|
||||
const secret = await loadSecret(client, task.secretRef);
|
||||
const request = prepareRequest(task, secret);
|
||||
const timeoutMs = Math.max(1000, (task.timeoutSec || 0) * 1000 || config.platformAuditNotificationRequestTimeoutMs);
|
||||
const attempt = task.attempts + 1;
|
||||
const result = await sendWebhook(request, timeoutMs);
|
||||
return markEventResult(client, task, request, result, attempt);
|
||||
} catch (error) {
|
||||
const attempt = task.attempts + 1;
|
||||
return markEventResult(client, task, null, {
|
||||
ok: false,
|
||||
httpCode: 0,
|
||||
responseSummary: '',
|
||||
errorMessage: error instanceof Error ? error.message : String(error),
|
||||
}, attempt);
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
|
||||
export async function processPlatformAuditNotificationBatch(limit = config.platformAuditNotificationBatchSize): Promise<ProcessResult> {
|
||||
const client = await pool.connect();
|
||||
let tasks: NotificationEventRow[] = [];
|
||||
let enqueued = 0;
|
||||
try {
|
||||
await client.query('begin');
|
||||
await recoverStaleProcessingEvents(client);
|
||||
enqueued = await enqueueNotificationEvents(client, limit);
|
||||
tasks = await claimDueEvents(client, limit);
|
||||
await client.query('commit');
|
||||
} catch (error) {
|
||||
await client.query('rollback');
|
||||
throw error;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
|
||||
const result: ProcessResult = { enqueued, processed: tasks.length, sent: 0, failed: 0, retrying: 0, discarded: 0 };
|
||||
for (const task of tasks) {
|
||||
const status = await processEvent(task);
|
||||
if (status === 'sent') result.sent += 1;
|
||||
if (status === 'failed') result.failed += 1;
|
||||
if (status === 'retrying') result.retrying += 1;
|
||||
if (status === 'discarded') result.discarded += 1;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
Reference in New Issue
Block a user