fix duplicate del/inv/rev insertion

kepler-core/src/models/delegation.rs

@@ -2,7 +2,7 @@ use crate::hash::Hash;
 use crate::types::{Facts, Resource};
 use crate::{events::Delegation, models::*, relationships::*, util};
 use kepler_lib::{authorization::KeplerDelegation, resolver::DID_METHODS};
-use sea_orm::{entity::prelude::*, ConnectionTrait};
+use sea_orm::{entity::prelude::*, sea_query::OnConflict, ConnectionTrait};
 use time::OffsetDateTime;
 
 #[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
@@ -227,7 +227,7 @@ async fn save<C: ConnectionTrait>(
     let hash: Hash = crate::hash::hash(&serialization);
 
     // save delegation
-    Entity::insert(ActiveModel::from(Model {
+    match Entity::insert(ActiveModel::from(Model {
         id: hash,
         delegator: delegation.delegator,
         delegatee: delegation.delegate,
@@ -237,8 +237,15 @@ async fn save<C: ConnectionTrait>(
         facts: None,
         serialization,
     }))
+    .on_conflict(OnConflict::column(Column::Id).do_nothing().to_owned())
     .exec(db)
-    .await?;
+    .await
+    {
+        Err(DbErr::RecordNotInserted) => return Ok(hash),
+        r => {
+            r?;
+        }
+    };
 
     // save abilities
     if !delegation.capabilities.is_empty() {
@@ -270,7 +277,6 @@ async fn save<C: ConnectionTrait>(
 }
 
 async fn save_actors<C: ConnectionTrait>(actors: &[&str], db: &C) -> Result<(), DbErr> {
-    use sea_orm::sea_query::OnConflict;
     match actor::Entity::insert_many(
         actors
             .iter()

kepler-core/src/models/invocation.rs

@@ -7,7 +7,7 @@ use super::super::{
 use crate::hash::Hash;
 use crate::types::{Facts, OrbitIdWrap, Resource};
 use kepler_lib::{authorization::KeplerInvocation, resolver::DID_METHODS};
-use sea_orm::{entity::prelude::*, Condition, ConnectionTrait, QueryOrder};
+use sea_orm::{entity::prelude::*, sea_query::OnConflict, Condition, ConnectionTrait, QueryOrder};
 use time::OffsetDateTime;
 
 #[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
@@ -191,15 +191,22 @@ async fn save<C: ConnectionTrait>(
     let hash = crate::hash::hash(&serialization);
     let issued_at = time.unwrap_or_else(OffsetDateTime::now_utc);
 
-    Entity::insert(ActiveModel::from(Model {
+    match Entity::insert(ActiveModel::from(Model {
         id: hash,
         issued_at,
         serialization,
         facts: None,
         invoker: invocation.invoker,
     }))
+    .on_conflict(OnConflict::column(Column::Id).do_nothing().to_owned())
     .exec(db)
-    .await?;
+    .await
+    {
+        Err(DbErr::RecordNotInserted) => return Ok(hash),
+        r => {
+            r?;
+        }
+    };
 
     // save invoked abilities
     if !invocation.capabilities.is_empty() {

kepler-core/src/models/revocation.rs

@@ -1,7 +1,7 @@
 use super::super::{events::Revocation, models::*, relationships::*};
 use crate::hash::{hash, Hash};
 use kepler_lib::authorization::KeplerRevocation;
-use sea_orm::{entity::prelude::*, ConnectionTrait};
+use sea_orm::{entity::prelude::*, sea_query::OnConflict, ConnectionTrait};
 use time::OffsetDateTime;
 
 #[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
@@ -95,14 +95,21 @@ pub(crate) async fn process<C: ConnectionTrait>(
         return Err(RevocationError::UnauthorizedRevoker(r.revoker).into());
     };
 
-    Entity::insert(ActiveModel::from(Model {
+    match Entity::insert(ActiveModel::from(Model {
         id: hash,
         serialization,
         revoker: r.revoker,
         revoked: delegation.id,
     }))
+    .on_conflict(OnConflict::column(Column::Id).do_nothing().to_owned())
     .exec(db)
-    .await?;
+    .await
+    {
+        Err(DbErr::RecordNotInserted) => return Ok(hash),
+        r => {
+            r?;
+        }
+    };
 
     if !r.parents.is_empty() {
         parent_delegations::Entity::insert_many(r.parents.into_iter().map(|p| {