rustango 0.61.13

//! Editable DB-override layer for translations — issue #532 Slice 1.
//!
//! `Translator::from_directory` loads `locales/<lang>.json` on boot and
//! that's the *only* ingest path: fixing a typo in `fr.json` means an
//! editor + a redeploy. This module adds the hybrid storage layer the
//! issue's Slice 1 calls for — a `rustango_translations` table whose
//! rows override the file catalogs, so operators (eventually via the
//! Slice 2 admin UI) can edit translations without shipping code.
//!
//! Lookup order in [`crate::i18n::Translator::translate`] becomes:
//!
//! 1. DB override (this table, loaded into the in-memory override map
//!    via [`refresh_overrides_pool`])
//! 2. File-loaded JSON catalog (read-only seed, unchanged)
//! 3. Default-locale catalog (existing fallback chain)
//!
//! Files stay the source of truth for shipped defaults;
//! [`seed_from_translator_pool`] copies them into the DB once
//! (`ON CONFLICT DO NOTHING`), and edits live as DB rows on top.
//!
//! The table is created by [`ensure_table_pool`] across all three
//! dialects (mirroring `audit` / `contenttypes`) rather than the
//! migration graph, so it works for any app holding a [`Pool`] —
//! tenancy or not. `migrate` creates it; tenancy apps keep it on the
//! registry pool.

use crate::i18n::Translator;
use crate::sql::{Auto, ExecError, Pool};
use crate::Model;

/// One editable translation, table `rustango_translations`. The
/// `(locale, key)` pair is unique — one editable value per key per
/// locale.
///
/// `managed = false`: created by [`ensure_table_pool`], not the
/// migration graph. `migrate` runs it on the single database or the
/// tenancy registry.
///
/// Registry scope: the overrides are one process-wide map, so a tenant
/// admin must not list them (#2360).
///
/// `created_at` / `updated_at` carry the Slice 2 audit trail. Mapped
/// onto the model as of #1464 so the ORM writes them: left DB-defaulted
/// they arrived legacy-shaped forever on an upgraded SQLite file, whose
/// `CURRENT_TIMESTAMP` default `ALTER TABLE` cannot replace.
#[derive(Model, Debug, Clone, serde::Serialize)]
#[rustango(table = "rustango_translations", managed = false, scope = "registry")]
pub struct Translation {
    #[rustango(primary_key)]
    pub id: Auto<i64>,
    #[rustango(max_length = 16)]
    pub locale: String,
    #[rustango(max_length = 200)]
    pub key: String,
    pub value: String,
    #[rustango(max_length = 200, default = "")]
    pub updated_by: String,
    #[rustango(auto_now_add)]
    pub created_at: Auto<chrono::DateTime<chrono::Utc>>,
    #[rustango(auto_now)]
    pub updated_at: Auto<chrono::DateTime<chrono::Utc>>,
}

/// `max_length` of `locale` and `key`.
const LOCALE_LEN: u32 = 16;
const KEY_LEN: u32 = 200;

/// `true` when `(locale, key)` fits the table; a longer catalog key stays
/// file-only rather than failing the rows beside it.
pub(crate) fn storable(locale: &str, key: &str) -> bool {
    let fits = |s: &str, max: u32| u32::try_from(s.chars().count()).is_ok_and(|n| n <= max);
    let ok = fits(locale, LOCALE_LEN) && fits(key, KEY_LEN);
    if !ok {
        tracing::warn!(
            target: "rustango::i18n",
            locale,
            key_chars = key.chars().count(),
            "translation key longer than {KEY_LEN} chars (or locale over {LOCALE_LEN}) \
             is not stored in the database; it keeps its file value"
        );
    }
    ok
}

/// The `CREATE TABLE` for this dialect.
///
/// Both timestamp columns take type *and* `DEFAULT` from
/// [`crate::sql::Dialect::timestamp_now_column`]. That also moves MySQL
/// off `TIMESTAMP` (whole seconds) onto `DATETIME(6)`, now that the ORM
/// binds microseconds into them — but `IF NOT EXISTS` leaves an
/// existing MySQL table truncating, which needs a migration.
///
/// **No `--` comments inside the returned string.** Its newlines
/// collapse before the driver sees it, so a `--` swallows the rest of
/// the statement and SQLite rejects it with `incomplete input`.
pub(crate) fn ddl(dialect: &dyn crate::sql::Dialect) -> String {
    let ts = dialect.timestamp_now_column();
    // Case-sensitive keys: a MySQL `_ci` default would make `Key` and `key` collide.
    let (locale, key) = (
        dialect.cs_text_type(LOCALE_LEN),
        dialect.cs_text_type(KEY_LEN),
    );
    match dialect.name() {
        "mysql" => format!(
            r"
CREATE TABLE IF NOT EXISTS `rustango_translations` (
    `id`         BIGINT AUTO_INCREMENT PRIMARY KEY,
    `locale`     {locale} NOT NULL,
    `key`        {key} NOT NULL,
    `value`      TEXT NOT NULL,
    `updated_by` VARCHAR(200) NOT NULL DEFAULT '',
    `created_at` {ts},
    `updated_at` {ts},
    CONSTRAINT `rustango_translations_locale_key_uq` UNIQUE (`locale`, `key`)
);
"
        ),
        "sqlite" => format!(
            r#"
CREATE TABLE IF NOT EXISTS "rustango_translations" (
    "id"         INTEGER PRIMARY KEY AUTOINCREMENT,
    "locale"     TEXT NOT NULL,
    "key"        TEXT NOT NULL,
    "value"      TEXT NOT NULL,
    "updated_by" TEXT NOT NULL DEFAULT '',
    "created_at" {ts},
    "updated_at" {ts},
    CONSTRAINT "rustango_translations_locale_key_uq" UNIQUE ("locale", "key")
);
"#
        ),
        // Postgres + any future dialect: the standard `"`-quoted DDL.
        _ => format!(
            r#"
CREATE TABLE IF NOT EXISTS "rustango_translations" (
    "id"         BIGSERIAL PRIMARY KEY,
    "locale"     {locale} NOT NULL,
    "key"        {key} NOT NULL,
    "value"      TEXT NOT NULL,
    "updated_by" VARCHAR(200) NOT NULL DEFAULT '',
    "created_at" {ts},
    "updated_at" {ts},
    CONSTRAINT "rustango_translations_locale_key_uq" UNIQUE ("locale", "key")
);
"#
        ),
    }
}

