memby/server/internal/store/viewer_playback.go

package store

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
)

// ViewerState is one viewer's answer about one title, in the shape of the Emby UserData
// block it stands in for. Zero values are the honest answer for a title nobody has
// touched, which is what lets a caller decorate an item it found nothing stored for
// without a special case.
type ViewerState struct {
	ItemID           string
	SeriesID         string
	SeasonID         string
	PositionTicks    int64
	RuntimeTicks     int64
	Played           bool
	PlayCount        int
	Favourite        bool
	HiddenFromResume bool
	LastPlayedAt     *time.Time
}

// PlayedFraction is the share of a title that must be behind the viewer for it to count as
// watched. It matches Emby's own default so a household cannot come to disagree with itself
// about whether an episode is finished depending on which viewer watched it.
const PlayedFraction = 0.9

// PlayedFromPosition decides whether a stop report completed the title.
//
// A runtime of zero means the length was not known rather than that the title is zero
// long, so it can never complete anything — the alternative is that every report with a
// missing duration marks something watched at the first second.
func PlayedFromPosition(positionTicks, runtimeTicks int64) bool {
	if runtimeTicks <= 0 || positionTicks <= 0 {
		return false
	}
	return float64(positionTicks) >= float64(runtimeTicks)*PlayedFraction
}

// RecordViewerPlayback writes a progress or stop report for a shadow viewer.
//
// A completed title is stored at position zero, the way Emby stores one: the position is
// what Continue Watching reads, and a finished episode left sitting at its last frame is
// one the row keeps offering to resume four seconds from the end. play_count only moves on
// the transition into played, so the ten-second reports either side of the threshold cannot
// count one viewing several times.
func (s *Store) RecordViewerPlayback(ctx context.Context, viewerID string, state ViewerState) error {
	if viewerID == "" || state.ItemID == "" {
		return fmt.Errorf("store: viewer playback: viewer and item are required")
	}
	position := state.PositionTicks
	if position < 0 {
		position = 0
	}
	runtime := state.RuntimeTicks
	if runtime < 0 {
		runtime = 0
	}
	if state.Played {
		position = 0
	}
	_, err := s.pool.Exec(ctx, `
		INSERT INTO viewer_playback_state (
			viewer_id, item_id, series_id, season_id,
			position_ticks, runtime_ticks, played, play_count, last_played_at, updated_at
		) VALUES (
			$1, $2,
			-- The series is read out of the shared catalogue rather than asked of Emby or
			-- carried by the television: it is already there, it is what orders this
			-- viewer's Continue Watching, and a report arrives every ten seconds.
			COALESCE(NULLIF($3, ''), (SELECT series_id FROM library_items WHERE id = $2), ''),
			$4, $5, $6, $7, CASE WHEN $7 THEN 1 ELSE 0 END, now(), now())
		ON CONFLICT (viewer_id, item_id) DO UPDATE SET
			series_id      = CASE WHEN excluded.series_id <> '' THEN excluded.series_id
			                      ELSE viewer_playback_state.series_id END,
			season_id      = CASE WHEN excluded.season_id <> '' THEN excluded.season_id
			                      ELSE viewer_playback_state.season_id END,
			position_ticks = excluded.position_ticks,
			runtime_ticks  = CASE WHEN excluded.runtime_ticks > 0 THEN excluded.runtime_ticks
			                      ELSE viewer_playback_state.runtime_ticks END,
			played         = excluded.played,
			play_count     = viewer_playback_state.play_count
			                 + CASE WHEN excluded.played AND NOT viewer_playback_state.played
			                        THEN 1 ELSE 0 END,
			last_played_at = now(),
			updated_at     = now()`,
		viewerID, state.ItemID, state.SeriesID, state.SeasonID,
		position, runtime, state.Played,
	)
	if err != nil {
		return fmt.Errorf("store: record viewer playback: %w", err)
	}
	return nil
}

