aphomie_db_client 0.3.9

use anyhow::bail;
use bms_dtos::TrendDataRaw;
use chrono::{DateTime, NaiveDateTime, TimeDelta, Utc};
use serde::{Deserialize, Serialize, ser::SerializeStruct};
use tokio_postgres::{Client, Config, NoTls, Row, ToStatement, types::ToSql};

type Result<T> = core::result::Result<T, anyhow::Error>;

pub struct AphomieDbClient {
    db_client: Client,
}

impl AphomieDbClient {
    pub async fn new(host: Option<&str>, port: Option<u16>) -> Result<Self> {
        let host = if let Some(v) = host { v } else { "localhost" };

        let port = if let Some(v) = port { v } else { 5432 };

        let mut config = Config::new();
        config
            .user("dat7")
            .password("dat7")
            .dbname("dat7")
            .host(host)
            .port(port);

        let (db_client, connection) = config.connect(NoTls).await?;

        tokio::spawn(async move {
            if let Err(e) = connection.await {
                eprintln!("connection error: {}", e);
            }
        });

        Ok(Self { db_client })
    }

    pub async fn query<T>(&self, statement: &T, params: &[&(dyn ToSql + Sync)]) -> Result<Vec<Row>>
    where
        T: ?Sized + ToStatement,
    {
        self.db_client
            .query(statement, params)
            .await
            .map_err(|e| anyhow::anyhow!("failed performing query: {:?}", e))
    }

    pub async fn insert_metadata(&self, metadata: &MetadataDb) -> Result<Vec<MetadataDb>> {
        insert_metadata(
            &self.db_client,
            &metadata.log_id,
            &metadata.source,
            &metadata.unit,
            &metadata.alias,
        )
        .await
        .map(|v| v.iter().map(|r| MetadataDb::from(r)).collect())
    }

    pub async fn get_metadata_by_source(&self, source: &str) -> Result<Vec<MetadataDb>> {
        get_metadata_by_source(&self.db_client, source)
            .await
            .map(|v| v.iter().map(|r| MetadataDb::from(r)).collect())
    }

    pub async fn get_latest_trend_like_source_and_alias(
        &self,
        source: &str,
        alias: &str,
    ) -> Result<Vec<MetadataTrendDataJoinDb>> {
        get_latest_trend_like_source_and_alias(&self.db_client, source, alias)
            .await
            .map(|v| v.iter().map(|r| MetadataTrendDataJoinDb::from(r)).collect())
    }

    pub async fn get_latest_trend_like_source(
        &self,
        source: &str,
    ) -> Result<Vec<MetadataTrendDataJoinDb>> {
        get_latest_trend_like_source(&self.db_client, source)
            .await
            .map(|v| v.iter().map(|r| MetadataTrendDataJoinDb::from(r)).collect())
    }

    pub async fn get_all_metadata(&self) -> Result<Vec<MetadataDb>> {
        get_all_metadata(&self.db_client)
            .await
            .map(|v| v.iter().map(|r| MetadataDb::from(r)).collect())
    }

    pub async fn insert_trend_data(&self, trend_data: &TrendDataDb) -> Result<Vec<TrendDataDb>> {
        insert_trend_data(
            &self.db_client,
            &trend_data.log_id,
            &trend_data.timestamp,
            &trend_data.value,
        )
        .await
        .map(|v| v.iter().map(|r| TrendDataDb::from(r)).collect())
    }

    pub async fn get_trend_data_by_period(
        &self,
        start: &DateTime<Utc>,
        end: &DateTime<Utc>,
    ) -> Result<Vec<TrendDataDb>> {
        get_trend_data_by_period(&self.db_client, start, end)
            .await
            .map(|v| v.iter().map(|r| TrendDataDb::from(r)).collect())
    }

    pub async fn get_trend_data_by_source_offset_limit(
        &self,
        source: &str,
        offset: i64,
        limit: i64,
    ) -> Result<Vec<MetadataTrendDataJoinDb>> {
        get_trend_data_by_source_offset_limit(&self.db_client, source, offset, limit)
            .await
            .map(|v| v.iter().map(|r| MetadataTrendDataJoinDb::from(r)).collect())
    }

