// Package store persists gateway sessions in Postgres. package store import ( "context" _ "embed" "encoding/json" "errors" "fmt" "strings" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) //go:embed schema.sql var schema string // ErrNotFound is returned when a token does not match a live session. var ErrNotFound = errors.New("store: session not found") func isNoRows(err error) bool { return errors.Is(err, pgx.ErrNoRows) } type Session struct { TokenHash []byte EmbyUserID string EmbyToken string Username string ServerID string DeviceID string DeviceName string ClientVersion string ClientProtocol string ClientCapabilities []string LastSeenAt time.Time } type KnownUser struct { ID string `json:"id"` Username string `json:"username"` LastSeen time.Time `json:"lastSeen"` } type KnownClient struct { 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 // exists only in Emby. An account exists here once it has at least one gateway session. // Devices deliberately omit both gateway and upstream credentials. type MembyAccount struct { ID string `json:"id"` Username string `json:"username"` CreatedAt time.Time `json:"createdAt"` LastSeen time.Time `json:"lastSeen"` Devices []MembyDevice `json:"devices"` RecommendationPreferences json.RawMessage `json:"-"` } 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"` Versions []DeviceVersion `json:"versions"` } // MembyAccounts returns only people who have signed in through Memby. It must not be // confused with an Emby user directory: users that exist solely in Emby are absent. func (s *Store) MembyAccounts(ctx context.Context) ([]MembyAccount, error) { rows, err := s.pool.Query(ctx, ` SELECT s.emby_user_id, s.username, s.device_id, s.device_name, s.client_version, s.client_protocol, s.client_capabilities, s.created_at, s.last_seen_at, COALESCE(o.preferences, '{}'::jsonb) FROM sessions s LEFT JOIN recommendation_onboarding o ON o.emby_user_id = s.emby_user_id ORDER BY s.last_seen_at DESC, s.emby_user_id, s.device_name`) if err != nil { return nil, fmt.Errorf("store: list Memby accounts: %w", err) } defer rows.Close() accounts := []MembyAccount{} byID := map[string]int{} for rows.Next() { var userID, username string var device MembyDevice var preferences []byte if err := rows.Scan( &userID, &username, &device.ID, &device.Name, &device.Version, &device.Protocol, &device.Capabilities, &device.SignedInAt, &device.LastSeen, &preferences, ); err != nil { return nil, fmt.Errorf("store: scan Memby account: %w", err) } index, ok := byID[userID] if !ok { index = len(accounts) byID[userID] = index accounts = append(accounts, MembyAccount{ ID: userID, Username: username, CreatedAt: device.SignedInAt, LastSeen: device.LastSeen, Devices: []MembyDevice{}, RecommendationPreferences: json.RawMessage(preferences), }) } account := &accounts[index] if device.SignedInAt.Before(account.CreatedAt) { account.CreatedAt = device.SignedInAt } if device.LastSeen.After(account.LastSeen) { account.LastSeen = device.LastSeen account.Username = username } account.Devices = append(account.Devices, device) } 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_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 { return nil, fmt.Errorf("store: list known clients: %w", err) } defer rows.Close() clients := []KnownClient{} deviceIDs := []string{} for rows.Next() { var client KnownClient 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) } 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. func (s *Store) KnownUsers(ctx context.Context) ([]KnownUser, error) { rows, err := s.pool.Query(ctx, ` SELECT DISTINCT ON (emby_user_id) emby_user_id, username, last_seen_at FROM sessions ORDER BY emby_user_id, last_seen_at DESC`) if err != nil { return nil, fmt.Errorf("store: list known users: %w", err) } defer rows.Close() users := []KnownUser{} for rows.Next() { var user KnownUser if err := rows.Scan(&user.ID, &user.Username, &user.LastSeen); err != nil { return nil, fmt.Errorf("store: scan known user: %w", err) } users = append(users, user) } return users, rows.Err() } type Store struct { pool *pgxpool.Pool } // defaultMaxConns is the floor on the connection pool. // // pgxpool's own default is max(4, GOMAXPROCS), which on the NAS this runs on is four — // and four is fewer than one household's televisions. Every one of them polls // /v1/status every ten seconds, and that poll alone reads the feature policy, the // notification preferences, the preference revision, the theme and the request // allowlist. Queries that each take a millisecond then queue behind each other for // hundreds, and the waiting is invisible: the query time is honest and the request is // slow anyway. A deployment that has stated pool_max_conns for itself keeps its own // answer — this replaces a default nobody chose, not a decision somebody made. const defaultMaxConns = 16 // minConns keeps a few connections open. Establishing one costs a TCP connection, TLS // where it is configured and Postgres' own backend fork, which is several milliseconds // paid by whichever request happened to arrive after an idle period — most often the // first thing somebody does after switching a television on. const minConns = 4 func Open(ctx context.Context, databaseURL string) (*Store, error) { config, err := pgxpool.ParseConfig(databaseURL) if err != nil { return nil, fmt.Errorf("store: parse database url: %w", err) } if !strings.Contains(databaseURL, "pool_max_conns") && config.MaxConns < defaultMaxConns { config.MaxConns = defaultMaxConns } if !strings.Contains(databaseURL, "pool_min_conns") && config.MinConns < minConns { config.MinConns = minConns } // Every query in the gateway passes through here, which is what makes a slow request // able to say how much of itself was Postgres. config.ConnConfig.Tracer = queryTracer{} pool, err := pgxpool.NewWithConfig(ctx, config) if err != nil { return nil, fmt.Errorf("store: connect: %w", err) } if err := pool.Ping(ctx); err != nil { pool.Close() return nil, fmt.Errorf("store: ping: %w", err) } return &Store{pool: pool}, nil } func (s *Store) Close() { s.pool.Close() } func (s *Store) Ping(ctx context.Context) error { return s.pool.Ping(ctx) } // Migrate applies the schema. It is idempotent, so it runs on every boot. func (s *Store) Migrate(ctx context.Context) error { if _, err := s.pool.Exec(ctx, schema); err != nil { return fmt.Errorf("store: migrate: %w", err) } return nil } // CreateSession records every signed-in TV without an account-level device cap. // Re-authenticating the same stable device replaces its token. // // 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 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 created, fmt.Errorf("store: lock user sessions: %w", err) } var previousHash []byte err = tx.QueryRow(ctx, ` SELECT token_hash FROM sessions WHERE emby_user_id = $1 AND device_id = $2`, sess.EmbyUserID, sess.DeviceID).Scan(&previousHash) if err != nil && !errors.Is(err, pgx.ErrNoRows) { 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, ` INSERT INTO sessions ( token_hash, emby_user_id, emby_token, username, server_id, device_id, device_name, client_version, client_protocol, client_capabilities ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) ON CONFLICT (emby_user_id, device_id) DO UPDATE SET token_hash = EXCLUDED.token_hash, emby_token = EXCLUDED.emby_token, username = EXCLUDED.username, server_id = EXCLUDED.server_id, device_name = EXCLUDED.device_name, client_version = EXCLUDED.client_version, client_protocol = EXCLUDED.client_protocol, client_capabilities = EXCLUDED.client_capabilities, last_seen_at = now()`, sess.TokenHash, sess.EmbyUserID, sess.EmbyToken, sess.Username, sess.ServerID, sess.DeviceID, sess.DeviceName, sess.ClientVersion, sess.ClientProtocol, sess.ClientCapabilities) if err != nil { return created, fmt.Errorf("store: create session: %w", err) } if err := tx.Commit(ctx); err != nil { return created, fmt.Errorf("store: commit session: %w", err) } 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) { var sess Session err := s.pool.QueryRow(ctx, ` SELECT token_hash, emby_user_id, emby_token, username, server_id, device_id, device_name, client_version, client_protocol, client_capabilities, last_seen_at FROM sessions WHERE token_hash = $1`, hash). Scan(&sess.TokenHash, &sess.EmbyUserID, &sess.EmbyToken, &sess.Username, &sess.ServerID, &sess.DeviceID, &sess.DeviceName, &sess.ClientVersion, &sess.ClientProtocol, &sess.ClientCapabilities, &sess.LastSeenAt) if errors.Is(err, pgx.ErrNoRows) { return Session{}, ErrNotFound } if err != nil { return Session{}, fmt.Errorf("store: load session: %w", err) } return sess, nil } // Touch records activity. Cheap enough to call on the auth path, and it is what the // idle-expiry sweep reads. func (s *Store) Touch(ctx context.Context, hash []byte) error { _, err := s.pool.Exec(ctx, `UPDATE sessions SET last_seen_at = now() WHERE token_hash = $1`, hash) return err } // UpdateSessionClientIdentity remembers the last non-empty identity supplied by a TV. // Headerless image requests can then still be attributed to the correct app build. func (s *Store) UpdateSessionClientIdentity( ctx context.Context, hash []byte, version, protocol string, capabilities []string, ) error { _, err := s.pool.Exec(ctx, ` UPDATE sessions SET client_version = CASE WHEN $2 <> '' THEN $2 ELSE client_version END, client_protocol = CASE WHEN $3 <> '' THEN $3 ELSE client_protocol END, client_capabilities = CASE WHEN cardinality($4::text[]) > 0 THEN $4 ELSE client_capabilities END, last_seen_at = now() WHERE token_hash = $1`, hash, version, protocol, capabilities) return err } func (s *Store) DeleteSession(ctx context.Context, hash []byte) error { _, err := s.pool.Exec(ctx, `DELETE FROM sessions WHERE token_hash = $1`, hash) return err } // SessionsForUser returns the TVs whose gateway tokens are still active for a user. func (s *Store) SessionsForUser(ctx context.Context, userID string) ([]Session, error) { rows, err := s.pool.Query(ctx, ` SELECT token_hash, emby_user_id, emby_token, username, server_id, device_id, device_name, client_version, client_protocol, client_capabilities, last_seen_at FROM sessions WHERE emby_user_id = $1 ORDER BY last_seen_at DESC, device_name`, userID) if err != nil { return nil, fmt.Errorf("store: list user sessions: %w", err) } defer rows.Close() var sessions []Session for rows.Next() { var sess Session if err := rows.Scan( &sess.TokenHash, &sess.EmbyUserID, &sess.EmbyToken, &sess.Username, &sess.ServerID, &sess.DeviceID, &sess.DeviceName, &sess.ClientVersion, &sess.ClientProtocol, &sess.ClientCapabilities, &sess.LastSeenAt, ); err != nil { return nil, fmt.Errorf("store: scan user session: %w", err) } sessions = append(sessions, sess) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("store: list user session rows: %w", err) } return sessions, nil } // DeleteUserDevice revokes one device while scoping the delete to the authenticated user. func (s *Store) DeleteUserDevice(ctx context.Context, userID, deviceID string) ([]byte, error) { var tokenHash []byte err := s.pool.QueryRow(ctx, ` DELETE FROM sessions WHERE emby_user_id = $1 AND device_id = $2 RETURNING token_hash`, userID, deviceID).Scan(&tokenHash) if errors.Is(err, pgx.ErrNoRows) { return nil, ErrNotFound } if err != nil { return nil, fmt.Errorf("store: delete user device: %w", err) } return tokenHash, nil } // DeleteUserSessions removes the Memby account's active gateway access while leaving // the upstream Emby user untouched. func (s *Store) DeleteUserSessions(ctx context.Context, userID string) ([][]byte, error) { rows, err := s.pool.Query(ctx, ` DELETE FROM sessions WHERE emby_user_id = $1 RETURNING token_hash`, userID) if err != nil { return nil, fmt.Errorf("store: delete user sessions: %w", err) } defer rows.Close() hashes := [][]byte{} for rows.Next() { var hash []byte if err := rows.Scan(&hash); err != nil { return nil, fmt.Errorf("store: scan deleted session: %w", err) } hashes = append(hashes, hash) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("store: read deleted sessions: %w", err) } if len(hashes) == 0 { return nil, ErrNotFound } return hashes, nil } // RenameUserDevice changes only the display name and keeps the session/token intact. func (s *Store) RenameUserDevice(ctx context.Context, userID, deviceID, deviceName string) error { tag, err := s.pool.Exec(ctx, ` UPDATE sessions SET device_name = $3 WHERE emby_user_id = $1 AND device_id = $2`, userID, deviceID, deviceName) if err != nil { return fmt.Errorf("store: rename user device: %w", err) } if tag.RowsAffected() == 0 { return ErrNotFound } return nil } // DeleteIdleSessions retires tokens unused for longer than idle, returning how many went. func (s *Store) DeleteIdleSessions(ctx context.Context, idle time.Duration) (int64, error) { tag, err := s.pool.Exec(ctx, `DELETE FROM sessions WHERE last_seen_at < now() - $1::interval`, fmt.Sprintf("%d seconds", int64(idle.Seconds()))) if err != nil { return 0, err } return tag.RowsAffected(), nil }