// SetViewerPlayed marks a title watched or unwatched by hand.
//
// Marking unwatched clears the position for the same reason marking watched does: the two
// are one statement about where this viewer stands with the title, and a cleared flag over
// a retained playhead would put it straight back into Continue Watching at the closing
// credits.
func (s *Store) SetViewerPlayed(ctx context.Context, viewerID, itemID string, played bool) error {
	_, err := s.pool.Exec(ctx, `
		INSERT INTO viewer_playback_state (
			viewer_id, item_id, position_ticks, played, play_count, last_played_at, updated_at
		) VALUES ($1, $2, 0, $3, CASE WHEN $3 THEN 1 ELSE 0 END,
		          CASE WHEN $3 THEN now() ELSE NULL END, now())
		ON CONFLICT (viewer_id, item_id) DO UPDATE SET
			position_ticks = 0,
			played         = excluded.played,
			play_count     = viewer_playback_state.play_count
			                 + CASE WHEN excluded.played AND NOT viewer_playback_state.played
			                        THEN 1 ELSE 0 END,
			last_played_at = CASE WHEN excluded.played
			                      THEN COALESCE(viewer_playback_state.last_played_at, now())
			                      ELSE viewer_playback_state.last_played_at END,
			updated_at     = now()`,
		viewerID, itemID, played)
	if err != nil {
		return fmt.Errorf("store: set viewer played: %w", err)
	}
	return nil
}

// SetViewerFavourite records a favourite that belongs to the person rather than to the
// Emby account, so one viewer's heart cannot appear on everybody else's launcher.
func (s *Store) SetViewerFavourite(ctx context.Context, viewerID, itemID string, favourite bool) error {
	_, err := s.pool.Exec(ctx, `
		INSERT INTO viewer_playback_state (viewer_id, item_id, favourite, updated_at)
		VALUES ($1, $2, $3, now())
		ON CONFLICT (viewer_id, item_id) DO UPDATE SET
			favourite = excluded.favourite, updated_at = now()`,
		viewerID, itemID, favourite)
	if err != nil {
		return fmt.Errorf("store: set viewer favourite: %w", err)
	}
	return nil
}

// HideViewerFromResume takes a title off this viewer's Continue Watching without claiming
// they watched it. The position is kept: hiding is a statement about the row, not about
// where they got to, and pressing Play again should still resume.
func (s *Store) HideViewerFromResume(ctx context.Context, viewerID, itemID string) error {
	_, err := s.pool.Exec(ctx, `
		INSERT INTO viewer_playback_state (viewer_id, item_id, hidden_from_resume, updated_at)
		VALUES ($1, $2, true, now())
		ON CONFLICT (viewer_id, item_id) DO UPDATE SET
			hidden_from_resume = true, updated_at = now()`,
		viewerID, itemID)
	if err != nil {
		return fmt.Errorf("store: hide from resume: %w", err)
	}
	return nil
}

// ViewerStateFor reads one title's state. A title with no row is not an error: it is a
// title this viewer has never touched, which is the ordinary case.
func (s *Store) ViewerStateFor(ctx context.Context, viewerID, itemID string) (ViewerState, error) {
	state := ViewerState{ItemID: itemID}
	err := s.pool.QueryRow(ctx, `
		SELECT series_id, season_id, position_ticks, runtime_ticks,
		       played, play_count, favourite, hidden_from_resume, last_played_at
		FROM viewer_playback_state WHERE viewer_id = $1 AND item_id = $2`,
		viewerID, itemID,
	).Scan(
		&state.SeriesID, &state.SeasonID, &state.PositionTicks, &state.RuntimeTicks,
		&state.Played, &state.PlayCount, &state.Favourite, &state.HiddenFromResume,
		&state.LastPlayedAt,
	)
	if errors.Is(err, pgx.ErrNoRows) {
		return state, nil
	}
	if err != nil {
		return ViewerState{}, fmt.Errorf("store: viewer state: %w", err)
	}
	return state, nil
}

