// Copyright 2019. The Tari Project
//
// Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
// following conditions are met:
//
// 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
// disclaimer.
//
// 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
// following disclaimer in the documentation and/or other materials provided with the distribution.
//
// 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
// products derived from this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
// INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
// WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
// USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

use crate::{
    output_manager_service::TxId,
    schema::{completed_transactions, inbound_transactions, outbound_transactions},
    transaction_service::{
        error::TransactionStorageError,
        storage::database::{
            CompletedTransaction,
            DbKey,
            DbKeyValuePair,
            DbValue,
            InboundTransaction,
            OutboundTransaction,
            TransactionBackend,
            TransactionStatus,
            WriteOperation,
        },
    },
};
use chrono::NaiveDateTime;
use diesel::{prelude::*, result::Error as DieselError, SqliteConnection};
use std::{
    collections::HashMap,
    convert::TryFrom,
    sync::{Arc, Mutex, MutexGuard},
};
use tari_comms::types::CommsPublicKey;
use tari_core::transactions::{tari_amount::MicroTari, types::PublicKey};
use tari_crypto::tari_utilities::ByteArray;

/// A Sqlite backend for the Transaction Service. The Backend is accessed via a connection pool to the Sqlite file.
#[derive(Clone)]
pub struct TransactionServiceSqliteDatabase {
    database_connection: Arc<Mutex<SqliteConnection>>,
}
impl TransactionServiceSqliteDatabase {
    pub fn new(database_connection: Arc<Mutex<SqliteConnection>>) -> Self {
        Self { database_connection }
    }

    fn insert(kvp: DbKeyValuePair, conn: MutexGuard<SqliteConnection>) -> Result<(), TransactionStorageError> {
        match kvp {
            DbKeyValuePair::PendingOutboundTransaction(k, v) => {
                if OutboundTransactionSql::find(k, false, &(*conn)).is_ok() {
                    return Err(TransactionStorageError::DuplicateOutput);
                }
                OutboundTransactionSql::try_from(*v)?.commit(&(*conn))?;
            },
            DbKeyValuePair::PendingInboundTransaction(k, v) => {
                if InboundTransactionSql::find(k, false, &(*conn)).is_ok() {
                    return Err(TransactionStorageError::DuplicateOutput);
                }
                InboundTransactionSql::try_from(*v)?.commit(&(*conn))?;
            },
            DbKeyValuePair::CompletedTransaction(k, v) => {
                if CompletedTransactionSql::find(k, false, &(*conn)).is_ok() {
                    return Err(TransactionStorageError::DuplicateOutput);
                }
                CompletedTransactionSql::try_from(*v)?.commit(&(*conn))?;
            },
        }
        Ok(())
    }

    fn remove(key: DbKey, conn: MutexGuard<SqliteConnection>) -> Result<Option<DbValue>, TransactionStorageError> {
        match key {
            DbKey::PendingOutboundTransaction(k) => match OutboundTransactionSql::find(k, false, &(*conn)) {
                Ok(v) => {
                    v.delete(&(*conn))?;
                    Ok(Some(DbValue::PendingOutboundTransaction(Box::new(
                        OutboundTransaction::try_from(v)?,
                    ))))
                },
                Err(TransactionStorageError::DieselError(DieselError::NotFound)) => Err(
                    TransactionStorageError::ValueNotFound(DbKey::PendingOutboundTransaction(k)),
                ),
                Err(e) => Err(e),
            },
            DbKey::PendingInboundTransaction(k) => match InboundTransactionSql::find(k, false, &(*conn)) {
                Ok(v) => {
                    v.delete(&(*conn))?;
                    Ok(Some(DbValue::PendingInboundTransaction(Box::new(
                        InboundTransaction::try_from(v)?,
                    ))))
                },
                Err(TransactionStorageError::DieselError(DieselError::NotFound)) => Err(
                    TransactionStorageError::ValueNotFound(DbKey::PendingOutboundTransaction(k)),
                ),
                Err(e) => Err(e),
            },
            DbKey::CompletedTransaction(k) => match CompletedTransactionSql::find(k, false, &(*conn)) {
                Ok(v) => {
                    v.delete(&(*conn))?;
                    Ok(Some(DbValue::CompletedTransaction(Box::new(
                        CompletedTransaction::try_from(v)?,
                    ))))
                },
                Err(TransactionStorageError::DieselError(DieselError::NotFound)) => {
                    Err(TransactionStorageError::ValueNotFound(DbKey::CompletedTransaction(k)))
                },
                Err(e) => Err(e),
            },
            DbKey::PendingOutboundTransactions => Err(TransactionStorageError::OperationNotSupported),
            DbKey::PendingInboundTransactions => Err(TransactionStorageError::OperationNotSupported),
            DbKey::CompletedTransactions => Err(TransactionStorageError::OperationNotSupported),
            DbKey::CancelledPendingOutboundTransactions => Err(TransactionStorageError::OperationNotSupported),
            DbKey::CancelledPendingInboundTransactions => Err(TransactionStorageError::OperationNotSupported),
            DbKey::CancelledCompletedTransactions => Err(TransactionStorageError::OperationNotSupported),
            DbKey::CancelledPendingOutboundTransaction(k) => match OutboundTransactionSql::find(k, true, &(*conn)) {
                Ok(v) => {
                    v.delete(&(*conn))?;
                    Ok(Some(DbValue::PendingOutboundTransaction(Box::new(
                        OutboundTransaction::try_from(v)?,
                    ))))
                },
                Err(TransactionStorageError::DieselError(DieselError::NotFound)) => Err(
                    TransactionStorageError::ValueNotFound(DbKey::CancelledPendingOutboundTransaction(k)),
                ),
                Err(e) => Err(e),
            },
            DbKey::CancelledPendingInboundTransaction(k) => match InboundTransactionSql::find(k, true, &(*conn)) {
                Ok(v) => {
                    v.delete(&(*conn))?;
                    Ok(Some(DbValue::PendingInboundTransaction(Box::new(
                        InboundTransaction::try_from(v)?,
                    ))))
                },
                Err(TransactionStorageError::DieselError(DieselError::NotFound)) => Err(
                    TransactionStorageError::ValueNotFound(DbKey::CancelledPendingOutboundTransaction(k)),
                ),
                Err(e) => Err(e),
            },
            DbKey::CancelledCompletedTransaction(k) => match CompletedTransactionSql::find(k, true, &(*conn)) {
                Ok(v) => {
                    v.delete(&(*conn))?;
                    Ok(Some(DbValue::CompletedTransaction(Box::new(
                        CompletedTransaction::try_from(v)?,
                    ))))
                },
                Err(TransactionStorageError::DieselError(DieselError::NotFound)) => Err(
                    TransactionStorageError::ValueNotFound(DbKey::CancelledCompletedTransaction(k)),
                ),
                Err(e) => Err(e),
            },
        }
    }
}

