package httpapi

import (
	"encoding/json"
	"sync"
	"time"
)

type queueEvent struct {
	ProjectID string    `json:"project_id"`
	Revision  int64     `json:"revision"`
	Type      string    `json:"type"`
	At        time.Time `json:"at"`
}

// eventPublisher is the boundary for replacing the per-process fallback with
// a shared Redis-backed bus when Kubernetes runs multiple API pods. The
// current MVP intentionally keeps the in-memory implementation local.
type eventPublisher interface {
	subscribe(projectID string) (<-chan []byte, func())
	publish(event queueEvent)
}

type eventHub struct {
	mu          sync.RWMutex
	subscribers map[string]map[chan []byte]struct{}
}

func newEventHub() *eventHub {
	return &eventHub{subscribers: make(map[string]map[chan []byte]struct{})}
}

func (h *eventHub) subscribe(projectID string) (<-chan []byte, func()) {
	ch := make(chan []byte, 8)
	h.mu.Lock()
	if h.subscribers[projectID] == nil {
		h.subscribers[projectID] = make(map[chan []byte]struct{})
	}
	h.subscribers[projectID][ch] = struct{}{}
	h.mu.Unlock()
	return ch, func() {
		h.mu.Lock()
		delete(h.subscribers[projectID], ch)
		if len(h.subscribers[projectID]) == 0 {
			delete(h.subscribers, projectID)
		}
		h.mu.Unlock()
	}
}

func (h *eventHub) publish(event queueEvent) {
	body, err := json.Marshal(event)
	if err != nil {
		return
	}
	h.mu.RLock()
	defer h.mu.RUnlock()
	for ch := range h.subscribers[event.ProjectID] {
		select {
		case ch <- body:
		default:
			// A slow client can recover from the authoritative snapshot/revision.
		}
	}
}