// ViewerStates reads a whole launcher's worth in one query.
//
// This is the read behind every decorated row, so it is one indexed lookup for several
// hundred cards rather than a request per card — the economy decorateItemRatings already
// makes for scores.
func (s *Store) ViewerStates(
	ctx context.Context, viewerID string, itemIDs []string,
) (map[string]ViewerState, error) {
	states := map[string]ViewerState{}
	if viewerID == "" || len(itemIDs) == 0 {
		return states, nil
	}
	rows, err := s.pool.Query(ctx, `
		SELECT item_id, series_id, season_id, position_ticks, runtime_ticks,
		       played, play_count, favourite, hidden_from_resume, last_played_at
		FROM viewer_playback_state
		WHERE viewer_id = $1 AND item_id = ANY($2)`, viewerID, itemIDs)
	if err != nil {
		return nil, fmt.Errorf("store: viewer states: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var state ViewerState
		if err := rows.Scan(
			&state.ItemID, &state.SeriesID, &state.SeasonID,
			&state.PositionTicks, &state.RuntimeTicks,
			&state.Played, &state.PlayCount, &state.Favourite, &state.HiddenFromResume,
			&state.LastPlayedAt,
		); err != nil {
			return nil, fmt.Errorf("store: scan viewer state: %w", err)
		}
		states[state.ItemID] = state
	}
	return states, rows.Err()
}

// ViewerResumeItems is this viewer's Continue Watching, most recently played first.
//
// It answers in item ids alone: the catalogue is shared by the household and is read from
// library_items or Emby, so duplicating a single field of metadata here would be a second
// copy free to go stale.
func (s *Store) ViewerResumeItems(ctx context.Context, viewerID string, limit int) ([]string, error) {
	if viewerID == "" {
		return nil, nil
	}
	if limit <= 0 {
		limit = 20
	}
	rows, err := s.pool.Query(ctx, `
		SELECT item_id FROM viewer_playback_state
		WHERE viewer_id = $1 AND position_ticks > 0 AND NOT played AND NOT hidden_from_resume
		ORDER BY last_played_at DESC NULLS LAST
		LIMIT $2`, viewerID, limit)
	if err != nil {
		return nil, fmt.Errorf("store: viewer resume items: %w", err)
	}
	defer rows.Close()
	ids := []string{}
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return nil, fmt.Errorf("store: scan resume item: %w", err)
		}
		ids = append(ids, id)
	}
	return ids, rows.Err()
}

// ViewerFavouriteItems is this viewer's favourites, most recently marked first.
func (s *Store) ViewerFavouriteItems(ctx context.Context, viewerID string, limit int) ([]string, error) {
	if viewerID == "" {
		return nil, nil
	}
	if limit <= 0 {
		limit = 50
	}
	rows, err := s.pool.Query(ctx, `
		SELECT item_id FROM viewer_playback_state
		WHERE viewer_id = $1 AND favourite
		ORDER BY updated_at DESC
		LIMIT $2`, viewerID, limit)
	if err != nil {
		return nil, fmt.Errorf("store: viewer favourites: %w", err)
	}
	defer rows.Close()
	ids := []string{}
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return nil, fmt.Errorf("store: scan favourite: %w", err)
		}
		ids = append(ids, id)
	}
	return ids, rows.Err()
}

