Artifact 864956a56946ce1f0f8689a29c4f44aa9dac8b10916991db206917273b9f1da7:
- File
cli/hub_serve.go
— part of check-in
[f3d5aa96bc]
at
2026-05-11 15:26:31
on branch trunk
— feat(hub): machine-level hub CLI + ServerName / LeafUpstream config (#153)
Introduces `edgesync hub serve` and extends the hub package so a hub can be named at the NATS level and optionally solicit a leaf connection upstream. This is the substrate piece; session/agent vocabulary lives in a consumer (sesh) above this layer.
hub.Config: - ServerName — sets opts.ServerName on the embedded NATS server; surfaces in leafnode handshakes, varz, gossip. Empty leaves nats-server's auto-generated random name. - LeafUpstream — optional URL to solicit a leafnode connection toward. Wired to opts.LeafNode.Remotes. Lets one hub act as a leaf of another, enabling multi-hub topologies. - NATSStoreDir default renamed `nats-store` → `messaging` — implementation- agnostic on disk; contents underneath are still NATS JetStream format. The config field name stays `NATSStoreDir` (Go code is honest about what it is).
hub.Hub.ServerName() — getter returning the live identity (Config.ServerName or the auto-generated name).
cmd: `edgesync hub serve --name=<server-name> --leaf-upstream=<url> ...` defaults --name to <hostname>-hub and repo path to ~/.edgesync/<name>/hub.repo.
docs/architecture/hub-package.md updated to reflect the new fields, the messaging/ default, and a generic repo-file naming convention (the package itself doesn't impose a filename; the CLI defaults to hub.repo).
Verified end-to-end: - Hub solicits a hub: bidirectional pub/sub flows across two hubs on this machine via the leafnode handshake. - 3-edge star: 1 root hub + 3 leaf processes — fan-in, cross-leaf, subject filter all behave as expected. - ServerName visible at server-side truth (h.server.Name()). - 31 tests pass, vet clean.
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> (user: [email protected], size: 2714) [annotate] [blame] [check-ins using]
package cli import ( "context" "fmt" "log/slog" "os" "os/signal" "path/filepath" "syscall" "github.com/danmestas/EdgeSync/hub" libfossilcli "github.com/danmestas/libfossil/cli" ) // HubCmd is the top-level command group for the in-process EdgeSync hub. type HubCmd struct { Serve HubServeCmd `cmd:"" help:"Run an embedded hub (NATS + fossil) at the machine level"` } type HubServeCmd struct { Name string `help:"NATS server name (defaults to <hostname>-hub)"` StoreDir string `help:"JetStream store dir (default sibling of repo)"` HTTPPort int `help:"Fossil HTTP port (0 = auto)" default:"0"` NATSClientPort int `help:"NATS client port (0 = auto)" default:"0"` NATSLeafPort int `help:"NATS leafnode port (0 = auto)" default:"0"` LeafUpstream string `help:"Solicit leafnode connection to this URL (e.g. nats-leaf://host:port)"` NobodyCaps string `help:"Caps for unauthenticated 'nobody' user (e.g. 'gio' for clone access)" default:"gio"` } func (c *HubServeCmd) Run(g *libfossilcli.Globals) error { name := c.Name if name == "" { host, err := os.Hostname() if err != nil { return fmt.Errorf("hostname: %w", err) } name = shortHost(host) + "-hub" } repoPath := g.Repo if repoPath == "" { home, err := os.UserHomeDir() if err != nil { return fmt.Errorf("home dir: %w", err) } repoPath = filepath.Join(home, ".edgesync", name, "hub.repo") if err := os.MkdirAll(filepath.Dir(repoPath), 0o755); err != nil { return fmt.Errorf("mkdir %s: %w", filepath.Dir(repoPath), err) } } ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer cancel() h, err := hub.NewHub(ctx, hub.Config{ RepoPath: repoPath, ServerName: name, NATSStoreDir: c.StoreDir, FossilHTTPPort: c.HTTPPort, NATSClientPort: c.NATSClientPort, NATSLeafPort: c.NATSLeafPort, LeafUpstream: c.LeafUpstream, NobodyCaps: c.NobodyCaps, }) if err != nil { return fmt.Errorf("hub: %w", err) } slog.Info("edgesync hub running", "name", h.ServerName(), "repo", repoPath, "http", "http://"+h.HTTPAddr(), "nats", h.NATSURL(), "leaf_upstream", h.LeafUpstream(), "sync_subject", h.FossilSyncSubject(), ) serveErr := make(chan error, 1) go func() { serveErr <- h.ServeHTTP(ctx) }() select { case <-ctx.Done(): case err := <-serveErr: if err != nil { slog.Error("hub HTTP server stopped", "err", err) } } slog.Info("shutting down hub", "name", name) return h.Stop() } // shortHost trims a FQDN to its first label, mirroring `hostname -s`. func shortHost(h string) string { for i := 0; i < len(h); i++ { if h[i] == '.' { return h[:i] } } return h }