feat(apply): server-side materialisation of desktop sync ops (phase 1.g.0) by InstaZDLL · Pull Request #26 · InstaZDLL/waveflow-server

InstaZDLL · 2026-06-04T20:29:44Z

Summary

Until now `sync_op` was append-only: the desktop pushed ops, the server stored them, but no consumer materialised them into the entity tables. This PR closes that gap with a synchronous apply pipeline that runs in the same transaction as the durable insert.

Why now: Phase 1.g.2 (web share preview) is merged, but desktop-created playlists are invisible to the share endpoint because they live only in `sync_op` log. This PR unblocks 1.g.3 desktop sharing.

What lands

Migration `20260604000000_apply_pipeline.sql`

`profile.canonical_id`, `library.canonical_id`, `playlist.canonical_id` — partial-unique-indexed TEXT, tenant-scoped
`track.file_hash` — non-unique index (for future track sync join)
`sync_op.profile_canonical_id` — routing key for apply
`user_liked_track(user_id, file_hash, liked_at)` — file-hash keyed
`user_track_rating(user_id, file_hash, rating, updated_at)` — same POPM byte shape as `track.rating`

NOT included: `playlist_track` materialisation. Desktop emits local BIGINT track ids that have no meaning server-side. Pending desktop emitting file_hash refs.

Apply pipeline (`src/apply.rs`)

`apply_op(conn, user_id, op, now)` called from `api::sync::push_ops` right after each freshly-inserted log row, inside the same transaction
A failure rolls the log row back too — better to refuse a push than leave an op the server can't honour
Routing: `playlist` / `library` resolve profile via `find_or_provision` (read-first, UPSERT on miss, race-safe); `liked_track` / `track_rating` skip profile lookup (file_hash keyed); unknown entities log + skip (forward-compat)

Tests (`tests/apply.rs`)

11 integration tests covering the contract:

Insert / replay-idempotent / set-field / delete for playlist + library
Rating set + delete + out-of-range rollback
Liked insert + delete round-trip
Missing profile_canonical_id keeps durable row but skips apply
Two canonical_ids → two distinct server profiles
Tenant isolation (user A's apply doesn't bleed into B)

Wire shape change

`SyncOpIn` and `SyncOp` gain a `profile_canonical_id: Option` field. Legacy clients (no field) still push successfully — durable log stores NULL, apply skips. Forward-only protocol bump.

Test plan

CI runs the 11 new apply tests + the 12 existing sync tests + share tests
`cargo check --all-targets` clean locally (0 warnings)
Manual: push playlist insert with profile_canonical_id, verify playlist row appears, then share_mint via canonical_id lookup (follow-up PR)

Follow-ups (deferred to keep this PR reviewable)

Canonical-id-aware share endpoint (`POST /api/v1/share/by-canonical/{profile}/{playlist}`) — ~100 LOC. Could land in this PR if requested.
Desktop changes: `profile.canonical_id` generation + drain task injecting `profile_canonical_id` per op (~300 LOC, separate WaveFlow PR).
`playlist_track` apply once desktop emits file_hash refs.

Summary by CodeRabbit

Nouvelles fonctionnalités
- Synchronisation serveur améliorée : les opérations sont appliquées atomiquement lors du push, avec mapping via un identifiant canonique de profil.
- Meilleure synchronisation des playlists, bibliothèques, titres aimés et notes, et appariement basé sur un hash de fichier.
Corrections
- Validation renforcée : erreurs de payload ou valeurs hors plage annulant l’opération pour éviter états partiels.
Tests
- Ajout de tests end-to-end couvrant matérialisation, idempotence et isolation multi-utilisateur.

