use std::str::FromStr;
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 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 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,
)
.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,
)
.whatever_context("Failed processing issues")?;
}
Ok(())
}
pub 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: serde::Serialize + for<'de> serde::Deserialize<'de>,
{
let last_operation_id = self
.storage
.get_last_processed_operation(rid, &id, typename)
.whatever_context("Unable to get last processed operation")?;
if let Some(ref 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(ref since_id) = last_operation_id {
let since_id = radicle::git::Oid::from_str(&since_id).whatever_context("context")?;
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.clone();
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())
.whatever_context("Unable to query for existing operation")?
{
tracing::debug!("Operation already exists in storage, skipping");
continue;
}
let repo_alias = self
.storage
.resolve_rid(rid)
.whatever_context("Unable to resolve repo alias")?;
new_operations.push(OperationEntry {
operation_id: stream_entry.id().to_string(),
rid: rid.to_string(),
repo_alias,
created_at: stream_entry.timestamp.as_secs() as i32,
actions: stream_entry
.actions
.iter()
.filter_map(|action| serde_json::to_value(action).ok())
.collect::<Vec<_>>(),
author: stream_entry.author.to_string(),
author_alias: aliases.alias(&stream_entry.author).map(Into::into),
typename: typename.to_string(),
});
last_processed_id = Some(stream_entry.id().to_string());
}
self.storage
.insert_timeline_entry(&TimelineEntry {
repo: rid.to_string(),
node: self.profile.id().to_string(),
cob_id: id.to_string(),
cob_title: title,
cob_status: status,
typename: typename.to_string(),
last_operation_id: last_processed_id.map(|id| id.to_string()),
operations: new_operations
.clone()
.into_iter()
.map(|s| s.operation_id.to_string())
.collect::<Vec<_>>(),
})
.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)
.whatever_context("Failed insert batch")?;
} else {
tracing::debug!("{id} No new operations found");
}
Ok(())
}
pub fn get_stats(&mut self) -> Result<crate::storage::StorageStats, S::Error> {
self.storage.get_stats()
}
}
#[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(())
}
}