diff --git a/AGENTS.md b/AGENTS.md index a28359c..d4ce576 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -115,8 +115,8 @@ The server is written from the MPRIS 2.2 specification. The Go MPRIS server libr | `client.go` | HTTP client with 10s timeout, 50MB response limit, auth header, request logging | | `types.go` | All ABS data types. Custom UnmarshalJSON for MediaMetadata.Series (handles both object and array). | | `libraries.go` | GetLibraries, GetPersonalized, GetLibraryItems (paginated), GetLibraryItemsBatch (10 concurrent), GetLibrarySeries, GetSeriesContents, FilterAudioLibraries (sample-based audio detection) | -| `playback.go` | StartPlaySession, StartEpisodePlaySession, SyncSession, CloseSession | -| `progress.go`, `me.go` | UpdateProgress, GetMediaProgress (with bookmarks), UpdateEpisodeProgress, GetMe (user + all media progress) | +| `playback.go` | StartPlaySession, StartEpisodePlaySession, SyncSession, CloseSession, CloseSessionNoSync (sends no position) | +| `progress.go`, `me.go` | UpdateProgress, UpdatePosition (no `isFinished` key), GetMediaProgress (with bookmarks), UpdateEpisodeProgress, GetMe (user + all media progress) | | `bookmarks.go` | CreateBookmark, UpdateBookmark, DeleteBookmark, GetBookmarks | ### Data (internal/db/) @@ -140,8 +140,10 @@ Styles from theme config. Format helpers (timestamps, durations). Components: He - Skeleton loading with generation-gated reveal delays (home: 500ms, library: 150ms) - Cache-first data loading through `internal/cache.Client`: home and library views hit the persistent cache first; misses trigger a background rebuild - `q` always quits, `esc`/left goes back (never quits). `h`/`l` seek only, arrows navigate only +- A quit (`q` or ctrl+c) returns only `tea.Quit` and keeps the session in the model. `Cleanup` sends the last position, and `main` waits for it. Bubble Tea does not wait for a command at a quit, so a quit must not send the position in a command. - `ctrl+p` opens the global command palette from any screen - Duration from ABS takes precedence over mpv-reported duration (streaming content reports 0) +- The listening time (`timeListened`) is the time of the clock between two position ticks of a player that plays (`trackListeningTime`). It is not the change of the position: at the speed 2.0, the position moves two seconds in each second. A test moves the clock with `Model.now`. - pine finds the end of a track through `errors.Is(err, player.ErrClosed)` on a position tick. spruce uses `PlayerEndMsg`, and a tick with `ErrClosed` can arrive before it. Do not match the text of a player error. ## Testing diff --git a/internal/abs/playback.go b/internal/abs/playback.go index 9fb7570..dbaaae4 100644 --- a/internal/abs/playback.go +++ b/internal/abs/playback.go @@ -89,7 +89,8 @@ func (c *Client) SyncSession(ctx context.Context, sessionID string, currentTime, return nil } -// CloseSession closes an active playback session. +// CloseSession closes an active playback session. The server stores the +// position and the listening time before it closes the session. func (c *Client) CloseSession(ctx context.Context, sessionID string, currentTime, timeListened float64) error { path := fmt.Sprintf("/api/session/%s/close", sessionID) body := syncSessionRequest{ @@ -103,3 +104,15 @@ func (c *Client) CloseSession(ctx context.Context, sessionID string, currentTime } return nil } + +// CloseSessionNoSync closes an active playback session and sends no position, +// so the server keeps the progress that it has. +func (c *Client) CloseSessionNoSync(ctx context.Context, sessionID string) error { + path := fmt.Sprintf("/api/session/%s/close", sessionID) + logger.Debug("closing playback session with no sync", "sessionID", sessionID) + _, err := c.do(ctx, http.MethodPost, path, nil) + if err != nil { + return fmt.Errorf("close session: %w", err) + } + return nil +} diff --git a/internal/abs/playback_test.go b/internal/abs/playback_test.go index c5aafb4..39b17a5 100644 --- a/internal/abs/playback_test.go +++ b/internal/abs/playback_test.go @@ -247,3 +247,26 @@ func TestCloseSessionHTTP(t *testing.T) { t.Errorf("timeListened = %f, want 180.0", body.TimeListened) } } + +func TestCloseSessionNoSyncHTTP(t *testing.T) { + var capturedBody []byte + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/session/session-abc-123/close" { + t.Errorf("path = %q, want /api/session/session-abc-123/close", r.URL.Path) + } + if r.Method != http.MethodPost { + t.Errorf("method = %q, want POST", r.Method) + } + capturedBody, _ = io.ReadAll(r.Body) + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + c := NewClient(srv.URL, "tok") + if err := c.CloseSessionNoSync(context.Background(), "session-abc-123"); err != nil { + t.Fatalf("CloseSessionNoSync() error: %v", err) + } + if len(capturedBody) != 0 { + t.Errorf("body = %q, want no body", capturedBody) + } +} diff --git a/internal/abs/progress.go b/internal/abs/progress.go index d8d76da..038ee3e 100644 --- a/internal/abs/progress.go +++ b/internal/abs/progress.go @@ -8,10 +8,13 @@ import ( ) // updateProgressRequest is the body for PATCH /api/me/progress/{id}. +// The server reads "isFinished": false as "mark as not finished": it removes +// the finished mark and sets the position to 0. With no isFinished key, the +// server sets or keeps the mark by its own rule for the position. type updateProgressRequest struct { CurrentTime float64 `json:"currentTime"` Progress float64 `json:"progress"` - IsFinished bool `json:"isFinished"` + IsFinished *bool `json:"isFinished,omitempty"` } // UpdateProgress updates the user's progress on a library item. @@ -20,7 +23,7 @@ func (c *Client) UpdateProgress(ctx context.Context, itemID string, currentTime, body := updateProgressRequest{ CurrentTime: currentTime, Progress: progress, - IsFinished: isFinished, + IsFinished: &isFinished, } _, err := c.do(ctx, http.MethodPatch, path, body) if err != nil { @@ -29,6 +32,24 @@ func (c *Client) UpdateProgress(ctx context.Context, itemID string, currentTime, return nil } +// UpdatePosition updates the user's position in a library item, or in a +// podcast episode if episodeID is not empty. It sends no isFinished key. +func (c *Client) UpdatePosition(ctx context.Context, itemID, episodeID string, currentTime, progress float64) error { + path := fmt.Sprintf("/api/me/progress/%s", itemID) + if episodeID != "" { + path += "/" + episodeID + } + body := updateProgressRequest{ + CurrentTime: currentTime, + Progress: progress, + } + _, err := c.do(ctx, http.MethodPatch, path, body) + if err != nil { + return fmt.Errorf("update position: %w", err) + } + return nil +} + // GetMediaProgress returns the user's media progress (including bookmarks) // for a specific library item. func (c *Client) GetMediaProgress(ctx context.Context, itemID string) (*MediaProgressWithBookmarks, error) { @@ -50,7 +71,7 @@ func (c *Client) UpdateEpisodeProgress(ctx context.Context, itemID, episodeID st body := updateProgressRequest{ CurrentTime: currentTime, Progress: progress, - IsFinished: isFinished, + IsFinished: &isFinished, } _, err := c.do(ctx, http.MethodPatch, path, body) if err != nil { diff --git a/internal/abs/progress_test.go b/internal/abs/progress_test.go index 8ae2f6a..fbc6c05 100644 --- a/internal/abs/progress_test.go +++ b/internal/abs/progress_test.go @@ -6,6 +6,7 @@ import ( "io" "net/http" "net/http/httptest" + "reflect" "testing" ) @@ -78,6 +79,47 @@ func TestUpdateProgressFinishedHTTP(t *testing.T) { } } +func TestUpdatePositionHTTP(t *testing.T) { + tests := []struct { + name string + episodeID string + wantPath string + }{ + {name: "book", wantPath: "/api/me/progress/li-001"}, + {name: "episode", episodeID: "ep-001", wantPath: "/api/me/progress/li-001/ep-001"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var capturedBody []byte + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != tt.wantPath { + t.Errorf("path = %q, want %s", r.URL.Path, tt.wantPath) + } + if r.Method != http.MethodPatch { + t.Errorf("method = %q, want PATCH", r.Method) + } + capturedBody, _ = io.ReadAll(r.Body) + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + c := NewClient(srv.URL, "tok") + if err := c.UpdatePosition(context.Background(), "li-001", tt.episodeID, 500.5, 0.45); err != nil { + t.Fatalf("UpdatePosition() error: %v", err) + } + + var body map[string]any + if err := json.Unmarshal(capturedBody, &body); err != nil { + t.Fatalf("failed to unmarshal request body: %v", err) + } + want := map[string]any{"currentTime": 500.5, "progress": 0.45} + if !reflect.DeepEqual(body, want) { + t.Errorf("body = %v, want %v (no isFinished key)", body, want) + } + }) + } +} + func TestGetMediaProgressHTTP(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/api/me/progress/li-001" { diff --git a/internal/app/e2e_test.go b/internal/app/e2e_test.go index 20c9e10..dde29f1 100644 --- a/internal/app/e2e_test.go +++ b/internal/app/e2e_test.go @@ -245,12 +245,12 @@ func TestE2E_DetailToPlayback(t *testing.T) { // The tick commands use time.After so they'll timeout in feedCmd — that's fine m = feedCmdChain(m, cmd, 3) - // Simulate position updates + // Simulate position updates: the first tick of the player, and then 10 + // seconds of play mp.mu.Lock() mp.position = 52.0 mp.mu.Unlock() - res, _ := m.Update(player.PositionMsg{Position: 52.0, Duration: 3600.0, Paused: false, Generation: 1}) - m = res.(Model) + m = listen(playTick(m, 0, 42.0), 52.0) if m.player.Position != 52.0 { t.Errorf("position = %f, want 52.0", m.player.Position) @@ -263,7 +263,7 @@ func TestE2E_DetailToPlayback(t *testing.T) { assertViewContains(t, m, "The Great Gatsby") // Trigger sync - res, cmd = m.Update(SyncTickMsg{}) + res, cmd := m.Update(SyncTickMsg{}) m = res.(Model) executeBatchCmds(cmd) @@ -1048,11 +1048,10 @@ func TestE2E_FullJourney(t *testing.T) { mp.mu.Lock() mp.position = 60.0 mp.mu.Unlock() - res, _ := m.Update(player.PositionMsg{Position: 60.0, Duration: 3600.0, Paused: false, Generation: 1}) - m = res.(Model) + m = playTick(playTick(m, 0, 59.0), 1, 60.0) - if m.timeListened == 0 { - t.Error("timeListened should be > 0 after position update") + if m.timeListened != 1.0 { + t.Errorf("timeListened = %f, want 1.0 after one second of play", m.timeListened) } // === Phase 7: Create bookmark === @@ -1062,7 +1061,7 @@ func TestE2E_FullJourney(t *testing.T) { assertAPICallMade(t, log, "POST", "/api/me/item/") // === Phase 8: Sync tick === - res, cmd = m.Update(SyncTickMsg{}) + res, cmd := m.Update(SyncTickMsg{}) m = res.(Model) executeBatchCmds(cmd) assertAPICallMade(t, log, "POST", "/api/session/sess-e2e/sync") @@ -1086,13 +1085,13 @@ func TestE2E_FullJourney(t *testing.T) { } // === Phase 11: Quit from Home (stops playback and exits) === - _, cmd = m.Update(tea.KeyMsg{Type: tea.KeyRunes, Runes: []rune{'q'}}) + res, cmd = m.Update(tea.KeyMsg{Type: tea.KeyRunes, Runes: []rune{'q'}}) if cmd == nil { t.Fatal("'q' on Home with empty back stack should quit") } - // Should batch stop commands with quit - executeBatchCmds(cmd) + // main calls Cleanup on the final model, which closes the session. + res.(Model).Cleanup() assertAPICallMade(t, log, "POST", "/api/session/sess-e2e/close") // === Verify API call sequence === @@ -1669,13 +1668,13 @@ func TestE2E_QQuitsAndStopsPlayback(t *testing.T) { m.player.Duration = 3600.0 // Press q — should quit and stop playback - _, cmd := m.Update(tea.KeyMsg{Type: tea.KeyRunes, Runes: []rune{'q'}}) + res, cmd := m.Update(tea.KeyMsg{Type: tea.KeyRunes, Runes: []rune{'q'}}) if cmd == nil { t.Fatal("q should return a command") } - // Execute the batch — should include stop + quit - executeBatchCmds(cmd) + // main calls Cleanup on the final model, which closes the session. + res.(Model).Cleanup() // Verify close session API was called assertAPICallMade(t, log, "POST", "/api/session/sess-e2e/close") diff --git a/internal/app/messages.go b/internal/app/messages.go index 822af3c..c5da3a7 100644 --- a/internal/app/messages.go +++ b/internal/app/messages.go @@ -62,6 +62,7 @@ type PlaySessionData struct { ItemID string EpisodeID string CurrentTime float64 + SyncedTime float64 // book position that the server has Duration float64 Title string Authors []string @@ -81,6 +82,17 @@ type QueueEntry struct { // SyncTickMsg fires every 30 seconds to sync playback progress with ABS. type SyncTickMsg struct{} +// SessionSyncedMsg carries the result of a sync of the playback session. +// NewSessionID is the session that pine opened because the server did not +// have the session SessionID. +type SessionSyncedMsg struct { + SessionID string + NewSessionID string + CurrentTime float64 + TimeListened float64 + Err error +} + // PlaybackErrorMsg carries an error from the playback lifecycle. type PlaybackErrorMsg struct { Err error diff --git a/internal/app/model.go b/internal/app/model.go index 13c32ba..7352cbf 100644 --- a/internal/app/model.go +++ b/internal/app/model.go @@ -55,6 +55,7 @@ type Model struct { itemID string episodeID string timeListened float64 + lastPlayingTick time.Time lastSyncPos float64 playGeneration uint64 chapters []abs.Chapter @@ -100,6 +101,9 @@ type Model struct { lastPlayedItemID string lastPlayedAuthors []string currentAuthors []string + + // now replaces time.Now in a test. + now func() time.Time } // New creates a new root model. If client is non-nil (authenticated), @@ -636,6 +640,9 @@ func (m Model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { } return m, cmd + case SessionSyncedMsg: + return m.handleSessionSynced(msg), nil + case player.PlayerLaunchErrMsg: if msg.Generation != m.playGeneration { logger.Debug("ignoring stale player launch error", "msgGen", msg.Generation, "currentGen", m.playGeneration) @@ -740,11 +747,9 @@ func (m Model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { return m, nil case tea.KeyMsg: + // A quit sends no command for the session: Cleanup sends the last + // position. See the quit with q below. if msg.Type == tea.KeyCtrlC { - if m.isPlaying() { - _, stopCmd := m.stopPlayback() - return m, tea.Batch(stopCmd, tea.Quit) - } return m, tea.Quit } if !m.isFiltering() && key.Matches(msg, m.keys.Help) { @@ -824,12 +829,11 @@ func (m Model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { m.toggleCaptions() return m, nil } - // Global quit: q always quits the app. + // Global quit: q always quits the app. The session stays in the model: + // main calls Cleanup on the final model, and Cleanup waits until the + // server has the last position. The program does not wait for a + // command, so the request of a command can be lost. if !m.isFiltering() && m.screen != ScreenLogin && key.Matches(msg, m.keys.Quit) { - if m.isPlaying() { - m, stopCmd := m.stopPlayback() - return m, tea.Batch(stopCmd, tea.Quit) - } return m, tea.Quit } // Global back: esc/left goes back but never quits. diff --git a/internal/app/model_test.go b/internal/app/model_test.go index 47af3cb..2c5abf3 100644 --- a/internal/app/model_test.go +++ b/internal/app/model_test.go @@ -857,12 +857,7 @@ func TestPositionMsgUpdatesPlayer(t *testing.T) { m.player.Playing = true m.player.Position = 10.0 - result, _ := m.Update(player.PositionMsg{ - Position: 20.0, - Duration: 3600.0, - Paused: false, - }) - rm := result.(Model) + rm := listen(playingNow(m), 20.0) if rm.player.Position != 20.0 { t.Errorf("player position = %f, want 20.0", rm.player.Position) @@ -896,8 +891,8 @@ func TestSyncTickWhenPlaying(t *testing.T) { result, cmd := m.Update(SyncTickMsg{}) rm := result.(Model) - if rm.lastSyncPos != 100.0 { - t.Errorf("lastSyncPos = %f, want 100.0", rm.lastSyncPos) + if rm.timeListened != 0 { + t.Errorf("timeListened = %f, want 0 (the sync takes it)", rm.timeListened) } if cmd == nil { t.Error("expected batch cmd from sync tick") @@ -1640,32 +1635,29 @@ func TestPositionMsgTracksTimeListened(t *testing.T) { m.player.Position = 100.0 m.timeListened = 0 - // Position advances by 5 seconds - result, _ := m.Update(player.PositionMsg{Position: 105.0, Duration: 3600.0, Paused: false}) - rm := result.(Model) + // The player plays for 5 seconds + rm := listen(playingNow(m), 105.0) if rm.timeListened != 5.0 { t.Errorf("timeListened = %f, want 5.0", rm.timeListened) } - // Position advances by 3 more seconds - result, _ = rm.Update(player.PositionMsg{Position: 108.0, Duration: 3600.0, Paused: false}) - rm = result.(Model) + // The player plays for 3 more seconds + rm = listen(rm, 108.0) if rm.timeListened != 8.0 { t.Errorf("timeListened = %f, want 8.0", rm.timeListened) } } -func TestPositionMsgBackwardSeekDoesNotAddTime(t *testing.T) { +func TestPositionMsgBackwardSeekAddsOnlyTimeOfPlay(t *testing.T) { m := newPlaybackTestModel() m.sessionID = "sess-123" m.player.Position = 100.0 m.timeListened = 5.0 - // Seek backward should not add time - result, _ := m.Update(player.PositionMsg{Position: 80.0, Duration: 3600.0, Paused: false}) - rm := result.(Model) - if rm.timeListened != 5.0 { - t.Errorf("timeListened = %f, want 5.0 (unchanged after backward)", rm.timeListened) + // In one second of play, the position moves backward by 20 seconds + rm := playTick(playingNow(m), 1, 80.0) + if rm.timeListened != 6.0 { + t.Errorf("timeListened = %f, want 6.0 (the second of play)", rm.timeListened) } } diff --git a/internal/app/playback.go b/internal/app/playback.go index 30a427a..c44ab8a 100644 --- a/internal/app/playback.go +++ b/internal/app/playback.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "math" "net/http" "runtime/debug" "strings" @@ -22,6 +23,18 @@ import ( const ( trackEndRolloverSlack = 2.0 propertyUnavailableMaxBackoffExp = 2 // 500ms << 2 = 2s + + // maxListeningTick is the longest time between two position ticks that + // pine counts as listening time. The ticks are 500 ms apart. After a longer + // time, for example a suspend of the machine, pine does not know how long + // the player played. + maxListeningTick = 5 * time.Second + + // syncMinChange is the smallest change of the position or of the + // listening time, in seconds, that pine sends to the server. A paused + // player can report a position that is a fraction of a second away from + // the position that the session started at. + syncMinChange = 1.0 ) func pineVersion() string { @@ -53,6 +66,38 @@ func (m Model) isPlaying() bool { return m.sessionID != "" } +// clock returns the current time. +func (m Model) clock() time.Time { + if m.now != nil { + return m.now() + } + return time.Now() +} + +// trackListeningTime adds the time since the last position tick to the +// listening time, if the player played at that tick and plays now. The +// listening time is the time of the clock. It is not the change of the +// position: at the speed 2.0, the position moves two seconds in each second. +func (m *Model) trackListeningTime(paused bool) { + if paused { + m.lastPlayingTick = time.Time{} + return + } + now := m.clock() + if !m.lastPlayingTick.IsZero() { + if played := now.Sub(m.lastPlayingTick); played > 0 && played <= maxListeningTick { + m.timeListened += played.Seconds() + } + } + m.lastPlayingTick = now +} + +// hasUnsyncedProgress returns true if the position or the listening time has +// a change that the server does not have. +func (m Model) hasUnsyncedProgress() bool { + return m.timeListened >= syncMinChange || math.Abs(m.player.Position-m.lastSyncPos) >= syncMinChange +} + func (m Model) isFiltering() bool { switch m.screen { case ScreenHome: @@ -101,10 +146,10 @@ func (m Model) handlePlayCmd(msg detail.PlayCmd) (Model, tea.Cmd) { } // Different item playing → stop it first - var stopCmd tea.Cmd + var closeCmd, stopCmd tea.Cmd if m.isPlaying() { logger.Info("switching playback", "from", m.itemID, "to", msg.Item.ID) - m, stopCmd = m.stopPlayback() + m, closeCmd, stopCmd = m.stopPlaybackCmds() } m.setSeriesContext(msg.Item) @@ -133,7 +178,7 @@ func (m Model) handlePlayCmd(msg detail.PlayCmd) (Model, tea.Cmd) { } if stopCmd != nil { - return m, tea.Batch(stopCmd, startCmd) + return m, tea.Batch(stopCmd, closeThen(closeCmd, startCmd)) } return m, startCmd } @@ -159,10 +204,10 @@ func (m Model) handlePlayEpisodeCmd(msg detail.PlayEpisodeCmd) (Model, tea.Cmd) } // Different item/episode playing → stop it first - var stopCmd tea.Cmd + var closeCmd, stopCmd tea.Cmd if m.isPlaying() { logger.Info("switching playback", "from", m.itemID+"/"+m.episodeID, "to", msg.Item.ID+"/"+msg.Episode.ID) - m, stopCmd = m.stopPlayback() + m, closeCmd, stopCmd = m.stopPlaybackCmds() } item := msg.Item @@ -187,6 +232,7 @@ func (m Model) handlePlayEpisodeCmd(msg detail.PlayEpisodeCmd) (Model, tea.Cmd) ItemID: item.ID, EpisodeID: episode.ID, CurrentTime: session.CurrentTime, + SyncedTime: session.CurrentTime, Duration: resolveEpisodeDuration(episode, session), Title: episode.Title, Authors: mprisAuthors(item), @@ -202,7 +248,7 @@ func (m Model) handlePlayEpisodeCmd(msg detail.PlayEpisodeCmd) (Model, tea.Cmd) } if stopCmd != nil { - return m, tea.Batch(stopCmd, startCmd) + return m, tea.Batch(stopCmd, closeThen(closeCmd, startCmd)) } return m, startCmd } @@ -219,10 +265,11 @@ func (m Model) handlePlaySessionMsg(msg PlaySessionMsg) (Model, tea.Cmd) { m.trackStartOffset = msg.Session.TrackStartOffset m.trackDuration = msg.Session.TrackDuration m.timeListened = 0 + m.lastPlayingTick = time.Time{} m.currentAuthors = msg.Session.Authors bookPos := msg.Session.CurrentTime + m.trackStartOffset - m.lastSyncPos = bookPos + m.lastSyncPos = msg.Session.SyncedTime // Update player model title/state m.player, _ = m.player.Update(player.StartPlayMsg{Title: msg.Session.Title}) @@ -320,6 +367,9 @@ func (m Model) handlePositionMsg(msg player.PositionMsg) (Model, tea.Cmd) { } if msg.Err != nil { + // pine does not know if the player plays, so the time until the next + // tick is not listening time. + m.lastPlayingTick = time.Time{} if targetPos, ok := m.trackEndRolloverTarget(msg.Err); ok { if m.player.Duration > 0 && targetPos >= m.player.Duration-trackEndRolloverSlack { m.player.Position = m.player.Duration @@ -355,6 +405,7 @@ func (m Model) handlePositionMsg(msg player.PositionMsg) (Model, tea.Cmd) { } m.propertyUnavailableCount = 0 + m.trackListeningTime(msg.Paused) // Convert track-relative position to book-global bookPos := msg.Position + m.trackStartOffset @@ -382,11 +433,6 @@ func (m Model) handlePositionMsg(msg player.PositionMsg) (Model, tea.Cmd) { return m, tea.Batch(mprisCmds...) } - // Track time listened (delta from last position) - if bookPos > m.player.Position { - m.timeListened += bookPos - m.player.Position - } - m.player.Position = bookPos wasPlaying := m.player.Playing m.player.Playing = !msg.Paused @@ -444,6 +490,7 @@ func (m Model) restartPlaybackAt(bookPos float64) (Model, tea.Cmd) { sessionID := m.sessionID currentTime := m.player.Position timeListened := m.timeListened + changed := m.hasUnsyncedProgress() duration := m.player.Duration title := m.player.Title authors := m.currentAuthors @@ -462,7 +509,7 @@ func (m Model) restartPlaybackAt(bookPos float64) (Model, tea.Cmd) { return m, func() tea.Msg { if client != nil && sessionID != "" { - _ = client.CloseSession(context.Background(), sessionID, currentTime, timeListened) + _ = closeSession(context.Background(), client, sessionID, currentTime, timeListened, changed) evictProgressCaches(cacheStore, itemID, libraryID, seriesID) } if mpvPlayer != nil { @@ -610,6 +657,7 @@ func (m Model) startRestoredEpisodePlaybackCmd(item abs.LibraryItem, episode abs ItemID: item.ID, EpisodeID: episode.ID, CurrentTime: session.CurrentTime, + SyncedTime: session.CurrentTime, Duration: resolveEpisodeDuration(episode, session), Title: episode.Title, Authors: mprisAuthors(item), @@ -633,15 +681,15 @@ func (m Model) handlePlaybackCompleted() (Model, tea.Cmd) { currentItemID := m.itemID client := m.client - m, stopCmd := m.stopPlayback() + m, closeCmd, stopCmd := m.stopPlaybackCmds() if hasNext { - return m.startQueuedEntry(next, stopCmd) + return m.startQueuedEntry(next, closeCmd, stopCmd) } if seriesID == "" || libraryID == "" || client == nil { - return m, stopCmd + return m, tea.Batch(closeCmd, stopCmd) } cmd := seriesContinueCmd(client, libraryID, seriesID, currentItemID) - return m, tea.Batch(stopCmd, cmd) + return m, tea.Batch(stopCmd, closeThen(closeCmd, cmd)) } func (m Model) skipToNextQueued() (Model, tea.Cmd) { @@ -652,11 +700,13 @@ func (m Model) skipToNextQueued() (Model, tea.Cmd) { if !hasNext { return m, nil } - m, stopCmd := m.stopPlayback() - return m.startQueuedEntry(next, stopCmd) + m, closeCmd, stopCmd := m.stopPlaybackCmds() + return m.startQueuedEntry(next, closeCmd, stopCmd) } -func (m Model) startQueuedEntry(next QueueEntry, stopCmd tea.Cmd) (Model, tea.Cmd) { +// startQueuedEntry starts the next entry of the queue after closeCmd, the +// close of the old session. +func (m Model) startQueuedEntry(next QueueEntry, closeCmd, stopCmd tea.Cmd) (Model, tea.Cmd) { var nextCmd tea.Cmd if next.Episode != nil { m, nextCmd = m.handlePlayEpisodeCmd(detail.PlayEpisodeCmd{ @@ -667,13 +717,7 @@ func (m Model) startQueuedEntry(next QueueEntry, stopCmd tea.Cmd) (Model, tea.Cm m, nextCmd = m.handlePlayCmd(detail.PlayCmd{Item: next.Item}) } - if stopCmd == nil { - return m, nextCmd - } - if nextCmd == nil { - return m, stopCmd - } - return m, tea.Batch(stopCmd, nextCmd) + return m, tea.Batch(stopCmd, closeThen(closeCmd, nextCmd)) } func buildBookPlaySessionMsg(client *cache.Client, session *abs.PlaySession, itemID, title string, authors []string, duration, bookPos float64) (PlaySessionMsg, error) { @@ -699,6 +743,7 @@ func buildBookPlaySessionMsg(client *cache.Client, session *abs.PlaySession, ite SessionID: session.ID, ItemID: itemID, CurrentTime: seekTime, + SyncedTime: session.CurrentTime, Duration: duration, Title: title, Authors: authors, @@ -749,20 +794,8 @@ func (m Model) handleSyncTick() (Model, tea.Cmd) { duration := m.player.Duration store := m.db - m.lastSyncPos = currentTime - cmds := []tea.Cmd{ syncTickCmd(), - func() tea.Msg { - if client != nil { - if err := client.SyncSession(context.Background(), sessionID, currentTime, timeListened); err != nil { - logger.Warn("failed to sync session", "sessionID", sessionID, "currentTime", currentTime, "timeListened", timeListened, "err", err) - } else { - logger.Debug("session synced", "sessionID", sessionID, "currentTime", currentTime, "timeListened", timeListened) - } - } - return nil - }, func() tea.Msg { if store != nil { if err := store.SaveListeningSession(db.ListeningSession{ @@ -781,9 +814,69 @@ func (m Model) handleSyncTick() (Model, tea.Cmd) { }, } + // A sync with an old position overwrites a newer position that a second + // device wrote. So no sync goes out while nothing changed. + if client != nil && m.hasUnsyncedProgress() { + // The server adds the timeListened of each sync to the session. So the + // sync takes the listening time, and handleSessionSynced gives it back + // if the sync fails. + m.timeListened = 0 + cmds = append(cmds, syncSessionCmd(client, m.playbackDevice(), sessionID, itemID, episodeID, currentTime, timeListened)) + } + return m, tea.Batch(cmds...) } +// syncSessionCmd sends the position and the listening time of a session to +// the server. The server answers 404 if it does not have the session: it +// loses its open sessions in a restart, and it closes a session after 36 +// hours with no change. The command then opens a new session for the item, +// and sends the position and the listening time to that session. +func syncSessionCmd(client *cache.Client, device abs.DeviceInfo, sessionID, itemID, episodeID string, currentTime, timeListened float64) tea.Cmd { + return func() tea.Msg { + ctx := context.Background() + msg := SessionSyncedMsg{SessionID: sessionID, CurrentTime: currentTime, TimeListened: timeListened} + msg.Err = client.SyncSession(ctx, sessionID, currentTime, timeListened) + if abs.IsHTTPStatus(msg.Err, http.StatusNotFound) { + logger.Info("session is gone on the server, opening a new session", "sessionID", sessionID, "itemID", itemID, "episodeID", episodeID) + var session *abs.PlaySession + if episodeID != "" { + session, msg.Err = client.StartEpisodePlaySession(ctx, itemID, episodeID, device) + } else { + session, msg.Err = client.StartPlaySession(ctx, itemID, device) + } + if msg.Err == nil { + msg.NewSessionID = session.ID + msg.Err = client.SyncSession(ctx, session.ID, currentTime, timeListened) + } + } + if msg.Err != nil { + logger.Warn("failed to sync session", "sessionID", sessionID, "newSessionID", msg.NewSessionID, "currentTime", currentTime, "timeListened", timeListened, "err", msg.Err) + } else { + logger.Debug("session synced", "sessionID", sessionID, "newSessionID", msg.NewSessionID, "currentTime", currentTime, "timeListened", timeListened) + } + return msg + } +} + +// handleSessionSynced records the result of a sync. An accepted sync gives the +// position that the server has. A failed sync gives its listening time back to +// the session, so that the next sync sends that time also. +func (m Model) handleSessionSynced(msg SessionSyncedMsg) Model { + if msg.SessionID != m.sessionID { + return m + } + if msg.NewSessionID != "" { + m.sessionID = msg.NewSessionID + } + if msg.Err != nil { + m.timeListened += msg.TimeListened + return m + } + m.lastSyncPos = msg.CurrentTime + return m +} + // evictProgressCaches drops cached data that embeds user progress for an item: // the progress record, the library item, the home shelves, and the series view. func evictProgressCaches(store *cache.Store, itemID, libraryID, seriesID string) { @@ -854,8 +947,44 @@ func (m Model) handleMarkFinished(msg detail.MarkFinishedCmd) (Model, tea.Cmd) { } } +// closeSession closes a session on the server. The server reads a position in +// a close as a progress update, and a position that is not at the end of a +// finished book removes the finished mark. So the close sends the position +// and the listening time only if changed is true: the server does not have +// them. +func closeSession(ctx context.Context, client *cache.Client, sessionID string, currentTime, timeListened float64, changed bool) error { + if !changed { + return client.CloseSessionNoSync(ctx, sessionID) + } + return client.CloseSession(ctx, sessionID, currentTime, timeListened) +} + +// closeThen returns a command that runs closeCmd, the close of a session, and +// then next. The server closes the open session of a device when that device +// starts a session, and it then answers 404 to the close of pine: the +// listening time that the close carries is lost. So a play request must not +// go out before the close completes or fails. +func closeThen(closeCmd, next tea.Cmd) tea.Cmd { + if next == nil { + return closeCmd + } + return func() tea.Msg { + closeCmd() + return next() + } +} + // stopPlayback fires all cleanup commands: update progress, close session, quit mpv. +// It sends no position if the server has the position and the listening time. func (m Model) stopPlayback() (Model, tea.Cmd) { + m, closeCmd, stopCmd := m.stopPlaybackCmds() + return m, tea.Batch(closeCmd, stopCmd) +} + +// stopPlaybackCmds stops the playback as stopPlayback does. It returns the +// close of the session apart from the other cleanup commands, so that a +// caller can run a play request after the close with closeThen. +func (m Model) stopPlaybackCmds() (Model, tea.Cmd, tea.Cmd) { logger.Info("stopping playback", "sessionID", m.sessionID, "itemID", m.itemID, "episodeID", m.episodeID, "position", m.player.Position, "duration", m.player.Duration, "timeListened", m.timeListened) client := m.client sessionID := m.sessionID @@ -863,6 +992,7 @@ func (m Model) stopPlayback() (Model, tea.Cmd) { episodeID := m.episodeID currentTime := m.player.Position timeListened := m.timeListened + changed := m.hasUnsyncedProgress() duration := m.player.Duration store := m.db mpvPlayer := m.mpv @@ -889,15 +1019,22 @@ func (m Model) stopPlayback() (Model, tea.Cmd) { evictProgressCaches(cacheStore, itemID, libraryID, seriesID) } + closeCmd := func() tea.Msg { + if client != nil && sessionID != "" { + if err := closeSession(context.Background(), client, sessionID, currentTime, timeListened, changed); err != nil { + logger.Warn("failed to close session", "sessionID", sessionID, "currentTime", currentTime, "timeListened", timeListened, "changed", changed, "err", err) + } else { + logger.Debug("session closed", "sessionID", sessionID, "currentTime", currentTime, "timeListened", timeListened, "changed", changed) + } + } + invalidateCache() + return nil + } + cmds := []tea.Cmd{ func() tea.Msg { - if client != nil && itemID != "" { - var err error - if episodeID != "" { - err = client.UpdateEpisodeProgress(context.Background(), itemID, episodeID, currentTime, progress, false) - } else { - err = client.UpdateProgress(context.Background(), itemID, currentTime, progress, false) - } + if changed && client != nil && itemID != "" { + err := client.UpdatePosition(context.Background(), itemID, episodeID, currentTime, progress) if err != nil { logger.Warn("failed to update progress on stop", "itemID", itemID, "episodeID", episodeID, "currentTime", currentTime, "progress", progress, "err", err) } else { @@ -907,17 +1044,6 @@ func (m Model) stopPlayback() (Model, tea.Cmd) { invalidateCache() return nil }, - func() tea.Msg { - if client != nil && sessionID != "" { - if err := client.CloseSession(context.Background(), sessionID, currentTime, timeListened); err != nil { - logger.Warn("failed to close session", "sessionID", sessionID, "currentTime", currentTime, "timeListened", timeListened, "err", err) - } else { - logger.Debug("session closed", "sessionID", sessionID, "currentTime", currentTime, "timeListened", timeListened) - } - } - invalidateCache() - return nil - }, func() tea.Msg { if store != nil && itemID != "" { if err := store.SaveListeningSession(db.ListeningSession{ @@ -939,7 +1065,7 @@ func (m Model) stopPlayback() (Model, tea.Cmd) { m.mprisTitleCmd(), } - return m, tea.Batch(cmds...) + return m, closeCmd, tea.Batch(cmds...) } func playSessionChapters(session *abs.PlaySession) []abs.Chapter { @@ -967,6 +1093,7 @@ func (m Model) Cleanup() { currentTime := m.player.Position timeListened := m.timeListened + changed := m.hasUnsyncedProgress() duration := m.player.Duration var progress float64 @@ -982,21 +1109,16 @@ func (m Model) Cleanup() { itemID := m.itemID if m.client != nil && m.sessionID != "" { - if err := m.client.CloseSession(ctx, m.sessionID, currentTime, timeListened); err != nil { - logger.Warn("cleanup: failed to close session", "sessionID", m.sessionID, "currentTime", currentTime, "timeListened", timeListened, "err", err) + if err := closeSession(ctx, m.client, m.sessionID, currentTime, timeListened, changed); err != nil { + logger.Warn("cleanup: failed to close session", "sessionID", m.sessionID, "currentTime", currentTime, "timeListened", timeListened, "changed", changed, "err", err) } else { - logger.Debug("cleanup: session closed", "sessionID", m.sessionID, "currentTime", currentTime, "timeListened", timeListened) + logger.Debug("cleanup: session closed", "sessionID", m.sessionID, "currentTime", currentTime, "timeListened", timeListened, "changed", changed) } evictProgressCaches(m.cacheStore, itemID, libraryID, seriesID) } - if m.client != nil && m.itemID != "" { - var err error - if m.episodeID != "" { - err = m.client.UpdateEpisodeProgress(ctx, m.itemID, m.episodeID, currentTime, progress, false) - } else { - err = m.client.UpdateProgress(ctx, m.itemID, currentTime, progress, false) - } + if changed && m.client != nil && m.itemID != "" { + err := m.client.UpdatePosition(ctx, m.itemID, m.episodeID, currentTime, progress) if err != nil { logger.Warn("cleanup: failed to update progress", "itemID", m.itemID, "episodeID", m.episodeID, "currentTime", currentTime, "progress", progress, "err", err) } else { diff --git a/internal/app/playback_test.go b/internal/app/playback_test.go index 9cff5d4..dbedd82 100644 --- a/internal/app/playback_test.go +++ b/internal/app/playback_test.go @@ -216,6 +216,8 @@ func TestPlaybackLifecycleIntegration(t *testing.T) { mp.position = 52.0 mp.duration = 3600.0 + // The first tick of the player, and then 10 seconds of play to 52.0 + m = advanceClock(listen(playTick(m, 0, 42.0), 51.0), 1) result, cmd = m.Update(player.PositionMsg{ Position: 52.0, Duration: 3600.0, @@ -227,7 +229,7 @@ func TestPlaybackLifecycleIntegration(t *testing.T) { if m.player.Position != 52.0 { t.Errorf("player position = %f, want 52.0", m.player.Position) } - // Delta from 42.0 to 52.0 = 10.0 seconds listened + // 10 seconds of play from 42.0 to 52.0 if m.timeListened != 10.0 { t.Errorf("timeListened = %f, want 10.0", m.timeListened) } @@ -235,14 +237,8 @@ func TestPlaybackLifecycleIntegration(t *testing.T) { t.Error("PositionMsg should reschedule TickCmd") } - // Another tick: advance to 57.0 - result, _ = m.Update(player.PositionMsg{ - Position: 57.0, - Duration: 3600.0, - Paused: false, - Generation: 1, - }) - m = result.(Model) + // 5 more seconds of play: advance to 57.0 + m = listen(m, 57.0) if m.player.Position != 57.0 { t.Errorf("player position = %f, want 57.0", m.player.Position) @@ -255,15 +251,16 @@ func TestPlaybackLifecycleIntegration(t *testing.T) { result, cmd = m.Update(SyncTickMsg{}) m = result.(Model) - if m.lastSyncPos != 57.0 { - t.Errorf("lastSyncPos = %f, want 57.0", m.lastSyncPos) - } if cmd == nil { t.Fatal("SyncTickMsg should return batch of sync cmds") } // Execute sync commands to trigger API calls - executeBatchCmds(cmd) + m, _ = feedCmd(m, cmd) + + if m.lastSyncPos != 57.0 { + t.Errorf("lastSyncPos = %f, want 57.0", m.lastSyncPos) + } // Verify SyncSession was called reqs = log.get() @@ -408,8 +405,8 @@ func TestPlayCmdPrefersLongerMediaMetadataChapters(t *testing.T) { } } -// TestPlaybackSeekBackwardThenForward tests that backward seeks don't inflate timeListened, -// and subsequent forward movement resumes tracking. +// TestPlaybackSeekBackwardThenForward tests that timeListened is the time of play +// before and after a backward seek, and not the change of the position. func TestPlaybackSeekBackwardThenForward(t *testing.T) { log := &apiLog{} srv := newMockABSServer(log) @@ -428,26 +425,24 @@ func TestPlaybackSeekBackwardThenForward(t *testing.T) { m.player.Position = 100.0 m.player.Duration = 3600.0 m.timeListened = 0 + m = playingNow(m) - // Forward: 100 → 110, adds 10s - result, _ := m.Update(player.PositionMsg{Position: 110.0, Duration: 3600.0}) - m = result.(Model) + // Forward: 100 → 110 in 10s of play, adds 10s + m = listen(m, 110.0) if m.timeListened != 10.0 { t.Errorf("timeListened = %f, want 10.0", m.timeListened) } - // Backward seek: 110 → 80, should NOT add time - result, _ = m.Update(player.PositionMsg{Position: 80.0, Duration: 3600.0}) - m = result.(Model) - if m.timeListened != 10.0 { - t.Errorf("timeListened = %f, want 10.0 (no change after backward seek)", m.timeListened) + // Backward seek: 110 → 80 in a tick of 500ms, adds the 500ms of play + m = playTick(m, 0.5, 80.0) + if m.timeListened != 10.5 { + t.Errorf("timeListened = %f, want 10.5 (the time of the tick)", m.timeListened) } - // Forward again: 80 → 90, adds 10s - result, _ = m.Update(player.PositionMsg{Position: 90.0, Duration: 3600.0}) - m = result.(Model) - if m.timeListened != 20.0 { - t.Errorf("timeListened = %f, want 20.0", m.timeListened) + // Forward again: 80 → 90 in 10s of play, adds 10s + m = listen(m, 90.0) + if m.timeListened != 20.5 { + t.Errorf("timeListened = %f, want 20.5", m.timeListened) } } @@ -1254,11 +1249,10 @@ func TestPlaybackMultipleSyncCycles(t *testing.T) { // Second sync tick result, cmd = m.Update(SyncTickMsg{}) - m = result.(Model) + m, _ = feedCmd(result.(Model), cmd) if m.lastSyncPos != 130.0 { t.Errorf("lastSyncPos = %f, want 130.0", m.lastSyncPos) } - executeBatchCmds(cmd) // Verify two sync calls were made reqs := log.get() diff --git a/internal/app/progress_sync_test.go b/internal/app/progress_sync_test.go new file mode 100644 index 0000000..af963ed --- /dev/null +++ b/internal/app/progress_sync_test.go @@ -0,0 +1,784 @@ +package app + +import ( + "encoding/json" + "fmt" + "math" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/Thelost77/pine/internal/abs" + "github.com/Thelost77/pine/internal/cache" + "github.com/Thelost77/pine/internal/config" + "github.com/Thelost77/pine/internal/player" + "github.com/Thelost77/pine/internal/screens/detail" + tea "github.com/charmbracelet/bubbletea" +) + +// newFakeABS starts a fake ABS server and returns its request log and a client +// for it. The server answers a POST to a path of sessions with that session. +// For each other request, it answers with the status that status returns. If +// status is nil or returns 0, it answers 200. +func newFakeABS(t *testing.T, sessions map[string]abs.PlaySession, status func(method, path string) int) (*apiLog, *cache.Client) { + t.Helper() + log := &apiLog{} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]interface{} + _ = json.NewDecoder(r.Body).Decode(&body) + log.record(r.Method, r.URL.Path, body) + + w.Header().Set("Content-Type", "application/json") + if session, ok := sessions[r.URL.Path]; ok && r.Method == http.MethodPost { + _ = json.NewEncoder(w).Encode(session) + return + } + if status != nil { + if code := status(r.Method, r.URL.Path); code != 0 { + w.WriteHeader(code) + return + } + } + _, _ = w.Write([]byte("{}")) + })) + t.Cleanup(srv.Close) + return log, cache.NewClient(abs.NewClient(srv.URL, "tok"), nil) +} + +// requestsTo returns the requests of the log with the method and the path. +func requestsTo(log *apiLog, method, path string) []apiRequest { + var found []apiRequest + for _, r := range log.get() { + if r.Method == method && r.Path == path { + found = append(found, r) + } + } + return found +} + +// timesListened returns the timeListened of each POST to the path. +func timesListened(log *apiLog, path string) []float64 { + var times []float64 + for _, r := range requestsTo(log, http.MethodPost, path) { + times = append(times, r.Body["timeListened"].(float64)) + } + return times +} + +// syncTick runs one sync tick and its commands. +func syncTick(m Model) Model { + result, cmd := m.Update(SyncTickMsg{}) + m, _ = feedCmd(result.(Model), cmd) + return m +} + +// advanceClock moves the clock of the model by seconds. After the first call, +// the clock of the model moves only in a call of advanceClock. +func advanceClock(m Model, seconds float64) Model { + now := m.clock().Add(time.Duration(seconds * float64(time.Second))) + m.now = func() time.Time { return now } + return m +} + +// playTick moves the clock of the model by seconds, and then sends a position +// tick of a player that plays at the position pos. +func playTick(m Model, seconds, pos float64) Model { + m = advanceClock(m, seconds) + result, _ := m.Update(player.PositionMsg{ + Position: pos - m.trackStartOffset, + Duration: m.player.Duration, + Generation: m.playGeneration, + }) + return result.(Model) +} + +// playingNow stops the clock of the model, and records that the player played +// at the last position tick. The time until the next position tick of a +// player that plays is then listening time. +func playingNow(m Model) Model { + m = advanceClock(m, 0) + m.lastPlayingTick = m.clock() + return m +} + +// listen plays to the position pos at the speed 1.0, with one position tick +// for each second. +func listen(m Model, pos float64) Model { + for m.player.Position < pos { + step := math.Min(1, pos-m.player.Position) + m = playTick(m, step, m.player.Position+step) + } + return m +} + +// listenAndSync plays to the position pos at the speed 1.0, and then runs one +// sync tick and its commands. +func listenAndSync(m Model, pos float64) Model { + return syncTick(listen(m, pos)) +} + +// newSyncTestModel returns a model that plays item-001 in the session sess-abc +// at the position 100. The server has that position. +func newSyncTestModel(client *cache.Client) Model { + m := NewWithPlayer(config.Default(), nil, client, nil, &mockPlayer{}) + m.sessionID = "sess-abc" + m.itemID = "item-001" + m.player.Playing = true + m.player.Position = 100 + m.player.Duration = 3600 + m.lastSyncPos = 100 + return playingNow(m) +} + +// TestSyncSendsListeningTimeSinceLastSync verifies that a sync sends the time +// listened since the last sync. The server adds the timeListened of each sync +// to the session, so the total time of the session must not be in a sync. +func TestSyncSendsListeningTimeSinceLastSync(t *testing.T) { + log, client := newFakeABS(t, nil, nil) + m := newSyncTestModel(client) + + m = listenAndSync(m, 110) + m = listenAndSync(m, 115) + + want := []float64{10, 5} + if got := timesListened(log, "/api/session/sess-abc/sync"); !reflect.DeepEqual(got, want) { + t.Errorf("timeListened of the syncs = %v, want %v", got, want) + } + if m.timeListened != 0 { + t.Errorf("timeListened = %v, want 0 after an accepted sync", m.timeListened) + } +} + +// TestSyncCarriesListeningTimeOfFailedSync verifies that the listening time of +// a sync that the server did not accept goes out again with the next sync. +func TestSyncCarriesListeningTimeOfFailedSync(t *testing.T) { + var syncs atomic.Int32 + log, client := newFakeABS(t, nil, func(method, path string) int { + if path == "/api/session/sess-abc/sync" && syncs.Add(1) == 1 { + return http.StatusInternalServerError + } + return 0 + }) + m := newSyncTestModel(client) + + m = listenAndSync(m, 110) + if m.timeListened != 10 { + t.Fatalf("timeListened = %v, want 10 after a failed sync", m.timeListened) + } + m = listenAndSync(m, 115) + + want := []float64{10, 15} + if got := timesListened(log, "/api/session/sess-abc/sync"); !reflect.DeepEqual(got, want) { + t.Errorf("timeListened of the syncs = %v, want %v", got, want) + } + if m.timeListened != 0 { + t.Errorf("timeListened = %v, want 0 after an accepted sync", m.timeListened) + } +} + +// TestCloseSendsListeningTimeSinceLastSync verifies that the close of a +// session sends only the time listened since the last sync. +func TestCloseSendsListeningTimeSinceLastSync(t *testing.T) { + log, client := newFakeABS(t, nil, nil) + m := newSyncTestModel(client) + + m = listenAndSync(m, 110) + _, cmd := listen(m, 115).stopPlayback() + executeBatchCmds(cmd) + + want := []float64{5} + if got := timesListened(log, "/api/session/sess-abc/close"); !reflect.DeepEqual(got, want) { + t.Errorf("timeListened of the close = %v, want %v", got, want) + } +} + +// TestSyncTickSendsNothingWhenNothingChanged verifies that a session sends no +// sync while it has no change since the last sync that the server accepted. A +// sync with an old position overwrites a newer position of a second device. +func TestSyncTickSendsNothingWhenNothingChanged(t *testing.T) { + const syncPath = "/api/session/sess-abc/sync" + log, client := newFakeABS(t, nil, nil) + m := newSyncTestModel(client) + + // The player is paused at the position that the server has. + result, _ := m.Update(player.PositionMsg{Position: 100, Duration: 3600, Paused: true}) + result, cmd := result.(Model).Update(SyncTickMsg{}) + if cmd == nil { + t.Fatal("the sync tick must continue while a session exists") + } + m, _ = feedCmd(result.(Model), cmd) + if n := len(requestsTo(log, http.MethodPost, syncPath)); n != 0 { + t.Fatalf("%d syncs with no change, want 0", n) + } + + // The user moves the position. One sync sends the new position. + m, _ = m.handleSeek(200) + m = syncTick(m) + m = syncTick(m) + syncs := requestsTo(log, http.MethodPost, syncPath) + if len(syncs) != 1 { + t.Fatalf("%d syncs after one seek, want 1", len(syncs)) + } + if got := syncs[0].Body["currentTime"]; got != 300.0 { + t.Errorf("currentTime of the sync = %v, want 300", got) + } + if m.lastSyncPos != 300 { + t.Errorf("lastSyncPos = %v, want 300", m.lastSyncPos) + } +} + +// TestSyncSendsPositionOfSeekToOtherTrack verifies that a session that starts +// at a position that the server does not have sends that position. +func TestSyncSendsPositionOfSeekToOtherTrack(t *testing.T) { + log, client := newFakeABS(t, map[string]abs.PlaySession{ + "/api/items/item-001/play": { + ID: "sess-next", + AudioTracks: []abs.AudioTrack{ + {Index: 0, StartOffset: 0, ContentURL: "/track0.mp3", Duration: 1800}, + {Index: 1, StartOffset: 1800, ContentURL: "/track1.mp3", Duration: 1800}, + }, + CurrentTime: 100, + }, + }, nil) + m := newSyncTestModel(client) + m.trackDuration = 1800 + + m, cmd := m.handleSeek(1800) + m, _ = feedCmd(m, cmd) + if m.sessionID != "sess-next" { + t.Fatalf("sessionID = %q, want sess-next", m.sessionID) + } + m = syncTick(m) + + syncs := requestsTo(log, http.MethodPost, "/api/session/sess-next/sync") + if len(syncs) != 1 { + t.Fatalf("%d syncs after a seek to a different track, want 1", len(syncs)) + } + if got := syncs[0].Body["currentTime"]; got != 1900.0 { + t.Errorf("currentTime of the sync = %v, want 1900", got) + } + if m.lastSyncPos != 1900 { + t.Errorf("lastSyncPos = %v, want 1900", m.lastSyncPos) + } +} + +// sessionGone answers 404 to each sync of the session sess-abc, as a server +// does after a restart, or after 36 hours with no change of the session. +func sessionGone(_, path string) int { + if path == "/api/session/sess-abc/sync" { + return http.StatusNotFound + } + return 0 +} + +// TestSyncOpensNewSessionAfter404 verifies that pine opens a new session for +// the item if the server does not have the session of a sync, and that the +// new session gets the position and the listening time of that sync. +func TestSyncOpensNewSessionAfter404(t *testing.T) { + tests := []struct { + name string + episodeID string + playPath string + }{ + {name: "book", playPath: "/api/items/item-001/play"}, + {name: "episode", episodeID: "ep-001", playPath: "/api/items/item-001/play/ep-001"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + log, client := newFakeABS(t, map[string]abs.PlaySession{ + tt.playPath: {ID: "sess-new", AudioTracks: []abs.AudioTrack{{ContentURL: "/audio.mp3", Duration: 3600}}}, + }, sessionGone) + m := newSyncTestModel(client) + m.episodeID = tt.episodeID + + m = listenAndSync(m, 130) + + want := []string{ + "POST /api/session/sess-abc/sync", + "POST " + tt.playPath, + "POST /api/session/sess-new/sync", + } + if got := summarizeRequests(log.get()); !reflect.DeepEqual(got, want) { + t.Fatalf("requests = %v, want %v", got, want) + } + body := log.get()[2].Body + if body["currentTime"] != 130.0 || body["timeListened"] != 30.0 { + t.Errorf("sync of the new session = %v, want currentTime 130 and timeListened 30", body) + } + if m.sessionID != "sess-new" { + t.Errorf("sessionID = %q, want sess-new", m.sessionID) + } + if m.timeListened != 0 || m.lastSyncPos != 130 { + t.Errorf("timeListened = %v, lastSyncPos = %v, want 0 and 130", m.timeListened, m.lastSyncPos) + } + + listenAndSync(m, 140) + if got := timesListened(log, "/api/session/sess-new/sync"); !reflect.DeepEqual(got, []float64{30, 10}) { + t.Errorf("timeListened of the syncs of the new session = %v, want [30 10]", got) + } + }) + } +} + +// TestSyncKeepsListeningTimeIfNewSessionFails verifies that the listening time +// stays with the session if pine cannot open a new session after a 404. +func TestSyncKeepsListeningTimeIfNewSessionFails(t *testing.T) { + _, client := newFakeABS(t, nil, func(method, path string) int { + if path == "/api/items/item-001/play" { + return http.StatusInternalServerError + } + return sessionGone(method, path) + }) + m := newSyncTestModel(client) + + m = listenAndSync(m, 130) + + if m.sessionID != "sess-abc" { + t.Errorf("sessionID = %q, want sess-abc", m.sessionID) + } + if m.timeListened != 30 || m.lastSyncPos != 100 { + t.Errorf("timeListened = %v, lastSyncPos = %v, want 30 and 100", m.timeListened, m.lastSyncPos) + } +} + +// TestStopSendsNoFinishedFlag verifies that the progress update of a stop has +// no isFinished key. ABS reads "isFinished": false as "mark as not finished". +// At the end of a book, that update removed the mark that the close of the +// session set, and it set the position to 0. +func TestStopSendsNoFinishedFlag(t *testing.T) { + stops := map[string]func(m Model){ + "end of the book": func(m Model) { + _, cmd := m.Update(player.PositionMsg{Err: fmt.Errorf("get time-pos: %w", player.ErrClosed)}) + executeBatchCmds(cmd) + }, + "cleanup": func(m Model) { m.Cleanup() }, + } + for name, stop := range stops { + t.Run(name, func(t *testing.T) { + log, client := newFakeABS(t, nil, nil) + m := newSyncTestModel(client) + m.player.Position = 3600 + m.trackDuration = 3600 + + stop(m) + + updates := requestsTo(log, http.MethodPatch, "/api/me/progress/item-001") + if len(updates) != 1 { + t.Fatalf("%d progress updates, want 1", len(updates)) + } + want := map[string]interface{}{"currentTime": 3600.0, "progress": 1.0} + if !reflect.DeepEqual(updates[0].Body, want) { + t.Errorf("progress update = %v, want %v", updates[0].Body, want) + } + }) + } +} + +// TestMarkFinishedSendsExplicitFlag verifies that the actions "mark as +// finished" and "mark as not finished" send the isFinished key. +func TestMarkFinishedSendsExplicitFlag(t *testing.T) { + tests := []struct { + name string + undo bool + want map[string]interface{} + }{ + {name: "mark as finished", want: map[string]interface{}{"currentTime": 3600.0, "progress": 1.0, "isFinished": true}}, + {name: "mark as not finished", undo: true, want: map[string]interface{}{"currentTime": 0.0, "progress": 0.0, "isFinished": false}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + log, client := newFakeABS(t, nil, nil) + m := newSyncTestModel(client) + + _, cmd := m.handleMarkFinished(detail.MarkFinishedCmd{Item: testLibraryItem("item-002", "1984"), Undo: tt.undo}) + cmd() + + updates := requestsTo(log, http.MethodPatch, "/api/me/progress/item-002") + if len(updates) != 1 { + t.Fatalf("%d progress updates, want 1", len(updates)) + } + if !reflect.DeepEqual(updates[0].Body, tt.want) { + t.Errorf("progress update = %v, want %v", updates[0].Body, tt.want) + } + }) + } +} + +// TestStopOfUntouchedSessionSendsNoProgress verifies that the stop of a +// session sends no position if the user listened to nothing and did not move +// the position. pine restores the last session at each start. For a finished +// book, the server starts that session at the position 0, and it reads each +// position that is not at the end as "not finished". +func TestStopOfUntouchedSessionSendsNoProgress(t *testing.T) { + stops := map[string]func(t *testing.T, m Model){ + "quit with q": func(t *testing.T, m Model) { quitProgram(t, m, quitKeys["q"])() }, + "quit with ctrl+c": func(t *testing.T, m Model) { quitProgram(t, m, quitKeys["ctrl+c"])() }, + "cleanup": func(_ *testing.T, m Model) { m.Cleanup() }, + "play of a different item": func(_ *testing.T, m Model) { + _, cmd := m.handlePlayCmd(detail.PlayCmd{Item: testLibraryItem("item-002", "1984")}) + executeBatchCmds(cmd) + }, + } + for name, stop := range stops { + t.Run(name, func(t *testing.T) { + log, client := newFakeABS(t, map[string]abs.PlaySession{ + "/api/items/item-001/play": {ID: "sess-abc", AudioTracks: []abs.AudioTrack{{ContentURL: "/audio.mp3", Duration: 3600}}}, + }, nil) + m := NewWithPlayer(config.Default(), nil, client, nil, &mockPlayer{paused: true}) + + item := testLibraryItem("item-001", "The Great Gatsby") + m, _ = feedCmd(m, m.startRestorePlaybackCmd(RestoreSessionMsg{Item: &item})) + if m.sessionID != "sess-abc" { + t.Fatalf("sessionID = %q, want sess-abc", m.sessionID) + } + result, _ := m.Update(player.PositionMsg{Duration: 3600, Paused: true, Generation: m.playGeneration}) + + stop(t, result.(Model)) + + if n := len(requestsTo(log, http.MethodPatch, "/api/me/progress/item-001")); n != 0 { + t.Errorf("%d progress updates, want 0", n) + } + closes := requestsTo(log, http.MethodPost, "/api/session/sess-abc/close") + if len(closes) != 1 { + t.Fatalf("%d closes of the session, want 1", len(closes)) + } + if len(closes[0].Body) != 0 { + t.Errorf("close of the session = %v, want no position", closes[0].Body) + } + }) + } +} + +// runUntilPlaySession runs cmd, and the commands that Update returns for +// their messages, as Bubble Tea does: each command of a batch runs in its own +// goroutine, and Update gets each message when it arrives. It returns at the +// first PlaySessionMsg. +func runUntilPlaySession(t *testing.T, m Model, cmd tea.Cmd) { + t.Helper() + msgs := make(chan tea.Msg) + done := make(chan struct{}) + defer close(done) + + var run func(cmd tea.Cmd) + run = func(cmd tea.Cmd) { + if cmd == nil { + return + } + go func() { + msg := cmd() + if batch, ok := msg.(tea.BatchMsg); ok { + for _, c := range batch { + run(c) + } + return + } + select { + case msgs <- msg: + case <-done: + } + }() + } + run(cmd) + + timeout := time.After(5 * time.Second) + for { + select { + case msg := <-msgs: + if _, ok := msg.(PlaySessionMsg); ok { + return + } + result, next := m.Update(msg) + m = result.(Model) + run(next) + case <-timeout: + t.Fatal("no PlaySessionMsg in 5 seconds") + } + } +} + +// TestOldSessionClosesBeforeNextPlay verifies that the play request of the +// next item goes out after the close of the old session completes or fails. +// A server closes the open session of a device when that device starts a +// session, and it then answers 404 to the close: the listening time that the +// close carries is lost. +func TestOldSessionClosesBeforeNextPlay(t *testing.T) { + current := testLibraryItem("item-001", "The Great Gatsby") + current.LibraryID = "lib-001" + current.Media.Metadata.Series = &abs.SeriesSequence{ID: "series-001", Name: "A Series", Sequence: "1"} + next := testLibraryItem("item-002", "1984") + next.LibraryID = "lib-001" + next.Media.Metadata.Series = &abs.SeriesSequence{ID: "series-001", Name: "A Series", Sequence: "2"} + podcast := testPodcastItem("pod-001", "Tech Talk Daily") + + endOfBook := func(m Model) (Model, tea.Cmd) { + m.player.Position = 3600 + m.trackDuration = 3600 + result, cmd := m.Update(player.PositionMsg{Err: fmt.Errorf("get time-pos: %w", player.ErrClosed)}) + return result.(Model), cmd + } + tests := []struct { + name string + closeStatus int + start func(m Model) (Model, tea.Cmd) + }{ + {name: "play of a different book", start: func(m Model) (Model, tea.Cmd) { + return m.handlePlayCmd(detail.PlayCmd{Item: next}) + }}, + {name: "play of an episode", start: func(m Model) (Model, tea.Cmd) { + return m.handlePlayEpisodeCmd(detail.PlayEpisodeCmd{Item: podcast, Episode: podcast.Media.Episodes[0]}) + }}, + {name: "next item of the queue at the end of a book", start: func(m Model) (Model, tea.Cmd) { + m.queue = []QueueEntry{{Item: next}} + return endOfBook(m) + }}, + {name: "skip to the next item of the queue", start: func(m Model) (Model, tea.Cmd) { + m.queue = []QueueEntry{{Item: next}} + return m.skipToNextQueued() + }}, + {name: "next book of the series at the end of a book", start: func(m Model) (Model, tea.Cmd) { + m.playbackSeriesID = "series-001" + m.playbackLibraryID = "lib-001" + return endOfBook(m) + }}, + {name: "close fails", closeStatus: http.StatusInternalServerError, start: func(m Model) (Model, tea.Cmd) { + return m.handlePlayCmd(detail.PlayCmd{Item: next}) + }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var mu sync.Mutex + closed, played, playedBeforeClose := false, false, false + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch { + case r.URL.Path == "/api/session/sess-abc/close": + time.Sleep(100 * time.Millisecond) + mu.Lock() + closed = true + mu.Unlock() + if tt.closeStatus != 0 { + w.WriteHeader(tt.closeStatus) + } + case r.Method == http.MethodPost && strings.Contains(r.URL.Path, "/play"): + mu.Lock() + played = true + playedBeforeClose = !closed + mu.Unlock() + _ = json.NewEncoder(w).Encode(abs.PlaySession{ + ID: "sess-next", + AudioTracks: []abs.AudioTrack{{ContentURL: "/audio.mp3", Duration: 3600}}, + }) + return + case r.URL.Path == "/api/libraries/lib-001/series/series-001": + _ = json.NewEncoder(w).Encode(abs.Series{ID: "series-001", Name: "A Series"}) + return + case strings.HasPrefix(r.URL.Path, "/api/libraries/lib-001/items"): + _ = json.NewEncoder(w).Encode(abs.LibraryItemsResponse{Results: []abs.LibraryItem{current, next}, Total: 2}) + return + } + _, _ = w.Write([]byte("{}")) + })) + defer srv.Close() + + m := newSyncTestModel(cache.NewClient(abs.NewClient(srv.URL, "tok"), nil)) + m, cmd := tt.start(m) + runUntilPlaySession(t, m, cmd) + + mu.Lock() + defer mu.Unlock() + if !played { + t.Fatal("no play request for the next item") + } + if playedBeforeClose { + t.Error("the play request of the next item went out before the close of the old session completed") + } + }) + } +} + +// quitProgram sends the key of a quit to the model, and then does what the +// program does at a quit. Bubble Tea runs each command of a batch in its own +// goroutine, and it returns at the QuitMsg: it does not wait for the other +// commands. main then calls Cleanup on the final model, and the process ends. +// quitProgram returns at that moment. The function that it returns waits for +// the commands that still run. +func quitProgram(t *testing.T, m Model, key tea.KeyMsg) (wait func()) { + t.Helper() + result, cmd := m.Update(key) + + var running sync.WaitGroup + var once sync.Once + quit := make(chan struct{}) + var run func(cmd tea.Cmd) + run = func(cmd tea.Cmd) { + if cmd == nil { + return + } + running.Add(1) + go func() { + defer running.Done() + switch msg := cmd().(type) { + case tea.BatchMsg: + for _, c := range msg { + run(c) + } + case tea.QuitMsg: + once.Do(func() { close(quit) }) + } + }() + } + run(cmd) + + select { + case <-quit: + case <-time.After(5 * time.Second): + t.Fatal("no QuitMsg in 5 seconds") + } + result.(Model).Cleanup() + return running.Wait +} + +// quitKeys are the keys that quit pine. +var quitKeys = map[string]tea.KeyMsg{ + "q": {Type: tea.KeyRunes, Runes: []rune{'q'}}, + "ctrl+c": {Type: tea.KeyCtrlC}, +} + +// TestQuitWaitsForLastPosition verifies that the server has the last position +// and the listening time of a session when the process of a quit ends. A +// request that a command of the quit sends is lost if the process ends before +// the server gets it. +func TestQuitWaitsForLastPosition(t *testing.T) { + for name, key := range quitKeys { + t.Run(name, func(t *testing.T) { + // The server needs 100 ms for each request. + answered := &apiLog{} + log, client := newFakeABS(t, nil, func(method, path string) int { + time.Sleep(100 * time.Millisecond) + answered.record(method, path, nil) + return 0 + }) + m := newSyncTestModel(client) + m.player.Position = 130 + m.timeListened = 30 + + wait := quitProgram(t, m, key) + atEnd := summarizeRequests(answered.get()) + wait() + + for _, want := range []string{"POST /api/session/sess-abc/close", "PATCH /api/me/progress/item-001"} { + found := false + for _, got := range atEnd { + found = found || got == want + } + if !found { + t.Errorf("requests that the server answered before the end of the process = %v, want %s", atEnd, want) + } + } + closes := requestsTo(log, http.MethodPost, "/api/session/sess-abc/close") + if len(closes) == 0 { + t.Fatal("no close of the session") + } + want := map[string]interface{}{"currentTime": 130.0, "timeListened": 30.0} + if !reflect.DeepEqual(closes[0].Body, want) { + t.Errorf("close of the session = %v, want %v", closes[0].Body, want) + } + }) + } +} + +// TestQuitSendsLastPositionOnce verifies that a quit sends the close of the +// session and the progress update one time. The server adds the timeListened +// of each close to the session. +func TestQuitSendsLastPositionOnce(t *testing.T) { + for name, key := range quitKeys { + t.Run(name, func(t *testing.T) { + log, client := newFakeABS(t, nil, nil) + m := newSyncTestModel(client) + m.player.Position = 130 + m.timeListened = 30 + + quitProgram(t, m, key)() + + if got := timesListened(log, "/api/session/sess-abc/close"); !reflect.DeepEqual(got, []float64{30}) { + t.Errorf("timeListened of the closes = %v, want [30]", got) + } + if n := len(requestsTo(log, http.MethodPatch, "/api/me/progress/item-001")); n != 1 { + t.Errorf("%d progress updates, want 1", n) + } + }) + } +} + +// TestListeningTimeIsTimeOfPlay verifies that the listening time is the time +// during which the player played, and not the change of the position. At the +// speed 2.0, the position moves two seconds in each second. The server adds +// the timeListened of each sync to the listening time of the user. +func TestListeningTimeIsTimeOfPlay(t *testing.T) { + // pausedTick moves the clock by seconds, and then sends a position tick + // of a paused player. + pausedTick := func(m Model, seconds float64) Model { + m = advanceClock(m, seconds) + result, _ := m.Update(player.PositionMsg{Position: m.player.Position, Duration: m.player.Duration, Paused: true}) + return result.(Model) + } + tests := []struct { + name string + play func(m Model) Model + want map[string]interface{} + }{ + { + name: "one minute at the speed 2.0", + play: func(m Model) Model { + for i := 0; i < 120; i++ { + m = playTick(m, 0.5, m.player.Position+1) + } + return m + }, + want: map[string]interface{}{"currentTime": 220.0, "timeListened": 60.0}, + }, + { + name: "a pause of one minute", + play: func(m Model) Model { + m = listen(m, 110) + for i := 0; i < 120; i++ { + m = pausedTick(m, 0.5) + } + // The tick after the pause starts the count again. + return listen(playTick(m, 0.5, 110), 120) + }, + want: map[string]interface{}{"currentTime": 120.0, "timeListened": 20.0}, + }, + { + name: "a tick after a suspend of the machine", + play: func(m Model) Model { + m = listen(m, 110) + m = playTick(m, 8*3600, 110.5) + return listen(m, 120.5) + }, + want: map[string]interface{}{"currentTime": 120.5, "timeListened": 20.0}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + log, client := newFakeABS(t, nil, nil) + m := syncTick(tt.play(newSyncTestModel(client))) + + syncs := requestsTo(log, http.MethodPost, "/api/session/sess-abc/sync") + if len(syncs) != 1 { + t.Fatalf("%d syncs, want 1", len(syncs)) + } + if !reflect.DeepEqual(syncs[0].Body, tt.want) { + t.Errorf("sync = %v, want %v", syncs[0].Body, tt.want) + } + if m.timeListened != 0 { + t.Errorf("timeListened = %v, want 0 after an accepted sync", m.timeListened) + } + }) + } +}