verae-mcp-timestamping-server-go/internal/embednats/server.go

package embednats

import (
	"fmt"
	"net"
	"strconv"
	"strings"
	"time"

	"git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/internal/ledger"
	"git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/internal/veraetime"
	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
)

const (
	// DefaultListen is used when VERAE_NATS_LISTEN is unset.
	DefaultListen = "127.0.0.1:4223"
	queueGroup    = "verae-ts"
)

// Options configures the in-process NATS server and subject handlers.
type Options struct {
	// Listen is host:port for client connections. Empty disables embed.
	// Port 0 picks an ephemeral port (tests).
	Listen        string
	Ledger        *ledger.FileLedger
	Global        *veraetime.Client
	Keeper        *veraetime.Keeper
	RequireBearer bool
}

// Server is an embedded nats-server with timestamp subject handlers.
type Server struct {
	ns *server.Server
	nc *nats.Conn
}

// Start runs an in-process nats-server on opts.Listen and registers handlers.
// When Listen is empty, returns (nil, nil).
func Start(opts Options) (*Server, error) {
	listen := strings.TrimSpace(opts.Listen)
	if listen == "" {
		return nil, nil
	}
	host, port, err := splitHostPort(listen)
	if err != nil {
		return nil, err
	}

	ns, err := server.NewServer(&server.Options{
		Host:   host,
		Port:   port,
		NoLog:  true,
		NoSigs: true,
	})
	if err != nil {
		return nil, fmt.Errorf("embednats: NewServer: %w", err)
	}
	go ns.Start()
	if !ns.ReadyForConnections(5 * time.Second) {
		ns.Shutdown()
		return nil, fmt.Errorf("embednats: not ready for connections on %s", listen)
	}

	nc, err := nats.Connect(ns.ClientURL())
	if err != nil {
		ns.Shutdown()
		return nil, fmt.Errorf("embednats: connect: %w", err)
	}

	h := &handlers{
		nc:            nc,
		ledger:        opts.Ledger,
		global:        opts.Global,
		keeper:        opts.Keeper,
		requireBearer: opts.RequireBearer,
	}
	if err := h.subscribe(); err != nil {
		nc.Close()
		ns.Shutdown()
		return nil, err
	}

	return &Server{ns: ns, nc: nc}, nil
}

// ClientURL returns the nats:// URL clients should dial.
func (s *Server) ClientURL() string {
	if s == nil || s.ns == nil {
		return ""
	}
	return s.ns.ClientURL()
}

// Shutdown stops handlers and the embedded server.
func (s *Server) Shutdown() {
	if s == nil {
		return
	}
	if s.nc != nil {
		s.nc.Close()
	}
	if s.ns != nil {
		s.ns.Shutdown()
		s.ns.WaitForShutdown()
	}
}

func splitHostPort(listen string) (string, int, error) {
	host, portStr, err := net.SplitHostPort(listen)
	if err != nil {
		return "", 0, fmt.Errorf("embednats: listen %q: %w", listen, err)
	}
	port, err := strconv.Atoi(portStr)
	if err != nil {
		return "", 0, fmt.Errorf("embednats: listen port %q: %w", portStr, err)
	}
	if host == "" {
		host = "127.0.0.1"
	}
	// nats-server treats port 0 as its default (4222), which is already taken
	// on ns1. Bind an ephemeral port and hand that number to the server.
	if port == 0 {
		ln, err := net.Listen("tcp", net.JoinHostPort(host, "0"))
		if err != nil {
			return "", 0, fmt.Errorf("embednats: ephemeral listen: %w", err)
		}
		port = ln.Addr().(*net.TCPAddr).Port
		ln.Close()
	}
	return host, port, nil
}