kych

dbinit.rs (10338B)

---

 //! KyCH database initialisation and migration.
 //!
 //! Creates the `oauth2gw` schema in a fresh database and brings an existing one
 //! up to the patch level of the installed SQL files.  Both are the same
 //! operation: every schema change is a numbered patch file
 //! (`oauth2gw-000N.sql`) that registers itself with the depesz `Versioning`
 //! bookkeeping in the `_v` schema, and this tool applies those that the
 //! database does not have yet, in order.
 //!
 //! It is idempotent and says nothing but which patches it applied, so that it
 //! can run unattended from the Debian package's postinst on every upgrade.
 //!
 //! Usage:
 //!   kych-dbinit [-c kych.conf] [--sql-dir DIR] [--dry-run]
 //!   kych-dbinit [-c kych.conf] --print-connection
 
 use std::collections::HashSet;
 use std::path::{Path, PathBuf};
 
 use anyhow::{Context, Result, anyhow, bail};
 use clap::Parser;
 use kych_oauth2_gateway_lib::config::{CONFIG_SOURCE, Config};
 use sqlx::{Connection, Executor, PgConnection, Row};
 use taler_config::{CommonArgs, taler_main};
 
 /// Bookkeeping, loaded before the first patch.  Idempotent, so it is applied
 /// unconditionally rather than guessed at.
 const VERSIONING_SQL: &str = "versioning.sql";
 
 /// Patch files are `<PATCH_PREFIX><number>.sql`, and the file name without the
 /// extension is the patch name they register.
 const PATCH_PREFIX: &str = "oauth2gw-";
 
 #[derive(Parser, Debug)]
 #[command(name = "kych-dbinit")]
 #[command(version)]
 #[command(about = "initialise and upgrade the KyCH database schema")]
 struct Args {
     #[command(flatten)]
     common: CommonArgs,
 
     /// Directory holding versioning.sql and the oauth2gw-*.sql patches
     /// (default: SQL_DIR from the configuration)
     #[arg(long)]
     sql_dir: Option<PathBuf>,
 
     /// Report what would be applied, without changing the database
     #[arg(long, short = 'n')]
     dry_run: bool,
 
     /// Print the connection the configuration names as PG* environment
     /// assignments and exit, without connecting.  This is how kych-dbconfig
     /// learns where the database is without having to parse kych.conf itself.
     #[arg(long)]
     print_connection: bool,
 }
 
 /// Exit status for "there is no database here (yet)": either it does not
 /// exist or the server cannot be reached.  The Debian package's postinst
 /// tells this apart from a real failure, because on a first install it is the
 /// expected outcome and must not make the installation fail.
 const EXIT_NO_DATABASE: i32 = 2;
 
 fn main() {
     let args = Args::parse();
 
     taler_main(CONFIG_SOURCE, args.common, async |cfg| {
         let config = Config::parse(cfg)?;
         if args.print_connection {
             print_connection(&config)?;
             return Ok(());
         }
         let sql_dir = match args.sql_dir {
             Some(dir) => dir,
             None => PathBuf::from(&config.sql_dir),
         };
         run(&config, &sql_dir, args.dry_run).await
     })
 }
 
 async fn run(config: &Config, sql_dir: &Path, dry_run: bool) -> Result<()> {
     let patches = available_patches(sql_dir)?;
     if patches.is_empty() {
         bail!(
             "no {}*.sql patch file in {}",
             PATCH_PREFIX,
             sql_dir.display()
         );
     }
 
     let mut conn = match PgConnection::connect_with(&config.database).await {
         Ok(conn) => conn,
         Err(err) if is_absent_database(&err) => {
             // Not an error worth a stack of context: on a first install this
             // is simply the state before kych-dbconfig has run.
             eprintln!(
                 "No database at {}: {}",
                 describe_target(config),
                 err
             );
             eprintln!("Run kych-dbconfig to create it.");
             std::process::exit(EXIT_NO_DATABASE);
         }
         Err(err) => return Err(err).context("Failed to connect to database"),
     };
 
     if dry_run {
         // Without the bookkeeping there is nothing to compare against, and
         // creating it would be a change: everything is pending in that case.
         let applied = if versioning_present(&mut conn).await? {
             applied_patches(&mut conn).await?
         } else {
             println!("{VERSIONING_SQL} would be loaded");
             HashSet::new()
         };
         let mut pending = 0;
         for (name, path) in &patches {
             if !applied.contains(name) {
                 println!("{} would be applied", path.display());
                 pending += 1;
             }
         }
         if pending == 0 {
             println!("database is at the latest patch level");
         }
         return Ok(());
     }
 
     // Idempotent: CREATE SCHEMA / TABLE IF NOT EXISTS and CREATE OR REPLACE
     // FUNCTION throughout, so an existing installation is left as it is.
     execute_sql_file(&mut conn, &sql_dir.join(VERSIONING_SQL)).await?;
 
     let applied = applied_patches(&mut conn).await?;
     let mut count = 0;
     for (name, path) in &patches {
         if applied.contains(name) {
             continue;
         }
         tracing::info!(target: "dbinit", "applying patch {}", name);
         execute_sql_file(&mut conn, path).await?;
         println!("applied {name}");
         count += 1;
     }
 
     if count == 0 {
         tracing::info!(target: "dbinit", "database is at the latest patch level");
     }
 
     conn.close().await.context("Failed to close connection")?;
     Ok(())
 }
 
 /// Whether the connection failed because there is nothing to connect to: the
 /// database does not exist (PostgreSQL `invalid_catalog_name`), or the server
 /// is not listening where the configuration says it is.  Anything else -- a
 /// rejected password, a TLS failure, a protocol error -- is a real problem
 /// and is reported as one.
 fn is_absent_database(err: &sqlx::Error) -> bool {
     match err {
         sqlx::Error::Database(db_err) => db_err.code().as_deref() == Some("3D000"),
         sqlx::Error::Io(_) => true,
         _ => false,
     }
 }
 
 /// Where the configured connection goes.  `?host=/some/dir` in the DATABASE
 /// URI makes sqlx connect over that Unix socket directory and leaves the TCP
 /// host at its default, so the socket has to be asked for first or the answer
 /// is a plausible-looking lie about localhost.
 fn connection_host(config: &Config) -> String {
     match config.database.get_socket() {
         Some(socket) => socket.to_string_lossy().into_owned(),
         None => config.database.get_host().to_owned(),
     }
 }
 
 /// Where we tried to connect, for a message that is useful without having to
 /// go and read the configuration file.
 fn describe_target(config: &Config) -> String {
     format!(
         "{}@{}:{}",
         config.database.get_database().unwrap_or("?"),
         connection_host(config),
         config.database.get_port()
     )
 }
 
 /// `PGDATABASE`/`PGHOST`/`PGPORT` assignments for a shell to eval.  Only
 /// these three, and deliberately not PGUSER: the caller connects as the
 /// PostgreSQL administrator, not as the daemon.
 fn print_connection(config: &Config) -> Result<()> {
     let name = config
         .database
         .get_database()
         .ok_or_else(|| anyhow!("DATABASE in the configuration names no database"))?;
     let host = connection_host(config);
     // A shell would mangle anything exotic, and a database or socket path
     // with a quote in it is a configuration mistake rather than a use case.
     for (name, value) in [("PGDATABASE", name), ("PGHOST", host.as_str())] {
         if value.contains(['\'', '"', '$', '`', '\\', '\n']) {
             bail!("{name} value '{value}' contains a character a shell would mangle");
         }
     }
     println!("PGDATABASE='{name}'");
     println!("PGHOST='{host}'");
     println!("PGPORT='{}'", config.database.get_port());
     Ok(())
 }
 
 /// The patch files in `sql_dir`, as (patch name, path), ordered by the number
 /// in the file name -- `oauth2gw-0002.sql` must not be applied before
 /// `oauth2gw-0001.sql`, and `register_patch` would refuse it anyway if it
 /// declared the earlier one as a requirement.
 fn available_patches(sql_dir: &Path) -> Result<Vec<(String, PathBuf)>> {
     let entries = std::fs::read_dir(sql_dir)
         .with_context(|| format!("Failed to read SQL directory {}", sql_dir.display()))?;
 
     let mut patches = Vec::new();
     for entry in entries {
         let path = entry
             .with_context(|| format!("Failed to read SQL directory {}", sql_dir.display()))?
             .path();
         let Some(name) = path.file_stem().and_then(|s| s.to_str()) else {
             continue;
         };
         if path.extension().and_then(|e| e.to_str()) != Some("sql") {
             continue;
         }
         let Some(number) = name.strip_prefix(PATCH_PREFIX) else {
             continue;
         };
         let number: u32 = number.parse().map_err(|_| {
             anyhow!(
                 "patch file {} does not end in a number",
                 path.display()
             )
         })?;
         patches.push((number, name.to_owned(), path));
     }
 
     patches.sort_by_key(|(number, _, _)| *number);
     Ok(patches
         .into_iter()
         .map(|(_, name, path)| (name, path))
         .collect())
 }
 
 async fn versioning_present(conn: &mut PgConnection) -> Result<bool> {
     let present: Option<String> =
         sqlx::query_scalar("SELECT to_regclass('_v.patches')::text")
             .fetch_one(&mut *conn)
             .await
             .context("Failed to look for the _v.patches table")?;
     Ok(present.is_some())
 }
 
 async fn applied_patches(conn: &mut PgConnection) -> Result<HashSet<String>> {
     let rows = sqlx::query("SELECT patch_name FROM _v.patches")
         .fetch_all(&mut *conn)
         .await
         .context("Failed to read the applied patches from _v.patches")?;
     Ok(rows
         .into_iter()
         .map(|row| row.get::<String, _>("patch_name"))
         .collect())
 }
 
 /// Run a whole SQL file.  The files carry their own `BEGIN`/`COMMIT`, which is
 /// what makes each patch all-or-nothing, so they must not be wrapped in
 /// another transaction here.
 async fn execute_sql_file(conn: &mut PgConnection, path: &Path) -> Result<()> {
     let sql = std::fs::read_to_string(path)
         .with_context(|| format!("Failed to read {}", path.display()))?;
     conn.execute(sqlx::raw_sql(&sql))
         .await
         .with_context(|| format!("Failed to apply {}", path.display()))?;
     Ok(())
 }