refactor: service layer, hot reload and domain cleanup

README.md

@@ -61,6 +61,16 @@ A lightweight HTTP service for monitoring periodic "heartbeat" pings ("bumps") a
 | `/bump/{id}/fail` | `POST`, `GET` | Manually mark heartbeat as failed |
 | `/healthz`        | `GET`         | Liveness probe                    |
 | `/metrics`        | `GET`         | Prometheus metrics endpoint       |
+| `/-/reload`       | `POST`        | Reload configuration              |
+
+## 🔄 Reloading Configuration
+
+Heartbeats supports hot-reload of the YAML config without restarting the process:
+
+- **Signal:** send `SIGHUP` to the process
+- **HTTP:** `POST /-/reload`
+
+Reloading re-reads the config file, validates it, updates receiver configuration, and reconciles heartbeat actors.
 
 ## ⚙️ Configuration
 

internal/app/run.go

@@ -12,11 +12,14 @@ import (
 	"github.com/containeroo/heartbeats/internal/config"
 	"github.com/containeroo/heartbeats/internal/debugserver"
 	"github.com/containeroo/heartbeats/internal/flag"
+	"github.com/containeroo/heartbeats/internal/handler"
 	"github.com/containeroo/heartbeats/internal/heartbeat"
 	"github.com/containeroo/heartbeats/internal/history"
 	"github.com/containeroo/heartbeats/internal/logging"
+	"github.com/containeroo/heartbeats/internal/metrics"
 	"github.com/containeroo/heartbeats/internal/notifier"
 	"github.com/containeroo/heartbeats/internal/server"
+	servicehistory "github.com/containeroo/heartbeats/internal/service/history"
 
 	"github.com/containeroo/tinyflags"
 )
@@ -26,8 +29,10 @@ func Run(ctx context.Context, webFS fs.FS, version, commit string, args []string
 	// Create a context to listen for shutdown signals
 	ctx, cancel := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM)
 	defer cancel()
+	// Create another context to listen for reload signals
+	reloadCh := make(chan os.Signal, 1)
+	signal.Notify(reloadCh, syscall.SIGHUP)
 
