rustlavel-db 0.8.1

//! What differs between one SQL database and another.
//!
//! The query builder, the schema builder and the migrator are written once; a
//! [`Dialect`] supplies the handful of things the databases genuinely disagree
//! about — how an identifier is quoted, what a bound parameter looks like, what
//! a column type is called, and how a generated key is read back.
//!
//! Everything a dialect answers is pure string generation, so all three are
//! tested without a database anywhere near them.

use rustlavel_core::{Error, Result};

/// A column type as the schema builder thinks of it, before any database has
/// had an opinion.
///
/// Logical rather than literal: `Timestamp` is `timestamptz` on PostgreSQL,
/// `datetime(6)` on MySQL and `datetime2` on SQL Server, and a migration should
/// not have to know that.
#[derive(Debug, Clone, PartialEq)]
pub enum ColumnType {
    /// The conventional auto-incrementing primary key.
    Id,
    /// A UUID primary key, defaulted by the database.
    UuidId,
    SmallInteger,
    Integer,
    BigInteger,
    /// Approximate; for money use [`ColumnType::Decimal`].
    Float,
    Decimal { precision: u32, scale: u32 },
    Boolean,
    String { length: u32 },
    Text,
    Json,
    Uuid,
    Date,
    Time,
    Timestamp,
    Binary,
    /// An escape hatch for a type the framework does not model.
    Raw(String),
}

/// How a database hands back the key it generated for an inserted row.
#[derive(Debug, Clone, PartialEq)]
pub enum ReturningStyle {
    /// `insert into … values (…) returning "id"` — PostgreSQL.
    Suffix,
    /// `insert into … (…) output inserted.[id] values (…)` — SQL Server puts it
    /// between the column list and `values`, so it cannot be appended.
    OutputClause,
    /// Not supported: the key is read with a second statement. MySQL.
    SeparateQuery(&'static str),
}

/// The differences between one SQL database and another.
pub trait Dialect: Send + Sync + std::fmt::Debug + 'static {
    /// `postgres`, `mysql`, `sqlserver`.
    fn name(&self) -> &'static str;

    /// Wrap one identifier so a keyword or an unusual name is still valid.
    ///
    /// The identifier has already been validated; this only quotes it.
    fn quote(&self, identifier: &str) -> String;

    /// The placeholder for the `position`-th bound parameter, counting from 1.
    fn placeholder(&self, position: usize) -> String;

    /// The type name for a logical column type.
    fn column_type(&self, kind: &ColumnType) -> String;

    /// The expression for "now", used by `timestamps()`.
    fn now(&self) -> &'static str;

    /// The expression that generates a UUID, when the database has one.
    fn uuid_default(&self) -> Option<&'static str>;

    /// How a generated key comes back from an insert.
    fn returning(&self) -> ReturningStyle;

    /// `limit … offset …`, in whatever form this database accepts.
    ///
    /// `ordered` says whether the query already has an `order by`, because SQL
    /// Server's paging syntax requires one and will not accept paging without.
    fn limit_offset(&self, limit: Option<i64>, offset: Option<i64>, ordered: bool) -> String;

    /// Whether `create table if not exists` is understood.
    fn supports_if_not_exists_table(&self) -> bool {
        true
    }

    /// Whether `create index if not exists` is understood.
    ///
    /// PostgreSQL has it; MySQL and SQL Server do not, so the schema builder
    /// emits a plain `create index` and a repeated migration would fail — which
    /// is correct, since migrations run once.
    fn supports_if_not_exists_index(&self) -> bool {
        false
    }

    /// Whether a `boolean` column really is one.
    ///
    /// MySQL stores it as `tinyint(1)` and SQL Server as `bit`, so both hand
    /// back a number where PostgreSQL hands back a boolean. The row decoder
    /// uses this to convert.
    fn booleans_are_integers(&self) -> bool {
        false
    }

    /// The longest identifier this database accepts.
    fn max_identifier_length(&self) -> usize {
        63
    }

    /// The DDL for the migration tracking table.
    ///
    /// The key column has to say `primary key`: MySQL refuses an
    /// `auto_increment` column that is not one, and a real server is the only
    /// thing that will tell you so.
    fn migrations_table_sql(&self, table: &str) -> String {
        format!(
            "create table if not exists {} (\n  \
             id {} primary key,\n  \
             name {} not null unique,\n  \
             batch {} not null,\n  \
             ran_at {} not null default {}\n)",
            self.quote(table),
            self.column_type(&ColumnType::Id),
            self.column_type(&ColumnType::String { length: 255 }),
            self.column_type(&ColumnType::Integer),
            self.column_type(&ColumnType::Timestamp),
            self.now()
        )
    }

