koibumi_box/

lib.rs

//! This crate is an inbox/outbox module for Koibumi, an experimental Bitmessage client.
//!
//! See [`koibumi`](https://crates.io/crates/koibumi) for more about the application.
//! See [Bitmessage](https://bitmessage.org/) for more about the protocol.

#![deny(unsafe_code)]
#![warn(missing_docs)]

use std::{
    collections::HashMap,
    convert::{TryFrom, TryInto},
    fmt, io,
    os::raw::c_int,
};

use log::debug;

use koibumi_core::{
    address::{Address, Error as AddressError, ParseError as ParseAddressError},
    crypto::{PrivateKey, PrivateKeyError},
    encoding::{self, Encoding},
    identity::{Features, Private as PrivateIdentity},
    io::{SizedReadFromExact, WriteTo},
    message::{self, InvHash},
    object,
    pow::{NonceTrialsPerByte, PayloadLengthExtraBytes},
    time::Time,
};
use koibumi_node::db;

const SQLITE_CONSTRAINT_PRIMARYKEY: c_int = 1555;

const SUBSCRIBERS: &str = "(Subscribers)";

/// This error indicates that an operation on boxes failed.
#[derive(Debug)]
pub enum Error {
    /// A SQLx error was caught during operation on boxes.
    /// The actual error caught is returned as a payload of this variant.
    SqlxError(sqlx::Error),
    /// Indicates that tags did not match.
    TagMismatch,
    /// Indicates that the construction from public keys failed.
    AddressError(AddressError),
    /// An error was caught during parsing a Bitmessage address.
    /// The actual error caught is returned as a payload of this variant.
    ParseAddressError(ParseAddressError),
    /// Indicates that the conversion from object to msg failed.
    /// The actual error caught is returned as a payload of this variant.
    TryIntoMsgError(object::TryIntoMsgError),
    /// Indicates that the conversion from object to broadcast failed.
    /// The actual error caught is returned as a payload of this variant.
    TryIntoBroadcastError(object::TryIntoBroadcastError),
    /// Indicates that the decryption failed.
    /// The actual error caught is returned as a payload of this variant.
    DecryptError(object::DecryptError),
    /// Indicates that conversion from message row to message failed.
    /// The actual error caught is returned as a payload of this variant.
    TryIntoMessageError(TryIntoMessageError),
    /// The item attempted to insert already exists.
    AlreadyExists,
    /// The item attempted to retrieve does not exists.
    NotExists,
    /// The identity is invalid.
    InvalidIdentity,
    /// The conversion to private keys failed.
    TryIntoPrivateKeysError(TryIntoPrivateKeysError),
}

impl fmt::Display for Error {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::SqlxError(err) => err.fmt(f),
            Self::TagMismatch => "tag mismatch".fmt(f),
            Self::AddressError(err) => err.fmt(f),
            Self::ParseAddressError(err) => err.fmt(f),
            Self::TryIntoMsgError(err) => err.fmt(f),
            Self::TryIntoBroadcastError(err) => err.fmt(f),
            Self::DecryptError(err) => err.fmt(f),
            Self::TryIntoMessageError(err) => err.fmt(f),
            Self::AlreadyExists => "already exists".fmt(f),
            Self::NotExists => "not exists".fmt(f),
            Self::InvalidIdentity => "invalid identity".fmt(f),
            Self::TryIntoPrivateKeysError(err) => err.fmt(f),
        }
    }
}

impl std::error::Error for Error {}

impl From<sqlx::Error> for Error {
    fn from(err: sqlx::Error) -> Self {
        Self::SqlxError(err)
    }
}

impl From<AddressError> for Error {
    fn from(err: AddressError) -> Self {
        Self::AddressError(err)
    }
}

impl From<ParseAddressError> for Error {
    fn from(err: ParseAddressError) -> Self {
        Self::ParseAddressError(err)
    }
}

impl From<object::TryIntoMsgError> for Error {
    fn from(err: object::TryIntoMsgError) -> Self {
        Self::TryIntoMsgError(err)
    }
}

impl From<object::TryIntoBroadcastError> for Error {
    fn from(err: object::TryIntoBroadcastError) -> Self {
        Self::TryIntoBroadcastError(err)
    }
}

