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

[InstaZDLL]

Server-side end-to-end of the multi-device sync protocol per RFC-001 §6.6. Closes the server half of Phase 1.f #133; the desktop integration (settings toggle, repository swap, Lamport clock, pending-ops queue, login flow) lands in a follow-up against the WaveFlow desktop repo.

Summary

• New sync_op BIGSERIAL log with double UNIQUE on (user_id, device_id, operation_id) (idempotency) and (user_id, device_id, lamport_ts) (monotonicity). device_sync_cursor tracks per-device read position; sync_compaction_watermark is the floor below which a pull returns 410 Gone. All three tables ON DELETE CASCADE via users(id).
• Three REST routes + one WebSocket under /api/v1/sync/*:
  • POST /ops — batch push. ON CONFLICT (user_id, device_id, operation_id) DO NOTHING absorbs idempotent replays; a (user_id, device_id, lamport_ts) unique violation surfaces as 409 + stored_max so a stale client can resync its clock. All-or-nothing transaction; broadcast fires only after commit so subscribers never see a rolled-back op.
  • GET /ops?since=N — pull. Read-only by contract — never advances the device cursor (that's POST /ack's sole job). 410 Gone + compacted_up_to when since > 0 AND since < watermark; since == 0 always allowed so fresh devices bootstrap cleanly.
  • POST /ack — the only path that writes device_sync_cursor. Buffered in memory + flushed every 5 s with a monotonic UPSERT guard (WHERE last_seen_id < EXCLUDED).
  • GET /ws?device_id=… — subscribes to a tokio broadcast::Sender. Per-frame user_id equality filter so cross-tenant ops never leave the socket. Pre-serialised envelope so N subscribers share one Arc<String>. On-disconnect flush so a tab close doesn't strand the last ACK in memory.
• sync::SyncHub holds the broadcast Sender + DashMap<(user_id, device_id), AckEntry> + handles. Live boot spawns the 5 s flusher + daily compaction task in main.rs. SyncHub::for_tests skips the loops so compaction tests step through flush_acks / compact_once deterministically.
• Compaction job: phase 1 flushes pending ACKs (so the MIN sees the latest device positions — closes the "stale Postgres cursor" race the RFC flags); phase 2 reads MIN excluding devices with last_seen_at < now - 90 d; phase 3 collapses superseded ops per (entity, entity_id, COALESCE(field, '')) at id <= MIN AND UPSERTs the watermark in the same Postgres transaction so a crash mid-compaction is atomic. Watermark monotonic by UPSERT WHERE clause.
• Wires AppState.sync everywhere it needs to land (handlers, WS, tests). Test harness gains spawn_app_with_sync so sync tests inject SyncHub::for_tests while the legacy helpers keep wiring a live hub (50 ms flush cadence) so the ack-after-flush path stays exercised in REST tests.

Test coverage

10 integration tests in tests/sync.rs:

1. push_then_pull_round_trip — POST → GET ?since=0 returns the row + correct last_id; empty pull at head.
2. idempotent_replay_returns_same_id — same operation_id twice yields the same id, one physical row.
3. lamport_regression_returns_409_with_stored_max — replay at a taken lamport_ts → 409 + stored_max + offending_lamport_ts.
4. pull_does_not_advance_cursor — GET never touches device_sync_cursor.
5. ack_writes_cursor_after_flush — ACK + 5 s flush UPSERTs; re-ACK at lower id refused (monotonic).
6. resurrected_device_returns_410 — planted watermark; since < watermark → 410, since == 0 still 200.
7. tenant_isolation_pull — user B's pull never sees user A's ops.
8. oversized_batch_is_rejected — > 1024 ops → 400.
9. compaction_collapses_old_ops_and_advances_watermark — three SETs over same field collapse to one; unrelated entity untouched; watermark = head_id; second pass is a no-op.
10. compaction_skips_stale_devices — device with last_seen_at = -200 d excluded from MIN, so the up-to-date device's MIN is used.
11. websocket_fans_out_to_same_tenant — WS subscriber sees a frame from a different device under the same user.
12. websocket_isolates_other_tenants — user B's WS does NOT receive user A's push.

Plus tests/openapi.rs updated so a future refactor that drops #[utoipa::path] from a sync handler trips the spec assertion.

Test plan

• CI green on Ubuntu (Postgres service container runs the full suite).
• CI green on Windows (compile + clippy gate only).
• DCO sign-off present.
• OpenAPI spec exposes /api/v1/sync/ops + /api/v1/sync/ack.

Out of scope

• Desktop integration (settings "Server mode" toggle, runtime repo swap, Lamport clock persistence, pending-ops queue, login + JWT in OS keyring) — lands in the WaveFlow repo.
• LISTEN/NOTIFY-backed broadcast across multiple server replicas — the current tokio::sync::broadcast is single-process. Multi-replica fan-out is a Phase 2 / production-hardening concern.
• WS pings / heartbeats — relying on the OS-level TCP timeout for now; tighten when we observe real-world stale connections.

Summary by CodeRabbit

• Nouvelles Fonctionnalités
  • Synchronisation multi‑appareils en temps réel via REST et WebSocket (/api/v1/sync/*) : push, pull, ack et diffusion live entre appareils du même utilisateur.
  • Compaction automatique avec watermark par utilisateur pour purger l’historique obsolète.
• Comportement
  • Opérations idempotentes avec horodatage logique (Lamport) ; régression Lamport → 409.
  • ACKs persistés périodiquement ; pulls n’avancent pas le curseur. Pulls sous watermark → 410 (sauf bootstrap since=0).
• Tests
  • Tests end‑to‑end couvrant push/pull/idempotence, Lamport, compaction et WebSocket.

[coderabbitai]

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: 001d165c-0a12-4ce1-9bd4-a9a1ea105de6

📥 Commits

Reviewing files that changed from the base of the PR and between 31e332c and 7dab9ac.

📒 Files selected for processing (1)

• CLAUDE.md

---

📝 Walkthrough

Walkthrough

Ajout complet d'une API de synchronisation multi‑appareils (Phase 1.f) : schéma SQL append‑only, SyncHub (broadcast + ACK buffering), flush/compaction périodiques, handlers REST/WS pour push/pull/ack, et tests E2E couvrant les scénarios critiques.

Changes

Synchronisation Multi‑Appareils (Phase 1.f)

Couche / Fichiers	Résumé
Schéma et dépendances migrations/20260601000000_sync_ops.sql, Cargo.toml	Création des tables sync_op, device_sync_cursor, sync_compaction_watermark. Ajustements Cargo pour WebSocket (axum ws), sqlx tls-rustls, tokio sync, et dépendances de test WebSocket.
SyncHub : modèles & cœur src/sync.rs	Types SyncOpIn/SyncOp/SyncBroadcast, SyncHub clonable, subscribe/broadcast, agrégat ACK atomique et API for_tests/pool.
Flush ACKs & compaction src/sync.rs	flush_acks (UPSERT monotone), boucles run_flush_loop/run_compaction_loop, compact_once (flush préalable, suppression d'ops supersedées, mise à jour monotone du watermark).
Handlers REST & WebSocket src/api/sync.rs	POST /ops (idempotence, 409 lamport_regression), GET /ops (pagination, resurrect guard 410), POST /ack (buffer), GET /ws (fan‑out, ack frames, flush on disconnect), row_to_op.
Module DB src/db.rs	insert_op_returning, fetch_op_by_operation_id, lamport_max, fetch_compacted_up_to, pull_ops_since.
Wiring AppState & démarrage src/lib.rs, src/main.rs, src/api/mod.rs	Export sync module, ajout AppState.sync: SyncHub, initialisation SyncHub::spawn dans main (conserver handles), et enregistrement du router sync avec auth layer.
Infrastructure & tests tests/support.rs, tests/openapi.rs, tests/sync.rs	Helpers spawn adaptés pour SyncHub, assertion OpenAPI pour /api/v1/sync/ops et /api/v1/sync/ack, et suite E2E couvrant push/pull/idempotence/lamport/ack/compaction/WS.
Documentation CLAUDE.md	Section Phase 1.f décrivant AppState.sync, API REST/WS, sémantique idempotente/monotone, ACK buffering et compaction.

Sequence Diagram(s)

sequenceDiagram
  participant Client
  participant API as "POST /api/v1/sync/ops"
  participant Hub as SyncHub
  participant DB as PostgreSQL
  participant WS as "WebSocket subscribers"
  Client->>API: POST batch ops
  API->>DB: INSERT op(s) transaction
  DB-->>API: RETURN inserted rows or CONFLICT
  API->>Hub: broadcast(newly_inserted ops)
  Hub->>WS: send pre-serialized frames to subscribers
  Note right of Hub: record_ack updates in-memory accumulator
  Hub->>DB: flush_acks (periodic UPSERT device_sync_cursor)
  Hub->>DB: compact_once (delete superseded ops & update watermark)

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related issues

• Phase 1.f — Sync ops protocol + WebSocket + desktop integration WaveFlow#133 : Ce PR implémente le protocole Phase 1.f (schéma, endpoints POST/GET/ACK/WS, SyncHub, compaction) décrit par l'issue.

Possibly related PRs

• InstaZDLL/waveflow-server#5 : chevauchement sur la configuration Cargo (sqlx tls-rustls).

Poem

🌊 Sync en marche, Lamport trace la voie,
Les ACK s'accumulent puis s'écrivent sans émoi,
Le hub chuchote "nouveau" aux sockets du même tenant,
La compaction balaie l'ancien, le watermark reste constant,
Tout avance monotone, sûr et persistant.

🚥 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 résume fidèlement la contribution principale : implémentation d'un journal append-only pour sync, broadcast WebSocket et compaction (Phase 1.f du RFC-001).
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 unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/1-f-sync-server

---

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

[coderabbitai]

Actionable comments posted: 7

🤖 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 `@CLAUDE.md`:
- Line 45: The documentation is inconsistent: AppState is documented as
"currently just the PgPool" but Phase 1.f introduces a live sync hub; update the
AppState description to include the new sync field by documenting that AppState
now holds sync: SyncHub (or a reference/wrapper to the SyncHub), briefly
describe its role (broadcast sender + DashMap of AckEntry + compaction/flush
responsibilities) and note any test-only construction (e.g., SyncHub::for_tests)
so contributors know how to wire it and instantiate AppState in tests.

In `@src/api/sync.rs`:
- Around line 195-213: The handler contains inline SQL (the INSERT that sets
insert_res in src/api/sync.rs) which violates the guideline—move this SQL into a
DB/repository function (e.g., create a new function insert_sync_op or
create_sync_operation in src/db.rs or a new src/db/sync.rs or inside
waveflow-core::repository::postgres::*), implement the same sqlx::query(...) and
.bind(...) logic there (accepting the same args: user_id, device_id,
op_in.operation_id, op_in.lamport_ts, entity, entity_id, op_in.field.as_deref(),
op_kind, op_in.payload.as_ref(), now and a &mut Tx/Pool as needed) and return
the same result type (Option/Result of the returned row). Then replace the
inline query in the handler (where insert_res is set) with a call to that new DB
function, passing the transaction/context; do the same refactor for the other
inline queries referenced in this file so handlers remain free of SQL.
- Around line 329-335: The DB error is being swallowed by `.unwrap_or(None)`
when fetching `compacted_up_to` which can incorrectly treat connection/timeouts
as "no watermark"; instead remove the `.unwrap_or(None)` and propagate the error
from `sqlx::query_scalar(...).bind(user_id).fetch_optional(pool).await` (e.g.,
use the `?` operator or map the error into the function's error type) so that
failures when reading `sync_compaction_watermark` surface to the caller; update
the code paths that use the `compacted` variable accordingly to handle the
Result/propagated error.

In `@src/main.rs`:
- Around line 83-97: The comment above the SyncHub::spawn call is inaccurate: it
claims "we don't store the handles" while the code captures them as _flush_task
and _compaction_task; update the comment to reflect that we keep the JoinHandles
in scope (stored in variables with leading underscores to avoid unused warnings)
so they remain alive for the runtime, or remove the mention entirely and state
that SyncHub::spawn returns the hub and two JoinHandles which are intentionally
retained as _flush_task and _compaction_task to keep tasks running.

In `@src/sync.rs`:
- Around line 314-315: The code currently increments the `applied` counter by 1
inside the `match res { Ok(_) => applied += 1, ... }` block, which counts UPSERT
executions rather than actual modified rows; change the logic to extract the
number of affected rows from the execution result (e.g., call `rows_affected()`
or the equivalent method on the DB result returned into `res`) and add that
value to `applied` instead of adding 1, ensuring you handle the `Ok(result)` arm
where you read `result.rows_affected()` and convert to the same integer type as
`applied`.

In `@tests/sync.rs`:
- Around line 227-236: The test uses a fragile tokio::time::sleep to assert no
cursor row is written after spawn_authenticated; replace the
background-task-based setup with the deterministic test harness: use
spawn_app_with_sync (or the app variant that accepts a SyncHub) and create a
SyncHub via SyncHub::for_tests, then call hub.flush_acks().await (and/or
compact_once as needed) before running the COUNT(*) query so the assertion that
GET /sync/ops never advances the cursor is deterministic; update references in
the test from spawn_authenticated to spawn_app_with_sync and ensure you call
SyncHub::for_tests and flush_acks() prior to the SQL check.
- Around line 239-312: The test ack_writes_cursor_after_flush is
timing-dependent: replace the live/background flusher usage (spawn_authenticated
/ implicit flusher) and the polling loop with a deterministic SyncHub test
harness; create a SyncHub via SyncHub::for_tests and start the app with
spawn_app_with_sync (instead of spawn_authenticated), perform the push and ACK
as before, then call sync.flush_acks().await to force the ack flush, and replace
the polling/fetch_optional loop with a single sqlx fetch_one (query_as or
query_scalar) bound to auth.user_id to assert last_seen == head_id; keep
existing checks that re-ACKing with a lower id doesn't regress.

🪄 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: 5c015c38-6db8-4c32-9773-7f1feb0151d1

📥 Commits

Reviewing files that changed from the base of the PR and between 5c126a3 and 911fcbb.

⛔ Files ignored due to path filters (1)

• Cargo.lock is excluded by !**/*.lock, !Cargo.lock, !*.lock

