admin/CC-Switch
1
0
Fork 0
mirror of https://github.com/farion1231/cc-switch.git synced 2026-08-04 19:45:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

storage(pi): isolate typed provider writes and canonical restore

Browse Source 
Replace the generic Provider save/upsert surface with strict typed
create/update/rename/endpoint operations. Keep aggregate hydration read-only,
preserve nullable endpoint timestamps end to end, and make service create own
the initial endpoint set atomically.

Restore SQL and binary backups only through UntrustedScratch, migrate and copy
fixed data columns into a fresh CanonicalStage, validate the canonical result,
then publish through the Backup API. The imported schema is never eligible to
become the live schema.

Old save_provider callsite classification
==========================================

Inventory authority: abandoned 5a385fc8 tree. The old definition at
src-tauri/src/database/dao/providers.rs:180 is deleted and is not a callsite.

Production callsites:

- src-tauri/src/commands/provider.rs:253 [create] Claude Desktop import creates
  one absent aggregate; it now strict-inserts the row and initial endpoints in
  one transaction.
- src-tauri/src/database/dao/providers.rs:638 [create] official seed first
  proves absence, then strict-creates; a racing insert is a conflict.
- src-tauri/src/database/dao/providers.rs:704 [create] on-demand seed first
  proves absence, then strict-creates; it cannot overwrite an existing row.
- src-tauri/src/services/omo.rs:291 [create] OMO import constructs a new
  aggregate and strict-creates it; OMO is not eligible for rename.
- src-tauri/src/services/provider/endpoints.rs:85 [update] endpoint last-used
  is not a Provider-row save; it now calls the exact touch endpoint operation.
- src-tauri/src/services/provider/live.rs:1567 [create/update] default live
  import is reconciliation: read first, then strict create or strict update.
- src-tauri/src/services/provider/live.rs:1743 [update] an existing OpenCode
  live provider follows the strict row-update branch.
- src-tauri/src/services/provider/live.rs:1770 [create] a new OpenCode live
  provider follows the strict aggregate-create branch.
- src-tauri/src/services/provider/live.rs:1825 [update] an existing OpenClaw
  live provider follows the strict row-update branch.
- src-tauri/src/services/provider/live.rs:1858 [create] a new OpenClaw live
  provider follows the strict aggregate-create branch.
- src-tauri/src/services/provider/live.rs:1900 [update] an existing Hermes live
  provider follows the strict row-update branch.
- src-tauri/src/services/provider/live.rs:1926 [create] a new Hermes live
  provider follows the strict aggregate-create branch.
- src-tauri/src/services/provider/mod.rs:2568 [create] ProviderService::add owns
  strict aggregate creation and all initial endpoints.
- src-tauri/src/services/provider/mod.rs:2680 [rename] an additive DB-only key
  change now uses the dedicated transactional rename after eligibility checks.
- src-tauri/src/services/provider/mod.rs:2711 [update] OMO edit updates exactly
  the existing main row after its live-file coordination.
- src-tauri/src/services/provider/mod.rs:2740 [update] additive-provider edit
  updates exactly the existing main row after resolving live ownership.
- src-tauri/src/services/provider/mod.rs:2750 [update] switch-mode edit updates
  exactly the existing main row and never inserts.
- src-tauri/src/services/provider/mod.rs:2948 [update] remove-from-live changes
  only the existing provider's live-managed marker.
- src-tauri/src/services/provider/mod.rs:3120 [update] switch backfill updates
  only the existing current provider row.
- src-tauri/src/services/provider/mod.rs:3174 [update] successful additive
  switch changes only the existing live-managed marker.
- src-tauri/src/services/provider/mod.rs:3315 [update] common-config migration
  updates only each already-read existing row.
- src-tauri/src/services/provider/mod.rs:3895 [update] Gemini credential scrub
  updates only each already-read existing row.
- src-tauri/src/services/provider/mod.rs:4082 [update] sort ordering is routed
  to the dedicated sort-index state operation, not row replacement.
- src-tauri/src/services/provider/mod.rs:4636 [create/update] universal-to-
  Claude reconciliation reads the target and selects strict create or update.
- src-tauri/src/services/provider/mod.rs:4651 [create/update] universal-to-
  Codex reconciliation reads the target and selects strict create or update.
- src-tauri/src/services/provider/mod.rs:4665 [create/update] universal-to-
  Gemini reconciliation reads the target and selects strict create or update.

Required indirect ownership paths:

- src-tauri/src/deeplink/provider.rs [create] the old indirect flow called
  ProviderService::add and then appended endpoints one by one. It now supplies
  every non-primary endpoint to one strict aggregate create, so hydration is
  complete atomically and a duplicate is zero-side-effect.
- [restore] no old generic-save callsite is reclassified as restore. Exact
  aggregate replacement exists only as the sealed
  restore_provider_aggregate_on_tx compensation primitive.

Test-only callsites:

Every item below is classified [test]. Each is fixture setup, not a production
write authority, and is migrated to a real ProviderService entry where the
behavior is under test or to the cfg(test)-only typed fixture reconciler where
the test merely needs pre-existing rows.

