feat(billing): add products subscriptions orders and usage

This commit is contained in:
Jesse_Chen
2026-08-06 20:15:08 +08:00
parent 6457c836fa
commit 7040fd998e
22 changed files with 2439 additions and 277 deletions
+31 -1
View File
@@ -13,11 +13,18 @@ import {
isSupabaseConfigurationError,
} from "@/lib/supabase/config";
import { createServerSupabaseClient } from "@/lib/supabase/server";
import { readSelfHostedIdentityConfig } from "@/modules/identity/config";
export const runtime = "nodejs";
const unfinishedRectificationStatuses = ["starting", "active", "paused", "confirming"] as const;
function adminEntryUrl(): string {
return process.env.AUTH_PROVIDER?.trim() === "self-hosted"
? `${readSelfHostedIdentityConfig(process.env).adminOrigin}/admin`
: "/admin";
}
function isMissingProfileColumn(error: { code?: string; message?: string } | null) {
const message = error?.message?.toLowerCase() ?? "";
return error?.code === "PGRST204"
@@ -109,8 +116,21 @@ export async function GET() {
profile,
Array.isArray(rectificationCaseRows) ? rectificationCaseRows : [],
);
const { data: activeSubscriptions, error: subscriptionError } = await admin
.from("user_subscriptions")
.select("id,status,starts_at,ends_at,product_code,product_version,product_snapshot,entitlement_snapshot")
.eq("user_id", userId)
.eq("status", "active")
.lte("starts_at", new Date().toISOString())
.gt("ends_at", new Date().toISOString())
.order("ends_at", { ascending: true })
.limit(1);
if (subscriptionError) {
return NextResponse.json({ error: "暂时无法读取会员状态" }, { status: 500 });
}
const activeSubscription = activeSubscriptions?.[0] ?? null;
const isAdmin = await isAdminUser(user);
const adminUrl = isAdmin ? "/admin/codes" : null;
const adminUrl = isAdmin ? adminEntryUrl() : null;
return NextResponse.json({
user: { id: user.id, email: user.email ?? null },
@@ -118,6 +138,16 @@ export async function GET() {
isAdmin,
adminUrl,
rectificationPriceCredits,
activeSubscription: activeSubscription ? {
id: activeSubscription.id,
status: activeSubscription.status,
startsAt: activeSubscription.starts_at,
endsAt: activeSubscription.ends_at,
productCode: activeSubscription.product_code,
productVersion: activeSubscription.product_version,
product: activeSubscription.product_snapshot,
entitlements: activeSubscription.entitlement_snapshot,
} : null,
hasConfirmedBirthTime: profile.birth_time_status === "confirmed"
&& typeof profile.active_birth_time === "string",
hasUsableBirthTime: (profile.birth_time_status === "accepted" || profile.birth_time_status === "confirmed")
+29 -11
View File
@@ -1,32 +1,43 @@
import { NextResponse } from "next/server";
import { z } from "zod";
import { requireAdminSession } from "@/lib/admin/auth";
import { runCodeRpc } from "@/lib/admin/codes";
import {
adminErrorResponse,
invalidQueryResponse,
requireHighRiskAdminMutation,
requestId,
} from "@/lib/admin/http";
export const runtime = "nodejs";
const paramsSchema = z.object({ id: z.string().uuid() });
const updateCodeSchema = z.object({
note: z.string().trim().max(500).nullable().optional(),
expiresAt: z.string().datetime({ offset: true }).nullable().optional(),
}).refine((value) => "note" in value || "expiresAt" in value, {
message: "至少提供一个可修改字段",
const revokeCodeSchema = z.object({
reason: z.string().trim().min(1).max(500),
});
const updateCodeSchema = z
.object({
note: z.string().trim().max(500).nullable().optional(),
expiresAt: z.string().datetime({ offset: true }).nullable().optional(),
reason: z.string().trim().min(1).max(500),
})
.refine((value) => "note" in value || "expiresAt" in value, {
message: "至少提供一个可修改字段",
});
export async function PATCH(
request: Request,
context: { params: Promise<{ id: string }> },
) {
try {
const session = await requireAdminSession("write");
const session = await requireHighRiskAdminMutation(
request,
"billing.adjustments.write",
);
const parsedParams = paramsSchema.safeParse(await context.params);
const parsedBody = updateCodeSchema.safeParse(await request.json().catch(() => null));
const parsedBody = updateCodeSchema.safeParse(
await request.json().catch(() => null),
);
if (!parsedParams.success || !parsedBody.success) {
return invalidQueryResponse();
}
@@ -41,6 +52,7 @@ export async function PATCH(
p_note: body.note ?? null,
p_set_expires_at: "expiresAt" in body,
p_expires_at: body.expiresAt ?? null,
p_reason: body.reason,
},
);
return NextResponse.json({ data: rows[0] });
@@ -54,14 +66,20 @@ export async function DELETE(
context: { params: Promise<{ id: string }> },
) {
try {
const session = await requireAdminSession("write");
const session = await requireHighRiskAdminMutation(
request,
"billing.adjustments.write",
);
const parsed = paramsSchema.safeParse(await context.params);
if (!parsed.success) return invalidQueryResponse();
const parsedBody = revokeCodeSchema.safeParse(
await request.json().catch(() => null),
);
if (!parsed.success || !parsedBody.success) return invalidQueryResponse();
const rows = await runCodeRpc(
"admin_revoke_redemption_code",
session,
requestId(request),
{ p_code_id: parsed.data.id },
{ p_code_id: parsed.data.id, p_reason: parsedBody.data.reason },
);
return NextResponse.json({ data: rows[0] });
} catch (error) {
+48 -21
View File
@@ -1,13 +1,18 @@
import { NextResponse } from "next/server";
import { z } from "zod";
import { requireAdminSession } from "@/lib/admin/auth";
import { mapCode, runCodeRpc, type RedemptionCodeRecord } from "@/lib/admin/codes";
import { requirePermission } from "@/lib/admin/auth";
import {
mapCode,
runCodeRpc,
type RedemptionCodeRecord,
} from "@/lib/admin/codes";
import { pageOffset, queryAdminRows } from "@/lib/admin/database";
import {
adminErrorResponse,
invalidQueryResponse,
parseListQuery,
requireHighRiskAdminMutation,
requestId,
} from "@/lib/admin/http";
import {
@@ -23,6 +28,7 @@ const createCodesSchema = z.object({
count: z.number().int().min(1).max(100),
expiresAt: z.string().datetime({ offset: true }).nullable().optional(),
note: z.string().trim().max(500).nullable().optional(),
reason: z.string().trim().min(1).max(500),
});
type CodeRow = {
@@ -59,7 +65,7 @@ function serializedCodeRow(row: CodeRow) {
export async function GET(request: Request) {
try {
await requireAdminSession();
await requirePermission("billing.orders.read");
const parsed = parseListQuery(request);
if (!parsed.success) return invalidQueryResponse(parsed.error.flatten());
const { page, pageSize, sort, order, q, status } = parsed.data;
@@ -67,12 +73,19 @@ export async function GET(request: Request) {
const conditions: string[] = [];
if (q) {
values.push(`%${q}%`);
conditions.push(`(c.code_mask ilike $${values.length} or c.note ilike $${values.length})`);
conditions.push(
`(c.code_mask ilike $${values.length} or c.note ilike $${values.length})`,
);
}
if (status && ["available", "expired", "redeemed", "revoked"].includes(status)) {
if (
status &&
["available", "expired", "redeemed", "revoked"].includes(status)
) {
const clauses = {
available: "c.redeemed_at is null and c.revoked_at is null and (c.expires_at is null or c.expires_at > now())",
expired: "c.redeemed_at is null and c.revoked_at is null and c.expires_at <= now()",
available:
"c.redeemed_at is null and c.revoked_at is null and (c.expires_at is null or c.expires_at > now())",
expired:
"c.redeemed_at is null and c.revoked_at is null and c.expires_at <= now()",
redeemed: "c.redeemed_at is not null",
revoked: "c.revoked_at is not null",
};
@@ -80,7 +93,8 @@ export async function GET(request: Request) {
}
values.push(pageSize, pageOffset(page, pageSize));
const sortColumn = sortColumns.get(sort ?? "createdAt") ?? "c.created_at";
const rows = await queryAdminRows<CodeRow>(`
const rows = await queryAdminRows<CodeRow>(
`
select c.id, c.code_mask, c.credits, c.expires_at, c.note,
c.created_at, c.redeemed_by, c.redeemed_email, c.redeemed_at,
c.revoked_by, c.revoked_at,
@@ -95,7 +109,9 @@ export async function GET(request: Request) {
${conditions.length ? `where ${conditions.join(" and ")}` : ""}
order by ${sortColumn} ${order === "asc" ? "asc" : "desc"}, c.id asc
limit $${values.length - 1} offset $${values.length}
`, values);
`,
values,
);
return NextResponse.json({
data: rows.map(serializedCodeRow),
total: Number(rows[0]?.total_count ?? 0),
@@ -107,10 +123,18 @@ export async function GET(request: Request) {
export async function POST(request: Request) {
try {
const session = await requireAdminSession("write");
const parsed = createCodesSchema.safeParse(await request.json().catch(() => null));
const session = await requireHighRiskAdminMutation(
request,
"billing.adjustments.write",
);
const parsed = createCodesSchema.safeParse(
await request.json().catch(() => null),
);
if (!parsed.success) return invalidQueryResponse(parsed.error.flatten());
const plainCodes = Array.from({ length: parsed.data.count }, generateRedeemCode);
const plainCodes = Array.from(
{ length: parsed.data.count },
generateRedeemCode,
);
const records = plainCodes.map((code) => ({
codeHash: hashRedeemCode(code),
codeMask: maskRedeemCode(code),
@@ -123,20 +147,23 @@ export async function POST(request: Request) {
"admin_create_redemption_codes",
session,
operationRequestId,
{ p_codes: records },
{ p_codes: records, p_reason: parsed.data.reason },
);
const byMask = new Map<string, RedemptionCodeRecord>(
stored.map((record) => [record.mask, record]),
);
return NextResponse.json({
data: {
id: operationRequestId,
generated: plainCodes.map((code) => ({
...(byMask.get(maskRedeemCode(code)) ?? {}),
code,
})),
return NextResponse.json(
{
data: {
id: operationRequestId,
generated: plainCodes.map((code) => ({
...(byMask.get(maskRedeemCode(code)) ?? {}),
code,
})),
},
},
}, { status: 201 });
{ status: 201 },
);
} catch (error) {
return adminErrorResponse(error);
}
@@ -1,20 +1,28 @@
import crypto from "node:crypto";
import { NextResponse } from "next/server";
import { z } from "zod";
import { requireAdminSession } from "@/lib/admin/auth";
import { requirePermission } from "@/lib/admin/auth";
import { isPostgresError, queryAdminRows } from "@/lib/admin/database";
import { adminErrorResponse } from "@/lib/admin/http";
import { adminErrorResponse, requireHighRiskAdminMutation } from "@/lib/admin/http";
import { suggestedEpayUrls } from "@/lib/epay/config";
import { encryptEpayKey } from "@/lib/epay/encryption";
import { assertConfiguredEpayUrl, assertPublicGatewayUrl } from "@/lib/epay/gateway-policy";
export const runtime = "nodejs";
const httpUrl = z.string().trim().min(1).max(2048).url().refine((value) => /^https?:\/\//i.test(value), "必须使用 HTTP(S)");
const epayUrl = z.string().trim().min(1).max(2048).url().refine((value) => {
try {
assertConfiguredEpayUrl(value);
return true;
} catch {
return false;
}
}, "必须使用 HTTPS;开发测试仅可显式启用 loopback HTTP");
const settingsSchema = z.object({
gatewayUrl: httpUrl,
gatewayUrl: epayUrl,
pid: z.string().trim().min(1).max(200),
notifyUrl: httpUrl,
returnUrl: httpUrl,
notifyUrl: epayUrl,
returnUrl: epayUrl,
siteName: z.string().trim().min(1).max(100),
chatEnabled: z.boolean(),
newKey: z.string().min(1).max(1000).optional(),
@@ -77,7 +85,7 @@ async function databaseRow() {
export async function GET() {
try {
await requireAdminSession("read");
await requirePermission("billing.orders.read");
const row = await databaseRow();
if (row) return NextResponse.json(publicSettings(row, "database"));
const settings = environmentSettings();
@@ -91,9 +99,14 @@ export async function GET() {
export async function PUT(request: Request) {
try {
const session = await requireAdminSession("write");
const session = await requireHighRiskAdminMutation(request, "billing.adjustments.write");
const parsed = settingsSchema.safeParse(await request.json().catch(() => null));
if (!parsed.success) return NextResponse.json({ error: "易支付配置参数不正确" }, { status: 400 });
try {
await assertPublicGatewayUrl(parsed.data.gatewayUrl);
} catch {
return NextResponse.json({ error: "易支付网关地址必须是允许的 HTTPS 公网地址" }, { status: 400 });
}
const existing = await databaseRow();
if (!existing && !parsed.data.newKey) {
@@ -110,7 +123,7 @@ export async function PUT(request: Request) {
`, [
session.user.id,
session.user.email,
session.role,
session.roles[0] ?? "admin",
crypto.randomUUID(),
parsed.data.gatewayUrl.replace(/\/+$/, ""),
parsed.data.pid,
@@ -1,8 +1,8 @@
import { NextResponse } from "next/server";
import { AdminAuthorizationError, requireAdminSession } from "@/lib/admin/auth";
import { adminErrorResponse } from "@/lib/admin/http";
import { AdminAuthorizationError } from "@/lib/admin/auth";
import { adminErrorResponse, requireAdminMutation } from "@/lib/admin/http";
import { epaySubmitUrl, readEpayConfig } from "@/lib/epay/config";
import { assertPublicGatewayUrl } from "@/lib/epay/gateway-policy";
import { probePublicEpayGateway } from "@/lib/epay/gateway-policy";
export const runtime = "nodejs";
@@ -10,31 +10,19 @@ function reachableStatus(status: number) {
return status >= 200 && status < 500;
}
export async function POST() {
export async function POST(request: Request) {
try {
await requireAdminSession("write");
await requireAdminMutation(request, "billing.adjustments.write");
const config = await readEpayConfig();
const submitUrl = epaySubmitUrl(config.gatewayUrl);
await assertPublicGatewayUrl(submitUrl);
const startedAt = performance.now();
let response = await fetch(submitUrl, {
method: "HEAD",
redirect: "manual",
signal: AbortSignal.timeout(8_000),
});
if (response.status === 405 || response.status === 501) {
response = await fetch(submitUrl, {
method: "GET",
redirect: "manual",
signal: AbortSignal.timeout(8_000),
});
}
const available = reachableStatus(response.status);
const status = await probePublicEpayGateway(submitUrl);
const available = reachableStatus(status);
return NextResponse.json({
available,
message: available ? "当前已保存的易支付配置可访问" : "当前已保存的易支付配置暂不可用",
latencyMs: Math.round(performance.now() - startedAt),
status: response.status,
status,
});
} catch (error) {
if (error instanceof AdminAuthorizationError) return adminErrorResponse(error);
+152
View File
@@ -0,0 +1,152 @@
import { NextResponse } from "next/server";
import { z } from "zod";
import { requirePermission } from "@/lib/admin/auth";
import { pageOffset, queryAdminRows } from "@/lib/admin/database";
import {
adminErrorResponse,
invalidQueryResponse,
parseListQuery,
requestId,
requireHighRiskAdminMutation,
} from "@/lib/admin/http";
export const runtime = "nodejs";
const adjustmentSchema = z.object({
id: z.string().uuid(),
action: z.enum(["retry_grant", "compensate", "record_refund"]),
expectedVersion: z.number().int().min(0),
reason: z.string().trim().min(1).max(500),
});
type Row = {
id: string;
order_no: string;
user_id: string;
email: string | null;
product_code: string | null;
product_version: number | null;
money_cents: number;
currency: string;
status: string;
grant_type: string | null;
grant_status: string;
grant_error: string | null;
adjustment_version: number;
refund_status: string;
refund_amount_cents: number | null;
refunded_at: Date | null;
paid_at: Date | null;
created_at: Date;
total_count: string;
};
type AdjustmentRow = {
id: string;
order_no: string;
status: string;
grant_status: string;
grant_error: string | null;
adjustment_version: number;
refund_status: string;
refund_amount_cents: number | null;
refunded_at: Date | null;
action_success: boolean;
};
export async function GET(request: Request) {
try {
await requirePermission("billing.orders.read");
const parsed = parseListQuery(request);
if (!parsed.success) return invalidQueryResponse(parsed.error.flatten());
const query = parsed.data.q ? `%${parsed.data.q}%` : null;
const rows = await queryAdminRows<Row>(
`select o.id,o.order_no,o.user_id,u.email,o.product_code,o.product_version,
o.money_cents,o.currency,o.status,o.grant_type,o.grant_status,o.grant_error,
o.adjustment_version,o.refund_status,o.refund_amount_cents,o.refunded_at,
o.paid_at,o.created_at,count(*) over()::text total_count
from public.payment_orders o
left join identity.users u on u.id=o.user_id
where ($1::text is null or o.order_no ilike $1 or u.email ilike $1 or o.user_id::text ilike $1)
and ($2::text is null or o.status=$2 or o.grant_status=$2 or o.refund_status=$2)
order by o.created_at desc limit $3 offset $4`,
[
query,
parsed.data.status ?? null,
parsed.data.pageSize,
pageOffset(parsed.data.page, parsed.data.pageSize),
],
);
return NextResponse.json({
data: rows.map((row) => ({
id: row.id,
orderNo: row.order_no,
userId: row.user_id,
email: row.email,
productCode: row.product_code,
productVersion: row.product_version,
moneyCents: row.money_cents,
currency: row.currency,
status: row.status,
grantType: row.grant_type,
grantStatus: row.grant_status,
grantError: row.grant_error,
adjustmentVersion: row.adjustment_version,
refundStatus: row.refund_status,
refundAmountCents: row.refund_amount_cents,
refundedAt: row.refunded_at?.toISOString() ?? null,
paidAt: row.paid_at?.toISOString() ?? null,
createdAt: row.created_at.toISOString(),
})),
total: Number(rows[0]?.total_count ?? 0),
});
} catch (error) {
return adminErrorResponse(error);
}
}
export async function POST(request: Request) {
try {
const session = await requireHighRiskAdminMutation(
request,
"billing.adjustments.write",
);
const parsed = adjustmentSchema.safeParse(
await request.json().catch(() => null),
);
if (!parsed.success) return invalidQueryResponse(parsed.error.flatten());
const rows = await queryAdminRows<AdjustmentRow>(
`select * from public.admin_adjust_order(
$1::uuid,$2::uuid,$3::text,$4::integer,$5::text,$6::text
)`,
[
session.user.id,
parsed.data.id,
parsed.data.action,
parsed.data.expectedVersion,
parsed.data.reason,
requestId(request),
],
);
const row = rows[0];
return NextResponse.json({
data: row
? {
id: row.id,
orderNo: row.order_no,
status: row.status,
grantStatus: row.grant_status,
grantError: row.grant_error,
adjustmentVersion: row.adjustment_version,
refundStatus: row.refund_status,
refundAmountCents: row.refund_amount_cents,
refundedAt: row.refunded_at?.toISOString() ?? null,
actionSuccess: row.action_success,
}
: null,
});
} catch (error) {
return adminErrorResponse(error);
}
}
+17 -100
View File
@@ -1,114 +1,31 @@
import { NextResponse } from "next/server";
import { z } from "zod";
import { requireAdminSession } from "@/lib/admin/auth";
import { queryAdminRows } from "@/lib/admin/database";
import { adminErrorResponse } from "@/lib/admin/http";
export const runtime = "nodejs";
const schema = z.object({
name: z.string().trim().min(1).max(80),
description: z.string().trim().max(500),
priceCents: z.number().int().positive().max(100_000_000),
credits: z.number().int().positive().max(10_000_000),
sortOrder: z.number().int().min(-100_000).max(100_000),
enabled: z.boolean(),
}).strict();
const updateSchema = schema.extend({ id: z.string().uuid() });
const idSchema = z.object({ id: z.string().uuid() }).strict();
const replacement = "/api/admin/products";
type PackageRow = {
id: string;
name: string;
description: string;
price_cents: number;
credits: number;
sort_order: number;
enabled: boolean;
created_at: Date;
updated_at: Date;
};
function output(row: PackageRow) {
return {
id: row.id,
name: row.name,
description: row.description,
priceCents: row.price_cents,
credits: row.credits,
sortOrder: row.sort_order,
enabled: row.enabled,
createdAt: row.created_at.toISOString(),
updatedAt: row.updated_at.toISOString(),
};
export function GET(request: Request) {
return NextResponse.redirect(new URL(replacement, request.url), 308);
}
export async function GET() {
try {
await requireAdminSession("read");
const rows = await queryAdminRows<PackageRow>(`
select id, name, description, price_cents, credits, sort_order, enabled, created_at, updated_at
from public.payment_packages
order by sort_order, created_at
`);
return NextResponse.json({ packages: rows.map(output) });
} catch (error) {
return adminErrorResponse(error);
}
function removedMutation() {
return NextResponse.json(
{
error: "旧套餐写接口已停用,请使用统一商品管理。",
replacement,
},
{ status: 410 },
);
}
export async function POST(request: Request) {
try {
const auth = await requireAdminSession("write");
const parsed = schema.safeParse(await request.json().catch(() => null));
if (!parsed.success) return NextResponse.json({ error: "套餐参数不正确" }, { status: 400 });
const p = parsed.data;
const rows = await queryAdminRows<PackageRow>(`
insert into public.payment_packages
(name, description, price_cents, credits, sort_order, enabled, created_by)
values ($1, $2, $3, $4, $5, $6, $7)
returning id, name, description, price_cents, credits, sort_order, enabled, created_at, updated_at
`, [p.name, p.description, p.priceCents, p.credits, p.sortOrder, p.enabled, auth.user.id]);
return NextResponse.json({ package: output(rows[0]) }, { status: 201 });
} catch (error) {
return adminErrorResponse(error);
}
export function POST() {
return removedMutation();
}
export async function PATCH(request: Request) {
try {
await requireAdminSession("write");
const parsed = updateSchema.safeParse(await request.json().catch(() => null));
if (!parsed.success) return NextResponse.json({ error: "套餐参数不正确" }, { status: 400 });
const p = parsed.data;
const rows = await queryAdminRows<PackageRow>(`
update public.payment_packages
set name = $2, description = $3, price_cents = $4, credits = $5,
sort_order = $6, enabled = $7, updated_at = clock_timestamp()
where id = $1
returning id, name, description, price_cents, credits, sort_order, enabled, created_at, updated_at
`, [p.id, p.name, p.description, p.priceCents, p.credits, p.sortOrder, p.enabled]);
if (!rows[0]) return NextResponse.json({ error: "套餐不存在" }, { status: 404 });
return NextResponse.json({ package: output(rows[0]) });
} catch (error) {
return adminErrorResponse(error);
}
export function PATCH() {
return removedMutation();
}
export async function DELETE(request: Request) {
try {
await requireAdminSession("write");
const parsed = idSchema.safeParse(await request.json().catch(() => null));
if (!parsed.success) return NextResponse.json({ error: "套餐参数不正确" }, { status: 400 });
const rows = await queryAdminRows<{ id: string }>(`
update public.payment_packages
set enabled = false, updated_at = clock_timestamp()
where id = $1
returning id
`, [parsed.data.id]);
if (!rows[0]) return NextResponse.json({ error: "套餐不存在" }, { status: 404 });
return NextResponse.json({ ok: true });
} catch (error) {
return adminErrorResponse(error);
}
export function DELETE() {
return removedMutation();
}
@@ -0,0 +1,105 @@
import { NextResponse } from "next/server";
import { z } from "zod";
import { requirePermission } from "@/lib/admin/auth";
import { pageOffset, queryAdminRows } from "@/lib/admin/database";
import { adminErrorResponse, invalidQueryResponse, parseListQuery, requestId, requireHighRiskAdminMutation } from "@/lib/admin/http";
export const runtime = "nodejs";
const entitlementSchema = z.object({
featureKey: z.enum(["chat.standard", "chat.premium", "rectification", "report.full", "report.export", "profile.extra"]),
allowanceType: z.enum(["access", "unlimited", "quota", "credits"]),
allowanceCount: z.number().int().positive().nullable(),
resetPeriod: z.enum(["none", "day", "month", "billing_period"]),
modelTier: z.enum(["standard", "premium", "internal"]).nullable().optional(),
fairUsePolicyId: z.string().trim().max(100).nullable().optional(),
metadata: z.record(z.string(), z.unknown()).default({}),
}).strict();
const saveSchema = z.object({
action: z.literal("save"),
id: z.string().uuid().nullable().optional(),
code: z.string().regex(/^[a-z][a-z0-9_]{1,79}$/),
name: z.string().trim().min(1).max(80),
description: z.string().trim().max(1000).default(""),
productType: z.enum(["credit_pack", "trial", "subscription"]),
billingPeriod: z.enum(["none", "day", "month", "year"]),
intervalCount: z.number().int().nonnegative(),
priceCents: z.number().int().positive(),
currency: z.string().length(3).default("CNY"),
enabled: z.boolean(),
sortOrder: z.number().int(),
oneTimePerUser: z.boolean(),
entitlements: z.array(entitlementSchema).min(1),
reason: z.string().trim().min(1).max(500),
}).strict();
const publishSchema = z.object({
action: z.literal("publish"),
id: z.string().uuid(),
reason: z.string().trim().min(1).max(500),
}).strict();
const mutationSchema = z.discriminatedUnion("action", [saveSchema, publishSchema]);
type ProductRow = {
id: string; code: string; version: number; name: string; description: string;
product_type: string; billing_period: string; interval_count: number; price_cents: number;
currency: string; enabled: boolean; status: string; sort_order: number; one_time_per_user: boolean;
effective_from: Date | null; effective_to: Date | null; updated_at: Date; entitlements: unknown;
total_count: string;
};
function output(row: ProductRow) {
return {
id: row.id, code: row.code, version: row.version, name: row.name, description: row.description,
productType: row.product_type, billingPeriod: row.billing_period, intervalCount: row.interval_count,
priceCents: row.price_cents, currency: row.currency, enabled: row.enabled, status: row.status,
sortOrder: row.sort_order, oneTimePerUser: row.one_time_per_user,
effectiveFrom: row.effective_from?.toISOString() ?? null,
effectiveTo: row.effective_to?.toISOString() ?? null,
updatedAt: row.updated_at.toISOString(), entitlements: row.entitlements,
};
}
export async function GET(request: Request) {
try {
await requirePermission("billing.products.read");
const parsed = parseListQuery(request);
if (!parsed.success) return invalidQueryResponse(parsed.error.flatten());
const q = parsed.data.q ? `%${parsed.data.q}%` : null;
const rows = await queryAdminRows<ProductRow>(`
select p.*,coalesce(jsonb_agg(jsonb_build_object(
'featureKey',e.feature_key,'allowanceType',e.allowance_type,'allowanceCount',e.allowance_count,
'resetPeriod',e.reset_period,'modelTier',e.model_tier,'fairUsePolicyId',e.fair_use_policy_id,'metadata',e.metadata
) order by e.feature_key) filter(where e.id is not null),'[]'::jsonb) entitlements,
count(*) over()::text total_count
from public.billing_products p left join public.product_entitlements e on e.product_id=p.id
where ($1::text is null or p.code ilike $1 or p.name ilike $1)
and ($2::text is null or p.status=$2)
group by p.id
order by p.updated_at desc limit $3 offset $4
`, [q, parsed.data.status ?? null, parsed.data.pageSize, pageOffset(parsed.data.page, parsed.data.pageSize)]);
return NextResponse.json({ data: rows.map(output), total: Number(rows[0]?.total_count ?? 0) });
} catch (error) { return adminErrorResponse(error); }
}
export async function POST(request: Request) {
try {
const body = mutationSchema.safeParse(await request.json().catch(() => null));
if (!body.success) return invalidQueryResponse(body.error.flatten());
const permission = body.data.action === "publish" ? "billing.products.publish" : "billing.products.write";
const session = await requireHighRiskAdminMutation(request, permission);
const rid = requestId(request);
if (body.data.action === "publish") {
const rows = await queryAdminRows<{ id: string }>("select public.admin_publish_product($1,$2,$3,$4) id", [session.user.id, body.data.id, body.data.reason, rid]);
return NextResponse.json({ data: { id: rows[0]!.id, requestId: rid } });
}
const value = body.data;
const rows = await queryAdminRows<{ id: string }>(`
select public.admin_save_product_draft($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14::jsonb,$15,$16) id
`, [session.user.id, value.id ?? null, value.code, value.name, value.description, value.productType,
value.billingPeriod, value.intervalCount, value.priceCents, value.currency, value.enabled, value.sortOrder,
value.oneTimePerUser, JSON.stringify(value.entitlements), value.reason, rid]);
return NextResponse.json({ data: { id: rows[0]!.id, requestId: rid } });
} catch (error) { return adminErrorResponse(error); }
}
@@ -0,0 +1,35 @@
import { NextResponse } from "next/server";
import { z } from "zod";
import { requirePermission } from "@/lib/admin/auth";
import { pageOffset, queryAdminRows } from "@/lib/admin/database";
import { adminErrorResponse, invalidQueryResponse, parseListQuery, requestId, requireHighRiskAdminMutation } from "@/lib/admin/http";
export const runtime = "nodejs";
const mutationSchema = z.object({
id: z.string().uuid(),
action: z.enum(["extend", "revoke"]),
days: z.number().int().min(1).max(3660).optional(),
expectedEndsAt: z.string().datetime(),
reason: z.string().trim().min(1).max(500),
}).strict();
type Row = { id:string; user_id:string; email:string|null; product_code:string; product_version:number; status:string; starts_at:Date; ends_at:Date; created_at:Date; total_count:string };
export async function GET(request: Request) {
try {
await requirePermission("billing.orders.read");
const parsed=parseListQuery(request); if(!parsed.success) return invalidQueryResponse(parsed.error.flatten());
const q=parsed.data.q?`%${parsed.data.q}%`:null;
const rows=await queryAdminRows<Row>(`select s.id,s.user_id,u.email,s.product_code,s.product_version,s.status,s.starts_at,s.ends_at,s.created_at,count(*) over()::text total_count
from public.user_subscriptions s left join identity.users u on u.id=s.user_id
where ($1::text is null or u.email ilike $1 or s.user_id::text ilike $1 or s.product_code ilike $1) and ($2::text is null or s.status=$2)
order by s.created_at desc limit $3 offset $4`,[q,parsed.data.status??null,parsed.data.pageSize,pageOffset(parsed.data.page,parsed.data.pageSize)]);
return NextResponse.json({data:rows.map(r=>({id:r.id,userId:r.user_id,email:r.email,productCode:r.product_code,productVersion:r.product_version,status:r.status,startsAt:r.starts_at.toISOString(),endsAt:r.ends_at.toISOString(),createdAt:r.created_at.toISOString()})),total:Number(rows[0]?.total_count??0)});
} catch(error){return adminErrorResponse(error);}
}
export async function POST(request: Request){
try{
const body=mutationSchema.safeParse(await request.json().catch(()=>null)); if(!body.success)return invalidQueryResponse(body.error.flatten());
const session=await requireHighRiskAdminMutation(request,"billing.adjustments.write"); const rid=requestId(request);
const rows=await queryAdminRows<{id:string}>("select public.admin_adjust_subscription($1,$2,$3,$4,$5,$6,$7) id",[session.user.id,body.data.id,body.data.action,body.data.days??null,body.data.expectedEndsAt,body.data.reason,rid]);
return NextResponse.json({data:{id:rows[0]!.id,requestId:rid}});
}catch(error){return adminErrorResponse(error);}
}
@@ -0,0 +1,7 @@
import { NextResponse } from "next/server";
import { requirePermission } from "@/lib/admin/auth";
import { pageOffset, queryAdminRows } from "@/lib/admin/database";
import { adminErrorResponse, invalidQueryResponse, parseListQuery } from "@/lib/admin/http";
export const runtime="nodejs";
type Row={id:string;user_id:string;email:string|null;request_id:string;feature_key:string;source:string;requested_model_id:string|null;actual_model_id:string|null;model_config_version:number|null;input_tokens:number;output_tokens:number;cost_microusd:string;duration_ms:number|null;created_at:Date;total_count:string};
export async function GET(request:Request){try{await requirePermission("billing.orders.read");const p=parseListQuery(request);if(!p.success)return invalidQueryResponse(p.error.flatten());const q=p.data.q?`%${p.data.q}%`:null;const rows=await queryAdminRows<Row>(`select l.id,l.user_id,u.email,l.request_id,l.feature_key,l.source,l.requested_model_id,l.actual_model_id,l.model_config_version,l.input_tokens,l.output_tokens,l.cost_microusd::text,l.duration_ms,l.created_at,count(*) over()::text total_count from public.usage_ledger l left join identity.users u on u.id=l.user_id where ($1::text is null or u.email ilike $1 or l.request_id ilike $1 or l.actual_model_id ilike $1) and ($2::text is null or l.source=$2 or l.feature_key=$2) order by l.created_at desc limit $3 offset $4`,[q,p.data.status??null,p.data.pageSize,pageOffset(p.data.page,p.data.pageSize)]);return NextResponse.json({data:rows.map(r=>({id:r.id,userId:r.user_id,email:r.email,requestId:r.request_id,featureKey:r.feature_key,source:r.source,requestedModelId:r.requested_model_id,actualModelId:r.actual_model_id,modelConfigVersion:r.model_config_version,inputTokens:r.input_tokens,outputTokens:r.output_tokens,costMicrousd:Number(r.cost_microusd),durationMs:r.duration_ms,createdAt:r.created_at.toISOString()})),total:Number(rows[0]?.total_count??0)});}catch(e){return adminErrorResponse(e)}}
+182 -15
View File
@@ -5,36 +5,189 @@ import { createAdminSupabaseClient } from "@/lib/supabase/admin";
import { createServerSupabaseClient } from "@/lib/supabase/server";
import { epaySign } from "@/lib/epay/sign";
import { readEpayAvailability } from "@/lib/epay/availability";
import { epaySubmitUrl, readEpayConfig, EpayConfigurationError } from "@/lib/epay/config";
import {
epaySubmitUrl,
readEpayConfig,
EpayConfigurationError,
} from "@/lib/epay/config";
import { assertPublicGatewayUrl } from "@/lib/epay/gateway-policy";
import { loadRuntimeFeatureFlags } from "@/lib/feature-flags";
export const runtime = "nodejs";
const schema = z.object({ packageId: z.string().uuid() });
const schema = z
.object({
productId: z.string().uuid().optional(),
packageId: z.string().uuid().optional(),
})
.refine(
(value) => Boolean(value.productId || value.packageId),
"product_required",
);
type ProductRow = {
id: string;
code: string;
version: number;
name: string;
description: string;
product_type: "credit_pack" | "trial" | "subscription";
billing_period: "none" | "day" | "month" | "year";
interval_count: number;
price_cents: number;
currency: string;
enabled: boolean;
status: string;
one_time_per_user: boolean;
effective_from: string | null;
effective_to: string | null;
};
type EntitlementRow = {
feature_key: string;
allowance_type: string;
allowance_count: number | null;
reset_period: string;
model_tier: string | null;
fair_use_policy_id: string | null;
metadata: unknown;
};
export async function POST(request: Request) {
try {
const client = await createServerSupabaseClient();
const { data: { user } } = await client.auth.getUser();
const {
data: { user },
} = await client.auth.getUser();
if (!user) return NextResponse.json({ error: "请先登录" }, { status: 401 });
const parsed = schema.safeParse(await request.json().catch(() => null));
if (!parsed.success) return NextResponse.json({ error: "请选择有效套餐" }, { status: 400 });
if (!parsed.success)
return NextResponse.json({ error: "请选择有效套餐" }, { status: 400 });
const availability = await readEpayAvailability();
if (!availability.enabled) return NextResponse.json({ error: "在线支付暂未开放", code: "EPAY_DISABLED" }, { status: 403 });
if (!availability.enabled)
return NextResponse.json(
{ error: "在线支付暂未开放", code: "EPAY_DISABLED" },
{ status: 403 },
);
const config = await readEpayConfig();
const submitUrl = epaySubmitUrl(config.gatewayUrl);
await assertPublicGatewayUrl(submitUrl);
const productId = parsed.data.productId ?? parsed.data.packageId!;
const admin = createAdminSupabaseClient();
const { data: pack, error: packError } = await admin.from("payment_packages").select("id,name,price_cents,credits,enabled").eq("id", parsed.data.packageId).eq("enabled", true).maybeSingle();
if (packError || !pack) return NextResponse.json({ error: "套餐不存在或已下架" }, { status: 404 });
const { data, error: productError } = await admin
.from("billing_products")
.select(
"id,code,version,name,description,product_type,billing_period,interval_count,price_cents,currency,enabled,status,one_time_per_user,effective_from,effective_to",
)
.eq("id", productId)
.maybeSingle();
const product = data as ProductRow | null;
const now = Date.now();
if (
productError ||
!product ||
!product.enabled ||
product.status !== "published" ||
(product.effective_from && Date.parse(product.effective_from) > now) ||
(product.effective_to && Date.parse(product.effective_to) <= now)
) {
return NextResponse.json(
{ error: "套餐不存在或已下架" },
{ status: 404 },
);
}
if (product.product_type !== "credit_pack") {
const flags = await loadRuntimeFeatureFlags(["billing.subscriptions"]);
if (!flags.get("billing.subscriptions")?.enabled) {
return NextResponse.json(
{ error: "订阅套餐暂未开放", code: "BILLING_SUBSCRIPTIONS_DISABLED" },
{ status: 403 },
);
}
}
if (product.one_time_per_user) {
const { data: redemption, error: redemptionError } = await admin
.from("user_product_redemptions")
.select("id")
.eq("user_id", user.id)
.eq("product_code", product.code)
.maybeSingle();
if (redemptionError)
return NextResponse.json(
{ error: "暂时无法核对体验资格" },
{ status: 500 },
);
if (redemption)
return NextResponse.json(
{ error: "该体验套餐每位用户仅限一次" },
{ status: 409 },
);
}
const { data: entitlementRows, error: entitlementError } = await admin
.from("product_entitlements")
.select(
"feature_key,allowance_type,allowance_count,reset_period,model_tier,fair_use_policy_id,metadata",
)
.eq("product_id", product.id);
if (entitlementError)
return NextResponse.json(
{ error: "暂时无法读取套餐权益" },
{ status: 500 },
);
const entitlements = ((entitlementRows ?? []) as EntitlementRow[]).map(
(item) => ({
featureKey: item.feature_key,
allowanceType: item.allowance_type,
allowanceCount: item.allowance_count,
resetPeriod: item.reset_period,
modelTier: item.model_tier,
fairUsePolicyId: item.fair_use_policy_id,
metadata: item.metadata,
}),
);
const credits =
entitlements.find((item) => item.allowanceType === "credits")
?.allowanceCount ?? 0;
const productSnapshot = {
id: product.id,
code: product.code,
version: product.version,
name: product.name,
description: product.description,
productType: product.product_type,
billingPeriod: product.billing_period,
intervalCount: product.interval_count,
priceCents: product.price_cents,
currency: product.currency,
oneTimePerUser: product.one_time_per_user,
};
const orderNo = `JY${Date.now().toString(36)}${crypto.randomBytes(10).toString("hex")}`;
const { error: orderError } = await admin.from("payment_orders").insert({ order_no: orderNo, user_id: user.id, package_id: pack.id, money_cents: pack.price_cents, credits: pack.credits });
if (orderError) return NextResponse.json({ error: "创建订单失败" }, { status: 500 });
const { error: orderError } = await admin.from("payment_orders").insert({
order_no: orderNo,
user_id: user.id,
package_id: product.product_type === "credit_pack" ? product.id : null,
product_id: product.id,
product_code: product.code,
product_version: product.version,
product_snapshot: productSnapshot,
entitlement_snapshot: entitlements,
money_cents: product.price_cents,
currency: product.currency,
credits,
grant_type:
product.product_type === "credit_pack"
? "credits"
: product.product_type,
grant_status: "pending",
});
if (orderError)
return NextResponse.json({ error: "创建订单失败" }, { status: 500 });
const params = {
money: (pack.price_cents / 100).toFixed(2),
name: pack.name,
money: (product.price_cents / 100).toFixed(2),
name: product.name,
notify_url: config.notifyUrl,
out_trade_no: orderNo,
pid: config.pid,
@@ -42,12 +195,26 @@ export async function POST(request: Request) {
sitename: config.siteName,
type: "alipay",
};
const signedParams = { ...params, sign: epaySign(params, config.key), sign_type: "MD5" };
const signedParams = {
...params,
sign: epaySign(params, config.key),
sign_type: "MD5",
};
const payUrl = new URL(submitUrl);
for (const [name, value] of Object.entries(signedParams)) payUrl.searchParams.set(name, value);
return NextResponse.json({ orderNo, payUrl: payUrl.toString(), qrCode: null });
for (const [name, value] of Object.entries(signedParams))
payUrl.searchParams.set(name, value);
return NextResponse.json({
orderNo,
payUrl: payUrl.toString(),
qrCode: null,
product: productSnapshot,
});
} catch (error) {
if (error instanceof EpayConfigurationError) return NextResponse.json({ error: "易支付尚未配置", code: "EPAY_NOT_CONFIGURED" }, { status: 503 });
if (error instanceof EpayConfigurationError)
return NextResponse.json(
{ error: "易支付尚未配置", code: "EPAY_NOT_CONFIGURED" },
{ status: 503 },
);
return NextResponse.json({ error: "创建支付失败" }, { status: 500 });
}
}
@@ -6,7 +6,7 @@ export const runtime = "nodejs";
const notify = createEpayNotifyHandler({
readConfig: readEpayConfig,
settle: async (args) => await createAdminSupabaseClient().rpc("settle_epay_order", args),
settle: async (args) => await createAdminSupabaseClient().rpc("settle_order", args),
});
export async function POST(request: Request) { return notify(request); }
@@ -1,5 +1,43 @@
import { NextResponse } from "next/server";
import { createAdminSupabaseClient } from "@/lib/supabase/admin";
import { createServerSupabaseClient } from "@/lib/supabase/server";
export const runtime = "nodejs";
export async function GET(request: Request) { const client = await createServerSupabaseClient(); const { data: { user } } = await client.auth.getUser(); if (!user) return NextResponse.json({ error: "请先登录" }, { status: 401 }); const orderNo = new URL(request.url).searchParams.get("orderNo"); if (!orderNo) return NextResponse.json({ error: "缺少订单号" }, { status: 400 }); const { data, error } = await createAdminSupabaseClient().from("payment_orders").select("order_no,status,credits,paid_at").eq("order_no", orderNo).eq("user_id", user.id).maybeSingle(); if (error) return NextResponse.json({ error: "暂时无法查询订单" }, { status: 500 }); if (!data) return NextResponse.json({ error: "订单不存在" }, { status: 404 }); return NextResponse.json({ orderNo: data.order_no, status: data.status, credits: data.credits, paidAt: data.paid_at }); }
export async function GET(request: Request) {
const client = await createServerSupabaseClient();
const { data: { user } } = await client.auth.getUser();
if (!user) return NextResponse.json({ error: "请先登录" }, { status: 401 });
const orderNo = new URL(request.url).searchParams.get("orderNo");
if (!orderNo) return NextResponse.json({ error: "缺少订单号" }, { status: 400 });
const admin = createAdminSupabaseClient();
const { data, error } = await admin
.from("payment_orders")
.select("order_no,status,credits,paid_at,grant_type,grant_status,grant_reference_id,grant_error,product_snapshot")
.eq("order_no", orderNo)
.eq("user_id", user.id)
.maybeSingle();
if (error) return NextResponse.json({ error: "暂时无法查询订单" }, { status: 500 });
if (!data) return NextResponse.json({ error: "订单不存在" }, { status: 404 });
let subscription = null;
if (data.grant_reference_id && (data.grant_type === "subscription" || data.grant_type === "trial")) {
const result = await admin
.from("user_subscriptions")
.select("id,status,starts_at,ends_at,product_code,product_version")
.eq("id", data.grant_reference_id)
.maybeSingle();
if (!result.error) subscription = result.data;
}
return NextResponse.json({
orderNo: data.order_no,
status: data.status,
credits: data.credits,
paidAt: data.paid_at,
grantType: data.grant_type,
grantStatus: data.grant_status,
grantError: data.grant_error,
product: data.product_snapshot,
subscription,
});
}
+99 -11
View File
@@ -1,28 +1,116 @@
import { NextResponse } from "next/server";
import { readEpayAvailability } from "@/lib/epay/availability";
import { loadRuntimeFeatureFlags } from "@/lib/feature-flags";
import { createAdminSupabaseClient } from "@/lib/supabase/admin";
export const runtime = "nodejs";
type EntitlementRow = {
feature_key: string;
allowance_type: string;
allowance_count: number | null;
reset_period: string;
model_tier: string | null;
fair_use_policy_id: string | null;
metadata: unknown;
};
type EntitlementRowWithProduct = EntitlementRow & { product_id: string };
type ProductRow = {
id: string;
code: string;
version: number;
name: string;
description: string;
product_type: "credit_pack" | "trial" | "subscription";
billing_period: "none" | "day" | "month" | "year";
interval_count: number;
price_cents: number;
currency: string;
sort_order: number;
effective_from: string | null;
effective_to: string | null;
};
export async function GET() {
const availability = await readEpayAvailability();
if (!availability.enabled) return NextResponse.json({ enabled: false, packages: [] });
if (!availability.enabled)
return NextResponse.json({ enabled: false, packages: [] });
const { data, error } = await createAdminSupabaseClient()
.from("payment_packages")
.select("id,name,description,price_cents,credits,sort_order")
const flags = await loadRuntimeFeatureFlags(["billing.subscriptions"]);
const subscriptionsEnabled =
flags.get("billing.subscriptions")?.enabled ?? false;
const admin = createAdminSupabaseClient();
const { data, error } = await admin
.from("billing_products")
.select(
"id,code,version,name,description,product_type,billing_period,interval_count,price_cents,currency,sort_order,effective_from,effective_to",
)
.eq("enabled", true)
.eq("status", "published")
.order("sort_order")
.order("created_at");
if (error) return NextResponse.json({ enabled: false, packages: [] });
const now = Date.now();
const products = ((data ?? []) as ProductRow[]).filter(
(product) =>
(!product.effective_from || Date.parse(product.effective_from) <= now) &&
(!product.effective_to || Date.parse(product.effective_to) > now) &&
(subscriptionsEnabled || product.product_type === "credit_pack"),
);
const productIds = products.map((product) => product.id);
const { data: entitlementData, error: entitlementError } =
productIds.length === 0
? { data: [] as EntitlementRowWithProduct[], error: null }
: await admin
.from("product_entitlements")
.select(
"product_id,feature_key,allowance_type,allowance_count,reset_period,model_tier,fair_use_policy_id,metadata",
)
.in("product_id", productIds);
if (entitlementError)
return NextResponse.json({ enabled: false, packages: [] });
const entitlementsByProduct = new Map<string, EntitlementRow[]>();
for (const entitlement of (entitlementData ??
[]) as EntitlementRowWithProduct[]) {
const current = entitlementsByProduct.get(entitlement.product_id) ?? [];
current.push(entitlement);
entitlementsByProduct.set(entitlement.product_id, current);
}
return NextResponse.json({
enabled: true,
packages: (data || []).map((item) => ({
id: item.id,
name: item.name,
description: item.description,
priceCents: item.price_cents,
credits: item.credits,
})),
packages: products.map((product) => {
const entitlements = entitlementsByProduct.get(product.id) ?? [];
const creditEntitlement = entitlements.find(
(item) => item.allowance_type === "credits",
);
return {
id: product.id,
productId: product.id,
code: product.code,
version: product.version,
name: product.name,
description: product.description,
productType: product.product_type,
billingPeriod: product.billing_period,
intervalCount: product.interval_count,
priceCents: product.price_cents,
currency: product.currency,
credits: creditEntitlement?.allowance_count ?? 0,
entitlements: entitlements.map((item) => ({
featureKey: item.feature_key,
allowanceType: item.allowance_type,
allowanceCount: item.allowance_count,
resetPeriod: item.reset_period,
modelTier: item.model_tier,
fairUsePolicyId: item.fair_use_policy_id,
metadata: item.metadata,
})),
};
}),
});
}
+19 -7
View File
@@ -183,6 +183,16 @@ type Account = {
isAdmin: boolean;
adminUrl: string | null;
rectificationPriceCredits: number;
activeSubscription: {
id: string;
status: string;
startsAt: string;
endsAt: string;
productCode: string;
productVersion: number;
product: { name?: string; productType?: string } | null;
entitlements: unknown;
} | null;
hasConfirmedBirthTime: boolean;
hasUsableBirthTime: boolean;
rectificationCase: AccountRectificationCaseState | null;
@@ -930,7 +940,7 @@ export default function Home() {
const [redeemMessage, setRedeemMessage] = useState("");
const [redeeming, setRedeeming] = useState(false);
const [paymentEnabled, setPaymentEnabled] = useState(false);
const [paymentPackages, setPaymentPackages] = useState<Array<{ id: string; name: string; description: string; priceCents: number; credits: number }>>([]);
const [paymentPackages, setPaymentPackages] = useState<Array<{ id: string; name: string; description: string; productType: "credit_pack" | "trial" | "subscription"; billingPeriod: string; intervalCount: number; priceCents: number; credits: number }>>([]);
const [paymentOrder, setPaymentOrder] = useState<{ orderNo: string; payUrl: string | null; qrCode: string | null; status: string } | null>(null);
const [paymentError, setPaymentError] = useState("");
const [payingPackageId, setPayingPackageId] = useState<string | null>(null);
@@ -1254,6 +1264,7 @@ export default function Home() {
hasConfirmedBirthTime: previewProfile.birthTimeStatus === "confirmed",
hasUsableBirthTime: previewProfile.birthTimeStatus === "accepted" || previewProfile.birthTimeStatus === "confirmed",
rectificationCase: null,
activeSubscription: null,
profile: previewProfile,
});
setModelCatalog(previewModelCatalog);
@@ -2415,7 +2426,7 @@ export default function Home() {
? { kind: "consult" as const, mode: "general_no_birth_time" as const, time: null }
: initialConsultationRoute;
if (account.credits <= 0) {
if (account.credits <= 0 && !account.activeSubscription) {
openAccountDialog("redeem", creditTrigger.current);
return false;
}
@@ -2520,6 +2531,7 @@ export default function Home() {
headers: { "content-type": "application/json" },
body: JSON.stringify({
requestId,
sessionId: currentSession.id,
modelId: currentSession.modelId,
name: profile.name,
consultationMode: consultationRoute.mode,
@@ -3082,7 +3094,7 @@ export default function Home() {
? onboardingStep === "name"
? presetMessageFinished ? "输入你的称呼" : "Jyotisha 正在输入…"
: "请先完成上方资料"
: account.credits === 0
: account.credits === 0 && !account.activeSubscription
? "余额不足,发送时将打开兑换码"
: "例如:未来半年是否适合换工作?"}
rows={1}
@@ -3235,7 +3247,7 @@ export default function Home() {
{activeAccountDialog === "redeem" && (
<>
<div className="redeem-balance"><span>当前余额</span><strong>{account.credits} 点</strong></div>
<div className="redeem-balance"><span>{account.activeSubscription ? "当前会员" : "当前余额"}</span><strong>{account.activeSubscription ? `${account.activeSubscription.product?.name || account.activeSubscription.productCode} · ${new Date(account.activeSubscription.endsAt).toLocaleDateString("zh-CN")} 到期` : `${account.credits} 点`}</strong></div>
<form className="redeem-form account-redeem-form" onSubmit={redeem}>
<label htmlFor="redeem-code">兑换码</label>
<div>
@@ -3246,9 +3258,9 @@ export default function Home() {
{redeemMessage && <p className="form-success" role="status">{redeemMessage}</p>}
</form>
{paymentEnabled && <div className="payment-packages">
<h3>套餐充值</h3>
{paymentPackages.map((item) => <div className="payment-package" key={item.id}><div><strong>{item.name}</strong><span>{item.description || `${item.credits} 点`}</span></div><b>¥{(item.priceCents / 100).toFixed(2)}</b><button className="button-secondary" type="button" onClick={() => void createPayment(item.id)} disabled={Boolean(payingPackageId)}>{payingPackageId === item.id ? "创建中" : "立即支付"}</button></div>)}
{paymentOrder && <div className="payment-status" role="status"><p>订单 {paymentOrder.orderNo}:{paymentOrder.status === "paid" ? "支付成功,点数已到账" : "等待支付"}</p></div>}
<h3>套餐与会员</h3>
{paymentPackages.map((item) => <div className="payment-package" key={item.id}><div><strong>{item.name}</strong><span>{item.description || (item.productType === "credit_pack" ? `${item.credits} 点` : "会员权益")}</span></div><b>¥{(item.priceCents / 100).toFixed(2)}</b><button className="button-secondary" type="button" onClick={() => void createPayment(item.id)} disabled={Boolean(payingPackageId)}>{payingPackageId === item.id ? "创建中" : "立即支付"}</button></div>)}
{paymentOrder && <div className="payment-status" role="status"><p>订单 {paymentOrder.orderNo}:{paymentOrder.status === "paid" ? "支付成功,权益已到账" : "等待支付"}</p></div>}
</div>}
{paymentError && <p className="form-error" role="alert">{paymentError}</p>}
</>
+100 -30
View File
@@ -1,8 +1,7 @@
import "server-only";
import { createAdminSupabaseClient } from "@/lib/supabase/admin";
import type { IdentityUser } from "@/modules/identity/contracts";
import type { AdminRole } from "./auth";
import { queryAdminRows } from "./database";
import type { AdminSession } from "./auth";
export type RedemptionCodeRecord = {
id: string;
@@ -23,57 +22,128 @@ type RpcCodeRow = {
id: string;
code_mask: string;
credits: number;
expires_at: string | null;
expires_at: Date | string | null;
note: string | null;
created_at: string;
created_at: Date | string;
redeemed_by: string | null;
redeemed_email: string | null;
redeemed_at: string | null;
redeemed_at: Date | string | null;
revoked_by: string | null;
revoked_at: string | null;
revoked_at: Date | string | null;
};
type CodeRpcArgs = {
admin_create_redemption_codes: {
p_codes: unknown;
p_reason: string;
};
admin_update_redemption_code: {
p_code_id: string;
p_set_note: boolean;
p_note: string | null;
p_set_expires_at: boolean;
p_expires_at: string | null;
p_reason: string;
};
admin_revoke_redemption_code: {
p_code_id: string;
p_reason: string;
};
};
function isoTimestamp(value: Date | string | null): string | null {
return value instanceof Date ? value.toISOString() : value;
}
export function codeStatus(row: RpcCodeRow): RedemptionCodeRecord["status"] {
const expiresAt = isoTimestamp(row.expires_at);
if (row.redeemed_at) return "redeemed";
if (row.revoked_at) return "revoked";
if (row.expires_at && Date.parse(row.expires_at) <= Date.now()) return "expired";
if (expiresAt && Date.parse(expiresAt) <= Date.now()) return "expired";
return "available";
}
export function mapCode(row: RpcCodeRow): RedemptionCodeRecord {
const expiresAt = isoTimestamp(row.expires_at);
const redeemedAt = isoTimestamp(row.redeemed_at);
const revokedAt = isoTimestamp(row.revoked_at);
return {
id: row.id,
mask: row.code_mask,
credits: row.credits,
expiresAt: row.expires_at,
expiresAt,
note: row.note,
createdAt: row.created_at,
createdAt: isoTimestamp(row.created_at)!,
redeemedBy: row.redeemed_by,
redeemedEmail: row.redeemed_email,
redeemedAt: row.redeemed_at,
redeemedAt,
revokedBy: row.revoked_by,
revokedAt: row.revoked_at,
status: codeStatus(row),
revokedAt,
status: codeStatus({
...row,
expires_at: expiresAt,
created_at: isoTimestamp(row.created_at)!,
redeemed_at: redeemedAt,
revoked_at: revokedAt,
}),
};
}
export async function runCodeRpc(
functionName:
| "admin_create_redemption_codes"
| "admin_update_redemption_code"
| "admin_revoke_redemption_code",
session: { user: IdentityUser; role: AdminRole },
export async function runCodeRpc<T extends keyof CodeRpcArgs>(
functionName: T,
session: Pick<AdminSession, "user" | "roles">,
id: string,
args: Record<string, unknown>,
args: CodeRpcArgs[T],
): Promise<RedemptionCodeRecord[]> {
const admin = createAdminSupabaseClient();
const { data, error } = await admin.rpc(functionName, {
p_actor_user_id: session.user.id,
p_actor_email: session.user.email,
p_actor_role: session.role,
p_request_id: id,
...args,
});
if (error) throw new Error(error.message);
return ((data ?? []) as RpcCodeRow[]).map(mapCode);
const common = [
session.user.id,
session.user.email,
session.roles[0] ?? "admin",
id,
];
let rows: RpcCodeRow[] = [];
switch (functionName) {
case "admin_create_redemption_codes": {
const input = args as CodeRpcArgs["admin_create_redemption_codes"];
rows = await queryAdminRows<RpcCodeRow>(
`select * from public.admin_create_redemption_codes(
$1::uuid,$2::text,$3::text,$4::text,$5::jsonb,$6::text
)`,
[...common, input.p_codes, input.p_reason],
);
break;
}
case "admin_update_redemption_code": {
const input = args as CodeRpcArgs["admin_update_redemption_code"];
rows = await queryAdminRows<RpcCodeRow>(
`select * from public.admin_update_redemption_code(
$1::uuid,$2::text,$3::text,$4::text,$5::uuid,
$6::boolean,$7::text,$8::boolean,$9::timestamptz,$10::text
)`,
[
...common,
input.p_code_id,
input.p_set_note,
input.p_note,
input.p_set_expires_at,
input.p_expires_at,
input.p_reason,
],
);
break;
}
case "admin_revoke_redemption_code": {
const input = args as CodeRpcArgs["admin_revoke_redemption_code"];
rows = await queryAdminRows<RpcCodeRow>(
`select * from public.admin_revoke_redemption_code(
$1::uuid,$2::text,$3::text,$4::text,$5::uuid,$6::text
)`,
[...common, input.p_code_id, input.p_reason],
);
break;
}
}
return rows.map(mapCode);
}
+112 -23
View File
@@ -6,15 +6,44 @@ const creditResultSchema = z.object({
error_code: z.string().nullable().optional(),
});
const authorizationSchema = z.object({
success: z.boolean(),
reservation_id: z.string().uuid().nullable(),
source: z.enum(["subscription", "credits"]).nullable(),
credits: z.number().int().nullable(),
subscription_id: z.string().uuid().nullable(),
reason: z.string().nullable(),
retry_after_seconds: z.number().int().nullable(),
});
const settlementSchema = z.object({
success: z.boolean(),
reservation_id: z.string().uuid().nullable(),
credits: z.number().int().nullable(),
error_code: z.string().nullable(),
});
type CreditRpcName = "begin_consultation_credit" | "complete_consultation_credit" | "cancel_consultation_credit";
type AccountingClient = {
rpc(
rpcName: CreditRpcName,
args: { p_user_id: string; p_request_id: string },
): PromiseLike<{ data: unknown; error: { message: string } | null }>;
rpc(rpcName: string, args: Record<string, unknown>): PromiseLike<{
data: unknown;
error: { message: string } | null;
}>;
};
export type CreditResult = z.infer<typeof creditResultSchema>;
export type UsageAuthorization = z.infer<typeof authorizationSchema>;
export type UsageSettlement = z.infer<typeof settlementSchema>;
export type ActualUsage = {
eventKey: string;
actualModelId: string;
modelConfigVersion?: number;
inputTokens: number;
outputTokens: number;
costMicrousd: number;
durationMs: number;
};
export class CreditRpcError extends Error {
readonly code: string;
@@ -26,32 +55,92 @@ export class CreditRpcError extends Error {
}
}
function first(value: unknown) {
return Array.isArray(value) ? value[0] : value;
}
async function runRpc<T>(
accounting: AccountingClient,
rpcName: string,
args: Record<string, unknown>,
schema: z.ZodType<T>,
): Promise<T> {
let lastError = "unknown_billing_error";
for (let attempt = 1; attempt <= 3; attempt += 1) {
try {
const { data, error } = await accounting.rpc(rpcName, args);
const parsed = schema.safeParse(first(data));
if (!error && parsed.success) return parsed.data;
lastError = error?.message || "invalid_billing_response";
} catch (error) {
lastError = error instanceof Error ? error.message : "billing_request_failed";
}
if (attempt < 3) await new Promise((resolve) => setTimeout(resolve, attempt * 150));
}
throw new CreditRpcError(lastError);
}
export async function runCreditRpc(
accounting: AccountingClient,
rpcName: CreditRpcName,
userId: string,
requestId: string,
): Promise<CreditResult> {
let lastError = "unknown_credit_error";
return runRpc(accounting, rpcName, {
p_user_id: userId,
p_request_id: requestId,
}, creditResultSchema);
}
for (let attempt = 1; attempt <= 3; attempt += 1) {
try {
const { data, error } = await accounting.rpc(rpcName, {
p_user_id: userId,
p_request_id: requestId,
});
const candidate = Array.isArray(data) ? data[0] : data;
const parsed = creditResultSchema.safeParse(candidate);
if (!error && parsed.success) return parsed.data;
lastError = error?.message || "invalid_credit_response";
} catch (error) {
lastError = error instanceof Error ? error.message : "credit_request_failed";
}
export async function authorizeUsage(
accounting: AccountingClient,
input: {
userId: string;
requestId: string;
featureKey: "chat.standard" | "chat.premium" | "rectification" | "report.full" | "report.export";
requestedModelId: string;
creditCost: number;
},
): Promise<UsageAuthorization> {
return runRpc(accounting, "authorize_usage", {
p_user_id: input.userId,
p_feature_key: input.featureKey,
p_requested_model_id: input.requestedModelId,
p_request_id: input.requestId,
p_credit_cost: input.creditCost,
}, authorizationSchema);
}
if (attempt < 3) {
await new Promise((resolve) => setTimeout(resolve, attempt * 150));
}
}
export async function completeUsage(
accounting: AccountingClient,
userId: string,
requestId: string,
usage: ActualUsage,
): Promise<UsageSettlement> {
return runRpc(accounting, "complete_usage", {
p_user_id: userId,
p_request_id: requestId,
p_actual_usage: {
eventKey: usage.eventKey,
actualModelId: usage.actualModelId,
modelConfigVersion: usage.modelConfigVersion,
inputTokens: Math.max(0, Math.trunc(usage.inputTokens)),
outputTokens: Math.max(0, Math.trunc(usage.outputTokens)),
costMicrousd: Math.max(0, Math.trunc(usage.costMicrousd)),
durationMs: Math.max(0, Math.trunc(usage.durationMs)),
},
}, settlementSchema);
}
throw new CreditRpcError(lastError);
export async function releaseUsage(
accounting: AccountingClient,
userId: string,
requestId: string,
reason: string,
): Promise<UsageSettlement> {
return runRpc(accounting, "release_usage", {
p_user_id: userId,
p_request_id: requestId,
p_reason: reason,
}, settlementSchema);
}
+18 -15
View File
@@ -1,3 +1,4 @@
import { assertConfiguredEpayUrl } from "./gateway-policy";
import { decryptEpayKey } from "./encryption-core";
export type EpaySettingsRow = {
@@ -17,23 +18,25 @@ export class EpayConfigurationError extends Error {
}
}
function validHttpUrl(value: string, label: string) {
function validEpayUrl(value: string, label: string, env: NodeJS.ProcessEnv) {
try {
const url = new URL(value);
if (!/^https?:$/.test(url.protocol)) throw new Error();
return url;
return assertConfiguredEpayUrl(value, env);
} catch {
throw new EpayConfigurationError(`${label} 无效`);
}
}
export function suggestedEpayUrls(siteAddress = process.env.SITE_ADDRESS) {
export function suggestedEpayUrls(
siteAddress = process.env.SITE_ADDRESS,
env: NodeJS.ProcessEnv = process.env,
) {
let base: URL;
try {
base = new URL(siteAddress?.trim() || "http://localhost:3000");
if (!/^https?:$/.test(base.protocol)) throw new Error();
base = validEpayUrl(siteAddress?.trim() || "", "站点地址", env);
} catch {
base = new URL("http://localhost:3000");
const allowLoopbackHttp = (env.NODE_ENV === "development" || env.NODE_ENV === "test")
&& ["true", "1"].includes(env.EPAY_ALLOW_INSECURE_LOOPBACK_HTTP?.trim().toLowerCase() || "");
base = new URL(allowLoopbackHttp ? "http://localhost:3000" : "https://localhost:3000");
}
return {
notifyUrl: new URL("/api/payment/epay/notify", base).toString(),
@@ -54,13 +57,13 @@ function completeConfig(values: {
returnUrl: string;
siteName: string;
chatEnabled: boolean;
}) {
}, env: NodeJS.ProcessEnv) {
if (!values.gateway || !values.pid || !values.key || !values.notifyUrl || !values.returnUrl || !values.siteName) {
throw new EpayConfigurationError();
}
const gatewayUrl = validHttpUrl(values.gateway.replace(/\/+$/, ""), "易支付网关地址");
validHttpUrl(values.notifyUrl, "异步通知地址");
validHttpUrl(values.returnUrl, "支付返回地址");
const gatewayUrl = validEpayUrl(values.gateway.replace(/\/+$/, ""), "易支付网关地址", env);
validEpayUrl(values.notifyUrl, "异步通知地址", env);
validEpayUrl(values.returnUrl, "支付返回地址", env);
return { gatewayUrl, pid: values.pid, key: values.key, notifyUrl: values.notifyUrl, returnUrl: values.returnUrl, siteName: values.siteName, chatEnabled: values.chatEnabled };
}
@@ -83,10 +86,10 @@ export async function resolveEpayConfig(
returnUrl: row.return_url.trim(),
siteName: row.site_name.trim(),
chatEnabled: row.chat_enabled,
});
}, env);
}
const defaults = suggestedEpayUrls(env.SITE_ADDRESS);
const defaults = suggestedEpayUrls(env.SITE_ADDRESS, env);
return completeConfig({
gateway: env.EPAY_GATEWAY_URL?.trim() || "",
pid: env.EPAY_PID?.trim() || "",
@@ -95,7 +98,7 @@ export async function resolveEpayConfig(
returnUrl: env.EPAY_RETURN_URL?.trim() || defaults.returnUrl,
siteName: env.EPAY_SITE_NAME?.trim() || "Jyotisha",
chatEnabled: environmentChatEnabled(env.EPAY_CHAT_ENABLED),
});
}, env);
}
export function epaySubmitUrl(gatewayUrl: URL) {
+223 -12
View File
@@ -1,4 +1,5 @@
import { promises as dns } from "node:dns";
import https from "node:https";
import { isIP } from "node:net";
const blockedHostnames = new Set([
@@ -6,11 +7,21 @@ const blockedHostnames = new Set([
"localhost.localdomain",
"metadata.google.internal",
]);
const blockedHostnameSuffixes = [".localhost", ".local", ".internal", ".lan", ".home", ".arpa"];
const defaultModelProviderOrigins = new Set(["https://api.openai.com"]);
const defaultLookup: HostLookup = (hostname, options) => dns.lookup(hostname, options);
type ResolvedAddress = Readonly<{ address: string; family: number }>;
type HostLookup = (
hostname: string,
options: { all: true; verbatim: true },
) => Promise<readonly ResolvedAddress[]>;
type ModelProviderEnvironment = Readonly<Record<string, string | undefined>>;
function blockedIpv4(address: string) {
const parts = address.split(".").map(Number);
if (parts.length !== 4 || parts.some((part) => !Number.isInteger(part) || part < 0 || part > 255)) return true;
const [a, b] = parts;
const [a, b, c] = parts;
return a === 0
|| a === 10
|| a === 127
@@ -19,7 +30,11 @@ function blockedIpv4(address: string) {
|| (a === 172 && b >= 16 && b <= 31)
|| (a === 192 && b === 0)
|| (a === 192 && b === 168)
|| (a === 192 && b === 88 && c === 99)
|| (a === 192 && b === 0 && c === 2)
|| (a === 198 && (b === 18 || b === 19))
|| (a === 198 && b === 51 && c === 100)
|| (a === 203 && b === 0 && c === 113)
|| a >= 224;
}
@@ -34,31 +49,227 @@ export function isPublicEpayAddress(address: string) {
&& !normalized.startsWith("fc")
&& !normalized.startsWith("fd")
&& !/^fe[89ab]/.test(normalized)
&& !normalized.startsWith("ff")
&& !normalized.startsWith("2001:db8:");
}
function normalizedHostname(url: URL) {
const hostname = url.hostname.toLowerCase().replace(/\.$/, "");
return hostname.startsWith("[") && hostname.endsWith("]") ? hostname.slice(1, -1) : hostname;
}
function isLoopbackHostname(hostname: string) {
if (hostname === "localhost" || hostname.endsWith(".localhost")) return true;
if (isIP(hostname) === 4) return hostname.split(".")[0] === "127";
return hostname === "::1" || hostname.toLowerCase().startsWith("::ffff:127.");
}
function insecureLoopbackHttpEnabled(environment: ModelProviderEnvironment) {
const nodeEnvironment = environment.NODE_ENV?.trim().toLowerCase();
const enabled = environment.EPAY_ALLOW_INSECURE_LOOPBACK_HTTP?.trim().toLowerCase();
return (nodeEnvironment === "development" || nodeEnvironment === "test")
&& (enabled === "true" || enabled === "1");
}
export function assertConfiguredEpayUrl(
value: URL | string,
environment: ModelProviderEnvironment = process.env,
) {
const url = value instanceof URL ? value : new URL(value);
const hostname = normalizedHostname(url);
const insecureLoopback = url.protocol === "http:"
&& insecureLoopbackHttpEnabled(environment)
&& isLoopbackHostname(hostname);
if ((url.protocol !== "https:" && !insecureLoopback) || url.username || url.password) {
throw new Error("易支付地址必须使用 HTTPS;开发测试仅可显式启用 loopback HTTP");
}
return url;
}
export function assertPublicEpayGateway(value: URL | string) {
const url = value instanceof URL ? value : new URL(value);
const hostname = url.hostname.toLowerCase().replace(/\.$/, "");
const hostname = normalizedHostname(url);
if (!/^https?:$/.test(url.protocol)
|| url.username
|| url.password
|| blockedHostnames.has(hostname)
|| hostname.endsWith(".localhost")
|| hostname.endsWith(".local")
|| blockedHostnameSuffixes.some((suffix) => hostname.endsWith(suffix))
|| (!isIP(hostname) && !hostname.includes("."))
|| (isIP(hostname) && !isPublicEpayAddress(hostname))) {
throw new Error("易支付网关地址不允许指向本机或内网");
throw new Error("网关地址不允许指向本机、内网或内部域名");
}
return url;
}
export async function assertPublicGatewayUrl(value: URL | string) {
async function resolvePublicUrl(value: URL | string, lookup: HostLookup) {
const url = assertPublicEpayGateway(value);
if (!isIP(url.hostname)) {
const addresses = await dns.lookup(url.hostname, { all: true, verbatim: true });
if (!addresses.length || addresses.some(({ address }) => !isPublicEpayAddress(address))) {
throw new Error("易支付网关地址不允许解析到本机或内网");
}
const hostname = normalizedHostname(url);
const addressFamily = isIP(hostname);
const addresses = addressFamily
? [{ address: hostname, family: addressFamily }]
: await lookup(hostname, { all: true, verbatim: true });
if (!addresses.length || addresses.some(({ address }) => !isPublicEpayAddress(address))) {
throw new Error("网关地址不允许解析到本机、内网或保留地址");
}
return url;
return { url, addresses };
}
export async function assertPublicGatewayUrl(
value: URL | string,
environment: ModelProviderEnvironment = process.env,
lookup: HostLookup = defaultLookup,
) {
const configured = assertConfiguredEpayUrl(value, environment);
if (configured.protocol === "http:") return configured;
return (await resolvePublicUrl(configured, lookup)).url;
}
function addAllowedOrigin(origins: Set<string>, value: string | undefined) {
if (!value) return;
try {
const url = assertPublicEpayGateway(value.trim());
if (url.protocol === "https:") origins.add(url.origin);
} catch {
// Invalid server-owned entries do not widen the allowlist.
}
}
function modelProviderOrigins(environment: ModelProviderEnvironment) {
const origins = new Set(defaultModelProviderOrigins);
environment.MODEL_PROVIDER_BASE_URL_ALLOWLIST?.split(",").forEach((value) => addAllowedOrigin(origins, value));
addAllowedOrigin(origins, environment.LLM_BASE_URL);
try {
const entries = JSON.parse(environment.LLM_MODELS_JSON ?? "[]") as unknown;
if (Array.isArray(entries)) {
for (const entry of entries) {
if (entry && typeof entry === "object" && "baseURL" in entry && typeof entry.baseURL === "string") {
addAllowedOrigin(origins, entry.baseURL);
}
}
}
} catch {
// A malformed environment catalog must not widen the allowlist.
}
return origins;
}
export async function assertAllowedModelProviderUrl(
value: URL | string,
environment: ModelProviderEnvironment = process.env,
lookup: HostLookup = defaultLookup,
) {
const resolved = await resolvePublicUrl(value, lookup);
if (resolved.url.protocol !== "https:" || !modelProviderOrigins(environment).has(resolved.url.origin)) {
throw new Error("模型供应商地址不在服务器允许列表中");
}
return resolved;
}
async function withinTimeout<T>(operation: Promise<T>, timeoutMs: number, message: string) {
let timer: ReturnType<typeof setTimeout> | undefined;
try {
return await Promise.race([
operation,
new Promise<T>((_resolve, reject) => {
timer = setTimeout(() => reject(new Error(message)), timeoutMs);
}),
]);
} finally {
if (timer) clearTimeout(timer);
}
}
async function requestPinnedHttps(
resolved: Awaited<ReturnType<typeof resolvePublicUrl>>,
method: "GET" | "HEAD",
headers: Readonly<Record<string, string>>,
timeoutMs: number,
maxResponseBytes: number,
label: string,
) {
const pinned = resolved.addresses[0]!;
const hostname = normalizedHostname(resolved.url);
return await new Promise<number>((resolve, reject) => {
const request = https.request(resolved.url, {
method,
agent: false,
headers,
servername: isIP(hostname) ? undefined : hostname,
lookup: (_hostname, _options, callback) => callback(null, pinned.address, pinned.family),
}, (response) => {
const status = response.statusCode ?? 0;
if (status >= 300 && status < 400) {
response.destroy();
reject(new Error(`${label}不允许重定向`));
return;
}
let bytes = 0;
response.on("data", (chunk: Buffer) => {
bytes += chunk.length;
if (bytes > maxResponseBytes) request.destroy(new Error(`${label}响应超过大小限制`));
});
response.on("end", () => resolve(status));
response.on("error", reject);
});
request.setTimeout(timeoutMs, () => request.destroy(new Error(`${label}超时`)));
request.on("error", reject);
request.end();
});
}
export async function probePublicEpayGateway(
value: URL | string,
options: { timeoutMs?: number; maxResponseBytes?: number; lookup?: HostLookup } = {},
) {
const timeoutMs = Math.min(Math.max(options.timeoutMs ?? 8_000, 1), 10_000);
const maxResponseBytes = Math.min(Math.max(options.maxResponseBytes ?? 64 * 1024, 1), 256 * 1024);
const startedAt = Date.now();
const resolved = await withinTimeout(
resolvePublicUrl(value, options.lookup ?? defaultLookup),
timeoutMs,
"易支付连接测试超时",
);
if (resolved.url.protocol !== "https:") throw new Error("易支付连接测试必须使用 HTTPS 公网地址");
const headStatus = await requestPinnedHttps(
resolved,
"HEAD",
{},
Math.max(1, timeoutMs - (Date.now() - startedAt)),
maxResponseBytes,
"易支付连接测试",
);
return headStatus === 405 || headStatus === 501
? requestPinnedHttps(
resolved,
"GET",
{},
Math.max(1, timeoutMs - (Date.now() - startedAt)),
maxResponseBytes,
"易支付连接测试",
)
: headStatus;
}
export async function probeAllowedModelProvider(
value: URL | string,
apiKey: string,
environment: ModelProviderEnvironment = process.env,
options: { timeoutMs?: number; maxResponseBytes?: number; lookup?: HostLookup } = {},
) {
const timeoutMs = Math.min(Math.max(options.timeoutMs ?? 8_000, 1), 10_000);
const maxResponseBytes = Math.min(Math.max(options.maxResponseBytes ?? 64 * 1024, 1), 256 * 1024);
const startedAt = Date.now();
const resolved = await withinTimeout(
assertAllowedModelProviderUrl(value, environment, options.lookup ?? defaultLookup),
timeoutMs,
"模型供应商连接测试超时",
);
return requestPinnedHttps(
resolved,
"GET",
{ accept: "application/json", authorization: `Bearer ${apiKey}` },
Math.max(1, timeoutMs - (Date.now() - startedAt)),
maxResponseBytes,
"模型供应商连接测试",
);
}
+40
View File
@@ -0,0 +1,40 @@
import "server-only";
import { queryAdminRows } from "@/lib/admin/database";
type FeatureFlagRow = {
flag_key: string;
enabled: boolean;
rollout_percentage: number;
config: Record<string, unknown>;
};
export type RuntimeFeatureFlag = {
enabled: boolean;
config: Record<string, unknown>;
};
type Cache = { expiresAt: number; flags: Map<string, RuntimeFeatureFlag> };
const state = globalThis as typeof globalThis & { jyotishaFeatureFlagCache?: Cache };
const cacheTtlMs = 15_000;
export async function loadRuntimeFeatureFlags(keys: readonly string[]) {
const cached = state.jyotishaFeatureFlagCache;
if (cached && cached.expiresAt > Date.now() && keys.every((key) => cached.flags.has(key))) return cached.flags;
const rows = await queryAdminRows<FeatureFlagRow>(`
select flag_key,enabled,rollout_percentage,config
from public.feature_flags
where status='published' and flag_key=any($1::text[])
`, [keys]);
const flags = new Map<string, RuntimeFeatureFlag>();
for (const key of keys) flags.set(key, { enabled: false, config: {} });
for (const row of rows) {
flags.set(row.flag_key, {
enabled: row.enabled && row.rollout_percentage === 100,
config: row.config,
});
}
state.jyotishaFeatureFlagCache = { expiresAt: Date.now() + cacheTtlMs, flags };
return flags;
}
@@ -0,0 +1,301 @@
begin;
create table if not exists public.billing_products (
id uuid primary key default gen_random_uuid(),
code text not null check (code ~ '^[a-z][a-z0-9_]{1,79}$'),
version integer not null check (version > 0),
name text not null check (char_length(name) between 1 and 80),
description text not null default '' check (char_length(description) <= 1000),
product_type text not null check (product_type in ('credit_pack', 'trial', 'subscription')),
billing_period text not null default 'none' check (billing_period in ('none', 'day', 'month', 'year')),
interval_count integer not null default 0 check (interval_count >= 0),
price_cents integer not null check (price_cents > 0),
currency text not null default 'CNY' check (currency = upper(currency) and char_length(currency) = 3),
enabled boolean not null default false,
status text not null default 'draft' check (status in ('draft', 'published', 'retired')),
sort_order integer not null default 0,
one_time_per_user boolean not null default false,
effective_from timestamptz,
effective_to timestamptz,
created_by uuid references auth.users(id) on delete set null,
updated_by uuid references auth.users(id) on delete set null,
created_at timestamptz not null default now(),
updated_at timestamptz not null default now(),
unique (code, version),
check ((product_type = 'credit_pack' and billing_period = 'none' and interval_count = 0)
or (product_type <> 'credit_pack' and billing_period <> 'none' and interval_count > 0)),
check (effective_to is null or effective_from is null or effective_to > effective_from)
);
create unique index if not exists billing_products_one_draft_per_code_idx
on public.billing_products(code) where status = 'draft';
create index if not exists billing_products_public_catalog_idx
on public.billing_products(enabled, status, effective_from, sort_order);
create table if not exists public.product_entitlements (
id uuid primary key default gen_random_uuid(),
product_id uuid not null references public.billing_products(id) on delete cascade,
feature_key text not null check (feature_key in ('chat.standard', 'chat.premium', 'rectification', 'report.full', 'report.export', 'profile.extra')),
allowance_type text not null check (allowance_type in ('access', 'unlimited', 'quota', 'credits')),
allowance_count integer check (allowance_count is null or allowance_count > 0),
reset_period text not null default 'none' check (reset_period in ('none', 'day', 'month', 'billing_period')),
model_tier text,
fair_use_policy_id text,
metadata jsonb not null default '{}'::jsonb check (jsonb_typeof(metadata) = 'object'),
unique (product_id, feature_key),
check ((allowance_type in ('quota', 'credits') and allowance_count is not null)
or (allowance_type in ('access', 'unlimited') and allowance_count is null))
);
create table if not exists public.user_subscriptions (
id uuid primary key default gen_random_uuid(),
user_id uuid not null references auth.users(id) on delete cascade,
product_id uuid not null references public.billing_products(id) on delete restrict,
source_order_id uuid not null unique references public.payment_orders(id) on delete restrict,
status text not null default 'pending' check (status in ('pending', 'active', 'expired', 'cancelled', 'revoked')),
starts_at timestamptz not null,
ends_at timestamptz not null,
activated_at timestamptz,
cancelled_at timestamptz,
revoked_at timestamptz,
auto_renew boolean not null default false,
product_code text not null,
product_version integer not null,
product_snapshot jsonb not null check (jsonb_typeof(product_snapshot) = 'object'),
entitlement_snapshot jsonb not null check (jsonb_typeof(entitlement_snapshot) = 'array'),
created_at timestamptz not null default now(),
updated_at timestamptz not null default now(),
check (ends_at > starts_at)
);
create index if not exists user_subscriptions_active_idx
on public.user_subscriptions(user_id, starts_at, ends_at) where status = 'active';
create table if not exists public.user_product_redemptions (
id uuid primary key default gen_random_uuid(),
user_id uuid not null references auth.users(id) on delete cascade,
product_id uuid not null references public.billing_products(id) on delete restrict,
product_code text not null,
source_order_id uuid not null unique references public.payment_orders(id) on delete restrict,
redeemed_at timestamptz not null default now(),
unique (user_id, product_code)
);
alter table public.payment_orders alter column package_id drop not null;
alter table public.payment_orders alter column credits set default 0;
alter table public.payment_orders drop constraint if exists payment_orders_credits_check;
alter table public.payment_orders add constraint payment_orders_credits_check check (credits >= 0);
alter table public.payment_orders drop constraint if exists payment_orders_status_check;
alter table public.payment_orders add constraint payment_orders_status_check
check (status in ('pending', 'paid', 'grant_pending', 'granted', 'failed', 'expired', 'refunded'));
alter table public.payment_orders
add column if not exists product_id uuid references public.billing_products(id) on delete restrict,
add column if not exists product_code text,
add column if not exists product_version integer,
add column if not exists currency text not null default 'CNY',
add column if not exists product_snapshot jsonb,
add column if not exists entitlement_snapshot jsonb,
add column if not exists grant_type text check (grant_type in ('credits', 'trial', 'subscription')),
add column if not exists grant_status text not null default 'pending' check (grant_status in ('pending', 'granted', 'failed', 'reversed')),
add column if not exists grant_reference_id uuid,
add column if not exists grant_error text,
add column if not exists granted_at timestamptz;
create index if not exists payment_orders_product_idx on public.payment_orders(product_id, created_at desc);
insert into public.billing_products (
id, code, version, name, description, product_type, billing_period, interval_count,
price_cents, enabled, status, sort_order, effective_from, created_by, updated_by
)
select p.id, 'legacy_credit_' || replace(p.id::text, '-', ''), 1, p.name, p.description,
'credit_pack', 'none', 0, p.price_cents, p.enabled, 'published', p.sort_order,
p.created_at, p.created_by, p.created_by
from public.payment_packages p
on conflict (id) do nothing;
insert into public.product_entitlements(product_id, feature_key, allowance_type, allowance_count, reset_period, metadata)
select p.id, 'chat.standard', 'credits', p.credits, 'none', jsonb_build_object('legacyPackageId', p.id)
from public.payment_packages p
on conflict (product_id, feature_key) do nothing;
update public.payment_orders o
set product_id = coalesce(o.product_id, o.package_id),
product_code = coalesce(o.product_code, bp.code),
product_version = coalesce(o.product_version, bp.version),
product_snapshot = coalesce(o.product_snapshot, jsonb_build_object(
'id', bp.id, 'code', bp.code, 'version', bp.version, 'name', bp.name,
'productType', bp.product_type, 'priceCents', bp.price_cents, 'currency', bp.currency)),
entitlement_snapshot = coalesce(o.entitlement_snapshot,
jsonb_build_array(jsonb_build_object('featureKey', 'chat.standard', 'allowanceType', 'credits', 'allowanceCount', o.credits))),
grant_type = coalesce(o.grant_type, 'credits'),
grant_status = case when o.status = 'paid' then 'granted' else coalesce(o.grant_status, 'pending') end,
granted_at = case when o.status = 'paid' then coalesce(o.granted_at, o.paid_at) else o.granted_at end
from public.billing_products bp
where o.package_id = bp.id;
insert into public.billing_products (
id, code, version, name, description, product_type, billing_period, interval_count,
price_cents, enabled, status, sort_order, one_time_per_user, effective_from
) values
('00000000-0000-4000-8000-000000000901', 'trial_7d', 1, '体验卡', '7 天内标准 AI 咨询 30 次', 'trial', 'day', 7, 990, true, 'published', 10, true, now()),
('00000000-0000-4000-8000-000000000902', 'standard_monthly', 1, '标准月卡', '会员期内标准 AI 咨询不限点数,受合理使用规则约束', 'subscription', 'month', 1, 9900, true, 'published', 20, false, now()),
('00000000-0000-4000-8000-000000000903', 'standard_yearly', 1, '标准年卡', '会员期内标准 AI 咨询不限点数,受合理使用规则约束', 'subscription', 'year', 1, 59900, true, 'published', 30, false, now()),
('00000000-0000-4000-8000-000000000904', 'pro_monthly', 1, 'Pro 月卡', '高级模型与高规格功能,完成数据验证后启用', 'subscription', 'month', 1, 29900, false, 'published', 40, false, now()),
('00000000-0000-4000-8000-000000000905', 'pro_yearly', 1, 'Pro 年卡', '高级模型与高规格功能,完成数据验证后启用', 'subscription', 'year', 1, 199900, false, 'published', 50, false, now())
on conflict (id) do nothing;
insert into public.product_entitlements(product_id, feature_key, allowance_type, allowance_count, reset_period, model_tier, fair_use_policy_id, metadata) values
('00000000-0000-4000-8000-000000000901', 'chat.standard', 'quota', 30, 'none', 'standard', 'trial_default', '{"minuteLimit":6,"dayLimit":30,"billingLimit":30}'::jsonb),
('00000000-0000-4000-8000-000000000902', 'chat.standard', 'unlimited', null, 'billing_period', 'standard', 'standard_monthly', '{"minuteLimit":6,"dayLimit":100,"billingLimit":2000}'::jsonb),
('00000000-0000-4000-8000-000000000902', 'rectification', 'quota', 1, 'billing_period', 'standard', null, '{}'::jsonb),
('00000000-0000-4000-8000-000000000902', 'report.full', 'quota', 1, 'billing_period', 'standard', null, '{}'::jsonb),
('00000000-0000-4000-8000-000000000903', 'chat.standard', 'unlimited', null, 'billing_period', 'standard', 'standard_yearly', '{"minuteLimit":6,"dayLimit":100,"billingLimit":24000}'::jsonb),
('00000000-0000-4000-8000-000000000903', 'rectification', 'quota', 12, 'billing_period', 'standard', null, '{"release":"monthly"}'::jsonb),
('00000000-0000-4000-8000-000000000903', 'report.full', 'quota', 12, 'billing_period', 'standard', null, '{"release":"monthly"}'::jsonb),
('00000000-0000-4000-8000-000000000904', 'chat.standard', 'unlimited', null, 'billing_period', 'premium', 'pro_monthly', '{"minuteLimit":10,"dayLimit":200,"billingLimit":5000}'::jsonb),
('00000000-0000-4000-8000-000000000904', 'chat.premium', 'quota', 300, 'billing_period', 'premium', null, '{}'::jsonb),
('00000000-0000-4000-8000-000000000905', 'chat.standard', 'unlimited', null, 'billing_period', 'premium', 'pro_yearly', '{"minuteLimit":10,"dayLimit":200,"billingLimit":60000}'::jsonb),
('00000000-0000-4000-8000-000000000905', 'chat.premium', 'quota', 3600, 'billing_period', 'premium', null, '{}'::jsonb)
on conflict (product_id, feature_key) do nothing;
create or replace function public.admin_save_product_draft(
p_actor_user_id uuid, p_product_id uuid, p_code text, p_name text, p_description text,
p_product_type text, p_billing_period text, p_interval_count integer, p_price_cents integer,
p_currency text, p_enabled boolean, p_sort_order integer, p_one_time_per_user boolean,
p_entitlements jsonb, p_reason text, p_request_id text
)
returns uuid
language plpgsql security definer set search_path = ''
as $$
declare v_id uuid; v_version integer; v_item jsonb;
begin
if not public.admin_has_permission(p_actor_user_id, 'billing.products.write') then
raise exception 'admin_permission_denied' using errcode='42501';
end if;
if char_length(btrim(coalesce(p_reason,''))) not between 1 and 500 then
raise exception 'admin_reason_required' using errcode='22023';
end if;
if jsonb_typeof(p_entitlements) <> 'array' then raise exception 'invalid_entitlements' using errcode='22023'; end if;
if p_product_id is null then
select coalesce(max(version),0)+1 into v_version from public.billing_products where code=p_code;
insert into public.billing_products(code,version,name,description,product_type,billing_period,interval_count,
price_cents,currency,enabled,status,sort_order,one_time_per_user,created_by,updated_by)
values (p_code,v_version,p_name,p_description,p_product_type,p_billing_period,p_interval_count,
p_price_cents,upper(p_currency),p_enabled,'draft',p_sort_order,p_one_time_per_user,p_actor_user_id,p_actor_user_id)
returning id into v_id;
else
update public.billing_products set name=p_name,description=p_description,product_type=p_product_type,
billing_period=p_billing_period,interval_count=p_interval_count,price_cents=p_price_cents,
currency=upper(p_currency),enabled=p_enabled,sort_order=p_sort_order,one_time_per_user=p_one_time_per_user,
updated_by=p_actor_user_id,updated_at=clock_timestamp()
where id=p_product_id and status='draft' returning id into v_id;
if v_id is null then raise exception 'product_draft_not_found' using errcode='22023'; end if;
end if;
delete from public.product_entitlements where product_id=v_id;
for v_item in select value from jsonb_array_elements(p_entitlements)
loop
insert into public.product_entitlements(product_id,feature_key,allowance_type,allowance_count,
reset_period,model_tier,fair_use_policy_id,metadata)
values (v_id,v_item->>'featureKey',v_item->>'allowanceType',(v_item->>'allowanceCount')::integer,
coalesce(v_item->>'resetPeriod','none'),v_item->>'modelTier',v_item->>'fairUsePolicyId',coalesce(v_item->'metadata','{}'::jsonb));
end loop;
insert into audit.admin_audit_logs(actor_user_id,actor_email,actor_role,action,target_type,target_id,
after_value,request_id,permission_used,reason)
select p_actor_user_id,lower(btrim(u.email)),'admin','billing.product.draft.save','billing_product',v_id,
jsonb_build_object('code',p_code,'name',p_name),p_request_id,'billing.products.write',btrim(p_reason)
from identity.users u where u.id=p_actor_user_id on conflict do nothing;
return v_id;
end; $$;
create or replace function public.admin_publish_product(
p_actor_user_id uuid, p_product_id uuid, p_reason text, p_request_id text
)
returns uuid
language plpgsql security definer set search_path = ''
as $$
declare v_product public.billing_products%rowtype;
begin
if not public.admin_has_permission(p_actor_user_id, 'billing.products.publish') then
raise exception 'admin_permission_denied' using errcode='42501';
end if;
if char_length(btrim(coalesce(p_reason,''))) not between 1 and 500 then raise exception 'admin_reason_required' using errcode='22023'; end if;
select * into v_product from public.billing_products where id=p_product_id and status='draft' for update;
if not found then raise exception 'product_draft_not_found' using errcode='22023'; end if;
if not exists(select 1 from public.product_entitlements where product_id=p_product_id) then raise exception 'product_entitlements_required' using errcode='23514'; end if;
update public.billing_products set enabled=false,status='retired',effective_to=coalesce(effective_to,clock_timestamp()),updated_at=clock_timestamp()
where code=v_product.code and status='published' and id<>p_product_id and effective_to is null;
update public.billing_products set status='published',effective_from=coalesce(effective_from,clock_timestamp()),updated_at=clock_timestamp()
where id=p_product_id;
insert into audit.admin_audit_logs(actor_user_id,actor_email,actor_role,action,target_type,target_id,
after_value,request_id,permission_used,reason)
select p_actor_user_id,lower(btrim(u.email)),'admin','billing.product.publish','billing_product',p_product_id,
jsonb_build_object('code',v_product.code,'version',v_product.version),p_request_id,'billing.products.publish',btrim(p_reason)
from identity.users u where u.id=p_actor_user_id on conflict do nothing;
return p_product_id;
end; $$;
create or replace function public.admin_adjust_subscription(
p_actor_user_id uuid, p_subscription_id uuid, p_action text, p_days integer,
p_expected_ends_at timestamptz, p_reason text, p_request_id text
)
returns uuid
language plpgsql security definer set search_path = ''
as $$
declare v_subscription public.user_subscriptions%rowtype;
begin
if not public.admin_has_permission(p_actor_user_id, 'billing.adjustments.write') then
raise exception 'admin_permission_denied' using errcode='42501';
end if;
if char_length(btrim(coalesce(p_reason,''))) not between 1 and 500 then raise exception 'admin_reason_required' using errcode='22023'; end if;
select * into v_subscription from public.user_subscriptions where id=p_subscription_id for update;
if not found then raise exception 'subscription_not_found' using errcode='22023'; end if;
if v_subscription.ends_at<>p_expected_ends_at then raise exception 'subscription_version_conflict' using errcode='40001'; end if;
if p_action='extend' and p_days between 1 and 3660 then
update public.user_subscriptions set ends_at=ends_at+make_interval(days=>p_days),updated_at=clock_timestamp() where id=p_subscription_id;
elsif p_action='revoke' then
update public.user_subscriptions set status='revoked',revoked_at=clock_timestamp(),updated_at=clock_timestamp() where id=p_subscription_id and status<>'revoked';
else
raise exception 'invalid_subscription_adjustment' using errcode='22023';
end if;
insert into audit.admin_audit_logs(actor_user_id,actor_email,actor_role,action,target_type,target_id,before_value,after_value,
request_id,permission_used,reason)
select p_actor_user_id,lower(btrim(u.email)),'admin','billing.subscription.'||p_action,'user_subscription',p_subscription_id,
jsonb_build_object('status',v_subscription.status,'endsAt',v_subscription.ends_at),
(select jsonb_build_object('status',s.status,'endsAt',s.ends_at) from public.user_subscriptions s where s.id=p_subscription_id),
p_request_id,'billing.adjustments.write',btrim(p_reason) from identity.users u where u.id=p_actor_user_id on conflict do nothing;
return p_subscription_id;
end; $$;
alter table public.billing_products enable row level security;
alter table public.product_entitlements enable row level security;
alter table public.user_subscriptions enable row level security;
alter table public.user_product_redemptions enable row level security;
revoke all on table public.billing_products, public.product_entitlements, public.user_subscriptions,
public.user_product_redemptions from public, anon, authenticated;
grant select on table public.billing_products, public.product_entitlements to anon, authenticated;
grant select on table public.user_subscriptions to authenticated;
grant all on table public.billing_products, public.product_entitlements, public.user_subscriptions,
public.user_product_redemptions to service_role;
drop policy if exists billing_products_public_select on public.billing_products;
create policy billing_products_public_select on public.billing_products for select to anon, authenticated
using (status='published' and enabled and effective_from <= now() and (effective_to is null or effective_to > now()));
drop policy if exists product_entitlements_public_select on public.product_entitlements;
create policy product_entitlements_public_select on public.product_entitlements for select to anon, authenticated
using (exists(select 1 from public.billing_products p where p.id=product_id and p.status='published' and p.enabled and p.effective_from <= now() and (p.effective_to is null or p.effective_to > now())));
drop policy if exists user_subscriptions_own_select on public.user_subscriptions;
create policy user_subscriptions_own_select on public.user_subscriptions for select to authenticated using ((select auth.uid())=user_id);
revoke all on function public.admin_save_product_draft(uuid,uuid,text,text,text,text,text,integer,integer,text,boolean,integer,boolean,jsonb,text,text),
public.admin_publish_product(uuid,uuid,text,text),public.admin_adjust_subscription(uuid,uuid,text,integer,timestamptz,text,text) from public,anon,authenticated;
grant execute on function public.admin_save_product_draft(uuid,uuid,text,text,text,text,text,integer,integer,text,boolean,integer,boolean,jsonb,text,text),
public.admin_publish_product(uuid,uuid,text,text),public.admin_adjust_subscription(uuid,uuid,text,integer,timestamptz,text,text) to service_role;
do $$ begin
if exists(select 1 from pg_roles where rolname='admin_runtime') then
grant select on table public.billing_products, public.product_entitlements, public.user_subscriptions,
public.user_product_redemptions to admin_runtime;
grant execute on function public.admin_save_product_draft(uuid,uuid,text,text,text,text,text,integer,integer,text,boolean,integer,boolean,jsonb,text,text),
public.admin_publish_product(uuid,uuid,text,text),public.admin_adjust_subscription(uuid,uuid,text,integer,timestamptz,text,text) to admin_runtime;
end if;
end $$;
commit;
@@ -0,0 +1,851 @@
begin;
alter table public.credit_transactions
drop constraint if exists credit_transactions_check,
drop constraint if exists credit_transactions_transaction_type_check,
drop constraint if exists credit_transactions_amount_check;
alter table public.credit_transactions
add constraint credit_transactions_transaction_type_check check (
transaction_type in ('redeem', 'reserve', 'refund', 'payment', 'compensation')
),
add constraint credit_transactions_amount_check check (
(transaction_type = 'reserve' and amount < 0)
or (transaction_type in ('redeem', 'refund', 'payment', 'compensation') and amount > 0)
);
alter table public.payment_orders
add column if not exists adjustment_version integer not null default 0,
add column if not exists refund_status text not null default 'none',
add column if not exists refund_amount_cents integer,
add column if not exists refunded_at timestamptz;
alter table public.payment_orders
drop constraint if exists payment_orders_adjustment_version_check,
drop constraint if exists payment_orders_refund_status_check,
drop constraint if exists payment_orders_refund_amount_check;
alter table public.payment_orders
add constraint payment_orders_adjustment_version_check check (adjustment_version >= 0),
add constraint payment_orders_refund_status_check check (refund_status in ('none', 'recorded')),
add constraint payment_orders_refund_amount_check check (
(refund_status = 'none' and refund_amount_cents is null and refunded_at is null)
or (refund_status = 'recorded' and refund_amount_cents = money_cents and refunded_at is not null)
);
update public.payment_orders o
set product_snapshot = jsonb_set(
case
when jsonb_typeof(o.product_snapshot) = 'object' then o.product_snapshot
else jsonb_build_object(
'id', p.id,
'code', p.code,
'version', p.version,
'name', p.name,
'description', p.description,
'productType', p.product_type,
'billingPeriod', p.billing_period,
'intervalCount', p.interval_count,
'priceCents', p.price_cents,
'currency', p.currency
)
end,
'{oneTimePerUser}',
to_jsonb(p.one_time_per_user),
true
)
from public.billing_products p
where o.product_id = p.id;
alter table public.payment_orders
drop constraint if exists payment_orders_product_snapshot_one_time_check;
alter table public.payment_orders
add constraint payment_orders_product_snapshot_one_time_check check (
product_id is null
or (
jsonb_typeof(product_snapshot) = 'object'
and jsonb_typeof(product_snapshot -> 'oneTimePerUser') = 'boolean'
)
);
create or replace function public.protect_payment_order_snapshot()
returns trigger
language plpgsql
set search_path = ''
as $$
begin
if new.user_id is distinct from old.user_id
or new.package_id is distinct from old.package_id
or new.product_id is distinct from old.product_id
or new.product_code is distinct from old.product_code
or new.product_version is distinct from old.product_version
or new.product_snapshot is distinct from old.product_snapshot
or new.entitlement_snapshot is distinct from old.entitlement_snapshot
or new.money_cents is distinct from old.money_cents
or new.currency is distinct from old.currency
or new.credits is distinct from old.credits then
raise exception 'payment_order_snapshot_immutable' using errcode = '55000';
end if;
return new;
end;
$$;
revoke all on function public.protect_payment_order_snapshot() from public, anon, authenticated, service_role;
drop trigger if exists payment_order_snapshot_immutable on public.payment_orders;
create trigger payment_order_snapshot_immutable
before update of user_id, package_id, product_id, product_code, product_version,
product_snapshot, entitlement_snapshot, money_cents, currency, credits
on public.payment_orders
for each row execute function public.protect_payment_order_snapshot();
create table if not exists public.usage_reservations (
id uuid primary key default gen_random_uuid(),
user_id uuid not null references auth.users(id) on delete cascade,
request_id text not null check (char_length(request_id) between 1 and 200),
request_fingerprint text check (request_fingerprint is null or request_fingerprint ~ '^[0-9a-f]{64}$'),
feature_key text not null check (feature_key in ('chat.standard', 'chat.premium', 'rectification', 'report.full', 'report.export', 'profile.extra')),
requested_model_id text,
resolved_model_id text,
model_config_version integer,
source text not null check (source in ('subscription', 'credits')),
subscription_id uuid references public.user_subscriptions(id) on delete restrict,
credit_amount integer not null default 0 check (credit_amount >= 0),
status text not null default 'reserved' check (status in ('reserved', 'completed', 'released')),
release_reason text,
reserved_at timestamptz not null default clock_timestamp(),
completed_at timestamptz,
released_at timestamptz,
unique (user_id, request_id)
);
create index if not exists usage_reservations_subscription_feature_idx
on public.usage_reservations(subscription_id, feature_key, reserved_at);
create index if not exists usage_reservations_user_created_idx
on public.usage_reservations(user_id, reserved_at desc);
create table if not exists public.usage_ledger (
id uuid primary key default gen_random_uuid(),
reservation_id uuid not null unique references public.usage_reservations(id) on delete restrict,
user_id uuid not null references auth.users(id) on delete cascade,
subscription_id uuid references public.user_subscriptions(id) on delete restrict,
request_id text not null,
feature_key text not null,
source text not null check (source in ('subscription', 'credits')),
requested_model_id text,
actual_model_id text,
model_config_version integer,
input_tokens integer not null default 0 check (input_tokens >= 0),
output_tokens integer not null default 0 check (output_tokens >= 0),
cost_microusd bigint not null default 0 check (cost_microusd >= 0),
duration_ms integer check (duration_ms is null or duration_ms >= 0),
metadata jsonb not null default '{}'::jsonb check (jsonb_typeof(metadata) = 'object'),
created_at timestamptz not null default clock_timestamp(),
unique (user_id, request_id)
);
create index if not exists usage_ledger_created_idx on public.usage_ledger(created_at desc);
create index if not exists usage_ledger_feature_created_idx on public.usage_ledger(feature_key, created_at desc);
create table if not exists public.usage_events (
id uuid primary key default gen_random_uuid(),
reservation_id uuid not null references public.usage_reservations(id) on delete restrict,
event_key text not null check (char_length(event_key) between 1 and 200),
payload_fingerprint text check (payload_fingerprint is null or payload_fingerprint ~ '^[0-9a-f]{64}$'),
actual_model_id text,
model_config_version integer,
input_tokens integer not null default 0 check (input_tokens >= 0),
output_tokens integer not null default 0 check (output_tokens >= 0),
cost_microusd bigint not null default 0 check (cost_microusd >= 0),
duration_ms integer check (duration_ms is null or duration_ms >= 0),
metadata jsonb not null default '{}'::jsonb check (jsonb_typeof(metadata) = 'object'),
created_at timestamptz not null default clock_timestamp(),
unique (reservation_id, event_key)
);
create or replace function public.authorize_usage(
p_user_id uuid, p_feature_key text, p_requested_model_id text, p_request_id text, p_credit_cost integer default 1
)
returns table (
success boolean, reservation_id uuid, source text, credits integer,
subscription_id uuid, reason text, retry_after_seconds integer
)
language plpgsql security definer set search_path = ''
as $$
declare
v_request_id text := btrim(p_request_id);
v_request_fingerprint text := pg_catalog.encode(pg_catalog.sha256(pg_catalog.convert_to(
pg_catalog.jsonb_build_object(
'featureKey', p_feature_key,
'requestedModelId', p_requested_model_id,
'creditCost', p_credit_cost
)::text,
'UTF8'
)), 'hex');
v_existing public.usage_reservations%rowtype;
v_subscription_id uuid;
v_redemption_id uuid;
v_subscription_starts_at timestamptz;
v_subscription_ends_at timestamptz;
v_subscription_billing_period text;
v_entitlement jsonb;
v_balance integer;
v_minute_limit integer;
v_day_limit integer;
v_billing_limit integer;
v_used integer;
v_reservation_id uuid;
v_allowance_type text;
v_allowance_count integer;
v_released_months integer;
v_allowed_model_tier text;
v_requested_model_tier text;
v_subscription_failure_id uuid;
v_subscription_failure_reason text;
v_subscription_retry_after integer;
v_subscriptions_enabled boolean := true;
begin
if p_user_id is null then return query select false,null::uuid,null::text,null::integer,null::uuid,'unauthorized'::text,null::integer; return; end if;
if v_request_id is null or char_length(v_request_id) not between 1 and 200 or p_credit_cost < 0 then
return query select false,null::uuid,null::text,null::integer,null::uuid,'invalid_request'::text,null::integer; return;
end if;
perform pg_advisory_xact_lock(hashtextextended(p_user_id::text || ':' || v_request_id,0));
select r.* into v_existing from public.usage_reservations r where r.user_id=p_user_id and r.request_id=v_request_id;
if found then
select p.credits into v_balance from public.profiles p where p.id=p_user_id;
if v_existing.request_fingerprint is distinct from v_request_fingerprint then
return query select false,v_existing.id,v_existing.source,v_balance,
v_existing.subscription_id,'request_conflict'::text,null::integer;
return;
end if;
return query select v_existing.status in ('reserved','completed'),v_existing.id,v_existing.source,v_balance,
v_existing.subscription_id,case when v_existing.status='released' then 'request_released' else null end,null::integer;
return;
end if;
if pg_catalog.to_regclass('public.feature_flags') is not null then
execute $flag$
select coalesce((
select enabled and rollout_percentage=100
from public.feature_flags
where flag_key='billing.subscriptions' and status='published'
),false)
$flag$ into v_subscriptions_enabled;
end if;
if v_subscriptions_enabled then
select s.id, s.starts_at, s.ends_at, s.product_snapshot->>'billingPeriod', e.value
into v_subscription_id, v_subscription_starts_at, v_subscription_ends_at,
v_subscription_billing_period, v_entitlement
from public.user_subscriptions s
cross join lateral jsonb_array_elements(s.entitlement_snapshot) e(value)
where s.user_id=p_user_id and s.status='active' and s.starts_at<=clock_timestamp() and s.ends_at>clock_timestamp()
and e.value->>'featureKey'=p_feature_key
order by s.ends_at asc limit 1 for update of s;
end if;
if v_subscription_id is not null then
v_subscription_failure_id := v_subscription_id;
v_allowance_type := v_entitlement->>'allowanceType';
v_allowance_count := nullif(v_entitlement->>'allowanceCount','')::integer;
if v_allowance_type='quota' and v_allowance_count is not null
and v_entitlement#>>'{metadata,release}'='monthly'
and v_subscription_billing_period='year' then
v_released_months := greatest(
1,
extract(year from age(clock_timestamp(),v_subscription_starts_at))::integer*12
+ extract(month from age(clock_timestamp(),v_subscription_starts_at))::integer
+ 1
);
v_allowance_count := least(v_allowance_count,v_released_months);
end if;
v_allowed_model_tier := v_entitlement->>'modelTier';
if p_requested_model_id is not null and v_allowed_model_tier is not null then
select v.model_tier into v_requested_model_tier
from public.model_configs c join public.model_config_versions v on v.config_id=c.id
where c.model_id=p_requested_model_id and v.status='published' and v.enabled limit 1;
if v_requested_model_tier is null or (v_allowed_model_tier='standard' and v_requested_model_tier='premium') then
v_subscription_failure_reason := 'model_not_included';
v_subscription_id := null;
end if;
end if;
v_minute_limit := nullif(v_entitlement#>>'{metadata,minuteLimit}','')::integer;
v_day_limit := nullif(v_entitlement#>>'{metadata,dayLimit}','')::integer;
v_billing_limit := nullif(v_entitlement#>>'{metadata,billingLimit}','')::integer;
if v_subscription_id is not null and v_minute_limit is not null then
select count(*) into v_used from public.usage_reservations r where r.subscription_id=v_subscription_id
and r.feature_key=p_feature_key and r.status in ('reserved','completed') and r.reserved_at>clock_timestamp()-interval '1 minute';
if v_used>=v_minute_limit then
v_subscription_failure_reason := 'fair_use_minute';
v_subscription_retry_after := 60;
v_subscription_id := null;
end if;
end if;
if v_subscription_id is not null and v_day_limit is not null then
select count(*) into v_used from public.usage_reservations r where r.subscription_id=v_subscription_id
and r.feature_key=p_feature_key and r.status in ('reserved','completed') and r.reserved_at>=date_trunc('day',clock_timestamp());
if v_used>=v_day_limit then
v_subscription_failure_reason := 'fair_use_day';
v_subscription_retry_after := greatest(1,extract(epoch from (date_trunc('day',clock_timestamp())+interval '1 day'-clock_timestamp()))::integer);
v_subscription_id := null;
end if;
end if;
if v_subscription_id is not null and v_billing_limit is not null then
select count(*) into v_used from public.usage_reservations r where r.subscription_id=v_subscription_id
and r.feature_key=p_feature_key and r.status in ('reserved','completed');
if v_used>=v_billing_limit then
v_subscription_failure_reason := 'fair_use_billing_period';
v_subscription_retry_after := greatest(1,extract(epoch from (v_subscription_ends_at-clock_timestamp()))::integer);
v_subscription_id := null;
end if;
end if;
if v_subscription_id is not null and v_allowance_type='quota' and v_allowance_count is not null then
select count(*) into v_used from public.usage_reservations r where r.subscription_id=v_subscription_id
and r.feature_key=p_feature_key and r.status in ('reserved','completed');
if v_used>=v_allowance_count then
v_subscription_failure_reason := 'feature_quota_exhausted';
v_subscription_id := null;
end if;
end if;
end if;
if v_subscription_id is not null then
insert into public.usage_reservations(
user_id,request_id,request_fingerprint,feature_key,requested_model_id,source,subscription_id
)
values(p_user_id,v_request_id,v_request_fingerprint,p_feature_key,p_requested_model_id,'subscription',v_subscription_id)
returning id into v_reservation_id;
select p.credits into v_balance from public.profiles p where p.id=p_user_id;
return query select true,v_reservation_id,'subscription'::text,v_balance,v_subscription_id,null::text,null::integer; return;
end if;
update public.profiles p set credits=p.credits-p_credit_cost,updated_at=clock_timestamp()
where p.id=p_user_id and p.credits>=p_credit_cost returning p.credits into v_balance;
if not found then
select p.credits into v_balance from public.profiles p where p.id=p_user_id;
return query select false,null::uuid,'credits'::text,v_balance,v_subscription_failure_id,
case when v_balance is null then 'profile_missing' else coalesce(v_subscription_failure_reason,'insufficient_credits') end,
case when v_balance is null then null::integer else v_subscription_retry_after end; return;
end if;
insert into public.usage_reservations(
user_id,request_id,request_fingerprint,feature_key,requested_model_id,source,credit_amount
)
values(p_user_id,v_request_id,v_request_fingerprint,p_feature_key,p_requested_model_id,'credits',p_credit_cost)
returning id into v_reservation_id;
if p_credit_cost>0 then
insert into public.credit_transactions(user_id,transaction_type,amount,balance_after,request_id,model)
values(p_user_id,'reserve',-p_credit_cost,v_balance,v_request_id,p_requested_model_id);
end if;
return query select true,v_reservation_id,'credits'::text,v_balance,null::uuid,null::text,null::integer;
end; $$;
create or replace function public.complete_usage(
p_user_id uuid, p_request_id text, p_actual_usage jsonb default '{}'::jsonb
)
returns table(success boolean, reservation_id uuid, credits integer, error_code text)
language plpgsql security definer set search_path = ''
as $$
declare
v_res public.usage_reservations%rowtype;
v_balance integer;
v_event_key text := coalesce(nullif(btrim(p_actual_usage->>'eventKey'),''),'request');
v_payload_fingerprint text := pg_catalog.encode(pg_catalog.sha256(pg_catalog.convert_to(
coalesce(p_actual_usage,'{}'::jsonb)::text,
'UTF8'
)), 'hex');
v_existing_payload_fingerprint text;
v_event_inserted boolean;
begin
perform pg_advisory_xact_lock(hashtextextended(p_user_id::text || ':' || btrim(p_request_id),0));
select * into v_res from public.usage_reservations where user_id=p_user_id and request_id=btrim(p_request_id) for update;
select p.credits into v_balance from public.profiles p where p.id=p_user_id;
if not found then return query select false,null::uuid,null::integer,'profile_missing'::text; return; end if;
if v_res.id is null then return query select false,null::uuid,v_balance,'request_missing'::text; return; end if;
if v_res.status='released' then return query select false,v_res.id,v_balance,'request_released'::text; return; end if;
insert into public.usage_events(reservation_id,event_key,payload_fingerprint,actual_model_id,model_config_version,
input_tokens,output_tokens,cost_microusd,duration_ms,metadata)
values(v_res.id,v_event_key,v_payload_fingerprint,p_actual_usage->>'actualModelId',nullif(p_actual_usage->>'modelConfigVersion','')::integer,
coalesce(nullif(p_actual_usage->>'inputTokens','')::integer,0),coalesce(nullif(p_actual_usage->>'outputTokens','')::integer,0),
coalesce(nullif(p_actual_usage->>'costMicrousd','')::bigint,0),nullif(p_actual_usage->>'durationMs','')::integer,
coalesce(p_actual_usage->'metadata','{}'::jsonb))
on conflict on constraint usage_events_reservation_id_event_key_key do nothing returning true into v_event_inserted;
if not coalesce(v_event_inserted,false) then
select e.payload_fingerprint into v_existing_payload_fingerprint
from public.usage_events e
where e.reservation_id=v_res.id and e.event_key=v_event_key;
if v_existing_payload_fingerprint is distinct from v_payload_fingerprint then
return query select false,v_res.id,v_balance,'event_payload_conflict'::text; return;
end if;
return query select true,v_res.id,v_balance,null::text; return;
end if;
update public.usage_reservations set status='completed',completed_at=coalesce(completed_at,clock_timestamp()),
resolved_model_id=coalesce(p_actual_usage->>'actualModelId',requested_model_id),
model_config_version=coalesce(nullif(p_actual_usage->>'modelConfigVersion','')::integer,model_config_version)
where id=v_res.id;
insert into public.usage_ledger as ledger(reservation_id,user_id,subscription_id,request_id,feature_key,source,
requested_model_id,actual_model_id,model_config_version,input_tokens,output_tokens,cost_microusd,duration_ms,metadata)
values(v_res.id,v_res.user_id,v_res.subscription_id,v_res.request_id,v_res.feature_key,v_res.source,
v_res.requested_model_id,coalesce(p_actual_usage->>'actualModelId',v_res.requested_model_id),
nullif(p_actual_usage->>'modelConfigVersion','')::integer,coalesce(nullif(p_actual_usage->>'inputTokens','')::integer,0),
coalesce(nullif(p_actual_usage->>'outputTokens','')::integer,0),coalesce(nullif(p_actual_usage->>'costMicrousd','')::bigint,0),
nullif(p_actual_usage->>'durationMs','')::integer,coalesce(p_actual_usage->'metadata','{}'::jsonb))
on conflict on constraint usage_ledger_reservation_id_key do update set
actual_model_id=coalesce(excluded.actual_model_id,ledger.actual_model_id),
model_config_version=coalesce(excluded.model_config_version,ledger.model_config_version),
input_tokens=ledger.input_tokens+excluded.input_tokens,
output_tokens=ledger.output_tokens+excluded.output_tokens,
cost_microusd=ledger.cost_microusd+excluded.cost_microusd,
duration_ms=coalesce(ledger.duration_ms,0)+coalesce(excluded.duration_ms,0),
metadata=ledger.metadata||excluded.metadata;
return query select true,v_res.id,v_balance,null::text;
end; $$;
create or replace function public.release_usage(
p_user_id uuid, p_request_id text, p_reason text
)
returns table(success boolean, reservation_id uuid, credits integer, error_code text)
language plpgsql security definer set search_path = ''
as $$
declare v_res public.usage_reservations%rowtype; v_balance integer;
begin
perform pg_advisory_xact_lock(hashtextextended(p_user_id::text || ':' || btrim(p_request_id),0));
select * into v_res from public.usage_reservations where user_id=p_user_id and request_id=btrim(p_request_id) for update;
select p.credits into v_balance from public.profiles p where p.id=p_user_id for update;
if not found then return query select false,null::uuid,null::integer,'profile_missing'::text; return; end if;
if v_res.id is null then return query select true,null::uuid,v_balance,null::text; return; end if;
if v_res.status='released' then return query select true,v_res.id,v_balance,null::text; return; end if;
if v_res.status='completed' then return query select false,v_res.id,v_balance,'request_completed'::text; return; end if;
if v_res.source='credits' and v_res.credit_amount>0 then
update public.profiles p set credits=p.credits+v_res.credit_amount,updated_at=clock_timestamp() where p.id=p_user_id returning p.credits into v_balance;
insert into public.credit_transactions(user_id,transaction_type,amount,balance_after,request_id,model)
values(p_user_id,'refund',v_res.credit_amount,v_balance,v_res.request_id,v_res.requested_model_id)
on conflict(user_id,transaction_type,request_id) do nothing;
end if;
update public.usage_reservations set status='released',released_at=clock_timestamp(),release_reason=left(coalesce(p_reason,'released'),500) where id=v_res.id;
return query select true,v_res.id,v_balance,null::text;
end; $$;
create or replace function public.settle_order(p_order_no text,p_trade_no text,p_money_cents integer,p_payload_hash text)
returns table(success boolean,status text,credits integer)
language plpgsql security definer set search_path = ''
as $$
declare
v_order public.payment_orders%rowtype;
v_balance integer;
v_entitlements jsonb;
v_product_id uuid;
v_product_code text;
v_product_version integer;
v_product_type text;
v_billing_period text;
v_interval_count integer;
v_snapshot_price_cents integer;
v_snapshot_currency text;
v_one_time_per_user boolean;
v_credit_grant integer;
v_start timestamptz;
v_end timestamptz;
v_now timestamptz;
v_subscription_id uuid;
v_redemption_id uuid;
v_overlap boolean;
v_grant_error text;
v_error_state text;
v_error_message text;
begin
if btrim(coalesce(p_trade_no,''))='' or btrim(coalesce(p_payload_hash,''))='' then
return query select false,'invalid'::text,null::integer; return;
end if;
select * into v_order from public.payment_orders where order_no=btrim(p_order_no) for update;
if not found or v_order.money_cents<>p_money_cents or v_order.currency<>'CNY' then
return query select false,'invalid'::text,null::integer; return;
end if;
perform pg_advisory_xact_lock(hashtextextended('payment-trade:'||p_trade_no,0));
if v_order.epay_trade_no is distinct from null and v_order.epay_trade_no<>p_trade_no then
return query select false,'transaction_mismatch'::text,null::integer; return;
end if;
if exists(select 1 from public.payment_orders o where o.epay_trade_no=p_trade_no and o.id<>v_order.id) then
return query select false,'transaction_conflict'::text,null::integer; return;
end if;
if v_order.grant_status='granted' then
return query select true,'paid'::text,v_order.credits; return;
end if;
update public.payment_orders set
status='grant_pending',
epay_trade_no=coalesce(epay_trade_no,p_trade_no),
raw_notify_payload_hash=coalesce(raw_notify_payload_hash,p_payload_hash),
paid_at=coalesce(paid_at,clock_timestamp())
where id=v_order.id;
perform pg_advisory_xact_lock(hashtextextended(v_order.user_id::text||':settle_order',0));
v_now := clock_timestamp();
<<grant_attempt>>
begin
if jsonb_typeof(v_order.product_snapshot)<>'object' or jsonb_typeof(v_order.entitlement_snapshot)<>'array' then
v_grant_error := 'invalid_snapshot'; exit grant_attempt;
end if;
v_entitlements := v_order.entitlement_snapshot;
v_product_id := nullif(v_order.product_snapshot->>'id','')::uuid;
v_product_code := nullif(v_order.product_snapshot->>'code','');
v_product_version := nullif(v_order.product_snapshot->>'version','')::integer;
v_product_type := nullif(v_order.product_snapshot->>'productType','');
v_billing_period := nullif(v_order.product_snapshot->>'billingPeriod','');
v_interval_count := nullif(v_order.product_snapshot->>'intervalCount','')::integer;
v_snapshot_price_cents := nullif(v_order.product_snapshot->>'priceCents','')::integer;
v_snapshot_currency := nullif(v_order.product_snapshot->>'currency','');
if jsonb_typeof(v_order.product_snapshot->'oneTimePerUser') is distinct from 'boolean' then
v_grant_error := 'invalid_snapshot'; exit grant_attempt;
end if;
v_one_time_per_user := (v_order.product_snapshot->>'oneTimePerUser')::boolean;
if v_product_id is null or v_product_id is distinct from v_order.product_id
or v_product_code is null or v_product_code is distinct from v_order.product_code
or v_product_version is null or v_product_version is distinct from v_order.product_version
or v_product_type not in ('credit_pack','trial','subscription')
or v_snapshot_price_cents is distinct from v_order.money_cents or v_snapshot_price_cents<>p_money_cents
or v_snapshot_currency is distinct from v_order.currency then
v_grant_error := 'invalid_snapshot'; exit grant_attempt;
end if;
if v_product_type='credit_pack' then
select coalesce(sum(nullif(e.value->>'allowanceCount','')::integer),0) into v_credit_grant
from jsonb_array_elements(v_entitlements) e(value) where e.value->>'allowanceType'='credits';
if v_credit_grant<=0 or v_credit_grant is distinct from v_order.credits then
v_grant_error := 'invalid_snapshot'; exit grant_attempt;
end if;
select p.credits into v_balance from public.profiles p where p.id=v_order.user_id for update;
if not found then v_grant_error := 'profile_missing'; exit grant_attempt; end if;
if v_one_time_per_user then
insert into public.user_product_redemptions(user_id,product_id,product_code,source_order_id)
values(v_order.user_id,v_product_id,v_product_code,v_order.id)
on conflict (user_id,product_code) do nothing
returning id into v_redemption_id;
if v_redemption_id is null then v_grant_error := 'one_time_limit'; exit grant_attempt; end if;
end if;
update public.profiles p set credits=p.credits+v_credit_grant,updated_at=clock_timestamp()
where p.id=v_order.user_id returning p.credits into v_balance;
insert into public.credit_transactions(user_id,transaction_type,amount,balance_after,request_id)
values(v_order.user_id,'payment',v_credit_grant,v_balance,v_order.order_no);
update public.payment_orders set status='paid',grant_status='granted',grant_type='credits',
granted_at=clock_timestamp(),grant_error=null where id=v_order.id;
return query select true,'paid'::text,v_credit_grant; return;
end if;
if v_billing_period not in ('day','month','year') or coalesce(v_interval_count,0)<=0 then
v_grant_error := 'invalid_period'; exit grant_attempt;
end if;
select exists(
select 1 from public.user_subscriptions s where s.user_id=v_order.user_id and s.status='active'
and s.starts_at<=v_now and s.ends_at>v_now and s.product_code<>v_product_code
) into v_overlap;
if v_overlap then v_grant_error := 'overlapping_subscription'; exit grant_attempt; end if;
select greatest(v_now,coalesce(max(s.ends_at),v_now)) into v_start
from public.user_subscriptions s where s.user_id=v_order.user_id and s.product_code=v_product_code and s.status='active';
v_end := case v_billing_period
when 'day' then v_start+make_interval(days=>v_interval_count)
when 'month' then v_start+make_interval(months=>v_interval_count)
when 'year' then v_start+make_interval(years=>v_interval_count)
else null end;
if v_one_time_per_user then
insert into public.user_product_redemptions(user_id,product_id,product_code,source_order_id)
values(v_order.user_id,v_product_id,v_product_code,v_order.id)
on conflict (user_id,product_code) do nothing
returning id into v_redemption_id;
if v_redemption_id is null then v_grant_error := 'one_time_limit'; exit grant_attempt; end if;
end if;
insert into public.user_subscriptions(user_id,product_id,source_order_id,status,starts_at,ends_at,activated_at,
product_code,product_version,product_snapshot,entitlement_snapshot)
values(v_order.user_id,v_product_id,v_order.id,'active',v_start,v_end,v_now,v_product_code,v_product_version,
v_order.product_snapshot,v_entitlements) returning id into v_subscription_id;
update public.payment_orders set status='paid',grant_status='granted',grant_type=v_product_type,
grant_reference_id=v_subscription_id,granted_at=clock_timestamp(),grant_error=null where id=v_order.id;
select p.credits into v_balance from public.profiles p where p.id=v_order.user_id;
return query select true,'paid'::text,v_balance; return;
exception when others then
get stacked diagnostics v_error_state=returned_sqlstate,v_error_message=message_text;
v_grant_error := left(v_error_state||':'||v_error_message,500);
end grant_attempt;
update public.payment_orders set status='grant_pending',grant_status='failed',
grant_type=case when v_product_type='credit_pack' then 'credits' else v_product_type end,
grant_reference_id=null,grant_error=left(coalesce(v_grant_error,'grant_failed'),500)
where id=v_order.id;
return query select false,
case when v_grant_error in ('invalid_snapshot','invalid_period','profile_missing','one_time_limit','overlapping_subscription')
then v_grant_error else 'grant_failed' end,
v_balance;
end; $$;
create or replace function public.settle_epay_order(p_order_no text,p_trade_no text,p_money_cents integer,p_payload_hash text)
returns table(success boolean,status text,credits integer)
language sql security definer set search_path = ''
as $$ select * from public.settle_order(p_order_no,p_trade_no,p_money_cents,p_payload_hash); $$;
create or replace function public.require_admin_redemption_reason()
returns trigger
language plpgsql
set search_path = ''
as $$
declare
v_reason text := nullif(btrim(current_setting('app.admin_redemption_reason', true)), '');
begin
if new.action in ('redemption_code.create','redemption_code.update','redemption_code.revoke') then
if v_reason is null or char_length(v_reason) > 500 then
raise exception 'admin_reason_required' using errcode='22023';
end if;
new.permission_used := 'billing.adjustments.write';
new.reason := v_reason;
end if;
return new;
end;
$$;
revoke all on function public.require_admin_redemption_reason() from public,anon,authenticated,service_role;
drop trigger if exists admin_redemption_reason_audit on audit.admin_audit_logs;
create trigger admin_redemption_reason_audit
before insert on audit.admin_audit_logs
for each row execute function public.require_admin_redemption_reason();
create or replace function public.admin_create_redemption_codes(
p_actor_user_id uuid,p_actor_email text,p_actor_role text,p_request_id text,p_codes jsonb,p_reason text
)
returns table(id uuid,code_mask text,credits integer,expires_at timestamptz,note text,created_at timestamptz,
redeemed_by uuid,redeemed_email text,redeemed_at timestamptz,revoked_by uuid,revoked_at timestamptz)
language plpgsql security definer set search_path = ''
as $$
begin
if not public.admin_has_permission(p_actor_user_id,'billing.adjustments.write') then
raise exception 'admin_permission_denied' using errcode='42501';
end if;
if char_length(btrim(coalesce(p_reason,''))) not between 1 and 500 then
raise exception 'admin_reason_required' using errcode='22023';
end if;
perform set_config('app.admin_redemption_reason',btrim(p_reason),true);
return query select * from public.admin_create_redemption_codes(
p_actor_user_id,p_actor_email,p_actor_role,p_request_id,p_codes
);
end;
$$;
create or replace function public.admin_update_redemption_code(
p_actor_user_id uuid,p_actor_email text,p_actor_role text,p_request_id text,p_code_id uuid,
p_set_note boolean,p_note text,p_set_expires_at boolean,p_expires_at timestamptz,p_reason text
)
returns table(id uuid,code_mask text,credits integer,expires_at timestamptz,note text,created_at timestamptz,
redeemed_by uuid,redeemed_email text,redeemed_at timestamptz,revoked_by uuid,revoked_at timestamptz)
language plpgsql security definer set search_path = ''
as $$
begin
if not public.admin_has_permission(p_actor_user_id,'billing.adjustments.write') then
raise exception 'admin_permission_denied' using errcode='42501';
end if;
if char_length(btrim(coalesce(p_reason,''))) not between 1 and 500 then
raise exception 'admin_reason_required' using errcode='22023';
end if;
perform set_config('app.admin_redemption_reason',btrim(p_reason),true);
return query select * from public.admin_update_redemption_code(
p_actor_user_id,p_actor_email,p_actor_role,p_request_id,p_code_id,
p_set_note,p_note,p_set_expires_at,p_expires_at
);
end;
$$;
create or replace function public.admin_revoke_redemption_code(
p_actor_user_id uuid,p_actor_email text,p_actor_role text,p_request_id text,p_code_id uuid,p_reason text
)
returns table(id uuid,code_mask text,credits integer,expires_at timestamptz,note text,created_at timestamptz,
redeemed_by uuid,redeemed_email text,redeemed_at timestamptz,revoked_by uuid,revoked_at timestamptz)
language plpgsql security definer set search_path = ''
as $$
begin
if not public.admin_has_permission(p_actor_user_id,'billing.adjustments.write') then
raise exception 'admin_permission_denied' using errcode='42501';
end if;
if char_length(btrim(coalesce(p_reason,''))) not between 1 and 500 then
raise exception 'admin_reason_required' using errcode='22023';
end if;
perform set_config('app.admin_redemption_reason',btrim(p_reason),true);
return query select * from public.admin_revoke_redemption_code(
p_actor_user_id,p_actor_email,p_actor_role,p_request_id,p_code_id
);
end;
$$;
revoke all on function public.admin_create_redemption_codes(uuid,text,text,text,jsonb) from service_role;
revoke all on function public.admin_update_redemption_code(uuid,text,text,text,uuid,boolean,text,boolean,timestamptz) from service_role;
revoke all on function public.admin_revoke_redemption_code(uuid,text,text,text,uuid) from service_role;
revoke all on function public.admin_create_redemption_codes(uuid,text,text,text,jsonb,text),
public.admin_update_redemption_code(uuid,text,text,text,uuid,boolean,text,boolean,timestamptz,text),
public.admin_revoke_redemption_code(uuid,text,text,text,uuid,text) from public,anon,authenticated,service_role;
do $$ begin if exists(select 1 from pg_roles where rolname='admin_runtime') then
revoke execute on function public.admin_create_redemption_codes(uuid,text,text,text,jsonb) from admin_runtime;
revoke execute on function public.admin_update_redemption_code(uuid,text,text,text,uuid,boolean,text,boolean,timestamptz) from admin_runtime;
revoke execute on function public.admin_revoke_redemption_code(uuid,text,text,text,uuid) from admin_runtime;
grant execute on function public.admin_create_redemption_codes(uuid,text,text,text,jsonb,text),
public.admin_update_redemption_code(uuid,text,text,text,uuid,boolean,text,boolean,timestamptz,text),
public.admin_revoke_redemption_code(uuid,text,text,text,uuid,text) to admin_runtime;
end if; end $$;
create or replace function public.admin_adjust_order(
p_actor_user_id uuid,p_order_id uuid,p_action text,p_expected_version integer,p_reason text,p_request_id text
)
returns table(
id uuid,order_no text,status text,grant_status text,grant_error text,adjustment_version integer,
refund_status text,refund_amount_cents integer,refunded_at timestamptz,action_success boolean
)
language plpgsql security definer set search_path = ''
as $$
declare
v_order public.payment_orders%rowtype;
v_actor_email text;
v_settle record;
v_balance integer;
v_one_time boolean;
v_redemption_id uuid;
v_action_success boolean := true;
begin
if not public.admin_has_permission(p_actor_user_id,'billing.adjustments.write') then
raise exception 'admin_permission_denied' using errcode='42501';
end if;
if p_action not in ('retry_grant','compensate','record_refund') then
raise exception 'admin_order_action_invalid' using errcode='22023';
end if;
if char_length(btrim(coalesce(p_reason,''))) not between 1 and 500 then
raise exception 'admin_reason_required' using errcode='22023';
end if;
if char_length(btrim(coalesce(p_request_id,''))) not between 1 and 200 then
raise exception 'admin_request_id_invalid' using errcode='22023';
end if;
select lower(btrim(u.email)) into v_actor_email from identity.users u where u.id=p_actor_user_id;
if v_actor_email is null then raise exception 'admin_user_not_found' using errcode='22023'; end if;
select * into v_order from public.payment_orders where payment_orders.id=p_order_id for update;
if not found then raise exception 'admin_order_not_found' using errcode='22023'; end if;
if exists(select 1 from audit.admin_audit_logs a where a.actor_user_id=p_actor_user_id
and a.request_id=btrim(p_request_id) and a.action='billing.order.'||p_action and a.target_id=p_order_id) then
return query select v_order.id,v_order.order_no,v_order.status,v_order.grant_status,v_order.grant_error,
v_order.adjustment_version,v_order.refund_status,v_order.refund_amount_cents,v_order.refunded_at,
v_order.grant_status='granted' or p_action='record_refund';
return;
end if;
if v_order.adjustment_version<>p_expected_version then
raise exception 'admin_order_version_conflict' using errcode='40001';
end if;
if p_action='retry_grant' then
if v_order.grant_status<>'failed' or v_order.epay_trade_no is null or v_order.raw_notify_payload_hash is null then
raise exception 'admin_order_retry_invalid' using errcode='23514';
end if;
select * into v_settle from public.settle_order(
v_order.order_no,v_order.epay_trade_no,v_order.money_cents,v_order.raw_notify_payload_hash
);
v_action_success := coalesce(v_settle.success,false);
elsif p_action='compensate' then
if v_order.grant_status<>'failed' or v_order.grant_type<>'credits'
or jsonb_typeof(v_order.product_snapshot)<>'object'
or jsonb_typeof(v_order.product_snapshot->'oneTimePerUser') is distinct from 'boolean' then
raise exception 'admin_order_compensation_invalid' using errcode='23514';
end if;
if nullif(v_order.product_snapshot->>'productType','') is distinct from 'credit_pack' or v_order.credits<=0 then
raise exception 'admin_order_compensation_invalid' using errcode='23514';
end if;
v_one_time := (v_order.product_snapshot->>'oneTimePerUser')::boolean;
select credits into v_balance from public.profiles where profiles.id=v_order.user_id for update;
if not found then raise exception 'profile_missing' using errcode='23514'; end if;
if v_one_time then
insert into public.user_product_redemptions(user_id,product_id,product_code,source_order_id)
values(v_order.user_id,v_order.product_id,v_order.product_code,v_order.id)
on conflict (user_id,product_code) do nothing returning user_product_redemptions.id into v_redemption_id;
if v_redemption_id is null then raise exception 'one_time_limit' using errcode='23514'; end if;
end if;
update public.profiles set credits=credits+v_order.credits,updated_at=clock_timestamp()
where profiles.id=v_order.user_id returning credits into v_balance;
insert into public.credit_transactions(user_id,transaction_type,amount,balance_after,request_id)
values(v_order.user_id,'compensation',v_order.credits,v_balance,'order-compensation:'||v_order.order_no);
update public.payment_orders set status='paid',grant_status='granted',grant_type='credits',
granted_at=clock_timestamp(),grant_error=null where payment_orders.id=v_order.id;
else
if v_order.refund_status<>'none' or v_order.status not in ('paid','granted') or v_order.paid_at is null then
raise exception 'admin_order_refund_invalid' using errcode='23514';
end if;
update public.payment_orders set status='refunded',refund_status='recorded',
refund_amount_cents=money_cents,refunded_at=clock_timestamp()
where payment_orders.id=v_order.id;
end if;
update public.payment_orders set adjustment_version=payment_orders.adjustment_version+1
where payment_orders.id=v_order.id returning * into v_order;
insert into audit.admin_audit_logs(actor_user_id,actor_email,actor_role,action,target_type,target_id,
before_value,after_value,request_id,permission_used,reason)
values(p_actor_user_id,v_actor_email,'admin','billing.order.'||p_action,'payment_order',v_order.id,
jsonb_build_object('expectedVersion',p_expected_version),
jsonb_build_object('orderNo',v_order.order_no,'status',v_order.status,'grantStatus',v_order.grant_status,
'grantError',v_order.grant_error,'adjustmentVersion',v_order.adjustment_version,
'refundStatus',v_order.refund_status,'refundAmountCents',v_order.refund_amount_cents,
'externalGatewayRefundAttempted',false,'actionSuccess',v_action_success),
btrim(p_request_id),'billing.adjustments.write',btrim(p_reason));
return query select v_order.id,v_order.order_no,v_order.status,v_order.grant_status,v_order.grant_error,
v_order.adjustment_version,v_order.refund_status,v_order.refund_amount_cents,v_order.refunded_at,v_action_success;
end;
$$;
revoke all on function public.admin_adjust_order(uuid,uuid,text,integer,text,text)
from public,anon,authenticated,service_role;
do $$ begin if exists(select 1 from pg_roles where rolname='admin_runtime') then
grant execute on function public.admin_adjust_order(uuid,uuid,text,integer,text,text) to admin_runtime;
end if; end $$;
create or replace function public.begin_consultation_credit(p_user_id uuid,p_request_id text)
returns table(success boolean,credits integer,error_code text)
language plpgsql security definer set search_path = ''
as $$
declare v_auth record;
begin
if exists(select 1 from public.consultation_requests where user_id=p_user_id and request_id=btrim(p_request_id)) then
return query select false,(select profiles.credits from public.profiles where id=p_user_id),'request_conflict'::text; return;
end if;
select * into v_auth from public.authorize_usage(p_user_id,'chat.standard',null,p_request_id,1);
if v_auth.success then insert into public.consultation_requests(user_id,request_id,status) values(p_user_id,btrim(p_request_id),'reserved'); end if;
return query select v_auth.success,v_auth.credits,v_auth.reason;
end; $$;
create or replace function public.complete_consultation_credit(p_user_id uuid,p_request_id text)
returns table(success boolean,credits integer,error_code text)
language plpgsql security definer set search_path = ''
as $$
declare v_result record;
begin
select * into v_result from public.complete_usage(p_user_id,p_request_id,'{}'::jsonb);
if v_result.success then update public.consultation_requests set status='completed',updated_at=clock_timestamp()
where user_id=p_user_id and request_id=btrim(p_request_id) and status='reserved'; end if;
return query select v_result.success,v_result.credits,v_result.error_code;
end; $$;
create or replace function public.cancel_consultation_credit(p_user_id uuid,p_request_id text)
returns table(success boolean,credits integer,error_code text)
language plpgsql security definer set search_path = ''
as $$
declare v_result record;
begin
select * into v_result from public.release_usage(p_user_id,p_request_id,'consultation_cancelled');
if v_result.success then update public.consultation_requests set status='cancelled',updated_at=clock_timestamp()
where user_id=p_user_id and request_id=btrim(p_request_id) and status='reserved'; end if;
return query select v_result.success,v_result.credits,v_result.error_code;
end; $$;
alter table public.usage_reservations enable row level security;
alter table public.usage_ledger enable row level security;
alter table public.usage_events enable row level security;
revoke all on table public.usage_reservations, public.usage_ledger, public.usage_events from public,anon,authenticated;
grant select on table public.usage_reservations, public.usage_ledger, public.usage_events to service_role;
revoke all on function public.authorize_usage(uuid,text,text,text,integer),public.complete_usage(uuid,text,jsonb),
public.release_usage(uuid,text,text),public.settle_order(text,text,integer,text),public.settle_epay_order(text,text,integer,text),
public.begin_consultation_credit(uuid,text),public.complete_consultation_credit(uuid,text),public.cancel_consultation_credit(uuid,text)
from public,anon,authenticated;
grant execute on function public.authorize_usage(uuid,text,text,text,integer),public.complete_usage(uuid,text,jsonb),
public.release_usage(uuid,text,text),public.settle_order(text,text,integer,text),public.settle_epay_order(text,text,integer,text),
public.begin_consultation_credit(uuid,text),public.complete_consultation_credit(uuid,text),public.cancel_consultation_credit(uuid,text)
to service_role;
do $$ begin if exists(select 1 from pg_roles where rolname='admin_runtime') then
grant select on table public.usage_reservations, public.usage_ledger, public.usage_events to admin_runtime;
end if; end $$;
commit;