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	领星聚合适配器（多子平台）

构建

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	原始报文、水位线、审计

CanonicalKeys.salesOrder(Platform.WALMART_SC, "oms-wm-001", "PO-123");
// → "WALMART_SC:oms-wm-001:PO-123"   直连与领星算出来完全一样

写操作要两把锁：

// 业务幂等键：不带 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 单次查询（不推进水位线）

适合人工排查、补偿。宿主自己决定拉哪个时间窗。

// 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 负责「翻页 + 落库 + 水位线」。整批成功才推进，失败不推进。

// 宿主实现这两个 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 与门面方法当前尚未实现（见第六节待办）。

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()：

// 领星上 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 步：实现能力接口。只实现真正支持的，其余不实现：

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，与直连/聚合器同一套注册机制：

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，按账户的子平台细化能力：

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

每个平台至少准备「正常 + 异常」两套。

写法

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。

执行

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 总体设计）