//! TRACES: FR-CAT-15 | NFR-P9
//! Soft delete, restore, and permanent delete against the remote.
//!
//! `dr_catalog::trash` owns the catalog side and does no I/O. This is the other
//! half: the remote `MOVE`/`DELETE`, run on a worker thread, paired with the
//! catalog record in the one order that is safe.
//!
//! # Ordering, which is the whole of the correctness here
//!
//! **Soft delete** — `MOVE` first, record second. The reverse would leave the
//! catalog claiming a file is trashed while it sits in the library; the scan
//! excludes the trash folder, so nothing would ever correct the row.
//!
//! **Restore** — `MOVE` first, record second, for the same reason mirrored.
//!
//! **Purge** — `DELETE` the file, then forget the row, then forget the
//! thumbnail. A crash between steps leaves a trashed row whose file is gone,
//! which the next empty resolves as already-deleted. The other order loses the
//! file silently: no row, no listing, and a scan that will never look in the
//! trash folder — a photograph consuming quota that nothing can find.
//!
//! # Partial failure is normal, not exceptional
//!
//! Forty files is forty requests, and one can fail on permissions while the
//! rest succeed. Every operation here is therefore per-image and reports what
//! actually happened rather than aborting the batch — a trash that gives up
//! halfway with no record of where it stopped is worse than one that reports
//! "38 of 40".

use std::path::PathBuf;
use std::sync::mpsc::Receiver;

use dr_catalog::{trash, Catalog};
use dr_sync::{Connection, RemoteError, RemoteId, RemotePath};

use dr_types::ImageId;

/// What a trash operation reports back to the UI.
#[derive(Debug)]
pub enum TrashMessage {
    /// One image finished, successfully or not.
    ///
    /// Per-image rather than per-batch so the UI can show progress on a large
    /// selection, and so a failure names the file it happened to.
    Progress {
        done: usize,
        total: usize,
        failed: usize,
    },
    /// The batch finished. `failed` names what did not work, for the status line.
    Done { moved: usize, failed: Vec<String> },
}

/// Which way an image is being moved.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Direction {
    /// Library → trash folder.
    ToTrash,
    /// Trash folder → where it came from.
    Restore,
}

/// One image to move, resolved before the worker starts.
///
/// Carries the paths the `MOVE` runs between, so the worker needs no catalog
/// access to do its half — the catalog is not `Send`, and the worker owns a
/// separate connection only for the write-back.
#[derive(Debug, Clone)]
pub struct Move {
    pub image_id: ImageId,
    /// `oc:fileid` where the catalog knows one — the identity `MOVE` preserves,
    /// which is what keeps the thumbnail and sidecar mapping attached across a
    /// trash and restore.
    ///
    /// Not how the file is addressed: the `MOVE` goes from `from` to `to`,
    /// because WebDAV exposes no fileid-addressable endpoint and the backend
    /// rejects a bare `RemoteId::Stable`. Carried so the plan records the
    /// identity it expects to survive.
    pub file_id: Option<u64>,
    pub from: String,
    pub to: String,
}

/// Plan a soft delete: where each image goes in the trash.
///
/// Reads the catalog, so it runs on the UI thread before the worker starts.
/// Skips images already trashed — re-trashing is a no-op, not an error, and the
/// UI can hand over a selection that overlaps the trash.
pub fn plan_trash(
    catalog: &Catalog,
    root: &str,
    images: &[ImageId],
) -> Result<Vec<Move>, dr_catalog::CatalogError> {
    let mut out = Vec::new();
    for &image in images {
        let row: Option<(String, Option<i64>, Option<i64>)> = catalog
            .connection()
            .query_row(
                "SELECT i.source_ref, r.file_id, i.trashed_at
                   FROM images i LEFT JOIN remote r ON r.image_id = i.id
                  WHERE i.id = ?1",
                [image.0 as i64],
                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
            )
            .ok();

        let Some((from, file_id, trashed_at)) = row else {
            continue;
        };
        if trashed_at.is_some() {
            continue;
        }

        out.push(Move {
            image_id: image,
            file_id: file_id.map(|v| v as u64),
            to: trash::trash_path(root, image, &from),
            from,
        });
    }
    Ok(out)
}

