561 lines
19 KiB
Rust
561 lines
19 KiB
Rust
//! Marketplaces: git repos of agents, skills, commands, hooks and plugins that
|
|
//! are fetched on the host and synced into containers. See
|
|
//! `docs/superpowers/specs/2026-09-27-marketplace-design.md`.
|
|
|
|
pub mod auth;
|
|
pub mod catalog;
|
|
pub mod diff;
|
|
pub mod git;
|
|
pub mod payload;
|
|
pub mod tree;
|
|
#[cfg(test)]
|
|
pub(crate) mod test_support;
|
|
|
|
use std::collections::{BTreeSet, HashMap};
|
|
use std::path::{Path, PathBuf};
|
|
use std::sync::Mutex;
|
|
|
|
use tokio::sync::oneshot;
|
|
|
|
use crate::models::marketplace::{
|
|
CatalogItem, ItemUpdate, Marketplace, MarketplaceInstall, MarketplaceSnapshot, SyncReport,
|
|
};
|
|
use crate::models::{AppSettings, Project};
|
|
use catalog::{item_fingerprint, parse_catalog};
|
|
use tree::GitTree;
|
|
|
|
/// Emitted after every container sync, payload `{ project_id, report }`.
|
|
pub const SYNC_FINISHED_EVENT: &str = "marketplace-sync-finished";
|
|
|
|
pub struct MarketplaceManager {
|
|
data_root: PathBuf,
|
|
snapshots: Mutex<HashMap<String, MarketplaceSnapshot>>,
|
|
reports: Mutex<HashMap<String, SyncReport>>,
|
|
gh_login_cancel: tokio::sync::Mutex<Option<oneshot::Sender<()>>>,
|
|
/// Serialises writers of the bare caches (fetch, pins, cache removal) so
|
|
/// concurrent refreshes never race on gix ref locks (pre-flight F11a).
|
|
repo_lock: tokio::sync::Mutex<()>,
|
|
}
|
|
|
|
/// Project ids become file names; anything outside this set is not persisted.
|
|
fn safe_file_stem(id: &str) -> bool {
|
|
!id.is_empty()
|
|
&& id.len() <= 128
|
|
&& id
|
|
.chars()
|
|
.all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
|
|
}
|
|
|
|
impl MarketplaceManager {
|
|
/// `data_root` is `<data_dir>/triple-c`.
|
|
pub fn new(data_root: PathBuf) -> Self {
|
|
Self {
|
|
data_root,
|
|
snapshots: Mutex::new(HashMap::new()),
|
|
reports: Mutex::new(HashMap::new()),
|
|
gh_login_cancel: tokio::sync::Mutex::new(None),
|
|
repo_lock: tokio::sync::Mutex::new(()),
|
|
}
|
|
}
|
|
|
|
pub fn data_root(&self) -> &Path {
|
|
&self.data_root
|
|
}
|
|
|
|
/// Hold while writing to any marketplace cache (fetch, `git::set_pins`,
|
|
/// removing a cache).
|
|
pub fn repo_lock(&self) -> &tokio::sync::Mutex<()> {
|
|
&self.repo_lock
|
|
}
|
|
|
|
pub fn snapshot(&self, marketplace_id: &str) -> Option<MarketplaceSnapshot> {
|
|
self.snapshots.lock().unwrap().get(marketplace_id).cloned()
|
|
}
|
|
|
|
pub fn put_snapshot(&self, snap: MarketplaceSnapshot) {
|
|
self.snapshots
|
|
.lock()
|
|
.unwrap()
|
|
.insert(snap.marketplace_id.clone(), snap);
|
|
}
|
|
|
|
pub fn remove_snapshot(&self, marketplace_id: &str) {
|
|
self.snapshots.lock().unwrap().remove(marketplace_id);
|
|
}
|
|
|
|
fn report_path(&self, project_id: &str) -> PathBuf {
|
|
self.data_root
|
|
.join("marketplace-sync")
|
|
.join(format!("{project_id}.json"))
|
|
}
|
|
|
|
pub fn report(&self, project_id: &str) -> Option<SyncReport> {
|
|
if let Some(r) = self.reports.lock().unwrap().get(project_id) {
|
|
return Some(r.clone());
|
|
}
|
|
if !safe_file_stem(project_id) {
|
|
return None;
|
|
}
|
|
let text = std::fs::read_to_string(self.report_path(project_id)).ok()?;
|
|
let report: SyncReport = serde_json::from_str(&text).ok()?;
|
|
self.reports
|
|
.lock()
|
|
.unwrap()
|
|
.insert(project_id.to_string(), report.clone());
|
|
Some(report)
|
|
}
|
|
|
|
pub fn put_report(&self, project_id: &str, report: SyncReport) {
|
|
self.reports
|
|
.lock()
|
|
.unwrap()
|
|
.insert(project_id.to_string(), report.clone());
|
|
if !safe_file_stem(project_id) {
|
|
return;
|
|
}
|
|
let path = self.report_path(project_id);
|
|
let write = || -> std::io::Result<()> {
|
|
std::fs::create_dir_all(path.parent().unwrap())?;
|
|
let tmp = path.with_extension("json.tmp");
|
|
std::fs::write(&tmp, serde_json::to_vec_pretty(&report).unwrap_or_default())?;
|
|
std::fs::rename(&tmp, &path)
|
|
};
|
|
if let Err(e) = write() {
|
|
log::warn!(
|
|
"Could not persist the marketplace sync report for {}: {}",
|
|
project_id,
|
|
e
|
|
);
|
|
}
|
|
}
|
|
|
|
/// Claim (`Some`) or release (`None`) the single gh-login slot. Claiming
|
|
/// fails while another login holds it.
|
|
pub async fn set_gh_login_cancel(&self, tx: Option<oneshot::Sender<()>>) -> bool {
|
|
let mut slot = self.gh_login_cancel.lock().await;
|
|
match tx {
|
|
Some(tx) => {
|
|
if slot.is_some() {
|
|
return false;
|
|
}
|
|
*slot = Some(tx);
|
|
true
|
|
}
|
|
None => {
|
|
*slot = None;
|
|
true
|
|
}
|
|
}
|
|
}
|
|
|
|
pub async fn cancel_gh_login(&self) {
|
|
if let Some(tx) = self.gh_login_cancel.lock().await.take() {
|
|
let _ = tx.send(());
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Head commit for a marketplace: the in-memory snapshot's, else the cache's.
|
|
pub fn head_for(mgr: &MarketplaceManager, m: &Marketplace) -> Option<String> {
|
|
mgr.snapshot(&m.id).and_then(|s| s.head_commit).or_else(|| {
|
|
git::cached_head(&git::cache_path(mgr.data_root(), &m.id))
|
|
.ok()
|
|
.flatten()
|
|
})
|
|
}
|
|
|
|
fn parse_at(repo: &Path, commit: &str) -> Result<Vec<CatalogItem>, String> {
|
|
let tree = GitTree::open(repo, commit)?;
|
|
Ok(parse_catalog(&tree))
|
|
}
|
|
|
|
/// Snapshot from the cache alone (no network): startup, and after an install
|
|
/// when nothing is in memory. `fetched_at` stays `None`.
|
|
pub fn load_cached_snapshot(
|
|
mgr: &MarketplaceManager,
|
|
marketplace: &Marketplace,
|
|
) -> MarketplaceSnapshot {
|
|
let repo = git::cache_path(mgr.data_root(), &marketplace.id);
|
|
let mut snap = MarketplaceSnapshot {
|
|
marketplace_id: marketplace.id.clone(),
|
|
..Default::default()
|
|
};
|
|
match git::cached_head(&repo) {
|
|
Ok(Some(head)) => match parse_at(&repo, &head) {
|
|
Ok(items) => {
|
|
snap.head_commit = Some(head);
|
|
snap.items = items;
|
|
}
|
|
Err(e) => snap.fetch_error = Some(format!("The cached copy could not be read: {e}")),
|
|
},
|
|
Ok(None) => {}
|
|
Err(e) => snap.fetch_error = Some(format!("The cached copy could not be read: {e}")),
|
|
}
|
|
snap
|
|
}
|
|
|
|
/// Keep the previous items and head (in memory, else from the cache) and
|
|
/// record why this refresh failed.
|
|
fn failed_snapshot(
|
|
mgr: &MarketplaceManager,
|
|
m: &Marketplace,
|
|
message: String,
|
|
) -> MarketplaceSnapshot {
|
|
let mut snap = mgr
|
|
.snapshot(&m.id)
|
|
.unwrap_or_else(|| load_cached_snapshot(mgr, m));
|
|
snap.fetch_error = Some(message);
|
|
mgr.put_snapshot(snap.clone());
|
|
snap
|
|
}
|
|
|
|
/// Refresh one marketplace: resolve the credential, fetch (blocking task, under
|
|
/// the repo lock), parse the catalog at head and store the snapshot. On failure
|
|
/// the previous items and head are kept and `fetch_error` is set.
|
|
pub async fn refresh_marketplace(
|
|
mgr: &MarketplaceManager,
|
|
settings: &AppSettings,
|
|
marketplace_id: &str,
|
|
) -> MarketplaceSnapshot {
|
|
let Some(m) = settings
|
|
.marketplaces
|
|
.iter()
|
|
.find(|m| m.id == marketplace_id)
|
|
.cloned()
|
|
else {
|
|
return MarketplaceSnapshot {
|
|
marketplace_id: marketplace_id.to_string(),
|
|
fetch_error: Some("This marketplace is no longer configured.".to_string()),
|
|
..Default::default()
|
|
};
|
|
};
|
|
let account = m
|
|
.account_id
|
|
.as_ref()
|
|
.and_then(|id| settings.marketplace_accounts.iter().find(|a| &a.id == id))
|
|
.cloned();
|
|
let cred = match &account {
|
|
Some(a) => match auth::resolve_credential(a).await {
|
|
Ok(c) => Some(c),
|
|
Err(e) => return failed_snapshot(mgr, &m, e),
|
|
},
|
|
None => None,
|
|
};
|
|
|
|
let repo = git::cache_path(mgr.data_root(), &m.id);
|
|
let (url, branch) = (m.url.clone(), m.branch.clone());
|
|
let joined = {
|
|
let _repo_guard = mgr.repo_lock.lock().await;
|
|
tokio::task::spawn_blocking(move || {
|
|
let head = git::fetch(&repo, &url, branch.as_deref(), cred)?;
|
|
let items = parse_at(&repo, &head).map_err(git::FetchError::Other)?;
|
|
Ok::<_, git::FetchError>((head, items))
|
|
})
|
|
.await
|
|
};
|
|
|
|
match joined {
|
|
Ok(Ok((head, items))) => {
|
|
let snap = MarketplaceSnapshot {
|
|
marketplace_id: m.id.clone(),
|
|
head_commit: Some(head),
|
|
fetched_at: Some(chrono::Utc::now().to_rfc3339()),
|
|
fetch_error: None,
|
|
items,
|
|
};
|
|
mgr.put_snapshot(snap.clone());
|
|
snap
|
|
}
|
|
Ok(Err(e)) => failed_snapshot(
|
|
mgr,
|
|
&m,
|
|
auth::describe_fetch_error(&e, account.as_ref(), &m.url),
|
|
),
|
|
Err(e) => failed_snapshot(mgr, &m, format!("The refresh task failed: {e}")),
|
|
}
|
|
}
|
|
|
|
fn item_changed(repo: &Path, inst: &MarketplaceInstall, head: &str) -> Result<bool, String> {
|
|
let old = GitTree::open(repo, &inst.commit)?;
|
|
let new = GitTree::open(repo, head)?;
|
|
Ok(item_fingerprint(&old, inst.kind, &inst.key)?
|
|
!= item_fingerprint(&new, inst.kind, &inst.key)?)
|
|
}
|
|
|
|
/// Every install (global + all projects) whose item fingerprint at head
|
|
/// differs from its pin. Installs whose pin is not in the cache are skipped.
|
|
pub fn compute_updates(
|
|
mgr: &MarketplaceManager,
|
|
settings: &AppSettings,
|
|
projects: &[Project],
|
|
) -> Vec<ItemUpdate> {
|
|
let mut seen = BTreeSet::new();
|
|
let mut out = Vec::new();
|
|
let all = settings
|
|
.global_marketplace_installs
|
|
.iter()
|
|
.chain(projects.iter().flat_map(|p| p.marketplace_installs.iter()));
|
|
for inst in all {
|
|
if !seen.insert((inst.item_ref(), inst.commit.clone())) {
|
|
continue;
|
|
}
|
|
let Some(m) = settings
|
|
.marketplaces
|
|
.iter()
|
|
.find(|m| m.id == inst.marketplace_id)
|
|
else {
|
|
continue;
|
|
};
|
|
let Some(head) = head_for(mgr, m) else {
|
|
continue;
|
|
};
|
|
if head == inst.commit {
|
|
continue;
|
|
}
|
|
let repo = git::cache_path(mgr.data_root(), &m.id);
|
|
match item_changed(&repo, inst, &head) {
|
|
Ok(true) => out.push(ItemUpdate {
|
|
item: inst.item_ref(),
|
|
pinned: inst.commit.clone(),
|
|
head,
|
|
}),
|
|
Ok(false) => {}
|
|
Err(e) => log::debug!("Update check skipped for {}: {}", inst.key, e),
|
|
}
|
|
}
|
|
out
|
|
}
|
|
|
|
/// All commits referenced by installs, per marketplace (for `git::set_pins`).
|
|
pub fn pins_by_marketplace(
|
|
settings: &AppSettings,
|
|
projects: &[Project],
|
|
) -> HashMap<String, Vec<String>> {
|
|
let mut map: HashMap<String, BTreeSet<String>> = HashMap::new();
|
|
let all = settings
|
|
.global_marketplace_installs
|
|
.iter()
|
|
.chain(projects.iter().flat_map(|p| p.marketplace_installs.iter()));
|
|
for inst in all {
|
|
map.entry(inst.marketplace_id.clone())
|
|
.or_default()
|
|
.insert(inst.commit.clone());
|
|
}
|
|
map.into_iter()
|
|
.map(|(k, v)| (k, v.into_iter().collect()))
|
|
.collect()
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::marketplace::test_support::GitFixture;
|
|
use crate::models::marketplace::{ItemKind, Marketplace, MarketplaceInstall};
|
|
|
|
fn settings_with(url: &str) -> AppSettings {
|
|
let mut s = AppSettings::default();
|
|
s.marketplaces.push(Marketplace {
|
|
id: "m1".into(),
|
|
name: "Test".into(),
|
|
url: url.into(),
|
|
branch: None,
|
|
account_id: None,
|
|
});
|
|
s
|
|
}
|
|
|
|
fn install(kind: ItemKind, key: &str, commit: &str) -> MarketplaceInstall {
|
|
MarketplaceInstall {
|
|
marketplace_id: "m1".into(),
|
|
kind,
|
|
key: key.into(),
|
|
commit: commit.into(),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn refresh_parses_the_catalog_at_head() {
|
|
let Some(fx) = GitFixture::new() else { return };
|
|
let c1 = fx.with_all_kinds();
|
|
let data = tempfile::tempdir().unwrap();
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
|
|
let snap = refresh_marketplace(&mgr, &settings_with(&fx.url()), "m1").await;
|
|
|
|
assert_eq!(snap.fetch_error, None);
|
|
assert_eq!(snap.head_commit.as_deref(), Some(c1.as_str()));
|
|
assert!(snap.fetched_at.is_some());
|
|
let mut keys: Vec<String> = snap
|
|
.items
|
|
.iter()
|
|
.map(|i| format!("{:?}:{}", i.kind, i.key))
|
|
.collect();
|
|
keys.sort();
|
|
assert_eq!(
|
|
keys,
|
|
vec![
|
|
"Agent:code-reviewer",
|
|
"Command:example-command",
|
|
"Hook:notify-on-stop",
|
|
"Plugin:example-plugin",
|
|
"Skill:example-skill",
|
|
]
|
|
);
|
|
assert_eq!(mgr.snapshot("m1"), Some(snap));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn refresh_failure_keeps_snapshot() {
|
|
let Some(fx) = GitFixture::new() else { return };
|
|
let c1 = fx.with_all_kinds();
|
|
let url = fx.url();
|
|
let data = tempfile::tempdir().unwrap();
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
let settings = settings_with(&url);
|
|
let first = refresh_marketplace(&mgr, &settings, "m1").await;
|
|
assert_eq!(first.fetch_error, None);
|
|
|
|
drop(fx); // the source repository disappears (offline, deleted, …)
|
|
let second = refresh_marketplace(&mgr, &settings, "m1").await;
|
|
|
|
assert!(second.fetch_error.is_some(), "expected a fetch error");
|
|
assert_eq!(second.head_commit.as_deref(), Some(c1.as_str()));
|
|
assert_eq!(second.items, first.items);
|
|
assert_eq!(second.fetched_at, first.fetched_at);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn refresh_waits_for_the_repo_lock() {
|
|
let Some(fx) = GitFixture::new() else { return };
|
|
let c1 = fx.with_all_kinds();
|
|
let data = tempfile::tempdir().unwrap();
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
let settings = settings_with(&fx.url());
|
|
|
|
let guard = mgr.repo_lock().lock().await;
|
|
let blocked = tokio::time::timeout(
|
|
std::time::Duration::from_millis(300),
|
|
refresh_marketplace(&mgr, &settings, "m1"),
|
|
)
|
|
.await;
|
|
assert!(blocked.is_err(), "refresh must not fetch while the repo lock is held");
|
|
assert!(
|
|
!git::cache_path(data.path(), "m1").exists(),
|
|
"nothing may touch the cache while the lock is held"
|
|
);
|
|
drop(guard);
|
|
|
|
let snap = refresh_marketplace(&mgr, &settings, "m1").await;
|
|
assert_eq!(snap.head_commit.as_deref(), Some(c1.as_str()));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn concurrent_refreshes_all_succeed() {
|
|
let Some(fx) = GitFixture::new() else { return };
|
|
let c1 = fx.with_all_kinds();
|
|
let data = tempfile::tempdir().unwrap();
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
let settings = settings_with(&fx.url());
|
|
|
|
let (a, b, c) = tokio::join!(
|
|
refresh_marketplace(&mgr, &settings, "m1"),
|
|
refresh_marketplace(&mgr, &settings, "m1"),
|
|
refresh_marketplace(&mgr, &settings, "m1"),
|
|
);
|
|
for snap in [a, b, c] {
|
|
assert_eq!(snap.fetch_error, None);
|
|
assert_eq!(snap.head_commit.as_deref(), Some(c1.as_str()));
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn cached_snapshot_loads_without_network() {
|
|
let Some(fx) = GitFixture::new() else { return };
|
|
let c1 = fx.with_all_kinds();
|
|
let data = tempfile::tempdir().unwrap();
|
|
let settings = settings_with(&fx.url());
|
|
{
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
refresh_marketplace(&mgr, &settings, "m1").await;
|
|
}
|
|
drop(fx);
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
let snap = load_cached_snapshot(&mgr, &settings.marketplaces[0]);
|
|
assert_eq!(snap.head_commit.as_deref(), Some(c1.as_str()));
|
|
assert_eq!(snap.items.len(), 5);
|
|
assert_eq!(snap.fetch_error, None);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn only_items_whose_own_files_changed_have_updates() {
|
|
let Some(fx) = GitFixture::new() else { return };
|
|
let c1 = fx.with_all_kinds();
|
|
fx.write(
|
|
"agents/code-reviewer.md",
|
|
"---\nname: code-reviewer\ndescription: Reviews code\n---\nReview harder.\n",
|
|
);
|
|
let c2 = fx.commit("tweak agent");
|
|
let data = tempfile::tempdir().unwrap();
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
let mut settings = settings_with(&fx.url());
|
|
settings.global_marketplace_installs = vec![
|
|
install(ItemKind::Agent, "code-reviewer", &c1),
|
|
install(ItemKind::Hook, "notify-on-stop", &c1),
|
|
];
|
|
let mut project = crate::models::Project::new("p".into(), vec![]);
|
|
project.marketplace_installs = vec![install(ItemKind::Skill, "example-skill", &c1)];
|
|
refresh_marketplace(&mgr, &settings, "m1").await;
|
|
|
|
let updates = compute_updates(&mgr, &settings, &[project]);
|
|
|
|
assert_eq!(updates.len(), 1, "{updates:?}");
|
|
assert_eq!(updates[0].item.key, "code-reviewer");
|
|
assert_eq!(updates[0].pinned, c1);
|
|
assert_eq!(updates[0].head, c2);
|
|
}
|
|
|
|
#[test]
|
|
fn pins_are_grouped_and_deduplicated_per_marketplace() {
|
|
let a = "a".repeat(40);
|
|
let b = "b".repeat(40);
|
|
let mut settings = settings_with("https://example.invalid/r.git");
|
|
settings.global_marketplace_installs = vec![
|
|
install(ItemKind::Agent, "x", &b),
|
|
install(ItemKind::Hook, "y", &a),
|
|
];
|
|
let mut project = crate::models::Project::new("p".into(), vec![]);
|
|
project.marketplace_installs = vec![install(ItemKind::Agent, "z", &a)];
|
|
let pins = pins_by_marketplace(&settings, &[project]);
|
|
assert_eq!(pins.get("m1"), Some(&vec![a.clone(), b.clone()]));
|
|
}
|
|
|
|
#[test]
|
|
fn reports_are_persisted_per_project() {
|
|
let data = tempfile::tempdir().unwrap();
|
|
let report = SyncReport {
|
|
installed: vec!["agent:x".into()],
|
|
..Default::default()
|
|
};
|
|
MarketplaceManager::new(data.path().to_path_buf()).put_report("proj-1", report.clone());
|
|
let fresh = MarketplaceManager::new(data.path().to_path_buf());
|
|
assert_eq!(fresh.report("proj-1"), Some(report));
|
|
assert_eq!(fresh.report("proj-2"), None);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn only_one_gh_login_may_hold_the_cancel_slot() {
|
|
let mgr = MarketplaceManager::new(std::env::temp_dir());
|
|
let (tx1, rx1) = tokio::sync::oneshot::channel();
|
|
let (tx2, _rx2) = tokio::sync::oneshot::channel();
|
|
assert!(mgr.set_gh_login_cancel(Some(tx1)).await);
|
|
assert!(!mgr.set_gh_login_cancel(Some(tx2)).await);
|
|
mgr.cancel_gh_login().await;
|
|
assert!(rx1.await.is_ok(), "cancel must signal the running login");
|
|
let (tx3, _rx3) = tokio::sync::oneshot::channel();
|
|
assert!(
|
|
mgr.set_gh_login_cancel(Some(tx3)).await,
|
|
"slot is free after cancel"
|
|
);
|
|
}
|
|
}
|