im_core/messages/

service.rs

pub struct MessageService<'a> {
    client: &'a crate::core::ImClient,
}

#[cfg(test)]
mod direct_send_result_identity_tests {
    use crate::internal::local_state::owner_scope::{
        direct_conversation_id_for_peer_scope, DirectPeerScope,
    };

    #[test]
    fn direct_send_result_normalization_uses_peer_scope_thread() {
        let scope = DirectPeerScope::new("user-bob", "Bob.AWiki.Test").expect("valid peer scope");
        let expected_thread_id = direct_conversation_id_for_peer_scope(&scope);
        let mut result = send_result("msg-peer-scope");

        super::normalize_direct_send_result_for_peer_scope(
            &mut result,
            Some(&scope),
            Some("bob.awiki.test"),
            Some("did:example:bob-current"),
        )
        .expect("normalize");

        match &result.message.thread {
            crate::messages::ThreadRef::Thread(thread) => {
                assert_eq!(thread.as_str(), expected_thread_id);
            }
            other => panic!("expected thread ref, got {other:?}"),
        }
        assert_eq!(
            attribute(&result, "peer_user_id").as_deref(),
            Some("user-bob")
        );
        assert_eq!(
            attribute(&result, "peer_full_handle").as_deref(),
            Some("bob.awiki.test")
        );
        assert_eq!(
            attribute(&result, "target_handle").as_deref(),
            Some("bob.awiki.test")
        );
        assert_eq!(
            attribute(&result, "resolved_target_did").as_deref(),
            Some("did:example:bob-current")
        );
        assert_eq!(
            attribute(&result, "peer_current_did").as_deref(),
            Some("did:example:bob-current")
        );
        assert_eq!(
            result.message.receiver.as_ref().map(|peer| peer.as_str()),
            Some("bob.awiki.test")
        );
    }

    #[test]
    fn direct_send_result_normalization_without_peer_scope_is_noop() {
        let mut result = send_result("msg-direct-did");

        super::normalize_direct_send_result_for_peer_scope(
            &mut result,
            None,
            Some("bob.awiki.test"),
            Some("did:example:bob"),
        )
        .expect("normalize");

        match &result.message.thread {
            crate::messages::ThreadRef::Direct(peer) => {
                assert_eq!(peer.as_str(), "did:example:bob");
            }
            other => panic!("expected direct ref, got {other:?}"),
        }
        assert!(attribute(&result, "peer_user_id").is_none());
    }

    fn send_result(message_id: &str) -> crate::messages::SendMessageResult {
        crate::messages::SendMessageResult {
            message: crate::messages::Message {
                id: crate::ids::MessageId::parse(message_id).unwrap(),
                thread: crate::messages::ThreadRef::Direct(
                    crate::ids::PeerRef::parse("did:example:bob", "").unwrap(),
                ),
                direction: crate::messages::MessageDirection::Outgoing,
                sender: crate::ids::PeerRef::parse("did:example:alice", "").unwrap(),
                receiver: Some(crate::ids::PeerRef::parse("bob.awiki.test", "").unwrap()),
                group: None,
                body: crate::messages::MessageBodyView::Text {
                    text: "hello".to_owned(),
                    kind: crate::messages::MessageKind::Text,
                },
                sent_at: Some("2026-07-04T00:00:00Z".to_owned()),
                received_at: None,
                metadata: crate::messages::MessageMetadata::default(),
            },
            delivery: crate::messages::DeliveryState::Accepted,
            warnings: Vec::new(),
        }
    }

    fn attribute(result: &crate::messages::SendMessageResult, key: &str) -> Option<String> {
        result
            .message
            .metadata
            .attributes
            .iter()
            .find(|attribute| attribute.key == key)
            .map(|attribute| attribute.value.clone())
    }
}

#[cfg(test)]
mod stale_delivery_target_tests {
    use serde_json::json;

    #[test]
    fn stale_delivery_target_from_error_extracts_current_binding() {
        let err = crate::ImError::Service {
            status_code: None,
            code: Some("1406".to_owned()),
            message: "stale did".to_owned(),
            data: Some(json!({
                "reason": "stale_did",
                "target_did": "did:example:bob-old",
                "current_did": "did:example:bob-current",
                "full_handle": "bob.awiki.test",
                "user_id": "user-bob"
            })),
        };

        let stale = super::stale_delivery_target_from_error(&err, "did:example:alice")
            .expect("stale target");

        assert_eq!(
            stale.current_did.as_deref(),
            Some("did:example:bob-current")
        );
        assert_eq!(stale.full_handle.as_deref(), Some("bob.awiki.test"));
        assert_eq!(stale.user_id.as_deref(), Some("user-bob"));
    }

    #[test]
    fn stale_delivery_target_from_error_ignores_plain_invalid_binding() {
        let err = crate::ImError::Service {
            status_code: None,
            code: Some("1406".to_owned()),
            message: "invalid target".to_owned(),
            data: Some(json!({
                "reason": "proof_mismatch",
                "current_did": "did:example:bob-current"
            })),
        };

        assert!(super::stale_delivery_target_from_error(&err, "did:example:alice").is_none());
    }
}

#[cfg(all(test, feature = "sqlite"))]
mod conversation_mark_read_request_tests {
    use serde_json::json;
    use std::fs;
    use std::path::PathBuf;

    #[test]
    fn mark_conversation_read_maps_direct_storage_and_remote_thread_separately() {
        let fixture = Fixture::new("mark-read-direct");
        let client = fixture.client();
        let request = crate::messages::MarkConversationReadRequest {
            conversation: crate::messages::ConversationReadRef::new("dm:did:example:bob").unwrap(),
            watermark: None,
            fallback_max_message_ids: Some(25),
        };

        let mapped = super::mark_read_input_for_conversation(&client, request).unwrap();

        match mapped.request.thread {
            crate::messages::ThreadRef::Thread(thread) => {
                assert_eq!(thread.as_str(), "dm:did:example:bob");
            }
            other => panic!("expected storage thread ref, got {other:?}"),
        }
        match mapped.remote_thread.expect("remote thread") {
            crate::messages::ThreadRef::Direct(peer) => {
                assert_eq!(peer.as_str(), "did:example:bob");
            }
            other => panic!("expected remote direct ref, got {other:?}"),
        }
        assert!(mapped.request.watermark.is_none());
        assert_eq!(mapped.request.fallback_max_message_ids, Some(25));
    }

    #[test]
    fn mark_conversation_read_maps_peer_scope_to_remote_direct_from_projection() {
        let fixture = Fixture::new("mark-read-peer-scope");
        let client = fixture.client();
        let peer_scope = crate::internal::local_state::owner_scope::DirectPeerScope::new(
            "user-bob",
            "bob.awiki.test",
        )
        .unwrap();
        let conversation_id =
            crate::internal::local_state::owner_scope::direct_conversation_id_for_peer_scope(
                &peer_scope,
            );
        fixture.seed_message(crate::internal::local_state::messages::MessageRecord {
            msg_id: "msg-peer-scope-read".to_owned(),
            conversation_id: conversation_id.clone(),
            thread_id: conversation_id.clone(),
            sender_did: "did:example:bob-current".to_owned(),
            receiver_did: "did:example:alice".to_owned(),
            metadata: json!({
                "peer_user_id": "user-bob",
                "peer_full_handle": "bob.awiki.test",
                "peer_current_did": "did:example:bob-current"
            })
            .to_string(),
            ..Fixture::message_record_defaults()
        });
        let request = crate::messages::MarkConversationReadRequest {
            conversation: crate::messages::ConversationReadRef::new(conversation_id.clone())
                .unwrap(),
            watermark: Some(crate::messages::ReadWatermark {
                last_read_message_id: None,
                last_read_thread_seq: Some("42".to_owned()),
                read_at: None,
            }),
            fallback_max_message_ids: None,
        };

        let mapped = super::mark_read_input_for_conversation(&client, request).unwrap();

        match mapped.request.thread {
            crate::messages::ThreadRef::Thread(thread) => {
                assert_eq!(thread.as_str(), conversation_id);
            }
            other => panic!("expected storage thread ref, got {other:?}"),
        }
        match mapped.remote_thread.expect("remote thread") {
            crate::messages::ThreadRef::Direct(peer) => {
                assert_eq!(peer.as_str(), "did:example:bob-current");
            }
            other => panic!("expected remote direct ref, got {other:?}"),
        }
        assert_eq!(
            mapped
                .request
                .watermark
                .as_ref()
                .and_then(|watermark| watermark.last_read_thread_seq.as_deref()),
            Some("42")
        );
    }

    #[test]
    fn mark_conversation_read_maps_group_storage_and_remote_thread_separately() {
        let fixture = Fixture::new("mark-read-group");
        let client = fixture.client();
        let request = crate::messages::MarkConversationReadRequest {
            conversation: crate::messages::ConversationReadRef::new("group:did:example:group")
                .unwrap(),
            watermark: None,
            fallback_max_message_ids: None,
        };

        let mapped = super::mark_read_input_for_conversation(&client, request).unwrap();

        match mapped.request.thread {
            crate::messages::ThreadRef::Thread(thread) => {
                assert_eq!(thread.as_str(), "group:did:example:group");
            }
            other => panic!("expected group storage thread ref, got {other:?}"),
        }
        match mapped.remote_thread.expect("remote thread") {
            crate::messages::ThreadRef::Group(group) => {
                assert_eq!(group.as_str(), "did:example:group");
            }
            other => panic!("expected remote group ref, got {other:?}"),
        }
    }

    struct Fixture {
        root: PathBuf,
    }

    impl Fixture {
        fn new(prefix: &str) -> Self {
            let nanos = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos();
            let root = std::env::temp_dir().join(format!(
                "im-core-mark-read-{prefix}-{}-{nanos}",
                std::process::id()
            ));
            let identity_root = root.join("identities");
            let identity_dir = identity_root.join("alice");
            fs::create_dir_all(&identity_dir).unwrap();
            fs::write(identity_root.join("default"), "alice\n").unwrap();
            fs::write(
                identity_root.join("registry.json"),
                r#"{
                  "default_identity": "alice",
                  "identities": [{
                    "id": "alice-id",
                    "did": "did:example:alice",
                    "local_alias": "alice",
                    "ready_for_auth": true,
                    "ready_for_messaging": true,
                    "device_id": "device-alice",
                    "missing": []
                  }]
                }"#,
            )
            .unwrap();
            Self { root }
        }

        fn client(&self) -> crate::core::ImClient {
            crate::core::ImCore::new(
                crate::ImCoreConfig {
                    service_base_url: crate::ServiceEndpoint::parse("https://example.test")
                        .unwrap(),
                    did_domain: "awiki.test".to_owned(),
                    user_service_endpoint: None,
                    message_service_endpoint: None,
                    mail_service_endpoint: None,
                    anp_service_endpoint: None,
                    anp_service_did: None,
                    ca_bundle: None,
                    transport_policy: crate::MessageTransportPolicy::HttpOnly,
                },
                crate::ImCorePaths {
                    identities: crate::paths::IdentityRegistryPaths {
                        identity_root_dir: self.root.join("identities"),
                        registry_path: self.root.join("identities").join("registry.json"),
                        default_identity_path: Some(self.root.join("identities").join("default")),
                    },
                    local_state: crate::paths::LocalStatePaths {
                        sqlite_path: self.root.join("local").join("im.sqlite"),
                    },
                    runtime: crate::paths::RuntimePaths {
                        cache_dir: self.root.join("cache"),
                        temp_dir: self.root.join("tmp"),
                    },
                },
            )
            .unwrap()
            .client(crate::identity::IdentitySelector::LocalAlias(
                "alice".to_owned(),
            ))
            .unwrap()
        }

        fn seed_message(&self, record: crate::internal::local_state::messages::MessageRecord) {
            let connection = crate::internal::local_state::open_writable(
                &self.root.join("local").join("im.sqlite"),
            )
            .unwrap();
            crate::internal::local_state::messages::upsert_message(&connection, &record).unwrap();
        }

        fn message_record_defaults() -> crate::internal::local_state::messages::MessageRecord {
            crate::internal::local_state::messages::MessageRecord {
                owner_identity_id: "alice-id".to_owned(),
                owner_did: "did:example:alice".to_owned(),
                direction: 0,
                content_type: "text/plain".to_owned(),
                content: "hello from projection".to_owned(),
                sent_at: "2026-07-05T00:00:00Z".to_owned(),
                stored_at: "2026-07-05T00:00:00Z".to_owned(),
                ..crate::internal::local_state::messages::MessageRecord::default()
            }
        }
    }
}

#[cfg(all(test, feature = "sqlite"))]
mod conversation_read_model_tests {
    use serde_json::json;
    use std::fs;
    use std::path::PathBuf;

    #[test]
    fn local_conversation_timeline_reads_projection_by_conversation_id() {
        let fixture = Fixture::new("conversation-timeline");
        let client = fixture.client();
        let connection =
            crate::internal::local_state::open_writable(&fixture.sqlite_path()).unwrap();
        crate::internal::local_state::messages::upsert_message(
            &connection,
            &crate::internal::local_state::messages::MessageRecord {
                msg_id: "msg-conversation-local".to_owned(),
                owner_identity_id: "alice-id".to_owned(),
                owner_did: "did:example:alice".to_owned(),
                conversation_id: "dm:did:example:bob".to_owned(),
                thread_id: "dm:did:example:bob".to_owned(),
                direction: 0,
                sender_did: "did:example:bob".to_owned(),
                receiver_did: "did:example:alice".to_owned(),
                content_type: "text/plain".to_owned(),
                content: "hello from projection".to_owned(),
                sent_at: "2026-07-05T00:00:00Z".to_owned(),
                stored_at: "2026-07-05T00:00:00Z".to_owned(),
                server_seq: Some(7),
                ..crate::internal::local_state::messages::MessageRecord::default()
            },
        )
        .unwrap();

        let page = client
            .messages()
            .local_conversation_timeline_with_metadata(
                crate::messages::ConversationReadRef::new("dm:did:example:bob").unwrap(),
                crate::messages::LocalHistoryQuery {
                    limit: crate::ids::PageLimit(20),
                    cursor: None,
                },
            )
            .unwrap();

        assert_eq!(page.items.len(), 1);
        assert_eq!(page.items[0].id.as_str(), "msg-conversation-local");
        assert_eq!(
            page.items[0].metadata.server_sequence,
            Some(7),
            "conversation timeline should read committed projection metadata"
        );
        assert!(!page.has_more);
    }

    #[test]
    fn sync_conversation_after_resolves_direct_conversation_to_syncable_thread() {
        let fixture = Fixture::new("conversation-sync-direct");
        let client = fixture.client();

        let resolved = super::resolve_service_conversation_thread(
            &client,
            &crate::messages::ConversationReadRef::new("dm:did:example:bob").unwrap(),
        )
        .unwrap();

        match resolved.thread {
            crate::messages::ThreadRef::Direct(peer) => {
                assert_eq!(peer.as_str(), "did:example:bob");
            }
            other => panic!("expected direct thread, got {other:?}"),
        }
        assert_eq!(resolved.resolved_did.as_deref(), Some("did:example:bob"));
        assert!(resolved.peer_scope.is_none());
    }

    #[test]
    fn sync_conversation_after_rejects_non_canonical_direct_alias() {
        let fixture = Fixture::new("conversation-sync-direct-alias");
        let client = fixture.client();

        let err = super::resolve_service_conversation_thread(
            &client,
            &crate::messages::ConversationReadRef::new("dm:bob.awiki.test").unwrap(),
        )
        .unwrap_err();

        assert!(matches!(
            err,
            crate::ImError::InvalidInput {
                field: Some(ref field),
                ..
            } if field == "conversation_id"
        ));
    }

    #[test]
    fn sync_conversation_after_rejects_legacy_sorted_direct_alias() {
        let fixture = Fixture::new("conversation-sync-sorted-alias");
        let client = fixture.client();

        let err = super::resolve_service_conversation_thread(
            &client,
            &crate::messages::ConversationReadRef::new("dm:did:example:alice:did:example:bob")
                .unwrap(),
        )
        .unwrap_err();

        assert!(matches!(
            err,
            crate::ImError::InvalidInput {
                field: Some(ref field),
                ..
            } if field == "conversation_id"
        ));
    }

    #[test]
    fn sync_conversation_after_public_path_does_not_route_as_raw_thread() {
        let fixture = Fixture::new("conversation-sync-public-path");
        let client = fixture.client();

        let err = client
            .messages()
            .sync_conversation_after(crate::messages::SyncConversationAfterRequest {
                conversation: crate::messages::ConversationReadRef::new("dm:did:example:bob")
                    .unwrap(),
                after_server_seq: Some("0".to_owned()),
                limit: Some(1),
            })
            .unwrap_err();

        assert!(
            !matches!(
                err,
                crate::ImError::UnsupportedCapability { ref capability }
                    if capability == "sync-thread-after-raw-thread"
            ),
            "sync_conversation_after must resolve conversation_id before calling thread-after"
        );
    }

    #[test]
    fn sync_conversation_after_resolves_group_conversation_to_syncable_thread() {
        let fixture = Fixture::new("conversation-sync-group");
        let client = fixture.client();

        let resolved = super::resolve_service_conversation_thread(
            &client,
            &crate::messages::ConversationReadRef::new("group:did:example:group").unwrap(),
        )
        .unwrap();

        match resolved.thread {
            crate::messages::ThreadRef::Group(group) => {
                assert_eq!(group.as_str(), "did:example:group");
            }
            other => panic!("expected group thread, got {other:?}"),
        }
        assert!(resolved.resolved_did.is_none());
        assert!(resolved.peer_scope.is_none());
    }

