0.2.82
This commit is contained in:
@@ -44,6 +44,10 @@ func (s *Server) adminRoutes() http.Handler {
|
||||
mux.Handle("POST /admin/api/accounts/{userID}/preferences/revisions/{revision}/restore",
|
||||
s.adminAuth(s.handleAdminRestorePreferences))
|
||||
mux.Handle("DELETE /admin/api/accounts/{userID}/recommendations", s.adminAuth(s.handleAdminResetRecommendations))
|
||||
mux.Handle("GET /admin/api/accounts/{userID}/viewers", s.adminAuth(s.handleAdminViewers))
|
||||
mux.Handle("POST /admin/api/accounts/{userID}/viewers", s.adminAuth(s.handleAdminViewers))
|
||||
mux.Handle("PUT /admin/api/accounts/{userID}/viewers/{viewerID}", s.adminAuth(s.handleAdminViewer))
|
||||
mux.Handle("DELETE /admin/api/accounts/{userID}/viewers/{viewerID}", s.adminAuth(s.handleAdminViewer))
|
||||
mux.Handle("PUT /admin/api/accounts/{userID}/recommendations/prompt", s.adminAuth(s.handleAdminPromptRecommendations))
|
||||
mux.Handle("PUT /admin/api/accounts/{userID}/themes", s.adminAuth(s.handleAdminUserThemes))
|
||||
mux.Handle("GET /admin/api/recommendations", s.adminAuth(s.handleAdminRecommendations))
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"github.com/ponzischeme89/memby/server/internal/store"
|
||||
)
|
||||
|
||||
// The people under one account, from the operator's side.
|
||||
//
|
||||
// It lives on the **account page** rather than on a rail entry of its own, which is the
|
||||
// whole of the design decision: a viewer only exists under an account, and a top-level
|
||||
// page would have to begin by asking which account is being talked about — a question the
|
||||
// page an operator reached this from has already answered. It is the arrangement the
|
||||
// per-account preference editor and the device list already take.
|
||||
//
|
||||
// A television can now do all of this for itself, so this is the operator's copy rather
|
||||
// than the only way in: what it is for is a household that has asked for help over the
|
||||
// phone, and the case a remote genuinely cannot reach — a viewer created on a set that has
|
||||
// since been unplugged.
|
||||
|
||||
type adminViewersResponse struct {
|
||||
Viewers []store.Viewer `json:"viewers"`
|
||||
// Whether the household's own switch is on. The page says so rather than quietly
|
||||
// offering controls whose effect nothing on any television would show: an operator who
|
||||
// has switched viewers off and then adds one has done something that looks like it
|
||||
// worked and did nothing.
|
||||
Enabled bool `json:"enabled"`
|
||||
// What the gateway will accept, so the console can stop offering Add at the same point
|
||||
// the television does rather than discovering the limit by being refused.
|
||||
MaxShadowViewers int `json:"maxShadowViewers"`
|
||||
}
|
||||
|
||||
func (s *Server) handleAdminViewers(w http.ResponseWriter, r *http.Request) {
|
||||
userID := strings.TrimSpace(r.PathValue("userID"))
|
||||
if userID == "" {
|
||||
writeError(w, http.StatusBadRequest, "user is required")
|
||||
return
|
||||
}
|
||||
if r.Method == http.MethodPost {
|
||||
s.handleAdminCreateViewer(w, r, userID)
|
||||
return
|
||||
}
|
||||
// The username is only used to name a main viewer that does not exist yet, and an
|
||||
// operator is not the right person to be naming somebody — an account that has never
|
||||
// had a request made against it gets the placeholder, and the television replaces it
|
||||
// with the real Emby name on its first sign-in.
|
||||
viewers, err := s.store.Viewers(r.Context(), userID, "")
|
||||
if err != nil {
|
||||
writeError(w, http.StatusInternalServerError, "could not load viewers")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, adminViewersResponse{
|
||||
Viewers: viewers,
|
||||
Enabled: s.viewersEnabled(r.Context()),
|
||||
MaxShadowViewers: store.MaxShadowViewers,
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Server) handleAdminCreateViewer(w http.ResponseWriter, r *http.Request, userID string) {
|
||||
var req viewerRequest
|
||||
if json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<10)).Decode(&req) != nil {
|
||||
writeError(w, http.StatusBadRequest, "malformed request body")
|
||||
return
|
||||
}
|
||||
if strings.TrimSpace(req.Name) == "" || len([]rune(req.Name)) > 40 {
|
||||
writeError(w, http.StatusBadRequest, "a name of up to 40 characters is required")
|
||||
return
|
||||
}
|
||||
viewer, err := s.store.CreateShadowViewer(r.Context(), userID, req.Name, req.ShortName, req.Colour)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadRequest, err.Error())
|
||||
return
|
||||
}
|
||||
// The televisions hold a cached list for viewerListTTL, so the write clears it here for
|
||||
// the same reason it does on the client-facing route: a person added from the console
|
||||
// must be pickable on the next request rather than at the end of the window.
|
||||
s.forgetViewers(userID)
|
||||
s.loggerFor(r.Context()).Info("viewer added by operator",
|
||||
"account", userID, "viewer", viewer.ID, "name", viewer.Name)
|
||||
writeJSON(w, http.StatusOK, viewer)
|
||||
}
|
||||
|
||||
func (s *Server) handleAdminViewer(w http.ResponseWriter, r *http.Request) {
|
||||
userID := strings.TrimSpace(r.PathValue("userID"))
|
||||
viewerID := strings.TrimSpace(r.PathValue("viewerID"))
|
||||
if userID == "" || viewerID == "" {
|
||||
writeError(w, http.StatusBadRequest, "user and viewer are required")
|
||||
return
|
||||
}
|
||||
|
||||
if r.Method == http.MethodDelete {
|
||||
if err := s.store.DeleteShadowViewer(r.Context(), userID, viewerID); err != nil {
|
||||
if errors.Is(err, store.ErrViewerNotFound) {
|
||||
writeError(w, http.StatusNotFound, "no such viewer")
|
||||
return
|
||||
}
|
||||
writeError(w, http.StatusInternalServerError, "could not remove that viewer")
|
||||
return
|
||||
}
|
||||
s.forgetViewers(userID)
|
||||
// Everything cached under this viewer's own key is now about nobody.
|
||||
if err := s.cache.InvalidateUser(r.Context(), viewerID); err != nil {
|
||||
s.loggerFor(r.Context()).Warn("cache invalidation failed", "error", err)
|
||||
}
|
||||
s.loggerFor(r.Context()).Info("viewer removed by operator",
|
||||
"account", userID, "viewer", viewerID)
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
return
|
||||
}
|
||||
|
||||
var req viewerRequest
|
||||
if json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<10)).Decode(&req) != nil {
|
||||
writeError(w, http.StatusBadRequest, "malformed request body")
|
||||
return
|
||||
}
|
||||
viewer, err := s.store.UpdateShadowViewer(
|
||||
r.Context(), userID, viewerID, req.Name, req.ShortName, req.Colour,
|
||||
)
|
||||
if err != nil {
|
||||
if errors.Is(err, store.ErrViewerNotFound) {
|
||||
// The main viewer lands here too, and that is the honest answer: its name is
|
||||
// the Emby account's, so as a *shadow* viewer to rename it does not exist.
|
||||
writeError(w, http.StatusNotFound, "no such viewer")
|
||||
return
|
||||
}
|
||||
writeError(w, http.StatusBadRequest, err.Error())
|
||||
return
|
||||
}
|
||||
s.forgetViewers(userID)
|
||||
s.loggerFor(r.Context()).Info("viewer renamed by operator",
|
||||
"account", userID, "viewer", viewer.ID, "name", viewer.Name)
|
||||
writeJSON(w, http.StatusOK, viewer)
|
||||
}
|
||||
@@ -177,7 +177,7 @@ func (s *Server) handleRowAnalytics(w http.ResponseWriter, r *http.Request, sess
|
||||
}
|
||||
for _, event := range events {
|
||||
if event.Event == store.RowEventSelect {
|
||||
_ = s.cache.InvalidateUser(r.Context(), sess.EmbyUserID)
|
||||
_ = s.cache.InvalidateUser(r.Context(), viewerKeyOf(r.Context(), sess))
|
||||
if s.forYou != nil {
|
||||
s.forYou.MarkDirty(context.WithoutCancel(r.Context()), sess)
|
||||
}
|
||||
|
||||
@@ -91,6 +91,8 @@ type Server struct {
|
||||
openSubtitles *opensubtitles.Client
|
||||
openSubtitlesKey string
|
||||
mdblistMu sync.Mutex
|
||||
// viewerLists spares every authenticated request a read of the account's people.
|
||||
viewerLists viewerListCache
|
||||
// mdblistSettingsCache spares every row and keystroke a settings read.
|
||||
mdblistSettingsCache mdblistSettingsCache
|
||||
// ratingsWarm fills and renews the durable rating cache behind the viewer, so a row
|
||||
@@ -267,6 +269,13 @@ func (s *Server) Routes() http.Handler {
|
||||
v1.Handle("PUT /v1/auth/devices/{deviceID}", s.authed(s.handleRenameDevice))
|
||||
v1.Handle("DELETE /v1/auth/devices/{deviceID}", s.authed(s.handleDeleteDevice))
|
||||
|
||||
// Viewers: the people under one account. The list is what the picker draws; every
|
||||
// other route learns who is watching from the X-Memby-Viewer header instead.
|
||||
v1.Handle("GET /v1/viewers", s.authed(s.handleViewers))
|
||||
v1.Handle("POST /v1/viewers", s.authed(s.handleViewers))
|
||||
v1.Handle("PUT /v1/viewers/{viewerID}", s.authed(s.handleViewer))
|
||||
v1.Handle("DELETE /v1/viewers/{viewerID}", s.authed(s.handleViewer))
|
||||
|
||||
v1.Handle("GET /v1/home", s.authed(s.handleHome))
|
||||
v1.Handle("GET /v1/heroes/active", s.authed(s.handleActiveHero))
|
||||
v1.Handle("GET /v1/screensaver", s.authed(s.handleScreensaver))
|
||||
@@ -435,6 +444,12 @@ func (s *Server) authed(h authedFunc) http.Handler {
|
||||
}
|
||||
sess = s.captureClientIdentity(r, sess)
|
||||
identify(r.Context(), sess)
|
||||
// Resolved once, here, and read out of the context by everything downstream. The
|
||||
// header short-circuits when it is absent, so a household running no viewers pays
|
||||
// a map lookup and nothing else.
|
||||
viewer := s.activeViewer(r.Context(), sess, r)
|
||||
identifyViewer(r.Context(), viewer)
|
||||
r = r.WithContext(withViewer(r.Context(), viewer))
|
||||
policy := s.updatePolicy.get()
|
||||
decision := appupdate.Decide(effectiveUpdatePolicy(policy), clientVersion(r))
|
||||
retireBelow := destructiveUpdateFloor(policy)
|
||||
@@ -446,7 +461,7 @@ func (s *Server) authed(h authedFunc) http.Handler {
|
||||
s.loggerFor(r.Context()).Error("required-update session delete failed", "error", err)
|
||||
}
|
||||
_ = s.cache.Delete(r.Context(), cache.SessionKey(hexHash(sess.TokenHash)))
|
||||
_ = s.cache.InvalidateUser(r.Context(), sess.EmbyUserID)
|
||||
s.invalidateAccountViews(r.Context(), sess)
|
||||
s.loggerFor(r.Context()).Info("signed out for required update",
|
||||
"device_id", sess.DeviceID,
|
||||
"from", clientLogValue(clientVersion(r)),
|
||||
|
||||
@@ -232,7 +232,7 @@ func (s *Server) handleLogout(w http.ResponseWriter, r *http.Request, sess store
|
||||
s.log.Error("session delete failed", "error", err)
|
||||
}
|
||||
_ = s.cache.Delete(r.Context(), cache.SessionKey(hexHash(sess.TokenHash)))
|
||||
_ = s.cache.InvalidateUser(r.Context(), sess.EmbyUserID)
|
||||
s.invalidateAccountViews(r.Context(), sess)
|
||||
s.loggerFor(r.Context()).Info("signed out", "device_id", sess.DeviceID)
|
||||
s.publishAdmin(r.Context(), adminevents.Event{
|
||||
Type: adminevents.TypeLogout,
|
||||
|
||||
@@ -27,6 +27,7 @@ const (
|
||||
featureGenreBrowser = "genre_browser"
|
||||
featureTVCalendar = "tv_calendar"
|
||||
featureWatchTimeDigest = "watch_time_digest"
|
||||
featureViewers = "viewers"
|
||||
)
|
||||
|
||||
type featureDefinition struct {
|
||||
@@ -146,6 +147,25 @@ var featureCatalogue = []featureDefinition{
|
||||
DefaultEnabled: true, MinimumProtocol: 1,
|
||||
Recovery: "Server-enforced; takes effect before the next summary is due.",
|
||||
},
|
||||
{
|
||||
// Default **off**, the stance the genre browser takes. This is the switch that
|
||||
// decides where a household's watched state is written, and a feature that
|
||||
// arrives already on is one every server running this build starts using before
|
||||
// anybody has decided to — so it is opted into rather than out of.
|
||||
//
|
||||
// Switching it on or off never deletes a viewer or their history: the rows stay
|
||||
// in Postgres and come back intact. Off, the gateway routes nobody's state
|
||||
// anywhere but Emby, which is the state a household was in before the feature
|
||||
// existed; on, a shadow viewer's watching goes to Memby and is picked up exactly
|
||||
// where they left it.
|
||||
Key: featureViewers, Name: "Viewers", Area: "Accounts",
|
||||
Description: "Let one Emby account hold several people, each with their own " +
|
||||
"Continue Watching, watched history and favourites. Off by default; turning " +
|
||||
"it off again returns every television to watching as the account itself, " +
|
||||
"without losing what anybody has watched.",
|
||||
DefaultEnabled: false, MinimumProtocol: 1, Capability: "viewers_v1",
|
||||
Recovery: "Takes effect on the next request; nothing a viewer has watched is lost.",
|
||||
},
|
||||
{
|
||||
Key: featureInstallPermission, Name: "Ask TVs for install permission", Area: "Setup",
|
||||
Description: "Ask a signed-in TV that cannot install its own updates to grant the " +
|
||||
|
||||
@@ -76,7 +76,7 @@ func (s *Server) handleBrowseItems(
|
||||
if genre != "" {
|
||||
filterKey = "genre:" + genre
|
||||
}
|
||||
key := cache.UserKey(sess.EmbyUserID, "browse:"+itemType+":"+filterKey+":"+itoa(offset)+":"+itoa(limit))
|
||||
key := cache.UserKey(viewerKeyOf(ctx, sess), "browse:"+itemType+":"+filterKey+":"+itoa(offset)+":"+itoa(limit))
|
||||
if raw, err := s.cache.Get(ctx, key); err == nil {
|
||||
w.Header().Set("X-Memby-Cache", "hit")
|
||||
writeRaw(w, http.StatusOK, raw)
|
||||
@@ -111,7 +111,7 @@ func (s *Server) handleBrowseItems(
|
||||
return
|
||||
}
|
||||
items := nonNil(result.Items)
|
||||
s.decorateItemRatings(ctx, items)
|
||||
s.decorateItems(ctx, items)
|
||||
|
||||
total := genreTotal(result.TotalRecordCount, offset, len(items), limit)
|
||||
|
||||
|
||||
@@ -1109,7 +1109,7 @@ func (s *Server) heroPremiereCandidates(ctx context.Context, now time.Time) []he
|
||||
// The imported catalogue is shared by the household and deliberately carries no user
|
||||
// data, so these cards arrive without ratings. Decorating them is one indexed read
|
||||
// and is what lets a premiere be ranked on the same terms as a film.
|
||||
s.decorateItemRatings(ctx, payloads)
|
||||
s.decorateItems(ctx, payloads)
|
||||
|
||||
byID := make(map[string]json.RawMessage, len(payloads))
|
||||
for _, raw := range payloads {
|
||||
|
||||
@@ -34,7 +34,7 @@ func (s *Server) handleActiveHero(w http.ResponseWriter, r *http.Request, sess s
|
||||
// slot. That is what makes an operator's change reachable immediately without dropping
|
||||
// anything else the household has cached: the old entry is not invalidated, it is
|
||||
// simply no longer named. See heroRevision.
|
||||
key := cache.UserKey(sess.EmbyUserID, "hero:active:v2:"+placement+":"+
|
||||
key := cache.UserKey(viewerKeyOf(r.Context(), sess), "hero:active:v2:"+placement+":"+
|
||||
heroRevision(s.currentHeroPolicy(r.Context()), sess.EmbyUserID, now, location))
|
||||
if raw, err := s.cache.Get(r.Context(), key); err == nil {
|
||||
w.Header().Set("X-Memby-Cache", "hit")
|
||||
@@ -74,7 +74,7 @@ func (s *Server) resolveActiveHero(ctx context.Context, sess store.Session, plac
|
||||
if err != nil {
|
||||
return activeHeroResponse{}, err
|
||||
}
|
||||
s.decorateItemRatings(ctx, result.Items)
|
||||
s.decorateItems(ctx, result.Items)
|
||||
rows := []recommend.Row{{ID: "hero-candidates-" + placement, Kind: "catalogue", Items: result.Items}}
|
||||
|
||||
policy := s.currentHeroPolicy(ctx)
|
||||
|
||||
@@ -83,7 +83,7 @@ func (s *Server) handleHome(w http.ResponseWriter, r *http.Request, sess store.S
|
||||
// four cards. See heroRevision.
|
||||
heroRev := heroRevision(s.currentHeroPolicy(ctx), sess.EmbyUserID, now, s.heroLocation())
|
||||
key := cache.UserKey(
|
||||
sess.EmbyUserID,
|
||||
viewerKeyOf(ctx, sess),
|
||||
"home:v4:"+itoa(limit)+":s"+strconv.FormatBool(sonarrSchedule)+
|
||||
":r"+strconv.FormatBool(radarrSchedule)+":h"+strconv.FormatBool(hero)+
|
||||
":hr"+heroRev+":d"+sess.DeviceID,
|
||||
@@ -135,7 +135,19 @@ func (s *Server) handleHome(w http.ResponseWriter, r *http.Request, sess store.S
|
||||
}()
|
||||
}
|
||||
|
||||
// The three rows that are answers about a *person* rather than about the library.
|
||||
// For the account's own viewer they are Emby's, exactly as they always were; for a
|
||||
// shadow viewer they are built from that viewer's own state, and Emby is asked only to
|
||||
// describe the titles. The fan-out, the failure counting and the merge below are
|
||||
// unchanged either way — this is a substitution of one fetch for another, not a second
|
||||
// code path through the launcher.
|
||||
viewer := viewerOf(ctx, sess)
|
||||
shadow := !viewer.IsMain() && s.store != nil
|
||||
|
||||
run("resume", &out.ContinueWatching, func(ctx context.Context) (*emby.ItemsResult, error) {
|
||||
if shadow {
|
||||
return s.viewerContinueRow(ctx, cred, viewer.ID, limit)
|
||||
}
|
||||
return s.emby.ResumeItems(ctx, cred, rowParams(url.Values{
|
||||
"Recursive": {"true"},
|
||||
"MediaTypes": {"Video"},
|
||||
@@ -143,6 +155,9 @@ func (s *Server) handleHome(w http.ResponseWriter, r *http.Request, sess store.S
|
||||
}, fieldsContinue))
|
||||
})
|
||||
run("favourites", &out.Favorites, func(ctx context.Context) (*emby.ItemsResult, error) {
|
||||
if shadow {
|
||||
return s.viewerFavouritesRow(ctx, cred, viewer.ID, limit)
|
||||
}
|
||||
return s.emby.Items(ctx, cred, rowParams(url.Values{
|
||||
"Filters": {"IsFavorite"},
|
||||
"IncludeItemTypes": {"Movie,Series"},
|
||||
@@ -153,6 +168,9 @@ func (s *Server) handleHome(w http.ResponseWriter, r *http.Request, sess store.S
|
||||
}, fieldsRow))
|
||||
})
|
||||
run("nextup", &out.NextUp, func(ctx context.Context) (*emby.ItemsResult, error) {
|
||||
if shadow {
|
||||
return s.viewerNextUpRow(ctx, cred, viewer.ID, limit)
|
||||
}
|
||||
return s.emby.NextUp(ctx, cred, rowParams(url.Values{
|
||||
"Limit": {itoa(limit)},
|
||||
}, fieldsContinue))
|
||||
@@ -163,7 +181,16 @@ func (s *Server) handleHome(w http.ResponseWriter, r *http.Request, sess store.S
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
played, err := s.recentlyPlayedSeries(timing.WithLabel(ctx, "emby.recent"), cred)
|
||||
// What orders the two halves of the merge. For a shadow viewer it is one grouped
|
||||
// query over their own state rather than a lookback over the account's plays —
|
||||
// the same question, asked of the system that holds the answer.
|
||||
var played map[string]time.Time
|
||||
var err error
|
||||
if shadow {
|
||||
played, err = s.store.ViewerWatchedSeries(ctx, viewer.ID, continuePlayLookback)
|
||||
} else {
|
||||
played, err = s.recentlyPlayedSeries(timing.WithLabel(ctx, "emby.recent"), cred)
|
||||
}
|
||||
if err != nil {
|
||||
s.loggerFor(ctx).Warn("recently played lookup failed", "error", err)
|
||||
return
|
||||
@@ -578,7 +605,7 @@ func (s *Server) handleSearch(w http.ResponseWriter, r *http.Request, sess store
|
||||
return
|
||||
}
|
||||
limit := queryInt(r, "limit", 40, 100)
|
||||
key := cache.UserKey(sess.EmbyUserID, "search:"+itoa(limit)+":"+term+":d"+sess.DeviceID)
|
||||
key := cache.UserKey(viewerKeyOf(ctx, sess), "search:"+itoa(limit)+":"+term+":d"+sess.DeviceID)
|
||||
|
||||
// Every search the tab performs is recorded here, before the cache is consulted, so a
|
||||
// query answered from Redis counts the same as one that reached Emby. The client also
|
||||
@@ -606,7 +633,7 @@ func (s *Server) handleSearch(w http.ResponseWriter, r *http.Request, sess store
|
||||
return
|
||||
}
|
||||
items := s.personalizeSearch(ctx, sess, term, result.Items, limit)
|
||||
s.decorateItemRatings(ctx, items)
|
||||
s.decorateItems(ctx, items)
|
||||
// Instant search fires a request per keystroke past the second character, so this is
|
||||
// DEBUG: it is the record of what somebody was looking for when nothing was found,
|
||||
// not something to carry in the normal log.
|
||||
|
||||
@@ -48,7 +48,7 @@ func (s *Server) handleItem(w http.ResponseWriter, r *http.Request, sess store.S
|
||||
writeError(w, http.StatusBadRequest, "item id is required")
|
||||
return
|
||||
}
|
||||
if raw, err := s.cache.Get(ctx, itemDetailKey(sess.EmbyUserID, itemID)); err == nil {
|
||||
if raw, err := s.cache.Get(ctx, itemDetailKey(viewerKeyOf(ctx, sess), itemID)); err == nil {
|
||||
w.Header().Set("X-Memby-Cache", "hit")
|
||||
writeRaw(w, http.StatusOK, raw)
|
||||
return
|
||||
@@ -88,7 +88,7 @@ func (s *Server) detailItem(
|
||||
// than after a second request. Anything not yet stored still arrives on
|
||||
// /ratings.
|
||||
decorated := []json.RawMessage{raw}
|
||||
s.decorateItemRatings(ctx, decorated)
|
||||
s.decorateItems(ctx, decorated)
|
||||
return decorated[0], nil
|
||||
})
|
||||
return item, err
|
||||
@@ -255,7 +255,7 @@ func (s *Server) handleSeriesEpisodes(w http.ResponseWriter, r *http.Request, se
|
||||
// Watching card is focused, and the page asks again when somebody presses. A
|
||||
// long-running show is a thousand records, so two of them is a real cost on the one
|
||||
// press that must feel free.
|
||||
key := cache.UserKey(sess.EmbyUserID, "series-episodes:"+seriesID)
|
||||
key := cache.UserKey(viewerKeyOf(ctx, sess), "series-episodes:"+seriesID)
|
||||
body, hit, err := s.cachedRead(ctx, key, s.cfg.ItemTTL,
|
||||
func(ctx context.Context) (json.RawMessage, error) {
|
||||
result, err := s.emby.Episodes(
|
||||
@@ -278,6 +278,11 @@ func (s *Server) handleSeriesEpisodes(w http.ResponseWriter, r *http.Request, se
|
||||
if items == nil {
|
||||
items = []json.RawMessage{}
|
||||
}
|
||||
// The ticks down the episode list are the clearest statement this app makes
|
||||
// about what somebody has seen, so they are the last place the account's
|
||||
// answer may be left standing. The cache key is the viewer's, so this is
|
||||
// stored per person rather than decorated on the way out.
|
||||
s.decorateViewerState(ctx, items)
|
||||
return json.Marshal(seriesEpisodesResponse{Items: items})
|
||||
})
|
||||
if err != nil {
|
||||
@@ -307,33 +312,78 @@ func (s *Server) handleTrailer(w http.ResponseWriter, r *http.Request, sess stor
|
||||
writeRaw(w, http.StatusOK, result.Items[0])
|
||||
}
|
||||
|
||||
// The four routes below are the whole of what Memby writes back to Emby about a person:
|
||||
// a favourite, a watched flag, a hidden resume item and a playback report. Each one now
|
||||
// asks who is watching first, because that is the entire promise of a shadow viewer — the
|
||||
// Emby account lends them the library and never learns what they did with it.
|
||||
|
||||
func (s *Server) handleFavorite(w http.ResponseWriter, r *http.Request, sess store.Session) {
|
||||
s.setFlag(w, r, sess, func(itemID string, value bool) (json.RawMessage, error) {
|
||||
s.setFlag(w, r, sess, func(viewer store.Viewer, itemID string, value bool) (json.RawMessage, error) {
|
||||
if !viewer.IsMain() {
|
||||
return s.setShadowFlag(r.Context(), viewer, itemID, func() error {
|
||||
return s.store.SetViewerFavourite(r.Context(), viewer.ID, itemID, value)
|
||||
})
|
||||
}
|
||||
return s.emby.SetFavorite(r.Context(), credentials(sess), itemID, value)
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Server) handlePlayed(w http.ResponseWriter, r *http.Request, sess store.Session) {
|
||||
s.setFlag(w, r, sess, func(itemID string, value bool) (json.RawMessage, error) {
|
||||
s.setFlag(w, r, sess, func(viewer store.Viewer, itemID string, value bool) (json.RawMessage, error) {
|
||||
if !viewer.IsMain() {
|
||||
return s.setShadowFlag(r.Context(), viewer, itemID, func() error {
|
||||
return s.store.SetViewerPlayed(r.Context(), viewer.ID, itemID, value)
|
||||
})
|
||||
}
|
||||
return s.emby.SetPlayed(r.Context(), credentials(sess), itemID, value)
|
||||
})
|
||||
}
|
||||
|
||||
// setShadowFlag applies a Memby-side mutation and answers in the shape Emby would have.
|
||||
//
|
||||
// Re-reading the row rather than describing the write is deliberate: the response is what
|
||||
// the television draws the card from, and a favourite pressed on a title that is also part
|
||||
// way through has to come back carrying the position as well as the heart.
|
||||
func (s *Server) setShadowFlag(
|
||||
ctx context.Context, viewer store.Viewer, itemID string, apply func() error,
|
||||
) (json.RawMessage, error) {
|
||||
if err := apply(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
state, err := s.store.ViewerStateFor(ctx, viewer.ID, itemID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return viewerUserData(state), nil
|
||||
}
|
||||
|
||||
func (s *Server) handleHideFromResume(w http.ResponseWriter, r *http.Request, sess store.Session) {
|
||||
itemID := r.PathValue("id")
|
||||
if itemID == "" {
|
||||
writeError(w, http.StatusBadRequest, "item id is required")
|
||||
return
|
||||
}
|
||||
userData, err := s.emby.HideFromResume(r.Context(), credentials(sess), itemID)
|
||||
viewer := s.activeViewer(r.Context(), sess, r)
|
||||
var userData json.RawMessage
|
||||
var err error
|
||||
if viewer.IsMain() {
|
||||
userData, err = s.emby.HideFromResume(r.Context(), credentials(sess), itemID)
|
||||
} else {
|
||||
userData, err = s.setShadowFlag(r.Context(), viewer, itemID, func() error {
|
||||
return s.store.HideViewerFromResume(r.Context(), viewer.ID, itemID)
|
||||
})
|
||||
}
|
||||
if err != nil {
|
||||
s.writeUpstreamError(r.Context(), w, err, "could not remove the item from Continue Watching")
|
||||
return
|
||||
}
|
||||
if err := s.cache.InvalidateUser(r.Context(), sess.EmbyUserID); err != nil {
|
||||
if err := s.cache.InvalidateUser(r.Context(), viewer.ID); err != nil {
|
||||
s.loggerFor(r.Context()).Warn("cache invalidation failed", "error", err)
|
||||
}
|
||||
if s.forYou != nil {
|
||||
// For You is built from the Emby account's own history, so a shadow viewer's press
|
||||
// says nothing about it. Marking it dirty would rebuild the main viewer's row out of
|
||||
// somebody else's choice.
|
||||
if s.forYou != nil && viewer.IsMain() {
|
||||
s.forYou.MarkDirty(r.Context(), sess)
|
||||
}
|
||||
writeRaw(w, http.StatusOK, userData)
|
||||
@@ -345,8 +395,9 @@ func (s *Server) setFlag(
|
||||
w http.ResponseWriter,
|
||||
r *http.Request,
|
||||
sess store.Session,
|
||||
apply func(itemID string, value bool) (json.RawMessage, error),
|
||||
apply func(viewer store.Viewer, itemID string, value bool) (json.RawMessage, error),
|
||||
) {
|
||||
viewer := s.activeViewer(r.Context(), sess, r)
|
||||
itemID := r.PathValue("id")
|
||||
if itemID == "" {
|
||||
writeError(w, http.StatusBadRequest, "item id is required")
|
||||
@@ -358,15 +409,18 @@ func (s *Server) setFlag(
|
||||
return
|
||||
}
|
||||
|
||||
userData, err := apply(itemID, req.Value)
|
||||
userData, err := apply(viewer, itemID, req.Value)
|
||||
if err != nil {
|
||||
s.writeUpstreamError(r.Context(), w, err, "could not update the item")
|
||||
return
|
||||
}
|
||||
if err := s.cache.InvalidateUser(r.Context(), sess.EmbyUserID); err != nil {
|
||||
if err := s.cache.InvalidateUser(r.Context(), viewer.ID); err != nil {
|
||||
s.loggerFor(r.Context()).Warn("cache invalidation failed", "error", err)
|
||||
}
|
||||
if s.forYou != nil {
|
||||
// For You is built from the Emby account's own history, so a shadow viewer's press
|
||||
// says nothing about it. Marking it dirty would rebuild the main viewer's row out of
|
||||
// somebody else's choice.
|
||||
if s.forYou != nil && viewer.IsMain() {
|
||||
s.forYou.MarkDirty(r.Context(), sess)
|
||||
}
|
||||
writeRaw(w, http.StatusOK, userData)
|
||||
|
||||
@@ -24,6 +24,7 @@ type requestIdentity struct {
|
||||
component string
|
||||
userID string
|
||||
user string
|
||||
viewer string
|
||||
device string
|
||||
client string
|
||||
protocol string
|
||||
@@ -73,6 +74,21 @@ func identify(ctx context.Context, sess store.Session) {
|
||||
}
|
||||
}
|
||||
|
||||
// identifyViewer names the person watching, where that is somebody other than the account
|
||||
// itself. A main viewer is deliberately not recorded: its name is already the "user" field,
|
||||
// and printing it twice on every line would say nothing.
|
||||
func identifyViewer(ctx context.Context, viewer store.Viewer) {
|
||||
identity := identityFrom(ctx)
|
||||
if identity == nil || viewer.IsMain() {
|
||||
return
|
||||
}
|
||||
if viewer.Name != "" {
|
||||
identity.viewer = viewer.Name
|
||||
} else {
|
||||
identity.viewer = viewer.ID
|
||||
}
|
||||
}
|
||||
|
||||
func (i *requestIdentity) attrs() []any {
|
||||
if i == nil {
|
||||
return nil
|
||||
@@ -124,6 +140,9 @@ func (i *requestIdentity) viewerAttrs() []any {
|
||||
if i.user != "" {
|
||||
attrs = append(attrs, "user", i.user)
|
||||
}
|
||||
if i.viewer != "" {
|
||||
attrs = append(attrs, "viewer", i.viewer)
|
||||
}
|
||||
if i.device != "" {
|
||||
attrs = append(attrs, "device", i.device)
|
||||
}
|
||||
|
||||
@@ -56,7 +56,7 @@ func (s *Server) handlePersonFilmography(w http.ResponseWriter, r *http.Request,
|
||||
writeError(w, http.StatusBadRequest, "person id is required")
|
||||
return
|
||||
}
|
||||
key := cache.UserKey(sess.EmbyUserID, "person-filmography:v1:"+personID)
|
||||
key := cache.UserKey(viewerKeyOf(ctx, sess), "person-filmography:v1:"+personID)
|
||||
body, hit, err := s.cachedRead(ctx, key, s.cfg.ItemTTL,
|
||||
func(ctx context.Context) (json.RawMessage, error) {
|
||||
result, err := s.emby.Items(ctx, credentials(sess), url.Values{
|
||||
|
||||
+122
-16
@@ -129,6 +129,7 @@ func (s *Server) handlePlayback(w http.ResponseWriter, r *http.Request, sess sto
|
||||
return
|
||||
}
|
||||
cred := credentials(sess)
|
||||
viewer := viewerOf(ctx, sess)
|
||||
|
||||
item, hinted := playbackHint(r, itemID)
|
||||
if !hinted {
|
||||
@@ -148,7 +149,7 @@ func (s *Server) handlePlayback(w http.ResponseWriter, r *http.Request, sess sto
|
||||
title := item.Name
|
||||
|
||||
if strings.EqualFold(item.Type, "Series") {
|
||||
episode, err := s.firstPlayableEpisode(ctx, cred, item.ID)
|
||||
episode, err := s.firstPlayableEpisode(ctx, cred, viewer, item.ID)
|
||||
if err != nil {
|
||||
s.writeUpstreamError(ctx, w, err, "could not find an episode to play")
|
||||
return
|
||||
@@ -163,6 +164,26 @@ func (s *Server) handlePlayback(w http.ResponseWriter, r *http.Request, sess sto
|
||||
}
|
||||
}
|
||||
|
||||
// Where a shadow viewer resumes from is Memby's answer, and it is taken here rather
|
||||
// than trusted from the card.
|
||||
//
|
||||
// The hint the television sends is read off a card this gateway already decorated with
|
||||
// this viewer's own state, so the two normally agree — but only normally. The store has
|
||||
// heard about the episode they were part-way through on the other television, and a
|
||||
// card is only as fresh as the last home refresh. This is also the value handed to
|
||||
// PlaybackInfo below, so taking it here fixes the negotiated stream as well as the
|
||||
// number sent back.
|
||||
if !viewer.IsMain() && s.store != nil {
|
||||
if state, err := s.store.ViewerStateFor(ctx, viewer.ID, target.ID); err == nil {
|
||||
target.UserData.PlaybackPositionTicks = state.PositionTicks
|
||||
} else {
|
||||
// Starting from the beginning is a recoverable disappointment; starting from
|
||||
// where somebody else got to is not.
|
||||
s.loggerFor(ctx).Warn("viewer resume position unavailable", "error", err)
|
||||
target.UserData.PlaybackPositionTicks = 0
|
||||
}
|
||||
}
|
||||
|
||||
var subtitleIndex *int
|
||||
if raw := strings.TrimSpace(r.URL.Query().Get("subtitleIndex")); raw != "" {
|
||||
if parsed, err := strconv.Atoi(raw); err == nil && parsed >= 0 {
|
||||
@@ -265,7 +286,26 @@ func playbackHint(r *http.Request, itemID string) (emby.Summary, bool) {
|
||||
}
|
||||
|
||||
// firstPlayableEpisode prefers the server's next-up choice and falls back to episode one.
|
||||
func (s *Server) firstPlayableEpisode(ctx context.Context, cred emby.Credentials, seriesID string) (*emby.Summary, error) {
|
||||
// The viewer is threaded in because "where does this series start" is a question about a
|
||||
// person, and Emby's NextUp answers it for the account. A shadow viewer's answer is their
|
||||
// own: the first episode they have not finished.
|
||||
func (s *Server) firstPlayableEpisode(
|
||||
ctx context.Context, cred emby.Credentials, viewer store.Viewer, seriesID string,
|
||||
) (*emby.Summary, error) {
|
||||
if !viewer.IsMain() && s.store != nil {
|
||||
episode, err := s.firstUnwatchedEpisodeFor(ctx, cred, viewer, seriesID)
|
||||
if err != nil {
|
||||
// Falling through to Emby's answer is wrong for this viewer, so it is not
|
||||
// done: starting somebody at the account's next episode is the leak this
|
||||
// feature exists to prevent.
|
||||
return nil, err
|
||||
}
|
||||
if episode != nil {
|
||||
return episode, nil
|
||||
}
|
||||
// Nothing recorded for this series yet: fall through and let Emby name its first
|
||||
// episode, which is the right answer for somebody who has never watched any of it.
|
||||
}
|
||||
nextUp, err := s.emby.NextUp(ctx, cred, url.Values{
|
||||
"SeriesId": {seriesID},
|
||||
"Limit": {"1"},
|
||||
@@ -376,6 +416,20 @@ func (s *Server) handleNextEpisode(w http.ResponseWriter, r *http.Request, sess
|
||||
title = series + " – " + title
|
||||
}
|
||||
|
||||
// Which episode follows is a property of the season and is the same for everybody;
|
||||
// how far into it *this* viewer already is, is not. The item payload is rewritten as
|
||||
// well as the summary, because the television draws the next-up banner from it.
|
||||
viewer := viewerOf(ctx, sess)
|
||||
if !viewer.IsMain() && s.store != nil {
|
||||
state, stateErr := s.store.ViewerStateFor(ctx, viewer.ID, next.ID)
|
||||
if stateErr != nil {
|
||||
s.loggerFor(ctx).Warn("viewer next-episode position unavailable", "error", stateErr)
|
||||
state = store.ViewerState{}
|
||||
}
|
||||
next.UserData.PlaybackPositionTicks = state.PositionTicks
|
||||
raw = injectItemUserData(raw, viewerUserData(state))
|
||||
}
|
||||
|
||||
// The metadata-only shape. Everything below this point is a PlaybackInfo negotiation,
|
||||
// and a client that said it does not want one yet must not be given one anyway.
|
||||
if !nextEpisodeWantsStream(r) {
|
||||
@@ -798,6 +852,41 @@ func seriesNameOf(raw json.RawMessage) string {
|
||||
return parsed.SeriesName
|
||||
}
|
||||
|
||||
// recordShadowPlayback is the other side of the playback report: the same three phases,
|
||||
// written to Memby instead of to Emby.
|
||||
//
|
||||
// Where a title is *finished* is decided here rather than by the television, for the
|
||||
// reason the gateway decides which subtitle comes on: Emby applies its own completion
|
||||
// threshold on the main viewer's behalf, and a shadow viewer must be judged by the same
|
||||
// rule or one household would disagree with itself about whether an episode is watched
|
||||
// depending on who watched it.
|
||||
//
|
||||
// A paused progress report still records the position. Pausing is where somebody leaves a
|
||||
// film, and the ten seconds between reports is exactly the window a set switched off at
|
||||
// the wall would otherwise lose.
|
||||
func (s *Server) recordShadowPlayback(
|
||||
ctx context.Context, viewer store.Viewer, phase string, report playbackReport,
|
||||
) error {
|
||||
if s.store == nil {
|
||||
return fmt.Errorf("no store for viewer playback")
|
||||
}
|
||||
// The pool's own tracer times this; nothing extra is recorded here.
|
||||
position := max64(report.PositionMs, 0) * ticksPerMillisecond
|
||||
runtime := max64(report.DurationMs, 0) * ticksPerMillisecond
|
||||
state := store.ViewerState{
|
||||
ItemID: report.ItemID,
|
||||
PositionTicks: position,
|
||||
RuntimeTicks: runtime,
|
||||
}
|
||||
// Only a stop can complete a title. A progress report crossing the threshold is
|
||||
// somebody still watching the closing minutes, and marking it played there would take
|
||||
// the episode out of Continue Watching underneath them.
|
||||
if phase == "stopped" {
|
||||
state.Played = store.PlayedFromPosition(position, runtime)
|
||||
}
|
||||
return s.store.RecordViewerPlayback(ctx, viewer.ID, state)
|
||||
}
|
||||
|
||||
// handlePlaybackReport forwards progress to Emby. Stopping invalidates the user's cache
|
||||
// so Continue Watching reflects the new position on the next home load.
|
||||
func (s *Server) handlePlaybackReport(w http.ResponseWriter, r *http.Request, sess store.Session) {
|
||||
@@ -826,12 +915,24 @@ func (s *Server) handlePlaybackReport(w http.ResponseWriter, r *http.Request, se
|
||||
"play_session_id", clientLogValue(report.PlaySessionID),
|
||||
)
|
||||
|
||||
err := s.emby.ReportPlayback(
|
||||
timing.WithLabel(r.Context(), "emby.report"),
|
||||
credentials(sess), phase, report.ItemID, report.MediaSourceID,
|
||||
report.PlaySessionID, report.PlayMethod, report.EventName,
|
||||
max64(report.PositionMs, 0)*ticksPerMillisecond, report.IsPaused,
|
||||
)
|
||||
// Who is watching decides where this goes, and it is the only place that decision is
|
||||
// made for progress. A shadow viewer's evening is Memby's: nothing below reaches
|
||||
// /Sessions/Playing, so the Emby account lending them the library never learns what
|
||||
// they watched or how far they got.
|
||||
viewer := s.activeViewer(r.Context(), sess, r)
|
||||
log = log.With("viewer", viewer.ID)
|
||||
|
||||
var err error
|
||||
if viewer.IsMain() {
|
||||
err = s.emby.ReportPlayback(
|
||||
timing.WithLabel(r.Context(), "emby.report"),
|
||||
credentials(sess), phase, report.ItemID, report.MediaSourceID,
|
||||
report.PlaySessionID, report.PlayMethod, report.EventName,
|
||||
max64(report.PositionMs, 0)*ticksPerMillisecond, report.IsPaused,
|
||||
)
|
||||
} else {
|
||||
err = s.recordShadowPlayback(r.Context(), viewer, phase, report)
|
||||
}
|
||||
if err != nil {
|
||||
log.Warn("playback report failed", "phase", phase, "error", err)
|
||||
// Progress is advisory and another reading follows in ten seconds. A final stop is
|
||||
@@ -876,7 +977,7 @@ func (s *Server) handlePlaybackReport(w http.ResponseWriter, r *http.Request, se
|
||||
|
||||
if phase == "stopped" {
|
||||
invalidate := timing.Start(r.Context(), "invalidate")
|
||||
if err := s.cache.InvalidateUser(r.Context(), sess.EmbyUserID); err != nil {
|
||||
if err := s.cache.InvalidateUser(r.Context(), viewer.ID); err != nil {
|
||||
s.loggerFor(r.Context()).Warn("cache invalidation failed", "error", err)
|
||||
}
|
||||
invalidate()
|
||||
@@ -891,10 +992,10 @@ func (s *Server) handlePlaybackReport(w http.ResponseWriter, r *http.Request, se
|
||||
// episode stops here rather than at the durable insert four upstream calls later;
|
||||
// the feature check is a cached read; and only then is anything asked of Emby.
|
||||
if shouldAutoFollowShow(phase, report.PositionMs, report.DurationMs) &&
|
||||
s.followChecks.claim(sess.EmbyUserID, report.ItemID) &&
|
||||
s.followChecks.claim(viewer.ID, report.ItemID) &&
|
||||
s.featureEnabled(r.Context(), featureAutomaticMyShows) {
|
||||
follow := timing.Start(r.Context(), "autofollow")
|
||||
response.AutoFollowedShowTitle = s.autoFollowContinuingShow(r.Context(), sess, report.ItemID)
|
||||
response.AutoFollowedShowTitle = s.autoFollowContinuingShow(r.Context(), sess, viewer, report.ItemID)
|
||||
follow()
|
||||
}
|
||||
writeJSON(w, http.StatusOK, response)
|
||||
@@ -907,7 +1008,12 @@ func shouldAutoFollowShow(phase string, positionMs, durationMs int64) bool {
|
||||
return phase != "started" && durationMs > 0 && positionMs >= (durationMs+1)/2
|
||||
}
|
||||
|
||||
func (s *Server) autoFollowContinuingShow(ctx context.Context, sess store.Session, episodeID string) string {
|
||||
// The Emby credential reads the catalogue; the viewer owns the list it is written to.
|
||||
// Following a show is a Memby preference and belongs to the person, so a shadow viewer
|
||||
// finishing an episode fills their own My Shows rather than the account's.
|
||||
func (s *Server) autoFollowContinuingShow(
|
||||
ctx context.Context, sess store.Session, viewer store.Viewer, episodeID string,
|
||||
) string {
|
||||
if s.sonarr == nil || s.store == nil {
|
||||
return ""
|
||||
}
|
||||
@@ -949,7 +1055,7 @@ func (s *Server) autoFollowContinuingShow(ctx context.Context, sess store.Sessio
|
||||
ItemID: episode.SeriesID, Title: seriesItem.Name, Year: seriesItem.ProductionYear,
|
||||
ImageTag: seriesItem.ImageTags["Primary"],
|
||||
}
|
||||
inserted, err := s.store.SaveUserShowIfAbsent(ctx, sess.EmbyUserID, show)
|
||||
inserted, err := s.store.SaveUserShowIfAbsent(ctx, viewer.ID, show)
|
||||
if err != nil {
|
||||
s.loggerFor(ctx).Warn("auto-follow save failed", "error", err)
|
||||
return ""
|
||||
@@ -957,7 +1063,7 @@ func (s *Server) autoFollowContinuingShow(ctx context.Context, sess store.Sessio
|
||||
if !inserted {
|
||||
return ""
|
||||
}
|
||||
prefs, err := s.store.NotificationPreferences(ctx, sess.EmbyUserID)
|
||||
prefs, err := s.store.NotificationPreferences(ctx, viewer.ID)
|
||||
if err != nil {
|
||||
s.loggerFor(ctx).Warn("auto-follow notification preferences unavailable", "error", err)
|
||||
return ""
|
||||
@@ -965,8 +1071,8 @@ func (s *Server) autoFollowContinuingShow(ctx context.Context, sess store.Sessio
|
||||
notification := notify.Notification{
|
||||
Kind: "auto-follow",
|
||||
Source: notifySourceAutoFollow,
|
||||
UserID: sess.EmbyUserID,
|
||||
Username: sess.Username,
|
||||
UserID: viewer.ID,
|
||||
Username: viewer.Name,
|
||||
Title: "Added to My Shows",
|
||||
Body: seriesItem.Name + " was added because you started watching it and it is still continuing.",
|
||||
ItemID: episode.SeriesID,
|
||||
|
||||
@@ -586,7 +586,7 @@ func (s *Server) handleRecommendationAction(
|
||||
writeError(w, http.StatusBadRequest, err.Error())
|
||||
return
|
||||
}
|
||||
_ = s.cache.InvalidateUser(r.Context(), sess.EmbyUserID)
|
||||
_ = s.cache.InvalidateUser(r.Context(), viewerKeyOf(r.Context(), sess))
|
||||
if s.forYou != nil {
|
||||
s.forYou.MarkDirty(context.WithoutCancel(r.Context()), sess)
|
||||
s.forYou.RefreshAsync(sess, false)
|
||||
@@ -640,7 +640,7 @@ func (s *Server) handleRecommendationPreferences(
|
||||
writeError(w, http.StatusInternalServerError, "could not save onboarding preferences")
|
||||
return
|
||||
}
|
||||
_ = s.cache.InvalidateUser(r.Context(), sess.EmbyUserID)
|
||||
_ = s.cache.InvalidateUser(r.Context(), viewerKeyOf(r.Context(), sess))
|
||||
if s.forYou != nil {
|
||||
s.forYou.MarkDirty(context.WithoutCancel(r.Context()), sess)
|
||||
s.forYou.RefreshAsync(sess, false)
|
||||
|
||||
@@ -231,7 +231,18 @@ func (s *Server) decorateHomeRatings(ctx context.Context, out *homeResponse) {
|
||||
}
|
||||
collections = append(collections,
|
||||
out.ContinueWatching, out.NextUp, out.Favorites, out.LatestMovies)
|
||||
s.decorateItems(ctx, collections...)
|
||||
}
|
||||
|
||||
// decorateItems is the one door items leave the gateway through.
|
||||
//
|
||||
// It attaches both of the things Memby knows about a title that Emby's payload does not
|
||||
// carry: the stored review scores, and — for a shadow viewer — whose progress this is. The
|
||||
// two are separate concerns and stayed separate functions, but every call site wanted both,
|
||||
// and a decoration added at seven sites is a decoration missing from the eighth.
|
||||
func (s *Server) decorateItems(ctx context.Context, collections ...[]json.RawMessage) {
|
||||
s.decorateItemRatings(ctx, collections...)
|
||||
s.decorateViewerState(ctx, collections...)
|
||||
}
|
||||
|
||||
// decorateItemRatings rewrites each item in place with whatever the database already
|
||||
@@ -293,7 +304,7 @@ func (s *Server) decorateRowRatings(ctx context.Context, rows []recommend.Row) {
|
||||
for _, row := range rows {
|
||||
collections = append(collections, row.Items)
|
||||
}
|
||||
s.decorateItemRatings(ctx, collections...)
|
||||
s.decorateItems(ctx, collections...)
|
||||
}
|
||||
|
||||
// ratingKeysForItems resolves Emby ids to external titles, preferring the index built by
|
||||
|
||||
@@ -40,7 +40,7 @@ func (s *Server) handleRelated(w http.ResponseWriter, r *http.Request, sess stor
|
||||
writeError(w, http.StatusBadRequest, "item id is required")
|
||||
return
|
||||
}
|
||||
key := cache.UserKey(sess.EmbyUserID, "related:v2:"+itemID)
|
||||
key := cache.UserKey(viewerKeyOf(ctx, sess), "related:v2:"+itemID)
|
||||
if raw, err := s.cache.Get(ctx, key); err == nil {
|
||||
w.Header().Set("X-Memby-Cache", "hit")
|
||||
writeRaw(w, http.StatusOK, raw)
|
||||
@@ -64,7 +64,7 @@ func (s *Server) handleRelated(w http.ResponseWriter, r *http.Request, sess stor
|
||||
return nil, err
|
||||
}
|
||||
items := nonNilRaws(recommend.Raws(related))
|
||||
s.decorateItemRatings(ctx, items)
|
||||
s.decorateItems(ctx, items)
|
||||
return json.Marshal(relatedResponse{
|
||||
Reasons: nonNilStrings(reasons),
|
||||
Items: items,
|
||||
|
||||
@@ -0,0 +1,174 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/url"
|
||||
"strings"
|
||||
|
||||
"github.com/ponzischeme89/memby/server/internal/emby"
|
||||
"github.com/ponzischeme89/memby/server/internal/store"
|
||||
"github.com/ponzischeme89/memby/server/internal/timing"
|
||||
)
|
||||
|
||||
// viewerRowLookupLimit bounds one row's worth of ids. A row is a shelf on a television and
|
||||
// nothing draws more than a screenful plus what the D-pad can reach.
|
||||
const viewerRowLookupLimit = 60
|
||||
|
||||
// itemsByID fetches a named set of titles and returns them **in the order asked for**.
|
||||
//
|
||||
// Emby answers an Ids= query in its own order, and for a shadow viewer the order is the
|
||||
// whole answer: Continue Watching is "what am I in the middle of, most recent first", and
|
||||
// that ranking was decided in Postgres out of this viewer's own history. Handing back
|
||||
// Emby's order would keep the right titles and throw away the reason they were chosen.
|
||||
//
|
||||
// The metadata is still Emby's. Only the viewing state belongs to Memby, which is why this
|
||||
// asks for the ordinary row fields and lets decorateItems replace the UserData afterwards.
|
||||
func (s *Server) itemsByID(
|
||||
ctx context.Context, cred emby.Credentials, ids []string, fields string,
|
||||
) (*emby.ItemsResult, error) {
|
||||
if len(ids) == 0 {
|
||||
return &emby.ItemsResult{}, nil
|
||||
}
|
||||
if len(ids) > viewerRowLookupLimit {
|
||||
ids = ids[:viewerRowLookupLimit]
|
||||
}
|
||||
result, err := s.emby.Items(ctx, cred, rowParams(url.Values{
|
||||
"Ids": {strings.Join(ids, ",")},
|
||||
"Recursive": {"true"},
|
||||
"Limit": {itoa(len(ids))},
|
||||
}, fields))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result.Items = orderItemsByID(result.Items, ids)
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// orderItemsByID puts a set of items back into the order they were asked for.
|
||||
//
|
||||
// A title the catalogue still names but Emby no longer answers for is dropped rather than
|
||||
// left as a gap — the row is drawn from what comes back, and a missing card is better than
|
||||
// one that cannot be opened. A title Emby volunteers that was not asked for is dropped too:
|
||||
// the ids are the answer, and anything else in the response is not part of it.
|
||||
func orderItemsByID(items []json.RawMessage, ids []string) []json.RawMessage {
|
||||
byID := make(map[string]json.RawMessage, len(items))
|
||||
for _, raw := range items {
|
||||
if id := itemIDOf(raw); id != "" {
|
||||
if _, seen := byID[id]; !seen {
|
||||
byID[id] = raw
|
||||
}
|
||||
}
|
||||
}
|
||||
ordered := make([]json.RawMessage, 0, len(ids))
|
||||
for _, id := range ids {
|
||||
if raw, ok := byID[id]; ok {
|
||||
ordered = append(ordered, raw)
|
||||
// Removed so a repeated id cannot draw the same card twice. The television
|
||||
// keys its rows by item id and throws on a duplicate.
|
||||
delete(byID, id)
|
||||
}
|
||||
}
|
||||
return ordered
|
||||
}
|
||||
|
||||
// viewerContinueRow is a shadow viewer's Continue Watching, built from their own playheads.
|
||||
func (s *Server) viewerContinueRow(
|
||||
ctx context.Context, cred emby.Credentials, viewerID string, limit int,
|
||||
) (*emby.ItemsResult, error) {
|
||||
ids, err := s.store.ViewerResumeItems(ctx, viewerID, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s.itemsByID(timing.WithLabel(ctx, "emby.viewer_resume"), cred, ids, fieldsContinue)
|
||||
}
|
||||
|
||||
// viewerNextUpRow is the next unwatched episode of each series this viewer is part-way
|
||||
// through. The ranking is Postgres's; Emby is only asked to describe the titles.
|
||||
func (s *Server) viewerNextUpRow(
|
||||
ctx context.Context, cred emby.Credentials, viewerID string, limit int,
|
||||
) (*emby.ItemsResult, error) {
|
||||
ids, err := s.store.ViewerNextUp(ctx, viewerID, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s.itemsByID(timing.WithLabel(ctx, "emby.viewer_nextup"), cred, ids, fieldsContinue)
|
||||
}
|
||||
|
||||
// viewerFavouritesRow is this viewer's own favourites rather than the account's.
|
||||
//
|
||||
// It is deliberately *not* re-sorted by name the way the Emby row is. A shadow viewer's
|
||||
// favourites are the ones they marked, and the order they marked them in is the only
|
||||
// ordering Memby has that means anything.
|
||||
func (s *Server) viewerFavouritesRow(
|
||||
ctx context.Context, cred emby.Credentials, viewerID string, limit int,
|
||||
) (*emby.ItemsResult, error) {
|
||||
ids, err := s.store.ViewerFavouriteItems(ctx, viewerID, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s.itemsByID(timing.WithLabel(ctx, "emby.viewer_favourites"), cred, ids, fieldsRow)
|
||||
}
|
||||
|
||||
// firstUnwatchedEpisodeFor is where a shadow viewer's series starts.
|
||||
//
|
||||
// It reads the series' episodes once from Emby and walks them against this viewer's own
|
||||
// played set, rather than asking Emby's NextUp — which answers for the account and is the
|
||||
// whole reason a shadow viewer pressing Play on a show they have never seen was being
|
||||
// dropped into the middle of somebody else's season.
|
||||
//
|
||||
// A part-watched episode wins over the first unwatched one: somebody eleven minutes into
|
||||
// an episode wants that episode, which is the same judgement the Continue Watching merge
|
||||
// makes.
|
||||
func (s *Server) firstUnwatchedEpisodeFor(
|
||||
ctx context.Context, cred emby.Credentials, viewer store.Viewer, seriesID string,
|
||||
) (*emby.Summary, error) {
|
||||
states, err := s.store.ViewerPlayedInSeries(ctx, viewer.ID, seriesID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(states) == 0 {
|
||||
// Never watched. Emby's own first episode is the right answer and costs the
|
||||
// caller nothing extra to ask for.
|
||||
return nil, nil
|
||||
}
|
||||
episodes, err := s.emby.Episodes(timing.WithLabel(ctx, "emby.viewer_series"), cred, seriesID, url.Values{
|
||||
"Fields": {"Overview,RunTimeTicks,SeriesName,ParentIndexNumber,IndexNumber"},
|
||||
"EnableUserData": {"false"},
|
||||
"EnableTotalRecordCount": {"false"},
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resume, err := s.store.ViewerStates(ctx, viewer.ID, episodeIDsOf(episodes.Items))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var firstUnplayed *emby.Summary
|
||||
for _, raw := range episodes.Items {
|
||||
summary, err := emby.Summarise(raw)
|
||||
if err != nil || summary.ID == "" {
|
||||
continue
|
||||
}
|
||||
state := resume[summary.ID]
|
||||
if state.PositionTicks > 0 && !state.Played {
|
||||
summary.UserData.PlaybackPositionTicks = state.PositionTicks
|
||||
return &summary, nil
|
||||
}
|
||||
if !state.Played && firstUnplayed == nil {
|
||||
episode := summary
|
||||
firstUnplayed = &episode
|
||||
}
|
||||
}
|
||||
return firstUnplayed, nil
|
||||
}
|
||||
|
||||
func episodeIDsOf(items []json.RawMessage) []string {
|
||||
ids := make([]string, 0, len(items))
|
||||
for _, raw := range items {
|
||||
if id := itemIDOf(raw); id != "" {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
}
|
||||
return ids
|
||||
}
|
||||
@@ -0,0 +1,172 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
|
||||
"github.com/ponzischeme89/memby/server/internal/store"
|
||||
)
|
||||
|
||||
// userDataItemField is the block a television draws a progress bar, a tick and a heart
|
||||
// from. For a shadow viewer it is written here rather than by Emby.
|
||||
const userDataItemField = "UserData"
|
||||
|
||||
// A launcher is a few hundred cards. This matches the ratings attach limit for the same
|
||||
// reason: it guards against a future row type asking for a thousand, not against anything
|
||||
// reached today.
|
||||
const viewerStateItemLimit = 600
|
||||
|
||||
// decorateViewerState replaces the UserData on every item with this viewer's own.
|
||||
//
|
||||
// It is the read half of what the four gated mutations are the write half of, and it rides
|
||||
// exactly where decorateItemRatings rides — one indexed read for a whole launcher, at every
|
||||
// point items leave the gateway. A card then draws the right progress bar as the row
|
||||
// appears, and nothing above this line has to know which viewer it is drawing for.
|
||||
//
|
||||
// Three things to preserve:
|
||||
//
|
||||
// A main viewer returns immediately. Their state is Emby's and is already on the payload,
|
||||
// so a household running no viewers pays one comparison for the whole launcher.
|
||||
//
|
||||
// **Every item is rewritten, not only the ones with something stored.** The UserData that
|
||||
// arrived from Emby is the *account's*, and leaving it in place on a title this viewer has
|
||||
// never touched is precisely the leak this feature exists to prevent: Alessandra would see
|
||||
// Matt's progress bars on everything neither of them had watched together. A title with no
|
||||
// row gets the zero state, which is the truth about it.
|
||||
//
|
||||
// A series and a season are aggregates, so they are rewritten from a *count* rather than
|
||||
// from a row of their own: Emby fills their block in from their children, and a shadow
|
||||
// viewer has no children Emby knows about. This was the last place they were still shown
|
||||
// the account's answer — a series ticked because somebody else had finished it.
|
||||
func (s *Server) decorateViewerState(ctx context.Context, collections ...[]json.RawMessage) {
|
||||
// The viewer is read from the context alone. Outside a request there is none, and the
|
||||
// zero session resolves to a main viewer, so a scheduled task or a test decorates
|
||||
// nothing rather than blanking what it was given.
|
||||
viewer := viewerOf(ctx, store.Session{})
|
||||
if viewer.IsMain() || s.store == nil {
|
||||
return
|
||||
}
|
||||
ids := itemIDsIn(collections, viewerStateItemLimit)
|
||||
if len(ids) == 0 {
|
||||
return
|
||||
}
|
||||
states, err := s.store.ViewerStates(ctx, viewer.ID, ids)
|
||||
if err != nil {
|
||||
// A state read that fails must not hand the viewer the account's watched state,
|
||||
// so every item is blanked rather than left as it arrived. A launcher with no
|
||||
// progress bars is a poor answer; one showing somebody else's is a wrong one.
|
||||
s.loggerFor(ctx).Warn("viewer state read failed", "error", err)
|
||||
states = map[string]store.ViewerState{}
|
||||
}
|
||||
// The aggregate half is a second read and is only paid for by a response that
|
||||
// actually carries a series or a season card. Its failure is the same failure the leaf
|
||||
// read has: an empty map, so every container is blanked rather than left carrying
|
||||
// somebody else's progress.
|
||||
seriesIDs, seasonIDs := containerIDsIn(collections)
|
||||
containers := map[string]store.ViewerAggregate{}
|
||||
if len(seriesIDs) > 0 || len(seasonIDs) > 0 {
|
||||
found, err := s.store.ViewerContainerStates(ctx, viewer.ID, seriesIDs, seasonIDs)
|
||||
if err != nil {
|
||||
s.loggerFor(ctx).Warn("viewer container state read failed", "error", err)
|
||||
} else {
|
||||
containers = found
|
||||
}
|
||||
}
|
||||
|
||||
for _, items := range collections {
|
||||
for index, raw := range items {
|
||||
id := itemIDOf(raw)
|
||||
if id == "" {
|
||||
continue
|
||||
}
|
||||
if isAggregateItem(raw) {
|
||||
items[index] = injectItemUserData(
|
||||
raw, viewerAggregateUserData(states[id], containers[id]),
|
||||
)
|
||||
continue
|
||||
}
|
||||
items[index] = injectItemUserData(raw, viewerUserData(states[id]))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// isAggregateItem reports whether an item's UserData describes its children rather than
|
||||
// itself. Emby fills a series' and a season's block in from their episodes.
|
||||
func isAggregateItem(raw json.RawMessage) bool {
|
||||
return itemTypeOf(raw) == "Series" || itemTypeOf(raw) == "Season"
|
||||
}
|
||||
|
||||
func itemTypeOf(raw json.RawMessage) string {
|
||||
var item struct {
|
||||
Type string `json:"Type"`
|
||||
}
|
||||
if json.Unmarshal(raw, &item) != nil {
|
||||
return ""
|
||||
}
|
||||
return item.Type
|
||||
}
|
||||
|
||||
// containerIDsIn collects the series and seasons a response is carrying.
|
||||
//
|
||||
// A **series** is keyed by its own id, because that is what its episodes carry as their
|
||||
// series id. A **season** cannot be: its episodes carry its id in their payload, but the
|
||||
// catalogue's indexed column is the series, so a season is looked up by its own id *and*
|
||||
// its series is asked for alongside — which is what makes one query answer for a season
|
||||
// card sitting on a page about a show the response also carries.
|
||||
func containerIDsIn(collections [][]json.RawMessage) (seriesIDs, seasonIDs []string) {
|
||||
seenSeries := map[string]bool{}
|
||||
seenSeasons := map[string]bool{}
|
||||
for _, items := range collections {
|
||||
for _, raw := range items {
|
||||
id := itemIDOf(raw)
|
||||
if id == "" {
|
||||
continue
|
||||
}
|
||||
switch itemTypeOf(raw) {
|
||||
case "Series":
|
||||
if !seenSeries[id] {
|
||||
seenSeries[id] = true
|
||||
seriesIDs = append(seriesIDs, id)
|
||||
}
|
||||
case "Season":
|
||||
if !seenSeasons[id] {
|
||||
seenSeasons[id] = true
|
||||
seasonIDs = append(seasonIDs, id)
|
||||
}
|
||||
if parent := seriesIDOf(raw); parent != "" && !seenSeries[parent] {
|
||||
seenSeries[parent] = true
|
||||
seriesIDs = append(seriesIDs, parent)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return seriesIDs, seasonIDs
|
||||
}
|
||||
|
||||
func seriesIDOf(raw json.RawMessage) string {
|
||||
var item struct {
|
||||
SeriesID string `json:"SeriesId"`
|
||||
}
|
||||
if json.Unmarshal(raw, &item) != nil {
|
||||
return ""
|
||||
}
|
||||
return item.SeriesID
|
||||
}
|
||||
|
||||
// injectItemUserData replaces one item's UserData block.
|
||||
//
|
||||
// It rewrites rather than merges: a partial overlay would leave whichever fields Memby had
|
||||
// nothing to say about carrying the account's values, which is the same leak from a
|
||||
// narrower angle.
|
||||
func injectItemUserData(raw, userData json.RawMessage) json.RawMessage {
|
||||
var members map[string]json.RawMessage
|
||||
if json.Unmarshal(raw, &members) != nil || members == nil {
|
||||
return raw
|
||||
}
|
||||
members[userDataItemField] = userData
|
||||
out, err := json.Marshal(members)
|
||||
if err != nil {
|
||||
return raw
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,414 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/ponzischeme89/memby/server/internal/store"
|
||||
)
|
||||
|
||||
// viewerHeader names the person watching, as distinct from the account streaming.
|
||||
//
|
||||
// It is a header rather than part of the session because a viewer does not belong to a
|
||||
// television: somebody starts an episode in the lounge and finishes it in the bedroom, and
|
||||
// switching between two people on one set must not be a re-authentication. The session
|
||||
// still answers "which Emby account is this and what may it read"; this answers "whose
|
||||
// evening is it", and the two are separate questions.
|
||||
const viewerHeader = "X-Memby-Viewer"
|
||||
|
||||
// The header is *stated, never inferred* — the stance Credentials.Gateway takes. An app
|
||||
// that predates viewers sends nothing and resolves to the account's main viewer, which is
|
||||
// exactly the behaviour it had before this existed; guessing from anything else would file
|
||||
// a household's ordinary watching under somebody who does not exist.
|
||||
|
||||
// How long an account's viewer list is trusted in memory.
|
||||
//
|
||||
// Every authenticated request resolves a viewer, and /v1/status alone is every open
|
||||
// television every ten seconds — a Postgres round trip each, for a list that changes when
|
||||
// somebody adds a person to the household. A write clears it, so the window is "how long
|
||||
// until another instance notices" rather than "how long until my change takes effect", the
|
||||
// bargain featurePolicyCache already makes.
|
||||
const viewerListTTL = 30 * time.Second
|
||||
|
||||
type viewerListCache struct {
|
||||
mu sync.Mutex
|
||||
entries map[string]viewerListEntry
|
||||
}
|
||||
|
||||
type viewerListEntry struct {
|
||||
viewers []store.Viewer
|
||||
loadedAt time.Time
|
||||
}
|
||||
|
||||
// viewersFor lists an account's viewers, from memory where it can.
|
||||
func (s *Server) viewersFor(ctx context.Context, sess store.Session) ([]store.Viewer, error) {
|
||||
if s.store == nil {
|
||||
return nil, errors.New("no store")
|
||||
}
|
||||
c := &s.viewerLists
|
||||
now := time.Now()
|
||||
c.mu.Lock()
|
||||
entry, ok := c.entries[sess.EmbyUserID]
|
||||
c.mu.Unlock()
|
||||
if ok && now.Sub(entry.loadedAt) < viewerListTTL {
|
||||
return entry.viewers, nil
|
||||
}
|
||||
|
||||
viewers, err := s.store.Viewers(ctx, sess.EmbyUserID, sess.Username)
|
||||
if err != nil {
|
||||
// A list that will not load is not evidence that the household has no viewers, so
|
||||
// a stale reading is preferred to none: losing it would silently move a shadow
|
||||
// viewer's playback back onto the Emby account, which is the one failure this
|
||||
// feature must never have.
|
||||
if ok {
|
||||
return entry.viewers, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
c.mu.Lock()
|
||||
if c.entries == nil {
|
||||
c.entries = map[string]viewerListEntry{}
|
||||
}
|
||||
c.entries[sess.EmbyUserID] = viewerListEntry{viewers: viewers, loadedAt: now}
|
||||
c.mu.Unlock()
|
||||
return viewers, nil
|
||||
}
|
||||
|
||||
// forgetViewers drops an account's cached list so a viewer added, renamed or removed is
|
||||
// live on the next request rather than at the end of the window.
|
||||
func (s *Server) forgetViewers(embyUserID string) {
|
||||
c := &s.viewerLists
|
||||
c.mu.Lock()
|
||||
delete(c.entries, embyUserID)
|
||||
c.mu.Unlock()
|
||||
}
|
||||
|
||||
// activeViewer resolves who is watching.
|
||||
//
|
||||
// Every unknown case resolves to the main viewer, and that is deliberate: this is on the
|
||||
// path of every authenticated request, and the failure it is protecting against — a
|
||||
// television left unable to do anything because a header could not be checked — is far
|
||||
// worse than a shadow viewer's episode being attributed to the account for one request.
|
||||
// The one thing it will not do is accept an id it could not confirm belongs to this
|
||||
// account, because that would let one household's television read another's viewing.
|
||||
func (s *Server) activeViewer(ctx context.Context, sess store.Session, r *http.Request) store.Viewer {
|
||||
fallback := store.Viewer{ID: sess.EmbyUserID, Name: sess.Username, Kind: store.ViewerMain}
|
||||
// The operator's switch is read here rather than at each of the four mutations,
|
||||
// because this is the one place a request learns who is watching: with it off there is
|
||||
// no shadow viewer to resolve to, so every branch downstream — the gated writes, the
|
||||
// substituted rows, the per-viewer cache keys — falls back to the account by
|
||||
// construction rather than by fifteen separate checks.
|
||||
if !s.viewersEnabled(ctx) {
|
||||
return fallback
|
||||
}
|
||||
requested := strings.TrimSpace(r.Header.Get(viewerHeader))
|
||||
if requested == "" || requested == sess.EmbyUserID {
|
||||
return fallback
|
||||
}
|
||||
viewers, err := s.viewersFor(ctx, sess)
|
||||
if err != nil {
|
||||
s.loggerFor(ctx).Warn("viewer list unavailable", "error", err)
|
||||
return fallback
|
||||
}
|
||||
for _, viewer := range viewers {
|
||||
if viewer.ID == requested {
|
||||
return viewer
|
||||
}
|
||||
}
|
||||
// An id this account does not own. It is logged rather than refused: the ordinary
|
||||
// cause is a television still holding a viewer somebody has since deleted, and
|
||||
// answering 403 to every request would leave that set unable to reach the picker that
|
||||
// would fix it.
|
||||
s.loggerFor(ctx).Warn("unknown viewer requested", "viewer", requested)
|
||||
return fallback
|
||||
}
|
||||
|
||||
// viewersEnabled reports whether the household is running viewers at all.
|
||||
//
|
||||
// Off is not a deletion. A viewer's rows stay in Postgres untouched and come back intact
|
||||
// when it is switched on again; what stops is the gateway routing anybody's watching
|
||||
// anywhere but Emby, which is exactly the state a household was in before this existed.
|
||||
func (s *Server) viewersEnabled(ctx context.Context) bool {
|
||||
return s.featureEnabled(ctx, featureViewers)
|
||||
}
|
||||
|
||||
// mainViewerOnly is what an account's list looks like with the feature switched off.
|
||||
//
|
||||
// It is a *shortened list* rather than an error or an empty one, because the television
|
||||
// decides whether to offer the picker by counting what it was sent: one viewer is an
|
||||
// account nobody has added anybody to, which is the reading that makes a switched-off
|
||||
// household look like one that never used the feature rather than like one whose picker
|
||||
// has broken.
|
||||
func mainViewerOnly(viewers []store.Viewer) []store.Viewer {
|
||||
for _, viewer := range viewers {
|
||||
if viewer.IsMain() {
|
||||
return []store.Viewer{viewer}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// --- carrying the viewer through one request --------------------------------
|
||||
|
||||
type viewerContextKey struct{}
|
||||
|
||||
// withViewer installs the resolved viewer for the rest of the request.
|
||||
//
|
||||
// It is resolved once, in [Server.authed], and read from the context everywhere else. The
|
||||
// alternative — every handler that needs a cache key calling activeViewer for itself —
|
||||
// is fifteen call sites that must each remember to, and the failure of forgetting one is
|
||||
// silent: the handler simply keys that view under the account, and one viewer is served
|
||||
// another viewer's rows. Resolving it at the boundary makes forgetting impossible.
|
||||
func withViewer(ctx context.Context, viewer store.Viewer) context.Context {
|
||||
return context.WithValue(ctx, viewerContextKey{}, viewer)
|
||||
}
|
||||
|
||||
// viewerOf reports who this request belongs to.
|
||||
//
|
||||
// A request with no viewer in its context is one that never passed through authed — a
|
||||
// scheduled task, a probe, a test — and the account's own id is the honest answer for it,
|
||||
// which is also the value every one of these keys held before viewers existed.
|
||||
func viewerOf(ctx context.Context, sess store.Session) store.Viewer {
|
||||
if viewer, ok := ctx.Value(viewerContextKey{}).(store.Viewer); ok && viewer.ID != "" {
|
||||
return viewer
|
||||
}
|
||||
return store.Viewer{ID: sess.EmbyUserID, Name: sess.Username, Kind: store.ViewerMain}
|
||||
}
|
||||
|
||||
// viewerKeyOf is the shorthand the cache keys use: the id everything about this person is
|
||||
// filed under. For the main viewer it is the Emby user id, so an existing household's
|
||||
// cached views keep the names they already had.
|
||||
func viewerKeyOf(ctx context.Context, sess store.Session) string {
|
||||
return viewerOf(ctx, sess).ID
|
||||
}
|
||||
|
||||
// viewerID is the key everything about a *person* is stored under — preferences,
|
||||
// notifications, followed shows, row statistics, recommendation profiles.
|
||||
//
|
||||
// For the main viewer it is the Emby user id, which is why this feature needed no
|
||||
// migration: an existing household's rows are already filed under exactly this value.
|
||||
func viewerID(viewer store.Viewer) string { return viewer.ID }
|
||||
|
||||
// --- the common state layer -------------------------------------------------
|
||||
|
||||
// viewerUserData renders one viewer's state in the shape of Emby's UserData block.
|
||||
//
|
||||
// This is the seam the client never sees. A television asks for a row and draws a progress
|
||||
// bar, a tick and a heart from UserData; whether that block came from Emby or from Postgres
|
||||
// is not a question anything above this line asks, which is what keeps viewers from
|
||||
// becoming a special case in every screen.
|
||||
func viewerUserData(state store.ViewerState) json.RawMessage {
|
||||
payload := map[string]any{
|
||||
"IsFavorite": state.Favourite,
|
||||
"Played": state.Played,
|
||||
"PlaybackPositionTicks": state.PositionTicks,
|
||||
"PlayCount": state.PlayCount,
|
||||
}
|
||||
if state.RuntimeTicks > 0 && state.PositionTicks > 0 {
|
||||
payload["PlayedPercentage"] = float64(state.PositionTicks) / float64(state.RuntimeTicks) * 100
|
||||
}
|
||||
if state.LastPlayedAt != nil {
|
||||
payload["LastPlayedDate"] = state.LastPlayedAt.UTC().Format(time.RFC3339)
|
||||
}
|
||||
raw, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return json.RawMessage(`{}`)
|
||||
}
|
||||
return raw
|
||||
}
|
||||
|
||||
// viewerAggregateUserData renders a series' or a season's block from a count of episodes.
|
||||
//
|
||||
// Emby fills those in from an item's children, and a shadow viewer has no children Emby
|
||||
// has ever heard of — so the count comes from the shared catalogue and this viewer's own
|
||||
// state. The favourite is the one field that is genuinely the container's own: somebody
|
||||
// marks a *show* a favourite, not the sum of its episodes, so it is read from the row
|
||||
// against the series id rather than derived.
|
||||
//
|
||||
// Two things are omitted rather than sent as zero, the rule the leaf block follows. A
|
||||
// container the catalogue cannot count for — a library not yet imported, a show it has
|
||||
// never seen — has no unwatched count, because "0 left" and "I cannot say" are different
|
||||
// answers and only one of them is true. And a container nothing has been watched of has no
|
||||
// last-played date.
|
||||
func viewerAggregateUserData(state store.ViewerState, aggregate store.ViewerAggregate) json.RawMessage {
|
||||
payload := map[string]any{
|
||||
"IsFavorite": state.Favourite,
|
||||
// A container is never resumable: what resumes is an episode, and Emby reports
|
||||
// zero here for the same reason.
|
||||
"PlaybackPositionTicks": 0,
|
||||
"PlayCount": aggregate.Played,
|
||||
// Played only where there is something to have finished. An empty catalogue must
|
||||
// not tick every show in the house.
|
||||
"Played": aggregate.Total > 0 && aggregate.Played >= aggregate.Total,
|
||||
}
|
||||
if aggregate.Total > 0 {
|
||||
unplayed := aggregate.Total - aggregate.Played
|
||||
if unplayed < 0 {
|
||||
unplayed = 0
|
||||
}
|
||||
payload["UnplayedItemCount"] = unplayed
|
||||
payload["PlayedPercentage"] = float64(aggregate.Played) / float64(aggregate.Total) * 100
|
||||
}
|
||||
if aggregate.LastPlayedAt != nil {
|
||||
payload["LastPlayedDate"] = aggregate.LastPlayedAt.UTC().Format(time.RFC3339)
|
||||
}
|
||||
raw, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return json.RawMessage(`{}`)
|
||||
}
|
||||
return raw
|
||||
}
|
||||
|
||||
// --- routes -----------------------------------------------------------------
|
||||
|
||||
type viewersResponse struct {
|
||||
Viewers []store.Viewer `json:"viewers"`
|
||||
Active string `json:"active"`
|
||||
}
|
||||
|
||||
type viewerRequest struct {
|
||||
Name string `json:"name"`
|
||||
ShortName string `json:"shortName"`
|
||||
Colour string `json:"colour"`
|
||||
}
|
||||
|
||||
func (s *Server) handleViewers(w http.ResponseWriter, r *http.Request, sess store.Session) {
|
||||
if r.Method == http.MethodPost {
|
||||
s.handleCreateViewer(w, r, sess)
|
||||
return
|
||||
}
|
||||
viewers, err := s.viewersFor(r.Context(), sess)
|
||||
if err != nil {
|
||||
s.writeUpstreamError(r.Context(), w, err, "could not load viewers")
|
||||
return
|
||||
}
|
||||
if !s.viewersEnabled(r.Context()) {
|
||||
viewers = mainViewerOnly(viewers)
|
||||
}
|
||||
writeJSON(w, http.StatusOK, viewersResponse{
|
||||
Viewers: viewers,
|
||||
Active: s.activeViewer(r.Context(), sess, r).ID,
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Server) handleCreateViewer(w http.ResponseWriter, r *http.Request, sess store.Session) {
|
||||
// A refusal rather than a silent success: the television is about to draw a card for
|
||||
// somebody, and an operator who has switched the feature off has said the household
|
||||
// does not use it. The wording names the reason, because a television has no log and
|
||||
// no support channel and that sentence is the whole diagnosis.
|
||||
if !s.viewersEnabled(r.Context()) {
|
||||
writeError(w, http.StatusForbidden, "viewers are switched off for this server")
|
||||
return
|
||||
}
|
||||
var req viewerRequest
|
||||
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<10)).Decode(&req); err != nil {
|
||||
writeError(w, http.StatusBadRequest, "malformed request body")
|
||||
return
|
||||
}
|
||||
if strings.TrimSpace(req.Name) == "" {
|
||||
writeError(w, http.StatusBadRequest, "a name is required")
|
||||
return
|
||||
}
|
||||
if len([]rune(req.Name)) > 40 {
|
||||
writeError(w, http.StatusBadRequest, "that name is too long")
|
||||
return
|
||||
}
|
||||
viewer, err := s.store.CreateShadowViewer(
|
||||
r.Context(), sess.EmbyUserID, req.Name, req.ShortName, req.Colour,
|
||||
)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadRequest, err.Error())
|
||||
return
|
||||
}
|
||||
s.forgetViewers(sess.EmbyUserID)
|
||||
s.loggerFor(r.Context()).Info("viewer added", "viewer", viewer.ID, "name", viewer.Name)
|
||||
writeJSON(w, http.StatusOK, viewer)
|
||||
}
|
||||
|
||||
func (s *Server) handleViewer(w http.ResponseWriter, r *http.Request, sess store.Session) {
|
||||
id := r.PathValue("viewerID")
|
||||
if id == "" {
|
||||
writeError(w, http.StatusBadRequest, "viewer id is required")
|
||||
return
|
||||
}
|
||||
if !s.viewersEnabled(r.Context()) {
|
||||
writeError(w, http.StatusForbidden, "viewers are switched off for this server")
|
||||
return
|
||||
}
|
||||
if r.Method == http.MethodDelete {
|
||||
if err := s.store.DeleteShadowViewer(r.Context(), sess.EmbyUserID, id); err != nil {
|
||||
if errors.Is(err, store.ErrViewerNotFound) {
|
||||
writeError(w, http.StatusNotFound, "no such viewer")
|
||||
return
|
||||
}
|
||||
s.writeUpstreamError(r.Context(), w, err, "could not remove that viewer")
|
||||
return
|
||||
}
|
||||
s.forgetViewers(sess.EmbyUserID)
|
||||
// Everything cached under this viewer's own key is now about nobody.
|
||||
if err := s.cache.InvalidateUser(r.Context(), id); err != nil {
|
||||
s.loggerFor(r.Context()).Warn("cache invalidation failed", "error", err)
|
||||
}
|
||||
s.loggerFor(r.Context()).Info("viewer removed", "viewer", id)
|
||||
writeJSON(w, http.StatusOK, map[string]bool{"removed": true})
|
||||
return
|
||||
}
|
||||
|
||||
var req viewerRequest
|
||||
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<10)).Decode(&req); err != nil {
|
||||
writeError(w, http.StatusBadRequest, "malformed request body")
|
||||
return
|
||||
}
|
||||
viewer, err := s.store.UpdateShadowViewer(
|
||||
r.Context(), sess.EmbyUserID, id, req.Name, req.ShortName, req.Colour,
|
||||
)
|
||||
if err != nil {
|
||||
if errors.Is(err, store.ErrViewerNotFound) {
|
||||
writeError(w, http.StatusNotFound, "no such viewer")
|
||||
return
|
||||
}
|
||||
writeError(w, http.StatusBadRequest, err.Error())
|
||||
return
|
||||
}
|
||||
s.forgetViewers(sess.EmbyUserID)
|
||||
writeJSON(w, http.StatusOK, viewer)
|
||||
}
|
||||
|
||||
// invalidateAccountViews drops the cached views of every viewer on an account.
|
||||
//
|
||||
// Signing a television out, or an operator resetting somebody, is a statement about the
|
||||
// account rather than about whoever happened to be watching — so invalidating the account's
|
||||
// own key alone would leave each shadow viewer's rows behind, to be served intact to the
|
||||
// next person who signs in on that set.
|
||||
//
|
||||
// The list is read directly rather than through the cache, because this is called at
|
||||
// exactly the moments the cached copy is least trustworthy, and it is best-effort: the
|
||||
// entries it misses expire on their own TTL, and nothing here is worth failing a sign-out
|
||||
// over.
|
||||
func (s *Server) invalidateAccountViews(ctx context.Context, sess store.Session) {
|
||||
if err := s.cache.InvalidateUser(ctx, sess.EmbyUserID); err != nil {
|
||||
s.loggerFor(ctx).Warn("cache invalidation failed", "error", err)
|
||||
}
|
||||
if s.store == nil {
|
||||
return
|
||||
}
|
||||
viewers, err := s.store.Viewers(ctx, sess.EmbyUserID, sess.Username)
|
||||
if err != nil {
|
||||
s.loggerFor(ctx).Warn("viewer list unavailable for invalidation", "error", err)
|
||||
return
|
||||
}
|
||||
for _, viewer := range viewers {
|
||||
if viewer.IsMain() {
|
||||
continue
|
||||
}
|
||||
if err := s.cache.InvalidateUser(ctx, viewer.ID); err != nil {
|
||||
s.loggerFor(ctx).Warn("viewer cache invalidation failed",
|
||||
"viewer", viewer.ID, "error", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,364 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/ponzischeme89/memby/server/internal/store"
|
||||
)
|
||||
|
||||
// The television has one UserData reader and it must not be able to tell where the block
|
||||
// came from. This pins the field names against the client's UserItemData, which is the
|
||||
// contract that keeps viewers from becoming a special case in every screen.
|
||||
func TestViewerUserDataIsShapedLikeEmbys(t *testing.T) {
|
||||
played := time.Date(2026, 8, 19, 21, 14, 0, 0, time.UTC)
|
||||
raw := viewerUserData(store.ViewerState{
|
||||
ItemID: "982173",
|
||||
PositionTicks: 15_420_000_000,
|
||||
RuntimeTicks: 30_840_000_000,
|
||||
PlayCount: 2,
|
||||
Favourite: true,
|
||||
LastPlayedAt: &played,
|
||||
})
|
||||
|
||||
var parsed struct {
|
||||
IsFavorite bool `json:"IsFavorite"`
|
||||
Played bool `json:"Played"`
|
||||
PlaybackPositionTicks int64 `json:"PlaybackPositionTicks"`
|
||||
PlayCount int `json:"PlayCount"`
|
||||
PlayedPercentage *float64 `json:"PlayedPercentage"`
|
||||
LastPlayedDate string `json:"LastPlayedDate"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &parsed); err != nil {
|
||||
t.Fatalf("decode viewer user data: %v", err)
|
||||
}
|
||||
if !parsed.IsFavorite {
|
||||
t.Error("favourite was not carried")
|
||||
}
|
||||
if parsed.Played {
|
||||
t.Error("a part-watched title was reported as played")
|
||||
}
|
||||
if parsed.PlaybackPositionTicks != 15_420_000_000 {
|
||||
t.Errorf("position = %d", parsed.PlaybackPositionTicks)
|
||||
}
|
||||
if parsed.PlayCount != 2 {
|
||||
t.Errorf("play count = %d", parsed.PlayCount)
|
||||
}
|
||||
if parsed.PlayedPercentage == nil || *parsed.PlayedPercentage < 49 || *parsed.PlayedPercentage > 51 {
|
||||
t.Errorf("played percentage = %v, want about 50", parsed.PlayedPercentage)
|
||||
}
|
||||
if parsed.LastPlayedDate != "2026-08-19T21:14:00Z" {
|
||||
t.Errorf("last played = %q", parsed.LastPlayedDate)
|
||||
}
|
||||
}
|
||||
|
||||
// A title nobody has touched has to render as untouched rather than as a card claiming a
|
||||
// zero-length progress bar, so the two optional fields are omitted rather than sent empty.
|
||||
func TestViewerUserDataOmitsWhatItDoesNotKnow(t *testing.T) {
|
||||
raw := viewerUserData(store.ViewerState{ItemID: "982173"})
|
||||
var fields map[string]any
|
||||
if err := json.Unmarshal(raw, &fields); err != nil {
|
||||
t.Fatalf("decode viewer user data: %v", err)
|
||||
}
|
||||
if _, ok := fields["PlayedPercentage"]; ok {
|
||||
t.Error("a percentage was claimed for a title with no runtime or position")
|
||||
}
|
||||
if _, ok := fields["LastPlayedDate"]; ok {
|
||||
t.Error("a play date was claimed for a title that has never been played")
|
||||
}
|
||||
if fields["Played"] != false || fields["IsFavorite"] != false {
|
||||
t.Errorf("untouched state rendered as %v", fields)
|
||||
}
|
||||
}
|
||||
|
||||
// IsMain is what every mutation branches on, so it is worth stating that it reads the
|
||||
// stored kind rather than guessing from the id.
|
||||
func TestViewerIsMainReadsTheStoredKind(t *testing.T) {
|
||||
if !(store.Viewer{ID: "abc", Kind: store.ViewerMain}).IsMain() {
|
||||
t.Error("a main viewer did not report as main")
|
||||
}
|
||||
if (store.Viewer{ID: "abc", Kind: store.ViewerShadow}).IsMain() {
|
||||
t.Error("a shadow viewer reported as main")
|
||||
}
|
||||
if (store.Viewer{ID: "abc"}).IsMain() {
|
||||
t.Error("a viewer with no kind reported as main")
|
||||
}
|
||||
}
|
||||
|
||||
// The ids are the answer, not just the selection: a shadow viewer's Continue Watching is
|
||||
// ordered by their own history in Postgres, and Emby answers an Ids= query in its own
|
||||
// order. Handing that back would keep the right titles and discard the reason for them.
|
||||
func TestOrderItemsByIDRestoresTheOrderAskedFor(t *testing.T) {
|
||||
item := func(id string) json.RawMessage {
|
||||
return json.RawMessage(`{"Id":"` + id + `","Name":"` + id + `"}`)
|
||||
}
|
||||
idsOf := func(items []json.RawMessage) []string {
|
||||
out := []string{}
|
||||
for _, raw := range items {
|
||||
out = append(out, itemIDOf(raw))
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
t.Run("emby's order is replaced", func(t *testing.T) {
|
||||
got := orderItemsByID(
|
||||
[]json.RawMessage{item("c"), item("a"), item("b")},
|
||||
[]string{"b", "c", "a"},
|
||||
)
|
||||
want := []string{"b", "c", "a"}
|
||||
if diff := idsOf(got); !equalStrings(diff, want) {
|
||||
t.Fatalf("order = %v, want %v", diff, want)
|
||||
}
|
||||
})
|
||||
|
||||
// A title the catalogue still names but Emby will not answer for leaves a card that
|
||||
// cannot be opened, so it is dropped instead.
|
||||
t.Run("a missing title is dropped", func(t *testing.T) {
|
||||
got := orderItemsByID([]json.RawMessage{item("a")}, []string{"a", "gone", "b"})
|
||||
if diff := idsOf(got); !equalStrings(diff, []string{"a"}) {
|
||||
t.Fatalf("order = %v, want [a]", diff)
|
||||
}
|
||||
})
|
||||
|
||||
// Every keyed list on the television throws on a repeated key, and a paging boundary
|
||||
// or a duplicated row is exactly where an id comes back twice.
|
||||
t.Run("a repeated id draws one card", func(t *testing.T) {
|
||||
got := orderItemsByID([]json.RawMessage{item("a"), item("a")}, []string{"a", "a"})
|
||||
if diff := idsOf(got); !equalStrings(diff, []string{"a"}) {
|
||||
t.Fatalf("order = %v, want one card", diff)
|
||||
}
|
||||
})
|
||||
|
||||
// Anything Emby volunteers that was not asked for is not part of the answer.
|
||||
t.Run("an unasked title is dropped", func(t *testing.T) {
|
||||
got := orderItemsByID([]json.RawMessage{item("a"), item("z")}, []string{"a"})
|
||||
if diff := idsOf(got); !equalStrings(diff, []string{"a"}) {
|
||||
t.Fatalf("order = %v, want [a]", diff)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// The account's UserData must be replaced outright rather than merged: a partial overlay
|
||||
// leaves whichever fields Memby had nothing to say about carrying the account's values.
|
||||
func TestInjectItemUserDataReplacesRatherThanMerges(t *testing.T) {
|
||||
raw := json.RawMessage(
|
||||
`{"Id":"1","Name":"Anatomy of a Fall",` +
|
||||
`"UserData":{"Played":true,"PlaybackPositionTicks":9999,"IsFavorite":true,"PlayCount":4}}`)
|
||||
out := injectItemUserData(raw, viewerUserData(store.ViewerState{ItemID: "1"}))
|
||||
|
||||
var parsed struct {
|
||||
Name string `json:"Name"`
|
||||
UserData struct {
|
||||
Played bool `json:"Played"`
|
||||
PlaybackPositionTicks int64 `json:"PlaybackPositionTicks"`
|
||||
IsFavorite bool `json:"IsFavorite"`
|
||||
PlayCount int `json:"PlayCount"`
|
||||
} `json:"UserData"`
|
||||
}
|
||||
if err := json.Unmarshal(out, &parsed); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
if parsed.Name != "Anatomy of a Fall" {
|
||||
t.Errorf("the rest of the item was disturbed: name = %q", parsed.Name)
|
||||
}
|
||||
if parsed.UserData.Played || parsed.UserData.IsFavorite {
|
||||
t.Error("the account's watched or favourite state survived")
|
||||
}
|
||||
if parsed.UserData.PlaybackPositionTicks != 0 || parsed.UserData.PlayCount != 0 {
|
||||
t.Errorf("the account's position or play count survived: %+v", parsed.UserData)
|
||||
}
|
||||
}
|
||||
|
||||
// A series and a season are answered from a count of episodes rather than from a row of
|
||||
// their own, so they have to be told apart from the leaf items around them.
|
||||
func TestAggregateItemsAreRecognised(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
itemType string
|
||||
want bool
|
||||
}{
|
||||
{"Series", true},
|
||||
{"Season", true},
|
||||
{"Episode", false},
|
||||
{"Movie", false},
|
||||
{"", false},
|
||||
} {
|
||||
raw := json.RawMessage(`{"Id":"1","Type":"` + tc.itemType + `"}`)
|
||||
if got := isAggregateItem(raw); got != tc.want {
|
||||
t.Errorf("isAggregateItem(%q) = %v, want %v", tc.itemType, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Everything downstream keys its cached views on this, so a request that never passed
|
||||
// through authed must still answer with the value those keys held before viewers existed.
|
||||
func TestViewerOfFallsBackToTheAccount(t *testing.T) {
|
||||
sess := store.Session{EmbyUserID: "emby-user-1", Username: "Matt"}
|
||||
viewer := viewerOf(context.Background(), sess)
|
||||
if !viewer.IsMain() || viewer.ID != "emby-user-1" {
|
||||
t.Fatalf("fallback viewer = %+v, want the account as main", viewer)
|
||||
}
|
||||
if got := viewerKeyOf(context.Background(), sess); got != "emby-user-1" {
|
||||
t.Fatalf("cache key = %q, want the Emby user id", got)
|
||||
}
|
||||
|
||||
shadow := store.Viewer{ID: "v0123", Name: "Alessandra", Kind: store.ViewerShadow}
|
||||
ctx := withViewer(context.Background(), shadow)
|
||||
if got := viewerKeyOf(ctx, sess); got != "v0123" {
|
||||
t.Fatalf("cache key = %q, want the shadow viewer", got)
|
||||
}
|
||||
}
|
||||
|
||||
func equalStrings(a, b []string) bool {
|
||||
if len(a) != len(b) {
|
||||
return false
|
||||
}
|
||||
for i := range a {
|
||||
if a[i] != b[i] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// The last place a shadow viewer was shown the account's answer. A series card draws its
|
||||
// tick and its "left to watch" from these fields, and Emby fills them in from children it
|
||||
// has never heard of for this person.
|
||||
func TestViewerAggregateUserDataCountsWhatIsLeft(t *testing.T) {
|
||||
played := time.Date(2026, 8, 18, 20, 5, 0, 0, time.UTC)
|
||||
raw := viewerAggregateUserData(
|
||||
store.ViewerState{ItemID: "series-1", Favourite: true},
|
||||
store.ViewerAggregate{Total: 10, Played: 4, LastPlayedAt: &played},
|
||||
)
|
||||
|
||||
var parsed struct {
|
||||
IsFavorite bool `json:"IsFavorite"`
|
||||
Played bool `json:"Played"`
|
||||
PlaybackPositionTicks int64 `json:"PlaybackPositionTicks"`
|
||||
UnplayedItemCount *int `json:"UnplayedItemCount"`
|
||||
PlayedPercentage *float64 `json:"PlayedPercentage"`
|
||||
LastPlayedDate string `json:"LastPlayedDate"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &parsed); err != nil {
|
||||
t.Fatalf("decode aggregate user data: %v", err)
|
||||
}
|
||||
// The favourite is the container's own — somebody marks a show, not the sum of its
|
||||
// episodes — so it comes from the row rather than from the count.
|
||||
if !parsed.IsFavorite {
|
||||
t.Error("the viewer's own favourite on the series was dropped")
|
||||
}
|
||||
if parsed.Played {
|
||||
t.Error("a part-watched series reported as finished")
|
||||
}
|
||||
if parsed.UnplayedItemCount == nil || *parsed.UnplayedItemCount != 6 {
|
||||
t.Errorf("unplayed = %v, want 6", parsed.UnplayedItemCount)
|
||||
}
|
||||
if parsed.PlayedPercentage == nil || *parsed.PlayedPercentage < 39 || *parsed.PlayedPercentage > 41 {
|
||||
t.Errorf("played percentage = %v, want about 40", parsed.PlayedPercentage)
|
||||
}
|
||||
// A container is never resumable; what resumes is an episode.
|
||||
if parsed.PlaybackPositionTicks != 0 {
|
||||
t.Errorf("a series carried a resume position: %d", parsed.PlaybackPositionTicks)
|
||||
}
|
||||
if parsed.LastPlayedDate != "2026-08-18T20:05:00Z" {
|
||||
t.Errorf("last played = %q", parsed.LastPlayedDate)
|
||||
}
|
||||
|
||||
finished := viewerAggregateUserData(
|
||||
store.ViewerState{}, store.ViewerAggregate{Total: 10, Played: 10},
|
||||
)
|
||||
var done map[string]any
|
||||
if err := json.Unmarshal(finished, &done); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
if done["Played"] != true {
|
||||
t.Errorf("a fully watched series did not report as played: %v", done)
|
||||
}
|
||||
if count, ok := done["UnplayedItemCount"].(float64); !ok || count != 0 {
|
||||
t.Errorf("unplayed on a finished series = %v, want 0", done["UnplayedItemCount"])
|
||||
}
|
||||
}
|
||||
|
||||
// "Nothing left to watch" and "I cannot say how much there is" are different answers, and
|
||||
// only one of them is true for a library the catalogue has not imported yet. Ticking every
|
||||
// show in the house is the worst thing this could do.
|
||||
func TestViewerAggregateUserDataSaysNothingItCannotCount(t *testing.T) {
|
||||
raw := viewerAggregateUserData(store.ViewerState{}, store.ViewerAggregate{})
|
||||
var fields map[string]any
|
||||
if err := json.Unmarshal(raw, &fields); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
if _, ok := fields["UnplayedItemCount"]; ok {
|
||||
t.Error("a count was claimed for a series the catalogue cannot count")
|
||||
}
|
||||
if _, ok := fields["PlayedPercentage"]; ok {
|
||||
t.Error("a percentage was claimed with nothing to divide by")
|
||||
}
|
||||
if _, ok := fields["LastPlayedDate"]; ok {
|
||||
t.Error("a play date was claimed for a series nobody has watched")
|
||||
}
|
||||
if fields["Played"] != false {
|
||||
t.Errorf("an uncountable series reported as watched: %v", fields)
|
||||
}
|
||||
}
|
||||
|
||||
// A season is looked up by its own id, and its series is asked for alongside it, because
|
||||
// the catalogue's indexed column is the series. Getting that wrong costs the season card
|
||||
// its count on exactly the page — a series detail page — where seasons appear.
|
||||
func TestContainerIDsCollectSeriesAndSeasons(t *testing.T) {
|
||||
items := []json.RawMessage{
|
||||
json.RawMessage(`{"Id":"ep-1","Type":"Episode","SeriesId":"show-1"}`),
|
||||
json.RawMessage(`{"Id":"show-1","Type":"Series"}`),
|
||||
json.RawMessage(`{"Id":"season-2","Type":"Season","SeriesId":"show-2"}`),
|
||||
json.RawMessage(`{"Id":"season-2","Type":"Season","SeriesId":"show-2"}`),
|
||||
json.RawMessage(`{"Id":"film-1","Type":"Movie"}`),
|
||||
}
|
||||
seriesIDs, seasonIDs := containerIDsIn([][]json.RawMessage{items})
|
||||
|
||||
if !equalStrings(seasonIDs, []string{"season-2"}) {
|
||||
t.Errorf("seasons = %v, want one season and no repeat", seasonIDs)
|
||||
}
|
||||
// show-2 is there because the season named it; show-1 because it is a card in its own
|
||||
// right. Neither the episode nor the film contributes a container.
|
||||
if !equalStrings(seriesIDs, []string{"show-1", "show-2"}) {
|
||||
t.Errorf("series = %v, want the series card and the season's parent", seriesIDs)
|
||||
}
|
||||
}
|
||||
|
||||
// Switched off, an account looks like one nobody has added anybody to — which is how the
|
||||
// television decides not to offer the picker. It must never look like an account whose
|
||||
// list failed to load.
|
||||
func TestMainViewerOnlyLeavesTheAccount(t *testing.T) {
|
||||
viewers := []store.Viewer{
|
||||
{ID: "emby-user-1", Name: "Matt", Kind: store.ViewerMain},
|
||||
{ID: "v01", Name: "Alessandra", Kind: store.ViewerShadow},
|
||||
{ID: "v02", Name: "Guest", Kind: store.ViewerShadow},
|
||||
}
|
||||
got := mainViewerOnly(viewers)
|
||||
if len(got) != 1 || !got[0].IsMain() || got[0].ID != "emby-user-1" {
|
||||
t.Fatalf("switched-off list = %+v, want the account alone", got)
|
||||
}
|
||||
if mainViewerOnly(nil) != nil {
|
||||
t.Error("a list with no main viewer invented one")
|
||||
}
|
||||
}
|
||||
|
||||
// The switch is the operator's and rides the ordinary feature machinery, so what is worth
|
||||
// pinning is that it is *in* the catalogue and gated on a capability — a household half of
|
||||
// whose televisions cannot choose between people must not be offered it.
|
||||
func TestViewersIsAnOperatorFeature(t *testing.T) {
|
||||
definition, ok := knownFeature(featureViewers)
|
||||
if !ok {
|
||||
t.Fatal("viewers is not in the feature catalogue")
|
||||
}
|
||||
if definition.Capability != "viewers_v1" {
|
||||
t.Errorf("capability = %q, want viewers_v1", definition.Capability)
|
||||
}
|
||||
// Off by default, and deliberately so: this is the switch deciding where a household's
|
||||
// watched state is written, and a feature that arrives already on is one every server
|
||||
// running this build starts using before anybody decided to.
|
||||
if definition.DefaultEnabled {
|
||||
t.Error("viewers defaults on; it is opted into rather than out of")
|
||||
}
|
||||
}
|
||||
@@ -111,17 +111,20 @@ func appendField(target []field, groups []string, attr slog.Attr) []field {
|
||||
// Anything unranked keeps the order the caller wrote it in, which is usually the order
|
||||
// that reads best for that particular event.
|
||||
var fieldRank = map[string]int{
|
||||
"component": 1,
|
||||
"user": 2,
|
||||
"device": 3,
|
||||
"client": 4,
|
||||
"correlation": 5,
|
||||
"play_session_id": 6,
|
||||
"protocol": 7,
|
||||
"method": 8,
|
||||
"path": 9,
|
||||
"status": 10,
|
||||
"duration": 11,
|
||||
"component": 1,
|
||||
"user": 2,
|
||||
// The person watching sits beside the account they watch through, because on a
|
||||
// household running viewers those are different answers and the line has to give both.
|
||||
"viewer": 3,
|
||||
"device": 4,
|
||||
"client": 5,
|
||||
"correlation": 6,
|
||||
"play_session_id": 7,
|
||||
"protocol": 8,
|
||||
"method": 9,
|
||||
"path": 10,
|
||||
"status": 11,
|
||||
"duration": 12,
|
||||
// Constant per process, so it belongs at the end of the line rather than in front
|
||||
// of the fields that differ between events.
|
||||
"version": 900,
|
||||
|
||||
@@ -108,6 +108,12 @@ CREATE INDEX IF NOT EXISTS library_items_genres_idx ON library_items USING GIN (
|
||||
CREATE INDEX IF NOT EXISTS library_items_type_created_idx ON library_items (type, date_created DESC);
|
||||
CREATE INDEX IF NOT EXISTS library_items_synced_idx ON library_items (synced_at);
|
||||
|
||||
-- Every episode of one series, which is what a viewer's Next Up walks and what a series
|
||||
-- card's watched count is computed from. Both run on the tail of an ordinary request, and
|
||||
-- without this each is a scan of every episode in the library.
|
||||
CREATE INDEX IF NOT EXISTS library_items_series_episodes_idx
|
||||
ON library_items (series_id) WHERE type = 'Episode';
|
||||
|
||||
-- Durable raw MDBList responses. Source selection and display formatting happen at read
|
||||
-- time, so changing the visible sources does not require another external API request.
|
||||
CREATE TABLE IF NOT EXISTS external_media_ratings (
|
||||
@@ -903,3 +909,76 @@ CREATE INDEX IF NOT EXISTS notification_log_user_idx
|
||||
ON notification_log (emby_user_id, occurred_at DESC) WHERE emby_user_id <> '';
|
||||
CREATE INDEX IF NOT EXISTS notification_log_status_idx ON notification_log (status, occurred_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS notification_log_kind_idx ON notification_log (kind, occurred_at DESC);
|
||||
|
||||
-- Viewers: the people using one Memby account.
|
||||
--
|
||||
-- A Memby account is the household's relationship with an Emby user; a viewer is one
|
||||
-- person under it. Every account has exactly one MAIN viewer, whose state is Emby's and
|
||||
-- which behaves exactly as the account did before viewers existed, and any number of
|
||||
-- SHADOW viewers whose state is Memby's alone.
|
||||
--
|
||||
-- The main viewer's id IS the Emby user id, and that is the whole of why this feature
|
||||
-- needed no migration. Every table in this schema keys a person by a bare emby_user_id
|
||||
-- with no foreign key behind it, so substituting a viewer id for it leaves an existing
|
||||
-- household's preferences, notifications, followed shows, search history and row stats
|
||||
-- exactly where they were. A shadow id is prefixed 'v' and is therefore distinguishable
|
||||
-- from Emby's 32-hex GUIDs by inspection, which is what makes that substitution safe.
|
||||
CREATE TABLE IF NOT EXISTS viewers (
|
||||
id TEXT PRIMARY KEY,
|
||||
emby_user_id TEXT NOT NULL,
|
||||
name TEXT NOT NULL,
|
||||
short_name TEXT NOT NULL DEFAULT '',
|
||||
colour TEXT NOT NULL DEFAULT '',
|
||||
kind TEXT NOT NULL, -- main | shadow
|
||||
pin_hash BYTEA,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS viewers_account_idx ON viewers (emby_user_id, created_at);
|
||||
|
||||
-- One main viewer per account, enforced rather than assumed: the main viewer is what a
|
||||
-- request falls back to, so an account with two of them would resolve differently
|
||||
-- depending on which row a query happened to return first.
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS viewers_account_main_idx
|
||||
ON viewers (emby_user_id) WHERE kind = 'main';
|
||||
|
||||
-- A shadow viewer's own viewing state, in the shape of the Emby UserData block it stands
|
||||
-- in for. Only the fields Memby actually renders are here: the Emby item id is the common
|
||||
-- identifier, so no library metadata is duplicated and nothing here needs invalidating
|
||||
-- when the catalogue changes.
|
||||
--
|
||||
-- There is deliberately no row for a main viewer. Their state lives in Emby, and a copy
|
||||
-- of it here would be a second answer free to disagree with the one the household's other
|
||||
-- Emby clients see.
|
||||
CREATE TABLE IF NOT EXISTS viewer_playback_state (
|
||||
viewer_id TEXT NOT NULL,
|
||||
item_id TEXT NOT NULL,
|
||||
series_id TEXT NOT NULL DEFAULT '',
|
||||
season_id TEXT NOT NULL DEFAULT '',
|
||||
position_ticks BIGINT NOT NULL DEFAULT 0,
|
||||
runtime_ticks BIGINT NOT NULL DEFAULT 0,
|
||||
played BOOLEAN NOT NULL DEFAULT false,
|
||||
play_count INT NOT NULL DEFAULT 0,
|
||||
favourite BOOLEAN NOT NULL DEFAULT false,
|
||||
hidden_from_resume BOOLEAN NOT NULL DEFAULT false,
|
||||
last_played_at TIMESTAMPTZ,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (viewer_id, item_id)
|
||||
);
|
||||
|
||||
-- Continue Watching for a shadow viewer is this index: what they are part-way through,
|
||||
-- most recent first. The partial predicate keeps it to the rows that row can draw from
|
||||
-- rather than to everything they have ever pressed Play on.
|
||||
CREATE INDEX IF NOT EXISTS viewer_playback_resume_idx
|
||||
ON viewer_playback_state (viewer_id, last_played_at DESC)
|
||||
WHERE position_ticks > 0 AND NOT played AND NOT hidden_from_resume;
|
||||
|
||||
-- Next Up walks a series' episodes for the newest completion; favourites are their own
|
||||
-- row, and both are asked for per viewer.
|
||||
CREATE INDEX IF NOT EXISTS viewer_playback_series_idx
|
||||
ON viewer_playback_state (viewer_id, series_id, last_played_at DESC)
|
||||
WHERE series_id <> '';
|
||||
|
||||
CREATE INDEX IF NOT EXISTS viewer_playback_favourite_idx
|
||||
ON viewer_playback_state (viewer_id, updated_at DESC) WHERE favourite;
|
||||
|
||||
@@ -0,0 +1,508 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// ViewerState is one viewer's answer about one title, in the shape of the Emby UserData
|
||||
// block it stands in for. Zero values are the honest answer for a title nobody has
|
||||
// touched, which is what lets a caller decorate an item it found nothing stored for
|
||||
// without a special case.
|
||||
type ViewerState struct {
|
||||
ItemID string
|
||||
SeriesID string
|
||||
SeasonID string
|
||||
PositionTicks int64
|
||||
RuntimeTicks int64
|
||||
Played bool
|
||||
PlayCount int
|
||||
Favourite bool
|
||||
HiddenFromResume bool
|
||||
LastPlayedAt *time.Time
|
||||
}
|
||||
|
||||
// PlayedFraction is the share of a title that must be behind the viewer for it to count as
|
||||
// watched. It matches Emby's own default so a household cannot come to disagree with itself
|
||||
// about whether an episode is finished depending on which viewer watched it.
|
||||
const PlayedFraction = 0.9
|
||||
|
||||
// PlayedFromPosition decides whether a stop report completed the title.
|
||||
//
|
||||
// A runtime of zero means the length was not known rather than that the title is zero
|
||||
// long, so it can never complete anything — the alternative is that every report with a
|
||||
// missing duration marks something watched at the first second.
|
||||
func PlayedFromPosition(positionTicks, runtimeTicks int64) bool {
|
||||
if runtimeTicks <= 0 || positionTicks <= 0 {
|
||||
return false
|
||||
}
|
||||
return float64(positionTicks) >= float64(runtimeTicks)*PlayedFraction
|
||||
}
|
||||
|
||||
// RecordViewerPlayback writes a progress or stop report for a shadow viewer.
|
||||
//
|
||||
// A completed title is stored at position zero, the way Emby stores one: the position is
|
||||
// what Continue Watching reads, and a finished episode left sitting at its last frame is
|
||||
// one the row keeps offering to resume four seconds from the end. play_count only moves on
|
||||
// the transition into played, so the ten-second reports either side of the threshold cannot
|
||||
// count one viewing several times.
|
||||
func (s *Store) RecordViewerPlayback(ctx context.Context, viewerID string, state ViewerState) error {
|
||||
if viewerID == "" || state.ItemID == "" {
|
||||
return fmt.Errorf("store: viewer playback: viewer and item are required")
|
||||
}
|
||||
position := state.PositionTicks
|
||||
if position < 0 {
|
||||
position = 0
|
||||
}
|
||||
runtime := state.RuntimeTicks
|
||||
if runtime < 0 {
|
||||
runtime = 0
|
||||
}
|
||||
if state.Played {
|
||||
position = 0
|
||||
}
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO viewer_playback_state (
|
||||
viewer_id, item_id, series_id, season_id,
|
||||
position_ticks, runtime_ticks, played, play_count, last_played_at, updated_at
|
||||
) VALUES (
|
||||
$1, $2,
|
||||
-- The series is read out of the shared catalogue rather than asked of Emby or
|
||||
-- carried by the television: it is already there, it is what orders this
|
||||
-- viewer's Continue Watching, and a report arrives every ten seconds.
|
||||
COALESCE(NULLIF($3, ''), (SELECT series_id FROM library_items WHERE id = $2), ''),
|
||||
$4, $5, $6, $7, CASE WHEN $7 THEN 1 ELSE 0 END, now(), now())
|
||||
ON CONFLICT (viewer_id, item_id) DO UPDATE SET
|
||||
series_id = CASE WHEN excluded.series_id <> '' THEN excluded.series_id
|
||||
ELSE viewer_playback_state.series_id END,
|
||||
season_id = CASE WHEN excluded.season_id <> '' THEN excluded.season_id
|
||||
ELSE viewer_playback_state.season_id END,
|
||||
position_ticks = excluded.position_ticks,
|
||||
runtime_ticks = CASE WHEN excluded.runtime_ticks > 0 THEN excluded.runtime_ticks
|
||||
ELSE viewer_playback_state.runtime_ticks END,
|
||||
played = excluded.played,
|
||||
play_count = viewer_playback_state.play_count
|
||||
+ CASE WHEN excluded.played AND NOT viewer_playback_state.played
|
||||
THEN 1 ELSE 0 END,
|
||||
last_played_at = now(),
|
||||
updated_at = now()`,
|
||||
viewerID, state.ItemID, state.SeriesID, state.SeasonID,
|
||||
position, runtime, state.Played,
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: record viewer playback: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// SetViewerPlayed marks a title watched or unwatched by hand.
|
||||
//
|
||||
// Marking unwatched clears the position for the same reason marking watched does: the two
|
||||
// are one statement about where this viewer stands with the title, and a cleared flag over
|
||||
// a retained playhead would put it straight back into Continue Watching at the closing
|
||||
// credits.
|
||||
func (s *Store) SetViewerPlayed(ctx context.Context, viewerID, itemID string, played bool) error {
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO viewer_playback_state (
|
||||
viewer_id, item_id, position_ticks, played, play_count, last_played_at, updated_at
|
||||
) VALUES ($1, $2, 0, $3, CASE WHEN $3 THEN 1 ELSE 0 END,
|
||||
CASE WHEN $3 THEN now() ELSE NULL END, now())
|
||||
ON CONFLICT (viewer_id, item_id) DO UPDATE SET
|
||||
position_ticks = 0,
|
||||
played = excluded.played,
|
||||
play_count = viewer_playback_state.play_count
|
||||
+ CASE WHEN excluded.played AND NOT viewer_playback_state.played
|
||||
THEN 1 ELSE 0 END,
|
||||
last_played_at = CASE WHEN excluded.played
|
||||
THEN COALESCE(viewer_playback_state.last_played_at, now())
|
||||
ELSE viewer_playback_state.last_played_at END,
|
||||
updated_at = now()`,
|
||||
viewerID, itemID, played)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: set viewer played: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// SetViewerFavourite records a favourite that belongs to the person rather than to the
|
||||
// Emby account, so one viewer's heart cannot appear on everybody else's launcher.
|
||||
func (s *Store) SetViewerFavourite(ctx context.Context, viewerID, itemID string, favourite bool) error {
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO viewer_playback_state (viewer_id, item_id, favourite, updated_at)
|
||||
VALUES ($1, $2, $3, now())
|
||||
ON CONFLICT (viewer_id, item_id) DO UPDATE SET
|
||||
favourite = excluded.favourite, updated_at = now()`,
|
||||
viewerID, itemID, favourite)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: set viewer favourite: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// HideViewerFromResume takes a title off this viewer's Continue Watching without claiming
|
||||
// they watched it. The position is kept: hiding is a statement about the row, not about
|
||||
// where they got to, and pressing Play again should still resume.
|
||||
func (s *Store) HideViewerFromResume(ctx context.Context, viewerID, itemID string) error {
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO viewer_playback_state (viewer_id, item_id, hidden_from_resume, updated_at)
|
||||
VALUES ($1, $2, true, now())
|
||||
ON CONFLICT (viewer_id, item_id) DO UPDATE SET
|
||||
hidden_from_resume = true, updated_at = now()`,
|
||||
viewerID, itemID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: hide from resume: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ViewerStateFor reads one title's state. A title with no row is not an error: it is a
|
||||
// title this viewer has never touched, which is the ordinary case.
|
||||
func (s *Store) ViewerStateFor(ctx context.Context, viewerID, itemID string) (ViewerState, error) {
|
||||
state := ViewerState{ItemID: itemID}
|
||||
err := s.pool.QueryRow(ctx, `
|
||||
SELECT series_id, season_id, position_ticks, runtime_ticks,
|
||||
played, play_count, favourite, hidden_from_resume, last_played_at
|
||||
FROM viewer_playback_state WHERE viewer_id = $1 AND item_id = $2`,
|
||||
viewerID, itemID,
|
||||
).Scan(
|
||||
&state.SeriesID, &state.SeasonID, &state.PositionTicks, &state.RuntimeTicks,
|
||||
&state.Played, &state.PlayCount, &state.Favourite, &state.HiddenFromResume,
|
||||
&state.LastPlayedAt,
|
||||
)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return state, nil
|
||||
}
|
||||
if err != nil {
|
||||
return ViewerState{}, fmt.Errorf("store: viewer state: %w", err)
|
||||
}
|
||||
return state, nil
|
||||
}
|
||||
|
||||
// ViewerStates reads a whole launcher's worth in one query.
|
||||
//
|
||||
// This is the read behind every decorated row, so it is one indexed lookup for several
|
||||
// hundred cards rather than a request per card — the economy decorateItemRatings already
|
||||
// makes for scores.
|
||||
func (s *Store) ViewerStates(
|
||||
ctx context.Context, viewerID string, itemIDs []string,
|
||||
) (map[string]ViewerState, error) {
|
||||
states := map[string]ViewerState{}
|
||||
if viewerID == "" || len(itemIDs) == 0 {
|
||||
return states, nil
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT item_id, series_id, season_id, position_ticks, runtime_ticks,
|
||||
played, play_count, favourite, hidden_from_resume, last_played_at
|
||||
FROM viewer_playback_state
|
||||
WHERE viewer_id = $1 AND item_id = ANY($2)`, viewerID, itemIDs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: viewer states: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var state ViewerState
|
||||
if err := rows.Scan(
|
||||
&state.ItemID, &state.SeriesID, &state.SeasonID,
|
||||
&state.PositionTicks, &state.RuntimeTicks,
|
||||
&state.Played, &state.PlayCount, &state.Favourite, &state.HiddenFromResume,
|
||||
&state.LastPlayedAt,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("store: scan viewer state: %w", err)
|
||||
}
|
||||
states[state.ItemID] = state
|
||||
}
|
||||
return states, rows.Err()
|
||||
}
|
||||
|
||||
// ViewerResumeItems is this viewer's Continue Watching, most recently played first.
|
||||
//
|
||||
// It answers in item ids alone: the catalogue is shared by the household and is read from
|
||||
// library_items or Emby, so duplicating a single field of metadata here would be a second
|
||||
// copy free to go stale.
|
||||
func (s *Store) ViewerResumeItems(ctx context.Context, viewerID string, limit int) ([]string, error) {
|
||||
if viewerID == "" {
|
||||
return nil, nil
|
||||
}
|
||||
if limit <= 0 {
|
||||
limit = 20
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT item_id FROM viewer_playback_state
|
||||
WHERE viewer_id = $1 AND position_ticks > 0 AND NOT played AND NOT hidden_from_resume
|
||||
ORDER BY last_played_at DESC NULLS LAST
|
||||
LIMIT $2`, viewerID, limit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: viewer resume items: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
ids := []string{}
|
||||
for rows.Next() {
|
||||
var id string
|
||||
if err := rows.Scan(&id); err != nil {
|
||||
return nil, fmt.Errorf("store: scan resume item: %w", err)
|
||||
}
|
||||
ids = append(ids, id)
|
||||
}
|
||||
return ids, rows.Err()
|
||||
}
|
||||
|
||||
// ViewerFavouriteItems is this viewer's favourites, most recently marked first.
|
||||
func (s *Store) ViewerFavouriteItems(ctx context.Context, viewerID string, limit int) ([]string, error) {
|
||||
if viewerID == "" {
|
||||
return nil, nil
|
||||
}
|
||||
if limit <= 0 {
|
||||
limit = 50
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT item_id FROM viewer_playback_state
|
||||
WHERE viewer_id = $1 AND favourite
|
||||
ORDER BY updated_at DESC
|
||||
LIMIT $2`, viewerID, limit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: viewer favourites: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
ids := []string{}
|
||||
for rows.Next() {
|
||||
var id string
|
||||
if err := rows.Scan(&id); err != nil {
|
||||
return nil, fmt.Errorf("store: scan favourite: %w", err)
|
||||
}
|
||||
ids = append(ids, id)
|
||||
}
|
||||
return ids, rows.Err()
|
||||
}
|
||||
|
||||
// ViewerWatchedSeries reports, per series, when this viewer last finished or watched an
|
||||
// episode of it. It is what orders a shadow viewer's Continue Watching, which merges
|
||||
// resumable items with the next unwatched episode of a series they are part-way through —
|
||||
// and a Next Up episode has no time of its own, so it is placed by its series.
|
||||
func (s *Store) ViewerWatchedSeries(
|
||||
ctx context.Context, viewerID string, limit int,
|
||||
) (map[string]time.Time, error) {
|
||||
watched := map[string]time.Time{}
|
||||
if viewerID == "" {
|
||||
return watched, nil
|
||||
}
|
||||
if limit <= 0 {
|
||||
limit = 40
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT series_id, max(last_played_at) AS played_at
|
||||
FROM viewer_playback_state
|
||||
WHERE viewer_id = $1 AND series_id <> '' AND last_played_at IS NOT NULL
|
||||
GROUP BY series_id
|
||||
ORDER BY played_at DESC
|
||||
LIMIT $2`, viewerID, limit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: viewer watched series: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var seriesID string
|
||||
var playedAt time.Time
|
||||
if err := rows.Scan(&seriesID, &playedAt); err != nil {
|
||||
return nil, fmt.Errorf("store: scan watched series: %w", err)
|
||||
}
|
||||
watched[seriesID] = playedAt
|
||||
}
|
||||
return watched, rows.Err()
|
||||
}
|
||||
|
||||
// ViewerPlayedInSeries reports which of a series' episodes this viewer has finished, which
|
||||
// is what Next Up walks to find the first one they have not.
|
||||
func (s *Store) ViewerPlayedInSeries(
|
||||
ctx context.Context, viewerID, seriesID string,
|
||||
) (map[string]bool, error) {
|
||||
played := map[string]bool{}
|
||||
if viewerID == "" || seriesID == "" {
|
||||
return played, nil
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT item_id, played FROM viewer_playback_state
|
||||
WHERE viewer_id = $1 AND series_id = $2`, viewerID, seriesID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: viewer played in series: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var itemID string
|
||||
var done bool
|
||||
if err := rows.Scan(&itemID, &done); err != nil {
|
||||
return nil, fmt.Errorf("store: scan played episode: %w", err)
|
||||
}
|
||||
played[itemID] = done
|
||||
}
|
||||
return played, rows.Err()
|
||||
}
|
||||
|
||||
// ViewerNextUp is the next unwatched episode of every series this viewer is part-way
|
||||
// through, the series they watched most recently first.
|
||||
//
|
||||
// It is computed entirely in Postgres, out of the shared catalogue and this viewer's own
|
||||
// state, because Emby's own NextUp answers for the *account* and there is nobody else to
|
||||
// ask. That also makes it cheap: the alternative — walking each series' episode list over
|
||||
// the wire — is one Emby request per show on the tail of the launcher.
|
||||
//
|
||||
// Three rules, each of which Emby's own answer also applies:
|
||||
//
|
||||
// An episode already resumable is left out, because it is in Continue Watching already and
|
||||
// the merge would otherwise offer the same show twice.
|
||||
//
|
||||
// Specials are not next episodes. Season 0 is a real season and a perfectly good thing to
|
||||
// watch, but it is not what "next" means, and a show whose specials sort first would never
|
||||
// offer anything else.
|
||||
//
|
||||
// A series with nothing unwatched left simply contributes no row rather than an empty one.
|
||||
func (s *Store) ViewerNextUp(ctx context.Context, viewerID string, limit int) ([]string, error) {
|
||||
if viewerID == "" {
|
||||
return nil, nil
|
||||
}
|
||||
if limit <= 0 {
|
||||
limit = 20
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
WITH watched AS (
|
||||
SELECT series_id, max(last_played_at) AS played_at
|
||||
FROM viewer_playback_state
|
||||
WHERE viewer_id = $1 AND series_id <> '' AND last_played_at IS NOT NULL
|
||||
GROUP BY series_id
|
||||
),
|
||||
episodes AS (
|
||||
SELECT li.id,
|
||||
li.series_id,
|
||||
COALESCE((li.payload->>'ParentIndexNumber')::int, 0) AS season,
|
||||
COALESCE((li.payload->>'IndexNumber')::int, 0) AS episode,
|
||||
w.played_at
|
||||
FROM library_items li
|
||||
JOIN watched w ON w.series_id = li.series_id
|
||||
WHERE li.type = 'Episode'
|
||||
AND COALESCE((li.payload->>'ParentIndexNumber')::int, 0) > 0
|
||||
),
|
||||
unplayed AS (
|
||||
SELECT e.id, e.played_at,
|
||||
row_number() OVER (
|
||||
PARTITION BY e.series_id ORDER BY e.season, e.episode, e.id
|
||||
) AS rank
|
||||
FROM episodes e
|
||||
LEFT JOIN viewer_playback_state vps
|
||||
ON vps.viewer_id = $1 AND vps.item_id = e.id
|
||||
WHERE COALESCE(vps.played, false) = false
|
||||
AND COALESCE(vps.position_ticks, 0) = 0
|
||||
)
|
||||
SELECT id FROM unplayed WHERE rank = 1
|
||||
ORDER BY played_at DESC
|
||||
LIMIT $2`, viewerID, limit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: viewer next up: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
ids := []string{}
|
||||
for rows.Next() {
|
||||
var id string
|
||||
if err := rows.Scan(&id); err != nil {
|
||||
return nil, fmt.Errorf("store: scan next up: %w", err)
|
||||
}
|
||||
ids = append(ids, id)
|
||||
}
|
||||
return ids, rows.Err()
|
||||
}
|
||||
|
||||
// ViewerAggregate is what a series or a season card says about a viewer: how much of it
|
||||
// there is, how much of it is behind them, and when they last watched any of it.
|
||||
//
|
||||
// It stands in for the block Emby fills in from an item's children, which is the one place
|
||||
// a shadow viewer was still shown the *account's* answer — a series card ticked because
|
||||
// somebody else had finished it. Emby computes it per Emby user and there is nobody to ask
|
||||
// for a person Emby has never heard of, so it is computed here out of the shared catalogue
|
||||
// and this viewer's own state.
|
||||
type ViewerAggregate struct {
|
||||
// Total is how many episodes the catalogue holds. Zero means the catalogue cannot
|
||||
// answer — a library not yet imported, or a series it has never seen — which is a
|
||||
// different thing from a series with nothing in it, and the caller must not print a
|
||||
// count for it.
|
||||
Total int
|
||||
// Played is how many of those this viewer has finished.
|
||||
Played int
|
||||
// LastPlayedAt is the most recent episode of it they touched, finished or not, which
|
||||
// is what orders a shelf.
|
||||
LastPlayedAt *time.Time
|
||||
}
|
||||
|
||||
// ViewerContainerStates aggregates a viewer's episode state per series *and* per season.
|
||||
//
|
||||
// One map keyed by container id serves both, because a series id and a season id are both
|
||||
// Emby GUIDs and cannot collide — so the caller looks an item up by its own id and does not
|
||||
// have to know which of the two it is holding.
|
||||
//
|
||||
// The query groups by the **pair** and the two rollups are done here rather than in SQL.
|
||||
// That is a deliberately dull query — no grouping sets, no second pass over the same rows —
|
||||
// and it is exact for both answers because a season belongs to exactly one series, so
|
||||
// summing a series' seasons is summing its episodes. It runs on the tail of every request
|
||||
// that serves a series card, which is why the index it reads
|
||||
// (library_items_series_episodes_idx) exists.
|
||||
//
|
||||
// One thing to know about it: a series is only counted completely if it was *asked* for.
|
||||
// A season whose series was not in seriesIDs contributes to a partial series total, which
|
||||
// is harmless only because nothing looks that series up — containerIDsIn asks for a
|
||||
// season's series alongside it precisely so the case cannot arise for anything drawn.
|
||||
func (s *Store) ViewerContainerStates(
|
||||
ctx context.Context, viewerID string, seriesIDs, seasonIDs []string,
|
||||
) (map[string]ViewerAggregate, error) {
|
||||
aggregates := map[string]ViewerAggregate{}
|
||||
if viewerID == "" || (len(seriesIDs) == 0 && len(seasonIDs) == 0) {
|
||||
return aggregates, nil
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT li.series_id,
|
||||
COALESCE(li.payload->>'SeasonId', '') AS season_id,
|
||||
count(*) AS total,
|
||||
count(*) FILTER (WHERE COALESCE(vps.played, false)) AS played,
|
||||
max(vps.last_played_at) AS last_played_at
|
||||
FROM library_items li
|
||||
LEFT JOIN viewer_playback_state vps
|
||||
ON vps.viewer_id = $1 AND vps.item_id = li.id
|
||||
WHERE li.type = 'Episode'
|
||||
AND (li.series_id = ANY($2) OR COALESCE(li.payload->>'SeasonId', '') = ANY($3))
|
||||
GROUP BY li.series_id, COALESCE(li.payload->>'SeasonId', '')`,
|
||||
viewerID, seriesIDs, seasonIDs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: viewer container states: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var seriesID, seasonID string
|
||||
var total, played int
|
||||
var lastPlayedAt *time.Time
|
||||
if err := rows.Scan(&seriesID, &seasonID, &total, &played, &lastPlayedAt); err != nil {
|
||||
return nil, fmt.Errorf("store: scan container state: %w", err)
|
||||
}
|
||||
// An episode filed under no series or no season contributes to neither rather than
|
||||
// to a row keyed on the empty string, which would be an aggregate about nothing.
|
||||
addViewerAggregate(aggregates, seriesID, total, played, lastPlayedAt)
|
||||
addViewerAggregate(aggregates, seasonID, total, played, lastPlayedAt)
|
||||
}
|
||||
return aggregates, rows.Err()
|
||||
}
|
||||
|
||||
// addViewerAggregate folds one season's worth of counting into a container's total.
|
||||
func addViewerAggregate(
|
||||
into map[string]ViewerAggregate, key string, total, played int, lastPlayedAt *time.Time,
|
||||
) {
|
||||
if key == "" {
|
||||
return
|
||||
}
|
||||
aggregate := into[key]
|
||||
aggregate.Total += total
|
||||
aggregate.Played += played
|
||||
if lastPlayedAt != nil &&
|
||||
(aggregate.LastPlayedAt == nil || lastPlayedAt.After(*aggregate.LastPlayedAt)) {
|
||||
aggregate.LastPlayedAt = lastPlayedAt
|
||||
}
|
||||
into[key] = aggregate
|
||||
}
|
||||
@@ -0,0 +1,238 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// ViewerKind separates the one viewer whose state is Emby's from the ones whose state is
|
||||
// Memby's. It is stated on the row rather than derived from whether an id looks like an
|
||||
// Emby GUID: the id shape is a safety property, not a source of truth, and a household
|
||||
// that arrived at an odd id must not silently change which viewer publishes.
|
||||
const (
|
||||
ViewerMain = "main"
|
||||
ViewerShadow = "shadow"
|
||||
)
|
||||
|
||||
// ErrViewerNotFound is returned when an id names no viewer of the account that asked.
|
||||
var ErrViewerNotFound = errors.New("store: viewer not found")
|
||||
|
||||
// MaxShadowViewers bounds an account's list. A picker is a row of cards on a television
|
||||
// and the D-pad has to reach the end of it; this is a limit on the UI, not on the schema.
|
||||
const MaxShadowViewers = 7
|
||||
|
||||
type Viewer struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
ShortName string `json:"shortName,omitempty"`
|
||||
Colour string `json:"colour,omitempty"`
|
||||
Kind string `json:"kind"`
|
||||
HasPIN bool `json:"hasPin"`
|
||||
CreatedAt time.Time `json:"createdAt"`
|
||||
}
|
||||
|
||||
// IsMain reports whether this viewer's state is published to Emby.
|
||||
func (v Viewer) IsMain() bool { return v.Kind == ViewerMain }
|
||||
|
||||
// NewShadowViewerID mints an id that cannot be mistaken for an Emby user id.
|
||||
//
|
||||
// Emby's are 32 hex characters. This is a "v" followed by 32 more, so the two are
|
||||
// distinguishable by inspection anywhere one is read out of a log line or a cache key —
|
||||
// which matters because a viewer id is substituted for an emby_user_id in twenty tables
|
||||
// that cannot tell the difference themselves.
|
||||
func NewShadowViewerID() (string, error) {
|
||||
buf := make([]byte, 16)
|
||||
if _, err := rand.Read(buf); err != nil {
|
||||
return "", fmt.Errorf("store: viewer id: %w", err)
|
||||
}
|
||||
return "v" + hex.EncodeToString(buf), nil
|
||||
}
|
||||
|
||||
// IsShadowViewerID reports whether an id belongs to the shadow namespace. Callers holding
|
||||
// no viewer record use it to answer "is this Emby's user or Memby's" cheaply.
|
||||
func IsShadowViewerID(id string) bool {
|
||||
return strings.HasPrefix(id, "v") && len(id) == 33
|
||||
}
|
||||
|
||||
// Viewers lists an account's viewers, main first and the rest in the order they were
|
||||
// added. The main viewer is created on demand: an account that predates this feature has
|
||||
// no row, and its first request must still resolve to something rather than to an error.
|
||||
func (s *Store) Viewers(ctx context.Context, embyUserID, username string) ([]Viewer, error) {
|
||||
if err := s.ensureMainViewer(ctx, embyUserID, username); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT id, name, short_name, colour, kind, pin_hash IS NOT NULL, created_at
|
||||
FROM viewers WHERE emby_user_id = $1
|
||||
ORDER BY kind = 'main' DESC, created_at, id`, embyUserID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: list viewers: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
viewers := []Viewer{}
|
||||
for rows.Next() {
|
||||
var v Viewer
|
||||
if err := rows.Scan(
|
||||
&v.ID, &v.Name, &v.ShortName, &v.Colour, &v.Kind, &v.HasPIN, &v.CreatedAt,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("store: scan viewer: %w", err)
|
||||
}
|
||||
viewers = append(viewers, v)
|
||||
}
|
||||
return viewers, rows.Err()
|
||||
}
|
||||
|
||||
// ensureMainViewer records the account's own viewer if it has none.
|
||||
//
|
||||
// The insert is ON CONFLICT DO NOTHING on the primary key, so two televisions signing in
|
||||
// at once cannot both create it, and the name is only ever set on the way in: the viewer
|
||||
// may have been renamed since, and an Emby username arriving on every request must not
|
||||
// overwrite that.
|
||||
func (s *Store) ensureMainViewer(ctx context.Context, embyUserID, username string) error {
|
||||
if strings.TrimSpace(embyUserID) == "" {
|
||||
return fmt.Errorf("store: main viewer: no account")
|
||||
}
|
||||
name := strings.TrimSpace(username)
|
||||
if name == "" {
|
||||
name = "Me"
|
||||
}
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO viewers (id, emby_user_id, name, kind)
|
||||
VALUES ($1, $1, $2, 'main')
|
||||
ON CONFLICT (id) DO NOTHING`, embyUserID, name)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: ensure main viewer: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ViewerFor resolves one viewer *of this account*.
|
||||
//
|
||||
// The account is part of the query rather than checked afterwards: the id arrives in a
|
||||
// request header, so this is the boundary at which one household's television is stopped
|
||||
// from naming another household's viewer.
|
||||
func (s *Store) ViewerFor(ctx context.Context, embyUserID, viewerID string) (Viewer, error) {
|
||||
var v Viewer
|
||||
err := s.pool.QueryRow(ctx, `
|
||||
SELECT id, name, short_name, colour, kind, pin_hash IS NOT NULL, created_at
|
||||
FROM viewers WHERE emby_user_id = $1 AND id = $2`, embyUserID, viewerID,
|
||||
).Scan(&v.ID, &v.Name, &v.ShortName, &v.Colour, &v.Kind, &v.HasPIN, &v.CreatedAt)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return Viewer{}, ErrViewerNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return Viewer{}, fmt.Errorf("store: viewer: %w", err)
|
||||
}
|
||||
return v, nil
|
||||
}
|
||||
|
||||
// CreateShadowViewer adds a person to an account.
|
||||
//
|
||||
// The count is taken inside the transaction, because the limit is the only thing standing
|
||||
// between a held D-pad on the add button and an unbounded picker.
|
||||
func (s *Store) CreateShadowViewer(
|
||||
ctx context.Context, embyUserID, name, shortName, colour string,
|
||||
) (Viewer, error) {
|
||||
name = strings.TrimSpace(name)
|
||||
if name == "" {
|
||||
return Viewer{}, fmt.Errorf("store: viewer name is required")
|
||||
}
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return Viewer{}, fmt.Errorf("store: begin create viewer: %w", err)
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
|
||||
var shadows int
|
||||
if err := tx.QueryRow(ctx, `
|
||||
SELECT count(*) FROM viewers WHERE emby_user_id = $1 AND kind = 'shadow'`,
|
||||
embyUserID,
|
||||
).Scan(&shadows); err != nil {
|
||||
return Viewer{}, fmt.Errorf("store: count viewers: %w", err)
|
||||
}
|
||||
if shadows >= MaxShadowViewers {
|
||||
return Viewer{}, fmt.Errorf("store: %d viewers is the limit", MaxShadowViewers)
|
||||
}
|
||||
|
||||
id, err := NewShadowViewerID()
|
||||
if err != nil {
|
||||
return Viewer{}, err
|
||||
}
|
||||
var v Viewer
|
||||
if err := tx.QueryRow(ctx, `
|
||||
INSERT INTO viewers (id, emby_user_id, name, short_name, colour, kind)
|
||||
VALUES ($1, $2, $3, $4, $5, 'shadow')
|
||||
RETURNING id, name, short_name, colour, kind, pin_hash IS NOT NULL, created_at`,
|
||||
id, embyUserID, name, strings.TrimSpace(shortName), strings.TrimSpace(colour),
|
||||
).Scan(&v.ID, &v.Name, &v.ShortName, &v.Colour, &v.Kind, &v.HasPIN, &v.CreatedAt); err != nil {
|
||||
return Viewer{}, fmt.Errorf("store: create viewer: %w", err)
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return Viewer{}, fmt.Errorf("store: commit create viewer: %w", err)
|
||||
}
|
||||
return v, nil
|
||||
}
|
||||
|
||||
// UpdateShadowViewer renames or re-colours a viewer. The main viewer is deliberately not
|
||||
// updatable here: its name is the Emby account's and belongs to Emby.
|
||||
func (s *Store) UpdateShadowViewer(
|
||||
ctx context.Context, embyUserID, viewerID, name, shortName, colour string,
|
||||
) (Viewer, error) {
|
||||
name = strings.TrimSpace(name)
|
||||
if name == "" {
|
||||
return Viewer{}, fmt.Errorf("store: viewer name is required")
|
||||
}
|
||||
var v Viewer
|
||||
err := s.pool.QueryRow(ctx, `
|
||||
UPDATE viewers SET name = $3, short_name = $4, colour = $5, updated_at = now()
|
||||
WHERE emby_user_id = $1 AND id = $2 AND kind = 'shadow'
|
||||
RETURNING id, name, short_name, colour, kind, pin_hash IS NOT NULL, created_at`,
|
||||
embyUserID, viewerID, name, strings.TrimSpace(shortName), strings.TrimSpace(colour),
|
||||
).Scan(&v.ID, &v.Name, &v.ShortName, &v.Colour, &v.Kind, &v.HasPIN, &v.CreatedAt)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return Viewer{}, ErrViewerNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return Viewer{}, fmt.Errorf("store: update viewer: %w", err)
|
||||
}
|
||||
return v, nil
|
||||
}
|
||||
|
||||
// DeleteShadowViewer removes a viewer and everything Memby held on their behalf.
|
||||
//
|
||||
// A main viewer can never be deleted through this route: it is the account's own, and an
|
||||
// account with no main viewer would have nothing to fall back to. The playback state goes
|
||||
// with the row rather than being left to a housekeeping task, because the whole of what it
|
||||
// describes is a person who no longer exists.
|
||||
func (s *Store) DeleteShadowViewer(ctx context.Context, embyUserID, viewerID string) error {
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: begin delete viewer: %w", err)
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
|
||||
tag, err := tx.Exec(ctx, `
|
||||
DELETE FROM viewers WHERE emby_user_id = $1 AND id = $2 AND kind = 'shadow'`,
|
||||
embyUserID, viewerID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: delete viewer: %w", err)
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return ErrViewerNotFound
|
||||
}
|
||||
if _, err := tx.Exec(ctx,
|
||||
`DELETE FROM viewer_playback_state WHERE viewer_id = $1`, viewerID); err != nil {
|
||||
return fmt.Errorf("store: delete viewer state: %w", err)
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return fmt.Errorf("store: commit delete viewer: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,100 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// A shadow id must never be mistaken for an Emby user id: the two are substituted for one
|
||||
// another in twenty tables that cannot tell the difference, so telling them apart by
|
||||
// inspection is the safety property the whole scheme rests on.
|
||||
func TestShadowViewerIDIsDistinguishableFromEmbyUserID(t *testing.T) {
|
||||
id, err := NewShadowViewerID()
|
||||
if err != nil {
|
||||
t.Fatalf("mint shadow id: %v", err)
|
||||
}
|
||||
if !IsShadowViewerID(id) {
|
||||
t.Fatalf("minted id %q not recognised as a shadow id", id)
|
||||
}
|
||||
// Emby's are 32 hex characters with no prefix.
|
||||
if IsShadowViewerID("8f14e45fceea167a5a36dedd4bea2543") {
|
||||
t.Fatal("an Emby user id was read as a shadow viewer")
|
||||
}
|
||||
if IsShadowViewerID("") || IsShadowViewerID("v") || IsShadowViewerID("viewer") {
|
||||
t.Fatal("a short string was read as a shadow viewer")
|
||||
}
|
||||
other, err := NewShadowViewerID()
|
||||
if err != nil {
|
||||
t.Fatalf("mint second shadow id: %v", err)
|
||||
}
|
||||
if other == id {
|
||||
t.Fatal("two minted ids collided")
|
||||
}
|
||||
}
|
||||
|
||||
func TestPlayedFromPosition(t *testing.T) {
|
||||
const hour = int64(36_000_000_000) // one hour in Emby ticks
|
||||
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
position int64
|
||||
runtime int64
|
||||
want bool
|
||||
}{
|
||||
{"finished", hour, hour, true},
|
||||
{"at the threshold", hour * 9 / 10, hour, true},
|
||||
{"just short of it", hour*9/10 - 1, hour, false},
|
||||
{"barely started", hour / 100, hour, false},
|
||||
// A runtime of zero means the length was not known, not that the title is zero
|
||||
// long. Reading it the other way marks everything watched at the first second.
|
||||
{"unknown runtime", hour, 0, false},
|
||||
{"nothing watched", 0, hour, false},
|
||||
{"negative position", -hour, hour, false},
|
||||
// Playing past the stated runtime is ordinary — a container whose duration is a
|
||||
// little short of its own last frame.
|
||||
{"past the end", hour * 2, hour, true},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
if got := PlayedFromPosition(tc.position, tc.runtime); got != tc.want {
|
||||
t.Fatalf("PlayedFromPosition(%d, %d) = %v, want %v",
|
||||
tc.position, tc.runtime, got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// The rollup a series card's tick and count are built from. It is done in Go rather than in
|
||||
// SQL — the query groups by the season/series pair and this folds it twice — so it is worth
|
||||
// pinning that both directions add up and that the date is the latest of them.
|
||||
func TestViewerAggregateRollup(t *testing.T) {
|
||||
earlier := time.Date(2026, 8, 1, 20, 0, 0, 0, time.UTC)
|
||||
later := time.Date(2026, 8, 18, 21, 30, 0, 0, time.UTC)
|
||||
aggregates := map[string]ViewerAggregate{}
|
||||
|
||||
// Two seasons of one show, folded into the series and kept apart per season.
|
||||
addViewerAggregate(aggregates, "show-1", 10, 10, &earlier)
|
||||
addViewerAggregate(aggregates, "season-1", 10, 10, &earlier)
|
||||
addViewerAggregate(aggregates, "show-1", 8, 3, &later)
|
||||
addViewerAggregate(aggregates, "season-2", 8, 3, &later)
|
||||
|
||||
series := aggregates["show-1"]
|
||||
if series.Total != 18 || series.Played != 13 {
|
||||
t.Errorf("series rollup = %d of %d, want 13 of 18", series.Played, series.Total)
|
||||
}
|
||||
if series.LastPlayedAt == nil || !series.LastPlayedAt.Equal(later) {
|
||||
t.Errorf("series last played = %v, want the later of the two", series.LastPlayedAt)
|
||||
}
|
||||
if got := aggregates["season-1"]; got.Total != 10 || got.Played != 10 {
|
||||
t.Errorf("season one = %d of %d, want 10 of 10", got.Played, got.Total)
|
||||
}
|
||||
if got := aggregates["season-2"]; got.Total != 8 || got.Played != 3 {
|
||||
t.Errorf("season two = %d of %d, want 3 of 8", got.Played, got.Total)
|
||||
}
|
||||
|
||||
// An episode filed under no series or no season contributes to neither, rather than to
|
||||
// a row keyed on the empty string — which would be an aggregate about nothing.
|
||||
addViewerAggregate(aggregates, "", 5, 5, &later)
|
||||
if _, ok := aggregates[""]; ok {
|
||||
t.Error("an unfiled episode produced an aggregate")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user