common-ecommerce/README.md

# 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<T>` | 分页载体，`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<SalesOrder> 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<SalesOrder> 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<Platform> platforms()  { return EnumSet.of(Platform.AMAZON_SC); }

    // 能力并集必须与实际实现一致
    @Override public EnumSet<Capability> capabilities() {
        return EnumSet.of(Capability.SALES_ORDER, Capability.PRODUCT,
                          Capability.MARK_SHIPPED, Capability.LABEL, Capability.BILLING);
    }

    @Override public Page<SalesOrder> 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<Platform> 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<SubChannel> 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<SalesOrder> 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 总体设计）