k-tv/crates/adapters/local-files/src/index.rs

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;

use chrono::Utc;
use tokio::sync::RwLock;
use tracing::{error, info};

use domain::MediaItemId;

use crate::config::LocalFilesConfig;
use crate::scanner::{scan_dir, LocalFileItem};

pub fn encode_id(rel_path: &str) -> MediaItemId {
    use base64::Engine as _;
    MediaItemId::new(
        base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(rel_path.as_bytes()),
    )
}

pub fn decode_id(id: &MediaItemId) -> Option<String> {
    use base64::Engine as _;
    let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
        .decode(id.as_ref())
        .ok()?;
    String::from_utf8(bytes).ok()
}

pub struct LocalIndex {
    items: Arc<RwLock<HashMap<MediaItemId, LocalFileItem>>>,
    pub root_dir: PathBuf,
    provider_id: String,
    pool: sqlx::SqlitePool,
}

impl LocalIndex {
    pub async fn new(
        config: &LocalFilesConfig,
        pool: sqlx::SqlitePool,
        provider_id: String,
    ) -> Self {
        let idx = Self {
            items: Arc::new(RwLock::new(HashMap::new())),
            root_dir: config.root_dir.clone(),
            provider_id,
            pool,
        };
        idx.load_from_db().await;
        idx
    }

    async fn load_from_db(&self) {
        #[derive(sqlx::FromRow)]
        struct Row {
            id: String,
            rel_path: String,
            title: String,
            duration_secs: i64,
            year: Option<i64>,
            tags: String,
            top_dir: String,
        }

        let rows = sqlx::query_as::<_, Row>(
            "SELECT id, rel_path, title, duration_secs, year, tags, top_dir \
             FROM local_files_index WHERE provider_id = ?",
        )
        .bind(&self.provider_id)
        .fetch_all(&self.pool)
        .await;

        match rows {
            Ok(rows) => {
                let mut map = self.items.write().await;
                for row in rows {
                    let tags: Vec<String> =
                        serde_json::from_str(&row.tags).unwrap_or_default();
                    let item = LocalFileItem {
                        rel_path: row.rel_path,
                        title: row.title,
                        duration_secs: row.duration_secs as u32,
                        year: row.year.map(|y| y as u16),
                        tags,
                        top_dir: row.top_dir,
                    };
                    map.insert(MediaItemId::new(row.id), item);
                }
                info!(
                    "Local files index [{}]: loaded {} items from DB",
                    self.provider_id,
                    map.len()
                );
            }
            Err(e) => {
                // Table might not exist yet on first run -- that's fine.
                tracing::debug!("Could not load local files index from DB: {}", e);
            }
        }
    }

    pub async fn rescan(&self) -> u32 {
        info!(
            "Local files [{}]: scanning {:?}",
            self.provider_id, self.root_dir
        );
        let new_items = scan_dir(&self.root_dir).await;
        let count = new_items.len() as u32;

        {
            let mut map = self.items.write().await;
            map.clear();
            for item in &new_items {
                let id = encode_id(&item.rel_path);
                map.insert(id, item.clone());
            }
        }

        if let Err(e) = self.save_to_db(&new_items).await {
            error!("Failed to persist local files index: {}", e);
        }

        info!(
            "Local files [{}]: indexed {} items",
            self.provider_id, count
        );
        count
    }

    async fn save_to_db(&self, items: &[LocalFileItem]) -> Result<(), sqlx::Error> {
        let mut tx = self.pool.begin().await?;

        sqlx::query("DELETE FROM local_files_index WHERE provider_id = ?")
            .bind(&self.provider_id)
            .execute(&mut *tx)
            .await?;

        let now = Utc::now().to_rfc3339();
        for item in items {
            let id = encode_id(&item.rel_path).into_inner();
            let tags_json =
                serde_json::to_string(&item.tags).unwrap_or_else(|_| "[]".into());
            sqlx::query(
                "INSERT INTO local_files_index \
                 (id, rel_path, title, duration_secs, year, tags, top_dir, scanned_at, provider_id) \
                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
            )
            .bind(&id)
            .bind(&item.rel_path)
            .bind(&item.title)
            .bind(item.duration_secs as i64)
            .bind(item.year.map(|y| y as i64))
            .bind(&tags_json)
            .bind(&item.top_dir)
            .bind(&now)
            .bind(&self.provider_id)
            .execute(&mut *tx)
            .await?;
        }

        tx.commit().await
    }

    pub async fn get(&self, id: &MediaItemId) -> Option<LocalFileItem> {
        self.items.read().await.get(id).cloned()
    }

    pub async fn get_all(&self) -> Vec<(MediaItemId, LocalFileItem)> {
        self.items
            .read()
            .await
            .iter()
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect()
    }

    pub async fn collections(&self) -> Vec<String> {
        let map = self.items.read().await;
        let mut seen = std::collections::HashSet::new();
        for item in map.values() {
            seen.insert(item.top_dir.clone());
        }
        let mut dirs: Vec<String> = seen.into_iter().collect();
        dirs.sort();
        dirs
    }
}