memby/server/internal/api/alerts.go

package api

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"time"
)

// Kinds are part of the wire contract: an unknown kind is rendered with the generic
// banner rather than dropped, so adding one never needs an app release.
const (
	alertKindSonarrAired  = "sonarr-aired"
	alertKindRadarrImport = "radarr-import"
	alertKindLibrarySync  = "library-updated"
	alertKindServerDown   = "server-unreachable"
	alertKindServerUp     = "server-restored"
	alertKindDeploying    = "server-deploying"

	// A TV shows one banner at a time; more than a few queued up is noise, not news.
	maxAlerts = 3
)

// clientAlert is a short-lived, informational nudge delivered on the /v1/status poll —
// the only channel the app already listens to while it is open. It carries no action:
// the client slides it in, shows it for a few seconds and forgets it.
type clientAlert struct {
	ID   string `json:"id"`
	Kind string `json:"kind"`
	// Label is the banner's eyebrow ("JUST AIRED", "NEW MOVIE ADDED"). It travels with
	// the alert so a new kind of news reads correctly on an app that predates it; a
	// client seeing no label falls back to its own wording.
	Label    string `json:"label,omitempty"`
	Title    string `json:"title"`
	Message  string `json:"message"`
	ItemID   string `json:"itemId,omitempty"`
	ImageTag string `json:"imageTag,omitempty"`
	// AiredAt is when the news happened — broadcast time for an episode, import time
	// for a movie. It orders the merged list; the client does not read it.
	AiredAt string `json:"airedAt,omitempty"`
}

// mergeAlerts interleaves the alert sources newest-first and trims to what a TV can
// usefully show. Each source is already sorted, but a movie imported ten minutes ago
// should still come before an episode that aired two hours back.
func mergeAlerts(groups ...[]clientAlert) []clientAlert {
	merged := make([]clientAlert, 0, maxAlerts)
	for _, group := range groups {
		merged = append(merged, group...)
	}
	sort.SliceStable(merged, func(i, j int) bool {
		return alertTime(merged[i]).After(alertTime(merged[j]))
	})
	if len(merged) > maxAlerts {
		merged = merged[:maxAlerts]
	}
	return merged
}

// alertTime is zero for an alert with no timestamp, which sorts it last rather than
// dropping it — an unorderable alert is still news.
func alertTime(alert clientAlert) time.Time {
	when, err := time.Parse(time.RFC3339, alert.AiredAt)
	if err != nil {
		return time.Time{}
	}
	return when
}

// Alerts that are *events* — a film imported, the library refreshed, Emby stopping and
// starting to answer — are published into one shared Redis list and served from it,
// rather than derived on each poll the way the Sonarr calendar is.
//
// A list with expiring entries rather than a push: the gateway holds no connection to a
// television, so there is nothing to push to. A window is what lets a set that was off
// or in the screensaver when the event happened still hear about it, and each TV dedupes
// by alert id so nobody is told twice.
const publishedAlertsCacheKey = "alerts:events:v1"

// More than this stored at once means something is generating events in bulk, and a
// viewer does not want twenty banners about it — the newest few are the news.
const maxStoredAlerts = 8

// storedAlert is a clientAlert with the moment it stops being news. Expiry is held with
// the record rather than as a Redis TTL because they share one key: the list outlives
// any single entry in it.
type storedAlert struct {
	Alert     clientAlert `json:"alert"`
	ExpiresAt time.Time   `json:"expiresAt"`
}

// publishAlert makes one event visible to every signed-in TV for window.
//
// Failures are logged and swallowed. A missed banner is not worth failing the thing that
// produced it — an import, a library sync, a health probe — none of which the viewer
// would want retried for the sake of a notice.
func (s *Server) publishAlert(ctx context.Context, alert clientAlert, window time.Duration) {
	if window <= 0 || alert.ID == "" || s.cache == nil {
		return
	}
	// Read-modify-write on one key, so producers running on their own schedules need
	// serialising against each other. They are rare enough that a mutex is the whole
	// answer; this instance is the only writer.
	s.alertMu.Lock()
	defer s.alertMu.Unlock()

	now := time.Now().UTC()
	stored := appendAlert(s.storedAlerts(ctx), alert, now.Add(window), now)
	body, err := json.Marshal(stored)
	if err != nil {
		s.loggerFor(ctx).Warn("alert encode failed", "error", err)
		return
	}
	// The key's own TTL is a floor sweep for a gateway that stops producing events; the
	// per-entry expiry is what actually decides what a client sees.
	if err := s.cache.Set(ctx, publishedAlertsCacheKey, body, window*2); err != nil {
		s.loggerFor(ctx).Warn("alert store failed", "error", err)
	}
}

// appendAlert is the pure half of publishing: prune what has expired, replace any earlier
// copy of the same alert rather than stacking duplicates, keep the newest few.
func appendAlert(
	existing []storedAlert, alert clientAlert, expiresAt, now time.Time,
) []storedAlert {
	kept := make([]storedAlert, 0, len(existing)+1)
	for _, entry := range existing {
		if entry.Alert.ID == alert.ID || !entry.ExpiresAt.After(now) {
			continue
		}
		kept = append(kept, entry)
	}
	kept = append(kept, storedAlert{Alert: alert, ExpiresAt: expiresAt})
	sort.SliceStable(kept, func(i, j int) bool {
		return alertTime(kept[i].Alert).After(alertTime(kept[j].Alert))
	})
	if len(kept) > maxStoredAlerts {
		kept = kept[:maxStoredAlerts]
	}
	return kept
}