/// Create `rustango_translations` if absent — idempotent, dispatched per
/// dialect. Call once at startup (and the Slice 2 `seed-translations`
/// verb calls it too).
///
/// # Errors
/// Driver / SQL failures other than the duplicate-object errors that
/// [`crate::sql::run_ddl_idempotent`] swallows. A refused collation upgrade
/// only warns: the table keeps working, case-insensitively.
pub async fn ensure_table_pool(pool: &Pool) -> Result<(), sqlx::Error> {
    crate::sql::run_ddl_idempotent(pool, &ddl(pool.dialect())).await?;
    if refused(pool) {
        return Ok(());
    }
    if let Err(e) = case_sensitive_keys(pool).await {
        tracing::warn!(
            error = %e,
            "rustango_translations keeps case-insensitive keys: moving them needs ALTER"
        );
        // Each pool tries once per process, not on every ensure call.
        refused_pools().push(crate::sql::PoolId::of(pool));
    }
    Ok(())
}

fn refused_pools() -> std::sync::MutexGuard<'static, Vec<crate::sql::PoolId>> {
    static REFUSED: std::sync::Mutex<Vec<crate::sql::PoolId>> = std::sync::Mutex::new(Vec::new());
    REFUSED
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
}

/// `true` when this pool's upgrade already failed; drops closed pools' entries.
fn refused(pool: &Pool) -> bool {
    let mut ids = refused_pools();
    ids.retain(|id| !id.is_closed());
    ids.iter().any(|id| id.is(pool))
}

/// A table an older release created took the database's collation, which
/// may ignore case: move its `locale` / `key` to [`ddl`]'s.
async fn case_sensitive_keys(pool: &Pool) -> Result<(), ExecError> {
    let d = pool.dialect();
    let Some(probe) = d.ci_columns_sql("rustango_translations") else {
        return Ok(());
    };
    let found: Vec<(String,)> = crate::sql::raw_query_pool(&probe.sql, probe.params, pool).await?;
    for (column,) in found {
        let len = match column.as_str() {
            "locale" => LOCALE_LEN,
            "key" => KEY_LEN,
            _ => continue,
        };
        let sql = format!(
            "ALTER TABLE {} MODIFY COLUMN {} {} NOT NULL",
            d.quote_ident("rustango_translations"),
            d.quote_ident(&column),
            d.cs_text_type(len),
        );
        crate::sql::raw_execute_pool(pool, &sql, Vec::new()).await?;
    }
    Ok(())
}

/// Every override row, for the admin list view / [`refresh_overrides_pool`].
///
/// # Errors
/// As the ORM fetch path ([`ExecError`]).
pub async fn all_pool(pool: &Pool) -> Result<Vec<Translation>, ExecError> {
    use crate::sql::FetcherPool as _;
    Translation::objects().fetch(pool).await
}

/// Insert-or-update the editable value for `(locale, key)` — the
/// "edit without redeploy" write. `updated_by` records who made the
/// change (operator id / username); pass `""` if not tracked.
///
/// # Errors
/// As the ORM upsert path ([`ExecError`]). A `key` over 200 chars or a
/// `locale` over 16 fails validation with `MaxLengthExceeded` before
/// any SQL runs; the seed and the editor skip those keys instead.
pub async fn upsert_pool(
    pool: &Pool,
    locale: &str,
    key: &str,
    value: &str,
    updated_by: &str,
) -> Result<(), ExecError> {
    let row = Translation {
        id: Auto::Unset,
        locale: locale.to_owned(),
        key: key.to_owned(),
        value: value.to_owned(),
        updated_by: updated_by.to_owned(),
        created_at: Auto::Unset,
        updated_at: Auto::Unset,
    };
    // `updated_at` is in the DO UPDATE list: an upsert that lands on
    // the conflict branch is an edit, and the audit trail is the point
    // of the column.
    Translation::bulk_upsert_pool(
        &[row],
        &["locale", "key"],
        &["value", "updated_by", "updated_at"],
        pool,
    )
    .await
}

