Market decentralized - database models

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

@@ -1,23 +1,21 @@
 -- Your SQL goes here
 
 CREATE TABLE market_offer (
-	id VARCHAR(255) NOT NULL PRIMARY KEY,
+	id VARCHAR(97) NOT NULL PRIMARY KEY,
 	properties TEXT NOT NULL,
 	constraints TEXT NOT NULL,
-	node_id VARCHAR(255) NOT NULL,
+	node_id VARCHAR(20) NOT NULL,
 
-	creation_time DATETIME NOT NULL,
-	addition_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+	creation_ts DATETIME NOT NULL,
 	expiration_time DATETIME NOT NULL
 );
 
 CREATE TABLE market_demand (
-	id VARCHAR(255) NOT NULL PRIMARY KEY,
+	id VARCHAR(97) NOT NULL PRIMARY KEY,
 	properties TEXT NOT NULL,
 	constraints TEXT NOT NULL,
-	node_id VARCHAR(255) NOT NULL,
+	node_id VARCHAR(20) NOT NULL,
 
-	creation_time DATETIME NOT NULL,
-	addition_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+	creation_ts DATETIME NOT NULL,
 	expiration_time DATETIME NOT NULL
 );

core/market/decentralized/src/api.rs

@@ -4,9 +4,8 @@ pub mod response;
 
 use serde::{Deserialize, Serialize};
 
-pub const DEFAULT_ACK_TIMEOUT: u32 = 60; // seconds
 pub const DEFAULT_EVENT_TIMEOUT: f32 = 0.0; // seconds
-pub const DEFAULT_REQUEST_TIMEOUT: f32 = 12.0;
+pub const DEFAULT_QUERY_TIMEOUT: f32 = 12.0;
 
 #[derive(Deserialize)]
 pub struct PathAgreement {
@@ -50,21 +49,10 @@ pub struct QueryTimeoutMaxEvents {
 
 #[inline(always)]
 pub(crate) fn default_query_timeout() -> Option<f32> {
-    Some(DEFAULT_REQUEST_TIMEOUT)
-}
-
-#[inline(always)]
-pub(crate) fn default_ack_timeout() -> u32 {
-    DEFAULT_ACK_TIMEOUT
+    Some(DEFAULT_QUERY_TIMEOUT)
 }
 
 #[inline(always)]
 pub(crate) fn default_event_timeout() -> Option<f32> {
     Some(DEFAULT_EVENT_TIMEOUT)
 }
-
-#[derive(Deserialize)]
-pub struct Timeout {
-    #[serde(default = "default_ack_timeout")]
-    pub timeout: u32,
-}

core/market/decentralized/src/api/provider.rs

@@ -26,6 +26,8 @@ pub fn register_endpoints(scope: Scope) -> Scope {
         .service(reject_proposal)
         .service(approve_agreement)
         .service(reject_agreement)
+        .service(terminate_agreement)
+        .service(get_agreement)
 }
 
 #[actix_web::post("/offers")]
@@ -52,13 +54,9 @@ async fn unsubscribe(
     path: Path<PathSubscription>,
     id: Identity,
 ) -> HttpResponse {
-    match market
-        .matcher
-        .get_offer(path.into_inner().subscription_id)
-        .await
-    {
-        Ok(Some(offer)) => response::ok(offer),
-        Ok(None) => response::not_found(),
+    let subscription_id = path.into_inner().subscription_id;
+    match market.unsubscribe_offer(subscription_id.clone(), id).await {
+        Ok(()) => response::ok(subscription_id),
         // TODO: Translate MatcherError to better HTTP response.
         Err(error) => response::server_error(&format!("{}", error)),
     }
@@ -120,3 +118,21 @@ async fn reject_agreement(
 ) -> HttpResponse {
     response::not_implemented()
 }
+
+#[actix_web::delete("/agreements/{agreement_id}/terminate")]
+async fn terminate_agreement(
+    market: Data<Arc<MarketService>>,
+    path: Path<PathAgreement>,
+    id: Identity,
+) -> HttpResponse {
+    response::not_implemented()
+}
+
+#[actix_web::get("/agreements/{agreement_id}")]
+async fn get_agreement(
+    market: Data<Arc<MarketService>>,
+    path: Path<PathAgreement>,
+    id: Identity,
+) -> HttpResponse {
+    response::not_implemented()
+}

core/market/decentralized/src/api/requestor.rs

@@ -56,13 +56,9 @@ async fn unsubscribe(
     path: Path<PathSubscription>,
     id: Identity,
 ) -> HttpResponse {
-    match market
-        .matcher
-        .get_demand(path.into_inner().subscription_id)
-        .await
-    {
-        Ok(Some(demand)) => response::ok(demand),
-        Ok(None) => response::not_found(),
+    let subscription_id = path.into_inner().subscription_id;
+    match market.unsubscribe_demand(subscription_id.clone(), id).await {
+        Ok(()) => response::ok(subscription_id),
         // TODO: Translate MatcherError to better HTTP response.
         Err(error) => response::server_error(&format!("{}", error)),
     }

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

@@ -20,9 +20,7 @@ pub struct Demand {
     pub node_id: String,
 
     /// Creation time of Demand on Requestor side.
-    pub creation_time: NaiveDateTime,
-    /// Time when this Demand was added to our local database.
-    pub addition_time: NaiveDateTime,
+    pub creation_ts: NaiveDateTime,
     /// Time when Demand expires set by Requestor.
     pub expiration_time: NaiveDateTime,
 }
@@ -46,8 +44,7 @@ impl Demand {
             properties,
             constraints,
             node_id,
-            creation_time: creation_time.clone(),
-            addition_time: creation_time.clone(),
+            creation_ts: creation_time.clone(),
             expiration_time: creation_time.clone(),
         })
     }
@@ -70,8 +67,7 @@ impl Demand {
             properties,
             constraints,
             node_id,
-            creation_time: creation_time.clone(),
-            addition_time: creation_time.clone(),
+            creation_ts: creation_time.clone(),
             expiration_time: creation_time.clone(),
         }
     }

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

@@ -20,9 +20,7 @@ pub struct Offer {
     pub node_id: String,
 
     /// Creation time of Offer on Provider side.
-    pub creation_time: NaiveDateTime,
-    /// Time when this Offer was added to our local database.
-    pub addition_time: NaiveDateTime,
+    pub creation_ts: NaiveDateTime,
     /// Time when Offer expires set by Provider.
     pub expiration_time: NaiveDateTime,
 }
@@ -49,8 +47,7 @@ impl Offer {
             properties,
             constraints,
             node_id,
-            creation_time: creation_time.clone(),
-            addition_time: creation_time.clone(),
+            creation_ts: creation_time.clone(),
             expiration_time: creation_time.clone(),
         })
     }
@@ -73,8 +70,7 @@ impl Offer {
             properties,
             constraints,
             node_id,
-            creation_time: creation_time.clone(),
-            addition_time: creation_time.clone(),
+            creation_ts: creation_time.clone(),
             expiration_time: creation_time.clone(),
         }
     }
