# 统一消费 SDK：预扣、扣减与异常恢复

本指南描述当前源码；合并 main 不会自动发布 npm、PyPI、Maven 或 Go SDK 版本标签。接入方须确认所安装的制品包含这些接口，再按对应版本发布流程升级。

适用于受信任的业务后端和 Worker。TypeScript、Go、Python、Java 复用现有 Nexus API，不新增 Nexus 服务、生产迁移或支付渠道配置。已有 SDK 低层 API 保持兼容；新的流程需要接入方升级 SDK 并显式使用。

## 接入方需要提供什么

1. 绑定组织、环境的 Nexus 客户端及该业务系统稳定的 `applicationId`。
2. 所有节点共享的持久化消费日志。可以直接使用业务系统已有 PostgreSQL 连接池，不需要新增数据库服务或 Nexus 环境变量。
3. 稳定业务编号、客户编号、交付类型、商品/能力或资源编号、预计用量。
4. 业务完成后，从受信任的业务后端或供应商取得的实际用量与执行结果。不要接受浏览器自行申报用量。

SDK 自动读取并冻结发布版本中的 Offer、Meter、Provider、能力版本和配额配置。价格和失败计费规则仍由 Nexus 服务端决定，调用方无需重复维护价格映射。

| 交付 | kind | 标识与数量 | 处理 |
| --- | --- | --- | --- |
| 内容、功能访问 | ACCESS | resourceId；数量固定 1 | 派发前检查访问资格，不创建钱包扣款 |
| API/AI 等商品用量包 | QUOTA | capabilityId + bundleProductId；预计/实际数量 | 传递已购商品上下文，协调预扣、FINAL 用量与结算 |
| 存储容量 | CAPACITY | resourceId；数量为字节 | 预留容量，提交实际占用；实际为 0 时释放 |
| 成员席位 | SEAT | resourceId + 稳定 memberId；数量固定 1 | 预留并确认席位；实际为 0 时释放 |

`METERED` 是显式选择能力现有钱包/额度计费策略的调用模式，不是第五类商品交付；禁止传入 bundleProductId。`QUOTA` 必须传入当前组织内有效的商品 UUID 作为 bundleProductId；后端拒绝格式错误、零 UUID、不存在或归属其他组织的商品，不能因为忘传商品标识而转为钱包消费。商品可以是单品或套餐，传入购买时对应的商品 ID。

**本次没有开启商品额度耗尽后的自动钱包补扣。** 通用预付商品额度不足时仍由后端拒绝，不能通过切换 kind、换业务编号或丢弃商品上下文绕过。`METERED` 适用于原本按能力策略消费的独立业务入口。

## 一次性数据库安装

在业务系统自己的数据库迁移中执行一次以下 SQL。SDK 不会在请求时自动建表，不应把每个调用放在尚未提交的外层事务中。

```sql
CREATE TABLE IF NOT EXISTS nexus_consumption_journal (
  journal_key text PRIMARY KEY,
  revision bigint NOT NULL CHECK (revision > 0),
  snapshot jsonb NOT NULL,
  updated_at timestamptz NOT NULL DEFAULT clock_timestamp()
);
```

这里的表属于接入业务系统，**不是 Nexus 服务端的生产迁移**。业务若已有数据库连接配置即可复用；SDK 不要求运维额外填写日志数据库环境变量。各节点使用同一个数据库和 applicationId；不要使用进程内 Map、本地文件、异步写缓存或每节点一份数据库。

自定义 Journal 须提供原子 `load` 与 `compareAndSet`；插入冲突或版本不匹配返回空，不能覆盖已有内容。CAS 写入必须在返回前可靠提交，不能返回成功后再依赖业务外层事务提交。数据库/日志内容属于受信任服务端数据，限制修改权限并纳入备份；未完成的记录不能清理。已完成记录至少保留整个业务重试、回调和对账周期，不能依靠短期 TTL 去重。

## TypeScript

```ts
import { NexusClient, PostgresConsumptionJournal } from '@rainlib/nexus-sdk';

// dbPool 是应用已有的 PostgreSQL Pool；凭证从应用现有配置注入。
const nexus = new NexusClient({
  baseUrl: config.nexusUrl, apiKey: config.nexusKey,
  tenantId: team.nexusTenantId, environmentId: team.nexusEnvironmentId,
});
const consume = nexus.consumption({
  applicationId: 'my-service', journal: new PostgresConsumptionJournal(dbPool),
});
const request = {
  businessId: job.id, customerAccountId: job.customerId,
  kind: 'QUOTA' as const, capabilityId: job.capabilityId,
  bundleProductId: job.purchasedProductId, estimatedQuantity: '100',
};
const result = await consume.run(request, async receipt => {
  // 先保存供应商请求标识，供超时或重启后查询；不在这里另调一次 Nexus Gateway。
  const output = await provider.execute(job, { idempotencyKey: receipt.journalKey });
  await saveBusinessResult(job.id, output);
  return { value: output, actualQuantity: output.authoritativeUsage };
});
// replayed=true 时不会重跑 callback；从业务数据库读取之前保存的结果。
const output = result.replayed ? await loadBusinessResult(job.id) : result.value;
```

