Skip to content
edgesync Artifact [8eaaf58750]
Login

Artifact [8eaaf58750]

Check-ins Using Download Hex

Artifact 8eaaf58750c105b7e76d879b7481945dfece5c484f8cd1f027aa15770ae1fdd4:

// Package hub hosts an EdgeSync hub in-process: a fossil hub repo, an
// embedded NATS server with JetStream, and an HTTP server exposing the
// fossil timeline UI / xfer endpoint. All libfossil and nats-server types
// are wrapped internally — the public API exposes only stdlib types and
// types defined here, so consumers can host a hub without importing
// libfossil or nats-server.
//
// The package lives inside the EdgeSync root module rather than a separate
// go.mod so that it can ship in lockstep with the leaf agent without a
// chicken-and-egg release. It can be promoted to its own module later if
// downstream consumers need to depend on it independently.
package hub

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"sync"
	"time"

	libfossil "github.com/danmestas/libfossil"
	natsserver "github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
)

// Config configures NewHub.
type Config struct {
	// RepoPath is the absolute path to hub.fossil. Created if absent.
	RepoPath string

	// ServerName is the NATS server identity. Surfaces in leafnode
	// handshakes, varz, and gossip. Empty leaves nats-server's
	// auto-generated random name.
	ServerName string

	// BootstrapUser is the libfossil user created at hub bootstrap.
	// Default: "hub".
	BootstrapUser string

	// NATSStoreDir is the JetStream storage directory. When empty, defaults
	// to filepath.Dir(RepoPath) + "/messaging" — implementation-agnostic on
	// disk; the contents underneath are still NATS JetStream format.
	NATSStoreDir string

	// FossilHTTPPort is the port to serve fossil HTTP on. 0 = auto-pick.
	FossilHTTPPort int

	// NATSClientPort is the embedded NATS server's client port.
	// 0 = auto-pick.
	NATSClientPort int

	// NATSLeafPort is the leafnode listener port for remote agents.
	// 0 = auto-pick.
	NATSLeafPort int

	// EnableWebSocket turns on the embedded NATS server's WebSocket
	// listener. Off by default — opt-in for security (no TLS/WSS is wired
	// up here; see issue #176). When true the listener binds on
	// 127.0.0.1:NATSWebSocketPort with NoTLS=true.
	//
	// A separate bool (rather than overloading NATSWebSocketPort) is
	// required because 0 already means "auto-pick" for the sibling port
	// knobs, leaving no room for a "disabled" sentinel.
	EnableWebSocket bool

	// NATSWebSocketPort is the embedded NATS server's WebSocket listener
	// port when EnableWebSocket is true. 0 = auto-pick. Ignored when
	// EnableWebSocket is false.
	NATSWebSocketPort int

	// LeafUpstream is an optional upstream leafnode URL to solicit a
	// connection toward (e.g. "nats-leaf://hub.example:7422"). When set,
	// this hub acts as a leaf of that upstream — subjects published locally
	// are forwarded upstream and vice versa. Empty = no upstream solicit.
	LeafUpstream string

	// NobodyCaps grants the libfossil-pre-populated 'nobody' user these
	// caps after hub bootstrap. Empty leaves nobody at libfossil's defaults
	// (no caps; unauthenticated requests are rejected). Set to "gio" to
	// allow unauthenticated clone/pull/push, the typical hub deployment.
	NobodyCaps string

	// DisableFossilSyncOverNATS turns off the hub's NATS subscriber for the
	// fossil sync subject. By default, NewHub connects a local NATS client
	// to the embedded server and subscribes to "fossil.<project-code>.sync"
	// so peer leaf agents can push xfer requests to the hub over NATS
	// (alongside the HTTP path). Set to true for read-only HTTP-only
	// mirrors that don't accept NATS pushes.
	//
	// Inverted-name (Disable*) shape over the more obvious ServeFossilSyncOverNATS
	// because Go bool zero-value is false; the production default is "enabled,"
	// which maps cleanly to "DisableX defaults false."
	DisableFossilSyncOverNATS bool

	// FossilSyncSubjectPrefix overrides the NATS subject prefix used by the
	// fossil sync subscriber. Empty defaults to "fossil" (matching the
	// leaf-agent default). The full subscribed subjects are
	// "<prefix>.<project-code>.sync" (xfer request/reply) and
	// "<prefix>.<project-code>.commit" (notification on new commits).
	FossilSyncSubjectPrefix string

	// ProjectCode optionally pins the repo's project-code on first create
	// (passed to libfossil.CreateOpts). When non-empty, must be 40-char
	// lowercase hex (^[0-9a-f]{40}$).
	//
	// When the repo at RepoPath already exists, this value (if non-empty)
	// must match the on-disk project-code or NewHub returns an error —
	// guards against silent topology drift when a caller updates their
	// config but an old repo lingers.
	//
	// Empty preserves current behavior: generate on create, accept whatever's
	// on disk on open. Set this in topologies where multiple hubs/leaves
	// must share a project-code so commits propagate over the same NATS
	// sync subject.
	ProjectCode string

	// SeedFromUpstream, when set and RepoPath doesn't yet exist on disk,
	// clones from this URL before opening. The cloned repo carries the
	// upstream's project-code, so children declaring a SeedFromUpstream
	// don't typically also need to set ProjectCode (though setting both
	// and asserting they match is allowed and cross-checked).
	//
	// URL is an EdgeSync hub HTTP xfer endpoint (e.g. "http://hub.local:8080/").
	// NATS-native clone is deferred until libfossil HandleSync supports the
	// clone protocol upstream.
	SeedFromUpstream string

	// CheckpointInterval is how often the hub runs a PASSIVE WAL checkpoint
	// against hub.fossil while serving. PASSIVE never blocks readers or
	// writers, so the file stays readable by vanilla fossil tooling
	// (fossil ui / fossil info / fossil timeline) without coordinating
	// with the hub. Zero defaults to defaultCheckpointInterval.
	//
	// No "disable" knob: vanilla-fossil readability is the contract this
	// Config exists to uphold. Set a long interval if you really want to
	// dilute the cadence; it should never be off.
	CheckpointInterval time.Duration
}