// ViewerWatchedSeries reports, per series, when this viewer last finished or watched an
// episode of it. It is what orders a shadow viewer's Continue Watching, which merges
// resumable items with the next unwatched episode of a series they are part-way through —
// and a Next Up episode has no time of its own, so it is placed by its series.
func (s *Store) ViewerWatchedSeries(
	ctx context.Context, viewerID string, limit int,
) (map[string]time.Time, error) {
	watched := map[string]time.Time{}
	if viewerID == "" {
		return watched, nil
	}
	if limit <= 0 {
		limit = 40
	}
	rows, err := s.pool.Query(ctx, `
		SELECT series_id, max(last_played_at) AS played_at
		FROM viewer_playback_state
		WHERE viewer_id = $1 AND series_id <> '' AND last_played_at IS NOT NULL
		GROUP BY series_id
		ORDER BY played_at DESC
		LIMIT $2`, viewerID, limit)
	if err != nil {
		return nil, fmt.Errorf("store: viewer watched series: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var seriesID string
		var playedAt time.Time
		if err := rows.Scan(&seriesID, &playedAt); err != nil {
			return nil, fmt.Errorf("store: scan watched series: %w", err)
		}
		watched[seriesID] = playedAt
	}
	return watched, rows.Err()
}

// ViewerPlayedInSeries reports which of a series' episodes this viewer has finished, which
// is what Next Up walks to find the first one they have not.
func (s *Store) ViewerPlayedInSeries(
	ctx context.Context, viewerID, seriesID string,
) (map[string]bool, error) {
	played := map[string]bool{}
	if viewerID == "" || seriesID == "" {
		return played, nil
	}
	rows, err := s.pool.Query(ctx, `
		SELECT item_id, played FROM viewer_playback_state
		WHERE viewer_id = $1 AND series_id = $2`, viewerID, seriesID)
	if err != nil {
		return nil, fmt.Errorf("store: viewer played in series: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var itemID string
		var done bool
		if err := rows.Scan(&itemID, &done); err != nil {
			return nil, fmt.Errorf("store: scan played episode: %w", err)
		}
		played[itemID] = done
	}
	return played, rows.Err()
}

// ViewerNextUp is the next unwatched episode of every series this viewer is part-way
// through, the series they watched most recently first.
//
// It is computed entirely in Postgres, out of the shared catalogue and this viewer's own
// state, because Emby's own NextUp answers for the *account* and there is nobody else to
// ask. That also makes it cheap: the alternative — walking each series' episode list over
// the wire — is one Emby request per show on the tail of the launcher.
//
// Three rules, each of which Emby's own answer also applies:
//
// An episode already resumable is left out, because it is in Continue Watching already and
// the merge would otherwise offer the same show twice.
//
// Specials are not next episodes. Season 0 is a real season and a perfectly good thing to
// watch, but it is not what "next" means, and a show whose specials sort first would never
// offer anything else.
//
// A series with nothing unwatched left simply contributes no row rather than an empty one.
func (s *Store) ViewerNextUp(ctx context.Context, viewerID string, limit int) ([]string, error) {
	if viewerID == "" {
		return nil, nil
	}
	if limit <= 0 {
		limit = 20
	}
	rows, err := s.pool.Query(ctx, `
		WITH watched AS (
			SELECT series_id, max(last_played_at) AS played_at
			FROM viewer_playback_state
			WHERE viewer_id = $1 AND series_id <> '' AND last_played_at IS NOT NULL
			GROUP BY series_id
		),
		episodes AS (
			SELECT li.id,
			       li.series_id,
			       COALESCE((li.payload->>'ParentIndexNumber')::int, 0) AS season,
			       COALESCE((li.payload->>'IndexNumber')::int, 0)       AS episode,
			       w.played_at
			FROM library_items li
			JOIN watched w ON w.series_id = li.series_id
			WHERE li.type = 'Episode'
			  AND COALESCE((li.payload->>'ParentIndexNumber')::int, 0) > 0
		),
		unplayed AS (
			SELECT e.id, e.played_at,
			       row_number() OVER (
			           PARTITION BY e.series_id ORDER BY e.season, e.episode, e.id
			       ) AS rank
			FROM episodes e
			LEFT JOIN viewer_playback_state vps
			       ON vps.viewer_id = $1 AND vps.item_id = e.id
			WHERE COALESCE(vps.played, false) = false
			  AND COALESCE(vps.position_ticks, 0) = 0
		)
		SELECT id FROM unplayed WHERE rank = 1
		ORDER BY played_at DESC
		LIMIT $2`, viewerID, limit)
	if err != nil {
		return nil, fmt.Errorf("store: viewer next up: %w", err)
	}
	defer rows.Close()
	ids := []string{}
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return nil, fmt.Errorf("store: scan next up: %w", err)
		}
		ids = append(ids, id)
	}
	return ids, rows.Err()
}