impl From<object::DecryptError> for Error {
    fn from(err: object::DecryptError) -> Self {
        Self::DecryptError(err)
    }
}

impl From<TryIntoMessageError> for Error {
    fn from(err: TryIntoMessageError) -> Self {
        Self::TryIntoMessageError(err)
    }
}

impl From<TryIntoPrivateKeysError> for Error {
    fn from(err: TryIntoPrivateKeysError) -> Self {
        Self::TryIntoPrivateKeysError(err)
    }
}

/// An object represents a contact.
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct Contact {
    address: Address,
}

impl Contact {
    /// Creates a contact from an address.
    pub fn new(address: Address) -> Self {
        Self { address }
    }

    /// Returns the address.
    pub fn address(&self) -> &Address {
        &self.address
    }
}

/// An object represents an user.
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct User {
    id: Vec<u8>,
    subscriptions: Vec<Address>,
    private_identities: Vec<PrivateIdentity>,
    contacts: Vec<Contact>,
    aliases: HashMap<String, String>,
}

impl User {
    /// Returns the ID.
    pub fn id(&self) -> &[u8] {
        &self.id
    }

    /// Returns the subscription list.
    pub fn subscriptions(&self) -> &[Address] {
        &self.subscriptions
    }

    /// Returns the subscription list as mutable reference.
    pub fn subscriptions_mut(&mut self) -> &mut Vec<Address> {
        &mut self.subscriptions
    }

    /// Returns the private identity list.
    pub fn private_identities(&self) -> &[PrivateIdentity] {
        &self.private_identities
    }

    /// Returns the private identity list as mutable reference.
    pub fn private_identities_mut(&mut self) -> &mut Vec<PrivateIdentity> {
        &mut self.private_identities
    }

    /// Finds the private identity by address.
    pub fn private_identity_by_address(&self, address: &Address) -> Option<&PrivateIdentity> {
        self.private_identities
            .iter()
            .find(|i| &i.address() == address)
    }

    /// Returns the contact list.
    pub fn contacts(&self) -> &[Contact] {
        &self.contacts
    }

    /// Returns the contact list as mutable reference.
    pub fn contacts_mut(&mut self) -> &mut Vec<Contact> {
        &mut self.contacts
    }

    /// Returns the aliases.
    pub fn aliases(&self) -> &HashMap<String, String> {
        &self.aliases
    }

    /// Returns the aliases as mutable reference.
    pub fn aliases_mut(&mut self) -> &mut HashMap<String, String> {
        &mut self.aliases
    }

    /// Returns the alias corresponding to the address.
    pub fn alias<'a>(&'a self, address: &'a str) -> &'a str {
        match self.aliases.get(address) {
            Some(alias) => alias,
            None => address,
        }
    }

    /// Returns the rich alias corresponding to the address.
    pub fn rich_alias(&self, address: &str) -> String {
        match self.aliases.get(address) {
            Some(alias) => format!("{} ({})", alias, address),
            None => address.to_string(),
        }
    }
}

impl From<User> for koibumi_node::User {
    fn from(user: User) -> Self {
        Self::new(user.id, user.subscriptions, user.private_identities)
    }
}

/// A message structure stored on database.
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct Message {
    user_id: Vec<u8>,
    object: message::Object,
    time: Time,
    to: Option<Address>,
    from: Address,
    encoding: Encoding,
    content: Vec<u8>,
    subject: String,
    read: bool,
}

impl Message {
    /// Returns the received time.
    pub fn time(&self) -> Time {
        self.time
    }

    /// Returns the receiver's Bitmessage address.
    pub fn to_address(&self) -> Option<&Address> {
        self.to.as_ref()
    }

    /// Returns the sender's Bitmessage address.
    pub fn from_address(&self) -> &Address {
        &self.from
    }

    /// Returns the encoding.
    pub fn encoding(&self) -> Encoding {
        self.encoding
    }

    /// Returns the content bytes.
    pub fn content(&self) -> &[u8] {
        &self.content
    }
}