// defaultCheckpointInterval is the period between background PASSIVE
// checkpoints when Config.CheckpointInterval is zero.
const defaultCheckpointInterval = 10 * time.Second

// Hub hosts an EdgeSync hub in-process. Construct one via NewHub, then call
// ServeHTTP in a goroutine. Use Stop to tear everything down.
//
// While serving, the hub runs a PASSIVE WAL checkpoint on hub.fossil every
// Config.CheckpointInterval (default 10s) so vanilla fossil tooling can read
// the file. After Stop returns nil, the on-disk hub.fossil is self-contained
// — libfossil's Close runs a TRUNCATE checkpoint, leaving no WAL/SHM
// sidecar.
type Hub struct {
	repo         *Repo
	server       *natsserver.Server
	httpListener net.Listener
	httpServer   *http.Server

	natsClient      *nats.Conn
	natsSyncSub     *nats.Subscription
	natsPeerSyncSub *nats.Subscription
	natsCommitSub   *nats.Subscription
	syncSubject     string
	peerSyncSubject string
	commitSubject   string
	peerID          string

	httpAddr     string
	natsURL      string
	leafURL      string
	websocketURL string

	checkpointStop chan struct{}
	checkpointDone chan struct{}

	catchupStop chan struct{}
	catchupDone chan struct{}

	ready chan struct{}

	stopOnce sync.Once
}

// publishCatchupDelay is how long after NewHub returns the hub waits
// before scanning past the persisted watermark and republishing any
// commits that landed while the hub was down. The delay gives NATS
// leaf clients (default ReconnectWait 2s in startFossilSyncSubscriber)
// time to reconnect after a hub restart on the same leaf port; without
// it, the catchup publish would land before peers reconnect and the
// notification would be lost.
//
// Bounded by the test window in TestCrossLeaf_HubRestart_RepublishesUnannouncedCommits;
// production tolerates a slightly longer wait if reconnect backoff is
// tuned wider.
const publishCatchupDelay = 3 * time.Second