-	// Parse and validate command-line flags.
 	flags, err := flag.ParseFlags(args, version)
 	if err != nil {
 		if tinyflags.IsHelpRequested(err) || tinyflags.IsVersionRequested(err) {
@@ -37,7 +42,6 @@ func Run(ctx context.Context, webFS fs.FS, version, commit string, args []string
 		return fmt.Errorf("CLI flags error: %w", err)
 	}
 
-	// Load and validate configuration.
 	cfg, err := config.LoadConfig(flags.ConfigPath)
 	if err != nil {
 		return fmt.Errorf("failed to load config: %w", err)
@@ -46,59 +50,80 @@ func Run(ctx context.Context, webFS fs.FS, version, commit string, args []string
 		return fmt.Errorf("invalid YAML config: %w", err)
 	}
 
-	// Setup logger
 	logger := logging.SetupLogger(flags.LogFormat, flags.Debug, w)
 	logger.Info("Starting Heartbeats", "version", version, "commit", commit)
 
-	// Log CLI overrides
 	if len(flags.OverriddenValues) > 0 {
 		logger.Info("CLI Overrides", "overrides", flags.OverriddenValues)
 	}
 
-	// Create history cache
-	history, err := history.InitializeHistory(flags)
+	histStore, err := history.InitializeHistory(flags.HistorySize)
 	if err != nil {
 		return fmt.Errorf("failed to initialize history: %w", err)
 	}
+	histRecorder := servicehistory.NewRecorder(histStore)
+	metricsReg := metrics.New(histStore)
 
-	// Inizalize notification
 	store := notifier.InitializeStore(cfg.Receivers, flags.SkipTLS, version, logger)
 	bufferSize := len(cfg.Heartbeats) // 1 slot per actor; each heartbeat sends max 1 notification concurrently
 	dispatcher := notifier.NewDispatcher(
 		store,
 		logger,
-		history,
+		histRecorder,
 		flags.RetryCount,
 		flags.RetryDelay,
 		bufferSize,
+		metricsReg,
 	)
 	go dispatcher.Run(ctx)
 
 	// Create Heartbeat Manager
-	mgr := heartbeat.NewManagerFromHeartbeatMap(
+	actorFactory := heartbeat.DefaultActorFactory{
+		Logger:     logger,
+		History:    histRecorder,
+		Metrics:    metricsReg,
+		DispatchCh: dispatcher.Mailbox(),
+	}
+	mgr, err := heartbeat.NewManagerFromHeartbeatMap(
 		ctx,
 		cfg.Heartbeats,
-		dispatcher.Mailbox(),
-		history,
 		logger,
+		actorFactory,
+	)
+	if err != nil {
+		return fmt.Errorf("failed to initialize heartbeats: %w", err)
+	}
+	mgr.StartAll()
+
+	reloadConfig := config.NewReloadFunc(flags.ConfigPath, flags.SkipTLS, version, logger, dispatcher, mgr)
+	go config.WatchReload(ctx, reloadCh, logger, reloadConfig)
+
+	api := handler.NewAPI(
+		version,
+		commit,
+		webFS,
+		flags.SiteRoot,
+		flags.RoutePrefix,
+		flags.Debug,
+		logger,
+		mgr,
+		histStore,
+		histRecorder,
+		dispatcher,
+		metricsReg,
+		reloadConfig,
 	)
 
 	// Run debug server if enabled
 	if flags.Debug {
-		debugserver.Run(ctx, flags.DebugServerPort, mgr, dispatcher, logger)
+		debugserver.Run(ctx, flags.DebugServerPort, api)
 	}
 
 	// Create server and run forever
 	router := server.NewRouter(
 		webFS,
-		flags.SiteRoot,
-		flags.RoutePrefix,
-		version,
-		mgr,
-		history,
-		dispatcher,
+		api,
 		logger,
-		flags.Debug,
 	)
 	if err := server.Run(ctx, flags.ListenAddr, router, logger); err != nil {
 		return fmt.Errorf("failed to run Heartbeats: %w", err)

internal/config/config.go

@@ -1,8 +1,11 @@
 package config
 
 import (
+	"context"
 	"fmt"
+	"log/slog"
 	"os"
+	"sync"
 
 	"github.com/containeroo/heartbeats/internal/heartbeat"
 	"github.com/containeroo/heartbeats/internal/notifier"
@@ -82,3 +85,76 @@ func (c *Config) Validate() error {
 	}
 	return nil
 }
+
+// Reload loads, validates, and applies the config to the dispatcher and manager.
+func Reload(
+	filePath string,
+	skipTLS bool,
+	version string,
+	logger *slog.Logger,
+	dispatcher *notifier.Dispatcher,
+	mgr *heartbeat.Manager,
+) (heartbeat.ReconcileResult, error) {
+	cfg, err := LoadConfig(filePath)
+	if err != nil {
+		return heartbeat.ReconcileResult{}, fmt.Errorf("failed to load config: %w", err)
+	}
+	if err := cfg.Validate(); err != nil {
+		return heartbeat.ReconcileResult{}, fmt.Errorf("invalid YAML config: %w", err)
+	}
+
+	newStore := notifier.InitializeStore(cfg.Receivers, skipTLS, version, logger)
+	dispatcher.UpdateStore(newStore)
+
+	res, err := mgr.Reconcile(cfg.Heartbeats)
+	if err != nil {
+		return res, err
+	}
+	return res, nil
+}
+
+// NewReloadFunc builds a reload function with logging and serialization.
+func NewReloadFunc(
+	filePath string,
+	skipTLS bool,
+	version string,
+	logger *slog.Logger,
+	dispatcher *notifier.Dispatcher,
+	mgr *heartbeat.Manager,
+) func() error {
+	var reloadMu sync.Mutex
+	return func() error {
+		reloadMu.Lock()
+		defer reloadMu.Unlock()
+
+		res, err := Reload(filePath, skipTLS, version, logger, dispatcher, mgr)
+		if err != nil {
+			return err
+		}
+		if res.Added+res.Updated+res.Removed > 0 {
+			logger.Info("heartbeats reloaded",
+				"added", res.Added,
+				"updated", res.Updated,
+				"removed", res.Removed,
+			)
+		}
+		return nil
+	}
+}
+
+// WatchReload listens for reload signals and invokes the reload function.
+func WatchReload(ctx context.Context, reloadCh <-chan os.Signal, logger *slog.Logger, reloadFn func() error) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-reloadCh:
+			logger.Info("reload requested")
+			if err := reloadFn(); err != nil {
+				logger.Error("reload failed", "err", err)
+			} else {
+				logger.Info("reload completed")
+			}
+		}
+	}
+}

internal/config/reload_test.go

@@ -0,0 +1,39 @@
+package config
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestWatchReload(t *testing.T) {
+	t.Parallel()
+
+	var buf strings.Builder
+	logger := slog.New(slog.NewTextHandler(&buf, nil))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	ch := make(chan os.Signal, 1)
+	done := make(chan struct{})
+
+	go WatchReload(ctx, ch, logger, func() error {
+		close(done)
+		return nil
+	})
+
+	ch <- os.Interrupt
+
+	select {
+	case <-done:
+		// ok
+	case <-time.After(time.Second):
+		require.Fail(t, "timeout waiting for reload")
+	}
+}

internal/debugserver/debugserver.go

@@ -3,23 +3,20 @@ package debugserver
 import (
 	"context"
 	"fmt"
-	"log/slog"
 	"net/http"
 
-	"github.com/containeroo/heartbeats/internal/handlers"
-	"github.com/containeroo/heartbeats/internal/heartbeat"
-	"github.com/containeroo/heartbeats/internal/notifier"
+	"github.com/containeroo/heartbeats/internal/handler"
 )
 
 // Run starts the local-only debug server for manual testing.
-func Run(ctx context.Context, port int, mgr *heartbeat.Manager, dispatcher *notifier.Dispatcher, logger *slog.Logger) {
+func Run(ctx context.Context, port int, api *handler.API) {
 	mux := http.NewServeMux()
 
-	mux.Handle("GET /internal/receiver/{id}", handlers.TestReceiverHandler(dispatcher, logger))
-	mux.Handle("GET /internal/heartbeat/{id}", handlers.TestHeartbeatHandler(mgr, logger))
+	mux.Handle("GET /internal/receiver/{id}", api.TestReceiverHandler())
+	mux.Handle("GET /internal/heartbeat/{id}", api.TestHeartbeatHandler())
 
 	addr := fmt.Sprintf("127.0.0.1:%d", port)
-	logger.Info("starting debug server", "listenAddr", addr)
+	api.Logger.Info("starting debug server", "listenAddr", addr)
 
 	server := &http.Server{
 		Addr:    addr,
@@ -35,7 +32,7 @@ func Run(ctx context.Context, port int, mgr *heartbeat.Manager, dispatcher *noti
 	// Serve requests on 127.0.0.1 until shutdown.
 	go func() {
 		if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
-			logger.Error("debug server error", "error", err)
+			api.Logger.Error("debug server error", "error", err)
 		}
 	}()
 }

internal/debugserver/debugserver_test.go

@@ -8,9 +8,12 @@ import (
 	"testing"
 	"time"
 
+	"github.com/containeroo/heartbeats/internal/handler"
 	"github.com/containeroo/heartbeats/internal/heartbeat"
 	"github.com/containeroo/heartbeats/internal/history"
+	"github.com/containeroo/heartbeats/internal/metrics"
 	"github.com/containeroo/heartbeats/internal/notifier"
+	servicehistory "github.com/containeroo/heartbeats/internal/service/history"
 	"github.com/stretchr/testify/require"
 )
 
@@ -22,8 +25,10 @@ func TestDebugServer_Run(t *testing.T) {
 	// Setup basic in-memory state
 	logger := slog.New(slog.NewTextHandler(&strings.Builder{}, nil))
 	hist := history.NewRingStore(5)
+	metricsReg := metrics.New(hist)
+	recorder := servicehistory.NewRecorder(hist)
 	store := notifier.NewReceiverStore()
-	disp := notifier.NewDispatcher(store, logger, hist, 1, 1*time.Millisecond, 10)
+	disp := notifier.NewDispatcher(store, logger, recorder, 1, 1*time.Millisecond, 10, metricsReg)
 
 	hbCfg := map[string]heartbeat.HeartbeatConfig{
 		"test-hb": {
@@ -33,11 +38,38 @@ func TestDebugServer_Run(t *testing.T) {
 			Receivers:   []string{"r1"},
 		},
 	}
-	mgr := heartbeat.NewManagerFromHeartbeatMap(ctx, hbCfg, disp.Mailbox(), hist, logger)
+	factory := heartbeat.DefaultActorFactory{
+		Logger:     logger,
+		History:    recorder,
+		Metrics:    metricsReg,
+		DispatchCh: disp.Mailbox(),
+	}
+	mgr, err := heartbeat.NewManagerFromHeartbeatMap(
+		ctx,
+		hbCfg,
+		logger,
+		factory,
+	)
+	require.NoError(t, err)
+	api := handler.NewAPI(
+		"test",
+		"test",
+		nil,
+		"",
+		"",
+		true,
+		logger,
+		mgr,
+		hist,
+		recorder,
+		disp,
+		nil,
+		nil,
+	)
 
 	// Pick a random free port
 	port := 8089
-	Run(ctx, port, mgr, disp, logger)
+	Run(ctx, port, api)
 
 	// Wait for server to start
 	time.Sleep(100 * time.Millisecond)