refactor: Move the core implementation to the shared package with no manual queries (#107)

packages/serverpod_offline_sync/lib/serverpod_offline_sync.dart

@@ -1,3 +1,16 @@
+export 'src/crdt/client_sync.dart';
+export 'src/crdt/exceptions.dart' hide PendingOutboundIntegrityViolation;
+export 'src/crdt/extensions.dart';
+export 'src/crdt/merge.dart';
+export 'src/crdt/scope_membership.dart';
+export 'src/crdt/sync.dart';
+export 'src/database/database.dart';
+export 'src/database/recorder.dart';
+export 'src/database/schema.dart';
+export 'src/database/session.dart' show CrdtDatabaseAccess, CrdtDatabaseSession;
+export 'src/database/tombstone.dart' show IncludeTombstonedRows;
 export 'src/generated/protocol.dart';
 export 'src/hlc/exceptions.dart';
 export 'src/hlc/hlc.dart';
+export 'src/managers/hlc.dart';
+export 'src/managers/scope.dart';

packages/serverpod_offline_sync/lib/src/crdt/client_sync.dart

@@ -0,0 +1,147 @@
+import 'dart:async';
+
+import 'package:serverpod_database/serverpod_database.dart';
+
+import '../database/session.dart';
+import '../generated/protocol.dart';
+import 'sync.dart';
+
+/// Opens the remote half of a CRDT sync stream.
+typedef CrdtSyncTransport =
+    Stream<CrdtSyncStreamEvent> Function({
+      required Stream<CrdtSyncStreamEvent> changes,
+      required bool once,
+    });
+
+/// A running continuous CRDT sync session.
+class CrdtSyncSession {
+  CrdtSyncSession._({
+    required this.done,
+    required Future<void> Function() cancel,
+  }) : _cancel = cancel;
+
+  /// Completes when the sync session ends normally or after [cancel].
+  final Future<void> done;
+
+  final Future<void> Function() _cancel;
+
+  /// Stops the sync session and closes the transport. This method is idempotent
+  /// and can be called multiple times.
+  Future<void> cancel() => _cancel();
+}
+
+/// High-level client sync helpers built on top of a CRDT stream transport.
+class CrdtSyncClient {
+  /// Creates a [CrdtSyncClient] for the provided [transport].
+  CrdtSyncClient(CrdtSyncTransport transport) : _transport = transport;
+
+  final CrdtSyncTransport _transport;
+
+  /// Pushes local pending changes for [session] to the remote peer once.
+  ///
+  /// The [session] must be wrapped in a [CrdtDatabaseSession].
+  Future<void> syncOnce(
+    DatabaseSession session, {
+    CrdtSyncOnMergeSuccess? onMergeSuccess,
+  }) async {
+    await _startSyncSession(session, once: true, onMergeSuccess: onMergeSuccess).done;
+  }
+
+  /// Keeps synchronizing [session] with the remote peer until cancelled through
+  /// [CrdtSyncSession.cancel] or the remote stream closes.
+  ///
+  /// The [session] must be wrapped in a [CrdtDatabaseSession].
+  CrdtSyncSession syncContinuously(
+    DatabaseSession session, {
+    CrdtSyncOnMergeSuccess? onMergeSuccess,
+  }) => _startSyncSession(session, once: false, onMergeSuccess: onMergeSuccess);
+
+  CrdtSyncSession _startSyncSession(
+    DatabaseSession session, {
+    required bool once,
+    CrdtSyncOnMergeSuccess? onMergeSuccess,
+  }) {
+    // The stream will be closed using [outboundChanges.closeOrSkip].
+    // ignore: close_sinks
+    final outboundChanges = StreamController<CrdtSyncStreamEvent>();
+    final doneCompleter = Completer<void>();
+    var cancelled = false;
+
+    try {
+      final remoteStream = _transport(
+        changes: outboundChanges.stream,
+        once: once,
+      );
+
+      final subscription = session.crdtDb
+          .sync(
+            inbound: remoteStream,
+            once: once,
+            onMergeSuccess: onMergeSuccess,
+            mode: CrdtSyncPeerMode.follower,
+          )
+          .listen(
+            (event) {
+              outboundChanges.addIfNotClosed(event);
+              if (once && event is CrdtSyncClose) {
+                unawaited(outboundChanges.closeOrSkip());
+              }
+            },
+            onDone: doneCompleter.completeOrSkip,
+            onError: doneCompleter.completeErrorOrSkip,
+            cancelOnError: true,
+          );
+
+      return CrdtSyncSession._(
+        done: doneCompleter.future.whenComplete(outboundChanges.closeOrSkip),
+        cancel: () async {
+          if (cancelled) return;
+          cancelled = true;
+
+          unawaited(outboundChanges.closeOrSkip());
+          if (doneCompleter.isCompleted) return;
+          const waitTimeout = Duration(milliseconds: 200);
+          // Prefer graceful shutdown: closing outbound lets the server finish
+          // and complete [done] via [subscription.onDone].
+          await doneCompleter.future.timeout(waitTimeout, onTimeout: () {});
+          await subscription.cancel().timeout(waitTimeout, onTimeout: () {});
+          doneCompleter.completeOrSkip();
+        },
+      );
+    } catch (_) {
+      unawaited(outboundChanges.closeOrSkip());
+      rethrow;
+    }
+  }
+}
+
+extension on StreamController<CrdtSyncStreamEvent> {
+  void addIfNotClosed(CrdtSyncStreamEvent event) {
+    if (isClosed) return;
+    add(event);
+  }
+
+  Future<void> closeOrSkip() async {
+    if (!isClosed) {
+      unawaited(close());
+    }
+    // The done event is only delivered once the stream has a listener. If the
+    // method stream never listened (e.g. opening it failed), awaiting [done]
+    // would hang forever.
+    if (hasListener) {
+      await done;
+    }
+  }
+}
+
+extension on Completer<void> {
+  void completeOrSkip() {
+    if (isCompleted) return;
+    complete();
+  }
+
+  void completeErrorOrSkip(Object error, StackTrace stackTrace) {
+    if (isCompleted) return;
+    completeError(error, stackTrace);
+  }
+}

packages/serverpod_offline_sync/lib/src/crdt/exceptions.dart

@@ -0,0 +1,178 @@
+import 'package:meta/meta.dart';
+import 'package:serverpod_serialization/serverpod_serialization.dart';
+
+import '../generated/protocol.dart';
+import '../hlc/hlc.dart';
+
+/// Base type for CRDT sync session failures.
+sealed class CrdtSyncException implements Exception {
+  const CrdtSyncException();
+}
+
+/// Thrown when the inbound sync stream closes before an expected event arrives.
+final class CrdtSyncStreamClosedException extends CrdtSyncException {
+  /// Creates a [CrdtSyncStreamClosedException].
+  const CrdtSyncStreamClosedException({required this.phase});
+
+  /// The protocol phase that was interrupted.
+  final String phase;
+
+  @override
+  String toString() =>
+      'CrdtSyncStreamClosedException: sync stream closed before $phase event.';
+}
+
+/// Thrown when the inbound sync stream delivers an event of the wrong type.
+final class CrdtSyncUnexpectedEventException<T extends CrdtSyncStreamEvent>
+    extends CrdtSyncException {
+  /// Creates a [CrdtSyncUnexpectedEventException].
+  const CrdtSyncUnexpectedEventException({
+    required this.expected,
+    required this.received,
+  });
+
+  /// A description of the expected event type or types.
+  final String expected;
+
+  /// The event that was received instead.
+  final CrdtSyncStreamEvent received;
+
+  @override
+  String toString() =>
+      'CrdtSyncUnexpectedEventException: expected $expected, but '
+      'received "${received.runtimeType.className}" instead.';
+}
+
+/// Thrown when the sync tables hash sent by a peer does not match locally.
+final class SyncTablesHashMismatchException extends CrdtSyncException {
+  /// Creates a [SyncTablesHashMismatchException].
+  const SyncTablesHashMismatchException({
+    required this.received,
+    required this.expected,
+  });
+
+  /// The hash received from the remote peer.
+  final String received;
+
+  /// The hash computed locally from the configured sync tables.
+  final String expected;
+
+  @override
+  String toString() =>
+      'SyncTablesHashMismatchException: schema hash mismatch. Received '
+      '"$received", expected "$expected". Ensure both sides are on the same '
+      'schema version before syncing.';
+}
+
+/// Thrown when sync observes terminal CRDT metadata/domain-row inconsistency.
+final class CrdtSyncIntegrityViolationException extends CrdtSyncException {
+  /// Creates a [CrdtSyncIntegrityViolationException].
+  const CrdtSyncIntegrityViolationException(this.violation);
+
+  /// The integrity violation that caused sync to stop.
+  final CrdtSyncIntegrityViolation violation;
+
+  @override
+  String toString() {
+    final persisted = violation.id == null
+        ? ''
+        : ' Persisted violation id: ${violation.id}.';
+    final row = '${violation.domainTableName}.${violation.uuidRowId}';
+    return switch (violation.type) {
+      CrdtSyncViolationType.ownershipCollision =>
+        'CrdtSyncIntegrityViolationException: '
+            '${violation.type.name}/${violation.operation.name} for $row '
+            'belongs to scope ${violation.ownerScopeUuid}, but sync attempted '
+            'scope ${violation.incomingScopeUuid}.$persisted',
+      CrdtSyncViolationType.missingDomainRow =>
+        'CrdtSyncIntegrityViolationException: '
+            '${violation.type.name}/${violation.operation.name} references '
+            'missing domain row $row for scope '
+            '${violation.incomingScopeUuid}.$persisted',
+      CrdtSyncViolationType.unauthorizedWrite =>
+        'CrdtSyncIntegrityViolationException: '
+            '${violation.type.name}/${violation.operation.name} rejected '
+            'write to $row in scope ${violation.incomingScopeUuid}.$persisted',
+    };
+  }
+}
+
+/// Thrown when a user attempts to act in a scope they are not a member of.
+final class CrdtScopeMembershipException implements Exception {
+  /// Creates a [CrdtScopeMembershipException].
+  const CrdtScopeMembershipException({
+    required this.userId,
+    required this.scopeId,
+  });
+
+  /// The authenticated user id supplied to `transactionForUser`.
+  final UuidValue userId;
+
+  /// The scope id the transaction attempted to act in.
+  final UuidValue scopeId;
+
+  @override
+  String toString() =>
+      'CrdtScopeMembershipException: user "$userId" is not a member of '
+      'scope "$scopeId".';
+}
+
+/// Thrown when a user attempts to write in a scope with a non-writable role.
+final class CrdtScopeRoleException implements Exception {
+  /// Creates a [CrdtScopeRoleException].
+  const CrdtScopeRoleException({
+    required this.userId,
+    required this.scopeId,
+    this.role,
+  });
+
+  /// The authenticated user id supplied to `transactionForUser`.
+  final UuidValue userId;
+
+  /// The scope id the transaction attempted to write in.
+  final UuidValue scopeId;
+
+  /// The role that does not allow writes.
+  final CrdtScopeRole? role;
+
+  @override
+  String toString() =>
+      'CrdtScopeRoleException: user "$userId" with role "$role" cannot write '
+      'in scope "$scopeId".';
+}
+
+@internal
+final class PendingOutboundIntegrityViolation implements Exception {
+  const PendingOutboundIntegrityViolation({
+    required this.crdtDataRowId,
+    required this.type,
+    required this.operation,
+    required this.tableName,
+    required this.rowId,
+    required this.ownerScopeId,
+    required this.incomingScopeUuid,
+    required this.uuidNodeId,
+    this.hlc,
+  });
+
+  final int? crdtDataRowId;
+  final CrdtSyncViolationType type;
+  final CrdtSyncViolationOperation operation;
+  final String tableName;
+  final UuidValue rowId;
+  final int? ownerScopeId;
+  final UuidValue incomingScopeUuid;
+  final UuidValue uuidNodeId;
+  final Hlc? hlc;
+}
+
+extension on Type {
+  /// The name of the class without the leading underscore and without the 'Impl' suffix.
+  String get className {
+    final name = toString();
+    if (name.startsWith('_') && name.endsWith('Impl')) {
+      return name.substring(1, name.length - 4);
+    }
+    return name;
+  }
+}

packages/serverpod_offline_sync/lib/src/crdt/extensions.dart

@@ -0,0 +1,79 @@
+import '../generated/protocol.dart';
+import '../hlc/hlc.dart';
+
+/// Last serialized index that still represents a visible CRDT row.
+///
+/// The [CrdtDataRowVisibility] is ordered by visibility, so the index is used
+/// to compare the visibility states. The last visible state in the enum is
+/// [CrdtDataRowVisibility.foreignKeyRestrictRestore].
+final crdtRowLastVisibleVisibilityIndex =
+    CrdtDataRowVisibility.foreignKeyRestrictRestore.index;
+
+/// Extensions for the [CrdtDataRowVisibility] enum.
+extension CrdtDataRowVisibilityExtension on CrdtDataRowVisibility {
+  /// Whether this visibility state hides the row from domain queries.
+  ///
+  bool get isHidden => index > crdtRowLastVisibleVisibilityIndex;
+}
+
+/// Extensions for the [CrdtDataField] class.
+extension CrdtDataFieldExtension on CrdtDataField {
+  /// The HLC represented by this field.
+  Hlc get hlc => node == null
+      ? (throw StateError(
+          'Fetch the CRDT field with the node included to get the HLC.',
+        ))
+      : toHlcForNode(node!.uuidNodeId);
+}
+
+/// Extensions for the [CrdtDataRow] class.
+extension CrdtDataRowExtension on CrdtDataRow {
+  /// Whether this row is currently hidden from domain queries.
+  bool get isHidden => visibility.isHidden;
+
+  /// The HLC represented by this row.
+  Hlc get hlc => node == null
+      ? (throw StateError(
+          'Fetch the CRDT row with the node included to get the HLC.',
+        ))
+      : toHlcForNode(node!.uuidNodeId);
+}
+
+/// Extensions for the [CrdtDataDeleted] class.
+extension CrdtDataDeletedExtension on CrdtDataDeleted {
+  /// Whether this visibility generation hides the row.
+  bool get isDeleted => clFlag.isEven;
+
+  /// The HLC represented by this visibility generation.
+  Hlc get hlc => node == null
+      ? (throw StateError(
+          'Fetch the CRDT visibility generation with the node included to get the HLC.',
+        ))
+      : toHlcForNode(node!.uuidNodeId);
+}
+
+/// Extensions for the [CrdtDataForeignKey] class.
+extension CrdtDataForeignKeyExtension on CrdtDataForeignKey {
+  /// Whether an active projection override exists for this FK field.
+  ///
+  /// Derived from [CrdtDataForeignKey.overrideReason]: an override is active if
+  /// and only if [overrideReason] is non-null. This is the authoritative test
+  /// for whether the domain row's FK value differs from the attempted value.
+  bool get hasOverride => overrideReason != null;
+}
+
+/// Extensions for the [CrdtDataDeletedReason] enum.
+extension CrdtDataDeletedReasonExtension on CrdtDataDeletedReason {
+  /// Whether this tombstone reason is synced across replicas. All reasons that
+  /// start with 'user' are synced.
+  bool get isSynced => name.startsWith('user');
+}
+
+/// CRDT write capability for a scope role.
+///
+/// The implicit personal scope resolves to [CrdtScopeRole.readWrite]. A missing
+/// role means no shared membership row was found and does not grant writes.
+extension CrdtScopeRoleWriteAccess on CrdtScopeRole? {
+  /// Whether this role may write CRDT rows in the scope.
+  bool get canWrite => this == CrdtScopeRole.readWrite;
+}

packages/serverpod_offline_sync/lib/src/crdt/integrity_violation.dart

@@ -0,0 +1,54 @@
+import 'package:serverpod_database/serverpod_database.dart';
+
+import '../generated/protocol.dart';
+
+/// Persists or updates the durable row for an integrity violation.
+Future<CrdtSyncIntegrityViolation> recordCrdtSyncIntegrityViolation(
+  DatabaseSession session, {
+  required CrdtSyncIntegrityViolation violation,
+  Transaction? transaction,
+}) async {
+  final existing = await CrdtSyncIntegrityViolation.db.findFirstRow(
+    session,
+    where: (t) =>
+        t.type.equals(violation.type) &
+        t.operation.equals(violation.operation) &
+        t.domainTableName.equals(violation.domainTableName) &
+        t.uuidRowId.equals(violation.uuidRowId) &
+        t.ownerScopeUuid.equals(violation.ownerScopeUuid) &
+        t.incomingScopeUuid.equals(violation.incomingScopeUuid),
+    transaction: transaction,
+  );
+
+  final now = DateTime.now().toUtc();
+  final newViolation = existing != null
+      ? existing.copyWith(
+          crdtDataRowId: violation.crdtDataRowId ?? existing.crdtDataRowId,
+          uuidNodeId: violation.uuidNodeId ?? existing.uuidNodeId,
+          hlcDatetime: violation.hlcDatetime,
+          hlcCounter: violation.hlcCounter,
+          lastSeenAt: now,
+          occurrences: existing.occurrences + 1,
+        )
+      : violation.copyWith(
+          firstSeenAt: now,
+          lastSeenAt: now,
+          occurrences: 1,
+        );
+
+  final persisted = await CrdtSyncIntegrityViolation.db.upsertRow(
+    session,
+    newViolation,
+    conflictColumns: (t) => [
+      t.type,
+      t.operation,
+      t.domainTableName,
+      t.uuidRowId,
+      t.ownerScopeUuid,
+      t.incomingScopeUuid,
+    ],
+    transaction: transaction,
+  );
+
+  return persisted!;
+}

packages/serverpod_offline_sync/lib/src/crdt/merge.dart

@@ -0,0 +1,191 @@
+import 'dart:async';
+
+import 'package:serverpod_database/serverpod_database.dart';
+import 'package:uuid/uuid.dart';
+
+import '../generated/protocol.dart';
+import '../hlc/hlc.dart';
+import 'exceptions.dart';
+
+/// A group of CRDT merge changes collected for one sync batch.
+typedef CrdtMergeSet = List<CrdtMergeChange>;
+
+/// Merge metadata extracted from a [CrdtMergeSet].
+typedef CrdtMergeMetadataLookup = ({
+  Map<String, Set<UuidValue>> rowIdsByTable,
+  Map<String, Set<String>> columnNamesByTable,
+});
+
+/// Extensions for stream iterators of [CrdtSyncStreamEvent].
+extension CrdtSyncStreamEventStreamExtension on StreamIterator<CrdtSyncStreamEvent> {
+  /// Collects the next framed sync batch from this iterator.
+  ///
+  /// Each batch is zero or more scope, handshake, and/or merge frames followed
+  /// by [CrdtSyncEndOfBatch]. If the stream is idle before a batch starts, an
+  /// empty batch is returned.
+  ///
+  /// When [allowCloseBeforeBatch] is true, returns `null` if the transport
+  /// stream closes before the next batch starts. [CrdtSyncClose] is an
+  /// intentional peer request to end the streaming session and returns `null`
+  /// even after frames for a partial batch have already arrived. The caller
+  /// must not persist checkpoint progress for such a discarded partial batch.
+  /// Otherwise, closing before a batch starts is treated as a
+  /// [CrdtSyncStreamClosedException].
+  ///
+  /// If the transport stream closes after a frame was already received without
+  /// a [CrdtSyncClose] control frame, the partial batch is still treated as an
+  /// error.
+  ///
+  /// An idle timeout only ends an *empty* batch: once any frame has been
+  /// collected the timeout is ignored and collection waits for the explicit
+  /// terminator.
+  Future<CrdtSyncCycleBatch?> collectNextBatch({
+    bool allowCloseBeforeBatch = false,
+  }) async {
+    final batch = CrdtSyncCycleBatch();
+
+    while (await moveNext()) {
+      switch (current) {
+        case final CrdtSyncScopeSet event:
+          batch.scopeSet = event;
+        case final CrdtSyncSinceHlc event:
+          batch.sinceHlcs[event.uuidScopeId] = event;
+        case CrdtSyncMergeChunk(:final changes):
+          batch.changes.addAll(changes);
+        case CrdtSyncIdleTimeout():
+          if (batch.isEmpty) return batch;
+        case CrdtSyncEndOfBatch():
+          return batch;
+        case CrdtSyncClose():
+          return null;
+        default:
+          throw CrdtSyncUnexpectedEventException(
+            expected: 'a sync cycle frame',
+            received: current,
+          );
+      }
+    }
+
+    if (batch.isEmpty && allowCloseBeforeBatch) return null;
+    throw const CrdtSyncStreamClosedException(phase: 'end-of-batch');
+  }
+
+  /// Moves the iterator to the next event and throws if the stream is closed or
+  /// the next event is not of type [T].
+  Future<T> moveAndThrowIfNot<T extends CrdtSyncStreamEvent>() async {
+    while (await moveNext()) {
+      if (current is CrdtSyncIdleTimeout) continue;
+      if (current is T) return current as T;
+      throw CrdtSyncUnexpectedEventException(
+        expected: '"$T"',
+        received: current,
+      );
+    }
+    throw CrdtSyncStreamClosedException(phase: '"$T"');
+  }
+}
+
+/// One sync cycle's inbound frames.
+class CrdtSyncCycleBatch {
+  /// The peer's scope announcement for this cycle, if it sent one.
+  CrdtSyncScopeSet? scopeSet;
+
+  /// The peer's resume vectors, keyed by scope.
+  final Map<UuidValue, CrdtSyncSinceHlc> sinceHlcs = {};
+
+  /// The peer's merge changes for this cycle.
+  final List<CrdtMergeChange> changes = [];
+
+  /// Whether the peer sent nothing this cycle (it was idle).
+  bool get isEmpty => scopeSet == null && sinceHlcs.isEmpty && changes.isEmpty;
+}
+
+/// Helpers for grouping merge changes into stream payload batches.
+extension CrdtMergeChangeStreamExtension on Stream<CrdtMergeChange> {
+  /// Emits lists with at most [batchSize] changes from this stream.
+  Stream<CrdtMergeSet> chunked(int batchSize) async* {
+    var batch = <CrdtMergeChange>[];
+    await for (final change in this) {
+      batch.add(change);
+      if (batch.length < batchSize) continue;
+      yield batch;
+      batch = <CrdtMergeChange>[];
+    }
+    if (batch.isNotEmpty) yield batch;
+  }
+}
+
+/// CRDT merge helpers for [CrdtMergeSet].
+extension CrdtMergeSetExtension on CrdtMergeSet {
+  /// The greatest HLC represented by the changes in this merge set.
+  Hlc? get maxHlc => fold<Hlc?>(
+    null,
+    (current, change) => change.hlc.maxBetween(current),
+  );
+
+  /// Insert changes in this set.
+  Iterable<CrdtMergeInsert> get inserts => whereType<CrdtMergeInsert>();
+
+  /// Update changes in this set.
+  Iterable<CrdtMergeUpdate> get updates => whereType<CrdtMergeUpdate>();
+
+  /// Delete changes in this set.
+  Iterable<CrdtMergeDelete> get deletes => whereType<CrdtMergeDelete>();
+
+  /// All merge changes in this set sorted by causal order.
+  List<CrdtMergeChange> get causallyOrderedChanges {
+    final operations = List<CrdtMergeChange>.from(this)
+      ..sort((left, right) => left.hlc.compareTo(right.hlc));
+    return operations;
+  }
+
+  /// Collects the row and column metadata needed to load merge state.
+  CrdtMergeMetadataLookup collectMetadataLookup({
+    required Map<String, Set<String>> columnNamesByTableName,
+  }) {
+    final rowIdsToLoadByTable = <String, Set<UuidValue>>{};
+    final columnNamesToLoadByTable = <String, Set<String>>{};
+
+    for (final change in this) {
+      final availableColumnNames = columnNamesByTableName[change.tableName];
+      if (availableColumnNames == null) continue;
+
+      rowIdsToLoadByTable.putIfAbsent(change.tableName, () => {}).add(change.uuidRowId);
+      columnNamesToLoadByTable.putIfAbsent(change.tableName, () => {}).addAll(
+        switch (change) {
+          CrdtMergeInsert() => availableColumnNames.difference({'id'}),
+          CrdtMergeUpdate() => [change.columnName],
+          _ => [],
+        },
+      );
+    }
+
+    return (
+      rowIdsByTable: rowIdsToLoadByTable,
+      columnNamesByTable: columnNamesToLoadByTable,
+    );
+  }
+}
+
+/// Convenience helpers shared by every merge change.
+extension CrdtMergeChangeExtension on CrdtMergeChange {
+  /// The HLC represented by this change.
+  Hlc get hlc => Hlc(hlcDatetime, hlcCounter, uuidNodeId);
+}
+
+/// Convenience helpers for insert changes.
+extension CrdtMergeInsertExtension on CrdtMergeInsert {
+  /// The database column payload represented by this change.
+  Map<String, Object?> get databaseColumns {
+    final payload = switch (data) {
+      final TableRow row => row.toJsonForDatabase(),
+      final Map<String, dynamic> map => map,
+      _ => throw StateError(
+        'Unsupported merge insert payload type for $tableName: '
+        '${data.runtimeType}. Expected TableRow or Map<String, dynamic>.',
+      ),
+    };
+
+    return Map<String, Object?>.from(payload as Map<String, dynamic>);
+  }
+}

packages/serverpod_offline_sync/lib/src/crdt/scope_membership.dart

@@ -0,0 +1,184 @@
+import 'package:meta/meta.dart';
+import 'package:serverpod_database/serverpod_database.dart' hide Protocol;
+import 'package:uuid/uuid.dart';
+
+import '../generated/protocol.dart';
+
+/// Resolves shared-scope membership from the `crdt_scope_members` table.
+///
+/// The table is `database: all`, so the same code runs on every node:
+/// authoritatively on the server (the source of truth) and against the local
+/// read-only cache on a follower. A user's personal scope is implicit — a user
+/// always belongs to the scope whose UUID equals their own user UUID — and
+/// needs no membership row. Explicit rows grant access to additional shared
+/// scopes.
+class CrdtScopeMembership {
+  const CrdtScopeMembership._();
+
+  /// Returns the scope UUIDs [userUuid] may sync.
+  ///
+  /// Includes the implicit personal scope and every explicit shared membership,
+  /// de-duplicated and sorted by UUID string so both peers iterate
+  /// deterministically.
+  static Future<List<UuidValue>> memberScopes(
+    DatabaseSession session,
+    UuidValue userUuid, {
+    Transaction? transaction,
+  }) async {
+    final grants = await memberGrants(
+      session,
+      userUuid,
+      transaction: transaction,
+    );
+    return [for (final grant in grants) grant.uuidScopeId];
+  }
+
+  /// Returns the authoritative scope grants for [userUuid].
+  ///
+  /// Like [memberScopes] but carries each scope's `role`. The implicit
+  /// personal scope is included with [CrdtScopeRole.readWrite]. Used to build the
+  /// authoritative [CrdtSyncScopeSet] announcement so a follower can project
+  /// roles into its local cache.
+  static Future<List<CrdtScopeGrant>> memberGrants(
+    DatabaseSession session,
+    UuidValue userUuid, {
+    Transaction? transaction,
+  }) async {
+    final memberships = await CrdtScopeMember.db.find(
+      session,
+      where: (t) => t.userUuid.equals(userUuid),
+      transaction: transaction,
+      include: CrdtScopeMember.include(scope: CrdtScope.include()),
+    );
+
+    final grantsByUuid = <String, CrdtScopeGrant>{};
+    for (final membership in memberships) {
+      final scopeUuid = membership.scope!.uuidScopeId;
+      grantsByUuid[scopeUuid.uuid] = CrdtScopeGrant(
+        uuidScopeId: scopeUuid,
+        role: membership.role,
+      );
+    }
+    // Assigned after the shared rows so the personal grant stays readWrite
+    // even if a stray membership row exists for it, matching [roleOf].
+    grantsByUuid[userUuid.uuid] = CrdtScopeGrant(
+      uuidScopeId: userUuid,
+      role: CrdtScopeRole.readWrite,
+    );
+
+    return grantsByUuid.values.toList()
+      ..sort((a, b) => a.uuidScopeId.uuid.compareTo(b.uuidScopeId.uuid));
+  }
+
+  /// The role [userUuid] holds in [scopeUuid], or null if none is recorded.
+  ///
+  /// The implicit personal scope resolves to [CrdtScopeRole.readWrite]. On the
+  /// server this reads authoritative membership; on a client it reads the
+  /// projected membership cache.
+  static Future<CrdtScopeRole?> roleOf(
+    DatabaseSession session, {
+    required UuidValue userUuid,
+    required UuidValue scopeUuid,
+    Transaction? transaction,
+  }) async {
+    if (userUuid == scopeUuid) return CrdtScopeRole.readWrite;
+
+    final scopeId = await _scopeIdForUuid(
+      session,
+      scopeUuid,
+      transaction: transaction,
+    );
+    if (scopeId == null) return null;
+
+    final membership = await CrdtScopeMember.db.findFirstRow(
+      session,
+      where: (t) => t.userUuid.equals(userUuid) & t.scopeId.equals(scopeId),
+      transaction: transaction,
+    );
+    return membership?.role;
+  }
+
+  /// Reconciles the local `crdt_scope_members` cache from an authoritative
+  /// [grants] announcement — a follower's read-only projection of its own
+  /// memberships.
+  ///
+  /// Upserts the materialized shared grants (the implicit personal scope is
+  /// skipped) and deletes rows for [userUuid] whose scope is no longer granted,
+  /// so a revoked or demoted membership does not linger offline. Scopes must
+  /// already be materialized locally; a grant whose scope is unknown is skipped.
+  @internal
+  static Future<void> projectFollowerMembership(
+    DatabaseSession session, {
+    required UuidValue userUuid,
+    required List<CrdtScopeGrant> grants,
+  }) async {
+    final keptScopeIds = <int>[];
+    final memberships = <CrdtScopeMember>[];
+
+    for (final grant in grants) {
+      if (grant.uuidScopeId == userUuid) continue; // personal scope is implicit
+      final scopeId = await _scopeIdForUuid(session, grant.uuidScopeId);
+      if (scopeId == null) continue; // not materialized locally yet
+
+      keptScopeIds.add(scopeId);
+      memberships.add(
+        CrdtScopeMember(
+          scopeId: scopeId,
+          userUuid: userUuid,
+          role: grant.role,
+        ),
+      );
+    }
+
+    if (memberships.isNotEmpty) {
+      await CrdtScopeMember.db.upsert(
+        session,
+        memberships,
+        conflictColumns: (t) => [t.userUuid, t.scopeId],
+        updateColumns: (t) => [t.role],
+        noReturn: true,
+      );
+    }
+
+    await CrdtScopeMember.db.deleteWhere(
+      session,
+      where: (t) =>
+          t.userUuid.equals(userUuid) &
+          (keptScopeIds.isEmpty
+              ? Constant.bool(true)
+              : t.scopeId.notInSet(keptScopeIds.toSet())),
+      noReturn: true,
+    );
+  }
+
+  /// Returns whether [userUuid] may act in [scopeUuid].
+  ///
+  /// The implicit personal scope is accepted without a membership row.
+  static Future<bool> isMember(
+    DatabaseSession session, {
+    required UuidValue userUuid,
+    required UuidValue scopeUuid,
+    Transaction? transaction,
+  }) async {
+    final role = await roleOf(
+      session,
+      userUuid: userUuid,
+      scopeUuid: scopeUuid,
+      transaction: transaction,
+    );
+    return role != null;
+  }
+}
+
+Future<int?> _scopeIdForUuid(
+  DatabaseSession session,
+  UuidValue scopeUuid, {
+  Transaction? transaction,
+}) async {
+  final scope = await CrdtScope.db.findFirstRow(
+    session,
+    where: (t) => t.uuidScopeId.equals(scopeUuid),
+    transaction: transaction,
+  );
+  return scope?.id;
+}

packages/serverpod_offline_sync/lib/src/crdt/scope_sync.dart

@@ -0,0 +1,273 @@
+import 'package:serverpod_database/serverpod_database.dart';
+import 'package:uuid/uuid.dart';
+
+import '../generated/protocol.dart';
+import '../hlc/hlc.dart';
+import '../managers/scope.dart';
+import 'extensions.dart';
+import 'merge.dart';
+import 'scope_membership.dart';
+
+/// How a peer decides which scopes it syncs.
+enum CrdtSyncPeerMode {
+  /// Dictates the scope set from authoritative membership.
+  authoritative,
+
+  /// Adopts the scope set announced by the authoritative peer.
+  follower,
+}
+
+/// De-duplicates [scopeIds] and sorts them by UUID string so both peers iterate
+/// scopes in the same deterministic order.
+List<UuidValue> _sortedUniqueScopeIds(Iterable<UuidValue> scopeIds) {
+  final byUuid = {for (final scopeId in scopeIds) scopeId.uuid: scopeId};
+  return byUuid.values.toList()..sort((a, b) => a.uuid.compareTo(b.uuid));
+}
+
+/// De-duplicates [grants] by scope and sorts them by scope UUID string.
+List<CrdtScopeGrant> _sortedUniqueGrants(Iterable<CrdtScopeGrant> grants) {
+  final byUuid = {for (final grant in grants) grant.uuidScopeId.uuid: grant};
+  return byUuid.values.toList()
+    ..sort((a, b) => a.uuidScopeId.uuid.compareTo(b.uuidScopeId.uuid));
+}
+
+/// Whether the grant list [a] last announced equals the current list [b].
+bool _grantsEqual(List<CrdtScopeGrant>? a, List<CrdtScopeGrant> b) {
+  if (a == null || a.length != b.length) return false;
+  for (var i = 0; i < b.length; i++) {
+    if (a[i].uuidScopeId != b[i].uuidScopeId || a[i].role != b[i].role) {
+      return false;
+    }
+  }
+  return true;
+}
+
+/// Owns the scope state of one sync session and every scope decision its driver
+/// (`CrdtSync.sync`) makes: which scopes are active, their handshake and
+/// checkpoint state, membership reconciliation, and per-frame authorization.
+///
+/// The driver owns the wire (yielding and reading frames, the once/continuous
+/// control flow); this owns the scopes. It needs only a [DatabaseSession] and
+/// the membership helpers, so it carries no protocol or transport concerns.
+class CrdtScopeSyncSession {
+  /// Creates a scope session for [userId] acting in [mode] against [_session].
+  ///
+  /// [peerNodeId] is the remote peer's CRDT node, learned from its connect
+  /// handshake before this session is constructed.
+  CrdtScopeSyncSession(
+    this._session, {
+    required UuidValue userId,
+    required CrdtSyncPeerMode mode,
+    required UuidValue peerNodeId,
+  }) : _userId = userId,
+       _mode = mode,
+       _peerNodeId = peerNodeId;
+
+  final DatabaseSession _session;
+  final UuidValue _userId;
+  final CrdtSyncPeerMode _mode;
+  final UuidValue _peerNodeId;
+
+  /// Session-lived scope manager. Reused across cycles so its in-memory cache
+  /// of already-materialized scopes survives, sparing a follower a per-cycle
+  /// `getOrCreate` round-trip for scopes it has already created this session.
+  late final CrdtScopeManager _scopeManager = CrdtScopeManager(_session);
+
+  /// The grant set this peer last announced (null until the first announcement).
+  List<CrdtScopeGrant>? _announcedGrants;
+
+  /// This peer's current local grants, refreshed by [reconcile].
+  List<CrdtScopeGrant> _localGrants = const [];
+
+  /// The grant set the peer last announced, adopted by [adoptPeerGrants].
+  List<CrdtScopeGrant> _peerGrants = const [];
+
+  /// The scopes cycled this round, sorted, materialized for a follower.
+  List<UuidValue> _activeScopeIds = const [];
+
+  /// UUID strings of [_activeScopeIds] and of [_peerGrants], for O(1) lookups.
+  Set<String> _activeUuids = const {};
+  Set<String> _peerUuids = const {};
+
+  /// Active scopes this peer may send local writes for.
+  Set<String> _writableUuids = const {};
+
+  /// Scopes for which this peer has already sent its [CrdtSyncSinceHlc].
+  final Set<String> _sinceHlcSent = {};
+
+  /// The peer's node id from its connect handshake.
+  UuidValue get peerNodeId => _peerNodeId;
+
+  /// What the peer has seen per scope (advances as we send), seeded from its
+  /// [CrdtSyncSinceHlc] — presence means the scope's handshake completed both
+  /// ways, so it can stream changes.
+  final Map<UuidValue, Map<UuidValue, Hlc>> _checkpointsByScope = {};
+
+  /// The grants this peer should announce this cycle.
+  List<CrdtScopeGrant> get localGrants => _localGrants;
+
+  /// The scopes cycled this round (sorted, deterministic on both peers).
+  List<UuidValue> get activeScopeIds => _activeScopeIds;
+
+  /// The authenticated user this sync session represents.
+  UuidValue get userId => _userId;
+
+  /// Whether this peer is authoritative for scope membership.
+  bool get isAuthoritative => _mode == CrdtSyncPeerMode.authoritative;
+
+  /// Whether the local grants changed since the last announcement.
+  bool get shouldAnnounce => !_grantsEqual(_announcedGrants, _localGrants);
+
+  /// Records that [localGrants] was just announced, so it is not re-sent until
+  /// it changes again.
+  void markAnnounced() => _announcedGrants = _localGrants;
+
+  /// Re-resolves local membership, then recomputes the active scope set.
+  ///
+  /// Used at the top of every data-loop cycle so an authoritative peer picks up
+  /// its own membership changes. Establishment instead recomputes from the
+  /// *announced* grants (via [adoptPeerGrants]) without re-reading, so a grant
+  /// that lands mid-handshake cannot make this peer handshake a different scope
+  /// count than it announced.
+  Future<void> reconcile() async {
+    _localGrants = _sortedUniqueGrants(await _resolveLocalGrants());
+    await _recomputeActiveScopes();
+    _refreshWritableScopes();
+  }
+
+  /// Adopts the peer's announced [peerGrants] and recomputes the active scope
+  /// set from the grants already resolved (no membership re-read). A follower
+  /// materializes the announced scopes locally and projects their roles into the
+  /// members cache (projection needs the scope rows to exist); an authoritative
+  /// peer keeps cycling its own membership.
+  Future<void> adoptPeerGrants(List<CrdtScopeGrant> peerGrants) async {
+    _peerGrants = _sortedUniqueGrants(peerGrants);
+    _peerUuids = {for (final g in _peerGrants) g.uuidScopeId.uuid};
+    await _recomputeActiveScopes();
+    if (_mode == CrdtSyncPeerMode.follower) {
+      await CrdtScopeMembership.projectFollowerMembership(
+        _session,
+        userUuid: _userId,
+        grants: _peerGrants,
+      );
+    }
+    _refreshWritableScopes();
+  }
+
+  /// Recomputes the active scope set from the current local and peer grants
+  /// (materializing a follower's adopted scopes), then prunes in-session state
+  /// for scopes that left the set — e.g. a membership the peer revoked.
+  Future<void> _recomputeActiveScopes() async {
+    _activeScopeIds = _sortedUniqueScopeIds(
+      await _resolveOrderedScopeIds(
+        localScopeIds: [for (final g in _localGrants) g.uuidScopeId],
+        peerScopeIds: [for (final g in _peerGrants) g.uuidScopeId],
+      ),
+    );
+    _activeUuids = {for (final s in _activeScopeIds) s.uuid};
+    _sinceHlcSent.removeWhere((uuid) => !_activeUuids.contains(uuid));
+    _checkpointsByScope.removeWhere((s, _) => !_activeUuids.contains(s.uuid));
+  }
+
+  void _refreshWritableScopes() {
+    if (_mode == CrdtSyncPeerMode.authoritative) {
+      _writableUuids = _activeUuids;
+      return;
+    }
+
+    // The peer's announced grants already carry the authoritative roles, so a
+    // follower derives writability in memory instead of re-reading the cache
+    // it projected from these same grants. The personal scope is always
+    // writable; consumers only consult scopes in the active set.
+    _writableUuids = {
+      _userId.uuid,
+      for (final grant in _peerGrants)
+        if (grant.role.canWrite) grant.uuidScopeId.uuid,
+    };
+  }
+
+  /// Marks that this peer is sending its [CrdtSyncSinceHlc] for [scopeId],
+  /// returning whether the scope was newly handshaked (so the driver sends one).
+  bool markHandshakeSent(UuidValue scopeId) => _sinceHlcSent.add(scopeId.uuid);
+
+  /// Records the peer's resume vector for [scopeId] from its [sinceHlc],
+  /// completing the scope's handshake from this peer's side.
+  void recordPeerHandshake(UuidValue scopeId, CrdtSyncSinceHlc sinceHlc) {
+    _checkpointsByScope[scopeId] = {
+      for (final checkpoint in sinceHlc.nodeCheckpoints) checkpoint.nodeId: checkpoint,
+    };
+  }
+
+  /// Checkpoints for writable scopes whose handshake completed both ways, keyed
+  /// by scope — the input to a pending-change collection pass.
+  Map<UuidValue, List<Hlc>> get sendableCheckpoints => {
+    for (final scopeId in _activeScopeIds)
+      if (_checkpointsByScope[scopeId] != null && _writableUuids.contains(scopeId.uuid))
+        scopeId: _checkpointsByScope[scopeId]!.values.toList(),
+  };
+
+  /// Whether any active scope still lacks the peer handshake state.
+  bool get hasIncompleteActiveHandshake => _activeScopeIds.any(
+    (scopeId) => _checkpointsByScope[scopeId] == null,
+  );
+
+  /// Advances the in-session checkpoint for [scopeId] past a just-sent [change],
+  /// so the next collection does not resend it.
+  void advanceCheckpoint(UuidValue scopeId, CrdtMergeChange change) {
+    final checkpoints = _checkpointsByScope[scopeId];
+    if (checkpoints == null) return;
+    checkpoints[change.uuidNodeId] = change.hlc.maxBetween(
+      checkpoints[change.uuidNodeId],
+    );
+  }
+
+  /// Whether this peer authorizes acting in [scopeId]: an authoritative peer
+  /// trusts its own membership, a follower trusts the peer's announced set. This
+  /// is the wire-side counterpart of the server's membership gate.
+  bool accepts(UuidValue scopeId) =>
+      (_mode == CrdtSyncPeerMode.follower ? _peerUuids : _activeUuids).contains(
+        scopeId.uuid,
+      );
+
+  /// The greatest checkpoint HLC tracked for [scopeId], or null if none.
+  Hlc? checkpointMaxOf(UuidValue scopeId) {
+    final checkpoints = _checkpointsByScope[scopeId];
+    if (checkpoints == null || checkpoints.isEmpty) return null;
+    return checkpoints.values.max;
+  }
+
+  /// Resolves the local grants this peer announces in its [CrdtSyncScopeSet].
+  ///
+  /// Authoritative peers enumerate [CrdtScopeMembership.memberGrants].
+  /// Followers send an empty set because the authoritative peer never widens
+  /// access from follower-reported scope state.
+  Future<List<CrdtScopeGrant>> _resolveLocalGrants() async {
+    switch (_mode) {
+      case CrdtSyncPeerMode.authoritative:
+        return CrdtScopeMembership.memberGrants(_session, _userId);
+      case CrdtSyncPeerMode.follower:
+        await _scopeManager.getOrCreate(_userId);
+        return const [];
+    }
+  }
+
+  /// Resolves the ordered lockstep scopes cycled this round.
+  ///
+  /// Authoritative peers dictate their own set; followers adopt the peer's
+  /// announced set, materializing each scope locally.
+  Future<List<UuidValue>> _resolveOrderedScopeIds({
+    required List<UuidValue> localScopeIds,
+    required List<UuidValue> peerScopeIds,
+  }) async {
+    switch (_mode) {
+      case CrdtSyncPeerMode.authoritative:
+        return localScopeIds;
+      case CrdtSyncPeerMode.follower:
+        final scopeIds = _sortedUniqueScopeIds(peerScopeIds);
+        for (final scopeId in scopeIds) {
+          await _scopeManager.getOrCreate(scopeId);
+        }
+        return scopeIds;
+    }
+  }
+}

packages/serverpod_offline_sync/lib/src/data/deleted.spy.yaml

@@ -0,0 +1,26 @@
+### CRDT visibility generation table.
+###
+### This table stores materialized monotone CLFlag metadata after a visibility
+### change event has been recorded.
+class: CrdtDataDeleted
+extends: BaseHlc
+table: crdt_data_tombstone
+database: all
+fields:
+  ### Row whose visibility is tracked.
+  row: CrdtDataRow?, relation(name=row_deleted, field=rowId, onDelete=Cascade)
+  rowId: int
+
+  ### The node that updated the field.
+  node: CrdtNode?, relation
+
+  ### Monotone causal-length flag. Odd values are visible and even values are deleted.
+  clFlag: int
+
+  ### Why the row entered its current visibility generation.
+  reason: CrdtDataDeletedReason
+
+indexes:
+  crdt_data_tombstone_row_idx:
+    fields: rowId
+    unique: true

packages/serverpod_offline_sync/lib/src/data/deleted_reason.spy.yaml

@@ -0,0 +1,9 @@
+### Why a synced user tombstone changed row visibility.
+enum: CrdtDataDeletedReason
+serialized: byIndex
+values:
+  - userInsert
+  - userReinsert
+  - userDelete
+  - userCascadeDelete
+  - uniqueLoser

packages/serverpod_offline_sync/lib/src/data/field.spy.yaml

@@ -0,0 +1,26 @@
+### CRDT data table.
+###
+### This table stores only the HLC timestamp for a field. During merge, if the
+### other HLC is greater, the actual data must be fetched from the table itself.
+class: CrdtDataField
+extends: BaseHlc
+table: crdt_data_fields
+database: all
+fields:
+  ### Row that the field belongs to.
+  row: CrdtDataRow?, relation(name=row_fields, onDelete=Cascade)
+
+  ### Column that the field belongs to.
+  column: CrdtSchemaColumn?, relation(onDelete=Cascade)
+
+  ### The node that updated the field.
+  node: CrdtNode?, relation
+
+  ### The foreign key information for the field, if this field holds a
+  ### reference to another row.
+  foreignKey: CrdtDataForeignKey?, relation(optional, name=field_foreign_key)
+
+indexes:
+  crdt_data_fields_row_column_idx:
+    fields: rowId, columnId
+    unique: true

packages/serverpod_offline_sync/lib/src/data/foreign_key.spy.yaml

@@ -0,0 +1,40 @@
+### CRDT field foreign key value and projection table.
+###
+### This table stores the user/attempted foreign key value for a CRDT field and
+### the materialized foreign key projection for that value. CrdtDataField stores
+### only HLC metadata, so FK columns store their durable attempted value here.
+###
+### Repairs must be recomputable from merged CRDT row, field, and tombstone
+### facts plus this attempted FK value. The projection columns record what was
+### materialized into the domain row; they do not decide business semantics or
+### create user field updates during merge.
+class: CrdtDataForeignKey
+table: crdt_data_foreign_key
+database: all
+fields:
+  ### The field that the foreign key belongs to.
+  field: CrdtDataField?, relation(name=field_foreign_key, field=fieldId, onDelete=Cascade)
+  fieldId: int
+
+  ### The user-attempted foreign key value that the CRDT field currently carries.
+  ###
+  ### This may be null for nullable foreign keys. This is durable FK field-value
+  ### storage for CRDT columns whose CrdtDataField row stores only HLC metadata.
+  attemptedValue: UuidValue?, required
+
+  ### The foreign key value that is currently visible in the actual data row
+  ### when an override is active (overrideReason is non-null).
+  visibleValue: UuidValue?
+
+  ### The authoritative indicator and reason for an active projection override.
+  ###
+  ### This is the single source of truth for whether an override is active:
+  ### non-null means an override is active; null means no override. It also
+  ### names the cause of the override. The FK resolver recomputes it from merged
+  ### row, field, and tombstone facts.
+  overrideReason: CrdtForeignKeyOverrideReason?
+
+indexes:
+  crdt_data_foreign_key_field_idx:
+    fields: fieldId
+    unique: true

packages/serverpod_offline_sync/lib/src/data/foreign_key_override_reason.spy.yaml

@@ -0,0 +1,15 @@
+### Why a foreign key projection override is active.
+enum: CrdtForeignKeyOverrideReason
+serialized: byIndex
+values:
+  ### The attempted parent target is hidden or missing and the FK projection
+  ### set the column to null.
+  - setNull
+  ### The attempted parent target is hidden or missing and the FK projection
+  ### set the column to the default value.
+  - setDefault
+  ### The attempted parent target is hidden or missing and the FK projection
+  ### cannot repair the value (restrict, no action, cascade, or the column is
+  ### non-nullable), so the row is hidden by projection while the attempt is
+  ### preserved.
+  - missingParent

packages/serverpod_offline_sync/lib/src/data/row.spy.yaml

@@ -0,0 +1,43 @@
+### CRDT data rows table.
+###
+### This table stores the data for all tables that are registered for CRDT
+### synchronization. It is used to interact with the CRDT data.
+###
+### The row is also the HLC timestamp for when it was inserted. This is used to
+### avoid creating one entry per field until a field is updated.
+class: CrdtDataRow
+extends: BaseHlc
+table: crdt_data_rows
+database: all
+fields:
+  ### Identifier for the scope that owns the data.
+  scope: CrdtScope?, relation(onDelete=Cascade)
+
+  ### Reference to the table this row belongs to.
+  tbl: CrdtSchemaTable?, relation(onDelete=Cascade)
+
+  ### Row identifier.
+  uuidRowId: UuidValue
+
+  ### The node that inserted the row with the HLC timestamp.
+  node: CrdtNode?, relation
+
+  ### Materialized visibility state for domain queries.
+  ###
+  ### Updated by user tombstone writes and foreign key projection. Visible
+  ### states are listed before hidden states in CrdtDataRowVisibility.
+  visibility: CrdtDataRowVisibility, default=userInsert
+
+  ### The synced user tombstone for this row.
+  ###
+  ### Only present after the first user visibility change event has been recorded.
+  ### Projection may keep a visible [visibility] while a user delete tombstone exists.
+  deleted: CrdtDataDeleted?, relation(name=row_deleted, optional)
+
+  ### The fields for this row.
+  fields: List<CrdtDataField>?, relation(name=row_fields)
+
+indexes:
+  crdt_data_rows_scope_tbl_row_idx:
+    fields: scopeId, tblId, uuidRowId
+    unique: true

packages/serverpod_offline_sync/lib/src/data/row_visibility.spy.yaml

@@ -0,0 +1,13 @@
+### Materialized visibility state for a CRDT row.
+###
+### Visible states are listed first, hidden states last. A row is visible when
+### its serialized index is less than or equal to [foreignKeyRestrictRestore].
+enum: CrdtDataRowVisibility
+serialized: byIndex
+values:
+  - userInsert
+  - userReinsert
+  - foreignKeyRestrictRestore
+  - userDelete
+  - userCascadeDelete
+  - foreignKeyCascade

packages/serverpod_offline_sync/lib/src/database/merge.dart

@@ -0,0 +1,976 @@
+part of 'recorder.dart';
+
+typedef _MergeRowKey = (String, UuidValue);
+typedef _MergeFieldKey = (String, UuidValue, String);
+typedef _DomainOwner = ({bool exists, int? scopeId});
+typedef _MergeContext = ({
+  Map<_MergeRowKey, CrdtDataRow> rows,
+  Map<_MergeFieldKey, CrdtDataField> fields,
+  Map<_MergeFieldKey, Hlc> incomingFieldHlcs,
+  Map<_MergeRowKey, CrdtDataDeleted> tombstones,
+  Map<_MergeRowKey, _DomainOwner> domainOwners,
+});
+typedef _MergeNodes = ({
+  Map<UuidValue, CrdtNode> nodesByUuid,
+  Map<UuidValue, CrdtScopeNode> scopeNodesByUuid,
+});
+
+/// Adds merge-specific behavior to [CrdtMutationRecorder].
+extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
+  CrdtUniqueConflictResolver get _uniqueConflictResolver =>
+      CrdtUniqueConflictResolver(this);
+
+  /// Locks the current user row so merges can serialize with other work.
+  Future<void> lockCurrentUser(Transaction transaction) async {
+    final user = _getEffectiveScope(transaction);
+    // Use a row lock without fetching the record since the merge path only
+    // needs serialization against concurrent work for the same user.
+    await CrdtScope.db.lockRows(
+      _session,
+      where: (t) => t.id.equals(user.id),
+      transaction: transaction,
+      lockMode: LockMode.forUpdate,
+      lockBehavior: LockBehavior.wait,
+    );
+  }
+
+  /// Merges remote CRDT changes into the current database.
+  Future<void> mergeChanges(
+    CrdtMergeSet mergeSet,
+    Transaction transaction,
+  ) async {
+    if (mergeSet.isEmpty) return;
+
+    final operations = mergeSet.causallyOrderedChanges;
+    final shouldProjectForeignKeys = _mergeOperationsMayAffectForeignKeys(
+      operations,
+    );
+    final currentUser = _getEffectiveScope(transaction);
+    final remoteNodes = await _findOrCreateNodesForMerge(
+      currentUser.id!,
+      {for (final change in operations) change.uuidNodeId},
+      transaction,
+    );
+    final nodesByUuid = remoteNodes.nodesByUuid;
+    final context = await _loadMergeContext(mergeSet, transaction);
+
+    for (final operation in operations) {
+      if (!_isCrdtTrackedTableName(operation.tableName)) {
+        continue;
+      }
+
+      switch (operation) {
+        case final CrdtMergeInsert insert:
+          await _applyMergeInsert(
+            insert,
+            nodesByUuid,
+            context,
+            transaction,
+          );
+        case final CrdtMergeUpdate update:
+          await _applyMergeUpdate(
+            update,
+            nodesByUuid,
+            context,
+            transaction,
+          );
+        case final CrdtMergeDelete delete:
+          await _applyMergeDelete(
+            delete,
+            nodesByUuid,
+            context,
+            transaction,
+          );
+      }
+    }
+
+    if (shouldProjectForeignKeys) {
+      await _projectForeignKeys(transaction);
+    }
+    await _updateHlcFromIncomingOperations(
+      operations,
+      remoteNodes,
+      transaction,
+    );
+  }
+
+  Future<void> _updateHlcFromIncomingOperations(
+    List<CrdtMergeChange> operations,
+    _MergeNodes remoteNodes,
+    Transaction transaction,
+  ) async {
+    final maxIncomingHlc = operations.fold<Hlc?>(
+      null,
+      (current, change) => change.hlc.maxBetween(current),
+    );
+    final scopeNodesToUpdate = <CrdtScopeNode>[];
+
+    if (maxIncomingHlc != null) {
+      final hlcManager = _getHlcManager(transaction);
+      if (maxIncomingHlc.nodeId == hlcManager.uuidNodeId) {
+        if (maxIncomingHlc > hlcManager.lastHlc) {
+          hlcManager.lastHlc = maxIncomingHlc;
+        }
+      } else {
+        hlcManager.merge(maxIncomingHlc);
+      }
+
+      await _persistCurrentNodeHlc(hlcManager, transaction);
+    }
+
+    final maxIncomingHlcByNode = <UuidValue, Hlc>{};
+    for (final operation in operations) {
+      maxIncomingHlcByNode[operation.uuidNodeId] = operation.hlc.maxBetween(
+        maxIncomingHlcByNode[operation.uuidNodeId],
+      );
+    }
+
+    for (final MapEntry(key: nodeId, value: incomingHlc)
+        in maxIncomingHlcByNode.entries) {
+      final remoteNode = remoteNodes.scopeNodesByUuid[nodeId];
+      if (remoteNode == null) continue;
+      final updatedScopeNode = remoteNode.copyWith(
+        lastReceivedHlc: incomingHlc.maxBetween(remoteNode.lastReceivedHlc),
+      );
+      if (updatedScopeNode.lastReceivedHlc == remoteNode.lastReceivedHlc) {
+        continue;
+      }
+
+      scopeNodesToUpdate.add(updatedScopeNode);
+      remoteNodes.scopeNodesByUuid[nodeId] = updatedScopeNode;
+    }
+
+    if (scopeNodesToUpdate.isNotEmpty) {
+      await CrdtScopeNode.db.update(
+        _session,
+        scopeNodesToUpdate,
+        columns: (t) => [t.lastReceivedHlc],
+        transaction: transaction,
+      );
+    }
+  }
+
+  Future<_MergeNodes> _findOrCreateNodesForMerge(
+    int userId,
+    Set<UuidValue> nodeIds,
+    Transaction transaction,
+  ) async {
+    if (nodeIds.isEmpty) {
+      return (
+        nodesByUuid: const <UuidValue, CrdtNode>{},
+        scopeNodesByUuid: const <UuidValue, CrdtScopeNode>{},
+      );
+    }
+
+    final nodesByUuid = {
+      for (final uuidNodeId in nodeIds)
+        uuidNodeId: await _findOrCreateNode(uuidNodeId, transaction),
+    };
+
+    final scopeNodesByUuid = {
+      for (final entry in nodesByUuid.entries)
+        entry.key: await _findOrCreateScopeNode(
+          userId,
+          entry.value.id!,
+          transaction,
+        ),
+    };
+
+    return (
+      nodesByUuid: nodesByUuid,
+      scopeNodesByUuid: scopeNodesByUuid,
+    );
+  }
+
+  Future<_MergeContext> _loadMergeContext(
+    CrdtMergeSet mergeSet,
+    Transaction transaction,
+  ) async {
+    final (:rowIdsByTable, :columnNamesByTable) = mergeSet.collectMetadataLookup(
+      columnNamesByTableName: _syncedTableColumnNamesForMerge,
+    );
+
+    final rows = <_MergeRowKey, CrdtDataRow>{};
+    final fields = <_MergeFieldKey, CrdtDataField>{};
+    final incomingFieldHlcs = {
+      for (final update in mergeSet.whereType<CrdtMergeUpdate>())
+        (update.tableName, update.uuidRowId, update.columnName): update.hlc,
+    };
+    final tombstones = <_MergeRowKey, CrdtDataDeleted>{};
+    // Owning scope per row, snapshotted once so the per-operation ownership
+    // checks never re-read the same immutable `scopeId`. Only rows that
+    // already have a tracker are pre-read here — those are exactly the rows
+    // whose ownership the update/delete/existing-insert paths verify. New
+    // rows go through the insert path, which proves ownership at the physical
+    // write and refreshes this entry afterwards.
+    final domainOwners = <_MergeRowKey, _DomainOwner>{};
+
+    for (final MapEntry(key: tableName, value: rowIds) in rowIdsByTable.entries) {
+      final loadedRows = await _findCrdtRows(
+        tableName,
+        rowIds,
+        transaction,
+        include: CrdtDataRow.include(
+          node: CrdtNode.include(),
+          deleted: CrdtDataDeleted.include(node: CrdtNode.include()),
+        ),
+      );
+
+      for (final row in loadedRows) {
+        rows[(tableName, row.uuidRowId)] = row;
+        if (row.deleted != null) {
+          tombstones[(tableName, row.uuidRowId)] = row.deleted!;
+        }
+      }
+
+      final trackedRowIds = loadedRows.map((row) => row.uuidRowId).toSet();
+      if (trackedRowIds.isNotEmpty) {
+        final ownerValues = await _readDomainColumnValues(
+          tableName,
+          trackedRowIds,
+          ['scopeId'],
+          transaction,
+        );
+        for (final rowId in trackedRowIds) {
+          final values = ownerValues[rowId];
+          domainOwners[(tableName, rowId)] = values == null
+              ? (exists: false, scopeId: null)
+              : (exists: true, scopeId: values['scopeId'] as int?);
+        }
+      }
+
+      final columnNames = columnNamesByTable[tableName];
+      if (columnNames == null || columnNames.isEmpty || loadedRows.isEmpty) continue;
+
+      final rowPks = loadedRows.map((row) => row.id).whereType<int>().toSet();
+      if (rowPks.length != loadedRows.length) {
+        throw StateError(
+          'Some merge metadata rows for table $tableName are missing '
+          'persisted identifiers.',
+        );
+      }
+
+      final columnIdsByName = _schemaColumnIds(_schema, tableName, columnNames);
+      if (columnIdsByName.isEmpty) continue;
+
+      final loadedFields = await CrdtDataField.db.find(
+        _session,
+        where: (t) =>
+            t.rowId.inSet(rowPks) & t.columnId.inSet(columnIdsByName.values.toSet()),
+        include: CrdtDataField.include(
+          column: CrdtSchemaColumn.include(),
+          node: CrdtNode.include(),
+        ),
+        transaction: transaction,
+      );
+
+      for (final field in loadedFields) {
+        final row = loadedRows.firstWhere((loadedRow) => loadedRow.id == field.rowId);
+        fields[(tableName, row.uuidRowId, field.column!.name)] = field;
+      }
+    }
+
+    return (
+      rows: rows,
+      fields: fields,
+      incomingFieldHlcs: incomingFieldHlcs,
+      tombstones: tombstones,
+      domainOwners: domainOwners,
+    );
+  }
+
+  Future<void> _applyMergeInsert(
+    CrdtMergeInsert insert,
+    Map<UuidValue, CrdtNode> remoteNodes,
+    _MergeContext context,
+    Transaction transaction,
+  ) async {
+    final rowKey = (insert.tableName, insert.uuidRowId);
+    final incomingHlc = insert.hlc;
+    final remoteNode = _requireRemoteNode(remoteNodes, insert.uuidNodeId);
+
+    final currentRow = context.rows[rowKey];
+    if (currentRow != null && incomingHlc <= currentRow.hlc) {
+      return;
+    }
+
+    final data = _sanitizeMergeRowData(insert.tableName, insert.databaseColumns);
+
+    if (currentRow == null) {
+      final mergingScope = _getEffectiveScope(transaction);
+      final mergingScopeId = mergingScope.id!;
+      try {
+        // Insert-first inside a savepoint: the domain primary key resolves
+        // creation races atomically, and `_applyMergeInsertForMissingRow`
+        // raises `_ForeignScopeRowCollision` when the row turns out to be owned
+        // by another scope. Rolling the savepoint back then leaves no CRDT
+        // metadata, released unique values, or foreign key attempts behind.
+        context.rows[rowKey] = await DatabaseUtil.runInTransactionOrSavepoint(
+          _db,
+          transaction,
+          (savepoint) async {
+            final resolvedInsert = await _uniqueConflictResolver
+                ._resolveForIncomingInsert(insert, data, context, savepoint);
+
+            final safeInsert = await _safeIncomingForeignKeyData(
+              insert.tableName,
+              insert.uuidRowId,
+              resolvedInsert.data,
+              savepoint,
+            );
+
+            final insertedRow = await _applyMergeInsertForMissingRow(
+              insert,
+              remoteNode,
+              incomingHlc,
+              safeInsert.data,
+              savepoint,
+            );
+            await _recordForeignKeyInsertAttempts(
+              insert.tableName,
+              {insert.uuidRowId},
+              savepoint,
+              safeInsert.attempts,
+            );
+            return insertedRow;
+          },
+        );
+        // The row is now owned by the merging scope (freshly inserted, or a
+        // same-scope recovery), so same-merge field updates resolve ownership
+        // from the cache instead of re-reading the domain row.
+        context.domainOwners[rowKey] = (exists: true, scopeId: mergingScopeId);
+      } on _ForeignScopeRowCollision catch (collision) {
+        await _throwOwnershipCollision(
+          operation: CrdtSyncViolationOperation.mergeInsert,
+          tableName: insert.tableName,
+          rowId: insert.uuidRowId,
+          owningScopeId: collision.owningScopeId,
+          incomingScope: mergingScope,
+          metadataRowId: null,
+          node: remoteNode,
+          hlc: insert.hlc,
+          transaction: transaction,
+        );
+      }
+    } else {
+      await _applyMergeInsertForExistingRow(
+        insert,
+        currentRow,
+        remoteNode,
+        incomingHlc,
+        data,
+        context,
+        transaction,
+      );
+    }
+  }
+
+  Future<void> _applyMergeUpdate(
+    CrdtMergeUpdate update,
+    Map<UuidValue, CrdtNode> remoteNodes,
+    _MergeContext context,
+    Transaction transaction,
+  ) async {
+    final rowKey = (update.tableName, update.uuidRowId);
+    final row = context.rows[rowKey];
+    if (row == null) return;
+    final remoteNode = _requireRemoteNode(remoteNodes, update.uuidNodeId);
+    if (!await _domainRowOwnedByEffectiveScope(
+      operation: CrdtSyncViolationOperation.mergeUpdate,
+      tableName: update.tableName,
+      rowId: update.uuidRowId,
+      context: context,
+      transaction: transaction,
+      metadataRowId: row.id,
+      node: remoteNode,
+      hlc: update.hlc,
+    )) {
+      return;
+    }
+
+    final shouldApply = await _shouldMergeFieldMetadataIfNewer(
+      tableName: update.tableName,
+      rowId: update.uuidRowId,
+      columnName: update.columnName,
+      row: row,
+      remoteNode: remoteNode,
+      incomingHlc: update.hlc,
+      fields: context.fields,
+      transaction: transaction,
+    );
+
+    if (!shouldApply) return;
+
+    final resolvedUpdates = await _uniqueConflictResolver._resolveForIncomingUpdate(
+      update,
+      row,
+      context,
+      transaction,
+    );
+    final safeUpdate = await _safeIncomingForeignKeyData(
+      update.tableName,
+      update.uuidRowId,
+      resolvedUpdates,
+      transaction,
+    );
+    await _updateDomainRows(
+      update.tableName,
+      {update.uuidRowId},
+      safeUpdate.data,
+      transaction,
+    );
+    await _recordForeignKeyAttemptsForRows(
+      update.tableName,
+      {update.uuidRowId},
+      resolvedUpdates.keys.toSet(),
+      transaction,
+      attemptedValues: safeUpdate.attempts,
+    );
+  }
+
+  Future<void> _applyMergeDelete(
+    CrdtMergeDelete delete,
+    Map<UuidValue, CrdtNode> remoteNodes,
+    _MergeContext context,
+    Transaction transaction,
+  ) async {
+    final rowKey = (delete.tableName, delete.uuidRowId);
+    final row = context.rows[rowKey];
+    if (row == null) return;
+    if (!delete.reason.isSynced) return;
+    final remoteNode = _requireRemoteNode(remoteNodes, delete.uuidNodeId);
+    if (!await _domainRowOwnedByEffectiveScope(
+      operation: CrdtSyncViolationOperation.mergeDelete,
+      tableName: delete.tableName,
+      rowId: delete.uuidRowId,
+      context: context,
+      transaction: transaction,
+      metadataRowId: row.id,
+      node: remoteNode,
+      hlc: delete.hlc,
+    )) {
+      return;
+    }
+
+    final currentTombstone = context.tombstones[rowKey];
+    final currentClFlag = currentTombstone?.clFlag ?? 1;
+    final currentHlc = currentTombstone?.hlc ?? row.hlc;
+    if (delete.clFlag < currentClFlag) return;
+    if (delete.clFlag == currentClFlag && delete.hlc <= currentHlc) return;
+
+    context.tombstones[rowKey] = await _upsertMergeTombstone(
+      row,
+      remoteNode,
+      delete.hlc,
+      delete.clFlag,
+      delete.reason,
+      currentTombstone,
+      transaction,
+    );
+
+    if (delete.reason.isSynced) {
+      await _applyRowVisibilityFromUserTombstone(
+        row,
+        context.tombstones[rowKey]!,
+        transaction,
+      );
+    }
+  }
+
+  Future<CrdtDataRow> _upsertMergeRow(
+    String tableName,
+    UuidValue rowId,
+    CrdtNode remoteNode,
+    Hlc incomingHlc,
+    CrdtDataRow? currentRow,
+    Transaction transaction,
+  ) async {
+    if (currentRow == null) {
+      final (tableId, _) = _schema[tableName]!;
+      final insertedRow = await CrdtDataRow.db.insertRow(
+        _session,
+        CrdtDataRow(
+          scopeId: _getHlcManager(transaction).normalizedScopeId,
+          tblId: tableId,
+          uuidRowId: rowId,
+          nodeId: remoteNode.id!,
+          hlcDatetime: incomingHlc.datetime,
+          hlcCounter: incomingHlc.counter,
+        ),
+        transaction: transaction,
+      );
+
+      return insertedRow.copyWith(node: remoteNode);
+    }
+
+    final updatedRow = currentRow.copyWith(
+      nodeId: remoteNode.id,
+      node: remoteNode,
+      hlcDatetime: incomingHlc.datetime,
+      hlcCounter: incomingHlc.counter,
+    );
+
+    await CrdtDataRow.db.update(
+      _session,
+      [updatedRow],
+      columns: (t) => [t.nodeId, t.hlcDatetime, t.hlcCounter],
+      transaction: transaction,
+    );
+
+    return updatedRow;
+  }
+
+  Future<CrdtDataField> _upsertMergeField(
+    CrdtDataRow row,
+    CrdtSchemaColumn column,
+    CrdtNode remoteNode,
+    Hlc incomingHlc,
+    CrdtDataField? currentField,
+    Transaction transaction,
+  ) async {
+    if (currentField == null) {
+      final insertedField = await CrdtDataField.db.insertRow(
+        _session,
+        CrdtDataField(
+          rowId: row.id!,
+          columnId: column.id!,
+          nodeId: remoteNode.id!,
+          hlcDatetime: incomingHlc.datetime,
+          hlcCounter: incomingHlc.counter,
+        ),
+        transaction: transaction,
+      );
+
+      return insertedField.copyWith(
+        row: row,
+        column: column,
+        node: remoteNode,
+      );
+    }
+
+    final updatedField = currentField.copyWith(
+      row: row,
+      column: column,
+      nodeId: remoteNode.id,
+      node: remoteNode,
+      hlcDatetime: incomingHlc.datetime,
+      hlcCounter: incomingHlc.counter,
+    );
+
+    await CrdtDataField.db.update(
+      _session,
+      [updatedField],
+      columns: (t) => [t.nodeId, t.hlcDatetime, t.hlcCounter],
+      transaction: transaction,
+    );
+    return updatedField;
+  }
+
+  Future<CrdtDataDeleted> _upsertMergeTombstone(
+    CrdtDataRow row,
+    CrdtNode remoteNode,
+    Hlc incomingHlc,
+    int clFlag,
+    CrdtDataDeletedReason reason,
+    CrdtDataDeleted? currentTombstone,
+    Transaction transaction,
+  ) async {
+    if (currentTombstone == null) {
+      final insertedTombstone = await CrdtDataDeleted.db.insertRow(
+        _session,
+        CrdtDataDeleted(
+          rowId: row.id!,
+          nodeId: remoteNode.id!,
+          hlcDatetime: incomingHlc.datetime,
+          hlcCounter: incomingHlc.counter,
+          clFlag: clFlag,
+          reason: reason,
+        ),
+        transaction: transaction,
+      );
+
+      return insertedTombstone.copyWith(
+        row: row,
+        node: remoteNode,
+      );
+    }
+
+    final updatedTombstone = currentTombstone.copyWith(
+      row: row,
+      nodeId: remoteNode.id,
+      node: remoteNode,
+      hlcDatetime: incomingHlc.datetime,
+      hlcCounter: incomingHlc.counter,
+      clFlag: clFlag,
+      reason: reason,
+    );
+
+    await CrdtDataDeleted.db.update(
+      _session,
+      [updatedTombstone],
+      columns: (t) => [t.nodeId, t.hlcDatetime, t.hlcCounter, t.clFlag, t.reason],
+      transaction: transaction,
+    );
+
+    return updatedTombstone;
+  }
+
+  Future<CrdtDataRow> _applyMergeInsertForMissingRow(
+    CrdtMergeInsert insert,
+    CrdtNode remoteNode,
+    Hlc incomingHlc,
+    Map<String, Object?> data,
+    Transaction transaction,
+  ) async {
+    final insertedCrdtRow = await _upsertMergeRow(
+      insert.tableName,
+      insert.uuidRowId,
+      remoteNode,
+      incomingHlc,
+      null,
+      transaction,
+    );
+
+    final tableRow = _requireMergeInsertTableRow(insert);
+    final patchedRow = _db.serializationManager.patchTableRow(tableRow, data);
+    final mergingScopeId = _getHlcManager(transaction).normalizedScopeId;
+    final scopedRow = _withScopeId(patchedRow, mergingScopeId);
+    try {
+      // Insert first: the domain primary key resolves creation races
+      // atomically. The inner savepoint keeps a violation from poisoning the
+      // surrounding transaction so the recovery update below can still run.
+      await DatabaseUtil.runInTransactionOrSavepoint(
+        _db,
+        transaction,
+        (savepoint) => _db.insertRow(scopedRow, transaction: savepoint),
+      );
+    } on DatabaseQueryException {
+      final owner = await _readDomainRowOwner(
+        insert.tableName,
+        insert.uuidRowId,
+        transaction,
+      );
+      // No row means the violation came from something other than the
+      // primary key (e.g. a racing unique value); preserve the failure.
+      if (!owner.exists) rethrow;
+      if (owner.scopeId != mergingScopeId) {
+        throw _ForeignScopeRowCollision(owner.scopeId);
+      }
+
+      // Same-scope recovery (a tracked row whose metadata was lost, or a
+      // race between this scope's own merges). scopeId is immutable and rows
+      // are only soft-deleted, so ownership observed here cannot change for
+      // the rest of the transaction and the update is safe.
+      await _db.updateRow(
+        scopedRow,
+        columns: scopedRow.table.managedColumns
+            .where((c) => data.containsKey(c.columnName))
+            .toList(),
+        transaction: transaction,
+      );
+    }
+
+    return insertedCrdtRow;
+  }
+
+  Future<void> _applyMergeInsertForExistingRow(
+    CrdtMergeInsert insert,
+    CrdtDataRow currentRow,
+    CrdtNode remoteNode,
+    Hlc incomingHlc,
+    Map<String, Object?> data,
+    _MergeContext context,
+    Transaction transaction,
+  ) async {
+    if (!await _domainRowOwnedByEffectiveScope(
+      operation: CrdtSyncViolationOperation.mergeInsert,
+      tableName: insert.tableName,
+      rowId: insert.uuidRowId,
+      context: context,
+      transaction: transaction,
+      metadataRowId: currentRow.id,
+      node: remoteNode,
+      hlc: insert.hlc,
+    )) {
+      return;
+    }
+
+    final (_, columnsByName) = _schema[insert.tableName]!;
+    final updatedValues = <String, Object?>{};
+    for (final MapEntry(key: columnName, value: schemaColumn)
+        in columnsByName.entries) {
+      if (columnName == 'id') continue;
+
+      final shouldApply = await _shouldMergeFieldMetadataIfNewer(
+        tableName: insert.tableName,
+        rowId: insert.uuidRowId,
+        columnName: columnName,
+        row: currentRow,
+        remoteNode: remoteNode,
+        incomingHlc: incomingHlc,
+        fields: context.fields,
+        transaction: transaction,
+        schemaColumn: schemaColumn,
+      );
+
+      if (!shouldApply) continue;
+      updatedValues[columnName] = data[columnName];
+    }
+
+    var updatesToApply = updatedValues;
+    if (updatesToApply.isNotEmpty) {
+      updatesToApply = await _uniqueConflictResolver._resolveForIncomingUpdates(
+        tableName: insert.tableName,
+        row: currentRow,
+        updates: updatesToApply,
+        context: context,
+        transaction: transaction,
+      );
+    }
+
+    if (updatesToApply.isNotEmpty) {
+      final safeUpdate = await _safeIncomingForeignKeyData(
+        insert.tableName,
+        insert.uuidRowId,
+        updatesToApply,
+        transaction,
+      );
+      await _updateDomainRows(
+        insert.tableName,
+        {insert.uuidRowId},
+        safeUpdate.data,
+        transaction,
+      );
+      await _recordForeignKeyAttemptsForRows(
+        insert.tableName,
+        {insert.uuidRowId},
+        updatesToApply.keys.toSet(),
+        transaction,
+        attemptedValues: safeUpdate.attempts,
+      );
+    }
+
+    final rowKey = (insert.tableName, insert.uuidRowId);
+    final currentTombstone = context.tombstones[rowKey];
+    if ((currentTombstone?.clFlag ?? 1) == 1 &&
+        incomingHlc > (currentTombstone?.hlc ?? currentRow.hlc)) {
+      context.tombstones[rowKey] = await _upsertMergeTombstone(
+        currentRow,
+        remoteNode,
+        incomingHlc,
+        1,
+        CrdtDataDeletedReason.userInsert,
+        currentTombstone,
+        transaction,
+      );
+      await _applyRowVisibilityFromUserTombstone(
+        currentRow,
+        context.tombstones[rowKey]!,
+        transaction,
+      );
+    }
+  }
+
+  Future<bool> _shouldMergeFieldMetadataIfNewer({
+    required String tableName,
+    required UuidValue rowId,
+    required String columnName,
+    required CrdtDataRow row,
+    required CrdtNode remoteNode,
+    required Hlc incomingHlc,
+    required Map<_MergeFieldKey, CrdtDataField> fields,
+    required Transaction transaction,
+    CrdtSchemaColumn? schemaColumn,
+  }) async {
+    final resolvedSchemaColumn =
+        schemaColumn ?? _schemaColumn(_schema, tableName, columnName);
+    if (resolvedSchemaColumn == null) return false;
+
+    final fieldKey = (tableName, rowId, columnName);
+    final currentField = fields[fieldKey];
+    final currentHlc = currentField?.hlc ?? row.hlc;
+    if (incomingHlc <= currentHlc) return false;
+
+    fields[fieldKey] = await _upsertMergeField(
+      row,
+      resolvedSchemaColumn,
+      remoteNode,
+      incomingHlc,
+      currentField,
+      transaction,
+    );
+    return true;
+  }
+
+  TableRow _requireMergeInsertTableRow(CrdtMergeInsert insert) {
+    final row = insert.data;
+    if (row is TableRow) return row;
+
+    throw StateError(
+      'Unsupported merge insert payload type for "${insert.tableName}": '
+      '${row.runtimeType}. Expected subclass of TableRow.',
+    );
+  }
+
+  CrdtNode _requireRemoteNode(
+    Map<UuidValue, CrdtNode> remoteNodes,
+    UuidValue uuidNodeId,
+  ) {
+    return remoteNodes[uuidNodeId] ??
+        (throw StateError(
+          'Remote node $uuidNodeId not found for merge.',
+        ));
+  }
+
+  Map<String, Object?> _sanitizeMergeRowData(
+    String tableName,
+    Map<String, Object?> data,
+  ) {
+    final columns = _context._columnsByTableAndName[tableName];
+    if (columns == null) return {};
+
+    return {
+      for (final MapEntry(key: columnName, value: value) in data.entries)
+        if (columnName != 'id' &&
+            columnName != 'scopeId' &&
+            columns.containsKey(columnName))
+          columnName: columns[columnName]!.columnType == ColumnType.uuid
+              ? _uuidValueFromDatabase(value)
+              : value,
+    };
+  }
+
+  Future<({bool exists, int? scopeId})> _readDomainRowOwner(
+    String tableName,
+    UuidValue rowId,
+    Transaction transaction,
+  ) async {
+    final values = await _readDomainColumnValues(
+      tableName,
+      {rowId},
+      ['scopeId'],
+      transaction,
+    );
+    final row = values[rowId];
+    if (row == null) return (exists: false, scopeId: null);
+    return (exists: true, scopeId: row['scopeId'] as int?);
+  }
+
+  /// Returns the owning scope of [rowId] from the merge snapshot taken in
+  /// [_loadMergeContext], reading the domain row only if it was not captured
+  /// there (e.g. a row a concurrent operation added since the snapshot).
+  Future<_DomainOwner> _resolveDomainOwner(
+    _MergeContext context,
+    String tableName,
+    UuidValue rowId,
+    Transaction transaction,
+  ) async {
+    final key = (tableName, rowId);
+    final cached = context.domainOwners[key];
+    if (cached != null) return cached;
+    final owner = await _readDomainRowOwner(tableName, rowId, transaction);
+    context.domainOwners[key] = owner;
+    return owner;
+  }
+
+  Future<bool> _domainRowOwnedByEffectiveScope({
+    required CrdtSyncViolationOperation operation,
+    required String tableName,
+    required UuidValue rowId,
+    required _MergeContext context,
+    required Transaction transaction,
+    int? metadataRowId,
+    CrdtNode? node,
+    Hlc? hlc,
+  }) async {
+    final owner = await _resolveDomainOwner(context, tableName, rowId, transaction);
+    final mergingScope = _getEffectiveScope(transaction);
+    if (owner.exists && owner.scopeId == mergingScope.id) return true;
+    if (!owner.exists) return false;
+
+    await _throwOwnershipCollision(
+      operation: operation,
+      tableName: tableName,
+      rowId: rowId,
+      owningScopeId: owner.scopeId,
+      incomingScope: mergingScope,
+      metadataRowId: metadataRowId,
+      node: node,
+      hlc: hlc,
+      transaction: transaction,
+    );
+  }
+
+  Future<Never> _throwOwnershipCollision({
+    required CrdtSyncViolationOperation operation,
+    required String tableName,
+    required UuidValue rowId,
+    required int? owningScopeId,
+    required CrdtScope incomingScope,
+    required int? metadataRowId,
+    required Transaction transaction,
+    CrdtNode? node,
+    Hlc? hlc,
+  }) async {
+    final now = DateTime.now().toUtc();
+    throw CrdtSyncIntegrityViolationException(
+      CrdtSyncIntegrityViolation(
+        type: CrdtSyncViolationType.ownershipCollision,
+        domainTableName: tableName,
+        uuidRowId: rowId,
+        ownerScopeUuid: await _scopeUuidForNormalizedId(owningScopeId, transaction),
+        incomingScopeUuid: incomingScope.uuidScopeId,
+        operation: operation,
+        uuidNodeId: node?.uuidNodeId,
+        crdtDataRowId: metadataRowId,
+        hlcDatetime: hlc?.datetime,
+        hlcCounter: hlc?.counter,
+        firstSeenAt: now,
+        lastSeenAt: now,
+        occurrences: 1,
+      ),
+    );
+  }
+
+  Future<UuidValue?> _scopeUuidForNormalizedId(
+    int? scopeId,
+    Transaction transaction,
+  ) async {
+    if (scopeId == null) return null;
+
+    final scope = await CrdtScope.db.findById(
+      _session,
+      scopeId,
+      transaction: transaction,
+    );
+    return scope?.uuidScopeId;
+  }
+
+  T _withScopeId<T extends TableRow>(T row, int scopeId) {
+    return (row as dynamic).copyWith(scopeId: scopeId) as T;
+  }
+}
+
+/// Signals that a merge insert lost a primary-key race to a row owned by
+/// another scope, so the savepoint around the insert application must be
+/// rolled back before the sync violation is recorded and thrown.
+class _ForeignScopeRowCollision implements Exception {
+  _ForeignScopeRowCollision(this.owningScopeId);
+
+  final int? owningScopeId;
+}
+
+extension on DatabaseSerializationManager {
+  /// Patches a [TableRow] with the given data.
+  T patchTableRow<T extends TableRow>(T row, Map<String, Object?> data) {
+    final rowJson = Map<String, dynamic>.from(row.toJson() as Map);
+    for (final column in row.table.managedColumns) {
+      if (column.columnName != 'id' &&
+          column.columnName != 'scopeId' &&
+          data.containsKey(column.columnName)) {
+        rowJson[column.fieldName] = data[column.columnName];
+      }
+    }
+    return deserialize<T>(rowJson);
+  }
+}