- src-tauri/src/codex_history_migration.rs:1442 [test] migration fixture setup.
- src-tauri/src/codex_history_migration.rs:1452 [test] migration fixture setup.
- src-tauri/src/codex_history_migration.rs:2174 [test] migration fixture setup.
- src-tauri/src/codex_history_migration.rs:2176 [test] migration fixture setup.
- src-tauri/src/codex_history_migration.rs:2199 [test] migration fixture setup.
- src-tauri/src/codex_history_migration.rs:2219 [test] migration fixture setup.
- src-tauri/src/codex_history_migration.rs:2247 [test] migration fixture setup.
- src-tauri/src/codex_history_migration.rs:2267 [test] migration fixture setup.
- src-tauri/src/codex_history_migration.rs:2288 [test] migration fixture setup.
- src-tauri/src/codex_history_migration.rs:2320 [test] migration fixture setup.
- src-tauri/src/codex_history_migration.rs:2393 [test] migration fixture setup.
- src-tauri/src/codex_history_migration.rs:2449 [test] migration fixture setup.
- src-tauri/src/codex_history_migration.rs:2498 [test] migration fixture setup.
- src-tauri/src/codex_history_migration.rs:2555 [test] migration fixture setup.
- src-tauri/src/codex_history_migration.rs:2604 [test] migration fixture setup.
- src-tauri/src/codex_history_migration.rs:2625 [test] migration fixture setup.
- src-tauri/src/database/dao/providers.rs:754 [test] DAO fixture setup.
- src-tauri/src/proxy/provider_router.rs:351 [test] router fixture setup.
- src-tauri/src/proxy/provider_router.rs:352 [test] router fixture setup.
- src-tauri/src/proxy/provider_router.rs:377 [test] router fixture setup.
- src-tauri/src/proxy/provider_router.rs:378 [test] router fixture setup.
- src-tauri/src/proxy/provider_router.rs:410 [test] router fixture setup.
- src-tauri/src/proxy/provider_router.rs:411 [test] router fixture setup.
- src-tauri/src/proxy/provider_router.rs:447 [test] router fixture setup.
- src-tauri/src/proxy/provider_router.rs:448 [test] router fixture setup.
- src-tauri/src/proxy/provider_router.rs:488 [test] router fixture setup.
- src-tauri/src/services/provider/mod.rs:485 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:586 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:813 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:825 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:1472 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:1607 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:1737 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:1945 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:1952 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:1978 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:2006 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:2056 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:2130 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:2167 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:2207 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:2235 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:2270 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:2320 [test] service fixture setup.
- src-tauri/src/services/provider/mod.rs:2362 [test] service fixture setup.
- src-tauri/src/services/proxy.rs:3762 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:3948 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:4034 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:4095 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:4114 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:4263 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:4341 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:4421 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:4533 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:4651 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:4787 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:5264 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:5320 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:5385 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:5387 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:5460 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:5462 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:5611 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:5613 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:5615 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:5698 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:5700 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:5998 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:6000 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:6173 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:6175 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:6417 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:6419 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:6553 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:6555 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:6635 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:6637 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:6919 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:7173 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:7175 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:7240 [test] proxy fixture setup.
- src-tauri/src/services/proxy.rs:7242 [test] proxy fixture setup.
- src-tauri/tests/profile_roundtrip.rs:112 [test] profile fixture create.
- src-tauri/tests/profile_roundtrip.rs:116 [test] profile fixture create.
- src-tauri/tests/profile_roundtrip.rs:126 [test] profile fixture create.
- src-tauri/tests/profile_roundtrip.rs:133 [test] profile fixture create.
- src-tauri/tests/profile_roundtrip.rs:292 [test] profile fixture create.
- src-tauri/tests/profile_roundtrip.rs:501 [test] profile fixture create.
- src-tauri/tests/profile_roundtrip.rs:505 [test] profile fixture create.
- src-tauri/tests/profile_roundtrip.rs:670 [test] profile fixture create.
- src-tauri/tests/profile_roundtrip.rs:677 [test] profile fixture create.
- src-tauri/tests/profile_roundtrip.rs:762 [test] Linux Desktop fixture create.
- src-tauri/tests/profile_roundtrip.rs:769 [test] Linux Desktop fixture create.
- src-tauri/tests/provider_commands.rs:69 [test] command fixture create.
- src-tauri/tests/provider_service.rs:2927 [test] service fixture create.
This commit is contained in:
SaladDay
2026-07-31 17:58:07 +00:00
parent 3dc188313e
commit 4f78451405
34 changed files with 6210 additions and 831 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/database/backup.rs
+2493 -149
View File
File diff suppressed because it is too large Load Diff
src-tauri/src/database/dao/mod.rs
+3
View File
@@ -4,12 +4,15 @@
pub mod failover;
pub mod mcp;
pub mod pi_projections;
pub mod profiles;
pub mod prompts;
pub mod provider_write;
pub mod providers;
pub mod providers_seed;
pub mod proxy;
pub mod settings;
pub mod skill_deployments;
pub mod skills;
pub mod stream_check;
pub mod universal_providers;
src-tauri/src/database/dao/pi_projections.rs
+204
View File
@@ -0,0 +1,204 @@
//! Device-local ownership ledger for exact keys in Pi's shared models.json.
// The projection writer is introduced in a later contract-ordered commit.
#![allow(dead_code)]
use crate::database::{lock_conn, Database};
use crate::error::AppError;
use indexmap::IndexMap;
use rusqlite::{params, OptionalExtension};
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct PiProviderProjection {
pub provider_id: String,
pub provider_key: String,
pub created_at: i64,
pub updated_at: i64,
}
fn decode_projection(row: &rusqlite::Row<'_>) -> rusqlite::Result<PiProviderProjection> {
Ok(PiProviderProjection {
provider_id: row.get(0)?,
provider_key: row.get(1)?,
created_at: row.get(2)?,
updated_at: row.get(3)?,
})
}
impl Database {
pub(crate) fn get_pi_projection(
&self,
provider_id: &str,
) -> Result<Option<PiProviderProjection>, AppError> {
let conn = lock_conn!(self.conn);
conn.query_row(
"SELECT provider_id, provider_key, created_at, updated_at
FROM pi_provider_projections WHERE provider_id = ?1",
[provider_id],
decode_projection,
)
.optional()
.map_err(|error| AppError::Database(error.to_string()))
}
pub(crate) fn get_pi_projection_for_key(
&self,
provider_key: &str,
) -> Result<Option<PiProviderProjection>, AppError> {
let conn = lock_conn!(self.conn);
conn.query_row(
"SELECT provider_id, provider_key, created_at, updated_at
FROM pi_provider_projections WHERE provider_key = ?1",
[provider_key],
decode_projection,
)
.optional()
.map_err(|error| AppError::Database(error.to_string()))
}
pub(crate) fn get_pi_projection_manifest(
&self,
) -> Result<IndexMap<String, PiProviderProjection>, AppError> {
let conn = lock_conn!(self.conn);
let mut stmt = conn
.prepare(
"SELECT provider_id, provider_key, created_at, updated_at
FROM pi_provider_projections ORDER BY provider_id",
)
.map_err(|error| AppError::Database(error.to_string()))?;
let rows = stmt
.query_map([], decode_projection)
.map_err(|error| AppError::Database(error.to_string()))?;
let mut manifest = IndexMap::new();
for row in rows {
let projection = row.map_err(|error| AppError::Database(error.to_string()))?;
manifest.insert(projection.provider_id.clone(), projection);
}
Ok(manifest)
}
/// Claim an exact key. Existing exact claims are idempotent; either-side
/// collisions fail and are never rewritten.
pub(crate) fn claim_pi_projection_key(
&self,
provider_id: &str,
provider_key: &str,
) -> Result<PiProviderProjection, AppError> {
if provider_id.trim().is_empty() || provider_key.trim().is_empty() {
return Err(AppError::Config(
"Pi projection provider id and key must be non-empty".to_string(),
));
}
let mut conn = lock_conn!(self.conn);
let tx = conn
.transaction()
.map_err(|error| AppError::Database(error.to_string()))?;
let by_provider = tx
.query_row(
"SELECT provider_id, provider_key, created_at, updated_at
FROM pi_provider_projections WHERE provider_id = ?1",
[provider_id],
decode_projection,
)
.optional()
.map_err(|error| AppError::Database(error.to_string()))?;
if let Some(existing) = by_provider {
if existing.provider_key != provider_key {
return Err(AppError::Config(format!(
"Pi provider '{provider_id}' already owns key '{}', not '{provider_key}'",
existing.provider_key
)));
}
tx.commit()
.map_err(|error| AppError::Database(error.to_string()))?;
return Ok(existing);
}
if let Some(existing_owner) = tx
.query_row(
"SELECT provider_id FROM pi_provider_projections WHERE provider_key = ?1",
[provider_key],
|row| row.get::<_, String>(0),
)
.optional()
.map_err(|error| AppError::Database(error.to_string()))?
{
return Err(AppError::Config(format!(
"Pi key '{provider_key}' is already owned by provider '{existing_owner}'"
)));
}
let now = chrono::Utc::now().timestamp_millis();
tx.execute(
"INSERT INTO pi_provider_projections
(provider_id, provider_key, created_at, updated_at)
VALUES (?1, ?2, ?3, ?3)",
params![provider_id, provider_key, now],
)
.map_err(|error| AppError::Database(error.to_string()))?;
tx.commit()
.map_err(|error| AppError::Database(error.to_string()))?;
Ok(PiProviderProjection {
provider_id: provider_id.to_string(),
provider_key: provider_key.to_string(),
created_at: now,
updated_at: now,
})
}
pub(crate) fn delete_pi_projection_key(
&self,
provider_id: &str,
expected_key: &str,
) -> Result<bool, AppError> {
let conn = lock_conn!(self.conn);
let removed = conn
.execute(
"DELETE FROM pi_provider_projections
WHERE provider_id = ?1 AND provider_key = ?2",
params![provider_id, expected_key],
)
.map_err(|error| AppError::Database(error.to_string()))?;
if removed == 0
&& conn
.query_row(
"SELECT 1 FROM pi_provider_projections WHERE provider_id = ?1",
[provider_id],
|_| Ok(()),
)
.optional()
.map_err(|error| AppError::Database(error.to_string()))?
.is_some()
{
return Err(AppError::Config(format!(
"refusing to delete Pi projection '{provider_id}': expected key changed"
)));
}
Ok(removed == 1)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn projection_claims_are_exact_idempotent_and_collision_safe() -> Result<(), AppError> {
let db = Database::memory()?;
let first = db.claim_pi_projection_key("provider-a", "native-a")?;
let repeated = db.claim_pi_projection_key("provider-a", "native-a")?;
assert_eq!(first, repeated);
assert!(db
.claim_pi_projection_key("provider-a", "native-b")
.is_err());
assert!(db
.claim_pi_projection_key("provider-b", "native-a")
.is_err());
assert_eq!(db.get_pi_projection_manifest()?.len(), 1);
assert!(db.delete_pi_projection_key("provider-a", "wrong").is_err());
assert!(db.get_pi_projection("provider-a")?.is_some());
assert!(db.delete_pi_projection_key("provider-a", "native-a")?);
assert!(db.get_pi_projection_for_key("native-a")?.is_none());
Ok(())
}
}
src-tauri/src/database/dao/provider_write.rs
+531
View File
@@ -0,0 +1,531 @@
use crate::database::{lock_conn, Database};
use crate::error::AppError;
use crate::provider::{ProviderMeta, ProviderMutationInput};
use crate::settings::CustomEndpoint;
use rusqlite::{params, OptionalExtension, Transaction};
use serde_json::Value;
use std::collections::HashSet;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProviderKey {
app_type: String,
id: String,
}
impl ProviderKey {
pub fn new(app_type: impl Into<String>, id: impl Into<String>) -> Result<Self, AppError> {
let app_type = app_type.into();
let id = id.into();
if app_type.trim().is_empty() || id.trim().is_empty() {
return Err(AppError::InvalidInput(
"provider app type and id must be non-empty".to_string(),
));
}
Ok(Self { app_type, id })
}
pub fn app_type(&self) -> &str {
&self.app_type
}
pub fn id(&self) -> &str {
&self.id
}
}
#[derive(Debug, Clone)]
pub struct ProviderRowUpdate {
name: String,
settings_config: Value,
website_url: Option<String>,
category: Option<String>,
created_at: Option<i64>,
notes: Option<String>,
meta: ProviderMeta,
icon: Option<String>,
icon_color: Option<String>,
}
impl ProviderRowUpdate {
pub fn from_input(input: &ProviderMutationInput) -> Result<Self, AppError> {
let meta = input.meta.clone().unwrap_or_default();
if !meta.custom_endpoints.is_empty() {
return Err(AppError::InvalidInput(
"provider update must not contain customEndpoints; use endpoint operations"
.to_string(),
));
}
Ok(Self {
name: input.name.clone(),
settings_config: input.settings_config.clone(),
website_url: input.website_url.clone(),
category: input.category.clone(),
created_at: input.created_at,
notes: input.notes.clone(),
meta,
icon: input.icon.clone(),
icon_color: input.icon_color.clone(),
})
}
}
#[derive(Debug, Clone)]
pub struct NewEndpoint {
url: String,
added_at: Option<i64>,
last_used: Option<i64>,
}
impl NewEndpoint {
pub fn new(
url: impl Into<String>,
added_at: Option<i64>,
last_used: Option<i64>,
) -> Result<Self, AppError> {
let url = url.into();
if url.trim().is_empty() {
return Err(AppError::InvalidInput(
"provider endpoint URL cannot be empty".to_string(),
));
}
Ok(Self {
url,
added_at,
last_used,
})
}
pub fn now(url: impl Into<String>) -> Result<Self, AppError> {
Self::new(url, Some(chrono::Utc::now().timestamp_millis()), None)
}
}
impl TryFrom<CustomEndpoint> for NewEndpoint {
type Error = AppError;
fn try_from(endpoint: CustomEndpoint) -> Result<Self, Self::Error> {
Self::new(endpoint.url, endpoint.added_at, endpoint.last_used)
}
}
#[derive(Debug, Clone)]
pub struct NewProviderAggregate {
key: ProviderKey,
row: ProviderRowUpdate,
sort_index: Option<usize>,
in_failover_queue: bool,
initial_endpoints: Vec<NewEndpoint>,
}
impl NewProviderAggregate {
pub fn from_input(app_type: &str, mut input: ProviderMutationInput) -> Result<Self, AppError> {
let endpoints = input
.meta
.as_mut()
.map(|meta| std::mem::take(&mut meta.custom_endpoints))
.unwrap_or_default();
let mut seen = HashSet::with_capacity(endpoints.len());
let mut initial_endpoints = Vec::with_capacity(endpoints.len());
for (key, endpoint) in endpoints {
if key != endpoint.url {
return Err(AppError::InvalidInput(format!(
"provider endpoint key '{key}' must match endpoint URL '{}'",
endpoint.url
)));
}
if !seen.insert(endpoint.url.clone()) {
return Err(AppError::InvalidInput(format!(
"duplicate initial provider endpoint '{}'",
endpoint.url
)));
}
initial_endpoints.push(endpoint.try_into()?);
}
let key = ProviderKey::new(app_type, input.id.clone())?;
let row = ProviderRowUpdate::from_input(&input)?;
Ok(Self {
key,
row,
sort_index: input.sort_index,
in_failover_queue: input.in_failover_queue,
initial_endpoints,
})
}
}
#[derive(Debug, Clone)]
pub struct RenameProvider {
source: ProviderKey,
target_id: String,
row: ProviderRowUpdate,
}
impl RenameProvider {
pub fn from_input(
source: ProviderKey,
input: &ProviderMutationInput,
) -> Result<Self, AppError> {
if !matches!(source.app_type(), "opencode" | "openclaw") {
return Err(AppError::InvalidInput(
"provider key changes are restricted to additive OpenCode/OpenClaw providers"
.to_string(),
));
}
if source.id() == input.id {
return Err(AppError::InvalidInput(
"provider rename requires a different target id".to_string(),
));
}
if input.id.trim().is_empty() {
return Err(AppError::InvalidInput(
"provider target id must be non-empty".to_string(),
));
}
Ok(Self {
source,
target_id: input.id.clone(),
row: ProviderRowUpdate::from_input(input)?,
})
}
}
fn encode_row(row: &ProviderRowUpdate) -> Result<(String, String), AppError> {
let settings_config = serde_json::to_string(&row.settings_config).map_err(|error| {
AppError::Database(format!("failed to serialize settings_config: {error}"))
})?;
let meta = serde_json::to_string(&row.meta).map_err(|error| {
AppError::Database(format!("failed to serialize provider meta: {error}"))
})?;
Ok((settings_config, meta))
}
fn insert_row(
tx: &Transaction<'_>,
key: &ProviderKey,
row: &ProviderRowUpdate,
sort_index: Option<usize>,
is_current: bool,
in_failover_queue: bool,
) -> Result<(), AppError> {
let (settings_config, meta) = encode_row(row)?;
tx.execute(
"INSERT INTO providers (
id, app_type, name, settings_config, website_url, category,
created_at, sort_index, notes, icon, icon_color, meta,
is_current, in_failover_queue
) VALUES (
?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14
)",
params![
key.id,
key.app_type,
row.name,
settings_config,
row.website_url,
row.category,
row.created_at,
sort_index,
row.notes,
row.icon,
row.icon_color,
meta,
is_current,
in_failover_queue,
],
)
.map_err(|error| AppError::Database(error.to_string()))?;
Ok(())
}
fn insert_endpoint(
tx: &Transaction<'_>,
key: &ProviderKey,
endpoint: &NewEndpoint,
) -> Result<(), AppError> {
tx.execute(
"INSERT INTO provider_endpoints
(provider_id, app_type, url, added_at, last_used)
VALUES (?1, ?2, ?3, ?4, ?5)",
params![
key.id,
key.app_type,
endpoint.url,
endpoint.added_at,
endpoint.last_used
],
)
.map_err(|error| AppError::Database(error.to_string()))?;
Ok(())
}
/// Exact aggregate replacement is sealed inside the DAO parent module. The
/// catalog compensation coordinator introduced with the ordered mutation
/// pipeline is the only intended caller.
#[allow(dead_code)]
pub(super) fn restore_provider_aggregate_on_tx(
tx: &Transaction<'_>,
key: &ProviderKey,
row: &ProviderRowUpdate,
sort_index: Option<usize>,
is_current: bool,
in_failover_queue: bool,
endpoints: &[NewEndpoint],
) -> Result<(), AppError> {
let updated = update_row(tx, key, row)?;
if updated != 1 {
return Err(AppError::NotFound(format!(
"provider '{}/{}'",
key.app_type, key.id
)));
}
tx.execute(
"DELETE FROM provider_endpoints WHERE provider_id = ?1 AND app_type = ?2",
params![key.id, key.app_type],
)
.map_err(|error| AppError::Database(error.to_string()))?;
for endpoint in endpoints {
insert_endpoint(tx, key, endpoint)?;
}
// State and order are maintained by their dedicated authorities. Exact
// compensation may restore their captured values without exposing them in
// ProviderRowUpdate.
tx.execute(
"UPDATE providers
SET sort_index = ?1, is_current = ?2, in_failover_queue = ?3
WHERE id = ?4 AND app_type = ?5",
params![
sort_index,
is_current,
in_failover_queue,
key.id,
key.app_type
],
)
.map_err(|error| AppError::Database(error.to_string()))?;
Ok(())
}
fn update_row(
tx: &Transaction<'_>,
key: &ProviderKey,
row: &ProviderRowUpdate,
) -> Result<usize, AppError> {
let (settings_config, meta) = encode_row(row)?;
tx.execute(
"UPDATE providers SET
name = ?1,
settings_config = ?2,
website_url = ?3,
category = ?4,
created_at = ?5,
notes = ?6,
icon = ?7,
icon_color = ?8,
meta = ?9
WHERE id = ?10 AND app_type = ?11",
params![
row.name,
settings_config,
row.website_url,
row.category,
row.created_at,
row.notes,
row.icon,
row.icon_color,
meta,
key.id,
key.app_type,
],
)
.map_err(|error| AppError::Database(error.to_string()))
}
impl Database {
pub fn create_provider(&self, input: NewProviderAggregate) -> Result<(), AppError> {
let mut conn = lock_conn!(self.conn);
let tx = conn
.transaction()
.map_err(|error| AppError::Database(error.to_string()))?;
insert_row(
&tx,
&input.key,
&input.row,
input.sort_index,
false,
input.in_failover_queue,
)?;
for endpoint in &input.initial_endpoints {
insert_endpoint(&tx, &input.key, endpoint)?;
}
tx.commit()
.map_err(|error| AppError::Database(error.to_string()))
}
pub fn update_provider(
&self,
key: &ProviderKey,
row: &ProviderRowUpdate,
) -> Result<(), AppError> {
let mut conn = lock_conn!(self.conn);
let tx = conn
.transaction()
.map_err(|error| AppError::Database(error.to_string()))?;
if update_row(&tx, key, row)? != 1 {
return Err(AppError::NotFound(format!(
"provider '{}/{}'",
key.app_type, key.id
)));
}
tx.commit()
.map_err(|error| AppError::Database(error.to_string()))
}
pub fn rename_db_only_additive_provider(&self, input: RenameProvider) -> Result<(), AppError> {
let mut conn = lock_conn!(self.conn);
let tx = conn
.transaction()
.map_err(|error| AppError::Database(error.to_string()))?;
let source_state = tx
.query_row(
"SELECT sort_index, is_current, in_failover_queue, category
FROM providers
WHERE id = ?1 AND app_type = ?2",
params![input.source.id, input.source.app_type],
|row| {
Ok((
row.get::<_, Option<usize>>(0)?,
row.get::<_, bool>(1)?,
row.get::<_, bool>(2)?,
row.get::<_, Option<String>>(3)?,
))
},
)
.optional()
.map_err(|error| AppError::Database(error.to_string()))?
.ok_or_else(|| {
AppError::NotFound(format!(
"provider '{}/{}'",
input.source.app_type, input.source.id
))
})?;
if matches!(source_state.3.as_deref(), Some("omo" | "omo-slim")) {
return Err(AppError::InvalidInput(
"OMO/OMO Slim providers cannot be renamed".to_string(),
));
}
let target = ProviderKey::new(&input.source.app_type, &input.target_id)?;
insert_row(
&tx,
&target,
&input.row,
source_state.0,
source_state.1,
source_state.2,
)?;
tx.execute(
"INSERT INTO provider_endpoints
(provider_id, app_type, url, added_at, last_used)
SELECT ?1, app_type, url, added_at, last_used
FROM provider_endpoints
WHERE provider_id = ?2 AND app_type = ?3
ORDER BY id",
params![target.id, input.source.id, input.source.app_type],
)
.map_err(|error| AppError::Database(error.to_string()))?;
if tx
.execute(
"DELETE FROM providers WHERE id = ?1 AND app_type = ?2",
params![input.source.id, input.source.app_type],
)
.map_err(|error| AppError::Database(error.to_string()))?
!= 1
{
return Err(AppError::NotFound(format!(
"provider '{}/{}'",
input.source.app_type, input.source.id
)));
}
tx.commit()
.map_err(|error| AppError::Database(error.to_string()))
}
pub fn add_provider_endpoint(
&self,
key: &ProviderKey,
endpoint: NewEndpoint,
) -> Result<(), AppError> {
let mut conn = lock_conn!(self.conn);
let tx = conn
.transaction()
.map_err(|error| AppError::Database(error.to_string()))?;
insert_endpoint(&tx, key, &endpoint)?;
tx.commit()
.map_err(|error| AppError::Database(error.to_string()))
}
pub fn remove_provider_endpoint(&self, key: &ProviderKey, url: &str) -> Result<(), AppError> {
let conn = lock_conn!(self.conn);
if conn
.execute(
"DELETE FROM provider_endpoints
WHERE provider_id = ?1 AND app_type = ?2 AND url = ?3",
params![key.id, key.app_type, url],
)
.map_err(|error| AppError::Database(error.to_string()))?
!= 1
{
return Err(AppError::NotFound(format!(
"provider endpoint '{}/{}/{}'",
key.app_type, key.id, url
)));
}
Ok(())
}
pub fn touch_provider_endpoint(
&self,
key: &ProviderKey,
url: &str,
at: i64,
) -> Result<(), AppError> {
let conn = lock_conn!(self.conn);
if conn
.execute(
"UPDATE provider_endpoints
SET last_used = ?1
WHERE provider_id = ?2 AND app_type = ?3 AND url = ?4",
params![at, key.id, key.app_type, url],
)
.map_err(|error| AppError::Database(error.to_string()))?
!= 1
{
return Err(AppError::NotFound(format!(
"provider endpoint '{}/{}/{}'",
key.app_type, key.id, url
)));
}
Ok(())
}
pub(crate) fn update_provider_sort_index(
&self,
key: &ProviderKey,
sort_index: usize,
) -> Result<(), AppError> {
let conn = lock_conn!(self.conn);
if conn
.execute(
"UPDATE providers SET sort_index = ?1 WHERE id = ?2 AND app_type = ?3",
params![sort_index, key.id, key.app_type],
)
.map_err(|error| AppError::Database(error.to_string()))?
!= 1
{
return Err(AppError::NotFound(format!(
"provider '{}/{}'",
key.app_type, key.id
)));
}
Ok(())
}
}
src-tauri/src/database/dao/providers.rs
+626 -352
View File
File diff suppressed because it is too large Load Diff
src-tauri/src/database/dao/skill_deployments.rs
+208
View File
@@ -0,0 +1,208 @@
//! Device-local evidence for Pi Skill deployments.
// Pi skill reconciliation consumes this ledger in a later contract-ordered commit.
#![allow(dead_code)]
use crate::database::{lock_conn, Database};
use crate::error::AppError;
use rusqlite::{params, OptionalExtension};
use serde::{Deserialize, Serialize};
use std::str::FromStr;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum SkillDeploymentMethod {
Symlink,
Copy,
}
impl SkillDeploymentMethod {
pub(crate) const fn as_str(self) -> &'static str {
match self {
Self::Symlink => "symlink",
Self::Copy => "copy",
}
}
}
impl FromStr for SkillDeploymentMethod {
type Err = AppError;
fn from_str(value: &str) -> Result<Self, Self::Err> {
match value {
"symlink" => Ok(Self::Symlink),
"copy" => Ok(Self::Copy),
_ => Err(AppError::Database(format!(
"unknown Pi Skill deployment method '{value}'"
))),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct SkillDeployment {
pub skill_id: String,
pub destination: String,
pub destination_key: String,
pub method: SkillDeploymentMethod,
pub source_identity: String,
pub deployed_digest: Option<String>,
pub created_at: i64,
pub updated_at: i64,
}
fn decode_deployment(row: &rusqlite::Row<'_>) -> rusqlite::Result<SkillDeployment> {
let method: String = row.get(3)?;
let method = method.parse().map_err(|error: AppError| {
rusqlite::Error::FromSqlConversionFailure(3, rusqlite::types::Type::Text, Box::new(error))
})?;
Ok(SkillDeployment {
skill_id: row.get(0)?,
destination: row.get(1)?,
destination_key: row.get(2)?,
method,
source_identity: row.get(4)?,
deployed_digest: row.get(5)?,
created_at: row.get(6)?,
updated_at: row.get(7)?,
})
}
impl Database {
pub(crate) fn get_pi_skill_deployment(
&self,
skill_id: &str,
destination_key: &str,
) -> Result<Option<SkillDeployment>, AppError> {
let conn = lock_conn!(self.conn);
conn.query_row(
"SELECT skill_id, destination, destination_key, method,
source_identity, deployed_digest, created_at, updated_at
FROM skill_deployments
WHERE app_type = 'pi' AND skill_id = ?1 AND destination_key = ?2",
params![skill_id, destination_key],
decode_deployment,
)
.optional()
.map_err(|error| AppError::Database(error.to_string()))
}
pub(crate) fn get_pi_skill_deployments(
&self,
skill_id: &str,
) -> Result<Vec<SkillDeployment>, AppError> {
let conn = lock_conn!(self.conn);
let mut stmt = conn
.prepare(
"SELECT skill_id, destination, destination_key, method,
source_identity, deployed_digest, created_at, updated_at
FROM skill_deployments
WHERE app_type = 'pi' AND skill_id = ?1
ORDER BY created_at, destination_key",
)
.map_err(|error| AppError::Database(error.to_string()))?;
let rows = stmt
.query_map([skill_id], decode_deployment)
.map_err(|error| AppError::Database(error.to_string()))?;
rows.map(|row| row.map_err(|error| AppError::Database(error.to_string())))
.collect()
}
pub(crate) fn save_pi_skill_deployment(
&self,
deployment: &SkillDeployment,
) -> Result<(), AppError> {
if deployment.skill_id.trim().is_empty()
|| deployment.destination.trim().is_empty()
|| deployment.destination_key.trim().is_empty()
|| deployment.source_identity.trim().is_empty()
{
return Err(AppError::Config(
"Pi Skill deployment identity fields must be non-empty".to_string(),
));
}
let conn = lock_conn!(self.conn);
conn.execute(
"INSERT INTO skill_deployments (
app_type, skill_id, destination, destination_key, method,
source_identity, deployed_digest, created_at, updated_at
) VALUES ('pi', ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
ON CONFLICT(app_type, skill_id, destination_key) DO UPDATE SET
destination = excluded.destination,
method = excluded.method,
source_identity = excluded.source_identity,
deployed_digest = excluded.deployed_digest,
updated_at = excluded.updated_at",
params![
deployment.skill_id,
deployment.destination,
deployment.destination_key,
deployment.method.as_str(),
deployment.source_identity,
deployment.deployed_digest,
deployment.created_at,
deployment.updated_at,
],
)
.map_err(|error| AppError::Database(error.to_string()))?;
Ok(())
}
pub(crate) fn delete_pi_skill_deployment(
&self,
skill_id: &str,
destination_key: &str,
) -> Result<bool, AppError> {
let conn = lock_conn!(self.conn);
conn.execute(
"DELETE FROM skill_deployments
WHERE app_type = 'pi' AND skill_id = ?1 AND destination_key = ?2",
params![skill_id, destination_key],
)
.map(|count| count == 1)
.map_err(|error| AppError::Database(error.to_string()))
}
}
#[cfg(test)]
mod tests {
use super::*;
fn deployment(skill_id: &str, destination_key: &str) -> SkillDeployment {
SkillDeployment {
skill_id: skill_id.into(),
destination: format!("/tmp/{destination_key}"),
destination_key: destination_key.into(),
method: SkillDeploymentMethod::Copy,
source_identity: format!("source:{skill_id}"),
deployed_digest: Some("sha256:initial".into()),
created_at: 10,
updated_at: 10,
}
}
#[test]
fn skill_ledger_preserves_created_at_and_rejects_destination_collision() -> Result<(), AppError>
{
let db = Database::memory()?;
db.save_pi_skill_deployment(&deployment("one", "destination"))?;
let mut updated = deployment("one", "destination");
updated.updated_at = 20;
updated.deployed_digest = Some("sha256:updated".into());
db.save_pi_skill_deployment(&updated)?;
let saved = db
.get_pi_skill_deployment("one", "destination")?
.expect("deployment");
assert_eq!(saved.created_at, 10);
assert_eq!(saved.updated_at, 20);
assert_eq!(saved.deployed_digest.as_deref(), Some("sha256:updated"));
assert!(db
.save_pi_skill_deployment(&deployment("two", "destination"))
.is_err());
assert_eq!(db.get_pi_skill_deployments("one")?.len(), 1);
assert!(db.delete_pi_skill_deployment("one", "destination")?);
Ok(())
}
}
src-tauri/src/database/dao/skills.rs
+68 -2
View File
@@ -109,11 +109,28 @@ impl Database {
pub fn save_skill(&self, skill: &InstalledSkill) -> Result<(), AppError> {
let conn = lock_conn!(self.conn);
conn.execute(
"INSERT OR REPLACE INTO skills
"INSERT INTO skills
(id, name, description, directory, repo_owner, repo_name, repo_branch,
readme_url, enabled_claude, enabled_codex, enabled_gemini, enabled_grokbuild, enabled_opencode, enabled_hermes,
installed_at, content_hash, updated_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17)",
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17)
ON CONFLICT(id) DO UPDATE SET
name = excluded.name,
description = excluded.description,
directory = excluded.directory,
repo_owner = excluded.repo_owner,
repo_name = excluded.repo_name,
repo_branch = excluded.repo_branch,
readme_url = excluded.readme_url,
enabled_claude = excluded.enabled_claude,
enabled_codex = excluded.enabled_codex,
enabled_gemini = excluded.enabled_gemini,
enabled_grokbuild = excluded.enabled_grokbuild,
enabled_opencode = excluded.enabled_opencode,
enabled_hermes = excluded.enabled_hermes,
installed_at = excluded.installed_at,
content_hash = excluded.content_hash,
updated_at = excluded.updated_at",
params![
skill.id,
skill.name,
@@ -262,3 +279,52 @@ impl Database {
Ok(count)
}
}
#[cfg(test)]
mod tests {
use super::*;
fn installed_skill() -> InstalledSkill {
InstalledSkill {
id: "owner/repo:skill".into(),
name: "Skill".into(),
description: Some("before".into()),
directory: "skill".into(),
repo_owner: Some("owner".into()),
repo_name: Some("repo".into()),
repo_branch: Some("main".into()),
readme_url: None,
apps: SkillApps::default(),
installed_at: 10,
content_hash: Some("sha256:before".into()),
updated_at: 11,
}
}
#[test]
fn legacy_skill_save_preserves_pi_desired_state() -> Result<(), AppError> {
let db = Database::memory()?;
let mut skill = installed_skill();
db.save_skill(&skill)?;
{
let conn = lock_conn!(db.conn);
conn.execute(
"UPDATE skills SET enabled_pi = 1 WHERE id = ?1",
[&skill.id],
)?;
}
skill.name = "Updated".into();
skill.content_hash = Some("sha256:after".into());
db.save_skill(&skill)?;
let conn = lock_conn!(db.conn);
let saved: (String, String, bool) = conn.query_row(
"SELECT name, content_hash, enabled_pi FROM skills WHERE id = ?1",
[&skill.id],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)?;
assert_eq!(saved, ("Updated".into(), "sha256:after".into(), true));
Ok(())
}
}
src-tauri/src/database/mod.rs
+45 -1
View File
@@ -32,6 +32,9 @@ mod schema;
mod tests;
// DAO 类型导出供外部使用
pub use dao::provider_write::{
NewEndpoint, NewProviderAggregate, ProviderKey, ProviderRowUpdate, RenameProvider,
};
pub(crate) use dao::providers_seed::{
is_official_seed_id, CLAUDE_DESKTOP_OFFICIAL_PROVIDER_ID, CODEX_OFFICIAL_PROVIDER_ID,
GROKBUILD_OFFICIAL_PROVIDER_ID,
@@ -53,7 +56,7 @@ use std::sync::Mutex;
/// 当前 Schema 版本号
/// 每次修改表结构时递增，并在 schema.rs 中添加相应的迁移逻辑
pub(crate) const SCHEMA_VERSION: i32 = 16;
pub(crate) const SCHEMA_VERSION: i32 = 17;
/// 安全地序列化 JSON，避免 unwrap panic
pub(crate) fn to_json_string<T: Serialize>(value: &T) -> Result<String, AppError> {
@@ -197,6 +200,11 @@ impl Database {
conn: Mutex::new(conn),
};
db.create_tables()?;
// Keep the test database structurally identical to a fresh production
// database. Marking the base DDL as current without running the
// migration chain creates a false-current schema and makes restore
// tests certify columns that do not actually exist.
db.apply_schema_migrations()?;
db.ensure_model_pricing_seeded()?;
Ok(db)
@@ -293,3 +301,39 @@ impl Database {
Ok(count == 0)
}
}
#[cfg(test)]
impl Database {
/// Test-fixture reconciliation helper. Production code cannot call this:
/// provider writes there must choose a typed create or update operation.
pub(crate) fn reconcile_provider_fixture(
&self,
app_type: &str,
provider: &crate::provider::Provider,
) -> Result<(), AppError> {
let mut input = crate::provider::ProviderMutationInput {
id: provider.id.clone(),
name: provider.name.clone(),
settings_config: provider.settings_config.clone(),
website_url: provider.website_url.clone(),
category: provider.category.clone(),
created_at: provider.created_at,
sort_index: provider.sort_index,
notes: provider.notes.clone(),
meta: provider.meta.clone(),
icon: provider.icon.clone(),
icon_color: provider.icon_color.clone(),
in_failover_queue: provider.in_failover_queue,
};
if self.get_provider_aggregate(app_type, &input.id)?.is_some() {
if let Some(meta) = input.meta.as_mut() {
meta.custom_endpoints.clear();
}
let key = ProviderKey::new(app_type, input.id.clone())?;
let row = ProviderRowUpdate::from_input(&input)?;
self.update_provider(&key, &row)
} else {
self.create_provider(NewProviderAggregate::from_input(app_type, input)?)
}
}
}
src-tauri/src/database/schema.rs
+808 -35
View File
@@ -6,6 +6,498 @@ use super::{lock_conn, Database, SCHEMA_VERSION};
use crate::error::AppError;
use rusqlite::{params, Connection};
use serde::Serialize;
use tempfile::NamedTempFile;
/// A disk-backed database whose schema was created from this binary's current
/// schema code. Its fields are intentionally private: untrusted connections
/// cannot be wrapped or converted into a publishable stage.
pub(super) struct CanonicalStage {
connection: Connection,
_file: NamedTempFile,
}
impl CanonicalStage {
pub(super) fn connection(&self) -> &Connection {
&self.connection
}
pub(super) fn connection_mut(&mut self) -> &mut Connection {
&mut self.connection
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum CanonicalRestoreClass {
MigrateAndValidate,
RebuildAndPreserveLocal,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct CanonicalColumnSpec {
pub name: &'static str,
pub data_type: &'static str,
pub not_null: bool,
pub default: Option<&'static str>,
pub pk_position: i64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct CanonicalIndexedColumnSpec {
pub name: &'static str,
pub collation: &'static str,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct CanonicalUniqueSpec {
pub columns: &'static [CanonicalIndexedColumnSpec],
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct CanonicalForeignKeySpec {
pub from: &'static [&'static str],
pub table: &'static str,
pub to: &'static [&'static str],
pub on_update: &'static str,
pub on_delete: &'static str,
pub match_type: &'static str,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum SchemaInvariantKind {
EndpointIdentityUnique,
EndpointParentForeignKey,
EndpointDeleteCascade,
ProjectionProviderKeyUnique,
SkillAppTypePiOnly,
SkillMethodAllowed,
SkillDestinationOwnedOnce,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct SchemaInvariantSpec {
pub name: &'static str,
pub kind: SchemaInvariantKind,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct CanonicalTableSpec {
pub name: &'static str,
pub definition: &'static str,
pub restore_class: CanonicalRestoreClass,
pub columns: &'static [CanonicalColumnSpec],
pub unique_tuples: &'static [CanonicalUniqueSpec],
pub foreign_keys: &'static [CanonicalForeignKeySpec],
pub invariants: &'static [SchemaInvariantSpec],
}
const PROVIDERS_COLUMNS: &[CanonicalColumnSpec] = &[
CanonicalColumnSpec {
name: "id",
data_type: "TEXT",
not_null: true,
default: None,
pk_position: 1,
},
CanonicalColumnSpec {
name: "app_type",
data_type: "TEXT",
not_null: true,
default: None,
pk_position: 2,
},
CanonicalColumnSpec {
name: "name",
data_type: "TEXT",
not_null: true,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "settings_config",
data_type: "TEXT",
not_null: true,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "website_url",
data_type: "TEXT",
not_null: false,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "category",
data_type: "TEXT",
not_null: false,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "created_at",
data_type: "INTEGER",
not_null: false,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "sort_index",
data_type: "INTEGER",
not_null: false,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "notes",
data_type: "TEXT",
not_null: false,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "icon",
data_type: "TEXT",
not_null: false,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "icon_color",
data_type: "TEXT",
not_null: false,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "meta",
data_type: "TEXT",
not_null: true,
default: Some("'{}'"),
pk_position: 0,
},
CanonicalColumnSpec {
name: "is_current",
data_type: "BOOLEAN",
not_null: true,
default: Some("0"),
pk_position: 0,
},
CanonicalColumnSpec {
name: "in_failover_queue",
data_type: "BOOLEAN",
not_null: true,
default: Some("0"),
pk_position: 0,
},
];
const PROVIDER_ENDPOINT_COLUMNS: &[CanonicalColumnSpec] = &[
CanonicalColumnSpec {
name: "id",
data_type: "INTEGER",
not_null: false,
default: None,
pk_position: 1,
},
CanonicalColumnSpec {
name: "provider_id",
data_type: "TEXT",
not_null: true,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "app_type",
data_type: "TEXT",
not_null: true,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "url",
data_type: "TEXT",
not_null: true,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "added_at",
data_type: "INTEGER",
not_null: false,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "last_used",
data_type: "INTEGER",
not_null: false,
default: None,
pk_position: 0,
},
];
const PI_PROJECTION_COLUMNS: &[CanonicalColumnSpec] = &[
CanonicalColumnSpec {
name: "provider_id",
data_type: "TEXT",
not_null: false,
default: None,
pk_position: 1,
},
CanonicalColumnSpec {
name: "provider_key",
data_type: "TEXT",
not_null: true,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "created_at",
data_type: "INTEGER",
not_null: true,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "updated_at",
data_type: "INTEGER",
not_null: true,
default: None,
pk_position: 0,
},
];
const SKILL_DEPLOYMENT_COLUMNS: &[CanonicalColumnSpec] = &[
CanonicalColumnSpec {
name: "app_type",
data_type: "TEXT",
not_null: true,
default: None,
pk_position: 1,
},
CanonicalColumnSpec {
name: "skill_id",
data_type: "TEXT",
not_null: true,
default: None,
pk_position: 2,
},
CanonicalColumnSpec {
name: "destination",
data_type: "TEXT",
not_null: true,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "destination_key",
data_type: "TEXT",
not_null: true,
default: None,
pk_position: 3,
},
CanonicalColumnSpec {
name: "method",
data_type: "TEXT",
not_null: true,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "source_identity",
data_type: "TEXT",
not_null: true,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "deployed_digest",
data_type: "TEXT",
not_null: false,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "created_at",
data_type: "INTEGER",
not_null: true,
default: None,
pk_position: 0,
},
CanonicalColumnSpec {
name: "updated_at",
data_type: "INTEGER",
not_null: true,
default: None,
pk_position: 0,
},
];
const ENDPOINT_IDENTITY_COLUMNS: &[CanonicalIndexedColumnSpec] = &[
CanonicalIndexedColumnSpec {
name: "provider_id",
collation: "BINARY",
},
CanonicalIndexedColumnSpec {
name: "app_type",
collation: "BINARY",
},
CanonicalIndexedColumnSpec {
name: "url",
collation: "BINARY",
},
];
const PROJECTION_KEY_COLUMNS: &[CanonicalIndexedColumnSpec] = &[CanonicalIndexedColumnSpec {
name: "provider_key",
collation: "BINARY",
}];
const SKILL_DESTINATION_COLUMNS: &[CanonicalIndexedColumnSpec] = &[
CanonicalIndexedColumnSpec {
name: "app_type",
collation: "BINARY",
},
CanonicalIndexedColumnSpec {
name: "destination_key",
collation: "BINARY",
},
];
const ENDPOINT_FOREIGN_KEYS: &[CanonicalForeignKeySpec] = &[CanonicalForeignKeySpec {
from: &["provider_id", "app_type"],
table: "providers",
to: &["id", "app_type"],
on_update: "NO ACTION",
on_delete: "CASCADE",
match_type: "NONE",
}];
const ENDPOINT_INVARIANTS: &[SchemaInvariantSpec] = &[
SchemaInvariantSpec {
name: "endpoint_identity_unique",
kind: SchemaInvariantKind::EndpointIdentityUnique,
},
SchemaInvariantSpec {
name: "endpoint_parent_fk",
kind: SchemaInvariantKind::EndpointParentForeignKey,
},
SchemaInvariantSpec {
name: "endpoint_delete_cascade",
kind: SchemaInvariantKind::EndpointDeleteCascade,
},
];
const PROJECTION_INVARIANTS: &[SchemaInvariantSpec] = &[SchemaInvariantSpec {
name: "provider_key_unique",
kind: SchemaInvariantKind::ProjectionProviderKeyUnique,
}];
const SKILL_INVARIANTS: &[SchemaInvariantSpec] = &[
SchemaInvariantSpec {
name: "app_type_pi_only",
kind: SchemaInvariantKind::SkillAppTypePiOnly,
},
SchemaInvariantSpec {
name: "method_symlink_or_copy",
kind: SchemaInvariantKind::SkillMethodAllowed,
},
SchemaInvariantSpec {
name: "destination_owned_once",
kind: SchemaInvariantKind::SkillDestinationOwnedOnce,
},
];
pub(crate) const CANONICAL_TABLE_SPECS: &[CanonicalTableSpec] = &[
CanonicalTableSpec {
name: "providers",
definition: "providers (
id TEXT NOT NULL,
app_type TEXT NOT NULL,
name TEXT NOT NULL,
settings_config TEXT NOT NULL,
website_url TEXT,
category TEXT,
created_at INTEGER,
sort_index INTEGER,
notes TEXT,
icon TEXT,
icon_color TEXT,
meta TEXT NOT NULL DEFAULT '{}',
is_current BOOLEAN NOT NULL DEFAULT 0,
in_failover_queue BOOLEAN NOT NULL DEFAULT 0,
PRIMARY KEY (id, app_type)
)",
restore_class: CanonicalRestoreClass::MigrateAndValidate,
columns: PROVIDERS_COLUMNS,
unique_tuples: &[],
foreign_keys: &[],
invariants: &[],
},
CanonicalTableSpec {
name: "provider_endpoints",
definition: "provider_endpoints (
id INTEGER PRIMARY KEY AUTOINCREMENT,
provider_id TEXT NOT NULL,
app_type TEXT NOT NULL,
url TEXT NOT NULL,
added_at INTEGER,
last_used INTEGER,
FOREIGN KEY (provider_id, app_type)
REFERENCES providers(id, app_type) ON DELETE CASCADE,
UNIQUE (provider_id, app_type, url)
)",
restore_class: CanonicalRestoreClass::MigrateAndValidate,
columns: PROVIDER_ENDPOINT_COLUMNS,
unique_tuples: &[CanonicalUniqueSpec {
columns: ENDPOINT_IDENTITY_COLUMNS,
}],
foreign_keys: ENDPOINT_FOREIGN_KEYS,
invariants: ENDPOINT_INVARIANTS,
},
CanonicalTableSpec {
name: "pi_provider_projections",
definition: "pi_provider_projections (
provider_id TEXT PRIMARY KEY,
provider_key TEXT NOT NULL UNIQUE,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
)",
restore_class: CanonicalRestoreClass::RebuildAndPreserveLocal,
columns: PI_PROJECTION_COLUMNS,
unique_tuples: &[CanonicalUniqueSpec {
columns: PROJECTION_KEY_COLUMNS,
}],
foreign_keys: &[],
invariants: PROJECTION_INVARIANTS,
},
CanonicalTableSpec {
name: "skill_deployments",
definition: "skill_deployments (
app_type TEXT NOT NULL CHECK (app_type = 'pi'),
skill_id TEXT NOT NULL,
destination TEXT NOT NULL,
destination_key TEXT NOT NULL,
method TEXT NOT NULL CHECK (method IN ('symlink', 'copy')),
source_identity TEXT NOT NULL,
deployed_digest TEXT,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
PRIMARY KEY (app_type, skill_id, destination_key),
UNIQUE (app_type, destination_key)
)",
restore_class: CanonicalRestoreClass::RebuildAndPreserveLocal,
columns: SKILL_DEPLOYMENT_COLUMNS,
unique_tuples: &[CanonicalUniqueSpec {
columns: SKILL_DESTINATION_COLUMNS,
}],
foreign_keys: &[],
invariants: SKILL_INVARIANTS,
},
];
#[derive(Serialize)]
struct LegacySkillMigrationRow {
@@ -14,6 +506,77 @@ struct LegacySkillMigrationRow {
}
impl Database {
/// Construct a publish-capable stage from an empty disk file using only
/// this binary's schema and migration code.
pub(super) fn current_canonical_stage() -> Result<CanonicalStage, AppError> {
let file = NamedTempFile::new().map_err(|error| AppError::IoContext {
context: "create canonical restore stage".to_string(),
source: error,
})?;
let connection =
Connection::open(file.path()).map_err(|error| AppError::Database(error.to_string()))?;
connection
.execute_batch(
"PRAGMA foreign_keys = ON;
PRAGMA trusted_schema = OFF;",
)
.map_err(|error| AppError::Database(error.to_string()))?;
// Starting from version zero exercises the normal migration chain and
// yields the exact same current objects as a fresh production database.
Self::create_tables_on_conn(&connection)?;
Self::set_user_version(&connection, 0)?;
Self::apply_schema_migrations_on_conn(&connection)?;
if Self::get_user_version(&connection)? != SCHEMA_VERSION {
return Err(AppError::Database(
"canonical stage factory did not reach the current schema version".to_string(),
));
}
Ok(CanonicalStage {
connection,
_file: file,
})
}
pub(crate) fn canonical_table_spec(
name: &str,
) -> Result<&'static CanonicalTableSpec, AppError> {
CANONICAL_TABLE_SPECS
.iter()
.find(|spec| spec.name == name)
.ok_or_else(|| AppError::Config(format!("unknown canonical table '{name}'")))
}
fn create_canonical_table_on_conn(
conn: &Connection,
name: &str,
if_not_exists: bool,
) -> Result<(), AppError> {
let spec = Self::canonical_table_spec(name)?;
Self::create_canonical_table_as_on_conn(conn, spec, name, if_not_exists)
}
fn create_canonical_table_as_on_conn(
conn: &Connection,
spec: &CanonicalTableSpec,
target_name: &str,
if_not_exists: bool,
) -> Result<(), AppError> {
let qualifier = if if_not_exists { " IF NOT EXISTS" } else { "" };
let definition = if target_name == spec.name {
spec.definition.to_string()
} else {
spec.definition.replacen(spec.name, target_name, 1)
};
conn.execute(&format!("CREATE TABLE{qualifier} {definition}"), [])
.map_err(|error| {
AppError::Database(format!(
"failed to create canonical table '{target_name}': {error}"
))
})?;
Ok(())
}
/// 创建所有数据库表
pub(crate) fn create_tables(&self) -> Result<(), AppError> {
let conn = lock_conn!(self.conn);
@@ -23,41 +586,10 @@ impl Database {
/// 在指定连接上创建表（供迁移和测试使用）
pub(crate) fn create_tables_on_conn(conn: &Connection) -> Result<(), AppError> {
// 1. Providers 表
conn.execute(
"CREATE TABLE IF NOT EXISTS providers (
id TEXT NOT NULL,
app_type TEXT NOT NULL,
name TEXT NOT NULL,
settings_config TEXT NOT NULL,
website_url TEXT,
category TEXT,
created_at INTEGER,
sort_index INTEGER,
notes TEXT,
icon TEXT,
icon_color TEXT,
meta TEXT NOT NULL DEFAULT '{}',
is_current BOOLEAN NOT NULL DEFAULT 0,
in_failover_queue BOOLEAN NOT NULL DEFAULT 0,
PRIMARY KEY (id, app_type)
)",
[],
)
.map_err(|e| AppError::Database(e.to_string()))?;
Self::create_canonical_table_on_conn(conn, "providers", true)?;
// 2. Provider Endpoints 表
conn.execute(
"CREATE TABLE IF NOT EXISTS provider_endpoints (
id INTEGER PRIMARY KEY AUTOINCREMENT,
provider_id TEXT NOT NULL,
app_type TEXT NOT NULL,
url TEXT NOT NULL,
added_at INTEGER,
FOREIGN KEY (provider_id, app_type) REFERENCES providers(id, app_type) ON DELETE CASCADE
)",
[],
)
.map_err(|e| AppError::Database(e.to_string()))?;
Self::create_canonical_table_on_conn(conn, "provider_endpoints", true)?;
// 3. MCP Servers 表
conn.execute(
@@ -97,6 +629,7 @@ impl Database {
enabled_grokbuild BOOLEAN NOT NULL DEFAULT 0,
enabled_opencode BOOLEAN NOT NULL DEFAULT 0,
enabled_hermes BOOLEAN NOT NULL DEFAULT 0,
enabled_pi BOOLEAN NOT NULL DEFAULT 0,
installed_at INTEGER NOT NULL DEFAULT 0,
content_hash TEXT,
updated_at INTEGER NOT NULL DEFAULT 0
@@ -105,6 +638,14 @@ impl Database {
)
.map_err(|e| AppError::Database(e.to_string()))?;
// Exact models.json ownership is device-local and must never be
// inferred from provider names, content, or prefixes.
Self::create_canonical_table_on_conn(conn, "pi_provider_projections", true)?;
// Pi Skill deployment ownership is independent from desired enablement
// in `skills.enabled_pi` and from live discovery.
Self::create_canonical_table_on_conn(conn, "skill_deployments", true)?;
// 6. Skill Repos 表
conn.execute(
"CREATE TABLE IF NOT EXISTS skill_repos (
@@ -511,6 +1052,13 @@ impl Database {
Self::migrate_v15_to_v16(conn)?;
Self::set_user_version(conn, 16)?;
}
16 => {
log::info!(
"迁移数据库从 v16 到 v17（添加 Pi aggregate 与设备本地 ledger）"
);
Self::migrate_v16_to_v17(conn)?;
Self::set_user_version(conn, 17)?;
}
_ => {
return Err(AppError::Database(format!(
"未知的数据库版本 {version}，无法迁移到 {SCHEMA_VERSION}"
@@ -1523,6 +2071,88 @@ impl Database {
crate::services::session_usage_codex::reset_codex_usage_on_conn(conn, &codex_dir)
}
/// v16 -> v17: add the Pi desired bit, lossless endpoint metadata, and
/// device-local ownership ledgers. No ownership is inferred during
/// migration; both ledgers intentionally start empty.
fn migrate_v16_to_v17(conn: &Connection) -> Result<(), AppError> {
if Self::table_exists(conn, "provider_endpoints")? {
Self::add_column_if_missing(conn, "provider_endpoints", "last_used", "INTEGER")?;
// Older builds allowed duplicate rows for one logical endpoint.
// Merge their timestamps before rebuilding from the canonical
// definition. The fixed-column copy makes FK/UNIQUE/collation
// semantics part of migration rather than an optional index patch.
conn.execute_batch(
"UPDATE provider_endpoints AS kept
SET added_at = (
SELECT MIN(other.added_at)
FROM provider_endpoints AS other
WHERE other.provider_id = kept.provider_id
AND other.app_type = kept.app_type
AND other.url = kept.url
),
last_used = (
SELECT MAX(other.last_used)
FROM provider_endpoints AS other
WHERE other.provider_id = kept.provider_id
AND other.app_type = kept.app_type
AND other.url = kept.url
)
WHERE kept.id = (
SELECT MIN(other.id)
FROM provider_endpoints AS other
WHERE other.provider_id = kept.provider_id
AND other.app_type = kept.app_type
AND other.url = kept.url
);
DELETE FROM provider_endpoints
WHERE id NOT IN (
SELECT MIN(id)
FROM provider_endpoints
GROUP BY provider_id, app_type, url
);",
)
.map_err(|error| AppError::Database(error.to_string()))?;
const REBUILT_ENDPOINTS: &str = "provider_endpoints_v17_canonical";
conn.execute(&format!("DROP TABLE IF EXISTS \"{REBUILT_ENDPOINTS}\""), [])
.map_err(|error| AppError::Database(error.to_string()))?;
let spec = Self::canonical_table_spec("provider_endpoints")?;
Self::create_canonical_table_as_on_conn(conn, spec, REBUILT_ENDPOINTS, false)?;
conn.execute(
&format!(
"INSERT INTO \"{REBUILT_ENDPOINTS}\"
(id, provider_id, app_type, url, added_at, last_used)
SELECT id, provider_id, app_type, url, added_at, last_used
FROM provider_endpoints"
),
[],
)
.map_err(|error| {
AppError::Database(format!(
"failed to copy provider_endpoints into canonical v17 table: {error}"
))
})?;
conn.execute("DROP TABLE provider_endpoints", [])
.map_err(|error| AppError::Database(error.to_string()))?;
conn.execute(
&format!("ALTER TABLE \"{REBUILT_ENDPOINTS}\" RENAME TO provider_endpoints"),
[],
)
.map_err(|error| AppError::Database(error.to_string()))?;
} else {
Self::create_canonical_table_on_conn(conn, "provider_endpoints", false)?;
}
if Self::table_exists(conn, "skills")? {
Self::add_column_if_missing(
conn,
"skills",
"enabled_pi",
"BOOLEAN NOT NULL DEFAULT 0",
)?;
}
Self::create_canonical_table_on_conn(conn, "pi_provider_projections", true)?;
Self::create_canonical_table_on_conn(conn, "skill_deployments", true)
}
/// 插入默认模型定价数据
/// 格式: (model_id, display_name, input, output, cache_read, cache_creation)
/// 注意: model_id 使用短横线格式（如 claude-haiku-4-5），与 API 返回的模型名称标准化后一致
@@ -2780,7 +3410,7 @@ impl Database {
Self::ensure_model_pricing_seeded_on_conn(&conn)
}
fn ensure_model_pricing_seeded_on_conn(conn: &Connection) -> Result<(), AppError> {
pub(crate) fn ensure_model_pricing_seeded_on_conn(conn: &Connection) -> Result<(), AppError> {
// 每次启动都执行 INSERT OR IGNORE，增量追加新模型；仅修复仍等于旧内置值的定价。
Self::seed_model_pricing(conn)?;
Self::repair_current_model_pricing(conn)
@@ -2938,6 +3568,73 @@ impl Database {
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn canonical_manifest_from_specs() -> serde_json::Value {
let tables = CANONICAL_TABLE_SPECS
.iter()
.map(|spec| {
json!({
"name": spec.name,
"restoreClass": spec.restore_class,
"columns": spec.columns.iter().map(|column| json!([
column.name,
column.data_type,
column.not_null,
column.default,
column.pk_position
])).collect::<Vec<_>>(),
"uniqueTuples": spec.unique_tuples.iter().map(|tuple| {
tuple.columns.iter().map(|column| {
json!([column.name, column.collation])
}).collect::<Vec<_>>()
}).collect::<Vec<_>>(),
"foreignKeys": spec.foreign_keys.iter().map(|foreign_key| json!({
"from": foreign_key.from,
"table": foreign_key.table,
"to": foreign_key.to,
"onUpdate": foreign_key.on_update,
"onDelete": foreign_key.on_delete,
"match": foreign_key.match_type
})).collect::<Vec<_>>(),
"checks": spec.invariants.iter().map(|invariant| invariant.name).collect::<Vec<_>>()
})
})
.collect::<Vec<_>>();
json!({
"manifestVersion": 1,
"schemaVersion": SCHEMA_VERSION,
"codeAuthority": "src-tauri/src/database/schema.rs",
"comparison": "semantic",
"tables": tables,
"futureUsageActivation": {
"commit": 13,
"proxy_request_logs.input_token_semantics": {
"type": "INTEGER",
"notNull": true,
"default": null,
"allowed": [1, 2, 3, 4]
},
"usage_daily_rollups.input_token_semantics": {
"type": "INTEGER",
"notNull": true,
"default": null,
"allowed": [2]
}
}
})
}
#[test]
fn canonical_schema_specs_match_review_manifest() -> Result<(), AppError> {
let expected: serde_json::Value = serde_json::from_str(include_str!(
"../../../tests/fixtures/pi/canonical-schema-manifest-v1.json"
))
.expect("parse canonical schema manifest");
assert_eq!(canonical_manifest_from_specs(), expected);
Ok(())
}
#[test]
fn migrate_v12_to_v13_adds_input_token_semantics_columns() -> Result<(), AppError> {
@@ -3080,7 +3777,7 @@ mod tests {
Database::apply_schema_migrations_on_conn(&conn)?;
assert_eq!(Database::get_user_version(&conn)?, 16);
assert_eq!(Database::get_user_version(&conn)?, SCHEMA_VERSION);
let counts: (i64, i64, i64, i64) = conn.query_row(
"SELECT
(SELECT COUNT(*) FROM proxy_request_logs WHERE data_source = 'codex_session'),
@@ -3093,4 +3790,80 @@ mod tests {
assert_eq!(counts, (0, 1, 0, 1));
Ok(())
}
#[test]
fn migrate_v16_to_v17_adds_pi_ledgers_without_inferred_ownership() -> Result<(), AppError> {
let conn = Connection::open_in_memory()?;
conn.execute_batch(
"CREATE TABLE providers (
id TEXT NOT NULL,
app_type TEXT NOT NULL,
PRIMARY KEY (id, app_type)
);
CREATE TABLE provider_endpoints (
id INTEGER PRIMARY KEY,
provider_id TEXT NOT NULL,
app_type TEXT NOT NULL,
url TEXT NOT NULL,
added_at INTEGER
);
CREATE TABLE skills (
id TEXT PRIMARY KEY,
enabled_codex BOOLEAN NOT NULL DEFAULT 0
);
INSERT INTO providers (id, app_type) VALUES ('provider', 'pi');
INSERT INTO provider_endpoints
(id, provider_id, app_type, url, added_at)
VALUES
(1, 'provider', 'pi', 'https://duplicate.test', 20),
(2, 'provider', 'pi', 'https://duplicate.test', 10);
INSERT INTO skills (id, enabled_codex) VALUES ('existing', 1);",
)?;
Database::set_user_version(&conn, 16)?;
Database::apply_schema_migrations_on_conn(&conn)?;
assert_eq!(Database::get_user_version(&conn)?, SCHEMA_VERSION);
assert!(Database::has_column(
&conn,
"provider_endpoints",
"last_used"
)?);
assert!(Database::has_column(&conn, "skills", "enabled_pi")?);
assert!(Database::table_exists(&conn, "pi_provider_projections")?);
assert!(Database::table_exists(&conn, "skill_deployments")?);
let desired: i64 = conn.query_row(
"SELECT enabled_pi FROM skills WHERE id = 'existing'",
[],
|row| row.get(0),
)?;
let ledgers: (i64, i64) = conn.query_row(
"SELECT
(SELECT COUNT(*) FROM pi_provider_projections),
(SELECT COUNT(*) FROM skill_deployments)",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
assert_eq!(desired, 0);
assert_eq!(ledgers, (0, 0));
let endpoint: (i64, Option<i64>) = conn.query_row(
"SELECT COUNT(*), MIN(added_at)
FROM provider_endpoints
WHERE provider_id = 'provider'
AND app_type = 'pi'
AND url = 'https://duplicate.test'",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
assert_eq!(endpoint, (1, Some(10)));
assert!(conn
.execute(
"INSERT INTO provider_endpoints
(provider_id, app_type, url, added_at)
VALUES ('provider', 'pi', 'https://duplicate.test', 30)",
[],
)
.is_err());
Ok(())
}
}