impl TransactionBackend for TransactionServiceSqliteDatabase {
    fn fetch(&self, key: &DbKey) -> Result<Option<DbValue>, TransactionStorageError> {
        let conn = acquire_lock!(self.database_connection);

        let result = match key {
            DbKey::PendingOutboundTransaction(t) => match OutboundTransactionSql::find(*t, false, &(*conn)) {
                Ok(o) => Some(DbValue::PendingOutboundTransaction(Box::new(
                    OutboundTransaction::try_from(o)?,
                ))),
                Err(TransactionStorageError::DieselError(DieselError::NotFound)) => None,
                Err(e) => return Err(e),
            },
            DbKey::PendingInboundTransaction(t) => match InboundTransactionSql::find(*t, false, &(*conn)) {
                Ok(o) => Some(DbValue::PendingInboundTransaction(Box::new(
                    InboundTransaction::try_from(o)?,
                ))),
                Err(TransactionStorageError::DieselError(DieselError::NotFound)) => None,
                Err(e) => return Err(e),
            },
            DbKey::CompletedTransaction(t) => match CompletedTransactionSql::find(*t, false, &(*conn)) {
                Ok(o) => Some(DbValue::CompletedTransaction(Box::new(CompletedTransaction::try_from(
                    o,
                )?))),
                Err(TransactionStorageError::DieselError(DieselError::NotFound)) => None,
                Err(e) => return Err(e),
            },
            DbKey::PendingOutboundTransactions => Some(DbValue::PendingOutboundTransactions(
                OutboundTransactionSql::index(&(*conn), false)?
                    .iter()
                    .fold(HashMap::new(), |mut acc, x| {
                        if let Ok(v) = OutboundTransaction::try_from((*x).clone()) {
                            acc.insert(x.tx_id as u64, v);
                        }
                        acc
                    }),
            )),
            DbKey::PendingInboundTransactions => Some(DbValue::PendingInboundTransactions(
                InboundTransactionSql::index(&(*conn), false)?
                    .iter()
                    .fold(HashMap::new(), |mut acc, x| {
                        if let Ok(v) = InboundTransaction::try_from((*x).clone()) {
                            acc.insert(x.tx_id as u64, v);
                        }
                        acc
                    }),
            )),
            DbKey::CompletedTransactions => Some(DbValue::CompletedTransactions(
                CompletedTransactionSql::index(&(*conn), false)?
                    .iter()
                    .fold(HashMap::new(), |mut acc, x| {
                        if let Ok(v) = CompletedTransaction::try_from((*x).clone()) {
                            acc.insert(x.tx_id as u64, v);
                        }
                        acc
                    }),
            )),
            DbKey::CancelledPendingOutboundTransactions => Some(DbValue::PendingOutboundTransactions(
                OutboundTransactionSql::index(&(*conn), true)?
                    .iter()
                    .fold(HashMap::new(), |mut acc, x| {
                        if let Ok(v) = OutboundTransaction::try_from((*x).clone()) {
                            acc.insert(x.tx_id as u64, v);
                        }
                        acc
                    }),
            )),
            DbKey::CancelledPendingInboundTransactions => Some(DbValue::PendingInboundTransactions(
                InboundTransactionSql::index(&(*conn), true)?
                    .iter()
                    .fold(HashMap::new(), |mut acc, x| {
                        if let Ok(v) = InboundTransaction::try_from((*x).clone()) {
                            acc.insert(x.tx_id as u64, v);
                        }
                        acc
                    }),
            )),
            DbKey::CancelledCompletedTransactions => Some(DbValue::CompletedTransactions(
                CompletedTransactionSql::index(&(*conn), true)?
                    .iter()
                    .fold(HashMap::new(), |mut acc, x| {
                        if let Ok(v) = CompletedTransaction::try_from((*x).clone()) {
                            acc.insert(x.tx_id as u64, v);
                        }
                        acc
                    }),
            )),
            DbKey::CancelledPendingOutboundTransaction(t) => match OutboundTransactionSql::find(*t, true, &(*conn)) {
                Ok(o) => Some(DbValue::PendingOutboundTransaction(Box::new(
                    OutboundTransaction::try_from(o)?,
                ))),
                Err(TransactionStorageError::DieselError(DieselError::NotFound)) => None,
                Err(e) => return Err(e),
            },
            DbKey::CancelledPendingInboundTransaction(t) => match InboundTransactionSql::find(*t, true, &(*conn)) {
                Ok(o) => Some(DbValue::PendingInboundTransaction(Box::new(
                    InboundTransaction::try_from(o)?,
                ))),
                Err(TransactionStorageError::DieselError(DieselError::NotFound)) => None,
                Err(e) => return Err(e),
            },
            DbKey::CancelledCompletedTransaction(t) => match CompletedTransactionSql::find(*t, true, &(*conn)) {
                Ok(o) => Some(DbValue::CompletedTransaction(Box::new(CompletedTransaction::try_from(
                    o,
                )?))),
                Err(TransactionStorageError::DieselError(DieselError::NotFound)) => None,
                Err(e) => return Err(e),
            },
        };

        Ok(result)
    }

