From 03f03576407749a4b430ad1fee36f0bd23df0bcf Mon Sep 17 00:00:00 2001
From: Mia <mia@parakeet.at>
Date: Wed, 29 Jan 2025 21:18:33 +0000
Subject: [PATCH] update repo state (rev & cid) on commit

---
 consumer/src/indexer/db.rs  | 23 +++++++++++++
 consumer/src/indexer/mod.rs | 67 +++++++++++++++++++++----------------
 2 files changed, 61 insertions(+), 29 deletions(-)

diff --git a/consumer/src/indexer/db.rs b/consumer/src/indexer/db.rs
index 7a37e87f..ebbf4feb 100644
--- a/consumer/src/indexer/db.rs
+++ b/consumer/src/indexer/db.rs
@@ -1,6 +1,7 @@
 use chrono::prelude::*;
 use diesel::prelude::*;
 use diesel_async::{AsyncPgConnection, RunQueryDsl};
+use ipld_core::cid::Cid;
 use parakeet_db::{models, schema, types};
 
 pub async fn upsert_actor(
@@ -37,3 +38,25 @@ pub async fn account_sync_status(
         .await
         .optional()
 }
+
+pub async fn update_repo_state(
+    conn: &mut AsyncPgConnection,
+    repo: &str,
+    rev: &str,
+    cid: Cid,
+) -> QueryResult<usize> {
+    diesel::update(schema::actors::table)
+        .set((
+            schema::actors::repo_rev.eq(rev),
+            schema::actors::repo_cid.eq(cid.to_string()),
+        ))
+        .filter(
+            schema::actors::did.eq(repo).and(
+                schema::actors::repo_rev
+                    .lt(rev)
+                    .or(schema::actors::repo_rev.is_null()),
+            ),
+        )
+        .execute(conn)
+        .await
+}
diff --git a/consumer/src/indexer/mod.rs b/consumer/src/indexer/mod.rs
index aac06f4c..41598a54 100644
--- a/consumer/src/indexer/mod.rs
+++ b/consumer/src/indexer/mod.rs
@@ -1,7 +1,7 @@
 use crate::firehose::{AtpAccountEvent, AtpCommitEvent, AtpIdentityEvent, FirehoseEvent};
 use crate::indexer::types::{CollectionType, RecordTypes};
 use diesel_async::pooled_connection::deadpool::Pool;
-use diesel_async::AsyncPgConnection;
+use diesel_async::{AsyncConnection, AsyncPgConnection};
 use futures::StreamExt;
 use ipld_core::cid::Cid;
 use parakeet_db::types::{ActorStatus, ActorSyncState};
@@ -103,39 +103,48 @@ async fn index_commit(conn: &mut AsyncPgConnection, commit: AtpCommitEvent) -> e
         }
     };
 
-    for op in &commit.ops {
-        let Some((collection_raw, rkey)) = op.path.split_once("/") else {
-            tracing::warn!("op contained invalid path {}", op.path);
-            continue;
-        };
-
-        let collection = CollectionType::from_str(collection_raw);
-        if collection == CollectionType::Unsupported {
-            tracing::debug!("{} {collection_raw} is unsupported", op.action);
-            continue;
+    conn.transaction::<_, diesel::result::Error, _>(|t| Box::pin(async move {
+        // ensure new repo rev > current rev, but only when the repo is synced
+        // (if repo is processing, they'll end up on the queue anyway)
+        if is_active && db::update_repo_state(t, &commit.repo, &commit.rev, commit.commit).await? == 0 {
+            tracing::warn!("Got a repo update older than the current rev. not processing ops.");
+            return Ok(false);
         }
 
-        if op.action == "create" || op.action == "update" {
-            if let Some(block) = op.cid.and_then(|cid| blocks.get(&cid)) {
-                let decoded: RecordTypes = match serde_ipld_dagcbor::from_slice(block) {
-                    Ok(decoded) => decoded,
-                    Err(err) => {
-                        tracing::error!("Failed to decode record: {err}");
-                        continue;
-                    }
-                };
-
-                dbg!(decoded);
-            } else {
-                tracing::error!("Missing Cid or the block was not found");
+        for op in &commit.ops {
+            let Some((collection_raw, rkey)) = op.path.split_once("/") else {
+                tracing::warn!("op contained invalid path {}", op.path);
+                continue;
+            };
+
+            let collection = CollectionType::from_str(collection_raw);
+            if collection == CollectionType::Unsupported {
+                tracing::debug!("{} {collection_raw} is unsupported", op.action);
                 continue;
             }
-        } else if op.action == "delete" {
-            //
-        } else {
-            tracing::warn!("op contained invalid action {}", op.action);
+
+            if op.action == "create" || op.action == "update" {
+                if let Some(block) = op.cid.and_then(|cid| blocks.get(&cid)) {
+                    let decoded: RecordTypes = match serde_ipld_dagcbor::from_slice(block) {
+                        Ok(decoded) => decoded,
+                        Err(err) => {
+                            tracing::error!("Failed to decode record: {err}");
+                            continue;
+                        }
+                    };
+                } else {
+                    tracing::error!("Missing Cid or the block was not found");
+                    continue;
+                }
+            } else if op.action == "delete" {
+                //
+            } else {
+                tracing::warn!("op contained invalid action {}", op.action);
+            }
         }
-    }
+
+        Ok(true)
+    })).await?;
 
     Ok(())
 }
-- 
2.54.0