feat: chat scaffold (#5244)

Description
---
- Refactor the contacts crate for better organization.
- Expand contacts with types, and capabilities for managing messaging
over the existing comms network
- Add cucumber tests to provide support for new functionality

Motivation and Context
---
We want to expand Tari's communication network to handle p2p chat
services. This PR sets us off in the right direction for the core Tari
processes to be able to handle message passing and storage. Currently
chat clients will be able to send, and receive messages when online or
offline. Utilizing store and forward for sending communications offline.

How Has This Been Tested?
---
Via cucumber only. The chat client within cucumber is currently the only
interface for trials.

Caveats and known issues
---
- [ ] Encryption at rest: We need to change the message storage to
support encryption at rest

---------

Co-authored-by: SW van Heerden <[email protected]>

applications/tari_console_wallet/log4rs_sample.yml

@@ -80,6 +80,23 @@ appenders:
     encoder:
       pattern: "{d(%Y-%m-%d %H:%M:%S.%f)} [{t}] [Thread:{I}] {l:5} {m}{n}"
 
+  # An appender named "contacts" that writes to a file with a custom pattern encoder
+  contacts:
+    kind: rolling_file
+    path: "{{log_dir}}/log/wallet/contacts.log"
+    policy:
+      kind: compound
+      trigger:
+        kind: size
+        limit: 10mb
+      roller:
+        kind: fixed_window
+        base: 1
+        count: 5
+        pattern: "{{log_dir}}/log/wallet/contacts.{}.log"
+    encoder:
+      pattern: "{d(%Y-%m-%d %H:%M:%S.%f)} [{t}] [Thread:{I}] {l:5} {m}{n}"
+
 # root (to base_layer)
 root:
   level: debug
@@ -116,6 +133,11 @@ loggers:
     appenders:
       - network
     additive: false
+  contacts:
+    level: info
+    appenders:
+      - contacts
+    additive: false
   comms::noise:
     level: error
     appenders:

applications/tari_console_wallet/src/ui/state/app_state.rs

@@ -42,7 +42,7 @@ use tari_comms::{
     net_address::{MultiaddressesWithStats, PeerAddressSource},
     peer_manager::{NodeId, Peer, PeerFeatures, PeerFlags},
 };
-use tari_contacts::contacts_service::{handle::ContactsLivenessEvent, storage::database::Contact};
+use tari_contacts::contacts_service::{handle::ContactsLivenessEvent, types::Contact};
 use tari_core::transactions::{
     tari_amount::{uT, MicroTari},
     transaction_components::OutputFeatures,

applications/tari_console_wallet/src/ui/state/wallet_event_monitor.rs

@@ -266,7 +266,7 @@ impl WalletEventMonitor {
                                     );
                                     self.trigger_contacts_refresh().await;
                                 }
-                                ContactsLivenessEvent::NetworkSilence => {}
+                                ContactsLivenessEvent::NetworkSilence => {},
                             }
                         }
                         Err(broadcast::error::RecvError::Lagged(n)) => {

applications/tari_console_wallet/src/ui/ui_contact.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: BSD-3-Clause
 
 use chrono::{DateTime, Local};
-use tari_contacts::contacts_service::storage::database::Contact;
+use tari_contacts::contacts_service::types::Contact;
 
 #[derive(Debug, Clone)]
 pub struct UiContact {

base_layer/contacts/Cargo.toml

@@ -8,34 +8,37 @@ edition = "2018"
 
 [dependencies]
 tari_common = { path = "../../common" }
+tari_common_sqlite = { path = "../../common_sqlite" }
 tari_common_types = {  path = "../../base_layer/common_types" }
 tari_comms = {  path = "../../comms/core" }
+tari_comms_dht = { path = "../../comms/dht" }
 tari_crypto = { version = "0.16.11"}
 tari_p2p = {  path = "../p2p", features = ["auto-update"] }
 tari_service_framework = {  path = "../service_framework" }
 tari_shutdown = {  path = "../../infrastructure/shutdown" }
-tari_common_sqlite = { path = "../../common_sqlite" }
 tari_utilities = "0.4.10"
 
-tokio = { version = "1.23", features = ["sync", "macros"] }
 chrono = { version = "0.4.19", default-features = false, features = ["serde"] }
 diesel = { version = "2.0.3", features = ["sqlite", "serde_json", "chrono", "64-column-tables"] }
 diesel_migrations = "2.0.0"
 futures = { version = "^0.3.1", features = ["compat", "std"] }
-libsqlite3-sys = { version = "0.25.1", features = ["bundled"], optional = true }
 log = "0.4.6"
+num-derive = "0.3.3"
+num-traits = "0.2.15"
+prost = "0.9"
 rand = "0.7.3"
 thiserror = "1.0.26"
+tokio = { version = "1.23", features = ["sync", "macros"] }
 tower = "0.4"
+uuid = { version = "1.3.1", features = ["v4"] }
 
 [dev-dependencies]
-tari_test_utils = {  path = "../../infrastructure/test_utils" }
 tari_comms_dht = {  path = "../../comms/dht", features = ["test-mocks"] }
+tari_test_utils = {  path = "../../infrastructure/test_utils" }
 tempfile = "3.1.0"
 
-[features]
-default=["bundled_sqlite"]
-bundled_sqlite = ["libsqlite3-sys"]
+[build-dependencies]
+tari_common = { path = "../../common" }
 
 [package.metadata.cargo-machete]
-ignored = ["libsqlite3-sys"] # this is so we can run cargo machete without getting false positive about macro dependancies
+ignored = ["prost"] # this is so we can run cargo machete without getting false positive about macro dependancies

base_layer/contacts/build.rs

@@ -0,0 +1,31 @@
+//  Copyright 2023. 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.
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    tari_common::build::ProtobufCompiler::new()
+        .proto_paths(&["proto"])
+        .include_paths(&["proto"])
+        .emit_rerun_if_changed_directives()
+        .compile()
+        .unwrap();
+    Ok(())
+}

base_layer/contacts/migrations/2023-03-14-101758_create_messages/down.sql

@@ -0,0 +1,3 @@
+DROP INDEX idx_messages_address;
+
+DROP TABLE IF EXISTS messages;

base_layer/contacts/migrations/2023-03-14-101758_create_messages/up.sql

@@ -0,0 +1,9 @@
+CREATE TABLE messages (
+    address    BLOB             NOT NULL,
+    message_id BLOB PRIMARY KEY NOT NULL,
+    body       BLOB             NOT NULL,
+    stored_at  TIMESTAMP        NOT NULL,
+    direction  INTEGER          NOT NULL
+);
+
+CREATE INDEX idx_messages_address ON messages (address);

base_layer/contacts/proto/message.proto

@@ -0,0 +1,18 @@
+// Copyright 2023 The Tari Project
+// SPDX-License-Identifier: BSD-3-Clause
+
+syntax = "proto3";
+package tari.contacts.chat;
+
+message Message {
+  bytes body = 1;
+  bytes address = 2;
+  DirectionEnum direction = 3;
+  uint64 stored_at = 4;
+  bytes message_id = 5;
+}
+
+enum DirectionEnum {
+  Inbound = 0;
+  Outbound = 1;
+}

base_layer/contacts/src/contacts_service/error.rs

@@ -23,6 +23,7 @@
 use diesel::result::Error as DieselError;
 use tari_common_sqlite::error::SqliteStorageError;
 use tari_comms::connectivity::ConnectivityError;
+use tari_comms_dht::outbound::DhtOutboundError;
 use tari_p2p::services::liveness::error::LivenessError;
 use tari_service_framework::reply_channel::TransportChannelError;
 use thiserror::Error;
@@ -44,6 +45,8 @@ pub enum ContactsServiceError {
     LivenessError(#[from] LivenessError),
     #[error("ConnectivityError error: `{0}`")]
     ConnectivityError(#[from] ConnectivityError),
+    #[error("Outbound comms error: `{0}`")]
+    OutboundCommsError(#[from] DhtOutboundError),
 }
 
 #[derive(Debug, Error)]

base_layer/contacts/src/contacts_service/handle.rs

@@ -35,7 +35,7 @@ use tower::Service;
 use crate::contacts_service::{
     error::ContactsServiceError,
     service::{ContactMessageType, ContactOnlineStatus},
-    storage::database::Contact,
+    types::{Contact, Message},
 };
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -130,6 +130,8 @@ pub enum ContactsServiceRequest {
     RemoveContact(TariAddress),
     GetContacts,
     GetContactOnlineStatus(Contact),
+    SendMessage(TariAddress, Message),
+    GetAllMessages(TariAddress),
 }
 
 #[derive(Debug)]
@@ -139,6 +141,8 @@ pub enum ContactsServiceResponse {
     Contact(Contact),
     Contacts(Vec<Contact>),
     OnlineStatus(ContactOnlineStatus),
+    Messages(Vec<Message>),
+    MessageSent,
 }
 
 #[derive(Clone)]
@@ -224,4 +228,26 @@ impl ContactsServiceHandle {
             _ => Err(ContactsServiceError::UnexpectedApiResponse),
         }
     }
+
+    pub async fn get_all_messages(&mut self, pk: TariAddress) -> Result<Vec<Message>, ContactsServiceError> {
+        match self
+            .request_response_service
+            .call(ContactsServiceRequest::GetAllMessages(pk))
+            .await??
+        {
+            ContactsServiceResponse::Messages(messages) => Ok(messages),
+            _ => Err(ContactsServiceError::UnexpectedApiResponse),
+        }
+    }
+
+    pub async fn send_message(&mut self, message: Message) -> Result<(), ContactsServiceError> {
+        match self
+            .request_response_service
+            .call(ContactsServiceRequest::SendMessage(message.address.clone(), message))
+            .await??
+        {
+            ContactsServiceResponse::MessageSent => Ok(()),
+            _ => Err(ContactsServiceError::UnexpectedApiResponse),
+        }
+    }
 }

base_layer/contacts/src/contacts_service/mod.rs

@@ -22,15 +22,18 @@
 
 pub mod error;
 pub mod handle;
+pub mod proto;
 pub mod service;
 pub mod storage;
+pub mod types;
 
-use std::time::Duration;
+use std::{sync::Arc, time::Duration};
 
 use futures::future;
 use log::*;
 use tari_comms::connectivity::ConnectivityRequester;
-use tari_p2p::services::liveness::LivenessHandle;
+use tari_comms_dht::Dht;
+use tari_p2p::{comms_connector::SubscriptionFactory, services::liveness::LivenessHandle};
 use tari_service_framework::{
     async_trait,
     reply_channel,
@@ -54,16 +57,23 @@ where T: ContactsBackend
     backend: Option<T>,
     contacts_auto_ping_interval: Duration,
     contacts_online_ping_window: usize,
+    subscription_factory: Arc<SubscriptionFactory>,
 }
 
 impl<T> ContactsServiceInitializer<T>
 where T: ContactsBackend
 {
-    pub fn new(backend: T, contacts_auto_ping_interval: Duration, online_ping_window: usize) -> Self {
+    pub fn new(
+        backend: T,
+        subscription_factory: Arc<SubscriptionFactory>,
+        contacts_auto_ping_interval: Duration,
+        online_ping_window: usize,
+    ) -> Self {
         Self {
             backend: Some(backend),
             contacts_auto_ping_interval,
             contacts_online_ping_window: online_ping_window,
+            subscription_factory,
         }
     }
 }
@@ -90,16 +100,20 @@ where T: ContactsBackend + 'static
 
         let contacts_auto_ping_interval = self.contacts_auto_ping_interval;
         let contacts_online_ping_window = self.contacts_online_ping_window;
+        let subscription_factory = self.subscription_factory.clone();
         context.spawn_when_ready(move |handles| async move {
             let liveness = handles.expect_handle::<LivenessHandle>();
             let connectivity = handles.expect_handle::<ConnectivityRequester>();
+            let dht = handles.expect_handle::<Dht>();
 
             let service = ContactsService::new(
                 ContactsDatabase::new(backend),
                 receiver,
                 handles.get_shutdown_signal(),
                 liveness,
                 connectivity,
+                dht,
+                subscription_factory,
                 publisher,
                 contacts_auto_ping_interval,
                 contacts_online_ping_window,