radicle_feed/

feed.rs

use radicle::cob::object::Storage;
use radicle::cob::stream::{CobRange, CobStream, Stream};
use radicle::cob::TypeName;
use radicle::node::AliasStore;
use radicle::profile::{Aliases, Profile};
use radicle::storage::git::paths;
use radicle::{issue, patch};
use serde::{Deserialize, Serialize};
use snafu::ResultExt;

use crate::entry::{OperationEntry, TimelineEntry};
use crate::radicle_extra::cob::CobInfo;
use crate::storage::FeedStorage;

pub struct FeedProcessor<S: FeedStorage> {
    storage: S,
    profile: Profile,
}

impl<S: FeedStorage> FeedProcessor<S> {
    pub fn new(storage: S, profile: Profile) -> Result<Self, S::Error> {
        Ok(Self { storage, profile })
    }

    pub async fn process_repository(
        &mut self,
        repo: &radicle::storage::RepositoryInfo,
    ) -> Result<(), snafu::Whatever> {
        let aliases = self.profile.aliases();
        let profile_storage = self.profile.storage.clone();

        tracing::info!("Processing repository: {}", repo.rid);

        let path = paths::repository(&profile_storage, &repo.rid);
        let repo_handle = radicle::storage::git::Repository::open(path.clone(), repo.rid)
            .whatever_context("Failed to open readonly repo")?;
        let git2_repo = radicle::git::raw::Repository::open(path)
            .whatever_context("Failed to open git2 repo")?;

        let patches = radicle::patch::Patches::open(&repo_handle)
            .whatever_context("Failed opening patches")?;
        let patch_ids = patches
            .as_ref()
            .types(&radicle::patch::TYPENAME)
            .whatever_context("Unable to load patch cob ids")?;
        for patch in patch_ids.into_keys() {
            let Some(stored_patch) = patches.get(&patch).ok().flatten() else {
                continue;
            };
            if stored_patch.title().is_empty() {
                continue;
            }
            self.process_cob::<patch::Action>(
                CobInfo {
                    id: patch,
                    title: stored_patch.title().to_string(),
                    status: stored_patch.state().to_string(),
                },
                &patch::TYPENAME,
                &git2_repo,
                &repo.rid,
                &aliases,
            )
            .await
            .whatever_context("Failed processing patches")?;
        }

        let issues =
            radicle::issue::Issues::open(&repo_handle).whatever_context("Failed opening issues")?;
        let issue_ids = issues
            .as_ref()
            .types(&issue::TYPENAME)
            .whatever_context("Unable to load issue cob ids")?;
        for issue in issue_ids.into_keys() {
            let Some(stored_issue) = issues.get(&issue).ok().flatten() else {
                continue;
            };
            if stored_issue.title().is_empty() {
                continue;
            }
            self.process_cob::<issue::Action>(
                CobInfo {
                    id: issue,
                    title: stored_issue.title().to_string(),
                    status: stored_issue.state().to_string(),
                },
                &issue::TYPENAME,
                &git2_repo,
                &repo.rid,
                &aliases,
            )
            .await
            .whatever_context("Failed processing issues")?;
        }

        Ok(())
    }

    pub async fn process_cob<A>(
        &mut self,
        CobInfo { id, title, status }: CobInfo,
        typename: &TypeName,
        git2_repo: &radicle::git::raw::Repository,
        rid: &radicle::prelude::RepoId,
        aliases: &Aliases,
    ) -> Result<(), snafu::Whatever>
    where
        A: Serialize + for<'de> Deserialize<'de>,
    {
        let last_operation_id = self
            .storage
            .get_last_processed_operation(rid, &id, typename)
            .await
            .whatever_context("Unable to get last processed operation")?;

        if let Some(since_id) = last_operation_id {
            tracing::debug!("{id} Last processed operation ID: {}", since_id);
        } else {
            tracing::debug!("{id} No previous operations found (first time processing)");
        }

        let stream = Stream::<A>::new(git2_repo, CobRange::new(typename, &id), typename.clone());

        let stream_entries = if let Some(since_id) = last_operation_id {
            let since_entries: Vec<_> = stream
                .since(since_id.into())
                .whatever_context("Unable to create since cob stream")?
                .filter_map(|s| s.ok())
                .filter(|entry| entry.id() != since_id.into()) // Filter out the already processed operation
                .collect();
            tracing::debug!(
                "{id} Found {} new operations since last processed",
                since_entries.len()
            );

            since_entries
        } else {
            let all_entries: Vec<_> = stream
                .all()
                .whatever_context("Unable to create all cob stream")?
                .filter_map(|s| s.ok())
                .collect();
            tracing::debug!("{id} Found {} total operations", all_entries.len());

            all_entries
        };

        if stream_entries.is_empty() {
            tracing::debug!("{id} No new operations to process");
            return Ok(());
        }

        let mut last_processed_id = last_operation_id;
        let mut new_operations = Vec::new();

        for stream_entry in stream_entries {
            tracing::debug!("Processing operation: {}", stream_entry.id());

            if self
                .storage
                .operation_exists(&stream_entry.id())
                .await
                .whatever_context("Unable to query for existing operation")?
            {
                tracing::debug!("Operation already exists in storage, skipping");
                continue;
            }

            new_operations.push(OperationEntry {
                operation_id: stream_entry.id(),
                cob_id: id,
                rid: *rid,
                timestamp: stream_entry.timestamp,
                actions: stream_entry
                    .actions
                    .iter()
                    .filter_map(|action| serde_json::to_string(&action).ok())
                    .collect::<Vec<_>>(),
                author: stream_entry.author,
                author_alias: aliases.alias(&stream_entry.author),
                typename: typename.clone(),
            });

            last_processed_id = Some(stream_entry.id().into());
        }

        self.storage
            .insert_timeline_entry(&TimelineEntry {
                repo: rid.to_owned(),
                node: self.profile.id().to_owned(),
                cob_id: id.to_owned(),
                cob_title: title,
                cob_status: status,
                typename: typename.to_owned(),
                last_operation_id: last_processed_id.map(Into::into),
                operations: new_operations
                    .clone()
                    .into_iter()
                    .map(|s| s.operation_id.to_string())
                    .collect::<Vec<_>>(),
            })
            .await
            .whatever_context("Insert timeline entry failed")?;

        if !new_operations.is_empty() {
            tracing::debug!(
                "Inserting {} new operations into storage",
                new_operations.len()
            );
            self.storage
                .insert_batch(&new_operations)
                .await
                .whatever_context("Failed insert batch")?;
        } else {
            tracing::debug!("{id} No new operations found");
        }

        Ok(())
    }

    /// Get storage statistics
    pub async fn get_stats(&self) -> Result<crate::storage::StorageStats, S::Error> {
        self.storage.get_stats().await
    }
}

/// Statistics about the processing operation
#[derive(Debug, Default)]
pub struct ProcessingStats {
    pub repositories_processed: usize,
    pub operations_found: usize,
    pub operations_stored: usize,
}

impl std::fmt::Display for ProcessingStats {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        writeln!(f, "=== Processing Statistics ===")?;
        writeln!(f, "Repositories processed: {}", self.repositories_processed)?;
        writeln!(f, "Operations found: {}", self.operations_found)?;
        writeln!(f, "Operations stored: {}", self.operations_stored)?;
        Ok(())
    }
}