lxmf-sdk 0.7.0

use crate::app::{Envelope, EnvelopeResponse, OperationRegistry};
use crate::backend::SdkBackend;
#[cfg(feature = "sdk-async")]
use crate::backend::SdkEventStream;
#[cfg(feature = "sdk-async")]
use crate::backend::{SdkBackendAsyncEvents, SdkBackendAsyncOps};
use crate::capability::{
    EffectiveLimits, NegotiationRequest, NegotiationResponse, ParityReference,
};
use crate::domain::{
    AttachmentDownloadChunk, AttachmentDownloadChunkRequest, AttachmentId, AttachmentListRequest,
    AttachmentListResult, AttachmentMeta, AttachmentStoreRequest, AttachmentUploadChunkAck,
    AttachmentUploadChunkRequest, AttachmentUploadCommitRequest, AttachmentUploadSession,
    AttachmentUploadStartRequest, ContactListRequest, ContactListResult, ContactRecord,
    ContactUpdateRequest, IdentityBootstrapRequest, IdentityBundle, IdentityImportRequest,
    IdentityRef, IdentityResolveRequest, MarkerCreateRequest, MarkerDeleteRequest,
    MarkerListRequest, MarkerListResult, MarkerRecord, MarkerUpdatePositionRequest,
    PaperDecodeResult, PaperMessageEnvelope, PresenceListRequest, PresenceListResult,
    RemoteCommandRequest, RemoteCommandResponse, RemoteCommandSession,
    RemoteCommandSessionListRequest, RemoteCommandSessionListResult, TelemetryPoint,
    TelemetryQuery, TopicCreateRequest, TopicId, TopicListRequest, TopicListResult,
    TopicPublishRequest, TopicRecord, TopicSubscriptionRequest, VoiceSessionId,
    VoiceSessionOpenRequest, VoiceSessionState, VoiceSessionUpdateRequest,
};
use crate::error::{code, ErrorCategory, SdkError};
use crate::event::{EventBatch, EventCursor, SdkEvent, Severity};
#[cfg(feature = "sdk-async")]
use crate::event::{EventSubscription, SubscriptionStart};
use crate::types::{
    Ack, AuthMode, CancelResult, ConfigPatch, DeliverySnapshot, DeliveryState, MessageId,
    RuntimeSnapshot, RuntimeState, SendRequest, ShutdownMode, TickBudget, TickResult,
};
use serde::de::DeserializeOwned;
use serde_json::{Map as JsonMap, Value as JsonValue};
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::RwLock;
use std::time::{SystemTime, UNIX_EPOCH};
use zeroize::Zeroizing;

#[path = "rpc/core_impl.rs"]
mod core_impl;
#[path = "rpc/domains_impl.rs"]
mod domains_impl;
#[path = "rpc/parsing.rs"]
mod parsing;
#[path = "rpc/transport.rs"]
mod transport;

pub struct RpcBackendClient {
    endpoint: String,
    client_instance_id: String,
    next_request_id: AtomicU64,
    negotiated_capabilities: RwLock<Vec<String>>,
    negotiated_limits: RwLock<Option<EffectiveLimits>>,
    manual_tick_cursor: RwLock<Option<EventCursor>>,
    session_auth: RwLock<SessionAuth>,
}

static NEXT_CLIENT_INSTANCE_ID: AtomicU64 = AtomicU64::new(1);

enum SessionAuth {
    LocalTrusted,
    Token {
        issuer: String,
        audience: String,
        shared_secret: Zeroizing<String>,
        ttl_secs: u64,
    },
    Mtls {
        ca_bundle_path: String,
        client_cert_path: Option<String>,
        client_key_path: Option<String>,
    },
}

#[derive(Clone)]
struct MtlsRequestAuth {
    ca_bundle_path: String,
    client_cert_path: Option<String>,
    client_key_path: Option<String>,
}