    fn contains(&self, key: &DbKey) -> Result<bool, TransactionStorageError> {
        let conn = acquire_lock!(self.database_connection);

        let result = match key {
            DbKey::PendingOutboundTransaction(k) => OutboundTransactionSql::find(*k, false, &(*conn)).is_ok(),
            DbKey::PendingInboundTransaction(k) => InboundTransactionSql::find(*k, false, &(*conn)).is_ok(),
            DbKey::CompletedTransaction(k) => CompletedTransactionSql::find(*k, false, &(*conn)).is_ok(),
            DbKey::PendingOutboundTransactions => false,
            DbKey::PendingInboundTransactions => false,
            DbKey::CompletedTransactions => false,
            DbKey::CancelledPendingOutboundTransactions => false,
            DbKey::CancelledPendingInboundTransactions => false,
            DbKey::CancelledCompletedTransactions => false,
            DbKey::CancelledPendingOutboundTransaction(k) => OutboundTransactionSql::find(*k, true, &(*conn)).is_ok(),
            DbKey::CancelledPendingInboundTransaction(k) => InboundTransactionSql::find(*k, true, &(*conn)).is_ok(),
            DbKey::CancelledCompletedTransaction(k) => CompletedTransactionSql::find(*k, true, &(*conn)).is_ok(),
        };

        Ok(result)
    }

    fn write(&self, op: WriteOperation) -> Result<Option<DbValue>, TransactionStorageError> {
        let conn = acquire_lock!(self.database_connection);

        match op {
            WriteOperation::Insert(kvp) => TransactionServiceSqliteDatabase::insert(kvp, conn).map(|_| None),

            WriteOperation::Remove(key) => TransactionServiceSqliteDatabase::remove(key, conn),
        }
    }

    fn transaction_exists(&self, tx_id: u64) -> Result<bool, TransactionStorageError> {
        let conn = acquire_lock!(self.database_connection);

        Ok(OutboundTransactionSql::find(tx_id, false, &(*conn)).is_ok() ||
            InboundTransactionSql::find(tx_id, false, &(*conn)).is_ok() ||
            CompletedTransactionSql::find(tx_id, false, &(*conn)).is_ok())
    }

    fn get_pending_transaction_counterparty_pub_key_by_tx_id(
        &self,
        tx_id: u64,
    ) -> Result<CommsPublicKey, TransactionStorageError>
    {
        let conn = acquire_lock!(self.database_connection);
        if let Ok(outbound_tx_sql) = OutboundTransactionSql::find(tx_id, false, &(*conn)) {
            let outbound_tx = OutboundTransaction::try_from(outbound_tx_sql)?;
            return Ok(outbound_tx.destination_public_key);
        }
        if let Ok(inbound_tx_sql) = InboundTransactionSql::find(tx_id, false, &(*conn)) {
            let inbound_tx = InboundTransaction::try_from(inbound_tx_sql)?;
            return Ok(inbound_tx.source_public_key);
        }

        Err(TransactionStorageError::ValuesNotFound)
    }

    fn complete_outbound_transaction(
        &self,
        tx_id: u64,
        completed_transaction: CompletedTransaction,
    ) -> Result<(), TransactionStorageError>
    {
        let conn = acquire_lock!(self.database_connection);

        if CompletedTransactionSql::find(tx_id, false, &(*conn)).is_ok() {
            return Err(TransactionStorageError::TransactionAlreadyExists);
        }

        match OutboundTransactionSql::find(tx_id, false, &(*conn)) {
            Ok(v) => {
                let completed_tx_sql = CompletedTransactionSql::try_from(completed_transaction)?;
                v.delete(&(*conn))?;
                completed_tx_sql.commit(&(*conn))?;
            },
            Err(TransactionStorageError::DieselError(DieselError::NotFound)) => {
                return Err(TransactionStorageError::ValueNotFound(
                    DbKey::PendingOutboundTransaction(tx_id),
                ))
            },
            Err(e) => return Err(e),
        };
        Ok(())
    }

    fn complete_inbound_transaction(
        &self,
        tx_id: u64,
        completed_transaction: CompletedTransaction,
    ) -> Result<(), TransactionStorageError>
    {
        let conn = acquire_lock!(self.database_connection);

        if CompletedTransactionSql::find(tx_id, false, &(*conn)).is_ok() {
            return Err(TransactionStorageError::TransactionAlreadyExists);
        }

        match InboundTransactionSql::find(tx_id, false, &(*conn)) {
            Ok(v) => {
                let completed_tx_sql = CompletedTransactionSql::try_from(completed_transaction)?;
                v.delete(&(*conn))?;
                completed_tx_sql.commit(&(*conn))?;
            },
            Err(TransactionStorageError::DieselError(DieselError::NotFound)) => {
                return Err(TransactionStorageError::ValueNotFound(
                    DbKey::PendingInboundTransaction(tx_id),
                ))
            },
            Err(e) => return Err(e),
        };
        Ok(())
    }

    fn broadcast_completed_transaction(&self, tx_id: u64) -> Result<(), TransactionStorageError> {
        let conn = acquire_lock!(self.database_connection);

        match CompletedTransactionSql::find(tx_id, false, &(*conn)) {
            Ok(v) => {
                if TransactionStatus::try_from(v.status)? == TransactionStatus::Completed {
                    let _ = v.update(
                        UpdateCompletedTransaction {
                            status: Some(TransactionStatus::Broadcast),
                            timestamp: None,
                            cancelled: None,
                        },
                        &(*conn),
                    )?;
                }
            },
            Err(TransactionStorageError::DieselError(DieselError::NotFound)) => {
                return Err(TransactionStorageError::ValueNotFound(DbKey::CompletedTransaction(
                    tx_id,
                )))
            },
            Err(e) => return Err(e),
        };
        Ok(())
    }