….g.0) Until now `sync_op` was append-only: the desktop pushed ops, the server stored them, but no consumer materialised them into the entity tables — so a desktop-created playlist was invisible to playlist sharing, library APIs, web views, and any other surface that reads from `playlist`/`library`/etc. This PR closes that gap with a synchronous apply pipeline that runs in the same transaction as the durable insert. ## Migration `20260604000000_apply_pipeline.sql` adds: - `profile.canonical_id`, `library.canonical_id`, `playlist.canonical_id` — partial-unique-indexed TEXT columns scoped to the parent tenant. Legacy server-created rows keep working (canonical_id NULL). - `track.file_hash` — non-unique index. Server has no track sync yet; the column is there for the future join with rating / liked tables. - `sync_op.profile_canonical_id` — routing key for apply. Legacy ops without it stay in the durable log but skip apply. - `user_liked_track(user_id, file_hash, liked_at)` — file-hash keyed so liked ops can land even before a corresponding track row exists. - `user_track_rating(user_id, file_hash, rating, updated_at)` — same shape as `track.rating` (raw POPM byte) so a future track sync can backfill via INNER JOIN on file_hash. Explicitly NOT included: `playlist_track` materialisation. Desktop emits these ops with local BIGINT track ids that have no meaning on the server. Until desktop emits file_hash refs + track sync ships, the apply path logs them as Skipped. ## Apply pipeline (`src/apply.rs`) `apply_op(conn, user_id, op, now) -> Result<ApplyOutcome, ApplyError>` is called from `api::sync::push_ops` right after each freshly-inserted log row, inside the same transaction. A failure rolls the log row back too — better to refuse a push than to leave an op the server can't honour. Routing: - `playlist` / `library` ops — resolve `profile_canonical_id` to a server `profile.id` via `find_or_provision` (read-first, UPSERT on miss, race-safe), then dispatch to the per-entity module. - `liked_track` / `track_rating` — keyed on `(user_id, file_hash)`; no profile lookup needed. - Anything else — `Skipped` (recognised + unsupported) or `Unknown` (forward-compat). Durable log always retains the row. ## Tests (`tests/apply.rs`) 11 integration tests covering the contract: - playlist / library insert materialises a row with the right defaults and canonical_id mapping - replay of the same operation_id is idempotent (one playlist row) - `set name` updates the field - `delete` removes the row - rating set persists; rating out-of-range short-circuits with rollback; rating delete clears - liked insert / delete round-trip - missing profile_canonical_id keeps the durable row but skips apply - two distinct profile_canonical_ids land in two distinct server profiles - tenant isolation: user A's apply doesn't bleed into user B's profile / playlist counts Signed-off-by: InstaZDLL <github.105mh@8shield.net>

coderabbitai · 2026-06-04T20:29:51Z

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: 879e794c-0a30-4d1a-ba89-ae6c678e32fa

📥 Commits

Reviewing files that changed from the base of the PR and between 7d86ff6 and efd8e3a.

📒 Files selected for processing (2)

src/apply.rs
tests/apply.rs

📝 Walkthrough

Walkthrough

Ce PR ajoute un pipeline server-side apply : migration DB pour canonical/file_hash, extension du protocole (profile_canonical_id), adaptation des requêtes, nouveau module apply (résolution de profil, handlers pour playlist/library/liked_track/track_rating) et tests E2E; push_ops applique ops dans la même transaction.

Changes

Pipeline Apply Phase 1.g

Layer / File(s)	Summary
Schéma et modèle de données `migrations/20260604000000_apply_pipeline.sql`	Migration SQL ajoutant `canonical_id` sur `profile/library/playlist`, `track.file_hash`, `sync_op.profile_canonical_id`, et création de `user_liked_track` et `user_track_rating` (PK `(user_id,file_hash)`, contraintes ON DELETE CASCADE).
Extension du protocole de synchronisation `src/sync.rs`	Ajout de `profile_canonical_id: Option<String>` à `SyncOpIn` et `SyncOp` (serde default).
Couche d'accès base de données `src/db.rs`	`insert_op_returning`, `fetch_op_by_operation_id`, `pull_ops_since` étendent SELECT/RETURNING et bindings pour `profile_canonical_id`.
Export module `src/lib.rs`	Ajout de `pub mod apply;`.
Intégration API et flux transactionnel `src/api/sync.rs`	`push_ops` appelle désormais `apply_op` après l'INSERT dans la même transaction ; échec d'application → rollback + 500 ; `row_to_op` inclut `profile_canonical_id`.
Noyau du pipeline apply et utilitaires `src/apply.rs` (types et routing)	Ajout de `ApplyError`/`ApplyOutcome`, `apply_op` (dispatch par entité), utilitaires JSON de parsing, résolution de profil via `find_or_provision`.
Handlers d'entités `src/apply.rs` (handlers)	Handlers pour `playlist` (insert/set/delete, tracks → Skipped), `library` (mirror), `liked_track` (insert/delete on (user_id,file_hash)), `track_rating` (validation 0..=255, upsert/delete).
Suite de tests end-to-end `tests/apply.rs`	Tests E2E couvrant matérialisation, idempotence, validations et rollback (rating/type), liked_track, cas sans `profile_canonical_id`, multi-canonical-id, isolation multi-utilisateur.

