reifydb_transaction/

change.rs

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

use OperationType::Delete;
use reifydb_core::{
	encoded::row::EncodedRow,
	interface::catalog::{
		authentication::{Authentication, AuthenticationId},
		binding::Binding,
		config::{Config, ConfigKey},
		dictionary::Dictionary,
		flow::{Flow, FlowId, FlowNodeId},
		handler::Handler,
		id::{
			BindingId, HandlerId, MigrationEventId, MigrationId, NamespaceId, ProcedureId, RingBufferId,
			SeriesId, SinkId, SourceId, TableId, TestId, ViewId,
		},
		identity::{GrantedRole, Identity, Role, RoleId},
		migration::{Migration, MigrationEvent},
		namespace::Namespace,
		policy::{Policy, PolicyId},
		procedure::Procedure,
		ringbuffer::RingBuffer,
		series::Series,
		shape::ShapeId,
		sink::Sink,
		source::Source,
		sumtype::SumType,
		table::Table,
		test::Test,
		view::View,
	},
	row::Ttl,
};
use reifydb_type::value::{dictionary::DictionaryId, identity::IdentityId, row_number::RowNumber, sumtype::SumTypeId};

use crate::TransactionId;

pub trait TransactionalChanges:
	TransactionalBindingChanges
	+ TransactionalDictionaryChanges
	+ TransactionalFlowChanges
	+ TransactionalHandlerChanges
	+ TransactionalMigrationChanges
	+ TransactionalNamespaceChanges
	+ TransactionalProcedureChanges
	+ TransactionalRingBufferChanges
	+ TransactionalRoleChanges
	+ TransactionalPolicyChanges
	+ TransactionalSeriesChanges
	+ TransactionalSinkChanges
	+ TransactionalSourceChanges
	+ TransactionalSumTypeChanges
	+ TransactionalTableChanges
	+ TransactionalTestChanges
	+ TransactionalAuthenticationChanges
	+ TransactionalIdentityChanges
	+ TransactionalGrantedRoleChanges
	+ TransactionalViewChanges
	+ TransactionalConfigChanges
	+ TransactionalRowTtlChanges
	+ TransactionalOperatorTtlChanges
{
}

pub trait TransactionalBindingChanges {
	fn find_binding(&self, id: BindingId) -> Option<&Binding>;

	fn find_binding_by_name(&self, namespace: NamespaceId, name: &str) -> Option<&Binding>;

	fn is_binding_deleted(&self, id: BindingId) -> bool;

	fn is_binding_deleted_by_name(&self, namespace: NamespaceId, name: &str) -> bool;
}

pub trait TransactionalRowTtlChanges {
	fn find_row_ttl(&self, shape: ShapeId) -> Option<&Ttl>;

	fn is_row_ttl_deleted(&self, shape: ShapeId) -> bool;
}

pub trait TransactionalOperatorTtlChanges {
	fn find_operator_ttl(&self, node: FlowNodeId) -> Option<&Ttl>;

	fn is_operator_ttl_deleted(&self, node: FlowNodeId) -> bool;
}

pub trait TransactionalConfigChanges {
	fn find_config(&self, key: ConfigKey) -> Option<&Config>;
}

pub trait TransactionalDictionaryChanges {
	fn find_dictionary(&self, id: DictionaryId) -> Option<&Dictionary>;

	fn find_dictionary_by_name(&self, namespace: NamespaceId, name: &str) -> Option<&Dictionary>;

	fn is_dictionary_deleted(&self, id: DictionaryId) -> bool;

	fn is_dictionary_deleted_by_name(&self, namespace: NamespaceId, name: &str) -> bool;
}

pub trait TransactionalNamespaceChanges {
	fn find_namespace(&self, id: NamespaceId) -> Option<&Namespace>;

	fn find_namespace_by_name(&self, name: &str) -> Option<&Namespace>;

	fn is_namespace_deleted(&self, id: NamespaceId) -> bool;

	fn is_namespace_deleted_by_name(&self, name: &str) -> bool;
}

pub trait TransactionalFlowChanges {
	fn find_flow(&self, id: FlowId) -> Option<&Flow>;

	fn find_flow_by_name(&self, namespace: NamespaceId, name: &str) -> Option<&Flow>;

	fn is_flow_deleted(&self, id: FlowId) -> bool;

	fn is_flow_deleted_by_name(&self, namespace: NamespaceId, name: &str) -> bool;
}