    fn mine_completed_transaction(&self, tx_id: u64) -> Result<(), TransactionStorageError> {
        let conn = acquire_lock!(self.database_connection);

        match CompletedTransactionSql::find(tx_id, false, &(*conn)) {
            Ok(v) => {
                let _ = v.update(
                    UpdateCompletedTransaction {
                        status: Some(TransactionStatus::Mined),
                        timestamp: None,
                        cancelled: None,
                    },
                    &(*conn),
                )?;
            },
            Err(TransactionStorageError::DieselError(DieselError::NotFound)) => {
                return Err(TransactionStorageError::ValueNotFound(DbKey::CompletedTransaction(
                    tx_id,
                )))
            },
            Err(e) => return Err(e),
        };
        Ok(())
    }

    fn cancel_completed_transaction(&self, tx_id: u64) -> Result<(), TransactionStorageError> {
        let conn = acquire_lock!(self.database_connection);
        match CompletedTransactionSql::find(tx_id, false, &(*conn)) {
            Ok(v) => {
                v.cancel(&(*conn))?;
            },
            Err(TransactionStorageError::DieselError(DieselError::NotFound)) => {
                return Err(TransactionStorageError::ValueNotFound(DbKey::CompletedTransaction(
                    tx_id,
                )));
            },
            Err(e) => return Err(e),
        };
        Ok(())
    }

    fn cancel_pending_transaction(&self, tx_id: u64) -> Result<(), TransactionStorageError> {
        let conn = acquire_lock!(self.database_connection);
        match InboundTransactionSql::find(tx_id, false, &(*conn)) {
            Ok(v) => {
                v.cancel(&(*conn))?;
            },
            Err(_) => {
                match OutboundTransactionSql::find(tx_id, false, &(*conn)) {
                    Ok(v) => {
                        v.cancel(&(*conn))?;
                    },
                    Err(TransactionStorageError::DieselError(DieselError::NotFound)) => {
                        return Err(TransactionStorageError::ValuesNotFound);
                    },
                    Err(e) => return Err(e),
                };
            },
        };
        Ok(())
    }

    fn mark_direct_send_success(&self, tx_id: u64) -> Result<(), TransactionStorageError> {
        let conn = acquire_lock!(self.database_connection);
        match InboundTransactionSql::find(tx_id, false, &(*conn)) {
            Ok(v) => {
                v.update(
                    UpdateInboundTransactionSql {
                        cancelled: None,
                        direct_send_success: Some(1i32),
                    },
                    &(*conn),
                )?;
            },
            Err(_) => {
                match OutboundTransactionSql::find(tx_id, false, &(*conn)) {
                    Ok(v) => {
                        v.update(
                            UpdateOutboundTransactionSql {
                                cancelled: None,
                                direct_send_success: Some(1i32),
                            },
                            &(*conn),
                        )?;
                    },
                    Err(TransactionStorageError::DieselError(DieselError::NotFound)) => {
                        return Err(TransactionStorageError::ValuesNotFound);
                    },
                    Err(e) => return Err(e),
                };
            },
        };
        Ok(())
    }

    #[cfg(feature = "test_harness")]
    fn update_completed_transaction_timestamp(
        &self,
        tx_id: u64,
        timestamp: NaiveDateTime,
    ) -> Result<(), TransactionStorageError>
    {
        let conn = acquire_lock!(self.database_connection);

        if let Ok(tx) = CompletedTransactionSql::find(tx_id, false, &(*conn)) {
            let _ = tx.update(
                UpdateCompletedTransaction {
                    status: None,
                    timestamp: Some(timestamp),
                    cancelled: None,
                },
                &(*conn),
            );
        }

        Ok(())
    }
}

#[derive(Clone, Debug, Queryable, Insertable, PartialEq)]
#[table_name = "inbound_transactions"]
struct InboundTransactionSql {
    tx_id: i64,
    source_public_key: Vec<u8>,
    amount: i64,
    receiver_protocol: String,
    message: String,
    timestamp: NaiveDateTime,
    cancelled: i32,
    direct_send_success: i32,
}

impl InboundTransactionSql {
    pub fn commit(&self, conn: &SqliteConnection) -> Result<(), TransactionStorageError> {
        diesel::insert_into(inbound_transactions::table)
            .values(self.clone())
            .execute(conn)?;
        Ok(())
    }

    pub fn index(
        conn: &SqliteConnection,
        cancelled: bool,
    ) -> Result<Vec<InboundTransactionSql>, TransactionStorageError>
    {
        Ok(inbound_transactions::table
            .filter(inbound_transactions::cancelled.eq(cancelled as i32))
            .load::<InboundTransactionSql>(conn)?)
    }

    pub fn find(
        tx_id: TxId,
        cancelled: bool,
        conn: &SqliteConnection,
    ) -> Result<InboundTransactionSql, TransactionStorageError>
    {
        Ok(inbound_transactions::table
            .filter(inbound_transactions::tx_id.eq(tx_id as i64))
            .filter(inbound_transactions::cancelled.eq(cancelled as i32))
            .first::<InboundTransactionSql>(conn)?)
    }

    pub fn delete(&self, conn: &SqliteConnection) -> Result<(), TransactionStorageError> {
        let num_deleted =
            diesel::delete(inbound_transactions::table.filter(inbound_transactions::tx_id.eq(&self.tx_id)))
                .execute(conn)?;

        if num_deleted == 0 {
            return Err(TransactionStorageError::ValuesNotFound);
        }

        Ok(())
    }

    pub fn update(
        &self,
        update: UpdateInboundTransactionSql,
        conn: &SqliteConnection,
    ) -> Result<(), TransactionStorageError>
    {
        let num_updated =
            diesel::update(inbound_transactions::table.filter(inbound_transactions::tx_id.eq(&self.tx_id)))
                .set(update)
                .execute(conn)?;

        if num_updated == 0 {
            return Err(TransactionStorageError::UnexpectedResult(
                "Database update error".to_string(),
            ));
        }

        Ok(())
    }

