//! TRACES: FR-CAT-4 | FR-CAT-6
//! Compiling a [`Selector`] into indexed SQL, and windowing the result.
//!
//! The UI never assembles SQL — it hands over a [`Query`] and receives a
//! window. Two properties matter:
//!
//! 1. **Nothing user-supplied is interpolated into SQL text.** Every value
//!    binds as a parameter; `LIKE` patterns have their wildcards escaped.
//! 2. **Predicates hit indices.** Filtering 50k images must stay interactive
//!    (FR-CAT-6), which means no expression over a column that would defeat
//!    its index.

use dr_types::{Availability, ColourLabel, DateSelector, FlagState, Selector};
use rusqlite::types::Value;

/// What to show, and in what order.
#[derive(Debug, Clone)]
pub struct Query {
    pub filter: Selector,
    pub sort: Sort,
    pub descending: bool,
}

impl Default for Query {
    fn default() -> Self {
        Query {
            filter: Selector::All,
            sort: Sort::CapturedAt,
            descending: true,
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Sort {
    CapturedAt,
    Added,
    FileName,
    Rating,
    /// Manual order within a collection. Falls back to capture time where the
    /// query is not scoped to one collection, since position is meaningless
    /// outside it.
    CollectionPosition,
}

impl Sort {
    /// The ORDER BY fragment. Fixed strings — never user input.
    ///
    /// Capture time sorts NULLs last regardless of direction: an image whose
    /// EXIF has not been read yet (metadata_state 1) should not lead the grid
    /// simply because its timestamp is unknown.
    fn sql(self, descending: bool) -> &'static str {
        match (self, descending) {
            (Sort::CapturedAt, false) => {
                "ORDER BY images.captured_at IS NULL, images.captured_at ASC, images.id ASC"
            }
            (Sort::CapturedAt, true) => {
                "ORDER BY images.captured_at IS NULL, images.captured_at DESC, images.id DESC"
            }
            (Sort::Added, false) => "ORDER BY images.added_at ASC, images.id ASC",
            (Sort::Added, true) => "ORDER BY images.added_at DESC, images.id DESC",
            (Sort::FileName, false) => "ORDER BY images.source_ref ASC, images.id ASC",
            (Sort::FileName, true) => "ORDER BY images.source_ref DESC, images.id DESC",
            (Sort::Rating, false) => "ORDER BY v.rating ASC, images.id ASC",
            (Sort::Rating, true) => "ORDER BY v.rating DESC, images.id DESC",
            (Sort::CollectionPosition, false) => {
                "ORDER BY cm.position IS NULL, cm.position ASC, images.captured_at ASC"
            }
            (Sort::CollectionPosition, true) => {
                "ORDER BY cm.position IS NULL, cm.position DESC, images.captured_at DESC"
            }
        }
    }

    /// Whether this sort needs the default-version join.
    fn needs_version(self) -> bool {
        matches!(self, Sort::Rating)
    }

    /// Whether this sort needs a collection-membership join.
    fn needs_membership(self) -> bool {
        matches!(self, Sort::CollectionPosition)
    }
}

/// A compiled WHERE clause plus its bound parameters.
///
/// Kept separate from the statement so `count` and `window` can share one
/// compilation.
#[derive(Debug, Default)]
pub struct Compiled {
    pub where_sql: String,
    pub params: Vec<Value>,
    /// True if the filter depends on capture time, and therefore on EXIF that
    /// a freshly scanned library may not have read yet. The UI surfaces this
    /// rather than silently under-reporting.
    pub needs_capture_time: bool,
}

/// Compile a selector to SQL against the `images` table.
///
/// `now` is passed rather than read from the clock so a rolling window is
/// reproducible in tests and consistent across one query.
pub fn compile(filter: &Selector, now: i64) -> Compiled {
    let mut params = Vec::new();
    let sql = if filter.is_unfiltered() {
        "1".to_string()
    } else {
        emit(filter, now, &mut params)
    };
    Compiled {
        where_sql: sql,
        params,
        needs_capture_time: filter.needs_capture_time(),
    }
}

fn emit(s: &Selector, now: i64, p: &mut Vec<Value>) -> String {
    match s {
        Selector::All => "1".into(),

        Selector::Collection(id) => {
            p.push(Value::Integer(id.0 as i64));
            format!(
                "EXISTS (SELECT 1 FROM collection_members m
                          WHERE m.image_id = images.id AND m.collection_id = ?{})",
                p.len()
            )
        }

        Selector::Folder {
            root,
            path,
            recursive,
        } => {
            p.push(Value::Integer(root.0 as i64));
            let root_ix = p.len();
            if *recursive {
                // Prefix match on the folder path. `like_prefix` escapes the
                // pattern metacharacters, so a folder literally named "50%"
                // matches itself and not everything.
                p.push(Value::Text(like_prefix(path)));
                format!(
                    "images.folder_id IN (
                        SELECT id FROM folders
                         WHERE root_id = ?{root_ix}
                           AND (path = ?{p} OR path LIKE ?{p} || '/%' ESCAPE '\\'))",
                    p = p.len()
                )
            } else {
                p.push(Value::Text(path.clone()));
                format!(
                    "images.folder_id IN (
                        SELECT id FROM folders WHERE root_id = ?{root_ix} AND path = ?{})",
                    p.len()
                )
            }
        }