    /// How a column is added in an `alter table`.
    ///
    /// PostgreSQL and MySQL say `add column`; SQL Server rejects the keyword on
    /// `add` while requiring it on `drop`, which is asymmetric enough that
    /// nobody guesses it right.
    fn add_column_clause(&self) -> &'static str {
        "add column"
    }

    /// Start a transaction.
    ///
    /// T-SQL wants the word `transaction`; the others are happy with `begin`
    /// alone and reject `begin transaction` is fine there too, but the bare
    /// form is what their documentation uses.
    fn begin_sql(&self) -> &'static str {
        "begin"
    }

    fn commit_sql(&self) -> &'static str {
        "commit"
    }

    fn rollback_sql(&self) -> &'static str {
        "rollback"
    }

    fn savepoint_sql(&self, name: &str) -> String {
        format!("savepoint {name}")
    }

    fn rollback_to_savepoint_sql(&self, name: &str) -> String {
        format!("rollback to savepoint {name}")
    }

    /// How a `select` claims the row it picks so no other worker takes it,
    /// as `(hint after the table, clause before the limit)`.
    ///
    /// A queue needs exactly this primitive and nothing weaker: a plain
    /// `select` then `update` is a read-then-write race that runs a job twice,
    /// and a lock without *skip* serialises every worker behind one row
    /// instead of letting them take the rows after it.
    ///
    /// **Two positions, because SQL Server's form is not a suffix.**
    /// PostgreSQL and MySQL append `for update skip locked` at the end;
    /// SQL Server says `with (updlock, readpast, rowlock)` immediately after
    /// the table name, and putting it at the end is a syntax error. Returning
    /// one string would have made the caller guess.
    ///
    /// The default is no locking at all, which is right only for a database
    /// that serialises writers by itself — see SQLite below.
    fn skip_locked(&self) -> (&'static str, &'static str) {
        ("", "")
    }

    /// The expression naming the schema this connection is working in.
    ///
    /// `information_schema` is standard; the way you ask "which schema am I in"
    /// is not.
    fn current_schema_expression(&self) -> &'static str;

    /// A query returning one row per table in the current schema, with the name
    /// in the first column.
    ///
    /// `migrate:fresh` enumerates and drops rather than running one clever
    /// statement, because only PostgreSQL has an anonymous block to put a loop
    /// in — and the enumerate-then-drop shape works identically everywhere.
    fn list_tables_sql(&self) -> &'static str;

    /// Turn off foreign key enforcement while tables are being dropped, so the
    /// order they come back in does not matter.
    fn disable_foreign_keys_sql(&self) -> Option<&'static str> {
        None
    }

    fn enable_foreign_keys_sql(&self) -> Option<&'static str> {
        None
    }

    /// Drop one table, including anything depending on it.
    fn drop_table_sql(&self, table: &str) -> String {
        format!("drop table if exists {}", self.quote(table))
    }
}

/// Quote a possibly-qualified name one part at a time.
pub fn quote_qualified(dialect: &dyn Dialect, name: &str) -> Result<String> {
    let parts: Result<Vec<String>> = name
        .split('.')
        .map(|part| {
            validate_identifier(part, dialect.max_identifier_length())
                .map(|_| dialect.quote(part))
        })
        .collect();
    Ok(parts?.join("."))
}

/// Reject anything that is not a plain identifier.
///
/// Identifiers cannot be sent as bound parameters, so every place the framework
/// interpolates one into SQL passes through here first.
pub fn validate_identifier(name: &str, max_length: usize) -> Result<()> {
    let valid = !name.is_empty()
        && name.len() <= max_length
        && name.chars().next().is_some_and(|c| c.is_ascii_alphabetic() || c == '_')
        && name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_');

    if valid {
        Ok(())
    } else {
        Err(Error::msg(format!(
            "`{name}` is not a valid SQL identifier. Identifiers may contain letters, digits and \
             underscores, must not start with a digit, and must be at most {max_length} characters."
        )))
    }
}

// --- PostgreSQL ---

#[derive(Debug, Default, Clone, Copy)]
pub struct Postgres;

