//! Example: Sync site.standard.publication and site.standard.document records.
//!
//! This example demonstrates using tapped to track publication and document
//! records from the ATProto network. It maintains an on-disk cache and
//! writes URL lists to disk on each change for inspection.
//!
//! Run with: `RUST_LOG=info cargo run --example standard_site_sync`

use std::collections::HashMap;
use std::fs;

use log::info;
use serde::{Deserialize, Serialize};
use tapped::{Event, RecordAction, RecordEvent, TapConfig, TapProcess};

/// A site.standard.publication record (subset of fields).
#[derive(Debug, Deserialize)]
struct Publication {
    url: String,
    name: String,
}

/// A site.standard.document record (subset of fields).
#[derive(Debug, Deserialize)]
struct Document {
    site: String,
    title: String,
    #[serde(default)]
    path: Option<String>,
}

/// Key for storing records in cache.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
struct RecordKey {
    did: String,
    collection: String,
    rkey: String,
}

impl RecordKey {
    fn new(event: &RecordEvent) -> Self {
        Self {
            did: event.did.clone(),
            collection: event.collection.clone(),
            rkey: event.rkey.clone(),
        }
    }
}

/// Cached record data.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
enum CachedRecord {
    Publication { url: String },
    Document { site: String, path: Option<String> },
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()?
        .block_on(async_main())
}

async fn async_main() -> Result<(), Box<dyn std::error::Error>> {
    env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info"))
        .format_timestamp_secs()
        .init();

    info!("Starting standard.site sync example");

    let config = TapConfig::builder()
        .database_url("sqlite://./tap-example.db")
        .bind(":2480")
        .signal_collection("site.standard.publication")
        .collection_filters(vec![
            "site.standard.publication".to_string(),
            "site.standard.document".to_string(),
        ])
        .disable_acks(false)
        .inherit_stdio(true)
        .build();

    info!("Spawning tap process...");

    // Spawn tap (looks for ./tap first, then tap on PATH)
    let process = TapProcess::spawn_default(config).await?;
    info!("Tap running at {}", process.url());

    let client = process.client()?;
    client.health().await?;
    info!("Tap is healthy!");

    let mut receiver = client.channel().await?;
    info!("Connected! Waiting for events...");

    // In-memory cache - load from disk if available
    let mut cache: HashMap<RecordKey, CachedRecord> = load_cache_from_disk();
    let (pub_count, doc_count) = count_cached(&cache);
    info!(
        "Loaded cache from disk: {} publications, {} documents",
        pub_count, doc_count
    );

    let mut live_count = 0u64;
    let mut backfill_count = 0u64;

    loop {
        match receiver.recv().await {
            Ok(received) => {
                if let Event::Record(ref record_event) = *received {
                    // Track live vs backfill
                    if record_event.live {
                        live_count += 1;
                    } else {
                        backfill_count += 1;
                    }

                    process_record_event(record_event, &mut cache);
                    write_output_files(&cache)?;

                    // Periodically show event source breakdown
                    if (live_count + backfill_count).is_multiple_of(100) {
                        info!(
                            "[Stats] Live events: {}, Backfill events: {}",
                            live_count, backfill_count
                        );
                    }
                } else if let Event::Identity(ref identity_event) = *received {
                    info!(
                        "[IDENTITY] {} -> {} (active: {})",
                        identity_event.did, identity_event.handle, identity_event.is_active
                    );
                }
                // Event is automatically acked when `received` is dropped here
            }
            Err(e) => {
                eprintln!("Error receiving event: {}", e);
                break;
            }
        }
    }

    Ok(())
}

