diff --git a/lib/app.dart b/lib/app.dart index 99c7abf1e..5d9b00633 100644 --- a/lib/app.dart +++ b/lib/app.dart @@ -6,6 +6,7 @@ import 'package:provider/provider.dart'; import 'ai/briefing.dart' show BriefingPeriod; import 'coach/coach_config.dart'; +import 'data/db.dart'; import 'l10n/app_localizations.dart'; import 'notify/notification_service.dart'; import 'notify/tap_router.dart'; @@ -46,11 +47,22 @@ class OpenStrapApp extends StatefulWidget { State createState() => _OpenStrapAppState(); } +/// Whether screen reads may use the reader connection at this lifecycle state. +/// Null (no report yet) counts as visible; the observer corrects it on the +/// first report. +@visibleForTesting +bool screenReadsForeground(AppLifecycleState? state) => + state != AppLifecycleState.hidden && + state != AppLifecycleState.paused && + state != AppLifecycleState.detached; + class _OpenStrapAppState extends State with WidgetsBindingObserver { @override void initState() { super.initState(); WidgetsBinding.instance.addObserver(this); + unawaited(LocalDb.setForeground( + screenReadsForeground(WidgetsBinding.instance.lifecycleState))); // Hand the BYOK provider config to AppState (briefing generation + the // key-aware AI notification schedule live there). WidgetsBinding.instance.addPostFrameCallback((_) { @@ -79,6 +91,7 @@ class _OpenStrapAppState extends State with WidgetsBindingObserver void didChangeAppLifecycleState(AppLifecycleState state) { final app = context.read(); if (state == AppLifecycleState.resumed) { + unawaited(LocalDb.setForeground(true)); // The user may have flipped our notification switch either way in OS // Settings while we were backgrounded. Drop the cached authorization // decision so the next present/schedule re-reads reality — a denial used @@ -114,10 +127,15 @@ class _OpenStrapAppState extends State with WidgetsBindingObserver // a lock screen looking current. unawaited(WidgetService.refresh(app.repo)); if (app.isPaired) app.openSession(); - } else if (state == AppLifecycleState.paused) { - // Backgrounded: hand the band to the iOS restore path so it can wake-and-drain - // in the background (no-op on Android, where the foreground service holds it). - app.pauseForBackground(); + } else if (state == AppLifecycleState.hidden || + state == AppLifecycleState.paused || + state == AppLifecycleState.detached) { + unawaited(LocalDb.setForeground(false)); + if (state == AppLifecycleState.paused) { + // Backgrounded: hand the band to the iOS restore path so it can wake-and-drain + // in the background (no-op on Android, where the foreground service holds it). + app.pauseForBackground(); + } } } diff --git a/lib/data/db.dart b/lib/data/db.dart index ed954c51b..917319008 100644 --- a/lib/data/db.dart +++ b/lib/data/db.dart @@ -236,7 +236,19 @@ class LocalDb { // object)` — a sustained crash burst (seen in the wild on background event // ingest). Reopen whenever the cached handle isn't actually open. if (db != null && db.isOpen) return db; - return _db = await _openOrRebuild(); + return _writerOpening ??= _openWriter(); + } + + static Future? _writerOpening; + + static Future _openWriter() async { + try { + // A reader must never retain a handle to a closed/replaced writer's file. + await _retireReader(drainAll: false); + return _db = await _openOrRebuild(); + } finally { + _writerOpening = null; + } } /// Set when [_openOrRebuild] had to quarantine an unopenable database and @@ -400,7 +412,7 @@ class LocalDb { // lost. Drop them here; the card is the one surface whose whole job is // telling the truth about a data-loss event. salvaged = Map.of( - await _mergeFromDbFile( + await _mergeFromDbFileImpl( quarantine, only: _salvageTables, tolerant: true, @@ -419,13 +431,172 @@ class LocalDb { } } - static Future close() async { - if (_db != null) { - await _db!.close(); - _db = null; + static Future close() => _withReaderInvalidated(() async { + // A cold open may have started before close, even without a screen lease. + await _writerOpening; + final db = _db; + if (db != null) { + await db.close(); + if (identical(_db, db)) _db = null; + } + }); + + static Database? _reader; + static Database? _readerWriter; + static Future? _readerOpening; + static Database? _readerFailedFor; + static bool _foreground = false; + static int _readerEpoch = 0; + static Future? _screenBarrier; + static Future? _readerClosing; + static final _readLeases = >{}; + static final _readerLeases = >{}; + + /// Bare acquisition for diagnostics. Screen queries use [withReadExecutor] + /// so teardown can wait across all awaits in a read. + static Future readExecutor() async { + while (_screenBarrier != null) { + await _screenBarrier; + } + return _acquireReadExecutor(); + } + + static Future _acquireReadExecutor() async { + final writer = await (_writerOpening ?? instance); + // Includes migrations, repair and any rebuild salvage. Existing leases may + // fall back to this writer while a store change waits for them to finish. + if (!_foreground || _screenBarrier != null) return writer; + final current = _reader; + if (current != null && current.isOpen && identical(_readerWriter, writer)) { + return current; + } + if (identical(_readerFailedFor, writer)) return writer; + final epoch = _readerEpoch; + final opening = _readerOpening ??= _openReader(writer, epoch); + final opened = await opening; + if (identical(_readerOpening, opening)) _readerOpening = null; + return _foreground && epoch == _readerEpoch && + identical(_db, writer) && identical(opened, _reader) + ? opened ?? writer + : writer; + } + + static Future _openReader(Database writer, int epoch) async { + Database? candidate; + try { + candidate = await (debugReaderOpener?.call(writer.path) ?? + openDatabase(writer.path, singleInstance: false)); + await candidate.execute('PRAGMA query_only=ON'); + // Read, never change, journal mode. A rollback-journal connection cannot + // provide the concurrent committed reads this connection exists for. + final mode = await candidate.rawQuery('PRAGMA journal_mode'); + if (mode.isEmpty || '${mode.first.values.first}'.toLowerCase() != 'wal') { + throw StateError('reader requires WAL'); + } + if (!_foreground || epoch != _readerEpoch || + !identical(_db, writer) || !writer.isOpen) { + await candidate.close(); + return null; + } + _reader = candidate; + _readerWriter = writer; + return candidate; + } catch (_) { + if (candidate != null) { + try { await candidate.close(); } catch (_) {} + } + if (epoch == _readerEpoch && identical(_db, writer)) { + _readerFailedFor = writer; + } + return null; + } + } + + static Future withReadExecutor( + Future Function(DatabaseExecutor db) read, + ) async { + while (_screenBarrier != null) { + await _screenBarrier; + } + // Register BEFORE acquisition: close also drains a read awaiting its opener + // and keeps the writer alive if that opener must fall back after invalidation. + final lease = Completer(); + _readLeases.add(lease); + try { + var db = await _acquireReadExecutor(); + if (!db.isOpen || (!identical(db, _db) && !identical(db, _reader))) { + db = await instance; + } + if (identical(db, _reader)) _readerLeases.add(lease); + return await read(db); + } finally { + _readerLeases.remove(lease); + _readLeases.remove(lease); + lease.complete(); + } + } + + static Future setForeground(bool foreground) { + _foreground = foreground; + return foreground ? Future.value() : _retireReader(); + } + + static Future _retireReader({bool drainAll = true}) { + _readerEpoch++; + _readerFailedFor = null; + final reader = _reader; + final opening = _readerOpening; + final previousClose = _readerClosing; + _reader = null; + _readerWriter = null; + final readerLeases = _readerLeases.toList(); + final allLeases = drainAll ? _readLeases.toList() : readerLeases; + // Track handle retirement separately from acquisition leases: a cold + // writer open can itself be awaited by a lease that close is draining. + final closing = () async { + await previousClose; + await Future.wait(readerLeases.map((lease) => lease.future)); + await opening; + if (identical(_readerOpening, opening)) _readerOpening = null; + // Read-only handle: Android may already have closed it natively. A failed + // close must not block a store change or surface from setForeground. + if (reader != null) { + try { await reader.close(); } catch (_) {} + } + }(); + _readerClosing = closing; + return () async { + try { + await Future.wait([ + closing, + ...allLeases.map((lease) => lease.future), + ]); + } finally { + if (identical(_readerClosing, closing)) _readerClosing = null; + } + }(); + } + + /// Block new screen leases while a reset/merge/close changes the store. + /// Existing acquisitions finish on the writer if their reader open goes stale. + static Future _withReaderInvalidated(Future Function() action) async { + final previous = _screenBarrier; + final done = Completer(); + _screenBarrier = done.future; + try { + if (previous != null) await previous; + await _retireReader(); + return await action(); + } finally { + if (identical(_screenBarrier, done.future)) _screenBarrier = null; + done.complete(); } } + /// Replaces only the second connection's open, never the writer's open. + @visibleForTesting + static Future Function(String path)? debugReaderOpener; + /// Runs a write against a guaranteed-open handle, reopening ONCE if the /// cached handle was closed under us mid-write. /// @@ -4613,8 +4784,8 @@ class LocalDb { /// The dates whose stored [key] is a wearable's estimate, or a row its /// column serves as unavailable (`metric_method.class`): not a measurement /// a plain trend line may draw. - static Future> metricNotOursDates(String key) async { - final db = await instance; + static Future> metricNotOursDates(String key, {DatabaseExecutor? executor}) async { + final db = executor ?? await instance; return { for (final r in await db.query('metric_method', columns: ['date'], @@ -4626,8 +4797,8 @@ class LocalDb { /// The algo version behind each day's `metric_series` scalars, oldest first. /// Days with no stamp are simply absent — never guessed at. - static Future>> metricSeriesVersions() async { - final db = await instance; + static Future>> metricSeriesVersions({DatabaseExecutor? executor}) async { + final db = executor ?? await instance; return db.query('metric_series_version', orderBy: 'date ASC'); } @@ -8028,8 +8199,9 @@ class LocalDb { int fromTs, int toTs, { int limit = 5000, + DatabaseExecutor? executor, }) async { - final db = await instance; + final db = executor ?? await instance; return db.query( 'events', where: 'ts >= ? AND ts < ?', @@ -8566,9 +8738,10 @@ class LocalDb { required List signals, required int fromSec, required int toSec, + DatabaseExecutor? executor, }) async { if (signals.isEmpty || toSec <= fromSec) return const {}; - final db = await instance; + final db = executor ?? await instance; final marks = List.filled(signals.length, '?').join(','); final rows = await db.rawQuery( 'SELECT DISTINCT device_id, start_ts, end_ts FROM device_coverage ' @@ -8892,8 +9065,8 @@ class LocalDb { ); } - static Future?> dayResult(String dayId) async { - final db = await instance; + static Future?> dayResult(String dayId, {DatabaseExecutor? executor}) async { + final db = executor ?? await instance; final rows = await db.query( 'day_result', where: 'day_id = ? AND algo_version <= ?', @@ -8938,8 +9111,8 @@ class LocalDb { } /// The N most recent days (newest day_id first), each at its LATEST version. - static Future>> recentDayResults(int limit) async { - final db = await instance; + static Future>> recentDayResults(int limit, {DatabaseExecutor? executor}) async { + final db = executor ?? await instance; // For each day_id pick MAX(algo_version), then join back for the full row. final rows = await db.rawQuery( 'SELECT r.* FROM day_result r ' @@ -9003,8 +9176,8 @@ class LocalDb { /// round trip. Rows: `{day_id, window_json}`. [before] keeps only days /// strictly earlier than that day_id. static Future>> sleepWindowRows(int limit, - {String? before}) async { - final db = await instance; + {String? before, DatabaseExecutor? executor}) async { + final db = executor ?? await instance; return db.rawQuery( 'SELECT r.day_id AS day_id, r.window_json AS window_json ' 'FROM day_result r ' @@ -9079,8 +9252,8 @@ class LocalDb { /// derived row) and a skip-marker day both render EMPTY, so neither must /// bound navigation. Skips are excluded via the `skipped` column, which /// `_markDaySkipped` sets alongside the `{skipped:true}` payload. - static Future> availableDayIds() async { - final db = await instance; + static Future> availableDayIds({DatabaseExecutor? executor}) async { + final db = executor ?? await instance; // Latest version per day (matches [dayResult]), then drop skip-markers. final rows = await db.rawQuery( 'SELECT r.day_id FROM day_result r ' @@ -9710,7 +9883,9 @@ class LocalDb { /// run inside `openDatabase` on iOS's CPU watchdog — not somewhere to be on /// the tail of a destructive user action). Views are `type='view'` and are /// not matched; the sqlite/Android internal tables are skipped by name. - static Future wipeAll() async { + static Future wipeAll() => _withReaderInvalidated(_wipeAll); + + static Future _wipeAll() async { final db = await instance; final rows = await db.rawQuery( "SELECT name FROM sqlite_master WHERE type = 'table' " @@ -10056,6 +10231,13 @@ class LocalDb { String path, { List? only, bool tolerant = false, + }) => _withReaderInvalidated(() => + _mergeFromDbFileImpl(path, only: only, tolerant: tolerant)); + + static Future> _mergeFromDbFileImpl( + String path, { + List? only, + bool tolerant = false, }) async { final src = await openDatabase(path, readOnly: true); final db = await instance; @@ -11045,8 +11227,9 @@ class LocalDb { String key, { int? limit, bool measuredOnly = false, + DatabaseExecutor? executor, }) async { - final db = await instance; + final db = executor ?? await instance; return db.query( 'metric_series', where: 'key = ? AND value IS NOT NULL${_validSql(key)}' @@ -11097,8 +11280,8 @@ class LocalDb { return [for (final r in rows.reversed) (r['value'] as num).toDouble()]; } - static Future?> baseline(String key) async { - final db = await instance; + static Future?> baseline(String key, {DatabaseExecutor? executor}) async { + final db = executor ?? await instance; final rows = await db.query( 'baselines', where: 'key = ?', @@ -12432,15 +12615,16 @@ class LocalDb { static Future>> sessionsInRange( int fromTs, - int toTs, - ) async { - final db = await instance; + int toTs, { + DatabaseExecutor? executor, + }) async { + final db = executor ?? await instance; return withoutFlagOffScores(await db.query( 'sessions', where: 'start_ts >= ? AND start_ts <= ?', whereArgs: [fromTs, toTs], orderBy: 'start_ts DESC', - )); + ), executor: db); } /// What a session's sensor scored: served empty, not stored empty. @@ -12466,10 +12650,10 @@ class LocalDb { /// ponytail: one window scan per session; batch it into one join if a long /// session list gets slow. static Future>> withoutFlagOffScores( - List> rows) async { + List> rows, {DatabaseExecutor? executor}) async { final out = >[]; for (final r in rows) { - out.add(await _flagOffOnly(r) + out.add(await _flagOffOnly(r, executor: executor) ? {...r, for (final c in _sensorScoreColumns) c: null} : r); } @@ -12479,7 +12663,8 @@ class LocalDb { /// Whether finished session [r]'s window has worn seconds and every one of /// them came from a device whose flag is now off (see /// [withoutFlagOffScores]). - static Future _flagOffOnly(Map r) async { + static Future _flagOffOnly(Map r, + {DatabaseExecutor? executor}) async { final s = (r['start_ts'] as num?)?.toInt(); final e = (r['end_ts'] as num?)?.toInt(); if (s == null || e == null || r['status'] != 'done') return false; @@ -12488,13 +12673,13 @@ class LocalDb { // wearable's) and once its raw is pruned. // ponytail: a stamped window the band also measured serves empty too, // not band-only; rescore it from the band's rows if that matters. - if ((await sessionSensorsOf(r['id'])) + if ((await sessionSensorsOf(r['id'], executor: executor)) .any((s) => !sessionSensorSources.contains(s))) { return true; } final on = {...sessionSensorSources, ?sessionWearableSource}; final sources = [ - for (final x in await (await instance).rawQuery( + for (final x in await (executor ?? await instance).rawQuery( 'SELECT DISTINCT source FROM decoded_onehz ' 'WHERE rec_ts >= ? AND rec_ts <= ? AND hr > 0', [s, e])) @@ -12530,8 +12715,9 @@ class LocalDb { } /// Every workout sensor stamped on session [id] ([stampSessionSensor]). - static Future> sessionSensorsOf(Object? id) async => { - for (final r in await (await instance).query('session_sensor', + static Future> sessionSensorsOf(Object? id, + {DatabaseExecutor? executor}) async => { + for (final r in await (executor ?? await instance).query('session_sensor', columns: ['source'], where: 'session_id = ?', whereArgs: [id])) r['source'] as String, }; diff --git a/lib/data/local_repository_impl.dart b/lib/data/local_repository_impl.dart index a6b79a5a8..c0fa8a075 100644 --- a/lib/data/local_repository_impl.dart +++ b/lib/data/local_repository_impl.dart @@ -98,7 +98,8 @@ class LocalRepositoryImpl extends LocalRepository { /// Decode a day_result row's payload bundle (latest algo_version), or null. Future?> _bundle(String date) async { - final row = await LocalDb.dayResult(date); + final row = await LocalDb.withReadExecutor( + (db) => LocalDb.dayResult(date, executor: db)); if (row == null) return null; final b = _decode(row['payload_json']); return b == null ? null : withoutWearableEstimates(b); @@ -117,7 +118,8 @@ class LocalRepositoryImpl extends LocalRepository { /// today's night is not the overnight yet (#448), so a partial night can't /// be served under the prior night's label. Future?> _latestBundle({String? skipDay}) async { - final rows = await LocalDb.recentDayResults(14); + final rows = await LocalDb.withReadExecutor( + (db) => LocalDb.recentDayResults(14, executor: db)); Map? newest, withScalars; for (final row in rows) { if (skipDay != null && row['day_id']?.toString() == skipDay) continue; @@ -155,21 +157,22 @@ class LocalRepositoryImpl extends LocalRepository { /// The stored artifact, UNGATED — only `_crossDay` and `getInsights` may /// read this, and only so `getInsights` can report why it withheld it. - Future?> _crossDayArtifact() async { - final r = await LocalDb.baseline('crossday'); + Future?> _crossDayArtifact() => + LocalDb.withReadExecutor((db) async { + final r = await LocalDb.baseline('crossday', executor: db); final cd = _decode(r?['payload_json']); if (cd == null) return null; // A wearable's estimated TRIMP anywhere in the crossday input (a ring's, // say) moves the load and the strain target built on it: the load's EWMA // runs over the input's whole span (90 day rows), not 42 days. Labelled // Estimated where shown, not dropped (whether it counts is the owner's). - final notOurs = await LocalDb.metricNotOursDates('trimp'); + final notOurs = await LocalDb.metricNotOursDates('trimp', executor: db); if (notOurs.isEmpty) return cd; final now = DateTime.now(); var from = dayLabelOf(DateTime(now.year, now.month, now.day - 90)); // 90 rows can span more than 90 calendar days: the input's oldest wins. final input = - _decode((await LocalDb.baseline('crossday_input'))?['payload_json']); + _decode((await LocalDb.baseline('crossday_input', executor: db))?['payload_json']); final inDays = input?['days']; if (inDays is List && inDays.isNotEmpty) { final first = (inDays.first as Map?)?['date']; @@ -181,7 +184,7 @@ class LocalRepositoryImpl extends LocalRepository { for (final k in const ['load', 'strain_coach']) if (cd[k] is Map) k: {...cd[k] as Map, 'estimated': true}, }; - } + }); Future?> _freshness(String key) async { final row = await LocalDb.computeFreshness(key); @@ -996,7 +999,8 @@ class LocalRepositoryImpl extends LocalRepository { } @override - Future> availableDays() => LocalDb.availableDayIds(); + Future> availableDays() => LocalDb.withReadExecutor( + (db) => LocalDb.availableDayIds(executor: db)); @override Future> getDaySleep(String date) => _daySleep(date); @@ -1770,14 +1774,16 @@ class LocalRepositoryImpl extends LocalRepository { } // Workouts + device events for that calendar day. - final sess = await LocalDb.sessionsInRange(dayStart, dayEnd); + final sess = await LocalDb.withReadExecutor( + (db) => LocalDb.sessionsInRange(dayStart, dayEnd, executor: db)); // Bounded BY THE DAY, in SQL. This used to pull `unuploadedEvents(limit: // 2000)` — `ORDER BY ts ASC LIMIT 2000`, i.e. the OLDEST 2000 rows — and // then filter that page down to this day. Once `events` held more than 2000 // rows the page could no longer reach recent days at all, so their markers // silently vanished from the timeline (the same oldest-N-vs-trailing-N // shape as the metricSeries(limit:) outage). - final dayEvents = await LocalDb.eventsInRange(dayStart, dayEnd); + final dayEvents = await LocalDb.withReadExecutor( + (db) => LocalDb.eventsInRange(dayStart, dayEnd, executor: db)); final events = >[ for (final e in dayEvents) { @@ -1879,7 +1885,8 @@ class LocalRepositoryImpl extends LocalRepository { @override Future>> sleepWindows( {int days = 60, String? before}) async { - final rows = await LocalDb.sleepWindowRows(days, before: before); + final rows = await LocalDb.withReadExecutor( + (db) => LocalDb.sleepWindowRows(days, before: before, executor: db)); final out = >[]; for (final r in rows) { final date = r['day_id'] as String?; @@ -1932,7 +1939,8 @@ class LocalRepositoryImpl extends LocalRepository { now.subtract(const Duration(days: 31)).millisecondsSinceEpoch ~/ 1000; final toSec = to ?? nowSec; - final manualRows = await LocalDb.sessionsInRange(fromSec, toSec); + final manualRows = await LocalDb.withReadExecutor( + (db) => LocalDb.sessionsInRange(fromSec, toSec, executor: db)); final manual = [for (final r in manualRows) _workoutOf(r)]; // Manual/live sessions are ALL there is now. There used to be a second @@ -2027,9 +2035,11 @@ class LocalRepositoryImpl extends LocalRepository { final key = _trendKey(metric); // A wearable's estimate is not drawn as a measured point, as on the // native cards ([withoutWearableEstimates]). - final notOurs = await LocalDb.metricNotOursDates(key); + final notOurs = await LocalDb.withReadExecutor( + (db) => LocalDb.metricNotOursDates(key, executor: db)); final rows = [ - for (final r in await LocalDb.metricSeries(key)) + for (final r in await LocalDb.withReadExecutor( + (db) => LocalDb.metricSeries(key, executor: db))) if (!notOurs.contains(r['date'])) r, ]; // THE PIN WINS. getToday serves the frozen morning headline for readiness, @@ -2058,7 +2068,8 @@ class LocalRepositoryImpl extends LocalRepository { // ONE read of metric_series_version, two consumers. `_algoBreaks` used to // fetch it privately; `coverage_devices` rides on the same rows because it // is the same table, the same day key and the same write. - final versions = await LocalDb.metricSeriesVersions(); + final versions = await LocalDb.withReadExecutor( + (db) => LocalDb.metricSeriesVersions(executor: db)); // WHO CONTRIBUTED TO EACH DAY'S SCALARS (final-plan §4.4). Days with no // stamp, and days stamped before schema 50, are simply ABSENT — never @@ -2109,7 +2120,8 @@ class LocalRepositoryImpl extends LocalRepository { // derived key — nothing here reconstructs it, and a day with no // `worn_min` row is simply absent. 'wear': [ - for (final r in await LocalDb.metricSeries('worn_min')) + for (final r in await LocalDb.withReadExecutor( + (db) => LocalDb.metricSeries('worn_min', executor: db))) {'t': _dateToEpoch(r['date'] as String), 'v': r['value']}, ], // L13 — WHERE THE MATHS CHANGED. The dates at which the algo version @@ -2126,11 +2138,12 @@ class LocalRepositoryImpl extends LocalRepository { // single-device install runs no additional statement, ever. 'coverage_recording': signals.isEmpty || distinctDevices.length < 2 ? const >{} - : await LocalDb.coverageDevicesByDay( + : await LocalDb.withReadExecutor((db) => LocalDb.coverageDevicesByDay( signals: signals.toList(), fromSec: oldestDaySec, toSec: _nowSec(), - ), + executor: db, + )), }; } diff --git a/test/wal_reader_connection_test.dart b/test/wal_reader_connection_test.dart new file mode 100644 index 000000000..43886c047 --- /dev/null +++ b/test/wal_reader_connection_test.dart @@ -0,0 +1,615 @@ +// E22 phase 1: RED contracts against upstream's writer-only implementation. +// Real files matter: separate :memory: handles do not share a WAL database. +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:path/path.dart' as p; +import 'package:sqflite_common/sqflite_logger.dart'; +import 'package:sqflite_common_ffi/sqflite_ffi.dart'; +import 'package:openstrap_edge/ble/ble_engine.dart' show DrainController; +import 'package:openstrap_edge/compute/derivation_engine.dart' show kAlgoVersion; +import 'package:openstrap_edge/data/db.dart'; +import 'package:openstrap_edge/data/day_label.dart'; +import 'package:openstrap_edge/data/local_repository_impl.dart'; +import 'package:openstrap_edge/data/models.dart'; + +const _wait = Duration(seconds: 2); +const _day = '2026-01-10'; +typedef _Sql = ({int id, int? txn, String sql, SqliteSqlCommandType type}); + +Future _putDay(String day, {double readiness = 61}) => + LocalDb.putDayResult( + dayId: day, algoVersion: kAlgoVersion, + payloadJson: jsonEncode({'scalars': {'readiness': readiness, 'rhr': 55}}), + windowJson: '{}', readiness: readiness, + series: {'rhr': 55}, + ); + +RawRecord _raw(int counter) => RawRecord( + counter: counter, packetType: 0x2f, hex: '2f18aabbccdd', + capturedAt: 1750000000000 + counter, recTs: 1750000000 + counter); +Sample _sample(int counter) => + Sample(tsEpoch: 1750000000 + counter, counter: counter, hr: 60, + rrIntervalsMs: [900, 910]); + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + final sql = <_Sql>[]; + final opens = []; + final closes = []; + late Directory dir; + late DatabaseFactory? savedFactory; + late String savedPath; + late String savedName; + late LocalRepositoryImpl repo; + var serial = 0; + + setUpAll(() async { + sqfliteFfiInit(); + savedFactory = databaseFactoryOrNull; + savedPath = await databaseFactoryFfi.getDatabasesPath(); + savedName = LocalDb.dbName; + dir = await Directory.systemTemp.createTemp('e22_wal_reader_'); + await databaseFactoryFfi.setDatabasesPath(dir.path); + // ignore: experimental_member_use + databaseFactory = SqfliteDatabaseFactoryLogger(databaseFactoryFfi, + options: SqfliteLoggerOptions(log: (event) { + if (event is SqfliteLoggerSqlEvent) { + sql.add((id: event.databaseId, txn: event.transactionId, + sql: event.sql.toLowerCase(), type: event.type)); + } else if (event is SqfliteLoggerBatchEvent) { + for (final op in event.operations) { + sql.add((id: event.databaseId, txn: event.transactionId, + sql: op.sql.toLowerCase(), type: op.type)); + } + } else if (event is SqfliteLoggerDatabaseOpenEvent) { + opens.add(event); + } else if (event is SqfliteLoggerDatabaseCloseEvent) { + closes.add(event.databaseId); + } + }), + ); + }); + setUp(() async { + await LocalDb.close(); + LocalDb.debugReaderOpener = null; + LocalDb.lastRebuild = null; + await LocalDb.setForeground(false); + LocalDb.dbName = 'reader_${serial++}.db'; + sql.clear(); + opens.clear(); + closes.clear(); + repo = LocalRepositoryImpl(getProfileMap: () => const {}); + await LocalDb.setForeground(true); + }); + tearDown(() async { + LocalDb.debugReaderOpener = null; + LocalDb.debugBeforeOrphanPinClear = null; + await LocalDb.setForeground(false); + await LocalDb.close(); + // Includes a handle opened by a delayed test opener if publication failed. + for (final event in opens) { + if (event.db?.isOpen ?? false) await event.db!.close(); + } + }); + tearDownAll(() async { + await LocalDb.close(); + LocalDb.dbName = savedName; + databaseFactoryOrNull = savedFactory; + await databaseFactoryFfi.setDatabasesPath(savedPath); + await dir.delete(recursive: true); + }); + + int idOf(Database db) => opens.lastWhere((e) => identical(e.db, db)).databaseId!; + + Future writerId() async => idOf(await LocalDb.instance); + + Future readerQueries(String table, int writer) async { + final hits = sql.where((e) => e.type == SqliteSqlCommandType.query && + RegExp('\\bfrom\\s+$table\\b').hasMatch(e.sql)).toList(); + expect(hits, isNotEmpty, reason: 'fixture must exercise $table'); + expect(hits.map((e) => e.id), everyElement(isNot(writer)), + reason: '$table screen SELECT must use the second connection'); + expect(hits.map((e) => e.id), everyElement(idOf(await LocalDb.readExecutor())), + reason: "$table must use LocalDb's owned reader handle"); + } + + // Release the writer even when a screen read times out on upstream, then + // drain that read before teardown. A failing RED test must not wedge SQLite. + Future holdingWrite(Future Function() body, + {bool rollback = false}) async { + final entered = Completer(); + final release = Completer(); + final writer = await LocalDb.instance; + final holding = writer.transaction((tx) async { + await tx.insert('day_result', { + 'day_id': '2026-01-11', 'algo_version': kAlgoVersion, + 'payload_json': '{"scalars":{"readiness":88}}', + 'window_json': '{}', 'computed_at': 1, 'readiness': 88, + }); + entered.complete(); + await release.future; + if (rollback) throw StateError('intentional rollback'); + }); + final settled = holding.then((_) => null, + onError: (Object e, StackTrace s) => e); + try { + await entered.future.timeout(_wait); + await body(); + } finally { + release.complete(); + final outcome = await settled.timeout(_wait); + if (!rollback && outcome != null) throw outcome; + if (rollback) expect(outcome, isA()); + } + } + + test('reader is a lazy second handle to the same WAL file', () async { + final writer = await LocalDb.instance; + expect(opens, hasLength(1)); + await LocalDb.setForeground(true); + expect(opens, hasLength(1), reason: 'foreground only permits lazy opening'); + final reader = await LocalDb.readExecutor(); + expect(identical(reader, writer), isFalse, + reason: 'upstream serializes all reads on the writer'); + expect(opens, hasLength(2)); + expect(reader.path, writer.path); + final files = await reader.rawQuery('PRAGMA database_list'); + expect(files.firstWhere((r) => r['name'] == 'main')['file'], writer.path); + final option = opens.last.options!; + expect(option.singleInstance, isFalse); + expect(option.readOnly, isFalse, reason: 'WAL may need to create -shm'); + expect(option.version, isNull); + expect(option.onCreate, isNull); + expect(option.onUpgrade, isNull); + expect(option.onOpen, isNull, reason: 'repair belongs only to writer'); + final readerId = idOf(reader); + final wal = sql.where((e) => e.id == readerId && + e.sql.startsWith('pragma journal_mode')).toList(); + expect(wal, isNotEmpty, reason: 'verify WAL before publishing reader'); + expect(wal.map((e) => e.type), everyElement(SqliteSqlCommandType.query)); + expect((await reader.rawQuery('PRAGMA journal_mode')).first.values.first, 'wal'); + }); + + test('cold reader waits for writer open and same-version schema repair', () async { + final writer = await LocalDb.instance; + final path = writer.path; + await writer.execute('DROP TABLE metric_method'); + await LocalDb.close(); + opens.clear(); + final reader = await LocalDb.readExecutor(); + expect((await reader.rawQuery('PRAGMA user_version')).first.values.first, + LocalDb.schemaVersion); + expect(await reader.rawQuery("SELECT name FROM sqlite_master " + "WHERE type='table' AND name='metric_method'"), hasLength(1)); + expect(opens, hasLength(2), reason: 'writer repair must precede second open'); + expect(opens.first.path, path); + expect(opens.first.options!.onOpen, isNotNull); + expect(opens.last.options!.onOpen, isNull); + }); + + test('query_only rejects a write through the reader', () async { + final reader = await LocalDb.readExecutor(); + await expectLater(reader.insert('sync_cursor', + {'name': 'reader_must_not_write', 'value': 'x', 'updated_at': 1}), + throwsA(isA())); + expect((await reader.rawQuery('PRAGMA query_only')).first.values.first, 1); + expect(await LocalDb.getCursor('reader_must_not_write'), isNull); + }); + + test('screen read completes during held write and sees only committed rows', () async { + await _putDay(_day); + // Open the reader before taking the write lock; opening is tested above. + await LocalDb.readExecutor(); + Future>? pending; + try { + await holdingWrite(() async { + pending = repo.getDayOverview(_day); + expect((await pending!.timeout(_wait))['readiness'], 61); + pending = repo.getDayOverview('2026-01-11'); + expect(await pending!.timeout(_wait), isEmpty, + reason: 'in-flight insert is invisible to the reader'); + }); + } finally { + await pending; + } + }); + + test('a new routed read sees a completed writer commit', () async { + await _putDay(_day); + final writer = await writerId(); + await repo.getDayOverview(_day); // establish the first snapshot + await _putDay('2026-01-11', readiness: 88); + sql.clear(); + expect((await repo.getDayOverview('2026-01-11'))['readiness'], 88); + await readerQueries('day_result', writer); + }); + + test('rolled-back rows never become visible on the reader', () async { + await _putDay(_day); + await LocalDb.readExecutor(); + Future>? pending; + try { + await holdingWrite(() async { + pending = repo.getDayOverview('2026-01-11'); + expect(await pending!.timeout(_wait), isEmpty); + }, rollback: true); + } finally { + await pending; + } + expect(await repo.getDayOverview('2026-01-11'), isEmpty); + }); + + test('failed reader open falls back to writer and returns stored values', () async { + await _putDay(_day); + var attempted = 0; + LocalDb.debugReaderOpener = (path) async { + attempted++; + throw StateError('simulated reader open failure'); + }; + final writer = await writerId(); + sql.clear(); + expect((await repo.getDayOverview(_day))['readiness'], 61); + expect(attempted, greaterThan(0), reason: 'must exercise the failed open'); + expect(sql.where((e) => e.sql.contains('from day_result')).map((e) => e.id), + everyElement(writer)); + expect(opens.where((e) => e.db?.isOpen ?? false), hasLength(1)); + }); + + test('failed reader open retries after background, not on every read', () async { + await _putDay(_day); + var attempted = 0; + LocalDb.debugReaderOpener = (path) async { + attempted++; + throw StateError('simulated reader open failure'); + }; + await repo.getDayOverview(_day); + await repo.getDayOverview(_day); + expect(attempted, 1, reason: 'a broken open is not retried per screen read'); + await LocalDb.setForeground(false); + await LocalDb.setForeground(true); + LocalDb.debugReaderOpener = null; + final writer = await LocalDb.instance; + final reader = await LocalDb.readExecutor(); + expect(identical(reader, writer), isFalse, reason: 'next foreground retries'); + }); + + test('a throwing reader close neither fails backgrounding nor blocks close', () async { + LocalDb.debugReaderOpener = (path) async => _CloseThrows( + await databaseFactory.openDatabase(path, + options: OpenDatabaseOptions(singleInstance: false))); + final writer = await LocalDb.instance; + final reader = await LocalDb.readExecutor(); + expect(reader, isA<_CloseThrows>()); + await LocalDb.setForeground(false); + await LocalDb.setForeground(true); + await LocalDb.close(); + expect(writer.isOpen, isFalse, reason: 'close must still reach the writer'); + }); + + test('non-WAL reader is rejected and closed before fallback', () async { + await _putDay(_day); + final writer = await LocalDb.instance; + await writer.rawQuery('PRAGMA journal_mode=DELETE'); + Database? candidate; + LocalDb.debugReaderOpener = (path) async => candidate = + await databaseFactory.openDatabase(path, + options: OpenDatabaseOptions(singleInstance: false)); + expect((await repo.getDayOverview(_day))['readiness'], 61); + expect(candidate, isNotNull, reason: 'must attempt the second open'); + expect(candidate!.isOpen, isFalse, reason: 'failed candidate must not leak'); + expect(await LocalDb.readExecutor(), same(writer)); + }); + + test('background reads use writer; resume permits a lazy fresh reader', () async { + final writer = await LocalDb.instance; + await LocalDb.setForeground(false); + expect(await LocalDb.readExecutor(), same(writer)); + expect(opens, hasLength(1)); + await LocalDb.setForeground(true); + expect(opens, hasLength(1)); + final reader = await LocalDb.readExecutor(); + expect(identical(reader, writer), isFalse); + await LocalDb.setForeground(false); + expect(reader.isOpen, isFalse); + expect(writer.isOpen, isTrue); + expect(await LocalDb.readExecutor(), same(writer)); + await LocalDb.setForeground(true); + expect(opens, hasLength(2), reason: 'resume does not eagerly reopen'); + final reopened = await LocalDb.readExecutor(); + expect(identical(reopened, reader), isFalse); + expect(identical(reopened, writer), isFalse); + }); + + test('LocalDb.close closes reader before writer', () async { + final writer = await writerId(); + final reader = idOf(await LocalDb.readExecutor()); + closes.clear(); + await LocalDb.close(); + expect(closes, [reader, writer], reason: 'reader must not outlive its file owner'); + }); + + for (final background in [true, false]) { + test('${background ? 'background' : 'close'} drains an in-flight read lease', () async { + await _putDay(_day); + final entered = Completer(); + final release = Completer(); + final reading = LocalDb.withReadExecutor((db) async { + entered.complete(); + await release.future; + return db.query('day_result', where: 'day_id = ?', whereArgs: [_day]); + }).then((rows) => rows, onError: (Object e, StackTrace s) => e); + await entered.future.timeout(_wait); + var finished = false; + final closing = (background ? LocalDb.setForeground(false) : LocalDb.close()) + .then((_) => finished = true); + Object? result; + try { + // Only a failure bound. The lease/release completers define ordering. + await Future.delayed(const Duration(milliseconds: 100)); + expect(finished, isFalse, reason: 'close must wait for the active lease'); + } finally { + release.complete(); + result = await reading.timeout(_wait); + await closing.timeout(_wait); + } + expect(result, isA>>(), + reason: 'the leased query must never run on a closed handle'); + expect(opens.where((e) => e.db?.isOpen ?? false), + hasLength(background ? 1 : 0)); + }); + + test('${background ? 'background' : 'close'} invalidates a pending reader open', () async { + await _putDay(_day); + final entered = Completer(); + final release = Completer(); + Database? candidate; + LocalDb.debugReaderOpener = (path) async { + final opened = await databaseFactory.openDatabase(path, + options: OpenDatabaseOptions(singleInstance: false)); + candidate = opened; + entered.complete(); + await release.future; + return opened; + }; + final reading = LocalDb.withReadExecutor((db) => db.query('day_result')) + .then((rows) => rows, onError: (Object e, StackTrace s) => e); + Future? closing; + Object? result; + try { + await entered.future.timeout(_wait, + onTimeout: () => fail('reader opener was never called')); + closing = background ? LocalDb.setForeground(false) : LocalDb.close(); + await Future.delayed(Duration.zero); + } finally { + release.complete(); + result = await reading.timeout(_wait); + await closing?.timeout(_wait); + } + expect(result, isA>>(), + reason: 'pending screen read must finish safely before shutdown'); + expect(candidate!.isOpen, isFalse, reason: 'stale opening must not leak'); + expect(opens.where((e) => e.db?.isOpen ?? false), + hasLength(background ? 1 : 0)); + }); + } + + test('wipe invalidates the reader and the next read sees the empty store', () async { + await _putDay(_day); + final reader = await LocalDb.readExecutor(); + await LocalDb.wipeAll(); + expect(reader.isOpen, isFalse, reason: 'store reset invalidates reader'); + expect(await repo.getDayOverview(_day), isEmpty); + expect(identical(await LocalDb.readExecutor(), reader), isFalse); + }); + + test('restore merge invalidates the reader and exposes restored rows', () async { + await _putDay(_day); + final writer = await LocalDb.instance; + // This backup is a real current-schema file, not a hand-built partial schema. + await writer.rawQuery('PRAGMA wal_checkpoint(TRUNCATE)'); + final backup = p.join(dir.path, 'restore_$serial.db'); + await File(writer.path).copy(backup); + await LocalDb.wipeAll(); + final reader = await LocalDb.readExecutor(); + await LocalDb.importFromDbFile(backup); + expect(reader.isOpen, isFalse, reason: 'merge invalidates old screen handle'); + expect((await repo.getDayOverview(_day))['readiness'], 61); + }); + + test('writer reopen retires an old reader before opening the file again', + () async { + // _openWriter retires the reader before _openOrRebuild touches the file, + // which is what keeps a rebuild from quarantining a file a reader still + // holds. A rebuild cannot be forced here: closing the live reader + // checkpoints its WAL frames back into the main file first. + await _putDay(_day); + final writer = await LocalDb.instance; + final reader = await LocalDb.readExecutor(); + expect(identical(reader, writer), isFalse, reason: 'needs a live reader'); + await writer.close(); + final reopened = await LocalDb.instance; + expect(identical(reopened, writer), isFalse); + expect(reader.isOpen, isFalse, reason: 'a reopened writer cannot keep the old reader'); + expect((await repo.getDayOverview(_day))['readiness'], 61); + expect(identical(await LocalDb.readExecutor(), reader), isFalse); + }); + + // Explicit screen allow-list. Default LocalDb callers below remain writers. + final routes = <({String name, Future Function() read, List tables})>[ + (name: '_bundle/dayResult', read: () => repo.getDayOverview(_day), + tables: ['day_result']), + (name: '_latestBundle/recentDayResults', read: () => repo.getDayOverview(todayLabel()), + tables: ['day_result']), + (name: '_crossDayArtifact/baseline', read: () => repo.getInsights(), + tables: ['baselines', 'metric_method']), + (name: 'availableDays/availableDayIds', read: () => repo.availableDays(), + tables: ['day_result']), + (name: 'sleepWindows/sleepWindowRows', read: () => repo.sleepWindows(), + tables: ['day_result']), + (name: 'getChart/series+provenance', read: () => repo.getChart('rhr'), + tables: ['metric_series', 'metric_series_version', 'metric_method']), + (name: 'getDayTimeline/events+sessions', read: () => repo.getDayTimeline(_day), + tables: ['day_result', 'events', 'sessions', 'session_sensor', 'decoded_onehz']), + (name: 'getSessions/sessionsInRange', read: () => repo.getSessions(from: 0, to: 1 << 40), + tables: ['sessions', 'session_sensor', 'decoded_onehz']), + ]; + for (final route in routes) { + test('routes ${route.name} through reader', () async { + await _putDay(_day); + await LocalDb.putBaseline('crossday', '{}'); + final writer = await LocalDb.instance; + await writer.insert('sessions', {'id': 's', 'start_ts': localDayStartSec(_day)!, + 'end_ts': localDayStartSec(_day)! + 60, 'status': 'done', + 'type': 'run', 'created_at': 1}); + final id = idOf(writer); + sql.clear(); + await route.read(); + for (final table in route.tables) { + await readerQueries(table, id); + } + }); + } + + test('routes getChart multi-device coverage through reader', () async { + await _putDay(_day); + final writer = await LocalDb.instance; + await writer.insert('metric_series_version', + {'date': _day, 'algo_version': kAlgoVersion, 'coverage_devices': 'a,b'}, + conflictAlgorithm: ConflictAlgorithm.replace); + final noon = localDayStartSec(_day)! + 12 * 3600; + await writer.insert('device_coverage', {'device_id': 'b', 'signal': 'hr', + 'start_ts': noon, 'end_ts': noon + 3600}); + final id = idOf(writer); + sql.clear(); + final chart = await repo.getChart('rhr', signals: {'hr'}); + expect(chart['coverage_recording'], isNotEmpty, reason: 'fixture must query coverage'); + await readerQueries('device_coverage', id); + }); + + test('shared LocalDb helpers still use writer by default for compute', () async { + await _putDay(_day); + await LocalDb.readExecutor(); + final writer = await writerId(); + sql.clear(); + await LocalDb.dayResult(_day); + await LocalDb.recentDayResults(2); + await LocalDb.availableDayIds(); + await LocalDb.sleepWindowRows(2); + await LocalDb.baseline('crossday'); + await LocalDb.metricSeries('rhr'); + await LocalDb.metricSeriesVersions(); + await LocalDb.metricNotOursDates('rhr'); + await LocalDb.eventsInRange(0, 1 << 40); + await LocalDb.sessionsInRange(0, 1 << 40); + expect(sql, isNotEmpty); + expect(sql.map((e) => e.id), everyElement(writer)); + }); + + test('getToday freshness refresh and reread stay on writer', () async { + await LocalDb.readExecutor(); + final writer = await writerId(); + sql.clear(); + final before = todayLabel(); + await repo.getToday(); // missing freshness triggers the real refresh + final after = todayLabel(); + final freshness = sql.where((e) => e.sql.contains('compute_freshness')).toList(); + expect(freshness.any((e) => e.type == SqliteSqlCommandType.insert), isTrue); + expect(freshness.where((e) => e.type == SqliteSqlCommandType.query).length, + greaterThanOrEqualTo(2)); + expect(freshness.map((e) => e.id), everyElement(writer)); + final refreshed = await LocalDb.computeFreshness('today'); + expect(jsonDecode(refreshed!['payload_json'] as String)['today_day'], anyOf(before, after)); + }); + + test('frozenHeadline orphan lookup and compare-delete stay on writer', () async { + await LocalDb.setCursor(LocalDb.kFrozenHeadlineCursor, + '{"day":"2026-01-01","value":71}'); + await LocalDb.readExecutor(); + final writer = await writerId(); + sql.clear(); + expect(await LocalDb.frozenHeadline(), isNull); + expect(sql.any((e) => e.type == SqliteSqlCommandType.delete && + e.sql.contains('sync_cursor')), isTrue); + expect(sql.map((e) => e.id), everyElement(writer)); + expect(await LocalDb.getCursor(LocalDb.kFrozenHeadlineCursor), isNull); + }); + + for (final failing in [false, true]) { + test('safe-trim ${failing ? 'rollback' : 'commit'} keeps FULL bracket on writer', () async { + await LocalDb.readExecutor(); + final writer = await writerId(); + final handle = await LocalDb.instance; + if (failing) { + await handle.execute('CREATE TRIGGER refuse_cursor BEFORE INSERT ON sync_cursor ' + "WHEN new.name = 'strap_trim' BEGIN SELECT RAISE(ABORT, 'refuse trim'); END"); + } + final d = DrainController(onRecord: (sample, raw) async {}, + onRecordsBatch: null, onArchive: null, log: (_) {}, + onCommit: (raws, samples, token, {archives, ecgRawPackets, deviceFamily}) => + LocalDb.commitSyncBatch(raws, samples, trimToken: token, + archives: archives, ecgRawPackets: ecgRawPackets, deviceFamily: deviceFamily)); + d.onHistoricalRecord(_raw(5001), _sample(5001), 24); + sql.clear(); + final durable = await d.commit(const [1, 2, 3, 4, 5, 6, 7, 8]); + expect(durable, !failing, reason: 'false must withhold the engine ACK'); + final bracket = sql.where((e) => e.sql.startsWith('pragma synchronous=')).toList(); + expect(bracket.map((e) => e.sql), ['pragma synchronous=full', 'pragma synchronous=normal']); + expect(bracket.map((e) => e.id), everyElement(writer)); + final boundary = sql.where((e) => e.sql.startsWith('begin') || + e.sql == 'commit' || e.sql == 'rollback').toList(); + expect(boundary, isNotEmpty); + expect(boundary.map((e) => e.id), everyElement(writer)); + expect(sql.indexOf(bracket.first), lessThan(sql.indexOf(boundary.first))); + expect(sql.indexOf(bracket.last), greaterThan(sql.indexOf(boundary.last))); + expect((await handle.rawQuery('PRAGMA synchronous')).first.values.first, 1); + expect(await LocalDb.getCursor('strap_trim'), failing ? null : '0102030405060708'); + final rows = await handle.query('decoded_onehz', where: 'counter = 5001'); + expect(rows, hasLength(failing ? 0 : 1)); + final beats = await handle.query('decoded_rr', + where: 'device_id = ? AND ts_ms = ?', whereArgs: ['', 1750005001000]); + expect(beats, hasLength(failing ? 0 : 2)); + expect(await LocalDb.getCursorInt('counter_hw'), failing ? null : 5001); + expect(d.bufferedRecords, failing ? 1 : 0); + if (!failing) { + final writes = sql.where((e) => e.type == SqliteSqlCommandType.insert && + (e.sql.contains('decoded_onehz') || e.sql.contains('decoded_rr') || + e.sql.contains('sync_cursor'))).toList(); + expect(writes.any((e) => e.sql.contains('decoded_onehz')), isTrue); + expect(writes.any((e) => e.sql.contains('decoded_rr')), isTrue); + expect(writes.any((e) => e.sql.contains('sync_cursor')), isTrue); + expect(writes.map((e) => e.id), everyElement(writer)); + expect(writes.first.txn, isNotNull); + expect(writes.map((e) => e.txn).toSet(), {writes.first.txn}, + reason: 'decoded seconds, beats and ACK cursor share the durable transaction'); + } + }); + } +} + +// A reader whose native handle is already gone: close() throws, as on Android. +class _CloseThrows implements Database { + _CloseThrows(this._db); + final Database _db; + @override + String get path => _db.path; + @override + bool get isOpen => _db.isOpen; + @override + Future execute(String sql, [List? arguments]) => + _db.execute(sql, arguments); + @override + Future>> rawQuery(String sql, + [List? arguments]) => + _db.rawQuery(sql, arguments); + @override + Future close() async { + await _db.close(); + throw StateError('simulated native close failure'); + } + @override + dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); +} + diff --git a/test/wal_reader_wiring_test.dart b/test/wal_reader_wiring_test.dart new file mode 100644 index 000000000..5fa067314 --- /dev/null +++ b/test/wal_reader_wiring_test.dart @@ -0,0 +1,115 @@ +// Structural guards for the app observer and the private radio ACK boundary. +// Database lifecycle and durability behavior run against real files in +// wal_reader_connection_test.dart. These guards cover their production callers +// without booting AppState's platform services or introducing another observer. +import 'dart:io'; + +import 'package:flutter/widgets.dart' show AppLifecycleState; +import 'package:flutter_test/flutter_test.dart'; +import 'package:openstrap_edge/app.dart' show screenReadsForeground; + +import 'support/dart_source.dart'; + +String _body(String code, String method) { + final call = RegExp('(?:void|Future)\\s+$method\\s*\\(') + .firstMatch(code)!; + var cursor = code.indexOf('(', call.start); + var depth = 0; + do { + if (code[cursor] == '(') depth++; + if (code[cursor] == ')') depth--; + cursor++; + } while (depth > 0); + final start = code.indexOf('{', cursor); + cursor = start; + do { + if (code[cursor] == '{') depth++; + if (code[cursor] == '}') depth--; + cursor++; + } while (depth > 0); + return code.substring(start + 1, cursor - 1); +} + +// Find the if/else-if branch that handles a given lifecycle state. Grouped +// OR conditions are allowed, as in the fork. Comments/strings are already gone. +String? _stateBranch(String code, String state) { + for (final condition in RegExp(r'\bif\s*\(', multiLine: true).allMatches(code)) { + var cursor = code.indexOf('(', condition.start); + final start = cursor; + var depth = 0; + do { + if (code[cursor] == '(') depth++; + if (code[cursor] == ')') depth--; + cursor++; + } while (depth > 0); + if (!RegExp('state\\s*==\\s*AppLifecycleState\\.$state\\b') + .hasMatch(code.substring(start, cursor))) { + continue; + } + final block = code.indexOf('{', cursor); + cursor = block; + do { + if (code[cursor] == '{') depth++; + if (code[cursor] == '}') depth--; + cursor++; + } while (depth > 0); + return code.substring(block + 1, cursor - 1); + } + return null; +} + +void main() { + final app = stripCommentsAndStrings(File('lib/app.dart').readAsStringSync()); + final lifecycle = _body(app, 'didChangeAppLifecycleState'); + + test('visible cold start enables lazy reads through the existing app observer', () { + final init = _body(app, 'initState'); + expect(init, contains('WidgetsBinding.instance.addObserver(this)')); + expect(init, matches(RegExp(r'LocalDb\.setForeground\(\s*screenReadsForeground\(\s*' + r'WidgetsBinding\.instance\.lifecycleState\s*\)\s*\)')), + reason: 'cold foreground start need not emit a resumed transition'); + }); + + test('cold-start lifecycle state maps to the reader foreground flag', () { + // Null: no lifecycle report yet; the observer corrects it on the first one. + expect(screenReadsForeground(null), isTrue); + expect(screenReadsForeground(AppLifecycleState.resumed), isTrue); + expect(screenReadsForeground(AppLifecycleState.inactive), isTrue); + expect(screenReadsForeground(AppLifecycleState.hidden), isFalse); + expect(screenReadsForeground(AppLifecycleState.paused), isFalse); + expect(screenReadsForeground(AppLifecycleState.detached), isFalse); + }); + + test('resume enables lazy reads in the app lifecycle observer', () { + final branch = _stateBranch(lifecycle, 'resumed'); + expect(branch, isNotNull); + expect(branch, matches(RegExp(r'LocalDb\.setForeground\(\s*true\s*\)')), + reason: 'resume must allow the next screen read to reopen the reader'); + }); + + for (final state in ['hidden', 'paused', 'detached']) { + test('$state closes reader through the app lifecycle observer', () { + final branch = _stateBranch(lifecycle, state); + expect(branch, isNotNull, reason: '$state must reach LocalDb'); + expect(branch, matches(RegExp(r'LocalDb\.setForeground\(\s*false\s*\)')), + reason: '$state must drain reads and close the second connection'); + }); + } + + test('HISTORY_END awaits durable commit and gates the verbatim ACK token', () { + final ble = stripCommentsAndStrings(File('lib/ble/ble_engine.dart').readAsStringSync()); + final marker = _body(ble, '_handleSyncMarker'); + final commit = RegExp(r'final\s+durable\s*=\s*await\s+d\.commit\(m\.token\)') + .firstMatch(marker); + final ack = RegExp(r'buildHistoryResultOk\([^;]*m\.token!', dotAll: true) + .firstMatch(marker); + expect(commit, isNotNull); + expect(ack, isNotNull); + expect(ack!.start, greaterThan(commit!.end)); + final gate = marker.substring(commit.end, ack.start); + expect(gate, contains('commitDurable: durable')); + expect(gate, matches(RegExp( + r'if\s*\(verdict\s*!=\s*TrimAckVerdict\.send\)\s*\{[^}]*return\s*;', + dotAll: true)), reason: 'failed/partial commit must return before ACK'); + }); +}