rustyroad 1.7.3

Rusty Road is a framework written in Rust that is based on Ruby on Rails. It is designed to provide the familiar conventions and ease of use of Ruby on Rails, while also taking advantage of the performance and efficiency of Rust.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

//! Explicit repair of legacy migration ledgers.

use crate::database::migrations::CustomMigrationError;
use crate::database::DatabaseConnection;

const COUNT_ROWS: &str = "SELECT COUNT(*) FROM _rustyroad_migrations";
const DELETE_STALE_ROWS: &str = "DELETE FROM _rustyroad_migrations WHERE id IN (
    SELECT id FROM (
        SELECT id, ROW_NUMBER() OVER (
            PARTITION BY name ORDER BY applied_at DESC, id DESC
        ) AS row_rank
        FROM _rustyroad_migrations
    ) ranked WHERE row_rank > 1
)";
const UNIQUE_INDEX: &str = "_rustyroad_migrations_name_direction_key";

/// Summarizes an idempotent migration-ledger repair.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RepairReport {
    /// Rows present before repair.
    pub rows_before: i64,
    /// Rows retained after repair.
    pub rows_after: i64,
    /// Superseded duplicate rows deleted by repair.
    pub rows_removed: u64,
}

/// Retains the newest row per migration identity and enforces ledger uniqueness.
///
/// The operation is transactional and idempotent. A failed delete or index build
/// rolls the entire repair back.
pub async fn repair(connection: &DatabaseConnection) -> Result<RepairReport, CustomMigrationError> {
    match connection {
        DatabaseConnection::Pg(pool) => {
            let mut transaction = pool.begin().await?;
            let before = sqlx::query_scalar::<_, i64>(COUNT_ROWS)
                .fetch_one(&mut *transaction)
                .await?;
            let removed = sqlx::query(DELETE_STALE_ROWS)
                .execute(&mut *transaction)
                .await?
                .rows_affected();
            let index = format!(
                "CREATE UNIQUE INDEX IF NOT EXISTS {UNIQUE_INDEX} \
                 ON _rustyroad_migrations (name, direction)"
            );
            sqlx::query(&index).execute(&mut *transaction).await?;
            let after = sqlx::query_scalar::<_, i64>(COUNT_ROWS)
                .fetch_one(&mut *transaction)
                .await?;
            transaction.commit().await?;
            Ok(RepairReport {
                rows_before: before,
                rows_after: after,
                rows_removed: removed,
            })
        }
        DatabaseConnection::Sqlite(pool) => {
            let mut transaction = pool.begin().await?;
            let before = sqlx::query_scalar::<_, i64>(COUNT_ROWS)
                .fetch_one(&mut *transaction)
                .await?;
            let removed = sqlx::query(DELETE_STALE_ROWS)
                .execute(&mut *transaction)
                .await?
                .rows_affected();
            let index = format!(
                "CREATE UNIQUE INDEX IF NOT EXISTS {UNIQUE_INDEX} \
                 ON _rustyroad_migrations (name, direction)"
            );
            sqlx::query(&index).execute(&mut *transaction).await?;
            let after = sqlx::query_scalar::<_, i64>(COUNT_ROWS)
                .fetch_one(&mut *transaction)
                .await?;
            transaction.commit().await?;
            Ok(RepairReport {
                rows_before: before,
                rows_after: after,
                rows_removed: removed,
            })
        }
        DatabaseConnection::MySql(pool) => {
            let mut transaction = pool.begin().await?;
            let before = sqlx::query_scalar::<_, i64>(COUNT_ROWS)
                .fetch_one(&mut *transaction)
                .await?;
            let removed = sqlx::query(DELETE_STALE_ROWS)
                .execute(&mut *transaction)
                .await?
                .rows_affected();
            let index_count = sqlx::query_scalar::<_, i64>(
                "SELECT COUNT(*) FROM information_schema.statistics \
                 WHERE table_schema = DATABASE() \
                   AND table_name = '_rustyroad_migrations' \
                   AND index_name = '_rustyroad_migrations_name_direction_key'",
            )
            .fetch_one(&mut *transaction)
            .await?;
            if index_count == 0 {
                let index = format!(
                    "CREATE UNIQUE INDEX {UNIQUE_INDEX} \
                     ON _rustyroad_migrations (name, direction)"
                );
                sqlx::query(&index).execute(&mut *transaction).await?;
            }
            let after = sqlx::query_scalar::<_, i64>(COUNT_ROWS)
                .fetch_one(&mut *transaction)
                .await?;
            transaction.commit().await?;
            Ok(RepairReport {
                rows_before: before,
                rows_after: after,
                rows_removed: removed,
            })
        }
    }
}