//! TRACES: FR-CAT-7 | FR-CAT-6 | NFR-R5
//! Collections: hierarchy, membership, and the edits the UI performs.
//!
//! The schema for this landed with the catalog (§2) and the cross-device merge
//! rules landed with [`crate::merge`]. What was missing is the layer between
//! them — the operations a user actually performs: make a collection, nest it,
//! drag images into it, take them out again.
//!
//! # Two independent structures
//!
//! **Hierarchy** is `collections.parent_id`: a collection inside a collection,
//! Lightroom's "collection set". A parent is an ordinary collection, not a
//! separate kind of thing, so a set can hold images of its own.
//!
//! **Membership** is `collection_members`, a join table. An image is in as many
//! collections as the user likes; nothing is moved or copied on disk, and no
//! collection owns an image. That is why dragging an image into a collection is
//! *additive* by default — it does not remove it from where it already is.
//!
//! These two are deliberately not entangled. Adding an image to a child does
//! **not** insert a row for the parent: a parent's contents are the union of
//! its own members and its descendants', computed on read by [`descendants`].
//! Materialising it instead would mean an add touching every ancestor, and a
//! reparent rewriting membership — both of which the merge layer would then
//! have to reconcile row by row.
//!
//! # Every edit bumps a revision
//!
//! [`crate::merge`] resolves conflicts by `revision`, a per-collection counter,
//! never by timestamp — a device with a skewed clock must not silently win. So
//! every mutation here goes through [`touch`], and none of them write
//! `modified` without also incrementing `revision`. Forgetting that is not a
//! local bug: it makes the *other* device lose work at the next sync.
//!
//! # Cycles
//!
//! Two ways a cycle can form, and both are refused rather than repaired:
//! parenting a collection under its own descendant ([`set_parent`]), and a
//! smart collection whose selector references itself ([`save_smart`]).
//! A cycle is unbounded recursion in the tree walk, so it must never be
//! representable in the database, not merely handled when drawn.

use rusqlite::{Connection, OptionalExtension};

use dr_types::{CollectionId, ImageId, Selector};

use crate::error::CatalogError;

/// Manual or smart, matching `collections.kind`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CollectionKind {
    /// A set the user assembled by hand. Membership lives in
    /// `collection_members`.
    Manual = 0,
    /// A saved filter. Membership is whatever its selector matches *now*, so
    /// there are no member rows and dropping images onto one is refused.
    Smart = 1,
}

impl CollectionKind {
    fn from_i64(v: i64) -> Self {
        match v {
            1 => CollectionKind::Smart,
            _ => CollectionKind::Manual,
        }
    }
}

/// One collection, as the sidebar draws it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Collection {
    pub id: CollectionId,
    /// Device-independent identity. The integer `id` is local and collides
    /// across devices; this is what a merge keys on.
    pub uuid: String,
    pub name: String,
    pub parent: Option<CollectionId>,
    pub kind: CollectionKind,
    /// The stored selector, for a smart collection.
    pub selector: Option<Selector>,
    /// Direct members only — not descendants'. The sidebar shows both, and
    /// conflating them makes an empty parent of full children look full.
    pub direct_count: usize,
}

/// A collection plus where it sits in the tree, flattened for display.
///
/// Slint models are flat, and a recursive `for` is not expressible in `.slint`,
/// so the tree is flattened here — in the place that already knows the
/// ordering rules — rather than in the UI.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TreeRow {
    pub collection: Collection,
    /// 0 for a root collection, incrementing per level.
    pub depth: usize,
    /// Whether this row has children, so the UI can draw a disclosure triangle
    /// without querying per row.
    pub has_children: bool,
}

/// Create a collection, optionally inside `parent`.
///
/// The UUID is generated here rather than taken from the caller: it is the
/// merge identity, and a caller that reuses one silently fuses two collections
/// on the next sync.
pub fn create(
    conn: &Connection,
    name: &str,
    parent: Option<CollectionId>,
    kind: CollectionKind,
) -> Result<CollectionId, CatalogError> {
    if let Some(p) = parent {
        // Fail before inserting rather than leaving an orphan pointing at a
        // parent that does not exist.
        require_exists(conn, p)?;
    }

    let now = now_secs();
    conn.execute(
        "INSERT INTO collections(uuid, name, parent_id, kind, created, revision, modified)
         VALUES (?1, ?2, ?3, ?4, ?5, 1, ?5)",
        rusqlite::params![
            new_uuid(),
            name.trim(),
            parent.map(|p| p.0 as i64),
            kind as i64,
            now,
        ],
    )?;
    Ok(CollectionId(conn.last_insert_rowid() as u64))
}

/// Rename a collection.
pub fn rename(conn: &Connection, id: CollectionId, name: &str) -> Result<(), CatalogError> {
    let n = conn.execute(
        "UPDATE collections SET name = ?2 WHERE id = ?1 AND deleted = 0",
        rusqlite::params![id.0 as i64, name.trim()],
    )?;
    if n == 0 {
        return Err(CatalogError::NoSuchCollection(id.0));
    }
    touch(conn, id)
}

/// Reparent a collection — the drag that restructures the tree.
///
/// `None` moves it to the top level. Refuses to create a cycle: parenting a
/// collection under its own descendant would make the tree walk recurse
/// forever, so it is rejected at the write rather than defended against at
/// every read.
pub fn set_parent(
    conn: &Connection,
    id: CollectionId,
    parent: Option<CollectionId>,
) -> Result<(), CatalogError> {
    require_exists(conn, id)?;

    if let Some(p) = parent {
        require_exists(conn, p)?;
        // A collection cannot be its own parent, and cannot descend from
        // itself. `descendants` includes `id`, which covers both cases in one
        // check.
        if descendants(conn, id)?.contains(&p) {
            return Err(CatalogError::CollectionCycle(id.0));
        }
    }

    conn.execute(
        "UPDATE collections SET parent_id = ?2 WHERE id = ?1",
        rusqlite::params![id.0 as i64, parent.map(|p| p.0 as i64)],
    )?;
    touch(conn, id)
}

/// Delete a collection, leaving a tombstone.
///
/// The row survives with `deleted = 1` because a merge against a device that
/// still holds the collection would otherwise resurrect it (§8.1). Children are
/// **promoted to the deleted collection's parent** rather than deleted with it:
/// the schema's `ON DELETE CASCADE` would take the whole subtree, and losing a
/// nested collection because its container was tidied away is not recoverable.
///
/// Member rows are dropped — they carry no independent identity, and the
/// tombstone is what merges.
pub fn delete(conn: &Connection, id: CollectionId) -> Result<(), CatalogError> {
    let parent: Option<i64> = conn
        .query_row(
            "SELECT parent_id FROM collections WHERE id = ?1 AND deleted = 0",
            [id.0 as i64],
            |r| r.get(0),
        )
        .optional()?
        .ok_or(CatalogError::NoSuchCollection(id.0))?;

    let tx = conn.unchecked_transaction()?;

    // Promote children first, so the subtree is detached before the tombstone.
    tx.execute(
        "UPDATE collections SET parent_id = ?2 WHERE parent_id = ?1",
        rusqlite::params![id.0 as i64, parent],
    )?;
    tx.execute(
        "DELETE FROM collection_members WHERE collection_id = ?1",
        [id.0 as i64],
    )?;
    tx.execute(
        "UPDATE collections
            SET deleted = 1, revision = revision + 1, modified = ?2
          WHERE id = ?1",
        rusqlite::params![id.0 as i64, now_secs()],
    )?;

    tx.commit()?;
    Ok(())
}

