gochat/backend/internal/ws/event_publisher.go

// Package ws provides the EventPublisher that unifies real-time event delivery
// across WebSocket Hub and SSE fallback channels. Services call PublishEvent
// after mutations (CreateConversation, SendMessage, etc.) to push events to
// all connected clients, mirroring Chatwoot's ActionCable broadcast pattern.
package ws

import (
	"context"
	"crypto/sha256"
	"encoding/json"
	"fmt"

	"github.com/gochat/gochat/internal/model"
	"github.com/gochat/gochat/internal/worker"
	applogger "github.com/gochat/gochat/pkg/logger"
	"gorm.io/gorm"
)

const taskTypeRealtimeEventPublish = "realtime:event_publish"

const (
	realtimeTargetAccount = "account"
	realtimeTargetToken   = "pubsub_token"
)

type realtimeEventPublishJob struct {
	AccountID   uint        `json:"account_id"`
	PubsubToken string      `json:"pubsub_token,omitempty"`
	Target      string      `json:"target,omitempty"`
	EventType   string      `json:"event_type"`
	Payload     interface{} `json:"payload"`
}

// EventPublisher routes business events to both WebSocket and SSE clients.
// It is the central point that services call after mutations to push real-time
// updates, matching Chatwoot's pattern where controllers broadcast events
// 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)
//
// 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)
	worker *worker.WorkerPool
}

// NewEventPublisher creates an EventPublisher with all delivery targets.
func NewEventPublisher(hub MessageHandler, sse *SSERegistry, relay *BroadcastRelay) *EventPublisher {
	return &EventPublisher{
		hub:   hub,
		sse:   sse,
		relay: relay,
	}
}

// NewEventPublisherLocal creates an EventPublisher with only local delivery
// (no Redis relay). Used for development/testing without Redis.
func NewEventPublisherLocal(hub MessageHandler, sse *SSERegistry) *EventPublisher {
	return &EventPublisher{
		hub: hub,
		sse: sse,
	}
}

// SetWorkerPool makes realtime delivery durable. The request path only stores
// the job; Redis failures are retried and remain visible in background_jobs.
func (p *EventPublisher) SetWorkerPool(wp *worker.WorkerPool) {
	p.worker = wp
	if wp != nil {
		wp.Register(taskTypeRealtimeEventPublish, p.performPublishJob)
	}
}

// PublishEvent publishes a real-time event to all delivery targets.
// This is the primary API that services call after business mutations.
//
// Parameters:
//   - accountID: the account scope for the event (each client only sees their account's events)
//   - eventType: dot-separated event name (e.g. "message.created", "conversation.updated")
//   - 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)
//
// Reference: Chatwoot controllers call broadcast_event after mutations,
// which triggers Wisper → ActionCable → Redis Pub/Sub relay.
func (p *EventPublisher) PublishEvent(accountID uint, eventType string, payload interface{}) error {
	if p.worker != nil {
		return p.enqueuePublishJobs(context.Background(), realtimePublishJobs(accountID, "", eventType, payload))
	}
	return p.publishEvent(accountID, eventType, payload)
}

func (p *EventPublisher) publishEvent(accountID uint, eventType string, payload interface{}) error {
	// Build the wire-format message (matching Chatwoot ActionCable event format)
	wsMsg := &WSMessage{
		Event:     eventType,
		Data:      payload,
		AccountID: accountID,
	}

	data, err := json.Marshal(wsMsg)
	if err != nil {
		return fmt.Errorf("marshal event %s: %w", eventType, err)
	}

	// 1. Deliver to WebSocket Hub (local clients)
	if p.hub != nil {
		p.hub.SendToAccount(accountID, data)
	}

	// 2. Deliver to SSE registry (local clients)
	if p.sse != nil {
		p.sse.SendToAccount(accountID, SSEEvent{
			Type:    eventType,
			Payload: payload,
		})
	}

	// 3. Publish to Redis Pub/Sub relay (cross-instance delivery)
	if p.relay != nil {
		room := accountRoomNameHelper(accountID)
		if err := p.relay.Publish(context.Background(), room, wsMsg); err != nil {
			return fmt.Errorf("publish event %s to account room: %w", eventType, err)
		}
	}
	return nil
}

