From 83b874322a4e56a65356326f31aa3fe236f32d30 Mon Sep 17 00:00:00 2001 From: sunlei Date: Mon, 27 Jul 2026 14:39:05 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A2=9E=E5=8A=A0TCP=20NATMap=20DDNS?= =?UTF-8?q?=E4=B8=8E=E7=AB=AF=E7=82=B9=E6=B6=88=E6=81=AF=E6=BA=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- API.md | 8 +- README.md | 6 +- .../admin-platform-config.module.ts | 2 + .../network-agent-mqtt.service.ts | 98 +++- .../network-ddns.service.ts | 101 +++- .../network-management.types.ts | 3 + ...twork-tcp-natmap-message-source.adapter.ts | 511 ++++++++++++++++++ .../network-tcp-natmap-source-eligibility.ts | 31 ++ ...em-message-delivery-coordinator.service.ts | 73 +-- .../network-agent-mqtt.service.spec.ts | 145 +++++ .../network-ddns.service.spec.ts | 175 ++++++ .../network-management.entity.spec.ts | 3 + ...-tcp-natmap-message-source.adapter.spec.ts | 286 ++++++++++ ...work-tcp-natmap-source-eligibility.spec.ts | 41 ++ .../qqbot-message-push.controller.spec.ts | 60 +- ...e-delivery-coordinator-integration.spec.ts | 26 +- .../system-message-delivery-runner.spec.ts | 39 +- .../system-message-fanout.service.spec.ts | 47 ++ .../system-message-source.registry.spec.ts | 16 +- 19 files changed, 1572 insertions(+), 99 deletions(-) create mode 100644 src/modules/admin/platform-config/network-management/network-tcp-natmap-message-source.adapter.ts create mode 100644 src/modules/admin/platform-config/network-management/network-tcp-natmap-source-eligibility.ts create mode 100644 test/admin/network-management/network-tcp-natmap-message-source.adapter.spec.ts create mode 100644 test/admin/network-management/network-tcp-natmap-source-eligibility.spec.ts diff --git a/API.md b/API.md index fa8ca58..082eff5 100644 --- a/API.md +++ b/API.md @@ -137,13 +137,13 @@ Wire contract 同时保留 schema-v1 兼容路径和严格 schema-v2。v2 外层 TCP current 必须同时满足 present、NATMap intent、synced、Router、DNAT、NATMap active、非空 generation、candidate/current/last-observed 同 tuple，以及接收时仍新鲜且未倒退的 lease；否则只保留候选/最近观测并撤下 current。candidate、current、last-observed、last-published、静态错误、Keeper 错误和 NATMap 错误分别持久化。事件按 `eventId` 幂等追加并记录 `udp_stun|tcp_natmap` mechanism。 -v2 endpoint event 与 matching reported 是两条独立 MQTT 消息。event、report 和 ownership 切换统一按 Agent state、channel、history 顺序加锁。正常 event-first 路径先提交 history，不提前创建会被 5 秒扫描器消费的 UDP Outbox；history 耐久保存来源 revision 和由 mechanism、tuple、observed/validated/valid-until 原始值生成的完整 lease identity，matching reported 只有在 revision、组/协议/mechanism、tuple 和未被 JS `Date` 截断的 lease 身份完全相同时，才在 current 同一事务中以原 `eventId` stage 现有 STUN 事件，提交后 wake。report-first 使用同一精确关联门禁。TCP 在本阶段只写历史，不复用 UDP STUN 消息源。组删除只有所有通道分别确认 exact absent 后才完成；先完成的通道单独软删除并清空活动键，仍有通道的组继续保留。v1 单 UDP 组继续使用 helper/router/route/Keeper/current 的原有 exact-absence 门禁。 +v2 endpoint event 与 matching reported 是两条独立 MQTT 消息。event、report 和 ownership 切换统一按 Agent state、channel、history 顺序加锁。正常 event-first 路径先提交 history，不提前创建会被 5 秒扫描器消费的 Outbox；history 耐久保存来源 revision 和由 mechanism、tuple、observed/validated/valid-until 原始值生成的完整 lease identity，matching reported 只有在 revision、组/协议/mechanism、tuple 和未被 JS `Date` 截断的 lease 身份完全相同时，才在 current 同一事务中以原 `eventId` stage 对应事件，提交后 wake。report-first 使用同一精确关联门禁。UDP 仅对有效端口变化使用既有 `network.stun.mapping-port-changed`，TCP 对公网 IPv4 或端口任一变化使用独立 `network.tcp.natmap-endpoint-changed`；首次发布、同 tuple 续租、撤下和同 tuple 恢复不 stage TCP 事件。组删除只有所有通道分别确认 exact absent 后才完成；先完成的通道单独软删除并清空活动键，仍有通道的组继续保留。v1 单 UDP 组继续使用 helper/router/route/Keeper/current 的原有 exact-absence 门禁。 Admin 首次进入网络管理页通过 HTTP 读取快照，随后使用 `/system/network/events/stream` 接收 `network-state-changed`。API 只在 `reported`、`status`、`events` 对应事务提交且语义状态实际变化后发出事件；MQTT QoS 1 重投、仅推进 `lastHeartbeatAt` 的状态心跳，以及相同 tuple 只推进 `validatedAt/validUntil` 的续租继续持久化，但不触发 history、SSE、DDNS 或消息 Outbox。公网 IP/端口、candidate/last-observed tuple、Keeper/NATMap/同步/错误/删除状态或 Agent 在线会话变化仍发布事件。一个入站消息不论改变多少通道都最多产生一次提交后 SSE。SSE 心跳只维持连接并复用最近一次真实状态事件 ID，尚无状态事件时显式发送空 ID；浏览器重连通过 `Last-Event-ID` 或 `lastEventId` 重放有限窗口，游标失效时收到一次 `snapshot-required` 并重新读取 HTTP 快照。前端不直接订阅 MQTT，也不使用定时轮询。 TCP 通道使用独立 NATMap 开关，UDP 通道使用独立 Keeper 开关；TCP 不提供 STUN，UDP 只有 `externalPort === internalPort` 时可启用 Keeper。`NETWORK_TCP_NATMAP_RELEASE_MODE=off|draining|canary|on` 控制 TCP 变更范围，canary 端口来自 `NETWORK_TCP_NATMAP_CANARY_PORTS`。当前端点只有在 `currentValidUntil` 未过期时才返回为可用值；租约过期不会删除 candidate、last-observed、last-published 或 `network_endpoint_history`。API 不直接访问小米路由器、不修改 NAS 路由；真实小米规则、FriendlyWrt DNAT/NATMap、raw UDP 与回程规则由固定 Agent/helper/Companion 处理。 -自动 DDNS 支持 `A` 和 `AAAA`。A 记录绑定 `sourceType=port_forward_ipv4` 与一条合格的同源端口 UDP Keeper；AAAA 记录固定使用 `sourceType=agent_ipv6`，不得携带 `portForwardId`。来源暂不可用时记录进入 `waiting_source`，不会向腾讯云写入空地址。协调器只修改腾讯云云解析 DNS 中已存在、已启用、默认线路且唯一的同类型记录，写入时保留 RecordId、线路和 TTL，并回读相同 RecordId 验证结果；DNS 值不包含端口。删除接口只删除本地自动更新绑定，不删除云端记录。Provider 状态接口只返回开关、配置完整性和官方 provider 标识，不返回 SecretId、SecretKey 或 SDK 原始错误。 +自动 DDNS 支持 `A` 和 `AAAA`。A 记录绑定 `sourceType=port_forward_ipv4` 与一条合格的 UDP Keeper 或 TCP NATMap 协议通道；来源选项返回 `groupId/protocol/mechanism`，名称为 `组名 / UDP Keeper` 或 `组名 / TCP NATMap`。AAAA 记录固定使用 `sourceType=agent_ipv6`，不得携带 `portForwardId`。来源暂不可用时记录进入 `waiting_source`，不会向腾讯云写入空地址。协调器只修改腾讯云云解析 DNS 中已存在、已启用、默认线路且唯一的同类型记录，写入时保留 RecordId、线路和 TTL，并回读相同 RecordId 验证结果；DNS 值不包含端口。仅端口变化不调用 provider；当 `syncStatus=synced`、`appliedAddress` 等于当前通道 IPv4 且当前公网端口有效时，DDNS 响应额外返回 `accessEndpoint=FQDN:port`。删除接口只删除本地自动更新绑定，不删除云端记录。Provider 状态接口只返回开关、配置完整性和官方 provider 标识，不返回 SecretId、SecretKey 或 SDK 原始错误。 Agent 状态响应额外包含可选的 `currentPublicIpv6/currentIpv6ObservedAt`。只有在线且未超过 `NETWORK_DDNS_AGENT_IPV6_MAX_AGE_MS` 的规范化全局 IPv6 才能作为 AAAA 来源；缺少 IPv6 不影响现有端口转发、UDP Keeper 或 Agent 在线状态。 @@ -471,9 +471,11 @@ QQBot 运行态包括 NapCat 容器登录、OneBot v11 反向 WebSocket、MQTT 响应仅返回管理契约字段：source definition/field/variable 白名单；通用 `subscription-options` 按来源声明的集合返回统一 `value/label/disabled/disabledReasonCode/dependsOnValue` 候选项；旧 STUN options 入口在兼容期内继续返回旧版 Admin 所需的专用字段。subscription 的 `sourceConfig` 只回显当前来源公开声明且值为字符串的字段；template、preview、binding/target 和 target option 仍使用各自视图白名单。不会返回 adapter、entity/repository、`activeKey`、digest、软删除字段、账号内部 ID、事件 payload/delivery/lease/retry 状态、凭据、access token、Provider/OneBot/MQTT 原始对象。系统事件只能通过 Nest 内部 Outbox stager 暂存，不存在 publish、event、delivery、fan-out、retry 或 worker HTTP 发布接口。 +TCP NATMap 独立消息源为 `network.tcp.natmap-endpoint-changed`、版本 `1`，订阅配置精确为 `{ tcpChannelId, ddnsRecordId }`，两个字段都必须是字符串 Snowflake ID，且 DDNS 必须是启用、未删除、绑定同一 TCP 通道的 A 记录。变量精确为 `endpoint`、`fqdn`、`publicIpv4`、`publicPort`、`previousPublicIpv4`、`previousPublicPort`、`portForwardName`；`endpoint` 使用事件端口冻结为 `FQDN:newPort`。DDNS 尚未同步到事件 IPv4 时为 `waiting_ddns`，当前 tuple 已变化或租约过期为 `superseded`，通道撤下、删除、停用 NATMap 或 DDNS 改绑为 `cancelled`。现有 STUN 来源定义与变量保持不变。 + #### 内部事件与投递生命周期 -- 只有一个既有有效端口直接变化为另一个有效端口的 `changed` 事件，才会在 endpoint-history 同一事务内以生产者 `eventId` 幂等写入 `qqbot_message_event` Outbox。首次 `published`、租约续期、仅 IPv4 变化、`withdrawn`、`restored`、重复事件和回滚事务都不产生消息事件；事务提交后才调用 `requestDrain()`。 +- UDP 只有一个既有有效端口直接变化为另一个有效端口的 `changed` 事件，才写既有 STUN Outbox；TCP 则在既有有效 tuple 的公网 IPv4 或端口任一变化时写独立 NATMap endpoint Outbox。首次 `published`、同 tuple 续期、`withdrawn`、同 tuple `restored`、重复事件和回滚事务都不产生 TCP 消息事件。事件使用生产者 `eventId` 幂等，matching report/event 的 history、current/baseline 与 Outbox 保持既有事务关联门禁，事务提交后才调用 `requestDrain()`。 - Outbox 扇出状态为 `accepted`、`processing`、`retry`、`completed`、`failed`；投递状态为 `waiting_ddns`、`pending`、`processing`、`retry`、`success`、`failed`、`superseded`、`cancelled`。每个 runner 每次最多领取 50 行并设置 30 秒租约；启动后立即恢复且每 5 秒扫描。过期 `processing` 租约可由重启后的进程重新领取。 - DDNS 未就绪的投递进入 `waiting_ddns` 并每 60 秒复检。只有对应 A 记录的 optimistic `synced` / `appliedAddress` 更新成功持久化后，`notifyDdnsSynced()` 才提前推进相关任务；发送前仍重检当前 endpoint、DDNS、订阅、绑定、目标和账号。更新 endpoint 使旧未完成任务成为 `superseded`，配置停用或删除使其成为 `cancelled`。 - 扇出和投递的临时错误从 10 秒开始指数退避，单次最长 15 分钟，并在事件发生 24 小时后截止。`sendStrictPlainText()` 只选择投递冻结的 `selfId`，把正文作为一个 OneBot `text` segment 发送，并在成功时关联 `qqbot_send_log`。数据库唯一键保证事件和事件-目标任务幂等；OneBot 超时可重试，因此外部结果不明确时仍是可能重复的至少一次投递，而不是跨系统恰好一次。 diff --git a/README.md b/README.md index 463f25e..1e7f80f 100644 --- a/README.md +++ b/README.md @@ -84,11 +84,11 @@ System 网络管理以 MySQL 中的逻辑端口转发组和 TCP/UDP 协议通道 MQTT v2 分别保存 candidate、current、last-observed、last-published、通道 report 原始 RFC3339Nano 水位，以及静态路径、Keeper 和 NATMap 错误。TCP current 只有在 desired present、NATMap intent、sync、Router、DNAT、NATMap 进程、generation、同 tuple 证据和新鲜 lease 全部通过时才发布；相同 tuple 续租只推进 validation/lease，不新增历史、不触发 DDNS、Outbox 或 Admin 刷新。同 revision 的旧 report 不得覆盖较新的通道状态；合法外层 snapshot 完成协调尝试后推进 applied revision 与独立 applied schema 水位，单通道 conflict 只隔离该通道。Agent 接受 v2 后会耐久锁存唯一 owner，在未来实现 Agent 可确认的降级握手前，API 单边降级固定拒绝。 -v2 endpoint event 通常先于 matching reported 到达，因此 event-first 只提交 history；UDP 端口变更必须等 matching current 在 reported 事务中提交后才写现有 STUN Outbox，TCP 在独立消息源上线前只写 `tcp_natmap` history。event、report 与 ownership 切换统一按 Agent state、channel、history 顺序加锁；history 耐久保存来源 revision 和完整 lease identity，只有 revision、tuple 与未被时间精度截断的 lease 身份全部匹配才允许 stage，避免迟到事件误投递。所有 SSE、DDNS reconcile、投递 wake 和 desired republish 都在数据库事务成功后最多触发一次；组只有所有协议通道都确认 exact absent 后才软删除。Agent 失联或 MQTT 暂不可用时合法请求仍保存为 pending，恢复后按 revision 自动收敛；瞬时数据库错误或 SUBACK 失败依赖持久会话重投，非法负载确认后丢弃。当前公网端点受 `currentValidUntil` 约束，过期后列表隐藏 current，但保留候选、最近观测、发布基线与历史。 +v2 endpoint event 通常先于 matching reported 到达，因此 event-first 先提交 history；UDP 有效端口变化写入既有 STUN Outbox，TCP 公网 IPv4 或端口任一变化写入独立的 `network.tcp.natmap-endpoint-changed` Outbox，二者都必须等 matching current 通过后才能 stage。event、report 与 ownership 切换统一按 Agent state、channel、history 顺序加锁；history 耐久保存来源 revision 和完整 lease identity，只有 revision、tuple 与未被时间精度截断的 lease 身份全部匹配才允许 stage，避免迟到事件误投递。首次发布、同 tuple 续租、撤下和同 tuple 恢复不会产生 TCP 消息事件。所有 SSE、DDNS reconcile、投递 wake 和 desired republish 都在数据库事务成功后最多触发一次；组只有所有协议通道都确认 exact absent 后才软删除。Agent 失联或 MQTT 暂不可用时合法请求仍保存为 pending，恢复后按 revision 自动收敛；瞬时数据库错误或 SUBACK 失败依赖持久会话重投，非法负载确认后丢弃。当前公网端点受 `currentValidUntil` 约束，过期后列表隐藏 current，但保留候选、最近观测、发布基线与历史。 -同一模块提供腾讯云云解析 DNS 的双栈自动 DDNS。A 记录只从合格 UDP Keeper 的有效公网 IPv4 取值，AAAA 记录只从在线 Agent 最近上报的全局 IPv6 取值；DNS 值始终不包含端口。协调器只修改已存在、已启用、默认线路且唯一的 A/AAAA 记录，保留 RecordId、线路和 TTL，并在写入后回读确认。删除 Admin 绑定只停止本地自动更新，不删除云端 DNS 记录。凭据只从 API 私有运行环境的 `NETWORK_DDNS_DNSPOD_SECRET_ID/SECRET_KEY` 读取，不进入 Admin、数据库、MQTT、Agent、日志或 Git；`DNSPOD` 是腾讯云官方 SDK 的技术服务名。 +同一模块提供腾讯云云解析 DNS 的双栈自动 DDNS。A 记录可从合格 UDP Keeper 或 TCP NATMap 通道的有效公网 IPv4 取值，AAAA 记录只从在线 Agent 最近上报的全局 IPv6 取值；DNS 值始终不包含端口。A 来源以 `组名 / UDP Keeper` 或 `组名 / TCP NATMap` 展示；同步地址等于当前通道 IPv4 且当前公网端口有效时，管理响应额外返回 `accessEndpoint=FQDN:port`。仅端口变化只更新该派生端点，不调用腾讯云。协调器只修改已存在、已启用、默认线路且唯一的 A/AAAA 记录，保留 RecordId、线路和 TTL，并在写入后回读确认。删除 Admin 绑定只停止本地自动更新，不删除云端 DNS 记录。凭据只从 API 私有运行环境的 `NETWORK_DDNS_DNSPOD_SECRET_ID/SECRET_KEY` 读取，不进入 Admin、数据库、MQTT、Agent、日志或 Git；`DNSPOD` 是腾讯云官方 SDK 的技术服务名。 -QQBot 系统消息推送由全局消息订阅和模板、账号范围发布绑定及耐久事件/投递共同管理；订阅表单字段、候选集合和资源匹配由所选系统消息源 adapter 声明，核心模块不写死 STUN 的 `portForwardId/ddnsRecordId`。账号接口严格使用路由 `selfId`，不会回退到其他机器人。群聊和私聊目标分别使用 `group` / `private`，Snowflake、QQ 账号和目标 ID 在 HTTP 与数据库边界始终保持字符串。系统事件只能通过内部 Outbox stager 暂存，不提供 publish/event/worker HTTP 路由；管理响应仅返回字段白名单，不暴露账号凭据、Provider/OneBot/MQTT 运行对象、原始事件载荷或内部持久化键。 +QQBot 系统消息推送由全局消息订阅和模板、账号范围发布绑定及耐久事件/投递共同管理；订阅表单字段、候选集合和资源匹配由所选系统消息源 adapter 声明，核心模块不写死 STUN 的 `portForwardId/ddnsRecordId` 或 TCP NATMap 的 `tcpChannelId/ddnsRecordId`。TCP 来源变量精确为 `endpoint/fqdn/publicIpv4/publicPort/previousPublicIpv4/previousPublicPort/portForwardName`，只有所选同通道 A 记录同步到事件 IPv4 后才能发送。账号接口严格使用路由 `selfId`，不会回退到其他机器人。群聊和私聊目标分别使用 `group` / `private`，Snowflake、QQ 账号和目标 ID 在 HTTP 与数据库边界始终保持字符串。系统事件只能通过内部 Outbox stager 暂存，不提供 publish/event/worker HTTP 路由；管理响应仅返回字段白名单，不暴露账号凭据、Provider/OneBot/MQTT 运行对象、原始事件载荷或内部持久化键。 `qqbot_message_event` 同时承担事件 Outbox：扇出状态为 `accepted`、`processing`、`retry`、`completed`、`failed`；按目标冻结的 `qqbot_message_delivery` 状态为 `waiting_ddns`、`pending`、`processing`、`retry`、`success`、`failed`、`superseded`、`cancelled`。扇出和投递每次各领取最多 50 行，处理租约为 30 秒；进程启动后立即恢复，并每 5 秒扫描一次。`waiting_ddns` 每 60 秒持久化复检；临时错误从 10 秒开始指数退避，单次最长 15 分钟，并以事件发生后 24 小时为截止时间。 diff --git a/src/modules/admin/platform-config/admin-platform-config.module.ts b/src/modules/admin/platform-config/admin-platform-config.module.ts index 1112212..fd2f77c 100644 --- a/src/modules/admin/platform-config/admin-platform-config.module.ts +++ b/src/modules/admin/platform-config/admin-platform-config.module.ts @@ -22,6 +22,7 @@ import { NetworkPortForwardGroupController } from '@/modules/admin/platform-conf import { NetworkPortForwardGroupService } from '@/modules/admin/platform-config/network-management/network-port-forward-group.service'; import { NetworkManagementService } from '@/modules/admin/platform-config/network-management/network-management.service'; import { NetworkStunMessageSourceAdapter } from '@/modules/admin/platform-config/network-management/network-stun-message-source.adapter'; +import { NetworkTcpNatmapMessageSourceAdapter } from '@/modules/admin/platform-config/network-management/network-tcp-natmap-message-source.adapter'; import { NetworkTcpReleasePolicyService } from '@/modules/admin/platform-config/network-management/network-tcp-release-policy.service'; import { SystemLogController } from '@/modules/admin/platform-config/system-log/system-log.controller'; import { SystemLogService } from '@/modules/admin/platform-config/system-log/system-log.service'; @@ -96,6 +97,7 @@ export const ADMIN_PLATFORM_CONFIG_PROVIDERS = [ NetworkDnsPodClient, NetworkDdnsService, NetworkStunMessageSourceAdapter, + NetworkTcpNatmapMessageSourceAdapter, NetworkTcpReleasePolicyService, NetworkAgentMqttService, ]; diff --git a/src/modules/admin/platform-config/network-management/network-agent-mqtt.service.ts b/src/modules/admin/platform-config/network-management/network-agent-mqtt.service.ts index 57eefc6..f849e26 100644 --- a/src/modules/admin/platform-config/network-management/network-agent-mqtt.service.ts +++ b/src/modules/admin/platform-config/network-management/network-agent-mqtt.service.ts @@ -1,3 +1,4 @@ +import { isIP } from 'node:net'; import { Inject, Injectable, @@ -788,14 +789,12 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { }); } - if ( - item.protocol === 'udp' && - currentTupleBefore !== currentTupleAfter && - currentTupleAfter - ) { - deliveryAccepted = - (await this.stageMatchingV2UdpHistory(manager, mapping)) || - deliveryAccepted; + if (currentTupleBefore !== currentTupleAfter && currentTupleAfter) { + const accepted = + item.protocol === 'tcp' + ? await this.stageMatchingV2TcpHistory(manager, mapping) + : await this.stageMatchingV2UdpHistory(manager, mapping); + deliveryAccepted = accepted || deliveryAccepted; } } } @@ -1190,6 +1189,23 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { ); } + private async stageMatchingV2TcpHistory( + manager: EntityManager, + mapping: NetworkPortForward, + ): Promise { + const histories = await manager.getRepository(NetworkEndpointHistory).find({ + order: { occurredAt: 'DESC', id: 'DESC' }, + take: 2, + where: { mappingId: mapping.id, mechanism: 'tcp_natmap' }, + }); + return await this.stageV2TcpEndpointChange( + manager, + mapping, + histories[0], + histories[1], + ); + } + private async stageV2UdpPortChange( manager: EntityManager, mapping: NetworkPortForward, @@ -1224,6 +1240,44 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { ); } + private async stageV2TcpEndpointChange( + manager: EntityManager, + mapping: NetworkPortForward, + currentHistory: NetworkEndpointHistory | undefined, + previousHistory: NetworkEndpointHistory | undefined, + ): Promise { + if ( + !currentHistory || + currentHistory.eventType !== 'changed' || + !previousHistory || + previousHistory.eventType === 'withdrawn' || + !this.isV2HistoryMatchingCurrent(mapping, currentHistory) || + isIP(previousHistory.publicIpv4 || '') !== 4 || + isIP(currentHistory.publicIpv4 || '') !== 4 || + !this.isValidPort(previousHistory.publicPort) || + !this.isValidPort(currentHistory.publicPort) || + (previousHistory.publicIpv4 === currentHistory.publicIpv4 && + previousHistory.publicPort === currentHistory.publicPort) + ) { + return false; + } + return ( + (await this.eventStager.stage(manager, { + eventId: currentHistory.eventId, + occurredAt: new Date(currentHistory.occurredAt).toISOString(), + payload: { + previousPublicIpv4: previousHistory.publicIpv4 as string, + previousPublicPort: previousHistory.publicPort, + publicIpv4: currentHistory.publicIpv4 as string, + publicPort: currentHistory.publicPort, + tcpChannelId: mapping.id, + }, + resourceKey: mapping.id, + sourceKey: 'network.tcp.natmap-endpoint-changed', + })) === 'accepted' + ); + } + private isV2HistoryMatchingCurrent( mapping: NetworkPortForward, history: NetworkEndpointHistory, @@ -1545,16 +1599,23 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { }); try { await repository.save(history); - const deliveryAccepted = this.canStageV2UdpEventAgainstCurrent( + const deliveryAccepted = this.canStageV2EventAgainstCurrent( mapping, event, ) - ? await this.stageV2UdpPortChange( - manager, - mapping, - history, - previousHistory || undefined, - ) + ? event.protocol === 'tcp' + ? await this.stageV2TcpEndpointChange( + manager, + mapping, + history, + previousHistory || undefined, + ) + : await this.stageV2UdpPortChange( + manager, + mapping, + history, + previousHistory || undefined, + ) : false; return { changed: true, deliveryAccepted }; } catch (error) { @@ -1564,13 +1625,14 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { }); } - private canStageV2UdpEventAgainstCurrent( + private canStageV2EventAgainstCurrent( mapping: NetworkPortForward, event: NetworkEndpointEventV2, ): boolean { + const expectedMechanism = + event.protocol === 'tcp' ? 'tcp_natmap' : 'udp_stun'; return ( - event.protocol === 'udp' && - event.mechanism === 'udp_stun' && + event.mechanism === expectedMechanism && event.type === 'changed' && !!event.endpoint && mapping.reportedRevision === String(event.revision) && diff --git a/src/modules/admin/platform-config/network-management/network-ddns.service.ts b/src/modules/admin/platform-config/network-management/network-ddns.service.ts index 26ef4c5..4b6deda 100644 --- a/src/modules/admin/platform-config/network-management/network-ddns.service.ts +++ b/src/modules/admin/platform-config/network-management/network-ddns.service.ts @@ -23,7 +23,9 @@ import { } from './network-dnspod.client'; import { NetworkManagementEventStreamService } from './network-management-event-stream.service'; import { NetworkPortForward } from './network-management.entity'; +import { NetworkPortForwardGroup } from './network-port-forward-group.entity'; import { classifyStunEndpointSource } from './network-source-eligibility'; +import { classifyTcpNatmapEndpointSource } from './network-tcp-natmap-source-eligibility'; import type { NetworkDdnsListQuery, NetworkDdnsRecordInput, @@ -85,6 +87,8 @@ export class NetworkDdnsService implements OnModuleInit, OnModuleDestroy { private readonly recordRepository: Repository, @InjectRepository(NetworkPortForward) private readonly mappingRepository: Repository, + @InjectRepository(NetworkPortForwardGroup) + private readonly groupRepository: Repository, @InjectRepository(NetworkAgentState) private readonly stateRepository: Repository, private readonly configService: ConfigService, @@ -166,11 +170,25 @@ export class NetworkDdnsService implements OnModuleInit, OnModuleDestroy { if (query.recordType !== 'A') { throwVbenError('DDNS 记录类型无效', HttpStatus.BAD_REQUEST); } - const mappings = await this.mappingRepository.find({ - order: { id: 'ASC', name: 'ASC' }, - where: { isDeleted: false }, - }); - return mappings.map((mapping) => this.portForwardSourceOption(mapping)); + const [mappings, groups] = await Promise.all([ + this.mappingRepository.find({ + order: { id: 'ASC', name: 'ASC' }, + where: { isDeleted: false }, + }), + this.groupRepository.find({ + order: { id: 'ASC', name: 'ASC' }, + where: { isDeleted: false }, + }), + ]); + const groupsById = new Map( + groups.map((group) => [String(group.id), group]), + ); + return mappings.map((mapping) => + this.portForwardSourceOption( + mapping, + groupsById.get(String(mapping.groupId)), + ), + ); } getProviderStatus() { @@ -638,7 +656,12 @@ export class NetworkDdnsService implements OnModuleInit, OnModuleDestroy { const mapping = await this.mappingRepository.findOne({ where: { id: record.portForwardId, isDeleted: false }, }); - if (mapping) return this.portForwardSourceOption(mapping); + if (mapping) { + const group = await this.groupRepository.findOne({ + where: { id: mapping.groupId, isDeleted: false }, + }); + return this.portForwardSourceOption(mapping, group || undefined); + } return this.missingPortForwardSourceOption(record.portForwardId); } return this.missingPortForwardSourceOption( @@ -648,21 +671,37 @@ export class NetworkDdnsService implements OnModuleInit, OnModuleDestroy { private portForwardSourceOption( mapping: NetworkPortForward, + group?: NetworkPortForwardGroup, ): NetworkDdnsSourceOption { - const { disabledReasonCode, eligible } = - classifyStunEndpointSource(mapping); + const mechanism = mapping.protocol === 'tcp' ? 'tcp_natmap' : 'udp_stun'; + const sourceEligibility = + mapping.protocol === 'tcp' + ? classifyTcpNatmapEndpointSource(mapping) + : classifyStunEndpointSource(mapping); + const disabledReasonCode = group + ? sourceEligibility.disabledReasonCode + : 'SOURCE_NOT_FOUND'; + const eligible = disabledReasonCode === null; const leaseValid = isIP(mapping.currentPublicIpv4 || '') === 4 && + isValidPort(mapping.currentPublicPort) && !!mapping.currentValidUntil && new Date(mapping.currentValidUntil).getTime() > Date.now(); const sourceUsable = eligible && leaseValid; return { currentAddress: sourceUsable ? mapping.currentPublicIpv4 || null : null, + ...(sourceUsable + ? { currentPort: mapping.currentPublicPort as number } + : {}), disabledReasonCode, - eligible: disabledReasonCode === null, + eligible, externalPort: mapping.externalPort, + groupId: String(mapping.groupId), id: String(mapping.id), - name: mapping.name, + mechanism, + name: `${group?.name || mapping.name} / ${ + mechanism === 'tcp_natmap' ? 'TCP NATMap' : 'UDP Keeper' + }`, observedAt: sourceUsable ? mapping.currentObservedAt || null : null, protocol: mapping.protocol, sourceType: 'port_forward_ipv4', @@ -844,9 +883,20 @@ export class NetworkDdnsService implements OnModuleInit, OnModuleDestroy { const mapping = await this.mappingRepository.findOne({ where: { id: portForwardId as string, isDeleted: false }, }); - if (!mapping || !classifyStunEndpointSource(mapping).eligible) { + const group = mapping + ? await this.groupRepository.findOne({ + where: { id: mapping.groupId, isDeleted: false }, + }) + : null; + const sourceEligible = + mapping?.protocol === 'tcp' + ? classifyTcpNatmapEndpointSource(mapping).eligible + : mapping + ? classifyStunEndpointSource(mapping).eligible + : false; + if (!mapping || !group || !sourceEligible) { throwVbenError( - 'A 记录来源必须是已启用 Keeper 的同端口 UDP 转发', + 'A 记录来源必须是已启用的 UDP Keeper 或 TCP NATMap 通道', HttpStatus.BAD_REQUEST, ); } @@ -890,14 +940,24 @@ export class NetworkDdnsService implements OnModuleInit, OnModuleDestroy { private async serializeRecord(record: NetworkDdnsRecord) { const source = await this.resolveRecordSource(record); + const fqdn = + record.subDomain === '@' + ? record.domain + : `${record.subDomain}.${record.domain}`; + const accessEndpoint = + record.recordType === 'A' && + record.syncStatus === 'synced' && + !!record.appliedAddress && + record.appliedAddress === source.currentAddress && + isValidPort(source.currentPort) + ? `${fqdn}:${source.currentPort}` + : null; return { + ...(accessEndpoint ? { accessEndpoint } : {}), appliedAddress: record.appliedAddress || null, domain: record.domain, enabled: record.enabled, - fqdn: - record.subDomain === '@' - ? record.domain - : `${record.subDomain}.${record.domain}`, + fqdn, id: String(record.id), lastErrorCode: record.lastErrorCode || null, lastErrorMessage: record.lastErrorMessage || null, @@ -1123,6 +1183,15 @@ function isValidDnsName( ); } +function isValidPort(value?: null | number): value is number { + return ( + typeof value === 'number' && + Number.isInteger(value) && + value >= 1 && + value <= 65_535 + ); +} + function normalizeGlobalIpv6(value?: null | string): null | string { if (!value || isIP(value) !== 6) return null; let normalized: string; diff --git a/src/modules/admin/platform-config/network-management/network-management.types.ts b/src/modules/admin/platform-config/network-management/network-management.types.ts index 6082662..8663b1c 100644 --- a/src/modules/admin/platform-config/network-management/network-management.types.ts +++ b/src/modules/admin/platform-config/network-management/network-management.types.ts @@ -77,10 +77,13 @@ export type NetworkDdnsListQuery = { export type NetworkDdnsSourceOption = { currentAddress: null | string; + currentPort?: number; disabledReasonCode: null | string; eligible: boolean; externalPort?: number; + groupId?: string; id: string; + mechanism?: EndpointMechanism; name: string; observedAt: null | KtDateTime; protocol?: PortForwardProtocol; diff --git a/src/modules/admin/platform-config/network-management/network-tcp-natmap-message-source.adapter.ts b/src/modules/admin/platform-config/network-management/network-tcp-natmap-message-source.adapter.ts new file mode 100644 index 0000000..cbcde7b --- /dev/null +++ b/src/modules/admin/platform-config/network-management/network-tcp-natmap-message-source.adapter.ts @@ -0,0 +1,511 @@ +import { isIP } from 'node:net'; +import { + Injectable, + type OnModuleDestroy, + type OnModuleInit, +} from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { SystemMessageSourceRegistry } from '@/modules/qqbot/core/application/message-push/system-message-source.registry'; +import { + SystemMessageContractError, + type SystemMessageDeliveryReadiness, + type SystemMessageScalar, + type SystemMessageSourceAdapter, + type SystemMessageSourceDefinition, + type SystemMessageSourceOptionsResponse, +} from '@/modules/qqbot/core/contract/message-push/qqbot-message-push.types'; +import { NetworkDdnsRecord } from './network-ddns.entity'; +import { NetworkPortForward } from './network-management.entity'; +import { NetworkPortForwardGroup } from './network-port-forward-group.entity'; +import { + classifyTcpNatmapEndpointSource, + type TcpNatmapEndpointSourceDisabledReason, +} from './network-tcp-natmap-source-eligibility'; + +const SOURCE_KEY = 'network.tcp.natmap-endpoint-changed'; +const SNOWFLAKE_ID_PATTERN = /^[1-9]\d{0,23}$/; +const SUBSCRIPTION_KEYS = ['ddnsRecordId', 'tcpChannelId']; +const EVENT_KEYS = [ + 'previousPublicIpv4', + 'previousPublicPort', + 'publicIpv4', + 'publicPort', + 'tcpChannelId', +]; + +type TcpNatmapSubscriptionConfig = { + ddnsRecordId: string; + tcpChannelId: string; +}; + +type TcpNatmapEventPayload = { + previousPublicIpv4: string; + previousPublicPort: number; + publicIpv4: string; + publicPort: number; + tcpChannelId: string; +}; + +type ResolvedSubscription = { + config: TcpNatmapSubscriptionConfig; + ddnsRecord: NetworkDdnsRecord; + group: NetworkPortForwardGroup; + mapping: NetworkPortForward; + sourceSummary: string; +}; + +@Injectable() +export class NetworkTcpNatmapMessageSourceAdapter + implements OnModuleDestroy, OnModuleInit, SystemMessageSourceAdapter +{ + private registered = false; + + readonly definition: SystemMessageSourceDefinition = { + description: '当 TCP NATMap 公网端点变更且 IPv4 DDNS 已同步时发送消息。', + displayName: 'TCP NATMap 端点变更', + sourceKey: SOURCE_KEY, + subscriptionFields: [ + { + key: 'tcpChannelId', + label: 'TCP NATMap 通道', + optionCollection: 'tcpChannels', + required: true, + type: 'select', + }, + { + dependsOn: 'tcpChannelId', + key: 'ddnsRecordId', + label: 'IPv4 DDNS 记录', + optionCollection: 'ddnsRecords', + required: true, + type: 'select', + }, + ], + variables: [ + { + description: '服务端组合的域名与公网端口', + example: 'pal.kwitsukasa.top:38213', + key: 'endpoint', + label: '访问端点', + type: 'string', + }, + { + description: '所选 DDNS 完整域名', + example: 'pal.kwitsukasa.top', + key: 'fqdn', + label: '域名', + type: 'string', + }, + { + description: '新的公网 IPv4', + example: '203.0.113.10', + key: 'publicIpv4', + label: '公网 IPv4', + type: 'string', + }, + { + description: '新的公网端口', + example: '38213', + key: 'publicPort', + label: '公网端口', + type: 'number', + }, + { + description: '变化前的公网 IPv4', + example: '198.51.100.9', + key: 'previousPublicIpv4', + label: '原公网 IPv4', + type: 'string', + }, + { + description: '变化前的公网端口', + example: '38111', + key: 'previousPublicPort', + label: '原公网端口', + type: 'number', + }, + { + description: '逻辑端口转发名称', + example: '帕鲁新世界', + key: 'portForwardName', + label: '端口转发名称', + type: 'string', + }, + ], + version: 1, + }; + + constructor( + @InjectRepository(NetworkPortForward) + private readonly mappingRepository: Repository, + @InjectRepository(NetworkPortForwardGroup) + private readonly groupRepository: Repository, + @InjectRepository(NetworkDdnsRecord) + private readonly ddnsRepository: Repository, + private readonly sourceRegistry: SystemMessageSourceRegistry, + ) {} + + onModuleInit(): void { + if (this.registered) return; + this.sourceRegistry.register(this); + this.registered = true; + } + + onModuleDestroy(): void { + if (!this.registered) return; + this.sourceRegistry.unregister(this.definition.sourceKey, this); + this.registered = false; + } + + eventResourceKey(payload: Record): string { + if (typeof payload.tcpChannelId !== 'string') { + throw new SystemMessageContractError('invalid_source_config'); + } + return payload.tcpChannelId; + } + + async normalizeSubscriptionConfig(input: unknown): Promise<{ + canonicalConfig: Record; + resourceKey: string; + sourceSummary: string; + }> { + const resolved = await this.resolveSubscription(input); + return { + canonicalConfig: resolved.config, + resourceKey: resolved.config.tcpChannelId, + sourceSummary: resolved.sourceSummary, + }; + } + + async inspectSubscription(config: Record): Promise<{ + invalidReasonCode: null | string; + sourceSummary: string; + valid: boolean; + }> { + try { + const resolved = await this.resolveSubscription(config); + return { + invalidReasonCode: null, + sourceSummary: resolved.sourceSummary, + valid: true, + }; + } catch (error) { + if (!(error instanceof SystemMessageContractError)) throw error; + return { + invalidReasonCode: error.code, + sourceSummary: '未选择有效的 TCP NATMap 通道与 DDNS', + valid: false, + }; + } + } + + async listSubscriptionOptions(): Promise { + const [mappings, groups, records] = await Promise.all([ + this.mappingRepository.find({ order: { id: 'ASC', name: 'ASC' } }), + this.groupRepository.find({ order: { id: 'ASC', name: 'ASC' } }), + this.ddnsRepository.find({ order: { id: 'ASC', name: 'ASC' } }), + ]); + const mappingsById = new Map( + mappings.map((mapping) => [String(mapping.id), mapping]), + ); + const groupsById = new Map( + groups.map((group) => [String(group.id), group]), + ); + return { + ddnsRecords: records.map((record) => { + const mapping = record.portForwardId + ? mappingsById.get(String(record.portForwardId)) + : undefined; + const group = mapping + ? groupsById.get(String(mapping.groupId)) + : undefined; + const disabledReasonCode = ddnsOptionReason(record, mapping, group); + return { + ...(record.portForwardId + ? { dependsOnValue: String(record.portForwardId) } + : {}), + disabled: disabledReasonCode !== null, + disabledReasonCode, + label: [record.name, ddnsFqdn(record), disabledReasonCode] + .filter((value) => value !== null) + .join(' · '), + value: String(record.id), + }; + }), + tcpChannels: mappings.map((mapping) => { + const group = groupsById.get(String(mapping.groupId)); + const source = classifyTcpNatmapEndpointSource(mapping); + const groupMissing = !group || group.isDeleted; + const disabledReasonCode = groupMissing + ? 'SOURCE_DELETING' + : source.disabledReasonCode; + return { + disabled: disabledReasonCode !== null, + disabledReasonCode, + label: `${group?.name || mapping.name} / ${mapping.protocol === 'tcp' ? 'TCP NATMap' : 'UDP Keeper'}`, + value: String(mapping.id), + }; + }), + }; + } + + subscriptionResourceKey(config: Record): null | string { + if (!isPlainRecord(config) || typeof config.tcpChannelId !== 'string') { + return null; + } + return config.tcpChannelId; + } + + validateEventPayload( + payload: Record, + ): Record { + assertExactKeys(payload, EVENT_KEYS); + const tcpChannelId = normalizeSnowflakeId(payload.tcpChannelId); + const publicIpv4 = normalizeIpv4(payload.publicIpv4); + const publicPort = normalizePort(payload.publicPort); + const previousPublicIpv4 = normalizeIpv4(payload.previousPublicIpv4); + const previousPublicPort = normalizePort(payload.previousPublicPort); + if ( + publicIpv4 === previousPublicIpv4 && + publicPort === previousPublicPort + ) { + throw new SystemMessageContractError('invalid_source_config'); + } + return { + previousPublicIpv4, + previousPublicPort, + publicIpv4, + publicPort, + tcpChannelId, + }; + } + + async resolveDelivery(input: { + eventPayload: Record; + subscriptionConfig: Record; + }): Promise { + let resolved: ResolvedSubscription; + try { + resolved = await this.resolveSubscription(input.subscriptionConfig); + } catch (error) { + if (!(error instanceof SystemMessageContractError)) throw error; + return { reasonCode: error.code, status: 'cancelled' }; + } + let event: TcpNatmapEventPayload; + try { + event = this.validateEventPayload( + input.eventPayload, + ) as TcpNatmapEventPayload; + } catch (error) { + if (!(error instanceof SystemMessageContractError)) throw error; + return { reasonCode: error.code, status: 'cancelled' }; + } + if (event.tcpChannelId !== resolved.config.tcpChannelId) { + return { reasonCode: 'invalid_source_config', status: 'cancelled' }; + } + if (!hasCurrentEndpointTuple(resolved.mapping)) { + return { reasonCode: 'endpoint_withdrawn', status: 'cancelled' }; + } + if ( + !hasFreshCurrentEndpoint(resolved.mapping) || + resolved.mapping.currentPublicIpv4 !== event.publicIpv4 || + resolved.mapping.currentPublicPort !== event.publicPort + ) { + return { reasonCode: 'endpoint_superseded', status: 'superseded' }; + } + const variables = deliveryVariables(resolved, event); + if ( + resolved.ddnsRecord.syncStatus !== 'synced' || + resolved.ddnsRecord.appliedAddress !== event.publicIpv4 + ) { + return { + reasonCode: 'ddns_not_synced', + status: 'waiting_ddns', + variables, + }; + } + return { reasonCode: null, status: 'ready', variables }; + } + + private async resolveSubscription( + input: unknown, + ): Promise { + assertExactKeys(input, SUBSCRIPTION_KEYS); + const config = { + ddnsRecordId: normalizeSnowflakeId(input.ddnsRecordId), + tcpChannelId: normalizeSnowflakeId(input.tcpChannelId), + }; + const mapping = await this.mappingRepository.findOne({ + where: { id: config.tcpChannelId }, + }); + if (!mapping) { + throw new SystemMessageContractError('mapping_not_found'); + } + const source = classifyTcpNatmapEndpointSource(mapping); + if (source.disabledReasonCode) { + throw new SystemMessageContractError( + messageSourceEligibilityReason(source.disabledReasonCode), + ); + } + const group = await this.groupRepository.findOne({ + where: { id: mapping.groupId }, + }); + if (!group || group.isDeleted) { + throw new SystemMessageContractError('mapping_not_managed'); + } + const ddnsRecord = await this.ddnsRepository.findOne({ + where: { id: config.ddnsRecordId }, + }); + const ddnsReason = ddnsMessageSourceReason(ddnsRecord, mapping); + if (ddnsReason) { + throw new SystemMessageContractError(ddnsReason); + } + const resolvedDdns = ddnsRecord as NetworkDdnsRecord; + return { + config, + ddnsRecord: resolvedDdns, + group, + mapping, + sourceSummary: `${group.name} / TCP NATMap · ${ddnsFqdn(resolvedDdns)}`, + }; + } +} + +function assertExactKeys( + value: unknown, + expectedKeys: readonly string[], +): asserts value is Record { + if ( + !isPlainRecord(value) || + value === null || + Object.keys(value).sort().join('\0') !== [...expectedKeys].sort().join('\0') + ) { + throw new SystemMessageContractError('invalid_source_config'); + } +} + +function isPlainRecord(value: unknown): value is Record { + return !!value && typeof value === 'object' && !Array.isArray(value); +} + +function normalizeSnowflakeId(value: unknown): string { + if (typeof value !== 'string' || !SNOWFLAKE_ID_PATTERN.test(value)) { + throw new SystemMessageContractError('invalid_source_config'); + } + return value; +} + +function normalizeIpv4(value: unknown): string { + if (typeof value !== 'string' || isIP(value) !== 4) { + throw new SystemMessageContractError('invalid_source_config'); + } + return value; +} + +function normalizePort(value: unknown): number { + if ( + typeof value !== 'number' || + !Number.isInteger(value) || + value < 1 || + value > 65_535 + ) { + throw new SystemMessageContractError('invalid_source_config'); + } + return value; +} + +function messageSourceEligibilityReason( + reason: TcpNatmapEndpointSourceDisabledReason, +): 'mapping_not_managed' | 'mapping_not_tcp' | 'natmap_disabled' { + switch (reason) { + case 'NATMAP_DISABLED': + return 'natmap_disabled'; + case 'SOURCE_DELETING': + return 'mapping_not_managed'; + case 'TCP_REQUIRED': + return 'mapping_not_tcp'; + } +} + +function ddnsOptionReason( + record: NetworkDdnsRecord, + mapping: NetworkPortForward | undefined, + group: NetworkPortForwardGroup | undefined, +): null | string { + if (record.isDeleted) return 'ddns_deleted'; + if (!record.enabled) return 'ddns_disabled'; + if (record.recordType !== 'A') return 'ddns_a_required'; + if (record.sourceType !== 'port_forward_ipv4') { + return 'ddns_source_type_invalid'; + } + if ( + !mapping || + String(record.portForwardId) !== String(mapping.id) || + !group || + group.isDeleted + ) { + return 'ddns_mapping_mismatch'; + } + return classifyTcpNatmapEndpointSource(mapping).disabledReasonCode; +} + +function ddnsMessageSourceReason( + record: NetworkDdnsRecord | null, + mapping: NetworkPortForward, +): + | 'ddns_disabled' + | 'ddns_mapping_mismatch' + | 'ddns_not_found' + | 'ddns_not_ipv4' + | null { + if (!record || record.isDeleted) return 'ddns_not_found'; + if (!record.enabled) return 'ddns_disabled'; + if (record.recordType !== 'A' || record.sourceType !== 'port_forward_ipv4') { + return 'ddns_not_ipv4'; + } + if (String(record.portForwardId) !== String(mapping.id)) { + return 'ddns_mapping_mismatch'; + } + return null; +} + +function ddnsFqdn(record: NetworkDdnsRecord): string { + const domain = record.domain.trim().toLowerCase().replace(/\.$/, ''); + const subDomain = record.subDomain.trim().toLowerCase().replace(/\.$/, ''); + return subDomain === '@' ? domain : `${subDomain}.${domain}`; +} + +function hasCurrentEndpointTuple(mapping: NetworkPortForward): boolean { + return ( + isIP(mapping.currentPublicIpv4 || '') === 4 && + typeof mapping.currentPublicPort === 'number' && + Number.isInteger(mapping.currentPublicPort) && + mapping.currentPublicPort >= 1 && + mapping.currentPublicPort <= 65_535 + ); +} + +function hasFreshCurrentEndpoint(mapping: NetworkPortForward): boolean { + return ( + hasCurrentEndpointTuple(mapping) && + !!mapping.currentValidUntil && + new Date(mapping.currentValidUntil).getTime() > Date.now() + ); +} + +function deliveryVariables( + resolved: ResolvedSubscription, + event: TcpNatmapEventPayload, +): Record { + const fqdn = ddnsFqdn(resolved.ddnsRecord); + return { + endpoint: `${fqdn}:${event.publicPort}`, + fqdn, + portForwardName: resolved.group.name, + previousPublicIpv4: event.previousPublicIpv4, + previousPublicPort: event.previousPublicPort, + publicIpv4: event.publicIpv4, + publicPort: event.publicPort, + }; +} diff --git a/src/modules/admin/platform-config/network-management/network-tcp-natmap-source-eligibility.ts b/src/modules/admin/platform-config/network-management/network-tcp-natmap-source-eligibility.ts new file mode 100644 index 0000000..5e18041 --- /dev/null +++ b/src/modules/admin/platform-config/network-management/network-tcp-natmap-source-eligibility.ts @@ -0,0 +1,31 @@ +import type { NetworkPortForward } from './network-management.entity'; + +export type TcpNatmapEndpointSourceDisabledReason = + | 'NATMAP_DISABLED' + | 'SOURCE_DELETING' + | 'TCP_REQUIRED'; + +export type TcpNatmapEndpointSourceEligibility = { + disabledReasonCode: null | TcpNatmapEndpointSourceDisabledReason; + eligible: boolean; +}; + +export function classifyTcpNatmapEndpointSource( + mapping: Pick< + NetworkPortForward, + 'desiredPresence' | 'isDeleted' | 'natmapDesiredEnabled' | 'protocol' + >, +): TcpNatmapEndpointSourceEligibility { + let disabledReasonCode: null | TcpNatmapEndpointSourceDisabledReason = null; + if (mapping.isDeleted || mapping.desiredPresence !== 'present') { + disabledReasonCode = 'SOURCE_DELETING'; + } else if (mapping.protocol !== 'tcp') { + disabledReasonCode = 'TCP_REQUIRED'; + } else if (!mapping.natmapDesiredEnabled) { + disabledReasonCode = 'NATMAP_DISABLED'; + } + return { + disabledReasonCode, + eligible: disabledReasonCode === null, + }; +} diff --git a/src/modules/qqbot/core/application/message-push/system-message-delivery-coordinator.service.ts b/src/modules/qqbot/core/application/message-push/system-message-delivery-coordinator.service.ts index 38ee6ed..1d9af34 100644 --- a/src/modules/qqbot/core/application/message-push/system-message-delivery-coordinator.service.ts +++ b/src/modules/qqbot/core/application/message-push/system-message-delivery-coordinator.service.ts @@ -18,6 +18,11 @@ import { SystemMessageDeliveryRunnerService } from './system-message-delivery-ru import { SystemMessageFanoutService } from './system-message-fanout.service'; const NETWORK_STUN_SOURCE = 'network.stun.mapping-port-changed'; +const NETWORK_TCP_NATMAP_SOURCE = 'network.tcp.natmap-endpoint-changed'; +const NETWORK_DDNS_MESSAGE_SOURCES = [ + NETWORK_STUN_SOURCE, + NETWORK_TCP_NATMAP_SOURCE, +] as const; @Injectable() export class SystemMessageDeliveryCoordinatorService @@ -92,39 +97,43 @@ export class SystemMessageDeliveryCoordinatorService ) return; const advanced = await this.dataSource.transaction(async (manager) => { - const subscriptions = await manager - .getRepository(QqbotMessageSubscription) - .find({ - where: { - enabled: true, - isDeleted: false, - sourceKey: NETWORK_STUN_SOURCE, + let affected = 0; + for (const sourceKey of NETWORK_DDNS_MESSAGE_SOURCES) { + const subscriptions = await manager + .getRepository(QqbotMessageSubscription) + .find({ + where: { + enabled: true, + isDeleted: false, + sourceKey, + }, + }); + const subscriptionIds = subscriptions + .filter( + (subscription) => + typeof subscription.sourceConfig?.ddnsRecordId === 'string' && + subscription.sourceConfig.ddnsRecordId === input.ddnsRecordId, + ) + .map((subscription) => subscription.id); + if (!subscriptionIds.length) continue; + const events = await manager + .getRepository(QqbotMessageEvent) + .find({ where: { sourceKey } }); + const eventIds = events + .filter((event) => event.payload.publicIpv4 === input.appliedAddress) + .map((event) => event.id); + if (!eventIds.length) continue; + const result = await manager.getRepository(QqbotMessageDelivery).update( + { + messageEventId: In(eventIds), + status: 'waiting_ddns', + subscriptionId: In(subscriptionIds), }, - }); - const subscriptionIds = subscriptions - .filter( - (subscription) => - typeof subscription.sourceConfig?.ddnsRecordId === 'string' && - subscription.sourceConfig.ddnsRecordId === input.ddnsRecordId, - ) - .map((subscription) => subscription.id); - if (!subscriptionIds.length) return 0; - const events = await manager - .getRepository(QqbotMessageEvent) - .find({ where: { sourceKey: NETWORK_STUN_SOURCE } }); - const eventIds = events - .filter((event) => event.payload.publicIpv4 === input.appliedAddress) - .map((event) => event.id); - if (!eventIds.length) return 0; - const result = await manager.getRepository(QqbotMessageDelivery).update( - { - messageEventId: In(eventIds), - status: 'waiting_ddns', - subscriptionId: In(subscriptionIds), - }, - { nextAttemptAt: new KtDateTime() }, - ); - return result.affected || 0; + { nextAttemptAt: new KtDateTime() }, + ); + affected += result.affected || 0; + } + return affected; }); if (advanced > 0) this.requestDrain(); } diff --git a/test/admin/network-management/network-agent-mqtt.service.spec.ts b/test/admin/network-management/network-agent-mqtt.service.spec.ts index fcfc683..6e3690e 100644 --- a/test/admin/network-management/network-agent-mqtt.service.spec.ts +++ b/test/admin/network-management/network-agent-mqtt.service.spec.ts @@ -2848,6 +2848,151 @@ describe('NetworkAgentMqttService', () => { expect(harness.deliveryCoordinator.requestDrain).not.toHaveBeenCalled(); }); + it('stages a TCP endpoint-change fact only after the matching report commits', async () => { + const harness = createV2Harness(); + const previousEndpoint = v2Endpoint('tcp_natmap', { + publicIpv4: '8.8.8.7', + publicPort: 45_100, + }); + harness.histories.push( + v2EndpointHistory({ + endpointIdentity: endpointLeaseIdentityV2(previousEndpoint), + endpointValidatedAt: new KtDateTime(previousEndpoint.validatedAt), + endpointValidUntil: new KtDateTime(previousEndpoint.validUntil), + eventId: 'v2-tcp-event-0', + mappingId: '101', + mechanism: 'tcp_natmap', + publicIpv4: previousEndpoint.publicIpv4, + publicPort: previousEndpoint.publicPort, + }), + ); + const currentEndpoint = v2Endpoint('tcp_natmap'); + await harness.service.consumeMessage( + 'kt/network/v2/agents/nas-main/events', + v2EndpointEvent({ + channelId: '101', + endpoint: currentEndpoint, + eventId: 'v2-tcp-event-1', + mechanism: 'tcp_natmap', + protocol: 'tcp', + }), + ); + + expect(harness.stagedEvents).toHaveLength(0); + expect(harness.deliveryCoordinator.requestDrain).not.toHaveBeenCalled(); + + await harness.service.consumeMessage( + 'kt/network/v2/agents/nas-main/reported', + v2Reported(harness), + ); + + expect(harness.stagedEvents).toEqual([ + { + eventId: 'v2-tcp-event-1', + occurredAt: '2099-07-27T00:00:16.000Z', + payload: { + previousPublicIpv4: '8.8.8.7', + previousPublicPort: 45_100, + publicIpv4: '8.8.8.8', + publicPort: 45_101, + tcpChannelId: '101', + }, + resourceKey: '101', + sourceKey: 'network.tcp.natmap-endpoint-changed', + }, + ]); + expect(harness.deliveryCoordinator.requestDrain).toHaveBeenCalledTimes(1); + }); + + it('stages a report-first TCP event when only the public IPv4 changes', async () => { + const harness = createV2Harness(); + const previousEndpoint = v2Endpoint('tcp_natmap', { + publicIpv4: '8.8.8.7', + }); + harness.histories.push( + v2EndpointHistory({ + endpointIdentity: endpointLeaseIdentityV2(previousEndpoint), + endpointValidatedAt: new KtDateTime(previousEndpoint.validatedAt), + endpointValidUntil: new KtDateTime(previousEndpoint.validUntil), + eventId: 'v2-tcp-event-0', + mappingId: '101', + mechanism: 'tcp_natmap', + publicIpv4: previousEndpoint.publicIpv4, + publicPort: previousEndpoint.publicPort, + }), + ); + await harness.service.consumeMessage( + 'kt/network/v2/agents/nas-main/reported', + v2Reported(harness), + ); + harness.deliveryCoordinator.requestDrain.mockClear(); + + await harness.service.consumeMessage( + 'kt/network/v2/agents/nas-main/events', + v2EndpointEvent({ + channelId: '101', + endpoint: v2Endpoint('tcp_natmap'), + eventId: 'v2-tcp-event-1', + mechanism: 'tcp_natmap', + protocol: 'tcp', + }), + ); + + expect(harness.stagedEvents[0]).toMatchObject({ + payload: { + previousPublicIpv4: '8.8.8.7', + previousPublicPort: 45_101, + publicIpv4: '8.8.8.8', + publicPort: 45_101, + tcpChannelId: '101', + }, + sourceKey: 'network.tcp.natmap-endpoint-changed', + }); + expect(harness.deliveryCoordinator.requestDrain).toHaveBeenCalledTimes(1); + }); + + it.each(['published', 'restored', 'withdrawn'] as const)( + 'does not stage a TCP %s lifecycle event', + async (type) => { + const harness = createV2Harness(); + const previousEndpoint = v2Endpoint('tcp_natmap', { + publicIpv4: '8.8.8.7', + publicPort: 45_100, + }); + harness.histories.push( + v2EndpointHistory({ + endpointIdentity: endpointLeaseIdentityV2(previousEndpoint), + eventId: 'v2-tcp-event-0', + mappingId: '101', + mechanism: 'tcp_natmap', + publicIpv4: previousEndpoint.publicIpv4, + publicPort: previousEndpoint.publicPort, + }), + ); + await harness.service.consumeMessage( + 'kt/network/v2/agents/nas-main/reported', + v2Reported(harness), + ); + harness.deliveryCoordinator.requestDrain.mockClear(); + + await harness.service.consumeMessage( + 'kt/network/v2/agents/nas-main/events', + v2EndpointEvent({ + channelId: '101', + endpoint: + type === 'withdrawn' ? undefined : v2Endpoint('tcp_natmap'), + eventId: `v2-tcp-${type}-1`, + mechanism: 'tcp_natmap', + protocol: 'tcp', + type, + }), + ); + + expect(harness.stagedEvents).toHaveLength(0); + expect(harness.deliveryCoordinator.requestDrain).not.toHaveBeenCalled(); + }, + ); + it('rolls back v2 history and Outbox together when staging fails', async () => { const harness = createV2Harness(); const currentEndpoint = v2Endpoint('udp_stun', { diff --git a/test/admin/network-management/network-ddns.service.spec.ts b/test/admin/network-management/network-ddns.service.spec.ts index 2cd7dec..e165dd5 100644 --- a/test/admin/network-management/network-ddns.service.spec.ts +++ b/test/admin/network-management/network-ddns.service.spec.ts @@ -11,10 +11,12 @@ import { } from '../../../src/modules/admin/platform-config/network-management/network-dnspod.client'; import type { NetworkManagementEventStreamService } from '../../../src/modules/admin/platform-config/network-management/network-management-event-stream.service'; import { NetworkPortForward } from '../../../src/modules/admin/platform-config/network-management/network-management.entity'; +import { NetworkPortForwardGroup } from '../../../src/modules/admin/platform-config/network-management/network-port-forward-group.entity'; type Harness = { client: jest.Mocked>; deliveryCoordinator: { notifyDdnsSynced: jest.Mock }; + group: NetworkPortForwardGroup; mapping: NetworkPortForward; recordUpdate: jest.Mock; records: NetworkDdnsRecord[]; @@ -42,6 +44,15 @@ function matchesUpdateCriterion(actual: unknown, expected: unknown): boolean { function createHarness(): Harness { const records: NetworkDdnsRecord[] = []; + const group = Object.assign(new NetworkPortForwardGroup(), { + externalPort: 45_678, + id: '10', + internalPort: 45_678, + isDeleted: false, + name: '公网服务', + protocolMode: 'udp', + targetIpv4: '192.168.31.224', + }); const mapping = Object.assign(new NetworkPortForward(), { currentObservedAt: new KtDateTime('2026-07-23T01:00:00.000Z'), currentPublicIpv4: '8.8.8.8', @@ -49,6 +60,7 @@ function createHarness(): Harness { currentValidUntil: new KtDateTime('2026-07-23T02:00:00.000Z'), desiredPresence: 'present', externalPort: 45_678, + groupId: group.id, id: '100', internalPort: 45_678, isDeleted: false, @@ -126,6 +138,11 @@ function createHarness(): Harness { findOne: async ({ where }) => where.id === mapping.id && !mapping.isDeleted ? mapping : null, } as unknown as Repository; + const groupRepository = { + find: async () => [group], + findOne: async ({ where }) => + where.id === group.id && !group.isDeleted ? group : null, + } as unknown as Repository; const stateRepository = { findOne: async () => state, } as unknown as Repository; @@ -157,6 +174,7 @@ function createHarness(): Harness { const service = new NetworkDdnsService( recordRepository, mappingRepository, + groupRepository, stateRepository, config, client as unknown as NetworkDnsPodClient, @@ -166,6 +184,7 @@ function createHarness(): Harness { return { client, deliveryCoordinator, + group, mapping, recordUpdate, records, @@ -223,7 +242,11 @@ describe('NetworkDdnsService', () => { expect.objectContaining({ currentAddress: '8.8.8.8', eligible: true, + groupId: '10', id: '100', + mechanism: 'udp_stun', + name: '公网服务 / UDP Keeper', + protocol: 'udp', sourceType: 'port_forward_ipv4', }), ]); @@ -239,6 +262,158 @@ describe('NetworkDdnsService', () => { ]); }); + it('accepts a managed TCP NATMap channel as an IPv4 DDNS source', async () => { + const harness = createHarness(); + harness.group.protocolMode = 'tcp'; + Object.assign(harness.mapping, { + keeperDesiredEnabled: false, + natmapDesiredEnabled: true, + protocol: 'tcp', + }); + + await expect( + harness.service.sourceOptions({ recordType: 'A' }), + ).resolves.toEqual([ + expect.objectContaining({ + currentAddress: '8.8.8.8', + eligible: true, + groupId: '10', + mechanism: 'tcp_natmap', + name: '公网服务 / TCP NATMap', + protocol: 'tcp', + }), + ]); + await expect( + harness.service.create({ + domain: 'kwitsukasa.top', + enabled: false, + name: 'TCP A', + portForwardId: '100', + recordType: 'A', + sourceType: 'port_forward_ipv4', + subDomain: 'tcp', + }), + ).resolves.toMatchObject({ + portForwardId: '100', + source: expect.objectContaining({ mechanism: 'tcp_natmap' }), + }); + }); + + it('derives accessEndpoint from a synchronized A record without writing DNS for a port-only change', async () => { + const harness = createHarness(); + harness.group.protocolMode = 'tcp'; + Object.assign(harness.mapping, { + keeperDesiredEnabled: false, + natmapDesiredEnabled: true, + protocol: 'tcp', + }); + await prepareEnabledA(harness); + Object.assign(harness.records[0], { + appliedAddress: '8.8.8.8', + providerRecordId: '300', + sourceAddress: '8.8.8.8', + syncStatus: 'synced', + }); + + await expect(harness.service.list()).resolves.toMatchObject({ + items: [{ accessEndpoint: 'pal.kwitsukasa.top:45678' }], + }); + harness.mapping.currentPublicPort = 45_679; + await harness.service.reconcileNow('200'); + + expect(harness.client.reconcile).not.toHaveBeenCalled(); + await expect(harness.service.list()).resolves.toMatchObject({ + items: [{ accessEndpoint: 'pal.kwitsukasa.top:45679' }], + }); + }); + + it('reconciles a changed TCP public IPv4 and exposes the endpoint only after provider readback', async () => { + const harness = createHarness(); + harness.group.protocolMode = 'tcp'; + Object.assign(harness.mapping, { + keeperDesiredEnabled: false, + natmapDesiredEnabled: true, + protocol: 'tcp', + }); + harness.client.reconcile.mockResolvedValueOnce({ + appliedAddress: '8.8.8.8', + changed: true, + providerRecordId: '300', + }); + await prepareEnabledA(harness); + Object.assign(harness.records[0], { + appliedAddress: '8.8.8.7', + providerRecordId: '300', + sourceAddress: '8.8.8.7', + syncStatus: 'synced', + }); + + await harness.service.reconcileNow('200'); + + expect(harness.client.reconcile).toHaveBeenCalledWith( + expect.objectContaining({ targetAddress: '8.8.8.8' }), + ); + await expect(harness.service.list()).resolves.toMatchObject({ + items: [ + { + accessEndpoint: 'pal.kwitsukasa.top:45678', + appliedAddress: '8.8.8.8', + }, + ], + }); + }); + + it('withdraws a TCP source to waiting_source without changing NATMap state', async () => { + const harness = createHarness(); + harness.group.protocolMode = 'tcp'; + Object.assign(harness.mapping, { + keeperDesiredEnabled: false, + natmapDesiredEnabled: true, + natmapStatus: 'active', + protocol: 'tcp', + }); + await prepareEnabledA(harness); + harness.mapping.currentPublicIpv4 = null; + harness.mapping.currentPublicPort = null; + + await harness.service.reconcileNow('200', true); + + expect(harness.records[0]).toMatchObject({ + sourceAddress: null, + syncStatus: 'waiting_source', + }); + expect(harness.mapping.natmapStatus).toBe('active'); + expect(harness.client.reconcile).not.toHaveBeenCalled(); + }); + + it('keeps TCP NATMap runtime state authoritative when DDNS synchronization fails', async () => { + const harness = createHarness(); + harness.group.protocolMode = 'tcp'; + Object.assign(harness.mapping, { + keeperDesiredEnabled: false, + natmapDesiredEnabled: true, + natmapStatus: 'active', + protocol: 'tcp', + }); + harness.client.reconcile.mockRejectedValueOnce( + new NetworkDnsPodClientError('DNSPOD_RATE_LIMITED', 'rate limited', true), + ); + await prepareEnabledA(harness); + + await harness.service.reconcileNow('200', true); + + expect(harness.records[0]).toMatchObject({ + lastErrorCode: 'provider_rate_limited', + syncStatus: 'failed', + }); + expect(harness.mapping).toMatchObject({ + currentPublicIpv4: '8.8.8.8', + currentPublicPort: 45_678, + natmapDesiredEnabled: true, + natmapStatus: 'active', + }); + }); + it('never exposes a residual lease from an ineligible IPv4 source', async () => { const harness = createHarness(); harness.mapping.keeperDesiredEnabled = false; diff --git a/test/admin/network-management/network-management.entity.spec.ts b/test/admin/network-management/network-management.entity.spec.ts index 48a704a..75f44f8 100644 --- a/test/admin/network-management/network-management.entity.spec.ts +++ b/test/admin/network-management/network-management.entity.spec.ts @@ -14,6 +14,7 @@ import { NetworkEndpointHistory } from '../../../src/modules/admin/platform-conf import { NetworkPortForwardGroup } from '../../../src/modules/admin/platform-config/network-management/network-port-forward-group.entity'; import { NetworkPortForward } from '../../../src/modules/admin/platform-config/network-management/network-management.entity'; import { NetworkManagementService } from '../../../src/modules/admin/platform-config/network-management/network-management.service'; +import { NetworkTcpNatmapMessageSourceAdapter } from '../../../src/modules/admin/platform-config/network-management/network-tcp-natmap-message-source.adapter'; describe('network management persistence module', () => { it('registers the five exact database entity tables', () => { @@ -225,6 +226,7 @@ describe('network management persistence module', () => { NetworkDdnsService, NetworkDnsPodClient, NetworkAgentMqttService, + NetworkTcpNatmapMessageSourceAdapter, ]), ); expect(providers).toEqual( @@ -233,6 +235,7 @@ describe('network management persistence module', () => { NetworkDdnsService, NetworkDnsPodClient, NetworkAgentMqttService, + NetworkTcpNatmapMessageSourceAdapter, ]), ); }); diff --git a/test/admin/network-management/network-tcp-natmap-message-source.adapter.spec.ts b/test/admin/network-management/network-tcp-natmap-message-source.adapter.spec.ts new file mode 100644 index 0000000..e3036d7 --- /dev/null +++ b/test/admin/network-management/network-tcp-natmap-message-source.adapter.spec.ts @@ -0,0 +1,286 @@ +import type { Repository } from 'typeorm'; +import { KtDateTime } from '../../../src/common'; +import { NetworkDdnsRecord } from '../../../src/modules/admin/platform-config/network-management/network-ddns.entity'; +import { NetworkPortForward } from '../../../src/modules/admin/platform-config/network-management/network-management.entity'; +import { NetworkPortForwardGroup } from '../../../src/modules/admin/platform-config/network-management/network-port-forward-group.entity'; +import { NetworkTcpNatmapMessageSourceAdapter } from '../../../src/modules/admin/platform-config/network-management/network-tcp-natmap-message-source.adapter'; +import { SystemMessageSourceRegistry } from '../../../src/modules/qqbot/core/application/message-push/system-message-source.registry'; + +type Harness = { + adapter: NetworkTcpNatmapMessageSourceAdapter; + ddns: NetworkDdnsRecord; + group: NetworkPortForwardGroup; + mapping: NetworkPortForward; + registry: SystemMessageSourceRegistry; +}; + +function createHarness(): Harness { + const group = Object.assign(new NetworkPortForwardGroup(), { + id: '2041700000000000001', + isDeleted: false, + name: '帕鲁新世界', + }); + const mapping = Object.assign(new NetworkPortForward(), { + currentPublicIpv4: '203.0.113.10', + currentPublicPort: 38213, + currentValidUntil: new KtDateTime('2026-07-27T13:00:00.000Z'), + desiredPresence: 'present' as const, + groupId: group.id, + id: '2041700000000000002', + isDeleted: false, + natmapDesiredEnabled: true, + name: group.name, + protocol: 'tcp' as const, + }); + const ddns = Object.assign(new NetworkDdnsRecord(), { + appliedAddress: '203.0.113.10', + domain: 'kwitsukasa.top', + enabled: true, + id: '2041700000000000003', + isDeleted: false, + name: '帕鲁 TCP 域名', + portForwardId: mapping.id, + recordType: 'A' as const, + sourceType: 'port_forward_ipv4' as const, + subDomain: 'pal', + syncStatus: 'synced' as const, + }); + const groups = [group]; + const mappings = [mapping]; + const records = [ddns]; + const groupRepository = { + find: jest.fn(async () => groups), + findOne: jest.fn( + async ({ where }) => groups.find((item) => item.id === where.id) || null, + ), + } as unknown as Repository; + const mappingRepository = { + find: jest.fn(async () => mappings), + findOne: jest.fn( + async ({ where }) => + mappings.find((item) => item.id === where.id) || null, + ), + } as unknown as Repository; + const ddnsRepository = { + find: jest.fn(async () => records), + findOne: jest.fn( + async ({ where }) => records.find((item) => item.id === where.id) || null, + ), + } as unknown as Repository; + const registry = new SystemMessageSourceRegistry(); + return { + adapter: new NetworkTcpNatmapMessageSourceAdapter( + mappingRepository, + groupRepository, + ddnsRepository, + registry, + ), + ddns, + group, + mapping, + registry, + }; +} + +function eventPayload(overrides: Record = {}) { + return { + previousPublicIpv4: '198.51.100.9', + previousPublicPort: 38111, + publicIpv4: '203.0.113.10', + publicPort: 38213, + tcpChannelId: '2041700000000000002', + ...overrides, + }; +} + +function subscriptionConfig(overrides: Record = {}) { + return { + ddnsRecordId: '2041700000000000003', + tcpChannelId: '2041700000000000002', + ...overrides, + }; +} + +describe('NetworkTcpNatmapMessageSourceAdapter', () => { + beforeEach(() => { + jest.useFakeTimers(); + jest.setSystemTime(new Date('2026-07-27T12:00:00.000Z')); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('registers an independent exact source definition', () => { + const { adapter, registry } = createHarness(); + adapter.onModuleInit(); + + expect(registry.get('network.tcp.natmap-endpoint-changed')).toBe(adapter); + expect(adapter.definition.subscriptionFields.map(({ key }) => key)).toEqual( + ['tcpChannelId', 'ddnsRecordId'], + ); + expect(adapter.definition.variables.map(({ key }) => key)).toEqual([ + 'endpoint', + 'fqdn', + 'publicIpv4', + 'publicPort', + 'previousPublicIpv4', + 'previousPublicPort', + 'portForwardName', + ]); + }); + + it('normalizes exactly two same-channel subscription fields', async () => { + const { adapter } = createHarness(); + + await expect( + adapter.normalizeSubscriptionConfig(subscriptionConfig()), + ).resolves.toEqual({ + canonicalConfig: subscriptionConfig(), + resourceKey: '2041700000000000002', + sourceSummary: '帕鲁新世界 / TCP NATMap · pal.kwitsukasa.top', + }); + await expect( + adapter.normalizeSubscriptionConfig( + subscriptionConfig({ portForwardId: '2041700000000000002' }), + ), + ).rejects.toMatchObject({ code: 'invalid_source_config' }); + }); + + it('exposes dependent TCP channel and same-channel DDNS options', async () => { + const { adapter } = createHarness(); + + await expect(adapter.listSubscriptionOptions()).resolves.toMatchObject({ + ddnsRecords: [ + { + dependsOnValue: '2041700000000000002', + disabled: false, + value: '2041700000000000003', + }, + ], + tcpChannels: [ + { + disabled: false, + label: '帕鲁新世界 / TCP NATMap', + value: '2041700000000000002', + }, + ], + }); + }); + + it('freezes the exact delivery variables from the event tuple and DDNS FQDN', async () => { + const { adapter } = createHarness(); + + await expect( + adapter.resolveDelivery({ + eventPayload: eventPayload(), + subscriptionConfig: subscriptionConfig(), + }), + ).resolves.toEqual({ + reasonCode: null, + status: 'ready', + variables: { + endpoint: 'pal.kwitsukasa.top:38213', + fqdn: 'pal.kwitsukasa.top', + portForwardName: '帕鲁新世界', + previousPublicIpv4: '198.51.100.9', + previousPublicPort: 38111, + publicIpv4: '203.0.113.10', + publicPort: 38213, + }, + }); + }); + + it('waits for matching DDNS, supersedes stale tuples, and cancels withdrawal or rebinding', async () => { + const waiting = createHarness(); + waiting.ddns.appliedAddress = '198.51.100.8'; + await expect( + waiting.adapter.resolveDelivery({ + eventPayload: eventPayload(), + subscriptionConfig: subscriptionConfig(), + }), + ).resolves.toMatchObject({ + reasonCode: 'ddns_not_synced', + status: 'waiting_ddns', + }); + + const expired = createHarness(); + expired.mapping.currentValidUntil = new KtDateTime( + '2026-07-27T11:59:59.999Z', + ); + await expect( + expired.adapter.resolveDelivery({ + eventPayload: eventPayload(), + subscriptionConfig: subscriptionConfig(), + }), + ).resolves.toEqual({ + reasonCode: 'endpoint_superseded', + status: 'superseded', + }); + + const superseded = createHarness(); + superseded.mapping.currentPublicPort = 39000; + await expect( + superseded.adapter.resolveDelivery({ + eventPayload: eventPayload(), + subscriptionConfig: subscriptionConfig(), + }), + ).resolves.toEqual({ + reasonCode: 'endpoint_superseded', + status: 'superseded', + }); + + const withdrawn = createHarness(); + withdrawn.mapping.currentPublicIpv4 = null; + withdrawn.mapping.currentPublicPort = null; + await expect( + withdrawn.adapter.resolveDelivery({ + eventPayload: eventPayload(), + subscriptionConfig: subscriptionConfig(), + }), + ).resolves.toEqual({ + reasonCode: 'endpoint_withdrawn', + status: 'cancelled', + }); + + const rebound = createHarness(); + rebound.ddns.portForwardId = '2041700000000000009'; + await expect( + rebound.adapter.resolveDelivery({ + eventPayload: eventPayload(), + subscriptionConfig: subscriptionConfig(), + }), + ).resolves.toMatchObject({ + reasonCode: 'ddns_mapping_mismatch', + status: 'cancelled', + }); + + const disabled = createHarness(); + disabled.mapping.natmapDesiredEnabled = false; + await expect( + disabled.adapter.resolveDelivery({ + eventPayload: eventPayload(), + subscriptionConfig: subscriptionConfig(), + }), + ).resolves.toEqual({ + reasonCode: 'natmap_disabled', + status: 'cancelled', + }); + }); + + it('rejects non-changing or malformed event tuples', () => { + const { adapter } = createHarness(); + + expect(() => + adapter.validateEventPayload( + eventPayload({ + previousPublicIpv4: '203.0.113.10', + previousPublicPort: 38213, + }), + ), + ).toThrow('invalid_source_config'); + expect(() => + adapter.validateEventPayload(eventPayload({ publicPort: '38213' })), + ).toThrow('invalid_source_config'); + }); +}); diff --git a/test/admin/network-management/network-tcp-natmap-source-eligibility.spec.ts b/test/admin/network-management/network-tcp-natmap-source-eligibility.spec.ts new file mode 100644 index 0000000..6eee716 --- /dev/null +++ b/test/admin/network-management/network-tcp-natmap-source-eligibility.spec.ts @@ -0,0 +1,41 @@ +import { classifyTcpNatmapEndpointSource } from '../../../src/modules/admin/platform-config/network-management/network-tcp-natmap-source-eligibility'; + +function source(overrides: Record = {}) { + return { + desiredPresence: 'present' as const, + isDeleted: false, + natmapDesiredEnabled: true, + protocol: 'tcp' as const, + ...overrides, + }; +} + +describe('classifyTcpNatmapEndpointSource', () => { + it('accepts only a managed TCP channel with NATMap enabled', () => { + expect(classifyTcpNatmapEndpointSource(source())).toEqual({ + disabledReasonCode: null, + eligible: true, + }); + }); + + it.each([ + [ + { desiredPresence: 'absent' }, + { disabledReasonCode: 'SOURCE_DELETING', eligible: false }, + ], + [ + { isDeleted: true }, + { disabledReasonCode: 'SOURCE_DELETING', eligible: false }, + ], + [ + { protocol: 'udp' }, + { disabledReasonCode: 'TCP_REQUIRED', eligible: false }, + ], + [ + { natmapDesiredEnabled: false }, + { disabledReasonCode: 'NATMAP_DISABLED', eligible: false }, + ], + ])('classifies %j without changing STUN semantics', (input, expected) => { + expect(classifyTcpNatmapEndpointSource(source(input))).toEqual(expected); + }); +}); diff --git a/test/modules/qqbot/message-push/qqbot-message-push.controller.spec.ts b/test/modules/qqbot/message-push/qqbot-message-push.controller.spec.ts index 40cfe7d..ecacd59 100644 --- a/test/modules/qqbot/message-push/qqbot-message-push.controller.spec.ts +++ b/test/modules/qqbot/message-push/qqbot-message-push.controller.spec.ts @@ -426,16 +426,24 @@ describe('QQBot message-push management controllers', () => { }); }); - it('resolves options and subscription config from arbitrary source metadata', async () => { + it('resolves dependent TCP NATMap options through the generic source endpoints', async () => { const definition = { - description: '后台任务完成', - displayName: '任务完成', - sourceKey: 'system.job.completed', + description: 'TCP NATMap 端点变更', + displayName: 'TCP NATMap 端点变更', + sourceKey: 'network.tcp.natmap-endpoint-changed', subscriptionFields: [ { - key: 'jobId', - label: '任务', - optionCollection: 'jobs', + key: 'tcpChannelId', + label: 'TCP NATMap 通道', + optionCollection: 'tcpChannels', + required: true, + type: 'select', + }, + { + dependsOn: 'tcpChannelId', + key: 'ddnsRecordId', + label: 'IPv4 DDNS 记录', + optionCollection: 'ddnsRecords', required: true, type: 'select', }, @@ -446,29 +454,47 @@ describe('QQBot message-push management controllers', () => { registry.get.mockReturnValueOnce({ definition, listSubscriptionOptions: jest.fn().mockResolvedValue({ - jobs: [ + ddnsRecords: [ { credential: 'must-not-leak', + dependsOnValue: 'tcp-channel-1', disabled: false, disabledReasonCode: null, - label: '夜间同步', - value: 'job-1', + label: 'Pal TCP · pal.example.test', + value: 'ddns-1', + }, + ], + tcpChannels: [ + { + disabled: false, + disabledReasonCode: null, + label: '帕鲁新世界 / TCP NATMap', + value: 'tcp-channel-1', }, ], }), }); const options = await request(apiUrl) .get( - '/qqbot/message-push/sources/system.job.completed/subscription-options', + '/qqbot/message-push/sources/network.tcp.natmap-endpoint-changed/subscription-options', ) .expect(200); expect(options.body.data).toEqual({ - jobs: [ + ddnsRecords: [ + { + dependsOnValue: 'tcp-channel-1', + disabled: false, + disabledReasonCode: null, + label: 'Pal TCP · pal.example.test', + value: 'ddns-1', + }, + ], + tcpChannels: [ { disabled: false, disabledReasonCode: null, - label: '夜间同步', - value: 'job-1', + label: '帕鲁新世界 / TCP NATMap', + value: 'tcp-channel-1', }, ], }); @@ -479,7 +505,8 @@ describe('QQBot message-push management controllers', () => { ...(subscriptionView() as unknown as Record), sourceConfig: { credential: 'must-not-leak', - jobId: 'job-1', + ddnsRecordId: 'ddns-1', + tcpChannelId: 'tcp-channel-1', }, sourceKey: definition.sourceKey, sourceName: definition.displayName, @@ -492,7 +519,8 @@ describe('QQBot message-push management controllers', () => { .get('/qqbot/message-push/subscriptions') .expect(200); expect(subscriptionsResponse.body.data.items[0].sourceConfig).toEqual({ - jobId: 'job-1', + ddnsRecordId: 'ddns-1', + tcpChannelId: 'tcp-channel-1', }); }); diff --git a/test/modules/qqbot/message-push/system-message-delivery-coordinator-integration.spec.ts b/test/modules/qqbot/message-push/system-message-delivery-coordinator-integration.spec.ts index 60345d7..ca0ea5e 100644 --- a/test/modules/qqbot/message-push/system-message-delivery-coordinator-integration.spec.ts +++ b/test/modules/qqbot/message-push/system-message-delivery-coordinator-integration.spec.ts @@ -6,6 +6,7 @@ import { QqbotMessageEvent } from '../../../../src/modules/qqbot/core/infrastruc import { QqbotMessageSubscription } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-subscription.entity'; const SOURCE_KEY = 'network.stun.mapping-port-changed'; +const TCP_SOURCE_KEY = 'network.tcp.natmap-endpoint-changed'; const NOW = new Date('2026-07-24T03:04:05.000Z'); type CoordinatorStore = { @@ -78,6 +79,13 @@ function createCoordinatorHarness(): CoordinatorHarness { status: 'waiting_ddns', subscriptionId: 'subscription-match', }, + { + id: 'delivery-tcp-match', + messageEventId: 'event-tcp-match', + nextAttemptAt: 'future-tcp-match', + status: 'waiting_ddns', + subscriptionId: 'subscription-tcp-match', + }, ], events: [ { @@ -100,6 +108,11 @@ function createCoordinatorHarness(): CoordinatorHarness { payload: { publicIpv4: 8_888 }, sourceKey: SOURCE_KEY, }, + { + id: 'event-tcp-match', + payload: { publicIpv4: '8.8.8.8' }, + sourceKey: TCP_SOURCE_KEY, + }, ], subscriptions: [ { @@ -109,6 +122,13 @@ function createCoordinatorHarness(): CoordinatorHarness { sourceConfig: { ddnsRecordId: '9007199254740993' }, sourceKey: SOURCE_KEY, }, + { + enabled: true, + id: 'subscription-tcp-match', + isDeleted: false, + sourceConfig: { ddnsRecordId: '9007199254740993' }, + sourceKey: TCP_SOURCE_KEY, + }, { enabled: false, id: 'subscription-disabled', @@ -263,7 +283,7 @@ describe('SystemMessageDeliveryCoordinatorService DDNS integration', () => { expect(harness.store.deliveries).toEqual( before.deliveries.map((delivery) => - delivery.id === 'delivery-match' + delivery.id === 'delivery-match' || delivery.id === 'delivery-tcp-match' ? { ...delivery, nextAttemptAt: new KtDateTime(NOW) } : delivery, ), @@ -271,6 +291,7 @@ describe('SystemMessageDeliveryCoordinatorService DDNS integration', () => { expect(harness.operations).toEqual([ 'transaction:start', 'delivery:update', + 'delivery:update', 'transaction:commit', 'drain', ]); @@ -282,6 +303,9 @@ describe('SystemMessageDeliveryCoordinatorService DDNS integration', () => { harness.store.deliveries.find( (delivery) => delivery.id === 'delivery-match', )!.status = 'processing'; + harness.store.deliveries.find( + (delivery) => delivery.id === 'delivery-tcp-match', + )!.status = 'processing'; const before = cloneStore(harness.store); await harness.coordinator.notifyDdnsSynced({ diff --git a/test/modules/qqbot/message-push/system-message-delivery-runner.spec.ts b/test/modules/qqbot/message-push/system-message-delivery-runner.spec.ts index 0b11037..69facd8 100644 --- a/test/modules/qqbot/message-push/system-message-delivery-runner.spec.ts +++ b/test/modules/qqbot/message-push/system-message-delivery-runner.spec.ts @@ -27,6 +27,7 @@ import { QqbotMessageSubscription } from '../../../../src/modules/qqbot/core/inf const NOW = new Date('2026-07-24T00:00:00.000Z'); const SOURCE_KEY = 'network.stun.mapping-port-changed'; +const TCP_SOURCE_KEY = 'network.tcp.natmap-endpoint-changed'; const RESOURCE_KEY = '9007199254740993'; type Store = { @@ -201,7 +202,7 @@ function matches( }); } -function setup(seed: Partial = {}) { +function setup(seed: Partial = {}, sourceKey = SOURCE_KEY) { let state: Store = { accounts: seed.accounts ?? [account()], bindings: seed.bindings ?? [binding()], @@ -215,7 +216,7 @@ function setup(seed: Partial = {}) { definition: { description: 'test', displayName: 'test', - sourceKey: SOURCE_KEY, + sourceKey, subscriptionFields: [], variables: [ { @@ -971,6 +972,40 @@ describe('System message delivery runner direct preflight contracts', () => { expect(harness.getState().deliveries[0].status).toBe('success'); }); + it('11 handles a due TCP NATMap waiting row through the generic source registry', async () => { + const harness = setup( + { + deliveries: [delivery({ status: 'waiting_ddns' })], + events: [ + event({ + payload: { + publicIpv4: '203.0.113.10', + publicPort: 45_101, + tcpChannelId: RESOURCE_KEY, + }, + sourceKey: TCP_SOURCE_KEY, + }), + ], + subscriptions: [ + subscription({ + sourceConfig: { + ddnsRecordId: '9007199254740995', + tcpChannelId: RESOURCE_KEY, + }, + sourceKey: TCP_SOURCE_KEY, + }), + ], + }, + TCP_SOURCE_KEY, + ); + + await harness.runner.runOnce(NOW); + + expect(harness.adapter.resolveDelivery).toHaveBeenCalledTimes(1); + expect(harness.sender.sendStrictPlainText).toHaveBeenCalledTimes(1); + expect(harness.getState().deliveries[0].status).toBe('success'); + }); + it.each([ ['endpoint_port_changed', 'superseded'], ['endpoint_ipv4_changed', 'superseded'], diff --git a/test/modules/qqbot/message-push/system-message-fanout.service.spec.ts b/test/modules/qqbot/message-push/system-message-fanout.service.spec.ts index 64b6eda..bcb8114 100644 --- a/test/modules/qqbot/message-push/system-message-fanout.service.spec.ts +++ b/test/modules/qqbot/message-push/system-message-fanout.service.spec.ts @@ -23,6 +23,7 @@ import { const NOW = new Date('2026-07-24T00:00:00.000Z'); const SOURCE_KEY = 'network.stun.mapping-port-changed'; +const TCP_SOURCE_KEY = 'network.tcp.natmap-endpoint-changed'; const JOB_SOURCE_KEY = 'system.job.completed'; const RESOURCE_KEY = '9007199254740993'; @@ -800,6 +801,52 @@ describe('SystemMessageFanoutService', () => { ); }); + it('fans out the independent TCP NATMap source without source-specific core logic', async () => { + const adapter = sourceAdapter(TCP_SOURCE_KEY); + adapter.eventResourceKey.mockImplementation((payload) => + typeof payload.tcpChannelId === 'string' ? payload.tcpChannelId : '', + ); + adapter.subscriptionResourceKey.mockImplementation((config) => + typeof config.tcpChannelId === 'string' ? config.tcpChannelId : null, + ); + adapter.resolveDelivery.mockResolvedValue({ + reasonCode: null, + status: 'ready', + variables: { endpoint: 'tcp.example.com:45101' }, + }); + const fixture = setup( + { + events: [ + event({ + payload: { + publicIpv4: '203.0.113.10', + publicPort: 45_101, + tcpChannelId: RESOURCE_KEY, + }, + sourceKey: TCP_SOURCE_KEY, + }), + ], + subscriptions: [ + subscription({ + sourceConfig: { + ddnsRecordId: '9007199254740995', + tcpChannelId: RESOURCE_KEY, + }, + sourceKey: TCP_SOURCE_KEY, + }), + ], + templates: [template({ sourceKey: TCP_SOURCE_KEY })], + }, + adapter, + ); + + await expect(fixture.service.runOnce(NOW)).resolves.toBe(1); + expect(fixture.deliveries()[0]).toMatchObject({ + renderedMessage: 'endpoint=tcp.example.com:45101', + variableSnapshot: { endpoint: 'tcp.example.com:45101' }, + }); + }); + it('schedules ready and waiting-DDNS snapshots without sending OneBot', async () => { const fixture = setup(); fixture.adapter.resolveDelivery.mockResolvedValueOnce({ diff --git a/test/modules/qqbot/message-push/system-message-source.registry.spec.ts b/test/modules/qqbot/message-push/system-message-source.registry.spec.ts index fcea49f..e87e3d9 100644 --- a/test/modules/qqbot/message-push/system-message-source.registry.spec.ts +++ b/test/modules/qqbot/message-push/system-message-source.registry.spec.ts @@ -42,20 +42,20 @@ describe('SystemMessageSourceRegistry', () => { it('sorts definitions and unregisters only the same adapter instance', () => { const registry = new SystemMessageSourceRegistry(); - const first = createAdapter('z.source'); - const second = createAdapter('a.source'); - const replacement = createAdapter('z.source'); + const first = createAdapter('network.tcp.natmap-endpoint-changed'); + const second = createAdapter('network.stun.mapping-port-changed'); + const replacement = createAdapter('network.tcp.natmap-endpoint-changed'); registry.register(first); registry.register(second); - registry.unregister('z.source', replacement); + registry.unregister('network.tcp.natmap-endpoint-changed', replacement); expect(registry.list().map((definition) => definition.sourceKey)).toEqual([ - 'a.source', - 'z.source', + 'network.stun.mapping-port-changed', + 'network.tcp.natmap-endpoint-changed', ]); - registry.unregister('z.source', first); + registry.unregister('network.tcp.natmap-endpoint-changed', first); expect(registry.list().map((definition) => definition.sourceKey)).toEqual([ - 'a.source', + 'network.stun.mapping-port-changed', ]); }); });