/// Add images to a manual collection. The drop half of drag-and-drop.
///
/// Additive and idempotent: an image already present keeps its original
/// `position` and `added` rather than jumping to the end, because re-dropping a
/// selection that overlaps what is already there is a normal thing to do and
/// must not reshuffle the collection.
///
/// Returns how many rows were genuinely new, which is what the UI reports —
/// "added 3 of 12" is the honest message when nine were already in.
pub fn add_images(
    conn: &Connection,
    id: CollectionId,
    images: &[ImageId],
) -> Result<usize, CatalogError> {
    let tx = conn.unchecked_transaction()?;
    let added = add_within(&tx, id, images)?;
    tx.commit()?;
    Ok(added)
}

/// [`add_images`] without opening a transaction.
///
/// Separate because SQLite has no nested `BEGIN`: [`move_images`] needs an add
/// and a removal inside *one* transaction, and calling the public form there
/// fails at runtime with "cannot start a transaction within a transaction".
fn add_within(
    conn: &Connection,
    id: CollectionId,
    images: &[ImageId],
) -> Result<usize, CatalogError> {
    let kind = kind_of(conn, id)?;
    if kind == CollectionKind::Smart {
        // A smart collection's membership *is* its selector. Writing member
        // rows would create a second, silently ignored source of truth.
        return Err(CatalogError::SmartCollectionNotEditable(id.0));
    }
    if images.is_empty() {
        return Ok(0);
    }

    let now = now_secs();

    // Append after whatever is already there, so a drop lands at the end in
    // the order the user dragged.
    let mut next: i64 = conn.query_row(
        "SELECT coalesce(max(position), -1) + 1 FROM collection_members
          WHERE collection_id = ?1",
        [id.0 as i64],
        |r| r.get(0),
    )?;

    let mut added = 0;
    {
        let mut stmt = conn.prepare(
            "INSERT INTO collection_members(collection_id, image_id, position, added)
             VALUES (?1, ?2, ?3, ?4)
             ON CONFLICT(collection_id, image_id) DO NOTHING",
        )?;
        for img in images {
            if stmt.execute(rusqlite::params![id.0 as i64, img.0 as i64, next, now])? > 0 {
                added += 1;
                next += 1;
            }
        }
    }

    // Only a real change is an edit. Bumping the revision for a no-op drop
    // would make an idle device win a merge against one that did real work.
    if added > 0 {
        touch(conn, id)?;
    }
    Ok(added)
}

/// Remove images from a collection.
///
/// Removes membership only — the images themselves are untouched, and stay in
/// every other collection they belong to. That asymmetry is the whole point of
/// a join table, and it is why this is "remove from collection" and never
/// "delete".
pub fn remove_images(
    conn: &Connection,
    id: CollectionId,
    images: &[ImageId],
) -> Result<usize, CatalogError> {
    let tx = conn.unchecked_transaction()?;
    let removed = remove_within(&tx, id, images)?;
    tx.commit()?;
    Ok(removed)
}

/// [`remove_images`] without opening a transaction. See [`add_within`].
fn remove_within(
    conn: &Connection,
    id: CollectionId,
    images: &[ImageId],
) -> Result<usize, CatalogError> {
    if images.is_empty() {
        return Ok(0);
    }
    let mut removed = 0;
    {
        let mut stmt = conn
            .prepare("DELETE FROM collection_members WHERE collection_id = ?1 AND image_id = ?2")?;
        for img in images {
            removed += stmt.execute(rusqlite::params![id.0 as i64, img.0 as i64])?;
        }
    }

    if removed > 0 {
        touch(conn, id)?;
    }
    Ok(removed)
}

/// Move images from one collection to another — a drag with the modifier held.
///
/// One transaction, so a failure cannot leave the images in neither place.
/// A move onto the collection they came from is a no-op rather than a
/// remove-then-add that would lose their positions.
pub fn move_images(
    conn: &Connection,
    from: CollectionId,
    to: CollectionId,
    images: &[ImageId],
) -> Result<usize, CatalogError> {
    if from == to || images.is_empty() {
        return Ok(0);
    }
    let tx = conn.unchecked_transaction()?;
    // Add first: if `to` turns out to be a smart collection the whole move
    // aborts with nothing removed, rather than emptying `from` into nowhere.
    let added = add_within(&tx, to, images)?;
    remove_within(&tx, from, images)?;
    tx.commit()?;
    Ok(added)
}

/// Reorder a manual collection: put `images` at the front, in the order given.
///
/// Manual position is what `Sort::Position` reads. Rewriting the whole run
/// rather than patching one row keeps positions dense, so there is no
/// renumbering pass to run later.
pub fn set_order(
    conn: &Connection,
    id: CollectionId,
    images: &[ImageId],
) -> Result<(), CatalogError> {
    let tx = conn.unchecked_transaction()?;
    {
        let mut stmt = tx.prepare(
            "UPDATE collection_members SET position = ?3
              WHERE collection_id = ?1 AND image_id = ?2",
        )?;
        for (i, img) in images.iter().enumerate() {
            stmt.execute(rusqlite::params![id.0 as i64, img.0 as i64, i as i64])?;
        }
    }
    tx.commit()?;
    touch(conn, id)
}

/// Whether this collection has a manual order worth reading.
///
/// The grid asks before choosing an `ORDER BY`, and the answer is narrower
/// than "is it manual". Two things have to be true.
///
/// It must be **manual**: a saved filter's contents are whatever its selector
/// matches now, in whatever order the query returns them, and there are no
/// member rows to carry a position.
///
/// And it must have **no children**. A parent's contents are the union of its
/// own members and its descendants' ([`descendants`]), and positions are only
/// ever assigned within one collection — so two children's positions are
/// unrelated integers, and interleaving them by value would order the grid by
/// a coincidence. Capture time is the honest answer for a set.
///
/// Kept here rather than assembled at the call site, so the rule has one name
/// and one test, and the grid does not have to know that "no children" is part
/// of it.
pub fn orders_manually(conn: &Connection, id: CollectionId) -> Result<bool, CatalogError> {
    if kind_of(conn, id)? != CollectionKind::Manual {
        return Ok(false);
    }

    let has_child: Option<i64> = conn
        .query_row(
            "SELECT 1 FROM collections
              WHERE parent_id = ?1 AND deleted = 0 LIMIT 1",
            [id.0 as i64],
            |r| r.get(0),
        )
        .optional()?;

    Ok(has_child.is_none())
}