## Python

```python
from nexus_instrumentation import NexusClient, NexusConfig, PostgresConsumptionJournal

nexus = NexusClient(NexusConfig(
    endpoint=config.nexus_url, api_key=config.nexus_key,
    tenant_id=team.nexus_tenant_id, environment_id=team.nexus_environment_id,
))
# psycopg_pool.ConnectionPool.connection 的上下文退出时必须提交事务。
consume = nexus.consumption(PostgresConsumptionJournal(db_pool.connection), "my-service")
request = dict(
    business_id=job.id, customer_account_id=job.customer_id, kind="QUOTA",
    capability_id=job.capability_id, bundle_product_id=job.purchased_product_id,
    estimated_quantity="100",
)
def work(receipt):
    output = provider.execute(job, idempotency_key=receipt["journal_key"])
    save_business_result(job.id, output)
    return dict(value=output, actual_quantity=output.authoritative_usage)
result = consume.run(request, work)
```

Python 的这一接口是同步接口；不要传入 async callback。在异步业务中可用受控的同步 Worker，或用 prepare/dispatch 与 complete 的分段方式，不要阻塞事件循环。

## Go

```go
// nexusClient 已通过 nexus.New 配置 TenantID、EnvironmentID、APIKey。
consume, err := nexusClient.Consumption(nexus.PostgresConsumptionJournal{DB: db}, "my-service")
if err != nil { return err }
request := nexus.ConsumptionRequest{
    BusinessID: job.ID, CustomerAccountID: job.CustomerID, Kind: "QUOTA",
    CapabilityID: job.CapabilityID, BundleProductID: job.PurchasedProductID,
    EstimatedQuantity: "100",
}
result, err := consume.Run(ctx, request, func(ctx context.Context, receipt nexus.ConsumptionResult) (nexus.ConsumptionOutcome, error) {
    output, err := provider.Execute(ctx, job, receipt.JournalKey)
    if err != nil { return nexus.ConsumptionOutcome{}, err }
    if err := saveBusinessResult(ctx, job.ID, output); err != nil { return nexus.ConsumptionOutcome{}, err }
    return nexus.ConsumptionOutcome{
        ConsumptionFacts: nexus.ConsumptionFacts{ActualQuantity: output.AuthoritativeUsage},
        Value: output,
    }, nil
})
// 根据 err 和 result.Replayed 返回已保存的业务结果。
```

Go 在 callback 已返回最终事实后，用有期限、脱离调用方取消信号的上下文保存事实并结算；调用方断开不等于免费取消已发生的用量。

## Java 17+

```java
var consume = nexusClient.consumption(new PostgresConsumptionJournal(dataSource), "my-service");
var request = Map.<String,Object>of(
    "business_id", job.id(), "customer_account_id", job.customerId(), "kind", "QUOTA",
    "capability_id", job.capabilityId(), "bundle_product_id", job.purchasedProductId(),
    "estimated_quantity", "100");
var result = consume.run(request, receipt -> {
    var output = provider.execute(job, receipt.get("journal_key").toString());
    saveBusinessResult(job.id(), output);
    return new NexusConsumptionWorkflow.Outcome<>(output.authoritativeUsage(), output);
});
```

四种语言的 PostgreSQL 适配器均不内置数据库驱动；复用应用当前驱动。Java 适配器自行提交 JDBC 事务后才确认 CAS。

## 分段执行与恢复

以下以 TypeScript 为例，其他语言方法相同，使用各语言的字段命名。

```ts
await consume.prepare(request); // 先预扣，适合队列准入；此时可 cancel。
await consume.dispatch(request, receipt => enqueueOnce(job, receipt.journalKey));
// 另一个节点收到可信任务结果后：
await consume.complete(request, { actualQuantity: '62', executionStatus: 'SUCCEEDED' });
// 在派发前取消：await consume.cancel(request)
// 网络或进程恢复：await consume.recover(request)
```

`run` 一步完成派发与结算；`prepare`/`dispatch`/`complete` 适合异步任务。分段模式中，dispatch 是一次性派发声明，**不保证消息队列和数据库之间恰好一次投递**；跨服务投递使用业务 Outbox，并让供应商或下游队列接受 journalKey 作为幂等标识。进程在声明后、真实发出前退出也可能留下未知结果，必须查询业务系统/供应商进行核对，不能自动重新调用 callback。

