radicle-feed 0.5.3

mod cobs;
mod ops;
mod repos;

use std::str::FromStr;

use radicle::cob::{ObjectId, Op, TypeName};
use radicle::profile::{Aliases, Profile};
use radicle::{issue, patch};
use snafu::ResultExt;

use crate::models::entry::TimelineEntry;
use crate::storage::FeedStorage;

pub struct FeedProcessor<'a, S: FeedStorage> {
    storage: &'a mut S,
    profile: Profile,
}

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

    pub fn process_repository(
        &mut self,
        repo: &radicle::storage::RepositoryInfo,
    ) -> Result<(), snafu::Whatever> {
        let (git2_repo, patches, issues, aliases) = {
            let repo_loader = repos::RepositoryLoader::new(&self.profile)?;
            let (repo_handle, git2_repo) = repo_loader.open_repository(&repo.rid)?;
            let patches = repo_loader.load_patches(&repo_handle)?;
            let issues = repo_loader.load_issues(&repo_handle)?;
            let aliases = self.profile.aliases();
            (git2_repo, patches, issues, aliases)
        };

        // Process patches
        for patch_id in patches {
            self.process_single_cob::<patch::Action>(
                &patch_id,
                &patch::TYPENAME,
                &git2_repo,
                &repo.rid,
                &aliases,
            )?;
        }

        // Process issues
        for issue_id in issues {
            self.process_single_cob::<issue::Action>(
                &issue_id,
                &issue::TYPENAME,
                &git2_repo,
                &repo.rid,
                &aliases,
            )?;
        }

        Ok(())
    }

    fn process_single_cob<A>(
        &mut self,
        id: &ObjectId,
        typename: &TypeName,
        git2_repo: &radicle::git::raw::Repository,
        rid: &radicle::prelude::RepoId,
        aliases: &Aliases,
    ) -> Result<(), snafu::Whatever>
    where
        A: serde::Serialize + for<'de> serde::Deserialize<'de> + Clone,
    {
        // Get the last processed operation
        let last_operation_id = self
            .storage
            .get_last_processed_operation(rid, &id, typename)
            .whatever_context("Unable to get last processed operation")?
            .and_then(|oid| {
                // To avoid missing operations in different storages we make sure to fallback
                // to `None` if we are unable to load the operation
                let oid: radicle::git::Oid = radicle::git::Oid::from_str(&oid).unwrap();
                Op::<A>::load(git2_repo, oid).map(|op| op.id()).ok()
            });

        // Use stream processor to get new entries
        let stream_processor = cobs::CobStreamProcessor::new(git2_repo, typename, id);
        let stream_entries = stream_processor.fetch_entries_since::<A>(last_operation_id)?;

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

        // Get the repo alias before building operations
        let repo_alias = self
            .storage
            .resolve_rid(rid)
            .whatever_context("Unable to resolve repo alias")?;

        // Build operations from stream entries
        let operation_builder = ops::OperationBuilder::new(&aliases, rid, typename, repo_alias);

        let (operations, last_processed_id) = operation_builder.build_operations::<A, _>(
            stream_entries,
            last_operation_id,
            self.storage,
        )?;

        // Write operations and timeline
        self.storage
            .insert_timeline_entry(&TimelineEntry {
                repo: rid.to_string(),
                node: self.profile.id().to_string(),
                cob_id: id.to_string(),
                typename: typename.to_string(),
                last_operation_id: last_processed_id.map(|l| l.to_string()),
                operations: operations
                    .iter()
                    .map(|op| op.operation_id.clone())
                    .collect(),
            })
            .whatever_context("Insert timeline entry failed")?;

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

        Ok(())
    }

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