    #[test]
    fn sync_conversation_after_resolves_peer_scope_from_projection_metadata() {
        let fixture = Fixture::new("conversation-sync-peer-scope");
        let client = fixture.client();
        let peer_scope = crate::internal::local_state::owner_scope::DirectPeerScope::new(
            "user-bob",
            "Bob.AWiki.Test",
        )
        .unwrap();
        let conversation_id =
            crate::internal::local_state::owner_scope::direct_conversation_id_for_peer_scope(
                &peer_scope,
            );
        fixture.seed_message(crate::internal::local_state::messages::MessageRecord {
            msg_id: "msg-peer-scope-sync".to_owned(),
            conversation_id: conversation_id.clone(),
            thread_id: conversation_id.clone(),
            sender_did: "did:example:bob-current".to_owned(),
            receiver_did: "did:example:alice".to_owned(),
            metadata: json!({
                "peer_user_id": "user-bob",
                "peer_full_handle": "bob.awiki.test",
                "peer_current_did": "did:example:bob-current"
            })
            .to_string(),
            ..Fixture::message_record_defaults()
        });

        let resolved = super::resolve_service_conversation_thread(
            &client,
            &crate::messages::ConversationReadRef::new(conversation_id).unwrap(),
        )
        .unwrap();

        match resolved.thread {
            crate::messages::ThreadRef::Direct(peer) => {
                assert_eq!(peer.as_str(), "did:example:bob-current");
            }
            other => panic!("expected direct thread, got {other:?}"),
        }
        assert_eq!(
            resolved.resolved_did.as_deref(),
            Some("did:example:bob-current")
        );
        let resolved_scope = resolved.peer_scope.expect("peer scope");
        assert_eq!(resolved_scope.user_id, "user-bob");
        assert_eq!(resolved_scope.full_handle, "bob.awiki.test");
    }

    #[test]
    fn sync_conversation_after_peer_scope_prefers_metadata_did_over_legacy_outgoing_rows() {
        let fixture = Fixture::new("conversation-sync-peer-scope-rotation");
        let client = fixture.client();
        let peer_scope = crate::internal::local_state::owner_scope::DirectPeerScope::new(
            "user-bob",
            "bob.awiki.test",
        )
        .unwrap();
        let conversation_id =
            crate::internal::local_state::owner_scope::direct_conversation_id_for_peer_scope(
                &peer_scope,
            );
        fixture.seed_message(crate::internal::local_state::messages::MessageRecord {
            msg_id: "msg-legacy-latest".to_owned(),
            conversation_id: conversation_id.clone(),
            thread_id: conversation_id.clone(),
            direction: 1,
            sender_did: "did:example:alice".to_owned(),
            receiver_did: "did:example:bob-old".to_owned(),
            sent_at: "2026-07-05T00:02:00Z".to_owned(),
            stored_at: "2026-07-05T00:02:00Z".to_owned(),
            ..Fixture::message_record_defaults()
        });
        fixture.seed_message(crate::internal::local_state::messages::MessageRecord {
            msg_id: "msg-metadata-older".to_owned(),
            conversation_id: conversation_id.clone(),
            thread_id: conversation_id.clone(),
            direction: 1,
            sender_did: "did:example:alice".to_owned(),
            receiver_did: "did:example:bob-old".to_owned(),
            sent_at: "2026-07-05T00:01:00Z".to_owned(),
            stored_at: "2026-07-05T00:01:00Z".to_owned(),
            metadata: json!({
                "peer_user_id": "user-bob",
                "peer_full_handle": "bob.awiki.test",
                "peer_current_did": "did:example:bob-current"
            })
            .to_string(),
            ..Fixture::message_record_defaults()
        });

        let resolved = super::resolve_service_conversation_thread(
            &client,
            &crate::messages::ConversationReadRef::new(conversation_id).unwrap(),
        )
        .unwrap();

        match resolved.thread {
            crate::messages::ThreadRef::Direct(peer) => {
                assert_eq!(peer.as_str(), "did:example:bob-current");
            }
            other => panic!("expected direct thread, got {other:?}"),
        }
        assert_eq!(
            resolved.resolved_did.as_deref(),
            Some("did:example:bob-current")
        );
    }

    #[test]
    fn sync_conversation_after_peer_scope_prefers_latest_incoming_did_over_stale_metadata() {
        let fixture = Fixture::new("conversation-sync-peer-scope-latest-incoming");
        let client = fixture.client();
        let peer_scope = crate::internal::local_state::owner_scope::DirectPeerScope::new(
            "user-bob",
            "bob.awiki.test",
        )
        .unwrap();
        let conversation_id =
            crate::internal::local_state::owner_scope::direct_conversation_id_for_peer_scope(
                &peer_scope,
            );
        fixture.seed_message(crate::internal::local_state::messages::MessageRecord {
            msg_id: "msg-stale-metadata-latest".to_owned(),
            conversation_id: conversation_id.clone(),
            thread_id: conversation_id.clone(),
            sender_did: "did:example:bob-current".to_owned(),
            receiver_did: "did:example:alice".to_owned(),
            sent_at: "2026-07-05T00:02:00Z".to_owned(),
            stored_at: "2026-07-05T00:02:00Z".to_owned(),
            metadata: json!({
                "peer_user_id": "user-bob",
                "peer_full_handle": "bob.awiki.test",
                "peer_current_did": "did:example:bob-old"
            })
            .to_string(),
            ..Fixture::message_record_defaults()
        });

        let resolved = super::resolve_service_conversation_thread(
            &client,
            &crate::messages::ConversationReadRef::new(conversation_id).unwrap(),
        )
        .unwrap();

        match resolved.thread {
            crate::messages::ThreadRef::Direct(peer) => {
                assert_eq!(peer.as_str(), "did:example:bob-current");
            }
            other => panic!("expected direct thread, got {other:?}"),
        }
        assert_eq!(
            resolved.resolved_did.as_deref(),
            Some("did:example:bob-current")
        );
    }

    struct Fixture {
        root: PathBuf,
    }

    impl Fixture {
        fn new(prefix: &str) -> Self {
            let nanos = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos();
            let root = std::env::temp_dir().join(format!(
                "im-core-service-{prefix}-{}-{nanos}",
                std::process::id()
            ));
            let identity_root = root.join("identities");
            let identity_dir = identity_root.join("alice");
            fs::create_dir_all(&identity_dir).unwrap();
            fs::create_dir_all(root.join("local")).unwrap();
            fs::write(identity_root.join("default"), "alice\n").unwrap();
            fs::write(
                identity_root.join("registry.json"),
                json!({
                    "default_identity": "alice",
                    "identities": [{
                        "id": "alice-id",
                        "did": "did:example:alice",
                        "local_alias": "alice",
                        "ready_for_auth": true,
                        "ready_for_messaging": true,
                        "missing": []
                    }]
                })
                .to_string(),
            )
            .unwrap();
            fs::write(identity_dir.join("did.json"), "{}").unwrap();
            Self { root }
        }

        fn client(&self) -> crate::core::ImClient {
            crate::core::ImCore::new(
                crate::ImCoreConfig {
                    service_base_url: crate::ServiceEndpoint::parse("https://example.test")
                        .unwrap(),
                    did_domain: "awiki.test".to_owned(),
                    user_service_endpoint: None,
                    message_service_endpoint: None,
                    mail_service_endpoint: None,
                    anp_service_endpoint: None,
                    anp_service_did: None,
                    ca_bundle: None,
                    transport_policy: crate::MessageTransportPolicy::HttpOnly,
                },
                crate::ImCorePaths {
                    identities: crate::IdentityRegistryPaths {
                        identity_root_dir: self.root.join("identities"),
                        registry_path: self.root.join("identities").join("registry.json"),
                        default_identity_path: Some(self.root.join("identities").join("default")),
                    },
                    local_state: crate::LocalStatePaths {
                        sqlite_path: self.sqlite_path(),
                    },
                    runtime: crate::RuntimePaths {
                        cache_dir: self.root.join("cache"),
                        temp_dir: self.root.join("tmp"),
                    },
                },
            )
            .unwrap()
            .client(crate::identity::IdentitySelector::LocalAlias(
                "alice".to_owned(),
            ))
            .unwrap()
        }

        fn sqlite_path(&self) -> PathBuf {
            self.root.join("local").join("im.sqlite")
        }

        fn seed_message(&self, record: crate::internal::local_state::messages::MessageRecord) {
            let connection =
                crate::internal::local_state::open_writable(&self.sqlite_path()).unwrap();
            crate::internal::local_state::messages::upsert_message(&connection, &record).unwrap();
        }

        fn message_record_defaults() -> crate::internal::local_state::messages::MessageRecord {
            crate::internal::local_state::messages::MessageRecord {
                owner_identity_id: "alice-id".to_owned(),
                owner_did: "did:example:alice".to_owned(),
                direction: 0,
                content_type: "text/plain".to_owned(),
                content: "hello from projection".to_owned(),
                sent_at: "2026-07-05T00:00:00Z".to_owned(),
                stored_at: "2026-07-05T00:00:00Z".to_owned(),
                ..crate::internal::local_state::messages::MessageRecord::default()
            }
        }
    }
}

#[cfg(all(test, feature = "sqlite"))]
mod direct_e2ee_async_persistence_tests {
    use serde_json::json;

    use crate::internal::local_state::owner_scope::DirectPeerScope;
    use crate::internal::secure_direct::send::{
        DirectSecureAttachmentLocalEffect, DirectSecureAttachmentSendResult,
        DirectSecureLocalEffect, DirectSecureTextSendResult,
    };

    #[tokio::test]
    async fn deferred_direct_e2ee_success_projection_uses_db_actor() {
        let fixture = Fixture::new("direct-e2ee-success-actor");
        let client = fixture.client();
        let sdk_result = sdk_result("msg-secure-actor", "accepted", Some(17));

        let result = super::persist_deferred_direct_e2ee_effect(
            &client,
            DirectSecureTextSendResult {
                sdk_result,
                queued_outbox_id: None,
                target_did: "did:example:bob".to_owned(),
                target_handle: None,
                peer_scope: None,
                text: "actor persisted secret".to_owned(),
                kind: crate::messages::MessageKind::Text,
                raw: Some(json!({ "accepted": true })),
                local_effect: DirectSecureLocalEffect::PersistOutgoing,
            },
            None,
            None,
        )
        .await
        .unwrap();

        assert_eq!(result.message.id.as_str(), "msg-secure-actor");
        let db = rusqlite::Connection::open(fixture.sqlite_path()).unwrap();
        let stored = db
            .query_row(
                r#"
SELECT content, is_e2ee, server_seq, metadata
FROM messages
WHERE owner_identity_id = 'alice-id' AND msg_id = 'msg-secure-actor'"#,
                [],
                |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, i64>(1)?,
                        row.get::<_, Option<i64>>(2)?,
                        row.get::<_, String>(3)?,
                    ))
                },
            )
            .unwrap();
        assert_eq!(stored.0, "actor persisted secret");
        assert_eq!(stored.1, 1);
        assert_eq!(stored.2, Some(17));
        let metadata: serde_json::Value = serde_json::from_str(&stored.3).unwrap();
        assert_eq!(metadata["security"], "direct-e2ee");
        assert_eq!(metadata["contains_sensitive"], false);
    }

    #[tokio::test]
    async fn deferred_direct_e2ee_peer_scope_projection_and_result_use_same_thread() {
        let fixture = Fixture::new("direct-e2ee-peer-scope");
        let client = fixture.client();
        let peer_scope = DirectPeerScope::new("user-bob", "bob.awiki.test").unwrap();
        let sdk_result = sdk_result("msg-secure-peer-scope", "accepted", Some(18));

        let result = super::persist_deferred_direct_e2ee_effect(
            &client,
            DirectSecureTextSendResult {
                sdk_result,
                queued_outbox_id: None,
                target_did: "did:example:bob-current".to_owned(),
                target_handle: Some("bob.awiki.test".to_owned()),
                peer_scope: Some(peer_scope),
                text: "peer scoped secret".to_owned(),
                kind: crate::messages::MessageKind::Text,
                raw: Some(json!({ "accepted": true })),
                local_effect: DirectSecureLocalEffect::PersistOutgoing,
            },
            None,
            None,
        )
        .await
        .unwrap();
        let result_thread = match &result.message.thread {
            crate::messages::ThreadRef::Thread(thread) => thread.as_str().to_owned(),
            other => panic!("expected peer-scope thread result, got {other:?}"),
        };

        let db = rusqlite::Connection::open(fixture.sqlite_path()).unwrap();
        let stored = db
            .query_row(
                r#"
SELECT conversation_id, receiver_did, metadata
FROM messages
WHERE owner_identity_id = 'alice-id' AND msg_id = 'msg-secure-peer-scope'"#,
                [],
                |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, String>(1)?,
                        row.get::<_, String>(2)?,
                    ))
                },
            )
            .unwrap();
        assert_eq!(stored.0, result_thread);
        assert_eq!(stored.1, "did:example:bob-current");
        let metadata: serde_json::Value = serde_json::from_str(&stored.2).unwrap();
        assert_eq!(metadata["security"], "direct-e2ee");
        assert_eq!(metadata["target_handle"], "bob.awiki.test");
        assert_eq!(metadata["peer_user_id"], "user-bob");
        assert_eq!(metadata["peer_full_handle"], "bob.awiki.test");
        assert_eq!(metadata["resolved_target_did"], "did:example:bob-current");
    }

    #[tokio::test]
    async fn deferred_direct_e2ee_attachment_peer_scope_projection_and_result_use_same_thread() {
        let fixture = Fixture::new("direct-e2ee-attachment-peer-scope");
        let client = fixture.client();
        let peer_scope = DirectPeerScope::new("user-bob", "bob.awiki.test").unwrap();
        let sdk_result =
            sdk_attachment_result("msg-secure-attachment-peer-scope", "accepted", Some(19));
        let redacted_manifest = json!({
            "primary_attachment_id": "att-peer",
            "attachments": [{
                "attachment_id": "att-peer",
                "filename": "peer.txt",
                "mime_type": "text/plain",
                "size": "4",
                "object_uri": "awiki://object/peer"
            }]
        });

        let result = super::persist_deferred_direct_e2ee_attachment_effect(
            &client,
            DirectSecureAttachmentSendResult {
                sdk_result,
                target_did: "did:example:bob-current".to_owned(),
                target_handle: Some("bob.awiki.test".to_owned()),
                peer_scope: Some(peer_scope),
                redacted_manifest,
                raw: Some(json!({ "accepted": true })),
                local_effect: DirectSecureAttachmentLocalEffect::PersistOutgoing,
            },
            None,
            None,
        )
        .await
        .unwrap();
        let result_thread = match &result.message.thread {
            crate::messages::ThreadRef::Thread(thread) => thread.as_str().to_owned(),
            other => panic!("expected peer-scope thread result, got {other:?}"),
        };

        let db = rusqlite::Connection::open(fixture.sqlite_path()).unwrap();
        let stored = db
            .query_row(
                r#"
SELECT conversation_id, receiver_did, metadata
FROM messages
WHERE owner_identity_id = 'alice-id' AND msg_id = 'msg-secure-attachment-peer-scope'"#,
                [],
                |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, String>(1)?,
                        row.get::<_, String>(2)?,
                    ))
                },
            )
            .unwrap();
        assert_eq!(stored.0, result_thread);
        assert_eq!(stored.1, "did:example:bob-current");
        let metadata: serde_json::Value = serde_json::from_str(&stored.2).unwrap();
        assert_eq!(metadata["security"], "direct-e2ee");
        assert_eq!(metadata["target_handle"], "bob.awiki.test");
        assert_eq!(metadata["peer_user_id"], "user-bob");
        assert_eq!(metadata["peer_full_handle"], "bob.awiki.test");
        assert_eq!(metadata["resolved_target_did"], "did:example:bob-current");
        assert_eq!(metadata["attachment_id"], "att-peer");
    }

    #[tokio::test]
    async fn deferred_direct_e2ee_pending_outbox_uses_db_actor() {
        let fixture = Fixture::new("direct-e2ee-outbox-actor");
        let client = fixture.client();
        let sdk_result = sdk_result("msg-secure-queued", "queued", None);
        let scope = crate::internal::store::e2ee_outbox::E2eeOutboxOwnerScope::for_client(&client);

        super::persist_deferred_direct_e2ee_effect(
            &client,
            DirectSecureTextSendResult {
                sdk_result,
                queued_outbox_id: Some("outbox-actor".to_owned()),
                target_did: "did:example:bob".to_owned(),
                target_handle: None,
                peer_scope: None,
                text: "queued actor secret".to_owned(),
                kind: crate::messages::MessageKind::Markdown,
                raw: None,
                local_effect: DirectSecureLocalEffect::QueueOutbox(
                    crate::internal::store::e2ee_outbox::E2eeOutboxRecord {
                        outbox_id: "outbox-actor".to_owned(),
                        owner_identity_id: scope.owner_identity_id,
                        owner_did: scope.owner_did,
                        credential_name: scope.credential_name,
                        peer_did: "did:example:bob".to_owned(),
                        original_type: "markdown".to_owned(),
                        plaintext: "queued actor secret".to_owned(),
                        local_status: "queued".to_owned(),
                        last_error_code: "pending-confirmation".to_owned(),
                        retry_hint: "retry".to_owned(),
                        ..Default::default()
                    },
                ),
            },
            None,
            None,
        )
        .await
        .unwrap();

        let db = rusqlite::Connection::open(fixture.sqlite_path()).unwrap();
        let stored = db
            .query_row(
                r#"
SELECT peer_did, original_type, plaintext, local_status, last_error_code
FROM e2ee_outbox
WHERE owner_identity_id = 'alice-id' AND outbox_id = 'outbox-actor'"#,
                [],
                |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, String>(1)?,
                        row.get::<_, String>(2)?,
                        row.get::<_, String>(3)?,
                        row.get::<_, String>(4)?,
                    ))
                },
            )
            .unwrap();
        assert_eq!(stored.0, "did:example:bob");
        assert_eq!(stored.1, "markdown");
        assert_eq!(stored.2, "queued actor secret");
        assert_eq!(stored.3, "queued");
        assert_eq!(stored.4, "pending-confirmation");
    }

    fn sdk_result(
        message_id: &str,
        delivery_state: &str,
        server_sequence: Option<i64>,
    ) -> crate::messages::SendMessageResult {
        crate::messages::SendMessageResult {
            message: crate::messages::Message {
                id: crate::ids::MessageId::parse(message_id).unwrap(),
                thread: crate::messages::ThreadRef::Direct(
                    crate::ids::PeerRef::parse("did:example:bob", "").unwrap(),
                ),
                direction: crate::messages::MessageDirection::Outgoing,
                sender: crate::ids::PeerRef::parse("did:example:alice", "").unwrap(),
                receiver: Some(crate::ids::PeerRef::parse("did:example:bob", "").unwrap()),
                group: None,
                body: crate::messages::MessageBodyView::Text {
                    text: "redacted from db builder".to_owned(),
                    kind: crate::messages::MessageKind::Text,
                },
                sent_at: Some("2026-05-24T00:00:00Z".to_owned()),
                received_at: None,
                metadata: crate::messages::MessageMetadata {
                    operation_id: Some(message_id.to_owned()),
                    delivery_state: Some(delivery_state.to_owned()),
                    send_state: None,
                    retry_plan: None,
                    server_sequence,
                    content_type: Some("text/plain".to_owned()),
                    conversation_identity: None,
                    attributes: vec![crate::messages::MessageMetadataAttribute {
                        key: "security".to_owned(),
                        value: "direct-e2ee".to_owned(),
                    }],
                },
            },
            delivery: crate::messages::DeliveryState::Accepted,
            warnings: Vec::new(),
        }
    }

    fn sdk_attachment_result(
        message_id: &str,
        delivery_state: &str,
        server_sequence: Option<i64>,
    ) -> crate::messages::SendMessageResult {
        crate::messages::SendMessageResult {
            message: crate::messages::Message {
                id: crate::ids::MessageId::parse(message_id).unwrap(),
                thread: crate::messages::ThreadRef::Direct(
                    crate::ids::PeerRef::parse("did:example:bob", "").unwrap(),
                ),
                direction: crate::messages::MessageDirection::Outgoing,
                sender: crate::ids::PeerRef::parse("did:example:alice", "").unwrap(),
                receiver: Some(crate::ids::PeerRef::parse("did:example:bob", "").unwrap()),
                group: None,
                body: crate::messages::MessageBodyView::Unsupported {
                    content_type: Some(
                        crate::attachments::manifest::attachment_manifest_content_type().to_owned(),
                    ),
                },
                sent_at: Some("2026-05-24T00:00:00Z".to_owned()),
                received_at: None,
                metadata: crate::messages::MessageMetadata {
                    operation_id: Some(message_id.to_owned()),
                    delivery_state: Some(delivery_state.to_owned()),
                    send_state: None,
                    retry_plan: None,
                    server_sequence,
                    content_type: Some(
                        crate::attachments::manifest::attachment_manifest_content_type().to_owned(),
                    ),
                    conversation_identity: None,
                    attributes: vec![crate::messages::MessageMetadataAttribute {
                        key: "security".to_owned(),
                        value: "direct-e2ee".to_owned(),
                    }],
                },
            },
            delivery: crate::messages::DeliveryState::Accepted,
            warnings: Vec::new(),
        }
    }

    struct Fixture {
        root: std::path::PathBuf,
    }

    impl Fixture {
        fn new(prefix: &str) -> Self {
            let nanos = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos();
            let root = std::env::temp_dir()
                .join(format!("im-core-{prefix}-{}-{nanos}", std::process::id()));
            let identity_root = root.join("identities");
            let identity_dir = identity_root.join("alice");
            std::fs::create_dir_all(&identity_dir).unwrap();
            std::fs::create_dir_all(root.join("local")).unwrap();
            std::fs::write(identity_root.join("default"), "alice\n").unwrap();
            std::fs::write(
                identity_root.join("registry.json"),
                json!({
                    "default_identity": "alice",
                    "identities": [{
                        "id": "alice-id",
                        "did": "did:example:alice",
                        "local_alias": "alice",
                        "ready_for_auth": true,
                        "ready_for_messaging": true,
                        "missing": []
                    }]
                })
                .to_string(),
            )
            .unwrap();
            std::fs::write(identity_dir.join("did.json"), "{}").unwrap();
            Self { root }
        }

        fn client(&self) -> crate::core::ImClient {
            crate::core::ImCore::new(
                crate::ImCoreConfig {
                    service_base_url: crate::ServiceEndpoint::parse("https://example.test")
                        .unwrap(),
                    did_domain: "awiki.test".to_owned(),
                    user_service_endpoint: None,
                    message_service_endpoint: None,
                    mail_service_endpoint: None,
                    anp_service_endpoint: None,
                    anp_service_did: None,
                    ca_bundle: None,
                    transport_policy: crate::MessageTransportPolicy::HttpOnly,
                },
                crate::ImCorePaths {
                    identities: crate::IdentityRegistryPaths {
                        identity_root_dir: self.root.join("identities"),
                        registry_path: self.root.join("identities").join("registry.json"),
                        default_identity_path: Some(self.root.join("identities").join("default")),
                    },
                    local_state: crate::LocalStatePaths {
                        sqlite_path: self.sqlite_path(),
                    },
                    runtime: crate::RuntimePaths {
                        cache_dir: self.root.join("cache"),
                        temp_dir: self.root.join("tmp"),
                    },
                },
            )
            .unwrap()
            .client(crate::identity::IdentitySelector::LocalAlias(
                "alice".to_owned(),
            ))
            .unwrap()
        }

        fn sqlite_path(&self) -> std::path::PathBuf {
            self.root.join("local").join("im.sqlite")
        }
    }
}