pub trait TransactionalTableChanges {
	fn find_table(&self, id: TableId) -> Option<&Table>;

	fn find_table_by_name(&self, namespace: NamespaceId, name: &str) -> Option<&Table>;

	fn is_table_deleted(&self, id: TableId) -> bool;

	fn is_table_deleted_by_name(&self, namespace: NamespaceId, name: &str) -> bool;
}

pub trait TransactionalProcedureChanges {
	fn find_procedure(&self, id: ProcedureId) -> Option<&Procedure>;

	fn find_procedure_by_name(&self, namespace: NamespaceId, name: &str) -> Option<&Procedure>;

	fn is_procedure_deleted(&self, id: ProcedureId) -> bool;

	fn is_procedure_deleted_by_name(&self, namespace: NamespaceId, name: &str) -> bool;
}

pub trait TransactionalTestChanges {
	fn find_test(&self, id: TestId) -> Option<&Test>;

	fn find_test_by_name(&self, namespace: NamespaceId, name: &str) -> Option<&Test>;

	fn is_test_deleted(&self, id: TestId) -> bool;

	fn is_test_deleted_by_name(&self, namespace: NamespaceId, name: &str) -> bool;
}

pub trait TransactionalRingBufferChanges {
	fn find_ringbuffer(&self, id: RingBufferId) -> Option<&RingBuffer>;

	fn find_ringbuffer_by_name(&self, namespace: NamespaceId, name: &str) -> Option<&RingBuffer>;

	fn is_ringbuffer_deleted(&self, id: RingBufferId) -> bool;

	fn is_ringbuffer_deleted_by_name(&self, namespace: NamespaceId, name: &str) -> bool;
}

pub trait TransactionalSeriesChanges {
	fn find_series(&self, id: SeriesId) -> Option<&Series>;

	fn find_series_by_name(&self, namespace: NamespaceId, name: &str) -> Option<&Series>;

	fn is_series_deleted(&self, id: SeriesId) -> bool;

	fn is_series_deleted_by_name(&self, namespace: NamespaceId, name: &str) -> bool;
}

pub trait TransactionalViewChanges {
	fn find_view(&self, id: ViewId) -> Option<&View>;

	fn find_view_by_name(&self, namespace: NamespaceId, name: &str) -> Option<&View>;

	fn is_view_deleted(&self, id: ViewId) -> bool;

	fn is_view_deleted_by_name(&self, namespace: NamespaceId, name: &str) -> bool;
}

pub trait TransactionalSumTypeChanges {
	fn find_sumtype(&self, id: SumTypeId) -> Option<&SumType>;

	fn find_sumtype_by_name(&self, namespace: NamespaceId, name: &str) -> Option<&SumType>;

	fn is_sumtype_deleted(&self, id: SumTypeId) -> bool;

	fn is_sumtype_deleted_by_name(&self, namespace: NamespaceId, name: &str) -> bool;
}

pub trait TransactionalHandlerChanges {
	fn find_handler_by_id(&self, id: HandlerId) -> Option<&Handler>;

	fn find_handler_by_name(&self, namespace: NamespaceId, name: &str) -> Option<&Handler>;

	fn is_handler_deleted(&self, id: HandlerId) -> bool;

	fn is_handler_deleted_by_name(&self, namespace: NamespaceId, name: &str) -> bool;
}

pub trait TransactionalIdentityChanges {
	fn find_identity(&self, id: IdentityId) -> Option<&Identity>;

	fn find_identity_by_name(&self, name: &str) -> Option<&Identity>;

	fn is_identity_deleted(&self, id: IdentityId) -> bool;

	fn is_identity_deleted_by_name(&self, name: &str) -> bool;
}

pub trait TransactionalRoleChanges {
	fn find_role(&self, id: RoleId) -> Option<&Role>;

	fn find_role_by_name(&self, name: &str) -> Option<&Role>;

	fn is_role_deleted(&self, id: RoleId) -> bool;

	fn is_role_deleted_by_name(&self, name: &str) -> bool;
}

pub trait TransactionalAuthenticationChanges {
	fn find_authentication(&self, id: AuthenticationId) -> Option<&Authentication>;

	fn find_authentication_by_identity_and_method(
		&self,
		identity: IdentityId,
		method: &str,
	) -> Option<&Authentication>;

