platform_core/
migrations.rs

1use crate::error::{AppError, AppResult, ErrorCode};
2use sqlx::PgPool;
3
4#[derive(Debug, Clone, Copy)]
5pub struct Migration {
6    pub name: &'static str,
7    pub sql: &'static str,
8}
9
10pub const PLATFORM_MIGRATIONS: &[Migration] = &[
11    Migration {
12        name: "platform/0001_create_platform_schema",
13        sql: include_str!("../migrations/0001_create_platform_schema.sql"),
14    },
15    Migration {
16        name: "platform/0002_create_outbox",
17        sql: include_str!("../migrations/0002_create_outbox.sql"),
18    },
19    Migration {
20        name: "platform/0003_extend_outbox_delivery_fields",
21        sql: include_str!("../migrations/0003_extend_outbox_delivery_fields.sql"),
22    },
23    Migration {
24        name: "platform/0004_add_outbox_summary_index",
25        sql: include_str!("../migrations/0004_add_outbox_summary_index.sql"),
26    },
27    Migration {
28        name: "platform/0005_create_execution_logs",
29        sql: include_str!("../migrations/0005_create_execution_logs.sql"),
30    },
31    Migration {
32        name: "platform/0006_create_story_events",
33        sql: include_str!("../migrations/0006_create_story_events.sql"),
34    },
35    Migration {
36        name: "platform/0007_create_config_schema",
37        sql: include_str!("../migrations/0007_create_config_schema.sql"),
38    },
39    Migration {
40        name: "platform/0008_create_remote_http_proxy_calls",
41        sql: include_str!("../migrations/0008_create_remote_http_proxy_calls.sql"),
42    },
43    Migration {
44        name: "platform/0009_add_story_query_indexes",
45        sql: include_str!("../migrations/0009_add_story_query_indexes.sql"),
46    },
47];
48
49pub async fn apply_migrations(pool: &PgPool, migrations: &[Migration]) -> AppResult<()> {
50    ensure_migration_table(pool).await?;
51
52    for migration in migrations {
53        apply_migration(pool, migration).await?;
54    }
55
56    Ok(())
57}
58
59async fn ensure_migration_table(pool: &PgPool) -> AppResult<()> {
60    sqlx::raw_sql(
61        r#"
62        create schema if not exists platform;
63
64        create table if not exists platform.schema_migrations (
65            name text primary key,
66            applied_at timestamptz not null default now()
67        );
68        "#,
69    )
70    .execute(pool)
71    .await
72    .map(|_| ())
73    .map_err(map_migration_error)
74}
75
76async fn apply_migration(pool: &PgPool, migration: &Migration) -> AppResult<()> {
77    let mut tx = pool.begin().await.map_err(map_migration_error)?;
78
79    let already_applied: Option<String> = sqlx::query_scalar(
80        r#"
81        select name
82        from platform.schema_migrations
83        where name = $1
84        "#,
85    )
86    .bind(migration.name)
87    .fetch_optional(&mut *tx)
88    .await
89    .map_err(map_migration_error)?;
90
91    if already_applied.is_some() {
92        tx.commit().await.map_err(map_migration_error)?;
93        return Ok(());
94    }
95
96    sqlx::raw_sql(migration.sql)
97        .execute(&mut *tx)
98        .await
99        .map_err(map_migration_error)?;
100
101    sqlx::query(
102        r#"
103        insert into platform.schema_migrations (name)
104        values ($1)
105        on conflict (name) do nothing
106        "#,
107    )
108    .bind(migration.name)
109    .execute(&mut *tx)
110    .await
111    .map_err(map_migration_error)?;
112
113    tx.commit().await.map_err(map_migration_error)
114}
115
116fn map_migration_error(source: sqlx::Error) -> AppError {
117    AppError::new(ErrorCode::Internal, "Database migration failed").with_source(source)
118}
platform_core/migrations.rs

platform_core/
migrations.rs
