Dynamic GraphQL document API

aquadoggo/Cargo.toml

@@ -18,6 +18,7 @@ edition = "2018"
 anyhow = "^1.0.58"
 async-graphql = "^3.0.38"
 async-graphql-axum = "^3.0.38"
+async-recursion = "^1.0.0"
 async-trait = "^0.1.56"
 axum = "^0.5.10"
 bamboo-rs-core-ed25519-yasmf = "^0.1.1"
@@ -30,8 +31,9 @@ futures = "^0.3.21"
 gql_client = "^1.0.6"
 lipmaa-link = "^0.2.2"
 log = "^0.4.17"
+once_cell = "^1.12.0"
 openssl-probe = "^0.1.5"
-p2panda-rs = { git = "https://github.com/p2panda/p2panda", rev = "e06fd08c45253d60fcd42778f59e946a9ed73f71" }
+p2panda-rs = { git = "https://github.com/p2panda/p2panda", rev = "c79622e7f8edc495d74a37825d5f72234a7b0c9e" }
 serde = { version = "^1.0.137", features = ["derive"] }
 sqlx = { version = "^0.6.0", features = [
     "any",
@@ -61,7 +63,7 @@ hex = "0.4.3"
 http = "^0.2.8"
 hyper = "^0.14.19"
 once_cell = "^1.12.0"
-p2panda-rs = { git = "https://github.com/p2panda/p2panda", rev = "e06fd08c45253d60fcd42778f59e946a9ed73f71", features = [
+p2panda-rs = { git = "https://github.com/p2panda/p2panda", rev = "c79622e7f8edc495d74a37825d5f72234a7b0c9e", features = [
     "testing",
 ] }
 rand = "^0.8.5"

aquadoggo/src/db/models/entry.rs

@@ -8,7 +8,7 @@ use sqlx::FromRow;
 ///
 /// We store the u64 integer values of `log_id` and `seq_num` as strings since SQLite doesn't
 /// support storing unsigned 64 bit integers.
-#[derive(FromRow, Debug, Serialize, Clone, PartialEq)]
+#[derive(FromRow, Debug, Serialize, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
 pub struct EntryRow {
     /// Public key of the author.

aquadoggo/src/db/models/task.rs

@@ -6,7 +6,7 @@ use sqlx::FromRow;
 /// Representation of a row from the `tasks` table as stored in the database.
 ///
 /// This table holds all "pending" tasks of the materialization service worker.
-#[derive(FromRow, Debug, Serialize, Clone, PartialEq)]
+#[derive(FromRow, Debug, Serialize, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
 pub struct TaskRow {
     /// Name of the task worker.

aquadoggo/src/db/provider.rs

@@ -13,7 +13,7 @@ use crate::db::request::{EntryArgsRequest, PublishEntryRequest};
 use crate::db::stores::{StorageEntry, StorageLog};
 use crate::db::Pool;
 use crate::errors::StorageProviderResult;
-use crate::graphql::client::NextEntryArguments;
+use crate::graphql::client::NextEntryArgumentsType;
 
 /// Sql based storage that implements `StorageProvider`.
 #[derive(Clone, Debug)]
@@ -32,9 +32,9 @@ impl SqlStorage {
 /// databases.
 #[async_trait]
 impl StorageProvider<StorageEntry, StorageLog, VerifiedOperation> for SqlStorage {
-    type EntryArgsResponse = NextEntryArguments;
+    type EntryArgsResponse = NextEntryArgumentsType;
     type EntryArgsRequest = EntryArgsRequest;
-    type PublishEntryResponse = NextEntryArguments;
+    type PublishEntryResponse = NextEntryArgumentsType;
     type PublishEntryRequest = PublishEntryRequest;
 
     /// Returns the related document for any entry.

aquadoggo/src/db/stores/entry.rs

@@ -22,7 +22,7 @@ use crate::db::provider::SqlStorage;
 ///
 /// This struct implements the `AsStorageEntry` trait which is required when constructing the
 /// `EntryStore`.
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct StorageEntry {
     entry_signed: EntrySigned,
     operation_encoded: OperationEncoded,

aquadoggo/src/db/stores/schema.rs

@@ -56,6 +56,8 @@ impl SchemaStore for SqlStorage {
     /// Get all Schema which have been published to this node.
     ///
     /// Returns an error if a fatal db error occured.
+    ///
+    /// Silently ignores incomplete or broken schema definitions.
     async fn get_all_schema(&self) -> Result<Vec<Schema>, SchemaStoreError> {
         let schema_views: Vec<SchemaView> = self
             .get_documents_by_schema(&SchemaId::new("schema_definition_v1")?)

aquadoggo/src/db/stores/test_utils.rs

@@ -4,15 +4,16 @@ use std::convert::TryFrom;
 use std::sync::Arc;
 
 use futures::Future;
+use log::{debug, error, info};
 use p2panda_rs::document::{DocumentBuilder, DocumentId, DocumentViewId};
 use p2panda_rs::entry::{sign_and_encode, Entry, EntrySigned};
 use p2panda_rs::hash::Hash;
 use p2panda_rs::identity::{Author, KeyPair};
 use p2panda_rs::operation::{
-    AsOperation, AsVerifiedOperation, Operation, OperationEncoded, OperationId, OperationValue,
-    PinnedRelation, PinnedRelationList, Relation, RelationList, VerifiedOperation,
+    AsOperation, AsVerifiedOperation, Operation, OperationEncoded, OperationFields, OperationId,
+    OperationValue, PinnedRelation, PinnedRelationList, Relation, RelationList, VerifiedOperation,
 };
-use p2panda_rs::schema::SchemaId;
+use p2panda_rs::schema::{FieldType, Schema, SchemaId};
 use p2panda_rs::storage_provider::traits::{OperationStore, StorageProvider};
 use p2panda_rs::test_utils::constants::{PRIVATE_KEY, SCHEMA_ID};
 use p2panda_rs::test_utils::fixtures::{operation, operation_fields};
@@ -22,12 +23,16 @@ use sqlx::Any;
 use tokio::runtime::Builder;
 use tokio::sync::Mutex;
 
+use crate::context::Context;
 use crate::db::provider::SqlStorage;
 use crate::db::request::{EntryArgsRequest, PublishEntryRequest};
 use crate::db::traits::DocumentStore;
 use crate::db::{connection_pool, create_database, run_pending_migrations, Pool};
-use crate::graphql::client::NextEntryArguments;
+use crate::graphql::client::NextEntryArgumentsType;
+use crate::materializer::tasks::{dependency_task, reduce_task, schema_task};
+use crate::materializer::TaskInput;
 use crate::test_helpers::TEST_CONFIG;
+use crate::{Configuration, SchemaProvider};
 
 /// The fields used as defaults in the tests.
 pub fn doggo_test_fields() -> Vec<(&'static str, OperationValue)> {
@@ -264,8 +269,15 @@ impl TestDatabaseRunner {
         runtime.block_on(async {
             // Initialise test database
             let pool = initialize_db().await;
+            let store = SqlStorage::new(pool);
+            let context = Context::new(
+                store.clone(),
+                Configuration::default(),
+                SchemaProvider::default(),
+            );
             let mut db = TestDatabase {
-                store: SqlStorage::new(pool),
+                context,
+                store,
                 test_data: TestData::default(),
             };
 
@@ -291,7 +303,7 @@ impl TestDatabaseRunner {
 
             // Panic here when test failed. The test fails within its own async task and stays
             // there, we need to propagate it further to inform the test runtime about the result
-            result.unwrap();
+            result.expect("The test failed");
         });
     }
 }
@@ -380,10 +392,97 @@ pub fn test_db(
 /// Container for `SqlStore` with access to the document ids and key_pairs used in the
 /// pre-populated database for testing.
 pub struct TestDatabase {
+    pub context: Context,
     pub store: SqlStorage,
     pub test_data: TestData,
 }
 
+impl TestDatabase {
+    /// Publish a document and materialise it in the store.
+    ///
+    /// Also runs dependency task for document.
+    pub async fn add_document(
+        &mut self,
+        schema_id: &SchemaId,
+        fields: OperationFields,
+        key_pair: &KeyPair,
+    ) -> DocumentViewId {
+        info!("Creating document for {}", schema_id);
+
+        // Get requested schema from store.
+        let schema = self
+            .context
+            .schema_provider
+            .get(schema_id)
+            .await
+            .expect("Schema not found");
+
+        // Build, publish and reduce create operation for document.
+        let create_op = Operation::new_create(schema.id().to_owned(), fields).unwrap();
+        let (entry_signed, _) = send_to_store(&self.store, &create_op, None, key_pair).await;
+        let input = TaskInput::new(Some(DocumentId::from(entry_signed.hash())), None);
+        let dependency_tasks = reduce_task(self.context.clone(), input.clone())
+            .await
+            .unwrap();
+
+        // Run dependency tasks
+        if let Some(tasks) = dependency_tasks {
+            for task in tasks {
+                dependency_task(self.context.clone(), task.input().to_owned())
+                    .await
+                    .unwrap();
+            }
+        }
+        DocumentViewId::from(entry_signed.hash())
+    }
+
+    /// Publish a schema and materialise it in the store.
+    pub async fn add_schema(
+        &mut self,
+        name: &str,
+        fields: Vec<(&str, FieldType)>,
+        key_pair: &KeyPair,
+    ) -> Schema {
+        info!("Creating schema {}", name);
+        let mut field_ids = Vec::new();
+
+        // Build and reduce schema field definitions
+        for field in fields {
+            let create_field_op = Schema::create_field(field.0, field.1.clone().into()).unwrap();
+            let (entry_signed, _) =
+                send_to_store(&self.store, &create_field_op, None, key_pair).await;
+
+            let input = TaskInput::new(Some(DocumentId::from(entry_signed.hash())), None);
+            reduce_task(self.context.clone(), input).await.unwrap();
+
+            info!("Added field '{}' ({})", field.0, field.1.serialise());
+            field_ids.push(DocumentViewId::from(entry_signed.hash()));
+        }
+
+        // Build and reduce schema definition
+        let create_schema_op =
+            Schema::create(name, "test schema description", field_ids.into()).unwrap();
+        let (entry_signed, _) = send_to_store(&self.store, &create_schema_op, None, key_pair).await;
+        let input = TaskInput::new(None, Some(DocumentViewId::from(entry_signed.hash())));
+        reduce_task(self.context.clone(), input.clone())
+            .await
+            .unwrap();
+
+        // Run schema task for this spec
+        schema_task(self.context.clone(), input).await.unwrap();
+
+        let view_id = DocumentViewId::from(entry_signed.hash());
+        let schema_id = SchemaId::Application(name.to_string(), view_id);
+
+        debug!("Done building {}", schema_id);
+        self.context
+            .schema_provider
+            .get(&schema_id)
+            .await
+            .expect("Failed adding schema to provider.")
+    }
+}
+
 /// Data collected when populating a `TestData` base in order to easily check values which
 /// would be otherwise hard or impossible to get through the store methods.
 #[derive(Default)]
@@ -453,7 +552,7 @@ pub async fn send_to_store(
     operation: &Operation,
     document_id: Option<&DocumentId>,
     key_pair: &KeyPair,
-) -> (EntrySigned, NextEntryArguments) {
+) -> (EntrySigned, NextEntryArgumentsType) {
     // Get an Author from the key_pair.
     let author = Author::try_from(key_pair.public_key().to_owned()).unwrap();
 
@@ -497,10 +596,13 @@ pub async fn send_to_store(
     // Also insert the operation into the store.
     let verified_operation =
         VerifiedOperation::new(&author, &entry_encoded.hash().into(), operation).unwrap();
-    store
+    match store
         .insert_operation(&verified_operation, &document_id)
         .await
-        .unwrap();
+    {
+        Ok(_) => {}
+        Err(err) => error!("Failed inserting operation: {}", err),
+    }
 
     (entry_encoded, publish_entry_response)
 }
@@ -559,10 +661,22 @@ impl TestDatabaseManager {
     }
 
     pub async fn create(&self, url: &str) -> TestDatabase {
-        // Initialise test database
         let pool = initialize_db_with_url(url).await;
+
+        // Initialise test store using pool.
+        let store = SqlStorage::new(pool.clone());
+
+        // Initialise context for store.
+        let context = Context::new(
+            store.clone(),
+            Configuration::default(),
+            SchemaProvider::default(),
+        );
+
+        // Initialise finished test database.
         let test_db = TestDatabase {
-            store: SqlStorage::new(pool.clone()),
+            context,
+            store,
             test_data: TestData::default(),
         };
         self.pools.lock().await.push(pool);

aquadoggo/src/db/traits/schema.rs

@@ -20,6 +20,6 @@ pub trait SchemaStore {
     /// Get all published Schema from storage.
     ///
     /// Returns a vector of Schema, or an empty vector if none were found. Returns
-    /// an error when a fatal storage error occured or a schema could not be constructed.  
+    /// an error when a fatal storage error occured or a schema could not be constructed.
     async fn get_all_schema(&self) -> Result<Vec<Schema>, SchemaStoreError>;
 }

aquadoggo/src/errors.rs

@@ -1,4 +1,25 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+use p2panda_rs::schema::{SchemaId, SchemaIdError};
+use thiserror::Error;
+
 /// A specialized result type for the storage provider.
 pub type StorageProviderResult<T> = anyhow::Result<T, Box<dyn std::error::Error + Send + Sync>>;
+
+/// Errors returned by schema service.
+#[derive(Error, Debug)]
+pub enum SchemaProviderError {
+    /// Schema service can only handle application schemas it has definitions for.
+    #[allow(dead_code)]
+    #[error("not a known application schema: {0}")]
+    UnknownApplicationSchema(SchemaId),
+
+    /// This operation has a requirement on the schema parameter.
+    #[allow(dead_code)]
+    #[error("invalid schema: {0}, {1}")]
+    InvalidSchema(SchemaId, String),
+
+    /// Schema service can only handle valid schema ids.
+    #[error(transparent)]
+    InvalidSchemaId(#[from] SchemaIdError),
+}