// publishedAlerts is the read side, called on every /v1/status poll.
func (s *Server) publishedAlerts(ctx context.Context) []clientAlert {
	return liveAlerts(s.storedAlerts(ctx), time.Now().UTC())
}

func (s *Server) storedAlerts(ctx context.Context) []storedAlert {
	// Redis is where these live, so no Redis means no news — never a failed status poll.
	// The status endpoint is what a client falls back on when everything else is broken;
	// it must not acquire a hard dependency for the sake of a banner.
	if s.cache == nil {
		return nil
	}
	raw, err := s.cache.Get(ctx, publishedAlertsCacheKey)
	if err != nil {
		return nil
	}
	var stored []storedAlert
	if json.Unmarshal(raw, &stored) != nil {
		return nil
	}
	return stored
}

// liveAlerts drops entries whose window has closed. Expiry is applied on read as well as
// on write, so an alert stops being offered on time even if no event follows it.
func liveAlerts(stored []storedAlert, now time.Time) []clientAlert {
	alerts := make([]clientAlert, 0, len(stored))
	for _, entry := range stored {
		if !entry.ExpiresAt.After(now) {
			continue
		}
		alerts = append(alerts, entry.Alert)
	}
	return alerts
}

// sonarrAiredAlerts reads the calendar through the same cache the five-day schedule row
// uses, so polling clients never cost a Sonarr request of their own.
func (s *Server) sonarrAiredAlerts(ctx context.Context) []clientAlert {
	if s.sonarr == nil || s.cfg.SonarrAlertWindow <= 0 {
		return nil
	}
	row, err := s.sonarrAiringTodayRow(ctx)
	if err != nil {
		s.loggerFor(ctx).Warn("sonarr alerts unavailable", "error", err)
		return nil
	}
	if row == nil {
		return nil
	}
	items := make([]sonarrScheduleItem, 0, len(row.Items))
	for _, raw := range row.Items {
		var item sonarrScheduleItem
		if json.Unmarshal(raw, &item) != nil {
			continue
		}
		items = append(items, item)
	}
	location := s.cfg.SonarrLocation
	if location == nil {
		location = time.Local
	}
	return buildSonarrAlerts(items, time.Now().In(location), s.cfg.SonarrAlertWindow)
}

// buildSonarrAlerts announces episodes that have aired but are not in Emby yet — the
// gap the viewer would otherwise experience as "it's Tuesday, so why isn't it there".
// Anything already downloaded is deliberately silent: it is on the home screen, which
// says it better than a banner would.
func buildSonarrAlerts(items []sonarrScheduleItem, now time.Time, window time.Duration) []clientAlert {
	if window <= 0 {
		return nil
	}
	type dated struct {
		alert clientAlert
		airs  time.Time
	}
	found := make([]dated, 0, len(items))
	for _, item := range items {
		if item.MembyAirsAt == "" {
			continue
		}
		airsAt, err := time.Parse(time.RFC3339, item.MembyAirsAt)
		if err != nil {
			continue
		}
		// Not yet aired, or aired so long ago that saying so is no longer news.
		if airsAt.After(now) || now.Sub(airsAt) > window {
			continue
		}
		message, ok := airedMessage(item, airsAt)
		if !ok {
			continue
		}
		found = append(found, dated{
			alert: clientAlert{
				ID:       item.ID + ":aired",
				Kind:     alertKindSonarrAired,
				Label:    "JUST AIRED",
				Title:    item.Name,
				Message:  message,
				ItemID:   item.ID,
				ImageTag: item.ImageTags["Primary"],
				AiredAt:  item.MembyAirsAt,
			},
			airs: airsAt,
		})
	}

	// Newest first: if several aired inside the window, the most recent is the one the
	// viewer is most likely to be waiting on.
	sort.SliceStable(found, func(i, j int) bool { return found[i].airs.After(found[j].airs) })
	if len(found) > maxAlerts {
		found = found[:maxAlerts]
	}
	alerts := make([]clientAlert, 0, len(found))
	for _, entry := range found {
		alerts = append(alerts, entry.alert)
	}
	return alerts
}

func airedMessage(item sonarrScheduleItem, airsAt time.Time) (string, bool) {
	episode := item.MembyEpisodeCode
	if item.MembyEpisodeTitle != "" {
		episode = fmt.Sprintf("%s — %s", episode, item.MembyEpisodeTitle)
	}
	switch item.MembyAvailability {
	case "downloading":
		return fmt.Sprintf("%s aired at %s and is downloading now.", episode, airsAt.Format("3:04 PM")), true
	case "awaiting":
		return fmt.Sprintf("%s aired at %s and will be in Emby soon.", episode, airsAt.Format("3:04 PM")), true
	default:
		// available (already watchable) and unmonitored (never coming) both have
		// nothing useful to announce.
		return "", false
	}
}