Align GoChat with Chatwoot frontend contracts

internal/ws/event_publisher.go

@@ -18,19 +18,20 @@ import (
 // after create/update/delete actions via Wisper → ActionCable.
 //
 // Architecture:
-//   Service.CreateX() → EventPublisher.PublishEvent(accountID, eventType, payload)
-//     → Hub.SendToAccount (WebSocket local delivery)
-//     → SSERegistry.SendToAccount (SSE local delivery)
-//     → BroadcastRelay.Publish (Redis Pub/Sub cross-instance delivery)
+//
+//	Service.CreateX() → EventPublisher.PublishEvent(accountID, eventType, payload)
+//	  → Hub.SendToAccount (WebSocket local delivery)
+//	  → SSERegistry.SendToAccount (SSE local delivery)
+//	  → BroadcastRelay.Publish (Redis Pub/Sub cross-instance delivery)
 //
 // Reference: Chatwoot uses Wisper (in-process pub/sub) + ActionCable (WebSocket)
 // + Redis Pub/Sub for cross-instance. GoChat uses EventPublisher as the unified
 // entry point, with Hub and SSERegistry as local delivery targets, and Redis
 // Pub/Sub relay for multi-instance fan-out.
 type EventPublisher struct {
-	hub   MessageHandler   // WebSocket Hub (local delivery)
-	sse   *SSERegistry     // SSE registry (local delivery)
-	relay *BroadcastRelay  // Redis Pub/Sub relay (cross-instance delivery)
+	hub   MessageHandler  // WebSocket Hub (local delivery)
+	sse   *SSERegistry    // SSE registry (local delivery)
+	relay *BroadcastRelay // Redis Pub/Sub relay (cross-instance delivery)
 }
 
 // NewEventPublisher creates an EventPublisher with all delivery targets.
@@ -60,9 +61,9 @@ func NewEventPublisherLocal(hub MessageHandler, sse *SSERegistry) *EventPublishe
 //   - payload: event data (will be JSON-encoded for transport)
 //
 // The event is routed to:
-//   1. WebSocket Hub → all locally connected WS clients for that account
-//   2. SSE Registry → all locally connected SSE clients for that account
-//   3. Redis Pub/Sub relay → all other GoChat instances (cross-instance delivery)
+//  1. WebSocket Hub → all locally connected WS clients for that account
+//  2. SSE Registry → all locally connected SSE clients for that account
+//  3. Redis Pub/Sub relay → all other GoChat instances (cross-instance delivery)
 //
 // Reference: Chatwoot controllers call broadcast_event after mutations,
 // which triggers Wisper → ActionCable → Redis Pub/Sub relay.
@@ -145,6 +146,47 @@ func (p *EventPublisher) PublishConversationEvent(accountID uint, conversationID
 	}
 }
 
