AWG: Added migration

src/admin/views.rs

@@ -84,6 +84,11 @@ fn config_display_entries(config: &AppConfig, sources: &ConfigSources) -> Vec<Co
             config.database_url.clone(),
             defaults.database_url.clone()
         ),
+        entry!(
+            migrate_sqlite,
+            config.migrate_sqlite.clone(),
+            defaults.migrate_sqlite.clone()
+        ),
         entry!(
             oidc_issuer,
             config.oidc_issuer.clone(),

src/config.rs

@@ -47,6 +47,15 @@ impl ConfigEntry {
     pub fn new(key: String, value: String) -> Self {
         Self { key, value }
     }
+
+    pub fn key_str(&self) -> &str {
+        &self.key
+    }
+
+    pub async fn get_by_key(db: &Database, key: &str) -> cot::db::Result<Option<Self>> {
+        let key = key.to_owned();
+        cot::db::query!(ConfigEntry, $key == key).get(db).await
+    }
 }
 
 // ---------------------------------------------------------------------------
@@ -87,6 +96,7 @@ pub mod db_migrations {
 
 pub struct ConfigSources {
     pub database_url: ConfigSource,
+    pub migrate_sqlite: ConfigSource,
     pub oidc_issuer: ConfigSource,
     pub oidc_client_id: ConfigSource,
     pub oidc_client_secret: ConfigSource,
@@ -116,6 +126,7 @@ impl Default for ConfigSources {
     fn default() -> Self {
         Self {
             database_url: ConfigSource::Default,
+            migrate_sqlite: ConfigSource::Default,
             oidc_issuer: ConfigSource::Default,
             oidc_client_id: ConfigSource::Default,
             oidc_client_secret: ConfigSource::Default,
@@ -194,8 +205,10 @@ macro_rules! impl_env_overrides {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct AppConfig {
-    /// SQLite connection URL.
+    /// SQLite or PostgreSQL connection URL.
     pub database_url: String,
+    /// Optional SQLite database path/URL copied into PostgreSQL on startup.
+    pub migrate_sqlite: String,
     /// OIDC issuer URL.
     pub oidc_issuer: String,
     /// OIDC client ID.
@@ -248,6 +261,7 @@ impl Default for AppConfig {
     fn default() -> Self {
         Self {
             database_url: "sqlite://amnezia-fellow.sqlite3?mode=rwc".into(),
+            migrate_sqlite: String::new(),
             oidc_issuer: String::new(),
             oidc_client_id: String::new(),
             oidc_client_secret: String::new(),
@@ -277,6 +291,7 @@ impl Default for AppConfig {
 
 impl_env_overrides!(
     database_url,
+    migrate_sqlite,
     oidc_issuer,
     oidc_client_id,
     oidc_client_secret,
@@ -355,6 +370,7 @@ impl AppConfig {
         }
 
         apply_db_field!(database_url);
+        apply_db_field!(migrate_sqlite);
         apply_db_field!(oidc_issuer);
         apply_db_field!(oidc_client_id);
         apply_db_field!(oidc_client_secret);

src/main.rs

@@ -4,12 +4,14 @@ mod auth;
 mod config;
 mod i18n;
 mod oidc;
+mod sqlite_migration;
 mod telegram;
 mod user;
 mod vpn;
 
 use std::sync::Arc;
 
+use async_trait::async_trait;
 use cot::auth::PasswordVerificationResult;
 use cot::cli::CliMetadata;
 use cot::common_types::Password;
@@ -21,7 +23,7 @@ use cot::db::Database;
 use cot::form::{Form, FormResult};
 use cot::html::Html;
 use cot::middleware::SessionMiddleware;
-use cot::project::RegisterAppsContext;
+use cot::project::{ProjectContext, RegisterAppsContext};
 use cot::request::extractors::{RequestForm, UrlQuery};
 use cot::response::IntoResponse;
 use cot::router::method::get;
@@ -181,11 +183,19 @@ struct AmneziaFellowApp {
     config: Arc<AppConfig>,
 }
 
+#[async_trait]
 impl App for AmneziaFellowApp {
     fn name(&self) -> &'static str {
         env!("CARGO_PKG_NAME")
     }
 
+    async fn init(&self, context: &mut ProjectContext) -> cot::Result<()> {
+        if let Some(db) = context.try_database() {
+            sqlite_migration::run_if_requested(&self.config, db).await?;
+        }
+        Ok(())
+    }
+
     fn router(&self) -> Router {
         Router::with_urls([
             Route::with_handler_and_name(
@@ -296,8 +306,11 @@ impl Project for AmneziaFellowProject {
                 "  Priority: env var > DB override > compiled default.\n",
                 "\n",
                 "  Database (required for most features):\n",
-                "    AMNEZIA_FELLOW_DATABASE_URL    SQLite connection URL\n",
+                "    AMNEZIA_FELLOW_DATABASE_URL    SQLite or PostgreSQL connection URL\n",
                 "                                   Example: sqlite:///data/amnezia-fellow.sqlite3?mode=rwc\n",
+                "                                   Example: postgresql://user:pass@postgres:5432/amnezia_fellow\n",
+                "    AMNEZIA_FELLOW_MIGRATE_SQLITE SQLite path/URL to import into PostgreSQL once\n",
+                "                                   Example: /data/amnezia-fellow.sqlite3\n",
                 "\n",
                 "  Server:\n",
                 "    AMNEZIA_FELLOW_LOG_LEVEL       Tracing filter (default: info)\n",

src/sqlite_migration.rs

@@ -0,0 +1,186 @@
+use std::path::Path;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use cot::db::migrations::{self, MigrationEngine, SyncDynMigration};
+use cot::db::{Database, Model};
+use cot::session::db::Session as CotSession;
+
+use crate::config::{AppConfig, ConfigEntry};
+use crate::user::{OidcLink, User};
+use crate::vpn::VpnClient;
+
+const MARKER_KEY: &str = "sqlite_migration_completed_at";
+const SOURCE_KEY: &str = "sqlite_migration_source";
+
+pub async fn run_if_requested(config: &AppConfig, target_db: &Database) -> cot::Result<()> {
+    let source = config.migrate_sqlite.trim();
+    if source.is_empty() {
+        return Ok(());
+    }
+
+    if !is_postgres_url(&config.database_url) {
+        tracing::warn!(
+            "AMNEZIA_FELLOW_MIGRATE_SQLITE is set but the active database is not PostgreSQL; skipping SQLite import"
+        );
+        return Ok(());
+    }
+
+    if ConfigEntry::get_by_key(target_db, MARKER_KEY)
+        .await
+        .map_err(internal)?
+        .is_some()
+    {
+        tracing::info!("SQLite import marker exists; skipping SQLite import");
+        return Ok(());
+    }
+
+    if target_has_app_data(target_db).await? {
+        return Err(cot::Error::internal(
+            "PostgreSQL target already contains amnezia-fellow data and has no SQLite import marker; refusing to merge automatically",
+        ));
+    }
+
+    let source_url = sqlite_source_url(source)?;
+    tracing::info!("Importing amnezia-fellow data from SQLite into PostgreSQL");
+
+    let source_db = Database::new(source_url)
+        .await
+        .map_err(|e| cot::Error::internal(format!("failed to open SQLite source: {e}")))?;
+
+    run_app_migrations(&source_db).await?;
+    copy_app_data(&source_db, target_db).await?;
+    reset_postgres_sequences(target_db).await?;
+    write_marker(target_db, source).await?;
+
+    tracing::info!("SQLite import into PostgreSQL completed");
+    Ok(())
+}
+
+fn is_postgres_url(url: &str) -> bool {
+    url.starts_with("postgresql:")
+}
+
+fn sqlite_source_url(source: &str) -> cot::Result<String> {
+    if source.starts_with("sqlite:") {
+        return Ok(source.to_owned());
+    }
+
+    if !Path::new(source).exists() {
+        return Err(cot::Error::internal(format!(
+            "SQLite import source does not exist: {source}"
+        )));
+    }
+
+    let prefix = if Path::new(source).is_absolute() {
+        "sqlite://"
+    } else {
+        "sqlite:"
+    };
+    Ok(format!("{prefix}{source}?mode=rw"))
+}
+
+async fn run_app_migrations(db: &Database) -> cot::Result<()> {
+    let engine = MigrationEngine::new(app_migrations())
+        .map_err(|e| cot::Error::internal(format!("failed to build migration engine: {e}")))?;
+    engine
+        .run(db)
+        .await
+        .map_err(|e| cot::Error::internal(format!("failed to migrate SQLite source: {e}")))
+}
+
+fn app_migrations() -> Vec<Box<SyncDynMigration>> {
+    let mut all = migrations::wrap_migrations(cot::session::db::migrations::MIGRATIONS);
+    all.extend(migrations::wrap_migrations(
+        crate::config::db_migrations::MIGRATIONS,
+    ));
+    all.extend(migrations::wrap_migrations(
+        crate::user::db_migrations::MIGRATIONS,
+    ));
+    all.extend(migrations::wrap_migrations(
+        crate::vpn::db_migrations::MIGRATIONS,
+    ));
+    all
+}
+
+async fn target_has_app_data(db: &Database) -> cot::Result<bool> {
+    let config_entries = ConfigEntry::objects().count(db).await.map_err(internal)?;
+    let users = User::objects().count(db).await.map_err(internal)?;
+    let oidc_links = OidcLink::objects().count(db).await.map_err(internal)?;
+    let clients = VpnClient::objects().count(db).await.map_err(internal)?;
+    Ok(config_entries > 0 || users > 0 || oidc_links > 0 || clients > 0)
+}
+
+async fn copy_app_data(source_db: &Database, target_db: &Database) -> cot::Result<()> {
+    for mut entry in ConfigEntry::objects().all(source_db).await.map_err(internal)? {
+        if matches!(
+            entry.key_str(),
+            "database_url" | "migrate_sqlite" | MARKER_KEY | SOURCE_KEY
+        ) {
+            continue;
+        }
+        entry.save(target_db).await.map_err(internal)?;
+    }
+
+    for mut session in CotSession::objects()
+        .all(source_db)
+        .await
+        .map_err(internal)?
+    {
+        session.save(target_db).await.map_err(internal)?;
+    }
+
+    for mut user in User::list_all(source_db).await.map_err(internal)? {
+        user.save(target_db).await.map_err(internal)?;
+    }
+
+    for mut link in OidcLink::objects().all(source_db).await.map_err(internal)? {
+        link.save(target_db).await.map_err(internal)?;
+    }
+
+    for mut client in VpnClient::list_all(source_db).await.map_err(internal)? {
+        client.save(target_db).await.map_err(internal)?;
+    }
+
+    Ok(())
+}
+
+async fn reset_postgres_sequences(db: &Database) -> cot::Result<()> {
+    for table in [
+        "cot_session__session",
+        "amnezia_fellow__user",
+        "amnezia_fellow__oidc_link",
+        "amnezia_fellow__vpn_client",
+    ] {
+        let sql = format!(
+            "SELECT setval(pg_get_serial_sequence('{table}', 'id'), \
+             COALESCE((SELECT MAX(id) FROM {table}), 1), \
+             (SELECT COUNT(*) > 0 FROM {table}))"
+        );
+        db.raw(&sql).await.map_err(internal)?;
+    }
+    Ok(())
+}
+
+async fn write_marker(db: &Database, source: &str) -> cot::Result<()> {
+    let mut marker = ConfigEntry::new(MARKER_KEY.to_owned(), now_unix_seconds().to_string());
+    marker.save(db).await.map_err(internal)?;
+
+    let mut source_entry = ConfigEntry::new(SOURCE_KEY.to_owned(), source.to_owned());
+    source_entry.save(db).await.map_err(internal)?;
+
+    let mut migrate_entry = ConfigEntry::new("migrate_sqlite".to_owned(), String::new());
+    migrate_entry.save(db).await.map_err(internal)?;
+    Ok(())
+}
+
+fn now_unix_seconds() -> i64 {
+    let seconds = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|duration| duration.as_secs())
+        .unwrap_or_default();
+    i64::try_from(seconds).unwrap_or(i64::MAX)
+}
+
+fn internal(error: impl std::fmt::Display) -> cot::Error {
+    cot::Error::internal(error.to_string())
+}