goodie/pkg/server/bus.go

package server

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"sync"
	"time"
)

// Bus implements a simple long-polling message bus for Discuss.
// Mirrors: odoo/addons/bus/models/bus.py ImBus
//
// Channels subscribe to notifications. A long-poll request blocks until
// a notification arrives or the timeout expires.
type Bus struct {
	mu       sync.Mutex
	channels map[int64][]chan busNotification
	lastID   int64
}

type busNotification struct {
	ID      int64       `json:"id"`
	Channel string      `json:"channel"`
	Message interface{} `json:"message"`
}

// NewBus creates a new message bus.
func NewBus() *Bus {
	return &Bus{
		channels: make(map[int64][]chan busNotification),
	}
}

// Notify sends a notification to all subscribers of a channel.
func (b *Bus) Notify(channelID int64, channel string, message interface{}) {
	b.mu.Lock()
	b.lastID++
	notif := busNotification{
		ID:      b.lastID,
		Channel: channel,
		Message: message,
	}
	subs := b.channels[channelID]
	b.mu.Unlock()

	for _, ch := range subs {
		select {
		case ch <- notif:
		default:
			// subscriber buffer full, skip
		}
	}
}

// Subscribe creates a subscription for a partner's channels.
func (b *Bus) Subscribe(partnerID int64) chan busNotification {
	ch := make(chan busNotification, 10)
	b.mu.Lock()
	b.channels[partnerID] = append(b.channels[partnerID], ch)
	b.mu.Unlock()
	return ch
}

// Unsubscribe removes a subscription.
func (b *Bus) Unsubscribe(partnerID int64, ch chan busNotification) {
	b.mu.Lock()
	defer b.mu.Unlock()
	subs := b.channels[partnerID]
	for i, s := range subs {
		if s == ch {
			b.channels[partnerID] = append(subs[:i], subs[i+1:]...)
			close(ch)
			return
		}
	}
}

// registerBusRoutes adds the long-polling endpoint.
func (s *Server) registerBusRoutes() {
	if s.bus == nil {
		s.bus = NewBus()
	}
	s.mux.HandleFunc("/longpolling/poll", s.handleBusPoll)
	s.mux.HandleFunc("/discuss/channel/messages", s.handleDiscussMessages)
	s.mux.HandleFunc("/discuss/channel/list", s.handleDiscussChannelList)
}

// handleBusPoll implements long-polling for real-time notifications.
// Mirrors: odoo/addons/bus/controllers/main.py poll()
func (s *Server) handleBusPoll(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
		return
	}

	sess := GetSession(r)
	if sess == nil {
		writeJSON(w, []interface{}{})
		return
	}

	// Get partner ID
	var partnerID int64
	s.pool.QueryRow(r.Context(),
		`SELECT COALESCE(partner_id, 0) FROM res_users WHERE id = $1`, sess.UID,
	).Scan(&partnerID)

	if partnerID == 0 {
		writeJSON(w, []interface{}{})
		return
	}

	// Subscribe and wait for notifications (max 30s)
	ch := s.bus.Subscribe(partnerID)
	defer s.bus.Unsubscribe(partnerID, ch)

	ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
	defer cancel()

	select {
	case notif := <-ch:
		writeJSON(w, []busNotification{notif})
	case <-ctx.Done():
		writeJSON(w, []interface{}{}) // timeout, empty response
	}
}

// handleDiscussMessages fetches messages for a channel via JSON-RPC.
func (s *Server) handleDiscussMessages(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
		return
	}

	sess := GetSession(r)
	if sess == nil {
		s.writeJSONRPC(w, nil, nil, &RPCError{Code: 100, Message: "Not authenticated"})
		return
	}

	var req JSONRPCRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		s.writeJSONRPC(w, nil, nil, &RPCError{Code: -32700, Message: "Parse error"})
		return
	}

	var params struct {
		ChannelID int64 `json:"channel_id"`
		Limit     int   `json:"limit"`
	}
	if err := json.Unmarshal(req.Params, &params); err != nil {
		s.writeJSONRPC(w, req.ID, nil, &RPCError{Code: -32602, Message: "Invalid params"})
		return
	}
	if params.Limit <= 0 {
		params.Limit = 50
	}

	rows, err := s.pool.Query(r.Context(),
		`SELECT m.id, m.body, m.date, m.author_id, COALESCE(p.name, '')
		 FROM mail_message m
		 LEFT JOIN res_partner p ON p.id = m.author_id
		 WHERE m.model = 'mail.channel' AND m.res_id = $1
		 ORDER BY m.id DESC LIMIT $2`, params.ChannelID, params.Limit)
	if err != nil {
		s.writeJSONRPC(w, req.ID, nil, &RPCError{Code: -32603, Message: fmt.Sprintf("Query: %v", err)})
		return
	}
	defer rows.Close()

	var messages []map[string]interface{}
	for rows.Next() {
		var id, authorID int64
		var body, authorName string
		var date interface{}
		if err := rows.Scan(&id, &body, &date, &authorID, &authorName); err != nil {
			continue
		}
		msg := map[string]interface{}{
			"id": id, "body": body, "date": date,
		}
		if authorID > 0 {
			msg["author_id"] = []interface{}{authorID, authorName}
		} else {
			msg["author_id"] = false
		}
		messages = append(messages, msg)
	}
	if messages == nil {
		messages = []map[string]interface{}{}
	}
	s.writeJSONRPC(w, req.ID, messages, nil)
}

// handleDiscussChannelList returns channels the current user is member of.
func (s *Server) handleDiscussChannelList(w http.ResponseWriter, r *http.Request) {
	sess := GetSession(r)
	if sess == nil {
		s.writeJSONRPC(w, nil, nil, &RPCError{Code: 100, Message: "Not authenticated"})
		return
	}

	var partnerID int64
	s.pool.QueryRow(r.Context(),
		`SELECT COALESCE(partner_id, 0) FROM res_users WHERE id = $1`, sess.UID,
	).Scan(&partnerID)

	rows, err := s.pool.Query(r.Context(),
		`SELECT c.id, c.name, c.channel_type,
		        (SELECT COUNT(*) FROM mail_channel_member WHERE channel_id = c.id) AS members
		 FROM mail_channel c
		 JOIN mail_channel_member cm ON cm.channel_id = c.id AND cm.partner_id = $1
		 WHERE c.active = true
		 ORDER BY c.last_message_date DESC NULLS LAST`, partnerID)
	if err != nil {
		log.Printf("discuss: channel list error: %v", err)
		writeJSON(w, []interface{}{})
		return
	}
	defer rows.Close()

	var channels []map[string]interface{}
	for rows.Next() {
		var id int64
		var name, channelType string
		var members int64
		if err := rows.Scan(&id, &name, &channelType, &members); err != nil {
			continue
		}
		channels = append(channels, map[string]interface{}{
			"id": id, "name": name, "channel_type": channelType, "member_count": members,
		})
	}
	if channels == nil {
		channels = []map[string]interface{}{}
	}
	writeJSON(w, channels)
}