package api import ( "context" "encoding/json" "net/http" "strconv" "github.com/jackc/pgx/v5/pgtype" "git.fabledsword.com/bvandeusen/minstrel/internal/apierror" "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" syncpkg "git.fabledsword.com/bvandeusen/minstrel/internal/sync" ) // SyncBatchLimit caps the number of change rows returned per call. // Clients that need more loop with the returned cursor. const SyncBatchLimit = 1000 // syncResponse is the wire shape returned by GET /api/library/sync. // upserts and deletes are separated by entity type so clients can apply // them to typed local tables. cursor is the largest library_changes.id // the server returned in this batch — clients pass it back as `since` // next time. type syncResponse struct { Cursor int64 `json:"cursor"` Upserts map[string][]json.RawMessage `json:"upserts"` Deletes map[string][]string `json:"deletes"` } // handleLibrarySync returns batched upserts + deletes since the supplied // cursor. Cursor is the last `library_changes.id` the client has seen. // Empty / zero / invalid cursor means "give me everything" (initial sync). // // Responses: // - 204 No Content — no changes since cursor // - 200 OK — JSON syncResponse with upserts, deletes, new cursor // - 410 Gone — cursor older than oldest log row; client must reset // - 401 / 500 — standard envelope errors func (h *handlers) handleLibrarySync(w http.ResponseWriter, r *http.Request) { user, ok := requireUser(w, r) if !ok { return } ctx := r.Context() since, err := strconv.ParseInt(r.URL.Query().Get("since"), 10, 64) if err != nil || since < 0 { since = 0 } q := dbq.New(h.pool) // 410 path: if the requested cursor is older than the oldest row, // tell the client to reset. (No compaction job in this slice; the // path exists for future compaction support.) if since > 0 { minCursor, err := q.GetMinLibraryChangeCursor(ctx) if err != nil { writeErrWithLog(w, h.logger, "library_sync: GetMinLibraryChangeCursor", err) return } if minCursor > 0 && since+1 < minCursor { writeErr(w, &apierror.Error{ Status: http.StatusGone, Code: "cursor_too_old", Message: "reset client cursor to 0 and resync", }) return } } changes, err := q.GetLibraryChangesSince(ctx, dbq.GetLibraryChangesSinceParams{ ID: since, Limit: SyncBatchLimit, }) if err != nil { writeErrWithLog(w, h.logger, "library_sync: GetLibraryChangesSince", err) return } if len(changes) == 0 { w.WriteHeader(http.StatusNoContent) return } // Group ids by entity type so we can fetch upsert payloads in batches. upsertIDs := map[syncpkg.EntityType][]string{} deleteIDs := map[syncpkg.EntityType][]string{} var maxID int64 for _, c := range changes { et := syncpkg.EntityType(c.EntityType) if c.ID > maxID { maxID = c.ID } if c.Op == string(syncpkg.OpUpsert) { upsertIDs[et] = append(upsertIDs[et], c.EntityID) } else { deleteIDs[et] = append(deleteIDs[et], c.EntityID) } } upserts, err := h.hydrateUpserts(ctx, q, user.ID, upsertIDs) if err != nil { writeErrWithLog(w, h.logger, "library_sync: hydrateUpserts", err) return } // Format deletes — convert EntityType keys to strings for JSON. deletes := make(map[string][]string, len(deleteIDs)) for et, ids := range deleteIDs { deletes[string(et)] = ids } writeJSON(w, http.StatusOK, syncResponse{ Cursor: maxID, Upserts: upserts, Deletes: deletes, }) } // hydrateUpserts loads the current row payloads for each upserted id, // keyed by entity type (string form). Returns json.RawMessage values so // each entity's full sqlc row shape passes through unchanged. // // Per-user entities (likes, playlists, playlist_tracks) are scoped to // userID — the change log is global but each user only sees rows that // concern them. func (h *handlers) hydrateUpserts( ctx context.Context, q *dbq.Queries, userID pgtype.UUID, ids map[syncpkg.EntityType][]string, ) (map[string][]json.RawMessage, error) { out := map[string][]json.RawMessage{} if rows := ids[syncpkg.EntityArtist]; len(rows) > 0 { uuids := stringsToUUIDs(rows) artists, err := q.GetArtistsByIDs(ctx, uuids) if err != nil { return nil, err } for _, a := range artists { b, _ := json.Marshal(a) out["artist"] = append(out["artist"], b) } } if rows := ids[syncpkg.EntityAlbum]; len(rows) > 0 { uuids := stringsToUUIDs(rows) albums, err := q.GetAlbumsByIDs(ctx, uuids) if err != nil { return nil, err } for _, a := range albums { b, _ := json.Marshal(a) out["album"] = append(out["album"], b) } } if rows := ids[syncpkg.EntityTrack]; len(rows) > 0 { uuids := stringsToUUIDs(rows) tracks, err := q.GetTracksByIDs(ctx, uuids) if err != nil { return nil, err } for _, t := range tracks { b, _ := json.Marshal(t) out["track"] = append(out["track"], b) } } userIDStr := syncpkg.FormatUUID(userID) if rows := ids[syncpkg.EntityLikeTrack]; len(rows) > 0 { out["like_track"] = scopedLikeRows(rows, userIDStr, "track_id") } if rows := ids[syncpkg.EntityLikeAlbum]; len(rows) > 0 { out["like_album"] = scopedLikeRows(rows, userIDStr, "album_id") } if rows := ids[syncpkg.EntityLikeArtist]; len(rows) > 0 { out["like_artist"] = scopedLikeRows(rows, userIDStr, "artist_id") } if rows := ids[syncpkg.EntityPlaylist]; len(rows) > 0 { uuids := stringsToUUIDs(rows) playlists, err := q.GetPlaylistsByIDs(ctx, uuids) if err != nil { return nil, err } for _, p := range playlists { // Filter: only return playlists owned by this user OR public ones if syncpkg.FormatUUID(p.UserID) != userIDStr && !p.IsPublic { continue } b, _ := json.Marshal(p) out["playlist"] = append(out["playlist"], b) } } if rows := ids[syncpkg.EntityPlaylistTrack]; len(rows) > 0 { // playlist_track ids are ":" — we don't know // playlist ownership here without a JOIN. Pass through; client // reconciles against its locally-cached playlists. var msgs []json.RawMessage for _, id := range rows { parts := splitOnce(id, ":") if len(parts) != 2 { continue } b, _ := json.Marshal(map[string]string{ "playlist_id": parts[0], "track_id": parts[1], }) msgs = append(msgs, b) } out["playlist_track"] = msgs } return out, nil } // scopedLikeRows filters composite-key like rows to those owned by userID // and emits {user_id, } JSON shapes. entityKey is "track_id", // "album_id", or "artist_id" depending on the like table. func scopedLikeRows(rows []string, userIDStr, entityKey string) []json.RawMessage { var msgs []json.RawMessage for _, id := range rows { parts := splitOnce(id, ":") if len(parts) != 2 || parts[0] != userIDStr { continue } b, _ := json.Marshal(map[string]string{ "user_id": parts[0], entityKey: parts[1], }) msgs = append(msgs, b) } return msgs } // stringsToUUIDs parses each string as a pgtype.UUID. Invalid entries // are skipped silently — they can't have come from a valid scan/mutation. func stringsToUUIDs(strs []string) []pgtype.UUID { out := make([]pgtype.UUID, 0, len(strs)) for _, s := range strs { u, ok := parseUUID(s) if !ok { continue } out = append(out, u) } return out } // splitOnce returns [before, after] split at the first occurrence of sep, // or [s] if sep doesn't appear. Used for composite-key parsing. func splitOnce(s, sep string) []string { for i := 0; i+len(sep) <= len(s); i++ { if s[i:i+len(sep)] == sep { return []string{s[:i], s[i+len(sep):]} } } return []string{s} }