DarkRoom/core/dr-catalog/src/rating.rs

//! TRACES: FR-CAT-5 | FR-CAT-6 | FR-CULL-4
//! Star ratings and pick/reject flags — the judgement a cull produces.
//!
//! # Why this hangs off `versions` rather than `images`
//!
//! The schema already carries `rating`, `label` and `flag` on `versions`, and
//! [`crate::query`] already compiles [`dr_types::Selector::Rating`] and
//! [`dr_types::Selector::Flag`] against the *default* version. What was
//! missing is that nothing ever created a version row: a scan inserts into
//! `images` and stops, so every image had no version, and therefore nowhere
//! to record a rating. The whole library sat permanently unrated with no way
//! out of that state.
//!
//! So this module's first job is [`ensure_default_versions`] — every image
//! gets exactly one default version, created at scan time and backfilled by
//! the v2 migration for libraries scanned before this existed.
//!
//! Keeping judgement on the version rather than the image is what makes
//! FR-CAT-12's virtual copies coherent: two crops of one frame are two
//! photographs to the photographer, and one may be a keeper while the other
//! is a reject. Hoisting the rating onto the image would force them to agree.
//!
//! # Unrated is a real state, not a zero
//!
//! `rating = 0` means *not yet judged*, and that is precisely what "filter to
//! unjudged" selects (FR-CULL-4). It is deliberately not conflated with "one
//! star" or with "rejected" — those are three different answers, and a cull
//! that cannot distinguish "I have not looked at this" from "I looked and it
//! is poor" cannot be resumed.

use rusqlite::{Connection, OptionalExtension};

use dr_types::{ColourLabel, FlagState, ImageId};

use crate::error::CatalogError;

/// Highest star rating. Five, as every photo tool has settled on.
pub const MAX_RATING: u8 = 5;

/// Name given to the version created for an image that has none.
///
/// Matches what [`crate::collections`] and the sidecar both expect to see for
/// the original, unmodified frame.
pub const DEFAULT_VERSION_NAME: &str = "Default";

/// The judgement recorded against one image's default version.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct Judgement {
    /// 0..=5. Zero means *unrated*, which is a state in its own right.
    pub rating: u8,
    pub flag: FlagState,
    /// TRACES: FR-CAT-5
    /// The colour label, or `None`. Not part of [`Judgement::is_judged`]:
    /// a label sorts photographs into piles of the photographer's own
    /// meaning — "to print", "send to Anna" — and says nothing about whether
    /// a frame has been culled, which is the question "unjudged" asks.
    pub label: Option<ColourLabel>,
}

impl Judgement {
    /// Whether this image has been judged at all.
    ///
    /// Either axis counts: a photographer who flags without starring, or stars
    /// without flagging, has still made a decision about the frame. "Filter to
    /// unjudged" (FR-CULL-4) is the negation of this, and getting it wrong
    /// means a resumed session re-presents work already done.
    pub fn is_judged(self) -> bool {
        self.rating > 0 || self.flag != FlagState::Unflagged
    }
}

/// TRACES: FR-NC-8 | FR-NC-9
/// The default version's uuid for a photograph the server knows by `file_id`.
///
/// # Why this is derived and not generated
///
/// A version's uuid is the identity a cross-device merge keys on. It used to
/// be minted at random per row, and the comment above this function used to
/// say that made it unique — which it did, and that was precisely the bug.
/// Two devices indexing the same library minted *different* uuids for the same
/// photograph, so the sidecar they shared ended up with two `default = 1`
/// blocks, `Version::merge` never saw a matching pair to reconcile, and a
/// rating made on one device was invisible on the other. `crate::merge` has
/// documented the consequence for keywords for as long as it has existed: a
/// uuid-keyed join across two catalogs unions nothing at all.
///
/// `oc:fileid` is the identity that *is* shared. The server assigns it, every
/// client pointed at that library sees the same integer, and it survives a
/// server-side rename and move — the same three properties that made
/// `crate::merge::ASSIGN_BY_FILE_ID` prefer it to a content hash.
///
/// # The layout
///
/// A UUIDv8 (RFC 9562: an application-defined layout) carrying the file id
/// verbatim across the four variable fields, with a fixed tag in the node
/// field saying what minted it. Verbatim rather than hashed so the mapping is
/// injective by construction: two file ids cannot collide, which a truncated
/// hash could, and a uuid read out of a sidecar can be traced back to the file
/// it belongs to by eye.
///
/// Every device computes this identically from the same integer, which is the
/// whole point — there is no negotiation and no first-writer-wins.
pub fn derived_version_uuid(file_id: i64) -> String {
    let id = file_id as u64;
    format!(
        // 32 + 16 + 12 + 4 = 64 bits of file id, then the tag.
        "{:08x}-{:04x}-8{:03x}-{:04x}-{:012x}",
        (id >> 32) as u32,
        (id >> 16) as u16,
        (id >> 4) as u16 & 0x0FFF,
        // The two high bits are the RFC's variant field and must be `0b10`;
        // the remaining fourteen carry the file id's last four bits.
        0x8000u16 | ((id as u16 & 0x000F) << 10),
        DERIVED_VERSION_TAG,
    )
}

/// The node field of a [`derived_version_uuid`], identifying what minted it.
///
/// Fixed and arbitrary. Its only job is to keep a derived uuid from colliding
/// with a randomly minted one and to make it recognisable in a sidecar read by
/// eye — `…-d0c5ec0de001` is visibly not a v4.
const DERIVED_VERSION_TAG: u64 = 0xd0c5_ec0d_e001;

/// Give every image without one a default version.
///
/// Idempotent, and cheap on the common path: the `NOT EXISTS` sub-select is
/// answered by the `versions_image` index, so a library that already has its
/// versions costs one indexed scan and writes nothing.
///
/// Returns how many were created, so a scan can log the backfill rather than
/// silently doing thousands of inserts.
///
/// The uuid comes from [`derived_version_uuid`] where the server has named the
/// file, so every device computes the same one; only a library with no server
/// behind it falls back to a generated id.
pub fn ensure_default_versions(conn: &Connection) -> Result<usize, CatalogError> {
    // One transaction for the batch. A backfill over a 24k-image library is
    // 24k inserts, and per-statement commits would make it minutes rather
    // than seconds.
    let tx = conn.unchecked_transaction()?;
    let n = ensure_default_versions_within(&tx)?;
    tx.commit()?;
    Ok(n)
}

