From d83f7fadefeb75392ec03cab12652ee83d39af1f Mon Sep 17 00:00:00 2001 From: Quick <31828688+Quick104@users.noreply.github.com> Date: Fri, 26 Jun 2026 12:27:12 -0400 Subject: [PATCH] fix(audiobooks): mirror ABS playback into live sessions (#203) --- cmd/silo/main.go | 2 + internal/audiobooks/abs/file_handler.go | 3 + internal/audiobooks/abs/handler.go | 8 + internal/audiobooks/abs/native_sessions.go | 226 +++++++++++++++++ .../audiobooks/abs/native_sessions_test.go | 229 ++++++++++++++++++ internal/audiobooks/abs/play_response.go | 49 ++-- internal/audiobooks/abs/progress.go | 2 + internal/audiobooks/service.go | 7 +- 8 files changed, 504 insertions(+), 22 deletions(-) create mode 100644 internal/audiobooks/abs/native_sessions.go create mode 100644 internal/audiobooks/abs/native_sessions_test.go diff --git a/cmd/silo/main.go b/cmd/silo/main.go index 55c5325e..63de4212 100644 --- a/cmd/silo/main.go +++ b/cmd/silo/main.go @@ -1881,6 +1881,8 @@ func main() { AccessResolver: audiobooks.NewABSAccessResolver(absUserRepo, userStoreProvider), Recs: recommendations.NewRepo(deps.DB), Detail: absDetailSvc, + SessionMgr: sessionMgr, + SessionSyncer: deps.SessionSyncer, } absH := audiobooksService.BuildABSHandler(absHDeps) deps.ABSHandler = absH diff --git a/internal/audiobooks/abs/file_handler.go b/internal/audiobooks/abs/file_handler.go index a181c16d..4c1ec280 100644 --- a/internal/audiobooks/abs/file_handler.go +++ b/internal/audiobooks/abs/file_handler.go @@ -172,6 +172,9 @@ func (h *Handler) handlePublicTrack(w http.ResponseWriter, r *http.Request) { http.Error(w, "session expired", http.StatusGone) return } + if h.beginNativePlaybackTransport(sid) { + defer h.endNativePlaybackTransport(sid) + } access, err := h.accessFilterForAuth(r.Context(), ctxAuth{UserID: sess.UserID, ProfileID: sess.ProfileID}) if err != nil { diff --git a/internal/audiobooks/abs/handler.go b/internal/audiobooks/abs/handler.go index 2e6655fc..ea59995d 100644 --- a/internal/audiobooks/abs/handler.go +++ b/internal/audiobooks/abs/handler.go @@ -258,6 +258,14 @@ type Dependencies struct { // SocketIO is the Socket.io server mounted at /abs/socket.io/. May be nil; // the route is only registered when a non-nil value is supplied. SocketIO SocketIOServer + // NativeSessions mirrors ABS playback into Silo's native playback session + // manager so shared live-session views, limits, and stale-session cleanup + // see Audiobookshelf-compatible clients. May be nil; ABS playback still + // functions, but admin live-session visibility is unavailable. + NativeSessions PlaybackSessionManager + // NativeSessionSyncer flushes native session-manager state into the shared + // admin live-session table after ABS play/sync/close events. + NativeSessionSyncer PlaybackSessionSyncer // CoverResolver translates a raw silo poster path (e.g. // "local/audiobooks/.../original.webp") into a fully-qualified URL // the ABS client can fetch. Optional; when nil, /api/items/{id}/cover diff --git a/internal/audiobooks/abs/native_sessions.go b/internal/audiobooks/abs/native_sessions.go new file mode 100644 index 00000000..9345353f --- /dev/null +++ b/internal/audiobooks/abs/native_sessions.go @@ -0,0 +1,226 @@ +package abs + +import ( + "context" + "errors" + "fmt" + "log/slog" + "net" + "net/http" + "strconv" + "strings" + + "github.com/Silo-Server/silo-server/internal/clientip" + "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/playback" +) + +// PlaybackSessionManager is the native Silo playback-session surface ABS needs +// to make compatibility playback visible to live admin/session monitoring. +type PlaybackSessionManager interface { + StartSessionWithFilesContext(ctx context.Context, userID int, profileID string, effectiveFileID int, requestedFileID int, method playback.PlayMethod, transcodeAudio bool) (*playback.Session, error) + UpdateProgress(sessionID string, position float64, isPaused bool) error + UpdateStreamState(sessionID string, state playback.SessionStreamState) error + BeginTransport(sessionID string) error + EndTransport(sessionID string) error + SetProgressPersistenceDisabled(sessionID string, disabled bool) error + StopSession(sessionID string) error +} + +// PlaybackSessionSyncer flushes the local native-session snapshot into the +// shared admin live-session table. +type PlaybackSessionSyncer interface { + SyncNow(ctx context.Context) error +} + +func (h *Handler) startNativePlaybackSession( + r *http.Request, + a ctxAuth, + files []*models.MediaFile, + startPosition float64, +) (*playback.Session, error) { + if h == nil || h.deps.NativeSessions == nil || len(files) == 0 { + return nil, nil + } + userID, err := strconv.Atoi(a.UserID) + if err != nil { + return nil, fmt.Errorf("invalid ABS user id %q: %w", a.UserID, err) + } + + file := files[0] + sessionCtx := playback.WithClientInfo(r.Context(), absPlaybackClientInfoFromRequest(r)) + session, err := h.deps.NativeSessions.StartSessionWithFilesContext( + sessionCtx, + userID, + a.ProfileID, + file.ID, + file.ID, + playback.PlayDirect, + false, + ) + if err != nil { + return nil, err + } + + if err := h.deps.NativeSessions.SetProgressPersistenceDisabled(session.ID, true); err != nil { + slog.Warn("abs play: disable native progress persistence failed", + "session_id", session.ID, "error", err) + } else { + session.DisableProgressPersistence = true + } + + if startPosition > 0 { + if err := h.deps.NativeSessions.UpdateProgress(session.ID, startPosition, false); err != nil { + slog.Warn("abs play: seed native session progress failed", + "session_id", session.ID, "position", startPosition, "error", err) + } else { + session.Position = startPosition + session.IsPaused = false + } + } + + streamBitrateKbps := 0 + if file.Bitrate > 0 { + streamBitrateKbps = file.Bitrate + } + if err := h.deps.NativeSessions.UpdateStreamState(session.ID, playback.SessionStreamState{ + PlayMethod: playback.PlayDirect, + BasePlayMethod: playback.PlayDirect, + ClientIP: requestClientIP(r), + ClientName: session.ClientName, + ClientVersion: session.ClientVersion, + ClientUserAgent: session.ClientUserAgent, + StreamBitrateKbps: streamBitrateKbps, + }); err != nil { + slog.Warn("abs play: update native session stream state failed", + "session_id", session.ID, "error", err) + } else { + session.PlayMethod = playback.PlayDirect + session.BasePlayMethod = playback.PlayDirect + session.ClientIP = requestClientIP(r) + session.StreamBitrateKbps = streamBitrateKbps + } + + h.syncNativeSessionsNow(r.Context(), "abs_start") + return session, nil +} + +func (h *Handler) updateNativePlaybackProgress(ctx context.Context, sessionID string, position float64) { + if h == nil || h.deps.NativeSessions == nil || sessionID == "" { + return + } + if err := h.deps.NativeSessions.UpdateProgress(sessionID, position, false); err != nil { + if !errors.Is(err, playback.ErrSessionNotFound) { + slog.Warn("abs session sync: update native session progress failed", + "session_id", sessionID, "position", position, "error", err) + } + return + } + h.syncNativeSessionsNow(ctx, "abs_progress") +} + +func (h *Handler) stopNativePlaybackSession(ctx context.Context, sessionID string) { + if h == nil || h.deps.NativeSessions == nil || sessionID == "" { + return + } + if err := h.deps.NativeSessions.StopSession(sessionID); err != nil { + if !errors.Is(err, playback.ErrSessionNotFound) { + slog.Warn("abs session close: stop native session failed", + "session_id", sessionID, "error", err) + } + return + } + h.syncNativeSessionsNow(ctx, "abs_close") +} + +func (h *Handler) beginNativePlaybackTransport(sessionID string) bool { + if h == nil || h.deps.NativeSessions == nil || sessionID == "" { + return false + } + if err := h.deps.NativeSessions.BeginTransport(sessionID); err != nil { + if !errors.Is(err, playback.ErrSessionNotFound) { + slog.Warn("abs public track: begin native transport failed", + "session_id", sessionID, "error", err) + } + return false + } + return true +} + +func (h *Handler) endNativePlaybackTransport(sessionID string) { + if h == nil || h.deps.NativeSessions == nil || sessionID == "" { + return + } + if err := h.deps.NativeSessions.EndTransport(sessionID); err != nil && !errors.Is(err, playback.ErrSessionNotFound) { + slog.Warn("abs public track: end native transport failed", + "session_id", sessionID, "error", err) + } +} + +func (h *Handler) syncNativeSessionsNow(ctx context.Context, reason string) { + if h == nil || h.deps.NativeSessionSyncer == nil { + return + } + if err := h.deps.NativeSessionSyncer.SyncNow(ctx); err != nil { + slog.Error("abs: failed to sync native playback sessions", "reason", reason, "error", err) + } +} + +func writeNativePlaybackStartError(w http.ResponseWriter, err error) { + switch { + case errors.Is(err, playback.ErrTooManyStreams): + http.Error(w, "too many concurrent streams", http.StatusTooManyRequests) + case errors.Is(err, playback.ErrTooManyTranscodes): + http.Error(w, "too many concurrent transcodes", http.StatusTooManyRequests) + default: + slog.Error("abs play: start native playback session failed", "error", err) + http.Error(w, "failed to start playback session", http.StatusInternalServerError) + } +} + +func absPlaybackClientInfoFromRequest(r *http.Request) playback.ClientInfo { + if r == nil { + return playback.ClientInfo{Name: "Audiobookshelf"} + } + name := firstHeaderValue(r, + "X-Silo-Client", + "X-Client-Name", + "X-Device-Name", + "X-Emby-Client", + ) + if name == "" { + name = "Audiobookshelf" + } + return playback.ClientInfo{ + Name: name, + Version: firstHeaderValue(r, + "X-Silo-Client-Version", + "X-Client-Version", + "X-Emby-Client-Version", + ), + UserAgent: r.UserAgent(), + } +} + +func firstHeaderValue(r *http.Request, names ...string) string { + for _, name := range names { + if value := strings.TrimSpace(r.Header.Get(name)); value != "" { + return value + } + } + return "" +} + +func requestClientIP(r *http.Request) string { + if r == nil { + return "" + } + if ip := clientip.FromContext(r.Context()); ip != "" { + return ip + } + host, _, err := net.SplitHostPort(r.RemoteAddr) + if err == nil { + return host + } + return strings.TrimSpace(r.RemoteAddr) +} diff --git a/internal/audiobooks/abs/native_sessions_test.go b/internal/audiobooks/abs/native_sessions_test.go new file mode 100644 index 00000000..ad832df8 --- /dev/null +++ b/internal/audiobooks/abs/native_sessions_test.go @@ -0,0 +1,229 @@ +package abs + +import ( + "context" + "encoding/json" + "net/http" + "strings" + "testing" + "time" + + "github.com/Silo-Server/silo-server/internal/catalog" + "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/playback" +) + +type playStartMediaStore struct { + noopMediaStore + item *models.MediaItem + files []*models.MediaFile +} + +func (s *playStartMediaStore) GetAudiobookByID(_ context.Context, id string, _ catalog.AccessFilter) (*models.MediaItem, error) { + if s.item != nil && s.item.ContentID == id { + return s.item, nil + } + return nil, nil +} + +func (s *playStartMediaStore) GetMediaFiles(_ context.Context, contentID string, _ catalog.AccessFilter) ([]*models.MediaFile, error) { + if s.item != nil && s.item.ContentID == contentID { + return s.files, nil + } + return nil, nil +} + +type recordingPlaybackSessionSyncer struct { + calls int +} + +func (s *recordingPlaybackSessionSyncer) SyncNow(context.Context) error { + s.calls++ + return nil +} + +func TestHandlePlayStartCreatesNativePlaybackSession(t *testing.T) { + now := time.Now() + media := &playStartMediaStore{ + item: &models.MediaItem{ + ContentID: "book-1", + Type: "audiobook", + Title: "Native Session Book", + UpdatedAt: now, + AddedAt: &now, + }, + files: []*models.MediaFile{{ + ID: 42, + ContentID: "book-1", + FilePath: "/tmp/book.mp3", + FileSize: 1024, + Duration: 3600, + Bitrate: 128, + CodecAudio: "mp3", + }}, + } + absSessions := &fakePlaybackSessionStore{} + nativeSessions := playback.NewSessionManager(0, 0) + syncer := &recordingPlaybackSessionSyncer{} + progress := &fakeProgressStore{row: &ProgressRow{ + UserID: "1", + ProfileID: "profile-1", + ContentID: "book-1", + CurrentSeconds: 123.5, + DurationSeconds: 3600, + UpdatedAt: now, + }} + h := New(Dependencies{ + MediaStore: media, + ProgressStore: progress, + PlaybackSessionStore: absSessions, + NativeSessions: nativeSessions, + NativeSessionSyncer: syncer, + }) + + rec := dispatchABSWithParams( + http.MethodPost, + "/api/items/book-1/play", + map[string]string{"libraryItemId": "book-1"}, + nil, + "1", + "profile-1", + h.handlePlayStart, + ) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", rec.Code, rec.Body.String()) + } + + var body map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatalf("decode response: %v", err) + } + sessionID, _ := body["id"].(string) + if sessionID == "" { + t.Fatalf("response id is empty: %#v", body["id"]) + } + native, err := nativeSessions.GetSession(sessionID) + if err != nil { + t.Fatalf("native session %q missing: %v", sessionID, err) + } + if native.MediaFileID != 42 || native.RequestedMediaFileID != 42 { + t.Fatalf("native file ids = (%d, %d), want (42, 42)", native.MediaFileID, native.RequestedMediaFileID) + } + if !native.DisableProgressPersistence { + t.Fatalf("native session should disable progress persistence") + } + if native.Position != 123.5 { + t.Fatalf("native position = %v, want 123.5", native.Position) + } + if syncer.calls == 0 { + t.Fatalf("native session syncer was not called") + } + absSession, err := absSessions.GetPlaybackSession(context.Background(), sessionID) + if err != nil { + t.Fatalf("ABS session %q missing: %v", sessionID, err) + } + if absSession.CurrentPositionSeconds != 123.5 { + t.Fatalf("ABS session position = %v, want 123.5", absSession.CurrentPositionSeconds) + } + tracks, _ := body["audioTracks"].([]any) + if len(tracks) != 1 { + t.Fatalf("audioTracks length = %d, want 1", len(tracks)) + } + track, _ := tracks[0].(map[string]any) + if got, _ := track["contentUrl"].(string); got == "" || !strings.Contains(got, "/abs/public/session/"+sessionID+"/track/1") { + t.Fatalf("contentUrl = %q, want session-scoped URL", got) + } +} + +func TestHandleSessionSyncUpdatesNativePlaybackSession(t *testing.T) { + media := &playStartMediaStore{ + item: &models.MediaItem{ContentID: "book-1", Type: "audiobook", Title: "Book", UpdatedAt: time.Now()}, + } + absSessions := &fakePlaybackSessionStore{} + nativeSessions := playback.NewSessionManager(0, 0) + native, err := nativeSessions.StartSessionWithFilesContext(context.Background(), 1, "profile-1", 42, 42, playback.PlayDirect, false) + if err != nil { + t.Fatalf("start native session: %v", err) + } + _ = absSessions.InsertPlaybackSession(context.Background(), ABSPlaybackSession{ + ID: native.ID, + UserID: "1", + ProfileID: "profile-1", + ContentID: "book-1", + }) + syncer := &recordingPlaybackSessionSyncer{} + h := New(Dependencies{ + MediaStore: media, + ProgressStore: &fakeProgressStore{}, + PlaybackSessionStore: absSessions, + NativeSessions: nativeSessions, + NativeSessionSyncer: syncer, + }) + + rec := dispatchABSWithParams( + http.MethodPatch, + "/api/session/"+native.ID, + map[string]string{"sid": native.ID}, + []byte(`{"currentTime":55.25,"timeListening":10}`), + "1", + "profile-1", + h.handleSessionSync, + ) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", rec.Code, rec.Body.String()) + } + updated, err := nativeSessions.GetSession(native.ID) + if err != nil { + t.Fatalf("native session missing: %v", err) + } + if updated.Position != 55.25 { + t.Fatalf("native position = %v, want 55.25", updated.Position) + } + if updated.IsPaused { + t.Fatalf("native session should be marked playing") + } + if syncer.calls == 0 { + t.Fatalf("native session syncer was not called") + } +} + +func TestHandleSessionCloseStopsNativePlaybackSession(t *testing.T) { + absSessions := &fakePlaybackSessionStore{} + nativeSessions := playback.NewSessionManager(0, 0) + native, err := nativeSessions.StartSessionWithFilesContext(context.Background(), 1, "profile-1", 42, 42, playback.PlayDirect, false) + if err != nil { + t.Fatalf("start native session: %v", err) + } + _ = absSessions.InsertPlaybackSession(context.Background(), ABSPlaybackSession{ + ID: native.ID, + UserID: "1", + ProfileID: "profile-1", + ContentID: "book-1", + }) + syncer := &recordingPlaybackSessionSyncer{} + h := New(Dependencies{ + MediaStore: noopMediaStore{}, + PlaybackSessionStore: absSessions, + NativeSessions: nativeSessions, + NativeSessionSyncer: syncer, + }) + + rec := dispatchABSWithParams( + http.MethodPost, + "/api/session/"+native.ID+"/close", + map[string]string{"sid": native.ID}, + nil, + "1", + "profile-1", + h.handleSessionClose, + ) + if rec.Code != http.StatusNoContent { + t.Fatalf("status = %d, want 204; body=%s", rec.Code, rec.Body.String()) + } + if _, err := nativeSessions.GetSession(native.ID); err == nil { + t.Fatalf("native session still exists after close") + } + if syncer.calls == 0 { + t.Fatalf("native session syncer was not called") + } +} diff --git a/internal/audiobooks/abs/play_response.go b/internal/audiobooks/abs/play_response.go index 5f702877..2b73e3ab 100644 --- a/internal/audiobooks/abs/play_response.go +++ b/internal/audiobooks/abs/play_response.go @@ -59,21 +59,45 @@ func (h *Handler) handlePlayStart(w http.ResponseWriter, r *http.Request) { return } + // currentTime seeds the audio element's initial position so cross-device + // resume works. Lookup is best-effort: any error returns position 0, + // which is always correct for a first listen. + // + // Note: we deliberately do NOT emit a "progress" (0.0-1.0) field on the + // play session response. The canonical continuum-plugin handler omits it + // too — "progress" belongs to /me/progress responses, not playbackSession. + // The spec's Phase 0 row mentioning "currentTime AND progress fields" was + // over-specified; matching the canonical wire shape is the load-bearing + // requirement. + var currentTime float64 + currentTime, err = resolveResumeTime(r.Context(), h.deps.ProgressStore, a.UserID, a.ProfileID, contentID) + if err != nil { + slog.Debug("play: progress lookup failed", "user", a.UserID, "item", contentID, "err", err) + // currentTime is already 0 on error path; safe to continue. + } + // The {ino} parameter used by handleFileStream is a 0-based file index, but // we want iOS clients to resolve it via the stable MD5 derivation. Emit inos // that handleFileStream can reverse without a database lookup. baseURL := h.absBaseURL(r) sessionID := ulid.Make().String() + if nativeSession, err := h.startNativePlaybackSession(r, a, files, currentTime); err != nil { + writeNativePlaybackStartError(w, err) + return + } else if nativeSession != nil { + sessionID = nativeSession.ID + } // Persist the session row so subsequent PATCH /session/{sid} heartbeats // and POST /session/{sid}/close can find it. Without this, the session // ID is returned to the client but every sync/close lookup 404s. if h.deps.PlaybackSessionStore != nil { sess := ABSPlaybackSession{ - ID: sessionID, - UserID: a.UserID, - ProfileID: a.ProfileID, - ContentID: contentID, + ID: sessionID, + UserID: a.UserID, + ProfileID: a.ProfileID, + ContentID: contentID, + CurrentPositionSeconds: currentTime, } if len(files) > 0 { fid := files[0].ID @@ -111,23 +135,6 @@ func (h *Handler) handlePlayStart(w http.ResponseWriter, r *http.Request) { dateStr := now.UTC().Format("2006-01-02") dayOfWeek := now.UTC().Weekday().String() - // currentTime seeds the audio element's initial position so cross-device - // resume works. Lookup is best-effort: any error returns position 0, - // which is always correct for a first listen. - // - // Note: we deliberately do NOT emit a "progress" (0.0-1.0) field on the - // play session response. The canonical continuum-plugin handler omits it - // too — "progress" belongs to /me/progress responses, not playbackSession. - // The spec's Phase 0 row mentioning "currentTime AND progress fields" was - // over-specified; matching the canonical wire shape is the load-bearing - // requirement. - var currentTime float64 - currentTime, err = resolveResumeTime(r.Context(), h.deps.ProgressStore, a.UserID, a.ProfileID, contentID) - if err != nil { - slog.Debug("play: progress lookup failed", "user", a.UserID, "item", contentID, "err", err) - // currentTime is already 0 on error path; safe to continue. - } - playbackSession := map[string]any{ "id": sessionID, "userId": a.UserID, diff --git a/internal/audiobooks/abs/progress.go b/internal/audiobooks/abs/progress.go index 9f202aa2..2804efff 100644 --- a/internal/audiobooks/abs/progress.go +++ b/internal/audiobooks/abs/progress.go @@ -376,6 +376,7 @@ func (h *Handler) handleSessionSync(w http.ResponseWriter, r *http.Request) { "session_id", sid, "content_id", sess.ContentID, "error", err) } } + h.updateNativePlaybackProgress(r.Context(), sid, p.CurrentTime) // Realtime push to other connected clients. h.publish(a.UserID, "user_item_progress_updated", map[string]any{ @@ -421,6 +422,7 @@ func (h *Handler) handleSessionClose(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), http.StatusInternalServerError) return } + h.stopNativePlaybackSession(r.Context(), sid) h.publish(a.UserID, "user_session_closed", map[string]any{ "id": sid, diff --git a/internal/audiobooks/service.go b/internal/audiobooks/service.go index 7553365b..0b6b5db4 100644 --- a/internal/audiobooks/service.go +++ b/internal/audiobooks/service.go @@ -7,6 +7,7 @@ import ( "github.com/Silo-Server/silo-server/internal/audiobooks/abs" "github.com/Silo-Server/silo-server/internal/audiobooks/abssocket" "github.com/Silo-Server/silo-server/internal/catalog" + "github.com/Silo-Server/silo-server/internal/playback" "github.com/Silo-Server/silo-server/internal/recommendations" "github.com/Silo-Server/silo-server/internal/scanner" @@ -54,7 +55,9 @@ type ABSHandlerDeps struct { // Detail resolves audiobook poster S3 paths into fully-qualified URLs // that ABS clients can fetch. Optional; when nil, /api/items/{id}/cover // 404s rather than redirecting to an unreachable storage path. - Detail *catalog.DetailService + Detail *catalog.DetailService + SessionMgr *playback.SessionManager + SessionSyncer abs.PlaybackSessionSyncer } // absAuthAdapter is the narrow slice of internal/auth that BuildABSHandler @@ -158,6 +161,8 @@ func (s *Service) BuildABSHandler(deps ABSHandlerDeps) *abs.Handler { SmartCollectionStore: smartCollectionStore, RSSFeedStore: rssFeedStore, SocketIO: socketServer, + NativeSessions: deps.SessionMgr, + NativeSessionSyncer: deps.SessionSyncer, CoverResolver: func(ctx context.Context, path, variant string) string { if deps.Detail == nil { return ""