feat(mobile): sync confirmed channel filter preferences

Co-authored-by: Carl <acda9e433d19dcd0e6b6840f7f4b98f3a56f1fab98049d444c087019e6d36560@buzz.block.builderlab.xyz>
Signed-off-by: Taylor Ho <taylorkmho@gmail.com>
This commit is contained in:
Taylor Ho
2026-10-19 09:11:43 -07:00
parent bf8b356f3b
commit 490f2e56c6
5 changed files with 809 additions and 0 deletions

View File

@@ -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<Map<String, dynamic>> 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 = <Completer<void>>[];
Future<T> run<T>(Future<T> Function() read) async {
if (_running >= 4) {
final turn = Completer<void>();
_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<int?, String>((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);
});
});

View File

@@ -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<List<NostrEvent>> Function() fetch;
final Future<void> 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<ChannelFilterStore> _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<String, dynamic>) {
throw const FormatException('Invalid filter preference');
}
final result = ChannelFilterStore.parse(decoded);
_createdAt = event.createdAt;
return result;
}
void dismissError() {
error = null;
_notify();
}
Future<void> 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<void> 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();
}
}

View File

@@ -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<ChannelFilterManager?>((
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<String, dynamic>,
);
}
} 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;
});

View File

@@ -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<String, dynamic> _record;
ChannelFilterStore._(this._record);
factory ChannelFilterStore.empty() =>
ChannelFilterStore._({'version': 1, 'groups': <String, dynamic>{}});
factory ChannelFilterStore.parse(Map<String, dynamic> 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<String, dynamic>;
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'] ??= <String, dynamic>{};
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<String, dynamic> toJson() =>
jsonDecode(jsonEncode(_record)) as Map<String, dynamic>;
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<String, dynamic>() ??
{
'v': 1,
'g': <String, dynamic>{
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<String, dynamic>();
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<String, dynamic>() ??
<String, dynamic>{};
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,
};

View File

@@ -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 <int?>[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<NostrEvent> history;
var failRead = false;
var failPublish = false;
var retainPublish = true;
var live = true;
var publishes = 0;
late List<ChannelFilterStore> 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<List<NostrEvent>> 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<List<NostrEvent>>();
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);
});
}