/// [`ensure_default_versions`] without opening a transaction.
///
/// Separate because SQLite has no nested `BEGIN`: [`crate::merge`] needs the
/// invariant restored *inside* the merge transaction — an incoming keyword
/// lands on a default version, so an image without one would silently drop it —
/// and calling the public form there fails at runtime with "cannot start a
/// transaction within a transaction". The same split, for the same reason, as
/// `collections::add_within`.
pub fn ensure_default_versions_within(conn: &Connection) -> Result<usize, CatalogError> {
    // The remote id travels with the image so the uuid can be derived from it.
    // A `LEFT JOIN`, because a library on a folder or a card has no `remote`
    // row at all and still needs its versions.
    let ids: Vec<(i64, Option<i64>)> = {
        let mut stmt = conn.prepare(
            "SELECT i.id, r.file_id FROM images i
             LEFT JOIN remote r ON r.image_id = i.id
              WHERE NOT EXISTS (SELECT 1 FROM versions v WHERE v.image_id = i.id)",
        )?;
        let found = stmt
            .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?
            .collect::<Result<Vec<_>, _>>()?;
        found
    };
    if ids.is_empty() {
        return Ok(0);
    }

    {
        let mut insert = conn.prepare(
            "INSERT INTO versions(image_id, uuid, name, is_default, rating, flag)
             VALUES (?1, ?2, ?3, 1, 0, 0)",
        )?;
        for (id, file_id) in &ids {
            insert.execute(rusqlite::params![
                id,
                version_uuid(*file_id),
                DEFAULT_VERSION_NAME
            ])?;
        }
    }

    Ok(ids.len())
}

/// The uuid to mint for a new default version.
///
/// Derived from the server's file id where there is one, so two devices agree
/// (FR-NC-8); generated where there is not.
///
/// # What the fallback costs
///
/// A library with no server behind it — a folder, a card — has no identity two
/// devices could both compute, so the split this derivation prevents is still
/// reachable there if that folder is synced by something else. That case is
/// repaired rather than prevented: `Sidecar::fuse_default_versions` folds the
/// rival defaults together the next time either device reads the file.
fn version_uuid(file_id: Option<i64>) -> String {
    match file_id {
        Some(id) => derived_version_uuid(id),
        None => new_uuid(),
    }
}

/// TRACES: FR-NC-8 | FR-NC-9
/// Move default versions minted before [`derived_version_uuid`] onto it.
///
/// Every catalog written by an earlier build holds a randomly minted uuid per
/// image, and its peers hold different ones for the same photographs. Deriving
/// the uuid only for *new* rows would leave every image already indexed —
/// which is all of them, on a library anybody has used — writing to the same
/// rival identity it always did.
///
/// Safe to run repeatedly: it selects only rows whose uuid is not already the
/// derived one, so a realigned catalog matches nothing and writes nothing.
///
/// # Why this cannot collide
///
/// `versions.uuid` is `UNIQUE`, and `remote.file_id` has a unique index of its
/// own, so two images cannot derive the same uuid. The one row that could
/// stand in the way is a *virtual copy* (FR-CAT-12) that already holds the
/// target — impossible to mint but not impossible to receive from a merge — so
/// the update is skipped where the target is taken rather than failing the
/// backfill and, with it, the catalog open.
///
/// # The sidecar side is not this function's business
///
/// Moving the catalog's uuid alone would leave the file's default under the
/// old one and the next write would add a rival rather than amend it. What
/// stops that is `library::amend` fusing onto the write's uuid before it looks
/// anything up, which renames the file's default to match. This end and that
/// one have to land together, and they do.
///
/// Returns how many rows moved.
pub fn align_default_version_uuids(conn: &Connection) -> Result<usize, CatalogError> {
    // Filtered in SQL rather than in the loop: every derived uuid ends in the
    // tag, so a catalog that has already been realigned selects no rows at all
    // and this costs one indexed pass instead of twenty-four thousand reads.
    let already = format!("%-{DERIVED_VERSION_TAG:012x}");
    let stale: Vec<(i64, i64)> = {
        let mut stmt = conn.prepare(
            "SELECT v.id, r.file_id
               FROM versions v
               JOIN remote r ON r.image_id = v.image_id
              WHERE v.is_default = 1 AND v.uuid NOT LIKE ?1",
        )?;
        let found = stmt
            .query_map([&already], |r| Ok((r.get(0)?, r.get(1)?)))?
            .collect::<Result<Vec<_>, _>>()?;
        found
    };
    if stale.is_empty() {
        return Ok(0);
    }

    let tx = conn.unchecked_transaction()?;
    let mut moved = 0usize;
    {
        // `OR IGNORE` covers the taken-target case described above: the row
        // keeps the uuid it has, which is the state this build has always
        // coped with, rather than aborting the transaction.
        let mut update = tx.prepare("UPDATE OR IGNORE versions SET uuid = ?2 WHERE id = ?1")?;
        for (row, file_id) in &stale {
            moved += update.execute(rusqlite::params![row, derived_version_uuid(*file_id)])?;
        }
    }
    tx.commit()?;
    Ok(moved)
}

/// TRACES: FR-CAT-13
/// How `versions.label` encodes a colour label, and back.
///
/// One place for both directions, so a label written by the XMP pull and a
/// label queried by the selector cannot drift apart: the query used to hold
/// its own copy of the forward mapping and nothing held the reverse.
pub fn label_code(l: ColourLabel) -> i64 {
    match l {
        ColourLabel::Red => 1,
        ColourLabel::Yellow => 2,
        ColourLabel::Green => 3,
        ColourLabel::Blue => 4,
        ColourLabel::Purple => 5,
    }
}

/// The colour a `versions.label` value names, or `None` for NULL and for a
/// code this build does not know.
pub fn label_from_code(code: Option<i64>) -> Option<ColourLabel> {
    Some(match code? {
        1 => ColourLabel::Red,
        2 => ColourLabel::Yellow,
        3 => ColourLabel::Green,
        4 => ColourLabel::Blue,
        5 => ColourLabel::Purple,
        _ => return None,
    })
}

/// The default version's row id for an image, creating one if it has none.
///
/// Every write path goes through this rather than assuming a version exists.
/// An image can arrive without one in two ways that are not worth trying to
/// prevent: a row inserted by a build predating this module, and a scan whose
/// version pass was interrupted between the image insert and the commit.
/// Failing a rating because of either would be the wrong answer — the user
/// pressed a key and expects a star.
pub fn default_version_id(conn: &Connection, image: ImageId) -> Result<i64, CatalogError> {
    let existing: Option<i64> = conn
        .query_row(
            "SELECT id FROM versions
              WHERE image_id = ?1
              ORDER BY is_default DESC, id ASC
              LIMIT 1",
            [image.0 as i64],
            |r| r.get(0),
        )
        .optional()?;

    if let Some(id) = existing {
        return Ok(id);
    }

    // Derived from the server's file id where there is one, so the version
    // this mints is the same one the photographer's other device will mint
    // (FR-NC-8). A miss here is a library with no server behind it.
    let file_id: Option<i64> = conn
        .query_row(
            "SELECT file_id FROM remote WHERE image_id = ?1",
            [image.0 as i64],
            |r| r.get(0),
        )
        .optional()?;

    conn.execute(
        "INSERT INTO versions(image_id, uuid, name, is_default, rating, flag)
         VALUES (?1, ?2, ?3, 1, 0, 0)",
        rusqlite::params![image.0 as i64, version_uuid(file_id), DEFAULT_VERSION_NAME],
    )?;
    Ok(conn.last_insert_rowid())
}

/// Set the star rating for one image, clamped to 0..=[`MAX_RATING`].
///
/// Clamped rather than rejected: the value comes from a keystroke or a click
/// on a star strip, and there is no useful error to show a photographer who
/// pressed a key. Out of range can only mean a UI bug, and losing the
/// keystroke would be a worse symptom than recording five.
pub fn set_rating(conn: &Connection, image: ImageId, rating: u8) -> Result<(), CatalogError> {
    let version = default_version_id(conn, image)?;
    conn.execute(
        "UPDATE versions SET rating = ?2 WHERE id = ?1",
        rusqlite::params![version, rating.min(MAX_RATING) as i64],
    )?;
    Ok(())
}

/// Set the pick/reject flag for one image.
pub fn set_flag(conn: &Connection, image: ImageId, flag: FlagState) -> Result<(), CatalogError> {
    let version = default_version_id(conn, image)?;
    conn.execute(
        "UPDATE versions SET flag = ?2 WHERE id = ?1",
        rusqlite::params![version, flag_code(flag)],
    )?;
    Ok(())
}

/// Apply a rating to many images in one transaction.
///
/// The bulk path exists because rating a selection is one gesture: the user
/// selects forty frames and presses `3`. Forty separate transactions would be
/// forty fsyncs for what is conceptually a single edit, and a crash partway
/// through would leave the selection half-rated.
pub fn set_rating_many(
    conn: &Connection,
    images: &[ImageId],
    rating: u8,
) -> Result<usize, CatalogError> {
    apply_many(conn, images, |conn, id| set_rating(conn, id, rating))
}

/// Apply a flag to many images in one transaction. See [`set_rating_many`].
pub fn set_flag_many(
    conn: &Connection,
    images: &[ImageId],
    flag: FlagState,
) -> Result<usize, CatalogError> {
    apply_many(conn, images, |conn, id| set_flag(conn, id, flag))
}

/// TRACES: FR-CAT-5
/// Set or clear the colour label for one image.
pub fn set_label(
    conn: &Connection,
    image: ImageId,
    label: Option<ColourLabel>,
) -> Result<(), CatalogError> {
    let version = default_version_id(conn, image)?;
    conn.execute(
        "UPDATE versions SET label = ?2 WHERE id = ?1",
        rusqlite::params![version, label.map(label_code)],
    )?;
    Ok(())
}

/// TRACES: FR-CAT-5
/// Set or clear a label on many images in one transaction — one keystroke
/// over a selection is one commit, as for [`set_rating_many`].
pub fn set_label_many(
    conn: &Connection,
    images: &[ImageId],
    label: Option<ColourLabel>,
) -> Result<usize, CatalogError> {
    apply_many(conn, images, |conn, id| set_label(conn, id, label))
}

/// TRACES: FR-CAT-5
/// What a label key does to a set of images: Lightroom's toggle.
///
/// Pressing the key for the label every one of them already carries takes it
/// off; otherwise every one of them gets it. Decided over the whole set
/// rather than per image, so a selection that was half red comes out all red
/// rather than inverted — the photographer pressed "red", and a key that
/// turned half of them red and the other half plain would be two answers to
/// one question.
pub fn toggled_label(
    current: impl IntoIterator<Item = Option<ColourLabel>>,
    pressed: ColourLabel,
) -> Option<ColourLabel> {
    let mut any = false;
    for label in current {
        any = true;
        if label != Some(pressed) {
            return Some(pressed);
        }
    }
    if any {
        None
    } else {
        Some(pressed)
    }
}

/// TRACES: FR-CAT-5 | FR-CAT-6
/// How the library divides by colour label, for the filter chips' counts.
///
/// Index 0 is unlabelled and index `n` the label whose code is `n`. The
/// same shape as [`rating_histogram`], and for the same reason the
/// unlabelled slot is what is left of [`judged_rows`]: an image without a
/// version row is unlabelled, not missing.
///
/// Only labelled rows are grouped. The join this replaced (2026-09-26)
/// probed `versions_judgement` per image and then read each version's row
/// for `label`, which the index does not carry -- 10 ms on the reference
/// library, on every label keystroke, to find that none of 23,500 images
/// had one. This walks the default versions in the index's order, which
/// is close to the table's, and groups the few that are labelled.
pub fn label_histogram(conn: &Connection) -> Result<[usize; 6], CatalogError> {
    let mut out = [0usize; 6];
    let mut stmt = conn.prepare_cached(
        "SELECT label, count(*) FROM versions
          WHERE is_default = 1 AND label IS NOT NULL
          GROUP BY label",
    )?;
    let rows = stmt.query_map([], |r| Ok((r.get::<_, i64>(0)?, r.get::<_, i64>(1)?)))?;
    let mut counted = 0usize;
    for (code, count) in rows.flatten() {
        // A code this build does not know counts as unlabelled, which is how
        // `label_from_code` reads it everywhere else.
        let slot = if label_from_code(Some(code)).is_some() {
            code as usize
        } else {
            0
        };
        out[slot] += count as usize;
        counted += count as usize;
    }
    out[0] += judged_rows(conn)?.saturating_sub(counted);
    Ok(out)
}

/// Shared bulk wrapper, so the two axes cannot drift in their commit
/// behaviour — a partially-committed rating and a fully-committed flag from
/// the same keystroke would be hard to explain and harder to notice.
fn apply_many(
    conn: &Connection,
    images: &[ImageId],
    mut one: impl FnMut(&Connection, ImageId) -> Result<(), CatalogError>,
) -> Result<usize, CatalogError> {
    if images.is_empty() {
        return Ok(0);
    }
    let tx = conn.unchecked_transaction()?;
    for id in images {
        one(&tx, *id)?;
    }
    tx.commit()?;
    Ok(images.len())
}

/// Read the judgement for one image.
///
/// An image with no version reads as unrated and unflagged rather than as an
/// error: that is exactly what it is.
pub fn judgement(conn: &Connection, image: ImageId) -> Result<Judgement, CatalogError> {
    let row: Option<(i64, i64, Option<i64>)> = conn
        .query_row(
            "SELECT rating, flag, label FROM versions
              WHERE image_id = ?1
              ORDER BY is_default DESC, id ASC
              LIMIT 1",
            [image.0 as i64],
            |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
        )
        .optional()?;

    Ok(match row {
        Some((rating, flag, label)) => Judgement {
            rating: rating.clamp(0, MAX_RATING as i64) as u8,
            flag: flag_from_code(flag),
            label: label_from_code(label),
        },
        None => Judgement::default(),
    })
}

/// Judgements for a window of images, in one statement.
///
/// The grid needs a star strip per cell, and one query per cell would be 120
/// round trips on every scroll — the same reasoning as
/// `collections_ui::sync_badges`. Images with no version simply do not appear
/// in the result, and the caller treats a miss as unrated.
pub fn judgements(
    conn: &Connection,
    images: &[ImageId],
) -> Result<std::collections::HashMap<ImageId, Judgement>, CatalogError> {
    let mut out = std::collections::HashMap::new();
    if images.is_empty() {
        return Ok(out);
    }

    // Placeholders are generated from the *count* of ids, never from any text
    // that came from outside — the same rule `read_cells_scoped` follows.
    let placeholders = std::iter::repeat_n("?", images.len())
        .collect::<Vec<_>>()
        .join(",");
    let sql = format!(
        "SELECT image_id, rating, flag, label FROM versions
          WHERE image_id IN ({placeholders}) AND is_default = 1"
    );

    let params: Vec<rusqlite::types::Value> = images
        .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((
            r.get::<_, i64>(0)?,
            r.get::<_, i64>(1)?,
            r.get::<_, i64>(2)?,
            r.get::<_, Option<i64>>(3)?,
        ))
    })?;

    for (image, rating, flag, label) in rows.flatten() {
        out.insert(
            ImageId(image as u64),
            Judgement {
                rating: rating.clamp(0, MAX_RATING as i64) as u8,
                flag: flag_from_code(flag),
                label: label_from_code(label),
            },
        );
    }
    Ok(out)
}

/// How the library divides by rating, for the filter bar's counts.
///
/// Index `n` is the number of images rated `n`, so index 0 is the unrated
/// count. Shown beside each filter button so the user can see there is
/// something behind it before narrowing to it — a filter that silently
/// empties the grid reads as a broken filter.
pub fn rating_histogram(conn: &Connection) -> Result<[usize; 6], CatalogError> {
    let mut out = [0usize; 6];

    // Only the rated rows are grouped; the unrated slot is what is left of
    // [`judged_rows`]. So an image whose version row is missing still counts
    // as unrated rather than vanishing from the totals -- the histogram has
    // to sum to the library size or it is not believable.
    //
    // It was one `images LEFT JOIN versions ... GROUP BY` until 2026-09-26:
    // a probe of `versions_judgement` per image and a sort of every row, to
    // put 22,000 of 23,500 in slot zero. 9 ms on every star keystroke on the
    // reference library; this is a pass over the index that sorts only the
    // rated few, and [`judged_rows`] is three index-only counts.
    let mut stmt = conn.prepare_cached(
        "SELECT rating, count(*) FROM versions
          WHERE is_default = 1 AND rating != 0
          GROUP BY rating",
    )?;
    let rows = stmt.query_map([], |r| Ok((r.get::<_, i64>(0)?, r.get::<_, i64>(1)?)))?;

    let mut counted = 0usize;
    for (rating, count) in rows.flatten() {
        if let Some(slot) = out.get_mut(rating.clamp(0, MAX_RATING as i64) as usize) {
            *slot += count as usize;
            counted += count as usize;
        }
    }
    out[0] += judged_rows(conn)?.saturating_sub(counted);
    Ok(out)
}

/// How many rows `images LEFT JOIN versions ON ... AND is_default = 1` has:
/// one per image with no default version, and one per default version for
/// the rest. The total both histograms divide up, and the unjudged slot is
/// what is left of it once the judged rows are counted.
///
/// Spelled as three counts rather than as that join because the join probes
/// `versions_judgement` once per image, where each count here is one pass
/// over an index without reading a row: the library size, the default
/// versions, and the images holding one. An image with two default versions
/// -- nothing prevents it -- is two rows of the join and one image of the
/// third count, so it adds one here exactly as it did there. A version
/// always belongs to an image; `foreign_keys` is on and deletes cascade.
///
/// `count(DISTINCT image_id)` alone in its statement: that is what lets
/// SQLite read the distinct values off the index's order instead of
/// building a temporary b-tree of them.
fn judged_rows(conn: &Connection) -> Result<usize, CatalogError> {
    let n: i64 = conn
        .prepare_cached(
            "SELECT (SELECT count(*) FROM images)
                  + (SELECT count(*) FROM versions WHERE is_default = 1)
                  - (SELECT count(DISTINCT image_id) FROM versions WHERE is_default = 1)",
        )?
        .query_row([], |r| r.get(0))?;
    Ok(n.max(0) as usize)
}

/// How many images carry each flag: `(picks, rejects)`.
pub fn flag_counts(conn: &Connection) -> Result<(usize, usize), CatalogError> {
    let picks: i64 = conn.query_row(
        "SELECT count(*) FROM versions WHERE is_default = 1 AND flag = 1",
        [],
        |r| r.get(0),
    )?;
    let rejects: i64 = conn.query_row(
        "SELECT count(*) FROM versions WHERE is_default = 1 AND flag = 2",
        [],
        |r| r.get(0),
    )?;
    Ok((picks as usize, rejects as usize))
}

/// The stored integer for a flag. Matches [`crate::query::flag_code`]'s
/// mapping — the two must agree or a filter will not find what a write stored.
fn flag_code(f: FlagState) -> i64 {
    match f {
        FlagState::Unflagged => 0,
        FlagState::Pick => 1,
        FlagState::Reject => 2,
    }
}

fn flag_from_code(v: i64) -> FlagState {
    match v {
        1 => FlagState::Pick,
        2 => FlagState::Reject,
        _ => FlagState::Unflagged,
    }
}