impl<'a> MessageService<'a> {
    pub(crate) fn new(client: &'a crate::core::ImClient) -> Self {
        Self { client }
    }

    pub fn send(
        &self,
        request: super::SendMessageRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        validate_body(&request.body)?;
        validate_send_mode(&request.target, &request.security)?;
        validate_attachment_security(&request.body, &request.security)?;
        validate_delegated_send_scope(&request)?;
        match (&request.target, &request.security) {
            (
                super::MessageTarget::Direct(_),
                super::MessageSecurityMode::E2eeRequired | super::MessageSecurityMode::SecureDirect,
            ) => self.send_direct_e2ee(resolve_send_request(self.client, request)?),
            (
                super::MessageTarget::Direct(_),
                super::MessageSecurityMode::DefaultPlain | super::MessageSecurityMode::Plain,
            ) if matches!(request.body, super::MessageBody::Attachment { .. }) => {
                self.send_plain_attachment(request)
            }
            (super::MessageTarget::Direct(_), _) => {
                let resolved = resolve_send_request(self.client, request)?;
                let direct_handle = resolved.direct_handle().map(str::to_owned);
                let peer_scope = resolved.peer_scope.clone();
                let mut result = crate::internal::message_runtime::direct::DirectTextSender::new(
                    self.client,
                    crate::internal::auth::session::FileSessionProvider::new(self.client),
                    crate::internal::transport::CoreHttpTransport::new(self.client),
                )
                .send(
                    crate::internal::message_runtime::direct::DirectTextSend {
                        request: resolved.request.clone(),
                        resolved_target_did: resolved.target_did.clone(),
                        credentials: None,
                    },
                )?;
                normalize_resolved_direct_send_result(&mut result.sdk_result, &resolved)?;
                #[cfg(feature = "sqlite")]
                match
                    crate::internal::message_runtime::local_projection::persist_direct_outgoing_result(
                        self.client,
                        &result.target_did,
                        direct_handle.as_deref(),
                        peer_scope.as_ref(),
                        &result.sdk_result,
                    )
                {
                    Ok(()) => self
                        .client
                        .emit_committed_local_message_projection("local_send"),
                    Err(err) => {
                        result
                            .sdk_result
                            .warnings
                            .push(format!("Failed to persist local message: {err}"));
                    }
                }
                Ok(result.sdk_result)
            }
            (
                super::MessageTarget::Group(_),
                super::MessageSecurityMode::E2eeRequired | super::MessageSecurityMode::GroupE2ee,
            ) => self.send_group_e2ee(request),
            (
                super::MessageTarget::Group(_),
                super::MessageSecurityMode::DefaultPlain | super::MessageSecurityMode::Plain,
            ) if matches!(request.body, super::MessageBody::Attachment { .. }) => {
                self.send_plain_attachment(request)
            }
            (super::MessageTarget::Group(_), _) => {
                let mut result = crate::internal::message_runtime::group::GroupTextSender::new(
                    self.client,
                    crate::internal::auth::session::FileSessionProvider::new(self.client),
                    crate::internal::transport::CoreHttpTransport::new(self.client),
                )
                .send(crate::internal::message_runtime::group::GroupTextSend {
                    request,
                    credentials: None,
                })?;
                #[cfg(feature = "sqlite")]
                match
                    crate::internal::message_runtime::local_projection::persist_group_outgoing_result(
                        self.client,
                        &result.group_did,
                        &result.sdk_result,
                    )
                {
                    Ok(()) => self
                        .client
                        .emit_committed_local_message_projection("local_send"),
                    Err(err) => {
                        result
                            .sdk_result
                            .warnings
                            .push(format!("Failed to persist local group message: {err}"));
                    }
                }
                Ok(result.sdk_result)
            }
        }
    }

    pub(crate) fn send_secure_attachment(
        &self,
        request: super::SendMessageRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        validate_body(&request.body)?;
        validate_send_mode(&request.target, &request.security)?;
        validate_secure_attachment_request(&request.body, &request.security)?;
        match (&request.target, &request.security) {
            (
                super::MessageTarget::Direct(_),
                super::MessageSecurityMode::E2eeRequired | super::MessageSecurityMode::SecureDirect,
            ) => self.send_direct_e2ee(resolve_send_request(self.client, request)?),
            (
                super::MessageTarget::Group(_),
                super::MessageSecurityMode::E2eeRequired | super::MessageSecurityMode::GroupE2ee,
            ) => self.send_group_e2ee(request),
            _ => Err(crate::ImError::unsupported("secure-attachment")),
        }
    }

    pub async fn send_async(
        &self,
        request: super::SendMessageRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        validate_body(&request.body)?;
        validate_send_mode(&request.target, &request.security)?;
        validate_attachment_security(&request.body, &request.security)?;
        validate_delegated_send_scope(&request)?;
        match (&request.target, &request.security) {
            (
                super::MessageTarget::Direct(_),
                super::MessageSecurityMode::E2eeRequired | super::MessageSecurityMode::SecureDirect,
            ) => {
                self.send_direct_e2ee_async(resolve_send_request_async(self.client, request).await?)
                    .await
            }
            (
                super::MessageTarget::Direct(_),
                super::MessageSecurityMode::DefaultPlain | super::MessageSecurityMode::Plain,
            ) if matches!(request.body, super::MessageBody::Attachment { .. }) => {
                self.send_plain_attachment_async(request).await
            }
            (super::MessageTarget::Direct(_), _) => {
                let resolved = resolve_send_request_async(self.client, request).await?;
                let direct_handle = resolved.direct_handle().map(str::to_owned);
                let peer_scope = resolved.peer_scope.clone();
                let mut result = crate::internal::message_runtime::direct::DirectTextSender::new(
                    self.client,
                    crate::internal::auth::session::FileSessionProvider::new(self.client),
                    crate::internal::transport::CoreHttpTransport::new(self.client),
                )
                .send_async(crate::internal::message_runtime::direct::DirectTextSend {
                    request: resolved.request.clone(),
                    resolved_target_did: resolved.target_did.clone(),
                    credentials: None,
                })
                .await?;
                normalize_resolved_direct_send_result(&mut result.sdk_result, &resolved)?;
                #[cfg(feature = "sqlite")]
                match
                    crate::internal::message_runtime::local_projection::persist_direct_outgoing_result_async(
                        self.client,
                        &result.target_did,
                        direct_handle.as_deref(),
                        peer_scope.as_ref(),
                        &result.sdk_result,
                    )
                    .await
                {
                    Ok(()) => self
                        .client
                        .emit_committed_local_message_projection("local_send"),
                    Err(err) => {
                        result
                            .sdk_result
                            .warnings
                            .push(format!("Failed to persist local message: {err}"));
                    }
                }
                Ok(result.sdk_result)
            }
            (
                super::MessageTarget::Group(_),
                super::MessageSecurityMode::E2eeRequired | super::MessageSecurityMode::GroupE2ee,
            ) => self.send_group_e2ee_async(request).await,
            (
                super::MessageTarget::Group(_),
                super::MessageSecurityMode::DefaultPlain | super::MessageSecurityMode::Plain,
            ) if matches!(request.body, super::MessageBody::Attachment { .. }) => {
                self.send_plain_attachment_async(request).await
            }
            (super::MessageTarget::Group(_), _) => {
                let mut result = crate::internal::message_runtime::group::GroupTextSender::new(
                    self.client,
                    crate::internal::auth::session::FileSessionProvider::new(self.client),
                    crate::internal::transport::CoreHttpTransport::new(self.client),
                )
                .send_async(crate::internal::message_runtime::group::GroupTextSend {
                    request,
                    credentials: None,
                })
                .await?;
                #[cfg(feature = "sqlite")]
                match
                    crate::internal::message_runtime::local_projection::persist_group_outgoing_result_async(
                        self.client,
                        &result.group_did,
                        &result.sdk_result,
                    )
                    .await
                {
                    Ok(()) => self
                        .client
                        .emit_committed_local_message_projection("local_send"),
                    Err(err) => {
                        result
                            .sdk_result
                            .warnings
                            .push(format!("Failed to persist local group message: {err}"));
                    }
                }
                Ok(result.sdk_result)
            }
        }
    }

    pub async fn send_conversation_text_async(
        &self,
        request: super::SendConversationTextRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        let body = super::MessageBody::Text {
            text: request.text,
            kind: if request.markdown {
                super::MessageKind::Markdown
            } else {
                super::MessageKind::Text
            },
        };
        let send_request = conversation_send_request(
            self.client,
            request.conversation,
            body,
            request.security,
            request.client_message_id,
            request.idempotency_key,
            request.wait_for_final_acceptance,
            request.delegated_signing,
        )?;
        self.send_conversation_request_async(send_request).await
    }

    pub async fn send_conversation_payload_async(
        &self,
        request: super::SendConversationPayloadRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        let send_request = conversation_send_request(
            self.client,
            request.conversation,
            super::MessageBody::Payload {
                payload: request.payload,
            },
            request.security,
            request.client_message_id,
            request.idempotency_key,
            request.wait_for_final_acceptance,
            request.delegated_signing,
        )?;
        self.send_conversation_request_async(send_request).await
    }

    async fn send_conversation_request_async(
        &self,
        resolved: ResolvedConversationSendRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        validate_plain_conversation_send(&resolved.request)?;
        let message_id = resolved
            .request
            .client_message_id
            .as_ref()
            .expect("conversation send request should always have a client message id")
            .clone();
        let operation_id = resolved.request.delivery.idempotency_key.as_deref();
        let target_did = resolved.target_did.as_deref();
        let target_handle = resolved.target_handle.as_deref();
        let peer_scope = resolved.peer_scope.as_ref();

        crate::internal::message_runtime::local_projection::persist_send_projection_async(
            self.client,
            &resolved.request.target,
            &resolved.request.body,
            &message_id,
            operation_id,
            super::DeliveryState::StoredLocally,
            target_did,
            target_handle,
            peer_scope,
        )
        .await?;
        self.client
            .emit_committed_local_message_projection("local_send_pending");

        match self.send_plain_conversation_network_async(&resolved).await {
            Ok(result) => Ok(result),
            Err(err) => {
                let mut failed_resolved = resolved.clone();
                let err = match self
                    .stale_delivery_retry_target_async(&resolved, &err)
                    .await
                {
                    Ok(Some(retry_resolved)) => {
                        match self
                            .send_plain_conversation_network_async(&retry_resolved)
                            .await
                        {
                            Ok(result) => return Ok(result),
                            Err(retry_err) => {
                                failed_resolved = retry_resolved;
                                retry_err
                            }
                        }
                    }
                    Ok(None) => err,
                    Err(resolve_err) => resolve_err,
                };
                self.persist_failed_conversation_send_projection_async(&failed_resolved, &err)
                    .await;
                Err(err)
            }
        }
    }

    async fn stale_delivery_retry_target_async(
        &self,
        resolved: &ResolvedConversationSendRequest,
        err: &crate::ImError,
    ) -> crate::ImResult<Option<ResolvedConversationSendRequest>> {
        let Some(stale) = stale_delivery_target_from_error(err, self.client.did().as_str()) else {
            return Ok(None);
        };
        let super::MessageTarget::Direct(_) = &resolved.request.target else {
            return Ok(None);
        };

        let mut current_did = stale.current_did;
        let mut full_handle = stale
            .full_handle
            .or_else(|| resolved.target_handle.clone())
            .or_else(|| {
                resolved
                    .peer_scope
                    .as_ref()
                    .map(|scope| scope.full_handle.clone())
            });
        let mut user_id = stale.user_id.or_else(|| {
            let handle = full_handle.as_deref()?;
            let scope = resolved.peer_scope.as_ref()?;
            (scope.full_handle.eq_ignore_ascii_case(handle)).then(|| scope.user_id.clone())
        });

        if let Some(handle) = full_handle.clone() {
            if current_did.is_none() || user_id.is_none() {
                let lookup = crate::internal::handle_discovery::resolve_direct_handle_async(
                    self.client,
                    &handle,
                )
                .await?;
                current_did = Some(lookup.target_did);
                full_handle = Some(lookup.full_handle);
                user_id = Some(lookup.user_id);
            }
        }

        let Some(current_did) = current_did
            .as_deref()
            .and_then(|did| canonical_peer_did_value(did, self.client.did().as_str()))
        else {
            return Ok(None);
        };
        if resolved
            .target_did
            .as_deref()
            .is_some_and(|did| did.trim() == current_did)
        {
            return Ok(None);
        }

        let full_handle = full_handle
            .map(|value| value.trim().to_owned())
            .filter(|value| !value.is_empty());
        let peer_scope = match (user_id, full_handle.as_ref()) {
            (Some(user_id), Some(handle)) => Some(
                crate::internal::local_state::owner_scope::DirectPeerScope::new(
                    user_id,
                    handle.clone(),
                )?,
            ),
            _ => None,
        };
        let target_peer = match full_handle.as_deref() {
            Some(handle) => crate::ids::PeerRef::parse(handle, "")?,
            None => crate::ids::PeerRef::parse(&current_did, "")?,
        };

        let mut retry = resolved.clone();
        retry.request.target = super::MessageTarget::Direct(target_peer);
        retry.target_did = Some(current_did);
        retry.target_handle = full_handle;
        retry.peer_scope = peer_scope;
        Ok(Some(retry))
    }

    async fn persist_failed_conversation_send_projection_async(
        &self,
        resolved: &ResolvedConversationSendRequest,
        err: &crate::ImError,
    ) {
        let Some(message_id) = resolved.request.client_message_id.as_ref() else {
            return;
        };
        let failed =
            crate::internal::message_runtime::local_projection::persist_send_projection_async(
                self.client,
                &resolved.request.target,
                &resolved.request.body,
                message_id,
                resolved.request.delivery.idempotency_key.as_deref(),
                super::DeliveryState::Failed {
                    reason: err.to_string(),
                },
                resolved.target_did.as_deref(),
                resolved.target_handle.as_deref(),
                resolved.peer_scope.as_ref(),
            )
            .await;
        match failed {
            Ok(_) => self
                .client
                .emit_committed_local_message_projection("local_send_failed"),
            Err(_persist_err) => {}
        }
    }

