Bridge channel dispatcher events to WebSocket EventPublisher for real-time delivery

backend/internal/app/bootstrap.go

@@ -14,6 +14,7 @@ import (
 	"github.com/gochat/gochat/internal/campaign"
 	"github.com/gochat/gochat/internal/canned"
 	"github.com/gochat/gochat/internal/channel"
+	"github.com/gochat/gochat/internal/wsevent"
 	emailchannel "github.com/gochat/gochat/internal/channel/email"
 	facebookchannel "github.com/gochat/gochat/internal/channel/facebook"
 	googlechannel "github.com/gochat/gochat/internal/channel/google"
@@ -727,6 +728,10 @@ func Bootstrap(env string) (*App, error) {
 	eventPublisher := wspkg.NewEventPublisher(wsHub, nil, wsRelay)
 	presenceTracker := wspkg.NewPresenceTracker(rdb, wsRelay)
 
+	// Bridge channel dispatcher events to the WebSocket EventPublisher so that
+	// message/conversation/contact events reach connected WS/SSE clients.
+	channelDispatcher.Register(wsevent.New(eventPublisher))
+
 	// Widget service + handler (M11 — WebWidget channel completion)
 	// hubTypingAdapter delegates typing events to the WS hub's direct broadcast,
 	// avoiding Redis dependency (the full TypingTracker requires Redis).

backend/internal/wsevent/bridge_listener.go

@@ -0,0 +1,102 @@
+// Package wsevent provides a bridge listener that forwards channel dispatcher
+// events to the WebSocket EventPublisher for real-time delivery to connected
+// clients (WebSocket + SSE + Redis Pub/Sub relay).
+//
+// This lives in a separate package to avoid an import cycle between
+// internal/channel (which defines EventListener) and internal/ws (which defines
+// EventPublisher).
+package wsevent
+
+import (
+	"context"
+
+	"github.com/gochat/gochat/internal/channel"
+	wspkg "github.com/gochat/gochat/internal/ws"
+	applogger "github.com/gochat/gochat/pkg/logger"
+)
+
+// BridgeListener implements channel.EventListener and forwards events to the
+// WebSocket EventPublisher.
+//
+// Problem this solves: MessageService.dispatchMessageEvent() dispatches events
+// to the channel.Dispatcher, but without this listener those events never reach
+// the WS Hub / SSE Registry / Redis Pub-Sub relay. The EventPublisher is the
+// component that actually delivers to connected clients — this listener is the
+// bridge.
+//
+// The string values of channel.EventType and wspkg event constants are identical
+// ("message.created", "conversation.updated", etc.), so we pass the type through
+// directly after casting to string.
+type BridgeListener struct {
+	publisher *wspkg.EventPublisher
+}
+
+// New creates a bridge listener that forwards channel dispatcher events to the
+// WebSocket EventPublisher.
+func New(publisher *wspkg.EventPublisher) *BridgeListener {
+	return &BridgeListener{publisher: publisher}
+}
+
+// Name returns the unique listener identifier.
+func (l *BridgeListener) Name() string {
+	return "ws_event_bridge"
+}
+
+// OnEvent forwards a channel dispatcher event to the WebSocket EventPublisher.
+// The event type string is passed through directly (channel.EventType and
+// wspkg event constants share the same string values).
+func (l *BridgeListener) OnEvent(ctx context.Context, event *channel.ChannelEvent) error {
+	eventType := string(event.Type)
+	if !isWSEventType(eventType) {
+		return nil
+	}
+
+	payload := event.Data
+	if payload == nil {
+		payload = map[string]interface{}{}
+	}
+
+	if event.ConversationID != 0 {
+		payload["conversation_id"] = event.ConversationID
+	}
+	if event.InboxID != 0 {
+		payload["inbox_id"] = event.InboxID
+	}
+
+	l.publisher.PublishEvent(event.AccountID, eventType, payload)
+
+	applogger.L().Debugf("ws_bridge: forwarded event %s for account %d", eventType, event.AccountID)
+	return nil
+}
+
+// isWSEventType returns true if the event type string corresponds to a
+// WebSocket/SSE event constant defined in wspkg.
+func isWSEventType(eventType string) bool {
+	switch eventType {
+	// Message events
+	case "message.created", "message.updated", "message.deleted":
+		return true
+	// Conversation events
+	case "conversation.created", "conversation.updated",
+		"conversation.resolved", "conversation.opened",
+		"conversation.assigned", "conversation.unassigned",
+		"conversation.status_changed", "conversation.typing_on",
+		"conversation.typing_off", "conversation.mentioned", "conversation.read":
+		return true
+	// Contact events
+	case "contact.created", "contact.updated", "contact.deleted", "contact.merged":
+		return true
+	// Inbox events
+	case "inbox.created", "inbox.updated", "inbox.deleted":
+		return true
+	// Presence / agent events
+	case "presence.update", "agent.typing_on", "agent.typing_off",
+		"agent.online", "agent.offline":
+		return true
+	// Notification events
+	case "notification.created", "notification.updated", "notification.deleted":
+		return true
+	default:
+		return false
+	}
+}

backend/scripts/verify_msg_e2e.py

@@ -0,0 +1,227 @@
+#!/usr/bin/env python3
+"""End-to-end message delivery verification for GoChat.
+
+Tests:
+1. Agent sends message via REST API → WebSocket receives message.created event
+2. Measures delivery latency (P50/P95/P99) and success rate over N messages
+3. Verifies event payload structure (event type, account_id, data.message)
+
+Usage:
+    python3 verify_msg_e2e.py [--host 127.0.0.1] [--port 3000] [--count 50]
+"""
+
+import argparse
+import asyncio
+import json
+import statistics
+import sys
+import time
+import urllib.request
+
+try:
+    import websockets
+except ImportError:
+    print("ERROR: pip install websockets", file=sys.stderr)
+    sys.exit(1)
+
+
+def login(host, port, email, password):
+    """Login and return Chatwoot auth headers (JWT access-token + client + uid)."""
+    url = f"http://{host}:{port}/auth/sign_in"
+    body = json.dumps({"email": email, "password": password}).encode()
+    req = urllib.request.Request(
+        url, data=body, method="POST",
+        headers={"Content-Type": "application/json"},
+    )
+    with urllib.request.urlopen(req, timeout=10) as resp:
+        headers = {k.lower(): v for k, v in resp.headers.items()}
+        body = json.loads(resp.read())
+    return {
+        "access-token": headers.get("access-token", ""),
+        "client": headers.get("client", ""),
+        "uid": headers.get("uid", ""),
+        "token-type": headers.get("token-type", "Bearer"),
+        "pubsub_token": body.get("data", {}).get("pubsub_token", ""),
+        "account_id": body.get("data", {}).get("account_id", 1),
+        "user": body.get("data", {}),
+    }
+
+
+def get_conversations(host, port, auth, account_id):
+    """List conversations to find one for testing."""
+    url = f"http://{host}:{port}/api/v1/accounts/{account_id}/conversations"
+    req = urllib.request.Request(url, method="GET", headers=auth_headers(auth))
+    with urllib.request.urlopen(req, timeout=10) as resp:
+        return json.loads(resp.read())
+
+
+def auth_headers(auth):
+    return {
+        "access-token": auth["access-token"],
+        "client": auth["client"],
+        "uid": auth["uid"],
+        "token-type": auth["token-type"],
+        "Content-Type": "application/json",
+    }
+
+
+def send_message(host, port, auth, account_id, conversation_id, content):
+    """POST a message via REST API, return (send_time, response_body)."""
+    url = (
+        f"http://{host}:{port}/api/v1/accounts/{account_id}"
+        f"/conversations/{conversation_id}/messages"
+    )
+    body = json.dumps({
+        "content": content,
+        "content_type": "text",
+        "message_type": "outgoing",
+        "private": False,
+    }).encode()
+    req = urllib.request.Request(url, data=body, method="POST", headers=auth_headers(auth))
+    send_time = time.monotonic()
+    with urllib.request.urlopen(req, timeout=10) as resp:
+        resp_body = json.loads(resp.read())
+    return send_time, resp_body
+
+
+async def ws_connect(host, port, auth, account_id):
+    """Connect to WebSocket /cable and subscribe to AccountChannel.
+
+    Uses ActionCable-compatible command format:
+      {"command":"subscribe","identifier":"{\"channel\":\"AccountChannel\",\"account_id\":1}"}
+    The client is auto-subscribed to the account room on connect (hub.go:180),
+    but we send an explicit subscribe for confirmation.
+    """
+    ws_url = f"ws://{host}:{port}/cable?token={auth['access-token']}"
+    ws = await websockets.connect(ws_url, max_size=2**20)
+    # Subscribe to AccountChannel (ActionCable format: identifier is JSON string)
+    identifier = json.dumps({"channel": "AccountChannel", "account_id": account_id})
+    sub_cmd = {"command": "subscribe", "identifier": identifier}
+    await ws.send(json.dumps(sub_cmd))
+    # Wait for subscribe confirmation or welcome
+    while True:
+        raw = await asyncio.wait_for(ws.recv(), timeout=5)
+        msg = json.loads(raw)
+        msg_type = msg.get("type", "")
+        if msg_type == "confirm_subscribe":
+            return ws
+        if msg.get("event") == "welcome":
+            # Auto-subscribed to account room; keep waiting for confirm
+            continue
+        if msg_type == "reject_subscribe":
+            raise RuntimeError(f"subscribe rejected: {msg.get('reason')}")
+        # Unknown message — assume connected (auto-subscribe may skip confirm)
+        return ws
+
+
+async def verify_message_delivery(host, port, count):
+    print(f"\n=== GoChat Message E2E Verification ===")
+    print(f"Target: http://{host}:{port}")
+    print(f"Messages: {count}")
+
+    # Step 1: Login
+    print("\n[1/4] Logging in as admin...")
+    auth = login(host, port, "admin@gochat.local", "changeme")
+    account_id = auth["account_id"]
+    print(f"  OK: user={auth['user']['name']}, account_id={account_id}")
+
+    # Step 2: Find a conversation
+    print("[2/4] Finding conversation...")
+    convos = get_conversations(host, port, auth, account_id)
+    payload = convos.get("data", {}).get("payload", [])
+    if not payload:
+        print("  FAIL: no conversations found. Run `gochat seed` first.")
+        return False
+    conv = payload[0]
+    conv_id = conv["id"]
+    sender = conv.get("meta", {}).get("sender", {}).get("name", "?")
+    print(f"  OK: conversation_id={conv_id} ({sender})")
+
+    # Step 3: Connect WebSocket
+    print("[3/4] Connecting WebSocket...")
+    try:
+        ws = await ws_connect(host, port, auth, account_id)
+    except Exception as e:
+        print(f"  FAIL: WS connect/subscribe failed: {e}")
+        return False
+    print(f"  OK: subscribed to AccountChannel")
+
+    # Step 4: Send messages and measure latency
+    print(f"[4/4] Sending {count} messages and measuring delivery latency...")
+    latencies = []
+    received = 0
+    missed = 0
+
+    for i in range(count):
+        content = f"e2e-test-{i:04d}-{int(time.time()*1000)}"
+        try:
+            send_time, resp = send_message(host, port, auth, account_id, conv_id, content)
+            msg_id = resp.get("id")
+        except Exception as e:
+            print(f"  [{i+1}] SEND FAIL: {e}")
+            missed += 1
+            continue
+
+        # Wait for message.created event on WS
+        try:
+            while True:
+                raw = await asyncio.wait_for(ws.recv(), timeout=5)
+                msg = json.loads(raw)
+                if msg.get("event") != "message.created":
+                    continue
+                # EventFrame uses 'payload' for data, WSMessage uses 'data'
+                data = msg.get("payload") or msg.get("data", {})
+                # message id is nested under data.message.id (dispatchMessageEvent
+                # sets event.Data["message"] = message)
+                msg_data = data.get("message", data)
+                if msg_data.get("id") == msg_id:
+                    recv_time = time.monotonic()
+                    latency_ms = (recv_time - send_time) * 1000
+                    latencies.append(latency_ms)
+                    received += 1
+                    break
+                # Different message — might be echo of our own or other; keep waiting
+        except asyncio.TimeoutError:
+            print(f"  [{i+1}] TIMEOUT: no WS event for msg_id={msg_id}")
+            missed += 1
+
+        if (i + 1) % 10 == 0:
+            print(f"  ...{i+1}/{count} sent, {received} received")
+
+    await ws.close()
+
+    # Results
+    print(f"\n=== Results ===")
+    print(f"Sent:      {count}")
+    print(f"Received:  {received}")
+    print(f"Missed:    {missed}")
+    if latencies:
+        latencies.sort()
+        p50 = latencies[len(latencies) // 2]
+        p95 = latencies[int(len(latencies) * 0.95)]
+        p99 = latencies[min(int(len(latencies) * 0.99), len(latencies) - 1)]
+        mean = statistics.mean(latencies)
+        print(f"\nLatency (ms):")
+        print(f"  Mean:  {mean:.1f}")
+        print(f"  P50:   {p50:.1f}")
+        print(f"  P95:   {p95:.1f}")
+        print(f"  P99:   {p99:.1f}")
+        print(f"  Min:   {latencies[0]:.1f}")
+        print(f"  Max:   {latencies[-1]:.1f}")
+        print(f"\nSuccess rate: {received/count*100:.1f}%")
+    return received == count
+
+
+def main():
+    parser = argparse.ArgumentParser(description="GoChat message E2E verification")
+    parser.add_argument("--host", default="127.0.0.1")
+    parser.add_argument("--port", type=int, default=3000)
+    parser.add_argument("--count", type=int, default=50, help="number of messages to send")
+    args = parser.parse_args()
+
+    ok = asyncio.run(verify_message_delivery(args.host, args.port, args.count))
+    sys.exit(0 if ok else 1)
+
+
+if __name__ == "__main__":
+    main()