/// Every image in a manual collection, in manual order.
///
/// The whole membership, not the filtered view the grid happens to be showing.
/// [`set_order`] renumbers exactly the images it is handed, so a caller that
/// reordered a filtered list would renumber those and leave every hidden image
/// on a stale position — two images sharing a position, and a grid that
/// rearranges itself when the filter comes off.
///
/// `image_id` breaks ties. Positions are dense in practice, but a merge can
/// deliver two devices' rows at the same position, and an order that is not
/// total is one the grid draws differently each time it reads it.
pub fn members_in_order(conn: &Connection, id: CollectionId) -> Result<Vec<ImageId>, CatalogError> {
    let mut stmt = conn.prepare(
        "SELECT image_id FROM collection_members
          WHERE collection_id = ?1
          ORDER BY position, image_id",
    )?;
    let rows = stmt
        .query_map([id.0 as i64], |r| Ok(ImageId(r.get::<_, i64>(0)? as u64)))?
        .collect::<Result<Vec<_>, _>>()?;
    Ok(rows)
}

/// Save a smart collection's selector.
///
/// Refuses a selector that references this collection, directly or through
/// another that leads back here — evaluating it would recurse forever.
pub fn save_smart(
    conn: &Connection,
    id: CollectionId,
    selector: &Selector,
) -> Result<(), CatalogError> {
    let mut referenced = Vec::new();
    selector.collections(&mut referenced);

    // Direct self-reference, and indirect: a referenced smart collection whose
    // own selector leads back to `id`.
    for r in &referenced {
        if *r == id || smart_reaches(conn, *r, id, 0)? {
            return Err(CatalogError::CollectionCycle(id.0));
        }
    }

    let json =
        serde_json::to_string(selector).map_err(|e| CatalogError::BadSelector(e.to_string()))?;
    let n = conn.execute(
        "UPDATE collections SET selector_json = ?2, kind = 1 WHERE id = ?1 AND deleted = 0",
        rusqlite::params![id.0 as i64, json],
    )?;
    if n == 0 {
        return Err(CatalogError::NoSuchCollection(id.0));
    }
    touch(conn, id)
}

/// Whether `from`'s selector can reach `target` by following references.
///
/// Depth-limited rather than visited-set tracking: the database already refuses
/// to store a cycle, so this defends against a hand-edited or merged-in row,
/// and a bound is enough for that. A merge can introduce a selector this device
/// never validated, which is exactly when the read path must not hang.
fn smart_reaches(
    conn: &Connection,
    from: CollectionId,
    target: CollectionId,
    depth: usize,
) -> Result<bool, CatalogError> {
    const MAX_DEPTH: usize = 16;
    if depth >= MAX_DEPTH {
        // Treat exhaustion as "yes, a cycle": refusing to save an
        // implausibly-nested selector is safer than storing one that hangs.
        return Ok(true);
    }

    let json: Option<String> = conn
        .query_row(
            "SELECT selector_json FROM collections WHERE id = ?1 AND deleted = 0",
            [from.0 as i64],
            |r| r.get(0),
        )
        .optional()?
        .flatten();

    let Some(json) = json else { return Ok(false) };
    let Ok(sel) = serde_json::from_str::<Selector>(&json) else {
        // An unreadable stored selector cannot be followed. Reporting "no
        // cycle" is right: the broken row is a separate problem, surfaced when
        // that collection is read.
        return Ok(false);
    };

    let mut refs = Vec::new();
    sel.collections(&mut refs);
    for r in refs {
        if r == target || smart_reaches(conn, r, target, depth + 1)? {
            return Ok(true);
        }
    }
    Ok(false)
}

/// Every collection an image belongs to.
///
/// What the grid needs to badge a cell with its collections, and what makes
/// "remove from this collection" distinguishable from "this image is only
/// here".
pub fn collections_for_image(
    conn: &Connection,
    image: ImageId,
) -> Result<Vec<CollectionId>, CatalogError> {
    let mut stmt = conn.prepare(
        "SELECT m.collection_id FROM collection_members m
           JOIN collections c ON c.id = m.collection_id
          WHERE m.image_id = ?1 AND c.deleted = 0
          ORDER BY c.name",
    )?;
    let rows = stmt
        .query_map([image.0 as i64], |r| {
            Ok(CollectionId(r.get::<_, i64>(0)? as u64))
        })?
        .collect::<Result<Vec<_>, _>>()?;
    Ok(rows)
}

/// One collection a set of images is filed in, and how much of that set is in
/// it.
///
/// `holding` is what makes a removal honest: with forty photographs selected
/// and three of them in "Iceland", the row has to say "3 of 40" or the user
/// reads it as "this selection is in Iceland" and takes all forty out of a
/// collection thirty-seven of them were never in.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Membership {
    pub id: CollectionId,
    pub name: String,
    /// How many of the images asked about are members. Never zero — a
    /// collection holding none of them is not returned at all.
    pub holding: usize,
}

/// Every collection the given images are filed in, with how many of them each
/// holds.
///
/// The read behind "which collections is this selection in, and take it out of
/// one" — the counterpart to [`collections_for_image`], which answers the same
/// question for a single photograph and does not need the counts.
///
/// Smart collections never appear: they have no `collection_members` rows, so
/// there is nothing to remove and offering it would be a button that does
/// nothing. Sorted by name, matching the sidebar.
///
/// # Why this is chunked
///
/// The image list is a *selection*, which a select-all makes as large as the
/// library. SQLite caps the number of bound parameters in one statement, so a
/// single `IN (...)` over every selected id fails outright on exactly the
/// gesture most likely to produce it. The counts are summed across chunks
/// rather than re-queried, so the result is the same as the unchunked query
/// would have given.
pub fn membership_of(
    conn: &Connection,
    images: &[ImageId],
) -> Result<Vec<Membership>, CatalogError> {
    if images.is_empty() {
        return Ok(Vec::new());
    }

    // Well under SQLite's default parameter cap, and large enough that an
    // ordinary selection is one round trip.
    const CHUNK: usize = 400;

    let mut totals: std::collections::HashMap<CollectionId, (String, usize)> =
        std::collections::HashMap::new();

    for chunk in images.chunks(CHUNK) {
        let placeholders = std::iter::repeat_n("?", chunk.len())
            .collect::<Vec<_>>()
            .join(",");
        // The placeholder list is built from the id *count*, never from user
        // text — the same construction `deep_count` uses.
        let sql = format!(
            "SELECT c.id, c.name, count(*)
               FROM collection_members m
               JOIN collections c ON c.id = m.collection_id
              WHERE m.image_id IN ({placeholders}) AND c.deleted = 0
              GROUP BY c.id, c.name"
        );
        let params: Vec<rusqlite::types::Value> = chunk
            .iter()
            .map(|i| rusqlite::types::Value::Integer(i.0 as i64))
            .collect();

        let mut stmt = conn.prepare(&sql)?;
        let rows = stmt.query_map(rusqlite::params_from_iter(params.iter()), |r| {
            Ok((
                CollectionId(r.get::<_, i64>(0)? as u64),
                r.get::<_, String>(1)?,
                r.get::<_, i64>(2)? as usize,
            ))
        })?;

        for row in rows {
            let (id, name, n) = row?;
            let entry = totals.entry(id).or_insert((name, 0));
            entry.1 += n;
        }
    }

    let mut out: Vec<Membership> = totals
        .into_iter()
        .map(|(id, (name, holding))| Membership { id, name, holding })
        .collect();
    // By name, then by id, so two collections sharing a name have a stable
    // order rather than the hash map's.
    out.sort_by(|a, b| {
        a.name
            .to_lowercase()
            .cmp(&b.name.to_lowercase())
            .then(a.id.0.cmp(&b.id.0))
    });
    Ok(out)
}

