reifydb_catalog/

error.rs

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2025 ReifyDB

use std::{
	fmt,
	fmt::{Display, Formatter},
};

use reifydb_core::{
	interface::catalog::config::{AcceptError, ConfigKey},
	key::kind::KeyKind,
};
use reifydb_runtime::hash::Hash128;
use reifydb_type::{
	error::{Diagnostic, Error, IntoDiagnostic},
	fragment::Fragment,
	value::r#type::Type,
};

#[derive(Debug, Clone, PartialEq)]
pub enum CatalogObjectKind {
	Namespace,
	Table,
	View,
	Flow,
	RingBuffer,
	Dictionary,
	Enum,
	Event,
	VirtualTable,
	Handler,
	Series,
	Tag,
	Identity,
	Role,
	Policy,
	Migration,
	Column,
	Source,
	Sink,
	Procedure,
	TestProcedure,
	Binding,
}

impl Display for CatalogObjectKind {
	fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
		match self {
			CatalogObjectKind::Namespace => f.write_str("namespace"),
			CatalogObjectKind::Table => f.write_str("table"),
			CatalogObjectKind::View => f.write_str("view"),
			CatalogObjectKind::Flow => f.write_str("flow"),
			CatalogObjectKind::RingBuffer => f.write_str("ring buffer"),
			CatalogObjectKind::Dictionary => f.write_str("dictionary"),
			CatalogObjectKind::Enum => f.write_str("enum"),
			CatalogObjectKind::Event => f.write_str("event"),
			CatalogObjectKind::VirtualTable => f.write_str("virtual table"),
			CatalogObjectKind::Handler => f.write_str("handler"),
			CatalogObjectKind::Series => f.write_str("series"),
			CatalogObjectKind::Tag => f.write_str("tag"),
			CatalogObjectKind::Identity => f.write_str("identity"),
			CatalogObjectKind::Role => f.write_str("role"),
			CatalogObjectKind::Policy => f.write_str("policy"),
			CatalogObjectKind::Migration => f.write_str("migration"),
			CatalogObjectKind::Column => f.write_str("column"),
			CatalogObjectKind::Source => f.write_str("source"),
			CatalogObjectKind::Sink => f.write_str("sink"),
			CatalogObjectKind::Procedure => f.write_str("procedure"),
			CatalogObjectKind::TestProcedure => f.write_str("test procedure"),
			CatalogObjectKind::Binding => f.write_str("binding"),
		}
	}
}

#[derive(Debug, thiserror::Error)]
pub enum CatalogError {
	#[error("{kind} `{namespace}::{name}` already exists")]
	AlreadyExists {
		kind: CatalogObjectKind,
		namespace: String,
		name: String,
		fragment: Fragment,
	},

	#[error("{kind} `{namespace}::{name}` not found")]
	NotFound {
		kind: CatalogObjectKind,
		namespace: String,
		name: String,
		fragment: Fragment,
	},

	#[error("migration `{name}` has no rollback body")]
	MigrationNoRollbackBody {
		name: String,
		fragment: Fragment,
	},

	#[error("migration `{name}` content has changed since registration: was {expected_hex}, now {actual_hex}")]
	MigrationHashMismatch {
		name: String,
		expected: Hash128,
		actual: Hash128,
		expected_hex: String,
		actual_hex: String,
		fragment: Fragment,
	},

	#[error("column `{column}` already exists in {kind} `{namespace}::{name}`")]
	ColumnAlreadyExists {
		kind: CatalogObjectKind,
		namespace: String,
		name: String,
		column: String,
		fragment: Fragment,
	},