impl Dialect for Postgres {
    fn name(&self) -> &'static str {
        "postgres"
    }

    fn quote(&self, identifier: &str) -> String {
        format!("\"{identifier}\"")
    }

    fn placeholder(&self, position: usize) -> String {
        format!("${position}")
    }

    fn column_type(&self, kind: &ColumnType) -> String {
        match kind {
            ColumnType::Id => "bigserial".into(),
            ColumnType::UuidId | ColumnType::Uuid => "uuid".into(),
            ColumnType::SmallInteger => "smallint".into(),
            ColumnType::Integer => "integer".into(),
            ColumnType::BigInteger => "bigint".into(),
            ColumnType::Float => "double precision".into(),
            ColumnType::Decimal { precision, scale } => format!("numeric({precision}, {scale})"),
            ColumnType::Boolean => "boolean".into(),
            ColumnType::String { length } => format!("varchar({length})"),
            ColumnType::Text => "text".into(),
            ColumnType::Json => "jsonb".into(),
            ColumnType::Date => "date".into(),
            ColumnType::Time => "time".into(),
            ColumnType::Timestamp => "timestamptz".into(),
            ColumnType::Binary => "bytea".into(),
            ColumnType::Raw(sql) => sql.clone(),
        }
    }

    fn now(&self) -> &'static str {
        "now()"
    }

    fn uuid_default(&self) -> Option<&'static str> {
        Some("gen_random_uuid()")
    }

    fn returning(&self) -> ReturningStyle {
        ReturningStyle::Suffix
    }

    fn limit_offset(&self, limit: Option<i64>, offset: Option<i64>, _ordered: bool) -> String {
        let mut out = String::new();
        if let Some(limit) = limit {
            out.push_str(&format!(" limit {}", limit.max(0)));
        }
        if let Some(offset) = offset {
            out.push_str(&format!(" offset {}", offset.max(0)));
        }
        out
    }

    fn supports_if_not_exists_index(&self) -> bool {
        true
    }

    fn skip_locked(&self) -> (&'static str, &'static str) {
        ("", " for update skip locked")
    }

    fn current_schema_expression(&self) -> &'static str {
        "current_schema()"
    }

    fn list_tables_sql(&self) -> &'static str {
        "select tablename from pg_tables where schemaname = current_schema()"
    }

    fn drop_table_sql(&self, table: &str) -> String {
        // `cascade` also removes the foreign keys pointing at it, which is why
        // PostgreSQL needs no enforcement switch.
        format!("drop table if exists {} cascade", self.quote(table))
    }
}

// --- SQLite ---

/// SQLite's SQL.
///
/// **This dialect compiles unconditionally; the driver beside it does not.**
/// A dialect is string formatting with no dependency at all, so the generated
/// SQL can be asserted in a build that cannot open a SQLite file — which is
/// how the other three are tested too.
///
/// SQLite is not a server and its SQL reflects that. Three differences are
/// load-bearing:
///
/// * **Types are advisory.** A column has a *type affinity*, not a type, and
///   a `varchar(255)` accepts a megabyte. The affinities below are the ones
///   the framework's [`ColumnType`]s map onto; the lengths are kept in the SQL
///   because they document intent and cost nothing.
/// * **`integer primary key` is the row id itself**, and auto-increments with
///   no keyword. `autoincrement` exists but only stops row ids being reused
///   after a delete, at the cost of a second table — which is not what the
///   other three do, so it is not asked for here.
/// * **Foreign keys are off unless switched on per connection.** That is a
///   SQLite default, not a choice this dialect can make; the driver turns them
///   on for every connection it opens.
#[derive(Debug, Default, Clone, Copy)]
pub struct Sqlite;

