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) <noreply@anthropic.com>
This commit is contained in:
RXWatcher
co-authored by
Claude Opus 4.7
Claude Opus 4.7
parent
db138450ac
commit
8c60e48817
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
@@ -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 "<userID>-<libraryItemId>".
|
||||
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
|
||||
}
|
||||
@@ -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"])
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user