// NewHub bootstraps the hub repo if absent, applies SQLite tunings, binds
// the HTTP listener (so HTTPAddr is immediately available), and starts the
// embedded NATS server. Does not yet serve HTTP — call ServeHTTP for that.
func NewHub(ctx context.Context, cfg Config) (*Hub, error) {
	if cfg.RepoPath == "" {
		return nil, errors.New("hub: Config.RepoPath is required")
	}
	if cfg.BootstrapUser == "" {
		cfg.BootstrapUser = "hub"
	}
	if cfg.NATSStoreDir == "" {
		cfg.NATSStoreDir = filepath.Join(filepath.Dir(cfg.RepoPath), "messaging")
	}

	if cfg.SeedFromUpstream != "" {
		if _, statErr := os.Stat(cfg.RepoPath); errors.Is(statErr, os.ErrNotExist) {
			if err := seedFromUpstream(ctx, cfg.RepoPath, cfg.SeedFromUpstream, cfg.ProjectCode); err != nil {
				return nil, err
			}
		}
	}

	libfossilRepo, err := openOrCreateRepo(cfg.RepoPath, cfg.BootstrapUser, cfg.ProjectCode)
	if err != nil {
		return nil, err
	}
	applySQLiteTuning(libfossilRepo)
	repo := newRepoFromHandle(libfossilRepo)

	if cfg.NobodyCaps != "" {
		if err := repo.SetUserCaps("nobody", cfg.NobodyCaps); err != nil {
			repo.Close()
			return nil, fmt.Errorf("hub: apply NobodyCaps: %w", err)
		}
	}

	httpAddr := fmt.Sprintf("127.0.0.1:%d", cfg.FossilHTTPPort)
	httpLn, err := net.Listen("tcp", httpAddr)
	if err != nil {
		repo.Close()
		return nil, fmt.Errorf("hub: bind HTTP listener on %s: %w", httpAddr, err)
	}

	natsServer, err := startNATS(cfg)
	if err != nil {
		httpLn.Close()
		repo.Close()
		return nil, err
	}

	leafAddr := ""
	if varz, vErr := natsServer.Varz(&natsserver.VarzOptions{}); vErr == nil && varz.LeafNode.Port > 0 {
		leafAddr = fmt.Sprintf("nats-leaf://127.0.0.1:%d", varz.LeafNode.Port)
	}

	wsURL := ""
	if cfg.EnableWebSocket {
		wsURL = natsServer.WebsocketURL()
	}

	h := &Hub{
		repo:         repo,
		server:       natsServer,
		httpListener: httpLn,
		httpAddr:     httpLn.Addr().String(),
		natsURL:      natsServer.ClientURL(),
		leafURL:      leafAddr,
		websocketURL: wsURL,
		ready:        make(chan struct{}),
	}

	// Build the HTTP mux + server here, in NewHub, rather than inside
	// ServeHTTP. Single-writer (this function), many-readers (ServeHTTP,
	// Stop). After NewHub returns, h.httpServer is immutable for the
	// hub's lifetime, so the Stop-vs-ServeHTTP read/write race that
	// existed when ServeHTTP wrote h.httpServer mid-goroutine is gone
	// by construction — no mutex, no atomic, no sync.Once required.
	// (Issue #187.)
	mux := http.NewServeMux()
	xfer := h.repo.handle.XferHandler()
	// Wrap XferHandler so commits arriving via an external `fossil push
	// <hub-http>` (which writes artifacts straight into the SQLite repo
	// via libfossil's xfer protocol, bypassing Repo.Commit) still trigger
	// the .commit auto-publish that Repo.Commit installs. publishNewCommits
	// is a no-op when nothing new landed (typical for pull/clone) or when
	// the publish hook isn't wired (DisableFossilSyncOverNATS), so it's
	// safe to fire on every request. Issue #160.
	mux.Handle("/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		xfer.ServeHTTP(w, r)
		h.repo.publishNewCommits()
	}))
	h.httpServer = &http.Server{Handler: mux}

	if !cfg.DisableFossilSyncOverNATS {
		if err := h.startFossilSyncSubscriber(cfg); err != nil {
			natsServer.Shutdown()
			natsServer.WaitForShutdown()
			httpLn.Close()
			repo.Close()
			return nil, err
		}
		if err := h.startCommitSubscriber(cfg); err != nil {
			if h.natsSyncSub != nil {
				_ = h.natsSyncSub.Unsubscribe()
			}
			if h.natsPeerSyncSub != nil {
				_ = h.natsPeerSyncSub.Unsubscribe()
			}
			if h.natsClient != nil {
				h.natsClient.Close()
			}
			natsServer.Shutdown()
			natsServer.WaitForShutdown()
			httpLn.Close()
			repo.Close()
			return nil, err
		}
	}

	h.startCheckpointLoop(cfg.CheckpointInterval)

	return h, nil
}

