package agent
import (
"context"
"crypto/sha256"
"fmt"
"log/slog"
"net"
"net/http"
"os"
"os/exec"
"path/filepath"
"time"
libfossil "github.com/danmestas/libfossil"
"github.com/danmestas/libfossil/simio"
"github.com/danmestas/EdgeSync/leaf/agent/notify"
"github.com/nats-io/nats.go"
)
// irohALPN is the ALPN protocol identifier for EdgeSync xfer over iroh.
const irohALPN = "/edgesync/xfer/1"
// syncTarget represents a single sync destination (NATS, iroh peer, etc.).
type syncTarget struct {
transport libfossil.Transport
label string // "nats", "iroh:<endpoint>", etc.
}
// Event represents an input to the agent state machine.
type Event int
const (
EventTimer Event = iota // poll interval elapsed
EventSyncNow // manual sync trigger
EventStop // shutdown requested
)
// ActionType classifies the result of processing an event.
type ActionType int
const (
ActionNone ActionType = iota
ActionSynced // a sync round was executed
ActionStopped // agent acknowledged stop
)
// Action is the result of processing a single event via Tick.
type Action struct {
Type ActionType
Result *libfossil.SyncResult // non-nil when Type == ActionSynced
Err error // non-nil on sync failure
}
// Agent manages the lifecycle of a leaf sync daemon: it opens a Fossil repo,
// connects to NATS, and periodically syncs via the xfer protocol.
type Agent struct {
config Config
clock simio.Clock
repo *libfossil.Repo
mesh *NATSMesh // nil when created via NewFromParts
conn *nats.Conn // nil when created via NewFromParts
transport libfossil.Transport
syncTargets []syncTarget // unified list of sync destinations
projectCode string
serverCode string
cancel context.CancelFunc
done chan struct{}
syncNow chan struct{} // buffer 1
irohSidecar *sidecar // nil when iroh is disabled
irohSock string // Unix socket path for iroh sidecar
irohEndpointID string // local iroh endpoint ID (set after sidecar ready)
notifyRepo *libfossil.Repo // nil when notify is disabled
notifySvc *notify.Service // nil when notify is disabled
}
// New creates a new Agent from the given configuration.
// It opens the Fossil repo and reads project-code / server-code.
// The embedded NATS server is NOT started here — that happens in Start()
// after iroh tunnels are established (tunnel ports must be known before
// NATS can be configured with leaf remotes).
func New(cfg Config) (*Agent, error) {
cfg.applyDefaults()
if err := cfg.validate(); err != nil {
return nil, err
}
r, err := libfossil.Open(cfg.RepoPath)
if err != nil {
return nil, fmt.Errorf("agent: open repo: %w", err)
}
projectCode, err := r.Config("project-code")
if err != nil {
r.Close()
return nil, fmt.Errorf("agent: read project-code: %w", err)
}
serverCode, err := r.Config("server-code")
if err != nil {
r.Close()
return nil, fmt.Errorf("agent: read server-code: %w", err)
}
// Create the mesh but don't start it yet. Start() will call
// startIrohSidecar (which establishes tunnels and populates
// tunnelPorts) before starting the NATS server.
mesh := &NATSMesh{
role: cfg.NATSRole,
upstream: cfg.NATSUpstream,
irohPeers: cfg.IrohPeers,
clientPort: cfg.NATSClientPort,
storeDir: cfg.NATSStoreDir,
}
a := &Agent{
config: cfg,
clock: cfg.Clock,
repo: r,
mesh: mesh,
projectCode: projectCode,
serverCode: serverCode,
syncNow: make(chan struct{}, 1),
}
// Initialize peer registry (log warnings, don't fail startup).
if err := a.ensurePeerRegistry(); err != nil {
a.logf("warning: ensurePeerRegistry: %v", err)
}
if err := a.seedPeerRegistry(); err != nil {
a.logf("warning: seedPeerRegistry: %v", err)
}
return a, nil
}
// NewFromParts creates an Agent from pre-built components without performing
// any I/O. Used by tests and the deterministic simulation harness.
func NewFromParts(cfg Config, r *libfossil.Repo, t libfossil.Transport, projectCode, serverCode string) *Agent {
cfg.applyDefaults()
return &Agent{
config: cfg,
clock: cfg.Clock,
repo: r,
transport: t,
projectCode: projectCode,
serverCode: serverCode,
syncNow: make(chan struct{}, 1),
}
}
// Tick processes a single event and returns the resulting action.
// This is the core state machine entry point. In simulation, the caller
// drives events directly via the primary transport. In production,
// pollLoop iterates all syncTargets.
func (a *Agent) Tick(ctx context.Context, ev Event) Action {
switch ev {
case EventTimer:
// BUGGIFY: skip a timer-triggered sync to test stale-state behavior.
if a.config.Buggify != nil && a.config.Buggify.Check("agent.runSync.earlyReturn", 0.05) {
return Action{Type: ActionNone}
}
result, err := a.runSync(ctx)
return Action{Type: ActionSynced, Result: result, Err: err}
case EventSyncNow:
result, err := a.runSync(ctx)
return Action{Type: ActionSynced, Result: result, Err: err}
case EventStop:
return Action{Type: ActionStopped}
default:
return Action{Type: ActionNone}
}
}
func (a *Agent) logf(format string, args ...any) {
if a.config.Logger != nil {
a.config.Logger(fmt.Sprintf(format, args...))
}
}
// Repo returns the agent's Fossil repository (for invariant checking in simulation).
func (a *Agent) Repo() *libfossil.Repo {
return a.repo
}
// Start launches the background poll loop and any configured server
// listeners. Call Stop to shut everything down.
//
// The startup order is critical for correct NATS leaf node handshakes:
// 1. Start iroh sidecar and establish tunnels (get local listener ports)
// 2. Start embedded NATS server with tunnel ports as leaf remotes
// 3. Connect NATS client and configure sync targets
// 4. Start HTTP/NATS serve listeners
// 5. Launch poll loop
func (a *Agent) Start() error {
a.logf("starting agent...")
ctx, cancel := context.WithCancel(context.Background())
a.cancel = cancel
// Step 1: iroh sidecar + tunnels (must happen before NATS start).
if err := a.startIrohSidecar(ctx); err != nil {
cancel()
return err
}
// Step 2+3: start embedded NATS mesh and connect client.
if err := a.startNATSMesh(); err != nil {
cancel()
return err
}
// Step 4: serve listeners.
if err := a.startNATSServe(ctx); err != nil {
cancel()
return err
}
if err := a.startHTTPServer(ctx); err != nil {
cancel()
return err
}
// Step 5: notify messaging service (after NATS is connected).
if err := a.startNotify(); err != nil {
a.logf("warning: notify service failed to start: %v", err)
// Non-fatal — agent still works without notify.
}
a.done = make(chan struct{})
go a.pollLoop(ctx)
return nil
}
// startNotify opens the notify.fossil repo and creates the notify Service
// with the agent's NATS connection. Non-fatal if notify.fossil doesn't exist.
func (a *Agent) startNotify() error {
if !a.config.NotifyEnabled {
return nil
}
path := a.config.NotifyRepoPath
r, err := libfossil.Open(path)
if err != nil {
return fmt.Errorf("open notify repo %s: %w", path, err)
}
a.notifyRepo = r
fromName := a.config.PeerID
from := a.irohEndpointID
if from == "" {
from = a.config.PeerID
}
svc, err := notify.NewService(notify.ServiceConfig{
Repo: r,
NATSConn: a.conn,
From: from,
FromName: fromName,
})
if err != nil {
r.Close()
a.notifyRepo = nil
return fmt.Errorf("create notify service: %w", err)
}
a.notifySvc = svc
a.logf("notify service started (repo=%s, nats=%v)", path, a.conn != nil)
return nil
}
// startNATSMesh starts the embedded NATS server and connects the NATS client.
// If iroh is active, passes sidecar info so the mesh can establish tunnels.
func (a *Agent) startNATSMesh() error {
if a.mesh == nil {
return nil
}
var sc *SidecarInfo
if a.irohSidecar != nil && a.irohEndpointID != "" {
sc = &SidecarInfo{
EndpointID: a.irohEndpointID,
SocketPath: a.irohSock,
}
}
clientURL, err := a.mesh.Start(sc)
if err != nil {
return fmt.Errorf("agent: nats mesh start: %w", err)
}
cfg := a.config
natsOpts := []nats.Option{
nats.Name("edgesync-leaf"),
nats.MaxReconnects(-1),
nats.ReconnectWait(2 * time.Second),
nats.RetryOnFailedConnect(true),
}
if cfg.CustomDialer != nil {
natsOpts = append(natsOpts, nats.SetCustomDialer(cfg.CustomDialer))
}
natsOpts = append(natsOpts, nats.DisconnectErrHandler(func(_ *nats.Conn, err error) {
if err != nil {
slog.Warn("NATS disconnected", "error", err)
if cfg.Logger != nil {
cfg.Logger("NATS disconnected: " + err.Error())
}
}
}))
natsOpts = append(natsOpts, nats.ReconnectHandler(func(_ *nats.Conn) {
slog.Info("NATS reconnected", "url", clientURL)
if cfg.Logger != nil {
cfg.Logger("NATS reconnected")
}
}))
nc, err := nats.Connect(clientURL, natsOpts...)
if err != nil {
a.mesh.Stop()
return fmt.Errorf("agent: nats connect: %w", err)
}
if nc.IsConnected() {
slog.Info("connected to NATS", "url", clientURL, "upstream", cfg.NATSUpstream, "role", cfg.NATSRole)
if cfg.Logger != nil {
cfg.Logger("connected to NATS: " + clientURL)
}
} else {
slog.Warn("NATS not yet reachable, will retry in background", "url", clientURL)
if cfg.Logger != nil {
cfg.Logger("warning: NATS not yet reachable at " + clientURL + ", will retry in background")
}
}
a.conn = nc
a.transport = NewNATSTransport(nc, a.projectCode, 0, cfg.SubjectPrefix)
return nil
}
// startNATSServe adds the NATS transport as a sync target and launches the
// NATS request/reply listener if configured.
func (a *Agent) startNATSServe(ctx context.Context) error {
// Add NATS as the primary sync target (always present when agent is built via New).
if a.transport != nil {
a.syncTargets = append(a.syncTargets, syncTarget{
transport: a.transport,
label: "nats",
})
}
if a.config.ServeNATSEnabled && a.conn != nil {
go func() {
subject := a.config.SubjectPrefix + "." + a.projectCode + ".sync"
if err := ServeNATS(ctx, a.conn, subject, a.repo); err != nil {
slog.Error("serve-nats stopped", "error", err)
}
}()
}
return nil
}
// startHTTPServer launches the public HTTP server if ServeHTTPAddr is set.
func (a *Agent) startHTTPServer(ctx context.Context) error {
if a.config.ServeHTTPAddr == "" {
return nil
}
go func() {
if err := a.serveHTTP(ctx); err != nil {
slog.Error("serve-http stopped", "error", err)
}
}()
return nil
}
// startIrohSidecar launches the iroh sidecar process, establishes outbound
// NATS tunnels, and registers iroh peers as sync targets. It creates a
// dedicated callback listener for sidecar-to-agent xfer traffic (decoupled
// from the public HTTP server).
//
// This runs BEFORE the NATS server starts. The leaf port is pre-reserved so
// the sidecar knows where NATS will listen for inbound leaf connections.
// Outbound tunnel ports are collected so the NATS server can be started with
// them as leaf remotes.
func (a *Agent) startIrohSidecar(ctx context.Context) error {
if !a.config.IrohEnabled {
return nil
}
binPath, err := a.findIrohBinary()
if err != nil {
return fmt.Errorf("agent: iroh: %w", err)
}
// Pre-reserve the NATS leaf port so we can tell the sidecar where NATS
// will listen before NATS actually starts. NATS will bind this exact port.
if a.mesh != nil {
if err := a.mesh.ReserveLeafPort(); err != nil {
return fmt.Errorf("agent: reserve leaf port: %w", err)
}
}
// Use a hash of the repo path to make the socket unique per agent instance,
// so multiple agents in the same process (tests) don't collide.
sockHash := fmt.Sprintf("%x", sha256.Sum256([]byte(a.config.RepoPath)))[:12]
a.irohSock = fmt.Sprintf("/tmp/iroh-%s.sock", sockHash)
// Always create a dedicated callback listener for sidecar-to-agent traffic.
// This is internal plumbing -- not the agent's public HTTP server.
callbackLn, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
return fmt.Errorf("agent: iroh callback listener: %w", err)
}
callbackMux := http.NewServeMux()
callbackMux.Handle("/", a.repo.XferHandler())
callbackSrv := &http.Server{Handler: callbackMux}
go func() {
<-ctx.Done()
callbackSrv.Close()
}()
go func() {
if err := callbackSrv.Serve(callbackLn); err != nil && err != http.ErrServerClosed {
slog.Error("iroh callback server stopped", "error", err)
}
}()
callbackURL := fmt.Sprintf("http://127.0.0.1:%d", callbackLn.Addr().(*net.TCPAddr).Port)
a.irohSidecar = &sidecar{
binPath: binPath,
socketPath: a.irohSock,
keyPath: a.config.IrohKeyPath,
callbackURL: callbackURL,
alpn: irohALPN,
}
// Pass the pre-reserved leaf port so the sidecar can accept inbound NATS
// leaf connections from remote peers.
if a.mesh != nil && a.mesh.LeafAddr() != "" {
a.irohSidecar.natsAddr = a.mesh.LeafAddr()
}
status, err := a.irohSidecar.Start(10 * time.Second)
if err != nil {
callbackSrv.Close()
return fmt.Errorf("agent: iroh sidecar: %w", err)
}
a.irohEndpointID = status.EndpointID
a.logf("iroh sidecar ready, endpoint_id=%s", status.EndpointID)
// Tunnel establishment is now handled by mesh.Start(SidecarInfo) in
// startNATSMesh. The mesh uses the EndpointID and socket path to
// establish tunnels before starting the NATS server.
// Register iroh peers as sync targets.
for _, peerID := range a.config.IrohPeers {
a.syncTargets = append(a.syncTargets, syncTarget{
transport: NewIrohTransport(a.irohSock, peerID),
label: "iroh:" + peerID,
})
}
// Monitor sidecar process liveness.
go a.monitorSidecar(ctx)
return nil
}
// Stop cancels the poll loop, waits for it to finish, closes the NATS
// connection, shuts down the mesh, and closes the repo.
func (a *Agent) Stop() error {
if a.cancel != nil {
a.cancel()
}
if a.done != nil {
<-a.done
}
if a.notifySvc != nil {
a.notifySvc.Close()
}
if a.notifyRepo != nil {
a.notifyRepo.Close()
}
if a.irohSidecar != nil {
a.irohSidecar.shutdown()
}
if a.conn != nil {
a.conn.Close()
}
if a.mesh != nil {
a.mesh.Stop()
}
if a.repo != nil {
return a.repo.Close()
}
return nil
}
// NotifyService returns the notify messaging service, or nil if notify is not enabled.
func (a *Agent) NotifyService() *notify.Service { return a.notifySvc }
// NotifyRepo returns the notify repository, or nil if notify is not enabled.
func (a *Agent) NotifyRepo() *libfossil.Repo { return a.notifyRepo }
// IrohEndpointID returns the local iroh endpoint ID, or "" if iroh is not enabled
// or the sidecar hasn't started yet.
func (a *Agent) IrohEndpointID() string { return a.irohEndpointID }
// IrohSocketPath returns the Unix socket path for the iroh sidecar, or "" if iroh is not enabled.
func (a *Agent) IrohSocketPath() string { return a.irohSock }
// MeshClientURL returns the NATS client URL the agent's embedded mesh
// server accepts client connections on. Empty when the agent has not
// run Start, or was constructed via NewFromParts (no mesh). Callers
// publish to this URL when they want to share the agent's NATS bus
// (e.g. for orchestration messages distinct from sync traffic).
func (a *Agent) MeshClientURL() string {
if a.mesh == nil {
return ""
}
return a.mesh.ClientURL()
}
// MeshLeafAddr returns the leaf-node listen address (host:port) where
// other agents' meshes can solicit upstream. Empty for NATSRoleLeaf
// agents (which never accept) and when the agent was constructed via
// NewFromParts. Callers pass this address as NATSUpstream when wiring
// remote agents into this agent's mesh.
func (a *Agent) MeshLeafAddr() string {
if a.mesh == nil {
return ""
}
return a.mesh.LeafAddr()
}
// SyncNow triggers an immediate sync round. It is non-blocking: if a sync
// trigger is already pending, the call is a no-op.
func (a *Agent) SyncNow() {
select {
case a.syncNow <- struct{}{}:
default:
}
}
// pollLoop runs until the context is cancelled, converting channel events
// into Tick calls and logging results.
func (a *Agent) pollLoop(ctx context.Context) {
defer close(a.done)
a.logf("poll loop started, interval %s", a.config.PollInterval)
for {
var ev Event
select {
case <-ctx.Done():
a.logf("poll loop stopped")
return
case <-a.clock.After(a.config.PollInterval):
ev = EventTimer
case <-a.syncNow:
ev = EventSyncNow
a.logf("manual sync triggered")
}
// BUGGIFY: skip a timer-triggered sync to test stale-state behavior.
if ev == EventTimer && a.config.Buggify != nil && a.config.Buggify.Check("agent.runSync.earlyReturn", 0.05) {
continue
}
// Run all sync targets, collect outcomes, then emit logs at
// the appropriate level. Per-target failures get demoted from
// ERROR to DEBUG when at least one other target succeeded —
// sync as a whole succeeded (e.g. round-0 NATS races where
// the HTTP target carries the real result), so isolated
// transport noise is not a user-actionable error.
outcomes := make([]syncOutcome, 0, len(a.syncTargets))
for _, target := range a.syncTargets {
result, err := a.repo.Sync(ctx, target.transport, a.buildSyncOpts())
outcomes = append(outcomes, syncOutcome{target: target, result: result, err: err})
}
a.emitSyncOutcomes(ctx, outcomes)
// Update peer registry once per poll cycle.
if err := a.updatePeerRegistryAfterSync(); err != nil {
a.logf("warning: updatePeerRegistryAfterSync: %v", err)
}
}
}
// buildSyncOpts constructs SyncOpts from the agent's config.
func (a *Agent) buildSyncOpts() libfossil.SyncOpts {
return libfossil.SyncOpts{
Push: a.config.Push,
Pull: a.config.Pull,
ProjectCode: a.projectCode,
ServerCode: a.serverCode,
User: a.config.User,
Password: a.config.Password,
PeerID: a.config.PeerID,
Buggify: a.config.Buggify,
UV: a.config.UV,
XTableSync: a.config.ServeNATSEnabled,
Private: a.config.Private,
Observer: a.config.Observer,
}
}
// runSync performs one sync cycle against the transport.
func (a *Agent) runSync(ctx context.Context) (*libfossil.SyncResult, error) {
return a.repo.Sync(ctx, a.transport, a.buildSyncOpts())
}
// monitorSidecar watches the sidecar process and logs if it exits unexpectedly.
func (a *Agent) monitorSidecar(ctx context.Context) {
if a.irohSidecar == nil {
return
}
select {
case <-a.irohSidecar.exited:
// Check if this was an expected shutdown.
select {
case <-ctx.Done():
return
default:
}
slog.Error("iroh sidecar exited unexpectedly", "error", a.irohSidecar.exitErr)
a.logf("iroh sidecar died: %v — iroh sync will fail until restart", a.irohSidecar.exitErr)
case <-ctx.Done():
return
}
}
// findIrohBinary looks for the iroh-sidecar binary. It checks, in order:
// 1. Config.IrohBinaryPath (explicit override)
// 2. Next to the leaf binary
// 3. In PATH
func (a *Agent) findIrohBinary() (string, error) {
if a.config.IrohBinaryPath != "" {
if _, err := os.Stat(a.config.IrohBinaryPath); err != nil {
return "", fmt.Errorf("iroh-sidecar binary not found at configured path %q: %w", a.config.IrohBinaryPath, err)
}
return a.config.IrohBinaryPath, nil
}
exe, err := os.Executable()
if err == nil {
candidate := filepath.Join(filepath.Dir(exe), "iroh-sidecar")
if _, err := os.Stat(candidate); err == nil {
return candidate, nil
}
}
path, err := exec.LookPath("iroh-sidecar")
if err != nil {
return "", fmt.Errorf("iroh-sidecar binary not found (not next to leaf binary, not in PATH)")
}
return path, nil
}