ceres-core 0.4.0

//! Sync service layer for portal synchronization logic.
//!
//! This module provides pure business logic for delta detection and sync statistics,
//! decoupled from I/O operations and CLI orchestration.

use std::sync::atomic::{AtomicUsize, Ordering};

/// Outcome of processing a single dataset during sync.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SyncOutcome {
    /// Dataset content hash matches existing - no changes needed
    Unchanged,
    /// Dataset content changed - embedding regenerated
    Updated,
    /// New dataset - first time seeing this dataset
    Created,
    /// Processing failed for this dataset
    Failed,
    /// Dataset skipped due to circuit breaker being open
    Skipped,
}

/// Statistics for a portal sync operation.
#[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize)]
pub struct SyncStats {
    pub unchanged: usize,
    pub updated: usize,
    pub created: usize,
    pub failed: usize,
    /// Number of datasets skipped due to circuit breaker being open.
    #[serde(default)]
    pub skipped: usize,
}

impl SyncStats {
    /// Creates a new empty stats tracker.
    pub fn new() -> Self {
        Self::default()
    }

    /// Records an outcome, incrementing the appropriate counter.
    pub fn record(&mut self, outcome: SyncOutcome) {
        match outcome {
            SyncOutcome::Unchanged => self.unchanged += 1,
            SyncOutcome::Updated => self.updated += 1,
            SyncOutcome::Created => self.created += 1,
            SyncOutcome::Failed => self.failed += 1,
            SyncOutcome::Skipped => self.skipped += 1,
        }
    }

    /// Returns the total number of processed datasets.
    pub fn total(&self) -> usize {
        self.unchanged + self.updated + self.created + self.failed + self.skipped
    }

    /// Returns the number of successfully processed datasets.
    pub fn successful(&self) -> usize {
        self.unchanged + self.updated + self.created
    }
}

// =============================================================================
// Sync Status and Result Types (for cancellation support)
// =============================================================================

/// Overall status of a sync operation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SyncStatus {
    /// Sync completed successfully with all datasets processed.
    Completed,
    /// Sync was cancelled but partial progress was saved.
    Cancelled,
}

impl SyncStatus {
    /// Returns the string representation for database storage.
    pub fn as_str(&self) -> &'static str {
        match self {
            SyncStatus::Completed => "completed",
            SyncStatus::Cancelled => "cancelled",
        }
    }

    /// Returns true if the sync was completed successfully.
    pub fn is_completed(&self) -> bool {
        matches!(self, SyncStatus::Completed)
    }

    /// Returns true if the sync was cancelled.
    pub fn is_cancelled(&self) -> bool {
        matches!(self, SyncStatus::Cancelled)
    }
}

/// Result of a sync operation including status and statistics.
#[derive(Debug, Clone)]
pub struct SyncResult {
    /// The final status of the sync operation.
    pub status: SyncStatus,
    /// Statistics about processed datasets.
    pub stats: SyncStats,
    /// Optional message providing context (e.g., cancellation reason).
    pub message: Option<String>,
}

impl SyncResult {
    /// Creates a completed sync result.
    pub fn completed(stats: SyncStats) -> Self {
        Self {
            status: SyncStatus::Completed,
            stats,
            message: None,
        }
    }

    /// Creates a cancelled sync result with partial statistics.
    pub fn cancelled(stats: SyncStats) -> Self {
        Self {
            status: SyncStatus::Cancelled,
            stats,
            message: Some("Operation cancelled - partial progress saved".to_string()),
        }
    }

    /// Returns true if the sync was fully successful.
    pub fn is_completed(&self) -> bool {
        self.status.is_completed()
    }

    /// Returns true if the sync was cancelled.
    pub fn is_cancelled(&self) -> bool {
        self.status.is_cancelled()
    }
}