// PublishConversationEvent publishes a real-time event scoped to a specific
// conversation within an account. This is used for events that should only
// reach clients viewing that conversation (e.g. typing indicators, message updates).
//
// Reference: Chatwoot ConversationChannel — events broadcast to clients
// subscribed to a specific conversation's channel.
func (p *EventPublisher) PublishConversationEvent(accountID uint, conversationID uint, 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 conversation event %s: %v", eventType, err)
		return
	}

	// 1. Deliver to WebSocket Hub (both account room and conversation room)
	if p.hub != nil {
		p.hub.SendToAccount(accountID, data)
		convRoom := conversationRoomNameHelper(accountID, conversationID)
		p.hub.SendToRoom(convRoom, data)
	}

	// 2. Deliver to SSE registry (conversation-filtered channels)
	if p.sse != nil {
		p.sse.SendToConversation(accountID, conversationID, SSEEvent{
			Type:    eventType,
			Payload: payload,
		})
	}

	// 3. Publish to Redis Pub/Sub relay (both account and conversation channels)
	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)
		}
	}
}

// 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{}) error {
	if p.worker != nil {
		return p.enqueuePublishJobs(context.Background(), realtimePublishJobs(accountID, pubsubToken, eventType, payload))
	}
	return p.publishWidgetEvent(accountID, pubsubToken, eventType, payload)
}

func (p *EventPublisher) publishWidgetEvent(accountID uint, pubsubToken string, eventType string, payload interface{}) error {
	if err := p.publishEvent(accountID, eventType, payload); err != nil {
		return err
	}
	if pubsubToken == "" {
		return nil
	}
	return p.publishTokenEvent(accountID, pubsubToken, eventType, payload)
}

func (p *EventPublisher) publishTokenEvent(accountID uint, pubsubToken string, eventType string, payload interface{}) error {
	wsMsg := &WSMessage{
		Event:     eventType,
		Data:      payload,
		AccountID: accountID,
	}

	data, err := json.Marshal(wsMsg)
	if err != nil {
		return fmt.Errorf("marshal widget event %s: %w", eventType, err)
	}

	if p.hub != nil {
		p.hub.SendToRoom(pubsubTokenRoomNameHelper(pubsubToken), data)
	}

	if p.relay != nil {
		if err := p.relay.Publish(context.Background(), pubsubTokenRoomNameHelper(pubsubToken), wsMsg); err != nil {
			return fmt.Errorf("publish widget event %s to token room: %w", eventType, err)
		}
	}
	return nil
}

func realtimePublishJobs(accountID uint, pubsubToken, eventType string, payload interface{}) []realtimeEventPublishJob {
	jobs := []realtimeEventPublishJob{{AccountID: accountID, Target: realtimeTargetAccount, EventType: eventType, Payload: payload}}
	if pubsubToken != "" {
		jobs = append(jobs, realtimeEventPublishJob{AccountID: accountID, PubsubToken: pubsubToken, Target: realtimeTargetToken, EventType: eventType, Payload: payload})
	}
	return jobs
}

func realtimePublishJobOptions(job realtimeEventPublishJob) ([]worker.EnqueueOption, error) {
	opts := []worker.EnqueueOption{worker.WithQueue("events"), worker.WithMaxAttempts(3)}
	if job.EventType == EventMessageCreated {
		encoded, err := json.Marshal(job)
		if err != nil {
			return nil, fmt.Errorf("marshal realtime publish job: %w", err)
		}
		opts = append(opts, worker.WithIdempotencyKey(fmt.Sprintf("realtime:message.created:%s:%x", job.Target, sha256.Sum256(encoded))))
	}
	return opts, nil
}