/// What kind of collection `id` is, or `None` if there is no such collection.
///
/// Cheaper than reading the whole [`Collection`] where the caller only needs to
/// know whether member rows exist — the grid asks this to decide whether manual
/// position is a thing it can order by, and a smart collection has no
/// `collection_members` rows to carry one.
pub fn kind(conn: &Connection, id: CollectionId) -> Result<Option<CollectionKind>, CatalogError> {
    let found = conn
        .query_row(
            "SELECT kind FROM collections WHERE id = ?1 AND deleted = 0",
            [id.0 as i64],
            |r| r.get::<_, i64>(0),
        )
        .optional()?;
    Ok(found.map(CollectionKind::from_i64))
}

/// TRACES: FR-UI-8
/// The device-independent name of a collection, from its local id.
///
/// The pair to [`id_for_uuid`], and the reason both exist: `collections.id` is
/// an autoincrement local to one catalog, so anything that travels between
/// devices — a place, a merge — has to say which collection it means in the
/// only vocabulary they share.
///
/// `None` for a collection that is not there, or has been tombstoned. A caller
/// writing down a scope treats that as "the whole library", which is the
/// harmless direction: the alternative is recording a name nothing can resolve.
pub fn uuid_of(conn: &Connection, id: CollectionId) -> Result<Option<String>, CatalogError> {
    Ok(conn
        .query_row(
            "SELECT uuid FROM collections WHERE id = ?1 AND deleted = 0",
            [id.0 as i64],
            |r| r.get::<_, String>(0),
        )
        .optional()?)
}

/// TRACES: FR-UI-8
/// The local id of a collection, from the name every device knows it by.
///
/// `None` where this device has never heard of it, or has deleted it — a place
/// recorded on the tablet inside a collection this machine has not yet merged.
/// The caller falls back to the whole library rather than to an empty grid.
pub fn id_for_uuid(conn: &Connection, uuid: &str) -> Result<Option<CollectionId>, CatalogError> {
    Ok(conn
        .query_row(
            "SELECT id FROM collections WHERE uuid = ?1 AND deleted = 0",
            [uuid],
            |r| r.get::<_, i64>(0),
        )
        .optional()?
        .map(|id| CollectionId(id as u64)))
}

/// A collection and everything beneath it, including itself.
///
/// Used for cycle checks and for scoping the grid to a parent: selecting a
/// collection set shows its children's images, which is what makes a set useful
/// rather than an empty folder.
///
/// Iterative rather than a recursive CTE so the depth guard is explicit: a
/// cycle introduced by a merge from another device must terminate here, not
/// spin inside SQLite.
pub fn descendants(
    conn: &Connection,
    root: CollectionId,
) -> Result<Vec<CollectionId>, CatalogError> {
    let mut out = vec![root];
    let mut frontier = vec![root];
    let mut guard = 0usize;

    let mut stmt =
        conn.prepare("SELECT id FROM collections WHERE parent_id = ?1 AND deleted = 0")?;

    while let Some(next) = frontier.pop() {
        guard += 1;
        if guard > MAX_TREE_NODES {
            // A cycle that reached us through a merge. Truncating is the only
            // non-hanging option, and the tree is a view — the user sees a
            // clipped subtree rather than a frozen window.
            log::warn!("collection tree walk exceeded {MAX_TREE_NODES} nodes; truncated");
            break;
        }

        let children = stmt
            .query_map([next.0 as i64], |r| {
                Ok(CollectionId(r.get::<_, i64>(0)? as u64))
            })?
            .collect::<Result<Vec<_>, _>>()?;

        for child in children {
            // Guards against a cycle reintroducing a node already visited.
            if !out.contains(&child) {
                out.push(child);
                frontier.push(child);
            }
        }
    }
    Ok(out)
}

/// Upper bound on nodes visited in one tree walk.
///
/// Not a limit on how many collections a user may have — it bounds a single
/// walk so a malformed tree cannot hang the UI thread.
const MAX_TREE_NODES: usize = 10_000;

/// The whole tree, flattened depth-first for the sidebar.
///
/// Siblings sort by name, so the order is stable across launches and across
/// devices — id order would differ per device, which is disorienting on the
/// same library seen from two machines.
///
/// One query for the rows and one for the counts, then assembled in memory:
/// a per-row count query would be one statement per collection, and the
/// sidebar redraws on every drop.
pub fn tree(conn: &Connection) -> Result<Vec<TreeRow>, CatalogError> {
    let mut stmt = conn.prepare(
        "SELECT c.id, c.uuid, c.name, c.parent_id, c.kind, c.selector_json,
                (SELECT count(*) FROM collection_members m WHERE m.collection_id = c.id)
           FROM collections c
          WHERE c.deleted = 0
          ORDER BY c.name COLLATE NOCASE, c.id",
    )?;

    let all: Vec<Collection> = stmt
        .query_map([], |r| {
            let selector = r
                .get::<_, Option<String>>(5)?
                .and_then(|j| serde_json::from_str(&j).ok());
            Ok(Collection {
                id: CollectionId(r.get::<_, i64>(0)? as u64),
                uuid: r.get(1)?,
                name: r.get(2)?,
                parent: r.get::<_, Option<i64>>(3)?.map(|v| CollectionId(v as u64)),
                kind: CollectionKind::from_i64(r.get(4)?),
                selector,
                direct_count: r.get::<_, i64>(6)? as usize,
            })
        })?
        .collect::<Result<Vec<_>, _>>()?;

    // Depth-first from the roots. A collection whose parent is missing or
    // deleted is treated as a root rather than dropped: an orphan must stay
    // reachable, or the images in it become invisible.
    let known: std::collections::HashSet<CollectionId> = all.iter().map(|c| c.id).collect();
    let mut out = Vec::with_capacity(all.len());
    let mut visited = std::collections::HashSet::new();

    fn walk(
        all: &[Collection],
        parent: Option<CollectionId>,
        depth: usize,
        visited: &mut std::collections::HashSet<CollectionId>,
        out: &mut Vec<TreeRow>,
    ) {
        for c in all.iter().filter(|c| c.parent == parent) {
            // A cycle from a merged-in row would otherwise recurse forever.
            if !visited.insert(c.id) {
                continue;
            }
            let has_children = all.iter().any(|k| k.parent == Some(c.id));
            out.push(TreeRow {
                collection: c.clone(),
                depth,
                has_children,
            });
            walk(all, Some(c.id), depth + 1, visited, out);
        }
    }

    walk(&all, None, 0, &mut visited, &mut out);

    // Orphans, after the real roots: their parent is gone, so they have no
    // place in the hierarchy but must not disappear.
    for c in &all {
        let orphaned = c.parent.is_some_and(|p| !known.contains(&p));
        if orphaned && visited.insert(c.id) {
            let has_children = all.iter().any(|k| k.parent == Some(c.id));
            out.push(TreeRow {
                collection: c.clone(),
                depth: 0,
                has_children,
            });
            walk(&all, Some(c.id), 1, &mut visited, &mut out);
        }
    }

    Ok(out)
}

