diff --git a/mobile/lib/features/channels/channel_filter/channel_filter_activity.dart b/mobile/lib/features/channels/channel_filter/channel_filter_activity.dart new file mode 100644 index 0000000000..3d625620e2 --- /dev/null +++ b/mobile/lib/features/channels/channel_filter/channel_filter_activity.dart @@ -0,0 +1,86 @@ +import 'dart:async'; +import 'dart:convert'; +import 'package:flutter/foundation.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:nostr/nostr.dart' as nostr; +import '../../../shared/relay/relay.dart'; +import '../channels_provider.dart'; + +/// Verification stays off the UI isolate. Null after successful history means +/// known empty; a failed/invalid response stays unknown and exposes Retry. +int? verifiedFilterActivity(List> records) { + int? latest; + final now = DateTime.now().millisecondsSinceEpoch ~/ 1000; + for (final record in records) { + nostr.Event.fromJson(jsonEncode(record)); + final at = record['created_at'] as int; + if (at <= now + 300 && (latest == null || at > latest)) latest = at; + } + return latest; +} + +class _ActivityReadPool { + int _running = 0; + final _waiting = >[]; + Future run(Future Function() read) async { + if (_running >= 4) { + final turn = Completer(); + _waiting.add(turn); + await turn.future; + } else { + _running++; + } + try { + return await read(); + } finally { + if (_waiting.isNotEmpty) { + _waiting.removeAt(0).complete(); + } else { + _running--; + } + } + } +} + +final _activityReadPoolProvider = Provider((ref) => _ActivityReadPool()); +final channelFilterActivityProvider = FutureProvider.autoDispose + .family((ref, channelId) async { + // Live messages and foreground refresh already update this owner. Re-read + // only the affected channel; never trust its unverified presentation stamp. + ref.watch( + channelsProvider.select( + (value) => value.value + ?.where((c) => c.id == channelId) + .firstOrNull + ?.lastMessageAt, + ), + ); + final session = ref.watch(relaySessionProvider.select((s) => s.status)); + if (session != SessionStatus.connected) { + throw StateError('Activity unavailable while disconnected'); + } + var disposed = false; + ref.onDispose(() => disposed = true); + final relay = ref.read(relaySessionProvider.notifier); + return ref.read(_activityReadPoolProvider).run(() async { + if (disposed) throw StateError('Activity request superseded'); + final events = await relay.fetchHistory( + NostrFilter( + kinds: EventKind.channelMessageEventKinds, + tags: { + '#h': [channelId], + }, + limit: 1, + ), + ); + final matching = events + .where( + (e) => + e.channelId == channelId && + EventKind.channelMessageEventKinds.contains(e.kind), + ) + .map((e) => e.toJson()) + .toList(); + return compute(verifiedFilterActivity, matching); + }); + }); diff --git a/mobile/lib/features/channels/channel_filter/channel_filter_manager.dart b/mobile/lib/features/channels/channel_filter/channel_filter_manager.dart new file mode 100644 index 0000000000..119fa9e9ec --- /dev/null +++ b/mobile/lib/features/channels/channel_filter/channel_filter_manager.dart @@ -0,0 +1,163 @@ +import 'dart:convert'; +import 'dart:math'; +import 'package:flutter/foundation.dart'; +import 'package:nostr/nostr.dart' as nostr; +import '../../../shared/relay/relay.dart'; +import '../channel_sort/channel_sort_manager.dart'; +import 'channel_filter_store.dart'; + +/// Whole-record LWW, with an explicit preflight and confirmed readback. Cached +/// data is display-only. No automatic publication or retry of failed edits. +class ChannelFilterManager extends ChangeNotifier { + final String pubkey; + final ChannelSortCrypto crypto; + final Future> Function() fetch; + final Future Function(String content, int createdAt) publish; + final void Function(ChannelFilterStore store) cache; + final bool Function(String group)? isGroupLive; + ChannelFilterStore store; + bool ready = false; + bool busy = false; + String? error; + bool _disposed = false; + int _createdAt = 0; + + ChannelFilterManager({ + required this.pubkey, + required this.crypto, + required this.fetch, + required this.publish, + required this.cache, + this.isGroupLive, + ChannelFilterStore? cached, + }) : store = cached ?? ChannelFilterStore.empty(); + void _notify() { + if (!_disposed) notifyListeners(); + } + + Future _read() async { + final events = await fetch(); + if (_disposed) throw StateError('Filter session changed'); + final candidates = + events + .where( + (e) => + e.kind == EventKind.readState && + e.pubkey == pubkey && + e.getTagValue('d') == 'channel-filter' && + e.createdAt <= + DateTime.now().millisecondsSinceEpoch ~/ 1000 + 300, + ) + .toList() + ..sort( + (a, b) => b.createdAt.compareTo(a.createdAt) != 0 + ? b.createdAt.compareTo(a.createdAt) + : a.id.compareTo(b.id), + ); + if (candidates.isEmpty) { + if (events.isNotEmpty) { + throw const FormatException('No valid filter head'); + } + _createdAt = 0; + return ChannelFilterStore.empty(); + } + final event = candidates.first; + nostr.Event.fromJson(jsonEncode(event.toJson())); + final decoded = jsonDecode(crypto.decrypt(event.content)); + if (decoded is! Map) { + throw const FormatException('Invalid filter preference'); + } + final result = ChannelFilterStore.parse(decoded); + _createdAt = event.createdAt; + return result; + } + + void dismissError() { + error = null; + _notify(); + } + + Future refresh() async { + if (_disposed || busy) return; + busy = true; + _notify(); + try { + store = await _read(); + ready = true; + error = null; + cache(store); + } catch (_) { + ready = false; + error = 'Could not load filters. Retry to enable changes.'; + } finally { + busy = false; + _notify(); + } + } + + Future save( + String group, + ChannelFilterMode mode, { + int? windowSeconds, + bool removeWindow = false, + }) async { + if (_disposed || !ready || busy) throw StateError('Filters are read-only'); + if (isGroupLive?.call(group) == false) { + throw StateError('Section no longer exists'); + } + busy = true; + error = null; + final previous = store; + try { + store = store.edit( + group, + mode, + windowSeconds: windowSeconds, + removeWindow: removeWindow, + remoteCreatedAt: _createdAt, + ); + _notify(); + final fresh = await _read(); + if (isGroupLive?.call(group) == false) { + throw StateError('Section no longer exists'); + } + final next = fresh.edit( + group, + mode, + windowSeconds: windowSeconds, + removeWindow: removeWindow, + remoteCreatedAt: _createdAt, + ); + final now = DateTime.now().millisecondsSinceEpoch ~/ 1000; + final createdAt = max(now, _createdAt + 1); + if (createdAt > now + 300) throw StateError('Preference clock is ahead'); + await publish(crypto.encrypt(jsonEncode(next.toJson())), createdAt); + if (_disposed) throw StateError('Filter session changed'); + final confirmed = await _read(); + if (confirmed.modeFor(group) != mode || + confirmed.windowFor(group) != next.windowFor(group) || + (removeWindow && + (confirmed.toJson()['activeWindowSeconds'] as Map?)?.containsKey( + group, + ) == + true)) { + throw StateError('Filter save not confirmed'); + } + store = confirmed; + cache(store); + } catch (_) { + store = previous; + error = 'Could not confirm the filter save.'; + rethrow; + } finally { + busy = false; + _notify(); + } + } + + @override + void dispose() { + _disposed = true; + super.dispose(); + } +} diff --git a/mobile/lib/features/channels/channel_filter/channel_filter_provider.dart b/mobile/lib/features/channels/channel_filter/channel_filter_provider.dart new file mode 100644 index 0000000000..e447a4c2c4 --- /dev/null +++ b/mobile/lib/features/channels/channel_filter/channel_filter_provider.dart @@ -0,0 +1,86 @@ +import 'dart:async'; +import 'dart:convert'; +import 'package:flutter/widgets.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:nostr/nostr.dart' as nostr; +import '../../../shared/community/community_provider.dart'; +import '../../../shared/relay/relay.dart'; +import '../../../shared/theme/theme_provider.dart'; +import '../channel_sort/channel_sort_manager.dart'; +import '../channel_sections/channel_sections_provider.dart'; +import '../channel_sort/channel_sort_storage.dart'; +import 'channel_filter_manager.dart'; +import 'channel_filter_store.dart'; + +final channelFilterProvider = Provider.autoDispose(( + ref, +) { + final config = ref.watch(relayConfigProvider); + final connected = ref.watch( + relaySessionProvider.select((s) => s.status == SessionStatus.connected), + ); + final community = ref.watch(activeCommunityProvider).unwrapPrevious().value; + final nsec = config.nsec?.trim(); + if (nsec == null || nsec.isEmpty || community == null) return null; + final prefs = ref.read(savedPrefsProvider); + final pubkey = nostr.Keys(nostr.Nip19.decode(payload: nsec).data).public; + final cacheKey = + 'buzz.channel-filter.v1:$pubkey:${Uri.encodeComponent(normalizeChannelSortRelayUrl(community.relayUrl))}'; + ChannelFilterStore? cached; + try { + final raw = prefs.getString(cacheKey); + if (raw != null) { + cached = ChannelFilterStore.parse( + jsonDecode(raw) as Map, + ); + } + } catch (_) { + /* A malformed display cache never grants write authority. */ + } + final session = ref.read(relaySessionProvider.notifier); + final signer = SignedEventRelay(session: session, nsec: nsec); + final manager = ChannelFilterManager( + isGroupLive: (group) => + !group.startsWith('section:') || + ref + .read(channelSectionsProvider) + .store + .sections + .any((s) => 'section:${s.id}' == group), + pubkey: pubkey, + crypto: ChannelSortCrypto(nsec, pubkey), + cached: cached, + fetch: () => session.fetchHistory( + NostrFilter( + kinds: const [EventKind.readState], + authors: [pubkey], + tags: const { + '#d': ['channel-filter'], + }, + limit: 1, + ), + ), + publish: (content, createdAt) async { + await signer.submit( + kind: EventKind.readState, + content: content, + tags: const [ + ['d', 'channel-filter'], + ['t', 'channel-filter'], + ], + createdAt: createdAt, + ); + }, + cache: (store) { + unawaited(prefs.setString(cacheKey, jsonEncode(store.toJson()))); + }, + ); + if (connected) scheduleMicrotask(manager.refresh); + ref.listen(appLifecycleProvider, (_, next) { + if (connected && next == AppLifecycleState.resumed) { + unawaited(manager.refresh()); + } + }); + ref.onDispose(manager.dispose); + return manager; +}); diff --git a/mobile/lib/features/channels/channel_filter/channel_filter_store.dart b/mobile/lib/features/channels/channel_filter/channel_filter_store.dart new file mode 100644 index 0000000000..3691ce4dfd --- /dev/null +++ b/mobile/lib/features/channels/channel_filter/channel_filter_store.dart @@ -0,0 +1,176 @@ +import 'dart:convert'; +import 'dart:math'; + +const defaultActiveWindowSeconds = 30 * 86400; + +enum ChannelFilterMode { all, unreads, active } + +bool validActiveWindow(Object? value) => + value is int && + value >= 86400 && + value <= 365 * 86400 && + value % 86400 == 0; + +/// Retains the entire encrypted record. Metadata is authoritative, including +/// null reset tombstones; legacy groups are only a compatibility projection. +class ChannelFilterStore { + final Map _record; + ChannelFilterStore._(this._record); + factory ChannelFilterStore.empty() => + ChannelFilterStore._({'version': 1, 'groups': {}}); + factory ChannelFilterStore.parse(Map record) { + if (record['version'] != 1 || + utf8.encode(jsonEncode(record)).length > 128 * 1024) { + throw const FormatException('Unsupported channel filter'); + } + final copy = jsonDecode(jsonEncode(record)) as Map; + final groups = copy['groups']; + if (groups is! Map || + groups.length > 104 || + groups.entries.any((e) => !_key(e.key) || !_mode(e.value))) { + throw const FormatException('Invalid filter groups'); + } + final windows = copy['activeWindowSeconds']; + if (windows != null && + (windows is! Map || + windows.length > 104 || + windows.entries.any( + (e) => !_key(e.key) || !validActiveWindow(e.value), + ))) { + throw const FormatException('Invalid activity windows'); + } + final meta = copy['meta']; + if (meta != null) { + if (meta is! Map || + meta['v'] != 1 || + meta.keys.any((k) => k != 'v' && k != 'g') || + (meta['g'] != null && meta['g'] is! Map)) { + throw const FormatException('Unsupported filter metadata'); + } + meta['g'] ??= {}; + var projectedCount = 0; + for (final entry in (meta['g'] as Map).entries) { + if (entry.value is List && + (entry.value as List).length == 3 && + (entry.value as List)[2] != null) { + projectedCount++; + } + final r = entry.value; + if (!_key(entry.key) || + r is! List || + r.length != 3 || + r[0] is! int || + r[0] < 0 || + r[0] > 9007199254740991 || + r[1] is! String || + !RegExp(r'^[0-9a-f]{16}$').hasMatch(r[1]) || + (r[2] != null && r[2] != 'unreads' && r[2] != 'active')) { + throw const FormatException('Invalid filter register'); + } + } + if (projectedCount > 104) { + throw const FormatException('Filter group budget exceeded'); + } + } + return ChannelFilterStore._(copy); + } + static bool _key(Object? k) => + k is String && k.trim().isNotEmpty && k.length <= 264 && k != '__proto__'; + static bool _mode(Object? v) => v == 'all' || v == 'unreads' || v == 'active'; + ChannelFilterMode modeFor(String group) { + final meta = _record['meta']; + final value = meta == null + ? (_record['groups'] as Map)[group] + : ((meta['g'] as Map)[group] as List?)?[2]; + return switch (value) { + 'unreads' => ChannelFilterMode.unreads, + 'active' => ChannelFilterMode.active, + _ => ChannelFilterMode.all, + }; + } + + int windowFor(String group) => + (_record['activeWindowSeconds'] as Map?)?[group] as int? ?? + defaultActiveWindowSeconds; + Map toJson() => + jsonDecode(jsonEncode(_record)) as Map; + + ChannelFilterStore edit( + String group, + ChannelFilterMode mode, { + int? windowSeconds, + bool removeWindow = false, + int remoteCreatedAt = 0, + }) { + if (!_key(group) || + (windowSeconds != null && + (!validActiveWindow(windowSeconds) || + mode != ChannelFilterMode.active))) { + throw const FormatException('Invalid filter choice'); + } + final data = toJson(); + final meta = + (data['meta'] as Map?)?.cast() ?? + { + 'v': 1, + 'g': { + for (final entry in (data['groups'] as Map).entries) + if (entry.value != 'all') + entry.key as String: [ + remoteCreatedAt * 1000, + '0000000000000000', + entry.value, + ], + }, + }; + final registers = (meta['g'] as Map).cast(); + var stamp = DateTime.now().millisecondsSinceEpoch; + for (final r in registers.values) { + stamp = max(stamp, ((r as List)[0] as int) + 1); + } + final random = Random.secure(); + final device = List.generate( + 8, + (_) => random.nextInt(256).toRadixString(16).padLeft(2, '0'), + ).join(); + registers[group] = [ + stamp, + device, + mode == ChannelFilterMode.all ? null : mode.name, + ]; + data['meta'] = {...meta, 'g': registers}; + data['groups'] = { + for (final e in registers.entries) + if ((e.value as List)[2] != null) e.key: e.value[2], + }; + final windows = + (data['activeWindowSeconds'] as Map?)?.cast() ?? + {}; + if (removeWindow) { + windows.remove(group); + } else if (windowSeconds != null) { + windows[group] = windowSeconds; + } + data['activeWindowSeconds'] = windows; + return ChannelFilterStore.parse(data); + } +} + +bool channelMatchesFilter({ + required ChannelFilterMode mode, + required bool unread, + required int? activitySeconds, + required bool activityKnown, + required int nowSeconds, + required int windowSeconds, + bool selected = false, +}) => + selected || + switch (mode) { + ChannelFilterMode.all => true, + ChannelFilterMode.unreads => unread, + ChannelFilterMode.active => + activitySeconds == null + ? !activityKnown + : activitySeconds + windowSeconds > nowSeconds, + }; diff --git a/mobile/test/features/channels/channel_filter/channel_filter_test.dart b/mobile/test/features/channels/channel_filter/channel_filter_test.dart new file mode 100644 index 0000000000..adc33f94c8 --- /dev/null +++ b/mobile/test/features/channels/channel_filter/channel_filter_test.dart @@ -0,0 +1,298 @@ +import 'dart:async'; +import 'dart:convert'; +import 'package:buzz/features/channels/channel_filter/channel_filter_manager.dart'; +import 'package:buzz/features/channels/channel_filter/channel_filter_store.dart'; +import 'package:buzz/features/channels/channel_filter/channel_filter_activity.dart'; +import 'package:buzz/features/channels/channel_sort/channel_sort_manager.dart'; +import 'package:buzz/shared/relay/relay.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:nostr/nostr.dart' as nostr; + +void main() { + test( + 'metadata wins and edits preserve reset tombstones, unknown fields and windows', + () { + final source = { + 'version': 1, + 'groups': {'channels': 'active'}, + 'future': {'keep': true}, + 'activeWindowSeconds': {'channels': 604800, 'forums': 86400}, + 'meta': { + 'v': 1, + 'g': { + 'channels': [12, '0000000000000000', null], + 'forums': [13, '0000000000000000', 'active'], + }, + }, + }; + final store = ChannelFilterStore.parse(source); + expect(store.modeFor('channels'), ChannelFilterMode.all); + final next = store.edit('dms', ChannelFilterMode.unreads).toJson(); + expect(next['future'], source['future']); + expect(next['activeWindowSeconds'], source['activeWindowSeconds']); + expect( + next['meta']['g']['channels'], + ((source['meta'] as Map)['g'] as Map)['channels'], + ); + expect(next['groups'], {'forums': 'active', 'dms': 'unreads'}); + expect(store.modeFor('dms'), ChannelFilterMode.all); + }, + ); + test( + 'All writes a tombstone, preserves duration; section deletion removes duration', + () { + final active = ChannelFilterStore.empty().edit( + 'section:abc', + ChannelFilterMode.active, + windowSeconds: 86400, + ); + final reset = active.edit('section:abc', ChannelFilterMode.all); + expect(reset.windowFor('section:abc'), 86400); + expect(reset.toJson()['meta']['g']['section:abc'][2], isNull); + expect( + reset + .edit('section:abc', ChannelFilterMode.all, removeWindow: true) + .toJson()['activeWindowSeconds'], + isEmpty, + ); + }, + ); + test( + 'rejects unsupported metadata, invalid durations and oversized records', + () { + for (final invalid in [0, 86399, 86401, 366 * 86400, '86400']) { + expect( + () => ChannelFilterStore.parse({ + 'version': 1, + 'groups': {}, + 'activeWindowSeconds': {'channels': invalid}, + }), + throwsFormatException, + ); + } + expect( + () => ChannelFilterStore.parse({ + 'version': 1, + 'groups': {}, + 'meta': {'v': 2, 'g': {}}, + }), + throwsFormatException, + ); + expect( + () => ChannelFilterStore.parse({ + 'version': 1, + 'groups': {}, + 'future': 'x' * (128 * 1024), + }), + throwsFormatException, + ); + }, + ); + test( + 'filter predicate covers selection, unread, unknown/empty activity and exact expiry', + () { + for (final mode in ChannelFilterMode.values) { + for (final selected in [false, true]) { + for (final unread in [false, true]) { + for (final known in [false, true]) { + for (final at in [null, 0, 1, 2]) { + final result = channelMatchesFilter( + mode: mode, + selected: selected, + unread: unread, + activityKnown: known, + activitySeconds: at, + nowSeconds: 11, + windowSeconds: 10, + ); + final expected = + selected || + mode == ChannelFilterMode.all || + (mode == ChannelFilterMode.unreads && unread) || + (mode == ChannelFilterMode.active && + (at == null ? !known : at > 1)); + expect( + result, + expected, + reason: '$mode $selected $unread $known $at', + ); + } + } + } + } + } + }, + ); + + late nostr.Keys keys; + late ChannelSortCrypto crypto; + late List history; + var failRead = false; + var failPublish = false; + var retainPublish = true; + var live = true; + var publishes = 0; + late List cached; + setUp(() { + keys = nostr.Keys.generate(); + crypto = ChannelSortCrypto(keys.nsec, keys.public); + history = []; + cached = []; + failRead = false; + failPublish = false; + retainPublish = true; + publishes = 0; + live = true; + }); + NostrEvent signed(String content, int at) => NostrEvent.fromJson( + nostr.Event.from( + kind: EventKind.readState, + tags: [ + ['d', 'channel-filter'], + ], + content: content, + secretKey: nostr.Nip19.decode(payload: keys.nsec).data, + createdAt: at, + verify: false, + ).toMap(), + ); + ChannelFilterManager manager({ + ChannelFilterStore? startup, + Future> Function()? fetch, + }) => ChannelFilterManager( + pubkey: keys.public, + crypto: crypto, + cached: startup, + cache: cached.add, + isGroupLive: (_) => live, + fetch: + fetch ?? + () async { + if (failRead) throw StateError('offline'); + return history; + }, + publish: (content, at) async { + publishes++; + if (failPublish) throw StateError('rejected'); + if (retainPublish) history = [signed(content, at)]; + }, + ); + test( + 'startup cache is display-only; failed reads cannot enable writes or publish', + () async { + failRead = true; + final subject = manager( + startup: ChannelFilterStore.empty().edit( + 'channels', + ChannelFilterMode.unreads, + ), + ); + await subject.refresh(); + expect(subject.store.modeFor('channels'), ChannelFilterMode.unreads); + expect(subject.ready, isFalse); + await expectLater( + subject.save('channels', ChannelFilterMode.all), + throwsStateError, + ); + expect(publishes, 0); + subject.dispose(); + }, + ); + test( + 'failed save rolls back; explicit retry confirms mode and duration together', + () async { + final subject = manager(); + await subject.refresh(); + failPublish = true; + await expectLater( + subject.save( + 'channels', + ChannelFilterMode.active, + windowSeconds: 604800, + ), + throwsStateError, + ); + expect(subject.store.modeFor('channels'), ChannelFilterMode.all); + expect(subject.busy, isFalse); + failPublish = false; + await subject.save( + 'channels', + ChannelFilterMode.active, + windowSeconds: 604800, + ); + expect(subject.store.modeFor('channels'), ChannelFilterMode.active); + expect(subject.store.windowFor('channels'), 604800); + expect(cached.last.windowFor('channels'), 604800); + subject.dispose(); + }, + ); + test('OK without matching readback is failure', () async { + final subject = manager(); + await subject.refresh(); + retainPublish = false; + await expectLater( + subject.save('channels', ChannelFilterMode.active, windowSeconds: 86400), + throwsStateError, + ); + expect(subject.store.modeFor('channels'), ChannelFilterMode.all); + expect(subject.error, isNotNull); + subject.dispose(); + }); + test( + 'preflight preserves another device edit and rejects removed-section retry', + () async { + final subject = manager(); + await subject.refresh(); + history = [ + signed( + crypto.encrypt( + jsonEncode( + ChannelFilterStore.empty() + .edit( + 'forums', + ChannelFilterMode.active, + windowSeconds: 86400, + ) + .toJson(), + ), + ), + 10, + ), + ]; + await subject.save('channels', ChannelFilterMode.unreads); + expect(subject.store.modeFor('forums'), ChannelFilterMode.active); + expect(subject.store.windowFor('forums'), 86400); + live = false; + await expectLater( + subject.save('section:gone', ChannelFilterMode.unreads), + throwsStateError, + ); + expect(publishes, 1); + subject.dispose(); + }, + ); + test( + 'disposed read never enables writes or persists a different session', + () async { + final pending = Completer>(); + final subject = manager(fetch: () => pending.future); + final loading = subject.refresh(); + subject.dispose(); + pending.complete([]); + await loading; + expect(cached, isEmpty); + expect(subject.ready, isFalse); + }, + ); + test('activity accepts verified events and rejects a forged timestamp', () { + final event = signed('activity', 100); + expect(verifiedFilterActivity([event.toJson()]), 100); + expect( + () => verifiedFilterActivity([ + {...event.toJson(), 'created_at': 200}, + ]), + throwsA(anything), + ); + expect(verifiedFilterActivity([]), isNull); + }); +}