	#[error(
		"column `{column}` type `{column_type}` does not match dictionary `{dictionary}` value type `{dictionary_value_type}`"
	)]
	DictionaryTypeMismatch {
		column: String,
		column_type: Type,
		dictionary: String,
		dictionary_value_type: Type,
		fragment: Fragment,
	},

	#[error("auto increment is not supported for type `{ty}`")]
	AutoIncrementInvalidType {
		column: String,
		ty: Type,
		fragment: Fragment,
	},

	#[error("policy `{policy}` already exists for column `{column}`")]
	ColumnPropertyAlreadyExists {
		policy: String,
		column: String,
	},

	#[error("{kind} `{namespace}` already has pending changes in this transaction")]
	AlreadyPendingInTransaction {
		kind: CatalogObjectKind,
		namespace: String,
		name: Option<String>,
		fragment: Fragment,
	},

	#[error("cannot update {kind} as it is marked for deletion")]
	CannotUpdateDeleted {
		kind: CatalogObjectKind,
		namespace: String,
		name: Option<String>,
		fragment: Fragment,
	},

	#[error("{kind} is already marked for deletion")]
	CannotDeleteAlreadyDeleted {
		kind: CatalogObjectKind,
		namespace: String,
		name: Option<String>,
		fragment: Fragment,
	},

	#[error("primary key must contain at least one column")]
	PrimaryKeyEmpty {
		fragment: Fragment,
	},

	#[error("column with ID {column_id} not found for primary key")]
	PrimaryKeyColumnNotFound {
		fragment: Fragment,
		column_id: u64,
	},

	#[error("subscription `{name}` already exists")]
	SubscriptionAlreadyExists {
		fragment: Fragment,
		name: String,
	},

	#[error("subscription `{name}` not found")]
	SubscriptionNotFound {
		fragment: Fragment,
		name: String,
	},

	#[error("column `{column}` not found in {kind} `{namespace}`.`{name}`")]
	ColumnNotFound {
		kind: CatalogObjectKind,
		namespace: String,
		name: String,
		column: String,
		fragment: Fragment,
	},

	#[error("cannot drop {kind} because it is in use")]
	InUse {
		kind: CatalogObjectKind,
		namespace: String,
		name: Option<String>,
		dependents: String,
		fragment: Fragment,
	},

	#[error(
		"cannot drop {kind} procedure `{name}`: native/FFI/WASM procedures are managed by the runtime registry, not DDL"
	)]
	CannotDropEphemeralProcedure {
		kind: String,
		name: String,
		fragment: Fragment,
	},

	#[error("cannot register {kind} procedure as ephemeral: only Native/FFI/WASM variants are accepted")]
	CannotRegisterPersistentAsEphemeral {
		kind: String,
	},

	#[error("unknown config key `{0}`")]
	ConfigStorageKeyNotFound(String),

	#[error("config value for key `{0}` cannot be none")]
	ConfigValueInvalid(String),

	#[error("config value for key `{key}` must be of type `{expected:?}`, got `{actual}`")]
	ConfigTypeMismatch {
		key: String,
		expected: Vec<Type>,
		actual: Type,
	},

	#[error("config value for key `{key}` is invalid: {reason}")]
	ConfigInvalidValue {
		key: String,
		reason: String,
	},

	#[error("unknown operation `{operation}` for {target_type} policy")]
	PolicyInvalidOperation {
		target_type: &'static str,
		operation: String,
		valid: &'static [&'static str],
		policy_name: Option<String>,
	},

	#[error("invalid binding config: {reason}")]
	InvalidBindingConfig {
		reason: String,
		fragment: Fragment,
	},
}

impl From<(ConfigKey, AcceptError)> for CatalogError {
	fn from((key, err): (ConfigKey, AcceptError)) -> Self {
		match err {
			AcceptError::TypeMismatch {
				expected,
				actual,
			} => CatalogError::ConfigTypeMismatch {
				key: key.to_string(),
				expected,
				actual,
			},
			AcceptError::InvalidValue(reason) => CatalogError::ConfigInvalidValue {
				key: key.to_string(),
				reason,
			},
		}
	}
}