📒 Files selected for processing (11)

• CLAUDE.md
• Cargo.toml
• migrations/20260601000000_sync_ops.sql
• src/api/mod.rs
• src/api/sync.rs
• src/lib.rs
• src/main.rs
• src/sync.rs
• tests/openapi.rs
• tests/support.rs
• tests/sync.rs

[coderabbitai]

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

CLAUDE.md (1)

45-45: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Clarifier l'exception since=0 pour le bootstrap.

La doc indique « 410 + compacted_up_to when since < watermark » mais ne précise pas que since=0 est permis pour le bootstrap initial (mentionné dans les objectifs du PR). Un développeur pourrait croire que since=0 déclenche systématiquement un 410 si le watermark > 0.

📝 Clarification suggérée

-GET /ops?since=N` pull (410 + `compacted_up_to` when `since < watermark`, read-only — never advances the cursor)
+GET /ops?since=N` pull (410 + `compacted_up_to` when `0 < since < watermark`; `since=0` allowed for bootstrap, read-only — never advances the cursor)

🤖 Prompt for 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.

In `@CLAUDE.md` at line 45, The docs for GET /ops?since=N are ambiguous about
since=0; update the sentence describing "410 + `compacted_up_to` when `since <
watermark`" to explicitly state that since=0 is permitted for initial bootstrap
and must not be treated as an automatic 410 even when `watermark > 0` — clarify
the server behavior for `since=0` (allowed for bootstrap initial sync, how
`compacted_up_to` is reported) so readers of the GET /ops?since=N,
`compacted_up_to`, and `watermark` behavior understand that `since=0` is a valid
bootstrap request.

🤖 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.

Outside diff comments:
In `@CLAUDE.md`:
- Line 45: The docs for GET /ops?since=N are ambiguous about since=0; update the
sentence describing "410 + `compacted_up_to` when `since < watermark`" to
explicitly state that since=0 is permitted for initial bootstrap and must not be
treated as an automatic 410 even when `watermark > 0` — clarify the server
behavior for `since=0` (allowed for bootstrap initial sync, how
`compacted_up_to` is reported) so readers of the GET /ops?since=N,
`compacted_up_to`, and `watermark` behavior understand that `since=0` is a valid
bootstrap request.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: 74589e67-904c-48f0-930a-43cc4bb0114f

📥 Commits

Reviewing files that changed from the base of the PR and between 911fcbb and 31e332c.

📒 Files selected for processing (6)

• CLAUDE.md
• src/api/sync.rs
• src/db.rs
• src/main.rs
• src/sync.rs
• tests/sync.rs