状态和行为：

| 状态 | 后续操作 |
| --- | --- |
| PREPARING | 使用原准备请求重试预扣；不重新取价或换商品 |
| RESERVED | 可派发或取消；容量/席位派发前重新检查是否仍有效 |
| DISPATCHING | 已声明业务派发；未知结果返回 CONSUMPTION_PENDING，不重跑、不自动释放 |
| SETTLING | 用最终事实和已确认的计量回执继续结算，不重新派发 |
| COMPLETED | 返回 replayed=true；读取业务系统自己的原结果 |
| CANCELLING | 恢复原释放操作 |
| CANCELLED | 不允许同一业务编号再次派发；新业务才使用新编号 |

已知失败也需要提供真实用量，例如 `{actualQuantity:'20', executionStatus:'FAILED', failureCode:'UPSTREAM_FAILED'}`。套餐按服务端冻结的失败计费规则决定扣减量，SDK 不把真实消耗强行改成 0。容量/席位通过 actualQuantity='0' 明确表示没有产生占用。

`CONSUMPTION_PENDING` 需要保留原业务编号并核对，不能等同于“未收费”。定时恢复 Worker 可分页读取日志中非终态记录，取 snapshot.request 中的原请求调用 recover；Python/Java 可直接使用 snake_case，TypeScript/Go 需要映射到对应类型。DISPATCHING 必须先取回权威结果再调用 complete。席位实际数量须为 0 或 1；非法数量在保存前拒绝，核对后仍可提交有效事实。最终事实一旦保存就不可更改，不同用量/执行结果返回 CONSUMPTION_CONFLICT。CAS 失败通常表示另一个节点已推进；重新读取原记录即可。

prepare、run 等在远端明确拒绝时也可能返回 Pending 并保留原始错误为 cause；检查 cause 的 Nexus 错误码区分额度不足、版本过期、凭证错误与临时网络故障。不能因为出现 Pending 就无限立即重试。旧版本价格失效不会自动切到新版本，需明确处理原业务与新业务之间的关系。

日志不保存业务输出、Prompt 或支付凭证。业务系统需要持久化供应商请求 ID、返回结果、可信用量与任务状态；仅依赖 callback 的内存结果无法在进程退出后恢复。

## 当前边界

- 需要 quantity_policy 归一化或媒体计量的能力在预扣前明确拒绝，继续使用已有 session/consumption-heartbeat 集成，以免先执行后发现计量方式不支持。
- 每次高层调用处理一个能力的一种计费 Meter，或一个资源。套餐中多能力需要分别指定已购商品上下文和能力，业务编号按子步骤派生；需要所有子步骤统一准入时继续使用已有 composite execution API，不能把多个 run 当作原子事务。
- 金额和用量使用十进制字符串，支持 numeric(30,9) 范围；不要先转 JS Number/float。ACCESS/SEAT 固定为 1；存储由业务提供准确字节数。
- 执行预扣 TTL 为 30–86400 秒，容量/席位为 1–3600 秒。这是预留期限，不是商品购买有效期。
- 本封装不自动续租或扩大授权。长会话、流式任务和预计用量不足时使用已有 extend / consumption-heartbeat API 与长期会话集成；业务应在授权范围内控制工作，不能等结算失败才无限消耗。
- 已通过 Nexus Gateway 自动计费的同一次模型请求，不要再次套用此消费封装，避免双重计费。
- ACCESS 结果需要业务服务执行内容/功能访问控制；每次独立访问使用新的业务编号，不能把已完成操作的 replayed 结果当作可长期缓存的访问授权。SDK 不会自动保护外部课程、对象存储或第三方系统。

## 验证

SDK 测试覆盖并发节点只派发一次、JSONB 排序、购买商品归属上下文、十进制精度、失效预留、异步完成、取消与未知结果恢复。共享样本位于 `sdk/conformance/consumption-workflow.json`，四种语言可读取同一格式的结算日志。

`TestUnifiedSDKPurchasedConsumptionAgainstPostgres` 在本地 Node、隔离 PostgreSQL 下验证 TypeScript SDK → HTTP protobuf → 实际 Nexus 服务，包括已购额度、结算成功回执丢失、恢复、取消和失败计费。通过 `NEXUS_TEST_DATABASE_URL` 指向专用且名称含 test 的临时数据库执行；未配置时显式跳过。这不代表生产支付渠道或供应商的真实交易验收。

## 当前用户的已购访问权限