impl Dialect for Sqlite {
    fn name(&self) -> &'static str {
        "sqlite"
    }

    fn quote(&self, identifier: &str) -> String {
        format!("\"{identifier}\"")
    }

    fn placeholder(&self, _position: usize) -> String {
        "?".into()
    }

    fn column_type(&self, kind: &ColumnType) -> String {
        match kind {
            // No `autoincrement`: `integer primary key` *is* the row id, and
            // it increments on its own. The schema builder appends the
            // `primary key`, which is what makes this the row id rather than
            // an ordinary integer column.
            ColumnType::Id => "integer".into(),
            ColumnType::SmallInteger | ColumnType::Integer | ColumnType::BigInteger => {
                "integer".into()
            }
            ColumnType::Float => "real".into(),
            // Affinity `numeric`, which keeps the value exact when it is
            // written as a string. SQLite has no true decimal; money in a
            // SQLite database belongs in an integer of minor units, the way
            // `rustlavel-ledger` stores it.
            ColumnType::Decimal { precision, scale } => format!("numeric({precision}, {scale})"),
            // SQLite has no boolean. 0 and 1, which is why
            // `booleans_are_integers` is true below.
            ColumnType::Boolean => "integer".into(),
            ColumnType::String { length } => format!("varchar({length})"),
            ColumnType::Text | ColumnType::Json => "text".into(),
            ColumnType::UuidId | ColumnType::Uuid => "text".into(),
            ColumnType::Date | ColumnType::Time | ColumnType::Timestamp => "text".into(),
            ColumnType::Binary => "blob".into(),
            ColumnType::Raw(sql) => sql.clone(),
        }
    }

    fn now(&self) -> &'static str {
        "current_timestamp"
    }

    fn uuid_default(&self) -> Option<&'static str> {
        // SQLite has no uuid function, and inventing one out of `randomblob`
        // would produce something that looks like a UUID without being one.
        // The application generates the id.
        None
    }

    fn returning(&self) -> ReturningStyle {
        // SQLite has had `returning` since 3.35, but the row id is what the
        // framework wants and `last_insert_rowid()` has always been there —
        // so this works against an old system SQLite as well as a new one.
        ReturningStyle::SeparateQuery("select last_insert_rowid()")
    }

    fn limit_offset(&self, limit: Option<i64>, offset: Option<i64>, _ordered: bool) -> String {
        let mut out = String::new();
        if let Some(limit) = limit {
            out.push_str(&format!(" limit {}", limit.max(0)));
        }
        if let Some(offset) = offset {
            // SQLite rejects `offset` without `limit`, which the other three
            // accept. -1 is its documented "no limit".
            if limit.is_none() {
                out.push_str(" limit -1");
            }
            out.push_str(&format!(" offset {}", offset.max(0)));
        }
        out
    }

    fn supports_if_not_exists_index(&self) -> bool {
        true
    }

    fn booleans_are_integers(&self) -> bool {
        true
    }

    fn max_identifier_length(&self) -> usize {
        // SQLite imposes no limit worth the name. Kept at PostgreSQL's 63 so a
        // schema that works here works there — the point of developing against
        // SQLite is that the result runs on a server afterwards.
        63
    }

    fn current_schema_expression(&self) -> &'static str {
        "'main'"
    }

    /// **Nothing, and here that is correct rather than missing.** SQLite has
    /// no row locks — a write transaction locks the whole database — so two
    /// workers cannot both claim one job however the `select` is written. What
    /// makes it safe is `begin immediate` below, not a clause here.
    fn skip_locked(&self) -> (&'static str, &'static str) {
        ("", "")
    }

    /// `begin immediate`, not the bare `begin` the other three use.
    ///
    /// SQLite's default transaction is *deferred*: it takes a read lock on the
    /// first `select` and only tries to upgrade at the first write. Two
    /// transactions that both read and then both try to upgrade deadlock, and
    /// SQLite breaks the tie by failing one with `SQLITE_BUSY` immediately —
    /// without waiting for `busy_timeout`, because waiting cannot help. Two
    /// queue workers are exactly that shape.
    ///
    /// `begin immediate` takes the write lock up front, so the second worker
    /// waits its turn instead of failing.
    fn begin_sql(&self) -> &'static str {
        "begin immediate"
    }

    fn list_tables_sql(&self) -> &'static str {
        // `sqlite_%` is reserved for SQLite's own bookkeeping, and dropping one
        // is an error rather than a no-op.
        "select name from sqlite_master where type = 'table' and name not like 'sqlite_%'"
    }

    fn disable_foreign_keys_sql(&self) -> Option<&'static str> {
        // Not a statement SQLite will honour inside a transaction — it is
        // silently ignored there. `migrate:fresh` drops outside one, which is
        // where it works.
        Some("pragma foreign_keys = off")
    }

    fn enable_foreign_keys_sql(&self) -> Option<&'static str> {
        Some("pragma foreign_keys = on")
    }
}

// --- MySQL ---

#[derive(Debug, Default, Clone, Copy)]
pub struct MySql;