/// Seed the DB layer from a [`Translator`]'s file catalogs. Existing
/// `(locale, key)` rows are left untouched (`ON CONFLICT DO NOTHING`),
/// so files stay the source of truth for shipped defaults and only
/// previously-unknown keys are added. Idempotent — safe to run on every
/// boot. Returns the number of file entries processed. A key longer than
/// the `key` column is skipped with a warning and keeps its file value.
///
/// The on-disk path is the caller's: load with
/// `Translator::from_directory(dir, default)?` then hand the result here
/// (keeps the filesystem error out of [`ExecError`]).
///
/// # Errors
/// As the ORM bulk-insert path ([`ExecError`]).
pub async fn seed_from_translator_pool(
    pool: &Pool,
    translator: &Translator,
) -> Result<usize, ExecError> {
    let rows: Vec<Translation> = translator
        .entries()
        .into_iter()
        .filter(|(locale, key, _)| storable(locale, key))
        .map(|(locale, key, value)| Translation {
            id: Auto::Unset,
            locale,
            key,
            value,
            updated_by: String::new(),
            created_at: Auto::Unset,
            updated_at: Auto::Unset,
        })
        .collect();
    let n = rows.len();
    if n > 0 {
        Translation::bulk_insert_or_ignore_pool(&rows, pool).await?;
    }
    Ok(n)
}

/// Delete every override row for `key` (all locales) — the Slice 2
/// editor's "delete key" action. Returns the number of rows removed (0
/// if the key had no overrides). The file-catalog default, if any, is
/// untouched, so a deleted key falls back to its shipped value.
///
/// # Errors
/// As the ORM delete path ([`ExecError`]).
pub async fn delete_key_pool(pool: &Pool, key: &str) -> Result<u64, ExecError> {
    use crate::core::Model as _; // brings `Translation::SCHEMA` into scope
    use crate::core::{DeleteQuery, Filter, Op, SqlValue, WhereExpr};
    crate::sql::delete_pool(
        pool,
        &DeleteQuery {
            model: Translation::SCHEMA,
            where_clause: WhereExpr::and_predicates(vec![Filter {
                column: "key",
                op: Op::Eq,
                value: SqlValue::from(key),
            }]),
        },
    )
    .await
}

/// Delete one locale's override for `key`, so that locale falls back to
/// the file catalog (#2091). Returns the rows removed (0 or 1).
///
/// # Errors
/// As the ORM delete path ([`ExecError`]).
pub async fn delete_override(pool: &Pool, locale: &str, key: &str) -> Result<u64, ExecError> {
    use crate::core::Model as _;
    use crate::core::{DeleteQuery, Filter, Op, SqlValue, WhereExpr};
    crate::sql::delete_pool(
        pool,
        &DeleteQuery {
            model: Translation::SCHEMA,
            where_clause: WhereExpr::and_predicates(vec![
                Filter {
                    column: "locale",
                    op: Op::Eq,
                    value: SqlValue::from(locale),
                },
                Filter {
                    column: "key",
                    op: Op::Eq,
                    value: SqlValue::from(key),
                },
            ]),
        },
    )
    .await
}

/// Translators the admin translations editor refreshes after each save.
static LIVE_TRANSLATORS: std::sync::Mutex<Vec<std::sync::Weak<Translator>>> =
    std::sync::Mutex::new(Vec::new());

/// Let the admin translations editor refresh `translator` after each save.
/// Only this process is refreshed: other replicas still need their own
/// [`refresh_overrides_pool`] (on a timer or a signal).
pub fn register_live_translator(translator: &std::sync::Arc<Translator>) {
    let mut live = LIVE_TRANSLATORS
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    live.retain(|w| w.strong_count() > 0);
    live.push(std::sync::Arc::downgrade(translator));
}

/// [`refresh_overrides_pool`] for every [`register_live_translator`] one.
///
/// # Errors
/// The first failed reload.
pub async fn refresh_live_translators(pool: &Pool) -> Result<(), ExecError> {
    let live: Vec<std::sync::Arc<Translator>> = LIVE_TRANSLATORS
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .iter()
        .filter_map(std::sync::Weak::upgrade)
        .collect();
    for t in live {
        refresh_overrides_pool(&t, pool).await?;
    }
    Ok(())
}

/// Reload `translator`'s override layer from the DB so subsequent
/// `translate(...)` calls reflect persisted edits — call after seeding,
/// on boot, and after each admin save. Returns the row count loaded.
///
/// # Errors
/// As the ORM fetch path ([`ExecError`]).
pub async fn refresh_overrides_pool(
    translator: &Translator,
    pool: &Pool,
) -> Result<usize, ExecError> {
    let rows = all_pool(pool).await?;
    let n = rows.len();
    translator.load_overrides(rows.into_iter().map(|t| (t.locale, t.key, t.value)));
    Ok(n)
}