/// Thread-safe wrapper for [`SyncStats`] using atomic counters.
///
/// This is useful for concurrent harvesting where multiple tasks
/// update statistics simultaneously without requiring a mutex.
///
/// # Example
///
/// ```
/// use ceres_core::sync::{AtomicSyncStats, SyncOutcome};
///
/// let stats = AtomicSyncStats::new();
///
/// // Can be safely called from multiple threads
/// stats.record(SyncOutcome::Created);
/// stats.record(SyncOutcome::Unchanged);
///
/// let snapshot = stats.to_stats();
/// assert_eq!(snapshot.created, 1);
/// assert_eq!(snapshot.unchanged, 1);
/// ```
pub struct AtomicSyncStats {
    unchanged: AtomicUsize,
    updated: AtomicUsize,
    created: AtomicUsize,
    failed: AtomicUsize,
    skipped: AtomicUsize,
}

impl AtomicSyncStats {
    /// Creates a new zeroed stats tracker.
    pub fn new() -> Self {
        Self {
            unchanged: AtomicUsize::new(0),
            updated: AtomicUsize::new(0),
            created: AtomicUsize::new(0),
            failed: AtomicUsize::new(0),
            skipped: AtomicUsize::new(0),
        }
    }

    /// Records an outcome, incrementing the appropriate counter atomically.
    pub fn record(&self, outcome: SyncOutcome) {
        match outcome {
            SyncOutcome::Unchanged => self.unchanged.fetch_add(1, Ordering::Relaxed),
            SyncOutcome::Updated => self.updated.fetch_add(1, Ordering::Relaxed),
            SyncOutcome::Created => self.created.fetch_add(1, Ordering::Relaxed),
            SyncOutcome::Failed => self.failed.fetch_add(1, Ordering::Relaxed),
            SyncOutcome::Skipped => self.skipped.fetch_add(1, Ordering::Relaxed),
        };
    }

    /// Converts atomic counters to a snapshot [`SyncStats`].
    pub fn to_stats(&self) -> SyncStats {
        SyncStats {
            unchanged: self.unchanged.load(Ordering::Relaxed),
            updated: self.updated.load(Ordering::Relaxed),
            created: self.created.load(Ordering::Relaxed),
            failed: self.failed.load(Ordering::Relaxed),
            skipped: self.skipped.load(Ordering::Relaxed),
        }
    }
}

impl Default for AtomicSyncStats {
    fn default() -> Self {
        Self::new()
    }
}

/// Result of delta detection for a dataset.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReprocessingDecision {
    /// The outcome classification for this dataset
    pub outcome: SyncOutcome,
    /// Human-readable reason for the decision
    pub reason: &'static str,
}

impl ReprocessingDecision {
    /// Returns true if this is a legacy record update (existing record without hash).
    pub fn is_legacy(&self) -> bool {
        self.reason == "legacy record without hash"
    }
}

/// Trait for delta detection strategies.
///
/// Implementations determine whether a dataset needs reprocessing based on
/// its existing and new content hashes. This enables different strategies
/// such as content-hash comparison, always-reprocess, or timestamp-based detection.
pub trait DeltaDetector: Send + Sync + Clone {
    /// Determines if a dataset needs reprocessing.
    ///
    /// Implementations must only return `Created`, `Updated`, or `Unchanged` outcomes.
    /// `Failed` and `Skipped` are reserved for runtime errors in the harvest pipeline.
    ///
    /// # Arguments
    /// * `existing_hash` - The stored content hash lookup result:
    ///   - `None` — dataset not found in the store (new dataset)
    ///   - `Some(None)` — dataset exists but has no stored hash (legacy record)
    ///   - `Some(Some(hash))` — dataset exists with the given content hash
    /// * `new_hash` - The computed content hash from the portal data
    fn needs_reprocessing(
        &self,
        existing_hash: Option<&Option<String>>,
        new_hash: &str,
    ) -> ReprocessingDecision;
}

/// Default delta detector using content hash comparison.
///
/// Compares the stored content hash with the newly computed hash to determine
/// if a dataset's embedding needs to be regenerated.
#[derive(Debug, Clone, Default)]
pub struct ContentHashDetector;

