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::{OptionExt, 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 stored_patch = patches
.get(&patch)
.ok()
.flatten()
.whatever_context(format!("Unable to load patch {}", patch))?;
let cob_title = stored_patch.title().to_string();
let cob_status = stored_patch.state().to_string();
self.process_cob::<patch::Action>(
&CobInfo {
id: patch,
title: cob_title,
status: cob_status,
},
&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 stored_issue = issues
.get(&issue)
.ok()
.flatten()
.whatever_context(format!("Unable to load issue {}", issue))?;
let cob_title = stored_issue.title().to_string();
let cob_status = stored_issue.state().to_string();
self.process_cob::<issue::Action>(
&CobInfo {
id: issue,
title: cob_title,
status: cob_status,
},
&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()) .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.to_owned(),
cob_status: status.to_owned(),
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(())
}
pub async fn get_stats(&self) -> Result<crate::storage::StorageStats, S::Error> {
self.storage.get_stats().await
}
}
#[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(())
}
}