        Selector::DateRange(d) => emit_date(d, now, p),

        Selector::Rating { min } => {
            p.push(Value::Integer(*min as i64));
            format!("{} >= ?{}", default_version_scalar("rating"), p.len())
        }

        Selector::Label(l) => {
            p.push(Value::Integer(label_code(*l)));
            format!("{} = ?{}", default_version_scalar("label"), p.len())
        }

        Selector::Flag(f) => {
            p.push(Value::Integer(flag_code(*f)));
            format!("{} = ?{}", default_version_scalar("flag"), p.len())
        }

        Selector::Keyword(k) => {
            p.push(Value::Text(k.clone()));
            format!(
                "EXISTS (SELECT 1 FROM keywords kw
                          JOIN versions kv ON kv.id = kw.version_id
                         WHERE kv.image_id = images.id AND kw.keyword = ?{})",
                p.len()
            )
        }

        Selector::Camera(c) => {
            p.push(Value::Text(c.clone()));
            format!("images.camera = ?{}", p.len())
        }

        Selector::Lens(l) => {
            p.push(Value::Text(l.clone()));
            format!("images.lens = ?{}", p.len())
        }

        Selector::IsoRange { min, max } => {
            p.push(Value::Integer(*min as i64));
            let lo = p.len();
            p.push(Value::Integer(*max as i64));
            format!("images.iso BETWEEN ?{lo} AND ?{}", p.len())
        }

        Selector::Availability(a) => {
            p.push(Value::Integer(availability_code(*a)));
            format!("images.availability = ?{}", p.len())
        }

        Selector::Text(t) => {
            // Substring over filename and keywords. A LIKE scan is adequate at
            // 50k; if free text over title and description becomes a real
            // workflow, FTS5 is the answer and it is additive.
            p.push(Value::Text(format!("%{}%", escape_like(t))));
            let ix = p.len();
            format!(
                "(images.source_ref LIKE ?{ix} ESCAPE '\\'
                  OR EXISTS (SELECT 1 FROM keywords kw
                              JOIN versions kv ON kv.id = kw.version_id
                             WHERE kv.image_id = images.id
                               AND kw.keyword LIKE ?{ix} ESCAPE '\\'))"
            )
        }

        // An empty conjunction is vacuously true; an empty disjunction matches
        // nothing. Both arise from a UI that lets every term be cleared, and
        // conflating them would show the whole library when the user meant the
        // opposite.
        Selector::All_(v) if v.is_empty() => "1".into(),
        Selector::Any(v) if v.is_empty() => "0".into(),

        Selector::All_(v) => join(v, " AND ", now, p),
        Selector::Any(v) => join(v, " OR ", now, p),
        Selector::Not(inner) => format!("NOT ({})", emit(inner, now, p)),
    }
}

fn join(items: &[Selector], op: &str, now: i64, p: &mut Vec<Value>) -> String {
    let parts: Vec<String> = items.iter().map(|s| emit(s, now, p)).collect();
    format!("({})", parts.join(op))
}