impl IntoDiagnostic for CatalogError {
	fn into_diagnostic(self) -> Diagnostic {
		match self {
			CatalogError::AlreadyExists {
				kind,
				namespace,
				name,
				fragment,
			} => {
				let (code, kind_str, help) = match kind {
					CatalogObjectKind::Namespace => (
						"CA_001",
						"namespace",
						"choose a different name or drop the existing namespace first",
					),
					CatalogObjectKind::Table => (
						"CA_003",
						"table",
						"choose a different name, drop the existing table or create table in a different namespace",
					),
					CatalogObjectKind::View => (
						"CA_003",
						"view",
						"choose a different name, drop the existing view or create view in a different namespace",
					),
					CatalogObjectKind::Flow => (
						"CA_030",
						"flow",
						"choose a different name, drop the existing flow or create flow in a different namespace",
					),
					CatalogObjectKind::RingBuffer => (
						"CA_005",
						"ring buffer",
						"choose a different name, drop the existing ring buffer or create ring buffer in a different namespace",
					),
					CatalogObjectKind::Dictionary => (
						"CA_006",
						"dictionary",
						"choose a different name, drop the existing dictionary or create dictionary in a different namespace",
					),
					CatalogObjectKind::Enum => (
						"CA_003",
						"enum",
						"choose a different name or drop the existing enum first",
					),
					CatalogObjectKind::Event => (
						"CA_003",
						"event",
						"choose a different name or drop the existing event first",
					),
					CatalogObjectKind::VirtualTable => (
						"CA_022",
						"virtual table",
						"choose a different name or unregister the existing virtual table first",
					),
					CatalogObjectKind::Handler => (
						"CA_003",
						"handler",
						"choose a different name or drop the existing handler first",
					),
					CatalogObjectKind::Series => (
						"CA_003",
						"series",
						"choose a different name or drop the existing series first",
					),
					CatalogObjectKind::Tag => (
						"CA_003",
						"tag",
						"choose a different name or drop the existing tag first",
					),
					CatalogObjectKind::Identity => (
						"CA_040",
						"identity",
						"choose a different name or drop the existing identity first",
					),
					CatalogObjectKind::Role => (
						"CA_041",
						"role",
						"choose a different name or drop the existing role first",
					),
					CatalogObjectKind::Policy => (
						"CA_042",
						"policy",
						"choose a different name or drop the existing policy first",
					),
					CatalogObjectKind::Migration => {
						("CA_046", "migration", "choose a different name for the migration")
					}
					CatalogObjectKind::Column => {
						("CA_003", "column", "ensure the column exists in the definition")
					}
					CatalogObjectKind::Source => (
						"CA_060",
						"source",
						"choose a different name, drop the existing source or create source in a different namespace",
					),
					CatalogObjectKind::Sink => (
						"CA_061",
						"sink",
						"choose a different name, drop the existing sink or create sink in a different namespace",
					),
					CatalogObjectKind::Procedure => (
						"CA_080",
						"procedure",
						"choose a different name, drop the existing procedure or create procedure in a different namespace",
					),
					CatalogObjectKind::TestProcedure => (
						"CA_081",
						"test procedure",
						"choose a different name or drop the existing test procedure first",
					),
					CatalogObjectKind::Binding => (
						"CA_087",
						"binding",
						"choose a different protocol key or drop the existing binding first",
					),
				};
				let message = if matches!(
					kind,
					CatalogObjectKind::Namespace | CatalogObjectKind::Migration
				) {
					format!("{} `{}` already exists", kind_str, name)
				} else {
					format!("{} `{}::{}` already exists", kind_str, namespace, name)
				};
				Diagnostic {
					code: code.to_string(),
					rql: None,
					message,
					fragment,
					label: Some(format!("duplicate {} definition", kind_str)),
					help: Some(help.to_string()),
					column: None,
					notes: vec![],
					cause: None,
					operator_chain: None,
				}
			}

			CatalogError::NotFound {
				kind,
				namespace,
				name,
				fragment,
			} => {
				let (code, kind_str, help) = match kind {
					CatalogObjectKind::Namespace => (
						"CA_002",
						"namespace",
						"make sure the namespace exists before using it or create it first".to_string(),
					),
					CatalogObjectKind::Table => (
						"CA_004",
						"table",
						"ensure the table exists or create it first using `CREATE TABLE`".to_string(),
					),
					CatalogObjectKind::View => (
						"CA_004",
						"view",
						"ensure the view exists or create it first using `CREATE VIEW`".to_string(),
					),
					CatalogObjectKind::Flow => (
						"CA_031",
						"flow",
						"ensure the flow exists or create it first using `CREATE FLOW`".to_string(),
					),
					CatalogObjectKind::RingBuffer => (
						"CA_006",
						"ring buffer",
						"ensure the ring buffer exists or create it first using `CREATE RING BUFFER`".to_string(),
					),
					CatalogObjectKind::Dictionary => (
						"CA_007",
						"dictionary",
						"ensure the dictionary exists or create it first using `CREATE DICTIONARY`".to_string(),
					),
					CatalogObjectKind::Enum => (
						"CA_002",
						"type",
						format!("create the enum first with `CREATE ENUM {}::{} {{ ... }}`", namespace, name),
					),
					CatalogObjectKind::Event => (
						"CA_002",
						"event",
						format!("create the event first with `CREATE EVENT {}::{} {{ ... }}`", namespace, name),
					),
					CatalogObjectKind::VirtualTable => (
						"CA_023",
						"virtual table",
						"ensure the virtual table is registered before using it".to_string(),
					),
					CatalogObjectKind::Handler => (
						"CA_004",
						"handler",
						"ensure the handler exists or create it first using `CREATE HANDLER`".to_string(),
					),
					CatalogObjectKind::Series => (
						"CA_024",
						"series",
						"ensure the series exists or create it first using `CREATE SERIES`".to_string(),
					),
					CatalogObjectKind::Tag => (
						"CA_002",
						"tag",
						format!("create the tag first with `CREATE TAG {}.{} {{ ... }}`", namespace, name),
					),
					CatalogObjectKind::Identity => (
						"CA_043",
						"identity",
						"ensure the identity exists or create it first using `CREATE IDENTITY`".to_string(),
					),
					CatalogObjectKind::Role => (
						"CA_044",
						"role",
						"ensure the role exists or create it first using `CREATE ROLE`".to_string(),
					),
					CatalogObjectKind::Policy => (
						"CA_045",
						"policy",
						"ensure the policy exists or create it first".to_string(),
					),
					CatalogObjectKind::Migration => (
						"CA_047",
						"migration",
						"ensure the migration exists or create it first using `CREATE MIGRATION`".to_string(),
					),
					CatalogObjectKind::Column => (
						"CA_004",
						"column",
						"ensure the column exists in the definition".to_string(),
					),
					CatalogObjectKind::Source => (
						"CA_062",
						"source",
						"ensure the source exists or create it first using `CREATE SOURCE`".to_string(),
					),
					CatalogObjectKind::Sink => (
						"CA_063",
						"sink",
						"ensure the sink exists or create it first using `CREATE SINK`".to_string(),
					),
					CatalogObjectKind::Procedure => (
						"CA_082",
						"procedure",
						"ensure the procedure exists or create it first using `CREATE PROCEDURE`".to_string(),
					),
					CatalogObjectKind::TestProcedure => (
						"CA_083",
						"test procedure",
						"ensure the test procedure exists or create it first using `CREATE TEST PROCEDURE`".to_string(),
					),
					CatalogObjectKind::Binding => (
						"CA_088",
						"binding",
						"ensure the binding exists or create it first using `CREATE <PROTOCOL> BINDING`".to_string(),
					),
				};
				let message = match kind {
					CatalogObjectKind::Namespace => {
						format!("{} `{}` not found", kind_str, namespace)
					}
					CatalogObjectKind::Migration => format!("{} `{}` not found", kind_str, name),
					_ => format!("{} `{}::{}` not found", kind_str, namespace, name),
				};
				let label_str = match kind {
					CatalogObjectKind::Namespace => "unknown namespace reference".to_string(),
					CatalogObjectKind::Enum => "unknown type".to_string(),
					CatalogObjectKind::Event => "unknown event reference".to_string(),
					_ => format!("unknown {} reference", kind_str),
				};
				Diagnostic {
					code: code.to_string(),
					rql: None,
					message,
					fragment,
					label: Some(label_str),
					help: Some(help),
					column: None,
					notes: vec![],
					cause: None,
					operator_chain: None,
				}
			}

			CatalogError::MigrationNoRollbackBody {
				name,
				fragment,
			} => Diagnostic {
				code: "CA_048".to_string(),
				rql: None,
				message: format!("migration `{}` has no rollback body", name),
				fragment,
				label: Some("no rollback body defined".to_string()),
				help: Some("define a ROLLBACK clause when creating the migration".to_string()),
				column: None,
				notes: vec![],
				cause: None,
				operator_chain: None,
			},

			CatalogError::MigrationHashMismatch {
				name,
				expected: _,
				actual: _,
				expected_hex,
				actual_hex,
				fragment,
			} => Diagnostic {
				code: "CA_049".to_string(),
				rql: None,
				message: format!(
					"migration `{}` content has changed since registration: was {}, now {}",
					name, expected_hex, actual_hex
				),
				fragment,
				label: Some("modified migration".to_string()),
				help: Some(
					"applied migrations are immutable; revert the change or register a new migration with a different name"
						.to_string(),
				),
				column: None,
				notes: vec![
					"the registered hash in the catalog does not match the hash of the supplied body+rollback".to_string(),
				],
				cause: None,
				operator_chain: None,
			},

			CatalogError::ColumnAlreadyExists {
				kind,
				namespace,
				name,
				column,
				fragment,
			} => {
				let kind_str = match kind {
					CatalogObjectKind::Table => "table",
					CatalogObjectKind::View => "view",
					_ => "object",
				};
				Diagnostic {
					code: "CA_005".to_string(),
					rql: None,
					message: format!(
						"column `{}` already exists in {} `{}::{}`",
						column, kind_str, namespace, name
					),
					fragment,
					label: Some("duplicate column definition".to_string()),
					help: Some("choose a different column name or drop the existing one first"
						.to_string()),
					column: None,
					notes: vec![],
					cause: None,
					operator_chain: None,
				}
			}

			CatalogError::DictionaryTypeMismatch {
				column,
				column_type,
				dictionary,
				dictionary_value_type,
				fragment,
			} => Diagnostic {
				code: "CA_008".to_string(),
				rql: None,
				message: format!(
					"column `{}` type `{}` does not match dictionary `{}` value type `{}`",
					column, column_type, dictionary, dictionary_value_type
				),
				fragment,
				label: Some("type mismatch".to_string()),
				help: Some(format!(
					"change the column type to `{}` to match the dictionary value type",
					dictionary_value_type
				)),
				column: None,
				notes: vec![],
				cause: None,
				operator_chain: None,
			},

			CatalogError::AutoIncrementInvalidType {
				column,
				ty,
				fragment,
			} => Diagnostic {
				code: "CA_006".to_string(),
				rql: None,
				message: format!("auto increment is not supported for type `{}`", ty),
				fragment,
				label: Some("invalid auto increment usage".to_string()),
				help: Some(format!(
					"auto increment is only supported for integer types (int1-16, uint1-16), column `{}` has type `{}`",
					column, ty
				)),
				column: None,
				notes: vec![],
				cause: None,
				operator_chain: None,
			},

			CatalogError::ColumnPropertyAlreadyExists {
				policy,
				column,
			} => Diagnostic {
				code: "CA_008".to_string(),
				rql: None,
				message: format!("policy `{:?}` already exists for column `{}`", policy, column),
				fragment: Fragment::None,
				label: Some("duplicate column policy".to_string()),
				help: Some("remove the existing policy first".to_string()),
				column: None,
				notes: vec![],
				cause: None,
				operator_chain: None,
			},

			CatalogError::AlreadyPendingInTransaction {
				kind,
				namespace,
				name,
				fragment,
			} => {
				let (code, message, label_str) = match kind {
					CatalogObjectKind::Namespace => (
						"CA_011",
						format!(
							"namespace `{}` already has pending changes in this transaction",
							namespace
						),
						"duplicate namespace modification in transaction",
					),
					CatalogObjectKind::Table => (
						"CA_012",
						format!(
							"table `{}::{}` already has pending changes in this transaction",
							namespace,
							name.as_deref().unwrap_or("")
						),
						"duplicate table modification in transaction",
					),
					CatalogObjectKind::View => (
						"CA_013",
						format!(
							"view `{}::{}` already has pending changes in this transaction",
							namespace,
							name.as_deref().unwrap_or("")
						),
						"duplicate view modification in transaction",
					),
					_ => (
						"CA_011",
						format!("{} already has pending changes in this transaction", kind),
						"duplicate modification in transaction",
					),
				};
				let kind_str = match kind {
					CatalogObjectKind::Namespace => "namespace",
					CatalogObjectKind::Table => "table",
					CatalogObjectKind::View => "view",
					_ => "object",
				};
				Diagnostic {
					code: code.to_string(),
					rql: None,
					message,
					fragment,
					label: Some(label_str.to_string()),
					help: Some(format!(
						"a {} can only be created, updated, or deleted once per transaction",
						kind_str
					)),
					column: None,
					notes: vec![
						"This usually indicates a programming error in transaction management"
							.to_string(),
						"Consider reviewing the transaction logic for duplicate operations"
							.to_string(),
					],
					cause: None,
					operator_chain: None,
				}
			}

			CatalogError::CannotUpdateDeleted {
				kind,
				namespace,
				name,
				fragment,
			} => {
				let (code, message, kind_str) = match kind {
					CatalogObjectKind::Namespace => (
						"CA_014",
						format!(
							"cannot update namespace `{}` as it is marked for deletion in this transaction",
							namespace
						),
						"namespace",
					),
					CatalogObjectKind::Table => (
						"CA_015",
						format!(
							"cannot update table `{}::{}` as it is marked for deletion in this transaction",
							namespace,
							name.as_deref().unwrap_or("")
						),
						"table",
					),
					CatalogObjectKind::View => (
						"CA_016",
						format!(
							"cannot update view `{}::{}` as it is marked for deletion in this transaction",
							namespace,
							name.as_deref().unwrap_or("")
						),
						"view",
					),
					_ => (
						"CA_014",
						format!(
							"cannot update {} as it is marked for deletion in this transaction",
							kind
						),
						"object",
					),
				};
				Diagnostic {
					code: code.to_string(),
					rql: None,
					message,
					fragment,
					label: Some(format!("attempted update on deleted {}", kind_str)),
					help: Some("remove the delete operation or skip the update".to_string()),
					column: None,
					notes: vec![format!(
						"A {} marked for deletion cannot be updated in the same transaction",
						kind_str
					)],
					cause: None,
					operator_chain: None,
				}
			}

			CatalogError::CannotDeleteAlreadyDeleted {
				kind,
				namespace,
				name,
				fragment,
			} => {
				let (code, message, kind_str) = match kind {
					CatalogObjectKind::Namespace => (
						"CA_017",
						format!(
							"namespace `{}` is already marked for deletion in this transaction",
							namespace
						),
						"namespace",
					),
					CatalogObjectKind::Table => (
						"CA_018",
						format!(
							"table `{}::{}` is already marked for deletion in this transaction",
							namespace,
							name.as_deref().unwrap_or("")
						),
						"table",
					),
					CatalogObjectKind::View => (
						"CA_019",
						format!(
							"view `{}::{}` is already marked for deletion in this transaction",
							namespace,
							name.as_deref().unwrap_or("")
						),
						"view",
					),
					_ => (
						"CA_017",
						format!("{} is already marked for deletion in this transaction", kind),
						"object",
					),
				};
				Diagnostic {
					code: code.to_string(),
					rql: None,
					message,
					fragment,
					label: Some(format!("duplicate {} deletion", kind_str)),
					help: Some("remove the duplicate delete operation".to_string()),
					column: None,
					notes: vec![format!("A {} can only be deleted once per transaction", kind_str)],
					cause: None,
					operator_chain: None,
				}
			}

			CatalogError::PrimaryKeyEmpty {
				fragment,
			} => Diagnostic {
				code: "CA_020".to_string(),
				rql: None,
				message: "primary key must contain at least one column".to_string(),
				fragment,
				label: Some("empty primary key definition".to_string()),
				help: Some("specify at least one column for the primary key".to_string()),
				column: None,
				notes: vec![],
				cause: None,
				operator_chain: None,
			},

			CatalogError::PrimaryKeyColumnNotFound {
				fragment,
				column_id,
			} => Diagnostic {
				code: "CA_021".to_string(),
				rql: None,
				message: format!("column with ID {} not found for primary key", column_id),
				fragment,
				label: Some("invalid column reference in primary key".to_string()),
				help: Some(
					"ensure all columns referenced in the primary key exist in the table or view"
						.to_string(),
				),
				column: None,
				notes: vec![],
				cause: None,
				operator_chain: None,
			},

			CatalogError::SubscriptionAlreadyExists {
				fragment,
				name,
			} => Diagnostic {
				code: "CA_010".to_string(),
				rql: None,
				message: format!("subscription `{}` already exists", name),
				fragment,
				label: Some("duplicate subscription definition".to_string()),
				help: Some(
					"choose a different name or close the existing subscription first".to_string()
				),
				column: None,
				notes: vec![],
				cause: None,
				operator_chain: None,
			},

			CatalogError::SubscriptionNotFound {
				fragment,
				name,
			} => Diagnostic {
				code: "CA_011".to_string(),
				rql: None,
				message: format!("subscription `{}` not found", name),
				fragment,
				label: Some("unknown subscription reference".to_string()),
				help: Some(
					"ensure the subscription exists or create it first using `CREATE SUBSCRIPTION`"
						.to_string(),
				),
				column: None,
				notes: vec![],
				cause: None,
				operator_chain: None,
			},

			CatalogError::ColumnNotFound {
				kind,
				namespace,
				name,
				column,
				fragment,
			} => {
				let kind_str = match kind {
					CatalogObjectKind::Table => "table",
					CatalogObjectKind::View => "view",
					_ => "object",
				};
				Diagnostic {
					code: "CA_039".to_string(),
					rql: None,
					message: format!(
						"column `{}` not found in {} `{}`.`{}`",
						column, kind_str, namespace, name
					),
					fragment,
					label: Some("unknown column reference".to_string()),
					help: Some("ensure the column exists in the table".to_string()),
					column: None,
					notes: vec![],
					cause: None,
					operator_chain: None,
				}
			}

			CatalogError::ConfigStorageKeyNotFound(key) => Diagnostic {
				code: "CA_050".to_string(),
				rql: None,
				message: format!("unknown config key `{}`", key),
				fragment: Fragment::None,
				label: Some("unknown config key".to_string()),
				help: Some("query system.config to see all registered configuration keys".to_string()),
				column: None,
				notes: vec![],
				cause: None,
				operator_chain: None,
			},

			CatalogError::ConfigValueInvalid(key) => Diagnostic {
				code: "CA_051".to_string(),
				rql: None,
				message: format!("config value for key `{}` cannot be none", key),
				fragment: Fragment::None,
				label: Some("invalid config value".to_string()),
				help: Some("provide a concrete value such as an integer or boolean".to_string()),
				column: None,
				notes: vec![],
				cause: None,
				operator_chain: None,
			},

			CatalogError::ConfigTypeMismatch {
				key,
				expected,
				actual,
			} => {
				let expected_str =
					expected.iter().map(|t| format!("`{:?}`", t)).collect::<Vec<_>>().join(", ");
				Diagnostic {
					code: "CA_052".to_string(),
					rql: None,
					message: format!(
						"config value for key `{}` must be of type {}, got `{}`",
						key, expected_str, actual
					),
					fragment: Fragment::None,
					label: Some("type mismatch".to_string()),
					help: Some(format!("provide a value of type {}", expected_str)),
					column: None,
					notes: vec![],
					cause: None,
					operator_chain: None,
				}
			}

			CatalogError::ConfigInvalidValue {
				key,
				reason,
			} => Diagnostic {
				code: "CA_053".to_string(),
				rql: None,
				message: format!("config value for key `{}` is invalid: {}", key, reason),
				fragment: Fragment::None,
				label: Some("invalid config value".to_string()),
				help: None,
				column: None,
				notes: vec![],
				cause: None,
				operator_chain: None,
			},

			CatalogError::InUse {
				kind,
				namespace,
				name,
				dependents,
				fragment,
			} => {
				let (code, label, help) = match kind {
					CatalogObjectKind::Dictionary => (
						"CA_032",
						"dictionary is in use",
						"drop or alter the dependent columns first, or use CASCADE to automatically drop all dependents",
					),
					CatalogObjectKind::Enum => (
						"CA_033",
						"enum is in use",
						"drop or alter the dependent columns first, or use CASCADE to automatically drop all dependents",
					),
					CatalogObjectKind::Event => (
						"CA_033",
						"event is in use",
						"drop or alter the dependent handlers first, or use CASCADE to automatically drop all dependents",
					),
					CatalogObjectKind::Namespace => (
						"CA_034",
						"namespace contains referenced objects",
						"drop or alter the dependent columns in other namespaces first",
					),
					CatalogObjectKind::Table => (
						"CA_035",
						"table is in use",
						"drop or alter the dependent flows first, or use CASCADE to automatically drop all dependents",
					),
					CatalogObjectKind::View => (
						"CA_036",
						"view is in use",
						"drop or alter the dependent flows first, or use CASCADE to automatically drop all dependents",
					),
					CatalogObjectKind::Flow => (
						"CA_037",
						"flow is in use",
						"drop or alter the dependent flows first, or use CASCADE to automatically drop all dependents",
					),
					CatalogObjectKind::RingBuffer => (
						"CA_038",
						"ring buffer is in use",
						"drop or alter the dependent flows first, or use CASCADE to automatically drop all dependents",
					),
					_ => (
						"CA_032",
						"object is in use",
						"drop or alter the dependents first, or use CASCADE to automatically drop all dependents",
					),
				};
				let message = if matches!(kind, CatalogObjectKind::Namespace) {
					format!(
						"cannot drop namespace '{}' because it contains objects referenced from other namespaces: {}",
						namespace, dependents
					)
				} else {
					format!(
						"cannot drop {} '{}::{}' because it is referenced by: {}",
						kind,
						namespace,
						name.as_deref().unwrap_or(""),
						dependents
					)
				};
				Diagnostic {
					code: code.to_string(),
					rql: None,
					message,
					fragment,
					label: Some(label.to_string()),
					help: Some(help.to_string()),
					column: None,
					notes: vec![],
					cause: None,
					operator_chain: None,
				}
			}

			CatalogError::CannotDropEphemeralProcedure {
				kind,
				name,
				fragment,
			} => Diagnostic {
				code: "CA_084".to_string(),
				rql: None,
				message: format!(
					"cannot drop {} procedure `{}`: native/FFI/WASM procedures are managed by the runtime registry, not DDL",
					kind, name
				),
				fragment,
				label: Some("cannot drop system-managed procedure".to_string()),
				help: Some(
					"native, FFI, and WASM procedures are repopulated on every boot from the runtime registry — remove them from the binary or plugin directory instead"
						.to_string(),
				),
				column: None,
				notes: vec![],
				cause: None,
				operator_chain: None,
			},

			CatalogError::CannotRegisterPersistentAsEphemeral {
				kind,
			} => Diagnostic {
				code: "CA_085".to_string(),
				rql: None,
				message: format!(
					"cannot register {} procedure as ephemeral: only Native/FFI/WASM variants are accepted",
					kind
				),
				fragment: Fragment::None,
				label: Some("variant not accepted by ephemeral registrar".to_string()),
				help: Some(
					"persistent Rql/Test procedures must be created via `CREATE PROCEDURE`, not via the ephemeral registrar"
						.to_string(),
				),
				column: None,
				notes: vec![],
				cause: None,
				operator_chain: None,
			},

			CatalogError::PolicyInvalidOperation {
				target_type,
				operation,
				valid,
				policy_name,
			} => {
				let where_clause = match policy_name {
					Some(name) => format!(" in policy `{}`", name),
					None => String::new(),
				};
				let help = if valid.is_empty() {
					format!(
						"{} policies currently have no enforceable operations — remove this policy or add an enforcement call site for it",
						target_type
					)
				} else {
					format!("valid operations for {} policy: {}", target_type, valid.join(", "))
				};
				Diagnostic {
					code: "CA_086".to_string(),
					rql: None,
					message: format!(
						"unknown operation `{}` for {} policy{}",
						operation, target_type, where_clause
					),
					fragment: Fragment::None,
					label: Some("unknown policy operation".to_string()),
					help: Some(help),
					column: None,
					notes: vec![
						"operation names are matched by exact string equality at enforcement time; unknown keys are silently skipped and effectively dead code"
							.to_string(),
					],
					cause: None,
					operator_chain: None,
				}
			}

			CatalogError::InvalidBindingConfig {
				reason,
				fragment,
			} => Diagnostic {
				code: "CA_089".to_string(),
				rql: None,
				message: format!("invalid binding config: {}", reason),
				fragment,
				label: Some("invalid binding config".to_string()),
				help: Some("check the protocol's required WITH keys and value constraints".to_string()),
				column: None,
				notes: vec![],
				cause: None,
				operator_chain: None,
			},
		}
	}
}

