use anyhow::bail;
use sqlx::PgPool;
use uuid::Uuid;
use crate::store::admin::{self as store, Actor};
async fn team_id(pool: &PgPool, slug: &str) -> anyhow::Result<Uuid> {
Ok(store::team_id_by_slug(pool, slug).await?)
}
pub async fn team_create(pool: &PgPool, slug: &str, name: Option<String>) -> anyhow::Result<()> {
let team = store::create_team(pool, Actor::Cli, slug, name).await?;
println!("team '{}' ready", team.slug);
Ok(())
}
pub async fn team_list(pool: &PgPool) -> anyhow::Result<()> {
let rows = store::list_teams(pool).await?;
if rows.is_empty() {
println!("(no teams yet — create one with `team create --slug <slug>`)");
}
for t in rows {
println!("{:<20} {:<30} {} agent(s)", t.slug, t.name, t.agents);
}
Ok(())
}
pub async fn team_capability(pool: &PgPool, team: &str, conversations: bool) -> anyhow::Result<()> {
let id = team_id(pool, team).await?;
store::set_conversations(pool, Actor::Cli, id, conversations).await?;
println!(
"team '{team}': conversations {}",
if conversations { "enabled" } else { "disabled" }
);
if conversations {
println!(
"Agents of this team now see create_conversation and the rest. Existing tools \
are unchanged."
);
}
Ok(())
}
pub async fn team_backend(pool: &PgPool, team: &str, backend: &str) -> anyhow::Result<()> {
let id = team_id(pool, team).await?;
store::set_default_backend(pool, Actor::Cli, id, backend).await?;
println!("team '{team}': new conversations are created on '{backend}'");
println!(
"Existing threads keep the backend they were created on. Nothing was migrated, and \
nothing will be by this command."
);
if backend == "jetstream" {
println!(
"Check before anyone writes: the stream exists (`ai-crew-sync team stream --team \
{team} --nats-url ...`) and the server was started with --nats-url. Without \
both, sends are accepted and stay pending."
);
}
Ok(())
}
#[derive(Clone, Copy, Debug)]
pub struct StreamQuotas {
pub max_bytes: i64,
pub max_messages: i64,
pub inbox_max_bytes: i64,
pub inbox_max_messages: i64,
}
impl Default for StreamQuotas {
fn default() -> Self {
use crate::store::jetstream::*;
Self {
max_bytes: DEFAULT_MAX_BYTES,
max_messages: DEFAULT_MAX_MESSAGES,
inbox_max_bytes: DEFAULT_INBOX_MAX_BYTES,
inbox_max_messages: DEFAULT_INBOX_MAX_MESSAGES,
}
}
}
pub async fn team_stream(
pool: &PgPool,
team: &str,
nats_url: &str,
credentials: Option<String>,
remove: bool,
quotas: StreamQuotas,
update: bool,
) -> anyhow::Result<()> {
use crate::store::jetstream::{
JetStreamBackend, Provisioned, QuotaChange, StreamKind, format_size,
};
let id = team_id(pool, team).await?;
let mut config = crate::store::jetstream::Config::new(nats_url.to_owned())
.with_limits(quotas.max_messages, quotas.max_bytes)
.with_inbox_limits(quotas.inbox_max_messages, quotas.inbox_max_bytes);
config.credentials = credentials;
config.validate_quotas()?;
if remove {
let (routed,): (String,) =
sqlx::query_as("SELECT default_backend FROM teams WHERE id = $1")
.bind(id)
.fetch_one(pool)
.await?;
if routed == "jetstream" {
anyhow::bail!(
"team '{team}' still creates its conversations on JetStream. Route it back \
with `team capability --backend postgres` first; deleting the stream now \
would drop bodies its threads still point at."
);
}
let (still_there,): (i64,) = sqlx::query_as(
"SELECT count(*) FROM conversations WHERE team_id = $1 AND backend = 'jetstream'",
)
.bind(id)
.fetch_one(pool)
.await?;
if still_there > 0 {
anyhow::bail!(
"{still_there} conversation(s) of team '{team}' still keep their bodies in \
this stream. Deleting it now would turn every one of those messages into \
a tombstone, and no rollback brings them back."
);
}
crate::store::jetstream::JetStreamBackend::deprovision(&config, id).await?;
println!("team '{team}': stream removed, with every body it held");
return Ok(());
}
let bodies = JetStreamBackend::provision(&config, id).await?;
let inbox = JetStreamBackend::provision_inbox(&config, id).await?;
let plan = [
(
bodies,
StreamKind::Bodies,
quotas.max_messages,
quotas.max_bytes,
),
(
inbox,
StreamKind::Inbox,
quotas.inbox_max_messages,
quotas.inbox_max_bytes,
),
];
let mut outcomes: Vec<(Provisioned, &str)> = Vec::with_capacity(2);
let mut kept = Vec::new();
if update {
let mut changes = Vec::new();
for (outcome, kind, want_messages, want_bytes) in &plan {
if outcome.differs_from(*want_messages, *want_bytes) {
changes.push((
*kind,
JetStreamBackend::check_update(&config, id, *kind).await?,
));
}
}
let additional: i64 = changes.iter().map(|(_, c)| c.additional_bytes()).sum();
if additional > 0 {
let account = JetStreamBackend::storage_account(&config).await?;
if !account.fits(additional) {
anyhow::bail!(
"the requested quotas need {} ({}) more reserved than the streams have now, \
and {}. Nothing was changed. Ask for smaller quotas, lower another \
stream's, remove one, or raise the broker's max_file_store.",
additional,
format_size(additional),
account.describe()
);
}
}
let mut applied: Vec<(StreamKind, &QuotaChange)> = Vec::new();
for (kind, change) in &changes {
match JetStreamBackend::apply_update(&config, id, *kind, change).await {
Ok(outcome) => {
applied.push((*kind, change));
outcomes.push((outcome, "updated"));
}
Err(e) => {
let mut restored = Vec::new();
for (done_kind, done) in &applied {
let previous = match done_kind {
StreamKind::Bodies => config
.clone()
.with_limits(done.current_max_messages, done.current_max_bytes),
StreamKind::Inbox => config.clone().with_inbox_limits(
done.current_max_messages,
done.current_max_bytes,
),
};
match JetStreamBackend::update_quotas(&previous, id, *done_kind).await {
Ok(_) => restored.push(done.name.clone()),
Err(back) => anyhow::bail!(
"updating '{}' failed ({e}) and restoring '{}' to its previous \
limits failed too ({back}); the team's quotas are now mixed. \
Re-run with --update-quotas once the cause is fixed.",
change.name,
done.name
),
}
}
if restored.is_empty() {
anyhow::bail!("{e}\nNothing was changed.");
}
anyhow::bail!(
"{e}\nRestored '{}' to its previous limits; nothing was changed.",
restored.join("', '")
);
}
}
}
}
for (outcome, _, want_messages, want_bytes) in plan {
if outcomes.iter().any(|(o, _)| o.name == outcome.name) {
continue;
}
if outcome.differs_from(want_messages, want_bytes) && !outcome.created {
kept.push(format!(
"'{}' keeps {} messages / {} (you asked for {} / {})",
outcome.name,
outcome.max_messages,
format_size(outcome.max_bytes),
want_messages,
format_size(want_bytes)
));
}
let verb = if outcome.created { "created" } else { "exists" };
outcomes.push((outcome, verb));
}
for (outcome, verb) in &outcomes {
let what = if outcome.name.starts_with("ACS_I_") {
"inbox references"
} else {
"bodies"
};
println!(
"team '{team}': '{}' ({what}) {verb}: up to {} messages / {} ({} bytes), holding {} \
messages / {}",
outcome.name,
outcome.max_messages,
format_size(outcome.max_bytes),
outcome.max_bytes,
outcome.messages,
format_size(outcome.bytes as i64)
);
if *verb == "updated" && outcome.over_ceiling() {
println!(
" note: '{}' grew past the new ceiling while it was being changed (a publisher \
got in between). Nothing was lost: the body stream refuses new writes until \
pruned, and a dropped inbox reference is rebuilt from Postgres.",
outcome.name
);
}
}
if !kept.is_empty() {
println!(
"Existing limits were kept: {}. Pass --update-quotas to apply the requested ones; \
a ceiling below what a stream already holds is refused.",
kept.join("; ")
);
}
println!(
"Reservation note: every stream's max_bytes counts against the broker's max_file_store \
from creation, used or not."
);
println!(
"This routes nobody. `team capability --team {team} --backend jetstream` is what \
sends new conversations there."
);
Ok(())
}
pub async fn conversations_migrate(
pool: &PgPool,
team: &str,
to: &str,
conversations: &[String],
nats_url: &str,
credentials: Option<String>,
apply: bool,
) -> anyhow::Result<()> {
use crate::store::migrate::{self, Direction};
let id = team_id(pool, team).await?;
let direction = Direction::parse(to)?;
let only: Vec<Uuid> = conversations
.iter()
.map(|c| {
c.trim()
.parse::<Uuid>()
.map_err(|_| anyhow::anyhow!("'{c}' is not a conversation id"))
})
.collect::<anyhow::Result<_>>()?;
let mut config = crate::store::jetstream::Config::new(nats_url.to_owned());
config.credentials = credentials;
let jetstream = crate::store::jetstream::JetStreamBackend::connect(&config, id).await?;
let plans = migrate::plan(pool, id, direction, &only).await?;
if plans.is_empty() {
println!("team '{team}': nothing matches");
return Ok(());
}
println!("team '{team}' → {}", direction.target());
for p in &plans {
println!(
" {} {:<30} {} message(s), {}{}",
p.conversation_id,
p.title.chars().take(30).collect::<String>(),
p.messages,
human_bytes(p.bytes),
match (&p.blocked, p.resuming) {
(Some(why), _) => format!(" — skipped: {why}"),
(None, true) => " — resuming an interrupted move".to_owned(),
(None, false) => String::new(),
}
);
}
if !apply {
println!();
println!("Dry run. Nothing was moved. Add --apply to run it.");
println!(
"This moves message bodies only. Attachments, memberships, receipts and ids \
stay exactly where and as they are."
);
println!(
"Each thread pauses writes only while its own tail is copied and verified; reads keep working throughout, and the rest of the bus is untouched."
);
return Ok(());
}
for p in plans.iter().filter(|p| p.blocked.is_none()) {
print!(" {} … ", p.conversation_id);
use std::io::Write as _;
let _ = std::io::stdout().flush();
match migrate::run(pool, &jetstream, id, p.conversation_id, direction).await {
Ok(o) => println!(
"moved {} message(s), {} already there, {} verified",
o.copied,
o.skipped,
human_bytes(o.bytes)
),
Err(e) => {
println!("FAILED: {e}");
match migrate::is_paused(pool, p.conversation_id).await {
Ok(false) => println!(
" Nothing was cut over for this thread and its writes are open again. Fix the cause and run the same command: what is already verified is not copied twice."
),
Ok(true) => println!(
" Nothing was cut over, but this thread is still paused under its open move. Fix the cause and run the same command: it resumes that move, copies nothing twice and reopens the thread."
),
Err(check) => println!(
" Nothing was cut over, and whether this thread is paused could not be checked ({check}). Run the same command once the database answers: it resumes an open move, or reports the thread as it is."
),
}
}
}
}
println!();
println!(
"Source bodies are kept. `conversations cleanup` drops them later, once you are sure you will not roll back."
);
Ok(())
}
pub async fn conversations_cleanup(
pool: &PgPool,
team: &str,
rollback_window_hours: i64,
apply: bool,
) -> anyhow::Result<()> {
let id = team_id(pool, team).await?;
let (count, bytes) =
crate::store::migrate::cleanup(pool, id, rollback_window_hours, apply).await?;
if count == 0 {
println!(
"team '{team}': nothing to drop (no move finished more than \
{rollback_window_hours}h ago)"
);
return Ok(());
}
if apply {
println!(
"team '{team}': dropped {count} source body(ies), {} freed in Postgres",
human_bytes(bytes)
);
println!("Rolling those threads back now needs the broker, not an older image.");
} else {
println!(
"team '{team}': {count} source body(ies) could be dropped, {} in Postgres",
human_bytes(bytes)
);
println!("Dry run. Add --apply to delete them.");
}
Ok(())
}
fn human_bytes(n: i64) -> String {
const UNITS: [&str; 4] = ["B", "KiB", "MiB", "GiB"];
let mut value = n as f64;
let mut unit = 0;
while value >= 1024.0 && unit < UNITS.len() - 1 {
value /= 1024.0;
unit += 1;
}
if unit == 0 {
format!("{n} B")
} else {
format!("{value:.1} {}", UNITS[unit])
}
}
pub async fn team_quota(pool: &PgPool, team: &str, bytes: Option<i64>) -> anyhow::Result<()> {
let id = team_id(pool, team).await?;
if let Some(b) = bytes
&& b <= 0
{
anyhow::bail!("a quota must be positive; omit --bytes to clear it");
}
sqlx::query("UPDATE teams SET attachment_bytes_limit = $1 WHERE id = $2")
.bind(bytes)
.bind(id)
.execute(pool)
.await?;
match bytes {
Some(b) => println!("team '{team}' attachment quota set to {}", human_bytes(b)),
None => println!("team '{team}' attachment quota cleared (unlimited)"),
}
Ok(())
}
pub async fn team_usage(pool: &PgPool, team: &str) -> anyhow::Result<()> {
let id = team_id(pool, team).await?;
let u = crate::store::quota::usage(pool, id).await?;
let quota = match u.attachment_bytes_limit {
Some(limit) => format!(
"{} of {} ({:.1}%)",
human_bytes(u.attachment_bytes),
human_bytes(limit),
u.percent_used().unwrap_or(0.0)
),
None => format!("{} (no quota set)", human_bytes(u.attachment_bytes)),
};
println!("team '{team}'");
println!(
" attachments {quota} across {} file(s)",
u.attachment_count
);
println!(" messages {}", u.messages);
println!(" note revisions {}", u.note_revisions);
println!(" task events {}", u.task_events);
if let Some(oldest) = u.oldest_message {
let days = (chrono::Utc::now() - oldest).num_days();
println!(" oldest message {days} day(s) ago");
}
let (backend,): (String,) = sqlx::query_as("SELECT default_backend FROM teams WHERE id = $1")
.bind(id)
.fetch_one(pool)
.await?;
let outbox = crate::store::outbox::status(pool, id).await?;
if backend != "postgres" || outbox.pending + outbox.leased + outbox.failed > 0 {
println!(" backend {backend}");
println!(
" publication {} pending, {} in flight, {} failed ({})",
outbox.pending,
outbox.leased,
outbox.failed,
human_bytes(outbox.pending_bytes)
);
if let Some(secs) = outbox.oldest_pending_seconds
&& secs > 300
{
println!(
" ⚠ the oldest unpublished message is {} minute(s) old. Check the broker \
and that a replica is draining (--publication-worker).",
secs / 60
);
}
if outbox.failed > 0 {
println!(
" ⚠ {} message(s) will not be published. Their slots stay so the gap is \
visible; readers are told.",
outbox.failed
);
}
}
if let Some(pct) = u.percent_used()
&& pct >= 80.0
{
println!();
println!(" ⚠ {pct:.0}% of the attachment quota is in use — raise it with");
println!(" `team quota --team {team} --bytes N`, or free space with `team prune`.");
}
Ok(())
}
pub async fn team_prune(pool: &PgPool, team: &str, days: i64, apply: bool) -> anyhow::Result<()> {
let id = team_id(pool, team).await?;
let report = crate::store::quota::prune(pool, id, days, !apply).await?;
let verb = if report.dry_run {
"would delete"
} else {
"deleted"
};
println!("team '{team}', anything older than {days} day(s):");
println!(" {verb} {} message(s)", report.messages);
println!(" {verb} {} note revision(s)", report.note_revisions);
println!(" {verb} {} task event(s)", report.task_events);
println!(
" {verb} attachments worth {}",
human_bytes(report.attachments_freed_bytes)
);
if report.dry_run {
println!();
println!("dry run — nothing was deleted. Re-run with --apply to do it.");
println!("Notes and tasks themselves are never pruned, only their history.");
}
Ok(())
}
pub async fn agent_add(
pool: &PgPool,
team: &str,
name: &str,
display_name: Option<String>,
issue_token: bool,
) -> anyhow::Result<()> {
let tid = team_id(pool, team).await?;
let agent = store::create_agent(pool, Actor::Cli, tid, name, display_name).await?;
println!("agent '{}' ready in team '{team}'", agent.name);
if issue_token {
token_issue(pool, team, &agent.name, None).await?;
}
Ok(())
}
pub async fn agent_list(pool: &PgPool, team: &str) -> anyhow::Result<()> {
let tid = team_id(pool, team).await?;
for a in store::list_agents(pool, tid).await? {
let flag = if a.disabled { " [disabled]" } else { "" };
println!(
"{:<24} {:<28} {} active token(s){flag}",
a.name,
a.display_name.unwrap_or_default(),
a.active_tokens
);
}
Ok(())
}
pub async fn agent_disable(pool: &PgPool, team: &str, name: &str) -> anyhow::Result<()> {
let tid = team_id(pool, team).await?;
match store::disable_agent(pool, Actor::Cli, tid, name).await {
Err(crate::error::BusError::NotFound(_)) => bail!("no agent '{name}' in team '{team}'"),
other => other?,
}
println!("agent '{name}' disabled; its tokens no longer authenticate");
Ok(())
}
pub async fn token_issue(
pool: &PgPool,
team: &str,
agent: &str,
label: Option<String>,
) -> anyhow::Result<()> {
let tid = team_id(pool, team).await?;
let issued = match store::issue_token(pool, Actor::Cli, tid, agent, label).await {
Err(crate::error::BusError::NotFound(_)) => {
bail!("no agent '{agent}' in team '{team}' — add it with `agent add` first")
}
other => other?,
};
println!();
println!("Token for {agent}@{team} — shown once, store it now:");
println!();
println!(" {}", issued.token);
println!();
Ok(())
}
pub async fn token_list(pool: &PgPool, team: &str) -> anyhow::Result<()> {
let tid = team_id(pool, team).await?;
for t in store::list_tokens(pool, tid).await? {
let used = t
.last_used_at
.map(|d| d.to_rfc3339_opts(chrono::SecondsFormat::Secs, true))
.unwrap_or_else(|| "never".into());
let flag = if t.revoked { " [revoked]" } else { "" };
println!(
"{} {:<20} {}… last used {used} {}{flag}",
t.id,
t.agent,
t.prefix,
t.label.unwrap_or_default()
);
}
Ok(())
}
pub async fn token_revoke(pool: &PgPool, id: Uuid) -> anyhow::Result<()> {
match store::revoke_token(pool, Actor::Cli, None, id).await {
Err(crate::error::BusError::NotFound(_)) => bail!("no token with id {id}"),
other => other?,
}
println!("token {id} revoked");
Ok(())
}
pub async fn admin_bootstrap(pool: &PgPool, label: Option<String>) -> anyhow::Result<()> {
let issued = store::grant_admin(pool, Actor::Cli, None, label).await?;
println!();
println!("Global administrative credential — shown once, store it now:");
println!();
println!(" {}", issued.token);
println!();
println!("Use it from your machine with `ai-crew-sync admin login --url <bus>`.");
match store::list_admins(pool, None).await {
Ok(rows) => {
let active = rows
.iter()
.filter(|c| c.team.is_none() && !c.revoked)
.count();
println!(
"{active} global credential(s) are now active; list them with `admin credential list`."
);
}
Err(e) => eprintln!("(could not count active credentials: {e})"),
}
Ok(())
}
pub async fn admin_credential_list(pool: &PgPool, team: Option<&str>) -> anyhow::Result<()> {
let tid = match team {
Some(slug) => Some(team_id(pool, slug).await?),
None => None,
};
let rows = store::list_admins(pool, tid).await?;
if rows.is_empty() {
println!("(no administrative credentials — mint the first with `admin bootstrap`)");
}
for c in rows {
print_admin_row(&c);
}
Ok(())
}
pub fn print_admin_row(c: &store::AdminRow) {
let used = c
.last_used_at
.map(|d| d.to_rfc3339_opts(chrono::SecondsFormat::Secs, true))
.unwrap_or_else(|| "never".into());
let scope = c.team.as_deref().unwrap_or("(global)");
let flag = if c.revoked { " [revoked]" } else { "" };
println!(
"{} {scope:<20} {}… last used {used} {}{flag}",
c.id,
c.prefix,
c.label.as_deref().unwrap_or_default()
);
}
pub async fn admin_credential_revoke(pool: &PgPool, id: Uuid) -> anyhow::Result<()> {
match store::revoke_admin(pool, Actor::Cli, None, id).await {
Err(crate::error::BusError::NotFound(_)) => {
bail!("no administrative credential with id {id}")
}
other => other?,
}
println!("administrative credential {id} revoked");
Ok(())
}
pub fn print_proxy_config(
format: &str,
role: Option<&str>,
project: Option<&str>,
profile: Option<&str>,
) {
let mut args: Vec<String> = vec!["mcp".into(), "proxy".into()];
for (flag, value) in [
("--role", role),
("--project", project),
("--profile", profile),
] {
if let Some(v) = value.map(str::trim).filter(|v| !v.is_empty()) {
args.push(flag.into());
args.push(v.into());
}
}
let exe = "ai-crew-sync";
match format {
"toml" => {
println!("# ~/.codex/config.toml (or <repo>/.codex/config.toml in a trusted project)");
println!("[mcp_servers.ai-crew-sync]");
println!("command = \"{exe}\"");
println!(
"args = [{}]",
args.iter()
.map(|a| format!("\"{a}\""))
.collect::<Vec<_>>()
.join(", ")
);
println!();
println!("# No token here: credentials come from your local profiles");
println!("# (`ai-crew-sync context profile add`), never from this file.");
}
_ => {
let cfg = serde_json::json!({
"mcpServers": {
"ai-crew-sync": { "command": exe, "args": args }
}
});
println!("{}", serde_json::to_string_pretty(&cfg).unwrap_or_default());
}
}
}
pub fn print_mcp_config(url: &str, token: &str, session: Option<&str>) {
let mut headers = serde_json::Map::new();
headers.insert("Authorization".into(), format!("Bearer {token}").into());
if let Some(session) = session.map(str::trim).filter(|s| !s.is_empty()) {
headers.insert(crate::auth::SESSION_HEADER.into(), session.into());
}
let cfg = serde_json::json!({
"mcpServers": {
"ai-crew-sync": {
"type": "http",
"url": url,
"headers": headers
}
}
});
println!("{}", serde_json::to_string_pretty(&cfg).unwrap());
}