// Copyright Claudio Mattera 2019.
// Distributed under the MIT License.
// See accompanying file License.txt, or online at
// https://opensource.org/licenses/MIT

use std::env;

use log::*;

use postgres::{Connection, TlsMode};

use url::Url;

use crate::error::StorageError;
use crate::storage::Storage;
use crate::types::{Timestamp, TimeSeries, Path, Username, PasswordHash};

#[derive(Debug)]
pub struct PostgresStorage {
    connection: Connection,
}

impl PostgresStorage {
    pub fn new(host: &str) -> Result<Self, StorageError> {
        let mut url = Url::parse(host)?;
        info!("Postgres database host: {}", url);

        if let Some(password_os) = env::var_os("POSTGRES_PASSWORD") {
            let password = password_os.into_string().map_err(|_| {
                StorageError::new(
                    "Invalid POSTGRES_PASSWORD environment variable content".to_string(),
                )
            })?;
            info!("Using Postgres password from POSTGRES_PASSWORD environment variable");
            url.set_password(Some(&password))
                .map_err(|_| StorageError::new("Invalid URL".to_string()))?;
        }

        debug!("Removing POSTGRES_PASSWORD from environment");
        env::remove_var("POSTGRES_PASSWORD");

        let connection_string = url.into_string();
        let connection = Connection::connect(connection_string, TlsMode::None)?;

        Ok(PostgresStorage { connection })
    }

    fn _store_data(
        self: &mut Self,
        path: &Path,
        time_series: TimeSeries,
    ) -> Result<(), StorageError> {
        debug!(
            "Inserting {1} new values for time-series {0}",
            path,
            time_series.len()
        );

        debug!("Opening transaction");
        let transaction = self.connection.transaction()?;

        for (instant, value) in time_series {
            transaction.execute(
                "INSERT INTO stanley_time_series (path, instant, value)
                VALUES ($1, $2, $3)",
                &[&path.to_str(), &instant, &value],
            )?;
        }
        debug!("Committing transaction");
        transaction.commit()?;

        Ok(())
    }

    fn _fetch_data(
        self: &Self,
        path: &Path,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
    ) -> Result<TimeSeries, StorageError> {
        debug!("Fetching values for time-series {0}", path);

        let statement = self.connection.prepare(
            "SELECT instant, value FROM stanley_time_series
            WHERE instant > $1 AND instant < $2 AND path=$3
            ORDER BY instant",
        )?;
        let rows = statement.query(&[
            &start.unwrap_or(Timestamp::min_value()),
            &end.unwrap_or(Timestamp::max_value()),
            &path.to_str(),
        ])?;

        let mut readings = vec![];
        for row in &rows {
            let instant: i64 = row.get(0);
            let value: f64 = row.get(1);
            readings.push((instant, value));
        }

        Ok(TimeSeries::from_vec(readings))
    }
}

impl Storage for PostgresStorage {
    fn store_data(
        self: &mut Self,
        path: &Path,
        time_series: TimeSeries,
    ) -> Result<(), StorageError> {
        self._store_data(path, time_series)
    }

    fn fetch_data(
        self: &Self,
        path: &Path,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
    ) -> Result<TimeSeries, StorageError> {
        self._fetch_data(path, start, end)
    }

    fn fetch_password_hash(
        self: &Self,
        username: &Username,
    ) -> Result<Option<PasswordHash>, StorageError> {
        let statement = self.connection.prepare(
            "SELECT password_hash FROM stanley_user
            WHERE username = $1",
        )?;
        let rows = statement.query(&[&username.to_str()])?;
        if rows.is_empty() {
            return Ok(None);
        }
        let row = rows.get(0);

        let stored_password_hash = PasswordHash::new(row.get(0));
        Ok(Some(stored_password_hash))
    }

    fn validate_permission(self: &Self, username: &Username, path: &Path) -> Result<bool, StorageError> {
        let statement = self.connection.prepare(
            "SELECT prefix FROM stanley_permission
            WHERE username = $1",
        )?;
        let rows = statement.query(&[&username.to_str()])?;
        for row in &rows {
            let prefix: String = row.get(0);
            if path.to_str().starts_with(&prefix) {
                return Ok(true);
            }
        }
        Ok(false)
    }
}