/// This error indicates that conversion from message row to message failed.
#[derive(Debug)]
pub enum TryIntoMessageError {
    /// A standard I/O error was caught during converting a message.
    /// The actual error caught is returned as a payload of this variant.
    IoError(io::Error),
    /// An error was caught during parsing a Bitmessage address.
    /// The actual error caught is returned as a payload of this variant.
    ParseAddressError(ParseAddressError),
    /// An error was caught during parsing a message encoding type.
    /// The actual error caught is returned as a payload of this variant.
    InvalidEncoding(encoding::InvalidEncoding),
}

impl fmt::Display for TryIntoMessageError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::IoError(err) => err.fmt(f),
            Self::ParseAddressError(err) => err.fmt(f),
            Self::InvalidEncoding(err) => err.fmt(f),
        }
    }
}

impl std::error::Error for TryIntoMessageError {}

impl From<io::Error> for TryIntoMessageError {
    fn from(err: io::Error) -> Self {
        Self::IoError(err)
    }
}

impl From<ParseAddressError> for TryIntoMessageError {
    fn from(err: ParseAddressError) -> Self {
        Self::ParseAddressError(err)
    }
}

impl From<encoding::InvalidEncoding> for TryIntoMessageError {
    fn from(err: encoding::InvalidEncoding) -> Self {
        Self::InvalidEncoding(err)
    }
}

impl TryFrom<MessageRow> for Message {
    type Error = TryIntoMessageError;

    fn try_from(value: MessageRow) -> Result<Self, <Self as TryFrom<MessageRow>>::Error> {
        let object = message::Object::sized_read_from_exact(value.object)?;

        let to = if value.to_address == SUBSCRIBERS {
            None
        } else {
            Some(value.to_address.parse::<Address>()?)
        };

        let from = value.from_address.parse::<Address>()?;

        let encoding: Encoding = (value.encoding as u64).try_into()?;

        Ok(Self {
            user_id: value.user,
            object,
            time: (value.time as u64).into(),
            to,
            from,
            encoding,
            content: value.content,
            subject: value.subject,
            read: value.read != 0,
        })
    }
}

#[derive(sqlx::FromRow, Clone)]
struct MessageRow {
    user: Vec<u8>,
    object: Vec<u8>,
    time: i64,
    to_address: String,
    from_address: String,
    encoding: i64,
    content: Vec<u8>,
    subject: String,
    read: i8,
}

impl From<Message> for MessageRow {
    fn from(value: Message) -> Self {
        let mut object = Vec::new();
        value.object.write_to(&mut object).unwrap();

        let time = if value.time.as_secs() > i64::MAX as u64 {
            i64::MAX
        } else {
            value.time.as_secs() as i64
        };

        let to_address = if let Some(address) = value.to {
            address.to_string()
        } else {
            SUBSCRIBERS.to_string()
        };

        Self {
            user: value.user_id,
            object,
            time,
            to_address,
            from_address: value.from.to_string(),
            encoding: value.encoding as i64,
            content: value.content,
            subject: value.subject,
            read: value.read as i8,
        }
    }
}

#[derive(Clone, PartialEq, Eq, Hash, Debug)]
struct PrivateKeys {
    user_id: Vec<u8>,
    enabled: bool,
    private_identity: PrivateIdentity,
}

#[derive(sqlx::FromRow, Clone)]
struct PrivateKeysRow {
    user: Vec<u8>,
    address: String,
    enabled: bool,
    features: i64,
    nonce_trials_per_byte: i64,
    payload_length_extra_bytes: i64,
    signing_key: Vec<u8>,
    encryption_key: Vec<u8>,
    chan: bool,
}

#[derive(Debug)]
enum TryIntoPrivateKeysRowError {
    OutOfRange,
}

impl fmt::Display for TryIntoPrivateKeysRowError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::OutOfRange => "out of range".fmt(f),
        }
    }
}

impl std::error::Error for TryIntoPrivateKeysRowError {}

impl TryFrom<PrivateKeys> for PrivateKeysRow {
    type Error = TryIntoPrivateKeysRowError;

