twonly/twonly-app
1
0
Fork 0
mirror of https://github.com/twonlyapp/twonly-app.git synced 2026-09-01 09:44:07 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

fixing multiple issues

Browse source
This commit is contained in:
otsmr 2026-08-30 09:25:15 +02:00
parent d80494c206
commit d89952c295
37 changed files with 2211 additions and 429 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

663
android/build/reports/problems/problems-report.html Normal file
View file

File diff suppressed because one or more lines are too long

79
ios/Runner/AppDelegate.swift
View file

@ -81,12 +81,13 @@ import workmanager_apple
} }
/// Withdraws only the native notifications whose message IDs Dart reports as /// Withdraws only the native notifications whose message IDs Dart reports as
/// opened. The notification service extension's final alert keeps APNs' request /// opened, and keeps the app icon badge in step with them. The notification
/// identifier, so both the identifier and our `notification_id` user-info field /// service extension's final alert keeps APNs' request identifier, so both the
/// have to be considered. /// identifier and our `notification_id` user-info field have to be considered.
class NativeNotificationChannel { class NativeNotificationChannel {
private static let channelName = "eu.twonly/notificationTap" private static let channelName = "eu.twonly/notificationTap"
private static let notificationIdsKey = "notification_ids" private static let notificationIdsKey = "notification_ids"
private static let badgeCountKey = "badge_count"
static func register(with registry: FlutterPluginRegistry) { static func register(with registry: FlutterPluginRegistry) {
guard let registrar = registry.registrar(forPlugin: "TwonlyNativeNotifications") else { guard let registrar = registry.registrar(forPlugin: "TwonlyNativeNotifications") else {
@ -97,23 +98,67 @@ class NativeNotificationChannel {
binaryMessenger: registrar.messenger() binaryMessenger: registrar.messenger()
) )
channel.setMethodCallHandler { call, result in channel.setMethodCallHandler { call, result in
guard call.method == "cancelNotifications" else { switch call.method {
case "cancelNotifications":
guard
let arguments = call.arguments as? [String: Any],
let values = arguments[notificationIdsKey] as? [String]
else {
result(
FlutterError(
code: "invalid_notification_ids",
message: "notification_ids must be a list of strings",
details: nil
))
return
}
removeNotifications(Set(values), completion: result)
case "setBadgeCount":
guard
let arguments = call.arguments as? [String: Any],
let count = arguments[badgeCountKey] as? Int
else {
result(
FlutterError(
code: "invalid_badge_count",
message: "badge_count must be an integer",
details: nil
))
return
}
setBadgeCount(count, completion: result)
default:
result(FlutterMethodNotImplemented) result(FlutterMethodNotImplemented)
return
} }
guard }
let arguments = call.arguments as? [String: Any], }
let values = arguments[notificationIdsKey] as? [String]
else { /// iOS only takes an app icon badge from a notification payload, so a badge
result( /// set by the notification service extension survives until the app itself
FlutterError( /// overwrites it. Dart pushes the pending event count here whenever the
code: "invalid_notification_ids", /// notification outbox changes and on every resume.
message: "notification_ids must be a list of strings", private static func setBadgeCount(_ count: Int, completion: @escaping FlutterResult) {
details: nil let badge = max(0, count)
)) guard #available(iOS 16.0, *) else {
return DispatchQueue.main.async {
UIApplication.shared.applicationIconBadgeNumber = badge
completion(nil)
}
return
}
UNUserNotificationCenter.current().setBadgeCount(badge) { error in
DispatchQueue.main.async {
if let error {
completion(
FlutterError(
code: "badge_count_failed",
message: error.localizedDescription,
details: nil
))
} else {
completion(nil)
}
} }
removeNotifications(Set(values), completion: result)
} }
} }

5
lib/app.dart
View file

@ -17,6 +17,7 @@ import 'package:twonly/src/model/json/onboarding_state.model.dart';
import 'package:twonly/src/providers/routing.provider.dart'; import 'package:twonly/src/providers/routing.provider.dart';
import 'package:twonly/src/providers/settings.provider.dart'; import 'package:twonly/src/providers/settings.provider.dart';
import 'package:twonly/src/services/intent/links.intent.dart'; import 'package:twonly/src/services/intent/links.intent.dart';
import 'package:twonly/src/services/notifications/native.notifications.dart';
import 'package:twonly/src/utils/keyvalue.dart'; import 'package:twonly/src/utils/keyvalue.dart';
import 'package:twonly/src/utils/log.dart'; import 'package:twonly/src/utils/log.dart';
import 'package:twonly/src/utils/pow.dart'; import 'package:twonly/src/utils/pow.dart';
@ -63,6 +64,10 @@ class _AppState extends State<App> with WidgetsBindingObserver {
unawaited( unawaited(
rust_api.RustApi.setBackground(inBackground: false), rust_api.RustApi.setBackground(inBackground: false),
); );
// The notification service extension wrote to the outbox while the app
// was suspended, and its Rust change broadcast never reached this
// process, so the badge has to be re-read on the way back in.
unawaited(NativeNotificationService.refreshBadgeCount());
} }
} else if (state == AppLifecycleState.paused) { } else if (state == AppLifecycleState.paused) {
_wasPaused = true; _wasPaused = true;

22
lib/core/bridge/api.dart
View file

@ -293,6 +293,22 @@ class RustApi {
alreadyReceivedMessageIds: alreadyReceivedMessageIds, alreadyReceivedMessageIds: alreadyReceivedMessageIds,
); );
/// Acknowledges the pending contact-request notifications and returns the
/// native notification IDs to withdraw.
static Future<List<String>> clearContactRequestNotifications() => RustLib
.instance
.api
.crateBridgeApiRustApiClearContactRequestNotifications();
/// Acknowledges every pending notification of a conversation the user just
/// opened and returns the native notification IDs to withdraw.
static Future<List<String>> clearConversationNotifications({
required String conversationId,
}) =>
RustLib.instance.api.crateBridgeApiRustApiClearConversationNotifications(
conversationId: conversationId,
);
static Future<void> close() => static Future<void> close() =>
RustLib.instance.api.crateBridgeApiRustApiClose(); RustLib.instance.api.crateBridgeApiRustApiClose();
@ -432,6 +448,12 @@ class RustApi {
static Future<FrbPlanBalance> loadPlanBalance() => static Future<FrbPlanBalance> loadPlanBalance() =>
RustLib.instance.api.crateBridgeApiRustApiLoadPlanBalance(); RustLib.instance.api.crateBridgeApiRustApiLoadPlanBalance();
/// The number of pending notification events. iOS cannot derive its app
/// icon badge from the delivered alerts, so the running app pushes this
/// into `UNUserNotificationCenter` whenever the outbox changes.
static Future<PlatformInt64> notificationBadgeCount() =>
RustLib.instance.api.crateBridgeApiRustApiNotificationBadgeCount();
static Future<void> notifyMessagesOpened({ static Future<void> notifyMessagesOpened({
required PlatformInt64 contactId, required PlatformInt64 contactId,
required List<String> messageIds, required List<String> messageIds,

14
lib/core/bridge/callbacks/log.dart Normal file
View file

@ -0,0 +1,14 @@
// This file is automatically generated, so please do not edit it.
// @generated by `flutter_rust_bridge`@ 2.12.0.
// ignore_for_file: invalid_use_of_internal_member, unused_import, unnecessary_import
import '../../frb_generated.dart';
import 'package:flutter_rust_bridge/flutter_rust_bridge_for_generated.dart';
// These functions are ignored because they are not marked as `pub`: `set_dart_sink`
// These types are ignored because they are neither used by any `pub` functions nor (for structs and enums) marked `#[frb(unignore)]`: `DartWriter`
// These function are ignored because they are on traits that is not defined in current crate (put an empty `#[frb]` on it to unignore): `clone`, `flush`, `make_writer`, `write`
Future<String> stripAnsi({required String input}) =>
RustLib.instance.api.crateBridgeCallbacksLogStripAnsi(input: input);

13
lib/core/bridge/groups.dart
View file

@ -25,17 +25,20 @@ Future<bool> addHiddenContact({required PlatformInt64 contactId}) => RustLib
Future<void> fetchGroupStatesForUnjoinedGroups() => Future<void> fetchGroupStatesForUnjoinedGroups() =>
RustLib.instance.api.crateBridgeGroupsFetchGroupStatesForUnjoinedGroups(); RustLib.instance.api.crateBridgeGroupsFetchGroupStatesForUnjoinedGroups();
Future<void> fetchMissingGroupPublicKeys() => Future<void> fetchMissingGroupPublicKeys({
RustLib.instance.api.crateBridgeGroupsFetchMissingGroupPublicKeys(); String? groupId,
required bool force,
}) => RustLib.instance.api.crateBridgeGroupsFetchMissingGroupPublicKeys(
groupId: groupId,
force: force,
);
Future<bool> manageAdminState({ Future<bool> manageAdminState({
required String groupId, required String groupId,
required List<int> groupPublicKey,
required PlatformInt64 contactId, required PlatformInt64 contactId,
required bool remove, required bool remove,
}) => RustLib.instance.api.crateBridgeGroupsManageAdminState( }) => RustLib.instance.api.crateBridgeGroupsManageAdminState(
groupId: groupId, groupId: groupId,
groupPublicKey: groupPublicKey,
contactId: contactId, contactId: contactId,
remove: remove, remove: remove,
); );
@ -66,11 +69,9 @@ Future<bool> addNewGroupMembers({
Future<bool> removeMemberFromGroup({ Future<bool> removeMemberFromGroup({
required String groupId, required String groupId,
required List<int> groupPublicKey,
required PlatformInt64 contactId, required PlatformInt64 contactId,
}) => RustLib.instance.api.crateBridgeGroupsRemoveMemberFromGroup( }) => RustLib.instance.api.crateBridgeGroupsRemoveMemberFromGroup(
groupId: groupId, groupId: groupId,
groupPublicKey: groupPublicKey,
contactId: contactId, contactId: contactId,
); );

363
lib/core/frb_generated.dart
View file

File diff suppressed because it is too large Load diff

1
lib/core/frb_generated.io.dart
View file

@ -7,6 +7,7 @@ import 'api/server/prekeys.dart';
import 'bridge.dart'; import 'bridge.dart';
import 'bridge/api.dart'; import 'bridge/api.dart';
import 'bridge/callbacks.dart'; import 'bridge/callbacks.dart';
import 'bridge/callbacks/log.dart';
import 'bridge/groups.dart'; import 'bridge/groups.dart';
import 'bridge/user_config.dart'; import 'bridge/user_config.dart';
import 'bridge/wrapper.dart'; import 'bridge/wrapper.dart';

1
lib/core/frb_generated.web.dart
View file

@ -10,6 +10,7 @@ import 'api/server/prekeys.dart';
import 'bridge.dart'; import 'bridge.dart';
import 'bridge/api.dart'; import 'bridge/api.dart';
import 'bridge/callbacks.dart'; import 'bridge/callbacks.dart';
import 'bridge/callbacks/log.dart';
import 'bridge/groups.dart'; import 'bridge/groups.dart';
import 'bridge/user_config.dart'; import 'bridge/user_config.dart';
import 'bridge/wrapper.dart'; import 'bridge/wrapper.dart';

76
lib/src/services/notifications/native.notifications.dart
View file

@ -2,6 +2,8 @@ import 'dart:async';
import 'dart:io' show Platform; import 'dart:io' show Platform;
import 'package:flutter/services.dart'; import 'package:flutter/services.dart';
import 'package:twonly/core/bridge/api.dart';
import 'package:twonly/core/bridge/wrapper/app_database.dart';
import 'package:twonly/src/utils/log.dart'; import 'package:twonly/src/utils/log.dart';
/// Taps on notifications rendered natively (Android `MessagingStyle`) arrive /// Taps on notifications rendered natively (Android `MessagingStyle`) arrive
@ -17,16 +19,27 @@ class NativeNotificationService {
static const String _conversationIdKey = 'conversation_id'; static const String _conversationIdKey = 'conversation_id';
static const String _notificationIdsKey = 'notification_ids'; static const String _notificationIdsKey = 'notification_ids';
static const String _badgeCountKey = 'badge_count';
/// The table Rust commits to whenever a notification event is recorded,
/// delivered or cleared.
static const String _outboxTable = 'notification_outbox';
static final StreamController<String?> _taps = static final StreamController<String?> _taps =
StreamController<String?>.broadcast(); StreamController<String?>.broadcast();
/// Lives for the whole process: the badge has to follow the outbox for as
/// long as the app runs.
// ignore: cancel_subscriptions
static StreamSubscription<List<String>>? _outboxChanges;
/// Emits the conversation id of every notification tapped while the app is /// Emits the conversation id of every notification tapped while the app is
/// running. A `null` value means the notification had no specific /// running. A `null` value means the notification had no specific
/// conversation and should only open the chats tab. /// conversation and should only open the chats tab.
static Stream<String?> get taps => _taps.stream; static Stream<String?> get taps => _taps.stream;
static void init() { static void init() {
_startBadgeSync();
if (!Platform.isAndroid) return; if (!Platform.isAndroid) return;
_channel.setMethodCallHandler((call) async { _channel.setMethodCallHandler((call) async {
if (call.method != 'onNotificationTapped') return; if (call.method != 'onNotificationTapped') return;
@ -34,6 +47,69 @@ class NativeNotificationService {
}); });
} }
/// Keeps the iOS app icon badge in sync with the pending notification
/// events. Only the notification service extension ever sets a badge on a
/// delivered alert, so without this the number stays at whatever the last
/// push reported even after every message has been read.
static void _startBadgeSync() {
if (!Platform.isIOS || _outboxChanges != null) return;
try {
_outboxChanges = RustAppDatabase.changes().listen(
(tables) {
// An empty batch means Rust could not say what changed, so the
// outbox has to be assumed stale as well.
if (tables.isNotEmpty && !tables.contains(_outboxTable)) return;
unawaited(refreshBadgeCount());
},
onError: (Object error) {
Log.error('Notification badge change stream failed: $error');
},
);
} catch (e) {
Log.error('Could not watch the notification outbox: $e');
return;
}
unawaited(refreshBadgeCount());
}
/// Reads the pending event count from Rust and writes it to the app icon.
static Future<void> refreshBadgeCount() async {
if (!Platform.isIOS) return;
try {
final count = await RustApi.notificationBadgeCount();
await _channel.invokeMethod<void>('setBadgeCount', {
_badgeCountKey: count,
});
} catch (e) {
Log.error('Could not update the app icon badge: $e');
}
}
/// Acknowledges every pending notification of a conversation the user just
/// opened and withdraws the alerts still on screen.
static Future<void> clearConversation(String conversationId) async {
if (conversationId.isEmpty) return;
try {
final notificationIds = await RustApi.clearConversationNotifications(
conversationId: conversationId,
);
await cancelNotifications(notificationIds);
} catch (e) {
Log.error('Could not clear the notifications of a conversation: $e');
}
}
/// Acknowledges the contact requests the user just looked at. They belong to
/// no conversation, so this is the only moment they can be cleared.
static Future<void> clearContactRequests() async {
try {
final notificationIds = await RustApi.clearContactRequestNotifications();
await cancelNotifications(notificationIds);
} catch (e) {
Log.error('Could not clear the contact request notifications: $e');
}
}
/// Returns the tap that launched the app, or `null` when the app was not /// Returns the tap that launched the app, or `null` when the app was not
/// started from a native notification. The result is consumed once. /// started from a native notification. The result is consumed once.
static Future<({String? conversationId})?> consumeInitialTap() async { static Future<({String? conversationId})?> consumeInitialTap() async {

14
lib/src/visual/views/chats/chat_messages.view.dart
View file

@ -118,9 +118,23 @@ class _ChatMessagesViewState extends State<ChatMessagesView>
textFieldFocus = FocusNode(); textFieldFocus = FocusNode();
WidgetsBinding.instance.addObserver(this); WidgetsBinding.instance.addObserver(this);
itemPositionsListener.itemPositions.addListener(_loadOlderWhenNeeded); itemPositionsListener.itemPositions.addListener(_loadOlderWhenNeeded);
// Opening a conversation acknowledges everything it has pending, including
// the events that survive `notifyMessagesOpened` such as reactions and
// media status updates. Without this they keep inflating the app badge.
unawaited(NativeNotificationService.clearConversation(widget.groupId));
initStreams(); initStreams();
} }
@override
void didChangeAppLifecycleState(AppLifecycleState state) {
super.didChangeAppLifecycleState(state);
if (state != AppLifecycleState.resumed) return;
if (!mounted || !(ModalRoute.of(context)?.isCurrent ?? false)) return;
// Notifications that arrived while this chat sat in the background are
// acknowledged as soon as the user looks at it again.
unawaited(NativeNotificationService.clearConversation(widget.groupId));
}
@override @override
void dispose() { void dispose() {
_subscriptions.cancelAll(); _subscriptions.cancelAll();

4
lib/src/visual/views/contact/add_new_contact.view.dart
View file

@ -15,6 +15,7 @@ import 'package:twonly/src/database/daos/user_discovery.dao.dart';
import 'package:twonly/src/database/tables/contacts.table.dart'; import 'package:twonly/src/database/tables/contacts.table.dart';
import 'package:twonly/src/database/twonly.db.dart'; import 'package:twonly/src/database/twonly.db.dart';
import 'package:twonly/src/services/api/utils.api.dart'; import 'package:twonly/src/services/api/utils.api.dart';
import 'package:twonly/src/services/notifications/native.notifications.dart';
import 'package:twonly/src/utils/misc.dart'; import 'package:twonly/src/utils/misc.dart';
import 'package:twonly/src/visual/components/alert.dialog.dart'; import 'package:twonly/src/visual/components/alert.dialog.dart';
import 'package:twonly/src/visual/components/profile_qr_code.comp.dart'; import 'package:twonly/src/visual/components/profile_qr_code.comp.dart';
@ -55,6 +56,9 @@ class _SearchUsernameView extends State<AddNewUserView> {
@override @override
void initState() { void initState() {
super.initState(); super.initState();
// Contact requests belong to no conversation, so this list is the only
// place they can be acknowledged and taken off the app badge.
unawaited(NativeNotificationService.clearContactRequests());
_contactsStream = twonlyDB.contactsDao.watchNotAcceptedContacts().listen( _contactsStream = twonlyDB.contactsDao.watchNotAcceptedContacts().listen(
(update) { (update) {
if (mounted) { if (mounted) {

14
lib/src/visual/views/groups/group.view.dart
View file

@ -54,6 +54,16 @@ class _GroupViewState extends State<GroupView> {
} }
Future<void> initAsync() async { Future<void> initAsync() async {
// Opening the group is the user's way of retrying: ask for every member
// key that is still missing, ignoring the per-member request interval.
// Without a member's key the admin actions on them stay hidden.
unawaited(
rust_groups.fetchMissingGroupPublicKeys(
groupId: widget.groupId,
force: true,
),
);
final groupStream = twonlyDB.groupsDao.watchGroup(widget.groupId); final groupStream = twonlyDB.groupsDao.watchGroup(widget.groupId);
groupSub = groupStream.listen((update) { groupSub = groupStream.listen((update) {
if (update != null) { if (update != null) {
@ -67,7 +77,9 @@ class _GroupViewState extends State<GroupView> {
setState(() { setState(() {
members = update; members = update;
members.sort( members.sort(
(b, a) => a.$2.memberState!.index.compareTo(b.$2.memberState!.index), (b, a) => (a.$2.memberState ?? MemberState.normal).index.compareTo(
(b.$2.memberState ?? MemberState.normal).index,
),
); );
}); });
}); });

7
lib/src/visual/views/groups/group_member.context.dart
View file

@ -38,7 +38,6 @@ class GroupMemberContextMenu extends StatelessWidget {
if (ok) { if (ok) {
if (!await rust_groups.manageAdminState( if (!await rust_groups.manageAdminState(
groupId: group.groupId, groupId: group.groupId,
groupPublicKey: member.groupPublicKey!,
contactId: contact.userId, contactId: contact.userId,
remove: false, remove: false,
)) { )) {
@ -59,7 +58,6 @@ class GroupMemberContextMenu extends StatelessWidget {
if (ok) { if (ok) {
if (!await rust_groups.manageAdminState( if (!await rust_groups.manageAdminState(
groupId: group.groupId, groupId: group.groupId,
groupPublicKey: member.groupPublicKey!,
contactId: contact.userId, contactId: contact.userId,
remove: true, remove: true,
)) { )) {
@ -79,7 +77,6 @@ class GroupMemberContextMenu extends StatelessWidget {
if (ok) { if (ok) {
if (!await rust_groups.removeMemberFromGroup( if (!await rust_groups.removeMemberFromGroup(
groupId: group.groupId, groupId: group.groupId,
groupPublicKey: member.groupPublicKey!,
contactId: contact.userId, contactId: contact.userId,
)) { )) {
if (context.mounted) { if (context.mounted) {
@ -149,7 +146,7 @@ class GroupMemberContextMenu extends StatelessWidget {
), ),
if (member.groupPublicKey != null && if (member.groupPublicKey != null &&
group.isGroupAdmin && group.isGroupAdmin &&
member.memberState == MemberState.normal) (member.memberState ?? MemberState.normal) == MemberState.normal)
ContextMenuItem( ContextMenuItem(
title: context.lang.makeAdmin, title: context.lang.makeAdmin,
onTap: () => _makeContactAdmin(navigator.context), onTap: () => _makeContactAdmin(navigator.context),
@ -163,7 +160,7 @@ class GroupMemberContextMenu extends StatelessWidget {
onTap: () => _removeContactAsAdmin(navigator.context), onTap: () => _removeContactAsAdmin(navigator.context),
icon: FontAwesomeIcons.key, icon: FontAwesomeIcons.key,
), ),
if (group.isGroupAdmin && member.groupPublicKey != null) if (group.isGroupAdmin)
ContextMenuItem( ContextMenuItem(
title: context.lang.removeFromGroup, title: context.lang.removeFromGroup,
onTap: () => _removeContactFromGroup(navigator.context), onTap: () => _removeContactFromGroup(navigator.context),

12
rust/.sqlx/query-2b5ae0640585d4a0c5c1c521103d2279469d8645feca81bdadb717b39ad7ccd1.json Normal file
View file

@ -0,0 +1,12 @@
{
"db_name": "SQLite",
"query": "\n UPDATE notification_outbox SET cleared_at = ?\n WHERE kind IN ('contact_request', 'accept_request') AND cleared_at IS NULL\n ",
"describe": {
"columns": [],
"parameters": {
"Right": 1
},
"nullable": []
},
"hash": "2b5ae0640585d4a0c5c1c521103d2279469d8645feca81bdadb717b39ad7ccd1"
}

12
rust/.sqlx/query-411a854abb8d23c104b7290c25313c679ef0361348c17d2ee740d30ad8c5a424.json
View file

@ -1,12 +0,0 @@
{
"db_name": "SQLite",
"query": "UPDATE group_members SET last_message = CAST(strftime('%s','now') AS INTEGER) WHERE group_id = ? AND contact_id = ?",
"describe": {
"columns": [],
"parameters": {
"Right": 2
},
"nullable": []
},
"hash": "411a854abb8d23c104b7290c25313c679ef0361348c17d2ee740d30ad8c5a424"
}

26
rust/.sqlx/query-5164aba96e6ee8f761644482263b30e96adf6a2704e98310de21fc832f7a187f.json Normal file
View file

@ -0,0 +1,26 @@
{
"db_name": "SQLite",
"query": "SELECT signal_version FROM contacts WHERE user_id = ?",
"describe": {
"columns": [
{
"name": "signal_version",
"ordinal": 0,
"type_info": "Text",
"origin": {
"Table": {
"table": "contacts",
"name": "signal_version"
}
}
}
],
"parameters": {
"Right": 1
},
"nullable": [
false
]
},
"hash": "5164aba96e6ee8f761644482263b30e96adf6a2704e98310de21fc832f7a187f"
}

12
rust/.sqlx/query-7cc39bf45c3395a1f71a40d027afb0b2b7a14f18fa0ed0c66e5c9cac5678f4d4.json Normal file
View file

@ -0,0 +1,12 @@
{
"db_name": "SQLite",
"query": "UPDATE contacts SET signal_version = 'v2', account_deleted = 0 WHERE user_id = ?",
"describe": {
"columns": [],
"parameters": {
"Right": 1
},
"nullable": []
},
"hash": "7cc39bf45c3395a1f71a40d027afb0b2b7a14f18fa0ed0c66e5c9cac5678f4d4"
}

26
rust/.sqlx/query-9841b7394adf2125090839e8381f8db55945d8230b5450aacc1620c975a59387.json Normal file
View file

@ -0,0 +1,26 @@
{
"db_name": "SQLite",
"query": "\n SELECT notification_id FROM notification_outbox\n WHERE kind IN ('contact_request', 'accept_request') AND cleared_at IS NULL\n ",
"describe": {
"columns": [
{
"name": "notification_id",
"ordinal": 0,
"type_info": "Text",
"origin": {
"Table": {
"table": "notification_outbox",
"name": "notification_id"
}
}
}
],
"parameters": {
"Right": 0
},
"nullable": [
false
]
},
"hash": "9841b7394adf2125090839e8381f8db55945d8230b5450aacc1620c975a59387"
}

6
rust/.sqlx/query-6d6a5fb10742309ab6e4242179ef86b64ac9f751f1c6d647d4ecfa5ecac5d332.json → rust/.sqlx/query-e15b811d31ca59847e3d4ee2eff71410100258a0a5fe894233d63b6ac770368f.json
View file

@ -1,6 +1,6 @@
{ {
"db_name": "SQLite", "db_name": "SQLite",
"query": "\n SELECT group_id, contact_id \n FROM group_members \n WHERE group_public_key IS NULL \n AND last_message >= CAST(strftime('%s','now','-2 days') AS INTEGER)\n ", "query": "\n SELECT members.group_id, members.contact_id\n FROM group_members AS members\n JOIN groups ON groups.group_id = members.group_id\n WHERE members.group_public_key IS NULL\n AND (members.member_state IS NULL OR members.member_state != 'leftGroup')\n AND groups.is_direct_chat = 0\n AND groups.left_group = 0\n AND (? IS NULL OR members.group_id = ?)\n ",
"describe": { "describe": {
"columns": [ "columns": [
{ {
@ -27,12 +27,12 @@
} }
], ],
"parameters": { "parameters": {
"Right": 0 "Right": 2
}, },
"nullable": [ "nullable": [
false, false,
false false
] ]
}, },
"hash": "6d6a5fb10742309ab6e4242179ef86b64ac9f751f1c6d647d4ecfa5ecac5d332" "hash": "e15b811d31ca59847e3d4ee2eff71410100258a0a5fe894233d63b6ac770368f"
} }

102
rust/src/api/messages/incoming.rs
View file

@ -203,6 +203,65 @@ pub(crate) async fn handle_request_new_pqc_prekeys(
)) ))
} }
/// Rebuilds a v2 Signal session for a peer that still speaks the legacy
/// protocol, so the decryption error queued for the message can be answered
/// with a session the peer can upgrade to.
///
/// Runs before the inbound transaction opens: it awaits a server round-trip and
/// `establish_signal_session` writes through the pool, and the app database
/// allows a single connection, so doing this under an open transaction would
/// deadlock until the acquire timeout.
async fn upgrade_legacy_session_to_v2(
ctx: &Arc<Context>,
database: &Arc<crate::database::app::AppDatabase>,
from_user_id: i64,
) -> Result<()> {
tracing::info!("Received legacy signal message; rejecting and upgrading session to v2");
let has_v2_session = {
let rust_database = ctx.rust_db.read().await.clone();
sqlx::query_scalar!(
"SELECT EXISTS(SELECT 1 FROM signal_sessions WHERE name = ? AND device_id = 1)",
from_user_id.to_string(),
)
.fetch_one(&rust_database.pool)
.await?
!= 0
};
let is_v2_contact = sqlx::query_scalar!(
"SELECT signal_version FROM contacts WHERE user_id = ?",
from_user_id,
)
.fetch_optional(&database.pool)
.await?
.is_some_and(|version| version == "v2");
if has_v2_session && is_v2_contact {
return Ok(());
}
if let Err(error) = ContactService::new(ctx)
.establish_signal_session(from_user_id, None)
.await
{
tracing::warn!(
from_user_id,
"failed to establish v2 signal session from server: {error}"
);
}
sqlx::query!(
"UPDATE contacts SET signal_version = 'v2' WHERE user_id = ?",
from_user_id,
)
.execute(&database.pool)
.await?;
database.notify_committed(["contacts"]);
Ok(())
}
#[tracing::instrument( #[tracing::instrument(
skip_all, skip_all,
fields( fields(
@ -242,6 +301,13 @@ pub(crate) async fn handle_decoded_server_message(
let database = ctx.app_db.read().await.clone(); let database = ctx.app_db.read().await.clone();
// Upgrading a legacy peer to a v2 session needs a server round-trip and
// writes through the pool itself, so it has to finish before the
// transaction below claims the single app-database connection.
if matches!(message_type, Type::Ciphertext | Type::PrekeyBundle) {
upgrade_legacy_session_to_v2(ctx, &database, from_user_id).await?;
}
let mut t = database.pool.begin().await?; let mut t = database.pool.begin().await?;
let claimed = Receipt::claim_received(&mut t, &message.receipt_id).await?; let claimed = Receipt::claim_received(&mut t, &message.receipt_id).await?;
@ -277,42 +343,6 @@ pub(crate) async fn handle_decoded_server_message(
handle_sender_delivery_receipt(&mut t, from_user_id, &message.receipt_id).await?; handle_sender_delivery_receipt(&mut t, from_user_id, &message.receipt_id).await?;
} }
Type::Ciphertext | Type::PrekeyBundle => { Type::Ciphertext | Type::PrekeyBundle => {
tracing::info!("Received legacy signal message; rejecting and upgrading session to v2");
let has_v2_session = {
let rust_database = ctx.rust_db.read().await.clone();
sqlx::query_scalar!(
"SELECT EXISTS(SELECT 1 FROM signal_sessions WHERE name = ? AND device_id = 1)",
from_user_id.to_string(),
)
.fetch_one(&rust_database.pool)
.await?
!= 0
};
let is_v2_contact = {
let contact = Contact::get_contact_by_id(&mut t, from_user_id).await?;
contact.as_ref().is_some_and(|c| c.signal_version == "v2")
};
if !has_v2_session || !is_v2_contact {
if let Err(error) = ContactService::new(ctx)
.establish_signal_session(from_user_id, None)
.await
{
tracing::warn!(
from_user_id,
"failed to establish v2 signal session from server: {error}"
);
}
sqlx::query!(
"UPDATE contacts SET signal_version = 'v2' WHERE user_id = ?",
from_user_id,
)
.execute(&mut *t)
.await?;
}
queue_decryption_error(&mut t, from_user_id, &message.receipt_id, 0).await?; queue_decryption_error(&mut t, from_user_id, &message.receipt_id, 0).await?;
sends_error_response = true; sends_error_response = true;
} }

21
rust/src/api/messages/incoming/errors.rs
View file

@ -46,14 +46,19 @@ pub(crate) async fn handle_error_message(
} }
Type::GroupNotFoundOrNotAMember => { Type::GroupNotFoundOrNotAMember => {
if let Some(group_id) = group_id { if let Some(group_id) = group_id {
GroupService::new(ctx) // The repair refreshes the group from the server and owns its
.handle_membership_error( // own transaction, so it has to run once this one commits.
t, let ctx = ctx.clone();
from_user_id, let group_id = group_id.to_owned();
group_id.to_owned(), let related_receipt_id = error.related_receipt_id;
error.related_receipt_id, tokio::spawn(async move {
) if let Err(error) = GroupService::new(&ctx)
.await?; .handle_membership_error(from_user_id, group_id, related_receipt_id)
.await
{
tracing::warn!("group membership repair failed: {error}");
}
});
} }
} }
Type::SessionOutOfSync | Type::UnknownMessageType => {} Type::SessionOutOfSync | Type::UnknownMessageType => {}

8
rust/src/api/messages/incoming/groups.rs
View file

@ -91,9 +91,7 @@ pub(crate) async fn handle_group_create(
.execute(&mut **t) .execute(&mut **t)
.await?; .await?;
GroupService::new(ctx) GroupService::spawn_state_refresh_and_announce(ctx, group_id.to_owned());
.refresh_group_state(t, group_id.to_owned(), true)
.await;
Ok(()) Ok(())
} }
@ -175,9 +173,7 @@ pub(crate) async fn handle_group_update(
let is_direct = Group::is_direct_chat(t, group_id).await?; let is_direct = Group::is_direct_chat(t, group_id).await?;
if !is_direct { if !is_direct {
GroupService::new(ctx) GroupService::spawn_state_refresh(ctx, Some(group_id.to_owned()));
.refresh_group_state(t, group_id.to_owned(), false)
.await;
} }
if update.group_action_type == "updatedGroupName" { if update.group_action_type == "updatedGroupName" {

56
rust/src/api/messages/incoming/messages.rs
View file

@ -298,7 +298,6 @@ pub(crate) struct PreparedQueuedReceipt {
pub contact_id: i64, pub contact_id: i64,
pub message_id: Option<String>, pub message_id: Option<String>,
pub contact_will_sends_receipt: i64, pub contact_will_sends_receipt: i64,
pub account_deleted: i64,
pub payload: Vec<u8>, pub payload: Vec<u8>,
pub wake_receiver: bool, pub wake_receiver: bool,
} }
@ -313,12 +312,11 @@ struct PreparedQueuedReceiptRow {
signal_version: String, signal_version: String,
} }
pub(crate) async fn prepare_queued_receipt_details( async fn load_queued_receipt_row(
ctx: &Arc<Context>, pool: &sqlx::SqlitePool,
receipt_id: &str, receipt_id: &str,
) -> Result<Option<PreparedQueuedReceipt>> { ) -> Result<Option<PreparedQueuedReceiptRow>> {
let app_db = ctx.app_db.read().await.clone(); Ok(sqlx::query_as!(
let row = sqlx::query_as!(
PreparedQueuedReceiptRow, PreparedQueuedReceiptRow,
r#" r#"
SELECT r.contact_id, r.message, r.message_id, r.contact_will_sends_receipt, SELECT r.contact_id, r.message, r.message_id, r.contact_will_sends_receipt,
@ -329,13 +327,39 @@ pub(crate) async fn prepare_queued_receipt_details(
"#, "#,
receipt_id, receipt_id,
) )
.fetch_optional(&app_db.pool) .fetch_optional(pool)
.await?; .await?)
}
let Some(row) = row else { pub(crate) async fn prepare_queued_receipt_details(
ctx: &Arc<Context>,
receipt_id: &str,
) -> Result<Option<PreparedQueuedReceipt>> {
let app_db = ctx.app_db.read().await.clone();
let Some(row) = load_queued_receipt_row(&app_db.pool, receipt_id).await? else {
return Ok(None); return Ok(None);
}; };
if row.account_deleted != 0 {
return Err(TwonlyError::Generic(format!(
"contact {} deleted their account",
row.contact_id
)));
}
prepare_queued_receipt_from_row(ctx, receipt_id, row)
.await
.map(Some)
}
/// Encrypts the queued payload. Fetches a prekey bundle from the server when
/// the contact has no v2 session yet, so the caller must have ruled out a
/// deleted account first.
async fn prepare_queued_receipt_from_row(
ctx: &Arc<Context>,
receipt_id: &str,
row: PreparedQueuedReceiptRow,
) -> Result<PreparedQueuedReceipt> {
let mut message = proto::Message::decode(row.message.as_slice()) let mut message = proto::Message::decode(row.message.as_slice())
.map_err(|error| TwonlyError::Generic(format!("invalid queued message: {error}")))?; .map_err(|error| TwonlyError::Generic(format!("invalid queued message: {error}")))?;
message.receipt_id = receipt_id.to_owned(); message.receipt_id = receipt_id.to_owned();
@ -366,14 +390,13 @@ pub(crate) async fn prepare_queued_receipt_details(
message.r#type = proto::message::Type::CiphertextV2 as i32; message.r#type = proto::message::Type::CiphertextV2 as i32;
} }
Ok(Some(PreparedQueuedReceipt { Ok(PreparedQueuedReceipt {
contact_id: row.contact_id, contact_id: row.contact_id,
message_id: row.message_id, message_id: row.message_id,
contact_will_sends_receipt: row.contact_will_sends_receipt, contact_will_sends_receipt: row.contact_will_sends_receipt,
account_deleted: row.account_deleted,
payload: message.encode_to_vec(), payload: message.encode_to_vec(),
wake_receiver: row.wake_receiver != 0, wake_receiver: row.wake_receiver != 0,
})) })
} }
pub(crate) async fn prepare_queued_receipt( pub(crate) async fn prepare_queued_receipt(
@ -402,16 +425,21 @@ pub(crate) async fn send_queued_receipt(ctx: &Arc<Context>, receipt_id: &str) ->
} }
let app_db = ctx.app_db.read().await.clone(); let app_db = ctx.app_db.read().await.clone();
let Some(receipt) = prepare_queued_receipt_details(ctx, receipt_id).await? else { let Some(row) = load_queued_receipt_row(&app_db.pool, receipt_id).await? else {
return Ok(()); return Ok(());
}; };
if receipt.account_deleted != 0 { // The server has no account for this contact any more, so preparing the
// payload would only fetch a prekey bundle that answers `UserIdNotFound`
// on every retry. Drop the receipt instead of queueing it forever.
if row.account_deleted != 0 {
Receipt::delete(&app_db.pool, receipt_id).await?; Receipt::delete(&app_db.pool, receipt_id).await?;
app_db.notify_committed(["receipts"]); app_db.notify_committed(["receipts"]);
return Ok(()); return Ok(());
} }
let receipt = prepare_queued_receipt_from_row(ctx, receipt_id, row).await?;
match Server::send_text_message( match Server::send_text_message(
ctx, ctx,
receipt.contact_id, receipt.contact_id,

5
rust/src/api/server/contacts.rs
View file

@ -15,11 +15,14 @@ use std::sync::Arc;
impl Server { impl Server {
pub async fn get_user_by_id_response(ctx: &Arc<Context>, user_id: i64) -> Result<Vec<u8>> { pub async fn get_user_by_id_response(ctx: &Arc<Context>, user_id: i64) -> Result<Vec<u8>> {
Self::application( // Sent for the contact so that a `UserIdNotFound` answer marks the
// account as deleted instead of failing every prekey bundle fetch.
Self::application_for_contact(
ctx, ctx,
client_to_server::application_data::ApplicationData::GetUserById( client_to_server::application_data::ApplicationData::GetUserById(
client_to_server::application_data::GetUserById { user_id }, client_to_server::application_data::GetUserById { user_id },
), ),
user_id,
) )
.await .await
} }

20
rust/src/bridge/api.rs
View file

@ -667,6 +667,26 @@ impl RustApi {
.await .await
} }
/// The number of pending notification events. iOS cannot derive its app
/// icon badge from the delivered alerts, so the running app pushes this
/// into `UNUserNotificationCenter` whenever the outbox changes.
pub async fn notification_badge_count() -> Result<i64> {
crate::services::notifications::badge_count(Context::get_static()?).await
}
/// Acknowledges every pending notification of a conversation the user just
/// opened and returns the native notification IDs to withdraw.
pub async fn clear_conversation_notifications(conversation_id: String) -> Result<Vec<String>> {
crate::services::notifications::clear_conversation(Context::get_static()?, &conversation_id)
.await
}
/// Acknowledges the pending contact-request notifications and returns the
/// native notification IDs to withdraw.
pub async fn clear_contact_request_notifications() -> Result<Vec<String>> {
crate::services::notifications::clear_contact_requests(Context::get_static()?).await
}
pub async fn send_contact_profile(contact_id: i64) -> Result<()> { pub async fn send_contact_profile(contact_id: i64) -> Result<()> {
let ctx = Context::get_static()?; let ctx = Context::get_static()?;
ContactService::new(ctx).send_profile(contact_id).await ContactService::new(ctx).send_profile(contact_id).await

19
rust/src/bridge/groups.rs
View file

@ -29,19 +29,14 @@ pub async fn fetch_group_states_for_unjoined_groups() -> Result<()> {
.await .await
} }
pub async fn fetch_missing_group_public_keys() -> Result<()> { pub async fn fetch_missing_group_public_keys(group_id: Option<String>, force: bool) -> Result<()> {
let ctx = crate::context::Context::get_static()?; let ctx = crate::context::Context::get_static()?;
GroupService::new(ctx) GroupService::new(ctx)
.fetch_missing_group_public_keys() .fetch_missing_group_public_keys(group_id, force)
.await .await
} }
pub async fn manage_admin_state( pub async fn manage_admin_state(group_id: String, contact_id: i64, remove: bool) -> Result<bool> {
group_id: String,
_group_public_key: Vec<u8>,
contact_id: i64,
remove: bool,
) -> Result<bool> {
let ctx = crate::context::Context::get_static()?; let ctx = crate::context::Context::get_static()?;
GroupService::new(ctx) GroupService::new(ctx)
.manage_admin(group_id, contact_id, remove) .manage_admin(group_id, contact_id, remove)
@ -72,14 +67,10 @@ pub async fn add_new_group_members(group_id: String, member_ids: Vec<i64>) -> Re
.await .await
} }
pub async fn remove_member_from_group( pub async fn remove_member_from_group(group_id: String, contact_id: i64) -> Result<bool> {
group_id: String,
group_public_key: Vec<u8>,
contact_id: i64,
) -> Result<bool> {
let ctx = crate::context::Context::get_static()?; let ctx = crate::context::Context::get_static()?;
GroupService::new(ctx) GroupService::new(ctx)
.remove_member(group_id, group_public_key, contact_id) .remove_member(group_id, contact_id)
.await .await
} }

52
rust/src/database/app/tables/groups.rs
View file

@ -472,27 +472,67 @@ pub struct MissingGroupPublicKeyRow {
pub contact_id: i64, pub contact_id: i64,
} }
/// Lists the members whose group public key is still unknown.
///
/// Direct chats are excluded: they never carry a group identity, so the peer
/// has no key to resend and would answer every request with an error. Left
/// groups and members who left are excluded for the same reason -- their key
/// is of no use any more.
#[derive(bon::Builder)] #[derive(bon::Builder)]
pub struct GetMissingGroupPublicKeys {} pub struct GetMissingGroupPublicKeys {
group_id: Option<String>,
}
impl GetMissingGroupPublicKeys { impl GetMissingGroupPublicKeys {
pub async fn fetch_all( pub async fn fetch_all(
self, self,
pool: &sqlx::Pool<Sqlite>, pool: &sqlx::Pool<Sqlite>,
) -> Result<Vec<MissingGroupPublicKeyRow>> { ) -> Result<Vec<MissingGroupPublicKeyRow>> {
let group_id = self.group_id.as_deref();
let rows = sqlx::query_as!( let rows = sqlx::query_as!(
MissingGroupPublicKeyRow, MissingGroupPublicKeyRow,
r#" r#"
SELECT group_id, contact_id SELECT members.group_id, members.contact_id
FROM group_members FROM group_members AS members
WHERE group_public_key IS NULL JOIN groups ON groups.group_id = members.group_id
AND last_message >= CAST(strftime('%s','now','-2 days') AS INTEGER) WHERE members.group_public_key IS NULL
"# AND (members.member_state IS NULL OR members.member_state != 'leftGroup')
AND groups.is_direct_chat = 0
AND groups.left_group = 0
AND (? IS NULL OR members.group_id = ?)
"#,
group_id,
group_id,
) )
.fetch_all(pool) .fetch_all(pool)
.await?; .await?;
Ok(rows) Ok(rows)
} }
pub async fn fetch_all_in_transaction(
self,
tr: &mut Transaction<'_, Sqlite>,
) -> Result<Vec<MissingGroupPublicKeyRow>> {
let group_id = self.group_id.as_deref();
let rows = sqlx::query_as!(
MissingGroupPublicKeyRow,
r#"
SELECT members.group_id, members.contact_id
FROM group_members AS members
JOIN groups ON groups.group_id = members.group_id
WHERE members.group_public_key IS NULL
AND (members.member_state IS NULL OR members.member_state != 'leftGroup')
AND groups.is_direct_chat = 0
AND groups.left_group = 0
AND (? IS NULL OR members.group_id = ?)
"#,
group_id,
group_id,
)
.fetch_all(&mut **tr)
.await?;
Ok(rows)
}
} }
fn current_unix_timestamp() -> Result<i64> { fn current_unix_timestamp() -> Result<i64> {

358
rust/src/frb_generated.rs
View file

@ -38,7 +38,7 @@ flutter_rust_bridge::frb_generated_boilerplate!(
default_rust_auto_opaque = RustAutoOpaqueMoi, default_rust_auto_opaque = RustAutoOpaqueMoi,
); );
pub(crate) const FLUTTER_RUST_BRIDGE_CODEGEN_VERSION: &str = "2.12.0"; pub(crate) const FLUTTER_RUST_BRIDGE_CODEGEN_VERSION: &str = "2.12.0";
pub(crate) const FLUTTER_RUST_BRIDGE_CODEGEN_CONTENT_HASH: i32 = -1780439173; pub(crate) const FLUTTER_RUST_BRIDGE_CODEGEN_CONTENT_HASH: i32 = 463166747;
// Section: executor // Section: executor
@ -258,12 +258,17 @@ fn wire__crate__bridge__groups__fetch_missing_group_public_keys_impl(
}; };
let mut deserializer = let mut deserializer =
flutter_rust_bridge::for_generated::SseDeserializer::new(message); flutter_rust_bridge::for_generated::SseDeserializer::new(message);
let api_group_id = <Option<String>>::sse_decode(&mut deserializer);
let api_force = <bool>::sse_decode(&mut deserializer);
deserializer.end(); deserializer.end();
move |context| async move { move |context| async move {
transform_result_sse::<_, flutter_rust_bridge::for_generated::anyhow::Error>( transform_result_sse::<_, flutter_rust_bridge::for_generated::anyhow::Error>(
(move || async move { (move || async move {
let output_ok = let output_ok = crate::bridge::groups::fetch_missing_group_public_keys(
crate::bridge::groups::fetch_missing_group_public_keys().await?; api_group_id,
api_force,
)
.await?;
Ok(output_ok) Ok(output_ok)
})() })()
.await, .await,
@ -520,7 +525,6 @@ fn wire__crate__bridge__groups__manage_admin_state_impl(
let mut deserializer = let mut deserializer =
flutter_rust_bridge::for_generated::SseDeserializer::new(message); flutter_rust_bridge::for_generated::SseDeserializer::new(message);
let api_group_id = <String>::sse_decode(&mut deserializer); let api_group_id = <String>::sse_decode(&mut deserializer);
let api__group_public_key = <Vec<u8>>::sse_decode(&mut deserializer);
let api_contact_id = <i64>::sse_decode(&mut deserializer); let api_contact_id = <i64>::sse_decode(&mut deserializer);
let api_remove = <bool>::sse_decode(&mut deserializer); let api_remove = <bool>::sse_decode(&mut deserializer);
deserializer.end(); deserializer.end();
@ -529,7 +533,6 @@ fn wire__crate__bridge__groups__manage_admin_state_impl(
(move || async move { (move || async move {
let output_ok = crate::bridge::groups::manage_admin_state( let output_ok = crate::bridge::groups::manage_admin_state(
api_group_id, api_group_id,
api__group_public_key,
api_contact_id, api_contact_id,
api_remove, api_remove,
) )
@ -565,7 +568,6 @@ fn wire__crate__bridge__groups__remove_member_from_group_impl(
let mut deserializer = let mut deserializer =
flutter_rust_bridge::for_generated::SseDeserializer::new(message); flutter_rust_bridge::for_generated::SseDeserializer::new(message);
let api_group_id = <String>::sse_decode(&mut deserializer); let api_group_id = <String>::sse_decode(&mut deserializer);
let api_group_public_key = <Vec<u8>>::sse_decode(&mut deserializer);
let api_contact_id = <i64>::sse_decode(&mut deserializer); let api_contact_id = <i64>::sse_decode(&mut deserializer);
deserializer.end(); deserializer.end();
move |context| async move { move |context| async move {
@ -573,7 +575,6 @@ fn wire__crate__bridge__groups__remove_member_from_group_impl(
(move || async move { (move || async move {
let output_ok = crate::bridge::groups::remove_member_from_group( let output_ok = crate::bridge::groups::remove_member_from_group(
api_group_id, api_group_id,
api_group_public_key,
api_contact_id, api_contact_id,
) )
.await?; .await?;
@ -805,6 +806,83 @@ fn wire__crate__bridge__api__rust_api_check_for_passwordless_notification_impl(
}, },
) )
} }
fn wire__crate__bridge__api__rust_api_clear_contact_request_notifications_impl(
port_: flutter_rust_bridge::for_generated::MessagePort,
ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr,
rust_vec_len_: i32,
data_len_: i32,
) {
FLUTTER_RUST_BRIDGE_HANDLER.wrap_async::<flutter_rust_bridge::for_generated::SseCodec, _, _, _>(
flutter_rust_bridge::for_generated::TaskInfo {
debug_name: "rust_api_clear_contact_request_notifications",
port: Some(port_),
mode: flutter_rust_bridge::for_generated::FfiCallMode::Normal,
},
move || {
let message = unsafe {
flutter_rust_bridge::for_generated::Dart2RustMessageSse::from_wire(
ptr_,
rust_vec_len_,
data_len_,
)
};
let mut deserializer =
flutter_rust_bridge::for_generated::SseDeserializer::new(message);
deserializer.end();
move |context| async move {
transform_result_sse::<_, flutter_rust_bridge::for_generated::anyhow::Error>(
(move || async move {
let output_ok =
crate::bridge::api::RustApi::clear_contact_request_notifications()
.await?;
Ok(output_ok)
})()
.await,
)
}
},
)
}
fn wire__crate__bridge__api__rust_api_clear_conversation_notifications_impl(
port_: flutter_rust_bridge::for_generated::MessagePort,
ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr,
rust_vec_len_: i32,
data_len_: i32,
) {
FLUTTER_RUST_BRIDGE_HANDLER.wrap_async::<flutter_rust_bridge::for_generated::SseCodec, _, _, _>(
flutter_rust_bridge::for_generated::TaskInfo {
debug_name: "rust_api_clear_conversation_notifications",
port: Some(port_),
mode: flutter_rust_bridge::for_generated::FfiCallMode::Normal,
},
move || {
let message = unsafe {
flutter_rust_bridge::for_generated::Dart2RustMessageSse::from_wire(
ptr_,
rust_vec_len_,
data_len_,
)
};
let mut deserializer =
flutter_rust_bridge::for_generated::SseDeserializer::new(message);
let api_conversation_id = <String>::sse_decode(&mut deserializer);
deserializer.end();
move |context| async move {
transform_result_sse::<_, flutter_rust_bridge::for_generated::anyhow::Error>(
(move || async move {
let output_ok =
crate::bridge::api::RustApi::clear_conversation_notifications(
api_conversation_id,
)
.await?;
Ok(output_ok)
})()
.await,
)
}
},
)
}
fn wire__crate__bridge__api__rust_api_close_impl( fn wire__crate__bridge__api__rust_api_close_impl(
port_: flutter_rust_bridge::for_generated::MessagePort, port_: flutter_rust_bridge::for_generated::MessagePort,
ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr, ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr,
@ -1835,6 +1913,42 @@ fn wire__crate__bridge__api__rust_api_load_plan_balance_impl(
}, },
) )
} }
fn wire__crate__bridge__api__rust_api_notification_badge_count_impl(
port_: flutter_rust_bridge::for_generated::MessagePort,
ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr,
rust_vec_len_: i32,
data_len_: i32,
) {
FLUTTER_RUST_BRIDGE_HANDLER.wrap_async::<flutter_rust_bridge::for_generated::SseCodec, _, _, _>(
flutter_rust_bridge::for_generated::TaskInfo {
debug_name: "rust_api_notification_badge_count",
port: Some(port_),
mode: flutter_rust_bridge::for_generated::FfiCallMode::Normal,
},
move || {
let message = unsafe {
flutter_rust_bridge::for_generated::Dart2RustMessageSse::from_wire(
ptr_,
rust_vec_len_,
data_len_,
)
};
let mut deserializer =
flutter_rust_bridge::for_generated::SseDeserializer::new(message);
deserializer.end();
move |context| async move {
transform_result_sse::<_, flutter_rust_bridge::for_generated::anyhow::Error>(
(move || async move {
let output_ok =
crate::bridge::api::RustApi::notification_badge_count().await?;
Ok(output_ok)
})()
.await,
)
}
},
)
}
fn wire__crate__bridge__api__rust_api_notify_messages_opened_impl( fn wire__crate__bridge__api__rust_api_notify_messages_opened_impl(
port_: flutter_rust_bridge::for_generated::MessagePort, port_: flutter_rust_bridge::for_generated::MessagePort,
ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr, ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr,
@ -3915,6 +4029,40 @@ fn wire__crate__bridge__wrapper__rust_utils_recover_secret_impl(
}, },
) )
} }
fn wire__crate__bridge__callbacks__log__strip_ansi_impl(
port_: flutter_rust_bridge::for_generated::MessagePort,
ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr,
rust_vec_len_: i32,
data_len_: i32,
) {
FLUTTER_RUST_BRIDGE_HANDLER.wrap_normal::<flutter_rust_bridge::for_generated::SseCodec, _, _>(
flutter_rust_bridge::for_generated::TaskInfo {
debug_name: "strip_ansi",
port: Some(port_),
mode: flutter_rust_bridge::for_generated::FfiCallMode::Normal,
},
move || {
let message = unsafe {
flutter_rust_bridge::for_generated::Dart2RustMessageSse::from_wire(
ptr_,
rust_vec_len_,
data_len_,
)
};
let mut deserializer =
flutter_rust_bridge::for_generated::SseDeserializer::new(message);
let api_input = <String>::sse_decode(&mut deserializer);
deserializer.end();
move |context| {
transform_result_sse::<_, ()>((move || {
let output_ok =
Result::<_, ()>::Ok(crate::bridge::callbacks::log::strip_ansi(&api_input))?;
Ok(output_ok)
})())
}
},
)
}
fn wire__crate__bridge__groups__update_chat_deletion_time_impl( fn wire__crate__bridge__groups__update_chat_deletion_time_impl(
port_: flutter_rust_bridge::for_generated::MessagePort, port_: flutter_rust_bridge::for_generated::MessagePort,
ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr, ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr,
@ -5596,101 +5744,105 @@ fn pde_ffi_dispatcher_primary_impl(
19 => wire__crate__bridge__api__rust_api_change_username_impl(port, ptr, rust_vec_len, data_len), 19 => wire__crate__bridge__api__rust_api_change_username_impl(port, ptr, rust_vec_len, data_len),
20 => wire__crate__bridge__api__rust_api_check_for_deleted_usernames_impl(port, ptr, rust_vec_len, data_len), 20 => wire__crate__bridge__api__rust_api_check_for_deleted_usernames_impl(port, ptr, rust_vec_len, data_len),
21 => wire__crate__bridge__api__rust_api_check_for_passwordless_notification_impl(port, ptr, rust_vec_len, data_len), 21 => wire__crate__bridge__api__rust_api_check_for_passwordless_notification_impl(port, ptr, rust_vec_len, data_len),
22 => wire__crate__bridge__api__rust_api_close_impl(port, ptr, rust_vec_len, data_len), 22 => wire__crate__bridge__api__rust_api_clear_contact_request_notifications_impl(port, ptr, rust_vec_len, data_len),
23 => wire__crate__bridge__api__rust_api_confirm_memories_upload_impl(port, ptr, rust_vec_len, data_len), 23 => wire__crate__bridge__api__rust_api_clear_conversation_notifications_impl(port, ptr, rust_vec_len, data_len),
24 => wire__crate__bridge__api__rust_api_connect_impl(port, ptr, rust_vec_len, data_len), 24 => wire__crate__bridge__api__rust_api_close_impl(port, ptr, rust_vec_len, data_len),
25 => wire__crate__bridge__api__rust_api_connection_state_impl(port, ptr, rust_vec_len, data_len), 25 => wire__crate__bridge__api__rust_api_confirm_memories_upload_impl(port, ptr, rust_vec_len, data_len),
26 => wire__crate__bridge__api__rust_api_delete_account_impl(port, ptr, rust_vec_len, data_len), 26 => wire__crate__bridge__api__rust_api_connect_impl(port, ptr, rust_vec_len, data_len),
27 => wire__crate__bridge__api__rust_api_delete_memory_impl(port, ptr, rust_vec_len, data_len), 27 => wire__crate__bridge__api__rust_api_connection_state_impl(port, ptr, rust_vec_len, data_len),
28 => wire__crate__bridge__api__rust_api_disable_memories_backup_impl(port, ptr, rust_vec_len, data_len), 28 => wire__crate__bridge__api__rust_api_delete_account_impl(port, ptr, rust_vec_len, data_len),
29 => wire__crate__bridge__api__rust_api_download_done_impl(port, ptr, rust_vec_len, data_len), 29 => wire__crate__bridge__api__rust_api_delete_memory_impl(port, ptr, rust_vec_len, data_len),
30 => wire__crate__bridge__api__rust_api_download_media_impl(port, ptr, rust_vec_len, data_len), 30 => wire__crate__bridge__api__rust_api_disable_memories_backup_impl(port, ptr, rust_vec_len, data_len),
31 => wire__crate__bridge__api__rust_api_download_pending_media_impl(port, ptr, rust_vec_len, data_len), 31 => wire__crate__bridge__api__rust_api_download_done_impl(port, ptr, rust_vec_len, data_len),
32 => wire__crate__bridge__api__rust_api_establish_signal_session_impl(port, ptr, rust_vec_len, data_len), 32 => wire__crate__bridge__api__rust_api_download_media_impl(port, ptr, rust_vec_len, data_len),
33 => wire__crate__bridge__api__rust_api_events_impl(port, ptr, rust_vec_len, data_len), 33 => wire__crate__bridge__api__rust_api_download_pending_media_impl(port, ptr, rust_vec_len, data_len),
34 => wire__crate__bridge__api__rust_api_force_ipa_check_impl(port, ptr, rust_vec_len, data_len), 34 => wire__crate__bridge__api__rust_api_establish_signal_session_impl(port, ptr, rust_vec_len, data_len),
35 => wire__crate__bridge__api__rust_api_get_memories_url_impl(port, ptr, rust_vec_len, data_len), 35 => wire__crate__bridge__api__rust_api_events_impl(port, ptr, rust_vec_len, data_len),
36 => wire__crate__bridge__api__rust_api_get_memories_usage_impl(port, ptr, rust_vec_len, data_len), 36 => wire__crate__bridge__api__rust_api_force_ipa_check_impl(port, ptr, rust_vec_len, data_len),
37 => wire__crate__bridge__api__rust_api_get_plan_balance_impl(port, ptr, rust_vec_len, data_len), 37 => wire__crate__bridge__api__rust_api_get_memories_url_impl(port, ptr, rust_vec_len, data_len),
38 => wire__crate__bridge__api__rust_api_get_proof_of_work_impl(port, ptr, rust_vec_len, data_len), 38 => wire__crate__bridge__api__rust_api_get_memories_usage_impl(port, ptr, rust_vec_len, data_len),
39 => wire__crate__bridge__api__rust_api_get_server_key_for_passwordless_recovery_impl(port, ptr, rust_vec_len, data_len), 39 => wire__crate__bridge__api__rust_api_get_plan_balance_impl(port, ptr, rust_vec_len, data_len),
40 => wire__crate__bridge__api__rust_api_get_user_by_id_impl(port, ptr, rust_vec_len, data_len), 40 => wire__crate__bridge__api__rust_api_get_proof_of_work_impl(port, ptr, rust_vec_len, data_len),
41 => wire__crate__bridge__api__rust_api_get_user_data_impl(port, ptr, rust_vec_len, data_len), 41 => wire__crate__bridge__api__rust_api_get_server_key_for_passwordless_recovery_impl(port, ptr, rust_vec_len, data_len),
42 => wire__crate__bridge__api__rust_api_get_user_id_from_username_impl(port, ptr, rust_vec_len, data_len), 42 => wire__crate__bridge__api__rust_api_get_user_by_id_impl(port, ptr, rust_vec_len, data_len),
43 => wire__crate__bridge__api__rust_api_insert_and_send_additional_data_impl(port, ptr, rust_vec_len, data_len), 43 => wire__crate__bridge__api__rust_api_get_user_data_impl(port, ptr, rust_vec_len, data_len),
44 => wire__crate__bridge__api__rust_api_insert_and_send_ask_about_user_impl(port, ptr, rust_vec_len, data_len), 44 => wire__crate__bridge__api__rust_api_get_user_id_from_username_impl(port, ptr, rust_vec_len, data_len),
45 => wire__crate__bridge__api__rust_api_insert_and_send_contact_share_impl(port, ptr, rust_vec_len, data_len), 45 => wire__crate__bridge__api__rust_api_insert_and_send_additional_data_impl(port, ptr, rust_vec_len, data_len),
46 => wire__crate__bridge__api__rust_api_insert_and_send_text_impl(port, ptr, rust_vec_len, data_len), 46 => wire__crate__bridge__api__rust_api_insert_and_send_ask_about_user_impl(port, ptr, rust_vec_len, data_len),
47 => wire__crate__bridge__api__rust_api_ipa_purchase_impl(port, ptr, rust_vec_len, data_len), 47 => wire__crate__bridge__api__rust_api_insert_and_send_contact_share_impl(port, ptr, rust_vec_len, data_len),
48 => wire__crate__bridge__api__rust_api_load_plan_balance_impl(port, ptr, rust_vec_len, data_len), 48 => wire__crate__bridge__api__rust_api_insert_and_send_text_impl(port, ptr, rust_vec_len, data_len),
49 => wire__crate__bridge__api__rust_api_notify_messages_opened_impl(port, ptr, rust_vec_len, data_len), 49 => wire__crate__bridge__api__rust_api_ipa_purchase_impl(port, ptr, rust_vec_len, data_len),
50 => wire__crate__bridge__api__rust_api_perform_passwordless_recovery_heartbeat_impl(port, ptr, rust_vec_len, data_len), 50 => wire__crate__bridge__api__rust_api_load_plan_balance_impl(port, ptr, rust_vec_len, data_len),
51 => wire__crate__bridge__api__rust_api_prepare_queued_message_impl(port, ptr, rust_vec_len, data_len), 51 => wire__crate__bridge__api__rust_api_notification_badge_count_impl(port, ptr, rust_vec_len, data_len),
52 => wire__crate__bridge__api__rust_api_register_impl(port, ptr, rust_vec_len, data_len), 52 => wire__crate__bridge__api__rust_api_notify_messages_opened_impl(port, ptr, rust_vec_len, data_len),
53 => wire__crate__bridge__api__rust_api_register_passwordless_notification_impl(port, ptr, rust_vec_len, data_len), 53 => wire__crate__bridge__api__rust_api_perform_passwordless_recovery_heartbeat_impl(port, ptr, rust_vec_len, data_len),
54 => wire__crate__bridge__api__rust_api_register_passwordless_recovery_impl(port, ptr, rust_vec_len, data_len), 54 => wire__crate__bridge__api__rust_api_prepare_queued_message_impl(port, ptr, rust_vec_len, data_len),
55 => wire__crate__bridge__api__rust_api_reload_configuration_impl(port, ptr, rust_vec_len, data_len), 55 => wire__crate__bridge__api__rust_api_register_impl(port, ptr, rust_vec_len, data_len),
56 => wire__crate__bridge__api__rust_api_remove_additional_user_impl(port, ptr, rust_vec_len, data_len), 56 => wire__crate__bridge__api__rust_api_register_passwordless_notification_impl(port, ptr, rust_vec_len, data_len),
57 => wire__crate__bridge__api__rust_api_report_user_impl(port, ptr, rust_vec_len, data_len), 57 => wire__crate__bridge__api__rust_api_register_passwordless_recovery_impl(port, ptr, rust_vec_len, data_len),
58 => wire__crate__bridge__api__rust_api_request_binary_impl(port, ptr, rust_vec_len, data_len), 58 => wire__crate__bridge__api__rust_api_reload_configuration_impl(port, ptr, rust_vec_len, data_len),
59 => wire__crate__bridge__api__rust_api_request_contact_by_username_impl(port, ptr, rust_vec_len, data_len), 59 => wire__crate__bridge__api__rust_api_remove_additional_user_impl(port, ptr, rust_vec_len, data_len),
60 => wire__crate__bridge__api__rust_api_request_media_reupload_impl(port, ptr, rust_vec_len, data_len), 60 => wire__crate__bridge__api__rust_api_report_user_impl(port, ptr, rust_vec_len, data_len),
61 => wire__crate__bridge__api__rust_api_request_memories_upload_impl(port, ptr, rust_vec_len, data_len), 61 => wire__crate__bridge__api__rust_api_request_binary_impl(port, ptr, rust_vec_len, data_len),
62 => wire__crate__bridge__api__rust_api_retransmit_all_messages_impl(port, ptr, rust_vec_len, data_len), 62 => wire__crate__bridge__api__rust_api_request_contact_by_username_impl(port, ptr, rust_vec_len, data_len),
63 => wire__crate__bridge__api__rust_api_send_binary_impl(port, ptr, rust_vec_len, data_len), 63 => wire__crate__bridge__api__rust_api_request_media_reupload_impl(port, ptr, rust_vec_len, data_len),
64 => wire__crate__bridge__api__rust_api_send_contact_profile_impl(port, ptr, rust_vec_len, data_len), 64 => wire__crate__bridge__api__rust_api_request_memories_upload_impl(port, ptr, rust_vec_len, data_len),
65 => wire__crate__bridge__api__rust_api_send_encrypted_content_impl(port, ptr, rust_vec_len, data_len), 65 => wire__crate__bridge__api__rust_api_retransmit_all_messages_impl(port, ptr, rust_vec_len, data_len),
66 => wire__crate__bridge__api__rust_api_send_encrypted_content_to_group_impl(port, ptr, rust_vec_len, data_len), 66 => wire__crate__bridge__api__rust_api_send_binary_impl(port, ptr, rust_vec_len, data_len),
67 => wire__crate__bridge__api__rust_api_send_queued_message_impl(port, ptr, rust_vec_len, data_len), 67 => wire__crate__bridge__api__rust_api_send_contact_profile_impl(port, ptr, rust_vec_len, data_len),
68 => wire__crate__bridge__api__rust_api_send_text_message_impl(port, ptr, rust_vec_len, data_len), 68 => wire__crate__bridge__api__rust_api_send_encrypted_content_impl(port, ptr, rust_vec_len, data_len),
69 => wire__crate__bridge__api__rust_api_send_typing_impl(port, ptr, rust_vec_len, data_len), 69 => wire__crate__bridge__api__rust_api_send_encrypted_content_to_group_impl(port, ptr, rust_vec_len, data_len),
70 => wire__crate__bridge__api__rust_api_set_background_impl(port, ptr, rust_vec_len, data_len), 70 => wire__crate__bridge__api__rust_api_send_queued_message_impl(port, ptr, rust_vec_len, data_len),
71 => wire__crate__bridge__api__rust_api_set_login_token_impl(port, ptr, rust_vec_len, data_len), 71 => wire__crate__bridge__api__rust_api_send_text_message_impl(port, ptr, rust_vec_len, data_len),
72 => wire__crate__bridge__api__rust_api_set_network_available_impl(port, ptr, rust_vec_len, data_len), 72 => wire__crate__bridge__api__rust_api_send_typing_impl(port, ptr, rust_vec_len, data_len),
73 => wire__crate__bridge__api__rust_api_submit_recovery_share_impl(port, ptr, rust_vec_len, data_len), 73 => wire__crate__bridge__api__rust_api_set_background_impl(port, ptr, rust_vec_len, data_len),
74 => wire__crate__bridge__api__rust_api_update_fcm_token_impl(port, ptr, rust_vec_len, data_len), 74 => wire__crate__bridge__api__rust_api_set_login_token_impl(port, ptr, rust_vec_len, data_len),
75 => wire__crate__bridge__api__rust_api_update_signed_pre_key_impl(port, ptr, rust_vec_len, data_len), 75 => wire__crate__bridge__api__rust_api_set_network_available_impl(port, ptr, rust_vec_len, data_len),
76 => wire__crate__bridge__api__rust_api_upload_pqc_pre_keys_impl(port, ptr, rust_vec_len, data_len), 76 => wire__crate__bridge__api__rust_api_submit_recovery_share_impl(port, ptr, rust_vec_len, data_len),
77 => wire__crate__bridge__wrapper__app_database__rust_app_database_changes_impl(port, ptr, rust_vec_len, data_len), 77 => wire__crate__bridge__api__rust_api_update_fcm_token_impl(port, ptr, rust_vec_len, data_len),
78 => wire__crate__bridge__wrapper__app_database__rust_app_database_execute_impl(port, ptr, rust_vec_len, data_len), 78 => wire__crate__bridge__api__rust_api_update_signed_pre_key_impl(port, ptr, rust_vec_len, data_len),
79 => wire__crate__bridge__wrapper__app_database__rust_app_database_legacy_import_complete_impl(port, ptr, rust_vec_len, data_len), 79 => wire__crate__bridge__api__rust_api_upload_pqc_pre_keys_impl(port, ptr, rust_vec_len, data_len),
80 => wire__crate__bridge__wrapper__app_database__rust_app_database_migrate_legacy_database_impl(port, ptr, rust_vec_len, data_len), 80 => wire__crate__bridge__wrapper__app_database__rust_app_database_changes_impl(port, ptr, rust_vec_len, data_len),
81 => wire__crate__bridge__wrapper__app_database__rust_app_database_select_impl(port, ptr, rust_vec_len, data_len), 81 => wire__crate__bridge__wrapper__app_database__rust_app_database_execute_impl(port, ptr, rust_vec_len, data_len),
82 => wire__crate__bridge__wrapper__backup__rust_backup_archive_create_backup_archive_impl(port, ptr, rust_vec_len, data_len), 82 => wire__crate__bridge__wrapper__app_database__rust_app_database_legacy_import_complete_impl(port, ptr, rust_vec_len, data_len),
83 => wire__crate__bridge__wrapper__backup__rust_backup_archive_get_backup_download_token_impl(port, ptr, rust_vec_len, data_len), 83 => wire__crate__bridge__wrapper__app_database__rust_app_database_migrate_legacy_database_impl(port, ptr, rust_vec_len, data_len),
84 => wire__crate__bridge__wrapper__backup__rust_backup_archive_restore_backup_archive_impl(port, ptr, rust_vec_len, data_len), 84 => wire__crate__bridge__wrapper__app_database__rust_app_database_select_impl(port, ptr, rust_vec_len, data_len),
85 => wire__crate__bridge__wrapper__backup__rust_backup_identity_get_backup_id_impl(port, ptr, rust_vec_len, data_len), 85 => wire__crate__bridge__wrapper__backup__rust_backup_archive_create_backup_archive_impl(port, ptr, rust_vec_len, data_len),
86 => wire__crate__bridge__wrapper__backup__rust_backup_identity_get_backup_password_keys_impl(port, ptr, rust_vec_len, data_len), 86 => wire__crate__bridge__wrapper__backup__rust_backup_archive_get_backup_download_token_impl(port, ptr, rust_vec_len, data_len),
87 => wire__crate__bridge__wrapper__backup__rust_backup_identity_get_identity_backup_bytes_impl(port, ptr, rust_vec_len, data_len), 87 => wire__crate__bridge__wrapper__backup__rust_backup_archive_restore_backup_archive_impl(port, ptr, rust_vec_len, data_len),
88 => wire__crate__bridge__wrapper__backup__rust_backup_identity_import_backup_password_keys_impl(port, ptr, rust_vec_len, data_len), 88 => wire__crate__bridge__wrapper__backup__rust_backup_identity_get_backup_id_impl(port, ptr, rust_vec_len, data_len),
89 => wire__crate__bridge__wrapper__backup__rust_backup_identity_restore_identity_backup_impl(port, ptr, rust_vec_len, data_len), 89 => wire__crate__bridge__wrapper__backup__rust_backup_identity_get_backup_password_keys_impl(port, ptr, rust_vec_len, data_len),
90 => wire__crate__bridge__wrapper__backup__rust_backup_identity_set_backup_password_keys_impl(port, ptr, rust_vec_len, data_len), 90 => wire__crate__bridge__wrapper__backup__rust_backup_identity_get_identity_backup_bytes_impl(port, ptr, rust_vec_len, data_len),
91 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_decrypt_cloud_media_key_impl(port, ptr, rust_vec_len, data_len), 91 => wire__crate__bridge__wrapper__backup__rust_backup_identity_import_backup_password_keys_impl(port, ptr, rust_vec_len, data_len),
92 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_encrypt_cloud_media_key_impl(port, ptr, rust_vec_len, data_len), 92 => wire__crate__bridge__wrapper__backup__rust_backup_identity_restore_identity_backup_impl(port, ptr, rust_vec_len, data_len),
93 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_get_login_token_impl(port, ptr, rust_vec_len, data_len), 93 => wire__crate__bridge__wrapper__backup__rust_backup_identity_set_backup_password_keys_impl(port, ptr, rust_vec_len, data_len),
94 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_get_signal_identity_impl(port, ptr, rust_vec_len, data_len), 94 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_decrypt_cloud_media_key_impl(port, ptr, rust_vec_len, data_len),
95 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_get_user_id_impl(port, ptr, rust_vec_len, data_len), 95 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_encrypt_cloud_media_key_impl(port, ptr, rust_vec_len, data_len),
96 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_import_serialized_impl(port, ptr, rust_vec_len, data_len), 96 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_get_login_token_impl(port, ptr, rust_vec_len, data_len),
97 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_import_signal_identity_impl(port, ptr, rust_vec_len, data_len), 97 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_get_signal_identity_impl(port, ptr, rust_vec_len, data_len),
98 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_remove_key_manager_impl(port, ptr, rust_vec_len, data_len), 98 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_get_user_id_impl(port, ptr, rust_vec_len, data_len),
99 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_serialize_impl(port, ptr, rust_vec_len, data_len), 99 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_import_serialized_impl(port, ptr, rust_vec_len, data_len),
100 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_set_user_id_impl(port, ptr, rust_vec_len, data_len), 100 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_import_signal_identity_impl(port, ptr, rust_vec_len, data_len),
101 => wire__crate__bridge__wrapper__signal__rust_signal_decrypt_impl(port, ptr, rust_vec_len, data_len), 101 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_remove_key_manager_impl(port, ptr, rust_vec_len, data_len),
102 => wire__crate__bridge__wrapper__signal__rust_signal_encrypt_impl(port, ptr, rust_vec_len, data_len), 102 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_serialize_impl(port, ptr, rust_vec_len, data_len),
103 => wire__crate__bridge__wrapper__signal__rust_signal_generate_bundle_impl(port, ptr, rust_vec_len, data_len), 103 => wire__crate__bridge__wrapper__key_manager__rust_key_manager_set_user_id_impl(port, ptr, rust_vec_len, data_len),
104 => wire__crate__bridge__wrapper__signal__rust_signal_generate_pqc_prekeys_impl(port, ptr, rust_vec_len, data_len), 104 => wire__crate__bridge__wrapper__signal__rust_signal_decrypt_impl(port, ptr, rust_vec_len, data_len),
105 => wire__crate__bridge__wrapper__signal__rust_signal_get_contact_public_key_impl(port, ptr, rust_vec_len, data_len), 105 => wire__crate__bridge__wrapper__signal__rust_signal_encrypt_impl(port, ptr, rust_vec_len, data_len),
106 => wire__crate__bridge__wrapper__signal__rust_signal_get_user_public_key_impl(port, ptr, rust_vec_len, data_len), 106 => wire__crate__bridge__wrapper__signal__rust_signal_generate_bundle_impl(port, ptr, rust_vec_len, data_len),
107 => wire__crate__bridge__wrapper__signal__rust_signal_process_prekey_bundle_impl(port, ptr, rust_vec_len, data_len), 107 => wire__crate__bridge__wrapper__signal__rust_signal_generate_pqc_prekeys_impl(port, ptr, rust_vec_len, data_len),
108 => wire__crate__bridge__wrapper__rust_utils_generate_shares_impl(port, ptr, rust_vec_len, data_len), 108 => wire__crate__bridge__wrapper__signal__rust_signal_get_contact_public_key_impl(port, ptr, rust_vec_len, data_len),
109 => wire__crate__bridge__wrapper__rust_utils_recover_secret_impl(port, ptr, rust_vec_len, data_len), 109 => wire__crate__bridge__wrapper__signal__rust_signal_get_user_public_key_impl(port, ptr, rust_vec_len, data_len),
110 => wire__crate__bridge__groups__update_chat_deletion_time_impl(port, ptr, rust_vec_len, data_len), 110 => wire__crate__bridge__wrapper__signal__rust_signal_process_prekey_bundle_impl(port, ptr, rust_vec_len, data_len),
111 => wire__crate__bridge__groups__update_group_name_impl(port, ptr, rust_vec_len, data_len), 111 => wire__crate__bridge__wrapper__rust_utils_generate_shares_impl(port, ptr, rust_vec_len, data_len),
113 => wire__crate__bridge__user_config__user_config_api_create_impl(port, ptr, rust_vec_len, data_len), 112 => wire__crate__bridge__wrapper__rust_utils_recover_secret_impl(port, ptr, rust_vec_len, data_len),
114 => wire__crate__bridge__user_config__user_config_api_import_json_impl(port, ptr, rust_vec_len, data_len), 113 => wire__crate__bridge__callbacks__log__strip_ansi_impl(port, ptr, rust_vec_len, data_len),
115 => wire__crate__bridge__user_config__user_config_api_load_impl(port, ptr, rust_vec_len, data_len), 114 => wire__crate__bridge__groups__update_chat_deletion_time_impl(port, ptr, rust_vec_len, data_len),
116 => wire__crate__bridge__user_config__user_config_api_save_impl(port, ptr, rust_vec_len, data_len), 115 => wire__crate__bridge__groups__update_group_name_impl(port, ptr, rust_vec_len, data_len),
117 => wire__crate__bridge__user_config__user_config_api_update_impl(port, ptr, rust_vec_len, data_len), 117 => wire__crate__bridge__user_config__user_config_api_create_impl(port, ptr, rust_vec_len, data_len),
118 => wire__crate__bridge__user_config__user_config_api_import_json_impl(port, ptr, rust_vec_len, data_len),
119 => wire__crate__bridge__user_config__user_config_api_load_impl(port, ptr, rust_vec_len, data_len),
120 => wire__crate__bridge__user_config__user_config_api_save_impl(port, ptr, rust_vec_len, data_len),
121 => wire__crate__bridge__user_config__user_config_api_update_impl(port, ptr, rust_vec_len, data_len),
_ => unreachable!(), _ => unreachable!(),
} }
} }
@ -5704,7 +5856,7 @@ fn pde_ffi_dispatcher_sync_impl(
// Codec=Pde (Serialization + dispatch), see doc to use other codecs // Codec=Pde (Serialization + dispatch), see doc to use other codecs
match func_id { match func_id {
18 => wire__crate__bridge__api__rust_api_api_base_url_impl(ptr, rust_vec_len, data_len), 18 => wire__crate__bridge__api__rust_api_api_base_url_impl(ptr, rust_vec_len, data_len),
112 => wire__crate__bridge__user_config__user_config_api_clone_impl( 116 => wire__crate__bridge__user_config__user_config_api_clone_impl(
ptr, ptr,
rust_vec_len, rust_vec_len,
data_len, data_len,

4
rust/src/services/contacts.rs
View file

@ -90,8 +90,10 @@ impl ContactService {
self.process_user_prekey_bundle(&user).await?; self.process_user_prekey_bundle(&user).await?;
let database = self.ctx.app_db.read().await.clone(); let database = self.ctx.app_db.read().await.clone();
// The server answered with a bundle, so a previous `UserIdNotFound`
// (or a manual mark) must not keep the contact blocked.
sqlx::query!( sqlx::query!(
"UPDATE contacts SET signal_version = 'v2' WHERE user_id = ?", "UPDATE contacts SET signal_version = 'v2', account_deleted = 0 WHERE user_id = ?",
user_id user_id
) )
.execute(&database.pool) .execute(&database.pool)

330
rust/src/services/groups/mod.rs
View file

@ -7,7 +7,9 @@ pub(crate) mod crypto;
pub(crate) mod model; pub(crate) mod model;
use crate::api::groups::GroupApi; use crate::api::groups::GroupApi;
use crate::api::messages::incoming::messages::queue_encrypted_content; use crate::api::messages::incoming::messages::{
queue_encrypted_content, retransmit_queued_receipts,
};
use crate::api::messages::outgoing::send_c2c_message_to_contact; use crate::api::messages::outgoing::send_c2c_message_to_contact;
use crate::api::proto::client::encrypted_content::GroupJoin; use crate::api::proto::client::encrypted_content::GroupJoin;
use crate::api::proto::client::{ use crate::api::proto::client::{
@ -28,19 +30,75 @@ use crate::utils::{current_time, new_uuid_v4};
use model::GroupRecord; use model::GroupRecord;
use prost::Message; use prost::Message;
use rand::{RngCore, SeedableRng}; use rand::{RngCore, SeedableRng};
use std::collections::BTreeSet; use std::collections::{BTreeSet, HashMap};
use std::sync::Arc; use std::sync::{Arc, LazyLock, Mutex};
use std::time::{Duration, Instant};
const DEFAULT_DELETE_MS: i64 = 86_400_000; const DEFAULT_DELETE_MS: i64 = 86_400_000;
/// How long to wait before asking the same member for its group public key
/// again. A peer that cannot answer -- an old client, a member who never comes
/// online -- would otherwise be re-asked on every reconnect and every group
/// update.
const PUBLIC_KEY_REQUEST_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60);
pub struct GroupService { pub struct GroupService {
ctx: Arc<Context>, ctx: Arc<Context>,
} }
/// Serializes group-state refreshes per group.
///
/// A refresh reads the server state, may repair it through `update_remote`, and
/// only then writes it locally. Two refreshes running at once read the same
/// `version_id`, so the second repair is rejected by the group server with 409.
/// The single app-database connection used to serialize this by accident, back
/// when the fetch ran inside the transaction; taking the network out of the
/// transaction is what makes an explicit lock necessary.
///
/// Acquire this *before* a database connection and never while holding one:
/// the reverse order deadlocks against the one-connection pool.
static GROUP_STATE_LOCKS: LazyLock<Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
fn group_state_lock(group_id: &str) -> Arc<tokio::sync::Mutex<()>> {
let mut locks = match GROUP_STATE_LOCKS.lock() {
Ok(locks) => locks,
Err(poisoned) => poisoned.into_inner(),
};
locks.entry(group_id.to_owned()).or_default().clone()
}
/// When each member was last asked to resend its group public key.
///
/// Process-local on purpose: the point is to keep one session from re-asking a
/// silent peer on every reconnect, not to remember the attempt across restarts.
/// `force` bypasses it, so the user can always retry by opening the group.
static PUBLIC_KEY_REQUESTS: LazyLock<Mutex<HashMap<(String, i64), Instant>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
/// Records a request and reports whether it should be sent at all.
fn claim_public_key_request(group_id: &str, contact_id: i64, force: bool) -> bool {
let mut requests = match PUBLIC_KEY_REQUESTS.lock() {
Ok(requests) => requests,
Err(poisoned) => poisoned.into_inner(),
};
let key = (group_id.to_owned(), contact_id);
let now = Instant::now();
if !force
&& requests
.get(&key)
.is_some_and(|sent| now.duration_since(*sent) < PUBLIC_KEY_REQUEST_INTERVAL)
{
return false;
}
requests.insert(key, now);
true
}
impl GroupService { impl GroupService {
pub async fn on_connected(&self) -> Result<()> { pub async fn on_connected(&self) -> Result<()> {
self.fetch_group_states_for_unjoined_groups().await?; self.fetch_group_states_for_unjoined_groups().await?;
self.fetch_missing_group_public_keys().await?; self.fetch_missing_group_public_keys(None, false).await?;
self.sync_flame_counters().await self.sync_flame_counters().await
} }
@ -101,14 +159,22 @@ impl GroupService {
Self { ctx: ctx.clone() } Self { ctx: ctx.clone() }
} }
/// Repairs a group after a peer reported that it does not know the group.
///
/// Owns its transaction instead of borrowing the inbound one: the refresh
/// below reaches the group server, which must not happen while the single
/// app-database connection is held.
pub(crate) async fn handle_membership_error( pub(crate) async fn handle_membership_error(
&self, &self,
t: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
from_user_id: i64, from_user_id: i64,
group_id: String, group_id: String,
related_receipt_id: String, related_receipt_id: String,
) -> Result<()> { ) -> Result<()> {
let _ = self.fetch_group_state_in_transaction(t, &group_id).await; let _ = self.fetch_group_state(group_id.clone()).await;
let database = self.ctx.app_db.read().await.clone();
let mut transaction = database.pool.begin().await?;
let t = &mut transaction;
let group = GroupRecord::load_in_transaction(t, &group_id).await?; let group = GroupRecord::load_in_transaction(t, &group_id).await?;
let is_still_member = sqlx::query_scalar!( let is_still_member = sqlx::query_scalar!(
@ -156,6 +222,9 @@ impl GroupService {
) )
.execute(&mut **t) .execute(&mut **t)
.await?; .await?;
transaction.commit().await?;
database.notify_committed(["receipts", "groups", "group_members"]);
Ok(()) Ok(())
} }
@ -229,24 +298,51 @@ impl GroupService {
Ok(true) Ok(true)
} }
/// Refreshes one group from the group server.
///
/// The round-trip happens before the transaction opens, under the group's
/// [`group_state_lock`]. The app database allows a single connection, so
/// holding it across network I/O starves every other database user until
/// the acquire timeout and surfaces as `pool timed out while waiting for an
/// open connection`.
pub async fn fetch_group_state(&self, group_id: String) -> Result<bool> { pub async fn fetch_group_state(&self, group_id: String) -> Result<bool> {
let lock = group_state_lock(&group_id);
let _guard = lock.lock().await;
let server = GroupApi::fetch_group_state(&group_id).await?;
let database = self.ctx.app_db.read().await.clone(); let database = self.ctx.app_db.read().await.clone();
let mut t = database.pool.begin().await?; let mut t = database.pool.begin().await?;
let updated = self let updated = self
.fetch_group_state_in_transaction(&mut t, &group_id) .apply_fetched_group_state(&mut t, &group_id, server)
.await?; .await?;
t.commit().await?; t.commit().await?;
database.notify_committed(["groups", "group_members", "contacts"]); database.notify_committed(["groups", "group_members", "contacts", "receipts"]);
// `apply_state` may have queued public-key requests. Nothing else
// flushes them here -- this is not an inbound-message path -- and they
// would otherwise wait for the next reconnect.
let ctx = self.ctx.clone();
tokio::spawn(async move {
if let Err(error) = retransmit_queued_receipts(&ctx).await {
tracing::warn!("failed to flush messages queued by a group refresh: {error}");
}
});
Ok(updated) Ok(updated)
} }
async fn fetch_group_state_in_transaction( /// Applies a group state that has already been fetched. Database work only,
/// apart from the admin-side `update_remote` repair below, which the
/// caller's group lock keeps free of races.
async fn apply_fetched_group_state(
&self, &self,
t: &mut sqlx::Transaction<'_, sqlx::Sqlite>, t: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
group_id: &str, group_id: &str,
server: Option<crate::api::proto::http_requests::GroupState>,
) -> Result<bool> { ) -> Result<bool> {
let group = GroupRecord::load_in_transaction(t, group_id).await?; let group = GroupRecord::load_in_transaction(t, group_id).await?;
let Some(server) = GroupApi::fetch_group_state(&group_id).await? else { let Some(server) = server else {
UpdateGroup::builder() UpdateGroup::builder()
.group_id(group_id.to_owned()) .group_id(group_id.to_owned())
.left_group(true) .left_group(true)
@ -352,6 +448,32 @@ impl GroupService {
self.fetch_group_state(group_id).await self.fetch_group_state(group_id).await
} }
/// Resolves the group public key a member signs its group-state appends
/// with. Our own key lives in the group record; everyone else's arrives
/// through `group_join` and may still be missing.
async fn member_public_key(
&self,
db: &Arc<crate::database::app::AppDatabase>,
group: &GroupRecord,
group_id: &str,
contact_id: i64,
) -> Result<Vec<u8>> {
if contact_id == self.ctx.user_id().await? {
return Ok(group.identity()?.identity_key().serialize().to_vec());
}
GetGroupPublicKey::builder()
.group_id(group_id.to_owned())
.contact_id(contact_id)
.build()
.fetch_pool(&db.pool)
.await?
.ok_or_else(|| {
TwonlyError::Generic(format!(
"group public key for contact {contact_id} not found"
))
})
}
pub async fn manage_admin( pub async fn manage_admin(
&self, &self,
group_id: String, group_id: String,
@ -361,17 +483,9 @@ impl GroupService {
let (db, g) = self.load_group(&group_id).await?; let (db, g) = self.load_group(&group_id).await?;
let (v, mut s) = GroupApi::load_state(&g).await?; let (v, mut s) = GroupApi::load_state(&g).await?;
let public_key = GetGroupPublicKey::builder() let public_key = self
.group_id(group_id.clone()) .member_public_key(&db, &g, &group_id, contact_id)
.contact_id(contact_id) .await?;
.build()
.fetch_pool(&db.pool)
.await?
.ok_or_else(|| {
TwonlyError::Generic(format!(
"group public key for contact {contact_id} not found"
))
})?;
if remove { if remove {
s.admin_ids.retain(|x| *x != contact_id) s.admin_ids.retain(|x| *x != contact_id)
@ -436,12 +550,13 @@ impl GroupService {
self.fetch_group_state(group_id).await self.fetch_group_state(group_id).await
} }
pub async fn remove_member( /// Removes a member from the group.
&self, ///
group_id: String, /// The member's group public key is only needed to revoke an admin's
public_key: Vec<u8>, /// signing rights, so a plain member can be removed without it. That
contact_id: i64, /// matters: a key that never arrived would otherwise make the member
) -> Result<bool> { /// unremovable.
pub async fn remove_member(&self, group_id: String, contact_id: i64) -> Result<bool> {
let (db, g) = self.load_group(&group_id).await?; let (db, g) = self.load_group(&group_id).await?;
let (v, mut s) = GroupApi::load_state(&g).await?; let (v, mut s) = GroupApi::load_state(&g).await?;
if !s.member_ids.contains(&contact_id) { if !s.member_ids.contains(&contact_id) {
@ -450,7 +565,15 @@ impl GroupService {
s.member_ids.retain(|x| *x != contact_id); s.member_ids.retain(|x| *x != contact_id);
let was_admin = s.admin_ids.contains(&contact_id); let was_admin = s.admin_ids.contains(&contact_id);
s.admin_ids.retain(|x| *x != contact_id); s.admin_ids.retain(|x| *x != contact_id);
GroupApi::update_remote(&g, v, &s, None, was_admin.then_some(public_key)).await?; let revoked_key = if was_admin {
Some(
self.member_public_key(&db, &g, &group_id, contact_id)
.await?,
)
} else {
None
};
GroupApi::update_remote(&g, v, &s, None, revoked_key).await?;
self.announce(&group_id, "removedMember", Some(contact_id), None, None) self.announce(&group_id, "removedMember", Some(contact_id), None, None)
.await?; .await?;
let mut tr = db.pool.begin().await?; let mut tr = db.pool.begin().await?;
@ -472,9 +595,7 @@ impl GroupService {
let user_id = self.ctx.user_id().await?; let user_id = self.ctx.user_id().await?;
let (_, group_state) = GroupApi::load_state(&group).await?; let (_, group_state) = GroupApi::load_state(&group).await?;
if group_state.admin_ids.contains(&user_id) { if group_state.admin_ids.contains(&user_id) {
let identity = group.identity()?; return self.remove_member(group_id, user_id).await;
let public_key = identity.identity_key().serialize().to_vec();
return self.remove_member(group_id, public_key, user_id).await;
} }
let identity = group.identity()?; let identity = group.identity()?;
@ -546,20 +667,61 @@ impl GroupService {
Ok(true) Ok(true)
} }
pub async fn refresh_group_state( /// Refreshes one group and then announces our own group public key to it.
&self, ///
tr: &mut sqlx::Transaction<'_, sqlx::Sqlite>, /// The order matters: `broadcast_group_public_key` resolves its recipients
group_id: String, /// from `group_members`, which holds nothing but the sender of the
created: bool, /// `group_create` until the refresh fills in the rest of the group.
) { /// Announcing first would reach that one member and leave every other
if created { /// member unable to promote or remove us. The announcement still runs when
let _ = self /// the refresh fails, so an offline group server costs reach, not delivery.
.fetch_group_states_for_unjoined_groups_in_transaction(tr) pub fn spawn_state_refresh_and_announce(ctx: &Arc<Context>, group_id: String) {
.await; let ctx = ctx.clone();
let _ = self.broadcast_group_public_key(tr, &group_id).await; tokio::spawn(async move {
} else { let service = Self::new(&ctx);
let _ = self.fetch_group_state_in_transaction(tr, &group_id).await; if let Err(error) = service.fetch_group_state(group_id.clone()).await {
} tracing::warn!(
group_id,
"group state refresh before announce failed: {error}"
);
}
if let Err(error) = service.announce_group_public_key(&group_id).await {
tracing::warn!(group_id, "group public key announcement failed: {error}");
}
});
}
/// Sends our group public key to every current member of `group_id`.
async fn announce_group_public_key(&self, group_id: &str) -> Result<()> {
let database = self.ctx.app_db.read().await.clone();
let mut transaction = database.pool.begin().await?;
self.broadcast_group_public_key(&mut transaction, group_id)
.await?;
transaction.commit().await?;
database.notify_committed(["receipts"]);
Ok(())
}
/// Schedules the server-side half of a group refresh.
///
/// Callers reach this from inside the inbound transaction, and a refresh
/// talks to the group server, so it cannot run there: it would pin the only
/// app-database connection across the network and, once it takes a group
/// lock, invert the lock order this module depends on. Running it detached
/// lets the caller commit first; every call site treats the refresh as best
/// effort already.
pub fn spawn_state_refresh(ctx: &Arc<Context>, group_id: Option<String>) {
let ctx = ctx.clone();
tokio::spawn(async move {
let service = Self::new(&ctx);
let result = match group_id {
Some(group_id) => service.fetch_group_state(group_id).await.map(|_| ()),
None => service.fetch_group_states_for_unjoined_groups().await,
};
if let Err(error) = result {
tracing::warn!("scheduled group state refresh failed: {error}");
}
});
} }
pub async fn broadcast_group_public_key( pub async fn broadcast_group_public_key(
@ -602,38 +764,47 @@ impl GroupService {
Ok(()) Ok(())
} }
/// Refreshes every group this client has not joined yet.
///
/// The ids are read on their own connection and each group is then
/// refreshed by [`Self::fetch_group_state`], so no connection is held
/// across the per-group round-trips.
pub async fn fetch_group_states_for_unjoined_groups(&self) -> Result<()> { pub async fn fetch_group_states_for_unjoined_groups(&self) -> Result<()> {
let db = self.ctx.app_db.read().await.clone(); let db = self.ctx.app_db.read().await.clone();
let mut t = db.pool.begin().await?;
self.fetch_group_states_for_unjoined_groups_in_transaction(&mut t)
.await?;
t.commit().await?;
Ok(())
}
pub async fn fetch_group_states_for_unjoined_groups_in_transaction(
&self,
t: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
) -> Result<()> {
let ids = GetUnjoinedGroups::builder() let ids = GetUnjoinedGroups::builder()
.build() .build()
.fetch_all(&mut **t) .fetch_all(&db.pool)
.await?; .await?;
for id in ids { for id in ids {
if let Err(e) = self.fetch_group_state_in_transaction(t, &id).await { if let Err(e) = self.fetch_group_state(id.clone()).await {
tracing::warn!(group_id = id, "group state refresh failed: {e}") tracing::warn!(group_id = id, "group state refresh failed: {e}")
} }
} }
Ok(()) Ok(())
} }
pub async fn fetch_missing_group_public_keys(&self) -> Result<()> { /// Asks every member whose group public key is still unknown to resend it.
///
/// Without the key an admin cannot promote or remove that member, because
/// the group server needs it to authorize the member's future signed
/// appends. Pass a `group_id` to limit the sweep to one group, and `force`
/// to ignore [`PUBLIC_KEY_REQUEST_INTERVAL`] -- the group view does, so the
/// user always has a way to retry by hand.
pub async fn fetch_missing_group_public_keys(
&self,
group_id: Option<String>,
force: bool,
) -> Result<()> {
let db = self.ctx.app_db.read().await.clone(); let db = self.ctx.app_db.read().await.clone();
let rows = GetMissingGroupPublicKeys::builder() let rows = GetMissingGroupPublicKeys::builder()
.maybe_group_id(group_id)
.build() .build()
.fetch_all(&db.pool) .fetch_all(&db.pool)
.await?; .await?;
for row in rows { for row in rows {
if !claim_public_key_request(&row.group_id, row.contact_id, force) {
continue;
}
send_c2c_message_to_contact() send_c2c_message_to_contact()
.ctx(&self.ctx) .ctx(&self.ctx)
.contact_id(row.contact_id) .contact_id(row.contact_id)
@ -651,6 +822,39 @@ impl GroupService {
Ok(()) Ok(())
} }
/// Same request, queued inside the caller's transaction.
///
/// [`Self::fetch_missing_group_public_keys`] sends immediately, which needs
/// its own database access and so cannot run while a transaction holds the
/// single app-database connection.
async fn queue_missing_group_public_key_requests(
t: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
group_id: &str,
) -> Result<()> {
let rows = GetMissingGroupPublicKeys::builder()
.group_id(group_id.to_owned())
.build()
.fetch_all_in_transaction(t)
.await?;
for row in rows {
if !claim_public_key_request(&row.group_id, row.contact_id, false) {
continue;
}
queue_encrypted_content(
t,
row.contact_id,
EncryptedContent {
group_id: Some(row.group_id),
resend_group_public_key: Some(encrypted_content::ResendGroupPublicKey {}),
..Default::default()
},
true,
)
.await?;
}
Ok(())
}
// --- Private helpers --- // --- Private helpers ---
async fn load_group( async fn load_group(
@ -783,6 +987,14 @@ impl GroupService {
.execute(t) .execute(t)
.await?; .await?;
} }
// The server state carries member ids but no per-member keys, and a
// member announces its own key only once, when it first learns the
// group exists -- so nobody announces themselves to a member who joined
// later. Asking here is what closes that gap for both sides: whoever
// refreshes first discovers the other with an empty key and requests it.
Self::queue_missing_group_public_key_requests(t, group_id).await?;
Ok(()) Ok(())
} }
} }

60
rust/src/services/notifications.rs
View file

@ -7,6 +7,7 @@ use crate::api::proto::client::{self as proto, encrypted_content};
use crate::api::runtime::ApiRuntime; use crate::api::runtime::ApiRuntime;
use crate::bridge::InitConfig; use crate::bridge::InitConfig;
use crate::context::{Context, RuntimeMode}; use crate::context::{Context, RuntimeMode};
use crate::database::app::AppDatabase;
use crate::error::Result; use crate::error::Result;
use crate::user_config::UserConfig; use crate::user_config::UserConfig;
use crate::utils::{current_time, milliseconds_to_seconds}; use crate::utils::{current_time, milliseconds_to_seconds};
@ -273,11 +274,10 @@ pub(crate) async fn record_incoming_event(
Ok(()) Ok(())
} }
pub async fn pending_batch(ctx: &Arc<Context>, locale: &str) -> Result<NotificationBatch> { /// A foreground chat can mark a message as opened before the native push
let database = ctx.app_db.read().await.clone(); /// worker gets around to rendering its durable outbox row. Clearing those
// A foreground chat can mark a message as opened before the native push /// stale rows keeps them from producing an alert or inflating the badge.
// worker gets around to rendering its durable outbox row. Clear those async fn clear_stale_opened(database: &Arc<AppDatabase>) -> Result<()> {
// stale rows first so they neither produce an alert nor inflate the badge.
let cleared_at = current_time().timestamp(); let cleared_at = current_time().timestamp();
let cleared = sqlx::query( let cleared = sqlx::query(
r#" r#"
@ -299,6 +299,26 @@ pub async fn pending_batch(ctx: &Arc<Context>, locale: &str) -> Result<Notificat
if cleared.rows_affected() != 0 { if cleared.rows_affected() != 0 {
database.notify_committed(["notification_outbox"]); database.notify_committed(["notification_outbox"]);
} }
Ok(())
}
/// The number of events the user has not dealt with yet. iOS has no way to
/// derive an app icon badge from the delivered alerts, so the running app has
/// to push this value into `UNUserNotificationCenter` itself.
pub async fn badge_count(ctx: &Arc<Context>) -> Result<i64> {
let database = ctx.app_db.read().await.clone();
clear_stale_opened(&database).await?;
let count = sqlx::query_scalar!(
r#"SELECT COUNT(*) AS "count: i64" FROM notification_outbox WHERE cleared_at IS NULL"#
)
.fetch_one(&database.pool)
.await?;
Ok(count)
}
pub async fn pending_batch(ctx: &Arc<Context>, locale: &str) -> Result<NotificationBatch> {
let database = ctx.app_db.read().await.clone();
clear_stale_opened(&database).await?;
let rows = sqlx::query_as!( let rows = sqlx::query_as!(
PendingRow, PendingRow,
r#" r#"
@ -487,6 +507,36 @@ pub async fn clear_conversation(ctx: &Arc<Context>, conversation_id: &str) -> Re
Ok(notification_ids) Ok(notification_ids)
} }
/// Clears the contact-request notifications. They carry no conversation, so
/// opening the request list is the only moment the user acknowledges them.
pub async fn clear_contact_requests(ctx: &Arc<Context>) -> Result<Vec<String>> {
let database = ctx.app_db.read().await.clone();
let mut transaction = database.pool.begin().await?;
let notification_ids = sqlx::query_scalar!(
r#"
SELECT notification_id FROM notification_outbox
WHERE kind IN ('contact_request', 'accept_request') AND cleared_at IS NULL
"#,
)
.fetch_all(&mut *transaction)
.await?;
let cleared_at = current_time().timestamp();
sqlx::query!(
r#"
UPDATE notification_outbox SET cleared_at = ?
WHERE kind IN ('contact_request', 'accept_request') AND cleared_at IS NULL
"#,
cleared_at,
)
.execute(&mut *transaction)
.await?;
transaction.commit().await?;
if !notification_ids.is_empty() {
database.notify_committed(["notification_outbox"]);
}
Ok(notification_ids)
}
/// Clears only notifications representing the messages that were actually /// Clears only notifications representing the messages that were actually
/// opened. Events that merely refer to the same message, such as reactions or /// opened. Events that merely refer to the same message, such as reactions or
/// media-status updates, remain independent notifications. /// media-status updates, remain independent notifications.

22
rust/tests/api.rs
View file

@ -389,22 +389,22 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> {
tracing::info!("Group chat deletion timer updated to 1 hour"); tracing::info!("Group chat deletion timer updated to 1 hour");
// 6. Promote tester_b to admin // 6. Promote tester_b to admin
// tester_a needs tester_b's public key to promote them. We simulate a message from tester_b // tester_a needs tester_b's public key to promote them. tester_b
// so that tester_a can request the missing public key. // announces it when it learns of the group, so this asserts the
// announcement arrived rather than repairing the state by hand.
{ {
let db_a = tester_a.context.app_db.read().await.clone(); let db_a = tester_a.context.app_db.read().await.clone();
sqlx::query!( let public_key = sqlx::query_scalar!(
"UPDATE group_members SET last_message = CAST(strftime('%s','now') AS INTEGER) WHERE group_id = ? AND contact_id = ?", "SELECT group_public_key FROM group_members WHERE group_id = ? AND contact_id = ?",
group_id, group_id,
tester_b.user_id tester_b.user_id
) )
.execute(&db_a.pool) .fetch_one(&db_a.pool)
.await?; .await?;
group_service_a.fetch_missing_group_public_keys().await?; assert!(
public_key.is_some(),
// Wait for tester_b to respond with the group join containing the public key "tester_a should have tester_b's group public key without asking for it"
// (We just wait a moment to let the messages exchange) );
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
} }
group_service_a group_service_a
@ -453,7 +453,7 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> {
// Note: tester_c is not an admin, so their public key is not needed to remove them. // Note: tester_c is not an admin, so their public key is not needed to remove them.
// We pass an empty vec![] instead of waiting for a key exchange. // We pass an empty vec![] instead of waiting for a key exchange.
group_service_a group_service_a
.remove_member(group_id.clone(), vec![], tester_c.user_id) .remove_member(group_id.clone(), tester_c.user_id)
.await?; .await?;
// tester_b should see tester_c removed // tester_b should see tester_c removed

13
rust/tests/api/group_resilience.rs
View file

@ -173,18 +173,11 @@ async fn test_admin_and_non_admin_leave_group() -> anyhow::Result<()> {
.wait_for_group_exists(&group_id, group_name) .wait_for_group_exists(&group_id, group_name)
.await?; .await?;
// Fetch missing public key for tester_b // Ask for anything the announcements did not deliver. Forced, so the
// per-member request interval cannot skip it.
{ {
let db_a = tester_a.context.app_db.read().await.clone();
sqlx::query!(
"UPDATE group_members SET last_message = CAST(strftime('%s','now') AS INTEGER) WHERE group_id = ? AND contact_id = ?",
group_id,
tester_b.user_id
)
.execute(&db_a.pool)
.await?;
GroupService::new(&tester_a.context) GroupService::new(&tester_a.context)
.fetch_missing_group_public_keys() .fetch_missing_group_public_keys(Some(group_id.clone()), true)
.await?; .await?;
tokio::time::sleep(std::time::Duration::from_millis(500)).await; tokio::time::sleep(std::time::Duration::from_millis(500)).await;
} }

28
rust/tests/api/notifications.rs
View file

@ -440,6 +440,34 @@ async fn test_notification_outbox_end_to_end() -> anyhow::Result<()> {
.is_empty()); .is_empty());
} }
//
// The badge the running app pushes into iOS is the same number the native
// extension renders, and contact requests are cleared by looking at the
// request list rather than by opening a conversation.
//
{
let before = tester_b.notification_batch("en").await?;
assert_eq!(
tester_b.notification_badge_count().await?,
before.badge_count
);
let removed = tester_b.clear_contact_request_notifications().await?;
assert!(
!removed.is_empty(),
"the pending contact request must be reported for withdrawal"
);
assert_eq!(
tester_b.notification_badge_count().await?,
before.badge_count - removed.len() as i64,
"acknowledging contact requests must reduce the badge"
);
assert!(tester_b
.clear_contact_request_notifications()
.await?
.is_empty());
}
// //
// A blocked contact is still decrypted and committed, but must never // A blocked contact is still decrypted and committed, but must never
// produce a notification. // produce a notification.

9
rust/tests/api/tester.rs
View file

@ -588,6 +588,15 @@ impl Tester {
Ok(notifications::clear_conversation(&self.context, conversation_id).await?) Ok(notifications::clear_conversation(&self.context, conversation_id).await?)
} }
pub async fn clear_contact_request_notifications(&self) -> anyhow::Result<Vec<String>> {
Ok(notifications::clear_contact_requests(&self.context).await?)
}
/// The value the running app pushes into the iOS app icon badge.
pub async fn notification_badge_count(&self) -> anyhow::Result<i64> {
Ok(notifications::badge_count(&self.context).await?)
}
/// Number of rows the outbox holds for one receipt, used to prove that a /// Number of rows the outbox holds for one receipt, used to prove that a
/// redelivered envelope cannot notify twice. /// redelivered envelope cannot notify twice.
pub async fn notification_rows_for_event(&self, event_id: &str) -> anyhow::Result<i64> { pub async fn notification_rows_for_event(&self, event_id: &str) -> anyhow::Result<i64> {

163
scripts/run_emulators.sh Executable file
View file

@ -0,0 +1,163 @@
#!/usr/bin/env bash
#
# Start `flutter run` for every currently running iOS/Android emulator,
# each in its own zellij tab of a fresh session, so the logs stay separate.
#
# Usage: scripts/run_emulators.sh [-s SESSION] [-w SECONDS] [-f] [-- <extra flutter run args>]
#
# -s SESSION zellij session name (default: twonly-run)
# -w SECONDS stagger between device starts, avoids build-dir races (default: 15)
# -f kill an existing session with the same name instead of aborting
#
# Everything after `--` is appended to each `flutter run` invocation,
# e.g. `scripts/run_emulators.sh -- --flavor dev --dart-define=FOO=bar`
set -euo pipefail
SESSION="twonly-run"
STAGGER=15
FORCE=0
while getopts ":s:w:fh" opt; do
case "$opt" in
s) SESSION="$OPTARG" ;;
w) STAGGER="$OPTARG" ;;
f) FORCE=1 ;;
h) sed -n '2,15p' "$0"; exit 0 ;;
\?) echo "unknown option: -$OPTARG" >&2; exit 2 ;;
:) echo "option -$OPTARG needs an argument" >&2; exit 2 ;;
esac
done
shift $((OPTIND - 1))
case "$STAGGER" in
''|*[!0-9]*) echo "-w needs a non-negative whole number of seconds" >&2; exit 2 ;;
esac
for bin in zellij flutter jq; do
command -v "$bin" >/dev/null || { echo "missing required tool: $bin" >&2; exit 1; }
done
PROJECT_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
ZELLIJ_CONFIG="${ZELLIJ_CONFIG_FILE:-${XDG_CONFIG_HOME:-$HOME/.config}/zellij/config.kdl}"
if zellij list-sessions -s 2>/dev/null | grep -qx "$SESSION"; then
if [ "$FORCE" -eq 1 ]; then
echo "deleting existing session '$SESSION'..."
zellij delete-session --force "$SESSION" >/dev/null
else
echo "zellij session '$SESSION' already exists. Attach with:" >&2
echo " zellij attach $SESSION" >&2
echo "or re-run with -f to replace it." >&2
exit 1
fi
fi
echo "querying flutter devices..."
DEVICES_JSON="$(flutter devices --machine)"
# Only running emulators/simulators on the ios or android platforms.
# bash 3.2 (macOS default) has no mapfile, so read the lines manually.
DEVICES=()
while IFS= read -r line; do
[ -n "$line" ] && DEVICES[${#DEVICES[@]}]="$line"
done < <(
printf '%s' "$DEVICES_JSON" | jq -r '
.[]
| select(.emulator == true and .isSupported == true)
| select(.targetPlatform | test("^(ios|android)"))
| "\(.id)\t\(.name)\t\(.targetPlatform)"
'
)
if [ "${#DEVICES[@]}" -eq 0 ]; then
echo "no running iOS or Android emulators found." >&2
echo "start one first, e.g. 'flutter emulators --launch <id>' or open Simulator.app." >&2
exit 1
fi
RUNDIR="$(mktemp -d "${TMPDIR:-/tmp}/twonly-run-XXXXXX")"
LAYOUT="$RUNDIR/layout.kdl"
# A custom layout replaces zellij's default UI, so re-declare the default tab
# template -- without it the tab bar and status bar are gone and the tabs,
# although they exist, are invisible and unswitchable-looking.
if grep -qE '^[[:space:]]*default_layout[[:space:]]+"compact"' "$ZELLIJ_CONFIG" 2>/dev/null; then
TAB_TEMPLATE=' default_tab_template {
children
pane size=1 borderless=true {
plugin location="zellij:compact-bar"
}
}'
else
TAB_TEMPLATE=' default_tab_template {
pane size=1 borderless=true {
plugin location="zellij:tab-bar"
}
children
pane size=2 borderless=true {
plugin location="zellij:status-bar"
}
}'
fi
{
echo 'layout {'
echo "$TAB_TEMPLATE"
} > "$LAYOUT"
index=0
USED_NAMES=()
for entry in "${DEVICES[@]}"; do
id="${entry%% *}"
rest="${entry#* }"
name="${rest%% *}"
platform="${rest#* }"
# Two clones of the same emulator image share a name, so identify Android
# tabs by their (unique) adb id and fall back to a counter elsewhere.
case "$platform" in
ios*) label="ios-$name" ;;
*) label="android-$id" ;;
esac
# tab names must survive KDL quoting; keep them boring
tab_name="$(printf '%s' "$label" | tr -c 'A-Za-z0-9._-' '-' | cut -c1-30)"
suffix=2
while printf '%s\n' ${USED_NAMES+"${USED_NAMES[@]}"} | grep -qx "$tab_name"; do
tab_name="$(printf '%s' "$label" | tr -c 'A-Za-z0-9._-' '-' | cut -c1-27)-$suffix"
suffix=$((suffix + 1))
done
USED_NAMES[${#USED_NAMES[@]}]="$tab_name"
wrapper="$RUNDIR/run-$index.sh"
{
echo '#!/usr/bin/env bash'
printf 'cd %q\n' "$PROJECT_ROOT"
if [ "$index" -gt 0 ] && [ "$STAGGER" -gt 0 ]; then
delay=$((index * STAGGER))
printf 'echo "waiting %ss so the builds do not fight over the build directory..."\n' "$delay"
printf 'sleep %s\n' "$delay"
fi
printf 'echo "=== %s (%s) ==="\n' "$name" "$id"
printf 'exec flutter run -d %q' "$id"
for arg in "$@"; do printf ' %q' "$arg"; done
printf '\n'
} > "$wrapper"
chmod +x "$wrapper"
{
printf ' tab name="%s" {\n' "$tab_name"
printf ' pane command="bash" {\n'
printf ' args "%s"\n' "$wrapper"
printf ' }\n'
printf ' }\n'
} >> "$LAYOUT"
echo " tab '$tab_name' -> $id"
index=$((index + 1))
done
echo '}' >> "$LAYOUT"
echo "starting zellij session '$SESSION' with $index tab(s)..."
exec zellij --session "$SESSION" --new-session-with-layout "$LAYOUT"