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() }