mirror of
https://github.com/kuhyx/utils.git
synced 2026-11-10 17:47:24 +01:00
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017McuA1nAi6jBT1EaS5BrVf
This commit is contained in:
commit
4b66703280
6
crdt_sync_dart/.gitignore
vendored
Normal file
6
crdt_sync_dart/.gitignore
vendored
Normal file
@ -0,0 +1,6 @@
|
||||
.dart_tool/
|
||||
.packages
|
||||
build/
|
||||
pubspec.lock
|
||||
coverage/
|
||||
*.iml
|
||||
21
crdt_sync_dart/LICENSE
Normal file
21
crdt_sync_dart/LICENSE
Normal file
@ -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.
|
||||
75
crdt_sync_dart/README.md
Normal file
75
crdt_sync_dart/README.md
Normal file
@ -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<String, Record>`, 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<String, dynamic>).map(
|
||||
(k, v) => MapEntry(k, Record.fromJson(v as Map<String, dynamic>)),
|
||||
),
|
||||
);
|
||||
```
|
||||
|
||||
`syncLog` pulls every other device's last-pushed log from
|
||||
`<pathPrefix>/<other-device-id>/...`, 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
|
||||
```
|
||||
11
crdt_sync_dart/analysis_options.yaml
Normal file
11
crdt_sync_dart/analysis_options.yaml
Normal file
@ -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
|
||||
8
crdt_sync_dart/lib/crdt_sync.dart
Normal file
8
crdt_sync_dart/lib/crdt_sync.dart
Normal file
@ -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';
|
||||
171
crdt_sync_dart/lib/src/github_client.dart
Normal file
171
crdt_sync_dart/lib/src/github_client.dart
Normal file
@ -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
|
||||
/// `<pathPrefix>/<deviceId>/<filename>` 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<String, String> get _headers => {
|
||||
'Authorization': 'Bearer $_token',
|
||||
'Accept': 'application/vnd.github+json',
|
||||
};
|
||||
|
||||
Uri _contentsUri(String path) =>
|
||||
Uri.parse('$_apiBase/repos/$owner/$repo/contents/$path');
|
||||
|
||||
Future<http.Response> _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<bool> _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<void> _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<List<String>> 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<String, dynamic>>()
|
||||
.map((e) => e['name'])
|
||||
.whereType<String>()
|
||||
.toList();
|
||||
}
|
||||
|
||||
/// Returns the decoded text content at [path], or null if unused.
|
||||
Future<String?> 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<String, dynamic>;
|
||||
final content = (decoded['content'] as String? ?? '').replaceAll('\n', '');
|
||||
return utf8.decode(base64.decode(content));
|
||||
}
|
||||
|
||||
Future<String?> _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<String, dynamic>;
|
||||
return decoded['sha'] as String?;
|
||||
}
|
||||
|
||||
/// Creates or updates the file at [path] with [text].
|
||||
Future<void> putFileText(
|
||||
String path,
|
||||
String text, {
|
||||
required String message,
|
||||
}) async {
|
||||
final sha = await _existingSha(path);
|
||||
final body = <String, dynamic>{
|
||||
'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();
|
||||
}
|
||||
102
crdt_sync_dart/lib/src/hlc.dart
Normal file
102
crdt_sync_dart/lib/src/hlc.dart
Normal file
@ -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<Hlc> {
|
||||
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:
|
||||
/// `<iso8601-millis>Z-<counter:hex4>-<nodeId>`.
|
||||
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)';
|
||||
}
|
||||
20
crdt_sync_dart/lib/src/log.dart
Normal file
20
crdt_sync_dart/lib/src/log.dart
Normal file
@ -0,0 +1,20 @@
|
||||
import 'record.dart';
|
||||
|
||||
/// A collection of [Record]s keyed by id.
|
||||
typedef Log = Map<String, Record>;
|
||||
|
||||
/// 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<String, Record>.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;
|
||||
}
|
||||
142
crdt_sync_dart/lib/src/record.dart
Normal file
142
crdt_sync_dart/lib/src/record.dart
Normal file
@ -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<String, Field> 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<String, dynamic> toJson() => {
|
||||
'id': id,
|
||||
'fields': fields.map(
|
||||
(name, field) => MapEntry(name, [field.$1, field.$2.toStr()]),
|
||||
),
|
||||
'deleted': deleted,
|
||||
'deletedHlc': deletedHlc?.toStr(),
|
||||
};
|
||||
|
||||
factory Record.fromJson(Map<String, dynamic> json) {
|
||||
final rawFields = json['fields'] as Map<String, dynamic>;
|
||||
final fields = rawFields.map((name, value) {
|
||||
final pair = value as List<dynamic>;
|
||||
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<String, Field> a, Map<String, Field> 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<String, Field>.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),
|
||||
);
|
||||
}
|
||||
48
crdt_sync_dart/lib/src/sync.dart
Normal file
48
crdt_sync_dart/lib/src/sync.dart
Normal file
@ -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 `<pathPrefix>/<other-device-id>/<filename>` for every device
|
||||
/// directory GitHub reports under [pathPrefix], merges each into
|
||||
/// [localLog] with [mergeLogs], then pushes this device's own merged result
|
||||
/// to `<pathPrefix>/<deviceId>/<filename>`.
|
||||
///
|
||||
/// [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<Log> 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<String, Record>.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;
|
||||
}
|
||||
16
crdt_sync_dart/pubspec.yaml
Normal file
16
crdt_sync_dart/pubspec.yaml
Normal file
@ -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
|
||||
195
crdt_sync_dart/test/github_client_test.dart
Normal file
195
crdt_sync_dart/test/github_client_test.dart
Normal file
@ -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<Object> 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<RepoNotFoundError>()),
|
||||
);
|
||||
});
|
||||
|
||||
test('throws GitHubSyncError on a non-2xx non-404', () async {
|
||||
final client = _client([_response(500)]);
|
||||
expect(
|
||||
() => client.getFileText('devices/pc/log.json'),
|
||||
throwsA(isA<GitHubSyncError>()),
|
||||
);
|
||||
});
|
||||
|
||||
test('throws GitHubSyncError on a network exception', () async {
|
||||
final client = _client([_networkError]);
|
||||
expect(
|
||||
() => client.getFileText('devices/pc/log.json'),
|
||||
throwsA(isA<GitHubSyncError>()),
|
||||
);
|
||||
});
|
||||
|
||||
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<RepoNotFoundError>()),
|
||||
);
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
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<RepoNotFoundError>()),
|
||||
);
|
||||
});
|
||||
|
||||
test('throws GitHubSyncError on a non-2xx non-404', () async {
|
||||
final client = _client([_response(500)]);
|
||||
expect(
|
||||
() => client.listDirectory('devices'),
|
||||
throwsA(isA<GitHubSyncError>()),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
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<RepoNotFoundError>()),
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
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<GitHubSyncError>()),
|
||||
);
|
||||
});
|
||||
|
||||
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<GitHubSyncError>()),
|
||||
);
|
||||
});
|
||||
|
||||
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<GitHubSyncError>()),
|
||||
);
|
||||
});
|
||||
});
|
||||
}
|
||||
117
crdt_sync_dart/test/hlc_test.dart
Normal file
117
crdt_sync_dart/test/hlc_test.dart
Normal file
@ -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,
|
||||
);
|
||||
});
|
||||
});
|
||||
}
|
||||
85
crdt_sync_dart/test/log_test.dart
Normal file
85
crdt_sync_dart/test/log_test.dart
Normal file
@ -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 = <String, Record>{
|
||||
'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 = <String, Record>{
|
||||
'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 = <String, Record>{
|
||||
'a': Record(id: 'a', fields: {'text': ('a', _make(100))}),
|
||||
};
|
||||
final remote = <String, Record>{
|
||||
'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 = <String, Record>{
|
||||
'a': Record(id: 'a', fields: {'text': ('old', _make(100))}),
|
||||
};
|
||||
final remote = <String, Record>{
|
||||
'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 = <String, Record>{
|
||||
'a': Record(id: 'a', fields: {}, deleted: true, deletedHlc: _make(200)),
|
||||
};
|
||||
final remote = <String, Record>{
|
||||
'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 = <String, Record>{
|
||||
'a': Record(id: 'a', fields: {'text': ('a', _make(100))}),
|
||||
};
|
||||
final remote = <String, Record>{
|
||||
'b': Record(id: 'b', fields: {'text': ('b', _make(100))}),
|
||||
};
|
||||
final localBefore = Map<String, Record>.from(local);
|
||||
final remoteBefore = Map<String, Record>.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));
|
||||
});
|
||||
});
|
||||
}
|
||||
141
crdt_sync_dart/test/record_test.dart
Normal file
141
crdt_sync_dart/test/record_test.dart
Normal file
@ -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));
|
||||
});
|
||||
});
|
||||
}
|
||||
192
crdt_sync_dart/test/sync_test.dart
Normal file
192
crdt_sync_dart/test/sync_test.dart
Normal file
@ -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<String, dynamic>).map(
|
||||
(id, data) => MapEntry(id, Record.fromJson(data as Map<String, dynamic>)),
|
||||
);
|
||||
|
||||
http.Response _response(int statusCode, [Object? jsonBody]) =>
|
||||
http.Response(jsonEncode(jsonBody ?? {}), statusCode);
|
||||
|
||||
class _PutCall {
|
||||
_PutCall(this.path, this.body);
|
||||
final String path;
|
||||
final Map<String, dynamic> body;
|
||||
}
|
||||
|
||||
/// Builds a [GitHubClient] backed by an in-memory router: GET
|
||||
/// `.../contents/<key>` 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<String, http.Response> 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<String, dynamic>),
|
||||
);
|
||||
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<String> 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 = <String, Record>{
|
||||
'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 = <String, Record>{
|
||||
'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');
|
||||
});
|
||||
});
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user