// startCheckpointLoop runs PASSIVE WAL checkpoints against the hub repo on a
// ticker so vanilla fossil tooling can read hub.fossil while the hub is
// serving. PASSIVE never blocks readers or writers; checkpoint errors are
// non-fatal (the next tick retries, and Hub.Stop's repo.Close runs a
// TRUNCATE checkpoint anyway).
func (h *Hub) startCheckpointLoop(interval time.Duration) {
	if interval <= 0 {
		interval = defaultCheckpointInterval
	}
	h.checkpointStop = make(chan struct{})
	h.checkpointDone = make(chan struct{})
	go func() {
		defer close(h.checkpointDone)
		t := time.NewTicker(interval)
		defer t.Stop()
		for {
			select {
			case <-h.checkpointStop:
				return
			case <-t.C:
				_ = h.repo.handle.Checkpoint(libfossil.CheckpointPassive)
			}
		}
	}()
}

// startFossilSyncSubscriber connects a local NATS client to the embedded
// server and subscribes to two subjects for incoming xfer payloads,
// dispatching both to the hub repo's HandleSync:
//
//  1. "<prefix>.<project-code>.sync" — the broadcast subject used by
//     external leaf-agent pushes. Joined as a NATS queue group keyed on
//     the project-code so that across a multi-hub cluster only ONE hub
//     responds to any given broadcast request. Without the queue group,
//     every hub on the project would answer and nats.Request would
//     non-deterministically pick the first reply — see #162.
//
//  2. "<prefix>.<project-code>.sync.<peer-id>" — a per-peer addressed
//     subject. Used by hub-to-hub pulls triggered by .commit
//     notifications: the puller addresses the originating peer
//     directly so the response is deterministic regardless of how many
//     other hubs share the project-code.
//
// peerID is the hub's NATS server name. Reconnects keep the same server
// name because the embedded server's name is stable across the client's
// reconnect window. Invoked by NewHub when Config.DisableFossilSyncOverNATS
// is false.
func (h *Hub) startFossilSyncSubscriber(cfg Config) error {
	projectCode, err := h.repo.handle.Config("project-code")
	if err != nil {
		return fmt.Errorf("hub: read project-code for fossil-sync subscriber: %w", err)
	}
	if projectCode == "" {
		return fmt.Errorf("hub: project-code is empty; refusing to start fossil-sync subscriber")
	}

	prefix := cfg.FossilSyncSubjectPrefix
	if prefix == "" {
		prefix = "fossil"
	}
	subject := prefix + "." + projectCode + ".sync"
	peerID := h.server.Name()
	if peerID == "" {
		return fmt.Errorf("hub: NATS server name is empty; cannot derive peer-id for addressed .sync subject")
	}
	peerSubject := subject + "." + peerID

	nc, err := nats.Connect(h.natsURL,
		nats.Name("edgesync-hub"),
		nats.MaxReconnects(-1),
		nats.ReconnectWait(2*time.Second),
		nats.RetryOnFailedConnect(true),
		nats.Timeout(5*time.Second),
		nats.NoEcho(), // we publish on .commit + .sync ourselves; don't self-loop
	)
	if err != nil {
		return fmt.Errorf("hub: connect local NATS for fossil-sync subscriber: %w", err)
	}

	handle := func(msg *nats.Msg) {
		respBytes, hsErr := h.repo.handle.HandleSync(context.Background(), msg.Data)
		if hsErr != nil {
			// Empty response signals failure to the caller without hanging.
			_ = msg.Respond([]byte{})
			return
		}
		_ = msg.Respond(respBytes)
	}

	// Queue-group scope is the project-code so cross-project hub clusters
	// (unlikely today but cheap to guarantee) don't share a queue.
	queue := "sync-" + projectCode
	sub, err := nc.QueueSubscribe(subject, queue, handle)
	if err != nil {
		nc.Close()
		return fmt.Errorf("hub: queue-subscribe %s: %w", subject, err)
	}

	peerSub, err := nc.Subscribe(peerSubject, handle)
	if err != nil {
		_ = sub.Unsubscribe()
		nc.Close()
		return fmt.Errorf("hub: subscribe %s: %w", peerSubject, err)
	}

	// Flush so the server has registered the subscriptions before NewHub
	// returns. Tests that publish immediately after NewHub depend on this.
	if err := nc.FlushTimeout(2 * time.Second); err != nil {
		_ = sub.Unsubscribe()
		_ = peerSub.Unsubscribe()
		nc.Close()
		return fmt.Errorf("hub: flush fossil-sync subscription: %w", err)
	}

	h.natsClient = nc
	h.natsSyncSub = sub
	h.natsPeerSyncSub = peerSub
	h.syncSubject = subject
	h.peerSyncSubject = peerSubject
	h.peerID = peerID
	return nil
}