fn emit_date(d: &DateSelector, now: i64, p: &mut Vec<Value>) -> String {
    match d {
        DateSelector::Between { from, to } => {
            p.push(Value::Integer(*from));
            let lo = p.len();
            p.push(Value::Integer(*to));
            // Half-open, so adjacent ranges neither overlap nor gap.
            format!(
                "(images.captured_at >= ?{lo} AND images.captured_at < ?{})",
                p.len()
            )
        }
        DateSelector::Rolling { days } => {
            let from = now - (*days as i64) * 86_400;
            p.push(Value::Integer(from));
            format!("images.captured_at >= ?{}", p.len())
        }
        DateSelector::CollectionSpan(id) => {
            p.push(Value::Integer(id.0 as i64));
            let ix = p.len();
            format!(
                "images.captured_at BETWEEN
                    (SELECT min(i2.captured_at) FROM images i2
                       JOIN collection_members m2 ON m2.image_id = i2.id
                      WHERE m2.collection_id = ?{ix})
                AND (SELECT max(i2.captured_at) FROM images i2
                       JOIN collection_members m2 ON m2.image_id = i2.id
                      WHERE m2.collection_id = ?{ix})"
            )
        }
    }
}

/// Rating, label, and flag live on the *default* version, not the image.
///
/// A correlated subquery rather than a join, so these compose inside `OR` and
/// `NOT` without the join multiplying rows.
fn default_version_scalar(col: &str) -> String {
    format!(
        "(SELECT dv.{col} FROM versions dv
           WHERE dv.image_id = images.id AND dv.is_default = 1 LIMIT 1)"
    )
}

/// Escape LIKE metacharacters so a literal `%` or `_` in user text matches
/// itself. Paired with `ESCAPE '\'` in every LIKE that uses it.
fn escape_like(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for c in s.chars() {
        if matches!(c, '%' | '_' | '\\') {
            out.push('\\');
        }
        out.push(c);
    }
    out
}

fn like_prefix(path: &str) -> String {
    escape_like(path.trim_end_matches('/'))
}

fn label_code(l: ColourLabel) -> i64 {
    crate::rating::label_code(l)
}

fn flag_code(f: FlagState) -> i64 {
    match f {
        FlagState::Unflagged => 0,
        FlagState::Pick => 1,
        FlagState::Reject => 2,
    }
}

/// The stored form of an availability. Shared with [`crate::walk`], which
/// writes the column this reads — two spellings of the same mapping would
/// filter for a state nothing ever writes.
pub(crate) fn availability_code(a: Availability) -> i64 {
    match a {
        Availability::MetadataOnly => 0,
        Availability::Preview => 1,
        Availability::Original => 2,
        Availability::Offline => 3,
    }
}

/// Build the full SELECT for a window of results.
///
/// Joins are added only where the sort needs them, so an unsorted-by-rating
/// grid query touches one table.
pub fn window_sql(q: &Query, compiled: &Compiled) -> String {
    let mut joins = String::new();
    if q.sort.needs_version() {
        joins.push_str(" LEFT JOIN versions v ON v.image_id = images.id AND v.is_default = 1");
    }
    if q.sort.needs_membership() {
        // Only meaningful when the filter scopes to one collection; elsewhere
        // position is NULL and the sort falls through to capture time.
        joins.push_str(" LEFT JOIN collection_members cm ON cm.image_id = images.id");
    }
    format!(
        "SELECT images.id, images.source_ref, images.availability, images.captured_at, \
                images.captured_offset, images.metadata_state \
         FROM images{joins} WHERE {} {} LIMIT ? OFFSET ?",
        compiled.where_sql,
        q.sort.sql(q.descending)
    )
}

/// Build the COUNT for the same filter.
pub fn count_sql(compiled: &Compiled) -> String {
    format!("SELECT count(*) FROM images WHERE {}", compiled.where_sql)
}

#[cfg(test)]
mod tests {
    use super::*;
    use dr_types::{CollectionId, RootId};

    #[test]
    fn unfiltered_compiles_to_a_constant() {
        let c = compile(&Selector::All, 0);
        assert_eq!(c.where_sql, "1");
        assert!(c.params.is_empty());
    }

    #[test]
    fn empty_conjunction_and_disjunction_differ() {
        // The distinction that decides whether clearing a filter shows
        // everything or nothing.
        assert_eq!(compile(&Selector::All_(vec![]), 0).where_sql, "1");
        assert_eq!(compile(&Selector::Any(vec![]), 0).where_sql, "0");
    }

