feat: 接线 serve 真实 auth/注册/管理/MQTT 与消息循环

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