// commitNotification is the JSON payload published on the .commit subject
// when a new commit lands. Peers decode it and pull the manifest from the
// originating peer via the addressed "<prefix>.<project-code>.sync.<peer-id>"
// subject. PeerID is the publisher hub's NATS server name; pullers MUST
// route requests through it rather than the broadcast .sync subject to
// avoid the multi-responder race documented in #162.
type commitNotification struct {
	RID    int64  `json:"rid"`
	UUID   string `json:"uuid"`
	PeerID string `json:"peer_id"`
}

// startCommitSubscriber subscribes to "<prefix>.<project-code>.commit" and,
// on receipt, triggers an xfer pull from peers on the sibling .sync
// subject. Also wires the hub repo's publish hook so successful commits
// publish a notification on the .commit subject.
//
// Requires startFossilSyncSubscriber to have run first (uses h.natsClient
// and shares the project-code / prefix derivation).
func (h *Hub) startCommitSubscriber(cfg Config) error {
	projectCode, err := h.repo.handle.Config("project-code")
	if err != nil {
		return fmt.Errorf("hub: read project-code for commit subscriber: %w", err)
	}
	if projectCode == "" {
		return fmt.Errorf("hub: project-code is empty; refusing to start commit subscriber")
	}
	prefix := cfg.FossilSyncSubjectPrefix
	if prefix == "" {
		prefix = "fossil"
	}
	commitSubject := prefix + "." + projectCode + ".commit"
	syncSubject := prefix + "." + projectCode + ".sync"
	ownPeerID := h.peerID

	sub, err := h.natsClient.Subscribe(commitSubject, func(msg *nats.Msg) {
		var n commitNotification
		if jsonErr := json.Unmarshal(msg.Data, &n); jsonErr != nil {
			log.Printf("hub commit subscriber: decode notification: %v", jsonErr)
			return
		}
		// PeerID is required: without it we'd have to fall back to the
		// broadcast .sync subject, which is exactly the multi-responder
		// race #162 fixes. Drop notifications missing it (would only
		// happen against an older publisher that pre-dates this fix).
		if n.PeerID == "" {
			log.Printf("hub commit subscriber: notification missing peer_id rid=%d uuid=%s; refusing broadcast pull (see #162)", n.RID, n.UUID)
			return
		}
		// Self-notifications can happen even with NoEcho when a
		// publish from this hub round-trips back via an upstream leaf.
		// Skip them so we don't try to pull from ourselves.
		if n.PeerID == ownPeerID {
			return
		}
		// Address the pull at the publishing peer directly via its
		// per-peer .sync subject so exactly that peer responds — no
		// race with stale subscribers on the broadcast subject.
		peerSyncSubject := syncSubject + "." + n.PeerID
		pullTransport := newNATSTransport(h.natsClient, peerSyncSubject, 0)
		// Pull is naturally idempotent — if we already have the uuid,
		// the sync round converges with zero rounds / zero blobs. So
		// we skip the up-front existence check and let the protocol
		// handle dedup.
		pullCtx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
		defer cancel()
		if _, syncErr := h.repo.handle.Sync(pullCtx, pullTransport, libfossil.SyncOpts{
			Pull:        true,
			ProjectCode: projectCode,
			PeerID:      "edgesync-hub",
		}); syncErr != nil {
			log.Printf("hub commit subscriber: pull triggered by notification rid=%d uuid=%s peer=%s: %v", n.RID, n.UUID, n.PeerID, syncErr)
		}
	})
	if err != nil {
		return fmt.Errorf("hub: subscribe %s: %w", commitSubject, err)
	}
	if flushErr := h.natsClient.FlushTimeout(2 * time.Second); flushErr != nil {
		_ = sub.Unsubscribe()
		return fmt.Errorf("hub: flush commit subscription: %w", flushErr)
	}

	// Wire publish hook on the repo so future commits notify peers.
	// PeerID is required so receivers can address their pull-back at
	// this specific hub rather than fanning out a broadcast request
	// that's vulnerable to the #162 multi-responder race.
	nc := h.natsClient
	h.repo.publish = func(rid int64, uuid string) {
		payload, jsonErr := json.Marshal(commitNotification{RID: rid, UUID: uuid, PeerID: ownPeerID})
		if jsonErr != nil {
			log.Printf("hub commit publish: encode notification rid=%d uuid=%s: %v", rid, uuid, jsonErr)
			return
		}
		if pubErr := nc.Publish(commitSubject, payload); pubErr != nil {
			log.Printf("hub commit publish: publish to %s rid=%d uuid=%s: %v", commitSubject, rid, uuid, pubErr)
		}
	}
	// Restore the watermark from the persisted value if present, else
	// snapshot the current max ci rid. Schedule a delayed catchup that
	// republishes any commits past the persisted watermark — handles
	// the crash-mid-publish case (commit landed via HTTP push, hub
	// died before the .commit notification went out) and any commit
	// made by an offline tool against the same repo file while the
	// hub was down.
	//
	// The delay matters: an immediate catchup would fire before NATS
	// leaf clients reconnect after a hub restart on the same leaf port,
	// and the notification would land in the void. publishCatchupDelay
	// gives clients time to reconnect first.
	h.repo.initPublishWatermark()
	h.startPublishCatchup()

	h.natsCommitSub = sub
	h.commitSubject = commitSubject
	return nil
}

