feat(billing): 订阅制后端 —— 手动购买 + 周期发放积分 + 到期失效

规格(按需求):用户扫码买一个订阅周期,有效期内每 N 天发一次积分,到期即失效,
不自动续费。N 与每次发放额度都在套餐里配,后台可改。

为什么不做自动续费:微信 Native 扫码支付没有代扣能力,真自动续费要走「委托代扣」
——另一套产品与资质。与其假装有,不如把"到期即失效"这个语义做扎实。

两个决定,都写进了代码注释:
- **发放语义是累加而非重置**。每次刷新写一条 grant 分录、余额累加。重置型
  (月度配额清零)会让「余额 = SUM(ledger)」这条对账不变量变复杂,且有误清
  用户自费积分的风险。
- **订阅开通放在 store.MarkOrderPaid 内**,而不是各调用方。回调与掉单补偿两条
  路都经过它,放这一处才没人能漏掉;按 orderID 幂等,重复调用无害。

复用而非另造:订阅单与积分包单走同一条支付链路(下单/回调/查单/掉单补偿),
只是 kind=sub 且 credits_micro=0——积分不在付款时给,由订阅按周期发。

定时器每 10 分钟扫一轮,语义与幂等都在 store.TickSubscription 里,与手动触发
共用,不会两处漂移。停机期间欠下的发放会一次性补齐。

