163952e
use std::time::Duration;
use export::{AppState, KenAuthError};
use models::dir::{dir_overwrite_group, dir_overwrite_users, dirs};
use models::repo::repos;
use models::room::rooms;
use models::space::{
space_bans, space_extras, space_group_permissions, space_groups, space_invite,
space_invite_groups, space_join_request, space_join_rule, space_member_extras,
space_member_groups, space_warn,
};
use models::system::{account_deletion, space_deletion};
use models::users::{
user, user_agreement_acceptances, user_blocks, user_external_identity, user_favorite_memes,
user_followers, user_notifications, user_preferences, user_profiles, user_push_subscriptions,
user_recent_mentions, user_saved_messages,
};
use proto::KenShellClient;
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QuerySelect, Set};
use serde::Serialize;
use time::OffsetDateTime;
use uuid::Uuid;
const RETRY_BASE_SECS: u64 = 30;
const RETRY_MAX_SECS: u64 = 3600;
#[derive(Debug, thiserror::Error)]
pub enum DeletionError {
#[error("database: {0}")]
Database(#[from] sea_orm::DbErr),
#[error("{0}")]
Conflict(String),
#[error("{0}")]
NotFound(String),
#[error("kenoauth: {0}")]
KenAuth(String),
#[error("kenshell: {0}")]
KenShell(String),
#[error("config: {0}")]
Config(String),
}
impl DeletionError {
pub fn is_retryable(&self) -> bool {
!matches!(
self,
Self::Conflict(_) | Self::NotFound(_) | Self::Config(_)
)
}
}
fn ken_auth(error: KenAuthError) -> DeletionError {
match error {
KenAuthError::NotFound => {
DeletionError::NotFound("account or organization not found".into())
}
KenAuthError::Rejected(message) => DeletionError::Conflict(message),
other => DeletionError::KenAuth(other.to_string()),
}
}
fn ken_shell(error: proto::KenShellError) -> DeletionError {
DeletionError::KenShell(error.to_string())
}
fn client(state: &AppState) -> Result<&dyn export::AuthApi, DeletionError> {
state
.kenauth
.as_deref()
.ok_or_else(|| DeletionError::Config("kenoauth is not configured".into()))
}
fn shell_client(state: &AppState) -> Result<KenShellClient, DeletionError> {
let cfg = &state.config.git;
KenShellClient::connect_with_key(
&cfg.kenshell_endpoint,
&cfg.kenshell_rpc_key,
Duration::from_secs(cfg.kenshell_timeout_secs.max(1)),
)
.map_err(ken_shell)
}
fn now() -> OffsetDateTime {
OffsetDateTime::now_utc()
}
fn grace_until(days: u32) -> OffsetDateTime {
now() + time::Duration::days(i64::from(days.max(1)))
}
fn retry_at(attempts: i32) -> OffsetDateTime {
let exponent = attempts.clamp(1, 8) as u32;
let delay = (RETRY_BASE_SECS.saturating_mul(1u64 << exponent)).min(RETRY_MAX_SECS);
now() + time::Duration::seconds(delay as i64)
}
#[derive(Debug, Clone, Serialize)]
pub struct AccountDeletionStatus {
pub status: String,
pub requested_at: OffsetDateTime,
pub purge_after: OffsetDateTime,
pub last_error: Option<String>,
}
#[derive(Debug, Clone, Serialize)]
pub struct SpaceDeletionStatus {
pub status: String,
pub requested_at: OffsetDateTime,
pub purge_after: OffsetDateTime,
pub last_error: Option<String>,
}
pub async fn request_account_deletion(
state: &AppState,
user_id: i64,
reason: &str,
) -> Result<AccountDeletionStatus, DeletionError> {
let db = state.database.write();
if let Some(existing) = account_deletion::Entity::find()
.filter(account_deletion::Column::UserId.eq(user_id))
.filter(account_deletion::Column::Status.eq("pending"))
.one(db)
.await?
{
return Ok(status_of_account(&existing));
}
let client = client(state)?;
let organizations = client
.list_organizations(
user_id,
user_id,
true,
proto::auth::common::v1::PageRequest {
page_size: 100,
page_token: String::new(),
},
)
.await
.map_err(ken_auth)?
.organizations;
for organization in organizations
.iter()
.filter(|organization| organization.owner_user_id == user_id)
{
let space_id = Uuid::parse_str(&organization.id)
.map_err(|_| DeletionError::Config("organization id is not a uuid".into()))?;
let pending = space_deletion::Entity::find()
.filter(space_deletion::Column::SpaceId.eq(space_id))
.filter(space_deletion::Column::Status.eq("pending"))
.one(db)
.await?;
if pending.is_none() {
return Err(DeletionError::Conflict(
"delete or transfer owned spaces before deleting the account".into(),
));
}
}
let status = client.get_user_status(user_id).await.map_err(ken_auth)?;
if status.deletion_requested_at.is_none() && status.deleted_at.is_none() {
client
.delete_account(user_id, reason)
.await
.map_err(ken_auth)?;
}
let now = now();
let purge_after = grace_until(state.config.user.deletion_grace_days);
let row = match account_deletion::Entity::find()
.filter(account_deletion::Column::UserId.eq(user_id))
.one(db)
.await?
{
Some(existing) => {
let mut active: account_deletion::ActiveModel = existing.into();
active.status = Set("pending".into());
active.requested_at = Set(now);
active.purge_after = Set(purge_after);
active.available_at = Set(purge_after);
active.last_error = Set(None);
active.attempts = Set(0);
active.updated_at = Set(now);
active.update(db).await?
}
None => {
account_deletion::ActiveModel {
id: Set(Uuid::now_v7()),
user_id: Set(user_id),
requested_at: Set(now),
purge_after: Set(purge_after),
status: Set("pending".into()),
last_error: Set(None),
attempts: Set(0),
available_at: Set(purge_after),
created_at: Set(now),
updated_at: Set(now),
}
.insert(db)
.await?
}
};
Ok(status_of_account(&row))
}
pub async fn cancel_account_deletion(state: &AppState, user_id: i64) -> Result<(), DeletionError> {
client(state)?
.cancel_account_deletion(user_id)
.await
.map_err(ken_auth)?;
account_deletion::Entity::delete_many()
.filter(account_deletion::Column::UserId.eq(user_id))
.exec(state.database.write())
.await?;
Ok(())
}
pub async fn account_deletion_status(
state: &AppState,
user_id: i64,
) -> Result<Option<AccountDeletionStatus>, DeletionError> {
Ok(account_deletion::Entity::find()
.filter(account_deletion::Column::UserId.eq(user_id))
.one(state.database.read())
.await?
.as_ref()
.map(status_of_account))
}
pub async fn request_space_deletion(
state: &AppState,
actor: i64,
space_id: Uuid,
reason: &str,
) -> Result<SpaceDeletionStatus, DeletionError> {
let db = state.database.write();
let client = client(state)?;
let organization = client
.get_organization(actor, &space_id.to_string(), "")
.await
.map_err(ken_auth)?
.organization
.ok_or_else(|| DeletionError::NotFound("organization not found".into()))?;
if organization.status != proto::auth::v1::OrganizationStatus::Archived as i32 {
client
.archive_organization(actor, &space_id.to_string(), reason)
.await
.map_err(ken_auth)?;
}
let now = now();
let purge_after = grace_until(state.config.space.deletion_grace_days);
let row = match space_deletion::Entity::find()
.filter(space_deletion::Column::SpaceId.eq(space_id))
.one(db)
.await?
{
Some(existing) => {
let mut active: space_deletion::ActiveModel = existing.into();
active.requested_by = Set(actor);
active.requested_at = Set(now);
active.purge_after = Set(purge_after);
active.status = Set("pending".into());
active.available_at = Set(purge_after);
active.last_error = Set(None);
active.attempts = Set(0);
active.updated_at = Set(now);
active.update(db).await?
}
None => {
space_deletion::ActiveModel {
id: Set(Uuid::now_v7()),
space_id: Set(space_id),
requested_by: Set(actor),
requested_at: Set(now),
purge_after: Set(purge_after),
status: Set("pending".into()),
last_error: Set(None),
attempts: Set(0),
available_at: Set(purge_after),
created_at: Set(now),
updated_at: Set(now),
}
.insert(db)
.await?
}
};
Ok(status_of_space(&row))
}
pub async fn cancel_space_deletion(
state: &AppState,
actor: i64,
space_id: Uuid,
) -> Result<(), DeletionError> {
client(state)?
.restore_organization(actor, &space_id.to_string())
.await
.map_err(ken_auth)?;
space_deletion::Entity::delete_many()
.filter(space_deletion::Column::SpaceId.eq(space_id))
.exec(state.database.write())
.await?;
Ok(())
}
pub async fn space_deletion_status(
state: &AppState,
space_id: Uuid,
) -> Result<Option<SpaceDeletionStatus>, DeletionError> {
Ok(space_deletion::Entity::find()
.filter(space_deletion::Column::SpaceId.eq(space_id))
.one(state.database.read())
.await?
.as_ref()
.map(status_of_space))
}
fn status_of_account(row: &account_deletion::Model) -> AccountDeletionStatus {
AccountDeletionStatus {
status: row.status.clone(),
requested_at: row.requested_at,
purge_after: row.purge_after,
last_error: row.last_error.clone(),
}
}
fn status_of_space(row: &space_deletion::Model) -> SpaceDeletionStatus {
SpaceDeletionStatus {
status: row.status.clone(),
requested_at: row.requested_at,
purge_after: row.purge_after,
last_error: row.last_error.clone(),
}
}
pub async fn run_worker(state: AppState) {
tracing::info!("deletion worker started");
let mut interval = tokio::time::interval(Duration::from_secs(30));
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
loop {
interval.tick().await;
if let Err(error) = purge_due(&state).await {
tracing::warn!(%error, "deletion worker pass failed");
}
}
}
async fn purge_due(state: &AppState) -> Result<(), DeletionError> {
let db = state.database.read();
let cutoff = now();
let spaces = space_deletion::Entity::find()
.filter(space_deletion::Column::Status.eq("pending"))
.filter(space_deletion::Column::AvailableAt.lte(cutoff))
.all(db)
.await?;
for row in spaces {
if let Err(error) = purge_space(state, &row).await {
record_space_failure(state, &row, &error).await?;
}
}
let accounts = account_deletion::Entity::find()
.filter(account_deletion::Column::Status.eq("pending"))
.filter(account_deletion::Column::AvailableAt.lte(cutoff))
.all(db)
.await?;
for row in accounts {
if let Err(error) = purge_account(state, &row).await {
record_account_failure(state, &row, &error).await?;
}
}
Ok(())
}
async fn purge_space(state: &AppState, row: &space_deletion::Model) -> Result<(), DeletionError> {
let client = client(state)?;
// Never delete an organization whose ownership changed after the request.
let organization = match client
.get_organization(0, &row.space_id.to_string(), "")
.await
{
Ok(response) => match response.organization {
Some(organization) => organization,
None => {
space_deletion::Entity::delete_by_id(row.id)
.exec(state.database.write())
.await?;
return Ok(());
}
},
Err(KenAuthError::NotFound) => {
space_deletion::Entity::delete_by_id(row.id)
.exec(state.database.write())
.await?;
return Ok(());
}
Err(error) => return Err(ken_auth(error)),
};
if organization.owner_user_id != row.requested_by {
client
.restore_organization(organization.owner_user_id, &row.space_id.to_string())
.await
.map_err(ken_auth)?;
space_deletion::Entity::delete_by_id(row.id)
.exec(state.database.write())
.await?;
return Ok(());
}
// 1. KenShell is the owner of repository/placement/LFS bytes.
let shell = shell_client(state)?;
let repos = shell
.list_repositories_for_org(&row.space_id.to_string())
.await
.map_err(ken_shell)?;
for repo in repos {
match shell.delete_repository(&repo.id).await {
Ok(()) => {}
Err(error) if error.is_not_found() => {}
Err(error) => return Err(ken_shell(error)),
}
}
// 2. Local KenApi resources become unreachable and org-scoped metadata is removed.
archive_local_space(state, row.space_id).await?;
// 3. The authority plane removes the organization for good.
client
.delete_organization(
row.requested_by,
&row.space_id.to_string(),
true,
"scheduled space deletion",
)
.await
.map_err(ken_auth)?;
space_deletion::Entity::delete_by_id(row.id)
.exec(state.database.write())
.await?;
tracing::info!(space_id = %row.space_id, "space purged");
Ok(())
}
async fn archive_local_space(state: &AppState, space_id: Uuid) -> Result<(), DeletionError> {
let db = state.database.write();
let now = now();
rooms::Entity::update_many()
.col_expr(
rooms::Column::IsActive,
sea_orm::sea_query::Expr::value(false),
)
.filter(rooms::Column::SpaceId.eq(Some(space_id)))
.exec(db)
.await?;
// Only the root nodes carry `space_id`; walk `parent_dir_id` to collect the
// whole tree, otherwise nested directories/repos survive the purge.
let mut dir_ids: Vec<Uuid> = dirs::Entity::find()
.filter(dirs::Column::SpaceId.eq(Some(space_id)))
.select_only()
.column(dirs::Column::Id)
.into_tuple()
.all(db)
.await?;
let mut seen: std::collections::HashSet<Uuid> = dir_ids.iter().copied().collect();
let mut frontier = dir_ids.clone();
while !frontier.is_empty() {
let children: Vec<Uuid> = dirs::Entity::find()
.filter(dirs::Column::ParentDirId.is_in(frontier))
.select_only()
.column(dirs::Column::Id)
.into_tuple()
.all(db)
.await?;
let mut next = Vec::new();
for id in children {
if seen.insert(id) {
dir_ids.push(id);
next.push(id);
}
}
frontier = next;
}
if !dir_ids.is_empty() {
dirs::Entity::update_many()
.col_expr(
dirs::Column::IsArchived,
sea_orm::sea_query::Expr::value(true),
)
.col_expr(
dirs::Column::ArchivedAt,
sea_orm::sea_query::Expr::value(Some(now)),
)
.filter(dirs::Column::Id.is_in(dir_ids.clone()))
.exec(db)
.await?;
repos::Entity::update_many()
.col_expr(
repos::Column::IsArchived,
sea_orm::sea_query::Expr::value(true),
)
.filter(repos::Column::DirId.is_in(dir_ids.clone()))
.exec(db)
.await?;
dir_overwrite_group::Entity::delete_many()
.filter(dir_overwrite_group::Column::DirId.is_in(dir_ids.clone()))
.exec(db)
.await?;
dir_overwrite_users::Entity::delete_many()
.filter(dir_overwrite_users::Column::DirId.is_in(dir_ids))
.exec(db)
.await?;
}
let group_ids: Vec<Uuid> = space_groups::Entity::find()
.filter(space_groups::Column::SpaceId.eq(space_id))
.select_only()
.column(space_groups::Column::Id)
.into_tuple()
.all(db)
.await?;
if !group_ids.is_empty() {
space_group_permissions::Entity::delete_many()
.filter(space_group_permissions::Column::GroupId.is_in(group_ids))
.exec(db)
.await?;
}
let invite_ids: Vec<Uuid> = space_invite::Entity::find()
.filter(space_invite::Column::SpaceId.eq(space_id))
.select_only()
.column(space_invite::Column::Id)
.into_tuple()
.all(db)
.await?;
if !invite_ids.is_empty() {
space_invite_groups::Entity::delete_many()
.filter(space_invite_groups::Column::InviteId.is_in(invite_ids))
.exec(db)
.await?;
}
space_member_groups::Entity::delete_many()
.filter(space_member_groups::Column::SpaceId.eq(space_id))
.exec(db)
.await?;
space_groups::Entity::delete_many()
.filter(space_groups::Column::SpaceId.eq(space_id))
.exec(db)
.await?;
space_invite::Entity::delete_many()
.filter(space_invite::Column::SpaceId.eq(space_id))
.exec(db)
.await?;
space_join_request::Entity::delete_many()
.filter(space_join_request::Column::SpaceId.eq(space_id))
.exec(db)
.await?;
space_join_rule::Entity::delete_many()
.filter(space_join_rule::Column::SpaceId.eq(space_id))
.exec(db)
.await?;
space_warn::Entity::delete_many()
.filter(space_warn::Column::SpaceId.eq(space_id))
.exec(db)
.await?;
space_bans::Entity::delete_many()
.filter(space_bans::Column::SpaceId.eq(space_id))
.exec(db)
.await?;
space_extras::Entity::delete_by_id(space_id)
.exec(db)
.await?;
space_member_extras::Entity::delete_many()
.filter(space_member_extras::Column::SpaceId.eq(space_id))
.exec(db)
.await?;
Ok(())
}
async fn purge_account(
state: &AppState,
row: &account_deletion::Model,
) -> Result<(), DeletionError> {
let client = client(state)?;
let organizations = client
.list_organizations(
row.user_id,
row.user_id,
true,
proto::auth::common::v1::PageRequest {
page_size: 100,
page_token: String::new(),
},
)
.await
.map_err(ken_auth)?
.organizations;
if organizations
.iter()
.any(|organization| organization.owner_user_id == row.user_id)
{
return Err(DeletionError::Conflict(
"owned spaces are still being purged".into(),
));
}
let db = state.database.write();
let now = now();
user_preferences::Entity::delete_many()
.filter(user_preferences::Column::UserId.eq(row.user_id))
.exec(db)
.await?;
user_profiles::Entity::delete_many()
.filter(user_profiles::Column::UserId.eq(row.user_id))
.exec(db)
.await?;
user_notifications::Entity::delete_many()
.filter(user_notifications::Column::UserId.eq(row.user_id))
.exec(db)
.await?;
user_push_subscriptions::Entity::delete_many()
.filter(user_push_subscriptions::Column::UserId.eq(row.user_id))
.exec(db)
.await?;
user_favorite_memes::Entity::delete_many()
.filter(user_favorite_memes::Column::UserId.eq(row.user_id))
.exec(db)
.await?;
user_recent_mentions::Entity::delete_many()
.filter(user_recent_mentions::Column::UserId.eq(row.user_id))
.exec(db)
.await?;
user_saved_messages::Entity::delete_many()
.filter(user_saved_messages::Column::UserId.eq(row.user_id))
.exec(db)
.await?;
user_external_identity::Entity::delete_many()
.filter(user_external_identity::Column::UserId.eq(Some(row.user_id)))
.exec(db)
.await?;
user_agreement_acceptances::Entity::delete_many()
.filter(user_agreement_acceptances::Column::UserId.eq(row.user_id))
.exec(db)
.await?;
user_blocks::Entity::delete_many()
.filter(
sea_orm::Condition::any()
.add(user_blocks::Column::BlockerId.eq(row.user_id))
.add(user_blocks::Column::BlockedId.eq(row.user_id)),
)
.exec(db)
.await?;
user_followers::Entity::delete_many()
.filter(
sea_orm::Condition::any()
.add(user_followers::Column::FollowerId.eq(row.user_id))
.add(user_followers::Column::FollowedId.eq(row.user_id)),
)
.exec(db)
.await?;
if let Some(user) = user::Entity::find_by_id(row.user_id).one(db).await? {
let mut active: user::ActiveModel = user.into();
active.is_restricted = Set(true);
active.local_avatar_url = Set(String::new());
active.visibility = Set(models::users::user::ProfileVisibility::Private);
active.last_login_at = Set(None);
active.last_login_ip_hash = Set(None);
active.updated_at = Set(now);
active.update(db).await?;
}
match client.purge_account(row.user_id).await {
Ok(()) | Err(KenAuthError::NotFound) => {}
Err(error) => return Err(ken_auth(error)),
}
account_deletion::Entity::delete_by_id(row.id)
.exec(db)
.await?;
tracing::info!(user_id = %row.user_id, "account purged");
Ok(())
}
async fn record_space_failure(
state: &AppState,
row: &space_deletion::Model,
error: &DeletionError,
) -> Result<(), DeletionError> {
let attempts = row.attempts.saturating_add(1);
let mut active: space_deletion::ActiveModel = row.clone().into();
active.attempts = Set(attempts);
active.available_at = Set(retry_at(attempts));
active.last_error = Set(Some(error.to_string()));
active.status = Set("pending".into());
active.updated_at = Set(now());
active.update(state.database.write()).await?;
tracing::warn!(space_id = %row.space_id, %error, attempts, "space purge failed; will retry");
Ok(())
}
async fn record_account_failure(
state: &AppState,
row: &account_deletion::Model,
error: &DeletionError,
) -> Result<(), DeletionError> {
let attempts = row.attempts.saturating_add(1);
let mut active: account_deletion::ActiveModel = row.clone().into();
active.attempts = Set(attempts);
active.available_at = Set(retry_at(attempts));
active.last_error = Set(Some(error.to_string()));
active.status = Set("pending".into());
active.updated_at = Set(now());
active.update(state.database.write()).await?;
tracing::warn!(user_id = %row.user_id, %error, attempts, "account purge failed; will retry");
Ok(())
}