feat: add leadership package

cmd/serve.go

@@ -3,6 +3,7 @@ package cmd
 import (
 	"github.com/formancehq/go-libs/v2/logging"
 	"github.com/formancehq/ledger/internal/api/common"
+	"github.com/formancehq/ledger/internal/leadership"
 	systemstore "github.com/formancehq/ledger/internal/storage/system"
 	"net/http"
 	"net/http/pprof"
@@ -108,6 +109,7 @@ func NewServeCommand() *cobra.Command {
 				}),
 				bus.NewFxModule(),
 				ballast.Module(serveConfiguration.ballastSize),
+				leadership.NewFXModule(),
 				api.Module(api.Config{
 					Version: Version,
 					Debug:   service.IsDebug(cmd),
@@ -122,15 +124,15 @@ func NewServeCommand() *cobra.Command {
 				}),
 				fx.Decorate(func(
 					params struct {
-					fx.In
+						fx.In
 
-					Handler          chi.Router
-					HealthController *health.HealthController
-					Logger           logging.Logger
+						Handler          chi.Router
+						HealthController *health.HealthController
+						Logger           logging.Logger
 
-					MeterProvider *metric.MeterProvider         `optional:"true"`
-					Exporter      *otlpmetrics.InMemoryExporter `optional:"true"`
-				},
+						MeterProvider *metric.MeterProvider         `optional:"true"`
+						Exporter      *otlpmetrics.InMemoryExporter `optional:"true"`
+					},
 				) chi.Router {
 					return assembleFinalRouter(
 						service.IsDebug(cmd),

internal/doc.go

@@ -1,2 +1,2 @@
-//go:generate gomarkdoc -o README.md --repository.default-branch main
+//go:generate gomarkdoc -o README.md --repository.default-branch main --repository.url https://github.com/formancehq/ledger
 package ledger

internal/leadership/broadcaster.go

@@ -0,0 +1,86 @@
+package leadership
+
+import (
+	"sync"
+)
+
+type listener struct {
+	channel chan Leadership
+}
+
+type Broadcaster struct {
+	mu *sync.Mutex
+	t  *Leadership
+
+	inner []listener
+	outer chan Leadership
+}
+
+func (h *Broadcaster) Actual() Leadership {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	if h.t == nil {
+		return Leadership{}
+	}
+	return *h.t
+}
+
+func (h *Broadcaster) Subscribe() (<-chan Leadership, func()) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	newChannel := make(chan Leadership, 1)
+	index := len(h.inner)
+	h.inner = append(h.inner, listener{
+		channel: newChannel,
+	})
+	if h.t != nil {
+		newChannel <- *h.t
+	}
+
+	return newChannel, func() {
+		h.mu.Lock()
+		defer h.mu.Unlock()
+
+		if index < len(h.inner)-1 {
+			h.inner = append(h.inner[:index], h.inner[index+1:]...)
+		} else {
+			h.inner = h.inner[:index]
+		}
+	}
+}
+
+func (h *Broadcaster) Broadcast(t Leadership) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	h.t = &t
+
+	for _, inner := range h.inner {
+		inner.channel <- t
+	}
+}
+
+func (h *Broadcaster) Close() {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	for _, inner := range h.inner {
+		close(inner.channel)
+	}
+}
+
+func (h *Broadcaster) CountListeners() int {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	return len(h.inner)
+}
+
+func NewSignal() *Broadcaster {
+	return &Broadcaster{
+		outer: make(chan Leadership),
+		mu:    &sync.Mutex{},
+	}
+}