impl DeltaDetector for ContentHashDetector {
    fn needs_reprocessing(
        &self,
        existing_hash: Option<&Option<String>>,
        new_hash: &str,
    ) -> ReprocessingDecision {
        match existing_hash {
            Some(Some(hash)) if hash == new_hash => {
                // Hash matches - content unchanged
                ReprocessingDecision {
                    outcome: SyncOutcome::Unchanged,
                    reason: "content hash matches",
                }
            }
            Some(Some(_)) => {
                // Hash exists but differs - content updated
                ReprocessingDecision {
                    outcome: SyncOutcome::Updated,
                    reason: "content hash changed",
                }
            }
            Some(None) => {
                // Exists but no hash (legacy data) - treat as update
                ReprocessingDecision {
                    outcome: SyncOutcome::Updated,
                    reason: "legacy record without hash",
                }
            }
            None => {
                // Not in existing data - new dataset
                ReprocessingDecision {
                    outcome: SyncOutcome::Created,
                    reason: "new dataset",
                }
            }
        }
    }
}

/// Delta detector that always triggers reprocessing.
///
/// Useful for full rebuilds where all embeddings should be regenerated
/// regardless of content changes.
#[derive(Debug, Clone, Default)]
pub struct AlwaysReprocessDetector;

impl DeltaDetector for AlwaysReprocessDetector {
    fn needs_reprocessing(
        &self,
        existing_hash: Option<&Option<String>>,
        _new_hash: &str,
    ) -> ReprocessingDecision {
        match existing_hash {
            Some(_) => ReprocessingDecision {
                outcome: SyncOutcome::Updated,
                reason: "always reprocess strategy",
            },
            None => ReprocessingDecision {
                outcome: SyncOutcome::Created,
                reason: "new dataset",
            },
        }
    }
}

/// Determines if a dataset needs reprocessing based on content hash comparison.
///
/// This is a convenience wrapper around [`ContentHashDetector`].
///
/// # Arguments
/// * `existing_hash` - The stored content hash for this dataset (None if new dataset)
/// * `new_hash` - The computed content hash from the portal data
///
/// # Returns
/// A `ReprocessingDecision` indicating whether embedding regeneration is needed
/// and the classification of this sync operation.
pub fn needs_reprocessing(
    existing_hash: Option<&Option<String>>,
    new_hash: &str,
) -> ReprocessingDecision {
    ContentHashDetector.needs_reprocessing(existing_hash, new_hash)
}

// =============================================================================
// Batch Harvest Types
// =============================================================================

/// Result of harvesting a single portal in batch mode.
#[derive(Debug, Clone)]
pub struct PortalHarvestResult {
    /// Portal name identifier.
    pub portal_name: String,
    /// Portal URL.
    pub portal_url: String,
    /// Sync statistics for this portal.
    pub stats: SyncStats,
    /// Error message if harvest failed, None if successful.
    pub error: Option<String>,
}

impl PortalHarvestResult {
    /// Creates a successful harvest result.
    pub fn success(name: String, url: String, stats: SyncStats) -> Self {
        Self {
            portal_name: name,
            portal_url: url,
            stats,
            error: None,
        }
    }

    /// Creates a failed harvest result.
    pub fn failure(name: String, url: String, error: String) -> Self {
        Self {
            portal_name: name,
            portal_url: url,
            stats: SyncStats::default(),
            error: Some(error),
        }
    }

    /// Returns true if the harvest was successful.
    pub fn is_success(&self) -> bool {
        self.error.is_none()
    }
}

/// Aggregated results from batch harvesting multiple portals.
#[derive(Debug, Clone, Default)]
pub struct BatchHarvestSummary {
    /// Results for each portal.
    pub results: Vec<PortalHarvestResult>,
}

impl BatchHarvestSummary {
    /// Creates a new empty summary.
    pub fn new() -> Self {
        Self::default()
    }

    /// Adds a portal harvest result.
    pub fn add(&mut self, result: PortalHarvestResult) {
        self.results.push(result);
    }

    /// Returns the count of successful harvests.
    pub fn successful_count(&self) -> usize {
        self.results.iter().filter(|r| r.is_success()).count()
    }

    /// Returns the count of failed harvests.
    pub fn failed_count(&self) -> usize {
        self.results.iter().filter(|r| !r.is_success()).count()
    }

    /// Returns the total number of datasets across all successful portals.
    pub fn total_datasets(&self) -> usize {
        self.results.iter().map(|r| r.stats.total()).sum()
    }

