//! Intentionally boring protocol reference. It has no platform dependency and
//! receives all domain-shaped values through reviewed configuration.

use std::io::{self, BufRead, Write};

use chrono::Utc;
use cm_station_connector_sdk::{
    ConnectorObservation, ConnectorSource, ConnectorToHostMessage, HostToConnectorMessage,
    JsonObject, ObservationProtocolIdentity,
};
use serde_json::{Map, Value};

fn now() -> String {
    Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
}

fn send(message: &ConnectorToHostMessage) -> Result<(), Box<dyn std::error::Error>> {
    let mut stdout = io::stdout().lock();
    serde_json::to_writer(&mut stdout, message)?;
    stdout.write_all(b"\n")?;
    stdout.flush()?;
    Ok(())
}

fn string(configuration: &JsonObject, key: &str) -> Result<String, Box<dyn std::error::Error>> {
    configuration
        .get(key)
        .and_then(Value::as_str)
        .map(str::to_owned)
        .ok_or_else(|| format!("configuration.{key} must be a string").into())
}

fn configured_or_now(configuration: &JsonObject, key: &str) -> String {
    configuration
        .get(key)
        .and_then(Value::as_str)
        .map(str::to_owned)
        .unwrap_or_else(now)
}

fn observations(
    configuration: &JsonObject,
    default_source_id: &str,
    default_protocol_type: &str,
    protocols: &[ObservationProtocolIdentity],
) -> Result<Vec<ConnectorObservation>, Box<dyn std::error::Error>> {
    let protocol = |protocol_type: &str| {
        protocols
            .iter()
            .find(|protocol| protocol.r#type == protocol_type)
            .cloned()
            .ok_or_else(|| format!("configuration selected undeclared protocol {protocol_type}"))
    };
    let Some(configured) = configuration.get("observations").and_then(Value::as_array) else {
        let source_event_id = string(configuration, "sourceEventId")?;
        let occurred_at = configured_or_now(configuration, "occurredAt");
        return Ok(vec![ConnectorObservation {
            source_event_id,
            protocol: protocol(default_protocol_type)?,
            source_id: default_source_id.to_owned(),
            observed_at: configuration
                .get("observedAt")
                .and_then(Value::as_str)
                .map(str::to_owned)
                .unwrap_or_else(|| occurred_at.clone()),
            occurred_at,
            payload: configuration
                .get("payload")
                .and_then(Value::as_object)
                .cloned()
                .unwrap_or_default(),
        }]);
    };
    configured
        .iter()
        .map(|value| {
            let item = value
                .as_object()
                .ok_or("configuration.observations entries must be objects")?;
            let required = |key: &str| {
                item.get(key)
                    .and_then(Value::as_str)
                    .map(str::to_owned)
                    .ok_or_else(|| format!("configuration.observations.{key} must be a string"))
            };
            let occurred_at = required("occurredAt")?;
            Ok(ConnectorObservation {
                source_event_id: required("sourceEventId")?,
                protocol: protocol(
                    item.get("protocolType")
                        .and_then(Value::as_str)
                        .unwrap_or(default_protocol_type),
                )?,
                source_id: item
                    .get("sourceId")
                    .and_then(Value::as_str)
                    .unwrap_or(default_source_id)
                    .to_owned(),
                observed_at: item
                    .get("observedAt")
                    .and_then(Value::as_str)
                    .unwrap_or(&occurred_at)
                    .to_owned(),
                occurred_at,
                payload: item
                    .get("payload")
                    .and_then(Value::as_object)
                    .cloned()
                    .unwrap_or_default(),
            })
        })
        .collect::<Result<Vec<_>, String>>()
        .map_err(Into::into)
}

fn receive_observation_receipt(
    lines: &mut impl Iterator<Item = io::Result<String>>,
    session_id: &str,
    source_event_id: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    loop {
        let message: HostToConnectorMessage =
            serde_json::from_str(&lines.next().ok_or("host closed protocol")??)?;
        match message {
            HostToConnectorMessage::ObservationReceipt {
                session_id: received,
                source_event_id: received_event,
                disposition,
                ..
            } if received == session_id && received_event == source_event_id => {
                if disposition == "rejected" {
                    return Err("host rejected reference observation".into());
                }
                return Ok(());
            }
            HostToConnectorMessage::Ping {
                session_id: received,
                nonce,
            } if received == session_id => send(&ConnectorToHostMessage::Pong {
                session_id: session_id.to_owned(),
                nonce,
                occurred_at: now(),
            })?,
            _ => return Err("unexpected host protocol message while awaiting receipt".into()),
        }
    }
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let stdin = io::stdin();
    let mut lines = stdin.lock().lines();
    let initialize: HostToConnectorMessage =
        serde_json::from_str(&lines.next().ok_or("missing initialize")??)?;
    let (session_id, identity, configuration, observation_protocols) = match initialize {
        HostToConnectorMessage::Initialize {
            session_id,
            protocol_version: 1,
            expected_identity,
            observation_protocols,
            configuration,
            ..
        } => (
            session_id,
            expected_identity,
            configuration,
            observation_protocols,
        ),
        _ => return Err("expected Station v1 initialize".into()),
    };
    send(&ConnectorToHostMessage::Hello {
        session_id: session_id.clone(),
        protocol_version: 1,
        identity,
    })?;
    send(&ConnectorToHostMessage::Status {
        session_id: session_id.clone(),
        status: "starting".into(),
        detail: None,
        occurred_at: now(),
    })?;

    let start: HostToConnectorMessage =
        serde_json::from_str(&lines.next().ok_or("missing start")??)?;
    if !matches!(start, HostToConnectorMessage::Start { session_id: ref received } if received == &session_id)
    {
        return Err("expected start for this process session".into());
    }
    send(&ConnectorToHostMessage::Status {
        session_id: session_id.clone(),
        status: "ready".into(),
        detail: None,
        occurred_at: now(),
    })?;

    let source_id = string(&configuration, "sourceId")?;
    let source_type = string(&configuration, "sourceType")?;
    let observation_protocol_type = string(&configuration, "observationProtocolType")?;
    let source = ConnectorSource {
        source_id: source_id.clone(),
        source_type,
        display_name: string(&configuration, "displayName")?,
        metadata: Map::new(),
    };
    send(&ConnectorToHostMessage::Sources {
        session_id: session_id.clone(),
        revision: 1,
        observed_at: configured_or_now(&configuration, "inventoryObservedAt"),
        sources: vec![source],
    })?;
    send(&ConnectorToHostMessage::Health {
        session_id: session_id.clone(),
        status: "healthy".into(),
        summary: None,
        occurred_at: now(),
        checks: vec![],
    })?;
    for observation in observations(
        &configuration,
        &source_id,
        &observation_protocol_type,
        &observation_protocols,
    )? {
        let source_event_id = observation.source_event_id.clone();
        send(&ConnectorToHostMessage::Observation {
            session_id: session_id.clone(),
            observation,
        })?;
        receive_observation_receipt(&mut lines, &session_id, &source_event_id)?;
    }

    loop {
        let message: HostToConnectorMessage =
            serde_json::from_str(&lines.next().ok_or("host closed protocol")??)?;
        match message {
            HostToConnectorMessage::Ping {
                session_id: received,
                nonce,
            } if received == session_id => send(&ConnectorToHostMessage::Pong {
                session_id: session_id.clone(),
                nonce,
                occurred_at: now(),
            })?,
            HostToConnectorMessage::Stop {
                session_id: received,
                ..
            } if received == session_id => {
                send(&ConnectorToHostMessage::Status {
                    session_id: session_id.clone(),
                    status: "stopping".into(),
                    detail: None,
                    occurred_at: now(),
                })?;
                send(&ConnectorToHostMessage::Status {
                    session_id,
                    status: "stopped".into(),
                    detail: None,
                    occurred_at: now(),
                })?;
                return Ok(());
            }
            _ => return Err("unexpected host protocol message".into()),
        }
    }
}
