use std::fmt::Debug;
use radicle::cob::object::Storage;
use radicle::cob::{Op, TypeName};
use radicle::crypto::PublicKey;
use radicle::issue::Issue;
use radicle::node::AliasStore;
use radicle::profile::{Aliases, Profile};
use radicle::storage::git::paths;
use radicle::{issue, patch};
use radicle_cob::change::store::Entry;
use radicle_cob::signatures::ExtendedSignature;
use radicle_cob::{CollaborativeObject, EntryId, Evaluate, ObjectId};
use serde::{Deserialize, Serialize};
use snafu::{whatever, ResultExt};
use crate::entry::{OperationEntry, TimelineEntry};
use crate::storage::FeedStorage;
pub trait HasCobInfo {
fn title(&self) -> String;
fn state(&self) -> String;
}
impl HasCobInfo for radicle::issue::Issue {
fn title(&self) -> String {
self.title().to_string()
}
fn state(&self) -> String {
self.state().to_string()
}
}
impl HasCobInfo for radicle::patch::Patch {
fn title(&self) -> String {
self.title().to_string()
}
fn state(&self) -> String {
self.state().to_string()
}
}
pub struct FeedProcessor<S: FeedStorage> {
storage: S,
profile: Profile,
}
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct ObjectSnapshot<T> {
pub entry_id: EntryId,
pub timestamp: u64,
pub author: PublicKey,
pub actions: Vec<serde_json::Value>,
pub state: T,
}
pub fn iterate_with_snapshots<T, S, A>(
cob: CollaborativeObject<T>,
store: &S,
) -> Result<Vec<ObjectSnapshot<T>>, snafu::Whatever>
where
A: Serialize + for<'de> Deserialize<'de>,
T: Evaluate<S> + Clone,
S: radicle::cob::object::Storage,
S: radicle::cob::change::Storage<
ObjectId = radicle::storage::Oid,
Parent = radicle::storage::Oid,
Signatures = ExtendedSignature,
>,
{
let mut snapshots = Vec::new();
let history = cob.history();
let root_entry = history.root();
let mut current_state =
T::init(root_entry, store).whatever_context("Unable to init root state")?;
let op = Op::<A>::load::<S>(store, *root_entry.id());
if let Err(err) = op {
whatever!("Skipping {}: {}", cob.id(), err);
}
let actions = op
.unwrap()
.actions
.into_iter()
.map(|action| serde_json::to_value(action))
.collect::<Result<Vec<serde_json::Value>, _>>()
.whatever_context("Unable to convert T into a json")?;
snapshots.push(ObjectSnapshot {
entry_id: *root_entry.id(),
timestamp: root_entry.timestamp,
author: root_entry.signature.key,
actions,
state: current_state.clone(),
});
let sorted_entries: Vec<&Entry<_, _, _>> = history.sorted(|a, b| a.cmp(b)).collect();
for entry in sorted_entries.iter().skip(1) {
let entry_id = *entry.id();
let op = Op::<A>::load::<S>(store, entry_id)
.whatever_context("Unable to load siblings operation")?;
let actions = op
.actions
.into_iter()
.map(|action| serde_json::to_value(action))
.collect::<Result<Vec<serde_json::Value>, _>>()
.whatever_context("Unable to convert T into a json")?;
let concurrent = history.children_of(&entry_id);
let siblings = concurrent
.iter()
.filter_map(|id| history.graph().get(id))
.map(|node| (&node.id, &node.value));
if current_state.apply(entry, siblings, store).is_ok() {
snapshots.push(ObjectSnapshot {
entry_id,
timestamp: entry.timestamp,
author: root_entry.signature.key,
actions,
state: current_state.clone(),
});
}
}
Ok(snapshots)
}
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 patches = radicle::patch::Patches::open(&repo_handle)
.whatever_context("Failed opening patches")?;
let patch_references = patches
.as_ref()
.types(&radicle::patch::TYPENAME)
.whatever_context("Unable to load patch cob ids")?;
for patch in patch_references.into_keys() {
if let Ok(Some(cob)) = radicle_cob::object::collaboration::get::<patch::Patch, _>(
&repo_handle,
&patch::TYPENAME,
&patch,
) {
let Ok(snapshots) =
iterate_with_snapshots::<_, _, patch::Patch>(cob, patches.as_ref())
else {
continue;
};
self.process_cob::<patch::Patch>(
&patch,
&patch::TYPENAME,
snapshots,
&repo.rid,
&aliases,
)
.whatever_context("Failed processing patches")?;
}
}
let issues =
radicle::issue::Issues::open(&repo_handle).whatever_context("Failed opening issues")?;
let issue_references = issues
.as_ref()
.types(&issue::TYPENAME)
.whatever_context("Unable to load issue cob ids")?;
for issue in issue_references.into_keys() {
if let Ok(Some(cob)) = radicle_cob::object::collaboration::get::<Issue, _>(
&repo_handle,
&issue::TYPENAME,
&issue,
) {
let Ok(snapshots) =
iterate_with_snapshots::<_, _, issue::Action>(cob, issues.as_ref())
else {
continue;
};
self.process_cob::<issue::Issue>(
&issue,
&issue::TYPENAME,
snapshots,
&repo.rid,
&aliases,
)
.whatever_context("Failed processing issues")?;
}
}
Ok(())
}
pub fn process_cob<A>(
&mut self,
cob_id: &ObjectId,
typename: &TypeName,
snapshots: Vec<ObjectSnapshot<A>>,
rid: &radicle::prelude::RepoId,
aliases: &Aliases,
) -> Result<(), snafu::Whatever>
where
A: serde::Serialize + for<'de> serde::Deserialize<'de> + HasCobInfo,
{
let mut new_operations = Vec::new();
for snapshot in snapshots {
tracing::debug!("Processing operation: {}", snapshot.entry_id);
if self
.storage
.operation_exists(&snapshot.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: snapshot.entry_id.to_string(),
rid: rid.to_string(),
repo_alias,
cob_title: snapshot.state.title().to_string(),
cob_status: snapshot.state.state().to_string(),
created_at: snapshot.timestamp as i32,
actions: snapshot.actions,
author: snapshot.author.to_string(),
author_alias: aliases.alias(&snapshot.author).map(Into::into),
typename: typename.to_string(),
});
}
self.storage
.insert_timeline_entry(&TimelineEntry {
repo: rid.to_string(),
node: self.profile.id().to_string(),
cob_id: cob_id.to_string(),
typename: typename.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!("{cob_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(())
}
}