	fn is_authentication_deleted(&self, id: AuthenticationId) -> bool;

	fn is_authentication_deleted_by_identity_and_method(&self, identity: IdentityId, method: &str) -> bool;
}

pub trait TransactionalGrantedRoleChanges {
	fn find_granted_role(&self, identity: IdentityId, role: RoleId) -> Option<&GrantedRole>;

	fn find_granted_roles_for_identity(&self, identity: IdentityId) -> Vec<&GrantedRole>;

	fn is_granted_role_deleted(&self, identity: IdentityId, role: RoleId) -> bool;
}

pub trait TransactionalPolicyChanges {
	fn find_policy(&self, id: PolicyId) -> Option<&Policy>;

	fn find_policy_by_name(&self, name: &str) -> Option<&Policy>;

	fn is_policy_deleted(&self, id: PolicyId) -> bool;

	fn is_policy_deleted_by_name(&self, name: &str) -> bool;
}

pub trait TransactionalSourceChanges {
	fn find_source(&self, id: SourceId) -> Option<&Source>;

	fn find_source_by_name(&self, namespace: NamespaceId, name: &str) -> Option<&Source>;

	fn is_source_deleted(&self, id: SourceId) -> bool;

	fn is_source_deleted_by_name(&self, namespace: NamespaceId, name: &str) -> bool;
}

pub trait TransactionalSinkChanges {
	fn find_sink(&self, id: SinkId) -> Option<&Sink>;

	fn find_sink_by_name(&self, namespace: NamespaceId, name: &str) -> Option<&Sink>;

	fn is_sink_deleted(&self, id: SinkId) -> bool;

	fn is_sink_deleted_by_name(&self, namespace: NamespaceId, name: &str) -> bool;
}

pub trait TransactionalMigrationChanges {
	fn find_migration(&self, id: MigrationId) -> Option<&Migration>;

	fn find_migration_by_name(&self, name: &str) -> Option<&Migration>;

	fn is_migration_deleted(&self, id: MigrationId) -> bool;

	fn is_migration_deleted_by_name(&self, name: &str) -> bool;
}

#[derive(Default, Debug, Clone)]
pub struct TransactionalCatalogChanges {
	pub txn_id: TransactionId,

	pub config: Vec<Change<Config>>,

	pub binding: Vec<Change<Binding>>,

	pub dictionary: Vec<Change<Dictionary>>,

	pub flow: Vec<Change<Flow>>,

	pub handler: Vec<Change<Handler>>,

	pub migration: Vec<Change<Migration>>,

	pub migration_event: Vec<Change<MigrationEvent>>,

	pub namespace: Vec<Change<Namespace>>,

	pub procedure: Vec<Change<Procedure>>,

	pub ringbuffer: Vec<Change<RingBuffer>>,

	pub series: Vec<Change<Series>>,

	pub sink: Vec<Change<Sink>>,

	pub source: Vec<Change<Source>>,
	pub sumtype: Vec<Change<SumType>>,

	pub test: Vec<Change<Test>>,

	pub table: Vec<Change<Table>>,

	pub identity: Vec<Change<Identity>>,

	pub authentication: Vec<Change<Authentication>>,

	pub role: Vec<Change<Role>>,

	pub granted_role: Vec<Change<GrantedRole>>,

	pub policy: Vec<Change<Policy>>,

	pub view: Vec<Change<View>>,

	pub row_ttl: Vec<Change<(ShapeId, Ttl)>>,

	pub operator_ttl: Vec<Change<(FlowNodeId, Ttl)>>,

	pub log: Vec<Operation>,
}

pub struct CatalogChangesSavepoint {
	binding_len: usize,
	config_len: usize,
	dictionary_len: usize,
	flow_len: usize,
	handler_len: usize,
	migration_len: usize,
	migration_event_len: usize,
	namespace_len: usize,
	procedure_len: usize,
	ringbuffer_len: usize,
	series_len: usize,
	sink_len: usize,
	source_len: usize,
	sumtype_len: usize,
	test_len: usize,
	table_len: usize,
	identity_len: usize,
	authentication_len: usize,
	role_len: usize,
	granted_role_len: usize,
	policy_len: usize,
	view_len: usize,
	row_ttl_len: usize,
	operator_ttl_len: usize,
	log_len: usize,
}

