bsodfather/crank
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: make registry postgres-first

Browse Source
This commit is contained in:
a.tolmachev
2026-03-25 17:19:54 +03:00
parent b32b702d67
commit d37d40a975
12 changed files with 622 additions and 265 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/mcpaas-registry/src/lib.rs
+2 -2
View File
@@ -1,7 +1,7 @@
mod error;
mod migrations;
mod model;
mod sqlite;
mod postgres;
pub use error::RegistryError;
pub use model::{
@@ -11,4 +11,4 @@ pub use model::{
SaveSampleMetadataRequest, YamlImportJob, YamlImportJobCompletion, YamlImportJobId,
YamlImportJobStatus,
};
pub use sqlite::SqliteRegistry;
pub use postgres::PostgresRegistry;
crates/mcpaas-registry/src/migrations.rs
+32 -38
View File
@@ -1,6 +1,6 @@
use sqlx::{SqlitePool, query};
use sqlx::{PgPool, query};
pub async fn apply_sqlite(pool: &SqlitePool) -> Result<(), sqlx::Error> {
pub async fn apply_postgres(pool: &PgPool) -> Result<(), sqlx::Error> {
query(
"create table if not exists operations (
id text primary key,
@@ -10,9 +10,9 @@ pub async fn apply_sqlite(pool: &SqlitePool) -> Result<(), sqlx::Error> {
status text not null,
current_draft_version integer not null default 1,
latest_published_version integer null,
created_at text not null,
updated_at text not null,
published_at text null
created_at timestamptz not null,
updated_at timestamptz not null,
published_at timestamptz null
)",
)
.execute(pool)
@@ -20,24 +20,23 @@ pub async fn apply_sqlite(pool: &SqlitePool) -> Result<(), sqlx::Error> {
query(
"create table if not exists operation_versions (
operation_id text not null,
operation_id text not null references operations(id) on delete cascade,
version integer not null,
status text not null,
target_json text not null,
input_schema_json text not null,
output_schema_json text not null,
input_mapping_json text not null,
output_mapping_json text not null,
execution_config_json text not null,
tool_description_json text not null,
samples_json text null,
generated_draft_json text null,
config_export_json text null,
target_json jsonb not null,
input_schema_json jsonb not null,
output_schema_json jsonb not null,
input_mapping_json jsonb not null,
output_mapping_json jsonb not null,
execution_config_json jsonb not null,
tool_description_json jsonb not null,
samples_json jsonb null,
generated_draft_json jsonb null,
config_export_json jsonb null,
change_note text null,
created_at text not null,
created_at timestamptz not null,
created_by text null,
primary key (operation_id, version),
foreign key (operation_id) references operations(id) on delete cascade
primary key (operation_id, version)
)",
)
.execute(pool)
@@ -45,11 +44,10 @@ pub async fn apply_sqlite(pool: &SqlitePool) -> Result<(), sqlx::Error> {
query(
"create table if not exists published_operations (
operation_id text primary key,
operation_id text primary key references operations(id) on delete cascade,
version integer not null,
published_at text not null,
published_at timestamptz not null,
published_by text null,
foreign key (operation_id) references operations(id) on delete cascade,
foreign key (operation_id, version) references operation_versions(operation_id, version) on delete cascade
)",
)
@@ -59,14 +57,13 @@ pub async fn apply_sqlite(pool: &SqlitePool) -> Result<(), sqlx::Error> {
query(
"create table if not exists operation_samples (
id text primary key,
operation_id text not null,
operation_id text not null references operations(id) on delete cascade,
version integer not null,
sample_kind text not null,
storage_ref text not null,
content_type text not null,
file_name text null,
created_at text not null,
foreign key (operation_id) references operations(id) on delete cascade,
created_at timestamptz not null,
foreign key (operation_id, version) references operation_versions(operation_id, version) on delete cascade
)",
)
@@ -76,14 +73,13 @@ pub async fn apply_sqlite(pool: &SqlitePool) -> Result<(), sqlx::Error> {
query(
"create table if not exists descriptors (
id text primary key,
operation_id text null,
operation_id text null references operations(id) on delete cascade,
version integer null,
descriptor_kind text not null,
storage_ref text not null,
source_name text null,
package_index_json text null,
created_at text not null,
foreign key (operation_id) references operations(id) on delete cascade,
package_index_json jsonb null,
created_at timestamptz not null,
foreign key (operation_id, version) references operation_versions(operation_id, version) on delete cascade
)",
)
@@ -95,9 +91,9 @@ pub async fn apply_sqlite(pool: &SqlitePool) -> Result<(), sqlx::Error> {
id text primary key,
name text not null unique,
kind text not null,
config_json text not null,
created_at text not null,
updated_at text not null
config_json jsonb not null,
created_at timestamptz not null,
updated_at timestamptz not null
)",
)
.execute(pool)
@@ -106,17 +102,15 @@ pub async fn apply_sqlite(pool: &SqlitePool) -> Result<(), sqlx::Error> {
query(
"create table if not exists yaml_import_jobs (
id text primary key,
source_sample_id text null,
source_sample_id text null references operation_samples(id) on delete set null,
status text not null,
format_version text not null,
mode text not null,
result_operation_id text null,
result_operation_id text null references operations(id) on delete set null,
result_version integer null,
error_text text null,
created_at text not null,
finished_at text null,
foreign key (source_sample_id) references operation_samples(id) on delete set null,
foreign key (result_operation_id) references operations(id) on delete set null
created_at timestamptz not null,
finished_at timestamptz null
)",
)
.execute(pool)
crates/mcpaas-registry/src/sqlite.rs → crates/mcpaas-registry/src/postgres.rs
+251 -147
View File
@@ -1,7 +1,11 @@
use mcpaas_core::{AuthProfile, OperationId, OperationStatus};
use serde::{Serialize, de::DeserializeOwned};
use serde_json::Value;
use sqlx::{Row, SqlitePool, sqlite::SqlitePoolOptions};
use sqlx::{
PgPool, Postgres, Row, Transaction,
postgres::{PgPoolOptions, PgRow},
types::Json,
};
use crate::{
error::RegistryError,
@@ -16,35 +20,44 @@ use crate::{
};
#[derive(Clone, Debug)]
pub struct SqliteRegistry {
pool: SqlitePool,
pub struct PostgresRegistry {
pool: PgPool,
}
impl SqliteRegistry {
impl PostgresRegistry {
pub async fn connect(database_url: &str) -> Result<Self, RegistryError> {
let pool = SqlitePoolOptions::new()
.max_connections(1)
.connect(database_url)
.await?;
sqlx::query("pragma foreign_keys = on")
.execute(&pool)
.await?;
let registry = Self { pool };
registry.migrate().await?;
Ok(registry)
Self::connect_in_schema(database_url, None).await
}
pub fn pool(&self) -> &SqlitePool {
pub fn pool(&self) -> &PgPool {
&self.pool
}
pub async fn migrate(&self) -> Result<(), RegistryError> {
migrations::apply_sqlite(&self.pool).await?;
migrations::apply_postgres(&self.pool).await?;
Ok(())
}
async fn connect_in_schema(
database_url: &str,
schema: Option<&str>,
) -> Result<Self, RegistryError> {
let pool = PgPoolOptions::new()
.max_connections(1)
.connect(database_url)
.await?;
if let Some(schema) = schema {
sqlx::query(&format!("set search_path to {schema}"))
.execute(&pool)
.await?;
}
let registry = Self { pool };
registry.migrate().await?;
Ok(registry)
}
pub async fn create_operation(
&self,
snapshot: &RegistryOperation,
@@ -77,7 +90,12 @@ impl SqliteRegistry {
created_at,
updated_at,
published_at
) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
) values (
$1, $2, $3, $4, $5, $6, $7,
$8::timestamptz,
$9::timestamptz,
$10::timestamptz
)",
)
.bind(snapshot.id.as_str())
.bind(&snapshot.name)
@@ -136,10 +154,10 @@ impl SqliteRegistry {
sqlx::query(
"update operations
set status = ?,
current_draft_version = ?,
updated_at = ?
where id = ?",
set status = $1,
current_draft_version = $2,
updated_at = $3::timestamptz
where id = $4",
)
.bind(serialize_enum_text(&request.snapshot.status, "status")?)
.bind(to_db_version(request.snapshot.version))
@@ -162,9 +180,9 @@ impl SqliteRegistry {
status,
current_draft_version,
latest_published_version,
created_at,
updated_at,
published_at
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at,
to_char(published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as published_at
from operations
order by name asc",
)
@@ -187,11 +205,11 @@ impl SqliteRegistry {
status,
current_draft_version,
latest_published_version,
created_at,
updated_at,
published_at
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at,
to_char(published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as published_at
from operations
where id = ?",
where id = $1",
)
.bind(operation_id.as_str())
.fetch_optional(&self.pool)
@@ -211,9 +229,9 @@ impl SqliteRegistry {
o.name,
o.display_name,
o.protocol,
o.created_at as operation_created_at,
o.updated_at as operation_updated_at,
o.published_at as operation_published_at,
to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at,
to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at,
to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at,
ov.version,
ov.status,
ov.target_json,
@@ -227,11 +245,11 @@ impl SqliteRegistry {
ov.generated_draft_json,
ov.config_export_json,
ov.change_note,
ov.created_at,
to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
ov.created_by
from operation_versions ov
join operations o on o.id = ov.operation_id
where ov.operation_id = ? and ov.version = ?",
where ov.operation_id = $1 and ov.version = $2",
)
.bind(operation_id.as_str())
.bind(to_db_version(version))
@@ -251,9 +269,9 @@ impl SqliteRegistry {
o.name,
o.display_name,
o.protocol,
o.created_at as operation_created_at,
o.updated_at as operation_updated_at,
o.published_at as operation_published_at,
to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at,
to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at,
to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at,
ov.version,
ov.status,
ov.target_json,
@@ -267,11 +285,11 @@ impl SqliteRegistry {
ov.generated_draft_json,
ov.config_export_json,
ov.change_note,
ov.created_at,
to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
ov.created_by
from operation_versions ov
join operations o on o.id = ov.operation_id
where ov.operation_id = ?
where ov.operation_id = $1
order by ov.version asc",
)
.bind(operation_id.as_str())
@@ -304,7 +322,7 @@ impl SqliteRegistry {
version,
published_at,
published_by
) values (?, ?, ?, ?)
) values ($1, $2, $3::timestamptz, $4)
on conflict(operation_id) do update set
version = excluded.version,
published_at = excluded.published_at,
@@ -319,8 +337,8 @@ impl SqliteRegistry {
sqlx::query(
"update operation_versions
set status = ?
where operation_id = ? and version = ?",
set status = $1
where operation_id = $2 and version = $3",
)
.bind(serialize_enum_text(&OperationStatus::Published, "status")?)
.bind(request.operation_id.as_str())
@@ -330,11 +348,11 @@ impl SqliteRegistry {
sqlx::query(
"update operations
set status = ?,
latest_published_version = ?,
published_at = ?,
updated_at = ?
where id = ?",
set status = $1,
latest_published_version = $2,
published_at = $3::timestamptz,
updated_at = $4::timestamptz
where id = $5",
)
.bind(serialize_enum_text(&OperationStatus::Published, "status")?)
.bind(to_db_version(request.version))
@@ -358,9 +376,9 @@ impl SqliteRegistry {
o.name,
o.display_name,
o.protocol,
o.created_at as operation_created_at,
o.updated_at as operation_updated_at,
o.published_at as operation_published_at,
to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at,
to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at,
to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at,
ov.version,
ov.status,
ov.target_json,
@@ -374,13 +392,13 @@ impl SqliteRegistry {
ov.generated_draft_json,
ov.config_export_json,
ov.change_note,
ov.created_at,
to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
ov.created_by
from published_operations po
join operation_versions ov
on ov.operation_id = po.operation_id and ov.version = po.version
join operations o on o.id = po.operation_id
where po.operation_id = ?",
where po.operation_id = $1",
)
.bind(operation_id.as_str())
.fetch_optional(&self.pool)
@@ -398,9 +416,9 @@ impl SqliteRegistry {
o.name,
o.display_name,
o.protocol,
o.created_at as operation_created_at,
o.updated_at as operation_updated_at,
o.published_at as operation_published_at,
to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at,
to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at,
to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at,
ov.version,
ov.status,
ov.target_json,
@@ -414,7 +432,7 @@ impl SqliteRegistry {
ov.generated_draft_json,
ov.config_export_json,
ov.change_note,
ov.created_at,
to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
ov.created_by
from published_operations po
join operation_versions ov
@@ -442,7 +460,7 @@ impl SqliteRegistry {
config_json,
created_at,
updated_at
) values (?, ?, ?, ?, ?, ?)
) values ($1, $2, $3, $4, $5::timestamptz, $6::timestamptz)
on conflict(id) do update set
name = excluded.name,
kind = excluded.kind,
@@ -452,7 +470,7 @@ impl SqliteRegistry {
.bind(request.profile.id.as_str())
.bind(&request.profile.name)
.bind(serialize_enum_text(&request.profile.kind, "kind")?)
.bind(serialize_json(&request.profile.config)?)
.bind(Json(serialize_json_value(&request.profile.config)?))
.bind(&request.profile.created_at)
.bind(&request.profile.updated_at)
.execute(&self.pool)
@@ -466,9 +484,15 @@ impl SqliteRegistry {
auth_profile_id: &mcpaas_core::AuthProfileId,
) -> Result<Option<AuthProfile>, RegistryError> {
let row = sqlx::query(
"select id, name, kind, config_json, created_at, updated_at
"select
id,
name,
kind,
config_json,
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at
from auth_profiles
where id = ?",
where id = $1",
)
.bind(auth_profile_id.as_str())
.fetch_optional(&self.pool)
@@ -479,7 +503,13 @@ impl SqliteRegistry {
pub async fn list_auth_profiles(&self) -> Result<Vec<AuthProfile>, RegistryError> {
let rows = sqlx::query(
"select id, name, kind, config_json, created_at, updated_at
"select
id,
name,
kind,
config_json,
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at
from auth_profiles
order by name asc",
)
@@ -503,7 +533,7 @@ impl SqliteRegistry {
content_type,
file_name,
created_at
) values (?, ?, ?, ?, ?, ?, ?, ?)
) values ($1, $2, $3, $4, $5, $6, $7, $8::timestamptz)
on conflict(id) do update set
operation_id = excluded.operation_id,
version = excluded.version,
@@ -544,9 +574,9 @@ impl SqliteRegistry {
storage_ref,
content_type,
file_name,
created_at
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at
from operation_samples
where operation_id = ? and version = ?
where operation_id = $1 and version = $2
order by created_at asc",
)
.bind(operation_id.as_str())
@@ -571,7 +601,7 @@ impl SqliteRegistry {
source_name,
package_index_json,
created_at
) values (?, ?, ?, ?, ?, ?, ?, ?)
) values ($1, $2, $3, $4, $5, $6, $7, $8::timestamptz)
on conflict(id) do update set
operation_id = excluded.operation_id,
version = excluded.version,
@@ -596,7 +626,7 @@ impl SqliteRegistry {
)?)
.bind(&request.descriptor.storage_ref)
.bind(request.descriptor.source_name.as_deref())
.bind(serialize_option_json(&request.descriptor.package_index)?)
.bind(serialize_option_json_value(&request.descriptor.package_index)?.map(Json))
.bind(&request.descriptor.created_at)
.execute(&self.pool)
.await?;
@@ -618,9 +648,9 @@ impl SqliteRegistry {
storage_ref,
source_name,
package_index_json,
created_at
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at
from descriptors
where operation_id = ? and version = ?
where operation_id = $1 and version = $2
order by created_at asc",
)
.bind(operation_id.as_str())
@@ -647,7 +677,7 @@ impl SqliteRegistry {
error_text,
created_at,
finished_at
) values (?, ?, ?, ?, ?, null, null, null, ?, null)",
) values ($1, $2, $3, $4, $5, null, null, null, $6::timestamptz, null)",
)
.bind(request.id.as_str())
.bind(request.source_sample_id.map(|value| value.as_str()))
@@ -671,12 +701,12 @@ impl SqliteRegistry {
) -> Result<(), RegistryError> {
let result = sqlx::query(
"update yaml_import_jobs
set status = ?,
result_operation_id = ?,
result_version = ?,
error_text = ?,
finished_at = ?
where id = ?",
set status = $1,
result_operation_id = $2,
result_version = $3,
error_text = $4,
finished_at = $5::timestamptz
where id = $6",
)
.bind(serialize_enum_text(&completion.status, "status")?)
.bind(
@@ -715,10 +745,10 @@ impl SqliteRegistry {
result_operation_id,
result_version,
error_text,
created_at,
finished_at
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at
from yaml_import_jobs
where id = ?",
where id = $1",
)
.bind(job_id.as_str())
.fetch_optional(&self.pool)
@@ -729,7 +759,7 @@ impl SqliteRegistry {
}
async fn insert_version_row(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
tx: &mut Transaction<'_, Postgres>,
snapshot: &RegistryOperation,
change_note: Option<&str>,
created_by: Option<&str>,
@@ -752,21 +782,24 @@ async fn insert_version_row(
change_note,
created_at,
created_by
) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
) values (
$1, $2, $3, $4, $5, $6, $7, $8,
$9, $10, $11, $12, $13, $14, $15::timestamptz, $16
)",
)
.bind(snapshot.id.as_str())
.bind(to_db_version(snapshot.version))
.bind(serialize_enum_text(&snapshot.status, "status")?)
.bind(serialize_json(&snapshot.target)?)
.bind(serialize_json(&snapshot.input_schema)?)
.bind(serialize_json(&snapshot.output_schema)?)
.bind(serialize_json(&snapshot.input_mapping)?)
.bind(serialize_json(&snapshot.output_mapping)?)
.bind(serialize_json(&snapshot.execution_config)?)
.bind(serialize_json(&snapshot.tool_description)?)
.bind(serialize_option_json(&snapshot.samples)?)
.bind(serialize_option_json(&snapshot.generated_draft)?)
.bind(serialize_option_json(&snapshot.config_export)?)
.bind(Json(serialize_json_value(&snapshot.target)?))
.bind(Json(serialize_json_value(&snapshot.input_schema)?))
.bind(Json(serialize_json_value(&snapshot.output_schema)?))
.bind(Json(serialize_json_value(&snapshot.input_mapping)?))
.bind(Json(serialize_json_value(&snapshot.output_mapping)?))
.bind(Json(serialize_json_value(&snapshot.execution_config)?))
.bind(Json(serialize_json_value(&snapshot.tool_description)?))
.bind(serialize_option_json_value(&snapshot.samples)?.map(Json))
.bind(serialize_option_json_value(&snapshot.generated_draft)?.map(Json))
.bind(serialize_option_json_value(&snapshot.config_export)?.map(Json))
.bind(change_note)
.bind(&snapshot.updated_at)
.bind(created_by)
@@ -804,7 +837,7 @@ fn assert_immutable_fields(
Ok(())
}
fn map_operation_summary(row: &sqlx::sqlite::SqliteRow) -> Result<OperationSummary, RegistryError> {
fn map_operation_summary(row: &PgRow) -> Result<OperationSummary, RegistryError> {
Ok(OperationSummary {
id: OperationId::new(row.try_get::<String, _>("id")?),
name: row.try_get("name")?,
@@ -816,7 +849,7 @@ fn map_operation_summary(row: &sqlx::sqlite::SqliteRow) -> Result<OperationSumma
"current_draft_version",
)?,
latest_published_version: row
.try_get::<Option<i64>, _>("latest_published_version")?
.try_get::<Option<i32>, _>("latest_published_version")?
.map(|value| from_db_version(value, "latest_published_version"))
.transpose()?,
created_at: row.try_get("created_at")?,
@@ -825,9 +858,7 @@ fn map_operation_summary(row: &sqlx::sqlite::SqliteRow) -> Result<OperationSumma
})
}
fn map_operation_version_record(
row: &sqlx::sqlite::SqliteRow,
) -> Result<OperationVersionRecord, RegistryError> {
fn map_operation_version_record(row: &PgRow) -> Result<OperationVersionRecord, RegistryError> {
let operation_id = OperationId::new(row.try_get::<String, _>("id")?);
let version = from_db_version(row.try_get("version")?, "version")?;
let status = deserialize_enum_text(&row.try_get::<String, _>("status")?, "status")?;
@@ -846,31 +877,36 @@ fn map_operation_version_record(
protocol: deserialize_enum_text(&row.try_get::<String, _>("protocol")?, "protocol")?,
status,
version,
target: deserialize_json(&row.try_get::<String, _>("target_json")?)?,
input_schema: deserialize_json(&row.try_get::<String, _>("input_schema_json")?)?,
output_schema: deserialize_json(&row.try_get::<String, _>("output_schema_json")?)?,
input_mapping: deserialize_json(&row.try_get::<String, _>("input_mapping_json")?)?,
output_mapping: deserialize_json(&row.try_get::<String, _>("output_mapping_json")?)?,
execution_config: deserialize_json(
&row.try_get::<String, _>("execution_config_json")?,
target: deserialize_json_value(row.try_get::<Json<Value>, _>("target_json")?.0)?,
input_schema: deserialize_json_value(
row.try_get::<Json<Value>, _>("input_schema_json")?.0,
)?,
tool_description: deserialize_json(
&row.try_get::<String, _>("tool_description_json")?,
output_schema: deserialize_json_value(
row.try_get::<Json<Value>, _>("output_schema_json")?.0,
)?,
input_mapping: deserialize_json_value(
row.try_get::<Json<Value>, _>("input_mapping_json")?.0,
)?,
output_mapping: deserialize_json_value(
row.try_get::<Json<Value>, _>("output_mapping_json")?.0,
)?,
execution_config: deserialize_json_value(
row.try_get::<Json<Value>, _>("execution_config_json")?.0,
)?,
tool_description: deserialize_json_value(
row.try_get::<Json<Value>, _>("tool_description_json")?.0,
)?,
samples: row
.try_get::<Option<String>, _>("samples_json")?
.as_deref()
.map(deserialize_json)
.try_get::<Option<Json<Value>>, _>("samples_json")?
.map(|value| deserialize_json_value(value.0))
.transpose()?,
generated_draft: row
.try_get::<Option<String>, _>("generated_draft_json")?
.as_deref()
.map(deserialize_json)
.try_get::<Option<Json<Value>>, _>("generated_draft_json")?
.map(|value| deserialize_json_value(value.0))
.transpose()?,
config_export: row
.try_get::<Option<String>, _>("config_export_json")?
.as_deref()
.map(deserialize_json)
.try_get::<Option<Json<Value>>, _>("config_export_json")?
.map(|value| deserialize_json_value(value.0))
.transpose()?,
created_at: row.try_get("operation_created_at")?,
updated_at: row.try_get("operation_updated_at")?,
@@ -879,20 +915,18 @@ fn map_operation_version_record(
})
}
fn map_auth_profile(row: &sqlx::sqlite::SqliteRow) -> Result<AuthProfile, RegistryError> {
fn map_auth_profile(row: &PgRow) -> Result<AuthProfile, RegistryError> {
Ok(AuthProfile {
id: mcpaas_core::AuthProfileId::new(row.try_get::<String, _>("id")?),
name: row.try_get("name")?,
kind: deserialize_enum_text(&row.try_get::<String, _>("kind")?, "kind")?,
config: deserialize_json(&row.try_get::<String, _>("config_json")?)?,
config: deserialize_json_value(row.try_get::<Json<Value>, _>("config_json")?.0)?,
created_at: row.try_get("created_at")?,
updated_at: row.try_get("updated_at")?,
})
}
fn map_sample_metadata(
row: &sqlx::sqlite::SqliteRow,
) -> Result<OperationSampleMetadata, RegistryError> {
fn map_sample_metadata(row: &PgRow) -> Result<OperationSampleMetadata, RegistryError> {
Ok(OperationSampleMetadata {
id: mcpaas_core::SampleId::new(row.try_get::<String, _>("id")?),
operation_id: OperationId::new(row.try_get::<String, _>("operation_id")?),
@@ -908,16 +942,14 @@ fn map_sample_metadata(
})
}
fn map_descriptor_metadata(
row: &sqlx::sqlite::SqliteRow,
) -> Result<DescriptorMetadata, RegistryError> {
fn map_descriptor_metadata(row: &PgRow) -> Result<DescriptorMetadata, RegistryError> {
Ok(DescriptorMetadata {
id: mcpaas_core::DescriptorId::new(row.try_get::<String, _>("id")?),
operation_id: row
.try_get::<Option<String>, _>("operation_id")?
.map(OperationId::new),
version: row
.try_get::<Option<i64>, _>("version")?
.try_get::<Option<i32>, _>("version")?
.map(|value| from_db_version(value, "version"))
.transpose()?,
descriptor_kind: deserialize_enum_text(
@@ -927,15 +959,13 @@ fn map_descriptor_metadata(
storage_ref: row.try_get("storage_ref")?,
source_name: row.try_get("source_name")?,
package_index: row
.try_get::<Option<String>, _>("package_index_json")?
.as_deref()
.map(deserialize_json)
.transpose()?,
.try_get::<Option<Json<Value>>, _>("package_index_json")?
.map(|value| value.0),
created_at: row.try_get("created_at")?,
})
}
fn map_yaml_import_job(row: &sqlx::sqlite::SqliteRow) -> Result<YamlImportJob, RegistryError> {
fn map_yaml_import_job(row: &PgRow) -> Result<YamlImportJob, RegistryError> {
Ok(YamlImportJob {
id: YamlImportJobId::new(row.try_get::<String, _>("id")?),
source_sample_id: row
@@ -948,7 +978,7 @@ fn map_yaml_import_job(row: &sqlx::sqlite::SqliteRow) -> Result<YamlImportJob, R
.try_get::<Option<String>, _>("result_operation_id")?
.map(OperationId::new),
result_version: row
.try_get::<Option<i64>, _>("result_version")?
.try_get::<Option<i32>, _>("result_version")?
.map(|value| from_db_version(value, "result_version"))
.transpose()?,
error_text: row.try_get("error_text")?,
@@ -957,16 +987,18 @@ fn map_yaml_import_job(row: &sqlx::sqlite::SqliteRow) -> Result<YamlImportJob, R
})
}
fn serialize_json<T: Serialize>(value: &T) -> Result<String, RegistryError> {
Ok(serde_json::to_string(value)?)
fn serialize_json_value<T: Serialize>(value: &T) -> Result<Value, RegistryError> {
Ok(serde_json::to_value(value)?)
}
fn serialize_option_json<T: Serialize>(value: &Option<T>) -> Result<Option<String>, RegistryError> {
value.as_ref().map(serialize_json).transpose()
fn serialize_option_json_value<T: Serialize>(
value: &Option<T>,
) -> Result<Option<Value>, RegistryError> {
value.as_ref().map(serialize_json_value).transpose()
}
fn deserialize_json<T: DeserializeOwned>(value: &str) -> Result<T, RegistryError> {
Ok(serde_json::from_str(value)?)
fn deserialize_json_value<T: DeserializeOwned>(value: Value) -> Result<T, RegistryError> {
Ok(serde_json::from_value(value)?)
}
fn serialize_enum_text<T: Serialize>(
@@ -987,17 +1019,24 @@ fn deserialize_enum_text<T: DeserializeOwned>(
.map_err(|_| RegistryError::InvalidEnumRepresentation { field })
}
fn to_db_version(value: u32) -> i64 {
i64::from(value)
fn to_db_version(value: u32) -> i32 {
value as i32
}
fn from_db_version(value: i64, field: &'static str) -> Result<u32, RegistryError> {
u32::try_from(value).map_err(|_| RegistryError::InvalidNumericValue { field, value })
fn from_db_version(value: i32, field: &'static str) -> Result<u32, RegistryError> {
u32::try_from(value).map_err(|_| RegistryError::InvalidNumericValue {
field,
value: i64::from(value),
})
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use std::{
collections::BTreeMap,
env,
time::{SystemTime, UNIX_EPOCH},
};
use mcpaas_core::{
ApiKeyHeaderAuthConfig, AuthConfig, AuthKind, AuthProfile, ConfigExport, ExecutionConfig,
@@ -1008,10 +1047,10 @@ mod tests {
use mcpaas_mapping::{MappingRule, MappingSet};
use mcpaas_schema::{Schema, SchemaKind};
use serde_json::json;
use sqlx::{Executor, PgPool, postgres::PgPoolOptions};
use crate::{
SqliteRegistry,
error::RegistryError,
PostgresRegistry, RegistryError,
model::{
CreateVersionRequest, CreateYamlImportJobRequest, DescriptorKind, DescriptorMetadata,
OperationSampleMetadata, PublishRequest, RegistryOperation, SampleKind,
@@ -1022,7 +1061,8 @@ mod tests {
#[tokio::test]
async fn stores_versions_and_published_operations() {
let registry = SqliteRegistry::connect("sqlite::memory:").await.unwrap();
let database = TestDatabase::new().await;
let registry = database.registry().await;
let operation_v1 = test_operation("op_rest_01", 1, OperationStatus::Draft);
registry
@@ -1076,11 +1116,14 @@ mod tests {
);
assert_eq!(published.version, 2);
assert!(published.is_published());
database.cleanup().await;
}
#[tokio::test]
async fn rejects_out_of_order_versions() {
let registry = SqliteRegistry::connect("sqlite::memory:").await.unwrap();
let database = TestDatabase::new().await;
let registry = database.registry().await;
let operation = test_operation("op_rest_02", 1, OperationStatus::Draft);
registry.create_operation(&operation, None).await.unwrap();
@@ -1103,11 +1146,14 @@ mod tests {
..
}
));
database.cleanup().await;
}
#[tokio::test]
async fn stores_auth_profiles_and_artifact_metadata() {
let registry = SqliteRegistry::connect("sqlite::memory:").await.unwrap();
let database = TestDatabase::new().await;
let registry = database.registry().await;
let operation = test_operation("op_rest_03", 1, OperationStatus::Draft);
registry.create_operation(&operation, None).await.unwrap();
@@ -1178,11 +1224,14 @@ mod tests {
assert_eq!(auth_profiles, vec![auth_profile]);
assert_eq!(samples, vec![input_sample]);
assert_eq!(descriptors, vec![descriptor]);
database.cleanup().await;
}
#[tokio::test]
async fn stores_and_finishes_yaml_import_jobs() {
let registry = SqliteRegistry::connect("sqlite::memory:").await.unwrap();
let database = TestDatabase::new().await;
let registry = database.registry().await;
let job_id = YamlImportJobId::new("job_yaml_01");
let operation = test_operation("op_rest_04", 1, OperationStatus::Draft);
@@ -1222,6 +1271,61 @@ mod tests {
assert_eq!(job.status, YamlImportJobStatus::Completed);
assert_eq!(job.result_version, Some(2));
assert_eq!(job.mode, ExportMode::Portable);
database.cleanup().await;
}
struct TestDatabase {
admin_pool: PgPool,
database_url: String,
schema: String,
}
impl TestDatabase {
async fn new() -> Self {
let database_url = env::var("TEST_DATABASE_URL")
.expect("TEST_DATABASE_URL must point to a reachable PostgreSQL database");
let admin_pool = PgPoolOptions::new()
.max_connections(1)
.connect(&database_url)
.await
.unwrap();
let schema = format!(
"test_registry_{}_{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
);
admin_pool
.execute(sqlx::query(&format!("create schema {schema}")))
.await
.unwrap();
Self {
admin_pool,
database_url,
schema,
}
}
async fn registry(&self) -> PostgresRegistry {
PostgresRegistry::connect_in_schema(&self.database_url, Some(&self.schema))
.await
.unwrap()
}
async fn cleanup(&self) {
self.admin_pool
.execute(sqlx::query(&format!(
"drop schema if exists {} cascade",
self.schema
)))
.await
.unwrap();
}
}
fn test_operation(id: &str, version: u32, status: OperationStatus) -> RegistryOperation {
@@ -1344,7 +1448,7 @@ mod tests {
export_mode: ExportMode::Portable,
}),
created_at: "2026-03-25T11:58:00Z".to_owned(),
updated_at: format!("2026-03-25T12:{:02}:00Z", version),
updated_at: format!("2026-03-25T12:{version:02}:00Z"),
published_at: None,
}
}