reifydb_catalog/store/retention_policy/

list.rs

// Copyright (c) reifydb.com 2025
// This file is licensed under the AGPL-3.0-or-later, see license.md file

use reifydb_core::{
	interface::{FlowNodeId, QueryTransaction, SourceId},
	key::{
		EncodableKey, OperatorRetentionPolicyKey, OperatorRetentionPolicyKeyRange, SourceRetentionPolicyKey,
		SourceRetentionPolicyKeyRange,
	},
	retention::RetentionPolicy,
};

use super::decode_retention_policy;
use crate::CatalogStore;

/// A source retention policy entry
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SourceRetentionPolicyEntry {
	pub source: SourceId,
	pub policy: RetentionPolicy,
}

/// An operator retention policy entry
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OperatorRetentionPolicyEntry {
	pub operator: FlowNodeId,
	pub policy: RetentionPolicy,
}

impl CatalogStore {
	/// List all retention policies for sources (tables, views, ring buffers)
	pub async fn list_source_retention_policies(
		rx: &mut impl QueryTransaction,
	) -> crate::Result<Vec<SourceRetentionPolicyEntry>> {
		let mut result = Vec::new();

		let batch = rx.range(SourceRetentionPolicyKeyRange::full_scan()).await?;

		for entry in batch.items {
			if let Some(key) = SourceRetentionPolicyKey::decode(&entry.key) {
				if let Some(policy) = decode_retention_policy(&entry.values) {
					result.push(SourceRetentionPolicyEntry {
						source: key.source,
						policy,
					});
				}
			}
		}

		Ok(result)
	}

	/// List all retention policies for operators
	pub async fn list_operator_retention_policies(
		rx: &mut impl QueryTransaction,
	) -> crate::Result<Vec<OperatorRetentionPolicyEntry>> {
		let mut result = Vec::new();

		let batch = rx.range(OperatorRetentionPolicyKeyRange::full_scan()).await?;

		for entry in batch.items {
			if let Some(key) = OperatorRetentionPolicyKey::decode(&entry.key) {
				if let Some(policy) = decode_retention_policy(&entry.values) {
					result.push(OperatorRetentionPolicyEntry {
						operator: key.operator,
						policy,
					});
				}
			}
		}

		Ok(result)
	}
}

#[cfg(test)]
mod tests {
	use reifydb_core::{
		interface::{RingBufferId, TableId, ViewId},
		retention::{CleanupMode, RetentionPolicy},
	};
	use reifydb_engine::test_utils::create_test_command_transaction;

	use super::*;
	use crate::store::retention_policy::create::{
		_create_operator_retention_policy, create_source_retention_policy,
	};

	#[tokio::test]
	async fn test_list_source_retention_policies_empty() {
		let mut txn = create_test_command_transaction().await;

		let policies = CatalogStore::list_source_retention_policies(&mut txn).await.unwrap();

		assert_eq!(policies.len(), 0);
	}

	#[tokio::test]
	async fn test_list_source_retention_policies_multiple() {
		let mut txn = create_test_command_transaction().await;

		// Create policies for different sources
		let table_source = SourceId::Table(TableId(1));
		let table_policy = RetentionPolicy::KeepVersions {
			count: 10,
			cleanup_mode: CleanupMode::Delete,
		};
		create_source_retention_policy(&mut txn, table_source, &table_policy).await.unwrap();

		let view_source = SourceId::View(ViewId(2));
		let view_policy = RetentionPolicy::KeepForever;
		create_source_retention_policy(&mut txn, view_source, &view_policy).await.unwrap();

		let ringbuffer_source = SourceId::RingBuffer(RingBufferId(3));
		let ringbuffer_policy = RetentionPolicy::KeepVersions {
			count: 50,
			cleanup_mode: CleanupMode::Drop,
		};
		create_source_retention_policy(&mut txn, ringbuffer_source, &ringbuffer_policy).await.unwrap();

		// List all policies
		let policies = CatalogStore::list_source_retention_policies(&mut txn).await.unwrap();

		assert_eq!(policies.len(), 3);

		// Verify each policy
		assert!(policies.iter().any(|p| p.source == table_source && p.policy == table_policy));
		assert!(policies.iter().any(|p| p.source == view_source && p.policy == view_policy));
		assert!(policies.iter().any(|p| p.source == ringbuffer_source && p.policy == ringbuffer_policy));
	}