/// Plan a restore: where each trashed image goes back to.
///
/// An image with no recorded origin is skipped rather than guessed at — putting
/// a photograph in the wrong folder is harder to notice, and harder to undo,
/// than leaving it in the trash.
pub fn plan_restore(
    catalog: &Catalog,
    images: &[ImageId],
) -> Result<Vec<Move>, dr_catalog::CatalogError> {
    let mut out = Vec::new();
    for &image in images {
        let row: Option<(String, Option<String>, Option<i64>)> = catalog
            .connection()
            .query_row(
                "SELECT i.source_ref, i.trashed_from, r.file_id
                   FROM images i LEFT JOIN remote r ON r.image_id = i.id
                  WHERE i.id = ?1 AND i.trashed_at IS NOT NULL",
                [image.0 as i64],
                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
            )
            .ok();

        let Some((from, Some(to), file_id)) = row else {
            continue;
        };
        out.push(Move {
            image_id: image,
            file_id: file_id.map(|v| v as u64),
            from,
            to,
        });
    }
    Ok(out)
}

/// Move images to or from the trash on a worker thread.
///
/// The catalog is written by the worker, after each successful move, so an
/// interrupted batch leaves the rows it completed correct rather than losing all
/// of them.
pub fn spawn_move(
    conn: Connection,
    moves: Vec<Move>,
    direction: Direction,
    catalog_path: PathBuf,
) -> Receiver<TrashMessage> {
    let (tx, rx) = std::sync::mpsc::channel();

    std::thread::spawn(move || {
        let total = moves.len();
        let rt = match crate::net_runtime::build() {
            Ok(rt) => rt,
            Err(e) => {
                let _ = tx.send(TrashMessage::Done {
                    moved: 0,
                    failed: vec![e.to_string()],
                });
                return;
            }
        };

        rt.block_on(async {
            let backend = match crate::remote::connect(&conn) {
                Ok(b) => b,
                Err(e) => {
                    let _ = tx.send(TrashMessage::Done {
                        moved: 0,
                        failed: vec![e.to_string()],
                    });
                    return;
                }
            };

            let mut succeeded: Vec<(ImageId, String)> = Vec::new();
            let mut failed: Vec<String> = Vec::new();

            for (i, mv) in moves.iter().enumerate() {
                // Addressed by path, not by `mv.file_id`: WebDAV has no
                // fileid-addressable endpoint, so a `RemoteId::Stable` here is
                // rejected by the backend. The fileid is an identity that the
                // MOVE preserves, not a way to name the source.
                let id = RemoteId::Path(RemotePath::new(&mv.from));

                match backend.move_to(&id, &RemotePath::new(&mv.to)).await {
                    Ok(()) => {
                        // The fileid is logged, not sent: if a restore later
                        // shows a missing thumbnail, this is the record of which
                        // identity the MOVE was supposed to carry across.
                        if let Some(f) = mv.file_id {
                            log::debug!("moved {} to {} as fileid {f}", mv.from, mv.to);
                        }
                        succeeded.push((mv.image_id, mv.to.clone()));
                    }
                    Err(e) => {
                        // Named by file, not by id: the user recognises the
                        // filename and cannot do anything with a row number.
                        let name = mv.from.rsplit('/').next().unwrap_or(&mv.from);
                        log::warn!("moving {name}: {e}");
                        failed.push(format!("{name}: {e}"));
                    }
                }

                let _ = tx.send(TrashMessage::Progress {
                    done: i + 1,
                    total,
                    failed: failed.len(),
                });
            }

            // Record after the moves, in one transaction. A crash before this
            // leaves the files moved and the catalog stale — recoverable,
            // because the next scan cannot see them in the trash folder and the
            // rows still point at paths that 404, which the UI reports.
            if !succeeded.is_empty() {
                match Catalog::open(&catalog_path) {
                    Ok(cat) => {
                        let result = match direction {
                            Direction::ToTrash => {
                                trash::record_trashed(cat.connection(), &succeeded, now_secs())
                            }
                            Direction::Restore => {
                                trash::record_restored(cat.connection(), &succeeded)
                            }
                        };
                        if let Err(e) = result {
                            log::warn!("recording trash state: {e}");
                            failed.push(format!("catalog: {e}"));
                        }
                    }
                    Err(e) => {
                        log::warn!("opening catalog to record trash state: {e}");
                        failed.push(format!("catalog: {e}"));
                    }
                }
            }

            let _ = tx.send(TrashMessage::Done {
                moved: succeeded.len(),
                failed,
            });
        });
    });

    rx
}

