feat: implement sqlite registry storage

Cargo.toml

@@ -25,6 +25,7 @@ axum = "0.8"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 serde_yaml = "0.9"
+sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio-rustls", "sqlite"] }
 thiserror = "2"
 tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
 tracing = "0.1"

TASKS.md

@@ -2,19 +2,21 @@
 
 ## Current
 
-### `feat/mapping-engine`
+### `feat/registry-storage`
 
 Status: completed
 
 DoD:
 
-- JSONPath parsing works
-- input and output mapping work
-- draft generation from samples works
+- migrations create tables from `database-schema`
+- version snapshots persist and load correctly
+- publish linkage works
+- auth profiles and artifact metadata are stored
+- registry integration tests run on a real SQLite database
 
 ## Next
 
-- `feat/registry-storage`
+- `feat/rest-vertical-slice`
 
 ## Backlog
 

crates/mcpaas-registry/Cargo.toml

@@ -11,5 +11,8 @@ mcpaas-mapping = { path = "../mcpaas-mapping" }
 mcpaas-schema = { path = "../mcpaas-schema" }
 serde.workspace = true
 serde_json.workspace = true
+sqlx.workspace = true
 thiserror.workspace = true
 
+[dev-dependencies]
+tokio.workspace = true

crates/mcpaas-registry/src/error.rs

@@ -0,0 +1,36 @@
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum RegistryError {
+    #[error(transparent)]
+    Storage(#[from] sqlx::Error),
+    #[error(transparent)]
+    Serialization(#[from] serde_json::Error),
+    #[error("operation {operation_id} already exists")]
+    OperationAlreadyExists { operation_id: String },
+    #[error("operation {operation_id} was not found")]
+    OperationNotFound { operation_id: String },
+    #[error("operation version {version} for {operation_id} was not found")]
+    OperationVersionNotFound { operation_id: String, version: u32 },
+    #[error("operation {operation_id} must start with version 1, got {version}")]
+    InvalidInitialVersion { operation_id: String, version: u32 },
+    #[error("operation {operation_id} expected next version {expected}, got {actual}")]
+    InvalidVersionSequence {
+        operation_id: String,
+        expected: u32,
+        actual: u32,
+    },
+    #[error("operation {operation_id} changed immutable field {field}")]
+    ImmutableOperationFieldChanged {
+        operation_id: String,
+        field: &'static str,
+    },
+    #[error("auth profile {auth_profile_id} was not found")]
+    AuthProfileNotFound { auth_profile_id: String },
+    #[error("yaml import job {job_id} was not found")]
+    YamlImportJobNotFound { job_id: String },
+    #[error("unsupported enum representation for field {field}")]
+    InvalidEnumRepresentation { field: &'static str },
+    #[error("invalid numeric value for field {field}: {value}")]
+    InvalidNumericValue { field: &'static str, value: i64 },
+}

crates/mcpaas-registry/src/lib.rs

@@ -1,3 +1,14 @@
-pub fn crate_name() -> &'static str {
-    "mcpaas-registry"
-}
+mod error;
+mod migrations;
+mod model;
+mod sqlite;
+
+pub use error::RegistryError;
+pub use model::{
+    CreateVersionRequest, CreateYamlImportJobRequest, DescriptorKind, DescriptorMetadata,
+    OperationSampleMetadata, OperationSummary, OperationVersionRecord, PublishRequest,
+    RegistryOperation, SampleKind, SaveAuthProfileRequest, SaveDescriptorMetadataRequest,
+    SaveSampleMetadataRequest, YamlImportJob, YamlImportJobCompletion, YamlImportJobId,
+    YamlImportJobStatus,
+};
+pub use sqlite::SqliteRegistry;

crates/mcpaas-registry/src/migrations.rs

@@ -0,0 +1,126 @@
+use sqlx::{SqlitePool, query};
+
+pub async fn apply_sqlite(pool: &SqlitePool) -> Result<(), sqlx::Error> {
+    query(
+        "create table if not exists operations (
+            id text primary key,
+            name text not null unique,
+            display_name text not null,
+            protocol text not null,
+            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
+        )",
+    )
+    .execute(pool)
+    .await?;
+
+    query(
+        "create table if not exists operation_versions (
+            operation_id text not null,
+            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,
+            change_note text null,
+            created_at text not null,
+            created_by text null,
+            primary key (operation_id, version),
+            foreign key (operation_id) references operations(id) on delete cascade
+        )",
+    )
+    .execute(pool)
+    .await?;
+
+    query(
+        "create table if not exists published_operations (
+            operation_id text primary key,
+            version integer not null,
+            published_at text 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
+        )",
+    )
+    .execute(pool)
+    .await?;
+
+    query(
+        "create table if not exists operation_samples (
+            id text primary key,
+            operation_id text not null,
+            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,
+            foreign key (operation_id, version) references operation_versions(operation_id, version) on delete cascade
+        )",
+    )
+    .execute(pool)
+    .await?;
+
+    query(
+        "create table if not exists descriptors (
+            id text primary key,
+            operation_id text null,
+            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,
+            foreign key (operation_id, version) references operation_versions(operation_id, version) on delete cascade
+        )",
+    )
+    .execute(pool)
+    .await?;
+
+    query(
+        "create table if not exists auth_profiles (
+            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
+        )",
+    )
+    .execute(pool)
+    .await?;
+
+    query(
+        "create table if not exists yaml_import_jobs (
+            id text primary key,
+            source_sample_id text null,
+            status text not null,
+            format_version text not null,
+            mode text not null,
+            result_operation_id text 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
+        )",
+    )
+    .execute(pool)
+    .await?;
+
+    Ok(())
+}