/// A generated version UUID, for a photograph no server has named.
///
/// Hand-rolled rather than pulling in the `uuid` crate for one function — the
/// same reasoning as the date maths in `library_ui`. It needs to be unique,
/// not cryptographically unguessable: it keys a merge, and an attacker who can
/// write to the sidecar has already won.
///
/// Seeded from the system clock and a per-process counter, so two versions
/// created inside the same nanosecond tick still differ.
///
/// **Unique is not the same as agreed**, which is the distinction that cost a
/// photographer a day of culling. Two devices calling this for the same
/// photograph get two different answers, and a merge keyed on the result then
/// has no pair to reconcile. Anything with a `file_id` behind it must use
/// [`derived_version_uuid`]; this is the fallback for libraries that have no
/// server to supply one.
fn new_uuid() -> String {
    use std::sync::atomic::{AtomicU64, Ordering};
    static COUNTER: AtomicU64 = AtomicU64::new(0);

    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos() as u64)
        .unwrap_or(0);
    let n = COUNTER.fetch_add(1, Ordering::Relaxed);

    // Mixed so successive ids do not share a long common prefix, which makes
    // them easier to tell apart when reading a sidecar by eye.
    let a = nanos ^ (n.wrapping_mul(0x9E37_79B9_7F4A_7C15));
    let b = nanos
        .rotate_left(32)
        .wrapping_add(n.wrapping_mul(0xBF58_476D_1CE4_E5B9));

    format!(
        "{:08x}-{:04x}-4{:03x}-{:04x}-{:012x}",
        (a >> 32) as u32,
        (a >> 16) as u16,
        (a & 0x0FFF) as u16,
        // Variant bits, so this is a well-formed v4-shaped UUID rather than
        // something that merely looks like one.
        ((b >> 48) as u16 & 0x3FFF) | 0x8000,
        b & 0xFFFF_FFFF_FFFF,
    )
}

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

    /// A catalog holding `n` images and nothing else — the state a scan
    /// leaves behind before this module runs.
    fn with_images(n: usize) -> 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 0..n {
            c.execute(
                "INSERT INTO images(root_id, source_ref, added_at) VALUES (1, ?1, 0)",
                [format!("img{i:03}.CR3")],
            )
            .unwrap();
        }
        cat
    }

    fn ids(cat: &Catalog) -> Vec<ImageId> {
        let mut stmt = cat
            .connection()
            .prepare("SELECT id FROM images ORDER BY id")
            .unwrap();
        stmt.query_map([], |r| Ok(ImageId(r.get::<_, i64>(0)? as u64)))
            .unwrap()
            .map(Result::unwrap)
            .collect()
    }

    #[test]
    fn every_scanned_image_gets_a_default_version() {
        // The gap this module exists to close: a scan inserted images and no
        // versions, so there was nowhere for a rating to go.
        let cat = with_images(5);
        assert_eq!(ensure_default_versions(cat.connection()).unwrap(), 5);

        let n: i64 = cat
            .connection()
            .query_row(
                "SELECT count(*) FROM versions WHERE is_default = 1",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n, 5);
    }

    #[test]
    fn images_enter_unrated_rather_than_at_one_star() {
        // "Not yet judged" is the state a cull starts from and resumes to.
        let cat = with_images(3);
        ensure_default_versions(cat.connection()).unwrap();

        for id in ids(&cat) {
            let j = judgement(cat.connection(), id).unwrap();
            assert_eq!(j.rating, 0);
            assert_eq!(j.flag, FlagState::Unflagged);
            assert!(!j.is_judged());
        }
    }

    #[test]
    fn backfilling_twice_creates_nothing_the_second_time() {
        // Runs on every scan, so a second pass must not double every version.
        let cat = with_images(4);
        assert_eq!(ensure_default_versions(cat.connection()).unwrap(), 4);
        assert_eq!(ensure_default_versions(cat.connection()).unwrap(), 0);

        let n: i64 = cat
            .connection()
            .query_row("SELECT count(*) FROM versions", [], |r| r.get(0))
            .unwrap();
        assert_eq!(n, 4, "one version per image, not two");
    }

    #[test]
    fn version_uuids_are_unique_across_a_batch() {
        // The uuid is the cross-device merge identity: two images sharing one
        // silently fuse their edits at the next sync.
        let cat = with_images(200);
        ensure_default_versions(cat.connection()).unwrap();

        let distinct: i64 = cat
            .connection()
            .query_row("SELECT count(DISTINCT uuid) FROM versions", [], |r| {
                r.get(0)
            })
            .unwrap();
        assert_eq!(distinct, 200);
    }

    #[test]
    fn a_label_round_trips_and_clears() {
        // TRACES: FR-CAT-5
        let cat = with_images(1);
        let id = ids(&cat)[0];
        set_label(cat.connection(), id, Some(ColourLabel::Green)).unwrap();
        assert_eq!(
            judgement(cat.connection(), id).unwrap().label,
            Some(ColourLabel::Green)
        );
        set_label(cat.connection(), id, None).unwrap();
        assert_eq!(judgement(cat.connection(), id).unwrap().label, None);
    }

    #[test]
    fn a_label_is_not_a_judgement() {
        // "Unjudged" is the cull's resume point; a label is a pile of the
        // photographer's own, and labelling a frame must not hide it there.
        let cat = with_images(1);
        let id = ids(&cat)[0];
        set_label(cat.connection(), id, Some(ColourLabel::Red)).unwrap();
        assert!(!judgement(cat.connection(), id).unwrap().is_judged());
    }

    #[test]
    fn labelling_a_selection_is_one_commit_and_reaches_every_image() {
        // TRACES: FR-CAT-5
        let cat = with_images(4);
        let all = ids(&cat);
        assert_eq!(
            set_label_many(cat.connection(), &all, Some(ColourLabel::Blue)).unwrap(),
            4
        );
        let found = judgements(cat.connection(), &all).unwrap();
        assert!(all
            .iter()
            .all(|id| found[id].label == Some(ColourLabel::Blue)));
        assert_eq!(
            label_histogram(cat.connection()).unwrap(),
            [0, 0, 0, 0, 4, 0]
        );
    }

    #[test]
    fn a_label_key_toggles_only_when_every_image_already_has_it() {
        // TRACES: FR-CAT-5
        use ColourLabel::*;
        assert_eq!(toggled_label([Some(Red), Some(Red)], Red), None);
        assert_eq!(toggled_label([Some(Red), None], Red), Some(Red));
        assert_eq!(toggled_label([Some(Blue)], Red), Some(Red));
        assert_eq!(toggled_label([], Red), Some(Red));
    }

    #[test]
    fn the_label_histogram_sums_to_the_library() {
        // TRACES: FR-CAT-6
        // Images without a version row count as unlabelled rather than
        // vanishing, as the rating histogram's do.
        let cat = with_images(3);
        let first = ids(&cat)[0];
        set_label(cat.connection(), first, Some(ColourLabel::Purple)).unwrap();
        let h = label_histogram(cat.connection()).unwrap();
        assert_eq!(h, [2, 0, 0, 0, 0, 1]);
        assert_eq!(h.iter().sum::<usize>(), 3);
    }

    #[test]
    fn a_rating_round_trips() {
        let cat = with_images(1);
        let id = ids(&cat)[0];
        set_rating(cat.connection(), id, 4).unwrap();
        assert_eq!(judgement(cat.connection(), id).unwrap().rating, 4);
    }

    #[test]
    fn rating_an_image_with_no_version_creates_one() {
        // A library scanned by a build predating this module, or a scan that
        // died between the image insert and the version pass. The keystroke
        // must still land.
        let cat = with_images(1);
        let id = ids(&cat)[0];
        // Deliberately *not* calling ensure_default_versions first.
        set_rating(cat.connection(), id, 3).unwrap();
        assert_eq!(judgement(cat.connection(), id).unwrap().rating, 3);
    }

    #[test]
    fn an_out_of_range_rating_is_clamped_rather_than_stored() {
        // A stored 9 would sort above five stars forever and no filter would
        // reach it.
        let cat = with_images(1);
        let id = ids(&cat)[0];
        set_rating(cat.connection(), id, 99).unwrap();
        assert_eq!(judgement(cat.connection(), id).unwrap().rating, MAX_RATING);
    }

    #[test]
    fn rating_back_to_zero_returns_an_image_to_unrated() {
        // Pressing 0 is how a mistake is undone, so it has to be reachable —
        // not a floor at one star.
        let cat = with_images(1);
        let id = ids(&cat)[0];
        set_rating(cat.connection(), id, 5).unwrap();
        set_rating(cat.connection(), id, 0).unwrap();

        let j = judgement(cat.connection(), id).unwrap();
        assert_eq!(j.rating, 0);
        assert!(!j.is_judged(), "back to unjudged, so a cull re-presents it");
    }

    #[test]
    fn flags_and_stars_are_independent_axes() {
        // Rejecting a four-star frame is a normal thing to do while culling,
        // and one axis must not clear the other.
        let cat = with_images(1);
        let id = ids(&cat)[0];
        set_rating(cat.connection(), id, 4).unwrap();
        set_flag(cat.connection(), id, FlagState::Reject).unwrap();

        let j = judgement(cat.connection(), id).unwrap();
        assert_eq!(j.rating, 4);
        assert_eq!(j.flag, FlagState::Reject);
    }

    #[test]
    fn a_flag_alone_counts_as_judged() {
        // Filter-to-unjudged must not re-present a frame the user already
        // picked, merely because they did not also star it.
        let cat = with_images(1);
        let id = ids(&cat)[0];
        set_flag(cat.connection(), id, FlagState::Pick).unwrap();
        assert!(judgement(cat.connection(), id).unwrap().is_judged());
    }

    #[test]
    fn a_bulk_rating_applies_to_the_whole_selection() {
        // One gesture: select forty, press 3.
        let cat = with_images(10);
        let all = ids(&cat);
        let chosen = &all[2..7];

        assert_eq!(set_rating_many(cat.connection(), chosen, 3).unwrap(), 5);

        for id in chosen {
            assert_eq!(judgement(cat.connection(), *id).unwrap().rating, 3);
        }
        // And nothing outside the selection moved.
        assert_eq!(judgement(cat.connection(), all[0]).unwrap().rating, 0);
        assert_eq!(judgement(cat.connection(), all[9]).unwrap().rating, 0);
    }

    #[test]
    fn a_bulk_write_over_an_empty_selection_is_a_no_op() {
        let cat = with_images(3);
        assert_eq!(set_rating_many(cat.connection(), &[], 5).unwrap(), 0);
        assert_eq!(
            set_flag_many(cat.connection(), &[], FlagState::Pick).unwrap(),
            0
        );
    }

    #[test]
    fn judgements_reads_a_whole_window_in_one_query() {
        // The grid draws a star strip per cell; one query per cell would be
        // 120 round trips on every scroll.
        let cat = with_images(6);
        let all = ids(&cat);
        set_rating(cat.connection(), all[1], 2).unwrap();
        set_flag(cat.connection(), all[3], FlagState::Pick).unwrap();

        let map = judgements(cat.connection(), &all).unwrap();
        assert_eq!(map.get(&all[1]).unwrap().rating, 2);
        assert_eq!(map.get(&all[3]).unwrap().flag, FlagState::Pick);
        // Never rated, so it is either absent or explicitly unrated — both
        // mean the same thing to the caller.
        assert_eq!(
            map.get(&all[5]).copied().unwrap_or_default(),
            Judgement::default()
        );
    }

    #[test]
    fn the_histogram_sums_to_the_library_size() {
        // A histogram that disagrees with the image count is not believable,
        // and the unrated bucket is the one a fresh library lives in.
        let cat = with_images(8);
        ensure_default_versions(cat.connection()).unwrap();
        let all = ids(&cat);
        set_rating(cat.connection(), all[0], 5).unwrap();
        set_rating(cat.connection(), all[1], 5).unwrap();
        set_rating(cat.connection(), all[2], 3).unwrap();

        let h = rating_histogram(cat.connection()).unwrap();
        assert_eq!(h[5], 2);
        assert_eq!(h[3], 1);
        assert_eq!(h[0], 5, "the rest are still unrated");
        assert_eq!(h.iter().sum::<usize>(), 8);
    }

    #[test]
    fn the_histogram_counts_images_with_no_version_as_unrated() {
        // They are unrated. Dropping them would make the counts disagree with
        // the grid, which is the failure the LEFT JOIN exists to prevent.
        let cat = with_images(4);
        // No ensure_default_versions call at all.
        let h = rating_histogram(cat.connection()).unwrap();
        assert_eq!(h[0], 4);
        assert_eq!(h.iter().sum::<usize>(), 4);
    }

    /// The rows of the join the histograms used to be spelled as, grouped the
    /// way `rating_histogram` groups them. What the counts must still agree
    /// with, in the states nothing in the schema prevents.
    fn by_join(cat: &Catalog, column: &str) -> Vec<(i64, i64)> {
        cat.connection()
            .prepare(&format!(
                "SELECT coalesce(v.{column}, 0) AS c, count(*)
                   FROM images i
                   LEFT JOIN versions v ON v.image_id = i.id AND v.is_default = 1
                  GROUP BY c ORDER BY c"
            ))
            .unwrap()
            .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))
            .unwrap()
            .map(Result::unwrap)
            .collect()
    }

    /// A library in every awkward state at once: an image with no version,
    /// one with only a virtual copy, one with two default versions, and
    /// values out of range on both axes.
    fn awkward() -> Catalog {
        let cat = with_images(8);
        ensure_default_versions(cat.connection()).unwrap();
        let all = ids(&cat);
        let c = cat.connection();
        set_rating(c, all[0], 5).unwrap();
        set_rating(c, all[1], 2).unwrap();
        set_label(c, all[1], Some(ColourLabel::Blue)).unwrap();
        c.execute(
            "DELETE FROM versions WHERE image_id = ?1",
            [all[2].0 as i64],
        )
        .unwrap();
        c.execute(
            "UPDATE versions SET is_default = 0 WHERE image_id = ?1",
            [all[3].0 as i64],
        )
        .unwrap();
        c.execute(
            "INSERT INTO versions(image_id, uuid, name, is_default, rating, label)
             VALUES (?1, 'second-default', 'Copy', 1, 4, 3)",
            [all[4].0 as i64],
        )
        .unwrap();
        c.execute(
            "UPDATE versions SET rating = -1, label = 9 WHERE image_id = ?1",
            [all[5].0 as i64],
        )
        .unwrap();
        c.execute(
            "UPDATE versions SET rating = 7, label = 0 WHERE image_id = ?1",
            [all[6].0 as i64],
        )
        .unwrap();
        cat
    }

    /// Fold the join's rows into slots the way the old code did.
    fn folded(rows: &[(i64, i64)], slot: impl Fn(i64) -> usize) -> [usize; 6] {
        let mut out = [0usize; 6];
        for &(code, n) in rows {
            out[slot(code)] += n as usize;
        }
        out
    }

    #[test]
    fn the_rating_histogram_agrees_with_the_join_it_replaced() {
        let cat = awkward();
        let expected = folded(&by_join(&cat, "rating"), |r| {
            r.clamp(0, MAX_RATING as i64) as usize
        });
        assert_eq!(rating_histogram(cat.connection()).unwrap(), expected);
        // Nine rows for eight images: the doubled default counts twice, as
        // it always has.
        assert_eq!(expected.iter().sum::<usize>(), 9);
    }

    #[test]
    fn the_label_histogram_agrees_with_the_join_it_replaced() {
        let cat = awkward();
        let expected = folded(&by_join(&cat, "label"), |code| {
            if label_from_code(Some(code)).is_some() {
                code as usize
            } else {
                0
            }
        });
        assert_eq!(label_histogram(cat.connection()).unwrap(), expected);
        assert_eq!(expected[3], 1, "the second default's label is counted");
        assert_eq!(expected.iter().sum::<usize>(), 9);
    }

    #[test]
    fn flag_counts_separate_picks_from_rejects() {
        let cat = with_images(5);
        let all = ids(&cat);
        set_flag(cat.connection(), all[0], FlagState::Pick).unwrap();
        set_flag(cat.connection(), all[1], FlagState::Pick).unwrap();
        set_flag(cat.connection(), all[2], FlagState::Reject).unwrap();

        assert_eq!(flag_counts(cat.connection()).unwrap(), (2, 1));
    }

    #[test]
    fn unflagging_removes_an_image_from_both_counts() {
        let cat = with_images(2);
        let all = ids(&cat);
        set_flag(cat.connection(), all[0], FlagState::Reject).unwrap();
        set_flag(cat.connection(), all[0], FlagState::Unflagged).unwrap();
        assert_eq!(flag_counts(cat.connection()).unwrap(), (0, 0));
    }

    #[test]
    fn a_rating_survives_the_selector_that_queries_it() {
        // The end-to-end property: what this module writes is what
        // `dr_catalog::query` compiles `Selector::Rating` to find. These are
        // two independent pieces of SQL and they must agree on where a rating
        // lives, or rating an image would appear to do nothing.
        use crate::Query;
        use dr_types::Selector;

        let cat = with_images(6);
        let all = ids(&cat);
        ensure_default_versions(cat.connection()).unwrap();
        set_rating(cat.connection(), all[0], 5).unwrap();
        set_rating(cat.connection(), all[1], 4).unwrap();
        set_rating(cat.connection(), all[2], 1).unwrap();

        let q = Query {
            filter: Selector::Rating { min: 4 },
            ..Default::default()
        };
        assert_eq!(cat.count(&q, 0).unwrap(), 2);
    }

    #[test]
    fn a_flag_survives_the_selector_that_queries_it() {
        // Same contract for the other axis: `flag_code` here and in `query`
        // are separate mappings and must not drift.
        use crate::Query;
        use dr_types::Selector;

        let cat = with_images(4);
        let all = ids(&cat);
        ensure_default_versions(cat.connection()).unwrap();
        set_flag(cat.connection(), all[0], FlagState::Pick).unwrap();
        set_flag(cat.connection(), all[1], FlagState::Reject).unwrap();

        let picks = Query {
            filter: Selector::Flag(FlagState::Pick),
            ..Default::default()
        };
        assert_eq!(cat.count(&picks, 0).unwrap(), 1);

        let rejects = Query {
            filter: Selector::Flag(FlagState::Reject),
            ..Default::default()
        };
        assert_eq!(cat.count(&rejects, 0).unwrap(), 1);
    }

    #[test]
    fn unjudged_is_reachable_as_a_filter() {
        // FR-CULL-4's "filter to unjudged", which is what lets a session
        // resume where it stopped.
        use crate::Query;
        use dr_types::Selector;

        let cat = with_images(5);
        let all = ids(&cat);
        ensure_default_versions(cat.connection()).unwrap();
        set_rating(cat.connection(), all[0], 2).unwrap();

        let q = Query {
            filter: Selector::Rating { min: 0 },
            ..Default::default()
        };
        // `min: 0` matches everything, so unjudged needs the negation.
        assert_eq!(cat.count(&q, 0).unwrap(), 5);

        let unrated = Query {
            filter: Selector::Not(Box::new(Selector::Rating { min: 1 })),
            ..Default::default()
        };
        assert_eq!(cat.count(&unrated, 0).unwrap(), 4);
    }
}

