feat(server): GET /api/library/sync endpoint

Returns batched upserts + deletes since the supplied cursor. Empty cursor returns full snapshot; subsequent calls pull deltas. Per-user entities (likes, playlists) are scoped to the authed user. Composite-key entities (likes, playlist_tracks) use stable string ids encoded by sync.EncodeLikeID / sync.EncodePlaylistTrackID. Behavior: 204 No Content - no changes since cursor 200 OK - JSON syncResponse {cursor, upserts, deletes} 410 Gone - cursor older than oldest log row; client must reset 401 / 500 - standard envelope errors Adds sqlc queries GetAlbumsByIDs, GetTracksByIDs, GetPlaylistsByIDs to mirror the existing GetArtistsByIDs. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-09 22:36:26 -04:00
parent 3959c85111
commit 9bf4b504b3
8 changed files with 641 additions and 0 deletions
@@ -0,0 +1,287 @@
+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 := uuidToString(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 uuidToString(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 "<playlist>:<track>" — 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, <entityKey>} 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
+}
+
+// uuidToString renders a pgtype.UUID as the canonical 8-4-4-4-12 hex
+// form. Returns "" if the UUID is not valid.
+func uuidToString(u pgtype.UUID) string {
+	if !u.Valid {
+		return ""
+	}
+	b := u.Bytes
+	return formatUUIDBytes(b)
+}
+
+func formatUUIDBytes(b [16]byte) string {
+	const hex = "0123456789abcdef"
+	out := make([]byte, 36)
+	pos := 0
+	for i := 0; i < 16; i++ {
+		if i == 4 || i == 6 || i == 8 || i == 10 {
+			out[pos] = '-'
+			pos++
+		}
+		out[pos] = hex[b[i]>>4]
+		out[pos+1] = hex[b[i]&0x0f]
+		pos += 2
+	}
+	return string(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}
+}