diff --git a/cmd/nixmsg/main.go b/cmd/nixmsg/main.go new file mode 100644 index 0000000..b335e86 --- /dev/null +++ b/cmd/nixmsg/main.go @@ -0,0 +1,25 @@ +package main + +import ( + "fmt" + "os" +) + +func main() { + if len(os.Args) < 2 { + fmt.Fprintln(os.Stderr, "usage: nixmsg ") + os.Exit(2) + } + switch os.Args[1] { + case "version": + cmdVersion(os.Args[2:]) + case "serve": + if err := cmdServe(os.Args[2:]); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + default: + fmt.Fprintf(os.Stderr, "unknown command: %s\n", os.Args[1]) + os.Exit(2) + } +} diff --git a/cmd/nixmsg/serve.go b/cmd/nixmsg/serve.go new file mode 100644 index 0000000..f9d53b6 --- /dev/null +++ b/cmd/nixmsg/serve.go @@ -0,0 +1,97 @@ +package main + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "os" + "os/signal" + "path/filepath" + "syscall" + "time" + + "git.asio.asia/nixevol/NixMsg/internal/config" + "git.asio.asia/nixevol/NixMsg/internal/store" + "git.asio.asia/nixevol/NixMsg/web" +) + +func cmdServe(_ []string) error { + cfgPath := config.PathFromEnv() + cfg, err := config.Load(cfgPath) + if err != nil { + return err + } + 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 { + wire() + + if err := os.MkdirAll(cfg.DataDir, 0o755); err != nil { + return fmt.Errorf("mkdir data_dir: %w", err) + } + + db, err := store.OpenWriter(cfg.DataDir, cfg.SQLiteSynchronous) + if err != nil { + return err + } + defer func() { _ = db.Close() }() + + if migErr := store.Migrate(db); migErr != nil { + return migErr + } + + mux := http.NewServeMux() + mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok")) + }) + // 确保前端资源被链接进二进制;完整静态托管由后续任务完善。 + _ = 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() + return err + } + + srv := &http.Server{ + Handler: mux, + ReadHeaderTimeout: 10 * time.Second, + } + + errCh := make(chan error, 1) + go func() { + errCh <- srv.Serve(ln) + }() + + select { + case <-ctx.Done(): + shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _ = srv.Shutdown(shutdownCtx) + 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 + } +} + +func writeListenAddr(dataDir, addr string) error { + path := filepath.Join(dataDir, "listen.addr") + return os.WriteFile(path, []byte(addr+"\n"), 0o644) +} diff --git a/cmd/nixmsg/serve_test.go b/cmd/nixmsg/serve_test.go new file mode 100644 index 0000000..8524111 --- /dev/null +++ b/cmd/nixmsg/serve_test.go @@ -0,0 +1,81 @@ +package main + +import ( + "context" + "io" + "net/http" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "git.asio.asia/nixevol/NixMsg/internal/config" +) + +func TestServeHealthzAndListenAddr(t *testing.T) { + t.Parallel() + + dataDir := t.TempDir() + cfgPath := filepath.Join(dataDir, "config.yaml") + cfgYAML := []byte("listen: \"127.0.0.1:0\"\ndata_dir: \"" + filepath.ToSlash(dataDir) + "\"\n") + if err := os.WriteFile(cfgPath, cfgYAML, 0o644); err != nil { + t.Fatal(err) + } + + cfg, err := config.Load(cfgPath) + if err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + errCh := make(chan error, 1) + go func() { + errCh <- runServe(ctx, cfg) + }() + + var addr string + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + b, readErr := os.ReadFile(filepath.Join(dataDir, "listen.addr")) + if readErr == nil { + addr = strings.TrimSpace(string(b)) + if addr != "" { + break + } + } + time.Sleep(20 * time.Millisecond) + } + if addr == "" { + t.Fatal("listen.addr not written") + } + + resp, err := http.Get("http://" + addr + "/healthz") + if err != nil { + t.Fatalf("healthz: %v", err) + } + defer func() { _ = resp.Body.Close() }() + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK { + t.Fatalf("status=%d body=%s", resp.StatusCode, body) + } + if string(body) != "ok" { + t.Fatalf("unexpected body: %q", body) + } + + if _, err := os.Stat(filepath.Join(dataDir, "nixmsg.db")); err != nil { + t.Fatalf("db missing: %v", err) + } + + cancel() + select { + case err := <-errCh: + if err != nil { + t.Fatalf("serve exit: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("serve did not stop") + } +} diff --git a/cmd/nixmsg/version.go b/cmd/nixmsg/version.go new file mode 100644 index 0000000..086cede --- /dev/null +++ b/cmd/nixmsg/version.go @@ -0,0 +1,13 @@ +package main + +import ( + "fmt" + "runtime" +) + +// Version 由构建注入;未注入时显示开发占位。 +var Version = "dev" + +func cmdVersion(_ []string) { + fmt.Printf("nixmsg %s %s/%s\n", Version, runtime.GOOS, runtime.GOARCH) +} diff --git a/cmd/nixmsg/wire.go b/cmd/nixmsg/wire.go new file mode 100644 index 0000000..06e9b86 --- /dev/null +++ b/cmd/nixmsg/wire.go @@ -0,0 +1,4 @@ +package main + +// wire 组装各业务模块。T0.1 仅占位,T0.4 起由各线注册。 +func wire() {} diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..acecf0a --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,122 @@ +package config + +import ( + "fmt" + "os" + + "go.yaml.in/yaml/v3" +) + +// Config 对应 DEVELOPMENT 第 11.1 节的配置文件。 +type Config struct { + Listen string `yaml:"listen"` + AdminListen string `yaml:"admin_listen"` + TLS TLSConfig `yaml:"tls"` + TrustedProxies []string `yaml:"trusted_proxies"` + DataDir string `yaml:"data_dir"` + Limits LimitsConfig `yaml:"limits"` + SessionIdleDays int `yaml:"session_idle_days"` + RecordRetentionDays int `yaml:"record_retention_days"` + IdempotencyHours int `yaml:"idempotency_hours"` + ReceiptRetentionDays int `yaml:"receipt_retention_days"` + SQLiteSynchronous string `yaml:"sqlite_synchronous"` + Metrics MetricsConfig `yaml:"metrics"` + Log LogConfig `yaml:"log"` +} + +type TLSConfig struct { + CertFile string `yaml:"cert_file"` + KeyFile string `yaml:"key_file"` + AllowPlaintext bool `yaml:"allow_plaintext"` +} + +type LimitsConfig struct { + MaxBodyBytes int `yaml:"max_body_bytes"` + MaxMetaBytes int `yaml:"max_meta_bytes"` + MaxFrameBytes int `yaml:"max_frame_bytes"` + MaxTTLSeconds int `yaml:"max_ttl_seconds"` + MaxScheduleSeconds int `yaml:"max_schedule_seconds"` + MaxGroupMembers int `yaml:"max_group_members"` + GraceSeconds int `yaml:"grace_seconds"` + AckTimeoutSeconds int `yaml:"ack_timeout_seconds"` + DeliveryWindow int `yaml:"delivery_window"` + ReceiptWindow int `yaml:"receipt_window"` + RequestsPerSecond int `yaml:"requests_per_second"` + MaxPendingPerSender int `yaml:"max_pending_per_sender"` + MaxPendingPerReceiver int `yaml:"max_pending_per_receiver"` +} + +type MetricsConfig struct { + Token string `yaml:"token"` +} + +type LogConfig struct { + Level string `yaml:"level"` +} + +// Default 返回与 DEVELOPMENT 11.1 示例一致的默认值。 +func Default() Config { + return Config{ + Listen: ":7443", + AdminListen: "", + TLS: TLSConfig{ + AllowPlaintext: false, + }, + TrustedProxies: nil, + DataDir: "./data", + Limits: LimitsConfig{ + MaxBodyBytes: 262144, + MaxMetaBytes: 4096, + MaxFrameBytes: 786432, + MaxTTLSeconds: 2592000, + MaxScheduleSeconds: 31536000, + MaxGroupMembers: 1000, + GraceSeconds: 60, + AckTimeoutSeconds: 300, + DeliveryWindow: 32, + ReceiptWindow: 64, + RequestsPerSecond: 50, + MaxPendingPerSender: 10000, + MaxPendingPerReceiver: 10000, + }, + SessionIdleDays: 30, + RecordRetentionDays: 7, + IdempotencyHours: 24, + ReceiptRetentionDays: 7, + SQLiteSynchronous: "FULL", + Metrics: MetricsConfig{}, + Log: LogConfig{ + Level: "info", + }, + } +} + +// Load 从路径加载 YAML 配置;空字段保留默认值。 +func Load(path string) (Config, error) { + cfg := Default() + data, err := os.ReadFile(path) + if err != nil { + return Config{}, fmt.Errorf("read config: %w", err) + } + if err := yaml.Unmarshal(data, &cfg); err != nil { + return Config{}, fmt.Errorf("parse config: %w", err) + } + if cfg.Listen == "" { + cfg.Listen = Default().Listen + } + if cfg.DataDir == "" { + cfg.DataDir = Default().DataDir + } + if cfg.SQLiteSynchronous == "" { + cfg.SQLiteSynchronous = Default().SQLiteSynchronous + } + return cfg, nil +} + +// PathFromEnv 返回 NIXMSG_CONFIG 或默认 ./config.yaml。 +func PathFromEnv() string { + if p := os.Getenv("NIXMSG_CONFIG"); p != "" { + return p + } + return "./config.yaml" +} diff --git a/internal/store/db.go b/internal/store/db.go new file mode 100644 index 0000000..d4ce685 --- /dev/null +++ b/internal/store/db.go @@ -0,0 +1,49 @@ +package store + +import ( + "database/sql" + "fmt" + "os" + "path/filepath" + "strings" + + _ "modernc.org/sqlite" +) + +// OpenWriter 按 DEVELOPMENT 7.7 打开写连接(带 _txlock=immediate)。 +func OpenWriter(dataDir, synchronous string) (*sql.DB, error) { + if err := os.MkdirAll(dataDir, 0o755); err != nil { + return nil, fmt.Errorf("mkdir data_dir: %w", err) + } + dsn, err := writeDSN(dataDir, synchronous) + if err != nil { + return nil, err + } + db, err := sql.Open("sqlite", dsn) + if err != nil { + return nil, fmt.Errorf("open sqlite: %w", err) + } + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) + if err := db.Ping(); err != nil { + _ = db.Close() + return nil, fmt.Errorf("ping sqlite: %w", err) + } + return db, nil +} + +func writeDSN(dataDir, synchronous string) (string, error) { + sync := strings.ToUpper(strings.TrimSpace(synchronous)) + if sync == "" { + sync = "FULL" + } + if sync != "FULL" && sync != "NORMAL" { + return "", fmt.Errorf("invalid sqlite_synchronous: %s", synchronous) + } + dbPath := filepath.ToSlash(filepath.Join(dataDir, "nixmsg.db")) + return fmt.Sprintf( + "file:%s?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=synchronous(%s)&_pragma=foreign_keys(ON)&_pragma=secure_delete(ON)&_txlock=immediate", + dbPath, + sync, + ), nil +} diff --git a/internal/store/db_test.go b/internal/store/db_test.go new file mode 100644 index 0000000..a43e472 --- /dev/null +++ b/internal/store/db_test.go @@ -0,0 +1,36 @@ +package store + +import ( + "path/filepath" + "testing" +) + +func TestOpenAndMigrate(t *testing.T) { + t.Parallel() + dir := t.TempDir() + db, err := OpenWriter(dir, "FULL") + if err != nil { + t.Fatal(err) + } + defer func() { _ = db.Close() }() + + if migErr := Migrate(db); migErr != nil { + t.Fatal(migErr) + } + if migErr := Migrate(db); migErr != nil { + t.Fatal(migErr) + } + + var n int + if scanErr := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations`).Scan(&n); scanErr != nil { + t.Fatal(scanErr) + } + if n != 1 { + t.Fatalf("want 1 migration row, got %d", n) + } + nested, openErr := OpenWriter(filepath.Join(dir, "nested"), "NORMAL") + if openErr != nil { + t.Fatal(openErr) + } + _ = nested.Close() +} diff --git a/internal/store/migrate.go b/internal/store/migrate.go new file mode 100644 index 0000000..5212480 --- /dev/null +++ b/internal/store/migrate.go @@ -0,0 +1,95 @@ +package store + +import ( + "database/sql" + "embed" + "fmt" + "io/fs" + "sort" + "strconv" + "strings" + "time" +) + +//go:embed migrations/*.sql +var migrationFS embed.FS + +// Migrate 应用尚未执行的迁移。T0.1 仅为空执行器加 0001 占位;完整备份与表结构见 T0.3。 +func Migrate(db *sql.DB) error { + if _, err := db.Exec(` +CREATE TABLE IF NOT EXISTS schema_migrations ( + version INTEGER PRIMARY KEY, + applied_at INTEGER NOT NULL +)`); err != nil { + return fmt.Errorf("ensure schema_migrations: %w", err) + } + + entries, err := fs.ReadDir(migrationFS, "migrations") + if err != nil { + return fmt.Errorf("read migrations: %w", err) + } + var names []string + for _, e := range entries { + if e.IsDir() || !strings.HasSuffix(e.Name(), ".sql") { + continue + } + names = append(names, e.Name()) + } + sort.Strings(names) + + for _, name := range names { + version, err := parseMigrationVersion(name) + if err != nil { + return err + } + var exists int + err = db.QueryRow(`SELECT 1 FROM schema_migrations WHERE version = ?`, version).Scan(&exists) + if err == nil { + continue + } + if err != sql.ErrNoRows { + return fmt.Errorf("check migration %d: %w", version, err) + } + + body, err := migrationFS.ReadFile("migrations/" + name) + if err != nil { + return fmt.Errorf("read migration %s: %w", name, err) + } + sqlText := strings.TrimSpace(string(body)) + tx, err := db.Begin() + if err != nil { + return fmt.Errorf("begin migration %d: %w", version, err) + } + if sqlText != "" { + if _, err := tx.Exec(sqlText); err != nil { + _ = tx.Rollback() + return fmt.Errorf("apply migration %d: %w", version, err) + } + } + if _, err := tx.Exec( + `INSERT INTO schema_migrations(version, applied_at) VALUES(?, ?)`, + version, + time.Now().UnixMilli(), + ); err != nil { + _ = tx.Rollback() + return fmt.Errorf("record migration %d: %w", version, err) + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit migration %d: %w", version, err) + } + } + return nil +} + +func parseMigrationVersion(name string) (int, error) { + base := strings.TrimSuffix(name, ".sql") + parts := strings.SplitN(base, "_", 2) + if len(parts) < 1 { + return 0, fmt.Errorf("invalid migration name: %s", name) + } + v, err := strconv.Atoi(parts[0]) + if err != nil { + return 0, fmt.Errorf("invalid migration version in %s: %w", name, err) + } + return v, nil +} diff --git a/internal/store/migrations/0001_init.sql b/internal/store/migrations/0001_init.sql new file mode 100644 index 0000000..e0ac49d --- /dev/null +++ b/internal/store/migrations/0001_init.sql @@ -0,0 +1 @@ +SELECT 1;