Changeset 0.23.1 (#154)

server/events/bus.go

@@ -10,9 +10,10 @@ import (
 // Handlers subscribe to patterns (exact or trailing wildcard).
 // Publishing fans out to all matching subscribers.
 type Bus struct {
-	mu   sync.RWMutex
-	subs map[string][]*subscription
-	seq  uint64 // subscription ID counter
+	mu            sync.RWMutex
+	subs          map[string][]*subscription
+	seq           uint64 // subscription ID counter
+	broadcastHook func(Event) // called after Publish for cross-pod fan-out; nil-safe
 }
 
 type subscription struct {
@@ -28,6 +29,15 @@ func NewBus() *Bus {
 	}
 }
 
+// SetBroadcastHook registers a function called after every Publish.
+// Used by the Postgres LISTEN/NOTIFY adapter to fan out across pods.
+// The hook is NOT called by publishLocal (avoids re-broadcast loops).
+func (b *Bus) SetBroadcastHook(fn func(Event)) {
+	b.mu.Lock()
+	b.broadcastHook = fn
+	b.mu.Unlock()
+}
+
 // Subscribe registers a handler for a label pattern.
 // Returns an unsubscribe function.
 //
@@ -60,7 +70,22 @@ func (b *Bus) Subscribe(pattern string, handler Handler) func() {
 // Publish dispatches an event to all matching subscribers.
 // Handlers are called synchronously in subscription order.
 // Use PublishAsync for non-blocking dispatch.
+// After local dispatch, calls the broadcastHook if set (Postgres fan-out).
 func (b *Bus) Publish(event Event) {
+	b.publishLocal(event)
+
+	b.mu.RLock()
+	hook := b.broadcastHook
+	b.mu.RUnlock()
+	if hook != nil {
+		hook(event)
+	}
+}
+
+// publishLocal dispatches to local subscribers only, without triggering
+// the broadcastHook. Used by the Postgres listener to re-publish remote
+// events without causing an infinite re-broadcast loop.
+func (b *Bus) publishLocal(event Event) {
 	b.mu.RLock()
 	var matched []Handler
 	for pattern, subs := range b.subs {
@@ -79,6 +104,7 @@ func (b *Bus) Publish(event Event) {
 
 // PublishAsync dispatches an event to all matching subscribers
 // in separate goroutines. Useful for I/O-heavy handlers.
+// Also calls the broadcastHook asynchronously if set.
 func (b *Bus) PublishAsync(event Event) {
 	b.mu.RLock()
 	var matched []Handler
@@ -89,11 +115,15 @@ func (b *Bus) PublishAsync(event Event) {
 			}
 		}
 	}
+	hook := b.broadcastHook
 	b.mu.RUnlock()
 
 	for _, h := range matched {
 		go h(event)
 	}
+	if hook != nil {
+		go hook(event)
+	}
 }
 
 // match checks if a concrete label matches a subscription pattern.

server/events/pg_broadcast.go

@@ -0,0 +1,130 @@
+package events
+
+// pg_broadcast.go — Postgres LISTEN/NOTIFY adapter for multi-pod fan-out.
+//
+// Problem: the in-process Bus is per-pod. With N backend replicas, an event
+// published on pod-0 (e.g. a chat message completion) is never seen by clients
+// connected to pod-1.
+//
+// Solution: for every event that should reach clients (DirToClient | DirBoth),
+// publish via pg_notify so every pod's listener goroutine receives it and
+// re-publishes into its local Bus → Hub → WebSocket clients.
+//
+// Key invariant: re-published (remote) events use publishLocal, which skips the
+// broadcastHook, preventing infinite re-broadcast loops.
+//
+// No-op when running SQLite (single process, in-process Bus is sufficient).
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/lib/pq"
+
+	"git.gobha.me/xcaliber/chat-switchboard/database"
+)
+
+const pgNotifyChannel = "switchboard_events"
+
+// maxNotifyBytes is the practical PG NOTIFY payload limit (8192 byte hard limit,
+// leave headroom for encoding overhead).
+const maxNotifyBytes = 7900
+
+// StartPGBroadcast wires the Bus to Postgres LISTEN/NOTIFY.
+// Call once after bus and database are both initialized.
+// No-op for SQLite deployments.
+func StartPGBroadcast(bus *Bus) {
+	if !database.IsPostgres() {
+		log.Println("[pg_broadcast] SQLite mode — skipping cross-pod broadcast")
+		return
+	}
+
+	// Start the listener goroutine (auto-reconnects on failure).
+	go broadcastListenLoop(bus)
+
+	// Hook into the Bus: after every local Publish, fan out to other pods.
+	bus.SetBroadcastHook(func(e Event) {
+		// Only events that clients care about need to cross pod boundaries.
+		dir := RouteFor(e.Label)
+		if dir == DirLocal {
+			return
+		}
+
+		// tool.result.* arrives from a client WS on a specific pod — that pod
+		// handles WaitFor() locally. No need (or benefit) to broadcast.
+		if dir == DirFromClient {
+			return
+		}
+
+		data, err := json.Marshal(e)
+		if err != nil {
+			return
+		}
+		if len(data) > maxNotifyBytes {
+			// Event payload too large for NOTIFY — this should never happen
+			// in practice (we notify IDs/metadata, not message bodies).
+			log.Printf("[pg_broadcast] event %q payload %d bytes exceeds limit — skipping broadcast", e.Label, len(data))
+			return
+		}
+
+		if _, err := database.DB.Exec("SELECT pg_notify($1, $2)", pgNotifyChannel, string(data)); err != nil {
+			log.Printf("[pg_broadcast] pg_notify failed: %v", err)
+		}
+	})
+
+	log.Println("[pg_broadcast] cross-pod broadcast enabled via Postgres LISTEN/NOTIFY")
+}
+
+// broadcastListenLoop runs forever, reconnecting on error.
+func broadcastListenLoop(bus *Bus) {
+	dsn := database.DSN()
+	for {
+		if err := runBroadcastListener(bus, dsn); err != nil {
+			log.Printf("[pg_broadcast] listener error: %v — reconnecting in 5s", err)
+		}
+		time.Sleep(5 * time.Second)
+	}
+}
+
+// runBroadcastListener opens a dedicated pq listener connection and feeds
+// received notifications into the local Bus via publishLocal.
+// Returns an error when the connection is lost (caller retries).
+func runBroadcastListener(bus *Bus, dsn string) error {
+	reportErr := func(ev pq.ListenerEventType, err error) {
+		if err != nil {
+			log.Printf("[pg_broadcast] pq listener event=%d err=%v", ev, err)
+		}
+	}
+
+	l := pq.NewListener(dsn, 5*time.Second, time.Minute, reportErr)
+	if err := l.Listen(pgNotifyChannel); err != nil {
+		return fmt.Errorf("LISTEN %s: %w", pgNotifyChannel, err)
+	}
+	defer l.Close()
+
+	log.Printf("[pg_broadcast] listener ready on channel %q", pgNotifyChannel)
+
+	for {
+		select {
+		case n, ok := <-l.Notify:
+			if !ok {
+				return fmt.Errorf("notify channel closed")
+			}
+			if n == nil {
+				// Keepalive ping from pq — ignore.
+				continue
+			}
+
+			var e Event
+			if err := json.Unmarshal([]byte(n.Extra), &e); err != nil {
+				log.Printf("[pg_broadcast] malformed notification: %v", err)
+				continue
+			}
+
+			// publishLocal bypasses the broadcastHook — no re-broadcast loop.
+			bus.publishLocal(e)
+		}
+	}
+}