/// Permanently delete trashed images: the file, then the row, then the thumbnail.
///
/// See the module preamble for why that order. `thumbs_dir` is passed so the
/// worker can drop the previews — the shards sync, so a stale entry would keep
/// serving a preview of a deleted photograph on every device.
pub fn spawn_purge(
    conn: Connection,
    images: Vec<ImageId>,
    paths: Vec<(ImageId, Option<u64>, String)>,
    catalog_path: PathBuf,
    thumbs_dir: PathBuf,
) -> Receiver<TrashMessage> {
    let (tx, rx) = std::sync::mpsc::channel();

    std::thread::spawn(move || {
        let total = paths.len();
        let rt = match crate::net_runtime::build() {
            Ok(rt) => rt,
            Err(e) => {
                let _ = tx.send(TrashMessage::Done {
                    moved: 0,
                    failed: vec![e.to_string()],
                });
                return;
            }
        };

        rt.block_on(async {
            let backend = match crate::remote::connect(&conn) {
                Ok(b) => b,
                Err(e) => {
                    let _ = tx.send(TrashMessage::Done {
                        moved: 0,
                        failed: vec![e.to_string()],
                    });
                    return;
                }
            };

            let mut deleted: Vec<ImageId> = Vec::new();
            let mut dead_thumbs: Vec<u64> = Vec::new();
            let mut failed: Vec<String> = Vec::new();

            for (i, (image, file_id, path)) in paths.iter().enumerate() {
                // By path — see `spawn_move`. `file_id` still matters below, as
                // the key the thumbnail shards are stored under.
                let id = RemoteId::Path(RemotePath::new(path));

                match backend.delete(&id, None).await {
                    Ok(()) => {
                        deleted.push(*image);
                        if let Some(f) = file_id {
                            dead_thumbs.push(*f);
                        }
                    }
                    // Already gone is the goal state, not a failure. Treating it
                    // as an error would wedge every future empty-trash on a file
                    // the user had removed by hand.
                    Err(e) if is_missing(&e) => {
                        log::debug!("{path} was already gone");
                        deleted.push(*image);
                        if let Some(f) = file_id {
                            dead_thumbs.push(*f);
                        }
                    }
                    Err(e) => {
                        let name = path.rsplit('/').next().unwrap_or(path);
                        log::warn!("deleting {name}: {e}");
                        failed.push(format!("{name}: {e}"));
                    }
                }

                let _ = tx.send(TrashMessage::Progress {
                    done: i + 1,
                    total,
                    failed: failed.len(),
                });
            }

            // Rows after the files — see `trash::purge_order`.
            if !deleted.is_empty() {
                match Catalog::open(&catalog_path) {
                    Ok(cat) => {
                        if let Err(e) = trash::forget(cat.connection(), &deleted) {
                            log::warn!("forgetting purged rows: {e}");
                            failed.push(format!("catalog: {e}"));
                        }
                    }
                    Err(e) => failed.push(format!("catalog: {e}")),
                }
            }

            // Thumbnails last. A failure here is logged and dropped: the
            // photographs are gone, which was the point, and a stale preview is
            // a cosmetic problem rather than a reason to report the delete
            // failed.
            if !dead_thumbs.is_empty() {
                match dr_thumbs::ThumbStore::open(&thumbs_dir) {
                    Ok(mut store) => match store.forget(&dead_thumbs) {
                        Ok(n) => log::info!("dropped {n} thumbnail(s) for purged images"),
                        Err(e) => log::warn!("dropping thumbnails: {e}"),
                    },
                    Err(e) => log::warn!("opening thumbnail store to drop previews: {e}"),
                }
            }

            let _ = tx.send(TrashMessage::Done {
                moved: deleted.len(),
                failed,
            });
        });

        let _ = images;
    });

    rx
}

