radicle_feed/
feed.rs

1mod cobs;
2mod ops;
3mod repos;
4
5use std::str::FromStr;
6
7use radicle::cob::{ObjectId, Op, TypeName};
8use radicle::profile::{Aliases, Profile};
9use radicle::{issue, patch};
10use snafu::ResultExt;
11
12use crate::models::entry::TimelineEntry;
13use crate::storage::FeedStorage;
14
15pub struct FeedProcessor<'a, S: FeedStorage> {
16    storage: &'a mut S,
17    profile: Profile,
18}
19
20impl<'a, S: FeedStorage> FeedProcessor<'a, S> {
21    pub fn new(storage: &'a mut S, profile: Profile) -> Result<Self, S::Error> {
22        Ok(Self { storage, profile })
23    }
24
25    pub fn process_repository(
26        &mut self,
27        repo: &radicle::storage::RepositoryInfo,
28    ) -> Result<(), snafu::Whatever> {
29        let (git2_repo, patches, issues, aliases) = {
30            let repo_loader = repos::RepositoryLoader::new(&self.profile)?;
31            let (repo_handle, git2_repo) = repo_loader.open_repository(&repo.rid)?;
32            let patches = repo_loader.load_patches(&repo_handle)?;
33            let issues = repo_loader.load_issues(&repo_handle)?;
34            let aliases = self.profile.aliases();
35            (git2_repo, patches, issues, aliases)
36        };
37
38        // Process patches
39        for patch_id in patches {
40            self.process_single_cob::<patch::Action>(
41                &patch_id,
42                &patch::TYPENAME,
43                &git2_repo,
44                &repo.rid,
45                &aliases,
46            )?;
47        }
48
49        // Process issues
50        for issue_id in issues {
51            self.process_single_cob::<issue::Action>(
52                &issue_id,
53                &issue::TYPENAME,
54                &git2_repo,
55                &repo.rid,
56                &aliases,
57            )?;
58        }
59
60        Ok(())
61    }
62
63    fn process_single_cob<A>(
64        &mut self,
65        id: &ObjectId,
66        typename: &TypeName,
67        git2_repo: &radicle::git::raw::Repository,
68        rid: &radicle::prelude::RepoId,
69        aliases: &Aliases,
70    ) -> Result<(), snafu::Whatever>
71    where
72        A: serde::Serialize + for<'de> serde::Deserialize<'de> + Clone,
73    {
74        // Get the last processed operation
75        let last_operation_id = self
76            .storage
77            .get_last_processed_operation(rid, &id, typename)
78            .whatever_context("Unable to get last processed operation")?
79            .and_then(|oid| {
80                // To avoid missing operations in different storages we make sure to fallback
81                // to `None` if we are unable to load the operation
82                let oid: radicle::git::Oid = radicle::git::Oid::from_str(&oid).unwrap();
83                Op::<A>::load(git2_repo, oid).map(|op| op.id()).ok()
84            });
85
86        // Use stream processor to get new entries
87        let stream_processor = cobs::CobStreamProcessor::new(git2_repo, typename, id);
88        let stream_entries = stream_processor.fetch_entries_since::<A>(last_operation_id)?;
89
90        if stream_entries.is_empty() {
91            tracing::debug!("{id} No new operations to process");
92            return Ok(());
93        }
94
95        // Get the repo alias before building operations
96        let repo_alias = self
97            .storage
98            .resolve_rid(rid)
99            .whatever_context("Unable to resolve repo alias")?;
100
101        // Build operations from stream entries
102        let operation_builder = ops::OperationBuilder::new(&aliases, rid, typename, repo_alias);
103
104        let (operations, last_processed_id) = operation_builder.build_operations::<A, _>(
105            stream_entries,
106            last_operation_id,
107            self.storage,
108        )?;
109
110        // Write operations and timeline
111        self.storage
112            .insert_timeline_entry(&TimelineEntry {
113                repo: rid.to_string(),
114                node: self.profile.id().to_string(),
115                cob_id: id.to_string(),
116                typename: typename.to_string(),
117                last_operation_id: last_processed_id.map(|l| l.to_string()),
118                operations: operations
119                    .iter()
120                    .map(|op| op.operation_id.clone())
121                    .collect(),
122            })
123            .whatever_context("Insert timeline entry failed")?;
124
125        if !operations.is_empty() {
126            tracing::debug!("Inserting {} new operations into storage", operations.len());
127            self.storage
128                .insert_batch(&operations)
129                .whatever_context("Failed insert batch")?;
130        }
131
132        Ok(())
133    }
134
135    /// Get storage statistics
136    pub fn get_stats(&mut self) -> Result<crate::storage::StorageStats, S::Error> {
137        self.storage.get_stats()
138    }
139}
radicle_feed/feed.rs

radicle_feed/
feed.rs