impl RpcBackendClient {
    pub fn new(endpoint: impl Into<String>) -> Self {
        Self {
            endpoint: endpoint.into(),
            client_instance_id: new_client_instance_id(),
            next_request_id: AtomicU64::new(1),
            negotiated_capabilities: RwLock::new(Vec::new()),
            negotiated_limits: RwLock::new(None),
            manual_tick_cursor: RwLock::new(None),
            session_auth: RwLock::new(SessionAuth::LocalTrusted),
        }
    }

    fn next_request_id(&self) -> u64 {
        self.next_request_id.fetch_add(1, Ordering::Relaxed)
    }

    fn next_message_id(&self) -> String {
        format!("sdk-{}-{}", self.client_instance_id, self.next_request_id())
    }

    fn now_seconds() -> u64 {
        SystemTime::now().duration_since(UNIX_EPOCH).map(|duration| duration.as_secs()).unwrap_or(0)
    }

    fn has_capability(&self, capability_id: &str) -> bool {
        self.negotiated_capabilities
            .read()
            .expect("negotiated_capabilities rwlock poisoned")
            .iter()
            .any(|capability| capability == capability_id)
    }

    fn negotiated_max_poll_events(&self) -> usize {
        self.negotiated_limits
            .read()
            .expect("negotiated_limits rwlock poisoned")
            .as_ref()
            .map(|limits| limits.max_poll_events.max(1))
            .unwrap_or(256)
    }
}

fn new_client_instance_id() -> String {
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|duration| duration.as_nanos())
        .unwrap_or(0);
    let serial = NEXT_CLIENT_INSTANCE_ID.fetch_add(1, Ordering::Relaxed);
    format!("{:032x}-{:x}-{:x}", now, std::process::id(), serial)
}

impl SdkBackend for RpcBackendClient {
    fn negotiate(&self, req: NegotiationRequest) -> Result<NegotiationResponse, SdkError> {
        self.negotiate_impl(req)
    }

    fn send(&self, req: SendRequest) -> Result<MessageId, SdkError> {
        self.send_impl(req)
    }

    fn cancel(&self, id: MessageId) -> Result<CancelResult, SdkError> {
        self.cancel_impl(id)
    }

    fn status(&self, id: MessageId) -> Result<Option<DeliverySnapshot>, SdkError> {
        self.status_impl(id)
    }

    fn configure(&self, expected_revision: u64, patch: ConfigPatch) -> Result<Ack, SdkError> {
        self.configure_impl(expected_revision, patch)
    }

    fn poll_events(&self, cursor: Option<EventCursor>, max: usize) -> Result<EventBatch, SdkError> {
        self.poll_events_impl(cursor, max)
    }

    fn snapshot(&self) -> Result<RuntimeSnapshot, SdkError> {
        self.snapshot_impl()
    }

    fn shutdown(&self, mode: ShutdownMode) -> Result<Ack, SdkError> {
        self.shutdown_impl(mode)
    }

    fn topic_create(&self, req: TopicCreateRequest) -> Result<TopicRecord, SdkError> {
        self.topic_create_impl(req)
    }

    fn topic_get(&self, topic_id: TopicId) -> Result<Option<TopicRecord>, SdkError> {
        self.topic_get_impl(topic_id)
    }

    fn topic_list(&self, req: TopicListRequest) -> Result<TopicListResult, SdkError> {
        self.topic_list_impl(req)
    }

    fn topic_subscribe(&self, req: TopicSubscriptionRequest) -> Result<Ack, SdkError> {
        self.topic_subscribe_impl(req)
    }

    fn topic_unsubscribe(&self, topic_id: TopicId) -> Result<Ack, SdkError> {
        self.topic_unsubscribe_impl(topic_id)
    }

    fn topic_publish(&self, req: TopicPublishRequest) -> Result<Ack, SdkError> {
        self.topic_publish_impl(req)
    }

    fn telemetry_query(&self, query: TelemetryQuery) -> Result<Vec<TelemetryPoint>, SdkError> {
        self.telemetry_query_impl(query)
    }

    fn telemetry_subscribe(&self, query: TelemetryQuery) -> Result<Ack, SdkError> {
        self.telemetry_subscribe_impl(query)
    }

