feat(sync): append-only ops log + websocket fan-out + compaction (Phase 1.f)

CLAUDE.md

@@ -36,12 +36,13 @@ CI runs the full suite only on Linux (service container); the Windows leg is a c
 ## Architecture & conventions
 
 - **Library/binary split.** `src/main.rs` is only runtime plumbing (load `.env`, init tracing, connect pool, run migrations, bind, serve with graceful shutdown). The router is built by `waveflow_server::app(config, state)` in `src/lib.rs` so integration tests spawn the *same* app in-process (`tests/support.rs::spawn_app`). Put logic behind `app()`, not in `main`.
-- **`AppState`** (`src/lib.rs`) holds the shared singletons threaded through every handler — currently just the `PgPool` (cheap to clone, `Arc`-backed). Add new singletons here.
+- **`AppState`** (`src/lib.rs`) holds the shared singletons threaded through every handler. Cheap to clone — every field is `Arc`-backed. Fields today: `db: PgPool`, `jwt_verifier: Arc<JwtVerifier>` (Phase 1.d.1), `stream_ctx: Option<Arc<StreamCtx>>` (Phase 1.e — `None` disables streaming), `sync: SyncHub` (Phase 1.f — broadcast `Sender` + `DashMap<(user_id, device_id), AckEntry>` + flush/compaction tasks; tests construct via `SyncHub::for_tests(pool)` which skips the background loops so `flush_acks` / `compact_once` can be driven by hand for determinism). Add new singletons here.
 - **API is one file per resource** under `src/api/`, each exposing a `router()` merged in `src/api/mod.rs`. `/health` (liveness, no DB) and `/ready` (DB-aware readiness) are unversioned infra probes; every real resource mounts under `/api/v1/`.
 - **No SQL in handlers.** SQL lives in the DB layer (`src/db.rs`) or in a `waveflow-core::repository::postgres::*` method; handlers stay pure HTTP orchestration. `db::ping` (`/ready`'s `SELECT 1`) and `db::users::create` are the in-tree pattern; everything tenant-scoped goes through `PostgresProfileRepository::*_for_user`. This mirrors the desktop's Tauri-command ↔ `waveflow-core` boundary.
 - **Tenancy is enforced at the storage layer.** Server handlers under `/api/v1/*` extract `UserId` from the `middleware::authenticate` middleware and call `*_for_user` methods only — never the single-tenant trait surface. `PostgresProfileRepository` deliberately does NOT implement `ProfileRepository`, so the compiler stops a careless `list_all()` from leaking another tenant's rows. Apply the same pattern when adding library / track / playlist repositories.
 - **Auth: JWT-only (Phase 1.d.2).** `middleware::authenticate` requires a Bearer JWT signed by the upstream Better Auth issuer. The middleware verifies the token via [`AppState::jwt_verifier`], then `db::users::find_or_provision_by_external_id(state.db, &sub, now_ms)` lazy-onboards the user on first request — a valid signature is the authoritative onboarding signal, so no separate `POST /api/v1/users` exists. Boot requires the full `WAVEFLOW_JWT_*` triple (`_JWKS_URL` / `_ISSUER` / `_AUDIENCE`); the legacy `X-User-Id` dev shim retired alongside `WAVEFLOW_DEV_AUTH`.
 - **Streaming (Phase 1.e).** Two endpoints: `POST /api/v1/profiles/{p}/libraries/{l}/tracks/{t}/stream-url` (JWT-authed) verifies tenant ownership and signs a short-lived (≤ 60 s) URL via [`stream_token::mint`]; `GET /api/v1/stream/{token}` is mounted OUTSIDE the JWT layer because browsers can't attach a Bearer to `<audio src>` — the HMAC in the token IS the auth. The stream handler canonicalises `<music_root>/<file_path>` and refuses anything resolving outside `WAVEFLOW_MUSIC_ROOT` (path-traversal guard via `std::fs::canonicalize` + prefix check). Range requests handled in-process: `Accept-Ranges: bytes`, 206 + `Content-Range` on partial, 416 on unsatisfiable. Both endpoints answer 503 when streaming is disabled at boot (`WAVEFLOW_MUSIC_ROOT` + `WAVEFLOW_STREAM_SECRET` unset). Until 1.f's sync ships, files must be placed manually under the music root.
+- **Sync (Phase 1.f).** Append-only `sync_op` log keyed on `BIGSERIAL id`; per-`(user, device)` UNIQUEs on `operation_id` (idempotency) and `lamport_ts` (monotonicity). Three REST routes + one WebSocket under `/api/v1/sync/*`: `POST /ops` push (idempotent replay via `ON CONFLICT operation_id DO NOTHING`, 409 + `stored_max` on lamport regression), `GET /ops?since=N` pull (410 + `compacted_up_to` when `0 < since < watermark`; `since=0` is the bootstrap path and always skips the guard so fresh devices can converge from an empty cursor — read-only, never advances the cursor), `POST /ack` (the **only** path that writes `device_sync_cursor`, buffered in memory + flushed every 5 s), `GET /ws?device_id=…` WebSocket fan-out (one broadcast channel, per-frame `user_id` filter so cross-tenant ops never leak). Live state in `sync::SyncHub`: tokio `broadcast::Sender` + `DashMap<(user_id, device_id), AckEntry>` + daily compaction task that flushes ACKs first then collapses superseded ops in the same Postgres transaction as the watermark UPSERT. Watermark monotonic by UPSERT `WHERE` clause. Stale devices (`last_seen_at < now - 90 d`) skipped from the compaction MIN. Tests drive `SyncHub::for_tests` (no background tasks) and call `flush_acks` / `compact_once` directly for determinism.
 - **Don't leak DB errors to unauthenticated probes.** `/ready` logs the sqlx error via `tracing::warn!` but returns a fixed sentinel body (`{status, db}`) so a load balancer never sees the connection-URL host or credentials. Apply the same discipline to any other unauthenticated endpoint.
 - **Migrations are immutable once merged.** They're embedded at compile time via `sqlx::migrate!("./migrations")` (`db::MIGRATOR`); the `_sqlx_migrations` table stores each file's checksum, so editing an applied migration makes the server refuse to start. Schema changes = a new dated migration file (`YYYYMMDDHHMMSS_name.sql`). Boot applies pending migrations *before* opening the listener, which is what makes `/ready` trustworthy.
 - **Schema parity with the desktop SQLite migrations.** Postgres tables mirror the shapes in the desktop repo's `src-tauri/migrations/app/` so `PostgresProfileRepository` and `SqliteProfileRepository` (in `waveflow-core`) satisfy the same trait against identical rows. Keep types compatible (e.g. `BIGSERIAL` ↔ SQLite `INTEGER PK`, epoch-millis `BIGINT` for timestamps).

Cargo.toml

@@ -32,7 +32,7 @@ waveflow-core = { git = "https://github.com/InstaZDLL/WaveFlow", rev = "25b9ada6
 # mandate `sslmode=require` (Neon, Supabase, Prisma Accelerate, RDS).
 # Pure-Rust rustls matches the convention reqwest uses elsewhere in
 # the project — no OpenSSL system dep to ship in the container image.
-sqlx = { version = "0.9", default-features = false, features = ["runtime-tokio", "postgres", "macros", "migrate", "chrono", "tls-rustls"] }
+sqlx = { version = "0.9", default-features = false, features = ["runtime-tokio", "postgres", "macros", "migrate", "chrono", "uuid", "tls-rustls"] }
 
 # Wall-clock timestamps for `created_at` / `last_used_at`. Same
 # crate the desktop uses so a shared timestamp helper stays cheap.
@@ -44,16 +44,29 @@ chrono = "0.4"
 # without a parallel paths(...) list to maintain. `utoipa-scalar`
 # serves the Scalar API reference (Stripe-style — modern, ~500 KB
 # bundle, dark mode, integrated search) from `/reference`.
-utoipa = { version = "5", features = ["axum_extras"] }
+utoipa = { version = "5", features = ["axum_extras", "uuid"] }
 utoipa-axum = "0.2"
 utoipa-scalar = { version = "0.3", features = ["axum"] }
 
 # HTTP layer. axum 0.8 is the line current for Tokio 1.45+ — the same
 # generation used by the desktop's reqwest 0.12, so we're not pulling
 # two competing hyper trees once `waveflow-core` lands as a dependency
-# in 1.b.2.
-axum = "0.8"
-tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal"] }
+# in 1.b.2. The `ws` feature gates the WebSocket extractor the
+# sync-broadcast endpoint (Phase 1.f) needs.
+axum = { version = "0.8", features = ["ws"] }
+tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal", "sync", "time"] }
+# Stream combinators for the WebSocket handler — `SinkExt::send` +
+# `StreamExt::next` on the split halves of an `axum::extract::ws::WebSocket`.
+futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] }
+# UUIDv4 for `sync_op.operation_id` — the idempotency key the client
+# generates and the server enforces uniqueness on per `(user_id,
+# device_id, operation_id)`. `v4` for the random generator, `serde`
+# for axum's `Json<…>` round-trip.
+uuid = { version = "1", features = ["v4", "serde"] }
+# Lock-free concurrent map for the in-memory ACK buffer in
+# `sync::SyncHub`. Per-key shard locks mean handler-side `record_ack`
+# never blocks behind the 5 s flusher.
+dashmap = "6"
 tower = "0.5"
 tower-http = { version = "0.6", features = ["trace", "request-id", "timeout", "util"] }
 
@@ -155,6 +168,21 @@ p256 = { version = "0.13", features = ["ecdsa", "pkcs8"] }
 # no cleanup boilerplate.
 tempfile = "3"
 
+# WebSocket client for the sync fan-out integration tests. The
+# tokio-tungstenite version that pairs with tokio 1.x ships only
+# rustls feature flags — `rustls-tls-webpki-roots` matches the
+# pure-Rust convention used elsewhere in the project.
+tokio-tungstenite = { version = "0.27", default-features = false, features = ["connect", "rustls-tls-webpki-roots"] }
+
+# `record_ack` takes a `uuid::Uuid`; the sync tests construct one per
+# op. Use the `serde` feature so the same value round-trips through
+# the JSON request body.
+uuid = { version = "1", features = ["v4", "serde"] }
+
+# Streaming tests already pulled `futures-util` transitively; the
+# sync WS test needs `StreamExt::next` + `SinkExt::send` directly.
+futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] }
+
 [profile.release]
 # Match the desktop app's release profile so behaviour parity holds at
 # the binary level. Single-codegen-unit + LTO trade compile time for