impl TransactionalCatalogChanges {
	pub fn savepoint(&self) -> CatalogChangesSavepoint {
		CatalogChangesSavepoint {
			binding_len: self.binding.len(),
			config_len: self.config.len(),
			dictionary_len: self.dictionary.len(),
			flow_len: self.flow.len(),
			handler_len: self.handler.len(),
			migration_len: self.migration.len(),
			migration_event_len: self.migration_event.len(),
			namespace_len: self.namespace.len(),
			procedure_len: self.procedure.len(),
			ringbuffer_len: self.ringbuffer.len(),
			series_len: self.series.len(),
			sink_len: self.sink.len(),
			source_len: self.source.len(),
			sumtype_len: self.sumtype.len(),
			test_len: self.test.len(),
			table_len: self.table.len(),
			identity_len: self.identity.len(),
			authentication_len: self.authentication.len(),
			role_len: self.role.len(),
			granted_role_len: self.granted_role.len(),
			policy_len: self.policy.len(),
			view_len: self.view.len(),
			row_ttl_len: self.row_ttl.len(),
			operator_ttl_len: self.operator_ttl.len(),
			log_len: self.log.len(),
		}
	}

	pub fn restore_savepoint(&mut self, sp: CatalogChangesSavepoint) {
		self.binding.truncate(sp.binding_len);
		self.config.truncate(sp.config_len);
		self.dictionary.truncate(sp.dictionary_len);
		self.flow.truncate(sp.flow_len);
		self.handler.truncate(sp.handler_len);
		self.migration.truncate(sp.migration_len);
		self.migration_event.truncate(sp.migration_event_len);
		self.namespace.truncate(sp.namespace_len);
		self.procedure.truncate(sp.procedure_len);
		self.ringbuffer.truncate(sp.ringbuffer_len);
		self.series.truncate(sp.series_len);
		self.sink.truncate(sp.sink_len);
		self.source.truncate(sp.source_len);
		self.sumtype.truncate(sp.sumtype_len);
		self.test.truncate(sp.test_len);
		self.table.truncate(sp.table_len);
		self.identity.truncate(sp.identity_len);
		self.authentication.truncate(sp.authentication_len);
		self.role.truncate(sp.role_len);
		self.granted_role.truncate(sp.granted_role_len);
		self.policy.truncate(sp.policy_len);
		self.view.truncate(sp.view_len);
		self.row_ttl.truncate(sp.row_ttl_len);
		self.operator_ttl.truncate(sp.operator_ttl_len);
		self.log.truncate(sp.log_len);
	}

	pub fn add_binding_change(&mut self, change: Change<Binding>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|b| b.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.binding.push(change);
		self.log.push(Operation::Binding {
			id,
			op,
		});
	}