    async fn send_plain_conversation_network_async(
        &self,
        resolved: &ResolvedConversationSendRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        match &resolved.request.target {
            super::MessageTarget::Direct(_) => {
                let mut result = crate::internal::message_runtime::direct::DirectTextSender::new(
                    self.client,
                    crate::internal::auth::session::FileSessionProvider::new(self.client),
                    crate::internal::transport::CoreHttpTransport::new(self.client),
                )
                .send_async(crate::internal::message_runtime::direct::DirectTextSend {
                    request: resolved.request.clone(),
                    resolved_target_did: resolved.target_did.clone(),
                    credentials: None,
                })
                .await?;
                normalize_direct_send_result_for_peer_scope(
                    &mut result.sdk_result,
                    resolved.peer_scope.as_ref(),
                    resolved.target_handle.as_deref(),
                    Some(result.target_did.as_str()),
                )?;
                #[cfg(feature = "sqlite")]
                match crate::internal::message_runtime::local_projection::persist_direct_outgoing_result_async(
                    self.client,
                    &result.target_did,
                    resolved.target_handle.as_deref(),
                    resolved.peer_scope.as_ref(),
                    &result.sdk_result,
                )
                .await
                {
                    Ok(()) => self
                        .client
                        .emit_committed_local_message_projection("local_send"),
                    Err(err) => {
                        result
                            .sdk_result
                            .warnings
                            .push(format!("Failed to persist local conversation message: {err}"));
                    }
                }
                Ok(result.sdk_result)
            }
            super::MessageTarget::Group(_) => {
                let mut result = crate::internal::message_runtime::group::GroupTextSender::new(
                    self.client,
                    crate::internal::auth::session::FileSessionProvider::new(self.client),
                    crate::internal::transport::CoreHttpTransport::new(self.client),
                )
                .send_async(crate::internal::message_runtime::group::GroupTextSend {
                    request: resolved.request.clone(),
                    credentials: None,
                })
                .await?;
                #[cfg(feature = "sqlite")]
                match crate::internal::message_runtime::local_projection::persist_group_outgoing_result_async(
                    self.client,
                    &result.group_did,
                    &result.sdk_result,
                )
                .await
                {
                    Ok(()) => self
                        .client
                        .emit_committed_local_message_projection("local_send"),
                    Err(err) => {
                        result
                            .sdk_result
                            .warnings
                            .push(format!("Failed to persist local conversation group message: {err}"));
                    }
                }
                Ok(result.sdk_result)
            }
        }
    }

    pub(crate) async fn send_secure_attachment_async(
        &self,
        request: super::SendMessageRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        validate_body(&request.body)?;
        validate_send_mode(&request.target, &request.security)?;
        validate_secure_attachment_request(&request.body, &request.security)?;
        match (&request.target, &request.security) {
            (
                super::MessageTarget::Direct(_),
                super::MessageSecurityMode::E2eeRequired | super::MessageSecurityMode::SecureDirect,
            ) => {
                self.send_direct_e2ee_async(resolve_send_request_async(self.client, request).await?)
                    .await
            }
            (
                super::MessageTarget::Group(_),
                super::MessageSecurityMode::E2eeRequired | super::MessageSecurityMode::GroupE2ee,
            ) => self.send_group_e2ee_async(request).await,
            _ => Err(crate::ImError::unsupported("secure-attachment")),
        }
    }

    fn send_plain_attachment(
        &self,
        request: super::SendMessageRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        let target = request.target.clone();
        let attachment = attachment_request_from_message_request(request)?;
        self.client
            .attachments()
            .send(target, attachment)
            .map(|result| result.message)
    }

    async fn send_plain_attachment_async(
        &self,
        request: super::SendMessageRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        let target = request.target.clone();
        let attachment = attachment_request_from_message_request(request)?;
        self.client
            .attachments()
            .send_async(target, attachment)
            .await
            .map(|result| result.message)
    }

    fn send_direct_e2ee(
        &self,
        resolved: ResolvedSendRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        #[cfg(all(feature = "sqlite", feature = "blocking"))]
        {
            send_direct_e2ee_with_client(self.client, resolved)
        }
        #[cfg(all(feature = "sqlite", not(feature = "blocking")))]
        {
            let _ = resolved;
            Err(crate::ImError::unsupported("sync-secure-direct-send"))
        }
        #[cfg(not(feature = "sqlite"))]
        {
            let _ = resolved;
            Err(crate::ImError::unsupported("secure-direct"))
        }
    }

    #[cfg(feature = "sqlite")]
    async fn send_direct_e2ee_async(
        &self,
        resolved: ResolvedSendRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        let target_handle = resolved.direct_handle().map(str::to_owned);
        let peer_scope = resolved.peer_scope.clone();
        if matches!(resolved.request.body, super::MessageBody::Attachment { .. }) {
            let committed =
                crate::internal::attachment_runtime::upload::AttachmentUploadRuntime::new(
                    self.client,
                    crate::internal::auth::session::FileSessionProvider::new(self.client),
                    crate::internal::transport::CoreHttpTransport::new(self.client),
                )
                .prepare_and_commit_object_async(
                    crate::internal::attachment_runtime::upload::AttachmentPrepareObjectInput {
                        target: resolved.request.target.clone(),
                        request: attachment_request_from_message_request(resolved.request.clone())?,
                        resolved_target_did: resolved.target_did.clone(),
                        message_security_profile: "direct-e2ee",
                    },
                )
                .await?;
            let async_result =
                crate::internal::secure_direct::async_send::AsyncDirectSecureTextSender::new(
                    self.client,
                    crate::internal::auth::session::FileSessionProvider::new(self.client),
                    crate::internal::transport::CoreHttpTransport::new(self.client),
                    crate::internal::transport::CoreHttpTransport::new(self.client),
                )
                .send_attachment_async_if_ready(
                    crate::internal::secure_direct::send::DirectSecureAttachmentSend {
                        request: resolved.request.clone(),
                        resolved_target_did: resolved.target_did.clone(),
                        target_handle: target_handle.clone(),
                        peer_scope: peer_scope.clone(),
                        committed: committed.clone(),
                        local_persistence:
                            crate::internal::secure_direct::send::DirectSecureLocalPersistence::Deferred,
                    },
                )
                .await?;
            if let Some(result) = async_result {
                return persist_deferred_direct_e2ee_attachment_effect(
                    self.client,
                    result,
                    target_handle.as_deref(),
                    peer_scope.as_ref(),
                )
                .await;
            }
            #[cfg(feature = "blocking")]
            {
                let client = self.client.clone();
                let result = crate::internal::runtime::worker::run_blocking(move || {
                    send_direct_e2ee_attachment_with_client_and_persistence(
                        &client,
                        resolved,
                        committed,
                        crate::internal::secure_direct::send::DirectSecureLocalPersistence::Deferred,
                    )
                })
                .await
                .map_err(|err| crate::ImError::Internal {
                    message: err.to_string(),
                })??;
                return persist_deferred_direct_e2ee_attachment_effect(
                    self.client,
                    result,
                    target_handle.as_deref(),
                    peer_scope.as_ref(),
                )
                .await;
            }
            #[cfg(not(feature = "blocking"))]
            {
                let _ = committed;
                return Err(crate::ImError::LocalStateUnavailable {
                    detail: "direct E2EE async attachment send requires an established local session; sync compatibility fallback is disabled".to_owned(),
                });
            }
        }
        let async_input = crate::internal::secure_direct::send::DirectSecureTextSend {
            request: resolved.request.clone(),
            resolved_target_did: resolved.target_did.clone(),
            target_handle: target_handle.clone(),
            peer_scope: peer_scope.clone(),
            local_persistence:
                crate::internal::secure_direct::send::DirectSecureLocalPersistence::Deferred,
        };
        match crate::internal::secure_direct::async_send::AsyncDirectSecureTextSender::new(
            self.client,
            crate::internal::auth::session::FileSessionProvider::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
        )
        .send_async_if_ready(async_input)
        .await?
        {
            crate::internal::secure_direct::async_send::AsyncDirectSecureSendOutcome::Sent(
                result,
            ) => {
                return persist_deferred_direct_e2ee_effect(
                    self.client,
                    result,
                    target_handle.as_deref(),
                    peer_scope.as_ref(),
                )
                .await
            }
            crate::internal::secure_direct::async_send::AsyncDirectSecureSendOutcome::Fallback(
                crate::internal::secure_direct::async_send::AsyncDirectSecureSendFallback::NoEstablishedSession,
            ) => {}
        }

        #[cfg(feature = "blocking")]
        {
            let client = self.client.clone();
            let result = crate::internal::runtime::worker::run_blocking(move || {
                send_direct_e2ee_with_client_and_persistence(
                    &client,
                    resolved,
                    crate::internal::secure_direct::send::DirectSecureLocalPersistence::Deferred,
                )
            })
            .await
            .map_err(|err| crate::ImError::Internal {
                message: err.to_string(),
            })??;
            persist_deferred_direct_e2ee_effect(
                self.client,
                result,
                target_handle.as_deref(),
                peer_scope.as_ref(),
            )
            .await
        }
        #[cfg(not(feature = "blocking"))]
        {
            let _ = resolved;
            Err(crate::ImError::LocalStateUnavailable {
                detail: "direct E2EE async send requires an established local session; sync compatibility fallback is disabled".to_owned(),
            })
        }
    }

    #[cfg(not(feature = "sqlite"))]
    async fn send_direct_e2ee_async(
        &self,
        resolved: ResolvedSendRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        let _ = resolved;
        Err(crate::ImError::unsupported("secure-direct"))
    }

    #[cfg(feature = "group-e2ee")]
    fn send_group_e2ee(
        &self,
        request: super::SendMessageRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        #[cfg(feature = "blocking")]
        {
            let session_provider =
                crate::internal::auth::session::FileSessionProvider::new(self.client);
            let mut transport = crate::internal::transport::CoreHttpTransport::new(self.client);
            crate::internal::group_e2ee::lifecycle::ensure_group_e2ee_service_available(
                self.client,
                &session_provider,
                &mut transport,
                crate::internal::group_e2ee::lifecycle::GroupE2eeServiceAvailabilityInput {
                    credentials: None,
                    service_did: None,
                    check_key_package: false,
                },
            )?;
            let provider =
                crate::internal::group_e2ee::storage::native_provider_for_client(self.client)?;
            if matches!(request.body, super::MessageBody::Attachment { .. }) {
                let committed =
                    crate::internal::attachment_runtime::upload::AttachmentUploadRuntime::new(
                        self.client,
                        crate::internal::auth::session::FileSessionProvider::new(self.client),
                        crate::internal::transport::CoreHttpTransport::new(self.client),
                    )
                    .prepare_and_commit_object(
                        crate::internal::attachment_runtime::upload::AttachmentPrepareObjectInput {
                            target: request.target.clone(),
                            request: attachment_request_from_message_request(request.clone())?,
                            resolved_target_did: None,
                            message_security_profile: "group-e2ee",
                        },
                    )?;
                return crate::internal::group_e2ee::runtime::GroupE2eeTextSender::new(
                    self.client,
                    crate::internal::auth::session::FileSessionProvider::new(self.client),
                    crate::internal::transport::CoreHttpTransport::new(self.client),
                    provider,
                )
                .send_attachment(
                    crate::internal::group_e2ee::runtime::GroupE2eeAttachmentSend {
                        request,
                        group_state_ref: None,
                        credentials: None,
                        committed,
                    },
                )
                .map(|result| result.sdk_result);
            }
            crate::internal::group_e2ee::runtime::GroupE2eeTextSender::new(
                self.client,
                crate::internal::auth::session::FileSessionProvider::new(self.client),
                crate::internal::transport::CoreHttpTransport::new(self.client),
                provider,
            )
            .send(crate::internal::group_e2ee::runtime::GroupE2eeTextSend {
                request,
                group_state_ref: None,
                credentials: None,
            })
            .map(|result| result.sdk_result)
        }
        #[cfg(not(feature = "blocking"))]
        {
            let _ = request;
            Err(crate::ImError::unsupported("sync-group-e2ee-send"))
        }
    }

    #[cfg(not(feature = "group-e2ee"))]
    fn send_group_e2ee(
        &self,
        _request: super::SendMessageRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        Err(crate::ImError::unsupported("group-e2ee"))
    }

    #[cfg(feature = "group-e2ee")]
    async fn send_group_e2ee_async(
        &self,
        request: super::SendMessageRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        let provider =
            crate::internal::group_e2ee::storage::native_provider_for_client(self.client)?;
        if matches!(request.body, super::MessageBody::Attachment { .. }) {
            let committed =
                crate::internal::attachment_runtime::upload::AttachmentUploadRuntime::new(
                    self.client,
                    crate::internal::auth::session::FileSessionProvider::new(self.client),
                    crate::internal::transport::CoreHttpTransport::new(self.client),
                )
                .prepare_and_commit_object_async(
                    crate::internal::attachment_runtime::upload::AttachmentPrepareObjectInput {
                        target: request.target.clone(),
                        request: attachment_request_from_message_request(request.clone())?,
                        resolved_target_did: None,
                        message_security_profile: "group-e2ee",
                    },
                )
                .await?;
            return crate::internal::group_e2ee::runtime::GroupE2eeTextSender::new(
                self.client,
                crate::internal::auth::session::FileSessionProvider::new(self.client),
                crate::internal::transport::CoreHttpTransport::new(self.client),
                provider,
            )
            .send_attachment_async(
                crate::internal::group_e2ee::runtime::GroupE2eeAttachmentSend {
                    request,
                    group_state_ref: None,
                    credentials: None,
                    committed,
                },
            )
            .await
            .map(|result| result.sdk_result);
        }
        match crate::internal::group_e2ee::runtime::GroupE2eeTextSender::new(
            self.client,
            crate::internal::auth::session::FileSessionProvider::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
            provider,
        )
        .send_async(crate::internal::group_e2ee::runtime::GroupE2eeTextSend {
            request: request.clone(),
            group_state_ref: None,
            credentials: None,
        })
        .await
        {
            Ok(result) => Ok(result.sdk_result),
            Err(err)
                if crate::internal::group_e2ee::runtime::is_group_e2ee_epoch_mismatch(&err) =>
            {
                #[cfg(feature = "blocking")]
                {
                    let client = self.client.clone();
                    crate::internal::runtime::worker::run_blocking(move || {
                        client.messages().send_group_e2ee(request)
                    })
                    .await
                    .map_err(|join| crate::ImError::Internal {
                        message: join.to_string(),
                    })?
                }
                #[cfg(not(feature = "blocking"))]
                {
                    let _ = request;
                    Err(err)
                }
            }
            Err(err) => Err(err),
        }
    }

    #[cfg(not(feature = "group-e2ee"))]
    async fn send_group_e2ee_async(
        &self,
        _request: super::SendMessageRequest,
    ) -> crate::ImResult<super::SendMessageResult> {
        Err(crate::ImError::unsupported("group-e2ee"))
    }

    pub fn inbox(
        &self,
        query: super::InboxQuery,
    ) -> crate::ImResult<crate::ids::Page<super::Message>> {
        self.inbox_with_metadata(query)
            .map(super::MessagePage::into_page)
    }

    pub async fn inbox_async(
        &self,
        query: super::InboxQuery,
    ) -> crate::ImResult<crate::ids::Page<super::Message>> {
        self.inbox_with_metadata_async(query)
            .await
            .map(super::MessagePage::into_page)
    }

    pub fn inbox_with_metadata(
        &self,
        query: super::InboxQuery,
    ) -> crate::ImResult<super::MessagePage> {
        crate::internal::message_runtime::read::MessageReadRuntime::new(
            self.client,
            crate::internal::auth::session::FileSessionProvider::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
        )
        .inbox(crate::internal::message_runtime::read::InboxRead { query })
        .map(message_page_from_read_result)?
    }

    pub async fn inbox_with_metadata_async(
        &self,
        query: super::InboxQuery,
    ) -> crate::ImResult<super::MessagePage> {
        crate::internal::message_runtime::read::MessageReadRuntime::new(
            self.client,
            crate::internal::auth::session::FileSessionProvider::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
        )
        .inbox_async(crate::internal::message_runtime::read::InboxRead { query })
        .await
        .map(message_page_from_read_result)?
    }

    pub fn history(
        &self,
        thread: super::ThreadRef,
        query: super::HistoryQuery,
    ) -> crate::ImResult<crate::ids::Page<super::Message>> {
        self.history_with_metadata(thread, query)
            .map(super::MessagePage::into_page)
    }

    pub async fn history_async(
        &self,
        thread: super::ThreadRef,
        query: super::HistoryQuery,
    ) -> crate::ImResult<crate::ids::Page<super::Message>> {
        self.history_with_metadata_async(thread, query)
            .await
            .map(super::MessagePage::into_page)
    }

    pub fn history_with_metadata(
        &self,
        thread: super::ThreadRef,
        query: super::HistoryQuery,
    ) -> crate::ImResult<super::MessagePage> {
        let resolved = resolve_history_thread(self.client, thread)?;
        crate::internal::message_runtime::read::MessageReadRuntime::new(
            self.client,
            crate::internal::auth::session::FileSessionProvider::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
        )
        .history(crate::internal::message_runtime::read::HistoryRead {
            thread: resolved.thread,
            query,
            resolved_peer_did: resolved.resolved_did.clone(),
            peer_scope: resolved.peer_scope.clone(),
        })
        .map(|result| {
            let mut page = message_page_from_read_result(result)?;
            #[cfg(feature = "sqlite")]
            if let Some((handle, current_did)) = resolved.handle_peer.as_ref() {
                if let Ok(dids) =
                    crate::internal::message_runtime::local_projection::peer_dids_for_handle(
                        self.client,
                        handle,
                        current_did,
                    )
                {
                    page.resolved_dids = dids
                        .into_iter()
                        .filter_map(|did| crate::ids::Did::parse(did).ok())
                        .collect();
                }
            }
            if page.resolved_dids.is_empty() {
                if let Some(did) = resolved.resolved_did {
                    if let Ok(did) = crate::ids::Did::parse(did) {
                        page.resolved_dids.push(did);
                    }
                }
            }
            Ok(page)
        })?
    }