/// How many images a collection holds counting its descendants.
///
/// Distinct images, not a sum: an image in both a parent and its child is one
/// photograph, and reporting two is the kind of small lie that makes a user
/// stop trusting the counts.
pub fn deep_count(conn: &Connection, id: CollectionId) -> Result<usize, CatalogError> {
    let ids = descendants(conn, id)?;
    let list: Vec<rusqlite::types::Value> = ids
        .iter()
        .map(|c| rusqlite::types::Value::Integer(c.0 as i64))
        .collect();

    // `carray` is not compiled in, so the placeholder list is built from the
    // id count — never from user text.
    let placeholders = std::iter::repeat_n("?", list.len())
        .collect::<Vec<_>>()
        .join(",");
    let sql = format!(
        "SELECT count(DISTINCT image_id) FROM collection_members
          WHERE collection_id IN ({placeholders})"
    );
    let n: i64 = conn.query_row(&sql, rusqlite::params_from_iter(list.iter()), |r| r.get(0))?;
    Ok(n as usize)
}

/// Bump `revision` and `modified` together.
///
/// Every mutation ends here. The merge layer compares revisions, so an edit
/// that updates `modified` alone is invisible to it — the other device keeps
/// its own version and the user's work vanishes at the next sync.
fn touch(conn: &Connection, id: CollectionId) -> Result<(), CatalogError> {
    conn.execute(
        "UPDATE collections SET revision = revision + 1, modified = ?2 WHERE id = ?1",
        rusqlite::params![id.0 as i64, now_secs()],
    )?;
    Ok(())
}

fn kind_of(conn: &Connection, id: CollectionId) -> Result<CollectionKind, CatalogError> {
    conn.query_row(
        "SELECT kind FROM collections WHERE id = ?1 AND deleted = 0",
        [id.0 as i64],
        |r| r.get::<_, i64>(0),
    )
    .optional()?
    .map(CollectionKind::from_i64)
    .ok_or(CatalogError::NoSuchCollection(id.0))
}

fn require_exists(conn: &Connection, id: CollectionId) -> Result<(), CatalogError> {
    let found: Option<i64> = conn
        .query_row(
            "SELECT 1 FROM collections WHERE id = ?1 AND deleted = 0",
            [id.0 as i64],
            |r| r.get(0),
        )
        .optional()?;
    found
        .map(|_| ())
        .ok_or(CatalogError::NoSuchCollection(id.0))
}

/// A random UUID, formatted as the canonical 8-4-4-4-12.
///
/// Hand-rolled rather than pulling in the `uuid` crate for one function, the
/// same reasoning as the connector's date parsing. Version 4 layout, seeded
/// from the OS via `getrandom` through `rusqlite`'s existing dependency-free
/// path — see below.
pub(crate) fn new_uuid() -> String {
    let b = random_bytes();
    // Version 4, variant 1, per RFC 4122 §4.4.
    let v6 = (b[6] & 0x0F) | 0x40;
    let v8 = (b[8] & 0x3F) | 0x80;
    format!(
        "{:02x}{:02x}{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-\
         {:02x}{:02x}{:02x}{:02x}{:02x}{:02x}",
        b[0],
        b[1],
        b[2],
        b[3],
        b[4],
        b[5],
        v6,
        b[7],
        v8,
        b[9],
        b[10],
        b[11],
        b[12],
        b[13],
        b[14],
        b[15]
    )
}

