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];
44
45pub async fn apply_migrations(pool: &PgPool, migrations: &[Migration]) -> AppResult<()> {
46    ensure_migration_table(pool).await?;
47
48    for migration in migrations {
49        apply_migration(pool, migration).await?;
50    }
51
52    Ok(())
53}
54
55async fn ensure_migration_table(pool: &PgPool) -> AppResult<()> {
56    sqlx::raw_sql(
57        r#"
58        create schema if not exists platform;
59
60        create table if not exists platform.schema_migrations (
61            name text primary key,
62            applied_at timestamptz not null default now()
63        );
64        "#,
65    )
66    .execute(pool)
67    .await
68    .map(|_| ())
69    .map_err(map_migration_error)
70}
71
72async fn apply_migration(pool: &PgPool, migration: &Migration) -> AppResult<()> {
73    let mut tx = pool.begin().await.map_err(map_migration_error)?;
74
75    let already_applied: Option<String> = sqlx::query_scalar(
76        r#"
77        select name
78        from platform.schema_migrations
79        where name = $1
80        "#,
81    )
82    .bind(migration.name)
83    .fetch_optional(&mut *tx)
84    .await
85    .map_err(map_migration_error)?;
86
87    if already_applied.is_some() {
88        tx.commit().await.map_err(map_migration_error)?;
89        return Ok(());
90    }
91
92    sqlx::raw_sql(migration.sql)
93        .execute(&mut *tx)
94        .await
95        .map_err(map_migration_error)?;
96
97    sqlx::query(
98        r#"
99        insert into platform.schema_migrations (name)
100        values ($1)
101        on conflict (name) do nothing
102        "#,
103    )
104    .bind(migration.name)
105    .execute(&mut *tx)
106    .await
107    .map_err(map_migration_error)?;
108
109    tx.commit().await.map_err(map_migration_error)
110}
111
112fn map_migration_error(source: sqlx::Error) -> AppError {
113    AppError::new(ErrorCode::Internal, "Database migration failed").with_source(source)
114}
platform_core/migrations.rs

platform_core/
migrations.rs