    pub async fn history_with_metadata_async(
        &self,
        thread: super::ThreadRef,
        query: super::HistoryQuery,
    ) -> crate::ImResult<super::MessagePage> {
        let resolved = resolve_history_thread_async(self.client, thread).await?;
        let mut page = crate::internal::message_runtime::read::MessageReadRuntime::new(
            self.client,
            crate::internal::auth::session::FileSessionProvider::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
        )
        .history_async(crate::internal::message_runtime::read::HistoryRead {
            thread: resolved.thread,
            query,
            resolved_peer_did: resolved.resolved_did.clone(),
            peer_scope: resolved.peer_scope.clone(),
        })
        .await
        .map(message_page_from_read_result)??;
        #[cfg(feature = "sqlite")]
        if let Some((handle, current_did)) = resolved.handle_peer.as_ref() {
            if let Ok(dids) =
                crate::internal::message_runtime::local_projection::peer_dids_for_handle_async(
                    self.client,
                    handle,
                    current_did,
                )
                .await
            {
                page.resolved_dids = dids
                    .into_iter()
                    .filter_map(|did| crate::ids::Did::parse(did).ok())
                    .collect();
            }
        }
        if page.resolved_dids.is_empty() {
            if let Some(did) = resolved.resolved_did {
                if let Ok(did) = crate::ids::Did::parse(did) {
                    page.resolved_dids.push(did);
                }
            }
        }
        Ok(page)
    }

    pub fn local_history(
        &self,
        thread: super::ThreadRef,
        query: super::LocalHistoryQuery,
    ) -> crate::ImResult<crate::ids::Page<super::Message>> {
        self.local_history_with_metadata(thread, query)
            .map(super::MessagePage::into_page)
    }

    pub async fn local_history_async(
        &self,
        thread: super::ThreadRef,
        query: super::LocalHistoryQuery,
    ) -> crate::ImResult<crate::ids::Page<super::Message>> {
        self.local_history_with_metadata_async(thread, query)
            .await
            .map(super::MessagePage::into_page)
    }

    pub fn local_history_with_metadata(
        &self,
        thread: super::ThreadRef,
        query: super::LocalHistoryQuery,
    ) -> crate::ImResult<super::MessagePage> {
        crate::internal::message_runtime::read::MessageReadRuntime::new(
            self.client,
            crate::internal::auth::session::FileSessionProvider::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
        )
        .local_history(crate::internal::message_runtime::read::LocalHistoryRead { thread, query })
        .map(message_page_from_read_result)?
    }

    pub async fn local_history_with_metadata_async(
        &self,
        thread: super::ThreadRef,
        query: super::LocalHistoryQuery,
    ) -> crate::ImResult<super::MessagePage> {
        crate::internal::message_runtime::read::MessageReadRuntime::new(
            self.client,
            crate::internal::auth::session::FileSessionProvider::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
        )
        .local_history_async(crate::internal::message_runtime::read::LocalHistoryRead {
            thread,
            query,
        })
        .await
        .map(message_page_from_read_result)?
    }

    pub fn local_conversation_timeline(
        &self,
        conversation: super::ConversationReadRef,
        query: super::LocalHistoryQuery,
    ) -> crate::ImResult<crate::ids::Page<super::Message>> {
        self.local_conversation_timeline_with_metadata(conversation, query)
            .map(super::MessagePage::into_page)
    }

    pub async fn local_conversation_timeline_async(
        &self,
        conversation: super::ConversationReadRef,
        query: super::LocalHistoryQuery,
    ) -> crate::ImResult<crate::ids::Page<super::Message>> {
        self.local_conversation_timeline_with_metadata_async(conversation, query)
            .await
            .map(super::MessagePage::into_page)
    }

    pub fn local_conversation_timeline_with_metadata(
        &self,
        conversation: super::ConversationReadRef,
        query: super::LocalHistoryQuery,
    ) -> crate::ImResult<super::MessagePage> {
        self.local_history_with_metadata(conversation.as_thread_ref()?, query)
    }

    pub async fn local_conversation_timeline_with_metadata_async(
        &self,
        conversation: super::ConversationReadRef,
        query: super::LocalHistoryQuery,
    ) -> crate::ImResult<super::MessagePage> {
        self.local_history_with_metadata_async(conversation.as_thread_ref()?, query)
            .await
    }

    pub fn mark_read(
        &self,
        ids: Vec<crate::ids::MessageId>,
    ) -> crate::ImResult<super::MarkReadResult> {
        #[cfg(feature = "blocking")]
        {
            crate::internal::message_runtime::mark_read::MessageMarkReadRuntime::new(
                self.client,
                crate::internal::auth::session::FileSessionProvider::new(self.client),
                crate::internal::transport::CoreHttpTransport::new(self.client),
            )
            .mark_read(crate::internal::message_runtime::mark_read::MarkReadInput {
                message_ids: ids,
            })
            .map(|result| result.sdk_result)
        }
        #[cfg(not(feature = "blocking"))]
        {
            let _ = ids;
            Err(crate::ImError::unsupported("sync-message-mark-read"))
        }
    }

    pub async fn mark_read_async(
        &self,
        ids: Vec<crate::ids::MessageId>,
    ) -> crate::ImResult<super::MarkReadResult> {
        crate::internal::message_runtime::mark_read::MessageMarkReadRuntime::new(
            self.client,
            crate::internal::auth::session::FileSessionProvider::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
        )
        .mark_read_async(crate::internal::message_runtime::mark_read::MarkReadInput {
            message_ids: ids,
        })
        .await
        .map(|result| result.sdk_result)
    }

    pub fn mark_thread_read(
        &self,
        request: super::MarkThreadReadRequest,
    ) -> crate::ImResult<super::MarkThreadReadResult> {
        #[cfg(feature = "blocking")]
        {
            crate::internal::message_runtime::mark_read::MessageMarkReadRuntime::new(
                self.client,
                crate::internal::auth::session::FileSessionProvider::new(self.client),
                crate::internal::transport::CoreHttpTransport::new(self.client),
            )
            .mark_thread_read(
                crate::internal::message_runtime::mark_read::MarkThreadReadInput {
                    request,
                    remote_thread: None,
                },
            )
            .map(|result| result.sdk_result)
        }
        #[cfg(not(feature = "blocking"))]
        {
            let _ = request;
            Err(crate::ImError::unsupported("sync-message-mark-thread-read"))
        }
    }

    pub async fn mark_thread_read_async(
        &self,
        request: super::MarkThreadReadRequest,
    ) -> crate::ImResult<super::MarkThreadReadResult> {
        crate::internal::message_runtime::mark_read::MessageMarkReadRuntime::new(
            self.client,
            crate::internal::auth::session::FileSessionProvider::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
        )
        .mark_thread_read_async(
            crate::internal::message_runtime::mark_read::MarkThreadReadInput {
                request,
                remote_thread: None,
            },
        )
        .await
        .map(|result| result.sdk_result)
    }

    pub fn mark_conversation_read(
        &self,
        request: super::MarkConversationReadRequest,
    ) -> crate::ImResult<super::MarkThreadReadResult> {
        #[cfg(feature = "blocking")]
        {
            let mapped = mark_read_input_for_conversation(self.client, request)?;
            crate::internal::message_runtime::mark_read::MessageMarkReadRuntime::new(
                self.client,
                crate::internal::auth::session::FileSessionProvider::new(self.client),
                crate::internal::transport::CoreHttpTransport::new(self.client),
            )
            .mark_thread_read(mapped)
            .map(|result| result.sdk_result)
        }
        #[cfg(not(feature = "blocking"))]
        {
            let _ = request;
            Err(crate::ImError::unsupported("sync-message-mark-thread-read"))
        }
    }

    pub async fn mark_conversation_read_async(
        &self,
        request: super::MarkConversationReadRequest,
    ) -> crate::ImResult<super::MarkThreadReadResult> {
        let mapped = mark_read_input_for_conversation_async(self.client, request).await?;
        crate::internal::message_runtime::mark_read::MessageMarkReadRuntime::new(
            self.client,
            crate::internal::auth::session::FileSessionProvider::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
        )
        .mark_thread_read_async(mapped)
        .await
        .map(|result| result.sdk_result)
    }

    pub fn sync_thread_after(
        &self,
        request: super::SyncThreadAfterRequest,
    ) -> crate::ImResult<super::SyncThreadAfterResult> {
        #[cfg(feature = "blocking")]
        {
            let resolved = resolve_history_thread(self.client, request.thread.clone())?;
            crate::internal::message_runtime::sync::MessageSyncRuntime::new(
                self.client,
                crate::internal::auth::session::FileSessionProvider::new(self.client),
                crate::internal::transport::CoreHttpTransport::new(self.client),
                crate::internal::transport::CoreHttpTransport::new(self.client),
            )
            .sync_thread_after(
                crate::internal::message_runtime::sync::SyncThreadAfterInput {
                    request: super::SyncThreadAfterRequest {
                        thread: resolved.thread,
                        ..request
                    },
                    resolved_peer_did: resolved.resolved_did,
                    peer_scope: resolved.peer_scope,
                },
            )
        }
        #[cfg(not(feature = "blocking"))]
        {
            let _ = request;
            Err(crate::ImError::unsupported("sync-thread-after"))
        }
    }

    pub async fn sync_thread_after_async(
        &self,
        request: super::SyncThreadAfterRequest,
    ) -> crate::ImResult<super::SyncThreadAfterResult> {
        let resolved = resolve_history_thread_async(self.client, request.thread.clone()).await?;
        crate::internal::message_runtime::sync::MessageSyncRuntime::new(
            self.client,
            crate::internal::auth::session::FileSessionProvider::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
        )
        .sync_thread_after_async(
            crate::internal::message_runtime::sync::SyncThreadAfterInput {
                request: super::SyncThreadAfterRequest {
                    thread: resolved.thread,
                    ..request
                },
                resolved_peer_did: resolved.resolved_did,
                peer_scope: resolved.peer_scope,
            },
        )
        .await
    }

    pub fn sync_conversation_after(
        &self,
        request: super::SyncConversationAfterRequest,
    ) -> crate::ImResult<super::SyncThreadAfterResult> {
        #[cfg(feature = "blocking")]
        {
            let resolved = resolve_service_conversation_thread(self.client, &request.conversation)?;
            crate::internal::message_runtime::sync::MessageSyncRuntime::new(
                self.client,
                crate::internal::auth::session::FileSessionProvider::new(self.client),
                crate::internal::transport::CoreHttpTransport::new(self.client),
                crate::internal::transport::CoreHttpTransport::new(self.client),
            )
            .sync_thread_after(
                crate::internal::message_runtime::sync::SyncThreadAfterInput {
                    request: super::SyncThreadAfterRequest {
                        thread: resolved.thread,
                        after_server_seq: request.after_server_seq,
                        limit: request.limit,
                    },
                    resolved_peer_did: resolved.resolved_did,
                    peer_scope: resolved.peer_scope,
                },
            )
        }
        #[cfg(not(feature = "blocking"))]
        {
            let _ = request;
            Err(crate::ImError::unsupported("sync-conversation-after"))
        }
    }

    pub async fn sync_conversation_after_async(
        &self,
        request: super::SyncConversationAfterRequest,
    ) -> crate::ImResult<super::SyncThreadAfterResult> {
        let resolved =
            resolve_service_conversation_thread_async(self.client, &request.conversation).await?;
        crate::internal::message_runtime::sync::MessageSyncRuntime::new(
            self.client,
            crate::internal::auth::session::FileSessionProvider::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
        )
        .sync_thread_after_async(
            crate::internal::message_runtime::sync::SyncThreadAfterInput {
                request: super::SyncThreadAfterRequest {
                    thread: resolved.thread,
                    after_server_seq: request.after_server_seq,
                    limit: request.limit,
                },
                resolved_peer_did: resolved.resolved_did,
                peer_scope: resolved.peer_scope,
            },
        )
        .await
    }

    pub fn sync_delta(
        &self,
        request: super::SyncDeltaRequest,
    ) -> crate::ImResult<super::SyncDeltaResult> {
        #[cfg(feature = "blocking")]
        {
            crate::internal::message_runtime::sync::MessageSyncRuntime::new(
                self.client,
                crate::internal::auth::session::FileSessionProvider::new(self.client),
                crate::internal::transport::CoreHttpTransport::new(self.client),
                crate::internal::transport::CoreHttpTransport::new(self.client),
            )
            .sync_delta(crate::internal::message_runtime::sync::SyncDeltaInput { request })
        }
        #[cfg(not(feature = "blocking"))]
        {
            let _ = request;
            Err(crate::ImError::unsupported("sync-delta"))
        }
    }

    pub async fn sync_delta_async(
        &self,
        request: super::SyncDeltaRequest,
    ) -> crate::ImResult<super::SyncDeltaResult> {
        crate::internal::message_runtime::sync::MessageSyncRuntime::new(
            self.client,
            crate::internal::auth::session::FileSessionProvider::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
            crate::internal::transport::CoreHttpTransport::new(self.client),
        )
        .sync_delta_async(crate::internal::message_runtime::sync::SyncDeltaInput { request })
        .await
    }

    pub fn conversations(
        &self,
        query: super::ConversationQuery,
    ) -> crate::ImResult<crate::ids::Page<super::Conversation>> {
        #[cfg(feature = "blocking")]
        {
            crate::internal::message_runtime::conversations::MessageConversationRuntime::new(
                self.client,
            )
            .conversations(query)
        }
        #[cfg(not(feature = "blocking"))]
        {
            let _ = query;
            Err(crate::ImError::unsupported("sync-message-conversations"))
        }
    }

    pub async fn conversations_async(
        &self,
        query: super::ConversationQuery,
    ) -> crate::ImResult<crate::ids::Page<super::Conversation>> {
        crate::internal::message_runtime::conversations::MessageConversationRuntime::new(
            self.client,
        )
        .conversations_async(query)
        .await
    }

    pub fn load_conversation_snapshot(
        &self,
    ) -> crate::ImResult<Option<super::ConversationListSnapshot>> {
        crate::internal::snapshot::conversation_snapshot::load_for_client(self.client)
    }

    pub async fn load_conversation_snapshot_async(
        &self,
    ) -> crate::ImResult<Option<super::ConversationListSnapshot>> {
        self.load_conversation_snapshot()
    }

    pub fn clear_conversation_snapshot(&self) -> crate::ImResult<()> {
        crate::internal::snapshot::conversation_snapshot::clear_for_client(self.client)
    }

    pub async fn clear_conversation_snapshot_async(&self) -> crate::ImResult<()> {
        self.clear_conversation_snapshot()
    }

    pub fn watch_conversation_patches(&self) -> crate::ImResult<super::ConversationPatchSession> {
        self.client
            .conversation_store()
            .watch_for_client(self.client)
    }

    pub async fn watch_conversation_patches_async(
        &self,
    ) -> crate::ImResult<super::ConversationPatchSession> {
        self.watch_conversation_patches()
    }

    pub fn repair_conversation_store(&self) -> crate::ImResult<super::ConversationStorePatch> {
        self.client
            .conversation_store()
            .repair_from_client(self.client)
    }

    pub async fn repair_conversation_store_async(
        &self,
    ) -> crate::ImResult<super::ConversationStorePatch> {
        self.repair_conversation_store()
    }

    pub fn watch_thread_patches(
        &self,
        thread: super::ThreadRef,
        limit: Option<u32>,
    ) -> crate::ImResult<super::ThreadMessagePatchSession> {
        self.client
            .message_store()
            .watch_for_client(self.client, thread, limit)
    }

    pub async fn watch_thread_patches_async(
        &self,
        thread: super::ThreadRef,
        limit: Option<u32>,
    ) -> crate::ImResult<super::ThreadMessagePatchSession> {
        self.watch_thread_patches(thread, limit)
    }

    pub fn watch_conversation_timeline_patches(
        &self,
        conversation: super::ConversationReadRef,
        limit: Option<u32>,
    ) -> crate::ImResult<super::ThreadMessagePatchSession> {
        self.watch_thread_patches(conversation.as_thread_ref()?, limit)
    }

    pub async fn watch_conversation_timeline_patches_async(
        &self,
        conversation: super::ConversationReadRef,
        limit: Option<u32>,
    ) -> crate::ImResult<super::ThreadMessagePatchSession> {
        self.watch_conversation_timeline_patches(conversation, limit)
    }

    pub fn repair_thread_store(
        &self,
        thread: super::ThreadRef,
        limit: Option<u32>,
    ) -> crate::ImResult<super::ThreadMessageStorePatch> {
        self.client.message_store().repair_thread_from_client(
            self.client,
            thread,
            limit.unwrap_or(100),
        )
    }

    pub async fn repair_thread_store_async(
        &self,
        thread: super::ThreadRef,
        limit: Option<u32>,
    ) -> crate::ImResult<super::ThreadMessageStorePatch> {
        self.repair_thread_store(thread, limit)
    }

    pub fn repair_conversation_timeline_store(
        &self,
        conversation: super::ConversationReadRef,
        limit: Option<u32>,
    ) -> crate::ImResult<super::ThreadMessageStorePatch> {
        self.repair_thread_store(conversation.as_thread_ref()?, limit)
    }

    pub async fn repair_conversation_timeline_store_async(
        &self,
        conversation: super::ConversationReadRef,
        limit: Option<u32>,
    ) -> crate::ImResult<super::ThreadMessageStorePatch> {
        self.repair_conversation_timeline_store(conversation, limit)
    }
}

#[cfg(all(feature = "sqlite", feature = "blocking"))]
fn send_direct_e2ee_with_client(
    client: &crate::core::ImClient,
    resolved: ResolvedSendRequest,
) -> crate::ImResult<super::SendMessageResult> {
    send_direct_e2ee_with_client_attachment_aware(
        client,
        resolved,
        crate::internal::secure_direct::send::DirectSecureLocalPersistence::LegacySqlite,
    )
}

#[cfg(all(feature = "sqlite", feature = "blocking"))]
fn send_direct_e2ee_with_client_and_persistence(
    client: &crate::core::ImClient,
    resolved: ResolvedSendRequest,
    local_persistence: crate::internal::secure_direct::send::DirectSecureLocalPersistence,
) -> crate::ImResult<crate::internal::secure_direct::send::DirectSecureTextSendResult> {
    if matches!(resolved.request.body, super::MessageBody::Attachment { .. }) {
        return Err(crate::ImError::unsupported(
            "direct-e2ee-attachment-text-result",
        ));
    }
    let target_handle = resolved.direct_handle().map(str::to_owned);
    let peer_scope = resolved.peer_scope.clone();
    let mut result = crate::internal::secure_direct::send::DirectSecureTextSender::new(
        client,
        crate::internal::auth::session::FileSessionProvider::new(client),
        crate::internal::transport::CoreHttpTransport::new(client),
        crate::internal::transport::CoreHttpTransport::new(client),
    )
    .send(crate::internal::secure_direct::send::DirectSecureTextSend {
        request: resolved.request,
        resolved_target_did: resolved.target_did,
        target_handle: target_handle.clone(),
        peer_scope: peer_scope.clone(),
        local_persistence,
    })?;
    apply_peer_scope_to_direct_secure_text_result(
        &mut result,
        target_handle.as_deref(),
        peer_scope.as_ref(),
    )?;
    Ok(result)
}