    fn attachment_store(&self, req: AttachmentStoreRequest) -> Result<AttachmentMeta, SdkError> {
        self.attachment_store_impl(req)
    }

    fn attachment_get(
        &self,
        attachment_id: AttachmentId,
    ) -> Result<Option<AttachmentMeta>, SdkError> {
        self.attachment_get_impl(attachment_id)
    }

    fn attachment_list(
        &self,
        req: AttachmentListRequest,
    ) -> Result<AttachmentListResult, SdkError> {
        self.attachment_list_impl(req)
    }

    fn attachment_delete(&self, attachment_id: AttachmentId) -> Result<Ack, SdkError> {
        self.attachment_delete_impl(attachment_id)
    }

    fn attachment_download(&self, attachment_id: AttachmentId) -> Result<Ack, SdkError> {
        self.attachment_download_impl(attachment_id)
    }

    fn attachment_upload_start(
        &self,
        req: AttachmentUploadStartRequest,
    ) -> Result<AttachmentUploadSession, SdkError> {
        self.attachment_upload_start_impl(req)
    }

    fn attachment_upload_chunk(
        &self,
        req: AttachmentUploadChunkRequest,
    ) -> Result<AttachmentUploadChunkAck, SdkError> {
        self.attachment_upload_chunk_impl(req)
    }

    fn attachment_upload_commit(
        &self,
        req: AttachmentUploadCommitRequest,
    ) -> Result<AttachmentMeta, SdkError> {
        self.attachment_upload_commit_impl(req)
    }

    fn attachment_download_chunk(
        &self,
        req: AttachmentDownloadChunkRequest,
    ) -> Result<AttachmentDownloadChunk, SdkError> {
        self.attachment_download_chunk_impl(req)
    }

    fn attachment_associate_topic(
        &self,
        attachment_id: AttachmentId,
        topic_id: TopicId,
    ) -> Result<Ack, SdkError> {
        self.attachment_associate_topic_impl(attachment_id, topic_id)
    }

    fn marker_create(&self, req: MarkerCreateRequest) -> Result<MarkerRecord, SdkError> {
        self.marker_create_impl(req)
    }

    fn marker_list(&self, req: MarkerListRequest) -> Result<MarkerListResult, SdkError> {
        self.marker_list_impl(req)
    }

    fn marker_update_position(
        &self,
        req: MarkerUpdatePositionRequest,
    ) -> Result<MarkerRecord, SdkError> {
        self.marker_update_position_impl(req)
    }

    fn marker_delete(&self, req: MarkerDeleteRequest) -> Result<Ack, SdkError> {
        self.marker_delete_impl(req)
    }

    fn identity_list(&self) -> Result<Vec<IdentityBundle>, SdkError> {
        self.identity_list_impl()
    }

    fn identity_announce_now(&self) -> Result<Ack, SdkError> {
        self.identity_announce_now_impl()
    }

    fn identity_presence_list(
        &self,
        req: PresenceListRequest,
    ) -> Result<PresenceListResult, SdkError> {
        self.identity_presence_list_impl(req)
    }

    fn identity_activate(&self, identity: IdentityRef) -> Result<Ack, SdkError> {
        self.identity_activate_impl(identity)
    }

    fn identity_import(&self, req: IdentityImportRequest) -> Result<IdentityBundle, SdkError> {
        self.identity_import_impl(req)
    }

    fn identity_export(&self, identity: IdentityRef) -> Result<IdentityImportRequest, SdkError> {
        self.identity_export_impl(identity)
    }

    fn identity_resolve(
        &self,
        req: IdentityResolveRequest,
    ) -> Result<Option<IdentityRef>, SdkError> {
        self.identity_resolve_impl(req)
    }

    fn identity_contact_update(
        &self,
        req: ContactUpdateRequest,
    ) -> Result<ContactRecord, SdkError> {
        self.identity_contact_update_impl(req)
    }

    fn identity_contact_list(
        &self,
        req: ContactListRequest,
    ) -> Result<ContactListResult, SdkError> {
        self.identity_contact_list_impl(req)
    }