    pub fn cancel(&self, conn: &SqliteConnection) -> Result<(), TransactionStorageError> {
        self.update(
            UpdateInboundTransactionSql {
                cancelled: Some(1i32),
                direct_send_success: None,
            },
            conn,
        )
    }
}

impl TryFrom<InboundTransaction> for InboundTransactionSql {
    type Error = TransactionStorageError;

    fn try_from(i: InboundTransaction) -> Result<Self, Self::Error> {
        Ok(Self {
            tx_id: i.tx_id as i64,
            source_public_key: i.source_public_key.to_vec(),
            amount: u64::from(i.amount) as i64,
            receiver_protocol: serde_json::to_string(&i.receiver_protocol)?,
            message: i.message,
            timestamp: i.timestamp,
            cancelled: i.cancelled as i32,
            direct_send_success: i.direct_send_success as i32,
        })
    }
}

impl TryFrom<InboundTransactionSql> for InboundTransaction {
    type Error = TransactionStorageError;

    fn try_from(i: InboundTransactionSql) -> Result<Self, Self::Error> {
        Ok(Self {
            tx_id: i.tx_id as u64,
            source_public_key: PublicKey::from_vec(&i.source_public_key)
                .map_err(|_| TransactionStorageError::ConversionError)?,
            amount: MicroTari::from(i.amount as u64),
            receiver_protocol: serde_json::from_str(&i.receiver_protocol)?,
            status: TransactionStatus::Pending,
            message: i.message,
            timestamp: i.timestamp,
            cancelled: i.cancelled != 0,
            direct_send_success: i.direct_send_success != 0,
        })
    }
}

#[derive(AsChangeset)]
#[table_name = "inbound_transactions"]
pub struct UpdateInboundTransactionSql {
    cancelled: Option<i32>,
    direct_send_success: Option<i32>,
}

/// A structure to represent a Sql compatible version of the OutboundTransaction struct
#[derive(Clone, Debug, Queryable, Insertable, PartialEq)]
#[table_name = "outbound_transactions"]
struct OutboundTransactionSql {
    tx_id: i64,
    destination_public_key: Vec<u8>,
    amount: i64,
    fee: i64,
    sender_protocol: String,
    message: String,
    timestamp: NaiveDateTime,
    cancelled: i32,
    direct_send_success: i32,
}

impl OutboundTransactionSql {
    pub fn commit(&self, conn: &SqliteConnection) -> Result<(), TransactionStorageError> {
        diesel::insert_into(outbound_transactions::table)
            .values(self.clone())
            .execute(conn)?;
        Ok(())
    }

    pub fn index(
        conn: &SqliteConnection,
        cancelled: bool,
    ) -> Result<Vec<OutboundTransactionSql>, TransactionStorageError>
    {
        Ok(outbound_transactions::table
            .filter(outbound_transactions::cancelled.eq(cancelled as i32))
            .load::<OutboundTransactionSql>(conn)?)
    }

    pub fn find(
        tx_id: TxId,
        cancelled: bool,
        conn: &SqliteConnection,
    ) -> Result<OutboundTransactionSql, TransactionStorageError>
    {
        Ok(outbound_transactions::table
            .filter(outbound_transactions::tx_id.eq(tx_id as i64))
            .filter(outbound_transactions::cancelled.eq(cancelled as i32))
            .first::<OutboundTransactionSql>(conn)?)
    }

    pub fn delete(&self, conn: &SqliteConnection) -> Result<(), TransactionStorageError> {
        let num_deleted =
            diesel::delete(outbound_transactions::table.filter(outbound_transactions::tx_id.eq(&self.tx_id)))
                .execute(conn)?;

        if num_deleted == 0 {
            return Err(TransactionStorageError::ValuesNotFound);
        }

        Ok(())
    }

    pub fn update(
        &self,
        update: UpdateOutboundTransactionSql,
        conn: &SqliteConnection,
    ) -> Result<(), TransactionStorageError>
    {
        let num_updated =
            diesel::update(outbound_transactions::table.filter(outbound_transactions::tx_id.eq(&self.tx_id)))
                .set(update)
                .execute(conn)?;

        if num_updated == 0 {
            return Err(TransactionStorageError::UnexpectedResult(
                "Database update error".to_string(),
            ));
        }

        Ok(())
    }

    pub fn cancel(&self, conn: &SqliteConnection) -> Result<(), TransactionStorageError> {
        self.update(
            UpdateOutboundTransactionSql {
                cancelled: Some(1i32),
                direct_send_success: None,
            },
            conn,
        )
    }
}

impl TryFrom<OutboundTransaction> for OutboundTransactionSql {
    type Error = TransactionStorageError;

    fn try_from(o: OutboundTransaction) -> Result<Self, Self::Error> {
        Ok(Self {
            tx_id: o.tx_id as i64,
            destination_public_key: o.destination_public_key.to_vec(),
            amount: u64::from(o.amount) as i64,
            fee: u64::from(o.fee) as i64,
            sender_protocol: serde_json::to_string(&o.sender_protocol)?,
            message: o.message,
            timestamp: o.timestamp,
            cancelled: o.cancelled as i32,
            direct_send_success: o.direct_send_success as i32,
        })
    }
}

impl TryFrom<OutboundTransactionSql> for OutboundTransaction {
    type Error = TransactionStorageError;

    fn try_from(o: OutboundTransactionSql) -> Result<Self, Self::Error> {
        Ok(Self {
            tx_id: o.tx_id as u64,
            destination_public_key: PublicKey::from_vec(&o.destination_public_key)
                .map_err(|_| TransactionStorageError::ConversionError)?,
            amount: MicroTari::from(o.amount as u64),
            fee: MicroTari::from(o.fee as u64),
            sender_protocol: serde_json::from_str(&o.sender_protocol)?,
            status: TransactionStatus::Pending,
            message: o.message,
            timestamp: o.timestamp,
            cancelled: o.cancelled != 0,
            direct_send_success: o.direct_send_success != 0,
        })
    }
}