    /// Returns the total number of portals processed.
    pub fn total_portals(&self) -> usize {
        self.results.len()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_sync_stats_default() {
        let stats = SyncStats::new();
        assert_eq!(stats.unchanged, 0);
        assert_eq!(stats.updated, 0);
        assert_eq!(stats.created, 0);
        assert_eq!(stats.failed, 0);
        assert_eq!(stats.skipped, 0);
    }

    #[test]
    fn test_sync_stats_record() {
        let mut stats = SyncStats::new();
        stats.record(SyncOutcome::Unchanged);
        stats.record(SyncOutcome::Updated);
        stats.record(SyncOutcome::Created);
        stats.record(SyncOutcome::Failed);
        stats.record(SyncOutcome::Skipped);

        assert_eq!(stats.unchanged, 1);
        assert_eq!(stats.updated, 1);
        assert_eq!(stats.created, 1);
        assert_eq!(stats.failed, 1);
        assert_eq!(stats.skipped, 1);
    }

    #[test]
    fn test_sync_stats_total() {
        let mut stats = SyncStats::new();
        stats.unchanged = 10;
        stats.updated = 5;
        stats.created = 3;
        stats.failed = 2;
        stats.skipped = 1;

        assert_eq!(stats.total(), 21);
    }

    #[test]
    fn test_sync_stats_successful() {
        let mut stats = SyncStats::new();
        stats.unchanged = 10;
        stats.updated = 5;
        stats.created = 3;
        stats.failed = 2;
        stats.skipped = 1;

        // successful does not include failed or skipped
        assert_eq!(stats.successful(), 18);
    }

    #[test]
    fn test_needs_reprocessing_unchanged() {
        let hash = "abc123".to_string();
        let existing = Some(Some(hash.clone()));
        let decision = needs_reprocessing(existing.as_ref(), &hash);

        assert_eq!(decision.outcome, SyncOutcome::Unchanged);
        assert_eq!(decision.reason, "content hash matches");
    }

    #[test]
    fn test_needs_reprocessing_updated() {
        let old_hash = "abc123".to_string();
        let new_hash = "def456";
        let existing = Some(Some(old_hash));
        let decision = needs_reprocessing(existing.as_ref(), new_hash);

        assert_eq!(decision.outcome, SyncOutcome::Updated);
        assert_eq!(decision.reason, "content hash changed");
    }

    #[test]
    fn test_needs_reprocessing_legacy() {
        let existing: Option<Option<String>> = Some(None);
        let decision = needs_reprocessing(existing.as_ref(), "new_hash");

        assert_eq!(decision.outcome, SyncOutcome::Updated);
        assert_eq!(decision.reason, "legacy record without hash");
    }

    #[test]
    fn test_needs_reprocessing_new() {
        let decision = needs_reprocessing(None, "new_hash");

        assert_eq!(decision.outcome, SyncOutcome::Created);
        assert_eq!(decision.reason, "new dataset");
    }

    #[test]
    fn test_is_legacy_true() {
        let existing: Option<Option<String>> = Some(None);
        let decision = needs_reprocessing(existing.as_ref(), "new_hash");

        assert!(decision.is_legacy());
    }

    #[test]
    fn test_is_legacy_false() {
        let decision = needs_reprocessing(None, "new_hash");
        assert!(!decision.is_legacy());

        let hash = "abc123".to_string();
        let existing = Some(Some(hash.clone()));
        let decision = needs_reprocessing(existing.as_ref(), &hash);
        assert!(!decision.is_legacy());
    }

    // =========================================================================
    // PortalHarvestResult tests
    // =========================================================================

    #[test]
    fn test_portal_harvest_result_success() {
        let stats = SyncStats {
            unchanged: 5,
            updated: 3,
            created: 2,
            failed: 0,
            skipped: 0,
        };
        let result = PortalHarvestResult::success(
            "test".to_string(),
            "https://example.com".to_string(),
            stats,
        );
        assert!(result.is_success());
        assert!(result.error.is_none());
        assert_eq!(result.stats.total(), 10);
        assert_eq!(result.portal_name, "test");
        assert_eq!(result.portal_url, "https://example.com");
    }

    #[test]
    fn test_portal_harvest_result_failure() {
        let result = PortalHarvestResult::failure(
            "test".to_string(),
            "https://example.com".to_string(),
            "Connection timeout".to_string(),
        );
        assert!(!result.is_success());
        assert_eq!(result.error, Some("Connection timeout".to_string()));
        assert_eq!(result.stats.total(), 0);
    }

    // =========================================================================
    // BatchHarvestSummary tests
    // =========================================================================

    #[test]
    fn test_batch_harvest_summary_empty() {
        let summary = BatchHarvestSummary::new();
        assert_eq!(summary.successful_count(), 0);
        assert_eq!(summary.failed_count(), 0);
        assert_eq!(summary.total_datasets(), 0);
        assert_eq!(summary.total_portals(), 0);
    }

    #[test]
    fn test_batch_harvest_summary_mixed_results() {
        let mut summary = BatchHarvestSummary::new();

        let stats1 = SyncStats {
            unchanged: 10,
            updated: 5,
            created: 3,
            failed: 2,
            skipped: 0,
        };
        summary.add(PortalHarvestResult::success(
            "a".into(),
            "https://a.com".into(),
            stats1,
        ));

        summary.add(PortalHarvestResult::failure(
            "b".into(),
            "https://b.com".into(),
            "error".into(),
        ));

        let stats2 = SyncStats {
            unchanged: 20,
            updated: 0,
            created: 0,
            failed: 0,
            skipped: 0,
        };
        summary.add(PortalHarvestResult::success(
            "c".into(),
            "https://c.com".into(),
            stats2,
        ));

        assert_eq!(summary.total_portals(), 3);
        assert_eq!(summary.successful_count(), 2);
        assert_eq!(summary.failed_count(), 1);
        assert_eq!(summary.total_datasets(), 40); // 20 + 20 + 0 (failed portal has 0)
    }

    #[test]
    fn test_batch_harvest_summary_all_successful() {
        let mut summary = BatchHarvestSummary::new();

        let stats = SyncStats {
            unchanged: 5,
            updated: 0,
            created: 5,
            failed: 0,
            skipped: 0,
        };
        summary.add(PortalHarvestResult::success(
            "portal1".into(),
            "https://portal1.com".into(),
            stats,
        ));

        assert_eq!(summary.successful_count(), 1);
        assert_eq!(summary.failed_count(), 0);
        assert_eq!(summary.total_datasets(), 10);
    }

    #[test]
    fn test_batch_harvest_summary_all_failed() {
        let mut summary = BatchHarvestSummary::new();

        summary.add(PortalHarvestResult::failure(
            "portal1".into(),
            "https://portal1.com".into(),
            "error1".into(),
        ));
        summary.add(PortalHarvestResult::failure(
            "portal2".into(),
            "https://portal2.com".into(),
            "error2".into(),
        ));

        assert_eq!(summary.successful_count(), 0);
        assert_eq!(summary.failed_count(), 2);
        assert_eq!(summary.total_datasets(), 0);
        assert_eq!(summary.total_portals(), 2);
    }

    // =========================================================================
    // AtomicSyncStats tests
    // =========================================================================

    #[test]
    fn test_atomic_sync_stats_new() {
        let stats = AtomicSyncStats::new();
        let result = stats.to_stats();
        assert_eq!(result.unchanged, 0);
        assert_eq!(result.updated, 0);
        assert_eq!(result.created, 0);
        assert_eq!(result.failed, 0);
        assert_eq!(result.skipped, 0);
    }

    #[test]
    fn test_atomic_sync_stats_record() {
        let stats = AtomicSyncStats::new();
        stats.record(SyncOutcome::Unchanged);
        stats.record(SyncOutcome::Updated);
        stats.record(SyncOutcome::Created);
        stats.record(SyncOutcome::Failed);
        stats.record(SyncOutcome::Skipped);

        let result = stats.to_stats();
        assert_eq!(result.unchanged, 1);
        assert_eq!(result.updated, 1);
        assert_eq!(result.created, 1);
        assert_eq!(result.failed, 1);
        assert_eq!(result.skipped, 1);
    }

    #[test]
    fn test_atomic_sync_stats_multiple_records() {
        let stats = AtomicSyncStats::new();
        for _ in 0..10 {
            stats.record(SyncOutcome::Unchanged);
        }
        for _ in 0..5 {
            stats.record(SyncOutcome::Updated);
        }

        let result = stats.to_stats();
        assert_eq!(result.unchanged, 10);
        assert_eq!(result.updated, 5);
        assert_eq!(result.total(), 15);
        assert_eq!(result.successful(), 15);
    }

    #[test]
    fn test_atomic_sync_stats_default() {
        let stats = AtomicSyncStats::default();
        let result = stats.to_stats();
        assert_eq!(result.total(), 0);
    }

    // =========================================================================
    // SyncStatus tests
    // =========================================================================

    #[test]
    fn test_sync_status_completed() {
        let status = SyncStatus::Completed;
        assert_eq!(status.as_str(), "completed");
        assert!(status.is_completed());
        assert!(!status.is_cancelled());
    }

    #[test]
    fn test_sync_status_cancelled() {
        let status = SyncStatus::Cancelled;
        assert_eq!(status.as_str(), "cancelled");
        assert!(!status.is_completed());
        assert!(status.is_cancelled());
    }

    // =========================================================================
    // SyncResult tests
    // =========================================================================

    #[test]
    fn test_sync_result_completed() {
        let stats = SyncStats {
            unchanged: 10,
            updated: 5,
            created: 3,
            failed: 0,
            skipped: 0,
        };
        let result = SyncResult::completed(stats);
        assert!(result.is_completed());
        assert!(!result.is_cancelled());
        assert!(result.message.is_none());
        assert_eq!(result.stats.total(), 18);
    }

    #[test]
    fn test_sync_result_cancelled() {
        let stats = SyncStats {
            unchanged: 5,
            updated: 2,
            created: 1,
            failed: 0,
            skipped: 0,
        };
        let result = SyncResult::cancelled(stats);
        assert!(!result.is_completed());
        assert!(result.is_cancelled());
        assert!(result.message.is_some());
        assert!(result.message.unwrap().contains("cancelled"));
        assert_eq!(result.stats.total(), 8);
    }

    // =========================================================================
    // DeltaDetector trait tests
    // =========================================================================

    #[test]
    fn test_content_hash_detector_unchanged() {
        let detector = ContentHashDetector;
        let hash = "abc123".to_string();
        let existing = Some(Some(hash.clone()));
        let decision = detector.needs_reprocessing(existing.as_ref(), &hash);

        assert_eq!(decision.outcome, SyncOutcome::Unchanged);
    }

    #[test]
    fn test_content_hash_detector_updated() {
        let detector = ContentHashDetector;
        let existing = Some(Some("old_hash".to_string()));
        let decision = detector.needs_reprocessing(existing.as_ref(), "new_hash");

        assert_eq!(decision.outcome, SyncOutcome::Updated);
    }

    #[test]
    fn test_content_hash_detector_new() {
        let detector = ContentHashDetector;
        let decision = detector.needs_reprocessing(None, "new_hash");

        assert_eq!(decision.outcome, SyncOutcome::Created);
    }

    #[test]
    fn test_always_reprocess_detector_existing() {
        let detector = AlwaysReprocessDetector;
        let hash = "abc123".to_string();
        let existing = Some(Some(hash.clone()));
        let decision = detector.needs_reprocessing(existing.as_ref(), &hash);

        assert_eq!(decision.outcome, SyncOutcome::Updated);
        assert_eq!(decision.reason, "always reprocess strategy");
    }

    #[test]
    fn test_always_reprocess_detector_new() {
        let detector = AlwaysReprocessDetector;
        let decision = detector.needs_reprocessing(None, "new_hash");

        assert_eq!(decision.outcome, SyncOutcome::Created);
        assert_eq!(decision.reason, "new dataset");
    }

    #[test]
    fn test_always_reprocess_detector_legacy() {
        let detector = AlwaysReprocessDetector;
        let existing: Option<Option<String>> = Some(None);
        let decision = detector.needs_reprocessing(existing.as_ref(), "new_hash");

        assert_eq!(decision.outcome, SyncOutcome::Updated);
        assert_eq!(decision.reason, "always reprocess strategy");
    }
}