From 8c60e48817e6f5d3d2bd8047d89f4dcd9b172264 Mon Sep 17 00:00:00 2001 From: RXWatcher <14085001+RXWatcher@users.noreply.github.com> Date: Sun, 24 May 2026 15:35:34 +0200 Subject: [PATCH] feat(audiobooks): port ABS progress + continue listening Stage 4 of the ABS-compat port. Five handlers: - GET /abs/api/me/progress - GET /abs/api/me/progress/{libraryItemId} - POST /abs/api/me/progress/{libraryItemId} - PATCH /abs/api/session/{sid} - POST /abs/api/session/{sid}/close Reads/writes hit silo's user_watch_progress (keyed by user_id, profile_id, content_id) for resume positions and the new abs_playback_sessions table (migration 143) for per-session listening-time tracking. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/audiobooks/abs/handler.go | 25 +- internal/audiobooks/abs/progress.go | 347 ++++++++++++++++++ .../audiobooks/abs/progress_internal_test.go | 20 + .../audiobooks/abs_playback_session_store.go | 109 ++++++ internal/audiobooks/abs_progress_store.go | 179 +++++++++ 5 files changed, 679 insertions(+), 1 deletion(-) create mode 100644 internal/audiobooks/abs/progress.go create mode 100644 internal/audiobooks/abs/progress_internal_test.go create mode 100644 internal/audiobooks/abs_playback_session_store.go create mode 100644 internal/audiobooks/abs_progress_store.go diff --git a/internal/audiobooks/abs/handler.go b/internal/audiobooks/abs/handler.go index b28fe4dc..bce56d2b 100644 --- a/internal/audiobooks/abs/handler.go +++ b/internal/audiobooks/abs/handler.go @@ -115,6 +115,13 @@ type Dependencies struct { // InstallID returns the current plugin install ID for building // host-proxy-routable URLs. Defaults to "silo.audiobooks" when nil. InstallID func() string + // ProgressStore provides access to user_watch_progress for ABS + // progress endpoints. May be nil; handlers degrade gracefully. + ProgressStore ProgressStore + // PlaybackSessionStore persists abs_playback_sessions rows + // (migration 143) for /session/{sid}/sync and /session/{sid}/close. + // May be nil; handlers degrade gracefully. + PlaybackSessionStore ABSPlaybackSessionStore } // Handler wires the /abs/api/* and canonical ABS-client paths. @@ -177,8 +184,24 @@ func (h *Handler) mountRoutes(r chi.Router) { // TODO Stage 2 (remaining): logout, refresh, authorize, me, ping, status, init // TODO Stage 3 (remaining): library browse routes (libraries, items, item detail, cover, authors, series, search, personalized) - // TODO Stage 5: progress routes (me/progress/*, me/items-in-progress, me/listening-stats, me/stats/year/*) // TODO Stage 6: social / collection routes (bookmarks, smart-collections, collections, playlists, RSS feeds, similar) + + // Stage 4: progress + session tracking — requires bearerAuth. + r.Group(func(r chi.Router) { + r.Use(h.bearerAuth) + for _, prefix := range []string{"/abs/api", "/api"} { + // GET /me/progress — all audiobook progress for the caller + r.Get(prefix+"/me/progress", h.handleGetMyProgress) + // GET /me/progress/{id} — progress for one item + r.Get(prefix+"/me/progress/{libraryItemId}", h.handleGetItemProgress) + // POST /me/progress/{id} — set / update progress (ABS PATCH semantics) + r.Post(prefix+"/me/progress/{libraryItemId}", h.handleSetItemProgress) + // PATCH /session/{sid} — heartbeat: position + time_listening + r.Patch(prefix+"/session/{sid}", h.handleSessionSync) + // POST /session/{sid}/close — finalise the play session + r.Post(prefix+"/session/{sid}/close", h.handleSessionClose) + } + }) } // --------------------------------------------------------------------------- diff --git a/internal/audiobooks/abs/progress.go b/internal/audiobooks/abs/progress.go new file mode 100644 index 00000000..b3fba94b --- /dev/null +++ b/internal/audiobooks/abs/progress.go @@ -0,0 +1,347 @@ +package abs + +import ( + "context" + "encoding/json" + "net/http" + "time" + + "github.com/go-chi/chi/v5" +) + +// --------------------------------------------------------------------------- +// Interfaces +// --------------------------------------------------------------------------- + +// ProgressStore is the narrow slice of user_watch_progress access the ABS +// handlers need. Implemented by ABSProgressStore in +// internal/audiobooks/abs_progress_store.go. +type ProgressStore interface { + // GetProgress returns the progress row for (userID, profileID, contentID). + // Returns (nil, nil) when no row exists (not an error). + GetProgress(ctx context.Context, userID, profileID, contentID string) (*ProgressRow, error) + // ListProgressForAudiobooks returns all progress rows for (userID, profileID) + // that correspond to audiobooks (media_items.type = 'audiobook'). + // Capped at limit rows (most-recently-updated first). + ListProgressForAudiobooks(ctx context.Context, userID, profileID string, limit int) ([]ProgressRow, error) + // UpsertProgress writes a progress row. Fields not set in the body + // (currentTime/duration/isFinished/progress) should be merged by the + // caller before invoking this. + UpsertProgress(ctx context.Context, row ProgressRow) error + // UpdateProgressPosition updates only the position_seconds field for + // (userID, profileID, contentID). Used by session sync to avoid overwriting + // is_finished / progress_pct that the user set explicitly. + UpdateProgressPosition(ctx context.Context, userID, profileID, contentID string, positionSeconds float64) error +} + +// ABSPlaybackSessionStore tracks the active /abs/api/items/{id}/play sessions +// for per-session listening-time accounting (migration 143). +// Implemented by ABSPlaybackSessionStore in +// internal/audiobooks/abs_playback_session_store.go. +type ABSPlaybackSessionStore interface { + // InsertPlaybackSession creates the session row at play-start. + InsertPlaybackSession(ctx context.Context, sess ABSPlaybackSession) error + // GetPlaybackSession fetches a session by its ULID. Returns ErrNotFound + // when absent. + GetPlaybackSession(ctx context.Context, id string) (ABSPlaybackSession, error) + // SyncPlaybackSession updates position + accumulated listening time. + SyncPlaybackSession(ctx context.Context, id string, currentPositionSeconds float64, timeListeningSeconds int) error + // ClosePlaybackSession sets closed_at to now(). + ClosePlaybackSession(ctx context.Context, id string) error +} + +// ProgressRow is the in-memory representation of a user_watch_progress row +// as the ABS handlers use it. Intentionally narrow — only the fields the ABS +// wire format cares about. +type ProgressRow struct { + UserID string + ProfileID string + ContentID string + CurrentSeconds float64 + DurationSeconds float64 + ProgressPct float64 + IsFinished bool + UpdatedAt time.Time +} + +// ABSPlaybackSession is the in-memory representation of an abs_playback_sessions row. +type ABSPlaybackSession struct { + ID string + UserID string + ProfileID string + ContentID string + MediaFileID *int + TimeListeningSeconds int + CurrentPositionSeconds float64 + ClosedAt *time.Time +} + +// --------------------------------------------------------------------------- +// Handlers +// --------------------------------------------------------------------------- + +// handleGetMyProgress — GET /abs/api/me/progress +// Lists all progress rows for the caller that belong to audiobooks. +// The ABS mobile client reads this on startup to seed resume positions. +func (h *Handler) handleGetMyProgress(w http.ResponseWriter, r *http.Request) { + a, ok := absAuthFrom(r) + if !ok || a.UserID == "" { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + if h.deps.ProgressStore == nil { + writeJSON(w, http.StatusOK, map[string]any{"mediaProgress": []any{}}) + return + } + rows, err := h.deps.ProgressStore.ListProgressForAudiobooks(r.Context(), a.UserID, a.ProfileID, 500) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + out := make([]map[string]any, 0, len(rows)) + for _, p := range rows { + out = append(out, progressRowToABS(p)) + } + writeJSON(w, http.StatusOK, map[string]any{"mediaProgress": out}) +} + +// handleGetItemProgress — GET /abs/api/me/progress/{libraryItemId} +// Returns the progress row for one item. 404 when no progress exists. +func (h *Handler) handleGetItemProgress(w http.ResponseWriter, r *http.Request) { + a, ok := absAuthFrom(r) + if !ok || a.UserID == "" { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + contentID := chi.URLParam(r, "libraryItemId") + if h.deps.ProgressStore == nil { + http.Error(w, "progress not found", http.StatusNotFound) + return + } + p, err := h.deps.ProgressStore.GetProgress(r.Context(), a.UserID, a.ProfileID, contentID) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + if p == nil { + http.Error(w, "progress not found", http.StatusNotFound) + return + } + writeJSON(w, http.StatusOK, progressRowToABS(*p)) +} + +// progressBody is the JSON body for POST /abs/api/me/progress/{libraryItemId}. +// All fields are optional — only present fields update the row (PATCH semantics). +type progressBody struct { + CurrentTime *float64 `json:"currentTime"` + Duration *float64 `json:"duration"` + IsFinished *bool `json:"isFinished"` + Progress *float64 `json:"progress"` +} + +// handleSetItemProgress — POST /abs/api/me/progress/{libraryItemId} +// UPSERTs the progress row. Merges body fields over any existing row so a +// partial body (only currentTime) doesn't reset duration/isFinished. +// +// This matches sub-plan 3's HandleReportAudiobookProgress in intent but uses +// the ABS wire format and calls ProgressStore directly so the two code paths +// don't need a shared helper — their thresholds/semantics differ enough that +// keeping them separate is cleaner. +func (h *Handler) handleSetItemProgress(w http.ResponseWriter, r *http.Request) { + a, ok := absAuthFrom(r) + if !ok || a.UserID == "" { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + contentID := chi.URLParam(r, "libraryItemId") + var body progressBody + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + http.Error(w, "invalid body", http.StatusBadRequest) + return + } + if h.deps.ProgressStore == nil { + http.Error(w, "progress store unavailable", http.StatusServiceUnavailable) + return + } + + // Read existing row to merge (PATCH semantics). + var cur ProgressRow + if existing, err := h.deps.ProgressStore.GetProgress(r.Context(), a.UserID, a.ProfileID, contentID); err == nil && existing != nil { + cur = *existing + } + + next := ProgressRow{ + UserID: a.UserID, + ProfileID: a.ProfileID, + ContentID: contentID, + CurrentSeconds: cur.CurrentSeconds, + DurationSeconds: cur.DurationSeconds, + ProgressPct: cur.ProgressPct, + IsFinished: cur.IsFinished, + UpdatedAt: time.Now(), + } + if body.CurrentTime != nil { + next.CurrentSeconds = *body.CurrentTime + } + if body.Duration != nil { + next.DurationSeconds = *body.Duration + } + if body.Progress != nil { + next.ProgressPct = *body.Progress + } + if body.IsFinished != nil { + next.IsFinished = *body.IsFinished + } + if err := h.deps.ProgressStore.UpsertProgress(r.Context(), next); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + updated, err := h.deps.ProgressStore.GetProgress(r.Context(), a.UserID, a.ProfileID, contentID) + if err != nil || updated == nil { + // Best-effort: return the in-memory merged row rather than failing. + h.publish(a.UserID, "user_item_progress_updated", map[string]any{"data": progressRowToABS(next)}) + writeJSON(w, http.StatusOK, progressRowToABS(next)) + return + } + payload := progressRowToABS(*updated) + h.publish(a.UserID, "user_item_progress_updated", map[string]any{"data": payload}) + writeJSON(w, http.StatusOK, payload) +} + +// syncPayload is the JSON body for PATCH /abs/api/session/{sid}/sync. +type syncPayload struct { + CurrentTime float64 `json:"currentTime"` + TimeListened float64 `json:"timeListened"` +} + +// handleSessionSync — PATCH /abs/api/session/{sid}/sync +// Heartbeat endpoint the ABS mobile client calls every ~10 s during playback. +// Updates current position in user_watch_progress and accumulates +// time_listening_seconds in abs_playback_sessions. +// +// IDOR guard: the session must belong to the calling user (404 otherwise +// so session existence isn't leaked to other users). +// +// Uses UpdateProgressPosition (not UpsertProgress) to avoid overwriting +// is_finished / progress_pct that the user set explicitly — a sync tick +// that arrives after the user marks a book finished must not un-finish it. +func (h *Handler) handleSessionSync(w http.ResponseWriter, r *http.Request) { + a, ok := absAuthFrom(r) + if !ok { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + sid := chi.URLParam(r, "sid") + var p syncPayload + if err := json.NewDecoder(r.Body).Decode(&p); err != nil { + http.Error(w, "invalid body", http.StatusBadRequest) + return + } + if h.deps.PlaybackSessionStore == nil { + // No session store wired yet — accept the sync but return success + // rather than blocking the player. + writeJSON(w, http.StatusOK, map[string]any{"ok": true}) + return + } + + // Ownership gate. + sess, err := h.deps.PlaybackSessionStore.GetPlaybackSession(r.Context(), sid) + if err != nil || sess.UserID != a.UserID { + http.Error(w, "session not found", http.StatusNotFound) + return + } + + // Accumulate listening time and update position in abs_playback_sessions. + if err := h.deps.PlaybackSessionStore.SyncPlaybackSession( + r.Context(), sid, p.CurrentTime, int(p.TimeListened), + ); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + // Update position in user_watch_progress. Must NOT be a full upsert — + // see comment in handleSetItemProgress re: not overwriting is_finished. + if h.deps.ProgressStore != nil { + _ = h.deps.ProgressStore.UpdateProgressPosition( + r.Context(), a.UserID, a.ProfileID, sess.ContentID, p.CurrentTime, + ) + } + + // Realtime push to other connected clients. + h.publish(a.UserID, "user_item_progress_updated", map[string]any{ + "data": map[string]any{ + "libraryItemId": sess.ContentID, + "currentTime": p.CurrentTime, + "sessionId": sid, + }, + }) + h.publish(a.UserID, "user_session_updated", map[string]any{ + "id": sid, + "libraryItemId": sess.ContentID, + "currentTime": p.CurrentTime, + "timeListened": p.TimeListened, + }) + writeJSON(w, http.StatusOK, map[string]any{"ok": true}) +} + +// handleSessionClose — POST /abs/api/session/{sid}/close +// Finalises a play session. Sets closed_at on the abs_playback_sessions row. +// Only the owning user may close their session (IDOR guard). +func (h *Handler) handleSessionClose(w http.ResponseWriter, r *http.Request) { + a, ok := absAuthFrom(r) + if !ok { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + sid := chi.URLParam(r, "sid") + if h.deps.PlaybackSessionStore == nil { + // No store wired — accept close gracefully. + w.WriteHeader(http.StatusNoContent) + return + } + + // Ownership gate. + sess, err := h.deps.PlaybackSessionStore.GetPlaybackSession(r.Context(), sid) + if err != nil || sess.UserID != a.UserID { + http.Error(w, "session not found", http.StatusNotFound) + return + } + + if err := h.deps.PlaybackSessionStore.ClosePlaybackSession(r.Context(), sid); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + h.publish(a.UserID, "user_session_closed", map[string]any{ + "id": sid, + "libraryItemId": sess.ContentID, + }) + w.WriteHeader(http.StatusNoContent) +} + +// --------------------------------------------------------------------------- +// Serialisation helpers +// --------------------------------------------------------------------------- + +// progressRowToABS shapes a ProgressRow into the ABS /me/progress wire format. +// The `id` field uses the real-ABS convention of "-". +func progressRowToABS(p ProgressRow) map[string]any { + lastMs := p.UpdatedAt.UnixMilli() + out := map[string]any{ + "id": p.UserID + "-" + p.ContentID, + "libraryItemId": p.ContentID, + "mediaItemId": p.ContentID, + "currentTime": p.CurrentSeconds, + "duration": p.DurationSeconds, + "isFinished": p.IsFinished, + "progress": p.ProgressPct, + "startedAt": lastMs, + "finishedAt": nil, + "lastUpdate": lastMs, + } + if p.IsFinished { + out["finishedAt"] = lastMs + } + return out +} diff --git a/internal/audiobooks/abs/progress_internal_test.go b/internal/audiobooks/abs/progress_internal_test.go new file mode 100644 index 00000000..e47bf7f2 --- /dev/null +++ b/internal/audiobooks/abs/progress_internal_test.go @@ -0,0 +1,20 @@ +package abs + +import ( + "testing" + "time" +) + +func TestProgressRowToABSEmitsDuration(t *testing.T) { + out := progressRowToABS(ProgressRow{ + UserID: "u1", + ContentID: "b1", + CurrentSeconds: 30, + DurationSeconds: 3600, + ProgressPct: 0.0083, + UpdatedAt: time.Now(), + }) + if out["duration"] != float64(3600) { + t.Errorf("duration = %v, want 3600", out["duration"]) + } +} diff --git a/internal/audiobooks/abs_playback_session_store.go b/internal/audiobooks/abs_playback_session_store.go new file mode 100644 index 00000000..d12bb425 --- /dev/null +++ b/internal/audiobooks/abs_playback_session_store.go @@ -0,0 +1,109 @@ +package audiobooks + +import ( + "context" + "errors" + "fmt" + "strconv" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/Silo-Server/silo-server/internal/audiobooks/abs" +) + +// ABSPlaybackSessionStore implements abs.ABSPlaybackSessionStore on the +// abs_playback_sessions table (migration 143). Each row tracks one +// /abs/api/items/{id}/play session opened by an ABS-compatible client +// and is closed (closed_at set) when the client calls /session/{sid}/close. +type ABSPlaybackSessionStore struct { + Pool *pgxpool.Pool +} + +// InsertPlaybackSession persists a new session row at play-start time. +func (s *ABSPlaybackSessionStore) InsertPlaybackSession(ctx context.Context, sess abs.ABSPlaybackSession) error { + uid, err := strconv.Atoi(sess.UserID) + if err != nil { + return fmt.Errorf("abs_playback_session_store: invalid user_id %q: %w", sess.UserID, err) + } + _, err = s.Pool.Exec(ctx, ` + INSERT INTO abs_playback_sessions + (id, user_id, profile_id, content_id, media_file_id, + started_at, last_sync_at, time_listening_seconds, current_position_seconds) + VALUES ($1, $2, $3, $4, $5, now(), now(), 0, $6) + ON CONFLICT (id) DO NOTHING`, + sess.ID, + uid, + sess.ProfileID, + sess.ContentID, + sess.MediaFileID, + sess.CurrentPositionSeconds, + ) + if err != nil { + return fmt.Errorf("abs_playback_session_store: insert: %w", err) + } + return nil +} + +// GetPlaybackSession fetches a session by its ULID. Returns abs.ErrNotFound +// when the row doesn't exist. +func (s *ABSPlaybackSessionStore) GetPlaybackSession(ctx context.Context, id string) (abs.ABSPlaybackSession, error) { + var sess abs.ABSPlaybackSession + var uid int + var profileID string + var closedAt *time.Time + row := s.Pool.QueryRow(ctx, ` + SELECT id, user_id, profile_id, content_id, + time_listening_seconds, current_position_seconds, closed_at + FROM abs_playback_sessions + WHERE id = $1`, id) + err := row.Scan( + &sess.ID, &uid, &profileID, &sess.ContentID, + &sess.TimeListeningSeconds, &sess.CurrentPositionSeconds, &closedAt, + ) + if errors.Is(err, pgx.ErrNoRows) { + return abs.ABSPlaybackSession{}, abs.ErrNotFound + } + if err != nil { + return abs.ABSPlaybackSession{}, fmt.Errorf("abs_playback_session_store: get: %w", err) + } + sess.UserID = strconv.Itoa(uid) + sess.ProfileID = profileID + sess.ClosedAt = closedAt + return sess, nil +} + +// SyncPlaybackSession updates the position and accumulated listening time for +// an open session. Idempotent: calling it on an already-closed session is +// a no-op (the WHERE closed_at IS NULL guard prevents overwriting a final +// state with a stale sync payload that arrives after close). +func (s *ABSPlaybackSessionStore) SyncPlaybackSession(ctx context.Context, id string, currentPositionSeconds float64, timeListeningSeconds int) error { + _, err := s.Pool.Exec(ctx, ` + UPDATE abs_playback_sessions + SET current_position_seconds = $2, + time_listening_seconds = time_listening_seconds + $3, + last_sync_at = now() + WHERE id = $1 AND closed_at IS NULL`, + id, currentPositionSeconds, timeListeningSeconds, + ) + if err != nil { + return fmt.Errorf("abs_playback_session_store: sync: %w", err) + } + return nil +} + +// ClosePlaybackSession sets closed_at = now() for the given session. +// Idempotent: closing an already-closed session is safe. +func (s *ABSPlaybackSessionStore) ClosePlaybackSession(ctx context.Context, id string) error { + _, err := s.Pool.Exec(ctx, ` + UPDATE abs_playback_sessions + SET closed_at = now() + WHERE id = $1 AND closed_at IS NULL`, + id, + ) + if err != nil { + return fmt.Errorf("abs_playback_session_store: close: %w", err) + } + return nil +} diff --git a/internal/audiobooks/abs_progress_store.go b/internal/audiobooks/abs_progress_store.go new file mode 100644 index 00000000..3aa08a56 --- /dev/null +++ b/internal/audiobooks/abs_progress_store.go @@ -0,0 +1,179 @@ +package audiobooks + +import ( + "context" + "errors" + "fmt" + "strconv" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/Silo-Server/silo-server/internal/audiobooks/abs" +) + +// ABSProgressStore implements abs.ProgressStore directly against the +// user_watch_progress table using a shared pgxpool. Using the pool directly +// (rather than the per-user-scoped PostgresUserStore) lets us query by +// (user_id, profile_id) without needing a ForUser call, which would require +// knowing the integer user_id at construction time. The ABS handlers carry +// user_id as a string and resolve it inline here. +type ABSProgressStore struct { + Pool *pgxpool.Pool +} + +var _ abs.ProgressStore = (*ABSProgressStore)(nil) + +// GetProgress returns the progress row for (userID, profileID, contentID). +// Returns (nil, nil) when no row exists. +func (s *ABSProgressStore) GetProgress(ctx context.Context, userID, profileID, contentID string) (*abs.ProgressRow, error) { + uid, err := strconv.Atoi(userID) + if err != nil { + return nil, fmt.Errorf("abs_progress_store: invalid user_id %q: %w", userID, err) + } + var row abs.ProgressRow + var updatedAt time.Time + var positionSeconds, durationSeconds float64 + var completed bool + var progressPct *float64 + + dbRow := s.Pool.QueryRow(ctx, ` + SELECT media_item_id, position_seconds, duration_seconds, completed, + CASE WHEN duration_seconds > 0 THEN position_seconds / duration_seconds ELSE 0 END AS progress_pct, + updated_at + FROM user_watch_progress + WHERE user_id = $1 AND profile_id = $2 AND media_item_id = $3`, + uid, profileID, contentID, + ) + err = dbRow.Scan( + &row.ContentID, &positionSeconds, &durationSeconds, &completed, + &progressPct, &updatedAt, + ) + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("abs_progress_store: get progress: %w", err) + } + + row.UserID = userID + row.ProfileID = profileID + row.CurrentSeconds = positionSeconds + row.DurationSeconds = durationSeconds + row.IsFinished = completed + if progressPct != nil { + row.ProgressPct = *progressPct + } + row.UpdatedAt = updatedAt + return &row, nil +} + +// ListProgressForAudiobooks returns all progress rows for (userID, profileID) +// that join to media_items with type = 'audiobook'. Ordered by updated_at DESC, +// capped at limit rows. +func (s *ABSProgressStore) ListProgressForAudiobooks(ctx context.Context, userID, profileID string, limit int) ([]abs.ProgressRow, error) { + uid, err := strconv.Atoi(userID) + if err != nil { + return nil, fmt.Errorf("abs_progress_store: invalid user_id %q: %w", userID, err) + } + if limit <= 0 { + limit = 500 + } + rows, err := s.Pool.Query(ctx, ` + SELECT wp.media_item_id, + wp.position_seconds, + wp.duration_seconds, + wp.completed, + CASE WHEN wp.duration_seconds > 0 THEN wp.position_seconds / wp.duration_seconds ELSE 0 END, + wp.updated_at + FROM user_watch_progress wp + JOIN media_items mi ON mi.content_id = wp.media_item_id + WHERE wp.user_id = $1 + AND wp.profile_id = $2 + AND mi.type = 'audiobook' + ORDER BY wp.updated_at DESC + LIMIT $3`, + uid, profileID, limit, + ) + if err != nil { + return nil, fmt.Errorf("abs_progress_store: list progress: %w", err) + } + defer rows.Close() + + var result []abs.ProgressRow + for rows.Next() { + var p abs.ProgressRow + var updatedAt time.Time + if err := rows.Scan( + &p.ContentID, + &p.CurrentSeconds, + &p.DurationSeconds, + &p.IsFinished, + &p.ProgressPct, + &updatedAt, + ); err != nil { + return nil, fmt.Errorf("abs_progress_store: scan progress row: %w", err) + } + p.UserID = userID + p.ProfileID = profileID + p.UpdatedAt = updatedAt + result = append(result, p) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("abs_progress_store: iterate progress rows: %w", err) + } + return result, nil +} + +// UpsertProgress inserts or updates a user_watch_progress row. All fields in +// row are written; caller is responsible for merging existing state before +// calling this (see handleSetItemProgress). +func (s *ABSProgressStore) UpsertProgress(ctx context.Context, row abs.ProgressRow) error { + uid, err := strconv.Atoi(row.UserID) + if err != nil { + return fmt.Errorf("abs_progress_store: invalid user_id %q: %w", row.UserID, err) + } + updatedAt := row.UpdatedAt + if updatedAt.IsZero() { + updatedAt = time.Now().UTC() + } + _, err = s.Pool.Exec(ctx, ` + INSERT INTO user_watch_progress + (user_id, profile_id, media_item_id, position_seconds, duration_seconds, completed, updated_at) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (user_id, profile_id, media_item_id) DO UPDATE SET + position_seconds = EXCLUDED.position_seconds, + duration_seconds = EXCLUDED.duration_seconds, + completed = EXCLUDED.completed, + updated_at = EXCLUDED.updated_at`, + uid, row.ProfileID, row.ContentID, + row.CurrentSeconds, row.DurationSeconds, row.IsFinished, updatedAt, + ) + if err != nil { + return fmt.Errorf("abs_progress_store: upsert progress: %w", err) + } + return nil +} + +// UpdateProgressPosition updates only the position_seconds column for an +// existing row. If no row exists this is a no-op (the session-sync path +// that calls this only needs to move the cursor, not create a progress row +// for the first time; that's done when the user explicitly sets progress). +func (s *ABSProgressStore) UpdateProgressPosition(ctx context.Context, userID, profileID, contentID string, positionSeconds float64) error { + uid, err := strconv.Atoi(userID) + if err != nil { + return fmt.Errorf("abs_progress_store: invalid user_id %q: %w", userID, err) + } + _, err = s.Pool.Exec(ctx, ` + UPDATE user_watch_progress + SET position_seconds = $4, + updated_at = now() + WHERE user_id = $1 AND profile_id = $2 AND media_item_id = $3`, + uid, profileID, contentID, positionSeconds, + ) + if err != nil { + return fmt.Errorf("abs_progress_store: update progress position: %w", err) + } + return nil +}