func (p *EventPublisher) enqueuePublishJobs(ctx context.Context, jobs []realtimeEventPublishJob) error {
	for _, job := range jobs {
		opts, err := realtimePublishJobOptions(job)
		if err != nil {
			return err
		}
		if _, err := p.worker.Enqueue(ctx, taskTypeRealtimeEventPublish, job, opts...); err != nil {
			applogger.L().Errorf("enqueue realtime event %s failed: %v", job.EventType, err)
			return err
		}
	}
	return nil
}

// EnqueueInTransaction atomically persists every target channel with the
// caller's business mutation. The returned jobs are safe to publish only after
// the transaction commits.
func (p *EventPublisher) EnqueueInTransaction(ctx context.Context, tx *gorm.DB, accountID uint, pubsubToken, eventType string, payload interface{}) ([]*model.BackgroundJob, error) {
	if p.worker == nil {
		return nil, worker.ErrWorkerDatabaseRequired
	}
	var createdJobs []*model.BackgroundJob
	for _, job := range realtimePublishJobs(accountID, pubsubToken, eventType, payload) {
		opts, err := realtimePublishJobOptions(job)
		if err != nil {
			return nil, err
		}
		backgroundJob, created, err := p.worker.EnqueueInTransaction(ctx, tx, taskTypeRealtimeEventPublish, job, opts...)
		if err != nil {
			return nil, err
		}
		if created {
			createdJobs = append(createdJobs, backgroundJob)
		}
	}
	return createdJobs, nil
}

// PublishEnqueued triggers Redis after the surrounding transaction commits.
// Database sweep recovery remains authoritative if this process exits here.
func (p *EventPublisher) PublishEnqueued(ctx context.Context, jobs []*model.BackgroundJob) {
	for _, job := range jobs {
		p.worker.Publish(ctx, job)
	}
}

func (p *EventPublisher) performPublishJob(ctx context.Context, backgroundJob *model.BackgroundJob) error {
	var job realtimeEventPublishJob
	if err := json.Unmarshal(backgroundJob.Payload, &job); err != nil {
		return worker.Permanent(fmt.Errorf("unmarshal realtime publish job: %w", err))
	}
	if job.AccountID == 0 || job.EventType == "" {
		return worker.Permanent(fmt.Errorf("invalid realtime publish job: account_id=%d event_type=%q", job.AccountID, job.EventType))
	}
	var err error
	switch job.Target {
	case realtimeTargetToken:
		err = p.publishTokenJob(job.AccountID, job.PubsubToken, job.EventType, job.Payload)
	default:
		err = p.publishAccountJob(job.AccountID, job.EventType, job.Payload)
	}
	if err != nil {
		applogger.L().Errorf("realtime event job %d failed for %s: %v", backgroundJob.ID, job.EventType, err)
	}
	return err
}

func (p *EventPublisher) publishAccountJob(accountID uint, eventType string, payload interface{}) error {
	if p.relay == nil {
		return p.publishEvent(accountID, eventType, payload)
	}
	if err := p.relay.Publish(context.Background(), accountRoomNameHelper(accountID), &WSMessage{Event: eventType, Data: payload, AccountID: accountID}); err != nil {
		return err
	}
	if p.sse != nil {
		p.sse.SendToAccount(accountID, SSEEvent{Type: eventType, Payload: payload})
	}
	return nil
}

func (p *EventPublisher) publishTokenJob(accountID uint, pubsubToken, eventType string, payload interface{}) error {
	if p.relay == nil {
		return p.publishTokenEvent(accountID, pubsubToken, eventType, payload)
	}
	return p.relay.Publish(context.Background(), pubsubTokenRoomNameHelper(pubsubToken), &WSMessage{Event: eventType, Data: payload, AccountID: accountID})
}

// accountRoomNameHelper generates the room name for an account channel.
func accountRoomNameHelper(accountID uint) string {
	return fmt.Sprintf("account_%d", accountID)
}

// 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)
}