eeff292
use std::hash::{DefaultHasher, Hash, Hasher};
use sea_orm::{
ConnectionTrait, DatabaseBackend, DatabaseConnection, DbErr, Statement, TransactionTrait,
};
const SCHEMA_PRE_SYNC: &str = include_str!("../../schema_pre_sync.sql");
const SCHEMA_EXTRAS: &str = include_str!("../../schema_extras.sql");
/// Bump whenever `sync_schema` gains a Rust-side step that is not captured by
/// the entity registry or by `schema_pre_sync.sql` / `schema_extras.sql`.
const SYNC_FORMAT_VERSION: u32 = 3;
pub const MIGRATION_LOCK_KEY: i64 = 5_241_915_330_551;
const SYNC_STATE_DDL: &str = "CREATE TABLE IF NOT EXISTS schema_sync_state (
id smallint PRIMARY KEY CHECK (id = 1),
fingerprint text NOT NULL,
live_fingerprint text,
synced_at timestamptz NOT NULL DEFAULT now()
)";
const SYNC_STATE_LIVE_COLUMN: &str =
"ALTER TABLE schema_sync_state ADD COLUMN IF NOT EXISTS live_fingerprint text";
const LIVE_FINGERPRINT_SQL: &str = r#"
SELECT COALESCE(md5(string_agg(entry, E'\n' ORDER BY entry)), '') AS fingerprint
FROM (
SELECT format('table:%s', table_name) AS entry
FROM information_schema.tables
WHERE table_schema = 'public' AND table_type = 'BASE TABLE'
UNION ALL
SELECT format('column:%s.%s:%s:%s:%s',
table_name, column_name, data_type, is_nullable,
COALESCE(column_default, ''))
FROM information_schema.columns
WHERE table_schema = 'public'
AND table_name IN (
SELECT table_name
FROM information_schema.tables
WHERE table_schema = 'public' AND table_type = 'BASE TABLE'
)
UNION ALL
SELECT format('index:%s:%s', tablename, indexdef)
FROM pg_indexes
WHERE schemaname = 'public'
UNION ALL
SELECT format('constraint:%s:%s', conrelid::regclass, pg_get_constraintdef(oid))
FROM pg_constraint
WHERE connamespace = (SELECT oid FROM pg_namespace WHERE nspname = 'public')
UNION ALL
SELECT format('enum:%s:%s', t.typname, e.enumlabel)
FROM pg_type t
JOIN pg_enum e ON e.enumtypid = t.oid
JOIN pg_namespace n ON n.oid = t.typnamespace
WHERE n.nspname = 'public'
) AS entries
"#;
pub async fn sync_schema(db: &DatabaseConnection) -> Result<(), DbErr> {
let builder = db.get_schema_registry("models::");
if db.get_database_backend() != DatabaseBackend::Postgres {
return builder.sync(db).await;
}
let mut hasher = DefaultHasher::new();
SYNC_FORMAT_VERSION.hash(&mut hasher);
SCHEMA_PRE_SYNC.hash(&mut hasher);
SCHEMA_EXTRAS.hash(&mut hasher);
format!("{builder:?}").hash(&mut hasher);
let fingerprint = format!("{:016x}", hasher.finish());
let txn = db.begin().await?;
txn.execute_unprepared(&format!(
"SELECT pg_advisory_xact_lock({MIGRATION_LOCK_KEY})"
))
.await?;
txn.execute_unprepared(SYNC_STATE_DDL).await?;
txn.execute_unprepared(SYNC_STATE_LIVE_COLUMN).await?;
let live = live_fingerprint(&txn).await?;
let stored = txn
.query_one_raw(Statement::from_string(
DatabaseBackend::Postgres,
"SELECT fingerprint, live_fingerprint FROM schema_sync_state WHERE id = 1",
))
.await?;
if let Some(row) = stored {
let stored_fingerprint = row.try_get::<String>("", "fingerprint")?;
let stored_live = row.try_get::<Option<String>>("", "live_fingerprint")?;
if stored_fingerprint == fingerprint && stored_live.as_deref() == Some(live.as_str()) {
txn.commit().await?;
return Ok(());
}
}
txn.execute_unprepared(SCHEMA_PRE_SYNC).await?;
builder.sync(&txn).await?;
txn.execute_unprepared(SCHEMA_EXTRAS).await?;
let live = live_fingerprint(&txn).await?;
let update = format!(
"INSERT INTO schema_sync_state (id, fingerprint, live_fingerprint) \
VALUES (1, '{fingerprint}', '{live}') \
ON CONFLICT (id) DO UPDATE SET fingerprint = EXCLUDED.fingerprint, \
live_fingerprint = EXCLUDED.live_fingerprint, synced_at = now()"
);
txn.execute_unprepared(&update).await?;
txn.commit().await
}
async fn live_fingerprint<C>(db: &C) -> Result<String, DbErr>
where
C: ConnectionTrait,
{
let row = db
.query_one_raw(Statement::from_string(
DatabaseBackend::Postgres,
LIVE_FINGERPRINT_SQL,
))
.await?;
match row {
Some(row) => row.try_get::<String>("", "fingerprint"),
None => Ok(String::new()),
}
}