reifydb-codec 0.10.0

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

//! Proves the tag namespace is actually single: the same value carried through the value codec, an
//! RBCF Any column, a row Any field, and a keycode key must present the same leading kind byte and
//! round-trip to the same value.

use std::sync::Arc;

use arrow_array::{LargeBinaryArray, RecordBatch};
use arrow_schema::Schema;
use reifydb_codec::{
	frame::{decode::decode_frames, encode::encode_frames, options::EncodeOptions},
	key::{deserializer::KeyDeserializer, serializer::KeySerializer},
	tag::ValueKind,
	value::encode_value,
};
use reifydb_value::value::{
	Value,
	blob::Blob,
	constraint::TypeConstraint,
	container::any_array::{self, any_array},
	date::Date,
	frame::frame::Frame,
	ordered_f64::OrderedF64,
	value_type::{
		ValueType,
		field::{from_field, named},
	},
};

fn cross_codec_values() -> Vec<Value> {
	vec![
		Value::none_of(ValueType::Duration),
		Value::none_of(ValueType::Option(Box::new(ValueType::Duration))),
		Value::Boolean(true),
		Value::Int4(-7),
		Value::Int8(i64::MAX),
		Value::Utf8("cross".to_string()),
		Value::Float8(OrderedF64::try_from(1.5).unwrap()),
		Value::Date(Date::from_days_since_epoch(19_000).unwrap()),
		Value::Blob(Blob::new(vec![9, 8, 7])),
	]
}

fn value_codec_kind_byte(value: &Value) -> u8 {
	encode_value(value).unwrap()[0]
}

fn key_kind_byte(value: &Value) -> u8 {
	let mut ser = KeySerializer::new();
	ser.extend_value(value);
	ser.finish().as_slice()[0]
}

#[test]
fn kind_byte_is_identical_across_codecs() {
	for value in cross_codec_values() {
		let expected = ValueKind::of_value(&value).byte();
		assert_eq!(value_codec_kind_byte(&value), expected, "value codec kind byte for {value:?}");
		assert_eq!(key_kind_byte(&value), expected, "keycode kind byte for {value:?}");
	}
}

#[test]
fn value_codec_and_rbcf_any_column_round_trip_identically() {
	let values: Vec<Value> = cross_codec_values().into_iter().map(|value| Value::List(vec![value])).collect();
	let (field, array) = named("c", ValueType::Any.into(), Arc::new(any_array(values.clone())));
	let frame = Frame::from(RecordBatch::try_new(Arc::new(Schema::new(vec![field])), vec![array]).unwrap());
	let bytes = encode_frames(&[frame], &EncodeOptions::fast()).unwrap();
	let decoded = decode_frames(&bytes).unwrap();
	let batch = &decoded[0].batch;
	let field = batch.schema_ref().field(0);
	assert_eq!(from_field(field).unwrap().value_type, Some(ValueType::Any), "expected any column, got {field:?}");
	let container = batch.column(0).as_any().downcast_ref::<LargeBinaryArray>().expect("expected any column");
	for (expected, actual) in values.iter().zip(any_array::values(container).iter()) {
		assert_eq!(expected, actual, "wrapped value and its none inner type through RBCF");
	}
}

#[test]
fn value_codec_and_row_any_field_round_trip_identically() {
	use reifydb_codec::row::shape::{RowFamily, RowShape, RowShapeField};

	let shape = RowShape::new(RowFamily::Table, vec![RowShapeField::unconstrained("v", ValueType::Any)]);
	for value in cross_codec_values() {
		let mut row = shape.allocate_table();
		let wrapped = Value::Any(Box::new(value.clone()));
		shape.set_value(&mut row, 0, &wrapped);
		let read = shape.get_value(&row, 0);
		match (&wrapped, &read) {
			(Value::Any(l), Value::Any(r)) => match (l.as_ref(), r.as_ref()) {
				(
					Value::None {
						inner: li,
					},
					Value::None {
						inner: ri,
					},
				) => assert_eq!(li, ri, "none inner type through row any field"),
				(l, r) => assert_eq!(l, r),
			},
			_ => panic!("expected any value back, got {read:?}"),
		}
	}
}

#[test]
fn keycode_round_trips_the_same_values() {
	for value in cross_codec_values() {
		let mut ser = KeySerializer::new();
		ser.extend_value(&value);
		let key = ser.finish();
		let mut de = KeyDeserializer::from_bytes(&key);
		let read = de.read_value().unwrap();
		match (&value, &read) {
			(
				Value::None {
					inner: l,
				},
				Value::None {
					inner: r,
				},
			) => assert_eq!(l, r, "none inner type through keycode"),
			(l, r) => assert_eq!(l, r),
		}
		assert!(de.is_empty());
	}
}

#[test]
fn constraint_extern_c_round_trips_base_types() {
	use reifydb_codec::constraint::{decode_type_constraint, encode_type_constraint};

	let types = [
		ValueType::Boolean,
		ValueType::Int4,
		ValueType::Utf8,
		ValueType::Option(Box::new(ValueType::Duration)),
		ValueType::Option(Box::new(ValueType::Option(Box::new(ValueType::Int8)))),
	];
	for ty in types {
		let tc = TypeConstraint::unconstrained(ty.clone());
		let extern_c = encode_type_constraint(&tc).unwrap();
		let back = decode_type_constraint(&extern_c).unwrap();
		assert_eq!(back.get_type(), ty, "constraint base type must round-trip");
	}
}