GKaszewski/k-tv
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: multi-instance provider support

Browse Source 
- provider_configs: add id TEXT PK; migrate existing rows (provider_type becomes id)
- local_files_index: add provider_id column + index; scope all queries per instance
- ProviderConfigRow: add id field; add get_by_id to trait
- LocalIndex:🆕 add provider_id param; all SQL scoped by provider_id
- factory: thread provider_id through build_local_files_bundle
- AppState.local_index: Option<Arc<LocalIndex>> → HashMap<String, Arc<LocalIndex>>
- admin_providers: restructured routes (POST /admin/providers create, PUT/DELETE /{id}, POST /test)
- admin_providers: use row.id as registry key for jellyfin and local_files
- files.rescan: optional ?provider=<id> query param
- frontend: add id to ProviderConfig, update api/hooks, new multi-instance panel UX
This commit is contained in:
Gabriel Kaszewski
2026-03-19 22:54:41 +01:00
parent 373e1c7c0a
commit 311fdd4006
14 changed files with 563 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
k-tv-backend/infra/src/factory.rs
View File

@@ -133,6 +133,7 @@ pub async fn build_local_files_bundle(
transcode_dir: Option<std::path::PathBuf>,
cleanup_ttl_hours: u32,
base_url: String,
provider_id: &str,
) -> FactoryResult<LocalFilesBundle> {
match pool {
#[cfg(feature = "sqlite")]
@@ -143,7 +144,7 @@ pub async fn build_local_files_bundle(
transcode_dir: transcode_dir.clone(),
cleanup_ttl_hours,
};
let idx = Arc::new(crate::LocalIndex::new(&cfg, sqlite_pool.clone()).await);
let idx = Arc::new(crate::LocalIndex::new(&cfg, sqlite_pool.clone(), provider_id.to_string()).await);
let tm = transcode_dir.as_ref().map(|td| {
std::fs::create_dir_all(td).ok();
crate::TranscodeManager::new(td.clone(), cleanup_ttl_hours)

24
k-tv-backend/infra/src/local_files/index.rs
View File

@@ -36,15 +36,17 @@ pub fn decode_id(id: &MediaItemId) -> Option<String> {
pub struct LocalIndex {
items: Arc<RwLock<HashMap<MediaItemId, LocalFileItem>>>,
pub root_dir: PathBuf,
provider_id: String,
pool: sqlx::SqlitePool,
}
impl LocalIndex {
/// Create the index, immediately loading persisted entries from SQLite.
pub async fn new(config: &LocalFilesConfig, pool: sqlx::SqlitePool) -> Self {
pub async fn new(config: &LocalFilesConfig, pool: sqlx::SqlitePool, provider_id: String) -> Self {
let idx = Self {
items: Arc::new(RwLock::new(HashMap::new())),
root_dir: config.root_dir.clone(),
provider_id,
pool,
};
idx.load_from_db().await;
@@ -65,8 +67,10 @@ impl LocalIndex {
}
let rows = sqlx::query_as::<_, Row>(
"SELECT id, rel_path, title, duration_secs, year, tags, top_dir FROM local_files_index",
"SELECT id, rel_path, title, duration_secs, year, tags, top_dir \
FROM local_files_index WHERE provider_id = ?",
)
.bind(&self.provider_id)
.fetch_all(&self.pool)
.await;
@@ -86,7 +90,7 @@ impl LocalIndex {
};
map.insert(MediaItemId::new(row.id), item);
}
info!("Local files index: loaded {} items from DB", map.len());
info!("Local files index [{}]: loaded {} items from DB", self.provider_id, map.len());
}
Err(e) => {
// Table might not exist yet on first run — that's fine.
@@ -100,7 +104,7 @@ impl LocalIndex {
/// Returns the number of items found. Called on startup (background task)
/// and via `POST /files/rescan`.
pub async fn rescan(&self) -> u32 {
info!("Local files: scanning {:?}", self.root_dir);
info!("Local files [{}]: scanning {:?}", self.provider_id, self.root_dir);
let new_items = scan_dir(&self.root_dir).await;
let count = new_items.len() as u32;
@@ -119,15 +123,16 @@ impl LocalIndex {
error!("Failed to persist local files index: {}", e);
}
info!("Local files: indexed {} items", count);
info!("Local files [{}]: indexed {} items", self.provider_id, count);
count
}
async fn save_to_db(&self, items: &[LocalFileItem]) -> Result<(), sqlx::Error> {
// Rebuild the table in one transaction.
// Rebuild the table in one transaction, scoped to this provider.
let mut tx = self.pool.begin().await?;
sqlx::query("DELETE FROM local_files_index")
sqlx::query("DELETE FROM local_files_index WHERE provider_id = ?")
.bind(&self.provider_id)
.execute(&mut *tx)
.await?;
@@ -137,8 +142,8 @@ impl LocalIndex {
let tags_json = serde_json::to_string(&item.tags).unwrap_or_else(|_| "[]".into());
sqlx::query(
"INSERT INTO local_files_index \
(id, rel_path, title, duration_secs, year, tags, top_dir, scanned_at) \
VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(id, rel_path, title, duration_secs, year, tags, top_dir, scanned_at, provider_id) \
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
)
.bind(&id)
.bind(&item.rel_path)
@@ -148,6 +153,7 @@ impl LocalIndex {
.bind(&tags_json)
.bind(&item.top_dir)
.bind(&now)
.bind(&self.provider_id)
.execute(&mut *tx)
.await?;
}

39
k-tv-backend/infra/src/provider_config_repository/sqlite.rs
View File

@@ -15,8 +15,8 @@ impl SqliteProviderConfigRepository {
#[async_trait]
impl ProviderConfigRepository for SqliteProviderConfigRepository {
async fn get_all(&self) -> DomainResult<Vec<ProviderConfigRow>> {
let rows: Vec<(String, String, i64, String)> = sqlx::query_as(
"SELECT provider_type, config_json, enabled, updated_at FROM provider_configs",
let rows: Vec<(String, String, String, i64, String)> = sqlx::query_as(
"SELECT id, provider_type, config_json, enabled, updated_at FROM provider_configs",
)
.fetch_all(&self.pool)
.await
@@ -24,7 +24,8 @@ impl ProviderConfigRepository for SqliteProviderConfigRepository {
Ok(rows
.into_iter()
.map(|(provider_type, config_json, enabled, updated_at)| ProviderConfigRow {
.map(|(id, provider_type, config_json, enabled, updated_at)| ProviderConfigRow {
id,
provider_type,
config_json,
enabled: enabled != 0,
@@ -33,15 +34,35 @@ impl ProviderConfigRepository for SqliteProviderConfigRepository {
.collect())
}
async fn get_by_id(&self, id: &str) -> DomainResult<Option<ProviderConfigRow>> {
let row: Option<(String, String, String, i64, String)> = sqlx::query_as(
"SELECT id, provider_type, config_json, enabled, updated_at FROM provider_configs WHERE id = ?",
)
.bind(id)
.fetch_optional(&self.pool)
.await
.map_err(|e| DomainError::RepositoryError(e.to_string()))?;
Ok(row.map(|(id, provider_type, config_json, enabled, updated_at)| ProviderConfigRow {
id,
provider_type,
config_json,
enabled: enabled != 0,
updated_at,
}))
}
async fn upsert(&self, row: &ProviderConfigRow) -> DomainResult<()> {
sqlx::query(
r#"INSERT INTO provider_configs (provider_type, config_json, enabled, updated_at)
VALUES (?, ?, ?, ?)
ON CONFLICT(provider_type) DO UPDATE SET
r#"INSERT INTO provider_configs (id, provider_type, config_json, enabled, updated_at)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
provider_type = excluded.provider_type,
config_json = excluded.config_json,
enabled = excluded.enabled,
updated_at = excluded.updated_at"#,
)
.bind(&row.id)
.bind(&row.provider_type)
.bind(&row.config_json)
.bind(row.enabled as i64)
@@ -52,9 +73,9 @@ impl ProviderConfigRepository for SqliteProviderConfigRepository {
Ok(())
}
async fn delete(&self, provider_type: &str) -> DomainResult<()> {
sqlx::query("DELETE FROM provider_configs WHERE provider_type = ?")
.bind(provider_type)
async fn delete(&self, id: &str) -> DomainResult<()> {
sqlx::query("DELETE FROM provider_configs WHERE id = ?")
.bind(id)
.execute(&self.pool)
.await
.map_err(|e| DomainError::RepositoryError(e.to_string()))?;