App v0.2.26 and gateway 0.1.20
Client: seek controls, Bazarr subtitle download and cast panel in the player; MDBList ratings strip; episode and schedule detail pages; series pace estimate; what's new panel; install-permission onboarding step; synced per-profile preferences; Emby outage banner. Gateway: rebuilt admin console (one fragment per page), preference history and restore, merged Continue Watching, Emby health probe, subtitle selection and Bazarr download, structured request logging with per-request identity, and embedded build version. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
ponzischeme89
co-authored by
Claude Opus 5
Claude Opus 5
parent
2675e6d82b
commit
4a4df7a73c
@@ -0,0 +1,20 @@
|
||||
package store
|
||||
|
||||
import "testing"
|
||||
|
||||
// The exemption is the safety of the whole rule: a build predating device naming calls
|
||||
// itself "Memby TV", and a household with two of those would otherwise watch each set
|
||||
// sign the other out every time somebody opened the app.
|
||||
func TestSupersedeNameRefusesNamesThatIdentifyNothing(t *testing.T) {
|
||||
for _, name := range []string{"", " ", DefaultDeviceName, "memby tv", " Memby TV "} {
|
||||
if got := supersedeName(name); got != "" {
|
||||
t.Fatalf("supersedeName(%q) = %q, want no match", name, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSupersedeNameMatchesOnTheNameSomebodyTyped(t *testing.T) {
|
||||
if got := supersedeName(" Living room "); got != "Living room" {
|
||||
t.Fatalf("supersedeName trimmed to %q", got)
|
||||
}
|
||||
}
|
||||
@@ -275,6 +275,38 @@ func (s *Store) LibraryGenres(
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// SeriesRef is the minimum needed to link an outside catalogue's show — Sonarr's, in
|
||||
// practice — to the Emby series the library holds, so a card built from that catalogue
|
||||
// can open the show's own page. Year is 0 when Emby does not know it.
|
||||
type SeriesRef struct {
|
||||
ID string
|
||||
Name string
|
||||
Year int
|
||||
}
|
||||
|
||||
// SeriesRefs lists every imported series. The catalogue is a household's, not a
|
||||
// provider's: a few hundred rows of three short columns, which is why this reads them
|
||||
// all rather than querying per title.
|
||||
func (s *Store) SeriesRefs(ctx context.Context) ([]SeriesRef, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT id, name, COALESCE(production_year, 0)
|
||||
FROM library_items
|
||||
WHERE type = 'Series'`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: series refs: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
out := []SeriesRef{}
|
||||
for rows.Next() {
|
||||
var ref SeriesRef
|
||||
if err := rows.Scan(&ref.ID, &ref.Name, &ref.Year); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, ref)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func lowerStrings(values []string) []string {
|
||||
out := make([]string, 0, len(values))
|
||||
for _, value := range values {
|
||||
|
||||
@@ -0,0 +1,389 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// UserPreferences is one viewer's TV settings, held server-side so they follow the person
|
||||
// rather than the television.
|
||||
//
|
||||
// The document is stored opaquely on purpose. The vocabulary — which keys exist, what
|
||||
// values are legal — lives in internal/api, next to the client contract it belongs to, so
|
||||
// adding a setting is one file and never a database migration. What the store owns is the
|
||||
// part the API cannot: the revision, and the guarantee that two writers cannot interleave.
|
||||
type UserPreferences struct {
|
||||
Preferences json.RawMessage `json:"preferences"`
|
||||
// Revision increases by one on every accepted write. It is what lets a television
|
||||
// notice, from the status poll alone, that something changed it elsewhere.
|
||||
Revision int64 `json:"revision"`
|
||||
UpdatedAt time.Time `json:"updatedAt"`
|
||||
// Source records who wrote it last: "device" or "admin". The TV shows nothing with
|
||||
// this, but an operator asking "why did their layout change" needs the answer.
|
||||
Source string `json:"source"`
|
||||
}
|
||||
|
||||
// ErrPreferencesConflict means the caller's revision was not the current one. The client
|
||||
// resolves it by reading the server's copy — never by retrying its own write, which is
|
||||
// how an admin push would get silently undone by whichever TV was slowest to notice.
|
||||
var ErrPreferencesConflict = errors.New("store: user preferences revision conflict")
|
||||
|
||||
// ForceRevision skips the revision check. Only the admin console passes it: an operator
|
||||
// pushing settings is deliberately overriding whatever the devices last agreed on.
|
||||
const ForceRevision int64 = -1
|
||||
|
||||
func emptyPreferences() UserPreferences {
|
||||
return UserPreferences{Preferences: json.RawMessage(`{}`), Source: "default"}
|
||||
}
|
||||
|
||||
func (s *Store) UserPreferences(ctx context.Context, userID string) (UserPreferences, error) {
|
||||
var result UserPreferences
|
||||
var raw []byte
|
||||
err := s.pool.QueryRow(ctx, `
|
||||
SELECT preferences, revision, updated_at, source
|
||||
FROM user_preferences WHERE emby_user_id = $1`, userID).
|
||||
Scan(&raw, &result.Revision, &result.UpdatedAt, &result.Source)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return emptyPreferences(), nil
|
||||
}
|
||||
if err != nil {
|
||||
return emptyPreferences(), fmt.Errorf("store: read user preferences: %w", err)
|
||||
}
|
||||
result.Preferences = json.RawMessage(raw)
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// AllUserPreferences is the admin console's read. Absent users are absent rather than
|
||||
// defaulted, so the page can distinguish "never saved anything" from "saved the defaults".
|
||||
func (s *Store) AllUserPreferences(ctx context.Context) (map[string]UserPreferences, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT emby_user_id, preferences, revision, updated_at, source FROM user_preferences`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: list user preferences: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
all := map[string]UserPreferences{}
|
||||
for rows.Next() {
|
||||
var userID string
|
||||
var value UserPreferences
|
||||
var raw []byte
|
||||
if err := rows.Scan(&userID, &raw, &value.Revision, &value.UpdatedAt, &value.Source); err != nil {
|
||||
return nil, fmt.Errorf("store: scan user preferences: %w", err)
|
||||
}
|
||||
value.Preferences = json.RawMessage(raw)
|
||||
all[userID] = value
|
||||
}
|
||||
return all, rows.Err()
|
||||
}
|
||||
|
||||
// PreferenceWrite is one attempt to change somebody's settings, and everything the
|
||||
// history needs to describe it afterwards.
|
||||
//
|
||||
// The attribution fields are on the write rather than looked up later because the answer
|
||||
// changes: a television is renamed, signed out, replaced. What the history has to be able
|
||||
// to say is what was true when the change was made.
|
||||
type PreferenceWrite struct {
|
||||
Preferences json.RawMessage
|
||||
// ExpectedRevision is the revision the writer believes it is editing, or
|
||||
// ForceRevision for a deliberate override.
|
||||
ExpectedRevision int64
|
||||
// Source is "device" or "admin".
|
||||
Source string
|
||||
DeviceID, DeviceName, ClientVersion string
|
||||
// RestoredFrom is the revision this document was taken from, when the write is a
|
||||
// restore. Zero otherwise.
|
||||
RestoredFrom int64
|
||||
}
|
||||
|
||||
// preferenceHistoryLimit is how many revisions are kept per person. This is a household:
|
||||
// a hundred changes is years of them, and the operator question history answers — "what
|
||||
// did I just do to their launcher, and can I undo it" — is asked about the recent end.
|
||||
const preferenceHistoryLimit = 100
|
||||
|
||||
// SetUserPreferences writes the document if write.ExpectedRevision still matches, and
|
||||
// returns what is now stored.
|
||||
//
|
||||
// The advisory lock and the re-read inside the transaction are the point: every
|
||||
// television in the house writes this row, and an admin push lands in the middle of them.
|
||||
// Without the lock, two TVs that both read revision 4 would both write revision 5 and one
|
||||
// person's choice would vanish with no error anywhere.
|
||||
//
|
||||
// The history row is written in the same transaction as the document, so there is no
|
||||
// state in which a revision exists and nothing records where it came from.
|
||||
func (s *Store) SetUserPreferences(
|
||||
ctx context.Context, userID string, write PreferenceWrite,
|
||||
) (UserPreferences, error) {
|
||||
preferences, expectedRevision, source := write.Preferences, write.ExpectedRevision, write.Source
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return UserPreferences{}, fmt.Errorf("store: begin user preferences write: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
if _, err := tx.Exec(ctx,
|
||||
`SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, "prefs:"+userID); err != nil {
|
||||
return UserPreferences{}, fmt.Errorf("store: lock user preferences: %w", err)
|
||||
}
|
||||
|
||||
var current int64
|
||||
err = tx.QueryRow(ctx,
|
||||
`SELECT revision FROM user_preferences WHERE emby_user_id = $1`, userID).Scan(¤t)
|
||||
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
|
||||
return UserPreferences{}, fmt.Errorf("store: read current user preferences: %w", err)
|
||||
}
|
||||
if expectedRevision != ForceRevision && current != expectedRevision {
|
||||
return UserPreferences{}, ErrPreferencesConflict
|
||||
}
|
||||
|
||||
next := UserPreferences{
|
||||
Preferences: preferences, Revision: current + 1,
|
||||
UpdatedAt: time.Now().UTC(), Source: source,
|
||||
}
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO user_preferences (emby_user_id, preferences, revision, updated_at, source)
|
||||
VALUES ($1, $2::jsonb, $3, $4, $5)
|
||||
ON CONFLICT (emby_user_id) DO UPDATE SET
|
||||
preferences = EXCLUDED.preferences, revision = EXCLUDED.revision,
|
||||
updated_at = EXCLUDED.updated_at, source = EXCLUDED.source`,
|
||||
userID, string(preferences), next.Revision, next.UpdatedAt, source); err != nil {
|
||||
return UserPreferences{}, fmt.Errorf("store: write user preferences: %w", err)
|
||||
}
|
||||
|
||||
var restoredFrom any
|
||||
if write.RestoredFrom > 0 {
|
||||
restoredFrom = write.RestoredFrom
|
||||
}
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO user_preference_revisions (
|
||||
emby_user_id, revision, preferences, source,
|
||||
device_id, device_name, client_version, restored_from, created_at)
|
||||
VALUES ($1, $2, $3::jsonb, $4, $5, $6, $7, $8, $9)
|
||||
ON CONFLICT (emby_user_id, revision) DO NOTHING`,
|
||||
userID, next.Revision, string(preferences), source,
|
||||
write.DeviceID, write.DeviceName, write.ClientVersion, restoredFrom, next.UpdatedAt,
|
||||
); err != nil {
|
||||
return UserPreferences{}, fmt.Errorf("store: write preference history: %w", err)
|
||||
}
|
||||
// Pruned here rather than on a schedule: this is the only writer, so it is the only
|
||||
// place the table can grow, and the acks of a revision nobody can see any more are
|
||||
// dead weight with it.
|
||||
if _, err := tx.Exec(ctx, `
|
||||
DELETE FROM user_preference_revisions
|
||||
WHERE emby_user_id = $1 AND revision <= $2 - $3`,
|
||||
userID, next.Revision, preferenceHistoryLimit,
|
||||
); err != nil {
|
||||
return UserPreferences{}, fmt.Errorf("store: prune preference history: %w", err)
|
||||
}
|
||||
// A device's most recent ack is never pruned, however far behind it has fallen. A
|
||||
// television switched off for a year is exactly the one an operator wants to see
|
||||
// described as "on revision 12" rather than as one that has never checked in.
|
||||
if _, err := tx.Exec(ctx, `
|
||||
DELETE FROM user_preference_acks stale
|
||||
WHERE stale.emby_user_id = $1 AND stale.revision <= $2 - $3
|
||||
AND EXISTS (
|
||||
SELECT 1 FROM user_preference_acks newer
|
||||
WHERE newer.emby_user_id = stale.emby_user_id
|
||||
AND newer.device_id = stale.device_id
|
||||
AND newer.revision > stale.revision)`,
|
||||
userID, next.Revision, preferenceHistoryLimit,
|
||||
); err != nil {
|
||||
return UserPreferences{}, fmt.Errorf("store: prune preference acks: %w", err)
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return UserPreferences{}, fmt.Errorf("store: commit user preferences: %w", err)
|
||||
}
|
||||
return next, nil
|
||||
}
|
||||
|
||||
// PreferenceRevision is one entry in the history: what was stored, who stored it, and
|
||||
// which televisions have since taken it.
|
||||
type PreferenceRevision struct {
|
||||
Revision int64 `json:"revision"`
|
||||
Preferences json.RawMessage `json:"-"`
|
||||
Source string `json:"source"`
|
||||
DeviceID string `json:"deviceId,omitempty"`
|
||||
DeviceName string `json:"deviceName,omitempty"`
|
||||
ClientVersion string `json:"clientVersion,omitempty"`
|
||||
RestoredFrom int64 `json:"restoredFrom,omitempty"`
|
||||
CreatedAt time.Time `json:"createdAt"`
|
||||
Acks []PreferenceAck `json:"acks"`
|
||||
}
|
||||
|
||||
// PreferenceAck is one television taking one revision.
|
||||
type PreferenceAck struct {
|
||||
DeviceID string `json:"deviceId"`
|
||||
DeviceName string `json:"deviceName"`
|
||||
ClientVersion string `json:"clientVersion,omitempty"`
|
||||
Revision int64 `json:"revision"`
|
||||
AckedAt time.Time `json:"ackedAt"`
|
||||
}
|
||||
|
||||
// UserPreferenceHistory returns the newest revisions first, each carrying the televisions
|
||||
// that fetched it.
|
||||
//
|
||||
// Two queries rather than a join: the history is small and bounded, and a join would
|
||||
// return the document — the largest column here — once per acknowledging device.
|
||||
func (s *Store) UserPreferenceHistory(
|
||||
ctx context.Context, userID string, limit int,
|
||||
) ([]PreferenceRevision, error) {
|
||||
if limit <= 0 || limit > preferenceHistoryLimit {
|
||||
limit = preferenceHistoryLimit
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT revision, preferences, source, device_id, device_name,
|
||||
client_version, COALESCE(restored_from, 0), created_at
|
||||
FROM user_preference_revisions
|
||||
WHERE emby_user_id = $1
|
||||
ORDER BY revision DESC
|
||||
LIMIT $2`, userID, limit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: read preference history: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
history := []PreferenceRevision{}
|
||||
byRevision := map[int64]int{}
|
||||
for rows.Next() {
|
||||
var entry PreferenceRevision
|
||||
var raw []byte
|
||||
if err := rows.Scan(&entry.Revision, &raw, &entry.Source, &entry.DeviceID,
|
||||
&entry.DeviceName, &entry.ClientVersion, &entry.RestoredFrom, &entry.CreatedAt,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("store: scan preference history: %w", err)
|
||||
}
|
||||
entry.Preferences = json.RawMessage(raw)
|
||||
entry.Acks = []PreferenceAck{}
|
||||
byRevision[entry.Revision] = len(history)
|
||||
history = append(history, entry)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("store: read preference history: %w", err)
|
||||
}
|
||||
if len(history) == 0 {
|
||||
return history, nil
|
||||
}
|
||||
|
||||
oldest := history[len(history)-1].Revision
|
||||
ackRows, err := s.pool.Query(ctx, `
|
||||
SELECT device_id, device_name, client_version, revision, acked_at
|
||||
FROM user_preference_acks
|
||||
WHERE emby_user_id = $1 AND revision >= $2
|
||||
ORDER BY acked_at`, userID, oldest)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: read preference acks: %w", err)
|
||||
}
|
||||
defer ackRows.Close()
|
||||
for ackRows.Next() {
|
||||
var ack PreferenceAck
|
||||
if err := ackRows.Scan(&ack.DeviceID, &ack.DeviceName,
|
||||
&ack.ClientVersion, &ack.Revision, &ack.AckedAt); err != nil {
|
||||
return nil, fmt.Errorf("store: scan preference ack: %w", err)
|
||||
}
|
||||
if index, ok := byRevision[ack.Revision]; ok {
|
||||
history[index].Acks = append(history[index].Acks, ack)
|
||||
}
|
||||
}
|
||||
return history, ackRows.Err()
|
||||
}
|
||||
|
||||
// PreferenceRevisionDocument is the stored document for one revision, which is what a
|
||||
// restore reads. Absent means pruned or never written, and the caller must say so rather
|
||||
// than silently restoring something else.
|
||||
func (s *Store) PreferenceRevisionDocument(
|
||||
ctx context.Context, userID string, revision int64,
|
||||
) (json.RawMessage, error) {
|
||||
var raw []byte
|
||||
err := s.pool.QueryRow(ctx, `
|
||||
SELECT preferences FROM user_preference_revisions
|
||||
WHERE emby_user_id = $1 AND revision = $2`, userID, revision).Scan(&raw)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: read preference revision: %w", err)
|
||||
}
|
||||
return json.RawMessage(raw), nil
|
||||
}
|
||||
|
||||
// PreferenceDeviceState is where one television has got to: the newest revision it has
|
||||
// fetched. A set that has never fetched one is absent, which the caller reports as such —
|
||||
// "has not taken anything yet" and "is on revision 3" are different situations.
|
||||
type PreferenceDeviceState struct {
|
||||
DeviceID string `json:"deviceId"`
|
||||
Revision int64 `json:"revision"`
|
||||
AckedAt time.Time `json:"ackedAt"`
|
||||
}
|
||||
|
||||
func (s *Store) PreferenceDeviceStates(
|
||||
ctx context.Context, userID string,
|
||||
) (map[string]PreferenceDeviceState, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT DISTINCT ON (device_id) device_id, revision, acked_at
|
||||
FROM user_preference_acks
|
||||
WHERE emby_user_id = $1
|
||||
ORDER BY device_id, revision DESC`, userID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: read preference device states: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
states := map[string]PreferenceDeviceState{}
|
||||
for rows.Next() {
|
||||
var state PreferenceDeviceState
|
||||
if err := rows.Scan(&state.DeviceID, &state.Revision, &state.AckedAt); err != nil {
|
||||
return nil, fmt.Errorf("store: scan preference device state: %w", err)
|
||||
}
|
||||
states[state.DeviceID] = state
|
||||
}
|
||||
return states, rows.Err()
|
||||
}
|
||||
|
||||
// RecordPreferenceAck notes that one television now holds one revision.
|
||||
//
|
||||
// It is written when a set *fetches* the document, which is the only proof there is that
|
||||
// it took it: the status poll tells every open TV the revision, but being told is not the
|
||||
// same as having adopted, and a set that is switched off or cannot reach the gateway is
|
||||
// exactly the one an operator is asking about.
|
||||
//
|
||||
// A device with no id is not recorded rather than recorded as "": the empty id would
|
||||
// collect every legacy client into one imaginary television.
|
||||
func (s *Store) RecordPreferenceAck(
|
||||
ctx context.Context, userID, deviceID, deviceName, clientVersion string, revision int64,
|
||||
) error {
|
||||
if userID == "" || deviceID == "" || revision <= 0 {
|
||||
return nil
|
||||
}
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO user_preference_acks (
|
||||
emby_user_id, device_id, revision, device_name, client_version, acked_at)
|
||||
VALUES ($1, $2, $3, $4, $5, now())
|
||||
ON CONFLICT (emby_user_id, device_id, revision) DO UPDATE SET
|
||||
device_name = EXCLUDED.device_name,
|
||||
client_version = EXCLUDED.client_version`,
|
||||
userID, deviceID, revision, deviceName, clientVersion)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: record preference ack: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// UserPreferenceRevision is what the status poll consults: the revision alone, never the
|
||||
// document. /v1/status runs every ten seconds for every open television, so this has to
|
||||
// stay a primary-key lookup returning one number — the TV only needs to know whether to
|
||||
// go and fetch the rest.
|
||||
func (s *Store) UserPreferenceRevision(ctx context.Context, userID string) (int64, error) {
|
||||
var revision int64
|
||||
err := s.pool.QueryRow(ctx,
|
||||
`SELECT revision FROM user_preferences WHERE emby_user_id = $1`, userID).Scan(&revision)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return 0, nil
|
||||
}
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("store: read preference revision: %w", err)
|
||||
}
|
||||
return revision, nil
|
||||
}
|
||||
@@ -0,0 +1,244 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// ErrMediaRatingsNotFound means no durable external-rating response has been stored yet.
|
||||
var ErrMediaRatingsNotFound = errors.New("store: media ratings not found")
|
||||
|
||||
// MediaRatings returns the raw provider response and the time it was fetched. Keeping the
|
||||
// raw response lets API settings select different sources without invalidating this cache.
|
||||
func (s *Store) MediaRatings(
|
||||
ctx context.Context, mediaType, provider, providerID string,
|
||||
) (json.RawMessage, time.Time, error) {
|
||||
var ratings []byte
|
||||
var fetchedAt time.Time
|
||||
err := s.pool.QueryRow(ctx, `
|
||||
SELECT ratings, fetched_at
|
||||
FROM external_media_ratings
|
||||
WHERE media_type = $1 AND provider = $2 AND provider_id = $3`,
|
||||
mediaType, provider, providerID).Scan(&ratings, &fetchedAt)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, time.Time{}, ErrMediaRatingsNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return nil, time.Time{}, fmt.Errorf("store: load media ratings: %w", err)
|
||||
}
|
||||
return json.RawMessage(ratings), fetchedAt, nil
|
||||
}
|
||||
|
||||
// RatingKey identifies one title at the external provider. The same key serves every
|
||||
// television and every viewer, which is why ratings are stored per title rather than
|
||||
// per Emby item — a series and each of its episodes share one row.
|
||||
type RatingKey struct {
|
||||
MediaType string // movie | show
|
||||
Provider string // tmdb | imdb
|
||||
ProviderID string
|
||||
}
|
||||
|
||||
func (k RatingKey) valid() bool {
|
||||
return k.MediaType != "" && k.Provider != "" && k.ProviderID != ""
|
||||
}
|
||||
|
||||
// MediaRatingsEntry is one stored provider response and its age.
|
||||
type MediaRatingsEntry struct {
|
||||
Ratings json.RawMessage
|
||||
FetchedAt time.Time
|
||||
}
|
||||
|
||||
// MediaRatingsBatch reads many stored responses in one query. Rows attach ratings to
|
||||
// every card they carry, so the per-title read of MediaRatings would otherwise mean a
|
||||
// round trip per poster.
|
||||
func (s *Store) MediaRatingsBatch(
|
||||
ctx context.Context, keys []RatingKey,
|
||||
) (map[RatingKey]MediaRatingsEntry, error) {
|
||||
found := make(map[RatingKey]MediaRatingsEntry, len(keys))
|
||||
mediaTypes, providers, providerIDs := ratingKeyColumns(keys)
|
||||
if len(mediaTypes) == 0 {
|
||||
return found, nil
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT media_type, provider, provider_id, ratings, fetched_at
|
||||
FROM external_media_ratings
|
||||
WHERE (media_type, provider, provider_id) IN (
|
||||
SELECT * FROM unnest($1::text[], $2::text[], $3::text[])
|
||||
)`, mediaTypes, providers, providerIDs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: load media ratings batch: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var key RatingKey
|
||||
var ratings []byte
|
||||
var fetchedAt time.Time
|
||||
if err := rows.Scan(&key.MediaType, &key.Provider, &key.ProviderID, &ratings, &fetchedAt); err != nil {
|
||||
return nil, fmt.Errorf("store: scan media ratings batch: %w", err)
|
||||
}
|
||||
found[key] = MediaRatingsEntry{Ratings: json.RawMessage(ratings), FetchedAt: fetchedAt}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("store: load media ratings batch: %w", err)
|
||||
}
|
||||
return found, nil
|
||||
}
|
||||
|
||||
// ratingKeyColumns flattens keys into the three parallel arrays unnest expects, dropping
|
||||
// incomplete ones so a single unresolved item cannot break the whole read.
|
||||
func ratingKeyColumns(keys []RatingKey) ([]string, []string, []string) {
|
||||
mediaTypes := make([]string, 0, len(keys))
|
||||
providers := make([]string, 0, len(keys))
|
||||
providerIDs := make([]string, 0, len(keys))
|
||||
seen := make(map[RatingKey]bool, len(keys))
|
||||
for _, key := range keys {
|
||||
if !key.valid() || seen[key] {
|
||||
continue
|
||||
}
|
||||
seen[key] = true
|
||||
mediaTypes = append(mediaTypes, key.MediaType)
|
||||
providers = append(providers, key.Provider)
|
||||
providerIDs = append(providerIDs, key.ProviderID)
|
||||
}
|
||||
return mediaTypes, providers, providerIDs
|
||||
}
|
||||
|
||||
// SaveItemRatingRef remembers which external title an Emby item is, so later rows do not
|
||||
// have to ask Emby for its ProviderIds again.
|
||||
func (s *Store) SaveItemRatingRef(ctx context.Context, itemID string, key RatingKey) error {
|
||||
if itemID == "" || !key.valid() {
|
||||
return nil
|
||||
}
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO item_rating_refs (item_id, media_type, provider, provider_id, updated_at)
|
||||
VALUES ($1, $2, $3, $4, now())
|
||||
ON CONFLICT (item_id) DO UPDATE SET
|
||||
media_type = EXCLUDED.media_type,
|
||||
provider = EXCLUDED.provider,
|
||||
provider_id = EXCLUDED.provider_id,
|
||||
updated_at = EXCLUDED.updated_at`,
|
||||
itemID, key.MediaType, key.Provider, key.ProviderID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: save item rating ref: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ItemRatingRefs resolves Emby item ids that have been looked up before.
|
||||
func (s *Store) ItemRatingRefs(ctx context.Context, itemIDs []string) (map[string]RatingKey, error) {
|
||||
found := make(map[string]RatingKey, len(itemIDs))
|
||||
if len(itemIDs) == 0 {
|
||||
return found, nil
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT item_id, media_type, provider, provider_id
|
||||
FROM item_rating_refs
|
||||
WHERE item_id = ANY($1)`, itemIDs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: load item rating refs: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var itemID string
|
||||
var key RatingKey
|
||||
if err := rows.Scan(&itemID, &key.MediaType, &key.Provider, &key.ProviderID); err != nil {
|
||||
return nil, fmt.Errorf("store: scan item rating refs: %w", err)
|
||||
}
|
||||
found[itemID] = key
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("store: load item rating refs: %w", err)
|
||||
}
|
||||
return found, nil
|
||||
}
|
||||
|
||||
// LibraryProviderIDs reads the provider identifiers the library import stored, following
|
||||
// an episode to its series — MDBList rates shows, never single episodes.
|
||||
//
|
||||
// The returned map is keyed by the *requested* item id and carries Emby's own
|
||||
// ProviderIds object, so the caller applies the same provider precedence it uses for a
|
||||
// live lookup rather than a second copy of that rule living in SQL.
|
||||
func (s *Store) LibraryProviderIDs(
|
||||
ctx context.Context, itemIDs []string,
|
||||
) (map[string]LibraryProviderRef, error) {
|
||||
found := make(map[string]LibraryProviderRef, len(itemIDs))
|
||||
if len(itemIDs) == 0 {
|
||||
return found, nil
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT item.id,
|
||||
item.type,
|
||||
CASE WHEN item.type = 'Episode' THEN series.payload -> 'ProviderIds'
|
||||
ELSE item.payload -> 'ProviderIds' END
|
||||
FROM library_items item
|
||||
LEFT JOIN library_items series
|
||||
ON item.type = 'Episode' AND series.id = item.series_id
|
||||
WHERE item.id = ANY($1)`, itemIDs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: load library provider ids: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var itemID, itemType string
|
||||
var raw []byte
|
||||
if err := rows.Scan(&itemID, &itemType, &raw); err != nil {
|
||||
return nil, fmt.Errorf("store: scan library provider ids: %w", err)
|
||||
}
|
||||
ids := map[string]string{}
|
||||
if len(raw) > 0 {
|
||||
// Emby writes provider ids as strings, but a hand-edited or future payload
|
||||
// need not: an undecodable object leaves the item unresolved rather than
|
||||
// failing the whole read.
|
||||
_ = json.Unmarshal(raw, &ids)
|
||||
}
|
||||
found[itemID] = LibraryProviderRef{Type: itemType, ProviderIDs: ids}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("store: load library provider ids: %w", err)
|
||||
}
|
||||
return found, nil
|
||||
}
|
||||
|
||||
// LibraryProviderRef is one imported item's type and external identifiers.
|
||||
type LibraryProviderRef struct {
|
||||
Type string
|
||||
ProviderIDs map[string]string
|
||||
}
|
||||
|
||||
// MediaRatingsStats reports how much of the durable cache exists and how much of it is
|
||||
// old enough to be refreshed, which is the only way the admin page can say whether the
|
||||
// integration is still spending external requests.
|
||||
func (s *Store) MediaRatingsStats(ctx context.Context, staleBefore time.Time) (total, stale int, err error) {
|
||||
err = s.pool.QueryRow(ctx, `
|
||||
SELECT count(*), count(*) FILTER (WHERE fetched_at < $1)
|
||||
FROM external_media_ratings`, staleBefore).Scan(&total, &stale)
|
||||
if err != nil {
|
||||
return 0, 0, fmt.Errorf("store: media ratings stats: %w", err)
|
||||
}
|
||||
return total, stale, nil
|
||||
}
|
||||
|
||||
// SaveMediaRatings upserts a successfully loaded provider response.
|
||||
func (s *Store) SaveMediaRatings(
|
||||
ctx context.Context, mediaType, provider, providerID string, ratings json.RawMessage,
|
||||
) error {
|
||||
if !json.Valid(ratings) {
|
||||
return errors.New("store: save media ratings: invalid JSON")
|
||||
}
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO external_media_ratings (media_type, provider, provider_id, ratings, fetched_at)
|
||||
VALUES ($1, $2, $3, $4, now())
|
||||
ON CONFLICT (media_type, provider, provider_id) DO UPDATE SET
|
||||
ratings = EXCLUDED.ratings,
|
||||
fetched_at = EXCLUDED.fetched_at`,
|
||||
mediaType, provider, providerID, ratings)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: save media ratings: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -39,6 +39,24 @@ CREATE INDEX IF NOT EXISTS sessions_last_seen_idx ON sessions (last_seen_at);
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS sessions_user_device_idx
|
||||
ON sessions (emby_user_id, device_id);
|
||||
|
||||
-- Every app build a television has been seen running.
|
||||
--
|
||||
-- A session row carries only the version in force right now, which is overwritten by the
|
||||
-- next call that reports a different one, so on its own the answer to "what has this set
|
||||
-- been running" is one value deep. This is keyed on device_id alone because the history
|
||||
-- belongs to the television rather than to whoever is signed into it, and it outlives a
|
||||
-- sign-out: the set is the same set when it comes back.
|
||||
CREATE TABLE IF NOT EXISTS device_versions (
|
||||
device_id TEXT NOT NULL,
|
||||
client_version TEXT NOT NULL,
|
||||
first_seen_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
last_seen_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (device_id, client_version)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS device_versions_recent_idx
|
||||
ON device_versions (device_id, last_seen_at DESC);
|
||||
|
||||
-- The imported library.
|
||||
--
|
||||
-- payload is Emby's item JSON verbatim, so rows served from here are byte-identical to
|
||||
@@ -69,6 +87,34 @@ 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);
|
||||
|
||||
-- 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 (
|
||||
media_type TEXT NOT NULL,
|
||||
provider TEXT NOT NULL,
|
||||
provider_id TEXT NOT NULL,
|
||||
ratings JSONB NOT NULL DEFAULT '[]'::jsonb,
|
||||
fetched_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (media_type, provider, provider_id)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS external_media_ratings_fetched_idx
|
||||
ON external_media_ratings (fetched_at);
|
||||
|
||||
-- Emby item id -> external provider identity, learned as televisions navigate.
|
||||
--
|
||||
-- The rating itself is keyed by the provider's id, which Emby only reveals in a
|
||||
-- ProviderIds lookup. Remembering the answer is what lets a home row attach ratings to
|
||||
-- forty cards from one indexed read instead of forty Emby requests, and it works for
|
||||
-- items the library import has not yet re-read.
|
||||
CREATE TABLE IF NOT EXISTS item_rating_refs (
|
||||
item_id TEXT PRIMARY KEY,
|
||||
media_type TEXT NOT NULL,
|
||||
provider TEXT NOT NULL,
|
||||
provider_id TEXT NOT NULL,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
-- One row per import, so the admin page can show what happened and when.
|
||||
CREATE TABLE IF NOT EXISTS sync_runs (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
@@ -294,3 +340,66 @@ CREATE UNIQUE INDEX IF NOT EXISTS for_you_candidates_user_rank_idx
|
||||
ON for_you_candidates (emby_user_id, base_rank);
|
||||
CREATE INDEX IF NOT EXISTS for_you_candidates_user_runtime_rank_idx
|
||||
ON for_you_candidates (emby_user_id, runtime_minutes, base_rank);
|
||||
|
||||
-- One viewer's TV settings, so they follow the person rather than the television. The
|
||||
-- document is opaque here on purpose: the vocabulary lives in internal/api next to the
|
||||
-- client contract, so adding a setting never needs a migration. What this table owns is
|
||||
-- the revision, which is how a TV notices from the status poll alone that an operator (or
|
||||
-- another television) changed something.
|
||||
CREATE TABLE IF NOT EXISTS user_preferences (
|
||||
emby_user_id TEXT PRIMARY KEY,
|
||||
preferences JSONB NOT NULL DEFAULT '{}'::jsonb,
|
||||
revision BIGINT NOT NULL DEFAULT 0,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
source TEXT NOT NULL DEFAULT 'device'
|
||||
);
|
||||
|
||||
-- Every accepted write of the document above, so the operator can read what changed, who
|
||||
-- changed it, and put a previous version back.
|
||||
--
|
||||
-- The document is stored whole rather than as a delta. A delta would have to be
|
||||
-- interpreted against a vocabulary that lives in internal/api and can gain a setting
|
||||
-- between two revisions, and restoring one would then mean replaying a chain; a whole
|
||||
-- document is restorable on its own and normalised on the way out. History is capped per
|
||||
-- person at write time (preferenceHistoryLimit) — this is a household, and the value of an
|
||||
-- entry falls off a cliff once nobody remembers the change.
|
||||
CREATE TABLE IF NOT EXISTS user_preference_revisions (
|
||||
emby_user_id TEXT NOT NULL,
|
||||
revision BIGINT NOT NULL,
|
||||
preferences JSONB NOT NULL,
|
||||
source TEXT NOT NULL DEFAULT 'device',
|
||||
-- Which television wrote it, captured at write time rather than joined from sessions:
|
||||
-- a set that has since been signed out still has to be nameable in the history.
|
||||
device_id TEXT NOT NULL DEFAULT '',
|
||||
device_name TEXT NOT NULL DEFAULT '',
|
||||
client_version TEXT NOT NULL DEFAULT '',
|
||||
-- The revision this one was restored from, when it was. Never a rewind: a restore is
|
||||
-- a new revision carrying an old document, because the revision is what tells a TV
|
||||
-- something changed and one that went backwards would leave every set believing it
|
||||
-- was already up to date.
|
||||
restored_from BIGINT,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (emby_user_id, revision)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS user_preference_revisions_user_idx
|
||||
ON user_preference_revisions (emby_user_id, revision DESC);
|
||||
|
||||
-- Which televisions have actually taken a revision, recorded when a set fetches the
|
||||
-- document rather than when the server writes it.
|
||||
--
|
||||
-- The status poll carries the revision to every open TV, but a TV being told is not a TV
|
||||
-- having adopted: it may be switched off, mid-film, or unable to reach /v1/preferences.
|
||||
-- The fetch is the only proof, so it is what writes here.
|
||||
CREATE TABLE IF NOT EXISTS user_preference_acks (
|
||||
emby_user_id TEXT NOT NULL,
|
||||
device_id TEXT NOT NULL,
|
||||
revision BIGINT NOT NULL,
|
||||
device_name TEXT NOT NULL DEFAULT '',
|
||||
client_version TEXT NOT NULL DEFAULT '',
|
||||
acked_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (emby_user_id, device_id, revision)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS user_preference_acks_user_revision_idx
|
||||
ON user_preference_acks (emby_user_id, revision DESC);
|
||||
|
||||
@@ -28,7 +28,7 @@ const MDBListSettingsKey = "mdblist_settings"
|
||||
|
||||
var defaultMDBListSources = []string{
|
||||
"imdb", "tomatoes", "audience", "metacritic", "letterboxd", "rogerebert",
|
||||
"tmdb", "trakt", "mal", "score", "score_average",
|
||||
"tmdb", "trakt", "mal", "anilist", "anidb", "kitsu", "score", "score_average",
|
||||
}
|
||||
|
||||
type MDBListSettings struct {
|
||||
|
||||
+212
-25
@@ -7,6 +7,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
@@ -42,12 +43,42 @@ type KnownUser struct {
|
||||
}
|
||||
|
||||
type KnownClient struct {
|
||||
DeviceName string `json:"deviceName"`
|
||||
Username string `json:"username"`
|
||||
Version string `json:"version"`
|
||||
Protocol string `json:"protocol"`
|
||||
Capabilities []string `json:"capabilities"`
|
||||
LastSeen time.Time `json:"lastSeen"`
|
||||
DeviceID string `json:"deviceId"`
|
||||
DeviceName string `json:"deviceName"`
|
||||
Username string `json:"username"`
|
||||
Version string `json:"version"`
|
||||
Protocol string `json:"protocol"`
|
||||
Capabilities []string `json:"capabilities"`
|
||||
LastSeen time.Time `json:"lastSeen"`
|
||||
Versions []DeviceVersion `json:"versions"`
|
||||
}
|
||||
|
||||
// DefaultDeviceName is what a television is called when its build predates device naming.
|
||||
// It is a placeholder rather than a name, which is why supersedeDevices refuses to treat
|
||||
// two sets wearing it as the same television.
|
||||
const DefaultDeviceName = "Memby TV"
|
||||
|
||||
// DeviceVersion is one app build a television has been seen running, newest first when
|
||||
// read back. It is deliberately per device rather than per session: the question it
|
||||
// answers — what has this set been running — is about the television.
|
||||
type DeviceVersion struct {
|
||||
Version string `json:"version"`
|
||||
FirstSeen time.Time `json:"firstSeen"`
|
||||
LastSeen time.Time `json:"lastSeen"`
|
||||
}
|
||||
|
||||
// SupersededDevice is a session a sign-in retired: the same television, under a device id
|
||||
// it no longer uses. The token hash comes back so its cache entry can go with it.
|
||||
type SupersededDevice struct {
|
||||
DeviceID string
|
||||
TokenHash []byte
|
||||
}
|
||||
|
||||
// SessionCreated reports what a sign-in displaced. ReplacedHash is the token this same
|
||||
// device id already held; Superseded is the rest of the same television's past lives.
|
||||
type SessionCreated struct {
|
||||
ReplacedHash []byte
|
||||
Superseded []SupersededDevice
|
||||
}
|
||||
|
||||
// MembyAccount is an account known to Memby, as opposed to an arbitrary user that
|
||||
@@ -63,13 +94,14 @@ type MembyAccount struct {
|
||||
}
|
||||
|
||||
type MembyDevice struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Version string `json:"version"`
|
||||
Protocol string `json:"protocol"`
|
||||
Capabilities []string `json:"capabilities"`
|
||||
SignedInAt time.Time `json:"signedInAt"`
|
||||
LastSeen time.Time `json:"lastSeen"`
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Version string `json:"version"`
|
||||
Protocol string `json:"protocol"`
|
||||
Capabilities []string `json:"capabilities"`
|
||||
SignedInAt time.Time `json:"signedInAt"`
|
||||
LastSeen time.Time `json:"lastSeen"`
|
||||
Versions []DeviceVersion `json:"versions"`
|
||||
}
|
||||
|
||||
// MembyAccounts returns only people who have signed in through Memby. It must not be
|
||||
@@ -123,14 +155,91 @@ func (s *Store) MembyAccounts(ctx context.Context) ([]MembyAccount, error) {
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("store: read Memby accounts: %w", err)
|
||||
}
|
||||
|
||||
deviceIDs := []string{}
|
||||
for _, account := range accounts {
|
||||
for _, device := range account.Devices {
|
||||
deviceIDs = append(deviceIDs, device.ID)
|
||||
}
|
||||
}
|
||||
// A history read failure costs the build list and nothing else — the devices and the
|
||||
// sign-out buttons are what this list is for.
|
||||
if history, err := s.DeviceVersions(ctx, deviceIDs); err == nil {
|
||||
for i := range accounts {
|
||||
for j := range accounts[i].Devices {
|
||||
accounts[i].Devices[j].Versions = history[accounts[i].Devices[j].ID]
|
||||
}
|
||||
}
|
||||
}
|
||||
return accounts, nil
|
||||
}
|
||||
|
||||
// DeviceVersions returns every build each of these televisions has been seen running,
|
||||
// most recently seen first. One query for a whole household: the console draws this
|
||||
// beside every device row and a query per device would grow with the house.
|
||||
func (s *Store) DeviceVersions(ctx context.Context, deviceIDs []string) (map[string][]DeviceVersion, error) {
|
||||
history := map[string][]DeviceVersion{}
|
||||
if len(deviceIDs) == 0 {
|
||||
return history, nil
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT device_id, client_version, first_seen_at, last_seen_at
|
||||
FROM device_versions
|
||||
WHERE device_id = ANY($1::text[])
|
||||
ORDER BY device_id, last_seen_at DESC`, deviceIDs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: list device versions: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var deviceID string
|
||||
var version DeviceVersion
|
||||
if err := rows.Scan(&deviceID, &version.Version, &version.FirstSeen, &version.LastSeen); err != nil {
|
||||
return nil, fmt.Errorf("store: scan device version: %w", err)
|
||||
}
|
||||
history[deviceID] = append(history[deviceID], version)
|
||||
}
|
||||
return history, rows.Err()
|
||||
}
|
||||
|
||||
// RecordDeviceVersion notes that a television is running this build. Called from the
|
||||
// sign-in and from the identity refresh on the auth path, so it must stay one statement:
|
||||
// every request a TV makes can reach it, and all but the first are a no-op write.
|
||||
func (s *Store) RecordDeviceVersion(ctx context.Context, deviceID, version string) error {
|
||||
if deviceID == "" || version == "" {
|
||||
return nil
|
||||
}
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO device_versions (device_id, client_version)
|
||||
VALUES ($1, $2)
|
||||
ON CONFLICT (device_id, client_version) DO UPDATE SET last_seen_at = now()`,
|
||||
deviceID, version)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: record device version: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteDeviceVersions retires a television's build history along with the television.
|
||||
// It is called wherever a device row is removed, so a set that is gone does not leave a
|
||||
// list of builds behind it with nothing to attach them to.
|
||||
func (s *Store) DeleteDeviceVersions(ctx context.Context, deviceIDs ...string) error {
|
||||
if len(deviceIDs) == 0 {
|
||||
return nil
|
||||
}
|
||||
_, err := s.pool.Exec(ctx,
|
||||
`DELETE FROM device_versions WHERE device_id = ANY($1::text[])`, deviceIDs)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: delete device versions: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// KnownClients gives the feature console compatibility evidence without exposing
|
||||
// gateway or Emby credentials. Stale sessions remain useful rollout information.
|
||||
func (s *Store) KnownClients(ctx context.Context) ([]KnownClient, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT device_name, username, client_version, client_protocol,
|
||||
SELECT device_id, device_name, username, client_version, client_protocol,
|
||||
client_capabilities, last_seen_at
|
||||
FROM sessions ORDER BY last_seen_at DESC LIMIT 100`)
|
||||
if err != nil {
|
||||
@@ -138,15 +247,26 @@ func (s *Store) KnownClients(ctx context.Context) ([]KnownClient, error) {
|
||||
}
|
||||
defer rows.Close()
|
||||
clients := []KnownClient{}
|
||||
deviceIDs := []string{}
|
||||
for rows.Next() {
|
||||
var client KnownClient
|
||||
if err := rows.Scan(&client.DeviceName, &client.Username, &client.Version,
|
||||
&client.Protocol, &client.Capabilities, &client.LastSeen); err != nil {
|
||||
if err := rows.Scan(&client.DeviceID, &client.DeviceName, &client.Username,
|
||||
&client.Version, &client.Protocol, &client.Capabilities,
|
||||
&client.LastSeen); err != nil {
|
||||
return nil, fmt.Errorf("store: scan known client: %w", err)
|
||||
}
|
||||
clients = append(clients, client)
|
||||
deviceIDs = append(deviceIDs, client.DeviceID)
|
||||
}
|
||||
return clients, rows.Err()
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("store: read known clients: %w", err)
|
||||
}
|
||||
if history, err := s.DeviceVersions(ctx, deviceIDs); err == nil {
|
||||
for i := range clients {
|
||||
clients[i].Versions = history[clients[i].DeviceID]
|
||||
}
|
||||
}
|
||||
return clients, nil
|
||||
}
|
||||
|
||||
// KnownUsers returns one entry per Emby user that has signed in to the gateway.
|
||||
@@ -252,16 +372,25 @@ func (s *Store) Migrate(ctx context.Context) error {
|
||||
|
||||
// CreateSession records every signed-in TV without an account-level device cap.
|
||||
// Re-authenticating the same stable device replaces its token.
|
||||
// The replaced hash is returned so its Redis entry can be invalidated immediately.
|
||||
func (s *Store) CreateSession(ctx context.Context, sess Session) ([]byte, error) {
|
||||
//
|
||||
// It also retires the same television's earlier identities. Sessions are unique per
|
||||
// (user, device id), so a second row for one set can only mean its device id changed —
|
||||
// a reinstall on a build that generated a random id, or an install predating the derived
|
||||
// one. Left alone, each of those keeps a row, an Emby device record and a build history of
|
||||
// its own, and one television reads as several. See supersedeDevices for the match rule.
|
||||
//
|
||||
// The replaced hash and the superseded rows come back so their Redis entries, their Emby
|
||||
// records and their build histories can be retired with them.
|
||||
func (s *Store) CreateSession(ctx context.Context, sess Session) (SessionCreated, error) {
|
||||
created := SessionCreated{}
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: begin session: %w", err)
|
||||
return created, fmt.Errorf("store: begin session: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
|
||||
if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, sess.EmbyUserID); err != nil {
|
||||
return nil, fmt.Errorf("store: lock user sessions: %w", err)
|
||||
return created, fmt.Errorf("store: lock user sessions: %w", err)
|
||||
}
|
||||
|
||||
var previousHash []byte
|
||||
@@ -270,7 +399,13 @@ func (s *Store) CreateSession(ctx context.Context, sess Session) ([]byte, error)
|
||||
WHERE emby_user_id = $1 AND device_id = $2`,
|
||||
sess.EmbyUserID, sess.DeviceID).Scan(&previousHash)
|
||||
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, fmt.Errorf("store: find device session: %w", err)
|
||||
return created, fmt.Errorf("store: find device session: %w", err)
|
||||
}
|
||||
created.ReplacedHash = previousHash
|
||||
|
||||
created.Superseded, err = supersedeDevices(ctx, tx, sess)
|
||||
if err != nil {
|
||||
return created, err
|
||||
}
|
||||
|
||||
_, err = tx.Exec(ctx, `
|
||||
@@ -293,12 +428,64 @@ func (s *Store) CreateSession(ctx context.Context, sess Session) ([]byte, error)
|
||||
sess.ServerID, sess.DeviceID, sess.DeviceName, sess.ClientVersion, sess.ClientProtocol,
|
||||
sess.ClientCapabilities)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: create session: %w", err)
|
||||
return created, fmt.Errorf("store: create session: %w", err)
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return nil, fmt.Errorf("store: commit session: %w", err)
|
||||
return created, fmt.Errorf("store: commit session: %w", err)
|
||||
}
|
||||
return previousHash, nil
|
||||
return created, nil
|
||||
}
|
||||
|
||||
// supersedeDevices deletes the rows this sign-in makes redundant: the same person, the
|
||||
// same television name, a device id the set no longer uses.
|
||||
//
|
||||
// The name is the only evidence there is. A television that changes its device id has
|
||||
// nothing else in common with its previous row — the token is new, the session is new, and
|
||||
// Emby has issued it a second device record. What it does keep is what somebody typed into
|
||||
// Settings → Devices, which is the name of a physical set in a house.
|
||||
//
|
||||
// DefaultDeviceName is exempt, and that exemption is the whole safety of the rule: a build
|
||||
// predating device naming calls itself "Memby TV", so matching on it would let the second
|
||||
// such set in a household delete the first every time it signed in.
|
||||
func supersedeDevices(ctx context.Context, tx pgx.Tx, sess Session) ([]SupersededDevice, error) {
|
||||
name := supersedeName(sess.DeviceName)
|
||||
if name == "" {
|
||||
return nil, nil
|
||||
}
|
||||
rows, err := tx.Query(ctx, `
|
||||
DELETE FROM sessions
|
||||
WHERE emby_user_id = $1 AND device_id <> $2 AND lower(device_name) = lower($3)
|
||||
RETURNING device_id, token_hash`,
|
||||
sess.EmbyUserID, sess.DeviceID, name)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store: supersede devices: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var superseded []SupersededDevice
|
||||
for rows.Next() {
|
||||
var device SupersededDevice
|
||||
if err := rows.Scan(&device.DeviceID, &device.TokenHash); err != nil {
|
||||
return nil, fmt.Errorf("store: scan superseded device: %w", err)
|
||||
}
|
||||
superseded = append(superseded, device)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("store: read superseded devices: %w", err)
|
||||
}
|
||||
return superseded, nil
|
||||
}
|
||||
|
||||
// supersedeName is the whole decision, kept apart from the query so it can be tested: the
|
||||
// name to match earlier rows on, or "" for a sign-in that must displace nothing. A blank
|
||||
// name and the compatibility default are both "this set did not say", which is no evidence
|
||||
// of identity at all — and acting on it would let one unnamed television in a household
|
||||
// sign the others out.
|
||||
func supersedeName(deviceName string) string {
|
||||
name := strings.TrimSpace(deviceName)
|
||||
if name == "" || strings.EqualFold(name, DefaultDeviceName) {
|
||||
return ""
|
||||
}
|
||||
return name
|
||||
}
|
||||
|
||||
func (s *Store) SessionByTokenHash(ctx context.Context, hash []byte) (Session, error) {
|
||||
|
||||
Reference in New Issue
Block a user