// startPublishCatchup schedules a one-shot publishNewCommits call after
// publishCatchupDelay. Caller-cancellable via the catchupStop channel
// owned by Hub so Stop tears it down cleanly even if it fires before the
// delay elapses.
//
// Best-effort by contract: if no peer reconnects within the delay window
// the catchup publish is lost (NATS Core is fire-and-forget with no
// per-subscriber queuing). Next commit through Repo.Commit or the HTTP
// xfer-push wrapper triggers another scan, so peers eventually catch up
// on activity; a long quiet period after a crash with no peers connected
// leaves the backlog hanging. Tracked as a follow-up — JetStream-durable
// .commit subject would close the gap.
func (h *Hub) startPublishCatchup() {
	h.catchupStop = make(chan struct{})
	h.catchupDone = make(chan struct{})
	go func() {
		defer close(h.catchupDone)
		select {
		case <-h.catchupStop:
			return
		case <-time.After(publishCatchupDelay):
			h.repo.publishNewCommits()
		}
	}()
}

// ServeHTTP runs the fossil HTTP server (timeline UI + xfer endpoint)
// against the listener bound by NewHub. Blocks until ctx is cancelled or
// Stop is called.
//
// The http.Server itself is constructed by NewHub; ServeHTTP only consumes
// it. This is by design: a single writer (NewHub) and many readers
// (ServeHTTP, Stop, …) means no concurrent write on h.httpServer, so no
// race between Stop and a not-yet-started ServeHTTP. Issue #187.
func (h *Hub) ServeHTTP(ctx context.Context) error {
	assert(h.httpServer != nil, "ServeHTTP called before NewHub finished (h.httpServer == nil)")
	assert(h.httpListener != nil, "ServeHTTP called before NewHub finished (h.httpListener == nil)")
	srv := h.httpServer

	go func() {
		<-ctx.Done()
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = srv.Shutdown(shutdownCtx)
	}()

	close(h.ready)

	if err := srv.Serve(h.httpListener); err != nil && err != http.ErrServerClosed {
		return fmt.Errorf("hub: serve HTTP: %w", err)
	}
	return nil
}