impl Dialect for MySql {
    fn name(&self) -> &'static str {
        "mysql"
    }

    fn quote(&self, identifier: &str) -> String {
        format!("`{identifier}`")
    }

    fn placeholder(&self, _position: usize) -> String {
        // MySQL binds by position in order, not by number.
        "?".into()
    }

    fn column_type(&self, kind: &ColumnType) -> String {
        match kind {
            // Signed, matching PostgreSQL's bigserial and SQL Server's bigint
            // identity. MySQL's convention is unsigned, but then a `bigint`
            // foreign key cannot reference it — MySQL requires the types to
            // match exactly, signedness included.
            ColumnType::Id => "bigint not null auto_increment".into(),
            // MySQL has no uuid type; 36 characters holds the canonical form.
            ColumnType::UuidId | ColumnType::Uuid => "char(36)".into(),
            ColumnType::SmallInteger => "smallint".into(),
            ColumnType::Integer => "int".into(),
            ColumnType::BigInteger => "bigint".into(),
            ColumnType::Float => "double".into(),
            ColumnType::Decimal { precision, scale } => format!("decimal({precision}, {scale})"),
            // `boolean` is an alias for tinyint(1); spelled out so the schema
            // says what the database actually stores.
            ColumnType::Boolean => "tinyint(1)".into(),
            ColumnType::String { length } => format!("varchar({length})"),
            ColumnType::Text => "text".into(),
            ColumnType::Json => "json".into(),
            ColumnType::Date => "date".into(),
            ColumnType::Time => "time".into(),
            // Fractional seconds are not the default and cannot be added later
            // without rewriting the table.
            ColumnType::Timestamp => "datetime(6)".into(),
            ColumnType::Binary => "longblob".into(),
            ColumnType::Raw(sql) => sql.clone(),
        }
    }

    fn now(&self) -> &'static str {
        "current_timestamp(6)"
    }

    fn uuid_default(&self) -> Option<&'static str> {
        // Only from MySQL 8.0.13, and only in an expression default; left off
        // so the schema builder does not emit something a 5.7 server rejects.
        None
    }

    fn returning(&self) -> ReturningStyle {
        ReturningStyle::SeparateQuery("select last_insert_id()")
    }

    fn limit_offset(&self, limit: Option<i64>, offset: Option<i64>, _ordered: bool) -> String {
        let mut out = String::new();
        match (limit, offset) {
            // MySQL cannot offset without a limit, so an offset alone gets the
            // largest limit the syntax allows.
            (None, Some(offset)) => {
                out.push_str(&format!(" limit 18446744073709551615 offset {}", offset.max(0)));
            }
            (Some(limit), offset) => {
                out.push_str(&format!(" limit {}", limit.max(0)));
                if let Some(offset) = offset {
                    out.push_str(&format!(" offset {}", offset.max(0)));
                }
            }
            (None, None) => {}
        }
        out
    }

    fn booleans_are_integers(&self) -> bool {
        true
    }

    fn max_identifier_length(&self) -> usize {
        64
    }

    fn current_schema_expression(&self) -> &'static str {
        // MySQL has no schemas separate from databases; the current database is
        // the schema.
        "database()"
    }

    /// The same words PostgreSQL uses, and **MySQL 8.0 or newer**.
    /// `skip locked` arrived in 8.0; on 5.7 this is a syntax error rather than
    /// a silent fallback, which is the better failure — a queue that quietly
    /// dropped the `skip` would run jobs twice.
    fn skip_locked(&self) -> (&'static str, &'static str) {
        ("", " for update skip locked")
    }

    fn list_tables_sql(&self) -> &'static str {
        "select table_name from information_schema.tables \
         where table_schema = database() and table_type = 'BASE TABLE'"
    }

    fn disable_foreign_keys_sql(&self) -> Option<&'static str> {
        Some("set foreign_key_checks = 0")
    }

    fn enable_foreign_keys_sql(&self) -> Option<&'static str> {
        Some("set foreign_key_checks = 1")
    }
}

// --- SQL Server ---

#[derive(Debug, Default, Clone, Copy)]
pub struct SqlServer;