+// PublishWidgetEvent publishes an event to the account room and the widget
+// contact's pubsub_token room. Chatwoot's widget ActionCable connector
+// subscribes to RoomChannel with pubsub_token, so widget-visible events must be
+// available on that token-scoped room in addition to the dashboard account room.
+func (p *EventPublisher) PublishWidgetEvent(accountID uint, pubsubToken string, eventType string, payload interface{}) {
+	wsMsg := &WSMessage{
+		Event:     eventType,
+		Data:      payload,
+		AccountID: accountID,
+	}
+
+	data, err := json.Marshal(wsMsg)
+	if err != nil {
+		applogger.L().Warnf("event publisher: failed to marshal widget event %s: %v", eventType, err)
+		return
+	}
+
+	if p.hub != nil {
+		p.hub.SendToAccount(accountID, data)
+		if pubsubToken != "" {
+			p.hub.SendToRoom(pubsubTokenRoomNameHelper(pubsubToken), data)
+		}
+	}
+
+	if p.sse != nil {
+		p.sse.SendToAccount(accountID, SSEEvent{Type: eventType, Payload: payload})
+	}
+
+	if p.relay != nil {
+		room := accountRoomNameHelper(accountID)
+		if err := p.relay.Publish(context.Background(), room, wsMsg); err != nil {
+			applogger.L().Warnf("event publisher: redis publish failed for %s: %v", eventType, err)
+		}
+		if pubsubToken != "" {
+			if err := p.relay.Publish(context.Background(), pubsubTokenRoomNameHelper(pubsubToken), wsMsg); err != nil {
+				applogger.L().Warnf("event publisher: redis publish failed for widget %s: %v", eventType, err)
+			}
+		}
+	}
+}
+
 // accountRoomNameHelper generates the room name for an account channel.
 func accountRoomNameHelper(accountID uint) string {
 	return fmt.Sprintf("account_%d", accountID)
@@ -153,4 +195,8 @@ func accountRoomNameHelper(accountID uint) string {
 // conversationRoomNameHelper generates the room name for a conversation channel.
 func conversationRoomNameHelper(accountID uint, conversationID uint) string {
 	return fmt.Sprintf("account_%d_conversation_%d", accountID, conversationID)
-}
+}
+
+func pubsubTokenRoomNameHelper(token string) string {
+	return fmt.Sprintf("pubsub_token_%s", token)
+}

internal/ws/event_publisher_test.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -76,7 +77,7 @@ func TestEventPublisher_ConversationEvent(t *testing.T) {
 	publisher := NewEventPublisherLocal(hub, sse)
 
 	// Subscribe SSE clients
-	chBroad := sse.Subscribe("sse_broad", 1, 100)  // no conv filter
+	chBroad := sse.Subscribe("sse_broad", 1, 100) // no conv filter
 	chFiltered := sse.Subscribe("sse_filtered", 1, 200)
 	sse.SubscribeConversation("sse_filtered", 42)
 
@@ -135,6 +136,8 @@ func TestEventPublisher_AllEventTypes(t *testing.T) {
 		EventInboxDeleted,
 		EventNotificationCreated,
 		EventNotificationUpdated,
+		EventNotificationDeleted,
+		EventAccountCacheInvalidated,
 	}
 
 	for _, eventType := range eventTypes {
@@ -221,6 +224,406 @@ func TestEventPublisher_WSMessageFormat(t *testing.T) {
 	assert.Equal(t, "test message", dataMap["content"])
 }
 
+func TestEventPublisher_MessageCreatedChatwootPayloadShape(t *testing.T) {
+	hub := newMockHandler()
+	sse := NewSSERegistry()
+	publisher := NewEventPublisherLocal(hub, sse)
+
+	ch := sse.Subscribe("message-created-fixture", 1, 100)
+	require.NotNil(t, ch)
+
+	payload := map[string]interface{}{
+		"id":                 501,
+		"conversation_id":    42,
+		"display_id":         42,
+		"account_id":         1,
+		"inbox_id":           7,
+		"message_type":       "outgoing",
+		"content_type":       "text",
+		"content":            "hello from agent",
+		"content_attributes": map[string]interface{}{"submitted_email": "contact@example.com"},
+		"sender": map[string]interface{}{
+			"id":    100,
+			"name":  "Agent Ada",
+			"type":  "user",
+			"email": "ada@example.com",
+		},
+		"attachments": []interface{}{
+			map[string]interface{}{
+				"id":              900,
+				"message_id":      501,
+				"file_type":       "image",
+				"data_url":        "https://cdn.example.test/image.png",
+				"download_url":    "https://cdn.example.test/image.png",
+				"conversation_id": 42,
+			},
+		},
+	}
+	publisher.PublishEvent(1, EventMessageCreated, payload)
+
+	data := hub.getAccountData(1)
+	require.NotNil(t, data)
+	var wsMsg WSMessage
+	require.NoError(t, json.Unmarshal(data, &wsMsg))
+	require.Equal(t, EventMessageCreated, wsMsg.Event)
+	require.Equal(t, uint(1), wsMsg.AccountID)
+
+	dataMap, ok := wsMsg.Data.(map[string]interface{})
+	require.True(t, ok)
+	require.Equal(t, float64(501), dataMap["id"])
+	require.Equal(t, float64(42), dataMap["conversation_id"])
+	require.Equal(t, "outgoing", dataMap["message_type"])
+	require.Equal(t, "text", dataMap["content_type"])
+	require.Equal(t, "hello from agent", dataMap["content"])
+	require.Contains(t, dataMap, "content_attributes")
+	require.Contains(t, dataMap, "sender")
+	require.Contains(t, dataMap, "attachments")
+	require.Equal(t, "Agent Ada", dataMap["sender"].(map[string]interface{})["name"])
+	attachments := dataMap["attachments"].([]interface{})
+	require.Len(t, attachments, 1)
+	require.Equal(t, float64(42), attachments[0].(map[string]interface{})["conversation_id"])
+
+	select {
+	case event := <-ch.Events:
+		require.Equal(t, EventMessageCreated, event.Type)
+		fixture := event.Payload.(map[string]interface{})
+		require.Equal(t, "hello from agent", fixture["content"])
+		require.Contains(t, fixture, "sender")
+		require.Contains(t, fixture, "attachments")
+	default:
+		t.Fatal("SSE channel should receive Chatwoot message.created payload")
+	}
+}
+
+func TestEventPublisher_MessageUpdatedChatwootPayloadShape(t *testing.T) {
+	hub := newMockHandler()
+	sse := NewSSERegistry()
+	publisher := NewEventPublisherLocal(hub, sse)
+
+	ch := sse.Subscribe("message-updated-fixture", 1, 100)
+	require.NotNil(t, ch)
+
+	payload := map[string]interface{}{
+		"id":              502,
+		"conversation_id": 42,
+		"account_id":      1,
+		"inbox_id":        7,
+		"message_type":    "outgoing",
+		"content_type":    "text",
+		"content":         "provider rejected",
+		"status":          "failed",
+		"content_attributes": map[string]interface{}{
+			"external_error": "provider rejected message",
+		},
+		"sender": map[string]interface{}{
+			"id":    100,
+			"name":  "Agent Ada",
+			"type":  "user",
+			"email": "ada@example.com",
+		},
+		"attachments": []interface{}{},
+	}
+	publisher.PublishEvent(1, EventMessageUpdated, payload)
+
+	data := hub.getAccountData(1)
+	require.NotNil(t, data)
+	var wsMsg WSMessage
+	require.NoError(t, json.Unmarshal(data, &wsMsg))
+	require.Equal(t, EventMessageUpdated, wsMsg.Event)
+	require.Equal(t, uint(1), wsMsg.AccountID)
+
+	dataMap, ok := wsMsg.Data.(map[string]interface{})
+	require.True(t, ok)
+	require.Equal(t, float64(502), dataMap["id"])
+	require.Equal(t, float64(42), dataMap["conversation_id"])
+	require.Equal(t, "failed", dataMap["status"])
+	require.Equal(t, "provider rejected", dataMap["content"])
+	contentAttrs := dataMap["content_attributes"].(map[string]interface{})
+	require.Equal(t, "provider rejected message", contentAttrs["external_error"])
+	require.Contains(t, dataMap, "sender")
+	require.Contains(t, dataMap, "attachments")
+
+	select {
+	case event := <-ch.Events:
+		require.Equal(t, EventMessageUpdated, event.Type)
+		fixture := event.Payload.(map[string]interface{})
+		require.Equal(t, 502, fixture["id"])
+		require.Equal(t, "failed", fixture["status"])
+	case <-time.After(time.Second):
+		t.Fatal("SSE channel should receive message.updated fixture")
+	}
+}
+
+func TestEventPublisher_ConversationMutationChatwootPayloadShape(t *testing.T) {
+	hub := newMockHandler()
+	sse := NewSSERegistry()
+	publisher := NewEventPublisherLocal(hub, sse)
+
+	ch := sse.Subscribe("conversation-mutation-fixture", 1, 100)
+	require.NotNil(t, ch)
+
+	payload := map[string]interface{}{
+		"id":          42,
+		"account_id":  1,
+		"inbox_id":    7,
+		"status":      "resolved",
+		"priority":    "urgent",
+		"labels":      []interface{}{"support", "urgent"},
+		"assignee_id": 100,
+		"team_id":     55,
+		"changed_attributes": map[string]interface{}{
+			"status":   map[string]interface{}{"from": "open", "to": "resolved"},
+			"priority": map[string]interface{}{"from": "medium", "to": "urgent"},
+		},
+		"meta": map[string]interface{}{
+			"assignee": map[string]interface{}{"id": 100, "name": "Agent Ada"},
+			"team":     map[string]interface{}{"id": 55, "name": "Support"},
+		},
+	}
+	publisher.PublishEvent(1, EventConversationUpdated, payload)
+
+	data := hub.getAccountData(1)
+	require.NotNil(t, data)
+	var wsMsg WSMessage
+	require.NoError(t, json.Unmarshal(data, &wsMsg))
+	require.Equal(t, EventConversationUpdated, wsMsg.Event)
+	require.Equal(t, uint(1), wsMsg.AccountID)
+
+	dataMap, ok := wsMsg.Data.(map[string]interface{})
+	require.True(t, ok)
+	require.Equal(t, float64(42), dataMap["id"])
+	require.Equal(t, "resolved", dataMap["status"])
+	require.Equal(t, "urgent", dataMap["priority"])
+	require.Contains(t, dataMap, "labels")
+	require.Contains(t, dataMap, "meta")
+	changes := dataMap["changed_attributes"].(map[string]interface{})
+	require.Equal(t, "resolved", changes["status"].(map[string]interface{})["to"])
+	require.Equal(t, "urgent", changes["priority"].(map[string]interface{})["to"])
+
+	select {
+	case event := <-ch.Events:
+		require.Equal(t, EventConversationUpdated, event.Type)
+		fixture := event.Payload.(map[string]interface{})
+		require.Equal(t, 42, fixture["id"])
+		require.Equal(t, "resolved", fixture["status"])
+	case <-time.After(time.Second):
+		t.Fatal("SSE channel should receive conversation.updated fixture")
+	}
+}
+
+func TestEventPublisher_TypingChatwootPayloadShape(t *testing.T) {
+	hub := newMockHandler()
+	sse := NewSSERegistry()
+	publisher := NewEventPublisherLocal(hub, sse)
+
+	ch := sse.Subscribe("typing-fixture", 1, 100)
+	sse.SubscribeConversation("typing-fixture", 42)
+	require.NotNil(t, ch)
+
+	payload := map[string]interface{}{
+		"conversation_id": 42,
+		"account_id":      1,
+		"user_id":         100,
+		"typing_status":   "on",
+		"is_private":      true,
+	}
+	publisher.PublishConversationEvent(1, 42, EventConversationTypingOn, payload)
+
+	data := hub.getAccountData(1)
+	require.NotNil(t, data)
+	var wsMsg WSMessage
+	require.NoError(t, json.Unmarshal(data, &wsMsg))
+	require.Equal(t, EventConversationTypingOn, wsMsg.Event)
+	dataMap := wsMsg.Data.(map[string]interface{})
+	require.Equal(t, float64(42), dataMap["conversation_id"])
+	require.Equal(t, float64(100), dataMap["user_id"])
+	require.Equal(t, "on", dataMap["typing_status"])
+	require.Equal(t, true, dataMap["is_private"])
+
+	select {
+	case event := <-ch.Events:
+		require.Equal(t, EventConversationTypingOn, event.Type)
+		fixture := event.Payload.(map[string]interface{})
+		require.Equal(t, 42, fixture["conversation_id"])
+		require.Equal(t, "on", fixture["typing_status"])
+	case <-time.After(time.Second):
+		t.Fatal("SSE channel should receive conversation.typing_on fixture")
+	}
+}
+
+func TestEventPublisher_PresenceUpdateChatwootPayloadShape(t *testing.T) {
+	hub := newMockHandler()
+	sse := NewSSERegistry()
+	publisher := NewEventPublisherLocal(hub, sse)
+
+	ch := sse.Subscribe("presence-fixture", 1, 100)
+	require.NotNil(t, ch)
+
+	payload := map[string]interface{}{
+		"account_id": 1,
+		"users":      map[uint]string{100: "busy"},
+		"contacts":   map[uint]string{501: "online"},
+	}
+	publisher.PublishEvent(1, EventPresenceUpdate, payload)
+
+	data := hub.getAccountData(1)
+	require.NotNil(t, data)
+	var wsMsg WSMessage
+	require.NoError(t, json.Unmarshal(data, &wsMsg))
+	require.Equal(t, EventPresenceUpdate, wsMsg.Event)
+	require.Equal(t, uint(1), wsMsg.AccountID)
+
+	dataMap := wsMsg.Data.(map[string]interface{})
+	require.Equal(t, float64(1), dataMap["account_id"])
+	users := dataMap["users"].(map[string]interface{})
+	require.Equal(t, "busy", users["100"])
+	contacts := dataMap["contacts"].(map[string]interface{})
+	require.Equal(t, "online", contacts["501"])
+
+	select {
+	case event := <-ch.Events:
+		require.Equal(t, EventPresenceUpdate, event.Type)
+		fixture := event.Payload.(map[string]interface{})
+		require.Equal(t, map[uint]string{100: "busy"}, fixture["users"])
+	case <-time.After(time.Second):
+		t.Fatal("SSE channel should receive presence.update fixture")
+	}
+}
+
+func TestEventPublisher_NotificationChatwootPayloadShape(t *testing.T) {
+	hub := newMockHandler()
+	sse := NewSSERegistry()
+	publisher := NewEventPublisherLocal(hub, sse)
+
+	ch := sse.Subscribe("notification-fixture", 1, 100)
+	require.NotNil(t, ch)
+
+	notification := map[string]interface{}{
+		"id":                   701,
+		"account_id":           1,
+		"user_id":              100,
+		"notification_type":    "conversation_created",
+		"primary_actor_type":   "Conversation",
+		"primary_actor_id":     42,
+		"secondary_actor_type": "Message",
+		"secondary_actor_id":   501,
+		"read_at":              nil,
+		"snoozed_until":        nil,
+		"created_at":           "2026-06-12T10:00:00Z",
+		"last_activity_at":     "2026-06-12T10:01:00Z",
+		"meta": map[string]interface{}{
+			"sender": map[string]interface{}{"id": 301, "name": "Visitor"},
+		},
+		"additional_attributes": map[string]interface{}{
+			"browser_title": "New customer message",
+		},
+		"primary_actor": map[string]interface{}{
+			"id":         42,
+			"display_id": 42,
+			"status":     "open",
+		},
+		"secondary_actor": map[string]interface{}{
+			"id":              501,
+			"conversation_id": 42,
+			"content":         "hello",
+		},
+		"user": map[string]interface{}{
+			"id":   100,
+			"name": "Agent Ada",
+		},
+	}
+
+	for _, eventType := range []string{EventNotificationCreated, EventNotificationUpdated, EventNotificationDeleted} {
+		t.Run(eventType, func(t *testing.T) {
+			payload := map[string]interface{}{
+				"notification": notification,
+				"unread_count": 3,
+				"count":        7,
+			}
+			publisher.PublishEvent(1, eventType, payload)
+
+			data := hub.getAccountData(1)
+			require.NotNil(t, data)
+			var wsMsg WSMessage
+			require.NoError(t, json.Unmarshal(data, &wsMsg))
+			require.Equal(t, eventType, wsMsg.Event)
+			require.Equal(t, uint(1), wsMsg.AccountID)
+
+			dataMap, ok := wsMsg.Data.(map[string]interface{})
+			require.True(t, ok)
+			require.Equal(t, float64(3), dataMap["unread_count"])
+			require.Equal(t, float64(7), dataMap["count"])
+
+			notificationMap := dataMap["notification"].(map[string]interface{})
+			require.Equal(t, float64(701), notificationMap["id"])
+			require.Equal(t, float64(1), notificationMap["account_id"])
+			require.Equal(t, "conversation_created", notificationMap["notification_type"])
+			require.Equal(t, "Conversation", notificationMap["primary_actor_type"])
+			require.Equal(t, float64(42), notificationMap["primary_actor_id"])
+			require.Equal(t, "Message", notificationMap["secondary_actor_type"])
+			require.Equal(t, float64(501), notificationMap["secondary_actor_id"])
+			require.Contains(t, notificationMap, "read_at")
+			require.Contains(t, notificationMap, "snoozed_until")
+			require.Contains(t, notificationMap, "meta")
+			require.Contains(t, notificationMap, "additional_attributes")
+			require.Contains(t, notificationMap, "primary_actor")
+			require.Contains(t, notificationMap, "secondary_actor")
+			require.Contains(t, notificationMap, "user")
+
+			select {
+			case event := <-ch.Events:
+				require.Equal(t, eventType, event.Type)
+				fixture := event.Payload.(map[string]interface{})
+				require.Equal(t, 3, fixture["unread_count"])
+				require.Equal(t, 7, fixture["count"])
+				require.Equal(t, notification, fixture["notification"])
+			case <-time.After(time.Second):
+				t.Fatalf("SSE channel should receive %s fixture", eventType)
+			}
+		})
+	}
+}
+
+func TestEventPublisher_AccountCacheInvalidatedChatwootPayloadShape(t *testing.T) {
+	hub := newMockHandler()
+	sse := NewSSERegistry()
+	publisher := NewEventPublisherLocal(hub, sse)
+
+	ch := sse.Subscribe("account-cache-fixture", 1, 100)
+	require.NotNil(t, ch)
+
+	payload := map[string]interface{}{
+		"cache_keys": map[string]interface{}{
+			"label": "labels-v2",
+			"inbox": "inboxes-v2",
+			"team":  "teams-v2",
+		},
+	}
+	publisher.PublishEvent(1, EventAccountCacheInvalidated, payload)
+
+	data := hub.getAccountData(1)
+	require.NotNil(t, data)
+	var wsMsg WSMessage
+	require.NoError(t, json.Unmarshal(data, &wsMsg))
+	require.Equal(t, EventAccountCacheInvalidated, wsMsg.Event)
+	require.Equal(t, uint(1), wsMsg.AccountID)
+
+	dataMap := wsMsg.Data.(map[string]interface{})
+	cacheKeys := dataMap["cache_keys"].(map[string]interface{})
+	require.Equal(t, "labels-v2", cacheKeys["label"])
+	require.Equal(t, "inboxes-v2", cacheKeys["inbox"])
+	require.Equal(t, "teams-v2", cacheKeys["team"])
+
+	select {
+	case event := <-ch.Events:
+		require.Equal(t, EventAccountCacheInvalidated, event.Type)
+		fixture := event.Payload.(map[string]interface{})
+		require.Equal(t, payload["cache_keys"], fixture["cache_keys"])
+	case <-time.After(time.Second):
+		t.Fatal("SSE channel should receive account.cache_invalidated fixture")
+	}
+}
+
 // === Mock MessageHandler reuse ===
 // The mockHandler type is defined in broadcast_test.go in the same package.
 // We reuse it here for EventPublisher tests. If needed, here's a duplicate
@@ -231,7 +634,7 @@ func TestEventPublisher_WSMessageFormat(t *testing.T) {
 
 // Additional mock for tracking room-based sends
 type mockRoomHandler struct {
-	mu      sync.Mutex
+	mu       sync.Mutex
 	accounts map[uint][]byte
 	rooms    map[string][]byte
 }
@@ -283,4 +686,26 @@ func TestEventPublisher_ConversationEvent_RoomDelivery(t *testing.T) {
 	convRoom := conversationRoomNameHelper(1, 42)
 	roomData := hub.getRoomData(convRoom)
 	require.NotNil(t, roomData, "Hub conversation room should receive event")
-}
+}
+
+func TestEventPublisher_WidgetEvent_PubsubTokenRoomDelivery(t *testing.T) {
+	hub := newMockRoomHandler()
+	sse := NewSSERegistry()
+	publisher := NewEventPublisherLocal(hub, sse)
+
+	payload := map[string]interface{}{"id": 7, "content": "hello widget"}
+	publisher.PublishWidgetEvent(1, "pubsub-123", EventMessageCreated, payload)
+
+	accountData := hub.getAccountData(1)
+	require.NotNil(t, accountData, "Hub account room should receive widget event")
+	var accountMsg WSMessage
+	require.NoError(t, json.Unmarshal(accountData, &accountMsg))
+	assert.Equal(t, EventMessageCreated, accountMsg.Event)
+
+	roomData := hub.getRoomData("pubsub_token_pubsub-123")
+	require.NotNil(t, roomData, "Hub pubsub_token room should receive widget event")
+	var roomMsg WSMessage
+	require.NoError(t, json.Unmarshal(roomData, &roomMsg))
+	assert.Equal(t, EventMessageCreated, roomMsg.Event)
+	assert.Equal(t, uint(1), roomMsg.AccountID)
+}

internal/ws/event_types.go

@@ -41,6 +41,7 @@ const (
 	EventContactUpdated = "contact.updated"
 	EventContactDeleted = "contact.deleted"
 	EventContactMerged  = "contact.merged"
+	EventCompanyUpdated = "company.updated"
 )
 
 // Notification events
@@ -78,23 +79,24 @@ const (
 
 // --- Protocol control event constants ---
 const (
-	EventSubscribeConfirm    = "subscribe.confirm"
-	EventUnsubscribeConfirm  = "unsubscribe.confirm"
-	EventSubscribeReject     = "subscribe.reject"
-	EventPingResponse        = "ping.response"
-	EventWelcome             = "welcome"
-	EventDisconnect          = "disconnect"
+	EventSubscribeConfirm   = "subscribe.confirm"
+	EventUnsubscribeConfirm = "unsubscribe.confirm"
+	EventSubscribeReject    = "subscribe.reject"
+	EventPingResponse       = "ping.response"
+	EventWelcome            = "welcome"
+	EventDisconnect         = "disconnect"
 )
 
 // --- Wire-format message types ---
 // WSMessage is the JSON envelope for all server→client WebSocket messages.
 // Mirrors Chatwoot's ActionCableBroadcastJob event format:
-//   {event: "message.created", data: {...}, account_id: 1, performer: {id, name, type, avatar_url}}
+//
+//	{event: "message.created", data: {...}, account_id: 1, performer: {id, name, type, avatar_url}}
 type WSMessage struct {
-	Event     string    `json:"event"`                // e.g. "message.created"
-	Data      any       `json:"data"`                 // event payload
-	AccountID uint      `json:"account_id,omitempty"` // target account
-	Performer *Performer `json:"performer,omitempty"` // who triggered the event
+	Event     string     `json:"event"`                // e.g. "message.created"
+	Data      any        `json:"data"`                 // event payload
+	AccountID uint       `json:"account_id,omitempty"` // target account
+	Performer *Performer `json:"performer,omitempty"`  // who triggered the event
 }
 
 // Performer describes who triggered an event, mirroring Chatwoot's
@@ -102,24 +104,24 @@ type WSMessage struct {
 type Performer struct {
 	ID        uint   `json:"id"`
 	Name      string `json:"name"`
-	Type      string `json:"type"`       // "user" or "contact"
+	Type      string `json:"type"` // "user" or "contact"
 	AvatarURL string `json:"avatar_url"`
 }
 
 // WSCommand is the JSON envelope for client→server WebSocket commands.
 // Mirrors Chatwoot's ActionCable command structure.
 type WSCommand struct {
-	Command string `json:"command"`              // subscribe, unsubscribe, ping, typing_on, typing_off, update_presence
-	Data    string `json:"data,omitempty"`       // JSON-encoded command payload
+	Command string `json:"command"`        // subscribe, unsubscribe, ping, typing_on, typing_off, update_presence
+	Data    string `json:"data,omitempty"` // JSON-encoded command payload
 }
 
 // SubscribeData is the payload for a subscribe command.
 // Mirrors Chatwoot RoomChannel identifier.
 type SubscribeData struct {
-	Channel        string `json:"channel"`                    // "AccountChannel" or "ConversationChannel"
-	AccountID      uint   `json:"account_id"`                 // required for both channels
-	PubsubToken    string `json:"pubsub_token,omitempty"`     // for contact-based auth
-	ConversationID uint   `json:"conversation_id,omitempty"`  // required for ConversationChannel
+	Channel        string `json:"channel"`                   // "AccountChannel" or "ConversationChannel"
+	AccountID      uint   `json:"account_id"`                // required for both channels
+	PubsubToken    string `json:"pubsub_token,omitempty"`    // for contact-based auth
+	ConversationID uint   `json:"conversation_id,omitempty"` // required for ConversationChannel
 }
 
 // TypingData is the payload for typing_on/typing_off commands.
@@ -142,13 +144,13 @@ const (
 // --- Redis key prefix constants ---
 const (
 	// Pub/Sub channel prefixes for cross-instance relay
-	RedisPrefixRoom    = "gochat:ws:room:"     // gochat:ws:room:account_{id}
-	RedisPrefixAccount = "gochat:ws:account:"  // gochat:ws:account:{id}
+	RedisPrefixRoom    = "gochat:ws:room:"    // gochat:ws:room:account_{id}
+	RedisPrefixAccount = "gochat:ws:account:" // gochat:ws:account:{id}
 
 	// Sorted set / hash keys for presence tracking
-	RedisKeyPresenceAgents    = "gochat:presence:agents"     // sorted set: score=timestamp, member=agent_id:account_id
-	RedisKeyPresenceContacts  = "gochat:presence:contacts"   // sorted set: score=timestamp, member=contact_id:account_id
-	RedisKeyPresenceStatus    = "gochat:presence:status"     // hash: field=id:account_id, value=status string
+	RedisKeyPresenceAgents   = "gochat:presence:agents"   // sorted set: score=timestamp, member=agent_id:account_id
+	RedisKeyPresenceContacts = "gochat:presence:contacts" // sorted set: score=timestamp, member=contact_id:account_id
+	RedisKeyPresenceStatus   = "gochat:presence:status"   // hash: field=id:account_id, value=status string
 
 	// Typing indicator keys with TTL
 	RedisKeyTyping = "gochat:typing:%d:%d" // gochat:typing:{account_id}:{conversation_id}
@@ -166,4 +168,4 @@ const (
 
 	// TypingTTLSec is how long a typing indicator persists before auto-expiry.
 	TypingTTLSec = 4
-)
+)

internal/ws/presence.go

@@ -59,15 +59,7 @@ func (p *PresenceTracker) SetAgentOnline(ctx context.Context, agentID, accountID
 		return fmt.Errorf("failed to set agent status: %w", err)
 	}
 
-	// Broadcast presence update event
-	msg := &WSMessage{
-		Event: EventAgentOnline,
-		Data: map[string]any{
-			"agent_id":   agentID,
-			"account_id": accountID,
-		},
-		AccountID: accountID,
-	}
+	msg := presenceUpdateMessage(accountID, map[uint]string{agentID: "online"}, nil)
 
 	if err := p.relay.PublishAccount(ctx, accountID, msg); err != nil {
 		logger.L().Warnf("ws presence: failed to broadcast agent online: %v", err)
@@ -94,15 +86,7 @@ func (p *PresenceTracker) SetAgentOffline(ctx context.Context, agentID, accountI
 		return fmt.Errorf("failed to remove agent status: %w", err)
 	}
 
-	// Broadcast offline event
-	msg := &WSMessage{
-		Event: EventAgentOffline,
-		Data: map[string]any{
-			"agent_id":   agentID,
-			"account_id": accountID,
-		},
-		AccountID: accountID,
-	}
+	msg := presenceUpdateMessage(accountID, map[uint]string{agentID: "offline"}, nil)
 
 	if err := p.relay.PublishAccount(ctx, accountID, msg); err != nil {
 		logger.L().Warnf("ws presence: failed to broadcast agent offline: %v", err)
@@ -129,16 +113,7 @@ func (p *PresenceTracker) SetAgentBusy(ctx context.Context, agentID, accountID u
 		return fmt.Errorf("failed to set agent status busy: %w", err)
 	}
 
-	// Broadcast presence update
-	msg := &WSMessage{
-		Event: EventPresenceUpdate,
-		Data: map[string]any{
-			"agent_id":   agentID,
-			"account_id": accountID,
-			"status":     "busy",
-		},
-		AccountID: accountID,
-	}
+	msg := presenceUpdateMessage(accountID, map[uint]string{agentID: "busy"}, nil)
 
 	if err := p.relay.PublishAccount(ctx, accountID, msg); err != nil {
 		logger.L().Warnf("ws presence: failed to broadcast agent busy: %v", err)
@@ -164,6 +139,11 @@ func (p *PresenceTracker) SetContactOnline(ctx context.Context, contactID, accou
 		return fmt.Errorf("failed to set contact status: %w", err)
 	}
 
+	msg := presenceUpdateMessage(accountID, nil, map[uint]string{contactID: "online"})
+	if err := p.relay.PublishAccount(ctx, accountID, msg); err != nil {
+		logger.L().Warnf("ws presence: failed to broadcast contact online: %v", err)
+	}
+
 	logger.L().Debugf("ws presence: contact %d online for account %d", contactID, accountID)
 	return nil
 }
@@ -180,10 +160,33 @@ func (p *PresenceTracker) SetContactOffline(ctx context.Context, contactID, acco
 		return fmt.Errorf("failed to remove contact status: %w", err)
 	}
 
+	msg := presenceUpdateMessage(accountID, nil, map[uint]string{contactID: "offline"})
+	if err := p.relay.PublishAccount(ctx, accountID, msg); err != nil {
+		logger.L().Warnf("ws presence: failed to broadcast contact offline: %v", err)
+	}
+
 	logger.L().Debugf("ws presence: contact %d offline for account %d", contactID, accountID)
 	return nil
 }
 
+func presenceUpdateMessage(accountID uint, users map[uint]string, contacts map[uint]string) *WSMessage {
+	if users == nil {
+		users = map[uint]string{}
+	}
+	if contacts == nil {
+		contacts = map[uint]string{}
+	}
+	return &WSMessage{
+		Event: EventPresenceUpdate,
+		Data: map[string]any{
+			"account_id": accountID,
+			"users":      users,
+			"contacts":   contacts,
+		},
+		AccountID: accountID,
+	}
+}
+
 // GetOnlineAgentsForAccount returns all currently online agent IDs for an account.
 // Only returns agents whose sorted-set score is within PresenceDurationAgentSec
 // of the current time (i.e., have sent a heartbeat recently).
@@ -255,6 +258,24 @@ func (p *PresenceTracker) CleanupExpired(ctx context.Context) error {
 	agentThreshold := float64(time.Now().Unix() - PresenceDurationAgentSec)
 	contactThreshold := float64(time.Now().Unix() - PresenceDurationContactSec)
 
+	expiredAgents, agentListErr := p.rdb.ZRangeByScore(ctx, RedisKeyPresenceAgents, &redis.ZRangeBy{
+		Min: "0",
+		Max: fmt.Sprintf("%f", agentThreshold),
+	}).Result()
+	if agentListErr != nil {
+		logger.L().Warnf("ws presence: failed to list expired agents: %v", agentListErr)
+		expiredAgents = nil
+	}
+
+	expiredContacts, contactListErr := p.rdb.ZRangeByScore(ctx, RedisKeyPresenceContacts, &redis.ZRangeBy{
+		Min: "0",
+		Max: fmt.Sprintf("%f", contactThreshold),
+	}).Result()
+	if contactListErr != nil {
+		logger.L().Warnf("ws presence: failed to list expired contacts: %v", contactListErr)
+		expiredContacts = nil
+	}
+
 	// Remove expired agents
 	removedAgents, err := p.rdb.ZRemRangeByScore(ctx, RedisKeyPresenceAgents,
 		"0", fmt.Sprintf("%f", agentThreshold)).Result()
@@ -269,6 +290,8 @@ func (p *PresenceTracker) CleanupExpired(ctx context.Context) error {
 		logger.L().Warnf("ws presence: failed to cleanup expired contacts: %v", err)
 	}
 
+	p.cleanupExpiredStatusesAndBroadcast(ctx, expiredAgents, expiredContacts)
+
 	if removedAgents > 0 || removedContacts > 0 {
 		logger.L().Infof("ws presence: cleanup removed %d agents, %d contacts",
 			removedAgents, removedContacts)
@@ -276,6 +299,61 @@ func (p *PresenceTracker) CleanupExpired(ctx context.Context) error {
 	return nil
 }
 
+func (p *PresenceTracker) cleanupExpiredStatusesAndBroadcast(ctx context.Context, expiredAgents, expiredContacts []string) {
+	updates := map[uint]struct {
+		users    map[uint]string
+		contacts map[uint]string
+	}{}
+
+	addUpdate := func(accountID uint) struct {
+		users    map[uint]string
+		contacts map[uint]string
+	} {
+		update := updates[accountID]
+		if update.users == nil {
+			update.users = map[uint]string{}
+		}
+		if update.contacts == nil {
+			update.contacts = map[uint]string{}
+		}
+		updates[accountID] = update
+		return update
+	}
+
+	for _, member := range expiredAgents {
+		agentID, accountID := parsePresenceMember(member)
+		if agentID == 0 || accountID == 0 {
+			continue
+		}
+		update := addUpdate(accountID)
+		update.users[agentID] = "offline"
+		updates[accountID] = update
+		if err := p.rdb.HDel(ctx, RedisKeyPresenceStatus, member).Err(); err != nil {
+			logger.L().Warnf("ws presence: failed to delete expired agent status %s: %v", member, err)
+		}
+	}
+
+	for _, member := range expiredContacts {
+		contactID, accountID := parsePresenceMember(member)
+		if contactID == 0 || accountID == 0 {
+			continue
+		}
+		update := addUpdate(accountID)
+		update.contacts[contactID] = "offline"
+		updates[accountID] = update
+		if err := p.rdb.HDel(ctx, RedisKeyPresenceStatus, member).Err(); err != nil {
+			logger.L().Warnf("ws presence: failed to delete expired contact status %s: %v", member, err)
+		}
+	}
+
+	for accountID, update := range updates {
+		msg := presenceUpdateMessage(accountID, update.users, update.contacts)
+		if err := p.relay.PublishAccount(ctx, accountID, msg); err != nil {
+			logger.L().Warnf("ws presence: failed to broadcast expired presence for account %d: %v", accountID, err)
+		}
+	}
+}
+
 // RefreshAgentPresence refreshes an agent's timestamp in the sorted set.
 // Called on each heartbeat to prevent the agent from being cleaned up.
 func (p *PresenceTracker) RefreshAgentPresence(ctx context.Context, agentID, accountID uint) error {

internal/ws/presence_test.go

@@ -2,6 +2,7 @@ package ws
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"testing"
 	"time"
@@ -187,6 +188,47 @@ func TestSetAgentBusy(t *testing.T) {
 	assert.Greater(t, score, 0.0, "busy agent 应仍在 sorted set 中")
 }
 
+func TestPresenceTracker_BroadcastsChatwootPresenceUpdatePayload(t *testing.T) {
+	_, rdb, tracker, _, cleanup := setupPresenceTest(t)
+	defer cleanup()
+
+	ctx := context.Background()
+	channel := RedisPrefixAccount + "5"
+	sub := rdb.Subscribe(ctx, channel)
+	defer sub.Close()
+	_, err := sub.Receive(ctx)
+	require.NoError(t, err)
+
+	require.NoError(t, tracker.SetAgentBusy(ctx, 10, 5))
+	msgCh := sub.Channel()
+	select {
+	case redisMsg := <-msgCh:
+		assert.Equal(t, channel, redisMsg.Channel)
+		var wsMsg WSMessage
+		require.NoError(t, json.Unmarshal([]byte(redisMsg.Payload), &wsMsg))
+		assert.Equal(t, EventPresenceUpdate, wsMsg.Event)
+		data := wsMsg.Data.(map[string]interface{})
+		assert.Equal(t, float64(5), data["account_id"])
+		users := data["users"].(map[string]interface{})
+		assert.Equal(t, "busy", users["10"])
+	case <-time.After(2 * time.Second):
+		t.Fatal("should receive agent presence.update payload")
+	}
+
+	require.NoError(t, tracker.SetContactOnline(ctx, 99, 5))
+	select {
+	case redisMsg := <-msgCh:
+		var wsMsg WSMessage
+		require.NoError(t, json.Unmarshal([]byte(redisMsg.Payload), &wsMsg))
+		assert.Equal(t, EventPresenceUpdate, wsMsg.Event)
+		data := wsMsg.Data.(map[string]interface{})
+		contacts := data["contacts"].(map[string]interface{})
+		assert.Equal(t, "online", contacts["99"])
+	case <-time.After(2 * time.Second):
+		t.Fatal("should receive contact presence.update payload")
+	}
+}
+
 // --- SetContactOnline / SetContactOffline 测试 ---
 
 func TestSetContactOnline(t *testing.T) {
@@ -466,6 +508,48 @@ func TestCleanupExpired(t *testing.T) {
 	assert.Equal(t, int64(0), contactCount, "过期 contact 应被清理")
 }
 
+func TestCleanupExpired_BroadcastsChatwootOfflinePayloadAndClearsStatus(t *testing.T) {
+	_, rdb, tracker, _, cleanup := setupPresenceTest(t)
+	defer cleanup()
+
+	ctx := context.Background()
+	channel := RedisPrefixAccount + "10"
+	sub := rdb.Subscribe(ctx, channel)
+	defer sub.Close()
+	_, err := sub.Receive(ctx)
+	require.NoError(t, err)
+
+	expiredAgentScore := float64(time.Now().Unix() - PresenceDurationAgentSec - 10)
+	expiredContactScore := float64(time.Now().Unix() - PresenceDurationContactSec - 10)
+	require.NoError(t, rdb.ZAdd(ctx, RedisKeyPresenceAgents, redis.Z{Score: expiredAgentScore, Member: "1:10"}).Err())
+	require.NoError(t, rdb.HSet(ctx, RedisKeyPresenceStatus, "1:10", "busy").Err())
+	require.NoError(t, rdb.ZAdd(ctx, RedisKeyPresenceContacts, redis.Z{Score: expiredContactScore, Member: "100:10"}).Err())
+	require.NoError(t, rdb.HSet(ctx, RedisKeyPresenceStatus, "100:10", "online").Err())
+
+	require.NoError(t, tracker.CleanupExpired(ctx))
+
+	select {
+	case redisMsg := <-sub.Channel():
+		require.Equal(t, channel, redisMsg.Channel)
+		var wsMsg WSMessage
+		require.NoError(t, json.Unmarshal([]byte(redisMsg.Payload), &wsMsg))
+		require.Equal(t, EventPresenceUpdate, wsMsg.Event)
+		data := wsMsg.Data.(map[string]interface{})
+		require.Equal(t, float64(10), data["account_id"])
+		users := data["users"].(map[string]interface{})
+		require.Equal(t, "offline", users["1"])
+		contacts := data["contacts"].(map[string]interface{})
+		require.Equal(t, "offline", contacts["100"])
+	case <-time.After(2 * time.Second):
+		t.Fatal("should receive presence.update offline payload for expired records")
+	}
+
+	_, err = rdb.HGet(ctx, RedisKeyPresenceStatus, "1:10").Result()
+	require.Equal(t, redis.Nil, err)
+	_, err = rdb.HGet(ctx, RedisKeyPresenceStatus, "100:10").Result()
+	require.Equal(t, redis.Nil, err)
+}
+
 func TestCleanupExpired_NoExpired(t *testing.T) {
 	_, _, tracker, _, cleanup := setupPresenceTest(t)
 	defer cleanup()