// ViewerAggregate is what a series or a season card says about a viewer: how much of it
// there is, how much of it is behind them, and when they last watched any of it.
//
// It stands in for the block Emby fills in from an item's children, which is the one place
// a shadow viewer was still shown the *account's* answer — a series card ticked because
// somebody else had finished it. Emby computes it per Emby user and there is nobody to ask
// for a person Emby has never heard of, so it is computed here out of the shared catalogue
// and this viewer's own state.
type ViewerAggregate struct {
	// Total is how many episodes the catalogue holds. Zero means the catalogue cannot
	// answer — a library not yet imported, or a series it has never seen — which is a
	// different thing from a series with nothing in it, and the caller must not print a
	// count for it.
	Total int
	// Played is how many of those this viewer has finished.
	Played int
	// LastPlayedAt is the most recent episode of it they touched, finished or not, which
	// is what orders a shelf.
	LastPlayedAt *time.Time
}

// ViewerContainerStates aggregates a viewer's episode state per series *and* per season.
//
// One map keyed by container id serves both, because a series id and a season id are both
// Emby GUIDs and cannot collide — so the caller looks an item up by its own id and does not
// have to know which of the two it is holding.
//
// The query groups by the **pair** and the two rollups are done here rather than in SQL.
// That is a deliberately dull query — no grouping sets, no second pass over the same rows —
// and it is exact for both answers because a season belongs to exactly one series, so
// summing a series' seasons is summing its episodes. It runs on the tail of every request
// that serves a series card, which is why the index it reads
// (library_items_series_episodes_idx) exists.
//
// One thing to know about it: a series is only counted completely if it was *asked* for.
// A season whose series was not in seriesIDs contributes to a partial series total, which
// is harmless only because nothing looks that series up — containerIDsIn asks for a
// season's series alongside it precisely so the case cannot arise for anything drawn.
func (s *Store) ViewerContainerStates(
	ctx context.Context, viewerID string, seriesIDs, seasonIDs []string,
) (map[string]ViewerAggregate, error) {
	aggregates := map[string]ViewerAggregate{}
	if viewerID == "" || (len(seriesIDs) == 0 && len(seasonIDs) == 0) {
		return aggregates, nil
	}
	rows, err := s.pool.Query(ctx, `
		SELECT li.series_id,
		       COALESCE(li.payload->>'SeasonId', '')               AS season_id,
		       count(*)                                            AS total,
		       count(*) FILTER (WHERE COALESCE(vps.played, false))  AS played,
		       max(vps.last_played_at)                             AS last_played_at
		FROM library_items li
		LEFT JOIN viewer_playback_state vps
		       ON vps.viewer_id = $1 AND vps.item_id = li.id
		WHERE li.type = 'Episode'
		  AND (li.series_id = ANY($2) OR COALESCE(li.payload->>'SeasonId', '') = ANY($3))
		GROUP BY li.series_id, COALESCE(li.payload->>'SeasonId', '')`,
		viewerID, seriesIDs, seasonIDs)
	if err != nil {
		return nil, fmt.Errorf("store: viewer container states: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var seriesID, seasonID string
		var total, played int
		var lastPlayedAt *time.Time
		if err := rows.Scan(&seriesID, &seasonID, &total, &played, &lastPlayedAt); err != nil {
			return nil, fmt.Errorf("store: scan container state: %w", err)
		}
		// An episode filed under no series or no season contributes to neither rather than
		// to a row keyed on the empty string, which would be an aggregate about nothing.
		addViewerAggregate(aggregates, seriesID, total, played, lastPlayedAt)
		addViewerAggregate(aggregates, seasonID, total, played, lastPlayedAt)
	}
	return aggregates, rows.Err()
}

// addViewerAggregate folds one season's worth of counting into a container's total.
func addViewerAggregate(
	into map[string]ViewerAggregate, key string, total, played int, lastPlayedAt *time.Time,
) {
	if key == "" {
		return
	}
	aggregate := into[key]
	aggregate.Total += total
	aggregate.Played += played
	if lastPlayedAt != nil &&
		(aggregate.LastPlayedAt == nil || lastPlayedAt.After(*aggregate.LastPlayedAt)) {
		aggregate.LastPlayedAt = lastPlayedAt
	}
	into[key] = aggregate
}