fn process_record_event(event: &RecordEvent, cache: &mut HashMap<RecordKey, CachedRecord>) {
    let key = RecordKey::new(event);
    let action_str = match event.action {
        RecordAction::Create => "CREATE",
        RecordAction::Update => "UPDATE",
        RecordAction::Delete => "DELETE",
        _ => "UNKNOWN",
    };

    // Show whether this is a live or backfill event
    let source = if event.live { "LIVE" } else { "BACKFILL" };

    info!(
        "[{} {}] {} {}/{}",
        action_str, source, event.did, event.collection, event.rkey
    );

    match event.action {
        RecordAction::Create | RecordAction::Update => {
            if let Some(ref record) = event.record {
                match event.collection.as_str() {
                    "site.standard.publication" => match record.deserialize_as::<Publication>() {
                        Ok(pub_record) => {
                            info!("Publication: {} ({})", pub_record.name, pub_record.url);
                            cache.insert(
                                key,
                                CachedRecord::Publication {
                                    url: pub_record.url,
                                },
                            );
                        }
                        Err(e) => {
                            log::error!("Failed to parse publication: {}", e);
                        }
                    },
                    "site.standard.document" => match record.deserialize_as::<Document>() {
                        Ok(doc_record) => {
                            let full_url = match &doc_record.path {
                                Some(path) if !path.is_empty() => {
                                    format!(
                                        "{}/{}",
                                        doc_record.site.trim_end_matches('/'),
                                        path.trim_start_matches('/')
                                    )
                                }
                                _ => doc_record.site.clone(),
                            };
                            info!("Document: {} ({})", doc_record.title, full_url);
                            cache.insert(
                                key,
                                CachedRecord::Document {
                                    site: doc_record.site,
                                    path: doc_record.path,
                                },
                            );
                        }
                        Err(e) => {
                            log::error!("Failed to parse document: {}", e);
                        }
                    },
                    _ => {}
                }
            }
        }
        RecordAction::Delete => {
            cache.remove(&key);
        }
        _ => {}
    }

    // Log cache stats
    let (pub_count, doc_count) = count_cached(cache);
    info!(
        "Cached: {} publications, {} documents",
        pub_count, doc_count
    );
}

fn count_cached(cache: &HashMap<RecordKey, CachedRecord>) -> (usize, usize) {
    let mut publications = 0;
    let mut documents = 0;

    for record in cache.values() {
        match record {
            CachedRecord::Publication { .. } => publications += 1,
            CachedRecord::Document { .. } => documents += 1,
        }
    }

    (publications, documents)
}

fn write_output_files(cache: &HashMap<RecordKey, CachedRecord>) -> Result<(), std::io::Error> {
    let mut publication_urls: Vec<&str> = Vec::new();
    let mut document_urls: Vec<String> = Vec::new();

    for record in cache.values() {
        match record {
            CachedRecord::Publication { url } => {
                publication_urls.push(url);
            }
            CachedRecord::Document { site, path } => {
                let full_url = match path {
                    Some(p) if !p.is_empty() => {
                        format!(
                            "{}/{}",
                            site.trim_end_matches('/'),
                            p.trim_start_matches('/')
                        )
                    }
                    _ => site.clone(),
                };
                document_urls.push(full_url);
            }
        }
    }

    publication_urls.sort();
    document_urls.sort();

    fs::write("publications.txt", publication_urls.join("\n"))?;

    // Write documents.txt (publication + document URLs combined)
    let mut all_urls: Vec<String> = publication_urls.iter().map(|s| s.to_string()).collect();
    all_urls.extend(document_urls);
    all_urls.sort();
    all_urls.dedup();
    fs::write("documents.txt", all_urls.join("\n"))?;

    // Write cache.json for persistence
    let cache_entries: Vec<(&RecordKey, &CachedRecord)> = cache.iter().collect();
    let cache_json = serde_json::to_string_pretty(&cache_entries).map_err(std::io::Error::other)?;
    fs::write("cache.json", cache_json)?;

    Ok(())
}

fn load_cache_from_disk() -> HashMap<RecordKey, CachedRecord> {
    match fs::read_to_string("cache.json") {
        Ok(content) => match serde_json::from_str::<Vec<(RecordKey, CachedRecord)>>(&content) {
            Ok(entries) => entries.into_iter().collect(),
            Err(e) => {
                log::warn!("Failed to parse cache.json: {}", e);
                HashMap::new()
            }
        },
        Err(_) => HashMap::new(),
    }
}
