Changeset 0.5.0 (#35)
This commit is contained in:
116
server/events/bus.go
Normal file
116
server/events/bus.go
Normal file
@@ -0,0 +1,116 @@
|
||||
package events
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// Bus is a labeled publish/subscribe event bus.
|
||||
// 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
|
||||
}
|
||||
|
||||
type subscription struct {
|
||||
id uint64
|
||||
pattern string
|
||||
handler Handler
|
||||
}
|
||||
|
||||
// NewBus creates a new event bus.
|
||||
func NewBus() *Bus {
|
||||
return &Bus{
|
||||
subs: make(map[string][]*subscription),
|
||||
}
|
||||
}
|
||||
|
||||
// Subscribe registers a handler for a label pattern.
|
||||
// Returns an unsubscribe function.
|
||||
//
|
||||
// Patterns:
|
||||
//
|
||||
// "chat.message.abc123" — exact match
|
||||
// "chat.message.*" — wildcard: matches chat.message.{anything}
|
||||
// "chat.*" — wildcard: matches chat.{anything}
|
||||
// "*" — matches all events
|
||||
func (b *Bus) Subscribe(pattern string, handler Handler) func() {
|
||||
b.mu.Lock()
|
||||
b.seq++
|
||||
sub := &subscription{id: b.seq, pattern: pattern, handler: handler}
|
||||
b.subs[pattern] = append(b.subs[pattern], sub)
|
||||
b.mu.Unlock()
|
||||
|
||||
return func() {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
subs := b.subs[pattern]
|
||||
for i, s := range subs {
|
||||
if s.id == sub.id {
|
||||
b.subs[pattern] = append(subs[:i], subs[i+1:]...)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Publish dispatches an event to all matching subscribers.
|
||||
// Handlers are called synchronously in subscription order.
|
||||
// Use PublishAsync for non-blocking dispatch.
|
||||
func (b *Bus) Publish(event Event) {
|
||||
b.mu.RLock()
|
||||
var matched []Handler
|
||||
for pattern, subs := range b.subs {
|
||||
if match(event.Label, pattern) {
|
||||
for _, s := range subs {
|
||||
matched = append(matched, s.handler)
|
||||
}
|
||||
}
|
||||
}
|
||||
b.mu.RUnlock()
|
||||
|
||||
for _, h := range matched {
|
||||
h(event)
|
||||
}
|
||||
}
|
||||
|
||||
// PublishAsync dispatches an event to all matching subscribers
|
||||
// in separate goroutines. Useful for I/O-heavy handlers.
|
||||
func (b *Bus) PublishAsync(event Event) {
|
||||
b.mu.RLock()
|
||||
var matched []Handler
|
||||
for pattern, subs := range b.subs {
|
||||
if match(event.Label, pattern) {
|
||||
for _, s := range subs {
|
||||
matched = append(matched, s.handler)
|
||||
}
|
||||
}
|
||||
}
|
||||
b.mu.RUnlock()
|
||||
|
||||
for _, h := range matched {
|
||||
go h(event)
|
||||
}
|
||||
}
|
||||
|
||||
// match checks if a concrete label matches a subscription pattern.
|
||||
//
|
||||
// "chat.message.abc" matches "chat.message.abc" (exact)
|
||||
// "chat.message.abc" matches "chat.message.*" (wildcard)
|
||||
// "chat.message.abc" matches "chat.*" (wildcard)
|
||||
// "chat.message.abc" matches "*" (global wildcard)
|
||||
func match(label, pattern string) bool {
|
||||
if pattern == "*" {
|
||||
return true
|
||||
}
|
||||
if pattern == label {
|
||||
return true
|
||||
}
|
||||
if !strings.HasSuffix(pattern, "*") {
|
||||
return false
|
||||
}
|
||||
prefix := pattern[:len(pattern)-1] // "chat.message." from "chat.message.*"
|
||||
return strings.HasPrefix(label, prefix)
|
||||
}
|
||||
123
server/events/bus_test.go
Normal file
123
server/events/bus_test.go
Normal file
@@ -0,0 +1,123 @@
|
||||
package events
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestMatch(t *testing.T) {
|
||||
tests := []struct {
|
||||
label, pattern string
|
||||
want bool
|
||||
}{
|
||||
{"chat.message.abc", "chat.message.abc", true},
|
||||
{"chat.message.abc", "chat.message.*", true},
|
||||
{"chat.message.abc", "chat.*", true},
|
||||
{"chat.message.abc", "*", true},
|
||||
{"chat.message.abc", "chat.message.xyz", false},
|
||||
{"chat.message.abc", "channel.message.*", false},
|
||||
{"chat.message.abc", "chat.message", false},
|
||||
{"ping", "ping", true},
|
||||
{"ping", "pong", false},
|
||||
{"plugin.hook.pre_completion", "plugin.hook.*", true},
|
||||
{"plugin.hook.pre_completion", "plugin.*", true},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
got := match(tt.label, tt.pattern)
|
||||
if got != tt.want {
|
||||
t.Errorf("match(%q, %q) = %v, want %v", tt.label, tt.pattern, got, tt.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBusPublishExact(t *testing.T) {
|
||||
bus := NewBus()
|
||||
var count int32
|
||||
|
||||
bus.Subscribe("chat.message.abc", func(e Event) {
|
||||
atomic.AddInt32(&count, 1)
|
||||
})
|
||||
|
||||
bus.Publish(Event{Label: "chat.message.abc", Payload: json.RawMessage(`{}`)})
|
||||
bus.Publish(Event{Label: "chat.message.xyz", Payload: json.RawMessage(`{}`)})
|
||||
|
||||
if atomic.LoadInt32(&count) != 1 {
|
||||
t.Errorf("expected 1 dispatch, got %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBusPublishWildcard(t *testing.T) {
|
||||
bus := NewBus()
|
||||
var count int32
|
||||
|
||||
bus.Subscribe("chat.message.*", func(e Event) {
|
||||
atomic.AddInt32(&count, 1)
|
||||
})
|
||||
|
||||
bus.Publish(Event{Label: "chat.message.abc", Payload: json.RawMessage(`{}`)})
|
||||
bus.Publish(Event{Label: "chat.message.xyz", Payload: json.RawMessage(`{}`)})
|
||||
bus.Publish(Event{Label: "chat.typing.abc", Payload: json.RawMessage(`{}`)})
|
||||
|
||||
if atomic.LoadInt32(&count) != 2 {
|
||||
t.Errorf("expected 2 dispatches, got %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBusUnsubscribe(t *testing.T) {
|
||||
bus := NewBus()
|
||||
var count int32
|
||||
|
||||
unsub := bus.Subscribe("test.event", func(e Event) {
|
||||
atomic.AddInt32(&count, 1)
|
||||
})
|
||||
|
||||
bus.Publish(Event{Label: "test.event", Payload: json.RawMessage(`{}`)})
|
||||
unsub()
|
||||
bus.Publish(Event{Label: "test.event", Payload: json.RawMessage(`{}`)})
|
||||
|
||||
if atomic.LoadInt32(&count) != 1 {
|
||||
t.Errorf("expected 1 dispatch after unsub, got %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBusGlobalWildcard(t *testing.T) {
|
||||
bus := NewBus()
|
||||
var count int32
|
||||
|
||||
bus.Subscribe("*", func(e Event) {
|
||||
atomic.AddInt32(&count, 1)
|
||||
})
|
||||
|
||||
bus.Publish(Event{Label: "chat.message.abc", Payload: json.RawMessage(`{}`)})
|
||||
bus.Publish(Event{Label: "system.notify", Payload: json.RawMessage(`{}`)})
|
||||
bus.Publish(Event{Label: "plugin.hook.pre_completion", Payload: json.RawMessage(`{}`)})
|
||||
|
||||
if atomic.LoadInt32(&count) != 3 {
|
||||
t.Errorf("expected 3 dispatches, got %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRouteFor(t *testing.T) {
|
||||
tests := []struct {
|
||||
label string
|
||||
want Direction
|
||||
}{
|
||||
{"chat.message.abc", DirBoth},
|
||||
{"chat.typing.abc", DirBoth},
|
||||
{"system.notify", DirToClient},
|
||||
{"plugin.hook.pre_completion", DirLocal},
|
||||
{"internal.db.write", DirLocal},
|
||||
{"unknown.event", DirLocal}, // default
|
||||
{"ping", DirFromClient},
|
||||
{"pong", DirToClient},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
got := RouteFor(tt.label)
|
||||
if got != tt.want {
|
||||
t.Errorf("RouteFor(%q) = %d, want %d", tt.label, got, tt.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
94
server/events/types.go
Normal file
94
server/events/types.go
Normal file
@@ -0,0 +1,94 @@
|
||||
package events
|
||||
|
||||
import "encoding/json"
|
||||
|
||||
// Event is the universal envelope for all bus communication.
|
||||
type Event struct {
|
||||
Label string `json:"event"`
|
||||
Room string `json:"room,omitempty"`
|
||||
Payload json.RawMessage `json:"payload"`
|
||||
Ts int64 `json:"ts"`
|
||||
|
||||
// Server-side metadata (not serialized to clients)
|
||||
SenderID string `json:"-"` // user ID of sender, empty for system events
|
||||
ConnID string `json:"-"` // WebSocket connection ID, empty for server-origin
|
||||
}
|
||||
|
||||
// Handler is a callback for bus subscriptions.
|
||||
type Handler func(Event)
|
||||
|
||||
// Direction controls which events cross the WebSocket bridge.
|
||||
type Direction int
|
||||
|
||||
const (
|
||||
DirLocal Direction = iota // bus-internal only (plugins, DB hooks)
|
||||
DirToClient // BE → FE
|
||||
DirFromClient // FE → BE
|
||||
DirBoth // bidirectional
|
||||
)
|
||||
|
||||
// routeTable defines the default routing for known event prefixes.
|
||||
// Events not listed default to DirLocal (server-only).
|
||||
var routeTable = map[string]Direction{
|
||||
// Chat events
|
||||
"chat.message.": DirBoth, // new messages
|
||||
"chat.typing.": DirBoth, // typing indicators
|
||||
"chat.updated.": DirToClient, // chat title/metadata changed
|
||||
"chat.deleted.": DirToClient, // chat removed
|
||||
|
||||
// Channel events
|
||||
"channel.message.": DirBoth,
|
||||
"channel.typing.": DirBoth,
|
||||
"channel.updated.": DirToClient,
|
||||
"channel.member.": DirToClient,
|
||||
|
||||
// User/presence
|
||||
"user.presence": DirToClient,
|
||||
"user.status": DirToClient,
|
||||
|
||||
// System
|
||||
"system.notify": DirToClient,
|
||||
"system.broadcast": DirToClient,
|
||||
|
||||
// Model/provider status
|
||||
"model.status": DirToClient,
|
||||
|
||||
// Plugin hooks — never cross the wire
|
||||
"plugin.hook.": DirLocal,
|
||||
"internal.": DirLocal,
|
||||
"db.": DirLocal,
|
||||
|
||||
// Heartbeat
|
||||
"ping": DirFromClient,
|
||||
"pong": DirToClient,
|
||||
}
|
||||
|
||||
// RouteFor returns the routing direction for a given event label.
|
||||
func RouteFor(label string) Direction {
|
||||
// Check exact match first
|
||||
if dir, ok := routeTable[label]; ok {
|
||||
return dir
|
||||
}
|
||||
// Check prefix match (longest prefix wins)
|
||||
best := ""
|
||||
bestDir := DirLocal
|
||||
for prefix, dir := range routeTable {
|
||||
if len(prefix) > len(best) && len(label) >= len(prefix) && label[:len(prefix)] == prefix {
|
||||
best = prefix
|
||||
bestDir = dir
|
||||
}
|
||||
}
|
||||
return bestDir
|
||||
}
|
||||
|
||||
// ShouldSendToClient returns true if this event should be forwarded to FE.
|
||||
func ShouldSendToClient(label string) bool {
|
||||
d := RouteFor(label)
|
||||
return d == DirToClient || d == DirBoth
|
||||
}
|
||||
|
||||
// ShouldAcceptFromClient returns true if this event is allowed from FE.
|
||||
func ShouldAcceptFromClient(label string) bool {
|
||||
d := RouteFor(label)
|
||||
return d == DirFromClient || d == DirBoth
|
||||
}
|
||||
279
server/events/ws.go
Normal file
279
server/events/ws.go
Normal file
@@ -0,0 +1,279 @@
|
||||
package events
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/gorilla/websocket"
|
||||
)
|
||||
|
||||
var upgrader = websocket.Upgrader{
|
||||
ReadBufferSize: 1024,
|
||||
WriteBufferSize: 1024,
|
||||
CheckOrigin: func(r *http.Request) bool { return true },
|
||||
}
|
||||
|
||||
const (
|
||||
writeWait = 10 * time.Second
|
||||
pongWait = 60 * time.Second
|
||||
pingPeriod = (pongWait * 9) / 10
|
||||
maxMsgSize = 4096
|
||||
)
|
||||
|
||||
// Hub manages WebSocket connections and bridges them to the Bus.
|
||||
type Hub struct {
|
||||
bus *Bus
|
||||
mu sync.RWMutex
|
||||
conns map[string]*Conn // connID → Conn
|
||||
}
|
||||
|
||||
// Conn represents a single WebSocket connection.
|
||||
type Conn struct {
|
||||
id string
|
||||
userID string
|
||||
rooms map[string]bool // rooms this connection is subscribed to
|
||||
ws *websocket.Conn
|
||||
send chan []byte
|
||||
hub *Hub
|
||||
unsubs []func() // bus unsubscribe functions
|
||||
}
|
||||
|
||||
// NewHub creates a Hub bound to an event bus.
|
||||
func NewHub(bus *Bus) *Hub {
|
||||
return &Hub{
|
||||
bus: bus,
|
||||
conns: make(map[string]*Conn),
|
||||
}
|
||||
}
|
||||
|
||||
// HandleWebSocket is a Gin handler for WebSocket upgrade.
|
||||
// Expects userID set in context (by auth middleware or query param).
|
||||
func (h *Hub) HandleWebSocket(c *gin.Context) {
|
||||
// Auth: check token from query param
|
||||
userID := c.GetString("user_id")
|
||||
if userID == "" {
|
||||
c.JSON(http.StatusUnauthorized, gin.H{"error": "authentication required"})
|
||||
return
|
||||
}
|
||||
|
||||
ws, err := upgrader.Upgrade(c.Writer, c.Request, nil)
|
||||
if err != nil {
|
||||
log.Printf("[ws] upgrade failed: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
connID := userID + "-" + time.Now().Format("150405.000")
|
||||
conn := &Conn{
|
||||
id: connID,
|
||||
userID: userID,
|
||||
rooms: make(map[string]bool),
|
||||
ws: ws,
|
||||
send: make(chan []byte, 64),
|
||||
hub: h,
|
||||
}
|
||||
|
||||
h.mu.Lock()
|
||||
h.conns[connID] = conn
|
||||
h.mu.Unlock()
|
||||
|
||||
log.Printf("[ws] connected: %s (user=%s, total=%d)", connID, userID, h.ConnCount())
|
||||
|
||||
// Subscribe this connection to bus events destined for clients
|
||||
conn.subscribeToBus()
|
||||
|
||||
// Publish presence
|
||||
h.bus.Publish(Event{
|
||||
Label: "user.presence",
|
||||
Payload: mustJSON(map[string]any{"user_id": userID, "status": "online"}),
|
||||
Ts: time.Now().UnixMilli(),
|
||||
SenderID: userID,
|
||||
})
|
||||
|
||||
// Start read/write pumps
|
||||
go conn.writePump()
|
||||
go conn.readPump()
|
||||
}
|
||||
|
||||
// ConnCount returns the number of active connections.
|
||||
func (h *Hub) ConnCount() int {
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
return len(h.conns)
|
||||
}
|
||||
|
||||
// ConnsByUser returns all connections for a given user.
|
||||
func (h *Hub) ConnsByUser(userID string) []*Conn {
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
var result []*Conn
|
||||
for _, c := range h.conns {
|
||||
if c.userID == userID {
|
||||
result = append(result, c)
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// removeConn cleans up a connection.
|
||||
func (h *Hub) removeConn(conn *Conn) {
|
||||
h.mu.Lock()
|
||||
delete(h.conns, conn.id)
|
||||
h.mu.Unlock()
|
||||
|
||||
// Unsubscribe from bus
|
||||
for _, unsub := range conn.unsubs {
|
||||
unsub()
|
||||
}
|
||||
|
||||
log.Printf("[ws] disconnected: %s (total=%d)", conn.id, h.ConnCount())
|
||||
|
||||
// Publish presence offline (only if no other conns for this user)
|
||||
if len(h.ConnsByUser(conn.userID)) == 0 {
|
||||
h.bus.Publish(Event{
|
||||
Label: "user.presence",
|
||||
Payload: mustJSON(map[string]any{"user_id": conn.userID, "status": "offline"}),
|
||||
Ts: time.Now().UnixMilli(),
|
||||
SenderID: conn.userID,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// subscribeToBus registers this conn as a bus subscriber for client-facing events.
|
||||
func (c *Conn) subscribeToBus() {
|
||||
unsub := c.hub.bus.Subscribe("*", func(e Event) {
|
||||
// Only forward events destined for clients
|
||||
if !ShouldSendToClient(e.Label) {
|
||||
return
|
||||
}
|
||||
|
||||
// Don't echo typing events back to the sender
|
||||
if e.Label == "chat.typing" || e.ConnID == c.id {
|
||||
if e.SenderID == c.userID && e.ConnID == c.id {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Room filtering: if event has a room, only send if conn is in that room
|
||||
if e.Room != "" && !c.rooms[e.Room] {
|
||||
return
|
||||
}
|
||||
|
||||
data, err := json.Marshal(e)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
select {
|
||||
case c.send <- data:
|
||||
default:
|
||||
// send buffer full, drop event
|
||||
}
|
||||
})
|
||||
|
||||
c.unsubs = append(c.unsubs, unsub)
|
||||
}
|
||||
|
||||
// JoinRoom adds this connection to a named room.
|
||||
func (c *Conn) JoinRoom(room string) {
|
||||
c.rooms[room] = true
|
||||
}
|
||||
|
||||
// LeaveRoom removes this connection from a named room.
|
||||
func (c *Conn) LeaveRoom(room string) {
|
||||
delete(c.rooms, room)
|
||||
}
|
||||
|
||||
// ── Read/Write Pumps ─────────────────────────
|
||||
|
||||
func (c *Conn) readPump() {
|
||||
defer func() {
|
||||
c.hub.removeConn(c)
|
||||
c.ws.Close()
|
||||
}()
|
||||
|
||||
c.ws.SetReadLimit(maxMsgSize)
|
||||
c.ws.SetReadDeadline(time.Now().Add(pongWait))
|
||||
c.ws.SetPongHandler(func(string) error {
|
||||
c.ws.SetReadDeadline(time.Now().Add(pongWait))
|
||||
return nil
|
||||
})
|
||||
|
||||
for {
|
||||
_, message, err := c.ws.ReadMessage()
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
|
||||
var event Event
|
||||
if err := json.Unmarshal(message, &event); err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
// Handle ping
|
||||
if event.Label == "ping" {
|
||||
pong, _ := json.Marshal(Event{Label: "pong", Ts: time.Now().UnixMilli()})
|
||||
select {
|
||||
case c.send <- pong:
|
||||
default:
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
// Validate: only accept events allowed from clients
|
||||
if !ShouldAcceptFromClient(event.Label) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Tag with sender info
|
||||
event.SenderID = c.userID
|
||||
event.ConnID = c.id
|
||||
if event.Ts == 0 {
|
||||
event.Ts = time.Now().UnixMilli()
|
||||
}
|
||||
|
||||
// Publish into bus
|
||||
c.hub.bus.Publish(event)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Conn) writePump() {
|
||||
ticker := time.NewTicker(pingPeriod)
|
||||
defer func() {
|
||||
ticker.Stop()
|
||||
c.ws.Close()
|
||||
}()
|
||||
|
||||
for {
|
||||
select {
|
||||
case msg, ok := <-c.send:
|
||||
c.ws.SetWriteDeadline(time.Now().Add(writeWait))
|
||||
if !ok {
|
||||
c.ws.WriteMessage(websocket.CloseMessage, []byte{})
|
||||
return
|
||||
}
|
||||
if err := c.ws.WriteMessage(websocket.TextMessage, msg); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
case <-ticker.C:
|
||||
c.ws.SetWriteDeadline(time.Now().Add(writeWait))
|
||||
if err := c.ws.WriteMessage(websocket.PingMessage, nil); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Helpers ──────────────────────────────────
|
||||
|
||||
func mustJSON(v any) json.RawMessage {
|
||||
data, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
return json.RawMessage(`{}`)
|
||||
}
|
||||
return data
|
||||
}
|
||||
Reference in New Issue
Block a user