add more integration tests

src/watcher.rs

@@ -24,6 +24,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Sync + Send>> {
     // Prepare the sender channel to send the messages to the retry manager
     let sender_clone = retry_manager.sender.clone();
     // Create new asynchronous task to control the failed events
+    // TODO: Assure the thread safety
     tokio::spawn(async move {
         let _ = retry_manager.exec().await;
     });

tests/watcher/bookmarks/mod.rs

@@ -3,3 +3,4 @@ mod fail_index;
 mod raw;
 mod utils;
 mod viewer;
+mod retry_bookmark;

tests/watcher/bookmarks/retry_bookmark.rs

@@ -0,0 +1,77 @@
+use std::time::Duration;
+
+use crate::watcher::utils::watcher::WatcherTest;
+use anyhow::Result;
+use pubky_app_specs::{traits::HashId, PubkyAppBookmark, PubkyAppUser};
+use pubky_common::crypto::Keypair;
+use pubky_nexus::events::{error::EventProcessorError, retry::event::RetryEvent, EventType};
+
+/// The user profile is stored in the homeserver. Missing the post to connect the bookmark
+// These types of tests (e.g., retry_xxxx) can be used to verify whether the `RetryManager`
+// cache correctly adds the events as expected.
+#[tokio_shared_rt::test(shared)]
+async fn test_homeserver_bookmark_cannot_index() -> Result<()> {
+    let mut test = WatcherTest::setup().await?;
+
+    let keypair = Keypair::random();
+    let user = PubkyAppUser {
+        bio: Some("test_homeserver_bookmark_cannot_index".to_string()),
+        image: None,
+        links: None,
+        name: "Watcher:IndexFail:Bookmark:User".to_string(),
+        status: None,
+    };
+    let user_id = test.create_user(&keypair, &user).await?;
+
+    // Use a placeholder parent post ID to intentionally avoid resolving it in the graph database
+    let fake_post_id = "0032QB10HCRHG";
+    let fake_user_id = "ba3e8qeby33uq9cughpxdf7bew9etn1eq8bc3yhwg7p1f54yaozy";
+    // Create parent post uri
+    let post_uri = format!("pubky://{fake_user_id}/pub/pubky.app/posts/{fake_post_id}");
+
+    // Create a bookmark content
+    let bookmark = PubkyAppBookmark {
+        uri: post_uri,
+        created_at: chrono::Utc::now().timestamp_millis(),
+    };
+    let bookmark_blob = serde_json::to_vec(&bookmark)?;
+    // Create the bookmark of the shadow user
+    let bookmark_id = bookmark.create_id();
+    let bookmark_url = format!(
+        "pubky://{}/pub/pubky.app/bookmarks/{}",
+        user_id, bookmark_id
+    );
+    // Put bookmark
+    test.put(&bookmark_url, bookmark_blob).await?;
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    let index_key = format!(
+        "{}:{}",
+        EventType::Put,
+        RetryEvent::generate_index_key(&bookmark_url).unwrap()
+    );
+
+    // Assert if the event is in the timeline
+    let timestamp = RetryEvent::check_uri(&index_key).await.unwrap();
+    assert!(timestamp.is_some());
+
+    // Assert if the event is in the state hash map
+    let event_retry = RetryEvent::get_from_index(&index_key).await.unwrap();
+    assert!(event_retry.is_some());
+
+    let event_state = event_retry.unwrap();
+
+    assert_eq!(event_state.retry_count, 0);
+
+    let dependency_uri = format!("{fake_user_id}:posts:{fake_post_id}");
+    match event_state.error_type {
+        EventProcessorError::MissingDependency { dependency } => {
+            assert_eq!(dependency.len(), 1);
+            assert_eq!(dependency[0], dependency_uri);
+        }
+        _ => assert!(false, "The error type has to be MissingDependency type"),
+    };
+
+
+    Ok(())
+}

tests/watcher/follows/mod.rs

@@ -8,3 +8,4 @@ mod put_friends;
 mod put_notification;
 mod put_sequential;
 mod utils;
+mod retry_follow;

tests/watcher/follows/retry_follow.rs

@@ -0,0 +1,66 @@
+use std::time::Duration;
+
+use crate::watcher::utils::watcher::WatcherTest;
+use anyhow::Result;
+use pubky_app_specs::PubkyAppUser;
+use pubky_common::crypto::Keypair;
+use pubky_nexus::events::{error::EventProcessorError, retry::event::RetryEvent, EventType};
+
+/// The user profile is stored in the homeserver. Missing the followee to connect with follower
+// These types of tests (e.g., retry_xxxx) can be used to verify whether the `RetryManager`
+// cache correctly adds the events as expected.
+#[tokio_shared_rt::test(shared)]
+async fn test_homeserver_follow_cannot_index() -> Result<()> {
+    let mut test = WatcherTest::setup().await?;
+
+    let followee_keypair = Keypair::random();
+    let followee_id = followee_keypair.public_key().to_z32();
+    // In that case, that user will act as a NotSyncUser or user not registered in pubky.app
+    // It will not have a profile.json
+    test.register_user(&followee_keypair).await?;
+
+    let follower_keypair = Keypair::random();
+    let follower_user = PubkyAppUser {
+        bio: Some("test_homeserver_follow_cannot_index".to_string()),
+        image: None,
+        links: None,
+        name: "Watcher:IndexFail:Follower".to_string(),
+        status: None,
+    };
+    let follower_id = test.create_user(&follower_keypair, &follower_user).await?;
+
+    // Mute the user
+    test.create_follow(&follower_id, &followee_id).await?;
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    let index_key = format!(
+        "{}:{}",
+        EventType::Put,
+        RetryEvent::generate_index_key(&format!("pubky://{follower_id}/pub/pubky.app/follows/{followee_id}")).unwrap()
+    );
+
+    // Assert if the event is in the timeline
+    let timestamp = RetryEvent::check_uri(&index_key).await.unwrap();
+    assert!(timestamp.is_some());
+
+    // Assert if the event is in the state hash map
+    let event_retry = RetryEvent::get_from_index(&index_key).await.unwrap();
+    assert!(event_retry.is_some());
+
+    let event_state = event_retry.unwrap();
+
+    assert_eq!(event_state.retry_count, 0);
+
+    let dependency_uri = format!("{followee_id}:user:profile.json");
+
+    match event_state.error_type {
+        EventProcessorError::MissingDependency { dependency } => {
+            assert_eq!(dependency.len(), 1);
+            assert_eq!(dependency[0], dependency_uri)
+        }
+        _ => assert!(false, "The error type has to be MissingDependency type"),
+    };
+
+
+    Ok(())
+}

tests/watcher/mutes/mod.rs

@@ -1,4 +1,5 @@
 mod del;
 mod fail_index;
+mod retry_mute;
 mod put;
 mod utils;

tests/watcher/mutes/retry_mute.rs

@@ -0,0 +1,66 @@
+use std::time::Duration;
+
+use crate::watcher::utils::watcher::WatcherTest;
+use anyhow::Result;
+use pubky_app_specs::PubkyAppUser;
+use pubky_common::crypto::Keypair;
+use pubky_nexus::events::{error::EventProcessorError, retry::event::RetryEvent, EventType};
+
+/// The user profile is stored in the homeserver. Missing the mutee to connect with muter
+// These types of tests (e.g., retry_xxxx) can be used to verify whether the `RetryManager`
+// cache correctly adds the events as expected.
+#[tokio_shared_rt::test(shared)]
+async fn test_homeserver_mute_cannot_index() -> Result<()> {
+    let mut test = WatcherTest::setup().await?;
+
+    let mutee_keypair = Keypair::random();
+    let mutee_id = mutee_keypair.public_key().to_z32();
+    // In that case, that user will act as a NotSyncUser or user not registered in pubky.app
+    // It will not have a profile.json
+    test.register_user(&mutee_keypair).await?;
+
+    let muter_keypair = Keypair::random();
+    let muter_user = PubkyAppUser {
+        bio: Some("test_homeserver_mute_cannot_index".to_string()),
+        image: None,
+        links: None,
+        name: "Watcher:IndexFail:Muter".to_string(),
+        status: None,
+    };
+    let muter_id = test.create_user(&muter_keypair, &muter_user).await?;
+
+    // Mute the user
+    test.create_mute(&muter_id, &mutee_id).await?;
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    let index_key = format!(
+        "{}:{}",
+        EventType::Put,
+        RetryEvent::generate_index_key(&format!("pubky://{muter_id}/pub/pubky.app/mutes/{mutee_id}")).unwrap()
+    );
+
+    // Assert if the event is in the timeline
+    let timestamp = RetryEvent::check_uri(&index_key).await.unwrap();
+    assert!(timestamp.is_some());
+
+    // Assert if the event is in the state hash map
+    let event_retry = RetryEvent::get_from_index(&index_key).await.unwrap();
+    assert!(event_retry.is_some());
+
+    let event_state = event_retry.unwrap();
+
+    assert_eq!(event_state.retry_count, 0);
+
+    let dependency_uri = format!("{mutee_id}:user:profile.json");
+
+    match event_state.error_type {
+        EventProcessorError::MissingDependency { dependency } => {
+            assert_eq!(dependency.len(), 1);
+            assert_eq!(dependency[0], dependency_uri)
+        }
+        _ => assert!(false, "The error type has to be MissingDependency type"),
+    };
+
+
+    Ok(())
+}

tests/watcher/posts/mod.rs

@@ -24,4 +24,8 @@ mod reply_notification;
 mod reply_repost;
 mod repost;
 mod repost_notification;
+mod retry_post;
+mod retry_reply;
+mod retry_repost;
+mod retry_all;
 pub mod utils;

tests/watcher/posts/retry_all.rs

@@ -0,0 +1,78 @@
+use std::time::Duration;
+
+use crate::watcher::utils::watcher::WatcherTest;
+use anyhow::Result;
+use pubky_app_specs::{PubkyAppPost, PubkyAppPostEmbed, PubkyAppPostKind, PubkyAppUser};
+use pubky_common::crypto::Keypair;
+use pubky_nexus::events::{error::EventProcessorError, retry::event::RetryEvent, EventType};
+
+/// The user profile is stored in the homeserver. Missing the post to connect the new one
+// These types of tests (e.g., retry_xxxx) can be used to verify whether the `RetryManager`
+// cache correctly adds the events as expected.
+#[tokio_shared_rt::test(shared)]
+async fn test_homeserver_post_with_reply_repost_cannot_index() -> Result<()> {
+    let mut test = WatcherTest::setup().await?;
+
+    let keypair = Keypair::random();
+
+    let user = PubkyAppUser {
+        bio: Some("test_homeserver_post_reply".to_string()),
+        image: None,
+        links: None,
+        name: "Watcher:IndexFail:PostRepost:User".to_string(),
+        status: None,
+    };
+
+    let user_id = test.create_user(&keypair, &user).await?;
+
+    // Use a placeholder parent post ID to intentionally avoid resolving it in the graph database
+    let reply_fake_post_id = "0032QB10HCRHG";
+    let repost_fake_post_id = "0032QB10HP6JJ";
+    // Create parent post uri
+    let reply_uri = format!("pubky://{user_id}/pub/pubky.app/posts/{reply_fake_post_id}");
+    let repost_uri = format!("pubky://{user_id}/pub/pubky.app/posts/{repost_fake_post_id}");
+
+    let repost_reply_post = PubkyAppPost {
+        content: "Watcher:IndexFail:PostRepost:User:Reply".to_string(),
+        kind: PubkyAppPostKind::Short,
+        parent: Some(reply_uri.clone()),
+        embed: Some(PubkyAppPostEmbed {
+            kind: PubkyAppPostKind::Short,
+            uri: repost_uri.clone(),
+        }),
+        attachments: None,
+    };
+
+    let repost_reply_post_id = test.create_post(&user_id, &repost_reply_post).await?;
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    let index_key = format!(
+        "{}:{}",
+        EventType::Put,
+        RetryEvent::generate_index_key(&format!("pubky://{user_id}/pub/pubky.app/posts/{repost_reply_post_id}")).unwrap()
+    );
+
+    // Assert if the event is in the timeline
+    let timestamp = RetryEvent::check_uri(&index_key).await.unwrap();
+    assert!(timestamp.is_some());
+
+    // Assert if the event is in the state hash map
+    let event_retry = RetryEvent::get_from_index(&index_key).await.unwrap();
+    assert!(event_retry.is_some());
+
+    let event_state = event_retry.unwrap();
+
+    assert_eq!(event_state.retry_count, 0);
+
+    match event_state.error_type {
+        EventProcessorError::MissingDependency { dependency } => {
+            assert_eq!(dependency.len(), 2);
+            assert_eq!(dependency[0], RetryEvent::generate_index_key(&reply_uri).unwrap());
+            assert_eq!(dependency[1], RetryEvent::generate_index_key(&repost_uri).unwrap());
+        }
+        _ => assert!(false, "The error type has to be MissingDependency type"),
+    };
+
+
+    Ok(())
+}

tests/watcher/posts/retry_post.rs

@@ -0,0 +1,64 @@
+use std::time::Duration;
+
+use crate::watcher::utils::watcher::WatcherTest;
+use anyhow::Result;
+use pubky_app_specs::{PubkyAppPost, PubkyAppPostKind};
+use pubky_common::crypto::Keypair;
+use pubky_nexus::events::{error::EventProcessorError, retry::event::RetryEvent, EventType};
+
+/// The user profile is stored in the homeserver. Missing the author to connect the post
+// These types of tests (e.g., retry_xxxx) can be used to verify whether the `RetryManager`
+// cache correctly adds the events as expected.
+#[tokio_shared_rt::test(shared)]
+async fn test_homeserver_post_cannot_index() -> Result<()> {
+    let mut test = WatcherTest::setup().await?;
+
+    let keypair = Keypair::random();
+    let user_id = keypair.public_key().to_z32();
+
+    // In that case, that user will act as a NotSyncUser or user not registered in pubky.app
+    // It will not have a profile.json
+    test.register_user(&keypair).await?;
+
+    let post = PubkyAppPost {
+        content: "Watcher:IndexFail:PostEvent:PostWithoutUser".to_string(),
+        kind: PubkyAppPostKind::Short,
+        parent: None,
+        embed: None,
+        attachments: None,
+    };
+
+    let post_id = test.create_post(&user_id, &post).await?;
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    let index_key = format!(
+        "{}:{}",
+        EventType::Put,
+        RetryEvent::generate_index_key(&format!("pubky://{user_id}/pub/pubky.app/posts/{post_id}")).unwrap()
+    );
+
+    // Assert if the event is in the timeline
+    let timestamp = RetryEvent::check_uri(&index_key).await.unwrap();
+    assert!(timestamp.is_some());
+
+    // Assert if the event is in the state hash map
+    let event_retry = RetryEvent::get_from_index(&index_key).await.unwrap();
+    assert!(event_retry.is_some());
+
+    let event_state = event_retry.unwrap();
+
+    assert_eq!(event_state.retry_count, 0);
+
+    let dependency_uri = format!("pubky://{user_id}/pub/pubky.app/profile.json");
+
+    match event_state.error_type {
+        EventProcessorError::MissingDependency { dependency } => {
+            assert_eq!(dependency.len(), 1);
+            assert_eq!(dependency[0], RetryEvent::generate_index_key(&dependency_uri).unwrap());
+        }
+        _ => assert!(false, "The error type has to be MissingDependency type"),
+    };
+
+
+    Ok(())
+}