#[cfg(all(feature = "sqlite", feature = "blocking"))]
fn send_direct_e2ee_with_client_attachment_aware(
    client: &crate::core::ImClient,
    resolved: ResolvedSendRequest,
    local_persistence: crate::internal::secure_direct::send::DirectSecureLocalPersistence,
) -> crate::ImResult<super::SendMessageResult> {
    if matches!(resolved.request.body, super::MessageBody::Attachment { .. }) {
        let committed = crate::internal::attachment_runtime::upload::AttachmentUploadRuntime::new(
            client,
            crate::internal::auth::session::FileSessionProvider::new(client),
            crate::internal::transport::CoreHttpTransport::new(client),
        )
        .prepare_and_commit_object(
            crate::internal::attachment_runtime::upload::AttachmentPrepareObjectInput {
                target: resolved.request.target.clone(),
                request: attachment_request_from_message_request(resolved.request.clone())?,
                resolved_target_did: resolved.target_did.clone(),
                message_security_profile: "direct-e2ee",
            },
        )?;
        return send_direct_e2ee_attachment_with_client_and_persistence(
            client,
            resolved,
            committed,
            local_persistence,
        )
        .map(|result| result.sdk_result);
    }
    send_direct_e2ee_with_client_and_persistence(client, resolved, local_persistence)
        .map(|result| result.sdk_result)
}

#[cfg(all(feature = "sqlite", feature = "blocking"))]
fn send_direct_e2ee_attachment_with_client_and_persistence(
    client: &crate::core::ImClient,
    resolved: ResolvedSendRequest,
    committed: crate::internal::attachment_runtime::upload::PreparedCommittedAttachment,
    local_persistence: crate::internal::secure_direct::send::DirectSecureLocalPersistence,
) -> crate::ImResult<crate::internal::secure_direct::send::DirectSecureAttachmentSendResult> {
    let target_handle = resolved.direct_handle().map(str::to_owned);
    let peer_scope = resolved.peer_scope.clone();
    let mut result = crate::internal::secure_direct::send::DirectSecureTextSender::new(
        client,
        crate::internal::auth::session::FileSessionProvider::new(client),
        crate::internal::transport::CoreHttpTransport::new(client),
        crate::internal::transport::CoreHttpTransport::new(client),
    )
    .send_attachment(
        crate::internal::secure_direct::send::DirectSecureAttachmentSend {
            request: resolved.request,
            resolved_target_did: resolved.target_did,
            target_handle: target_handle.clone(),
            peer_scope: peer_scope.clone(),
            committed,
            local_persistence,
        },
    )?;
    apply_peer_scope_to_direct_secure_attachment_result(
        &mut result,
        target_handle.as_deref(),
        peer_scope.as_ref(),
    )?;
    Ok(result)
}

#[cfg(feature = "sqlite")]
async fn persist_deferred_direct_e2ee_effect(
    client: &crate::core::ImClient,
    mut result: crate::internal::secure_direct::send::DirectSecureTextSendResult,
    target_handle: Option<&str>,
    peer_scope: Option<&crate::internal::local_state::owner_scope::DirectPeerScope>,
) -> crate::ImResult<super::SendMessageResult> {
    apply_peer_scope_to_direct_secure_text_result(&mut result, target_handle, peer_scope)?;
    match result.local_effect {
        crate::internal::secure_direct::send::DirectSecureLocalEffect::None => {}
        crate::internal::secure_direct::send::DirectSecureLocalEffect::PersistOutgoing => {
            match
                crate::internal::message_runtime::local_projection::persist_direct_e2ee_outgoing_async(
                    client,
                    &result.target_did,
                    result.target_handle.as_deref(),
                    result.peer_scope.as_ref(),
                    &result.text,
                    &result.kind,
                    &result.sdk_result,
                )
                .await
            {
                Ok(()) => client.emit_committed_local_message_projection("local_send"),
                Err(err) => {
                    result
                        .sdk_result
                        .warnings
                        .push(format!("Failed to persist local secure direct message: {err}"));
                }
            }
        }
        crate::internal::secure_direct::send::DirectSecureLocalEffect::QueueOutbox(record) => {
            client
                .core_inner()
                .local_state_db()
                .await?
                .queue_e2ee_outbox(record)
                .await?;
        }
    }
    Ok(result.sdk_result)
}

#[cfg(feature = "sqlite")]
async fn persist_deferred_direct_e2ee_attachment_effect(
    client: &crate::core::ImClient,
    mut result: crate::internal::secure_direct::send::DirectSecureAttachmentSendResult,
    target_handle: Option<&str>,
    peer_scope: Option<&crate::internal::local_state::owner_scope::DirectPeerScope>,
) -> crate::ImResult<super::SendMessageResult> {
    apply_peer_scope_to_direct_secure_attachment_result(&mut result, target_handle, peer_scope)?;
    match result.local_effect {
        crate::internal::secure_direct::send::DirectSecureAttachmentLocalEffect::None => {}
        crate::internal::secure_direct::send::DirectSecureAttachmentLocalEffect::PersistOutgoing => {
            match
                crate::internal::message_runtime::local_projection::persist_direct_e2ee_attachment_outgoing_async(
                    client,
                    &result.target_did,
                    result.target_handle.as_deref(),
                    result.peer_scope.as_ref(),
                    &result.redacted_manifest,
                    &result.sdk_result,
                )
                .await
            {
                Ok(()) => client.emit_committed_local_message_projection("local_send"),
                Err(err) => {
                    result
                        .sdk_result
                        .warnings
                        .push(format!("Failed to persist local secure direct attachment: {err}"));
                }
            }
        }
    }
    Ok(result.sdk_result)
}

fn apply_peer_scope_to_direct_secure_text_result(
    result: &mut crate::internal::secure_direct::send::DirectSecureTextSendResult,
    target_handle: Option<&str>,
    peer_scope: Option<&crate::internal::local_state::owner_scope::DirectPeerScope>,
) -> crate::ImResult<()> {
    let handle = target_handle
        .map(str::trim)
        .filter(|value| !value.is_empty());
    if result.target_handle.is_none() {
        result.target_handle = handle.map(str::to_owned);
    }
    if result.peer_scope.is_none() {
        result.peer_scope = peer_scope.cloned();
    }
    normalize_direct_send_result_for_peer_scope(
        &mut result.sdk_result,
        result.peer_scope.as_ref(),
        result.target_handle.as_deref(),
        Some(result.target_did.as_str()),
    )
}

fn apply_peer_scope_to_direct_secure_attachment_result(
    result: &mut crate::internal::secure_direct::send::DirectSecureAttachmentSendResult,
    target_handle: Option<&str>,
    peer_scope: Option<&crate::internal::local_state::owner_scope::DirectPeerScope>,
) -> crate::ImResult<()> {
    let handle = target_handle
        .map(str::trim)
        .filter(|value| !value.is_empty());
    if result.target_handle.is_none() {
        result.target_handle = handle.map(str::to_owned);
    }
    if result.peer_scope.is_none() {
        result.peer_scope = peer_scope.cloned();
    }
    normalize_direct_send_result_for_peer_scope(
        &mut result.sdk_result,
        result.peer_scope.as_ref(),
        result.target_handle.as_deref(),
        Some(result.target_did.as_str()),
    )
}

#[derive(Clone)]
struct ResolvedSendRequest {
    request: super::SendMessageRequest,
    target_did: Option<String>,
    peer_scope: Option<crate::internal::local_state::owner_scope::DirectPeerScope>,
}

#[derive(Clone)]
struct ResolvedConversationSendRequest {
    request: super::SendMessageRequest,
    target_did: Option<String>,
    target_handle: Option<String>,
    peer_scope: Option<crate::internal::local_state::owner_scope::DirectPeerScope>,
}

impl ResolvedSendRequest {
    fn direct_handle(&self) -> Option<&str> {
        self.peer_scope
            .as_ref()
            .map(|scope| scope.full_handle.as_str())
            .or_else(|| direct_handle_from_target(&self.request.target))
    }
}

pub(crate) fn normalize_direct_send_result_for_peer_scope(
    result: &mut super::SendMessageResult,
    peer_scope: Option<&crate::internal::local_state::owner_scope::DirectPeerScope>,
    target_handle: Option<&str>,
    target_did: Option<&str>,
) -> crate::ImResult<()> {
    let Some(scope) = peer_scope else {
        return Ok(());
    };
    let thread_id = crate::messages::direct_peer_scope_thread_id(
        scope.user_id.as_str(),
        scope.full_handle.as_str(),
    )?;
    result.message.thread = super::ThreadRef::Thread(thread_id);
    upsert_message_attribute(
        &mut result.message.metadata.attributes,
        "peer_user_id",
        scope.user_id.as_str(),
    );
    upsert_message_attribute(
        &mut result.message.metadata.attributes,
        "peer_full_handle",
        scope.full_handle.as_str(),
    );
    if let Some(handle) = target_handle
        .map(str::trim)
        .filter(|value| !value.is_empty())
    {
        upsert_message_attribute(
            &mut result.message.metadata.attributes,
            "target_handle",
            handle,
        );
    }
    if let Some(did) = target_did.map(str::trim).filter(|value| !value.is_empty()) {
        upsert_message_attribute(
            &mut result.message.metadata.attributes,
            "resolved_target_did",
            did,
        );
        upsert_message_attribute(
            &mut result.message.metadata.attributes,
            "peer_current_did",
            did,
        );
    }
    Ok(())
}

fn conversation_send_request(
    client: &crate::core::ImClient,
    conversation: super::ConversationReadRef,
    body: super::MessageBody,
    security: super::MessageSecurityMode,
    client_message_id: Option<crate::ids::MessageId>,
    idempotency_key: Option<String>,
    wait_for_final_acceptance: bool,
    delegated_signing: Option<super::DelegatedSigningOptions>,
) -> crate::ImResult<ResolvedConversationSendRequest> {
    let resolved = resolve_service_conversation_thread(client, &conversation)?;
    let client_message_id = match client_message_id {
        Some(id) => id,
        None => crate::ids::MessageId::parse(format!(
            "msg-{}",
            crate::internal::wire::common::generate_operation_id()
        ))?,
    };
    let idempotency_key = idempotency_key
        .map(|value| value.trim().to_owned())
        .filter(|value| !value.is_empty())
        .or_else(|| Some(format!("op-{}", client_message_id.as_str())));
    let resolved_target = conversation_send_target(resolved)?;
    let request = super::SendMessageRequest {
        target: resolved_target.target,
        body,
        security,
        client_message_id: Some(client_message_id),
        delivery: super::MessageDeliveryOptions {
            idempotency_key,
            wait_for_final_acceptance,
        },
        delegated_signing,
    };
    Ok(ResolvedConversationSendRequest {
        request,
        target_did: resolved_target.target_did,
        target_handle: resolved_target.target_handle,
        peer_scope: resolved_target.peer_scope,
    })
}

fn conversation_send_target(
    resolved: ResolvedConversationServiceThread,
) -> crate::ImResult<ResolvedConversationSendTarget> {
    match resolved.thread {
        super::ThreadRef::Direct(peer) => {
            let target_did = resolved.resolved_did.clone().or_else(|| {
                peer.as_str()
                    .starts_with("did:")
                    .then(|| peer.as_str().to_owned())
            });
            let target_handle = resolved
                .peer_scope
                .as_ref()
                .map(|scope| scope.full_handle.clone());
            let target_peer = target_handle
                .as_deref()
                .map(|handle| crate::ids::PeerRef::parse(handle, ""))
                .transpose()?
                .unwrap_or(peer);
            Ok(ResolvedConversationSendTarget {
                target: super::MessageTarget::Direct(target_peer),
                target_did,
                target_handle,
                peer_scope: resolved.peer_scope,
            })
        }
        super::ThreadRef::Group(group) => Ok(ResolvedConversationSendTarget {
            target: super::MessageTarget::Group(group),
            target_did: None,
            target_handle: None,
            peer_scope: None,
        }),
        super::ThreadRef::Thread(_) => Err(crate::ImError::invalid_input(
            Some("conversation_id".to_owned()),
            "send_conversation requires a canonical dm: or group: conversation_id",
        )),
    }
}

pub(crate) fn resolve_conversation_send_target(
    client: &crate::core::ImClient,
    conversation: &super::ConversationReadRef,
) -> crate::ImResult<ResolvedConversationSendTarget> {
    conversation_send_target(resolve_service_conversation_thread(client, conversation)?)
}

fn validate_plain_conversation_send(request: &super::SendMessageRequest) -> crate::ImResult<()> {
    validate_body(&request.body)?;
    if matches!(request.body, super::MessageBody::Attachment { .. }) {
        return Err(crate::ImError::unsupported("conversation-attachment-send"));
    }
    match request.security {
        super::MessageSecurityMode::DefaultPlain | super::MessageSecurityMode::Plain => {}
        super::MessageSecurityMode::E2eeRequired
        | super::MessageSecurityMode::SecureDirect
        | super::MessageSecurityMode::GroupE2ee => {
            return Err(crate::ImError::unsupported(
                "secure-conversation-send-local-echo",
            ));
        }
    }
    validate_send_mode(&request.target, &request.security)?;
    validate_delegated_send_scope(request)
}

fn normalize_resolved_direct_send_result(
    result: &mut super::SendMessageResult,
    resolved: &ResolvedSendRequest,
) -> crate::ImResult<()> {
    normalize_direct_send_result_for_peer_scope(
        result,
        resolved.peer_scope.as_ref(),
        resolved.direct_handle(),
        resolved.target_did.as_deref(),
    )
}

fn upsert_message_attribute(
    attributes: &mut Vec<super::MessageMetadataAttribute>,
    key: &str,
    value: &str,
) {
    let value = value.trim();
    if value.is_empty() {
        return;
    }
    if let Some(attribute) = attributes.iter_mut().find(|attribute| attribute.key == key) {
        attribute.value = value.to_owned();
        return;
    }
    attributes.push(super::MessageMetadataAttribute {
        key: key.to_owned(),
        value: value.to_owned(),
    });
}

fn resolve_send_request(
    client: &crate::core::ImClient,
    request: super::SendMessageRequest,
) -> crate::ImResult<ResolvedSendRequest> {
    let (target_did, peer_scope) = match &request.target {
        super::MessageTarget::Direct(peer) => {
            let resolved = resolve_direct_peer(client, peer)?;
            (resolved.target_did, resolved.peer_scope)
        }
        super::MessageTarget::Group(_) => (None, None),
    };
    Ok(ResolvedSendRequest {
        request,
        target_did,
        peer_scope,
    })
}

async fn resolve_send_request_async(
    client: &crate::core::ImClient,
    request: super::SendMessageRequest,
) -> crate::ImResult<ResolvedSendRequest> {
    let (target_did, peer_scope) = match &request.target {
        super::MessageTarget::Direct(peer) => {
            let resolved = resolve_direct_peer_async(client, peer).await?;
            (resolved.target_did, resolved.peer_scope)
        }
        super::MessageTarget::Group(_) => (None, None),
    };
    Ok(ResolvedSendRequest {
        request,
        target_did,
        peer_scope,
    })
}

struct ResolvedDirectPeer {
    target_did: Option<String>,
    peer_scope: Option<crate::internal::local_state::owner_scope::DirectPeerScope>,
}

fn resolve_direct_peer(
    client: &crate::core::ImClient,
    peer: &crate::ids::PeerRef,
) -> crate::ImResult<ResolvedDirectPeer> {
    let raw = peer.as_str().trim();
    if raw.is_empty() || raw.starts_with("did:") {
        return Ok(ResolvedDirectPeer {
            target_did: None,
            peer_scope: None,
        });
    }
    let lookup = crate::internal::handle_discovery::resolve_direct_handle(client, raw)?;
    Ok(ResolvedDirectPeer {
        target_did: Some(lookup.target_did),
        peer_scope: Some(
            crate::internal::local_state::owner_scope::DirectPeerScope::new(
                lookup.user_id,
                lookup.full_handle,
            )?,
        ),
    })
}

async fn resolve_direct_peer_async(
    client: &crate::core::ImClient,
    peer: &crate::ids::PeerRef,
) -> crate::ImResult<ResolvedDirectPeer> {
    let raw = peer.as_str().trim();
    if raw.is_empty() || raw.starts_with("did:") {
        return Ok(ResolvedDirectPeer {
            target_did: None,
            peer_scope: None,
        });
    }
    let lookup =
        crate::internal::handle_discovery::resolve_direct_handle_async(client, raw).await?;
    Ok(ResolvedDirectPeer {
        target_did: Some(lookup.target_did),
        peer_scope: Some(
            crate::internal::local_state::owner_scope::DirectPeerScope::new(
                lookup.user_id,
                lookup.full_handle,
            )?,
        ),
    })
}

fn validate_send_mode(
    target: &super::MessageTarget,
    security: &super::MessageSecurityMode,
) -> crate::ImResult<()> {
    match (target, security) {
        (_, super::MessageSecurityMode::DefaultPlain | super::MessageSecurityMode::Plain) => Ok(()),
        (
            super::MessageTarget::Direct(_),
            super::MessageSecurityMode::E2eeRequired | super::MessageSecurityMode::SecureDirect,
        ) => Ok(()),
        (super::MessageTarget::Group(_), super::MessageSecurityMode::SecureDirect) => {
            Err(crate::ImError::unsupported("secure-direct"))
        }
        (
            super::MessageTarget::Group(_),
            super::MessageSecurityMode::E2eeRequired | super::MessageSecurityMode::GroupE2ee,
        ) => validate_group_e2ee_security(),
        (super::MessageTarget::Direct(_), super::MessageSecurityMode::GroupE2ee) => {
            Err(crate::ImError::unsupported("group-e2ee"))
        }
    }
}