packages/serverpod_offline_sync/lib/src/database/merge_utils/unique_resolver.dart

@@ -0,0 +1,383 @@
+part of '../recorder.dart';
+
+typedef _UniqueConflict = ({
+  CrdtDataRow row,
+  _UniqueIndexConflictRelease uniqueIndex,
+});
+
+/// Resolves unique constraint conflicts that appear while merging CRDT changes.
+class CrdtUniqueConflictResolver {
+  /// Creates a resolver for the merge recorder.
+  CrdtUniqueConflictResolver(this._recorder);
+
+  final CrdtMutationRecorder _recorder;
+
+  /// Resolves unique conflicts for an incoming insert.
+  Future<({Map<String, Object?> data, bool changed})> _resolveForIncomingInsert(
+    CrdtMergeInsert insert,
+    Map<String, Object?> data,
+    _MergeContext context,
+    Transaction transaction,
+  ) async {
+    final resolvedData = Map<String, Object?>.from(data);
+    var changed = false;
+    final conflicts = await _findVisibleUniqueConflicts(
+      tableName: insert.tableName,
+      rowId: insert.uuidRowId,
+      values: data,
+      transaction: transaction,
+    );
+
+    for (final conflict in conflicts) {
+      final incomingClaimHlc = _incomingInsertUniqueClaim(
+        insert: insert,
+        uniqueIndex: conflict.uniqueIndex,
+        context: context,
+      );
+      final conflictClaimHlc = await _uniqueIndexHlc(
+        tableName: insert.tableName,
+        row: conflict.row,
+        uniqueIndex: conflict.uniqueIndex,
+        fields: context.fields,
+        transaction: transaction,
+      );
+
+      if (incomingClaimHlc > conflictClaimHlc) {
+        resolvedData.addAll(
+          _uniqueConflictFreeValuesForData(
+            insert.tableName,
+            insert.uuidRowId,
+            resolvedData,
+            conflict.uniqueIndex,
+          ),
+        );
+        changed = true;
+      } else {
+        await _releaseUniqueConflictForRow(
+          tableName: insert.tableName,
+          rowId: conflict.row.uuidRowId,
+          uniqueIndex: conflict.uniqueIndex,
+          transaction: transaction,
+        );
+      }
+    }
+
+    return (data: resolvedData, changed: changed);
+  }
+
+  Hlc _incomingInsertUniqueClaim({
+    required CrdtMergeInsert insert,
+    required _UniqueIndexConflictRelease uniqueIndex,
+    required _MergeContext context,
+  }) {
+    var uniqueHlc = insert.hlc;
+    for (final columnName in uniqueIndex.indexedColumns) {
+      uniqueHlc = uniqueHlc.maxBetween(
+        context.incomingFieldHlcs[(
+          insert.tableName,
+          insert.uuidRowId,
+          columnName,
+        )],
+      );
+    }
+    return uniqueHlc;
+  }
+
+  /// Resolves unique conflicts for a single incoming field update.
+  Future<Map<String, Object?>> _resolveForIncomingUpdate(
+    CrdtMergeUpdate update,
+    CrdtDataRow row,
+    _MergeContext context,
+    Transaction transaction,
+  ) {
+    return _resolveForIncomingUpdates(
+      tableName: update.tableName,
+      row: row,
+      updates: {update.columnName: update.value},
+      context: context,
+      transaction: transaction,
+    );
+  }
+
+  /// Resolves unique conflicts for one or more incoming field updates.
+  Future<Map<String, Object?>> _resolveForIncomingUpdates({
+    required String tableName,
+    required CrdtDataRow row,
+    required Map<String, Object?> updates,
+    required _MergeContext context,
+    required Transaction transaction,
+  }) async {
+    if (updates.keys.every(
+      (columnName) => !_isUniqueIndexedColumn(tableName, columnName),
+    )) {
+      return updates;
+    }
+
+    final tableDefinition = _recorder._tableDefinitionsByName[tableName];
+    if (tableDefinition == null) {
+      return updates;
+    }
+
+    final uniqueColumnNamesToRead = {
+      for (final uniqueIndex in _recorder._uniqueIndexesForTable(tableDefinition))
+        ...uniqueIndex.indexedColumns,
+    };
+    if (uniqueColumnNamesToRead.isEmpty) {
+      return updates;
+    }
+
+    final currentValues = await _recorder._readDomainColumnValues(
+      tableName,
+      {row.uuidRowId},
+      uniqueColumnNamesToRead.toList(),
+      transaction,
+    );
+    final values = {
+      ...?currentValues[row.uuidRowId],
+      ...updates,
+    };
+
+    final conflicts = await _findVisibleUniqueConflicts(
+      tableName: tableName,
+      rowId: row.uuidRowId,
+      values: values,
+      transaction: transaction,
+    );
+
+    final resolvedUpdates = Map<String, Object?>.from(updates);
+    for (final conflict in conflicts) {
+      final rowClaimHlc = await _uniqueIndexHlc(
+        tableName: tableName,
+        row: row,
+        uniqueIndex: conflict.uniqueIndex,
+        fields: context.fields,
+        transaction: transaction,
+      );
+      final conflictClaimHlc = await _uniqueIndexHlc(
+        tableName: tableName,
+        row: conflict.row,
+        uniqueIndex: conflict.uniqueIndex,
+        fields: context.fields,
+        transaction: transaction,
+      );
+
+      if (rowClaimHlc > conflictClaimHlc) {
+        final releasedValues = _uniqueConflictFreeValuesForData(
+          tableName,
+          row.uuidRowId,
+          values,
+          conflict.uniqueIndex,
+        );
+        for (final columnName in conflict.uniqueIndex.releaseColumnNames) {
+          if (releasedValues.containsKey(columnName)) {
+            resolvedUpdates[columnName] = releasedValues[columnName];
+            values[columnName] = releasedValues[columnName];
+          }
+        }
+      } else {
+        await _releaseUniqueConflictForRow(
+          tableName: tableName,
+          rowId: conflict.row.uuidRowId,
+          uniqueIndex: conflict.uniqueIndex,
+          transaction: transaction,
+        );
+      }
+    }
+
+    return resolvedUpdates;
+  }
+
+  Map<String, Object?> _uniqueConflictFreeValuesForData(
+    String tableName,
+    UuidValue rowId,
+    Map<String, Object?> data,
+    _UniqueIndexConflictRelease uniqueIndex,
+  ) {
+    final tableDefinition = _recorder._tableDefinitionsByName[tableName];
+    if (tableDefinition == null) return data;
+
+    final released = Map<String, Object?>.from(data);
+    final values = {
+      for (final columnName in uniqueIndex.indexedColumns)
+        columnName: released[columnName],
+    };
+    if (values.values.any((value) => value == null)) return released;
+
+    for (final column in uniqueIndex.releaseColumns) {
+      released[column.columnName] = _recorder._conflictFreeValue(
+        column,
+        values[column.columnName],
+        tableName,
+        rowId,
+        'conflict',
+      );
+    }
+    return released;
+  }
+
+  Future<void> _releaseUniqueConflictForRow({
+    required String tableName,
+    required UuidValue rowId,
+    required _UniqueIndexConflictRelease uniqueIndex,
+    required Transaction transaction,
+  }) async {
+    final valuesByRowId = await _recorder._readDomainColumnValues(
+      tableName,
+      {rowId},
+      uniqueIndex.indexedColumns,
+      transaction,
+    );
+    final values = valuesByRowId[rowId];
+    if (values == null || values.values.any((value) => value == null)) return;
+
+    final releasedValues = _uniqueConflictFreeValuesForData(
+      tableName,
+      rowId,
+      values,
+      uniqueIndex,
+    );
+    await _recorder._updateDomainRows(tableName, {rowId}, releasedValues, transaction);
+  }
+
+  Future<List<_UniqueConflict>> _findVisibleUniqueConflicts({
+    required String tableName,
+    required UuidValue rowId,
+    required Map<String, Object?> values,
+    required Transaction transaction,
+  }) async {
+    final tableDefinition = _recorder._tableDefinitionsByName[tableName];
+    if (tableDefinition == null) return const [];
+
+    final uniqueIndexes = _recorder._uniqueIndexesForTable(tableDefinition);
+    if (uniqueIndexes.isEmpty) return const [];
+
+    final uniqueIndexesByConflictId = <UuidValue, Set<_UniqueIndexConflictRelease>>{};
+    for (final uniqueIndex in uniqueIndexes) {
+      final columnValues = {
+        for (final columnName in uniqueIndex.indexedColumns)
+          if (values.containsKey(columnName)) columnName: values[columnName],
+      };
+      if (uniqueIndex.scoped) {
+        columnValues['scopeId'] = _recorder
+            ._getHlcManager(transaction)
+            .normalizedScopeId;
+      }
+      final expectedValueCount =
+          uniqueIndex.indexedColumns.length + (uniqueIndex.scoped ? 1 : 0);
+      if (columnValues.length != expectedValueCount) continue;
+      if (columnValues.values.any((value) => value == null)) continue;
+
+      final uniqueConflictIds = await _findVisibleRowsByUniqueValues(
+        tableName: tableName,
+        rowId: rowId,
+        values: columnValues,
+        transaction: transaction,
+      );
+      for (final conflictId in uniqueConflictIds) {
+        uniqueIndexesByConflictId.putIfAbsent(conflictId, () => {}).add(uniqueIndex);
+      }
+    }
+
+    if (uniqueIndexesByConflictId.isEmpty) return const [];
+    final conflictRows = await _recorder._findCrdtRows(
+      tableName,
+      uniqueIndexesByConflictId.keys.toSet(),
+      transaction,
+      include: CrdtDataRow.include(
+        node: CrdtNode.include(),
+        deleted: CrdtDataDeleted.include(node: CrdtNode.include()),
+      ),
+    );
+    return [
+      for (final conflictRow in conflictRows)
+        for (final uniqueIndex in uniqueIndexesByConflictId[conflictRow.uuidRowId]!)
+          (
+            row: conflictRow,
+            uniqueIndex: uniqueIndex,
+          ),
+    ];
+  }
+
+  /// Returns the HLC for this row's current claim on [uniqueIndex].
+  ///
+  /// Composite indexes use the newest field HLC among their columns; rows with
+  /// no field metadata fall back to the row insert HLC. Missing field metadata
+  /// is loaded lazily because conflicting rows may not be part of the preloaded
+  /// merge context.
+  Future<Hlc> _uniqueIndexHlc({
+    required String tableName,
+    required CrdtDataRow row,
+    required _UniqueIndexConflictRelease uniqueIndex,
+    required Map<_MergeFieldKey, CrdtDataField> fields,
+    required Transaction transaction,
+  }) async {
+    final uniqueColumnNames = {
+      ...uniqueIndex.indexedColumns,
+    };
+    final missingColumnIds = <int>{};
+    final maxFieldHlc = <Hlc>[];
+    for (final columnName in uniqueColumnNames) {
+      final field = fields[(tableName, row.uuidRowId, columnName)];
+      if (field == null) {
+        final columnId = _schemaColumn(_recorder._schema, tableName, columnName)?.id;
+        if (columnId != null) missingColumnIds.add(columnId);
+        continue;
+      }
+
+      maxFieldHlc.add(field.hlc);
+    }
+
+    if (missingColumnIds.isNotEmpty && row.id != null) {
+      final loadedFields = await CrdtDataField.db.find(
+        _recorder._session,
+        where: (t) => t.rowId.equals(row.id) & t.columnId.inSet(missingColumnIds),
+        include: CrdtDataField.include(
+          column: CrdtSchemaColumn.include(),
+          node: CrdtNode.include(),
+        ),
+        transaction: transaction,
+      );
+
+      for (final field in loadedFields) {
+        final columnName = field.column?.name;
+        if (columnName == null) continue;
+        fields[(tableName, row.uuidRowId, columnName)] = field;
+        maxFieldHlc.add(field.hlc);
+      }
+    }
+
+    if (maxFieldHlc.isEmpty) return row.hlc;
+    return maxFieldHlc.reduce((left, right) => left.maxBetween(right));
+  }
+
+  Future<Set<UuidValue>> _findVisibleRowsByUniqueValues({
+    required String tableName,
+    required UuidValue rowId,
+    required Map<String, Object?> values,
+    required Transaction transaction,
+  }) async {
+    return _recorder._findVisibleDomainRowIdsWhere(
+      tableName: tableName,
+      predicates: [
+        for (final MapEntry(key: columnName, value: value) in values.entries)
+          _domainColumnPredicate(columnName, value),
+        _domainColumnNotPredicate('id', rowId),
+      ],
+      transaction: transaction,
+    );
+  }
+
+  bool _isUniqueIndexedColumn(String tableName, String columnName) {
+    final tableDefinition = _recorder._tableDefinitionsByName[tableName];
+    if (tableDefinition == null) return false;
+    return _recorder
+        ._uniqueIndexesForTable(tableDefinition)
+        .any((i) => i.indexedColumns.contains(columnName));
+  }
+}
+
+extension on _UniqueIndexConflictRelease {
+  Set<String> get releaseColumnNames => {
+    for (final column in releaseColumns) column.columnName,
+  };
+}