#[derive(AsChangeset)]
#[table_name = "outbound_transactions"]
pub struct UpdateOutboundTransactionSql {
    cancelled: Option<i32>,
    direct_send_success: Option<i32>,
}

/// A structure to represent a Sql compatible version of the CompletedTransaction struct
#[derive(Clone, Debug, Queryable, Insertable, PartialEq)]
#[table_name = "completed_transactions"]
struct CompletedTransactionSql {
    tx_id: i64,
    source_public_key: Vec<u8>,
    destination_public_key: Vec<u8>,
    amount: i64,
    fee: i64,
    transaction_protocol: String,
    status: i32,
    message: String,
    timestamp: NaiveDateTime,
    cancelled: i32,
}

impl CompletedTransactionSql {
    pub fn commit(&self, conn: &SqliteConnection) -> Result<(), TransactionStorageError> {
        diesel::insert_into(completed_transactions::table)
            .values(self.clone())
            .execute(conn)?;
        Ok(())
    }

    pub fn index(
        conn: &SqliteConnection,
        cancelled: bool,
    ) -> Result<Vec<CompletedTransactionSql>, TransactionStorageError>
    {
        Ok(completed_transactions::table
            .filter(completed_transactions::cancelled.eq(cancelled as i32))
            .load::<CompletedTransactionSql>(conn)?)
    }

    pub fn find(
        tx_id: TxId,
        cancelled: bool,
        conn: &SqliteConnection,
    ) -> Result<CompletedTransactionSql, TransactionStorageError>
    {
        Ok(completed_transactions::table
            .filter(completed_transactions::tx_id.eq(tx_id as i64))
            .filter(completed_transactions::cancelled.eq(cancelled as i32))
            .first::<CompletedTransactionSql>(conn)?)
    }

    pub fn delete(&self, conn: &SqliteConnection) -> Result<(), TransactionStorageError> {
        let num_deleted =
            diesel::delete(completed_transactions::table.filter(completed_transactions::tx_id.eq(&self.tx_id)))
                .execute(conn)?;

        if num_deleted == 0 {
            return Err(TransactionStorageError::ValuesNotFound);
        }

        Ok(())
    }

    pub fn update(
        &self,
        updated_tx: UpdateCompletedTransaction,
        conn: &SqliteConnection,
    ) -> Result<CompletedTransactionSql, TransactionStorageError>
    {
        let num_updated =
            diesel::update(completed_transactions::table.filter(completed_transactions::tx_id.eq(&self.tx_id)))
                .set(UpdateCompletedTransactionSql::from(updated_tx))
                .execute(conn)?;

        if num_updated == 0 {
            return Err(TransactionStorageError::UnexpectedResult(
                "Database update error".to_string(),
            ));
        }

        Ok(CompletedTransactionSql::find(self.tx_id as u64, false, conn)?)
    }

    pub fn cancel(&self, conn: &SqliteConnection) -> Result<(), TransactionStorageError> {
        let num_updated =
            diesel::update(completed_transactions::table.filter(completed_transactions::tx_id.eq(&self.tx_id)))
                .set(UpdateCompletedTransactionSql {
                    status: None,
                    timestamp: None,
                    cancelled: Some(1i32),
                })
                .execute(conn)?;

        if num_updated == 0 {
            return Err(TransactionStorageError::UnexpectedResult(
                "Database update error".to_string(),
            ));
        }

        Ok(())
    }
}

impl TryFrom<CompletedTransaction> for CompletedTransactionSql {
    type Error = TransactionStorageError;

    fn try_from(c: CompletedTransaction) -> Result<Self, Self::Error> {
        Ok(Self {
            tx_id: c.tx_id as i64,
            source_public_key: c.source_public_key.to_vec(),
            destination_public_key: c.destination_public_key.to_vec(),
            amount: u64::from(c.amount) as i64,
            fee: u64::from(c.fee) as i64,
            transaction_protocol: serde_json::to_string(&c.transaction)?,
            status: c.status as i32,
            message: c.message,
            timestamp: c.timestamp,
            cancelled: c.cancelled as i32,
        })
    }
}

impl TryFrom<CompletedTransactionSql> for CompletedTransaction {
    type Error = TransactionStorageError;

    fn try_from(c: CompletedTransactionSql) -> Result<Self, Self::Error> {
        Ok(Self {
            tx_id: c.tx_id as u64,
            source_public_key: PublicKey::from_vec(&c.source_public_key)
                .map_err(|_| TransactionStorageError::ConversionError)?,
            destination_public_key: PublicKey::from_vec(&c.destination_public_key)
                .map_err(|_| TransactionStorageError::ConversionError)?,
            amount: MicroTari::from(c.amount as u64),
            fee: MicroTari::from(c.fee as u64),
            transaction: serde_json::from_str(&c.transaction_protocol)?,
            status: TransactionStatus::try_from(c.status)?,
            message: c.message,
            timestamp: c.timestamp,
            cancelled: c.cancelled != 0,
        })
    }
}

/// These are the fields that can be updated for a Completed Transaction
pub struct UpdateCompletedTransaction {
    status: Option<TransactionStatus>,
    timestamp: Option<NaiveDateTime>,
    cancelled: Option<bool>,
}

#[derive(AsChangeset)]
#[table_name = "completed_transactions"]
pub struct UpdateCompletedTransactionSql {
    status: Option<i32>,
    timestamp: Option<NaiveDateTime>,
    cancelled: Option<i32>,
}

/// Map a Rust friendly UpdateCompletedTransaction to the Sql data type form
impl From<UpdateCompletedTransaction> for UpdateCompletedTransactionSql {
    fn from(u: UpdateCompletedTransaction) -> Self {
        Self {
            status: u.status.map(|s| s as i32),
            timestamp: u.timestamp,
            cancelled: u.cancelled.map(|c| c as i32),
        }
    }
}