fn validate_body(body: &super::MessageBody) -> crate::ImResult<()> {
    match body {
        super::MessageBody::Text { text, .. } if text.trim().is_empty() => {
            Err(crate::ImError::invalid_input(
                Some("text".to_string()),
                "text message must not be empty",
            ))
        }
        super::MessageBody::Text { .. } => Ok(()),
        super::MessageBody::Payload { payload } if !payload.is_object() => {
            Err(crate::ImError::invalid_input(
                Some("payload".to_string()),
                "message payload must be a JSON object",
            ))
        }
        super::MessageBody::Payload { .. } => Ok(()),
        super::MessageBody::Attachment { .. } => Ok(()),
    }
}

fn validate_attachment_security(
    body: &super::MessageBody,
    security: &super::MessageSecurityMode,
) -> crate::ImResult<()> {
    if !matches!(body, super::MessageBody::Attachment { .. }) {
        return Ok(());
    }
    match security {
        super::MessageSecurityMode::DefaultPlain
        | super::MessageSecurityMode::Plain
        | super::MessageSecurityMode::E2eeRequired
        | super::MessageSecurityMode::SecureDirect
        | super::MessageSecurityMode::GroupE2ee => Ok(()),
    }
}

fn validate_delegated_send_scope(request: &super::SendMessageRequest) -> crate::ImResult<()> {
    if request.delegated_signing.is_none() {
        return Ok(());
    }
    if !matches!(request.target, super::MessageTarget::Direct(_)) {
        return Err(crate::ImError::unsupported("delegated-group-send"));
    }
    if matches!(request.body, super::MessageBody::Attachment { .. }) {
        return Err(crate::ImError::unsupported("delegated-attachment-send"));
    }
    match request.security {
        super::MessageSecurityMode::DefaultPlain | super::MessageSecurityMode::Plain => Ok(()),
        super::MessageSecurityMode::E2eeRequired | super::MessageSecurityMode::SecureDirect => {
            Err(crate::ImError::unsupported("delegated-e2ee-send"))
        }
        super::MessageSecurityMode::GroupE2ee => {
            Err(crate::ImError::unsupported("delegated-group-e2ee-send"))
        }
    }
}

fn validate_secure_attachment_request(
    body: &super::MessageBody,
    security: &super::MessageSecurityMode,
) -> crate::ImResult<()> {
    if !matches!(body, super::MessageBody::Attachment { .. }) {
        return Err(crate::ImError::unsupported("secure-attachment"));
    }
    match security {
        super::MessageSecurityMode::E2eeRequired
        | super::MessageSecurityMode::SecureDirect
        | super::MessageSecurityMode::GroupE2ee => Ok(()),
        super::MessageSecurityMode::DefaultPlain | super::MessageSecurityMode::Plain => {
            Err(crate::ImError::unsupported("secure-attachment"))
        }
    }
}

fn attachment_request_from_message_request(
    request: super::SendMessageRequest,
) -> crate::ImResult<crate::attachments::AttachmentSendRequest> {
    match request.body {
        super::MessageBody::Attachment {
            input,
            caption,
            mention_payload,
            mime_type,
            filename,
        } => Ok(crate::attachments::AttachmentSendRequest {
            input,
            caption,
            mention_payload,
            mime_type,
            filename,
            delivery: request.delivery,
            security: crate::messages::MessageSecurityMode::DefaultPlain,
        }),
        _ => Err(crate::ImError::unsupported("attachments")),
    }
}

#[cfg(feature = "group-e2ee")]
fn validate_group_e2ee_security() -> crate::ImResult<()> {
    Ok(())
}

#[cfg(not(feature = "group-e2ee"))]
fn validate_group_e2ee_security() -> crate::ImResult<()> {
    Err(crate::ImError::unsupported("group-e2ee"))
}

fn message_kind_from_result(
    result: &super::SendMessageResult,
) -> crate::ImResult<super::MessageKind> {
    match &result.message.body {
        super::MessageBodyView::Text { kind, .. } => Ok(kind.clone()),
        super::MessageBodyView::Payload { .. } => Err(crate::ImError::unsupported(
            "message-projection-body:application/json",
        )),
        super::MessageBodyView::Unsupported { content_type } => {
            Err(crate::ImError::unsupported(format!(
                "message-projection-body:{}",
                content_type.as_deref().unwrap_or("unknown")
            )))
        }
    }
}

fn direct_handle_from_target(target: &super::MessageTarget) -> Option<&str> {
    match target {
        super::MessageTarget::Direct(peer) if !peer.as_str().starts_with("did:") => {
            Some(peer.as_str())
        }
        _ => None,
    }
}

struct StaleDeliveryTarget {
    current_did: Option<String>,
    full_handle: Option<String>,
    user_id: Option<String>,
}

fn stale_delivery_target_from_error(
    err: &crate::ImError,
    owner_did: &str,
) -> Option<StaleDeliveryTarget> {
    let crate::ImError::Service {
        code: Some(code),
        data: Some(data),
        ..
    } = err
    else {
        return None;
    };
    if code != "1406" {
        return None;
    }
    if data.get("reason").and_then(serde_json::Value::as_str) != Some("stale_did") {
        return None;
    }
    let current_did = data
        .get("current_did")
        .and_then(serde_json::Value::as_str)
        .and_then(|did| canonical_peer_did_value(did, owner_did));
    let full_handle = data
        .get("full_handle")
        .and_then(serde_json::Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(ToOwned::to_owned);
    let user_id = data
        .get("user_id")
        .and_then(serde_json::Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(ToOwned::to_owned);
    (current_did.is_some() || full_handle.is_some()).then_some(StaleDeliveryTarget {
        current_did,
        full_handle,
        user_id,
    })
}

struct ResolvedHistoryThread {
    thread: super::ThreadRef,
    resolved_did: Option<String>,
    handle_peer: Option<(String, String)>,
    peer_scope: Option<crate::internal::local_state::owner_scope::DirectPeerScope>,
}

#[derive(Debug)]
struct ResolvedConversationServiceThread {
    thread: super::ThreadRef,
    resolved_did: Option<String>,
    peer_scope: Option<crate::internal::local_state::owner_scope::DirectPeerScope>,
}

#[derive(Debug, Clone)]
pub(crate) struct ResolvedConversationSendTarget {
    pub(crate) target: super::MessageTarget,
    pub(crate) target_did: Option<String>,
    pub(crate) target_handle: Option<String>,
    pub(crate) peer_scope: Option<crate::internal::local_state::owner_scope::DirectPeerScope>,
}

fn resolve_history_thread(
    client: &crate::core::ImClient,
    thread: super::ThreadRef,
) -> crate::ImResult<ResolvedHistoryThread> {
    let super::ThreadRef::Direct(peer) = thread else {
        return Ok(ResolvedHistoryThread {
            thread,
            resolved_did: None,
            handle_peer: None,
            peer_scope: None,
        });
    };
    if peer.as_str().starts_with("did:") {
        return Ok(ResolvedHistoryThread {
            thread: super::ThreadRef::Direct(peer.clone()),
            resolved_did: Some(peer.as_str().to_owned()),
            handle_peer: None,
            peer_scope: None,
        });
    }
    let lookup = crate::internal::handle_discovery::resolve_direct_handle(client, peer.as_str())?;
    let did = lookup.target_did;
    let full_handle = lookup.full_handle;
    let peer_scope = crate::internal::local_state::owner_scope::DirectPeerScope::new(
        lookup.user_id,
        full_handle.clone(),
    )?;
    Ok(ResolvedHistoryThread {
        thread: super::ThreadRef::Direct(crate::ids::PeerRef::parse(&did, "")?),
        resolved_did: Some(did.clone()),
        handle_peer: Some((full_handle, did)),
        peer_scope: Some(peer_scope),
    })
}

fn resolve_service_conversation_thread(
    client: &crate::core::ImClient,
    conversation: &super::ConversationReadRef,
) -> crate::ImResult<ResolvedConversationServiceThread> {
    let conversation_id = conversation.conversation_id.trim();
    if let Some(group) = conversation_id.strip_prefix("group:") {
        let group = crate::ids::GroupRef::parse(group)?;
        return Ok(ResolvedConversationServiceThread {
            thread: super::ThreadRef::Group(group),
            resolved_did: None,
            peer_scope: None,
        });
    }
    if let Some(peer) = conversation_id.strip_prefix("dm:") {
        if peer.starts_with("peer-scope:") {
            return resolve_peer_scope_service_conversation_thread(client, conversation);
        }
        if !is_single_did_ref(peer) {
            return Err(unresolvable_service_conversation(
                &conversation.conversation_id,
            ));
        }
        let peer = crate::ids::PeerRef::parse(peer, "")?;
        let resolved_did = peer
            .as_str()
            .starts_with("did:")
            .then(|| peer.as_str().to_owned());
        return Ok(ResolvedConversationServiceThread {
            thread: super::ThreadRef::Direct(peer),
            resolved_did,
            peer_scope: None,
        });
    }
    Err(crate::ImError::invalid_input(
        Some("conversation_id".to_owned()),
        "conversation service operations require a canonical dm: or group: conversation_id",
    ))
}

fn mark_read_input_for_conversation(
    client: &crate::core::ImClient,
    request: super::MarkConversationReadRequest,
) -> crate::ImResult<crate::internal::message_runtime::mark_read::MarkThreadReadInput> {
    let remote_thread = resolve_service_conversation_thread(client, &request.conversation)?.thread;
    Ok(
        crate::internal::message_runtime::mark_read::MarkThreadReadInput {
            request: mark_thread_read_request_for_conversation(request)?,
            remote_thread: Some(remote_thread),
        },
    )
}

fn mark_thread_read_request_for_conversation(
    request: super::MarkConversationReadRequest,
) -> crate::ImResult<super::MarkThreadReadRequest> {
    Ok(super::MarkThreadReadRequest {
        thread: request.conversation.as_thread_ref()?,
        watermark: request.watermark,
        fallback_max_message_ids: request.fallback_max_message_ids,
    })
}

async fn resolve_service_conversation_thread_async(
    client: &crate::core::ImClient,
    conversation: &super::ConversationReadRef,
) -> crate::ImResult<ResolvedConversationServiceThread> {
    resolve_service_conversation_thread(client, conversation)
}

async fn mark_read_input_for_conversation_async(
    client: &crate::core::ImClient,
    request: super::MarkConversationReadRequest,
) -> crate::ImResult<crate::internal::message_runtime::mark_read::MarkThreadReadInput> {
    let remote_thread = resolve_service_conversation_thread_async(client, &request.conversation)
        .await?
        .thread;
    Ok(
        crate::internal::message_runtime::mark_read::MarkThreadReadInput {
            request: mark_thread_read_request_for_conversation(request)?,
            remote_thread: Some(remote_thread),
        },
    )
}

#[cfg(feature = "sqlite")]
fn resolve_peer_scope_service_conversation_thread(
    client: &crate::core::ImClient,
    conversation: &super::ConversationReadRef,
) -> crate::ImResult<ResolvedConversationServiceThread> {
    let records = conversation_projection_records(client, conversation)?;
    let Some(peer_did) = peer_current_did_from_records(&records, client.did().as_str()) else {
        return Err(unresolvable_service_conversation(
            &conversation.conversation_id,
        ));
    };
    Ok(ResolvedConversationServiceThread {
        thread: super::ThreadRef::Direct(crate::ids::PeerRef::parse(&peer_did, "")?),
        resolved_did: Some(peer_did),
        peer_scope: records.iter().find_map(peer_scope_from_record),
    })
}

#[cfg(not(feature = "sqlite"))]
fn resolve_peer_scope_service_conversation_thread(
    _client: &crate::core::ImClient,
    conversation: &super::ConversationReadRef,
) -> crate::ImResult<ResolvedConversationServiceThread> {
    Err(unresolvable_service_conversation(
        &conversation.conversation_id,
    ))
}

#[cfg(feature = "sqlite")]
fn conversation_projection_records(
    client: &crate::core::ImClient,
    conversation: &super::ConversationReadRef,
) -> crate::ImResult<Vec<crate::internal::local_state::messages::MessageRecord>> {
    let connection = crate::internal::local_state::open_writable(
        &client.core_inner().sdk_paths().local_state.sqlite_path,
    )?;
    let page =
        crate::internal::local_state::messages::list_messages_for_thread_ref_for_owner_identity(
            &connection,
            client.current_identity().id.as_str(),
            client.did().as_str(),
            &conversation.as_thread_ref()?,
            20,
            None,
        )?;
    Ok(page.records)
}

#[cfg(feature = "sqlite")]
fn peer_current_did_from_records(
    records: &[crate::internal::local_state::messages::MessageRecord],
    owner_did: &str,
) -> Option<String> {
    records
        .iter()
        .find_map(|record| incoming_participant_peer_did_from_record(record, owner_did))
        .or_else(|| {
            records
                .iter()
                .find_map(|record| metadata_peer_current_did_from_record(record, owner_did))
        })
        .or_else(|| {
            records
                .iter()
                .find_map(|record| participant_peer_did_from_record(record, owner_did))
        })
}

#[cfg(feature = "sqlite")]
fn metadata_peer_current_did_from_record(
    record: &crate::internal::local_state::messages::MessageRecord,
    owner_did: &str,
) -> Option<String> {
    let metadata = parse_message_metadata(&record.metadata);
    for candidate in [
        metadata_value(&metadata, "peer_current_did"),
        metadata_value(&metadata, "resolved_target_did"),
    ] {
        if let Some(candidate) = canonical_peer_did(candidate?, owner_did) {
            return Some(candidate.to_owned());
        }
    }
    None
}

#[cfg(feature = "sqlite")]
fn incoming_participant_peer_did_from_record(
    record: &crate::internal::local_state::messages::MessageRecord,
    owner_did: &str,
) -> Option<String> {
    if record.direction != 0 {
        return None;
    }
    participant_peer_did_from_record(record, owner_did)
}

#[cfg(feature = "sqlite")]
fn participant_peer_did_from_record(
    record: &crate::internal::local_state::messages::MessageRecord,
    owner_did: &str,
) -> Option<String> {
    for candidate in [record.sender_did.as_str(), record.receiver_did.as_str()] {
        if let Some(candidate) = canonical_peer_did(candidate, owner_did) {
            return Some(candidate.to_owned());
        }
    }
    None
}

fn canonical_peer_did<'a>(candidate: &'a str, owner_did: &str) -> Option<&'a str> {
    let candidate = candidate.trim();
    (is_single_did_ref(candidate) && candidate != owner_did).then_some(candidate)
}

fn canonical_peer_did_value(candidate: &str, owner_did: &str) -> Option<String> {
    canonical_peer_did(candidate, owner_did).map(ToOwned::to_owned)
}

fn is_single_did_ref(value: &str) -> bool {
    let value = value.trim();
    value.starts_with("did:") && !value["did:".len()..].contains(":did:")
}

#[cfg(feature = "sqlite")]
fn peer_scope_from_record(
    record: &crate::internal::local_state::messages::MessageRecord,
) -> Option<crate::internal::local_state::owner_scope::DirectPeerScope> {
    let metadata = parse_message_metadata(&record.metadata);
    crate::internal::local_state::owner_scope::DirectPeerScope::new(
        metadata_value(&metadata, "peer_user_id")?,
        metadata_value(&metadata, "peer_full_handle")?,
    )
    .ok()
}

#[cfg(feature = "sqlite")]
fn parse_message_metadata(metadata: &str) -> serde_json::Map<String, serde_json::Value> {
    serde_json::from_str::<serde_json::Value>(metadata)
        .ok()
        .and_then(|value| value.as_object().cloned())
        .unwrap_or_default()
}

#[cfg(feature = "sqlite")]
fn metadata_value<'a>(
    metadata: &'a serde_json::Map<String, serde_json::Value>,
    key: &str,
) -> Option<&'a str> {
    metadata
        .get(key)
        .and_then(serde_json::Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
}

fn unresolvable_service_conversation(conversation_id: &str) -> crate::ImError {
    crate::ImError::invalid_input(
        Some("conversation_id".to_owned()),
        format!(
            "conversation_id {conversation_id} cannot be resolved to a service direct or group thread"
        ),
    )
}

async fn resolve_history_thread_async(
    client: &crate::core::ImClient,
    thread: super::ThreadRef,
) -> crate::ImResult<ResolvedHistoryThread> {
    let super::ThreadRef::Direct(peer) = thread else {
        return Ok(ResolvedHistoryThread {
            thread,
            resolved_did: None,
            handle_peer: None,
            peer_scope: None,
        });
    };
    if peer.as_str().starts_with("did:") {
        return Ok(ResolvedHistoryThread {
            thread: super::ThreadRef::Direct(peer.clone()),
            resolved_did: Some(peer.as_str().to_owned()),
            handle_peer: None,
            peer_scope: None,
        });
    }
    let lookup =
        crate::internal::handle_discovery::resolve_direct_handle_async(client, peer.as_str())
            .await?;
    let did = lookup.target_did;
    let full_handle = lookup.full_handle;
    let peer_scope = crate::internal::local_state::owner_scope::DirectPeerScope::new(
        lookup.user_id,
        full_handle.clone(),
    )?;
    Ok(ResolvedHistoryThread {
        thread: super::ThreadRef::Direct(crate::ids::PeerRef::parse(&did, "")?),
        resolved_did: Some(did.clone()),
        handle_peer: Some((full_handle, did)),
        peer_scope: Some(peer_scope),
    })
}