    fn try_from(value: PrivateKeys) -> Result<Self, <Self as TryFrom<PrivateKeys>>::Error> {
        let nonce_trials_per_byte = value.private_identity.nonce_trials_per_byte().as_u64();
        if nonce_trials_per_byte > i64::MAX as u64 {
            return Err(TryIntoPrivateKeysRowError::OutOfRange);
        }
        let payload_length_extra_bytes =
            value.private_identity.payload_length_extra_bytes().as_u64();
        if payload_length_extra_bytes > i64::MAX as u64 {
            return Err(TryIntoPrivateKeysRowError::OutOfRange);
        }
        Ok(Self {
            user: value.user_id,
            address: value.private_identity.address().to_string(),
            enabled: value.enabled,
            features: value.private_identity.features().bits() as i64,
            nonce_trials_per_byte: nonce_trials_per_byte as i64,
            payload_length_extra_bytes: payload_length_extra_bytes as i64,
            signing_key: value
                .private_identity
                .private_signing_key()
                .as_ref()
                .to_vec(),
            encryption_key: value
                .private_identity
                .private_encryption_key()
                .as_ref()
                .to_vec(),
            chan: value.private_identity.chan(),
        })
    }
}

/// The conversion to private keys failed.
#[derive(Debug)]
pub enum TryIntoPrivateKeysError {
    /// There was an out of range parameter.
    OutOfRange,
    /// Indicates that parsing an address failed.
    ParseAddressError(ParseAddressError),
    /// Indicates array length mismatch.
    TryFromSliceError(std::array::TryFromSliceError),
    /// Indicates that the operation on a private key failed.
    PrivateKeyError(PrivateKeyError),
    /// Indicates that the construction of the address from public keys failed.
    AddressError(AddressError),
}

impl fmt::Display for TryIntoPrivateKeysError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::OutOfRange => "out of range".fmt(f),
            Self::ParseAddressError(err) => err.fmt(f),
            Self::TryFromSliceError(err) => err.fmt(f),
            Self::PrivateKeyError(err) => err.fmt(f),
            Self::AddressError(err) => err.fmt(f),
        }
    }
}

impl std::error::Error for TryIntoPrivateKeysError {}

impl From<ParseAddressError> for TryIntoPrivateKeysError {
    fn from(err: ParseAddressError) -> Self {
        Self::ParseAddressError(err)
    }
}

impl From<std::array::TryFromSliceError> for TryIntoPrivateKeysError {
    fn from(err: std::array::TryFromSliceError) -> Self {
        Self::TryFromSliceError(err)
    }
}

impl From<PrivateKeyError> for TryIntoPrivateKeysError {
    fn from(err: PrivateKeyError) -> Self {
        Self::PrivateKeyError(err)
    }
}

impl From<AddressError> for TryIntoPrivateKeysError {
    fn from(err: AddressError) -> Self {
        Self::AddressError(err)
    }
}
impl TryFrom<PrivateKeysRow> for PrivateKeys {
    type Error = TryIntoPrivateKeysError;

    fn try_from(value: PrivateKeysRow) -> Result<Self, <Self as TryFrom<PrivateKeysRow>>::Error> {
        let address = value.address.parse::<Address>()?;
        if value.features < 0 || value.features > u32::MAX as i64 {
            return Err(TryIntoPrivateKeysError::OutOfRange);
        }
        let features = Features::from_bits_retain(value.features as u32);
        if value.nonce_trials_per_byte < 0 {
            return Err(TryIntoPrivateKeysError::OutOfRange);
        }
        let nonce_trials_per_byte = NonceTrialsPerByte::from(value.nonce_trials_per_byte as u64);
        if value.payload_length_extra_bytes < 0 {
            return Err(TryIntoPrivateKeysError::OutOfRange);
        }
        let payload_length_extra_bytes =
            PayloadLengthExtraBytes::from(value.payload_length_extra_bytes as u64);
        let bytes: &[u8] = value.signing_key.as_ref();
        let private_signing_key = PrivateKey::new(bytes.try_into()?)?;
        let bytes: &[u8] = value.encryption_key.as_ref();
        let private_encryption_key = PrivateKey::new(bytes.try_into()?)?;

        let private_identity = PrivateIdentity::new(
            address.version(),
            address.stream(),
            features,
            nonce_trials_per_byte,
            payload_length_extra_bytes,
            private_signing_key,
            private_encryption_key,
            value.chan,
        )?;
        Ok(Self {
            user_id: value.user,
            enabled: value.enabled,
            private_identity,
        })
    }
}