后台角色权限不代表某个买家的使用权限。商品交付项可指定 `usage_scope=PERSONAL|TEAM|SEAT`，个人使用绑定购买时的已验证用户，团队共享要求有效客户成员，指定席位要求本人的有效分配。未配置的新字段按旧账户合同兼容，不回填历史购买者。

业务后端调用 `POST /v1/entitlements/user-access:check`，传付款账户、资源或能力、动作和业务编号。用户身份来自用户 JWT；应用密钥代用户调用时，额外传 `X-Nexus-Subject-Token`，由 Nexus 验签并重新解析本地用户与成员关系。不能用裸 `user_id` 替代 JWT，也不能把服务密钥放到浏览器。令牌只存在请求上下文，不进入消费 journal。

TypeScript `client.deliveries.withUserAccess`、Go `client.WithUserAccess`、Python `client.with_user_access`、Java `client.withUserAccess` 均在明确允许后才执行回调。Go 使用 `WithUserSubject(ctx, token)`；其他语言使用独立请求客户端的 `subjectToken`／`subject_token`。每次真正业务访问均调用，不缓存长期允许结果。

用量包继续调用上述统一消费 SDK；内容使用权限检查，容量/席位仍需对应预留和确认。到期、退款、移除成员后新准入拒绝，已预扣的协调执行派发前也重新校验，清理和结算继续沿原记录恢复。已分配席位不因池余量为零而失效。完整字段、兼容规则和四语言示例见仓库 `docs/integrations/purchased-user-access.md`。这不代表外部课程系统已经部署强制拦截；它的内容、下载和播放服务端入口必须接入。

## 业务前端：是否已购买与购买弹窗

通过请求独立的可信用户身份调用 `POST /v1/entitlements/purchase-context:check`。必填 `product_id`；可传已授权的 `customer_account_id`；不传时查本人本组织/环境的有效账户，多个账户要求先选择。附加 `resource_id + action` 或 `capability_id + action=consume` 才返回具体目标的 `access_checked/can_use/access`；只查商品不能直接放行组合中的全部功能。

四语言 SDK 支持 `getPurchaseContext`（TS 在 `deliveries` 下）、`get_purchase_context`（Python）、`GetPurchaseContext`（Go）。响应 `next_action` 为 PURCHASE 且 `purchasable=true` 才弹购买框；RESUME_PAYMENT 恢复待付款原订单；WAIT 等待支付/履约/权益开始；CONTACT_ADMIN 处理席位或成员关系；SELECT_ACCOUNT 选择账户；CHECK_ACCESS 指定目标继续查；CONTACT_MERCHANT 联系商家开通账户或处理下架/暂停；USE 仍在业务入口再次强制校验。未登录先登录，错误展示重试，不把 401/403/网络失败视为未购买。

`checkout_path` 是绑定组织、环境、商品和已选账户的站内路径，新购只允许当前启用的已发布商店与实时可售商品。用 SDK 的 `purchaseContextURL`（TS）、`purchase_context_url`（Python）、`PurchaseContext.CheckoutURL`（Go）、`NexusClient.purchaseContextUrl`（Java）配合业务配置的可信 Nexus 站点 origin 生成 URL；不新增 Nexus 运维环境变量、不接受用户提供的跳转域名、不在 URL 中携带令牌。原订单恢复入口只查看和继续原记录，不能重复创建支付。付款回跳后重新查询，以服务端决策为准。

底部 Workbench 蓝图“购买状态与收银台引导”和接入中心提供四语言代码。详细字段、账户与兼容边界见仓库 `docs/integrations/purchased-user-access.md`。这份 SDK 能提供判断与引导数据；业务应用仍需实现自己的弹窗 UI，并在内容/功能/API 入口强制校验和预扣。

### Redis 与购买查询

API节点自动复用现有Redis配置：商品公共投影最长缓存30秒，购买引导最长5秒，并在已知开始/到期时间前截止。部署前需执行迁移158；业务变更与分片数据库版本推进在同一事务内提交（读取已提交计数之和，避免组织内单行写热点），使多个节点不依赖消息通知即可避开旧缓存。缓存键按组织、环境、可信用户、账户、商品、目标和动作隔离，不存JWT或放行回执。SDK和业务弹窗不需要增加缓存参数或生产环境变量。

具体资源/能力的权限校验、预扣和派发仍读取权威数据；购买缓存命中不能绕过到期、退款、成员移除或席位检查。Redis故障回源，数据库校验失败返回错误；HTTP买家响应仍禁止浏览器/CDN缓存。该机制减少重复购买与商品查询，不代表已经完成生产负载压测。


Customer identity and first activation: see `docs/public/customer-onboarding.md` and the bottom workbench “Identity and customer activation” recipe. AUTO provisions a personal billing customer only; purchased access still requires fulfillment and live authorization.