fn message_page_from_read_result(
    result: crate::internal::message_runtime::read::ReadPageResult,
) -> crate::ImResult<super::MessagePage> {
    let source = result
        .raw
        .get("source")
        .and_then(serde_json::Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(ToOwned::to_owned);
    let resolved_dids = result
        .raw
        .get("resolved_dids")
        .and_then(serde_json::Value::as_array)
        .into_iter()
        .flatten()
        .filter_map(|value| value.as_str())
        .filter_map(|did| crate::ids::Did::parse(did).ok())
        .collect();
    let warnings = result
        .raw
        .get("warnings")
        .and_then(serde_json::Value::as_array)
        .into_iter()
        .flatten()
        .filter_map(|value| value.as_str())
        .map(ToOwned::to_owned)
        .collect();
    Ok(super::MessagePage {
        items: result.page.items,
        next_cursor: result.page.next_cursor,
        has_more: result.page.has_more,
        source,
        resolved_dids,
        warnings,
    })
}

#[cfg(all(test, feature = "group-e2ee"))]
mod group_e2ee_public_send_tests {
    use std::fs;
    use std::io::{Read, Write};
    use std::net::{TcpListener, TcpStream};
    use std::path::{Path, PathBuf};
    use std::sync::{Arc, Mutex};
    use std::thread;
    use std::time::{Duration, Instant};

    use anp::group_e2ee::operations::{CreateGroupInput, FinalizeCommitInput};
    use serde_json::{json, Value};

    use crate::internal::group_e2ee::provider::GroupMlsProvider;

    #[cfg(feature = "blocking")]
    #[test]
    fn public_group_e2ee_send_uses_native_provider_and_sends_cipher_only() {
        let fixture = Fixture::new();
        let server = RpcTestServer::spawn(vec![
            json!({
                "group_state_ref": {
                    "group_did": "did:wba:awiki.test:groups:group-e2ee-preflight",
                    "group_state_version": "preflight-state"
                }
            }),
            json!({
                "group_state_ref": {
                    "group_did": fixture.group_did,
                    "group_state_version": "service-state-1"
                }
            }),
            json!({
                "accepted": true,
                "final_acceptance": true,
                "group_did": fixture.group_did,
                "message_id": "server-message-id",
                "operation_id": "op-public-group-e2ee",
                "group_event_seq": "91",
                "group_state_version": "service-state-2",
                "accepted_at": "2026-05-21T00:00:00Z"
            }),
        ]);
        let core =
            crate::core::ImCore::new(fixture.config(server.base_url()), fixture.paths()).unwrap();
        let client = core
            .client(crate::identity::IdentitySelector::LocalAlias(
                "alice".to_owned(),
            ))
            .unwrap();
        prepare_local_mls_group(&client, &fixture.group_did);

        let result = client
            .messages()
            .send(crate::messages::SendMessageRequest {
                target: crate::messages::MessageTarget::Group(
                    crate::ids::GroupRef::parse(&fixture.group_did).unwrap(),
                ),
                body: crate::messages::MessageBody::Text {
                    text: "public group secret".to_owned(),
                    kind: crate::messages::MessageKind::Text,
                },
                security: crate::messages::MessageSecurityPolicy::E2eeRequired,
                client_message_id: Some(
                    crate::ids::MessageId::parse("msg-public-group-e2ee").unwrap(),
                ),
                delivery: crate::messages::MessageDeliveryOptions {
                    idempotency_key: Some("op-public-group-e2ee".to_owned()),
                    wait_for_final_acceptance: true,
                },
                delegated_signing: None,
            })
            .unwrap();

        assert_eq!(result.message.metadata.server_sequence, Some(91));
        assert!(matches!(
            result.delivery,
            crate::messages::DeliveryState::Accepted
        ));
        let requests = server.requests();
        assert_eq!(requests.len(), 3);
        assert_eq!(requests[0].rpc_method, "group.e2ee.head");
        assert_eq!(
            requests[0].params["body"]["group_did"],
            "did:wba:awiki.test:groups:group-e2ee-preflight"
        );
        assert_eq!(requests[1].rpc_method, "group.e2ee.head");
        assert_eq!(requests[1].params["body"]["group_did"], fixture.group_did);
        assert_eq!(requests[2].rpc_method, "group.e2ee.send");
        assert_eq!(
            requests[2].params["meta"]["security_profile"],
            anp::group_e2ee::SECURITY_PROFILE
        );
        assert_eq!(
            requests[2].params["meta"]["content_type"],
            anp::group_e2ee::GROUP_CIPHER_CONTENT_TYPE
        );
        assert_eq!(
            requests[2].params["body"]["group_state_ref"]["group_state_version"],
            "service-state-1"
        );
        assert!(
            requests[2].params["body"]["private_message_b64u"]
                .as_str()
                .map(|value| !value.trim().is_empty())
                .unwrap_or(false),
            "group.e2ee.send must carry an MLS private message"
        );
        let encoded_send = serde_json::to_string(&requests[2].params).unwrap();
        assert!(!encoded_send.contains("public group secret"));
        assert!(!encoded_send.contains("application_plaintext"));
        assert!(!encoded_send.contains("provider"));
        assert!(!encoded_send.contains("StorageProvider"));
        assert!(!encoded_send.contains("mls_state.sqlite"));
        assert!(!encoded_send.contains("openmls_group_id_b64u"));
    }

    #[cfg(not(feature = "blocking"))]
    #[test]
    fn public_group_e2ee_sync_send_fails_closed_by_default() {
        let fixture = Fixture::new();
        let core = crate::core::ImCore::new(
            fixture.config("https://example.test".to_owned()),
            fixture.paths(),
        )
        .unwrap();
        let client = core
            .client(crate::identity::IdentitySelector::LocalAlias(
                "alice".to_owned(),
            ))
            .unwrap();

        let result = client.messages().send(crate::messages::SendMessageRequest {
            target: crate::messages::MessageTarget::Group(
                crate::ids::GroupRef::parse(&fixture.group_did).unwrap(),
            ),
            body: crate::messages::MessageBody::Text {
                text: "public group secret".to_owned(),
                kind: crate::messages::MessageKind::Text,
            },
            security: crate::messages::MessageSecurityPolicy::E2eeRequired,
            client_message_id: Some(
                crate::ids::MessageId::parse("msg-public-group-e2ee-sync").unwrap(),
            ),
            delivery: crate::messages::MessageDeliveryOptions::default(),
            delegated_signing: None,
        });

        assert!(matches!(
            result,
            Err(crate::ImError::UnsupportedCapability { capability }) if capability == "sync-group-e2ee-send"
        ));
    }

    #[tokio::test]
    async fn public_group_e2ee_send_async_uses_async_transport_and_db_actor_projection() {
        let fixture = Fixture::new();
        let server = RpcTestServer::spawn(vec![
            json!({
                "group_state_ref": {
                    "group_did": fixture.group_did,
                    "group_state_version": "service-state-async-1"
                }
            }),
            json!({
                "accepted": true,
                "final_acceptance": true,
                "group_did": fixture.group_did,
                "message_id": "server-message-async-id",
                "operation_id": "op-public-group-e2ee-async",
                "group_event_seq": "92",
                "group_state_version": "service-state-async-2",
                "accepted_at": "2026-05-21T00:00:00Z"
            }),
        ]);
        let core = crate::core::ImCore::open(fixture.config(server.base_url()), fixture.paths())
            .await
            .unwrap();
        let client = core
            .client_async(crate::identity::IdentitySelector::LocalAlias(
                "alice".to_owned(),
            ))
            .await
            .unwrap();
        prepare_local_mls_group(&client, &fixture.group_did);

        let result = client
            .messages()
            .send_async(crate::messages::SendMessageRequest {
                target: crate::messages::MessageTarget::Group(
                    crate::ids::GroupRef::parse(&fixture.group_did).unwrap(),
                ),
                body: crate::messages::MessageBody::Text {
                    text: "async group secret".to_owned(),
                    kind: crate::messages::MessageKind::Text,
                },
                security: crate::messages::MessageSecurityPolicy::E2eeRequired,
                client_message_id: Some(
                    crate::ids::MessageId::parse("msg-public-group-e2ee-async").unwrap(),
                ),
                delivery: crate::messages::MessageDeliveryOptions {
                    idempotency_key: Some("op-public-group-e2ee-async".to_owned()),
                    wait_for_final_acceptance: true,
                },
                delegated_signing: None,
            })
            .await
            .unwrap();

        assert_eq!(result.message.metadata.server_sequence, Some(92));
        let requests = server.requests();
        assert_eq!(requests.len(), 2);
        assert_eq!(requests[0].rpc_method, "group.e2ee.head");
        assert_eq!(requests[0].params["body"]["group_did"], fixture.group_did);
        assert_eq!(requests[1].rpc_method, "group.e2ee.send");
        assert_eq!(
            requests[1].params["meta"]["content_type"],
            anp::group_e2ee::GROUP_CIPHER_CONTENT_TYPE
        );
        let encoded_send = serde_json::to_string(&requests[1].params).unwrap();
        assert!(!encoded_send.contains("async group secret"));

        let stored = stored_group_message(&fixture, &client, result.message.id.as_str());
        assert_eq!(stored.thread_id, format!("group:{}", fixture.group_did));
        assert_eq!(stored.group_did, fixture.group_did);
        assert_eq!(stored.content, "async group secret");
        assert!(stored.is_e2ee);
        assert_eq!(stored.server_seq, Some(92));
        let metadata: Value = serde_json::from_str(&stored.metadata).unwrap();
        assert_eq!(metadata["security"], "group-e2ee");
    }

    #[test]
    fn public_group_e2ee_mode_still_rejects_direct_targets() {
        let fixture = Fixture::new();
        let core = crate::core::ImCore::new(
            fixture.config("https://example.test".to_owned()),
            fixture.paths(),
        )
        .unwrap();
        let client = core
            .client(crate::identity::IdentitySelector::LocalAlias(
                "alice".to_owned(),
            ))
            .unwrap();

        let result = client.messages().send(crate::messages::SendMessageRequest {
            target: crate::messages::MessageTarget::Direct(
                crate::ids::PeerRef::parse("did:example:bob", "").unwrap(),
            ),
            body: crate::messages::MessageBody::Text {
                text: "not group".to_owned(),
                kind: crate::messages::MessageKind::Text,
            },
            security: crate::messages::MessageSecurityMode::GroupE2ee,
            client_message_id: None,
            delivery: crate::messages::MessageDeliveryOptions::default(),
            delegated_signing: None,
        });

        assert!(matches!(
            result,
            Err(crate::ImError::UnsupportedCapability { capability }) if capability == "group-e2ee"
        ));
    }

    fn prepare_local_mls_group(client: &crate::core::ImClient, group_did: &str) {
        let provider = crate::internal::group_e2ee::storage::native_provider_for_client(client)
            .expect("native provider");
        let prepared = provider
            .create_group_prepare(CreateGroupInput {
                creator_did: client.did().as_str().to_owned(),
                device_id: crate::internal::group_e2ee::DEFAULT_GROUP_MLS_DEVICE_ID.to_owned(),
                group_did: group_did.to_owned(),
                operation_id: "op-public-group-e2ee-create".to_owned(),
                request_id: "req-public-group-e2ee-create".to_owned(),
                pending_commit_id: Some("pc-public-group-e2ee-create".to_owned()),
            })
            .expect("create group");
        provider
            .finalize_commit(FinalizeCommitInput {
                pending_commit_id: prepared.pending_commit_id,
                request_id: "req-public-group-e2ee-finalize".to_owned(),
            })
            .expect("finalize group");
    }

    struct Fixture {
        root: PathBuf,
        group_did: String,
    }

    impl Fixture {
        fn new() -> Self {
            let root = unique_temp_root();
            write_identity_fixture(&root, "alice", "did:example:alice");
            Self {
                root,
                group_did: "did:example:groups:public-e2ee".to_owned(),
            }
        }

        fn config(&self, base_url: String) -> crate::ImCoreConfig {
            crate::ImCoreConfig {
                service_base_url: crate::ServiceEndpoint::parse(base_url).unwrap(),
                did_domain: "awiki.test".to_owned(),
                user_service_endpoint: None,
                message_service_endpoint: None,
                mail_service_endpoint: None,
                anp_service_endpoint: None,
                anp_service_did: None,
                ca_bundle: None,
                transport_policy: crate::MessageTransportPolicy::HttpOnly,
            }
        }

        fn paths(&self) -> crate::ImCorePaths {
            crate::ImCorePaths {
                identities: crate::IdentityRegistryPaths {
                    identity_root_dir: self.root.join("identities"),
                    registry_path: self.root.join("identities").join("registry.json"),
                    default_identity_path: Some(self.root.join("identities").join("default")),
                },
                local_state: crate::LocalStatePaths {
                    sqlite_path: self.root.join("local").join("im.sqlite"),
                },
                runtime: crate::RuntimePaths {
                    cache_dir: self.root.join("cache"),
                    temp_dir: self.root.join("tmp"),
                },
            }
        }
    }

    fn write_identity_fixture(root: &Path, alias: &str, did: &str) {
        let identity_root = root.join("identities");
        let identity_dir = identity_root.join(alias);
        fs::create_dir_all(&identity_dir).unwrap();
        fs::write(identity_root.join("default"), format!("{alias}\n")).unwrap();
        fs::write(
            identity_root.join("registry.json"),
            json!({
                "default_identity": alias,
                "identities": [{
                    "id": "alice-id",
                    "did": did,
                    "local_alias": alias,
                    "device_id": crate::internal::group_e2ee::DEFAULT_GROUP_MLS_DEVICE_ID,
                    "ready_for_auth": true,
                    "ready_for_messaging": true,
                    "missing": []
                }]
            })
            .to_string(),
        )
        .unwrap();
        let bundle = anp::authentication::create_did_wba_document(
            "awiki.test",
            anp::authentication::DidDocumentOptions {
                path_segments: vec!["user".to_owned()],
                domain: Some("awiki.test".to_owned()),
                challenge: Some("group-e2ee-public-send-test".to_owned()),
                ..anp::authentication::DidDocumentOptions::default()
            },
        )
        .unwrap();
        fs::write(
            identity_dir.join("did.json"),
            serde_json::to_vec_pretty(&bundle.did_document).unwrap(),
        )
        .unwrap();
        fs::write(
            identity_dir.join("private.key"),
            bundle.private_key_pem("key-1").unwrap(),
        )
        .unwrap();
        fs::write(
            identity_dir.join("auth.json"),
            r#"{"jwt_token":"test-token"}"#,
        )
        .unwrap();
    }

    #[derive(Debug)]
    struct StoredGroupMessage {
        thread_id: String,
        group_did: String,
        content: String,
        is_e2ee: bool,
        server_seq: Option<i64>,
        metadata: String,
    }

    fn stored_group_message(
        fixture: &Fixture,
        client: &crate::core::ImClient,
        message_id: &str,
    ) -> StoredGroupMessage {
        let connection =
            rusqlite::Connection::open(fixture.root.join("local").join("im.sqlite")).unwrap();
        connection
            .query_row(
                r#"
SELECT thread_id, group_did, content, is_e2ee, server_seq, metadata
FROM messages
WHERE owner_identity_id = ?1 AND owner_did = ?2 AND msg_id = ?3
"#,
                rusqlite::params![
                    client.current_identity().id.as_str(),
                    client.did().as_str(),
                    message_id
                ],
                |row| {
                    Ok(StoredGroupMessage {
                        thread_id: row.get(0)?,
                        group_did: row.get(1)?,
                        content: row.get(2)?,
                        is_e2ee: row.get::<_, i64>(3)? != 0,
                        server_seq: row.get(4)?,
                        metadata: row.get(5)?,
                    })
                },
            )
            .unwrap()
    }

    #[derive(Debug, Clone)]
    struct CapturedRpc {
        rpc_method: String,
        params: Value,
    }

    struct RpcTestServer {
        address: String,
        requests: Arc<Mutex<Vec<CapturedRpc>>>,
        join: Option<thread::JoinHandle<()>>,
    }

    impl RpcTestServer {
        fn spawn(responses: Vec<Value>) -> Self {
            let listener = TcpListener::bind("127.0.0.1:0").unwrap();
            listener.set_nonblocking(true).unwrap();
            let address = format!("http://{}", listener.local_addr().unwrap());
            let requests = Arc::new(Mutex::new(Vec::new()));
            let server_requests = Arc::clone(&requests);
            let join = thread::spawn(move || {
                let deadline = Instant::now() + Duration::from_secs(10);
                for response in responses {
                    let mut stream = accept_before_deadline(&listener, deadline);
                    let request = read_rpc_request(&mut stream);
                    let id = request.id.clone();
                    server_requests.lock().unwrap().push(CapturedRpc {
                        rpc_method: request.rpc_method,
                        params: request.params,
                    });
                    write_rpc_response(&mut stream, id, response);
                }
            });
            Self {
                address,
                requests,
                join: Some(join),
            }
        }

        fn base_url(&self) -> String {
            self.address.clone()
        }

        fn requests(&self) -> Vec<CapturedRpc> {
            self.requests.lock().unwrap().clone()
        }
    }

    impl Drop for RpcTestServer {
        fn drop(&mut self) {
            if let Some(join) = self.join.take() {
                let _ = join.join();
            }
        }
    }

    #[derive(Debug)]
    struct RawRpcRequest {
        rpc_method: String,
        params: Value,
        id: Value,
    }

    fn accept_before_deadline(listener: &TcpListener, deadline: Instant) -> TcpStream {
        loop {
            match listener.accept() {
                Ok((stream, _)) => return stream,
                Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
                    assert!(Instant::now() < deadline, "timed out waiting for RPC");
                    thread::sleep(Duration::from_millis(10));
                }
                Err(err) => panic!("accept RPC request: {err}"),
            }
        }
    }

    fn read_rpc_request(stream: &mut TcpStream) -> RawRpcRequest {
        stream
            .set_read_timeout(Some(Duration::from_secs(5)))
            .unwrap();
        let mut raw = Vec::new();
        let mut buffer = [0_u8; 4096];
        let header_end = loop {
            let count = stream.read(&mut buffer).unwrap();
            assert!(count > 0, "RPC request closed before headers");
            raw.extend_from_slice(&buffer[..count]);
            if let Some(index) = raw.windows(4).position(|window| window == b"\r\n\r\n") {
                break index;
            }
        };
        let headers_text = std::str::from_utf8(&raw[..header_end]).unwrap();
        let headers = headers_text
            .lines()
            .skip(1)
            .filter_map(|line| {
                let (key, value) = line.split_once(':')?;
                Some((key.trim().to_ascii_lowercase(), value.trim().to_owned()))
            })
            .collect::<std::collections::BTreeMap<_, _>>();
        let content_length = headers
            .get("content-length")
            .and_then(|value| value.parse::<usize>().ok())
            .unwrap_or_default();
        let body_start = header_end + 4;
        while raw.len() < body_start + content_length {
            let count = stream.read(&mut buffer).unwrap();
            assert!(count > 0, "RPC request closed before body");
            raw.extend_from_slice(&buffer[..count]);
        }
        let payload: Value = serde_json::from_slice(&raw[body_start..body_start + content_length])
            .expect("json rpc request body");
        RawRpcRequest {
            rpc_method: payload["method"].as_str().unwrap().to_owned(),
            params: payload["params"].clone(),
            id: payload["id"].clone(),
        }
    }

    fn write_rpc_response(stream: &mut TcpStream, id: Value, result: Value) {
        let body = json!({
            "jsonrpc": "2.0",
            "id": id,
            "result": result,
        })
        .to_string();
        write!(
            stream,
            "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
            body.len(),
            body
        )
        .unwrap();
    }

    fn unique_temp_root() -> PathBuf {
        let nanos = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        std::env::temp_dir().join(format!(
            "im-core-group-e2ee-public-send-{}-{nanos}",
            std::process::id()
        ))
    }
}