memby/server/internal/store/store.go

// 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
}