	pub fn add_config_change(&mut self, change: Change<Config>) {
		let key = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|c| c.key)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.config.push(change);
		self.log.push(Operation::Config {
			key,
			op,
		});
	}

	pub fn add_dictionary_change(&mut self, change: Change<Dictionary>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|d| d.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.dictionary.push(change);
		self.log.push(Operation::Dictionary {
			id,
			op,
		});
	}

	pub fn add_flow_change(&mut self, change: Change<Flow>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|f| f.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.flow.push(change);
		self.log.push(Operation::Flow {
			id,
			op,
		});
	}

	pub fn add_namespace_change(&mut self, change: Change<Namespace>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|s| s.id())
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.namespace.push(change);
		self.log.push(Operation::Namespace {
			id,
			op,
		});
	}

	pub fn add_handler_change(&mut self, change: Change<Handler>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|h| h.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.handler.push(change);
		self.log.push(Operation::Handler {
			id,
			op,
		});
	}

	pub fn add_migration_change(&mut self, change: Change<Migration>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|m| m.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.migration.push(change);
		self.log.push(Operation::Migration {
			id,
			op,
		});
	}

	pub fn add_migration_event_change(&mut self, change: Change<MigrationEvent>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|e| e.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.migration_event.push(change);
		self.log.push(Operation::MigrationEvent {
			id,
			op,
		});
	}

	pub fn add_procedure_change(&mut self, change: Change<Procedure>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|p| p.id())
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.procedure.push(change);
		self.log.push(Operation::Procedure {
			id,
			op,
		});
	}

	pub fn add_test_change(&mut self, change: Change<Test>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|t| t.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.test.push(change);
		self.log.push(Operation::Test {
			id,
			op,
		});
	}

	pub fn add_ringbuffer_change(&mut self, change: Change<RingBuffer>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|rb| rb.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.ringbuffer.push(change);
		self.log.push(Operation::RingBuffer {
			id,
			op,
		});
	}

	pub fn add_series_change(&mut self, change: Change<Series>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|s| s.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.series.push(change);
		self.log.push(Operation::Series {
			id,
			op,
		});
	}

	pub fn add_sink_change(&mut self, change: Change<Sink>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|s| s.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.sink.push(change);
		self.log.push(Operation::Sink {
			id,
			op,
		});
	}

	pub fn add_source_change(&mut self, change: Change<Source>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|s| s.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.source.push(change);
		self.log.push(Operation::Source {
			id,
			op,
		});
	}

	pub fn add_table_change(&mut self, change: Change<Table>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|t| t.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.table.push(change);
		self.log.push(Operation::Table {
			id,
			op,
		});
	}

	pub fn add_view_change(&mut self, change: Change<View>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|v| v.id())
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.view.push(change);
		self.log.push(Operation::View {
			id,
			op,
		});
	}

	pub fn add_sumtype_change(&mut self, change: Change<SumType>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|s| s.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.sumtype.push(change);
		self.log.push(Operation::SumType {
			id,
			op,
		});
	}

	pub fn add_identity_change(&mut self, change: Change<Identity>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|u| u.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.identity.push(change);
		self.log.push(Operation::Identity {
			id,
			op,
		});
	}

	pub fn add_role_change(&mut self, change: Change<Role>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|r| r.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.role.push(change);
		self.log.push(Operation::Role {
			id,
			op,
		});
	}

	pub fn add_granted_role_change(&mut self, change: Change<GrantedRole>) {
		let identity = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|ur| ur.identity)
			.expect("Change must have either pre or post state");
		let role = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|ur| ur.role_id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.granted_role.push(change);
		self.log.push(Operation::GrantedRole {
			identity,
			role,
			op,
		});
	}

	pub fn add_authentication_change(&mut self, change: Change<Authentication>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|a| a.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.authentication.push(change);
		self.log.push(Operation::Authentication {
			id,
			op,
		});
	}

	pub fn add_policy_change(&mut self, change: Change<Policy>) {
		let id = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|p| p.id)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.policy.push(change);
		self.log.push(Operation::Policy {
			id,
			op,
		});
	}

	pub fn add_row_ttl_change(&mut self, change: Change<(ShapeId, Ttl)>) {
		let shape = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|(s, _)| *s)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.row_ttl.push(change);
		self.log.push(Operation::Ttl {
			shape,
			op,
		});
	}

	pub fn add_operator_ttl_change(&mut self, change: Change<(FlowNodeId, Ttl)>) {
		let node = change
			.post
			.as_ref()
			.or(change.pre.as_ref())
			.map(|(n, _)| *n)
			.expect("Change must have either pre or post state");
		let op = change.op;
		self.operator_ttl.push(change);
		self.log.push(Operation::OperatorTtl {
			node,
			op,
		});
	}
}

#[derive(Debug, Clone)]
pub struct Change<T> {
	pub pre: Option<T>,

	pub post: Option<T>,

	pub op: OperationType,
}

#[derive(Debug, Clone, Copy, PartialEq)]
pub enum OperationType {
	Create,
	Update,
	Delete,
}

#[derive(Debug, Clone)]
pub enum Operation {
	Binding {
		id: BindingId,
		op: OperationType,
	},
	Config {
		key: ConfigKey,
		op: OperationType,
	},
	Dictionary {
		id: DictionaryId,
		op: OperationType,
	},
	Flow {
		id: FlowId,
		op: OperationType,
	},
	Handler {
		id: HandlerId,
		op: OperationType,
	},
	Migration {
		id: MigrationId,
		op: OperationType,
	},
	MigrationEvent {
		id: MigrationEventId,
		op: OperationType,
	},
	Namespace {
		id: NamespaceId,
		op: OperationType,
	},
	Procedure {
		id: ProcedureId,
		op: OperationType,
	},
	RingBuffer {
		id: RingBufferId,
		op: OperationType,
	},
	Series {
		id: SeriesId,
		op: OperationType,
	},
	Sink {
		id: SinkId,
		op: OperationType,
	},
	Source {
		id: SourceId,
		op: OperationType,
	},
	SumType {
		id: SumTypeId,
		op: OperationType,
	},
	Test {
		id: TestId,
		op: OperationType,
	},
	Table {
		id: TableId,
		op: OperationType,
	},
	Identity {
		id: IdentityId,
		op: OperationType,
	},
	Authentication {
		id: AuthenticationId,
		op: OperationType,
	},
	Role {
		id: RoleId,
		op: OperationType,
	},
	GrantedRole {
		identity: IdentityId,
		role: RoleId,
		op: OperationType,
	},
	Policy {
		id: PolicyId,
		op: OperationType,
	},
	View {
		id: ViewId,
		op: OperationType,
	},
	Ttl {
		shape: ShapeId,
		op: OperationType,
	},
	OperatorTtl {
		node: FlowNodeId,
		op: OperationType,
	},
}