crates/mcpaas-registry/src/model.rs

@@ -0,0 +1,176 @@
+use mcpaas_core::{
+    AuthProfile, DescriptorId, ExportMode, Operation, OperationId, OperationStatus, Protocol,
+    SampleId,
+};
+use mcpaas_mapping::MappingSet;
+use mcpaas_schema::Schema;
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+macro_rules! define_registry_id {
+    ($name:ident) => {
+        #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
+        pub struct $name(String);
+
+        impl $name {
+            pub fn new(value: impl Into<String>) -> Self {
+                Self(value.into())
+            }
+
+            pub fn as_str(&self) -> &str {
+                &self.0
+            }
+        }
+
+        impl From<String> for $name {
+            fn from(value: String) -> Self {
+                Self(value)
+            }
+        }
+
+        impl From<&str> for $name {
+            fn from(value: &str) -> Self {
+                Self(value.to_owned())
+            }
+        }
+    };
+}
+
+define_registry_id!(YamlImportJobId);
+
+pub type RegistryOperation = Operation<Schema, MappingSet>;
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OperationSummary {
+    pub id: OperationId,
+    pub name: String,
+    pub display_name: String,
+    pub protocol: Protocol,
+    pub status: OperationStatus,
+    pub current_draft_version: u32,
+    pub latest_published_version: Option<u32>,
+    pub created_at: String,
+    pub updated_at: String,
+    pub published_at: Option<String>,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct OperationVersionRecord {
+    pub operation_id: OperationId,
+    pub version: u32,
+    pub status: OperationStatus,
+    pub change_note: Option<String>,
+    pub created_at: String,
+    pub created_by: Option<String>,
+    pub snapshot: RegistryOperation,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum SampleKind {
+    InputJson,
+    OutputJson,
+    YamlImportSource,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OperationSampleMetadata {
+    pub id: SampleId,
+    pub operation_id: OperationId,
+    pub version: u32,
+    pub sample_kind: SampleKind,
+    pub storage_ref: String,
+    pub content_type: String,
+    pub file_name: Option<String>,
+    pub created_at: String,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum DescriptorKind {
+    ProtoUpload,
+    DescriptorSet,
+    ReflectionSnapshot,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct DescriptorMetadata {
+    pub id: DescriptorId,
+    pub operation_id: Option<OperationId>,
+    pub version: Option<u32>,
+    pub descriptor_kind: DescriptorKind,
+    pub storage_ref: String,
+    pub source_name: Option<String>,
+    pub package_index: Option<Value>,
+    pub created_at: String,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum YamlImportJobStatus {
+    Pending,
+    Completed,
+    Failed,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct YamlImportJob {
+    pub id: YamlImportJobId,
+    pub source_sample_id: Option<SampleId>,
+    pub status: YamlImportJobStatus,
+    pub format_version: String,
+    pub mode: ExportMode,
+    pub result_operation_id: Option<OperationId>,
+    pub result_version: Option<u32>,
+    pub error_text: Option<String>,
+    pub created_at: String,
+    pub finished_at: Option<String>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct YamlImportJobCompletion {
+    pub status: YamlImportJobStatus,
+    pub result_operation_id: Option<OperationId>,
+    pub result_version: Option<u32>,
+    pub error_text: Option<String>,
+    pub finished_at: String,
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct CreateVersionRequest<'a> {
+    pub snapshot: &'a RegistryOperation,
+    pub change_note: Option<&'a str>,
+    pub created_by: Option<&'a str>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct PublishRequest<'a> {
+    pub operation_id: &'a OperationId,
+    pub version: u32,
+    pub published_at: &'a str,
+    pub published_by: Option<&'a str>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct CreateYamlImportJobRequest<'a> {
+    pub id: &'a YamlImportJobId,
+    pub source_sample_id: Option<&'a SampleId>,
+    pub format_version: &'a str,
+    pub mode: ExportMode,
+    pub created_at: &'a str,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct SaveAuthProfileRequest<'a> {
+    pub profile: &'a AuthProfile,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct SaveSampleMetadataRequest<'a> {
+    pub sample: &'a OperationSampleMetadata,
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct SaveDescriptorMetadataRequest<'a> {
+    pub descriptor: &'a DescriptorMetadata,
+}

docs/database-schema.md

@@ -12,6 +12,8 @@
 
 Конфигурация operation не должна храниться только в одной "живой" записи. Каждое существенное изменение должно приводить к появлению новой версии конфигурации.
 
+Для MVP в registry version snapshot хранит protocol-specific конфигурацию, схемы, mapping и execution settings. Поля identity и listing view (`name`, `display_name`, `protocol`) считаются стабильными и хранятся в `operations`.
+
 ### 2.2. Published и draft разделяются логически
 
 - `draft` может меняться;
@@ -28,6 +30,14 @@
 
 Для MVP рекомендуется отдельная таблица `auth_profiles`, где metadata и secret refs отделены от operation versions.
 
+### 2.5. SQLite-адаптация допустима
+
+Канонической схемой остается PostgreSQL-совместимая модель, но MVP-реализация registry может использовать SQLite. В таком случае:
+
+- `jsonb` хранится как `text` с JSON-сериализацией;
+- `timestamptz` хранится как `text`;
+- внешние ключи и version snapshot semantics сохраняются без изменения.
+
 ## 3. Основные таблицы
 
 Минимальный набор таблиц: