knotserver: introduce notifier package

Notifier is a semaphore that can be used to indicate a change in a
resource.

Signed-off-by: oppiliappan <[e-mail]>

knotserver/db/oplog.go

···
 
 import (
 	"fmt"
+
+	"tangled.sh/tangled.sh/core/knotserver/notifier"
 )
 
 type Op struct {
···
 	Ref    string // the reference being updated
 }
 
-func (d *DB) InsertOp(op Op) error {
+func (d *DB) InsertOp(op Op, notifier *notifier.Notifier) error {
 	_, err := d.db.Exec(
 		`insert into oplog (tid, did, repo, old_sha, new_sha, ref) values (?, ?, ?, ?, ?, ?)`,
 		op.Tid,
···
 		op.NewSha,
 		op.Ref,
 	)
-	return err
+	if err != nil {
+		return err
+	}
+
+	notifier.NotifyAll()
+	return nil
 }
 
 func (d *DB) GetOps(cursor string) ([]Op, error) {

knotserver/handler.go

···
 	"tangled.sh/tangled.sh/core/jetstream"
 	"tangled.sh/tangled.sh/core/knotserver/config"
 	"tangled.sh/tangled.sh/core/knotserver/db"
+	"tangled.sh/tangled.sh/core/knotserver/notifier"
 	"tangled.sh/tangled.sh/core/rbac"
 )
 
···
 	jc *jetstream.JetstreamClient
 	e  *rbac.Enforcer
 	l  *slog.Logger
+	n  *notifier.Notifier
 
 	// init is a channel that is closed when the knot has been initailized
 	// i.e. when the first user (knot owner) has been added.
···
 	knotInitialized bool
 }
 
-func Setup(ctx context.Context, c *config.Config, db *db.DB, e *rbac.Enforcer, jc *jetstream.JetstreamClient, l *slog.Logger) (http.Handler, error) {
+func Setup(ctx context.Context, c *config.Config, db *db.DB, e *rbac.Enforcer, jc *jetstream.JetstreamClient, l *slog.Logger, n *notifier.Notifier) (http.Handler, error) {
 	r := chi.NewRouter()
 
 	h := Handle{
···
 		e:    e,
 		l:    l,
 		jc:   jc,
+		n:    n,
 		init: make(chan struct{}),
 	}
 

knotserver/internal.go

···
 	"github.com/go-chi/chi/v5/middleware"
 	"tangled.sh/tangled.sh/core/knotserver/config"
 	"tangled.sh/tangled.sh/core/knotserver/db"
+	"tangled.sh/tangled.sh/core/knotserver/notifier"
 	"tangled.sh/tangled.sh/core/rbac"
 )
 
···
 	c  *config.Config
 	e  *rbac.Enforcer
 	l  *slog.Logger
+	n  *notifier.Notifier
 }
 
 func (h *InternalHandle) PushAllowed(w http.ResponseWriter, r *http.Request) {
···
 	}
 
 	for _, op := range ops {
-		err := h.db.InsertOp(op)
+		err := h.db.InsertOp(op, h.n)
 		if err != nil {
 			l.Error("failed to insert op", "err", err, "op", op)
 			continue
···
 	return
 }
 
-func Internal(ctx context.Context, c *config.Config, db *db.DB, e *rbac.Enforcer, l *slog.Logger) http.Handler {
+func Internal(ctx context.Context, c *config.Config, db *db.DB, e *rbac.Enforcer, l *slog.Logger, n *notifier.Notifier) http.Handler {
 	r := chi.NewRouter()
 
 	h := InternalHandle{
···
 		c,
 		e,
 		l,
+		n,
 	}
 
 	r.Get("/push-allowed", h.PushAllowed)

knotserver/notifier/notifier.go

···
+package notifier
+
+import (
+	"sync"
+)
+
+type Notifier struct {
+	subscribers map[chan struct{}]struct{}
+	mu          sync.Mutex
+}
+
+func New() Notifier {
+	return Notifier{
+		subscribers: make(map[chan struct{}]struct{}),
+	}
+}
+
+func (n *Notifier) Subscribe() chan struct{} {
+	ch := make(chan struct{}, 1)
+	n.mu.Lock()
+	n.subscribers[ch] = struct{}{}
+	n.mu.Unlock()
+	return ch
+}
+
+func (n *Notifier) Unsubscribe(ch chan struct{}) {
+	n.mu.Lock()
+	delete(n.subscribers, ch)
+	close(ch)
+	n.mu.Unlock()
+}
+
+func (n *Notifier) NotifyAll() {
+	n.mu.Lock()
+	for ch := range n.subscribers {
+		select {
+		case ch <- struct{}{}:
+		default:
+			// avoid blocking if channel is full
+		}
+	}
+	n.mu.Unlock()
+}

knotserver/server.go

···
 	"tangled.sh/tangled.sh/core/jetstream"
 	"tangled.sh/tangled.sh/core/knotserver/config"
 	"tangled.sh/tangled.sh/core/knotserver/db"
+	"tangled.sh/tangled.sh/core/knotserver/notifier"
 	"tangled.sh/tangled.sh/core/log"
 	"tangled.sh/tangled.sh/core/rbac"
 )
···
 		logger.Error("failed to setup jetstream", "error", err)
 	}
 
-	mux, err := Setup(ctx, c, db, e, jc, logger)
+	notifier := notifier.New()
+
+	mux, err := Setup(ctx, c, db, e, jc, logger, &notifier)
 	if err != nil {
 		return fmt.Errorf("failed to setup server: %w", err)
 	}
-	imux := Internal(ctx, c, db, e, iLogger)
+
+	imux := Internal(ctx, c, db, e, iLogger, &notifier)
 
 	logger.Info("starting internal server", "address", c.Server.InternalListenAddr)
 	go http.ListenAndServe(c.Server.InternalListenAddr, imux)