use crate::extension::Extension;
use crate::migrate;
use sqlx::SqlitePool;
use std::collections::HashMap;
use std::sync::{Arc, RwLock};
use tokio::sync::RwLock as AsyncRwLock;
#[derive(Debug, Clone, Copy)]
pub struct RuntimeStatus {
pub enabled: bool,
pub purged: bool,
}
impl RuntimeStatus {
pub fn active(self) -> bool {
self.enabled && !self.purged
}
}
pub struct ExtensionRegistry {
extensions: RwLock<Vec<Arc<dyn Extension>>>,
runtime: AsyncRwLock<HashMap<String, RuntimeStatus>>,
}
impl ExtensionRegistry {
pub fn new(extensions: Vec<Arc<dyn Extension>>) -> Self {
Self {
extensions: RwLock::new(extensions),
runtime: AsyncRwLock::new(HashMap::new()),
}
}
pub fn find(&self, id: &str) -> Option<Arc<dyn Extension>> {
self.extensions
.read()
.unwrap()
.iter()
.find(|e| e.id() == id)
.cloned()
}
pub fn iter(&self) -> Vec<Arc<dyn Extension>> {
self.extensions.read().unwrap().clone()
}
pub async fn register_and_activate(&self, ext: Arc<dyn Extension>) -> String {
let id = ext.id().to_string();
self.extensions.write().unwrap().push(ext);
self.runtime
.write()
.await
.insert(id.clone(), RuntimeStatus { enabled: true, purged: false });
id
}
pub async fn seed_runtime_state(
&self,
pool: &SqlitePool,
toml_enabled: &[String],
) -> anyhow::Result<()> {
let snapshot = self.iter();
let mut entries = Vec::new();
for ext in &snapshot {
let row: Option<(i64, i64)> =
sqlx::query_as("SELECT enabled, purged FROM extension_state WHERE extension_id = ?")
.bind(ext.id())
.fetch_optional(pool)
.await?;
let status = match row {
Some((e, p)) => RuntimeStatus {
enabled: e != 0,
purged: p != 0,
},
None => {
let enabled = toml_enabled.is_empty()
|| toml_enabled.iter().any(|id| id == ext.id());
sqlx::query(
"INSERT INTO extension_state (extension_id, enabled, purged)
VALUES (?1, ?2, 0)",
)
.bind(ext.id())
.bind(if enabled { 1i64 } else { 0 })
.execute(pool)
.await?;
RuntimeStatus {
enabled,
purged: false,
}
}
};
entries.push((ext.id().to_string(), status));
}
let mut cache = self.runtime.write().await;
cache.clear();
for (id, status) in entries {
cache.insert(id, status);
}
Ok(())
}
pub async fn status_of(&self, id: &str) -> Option<RuntimeStatus> {
self.runtime.read().await.get(id).copied()
}
pub async fn is_active(&self, id: &str) -> bool {
self.status_of(id).await.map(|s| s.active()).unwrap_or(false)
}
pub async fn status_snapshot(&self) -> HashMap<String, RuntimeStatus> {
self.runtime.read().await.clone()
}
pub async fn set_enabled(
&self,
pool: &SqlitePool,
id: &str,
enabled: bool,
) -> anyhow::Result<Option<RuntimeStatus>> {
let prev = self.status_of(id).await;
sqlx::query(
"INSERT INTO extension_state (extension_id, enabled, purged, disabled_at)
VALUES (?1, ?2, 0,
CASE WHEN ?2 = 0 THEN strftime('%Y-%m-%dT%H:%M:%fZ','now') ELSE NULL END)
ON CONFLICT(extension_id) DO UPDATE SET
enabled = ?2,
disabled_at = CASE WHEN ?2 = 0
THEN strftime('%Y-%m-%dT%H:%M:%fZ','now') ELSE NULL END",
)
.bind(id)
.bind(if enabled { 1i64 } else { 0 })
.execute(pool)
.await?;
if let Some(s) = self.runtime.write().await.get_mut(id) {
s.enabled = enabled;
}
Ok(prev)
}
pub async fn set_purged(
&self,
pool: &SqlitePool,
id: &str,
purged: bool,
) -> anyhow::Result<()> {
sqlx::query(
"INSERT INTO extension_state (extension_id, enabled, purged)
VALUES (?1, 0, ?2)
ON CONFLICT(extension_id) DO UPDATE SET purged = ?2, enabled = CASE WHEN ?2 = 1 THEN 0 ELSE enabled END",
)
.bind(id)
.bind(if purged { 1i64 } else { 0 })
.execute(pool)
.await?;
if let Some(s) = self.runtime.write().await.get_mut(id) {
s.purged = purged;
if purged {
s.enabled = false;
}
}
Ok(())
}
pub async fn run_migrations(
&self,
pool: &SqlitePool,
toml_enabled: &[String],
) -> anyhow::Result<()> {
migrate::run_migrations(pool, "_core", migrate::CORE_MIGRATIONS).await?;
self.seed_runtime_state(pool, toml_enabled).await?;
let snapshot = self.iter();
for ext in &snapshot {
let purged = self
.status_of(ext.id())
.await
.map(|s| s.purged)
.unwrap_or(false);
if !purged {
migrate::run_migrations(pool, ext.id(), &ext.migrations()).await?;
}
}
Ok(())
}
pub async fn rerun_migrations(&self, pool: &SqlitePool, id: &str) -> anyhow::Result<()> {
let Some(ext) = self.find(id) else {
return Ok(());
};
let migrations = ext.migrations();
sqlx::query("DELETE FROM schema_migrations WHERE extension = ?")
.bind(id)
.execute(pool)
.await?;
migrate::run_migrations(pool, id, &migrations).await?;
Ok(())
}
}