/// 16 bytes from the OS.
///
/// A UUID that collides fuses two different collections on the next merge, so
/// this reads the system CSPRNG rather than mixing a clock with an address.
/// The fallback path exists so a UUID is always produced, and it is *logged*:
/// silently degrading identity quality is how a collision becomes unexplainable
/// later.
fn random_bytes() -> [u8; 16] {
    use std::io::Read as _;
    let mut buf = [0u8; 16];

    match std::fs::File::open("/dev/urandom").and_then(|mut f| f.read_exact(&mut buf)) {
        Ok(()) => return buf,
        Err(e) => log::warn!("no OS randomness for a collection UUID ({e}); using a weak fallback"),
    }

    // Last resort: time and an address, which is unique enough within one
    // process to avoid fusing two collections created in one session.
    let t = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let addr = &buf as *const _ as usize as u128;
    let mixed = t ^ (addr << 64) ^ std::process::id() as u128;
    buf.copy_from_slice(&mixed.to_le_bytes());
    buf
}

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::*;
    use crate::Catalog;

    fn seeded() -> Catalog {
        let cat = Catalog::in_memory().unwrap();
        let c = cat.connection();
        c.execute(
            "INSERT INTO roots(id, kind, label) VALUES (1, 'local', 'lib')",
            [],
        )
        .unwrap();
        for i in 1..=6i64 {
            c.execute(
                "INSERT INTO images(id, root_id, source_ref, added_at)
                 VALUES (?1, 1, ?2, 0)",
                rusqlite::params![i, format!("img{i}.CR3")],
            )
            .unwrap();
        }
        cat
    }

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

    #[test]
    fn a_manual_leaf_orders_manually() {
        let cat = seeded();
        let c = cat.connection();
        let id = create(c, "Trip", None, CollectionKind::Manual).unwrap();
        assert!(orders_manually(c, id).unwrap());
    }

    #[test]
    fn a_saved_filter_has_no_manual_order() {
        let cat = seeded();
        let c = cat.connection();
        let id = create(c, "Picks", None, CollectionKind::Smart).unwrap();
        assert!(
            !orders_manually(c, id).unwrap(),
            "a smart collection has no member rows to carry a position"
        );
    }

    #[test]
    fn a_parent_has_no_manual_order_of_its_own() {
        let cat = seeded();
        let c = cat.connection();
        let parent = create(c, "2026", None, CollectionKind::Manual).unwrap();
        let child = create(c, "Iceland", Some(parent), CollectionKind::Manual).unwrap();

        assert!(
            !orders_manually(c, parent).unwrap(),
            "a set shows its descendants' images, whose positions are unrelated"
        );
        assert!(
            orders_manually(c, child).unwrap(),
            "the child is still a leaf and still orders manually"
        );
    }

    #[test]
    fn deleting_the_last_child_gives_the_parent_its_order_back() {
        let cat = seeded();
        let c = cat.connection();
        let parent = create(c, "2026", None, CollectionKind::Manual).unwrap();
        let child = create(c, "Iceland", Some(parent), CollectionKind::Manual).unwrap();
        assert!(!orders_manually(c, parent).unwrap());

        delete(c, child).unwrap();
        assert!(
            orders_manually(c, parent).unwrap(),
            "a tombstoned child must not keep counting as a child"
        );
    }

    #[test]
    fn members_come_back_in_the_order_they_were_put_in() {
        let cat = seeded();
        let c = cat.connection();
        let id = create(c, "Trip", None, CollectionKind::Manual).unwrap();
        add_images(c, id, &[img(3), img(1), img(2)]).unwrap();

        assert_eq!(
            members_in_order(c, id).unwrap(),
            vec![img(3), img(1), img(2)],
            "adding assigns position in the order given, and reading returns it"
        );
    }

    #[test]
    fn members_come_back_in_the_order_set_order_wrote() {
        let cat = seeded();
        let c = cat.connection();
        let id = create(c, "Trip", None, CollectionKind::Manual).unwrap();
        add_images(c, id, &[img(1), img(2), img(3)]).unwrap();

        set_order(c, id, &[img(3), img(2), img(1)]).unwrap();
        assert_eq!(
            members_in_order(c, id).unwrap(),
            vec![img(3), img(2), img(1)]
        );
    }

    #[test]
    fn members_in_order_is_total_even_where_positions_collide() {
        let cat = seeded();
        let c = cat.connection();
        let id = create(c, "Trip", None, CollectionKind::Manual).unwrap();
        add_images(c, id, &[img(1), img(2), img(3)]).unwrap();

        // What a merge can deliver: two devices' rows landing on one position.
        c.execute(
            "UPDATE collection_members SET position = 0 WHERE collection_id = ?1",
            [id.0 as i64],
        )
        .unwrap();

        assert_eq!(
            members_in_order(c, id).unwrap(),
            vec![img(1), img(2), img(3)],
            "image_id breaks the tie, so two reads cannot disagree"
        );
    }

    #[test]
    fn a_created_collection_appears_in_the_tree() {
        let cat = seeded();
        let c = cat.connection();
        create(c, "Iceland", None, CollectionKind::Manual).unwrap();

        let rows = tree(c).unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].collection.name, "Iceland");
        assert_eq!(rows[0].depth, 0);
    }

    #[test]
    fn uuids_are_distinct_per_collection() {
        // Two collections sharing a UUID fuse into one at the next merge, and
        // the user loses one of them.
        let cat = seeded();
        let c = cat.connection();
        let a = create(c, "A", None, CollectionKind::Manual).unwrap();
        let b = create(c, "B", None, CollectionKind::Manual).unwrap();

        let uuid = |id: CollectionId| -> String {
            c.query_row(
                "SELECT uuid FROM collections WHERE id = ?1",
                [id.0 as i64],
                |r| r.get(0),
            )
            .unwrap()
        };
        assert_ne!(uuid(a), uuid(b));
        // Canonical 8-4-4-4-12 layout, which is what the merge layer compares.
        assert_eq!(uuid(a).len(), 36);
    }

    #[test]
    fn nesting_reports_depth_and_children() {
        let cat = seeded();
        let c = cat.connection();
        let trips = create(c, "Trips", None, CollectionKind::Manual).unwrap();
        create(c, "Iceland", Some(trips), CollectionKind::Manual).unwrap();

        let rows = tree(c).unwrap();
        assert_eq!(rows.len(), 2);
        assert_eq!(rows[0].collection.name, "Trips");
        assert!(
            rows[0].has_children,
            "a parent must draw a disclosure arrow"
        );
        assert_eq!(rows[1].depth, 1, "the child is indented one level");
    }

    #[test]
    fn siblings_sort_by_name_not_by_id() {
        // Id order differs per device; the same library on two machines must
        // not present its collections in two different orders.
        let cat = seeded();
        let c = cat.connection();
        create(c, "Zebra", None, CollectionKind::Manual).unwrap();
        create(c, "Antelope", None, CollectionKind::Manual).unwrap();

        let names: Vec<String> = tree(c)
            .unwrap()
            .iter()
            .map(|r| r.collection.name.clone())
            .collect();
        assert_eq!(names, vec!["Antelope", "Zebra"]);
    }

    #[test]
    fn an_image_can_be_in_several_collections_at_once() {
        // The property the whole join table exists for: no collection owns an
        // image.
        let cat = seeded();
        let c = cat.connection();
        let a = create(c, "Portfolio", None, CollectionKind::Manual).unwrap();
        let b = create(c, "Print queue", None, CollectionKind::Manual).unwrap();

        add_images(c, a, &[img(1), img(2)]).unwrap();
        add_images(c, b, &[img(2)]).unwrap();

        let of_2 = collections_for_image(c, img(2)).unwrap();
        assert_eq!(of_2.len(), 2);
        assert!(of_2.contains(&a) && of_2.contains(&b));
    }

    #[test]
    fn dropping_the_same_images_twice_adds_nothing_and_reorders_nothing() {
        // Re-dropping an overlapping selection is normal; it must not shuffle
        // what is already in the collection.
        let cat = seeded();
        let c = cat.connection();
        let a = create(c, "Set", None, CollectionKind::Manual).unwrap();

        assert_eq!(add_images(c, a, &[img(1), img(2)]).unwrap(), 2);
        // Only image 3 is new.
        assert_eq!(add_images(c, a, &[img(1), img(2), img(3)]).unwrap(), 1);

        let positions: Vec<(i64, i64)> = {
            let mut stmt = c
                .prepare(
                    "SELECT image_id, position FROM collection_members
                      WHERE collection_id = ?1 ORDER BY position",
                )
                .unwrap();
            stmt.query_map([a.0 as i64], |r| Ok((r.get(0)?, r.get(1)?)))
                .unwrap()
                .map(Result::unwrap)
                .collect()
        };
        assert_eq!(positions, vec![(1, 0), (2, 1), (3, 2)]);
    }

    #[test]
    fn removing_from_one_collection_leaves_the_others() {
        let cat = seeded();
        let c = cat.connection();
        let a = create(c, "A", None, CollectionKind::Manual).unwrap();
        let b = create(c, "B", None, CollectionKind::Manual).unwrap();
        add_images(c, a, &[img(1)]).unwrap();
        add_images(c, b, &[img(1)]).unwrap();

        remove_images(c, a, &[img(1)]).unwrap();

        assert_eq!(collections_for_image(c, img(1)).unwrap(), vec![b]);
        // And the image itself is untouched — this is not a delete.
        let n: i64 = c
            .query_row("SELECT count(*) FROM images WHERE id = 1", [], |r| r.get(0))
            .unwrap();
        assert_eq!(n, 1);
    }

    #[test]
    fn every_edit_bumps_the_revision() {
        // The merge layer resolves by revision. An edit that does not bump it
        // loses to an idle device at the next sync.
        let cat = seeded();
        let c = cat.connection();
        let a = create(c, "A", None, CollectionKind::Manual).unwrap();
        let rev = |c: &Connection| -> i64 {
            c.query_row(
                "SELECT revision FROM collections WHERE id = ?1",
                [a.0 as i64],
                |r| r.get(0),
            )
            .unwrap()
        };

        let start = rev(c);
        add_images(c, a, &[img(1)]).unwrap();
        let after_add = rev(c);
        assert!(after_add > start, "an add is an edit");

        rename(c, a, "Renamed").unwrap();
        assert!(rev(c) > after_add, "a rename is an edit");

        remove_images(c, a, &[img(1)]).unwrap();
        assert!(rev(c) > after_add + 1, "a removal is an edit");
    }

    #[test]
    fn a_drop_that_adds_nothing_does_not_bump_the_revision() {
        // Otherwise an idle device that re-dropped the same images would win a
        // merge against one that did real work.
        let cat = seeded();
        let c = cat.connection();
        let a = create(c, "A", None, CollectionKind::Manual).unwrap();
        add_images(c, a, &[img(1)]).unwrap();

        let before: i64 = c
            .query_row(
                "SELECT revision FROM collections WHERE id = ?1",
                [a.0 as i64],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(add_images(c, a, &[img(1)]).unwrap(), 0);
        let after: i64 = c
            .query_row(
                "SELECT revision FROM collections WHERE id = ?1",
                [a.0 as i64],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(before, after);
    }

    #[test]
    fn a_collection_cannot_be_parented_under_its_own_descendant() {
        // A cycle is unbounded recursion in the tree walk, so it must not be
        // storable at all.
        let cat = seeded();
        let c = cat.connection();
        let top = create(c, "Top", None, CollectionKind::Manual).unwrap();
        let mid = create(c, "Mid", Some(top), CollectionKind::Manual).unwrap();
        let leaf = create(c, "Leaf", Some(mid), CollectionKind::Manual).unwrap();

        assert!(matches!(
            set_parent(c, top, Some(leaf)),
            Err(CatalogError::CollectionCycle(_))
        ));
        // And the direct case.
        assert!(matches!(
            set_parent(c, top, Some(top)),
            Err(CatalogError::CollectionCycle(_))
        ));
        // The tree is unchanged and still walks.
        assert_eq!(tree(c).unwrap().len(), 3);
    }

    #[test]
    fn reparenting_to_the_top_level_is_allowed() {
        let cat = seeded();
        let c = cat.connection();
        let top = create(c, "Top", None, CollectionKind::Manual).unwrap();
        let child = create(c, "Child", Some(top), CollectionKind::Manual).unwrap();

        set_parent(c, child, None).unwrap();
        let rows = tree(c).unwrap();
        assert!(rows.iter().all(|r| r.depth == 0));
        assert!(!rows.iter().any(|r| r.has_children));
    }

    #[test]
    fn deleting_a_parent_promotes_its_children_rather_than_taking_them() {
        // The schema's ON DELETE CASCADE would remove the whole subtree.
        // Losing a nested collection because its container was tidied away is
        // not recoverable.
        let cat = seeded();
        let c = cat.connection();
        let top = create(c, "Top", None, CollectionKind::Manual).unwrap();
        let mid = create(c, "Mid", Some(top), CollectionKind::Manual).unwrap();
        add_images(c, mid, &[img(1)]).unwrap();

        delete(c, top).unwrap();

        let rows = tree(c).unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].collection.id, mid);
        assert_eq!(rows[0].depth, 0, "promoted to the top level");
        assert_eq!(rows[0].collection.direct_count, 1, "members survived");
    }

    #[test]
    fn a_deleted_collection_leaves_a_tombstone() {
        // Without it, merging with a device that still holds the collection
        // resurrects it (§8.1).
        let cat = seeded();
        let c = cat.connection();
        let a = create(c, "Gone", None, CollectionKind::Manual).unwrap();
        delete(c, a).unwrap();

        assert!(tree(c).unwrap().is_empty(), "hidden from the sidebar");
        let (deleted, rev): (i64, i64) = c
            .query_row(
                "SELECT deleted, revision FROM collections WHERE id = ?1",
                [a.0 as i64],
                |r| Ok((r.get(0)?, r.get(1)?)),
            )
            .unwrap();
        assert_eq!(deleted, 1, "the row survives as a tombstone");
        assert!(rev > 1, "deletion competes on revision like any other edit");
    }

    #[test]
    fn a_parent_counts_its_descendants_images_without_double_counting() {
        // A set of collections must not read as empty, and an image in both a
        // parent and a child is one photograph.
        let cat = seeded();
        let c = cat.connection();
        let trips = create(c, "Trips", None, CollectionKind::Manual).unwrap();
        let iceland = create(c, "Iceland", Some(trips), CollectionKind::Manual).unwrap();
        let japan = create(c, "Japan", Some(trips), CollectionKind::Manual).unwrap();

        add_images(c, trips, &[img(1)]).unwrap();
        add_images(c, iceland, &[img(1), img(2)]).unwrap();
        add_images(c, japan, &[img(3)]).unwrap();

        // Direct: only image 1. Deep: images 1, 2, 3 — with 1 counted once.
        assert_eq!(deep_count(c, trips).unwrap(), 3);
        assert_eq!(deep_count(c, iceland).unwrap(), 2);
    }

    #[test]
    fn moving_images_is_atomic_across_both_collections() {
        let cat = seeded();
        let c = cat.connection();
        let a = create(c, "From", None, CollectionKind::Manual).unwrap();
        let b = create(c, "To", None, CollectionKind::Manual).unwrap();
        add_images(c, a, &[img(1), img(2)]).unwrap();

        move_images(c, a, b, &[img(1)]).unwrap();

        assert_eq!(collections_for_image(c, img(1)).unwrap(), vec![b]);
        assert_eq!(collections_for_image(c, img(2)).unwrap(), vec![a]);
    }

    #[test]
    fn moving_onto_the_source_collection_changes_nothing() {
        // A remove-then-add would drop the images' manual positions.
        let cat = seeded();
        let c = cat.connection();
        let a = create(c, "A", None, CollectionKind::Manual).unwrap();
        add_images(c, a, &[img(1), img(2)]).unwrap();

        assert_eq!(move_images(c, a, a, &[img(1)]).unwrap(), 0);
        assert_eq!(collections_for_image(c, img(1)).unwrap(), vec![a]);
    }

    #[test]
    fn images_cannot_be_dropped_onto_a_smart_collection() {
        // Its membership *is* its selector; member rows would be a second,
        // silently ignored source of truth.
        let cat = seeded();
        let c = cat.connection();
        let s = create(c, "Five star", None, CollectionKind::Smart).unwrap();

        assert!(matches!(
            add_images(c, s, &[img(1)]),
            Err(CatalogError::SmartCollectionNotEditable(_))
        ));
    }

    #[test]
    fn a_smart_collection_cannot_reference_itself() {
        let cat = seeded();
        let c = cat.connection();
        let s = create(c, "Recursive", None, CollectionKind::Smart).unwrap();

        assert!(matches!(
            save_smart(c, s, &Selector::Collection(s)),
            Err(CatalogError::CollectionCycle(_))
        ));
    }

    #[test]
    fn a_smart_collection_cannot_reference_itself_through_another() {
        // The indirect case: A references B, and B is then pointed back at A.
        let cat = seeded();
        let c = cat.connection();
        let a = create(c, "A", None, CollectionKind::Smart).unwrap();
        let b = create(c, "B", None, CollectionKind::Smart).unwrap();

        save_smart(c, b, &Selector::Collection(a)).unwrap();
        assert!(matches!(
            save_smart(c, a, &Selector::Collection(b)),
            Err(CatalogError::CollectionCycle(_))
        ));
    }

    #[test]
    fn a_saved_selector_round_trips_into_the_tree() {
        let cat = seeded();
        let c = cat.connection();
        let s = create(c, "Keepers", None, CollectionKind::Smart).unwrap();
        let sel = Selector::Rating { min: 4 };
        save_smart(c, s, &sel).unwrap();

        let row = tree(c).unwrap().into_iter().next().unwrap();
        assert_eq!(row.collection.kind, CollectionKind::Smart);
        assert_eq!(row.collection.selector, Some(sel));
    }

    #[test]
    fn manual_order_can_be_rewritten() {
        let cat = seeded();
        let c = cat.connection();
        let a = create(c, "Sequence", None, CollectionKind::Manual).unwrap();
        add_images(c, a, &[img(1), img(2), img(3)]).unwrap();

        set_order(c, a, &[img(3), img(1), img(2)]).unwrap();

        let order: Vec<i64> = {
            let mut stmt = c
                .prepare(
                    "SELECT image_id FROM collection_members
                      WHERE collection_id = ?1 ORDER BY position",
                )
                .unwrap();
            stmt.query_map([a.0 as i64], |r| r.get(0))
                .unwrap()
                .map(Result::unwrap)
                .collect()
        };
        assert_eq!(order, vec![3, 1, 2]);
    }

    #[test]
    fn editing_a_missing_collection_is_an_error_not_a_silent_no_op() {
        let cat = seeded();
        let c = cat.connection();
        let ghost = CollectionId(999);

        assert!(matches!(
            rename(c, ghost, "x"),
            Err(CatalogError::NoSuchCollection(999))
        ));
        assert!(matches!(
            add_images(c, ghost, &[img(1)]),
            Err(CatalogError::NoSuchCollection(999))
        ));
        assert!(matches!(
            create(c, "child", Some(ghost), CollectionKind::Manual),
            Err(CatalogError::NoSuchCollection(999))
        ));
    }

    #[test]
    fn a_collection_under_a_tombstoned_parent_stays_reachable() {
        // A merge from another device can tombstone a parent this device still
        // has children under — `delete` promotes them, but a merge writes the
        // tombstone directly. Dropping the child from the tree would make the
        // images inside invisible with no way to reach them.
        let cat = seeded();
        let c = cat.connection();
        let top = create(c, "Top", None, CollectionKind::Manual).unwrap();
        let child = create(c, "Child", Some(top), CollectionKind::Manual).unwrap();
        add_images(c, child, &[img(1)]).unwrap();

        // What a merge does: tombstone the parent without touching children.
        c.execute(
            "UPDATE collections SET deleted = 1 WHERE id = ?1",
            [top.0 as i64],
        )
        .unwrap();

        let rows = tree(c).unwrap();
        assert_eq!(rows.len(), 1, "the child survives its parent");
        assert_eq!(rows[0].collection.id, child);
        assert_eq!(rows[0].depth, 0, "shown at the top level");
        assert_eq!(rows[0].collection.direct_count, 1);
    }

    #[test]
    fn a_cyclic_tree_from_a_merge_does_not_hang_the_walk() {
        // The write path refuses cycles, but a merged-in row from another
        // device was never validated here. The read path must terminate.
        let cat = seeded();
        let c = cat.connection();
        let a = create(c, "A", None, CollectionKind::Manual).unwrap();
        let b = create(c, "B", Some(a), CollectionKind::Manual).unwrap();
        // Force the cycle behind the API's back.
        c.execute(
            "UPDATE collections SET parent_id = ?2 WHERE id = ?1",
            rusqlite::params![a.0 as i64, b.0 as i64],
        )
        .unwrap();

        // Neither walk may recurse forever.
        let rows = tree(c).unwrap();
        assert!(rows.len() <= 2);
        let d = descendants(c, a).unwrap();
        assert!(d.len() <= 2);
    }

    #[test]
    fn membership_says_how_many_of_the_selection_each_collection_holds() {
        // The count is the whole point: "3 of 40" is what stops a user taking
        // forty photographs out of a collection thirty-seven were never in.
        let cat = seeded();
        let c = cat.connection();
        let iceland = create(c, "Iceland", None, CollectionKind::Manual).unwrap();
        let best = create(c, "Best", None, CollectionKind::Manual).unwrap();
        add_images(c, iceland, &[img(1), img(2), img(3)]).unwrap();
        add_images(c, best, &[img(1)]).unwrap();

        let m = membership_of(c, &[img(1), img(2), img(3), img(4)]).unwrap();
        assert_eq!(m.len(), 2);
        // Sorted by name, so "Best" comes before "Iceland".
        assert_eq!(m[0].id, best);
        assert_eq!(m[0].holding, 1);
        assert_eq!(m[1].id, iceland);
        assert_eq!(m[1].holding, 3);
    }

    #[test]
    fn membership_omits_collections_holding_none_of_them() {
        // A row offering to remove images that are not there would be a button
        // that does nothing, which is worse than an absent one.
        let cat = seeded();
        let c = cat.connection();
        let a = create(c, "A", None, CollectionKind::Manual).unwrap();
        create(c, "Empty", None, CollectionKind::Manual).unwrap();
        add_images(c, a, &[img(1)]).unwrap();

        let m = membership_of(c, &[img(1)]).unwrap();
        assert_eq!(m.len(), 1);
        assert_eq!(m[0].id, a);
    }

    #[test]
    fn membership_of_nothing_is_nothing() {
        let cat = seeded();
        assert!(membership_of(cat.connection(), &[]).unwrap().is_empty());
    }

    #[test]
    fn membership_skips_a_deleted_collection() {
        // The tombstone survives the delete, and its member rows are dropped —
        // but a merge can leave rows behind, and the sheet must not offer a
        // collection the sidebar does not draw.
        let cat = seeded();
        let c = cat.connection();
        let a = create(c, "A", None, CollectionKind::Manual).unwrap();
        add_images(c, a, &[img(1)]).unwrap();
        delete(c, a).unwrap();

        assert!(membership_of(c, &[img(1)]).unwrap().is_empty());
    }

    #[test]
    fn membership_sums_across_chunks() {
        // The chunking exists for a select-all, which is exactly the gesture
        // that would otherwise exceed SQLite's parameter cap. A count that was
        // per-chunk rather than summed would under-report on the one selection
        // large enough to need it.
        let cat = seeded();
        let c = cat.connection();
        let a = create(c, "A", None, CollectionKind::Manual).unwrap();
        // Past the fixture's six, so the member rows have images to point at.
        for i in 7..=900i64 {
            c.execute(
                "INSERT INTO images(id, root_id, source_ref, added_at)
                 VALUES (?1, 1, ?2, 0)",
                rusqlite::params![i, format!("img{i}.CR3")],
            )
            .unwrap();
        }
        let ids: Vec<ImageId> = (1..=900).map(img).collect();
        add_images(c, a, &ids).unwrap();

        let m = membership_of(c, &ids).unwrap();
        assert_eq!(m.len(), 1);
        assert_eq!(m[0].holding, 900, "counted across every chunk");
    }
}