Merge pull request #46 from openlawlibrary/ndusan/insert-changes-fixes

fix(rdf): fixes to insert history command

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "stelae"
 description = "A collection of tools in Rust and Python for preserving, authenticating, and accessing laws in perpetuity."
-version = "0.3.0-alpha.4"
+version = "0.3.0-alpha.5"
 edition = "2021"
 readme = "README.md"
 license = "AGPL-3.0"

migrations/sqlite/20240115152953_initial_db.down.sql

@@ -2,7 +2,17 @@
 PRAGMA foreign_keys = OFF;
 
 DROP INDEX IF EXISTS changed_library_document_library_mpath_idx;
+DROP INDEX IF EXISTS library_change_status_idx;
+DROP INDEX IF EXISTS library_change_publication_version_idx;
 DROP INDEX IF EXISTS library_change_library_mpath_idx;
+
+DROP INDEX IF EXISTS publication_version_id_idx;
+
+DROP INDEX IF EXISTS publication_has_publication_versions_publication_version_id_idx;
+DROP INDEX IF EXISTS publication_has_publication_versions_publication_id_idx;
+
+DROP INDEX IF EXISTS document_change_doc_mpath_pub_id_status_idx;
+DROP INDEX IF EXISTS document_change_publication_version_idx;
 DROP INDEX IF EXISTS document_change_doc_mpath_idx;
 
 DROP TABLE IF EXISTS changed_library_document;

migrations/sqlite/20240115152953_initial_db.up.sql

@@ -11,14 +11,18 @@ CREATE TABLE document_element (
     doc_mpath TEXT,
     url TEXT,
     doc_id TEXT,
+    stele TEXT,
     CONSTRAINT fk_doc_id
         FOREIGN KEY (doc_id)
         REFERENCES document(doc_id),
-    PRIMARY KEY (doc_mpath)
+    PRIMARY KEY (doc_mpath),
+    CONSTRAINT fk_stele FOREIGN KEY (stele) REFERENCES stele(name) ON DELETE CASCADE
 );
 CREATE TABLE library (
     mpath TEXT PRIMARY KEY,
-    url TEXT
+    url TEXT,
+    stele TEXT,
+    CONSTRAINT fk_stele FOREIGN KEY (stele) REFERENCES stele(name) ON DELETE CASCADE
 );
 CREATE TABLE publication (
     id TEXT,
@@ -80,7 +84,16 @@ CREATE TABLE document_change (
         ON DELETE CASCADE,
     PRIMARY KEY (id)
 );
+
 CREATE INDEX document_change_doc_mpath_idx ON document_change(doc_mpath COLLATE NOCASE);
+CREATE INDEX document_change_publication_version_idx ON document_change(publication_version_id);
+CREATE INDEX document_change_doc_mpath_pub_id_status_idx ON document_change(doc_mpath COLLATE NOCASE, status);
+
+CREATE INDEX publication_has_publication_versions_publication_id_idx ON publication_has_publication_versions(publication_id);
+CREATE INDEX publication_has_publication_versions_publication_version_id_idx ON publication_has_publication_versions(publication_version_id);
+
+CREATE INDEX publication_version_id_idx ON publication_version(id);
+
 CREATE TABLE library_change (
     publication_version_id TEXT,
     status TEXT,
@@ -101,6 +114,9 @@ CREATE TABLE changed_library_document (
     PRIMARY KEY (document_change_id, library_mpath)
 );
 CREATE INDEX library_change_library_mpath_idx ON library_change(library_mpath COLLATE NOCASE);
+CREATE INDEX library_change_publication_version_idx ON library_change(publication_version_id);
+CREATE INDEX library_change_status_idx ON library_change(library_mpath COLLATE NOCASE, status);
+
 CREATE INDEX changed_library_document_library_mpath_idx ON changed_library_document(library_mpath COLLATE NOCASE);
 
 PRAGMA optimize;

src/db/models/changed_library_document/manager.rs

@@ -25,6 +25,7 @@ impl super::TxManager for DatabaseTransaction {
             });
             let query = query_builder.build();
             query.execute(&mut *self.tx).await?;
+            query_builder.reset();
         }
         Ok(())
     }

src/db/models/document_element/manager.rs