/// TRACES: FR-NC-8 | FR-NC-9
/// The identity two devices have to agree on without talking to each other.
#[cfg(test)]
mod derived_identity {
    use super::*;
    use crate::Catalog;

    /// A catalog whose images the server has named, as a remote scan leaves it.
    fn with_remote_images(file_ids: &[i64]) -> Catalog {
        let cat = Catalog::in_memory().unwrap();
        let c = cat.connection();
        c.execute(
            "INSERT INTO roots(id, kind, label) VALUES (1, 'remote', 'lib')",
            [],
        )
        .unwrap();
        for (i, file_id) in file_ids.iter().enumerate() {
            c.execute(
                "INSERT INTO images(root_id, source_ref, added_at) VALUES (1, ?1, 0)",
                [format!("img{i:03}.CR3")],
            )
            .unwrap();
            let image = c.last_insert_rowid();
            c.execute(
                "INSERT INTO remote(image_id, file_id) VALUES (?1, ?2)",
                rusqlite::params![image, file_id],
            )
            .unwrap();
        }
        cat
    }

    fn default_uuids(cat: &Catalog) -> Vec<String> {
        let mut stmt = cat
            .connection()
            .prepare("SELECT uuid FROM versions WHERE is_default = 1 ORDER BY image_id")
            .unwrap();
        stmt.query_map([], |r| r.get(0))
            .unwrap()
            .map(Result::unwrap)
            .collect()
    }

