radicle-feed 0.1.1

use log::debug;
use serde::{Deserialize, Serialize};

use radicle::cob::object::Storage;
use radicle::cob::stream::{CobRange, CobStream, Stream};
use radicle::cob::TypeName;
use radicle::profile::{Aliases, Profile};
use radicle::storage::git::paths;
use radicle::storage::ReadStorage;
use radicle::{issue, patch};

use crate::entry::{ActionEntry, Author};
use crate::storage::FeedStorage;

/// Main feed processor that processes repositories and stores actions using a generic storage backend
pub struct FeedProcessor<S: FeedStorage> {
    storage: S,
    profile: Profile,
}

impl<S: FeedStorage> FeedProcessor<S> {
    /// Create a new feed processor with the given storage backend
    pub fn new(mut storage: S, profile: Profile) -> Result<Self, S::Error> {
        storage.initialize()?;

        Ok(Self { storage, profile })
    }

    /// Process all delegate repositories and store new actions
    pub fn process_repositories(&mut self) -> Result<ProcessingStats, Box<dyn std::error::Error>> {
        let aliases = self.profile.aliases();
        let profile_storage = self.profile.storage.clone();

        let repos = profile_storage
            .repositories()?
            .into_iter()
            .collect::<Vec<_>>();

        let mut stats = ProcessingStats::default();
        let mut new_operations = Vec::new();

        for repo in repos {
            debug!("Processing repository: {}", repo.rid);
            stats.repositories_processed += 1;

            let path = paths::repository(&profile_storage, &repo.rid);
            let repo_handle = radicle::storage::git::Repository::open(path.clone(), repo.rid)?;
            let git2_repo = radicle::git::raw::Repository::open(path)?;

            // Process patches
            let patches = radicle::patch::Patches::open(&repo_handle)?;
            let patch_stats = self.process_cobs::<patch::Action, _>(
                patches.as_ref(),
                &patch::TYPENAME,
                &git2_repo,
                &repo.rid,
                &aliases,
                &mut new_operations,
            )?;
            stats.operations_found += patch_stats.operations_found;

            // Process issues
            let issues = radicle::issue::Issues::open(&repo_handle)?;
            let issue_stats = self.process_cobs::<issue::Action, _>(
                issues.as_ref(),
                &issue::TYPENAME,
                &git2_repo,
                &repo.rid,
                &aliases,
                &mut new_operations,
            )?;
            stats.operations_found += issue_stats.operations_found;
        }

        // Insert new actions into storage
        if !new_operations.is_empty() {
            debug!(
                "Inserting {} new operations into storage",
                new_operations.len()
            );
            self.storage.insert_batch(&new_operations)?;
            stats.operations_stored = new_operations.len();
        } else {
            debug!("No new operations found");
        }

        Ok(stats)
    }

    pub fn process_cobs<A, Store>(
        &mut self,
        store: &Store,
        typename: &TypeName,
        git2_repo: &radicle::git::raw::Repository,
        rid: &radicle::prelude::RepoId,
        aliases: &Aliases,
        new_actions: &mut Vec<ActionEntry>,
    ) -> Result<ProcessingStats, Box<dyn std::error::Error>>
    where
        A: Serialize + for<'de> Deserialize<'de>,
        Store: Storage,
    {
        let mut stats = ProcessingStats::default();
        let cob_ids = store.types(typename)?;

        for cob_id in cob_ids.into_keys() {
            debug!("Processing {}: {}", typename, cob_id);

            // Get the last processed operation ID for this COB
            let last_operation_id = self
                .storage
                .get_last_processed_operation(rid, &cob_id, typename)?;

            if let Some(ref since_id) = last_operation_id {
                debug!("Last processed operation ID: {}", since_id);
            } else {
                debug!("No previous operations found (first time processing)");
            }

            // Create stream for this COB
            let stream = Stream::<A>::new(
                git2_repo,
                CobRange::new(typename, &cob_id),
                typename.clone(),
            );

            // Get actions - use since() if we have a last processed ID, otherwise get all
            let stream_entries = if let Some(since_id) = last_operation_id {
                let since_entries: Vec<_> = stream
                    .since(since_id)?
                    .filter_map(|s| s.ok())
                    .filter(|entry| entry.id() != since_id) // Filter out the already processed action
                    .collect();
                debug!(
                    "Found {} new operations since last processed",
                    since_entries.len(),
                );
                since_entries
            } else {
                let all_entries: Vec<_> = stream.all()?.filter_map(|s| s.ok()).collect();
                debug!("Found {} total operations", all_entries.len());
                all_entries
            };

            if stream_entries.is_empty() {
                debug!("No new operations to process");
                continue;
            }

            let mut last_processed_id = last_operation_id;
            let mut operations_added = 0;

            for stream_entry in stream_entries {
                debug!("Processing operation: {}", stream_entry.id());

                // Check if this operation is already in storage to avoid duplicates
                if self.storage.operation_exists(&stream_entry.id())? {
                    debug!("Operation already exists in storage, skipping");
                    continue;
                }

                for action in &stream_entry.actions {
                    new_actions.push(ActionEntry {
                        operation_id: stream_entry.id(),
                        cob_id,
                        rid: *rid,
                        timestamp: stream_entry.timestamp,
                        action: serde_json::to_string(action)?,
                        author: Author::new(&stream_entry.author.into(), aliases),
                        typename: typename.clone(),
                    });
                    operations_added += 1;
                    stats.operations_found += 1;
                }

                // Update the last processed ID
                last_processed_id = Some(stream_entry.id());
            }

            debug!("Added {} new operations", operations_added);

            // Update the tracking table with the latest operation ID only if we processed something
            if let Some(latest_id) = last_processed_id {
                self.storage
                    .update_last_processed_operation(rid, &cob_id, typename, &latest_id)?;
                debug!("Updated last processed ID to: {}", latest_id);
            }
        }

        Ok(stats)
    }

    /// Get a reference to the underlying storage
    pub fn storage(&self) -> &S {
        &self.storage
    }
}

/// Statistics about the processing operation
#[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(())
    }
}

/// Convenience function to process repositories with in-memory storage
pub fn process_with_memory(
) -> Result<(ProcessingStats, Vec<ActionEntry>), Box<dyn std::error::Error>> {
    use crate::storage::MemoryStorage;

    let profile = Profile::load()?;
    let storage = MemoryStorage::new();
    let mut processor = FeedProcessor::new(storage, profile)?;
    let stats = processor.process_repositories()?;
    let actions = processor.storage().load_all_sorted()?;
    Ok((stats, actions))
}