#[cfg(test)]
mod test {
    #[cfg(feature = "test_harness")]
    use crate::transaction_service::storage::sqlite_db::UpdateCompletedTransaction;
    use crate::transaction_service::storage::{
        database::{CompletedTransaction, InboundTransaction, OutboundTransaction, TransactionStatus},
        sqlite_db::{CompletedTransactionSql, InboundTransactionSql, OutboundTransactionSql},
    };
    use chrono::Utc;
    use diesel::{Connection, SqliteConnection};
    use rand::rngs::OsRng;
    use std::convert::TryFrom;
    use tari_core::transactions::{
        tari_amount::MicroTari,
        transaction::{OutputFeatures, Transaction, UnblindedOutput},
        transaction_protocol::sender::TransactionSenderMessage,
        types::{CryptoFactories, HashDigest, PrivateKey, PublicKey},
        ReceiverTransactionProtocol,
        SenderTransactionProtocol,
    };
    use tari_crypto::keys::{PublicKey as PublicKeyTrait, SecretKey as SecretKeyTrait};
    use tari_test_utils::random::string;
    use tempdir::TempDir;

    #[test]
    fn test_crud() {
        let factories = CryptoFactories::default();
        let db_name = format!("{}.sqlite3", string(8).as_str());
        let temp_dir = TempDir::new(string(8).as_str()).unwrap();
        let db_folder = temp_dir.path().to_str().unwrap().to_string();
        let db_path = format!("{}{}", db_folder, db_name);

        embed_migrations!("./migrations");
        let conn = SqliteConnection::establish(&db_path).unwrap_or_else(|_| panic!("Error connecting to {}", db_path));

        embedded_migrations::run_with_output(&conn, &mut std::io::stdout()).expect("Migration failed");

        conn.execute("PRAGMA foreign_keys = ON").unwrap();

        let mut builder = SenderTransactionProtocol::builder(1);
        let amount = MicroTari::from(10_000);
        let input = UnblindedOutput::new(MicroTari::from(100_000), PrivateKey::random(&mut OsRng), None);
        builder
            .with_lock_height(0)
            .with_fee_per_gram(MicroTari::from(177))
            .with_offset(PrivateKey::random(&mut OsRng))
            .with_private_nonce(PrivateKey::random(&mut OsRng))
            .with_amount(0, amount)
            .with_message("Yo!".to_string())
            .with_input(
                input.as_transaction_input(&factories.commitment, OutputFeatures::default()),
                input.clone(),
            )
            .with_change_secret(PrivateKey::random(&mut OsRng));

        let stp = builder.build::<HashDigest>(&factories).unwrap();

        let outbound_tx1 = OutboundTransaction {
            tx_id: 1u64,
            destination_public_key: PublicKey::from_secret_key(&PrivateKey::random(&mut OsRng)),
            amount,
            fee: stp.clone().get_fee_amount().unwrap(),
            sender_protocol: stp.clone(),
            status: TransactionStatus::Pending,
            message: "Yo!".to_string(),
            timestamp: Utc::now().naive_utc(),
            cancelled: false,
            direct_send_success: false,
        };

        let outbound_tx2 = OutboundTransactionSql::try_from(OutboundTransaction {
            tx_id: 2u64,
            destination_public_key: PublicKey::from_secret_key(&PrivateKey::random(&mut OsRng)),
            amount,
            fee: stp.clone().get_fee_amount().unwrap(),
            sender_protocol: stp.clone(),
            status: TransactionStatus::Pending,
            message: "Hey!".to_string(),
            timestamp: Utc::now().naive_utc(),
            cancelled: false,
            direct_send_success: false,
        })
        .unwrap();

        OutboundTransactionSql::from(OutboundTransactionSql::try_from(outbound_tx1.clone()).unwrap())
            .commit(&conn)
            .unwrap();
        OutboundTransactionSql::from(outbound_tx2.clone())
            .commit(&conn)
            .unwrap();

        let outbound_txs = OutboundTransactionSql::index(&conn, false).unwrap();
        assert_eq!(outbound_txs.len(), 2);

        let returned_outbound_tx =
            OutboundTransaction::try_from(OutboundTransactionSql::find(1u64, false, &conn).unwrap()).unwrap();
        assert_eq!(
            OutboundTransactionSql::try_from(returned_outbound_tx).unwrap(),
            OutboundTransactionSql::try_from(outbound_tx1.clone()).unwrap()
        );

        let rtp = ReceiverTransactionProtocol::new(
            TransactionSenderMessage::Single(Box::new(stp.clone().build_single_round_message().unwrap())),
            PrivateKey::random(&mut OsRng),
            PrivateKey::random(&mut OsRng),
            OutputFeatures::default(),
            &factories,
        );

        let inbound_tx1 = InboundTransaction {
            tx_id: 2,
            source_public_key: PublicKey::from_secret_key(&PrivateKey::random(&mut OsRng)),
            amount,
            receiver_protocol: rtp.clone(),
            status: TransactionStatus::Pending,
            message: "Yo!".to_string(),
            timestamp: Utc::now().naive_utc(),
            cancelled: false,
            direct_send_success: false,
        };
        let inbound_tx2 = InboundTransaction {
            tx_id: 3,
            source_public_key: PublicKey::from_secret_key(&PrivateKey::random(&mut OsRng)),
            amount,
            receiver_protocol: rtp.clone(),
            status: TransactionStatus::Pending,
            message: "Hey!".to_string(),
            timestamp: Utc::now().naive_utc(),
            cancelled: false,
            direct_send_success: false,
        };

        InboundTransactionSql::try_from(inbound_tx1.clone())
            .unwrap()
            .commit(&conn)
            .unwrap();
        InboundTransactionSql::try_from(inbound_tx2)
            .unwrap()
            .commit(&conn)
            .unwrap();

        let inbound_txs = InboundTransactionSql::index(&conn, false).unwrap();
        assert_eq!(inbound_txs.len(), 2);

        let returned_inbound_tx =
            InboundTransaction::try_from(InboundTransactionSql::find(2u64, false, &conn).unwrap()).unwrap();
        assert_eq!(
            InboundTransactionSql::try_from(returned_inbound_tx).unwrap(),
            InboundTransactionSql::try_from(inbound_tx1.clone()).unwrap()
        );

        let tx = Transaction::new(vec![], vec![], vec![], PrivateKey::random(&mut OsRng));

        let completed_tx1 = CompletedTransaction {
            tx_id: 2,
            source_public_key: PublicKey::from_secret_key(&PrivateKey::random(&mut OsRng)),
            destination_public_key: PublicKey::from_secret_key(&PrivateKey::random(&mut OsRng)),
            amount,
            fee: MicroTari::from(100),
            transaction: tx.clone(),
            status: TransactionStatus::Mined,
            message: "Yo!".to_string(),
            timestamp: Utc::now().naive_utc(),
            cancelled: false,
        };
        let completed_tx2 = CompletedTransaction {
            tx_id: 3,
            source_public_key: PublicKey::from_secret_key(&PrivateKey::random(&mut OsRng)),
            destination_public_key: PublicKey::from_secret_key(&PrivateKey::random(&mut OsRng)),
            amount,
            fee: MicroTari::from(100),
            transaction: tx.clone(),
            status: TransactionStatus::Broadcast,
            message: "Hey!".to_string(),
            timestamp: Utc::now().naive_utc(),
            cancelled: false,
        };

        CompletedTransactionSql::try_from(completed_tx1.clone())
            .unwrap()
            .commit(&conn)
            .unwrap();
        assert!(CompletedTransactionSql::try_from(completed_tx1.clone())
            .unwrap()
            .commit(&conn)
            .is_err());

        CompletedTransactionSql::try_from(completed_tx2.clone())
            .unwrap()
            .commit(&conn)
            .unwrap();

        let completed_txs = CompletedTransactionSql::index(&conn, false).unwrap();
        assert_eq!(completed_txs.len(), 2);

        let returned_completed_tx =
            CompletedTransaction::try_from(CompletedTransactionSql::find(2u64, false, &conn).unwrap()).unwrap();
        assert_eq!(
            CompletedTransactionSql::try_from(returned_completed_tx).unwrap(),
            CompletedTransactionSql::try_from(completed_tx1.clone()).unwrap()
        );

        assert!(InboundTransactionSql::find(inbound_tx1.tx_id, false, &conn).is_ok());
        InboundTransactionSql::try_from(inbound_tx1.clone())
            .unwrap()
            .delete(&conn)
            .unwrap();
        assert!(InboundTransactionSql::try_from(inbound_tx1.clone())
            .unwrap()
            .delete(&conn)
            .is_err());
        assert!(InboundTransactionSql::find(inbound_tx1.tx_id, false, &conn).is_err());

        assert!(OutboundTransactionSql::find(inbound_tx1.tx_id, false, &conn).is_ok());
        OutboundTransactionSql::try_from(outbound_tx1.clone())
            .unwrap()
            .delete(&conn)
            .unwrap();
        assert!(OutboundTransactionSql::try_from(outbound_tx1.clone())
            .unwrap()
            .delete(&conn)
            .is_err());
        assert!(OutboundTransactionSql::find(outbound_tx1.tx_id, false, &conn).is_err());

        assert!(CompletedTransactionSql::find(completed_tx1.tx_id, false, &conn).is_ok());
        CompletedTransactionSql::try_from(completed_tx1.clone())
            .unwrap()
            .delete(&conn)
            .unwrap();
        assert!(CompletedTransactionSql::try_from(completed_tx1.clone())
            .unwrap()
            .delete(&conn)
            .is_err());
        assert!(CompletedTransactionSql::find(completed_tx1.tx_id, false, &conn).is_err());

        InboundTransactionSql::try_from(inbound_tx1.clone())
            .unwrap()
            .commit(&conn)
            .unwrap();

        assert!(InboundTransactionSql::find(inbound_tx1.tx_id, true, &conn).is_err());
        InboundTransactionSql::try_from(inbound_tx1.clone())
            .unwrap()
            .cancel(&conn)
            .unwrap();
        assert!(InboundTransactionSql::find(inbound_tx1.tx_id, false, &conn).is_err());
        assert!(InboundTransactionSql::find(inbound_tx1.tx_id, true, &conn).is_ok());

        OutboundTransactionSql::from(OutboundTransactionSql::try_from(outbound_tx1.clone()).unwrap())
            .commit(&conn)
            .unwrap();

        assert!(OutboundTransactionSql::find(outbound_tx1.tx_id, true, &conn).is_err());
        OutboundTransactionSql::try_from(outbound_tx1.clone())
            .unwrap()
            .cancel(&conn)
            .unwrap();
        assert!(InboundTransactionSql::find(inbound_tx1.tx_id, false, &conn).is_err());
        assert!(InboundTransactionSql::find(inbound_tx1.tx_id, true, &conn).is_ok());

        CompletedTransactionSql::try_from(completed_tx1.clone())
            .unwrap()
            .commit(&conn)
            .unwrap();

        assert!(CompletedTransactionSql::find(completed_tx1.tx_id, true, &conn).is_err());
        CompletedTransactionSql::try_from(completed_tx1.clone())
            .unwrap()
            .cancel(&conn)
            .unwrap();
        assert!(CompletedTransactionSql::find(completed_tx1.tx_id, false, &conn).is_err());
        assert!(CompletedTransactionSql::find(completed_tx1.tx_id, true, &conn).is_ok());

        #[cfg(feature = "test_harness")]
        let updated_tx = CompletedTransactionSql::find(completed_tx2.tx_id, false, &conn)
            .unwrap()
            .update(
                UpdateCompletedTransaction {
                    status: Some(TransactionStatus::Mined),
                    timestamp: None,
                    cancelled: None,
                },
                &conn,
            )
            .unwrap();
        #[cfg(feature = "test_harness")]
        assert_eq!(updated_tx.status, 2);
    }
}
