feat(server): emit playlist.system_rebuilt on daily + manual system-playlist rebuild
The daily 03:00 scheduler rebuild (and the manual refresh endpoint) replace a user's system playlists + You-might-like rows but published no event, so a client left open across the rebuild served yesterday's snapshot until a manual reload — the stale-tab case behind #968. Add a user-scoped playlist.system_rebuilt event (envelope {kind,user_id,data:{}}) from both the scheduler (bus threaded into NewScheduler) and handleSystemPlaylistRefresh. Clients consume it to invalidate home / system-playlist views and proactively re-pull a stale active queue. Issue #968. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -242,7 +242,12 @@ func run() error {
|
|||||||
// active user's daily build at 03:00 in their stored timezone.
|
// active user's daily build at 03:00 in their stored timezone.
|
||||||
// Replaces the 24h-anchored cron loop (removed in the next commit
|
// Replaces the 24h-anchored cron loop (removed in the next commit
|
||||||
// of this arc).
|
// of this arc).
|
||||||
playlistScheduler, err := playlists.NewScheduler(pool, logger.With("component", "playlist_scheduler"), cfg.Storage.DataDir)
|
playlistScheduler, err := playlists.NewScheduler(
|
||||||
|
pool,
|
||||||
|
logger.With("component", "playlist_scheduler"),
|
||||||
|
cfg.Storage.DataDir,
|
||||||
|
bus,
|
||||||
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("init playlist scheduler: %w", err)
|
return fmt.Errorf("init playlist scheduler: %w", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -88,6 +88,21 @@ func (h *handlers) publishPlaylistEvent(kind string, ownerID, playlistID pgtype.
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// publishSystemRebuilt notifies the owner's clients that their system
|
||||||
|
// playlists (and You-might-like rows) were regenerated, so they invalidate
|
||||||
|
// the home / system-playlist providers and a stale active queue can re-pull.
|
||||||
|
// Mirrors the daily scheduler's event; fired here from the manual refresh.
|
||||||
|
func (h *handlers) publishSystemRebuilt(userID pgtype.UUID) {
|
||||||
|
if h.eventbus == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
h.eventbus.Publish(eventbus.Event{
|
||||||
|
Kind: "playlist.system_rebuilt",
|
||||||
|
UserID: uuidToString(userID),
|
||||||
|
Data: map[string]any{},
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
// publishRequestStatusChanged broadcasts a Lidarr request status flip to
|
// publishRequestStatusChanged broadcasts a Lidarr request status flip to
|
||||||
// the request's original requester so their /requests page reflects the
|
// the request's original requester so their /requests page reflects the
|
||||||
// new state without manual refresh. Admin actors (approve / reject) still
|
// new state without manual refresh. Admin actors (approve / reject) still
|
||||||
|
|||||||
@@ -73,6 +73,8 @@ func (h *handlers) handleSystemPlaylistRefresh(w http.ResponseWriter, r *http.Re
|
|||||||
writeErr(w, apierror.InternalMsg("build failed", err))
|
writeErr(w, apierror.InternalMsg("build failed", err))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
// #968: announce the rebuild so the user's other clients refresh.
|
||||||
|
h.publishSystemRebuilt(user.ID)
|
||||||
q := dbq.New(h.pool)
|
q := dbq.New(h.pool)
|
||||||
v := kind
|
v := kind
|
||||||
pl, err := q.GetSystemPlaylistByVariantForUser(r.Context(),
|
pl, err := q.GetSystemPlaylistByVariantForUser(r.Context(),
|
||||||
|
|||||||
@@ -28,6 +28,7 @@ import (
|
|||||||
"github.com/jackc/pgx/v5/pgxpool"
|
"github.com/jackc/pgx/v5/pgxpool"
|
||||||
|
|
||||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||||
|
"git.fabledsword.com/bvandeusen/minstrel/internal/eventbus"
|
||||||
"git.fabledsword.com/bvandeusen/minstrel/internal/taste"
|
"git.fabledsword.com/bvandeusen/minstrel/internal/taste"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -47,14 +48,21 @@ type Scheduler struct {
|
|||||||
pool *pgxpool.Pool
|
pool *pgxpool.Pool
|
||||||
logger *slog.Logger
|
logger *slog.Logger
|
||||||
dataDir string
|
dataDir string
|
||||||
|
bus *eventbus.Bus // #968: announce rebuilds so clients self-refresh
|
||||||
|
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
jobs map[pgtype.UUID]uuid.UUID // user_id → gocron job id
|
jobs map[pgtype.UUID]uuid.UUID // user_id → gocron job id
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewScheduler builds an idle scheduler. Caller must invoke Start
|
// NewScheduler builds an idle scheduler. Caller must invoke Start
|
||||||
// before any builds fire.
|
// before any builds fire. bus may be nil (tests); rebuild events are
|
||||||
func NewScheduler(pool *pgxpool.Pool, logger *slog.Logger, dataDir string) (*Scheduler, error) {
|
// then skipped.
|
||||||
|
func NewScheduler(
|
||||||
|
pool *pgxpool.Pool,
|
||||||
|
logger *slog.Logger,
|
||||||
|
dataDir string,
|
||||||
|
bus *eventbus.Bus,
|
||||||
|
) (*Scheduler, error) {
|
||||||
g, err := gocron.NewScheduler()
|
g, err := gocron.NewScheduler()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("init gocron: %w", err)
|
return nil, fmt.Errorf("init gocron: %w", err)
|
||||||
@@ -64,6 +72,7 @@ func NewScheduler(pool *pgxpool.Pool, logger *slog.Logger, dataDir string) (*Sch
|
|||||||
pool: pool,
|
pool: pool,
|
||||||
logger: logger,
|
logger: logger,
|
||||||
dataDir: dataDir,
|
dataDir: dataDir,
|
||||||
|
bus: bus,
|
||||||
jobs: map[pgtype.UUID]uuid.UUID{},
|
jobs: map[pgtype.UUID]uuid.UUID{},
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
@@ -234,7 +243,26 @@ func (s *Scheduler) rebuildUserDaily(ctx context.Context, userID pgtype.UUID, no
|
|||||||
if err := BuildSystemPlaylists(ctx, s.pool, s.logger, userID, now, s.dataDir); err != nil {
|
if err := BuildSystemPlaylists(ctx, s.pool, s.logger, userID, now, s.dataDir); err != nil {
|
||||||
s.logger.Warn("scheduler: build failed",
|
s.logger.Warn("scheduler: build failed",
|
||||||
"user_id", uuidStringPL(userID), "err", err)
|
"user_id", uuidStringPL(userID), "err", err)
|
||||||
|
return
|
||||||
}
|
}
|
||||||
|
// #968: tell the user's connected clients their home content was
|
||||||
|
// regenerated, so a tab/app left open across the rebuild refreshes its
|
||||||
|
// system-playlist + You-might-like views (and a stale active queue can
|
||||||
|
// re-pull) instead of serving yesterday's snapshot until a manual reload.
|
||||||
|
s.publishRebuilt(userID)
|
||||||
|
}
|
||||||
|
|
||||||
|
// publishRebuilt broadcasts a user-scoped "system playlists rebuilt" event.
|
||||||
|
// No-op when the bus is nil (test construction).
|
||||||
|
func (s *Scheduler) publishRebuilt(userID pgtype.UUID) {
|
||||||
|
if s.bus == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
s.bus.Publish(eventbus.Event{
|
||||||
|
Kind: "playlist.system_rebuilt",
|
||||||
|
UserID: uuidStringPL(userID),
|
||||||
|
Data: map[string]any{},
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// Stop drains gocron and stops the scheduler loop.
|
// Stop drains gocron and stops the scheduler loop.
|
||||||
|
|||||||
Reference in New Issue
Block a user