8 组测试。其中一条当场抓到真 bug:开通时原本无条件发一笔,同一订单重复开通
(回调重推/查单赛跑)会因序号自增绕过幂等索引,白送积分。改为开通也走与定时器
同一套排期判断——排期天然幂等。教训:幂等键要锚在业务时间轴上,不能靠自增序号。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Blizzard
2026-07-20 16:11:34 +08:00
parent 466959d3c3
commit 80d4aaf4ba
9 changed files with 836 additions and 96 deletions
+76 -70
View File
@@ -13,12 +13,12 @@ import (
"github.com/prometheus/client_golang/prometheus/promhttp"
"go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin"
"github.com/sundynix/sundynix-shared/blob"
"github.com/sundynix/sundynix-gateway/internal/handler"
"github.com/sundynix/sundynix-gateway/internal/middleware"
"github.com/sundynix/sundynix-gateway/internal/nats"
"github.com/sundynix/sundynix-gateway/internal/store"
"github.com/sundynix/sundynix-gateway/internal/webui"
"github.com/sundynix/sundynix-shared/blob"
)
// New 构建带有 Guardrail / 限流中间件的 Gin 引擎。
@@ -28,18 +28,19 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob.
r.Use(otelgin.Middleware("sundynix-gateway")) // OTel: HTTP server span(链路根 + 提取上游 traceparent
r.Use(middleware.RequestID()) // 生成/透传 X-Request-ID(日志关联)
r.Use(middleware.Observe()) // Prometheus 指标 + 结构化访问日志(替代 gin 默认文本日志)
r.Use(cors()) // 桌面端/浏览器跨源访问
r.Use(middleware.Auth()) // 解析 Bearer JWT,注入已验证 userID(非阻断)——须在限流前,供按用户限流
r.Use(middleware.TenantContext(db)) // 多租户:注入当前 tenant_id(已登录才解析;须在 Auth 之后)
r.Use(middleware.SpaceContext(db)) // 共享工作区:注入当前 space_id(须在 TenantContext 之后)
r.Use(middleware.RateLimit(cache)) // 已认证按用户限流,否则按 IP(企业网多人共享 IP 不再互相拖累)
r.Use(middleware.Guardrail(db)) // Harness: Input Guardrail(命中落库 guardrail_event
r.Use(cors()) // 桌面端/浏览器跨源访问
r.Use(middleware.Auth()) // 解析 Bearer JWT,注入已验证 userID(非阻断)——须在限流前,供按用户限流
r.Use(middleware.TenantContext(db)) // 多租户:注入当前 tenant_id(已登录才解析;须在 Auth 之后)
r.Use(middleware.SpaceContext(db)) // 共享工作区:注入当前 space_id(须在 TenantContext 之后)
r.Use(middleware.RateLimit(cache)) // 已认证按用户限流,否则按 IP(企业网多人共享 IP 不再互相拖累)
r.Use(middleware.Guardrail(db)) // Harness: Input Guardrail(命中落库 guardrail_event
h := handler.New(db, cache, bus, blobStore)
// 微信支付渠道装配:DB 配置优先(admin 控制面热重载)→ env 兜底 → 隐藏。失败不阻断启动。
h.InitWechat(context.Background())
// 掉单补偿定时器:周期扫 pending 微信单确认到账(用户扫完码关页面也能补入账)。
h.StartReconcile(context.Background())
h.StartSubscriptionTicker(context.Background()) // 订阅按周期发放积分 + 到期置失效
// 可观测性根端点:Prometheus 抓取 + k8s 存活/就绪探针(不挂业务中间件鉴权)。
r.GET("/metrics", gin.WrapH(promhttp.Handler()))
@@ -49,73 +50,75 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob.
api := r.Group("/api/v1")
{
// —— 公开:鉴权端点 / 健康 / 按 task_id 寻址的 SSE 与导出(EventSource/下载无法带 Bearer)——
api.POST("/auth/register", h.Register) // 注册 + 签发 JWT
api.POST("/auth/login", h.Login) // 登录 + 签发 JWT
api.GET("/auth/me", h.Me) // 当前登录用户(无效令牌 → 401)
api.GET("/health", h.Health) // 依赖健康聚合(顶栏五盏灯)
api.GET("/tasks/:id/stream", h.StreamTask) // SSE 回流 Token Streamtask_id 寻址)
api.GET("/tasks/:id/exec", h.StreamExec) // SSE 回流执行轨迹(task_id 寻址)
api.GET("/kb/ingest/:id/stream", h.KbIngestStream) // 入库进度 SSEjob_id 寻址)
api.GET("/reports/:id/export", h.ExportReport) // 按需导出(report_id 寻址)
api.GET("/reports/:id/download", h.ExportReport) // 兼容旧入口(默认 docx
api.POST("/auth/register", h.Register) // 注册 + 签发 JWT
api.POST("/auth/login", h.Login) // 登录 + 签发 JWT
api.GET("/auth/me", h.Me) // 当前登录用户(无效令牌 → 401)
api.GET("/health", h.Health) // 依赖健康聚合(顶栏五盏灯)
api.GET("/tasks/:id/stream", h.StreamTask) // SSE 回流 Token Streamtask_id 寻址)
api.GET("/tasks/:id/exec", h.StreamExec) // SSE 回流执行轨迹(task_id 寻址)
api.GET("/kb/ingest/:id/stream", h.KbIngestStream) // 入库进度 SSEjob_id 寻址)
api.GET("/reports/:id/export", h.ExportReport) // 按需导出(report_id 寻址)
api.GET("/reports/:id/download", h.ExportReport) // 兼容旧入口(默认 docx
api.POST("/billing/callback/:channel", h.PaymentCallback) // 支付回调(渠道服务器带不了 Bearer;渠道验签是唯一的门)
// —— 受保护:owner 作用域业务,必须携带有效 JWT ——
p := api.Group("", middleware.RequireAuth())
{
p.POST("/tasks", middleware.RequireTenantRole(db, store.RoleMember), h.SubmitTask) // 提交任务(烧租户积分):viewer 只读拦下
p.GET("/tasks/:id", h.TaskStatus) // 任务生命周期状态(UI 轮询 submitted/running/done/failed/timeout/waiting/rejected
p.POST("/tasks/:id/approve", middleware.Audit(db), h.ApproveTask) // HITL 人工审批决定(批准/拒绝,审计)
p.GET("/tenants/current", h.TenantCurrent) // 当前租户上下文 + 角色 + 可花余额(多租户)
p.GET("/me/tenants", h.MyTenantsList) // 我所属租户(供切换)
p.POST("/me/tenants", h.CreateMyTenant) // 自助建组织(创建者即 owner,建完切入)
p.POST("/me/tenant", h.SwitchTenant) // 切换当前活跃租户
p.GET("/tasks/:id", h.TaskStatus) // 任务生命周期状态(UI 轮询 submitted/running/done/failed/timeout/waiting/rejected
p.POST("/tasks/:id/approve", middleware.Audit(db), h.ApproveTask) // HITL 人工审批决定(批准/拒绝,审计)
p.GET("/tenants/current", h.TenantCurrent) // 当前租户上下文 + 角色 + 可花余额(多租户)
p.GET("/me/tenants", h.MyTenantsList) // 我所属租户(供切换)
p.POST("/me/tenants", h.CreateMyTenant) // 自助建组织(创建者即 owner,建完切入)
p.POST("/me/tenant", h.SwitchTenant) // 切换当前活跃租户
// 租户成员自助管理(薄 Web 面):作用于活跃租户,看 ≥viewer、写 ≥admin + 审计。
p.GET("/tenants/current/members", middleware.RequireTenantRole(db, store.RoleViewer), h.TenantMembers)
p.POST("/tenants/current/members", middleware.RequireTenantRole(db, store.RoleAdmin), middleware.Audit(db), h.TenantAddMember)
p.PUT("/tenants/current/members/:uid", middleware.RequireTenantRole(db, store.RoleAdmin), middleware.Audit(db), h.TenantSetMemberRole)
p.DELETE("/tenants/current/members/:uid", middleware.RequireTenantRole(db, store.RoleAdmin), middleware.Audit(db), h.TenantRemoveMember)
p.GET("/me/usage", h.MyUsage) // 我的用量明细(余额 + 趋势 + 最近消耗)
p.GET("/tasks/:id/eval", h.TaskEval) // 自动化评测结果(综合/质量/忠实度/分级)
p.PUT("/memory", h.SetMemory) // 偏好记忆登记(→ mcp-go memory_upsert
p.GET("/memory", h.ListMemory) // 列出当前用户偏好(记忆面板)
p.DELETE("/memory", h.DeleteMemory) // 软删一条偏好(?key=
p.GET("/kb/list", h.KbList) // 当前空间的知识库列表(共享工作区)
p.GET("/me/usage", h.MyUsage) // 我的用量明细(余额 + 趋势 + 最近消耗)
p.GET("/tasks/:id/eval", h.TaskEval) // 自动化评测结果(综合/质量/忠实度/分级)
p.PUT("/memory", h.SetMemory) // 偏好记忆登记(→ mcp-go memory_upsert
p.GET("/memory", h.ListMemory) // 列出当前用户偏好(记忆面板)
p.DELETE("/memory", h.DeleteMemory) // 软删一条偏好(?key=
p.GET("/kb/list", h.KbList) // 当前空间的知识库列表(共享工作区)
p.POST("/kb/create", middleware.RequireSpaceRole(db, store.RoleMember), h.KbCreate) // 新建知识库:空间只读 viewer 拦下
p.POST("/kb/ingest", middleware.RequireSpaceRole(db, store.RoleMember), h.KbIngest) // 文本入库:viewer 拦下
p.POST("/kb/ingest_file", middleware.RequireSpaceRole(db, store.RoleMember), h.KbIngestFile) // 文件入库:viewer 拦下
p.POST("/kb/search", h.KbSearch) // 检索台(读,全员)
p.GET("/kb/vault", h.KbVault) // 文库列表
p.GET("/kb/doc", h.KbDoc) // 取单篇文档
p.POST("/kb/search", h.KbSearch) // 检索台(读,全员)
p.GET("/kb/vault", h.KbVault) // 文库列表
p.GET("/kb/doc", h.KbDoc) // 取单篇文档
p.DELETE("/kb/doc", middleware.RequireSpaceRole(db, store.RoleMember), h.KbDeleteDoc) // 级联删文档:viewer 拦下
// Prompt 控制面(平台级配置:建版本 → 激活 → 控制面热下发各服务)
p.GET("/prompts", h.PromptList) // 列出全部版本 + 可配键
p.POST("/prompts/version", h.PromptCreateVersion) // 建新版本(不自动激活)
p.POST("/prompts/activate", middleware.Audit(db), h.PromptActivate) // 激活某版本 → 广播热更新(审计)
p.POST("/prompts/deactivate", middleware.Audit(db), h.PromptDeactivate) // 撤销激活 → 回退代码默认(热,审计)
p.GET("/kb/links", h.KbLinks) // 某库双链
p.POST("/kb/note", middleware.RequireSpaceRole(db, store.RoleMember), h.KbSaveNote) // 新建/编辑笔记:viewer 拦下
p.GET("/kb/graph", h.KbGraph) // 知识图谱三元组
p.GET("/agents", h.AgentList) // 当前空间的编排列表(共享工作区,含创建人)
p.GET("/prompts", h.PromptList) // 列出全部版本 + 可配键
p.POST("/prompts/version", h.PromptCreateVersion) // 建新版本(不自动激活)
p.POST("/prompts/activate", middleware.Audit(db), h.PromptActivate) // 激活某版本 → 广播热更新(审计)
p.POST("/prompts/deactivate", middleware.Audit(db), h.PromptDeactivate) // 撤销激活 → 回退代码默认(热,审计)
p.GET("/kb/links", h.KbLinks) // 某库双链
p.POST("/kb/note", middleware.RequireSpaceRole(db, store.RoleMember), h.KbSaveNote) // 新建/编辑笔记:viewer 拦下
p.GET("/kb/graph", h.KbGraph) // 知识图谱三元组
p.GET("/agents", h.AgentList) // 当前空间的编排列表(共享工作区,含创建人)
p.POST("/agents", middleware.RequireSpaceRole(db, store.RoleMember), h.AgentSave) // 保存/更新编排:空间只读 viewer 拦下
p.DELETE("/agents", middleware.RequireSpaceRole(db, store.RoleMember), h.AgentDelete) // 删除编排:viewer 拦下(删他人另需 admin,见 handler
// 共享工作区(Space):切换 / 列表 / 建 / 成员管理(增量3)
p.GET("/me/spaces", h.SpacesList) // 活跃租户内我所属的空间(供切换)
p.POST("/me/space", h.SwitchSpace) // 切换活跃空间
p.GET("/spaces/current", h.SpaceCurrent) // 当前空间上下文 + 我的角色
p.POST("/spaces", middleware.RequireTenantRole(db, store.RoleMember), h.SpaceCreate) // 建空间:租户只读 viewer 拦下
p.GET("/spaces/:id/members", h.SpaceMembers) // 空间成员列表
p.POST("/spaces/:id/members", h.SpaceAddMember) // 拉人进空间(handler 内校验空间 admin
p.PUT("/spaces/:id/members/:uid", h.SpaceSetMemberRole) // 改空间成员角色
p.DELETE("/spaces/:id/members/:uid", h.SpaceRemoveMember) // 移除空间成员
p.POST("/spaces/:id/archive", h.SpaceArchive) // 归档空间
p.GET("/me/spaces", h.SpacesList) // 活跃租户内我所属的空间(供切换)
p.POST("/me/space", h.SwitchSpace) // 切换活跃空间
p.GET("/spaces/current", h.SpaceCurrent) // 当前空间上下文 + 我的角色
p.POST("/spaces", middleware.RequireTenantRole(db, store.RoleMember), h.SpaceCreate) // 建空间:租户只读 viewer 拦下
p.GET("/spaces/:id/members", h.SpaceMembers) // 空间成员列表
p.POST("/spaces/:id/members", h.SpaceAddMember) // 拉人进空间(handler 内校验空间 admin
p.PUT("/spaces/:id/members/:uid", h.SpaceSetMemberRole) // 改空间成员角色
p.DELETE("/spaces/:id/members/:uid", h.SpaceRemoveMember) // 移除空间成员
p.POST("/spaces/:id/archive", h.SpaceArchive) // 归档空间
p.POST("/reports", middleware.RequireTenantRole(db, store.RoleMember), h.GenerateReport) // 报告生成(同样烧租户积分):viewer 只读拦下
p.GET("/billing", h.Billing)
// 充值(P5.1 兑换码 + P5.2 微信 Native):动钱的 ≥member + 审计;查询全员可看。
p.GET("/billing/packs", h.BillingPacks)
p.GET("/billing/sub-plans", h.BillingSubPlans) // 在售订阅套餐
p.GET("/billing/subscription", h.MySubscription) // 我的订阅(到期时间/发放次数)
p.GET("/billing/orders", h.BillingOrders)
p.GET("/billing/orders/:id", h.BillingOrderStatus) // 轮询单态(pending 时顺路主动查单确认)
p.POST("/billing/redeem", middleware.RequireTenantRole(db, store.RoleMember), middleware.Audit(db), h.BillingRedeem)
@@ -133,9 +136,9 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob.
admin.POST("/models/:id/active", h.SetActiveModel)
admin.DELETE("/models/:id", h.DeleteModel)
admin.POST("/models/test", h.TestModel)
admin.GET("/pricing", h.ListPricing) // 各模型计价(token↔真钱 + 积分权重)
admin.PUT("/pricing", h.SavePricing) // 设置某模型输入/输出单价 + 积分权重
admin.GET("/billing-config", h.BillingConfig) // 全局计费规则(token→积分汇率 + 硬拦截开关)
admin.GET("/pricing", h.ListPricing) // 各模型计价(token↔真钱 + 积分权重)
admin.PUT("/pricing", h.SavePricing) // 设置某模型输入/输出单价 + 积分权重
admin.GET("/billing-config", h.BillingConfig) // 全局计费规则(token→积分汇率 + 硬拦截开关)
admin.PUT("/billing-config", h.SaveBillingConfig)
admin.POST("/credits/grant", h.GrantCredits) // 给租户充值/发放积分
// 支付配置面(P5.1/P5.2):兑换码生成/查看 + 积分包配置 + 微信支付配置(DB 热生效)
@@ -145,32 +148,35 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob.
admin.PUT("/packs", h.AdminSavePack)
admin.GET("/payment/wechat", h.AdminGetWechatPay)
admin.PUT("/payment/wechat", h.AdminSaveWechatPay)
admin.GET("/orders", h.AdminOrders) // 全平台充值订单流 + 状态计数
admin.GET("/orders/reconcile", h.AdminReconcile) // 日终对账:paid 单 ↔ 账本 grant
admin.GET("/sub-plans", h.AdminSubPlans) // 订阅套餐(含下架)
admin.PUT("/sub-plans", h.AdminSaveSubPlan) // 配价格/时长/发放节奏
admin.GET("/subscriptions", h.AdminSubscriptions) // 全平台订阅观测
admin.GET("/orders", h.AdminOrders) // 全平台充值订单流 + 状态计数
admin.GET("/orders/reconcile", h.AdminReconcile) // 日终对账:paid 单 ↔ 账本 grant
admin.POST("/orders/:id/refund", h.AdminRefundOrder) // 人工退款:置 refunded + adjust 负分录 + 回退余额(审计)
// 多租户成员管理(平台运维口径)
admin.GET("/tenants", h.AdminTenants) // 租户目录(成员数+余额)
admin.POST("/tenants", h.AdminCreateTenant) // 新建租户(可选指定 owner
admin.GET("/tenants", h.AdminTenants) // 租户目录(成员数+余额)
admin.POST("/tenants", h.AdminCreateTenant) // 新建租户(可选指定 owner
admin.PUT("/tenants/:id/shared-billing", h.AdminSetSharedBilling) // 共享计费开关
admin.PUT("/tenants/:id/plan", h.AdminSetTenantPlan) // 方案等级
admin.PUT("/tenants/:id/status", h.AdminSetTenantStatus) // 租户状态
admin.GET("/tenants/:id/members", h.AdminMembers) // 成员列表
admin.POST("/tenants/:id/members", h.AdminAddMember) // 按邮箱加成员
admin.PUT("/tenants/:id/members/:uid", h.AdminSetMemberRole) // 改角色
admin.GET("/tenants/:id/members", h.AdminMembers) // 成员列表
admin.POST("/tenants/:id/members", h.AdminAddMember) // 按邮箱加成员
admin.PUT("/tenants/:id/members/:uid", h.AdminSetMemberRole) // 改角色
admin.DELETE("/tenants/:id/members/:uid", h.AdminRemoveMember) // 移除成员
admin.GET("/status", h.AdminStatus) // 服务状态:基建/服务探活 + MCP 工具注册
admin.GET("/overview", h.AdminOverview) // 系统级聚合:全平台用户/任务/评测/模型态/提示词态/健康
admin.GET("/tasks", h.AdminTasks) // 全平台任务/运行观测(状态/租户筛 + HITL 待审批)
admin.GET("/tasks/:id", h.AdminTaskDetail) // 任务下钻:DSL/输出/轨迹/评测(跨租户,走 WithoutTenant
admin.GET("/spaces", h.AdminSpaces) // 全平台空间观测(跨租户 Space + 成员数)
admin.GET("/usage", h.AdminUsage) // 用量/积分/成本:全平台按天趋势 + 租户排行 / 单租户余额
admin.GET("/evals", h.AdminEvals) // 自动评测观测:质量趋势 + 计数 + 错题本(真数据)
admin.GET("/datasources", h.AdminDatasources) // 数据源清单:全平台知识库 + 文档数(真数据)
admin.POST("/kb/search", h.AdminKbSearch) // 检索试验台:按完整作用域键跨租户检索(支持单路 mode 对比)
admin.POST("/migrate-kb-storage", h.MigrateKBStorage) // 增量3:存量 KB 三库 owner/kb→space/kb 重灌(一次性)
admin.GET("/audit", h.AuditList) // 敏感操作审计流(倒序,翻页)
admin.GET("/guardrail-events", h.GuardrailEvents) // 护栏命中安全事件流(倒序,翻页)
admin.GET("/status", h.AdminStatus) // 服务状态:基建/服务探活 + MCP 工具注册
admin.GET("/overview", h.AdminOverview) // 系统级聚合:全平台用户/任务/评测/模型态/提示词态/健康
admin.GET("/tasks", h.AdminTasks) // 全平台任务/运行观测(状态/租户筛 + HITL 待审批)
admin.GET("/tasks/:id", h.AdminTaskDetail) // 任务下钻:DSL/输出/轨迹/评测(跨租户,走 WithoutTenant
admin.GET("/spaces", h.AdminSpaces) // 全平台空间观测(跨租户 Space + 成员数)
admin.GET("/usage", h.AdminUsage) // 用量/积分/成本:全平台按天趋势 + 租户排行 / 单租户余额
admin.GET("/evals", h.AdminEvals) // 自动评测观测:质量趋势 + 计数 + 错题本(真数据)
admin.GET("/datasources", h.AdminDatasources) // 数据源清单:全平台知识库 + 文档数(真数据)
admin.POST("/kb/search", h.AdminKbSearch) // 检索试验台:按完整作用域键跨租户检索(支持单路 mode 对比)
admin.POST("/migrate-kb-storage", h.MigrateKBStorage) // 增量3:存量 KB 三库 owner/kb→space/kb 重灌(一次性)
admin.GET("/audit", h.AuditList) // 敏感操作审计流(倒序,翻页)
admin.GET("/guardrail-events", h.GuardrailEvents) // 护栏命中安全事件流(倒序,翻页)
}
}