memby/server/internal/store/credits.go

package store

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

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

const CreditsSettingsKey = "credits_settings"

// CreditsSettings is the durable operator tuning for candidate generation. Environment
// values remain the first-run defaults; this row exists only after the console saves an
// explicit choice.
type CreditsSettings struct {
	CandidateLimit   int       `json:"candidateLimit"`
	PrefetchEpisodes int       `json:"prefetchEpisodes"`
	MaxPrefetch      int       `json:"maxPrefetch"`
	RetryHours       int       `json:"retryHours"`
	UpdatedAt        time.Time `json:"updatedAt"`
}

func (s *Store) CreditsSettings(ctx context.Context) (CreditsSettings, bool, error) {
	var raw []byte
	err := s.pool.QueryRow(ctx, `SELECT value FROM app_settings WHERE key = $1`, CreditsSettingsKey).Scan(&raw)
	if errors.Is(err, pgx.ErrNoRows) {
		return CreditsSettings{}, false, nil
	}
	if err != nil {
		return CreditsSettings{}, false, fmt.Errorf("store: read credits settings: %w", err)
	}
	var settings CreditsSettings
	if err := json.Unmarshal(raw, &settings); err != nil {
		return CreditsSettings{}, false, fmt.Errorf("store: decode credits settings: %w", err)
	}
	return settings, true, nil
}

func (s *Store) SetCreditsSettings(ctx context.Context, settings CreditsSettings) error {
	settings.UpdatedAt = time.Now().UTC()
	raw, err := json.Marshal(settings)
	if err != nil {
		return err
	}
	_, err = s.pool.Exec(ctx, `
		INSERT INTO app_settings (key, value, updated_at)
		VALUES ($1, $2::jsonb, now())
		ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`,
		CreditsSettingsKey, string(raw))
	if err != nil {
		return fmt.Errorf("store: write credits settings: %w", err)
	}
	return nil
}

// CreditsScanHistoryRow is one completed scan, enriched with current library names for the
// console. Names are not copied into the history table, so metadata corrections appear here.
type CreditsScanHistoryRow struct {
	ID         int64     `json:"id"`
	ItemID     string    `json:"itemId"`
	ItemName   string    `json:"itemName"`
	SeriesName string    `json:"seriesName"`
	SeriesID   string    `json:"seriesId"`
	Season     int       `json:"season"`
	Episode    int       `json:"episode"`
	Reason     string    `json:"reason"`
	Priority   int       `json:"priority"`
	Outcome    string    `json:"outcome"`
	MarkerMs   int64     `json:"markerMs"`
	Confidence float64   `json:"confidence"`
	Method     string    `json:"method"`
	Frames     int       `json:"frames"`
	Error      string    `json:"error"`
	StartedAt  time.Time `json:"startedAt"`
	FinishedAt time.Time `json:"finishedAt"`
	DurationMs int64     `json:"durationMs"`
}

func (s *Store) SaveCreditsScanAttempt(ctx context.Context, row CreditsScanHistoryRow) error {
	duration := row.FinishedAt.Sub(row.StartedAt).Milliseconds()
	if duration < 0 {
		duration = 0
	}
	_, err := s.pool.Exec(ctx, `
		INSERT INTO credits_scan_history (
			item_id, series_id, season_number, episode_number, reason, priority, outcome,
			marker_ms, confidence, method, frames_sampled, error_text,
			started_at, finished_at, duration_ms
		) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)`,
		row.ItemID, row.SeriesID, row.Season, row.Episode, row.Reason, row.Priority,
		row.Outcome, row.MarkerMs, row.Confidence, row.Method, row.Frames, row.Error,
		row.StartedAt, row.FinishedAt, duration)
	if err != nil {
		return fmt.Errorf("store: save credits scan history: %w", err)
	}
	return nil
}

