// Package mqttnotifier publishes playlist-changed notifications to MQTT. // It is safe for concurrent use and applies per-screen debouncing so that // rapid edits within a 2-second window produce at most one MQTT message. package mqttnotifier import ( "fmt" "sync" "time" mqtt "github.com/eclipse/paho.mqtt.golang" ) const ( // debounceDuration is the minimum time between two publish calls for the // same screen. Any change arriving within this window resets the timer. debounceDuration = 2 * time.Second ) // Notifier publishes "playlist-changed" MQTT messages with per-screen debounce. // If no broker URL is configured it behaves as a no-op (all methods are safe // to call and do nothing). type Notifier struct { client mqtt.Client // nil when disabled mu sync.Mutex timers map[string]*time.Timer // keyed by screenSlug } // New creates a Notifier connected to broker (e.g. "tcp://mosquitto:1883"). // username/password may be empty. Returns a no-op Notifier when broker == "". func New(broker, username, password string) *Notifier { n := &Notifier{timers: make(map[string]*time.Timer)} if broker == "" { return n } opts := mqtt.NewClientOptions(). AddBroker(broker). SetClientID("morz-backend"). SetCleanSession(true). SetAutoReconnect(true). SetConnectRetry(true). SetConnectRetryInterval(10 * time.Second) if username != "" { opts.SetUsername(username) opts.SetPassword(password) } n.client = mqtt.NewClient(opts) n.client.Connect() // non-blocking; paho retries in background return n } // Topic returns the MQTT topic for a screen's playlist-changed notification. func Topic(screenSlug string) string { return fmt.Sprintf("signage/screen/%s/playlist-changed", screenSlug) } // NotifyChanged schedules a publish for screenSlug. If another call for the // same screen arrives within debounceDuration, the timer is reset (debounce). // The method returns immediately; the actual publish happens in a goroutine. func (n *Notifier) NotifyChanged(screenSlug string) { if n.client == nil { return } n.mu.Lock() defer n.mu.Unlock() // Reset existing timer if present (debounce). if t, ok := n.timers[screenSlug]; ok { t.Stop() } n.timers[screenSlug] = time.AfterFunc(debounceDuration, func() { n.mu.Lock() delete(n.timers, screenSlug) n.mu.Unlock() n.publish(screenSlug) }) } func (n *Notifier) publish(screenSlug string) { topic := Topic(screenSlug) payload := []byte(fmt.Sprintf(`{"ts":%d}`, time.Now().UnixMilli())) token := n.client.Publish(topic, 0, false, payload) token.WaitTimeout(3 * time.Second) // Errors are silently dropped — the 60 s polling in the agent is the fallback. } // Close disconnects the MQTT client gracefully. func (n *Notifier) Close() { if n.client == nil { return } n.mu.Lock() for _, t := range n.timers { t.Stop() } n.timers = make(map[string]*time.Timer) n.mu.Unlock() n.client.Disconnect(250) }