/// An inbox/outbox manager.
#[derive(Debug)]
pub struct Manager {
    pool: db::SqlitePool,
}

impl Manager {
    /// Constructs an inbox/outbox manager from database connection pool.
    pub async fn new(pool: db::SqlitePool) -> Result<Manager, Error> {
        sqlx::query(
            "CREATE TABLE IF NOT EXISTS users (
                id BLOB NOT NULL PRIMARY KEY,
                enabled INTEGER NOT NULL,
                name TEXT NOT NULL
            )",
        )
        .execute(pool.write())
        .await?;

        sqlx::query(
            "CREATE TABLE IF NOT EXISTS subscriptions (
                user BLOB NOT NULL,
                address TEXT NOT NULL,
                enabled INTEGER NOT NULL,
                PRIMARY KEY(user, address)
            )",
        )
        .execute(pool.write())
        .await?;

        sqlx::query(
            "CREATE TABLE IF NOT EXISTS messages (
                user BLOB NOT NULL,
                hash BLOB NOT NULL,
                object BLOB NOT NULL,
                time INTEGER NOT NULL,
                to_address TEXT NOT NULL,
                from_address TEXT NOT NULL,
                encoding INTEGER NOT NULL,
                content BLOB NOT NULL,
                subject TEXT NOT NULL,
                read INTEGER NOT NULL,
                PRIMARY KEY(user, hash)
            )",
        )
        .execute(pool.write())
        .await?;

        /*
        sqlx::query(
            "CREATE TABLE IF NOT EXISTS public_keys (
                user BLOB NOT NULL,
                address TEXT NOT NULL,
                features INTEGER NOT NULL,
                signing_key BLOB NOT NULL,
                encryption_key BLOB NOT NULL,
                nonce_trials_per_byte INTEGER NOT NULL,
                payload_length_extra_bytes INTEGER NOT NULL,
                hash BLOB NOT NULL,
                object BLOB NOT NULL,
                expires INTEGER NOT NULL,
                PRIMARY KEY(user, address)
            )",
        )
        .execute(pool.write())
        .await?;
        */

        sqlx::query(
            "CREATE TABLE IF NOT EXISTS private_keys (
                user BLOB NOT NULL,
                address TEXT NOT NULL,
                enabled INTEGER NOT NULL,
                features INTEGER NOT NULL,
                nonce_trials_per_byte INTEGER NOT NULL,
                payload_length_extra_bytes INTEGER NOT NULL,
                signing_key BLOB NOT NULL,
                encryption_key BLOB NOT NULL,
                chan INTEGER NOT NULL,
                PRIMARY KEY(user, address)
            )",
        )
        .execute(pool.write())
        .await?;

        sqlx::query(
            "CREATE TABLE IF NOT EXISTS contacts (
                user BLOB NOT NULL,
                address TEXT NOT NULL,
                enabled INTEGER NOT NULL,
                PRIMARY KEY(user, address)
            )",
        )
        .execute(pool.write())
        .await?;

        sqlx::query(
            "CREATE TABLE IF NOT EXISTS aliases (
                user BLOB NOT NULL,
                address TEXT NOT NULL,
                alias TEXT NOT NULL,
                PRIMARY KEY(user, address)
            )",
        )
        .execute(pool.write())
        .await?;

        Ok(Self { pool })
    }

    /// Retrieves a user object specified by its ID.
    pub async fn user(&self, id: &[u8]) -> Result<User, Error> {
        let list = sqlx::query_as::<sqlx::Sqlite, (i8, String)>(
            "SELECT enabled, name FROM users WHERE id=?1",
        )
        .bind(id)
        .fetch_all(self.pool.read())
        .await?;
        if list.is_empty() {
            return Err(Error::NotExists);
        }
        let subscriptions = sqlx::query_as::<sqlx::Sqlite, (String, i8)>(
            "SELECT address, enabled FROM subscriptions WHERE user=?1",
        )
        .bind(id)
        .fetch_all(self.pool.read())
        .await?;
        let mut slist = Vec::new();
        for subscription in subscriptions {
            let address = subscription.0.parse::<Address>();
            if address.is_err() {
                continue;
            }
            slist.push(address.unwrap());
        }

        let private_identities = sqlx::query_as::<sqlx::Sqlite, PrivateKeysRow>(
            "SELECT user, address, enabled,
            features, nonce_trials_per_byte, payload_length_extra_bytes,
            signing_key, encryption_key, chan
            FROM private_keys
            WHERE user=?1",
        )
        .bind(id)
        .fetch_all(self.pool.read())
        .await?;
        let mut ilist = Vec::new();
        for i in private_identities {
            let private_keys: PrivateKeys = i.try_into()?;
            ilist.push(private_keys.private_identity);
        }

        let contacts = sqlx::query_as::<sqlx::Sqlite, (String,)>(
            "SELECT address
            FROM contacts
            WHERE user=?1",
        )
        .bind(id)
        .fetch_all(self.pool.read())
        .await?;
        let mut clist = Vec::new();
        for i in contacts {
            let address = i.0.parse::<Address>()?;
            let contact = Contact { address };
            clist.push(contact);
        }

        let aliases = sqlx::query_as::<sqlx::Sqlite, (String, String)>(
            "SELECT address, alias FROM aliases WHERE user=?1",
        )
        .bind(id)
        .fetch_all(self.pool.read())
        .await?;
        let mut amap = HashMap::new();
        for alias in aliases {
            let address = alias.0.parse::<Address>();
            if address.is_err() {
                continue;
            }
            amap.insert(alias.0.clone(), alias.1.clone());
        }

        Ok(User {
            id: id.to_vec(),
            subscriptions: slist,
            private_identities: ilist,
            contacts: clist,
            aliases: amap,
        })
    }

    /// Inserts a user object.
    pub async fn add_user(&self, id: &[u8], name: &str) -> Result<(), Error> {
        // XXX guard
        if let Ok(_user) = self.user(id).await {
            return Err(Error::AlreadyExists);
        }

        let result = sqlx::query(
            "INSERT INTO users (
                id, enabled, name
            ) VALUES (?1, ?2, ?3)",
        )
        .bind(id)
        .bind(true)
        .bind(name)
        .execute(self.pool.write())
        .await;
        match result {
            Ok(_) => Ok(()),
            Err(sqlx::Error::Database(err)) => {
                if let Some(code) = err.code() {
                    debug!("code: {}", code);
                    if code == SQLITE_CONSTRAINT_PRIMARYKEY.to_string() {
                        return Err(Error::AlreadyExists);
                    }
                }
                Err(Error::SqlxError(sqlx::Error::Database(err)))
            }
            Err(err) => Err(Error::SqlxError(err)),
        }
    }

    /// Insert a subscription address for a user.
    pub async fn subscribe(&self, user_id: &[u8], address: &Address) -> Result<(), Error> {
        let result = sqlx::query(
            "INSERT INTO subscriptions (
                user, address, enabled
            ) VALUES (?1, ?2, ?3)",
        )
        .bind(user_id)
        .bind(address.to_string())
        .bind(true)
        .execute(self.pool.write())
        .await;
        match result {
            Ok(_) => Ok(()),
            Err(sqlx::Error::Database(err)) => {
                if let Some(code) = err.code() {
                    if code == SQLITE_CONSTRAINT_PRIMARYKEY.to_string() {
                        return Err(Error::AlreadyExists);
                    }
                }
                Err(Error::SqlxError(sqlx::Error::Database(err)))
            }
            Err(err) => Err(Error::SqlxError(err)),
        }
    }

    /// Retrieves a message specified by a user ID and an inventory hash.
    pub async fn get_message(
        &self,
        user_id: &[u8],
        hash: &InvHash,
    ) -> Result<Option<Message>, Error> {
        let list =
            sqlx::query_as::<sqlx::Sqlite, MessageRow>("SELECT user, object, time, to_address,
                from_address, encoding, content, subject, read FROM messages WHERE user=?1 AND hash=?2")
                .bind(user_id)
                .bind(hash.as_ref())
                .fetch_all(self.pool.read())
                .await?;
        if list.is_empty() {
            return Ok(None);
        }
        Ok(Some(list[0].clone().try_into()?))
    }

    async fn insert_message(&self, message: &Message) -> Result<(), Error> {
        let hash = message.object.inv_hash();
        let row: MessageRow = message.clone().into();

        sqlx::query(
            "INSERT INTO messages (
                    user, hash, object, time, to_address, from_address, encoding, content, subject, read
                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
        )
        .bind(row.user)
        .bind(hash.as_ref())
        .bind(row.object)
        .bind(row.time)
        .bind(row.to_address)
        .bind(row.from_address)
        .bind(row.encoding)
        .bind(row.content)
        .bind(row.subject)
        .bind(row.read)
        .execute(self.pool.write())
        .await?;
        Ok(())
    }

    /// Inserts a msg object.
    pub async fn insert_msg(
        &self,
        user_id: Vec<u8>,
        identity: &PrivateIdentity,
        object: message::Object,
    ) -> Result<Message, Error> {
        let msg = object::Msg::try_from(object.clone())?;
        println!("****** MSG FOUND ******"); // DEBUG
        let content = msg.decrypt(object.header(), &identity)?;

        let subject = if content.encoding() == encoding::Encoding::Simple {
            if let Ok(simple) = encoding::Simple::try_from(content.message()) {
                String::from_utf8_lossy(simple.subject()).to_string()
            } else {
                "(No subject)".to_string()
            }
        } else {
            "(No subject)".to_string()
        };

        let message = Message {
            user_id,
            object,
            time: Time::now(),
            to: Some(identity.address()),
            from: content.address()?,
            encoding: content.encoding(),
            content: content.message().to_vec(),
            subject,
            read: false,
        };

        self.insert_message(&message).await?;

        Ok(message)
    }

    /// Inserts a broadcast object.
    pub async fn insert_broadcast(
        &self,
        user_id: Vec<u8>,
        address: Address,
        object: message::Object,
    ) -> Result<Message, Error> {
        let broadcast = object::Broadcast::try_from(object.clone())?;
        match &broadcast {
            object::Broadcast::V4(_) => (),
            object::Broadcast::V5(v5) => {
                if address.broadcast_tag() != *v5.tag() {
                    return Err(Error::TagMismatch);
                }
                println!("****** BROADCAST V5 FOUND ******"); // DEBUG
            }
        }
        let content = broadcast.decrypt(object.header(), &address)?;

        let subject = if content.encoding() == encoding::Encoding::Simple {
            if let Ok(simple) = encoding::Simple::try_from(content.message()) {
                String::from_utf8_lossy(simple.subject()).to_string()
            } else {
                "(No subject)".to_string()
            }
        } else {
            "(No subject)".to_string()
        };

        let message = Message {
            user_id,
            object,
            time: Time::now(),
            to: None,
            from: address,
            encoding: content.encoding(),
            content: content.message().to_vec(),
            subject,
            read: false,
        };

        self.insert_message(&message).await?;

        Ok(message)
    }

    /// Retrieves a list of message entries for a user specified by a user ID.
    pub async fn message_list(&self, user_id: &[u8]) -> Result<Vec<MessageEntry>, Error> {
        let list = sqlx::query_as::<sqlx::Sqlite, (Vec<u8>, i64, String, String, String, i8)>(
            "SELECT hash, time, to_address,
                from_address, subject, read FROM messages WHERE user=?1 ORDER BY time DESC",
        )
        .bind(user_id)
        .fetch_all(self.pool.read())
        .await?;
        let mut r: Vec<MessageEntry> = Vec::new();
        for elem in list {
            if elem.0.len() != 32 {
                continue;
            }
            let hash: &[u8] = elem.0.as_ref();
            let hash: [u8; 32] = hash.try_into().unwrap();
            let hash = InvHash::new(hash);

            let time = Time::new(elem.1 as u64);

            let to = if elem.2 == SUBSCRIBERS {
                None
            } else {
                let to = elem.2.parse::<Address>();
                if to.is_err() {
                    continue;
                }
                Some(to.unwrap())
            };

            let from = elem.3.parse::<Address>();
            if from.is_err() {
                continue;
            }
            let from = from.unwrap();

            let entry = MessageEntry {
                hash,
                time,
                to,
                from,
                subject: elem.4,
                read: elem.5 != 0,
            };
            r.push(entry);
        }
        Ok(r)
    }

    /// Sets the read flag of the message specified by a user ID and an inventory hash.
    pub async fn set_read(&self, user_id: &[u8], hash: &InvHash, read: bool) -> Result<(), Error> {
        sqlx::query("UPDATE messages SET read=?1 WHERE user=?2 and hash=?3")
            .bind(read)
            .bind(user_id)
            .bind(hash.as_ref())
            .execute(self.pool.write())
            .await?;
        Ok(())
    }

    /// Adds private identity in database.
    pub async fn add_private_identity(
        &self,
        user_id: &[u8],
        identity: PrivateIdentity,
    ) -> Result<(), Error> {
        if identity.nonce_trials_per_byte().as_u64() > i64::MAX as u64 {
            return Err(Error::InvalidIdentity);
        }
        if identity.payload_length_extra_bytes().as_u64() > i64::MAX as u64 {
            return Err(Error::InvalidIdentity);
        }

        sqlx::query(
            "INSERT INTO private_keys (
                    user, address, enabled,
                    features, nonce_trials_per_byte, payload_length_extra_bytes,
                    signing_key, encryption_key, chan
                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
        )
        .bind(user_id)
        .bind(identity.address().to_string())
        .bind(true)
        .bind(identity.features().bits() as i64)
        .bind(identity.nonce_trials_per_byte().as_u64() as i64)
        .bind(identity.payload_length_extra_bytes().as_u64() as i64)
        .bind(identity.private_signing_key().as_ref())
        .bind(identity.private_encryption_key().as_ref())
        .bind(identity.chan())
        .execute(self.pool.write())
        .await?;

        Ok(())
    }

    /// Insert a contact address for a user.
    pub async fn add_contact(&self, user_id: &[u8], contact: &Contact) -> Result<(), Error> {
        let result = sqlx::query(
            "INSERT INTO contacts (
                user, address, enabled
            ) VALUES (?1, ?2, ?3)",
        )
        .bind(user_id)
        .bind(contact.address.to_string())
        .bind(true)
        .execute(self.pool.write())
        .await;
        match result {
            Ok(_) => Ok(()),
            Err(sqlx::Error::Database(err)) => {
                if let Some(code) = err.code() {
                    if code == SQLITE_CONSTRAINT_PRIMARYKEY.to_string() {
                        return Err(Error::AlreadyExists);
                    }
                }
                Err(Error::SqlxError(sqlx::Error::Database(err)))
            }
            Err(err) => Err(Error::SqlxError(err)),
        }
    }

    /// Insert an alias.
    pub async fn add_alias(
        &self,
        user_id: &[u8],
        address: &Address,
        alias: &str,
    ) -> Result<(), Error> {
        let result = sqlx::query(
            "INSERT INTO aliases (
                user, address, alias
            ) VALUES (?1, ?2, ?3)",
        )
        .bind(user_id)
        .bind(address.to_string())
        .bind(alias)
        .execute(self.pool.write())
        .await;
        match result {
            Ok(_) => Ok(()),
            Err(sqlx::Error::Database(err)) => {
                if let Some(code) = err.code() {
                    if code == SQLITE_CONSTRAINT_PRIMARYKEY.to_string() {
                        return Err(Error::AlreadyExists);
                    }
                }
                Err(Error::SqlxError(sqlx::Error::Database(err)))
            }
            Err(err) => Err(Error::SqlxError(err)),
        }
    }
}

/// A message object without its content.
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct MessageEntry {
    hash: InvHash,
    time: Time,
    to: Option<Address>,
    from: Address,
    subject: String,
    read: bool,
}

impl MessageEntry {
    /// Returns the inventory hash.
    pub fn hash(&self) -> &InvHash {
        &self.hash
    }

    /// Returns the subject.
    pub fn subject(&self) -> &str {
        &self.subject
    }

    /// Returns the read flag.
    pub fn read(&self) -> bool {
        self.read
    }

    /// Sets the read flag.
    pub fn set_read(&mut self, read: bool) {
        self.read = read;
    }
}

impl From<&Message> for MessageEntry {
    fn from(message: &Message) -> MessageEntry {
        Self {
            hash: message.object.inv_hash(),
            time: message.time,
            to: message.to.clone(),
            from: message.from.clone(),
            subject: message.subject.clone(),
            read: message.read,
        }
    }
}