package store import ( "context" "fmt" "strings" "time" ) type UserShow struct { ItemID string `json:"itemId"` Title string `json:"title"` Year *int `json:"year,omitempty"` ImageTag string `json:"imageTag,omitempty"` AddedAt time.Time `json:"addedAt"` } type NotificationPreferences struct { Enabled bool `json:"enabled"` ShowReturnAlerts bool `json:"showReturnAlerts"` SonarrAlerts bool `json:"sonarrAlerts"` RadarrAlerts bool `json:"radarrAlerts"` UpdateAlerts bool `json:"updateAlerts"` LibraryAlerts bool `json:"libraryAlerts"` SystemAlerts bool `json:"systemAlerts"` // WatchTimeDigest is the weekly viewing summary. It is its own switch rather than part // of SystemAlerts because it is the only notification here that is about the viewer // rather than about the library or the server, and somebody who wants to be told a show // was cancelled may well not want to be told how long they spent watching it. WatchTimeDigest bool `json:"watchTimeDigest"` LeadDays int `json:"leadDays"` } func DefaultNotificationPreferences() NotificationPreferences { return NotificationPreferences{ Enabled: true, ShowReturnAlerts: true, SonarrAlerts: true, RadarrAlerts: true, UpdateAlerts: true, LibraryAlerts: true, SystemAlerts: true, WatchTimeDigest: true, LeadDays: 7, } } type UserNotification struct { ID int64 `json:"id"` Kind string `json:"kind"` ItemID string `json:"itemId,omitempty"` Title string `json:"title"` Message string `json:"message"` EventAt *time.Time `json:"eventAt,omitempty"` CreatedAt time.Time `json:"createdAt"` ReadAt *time.Time `json:"readAt,omitempty"` } // SonarrSeriesStatus is one daily observation. SeriesKey prefers Sonarr's stable TVDB id; // the local Sonarr id is retained for diagnosis and as a fallback when TVDB has no answer. type SonarrSeriesStatus struct { SeriesKey string SonarrSeriesID int TVDBID int Title string Year int Status string ObservedAt time.Time } type SonarrSeriesStatusChange struct { HistoryID int64 PreviousStatus string Current SonarrSeriesStatus } // LatestSonarrSeriesStatuses returns Sonarr's most recently observed lifecycle for each // TVDB id. The history table is change-only, so the newest row is also the current value. func (s *Store) LatestSonarrSeriesStatuses( ctx context.Context, tvdbIDs []int, ) (map[int]string, error) { statuses := map[int]string{} if len(tvdbIDs) == 0 { return statuses, nil } rows, err := s.pool.Query(ctx, ` SELECT DISTINCT ON (tvdb_id) tvdb_id, status FROM sonarr_series_status_history WHERE tvdb_id = ANY($1) AND tvdb_id > 0 ORDER BY tvdb_id, observed_at DESC, id DESC`, tvdbIDs) if err != nil { return nil, fmt.Errorf("store: read latest Sonarr series statuses: %w", err) } defer rows.Close() for rows.Next() { var tvdbID int var status string if err := rows.Scan(&tvdbID, &status); err != nil { return nil, fmt.Errorf("store: scan latest Sonarr series status: %w", err) } statuses[tvdbID] = status } return statuses, rows.Err() } // SonarrSeriesStatusRevision is an opaque, monotonic version of the stored lifecycle // catalogue. It changes whenever a scan records a first sighting or a transition. func (s *Store) SonarrSeriesStatusRevision(ctx context.Context) (int64, error) { var revision int64 if err := s.pool.QueryRow(ctx, `SELECT COALESCE(max(id), 0) FROM sonarr_series_status_history`).Scan(&revision); err != nil { return 0, fmt.Errorf("store: read Sonarr series status revision: %w", err) } return revision, nil } func (s *Store) SaveUserShow(ctx context.Context, userID string, show UserShow) error { _, err := s.pool.Exec(ctx, ` INSERT INTO user_shows (emby_user_id, item_id, title, year, image_tag) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (emby_user_id, item_id) DO UPDATE SET title = EXCLUDED.title, year = EXCLUDED.year, image_tag = EXCLUDED.image_tag`, userID, show.ItemID, show.Title, show.Year, show.ImageTag) return err } // SaveUserShowIfAbsent is the auto-follow path. Manual saves intentionally refresh // metadata, while playback must only announce a show when it actually added it. func (s *Store) SaveUserShowIfAbsent(ctx context.Context, userID string, show UserShow) (bool, error) { result, err := s.pool.Exec(ctx, ` INSERT INTO user_shows (emby_user_id, item_id, title, year, image_tag) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (emby_user_id, item_id) DO NOTHING`, userID, show.ItemID, show.Title, show.Year, show.ImageTag) if err != nil { return false, err } return result.RowsAffected() == 1, nil } func (s *Store) DeleteUserShow(ctx context.Context, userID, itemID string) error { _, err := s.pool.Exec(ctx, ` WITH removed AS ( DELETE FROM user_shows WHERE emby_user_id = $1 AND item_id = $2 RETURNING item_id ) UPDATE user_notifications SET dismissed_at = now() WHERE emby_user_id = $1 AND item_id IN (SELECT item_id FROM removed) AND dismissed_at IS NULL`, userID, itemID) return err } func (s *Store) UserShows(ctx context.Context, userID string) ([]UserShow, error) { rows, err := s.pool.Query(ctx, ` SELECT item_id, title, year, image_tag, added_at FROM user_shows WHERE emby_user_id = $1 ORDER BY added_at, lower(title)`, userID) if err != nil { return nil, fmt.Errorf("store: list user shows: %w", err) } defer rows.Close() shows := []UserShow{} for rows.Next() { var show UserShow if err := rows.Scan(&show.ItemID, &show.Title, &show.Year, &show.ImageTag, &show.AddedAt); err != nil { return nil, fmt.Errorf("store: scan user show: %w", err) } shows = append(shows, show) } return shows, rows.Err() } // RecordSonarrSeriesStatuses appends first sightings and changes. The first complete scan // seeds a baseline; a first sighting on later scans is returned with an empty previous // status so the lifecycle scanner can announce a show newly added to Sonarr. func (s *Store) RecordSonarrSeriesStatuses( ctx context.Context, observations []SonarrSeriesStatus, ) ([]SonarrSeriesStatusChange, error) { tx, err := s.pool.Begin(ctx) if err != nil { return nil, fmt.Errorf("store: begin Sonarr status history: %w", err) } defer tx.Rollback(ctx) var seeded bool if err := tx.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM sonarr_lifecycle_scan_state)`).Scan(&seeded); err != nil { return nil, fmt.Errorf("store: read Sonarr lifecycle seed: %w", err) } rows, err := tx.Query(ctx, ` SELECT DISTINCT ON (series_key) series_key, status FROM sonarr_series_status_history ORDER BY series_key, observed_at DESC, id DESC`) if err != nil { return nil, fmt.Errorf("store: read Sonarr status history: %w", err) } previous := map[string]string{} for rows.Next() { var key, status string if err := rows.Scan(&key, &status); err != nil { rows.Close() return nil, fmt.Errorf("store: scan Sonarr status history: %w", err) } previous[key] = status } if err := rows.Err(); err != nil { rows.Close() return nil, fmt.Errorf("store: iterate Sonarr status history: %w", err) } rows.Close() changes := []SonarrSeriesStatusChange{} seen := map[string]bool{} for _, observation := range observations { observation.SeriesKey = strings.TrimSpace(observation.SeriesKey) observation.Title = strings.TrimSpace(observation.Title) observation.Status = strings.ToLower(strings.TrimSpace(observation.Status)) if observation.SeriesKey == "" || observation.Title == "" || observation.Status == "" || seen[observation.SeriesKey] { continue } seen[observation.SeriesKey] = true prior, known := previous[observation.SeriesKey] if known && strings.EqualFold(prior, observation.Status) { continue } if observation.ObservedAt.IsZero() { observation.ObservedAt = time.Now() } var historyID int64 err := tx.QueryRow(ctx, ` INSERT INTO sonarr_series_status_history (series_key, sonarr_series_id, tvdb_id, title, year, status, observed_at) VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING id`, observation.SeriesKey, observation.SonarrSeriesID, observation.TVDBID, observation.Title, observation.Year, observation.Status, observation.ObservedAt, ).Scan(&historyID) if err != nil { return nil, fmt.Errorf("store: append Sonarr status history: %w", err) } if known || seeded { changes = append(changes, SonarrSeriesStatusChange{ HistoryID: historyID, PreviousStatus: prior, Current: observation, }) } previous[observation.SeriesKey] = observation.Status } if !seeded { if _, err := tx.Exec(ctx, `INSERT INTO sonarr_lifecycle_scan_state (singleton) VALUES (true) ON CONFLICT DO NOTHING`); err != nil { return nil, fmt.Errorf("store: seed Sonarr lifecycle scan: %w", err) } } if err := tx.Commit(ctx); err != nil { return nil, fmt.Errorf("store: commit Sonarr status history: %w", err) } return changes, nil } func (s *Store) NotificationPreferences(ctx context.Context, userID string) (NotificationPreferences, error) { prefs := DefaultNotificationPreferences() err := s.pool.QueryRow(ctx, ` SELECT enabled, show_return_alerts, sonarr_alerts, radarr_alerts, update_alerts, library_alerts, system_alerts, watch_time_digest, lead_days FROM user_notification_preferences WHERE emby_user_id = $1`, userID). Scan(&prefs.Enabled, &prefs.ShowReturnAlerts, &prefs.SonarrAlerts, &prefs.RadarrAlerts, &prefs.UpdateAlerts, &prefs.LibraryAlerts, &prefs.SystemAlerts, &prefs.WatchTimeDigest, &prefs.LeadDays) if err != nil && !isNoRows(err) { return prefs, fmt.Errorf("store: notification preferences: %w", err) } return prefs, nil } func (s *Store) SetNotificationPreferences( ctx context.Context, userID string, prefs NotificationPreferences, ) error { if prefs.LeadDays < 1 { prefs.LeadDays = 1 } if prefs.LeadDays > 30 { prefs.LeadDays = 30 } _, err := s.pool.Exec(ctx, ` INSERT INTO user_notification_preferences (emby_user_id, enabled, show_return_alerts, sonarr_alerts, radarr_alerts, update_alerts, library_alerts, system_alerts, watch_time_digest, lead_days) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) ON CONFLICT (emby_user_id) DO UPDATE SET enabled = EXCLUDED.enabled, show_return_alerts = EXCLUDED.show_return_alerts, sonarr_alerts = EXCLUDED.sonarr_alerts, radarr_alerts = EXCLUDED.radarr_alerts, update_alerts = EXCLUDED.update_alerts, library_alerts = EXCLUDED.library_alerts, system_alerts = EXCLUDED.system_alerts, watch_time_digest = EXCLUDED.watch_time_digest, lead_days = EXCLUDED.lead_days, updated_at = now()`, userID, prefs.Enabled, prefs.ShowReturnAlerts, prefs.SonarrAlerts, prefs.RadarrAlerts, prefs.UpdateAlerts, prefs.LibraryAlerts, prefs.SystemAlerts, prefs.WatchTimeDigest, prefs.LeadDays) return err } func (s *Store) AllNotificationPreferences(ctx context.Context) (map[string]NotificationPreferences, error) { rows, err := s.pool.Query(ctx, ` SELECT emby_user_id, enabled, show_return_alerts, sonarr_alerts, radarr_alerts, update_alerts, library_alerts, system_alerts, watch_time_digest, lead_days FROM user_notification_preferences`) if err != nil { return nil, fmt.Errorf("store: list notification preferences: %w", err) } defer rows.Close() result := map[string]NotificationPreferences{} for rows.Next() { prefs := DefaultNotificationPreferences() var userID string if err := rows.Scan(&userID, &prefs.Enabled, &prefs.ShowReturnAlerts, &prefs.SonarrAlerts, &prefs.RadarrAlerts, &prefs.UpdateAlerts, &prefs.LibraryAlerts, &prefs.SystemAlerts, &prefs.WatchTimeDigest, &prefs.LeadDays); err != nil { return nil, fmt.Errorf("store: scan notification preferences: %w", err) } result[userID] = prefs } return result, rows.Err() } // UpsertNotification writes one notification into a viewer's own list. // // It reports whether a row was actually inserted, which is what separates the two answers // the source key produces: a genuine delivery, and a repeat of one already sitting in // somebody's list. Both are ordinary — the digest job runs hourly and re-sends the same // weekly key all evening on purpose — but the notification log has to be able to tell them // apart, or every catch-up run would read as a second summary nobody received. func (s *Store) UpsertNotification( ctx context.Context, userID, sourceKey, kind, itemID, title, message string, eventAt *time.Time, ) (bool, error) { tag, err := s.pool.Exec(ctx, ` INSERT INTO user_notifications (emby_user_id, source_key, kind, item_id, title, message, event_at) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (emby_user_id, source_key) DO NOTHING`, userID, sourceKey, kind, itemID, title, message, eventAt) if err != nil { return false, err } return tag.RowsAffected() > 0, nil } func (s *Store) UserNotifications(ctx context.Context, userID string) ([]UserNotification, error) { rows, err := s.pool.Query(ctx, ` SELECT id, kind, item_id, title, message, event_at, created_at, read_at FROM user_notifications WHERE emby_user_id = $1 AND dismissed_at IS NULL ORDER BY created_at DESC LIMIT 100`, userID) if err != nil { return nil, fmt.Errorf("store: list notifications: %w", err) } defer rows.Close() notifications := []UserNotification{} for rows.Next() { var notification UserNotification if err := rows.Scan( ¬ification.ID, ¬ification.Kind, ¬ification.ItemID, ¬ification.Title, ¬ification.Message, ¬ification.EventAt, ¬ification.CreatedAt, ¬ification.ReadAt, ); err != nil { return nil, fmt.Errorf("store: scan notification: %w", err) } notifications = append(notifications, notification) } return notifications, rows.Err() } func (s *Store) MarkNotificationRead(ctx context.Context, userID string, id int64) error { _, err := s.pool.Exec(ctx, ` UPDATE user_notifications SET read_at = COALESCE(read_at, now()) WHERE id = $1 AND emby_user_id = $2`, id, userID) return err } // MarkNotificationUnread puts a notification back to new. // // The counterpart to MarkNotificationRead, and deliberately a plain assignment rather than // that one's COALESCE: read is sticky because it is set by merely looking at a row, so a // second glance must not move the timestamp, while unread is only ever the viewer saying so // and means exactly one thing. func (s *Store) MarkNotificationUnread(ctx context.Context, userID string, id int64) error { _, err := s.pool.Exec(ctx, ` UPDATE user_notifications SET read_at = NULL WHERE id = $1 AND emby_user_id = $2`, id, userID) return err } func (s *Store) DismissNotification(ctx context.Context, userID string, id int64) error { _, err := s.pool.Exec(ctx, ` UPDATE user_notifications SET dismissed_at = now() WHERE id = $1 AND emby_user_id = $2`, id, userID) return err } // DismissNotifications clears several of one viewer's notifications at once and reports how // many rows it actually took. // // The count is the whole reason this is a route rather than the client's loop: "cleared 7" // is what the log line and the activity record are worth reading for, and a television // counting the requests it made would be counting what it asked for rather than what // happened — a row somebody dismissed on another set in the meantime is one this must not // claim. Already-dismissed rows are excluded rather than re-stamped, so a repeated press // honestly reports nothing left to clear. func (s *Store) DismissNotifications( ctx context.Context, userID string, ids []int64, ) (int, error) { if len(ids) == 0 { return 0, nil } tag, err := s.pool.Exec(ctx, ` UPDATE user_notifications SET dismissed_at = now() WHERE emby_user_id = $1 AND id = ANY($2::bigint[]) AND dismissed_at IS NULL`, userID, ids) if err != nil { return 0, fmt.Errorf("store: dismiss notifications: %w", err) } return int(tag.RowsAffected()), nil }