Sequence Diagrams

sequenceDiagram
    participant Client
    participant PushOps as push_ops handler
    participant DB as Postgres
    participant Apply as apply_op

    Client->>PushOps: POST /api/v1/sync/ops (SyncOpIn)
    PushOps->>DB: INSERT INTO sync_op RETURNING *
    DB-->>PushOps: row (incl. profile_canonical_id)
    PushOps->>PushOps: row_to_op -> SyncOp
    PushOps->>Apply: apply_op(user_id, SyncOp)
    Apply->>DB: SELECT profile WHERE canonical_id=...
    DB-->>Apply: profile.id (or none)
    Apply->>DB: INSERT/UPDATE/DELETE entity tables (playlist/library/user_liked_track/user_track_rating)
    alt Apply success
        DB-->>Apply: OK
        Apply-->>PushOps: ApplyOutcome::Applied
        PushOps->>DB: COMMIT
        PushOps-->>Client: 200 Accepted
    else Apply failure
        Apply-->>PushOps: ApplyError
        PushOps->>DB: ROLLBACK
        PushOps-->>Client: 500 Internal Error
    end

Estimated Code Review Effort

🎯 4 (Complex) | ⏱️ ~45 minutes

Possibly Related PRs

InstaZDLL/waveflow-server#19: Changements autour de push_ops / sync_op et du flux push↔commit, très lié à l'appel apply_op transactionnel ajouté ici.
InstaZDLL/waveflow-server#10: Modifications antérieures du modèle playlist et migrations connexes sur lesquelles la résolution canonical_id s'appuie.

Poem

Dans le log les ops s'alignent, puis l'apply se lance,
Canonical éclaire le profil, la table prend naissance,
Likes et notes trouvent leur case avec confiance,
Si l'apply échoue, tout rebondit en cadence,
Transactionnel, précis — la base garde la confiance.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	Le titre décrit précisément le changement principal : introduction d'un pipeline apply côté serveur pour matérialiser les opérations de sync desktop, avec indication claire de la phase.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

📝 Generate docstrings

Create stacked PR
Commit on current branch

🧪 Generate unit tests (beta)

Create PR with unit tests
Commit unit tests in branch feat/1-g-0-apply-pipeline

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

Signed-off-by: InstaZDLL <github.105mh@8shield.net>

coderabbitai

