reifydb_core/interface/catalog/

flow.rs

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

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

use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Visitor};

use crate::{flow::dag::FlowDag, interface::catalog::id::NamespaceId};

#[repr(transparent)]
#[derive(Debug, Copy, Clone, PartialOrd, PartialEq, Ord, Eq, Hash)]
pub struct FlowId(pub u64);

impl Deref for FlowId {
	type Target = u64;

	fn deref(&self) -> &Self::Target {
		&self.0
	}
}

impl PartialEq<u64> for FlowId {
	fn eq(&self, other: &u64) -> bool {
		self.0.eq(other)
	}
}

impl From<FlowId> for u64 {
	fn from(value: FlowId) -> Self {
		value.0
	}
}

impl FlowId {
	#[inline]
	pub fn to_u64(self) -> u64 {
		self.0
	}
}

impl From<u64> for FlowId {
	fn from(value: u64) -> Self {
		Self(value)
	}
}

impl Serialize for FlowId {
	fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
	where
		S: Serializer,
	{
		serializer.serialize_u64(self.0)
	}
}

impl<'de> Deserialize<'de> for FlowId {
	fn deserialize<D>(deserializer: D) -> Result<FlowId, D::Error>
	where
		D: Deserializer<'de>,
	{
		struct U64Visitor;

		impl Visitor<'_> for U64Visitor {
			type Value = FlowId;

			fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
				formatter.write_str("an unsigned 64-bit number")
			}

			fn visit_u64<E>(self, value: u64) -> Result<Self::Value, E> {
				Ok(FlowId(value))
			}
		}

		deserializer.deserialize_u64(U64Visitor)
	}
}

#[repr(transparent)]
#[derive(Debug, Copy, Clone, PartialOrd, PartialEq, Ord, Eq, Hash)]
pub struct OperatorId(pub u64);

impl Display for OperatorId {
	fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
		Display::fmt(&self.0, f)
	}
}

impl Deref for OperatorId {
	type Target = u64;

	fn deref(&self) -> &Self::Target {
		&self.0
	}
}

impl PartialEq<u64> for OperatorId {
	fn eq(&self, other: &u64) -> bool {
		self.0.eq(other)
	}
}

impl From<OperatorId> for u64 {
	fn from(value: OperatorId) -> Self {
		value.0
	}
}

impl OperatorId {
	#[inline]
	pub fn to_u64(self) -> u64 {
		self.0
	}
}

impl From<&OperatorId> for OperatorId {
	fn from(value: &OperatorId) -> Self {
		*value
	}
}

impl From<u64> for OperatorId {
	fn from(value: u64) -> Self {
		Self(value)
	}
}

impl Serialize for OperatorId {
	fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
	where
		S: Serializer,
	{
		serializer.serialize_u64(self.0)
	}
}

impl<'de> Deserialize<'de> for OperatorId {
	fn deserialize<D>(deserializer: D) -> Result<OperatorId, D::Error>
	where
		D: Deserializer<'de>,
	{
		struct U64Visitor;

		impl Visitor<'_> for U64Visitor {
			type Value = OperatorId;

			fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
				formatter.write_str("an unsigned 64-bit number")
			}

			fn visit_u64<E>(self, value: u64) -> Result<Self::Value, E> {
				Ok(OperatorId(value))
			}
		}

		deserializer.deserialize_u64(U64Visitor)
	}
}

#[repr(transparent)]
#[derive(Debug, Copy, Clone, PartialOrd, PartialEq, Ord, Eq, Hash)]
pub struct FlowEdgeId(pub u64);

impl Deref for FlowEdgeId {
	type Target = u64;

	fn deref(&self) -> &Self::Target {
		&self.0
	}
}

impl PartialEq<u64> for FlowEdgeId {
	fn eq(&self, other: &u64) -> bool {
		self.0.eq(other)
	}
}

impl From<FlowEdgeId> for u64 {
	fn from(value: FlowEdgeId) -> Self {
		value.0
	}
}

impl From<u64> for FlowEdgeId {
	fn from(value: u64) -> Self {
		Self(value)
	}
}

impl Serialize for FlowEdgeId {
	fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
	where
		S: Serializer,
	{
		serializer.serialize_u64(self.0)
	}
}

impl<'de> Deserialize<'de> for FlowEdgeId {
	fn deserialize<D>(deserializer: D) -> Result<FlowEdgeId, D::Error>
	where
		D: Deserializer<'de>,
	{
		struct U64Visitor;

		impl Visitor<'_> for U64Visitor {
			type Value = FlowEdgeId;

			fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
				formatter.write_str("an unsigned 64-bit number")
			}

			fn visit_u64<E>(self, value: u64) -> Result<Self::Value, E> {
				Ok(FlowEdgeId(value))
			}
		}

		deserializer.deserialize_u64(U64Visitor)
	}
}

#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
pub enum FlowStatus {
	Active,
	Paused,
	Failed,
}

impl FlowStatus {
	pub fn to_u8(self) -> u8 {
		match self {
			FlowStatus::Active => 0,
			FlowStatus::Paused => 1,
			FlowStatus::Failed => 2,
		}
	}

	pub fn from_u8(value: u8) -> Self {
		match value {
			0 => FlowStatus::Active,
			1 => FlowStatus::Paused,
			2 => FlowStatus::Failed,
			_ => FlowStatus::Failed,
		}
	}
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Flow {
	pub id: FlowId,
	pub namespace: NamespaceId,
	pub name: String,
	pub status: FlowStatus,
}

#[derive(Debug, Clone, PartialEq)]
pub struct FlowEntry {
	pub flow: Flow,
	pub dag: FlowDag,
}

#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct FlowEdge {
	pub id: FlowEdgeId,
	pub flow: FlowId,
	pub source: OperatorId,
	pub target: OperatorId,
}

impl FlowEdge {
	pub fn new(
		id: impl Into<FlowEdgeId>,
		flow: impl Into<FlowId>,
		source: impl Into<OperatorId>,
		target: impl Into<OperatorId>,
	) -> Self {
		Self {
			id: id.into(),
			flow: flow.into(),
			source: source.into(),
			target: target.into(),
		}
	}
}