platform/event-log/src/schema.rs

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

use sqlx::SqliteConnection;

use crate::StoreError;

pub(crate) mod terminal_display;

static MIGRATIONS: sqlx::migrate::Migrator = sqlx::migrate!("./migrations");

pub(crate) async fn initialize_connection(
    connection: &mut SqliteConnection,
) -> Result<(), StoreError> {
    let application_id: i64 = sqlx::query_scalar("PRAGMA application_id")
        .fetch_one(&mut *connection)
        .await?;
    let version: i64 = sqlx::query_scalar("PRAGMA user_version")
        .fetch_one(&mut *connection)
        .await?;
    let tables: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM sqlite_schema WHERE type = 'table' AND name NOT LIKE 'sqlite_%'",
    )
    .fetch_one(&mut *connection)
    .await?;
    if !((application_id == 0 && version == 0 && tables == 0)
        || (application_id == 0x4d414b52 && matches!(version, 0..=1)))
    {
        return Err(StoreError::UnsupportedDatabase);
    }
    let ledger: bool = sqlx::query_scalar(
        "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE type = 'table' AND name = '_sqlx_migrations')",
    )
    .fetch_one(&mut *connection)
    .await?;
    // Version zero is only a genuinely empty or interrupted initial setup.
    // Existing data is never adopted by creating a new migration ledger.
    if (version == 0 && tables != i64::from(ledger)) || (version == 1 && !ledger) {
        return Err(StoreError::UnsupportedDatabase);
    }
    if ledger {
        let applied: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM _sqlx_migrations")
            .fetch_one(&mut *connection)
            .await?;
        if (version == 0) != (applied == 0) {
            return Err(StoreError::UnsupportedDatabase);
        }
    }
    // A fresh file needs pointer-map pages before SQLx creates its first table.
    // Reclamation can then return space in small steps without a full VACUUM.
    if tables == 0 {
        sqlx::query("PRAGMA auto_vacuum = INCREMENTAL")
            .execute(&mut *connection)
            .await?;
    }
    // Establish Rust identity before SQLx creates its migration ledger, so a
    // crash before migration 1 is distinguishable from an unrelated database.
    if application_id == 0 {
        sqlx::query("PRAGMA application_id = 1296124754")
            .execute(&mut *connection)
            .await?;
    }
    crate::sqlite_functions::register(connection).await?;
    terminal_display::register(connection, true).await?;
    let migrated = MIGRATIONS.run_direct(None, &mut *connection, false).await;
    let removed = terminal_display::register(connection, false).await;
    migrated?;
    removed?;
    sqlx::raw_sql("PRAGMA journal_mode = WAL; PRAGMA synchronous = FULL;")
        .execute(&mut *connection)
        .await?;
    crate::message_sources::initialize(connection).await?;
    crate::sessions::initialize_execution(connection).await?;
    crate::transcript::initialize_cache(connection).await?;
    Ok(())
}

#[cfg(test)]
#[path = "../tests/migrations/prior_artifacts.rs"]
mod artifact_payload_tests;