nixevol/NixMsg
1
0
Fork 0
Code Issues 59 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 实现 version/serve 骨架、迁移占位与健康检查

Browse Source
This commit is contained in:
Nixevol
2026-09-30 05:50:37 +08:00
parent 857ba893fd
commit 27fccc2e2c
10 changed files with 523 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/nixmsg/main.go
+25
View File
@@ -0,0 +1,25 @@
package main
import (
"fmt"
"os"
)
func main() {
if len(os.Args) < 2 {
fmt.Fprintln(os.Stderr, "usage: nixmsg <command>")
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)
}
}
cmd/nixmsg/serve.go
+97
View File
@@ -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)
}
cmd/nixmsg/serve_test.go
+81
View File
@@ -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")
}
}
cmd/nixmsg/version.go
+13
View File
@@ -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)
}
cmd/nixmsg/wire.go
+4
View File
@@ -0,0 +1,4 @@
package main
// wire 组装各业务模块。T0.1 仅占位，T0.4 起由各线注册。
func wire() {}
internal/config/config.go
+122
View File
@@ -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"
}
internal/store/db.go
+49
View File
@@ -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
}
internal/store/db_test.go
+36
View File
@@ -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()
}
internal/store/migrate.go
+95
View File
@@ -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
}
internal/store/migrations/0001_init.sql
+1
View File
@@ -0,0 +1 @@
SELECT 1;