surrealdb-core 3.3.1

use anyhow::{Result, bail};
use reblessive::tree::Stk;
use surrealdb_strand::TableName;
use uuid::Uuid;

use crate::catalog::providers::TableProvider;
use crate::catalog::{Error as CatalogError, TableDefinition};
use crate::ctx::FrozenContext;
use crate::dbs::Options;
use crate::doc::CursorDoc;
use crate::exec::Error as ExecError;
use crate::expr::Base;
use crate::expr::statements::remove::table::RemoveTableStatement;
use crate::iam::{Action, ResourceKind};
use crate::key::schema::{ForeignTableKey, TblRoot};
use crate::legacy::expr::statements::remove::retire_table_indexes;
use crate::legacy::{expr_to_ident, kill_table_subscriptions};
use crate::val::Value;

/// Process this type returning a computed simple Value
pub(crate) async fn remove_table_statement_compute(
	this: &RemoveTableStatement,
	stk: &mut Stk,
	ctx: &FrozenContext,
	opt: &Options,
	doc: Option<&CursorDoc>,
) -> Result<Value> {
	// Allowed to run?
	ctx.is_allowed(opt, Action::Edit, ResourceKind::Table, Base::Db)?;
	// Compute the name
	let name = TableName::new(expr_to_ident(stk, ctx, opt, doc, &this.name, "table name").await?);
	// Get the NS and DB
	let (ns_name, db_name) = opt.ns_db()?;
	let (ns, db) = ctx.expect_ns_db_ids(opt).await?;
	// Get the transaction
	let txn = ctx.tx();
	// Get the defined table
	let Some(tb) = txn.get_tb(ns, db, &name, None).await? else {
		if this.if_exists {
			return Ok(Value::None);
		}

		return Err(CatalogError::TbNotFound {
			name,
		}
		.into());
	};
	// Get the foreign tables
	let fts = txn.all_tb_views(ns, db, &name, None).await?;

	if !fts.is_empty() {
		let mut message =
			format!("Cannot delete table `{name}` on which a view is defined, table(s) `");
		for (idx, f) in fts.iter().enumerate() {
			if idx != 0 {
				message.push_str("`, `")
			}
			message.push_str(f.name.as_str());
		}

		message.push_str("` are defined as a view on this table.");

		bail!(ExecError::Query {
			message
		});
	}

	// A lightweight relation's edges live under its endpoint tables — in
	// both directions. Removing a table that a non-empty lightweight
	// relation names as an endpoint would orphan pointer keys the
	// record-less scan can no longer enumerate (its IN-side keys vanish
	// with this table's subspace, leaving the OUT-side keys unreachable),
	// so the relation must be emptied first.
	for other in txn.all_tb(ns, db, None).await?.iter() {
		let Some(rel) = crate::kvs::lightweight::lightweight_relation(&other.table_type) else {
			continue;
		};
		if other.name == name || (!rel.from.contains(&name) && !rel.to.contains(&name)) {
			continue;
		}
		anyhow::ensure!(
			crate::kvs::lightweight::lightweight_relation_is_empty(&txn, ns, db, &other.name, rel)
				.await?,
			ExecError::Thrown(format!(
				"cannot remove `{name}`: the LIGHTWEIGHT relation `{}` uses it as an endpoint \
				 and still holds edges — delete them first (e.g. `DELETE {}`)",
				other.name, other.name
			))
		);
	}

	// A lightweight relation's edges live under its endpoint tables'
	// subspaces, which deleting this table's own key range cannot reach —
	// they would silently survive as orphaned adjacency. Deleting them here
	// would be an unbounded scan inside one DDL transaction, so a non-empty
	// lightweight relation must be emptied first.
	if let Some(rel) = crate::kvs::lightweight::lightweight_relation(&tb.table_type) {
		anyhow::ensure!(
			crate::kvs::lightweight::lightweight_relation_is_empty(&txn, ns, db, &name, rel)
				.await?,
			ExecError::Thrown(format!(
				"cannot remove the LIGHTWEIGHT relation `{name}` while it holds edges: delete \
				 them first (e.g. `DELETE {name}`)"
			))
		);
	}

	// Retire index state before deleting the table definition. Durable
	// cleanup is transactional; local builder aborts are deferred until commit.
	retire_table_indexes(ctx, &txn, ns, db, &tb, this.expunge).await?;
	// Every subscription on the table is about to lose its keys, so each one
	// is owed a KILLED. Compiling is total, so this reaches subscriptions
	// whose text no longer parses too: the clients most in need of being
	// told, since nothing else will ever wake them.
	kill_table_subscriptions(ctx, &txn, ns, db, &name).await?;

	// Delete the definition
	if this.expunge {
		txn.clr_tb(ns_name, db_name, &name).await?
	} else {
		txn.del_tb(ns_name, db_name, &name).await?
	};

	// Remove the resource data
	let key = TblRoot {
		ns,
		db,
		tb: std::borrow::Cow::Borrowed(&name),
	};
	if this.expunge {
		txn.clr_prefix_key(&key).await?
	} else {
		txn.del_prefix_key(&key).await?
	};
	// The table's doc-ID mappings went with the rest of its keyspace, and a
	// redefinition restarts id allocation from scratch. The process-wide
	// adjacency resolve cache needs no eviction for that: it is keyed on
	// the table's never-reused `TableId`, so this generation's entries are
	// unreachable once the definition is gone.
	// Check if this is a foreign table
	if let Some(tables) = tb.view.as_ref().map(|v| v.source_tables()) {
		// Process each foreign table
		for ft in tables.iter() {
			// Save the view config
			let key = ForeignTableKey {
				ns,
				db,
				tb: std::borrow::Cow::Borrowed(ft),
				ft: std::borrow::Cow::Borrowed(&name),
			};
			txn.del_key(&key).await?;
			// Refresh the table cache for foreign tables
			let foreign_tb = txn.expect_tb(ns, db, ft).await?;
			txn.replace_tb(
				ns_name,
				db_name,
				&TableDefinition {
					cache_tables_ts: Uuid::now_v7(),
					..(*foreign_tb).clone()
				},
			)
			.await?;
		}
	}
	// The table (and its committed `cache_lives_ts`) is being removed, so
	// there is nothing to invalidate: open subscriptions were queued a KILLED
	// notification above, and a re-created table gets a fresh
	// `cache_lives_ts`, so the live-query cache cannot serve stale entries.
	// Clear the transaction cache
	txn.clear_cache();
	// Ok all good
	Ok(Value::None)
}