feat: Implement shared scopes with role enforcement (#90)
This commit is contained in:
Marcelo Mendonça Soares
GitHub
parent
517abde4a1
commit
8acfa7f269
28 changed files with 4868 additions and 485 deletions
|
|
@ -1,6 +1,7 @@
|
|||
export 'src/crdt/client_sync.dart';
|
||||
export 'src/crdt/exceptions.dart' hide PendingOutboundIntegrityViolation;
|
||||
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';
|
||||
|
|
|
|||
|
|
@ -68,7 +68,12 @@ class CrdtSyncClient {
|
|||
);
|
||||
|
||||
final subscription = session.crdtDb
|
||||
.sync(inbound: remoteStream, once: once, onMergeSuccess: onMergeSuccess)
|
||||
.sync(
|
||||
inbound: remoteStream,
|
||||
once: once,
|
||||
onMergeSuccess: onMergeSuccess,
|
||||
mode: CrdtSyncPeerMode.follower,
|
||||
)
|
||||
.listen(
|
||||
outboundChanges.addIfNotClosed,
|
||||
onDone: doneCompleter.completeOrSkip,
|
||||
|
|
|
|||
|
|
@ -43,26 +43,6 @@ final class CrdtSyncUnexpectedEventException<T extends CrdtSyncStreamEvent>
|
|||
'received "${received.runtimeType.className}" instead.';
|
||||
}
|
||||
|
||||
/// Thrown when a peer's since-HLC frame does not reference the local node id.
|
||||
final class CrdtSyncInvalidSinceHlcException extends CrdtSyncException {
|
||||
/// Creates a [CrdtSyncInvalidSinceHlcException].
|
||||
const CrdtSyncInvalidSinceHlcException({
|
||||
required this.receivedNodeId,
|
||||
required this.expectedNodeId,
|
||||
});
|
||||
|
||||
/// The node id carried by the peer's since-HLC frame.
|
||||
final UuidValue receivedNodeId;
|
||||
|
||||
/// The local node id that was expected.
|
||||
final UuidValue expectedNodeId;
|
||||
|
||||
@override
|
||||
String toString() =>
|
||||
'CrdtSyncInvalidSinceHlcException: peer since HLC node id '
|
||||
'"$receivedNodeId" does not match local node id "$expectedNodeId".';
|
||||
}
|
||||
|
||||
/// Thrown when the sync tables hash sent by a peer does not match locally.
|
||||
final class SyncTablesHashMismatchException extends CrdtSyncException {
|
||||
/// Creates a [SyncTablesHashMismatchException].
|
||||
|
|
@ -109,10 +89,58 @@ final class CrdtSyncIntegrityViolationException extends CrdtSyncException {
|
|||
'${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({
|
||||
|
|
|
|||
|
|
@ -69,3 +69,12 @@ extension CrdtDataDeletedReasonExtension on CrdtDataDeletedReason {
|
|||
/// 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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -20,38 +20,53 @@ typedef CrdtMergeMetadataLookup = ({
|
|||
extension CrdtSyncStreamEventStreamExtension on StreamIterator<CrdtSyncStreamEvent> {
|
||||
/// Collects the next framed sync batch from this iterator.
|
||||
///
|
||||
/// Each batch is zero or more [CrdtSyncMergeChunk] events followed by
|
||||
/// [CrdtSyncEndOfBatch]. If the stream is idle before a batch starts, an
|
||||
/// 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 stream closes
|
||||
/// before the next batch starts. Otherwise, closing before a batch starts is
|
||||
/// treated as a [CrdtSyncStreamClosedException].
|
||||
/// 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 stream closes after a merge change was already received, the
|
||||
/// partial batch is still treated as an error.
|
||||
Future<CrdtMergeSet?> collectNextBatch({
|
||||
/// 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 mergeSet = <CrdtMergeChange>[];
|
||||
final batch = CrdtSyncCycleBatch();
|
||||
|
||||
while (await moveNext()) {
|
||||
switch (current) {
|
||||
case CrdtSyncMergeChunk(changes: final changes):
|
||||
mergeSet.addAll(changes);
|
||||
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 (mergeSet.isEmpty) return mergeSet;
|
||||
if (batch.isEmpty) return batch;
|
||||
case CrdtSyncEndOfBatch():
|
||||
return mergeSet;
|
||||
return batch;
|
||||
case CrdtSyncClose():
|
||||
return null;
|
||||
default:
|
||||
throw CrdtSyncUnexpectedEventException(
|
||||
expected: '"CrdtSyncMergeChunk" or "CrdtSyncEndOfBatch"',
|
||||
expected: 'a sync cycle frame',
|
||||
received: current,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if (mergeSet.isEmpty && allowCloseBeforeBatch) return null;
|
||||
if (batch.isEmpty && allowCloseBeforeBatch) return null;
|
||||
throw const CrdtSyncStreamClosedException(phase: 'end-of-batch');
|
||||
}
|
||||
|
||||
|
|
@ -70,6 +85,21 @@ extension CrdtSyncStreamEventStreamExtension on StreamIterator<CrdtSyncStreamEve
|
|||
}
|
||||
}
|
||||
|
||||
/// 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.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,204 @@
|
|||
import 'package:meta/meta.dart';
|
||||
import 'package:serverpod_database/serverpod_database.dart' hide Protocol;
|
||||
import 'package:uuid/uuid.dart';
|
||||
|
||||
import '../protocol/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.
|
||||
///
|
||||
/// Queries are raw SQL rather than the generated ORM on purpose: `find<T>`
|
||||
/// resolves the table through the session's serialization manager keyed by the
|
||||
/// model [Type], and a server session only registers the server-package model
|
||||
/// classes — never this client-package [CrdtScopeMember]. Raw SQL is
|
||||
/// serialization-manager-agnostic and runs identically on both ends, matching
|
||||
/// the rest of the CRDT engine's cross-node queries.
|
||||
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 encodedUserUuid = ValueEncoder.instance.convert(userUuid);
|
||||
final result = await session.db.unsafeQuery(
|
||||
'SELECT s."${CrdtScope.t.uuidScopeId.columnName}", '
|
||||
'm."${CrdtScopeMember.t.role.columnName}" '
|
||||
'FROM "${CrdtScopeMember.t.tableName}" m '
|
||||
'JOIN "${CrdtScope.t.tableName}" s '
|
||||
'ON s."${CrdtScope.t.id.columnName}" = m."${CrdtScopeMember.t.scopeId.columnName}" '
|
||||
'WHERE m."${CrdtScopeMember.t.userUuid.columnName}" = $encodedUserUuid',
|
||||
transaction: transaction,
|
||||
);
|
||||
|
||||
final grantsByUuid = <String, CrdtScopeGrant>{};
|
||||
for (final row in result) {
|
||||
final scopeUuid = Protocol().deserialize<UuidValue>(row[0]);
|
||||
grantsByUuid[scopeUuid.uuid] = CrdtScopeGrant(
|
||||
uuidScopeId: scopeUuid,
|
||||
role: _roleFromDatabase(row[1]),
|
||||
);
|
||||
}
|
||||
// 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 encodedUserUuid = ValueEncoder.instance.convert(userUuid);
|
||||
final encodedScopeId = ValueEncoder.instance.convert(scopeId);
|
||||
final result = await session.db.unsafeQuery(
|
||||
'SELECT "${CrdtScopeMember.t.role.columnName}" '
|
||||
'FROM "${CrdtScopeMember.t.tableName}" '
|
||||
'WHERE "${CrdtScopeMember.t.userUuid.columnName}" = $encodedUserUuid '
|
||||
'AND "${CrdtScopeMember.t.scopeId.columnName}" = $encodedScopeId '
|
||||
'LIMIT 1',
|
||||
transaction: transaction,
|
||||
);
|
||||
return result.isEmpty ? null : _roleFromDatabase(result.first[0]);
|
||||
}
|
||||
|
||||
/// Reconciles the local `crdt_scope_members` cache from an authoritative
|
||||
/// [grants] announcement — a follower's read-only projection of its own
|
||||
/// memberships.
|
||||
///
|
||||
/// Upserts a row per shared grant (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 encodedUserUuid = ValueEncoder.instance.convert(userUuid);
|
||||
final keptScopeIds = <int>[];
|
||||
|
||||
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);
|
||||
final encodedScopeId = ValueEncoder.instance.convert(scopeId);
|
||||
final encodedRole = ValueEncoder.instance.convert(grant.role.toJson());
|
||||
await session.db.unsafeExecute(
|
||||
'INSERT INTO "${CrdtScopeMember.t.tableName}" '
|
||||
'("${CrdtScopeMember.t.scopeId.columnName}", '
|
||||
'"${CrdtScopeMember.t.userUuid.columnName}", '
|
||||
'"${CrdtScopeMember.t.role.columnName}") '
|
||||
'VALUES ($encodedScopeId, $encodedUserUuid, $encodedRole) '
|
||||
'ON CONFLICT ("${CrdtScopeMember.t.userUuid.columnName}", '
|
||||
'"${CrdtScopeMember.t.scopeId.columnName}") '
|
||||
'DO UPDATE SET "${CrdtScopeMember.t.role.columnName}" = $encodedRole',
|
||||
);
|
||||
}
|
||||
|
||||
final notInClause = keptScopeIds.isEmpty
|
||||
? ''
|
||||
: 'AND "${CrdtScopeMember.t.scopeId.columnName}" NOT IN '
|
||||
'(${keptScopeIds.map(ValueEncoder.instance.convert).join(', ')})';
|
||||
await session.db.unsafeExecute(
|
||||
'DELETE FROM "${CrdtScopeMember.t.tableName}" '
|
||||
'WHERE "${CrdtScopeMember.t.userUuid.columnName}" = $encodedUserUuid '
|
||||
'$notInClause',
|
||||
);
|
||||
}
|
||||
|
||||
/// 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;
|
||||
}
|
||||
}
|
||||
|
||||
CrdtScopeRole _roleFromDatabase(Object? value) {
|
||||
if (value == null) {
|
||||
throw StateError('crdt_scope_members.role must not be null.');
|
||||
}
|
||||
return CrdtScopeRole.fromJson(value as String);
|
||||
}
|
||||
|
||||
Future<int?> _scopeIdForUuid(
|
||||
DatabaseSession session,
|
||||
UuidValue scopeUuid, {
|
||||
Transaction? transaction,
|
||||
}) async {
|
||||
final encodedScopeUuid = ValueEncoder.instance.convert(scopeUuid);
|
||||
final result = await session.db.unsafeQuery(
|
||||
'SELECT "${CrdtScope.t.id.columnName}" FROM "${CrdtScope.t.tableName}" '
|
||||
'WHERE "${CrdtScope.t.uuidScopeId.columnName}" = $encodedScopeUuid '
|
||||
'LIMIT 1',
|
||||
transaction: transaction,
|
||||
);
|
||||
return result.isEmpty ? null : result.first[0] as int;
|
||||
}
|
||||
|
|
@ -0,0 +1,273 @@
|
|||
import 'package:serverpod_database/serverpod_database.dart';
|
||||
import 'package:serverpod_offline_sync_shared/serverpod_offline_sync_shared.dart';
|
||||
import 'package:uuid/uuid.dart';
|
||||
|
||||
import '../managers/scope.dart';
|
||||
import '../protocol/protocol.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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -13,9 +13,14 @@ import 'exceptions.dart';
|
|||
import 'extensions.dart';
|
||||
import 'integrity_violation.dart';
|
||||
import 'merge.dart';
|
||||
import 'scope_membership.dart';
|
||||
import 'scope_sync.dart';
|
||||
|
||||
export 'scope_sync.dart' show CrdtSyncPeerMode;
|
||||
|
||||
/// Callback function for when a merge is successful.
|
||||
typedef CrdtSyncOnMergeSuccess = FutureOr<void> Function(Hlc syncedHlc);
|
||||
typedef CrdtSyncOnMergeSuccess =
|
||||
FutureOr<void> Function(UuidValue scopeUuid, Hlc syncedHlc);
|
||||
|
||||
/// A tuple representing the ownership of a domain row.
|
||||
typedef DomainRowOwner = ({bool exists, int? scopeId});
|
||||
|
|
@ -123,7 +128,7 @@ class CrdtSync {
|
|||
'${uuid.v5(Namespace.oid.value, canonicalSignature)}';
|
||||
}
|
||||
|
||||
/// Streams local CRDT changes that a peer node has not seen yet.
|
||||
/// Streams pending changes for every scope in [checkpointsByScopeUuid].
|
||||
///
|
||||
/// Changes are emitted in insert, update, then delete order. Domain row and
|
||||
/// column payloads are resolved incrementally as each change is yielded.
|
||||
|
|
@ -133,20 +138,39 @@ class CrdtSync {
|
|||
/// in the domain table. Peers need the attempted fact to converge; local FK
|
||||
/// projection materializes only the safe visible value into domain tables.
|
||||
///
|
||||
/// All changes for nodes that are not present in the [nodeCheckpoints] list
|
||||
/// are collected and emitted. Passing an empty list will collect all changes.
|
||||
/// Resolves the scopes' internal ids once and keeps their checkpoint vectors
|
||||
/// scoped by those internal ids. Node ids are stable per replica and may
|
||||
/// appear in multiple scopes, so checkpoint filtering must compare both
|
||||
/// `scopeId` and `uuidNodeId`. This still runs one query per change kind for
|
||||
/// the whole pass. Per-row ownership and integrity checks resolve against
|
||||
/// each row's own scope.
|
||||
///
|
||||
/// All changes for nodes that are not present in a scope's checkpoint list
|
||||
/// are collected and emitted. Passing an empty list for a scope will collect
|
||||
/// all of its changes. Each change carries its [CrdtMergeChange.uuidScopeId].
|
||||
Stream<CrdtMergeChange> collectPendingChanges(
|
||||
DatabaseSession session, {
|
||||
required UuidValue userId,
|
||||
required UuidValue peerNodeId,
|
||||
required List<Hlc> nodeCheckpoints,
|
||||
required Map<UuidValue, List<Hlc>> checkpointsByScopeUuid,
|
||||
}) async* {
|
||||
final crdtUser = await CrdtScopeManager(session).getOrCreate(userId);
|
||||
if (checkpointsByScopeUuid.isEmpty) return;
|
||||
|
||||
final scopes = await CrdtScope.db.find(
|
||||
session,
|
||||
where: (t) => t.uuidScopeId.inSet(checkpointsByScopeUuid.keys.toSet()),
|
||||
);
|
||||
final scopeUuidById = {
|
||||
for (final scope in scopes) scope.id!: scope.uuidScopeId,
|
||||
};
|
||||
final checkpointsByScopeId = {
|
||||
for (final scope in scopes)
|
||||
scope.id!: checkpointsByScopeUuid[scope.uuidScopeId] ?? const <Hlc>[],
|
||||
};
|
||||
|
||||
try {
|
||||
await for (final change in _streamPendingChanges(
|
||||
session,
|
||||
crdtUser,
|
||||
nodeCheckpoints,
|
||||
scopeUuidById,
|
||||
checkpointsByScopeId,
|
||||
)) {
|
||||
yield change;
|
||||
}
|
||||
|
|
@ -155,29 +179,31 @@ class CrdtSync {
|
|||
}
|
||||
}
|
||||
|
||||
/// Creates the [CrdtSyncSinceHlc] checkpoint for the post-connect handshake.
|
||||
/// Creates the [CrdtSyncSinceHlc] checkpoint for a scope handshake.
|
||||
///
|
||||
/// [CrdtSyncSinceHlc.nodeCheckpoints] reflects the latest change this node
|
||||
/// has received from each known node, tagged with the source node id.
|
||||
Future<CrdtSyncSinceHlc> createSyncSinceHlc(
|
||||
DatabaseSession session, {
|
||||
required UuidValue userId,
|
||||
required UuidValue peerNodeId,
|
||||
required UuidValue scopeId,
|
||||
}) async {
|
||||
final crdtUser = await CrdtScopeManager(session).getOrCreate(userId);
|
||||
final scope = await CrdtScopeManager(session).getOrCreate(scopeId);
|
||||
final localNodeId = scope.currentNode!.uuidNodeId;
|
||||
|
||||
final nodes = await CrdtNode.db.find(
|
||||
final scopeNodes = await CrdtScopeNode.db.find(
|
||||
session,
|
||||
where: (t) =>
|
||||
t.scopeId.equals(crdtUser.id) &
|
||||
t.uuidNodeId.notEquals(crdtUser.currentNode!.uuidNodeId),
|
||||
t.scopeId.equals(scope.id) & t.nodeId.notEquals(scope.currentNodeId),
|
||||
include: CrdtScopeNode.include(node: CrdtNode.include()),
|
||||
);
|
||||
|
||||
return CrdtSyncSinceHlc(
|
||||
uuidScopeId: scopeId,
|
||||
nodeCheckpoints: [
|
||||
// The local node is always included to avoid collecting its own changes.
|
||||
Hlc.now(crdtUser.currentNode!.uuidNodeId),
|
||||
for (final node in nodes) node.lastReceivedHlc ?? Hlc.zero(node.uuidNodeId),
|
||||
Hlc.now(localNodeId),
|
||||
for (final scopeNode in scopeNodes)
|
||||
scopeNode.lastReceivedHlc ?? Hlc.zero(scopeNode.node!.uuidNodeId),
|
||||
],
|
||||
);
|
||||
}
|
||||
|
|
@ -192,114 +218,111 @@ class CrdtSync {
|
|||
///
|
||||
/// Returns the greatest HLC synced in the batch, or `null` if the batch is
|
||||
/// empty.
|
||||
Future<Hlc?> mergeInboundBatch(
|
||||
Future<Hlc?> _mergeInboundBatch(
|
||||
DatabaseSession session, {
|
||||
required UuidValue userId,
|
||||
required UuidValue scopeId,
|
||||
required UuidValue otherNodeId,
|
||||
required CrdtMergeSet mergeSet,
|
||||
}) async {
|
||||
if (mergeSet.isEmpty) return null;
|
||||
final maxSyncedHlc = mergeSet.maxHlc;
|
||||
final crdtDb = _openCrdtDatabase(session);
|
||||
await crdtDb.mergeChanges(mergeSet, scopeId: userId);
|
||||
await crdtDb.mergeChanges(mergeSet, scopeId: scopeId);
|
||||
if (maxSyncedHlc != null) {
|
||||
await crdtDb.recordSyncCheckpoint(otherNodeId, maxSyncedHlc, userId: userId);
|
||||
await crdtDb.recordSyncCheckpoint(otherNodeId, maxSyncedHlc, userId: scopeId);
|
||||
}
|
||||
return maxSyncedHlc;
|
||||
}
|
||||
|
||||
/// Streams a framed outbound sync batch of merge changes.
|
||||
Stream<CrdtSyncStreamEvent> streamOutboundBatch(
|
||||
DatabaseSession session, {
|
||||
required UuidValue userId,
|
||||
required UuidValue peerNodeId,
|
||||
required List<Hlc> nodeCheckpoints,
|
||||
}) async* {
|
||||
yield* collectPendingChanges(
|
||||
session,
|
||||
userId: userId,
|
||||
peerNodeId: peerNodeId,
|
||||
nodeCheckpoints: nodeCheckpoints,
|
||||
)
|
||||
.chunked(_syncBatchSize)
|
||||
.map(
|
||||
(changes) => CrdtSyncMergeChunk(changes: changes),
|
||||
);
|
||||
yield CrdtSyncEndOfBatch();
|
||||
}
|
||||
|
||||
/// Runs a symmetric CRDT sync session over a bidirectional event stream.
|
||||
///
|
||||
/// Both peers send [CrdtSyncConnect], read the peer connect frame, then send
|
||||
/// and read [CrdtSyncSinceHlc] once. After that, both sides exchange framed
|
||||
/// data batches using the peer's per-node checkpoints. Sent changes advance
|
||||
/// the corresponding in-memory checkpoints for subsequent sends within the
|
||||
/// session.
|
||||
/// Both peers exchange [CrdtSyncConnect] and a lockstep [CrdtSyncScopeSet]
|
||||
/// before the data loop. Each cycle sends a combined batch — scope
|
||||
/// announcement when grants changed, [CrdtSyncSinceHlc] for newly active
|
||||
/// scopes, and [CrdtSyncMergeChunk]s — closed by [CrdtSyncEndOfBatch] when
|
||||
/// anything was sent. Inbound frames are de-multiplexed by collectNextBatch
|
||||
/// until [CrdtSyncEndOfBatch] when this peer sent a batch, or an idle timeout
|
||||
/// when both peers had nothing to send.
|
||||
///
|
||||
/// If a batch fails to merge, the exception propagates and the stream closes.
|
||||
/// The next sync attempt resumes from the last persisted checkpoint on the
|
||||
/// side that failed to merge.
|
||||
///
|
||||
/// When [once] is true, each peer sends one batch, reads one batch, sends a
|
||||
/// [CrdtSyncClose] frame and awaits for the other to send its closing event
|
||||
/// to complete the session.
|
||||
/// When [once] is true the loop may run an extra cycle after handshakes
|
||||
/// complete so merge data can flow; then it closes symmetrically. Continuous
|
||||
/// mode loops with [_continuousSyncInterval] between idle cycles.
|
||||
Stream<CrdtSyncStreamEvent> sync(
|
||||
DatabaseSession session, {
|
||||
required UuidValue userId,
|
||||
required Stream<CrdtSyncStreamEvent> inbound,
|
||||
required CrdtSyncPeerMode mode,
|
||||
bool once = false,
|
||||
CrdtSyncOnMergeSuccess? onMergeSuccess,
|
||||
}) async* {
|
||||
// Idle timeouts are a continuous-only affordance: they let an idle cycle
|
||||
// settle into an empty batch without closing the stream. A `once` session
|
||||
// is strictly lockstep — every batch ends with a [CrdtSyncEndOfBatch] and
|
||||
// the session with a [CrdtSyncClose] — so it must wait for those end frames
|
||||
// rather than truncate a slow peer's batch on a timeout.
|
||||
final inboundIterator = StreamIterator(
|
||||
// Inject a timeout event if the inbound stream is idle for too long to
|
||||
// allow returning an empty batch and waiting for a change without closing
|
||||
// the stream.
|
||||
inbound.timeout(
|
||||
const Duration(seconds: 1),
|
||||
onTimeout: (sink) => sink.add(CrdtSyncIdleTimeout()),
|
||||
),
|
||||
once
|
||||
? inbound
|
||||
: inbound.timeout(
|
||||
const Duration(seconds: 1),
|
||||
onTimeout: (sink) => sink.add(CrdtSyncIdleTimeout()),
|
||||
),
|
||||
);
|
||||
|
||||
var sessionCompleted = false;
|
||||
try {
|
||||
final crdtUser = await CrdtScopeManager(session).getOrCreate(userId);
|
||||
final localNodeId = crdtUser.currentNode!.uuidNodeId;
|
||||
|
||||
final scope = await CrdtScopeManager(session).getOrCreate(userId);
|
||||
final localNodeId = scope.currentNode!.uuidNodeId;
|
||||
yield CrdtSyncConnect(
|
||||
localNodeId: localNodeId,
|
||||
syncTablesHash: currentSyncTablesHash,
|
||||
);
|
||||
|
||||
final peerConnect = await inboundIterator.moveAndThrowIfNot<CrdtSyncConnect>();
|
||||
final peerNodeId = peerConnect.localNodeId;
|
||||
_validateSyncTablesHash(peerConnect.syncTablesHash);
|
||||
|
||||
yield await createSyncSinceHlc(
|
||||
final scopes = CrdtScopeSyncSession(
|
||||
session,
|
||||
userId: userId,
|
||||
peerNodeId: peerNodeId,
|
||||
mode: mode,
|
||||
peerNodeId: peerConnect.localNodeId,
|
||||
);
|
||||
|
||||
final peerSinceHlc = await inboundIterator.moveAndThrowIfNot<CrdtSyncSinceHlc>();
|
||||
final nodeCheckpoints = {
|
||||
for (final checkpoint in peerSinceHlc.nodeCheckpoints)
|
||||
checkpoint.nodeId: checkpoint,
|
||||
};
|
||||
await scopes.reconcile();
|
||||
yield CrdtSyncScopeSet(scopes: scopes.localGrants);
|
||||
scopes.markAnnounced();
|
||||
final peerScopeSet = await inboundIterator.moveAndThrowIfNot<CrdtSyncScopeSet>();
|
||||
await scopes.adoptPeerGrants(peerScopeSet.scopes);
|
||||
|
||||
while (true) {
|
||||
await scopes.reconcile();
|
||||
|
||||
final hadSendableCheckpoints = scopes.sendableCheckpoints.isNotEmpty;
|
||||
var hasChanges = false;
|
||||
final outboundScopes = <UuidValue>{};
|
||||
|
||||
if (scopes.shouldAnnounce) {
|
||||
yield CrdtSyncScopeSet(scopes: scopes.localGrants);
|
||||
scopes.markAnnounced();
|
||||
hasChanges = true;
|
||||
}
|
||||
|
||||
for (final scopeId in scopes.activeScopeIds) {
|
||||
if (!scopes.markHandshakeSent(scopeId)) continue;
|
||||
yield await createSyncSinceHlc(session, scopeId: scopeId);
|
||||
hasChanges = true;
|
||||
}
|
||||
|
||||
final pendingLocalChanges = collectPendingChanges(
|
||||
session,
|
||||
userId: userId,
|
||||
peerNodeId: peerNodeId,
|
||||
nodeCheckpoints: nodeCheckpoints.values.toList(),
|
||||
checkpointsByScopeUuid: scopes.sendableCheckpoints,
|
||||
);
|
||||
|
||||
var hasChanges = false;
|
||||
await for (final changes in pendingLocalChanges.chunked(_syncBatchSize)) {
|
||||
hasChanges = true;
|
||||
for (final change in changes) {
|
||||
final lastCheckpoint = nodeCheckpoints[change.uuidNodeId];
|
||||
nodeCheckpoints[change.uuidNodeId] = change.hlc.maxBetween(lastCheckpoint);
|
||||
scopes.advanceCheckpoint(change.uuidScopeId, change);
|
||||
outboundScopes.add(change.uuidScopeId);
|
||||
}
|
||||
yield CrdtSyncMergeChunk(changes: changes);
|
||||
}
|
||||
|
|
@ -307,28 +330,32 @@ class CrdtSync {
|
|||
yield CrdtSyncEndOfBatch();
|
||||
}
|
||||
|
||||
final inboundMergeSet = await inboundIterator.collectNextBatch(
|
||||
final batch = await inboundIterator.collectNextBatch(
|
||||
allowCloseBeforeBatch: !once,
|
||||
);
|
||||
if (inboundMergeSet == null) {
|
||||
if (batch == null) {
|
||||
if (once) {
|
||||
yield CrdtSyncClose();
|
||||
}
|
||||
sessionCompleted = true;
|
||||
return;
|
||||
}
|
||||
|
||||
final lastReceivedFromPeer = await mergeInboundBatch(
|
||||
await _applyCycleBatch(
|
||||
session,
|
||||
userId: userId,
|
||||
otherNodeId: peerNodeId,
|
||||
mergeSet: inboundMergeSet,
|
||||
scopes,
|
||||
batch,
|
||||
outboundScopes,
|
||||
onMergeSuccess,
|
||||
);
|
||||
|
||||
if (hasChanges || inboundMergeSet.isNotEmpty) {
|
||||
await onMergeSuccess?.call(
|
||||
nodeCheckpoints.values.max.maxBetween(lastReceivedFromPeer),
|
||||
);
|
||||
}
|
||||
|
||||
if (once) {
|
||||
if (scopes.hasIncompleteActiveHandshake) {
|
||||
continue;
|
||||
}
|
||||
if (!hadSendableCheckpoints && scopes.sendableCheckpoints.isNotEmpty) {
|
||||
continue;
|
||||
}
|
||||
yield CrdtSyncClose();
|
||||
await inboundIterator.moveAndThrowIfNot<CrdtSyncClose>();
|
||||
sessionCompleted = true;
|
||||
|
|
@ -349,6 +376,17 @@ class CrdtSync {
|
|||
// Wait for the configured interval before checking for local changes again.
|
||||
await Future<void>.delayed(_continuousSyncInterval);
|
||||
}
|
||||
} on CrdtSyncStreamClosedException {
|
||||
// A continuous session ending is normal: the peer closed its outbound
|
||||
// (typically a cancel). Whether the stream closed between batches or
|
||||
// mid-batch, nothing is left to do — complete gracefully so the `finally`
|
||||
// skips the force-cancel that races WebSocket teardown and surfaces
|
||||
// spurious "connection closed" errors on the peer. A `once` session has a
|
||||
// completion contract (full handshake plus the symmetric Close), so a
|
||||
// close before that is a real truncation and must propagate.
|
||||
if (once) rethrow;
|
||||
sessionCompleted = true;
|
||||
return;
|
||||
} finally {
|
||||
// Cancelling inbound on normal completion races with WebSocket stream
|
||||
// teardown and produces "connection closed" errors on the peer. On normal
|
||||
|
|
@ -363,6 +401,112 @@ class CrdtSync {
|
|||
}
|
||||
}
|
||||
|
||||
Future<void> _applyCycleBatch(
|
||||
DatabaseSession session,
|
||||
CrdtScopeSyncSession scopes,
|
||||
CrdtSyncCycleBatch batch,
|
||||
Set<UuidValue> outboundScopes,
|
||||
CrdtSyncOnMergeSuccess? onMergeSuccess,
|
||||
) async {
|
||||
if (batch.scopeSet != null) {
|
||||
await scopes.adoptPeerGrants(batch.scopeSet!.scopes);
|
||||
}
|
||||
for (final entry in batch.sinceHlcs.entries) {
|
||||
if (scopes.accepts(entry.key)) {
|
||||
scopes.recordPeerHandshake(entry.key, entry.value);
|
||||
}
|
||||
}
|
||||
|
||||
final mergedScopes = <UuidValue>{};
|
||||
final changesByScope = <UuidValue, List<CrdtMergeChange>>{};
|
||||
for (final change in batch.changes) {
|
||||
changesByScope.putIfAbsent(change.uuidScopeId, () => []).add(change);
|
||||
}
|
||||
for (final entry in changesByScope.entries) {
|
||||
final scopeId = entry.key;
|
||||
if (!scopes.accepts(scopeId)) continue;
|
||||
mergedScopes.add(scopeId);
|
||||
if (scopes.isAuthoritative) {
|
||||
await _assertCanMergeInboundScope(
|
||||
session,
|
||||
scopeId: scopeId,
|
||||
userId: scopes.userId,
|
||||
changes: entry.value,
|
||||
);
|
||||
}
|
||||
final receivedHlc = await _mergeInboundBatch(
|
||||
session,
|
||||
scopeId: scopeId,
|
||||
otherNodeId: scopes.peerNodeId,
|
||||
mergeSet: entry.value,
|
||||
);
|
||||
await _reportMerge(onMergeSuccess, scopes, scopeId, receivedHlc);
|
||||
}
|
||||
for (final scopeId in outboundScopes.difference(mergedScopes)) {
|
||||
await _reportMerge(onMergeSuccess, scopes, scopeId, null);
|
||||
}
|
||||
}
|
||||
|
||||
Future<void> _assertCanMergeInboundScope(
|
||||
DatabaseSession session, {
|
||||
required UuidValue scopeId,
|
||||
required UuidValue userId,
|
||||
required List<CrdtMergeChange> changes,
|
||||
}) async {
|
||||
if (changes.isEmpty || userId == scopeId) return;
|
||||
|
||||
final role = await CrdtScopeMembership.roleOf(
|
||||
session,
|
||||
userUuid: userId,
|
||||
scopeUuid: scopeId,
|
||||
);
|
||||
if (role.canWrite) return;
|
||||
|
||||
final firstChange = changes.first;
|
||||
final now = DateTime.now().toUtc();
|
||||
final violation = CrdtSyncIntegrityViolation(
|
||||
type: CrdtSyncViolationType.unauthorizedWrite,
|
||||
domainTableName: firstChange.tableName,
|
||||
uuidRowId: firstChange.uuidRowId,
|
||||
ownerScopeUuid: null,
|
||||
incomingScopeUuid: scopeId,
|
||||
operation: _operationForInboundChange(firstChange),
|
||||
uuidNodeId: firstChange.uuidNodeId,
|
||||
crdtDataRowId: null,
|
||||
hlcDatetime: firstChange.hlcDatetime,
|
||||
hlcCounter: firstChange.hlcCounter,
|
||||
firstSeenAt: now,
|
||||
lastSeenAt: now,
|
||||
occurrences: 1,
|
||||
);
|
||||
final persisted = await recordCrdtSyncIntegrityViolation(
|
||||
session,
|
||||
violation: violation,
|
||||
);
|
||||
throw CrdtSyncIntegrityViolationException(persisted);
|
||||
}
|
||||
|
||||
CrdtSyncViolationOperation _operationForInboundChange(CrdtMergeChange change) {
|
||||
return switch (change) {
|
||||
CrdtMergeInsert() => CrdtSyncViolationOperation.mergeInsert,
|
||||
CrdtMergeUpdate() => CrdtSyncViolationOperation.mergeUpdate,
|
||||
CrdtMergeDelete() => CrdtSyncViolationOperation.mergeDelete,
|
||||
};
|
||||
}
|
||||
|
||||
/// Reports a successful merge for [scopeId] to [onMergeSuccess], combining the
|
||||
/// scope's checkpoint high-water mark with the [receivedHlc] just merged.
|
||||
Future<void> _reportMerge(
|
||||
CrdtSyncOnMergeSuccess? onMergeSuccess,
|
||||
CrdtScopeSyncSession scopes,
|
||||
UuidValue scopeId,
|
||||
Hlc? receivedHlc,
|
||||
) async {
|
||||
final checkpointMax = scopes.checkpointMaxOf(scopeId);
|
||||
if (checkpointMax == null) return;
|
||||
await onMergeSuccess?.call(scopeId, checkpointMax.maxBetween(receivedHlc));
|
||||
}
|
||||
|
||||
/// Drains [iterator] until the peer closes the stream.
|
||||
///
|
||||
/// Used to settle the inbound transport after the `once` close handshake so
|
||||
|
|
@ -400,26 +544,26 @@ class CrdtSync {
|
|||
|
||||
Stream<CrdtMergeChange> _streamPendingChanges(
|
||||
DatabaseSession session,
|
||||
CrdtScope crdtUser,
|
||||
List<Hlc> nodeCheckpoints,
|
||||
Map<int, UuidValue> scopeUuidById,
|
||||
Map<int, List<Hlc>> checkpointsByScopeId,
|
||||
) async* {
|
||||
// Domain ownership is immutable while a collection runs, so read each
|
||||
// row's owner at most once across all three streams.
|
||||
final ownerCache = DomainRowOwnerCache();
|
||||
yield* _streamInserts(session, crdtUser, nodeCheckpoints, ownerCache);
|
||||
yield* _streamUpdates(session, crdtUser, nodeCheckpoints, ownerCache);
|
||||
yield* _streamDeletes(session, crdtUser, nodeCheckpoints, ownerCache);
|
||||
yield* _streamInserts(session, scopeUuidById, checkpointsByScopeId, ownerCache);
|
||||
yield* _streamUpdates(session, scopeUuidById, checkpointsByScopeId, ownerCache);
|
||||
yield* _streamDeletes(session, scopeUuidById, checkpointsByScopeId, ownerCache);
|
||||
}
|
||||
|
||||
Stream<CrdtMergeInsert> _streamInserts(
|
||||
DatabaseSession session,
|
||||
CrdtScope crdtUser,
|
||||
List<Hlc> nodeCheckpoints,
|
||||
Map<int, UuidValue> scopeUuidById,
|
||||
Map<int, List<Hlc>> checkpointsByScopeId,
|
||||
DomainRowOwnerCache ownerCache,
|
||||
) async* {
|
||||
final rows = await CrdtDataRow.db.find(
|
||||
session,
|
||||
where: (t) => _rowHlcAfterFilter(t, crdtUser, nodeCheckpoints),
|
||||
where: (t) => _rowHlcAfterFilter(t, checkpointsByScopeId),
|
||||
include: CrdtDataRow.include(
|
||||
tbl: CrdtSchemaTable.include(),
|
||||
node: CrdtNode.include(),
|
||||
|
|
@ -439,6 +583,9 @@ class CrdtSync {
|
|||
final dartName = _classNamesByTableName[tableName];
|
||||
if (dartName == null) continue;
|
||||
|
||||
final scopeId = row.scopeId;
|
||||
final scopeUuid = scopeUuidById[scopeId]!;
|
||||
|
||||
final domainRow = await _fetchDomainRow(
|
||||
session,
|
||||
tableName,
|
||||
|
|
@ -446,7 +593,7 @@ class CrdtSync {
|
|||
table,
|
||||
dartName,
|
||||
foreignKeyAttemptFieldsByRowId[row.id!],
|
||||
crdtUser.id!,
|
||||
scopeId,
|
||||
ownerCache,
|
||||
);
|
||||
if (!domainRow.exists) {
|
||||
|
|
@ -457,12 +604,12 @@ class CrdtSync {
|
|||
tableName: tableName,
|
||||
rowId: row.uuidRowId,
|
||||
ownerScopeId: null,
|
||||
incomingScopeUuid: crdtUser.uuidScopeId,
|
||||
incomingScopeUuid: scopeUuid,
|
||||
uuidNodeId: row.node!.uuidNodeId,
|
||||
hlc: row.hlc,
|
||||
);
|
||||
}
|
||||
if (domainRow.ownerScopeId != crdtUser.id) {
|
||||
if (domainRow.ownerScopeId != scopeId) {
|
||||
_throwPendingIntegrityViolation(
|
||||
crdtDataRowId: row.id,
|
||||
type: CrdtSyncViolationType.ownershipCollision,
|
||||
|
|
@ -470,13 +617,14 @@ class CrdtSync {
|
|||
tableName: tableName,
|
||||
rowId: row.uuidRowId,
|
||||
ownerScopeId: domainRow.ownerScopeId,
|
||||
incomingScopeUuid: crdtUser.uuidScopeId,
|
||||
incomingScopeUuid: scopeUuid,
|
||||
uuidNodeId: row.node!.uuidNodeId,
|
||||
hlc: row.hlc,
|
||||
);
|
||||
}
|
||||
|
||||
yield CrdtMergeInsert(
|
||||
uuidScopeId: scopeUuid,
|
||||
hlcDatetime: row.hlcDatetime,
|
||||
hlcCounter: row.hlcCounter,
|
||||
tableName: tableName,
|
||||
|
|
@ -489,13 +637,13 @@ class CrdtSync {
|
|||
|
||||
Stream<CrdtMergeUpdate> _streamUpdates(
|
||||
DatabaseSession session,
|
||||
CrdtScope crdtUser,
|
||||
List<Hlc> nodeCheckpoints,
|
||||
Map<int, UuidValue> scopeUuidById,
|
||||
Map<int, List<Hlc>> checkpointsByScopeId,
|
||||
DomainRowOwnerCache ownerCache,
|
||||
) async* {
|
||||
final fields = await CrdtDataField.db.find(
|
||||
session,
|
||||
where: (t) => _fieldHlcAfterFilter(t, crdtUser, nodeCheckpoints),
|
||||
where: (t) => _fieldHlcAfterFilter(t, checkpointsByScopeId),
|
||||
include: CrdtDataField.include(
|
||||
row: CrdtDataRow.include(tbl: CrdtSchemaTable.include()),
|
||||
column: CrdtSchemaColumn.include(),
|
||||
|
|
@ -513,6 +661,8 @@ class CrdtSync {
|
|||
continue;
|
||||
}
|
||||
|
||||
final scopeId = field.row!.scopeId;
|
||||
final scopeUuid = scopeUuidById[scopeId]!;
|
||||
final columnName = field.column!.name;
|
||||
final columnValue = await _fetchOwnedColumnValue(
|
||||
session,
|
||||
|
|
@ -520,7 +670,7 @@ class CrdtSync {
|
|||
field.row!.uuidRowId,
|
||||
columnName,
|
||||
field.foreignKey,
|
||||
crdtUser.id!,
|
||||
scopeId,
|
||||
ownerCache,
|
||||
);
|
||||
if (!columnValue.exists) {
|
||||
|
|
@ -531,12 +681,12 @@ class CrdtSync {
|
|||
tableName: tableName,
|
||||
rowId: field.row!.uuidRowId,
|
||||
ownerScopeId: null,
|
||||
incomingScopeUuid: crdtUser.uuidScopeId,
|
||||
incomingScopeUuid: scopeUuid,
|
||||
uuidNodeId: field.node!.uuidNodeId,
|
||||
hlc: field.hlc,
|
||||
);
|
||||
}
|
||||
if (columnValue.ownerScopeId != crdtUser.id) {
|
||||
if (columnValue.ownerScopeId != scopeId) {
|
||||
_throwPendingIntegrityViolation(
|
||||
crdtDataRowId: field.row!.id,
|
||||
type: CrdtSyncViolationType.ownershipCollision,
|
||||
|
|
@ -544,13 +694,14 @@ class CrdtSync {
|
|||
tableName: tableName,
|
||||
rowId: field.row!.uuidRowId,
|
||||
ownerScopeId: columnValue.ownerScopeId,
|
||||
incomingScopeUuid: crdtUser.uuidScopeId,
|
||||
incomingScopeUuid: scopeUuid,
|
||||
uuidNodeId: field.node!.uuidNodeId,
|
||||
hlc: field.hlc,
|
||||
);
|
||||
}
|
||||
|
||||
yield CrdtMergeUpdate(
|
||||
uuidScopeId: scopeUuid,
|
||||
hlcDatetime: field.hlcDatetime,
|
||||
hlcCounter: field.hlcCounter,
|
||||
tableName: tableName,
|
||||
|
|
@ -564,13 +715,13 @@ class CrdtSync {
|
|||
|
||||
Stream<CrdtMergeDelete> _streamDeletes(
|
||||
DatabaseSession session,
|
||||
CrdtScope crdtUser,
|
||||
List<Hlc> nodeCheckpoints,
|
||||
Map<int, UuidValue> scopeUuidById,
|
||||
Map<int, List<Hlc>> checkpointsByScopeId,
|
||||
DomainRowOwnerCache ownerCache,
|
||||
) async* {
|
||||
final tombstones = await CrdtDataDeleted.db.find(
|
||||
session,
|
||||
where: (t) => _tombstoneHlcAfterFilter(t, crdtUser, nodeCheckpoints),
|
||||
where: (t) => _tombstoneHlcAfterFilter(t, checkpointsByScopeId),
|
||||
include: CrdtDataDeleted.include(
|
||||
row: CrdtDataRow.include(tbl: CrdtSchemaTable.include()),
|
||||
node: CrdtNode.include(),
|
||||
|
|
@ -582,13 +733,16 @@ class CrdtSync {
|
|||
|
||||
final tableName = tombstone.row!.tbl!.name;
|
||||
if (!_syncTablesByName.containsKey(tableName)) continue;
|
||||
|
||||
final scopeId = tombstone.row!.scopeId;
|
||||
final scopeUuid = scopeUuidById[scopeId]!;
|
||||
final owner = await _readDomainRowOwner(
|
||||
session,
|
||||
tableName,
|
||||
tombstone.row!.uuidRowId,
|
||||
ownerCache,
|
||||
);
|
||||
if (owner.exists && owner.scopeId != crdtUser.id) {
|
||||
if (owner.exists && owner.scopeId != scopeId) {
|
||||
_throwPendingIntegrityViolation(
|
||||
crdtDataRowId: tombstone.row!.id,
|
||||
type: CrdtSyncViolationType.ownershipCollision,
|
||||
|
|
@ -596,13 +750,14 @@ class CrdtSync {
|
|||
tableName: tableName,
|
||||
rowId: tombstone.row!.uuidRowId,
|
||||
ownerScopeId: owner.scopeId,
|
||||
incomingScopeUuid: crdtUser.uuidScopeId,
|
||||
incomingScopeUuid: scopeUuid,
|
||||
uuidNodeId: tombstone.node!.uuidNodeId,
|
||||
hlc: tombstone.hlc,
|
||||
);
|
||||
}
|
||||
|
||||
yield CrdtMergeDelete(
|
||||
uuidScopeId: scopeUuid,
|
||||
hlcDatetime: tombstone.hlcDatetime,
|
||||
hlcCounter: tombstone.hlcCounter,
|
||||
tableName: tableName,
|
||||
|
|
@ -616,41 +771,41 @@ class CrdtSync {
|
|||
|
||||
Expression _rowHlcAfterFilter(
|
||||
CrdtDataRowTable t,
|
||||
CrdtScope crdtUser,
|
||||
List<Hlc> nodeCheckpoints,
|
||||
Map<int, List<Hlc>> checkpointsByScopeId,
|
||||
) =>
|
||||
t.scopeId.equals(crdtUser.id) &
|
||||
t.node.afterAnyCheckpointFilter(
|
||||
t.scopeId.inSet(checkpointsByScopeId.keys.toSet()) &
|
||||
_afterAnyScopeCheckpointFilter(
|
||||
t.scopeId,
|
||||
t.node.uuidNodeId,
|
||||
t.hlcDatetime,
|
||||
t.hlcCounter,
|
||||
nodeCheckpoints,
|
||||
checkpointsByScopeId,
|
||||
);
|
||||
|
||||
Expression _fieldHlcAfterFilter(
|
||||
CrdtDataFieldTable t,
|
||||
CrdtScope crdtUser,
|
||||
List<Hlc> nodeCheckpoints,
|
||||
Map<int, List<Hlc>> checkpointsByScopeId,
|
||||
) =>
|
||||
t.row.scopeId.equals(crdtUser.id) &
|
||||
t.node.afterAnyCheckpointFilter(
|
||||
t.row.scopeId.inSet(checkpointsByScopeId.keys.toSet()) &
|
||||
_afterAnyScopeCheckpointFilter(
|
||||
t.row.scopeId,
|
||||
t.node.uuidNodeId,
|
||||
t.hlcDatetime,
|
||||
t.hlcCounter,
|
||||
nodeCheckpoints,
|
||||
checkpointsByScopeId,
|
||||
);
|
||||
|
||||
Expression _tombstoneHlcAfterFilter(
|
||||
CrdtDataDeletedTable t,
|
||||
CrdtScope crdtUser,
|
||||
List<Hlc> nodeCheckpoints,
|
||||
Map<int, List<Hlc>> checkpointsByScopeId,
|
||||
) =>
|
||||
t.row.scopeId.equals(crdtUser.id) &
|
||||
t.node.afterAnyCheckpointFilter(
|
||||
t.row.scopeId.inSet(checkpointsByScopeId.keys.toSet()) &
|
||||
_afterAnyScopeCheckpointFilter(
|
||||
t.row.scopeId,
|
||||
t.node.uuidNodeId,
|
||||
t.hlcDatetime,
|
||||
t.hlcCounter,
|
||||
nodeCheckpoints,
|
||||
checkpointsByScopeId,
|
||||
);
|
||||
|
||||
/// Loads a domain row for outbound insert sync.
|
||||
|
|
@ -959,27 +1114,31 @@ class CrdtSync {
|
|||
}
|
||||
}
|
||||
|
||||
extension on CrdtNodeTable {
|
||||
Expression afterAnyCheckpointFilter(
|
||||
ColumnUuid uuidNodeId,
|
||||
ColumnDateTime hlcDatetime,
|
||||
ColumnInt hlcCounter,
|
||||
List<Hlc> nodeCheckpoints,
|
||||
) {
|
||||
if (nodeCheckpoints.isEmpty) return Constant.bool(true);
|
||||
|
||||
final caseExpression = Case();
|
||||
for (final checkpoint in nodeCheckpoints) {
|
||||
Expression _afterAnyScopeCheckpointFilter(
|
||||
ColumnInt scopeId,
|
||||
ColumnUuid uuidNodeId,
|
||||
ColumnDateTime hlcDatetime,
|
||||
ColumnInt hlcCounter,
|
||||
Map<int, List<Hlc>> checkpointsByScopeId,
|
||||
) {
|
||||
final caseExpression = Case();
|
||||
var hasCheckpoint = false;
|
||||
for (final MapEntry(key: normalizedScopeId, value: checkpoints)
|
||||
in checkpointsByScopeId.entries) {
|
||||
for (final checkpoint in checkpoints) {
|
||||
hasCheckpoint = true;
|
||||
caseExpression.when(
|
||||
uuidNodeId.equals(checkpoint.nodeId),
|
||||
scopeId.equals(normalizedScopeId) & uuidNodeId.equals(checkpoint.nodeId),
|
||||
then:
|
||||
(hlcDatetime > checkpoint.datetime) |
|
||||
(hlcDatetime.equals(checkpoint.datetime) &
|
||||
(hlcCounter > checkpoint.counter)),
|
||||
);
|
||||
}
|
||||
return caseExpression.orElse(Constant.bool(true));
|
||||
}
|
||||
return hasCheckpoint
|
||||
? caseExpression.orElse(Constant.bool(true))
|
||||
: Constant.bool(true);
|
||||
}
|
||||
|
||||
extension on Map<String, dynamic> {
|
||||
|
|
|
|||
|
|
@ -2,13 +2,17 @@
|
|||
// flag internal Serverpod APIs used by generated code.
|
||||
// ignore_for_file: invalid_use_of_internal_member
|
||||
|
||||
import 'dart:async';
|
||||
|
||||
import 'package:serverpod_database/serverpod_database.dart';
|
||||
import 'package:serverpod_offline_sync_shared/serverpod_offline_sync_shared.dart';
|
||||
import 'package:uuid/uuid.dart';
|
||||
|
||||
import '../crdt/exceptions.dart';
|
||||
import '../crdt/extensions.dart';
|
||||
import '../crdt/integrity_violation.dart';
|
||||
import '../crdt/merge.dart';
|
||||
import '../crdt/scope_membership.dart';
|
||||
import '../crdt/sync.dart';
|
||||
import '../protocol/protocol.dart';
|
||||
import 'recorder.dart';
|
||||
|
|
@ -20,6 +24,9 @@ part 'scope.dart';
|
|||
/// Map of transaction hashes to the scope they are associated with.
|
||||
final scopeForTransaction = <Transaction, CrdtScope>{};
|
||||
|
||||
/// Map of transaction hashes to the authenticated user associated with them.
|
||||
final userForTransaction = <Transaction, UuidValue>{};
|
||||
|
||||
/// Database proxy that runs insert/update/delete ORM operations inside a
|
||||
/// transaction to record each change in the CRDT tables.
|
||||
class CrdtDatabase implements Database {
|
||||
|
|
@ -111,6 +118,7 @@ class CrdtDatabase implements Database {
|
|||
/// Runs a symmetric CRDT sync session over a bidirectional event stream.
|
||||
Stream<CrdtSyncStreamEvent> sync({
|
||||
required Stream<CrdtSyncStreamEvent> inbound,
|
||||
required CrdtSyncPeerMode mode,
|
||||
UuidValue? userId,
|
||||
bool once = false,
|
||||
CrdtSyncOnMergeSuccess? onMergeSuccess,
|
||||
|
|
@ -123,6 +131,7 @@ class CrdtDatabase implements Database {
|
|||
inbound: inbound,
|
||||
once: once,
|
||||
onMergeSuccess: onMergeSuccess,
|
||||
mode: mode,
|
||||
);
|
||||
}
|
||||
|
||||
|
|
@ -184,20 +193,26 @@ class CrdtDatabase implements Database {
|
|||
_delegate.serializationManager;
|
||||
|
||||
/// Merges [where] with the CRDT visibility predicates scoped to the queried
|
||||
/// tables and the user associated with [transaction] (or the persistent
|
||||
/// user). Without a user scope (e.g. server-side admin reads), a row is
|
||||
/// only hidden when every user tracking it has it hidden.
|
||||
Expression? _whereVisibleWithTombstone<T extends TableRow>(
|
||||
/// tables and the user associated with [transaction] (or the persistent user).
|
||||
///
|
||||
/// Read filters are membership-wide. Write filters stay pinned to the acting
|
||||
/// scope so mutations cannot cross scope boundaries.
|
||||
Future<Expression?> _whereVisibleWithTombstone<T extends TableRow>(
|
||||
Expression? where,
|
||||
Include? include,
|
||||
Transaction? transaction,
|
||||
) {
|
||||
Transaction? transaction, {
|
||||
required bool membershipWide,
|
||||
}) async {
|
||||
final scopeIds = await _scopeIdsForQueries(
|
||||
transaction,
|
||||
membershipWide: membershipWide,
|
||||
);
|
||||
return mergeWhereWithTombstone<T>(
|
||||
serializationManager,
|
||||
where,
|
||||
include,
|
||||
tableIdForName: _recorder.tableIdForName,
|
||||
scopeId: () => _recorder.scopeForQueries(transaction)?.id,
|
||||
scopeIds: () => scopeIds,
|
||||
);
|
||||
}
|
||||
|
||||
|
|
@ -216,7 +231,12 @@ class CrdtDatabase implements Database {
|
|||
}) async {
|
||||
await _ensureInitialized();
|
||||
final result = await _delegate.find<T>(
|
||||
where: _whereVisibleWithTombstone<T>(where, include, transaction),
|
||||
where: await _whereVisibleWithTombstone<T>(
|
||||
where,
|
||||
include,
|
||||
transaction,
|
||||
membershipWide: true,
|
||||
),
|
||||
limit: limit,
|
||||
offset: offset,
|
||||
orderBy: orderBy,
|
||||
|
|
@ -244,7 +264,12 @@ class CrdtDatabase implements Database {
|
|||
final table = serializationManager.getTableForType(T);
|
||||
final where = table?.id.equals(id);
|
||||
final result = await _delegate.findFirstRow<T>(
|
||||
where: _whereVisibleWithTombstone<T>(where, include, transaction),
|
||||
where: await _whereVisibleWithTombstone<T>(
|
||||
where,
|
||||
include,
|
||||
transaction,
|
||||
membershipWide: true,
|
||||
),
|
||||
transaction: transaction,
|
||||
include: include,
|
||||
lockMode: lockMode,
|
||||
|
|
@ -268,7 +293,12 @@ class CrdtDatabase implements Database {
|
|||
}) async {
|
||||
await _ensureInitialized();
|
||||
final result = await _delegate.findFirstRow<T>(
|
||||
where: _whereVisibleWithTombstone<T>(where, include, transaction),
|
||||
where: await _whereVisibleWithTombstone<T>(
|
||||
where,
|
||||
include,
|
||||
transaction,
|
||||
membershipWide: true,
|
||||
),
|
||||
offset: offset,
|
||||
orderBy: orderBy,
|
||||
orderByList: orderByList,
|
||||
|
|
@ -289,6 +319,7 @@ class CrdtDatabase implements Database {
|
|||
List<T> rows, {
|
||||
Transaction? transaction,
|
||||
bool ignoreConflicts = false,
|
||||
bool noReturn = false,
|
||||
}) async {
|
||||
if (rows.isEmpty) return [];
|
||||
await _ensureInitialized();
|
||||
|
|
@ -348,6 +379,7 @@ class CrdtDatabase implements Database {
|
|||
List<Column>? updateColumns,
|
||||
Expression? updateWhere,
|
||||
Transaction? transaction,
|
||||
bool noReturn = false,
|
||||
}) async {
|
||||
await _ensureInitialized();
|
||||
// TODO: Implement CRDT upsert.
|
||||
|
|
@ -381,6 +413,7 @@ class CrdtDatabase implements Database {
|
|||
List<T> rows, {
|
||||
List<Column>? columns,
|
||||
Transaction? transaction,
|
||||
bool noReturn = false,
|
||||
}) async {
|
||||
if (rows.isEmpty) return [];
|
||||
await _ensureInitialized();
|
||||
|
|
@ -444,7 +477,12 @@ class CrdtDatabase implements Database {
|
|||
final where = row.table.id.equals(row.id);
|
||||
final updatedRows = await _delegate.updateWhere<T>(
|
||||
columnValues: columnValues,
|
||||
where: _whereVisibleWithTombstone<T>(where, null, transaction)!,
|
||||
where: (await _whereVisibleWithTombstone<T>(
|
||||
where,
|
||||
null,
|
||||
transaction,
|
||||
membershipWide: false,
|
||||
))!,
|
||||
transaction: transaction,
|
||||
);
|
||||
|
||||
|
|
@ -489,6 +527,7 @@ class CrdtDatabase implements Database {
|
|||
List<Column>? orderByList,
|
||||
bool orderDescending = false,
|
||||
Transaction? transaction,
|
||||
bool noReturn = false,
|
||||
}) async {
|
||||
await _ensureInitialized();
|
||||
if (_recorder.isCrdtTracked<T>()) {
|
||||
|
|
@ -501,7 +540,12 @@ class CrdtDatabase implements Database {
|
|||
(tx) async {
|
||||
final result = await _delegate.updateWhere<T>(
|
||||
columnValues: columnValues,
|
||||
where: _whereVisibleWithTombstone<T>(where, null, tx)!,
|
||||
where: (await _whereVisibleWithTombstone<T>(
|
||||
where,
|
||||
null,
|
||||
tx,
|
||||
membershipWide: false,
|
||||
))!,
|
||||
limit: limit,
|
||||
offset: offset,
|
||||
orderBy: orderBy,
|
||||
|
|
@ -529,6 +573,7 @@ class CrdtDatabase implements Database {
|
|||
List<Column>? orderByList,
|
||||
bool orderDescending = false,
|
||||
Transaction? transaction,
|
||||
bool noReturn = false,
|
||||
}) async {
|
||||
if (rows.isEmpty) return [];
|
||||
await _ensureInitialized();
|
||||
|
|
@ -572,6 +617,7 @@ class CrdtDatabase implements Database {
|
|||
List<Column>? orderByList,
|
||||
bool orderDescending = false,
|
||||
Transaction? transaction,
|
||||
bool noReturn = false,
|
||||
}) async {
|
||||
await _ensureInitialized();
|
||||
if (!_recorder.isCrdtTracked<T>()) {
|
||||
|
|
@ -591,7 +637,12 @@ class CrdtDatabase implements Database {
|
|||
transaction,
|
||||
(tx) async {
|
||||
final rows = await _delegate.find<T>(
|
||||
where: _whereVisibleWithTombstone<T>(where, null, tx),
|
||||
where: await _whereVisibleWithTombstone<T>(
|
||||
where,
|
||||
null,
|
||||
tx,
|
||||
membershipWide: false,
|
||||
),
|
||||
orderBy: orderBy,
|
||||
orderByList: orderByList,
|
||||
// Remove this once the deprecated member is removed.
|
||||
|
|
@ -615,7 +666,12 @@ class CrdtDatabase implements Database {
|
|||
}) async {
|
||||
await _ensureInitialized();
|
||||
return _delegate.count<T>(
|
||||
where: _whereVisibleWithTombstone<T>(where, null, transaction),
|
||||
where: await _whereVisibleWithTombstone<T>(
|
||||
where,
|
||||
null,
|
||||
transaction,
|
||||
membershipWide: true,
|
||||
),
|
||||
limit: limit,
|
||||
useCache: useCache,
|
||||
transaction: transaction,
|
||||
|
|
@ -631,7 +687,12 @@ class CrdtDatabase implements Database {
|
|||
}) async {
|
||||
await _ensureInitialized();
|
||||
return _delegate.lockRows<T>(
|
||||
where: _whereVisibleWithTombstone<T>(where, null, transaction)!,
|
||||
where: (await _whereVisibleWithTombstone<T>(
|
||||
where,
|
||||
null,
|
||||
transaction,
|
||||
membershipWide: false,
|
||||
))!,
|
||||
lockMode: lockMode,
|
||||
transaction: transaction,
|
||||
lockBehavior: lockBehavior,
|
||||
|
|
@ -654,31 +715,100 @@ class CrdtDatabase implements Database {
|
|||
|
||||
/// Executes the [transactionFunction] in a transaction with the provided [userId].
|
||||
///
|
||||
/// The [userId] is used to record the CRDT operations in the database. This method
|
||||
/// must be used instead of [transaction] when performing operations on the server,
|
||||
/// where the database knows multiple users.
|
||||
/// Without [scopeId], writes act in the user's personal scope. With [scopeId],
|
||||
/// [userId] stays the authenticated identity and [scopeId] is the scope being
|
||||
/// acted in; the pair is checked before the transaction starts.
|
||||
Future<R> transactionForUser<R>(
|
||||
UuidValue userId,
|
||||
TransactionFunction<R> transactionFunction, {
|
||||
UuidValue? scopeId,
|
||||
TransactionSettings? settings,
|
||||
}) async {
|
||||
await _ensureInitialized();
|
||||
// Ensure that the user exists with a node before starting the transaction.
|
||||
final user = await _recorder.getOrCreateScope(userId);
|
||||
final effectiveScopeId = scopeId ?? userId;
|
||||
await _assertCanActInScope(userId, effectiveScopeId);
|
||||
|
||||
// Ensure that the scope exists with a node before starting the transaction.
|
||||
final scope = await _recorder.getOrCreateScope(effectiveScopeId);
|
||||
|
||||
return transaction<R>(
|
||||
(tx) async {
|
||||
try {
|
||||
scopeForTransaction[tx] = user;
|
||||
scopeForTransaction[tx] = scope;
|
||||
userForTransaction[tx] = userId;
|
||||
return await transactionFunction(tx);
|
||||
} finally {
|
||||
scopeForTransaction.remove(tx);
|
||||
userForTransaction.remove(tx);
|
||||
}
|
||||
},
|
||||
settings: settings,
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> _assertCanActInScope(UuidValue userId, UuidValue scopeId) async {
|
||||
if (userId == scopeId) return;
|
||||
|
||||
// Authoritative membership: the source of truth on the server, the
|
||||
// read-only cache on a follower (empty until populated). A null role means
|
||||
// no membership row at all; a non-writable role means membership without
|
||||
// write access.
|
||||
final role = await CrdtScopeMembership.roleOf(
|
||||
_delegate.session,
|
||||
userUuid: userId,
|
||||
scopeUuid: scopeId,
|
||||
);
|
||||
if (role == null) {
|
||||
throw CrdtScopeMembershipException(userId: userId, scopeId: scopeId);
|
||||
}
|
||||
if (!role.canWrite) {
|
||||
throw CrdtScopeRoleException(userId: userId, scopeId: scopeId, role: role);
|
||||
}
|
||||
}
|
||||
|
||||
Future<List<int>?> _scopeIdsForQueries(
|
||||
Transaction? transaction, {
|
||||
required bool membershipWide,
|
||||
}) async {
|
||||
if (!membershipWide) {
|
||||
return _actingScopeIdsForQueries(transaction);
|
||||
}
|
||||
|
||||
final userId = _userIdForQueries(transaction);
|
||||
if (userId == null) return null;
|
||||
|
||||
// On the server this is authoritative membership; on a persistent client it
|
||||
// is the server-projected membership cache.
|
||||
return _scopeIdsForUuids(
|
||||
await CrdtScopeMembership.memberScopes(_delegate.session, userId),
|
||||
);
|
||||
}
|
||||
|
||||
List<int>? _actingScopeIdsForQueries(Transaction? transaction) {
|
||||
final scopeId = _recorder.scopeForQueries(transaction)?.id;
|
||||
return scopeId == null ? null : [scopeId];
|
||||
}
|
||||
|
||||
UuidValue? _userIdForQueries(Transaction? transaction) {
|
||||
if (transaction != null) {
|
||||
final userId = userForTransaction[transaction];
|
||||
if (userId != null) return userId;
|
||||
}
|
||||
return _recorder.persistentUserId;
|
||||
}
|
||||
|
||||
Future<List<int>> _scopeIdsForUuids(List<UuidValue> scopeUuids) async {
|
||||
if (scopeUuids.isEmpty) return [];
|
||||
final scopes = await CrdtScope.db.find(
|
||||
_delegate.session,
|
||||
where: (t) => t.uuidScopeId.inSet(scopeUuids.toSet()),
|
||||
);
|
||||
return [
|
||||
for (final scope in scopes)
|
||||
if (scope.id != null) scope.id!,
|
||||
];
|
||||
}
|
||||
|
||||
Future<UuidValue> _requireUserId(UuidValue? userId) async {
|
||||
return userId ??
|
||||
_recorder.persistentUserId ??
|
||||
|
|
|
|||
|
|
@ -10,6 +10,10 @@ typedef _MergeContext = ({
|
|||
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 {
|
||||
|
|
@ -47,6 +51,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
{for (final change in operations) change.uuidNodeId},
|
||||
transaction,
|
||||
);
|
||||
final nodesByUuid = remoteNodes.nodesByUuid;
|
||||
final context = await _loadMergeContext(mergeSet, transaction);
|
||||
|
||||
for (final operation in operations) {
|
||||
|
|
@ -58,21 +63,21 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
case final CrdtMergeInsert insert:
|
||||
await _applyMergeInsert(
|
||||
insert,
|
||||
remoteNodes,
|
||||
nodesByUuid,
|
||||
context,
|
||||
transaction,
|
||||
);
|
||||
case final CrdtMergeUpdate update:
|
||||
await _applyMergeUpdate(
|
||||
update,
|
||||
remoteNodes,
|
||||
nodesByUuid,
|
||||
context,
|
||||
transaction,
|
||||
);
|
||||
case final CrdtMergeDelete delete:
|
||||
await _applyMergeDelete(
|
||||
delete,
|
||||
remoteNodes,
|
||||
nodesByUuid,
|
||||
context,
|
||||
transaction,
|
||||
);
|
||||
|
|
@ -91,14 +96,14 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
|
||||
Future<void> _updateHlcFromIncomingOperations(
|
||||
List<CrdtMergeChange> operations,
|
||||
Map<UuidValue, CrdtNode> remoteNodes,
|
||||
_MergeNodes remoteNodes,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final maxIncomingHlc = operations.fold<Hlc?>(
|
||||
null,
|
||||
(current, change) => change.hlc.maxBetween(current),
|
||||
);
|
||||
final nodesToUpdate = <CrdtNode>[];
|
||||
final scopeNodesToUpdate = <CrdtScopeNode>[];
|
||||
|
||||
if (maxIncomingHlc != null) {
|
||||
final hlcManager = _getHlcManager(transaction);
|
||||
|
|
@ -110,7 +115,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
hlcManager.merge(maxIncomingHlc);
|
||||
}
|
||||
|
||||
nodesToUpdate.add(hlcManager.getNode());
|
||||
await _persistCurrentNodeHlc(hlcManager, transaction);
|
||||
}
|
||||
|
||||
final maxIncomingHlcByNode = <UuidValue, Hlc>{};
|
||||
|
|
@ -122,66 +127,59 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
|
||||
for (final MapEntry(key: nodeId, value: incomingHlc)
|
||||
in maxIncomingHlcByNode.entries) {
|
||||
final remoteNode = remoteNodes[nodeId];
|
||||
final remoteNode = remoteNodes.scopeNodesByUuid[nodeId];
|
||||
if (remoteNode == null) continue;
|
||||
final updatedNode = remoteNode.copyWith(
|
||||
final updatedScopeNode = remoteNode.copyWith(
|
||||
lastReceivedHlc: incomingHlc.maxBetween(remoteNode.lastReceivedHlc),
|
||||
);
|
||||
if (updatedNode.lastReceivedHlc == remoteNode.lastReceivedHlc) continue;
|
||||
if (updatedScopeNode.lastReceivedHlc == remoteNode.lastReceivedHlc) {
|
||||
continue;
|
||||
}
|
||||
|
||||
nodesToUpdate.add(updatedNode);
|
||||
remoteNodes[nodeId] = updatedNode;
|
||||
scopeNodesToUpdate.add(updatedScopeNode);
|
||||
remoteNodes.scopeNodesByUuid[nodeId] = updatedScopeNode;
|
||||
}
|
||||
|
||||
if (nodesToUpdate.isNotEmpty) {
|
||||
await CrdtNode.db.update(
|
||||
if (scopeNodesToUpdate.isNotEmpty) {
|
||||
await CrdtScopeNode.db.update(
|
||||
_session,
|
||||
nodesToUpdate,
|
||||
scopeNodesToUpdate,
|
||||
columns: (t) => [t.lastReceivedHlc],
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Future<Map<UuidValue, CrdtNode>> _findOrCreateNodesForMerge(
|
||||
Future<_MergeNodes> _findOrCreateNodesForMerge(
|
||||
int userId,
|
||||
Set<UuidValue> nodeIds,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
if (nodeIds.isEmpty) return {};
|
||||
|
||||
var nodes = await CrdtNode.db.find(
|
||||
_session,
|
||||
where: (t) => t.scopeId.equals(userId) & t.uuidNodeId.inSet(nodeIds),
|
||||
transaction: transaction,
|
||||
);
|
||||
final existingNodeIds = nodes.map((node) => node.uuidNodeId).toSet();
|
||||
final missingNodeIds = nodeIds.difference(existingNodeIds);
|
||||
|
||||
if (missingNodeIds.isNotEmpty) {
|
||||
await CrdtNode.db.insert(
|
||||
_session,
|
||||
[
|
||||
for (final uuidNodeId in missingNodeIds)
|
||||
CrdtNode(
|
||||
scopeId: userId,
|
||||
uuidNodeId: uuidNodeId,
|
||||
),
|
||||
],
|
||||
transaction: transaction,
|
||||
ignoreConflicts: true,
|
||||
);
|
||||
|
||||
nodes = await CrdtNode.db.find(
|
||||
_session,
|
||||
where: (t) => t.scopeId.equals(userId) & t.uuidNodeId.inSet(nodeIds),
|
||||
transaction: transaction,
|
||||
if (nodeIds.isEmpty) {
|
||||
return (
|
||||
nodesByUuid: const <UuidValue, CrdtNode>{},
|
||||
scopeNodesByUuid: const <UuidValue, CrdtScopeNode>{},
|
||||
);
|
||||
}
|
||||
|
||||
return {
|
||||
for (final node in nodes) node.uuidNodeId: node,
|
||||
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(
|
||||
|
|
|
|||
|
|
@ -357,28 +357,23 @@ class CrdtMutationRecorder {
|
|||
UuidValue otherNodeId,
|
||||
Hlc syncedHlc,
|
||||
) async {
|
||||
final user = await _scopeManager.getOrCreate(userId);
|
||||
final scope = await _scopeManager.getOrCreate(userId);
|
||||
await _db.transaction((transaction) async {
|
||||
var node = await CrdtNode.db.findFirstRow(
|
||||
_session,
|
||||
where: (t) => t.scopeId.equals(user.id) & t.uuidNodeId.equals(otherNodeId),
|
||||
transaction: transaction,
|
||||
final node = await _findOrCreateNode(otherNodeId, transaction);
|
||||
final scopeNode = await _findOrCreateScopeNode(
|
||||
scope.id!,
|
||||
node.id!,
|
||||
transaction,
|
||||
);
|
||||
|
||||
node ??= await CrdtNode.db.insertRow(
|
||||
_session,
|
||||
CrdtNode(scopeId: user.id!, uuidNodeId: otherNodeId),
|
||||
transaction: transaction,
|
||||
);
|
||||
|
||||
final currentSyncHlc = node.lastReceivedHlc;
|
||||
final currentSyncHlc = scopeNode.lastReceivedHlc;
|
||||
if (currentSyncHlc != null && currentSyncHlc >= syncedHlc) {
|
||||
return;
|
||||
}
|
||||
|
||||
await CrdtNode.db.updateRow(
|
||||
await CrdtScopeNode.db.updateRow(
|
||||
_session,
|
||||
node.copyWith(lastReceivedHlc: syncedHlc),
|
||||
scopeNode.copyWith(lastReceivedHlc: syncedHlc),
|
||||
columns: (t) => [t.lastReceivedHlc],
|
||||
transaction: transaction,
|
||||
);
|
||||
|
|
@ -607,11 +602,13 @@ WHERE c."id" IN ($whereRowIds)
|
|||
if (rows.isEmpty) return;
|
||||
if (!isCrdtTracked<T>(rows.first.table)) return;
|
||||
|
||||
final hlcManager = _getHlcManager(transaction);
|
||||
await record(
|
||||
rows.first.table.tableName,
|
||||
rows.uuidRowIds,
|
||||
_getHlcManager(transaction),
|
||||
hlcManager,
|
||||
);
|
||||
await _persistCurrentNodeHlc(hlcManager, transaction);
|
||||
}
|
||||
|
||||
CrdtDataRow _newCrdtDataRow(
|
||||
|
|
@ -1331,6 +1328,72 @@ WHERE "id" IN (${_sqlLiteralList(rowIds)})
|
|||
);
|
||||
}
|
||||
|
||||
Future<void> _persistCurrentNodeHlc(
|
||||
HlcManager hlcManager,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
await CrdtNode.db.updateRow(
|
||||
_session,
|
||||
hlcManager.getNode(),
|
||||
columns: (t) => [t.lastHlc],
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
|
||||
Future<CrdtNode> _findOrCreateNode(
|
||||
UuidValue uuidNodeId,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
var node = await CrdtNode.db.findFirstRow(
|
||||
_session,
|
||||
where: (t) => t.uuidNodeId.equals(uuidNodeId),
|
||||
transaction: transaction,
|
||||
);
|
||||
if (node != null) return node;
|
||||
|
||||
await CrdtNode.db.insert(
|
||||
_session,
|
||||
[CrdtNode(uuidNodeId: uuidNodeId)],
|
||||
transaction: transaction,
|
||||
ignoreConflicts: true,
|
||||
);
|
||||
node = await CrdtNode.db.findFirstRow(
|
||||
_session,
|
||||
where: (t) => t.uuidNodeId.equals(uuidNodeId),
|
||||
transaction: transaction,
|
||||
);
|
||||
return node ?? (throw StateError('Could not create CRDT node "$uuidNodeId".'));
|
||||
}
|
||||
|
||||
Future<CrdtScopeNode> _findOrCreateScopeNode(
|
||||
int scopeId,
|
||||
int nodeId,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
var scopeNode = await CrdtScopeNode.db.findFirstRow(
|
||||
_session,
|
||||
where: (t) => t.scopeId.equals(scopeId) & t.nodeId.equals(nodeId),
|
||||
transaction: transaction,
|
||||
);
|
||||
if (scopeNode != null) return scopeNode;
|
||||
|
||||
await CrdtScopeNode.db.insert(
|
||||
_session,
|
||||
[CrdtScopeNode(scopeId: scopeId, nodeId: nodeId)],
|
||||
transaction: transaction,
|
||||
ignoreConflicts: true,
|
||||
);
|
||||
scopeNode = await CrdtScopeNode.db.findFirstRow(
|
||||
_session,
|
||||
where: (t) => t.scopeId.equals(scopeId) & t.nodeId.equals(nodeId),
|
||||
transaction: transaction,
|
||||
);
|
||||
return scopeNode ??
|
||||
(throw StateError(
|
||||
'Could not create CRDT scope-node row for scope $scopeId and node $nodeId.',
|
||||
));
|
||||
}
|
||||
|
||||
CrdtScope _getEffectiveScope(Transaction transaction) {
|
||||
return scopeForQueries(transaction) ??
|
||||
(throw StateError(
|
||||
|
|
|
|||
|
|
@ -170,3 +170,48 @@ extension _CrdtDatabaseScope on CrdtDatabase {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Extension on [Table] that exposes [scopeEquals] for use in `where`
|
||||
/// clauses to restrict membership-wide reads to a single scope.
|
||||
extension ScopeUuidFilter on Table {
|
||||
/// Expression that narrows a membership-wide read to [scopeUuid].
|
||||
///
|
||||
/// Reads are membership-wide and strip the database-local `scopeId`, so a
|
||||
/// caller holding several scopes cannot otherwise restrict a query to one scope
|
||||
/// or tell which scope a row came from. The placeholder is resolved to a
|
||||
/// filter on the reserved `scopeId` column before the query runs. When the
|
||||
/// scope UUID is not known locally the filter matches no rows.
|
||||
///
|
||||
/// Example:
|
||||
/// ```dart
|
||||
/// // Returns only persons in the shared scope.
|
||||
/// final sharedPeople = await Person.db.find(
|
||||
/// session,
|
||||
/// where: (t) => t.scopeEquals(sharedScopeUuid),
|
||||
/// );
|
||||
///
|
||||
/// // Combines with other filters.
|
||||
/// final aliceInShared = await Person.db.find(
|
||||
/// session,
|
||||
/// where: (t) => t.name.equals('Alice') & t.scopeEquals(sharedScopeUuid),
|
||||
/// );
|
||||
/// ```
|
||||
Expression scopeEquals(UuidValue scopeUuid) =>
|
||||
_crdtScope.uuidScopeId.equals(scopeUuid);
|
||||
|
||||
CrdtScopeTable get _crdtScope {
|
||||
return createRelationTable<CrdtScopeTable>(
|
||||
relationFieldName: '${tableName}_crdt_scope',
|
||||
field: _scopeIdColumn,
|
||||
foreignField: CrdtScope.t.id,
|
||||
tableRelation: tableRelation,
|
||||
createTable: (foreignTableRelation) =>
|
||||
CrdtScopeTable(tableRelation: foreignTableRelation),
|
||||
);
|
||||
}
|
||||
|
||||
Column get _scopeIdColumn => columns.singleWhere(
|
||||
(column) => column.columnName == 'scopeId' && column is ColumnInt,
|
||||
orElse: () => throw StateError('Table "$tableName" has no scopeId column.'),
|
||||
);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,11 +8,11 @@ import '../protocol/protocol.dart';
|
|||
/// null when the table is not registered for CRDT synchronization.
|
||||
typedef CrdtTableIdResolver = int? Function(String tableName);
|
||||
|
||||
/// Resolves the [CrdtScope] id scoping the current query. Returns null when
|
||||
/// no scope is available (e.g. server-side reads outside `transactionForUser`
|
||||
/// Resolves the [CrdtScope] ids scoping the current query. Returns null when no
|
||||
/// scope is available (e.g. server-side reads outside `transactionForUser`
|
||||
/// without a persistent user), in which case the query is an admin read: rows
|
||||
/// are not isolated and a row is hidden only when its owning scope hides it.
|
||||
typedef CrdtScopeIdResolver = int? Function();
|
||||
typedef CrdtScopeIdsResolver = List<int>? Function();
|
||||
|
||||
/// Private sentinel class used to signal that the CRDT visibility filter
|
||||
/// should be bypassed. Detected via `is` type check instead of string
|
||||
|
|
@ -27,29 +27,26 @@ final class _IncludeHiddenSentinel extends Expression<String> {
|
|||
|
||||
/// Extension on [Table] that exposes [includeHiddenRows] for use in `where`
|
||||
/// clauses to bypass the CRDT tombstone/visibility filter.
|
||||
///
|
||||
/// When included in a `where` clause, all rows are returned — including those
|
||||
/// that are tombstoned (soft-deleted) — instead of being hidden. The placeholder
|
||||
/// is stripped before the query reaches the database, so no invalid SQL is
|
||||
/// generated.
|
||||
///
|
||||
/// Example:
|
||||
/// ```dart
|
||||
/// // Returns all persons, including tombstoned ones.
|
||||
/// final allPersons = await Person.db.find(
|
||||
/// session,
|
||||
/// where: (t) => t.includeHiddenRows,
|
||||
/// );
|
||||
///
|
||||
/// // Returns all persons named 'Alice', including tombstoned ones.
|
||||
/// final alicePersons = await Person.db.find(
|
||||
/// session,
|
||||
/// where: (t) => t.name.equals('Alice') & t.includeHiddenRows,
|
||||
/// );
|
||||
/// ```
|
||||
extension IncludeTombstonedRows on Table {
|
||||
/// Placeholder expression that signals the CRDT visibility filter should be
|
||||
/// skipped, returning tombstoned rows alongside live rows.
|
||||
/// Expression that bypasses the CRDT tombstone/visibility filter.
|
||||
///
|
||||
/// When included in a `where` clause, all rows are returned — including those
|
||||
/// that are tombstoned (soft-deleted) — instead of being hidden.
|
||||
///
|
||||
/// Example:
|
||||
/// ```dart
|
||||
/// // Returns all persons, including tombstoned ones.
|
||||
/// final allPersons = await Person.db.find(
|
||||
/// session,
|
||||
/// where: (t) => t.includeHiddenRows,
|
||||
/// );
|
||||
///
|
||||
/// // Returns all persons named 'Alice', including tombstoned ones.
|
||||
/// final alicePersons = await Person.db.find(
|
||||
/// session,
|
||||
/// where: (t) => t.name.equals('Alice') & t.includeHiddenRows,
|
||||
/// );
|
||||
/// ```
|
||||
Expression get includeHiddenRows => const _IncludeHiddenSentinel();
|
||||
}
|
||||
|
||||
|
|
@ -60,8 +57,8 @@ extension IncludeTombstonedRows on Table {
|
|||
/// is null, only the root table and [where] are merged.
|
||||
///
|
||||
/// Each predicate only considers [CrdtDataRow]s recorded for the queried table
|
||||
/// ([tableIdForName]) and, when a scope is available ([scopeId]), isolates the
|
||||
/// query to that scope's rows. This prevents a tombstone recorded for another
|
||||
/// ([tableIdForName]) and, when scopes are available ([scopeIds]), isolates the
|
||||
/// query to those scopes' rows. This prevents a tombstone recorded for another
|
||||
/// table or scope from masking an unrelated row that reuses the same UUID.
|
||||
/// Without a scope (admin reads), rows are not isolated and a row stays
|
||||
/// visible as long as its owning scope still sees it.
|
||||
|
|
@ -85,13 +82,13 @@ Expression? mergeWhereWithTombstone<T extends TableRow>(
|
|||
Expression? where,
|
||||
Include? include, {
|
||||
required CrdtTableIdResolver tableIdForName,
|
||||
required CrdtScopeIdResolver scopeId,
|
||||
required CrdtScopeIdsResolver scopeIds,
|
||||
}) {
|
||||
final includeObjectPredicates = _walkIncludeGraphForTombstone(
|
||||
include,
|
||||
null,
|
||||
tableIdForName: tableIdForName,
|
||||
scopeId: scopeId,
|
||||
scopeIds: scopeIds,
|
||||
);
|
||||
|
||||
final rootTable = serializationManager.getTableForType(T);
|
||||
|
|
@ -99,7 +96,7 @@ Expression? mergeWhereWithTombstone<T extends TableRow>(
|
|||
? where
|
||||
: _mergeWhereOptional(
|
||||
where,
|
||||
rootTable?.whereVisibleOnCrdtRow(tableIdForName, scopeId),
|
||||
rootTable?.whereVisibleOnCrdtRow(tableIdForName, scopeIds),
|
||||
);
|
||||
merged = _mergeWhereOptional(merged, includeObjectPredicates);
|
||||
return merged;
|
||||
|
|
@ -111,7 +108,7 @@ Expression? _walkIncludeGraphForTombstone(
|
|||
Include? inc,
|
||||
Expression? includeObjectPredicates, {
|
||||
required CrdtTableIdResolver tableIdForName,
|
||||
required CrdtScopeIdResolver scopeId,
|
||||
required CrdtScopeIdsResolver scopeIds,
|
||||
}) {
|
||||
if (inc == null) return includeObjectPredicates;
|
||||
if (inc is IncludeList) {
|
||||
|
|
@ -119,14 +116,14 @@ Expression? _walkIncludeGraphForTombstone(
|
|||
if (!_includesHiddenSentinel(inc.where)) {
|
||||
inc.where = _mergeWhereOptional(
|
||||
inc.where,
|
||||
inc.table.whereVisibleOnCrdtRow(tableIdForName, scopeId),
|
||||
inc.table.whereVisibleOnCrdtRow(tableIdForName, scopeIds),
|
||||
);
|
||||
}
|
||||
return _walkIncludeGraphForTombstone(
|
||||
inc.include,
|
||||
includeObjectPredicates,
|
||||
tableIdForName: tableIdForName,
|
||||
scopeId: scopeId,
|
||||
scopeIds: scopeIds,
|
||||
);
|
||||
}
|
||||
var acc = includeObjectPredicates;
|
||||
|
|
@ -140,7 +137,7 @@ Expression? _walkIncludeGraphForTombstone(
|
|||
nested,
|
||||
acc,
|
||||
tableIdForName: tableIdForName,
|
||||
scopeId: scopeId,
|
||||
scopeIds: scopeIds,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
|
@ -150,14 +147,14 @@ Expression? _walkIncludeGraphForTombstone(
|
|||
if (childTable != null) {
|
||||
acc = _mergeWhereOptional(
|
||||
acc,
|
||||
childTable.whereVisibleOnCrdtRow(tableIdForName, scopeId),
|
||||
childTable.whereVisibleOnCrdtRow(tableIdForName, scopeIds),
|
||||
);
|
||||
}
|
||||
acc = _walkIncludeGraphForTombstone(
|
||||
nested,
|
||||
acc,
|
||||
tableIdForName: tableIdForName,
|
||||
scopeId: scopeId,
|
||||
scopeIds: scopeIds,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
@ -190,15 +187,15 @@ extension on Table {
|
|||
/// registered for CRDT synchronization. Keeps rows when the id is null
|
||||
/// (unmatched [IncludeObject] joins) or no hidden CRDT row matches.
|
||||
///
|
||||
/// With a scope, rows are isolated to it (`scopeId = <scope>`) and checked
|
||||
/// against the scope's tombstones with a correlated `NOT EXISTS` probe
|
||||
/// covered by the (scopeId, tblId, uuidRowId) unique index. Without a
|
||||
/// With scopes, rows are isolated to them (`scopeId IN (<scopes>)`) and
|
||||
/// checked against those scopes' tombstones with a correlated `NOT EXISTS`
|
||||
/// probe covered by the (scopeId, tblId, uuidRowId) unique index. Without a
|
||||
/// scope (admin reads), there is no isolation filter and the probe is
|
||||
/// keyed by the row's own `scopeId` column instead — covered by the same
|
||||
/// index — so a row stays visible while its owning scope sees it.
|
||||
Expression? whereVisibleOnCrdtRow(
|
||||
CrdtTableIdResolver tableIdForName,
|
||||
CrdtScopeIdResolver scopeId,
|
||||
CrdtScopeIdsResolver scopeIds,
|
||||
) {
|
||||
if (id is! ColumnUuid) return null;
|
||||
final tableId = tableIdForName(tableName);
|
||||
|
|
@ -214,19 +211,26 @@ extension on Table {
|
|||
));
|
||||
|
||||
final crdtRow = CrdtDataRow.t;
|
||||
final effectiveScopeId = scopeId();
|
||||
final effectiveScopeIds = scopeIds();
|
||||
if (effectiveScopeIds != null && effectiveScopeIds.isEmpty) {
|
||||
return id.equals(null) | const Expression('FALSE');
|
||||
}
|
||||
|
||||
final scopeFilter = effectiveScopeIds == null
|
||||
? '${crdtRow.scopeId} = $scopeColumn'
|
||||
: '${crdtRow.scopeId} IN (${_sqlIntList(effectiveScopeIds)})';
|
||||
final notExistsExpr = Expression(
|
||||
'NOT EXISTS '
|
||||
'(SELECT 1 FROM "${crdtRow.tableName}" '
|
||||
'WHERE ${crdtRow.scopeId} = ${effectiveScopeId ?? scopeColumn} '
|
||||
'WHERE $scopeFilter '
|
||||
'AND ${crdtRow.tblId} = $tableId '
|
||||
'AND ${crdtRow.uuidRowId} = $id '
|
||||
'AND ${crdtRow.visibility} > $crdtRowLastVisibleVisibilityIndex)',
|
||||
);
|
||||
|
||||
return id.equals(null) |
|
||||
((effectiveScopeId != null)
|
||||
? (scopeColumn.equals(effectiveScopeId) & notExistsExpr)
|
||||
((effectiveScopeIds != null)
|
||||
? (scopeColumn.inSet(effectiveScopeIds.toSet()) & notExistsExpr)
|
||||
: notExistsExpr);
|
||||
}
|
||||
|
||||
|
|
@ -239,3 +243,7 @@ extension on Table {
|
|||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
String _sqlIntList(Iterable<int> values) {
|
||||
return values.map(ValueEncoder.instance.convert).join(', ');
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ class HlcManager {
|
|||
scope.uuidScopeId,
|
||||
scope.id!,
|
||||
scope.currentNodeId!,
|
||||
scope.currentNode!.lastReceivedHlc ?? Hlc.zero(scope.currentNode!.uuidNodeId),
|
||||
scope.currentNode!.lastHlc ?? Hlc.zero(scope.currentNode!.uuidNodeId),
|
||||
);
|
||||
}
|
||||
|
||||
|
|
@ -52,8 +52,7 @@ class HlcManager {
|
|||
/// Converts this manager state to the persisted current-node model.
|
||||
CrdtNode getNode() => CrdtNode(
|
||||
id: normalizedNodeId,
|
||||
scopeId: normalizedScopeId,
|
||||
uuidNodeId: uuidNodeId,
|
||||
lastReceivedHlc: lastHlc,
|
||||
lastHlc: lastHlc,
|
||||
);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -12,6 +12,9 @@ class CrdtScopeManager {
|
|||
|
||||
final Map<UuidValue, CrdtScope> _instances = {};
|
||||
|
||||
/// Canonical [CrdtNode] shared across all scopes on this database.
|
||||
CrdtNode? _cachedCurrentNode;
|
||||
|
||||
/// Returns the cached [CrdtScope] for the given scope ID.
|
||||
CrdtScope getCached(UuidValue uuidScopeId) =>
|
||||
_instances[uuidScopeId] ??
|
||||
|
|
@ -23,6 +26,7 @@ class CrdtScopeManager {
|
|||
/// Clears the in-memory cache so state is reloaded from the store.
|
||||
void clearCache() {
|
||||
_instances.clear();
|
||||
_cachedCurrentNode = null;
|
||||
}
|
||||
|
||||
/// Returns the [CrdtScope] for the given scope ID.
|
||||
|
|
@ -51,26 +55,101 @@ class CrdtScopeManager {
|
|||
transaction: transaction,
|
||||
);
|
||||
|
||||
if (scope.currentNode != null) {
|
||||
return scope;
|
||||
var currentNode = await _getOrCreateCurrentNode(transaction);
|
||||
|
||||
currentNode = await _preserveLatestCurrentNodeHlc(
|
||||
currentNode,
|
||||
scope.currentNode,
|
||||
transaction,
|
||||
);
|
||||
if (scope.currentNodeId != currentNode.id) {
|
||||
await CrdtScope.db.attachRow.currentNode(
|
||||
_session,
|
||||
scope,
|
||||
currentNode,
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
|
||||
final currentNode = await CrdtNode.db.insertRow(
|
||||
_session,
|
||||
CrdtNode(scopeId: scope.id!),
|
||||
transaction: transaction,
|
||||
);
|
||||
await _ensureScopeNode(scope.id!, currentNode.id!, transaction);
|
||||
|
||||
await CrdtScope.db.attachRow.currentNode(
|
||||
_session,
|
||||
scope,
|
||||
currentNode,
|
||||
transaction: transaction,
|
||||
);
|
||||
_cachedCurrentNode = currentNode;
|
||||
|
||||
return scope.copyWith(
|
||||
currentNodeId: currentNode.id,
|
||||
currentNode: currentNode,
|
||||
);
|
||||
}
|
||||
|
||||
Future<CrdtNode> _getOrCreateCurrentNode(Transaction transaction) async {
|
||||
final cachedId = _cachedCurrentNode?.id;
|
||||
if (cachedId != null) {
|
||||
final node = await CrdtNode.db.findById(
|
||||
_session,
|
||||
cachedId,
|
||||
transaction: transaction,
|
||||
);
|
||||
if (node != null) return node;
|
||||
_cachedCurrentNode = null;
|
||||
}
|
||||
|
||||
final existingScope = await CrdtScope.db.findFirstRow(
|
||||
_session,
|
||||
where: (t) => t.currentNodeId.notEquals(null),
|
||||
orderBy: (t) => t.id,
|
||||
include: CrdtScope.include(currentNode: CrdtNode.include()),
|
||||
transaction: transaction,
|
||||
);
|
||||
|
||||
final existingNode = existingScope?.currentNode;
|
||||
if (existingNode != null) {
|
||||
return _cachedCurrentNode = existingNode;
|
||||
}
|
||||
|
||||
return _cachedCurrentNode = await CrdtNode.db.insertRow(
|
||||
_session,
|
||||
CrdtNode(),
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
|
||||
Future<CrdtNode> _preserveLatestCurrentNodeHlc(
|
||||
CrdtNode currentNode,
|
||||
CrdtNode? scopeCurrentNode,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final scopeLastHlc = scopeCurrentNode?.lastHlc;
|
||||
if (scopeLastHlc == null || scopeCurrentNode?.id == currentNode.id) {
|
||||
return currentNode;
|
||||
}
|
||||
|
||||
final currentLastHlc = currentNode.lastHlc;
|
||||
if (currentLastHlc != null && currentLastHlc >= scopeLastHlc) {
|
||||
return currentNode;
|
||||
}
|
||||
|
||||
final updatedNode = currentNode.copyWith(
|
||||
lastHlc: scopeLastHlc.copyWith(nodeId: currentNode.uuidNodeId),
|
||||
);
|
||||
await CrdtNode.db.updateRow(
|
||||
_session,
|
||||
updatedNode,
|
||||
columns: (t) => [t.lastHlc],
|
||||
transaction: transaction,
|
||||
);
|
||||
return updatedNode;
|
||||
}
|
||||
|
||||
Future<void> _ensureScopeNode(
|
||||
int scopeId,
|
||||
int nodeId,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
await CrdtScopeNode.db.insert(
|
||||
_session,
|
||||
[CrdtScopeNode(scopeId: scopeId, nodeId: nodeId)],
|
||||
transaction: transaction,
|
||||
ignoreConflicts: true,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ class CrdtSyncEndpoint extends Endpoint {
|
|||
userId: UuidValue.withValidation(session.authenticated!.userIdentifier),
|
||||
inbound: changes,
|
||||
once: once,
|
||||
mode: CrdtSyncPeerMode.authoritative,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,6 +3,9 @@ class: CrdtMergeChange
|
|||
sealed: true
|
||||
extends: BaseHlc
|
||||
fields:
|
||||
### The scope this change belongs to.
|
||||
uuidScopeId: UuidValue
|
||||
|
||||
### The table receiving the change.
|
||||
tableName: String
|
||||
|
||||
|
|
|
|||
|
|
@ -1,15 +1,12 @@
|
|||
### CRDT nodes table.
|
||||
###
|
||||
### This table stores all nodes that have been seen.
|
||||
### This table stores stable CRDT replica identities that have been seen.
|
||||
class: CrdtNode
|
||||
table: crdt_nodes
|
||||
database: all
|
||||
fields:
|
||||
### The scope that owns the node.
|
||||
scope: CrdtScope?, relation(name=scope_nodes, onDelete=Cascade)
|
||||
|
||||
### Unique identifier for the node.
|
||||
### Stable unique identifier for the replica/database install.
|
||||
uuidNodeId: UuidValue, default=random_v7, unique
|
||||
|
||||
### The last HLC timestamp when the node has synchronized.
|
||||
lastReceivedHlc: Hlc?
|
||||
### Last HLC generated by the local replica when this row is the current node.
|
||||
lastHlc: Hlc?
|
||||
|
|
|
|||
|
|
@ -12,5 +12,5 @@ fields:
|
|||
### The current node id of the scope.
|
||||
currentNode: CrdtNode?, relation(optional)
|
||||
|
||||
### The nodes associated with the scope.
|
||||
nodes: List<CrdtNode>?, relation(name=scope_nodes)
|
||||
### The nodes associated with the scope and their per-scope checkpoints.
|
||||
nodes: List<CrdtScopeNode>?, relation(name=scope_nodes)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,31 @@
|
|||
### Authorization membership for shared CRDT scopes.
|
||||
###
|
||||
### `database: all`, so the table exists on every node and membership resolves
|
||||
### with the same code on both ends. The server is the authoritative writer; a
|
||||
### client copy (once populated) is a read-only cache for UI and offline role
|
||||
### checks.
|
||||
###
|
||||
### Personal-scope membership is implicit: a user always belongs to the scope
|
||||
### whose UUID equals their auth user UUID. Rows in this table represent shared
|
||||
### scopes only.
|
||||
class: CrdtScopeMember
|
||||
table: crdt_scope_members
|
||||
database: all
|
||||
fields:
|
||||
### Shared scope this membership grants access to.
|
||||
scope: CrdtScope?, relation(onDelete=Cascade)
|
||||
|
||||
### Auth user UUID that may access the scope.
|
||||
userUuid: UuidValue
|
||||
|
||||
### CRDT access role for this shared-scope membership.
|
||||
role: CrdtScopeRole
|
||||
|
||||
indexes:
|
||||
### Leads with userUuid so the per-cycle membership resolution
|
||||
### (`WHERE userUuid = ?`) is index-covered, while still enforcing one row
|
||||
### per (scope, user). `isMember` (`WHERE userUuid AND scopeId`) is covered
|
||||
### too.
|
||||
crdt_scope_member_unique_idx:
|
||||
fields: userUuid, scopeId
|
||||
unique: true
|
||||
Some files were not shown because too many files have changed in this diff Show more
Loading…
Reference in a new issue