kmp-adapter-embedded 0.1.5

use std::collections::BTreeMap;
use std::time::{Duration, SystemTime};

use kmp_domain::{
    NodeDetailProjection, NodeProjection, PortError, ProjectionCheckpoint, Provenance,
    RelationExplanation, SourceKind,
};
use serde::{Deserialize, Serialize, de::DeserializeOwned};

pub(crate) fn encode<T: Serialize>(what: &str, value: &T) -> Result<Vec<u8>, PortError> {
    serde_json::to_vec(value).map_err(|error| {
        PortError::InvalidState(format!("embedded store could not encode {what}: {error}"))
    })
}

pub(crate) fn decode<T: DeserializeOwned>(what: &str, bytes: &[u8]) -> Result<T, PortError> {
    serde_json::from_slice(bytes).map_err(|error| {
        PortError::InvalidState(format!("embedded store could not decode {what}: {error}"))
    })
}

#[derive(Debug, Serialize, Deserialize)]
pub(crate) struct NodeRecord {
    node_id: String,
    node_kind: String,
    title: String,
    summary: String,
    status: String,
    labels: Vec<String>,
    properties: BTreeMap<String, String>,
    source_kind: Option<String>,
    source_agent: Option<String>,
    observed_at: Option<String>,
}

impl From<NodeProjection> for NodeRecord {
    fn from(node: NodeProjection) -> Self {
        let (source_kind, source_agent, observed_at) = match node.provenance {
            Some(ref provenance) => (
                Some(provenance.source_kind().as_str().to_string()),
                provenance.source_agent().map(str::to_string),
                provenance.observed_at().map(str::to_string),
            ),
            None => (None, None, None),
        };
        Self {
            node_id: node.node_id,
            node_kind: node.node_kind,
            title: node.title,
            summary: node.summary,
            status: node.status,
            labels: node.labels,
            properties: node.properties,
            source_kind,
            source_agent,
            observed_at,
        }
    }
}

impl NodeRecord {
    pub(crate) fn into_projection(self) -> Result<NodeProjection, PortError> {
        let provenance = match self.source_kind {
            Some(kind) => {
                let kind = SourceKind::parse(&kind).map_err(|error| {
                    PortError::InvalidState(format!(
                        "embedded store node `{}` has invalid provenance source kind: {error}",
                        self.node_id
                    ))
                })?;
                let mut provenance = Provenance::new(kind);
                if let Some(agent) = self.source_agent {
                    provenance = provenance.with_source_agent(agent);
                }
                if let Some(observed_at) = self.observed_at {
                    provenance = provenance.with_observed_at(observed_at);
                }
                Some(provenance)
            }
            None => None,
        };
        Ok(NodeProjection {
            node_id: self.node_id,
            node_kind: self.node_kind,
            title: self.title,
            summary: self.summary,
            status: self.status,
            labels: self.labels,
            properties: self.properties,
            provenance,
        })
    }
}

#[derive(Debug, Serialize, Deserialize)]
pub(crate) struct DetailRecord {
    node_id: String,
    detail: String,
    content_hash: String,
    revision: u64,
}

impl From<NodeDetailProjection> for DetailRecord {
    fn from(detail: NodeDetailProjection) -> Self {
        Self {
            node_id: detail.node_id,
            detail: detail.detail,
            content_hash: detail.content_hash,
            revision: detail.revision,
        }
    }
}

impl From<DetailRecord> for NodeDetailProjection {
    fn from(record: DetailRecord) -> Self {
        Self {
            node_id: record.node_id,
            detail: record.detail,
            content_hash: record.content_hash,
            revision: record.revision,
        }
    }
}

#[derive(Debug, Serialize, Deserialize)]
pub(crate) struct AggregateRecord {
    pub(crate) revision: u64,
    pub(crate) content_hash: String,
}

#[derive(Debug, Serialize, Deserialize)]
pub(crate) struct CheckpointRecord {
    consumer_name: String,
    stream_name: String,
    last_subject: String,
    last_event_id: String,
    last_correlation_id: String,
    last_occurred_at: String,
    processed_events: u64,
    updated_at_millis: u64,
}

impl From<ProjectionCheckpoint> for CheckpointRecord {
    fn from(checkpoint: ProjectionCheckpoint) -> Self {
        let updated_at_millis = checkpoint
            .updated_at
            .duration_since(SystemTime::UNIX_EPOCH)
            .map(|duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX))
            .unwrap_or(0);
        Self {
            consumer_name: checkpoint.consumer_name,
            stream_name: checkpoint.stream_name,
            last_subject: checkpoint.last_subject,
            last_event_id: checkpoint.last_event_id,
            last_correlation_id: checkpoint.last_correlation_id,
            last_occurred_at: checkpoint.last_occurred_at,
            processed_events: checkpoint.processed_events,
            updated_at_millis,
        }
    }
}

impl From<CheckpointRecord> for ProjectionCheckpoint {
    fn from(record: CheckpointRecord) -> Self {
        Self {
            consumer_name: record.consumer_name,
            stream_name: record.stream_name,
            last_subject: record.last_subject,
            last_event_id: record.last_event_id,
            last_correlation_id: record.last_correlation_id,
            last_occurred_at: record.last_occurred_at,
            processed_events: record.processed_events,
            updated_at: SystemTime::UNIX_EPOCH + Duration::from_millis(record.updated_at_millis),
        }
    }
}

pub(crate) fn encode_explanation(explanation: &RelationExplanation) -> Result<Vec<u8>, PortError> {
    encode("relation explanation", &explanation.to_properties())
}

pub(crate) fn decode_explanation(bytes: &[u8]) -> Result<RelationExplanation, PortError> {
    let properties: BTreeMap<String, String> = decode("relation explanation", bytes)?;
    RelationExplanation::from_properties(&properties).map_err(|error| {
        PortError::InvalidState(format!(
            "embedded store relation explanation could not be rebuilt: {error}"
        ))
    })
}