    #[test]
    fn values_bind_rather_than_interpolate() {
        // The injection guard: a hostile keyword must appear in params, never
        // in SQL text.
        let evil = "'; DROP TABLE images; --";
        let c = compile(&Selector::Keyword(evil.into()), 0);
        assert!(!c.where_sql.contains("DROP"));
        assert_eq!(c.params, vec![Value::Text(evil.into())]);
    }

    #[test]
    fn like_metacharacters_are_escaped() {
        // A search for "50%" must not match everything containing "50".
        let c = compile(&Selector::Text("50%".into()), 0);
        assert_eq!(c.params, vec![Value::Text("%50\\%%".into())]);
        assert!(c.where_sql.contains("ESCAPE"));
    }

    #[test]
    fn a_backslash_in_search_text_is_itself_escaped() {
        let c = compile(&Selector::Text("a\\b".into()), 0);
        assert_eq!(c.params, vec![Value::Text("%a\\\\b%".into())]);
    }

    #[test]
    fn rolling_window_resolves_against_supplied_now() {
        // Passed in rather than read from the clock, so the window is stable
        // across one query and reproducible in a test.
        let now = 1_000_000i64;
        let c = compile(
            &Selector::DateRange(DateSelector::Rolling { days: 90 }),
            now,
        );
        assert_eq!(c.params, vec![Value::Integer(now - 90 * 86_400)]);
    }

    #[test]
    fn between_is_half_open() {
        let c = compile(
            &Selector::DateRange(DateSelector::Between { from: 10, to: 20 }),
            0,
        );
        // Half-open so adjacent day buckets neither overlap nor leave a gap.
        assert!(c.where_sql.contains(">= ?1"));
        assert!(c.where_sql.contains("< ?2"));
    }

    #[test]
    fn nested_composition_numbers_parameters_in_order() {
        let s = Selector::All_(vec![
            Selector::Rating { min: 4 },
            Selector::Any(vec![
                Selector::Camera("X-T5".into()),
                Selector::Not(Box::new(Selector::Lens("XF 35".into()))),
            ]),
        ]);
        let c = compile(&s, 0);
        assert_eq!(
            c.params,
            vec![
                Value::Integer(4),
                Value::Text("X-T5".into()),
                Value::Text("XF 35".into()),
            ]
        );
        assert!(c.where_sql.contains("?1"));
        assert!(c.where_sql.contains("?2"));
        assert!(c.where_sql.contains("?3"));
    }

    #[test]
    fn recursive_folder_matches_the_folder_itself_and_below() {
        let c = compile(
            &Selector::Folder {
                root: RootId(1),
                path: "2026/08".into(),
                recursive: true,
            },
            0,
        );
        // Both branches: the folder's own images and those in subfolders.
        assert!(c.where_sql.contains("path = ?2"));
        assert!(c.where_sql.contains("|| '/%'"));
    }

    #[test]
    fn collection_span_binds_its_id_once_and_reuses_it() {
        let c = compile(
            &Selector::DateRange(DateSelector::CollectionSpan(CollectionId(7))),
            0,
        );
        assert_eq!(c.params, vec![Value::Integer(7)]);
    }

    #[test]
    fn capture_time_dependency_is_reported() {
        let c = compile(&Selector::DateRange(DateSelector::Rolling { days: 7 }), 0);
        assert!(c.needs_capture_time);
        let c = compile(&Selector::Rating { min: 5 }, 0);
        assert!(!c.needs_capture_time);
    }

    #[test]
    fn capture_sort_puts_unknown_timestamps_last_in_both_directions() {
        // An image whose EXIF has not been read yet must not lead the grid
        // just because its timestamp is NULL.
        assert!(Sort::CapturedAt.sql(true).contains("IS NULL"));
        assert!(Sort::CapturedAt.sql(false).contains("IS NULL"));
    }

    #[test]
    fn window_sql_joins_only_when_the_sort_needs_it() {
        let c = compile(&Selector::All, 0);
        let plain = window_sql(
            &Query {
                filter: Selector::All,
                sort: Sort::CapturedAt,
                descending: true,
            },
            &c,
        );
        assert!(!plain.contains("JOIN"));

        let rated = window_sql(
            &Query {
                filter: Selector::All,
                sort: Sort::Rating,
                descending: true,
            },
            &c,
        );
        assert!(rated.contains("JOIN versions"));
    }
}