From 0efcfbb8b6a5dbe2a70b0c3b80326bd01a40009c Mon Sep 17 00:00:00 2001 From: Nixevol Date: Wed, 30 Sep 2026 07:56:25 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8E=A5=E7=BA=BF=20serve=20=E7=9C=9F?= =?UTF-8?q?=E5=AE=9E=20auth/=E6=B3=A8=E5=86=8C/=E7=AE=A1=E7=90=86/MQTT=20?= =?UTF-8?q?=E4=B8=8E=E6=B6=88=E6=81=AF=E5=BE=AA=E7=8E=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/nixmsg/serve.go | 315 +++++++++++++++++++++------- cmd/nixmsg/uplink.go | 83 ++++++++ cmd/nixmsg/wire.go | 12 +- cmd/nixmsg/wire_integration_test.go | 313 +++++++++++++++++++++++++++ docs/DEVIATIONS.md | 23 ++ internal/app/message/conn.go | 11 + internal/broker/broker.go | 7 + internal/broker/hooks.go | 10 + internal/broker/session.go | 5 + 9 files changed, 699 insertions(+), 80 deletions(-) create mode 100644 cmd/nixmsg/uplink.go create mode 100644 cmd/nixmsg/wire_integration_test.go diff --git a/cmd/nixmsg/serve.go b/cmd/nixmsg/serve.go index f625b0d..8ee23fe 100644 --- a/cmd/nixmsg/serve.go +++ b/cmd/nixmsg/serve.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "io/fs" "log/slog" "net" "net/http" @@ -14,7 +15,18 @@ import ( "syscall" "time" + "git.asio.asia/nixevol/NixMsg/internal/admin" + "git.asio.asia/nixevol/NixMsg/internal/app/group" + "git.asio.asia/nixevol/NixMsg/internal/app/identity" + "git.asio.asia/nixevol/NixMsg/internal/app/message" + "git.asio.asia/nixevol/NixMsg/internal/app/port" + "git.asio.asia/nixevol/NixMsg/internal/app/presence" + "git.asio.asia/nixevol/NixMsg/internal/auth" + "git.asio.asia/nixevol/NixMsg/internal/broker" "git.asio.asia/nixevol/NixMsg/internal/config" + "git.asio.asia/nixevol/NixMsg/internal/httpx" + "git.asio.asia/nixevol/NixMsg/internal/listener" + "git.asio.asia/nixevol/NixMsg/internal/metrics" "git.asio.asia/nixevol/NixMsg/internal/store" "git.asio.asia/nixevol/NixMsg/web" ) @@ -25,20 +37,16 @@ func cmdServe(_ []string) error { if err != nil { return err } - // P-WIRE-BEGIN if err := cfg.Validate(); err != nil { return err } setupJSONLogger(cfg.Log) - // P-WIRE-END ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() return runServe(ctx, cfg) } func runServe(ctx context.Context, cfg config.Config) error { - deps := wire() - if err := os.MkdirAll(cfg.DataDir, 0o755); err != nil { return fmt.Errorf("mkdir data_dir: %w", err) } @@ -49,7 +57,6 @@ func runServe(ctx context.Context, cfg config.Config) error { } defer func() { _ = db.Close() }() - // P-WIRE-BEGIN ok, err := store.HasAdminPassword(ctx, db.Write) if err != nil { return err @@ -57,87 +64,250 @@ func runServe(ctx context.Context, cfg config.Config) error { if !ok { return errors.New("admin password not initialized; run: nixmsg admin init") } - // P-WIRE-END - // 启动恢复入口已挂上（假实现为空操作）；M 线替换 message.Service 后生效。 - if recoverErr := deps.Messages.RecoverOnStart(ctx); recoverErr != nil { + hashPool := auth.NewPool() + sessionTokens := auth.NewSessionTokens() + apiTokens := auth.NewAPITokens() + loginLocks := auth.NewLoginLocks() + memConns := message.NewMemoryConns() + msgLim := message.LimitsFromFullConfig(cfg) + + login := broker.NewLogin(broker.LoginOptions{ + DB: db, + Pool: hashPool, + Tokens: sessionTokens, + Locks: loginLocks, + IdleDays: cfg.SessionIdleDays, + }) + + // 先用空下行建 message，拿到 App 指针；broker 建好后再 WithDownlink 重建并回填 uplink.msg。 + msgApp := message.New(db, msgLim, hashPool, + message.WithLocks(loginLocks), + message.WithConnRegistry(memConns), + ) + uplink := &appUplink{ + msg: msgApp, + conns: memConns, + next: port.StubUplinkHandler{}, + log: slog.Default(), + } + sess := broker.NewSession(broker.SessionOptions{ + Login: login, + Inner: uplink, + Limits: broker.HelloLimits{ + MaxBodyBytes: cfg.Limits.MaxBodyBytes, + MaxMetaBytes: cfg.Limits.MaxMetaBytes, + MaxFrameBytes: cfg.Limits.MaxFrameBytes, + MaxTTLSeconds: int64(cfg.Limits.MaxTTLSeconds), + MaxScheduleSeconds: int64(cfg.Limits.MaxScheduleSeconds), + AckTimeoutSeconds: int64(cfg.Limits.AckTimeoutSeconds), + ServerVersion: Version, + }, + Logger: slog.Default(), + }) + + brk, err := broker.New(broker.Options{ + Authenticator: login, + Uplink: sess, + Logger: slog.Default(), + OnPublishDropped: func(dropCtx context.Context, endpointID string, connID port.ConnID, payload []byte) { + if dropErr := msgApp.OnPublishDropped(dropCtx, endpointID, connID, payload); dropErr != nil { + slog.Error("on publish dropped", "endpoint", endpointID, "err", dropErr) + } + }, + }) + if err != nil { + return fmt.Errorf("broker: %w", err) + } + defer func() { _ = brk.Close() }() + sess.Attach(brk) + + msgApp = message.New(db, msgLim, hashPool, + message.WithLocks(loginLocks), + message.WithConnRegistry(memConns), + message.WithDownlink(brk), + ) + uplink.msg = msgApp + + presApp := presence.New(presence.Config{ + DB: db, + Downlink: brk, + Conns: &presenceConnTable{conns: memConns}, + }) + sess.SetPresence(presApp) + + idApp := identity.New(identity.Config{ + DB: db, + Hash: hashPool, + Locks: loginLocks, + Sessions: sessionTokens, + MaxScheduleSeconds: int64(cfg.Limits.MaxScheduleSeconds), + Logger: slog.Default(), + ConnControl: brk, + }) + _ = group.New(group.Config{ + DB: db, + Talk: idApp, + Online: presApp, + Downlink: brk, + MaxGroupMembers: cfg.Limits.MaxGroupMembers, + }) + + if recoverErr := msgApp.RecoverOnStart(ctx); recoverErr != nil { return fmt.Errorf("message recover: %w", recoverErr) } - // 其余 deps 供后续 admin / broker / httpx 接线；此处显式引用避免未使用告警。 - _ = deps.Identity - _ = deps.Groups - _ = deps.Presence - _ = deps.Downlink - _ = deps.Conns - _ = deps.Uplink - _ = deps.HashPool - _ = deps.SessionTokens - _ = deps.APITokens - _ = deps.LoginLocks - mux := http.NewServeMux() - mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { - w.WriteHeader(http.StatusOK) - _, _ = w.Write([]byte("ok")) + trustedNets := httpx.ParseCIDRs(cfg.TrustedProxies) + adminHandler := admin.New(admin.Deps{ + DB: db, + Hash: hashPool, + Tokens: apiTokens, + Locks: loginLocks, + Logger: slog.Default(), + TrustedProxies: trustedNets, + KickEndpoint: func(kickCtx context.Context, endpointID string) (bool, error) { + if _, found := brk.ConnInfoOf(endpointID); !found { + return false, nil + } + if kickErr := sess.Kick(kickCtx, endpointID); kickErr != nil { + return false, kickErr + } + return true, nil + }, }) - // P-WIRE-BEGIN - mux.HandleFunc("GET /readyz", func(w http.ResponseWriter, r *http.Request) { - if readyErr := db.Ready(r.Context()); readyErr != nil { - slog.Error("readyz failed", "err", readyErr) - http.Error(w, "not ready", http.StatusServiceUnavailable) - return + + metricsReg := metrics.New() + buildHandlers := func(proxies *listener.ProxySet) listener.Handlers { + return listener.Handlers{ + MQTT: brk.WSHandler(proxies), + ClientAPI: idApp.Handler(), + AdminAPI: adminHandler, + Metrics: metricsReg.Handler(), + MetricsToken: cfg.Metrics.Token, + Static: staticFileHandler(), + Healthz: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok")) + }), + Readyz: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if readyErr := db.Ready(r.Context()); readyErr != nil { + slog.Error("readyz failed", "err", readyErr) + http.Error(w, "not ready", http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok")) + }), } - w.WriteHeader(http.StatusOK) - _, _ = w.Write([]byte("ok")) - }) - // P-WIRE-END - // 确保前端资源被链接进二进制；完整静态托管由后续任务完善。 - _ = web.Dist() - - ln, err := net.Listen("tcp", cfg.Listen) - if err != nil { - return fmt.Errorf("listen %s: %w", cfg.Listen, err) } - if err := writeListenAddr(cfg.DataDir, ln.Addr().String()); err != nil { - _ = ln.Close() + handlers := buildHandlers(nil) + shared := cfg.AdminListen == "" + var clientHandler, adminHTTP http.Handler + if shared { + clientHandler = listener.NewMux(listener.RoleShared, handlers) + } else { + clientHandler = listener.NewMux(listener.RoleClient, handlers) + adminHTTP = listener.NewMux(listener.RoleAdmin, handlers) + } + + lnOpts := listener.Options{ + Listen: cfg.Listen, + AdminListen: cfg.AdminListen, + DataDir: cfg.DataDir, + CertFile: cfg.TLS.CertFile, + KeyFile: cfg.TLS.KeyFile, + AllowPlaintext: cfg.TLS.AllowPlaintext, + TrustedProxies: cfg.TrustedProxies, + ClientHandler: clientHandler, + AdminHandler: adminHTTP, + OnMQTT: func(conn net.Conn) { + _ = brk.AttachTCP(conn) + }, + Logger: slog.Default(), + } + lnSrv, err := listener.New(lnOpts) + if err != nil { + return fmt.Errorf("listener: %w", err) + } + handlers = buildHandlers(lnSrv.Proxies()) + if shared { + lnOpts.ClientHandler = listener.NewMux(listener.RoleShared, handlers) + } else { + lnOpts.ClientHandler = listener.NewMux(listener.RoleClient, handlers) + lnOpts.AdminHandler = listener.NewMux(listener.RoleAdmin, handlers) + } + lnSrv, err = listener.New(lnOpts) + if err != nil { + return fmt.Errorf("listener: %w", err) + } + + if err := lnSrv.Start(ctx); err != nil { return err } + defer func() { _ = lnSrv.Close() }() - srv := &http.Server{ - Handler: mux, - ReadHeaderTimeout: 10 * time.Second, + if err := writeListenAddr(cfg.DataDir, lnSrv.ListenAddr()); err != nil { + return err + } + if cfg.AdminListen != "" && lnSrv.AdminAddr() != "" { + if err := os.WriteFile(filepath.Join(cfg.DataDir, "admin.addr"), []byte(lnSrv.AdminAddr()+"\n"), 0o644); err != nil { + return err + } } - errCh := make(chan error, 1) - go func() { - errCh <- srv.Serve(ln) - }() + loopCtx, loopCancel := context.WithCancel(ctx) + defer loopCancel() + go messageLoops(loopCtx, msgApp, memConns) - select { - case <-ctx.Done(): - // P-WIRE-BEGIN - // 先停止接受新连接，再等写队列最多 10 秒，然后断开并退出。 - shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) - defer cancel() - _ = srv.Shutdown(shutdownCtx) - drainCtx, drainCancel := context.WithTimeout(context.Background(), 10*time.Second) - defer drainCancel() - if drainErr := db.Queue.Drain(drainCtx); drainErr != nil && !errors.Is(drainErr, context.DeadlineExceeded) { - slog.Error("write queue drain", "err", drainErr) - } - // P-WIRE-END - serveErr := <-errCh - if serveErr != nil && !errors.Is(serveErr, http.ErrServerClosed) { - return serveErr - } - return nil - case serveErr := <-errCh: - if errors.Is(serveErr, http.ErrServerClosed) { - return nil - } - return serveErr + <-ctx.Done() + loopCancel() + _ = lnSrv.Close() + drainCtx, drainCancel := context.WithTimeout(context.Background(), 10*time.Second) + defer drainCancel() + if drainErr := db.Queue.Drain(drainCtx); drainErr != nil && !errors.Is(drainErr, context.DeadlineExceeded) { + slog.Error("write queue drain", "err", drainErr) } + return nil +} + +func messageLoops(ctx context.Context, msgApp *message.App, conns *message.MemoryConns) { + t := time.NewTicker(time.Second) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + nowMs := time.Now().UnixMilli() + if _, err := msgApp.DispatchDue(ctx, nowMs, 100); err != nil { + slog.Error("dispatch due", "err", err) + } + for ep, live := range conns.Snapshot() { + if err := msgApp.PushPending(ctx, ep, live.ConnID); err != nil { + slog.Debug("push pending", "endpoint", ep, "err", err) + } + } + if err := msgApp.CleanupOnce(ctx, nowMs); err != nil { + slog.Error("cleanup once", "err", err) + } + } + } +} + +func staticFileHandler() http.Handler { + root := web.Dist() + for _, prefix := range []string{"dist", "stub"} { + sub, err := fs.Sub(root, prefix) + if err != nil { + continue + } + if f, err := sub.Open("index.html"); err == nil { + _ = f.Close() + return http.FileServer(http.FS(sub)) + } + } + return http.FileServer(http.FS(root)) } func writeListenAddr(dataDir, addr string) error { @@ -145,7 +315,6 @@ func writeListenAddr(dataDir, addr string) error { return os.WriteFile(path, []byte(addr+"\n"), 0o644) } -// P-WIRE-BEGIN func setupJSONLogger(cfg config.LogConfig) { level := slog.LevelInfo switch strings.ToLower(strings.TrimSpace(cfg.Level)) { @@ -159,5 +328,3 @@ func setupJSONLogger(cfg config.LogConfig) { h := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: level}) slog.SetDefault(slog.New(h)) } - -// P-WIRE-END diff --git a/cmd/nixmsg/uplink.go b/cmd/nixmsg/uplink.go new file mode 100644 index 0000000..e9631a7 --- /dev/null +++ b/cmd/nixmsg/uplink.go @@ -0,0 +1,83 @@ +package main + +import ( + "context" + "log/slog" + + "git.asio.asia/nixevol/NixMsg/internal/app/message" + "git.asio.asia/nixevol/NixMsg/internal/app/port" +) + +// appUplink 把 broker 生命周期接到消息连接表与投递推送。 +// 其余业务上行帧暂转交 next（可为空 Stub）；完整 HandleUplink 分发留后续波次。 +type appUplink struct { + msg *message.App + conns *message.MemoryConns + next port.UplinkHandler + log *slog.Logger +} + +func (u *appUplink) OnSessionEstablished(ctx context.Context, conn port.ConnInfo) error { + u.conns.Set(conn.EndpointID, message.LiveConn{ + ConnID: conn.ConnID, + MaxPacketSize: conn.MaxPacketSize, + }) + if u.next != nil { + return u.next.OnSessionEstablished(ctx, conn) + } + return nil +} + +func (u *appUplink) OnHandshakeComplete(ctx context.Context, hs port.HandshakeInfo) error { + live := message.LiveConn{ + ConnID: hs.ConnID, + MaxReceiveBytes: hs.MaxReceiveBytes, + MaxPacketSize: hs.MaxPacketSize, + } + u.conns.Set(hs.EndpointID, live) + if err := u.msg.OnHandshakeComplete(ctx, hs.EndpointID, live); err != nil { + u.log.Error("message handshake", "endpoint", hs.EndpointID, "err", err) + return err + } + if u.next != nil { + return u.next.OnHandshakeComplete(ctx, hs) + } + return nil +} + +func (u *appUplink) OnDisconnect(ctx context.Context, conn port.ConnInfo, reason port.DisconnectReason) { + live, ok := u.conns.Current(conn.EndpointID) + isCurrent := ok && live.ConnID == conn.ConnID + if err := u.msg.OnDisconnect(ctx, conn.EndpointID, conn.ConnID, isCurrent); err != nil { + u.log.Error("message disconnect", "endpoint", conn.EndpointID, "err", err) + } + u.conns.Clear(conn.EndpointID, conn.ConnID) + if u.next != nil { + u.next.OnDisconnect(ctx, conn, reason) + } +} + +func (u *appUplink) HandleUplink(ctx context.Context, conn port.ConnInfo, payload []byte) error { + if u.next != nil { + return u.next.HandleUplink(ctx, conn, payload) + } + return nil +} + +// presenceConnTable 把消息连接表暴露给 presence.ConnTable。 +type presenceConnTable struct { + conns *message.MemoryConns +} + +func (p *presenceConnTable) IsOnline(endpointID string) bool { + _, ok := p.conns.Current(endpointID) + return ok +} + +func (p *presenceConnTable) CurrentConn(endpointID string) (port.ConnID, bool) { + live, ok := p.conns.Current(endpointID) + if !ok { + return "", false + } + return live.ConnID, true +} diff --git a/cmd/nixmsg/wire.go b/cmd/nixmsg/wire.go index 283d6b0..0fed93a 100644 --- a/cmd/nixmsg/wire.go +++ b/cmd/nixmsg/wire.go @@ -9,7 +9,7 @@ import ( "git.asio.asia/nixevol/NixMsg/internal/auth" ) -// appDeps 是 serve 组装出的模块依赖。各线在后续任务中替换假实现为真实实现。 +// appDeps 是组装出的模块依赖（测试与骨架仍可调用 wire）。 type appDeps struct { HashPool auth.HashPool SessionTokens auth.SessionTokens @@ -26,13 +26,13 @@ type appDeps struct { Conns port.ConnControl } -// wire 组装各业务模块的骨架依赖（T0.4：假实现；后续各线替换）。 +// wire 返回真实 auth 实现与空业务桩（无 DB 时的轻量装配；正式 serve 走 assembleRuntime）。 func wire() appDeps { return appDeps{ - HashPool: auth.NewStubHashPool(), - SessionTokens: auth.NewStubSessionTokens(), - APITokens: auth.NewStubAPITokens(), - LoginLocks: auth.NewStubLoginLocks(), + HashPool: auth.NewPool(), + SessionTokens: auth.NewSessionTokens(), + APITokens: auth.NewAPITokens(), + LoginLocks: auth.NewLoginLocks(), Messages: message.NewStub(), Identity: identity.NewStub(), diff --git a/cmd/nixmsg/wire_integration_test.go b/cmd/nixmsg/wire_integration_test.go new file mode 100644 index 0000000..a4fb909 --- /dev/null +++ b/cmd/nixmsg/wire_integration_test.go @@ -0,0 +1,313 @@ +package main + +import ( + "bytes" + "context" + "database/sql" + "encoding/json" + "io" + "net/http" + "net/http/cookiejar" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "git.asio.asia/nixevol/NixMsg/internal/config" + "git.asio.asia/nixevol/NixMsg/internal/protocol" + "git.asio.asia/nixevol/NixMsg/internal/store" + "git.asio.asia/nixevol/NixMsg/test/harness" + "github.com/mochi-mqtt/server/v2/packets" +) + +func TestWireAdminLoginRegisterMQTTHandshake(t *testing.T) { + dataDir := t.TempDir() + cfgPath := writeTestConfig(t, dataDir) + initAdminForTest(t, dataDir) + enableRegistration(t, dataDir, "wire-code-99") + + cfg, err := config.Load(cfgPath) + if err != nil { + t.Fatal(err) + } + if vErr := cfg.Validate(); vErr != nil { + t.Fatal(vErr) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + errCh := make(chan error, 1) + go func() { errCh <- runServe(ctx, cfg) }() + + addr := waitListenAddr(t, dataDir, 15*time.Second) + base := "http://" + addr + + // 1) admin init 后真实进程可管理登录 + jar, err := cookiejar.New(nil) + if err != nil { + t.Fatal(err) + } + client := &http.Client{Jar: jar, Timeout: 10 * time.Second} + loginBody, _ := json.Marshal(map[string]string{ + "username": "admin", + "password": "test-admin-password-xx", + }) + resp, err := client.Post(base+"/api/admin/login", "application/json", bytes.NewReader(loginBody)) + if err != nil { + t.Fatalf("admin login: %v", err) + } + body, _ := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("admin login status=%d body=%s", resp.StatusCode, body) + } + var loginEnv struct { + OK bool `json:"ok"` + } + if uErr := json.Unmarshal(body, &loginEnv); uErr != nil || !loginEnv.OK { + t.Fatalf("admin login resp=%s", body) + } + + // 2) 已写入注册开关与安全码后可注册 + regBody := `{"registration_code":"wire-code-99","id":"ep_wire1","login_password":"password12","name":"接线端"}` + regResp, err := http.Post(base+"/api/client/register", "application/json", strings.NewReader(regBody)) + if err != nil { + t.Fatalf("register: %v", err) + } + regBytes, _ := io.ReadAll(regResp.Body) + _ = regResp.Body.Close() + if regResp.StatusCode != http.StatusOK { + t.Fatalf("register status=%d body=%s", regResp.StatusCode, regBytes) + } + var regEnv struct { + OK bool `json:"ok"` + Data struct { + ID string `json:"id"` + } `json:"data"` + } + if uErr := json.Unmarshal(regBytes, ®Env); uErr != nil || !regEnv.OK || regEnv.Data.ID != "ep_wire1" { + t.Fatalf("register resp=%s", regBytes) + } + + // 3) 注册出的端用密码完成 MQTT 握手并拿到 session_token + tok := mqttPasswordHandshake(t, base, "ep_wire1", "password12") + if tok == "" || !strings.HasPrefix(tok, protocol.SessionTokenPrefix) { + t.Fatalf("session_token=%q", tok) + } + + cancel() + select { + case err := <-errCh: + if err != nil { + t.Fatalf("serve exit: %v", err) + } + case <-time.After(15 * time.Second): + t.Fatal("serve did not stop") + } +} + +func mqttPasswordHandshake(t *testing.T, httpBase, endpointID, password string) string { + t.Helper() + mc, err := harness.DialMQTTWebSocket(httpBase, 10*time.Second) + if err != nil { + t.Fatalf("dial mqtt ws: %v", err) + } + defer func() { _ = mc.Close() }() + + pk := packets.Packet{ + FixedHeader: packets.FixedHeader{Type: packets.Connect}, + ProtocolVersion: 5, + Connect: packets.ConnectParams{ + ProtocolName: []byte("MQTT"), + Clean: true, + ClientIdentifier: endpointID, + Keepalive: 30, + UsernameFlag: true, + Username: []byte(endpointID), + PasswordFlag: true, + Password: []byte(password), + }, + } + var buf bytes.Buffer + if encErr := pk.ConnectEncode(&buf); encErr != nil { + t.Fatal(encErr) + } + if sendErr := mc.Send(buf.Bytes()); sendErr != nil { + t.Fatal(sendErr) + } + ack, err := mc.Recv() + if err != nil { + t.Fatalf("connack: %v", err) + } + if len(ack) < 2 || ack[0]>>4 != packets.Connack { + t.Fatalf("want CONNACK, got %x", ack) + } + // MQTT5 CONNACK: remaining length, flags, reason code + reason := byte(0) + if len(ack) >= 4 { + reason = ack[3] + } + if reason != 0 { + t.Fatalf("connack reason=%d raw=%x", reason, ack) + } + + sub := packets.Packet{ + FixedHeader: packets.FixedHeader{Type: packets.Subscribe, Qos: 1}, + ProtocolVersion: 5, + PacketID: 1, + Filters: packets.Subscriptions{ + {Filter: "nix/c/" + endpointID + "/down", Qos: 1}, + }, + } + buf.Reset() + if err := sub.SubscribeEncode(&buf); err != nil { + t.Fatal(err) + } + if err := mc.Send(buf.Bytes()); err != nil { + t.Fatal(err) + } + if _, err := mc.Recv(); err != nil { // SUBACK + t.Fatalf("suback: %v", err) + } + + hello, _ := protocol.Marshal(protocol.Hello{ + V: protocol.Version, Type: protocol.TypeHello, RID: "h1", + }) + pub := packets.Packet{ + FixedHeader: packets.FixedHeader{Type: packets.Publish, Qos: 1}, + ProtocolVersion: 5, + TopicName: "nix/c/" + endpointID + "/up", + PacketID: 2, + Payload: hello, + } + buf.Reset() + if err := pub.PublishEncode(&buf); err != nil { + t.Fatal(err) + } + if err := mc.Send(buf.Bytes()); err != nil { + t.Fatal(err) + } + + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + raw, err := mc.Recv() + if err != nil { + t.Fatalf("recv down: %v", err) + } + if len(raw) < 2 { + continue + } + typ := raw[0] >> 4 + if typ == packets.Puback || typ == packets.Pingresp { + continue + } + if typ != packets.Publish { + continue + } + payload, err := decodePublishPayload(raw) + if err != nil { + t.Fatalf("publish decode: %v raw=%x", err, raw) + } + var m map[string]any + if err := json.Unmarshal(payload, &m); err != nil { + t.Fatalf("json: %v payload=%s", err, payload) + } + if m["type"] == "resp" { + if tok := extractSessionToken(m); tok != "" { + return tok + } + t.Fatalf("hello resp without token: %v", m) + } + } + t.Fatal("timeout waiting hello resp") + return "" +} + +func decodePublishPayload(raw []byte) ([]byte, error) { + rem, n, err := decodeRemainingLength(raw[1:]) + if err != nil { + return nil, err + } + body := raw[1+n:] + if len(body) != rem { + return nil, io.ErrUnexpectedEOF + } + pk := packets.Packet{ + ProtocolVersion: 5, + FixedHeader: packets.FixedHeader{ + Type: packets.Publish, + Remaining: rem, + Qos: (raw[0] >> 1) & 0x3, + }, + } + if err := pk.PublishDecode(body); err != nil { + return nil, err + } + return pk.Payload, nil +} + +func decodeRemainingLength(b []byte) (value int, n int, err error) { + var mul uint32 = 1 + var v uint32 + for i := 0; i < len(b) && i < 4; i++ { + v += uint32(b[i]&127) * mul + n++ + if b[i]&128 == 0 { + return int(v), n, nil + } + mul *= 128 + } + return 0, 0, io.ErrUnexpectedEOF +} + +func extractSessionToken(m map[string]any) string { + if m["ok"] != true { + return "" + } + data, _ := m["data"].(map[string]any) + tok, _ := data["session_token"].(string) + return tok +} + +func enableRegistration(t *testing.T, dataDir, code string) { + t.Helper() + db, err := store.Open(dataDir, "FULL") + if err != nil { + t.Fatal(err) + } + defer func() { _ = db.Close() }() + now := time.Now().UnixMilli() + err = db.Queue.Do(context.Background(), func(tx *sql.Tx) error { + if _, e := tx.Exec(`INSERT INTO settings(key, value, updated_at) VALUES(?,?,?) +ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at`, + "registration_enabled", "1", now); e != nil { + return e + } + _, e := tx.Exec(`INSERT INTO settings(key, value, updated_at) VALUES(?,?,?) +ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at`, + "registration_code", code, now) + return e + }) + if err != nil { + t.Fatal(err) + } +} + +func waitListenAddr(t *testing.T, dataDir string, timeout time.Duration) string { + t.Helper() + deadline := time.Now().Add(timeout) + path := filepath.Join(dataDir, "listen.addr") + for time.Now().Before(deadline) { + b, err := os.ReadFile(path) + if err == nil { + addr := strings.TrimSpace(string(b)) + if addr != "" { + return addr + } + } + time.Sleep(20 * time.Millisecond) + } + t.Fatal("listen.addr not written") + return "" +} diff --git a/docs/DEVIATIONS.md b/docs/DEVIATIONS.md index 40775d5..5e1163b 100644 --- a/docs/DEVIATIONS.md +++ b/docs/DEVIATIONS.md @@ -43,6 +43,29 @@ - 备选方案：docker 目标仅 echo 提示。 - 影响：镜像发布流程仍由 Q4 定稿。 +### L-WIRE 2026-09-30 + +1. **serve 真实接线范围** + - 原条款：TASKS 总控接线；listener/broker/admin/注册/消息周期循环挂到进程。 + - 实际做法：`cmd/nixmsg/serve.go` 注入真实 `auth.NewPool`/`NewSessionTokens`/`NewAPITokens`/`NewLoginLocks`，挂注册与管理路由（踢线调 `Session.Kick`），listener 识别 HTTP/WebSocket/`OnMQTT` 裸 TCP，broker `Login`+`Session`，握手/断线/`OnPublishDropped` 接到 `message.App`，周期 `DispatchDue`/`PushPending`/`CleanupOnce`，下行 `PublishDown`。集成测覆盖：管理登录、写 settings 后注册、密码 MQTT 握手拿 `session_token`。 + - 原因：第二波收尾；三件验收必须通。 + - 备选方案：分文件多阶段接线。 + - 影响：进程可对外登录/注册/握手。 + +2. **未接线 / 未完成部分** + - 原条款：上行应用帧完整分发到 identity/group/presence/message 业务方法。 + - 实际做法：`Session` 处理 hello/logout；其余 `HandleUplink` 仍为 `StubUplinkHandler`（不解析 send/ack/group 等）。管理注册设置 HTTP（A3）未实现，测试直接写 `settings` 表。群/在线模块已构造并注入 Downlink，但无上行入口调用。 + - 原因：本波强制三件验收；完整协议分发属后续波次。 + - 备选方案：本波同时实现 HandleUplink 大 multiplex。 + - 影响：端连上后除握手/logout 外业务帧尚无 resp；F03–F16 等仍依赖后续接线。 + +3. **listen.addr 始终写入** + - 原条款：listener N1 仅端口 0 写文件；T0.1 超集为启动即写。 + - 实际做法：listener 按 N1 写端口 0；serve 成功后再强制写一次 `listen.addr`（及分离时的 `admin.addr`）。 + - 原因：与 harness / T0.1 一致。 + - 备选方案：改 listener 始终写。 + - 影响：固定端口也会有地址文件。 + ### T0.2 2026-09-30 1. **请求指纹规范化格式** diff --git a/internal/app/message/conn.go b/internal/app/message/conn.go index 382a464..a828040 100644 --- a/internal/app/message/conn.go +++ b/internal/app/message/conn.go @@ -67,6 +67,17 @@ func (c *MemoryConns) Current(endpointID string) (LiveConn, bool) { return v, ok } +// Snapshot 返回当前连接表副本（供推送循环遍历）。 +func (c *MemoryConns) Snapshot() map[string]LiveConn { + c.mu.RLock() + defer c.mu.RUnlock() + out := make(map[string]LiveConn, len(c.m)) + for k, v := range c.m { + out[k] = v + } + return out +} + // RecordingDownlink 记录下行发布，供测试断言。 type RecordingDownlink struct { mu sync.Mutex diff --git a/internal/broker/broker.go b/internal/broker/broker.go index bc1fd8f..f4989b3 100644 --- a/internal/broker/broker.go +++ b/internal/broker/broker.go @@ -58,11 +58,16 @@ func (AllowAuthenticator) Authenticate(context.Context, string, []byte, string) return AuthResult{OK: true}, nil } +// PublishDroppedFunc 下行未写入发送队列时回调（对接消息线 OnPublishDropped）。 +type PublishDroppedFunc func(ctx context.Context, endpointID string, connID port.ConnID, payload []byte) + // Options 装配 broker。 type Options struct { Authenticator Authenticator Uplink port.UplinkHandler Logger *slog.Logger + // OnPublishDropped 可选；nil 时仅打 debug 日志。 + OnPublishDropped PublishDroppedFunc } // Broker 内置 mochi，不自带监听端口。 @@ -71,6 +76,7 @@ type Broker struct { auth Authenticator uplink port.UplinkHandler log *slog.Logger + onDrop PublishDroppedFunc hook *nixHook @@ -144,6 +150,7 @@ func New(opts Options) (*Broker, error) { auth: auth, uplink: uplink, log: log, + onDrop: opts.OnPublishDropped, current: make(map[string]*connState), byClient: make(map[*mqtt.Client]*connState), queues: make(map[string]*uplinkQueue), diff --git a/internal/broker/hooks.go b/internal/broker/hooks.go index 58b4b57..d503274 100644 --- a/internal/broker/hooks.go +++ b/internal/broker/hooks.go @@ -159,6 +159,16 @@ func (h *nixHook) OnSubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes [ func (h *nixHook) OnPublishDropped(cl *mqtt.Client, pk packets.Packet) { h.b.log.Debug("publish dropped", "client", cl.ID, "topic", pk.TopicName, "size", len(pk.Payload)) + if h.b.onDrop == nil { + return + } + h.b.connsMu.RLock() + st := h.b.byClient[cl] + h.b.connsMu.RUnlock() + if st == nil { + return + } + h.b.onDrop(context.Background(), st.endpointID, st.connID, append([]byte(nil), pk.Payload...)) } func (h *nixHook) OnSessionEstablished(cl *mqtt.Client, _ packets.Packet) { diff --git a/internal/broker/session.go b/internal/broker/session.go index d714751..52bf739 100644 --- a/internal/broker/session.go +++ b/internal/broker/session.go @@ -101,6 +101,11 @@ func (s *Session) Attach(b *Broker) { s.b = b } +// SetPresence 接线时在创建最终 presence 实现后注入（可替换占位）。 +func (s *Session) SetPresence(p PresenceSink) { + s.presence = p +} + func (s *Session) OnSessionEstablished(ctx context.Context, conn port.ConnInfo) error { if s.b != nil { s.b.startHandshakeDeadline(conn.EndpointID, conn.ConnID, handshakeTimeout)