impl Dialect for SqlServer {
    fn name(&self) -> &'static str {
        "sqlserver"
    }

    fn quote(&self, identifier: &str) -> String {
        format!("[{identifier}]")
    }

    fn placeholder(&self, position: usize) -> String {
        format!("@P{position}")
    }

    fn column_type(&self, kind: &ColumnType) -> String {
        match kind {
            ColumnType::Id => "bigint identity(1,1)".into(),
            ColumnType::UuidId | ColumnType::Uuid => "uniqueidentifier".into(),
            ColumnType::SmallInteger => "smallint".into(),
            ColumnType::Integer => "int".into(),
            ColumnType::BigInteger => "bigint".into(),
            ColumnType::Float => "float".into(),
            ColumnType::Decimal { precision, scale } => format!("decimal({precision}, {scale})"),
            ColumnType::Boolean => "bit".into(),
            // `n` prefixed: the framework speaks UTF-8, and nvarchar is the type
            // that stores it without a collation surprise.
            ColumnType::String { length } => format!("nvarchar({length})"),
            ColumnType::Text | ColumnType::Json => "nvarchar(max)".into(),
            ColumnType::Date => "date".into(),
            ColumnType::Time => "time".into(),
            ColumnType::Timestamp => "datetime2".into(),
            ColumnType::Binary => "varbinary(max)".into(),
            ColumnType::Raw(sql) => sql.clone(),
        }
    }

    fn now(&self) -> &'static str {
        "sysutcdatetime()"
    }

    fn uuid_default(&self) -> Option<&'static str> {
        Some("newid()")
    }

    fn returning(&self) -> ReturningStyle {
        ReturningStyle::OutputClause
    }

    fn limit_offset(&self, limit: Option<i64>, offset: Option<i64>, ordered: bool) -> String {
        if limit.is_none() && offset.is_none() {
            return String::new();
        }

        // `offset … fetch next …` is only legal after an `order by`, so an
        // unordered paged query gets a placeholder ordering rather than a
        // syntax error the caller cannot explain.
        let mut out = String::new();
        if !ordered {
            out.push_str(" order by (select null)");
        }
        out.push_str(&format!(" offset {} rows", offset.unwrap_or(0).max(0)));
        if let Some(limit) = limit {
            out.push_str(&format!(" fetch next {} rows only", limit.max(0)));
        }
        out
    }

    fn supports_if_not_exists_table(&self) -> bool {
        false
    }

    fn booleans_are_integers(&self) -> bool {
        true
    }

    fn max_identifier_length(&self) -> usize {
        128
    }

    fn migrations_table_sql(&self, table: &str) -> String {
        // No `if not exists`; the catalogue is checked instead.
        format!(
            "if object_id('{table}', 'U') is null create table {} (\n  \
             [id] bigint identity(1,1) primary key,\n  \
             [name] nvarchar(255) not null unique,\n  \
             [batch] int not null,\n  \
             [ran_at] datetime2 not null default sysutcdatetime()\n)",
            self.quote(table)
        )
    }

    fn add_column_clause(&self) -> &'static str {
        "add"
    }

    fn begin_sql(&self) -> &'static str {
        "begin transaction"
    }

    fn commit_sql(&self) -> &'static str {
        "commit transaction"
    }

    fn rollback_sql(&self) -> &'static str {
        "rollback transaction"
    }

    fn savepoint_sql(&self, name: &str) -> String {
        // T-SQL has no `savepoint` keyword; a named save point is made and
        // returned to with `transaction`.
        format!("save transaction {name}")
    }

    fn rollback_to_savepoint_sql(&self, name: &str) -> String {
        format!("rollback transaction {name}")
    }

    fn current_schema_expression(&self) -> &'static str {
        "schema_name()"
    }

    /// A table hint, which is why this returns two pieces rather than one.
    /// `updlock` takes the update lock the row is about to need, `readpast` is
    /// T-SQL's `skip locked`, and `rowlock` stops the engine escalating to a
    /// page or table lock and serialising every worker.
    fn skip_locked(&self) -> (&'static str, &'static str) {
        (" with (updlock, readpast, rowlock)", "")
    }

    fn list_tables_sql(&self) -> &'static str {
        // `is_ms_shipped = 0` excludes the system tables SQL Server keeps in
        // some databases; without it, `migrate:fresh` pointed at `master` would
        // try to drop Microsoft's own.
        "select t.name from sys.tables t \
         where t.is_ms_shipped = 0 and schema_name(t.schema_id) = schema_name()"
    }

    fn disable_foreign_keys_sql(&self) -> Option<&'static str> {
        // Undocumented but long-standing: applies to every table at once.
        Some("exec sp_MSforeachtable 'alter table ? nocheck constraint all'")
    }

    fn enable_foreign_keys_sql(&self) -> Option<&'static str> {
        Some("exec sp_MSforeachtable 'alter table ? with check check constraint all'")
    }
}

