From 5be99ab11d1657ec04edd65508b8c4ecae4da906 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sat, 11 Jul 2026 22:19:34 +0200 Subject: [PATCH] Add LogStore local-persistence layer to crdt_sync (v0.2.0) crdt_sync_dart: new domain-agnostic LogStore -- an in-memory Log with a reactive changes stream and upsert/delete(sticky tombstone)/replaceAll, behind an injectable LogPersistence port -- plus logToJson/logFromJson. FileLogPersistence (atomic temp-file-then-rename, generalizing diet_guard's LogStorageService) lives behind a separate crdt_sync_io.dart entrypoint so the core barrel stays pure Dart / web-safe. crdt-sync (Python): mirror dump_log/load_log/read_log/write_log sharing the same on-disk JSON format so a log written on one side parses on the other. Load/save only -- the reactive store is a Dart/Flutter convenience with no Python equivalent by design (Python consumers are headless sync ticks). Querying stays app-side: a Record's fields are opaque to the library, so filtering/sorting/search is the caller's job over LogStore.values. Both packages bumped to 0.2.0. 83 Dart tests + 71 Python tests (100% branch coverage) green. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_019KwUCPESvE5CjCtmUFz8xQ --- crdt-sync/README.md | 20 ++ crdt-sync/crdt_sync/__init__.py | 5 + crdt-sync/crdt_sync/_store.py | 70 +++++++ crdt-sync/crdt_sync/tests/test_store.py | 83 +++++++++ crdt-sync/pyproject.toml | 2 +- crdt_sync_dart/README.md | 36 ++++ crdt_sync_dart/lib/crdt_sync.dart | 1 + crdt_sync_dart/lib/crdt_sync_io.dart | 8 + crdt_sync_dart/lib/src/io_store.dart | 40 ++++ crdt_sync_dart/lib/src/store.dart | 183 ++++++++++++++++++ crdt_sync_dart/pubspec.yaml | 2 +- crdt_sync_dart/test/store_test.dart | 236 ++++++++++++++++++++++++ 12 files changed, 684 insertions(+), 2 deletions(-) create mode 100644 crdt-sync/crdt_sync/_store.py create mode 100644 crdt-sync/crdt_sync/tests/test_store.py create mode 100644 crdt_sync_dart/lib/crdt_sync_io.dart create mode 100644 crdt_sync_dart/lib/src/io_store.dart create mode 100644 crdt_sync_dart/lib/src/store.dart create mode 100644 crdt_sync_dart/test/store_test.dart diff --git a/crdt-sync/README.md b/crdt-sync/README.md index 9dab170..3ffe6e2 100644 --- a/crdt-sync/README.md +++ b/crdt-sync/README.md @@ -90,6 +90,26 @@ pull-all/merge/push-own pattern diet_guard's `_sync.py` already used, made domain-agnostic via the `encode`/`decode` callbacks so callers keep their own on-disk JSON shape instead of `crdt-sync` dictating one. +## Local persistence + +For persisting a `Log` to disk, `crdt_sync` ships load/save helpers that share +a canonical on-disk JSON shape with the Dart `crdt_sync_dart.LogStore`, so a log +written by one language parses in the other: + +```python +from crdt_sync import dump_log, load_log, read_log, write_log + +write_log(path, log) # atomic temp-file-then-rename +log = read_log(path) # empty on a missing or corrupt file +text = dump_log(log) # -> canonical JSON string +log = load_log(text) # <- parse it back +``` + +These are load/save only. The Dart side additionally offers a *reactive* +`LogStore` (a change stream to drive a live UI); that's a Flutter-only +convenience and has no Python equivalent by design — Python consumers here are +headless sync ticks. + ## Development ```bash diff --git a/crdt-sync/crdt_sync/__init__.py b/crdt-sync/crdt_sync/__init__.py index e6d96d2..031b85c 100644 --- a/crdt-sync/crdt_sync/__init__.py +++ b/crdt-sync/crdt_sync/__init__.py @@ -6,6 +6,7 @@ from crdt_sync._github import GitHubSyncClient, GitHubSyncError, RepoNotFoundErr from crdt_sync._hlc import Hlc from crdt_sync._log import Log, merge_logs from crdt_sync._record import Field, Record, merge_field, merge_record +from crdt_sync._store import dump_log, load_log, read_log, write_log from crdt_sync._sync import sync_log __all__ = [ @@ -16,8 +17,12 @@ __all__ = [ "Log", "Record", "RepoNotFoundError", + "dump_log", + "load_log", "merge_field", "merge_logs", "merge_record", + "read_log", "sync_log", + "write_log", ] diff --git a/crdt-sync/crdt_sync/_store.py b/crdt-sync/crdt_sync/_store.py new file mode 100644 index 0000000..10f4572 --- /dev/null +++ b/crdt-sync/crdt_sync/_store.py @@ -0,0 +1,70 @@ +"""Local JSON persistence for a ``Log``. + +Mirrors ``crdt_sync_dart``'s ``logToJson``/``FileLogPersistence`` serialization +so a log written by the Dart side parses here and vice versa. Deliberately +*non-reactive*: the change-stream/reactive store is a Dart/Flutter-only +convenience (it exists to drive a live UI). On the Python side, where consumers +are headless sync ticks, this is plain load and (atomic) save. +""" + +from __future__ import annotations + +import json +import os +from typing import TYPE_CHECKING + +from crdt_sync._record import Record + +if TYPE_CHECKING: + from pathlib import Path + + from crdt_sync._log import Log + + +def dump_log(log: Log) -> str: + """Serialize ``log`` to canonical JSON text (id -> record dict).""" + return json.dumps({rid: record.to_dict() for rid, record in log.items()}) + + +def load_log(text: str) -> Log: + """Parse text produced by :func:`dump_log` back into a ``Log``. + + Raises ``ValueError`` (bad JSON), ``KeyError`` or ``TypeError`` (valid + JSON, wrong shape) on a corrupt payload -- callers reading from disk go + through :func:`read_log`, which turns those into an empty log. + """ + raw = json.loads(text) + if not isinstance(raw, dict): + msg = f"expected a JSON object mapping id -> record, got {type(raw).__name__}" + raise TypeError(msg) + return {rid: Record.from_dict(value) for rid, value in raw.items()} + + +def read_log(path: Path) -> Log: + """Read a ``Log`` from ``path``; empty on a missing or unparsable file. + + A truncated or corrupt file can never raise here: it is treated as "no log + yet", matching the Dart store's defensive load so a half-written file can't + brick a consumer. + """ + try: + text = path.read_text(encoding="utf-8") + except FileNotFoundError: + return {} + try: + return load_log(text) + except (ValueError, KeyError, TypeError): + return {} + + +def write_log(path: Path, log: Log) -> None: + """Write ``log`` to ``path`` atomically (temp file then rename). + + Writes to a per-process temp file and ``replace``s it over the real path, + so a concurrent reader never observes a half-written file. Mirrors the + Dart ``FileLogPersistence`` write scheme. + """ + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_name(f"{path.name}.{os.getpid()}.tmp") + tmp.write_text(dump_log(log), encoding="utf-8") + tmp.replace(path) diff --git a/crdt-sync/crdt_sync/tests/test_store.py b/crdt-sync/crdt_sync/tests/test_store.py new file mode 100644 index 0000000..5955f34 --- /dev/null +++ b/crdt-sync/crdt_sync/tests/test_store.py @@ -0,0 +1,83 @@ +"""Tests for local Log persistence (dump/load and atomic file read/write).""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +import pytest + +from crdt_sync import ( + Record, + dump_log, + load_log, + read_log, + write_log, +) + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + + from crdt_sync import Hlc, Log + + +def _sample_log(make_hlc: Callable[..., Hlc]) -> Log: + return { + "a": Record(id="a", fields={"text": ("alpha", make_hlc(100))}), + "b": Record(id="b", fields={}, deleted=True, deleted_hlc=make_hlc(200)), + } + + +class TestDumpLoad: + def test_round_trips_a_log(self, make_hlc: Callable[..., Hlc]) -> None: + log = _sample_log(make_hlc) + assert load_log(dump_log(log)) == log + + def test_load_raises_on_non_json(self) -> None: + # json.loads raises JSONDecodeError (a ValueError subclass). + with pytest.raises(json.JSONDecodeError): + load_log("{not json") + + def test_load_raises_typeerror_on_wrong_shape(self) -> None: + # Valid JSON, but a list where an id->record map is expected. + with pytest.raises(TypeError): + load_log("[]") + + +class TestReadWrite: + def test_write_then_read_round_trips( + self, tmp_path: Path, make_hlc: Callable[..., Hlc] + ) -> None: + log = _sample_log(make_hlc) + path = tmp_path / "nested" / "log.json" + write_log(path, log) + assert read_log(path) == log + + def test_write_creates_parent_directories( + self, tmp_path: Path, make_hlc: Callable[..., Hlc] + ) -> None: + path = tmp_path / "deep" / "nested" / "log.json" + write_log(path, _sample_log(make_hlc)) + assert path.exists() + + def test_write_overwrites_atomically( + self, tmp_path: Path, make_hlc: Callable[..., Hlc] + ) -> None: + path = tmp_path / "log.json" + write_log(path, {"a": Record(id="a", fields={})}) + write_log(path, _sample_log(make_hlc)) + assert set(read_log(path)) == {"a", "b"} + + def test_read_of_missing_file_is_empty(self, tmp_path: Path) -> None: + assert read_log(tmp_path / "nope.json") == {} + + def test_read_of_corrupt_file_is_empty(self, tmp_path: Path) -> None: + path = tmp_path / "log.json" + path.write_text("{not json", encoding="utf-8") + assert read_log(path) == {} + + def test_read_of_wrong_shape_file_is_empty(self, tmp_path: Path) -> None: + path = tmp_path / "log.json" + path.write_text("[]", encoding="utf-8") + assert read_log(path) == {} diff --git a/crdt-sync/pyproject.toml b/crdt-sync/pyproject.toml index 056469e..882a888 100644 --- a/crdt-sync/pyproject.toml +++ b/crdt-sync/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "crdt-sync" -version = "0.1.0" +version = "0.2.0" description = "Shared CRDT merge scheme + GitHub-Contents-API sync transport for cross-device apps" requires-python = ">=3.10" dependencies = ["requests"] diff --git a/crdt_sync_dart/README.md b/crdt_sync_dart/README.md index a466940..3d1fd00 100644 --- a/crdt_sync_dart/README.md +++ b/crdt_sync_dart/README.md @@ -66,6 +66,42 @@ final merged = await syncLog( `//...`, merges each into the local log with `mergeLogs`, then pushes this device's own merged result back up. +## Local persistence (`LogStore`) + +The merge scheme above is in-memory only. For an app that needs to keep its +`Log` on disk and drive a live UI off it, this package also ships a +domain-agnostic `LogStore`: + +```dart +import 'package:crdt_sync/crdt_sync.dart'; +import 'package:crdt_sync/crdt_sync_io.dart'; // FileLogPersistence (dart:io) + +final store = LogStore( + persistence: FileLogPersistence(File('$docsDir/notes.json')), + nodeId: 'phone', +); +await store.load(); + +// Reactive read: re-derive a domain view whenever anything changes. Filtering, +// sorting and search live here, in the app -- the store is field-agnostic. +final Stream> visible = store.changes.map( + (_) => store.values.where((r) => !r.deleted).toList(), +); + +await store.upsert(Record(id: 'n1', fields: {'text': ('hi', store.nextHlc())})); +await store.delete('n1'); // sticky tombstone, not a hard remove + +// After a sync tick, swap in the merged result: +await store.replaceAll(await syncLog(/* ... */, localLog: store.snapshot())); +``` + +`LogStore` is pure Dart (import it from the main barrel); the filesystem-backed +`FileLogPersistence` lives behind the separate `crdt_sync_io.dart` entrypoint so +the core stays web-safe. Querying is deliberately **not** in the library — a +`Record`'s fields are an opaque map, so the app filters `store.values` itself. +The Python side mirrors the on-disk format via `crdt_sync.dump_log`/`read_log` +(load/save only; no reactive stream — that's a Dart/Flutter convenience). + ## Development ```bash diff --git a/crdt_sync_dart/lib/crdt_sync.dart b/crdt_sync_dart/lib/crdt_sync.dart index ed94c6d..6a934fb 100644 --- a/crdt_sync_dart/lib/crdt_sync.dart +++ b/crdt_sync_dart/lib/crdt_sync.dart @@ -5,4 +5,5 @@ export 'src/github_client.dart'; export 'src/hlc.dart'; export 'src/log.dart'; export 'src/record.dart'; +export 'src/store.dart'; export 'src/sync.dart'; diff --git a/crdt_sync_dart/lib/crdt_sync_io.dart b/crdt_sync_dart/lib/crdt_sync_io.dart new file mode 100644 index 0000000..6b8ecc0 --- /dev/null +++ b/crdt_sync_dart/lib/crdt_sync_io.dart @@ -0,0 +1,8 @@ +/// Filesystem-backed persistence for `LogStore`. +/// +/// Import this instead of (in addition to) `package:crdt_sync/crdt_sync.dart` +/// on platforms with a filesystem (mobile, desktop, server). It pulls in +/// `dart:io`, so it is unavailable on web; the main barrel stays pure Dart. +library; + +export 'src/io_store.dart'; diff --git a/crdt_sync_dart/lib/src/io_store.dart b/crdt_sync_dart/lib/src/io_store.dart new file mode 100644 index 0000000..89795db --- /dev/null +++ b/crdt_sync_dart/lib/src/io_store.dart @@ -0,0 +1,40 @@ +import 'dart:io'; + +import 'store.dart'; + +/// A [LogPersistence] backed by a single JSON file. +/// +/// Writes atomically -- to a per-process temp file, then `rename`d over the +/// real path -- so a concurrent reader (e.g. a background sync isolate writing +/// while the foreground app also writes) never observes a half-written file. +/// The pid in the temp name keeps two writers from colliding. This generalizes +/// diet_guard's hand-rolled `LogStorageService` write scheme. +/// +/// Lives behind the `package:crdt_sync/crdt_sync_io.dart` entrypoint, not the +/// main barrel, because it imports `dart:io` and so is unavailable on web -- +/// the core [LogStore] stays pure Dart. +class FileLogPersistence implements LogPersistence { + /// Persists to [file]. The app chooses the path (e.g. via `path_provider`); + /// the library takes no dependency on where app documents live. + FileLogPersistence(this._file); + + final File _file; + + @override + Future read() async { + if (!_file.existsSync()) return null; + try { + return await _file.readAsString(); + } on FileSystemException { + return null; + } + } + + @override + Future write(String text) async { + await _file.parent.create(recursive: true); + final tmp = File('${_file.path}.$pid.tmp'); + await tmp.writeAsString(text); + await tmp.rename(_file.path); + } +} diff --git a/crdt_sync_dart/lib/src/store.dart b/crdt_sync_dart/lib/src/store.dart new file mode 100644 index 0000000..70b7a26 --- /dev/null +++ b/crdt_sync_dart/lib/src/store.dart @@ -0,0 +1,183 @@ +import 'dart:async'; +import 'dart:convert'; + +import 'hlc.dart'; +import 'log.dart'; +import 'record.dart'; + +/// Serializes a [Log] to canonical JSON text: a top-level object mapping each +/// record id to its [Record.toJson]. Mirrors `crdt_sync._store.dump_log` on +/// the Python side, so a log written by one language parses in the other. +String logToJson(Log log) => jsonEncode({ + for (final entry in log.entries) entry.key: entry.value.toJson(), +}); + +/// Parses text produced by [logToJson] back into a [Log]. +/// +/// Throws [FormatException] (bad JSON) or [TypeError] (valid JSON, wrong +/// shape) on a corrupt payload -- callers persisting to disk should treat +/// both as "start empty", see [LogPersistence]-backed [LogStore.load]. +Log logFromJson(String text) { + final raw = jsonDecode(text) as Map; + return raw.map( + (id, value) => + MapEntry(id, Record.fromJson((value as Map).cast())), + ); +} + +/// A durable, string-in/string-out persistence port for a [LogStore]. +/// +/// Kept deliberately free of `dart:io` so [LogStore] stays pure Dart (usable +/// on web and trivially faked in tests). A filesystem-backed implementation +/// lives behind the separate `package:crdt_sync/crdt_sync_io.dart` entrypoint; +/// tests inject an in-memory one. +abstract class LogPersistence { + /// Returns the stored text, or `null` if nothing has been persisted yet. + Future read(); + + /// Overwrites the stored text with [text]. + Future write(String text); +} + +/// An in-memory [Log] with reactive change notifications, backed by a +/// [LogPersistence] port. +/// +/// This is the local-persistence half that the merge scheme itself +/// (`Hlc`/`Record`/`Log`) deliberately omits. It is domain-agnostic: +/// filtering, sorting and searching over [values] is the caller's job -- the +/// store knows nothing about what fields a record carries. A consumer's typical +/// reactive read is `store.changes.map((_) => store.values.where(pred))`. +/// +/// The store owns this node's logical clock: every mutation it performs +/// (a [delete] tombstone) is stamped with a monotonic [Hlc] for [nodeId], and +/// callers building their own records should take field clocks from [nextHlc] +/// so the whole device shares one monotonic sequence. +class LogStore { + /// Creates a store for [nodeId] persisting through [persistence]. + /// + /// Call [load] once before reading [values] to hydrate from storage. + // Dart forbids private named params, so these can't be initializing + // formals; assign them explicitly. + // ignore_for_file: prefer_initializing_formals + LogStore({required LogPersistence persistence, required String nodeId}) + : _persistence = persistence, + _nodeId = nodeId; + + final LogPersistence _persistence; + final String _nodeId; + final Log _log = {}; + final StreamController _changes = StreamController.broadcast(); + Hlc? _lastHlc; + + /// Fires (with no payload) after every mutation that reaches storage. + /// + /// A UI listens here and re-derives its view from [values]; the event + /// carries nothing because the store is not the place that knows how to + /// shape a domain view. + Stream get changes => _changes.stream; + + /// The current records, tombstones included. Callers filter as needed. + Iterable get values => _log.values; + + /// Returns the record for [id], or `null` if absent. + Record? get(String id) => _log[id]; + + /// An unmodifiable snapshot of the whole log, e.g. to hand to `syncLog`. + Log snapshot() => Map.unmodifiable(_log); + + /// Returns the next monotonic [Hlc] for this node, advancing the clock. + /// + /// Callers stamp their own record fields with this so a device's writes and + /// the store's own delete tombstones share one strictly increasing sequence. + Hlc nextHlc() { + final next = Hlc.newTick(_nodeId, previous: _lastHlc); + _lastHlc = next; + return next; + } + + /// Hydrates the in-memory log from storage, returning a snapshot. + /// + /// A missing or corrupt payload is treated as an empty log (never an error), + /// so a truncated write can't brick the app -- it just loses the unwritten + /// tail, exactly as the file-storage mirror in diet_guard does. + Future load() async { + final text = await _persistence.read(); + _log.clear(); + if (text != null) { + try { + _log.addAll(logFromJson(text)); + } on FormatException { + _log.clear(); + } on TypeError { + _log.clear(); + } + } + _trackLatestHlc(); + return snapshot(); + } + + /// Inserts or replaces [record] and persists. + /// + /// The caller owns the record's field clocks (take them from [nextHlc]); + /// the store does not re-stamp them, so an unchanged re-upsert is a no-op + /// under a later merge. + Future upsert(Record record) async { + _log[record.id] = record; + _trackHlcsOf(record); + await _persist(); + } + + /// Tombstones the record with [id] in place (sticky CRDT delete) and + /// persists. No-op if [id] is absent or already deleted. + /// + /// Tombstoning rather than removing is what stops a delete from being + /// silently undone when an older, still-present copy is pulled from another + /// device on the next sync. + Future delete(String id) async { + final existing = _log[id]; + if (existing == null || existing.deleted) return; + _log[id] = Record( + id: existing.id, + fields: existing.fields, + deleted: true, + deletedHlc: nextHlc(), + ); + await _persist(); + } + + /// Replaces the entire log (e.g. with a post-merge result from `syncLog`) + /// and persists. + Future replaceAll(Log merged) async { + _log + ..clear() + ..addAll(merged); + _trackLatestHlc(); + await _persist(); + } + + /// Releases the [changes] stream. The store is unusable afterwards. + Future close() => _changes.close(); + + Future _persist() async { + await _persistence.write(logToJson(_log)); + if (!_changes.isClosed) _changes.add(null); + } + + void _trackLatestHlc() { + for (final record in _log.values) { + _trackHlcsOf(record); + } + } + + void _trackHlcsOf(Record record) { + for (final field in record.fields.values) { + _bumpLastHlc(field.$2); + } + final deletedHlc = record.deletedHlc; + if (deletedHlc != null) _bumpLastHlc(deletedHlc); + } + + void _bumpLastHlc(Hlc candidate) { + if (_lastHlc == null || candidate > _lastHlc!) _lastHlc = candidate; + } +} diff --git a/crdt_sync_dart/pubspec.yaml b/crdt_sync_dart/pubspec.yaml index f1d6505..d8fbb39 100644 --- a/crdt_sync_dart/pubspec.yaml +++ b/crdt_sync_dart/pubspec.yaml @@ -2,7 +2,7 @@ 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 +version: 0.2.0 publish_to: 'none' environment: diff --git a/crdt_sync_dart/test/store_test.dart b/crdt_sync_dart/test/store_test.dart new file mode 100644 index 0000000..d7dcfe1 --- /dev/null +++ b/crdt_sync_dart/test/store_test.dart @@ -0,0 +1,236 @@ +import 'dart:io'; + +import 'package:crdt_sync/crdt_sync.dart'; +import 'package:crdt_sync/crdt_sync_io.dart'; +import 'package:test/test.dart'; + +Hlc _make(int wallTimeMs, {int counter = 0, String nodeId = 'node-a'}) => + Hlc(wallTimeMs: wallTimeMs, counter: counter, nodeId: nodeId); + +Record _rec(String id, String text, Hlc hlc) => + Record(id: id, fields: {'text': (text, hlc)}); + +/// An in-memory [LogPersistence] so the core store is testable without disk. +class _MemoryPersistence implements LogPersistence { + String? text; + int writes = 0; + + @override + Future read() async => text; + + @override + Future write(String value) async { + text = value; + writes++; + } +} + +void main() { + group('logToJson / logFromJson', () { + test('round-trips a log', () { + final log = { + 'a': _rec('a', 'alpha', _make(100)), + 'b': Record(id: 'b', fields: {}, deleted: true, deletedHlc: _make(200)), + }; + expect(logFromJson(logToJson(log)), equals(log)); + }); + + test('throws FormatException on non-JSON text', () { + expect(() => logFromJson('{not json'), throwsFormatException); + }); + + test('throws TypeError on valid JSON of the wrong shape', () { + expect(() => logFromJson('[]'), throwsA(isA())); + }); + }); + + group('LogStore.load', () { + test('is empty when nothing was ever persisted', () async { + final store = LogStore(persistence: _MemoryPersistence(), nodeId: 'n'); + expect(await store.load(), isEmpty); + expect(store.values, isEmpty); + }); + + test('hydrates from a valid stored payload', () async { + final persistence = _MemoryPersistence() + ..text = logToJson({'a': _rec('a', 'alpha', _make(100))}); + final store = LogStore(persistence: persistence, nodeId: 'n'); + final log = await store.load(); + expect(log.keys, ['a']); + expect(store.get('a')!.fields['text']!.$1, 'alpha'); + }); + + test('treats a syntactically corrupt payload as empty', () async { + final persistence = _MemoryPersistence()..text = '{not json'; + final store = LogStore(persistence: persistence, nodeId: 'n'); + expect(await store.load(), isEmpty); + }); + + test('treats a wrong-shape payload as empty', () async { + final persistence = _MemoryPersistence()..text = '[]'; + final store = LogStore(persistence: persistence, nodeId: 'n'); + expect(await store.load(), isEmpty); + }); + }); + + group('LogStore mutations', () { + test('upsert stores the record and persists', () async { + final persistence = _MemoryPersistence(); + final store = LogStore(persistence: persistence, nodeId: 'n'); + await store.load(); + final record = _rec('a', 'alpha', store.nextHlc()); + await store.upsert(record); + expect(store.get('a'), record); + expect(persistence.writes, 1); + expect(logFromJson(persistence.text!)['a'], record); + }); + + test('delete tombstones in place and persists', () async { + final store = LogStore(persistence: _MemoryPersistence(), nodeId: 'n'); + await store.load(); + await store.upsert(_rec('a', 'alpha', store.nextHlc())); + await store.delete('a'); + final tombstone = store.get('a')!; + expect(tombstone.deleted, isTrue); + expect(tombstone.deletedHlc, isNotNull); + expect(tombstone.fields['text']!.$1, 'alpha'); // fields retained + }); + + test('delete is a no-op for an absent id', () async { + final persistence = _MemoryPersistence(); + final store = LogStore(persistence: persistence, nodeId: 'n'); + await store.load(); + await store.delete('missing'); + expect(persistence.writes, 0); + }); + + test('delete is a no-op for an already-tombstoned record', () async { + final persistence = _MemoryPersistence(); + final store = LogStore(persistence: persistence, nodeId: 'n'); + await store.load(); + await store.upsert(_rec('a', 'alpha', store.nextHlc())); + await store.delete('a'); + final writesAfterFirstDelete = persistence.writes; + await store.delete('a'); + expect(persistence.writes, writesAfterFirstDelete); + }); + + test('replaceAll swaps the whole log and persists', () async { + final persistence = _MemoryPersistence(); + final store = LogStore(persistence: persistence, nodeId: 'n'); + await store.load(); + await store.upsert(_rec('a', 'alpha', store.nextHlc())); + final merged = {'b': _rec('b', 'beta', _make(500))}; + await store.replaceAll(merged); + expect(store.values.map((r) => r.id), ['b']); + expect(logFromJson(persistence.text!), merged); + }); + }); + + group('LogStore clock & snapshot', () { + test('nextHlc is strictly monotonic', () async { + final store = LogStore(persistence: _MemoryPersistence(), nodeId: 'n'); + final first = store.nextHlc(); + final second = store.nextHlc(); + expect(second > first, isTrue); + }); + + test('nextHlc stays ahead of hydrated field clocks', () async { + final far = _make(1 << 40); // a clock far in the future + final persistence = _MemoryPersistence() + ..text = logToJson({'a': _rec('a', 'alpha', far)}); + final store = LogStore(persistence: persistence, nodeId: 'n'); + await store.load(); + expect(store.nextHlc() > far, isTrue); + }); + + test('snapshot is unmodifiable', () async { + final store = LogStore(persistence: _MemoryPersistence(), nodeId: 'n'); + await store.load(); + expect( + () => store.snapshot()['x'] = _rec('x', 'y', _make(1)), + throwsUnsupportedError, + ); + }); + + test('get returns null for an absent id', () async { + final store = LogStore(persistence: _MemoryPersistence(), nodeId: 'n'); + await store.load(); + expect(store.get('nope'), isNull); + }); + }); + + group('LogStore.changes', () { + test('fires once per mutation', () async { + final store = LogStore(persistence: _MemoryPersistence(), nodeId: 'n'); + await store.load(); + final events = []; + final sub = store.changes.listen(events.add); + await store.upsert(_rec('a', 'alpha', store.nextHlc())); + await store.delete('a'); + await Future.delayed(Duration.zero); + expect(events.length, 2); + await sub.cancel(); + }); + + test('close ends the stream and later writes do not throw', () async { + final persistence = _MemoryPersistence(); + final store = LogStore(persistence: persistence, nodeId: 'n'); + await store.load(); + await store.close(); + // isClosed guard: the write still lands, the (closed) stream is skipped. + await store.upsert(_rec('a', 'alpha', _make(1))); + expect(persistence.writes, 1); + }); + }); + + group('FileLogPersistence', () { + late Directory dir; + + setUp(() async { + dir = await Directory.systemTemp.createTemp('crdt_sync_store_test'); + }); + + tearDown(() async { + if (dir.existsSync()) await dir.delete(recursive: true); + }); + + test('read returns null when the file does not exist', () async { + final persistence = FileLogPersistence(File('${dir.path}/nope.json')); + expect(await persistence.read(), isNull); + }); + + test('write then read round-trips and creates parent dirs', () async { + final path = '${dir.path}/nested/deep/log.json'; + final persistence = FileLogPersistence(File(path)); + await persistence.write('hello'); + expect(await persistence.read(), 'hello'); + expect(File(path).existsSync(), isTrue); + }); + + test('write overwrites atomically', () async { + final file = File('${dir.path}/log.json'); + final persistence = FileLogPersistence(file); + await persistence.write('first'); + await persistence.write('second'); + expect(await persistence.read(), 'second'); + }); + + test('backs a LogStore end to end', () async { + final file = File('${dir.path}/store.json'); + final store = LogStore( + persistence: FileLogPersistence(file), + nodeId: 'device-1', + ); + await store.load(); + await store.upsert(_rec('a', 'alpha', store.nextHlc())); + + final reopened = LogStore( + persistence: FileLogPersistence(file), + nodeId: 'device-1', + ); + final log = await reopened.load(); + expect(log['a']!.fields['text']!.$1, 'alpha'); + }); + }); +}