nyaKazuha/BlueArchiveToolkit
1
0
Fork 0
mirror of https://github.com/Yuyi-Oak/BlueArchiveToolkit.git synced 2026-09-18 06:34:54 +08:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

fix(sqlite): 收口长期状态数据库版本化迁移契约
bat-rust / Build and test Rust (push) Canceled after 0s
Details
bat-rust / Build and test Go API (push) Canceled after 0s
Details

Browse Source
This commit is contained in:
nyaKazuha
2026-09-16 21:02:22 +08:00
parent 99355effe4
commit 7f7d757f15
10 changed files with 2487 additions and 421 deletions
Download Patch File Download Diff File Expand all files Collapse all files
infrastructure/src/glossary.rs
+686 -140
View File
@@ -3,6 +3,9 @@
use crate::path_security::{
ensure_safe_directory_path, lexical_absolute, set_file_mode, STATE_FILE_MODE,
};
use crate::sqlite_migration::{
self, ExpectedColumn, ExpectedIndex, ExpectedTable, SqliteSchemaSnapshot,
};
use async_trait::async_trait;
use bat_core::domain::{
evaluate_glossary, validate_glossary_draft, GlossaryEvaluation, GlossaryHistoryRecord,
@@ -12,7 +15,7 @@ use bat_core::domain::{
use bat_core::repositories::GlossaryRepository;
use bat_core::{Error, Result};
use serde::de::DeserializeOwned;
use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode};
use sqlx::{Row, SqlitePool};
use std::path::{Path, PathBuf};
use std::str::FromStr;
@@ -76,15 +79,20 @@ impl SqliteGlossaryRepository {
absolute.display()
)));
}
if metadata.len() > 0 {
let snapshot =
sqlite_migration::read_only_preflight(&absolute, GLOSSARY_SCHEMA_COMPONENT)
.await
.map_err(db_error)?;
classify_glossary_schema(&snapshot)?;
}
}
let options = SqliteConnectOptions::from_str(&format!("sqlite://{}", absolute.display()))
.map_err(|error| Error::Other(error.into()))?
.create_if_missing(create_if_missing)
.journal_mode(SqliteJournalMode::Wal)
.busy_timeout(Duration::from_secs(30));
let pool = SqlitePoolOptions::new()
.max_connections(1)
.connect_with(options)
let pool = sqlite_migration::connect_writable_pool(options)
.await
.map_err(db_error)?;
if create_if_missing {
@@ -97,119 +105,86 @@ impl SqliteGlossaryRepository {
}
async fn init_schema(&self) -> Result<()> {
sqlx::query(
"CREATE TABLE IF NOT EXISTS schema_migrations (
component TEXT PRIMARY KEY NOT NULL,
version INTEGER NOT NULL CHECK(version >= 1)
)",
)
.execute(&self.pool)
.await
.map_err(db_error)?;
sqlx::query(
"CREATE TABLE IF NOT EXISTS glossary_terms (
term_id TEXT PRIMARY KEY NOT NULL,
source_term TEXT NOT NULL,
aliases_json TEXT NOT NULL,
recommended_translation TEXT NOT NULL,
allowed_translations_json TEXT NOT NULL,
source_language TEXT,
target_language TEXT,
category TEXT,
priority INTEGER NOT NULL,
scope_json TEXT NOT NULL,
review_status TEXT NOT NULL,
source_kind TEXT NOT NULL,
source_ref TEXT,
source_author TEXT,
source_note TEXT,
source_observed_unix_seconds INTEGER NOT NULL,
created_unix_seconds INTEGER NOT NULL,
updated_unix_seconds INTEGER NOT NULL,
CHECK(length(term_id) > 0),
CHECK(length(source_term) > 0),
CHECK(length(recommended_translation) > 0),
CHECK(review_status IN ('draft', 'approved', 'deprecated', 'rejected')),
CHECK(source_kind IN ('manual', 'imported'))
)",
)
.execute(&self.pool)
.await
.map_err(db_error)?;
ensure_column(
&self.pool,
"glossary_terms",
"source_observed_unix_seconds",
"INTEGER NOT NULL DEFAULT 1",
)
.await?;
sqlx::query(
"CREATE TABLE IF NOT EXISTS glossary_term_history (
history_id TEXT PRIMARY KEY NOT NULL,
term_id TEXT NOT NULL,
action TEXT NOT NULL,
reviewer TEXT,
reason TEXT,
source_json TEXT NOT NULL,
review_status TEXT NOT NULL,
snapshot_json TEXT NOT NULL,
observed_unix_seconds INTEGER NOT NULL,
FOREIGN KEY(term_id) REFERENCES glossary_terms(term_id)
)",
)
.execute(&self.pool)
.await
.map_err(db_error)?;
sqlx::query(
"CREATE TABLE IF NOT EXISTS glossary_term_deletions (
deletion_id TEXT PRIMARY KEY NOT NULL,
term_id TEXT NOT NULL,
reviewer TEXT NOT NULL,
reason TEXT NOT NULL,
source_json TEXT NOT NULL,
snapshot_json TEXT NOT NULL,
history_json TEXT NOT NULL,
observed_unix_seconds INTEGER NOT NULL
)",
)
.execute(&self.pool)
.await
.map_err(db_error)?;
sqlx::query(
"CREATE INDEX IF NOT EXISTS idx_glossary_status
ON glossary_terms(review_status, priority DESC, term_id)",
)
.execute(&self.pool)
.await
.map_err(db_error)?;
sqlx::query(
"CREATE INDEX IF NOT EXISTS idx_glossary_source_term
ON glossary_terms(source_term)",
)
.execute(&self.pool)
.await
.map_err(db_error)?;
let current: Option<i64> =
sqlx::query_scalar("SELECT version FROM schema_migrations WHERE component = ?1")
.bind(GLOSSARY_SCHEMA_COMPONENT)
.fetch_optional(&self.pool)
self.init_schema_with_failure(None).await
}
#[cfg(test)]
async fn init_schema_with_test_failure(&self, fail_after_step: usize) -> Result<()> {
self.init_schema_with_failure(Some(fail_after_step)).await
}
async fn init_schema_with_failure(&self, fail_after_step: Option<usize>) -> Result<()> {
let mut transaction = sqlite_migration::begin_immediate(&self.pool)
.await
.map_err(db_error)?;
let result = self
.migrate_in_transaction(&mut transaction, fail_after_step)
.await;
match result {
Ok(()) => transaction.commit().await.map_err(db_error),
Err(error) => {
let _ = transaction.rollback().await;
Err(error)
}
}
}
async fn migrate_in_transaction(
&self,
transaction: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
fail_after_step: Option<usize>,
) -> Result<()> {
let snapshot =
sqlite_migration::snapshot_connection(transaction.as_mut(), GLOSSARY_SCHEMA_COMPONENT)
.await
.map_err(db_error)?;
if current.is_some_and(|version| version > i64::from(GLOSSARY_SCHEMA_VERSION)) {
return Err(Error::InvalidArgument(format!(
"不支持的 Glossary schema 版本：{}",
current.unwrap_or_default()
)));
match classify_glossary_schema(&snapshot)? {
GlossarySchemaState::Empty => {
create_glossary_schema(transaction, fail_after_step).await?;
sqlite_migration::write_component_version(
transaction,
GLOSSARY_SCHEMA_COMPONENT,
GLOSSARY_SCHEMA_VERSION,
)
.await
.map_err(db_error)?;
}
GlossarySchemaState::Version(version) if version == GLOSSARY_SCHEMA_VERSION => {
if snapshot.component_version.is_none() {
if !snapshot
.tables
.contains_key(sqlite_migration::SCHEMA_MIGRATIONS_TABLE)
{
sqlite_migration::create_schema_migrations_table(transaction)
.await
.map_err(db_error)?;
}
sqlite_migration::write_component_version(
transaction,
GLOSSARY_SCHEMA_COMPONENT,
GLOSSARY_SCHEMA_VERSION,
)
.await
.map_err(db_error)?;
}
}
GlossarySchemaState::Version(version) => {
return Err(invalid_glossary_schema(format!(
"内部不支持的迁移起点 {version}"
)));
}
}
let final_snapshot =
sqlite_migration::snapshot_connection(transaction.as_mut(), GLOSSARY_SCHEMA_COMPONENT)
.await
.map_err(db_error)?;
if final_snapshot.component_version != Some(i64::from(GLOSSARY_SCHEMA_VERSION))
|| !matches_glossary_fingerprint(&final_snapshot)
{
return Err(invalid_glossary_schema(
"migration 结果与当前 schema fingerprint 不一致".to_string(),
));
}
sqlx::query(
"INSERT INTO schema_migrations(component, version) VALUES (?1, ?2)
ON CONFLICT(component) DO UPDATE SET version = excluded.version",
)
.bind(GLOSSARY_SCHEMA_COMPONENT)
.bind(i64::from(GLOSSARY_SCHEMA_VERSION))
.execute(&self.pool)
.await
.map_err(db_error)?;
Ok(())
}
@@ -743,6 +718,450 @@ fn row_to_history(row: sqlx::sqlite::SqliteRow) -> Result<GlossaryHistoryRecord>
})
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum GlossarySchemaState {
Empty,
Version(u32),
}
fn classify_glossary_schema(snapshot: &SqliteSchemaSnapshot) -> Result<GlossarySchemaState> {
if snapshot.is_empty() {
return Ok(GlossarySchemaState::Empty);
}
if let Some(observed) = snapshot.component_version {
if observed > i64::from(GLOSSARY_SCHEMA_VERSION) {
return Err(invalid_glossary_schema(format!(
"不支持的 Glossary schema 版本：observed={observed}, supported={GLOSSARY_SCHEMA_VERSION}"
)));
}
if observed < 1 {
return Err(invalid_glossary_schema(format!(
"schema version 无效：observed={observed}, supported=1..={GLOSSARY_SCHEMA_VERSION}"
)));
}
}
if (matches_glossary_fingerprint(snapshot)
|| (snapshot.component_version.is_none()
&& matches_glossary_component_fingerprint(snapshot)))
&& snapshot
.component_version
.is_none_or(|observed| observed == i64::from(GLOSSARY_SCHEMA_VERSION))
{
return Ok(GlossarySchemaState::Version(GLOSSARY_SCHEMA_VERSION));
}
match snapshot.component_version {
Some(observed) => Err(invalid_glossary_schema(format!(
"schema version 与实际结构不一致：observed={observed}, supported={GLOSSARY_SCHEMA_VERSION}"
))),
None => Err(invalid_glossary_schema(
"未识别的 legacy schema，拒绝静默修复".to_string(),
)),
}
}
fn invalid_glossary_schema(detail: String) -> Error {
Error::InvalidArgument(format!("Glossary schema 无效：{detail}"))
}
fn matches_glossary_fingerprint(snapshot: &SqliteSchemaSnapshot) -> bool {
let tables = [
sqlite_migration::schema_migrations_table(),
ExpectedTable {
name: "glossary_terms",
columns: &GLOSSARY_TERM_COLUMNS,
},
ExpectedTable {
name: "glossary_term_history",
columns: &GLOSSARY_HISTORY_COLUMNS,
},
ExpectedTable {
name: "glossary_term_deletions",
columns: &GLOSSARY_DELETION_COLUMNS,
},
];
let indexes = [
ExpectedIndex {
table: "glossary_terms",
name: "idx_glossary_status",
columns: &["review_status", "priority", "term_id"],
},
ExpectedIndex {
table: "glossary_terms",
name: "idx_glossary_source_term",
columns: &["source_term"],
},
];
matches_glossary_fingerprint_with_tables(snapshot, &tables, &indexes)
}
fn matches_glossary_component_fingerprint(snapshot: &SqliteSchemaSnapshot) -> bool {
let tables = [
ExpectedTable {
name: "glossary_terms",
columns: &GLOSSARY_TERM_COLUMNS,
},
ExpectedTable {
name: "glossary_term_history",
columns: &GLOSSARY_HISTORY_COLUMNS,
},
ExpectedTable {
name: "glossary_term_deletions",
columns: &GLOSSARY_DELETION_COLUMNS,
},
];
let indexes = [
ExpectedIndex {
table: "glossary_terms",
name: "idx_glossary_status",
columns: &["review_status", "priority", "term_id"],
},
ExpectedIndex {
table: "glossary_terms",
name: "idx_glossary_source_term",
columns: &["source_term"],
},
];
matches_glossary_fingerprint_with_tables(snapshot, &tables, &indexes)
}
fn matches_glossary_fingerprint_with_tables(
snapshot: &SqliteSchemaSnapshot,
tables: &[ExpectedTable<'_>],
indexes: &[ExpectedIndex<'_>],
) -> bool {
sqlite_migration::matches_fingerprint(snapshot, tables, indexes)
}
const GLOSSARY_TERM_COLUMNS: [ExpectedColumn<'static>; 18] = [
ExpectedColumn {
name: "term_id",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: true,
},
ExpectedColumn {
name: "source_term",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "aliases_json",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "recommended_translation",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "allowed_translations_json",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "source_language",
data_type: "TEXT",
not_null: false,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "target_language",
data_type: "TEXT",
not_null: false,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "category",
data_type: "TEXT",
not_null: false,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "priority",
data_type: "INTEGER",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "scope_json",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "review_status",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "source_kind",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "source_ref",
data_type: "TEXT",
not_null: false,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "source_author",
data_type: "TEXT",
not_null: false,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "source_note",
data_type: "TEXT",
not_null: false,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "source_observed_unix_seconds",
data_type: "INTEGER",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "created_unix_seconds",
data_type: "INTEGER",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "updated_unix_seconds",
data_type: "INTEGER",
not_null: true,
default_value: None,
primary_key: false,
},
];
const GLOSSARY_HISTORY_COLUMNS: [ExpectedColumn<'static>; 9] = [
ExpectedColumn {
name: "history_id",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: true,
},
ExpectedColumn {
name: "term_id",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "action",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "reviewer",
data_type: "TEXT",
not_null: false,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "reason",
data_type: "TEXT",
not_null: false,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "source_json",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "review_status",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "snapshot_json",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "observed_unix_seconds",
data_type: "INTEGER",
not_null: true,
default_value: None,
primary_key: false,
},
];
const GLOSSARY_DELETION_COLUMNS: [ExpectedColumn<'static>; 8] = [
ExpectedColumn {
name: "deletion_id",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: true,
},
ExpectedColumn {
name: "term_id",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "reviewer",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "reason",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "source_json",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "snapshot_json",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "history_json",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "observed_unix_seconds",
data_type: "INTEGER",
not_null: true,
default_value: None,
primary_key: false,
},
];
async fn create_glossary_schema(
transaction: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
fail_after_step: Option<usize>,
) -> Result<()> {
let steps = [
"CREATE TABLE schema_migrations (
component TEXT PRIMARY KEY NOT NULL,
version INTEGER NOT NULL CHECK(version >= 1)
)",
"CREATE TABLE glossary_terms (
term_id TEXT PRIMARY KEY NOT NULL,
source_term TEXT NOT NULL,
aliases_json TEXT NOT NULL,
recommended_translation TEXT NOT NULL,
allowed_translations_json TEXT NOT NULL,
source_language TEXT,
target_language TEXT,
category TEXT,
priority INTEGER NOT NULL,
scope_json TEXT NOT NULL,
review_status TEXT NOT NULL,
source_kind TEXT NOT NULL,
source_ref TEXT,
source_author TEXT,
source_note TEXT,
source_observed_unix_seconds INTEGER NOT NULL,
created_unix_seconds INTEGER NOT NULL,
updated_unix_seconds INTEGER NOT NULL,
CHECK(length(term_id) > 0),
CHECK(length(source_term) > 0),
CHECK(length(recommended_translation) > 0),
CHECK(review_status IN ('draft', 'approved', 'deprecated', 'rejected')),
CHECK(source_kind IN ('manual', 'imported'))
)",
"CREATE TABLE glossary_term_history (
history_id TEXT PRIMARY KEY NOT NULL,
term_id TEXT NOT NULL,
action TEXT NOT NULL,
reviewer TEXT,
reason TEXT,
source_json TEXT NOT NULL,
review_status TEXT NOT NULL,
snapshot_json TEXT NOT NULL,
observed_unix_seconds INTEGER NOT NULL,
FOREIGN KEY(term_id) REFERENCES glossary_terms(term_id)
)",
"CREATE TABLE glossary_term_deletions (
deletion_id TEXT PRIMARY KEY NOT NULL,
term_id TEXT NOT NULL,
reviewer TEXT NOT NULL,
reason TEXT NOT NULL,
source_json TEXT NOT NULL,
snapshot_json TEXT NOT NULL,
history_json TEXT NOT NULL,
observed_unix_seconds INTEGER NOT NULL
)",
"CREATE INDEX idx_glossary_status
ON glossary_terms(review_status, priority DESC, term_id)",
"CREATE INDEX idx_glossary_source_term
ON glossary_terms(source_term)",
];
for (index, statement) in steps.iter().enumerate() {
sqlx::query(statement)
.execute(&mut **transaction)
.await
.map_err(db_error)?;
if fail_after_step == Some(index + 1) {
return Err(Error::Other(anyhow::anyhow!(
"Glossary migration failed after step {}",
index + 1
)));
}
}
Ok(())
}
fn parse_json<T: DeserializeOwned>(value: String) -> Result<T> {
serde_json::from_str(&value).map_err(|error| Error::Serialization(error.to_string()))
}
@@ -769,35 +1188,10 @@ fn db_error(error: sqlx::Error) -> Error {
Error::Other(error.into())
}
async fn ensure_column(
pool: &SqlitePool,
table: &str,
column: &str,
definition: &str,
) -> Result<()> {
let columns = sqlx::query(&format!("PRAGMA table_info({table})"))
.fetch_all(pool)
.await
.map_err(db_error)?;
let exists = columns.iter().any(|row| {
row.try_get::<String, _>("name")
.map(|name| name == column)
.unwrap_or(false)
});
if !exists {
sqlx::query(&format!(
"ALTER TABLE {table} ADD COLUMN {column} {definition}"
))
.execute(pool)
.await
.map_err(db_error)?;
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use sqlx::sqlite::SqlitePoolOptions;
use std::collections::BTreeMap;
fn draft(status: GlossaryReviewStatus) -> GlossaryTermDraft {
@@ -825,6 +1219,158 @@ mod tests {
}
}
async fn raw_repository(
path: &std::path::Path,
create_if_missing: bool,
) -> SqliteGlossaryRepository {
let options = SqliteConnectOptions::from_str(&format!("sqlite://{}", path.display()))
.unwrap()
.create_if_missing(create_if_missing);
let pool = SqlitePoolOptions::new()
.max_connections(1)
.connect_with(options)
.await
.unwrap();
SqliteGlossaryRepository { pool }
}
#[tokio::test]
async fn sqlite_glossary_preserves_data_across_reopen_and_missing_row() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join(GLOSSARY_REPOSITORY_FILE);
let repository = SqliteGlossaryRepository::new(&path).await.unwrap();
repository
.add(draft(GlossaryReviewStatus::Draft))
.await
.unwrap();
repository
.review(
"term-sensei",
GlossaryReviewStatus::Approved,
"reviewer",
Some("accepted".to_string()),
)
.await
.unwrap();
sqlx::query("DROP TABLE schema_migrations")
.execute(&repository.pool)
.await
.unwrap();
repository.pool.close().await;
let reopened = SqliteGlossaryRepository::open(&path).await.unwrap();
let term = reopened.find("term-sensei").await.unwrap();
assert_eq!(term.definition.recommended_translation, "老师");
assert_eq!(term.history.len(), 2);
assert_eq!(term.history[1].action, "approved");
assert_eq!(
reopened.summary().await.unwrap().schema_version,
GLOSSARY_SCHEMA_VERSION
);
}
#[tokio::test]
async fn sqlite_glossary_future_schema_is_read_only_failure() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join(GLOSSARY_REPOSITORY_FILE);
let repository = SqliteGlossaryRepository::new(&path).await.unwrap();
sqlx::query("UPDATE schema_migrations SET version = ?2 WHERE component = ?1")
.bind(GLOSSARY_SCHEMA_COMPONENT)
.bind(i64::from(GLOSSARY_SCHEMA_VERSION) + 1)
.execute(&repository.pool)
.await
.unwrap();
repository.pool.close().await;
let before = std::fs::read(&path).unwrap();
let wal_path = std::path::PathBuf::from(format!("{}-wal", path.display()));
let shm_path = std::path::PathBuf::from(format!("{}-shm", path.display()));
let wal_before = std::fs::read(&wal_path).ok();
let shm_before = std::fs::read(&shm_path).ok();
let error = SqliteGlossaryRepository::new(&path).await.unwrap_err();
assert!(error.to_string().contains("不支持的 Glossary schema"));
assert_eq!(std::fs::read(&path).unwrap(), before);
assert_eq!(std::fs::read(&wal_path).ok(), wal_before);
assert_eq!(std::fs::read(&shm_path).ok(), shm_before);
}
#[tokio::test]
async fn sqlite_glossary_unknown_schema_fails_closed() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join(GLOSSARY_REPOSITORY_FILE);
let repository = raw_repository(&path, true).await;
sqlx::query(
"CREATE TABLE schema_migrations (
component TEXT PRIMARY KEY NOT NULL,
version INTEGER NOT NULL CHECK(version >= 1)
)",
)
.execute(&repository.pool)
.await
.unwrap();
sqlx::query("INSERT INTO schema_migrations(component, version) VALUES (?1, 1)")
.bind(GLOSSARY_SCHEMA_COMPONENT)
.execute(&repository.pool)
.await
.unwrap();
sqlx::query("CREATE TABLE glossary_terms (term_id TEXT PRIMARY KEY)")
.execute(&repository.pool)
.await
.unwrap();
repository.pool.close().await;
let before = std::fs::read(&path).unwrap();
let error = SqliteGlossaryRepository::open(&path).await.unwrap_err();
assert!(error
.to_string()
.contains("schema version 与实际结构不一致"));
assert_eq!(std::fs::read(&path).unwrap(), before);
}
#[tokio::test]
async fn sqlite_glossary_failed_new_schema_rolls_back_and_retries() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join(GLOSSARY_REPOSITORY_FILE);
let repository = raw_repository(&path, true).await;
assert!(repository.init_schema_with_test_failure(3).await.is_err());
let table_count: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM sqlite_master WHERE type = 'table'")
.fetch_one(&repository.pool)
.await
.unwrap();
assert_eq!(table_count, 0);
repository.init_schema().await.unwrap();
let version: i64 =
sqlx::query_scalar("SELECT version FROM schema_migrations WHERE component = ?1")
.bind(GLOSSARY_SCHEMA_COMPONENT)
.fetch_one(&repository.pool)
.await
.unwrap();
assert_eq!(version, 1);
}
#[tokio::test]
async fn sqlite_glossary_concurrent_new_open_has_one_current_schema() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join(GLOSSARY_REPOSITORY_FILE);
let (left, right) = tokio::join!(
SqliteGlossaryRepository::new(&path),
SqliteGlossaryRepository::new(&path)
);
assert!(left.is_ok(), "left open failed: {left:?}");
assert!(right.is_ok(), "right open failed: {right:?}");
let repository = left.unwrap();
drop(right);
assert_eq!(
repository.summary().await.unwrap().schema_version,
GLOSSARY_SCHEMA_VERSION
);
repository.pool.close().await;
let reopened = SqliteGlossaryRepository::open(&path).await.unwrap();
assert_eq!(
reopened.summary().await.unwrap().schema_version,
GLOSSARY_SCHEMA_VERSION
);
}
#[tokio::test]
async fn sqlite_glossary_preserves_history_and_only_approved_terms_match() {
let temp = tempfile::TempDir::new().unwrap();
infrastructure/src/lib.rs
+1
View File
@@ -31,6 +31,7 @@ pub mod path_security;
pub mod release_flow;
pub mod release_ops;
pub mod resources;
mod sqlite_migration;
pub mod translation_memory;
pub mod translation_tasks;
pub mod translation_worker;
infrastructure/src/sqlite_migration.rs
+377
View File
@@ -0,0 +1,377 @@
//! Shared, deliberately small SQLite schema-migration primitives.
//!
//! Component owners still define their own schema fingerprints and migration
//! steps. This module only owns the read-only preflight, schema snapshot, and
//! writer-lock mechanics shared by the long-lived SQLite stores.
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use sqlx::{Row, SqliteConnection, SqlitePool};
use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
use std::str::FromStr;
use std::time::Duration;
pub const SCHEMA_MIGRATIONS_TABLE: &str = "schema_migrations";
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SqliteColumn {
pub data_type: String,
pub not_null: bool,
pub default_value: Option<String>,
pub primary_key: bool,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SqliteSchemaSnapshot {
/// Non-internal SQLite objects, including tables, indexes, views, and
/// triggers. Internal `sqlite_autoindex_*` objects are omitted.
pub objects: BTreeSet<(String, String)>,
pub tables: BTreeMap<String, BTreeMap<String, SqliteColumn>>,
pub indexes: BTreeMap<String, BTreeMap<String, Vec<String>>>,
pub component_version: Option<i64>,
}
impl SqliteSchemaSnapshot {
pub fn is_empty(&self) -> bool {
self.objects.is_empty()
}
}
#[derive(Debug, Clone, Copy)]
pub struct ExpectedColumn<'a> {
pub name: &'a str,
pub data_type: &'a str,
pub not_null: bool,
pub default_value: Option<&'a str>,
pub primary_key: bool,
}
#[derive(Debug, Clone, Copy)]
pub struct ExpectedTable<'a> {
pub name: &'a str,
pub columns: &'a [ExpectedColumn<'a>],
}
#[derive(Debug, Clone, Copy)]
pub struct ExpectedIndex<'a> {
pub table: &'a str,
pub name: &'a str,
pub columns: &'a [&'a str],
}
/// Opens an existing database with SQLite's read-only flag and snapshots its
/// schema before a writable connection can perform any mutation.
pub async fn read_only_preflight(
path: &Path,
component: &str,
) -> Result<SqliteSchemaSnapshot, sqlx::Error> {
let has_wal_sidecar =
sidecar_path(path, "-wal").exists() || sidecar_path(path, "-shm").exists();
let options = SqliteConnectOptions::from_str(&format!("sqlite://{}", path.display()))?
.read_only(true)
.create_if_missing(false)
// A cleanly closed WAL database has all committed pages in the main
// file. Immutable read-only mode prevents SQLite from creating a new
// `-shm` sidecar during future-schema rejection. Live WAL sidecars
// must remain visible to the preflight reader.
.immutable(!has_wal_sidecar)
.busy_timeout(Duration::from_secs(30));
let pool = SqlitePoolOptions::new()
.max_connections(1)
.connect_with(options)
.await?;
let snapshot = {
let mut connection = pool.acquire().await?;
snapshot_connection(&mut connection, component).await
};
pool.close().await;
snapshot
}
/// Connects a writable single-connection pool, retrying the SQLite-specific
/// exclusive lock needed when a connection switches an existing database to
/// WAL mode. SQLite's busy timeout cannot wait for that PRAGMA, so the retry
/// belongs around connection establishment rather than only around writes.
pub async fn connect_writable_pool(
options: SqliteConnectOptions,
) -> Result<SqlitePool, sqlx::Error> {
const MAX_ATTEMPTS: usize = 32;
for attempt in 0..=MAX_ATTEMPTS {
match SqlitePoolOptions::new()
.max_connections(1)
.connect_with(options.clone())
.await
{
Ok(pool) => return Ok(pool),
Err(error) if attempt < MAX_ATTEMPTS && is_sqlite_lock_error(&error) => {
let delay_millis = (25 * (attempt as u64 + 1)).min(250);
tokio::time::sleep(Duration::from_millis(delay_millis)).await;
}
Err(error) => return Err(error),
}
}
unreachable!("SQLite connection retry loop always returns")
}
fn is_sqlite_lock_error(error: &sqlx::Error) -> bool {
error.to_string().contains("database is locked")
}
/// Snapshots the schema using an already-open connection. The caller may use
/// this both for read-only preflight and inside the migration transaction.
pub async fn snapshot_connection(
connection: &mut SqliteConnection,
component: &str,
) -> Result<SqliteSchemaSnapshot, sqlx::Error> {
let object_rows = sqlx::query(
"SELECT type, name FROM sqlite_master
WHERE name NOT LIKE 'sqlite_%'
ORDER BY type, name",
)
.fetch_all(&mut *connection)
.await?;
let mut objects = BTreeSet::new();
let mut table_names = BTreeSet::new();
for row in object_rows {
let object_type: String = row.try_get("type")?;
let name: String = row.try_get("name")?;
if object_type == "table" {
table_names.insert(name.clone());
}
objects.insert((object_type, name));
}
let mut tables = BTreeMap::new();
let mut indexes = BTreeMap::new();
for table in table_names {
let quoted_table = quote_identifier(&table);
let column_rows = sqlx::query(&format!("PRAGMA table_info({quoted_table})"))
.fetch_all(&mut *connection)
.await?;
let mut columns = BTreeMap::new();
for row in column_rows {
let name: String = row.try_get("name")?;
let data_type: String = row.try_get("type")?;
let not_null: i64 = row.try_get("notnull")?;
let default_value: Option<String> = row.try_get("dflt_value")?;
let primary_key: i64 = row.try_get("pk")?;
columns.insert(
name,
SqliteColumn {
data_type,
not_null: not_null != 0,
default_value,
primary_key: primary_key != 0,
},
);
}
tables.insert(table.clone(), columns);
let index_rows = sqlx::query(&format!("PRAGMA index_list({quoted_table})"))
.fetch_all(&mut *connection)
.await?;
let mut table_indexes = BTreeMap::new();
for row in index_rows {
let index_name: String = row.try_get("name")?;
if index_name.starts_with("sqlite_autoindex_") {
continue;
}
let quoted_index = quote_identifier(&index_name);
let index_columns = sqlx::query(&format!("PRAGMA index_info({quoted_index})"))
.fetch_all(&mut *connection)
.await?;
let mut columns = Vec::new();
for index_column in index_columns {
let sequence: i64 = index_column.try_get("seqno")?;
let name: Option<String> = index_column.try_get("name")?;
if sequence < 0 {
continue;
}
let name = name.ok_or_else(|| {
sqlx::Error::Protocol(format!(
"SQLite index {index_name} has an unnamed column"
))
})?;
columns.push((sequence, name));
}
columns.sort_by_key(|(sequence, _)| *sequence);
table_indexes.insert(
index_name,
columns
.into_iter()
.map(|(_, name)| name)
.collect::<Vec<_>>(),
);
}
if !table_indexes.is_empty() {
indexes.insert(table, table_indexes);
}
}
let component_version = if tables.contains_key(SCHEMA_MIGRATIONS_TABLE) {
sqlx::query_scalar("SELECT version FROM schema_migrations WHERE component = ?1")
.bind(component)
.fetch_optional(&mut *connection)
.await?
} else {
None
};
Ok(SqliteSchemaSnapshot {
objects,
tables,
indexes,
component_version,
})
}
/// Starts a real SQLite writer transaction. `BEGIN IMMEDIATE` serializes DDL
/// migration writers instead of allowing two preflight results to race.
pub async fn begin_immediate(
pool: &SqlitePool,
) -> Result<sqlx::Transaction<'static, sqlx::Sqlite>, sqlx::Error> {
pool.begin_with("BEGIN IMMEDIATE").await
}
pub async fn write_component_version(
transaction: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
component: &str,
version: u32,
) -> Result<(), sqlx::Error> {
sqlx::query(
"INSERT INTO schema_migrations(component, version) VALUES (?1, ?2)
ON CONFLICT(component) DO UPDATE SET version = excluded.version",
)
.bind(component)
.bind(i64::from(version))
.execute(&mut **transaction)
.await?;
Ok(())
}
pub async fn create_schema_migrations_table(
transaction: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
) -> Result<(), sqlx::Error> {
sqlx::query(
"CREATE TABLE IF NOT EXISTS schema_migrations (
component TEXT PRIMARY KEY NOT NULL,
version INTEGER NOT NULL CHECK(version >= 1)
)",
)
.execute(&mut **transaction)
.await?;
Ok(())
}
pub fn matches_fingerprint(
snapshot: &SqliteSchemaSnapshot,
expected_tables: &[ExpectedTable<'_>],
expected_indexes: &[ExpectedIndex<'_>],
) -> bool {
let expected_table_names = expected_tables
.iter()
.map(|table| table.name)
.collect::<BTreeSet<_>>();
let actual_table_names: BTreeSet<&str> = snapshot.tables.keys().map(String::as_str).collect();
if actual_table_names != expected_table_names {
return false;
}
let expected_object_names = expected_tables
.iter()
.map(|table| ("table", table.name))
.chain(expected_indexes.iter().map(|index| ("index", index.name)))
.collect::<BTreeSet<_>>();
let actual_object_names = snapshot
.objects
.iter()
.map(|(object_type, name)| (object_type.as_str(), name.as_str()))
.collect::<BTreeSet<_>>();
if actual_object_names != expected_object_names {
return false;
}
for table in expected_tables {
let Some(actual_columns) = snapshot.tables.get(table.name) else {
return false;
};
if actual_columns.len() != table.columns.len() {
return false;
}
for expected in table.columns {
let Some(actual) = actual_columns.get(expected.name) else {
return false;
};
if actual.data_type.to_ascii_uppercase() != expected.data_type
|| actual.not_null != expected.not_null
|| actual.primary_key != expected.primary_key
|| normalize_default(actual.default_value.as_deref())
!= normalize_default(expected.default_value)
{
return false;
}
}
}
let expected_indexes = expected_indexes
.iter()
.map(|index| {
(
index.table.to_string(),
index.name.to_string(),
index
.columns
.iter()
.map(|column| (*column).to_string())
.collect::<Vec<_>>(),
)
})
.collect::<BTreeSet<_>>();
let actual_indexes = snapshot
.indexes
.iter()
.flat_map(|(table, indexes)| {
indexes
.iter()
.map(|(name, columns)| (table.clone(), name.clone(), columns.clone()))
})
.collect::<BTreeSet<_>>();
actual_indexes == expected_indexes
}
pub fn schema_migrations_table() -> ExpectedTable<'static> {
ExpectedTable {
name: SCHEMA_MIGRATIONS_TABLE,
columns: &[
ExpectedColumn {
name: "component",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: true,
},
ExpectedColumn {
name: "version",
data_type: "INTEGER",
not_null: true,
default_value: None,
primary_key: false,
},
],
}
}
fn normalize_default(value: Option<&str>) -> Option<String> {
value.map(|value| value.trim().to_ascii_lowercase())
}
fn quote_identifier(value: &str) -> String {
format!("\"{}\"", value.replace('"', "\"\""))
}
fn sidecar_path(path: &Path, suffix: &str) -> std::path::PathBuf {
std::path::PathBuf::from(format!("{}{}", path.display(), suffix))
}
infrastructure/src/translation_memory.rs
+546 -89
View File
@@ -1,6 +1,9 @@
//! 跨 official release 的 Translation Memory SQLite 仓储。
use crate::path_security::{set_file_mode, STATE_FILE_MODE};
use crate::sqlite_migration::{
self, ExpectedColumn, ExpectedIndex, ExpectedTable, SqliteSchemaSnapshot,
};
use async_trait::async_trait;
use bat_core::domain::{
TranslationMemoryContext, TranslationMemoryDraft, TranslationMemoryEntry,
@@ -10,7 +13,7 @@ use bat_core::domain::{
use bat_core::repositories::TranslationMemoryRepository;
use bat_core::{Error, Result};
use serde::de::DeserializeOwned;
use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode};
use sqlx::{Row, SqlitePool};
use std::fs;
use std::path::{Path, PathBuf};
@@ -85,14 +88,24 @@ impl SqliteTranslationMemoryRepository {
return Err(Error::NotFound(absolute.display().to_string()));
}
if let Ok(metadata) = fs::symlink_metadata(&absolute) {
if metadata.len() > 0 {
let snapshot = sqlite_migration::read_only_preflight(
&absolute,
TRANSLATION_MEMORY_SCHEMA_COMPONENT,
)
.await
.map_err(db_error)?;
classify_translation_memory_schema(&snapshot)?;
}
}
let options = SqliteConnectOptions::from_str(&format!("sqlite://{}", absolute.display()))
.map_err(|error| Error::Other(error.into()))?
.create_if_missing(create_if_missing)
.journal_mode(SqliteJournalMode::Wal)
.busy_timeout(Duration::from_secs(30));
let pool = SqlitePoolOptions::new()
.max_connections(1)
.connect_with(options)
let pool = sqlite_migration::connect_writable_pool(options)
.await
.map_err(db_error)?;
set_file_mode(&absolute, STATE_FILE_MODE, "Translation Memory 数据库")
@@ -103,98 +116,93 @@ impl SqliteTranslationMemoryRepository {
}
async fn init_schema(&self) -> Result<()> {
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS schema_migrations (
component TEXT PRIMARY KEY NOT NULL,
version INTEGER NOT NULL CHECK(version >= 1)
)
"#,
)
.execute(&self.pool)
.await
.map_err(db_error)?;
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS translation_memory (
record_id TEXT PRIMARY KEY NOT NULL,
source_text TEXT NOT NULL,
source_hash TEXT NOT NULL,
normalized_source_text TEXT NOT NULL,
source_context_json TEXT NOT NULL,
source_context_hash TEXT NOT NULL,
translated_text TEXT NOT NULL,
translation_source_kind TEXT NOT NULL,
trust_status TEXT NOT NULL,
official_release_id TEXT NOT NULL,
source_trace_json TEXT NOT NULL,
provider TEXT,
provider_run_id TEXT,
created_unix_seconds INTEGER NOT NULL,
updated_unix_seconds INTEGER NOT NULL,
trusted_unix_seconds INTEGER,
trusted_by TEXT,
trusted_reason TEXT,
supersedes_record_id TEXT,
superseded_by_record_id TEXT,
CHECK (length(source_text) > 0),
CHECK (length(source_hash) > 0),
CHECK (length(source_context_hash) > 0),
CHECK (length(official_release_id) > 0),
CHECK (translation_source_kind IN ('provider', 'manual', 'imported')),
CHECK (trust_status IN ('candidate', 'trusted', 'superseded', 'rejected'))
)
"#,
)
.execute(&self.pool)
.await
.map_err(db_error)?;
sqlx::query(
"CREATE INDEX IF NOT EXISTS idx_translation_memory_source_hash \
ON translation_memory(source_hash)",
)
.execute(&self.pool)
.await
.map_err(db_error)?;
sqlx::query(
"CREATE INDEX IF NOT EXISTS idx_translation_memory_normalized_source \
ON translation_memory(normalized_source_text)",
)
.execute(&self.pool)
.await
.map_err(db_error)?;
sqlx::query(
"CREATE INDEX IF NOT EXISTS idx_translation_memory_context \
ON translation_memory(source_hash, source_context_hash)",
)
.execute(&self.pool)
.await
.map_err(db_error)?;
self.init_schema_with_failure(None).await
}
let current: Option<i64> =
sqlx::query_scalar("SELECT version FROM schema_migrations WHERE component = ?1")
.bind(TRANSLATION_MEMORY_SCHEMA_COMPONENT)
.fetch_optional(&self.pool)
#[cfg(test)]
async fn init_schema_with_test_failure(&self, fail_after_step: usize) -> Result<()> {
self.init_schema_with_failure(Some(fail_after_step)).await
}
async fn init_schema_with_failure(&self, fail_after_step: Option<usize>) -> Result<()> {
let mut transaction = sqlite_migration::begin_immediate(&self.pool)
.await
.map_err(db_error)?;
let result = self
.migrate_in_transaction(&mut transaction, fail_after_step)
.await;
match result {
Ok(()) => transaction.commit().await.map_err(db_error),
Err(error) => {
let _ = transaction.rollback().await;
Err(error)
}
}
}
async fn migrate_in_transaction(
&self,
transaction: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
fail_after_step: Option<usize>,
) -> Result<()> {
let snapshot = sqlite_migration::snapshot_connection(
transaction.as_mut(),
TRANSLATION_MEMORY_SCHEMA_COMPONENT,
)
.await
.map_err(db_error)?;
match classify_translation_memory_schema(&snapshot)? {
TranslationMemorySchemaState::Empty => {
create_translation_memory_schema(transaction, fail_after_step).await?;
sqlite_migration::write_component_version(
transaction,
TRANSLATION_MEMORY_SCHEMA_COMPONENT,
TRANSLATION_MEMORY_SCHEMA_VERSION,
)
.await
.map_err(db_error)?;
if current.is_some_and(|version| version > i64::from(TRANSLATION_MEMORY_SCHEMA_VERSION)) {
return Err(Error::InvalidArgument(format!(
"不支持的 Translation Memory schema 版本：{}",
current.unwrap_or_default()
)));
}
TranslationMemorySchemaState::Version(version)
if version == TRANSLATION_MEMORY_SCHEMA_VERSION =>
{
if snapshot.component_version.is_none() {
if !snapshot
.tables
.contains_key(sqlite_migration::SCHEMA_MIGRATIONS_TABLE)
{
sqlite_migration::create_schema_migrations_table(transaction)
.await
.map_err(db_error)?;
}
sqlite_migration::write_component_version(
transaction,
TRANSLATION_MEMORY_SCHEMA_COMPONENT,
TRANSLATION_MEMORY_SCHEMA_VERSION,
)
.await
.map_err(db_error)?;
}
}
TranslationMemorySchemaState::Version(version) => {
return Err(invalid_translation_memory_schema(format!(
"内部不支持的迁移起点 {version}"
)));
}
}
sqlx::query(
r#"
INSERT INTO schema_migrations(component, version)
VALUES (?1, ?2)
ON CONFLICT(component) DO UPDATE SET version = excluded.version
"#,
let final_snapshot = sqlite_migration::snapshot_connection(
transaction.as_mut(),
TRANSLATION_MEMORY_SCHEMA_COMPONENT,
)
.bind(TRANSLATION_MEMORY_SCHEMA_COMPONENT)
.bind(i64::from(TRANSLATION_MEMORY_SCHEMA_VERSION))
.execute(&self.pool)
.await
.map_err(db_error)?;
if final_snapshot.component_version != Some(i64::from(TRANSLATION_MEMORY_SCHEMA_VERSION))
|| !matches_translation_memory_fingerprint(&final_snapshot)
{
return Err(invalid_translation_memory_schema(
"migration 结果与当前 schema fingerprint 不一致".to_string(),
));
}
Ok(())
}
@@ -219,6 +227,297 @@ impl SqliteTranslationMemoryRepository {
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum TranslationMemorySchemaState {
Empty,
Version(u32),
}
fn classify_translation_memory_schema(
snapshot: &SqliteSchemaSnapshot,
) -> Result<TranslationMemorySchemaState> {
if snapshot.is_empty() {
return Ok(TranslationMemorySchemaState::Empty);
}
if let Some(observed) = snapshot.component_version {
if observed > i64::from(TRANSLATION_MEMORY_SCHEMA_VERSION) {
return Err(invalid_translation_memory_schema(format!(
"不支持的 Translation Memory schema 版本：observed={observed}, supported={TRANSLATION_MEMORY_SCHEMA_VERSION}"
)));
}
if observed < 1 {
return Err(invalid_translation_memory_schema(format!(
"schema version 无效：observed={observed}, supported=1..={TRANSLATION_MEMORY_SCHEMA_VERSION}"
)));
}
}
if (matches_translation_memory_fingerprint(snapshot)
|| (snapshot.component_version.is_none()
&& matches_translation_memory_component_fingerprint(snapshot)))
&& snapshot
.component_version
.is_none_or(|observed| observed == i64::from(TRANSLATION_MEMORY_SCHEMA_VERSION))
{
return Ok(TranslationMemorySchemaState::Version(
TRANSLATION_MEMORY_SCHEMA_VERSION,
));
}
match snapshot.component_version {
Some(observed) => Err(invalid_translation_memory_schema(format!(
"schema version 与实际结构不一致：observed={observed}, supported={TRANSLATION_MEMORY_SCHEMA_VERSION}"
))),
None => Err(invalid_translation_memory_schema(
"未识别的 legacy schema，拒绝静默修复".to_string(),
)),
}
}
fn invalid_translation_memory_schema(detail: String) -> Error {
Error::InvalidArgument(format!("Translation Memory schema 无效：{detail}"))
}
fn matches_translation_memory_fingerprint(snapshot: &SqliteSchemaSnapshot) -> bool {
matches_translation_memory_fingerprint_with_migrations(snapshot, true)
}
fn matches_translation_memory_component_fingerprint(snapshot: &SqliteSchemaSnapshot) -> bool {
matches_translation_memory_fingerprint_with_migrations(snapshot, false)
}
fn matches_translation_memory_fingerprint_with_migrations(
snapshot: &SqliteSchemaSnapshot,
include_migrations: bool,
) -> bool {
let tables = [ExpectedTable {
name: "translation_memory",
columns: &TRANSLATION_MEMORY_COLUMNS,
}];
let indexes = [
ExpectedIndex {
table: "translation_memory",
name: "idx_translation_memory_source_hash",
columns: &["source_hash"],
},
ExpectedIndex {
table: "translation_memory",
name: "idx_translation_memory_normalized_source",
columns: &["normalized_source_text"],
},
ExpectedIndex {
table: "translation_memory",
name: "idx_translation_memory_context",
columns: &["source_hash", "source_context_hash"],
},
];
if include_migrations {
let tables = [sqlite_migration::schema_migrations_table(), tables[0]];
return sqlite_migration::matches_fingerprint(snapshot, &tables, &indexes);
}
sqlite_migration::matches_fingerprint(snapshot, &tables, &indexes)
}
const TRANSLATION_MEMORY_COLUMNS: [ExpectedColumn<'static>; 20] = [
ExpectedColumn {
name: "record_id",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: true,
},
ExpectedColumn {
name: "source_text",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "source_hash",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "normalized_source_text",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "source_context_json",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "source_context_hash",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "translated_text",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "translation_source_kind",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "trust_status",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "official_release_id",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "source_trace_json",
data_type: "TEXT",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "provider",
data_type: "TEXT",
not_null: false,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "provider_run_id",
data_type: "TEXT",
not_null: false,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "created_unix_seconds",
data_type: "INTEGER",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "updated_unix_seconds",
data_type: "INTEGER",
not_null: true,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "trusted_unix_seconds",
data_type: "INTEGER",
not_null: false,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "trusted_by",
data_type: "TEXT",
not_null: false,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "trusted_reason",
data_type: "TEXT",
not_null: false,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "supersedes_record_id",
data_type: "TEXT",
not_null: false,
default_value: None,
primary_key: false,
},
ExpectedColumn {
name: "superseded_by_record_id",
data_type: "TEXT",
not_null: false,
default_value: None,
primary_key: false,
},
];
async fn create_translation_memory_schema(
transaction: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
fail_after_step: Option<usize>,
) -> Result<()> {
let steps = [
"CREATE TABLE schema_migrations (
component TEXT PRIMARY KEY NOT NULL,
version INTEGER NOT NULL CHECK(version >= 1)
)",
"CREATE TABLE translation_memory (
record_id TEXT PRIMARY KEY NOT NULL,
source_text TEXT NOT NULL,
source_hash TEXT NOT NULL,
normalized_source_text TEXT NOT NULL,
source_context_json TEXT NOT NULL,
source_context_hash TEXT NOT NULL,
translated_text TEXT NOT NULL,
translation_source_kind TEXT NOT NULL,
trust_status TEXT NOT NULL,
official_release_id TEXT NOT NULL,
source_trace_json TEXT NOT NULL,
provider TEXT,
provider_run_id TEXT,
created_unix_seconds INTEGER NOT NULL,
updated_unix_seconds INTEGER NOT NULL,
trusted_unix_seconds INTEGER,
trusted_by TEXT,
trusted_reason TEXT,
supersedes_record_id TEXT,
superseded_by_record_id TEXT,
CHECK (length(source_text) > 0),
CHECK (length(source_hash) > 0),
CHECK (length(source_context_hash) > 0),
CHECK (length(official_release_id) > 0),
CHECK (translation_source_kind IN ('provider', 'manual', 'imported')),
CHECK (trust_status IN ('candidate', 'trusted', 'superseded', 'rejected'))
)",
"CREATE INDEX idx_translation_memory_source_hash
ON translation_memory(source_hash)",
"CREATE INDEX idx_translation_memory_normalized_source
ON translation_memory(normalized_source_text)",
"CREATE INDEX idx_translation_memory_context
ON translation_memory(source_hash, source_context_hash)",
];
for (index, statement) in steps.iter().enumerate() {
sqlx::query(statement)
.execute(&mut **transaction)
.await
.map_err(db_error)?;
if fail_after_step == Some(index + 1) {
return Err(Error::Other(anyhow::anyhow!(
"Translation Memory migration failed after step {}",
index + 1
)));
}
}
Ok(())
}
#[async_trait]
impl TranslationMemoryRepository for SqliteTranslationMemoryRepository {
async fn upsert_candidate(
@@ -721,6 +1020,7 @@ fn ensure_safe_tm_parent(parent: &Path) -> Result<()> {
mod tests {
use super::*;
use bat_core::domain::TranslationMemorySourceTrace;
use sqlx::sqlite::SqlitePoolOptions;
fn draft(release: &str, source: &str, translated: &str) -> TranslationMemoryDraft {
let source_trace = TranslationMemorySourceTrace {
@@ -762,6 +1062,163 @@ mod tests {
}
}
async fn raw_repository(
path: &std::path::Path,
create_if_missing: bool,
) -> SqliteTranslationMemoryRepository {
let options = SqliteConnectOptions::from_str(&format!("sqlite://{}", path.display()))
.unwrap()
.create_if_missing(create_if_missing);
let pool = SqlitePoolOptions::new()
.max_connections(1)
.connect_with(options)
.await
.unwrap();
SqliteTranslationMemoryRepository { pool }
}
#[tokio::test]
async fn sqlite_translation_memory_preserves_data_across_reopen_and_missing_row() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join(TRANSLATION_MEMORY_REPOSITORY_FILE);
let repository = SqliteTranslationMemoryRepository::new(&path).await.unwrap();
let entry = repository
.upsert_candidate(draft("release-1", "Hello", "你好"))
.await
.unwrap();
let trusted = repository
.confirm(&entry.record_id, "reviewer", Some("accepted".to_string()))
.await
.unwrap();
sqlx::query("DROP TABLE schema_migrations")
.execute(&repository.pool)
.await
.unwrap();
repository.pool.close().await;
let reopened = SqliteTranslationMemoryRepository::open(&path)
.await
.unwrap();
let restored = reopened.find(&trusted.record_id).await.unwrap();
assert_eq!(restored.translated_text, "你好");
assert_eq!(restored.trust_status, TranslationMemoryTrustStatus::Trusted);
assert_eq!(restored.source_trace.official_release_id, "release-1");
assert_eq!(
reopened.summary().await.unwrap().schema_version,
TRANSLATION_MEMORY_SCHEMA_VERSION
);
}
#[tokio::test]
async fn sqlite_translation_memory_future_schema_is_read_only_failure() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join(TRANSLATION_MEMORY_REPOSITORY_FILE);
let repository = SqliteTranslationMemoryRepository::new(&path).await.unwrap();
sqlx::query("UPDATE schema_migrations SET version = ?2 WHERE component = ?1")
.bind(TRANSLATION_MEMORY_SCHEMA_COMPONENT)
.bind(i64::from(TRANSLATION_MEMORY_SCHEMA_VERSION) + 1)
.execute(&repository.pool)
.await
.unwrap();
repository.pool.close().await;
let before = std::fs::read(&path).unwrap();
let wal_path = std::path::PathBuf::from(format!("{}-wal", path.display()));
let shm_path = std::path::PathBuf::from(format!("{}-shm", path.display()));
let wal_before = std::fs::read(&wal_path).ok();
let shm_before = std::fs::read(&shm_path).ok();
let error = SqliteTranslationMemoryRepository::new(&path)
.await
.unwrap_err();
assert!(error
.to_string()
.contains("不支持的 Translation Memory schema"));
assert_eq!(std::fs::read(&path).unwrap(), before);
assert_eq!(std::fs::read(&wal_path).ok(), wal_before);
assert_eq!(std::fs::read(&shm_path).ok(), shm_before);
}
#[tokio::test]
async fn sqlite_translation_memory_unknown_schema_fails_closed() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join(TRANSLATION_MEMORY_REPOSITORY_FILE);
let repository = raw_repository(&path, true).await;
sqlx::query(
"CREATE TABLE schema_migrations (
component TEXT PRIMARY KEY NOT NULL,
version INTEGER NOT NULL CHECK(version >= 1)
)",
)
.execute(&repository.pool)
.await
.unwrap();
sqlx::query("INSERT INTO schema_migrations(component, version) VALUES (?1, 1)")
.bind(TRANSLATION_MEMORY_SCHEMA_COMPONENT)
.execute(&repository.pool)
.await
.unwrap();
sqlx::query("CREATE TABLE translation_memory (record_id TEXT PRIMARY KEY)")
.execute(&repository.pool)
.await
.unwrap();
repository.pool.close().await;
let before = std::fs::read(&path).unwrap();
let error = SqliteTranslationMemoryRepository::open(&path)
.await
.unwrap_err();
assert!(error
.to_string()
.contains("schema version 与实际结构不一致"));
assert_eq!(std::fs::read(&path).unwrap(), before);
}
#[tokio::test]
async fn sqlite_translation_memory_failed_new_schema_rolls_back_and_retries() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join(TRANSLATION_MEMORY_REPOSITORY_FILE);
let repository = raw_repository(&path, true).await;
assert!(repository.init_schema_with_test_failure(2).await.is_err());
let table_count: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM sqlite_master WHERE type = 'table'")
.fetch_one(&repository.pool)
.await
.unwrap();
assert_eq!(table_count, 0);
repository.init_schema().await.unwrap();
let version: i64 =
sqlx::query_scalar("SELECT version FROM schema_migrations WHERE component = ?1")
.bind(TRANSLATION_MEMORY_SCHEMA_COMPONENT)
.fetch_one(&repository.pool)
.await
.unwrap();
assert_eq!(version, 1);
}
#[tokio::test]
async fn sqlite_translation_memory_concurrent_new_open_has_one_current_schema() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join(TRANSLATION_MEMORY_REPOSITORY_FILE);
let (left, right) = tokio::join!(
SqliteTranslationMemoryRepository::new(&path),
SqliteTranslationMemoryRepository::new(&path)
);
assert!(left.is_ok(), "left open failed: {left:?}");
assert!(right.is_ok(), "right open failed: {right:?}");
let repository = left.unwrap();
drop(right);
assert_eq!(
repository.summary().await.unwrap().schema_version,
TRANSLATION_MEMORY_SCHEMA_VERSION
);
repository.pool.close().await;
let reopened = SqliteTranslationMemoryRepository::open(&path)
.await
.unwrap();
assert_eq!(
reopened.summary().await.unwrap().schema_version,
TRANSLATION_MEMORY_SCHEMA_VERSION
);
}
#[tokio::test]
async fn initializes_schema_and_reuses_trusted_entry_across_releases() {
let temp = tempfile::TempDir::new().unwrap();
infrastructure/src/translation_tasks.rs
+827 -171
View File
File diff suppressed because it is too large Load Diff