impl From<CatalogError> for Error {
	fn from(err: CatalogError) -> Self {
		Error(Box::new(err.into_diagnostic()))
	}
}

#[derive(Debug, thiserror::Error)]
pub enum CatalogChangeError {
	#[error("failed to decode {kind:?} key while applying replicated catalog change")]
	KeyDecodeFailed {
		kind: KeyKind,
	},

	#[error("unrecognized key kind (raw: {raw:?})")]
	UnrecognizedKey {
		raw: Vec<u8>,
	},
}

impl IntoDiagnostic for CatalogChangeError {
	fn into_diagnostic(self) -> Diagnostic {
		match self {
			CatalogChangeError::KeyDecodeFailed {
				kind,
			} => Diagnostic {
				code: "CA_070".to_string(),
				rql: None,
				message: format!("failed to decode {:?} key while applying replicated catalog change", kind),
				fragment: Fragment::None,
				label: Some("key decode failure during replication".to_string()),
				help: Some(
					"this indicates a protocol mismatch between primary and replica — ensure both nodes are running the same version".to_string(),
				),
				column: None,
				notes: vec![],
				cause: None,
				operator_chain: None,
			},
			CatalogChangeError::UnrecognizedKey {
				raw,
			} => Diagnostic {
				code: "CA_071".to_string(),
				rql: None,
				message: format!("unrecognized key kind (raw: {:?})", raw),
				fragment: Fragment::None,
				label: Some("unrecognized key kind during replication".to_string()),
				help: Some(
					"this indicates state inconsistency — ensure primary and replica are running the same version".to_string(),
				),
				column: None,
				notes: vec![],
				cause: None,
				operator_chain: None,
			},
		}
	}
}

impl From<CatalogChangeError> for Error {
	fn from(err: CatalogChangeError) -> Self {
		Error(Box::new(err.into_diagnostic()))
	}
}