package app import ( "context" "fmt" "io" "io/fs" "log/slog" "net" "net/http" "os" "path/filepath" "strings" "time" "github.com/havline/havline/internal/acme" "github.com/havline/havline/internal/api" "github.com/havline/havline/internal/auth" "github.com/havline/havline/internal/backup" "github.com/havline/havline/internal/certificate" "github.com/havline/havline/internal/chinacidr" "github.com/havline/havline/internal/cloudflared" "github.com/havline/havline/internal/config" "github.com/havline/havline/internal/db" "github.com/havline/havline/internal/ddns" "github.com/havline/havline/internal/discovery" "github.com/havline/havline/internal/frp" "github.com/havline/havline/internal/logstore" "github.com/havline/havline/internal/metrics" "github.com/havline/havline/internal/monitor" "github.com/havline/havline/internal/nginx" "github.com/havline/havline/internal/notify" "github.com/havline/havline/internal/proxy" "github.com/havline/havline/internal/scheduler" "github.com/havline/havline/internal/secret" "github.com/havline/havline/internal/service" "github.com/havline/havline/internal/settings" "github.com/havline/havline/internal/terminal" "github.com/havline/havline/internal/traffic" "github.com/havline/havline/internal/updater" ) type App struct { cfg config.Config logger *slog.Logger server *http.Server nginx *nginx.Manager frpMulti *frp.Service cloudflare *cloudflared.Service scheduler *scheduler.Scheduler shutdownFn context.CancelFunc db interface{ Close() error } } func New(cfg config.Config, staticFS fs.FS, migrationsDir string) (*App, error) { appLogWriter := logstore.NewAppLogWriter(cfg.LogsDir()) logger := slog.New(slog.NewJSONHandler(io.MultiWriter(os.Stdout, appLogWriter), &slog.HandlerOptions{Level: slog.LevelInfo})) slog.SetDefault(logger) for _, dir := range []string{cfg.DataDir, cfg.NginxDir(), cfg.LogsDir(), cfg.CertsDir(), cfg.FrpDir()} { if err := os.MkdirAll(dir, 0o755); err != nil { return nil, err } } // 会话密钥:显式配置优先;未配置则在数据目录里读/生成并持久化 // (它同时是 secret box 的加密密钥,所以只在未配置时才生成,不替换已配置的值) secretSource, err := cfg.ResolveSessionSecret() if err != nil { return nil, err } logger.Info("会话密钥已就绪", "module", "SYSTEM", "source", secretSource) // updater Token 只给「一键升级」sidecar 用;原生版与飞牛 Docker 版都显式把 socket 置空, // 此时不能去写 /run/havline-updater:原生版以非 root 运行,权限不足会让应用直接起不来。 updaterToken, updaterTokenSource, err := updater.EnsureTokenIfEnabled(cfg.UpdaterSocket, cfg.UpdaterToken, cfg.UpdaterTokenFile) if err != nil { return nil, err } if updaterToken != "" { logger.Info("updater Token 已就绪", "module", "SYSTEM", "source", updaterTokenSource) } conn, err := db.Open(cfg.DBPath()) if err != nil { return nil, err } if err := db.Migrate(context.Background(), conn, migrationsDir); err != nil { conn.Close() return nil, err } secretBox, err := secret.NewBox(cfg.SessionSecret) if err != nil { conn.Close() return nil, err } authSvc := auth.New(conn) if err := authSvc.EnsureDefaultAdmin(context.Background(), logger); err != nil { conn.Close() return nil, fmt.Errorf("bootstrap admin: %w", err) } proxyStore := proxy.NewStore(conn) settingsStore := settings.NewStore(conn) // 时区必须在任何业务开始前生效:通知时间戳、日志时间、调度判断都依赖 time.Local。 // 设置项缺失或写坏时退回 TZ 环境变量,不让应用起不来。 if loc, err := settings.ApplyTimezone(context.Background(), settingsStore); err != nil { logger.Warn("应用时区设置失败,沿用系统 TZ", "module", "SYSTEM", "error", err.Error()) } else { logger.Info("时区已应用", "module", "SYSTEM", "timezone", loc.String()) } ddnsStore := ddns.NewStore(conn) certStore := certificate.NewStore(conn) nginxMgr := nginx.NewManager(cfg, logger.With("module", "NGINX")) if err := nginxMgr.EnsureDirs(); err != nil { conn.Close() return nil, err } proxySvc := service.NewProxyService(cfg, conn, proxyStore, certStore, settingsStore, nginxMgr) chinaCIDRSvc := chinacidr.New(cfg, settingsStore, nginxMgr, proxyStore, logger.With("module", "CHINA_CIDR")) notifySvc := notify.New(settingsStore, secretBox, logger) proxySvc.SetNotify(notifySvc) ddnsSvc := ddns.NewService(ddnsStore, settingsStore, secretBox, logger, notifySvc) acmeSvc := acme.NewService(cfg, certStore, ddnsSvc, settingsStore, proxySvc, logger, notifySvc) backupSvc := backup.New(cfg.DataDir) // 注入数据库连接:导出时用 VACUUM INTO 拿一致性快照,而不是直接打包正在写的文件 backupSvc.SetDB(conn) // 指标历史:与巡检同生命周期,采样跟随巡检轮次 metricsStore := metrics.NewStore(conn) discoverySvc := discovery.New() // 新增:多服务端 FRP 模块(接管原网关模式) frpMultiSvc := frp.NewService(cfg, frp.NewStore(conn), secretBox, logger, certStore) if err := frpMultiSvc.EnsureDirs(); err != nil { conn.Close() return nil, err } // 记录/恢复 frpc 运行状态:应用被停用再启用后要能自动回到运行状态 frpMultiSvc.SetSettingsStore(settingsStore) cloudflareSvc := cloudflared.NewService(cfg, cloudflared.NewStore(conn), secretBox, logger, proxySvc) if err := cloudflareSvc.EnsureDirs(); err != nil { conn.Close() return nil, err } proxySvc.SetExitSync(cloudflareSvc) startedAt := time.Now().UTC().Format(time.RFC3339) trafficCollector := traffic.NewCollector(conn, filepath.Join(cfg.LogsDir(), "access.log"), notifySvc) // 删规则时连带清掉它的流量累计,否则同名规则重建会继承上一轮的读数 proxySvc.SetTrafficCleaner(trafficCollector) updaterClient := updater.NewClient(cfg.UpdaterSocket, updaterToken) // 内置 SSH 终端:会话管理借用 frp 的 SSH 凭据与 known_hosts;开关默认关闭,见 settings.KeyTerminalEnabled // 会话审计落 SQLite:谁、何时、连了哪台、多久、为什么断开 terminalAuditStore := terminal.NewAuditStore(conn) // 常用命令:右侧面板的快捷命令,落库后换浏览器也在 terminalCommandStore := terminal.NewCommandStore(conn) terminalManager := terminal.NewManager(logger, frpMultiSvc.OpenSSHTerminal, terminal.Options{}) terminalManager.SetAuditSink(terminalAuditStore) // 会话录制:默认关闭,开启后按 asciinema v2 把输出写进 logs/terminal terminalRecordingStore := terminal.NewRecordingStore(filepath.Join(cfg.LogsDir(), "terminal")) if err := terminalRecordingStore.EnsureDir(); err != nil { logger.Warn("终端录制目录创建失败", "module", "TERMINAL", "error", err.Error()) } terminalManager.SetRecorderFactory(func(sessionID string, opts terminal.CreateOptions) (terminal.OutputRecorder, error) { title := fmt.Sprintf("服务端 #%d", opts.ServerID) return terminal.NewCastRecorder(terminalRecordingStore, sessionID, title, opts.Cols, opts.Rows, time.Now()) }) handler := api.NewRouter(api.Deps{ Config: cfg, Logger: logger, Auth: authSvc, Proxy: proxySvc, DDNS: ddnsSvc, ACME: acmeSvc, Settings: settingsStore, Notify: notifySvc, ChinaCIDR: chinaCIDRSvc, Backup: backupSvc, Metrics: metricsStore, Discovery: discoverySvc, FRPMulti: frpMultiSvc, Cloudflare: cloudflareSvc, Traffic: trafficCollector, Updater: updaterClient, StaticFS: staticFS, StartedAt: startedAt, Terminal: terminalManager, TerminalAudits: terminalAuditStore, TerminalCommands: terminalCommandStore, TerminalRecordings: terminalRecordingStore, }) if prefix := strings.TrimSpace(cfg.GatewayPrefix); prefix != "" { handler = stripPrefixIfPresent(prefix, handler) } server := &http.Server{ Addr: cfg.ListenAddr, Handler: handler, ReadHeaderTimeout: 10 * time.Second, // 读超时放到 5 分钟:备份恢复要上传最大 64MB;不设 WriteTimeout,否则日志 SSE 会被中途切断 ReadTimeout: 5 * time.Minute, IdleTimeout: 2 * time.Minute, } ctx, cancel := context.WithCancel(context.Background()) // 终端会话回收:空闲超时、超长会话与已结束会话都靠它清理 terminalManager.StartSweeper(ctx, time.Minute) frpMultiSvc.SetBaseContext(ctx) // cloudflared 子进程跟随应用 ctx:应用退出时不会留下孤儿进程 cloudflareSvc.SetBaseContext(ctx) intervalMinutes, _ := settingsStore.GetInt(ctx, settings.KeyDDNSCheckInterval) if intervalMinutes <= 0 { intervalMinutes = 5 } chinaHours, _ := settingsStore.GetInt(ctx, settings.KeyChinaCIDRUpdateHours) if chinaHours <= 0 { chinaHours = 24 } // 巡检间隔比其它任务短:「连续 N 次不通」需要足够采样次数才能判定 monitorChecker := monitor.New(settingsStore, notifySvc, logger, monitorSource{frp: frpMultiSvc, cloudflare: cloudflareSvc, logger: logger, traffic: trafficCollector}, healthRecorder{frp: frpMultiSvc}, metricsStore) // 自动备份:启动时会先跑一轮,靠「最新备份是否在 24h 内」避免频繁重启重复备份 backupRunner := backup.NewAutoRunner(backupSvc, settingsStore, notifySvc, logger) sched := scheduler.New( scheduler.Job{ Name: "ddns", Interval: time.Duration(intervalMinutes) * time.Minute, Run: ddnsSvc.Tick, }, scheduler.Job{ Name: "acme", Interval: 24 * time.Hour, Run: acmeSvc.Tick, }, scheduler.Job{ Name: "china_cidr", Interval: time.Duration(chinaHours) * time.Hour, Run: chinaCIDRSvc.Tick, }, scheduler.Job{ Name: "monitor", Interval: 2 * time.Minute, Run: monitorChecker.Run, }, scheduler.Job{ Name: "backup", Interval: backup.AutoInterval, Run: backupRunner.Run, }, scheduler.Job{ Name: "terminal_audit", Interval: 24 * time.Hour, // 审计只留 90 天:清理比会话回收慢得多,单独挂一个日任务即可 Run: func(ctx context.Context) { deleted, err := terminalAuditStore.PruneAudits(ctx) if err != nil { logger.Warn("终端审计清理失败", "module", "TERMINAL", "error", err.Error()) } if deleted > 0 { logger.Info("终端审计清理完成", "module", "TERMINAL", "deleted", deleted) } removed, err := terminalRecordingStore.Prune(time.Now().Add(-terminal.AuditRetention)) if err != nil { logger.Warn("终端录制清理失败", "module", "TERMINAL", "error", err.Error()) } if removed > 0 { logger.Info("终端录制清理完成", "module", "TERMINAL", "deleted", removed) } }, }, scheduler.Job{ Name: "session_prune", Interval: 24 * time.Hour, // 过期会话否则只能等自己被访问时删掉自己,长期不登录的设备会一直留在表里 Run: func(ctx context.Context) { removed, err := authSvc.PruneExpiredSessions(ctx) if err != nil { logger.Warn("过期会话清理失败", "module", "AUTH", "error", err.Error()) return } if removed > 0 { logger.Info("过期会话清理完成", "module", "AUTH", "deleted", removed) } }, }, scheduler.Job{ Name: "log_rotate", Interval: time.Hour, // Nginx 自己不会轮转:不切的话 access.log 会一直长,同时读日志的 // 几个接口(今日统计、小时曲线)会越来越慢。 Run: func(ctx context.Context) { rotateNginxLogs(ctx, cfg, nginxMgr, settingsStore, logger) }, }, ) sched.Start(ctx) trafficCollector.Start(ctx) app := &App{ cfg: cfg, logger: logger, server: server, nginx: nginxMgr, frpMulti: frpMultiSvc, cloudflare: cloudflareSvc, scheduler: sched, shutdownFn: cancel, db: conn, } // 就绪门:Nginx 首轮配置应用完之前不放后台任务。 // 否则 Agent 路由下发、Cloudflare ingress 同步会在启动瞬间各触发一次 // Nginx reload,和 bootstrap 抢同一份配置与 pid。 ready := make(chan struct{}) if err := app.bootstrapNginx(context.Background(), proxySvc); err != nil { cancel() conn.Close() return nil, err } close(ready) go waitReady(ctx, ready, func() { if err := frpMultiSvc.ReconcileRuntime(ctx); err != nil { logger.Warn("initial frp runtime reconcile skipped", "module", "FRP", "error", err.Error()) } if err := frpMultiSvc.AutoStart(ctx); err != nil { logger.Warn("initial frp autostart skipped", "module", "FRP", "error", err.Error()) } cloudflareSvc.StartAuto(ctx) }) go waitReady(ctx, ready, func() { frpMultiSvc.StartAgentTunnels(ctx) }) go waitReady(ctx, ready, func() { frpMultiSvc.StartAgentCertificateRenewal(ctx) }) listen := strings.TrimSpace(cfg.ListenAddr) if listen == "" { listen = strings.TrimSpace(cfg.GatewaySocket) } logger.Info("application started", "module", "SYSTEM", "listen", listen) return app, nil } // waitReady 等到 ready 关闭(或 ctx 取消)后再执行 fn。 func waitReady(ctx context.Context, ready <-chan struct{}, fn func()) { select { case <-ready: case <-ctx.Done(): return } if ctx.Err() != nil { return } fn() } // rotateNginxLogs 轮转 access.log 与 error.log。 // 保留天数接「设置 → 常规 → 日志保留天数」,份数用 logstore 默认值兜底。 func rotateNginxLogs(ctx context.Context, cfg config.Config, nginxMgr *nginx.Manager, store *settings.Store, logger *slog.Logger) { retentionDays, err := store.GetInt(ctx, settings.KeyLogRetentionDays) if err != nil || retentionDays <= 0 { retentionDays = 30 } logsDir := cfg.LogsDir() for _, name := range []string{"access.log", "error.log"} { rotator := logstore.NewRotator(filepath.Join(logsDir, name), retentionDays, func() error { return nginxMgr.ReopenLogs(ctx) }) rotated, err := rotator.RotateIfNeeded(time.Now()) if err != nil { logger.Warn("日志轮转失败", "module", "LOGSTORE", "file", name, "error", err.Error()) continue } if rotated { logger.Info("日志已轮转", "module", "LOGSTORE", "file", name) } } } func (a *App) bootstrapNginx(ctx context.Context, proxySvc *service.ProxyService) error { if err := proxySvc.ReloadAll(ctx); err != nil { a.logger.Warn("initial nginx apply skipped", "module", "NGINX", "error", err.Error()) } return nil } func (a *App) Run() error { socket := strings.TrimSpace(a.cfg.GatewaySocket) if socket == "" { return a.server.ListenAndServe() } listeners := make([]net.Listener, 0, 2) // 原生版显式把 TCP 地址置空,只保留 Unix Socket,避免和 Docker 版抢 6893。 if addr := strings.TrimSpace(a.server.Addr); addr != "" { tcp, err := net.Listen("tcp", addr) if err != nil { return err } defer tcp.Close() listeners = append(listeners, tcp) } if err := os.Remove(socket); err != nil && !os.IsNotExist(err) { return err } unixListener, err := net.Listen("unix", socket) if err != nil { return err } defer unixListener.Close() defer os.Remove(socket) if err := os.Chmod(socket, 0o660); err != nil { return err } listeners = append(listeners, unixListener) errCh := make(chan error, len(listeners)) for _, listener := range listeners { go func(l net.Listener) { errCh <- a.server.Serve(l) }(listener) } return <-errCh } func (a *App) Shutdown(ctx context.Context) error { a.logger.Info("application shutting down", "module", "SYSTEM") if a.shutdownFn != nil { a.shutdownFn() } if err := a.server.Shutdown(ctx); err != nil { return err } if socket := strings.TrimSpace(a.cfg.GatewaySocket); socket != "" { _ = os.Remove(socket) } if err := a.nginx.Stop(ctx); err != nil { a.logger.Error("nginx stop failed", "module", "NGINX", "error", err.Error()) } if a.frpMulti != nil { // 退出时保留「上次在运行」标记,供下次启用应用时自动恢复 frpc if err := a.frpMulti.StopForShutdown(ctx); err != nil { a.logger.Error("frpc stop failed", "module", "FRP", "error", err.Error()) } } // cloudflared 不保留「上次在运行」:AutoStart 是用户的显式选择, // 这里只负责把当前子进程收干净,避免退出后残留进程占着 Cloudflare 连接。 // 放在 nginx 之后:ingress 还指向本机 Nginx,先收隧道再做别的没有意义。 if a.cloudflare != nil { a.cloudflare.StopAll(ctx) } if a.db != nil { return a.db.Close() } return nil } func stripPrefixIfPresent(prefix string, next http.Handler) http.Handler { prefix = strings.TrimRight(prefix, "/") return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path == prefix || strings.HasPrefix(r.URL.Path, prefix+"/") { r.URL.Path = strings.TrimPrefix(r.URL.Path, prefix) if r.URL.Path == "" { r.URL.Path = "/" } } next.ServeHTTP(w, r) }) } func ResolveMigrationsDir() (string, error) { candidates := []string{ "migrations", filepath.Join("..", "migrations"), filepath.Join("..", "..", "migrations"), } for _, c := range candidates { if info, err := os.Stat(c); err == nil && info.IsDir() { return c, nil } } return "", fmt.Errorf("migrations directory not found") }