impl TransactionalCatalogChanges {
	pub fn new(txn_id: TransactionId) -> Self {
		Self {
			txn_id,
			binding: Vec::new(),
			config: Vec::new(),
			dictionary: Vec::new(),
			flow: Vec::new(),
			handler: Vec::new(),
			migration: Vec::new(),
			migration_event: Vec::new(),
			namespace: Vec::new(),
			procedure: Vec::new(),
			ringbuffer: Vec::new(),
			series: Vec::new(),
			sink: Vec::new(),
			source: Vec::new(),
			sumtype: Vec::new(),
			test: Vec::new(),
			table: Vec::new(),
			identity: Vec::new(),
			authentication: Vec::new(),
			role: Vec::new(),
			granted_role: Vec::new(),
			policy: Vec::new(),
			view: Vec::new(),
			row_ttl: Vec::new(),
			operator_ttl: Vec::new(),
			log: Vec::new(),
		}
	}

	pub fn table_exists(&self, id: TableId) -> bool {
		self.get_table(id).is_some()
	}

	pub fn get_table(&self, id: TableId) -> Option<&Table> {
		for change in self.table.iter().rev() {
			if let Some(table) = &change.post {
				if table.id == id {
					return Some(table);
				}
			} else if let Some(table) = &change.pre
				&& table.id == id && change.op == Delete
			{
				return None;
			}
		}
		None
	}

	pub fn view_exists(&self, id: ViewId) -> bool {
		self.get_view(id).is_some()
	}

	pub fn get_view(&self, id: ViewId) -> Option<&View> {
		for change in self.view.iter().rev() {
			if let Some(view) = &change.post {
				if view.id() == id {
					return Some(view);
				}
			} else if let Some(view) = &change.pre
				&& view.id() == id && change.op == Delete
			{
				return None;
			}
		}
		None
	}

	pub fn get_row_ttl(&self, shape: ShapeId) -> Option<&Ttl> {
		for change in self.row_ttl.iter().rev() {
			if let Some((s, ttl)) = &change.post {
				if *s == shape {
					return Some(ttl);
				}
			} else if let Some((s, _)) = &change.pre
				&& *s == shape && change.op == Delete
			{
				return None;
			}
		}
		None
	}

	pub fn get_pending_changes(&self) -> &[Operation] {
		&self.log
	}

	pub fn txn_id(&self) -> TransactionId {
		self.txn_id
	}

	pub fn namespace(&self) -> &[Change<Namespace>] {
		&self.namespace
	}

	pub fn table(&self) -> &[Change<Table>] {
		&self.table
	}

	pub fn view(&self) -> &[Change<View>] {
		&self.view
	}

	pub fn clear(&mut self) {
		self.binding.clear();
		self.config.clear();
		self.dictionary.clear();
		self.flow.clear();
		self.handler.clear();
		self.migration.clear();
		self.migration_event.clear();
		self.namespace.clear();
		self.procedure.clear();
		self.ringbuffer.clear();
		self.series.clear();
		self.sink.clear();
		self.source.clear();
		self.sumtype.clear();
		self.test.clear();
		self.table.clear();
		self.identity.clear();
		self.authentication.clear();
		self.role.clear();
		self.granted_role.clear();
		self.policy.clear();
		self.view.clear();
		self.log.clear();
	}
}

#[derive(Debug, Clone)]
pub struct TableRowInsertion {
	pub table_id: TableId,
	pub row_number: RowNumber,
	pub encoded: EncodedRow,
}

#[derive(Debug, Clone)]
pub enum RowChange {
	TableInsert(TableRowInsertion),
}