    pub async fn get_metadata_join_trend_data_offset_limit(
        &self,
        offset: i64,
        limit: i64,
    ) -> Result<Vec<MetadataTrendDataJoinDb>> {
        get_metadata_join_trend_data_offset_limit(&self.db_client, offset, limit)
            .await
            .map(|v| v.iter().map(|r| MetadataTrendDataJoinDb::from(r)).collect())
    }

    pub async fn get_latest_trend_data(&self) -> Result<Option<TrendDataDb>> {
        match get_latest_trend_data(&self.db_client).await.map(|v| {
            v.iter()
                .map(|e| TrendDataDb::from(e))
                .collect::<Vec<TrendDataDb>>()
        }) {
            Ok(v) => Ok(v.get(0).map(|e| e.clone())),
            Err(e) => bail!(e),
        }
    }

    pub async fn get_latest_trend_by_log_id(&self, log_id: i64) -> Result<Option<TrendDataDb>> {
        get_latest_trend_by_log_id(&self.db_client, log_id)
            .await
            .map(|v| {
                v.iter()
                    .map(|r| TrendDataDb::from(r))
                    .collect::<Vec<TrendDataDb>>()
            })
            .map(|e| e.get(0).map(|e| e.clone()))
    }

    pub async fn get_metadata_by_log_id(&self, log_id: i64) -> Result<Option<MetadataDb>> {
        get_metadata_by_id(&self.db_client, log_id)
            .await
            .map(|v| {
                v.iter()
                    .map(|r| MetadataDb::from(r))
                    .collect::<Vec<MetadataDb>>()
            })
            .map(|e| e.get(0).map(|e| e.clone()))
    }

    pub async fn get_trends_like_source_and_alias_in_dt_range_count(
        &self,
        source: &str,
        alias: &str,
        start: &DateTime<Utc>,
        end: &DateTime<Utc>,
        offset: i64,
        limit: i64,
    ) -> Result<Vec<MetadataTrendDataJoinDb>> {
        get_trends_like_source_and_alias_count(
            &self.db_client,
            source,
            alias,
            start,
            end,
            offset,
            limit,
        )
        .await
        .map(|v| v.iter().map(|r| MetadataTrendDataJoinDb::from(r)).collect())
    }
}

async fn insert_metadata(
    client: &Client,
    log_id: &i64,
    source: &str,
    unit: &str,
    alias: &Option<String>,
) -> Result<Vec<Row>> {
    let stmt =
        "INSERT INTO metadata (log_id, source, unit, alias) VALUES ($1, $2, $3, $4) RETURNING *;";

    match client.query(stmt, &[log_id, &source, &unit, &alias]).await {
        Ok(v) => Ok(v),
        Err(e) => {
            bail!("error querying database: {e}");
        }
    }
}

async fn get_metadata_by_source(client: &Client, source: &str) -> Result<Vec<Row>> {
    let stmt = "SELECT * FROM metadata WHERE source = $1;";

    match client.query(stmt, &[&source]).await {
        Ok(v) => Ok(v),
        Err(e) => {
            bail!("error querying database: {e}");
        }
    }
}

async fn get_latest_trend_like_source_and_alias(
    client: &Client,
    source: &str,
    alias: &str,
) -> Result<Vec<Row>> {
    let stmt = "SELECT m.log_id, m.source, m.unit, m.alias, t.timestamp, t.value FROM metadata m JOIN trend_data t on t.log_id = m.log_id WHERE m.source LIKE $1 AND m.alias = $2 ORDER BY t.timestamp DESC LIMIT 1;";

    match client.query(stmt, &[&source, &alias]).await {
        Ok(v) => Ok(v),
        Err(e) => {
            bail!("error querying database: {e}");
        }
    }
}

async fn get_latest_trend_like_source(client: &Client, source: &str) -> Result<Vec<Row>> {
    let stmt = "SELECT m.log_id, m.source, m.unit, m.alias, t.timestamp, t.value FROM metadata m JOIN trend_data t on t.log_id = m.log_id WHERE m.source LIKE $1 ORDER BY t.timestamp DESC LIMIT 1;";

    match client.query(stmt, &[&source]).await {
        Ok(v) => Ok(v),
        Err(e) => {
            bail!("error querying database: {e}");
        }
    }
}