@@ -12,18 +12,19 @@ impl super::Manager for DatabaseConnection {
     ///
     /// # Errors
     /// Errors if can't establish a connection to the database.
-    async fn find_doc_mpath_by_url(&self, url: &str) -> anyhow::Result<String> {
+    async fn find_doc_mpath_by_url(&self, url: &str, stele: &str) -> anyhow::Result<String> {
         let statement = "
             SELECT de.doc_mpath
             FROM document_element de
-            WHERE de.url = $1
+            WHERE de.url = $1 AND de.stele = $2
             LIMIT 1
         ";
         let row = match self.kind {
             DatabaseKind::Postgres | DatabaseKind::Sqlite => {
                 let mut connection = self.pool.acquire().await?;
                 sqlx::query_as::<_, (String,)>(statement)
                     .bind(url)
+                    .bind(stele)
                     .fetch_one(&mut *connection)
                     .await?
             }
@@ -39,14 +40,16 @@ impl super::TxManager for DatabaseTransaction {
     /// # Errors
     /// Errors if the document elements cannot be inserted into the database.
     async fn insert_bulk(&mut self, document_elements: Vec<DocumentElement>) -> anyhow::Result<()> {
-        let mut query_builder =
-            QueryBuilder::new("INSERT OR IGNORE INTO document_element ( doc_mpath, url, doc_id ) ");
+        let mut query_builder = QueryBuilder::new(
+            "INSERT OR IGNORE INTO document_element ( doc_mpath, url, doc_id, stele ) ",
+        );
         for chunk in document_elements.chunks(BATCH_SIZE) {
             query_builder.push_values(chunk, |mut bindings, de| {
                 bindings
                     .push_bind(&de.doc_mpath)
                     .push_bind(&de.url)
-                    .push_bind(&de.doc_id);
+                    .push_bind(&de.doc_id)
+                    .push_bind(&de.stele);
             });
             let query = query_builder.build();
             query.execute(&mut *self.tx).await?;

src/db/models/document_element/mod.rs

@@ -7,7 +7,7 @@ pub mod manager;
 #[async_trait]
 pub trait Manager {
     /// Find one document materialized path by url.
-    async fn find_doc_mpath_by_url(&self, url: &str) -> anyhow::Result<String>;
+    async fn find_doc_mpath_by_url(&self, url: &str, stele: &str) -> anyhow::Result<String>;
 }
 
 /// Trait for managing transactional document elements.
@@ -26,16 +26,19 @@ pub struct DocumentElement {
     pub url: String,
     /// Unique document identifier.
     pub doc_id: String,
+    /// Reference to the stele.
+    pub stele: String,
 }
 
 impl DocumentElement {
     /// Create a new document element.
     #[must_use]
-    pub const fn new(doc_mpath: String, url: String, doc_id: String) -> Self {
+    pub const fn new(doc_mpath: String, url: String, doc_id: String, stele: String) -> Self {
         Self {
             doc_mpath,
             url,
             doc_id,
+            stele,
         }
     }
 }

src/db/models/library/manager.rs

@@ -10,18 +10,19 @@ impl super::Manager for DatabaseConnection {
     ///
     /// # Errors
     /// Errors if can't establish a connection to the database.
-    async fn find_lib_mpath_by_url(&self, url: &str) -> anyhow::Result<String> {
+    async fn find_lib_mpath_by_url(&self, url: &str, stele: &str) -> anyhow::Result<String> {
         let statement = "
             SELECT l.mpath
             FROM library l
-            WHERE l.url = $1
+            WHERE l.url = $1 AND l.stele = $2
             LIMIT 1
         ";
         let row = match self.kind {
             DatabaseKind::Postgres | DatabaseKind::Sqlite => {
                 let mut connection = self.pool.acquire().await?;
                 sqlx::query_as::<_, (String,)>(statement)
                     .bind(url)
+                    .bind(stele)
                     .fetch_one(&mut *connection)
                     .await?
             }
@@ -37,13 +38,18 @@ impl super::TxManager for DatabaseTransaction {
     /// # Errors
     /// Errors if the libraries cannot be inserted into the database.
     async fn insert_bulk(&mut self, libraries: Vec<Library>) -> anyhow::Result<()> {
-        let mut query_builder = QueryBuilder::new("INSERT OR IGNORE INTO library ( mpath, url ) ");
+        let mut query_builder =
+            QueryBuilder::new("INSERT OR IGNORE INTO library ( mpath, url, stele ) ");
         for chunk in libraries.chunks(BATCH_SIZE) {
             query_builder.push_values(chunk, |mut bindings, lb| {
-                bindings.push_bind(&lb.mpath).push_bind(&lb.url);
+                bindings
+                    .push_bind(&lb.mpath)
+                    .push_bind(&lb.url)
+                    .push_bind(&lb.stele);
             });
             let query = query_builder.build();
             query.execute(&mut *self.tx).await?;
+            query_builder.reset();
         }
         Ok(())
     }

src/db/models/library/mod.rs

@@ -7,7 +7,7 @@ pub mod manager;
 #[async_trait]
 pub trait Manager {
     /// Find one library materialized path by url.
-    async fn find_lib_mpath_by_url(&self, url: &str) -> anyhow::Result<String>;
+    async fn find_lib_mpath_by_url(&self, url: &str, stele: &str) -> anyhow::Result<String>;
 }
 
 /// Trait for managing transactions on publication versions.
@@ -24,12 +24,14 @@ pub struct Library {
     pub mpath: String,
     /// Url to the collection.
     pub url: String,
+    /// Reference to the stele.
+    pub stele: String,
 }
 
 impl Library {
     /// Create a new library.
     #[must_use]
-    pub const fn new(mpath: String, url: String) -> Self {
-        Self { mpath, url }
+    pub const fn new(mpath: String, url: String, stele: String) -> Self {
+        Self { mpath, url, stele }
     }
 }

src/db/models/library_change/manager.rs

@@ -85,6 +85,7 @@ impl super::TxManager for DatabaseTransaction {
             });
             let query = query_builder.build();
             query.execute(&mut *self.tx).await?;
+            query_builder.reset();
         }
         Ok(())
     }

src/history/changes.rs

@@ -351,8 +351,11 @@ async fn insert_document_changes(
         let doc_id =
             pub_graph.literal_from_triple_matching(Some(version), Some(oll::docId), None)?;
         document::TxManager::create(tx, &doc_id).await?;
-        let changes_uri =
-            pub_graph.iri_from_triple_matching(Some(version), Some(oll::hasChanges), None)?;
+        let Ok(changes_uri) =
+            pub_graph.iri_from_triple_matching(Some(version), Some(oll::hasChanges), None)
+        else {
+            continue;
+        };
         let changes = Bag::new(pub_graph, changes_uri);
         for change in changes.items()? {
             let doc_mpath = pub_graph.literal_from_triple_matching(
@@ -366,6 +369,7 @@ async fn insert_document_changes(
                 doc_mpath.clone(),
                 url.clone(),
                 doc_id.clone(),
+                publication.stele.clone(),
             ));
             let reason = pub_graph
                 .literal_from_triple_matching(Some(&change), Some(oll::reason), None)
@@ -425,7 +429,11 @@ async fn insert_library_changes(
         let el_status =
             pub_graph.literal_from_triple_matching(Some(version), Some(oll::status), None)?;
         let library_status = Status::from_string(&el_status)?;
-        library_bulk.push(Library::new(library_mpath.clone(), url.clone()));
+        library_bulk.push(Library::new(
+            library_mpath.clone(),
+            url.clone(),
+            publication.stele.clone(),
+        ));
         let pub_version_hash =
             md5::compute(publication.name.clone() + &codified_date + &publication.stele);
         library_changes_bulk.push(LibraryChange::new(

src/history/rdf/graph.rs

@@ -115,7 +115,9 @@ impl StelaeGraph {
         let triple = self
             .triples_matching_inner(subject, predicate, object)
             .next()
-            .context("Expected to find triple matching")?;
+            .context(format!(
+                "Expected to find triple matching s={subject:?}, p={predicate:?}, o={object:?}"
+            ))?;
         Ok(triple?)
     }
 

src/server/api/versions/mod.rs

@@ -157,7 +157,8 @@ async fn publication_versions(
 ) -> Vec<response::Version> {
     tracing::debug!("Fetching publication versions for '{url}'");
     let mut versions = vec![];
-    let doc_mpath = document_element::Manager::find_doc_mpath_by_url(db, &url).await;
+    let doc_mpath =
+        document_element::Manager::find_doc_mpath_by_url(db, &url, &publication.stele).await;
     if let Ok(mpath) = doc_mpath {
         let doc_versions =
             document_change::Manager::find_all_document_versions_by_mpath_and_publication(
@@ -169,7 +170,7 @@ async fn publication_versions(
             .unwrap_or_default();
         versions = doc_versions.into_iter().map(Into::into).collect();
     } else {
-        let lib_mpath = library::Manager::find_lib_mpath_by_url(db, &url).await;
+        let lib_mpath = library::Manager::find_lib_mpath_by_url(db, &url, &publication.stele).await;
         if let Ok(mpath) = lib_mpath {
             let coll_versions =
                 library_change::Manager::find_all_collection_versions_by_mpath_and_publication(