From 4b66703280cffb0e757e9b7aa87cc0ab77050a16 Mon Sep 17 00:00:00 2001 From: Krzysztof Rudnicki Date: Sun, 5 Jul 2026 20:48:07 +0200 Subject: [PATCH] Initial implementation: unified CRDT merge scheme + GitHub sync transport Pure-Dart mirror of the Python crdt-sync package, module-for-module and test-for-test, so the merge algorithm canonically agrees on both sides of a sync between a Dart client and a Python one. No Flutter SDK dependency (package:http only) per Workstream D's scoping decision (2026-07-04). 59 tests, clean dart analyze. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_017McuA1nAi6jBT1EaS5BrVf --- crdt_sync_dart/.gitignore | 6 + crdt_sync_dart/LICENSE | 21 +++ crdt_sync_dart/README.md | 75 ++++++++ crdt_sync_dart/analysis_options.yaml | 11 ++ crdt_sync_dart/lib/crdt_sync.dart | 8 + crdt_sync_dart/lib/src/github_client.dart | 171 +++++++++++++++++ crdt_sync_dart/lib/src/hlc.dart | 102 ++++++++++ crdt_sync_dart/lib/src/log.dart | 20 ++ crdt_sync_dart/lib/src/record.dart | 142 ++++++++++++++ crdt_sync_dart/lib/src/sync.dart | 48 +++++ crdt_sync_dart/pubspec.yaml | 16 ++ crdt_sync_dart/test/github_client_test.dart | 195 ++++++++++++++++++++ crdt_sync_dart/test/hlc_test.dart | 117 ++++++++++++ crdt_sync_dart/test/log_test.dart | 85 +++++++++ crdt_sync_dart/test/record_test.dart | 141 ++++++++++++++ crdt_sync_dart/test/sync_test.dart | 192 +++++++++++++++++++ 16 files changed, 1350 insertions(+) create mode 100644 crdt_sync_dart/.gitignore create mode 100644 crdt_sync_dart/LICENSE create mode 100644 crdt_sync_dart/README.md create mode 100644 crdt_sync_dart/analysis_options.yaml create mode 100644 crdt_sync_dart/lib/crdt_sync.dart create mode 100644 crdt_sync_dart/lib/src/github_client.dart create mode 100644 crdt_sync_dart/lib/src/hlc.dart create mode 100644 crdt_sync_dart/lib/src/log.dart create mode 100644 crdt_sync_dart/lib/src/record.dart create mode 100644 crdt_sync_dart/lib/src/sync.dart create mode 100644 crdt_sync_dart/pubspec.yaml create mode 100644 crdt_sync_dart/test/github_client_test.dart create mode 100644 crdt_sync_dart/test/hlc_test.dart create mode 100644 crdt_sync_dart/test/log_test.dart create mode 100644 crdt_sync_dart/test/record_test.dart create mode 100644 crdt_sync_dart/test/sync_test.dart diff --git a/crdt_sync_dart/.gitignore b/crdt_sync_dart/.gitignore new file mode 100644 index 0000000..08e2419 --- /dev/null +++ b/crdt_sync_dart/.gitignore @@ -0,0 +1,6 @@ +.dart_tool/ +.packages +build/ +pubspec.lock +coverage/ +*.iml diff --git a/crdt_sync_dart/LICENSE b/crdt_sync_dart/LICENSE new file mode 100644 index 0000000..d1aa637 --- /dev/null +++ b/crdt_sync_dart/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Krzysztof Rudnicki + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/crdt_sync_dart/README.md b/crdt_sync_dart/README.md new file mode 100644 index 0000000..a466940 --- /dev/null +++ b/crdt_sync_dart/README.md @@ -0,0 +1,75 @@ +# crdt_sync + +Shared CRDT merge scheme + GitHub-Contents-API sync transport, extracted for +reuse across four personal apps that each need cross-device sync: `todo`, +`diet_guard`'s Flutter app, `wake_alarm`'s `phone_app`, and screen-locker's +`workout_app`. Pure Dart -- no Flutter SDK dependency (`package:http` only) +-- so it works in both Flutter apps and plain Dart tooling. Mirrors the +[Python `crdt-sync`](https://github.com/kuhyx/crdt-sync) package +module-for-module and test-for-test, so the merge algorithm canonically +agrees on both sides of any sync between a Dart client and a Python one. + +## The merge scheme + +See the Python package's README for the full rationale; in short, a generic +**LWW-map-with-sticky-remove**: + +- `Hlc` -- a Hybrid Logical Clock `(wallTimeMs, counter, nodeId)`, totally + ordered, monotonic even across clock skew. +- `Record` -- an id, a per-field last-writer-wins map (`fieldName -> + (value, Hlc)`), and a sticky `deleted` flag (`deleted = a.deleted || + b.deleted`, never the other way -- a delete can never be silently undone). +- `Log` -- `Map`, merged by taking the union of ids and + merging shared ones. + +All three merges are commutative and idempotent -- see `test/log_test.dart` +for the concrete convergence-property tests. + +## Install + +Pin to a tag once this is pushed: + +```yaml +dependencies: + crdt_sync: + git: + url: https://github.com/kuhyx/crdt_sync_dart + ref: v0.1.0 +``` + +## Usage + +```dart +import 'package:crdt_sync/crdt_sync.dart'; + +final nodeId = 'phone'; +final clock = Hlc.newTick(nodeId); +final record = Record( + id: 'abc123', + fields: {'text': ('buy milk', clock)}, +); + +final client = GitHubClient(owner: 'kuhyx', repo: 'my-app-sync', token: token); +final merged = await syncLog( + client: client, + deviceId: nodeId, + pathPrefix: 'devices', + localLog: {record.id: record}, + encode: (log) => jsonEncode(log.map((k, r) => MapEntry(k, r.toJson()))), + decode: (text) => (jsonDecode(text) as Map).map( + (k, v) => MapEntry(k, Record.fromJson(v as Map)), + ), +); +``` + +`syncLog` pulls every other device's last-pushed log from +`//...`, merges each into the local log with +`mergeLogs`, then pushes this device's own merged result back up. + +## Development + +```bash +dart pub get +dart analyze +dart test +``` diff --git a/crdt_sync_dart/analysis_options.yaml b/crdt_sync_dart/analysis_options.yaml new file mode 100644 index 0000000..e9907b0 --- /dev/null +++ b/crdt_sync_dart/analysis_options.yaml @@ -0,0 +1,11 @@ +# This file configures the analyzer for this pure-Dart package (no Flutter +# dependency, so `package:lints` rather than `package:flutter_lints`). + +include: package:lints/recommended.yaml + +linter: + rules: + - prefer_single_quotes + - always_declare_return_types + - avoid_dynamic_calls + - unawaited_futures diff --git a/crdt_sync_dart/lib/crdt_sync.dart b/crdt_sync_dart/lib/crdt_sync.dart new file mode 100644 index 0000000..ed94c6d --- /dev/null +++ b/crdt_sync_dart/lib/crdt_sync.dart @@ -0,0 +1,8 @@ +/// Shared CRDT merge scheme + GitHub-Contents-API sync transport. +library; + +export 'src/github_client.dart'; +export 'src/hlc.dart'; +export 'src/log.dart'; +export 'src/record.dart'; +export 'src/sync.dart'; diff --git a/crdt_sync_dart/lib/src/github_client.dart b/crdt_sync_dart/lib/src/github_client.dart new file mode 100644 index 0000000..c30df5b --- /dev/null +++ b/crdt_sync_dart/lib/src/github_client.dart @@ -0,0 +1,171 @@ +import 'dart:convert'; + +import 'package:http/http.dart' as http; + +const _apiBase = 'https://api.github.com'; +const _httpNotFound = 404; + +/// Raised for a GitHub API failure the caller must not silently ignore. +class GitHubSyncError implements Exception { + GitHubSyncError(this.message); + + final String message; + + @override + String toString() => 'GitHubSyncError: $message'; +} + +/// Raised when the configured repo itself is unreachable. +/// +/// Distinguished from a path-404 (nothing pushed to that path yet, which is +/// benign -- it just means no other device has synced before) so the caller +/// can tell "the repo name is wrong or the token isn't scoped to it" apart +/// from "no other device has synced yet". +class RepoNotFoundError extends GitHubSyncError { + RepoNotFoundError(super.message); +} + +/// Minimal GitHub Contents API client used as dumb file storage. +/// +/// GitHub is used purely as file storage via the REST Contents API, not a +/// git clone -- there is no working tree and no git-level merge; the only +/// merge is the domain-level one in `mergeLogs`. Mirrors +/// `crdt_sync._github.GitHubSyncClient` on the Python side: unlike `todo`'s +/// original `GitHubClient`, [listDirectory] returns every entry name +/// regardless of type (not just files), because `crdt_sync`'s +/// `//` layout needs to discover device +/// *subdirectories*; and [putFileText] resolves its own current SHA rather +/// than requiring the caller to track it. +class GitHubClient { + GitHubClient({ + required this.owner, + required this.repo, + required String token, + http.Client? httpClient, + }) // Dart forbids private named params, so this can't be an initializing + // formal; assign it explicitly. + // ignore: prefer_initializing_formals + : _token = token, + _http = httpClient ?? http.Client(); + + final String owner; + final String repo; + final String _token; + final http.Client _http; + + Map get _headers => { + 'Authorization': 'Bearer $_token', + 'Accept': 'application/vnd.github+json', + }; + + Uri _contentsUri(String path) => + Uri.parse('$_apiBase/repos/$owner/$repo/contents/$path'); + + Future _get(String path) async { + try { + return await _http.get(_contentsUri(path), headers: _headers); + } on http.ClientException catch (exc) { + throw GitHubSyncError('network error reading $path: $exc'); + } + } + + Future _repoExists() async { + try { + final res = await _http.get( + Uri.parse('$_apiBase/repos/$owner/$repo'), + headers: _headers, + ); + return res.statusCode >= 200 && res.statusCode < 300; + } on http.ClientException { + return false; + } + } + + Future _raiseForMissingPath(String path) async { + if (!await _repoExists()) { + throw RepoNotFoundError( + '$owner/$repo not found, private without access, or the token ' + 'lacks contents permission (while reading $path)', + ); + } + } + + /// Returns the entry names directly under [path] (empty if unused). + /// + /// Includes both files and subdirectories -- see the class doc for why + /// this differs from a plain "list files" helper. + Future> listDirectory(String path) async { + final res = await _get(path); + if (res.statusCode == _httpNotFound) { + await _raiseForMissingPath(path); + return []; + } + if (res.statusCode < 200 || res.statusCode >= 300) { + throw GitHubSyncError('GET $path (list) failed: ${res.statusCode}'); + } + final decoded = jsonDecode(res.body); + if (decoded is! List) return []; + return decoded + .whereType>() + .map((e) => e['name']) + .whereType() + .toList(); + } + + /// Returns the decoded text content at [path], or null if unused. + Future getFileText(String path) async { + final res = await _get(path); + if (res.statusCode == _httpNotFound) { + await _raiseForMissingPath(path); + return null; + } + if (res.statusCode < 200 || res.statusCode >= 300) { + throw GitHubSyncError('GET $path failed: ${res.statusCode}'); + } + final decoded = jsonDecode(res.body) as Map; + final content = (decoded['content'] as String? ?? '').replaceAll('\n', ''); + return utf8.decode(base64.decode(content)); + } + + Future _existingSha(String path) async { + final res = await _get(path); + if (res.statusCode == _httpNotFound) { + await _raiseForMissingPath(path); + return null; + } + if (res.statusCode < 200 || res.statusCode >= 300) { + throw GitHubSyncError('GET $path (for sha) failed: ${res.statusCode}'); + } + final decoded = jsonDecode(res.body) as Map; + return decoded['sha'] as String?; + } + + /// Creates or updates the file at [path] with [text]. + Future putFileText( + String path, + String text, { + required String message, + }) async { + final sha = await _existingSha(path); + final body = { + 'message': message, + 'content': base64.encode(utf8.encode(text)), + 'sha': ?sha, + }; + late final http.Response res; + try { + res = await _http.put( + _contentsUri(path), + headers: _headers, + body: jsonEncode(body), + ); + } on http.ClientException catch (exc) { + throw GitHubSyncError('network error pushing $path: $exc'); + } + if (res.statusCode < 200 || res.statusCode >= 300) { + throw GitHubSyncError('PUT $path failed: ${res.statusCode}'); + } + } + + void close() => _http.close(); +} diff --git a/crdt_sync_dart/lib/src/hlc.dart b/crdt_sync_dart/lib/src/hlc.dart new file mode 100644 index 0000000..aef2fc4 --- /dev/null +++ b/crdt_sync_dart/lib/src/hlc.dart @@ -0,0 +1,102 @@ +/// A Hybrid Logical Clock: a totally-ordered, monotonic per-node timestamp. +/// +/// Combines wall-clock time with a logical counter so two ticks issued by +/// the same node in the same millisecond still get a strict order, and two +/// ticks from different nodes are always comparable (the node id breaks +/// ties), without requiring synchronized clocks across devices. Mirrors +/// `crdt_sync._hlc.Hlc` on the Python side field-for-field so the two +/// serialize to and parse the same wire format. +class Hlc implements Comparable { + const Hlc({ + required this.wallTimeMs, + required this.counter, + required this.nodeId, + }); + + final int wallTimeMs; + final int counter; + final String nodeId; + + static const _isoPrefixLen = 23; // "YYYY-MM-DDTHH:MM:SS.mmm" + + /// Returns the next clock value for [nodeId]. + /// + /// Passing [previous] (this node's last-issued clock) is what makes the + /// clock monotonic even when the wall clock hasn't advanced (or has gone + /// backwards). [wallTimeMsOverride] is exposed for deterministic tests. + static Hlc newTick(String nodeId, {Hlc? previous, int? wallTimeMsOverride}) { + final now = + wallTimeMsOverride ?? DateTime.now().toUtc().millisecondsSinceEpoch; + if (previous == null) { + return Hlc(wallTimeMs: now, counter: 0, nodeId: nodeId); + } + final newWall = now > previous.wallTimeMs ? now : previous.wallTimeMs; + final counter = newWall == previous.wallTimeMs ? previous.counter + 1 : 0; + return Hlc(wallTimeMs: newWall, counter: counter, nodeId: nodeId); + } + + /// Serializes to a lexicographically sortable string: + /// `Z--`. + String toStr() { + final dt = DateTime.fromMillisecondsSinceEpoch(wallTimeMs, isUtc: true); + String two(int n) => n.toString().padLeft(2, '0'); + final iso = + '${dt.year.toString().padLeft(4, '0')}-${two(dt.month)}-${two(dt.day)}' + 'T${two(dt.hour)}:${two(dt.minute)}:${two(dt.second)}' + '.${(wallTimeMs % 1000).toString().padLeft(3, '0')}'; + final counterHex = counter.toRadixString(16).padLeft(4, '0'); + return '${iso}Z-$counterHex-$nodeId'; + } + + /// Parses the format produced by [toStr]. Throws [FormatException]. + static Hlc fromStr(String text) { + final zIndex = text.indexOf('Z-'); + if (zIndex == -1 || zIndex != _isoPrefixLen) { + throw FormatException('not a valid Hlc string', text); + } + final isoPart = text.substring(0, zIndex); + final rest = text.substring(zIndex + 2); + final dashIndex = rest.indexOf('-'); + if (dashIndex == -1) { + throw FormatException('not a valid Hlc string', text); + } + final counterHex = rest.substring(0, dashIndex); + final nodeId = rest.substring(dashIndex + 1); + final dt = DateTime.parse('${isoPart}Z'); + return Hlc( + wallTimeMs: dt.millisecondsSinceEpoch, + counter: int.parse(counterHex, radix: 16), + nodeId: nodeId, + ); + } + + @override + int compareTo(Hlc other) { + if (wallTimeMs != other.wallTimeMs) { + return wallTimeMs.compareTo(other.wallTimeMs); + } + if (counter != other.counter) { + return counter.compareTo(other.counter); + } + return nodeId.compareTo(other.nodeId); + } + + bool operator >(Hlc other) => compareTo(other) > 0; + bool operator >=(Hlc other) => compareTo(other) >= 0; + bool operator <(Hlc other) => compareTo(other) < 0; + bool operator <=(Hlc other) => compareTo(other) <= 0; + + @override + bool operator ==(Object other) => + other is Hlc && + wallTimeMs == other.wallTimeMs && + counter == other.counter && + nodeId == other.nodeId; + + @override + int get hashCode => Object.hash(wallTimeMs, counter, nodeId); + + @override + String toString() => + 'Hlc(wallTimeMs: $wallTimeMs, counter: $counter, nodeId: $nodeId)'; +} diff --git a/crdt_sync_dart/lib/src/log.dart b/crdt_sync_dart/lib/src/log.dart new file mode 100644 index 0000000..993e416 --- /dev/null +++ b/crdt_sync_dart/lib/src/log.dart @@ -0,0 +1,20 @@ +import 'record.dart'; + +/// A collection of [Record]s keyed by id. +typedef Log = Map; + +/// Returns the union of [local] and [remote], merging shared ids. +/// +/// Commutative and idempotent: `mergeLogs(a, b) == mergeLogs(b, a)` and +/// `mergeLogs(x, x) == x`, so pull order between devices never matters and a +/// repeated sync tick is a no-op. +Log mergeLogs(Log local, Log remote) { + final merged = Map.from(local); + for (final entry in remote.entries) { + final existing = merged[entry.key]; + merged[entry.key] = existing == null + ? entry.value + : mergeRecord(existing, entry.value); + } + return merged; +} diff --git a/crdt_sync_dart/lib/src/record.dart b/crdt_sync_dart/lib/src/record.dart new file mode 100644 index 0000000..5b31756 --- /dev/null +++ b/crdt_sync_dart/lib/src/record.dart @@ -0,0 +1,142 @@ +import 'hlc.dart'; + +/// A field value paired with the clock it was written at. +typedef Field = (Object?, Hlc); + +/// One CRDT record: an id, per-field LWW values, and a sticky tombstone. +/// +/// Mirrors `crdt_sync.Record` on the Python side. Deliberately +/// domain-agnostic -- translating an app's own model (a note, a food-log +/// entry, a workout entry, a wake-time value) to and from [Record] is each +/// app's job, not this library's. +class Record { + const Record({ + required this.id, + required this.fields, + this.deleted = false, + this.deletedHlc, + }); + + /// Stable identifier, unique within one `Log`. + final String id; + + /// Field name -> `(value, Hlc)`. Concurrent edits to the same field + /// converge to whichever side has the greater [Hlc]. + final Map fields; + + /// Whether this record is tombstoned. Deletion is monotonic: once true on + /// either side of a merge, it stays true. + final bool deleted; + + /// The clock value of the delete, if any -- informational/ordering only; + /// it does not participate in the merge decision (see [mergeRecord]). + final Hlc? deletedHlc; + + Map toJson() => { + 'id': id, + 'fields': fields.map( + (name, field) => MapEntry(name, [field.$1, field.$2.toStr()]), + ), + 'deleted': deleted, + 'deletedHlc': deletedHlc?.toStr(), + }; + + factory Record.fromJson(Map json) { + final rawFields = json['fields'] as Map; + final fields = rawFields.map((name, value) { + final pair = value as List; + return MapEntry(name, (pair[0], Hlc.fromStr(pair[1] as String))); + }); + final deletedHlcStr = json['deletedHlc'] as String?; + return Record( + id: json['id'] as String, + fields: fields, + deleted: json['deleted'] as bool, + deletedHlc: deletedHlcStr == null ? null : Hlc.fromStr(deletedHlcStr), + ); + } + + @override + bool operator ==(Object other) { + if (other is! Record) return false; + return id == other.id && + deleted == other.deleted && + deletedHlc == other.deletedHlc && + _fieldsEqual(fields, other.fields); + } + + static bool _fieldsEqual(Map a, Map b) { + if (a.length != b.length) return false; + for (final entry in a.entries) { + if (b[entry.key] != entry.value) return false; + } + return true; + } + + @override + int get hashCode => Object.hash( + id, + deleted, + deletedHlc, + Object.hashAllUnordered( + fields.entries.map((e) => Object.hash(e.key, e.value)), + ), + ); + + @override + String toString() => + 'Record(id: $id, fields: $fields, deleted: $deleted, ' + 'deletedHlc: $deletedHlc)'; +} + +/// Returns whichever of [a], [b] has the greater [Hlc]. +/// +/// Ties (equal [Hlc]) keep [a]; two ticks never compare equal across +/// different nodes, so a tie only happens when both sides are literally the +/// same tick re-merged, in which case the value is expected to be identical +/// anyway. +Field mergeField(Field a, Field b) => a.$2 >= b.$2 ? a : b; + +Hlc? _mergeDeletedHlc(Record a, Record b) { + if (a.deleted && b.deleted) { + if (a.deletedHlc == null) return b.deletedHlc; + if (b.deletedHlc == null) return a.deletedHlc; + return a.deletedHlc! >= b.deletedHlc! ? a.deletedHlc : b.deletedHlc; + } + if (a.deleted) return a.deletedHlc; + if (b.deleted) return b.deletedHlc; + return null; +} + +/// Merges two versions of the same record. +/// +/// Per-field last-writer-wins over the union of field names, plus a sticky +/// delete: `deleted = a.deleted || b.deleted`, never the other way around. A +/// tombstone can never be resurrected by merging in an older, non-deleted +/// copy pulled from a device that hasn't seen the delete yet. +/// +/// Commutative and idempotent: `mergeRecord(a, b) == mergeRecord(b, a)` and +/// `mergeRecord(a, a) == a`. +/// +/// Throws [ArgumentError] if `a.id != b.id` -- merging two different +/// records is a caller bug, not a case to silently paper over. +Record mergeRecord(Record a, Record b) { + if (a.id != b.id) { + throw ArgumentError( + 'cannot merge records with different ids: ${a.id} != ${b.id}', + ); + } + final mergedFields = Map.from(a.fields); + for (final entry in b.fields.entries) { + final existing = mergedFields[entry.key]; + mergedFields[entry.key] = existing == null + ? entry.value + : mergeField(existing, entry.value); + } + return Record( + id: a.id, + fields: mergedFields, + deleted: a.deleted || b.deleted, + deletedHlc: _mergeDeletedHlc(a, b), + ); +} diff --git a/crdt_sync_dart/lib/src/sync.dart b/crdt_sync_dart/lib/src/sync.dart new file mode 100644 index 0000000..946729b --- /dev/null +++ b/crdt_sync_dart/lib/src/sync.dart @@ -0,0 +1,48 @@ +import 'github_client.dart'; +import 'log.dart'; +import 'record.dart'; + +const _defaultFilename = 'log.json'; + +/// Runs one full sync tick: pull every other device's log, merge, push. +/// +/// Pulls from `//` for every device +/// directory GitHub reports under [pathPrefix], merges each into +/// [localLog] with [mergeLogs], then pushes this device's own merged result +/// to `//`. +/// +/// [encode] serializes a merged log for pushing. [decode] parses a remote +/// device's pushed text back into a log; throwing [FormatException] is +/// treated as a corrupt/unparsable push, and that device is skipped for +/// this tick rather than aborting the whole sync. +Future syncLog({ + required GitHubClient client, + required String deviceId, + required String pathPrefix, + required Log localLog, + required String Function(Log log) encode, + required Log Function(String text) decode, + String filename = _defaultFilename, + String commitMessage = 'crdt_sync: update log', +}) async { + var mergedLog = Map.from(localLog); + for (final otherDeviceId in await client.listDirectory(pathPrefix)) { + if (otherDeviceId == deviceId) continue; + final text = await client.getFileText( + '$pathPrefix/$otherDeviceId/$filename', + ); + if (text == null) continue; + try { + mergedLog = mergeLogs(mergedLog, decode(text)); + } on FormatException { + continue; + } + } + + await client.putFileText( + '$pathPrefix/$deviceId/$filename', + encode(mergedLog), + message: commitMessage, + ); + return mergedLog; +} diff --git a/crdt_sync_dart/pubspec.yaml b/crdt_sync_dart/pubspec.yaml new file mode 100644 index 0000000..f1d6505 --- /dev/null +++ b/crdt_sync_dart/pubspec.yaml @@ -0,0 +1,16 @@ +name: crdt_sync +description: >- + Shared CRDT merge scheme + GitHub-Contents-API sync transport for + cross-device apps. Pure Dart, no Flutter dependency. +version: 0.1.0 +publish_to: 'none' + +environment: + sdk: ^3.12.2 + +dependencies: + http: ^1.6.0 + +dev_dependencies: + lints: ^5.0.0 + test: ^1.25.0 diff --git a/crdt_sync_dart/test/github_client_test.dart b/crdt_sync_dart/test/github_client_test.dart new file mode 100644 index 0000000..6f70145 --- /dev/null +++ b/crdt_sync_dart/test/github_client_test.dart @@ -0,0 +1,195 @@ +import 'dart:convert'; + +import 'package:crdt_sync/crdt_sync.dart'; +import 'package:http/http.dart' as http; +import 'package:http/testing.dart' as http_testing; +import 'package:test/test.dart'; + +http.Response _response(int statusCode, [Object? jsonBody]) => + http.Response(jsonEncode(jsonBody ?? {}), statusCode); + +/// Sentinel meaning "this GET throws a network exception", since a real +/// [http.Response] can't carry an invalid status code. +final _networkError = Object(); + +/// Builds a client whose GETs return [getResponses] in call order (each +/// either an [http.Response] or [_networkError]), and whose PUT returns +/// [putResponse] (or throws [putError] if network fails). +GitHubClient _client( + List getResponses, { + http.Response? putResponse, + Object? putError, +}) { + var getIndex = 0; + final mock = http_testing.MockClient((request) async { + if (request.method == 'GET') { + final response = getResponses[getIndex]; + getIndex++; + if (identical(response, _networkError)) { + throw http.ClientException('offline'); + } + return response as http.Response; + } + if (request.method == 'PUT') { + if (putError != null) throw putError; + return putResponse!; + } + throw UnsupportedError('unexpected method ${request.method}'); + }); + return GitHubClient( + owner: 'kuhyx', + repo: 'crdt-sync-demo', + token: 'fake-token', + httpClient: mock, + ); +} + +void main() { + group('getFileText', () { + test('returns decoded content on success', () async { + final encoded = base64.encode(utf8.encode('hello world')); + final client = _client([ + _response(200, {'content': encoded}), + ]); + expect(await client.getFileText('devices/pc/log.json'), 'hello world'); + }); + + test('returns null for an unused path on a real repo', () async { + final client = _client([_response(404), _response(200)]); + expect(await client.getFileText('devices/phone/log.json'), isNull); + }); + + test('throws RepoNotFoundError when the repo itself is missing', () async { + final client = _client([_response(404), _response(404)]); + expect( + () => client.getFileText('devices/pc/log.json'), + throwsA(isA()), + ); + }); + + test('throws GitHubSyncError on a non-2xx non-404', () async { + final client = _client([_response(500)]); + expect( + () => client.getFileText('devices/pc/log.json'), + throwsA(isA()), + ); + }); + + test('throws GitHubSyncError on a network exception', () async { + final client = _client([_networkError]); + expect( + () => client.getFileText('devices/pc/log.json'), + throwsA(isA()), + ); + }); + + test( + 'treats a network error during the repo check as repo missing', + () async { + final client = _client([_response(404), _networkError]); + expect( + () => client.getFileText('devices/pc/log.json'), + throwsA(isA()), + ); + }, + ); + }); + + group('listDirectory', () { + test('returns entry names regardless of type', () async { + final payload = [ + {'name': 'pc', 'type': 'dir'}, + {'name': 'phone', 'type': 'dir'}, + {'not_a_name': 'x'}, + ]; + final client = _client([_response(200, payload)]); + expect(await client.listDirectory('devices'), ['pc', 'phone']); + }); + + test('returns an empty list when the response is not a list', () async { + final client = _client([ + _response(200, {'unexpected': 'shape'}), + ]); + expect(await client.listDirectory('devices'), isEmpty); + }); + + test('returns an empty list for an unused path on a real repo', () async { + final client = _client([_response(404), _response(200)]); + expect(await client.listDirectory('devices'), isEmpty); + }); + + test('throws RepoNotFoundError when the repo itself is missing', () async { + final client = _client([_response(404), _response(404)]); + expect( + () => client.listDirectory('devices'), + throwsA(isA()), + ); + }); + + test('throws GitHubSyncError on a non-2xx non-404', () async { + final client = _client([_response(500)]); + expect( + () => client.listDirectory('devices'), + throwsA(isA()), + ); + }); + }); + + group('putFileText', () { + test('creates a new file with no sha when none existed', () async { + final client = _client([ + _response(404), + _response(200), + ], putResponse: _response(201)); + await client.putFileText('devices/pc/log.json', '{}', message: 'm'); + }); + + test('updates an existing file by including its sha', () async { + final client = _client([ + _response(200, {'sha': 'abc123'}), + ], putResponse: _response(200)); + await client.putFileText('devices/pc/log.json', '{}', message: 'm'); + }); + + test( + 'throws RepoNotFoundError when checking sha on a missing repo', + () async { + final client = _client([_response(404), _response(404)]); + expect( + () => client.putFileText('devices/pc/log.json', '{}', message: 'm'), + throwsA(isA()), + ); + }, + ); + + test('throws GitHubSyncError when the sha check itself fails', () async { + final client = _client([_response(500)]); + expect( + () => client.putFileText('devices/pc/log.json', '{}', message: 'm'), + throwsA(isA()), + ); + }); + + test('throws GitHubSyncError on a put network exception', () async { + final client = _client([ + _response(404), + _response(200), + ], putError: http.ClientException('offline')); + expect( + () => client.putFileText('devices/pc/log.json', '{}', message: 'm'), + throwsA(isA()), + ); + }); + + test('throws GitHubSyncError on a put non-2xx response', () async { + final client = _client([ + _response(404), + _response(200), + ], putResponse: _response(422)); + expect( + () => client.putFileText('devices/pc/log.json', '{}', message: 'm'), + throwsA(isA()), + ); + }); + }); +} diff --git a/crdt_sync_dart/test/hlc_test.dart b/crdt_sync_dart/test/hlc_test.dart new file mode 100644 index 0000000..234e86c --- /dev/null +++ b/crdt_sync_dart/test/hlc_test.dart @@ -0,0 +1,117 @@ +import 'package:crdt_sync/crdt_sync.dart'; +import 'package:test/test.dart'; + +Hlc _make(int wallTimeMs, {int counter = 0, String nodeId = 'node-a'}) => + Hlc(wallTimeMs: wallTimeMs, counter: counter, nodeId: nodeId); + +void main() { + group('ordering', () { + test('greater wall time wins', () { + expect(_make(200) > _make(100), isTrue); + }); + + test('equal wall time, greater counter wins', () { + expect(_make(100, counter: 2) > _make(100, counter: 1), isTrue); + }); + + test('equal wall time and counter breaks tie on nodeId', () { + expect( + _make(100, counter: 1, nodeId: 'b') > + _make(100, counter: 1, nodeId: 'a'), + isTrue, + ); + }); + + test('identical clocks compare equal', () { + expect( + _make(100, counter: 1, nodeId: 'a'), + equals(_make(100, counter: 1, nodeId: 'a')), + ); + }); + }); + + group('newTick', () { + test('first tick has counter zero', () { + final tick = Hlc.newTick('node-a', wallTimeMsOverride: 1000); + expect(tick, equals(_make(1000))); + }); + + test('advancing wall clock resets counter', () { + final previous = _make(1000, counter: 5); + final tick = Hlc.newTick( + 'node-a', + previous: previous, + wallTimeMsOverride: 2000, + ); + expect(tick, equals(_make(2000))); + }); + + test('stalled wall clock increments counter', () { + final previous = _make(1000, counter: 5); + final tick = Hlc.newTick( + 'node-a', + previous: previous, + wallTimeMsOverride: 1000, + ); + expect(tick, equals(_make(1000, counter: 6))); + }); + + test('regressed wall clock still advances monotonically', () { + final previous = _make(1000, counter: 5); + final tick = Hlc.newTick( + 'node-a', + previous: previous, + wallTimeMsOverride: 500, + ); + expect(tick, equals(_make(1000, counter: 6))); + }); + + test('defaults to the real clock when unset', () { + final tick = Hlc.newTick('node-a'); + expect(tick.nodeId, 'node-a'); + expect(tick.wallTimeMs, greaterThan(0)); + }); + }); + + group('string round trip', () { + test('round trips through toStr and fromStr', () { + final original = _make(1751000123456, counter: 7, nodeId: 'phone'); + expect(Hlc.fromStr(original.toStr()), equals(original)); + }); + + test('toStr is lexicographically sortable by wall time', () { + expect(_make(1000).toStr().compareTo(_make(2000).toStr()), lessThan(0)); + }); + + test('toStr is lexicographically sortable by counter', () { + expect( + _make( + 1000, + counter: 1, + ).toStr().compareTo(_make(1000, counter: 200).toStr()), + lessThan(0), + ); + }); + + test('fromStr rejects a missing Z separator', () { + expect( + () => Hlc.fromStr('not-a-valid-clock-string-at-all'), + throwsFormatException, + ); + }); + + test('fromStr rejects a wrong-length iso prefix', () { + expect( + () => Hlc.fromStr('2026-07-05T12:00:00Z-0000-node-a'), + throwsFormatException, + ); + }); + + test('fromStr rejects a missing node id separator', () { + expect( + () => Hlc.fromStr('2026-07-05T12:00:00.000Z-0000'), + throwsFormatException, + ); + }); + }); +} diff --git a/crdt_sync_dart/test/log_test.dart b/crdt_sync_dart/test/log_test.dart new file mode 100644 index 0000000..7be4532 --- /dev/null +++ b/crdt_sync_dart/test/log_test.dart @@ -0,0 +1,85 @@ +import 'package:crdt_sync/crdt_sync.dart'; +import 'package:test/test.dart'; + +Hlc _make(int wallTimeMs, {int counter = 0, String nodeId = 'node-a'}) => + Hlc(wallTimeMs: wallTimeMs, counter: counter, nodeId: nodeId); + +(Log, Log) _sampleLogs() { + final a = { + 'shared': Record(id: 'shared', fields: {'text': ('a-version', _make(100))}), + 'only-in-a': Record(id: 'only-in-a', fields: {'text': ('a', _make(50))}), + }; + final b = { + 'shared': Record(id: 'shared', fields: {'text': ('b-version', _make(200))}), + 'only-in-b': Record(id: 'only-in-b', fields: {'text': ('b', _make(50))}), + }; + return (a, b); +} + +void main() { + group('mergeLogs', () { + test('returns the union of disjoint ids', () { + final local = { + 'a': Record(id: 'a', fields: {'text': ('a', _make(100))}), + }; + final remote = { + 'b': Record(id: 'b', fields: {'text': ('b', _make(100))}), + }; + expect(mergeLogs(local, remote), {...local, ...remote}); + }); + + test('merges a shared id field by field', () { + final local = { + 'a': Record(id: 'a', fields: {'text': ('old', _make(100))}), + }; + final remote = { + 'a': Record(id: 'a', fields: {'text': ('new', _make(200))}), + }; + final merged = mergeLogs(local, remote); + expect(merged['a']!.fields['text'], equals(('new', _make(200)))); + }); + + test('a delete on one side survives the merge', () { + final local = { + 'a': Record(id: 'a', fields: {}, deleted: true, deletedHlc: _make(200)), + }; + final remote = { + 'a': Record(id: 'a', fields: {'text': ('still here', _make(100))}), + }; + expect(mergeLogs(local, remote)['a']!.deleted, isTrue); + }); + + test('does not mutate either input', () { + final local = { + 'a': Record(id: 'a', fields: {'text': ('a', _make(100))}), + }; + final remote = { + 'b': Record(id: 'b', fields: {'text': ('b', _make(100))}), + }; + final localBefore = Map.from(local); + final remoteBefore = Map.from(remote); + mergeLogs(local, remote); + expect(local, equals(localBefore)); + expect(remote, equals(remoteBefore)); + }); + }); + + group('convergence properties', () { + test('is commutative', () { + final (a, b) = _sampleLogs(); + expect(mergeLogs(a, b), equals(mergeLogs(b, a))); + }); + + test('is idempotent', () { + final (a, _) = _sampleLogs(); + expect(mergeLogs(a, a), equals(a)); + }); + + test('repeated merge of the same remote is a no-op', () { + final (a, b) = _sampleLogs(); + final once = mergeLogs(a, b); + final twice = mergeLogs(once, b); + expect(once, equals(twice)); + }); + }); +} diff --git a/crdt_sync_dart/test/record_test.dart b/crdt_sync_dart/test/record_test.dart new file mode 100644 index 0000000..24bebaa --- /dev/null +++ b/crdt_sync_dart/test/record_test.dart @@ -0,0 +1,141 @@ +import 'package:crdt_sync/crdt_sync.dart'; +import 'package:test/test.dart'; + +Hlc _make(int wallTimeMs, {int counter = 0, String nodeId = 'node-a'}) => + Hlc(wallTimeMs: wallTimeMs, counter: counter, nodeId: nodeId); + +void main() { + group('mergeField', () { + test('greater Hlc wins', () { + final older = ('a', _make(100)); + final newer = ('b', _make(200)); + expect(mergeField(older, newer), equals(newer)); + expect(mergeField(newer, older), equals(newer)); + }); + + test('equal Hlc keeps the first argument', () { + final clock = _make(100); + final a = ('a', clock); + final b = ('a', clock); + expect(mergeField(a, b), equals(a)); + }); + }); + + group('mergeRecord', () { + test('throws when ids differ', () { + final a = Record(id: 'a', fields: {}); + final b = Record(id: 'b', fields: {}); + expect(() => mergeRecord(a, b), throwsArgumentError); + }); + + test('merges disjoint fields', () { + final a = Record(id: 'x', fields: {'text': ('hello', _make(100))}); + final b = Record(id: 'x', fields: {'priority': ('high', _make(50))}); + final merged = mergeRecord(a, b); + expect(merged.fields, { + 'text': ('hello', _make(100)), + 'priority': ('high', _make(50)), + }); + }); + + test('shared field keeps the later write', () { + final a = Record(id: 'x', fields: {'text': ('old', _make(100))}); + final b = Record(id: 'x', fields: {'text': ('new', _make(200))}); + final merged = mergeRecord(a, b); + expect(merged.fields['text'], equals(('new', _make(200)))); + }); + + test('delete is sticky against an older non-deleted copy', () { + final deleted = Record( + id: 'x', + fields: {}, + deleted: true, + deletedHlc: _make(200), + ); + final stale = Record( + id: 'x', + fields: {'text': ('resurrected', _make(100))}, + ); + expect(mergeRecord(deleted, stale).deleted, isTrue); + expect(mergeRecord(stale, deleted).deleted, isTrue); + }); + + test('neither side deleted stays not deleted', () { + final a = Record(id: 'x', fields: {}); + final b = Record(id: 'x', fields: {}); + final merged = mergeRecord(a, b); + expect(merged.deleted, isFalse); + expect(merged.deletedHlc, isNull); + }); + + test('both sides deleted keeps the later delete clock', () { + final a = Record( + id: 'x', + fields: {}, + deleted: true, + deletedHlc: _make(100), + ); + final b = Record( + id: 'x', + fields: {}, + deleted: true, + deletedHlc: _make(200), + ); + expect(mergeRecord(a, b).deletedHlc, equals(_make(200))); + }); + + test('both deleted but one side missing a clock keeps the other', () { + final a = Record(id: 'x', fields: {}, deleted: true); + final b = Record( + id: 'x', + fields: {}, + deleted: true, + deletedHlc: _make(200), + ); + expect(mergeRecord(a, b).deletedHlc, equals(_make(200))); + expect(mergeRecord(b, a).deletedHlc, equals(_make(200))); + }); + + test('is commutative', () { + final recordA = Record(id: 'x', fields: {'text': ('a', _make(100))}); + final recordB = Record( + id: 'x', + fields: {'text': ('b', _make(200)), 'extra': ('e', _make(1))}, + ); + expect( + mergeRecord(recordA, recordB), + equals(mergeRecord(recordB, recordA)), + ); + }); + + test('is idempotent', () { + final record = Record(id: 'x', fields: {'text': ('a', _make(100))}); + expect(mergeRecord(record, record), equals(record)); + }); + }); + + group('Record JSON round trip', () { + test('round trips a record with fields', () { + final record = Record( + id: 'x', + fields: {'text': ('hello', _make(100, nodeId: 'pc'))}, + ); + expect(Record.fromJson(record.toJson()), equals(record)); + }); + + test('round trips a deleted record', () { + final record = Record( + id: 'x', + fields: {}, + deleted: true, + deletedHlc: _make(200), + ); + expect(Record.fromJson(record.toJson()), equals(record)); + }); + + test('round trips a record with no delete clock', () { + final record = Record(id: 'x', fields: {}); + expect(Record.fromJson(record.toJson()), equals(record)); + }); + }); +} diff --git a/crdt_sync_dart/test/sync_test.dart b/crdt_sync_dart/test/sync_test.dart new file mode 100644 index 0000000..c906d80 --- /dev/null +++ b/crdt_sync_dart/test/sync_test.dart @@ -0,0 +1,192 @@ +import 'dart:convert'; + +import 'package:crdt_sync/crdt_sync.dart'; +import 'package:http/http.dart' as http; +import 'package:http/testing.dart' as http_testing; +import 'package:test/test.dart'; + +Hlc _make(int wallTimeMs, {int counter = 0, String nodeId = 'node-a'}) => + Hlc(wallTimeMs: wallTimeMs, counter: counter, nodeId: nodeId); + +String _encode(Log log) => + jsonEncode(log.map((id, record) => MapEntry(id, record.toJson()))); + +Log _decode(String text) => (jsonDecode(text) as Map).map( + (id, data) => MapEntry(id, Record.fromJson(data as Map)), +); + +http.Response _response(int statusCode, [Object? jsonBody]) => + http.Response(jsonEncode(jsonBody ?? {}), statusCode); + +class _PutCall { + _PutCall(this.path, this.body); + final String path; + final Map body; +} + +/// Builds a [GitHubClient] backed by an in-memory router: GET +/// `.../contents/` returns `contentResponses[key]` (404 if absent), the +/// bare repo-existence GET always succeeds, and every PUT is recorded into +/// [putCalls] and answered with 200. +({GitHubClient client, List<_PutCall> putCalls}) _client({ + Map contentResponses = const {}, +}) { + final putCalls = <_PutCall>[]; + final mock = http_testing.MockClient((request) async { + final path = request.url.path; + if (!path.contains('/contents/')) { + return _response(200); + } + final key = path.split('/contents/').last; + if (request.method == 'PUT') { + putCalls.add( + _PutCall(key, jsonDecode(request.body) as Map), + ); + return _response(200); + } + return contentResponses[key] ?? _response(404); + }); + final client = GitHubClient( + owner: 'kuhyx', + repo: 'crdt-sync-demo', + token: 'fake-token', + httpClient: mock, + ); + return (client: client, putCalls: putCalls); +} + +http.Response _directoryOf(List deviceIds) => + _response(200, deviceIds.map((id) => {'name': id, 'type': 'dir'}).toList()); + +http.Response _fileContaining(String text) => + _response(200, {'content': base64.encode(utf8.encode(text))}); + +void main() { + group('syncLog', () { + test('pushes the local log when no other devices have synced', () async { + final localLog = { + 'a': Record(id: 'a', fields: {'text': ('hello', _make(100))}), + }; + final (:client, :putCalls) = _client( + contentResponses: {'devices': _directoryOf([])}, + ); + + final merged = await syncLog( + client: client, + deviceId: 'pc', + pathPrefix: 'devices', + localLog: localLog, + encode: _encode, + decode: _decode, + ); + + expect(merged, equals(localLog)); + expect(putCalls, hasLength(1)); + expect(putCalls.single.path, 'devices/pc/log.json'); + }); + + test('skips its own device id when listing', () async { + final (:client, :putCalls) = _client( + contentResponses: { + 'devices': _directoryOf(['pc', 'phone']), + 'devices/phone/log.json': _fileContaining('{}'), + }, + ); + + await syncLog( + client: client, + deviceId: 'pc', + pathPrefix: 'devices', + localLog: {}, + encode: _encode, + decode: _decode, + ); + + expect(putCalls, hasLength(1)); + }); + + test('skips a device with no pushed file yet', () async { + final (:client, :putCalls) = _client( + contentResponses: { + 'devices': _directoryOf(['phone']), + }, + ); + + final merged = await syncLog( + client: client, + deviceId: 'pc', + pathPrefix: 'devices', + localLog: {}, + encode: _encode, + decode: _decode, + ); + + expect(merged, isEmpty); + expect(putCalls.single.body['content'], base64.encode(utf8.encode('{}'))); + }); + + test('skips a device whose pushed file is corrupt', () async { + final (:client, :putCalls) = _client( + contentResponses: { + 'devices': _directoryOf(['phone']), + 'devices/phone/log.json': _fileContaining('{not valid json'), + }, + ); + + final merged = await syncLog( + client: client, + deviceId: 'pc', + pathPrefix: 'devices', + localLog: {}, + encode: _encode, + decode: _decode, + ); + + expect(merged, isEmpty); + expect(putCalls, hasLength(1)); + }); + + test('merges in a remote device\'s entries', () async { + final remoteLog = { + 'b': Record(id: 'b', fields: {'text': ('from phone', _make(100))}), + }; + final (:client, :putCalls) = _client( + contentResponses: { + 'devices': _directoryOf(['phone']), + 'devices/phone/log.json': _fileContaining(_encode(remoteLog)), + }, + ); + + final merged = await syncLog( + client: client, + deviceId: 'pc', + pathPrefix: 'devices', + localLog: {}, + encode: _encode, + decode: _decode, + ); + + expect(merged, equals(remoteLog)); + }); + + test('uses a custom filename and commit message', () async { + final (:client, :putCalls) = _client( + contentResponses: {'devices': _directoryOf([])}, + ); + + await syncLog( + client: client, + deviceId: 'pc', + pathPrefix: 'devices', + localLog: {}, + encode: _encode, + decode: _decode, + filename: 'notes.json', + commitMessage: 'custom message', + ); + + expect(putCalls.single.path, 'devices/pc/notes.json'); + expect(putCalls.single.body['message'], 'custom message'); + }); + }); +}