    /// The whole point: the same photograph, indexed independently on two
    /// devices, gets one identity. This used to be two.
    #[test]
    fn two_devices_derive_the_same_uuid_for_one_photograph() {
        let laptop = with_remote_images(&[4_812]);
        let tablet = with_remote_images(&[4_812]);
        ensure_default_versions(laptop.connection()).unwrap();
        ensure_default_versions(tablet.connection()).unwrap();

        assert_eq!(default_uuids(&laptop), default_uuids(&tablet));
    }

    /// And different photographs must still be told apart — the property the
    /// random uuid did have, which this must not give up to gain agreement.
    #[test]
    fn different_photographs_keep_different_uuids() {
        let cat = with_remote_images(&[1, 2, 3, 0x7FFF_FFFF_FFFF_FFFF]);
        ensure_default_versions(cat.connection()).unwrap();

        let mut uuids = default_uuids(&cat);
        let before = uuids.len();
        uuids.sort();
        uuids.dedup();
        assert_eq!(uuids.len(), before, "two photographs share an identity");
    }

    /// The file id has to survive the layout intact, or two ids that differ
    /// only in the bits it drops would collide.
    #[test]
    fn the_whole_file_id_is_carried() {
        // A pair differing only in the low four bits, and a pair differing
        // only in the high thirty-two — the two places a sloppy layout loses
        // information.
        assert_ne!(derived_version_uuid(0x10), derived_version_uuid(0x1F));
        assert_ne!(
            derived_version_uuid(0x0000_0001_0000_0000),
            derived_version_uuid(0x0000_0002_0000_0000)
        );
        assert_ne!(derived_version_uuid(0), derived_version_uuid(-1));
    }

