From 06e6be3c390166d37c6f9de56253f4315dc8a4a9 Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Thu, 9 Jul 2026 14:57:55 +0800 Subject: [PATCH] fix: align channel copy and gateway connection state --- api/src/channels/channels.service.spec.ts | 2 +- api/src/channels/channels.service.ts | 4 +- .../first-version-development-requirements.md | 8 +- docs/system-functional-test-cases.md | 8 +- docs/testing-progress.md | 14 ++ gateway/internal/control/server.go | 121 ++++------ gateway/internal/control/server_test.go | 69 +++--- .../internal/upstream/connection_loss_test.go | 15 ++ gateway/internal/upstream/manager.go | 219 +++++++++++++++--- src/apps/LoginPage.tsx | 2 +- src/apps/admin/AdminUsersPage.tsx | 2 +- src/apps/client/ClientUsersPage.tsx | 2 +- 12 files changed, 309 insertions(+), 157 deletions(-) diff --git a/api/src/channels/channels.service.spec.ts b/api/src/channels/channels.service.spec.ts index 1228a29..25e5a4b 100644 --- a/api/src/channels/channels.service.spec.ts +++ b/api/src/channels/channels.service.spec.ts @@ -569,7 +569,7 @@ describe('ChannelsService', () => { const copied = await service.copyChannel('channel-1', { operatorId: 'admin-1' }); - expect(copied).toEqual(expect.objectContaining({ id: 'channel-copy', name: '主通道副本' })); + expect(copied).toEqual(expect.objectContaining({ id: 'channel-copy', name: '主通道副本', status: 'disabled' })); expect(prisma.$transaction).toHaveBeenCalled(); }); diff --git a/api/src/channels/channels.service.ts b/api/src/channels/channels.service.ts index 42df7d5..80dfa8b 100644 --- a/api/src/channels/channels.service.ts +++ b/api/src/channels/channels.service.ts @@ -356,7 +356,7 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy { cmppVersion: source.cmppVersion, rateLimitPerSecond: source.rateLimitPerSecond, unitPrice: source.unitPrice, - status: source.status, + status: 'disabled', config: source.config as Prisma.InputJsonValue | undefined, reportFields: { create: source.reportFields.map((field) => ({ @@ -396,6 +396,8 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy { detail: { sourceChannelId: source.id, sourceCode: source.code, + sourceStatus: source.status, + copiedStatus: 'disabled', copiedReportFields: source.reportFields.length, copiedReportMaterials: reportMaterials.length, } as Prisma.InputJsonValue, diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index 1515838..c63251b 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -415,12 +415,12 @@ - 支持新增、编辑、删除、启用、停用短信通道。 - 删除通道采用软删除或停用归档，不能破坏历史发送、报备、日志外键；删除、启用、停用、复制等高影响操作必须二次确认并写系统日志。 -- 支持复制通道：复制后新建一个通道，除 id/code 自动生成外，通道配置、CMPP 参数、通道报备字段、签名/引流报备材料和个性化字段配置均需从源通道复制；名称默认追加“副本”。 +- 支持复制通道：复制后新建一个通道，除 id/code 自动生成外，通道配置、CMPP 参数、通道报备字段、签名/引流报备材料和个性化字段配置均需从源通道复制；名称默认追加“副本”；若源通道为 active，新副本默认保存为 disabled，避免复制后立即占用上游连接。 - 支持发送测试短信。 - 支持查看通道成功率、未知率、失败率、累计发送量。 - 支持进入通道报备详情。 - 通道报备详情页中，签名下的引流信息默认收起，用户点击后展开；展开/收起只影响页面展示，不改变报备数据。 -- 通道列表状态区域展示“连接日志”入口；点击后弹窗展示真实连接日志，包括连接请求、连接成功、断开、心跳、重连、异常等事件，日志来源于 Gateway 回写或 OperationLog。 +- 通道列表状态区域展示“连接日志”入口；点击后弹窗展示真实连接日志，包括连接请求、连接成功、断开、心跳、重连、异常等事件，日志来源于 Gateway 回写或 OperationLog。通道连接状态、连接数和最近错误必须来自 Gateway 真实上游连接池回写，不能以一次性探测拨号成功代替长连接在线状态。 - 通道操作按钮应保持一致的两列布局，报备详情、编辑、复制、发送测试、启停、删除等操作文案清晰。 ### 5.14 运营端通道组管理 @@ -1331,7 +1331,7 @@ ### 1. 登录入口 - 客户端和运营端必须使用独立登录页面：客户端 `/client/login`，运营端 `/admin/login`。 -- 两端登录均输入邮箱或手机号、密码和图形验证码。 +- 两端登录均输入用户名、邮箱或手机号，外加密码和图形验证码。 - 登录接口必须调用真实 NestJS API，不允许前端静态用户、localStorage mock 或纯前端验证码作为验收依据。 - 运营端登录仅允许平台管理员；客户端登录仅允许已关联企业的企业管理员。 @@ -1347,7 +1347,7 @@ - 运营端和客户端用户管理均需接入真实 API，支持添加、编辑、启用/禁用、删除、修改密码。 - 启用/禁用、删除必须弹窗确认。 - 删除采用软删除，历史系统日志、审核记录和业务记录仍可追溯；删除后用户不可登录且列表默认不展示。 -- 客户端用户字段必须包含邮箱和手机号，并支持邮箱或手机号登录。 +- 客户端用户字段必须包含邮箱和手机号，并支持用户名、邮箱或手机号登录。 - 所有创建、编辑、启用、禁用、删除、修改密码动作必须写系统日志。 ### 4. fail2ban diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index b77cf18..f5949db 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -247,6 +247,7 @@ - 通道协议默认为 CMPP，CMPP 版本默认 2.0，且可选择 2.0 或 3.0。 - 通道限速保存正确。 - 通道真实保存 `desiredConnections/windowSize`，后续 Gateway `ConnectChannel` 与 `SubmitCommand.upstream` 使用该配置。 + - 通道列表和通道组页面展示的连接状态、连接数和最近错误来自 Gateway 真实上游连接池回写；一次性连接探测成功不能展示为 `connected`。 - 通道测试短信必须调用真实 NestJS API，创建独立 `SmsMessageRecord` 和 `SmsSubmitRecord`，并向 Redis Stream `gateway.submit.commands` 写入 `SubmitCommand`；短信记录页面能查询到该测试短信。 - 通道测试短信不得绑定企业、企业应用或 `SmsBatchTask` 发送任务；submit result 和 receipt 只更新短信记录、提交记录、回执记录和分片审计，不向客户侧推送 Deliver。 - 若通道未启用或没有在线 CMPP 连接，测试短信 API 返回明确错误，不能只在前端假提示成功。 @@ -442,6 +443,7 @@ - 预期结果： - 后端创建新通道，新通道 id/code 与源通道不同，名称默认追加“副本”。 - 通道配置、CMPP 参数、报备字段、签名/引流报备材料和个性化字段与源通道一致。 + - 若源通道为 active，新通道默认保存为 disabled，且不会立即触发上游真实连接。 - 复制动作写入系统日志。 - 复制后新通道可继续编辑、启停、删除，不影响源通道。 @@ -3090,13 +3092,13 @@ npm run verify:phase8 | 用例编号 | 操作 | 预期结果 | | --- | --- | --- | -| TC-AUTH-001 | 打开 `/admin/login`，输入平台管理员邮箱或手机号、密码和正确图形验证码。 | 登录成功进入运营端；session 用户角色为 `platform_admin`；后续运营端 API 使用真实后端。 | +| TC-AUTH-001 | 打开 `/admin/login`，输入平台管理员用户名、邮箱或手机号，外加密码和正确图形验证码。 | 登录成功进入运营端；session 用户角色为 `platform_admin`；后续运营端 API 使用真实后端。 | | TC-AUTH-002 | 使用企业管理员账号登录 `/admin/login`。 | 登录失败；返回“仅平台管理员可登录运营端”类错误；失败次数累计。 | -| TC-AUTH-003 | 打开 `/client/login`，输入已关联企业的企业管理员邮箱或手机号、密码和正确图形验证码。 | 登录成功进入客户端；客户端 API 自动携带当前企业 `tenantId`；Dashboard、用户、日志等仅展示当前企业数据。 | +| TC-AUTH-003 | 打开 `/client/login`，输入已关联企业的企业管理员用户名、邮箱或手机号，外加密码和正确图形验证码。 | 登录成功进入客户端；客户端 API 自动携带当前企业 `tenantId`；Dashboard、用户、日志等仅展示当前企业数据。 | | TC-AUTH-004 | 使用平台管理员或未关联企业的用户登录 `/client/login`。 | 登录失败；不进入客户端。 | | TC-AUTH-005 | 同一用户连续输错密码 5 次。 | 第 5 次后用户锁定 24 小时；锁定期内正确密码也被拒绝；系统记录失败次数和锁定时间。 | | TC-AUTH-006 | 输入错误或过期图形验证码登录。 | 返回 400 可读错误；必须刷新验证码后重试。 | -| TC-USER-ADMIN-001 | 运营端新增平台管理员，填写邮箱或手机号、初始密码。 | 创建成功；用户无 `tenantId`；可登录运营端；写 `user.created` 日志。 | +| TC-USER-ADMIN-001 | 运营端新增平台管理员，填写用户名/登录账号、邮箱或手机号、初始密码。 | 创建成功；用户无 `tenantId`；可登录运营端；写 `user.created` 日志。 | | TC-USER-ADMIN-002 | 运营端新增企业管理员但不选择企业。 | 返回 400；不创建用户。 | | TC-USER-ADMIN-003 | 运营端新增企业管理员并选择企业。 | 创建成功；用户关联企业；可登录客户端；客户端数据按该企业隔离。 | | TC-USER-ADMIN-004 | 运营端编辑用户、启用/禁用、删除、修改密码。 | 编辑和改密调用真实 API；启停/删除有确认弹窗；删除后列表不展示且不可登录；均写系统日志。 | diff --git a/docs/testing-progress.md b/docs/testing-progress.md index 2769b4d..dc8ca6d 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -8,6 +8,19 @@ - 已执行：`npm --prefix api test -- channels.service.spec.ts --runInBand`、`npm --prefix api run build`、`npm run build`。待生产部署后用指定号码做一次真实发送验证，并回查 DB/短信记录。 - 2026-07-09 追加：按产品边界收窄通道测试短信，`SmsMessageRecord/SmsSubmitRecord/SmsReceiptRecord/SmsMessageSegmentAudit` 支持 `tenantId/batchTaskId` 为空；通道测试只记录短信、提交结果和回执，不进入企业账务、发送任务进度、客户侧 Deliver 推送或业务补发。 +## 2026-07-09 通道复制默认停用与真实连接池状态回写修复 + +- 生产验证发现当前 3 个赛邮行业通道中有 2 个由首个 active 通道复制而来；数据库 `CmppConnectionState` 曾显示 3 条通道都为 `connected/currentConnections=1`，但生产机 `ss/netstat` 与上游平台都只能看到 1 条真实 TCP 长连接。 +- 根因一：`POST /api/admin/channels/:id/copy` 会继承源通道 `status=active`，复制后的通道创建完成后立即参与 Gateway 连接流程，和“复制只是拷贝配置，不应自动上线”的产品预期不符。 +- 根因二：Gateway `/connections/connect` 之前只做一次性 `DialCMPP` 探测，探测成功后就把 `connected/currentConnections=desiredConnections` 回写给 NestJS；该探测连接随后立即断开，导致页面状态与真实上游连接池不一致。 +- 已修复： + - 通道复制后统一保存为 `disabled`，复制日志补充 `sourceStatus` 和 `copiedStatus`，避免 active 通道副本自动占用上游连接。 + - Gateway `ConnectChannel` 改为直接建立/复用真实上游连接池，并由连接池回写 `connected/failed/disconnected`、`currentConnections`、最近错误和断开时间；连接丢失时状态随真实连接数变化更新。 +- 测试口径同步： + - `TC-ADMIN-003` 增加“连接状态必须来自真实上游连接池回写”的要求。 + - `TC-ADMIN-016` 明确复制 active 通道后副本默认 `disabled`，且不会立即触发上游真实连接。 +- 已执行：`npm --prefix api test -- channels.service.spec.ts --runInBand`、`npm --prefix api run build`、`go test ./internal/control ./internal/upstream`、`go build -o ..\\dist\\cmpp-gateway .\\cmd\\gateway`。 + ## 2026-07-09 线上通道 CMPP 版本修复 - 线上生产验证发现 3 个赛邮行业通道配置均指向 `121.40.172.212:7890`，其中 2 个已触发 Gateway 真实连接并失败，`CmppConnectionState.lastError` 为 `packetWriter.ReadBytes error: ReadBytes reads 14 bytes, not equal to 16 we expected`。 @@ -468,6 +481,7 @@ GET /api/admin/system-logs?page=1&pageSize=2 - 新增用户登录字段和 fail2ban 持久化字段：`email`、`phone`、`failedLoginCount`、`lockedUntil`、`lastLoginAt`、`deletedAt`。 - 登录入口拆分为 `/client/login` 和 `/admin/login`，两端均调用真实验证码和登录 API。 +- 用户登录入口和用户表单统一文案为“用户名/登录账号”，提示可用用户名、邮箱或手机号登录。 - 运营端登录仅允许 `platform_admin`；客户端登录仅允许已关联企业的 `enterprise_admin`。 - 运营端用户管理接入真实 `/api/admin/users`，支持平台管理员和企业管理员的新增、编辑、启用/禁用、删除、改密；企业管理员必须关联企业。 - 客户端用户管理接入真实 `/api/client/users`，所有操作继承当前登录企业 `tenantId`。 diff --git a/gateway/internal/control/server.go b/gateway/internal/control/server.go index 6060c66..48d8a1e 100644 --- a/gateway/internal/control/server.go +++ b/gateway/internal/control/server.go @@ -1,24 +1,18 @@ package control import ( - "bytes" "context" "encoding/json" "fmt" "net/http" - "strings" "time" "cmpp-platform/gateway/internal/inbound" "cmpp-platform/gateway/internal/queue" "cmpp-platform/gateway/internal/upstream" - - cmpp "github.com/bigwhite/gocmpp" ) -const defaultConnectTimeout = 5 * time.Second - -type DialFunc func(context.Context, ConnectChannelCommand) error +type ConnectFunc func(context.Context, ConnectChannelCommand) (ConnectionStateCallback, error) type ConnectChannelCommand struct { SchemaVersion string `json:"schemaVersion"` @@ -59,7 +53,7 @@ type ConnectionStateCallback struct { type Server struct { APIBaseURL string HTTPClient *http.Client - Dial DialFunc + Connect ConnectFunc Upstream *upstream.Manager RecoveryCandidates func(context.Context) ([]inbound.DownstreamPresence, error) RecoveryStatuses func(context.Context) ([]inbound.DownstreamRecoveryStatus, error) @@ -74,12 +68,12 @@ func Register(mux *http.ServeMux, server Server) { if server.HTTPClient == nil { server.HTTPClient = &http.Client{Timeout: 10 * time.Second} } - if server.Dial == nil { - server.Dial = DialCMPP - } if server.Upstream == nil { server.Upstream = &upstream.Manager{APIBaseURL: server.APIBaseURL, HTTPClient: server.HTTPClient} } + if server.Connect == nil { + server.Connect = server.connectChannel + } mux.HandleFunc("/connections/connect", server.handleConnectChannel) mux.HandleFunc("/upstream/submit", server.handleUpstreamSubmit) mux.HandleFunc("/downstream/receipt", server.handleDownstreamReceipt) @@ -105,26 +99,9 @@ func (s Server) handleConnectChannel(w http.ResponseWriter, r *http.Request) { return } - status := ConnectionStateCallback{ - ChannelID: command.ChannelID, - ConnectionID: command.ConnectionID, - DesiredConnections: desiredConnections(command.DesiredConnections), - } - if err := s.Dial(r.Context(), command); err != nil { - status.Status = "failed" - status.CurrentConnections = 0 - status.LastDisconnectedAt = time.Now().UTC().Format(time.RFC3339Nano) - status.LastError = err.Error() - } else { - now := time.Now().UTC().Format(time.RFC3339Nano) - status.Status = "connected" - status.CurrentConnections = status.DesiredConnections - status.LastConnectedAt = now - status.LastHeartbeatAt = now - } - - if err := s.postConnectionState(r.Context(), status); err != nil { - http.Error(w, fmt.Sprintf("failed to callback api: %v", err), http.StatusBadGateway) + status, err := s.Connect(r.Context(), command) + if err != nil { + http.Error(w, fmt.Sprintf("failed to establish upstream pool: %v", err), http.StatusBadGateway) return } @@ -258,55 +235,43 @@ func (s Server) handleDownstreamRecoveryOverview(w http.ResponseWriter, r *http. }) } -func DialCMPP(ctx context.Context, command ConnectChannelCommand) error { - ctx, cancel := context.WithTimeout(ctx, defaultConnectTimeout) - defer cancel() - - version := cmpp.V30 - if strings.HasPrefix(command.Channel.CMPPVersion, "2") { - version = cmpp.V20 - } - - client := cmpp.NewClient(version) - defer client.Disconnect() - - done := make(chan error, 1) - go func() { - addr := fmt.Sprintf("%s:%d", command.Channel.GatewayHost, command.Channel.GatewayPort) - done <- client.Connect(addr, command.Channel.Account, command.Channel.PasswordCipher, defaultConnectTimeout) - }() - - select { - case <-ctx.Done(): - return ctx.Err() - case err := <-done: - return err - } -} - -func (s Server) postConnectionState(ctx context.Context, state ConnectionStateCallback) error { - apiBaseURL := strings.TrimRight(s.APIBaseURL, "/") - if apiBaseURL == "" { - apiBaseURL = "http://127.0.0.1:3000/api" - } - payload, err := json.Marshal(state) +func (s Server) connectChannel(ctx context.Context, command ConnectChannelCommand) (ConnectionStateCallback, error) { + state, err := s.Upstream.ConnectChannel(ctx, queue.ConnectChannelCommand{ + SchemaVersion: command.SchemaVersion, + MessageType: queue.MessageTypeConnectChannel, + TraceID: command.TraceID, + ChannelID: command.ChannelID, + ConnectionID: command.ConnectionID, + CreatedAt: time.Now().UTC(), + Reason: command.Reason, + DesiredConnections: command.DesiredConnections, + Channel: queue.ConnectChannelConfig{ + Code: command.Channel.Code, + Name: command.Channel.Name, + GatewayHost: command.Channel.GatewayHost, + GatewayPort: command.Channel.GatewayPort, + Account: command.Channel.Account, + PasswordCipher: command.Channel.PasswordCipher, + SrcID: command.Channel.SrcID, + CMPPVersion: command.Channel.CMPPVersion, + RateLimitPerSecond: command.Channel.RateLimitPerSecond, + }, + }) if err != nil { - return err + return ConnectionStateCallback{}, err } - req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiBaseURL+"/admin/gateway/connections", bytes.NewReader(payload)) - if err != nil { - return err - } - req.Header.Set("Content-Type", "application/json") - resp, err := s.HTTPClient.Do(req) - if err != nil { - return err - } - defer resp.Body.Close() - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return fmt.Errorf("api returned %s", resp.Status) - } - return nil + return ConnectionStateCallback{ + ChannelID: state.ChannelID, + ConnectionID: state.ConnectionID, + Status: state.Status, + DesiredConnections: state.DesiredConnections, + CurrentConnections: state.CurrentConnections, + LastConnectedAt: state.LastConnectedAt, + LastDisconnectedAt: state.LastDisconnectedAt, + LastHeartbeatAt: state.LastHeartbeatAt, + ReconnectCount: state.ReconnectCount, + LastError: state.LastError, + }, nil } func validateConnectChannelCommand(command ConnectChannelCommand) error { diff --git a/gateway/internal/control/server_test.go b/gateway/internal/control/server_test.go index d5c1329..f12b441 100644 --- a/gateway/internal/control/server_test.go +++ b/gateway/internal/control/server_test.go @@ -13,20 +13,16 @@ import ( ) func TestConnectChannelCallbacksConnectedState(t *testing.T) { - var callback ConnectionStateCallback - api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.URL.Path != "/api/admin/gateway/connections" { - t.Fatalf("unexpected callback path: %s", r.URL.Path) - } - if err := json.NewDecoder(r.Body).Decode(&callback); err != nil { - t.Fatalf("decode callback: %v", err) - } - w.WriteHeader(http.StatusOK) - })) - defer api.Close() - - handler := handlerWithDial(api.URL+"/api", func(context.Context, ConnectChannelCommand) error { - return nil + handler := handlerWithConnect(func(context.Context, ConnectChannelCommand) (ConnectionStateCallback, error) { + return ConnectionStateCallback{ + ChannelID: "channel-1", + ConnectionID: "channel-1:primary", + Status: "connected", + DesiredConnections: 2, + CurrentConnections: 2, + LastConnectedAt: time.Now().UTC().Format(time.RFC3339Nano), + LastHeartbeatAt: time.Now().UTC().Format(time.RFC3339Nano), + }, nil }) resp := httptest.NewRecorder() @@ -36,6 +32,10 @@ func TestConnectChannelCallbacksConnectedState(t *testing.T) { if resp.Code != http.StatusOK { t.Fatalf("unexpected response status: %d body=%s", resp.Code, resp.Body.String()) } + var callback ConnectionStateCallback + if err := json.Unmarshal(resp.Body.Bytes(), &callback); err != nil { + t.Fatalf("decode response: %v", err) + } if callback.Status != "connected" || callback.CurrentConnections != 2 || callback.DesiredConnections != 2 { t.Fatalf("unexpected callback state: %+v", callback) } @@ -48,17 +48,16 @@ func TestConnectChannelCallbacksConnectedState(t *testing.T) { } func TestConnectChannelCallbacksFailedState(t *testing.T) { - var callback ConnectionStateCallback - api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if err := json.NewDecoder(r.Body).Decode(&callback); err != nil { - t.Fatalf("decode callback: %v", err) - } - w.WriteHeader(http.StatusOK) - })) - defer api.Close() - - handler := handlerWithDial(api.URL+"/api", func(context.Context, ConnectChannelCommand) error { - return errTestDial + handler := handlerWithConnect(func(context.Context, ConnectChannelCommand) (ConnectionStateCallback, error) { + return ConnectionStateCallback{ + ChannelID: "channel-1", + ConnectionID: "channel-1:primary", + Status: "failed", + DesiredConnections: 2, + CurrentConnections: 0, + LastDisconnectedAt: time.Now().UTC().Format(time.RFC3339Nano), + LastError: errTestConnect.Error(), + }, nil }) resp := httptest.NewRecorder() @@ -68,14 +67,18 @@ func TestConnectChannelCallbacksFailedState(t *testing.T) { if resp.Code != http.StatusOK { t.Fatalf("unexpected response status: %d body=%s", resp.Code, resp.Body.String()) } + var callback ConnectionStateCallback + if err := json.Unmarshal(resp.Body.Bytes(), &callback); err != nil { + t.Fatalf("decode response: %v", err) + } if callback.Status != "failed" || callback.CurrentConnections != 0 || callback.LastError == "" { t.Fatalf("unexpected callback state: %+v", callback) } } func TestConnectChannelRejectsInvalidCommand(t *testing.T) { - handler := handlerWithDial("", func(context.Context, ConnectChannelCommand) error { - return nil + handler := handlerWithConnect(func(context.Context, ConnectChannelCommand) (ConnectionStateCallback, error) { + return ConnectionStateCallback{}, nil }) resp := httptest.NewRecorder() @@ -180,16 +183,16 @@ func TestRecoveryOverviewEndpointReturnsCombinedView(t *testing.T) { } } -type testDialError struct{} +type testConnectError struct{} -func (testDialError) Error() string { - return "dial failed" +func (testConnectError) Error() string { + return "connect failed" } -var errTestDial testDialError +var errTestConnect testConnectError -func handlerWithDial(apiBaseURL string, dial DialFunc) http.Handler { - return handlerWithServer(Server{APIBaseURL: apiBaseURL, Dial: dial}) +func handlerWithConnect(connect ConnectFunc) http.Handler { + return handlerWithServer(Server{Connect: connect}) } func handlerWithServer(server Server) http.Handler { diff --git a/gateway/internal/upstream/connection_loss_test.go b/gateway/internal/upstream/connection_loss_test.go index db08bf3..c1dfd46 100644 --- a/gateway/internal/upstream/connection_loss_test.go +++ b/gateway/internal/upstream/connection_loss_test.go @@ -1,14 +1,26 @@ package upstream import ( + "context" "errors" "testing" ) func TestHandleConnectionLossNotifiesPendingSubmitters(t *testing.T) { + var reported ConnectionState conn := &connection{ pending: make(map[uint32]chan submitPartResponse), + pool: &connectionPool{ + channelID: "channel-1", + connectionID: "channel-1:primary", + config: normalizeUpstreamConfig(queueUpstreamConfigForTest()), + reporter: func(_ context.Context, state ConnectionState) error { + reported = state + return nil + }, + }, } + conn.pool.conns = []*connection{conn} waiter := make(chan submitPartResponse, 1) conn.pending[7] = waiter @@ -30,6 +42,9 @@ func TestHandleConnectionLossNotifiesPendingSubmitters(t *testing.T) { if len(conn.pending) != 0 { t.Fatalf("expected pending map to be reset, got %d entries", len(conn.pending)) } + if reported.Status != "disconnected" || reported.CurrentConnections != 0 { + t.Fatalf("unexpected reported state: %+v", reported) + } } func TestTemporaryReadTimeoutDetection(t *testing.T) { diff --git a/gateway/internal/upstream/manager.go b/gateway/internal/upstream/manager.go index 1b0755f..c0f900d 100644 --- a/gateway/internal/upstream/manager.go +++ b/gateway/internal/upstream/manager.go @@ -33,6 +33,19 @@ type Manager struct { conns map[string]*connectionPool } +type ConnectionState struct { + ChannelID string `json:"channelId"` + ConnectionID string `json:"connectionId"` + Status string `json:"status"` + DesiredConnections int `json:"desiredConnections"` + CurrentConnections int `json:"currentConnections"` + LastConnectedAt string `json:"lastConnectedAt,omitempty"` + LastDisconnectedAt string `json:"lastDisconnectedAt,omitempty"` + LastHeartbeatAt string `json:"lastHeartbeatAt,omitempty"` + ReconnectCount int `json:"reconnectCount,omitempty"` + LastError string `json:"lastError,omitempty"` +} + func (m *Manager) Submit(ctx context.Context, cmd queue.SubmitCommand) (queue.SubmitResult, error) { if err := validateSubmitCommand(cmd); err != nil { result := submitResult(cmd, 0, "", "rejected", "INVALID_COMMAND", err.Error()) @@ -64,28 +77,54 @@ func (m *Manager) Submit(ctx context.Context, cmd queue.SubmitCommand) (queue.Su return result, nil } +func (m *Manager) ConnectChannel(ctx context.Context, command queue.ConnectChannelCommand) (ConnectionState, error) { + if err := validateConnectChannelCommand(command); err != nil { + return ConnectionState{}, err + } + + m.mu.Lock() + m.ensureDefaultsLocked() + pool := m.conns[command.ChannelID] + config := normalizeUpstreamConfig(queue.UpstreamConfig{ + GatewayHost: command.Channel.GatewayHost, + GatewayPort: command.Channel.GatewayPort, + Account: command.Channel.Account, + PasswordCipher: command.Channel.PasswordCipher, + CMPPVersion: command.Channel.CMPPVersion, + DesiredConnections: command.DesiredConnections, + WindowSize: 16, + }) + if pool == nil || !pool.matches(config) { + if pool != nil { + pool.close() + } + pool = m.newConnectionPool(command.ChannelID, command.ConnectionID, config) + m.conns[command.ChannelID] = pool + } + m.mu.Unlock() + + if err := pool.ensureConnected(); err != nil { + _ = pool.reportState(ctx, "failed", err) + m.mu.Lock() + delete(m.conns, command.ChannelID) + m.mu.Unlock() + return pool.snapshotState("failed", err), nil + } + return pool.snapshotState("connected", nil), nil +} + func (m *Manager) connectionFor(cmd queue.SubmitCommand) (*connectionPool, error) { m.mu.Lock() defer m.mu.Unlock() - if m.HTTPClient == nil { - m.HTTPClient = &http.Client{Timeout: defaultHTTPTimeout} - } - if m.conns == nil { - m.conns = make(map[string]*connectionPool) - } + m.ensureDefaultsLocked() pool := m.conns[cmd.ChannelID] if pool == nil || !pool.matches(cmd.Upstream) { if pool != nil { pool.close() } - pool = &connectionPool{ - channelID: cmd.ChannelID, - config: normalizeUpstreamConfig(cmd.Upstream), - apiBaseURL: m.APIBaseURL, - httpClient: m.HTTPClient, - } + pool = m.newConnectionPool(cmd.ChannelID, defaultChannelConnectionID(cmd.ChannelID), normalizeUpstreamConfig(cmd.Upstream)) m.conns[cmd.ChannelID] = pool } if err := pool.ensureConnected(); err != nil { @@ -103,11 +142,35 @@ func (m *Manager) post(ctx context.Context, path string, payload any) error { return postJSON(ctx, client, m.APIBaseURL, path, payload) } +func (m *Manager) ensureDefaultsLocked() { + if m.HTTPClient == nil { + m.HTTPClient = &http.Client{Timeout: defaultHTTPTimeout} + } + if m.conns == nil { + m.conns = make(map[string]*connectionPool) + } +} + +func (m *Manager) newConnectionPool(channelID string, connectionID string, config queue.UpstreamConfig) *connectionPool { + return &connectionPool{ + channelID: channelID, + connectionID: connectionID, + config: config, + apiBaseURL: m.APIBaseURL, + httpClient: m.HTTPClient, + reporter: func(ctx context.Context, state ConnectionState) error { + return m.post(ctx, "/admin/gateway/connections", state) + }, + } +} + type connectionPool struct { - channelID string - config queue.UpstreamConfig - apiBaseURL string - httpClient *http.Client + channelID string + connectionID string + config queue.UpstreamConfig + apiBaseURL string + httpClient *http.Client + reporter func(context.Context, ConnectionState) error mu sync.Mutex conns []*connection @@ -119,19 +182,23 @@ func (p *connectionPool) matches(config queue.UpstreamConfig) bool { } func (p *connectionPool) ensureConnected() error { - p.mu.Lock() - defer p.mu.Unlock() - desired := p.config.DesiredConnections if desired <= 0 { desired = 1 } - for len(p.conns) < desired { + connectedAny := false + for { + p.mu.Lock() + if len(p.conns) >= desired { + p.mu.Unlock() + break + } index := len(p.conns) conn := &connection{ channelID: p.channelID, config: p.config, index: index, + pool: p, apiBaseURL: p.apiBaseURL, httpClient: p.httpClient, window: make(chan struct{}, p.config.WindowSize), @@ -139,12 +206,22 @@ func (p *connectionPool) ensureConnected() error { tracker: make(map[uint64]queue.SubmitCommand), longUplink: make(map[string]*longUplinkAssembly), } - if err := conn.ensureConnected(); err != nil { + p.mu.Unlock() + + connected, err := conn.ensureConnected() + if err != nil { conn.close() - p.closeLocked() + p.close() return err } + + p.mu.Lock() p.conns = append(p.conns, conn) + p.mu.Unlock() + connectedAny = connectedAny || connected + } + if connectedAny { + _ = p.reportState(context.Background(), "connected", nil) } return nil } @@ -198,7 +275,7 @@ func (p *connectionPool) acquireConnection(ctx context.Context) (*connection, fu for { if conn, release := p.tryAcquireConnection(); conn != nil { - if err := conn.ensureConnected(); err != nil { + if connected, err := conn.ensureConnected(); err != nil { release() select { case <-waitCtx.Done(): @@ -206,6 +283,8 @@ func (p *connectionPool) acquireConnection(ctx context.Context) (*connection, fu case <-ticker.C: continue } + } else if connected { + _ = p.reportState(context.Background(), "connected", nil) } return conn, release, nil } @@ -236,21 +315,66 @@ func (p *connectionPool) tryAcquireConnection() (*connection, func()) { func (p *connectionPool) close() { p.mu.Lock() - defer p.mu.Unlock() - p.closeLocked() -} - -func (p *connectionPool) closeLocked() { - for _, conn := range p.conns { + conns := p.conns + p.conns = nil + p.mu.Unlock() + for _, conn := range conns { conn.close() } - p.conns = nil +} + +func (p *connectionPool) reportState(ctx context.Context, status string, stateErr error) error { + if p.reporter == nil { + return nil + } + return p.reporter(ctx, p.snapshotState(status, stateErr)) +} + +func (p *connectionPool) snapshotState(status string, stateErr error) ConnectionState { + now := time.Now().UTC().Format(time.RFC3339Nano) + state := ConnectionState{ + ChannelID: p.channelID, + ConnectionID: p.connectionID, + Status: status, + DesiredConnections: p.config.DesiredConnections, + CurrentConnections: p.countActiveConnections(), + } + if state.DesiredConnections <= 0 { + state.DesiredConnections = 1 + } + switch status { + case "connected": + state.LastConnectedAt = now + state.LastHeartbeatAt = now + case "disconnected", "failed": + state.LastDisconnectedAt = now + } + if stateErr != nil { + state.LastError = stateErr.Error() + } + return state +} + +func (p *connectionPool) countActiveConnections() int { + p.mu.Lock() + defer p.mu.Unlock() + count := 0 + for _, conn := range p.conns { + conn.mu.Lock() + active := conn.client != nil && !conn.closed + conn.mu.Unlock() + if active { + count += 1 + } + } + return count } type connection struct { channelID string config queue.UpstreamConfig index int + pool *connectionPool apiBaseURL string httpClient *http.Client @@ -274,23 +398,23 @@ func (c *connection) matches(config queue.UpstreamConfig) bool { return c.config == normalizeUpstreamConfig(config) } -func (c *connection) ensureConnected() error { +func (c *connection) ensureConnected() (bool, error) { c.mu.Lock() defer c.mu.Unlock() if c.client != nil && !c.closed { - return nil + return false, nil } client := cmpp.NewClient(protocolVersion(c.config.CMPPVersion)) addr := fmt.Sprintf("%s:%d", c.config.GatewayHost, c.config.GatewayPort) if err := client.Connect(addr, c.config.Account, c.config.PasswordCipher, defaultConnectTimeout); err != nil { client.Disconnect() - return err + return false, err } c.client = client c.closed = false go c.readLoop() - return nil + return true, nil } func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, part submitPart) (uint32, string, queue.SubmitResult, error) { @@ -540,6 +664,13 @@ func (c *connection) handleConnectionLoss(err error) { default: } } + if c.pool != nil { + status := "disconnected" + if c.pool.countActiveConnections() > 0 { + status = "connected" + } + _ = c.pool.reportState(context.Background(), status, err) + } } func submitResult(cmd queue.SubmitCommand, sequenceID uint32, gatewayMessageID string, status string, code string, message string) queue.SubmitResult { @@ -599,6 +730,26 @@ func validateSubmitCommand(cmd queue.SubmitCommand) error { return nil } +func validateConnectChannelCommand(command queue.ConnectChannelCommand) error { + if command.MessageType != queue.MessageTypeConnectChannel { + return fmt.Errorf("unsupported messageType %q", command.MessageType) + } + if command.ChannelID == "" || command.ConnectionID == "" { + return fmt.Errorf("channelId and connectionId are required") + } + if command.Channel.GatewayHost == "" || command.Channel.GatewayPort <= 0 { + return fmt.Errorf("gatewayHost and gatewayPort are required") + } + if command.Channel.Account == "" || command.Channel.PasswordCipher == "" { + return fmt.Errorf("account and passwordCipher are required") + } + return nil +} + +func defaultChannelConnectionID(channelID string) string { + return fmt.Sprintf("%s:primary", channelID) +} + func normalizeUpstreamConfig(config queue.UpstreamConfig) queue.UpstreamConfig { if config.DesiredConnections <= 0 { config.DesiredConnections = 1 diff --git a/src/apps/LoginPage.tsx b/src/apps/LoginPage.tsx index 283f37a..c1f3a0b 100644 --- a/src/apps/LoginPage.tsx +++ b/src/apps/LoginPage.tsx @@ -73,7 +73,7 @@ export function LoginPage({ portal }: LoginPageProps) {
- setLogin(event.target.value)} placeholder="请输入邮箱或手机号" value={login} /> + setLogin(event.target.value)} placeholder="请输入用户名、邮箱或手机号" value={login} /> setPassword(event.target.value)} placeholder="请输入密码" type="password" value={password} />
setCaptchaText(event.target.value)} placeholder="请输入计算结果" value={captchaText} /> diff --git a/src/apps/admin/AdminUsersPage.tsx b/src/apps/admin/AdminUsersPage.tsx index eefd30d..b1d5638 100644 --- a/src/apps/admin/AdminUsersPage.tsx +++ b/src/apps/admin/AdminUsersPage.tsx @@ -195,7 +195,7 @@ export function AdminUsersPage() { updateField('displayName', event.target.value)} value={form.displayName} /> updateField('email', event.target.value)} value={form.email} /> updateField('phone', event.target.value)} value={form.phone} /> - updateField('username', event.target.value)} value={form.username} /> + updateField('username', event.target.value)} value={form.username} /> updateField('displayName', event.target.value)} placeholder="请输入用户名" value={form.displayName} /> updateField('email', event.target.value)} placeholder="请输入邮箱" value={form.email} /> updateField('phone', event.target.value)} placeholder="请输入手机号" value={form.phone} /> - updateField('username', event.target.value)} value={form.username} /> + updateField('username', event.target.value)} value={form.username} /> {creating ? updateField('password', event.target.value)} type="password" value={form.password} /> : null}