retrieveMessages method

Future<List<SFMessage>> retrieveMessages({ PeerId? targetPeerId, PeerId? fromServer, String? folderPath, int? fromSequence, int? maxMessages, MessagePriority? minPriority, bool throwOnFailure = false, })

Retrieve pending messages from S&F server

targetPeerId Optional: retrieve from another peer's public mailbox (default: own mailbox) fromServer Optional: specify which server to retrieve from folderPath Which folder to retrieve from (default: 'inbox') fromSequence Cursor position (for multi-reader mailboxes) maxMessages Maximum number of messages to retrieve minPriority Minimum priority level to retrieve throwOnFailure When true, a retrieval that fails (no server, a timeout, a broken connection) throws RetrieveFailedException instead of returning an empty list, so the caller can tell "the mailbox is empty" from "the mailbox could not be read" and try again.

Implementation

Future<List<SFMessage>> retrieveMessages({
  PeerId? targetPeerId,  // NEW: retrieve from another peer's public mailbox
  PeerId? fromServer,
  String? folderPath,
  int? fromSequence,
  int? maxMessages,
  MessagePriority? minPriority,
  bool throwOnFailure = false,
}) async {
  final effectivePeerId = targetPeerId ?? host.id;
  final isOwnMailbox = effectivePeerId == host.id;

  _logger.info('📥 [SFClient] retrieveMessages called');
  _logger.info('📥 [SFClient] Target peer: ${effectivePeerId.toBase58()}');
  _logger.info('📥 [SFClient] Folder: ${folderPath ?? 'inbox'}');
  _logger.info('📥 [SFClient] Own mailbox: $isOwnMailbox');
  _logger.info('📥 [SFClient] FromSeq: $fromSequence, MaxMsgs: $maxMessages');

  // Determine which server to query
  _logger.info('📥 [SFClient] Selecting server...');
  final serverId = fromServer ??
      await _serverSelector.selectServer(config.preferredServers);

  if (serverId == null) {
    _logger.warning('❌ [SFClient] No S&F server available for retrieval');
    if (throwOnFailure) throw const RetrieveFailedException('No S&F server available');
    return [];
  }

  _logger.info('✅ [SFClient] Server selected: ${serverId.toBase58()}');

  P2PStream? stream;  // Declare outside try
  try {
    // Open stream to server using MAA (Mail Access Agent) protocol
    _logger.info('🌐 [SFClient] Opening stream to server (MAA protocol)...');
    final context = Context();
    stream = await host.newStream(
      serverId,
      [AccessHandler.protocolId],
      context,
    ).timeout(config.connectionTimeout);

    _logger.info('✅ [SFClient] Stream opened successfully');
    _logger.info('📤 [SFClient] Sending retrieve request to server...');

    // Retrieve messages via MAA (pass targetPeerId to protocol handler)
    final response = await AccessHandler.retrieveMessages(
      stream,
      effectivePeerId,  // Request messages for this peer (could be us or another peer)
      folderPath: folderPath,
      fromSequence: fromSequence,
      maxMessages: maxMessages,
      minPriority: minPriority,
    ).timeout(config.messageTimeout);

    _logger.info('✅ [SFClient] Retrieved ${response.messages.length} messages from ${folderPath ?? 'inbox'}');

    // Protect the S&F server connection so it won't be closed by cleanup
    host.connManager.protect(serverId, 'ricochet-sf-server');
    _logger.fine('Protected connection to S&F server ${serverId.toBase58().substring(0, 12)}...');

    for (int i = 0; i < response.messages.length; i++) {
      final msg = response.messages[i];
      _logger.info('✅ [SFClient] Message $i: ID=${msg.messageId.substring(0, 8)}, Folder=${msg.folderPath ?? "null"}');
    }

    // Open sealed payloads against the envelope each message arrived in,
    // so a ciphertext the server moved or relabelled is refused.
    final messages = <SFMessage>[];
    for (final message in response.messages) {
      try {
        messages.add(await openIfEncrypted(message, encryptor));
      } catch (e) {
        // A payload that does not open is returned sealed rather than
        // dropped: the caller still sees the message, and can delete it or
        // pass it on. Failing the loop instead would cost the caller every
        // other message in the batch, so one unopenable ciphertext (from a
        // sender using the wrong key, say) would stall the mailbox.
        _logger.warning(
            'message ${message.messageId} did not open, returning it sealed: $e');
        messages.add(message);
      }
    }

    // Emit to incoming messages stream
    for (final message in messages) {
      _incomingMessagesController.add(message);
    }

    return messages;

  } on TimeoutException {
    _logger.warning('Retrieval timed out');
    if (throwOnFailure) throw const RetrieveFailedException('Retrieval timed out');
    return [];
  } on RetrieveRefusedException catch (e) {
    // Not an empty mailbox: the caller is told so, rather than finding
    // nothing and concluding nothing was ever published there.
    _logger.warning('Retrieve refused: $e');
    rethrow;
  } catch (e) {
    _logger.warning('Failed to retrieve messages: $e');
    if (throwOnFailure) throw RetrieveFailedException('$e');
    return [];
  } finally {
    // Always close the stream after use
    if (stream != null && !stream.isClosed) {
      try {
        await stream.close();
      } catch (e) {
        _logger.fine('Error closing stream: $e');
      }
    }
  }
}