@@ -94,7 +90,6 @@ impl Offer {
     }
 }
 
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -132,4 +127,3 @@ mod tests {
         assert!(Offer::from(&offer).is_ok());
     }
 }
-

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

@@ -15,11 +15,11 @@ use ya_client::model::ErrorMessage;
 
 #[derive(Error, Debug)]
 pub enum SubscriptionParseError {
-    #[error("Subscription id [{}] has invalid format.", .0)]
+    #[error("Subscription id [{0}] has invalid format.")]
     InvalidFormat(String),
-    #[error("Subscription id [{}] contains non hexadecimal characters.", .0)]
+    #[error("Subscription id [{0}] contains non hexadecimal characters.")]
     NotHexadecimal(String),
-    #[error("Subscription id [{}] has invalid length.", .0)]
+    #[error("Subscription id [{0}] has invalid length.")]
     InvalidLength(String),
 }
 

core/market/decentralized/src/db/schema.rs

@@ -4,8 +4,7 @@ table! {
         properties -> Text,
         constraints -> Text,
         node_id -> Text,
-        creation_time -> Timestamp,
-        addition_time -> Timestamp,
+        creation_ts -> Timestamp,
         expiration_time -> Timestamp,
     }
 }
@@ -16,8 +15,7 @@ table! {
         properties -> Text,
         constraints -> Text,
         node_id -> Text,
-        creation_time -> Timestamp,
-        addition_time -> Timestamp,
+        creation_ts -> Timestamp,
         expiration_time -> Timestamp,
     }
 }

core/market/decentralized/src/market.rs

@@ -101,10 +101,11 @@ impl MarketService {
         let offer = ModelOffer::from_new(&offer, &id);
         let subscription_id = offer.id.to_string();
 
+        self.matcher.subscribe_offer(&offer).await?;
         self.provider_negotiation_engine
             .subscribe_offer(&offer)
             .await?;
-        self.matcher.subscribe_offer(&offer).await?;
+
         Ok(subscription_id)
     }
 
@@ -129,11 +130,10 @@ impl MarketService {
         let demand = ModelDemand::from_new(&demand, &id);
         let subscription_id = demand.id.to_string();
 
+        self.matcher.subscribe_demand(&demand).await?;
         self.requestor_negotiation_engine
             .subscribe_demand(&demand)
             .await?;
-
-        self.matcher.subscribe_demand(&demand).await?;
         Ok(subscription_id)
     }