Add database most important tables

.cargo/config

@@ -1,2 +1,7 @@
 [build]
 rustflags = ["-C", "target-feature=+crt-static"]
+
+
+[cargo-new]
+name = "Golem Factory"
+email = "contact@golem.network"

.gitignore

@@ -7,7 +7,7 @@
 
 ## Workdir
 workdir/*
-core/market-decentralized/tests/test-workdir
+core/market-decentralized/tests/test-workdir/*
 
 # IDEs
 .idea

Cargo.toml

@@ -6,6 +6,7 @@ members = [
     "core/gftp",
     "core/identity",
     "core/market-forwarding",
+    "core/market-decentralized",
     "core/model",
     "core/net",
     "core/payment",
@@ -37,6 +38,7 @@ ya-activity = { path = "core/activity" }
 ya-payment = { path = "core/payment" }
 ya-identity = { path = "core/identity" }
 ya-market-forwarding = { path = "core/market-forwarding" }
+ya-market-decentralized = { path = "core/market-decentralized" }
 ya-net = { path = "core/net" }
 
 ## CORE UTILS

agent/provider/Cargo.toml

@@ -19,7 +19,7 @@ actix-rt = "1.0"
 actix_derive = "0.2.0"
 anyhow = "1.0"
 bigdecimal = "0.1.0"
-derive_more = "0.99.3"
+derive_more = "0.99.5"
 dialoguer = "0.5.0"
 dotenv = "0.15.0"
 env_logger = "0.7"

core/market-decentralized/Cargo.toml

@@ -0,0 +1,44 @@
+[package]
+name = "ya-market-decentralized"
+version = "0.1.0"
+authors = ["nieznanysprawiciel <nieznany.sprawiciel@gmail.com>"]
+edition = "2018"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[features]
+market-test-suite = []
+
+[dependencies]
+ya-client = "0.3"
+ya-client-model = "0.1"
+ya-core-model = { version = "0.1", features = ["market"] }
+ya-persistence = "0.2"
+ya-service-api = "0.1"
+ya-service-bus = "0.2"
+ya-service-api-interfaces = "0.1"
+
+actix-web = "2.0"
+anyhow = "1.0"
+async-trait = "0.1"
+chrono = { version = "0.4", features = ["serde"] }
+derive_more = "0.99.5"
+diesel = { version = "1.4", features = ["chrono", "sqlite", "r2d2"] }
+diesel_migrations = "1.4"
+futures = "0.3"
+lazy_static = "1.4"
+libsqlite3-sys = { version = "0.9.1", features = ["bundled"] }
+log = "0.4"
+r2d2 = "0.8"
+serde = { version = "^1.0", features = ["derive"] }
+thiserror = "1.0"
+tokio = { version = "0.2", features = ["time"] }
+
+[dev-dependencies]
+actix-rt = "1.0.0"
+anyhow = "1.0"
+env_logger = "0.7"
+serde_json = "1.0"
+tokio = { version = "0.2", features = ["macros"] }
+
+ya-core-model = { version = "0.1", features = ["net"] }

core/market-decentralized/diesel.toml

@@ -0,0 +1,5 @@
+# For documentation on how to configure this file,
+# see diesel.rs/guides/configuring-diesel-cli
+
+[print_schema]
+file = "src/db/schema.rs"

core/market-decentralized/examples/discovery_usage.rs

@@ -0,0 +1,97 @@
+use anyhow;
+use async_trait::async_trait;
+use chrono::Utc;
+use serde_json::json;
+use std::sync::Arc;
+use tokio::runtime::Runtime;
+
+use ya_client_model::market::Offer;
+use ya_market_decentralized::protocol::callbacks::HandlerSlot;
+use ya_market_decentralized::protocol::{
+    Discovery, DiscoveryBuilder, DiscoveryError, DiscoveryFactory, DiscoveryInitError,
+    OfferReceived, RetrieveOffers,
+};
+
+// =========================================== //
+// TODO: Remove this example after implementing Discovery
+// =========================================== //
+
+/// Example implementation of Discovery.
+struct DiscoveryExample {
+    offer_received: HandlerSlot<OfferReceived>,
+    retrieve_offers: HandlerSlot<RetrieveOffers>,
+}
+
+impl DiscoveryFactory for DiscoveryExample {
+    fn new(mut builder: DiscoveryBuilder) -> Result<Arc<dyn Discovery>, DiscoveryInitError> {
+        let offer_received = builder.offer_received_handler()?;
+        let retrieve_offers = builder.retrieve_offers_handler()?;
+
+        let discovery = DiscoveryExample {
+            offer_received,
+            retrieve_offers,
+        };
+        Ok(Arc::new(discovery))
+    }
+}
+
+#[async_trait]
+impl Discovery for DiscoveryExample {
+    async fn bind_gsb(&self, _prefix: String) -> Result<(), DiscoveryInitError> {
+        Ok(())
+    }
+
+    async fn broadcast_offer(&self, offer: Offer) -> Result<(), DiscoveryError> {
+        Ok(self
+            .offer_received
+            .call(format!("caller"), OfferReceived { offer })
+            .await?)
+    }
+
+    async fn retrieve_offers(&self) -> Result<Vec<Offer>, DiscoveryError> {
+        Ok(self
+            .retrieve_offers
+            .call(
+                format!("caller"),
+                RetrieveOffers {
+                    newer_than: Utc::now(),
+                },
+            )
+            .await?)
+    }
+}
+
+#[actix_rt::main]
+async fn main() -> anyhow::Result<()> {
+    env_logger::init();
+
+    let builder = DiscoveryBuilder::new()
+        .bind_offer_received(move |msg: OfferReceived| async move {
+            log::info!("Offer from [{}] received.", msg.offer.offer_id.unwrap());
+            Ok(())
+        })
+        .bind_retrieve_offers(move |_msg: RetrieveOffers| async move {
+            log::info!("Offers request received.");
+            Ok(vec![])
+        });
+    let discovery = DiscoveryExample::new(builder)?;
+    let dicovery_clone = discovery.clone();
+
+    std::thread::spawn(move || {
+        let offer = mock_offer(format!("Caller-thread"));
+
+        let mut rt = Runtime::new().unwrap();
+        rt.block_on(dicovery_clone.broadcast_offer(offer))
+    });
+
+    let offer = mock_offer(format!("Caller-local"));
+    discovery.broadcast_offer(offer).await?;
+
+    Ok(())
+}
+
+fn mock_offer(caller: String) -> Offer {
+    let mut offer = Offer::new(json!({}), format!(""));
+    offer.offer_id = Some(caller);
+    offer
+}

core/market-decentralized/migrations/2020-05-15-093513_init-db/down.sql

@@ -0,0 +1,9 @@
+-- This file should undo anything in `up.sql`
+
+DROP TABLE "market_agreement";
+DROP TABLE "market_agreement_state";
+DROP TABLE "market_agreement_event";
+DROP TABLE "market_agreement_event_type";
+
+DROP TABLE "market_offer";
+DROP TABLE "market_demand";

core/market-decentralized/migrations/2020-05-15-093513_init-db/up.sql

@@ -0,0 +1,63 @@
+-- Your SQL goes here
+
+CREATE TABLE market_agreement (
+	id VARCHAR(255) NOT NULL PRIMARY KEY,
+	state_id INTEGER NOT NULL,
+	demand_node_id VARCHAR(255) NOT NULL,
+	demand_properties TEXT NOT NULL,
+	demand_constraints TEXT NOT NULL,
+	offer_node_id VARCHAR(255) NOT NULL,
+	offer_properties TEXT NOT NULL,
+	offer_constraints TEXT NOT NULL,
+	valid_to DATETIME NOT NULL,
+	approved_date DATETIME,
+	proposed_signature TEXT NOT NULL,
+	approved_signature TEXT NOT NULL,
+	committed_signature TEXT,
+    FOREIGN KEY(state_id) REFERENCES market_agreement_state (id)
+);
+
+
+CREATE TABLE market_agreement_state(
+	id INTEGER NOT NULL PRIMARY KEY,
+	name VARCHAR(50) NOT NULL
+);
+
+INSERT INTO market_agreement_state(id, name)
+values
+       (0, 'Proposal'),
+       (1, 'Pending'),
+       (10, 'Approved'),
+       (40, 'Canceled'),
+       (41, 'Rejected'),
+       (42, 'Expired'),
+       (50, 'Terminated');
+
+
+CREATE TABLE market_agreement_event(
+	id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
+	agreement_id VARCHAR(255) NOT NULL,
+	event_date DATETIME NOT NULL,
+	event_type_id INTEGER NOT NULL,
+    FOREIGN KEY(agreement_id) REFERENCES market_agreement (id),
+    FOREIGN KEY(event_type_id) REFERENCES market_agreement_event_type (id)
+);
+
+CREATE TABLE market_agreement_event_type(
+	id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
+	name VARCHAR(50) NOT NULL
+);
+
+CREATE TABLE market_offer (
+	id VARCHAR(255) NOT NULL PRIMARY KEY,
+	properties TEXT NOT NULL,
+	constraints TEXT NOT NULL,
+	node_id VARCHAR(255) NOT NULL
+);
+
+CREATE TABLE market_demand (
+	id VARCHAR(255) NOT NULL PRIMARY KEY,
+	properties TEXT NOT NULL,
+	constraints TEXT NOT NULL,
+	node_id VARCHAR(255) NOT NULL
+);

core/market-decentralized/src/db.rs

@@ -0,0 +1,4 @@
+pub mod dao;
+pub mod models;
+pub mod schema;
+

core/market-decentralized/src/db/dao.rs

@@ -0,0 +1,2 @@
+mod offer;
+mod demand;

core/market-decentralized/src/db/dao/demand.rs

@@ -0,0 +1,14 @@
+use ya_persistence::executor::{do_with_transaction, readonly_transaction, AsDao, PoolType};
+
+
+#[allow(unused)]
+pub struct DemandDao<'c> {
+    pool: &'c PoolType,
+}
+
+impl<'c> AsDao<'c> for DemandDao<'c> {
+    fn as_dao(pool: &'c PoolType) -> Self {
+        Self { pool }
+    }
+}
+

core/market-decentralized/src/db/dao/offer.rs

@@ -0,0 +1,18 @@
+use ya_persistence::executor::{do_with_transaction, readonly_transaction, AsDao, PoolType};
+
+
+#[allow(unused)]
+pub struct OfferDao<'c> {
+    pool: &'c PoolType,
+}
+
+impl<'c> AsDao<'c> for OfferDao<'c> {
+    fn as_dao(pool: &'c PoolType) -> Self {
+        Self { pool }
+    }
+}
+
+impl<'c> OfferDao<'c> {
+    //pub async fn insert_offer()
+}
+

core/market-decentralized/src/db/models.rs

@@ -0,0 +1,4 @@
+mod offer;
+mod demand;
+
+

core/market-decentralized/src/db/models/offer.rs

@@ -0,0 +1,14 @@
+use diesel::prelude::*;
+
+use crate::db::schema::market_offer;
+
+
+#[derive(Debug, Identifiable, Insertable)]
+#[table_name = "market_offer"]
+pub struct Offer {
+    pub id: String,
+    pub properties: String,
+    pub constraints: String,
+    pub node_id: Option<String>,
+}
+