forked from wangziqi/gongxue-base
feat: add platform automatic billing worker
This commit is contained in:
Codex
@@ -18,6 +18,10 @@ export interface WorkerConfig {
|
||||
providerBillBatchSize: number;
|
||||
providerBillWorkerId: string;
|
||||
providerBillClaimStaleSeconds: number;
|
||||
platformBillingBatchSize: number;
|
||||
platformBillingDaysAhead: number;
|
||||
platformBillingDueDays: number;
|
||||
platformBillingWorkerId: string;
|
||||
assetBatchSize: number;
|
||||
assetMinAgeSeconds: number;
|
||||
assetRecheckIntervalSeconds: number;
|
||||
@@ -166,6 +170,10 @@ const loadedConfig: WorkerConfig = {
|
||||
providerBillBatchSize: envNumber('WORKER_PROVIDER_BILL_BATCH_SIZE', 5),
|
||||
providerBillWorkerId: envString('WORKER_PROVIDER_BILL_ID', `provider-bills-${process.pid}`),
|
||||
providerBillClaimStaleSeconds: envNumber('WORKER_PROVIDER_BILL_CLAIM_STALE_SECONDS', 15 * 60),
|
||||
platformBillingBatchSize: envNumber('WORKER_PLATFORM_BILLING_BATCH_SIZE', 50),
|
||||
platformBillingDaysAhead: envNumber('WORKER_PLATFORM_BILLING_DAYS_AHEAD', 45),
|
||||
platformBillingDueDays: envNumber('WORKER_PLATFORM_BILLING_DUE_DAYS', 15),
|
||||
platformBillingWorkerId: envString('WORKER_PLATFORM_BILLING_ID', `platform-billing-${process.pid}`),
|
||||
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),
|
||||
|
||||
@@ -42,6 +42,15 @@ async function runOnce() {
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'platform-billing') {
|
||||
const { processPlatformBillingBatch } = await import('./jobs/platform-billing.js');
|
||||
const result = await processPlatformBillingBatch();
|
||||
console.log(
|
||||
`[worker] platform-billing batch processed=${result.processed}`
|
||||
+ ` created=${result.created} skipped=${result.skipped} failed=${result.failed}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (job === 'assets') {
|
||||
const result = await processAssetBatch();
|
||||
console.log(
|
||||
|
||||
289
apps/worker/src/jobs/platform-billing.ts
Normal file
289
apps/worker/src/jobs/platform-billing.ts
Normal file
@@ -0,0 +1,289 @@
|
||||
import type pg from 'pg';
|
||||
import { pool } from '../db.js';
|
||||
import { config } from '../config.js';
|
||||
import {
|
||||
createInvoiceNo,
|
||||
recalculateInvoiceTotals,
|
||||
} from '../../../api/src/features/platform-admin/service.js';
|
||||
|
||||
interface BillingCandidate {
|
||||
tenantId: string;
|
||||
tenantSlug: string;
|
||||
tenantName: string;
|
||||
subscriptionId: string;
|
||||
planCode: string;
|
||||
planName: string | null;
|
||||
startsAt: string | null;
|
||||
expiresAt: string | null;
|
||||
billingCycle: string | null;
|
||||
amountCents: number;
|
||||
}
|
||||
|
||||
interface PlatformBillingWorkerResult {
|
||||
processed: number;
|
||||
created: number;
|
||||
skipped: number;
|
||||
failed: number;
|
||||
}
|
||||
|
||||
function positiveInteger(value: number, fallback: number, max: number) {
|
||||
if (!Number.isFinite(value) || value <= 0) return fallback;
|
||||
return Math.min(Math.trunc(value), max);
|
||||
}
|
||||
|
||||
function nonNegativeInteger(value: number, fallback: number, max: number) {
|
||||
if (!Number.isFinite(value) || value < 0) return fallback;
|
||||
return Math.min(Math.trunc(value), max);
|
||||
}
|
||||
|
||||
function dateText(value: unknown) {
|
||||
if (!value) return null;
|
||||
if (value instanceof Date) return value.toISOString().slice(0, 10);
|
||||
return String(value).slice(0, 10);
|
||||
}
|
||||
|
||||
function dueDateText(days: number) {
|
||||
const now = new Date();
|
||||
now.setUTCDate(now.getUTCDate() + nonNegativeInteger(days, 15, 365));
|
||||
return now.toISOString().slice(0, 10);
|
||||
}
|
||||
|
||||
function errorMessage(error: unknown) {
|
||||
return error instanceof Error ? error.message : String(error);
|
||||
}
|
||||
|
||||
function errorCode(error: unknown) {
|
||||
return typeof error === 'object' && error !== null && 'code' in error
|
||||
? String((error as { code?: unknown }).code || 'PLATFORM_BILLING_WORKER_ERROR')
|
||||
: 'PLATFORM_BILLING_WORKER_ERROR';
|
||||
}
|
||||
|
||||
function truncate(value: unknown, max = 1900) {
|
||||
return String(value ?? '').slice(0, max);
|
||||
}
|
||||
|
||||
async function loadSubscriptionBillingCandidates(params: {
|
||||
limit: number;
|
||||
daysAhead: number;
|
||||
}) {
|
||||
const result = await pool.query<BillingCandidate>(
|
||||
`
|
||||
select t.id as "tenantId",
|
||||
t.slug::text as "tenantSlug",
|
||||
t.name as "tenantName",
|
||||
s.id as "subscriptionId",
|
||||
s.plan_code as "planCode",
|
||||
p.name as "planName",
|
||||
s.starts_at as "startsAt",
|
||||
s.expires_at as "expiresAt",
|
||||
s.billing_cycle as "billingCycle",
|
||||
s.amount_cents as "amountCents"
|
||||
from public.tenant_subscriptions s
|
||||
join public.tenants t on t.id = s.tenant_id
|
||||
left join public.platform_saas_plans p on p.code = s.plan_code
|
||||
left join lateral (
|
||||
select id
|
||||
from public.tenant_invoices i
|
||||
where i.tenant_id = s.tenant_id
|
||||
and i.invoice_type = 'subscription'
|
||||
and i.status <> 'void'
|
||||
and i.metadata->>'subscriptionId' = s.id::text
|
||||
limit 1
|
||||
) existing on true
|
||||
where t.status = 'active'
|
||||
and s.status in ('trial', 'active', 'past_due')
|
||||
and existing.id is null
|
||||
and s.expires_at is not null
|
||||
and s.expires_at <= now() + ($2::integer * interval '1 day')
|
||||
order by s.expires_at asc nulls last, t.created_at asc
|
||||
limit $1
|
||||
`,
|
||||
[params.limit, params.daysAhead],
|
||||
);
|
||||
return result.rows;
|
||||
}
|
||||
|
||||
async function createInvoiceForCandidate(
|
||||
client: pg.PoolClient,
|
||||
candidate: BillingCandidate,
|
||||
dueDate: string,
|
||||
) {
|
||||
const invoiceNo = createInvoiceNo();
|
||||
const amountCents = Math.max(0, Math.trunc(Number(candidate.amountCents || 0)));
|
||||
if (amountCents <= 0) {
|
||||
throw Object.assign(new Error('Subscription amount must be greater than zero'), {
|
||||
code: 'SUBSCRIPTION_AMOUNT_REQUIRED',
|
||||
});
|
||||
}
|
||||
|
||||
const invoiceResult = await client.query<{
|
||||
id: string;
|
||||
tenantId: string;
|
||||
invoiceNo: string;
|
||||
status: string;
|
||||
totalCents: number;
|
||||
balanceCents: number;
|
||||
}>(
|
||||
`
|
||||
insert into public.tenant_invoices (
|
||||
tenant_id, invoice_no, invoice_type, status, currency,
|
||||
subtotal_cents, discount_cents, tax_cents, total_cents,
|
||||
paid_cents, balance_cents, billing_period_start, billing_period_end,
|
||||
due_date, issued_at, note, metadata
|
||||
)
|
||||
values (
|
||||
$1, $2, 'subscription', 'issued', 'CNY',
|
||||
$3, 0, 0, $3,
|
||||
0, $3, $4::date, $5::date,
|
||||
$6::date, now(), $7, $8::jsonb
|
||||
)
|
||||
returning id, tenant_id as "tenantId", invoice_no as "invoiceNo",
|
||||
status, total_cents as "totalCents", balance_cents as "balanceCents"
|
||||
`,
|
||||
[
|
||||
candidate.tenantId,
|
||||
invoiceNo,
|
||||
amountCents,
|
||||
dateText(candidate.startsAt),
|
||||
dateText(candidate.expiresAt),
|
||||
dueDate,
|
||||
'平台自动生成订阅服务费账单',
|
||||
JSON.stringify({
|
||||
source: 'subscription_auto',
|
||||
subscriptionId: candidate.subscriptionId,
|
||||
tenantSlug: candidate.tenantSlug,
|
||||
workerId: config.platformBillingWorkerId,
|
||||
}),
|
||||
],
|
||||
);
|
||||
const invoice = invoiceResult.rows[0];
|
||||
|
||||
await client.query(
|
||||
`
|
||||
insert into public.tenant_invoice_items (
|
||||
tenant_id, invoice_id, item_type, description,
|
||||
quantity, unit_amount_cents, amount_cents, metadata
|
||||
)
|
||||
values ($1, $2, 'subscription', $3, 1, $4, $4, $5::jsonb)
|
||||
`,
|
||||
[
|
||||
candidate.tenantId,
|
||||
invoice.id,
|
||||
`${candidate.planName || candidate.planCode} ${candidate.planCode}`,
|
||||
amountCents,
|
||||
JSON.stringify({
|
||||
subscriptionId: candidate.subscriptionId,
|
||||
planCode: candidate.planCode,
|
||||
source: 'subscription_auto',
|
||||
}),
|
||||
],
|
||||
);
|
||||
|
||||
await recalculateInvoiceTotals(client, invoice.id);
|
||||
await client.query(
|
||||
`
|
||||
insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
|
||||
values ($1, null, 'platform.invoice.subscription_auto_created', 'tenant_invoice', $2, $3::jsonb)
|
||||
`,
|
||||
[
|
||||
candidate.tenantId,
|
||||
invoice.id,
|
||||
JSON.stringify({
|
||||
subscriptionId: candidate.subscriptionId,
|
||||
planCode: candidate.planCode,
|
||||
invoiceNo: invoice.invoiceNo,
|
||||
amountCents,
|
||||
dueDate,
|
||||
workerId: config.platformBillingWorkerId,
|
||||
}),
|
||||
],
|
||||
);
|
||||
return invoice;
|
||||
}
|
||||
|
||||
async function processCandidate(candidate: BillingCandidate, dueDate: string) {
|
||||
const client = await pool.connect();
|
||||
try {
|
||||
await client.query('begin');
|
||||
|
||||
const subscriptionLock = await client.query<{ id: string; status: string }>(
|
||||
`
|
||||
select id, status
|
||||
from public.tenant_subscriptions
|
||||
where id = $1 and tenant_id = $2
|
||||
for update
|
||||
`,
|
||||
[candidate.subscriptionId, candidate.tenantId],
|
||||
);
|
||||
const lockedSubscription = subscriptionLock.rows[0];
|
||||
if (!lockedSubscription || !['trial', 'active', 'past_due'].includes(lockedSubscription.status)) {
|
||||
await client.query('rollback');
|
||||
return 'skipped' as const;
|
||||
}
|
||||
|
||||
const existing = await client.query<{ id: string }>(
|
||||
`
|
||||
select id
|
||||
from public.tenant_invoices
|
||||
where tenant_id = $1
|
||||
and invoice_type = 'subscription'
|
||||
and status <> 'void'
|
||||
and metadata->>'subscriptionId' = $2
|
||||
limit 1
|
||||
`,
|
||||
[candidate.tenantId, candidate.subscriptionId],
|
||||
);
|
||||
if (existing.rows[0]) {
|
||||
await client.query('rollback');
|
||||
return 'skipped' as const;
|
||||
}
|
||||
|
||||
await createInvoiceForCandidate(client, candidate, dueDate);
|
||||
await client.query('commit');
|
||||
return 'created' as const;
|
||||
} catch (error) {
|
||||
await client.query('rollback').catch(() => {});
|
||||
await pool.query(
|
||||
`
|
||||
insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
|
||||
values ($1, null, 'platform.invoice.subscription_auto_failed', 'tenant_subscription', $2, $3::jsonb)
|
||||
`,
|
||||
[
|
||||
candidate.tenantId,
|
||||
candidate.subscriptionId,
|
||||
JSON.stringify({
|
||||
code: errorCode(error),
|
||||
message: truncate(errorMessage(error)),
|
||||
workerId: config.platformBillingWorkerId,
|
||||
}),
|
||||
],
|
||||
).catch(() => {});
|
||||
return 'failed' as const;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
|
||||
export async function processPlatformBillingBatch(options: {
|
||||
limit?: number;
|
||||
daysAhead?: number;
|
||||
dueDays?: number;
|
||||
} = {}): Promise<PlatformBillingWorkerResult> {
|
||||
const limit = positiveInteger(options.limit ?? config.platformBillingBatchSize, 50, 500);
|
||||
const daysAhead = nonNegativeInteger(options.daysAhead ?? config.platformBillingDaysAhead, 45, 1095);
|
||||
const dueDate = dueDateText(options.dueDays ?? config.platformBillingDueDays);
|
||||
const candidates = await loadSubscriptionBillingCandidates({ limit, daysAhead });
|
||||
const result: PlatformBillingWorkerResult = {
|
||||
processed: candidates.length,
|
||||
created: 0,
|
||||
skipped: 0,
|
||||
failed: 0,
|
||||
};
|
||||
|
||||
for (const candidate of candidates) {
|
||||
const status = await processCandidate(candidate, dueDate);
|
||||
result[status] += 1;
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
Reference in New Issue
Block a user