// Ready returns a channel closed once ServeHTTP has finished configuring
// the HTTP server and is about to enter the Accept loop. Callers that
// advertise the hub's URLs externally — e.g. session state JSON or peer
// discovery — should wait on Ready before publishing, to close the window
// where the listener is bound at the kernel level but the goroutine running
// ServeHTTP hasn't started processing requests yet.
//
// The TCP listener is bound earlier (inside NewHub), so TCP-level connects
// succeed immediately after NewHub returns. Ready closes after the
// http.Server is wired up and just before srv.Serve runs, giving consumers
// a clean barrier for "HTTP request processing is about to be live."
//
// Safe to call from any goroutine; safe to call multiple times. Returns
// the same channel each time. The channel never closes if ServeHTTP is
// never called.
func (h *Hub) Ready() <-chan struct{} { return h.ready }

// Stop tears down the embedded NATS server and any HTTP listeners. Safe to
// call multiple times.
//
// h.httpServer and h.httpListener are populated by NewHub and immutable for
// the hub's lifetime, so Stop reads them without coordination with a
// concurrent ServeHTTP. The asserts surface a programmer-bug if Stop is
// somehow called against a Hub the constructor didn't finish (today
// impossible — NewHub only returns a *Hub on success). Issue #187.
func (h *Hub) Stop() error {
	assert(h.httpServer != nil, "Stop called on a Hub NewHub did not finish constructing (h.httpServer == nil)")
	assert(h.httpListener != nil, "Stop called on a Hub NewHub did not finish constructing (h.httpListener == nil)")
	var firstErr error
	h.stopOnce.Do(func() {
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		if err := h.httpServer.Shutdown(shutdownCtx); err != nil && err != http.ErrServerClosed {
			firstErr = fmt.Errorf("hub: shutdown HTTP: %w", err)
		}
		// Shutdown only closes listeners srv.Serve was actually called
		// against. If ServeHTTP never ran, the bound listener from
		// NewHub is still open; close it explicitly. Idempotent — a
		// second Close on an already-closed listener returns an error
		// which we deliberately ignore (Stop is itself idempotent via
		// stopOnce above).
		_ = h.httpListener.Close()
		if h.natsCommitSub != nil {
			_ = h.natsCommitSub.Unsubscribe()
		}
		if h.natsSyncSub != nil {
			_ = h.natsSyncSub.Unsubscribe()
		}
		if h.natsPeerSyncSub != nil {
			_ = h.natsPeerSyncSub.Unsubscribe()
		}
		if h.natsClient != nil {
			h.natsClient.Close()
		}
		if h.server != nil {
			h.server.Shutdown()
			h.server.WaitForShutdown()
		}
		if h.catchupStop != nil {
			close(h.catchupStop)
			<-h.catchupDone
		}
		if h.checkpointStop != nil {
			close(h.checkpointStop)
			<-h.checkpointDone
		}
		if h.repo != nil {
			if err := h.repo.Close(); err != nil && firstErr == nil {
				firstErr = err
			}
		}
	})
	return firstErr
}

// NATSURL returns the embedded NATS server's client URL.
func (h *Hub) NATSURL() string { return h.natsURL }

// ServerName returns the NATS server identity (Config.ServerName, or the
// auto-generated random name if Config.ServerName was empty).
func (h *Hub) ServerName() string { return h.server.Name() }

// LeafURL returns this hub's leafnode listener URL — the URL remote leaves
// dial as their upstream. Empty if no leaf port is configured. Distinct
// from Config.LeafUpstream, which is the input URL this hub itself solicits.
func (h *Hub) LeafURL() string { return h.leafURL }

// NATSWebSocketURL returns the embedded NATS server's WebSocket client URL
// (e.g. "ws://127.0.0.1:NNNN") when Config.EnableWebSocket is true, or "" when
// WebSocket is disabled (the default). No TLS/WSS is wired up — see issue #176.
func (h *Hub) NATSWebSocketURL() string { return h.websocketURL }

// HTTPAddr returns the host:port the fossil HTTP server is bound to.
func (h *Hub) HTTPAddr() string { return h.httpAddr }

// NumLeafs returns the current count of active leafnode connections to
// this hub. Consumers can poll this to implement auto-shutdown behavior
// when the last remote leaf disconnects (e.g. sesh's "hub runs until the
// last session closes" lifecycle).
func (h *Hub) NumLeafs() int { return h.server.NumLeafNodes() }