/// Build a dialect from its name.
pub fn by_name(name: &str) -> Result<Box<dyn Dialect>> {
    match name.to_ascii_lowercase().as_str() {
        "postgres" | "postgresql" | "pgsql" => Ok(Box::new(Postgres)),
        "mysql" | "mariadb" => Ok(Box::new(MySql)),
        "sqlserver" | "mssql" => Ok(Box::new(SqlServer)),
        other => Err(Error::msg(format!(
            "`{other}` is not a database this framework speaks. Available: postgres, mysql, sqlserver."
        ))),
    }
}

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

    fn all() -> Vec<Box<dyn Dialect>> {
        vec![Box::new(Postgres), Box::new(MySql), Box::new(SqlServer)]
    }

    #[test]
    fn each_dialect_quotes_the_way_its_database_expects() {
        assert_eq!(Postgres.quote("users"), "\"users\"");
        assert_eq!(MySql.quote("users"), "`users`");
        assert_eq!(SqlServer.quote("users"), "[users]");
    }

    #[test]
    fn placeholders_differ_in_kind_not_just_spelling() {
        assert_eq!(Postgres.placeholder(1), "$1");
        assert_eq!(Postgres.placeholder(3), "$3");

        // MySQL binds positionally, so every placeholder is the same token.
        assert_eq!(MySql.placeholder(1), "?");
        assert_eq!(MySql.placeholder(3), "?");

        assert_eq!(SqlServer.placeholder(3), "@P3");
    }

    #[test]
    fn a_qualified_name_is_quoted_one_part_at_a_time() {
        assert_eq!(
            quote_qualified(&Postgres, "public.users").unwrap(),
            "\"public\".\"users\""
        );
        assert_eq!(quote_qualified(&MySql, "shop.orders").unwrap(), "`shop`.`orders`");
        assert_eq!(quote_qualified(&SqlServer, "dbo.users").unwrap(), "[dbo].[users]");
    }

    #[test]
    fn an_injected_identifier_is_rejected_by_every_dialect() {
        for dialect in all() {
            for hostile in ["users; drop table users", "a b", "1abc", "", "us\"er"] {
                assert!(
                    quote_qualified(dialect.as_ref(), hostile).is_err(),
                    "{} accepted {hostile:?}",
                    dialect.name()
                );
            }
        }
    }

    #[test]
    fn identifier_length_limits_follow_the_database() {
        let long = "a".repeat(100);

        assert!(validate_identifier(&long, Postgres.max_identifier_length()).is_err());
        assert!(validate_identifier(&long, MySql.max_identifier_length()).is_err());
        assert!(validate_identifier(&long, SqlServer.max_identifier_length()).is_ok());
    }

    #[test]
    fn the_key_column_is_auto_incrementing_everywhere() {
        assert_eq!(Postgres.column_type(&ColumnType::Id), "bigserial");
        assert_eq!(MySql.column_type(&ColumnType::Id), "bigint not null auto_increment");
        assert_eq!(SqlServer.column_type(&ColumnType::Id), "bigint identity(1,1)");
    }

    #[test]
    fn text_and_json_map_to_what_each_database_actually_has() {
        assert_eq!(Postgres.column_type(&ColumnType::Json), "jsonb");
        assert_eq!(MySql.column_type(&ColumnType::Json), "json");
        // SQL Server has no JSON type; it stores the document as text.
        assert_eq!(SqlServer.column_type(&ColumnType::Json), "nvarchar(max)");
    }

    #[test]
    fn a_string_column_carries_its_length_everywhere() {
        let kind = ColumnType::String { length: 120 };

        assert_eq!(Postgres.column_type(&kind), "varchar(120)");
        assert_eq!(MySql.column_type(&kind), "varchar(120)");
        assert_eq!(SqlServer.column_type(&kind), "nvarchar(120)");
    }

    #[test]
    fn paging_uses_each_databases_own_syntax() {
        assert_eq!(Postgres.limit_offset(Some(10), Some(20), true), " limit 10 offset 20");
        assert_eq!(MySql.limit_offset(Some(10), Some(20), true), " limit 10 offset 20");
        assert_eq!(
            SqlServer.limit_offset(Some(10), Some(20), true),
            " offset 20 rows fetch next 10 rows only"
        );
    }

    #[test]
    fn sql_server_supplies_an_ordering_when_paging_has_none() {
        // `offset` is a syntax error without `order by`, and a caller cannot
        // debug an error the builder could have avoided.
        let paged = SqlServer.limit_offset(Some(10), None, false);
        assert!(paged.starts_with(" order by (select null)"), "{paged}");

        // With an ordering already present, none is added.
        assert!(!SqlServer.limit_offset(Some(10), None, true).contains("order by"));
    }

    #[test]
    fn mysql_cannot_offset_without_a_limit() {
        let offset_only = MySql.limit_offset(None, Some(20), true);

        assert!(offset_only.contains("limit 18446744073709551615"), "{offset_only}");
        assert!(offset_only.ends_with("offset 20"));
    }

    #[test]
    fn no_paging_produces_no_clause() {
        for dialect in all() {
            assert_eq!(dialect.limit_offset(None, None, true), "", "{}", dialect.name());
        }
    }

    #[test]
    fn generated_keys_come_back_differently() {
        assert_eq!(Postgres.returning(), ReturningStyle::Suffix);
        assert_eq!(SqlServer.returning(), ReturningStyle::OutputClause);
        assert_eq!(
            MySql.returning(),
            ReturningStyle::SeparateQuery("select last_insert_id()")
        );
    }

    #[test]
    fn the_migration_table_is_valid_for_each_database() {
        let postgres = Postgres.migrations_table_sql("rustlavel_migrations");
        assert!(postgres.contains("create table if not exists \"rustlavel_migrations\""));
        assert!(postgres.contains("bigserial primary key"));

        let mysql = MySql.migrations_table_sql("rustlavel_migrations");
        assert!(mysql.contains("`rustlavel_migrations`"));
        // MySQL rejects an auto_increment column that is not a key.
        assert!(mysql.contains("auto_increment primary key"), "{mysql}");

        // SQL Server has no `if not exists`, so it checks the catalogue.
        let sqlserver = SqlServer.migrations_table_sql("rustlavel_migrations");
        assert!(sqlserver.starts_with("if object_id("));
        assert!(sqlserver.contains("identity(1,1)"));
    }

    #[test]
    fn transaction_control_uses_each_databases_own_words() {
        // `begin` alone is a syntax error in T-SQL, which would have broken
        // every transaction on SQL Server.
        assert_eq!(Postgres.begin_sql(), "begin");
        assert_eq!(MySql.begin_sql(), "begin");
        assert_eq!(SqlServer.begin_sql(), "begin transaction");

        assert_eq!(SqlServer.commit_sql(), "commit transaction");
        assert_eq!(SqlServer.rollback_sql(), "rollback transaction");
        assert_eq!(SqlServer.savepoint_sql("sp1"), "save transaction sp1");
        assert_eq!(SqlServer.rollback_to_savepoint_sql("sp1"), "rollback transaction sp1");

        assert_eq!(Postgres.savepoint_sql("sp1"), "savepoint sp1");
        assert_eq!(Postgres.rollback_to_savepoint_sql("sp1"), "rollback to savepoint sp1");
    }

    /// The clause a queue uses to claim a row, per database.
    ///
    /// Written as a table because the shapes genuinely differ: two of them
    /// append a clause, SQL Server puts a hint after the table name, and
    /// SQLite has no row locks at all. A queue that assumed one shape is what
    /// this test exists to stop.
    #[test]
    fn each_database_claims_a_row_in_its_own_way() {
        assert_eq!(Postgres.skip_locked(), ("", " for update skip locked"));
        assert_eq!(MySql.skip_locked(), ("", " for update skip locked"));
        assert_eq!(SqlServer.skip_locked(), (" with (updlock, readpast, rowlock)", ""));
        // No clause — SQLite locks the whole database for a writer, so the
        // guarantee comes from `begin immediate` instead.
        assert_eq!(Sqlite.skip_locked(), ("", ""));
        assert_eq!(Sqlite.begin_sql(), "begin immediate");

        // The hint and the clause are never both set: a statement builder
        // appends both, and a database wanting two locks would be a mistake in
        // one of the rows above.
        for dialect in [&Postgres as &dyn Dialect, &MySql, &SqlServer, &Sqlite] {
            let (hint, clause) = dialect.skip_locked();
            assert!(
                hint.is_empty() || clause.is_empty(),
                "{} sets both a hint and a clause",
                dialect.name()
            );
        }
    }

    #[test]
    fn schema_expressions_are_what_each_database_calls_them() {
        assert_eq!(Postgres.current_schema_expression(), "current_schema()");
        assert_eq!(MySql.current_schema_expression(), "database()");
        assert_eq!(SqlServer.current_schema_expression(), "schema_name()");
    }

    #[test]
    fn every_dialect_can_enumerate_its_own_tables() {
        for dialect in all() {
            let sql = dialect.list_tables_sql();

            assert!(sql.starts_with("select "), "{}: {sql}", dialect.name());
            // The query must be scoped to the current schema, or `migrate:fresh`
            // would reach into someone else's database.
            assert!(
                sql.contains("current_schema()")
                    || sql.contains("database()")
                    || sql.contains("schema_name()"),
                "{} does not scope its table list: {sql}",
                dialect.name()
            );
        }
    }

    #[test]
    fn sql_server_adds_a_column_without_saying_column() {
        // Confirmed against a live server: `add column` is a syntax error there,
        // while `drop column` is required. The asymmetry is real.
        assert_eq!(Postgres.add_column_clause(), "add column");
        assert_eq!(MySql.add_column_clause(), "add column");
        assert_eq!(SqlServer.add_column_clause(), "add");
    }

    #[test]
    fn sql_server_never_lists_microsofts_own_tables() {
        // `master` ships system tables in dbo; dropping those is not what
        // `migrate:fresh` is for.
        assert!(SqlServer.list_tables_sql().contains("is_ms_shipped = 0"));
    }

    #[test]
    fn dropping_a_table_takes_its_dependants_with_it() {
        // PostgreSQL says so explicitly; the others need the enforcement
        // switched off around the whole run instead.
        assert!(Postgres.drop_table_sql("users").ends_with("cascade"));
        assert!(Postgres.disable_foreign_keys_sql().is_none());

        assert_eq!(MySql.drop_table_sql("users"), "drop table if exists `users`");
        assert!(MySql.disable_foreign_keys_sql().is_some());
        assert!(MySql.enable_foreign_keys_sql().is_some());

        assert_eq!(SqlServer.drop_table_sql("users"), "drop table if exists [users]");
        assert!(SqlServer.disable_foreign_keys_sql().is_some());
    }

    #[test]
    fn dialects_are_found_by_the_names_people_use() {
        for (name, expected) in [
            ("postgres", "postgres"),
            ("postgresql", "postgres"),
            ("mysql", "mysql"),
            ("mariadb", "mysql"),
            ("sqlserver", "sqlserver"),
            ("mssql", "sqlserver"),
            ("MySQL", "mysql"),
        ] {
            assert_eq!(by_name(name).unwrap().name(), expected, "for {name}");
        }
    }

    #[test]
    fn an_unknown_database_lists_the_ones_that_exist() {
        let error = by_name("oracle").unwrap_err().to_string();

        assert!(error.contains("postgres, mysql, sqlserver"), "{error}");
    }
}