    /// Well-formed, and recognisably not a generated one.
    #[test]
    fn a_derived_uuid_is_a_well_formed_v8() {
        let uuid = derived_version_uuid(4_812);
        let fields: Vec<&str> = uuid.split('-').collect();
        assert_eq!(fields.len(), 5);
        assert_eq!(
            fields.iter().map(|f| f.len()).collect::<Vec<_>>(),
            vec![8, 4, 4, 4, 12]
        );
        assert!(fields[2].starts_with('8'), "version nibble: {uuid}");
        // The RFC's variant field is the two high bits of the fourth group,
        // and must read `0b10` — so the first hex digit is 8, 9, a or b.
        assert!(
            matches!(fields[3].as_bytes()[0], b'8' | b'9' | b'a' | b'b'),
            "variant: {uuid}"
        );
        assert!(uuid.ends_with("d0c5ec0de001"), "tag: {uuid}");
    }

    /// A library with no server behind it has no shared identity to derive,
    /// and must still get a version rather than failing.
    #[test]
    fn a_library_with_no_server_still_gets_its_versions() {
        let cat = Catalog::in_memory().unwrap();
        let c = cat.connection();
        c.execute(
            "INSERT INTO roots(id, kind, label) VALUES (1, 'local', 'lib')",
            [],
        )
        .unwrap();
        c.execute(
            "INSERT INTO images(root_id, source_ref, added_at) VALUES (1, 'a.CR3', 0)",
            [],
        )
        .unwrap();

        assert_eq!(ensure_default_versions(c).unwrap(), 1);
        assert_eq!(default_uuids(&cat).len(), 1);
    }

    /// The repair. A catalog written by an earlier build holds randomly minted
    /// uuids, and leaving them there would mean every image already indexed —
    /// which is all of them — kept writing to its own rival identity.
    #[test]
    fn a_catalog_from_an_earlier_build_is_realigned() {
        let cat = with_remote_images(&[4_812, 4_813]);
        let c = cat.connection();
        // As the old code left it.
        for (i, image) in [1i64, 2].iter().enumerate() {
            c.execute(
                "INSERT INTO versions(image_id, uuid, name, is_default, rating, flag)
                 VALUES (?1, ?2, 'Default', 1, ?3, 0)",
                rusqlite::params![image, format!("random-{i}"), (i + 1) as i64],
            )
            .unwrap();
        }

        assert_eq!(align_default_version_uuids(c).unwrap(), 2);
        assert_eq!(
            default_uuids(&cat),
            vec![derived_version_uuid(4_812), derived_version_uuid(4_813)]
        );

        // The judgement travels with the row — a realignment that dropped the
        // ratings would be a worse bug than the one it fixes.
        let ratings: Vec<i64> = {
            let mut stmt = c
                .prepare("SELECT rating FROM versions ORDER BY image_id")
                .unwrap();
            let v = stmt
                .query_map([], |r| r.get(0))
                .unwrap()
                .map(Result::unwrap)
                .collect();
            v
        };
        assert_eq!(ratings, vec![1, 2]);
    }

    /// Runs on every catalog open, so a second pass must select nothing and
    /// write nothing.
    #[test]
    fn realigning_twice_changes_nothing_the_second_time() {
        let cat = with_remote_images(&[4_812]);
        ensure_default_versions(cat.connection()).unwrap();

        assert_eq!(
            align_default_version_uuids(cat.connection()).unwrap(),
            0,
            "a freshly derived catalog must match nothing"
        );
        let before = default_uuids(&cat);
        align_default_version_uuids(cat.connection()).unwrap();
        assert_eq!(default_uuids(&cat), before);
    }

    /// A virtual copy (FR-CAT-12) that already holds the target uuid must not
    /// take the backfill — and with it the catalog open — down with it.
    #[test]
    fn a_taken_target_leaves_the_row_where_it_is() {
        let cat = with_remote_images(&[4_812]);
        let c = cat.connection();
        c.execute(
            "INSERT INTO versions(image_id, uuid, name, is_default, rating, flag)
             VALUES (1, 'random', 'Default', 1, 0, 0)",
            [],
        )
        .unwrap();
        c.execute(
            "INSERT INTO versions(image_id, uuid, name, is_default, rating, flag)
             VALUES (1, ?1, 'For print', 0, 0, 0)",
            [derived_version_uuid(4_812)],
        )
        .unwrap();

        assert_eq!(align_default_version_uuids(c).unwrap(), 0);
        assert_eq!(default_uuids(&cat), vec!["random".to_string()]);
    }

    /// The backfill is what actually runs this, so it has to be wired in.
    #[test]
    fn opening_a_catalog_realigns_it() {
        let cat = with_remote_images(&[4_812]);
        cat.connection()
            .execute(
                "INSERT INTO versions(image_id, uuid, name, is_default, rating, flag)
                 VALUES (1, 'random', 'Default', 1, 3, 0)",
                [],
            )
            .unwrap();

        crate::schema::backfill(cat.connection()).unwrap();
        assert_eq!(default_uuids(&cat), vec![derived_version_uuid(4_812)]);
    }
}