// FossilSyncSubject returns the NATS subject the hub subscribes to for
// incoming fossil-sync xfer requests, or "" if Config.DisableFossilSyncOverNATS
// was set. Format: "<prefix>.<project-code>.sync".
func (h *Hub) FossilSyncSubject() string { return h.syncSubject }

// FossilCommitSubject returns the NATS subject the hub publishes on
// after a successful commit, and subscribes to for peer notifications.
// Empty if Config.DisableFossilSyncOverNATS was set. Format:
// "<prefix>.<project-code>.commit".
func (h *Hub) FossilCommitSubject() string { return h.commitSubject }

// Repo returns the underlying fossil-repo handle the hub serves from. Use
// this to share the handle with another component in the same process —
// e.g. a coexisting CLI command path that needs user/commit/read access
// without re-opening the file. Don't call Close on the returned *Repo
// while the Hub is still serving; use Hub.Stop for teardown.
func (h *Hub) Repo() *Repo { return h.repo }

func openOrCreateRepo(path, bootstrapUser, projectCode string) (*libfossil.Repo, error) {
	if r, err := libfossil.Open(path); err == nil {
		if projectCode != "" {
			onDisk, cfgErr := r.Config("project-code")
			if cfgErr != nil {
				_ = r.Close()
				return nil, fmt.Errorf("hub: read project-code from existing repo at %s: %w", path, cfgErr)
			}
			if onDisk != projectCode {
				_ = r.Close()
				return nil, fmt.Errorf("hub: existing repo at %s has project-code %q, Config.ProjectCode is %q (config drift)", path, onDisk, projectCode)
			}
		}
		return r, nil
	}
	r, err := libfossil.Create(path, libfossil.CreateOpts{
		User:        bootstrapUser,
		ProjectCode: projectCode,
	})
	if err != nil {
		return nil, fmt.Errorf("hub: create repo at %s: %w", path, err)
	}
	return r, nil
}

func startNATS(cfg Config) (*natsserver.Server, error) {
	opts := &natsserver.Options{
		Host:       "127.0.0.1",
		Port:       portOrAuto(cfg.NATSClientPort),
		ServerName: cfg.ServerName,
		NoLog:      true,
		NoSigs:     true,
		JetStream:  true,
		StoreDir:   cfg.NATSStoreDir,
	}
	opts.LeafNode.Host = "127.0.0.1"
	opts.LeafNode.Port = portOrAuto(cfg.NATSLeafPort)

	if cfg.EnableWebSocket {
		opts.Websocket = natsserver.WebsocketOpts{
			Host:  "127.0.0.1",
			Port:  portOrAuto(cfg.NATSWebSocketPort),
			NoTLS: true, // TLS/WSS out of scope — issue #176
		}
	}

	if cfg.LeafUpstream != "" {
		u, err := url.Parse(cfg.LeafUpstream)
		if err != nil {
			return nil, fmt.Errorf("hub: parse LeafUpstream %q: %w", cfg.LeafUpstream, err)
		}
		opts.LeafNode.Remotes = []*natsserver.RemoteLeafOpts{
			{URLs: []*url.URL{u}},
		}
	}

	srv, err := natsserver.NewServer(opts)
	if err != nil {
		return nil, fmt.Errorf("hub: create NATS server: %w", err)
	}
	srv.Start()
	if !srv.ReadyForConnections(5 * time.Second) {
		srv.Shutdown()
		return nil, fmt.Errorf("hub: NATS server not ready within 5s")
	}
	return srv, nil
}

func portOrAuto(p int) int {
	if p == 0 {
		return -1 // nats-server convention for "auto-pick"
	}
	return p
}

// applySQLiteTuning sets a 30s SQLite busy_timeout so concurrent operations
// retry on SQLITE_BUSY rather than failing fast.
//
// Mirrors leaf/agent's tuning. Notably does NOT cap MaxOpenConns: hub
// repos serve concurrent clones, and capping the pool deadlocks libfossil's
// clone path (issue #120).
func applySQLiteTuning(r *libfossil.Repo) {
	_, _ = r.DB().Exec(`PRAGMA busy_timeout = 30000`)
}