# common-ecommerce 多渠道电商履约对接 SDK。把亚马逊、沃尔玛、TikTok、Temu、领星聚合等渠道的商品、订单、履约、标发、面单、账单、售后统一成一套 API。 ## 定位 纯 SDK 库,嵌入宿主(OMS)进程运行。**不访问数据库、不访问 Redis、不访问 MQ、不私起线程或定时器**。所有外部副作用通过 SPI 注入,由宿主实现。 ``` ① 统一入口 ChannelSdk / 各能力门面 ② 能力接口 ProductCapable / SalesOrderCapable / FulfillmentCapable / ... ③ 平台适配 amazon-sc / walmart-sc / tiktok / temu / lingxing / ... ④ 基础设施 HttpTransport / Signer / TokenManager / RateLimiter / SyncEngine / Idempotency ⑤ 宿主 SPI AccountRepository / TokenStore / SyncCursorStore / Sink / Clock / LockProvider / ... ``` ## 模块 | 模块 | 职责 | |:---|:---| | `common-ecommerce-core` | 领域模型、能力接口、门面、管道、SPI。**零第三方运行时依赖** | | `common-ecommerce-http` | HTTP 传输、签名、令牌管理、JSON 编解码 | | `common-ecommerce-testkit` | InMemory SPI、FakeTransport、夹具加载、断言工具 | | `common-ecommerce-lingxing` | 领星聚合适配器(多子平台) | ## 构建 ```bash mvn clean test ``` 要求 JDK 1.8、Maven 3.6+。 --- ## 一、包职责 ``` apelet.common.ecommerce ├── model 领域模型与标识计算(core) ├── capability 能力接口,适配器实现这些(core) ├── error 统一异常体系(core) ├── sdk 门面、上下文、适配器工厂接口(core) ├── sync 增量同步引擎与水位线计算(core) ├── spi 宿主必须实现的扩展点(core) ├── http HTTP 传输、JSON、签名、令牌(http) ├── testkit 测试替身与夹具(testkit) └── lingxing 领星适配器(lingxing) ``` 包名统一前缀 `apelet.common.ecommerce.`,下表省略前缀。 ### `model` —— 领域模型 所有对象**不可变**,用 Builder 构造,无 setter。Java 8 不引入 record/sealed。 | 类 | 职责 | |:---|:---| | `Platform` | 业务平台枚举。**不含 provider**,直连与聚合器映射到同一值 | | `ChannelProvider` | 接入通道:DIRECT / LINGXING / TEAPPLIX / CUSTOM | | `Capability` | 能力枚举,适配器声明、门面校验 | | `FulfillmentMode` | A/C/D 三种履约模式(B 平台仓只读,不在枚举内) | | `ProductSyncStrategy` | 增量 / 全量 diff / 报告 | | `AccountKey` | `provider + platform + omsStoreId`,所有调用的隔离维度 | | `ChannelAccount` | 渠道账户配置。只存 `credentialRef`,**不存密钥** | | `CanonicalKeys` | 三级标识与指纹计算,见下方「标识体系」 | | `PlatformTime` | 三轴时间:UTC 真值 + 市场当地 + 北京时间 | | `Money` | `BigDecimal + Currency`,跨币种运算直接拒绝 | | `SalesOrder` / `OrderLine` | 3P 销售订单 | | `ProductListing` | 商品 / Listing | | `SalesOrderQuery` / `ProductQuery` | 查询条件(时间窗 + 分页游标) | | `Page` | 分页载体,`nextCursor == null` 表示结束 | ### `capability` —— 能力接口 适配器实现哪些接口就声明哪些能力;未实现的方法抛 `UnsupportedCapabilityException`,不静默返回空。 | 接口 | 关键方法 | |:---|:---| | `PlatformAdapter` | 基座:`adapterId()` / `provider()` / `platforms()` / `capabilities()` / `supports(account, cap)` | | `AggregatorCapable` | 聚合器扩展:`listSubChannels(account)`,按子平台细化能力 | | `SalesOrderCapable` | `pull(ctx, query)` / `get(ctx, orderId)` | | `ProductCapable` | `syncStrategy()` / `pull(ctx, query)` / `get(ctx, sku)` | ### `error` —— 异常体系 全部继承 `ChannelException`,带 `platform / provider / omsStoreId / operation / requestId / platformCode / retryable / rawResponse`。 | 异常 | 触发 | 可否重试 | |:---|:---|:---:| | `AuthException` | 401/403、token 失效 | 刷新后 1 次 | | `RateLimitException` | 429,携带 `retryAfterMillis()` | 是 | | `TransientException` | 5xx / 网络 / 超时 | 是 | | `ValidationException` | 4xx 参数或映射错误 | 否 | | `PlatformBusinessException` | 平台业务错误码 | 视码 | | `ChannelException.UnsupportedCapabilityException` | 能力缺失 | 否 | | `IdempotencyConflictException` | 重复写 | 否,返回已完成结果 | | `CrossProviderConflictException` | 已由另一 provider 完成 | 否,人工确认 | **注意** `UnsupportedCapabilityException` 是 `ChannelException` 的静态内部类,用 `ChannelException.UnsupportedCapabilityException` 引用。 **同名类** `Responses` 有两个:`http.Responses`(状态码 → 异常)与 `testkit.Responses`(快捷构造响应)。测试里注意 import 哪个。 ### `sdk` —— 门面与上下文 | 类 | 职责 | |:---|:---| | `ChannelContext` | 单次调用上下文:`account` + `requestId` + `clock`。**由 SDK 创建,适配器不得自造** | | `AdapterFactory` | 适配器工厂:`adapterId()` / `provider()` / `platforms()` / `create(AdapterConfig)` | | `ChannelSdk` | 统一门面。**当前为骨架,见「调用方式」中的状态说明** | ### `sync` —— 增量同步 | 类 | 职责 | |:---|:---| | `SyncSpec` | 同步维度:`provider + accountKey + stream + overlap` | | `SyncResult` | 单次结果:页数、条目数、水位线是否推进 | | `SyncEngine` | 分页循环 + 落库 + 水位线推进 | ### `spi` —— 宿主扩展点 **SDK 只定义,宿主实现。SDK 内部不碰 JDBC / Redis / MQ。** | SPI | 作用 | 典型实现 | |:---|:---|:---| | `ChannelAccountRepository` | 读渠道账户 | MySQL | | `SecretProvider` | 按 `credentialRef` 取密钥 | KMS / Vault | | `ChannelSecret` | 密钥载体(appKey/appSecret/clientId/…),只提供 `fingerprint()` 供日志 | | `TokenStore` | access/refresh token 读写 | MySQL / Redis(加密) | | `AccessToken` | 令牌载体,`needsRefresh(now, earlySec)` 判断提前续期 | | `SyncCursorStore` | 水位线读写 | MySQL | | `Sink` | 领域对象落库(当前 `upsertSalesOrders` / `upsertProducts`) | MyBatis | | `LockProvider` | token 刷新防击穿、账户级互斥 | Redisson / DB | ### `http` —— 出入网层 | 类 | 职责 | |:---|:---| | `HttpTransport` | 传输接口,默认 `OkHttpTransport`(连接 5s / 读 30s),测试用 `FakeTransport` | | `HttpRequest` / `HttpResponse` | 统一请求响应,保留原始 header/body 供留痕 | | `JsonCodec` | JSON 抽象,`JacksonJsonCodec` 提供点路径取值 `nodeAt("data.list")` | | `Signer` | 签名接口,各平台实现 | | `TokenManager` | 令牌模板:账户隔离、提前续期、`LockProvider` 防击穿、日志只打指纹 | | `Responses` | 状态码 → 异常映射(`throwIfNotSuccess`) | ### `testkit` —— 测试口子 | 类 | 职责 | |:---|:---| | `FakeTransport` | 按 URL 子串 + 方法路由,直接返回录制响应,记录全部请求 | | `InMemoryAccounts` | 内存版账户仓库 + 密钥提供者 | | `InMemoryTokenStore` / `InMemoryCursorStore` / `InMemorySink` / `InMemoryLockProvider` | 全套内存 SPI | | `Fixtures` | 夹具加载,约定 `fixtures/{platform}/{operation}/{case}.json` | | `Responses` / `TestData` | 响应与同步测试的快捷构造 | ### `lingxing` —— 领星适配器 | 类 | 职责 | |:---|:---| | `LingxingAdapter` | `SalesOrderCapable` + `AggregatorCapable`,订单列表拉取 | | `LingxingConstants` | 端点、限制(31 天 / length≤500)、`platform_code → Platform` 映射 | | `LingxingSigner` | 公共参数 ASCII 升序 + appSecret 前后包裹 + MD5 | | `LingxingTokenClient` | 换取 access_token | | `LingxingOrderMapper` | 领星报文 → `SalesOrder` | | `dto/*` | 领星原始报文结构(Jackson 映射) | | `LingxingAdapterFactory` | 工厂注册 | --- ## 二、标识体系(务必先读) provider **不得进入业务主键**,否则直连与聚合器切换会把同一订单变成两条。 | 标识 | 组成 | 用途 | |:---|:---|:---| | `omsStoreId` | OMS 渠道店铺 ID | 业务归属 | | `canonicalKey` | `platform + omsStoreId + 平台实体 ID` | upsert、去重、状态流转 | | `sourceKey` | `provider + platform + externalStoreId + 平台实体 ID` | 原始报文、水位线、审计 | ```java CanonicalKeys.salesOrder(Platform.WALMART_SC, "oms-wm-001", "PO-123"); // → "WALMART_SC:oms-wm-001:PO-123" 直连与领星算出来完全一样 ``` 写操作要两把锁: ```java // 业务幂等键:不带 provider,直连/领星共用,同一业务动作只能成功一次 CanonicalKeys.businessIdempotencyKey(Platform.WALMART_SC, "oms-wm-001", "PO-1:pkg-1", "shipping"); // provider 请求键:带 provider,单 provider 内重试去重 CanonicalKeys.providerRequestKey(ChannelProvider.DIRECT, Platform.WALMART_SC, "wm-seller-01", "shipping", "PO-1:pkg-1"); ``` --- ## 三、调用方式 ### 状态说明 `ChannelSdk` 门面目前是骨架(`builder().build()` 返回空对象,门面方法尚未挂接)。 **当前可用的调用路径是「适配器 + SyncEngine」**,即下方 3.1 / 3.2。3.3 是门面接入后的目标形态。 ### 3.1 单次查询(不推进水位线) 适合人工排查、补偿。宿主自己决定拉哪个时间窗。 ```java // 1) 宿主实现 SPI ChannelAccountRepository accountRepo = ...; // MySQL SecretProvider secretProvider = ...; // KMS TokenStore tokenStore = ...; // 加密存储 LockProvider lockProvider = ...; // Redisson / DB // 2) 创建适配器(也可经 LingxingAdapterFactory.create) LingxingAdapter adapter = new LingxingAdapter( LingxingConstants.DEFAULT_BASE_URL, new OkHttpTransport("common-ecommerce/1.0.0 (lingxing)"), new JacksonJsonCodec(), secretProvider, tokenStore, lockProvider, Clock.systemUTC()); // 3) 取账户 → 组装调用上下文 ChannelAccount account = accountRepo .findByKey("LINGXING", "WALMART_SC", "oms-wm-001") .orElseThrow(() -> new IllegalStateException("account not configured")); ChannelContext ctx = ChannelContext.of(account, UUID.randomUUID().toString(), Clock.systemUTC()); // 4) 调用 Page page = adapter.pull(ctx, SalesOrderQuery.builder() .updatedAfter(Instant.parse("2026-10-25T00:00:00Z")) // 必填,UTC .updatedBefore(Instant.parse("2026-10-26T00:00:00Z")) // 可选,缺省取 now .pageSize(100) // 平台上限内自动截断 .build()); for (SalesOrder order : page.items()) { // order.channelOrderId() 是平台单号,可直接算 canonicalKey } ``` 翻页:拿 `page.nextCursor()` 填回 `query.toBuilder().cursor(...)`(`SalesOrderQuery` 提供 `toBuilder()`),`nextCursor == null` 表示结束。 ### 3.2 增量同步(推进水位线) 适配器负责「怎么拉一页」,`SyncEngine` 负责「翻页 + 落库 + 水位线」。整批成功才推进,失败不推进。 ```java // 宿主实现这两个 SPI(测试时用 InMemoryCursorStore / InMemorySink) SyncCursorStore cursorStore = ...; Sink sink = ...; SyncSpec spec = SyncSpec.of( ChannelProvider.LINGXING, "oms-wm-001", SyncSpec.STREAM_SALES_ORDERS, Duration.ofMinutes(10)); // 重叠窗口,防边界丢单 Instant to = Instant.now(); SyncResult result = new SyncEngine(cursorStore, sink).syncSalesOrders(spec, to, (from, cursor) -> adapter.pull(ctx, SalesOrderQuery.builder() .updatedAfter(from) .updatedBefore(to) .cursor(cursor) .build())); result.pages(); // 拉了几页 result.items(); // 落了几个实体 result.watermarkAdvanced(); // true == 水位线已推进到 to ``` 要点: - **起点计算**:`from = lastWatermark - overlap`,由 `SyncSpec.from(watermark)` 完成。 - **必须先有水位线**:`cursorStore.get(...)` 返回 null 时 `SyncEngine` 抛 `IllegalStateException`,提醒先做全量建底(首次全量用创建时间回溯,成功后调 `advance`)。 - **水位线维度**:`provider + accountKey + stream`,直连与领星各存各的,不能复用。 - **调度归宿主**:SDK 不持有线程和定时器,由 OMS 调度器触发 `syncSalesOrders`。 ### 3.3 门面形态(接入后) > ⚠️ 下面这段是**目标形态**,`ChannelSdk.Builder` 的 `register` / `accountRepository` 与门面方法当前尚未实现(见第六节待办)。 ```java ChannelSdk sdk = ChannelSdk.builder() .register(new LingxingAdapterFactory(secretProvider, tokenStore, lockProvider, Clock.systemUTC())) .accountRepository(accountRepo) .build(); AccountKey key = AccountKey.of(ChannelProvider.LINGXING, Platform.WALMART_SC, "oms-wm-001"); // 能力矩阵只读,不触发真实 API CapabilityReport caps = sdk.capabilities(key); if (!caps.supports(Capability.SALES_ORDER)) { ... } // 路由完全由调用方决定,SDK 不推荐、不自动兜底 Page page = sdk.salesOrders(key).pull(query); ``` ### 3.4 能力检查 聚合器必须按子平台判断,不能只用 `capabilities()`: ```java // 领星上 eBay 的账户:只有订单,没有账单/售后 AccountKey ebay = AccountKey.of(ChannelProvider.LINGXING, Platform.EBAY, "oms-ebay-001"); adapter.supports(ebayAccount, Capability.BILLING); // false ``` 缺失能力时抛 `ChannelException.UnsupportedCapabilityException`,由调用方走 3PL / 人工兜底,**SDK 不会自动换 provider**。 --- ## 四、扩展方式 ### 4.1 新增一个直连平台 以新增 `amazon-sc` 为例,**不改 core、不改门面、不改宿主**。 **第 1 步:建模块**,`pom.xml` 只依赖 core 与 http。 **第 2 步:实现能力接口**。只实现真正支持的,其余不实现: ```java public final class AmazonScAdapter implements SalesOrderCapable, FulfillmentCapable, LabelCapable { @Override public String adapterId() { return "amazon-sc"; } @Override public ChannelProvider provider() { return ChannelProvider.DIRECT; } @Override public Set platforms() { return EnumSet.of(Platform.AMAZON_SC); } // 能力并集必须与实际实现一致 @Override public EnumSet capabilities() { return EnumSet.of(Capability.SALES_ORDER, Capability.PRODUCT, Capability.MARK_SHIPPED, Capability.LABEL, Capability.BILLING); } @Override public Page pull(ChannelContext ctx, SalesOrderQuery query) { ... } @Override public SalesOrder get(ChannelContext ctx, String orderId) { ... } } ``` **第 3 步:写签名器**,实现 `Signer`(LWA 只需 `Authorization: Bearer`,签名器可为空实现)。 **第 4 步:写报文映射**,平台 JSON → 统一模型。**平台字段名只出现在 DTO 和 Mapper 里**。 **第 5 步:实现 `AdapterFactory`**,与直连/聚合器同一套注册机制: ```java public final class AmazonScAdapterFactory implements AdapterFactory { @Override public String adapterId() { return "amazon-sc"; } @Override public ChannelProvider provider() { return ChannelProvider.DIRECT; } @Override public EnumSet platforms() { return EnumSet.of(Platform.AMAZON_SC); } @Override public PlatformAdapter create(AdapterConfig config) { return new AmazonScAdapter(...); } } ``` **第 6 步:加契约测试**(见第五节)。 **第 7 步:`ChannelSdk.builder().register(...)`**,完成。 ### 4.2 新增一个聚合器 与直连的唯一区别是**实现 `AggregatorCapable`**,按账户的子平台细化能力: ```java public final class MyAggregatorAdapter implements SalesOrderCapable, AggregatorCapable { @Override public boolean supports(ChannelAccount account, Capability capability) { if (!capabilities().contains(capability)) return false; // 按 account.platformCode() 判断,不能一刀切 return isSupportedBySubPlatform(account.platformCode(), capability); } @Override public List listSubChannels(ChannelAccount account) { ... } } ``` **不要**为每个子平台新建模块——子平台是配置(`platformCode + externalStoreId`),不是代码。 ### 4.3 新增一个能力(如退款) 1. `Capability` 加枚举值; 2. `capability/` 下加接口(可选动作给 `default` 方法抛 `UnsupportedCapabilityException`,适配器只覆写支持的); 3. 适配器声明能力并实现; 4. 领域模型加实体(`Refund` / `ReturnOrder` …); 5. `Sink` 加对应 `upsert`,testkit 加 `InMemory` 实现。 **不会改动的**:`PlatformAdapter`、`ChannelContext`、`AccountKey`、`CanonicalKeys`、`SyncEngine`。 ### 4.4 新增一个 SPI 宿主可能需要新的副作用出口(如 `RefundLedger`、`RawPayloadStore`)。 1. `spi/` 下定义**纯接口**,不引入 JDBC/Redis 类型; 2. 提供 `InMemory` 实现进 testkit,保证适配器可脱离宿主测试; 3. 通过 `ChannelSdk.Builder` 或适配器构造器注入。 **判断标准**:是否需要持久化或跨进程状态?是 → 新 SPI;否 → 放管道层(http)。 ### 4.5 扩展红线 | 不做 | 原因 | |:---|:---| | `core` 引 Spring / JDBC / Redis / MQ | SDK 要能脱离宿主单测 | | 把平台端点、字段名做成配置表 | 协议是代码,配置化会让线上线下跑不同协议 | | 密钥写入 `ChannelAccount` | 只能存 `credentialRef`,密钥走 `SecretProvider` | | provider 进 canonicalKey | 直连/聚合器切换会产生重复实体 | | 在适配器里 `new Thread` / 定时器 | 调度归宿主,SDK 不许私起线程 | | 能力声明与实际实现不一致 | 验收项,缺失能力必须显式抛异常 | | 写操作盲目重试 | 必须有幂等键;退款/标发超时先回查 | --- ## 五、测试 ### 分层 | 层级 | 测什么 | 依赖 | |:---|:---|:---| | 单元测试 | 签名、canonicalKey、时间三轴/DST、错误映射 | 纯计算,毫秒级 | | **契约测试**(核心) | 请求是否构造对、响应是否解析对 | 录制夹具 | | 集成测试 | 重试、限流、401 刷新 | `FakeTransport` | | 沙箱联调 | 端到端 | 手动,不进 CI | 适配器测试**不依赖网络、数据库、Spring**。 ### 夹具约定 ``` src/test/resources/fixtures/{platform}/{operation}/{case}.json ``` 例:`lingxing/src/test/resources/fixtures/lingxing/orderList/success.json` 每个平台至少准备「正常 + 异常」两套。 ### 写法 ```java FakeTransport transport = FakeTransport.create() .route(LingxingConstants.PATH_ACCESS_TOKEN, "POST", r -> Responses.ok(tokenJson)) .route(LingxingConstants.PATH_ORDER_LIST, "POST", r -> Responses.ok(orderJson)); LingxingAdapter adapter = new LingxingAdapter(LingxingConstants.DEFAULT_BASE_URL, transport, new JacksonJsonCodec(), accounts, tokenStore, locks, Clock.fixed(Instant.parse("2026-10-29T00:00:00Z"), ZoneOffset.UTC)); ChannelContext ctx = ChannelContext.of(account, "req-1", clock); Page page = adapter.pull(ctx, SalesOrderQuery.builder() .updatedAfter(Instant.parse("2026-10-25T00:00:00Z")) .updatedBefore(Instant.parse("2026-10-26T00:00:00Z")) .build()); // 断言请求构造 HttpRequest last = transport.lastRequest(); assertTrue(last.url().contains("sign=")); assertTrue(last.body().contains("\"platform_code\":10008")); // 断言解析结果 assertEquals("PO-123456789", page.items().get(0).channelOrderId()); ``` 完整示例见 `lingxing/src/test/java/.../LingxingAdapterTest.java`。 ### 执行 ```bash mvn test # 全部 mvn -pl lingxing test # 单模块 mvn -pl lingxing -Dtest=LingxingAdapterTest test ``` --- ## 六、待办与待核对 **待实现** | 项 | 说明 | |:---|:---| | `ChannelSdk` 门面 | 当前为空壳,`create` 与门面方法待挂接 | | `ProviderResolver` | 按 `AccountKey` 解析适配器的注册表 | | `IdempotencyStore` + 双锁 | 写操作幂等尚未装配 | | `RawPayloadStore` | 原始报文留痕 | | `FakeDirectAdapter` | 验证 core 抽象未被聚合器带偏 | | 其余由文档 `Spec 0` 定义的能力 | 履约、面单、账单、售后、Webhook、异步报表 | **待核对(需官方文档 + 真实沙箱,联调时集中修改)** | 位置 | 内容 | |:---|:---| | `LingxingOrderDto` | 字段名:`platform_order_no` / `platform_order_names` / `purchase_time` / `item_list` | | `LingxingSigner` | 签名大小写、是否排除空值参数 | | `LingxingTokenClient` | 令牌接口请求体字段名与响应嵌套层级 | | `LingxingAdapter.get()` | 单订单详情接口是否存在 | ## 设计文档 - OMS 侧 `cross-border-fulfillment-sdk-design.md`(Spec 0 总体设计)