func (s *Store) RecentCreditsScanTimes(
	ctx context.Context, itemIDs []string, since time.Time,
) (map[string]time.Time, error) {
	out := map[string]time.Time{}
	if len(itemIDs) == 0 {
		return out, nil
	}
	rows, err := s.pool.Query(ctx, `
		SELECT item_id, max(finished_at)
		FROM credits_scan_history
		WHERE item_id = ANY($1) AND finished_at >= $2
		  AND outcome IN ('no_match', 'failed')
		GROUP BY item_id`, itemIDs, since.UTC())
	if err != nil {
		return nil, fmt.Errorf("store: recent credits scans: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var itemID string
		var finished time.Time
		if err := rows.Scan(&itemID, &finished); err != nil {
			return nil, err
		}
		out[itemID] = finished
	}
	return out, rows.Err()
}

func (s *Store) CreditsScanHistory(ctx context.Context, limit int) ([]CreditsScanHistoryRow, error) {
	if limit <= 0 {
		limit = 100
	}
	rows, err := s.pool.Query(ctx, `
		SELECT h.id, h.item_id,
		       coalesce(i.payload->>'Name', ''), coalesce(i.payload->>'SeriesName', ''),
		       h.series_id, h.season_number, h.episode_number, h.reason, h.priority,
		       h.outcome, h.marker_ms, h.confidence, h.method, h.frames_sampled,
		       h.error_text, h.started_at, h.finished_at, h.duration_ms
		FROM credits_scan_history h
		LEFT JOIN library_items i ON i.id = h.item_id
		ORDER BY h.finished_at DESC
		LIMIT $1`, limit)
	if err != nil {
		return nil, fmt.Errorf("store: credits scan history: %w", err)
	}
	defer rows.Close()
	out := make([]CreditsScanHistoryRow, 0, limit)
	for rows.Next() {
		var row CreditsScanHistoryRow
		if err := rows.Scan(&row.ID, &row.ItemID, &row.ItemName, &row.SeriesName,
			&row.SeriesID, &row.Season, &row.Episode, &row.Reason, &row.Priority,
			&row.Outcome, &row.MarkerMs, &row.Confidence, &row.Method, &row.Frames,
			&row.Error, &row.StartedAt, &row.FinishedAt, &row.DurationMs); err != nil {
			return nil, err
		}
		out = append(out, row)
	}
	return out, rows.Err()
}

// The database half of credits marking.
//
// Marker queries keep the playback path cheap. Scan history is operational data on the
// background worker path only; queue transitions remain entirely in memory.

// CreditsMarkerRow is one stored marker.
type CreditsMarkerRow struct {
	ItemID           string
	MediaFingerprint string
	CreditsStartMs   int64
	Confidence       float64
	DetectionMethod  string
	SeriesID         string
	Season           int
	CreatedAt        time.Time
	UpdatedAt        time.Time
}

// CreditsMarker reads the marker for one media version. Absence is an ordinary answer.
func (s *Store) CreditsMarker(
	ctx context.Context, itemID, fingerprint string,
) (CreditsMarkerRow, bool, error) {
	var row CreditsMarkerRow
	err := s.pool.QueryRow(ctx, `
		SELECT item_id, media_fingerprint, credits_start_ms, confidence, detection_method,
		       series_id, season_number, created_at, updated_at
		FROM credits_markers
		WHERE item_id = $1 AND media_fingerprint = $2`, itemID, fingerprint).
		Scan(&row.ItemID, &row.MediaFingerprint, &row.CreditsStartMs, &row.Confidence,
			&row.DetectionMethod, &row.SeriesID, &row.Season, &row.CreatedAt, &row.UpdatedAt)
	if errors.Is(err, pgx.ErrNoRows) {
		return CreditsMarkerRow{}, false, nil
	}
	if err != nil {
		return CreditsMarkerRow{}, false, fmt.Errorf("store: credits marker: %w", err)
	}
	return row, true, nil
}

// SaveCreditsMarker upserts one marker. This is the single write the whole subsystem makes,
// and the caller has already decided that the new evidence is worth it — the stability rule
// lives in the credits package beside the confidence model it depends on, not here.
//
// created_at is preserved on conflict so a marker's age remains the age of the finding rather
// than of the last time something confirmed it.
func (s *Store) SaveCreditsMarker(ctx context.Context, row CreditsMarkerRow) error {
	_, err := s.pool.Exec(ctx, `
		INSERT INTO credits_markers (
			item_id, media_fingerprint, credits_start_ms, confidence, detection_method,
			series_id, season_number, created_at, updated_at
		) VALUES ($1, $2, $3, $4, $5, $6, $7, now(), now())
		ON CONFLICT (item_id, media_fingerprint) DO UPDATE SET
			credits_start_ms = EXCLUDED.credits_start_ms,
			confidence       = EXCLUDED.confidence,
			detection_method = EXCLUDED.detection_method,
			series_id        = EXCLUDED.series_id,
			season_number    = EXCLUDED.season_number,
			updated_at       = now()`,
		row.ItemID, row.MediaFingerprint, row.CreditsStartMs, row.Confidence,
		row.DetectionMethod, row.SeriesID, row.Season)
	if err != nil {
		return fmt.Errorf("store: save credits marker: %w", err)
	}
	return nil
}

// CreditsSeasonMarkers reads what is already known about a season, best evidence first.
//
// This is the single most valuable query in the subsystem. Credits within a season begin at
// a consistent point, so two decided episodes turn the next one's ten-minute tail scan into a
// three-minute one — which is most of the difference between a feature that is affordable on
// a NAS and one that is not.
func (s *Store) CreditsSeasonMarkers(
	ctx context.Context, seriesID string, season, limit int,
) ([]CreditsMarkerRow, error) {
	if seriesID == "" || limit <= 0 {
		return nil, nil
	}
	rows, err := s.pool.Query(ctx, `
		SELECT item_id, media_fingerprint, credits_start_ms, confidence, detection_method,
		       series_id, season_number, created_at, updated_at
		FROM credits_markers
		WHERE series_id = $1 AND season_number = $2
		ORDER BY confidence DESC, updated_at DESC
		LIMIT $3`, seriesID, season, limit)
	if err != nil {
		return nil, fmt.Errorf("store: credits season markers: %w", err)
	}
	defer rows.Close()

	out := make([]CreditsMarkerRow, 0, limit)
	for rows.Next() {
		var row CreditsMarkerRow
		if err := rows.Scan(&row.ItemID, &row.MediaFingerprint, &row.CreditsStartMs,
			&row.Confidence, &row.DetectionMethod, &row.SeriesID, &row.Season,
			&row.CreatedAt, &row.UpdatedAt); err != nil {
			return nil, err
		}
		out = append(out, row)
	}
	return out, rows.Err()
}

// CreditsWatchRow is one episode one viewer played, as candidate generation needs it.
type CreditsWatchRow struct {
	UserKey   string
	SeriesID  string
	Season    int
	Episode   int
	WatchedAt time.Time
	Completed bool
}

// CreditsRecentWatches is the demand signal, and the whole of it: one indexed read of
// sessions Tracearr has already imported.
//
// Nothing is written here and no new ingestion exists — the For You import already maintains
// this table and already resolves its rows to Emby ids. That reuse is why demand-driven
// candidate generation costs the gateway a single query every ten minutes rather than a
// second Tracearr integration.
//
// Episodes only, and only where the series resolved to something in Emby: a session that
// could not be matched cannot produce a scannable candidate, and filtering in SQL keeps the
// unmatched majority of an old library out of Go entirely.
func (s *Store) CreditsRecentWatches(
	ctx context.Context, since time.Time, limit int,
) ([]CreditsWatchRow, error) {
	if limit <= 0 {
		limit = 500
	}
	rows, err := s.pool.Query(ctx, `
		SELECT
			coalesce(nullif(tracearr_user_id, ''), lower(username)) AS user_key,
			emby_series_id,
			coalesce(season_number, 0),
			coalesce(episode_number, 0),
			coalesce(stopped_at, started_at) AS watched_at,
			watched OR (total_duration_ms > 0
				AND progress_ms::float8 / total_duration_ms::float8 >= 0.9) AS completed
		FROM tracearr_sessions
		WHERE lower(media_type) = 'episode'
		  AND emby_series_id <> ''
		  AND episode_number IS NOT NULL AND episode_number > 0
		  AND coalesce(stopped_at, started_at) >= $1
		  AND coalesce(nullif(tracearr_user_id, ''), lower(username)) <> ''
		ORDER BY coalesce(stopped_at, started_at) DESC
		LIMIT $2`, since.UTC(), limit)
	if err != nil {
		return nil, fmt.Errorf("store: credits recent watches: %w", err)
	}
	defer rows.Close()

	out := make([]CreditsWatchRow, 0, 64)
	for rows.Next() {
		var row CreditsWatchRow
		if err := rows.Scan(&row.UserKey, &row.SeriesID, &row.Season,
			&row.Episode, &row.WatchedAt, &row.Completed); err != nil {
			return nil, err
		}
		out = append(out, row)
	}
	return out, rows.Err()
}

// CreditsStopRow is one viewer leaving one episode.
type CreditsStopRow struct {
	UserKey     string
	PositionMs  int64
	RuntimeMs   int64
	NextEpisode bool
}

// CreditsStops reads where the household stopped one episode.
//
// The behavioural detector's entire input, and it needs no new table: Tracearr already
// records progress and completion per session. NextEpisode is derived rather than stored —
// a session for the following episode of the same series starting within a couple of minutes
// of this one ending is an auto-advance, which is the strongest form of this signal because
// it says the viewer was unambiguously looking at credits rather than deciding to stop.
func (s *Store) CreditsStops(ctx context.Context, itemID string) ([]CreditsStopRow, error) {
	if itemID == "" {
		return nil, nil
	}
	rows, err := s.pool.Query(ctx, `
		WITH plays AS (
			SELECT
				coalesce(nullif(tracearr_user_id, ''), lower(username)) AS user_key,
				emby_series_id,
				season_number,
				episode_number,
				progress_ms,
				total_duration_ms,
				coalesce(stopped_at, started_at) AS ended_at
			FROM tracearr_sessions
			WHERE emby_item_id = $1
			  AND progress_ms > 0
			  AND total_duration_ms > 0
			  AND coalesce(nullif(tracearr_user_id, ''), lower(username)) <> ''
		)
		SELECT
			plays.user_key,
			plays.progress_ms,
			plays.total_duration_ms,
			EXISTS (
				SELECT 1 FROM tracearr_sessions following
				WHERE following.emby_series_id = plays.emby_series_id
				  AND coalesce(nullif(following.tracearr_user_id, ''), lower(following.username))
				      = plays.user_key
				  AND following.season_number = plays.season_number
				  AND following.episode_number = plays.episode_number + 1
				  AND following.started_at BETWEEN plays.ended_at - interval '30 seconds'
				                              AND plays.ended_at + interval '3 minutes'
			) AS next_episode
		FROM plays
		LIMIT 200`, itemID)
	if err != nil {
		return nil, fmt.Errorf("store: credits stops: %w", err)
	}
	defer rows.Close()

	out := make([]CreditsStopRow, 0, 16)
	for rows.Next() {
		var row CreditsStopRow
		if err := rows.Scan(&row.UserKey, &row.PositionMs, &row.RuntimeMs,
			&row.NextEpisode); err != nil {
			return nil, err
		}
		out = append(out, row)
	}
	return out, rows.Err()
}

// CreditsEpisodeRow is one episode's position in its series.
type CreditsEpisodeRow struct {
	ItemID   string
	SeriesID string
	Season   int
	Episode  int
}

// CreditsSeriesEpisodes reads the numbering of every episode of the given series.
//
// One query for the handful of series a household is currently watching, rather than a
// lookup per candidate. The result becomes an in-memory index, so the look-ahead arithmetic
// — including stepping across a season boundary, which is exactly when somebody is most
// likely to keep going — is pure and testable with no database at all.
func (s *Store) CreditsSeriesEpisodes(
	ctx context.Context, seriesIDs []string,
) ([]CreditsEpisodeRow, error) {
	if len(seriesIDs) == 0 {
		return nil, nil
	}
	rows, err := s.pool.Query(ctx, `
		SELECT
			id,
			series_id,
			coalesce((payload->>'ParentIndexNumber')::int, 0),
			coalesce((payload->>'IndexNumber')::int, 0)
		FROM library_items
		WHERE type = 'Episode'
		  AND series_id = ANY($1)
		  AND payload->>'IndexNumber' IS NOT NULL`, seriesIDs)
	if err != nil {
		return nil, fmt.Errorf("store: credits series episodes: %w", err)
	}
	defer rows.Close()

	out := make([]CreditsEpisodeRow, 0, 128)
	for rows.Next() {
		var row CreditsEpisodeRow
		if err := rows.Scan(&row.ItemID, &row.SeriesID, &row.Season, &row.Episode); err != nil {
			return nil, err
		}
		if row.Episode <= 0 {
			continue
		}
		out = append(out, row)
	}
	return out, rows.Err()
}