async fn get_metadata_by_id(client: &Client, log_id: i64) -> Result<Vec<Row>> {
    let stmt = "SELECT * FROM metadata WHERE log_id = $1;";

    match client.query(stmt, &[&log_id]).await {
        Ok(v) => Ok(v),
        Err(e) => {
            bail!("error querying database: {e}");
        }
    }
}

async fn get_latest_trend_by_log_id(client: &Client, log_id: i64) -> Result<Vec<Row>> {
    let stmt = "SELECT * FROM trend_data WHERE log_id = $1 ORDER BY timestamp DESC LIMIT 1;";

    match client.query(stmt, &[&log_id]).await {
        Ok(v) => Ok(v),
        Err(e) => {
            bail!("error querying database: {e}");
        }
    }
}

async fn get_all_metadata(client: &Client) -> Result<Vec<Row>> {
    let stmt = "SELECT * FROM metadata;";

    match client.query(stmt, &[]).await {
        Ok(v) => Ok(v),
        Err(e) => {
            bail!("error querying database: {e}");
        }
    }
}

async fn get_trend_data_by_period(
    client: &Client,
    start: &DateTime<Utc>,
    end: &DateTime<Utc>,
) -> Result<Vec<Row>> {
    let stmt = "SELECT * FROM trend_data WHERE timestamp <= $1 AND timestamp >= $2;";

    match client.query(stmt, &[&end, &start]).await {
        Ok(v) => Ok(v),
        Err(e) => {
            bail!("error querying database: {e}");
        }
    }
}

async fn get_trend_data_by_source_offset_limit(
    client: &Client,
    source: &str,
    offset: i64,
    limit: i64,
) -> Result<Vec<Row>> {
    let stmt = "SELECT t.log_id, t.timestamp, t.value, m.source, m.unit FROM metadata m JOIN trend_data t ON m.log_id = t.log_id WHERE source LIKE $1 ORDER BY t.log_id, t.timestamp OFFSET $2 LIMIT $3;";

    match client.query(stmt, &[&source, &offset, &limit]).await {
        Ok(v) => Ok(v),
        Err(e) => {
            bail!("error querying database: {e}");
        }
    }
}

async fn get_metadata_join_trend_data_offset_limit(
    client: &Client,
    offset: i64,
    limit: i64,
) -> Result<Vec<Row>> {
    let stmt = "SELECT t.log_id, t.timestamp, t.value, m.source, m.unit FROM metadata m JOIN trend_data t ON m.log_id = t.log_id ORDER BY t.timestamp, t.log_id OFFSET $1 LIMIT $2;";

    match client.query(stmt, &[&offset, &limit]).await {
        Ok(v) => Ok(v),
        Err(e) => {
            bail!("error querying database: {e}");
        }
    }
}

async fn get_latest_trend_data(client: &Client) -> Result<Vec<Row>> {
    let stmt = "SELECT * FROM trend_data ORDER BY timestamp DESC LIMIT 1";

    match client.query(stmt, &[]).await {
        Ok(v) => Ok(v),
        Err(e) => {
            bail!("error querying database: {e}");
        }
    }
}

async fn insert_trend_data(
    client: &Client,
    log_id: &i64,
    timestamp: &DateTime<Utc>,
    value: &Option<f64>,
) -> Result<Vec<Row>> {
    let stmt = "INSERT INTO trend_data (log_id, timestamp, value) VALUES ($1, $2, $3) RETURNING *;";

    match client.query(stmt, &[log_id, &timestamp, &value]).await {
        Ok(v) => Ok(v),
        Err(e) => {
            bail!("error querying database: {e}");
        }
    }
}

async fn get_trends_like_source_and_alias_count(
    client: &Client,
    source: &str,
    alias: &str,
    start: &DateTime<Utc>,
    end: &DateTime<Utc>,
    offset: i64,
    limit: i64,
) -> Result<Vec<Row>> {
    let stmt = "
SELECT m.log_id, m.source, m.unit, m.alias, t.timestamp, t.value 
FROM metadata m JOIN trend_data t on t.log_id = m.log_id 
WHERE 1=1
    AND m.source LIKE $1 
    AND m.alias LIKE $2
    AND t.timestamp >= $3
    AND t.timestamp < $4
ORDER BY t.timestamp OFFSET $5 LIMIT $6;
";

    match client
        .query(stmt, &[&source, &alias, start, end, &offset, &limit])
        .await
    {
        Ok(v) => Ok(v),
        Err(e) => {
            bail!("error querying database: {e}");
        }
    }
}