/// Whether a remote error means the object is not there.
///
/// Kept next to its use rather than in `dr-catalog`: it inspects a
/// `dr_sync::RemoteError`, and the catalog crate neither sees nor should see
/// that type.
fn is_missing(e: &RemoteError) -> bool {
    matches!(e, RemoteError::NotFound(_))
}

fn now_secs() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}

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

    fn seeded() -> Catalog {
        let cat = Catalog::in_memory().unwrap();
        let c = cat.connection();
        c.execute(
            "INSERT INTO roots(id, kind, label) VALUES (1, 'remote', 'PhotosRaw')",
            [],
        )
        .unwrap();
        for i in 1..=3i64 {
            c.execute(
                "INSERT INTO images(id, root_id, source_ref, file_size, added_at)
                 VALUES (?1, 1, ?2, 1000, 0)",
                rusqlite::params![i, format!("PhotosRaw/2019/IMG_{i:04}.CR2")],
            )
            .unwrap();
            c.execute(
                "INSERT INTO remote(image_id, file_id) VALUES (?1, ?2)",
                rusqlite::params![i, 500 + i],
            )
            .unwrap();
        }
        cat
    }

    fn img(i: u64) -> ImageId {
        ImageId(i)
    }

    #[test]
    fn a_trash_plan_targets_the_trash_folder_and_keeps_the_stable_id() {
        let cat = seeded();
        let plan = plan_trash(&cat, "PhotosRaw", &[img(1)]).unwrap();

        assert_eq!(plan.len(), 1);
        assert_eq!(plan[0].from, "PhotosRaw/2019/IMG_0001.CR2");
        assert!(plan[0].to.contains(".darkroom-trash"), "{}", plan[0].to);
        // The stable id is what makes the move keep the thumbnail attached.
        assert_eq!(plan[0].file_id, Some(501));
    }

    #[test]
    fn an_already_trashed_image_is_not_trashed_again() {
        // The selection can overlap what is already in the trash; a second move
        // would relocate the file *within* the trash and lose its origin.
        let cat = seeded();
        let plan = plan_trash(&cat, "PhotosRaw", &[img(1)]).unwrap();
        trash::record_trashed(cat.connection(), &[(img(1), plan[0].to.clone())], 100).unwrap();

        assert!(plan_trash(&cat, "PhotosRaw", &[img(1)]).unwrap().is_empty());
    }

    #[test]
    fn a_restore_plan_sends_each_image_back_where_it_came_from() {
        let cat = seeded();
        let plan = plan_trash(&cat, "PhotosRaw", &[img(1)]).unwrap();
        trash::record_trashed(cat.connection(), &[(img(1), plan[0].to.clone())], 100).unwrap();

        let back = plan_restore(&cat, &[img(1)]).unwrap();
        assert_eq!(back.len(), 1);
        assert_eq!(back[0].to, "PhotosRaw/2019/IMG_0001.CR2");
        // And it moves *from* the trash.
        assert!(back[0].from.contains(".darkroom-trash"));
    }

    #[test]
    fn restoring_an_untrashed_image_plans_nothing() {
        let cat = seeded();
        assert!(plan_restore(&cat, &[img(2)]).unwrap().is_empty());
    }

    #[test]
    fn a_missing_image_is_skipped_rather_than_planned_against_nothing() {
        // The grid's selection can outlive a rescan that removed a row.
        let cat = seeded();
        assert!(plan_trash(&cat, "PhotosRaw", &[img(999)])
            .unwrap()
            .is_empty());
        assert!(plan_restore(&cat, &[img(999)]).unwrap().is_empty());
    }

    #[test]
    fn a_not_found_on_delete_counts_as_deleted() {
        // Otherwise one hand-removed file wedges every future empty-trash.
        assert!(is_missing(&RemoteError::NotFound("x".into())));
        assert!(!is_missing(&RemoteError::Unsupported("x")));
    }

    #[test]
    fn planning_over_an_empty_selection_is_a_no_op() {
        let cat = seeded();
        assert!(plan_trash(&cat, "PhotosRaw", &[]).unwrap().is_empty());
        assert!(plan_restore(&cat, &[]).unwrap().is_empty());
    }
}