add c2c testing

lib/src/services/api.service.dart

@@ -164,6 +164,9 @@ class ApiService {
   }
 
   Future<void> onClosed() async {
+    if (kDebugMode) {
+      print('API onClosed called');
+    }
     if (_channel == null) return;
     Log.info('websocket connection closed');
     _channel = null;
@@ -251,11 +254,17 @@ class ApiService {
   bool get isConnected => _channel != null && _channel!.closeCode == null;
 
   Future<void> _onDone() async {
+    if (kDebugMode) {
+      print('API _onDone called');
+    }
     _reconnectionDelay = 3;
     await onClosed();
   }
 
   Future<void> _onError(dynamic e) async {
+    if (kDebugMode) {
+      print('API _onError called: $e');
+    }
     if (e.toString().contains('Failed host lookup')) {
       Log.info('WebSocket connection failed: Host not reachable.');
     } else {
@@ -265,8 +274,14 @@ class ApiService {
   }
 
   Future<void> _onData(dynamic msgBuffer) async {
+    if (kDebugMode) {
+      print('API _onData received: $msgBuffer');
+    }
     try {
-      final msg = server.ServerToClient.fromBuffer(msgBuffer as Uint8List);
+      if (msgBuffer is! Uint8List) {
+        msgBuffer = Uint8List.fromList(msgBuffer as List<int>);
+      }
+      final msg = server.ServerToClient.fromBuffer(msgBuffer);
       if (msg.v0.hasResponse()) {
         final completer = _pendingRequests.remove(msg.v0.seq);
         if (completer != null && !completer.isCompleted) {

test/features/link_parser_test.dart

@@ -102,6 +102,7 @@ void main() {
       ),
       LinkParserTest(
         title: 'twonly Public Launch',
+        siteName: 'twonly',
         desc:
             'After about a year of development, twonly is finally ready for its public launch.',
         url: 'https://twonly.eu/en/blog/2026-public-launch.html',

test/mocks/platform_channels.dart

@@ -0,0 +1,85 @@
+// ignore_for_file: avoid_print, avoid_dynamic_calls
+
+import 'dart:async';
+import 'package:flutter/services.dart';
+import 'package:flutter_test/flutter_test.dart';
+
+void setupPlatformChannelMocks() {
+  final secureStorageMock = <String, String>{};
+  Future<dynamic> mockHandler(MethodCall methodCall) async {
+    final userId = Zone.current[#userId] as int?;
+    final keyPrefix = userId != null ? '${userId}_' : '';
+    print(
+      'DEBUG: mockHandler: method=${methodCall.method}, key=${methodCall.arguments?['key']}, userId=$userId, prefix=$keyPrefix',
+    );
+    if (methodCall.method == 'read') {
+      final key = methodCall.arguments['key'] as String;
+      return secureStorageMock[keyPrefix + key];
+    } else if (methodCall.method == 'write') {
+      final key = methodCall.arguments['key'] as String;
+      final value = methodCall.arguments['value'] as String;
+      secureStorageMock[keyPrefix + key] = value;
+      return true;
+    } else if (methodCall.method == 'delete') {
+      final key = methodCall.arguments['key'] as String;
+      secureStorageMock.remove(keyPrefix + key);
+      return true;
+    } else if (methodCall.method == 'readAll') {
+      final result = <String, String>{};
+      secureStorageMock.forEach((k, v) {
+        if (k.startsWith(keyPrefix)) {
+          result[k.substring(keyPrefix.length)] = v;
+        }
+      });
+      return result;
+    } else if (methodCall.method == 'deleteAll') {
+      if (userId != null) {
+        secureStorageMock.removeWhere((k, v) => k.startsWith(keyPrefix));
+      } else {
+        secureStorageMock.clear();
+      }
+      return true;
+    }
+    return null;
+  }
+
+  TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger
+      .setMockMethodCallHandler(
+        const MethodChannel(
+          'plugins.it_crowd.double_tapp/flutter_secure_storage',
+        ),
+        mockHandler,
+      );
+  TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger
+      .setMockMethodCallHandler(
+        const MethodChannel('plugins.it_nomads.com/flutter_secure_storage'),
+        mockHandler,
+      );
+  TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger
+      .setMockMethodCallHandler(
+        const MethodChannel('dev.fluttercommunity.plus/connectivity'),
+        (call) async {
+          if (call.method == 'check') {
+            return ['wifi'];
+          }
+          return null;
+        },
+      );
+  TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger
+      .setMockMethodCallHandler(
+        const MethodChannel(
+          'be.tramesch.workmanager/foreground_channel_workmanager',
+        ),
+        (call) async => true,
+      );
+  TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger
+      .setMockMethodCallHandler(
+        const MethodChannel('com.bbflight.background_downloader'),
+        (call) async {
+          if (call.method == 'enqueue') {
+            return true;
+          }
+          return null;
+        },
+      );
+}

test/mocks/test_client.dart

@@ -0,0 +1,300 @@
+// ignore_for_file: avoid_dynamic_calls
+
+import 'dart:async';
+import 'dart:io';
+
+import 'package:collection/collection.dart';
+import 'package:drift/drift.dart' hide isNotNull, isNull;
+import 'package:fixnum/fixnum.dart';
+import 'package:twonly/src/constants/secure_storage.keys.dart';
+import 'package:twonly/src/database/tables/messages.table.dart';
+import 'package:twonly/src/database/twonly.db.dart';
+import 'package:twonly/src/model/json/userdata.model.dart';
+import 'package:twonly/src/model/protobuf/client/generated/messages.pb.dart' as pb;
+import 'package:twonly/src/model/protobuf/client/generated/push_notification.pb.dart';
+import 'package:twonly/src/services/api.service.dart';
+import 'package:twonly/src/services/api/messages.api.dart';
+import 'package:twonly/src/services/notifications/pushkeys.notifications.dart';
+import 'package:twonly/src/services/signal/identity.signal.dart';
+import 'package:twonly/src/services/signal/session.signal.dart';
+import 'package:twonly/src/services/user.service.dart';
+import 'package:twonly/src/utils/log.dart';
+import 'package:twonly/src/utils/pow.dart';
+
+import 'user_environment.dart';
+
+class RealHttpOverrides extends HttpOverrides {
+  @override
+  HttpClient createHttpClient(SecurityContext? context) {
+    return super.createHttpClient(context)
+      ..badCertificateCallback = (cert, host, port) {
+        return true;
+      };
+  }
+}
+
+class TestClient {
+  TestClient(this.localIdSeed) {
+    final timeStr = DateTime.now().millisecondsSinceEpoch.toString();
+    username = 't_${timeStr.substring(timeStr.length - 6)}$localIdSeed';
+  }
+  late final UserEnvironment env;
+  late final ApiService api;
+  final int localIdSeed;
+  late String username;
+  Group? defaultGroup;
+  int realUserId = 0;
+
+  Future<void> init() async {
+    env = await UserEnvironment.create(localIdSeed, username);
+    api = ApiService();
+
+    await run(() async {
+      await createIfNotExistsSignalIdentity();
+
+      Log.info('Connecting to API...');
+      final connected = await api.connect();
+      Log.info('Connected: $connected');
+      if (!connected) throw Exception('Failed to connect to API');
+
+      Log.info('Requesting POW...');
+      final powRes = await api.getProofOfWork();
+      Log.info('POW result: $powRes');
+      if (powRes.$1 == null) throw Exception('Failed to get POW');
+
+      final prefix = powRes.$1!.prefix;
+      final difficulty = powRes.$1!.difficulty.toInt();
+      final proof = await calculatePoW(prefix, difficulty);
+
+      final regRes = await api.register(username, '', proof);
+      if (regRes.isError) {
+        throw Exception('Registration failed: ${regRes.error}');
+      }
+
+      realUserId = regRes.value.userid.toInt() as int;
+
+      final userData = UserData(
+        userId: realUserId,
+        username: username,
+        displayName: username,
+        subscriptionPlan: 'Free',
+        currentSetupPage: null,
+      );
+      // ignore: cascade_invocations
+      userData.appVersion = 100;
+      await UserService.save(userData);
+
+      await api.authenticate();
+      await signalGetPreKeys();
+    });
+  }
+
+  Future<void> initContact(TestClient other) async {
+    await run(() async {
+      await env.db.contactsDao.insertContact(
+        ContactsCompanion.insert(
+          userId: Value(other.realUserId),
+          username: other.username,
+          accepted: const Value(true),
+        ),
+      );
+      defaultGroup = await env.db.groupsDao.createNewDirectChat(
+        other.realUserId,
+        GroupsCompanion(groupName: Value(other.username)),
+      );
+
+      final dummyPushKeys = [
+        PushUser()
+          ..userId = Int64(other.realUserId)
+          ..pushKeys.add(
+            PushKey()
+              ..key = Uint8List(32)
+              ..id = Int64(12345)
+              ..createdAtUnixTimestamp = Int64(
+                DateTime.now().millisecondsSinceEpoch,
+              ),
+          ),
+        PushUser()
+          ..userId = Int64(realUserId)
+          ..pushKeys.add(
+            PushKey()
+              ..key = Uint8List(32)
+              ..id = Int64(67890)
+              ..createdAtUnixTimestamp = Int64(
+                DateTime.now().millisecondsSinceEpoch,
+              ),
+          ),
+      ];
+      await setPushKeys(SecureStorageKeys.sendingPushKeys, dummyPushKeys);
+      await setPushKeys(SecureStorageKeys.receivingPushKeys, dummyPushKeys);
+
+      final userData = await api.getUserById(other.realUserId);
+      final sessionStarted = await processSignalUserData(userData!);
+      if (!sessionStarted) throw Exception('Failed to start session');
+    });
+  }
+
+  Future<T> run<T>(Future<T> Function() computation) {
+    return runInZone(env, api, computation);
+  }
+
+  Future<Message> sendText(TestClient target, String text) async {
+    return run(() async {
+      final m = await env.db.messagesDao.insertMessage(
+        MessagesCompanion(
+          groupId: Value(defaultGroup!.groupId),
+          content: Value(text),
+          type: Value(MessageType.text.name),
+        ),
+      );
+      await sendCipherText(
+        target.realUserId,
+        pb.EncryptedContent(
+          groupId: defaultGroup!.groupId,
+          textMessage: pb.EncryptedContent_TextMessage()
+            ..senderMessageId = m!.messageId
+            ..text = text,
+        ),
+        messageId: m.messageId,
+      );
+      return m;
+    });
+  }
+
+  Future<void> sendEncryptedContent(
+    TestClient target,
+    pb.EncryptedContent content,
+  ) async {
+    await run(() async {
+      await sendCipherText(target.realUserId, content);
+    });
+  }
+
+  Future<void> sendGroupJoin(
+    TestClient target,
+    String groupId,
+    List<int> groupPublicKey,
+  ) async {
+    final content = pb.EncryptedContent()
+      ..groupId = groupId
+      ..groupJoin = (pb.EncryptedContent_GroupJoin()
+        ..groupPublicKey = groupPublicKey);
+    await sendEncryptedContent(target, content);
+  }
+
+  Future<void> sendResendGroupPublicKey(
+    TestClient target,
+    String groupId,
+  ) async {
+    final content = pb.EncryptedContent()
+      ..groupId = groupId
+      ..resendGroupPublicKey = pb.EncryptedContent_ResendGroupPublicKey();
+    await sendEncryptedContent(target, content);
+  }
+
+  Future<void> sendErrorMessages(
+    TestClient target,
+    pb.EncryptedContent_ErrorMessages_Type type,
+    String receiptId,
+  ) async {
+    final content = pb.EncryptedContent()
+      ..errorMessages = (pb.EncryptedContent_ErrorMessages()
+        ..type = type
+        ..relatedReceiptId = receiptId);
+    await sendEncryptedContent(target, content);
+  }
+
+  Future<void> sendUserDiscoveryRequest(
+    TestClient target,
+    List<int> version,
+  ) async {
+    final content = pb.EncryptedContent()
+      ..userDiscoveryRequest = (pb.EncryptedContent_UserDiscoveryRequest()
+        ..currentVersion = version);
+    await sendEncryptedContent(target, content);
+  }
+
+  Future<void> sendUserDiscoveryUpdate(
+    TestClient target,
+    List<List<int>> messages,
+  ) async {
+    final content = pb.EncryptedContent()
+      ..userDiscoveryUpdate = (pb.EncryptedContent_UserDiscoveryUpdate()
+        ..messages.addAll(messages));
+    await sendEncryptedContent(target, content);
+  }
+
+  Future<void> sendKeyVerificationProof(
+    TestClient target,
+    List<int> mac,
+  ) async {
+    final content = pb.EncryptedContent()
+      ..keyVerificationProof = (pb.EncryptedContent_KeyVerificationProof()
+        ..calculatedMac = mac);
+    await sendEncryptedContent(target, content);
+  }
+
+  Future<void> sendDeliveryReceipt(TestClient target, String receiptId) async {
+    final msg = pb.Message()
+      ..type = pb.Message_Type.SENDER_DELIVERY_RECEIPT
+      ..receiptId = receiptId;
+    await api.sendTextMessage(target.realUserId, msg.writeToBuffer(), null);
+  }
+
+  Future<void> sendReaction(TestClient target, String targetMessageId, String emoji, {bool remove = false}) async {
+    final content = pb.EncryptedContent()
+      ..groupId = defaultGroup!.groupId
+      ..reaction = (pb.EncryptedContent_Reaction()
+        ..targetMessageId = targetMessageId
+        ..emoji = emoji
+        ..remove = remove);
+    await sendEncryptedContent(target, content);
+  }
+
+  Future<void> sendMessageUpdate(TestClient target, String targetMessageId, pb.EncryptedContent_MessageUpdate_Type type, {String? text}) async {
+    final update = pb.EncryptedContent_MessageUpdate()
+      ..type = type
+      ..senderMessageId = targetMessageId
+      ..timestamp = Int64(DateTime.now().millisecondsSinceEpoch);
+    if (text != null) update.text = text;
+    final content = pb.EncryptedContent()
+      ..groupId = defaultGroup!.groupId
+      ..messageUpdate = update;
+    await sendEncryptedContent(target, content);
+  }
+
+  Future<void> sendMedia(TestClient target, String senderMessageId, pb.EncryptedContent_Media_Type type) async {
+    final content = pb.EncryptedContent()
+      ..groupId = defaultGroup!.groupId
+      ..media = (pb.EncryptedContent_Media()
+        ..senderMessageId = senderMessageId
+        ..type = type
+        ..requiresAuthentication = false
+        ..timestamp = Int64(DateTime.now().millisecondsSinceEpoch));
+    await sendEncryptedContent(target, content);
+  }
+
+  Future<Message> expectMessage(bool Function(Message) predicate) async {
+    for (var i = 0; i < 500; i++) {
+      final msg = await run(() async {
+        final msgs = await env.db.select(env.db.messages).get();
+        return msgs.firstWhereOrNull(predicate);
+      });
+      if (msg != null) return msg;
+      await Future.delayed(const Duration(milliseconds: 10));
+    }
+    throw Exception('Message matching predicate not received');
+  }
+
+  Future<Reaction> expectReaction(String messageId, String emoji) async {
+    for (var i = 0; i < 500; i++) {
+      final reaction = await run(() async {
+         final reactions = await (env.db.select(env.db.reactions)..where((t) => t.messageId.equals(messageId))).get();
+         return reactions.firstWhereOrNull((r) => r.emoji == emoji);
+      });
+      if (reaction != null) return reaction;
+      await Future.delayed(const Duration(milliseconds: 10));
+    }
+    throw Exception('Reaction $emoji not received on message $messageId');
+  }
+}

test/mocks/user_environment.dart

@@ -0,0 +1,124 @@
+import 'dart:async';
+import 'dart:io';
+
+import 'package:drift/drift.dart' hide isNotNull, isNull;
+import 'package:drift/native.dart';
+import 'package:libsignal_protocol_dart/libsignal_protocol_dart.dart';
+import 'package:twonly/src/database/twonly.db.dart';
+import 'package:twonly/src/model/json/userdata.model.dart';
+import 'package:twonly/src/services/api.service.dart';
+import 'package:twonly/src/services/user.service.dart';
+import 'package:twonly/src/utils/keyvalue.dart';
+
+base class ZoneIOOverrides extends IOOverrides {
+  @override
+  File createFile(String path) {
+    final userId = Zone.current[#userId] as int?;
+    if (userId != null) {
+      final newPath = _rewritePath(path, userId);
+      return super.createFile(newPath);
+    }
+    return super.createFile(path);
+  }
+
+  @override
+  Directory createDirectory(String path) {
+    final userId = Zone.current[#userId] as int?;
+    if (userId != null) {
+      final newPath = _rewritePath(path, userId);
+      return super.createDirectory(newPath);
+    }
+    return super.createDirectory(path);
+  }
+
+  String _rewritePath(String path, int userId) {
+    if (path.contains('/user_$userId') || path.contains('/$userId')) {
+      return path;
+    }
+    if (path.contains('/keyvalue/')) {
+      return path.replaceFirst('/keyvalue/', '/keyvalue/user_$userId/');
+    }
+    return path;
+  }
+}
+
+Future<T> runInZone<T>(
+  UserEnvironment env,
+  ApiService api,
+  Future<T> Function() computation,
+) {
+  return IOOverrides.runWithIOOverrides(
+    () => runZoned(
+      computation,
+      zoneValues: {
+        #twonlyDB: env.db,
+        #userService: env.userService,
+        #apiService: api,
+        #userId: env.userId,
+      },
+    ),
+    ZoneIOOverrides(),
+  );
+}
+
+class UserEnvironment {
+  UserEnvironment({
+    required this.userId,
+    required this.username,
+    required this.db,
+    required this.userService,
+    required this.identityKeyPair,
+    required this.registrationId,
+  });
+  final int userId;
+  final String username;
+  final TwonlyDB db;
+  final UserService userService;
+  final IdentityKeyPair identityKeyPair;
+  final int registrationId;
+
+  static Future<UserEnvironment> create(
+    int userId,
+    String username,
+  ) async {
+    final db = TwonlyDB.forTesting(
+      DatabaseConnection(
+        NativeDatabase.memory(),
+        closeStreamsSynchronously: true,
+      ),
+    );
+
+    final us = UserService();
+    // ignore: cascade_invocations
+    us.currentUser = UserData(
+      userId: userId,
+      username: username,
+      displayName: '$username Display',
+      subscriptionPlan: 'Free',
+      currentSetupPage: null,
+    )..appVersion = 100;
+
+    // ignore: cascade_invocations
+    us.isUserCreated = true;
+
+    final identityKeyPair = generateIdentityKeyPair();
+    final registrationId = generateRegistrationId(true);
+
+    // Save to keyvalue store using zone so it is isolated per-user
+    await runZoned(
+      () => KeyValueStore.put('user', us.currentUser.toJson()),
+      zoneValues: {
+        #userId: userId,
+      },
+    );
+
+    return UserEnvironment(
+      userId: userId,
+      username: username,
+      db: db,
+      userService: us,
+      identityKeyPair: identityKeyPair,
+      registrationId: registrationId,
+    );
+  }
+}

test/mocks/workmanager.dart

@@ -0,0 +1,29 @@
+import 'package:workmanager_platform_interface/workmanager_platform_interface.dart';
+
+class MockWorkmanagerPlatform extends WorkmanagerPlatform {
+  @override
+  Future<void> initialize(
+    Function callbackDispatcher, {
+    bool isInDebugMode = false,
+  }) async {}
+
+  @override
+  Future<void> registerOneOffTask(
+    String uniqueName,
+    String taskName, {
+    Map<String, dynamic>? inputData,
+    Duration? initialDelay,
+    Constraints? constraints,
+    ExistingWorkPolicy? existingWorkPolicy,
+    BackoffPolicy? backoffPolicy,
+    Duration? backoffPolicyDelay,
+    String? tag,
+    OutOfQuotaPolicy? outOfQuotaPolicy,
+  }) async {}
+
+  @override
+  Future<void> cancelByUniqueName(String uniqueName) async {}
+
+  @override
+  dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
+}

test/services/api_and_c2c_test.dart

@@ -0,0 +1,242 @@
+import 'dart:async';
+import 'dart:io';
+
+import 'package:drift/drift.dart' hide isNotNull, isNull;
+import 'package:flutter/services.dart';
+import 'package:flutter_rust_bridge/flutter_rust_bridge_for_generated.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:twonly/core/bridge.dart' as bridge;
+import 'package:twonly/core/frb_generated.dart';
+import 'package:twonly/globals.dart';
+import 'package:twonly/locator.dart';
+import 'package:twonly/src/callbacks/callbacks.dart';
+import 'package:twonly/src/database/tables/messages.table.dart';
+import 'package:twonly/src/database/twonly.db.dart';
+import 'package:twonly/src/model/protobuf/client/generated/messages.pb.dart'
+    as pb;
+import 'package:twonly/src/services/api.service.dart';
+import 'package:twonly/src/services/user.service.dart';
+import 'package:workmanager/workmanager.dart';
+
+import '../mocks/platform_channels.dart';
+import '../mocks/test_client.dart';
+import '../mocks/workmanager.dart';
+
+void main() {
+  if (!Platform.isMacOS) {
+    return;
+  }
+
+  TestWidgetsFlutterBinding.ensureInitialized();
+
+  late Directory tempDir;
+
+  setUpAll(() async {
+    // Log.init();
+    driftRuntimeOptions.dontWarnAboutMultipleDatabases = true;
+    WorkmanagerPlatform.instance = MockWorkmanagerPlatform();
+    tempDir = Directory.systemTemp.createTempSync('twonly_messaging_test_');
+    AppEnvironment.initTesting(
+      customCacheDir: tempDir.path,
+      customSupportDir: tempDir.path,
+    );
+
+    final dylibPath =
+        '${Directory.current.path}/rust/target/debug/librust_lib_twonly.dylib';
+    if (File(dylibPath).existsSync()) {
+      await RustLib.init(externalLibrary: ExternalLibrary.open(dylibPath));
+    } else {
+      await RustLib.init();
+    }
+    await initFlutterCallbacksForRust();
+
+    await bridge.initializeTwonlyFlutter(
+      config: bridge.InitConfig(
+        databaseDir: tempDir.path,
+        dataDir: tempDir.path,
+      ),
+    );
+
+    if (locator.isRegistered<TwonlyDB>()) await locator.unregister<TwonlyDB>();
+    if (locator.isRegistered<UserService>()) {
+      await locator.unregister<UserService>();
+    }
+    if (locator.isRegistered<ApiService>()) {
+      await locator.unregister<ApiService>();
+    }
+
+    locator
+      ..registerFactory<TwonlyDB>(() {
+        final db = Zone.current[#twonlyDB] as TwonlyDB?;
+        if (db != null) return db;
+        throw StateError('No TwonlyDB in active Zone.');
+      })
+      ..registerFactory<UserService>(() {
+        final us = Zone.current[#userService] as UserService?;
+        if (us != null) return us;
+        throw StateError('No UserService in active Zone.');
+      })
+      ..registerFactory<ApiService>(() {
+        final api = Zone.current[#apiService] as ApiService?;
+        if (api != null) return api;
+        throw StateError('No ApiService in active Zone.');
+      });
+  });
+
+  tearDownAll(() async {
+    if (tempDir.existsSync()) {
+      try {
+        tempDir.deleteSync(recursive: true);
+      } catch (_) {}
+    }
+  });
+
+  group('C2C Messaging Protocol - Real API Stress Tests', () {
+    late TestClient clientA;
+    late TestClient clientB;
+
+    setUp(() async {
+      setupPlatformChannelMocks();
+      HttpOverrides.global = RealHttpOverrides();
+      TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger
+          .setMockMethodCallHandler(
+            const MethodChannel('dev.fluttercommunity.plus/package_info'),
+            (call) async {
+              return {
+                'appName': 'twonly',
+                'packageName': 'eu.twonly.app',
+                'version': '1.0.0',
+                'buildNumber': '100',
+              };
+            },
+          );
+      await Workmanager().initialize(() {});
+
+      clientA = TestClient(1001);
+      clientB = TestClient(2002);
+
+      await clientA.init();
+      await clientB.init();
+
+      await clientA.initContact(clientB);
+      await clientB.initContact(clientA);
+    });
+
+    tearDown(() async {
+      await clientA.run(() async => clientA.api.close(null));
+      await clientB.run(() async => clientB.api.close(null));
+      await clientA.env.db.close();
+      await clientB.env.db.close();
+    });
+
+    test('C2C: Text Message Send & Receive', () async {
+      const text = 'Hello User B!';
+      await clientA.sendText(clientB, text);
+
+      final receivedMsg = await clientB.expectMessage(
+        (m) => m.content == text && m.senderId == clientA.realUserId,
+      );
+
+      expect(receivedMsg.content, text);
+      expect(receivedMsg.senderId, clientA.realUserId);
+      expect(receivedMsg.type, MessageType.text.name);
+    });
+
+    test('C2C: GroupJoin & ResendGroupPublicKey', () async {
+      // clientA creates a fake group ID
+      const groupId = 'test-group-id';
+
+      // clientA asks clientB for public key
+      await clientA.sendResendGroupPublicKey(clientB, groupId);
+
+      // We expect clientB to process it without crashing.
+      // Wait for it to settle by sending a text message and awaiting it.
+      await clientA.sendText(clientB, 'sync1');
+      await clientB.expectMessage((m) => m.content == 'sync1');
+
+      // clientA sends GroupJoin
+      await clientA.sendGroupJoin(clientB, groupId, [1, 2, 3]);
+
+      await clientA.sendText(clientB, 'sync2');
+      await clientB.expectMessage((m) => m.content == 'sync2');
+    });
+
+    test('C2C: ErrorMessages', () async {
+      await clientA.sendErrorMessages(
+        clientB,
+        pb.EncryptedContent_ErrorMessages_Type.SESSION_OUT_OF_SYNC,
+        'fake-receipt',
+      );
+      await clientA.sendText(clientB, 'sync3');
+      await clientB.expectMessage((m) => m.content == 'sync3');
+    });
+
+    test('C2C: UserDiscoveryRequest & Update', () async {
+      await clientA.sendUserDiscoveryRequest(clientB, [1]);
+      await clientA.sendUserDiscoveryUpdate(clientB, [
+        [1, 2],
+      ]);
+      await clientA.sendText(clientB, 'sync4');
+      await clientB.expectMessage((m) => m.content == 'sync4');
+    });
+
+    test('C2C: KeyVerificationProof', () async {
+      await clientA.sendKeyVerificationProof(clientB, [0, 0, 0]);
+      await clientA.sendText(clientB, 'sync5');
+      await clientB.expectMessage((m) => m.content == 'sync5');
+    });
+
+    test('C2C: SENDER_DELIVERY_RECEIPT', () async {
+      await clientA.sendDeliveryReceipt(clientB, 'fake-receipt-999');
+      await clientA.sendText(clientB, 'sync6');
+      await clientB.expectMessage((m) => m.content == 'sync6');
+    });
+
+    test('C2C: Reaction Send & Receive', () async {
+      final msgA = await clientA.sendText(clientB, 'Message to react to');
+      await clientB.expectMessage((m) => m.content == 'Message to react to');
+
+      await clientB.sendReaction(clientA, msgA.messageId, '👍');
+      final receivedReaction = await clientA.expectReaction(
+        msgA.messageId,
+        '👍',
+      );
+      expect(receivedReaction.emoji, '👍');
+    });
+
+    test('C2C: MessageUpdate (Edit & Delete)', () async {
+      final msgA = await clientA.sendText(clientB, 'Message to edit');
+      await clientB.expectMessage((m) => m.content == 'Message to edit');
+
+      await clientA.sendMessageUpdate(
+        clientB,
+        msgA.messageId,
+        pb.EncryptedContent_MessageUpdate_Type.EDIT_TEXT,
+        text: 'Edited text',
+      );
+
+      final editedMsg = await clientB.expectMessage(
+        (m) => m.content == 'Edited text' && m.messageId == msgA.messageId,
+      );
+      expect(editedMsg.content, 'Edited text');
+
+      await clientA.sendMessageUpdate(
+        clientB,
+        msgA.messageId,
+        pb.EncryptedContent_MessageUpdate_Type.DELETE,
+      );
+      await clientA.sendText(clientB, 'sync7');
+      await clientB.expectMessage((m) => m.content == 'sync7');
+    });
+
+    test('C2C: Media Message', () async {
+      await clientA.sendMedia(
+        clientB,
+        'media-msg-id',
+        pb.EncryptedContent_Media_Type.IMAGE,
+      );
+      await clientA.sendText(clientB, 'sync8');
+      await clientB.expectMessage((m) => m.content == 'sync8');
+    });
+  });
+}