From 5c1bf75bf92e83e3bc4605f9206b1392056aa9d4 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Tue, 10 Nov 2026 15:38:54 +0100 Subject: [PATCH] Bulk note writes and assign-to-repo helper NoteRepository.bulkUpdate sets status/priority on many notes in one store write (mergeLogs + replaceAll; only changed fields plus updated_at get a clock). saveReplacing saves one note and tombstones others in one write. assignToRepo moves plain notes into the repo's open note as "## title" sections via addRepoTask, demoting body headings below "##" so a template note stays one section; notes already in a repo are refused. The seeded watch stream moved to log_watch.dart for the 250-line cap. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01WzxdWXdsB7CQPNXVN829G1 --- lib/data/log_watch.dart | 22 ++++++ lib/data/note_bulk.dart | 59 +++++++++++++++ lib/data/note_repository.dart | 62 ++++++++------- lib/data/repo_assign.dart | 75 +++++++++++++++++++ pubspec.yaml | 2 +- test/fake_note_repository.dart | 34 ++++++++- test/note_repository_bulk_test.dart | 112 ++++++++++++++++++++++++++++ test/repo_assign_test.dart | 109 +++++++++++++++++++++++++++ 8 files changed, 446 insertions(+), 29 deletions(-) create mode 100644 lib/data/log_watch.dart create mode 100644 lib/data/note_bulk.dart create mode 100644 lib/data/repo_assign.dart create mode 100644 test/note_repository_bulk_test.dart create mode 100644 test/repo_assign_test.dart diff --git a/lib/data/log_watch.dart b/lib/data/log_watch.dart new file mode 100644 index 0000000..be2b150 --- /dev/null +++ b/lib/data/log_watch.dart @@ -0,0 +1,22 @@ +/// A seeded stream over the note log's change events. +library; + +import 'dart:async'; + +/// A stream that emits `compute()` synchronously when first listened to (the +/// seed), then again on every event of [changes]. Seeding in `onListen` -- +/// rather than lazily in an `async*` body -- avoids a race where a write +/// between `listen` and the first generator tick would swallow the initial +/// snapshot. +Stream watchLog(Stream changes, T Function() compute) { + StreamSubscription? sub; + late final StreamController controller; + controller = StreamController( + onListen: () { + controller.add(compute()); + sub = changes.listen((_) => controller.add(compute())); + }, + onCancel: () => sub?.cancel(), + ); + return controller.stream; +} diff --git a/lib/data/note_bulk.dart b/lib/data/note_bulk.dart new file mode 100644 index 0000000..ce0b387 --- /dev/null +++ b/lib/data/note_bulk.dart @@ -0,0 +1,59 @@ +/// Partial logs for the multi-note writes: merged in one store write. +/// +/// Each write re-serialises the whole log (over a megabyte on a real +/// device), so a bulk action must be one partial log merged once, never one +/// upsert per note. Merging is per field, so a field left out keeps its value +/// and its clock (`created_by` included). +library; + +import 'package:crdt_sync/crdt_sync.dart'; +import 'package:todo/data/note.dart'; +import 'package:todo/data/note_record.dart'; + +/// Partial records setting [priority] and/or [status] on the live notes of +/// [log] among [ids], all at [hlc]. +/// +/// Only fields whose value changes are written, plus `updated_at` = [now] +/// (so "last updated" sorting and newer-wins import see the edit). A note +/// with nothing to change, an unknown id and a deleted note are left out. +Log planBulkUpdate( + Log log, + Set ids, { + required Hlc hlc, + required DateTime now, + Priority? priority, + Status? status, +}) { + final plan = {}; + for (final id in ids) { + final current = log[id]; + if (current == null || current.deleted) continue; + final note = noteFromRecord(current); + final next = note.copyWith(priority: priority, status: status); + if (next.priority == note.priority && next.status == note.status) continue; + plan[id] = Record( + id: id, + fields: { + for (final MapEntry(:key, :value) in noteFields( + next.copyWith(updatedAt: now), + hlc, + ).entries) + if (current.fields[key]?.$1 != value.$1) key: value, + }, + ); + } + return plan; +} + +/// [note] at [hlc] plus a tombstone for each live note of [log] in +/// [removed] (other than [note] itself). +/// +/// A merge, not a replace: a field [note] leaves unset keeps its stored +/// value, so this suits notes that only gain content (the repo note an +/// assign appends to), not edits that clear a field. +Log planReplace(Log log, Note note, Iterable removed, Hlc hlc) => { + for (final id in removed) + if (log[id] case final r? when !r.deleted && id != note.id) + id: Record(id: id, fields: const {}, deleted: true, deletedHlc: hlc), + note.id: Record(id: note.id, fields: noteFields(note, hlc)), +}; diff --git a/lib/data/note_repository.dart b/lib/data/note_repository.dart index aa27015..d6eff76 100644 --- a/lib/data/note_repository.dart +++ b/lib/data/note_repository.dart @@ -1,8 +1,8 @@ -import 'dart:async'; - import 'package:crdt_sync/crdt_sync.dart'; import 'package:todo/data/created_by_backfill.dart'; +import 'package:todo/data/log_watch.dart'; import 'package:todo/data/note.dart'; +import 'package:todo/data/note_bulk.dart'; import 'package:todo/data/note_import.dart'; import 'package:todo/data/note_query.dart'; import 'package:todo/data/note_record.dart'; @@ -134,15 +134,40 @@ class NoteRepository { final plan = planCreatedByBackfill(_store.snapshot(), claudeNodes); if (dryRun || plan.isEmpty) return plan; final hlc = _store.nextHlc(); - await _store.replaceAll( - mergeLogs(_store.snapshot(), { - for (final e in plan.entries) - e.key: Record(id: e.key, fields: {kCreatedByField: (e.value, hlc)}), - }), - ); + await _patch({ + for (final e in plan.entries) + e.key: Record(id: e.key, fields: {kCreatedByField: (e.value, hlc)}), + }); return plan; } + /// Sets [priority] and/or [status] on the live notes [ids] in ONE store + /// write (see [planBulkUpdate]); a no-op when nothing changes. + Future bulkUpdate( + Set ids, { + Priority? priority, + Status? status, + }) => _patch( + planBulkUpdate( + _store.snapshot(), + ids, + hlc: _store.nextHlc(), + now: DateTime.now(), + priority: priority, + status: status, + ), + ); + + /// Saves [note] and deletes [removed] in ONE store write ([planReplace]). + Future saveReplacing(Note note, Iterable removed) => + _patch(planReplace(_store.snapshot(), note, removed, _store.nextHlc())); + + /// Merges a partial log in one write: untouched fields keep their clocks. + Future _patch(Log partial) async { + if (partial.isEmpty) return; + await _store.replaceAll(mergeLogs(_store.snapshot(), partial)); + } + /// Soft-deletes a note. The log keeps a sticky tombstone so the deletion /// propagates on the next sync instead of resurrecting the record. Future delete(String id) => _store.delete(id); @@ -189,27 +214,10 @@ class NoteRepository { Stream> watchNotes({ NoteSort sort = NoteSort.priorityDesc, NoteFilter filter = const NoteFilter(), - }) => _watch(() => _query(sort, filter)); + }) => watchLog(_store.changes, () => _query(sort, filter)); /// Emits the live count of non-deleted notes immediately and on each change. - Stream watchCount() => _watch(_liveCount); - - /// A stream that emits `compute()` synchronously when first listened to (the - /// seed), then again on every log change. Seeding in `onListen` — rather than - /// lazily in an `async*` body — avoids a race where a write between `listen` - /// and the first generator tick would swallow the initial snapshot. - Stream _watch(T Function() compute) { - StreamSubscription? sub; - late final StreamController controller; - controller = StreamController( - onListen: () { - controller.add(compute()); - sub = _store.changes.listen((_) => controller.add(compute())); - }, - onCancel: () => sub?.cancel(), - ); - return controller.stream; - } + Stream watchCount() => watchLog(_store.changes, _liveCount); /// This device's stable node id. Names its file in the sync repo so two /// devices never write the same path. diff --git a/lib/data/repo_assign.dart b/lib/data/repo_assign.dart new file mode 100644 index 0000000..ad9bfb1 --- /dev/null +++ b/lib/data/repo_assign.dart @@ -0,0 +1,75 @@ +/// Assigning plain notes to a repo from the UI: each becomes a `## title` +/// section of the repo's one open note ([addRepoTask]), and the original goes. +/// +/// Only the data layer: nothing here talks to the board or starts a task. +library; + +import 'package:todo/data/note.dart'; +import 'package:todo/data/note_repository.dart'; +import 'package:todo/data/repo_note.dart'; + +final _repoShape = RegExp(r'^[\w.-]+/[\w.-]+$'); +final _heading = RegExp(r'^(#+)(\s.*)$'); + +/// Whether [value] looks like `owner/name`. +bool isRepoName(String value) => _repoShape.hasMatch(value.trim()); + +/// [body] with every heading outside code fences pushed below `##`. +/// +/// A repo note splits into tasks at `##` headings, so a template note's own +/// `## what` / `## must` would otherwise turn into tasks of their own. +String demoteHeadings(String body) { + var fenced = false; + final out = []; + for (final line in body.split('\n')) { + if (line.trimLeft().startsWith('```')) fenced = !fenced; + final m = fenced ? null : _heading.firstMatch(line); + if (m == null) { + out.add(line); + continue; + } + final level = m.group(1)!.length; + out.add('${'#' * (level < 2 ? 3 : level + 1)}${m.group(2)}'); + } + return out.join('\n'); +} + +/// Moves the plain [notes] into [repo]'s open note in one write, creating +/// that note when the repo has none; returns the notes refused because +/// they already belong to a repo (left untouched). +Future> assignToRepo( + NoteRepository repository, + List notes, + String repo, +) async { + final refused = [ + for (final n in notes) + if (n.repo != null) n, + ]; + final plain = [ + for (final n in notes) + if (n.repo == null) n, + ]; + if (plain.isEmpty) return refused; + final first = splitNote(plain.first.text); + var target = (await addRepoTask( + repository, + repo: repo, + title: first.title, + body: demoteHeadings(first.body), + priority: plain.first.priority, + createdBy: 'kuhy', + )).note; + for (final n in plain.skip(1)) { + final (:title, :body) = splitNote(n.text); + target = target.copyWith( + text: '${target.text}\n\n${repoSection(title, demoteHeadings(body))}', + priority: maxPriority(target.priority, n.priority), + ); + } + await repository.saveReplacing( + target.copyWith(updatedAt: DateTime.now()), + plain.map((n) => n.id), + ); + return refused; +} diff --git a/pubspec.yaml b/pubspec.yaml index 0122543..882df80 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -16,7 +16,7 @@ publish_to: 'none' # Remove this line if you wish to publish to pub.dev # https://developer.apple.com/library/archive/documentation/General/Reference/InfoPlistKeyReference/Articles/CoreFoundationKeys.html # In Windows, build-name is used as the major, minor, and patch parts # of the product and file versions while build-number is used as the build suffix. -version: 1.0.66+1 +version: 1.0.67+1 environment: sdk: ^3.12.2 diff --git a/test/fake_note_repository.dart b/test/fake_note_repository.dart index 20b27ce..3ad3214 100644 --- a/test/fake_note_repository.dart +++ b/test/fake_note_repository.dart @@ -29,6 +29,11 @@ class FakeNoteRepository implements NoteRepository { NoteSort? lastSort; NoteFilter? lastFilter; + /// Calls to [upsert] and to the one-write [bulkUpdate], so a bulk action + /// can be shown to be one call rather than one write per note. + int upsertCalls = 0; + int bulkCalls = 0; + /// Emits the current snapshot to a new subscriber first (so late-binding /// [StreamBuilder]s get the seed), then forwards subsequent changes. Stream> _snapshots() async* { @@ -46,12 +51,37 @@ class FakeNoteRepository implements NoteRepository { @override Future upsert(Note note) async { + upsertCalls++; _notes ..removeWhere((n) => n.id == note.id) ..add(note); _emit(); } + @override + Future bulkUpdate( + Set ids, { + Priority? priority, + Status? status, + }) async { + bulkCalls++; + for (final (i, n) in _notes.indexed.toList()) { + if (ids.contains(n.id)) { + _notes[i] = n.copyWith(priority: priority, status: status); + } + } + _emit(); + } + + @override + Future saveReplacing(Note note, Iterable removed) async { + final gone = {...removed, note.id}; + _notes + ..removeWhere((n) => gone.contains(n.id)) + ..add(note); + _emit(); + } + @override Future delete(String id) async { _notes.removeWhere((n) => n.id == id); @@ -83,7 +113,9 @@ class FakeNoteRepository implements NoteRepository { Future> listNotes({ NoteSort sort = NoteSort.modifiedDesc, NoteFilter filter = const NoteFilter(), - }) async => List.unmodifiable(_notes); + }) async => List.unmodifiable( + applyFilter ? _notes.where((n) => matchesFilter(n, filter)) : _notes, + ); @override Stream> watchNotes({ diff --git a/test/note_repository_bulk_test.dart b/test/note_repository_bulk_test.dart new file mode 100644 index 0000000..54a5f1f --- /dev/null +++ b/test/note_repository_bulk_test.dart @@ -0,0 +1,112 @@ +/// Bulk status/priority and save-replacing: one store write each. +library; + +import 'package:crdt_sync/crdt_sync.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:todo/data/note.dart'; +import 'package:todo/data/note_repository.dart'; + +import 'note_repository_helpers.dart'; + +/// Counts log writes, so "one write" is measured, not assumed. +class _CountingPersistence implements LogPersistence { + final _inner = MemoryPersistence(); + int writes = 0; + + @override + Future read() => _inner.read(); + + @override + Future write(String text) { + writes++; + return _inner.write(text); + } +} + +Future<(NoteRepository, _CountingPersistence)> _open() async { + final p = _CountingPersistence(); + final repo = await NoteRepository.openWith(persistence: p, nodeId: 'dev'); + addTearDown(repo.close); + return (repo, p); +} + +Hlc _clock(NoteRepository repo, String id, String field) => + repo.exportLog()[id]!.fields[field]!.$2; + +void main() { + test('sets status and priority on every id in ONE write', () async { + final (repo, p) = await _open(); + for (final id in ['a', 'b', 'c']) { + await repo.upsert( + noteFixture(id, 'note $id').copyWith(createdBy: 'kuhy'), + ); + } + final textClock = _clock(repo, 'a', 'text'); + final createdClock = _clock(repo, 'a', 'created_at'); + p.writes = 0; + + await repo.bulkUpdate( + {'a', 'b', 'c'}, + status: Status.inProgress, + priority: Priority.high, + ); + + expect(p.writes, 1); + final notes = {for (final n in await repo.listNotes()) n.id: n}; + for (final n in notes.values) { + expect(n.status, Status.inProgress); + expect(n.priority, Priority.high); + expect(n.createdBy, 'kuhy'); + } + expect(_clock(repo, 'a', 'text'), textClock); + expect(_clock(repo, 'a', 'created_at'), createdClock); + expect(_clock(repo, 'a', 'status') > textClock, isTrue); + expect(_clock(repo, 'a', 'updated_at') > textClock, isTrue); + }); + + test('touches only fields whose value changes', () async { + final (repo, _) = await _open(); + await repo.upsert(noteFixture('a', 'A', priority: Priority.high)); + final priorityClock = _clock(repo, 'a', 'priority'); + + await repo.bulkUpdate({'a'}, priority: Priority.high, status: Status.done); + + expect(_clock(repo, 'a', 'priority'), priorityClock); + expect((await repo.listNotes()).single.status, Status.done); + }); + + test('no write when nothing changes or ids are unknown/deleted', () async { + final (repo, p) = await _open(); + await repo.upsert(noteFixture('a', 'A')); + await repo.upsert(noteFixture('gone', 'G')); + await repo.delete('gone'); + p.writes = 0; + + await repo.bulkUpdate({'a'}, priority: Priority.medium); + await repo.bulkUpdate({'missing', 'gone'}, status: Status.done); + await repo.bulkUpdate({'a'}); + + expect(p.writes, 0); + expect(repo.exportLog().containsKey('missing'), isFalse); + expect(repo.exportLog()['gone']!.fields.containsKey('status'), isTrue); + }); + + test('saveReplacing saves one note and tombstones the rest', () async { + final (repo, p) = await _open(); + await repo.upsert(noteFixture('a', 'A').copyWith(createdBy: 'kuhy')); + await repo.upsert(noteFixture('b', 'B')); + await repo.upsert(noteFixture('keep', 'K')); + p.writes = 0; + + final saved = noteFixture('a', 'A2', status: Status.inProgress); + await repo.saveReplacing(saved, ['b', 'missing']); + + expect(p.writes, 1); + final notes = {for (final n in await repo.listNotes()) n.id: n}; + expect(notes.keys, unorderedEquals(['a', 'keep'])); + expect(notes['a']!.text, 'A2'); + expect(notes['a']!.createdBy, 'kuhy'); + expect(repo.exportLog()['b']!.deleted, isTrue); + expect(repo.exportLog().containsKey('missing'), isFalse); + }); +} diff --git a/test/repo_assign_test.dart b/test/repo_assign_test.dart new file mode 100644 index 0000000..a00a9e3 --- /dev/null +++ b/test/repo_assign_test.dart @@ -0,0 +1,109 @@ +/// Assigning plain notes to a repo: sections in its one open note. +library; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:todo/data/note.dart'; +import 'package:todo/data/note_repository.dart'; +import 'package:todo/data/repo_assign.dart'; +import 'package:todo/data/repo_note_split.dart'; + +import 'note_repository_helpers.dart'; + +Future _repo(List seed) async { + final repo = await NoteRepository.openInMemory(); + addTearDown(repo.close); + for (final n in seed) { + await repo.upsert(n); + } + return repo; +} + +Note _task(String id, String text, String repo, {Status? status}) => + noteFixture( + id, + text, + status: status ?? Status.todo, + createdAt: DateTime(2026), + ).copyWith(repo: repo, doneCheck: 'true'); + +void main() { + test('appends each plain note as one section and removes it', () async { + final repo = await _repo([ + _task('r', 'app todos\n\n## Old\nbody', 'kuhyx/app'), + _task('closed', 'app todos\n\n## x', 'kuhyx/app', status: Status.done), + noteFixture('p1', 'Fix login\n\n## what\n_why_\nit 500s'), + noteFixture('p2', 'Tidy', priority: Priority.high), + ]); + final plain = (await repo.listNotes()).where((n) => n.repo == null).toList() + ..sort((a, b) => a.id.compareTo(b.id)); + + // Sections land in the order given (the UI passes selection order). + final refused = await assignToRepo(repo, plain, 'kuhyx/app'); + + expect(refused, isEmpty); + final notes = {for (final n in await repo.listNotes()) n.id: n}; + expect(notes.keys, unorderedEquals(['r', 'closed'])); + final text = notes['r']!.text; + expect(openSections(text), ['Old', 'Fix login', 'Tidy']); + expect(text, contains('### what\n_why_\nit 500s')); + expect(notes['r']!.priority, Priority.high); + expect(notes['r']!.doneCheck, 'true'); + }); + + test('creates the repo note when the repo has none open', () async { + final repo = await _repo([noteFixture('p', 'Ship it\nnow')]); + + await assignToRepo(repo, await repo.listNotes(), 'kuhyx/new-app'); + + final made = (await repo.listNotes()).single; + expect(made.id, isNot('p')); + expect(made.repo, 'kuhyx/new-app'); + expect(made.status, Status.todo); + expect(made.createdBy, 'kuhy'); + expect(made.text, 'new-app todos\n\n## Ship it\nnow'); + }); + + test('refuses notes that already have a repo, moves the rest', () async { + final repo = await _repo([ + _task('t', 'other todos\n\n## A', 'kuhyx/other'), + noteFixture('p', 'Plain'), + ]); + + final refused = await assignToRepo( + repo, + await repo.listNotes(), + 'kuhyx/app', + ); + + expect(refused.map((n) => n.id), ['t']); + final notes = {for (final n in await repo.listNotes()) n.id: n}; + expect(notes['t']!.text, 'other todos\n\n## A'); + expect(notes.containsKey('p'), isFalse); + }); + + test('only repo notes: a no-op that returns them all', () async { + final repo = await _repo([_task('t', 'other todos', 'kuhyx/other')]); + final before = repo.exportLog(); + final refused = await assignToRepo( + repo, + await repo.listNotes(), + 'kuhyx/app', + ); + expect(refused, hasLength(1)); + expect(repo.exportLog(), before); + }); + + test('demoteHeadings keeps a body inside its section', () { + expect( + demoteHeadings('# a\n## b\n### c\n```\n## code\n```\ntext #x'), + '### a\n### b\n#### c\n```\n## code\n```\ntext #x', + ); + }); + + test('isRepoName accepts owner/name only', () { + expect(isRepoName('kuhyx/todo-app'), isTrue); + expect(isRepoName('kuhyx/'), isFalse); + expect(isRepoName('todo'), isFalse); + expect(isRepoName('a/b/c'), isFalse); + }); +}