    fn identity_bootstrap(&self, req: IdentityBootstrapRequest) -> Result<ContactRecord, SdkError> {
        self.identity_bootstrap_impl(req)
    }

    fn paper_encode(&self, message_id: MessageId) -> Result<PaperMessageEnvelope, SdkError> {
        self.paper_encode_impl(message_id)
    }

    fn paper_decode(&self, envelope: PaperMessageEnvelope) -> Result<Ack, SdkError> {
        self.paper_decode_impl(envelope)
    }

    fn paper_decode_with_metadata(
        &self,
        envelope: PaperMessageEnvelope,
    ) -> Result<PaperDecodeResult, SdkError> {
        self.paper_decode_with_metadata_impl(envelope)
    }

    fn command_invoke(&self, req: RemoteCommandRequest) -> Result<RemoteCommandResponse, SdkError> {
        self.command_invoke_impl(req)
    }

    fn command_reply(
        &self,
        correlation_id: String,
        reply: RemoteCommandResponse,
    ) -> Result<Ack, SdkError> {
        self.command_reply_impl(correlation_id, reply)
    }

    fn command_session_get(
        &self,
        correlation_id: String,
    ) -> Result<Option<RemoteCommandSession>, SdkError> {
        self.command_session_get_impl(correlation_id)
    }

    fn command_session_list(
        &self,
        req: RemoteCommandSessionListRequest,
    ) -> Result<RemoteCommandSessionListResult, SdkError> {
        self.command_session_list_impl(req)
    }

    fn voice_session_open(&self, req: VoiceSessionOpenRequest) -> Result<VoiceSessionId, SdkError> {
        self.voice_session_open_impl(req)
    }

    fn voice_session_update(
        &self,
        req: VoiceSessionUpdateRequest,
    ) -> Result<VoiceSessionState, SdkError> {
        self.voice_session_update_impl(req)
    }

    fn voice_session_close(&self, session_id: VoiceSessionId) -> Result<Ack, SdkError> {
        self.voice_session_close_impl(session_id)
    }

    fn operation_registry(&self) -> Result<OperationRegistry, SdkError> {
        self.operation_registry_impl()
    }

    fn envelope_execute(&self, envelope: Envelope) -> Result<EnvelopeResponse, SdkError> {
        self.envelope_execute_impl(envelope)
    }

    fn tick(&self, budget: TickBudget) -> Result<TickResult, SdkError> {
        self.tick_impl(budget)
    }
}

#[cfg(feature = "sdk-async")]
impl SdkBackendAsyncOps for RpcBackendClient {
    fn negotiate_async(
        &self,
        req: NegotiationRequest,
    ) -> crate::SdkBoxFuture<'_, NegotiationResponse> {
        Box::pin(async move { self.negotiate_async_impl(req).await })
    }

    fn send_async(&self, req: SendRequest) -> crate::SdkBoxFuture<'_, MessageId> {
        Box::pin(async move { self.send_async_impl(req).await })
    }

    fn status_async(&self, id: MessageId) -> crate::SdkBoxFuture<'_, Option<DeliverySnapshot>> {
        Box::pin(async move { self.status_async_impl(id).await })
    }

    fn snapshot_async(&self) -> crate::SdkBoxFuture<'_, RuntimeSnapshot> {
        Box::pin(async move { self.snapshot_async_impl().await })
    }

    fn shutdown_async(&self, mode: ShutdownMode) -> crate::SdkBoxFuture<'_, Ack> {
        Box::pin(async move { self.shutdown_async_impl(mode).await })
    }
}

#[cfg(feature = "sdk-async")]
impl SdkBackendAsyncEvents for RpcBackendClient {
    fn subscribe_events(&self, start: SubscriptionStart) -> Result<EventSubscription, SdkError> {
        self.subscribe_events_impl(start)
    }

    fn open_event_stream(
        &self,
        subscription: &EventSubscription,
    ) -> Result<Option<SdkEventStream>, SdkError> {
        self.open_event_stream_impl(subscription)
    }
}