feat(server): sync.LogChange helper + EntityType/Op constants
New package internal/sync. LogChange writes a library_changes row inside the supplied tx. Encode helpers produce stable composite ids for like_* and playlist_track entries. Subsequent commits wire LogChange into the scanner / likes / playlists services. Also: dbtest.dataTables now includes library_changes so test isolation holds across runs. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -54,6 +54,7 @@ var dataTables = []string{
|
|||||||
"lidarr_quarantine",
|
"lidarr_quarantine",
|
||||||
"playlist_tracks",
|
"playlist_tracks",
|
||||||
"playlists",
|
"playlists",
|
||||||
|
"library_changes", // M7 #357 — must reset to keep cursor isolated per test
|
||||||
"tracks",
|
"tracks",
|
||||||
"albums",
|
"albums",
|
||||||
"artists",
|
"artists",
|
||||||
|
|||||||
@@ -0,0 +1,42 @@
|
|||||||
|
package sync
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
|
||||||
|
"github.com/jackc/pgx/v5"
|
||||||
|
|
||||||
|
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||||
|
)
|
||||||
|
|
||||||
|
// LogChange writes a row to library_changes inside the supplied transaction.
|
||||||
|
// Callers MUST pass the same tx that performed the underlying mutation —
|
||||||
|
// the change row and the mutation must commit or roll back together.
|
||||||
|
//
|
||||||
|
// entityID is the string form of the row's primary key. For composite-key
|
||||||
|
// entities (likes, playlist_tracks), use EncodeLikeID / EncodePlaylistTrackID
|
||||||
|
// to produce stable strings.
|
||||||
|
func LogChange(ctx context.Context, tx pgx.Tx, entityType EntityType, entityID string, op Op) error {
|
||||||
|
q := dbq.New(tx)
|
||||||
|
if err := q.InsertLibraryChange(ctx, dbq.InsertLibraryChangeParams{
|
||||||
|
EntityType: string(entityType),
|
||||||
|
EntityID: entityID,
|
||||||
|
Op: string(op),
|
||||||
|
}); err != nil {
|
||||||
|
return fmt.Errorf("sync.LogChange(%s, %s, %s): %w", entityType, entityID, op, err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// EncodeLikeID joins a user UUID and an entity UUID into the stable
|
||||||
|
// composite identifier used for like_* rows in library_changes. Both
|
||||||
|
// sides are UUID strings so no escaping is needed.
|
||||||
|
func EncodeLikeID(userID, entityID string) string {
|
||||||
|
return userID + ":" + entityID
|
||||||
|
}
|
||||||
|
|
||||||
|
// EncodePlaylistTrackID joins a playlist UUID and a track UUID into the
|
||||||
|
// stable composite identifier used for playlist_track rows.
|
||||||
|
func EncodePlaylistTrackID(playlistID, trackID string) string {
|
||||||
|
return playlistID + ":" + trackID
|
||||||
|
}
|
||||||
@@ -0,0 +1,86 @@
|
|||||||
|
package sync_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"io"
|
||||||
|
"log/slog"
|
||||||
|
"os"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/jackc/pgx/v5/pgxpool"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
|
||||||
|
"git.fabledsword.com/bvandeusen/minstrel/internal/db"
|
||||||
|
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||||
|
"git.fabledsword.com/bvandeusen/minstrel/internal/dbtest"
|
||||||
|
"git.fabledsword.com/bvandeusen/minstrel/internal/sync"
|
||||||
|
)
|
||||||
|
|
||||||
|
func testPool(t *testing.T) *pgxpool.Pool {
|
||||||
|
t.Helper()
|
||||||
|
if testing.Short() {
|
||||||
|
t.Skip("skipping integration test in -short mode")
|
||||||
|
}
|
||||||
|
dsn := os.Getenv("MINSTREL_TEST_DATABASE_URL")
|
||||||
|
if dsn == "" {
|
||||||
|
t.Skip("MINSTREL_TEST_DATABASE_URL not set")
|
||||||
|
}
|
||||||
|
if err := db.Migrate(dsn, slog.New(slog.NewTextHandler(io.Discard, nil))); err != nil {
|
||||||
|
t.Fatalf("migrate: %v", err)
|
||||||
|
}
|
||||||
|
pool, err := pgxpool.New(context.Background(), dsn)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("pool: %v", err)
|
||||||
|
}
|
||||||
|
t.Cleanup(pool.Close)
|
||||||
|
dbtest.ResetDB(t, pool)
|
||||||
|
return pool
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestLogChange_WritesRowInTx(t *testing.T) {
|
||||||
|
pool := testPool(t)
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
tx, err := pool.Begin(ctx)
|
||||||
|
require.NoError(t, err)
|
||||||
|
defer tx.Rollback(ctx)
|
||||||
|
|
||||||
|
require.NoError(t, sync.LogChange(ctx, tx, sync.EntityArtist, "artist-123", sync.OpUpsert))
|
||||||
|
require.NoError(t, tx.Commit(ctx))
|
||||||
|
|
||||||
|
q := dbq.New(pool)
|
||||||
|
rows, err := q.GetLibraryChangesSince(ctx, dbq.GetLibraryChangesSinceParams{ID: 0, Limit: 10})
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.NotEmpty(t, rows)
|
||||||
|
last := rows[len(rows)-1]
|
||||||
|
require.Equal(t, "artist", last.EntityType)
|
||||||
|
require.Equal(t, "artist-123", last.EntityID)
|
||||||
|
require.Equal(t, "upsert", last.Op)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestLogChange_RollbackDoesNotPersist(t *testing.T) {
|
||||||
|
pool := testPool(t)
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
q := dbq.New(pool)
|
||||||
|
cursorBefore, err := q.GetMaxLibraryChangeCursor(ctx)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
tx, err := pool.Begin(ctx)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.NoError(t, sync.LogChange(ctx, tx, sync.EntityAlbum, "album-456", sync.OpDelete))
|
||||||
|
require.NoError(t, tx.Rollback(ctx))
|
||||||
|
|
||||||
|
cursorAfter, err := q.GetMaxLibraryChangeCursor(ctx)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.Equal(t, cursorBefore, cursorAfter, "rollback must not advance cursor")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Pure unit tests — run even with -short.
|
||||||
|
func TestEncodeLikeID(t *testing.T) {
|
||||||
|
require.Equal(t, "u1:t2", sync.EncodeLikeID("u1", "t2"))
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEncodePlaylistTrackID(t *testing.T) {
|
||||||
|
require.Equal(t, "p1:t2", sync.EncodePlaylistTrackID("p1", "t2"))
|
||||||
|
}
|
||||||
@@ -0,0 +1,40 @@
|
|||||||
|
// Package sync supports the Flutter delta-sync endpoint by writing an
|
||||||
|
// append-only library_changes log alongside every mutation on tracked
|
||||||
|
// entities (artists, albums, tracks, likes, playlists, playlist_tracks).
|
||||||
|
//
|
||||||
|
// Callers MUST pass the same transaction that performed the underlying
|
||||||
|
// mutation — the change row and the mutation must commit or roll back
|
||||||
|
// together. See LogChange in changes.go.
|
||||||
|
package sync
|
||||||
|
|
||||||
|
// EntityType enumerates the entity kinds whose mutations are tracked
|
||||||
|
// by the library_changes log. Values must stay in sync with the CHECK
|
||||||
|
// constraint on library_changes.entity_type (migration 0025).
|
||||||
|
type EntityType string
|
||||||
|
|
||||||
|
const (
|
||||||
|
EntityArtist EntityType = "artist"
|
||||||
|
EntityAlbum EntityType = "album"
|
||||||
|
EntityTrack EntityType = "track"
|
||||||
|
EntityLikeTrack EntityType = "like_track"
|
||||||
|
EntityLikeAlbum EntityType = "like_album"
|
||||||
|
EntityLikeArtist EntityType = "like_artist"
|
||||||
|
EntityPlaylist EntityType = "playlist"
|
||||||
|
EntityPlaylistTrack EntityType = "playlist_track"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Op is upsert or delete. Matches the CHECK constraint in migration 0025.
|
||||||
|
type Op string
|
||||||
|
|
||||||
|
const (
|
||||||
|
OpUpsert Op = "upsert"
|
||||||
|
OpDelete Op = "delete"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Change is the wire shape returned by the sync endpoint to clients.
|
||||||
|
type Change struct {
|
||||||
|
ID int64 `json:"id"`
|
||||||
|
EntityType EntityType `json:"entity_type"`
|
||||||
|
EntityID string `json:"entity_id"`
|
||||||
|
Op Op `json:"op"`
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user