package store import ( "context" "fmt" "strings" "time" ) // MediaRequest is one viewer's ask, as recorded. // // The status a card shows is not here: it is derived per read from the *arrs and the // library, for the reason the schema gives. The one exception is LastStatus, which is not // the card's status but the memory of it — the only way to notice that something changed. type MediaRequest struct { MediaType string `json:"mediaType"` ForeignID int `json:"foreignId"` Title string `json:"title"` Year int `json:"year,omitempty"` PosterURL string `json:"posterUrl,omitempty"` RequestedAt time.Time `json:"requestedAt"` // LastStatus is the state this request was in the last time anything looked, and it is // the only piece of request state that is stored. See the schema comment: an arrival is // a difference between two observations rather than a property of one, and the viewer is // told about it once. LastStatus string `json:"-"` } // OwnedMediaRequest is a stored ask with the person who made it, which the per-viewer read // does not need to carry because it was asked for by user. The ready sweep looks at the // whole household in one query, so there it is the point. type OwnedMediaRequest struct { MediaRequest UserID string } // RequestUsage is the operator-facing use of the request feature. A recorded request is // one that Memby has already handed to Radarr or Sonarr; their later download state is // deliberately not duplicated here. type RequestUsage struct { UserID string `json:"userId"` Requests int64 `json:"requests"` LastRequest time.Time `json:"lastRequest,omitempty"` } func (s *Store) RequestUsage(ctx context.Context) ([]RequestUsage, error) { rows, err := s.pool.Query(ctx, `SELECT emby_user_id, count(*), max(requested_at) FROM media_requests GROUP BY emby_user_id`) if err != nil { return nil, fmt.Errorf("store: request usage: %w", err) } defer rows.Close() out := []RequestUsage{} for rows.Next() { var value RequestUsage if err := rows.Scan(&value.UserID, &value.Requests, &value.LastRequest); err != nil { return nil, err } out = append(out, value) } return out, rows.Err() } // MediaRequestLimit caps what one viewer's page will read back. A household that has been // asking for things for two years should not turn the page into an unbounded query, and // nobody scrolls past the most recent hundred by remote. const MediaRequestLimit = 100 // SaveMediaRequest records an ask, or refreshes one already held. // // The repeat is deliberately an update rather than a no-op: asking again is how somebody // says they still want it, and the page is ordered by when they asked. func (s *Store) SaveMediaRequest(ctx context.Context, userID string, req MediaRequest) error { userID = strings.TrimSpace(userID) if userID == "" || req.ForeignID <= 0 { return fmt.Errorf("store: media request needs a user and a foreign id") } // last_status is deliberately absent from the UPDATE. Asking again is somebody saying // they still want it, not a reason to re-announce an arrival they were already told // about — and re-seeding it here would make a second press of Request the way to make // the gateway repeat itself. _, err := s.pool.Exec(ctx, ` INSERT INTO media_requests (emby_user_id, media_type, foreign_id, title, year, poster_url, last_status, requested_at) VALUES ($1, $2, $3, $4, $5, $6, $7, now()) ON CONFLICT (emby_user_id, media_type, foreign_id) DO UPDATE SET title = EXCLUDED.title, year = EXCLUDED.year, poster_url = EXCLUDED.poster_url, requested_at = now()`, userID, req.MediaType, req.ForeignID, req.Title, req.Year, req.PosterURL, req.LastStatus) if err != nil { return fmt.Errorf("store: save media request: %w", err) } return nil } // MediaRequests returns one viewer's asks, most recent first. func (s *Store) MediaRequests(ctx context.Context, userID string) ([]MediaRequest, error) { rows, err := s.pool.Query(ctx, ` SELECT media_type, foreign_id, title, year, poster_url, last_status, requested_at FROM media_requests WHERE emby_user_id = $1 ORDER BY requested_at DESC LIMIT $2`, strings.TrimSpace(userID), MediaRequestLimit) if err != nil { return nil, fmt.Errorf("store: read media requests: %w", err) } defer rows.Close() requests := []MediaRequest{} for rows.Next() { var req MediaRequest if err := rows.Scan( &req.MediaType, &req.ForeignID, &req.Title, &req.Year, &req.PosterURL, &req.LastStatus, &req.RequestedAt, ); err != nil { return nil, fmt.Errorf("store: scan media request: %w", err) } requests = append(requests, req) } return requests, rows.Err() } // DeleteMediaRequest removes a viewer's ask from their own page. // // It deliberately does not touch Radarr or Sonarr. The title may well have been downloaded // by now, and other people may have asked for it too — this only says the viewer no longer // wants it listed among theirs. func (s *Store) DeleteMediaRequest( ctx context.Context, userID, mediaType string, foreignID int, ) error { _, err := s.pool.Exec(ctx, ` DELETE FROM media_requests WHERE emby_user_id = $1 AND media_type = $2 AND foreign_id = $3`, strings.TrimSpace(userID), mediaType, foreignID) if err != nil { return fmt.Errorf("store: delete media request: %w", err) } return nil } // AllMediaRequests reads the whole household's asks, newest first, with the person attached. // // The per-viewer read above is what a page needs; this is what the ready sweep needs, and // the difference is worth one query rather than one per account: a household of six with // eighty requests between them is one read, and the sweep has to look at all of them anyway // because two people can be waiting for the same film. // // It is bounded like the per-viewer read. A sweep that fell behind on a household which had // been asking for things for two years must not become an unbounded query on a timer. func (s *Store) AllMediaRequests(ctx context.Context, limit int) ([]OwnedMediaRequest, error) { if limit <= 0 { limit = MediaRequestSweepLimit } rows, err := s.pool.Query(ctx, ` SELECT emby_user_id, media_type, foreign_id, title, year, poster_url, last_status, requested_at FROM media_requests ORDER BY requested_at DESC LIMIT $1`, limit) if err != nil { return nil, fmt.Errorf("store: read all media requests: %w", err) } defer rows.Close() requests := []OwnedMediaRequest{} for rows.Next() { var req OwnedMediaRequest if err := rows.Scan( &req.UserID, &req.MediaType, &req.ForeignID, &req.Title, &req.Year, &req.PosterURL, &req.LastStatus, &req.RequestedAt, ); err != nil { return nil, fmt.Errorf("store: scan media request: %w", err) } requests = append(requests, req) } return requests, rows.Err() } // MediaRequestSweepLimit caps what one pass of the ready sweep will look at. const MediaRequestSweepLimit = 500 // RequestReadySurfaceWindow is how long a freshly arrived request stays pinned to the front // of Continue Watching with a REQUEST READY tag if the viewer never plays it. Past this the // age-out clears it: an arrival nobody has acted on in a fortnight is no longer news, and a // pin that outstays its welcome is worse than one that lapses. const RequestReadySurfaceWindow = 14 * 24 * time.Hour // SurfacedReadyRequest is one arrival still worth pinning: which title, and the Emby item id // the card opens and dedupes against. type SurfacedReadyRequest struct { MediaType string ForeignID int Title string Year int PosterURL string ItemID string SurfacedAt time.Time } // MarkRequestReadySurfaced records that a request has become watchable and should be pinned. // // The guard is the whole of the idempotency: the ready sweep calls this on every observed // transition to "available", and only the first one — before anything has cleared it — takes // effect. A request cleared by playback or the age-out is never re-pinned here; a genuinely // new ask is a fresh row with both timestamps null. func (s *Store) MarkRequestReadySurfaced( ctx context.Context, userID, mediaType string, foreignID int, itemID string, ) error { itemID = strings.TrimSpace(itemID) if itemID == "" { return fmt.Errorf("store: request ready needs an item id") } _, err := s.pool.Exec(ctx, ` UPDATE media_requests SET ready_surfaced_at = now(), ready_item_id = $4 WHERE emby_user_id = $1 AND media_type = $2 AND foreign_id = $3 AND ready_surfaced_at IS NULL AND ready_cleared_at IS NULL`, strings.TrimSpace(userID), mediaType, foreignID, itemID) if err != nil { return fmt.Errorf("store: mark request ready surfaced: %w", err) } return nil } // SurfacedReadyRequests reads one viewer's still-pinned arrivals, newest first. // // The window and the cleared check are both in the query so the caller never has to think // about either: a row past RequestReadySurfaceWindow, or one playback has cleared, simply // does not come back. This is the read Home makes beside its Emby fan-out. func (s *Store) SurfacedReadyRequests( ctx context.Context, userID string, ) ([]SurfacedReadyRequest, error) { cutoff := time.Now().Add(-RequestReadySurfaceWindow) rows, err := s.pool.Query(ctx, ` SELECT media_type, foreign_id, title, year, poster_url, ready_item_id, ready_surfaced_at FROM media_requests WHERE emby_user_id = $1 AND ready_surfaced_at IS NOT NULL AND ready_cleared_at IS NULL AND ready_surfaced_at > $2 AND ready_item_id <> '' ORDER BY ready_surfaced_at DESC`, strings.TrimSpace(userID), cutoff) if err != nil { return nil, fmt.Errorf("store: read surfaced ready requests: %w", err) } defer rows.Close() out := []SurfacedReadyRequest{} for rows.Next() { var req SurfacedReadyRequest if err := rows.Scan( &req.MediaType, &req.ForeignID, &req.Title, &req.Year, &req.PosterURL, &req.ItemID, &req.SurfacedAt, ); err != nil { return nil, fmt.Errorf("store: scan surfaced ready request: %w", err) } out = append(out, req) } return out, rows.Err() } // ClearRequestReadySurfaced retires the pin for whichever of a viewer's requests point at // the given Emby item id — the film itself, or the series an episode belongs to. It returns // how many rows it touched so the caller can skip a cache invalidation that would change // nothing. func (s *Store) ClearRequestReadySurfaced( ctx context.Context, userID string, itemIDs ...string, ) (int64, error) { ids := make([]string, 0, len(itemIDs)) for _, id := range itemIDs { if id = strings.TrimSpace(id); id != "" { ids = append(ids, id) } } if len(ids) == 0 { return 0, nil } tag, err := s.pool.Exec(ctx, ` UPDATE media_requests SET ready_cleared_at = now() WHERE emby_user_id = $1 AND ready_item_id = ANY($2) AND ready_surfaced_at IS NOT NULL AND ready_cleared_at IS NULL`, strings.TrimSpace(userID), ids) if err != nil { return 0, fmt.Errorf("store: clear request ready surfaced: %w", err) } return tag.RowsAffected(), nil } // ExpireStaleReadyRequests clears every pin past RequestReadySurfaceWindow across the whole // household in one statement, and returns the users whose Home cache is now stale. The ready // sweep runs this so SurfacedReadyRequests never has to lean on its own window clause to // hide a row that should have been retired days ago. func (s *Store) ExpireStaleReadyRequests(ctx context.Context) ([]string, error) { cutoff := time.Now().Add(-RequestReadySurfaceWindow) rows, err := s.pool.Query(ctx, ` UPDATE media_requests SET ready_cleared_at = now() WHERE ready_surfaced_at IS NOT NULL AND ready_cleared_at IS NULL AND ready_surfaced_at <= $1 RETURNING emby_user_id`, cutoff) if err != nil { return nil, fmt.Errorf("store: expire stale ready requests: %w", err) } defer rows.Close() seen := map[string]bool{} users := []string{} for rows.Next() { var userID string if err := rows.Scan(&userID); err != nil { return nil, err } if !seen[userID] { seen[userID] = true users = append(users, userID) } } return users, rows.Err() } // SetMediaRequestStatus records what a request was last seen doing. // // Written only when the state actually moved, so a sweep over a household where nothing has // changed — which is almost every sweep — costs no writes at all. func (s *Store) SetMediaRequestStatus( ctx context.Context, userID, mediaType string, foreignID int, status string, ) error { _, err := s.pool.Exec(ctx, ` UPDATE media_requests SET last_status = $4 WHERE emby_user_id = $1 AND media_type = $2 AND foreign_id = $3`, strings.TrimSpace(userID), mediaType, foreignID, status) if err != nil { return fmt.Errorf("store: set media request status: %w", err) } return nil }