feature (journal): improve happy path speed

server/pkg/journal/index.go

@@ -2,12 +2,14 @@ package journal
 
 import (
 	"sync"
+	"sync/atomic"
 	"time"
 )
 
 var (
-	journal = NewRing[IObservation](5000, Observation[Nop]{ At: time.Now().UTC() })
-	gate    = sync.NewCond(&sync.Mutex{})
+	journal     = NewRing[IObservation](5000, Observation[Nop]{At: time.Now().UTC()})
+	gate        = sync.NewCond(&sync.Mutex{})
+	subscribers = atomic.Int64{}
 )
 
 type Payload interface {

server/pkg/journal/observation.go

@@ -15,6 +15,7 @@ type Observation[T Payload] struct {
 	Trace   tracer.TraceContext
 	Done    bool
 	Error   error
+	Session map[string]string
 	Emit    func(Observation[T])
 }
 
@@ -23,11 +24,17 @@ type IObservation interface {
 }
 
 func NewObservation[T Payload](ob Observation[T]) Observation[T] {
+	if ob.Emit == nil {
+		return ob
+	}
 	ob.Emit(ob)
 	return ob
 }
 
 func (this Observation[T]) Close(res http.ResponseWriter) {
+	if this.Emit == nil {
+		return
+	}
 	this.At = time.Now().UTC()
 	this.Done = true
 	if obj, ok := res.(interface{ Status() int }); ok {

server/pkg/journal/publish.go

@@ -13,6 +13,9 @@ import (
 )
 
 func PublishFS(ctx *App, req *http.Request, topic string, path string, target ...string) Observation[FileOp] {
+	if subscribers.Load() == 0 {
+		return NewObservation(Observation[FileOp]{})
+	}
 	return NewObservation(Observation[FileOp]{
 		At:   time.Now().UTC(),
 		Kind: "fs",
@@ -23,30 +26,39 @@ func PublishFS(ctx *App, req *http.Request, topic string, path string, target ..
 			Target:    strings.Join(target, ","),
 			StorageID: GenerateID(ctx.Session),
 		},
-		Trace: tracer.Extract(req),
-		Emit:  emit[FileOp],
+		Session: ctx.Session,
+		Trace:   tracer.Extract(req),
+		Emit:    emit[FileOp],
 	})
 }
 
 func PublishSession(ctx *App, req *http.Request, cmd string) Observation[SessionOp] {
+	if subscribers.Load() == 0 {
+		return NewObservation(Observation[SessionOp]{})
+	}
 	return NewObservation(Observation[SessionOp]{
 		At:   time.Now().UTC(),
 		Kind: "session",
 		Payload: SessionOp{
 			Operation: cmd,
 		},
-		Trace: tracer.Extract(req),
-		Emit:  emit[SessionOp],
+		Session: ctx.Session,
+		Trace:   tracer.Extract(req),
+		Emit:    emit[SessionOp],
 	})
 }
 
 func PublishEvent(ctx *App, req *http.Request, data EventOp) Observation[EventOp] {
+	if subscribers.Load() == 0 {
+		return NewObservation(Observation[EventOp]{})
+	}
 	return NewObservation(Observation[EventOp]{
-		At:   time.Now().UTC(),
-		Kind: "event",
+		At:      time.Now().UTC(),
+		Kind:    "event",
 		Payload: data,
-		Trace: tracer.Extract(req),
-		Emit:  emit[EventOp],
+		Trace:   tracer.Extract(req),
+		Session: ctx.Session,
+		Emit:    emit[EventOp],
 	})
 }
 

server/pkg/journal/subscribe.go

@@ -8,7 +8,9 @@ import (
 
 func Subscribe[T Payload](ctx context.Context, since time.Time, filter func(Observation[T]) bool) <-chan []Observation[T] {
 	out := make(chan []Observation[T])
+	subscribers.Add(1)
 	go func() {
+		defer subscribers.Add(-1)
 		for firstRun, checkpoint := true, since; ; firstRun = false {
 			changes := make([]Observation[T], 0, 10)
 			gate.L.Lock()