#[derive(Debug, Clone)]
pub struct MetadataTrendDataJoinDb {
    pub log_id: i64,
    pub source: String,
    pub unit: String,
    pub timestamp: DateTime<Utc>,
    pub value: Option<f64>,
    pub alias: Option<String>,
}

#[derive(Debug, Clone)]
pub struct TrendDataDb {
    pub log_id: i64,
    pub timestamp: DateTime<Utc>,
    pub value: Option<f64>,
}

impl Serialize for MetadataTrendDataJoinDb {
    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        let mut s = serializer.serialize_struct("TrendDataDb", 3)?;
        s.serialize_field("log_id", &self.log_id)?;
        s.serialize_field("timestamp", &self.timestamp.to_rfc3339())?;
        s.serialize_field("value", &self.value)?;
        s.serialize_field("source", &self.source)?;
        s.serialize_field("unit", &self.unit)?;
        s.serialize_field("alias", &self.alias)?;
        s.end()
    }
}

impl Serialize for TrendDataDb {
    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        let mut s = serializer.serialize_struct("TrendDataDb", 3)?;
        s.serialize_field("log_id", &self.log_id)?;
        s.serialize_field("timestamp", &self.timestamp.to_rfc3339())?;
        s.serialize_field("value", &self.value)?;
        s.end()
    }
}

#[derive(Clone, Serialize, Deserialize, Debug)]
pub struct MetadataDb {
    pub log_id: i64,
    pub source: String,
    pub unit: String,
    pub alias: Option<String>,
}

impl From<&Row> for MetadataTrendDataJoinDb {
    fn from(value: &Row) -> Self {
        let log_id: i64 = value.get("log_id");
        let source: String = value.get("source");
        let unit: String = value.get("unit");
        let timestamp: DateTime<Utc> = value.get("timestamp");
        let alias: Option<String> = value.get("alias");
        let value: Option<f64> = value.get("value");

        MetadataTrendDataJoinDb {
            value,
            log_id,
            source,
            unit,
            timestamp,
            alias,
        }
    }
}

impl From<&Row> for MetadataDb {
    fn from(value: &Row) -> MetadataDb {
        let log_id: i64 = value.get("log_id");
        let source: String = value.get("source");
        let unit: String = value.get("unit");
        let alias: Option<String> = value.get("alias");

        MetadataDb {
            log_id,
            source,
            unit,
            alias,
        }
    }
}

impl TryFrom<&TrendDataRaw> for TrendDataDb {
    type Error = anyhow::Error;

    fn try_from(value: &TrendDataRaw) -> core::result::Result<Self, Self::Error> {
        let naive_timestamp = NaiveDateTime::parse_from_str(&value.timestamp, "%Y-%m-%d %H:%M:%S");
        if let Err(e) = naive_timestamp {
            let timestamp = &value.timestamp;
            bail!("error parsing {timestamp}: {e}");
        }
        let mut naive_timestamp = naive_timestamp.unwrap();
        match value.timestamp_tzinfo {
            248 => {
                // CET
                naive_timestamp = naive_timestamp - TimeDelta::hours(1);
            }
            241 => {
                // CEST
                naive_timestamp = naive_timestamp - TimeDelta::hours(2);
            }
            _ => {
                let timezone = value.timestamp_tzinfo;
                bail!("did not recognize timestamp timezone {timezone}");
            }
        }

        let utc_time: DateTime<Utc> = naive_timestamp.and_local_timezone(Utc).unwrap();

        Ok(TrendDataDb {
            log_id: value.externallogid,
            timestamp: utc_time,
            value: value.value,
        })
    }
}

impl From<&Row> for TrendDataDb {
    fn from(row: &Row) -> TrendDataDb {
        let log_id: i64 = row.get("log_id");
        let timestamp: DateTime<Utc> = row.get("timestamp");
        let value: Option<f64> = row.get("value");

        TrendDataDb {
            log_id,
            timestamp,
            value,
        }
    }
}