	#[tokio::test]
	async fn test_list_operator_retention_policies_empty() {
		let mut txn = create_test_command_transaction().await;

		let policies = CatalogStore::list_operator_retention_policies(&mut txn).await.unwrap();

		assert_eq!(policies.len(), 0);
	}

	#[tokio::test]
	async fn test_list_operator_retention_policies_multiple() {
		let mut txn = create_test_command_transaction().await;

		// Create policies for different operators
		let operator1 = FlowNodeId(100);
		let policy1 = RetentionPolicy::KeepVersions {
			count: 5,
			cleanup_mode: CleanupMode::Delete,
		};
		_create_operator_retention_policy(&mut txn, operator1, &policy1).await.unwrap();

		let operator2 = FlowNodeId(200);
		let policy2 = RetentionPolicy::KeepForever;
		_create_operator_retention_policy(&mut txn, operator2, &policy2).await.unwrap();

		let operator3 = FlowNodeId(300);
		let policy3 = RetentionPolicy::KeepVersions {
			count: 3,
			cleanup_mode: CleanupMode::Drop,
		};
		_create_operator_retention_policy(&mut txn, operator3, &policy3).await.unwrap();

		// List all policies
		let policies = CatalogStore::list_operator_retention_policies(&mut txn).await.unwrap();

		assert_eq!(policies.len(), 3);

		// Verify each policy
		assert!(policies.iter().any(|p| p.operator == operator1 && p.policy == policy1));
		assert!(policies.iter().any(|p| p.operator == operator2 && p.policy == policy2));
		assert!(policies.iter().any(|p| p.operator == operator3 && p.policy == policy3));
	}

	#[tokio::test]
	async fn test_list_source_retention_policies_after_updates() {
		let mut txn = create_test_command_transaction().await;

		let source = SourceId::Table(TableId(42));

		// Create initial policy
		let policy1 = RetentionPolicy::KeepForever;
		create_source_retention_policy(&mut txn, source, &policy1).await.unwrap();

		let policies = CatalogStore::list_source_retention_policies(&mut txn).await.unwrap();
		assert_eq!(policies.len(), 1);
		assert_eq!(policies[0].policy, policy1);

		// Update policy
		let policy2 = RetentionPolicy::KeepVersions {
			count: 20,
			cleanup_mode: CleanupMode::Drop,
		};
		create_source_retention_policy(&mut txn, source, &policy2).await.unwrap();

		// Should still have only 1 entry (updated, not added)
		let policies = CatalogStore::list_source_retention_policies(&mut txn).await.unwrap();
		assert_eq!(policies.len(), 1);
		assert_eq!(policies[0].policy, policy2);
	}

	#[tokio::test]
	async fn test_list_operator_retention_policies_after_updates() {
		let mut txn = create_test_command_transaction().await;

		let operator = FlowNodeId(999);

		// Create initial policy
		let policy1 = RetentionPolicy::KeepVersions {
			count: 3,
			cleanup_mode: CleanupMode::Delete,
		};
		_create_operator_retention_policy(&mut txn, operator, &policy1).await.unwrap();

		let policies = CatalogStore::list_operator_retention_policies(&mut txn).await.unwrap();
		assert_eq!(policies.len(), 1);
		assert_eq!(policies[0].policy, policy1);

		// Update policy
		let policy2 = RetentionPolicy::KeepForever;
		_create_operator_retention_policy(&mut txn, operator, &policy2).await.unwrap();

		// Should still have only 1 entry (updated, not added)
		let policies = CatalogStore::list_operator_retention_policies(&mut txn).await.unwrap();
		assert_eq!(policies.len(), 1);
		assert_eq!(policies[0].policy, policy2);
	}
}