forked from wangziqi/gongxue-base
feat: add platform dunning notifications
This commit is contained in:
Codex
@@ -13,6 +13,7 @@
|
||||
"provider-bills:once": "tsx src/index.ts --once --job provider-bills",
|
||||
"platform-billing:once": "tsx src/index.ts --once --job platform-billing",
|
||||
"platform-dunning:once": "tsx src/index.ts --once --job platform-dunning",
|
||||
"platform-dunning-notifications:once": "tsx src/index.ts --once --job platform-dunning-notifications",
|
||||
"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",
|
||||
|
||||
@@ -24,6 +24,11 @@ export interface WorkerConfig {
|
||||
platformBillingWorkerId: string;
|
||||
platformDunningBatchSize: number;
|
||||
platformDunningWorkerId: string;
|
||||
platformDunningNotificationBatchSize: number;
|
||||
platformDunningNotificationMaxAttempts: number;
|
||||
platformDunningNotificationBackoffSeconds: number[];
|
||||
platformDunningNotificationRequestTimeoutMs: number;
|
||||
platformDunningNotificationAllowInsecureLocalhost: boolean;
|
||||
platformAuditAlertBatchSize: number;
|
||||
platformAuditAlertWorkerId: string;
|
||||
platformAuditAlertLookbackDays: number;
|
||||
@@ -119,6 +124,9 @@ function validateProductionConfig(nextConfig: WorkerConfig) {
|
||||
if (nextConfig.platformAuditNotificationAllowInsecureLocalhost) {
|
||||
failures.push('WORKER_PLATFORM_AUDIT_NOTIFICATION_ALLOW_INSECURE_LOCALHOST=true is not allowed in production workers');
|
||||
}
|
||||
if (nextConfig.platformDunningNotificationAllowInsecureLocalhost) {
|
||||
failures.push('WORKER_PLATFORM_DUNNING_NOTIFICATION_ALLOW_INSECURE_LOCALHOST=true is not allowed in production workers');
|
||||
}
|
||||
const scannerModes = nextConfig.assetSecurityScanner
|
||||
.split(',')
|
||||
.map(item => item.trim().toLowerCase())
|
||||
@@ -189,6 +197,13 @@ const loadedConfig: WorkerConfig = {
|
||||
platformBillingWorkerId: envString('WORKER_PLATFORM_BILLING_ID', `platform-billing-${process.pid}`),
|
||||
platformDunningBatchSize: envNumber('WORKER_PLATFORM_DUNNING_BATCH_SIZE', 100),
|
||||
platformDunningWorkerId: envString('WORKER_PLATFORM_DUNNING_ID', `platform-dunning-${process.pid}`),
|
||||
platformDunningNotificationBatchSize: envNumber('WORKER_PLATFORM_DUNNING_NOTIFICATION_BATCH_SIZE', 50),
|
||||
platformDunningNotificationMaxAttempts: envNumber('WORKER_PLATFORM_DUNNING_NOTIFICATION_MAX_ATTEMPTS', 5),
|
||||
platformDunningNotificationBackoffSeconds: envList('WORKER_PLATFORM_DUNNING_NOTIFICATION_BACKOFF_SECONDS', '10,60,300,900,1800')
|
||||
.map((value: string) => Number(value))
|
||||
.filter((value: number) => Number.isFinite(value) && value > 0),
|
||||
platformDunningNotificationRequestTimeoutMs: envNumber('WORKER_PLATFORM_DUNNING_NOTIFICATION_REQUEST_TIMEOUT_MS', 10_000),
|
||||
platformDunningNotificationAllowInsecureLocalhost: envBoolean('WORKER_PLATFORM_DUNNING_NOTIFICATION_ALLOW_INSECURE_LOCALHOST', false),
|
||||
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),
|
||||
|
||||
@@ -61,6 +61,16 @@ async function runOnce() {
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-dunning-notifications') {
|
||||
const { processPlatformDunningNotificationBatch } = await import('./jobs/platform-dunning-notifications.js');
|
||||
const result = await processPlatformDunningNotificationBatch();
|
||||
console.log(
|
||||
`[worker] platform-dunning-notifications batch enqueued=${result.enqueued}`
|
||||
+ ` processed=${result.processed} sent=${result.sent} failed=${result.failed}`
|
||||
+ ` retrying=${result.retrying} discarded=${result.discarded}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-audit-alerts') {
|
||||
const { processPlatformAuditAlertBatch } = await import('./jobs/platform-audit-alerts.js');
|
||||
const result = await processPlatformAuditAlertBatch();
|
||||
|
||||
702
apps/worker/src/jobs/platform-dunning-notifications.ts
Normal file
702
apps/worker/src/jobs/platform-dunning-notifications.ts
Normal file
@@ -0,0 +1,702 @@
|
||||
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 DunningChannelRow {
|
||||
id: string;
|
||||
channelCode: string;
|
||||
name: string;
|
||||
provider: NotificationProvider;
|
||||
webhookUrl: string;
|
||||
secretRef: string | null;
|
||||
reminderTypes: string[];
|
||||
reminderChannels: string[];
|
||||
minReminderLevel: number;
|
||||
tenantIds: string[];
|
||||
timeoutSec: number | null;
|
||||
}
|
||||
|
||||
interface DunningReminderRow {
|
||||
id: string;
|
||||
tenantId: string;
|
||||
invoiceId: string;
|
||||
reminderType: string;
|
||||
channel: string;
|
||||
reminderLevel: number;
|
||||
}
|
||||
|
||||
interface DunningEventRow {
|
||||
id: string;
|
||||
channelId: string;
|
||||
channelCode: string;
|
||||
channelName: string;
|
||||
reminderId: string;
|
||||
invoiceId: string;
|
||||
tenantId: string;
|
||||
provider: NotificationProvider;
|
||||
attempts: number;
|
||||
webhookUrl: string;
|
||||
secretRef: string | null;
|
||||
timeoutSec: number | null;
|
||||
tenantSlug: string;
|
||||
tenantName: string;
|
||||
legalName: string | null;
|
||||
billingStatus: string;
|
||||
invoiceNo: string;
|
||||
invoiceType: string;
|
||||
invoiceStatus: string;
|
||||
currency: string;
|
||||
totalCents: number;
|
||||
paidCents: number;
|
||||
balanceCents: number;
|
||||
dueDate: string | null;
|
||||
issuedAt: string | null;
|
||||
reminderType: string;
|
||||
reminderChannel: string;
|
||||
reminderStatus: string;
|
||||
reminderDate: string;
|
||||
reminderLevel: number;
|
||||
balanceCentsSnapshot: number;
|
||||
message: string | null;
|
||||
reminderMetadata: Record<string, unknown> | null;
|
||||
billingName: string | null;
|
||||
contactName: string | null;
|
||||
contactPhone: string | null;
|
||||
contactEmail: string | null;
|
||||
}
|
||||
|
||||
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 truncate(value: unknown, max = 1900) {
|
||||
return String(value ?? '').slice(0, max);
|
||||
}
|
||||
|
||||
function maskPhone(value: string | null) {
|
||||
if (!value) return null;
|
||||
const digits = value.replace(/\D/g, '');
|
||||
if (digits.length < 7) return '***';
|
||||
return `${digits.slice(0, 3)}****${digits.slice(-4)}`;
|
||||
}
|
||||
|
||||
function maskEmail(value: string | null) {
|
||||
if (!value) return null;
|
||||
const [name, domain] = value.split('@');
|
||||
if (!name || !domain) return '***';
|
||||
return `${name.slice(0, 2)}***@${domain}`;
|
||||
}
|
||||
|
||||
function redactDunningNotificationValue(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 => redactDunningNotificationValue(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] = redactDunningNotificationValue(item, key, depth + 1);
|
||||
}
|
||||
return output;
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
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 dunning notification webhook URL is invalid');
|
||||
}
|
||||
const isLocalhost = ['127.0.0.1', 'localhost', '::1'].includes(url.hostname);
|
||||
if (url.protocol !== 'https:' && !(config.platformDunningNotificationAllowInsecureLocalhost && isLocalhost)) {
|
||||
throw new Error('Platform dunning 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 dunningPayload(task: DunningEventRow) {
|
||||
return {
|
||||
event: 'platform.invoice.dunning_reminder',
|
||||
tenant: {
|
||||
id: task.tenantId,
|
||||
slug: task.tenantSlug,
|
||||
name: task.tenantName,
|
||||
legalName: task.legalName,
|
||||
billingStatus: task.billingStatus,
|
||||
},
|
||||
invoice: {
|
||||
id: task.invoiceId,
|
||||
invoiceNo: task.invoiceNo,
|
||||
invoiceType: task.invoiceType,
|
||||
status: task.invoiceStatus,
|
||||
currency: task.currency,
|
||||
totalCents: Number(task.totalCents || 0),
|
||||
paidCents: Number(task.paidCents || 0),
|
||||
balanceCents: Number(task.balanceCents || 0),
|
||||
dueDate: task.dueDate,
|
||||
issuedAt: task.issuedAt,
|
||||
},
|
||||
reminder: {
|
||||
id: task.reminderId,
|
||||
type: task.reminderType,
|
||||
channel: task.reminderChannel,
|
||||
status: task.reminderStatus,
|
||||
date: task.reminderDate,
|
||||
level: Number(task.reminderLevel || 0),
|
||||
balanceCentsSnapshot: Number(task.balanceCentsSnapshot || 0),
|
||||
message: task.message,
|
||||
metadata: redactDunningNotificationValue(task.reminderMetadata || {}),
|
||||
},
|
||||
billingContact: {
|
||||
billingName: task.billingName,
|
||||
contactName: task.contactName,
|
||||
phoneMasked: maskPhone(task.contactPhone),
|
||||
emailMasked: maskEmail(task.contactEmail),
|
||||
},
|
||||
source: {
|
||||
channelId: task.channelId,
|
||||
channelCode: task.channelCode,
|
||||
eventId: task.id,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function dunningMarkdown(task: DunningEventRow) {
|
||||
return [
|
||||
`### SaaS 服务费催缴：${task.tenantName}`,
|
||||
`- 租户：${task.tenantName} (${task.tenantSlug})`,
|
||||
`- 账单号：${task.invoiceNo}`,
|
||||
`- 账单状态：${task.invoiceStatus}`,
|
||||
`- 未结清金额：${Math.round(Number(task.balanceCents || 0)) / 100} ${task.currency}`,
|
||||
`- 到期日：${task.dueDate || '未设置'}`,
|
||||
`- 催缴类型：${task.reminderType}`,
|
||||
`- 催缴级别：${task.reminderLevel}`,
|
||||
`- 联系人：${task.contactName || '未设置'} ${maskPhone(task.contactPhone) || ''}`,
|
||||
`- 催缴记录ID：${task.reminderId}`,
|
||||
`- 备注：${task.message || '无'}`,
|
||||
].join('\n');
|
||||
}
|
||||
|
||||
function prepareRequest(task: DunningEventRow, 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 = dunningMarkdown(task);
|
||||
|
||||
if (provider === 'dingtalk') {
|
||||
return {
|
||||
provider,
|
||||
url: appendQuery(target, dingtalkSign(secretValue)),
|
||||
headers,
|
||||
body: {
|
||||
msgtype: 'markdown',
|
||||
markdown: {
|
||||
title: 'SaaS 服务费催缴',
|
||||
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: 'SaaS 服务费催缴' }, template: 'orange' },
|
||||
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: dunningPayload(task),
|
||||
};
|
||||
}
|
||||
|
||||
async function loadEnabledChannels(client: pg.PoolClient) {
|
||||
const result = await client.query<DunningChannelRow>(
|
||||
`
|
||||
select id, channel_code as "channelCode", name, provider,
|
||||
webhook_url as "webhookUrl", secret_ref as "secretRef",
|
||||
reminder_types as "reminderTypes", reminder_channels as "reminderChannels",
|
||||
min_reminder_level as "minReminderLevel", tenant_ids as "tenantIds",
|
||||
timeout_sec as "timeoutSec"
|
||||
from public.platform_dunning_notification_channels
|
||||
where enabled = true
|
||||
order by min_reminder_level asc, created_at asc
|
||||
`,
|
||||
);
|
||||
return result.rows;
|
||||
}
|
||||
|
||||
async function loadPendingReminders(client: pg.PoolClient, limit: number) {
|
||||
const result = await client.query<DunningReminderRow>(
|
||||
`
|
||||
select id, tenant_id as "tenantId", invoice_id as "invoiceId",
|
||||
reminder_type as "reminderType", channel, reminder_level as "reminderLevel"
|
||||
from public.tenant_invoice_reminders
|
||||
where status in ('pending', 'failed')
|
||||
and reminder_type in ('due_soon', 'overdue', 'final_notice', 'manual')
|
||||
order by reminder_level desc, 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 reminders = await loadPendingReminders(client, Math.max(limit * 4, 50));
|
||||
let enqueued = 0;
|
||||
|
||||
for (const reminder of reminders) {
|
||||
for (const channel of channels) {
|
||||
if (enqueued >= limit) return enqueued;
|
||||
if (!channel.reminderTypes.includes(reminder.reminderType)) continue;
|
||||
if (!channel.reminderChannels.includes(reminder.channel)) continue;
|
||||
if (Number(reminder.reminderLevel || 0) < Number(channel.minReminderLevel || 1)) continue;
|
||||
if (channel.tenantIds.length && !channel.tenantIds.includes(reminder.tenantId)) continue;
|
||||
|
||||
const result = await client.query(
|
||||
`
|
||||
insert into public.platform_dunning_notification_events (
|
||||
channel_id, reminder_id, invoice_id, tenant_id, provider, status,
|
||||
request_payload, metadata
|
||||
)
|
||||
values ($1, $2, $3, $4, $5, 'pending', $6::jsonb, $7::jsonb)
|
||||
on conflict (channel_id, reminder_id) do nothing
|
||||
returning id
|
||||
`,
|
||||
[
|
||||
channel.id,
|
||||
reminder.id,
|
||||
reminder.invoiceId,
|
||||
reminder.tenantId,
|
||||
channel.provider,
|
||||
JSON.stringify({
|
||||
channelCode: channel.channelCode,
|
||||
reminderId: reminder.id,
|
||||
invoiceId: reminder.invoiceId,
|
||||
tenantId: reminder.tenantId,
|
||||
reminderType: reminder.reminderType,
|
||||
}),
|
||||
JSON.stringify({ enqueuedBy: 'platform-dunning-notifications-worker' }),
|
||||
],
|
||||
);
|
||||
if (result.rowCount) enqueued += 1;
|
||||
}
|
||||
}
|
||||
return enqueued;
|
||||
}
|
||||
|
||||
async function claimDueEvents(client: pg.PoolClient, limit: number) {
|
||||
const result = await client.query<DunningEventRow>(
|
||||
`
|
||||
with due as (
|
||||
select e.id
|
||||
from public.platform_dunning_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_dunning_notification_events e
|
||||
set status = 'processing',
|
||||
last_attempt_at = now(),
|
||||
updated_at = now()
|
||||
from due
|
||||
join public.platform_dunning_notification_channels c on true
|
||||
join public.tenant_invoice_reminders r on true
|
||||
join public.tenant_invoices i on true
|
||||
join public.tenants t on true
|
||||
left join public.tenant_billing_profiles bp on bp.tenant_id = t.id
|
||||
where e.id = due.id
|
||||
and c.id = e.channel_id
|
||||
and r.id = e.reminder_id
|
||||
and i.id = e.invoice_id
|
||||
and t.id = e.tenant_id
|
||||
returning e.id, e.channel_id as "channelId", c.channel_code as "channelCode",
|
||||
c.name as "channelName", e.reminder_id as "reminderId",
|
||||
e.invoice_id as "invoiceId", e.tenant_id as "tenantId",
|
||||
e.provider, e.attempts, c.webhook_url as "webhookUrl",
|
||||
c.secret_ref as "secretRef", c.timeout_sec as "timeoutSec",
|
||||
t.slug::text as "tenantSlug", t.name as "tenantName",
|
||||
t.legal_name as "legalName", t.billing_status as "billingStatus",
|
||||
i.invoice_no as "invoiceNo", i.invoice_type as "invoiceType",
|
||||
i.status as "invoiceStatus", i.currency, i.total_cents as "totalCents",
|
||||
i.paid_cents as "paidCents", i.balance_cents as "balanceCents",
|
||||
i.due_date as "dueDate", i.issued_at as "issuedAt",
|
||||
r.reminder_type as "reminderType", r.channel as "reminderChannel",
|
||||
r.status as "reminderStatus", r.reminder_date as "reminderDate",
|
||||
r.reminder_level as "reminderLevel",
|
||||
r.balance_cents_snapshot as "balanceCentsSnapshot",
|
||||
r.message, r.metadata as "reminderMetadata",
|
||||
bp.billing_name as "billingName", bp.contact_name as "contactName",
|
||||
bp.contact_phone as "contactPhone", bp.contact_email as "contactEmail"
|
||||
`,
|
||||
[limit],
|
||||
);
|
||||
return result.rows;
|
||||
}
|
||||
|
||||
async function recoverStaleProcessingEvents(client: pg.PoolClient) {
|
||||
const staleMs = Math.max(config.platformDunningNotificationRequestTimeoutMs * 3, 60_000);
|
||||
await client.query(
|
||||
`
|
||||
update public.platform_dunning_notification_events
|
||||
set status = 'retrying',
|
||||
next_attempt_at = now(),
|
||||
last_error = coalesce(last_error, 'Recovered stale processing platform dunning 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 dunning 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: DunningEventRow,
|
||||
request: PreparedRequest | null,
|
||||
result: SendResult,
|
||||
attempt: number,
|
||||
) {
|
||||
const backoff = config.platformDunningNotificationBackoffSeconds[Math.min(attempt - 1, config.platformDunningNotificationBackoffSeconds.length - 1)] || 60;
|
||||
const requestPayload = {
|
||||
provider: request?.provider || task.provider,
|
||||
urlHost: request ? new URL(request.url).hostname : null,
|
||||
body: request?.body ? redactDunningNotificationValue(request.body) : null,
|
||||
};
|
||||
|
||||
if (result.ok) {
|
||||
await client.query(
|
||||
`
|
||||
update public.platform_dunning_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)],
|
||||
);
|
||||
|
||||
await client.query(
|
||||
`
|
||||
update public.tenant_invoice_reminders
|
||||
set status = 'sent',
|
||||
sent_at = coalesce(sent_at, now()),
|
||||
metadata = metadata || $2::jsonb,
|
||||
updated_at = now()
|
||||
where id = $1
|
||||
and status in ('pending', 'failed')
|
||||
`,
|
||||
[
|
||||
task.reminderId,
|
||||
JSON.stringify({
|
||||
externalNotification: {
|
||||
status: 'sent',
|
||||
eventId: task.id,
|
||||
channelCode: task.channelCode,
|
||||
sentAt: new Date().toISOString(),
|
||||
},
|
||||
}),
|
||||
],
|
||||
);
|
||||
return 'sent';
|
||||
}
|
||||
|
||||
const terminal = attempt >= config.platformDunningNotificationMaxAttempts;
|
||||
await client.query(
|
||||
`
|
||||
update public.platform_dunning_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),
|
||||
],
|
||||
);
|
||||
|
||||
if (terminal) {
|
||||
await client.query(
|
||||
`
|
||||
update public.tenant_invoice_reminders
|
||||
set status = 'failed',
|
||||
metadata = metadata || $2::jsonb,
|
||||
updated_at = now()
|
||||
where id = $1
|
||||
and status in ('pending', 'failed')
|
||||
`,
|
||||
[
|
||||
task.reminderId,
|
||||
JSON.stringify({
|
||||
externalNotification: {
|
||||
status: 'failed',
|
||||
eventId: task.id,
|
||||
channelCode: task.channelCode,
|
||||
failedAt: new Date().toISOString(),
|
||||
lastError: truncate(result.errorMessage, 500),
|
||||
},
|
||||
}),
|
||||
],
|
||||
);
|
||||
}
|
||||
|
||||
return terminal ? 'failed' : 'retrying';
|
||||
}
|
||||
|
||||
async function discardEvent(client: pg.PoolClient, task: DunningEventRow, message: string) {
|
||||
await client.query(
|
||||
`
|
||||
update public.platform_dunning_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: DunningEventRow) {
|
||||
const client = await pool.connect();
|
||||
try {
|
||||
if (!PROVIDERS.includes(task.provider)) {
|
||||
return discardEvent(client, task, 'Unsupported platform dunning notification provider');
|
||||
}
|
||||
const secret = await loadSecret(client, task.secretRef);
|
||||
const request = prepareRequest(task, secret);
|
||||
const timeoutMs = Math.max(1000, (task.timeoutSec || 0) * 1000 || config.platformDunningNotificationRequestTimeoutMs);
|
||||
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 processPlatformDunningNotificationBatch(limit = config.platformDunningNotificationBatchSize): Promise<ProcessResult> {
|
||||
const client = await pool.connect();
|
||||
let tasks: DunningEventRow[] = [];
|
||||
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