Actionable comments posted: 3

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@src/apply.rs`:
- Around line 212-217: La fonction payload_optional_string silencieusement
convertit un type invalide en None; changez sa signature et son comportement
pour renvoyer Result<Option<String>, InvalidPayload> (ou le type d'erreur
existant, ex. ApplyError::InvalidPayload) et validez explicitement la
présence/valeur: si payload.get(key) est None => Ok(None); si c'est Value::Null
=> Ok(None) (ou l'équivalent pour un null explicite selon le contexte d'appel);
si c'est une string => Ok(Some(string.to_owned())); sinon => Err(InvalidPayload{
key, expected: "string | null" }); modifiez les appels existants de
payload_optional_string pour propager l'erreur (utiliser ?), en faisant
référence à la fonction payload_optional_string et au type d'erreur
InvalidPayload/ApplyError dans votre code.

In `@tests/apply.rs`:
- Around line 122-128: Replace the unnecessary clone when passing a single-item
slice to push: change the call that currently builds a slice with
&[body.clone()] to use std::slice::from_ref(&body) so you pass a &[BodyType]
without cloning; update both push(&auth.base, &auth.token, &[body.clone()]) and
the later duplicate to push(&auth.base, &auth.token,
std::slice::from_ref(&body)) (referencing the push function and the local body
variable to locate the sites).
- Around line 22-47: La fonction de test op dépasse la limite de paramètres de
Clippy (8 args) ; corrige en choisissant soit créer un petit builder struct
(p.ex. OpBuilder avec méthodes chainables build() pour produire la Value) et
remplacer les appels à op par OpBuilder::new(...)..., soit simplement ajouter
#[allow(clippy::too_many_arguments)] directement au-dessus de la définition de
la fonction op pour désactiver l'avertissement de Clippy pour cet helper de test
; faites la modification sur la fonction nommée op et/ou introduisez OpBuilder
si vous préférez refactoriser.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

Push a commit to this branch (recommended)
Create a new PR with the fixes

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: e5e58ebb-fc4e-49bb-abeb-58ebd37d5d98

📥 Commits

Reviewing files that changed from the base of the PR and between 9bab9d0 and 7d86ff6.

📒 Files selected for processing (7)

migrations/20260604000000_apply_pipeline.sql
src/api/sync.rs
src/apply.rs
src/db.rs
src/lib.rs
src/sync.rs
tests/apply.rs

Three findings from CI + CodeRabbit on PR #26: payload_optional_string was silently coercing a non-string, non-null value (e.g. a number) to None — turning a desktop bug into a "clear this field" UPDATE on the server. Changed the signature to return Result<Option<String>, ApplyError> with an explicit InvalidPayload error for the typed-mismatch case, and propagated `?` to every call site (playlist + library insert / set_field). Null and absent stay valid "clear" signals. Clippy too_many_arguments on the test op() helper — folding 8 args into a builder would obscure call sites that read like a wire-format dump. #[allow] locally with a comment explaining the trade-off. Clippy cloned_ref_to_slice_refs on &[body.clone()] in the replay test — swapped to std::slice::from_ref(&body) on both calls. Plus a new test pinning the typed-payload error path so a future regression on payload_optional_string surfaces as a failed test instead of as a silently-cleared description column. Signed-off-by: InstaZDLL <github.105mh@8shield.net>

InstaZDLL · 2026-06-04T21:01:38Z

Round 1 CR addressed in efd8e3a.

1. Clippy too_many_arguments on op() — added #[allow(clippy::too_many_arguments)] with a comment. Folding into a builder would obscure the wire-format-dump call sites; the helper is fixture-only.

2. Clippy cloned_ref_to_slice_refs — both &[body.clone()] swapped to std::slice::from_ref(&body).

3. @coderabbitai payload_optional_string silent coercion — sharp finding, confirmed it was a real bug. A desktop payload like { "description": 42 } would have silently cleared the column instead of erroring. Changed the signature to Result<Option<String>, ApplyError>:

key absent → Ok(None)
value null → Ok(None) (valid explicit-clear signal)
value is a string → Ok(Some(...))
value is anything else → Err(InvalidPayload) with the JSON kind in the message

Propagated ? to all 8 call sites (playlist + library insert + set_field). Added a new test playlist_insert_with_wrong_type_for_optional_string_is_rejected pinning the rejection + transaction rollback so a future regression surfaces as a failed test instead of a silently-cleared column.

Local validation: cargo fmt clean, cargo clippy --all-targets --all-features -- -D warnings clean, cargo test --test apply → 12/12. Full suite cargo test → 129/129. (Postgres is now running locally, hence the move from "trust CI" to "locally validated".)

coderabbitai · 2026-06-04T21:02:22Z

Tip

For best results, initiate chat on the files or code changes.

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Check the state of apply.rs after the fixes, focusing on payload_optional_string
gh pr view 26 --repo InstaZDLL/waveflow-server --json headRefName,headRefOid
git show efd8e3a --stat 2>/dev/null || echo "commit not found directly, checking branch"
# Show the current payload_optional_string implementation
rg -n "payload_optional_string" --type rust -A 10

Length of output: 7043

🏁 Script executed:

#!/bin/bash
# Check the new test and the fixed implementation
rg -n "playlist_insert_with_wrong_type" --type rust -A 20
rg -n "fn payload_optional_string" --type rust -A 20

Length of output: 2309

🏁 Script executed:

#!/bin/bash
# Check the rollback assertion in the new test
rg -n "playlist_insert_with_wrong_type" --type rust -A 35

Length of output: 2149

@InstaZDLL : Tout est propre — les trois points sont correctement traités.

payload_optional_string : l'implémentation finale est exactement ce qu'il fallait — la branche other produit Err(InvalidPayload) avec le kind JSON dans le message, les deux cas None légitimes (absent / null) restent inchangés, et les ? sont propagés sur les 8 sites d'appel.
Test de régression : le double assert (HTTP 5xx + sync_op count = 0) épingle à la fois le rejet et le rollback transactionnel — c'est la couverture correcte pour ce cas.
Clippy : les deux corrections (#[allow] avec commentaire, std::slice::from_ref) sont sans reproche.

LGTM.

(╯°□°）╯ 🐇 ✓

🧠 Learnings used

Learnt from: CR
Repo: InstaZDLL/waveflow-server PR: 0
File: CLAUDE.md:0-0
Timestamp: 2026-05-31T13:47:30.071Z
Learning: Implement sync (`Phase 1.f`) with an append-only `sync_op` log, three REST routes + one WebSocket under `/api/v1/sync/*`: `POST /ops` (idempotent replay via `ON CONFLICT operation_id DO NOTHING`, 409 on lamport regression), `GET /ops?since=N` (410 + watermark when stale; `since=0` is bootstrap and always skips the guard), `POST /ack` (only path updating `device_sync_cursor`, buffered + flushed every 5 s), `GET /ws?device_id=…` WebSocket (broadcast + per-frame user filtering). Live state in `sync::SyncHub` with tokio broadcast channel + DashMap per `(user_id, device_id)` + daily compaction task. Tests use `SyncHub::for_tests` and drive `flush_acks` / `compact_once` for determinism

InstaZDLL · 2026-06-04T21:09:02Z

@coderabbitai review

coderabbitai · 2026-06-04T21:09:07Z

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

style(apply): cargo fmt

7d86ff6

Signed-off-by: InstaZDLL <github.105mh@8shield.net>

InstaZDLL marked this pull request as ready for review June 4, 2026 20:37

coderabbitai Bot requested changes Jun 4, 2026

View reviewed changes

Comment thread src/apply.rs Outdated

Comment thread tests/apply.rs

Comment thread tests/apply.rs Outdated

InstaZDLL self-assigned this Jun 4, 2026

coderabbitai Bot approved these changes Jun 4, 2026

View reviewed changes

InstaZDLL merged commit f4215df into main Jun 4, 2026
8 checks passed

InstaZDLL deleted the feat/1-g-0-apply-pipeline branch June 4, 2026 21:12

InstaZDLL mentioned this pull request Jun 4, 2026

feat(share): mint + revoke by canonical id (phase 1.g.1b) #27

Merged

4 tasks

This was referenced Jun 6, 2026

feat(playlist_track): server-side materialisation + share preview (phase 1.j.a) #32

Merged

feat(track-sync): apply pipeline for the track entity (phase 4.d.0.2) #35

Merged

Navigation Menu

feat(apply): server-side materialisation of desktop sync ops (phase 1.g.0) - #26

feat(apply): server-side materialisation of desktop sync ops (phase 1.g.0)#26
InstaZDLL merged 3 commits into
mainfrom
feat/1-g-0-apply-pipeline

InstaZDLL commented Jun 4, 2026 •

edited by coderabbitai Bot

Loading

Uh oh!

coderabbitai Bot commented Jun 4, 2026 •

edited

Loading

Walkthrough

Changes

Sequence Diagrams

Estimated Code Review Effort

Possibly Related PRs

Poem

Uh oh!

coderabbitai Bot left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

InstaZDLL commented Jun 4, 2026

Uh oh!

coderabbitai Bot commented Jun 4, 2026

Uh oh!

InstaZDLL commented Jun 4, 2026

Uh oh!

coderabbitai Bot commented Jun 4, 2026 •

edited

Loading

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Conversation

InstaZDLL commented Jun 4, 2026 • edited by coderabbitai Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

What lands

Migration `20260604000000_apply_pipeline.sql`

Apply pipeline (`src/apply.rs`)

Tests (`tests/apply.rs`)

Wire shape change

Test plan

Follow-ups (deferred to keep this PR reviewable)

Summary by CodeRabbit

Uh oh!

coderabbitai Bot commented Jun 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Walkthrough

Changes

Sequence Diagrams

Estimated Code Review Effort

Possibly Related PRs

Poem

Uh oh!

coderabbitai Bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

InstaZDLL commented Jun 4, 2026

Uh oh!

coderabbitai Bot commented Jun 4, 2026

Uh oh!

InstaZDLL commented Jun 4, 2026

Uh oh!

coderabbitai Bot commented Jun 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

InstaZDLL commented Jun 4, 2026 •

edited by coderabbitai Bot

Loading

coderabbitai Bot commented Jun 4, 2026 •

edited

Loading

coderabbitai Bot commented Jun 4, 2026 •

edited

Loading