reifydb-sdk 0.9.3

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

use core::{slice, str};

use reifydb_codec::tag::ValueKind;
use reifydb_value::{
	reifydb_assertions,
	value::{date::Date, datetime::DateTime, diff_type::DiffType, duration::Duration, time::Time},
};

use crate::{
	common::extern_c::wire::{
		buffer::ExternCBuffer,
		columns::{ExternCColumn, ExternCColumns},
	},
	flow::extern_c::wire::change::{ExternCChange, ExternCDiff, ExternCOrigin},
};

#[derive(Clone, Copy)]
pub struct BorrowedChange<'a> {
	extern_c: &'a ExternCChange,
}

impl<'a> BorrowedChange<'a> {
	/// # Safety
	///
	/// `ptr` must be non-null and point to a valid `ExternCChange` whose backing
	/// buffers remain live for the lifetime `'a`.
	pub unsafe fn from_raw(ptr: *const ExternCChange) -> Self {
		reifydb_assertions! {
			assert!(!ptr.is_null(), "BorrowedChange::from_raw: null pointer");
		}
		Self {
			// SAFETY: the `from_raw` contract above makes `ptr` non-null and a live, initialized
			// `ExternCChange` that outlives `'a`.
			extern_c: unsafe { &*ptr },
		}
	}

	pub fn origin(&self) -> ExternCOrigin {
		self.extern_c.origin
	}

	pub fn version(&self) -> u64 {
		self.extern_c.version
	}

	pub fn changed_at_nanos(&self) -> u64 {
		self.extern_c.changed_at
	}

	pub fn diff_count(&self) -> usize {
		self.extern_c.diff_count
	}

	pub fn diffs(&self) -> impl Iterator<Item = BorrowedDiff<'a>> + 'a {
		let count = self.extern_c.diff_count;
		let base = self.extern_c.diffs;
		(0..count).map(move |i| {
			// SAFETY: `base` is the `diff_count`-element `ExternCDiff` array `marshal_change` wrote and
			// fully initialized; `i < count` keeps the offset inside it.
			let diff: &'a ExternCDiff = unsafe { &*base.add(i) };
			BorrowedDiff {
				extern_c: diff,
			}
		})
	}
}

#[derive(Clone, Copy)]
pub struct BorrowedDiff<'a> {
	extern_c: &'a ExternCDiff,
}

impl<'a> BorrowedDiff<'a> {
	pub fn kind(&self) -> DiffType {
		self.extern_c.diff_type
	}

	pub fn pre(&self) -> BorrowedColumns<'a> {
		BorrowedColumns {
			extern_c: &self.extern_c.pre,
		}
	}

	pub fn post(&self) -> BorrowedColumns<'a> {
		BorrowedColumns {
			extern_c: &self.extern_c.post,
		}
	}
}

#[derive(Clone, Copy)]
pub struct BorrowedColumns<'a> {
	extern_c: &'a ExternCColumns,
}

impl<'a> BorrowedColumns<'a> {
	/// # Safety
	/// - `ptr` must be non-null and point at a `ExternCColumns` whose buffer pointers are valid for at least `'a`.
	pub unsafe fn from_extern_c(ptr: *const ExternCColumns) -> Self {
		reifydb_assertions! {
			assert!(!ptr.is_null(), "BorrowedColumns::from_extern_c: null pointer");
		}
		Self {
			// SAFETY: the `from_extern_c` contract above makes `ptr` non-null and a live, initialized
			// `ExternCColumns` that outlives `'a`.
			extern_c: unsafe { &*ptr },
		}
	}

	pub fn row_count(&self) -> usize {
		self.extern_c.row_count
	}

	pub fn column_count(&self) -> usize {
		self.extern_c.column_count
	}

	pub fn is_empty(&self) -> bool {
		self.extern_c.row_count == 0 && self.extern_c.column_count == 0
	}

	pub fn row_numbers(&self) -> &'a [u64] {
		if self.extern_c.row_numbers.is_null() || self.extern_c.row_count == 0 {
			&[]
		} else {
			// SAFETY: `row_numbers` is non-null here and a non-empty system sidecar holds exactly
			// `row_count` entries (`Columns::with_system` asserts it); `RowNumber` is
			// `repr(transparent)` over `u64`.
			unsafe { slice::from_raw_parts(self.extern_c.row_numbers, self.extern_c.row_count) }
		}
	}

	pub fn time(&self) -> &'a [u64] {
		if self.extern_c.time.is_null() || self.extern_c.row_count == 0 {
			&[]
		} else {
			// SAFETY: `time` is non-null here and a non-empty system sidecar holds exactly `row_count`
			// entries (`Columns::with_system` asserts it); `DateTime` is `repr(transparent)` over
			// `u64`.
			unsafe { slice::from_raw_parts(self.extern_c.time, self.extern_c.row_count) }
		}
	}

	pub fn columns(&self) -> impl Iterator<Item = BorrowedColumn<'a>> + 'a {
		let count = self.extern_c.column_count;
		let base = self.extern_c.columns;
		(0..count).map(move |i| {
			// SAFETY: `base` is the `column_count`-element `ExternCColumn` array `marshal_columns` wrote
			// and fully initialized; `i < count` keeps the offset inside it.
			let col: &'a ExternCColumn = unsafe { &*base.add(i) };
			BorrowedColumn {
				extern_c: col,
			}
		})
	}

	pub fn column(&self, name: &str) -> Option<BorrowedColumn<'a>> {
		self.columns().find(|c| c.name() == name)
	}

	pub fn column_at_index(&self, idx: usize) -> Option<BorrowedColumn<'a>> {
		if idx >= self.extern_c.column_count {
			return None;
		}
		// SAFETY: `idx < column_count` was checked above, so the offset stays inside the initialized
		// `ExternCColumn` array.
		let col: &'a ExternCColumn = unsafe { &*self.extern_c.columns.add(idx) };
		Some(BorrowedColumn {
			extern_c: col,
		})
	}

	pub fn index_of(&self, name: &str) -> Option<usize> {
		self.columns().position(|c| c.name() == name)
	}
}

#[derive(Clone, Copy)]
pub struct BorrowedColumn<'a> {
	extern_c: &'a ExternCColumn,
}

impl<'a> BorrowedColumn<'a> {
	pub fn name(&self) -> &'a str {
		// SAFETY: `self.extern_c` came from a `BorrowedChange::from_raw` / `BorrowedColumns::from_extern_c`
		// caller, whose contract keeps every buffer this `ExternCColumn` describes live for `'a`.
		unsafe { read_buffer_str(&self.extern_c.name) }
	}

	pub fn type_code(&self) -> ValueKind {
		self.extern_c.data.type_code
	}

	pub fn row_count(&self) -> usize {
		self.extern_c.data.row_count
	}

	pub fn data_bytes(&self) -> &'a [u8] {
		// SAFETY: `self.extern_c` came from a `BorrowedChange::from_raw` / `BorrowedColumns::from_extern_c`
		// caller, whose contract keeps every buffer this `ExternCColumn` describes live for `'a`.
		unsafe { read_buffer(&self.extern_c.data.data) }
	}

	pub fn offsets(&self) -> &'a [u64] {
		let buf = &self.extern_c.data.offsets;
		if buf.ptr.is_null() || buf.len == 0 {
			&[]
		} else {
			let count = buf.len / core::mem::size_of::<u64>();
			// SAFETY: `buf.ptr` is non-null here and every offsets buffer is either a marshalled
			// `&[u64]` or an 8-aligned arena block; `count` floors `buf.len / 8`, so that many
			// initialized `u64` fit.
			unsafe { slice::from_raw_parts(buf.ptr as *const u64, count) }
		}
	}

	pub fn defined_bitvec(&self) -> &'a [u8] {
		// SAFETY: `self.extern_c` came from a `BorrowedChange::from_raw` / `BorrowedColumns::from_extern_c`
		// caller, whose contract keeps every buffer this `ExternCColumn` describes live for `'a`.
		unsafe { read_buffer(&self.extern_c.data.defined_bitvec) }
	}

	/// # Safety
	///
	/// The caller must ensure the column's underlying bytes are a valid,
	/// properly aligned array of `T` for the column's row count.
	pub unsafe fn as_slice<T: Copy>(&self) -> Option<&'a [T]> {
		let bytes = self.data_bytes();
		let count = self.row_count();
		let elem = core::mem::size_of::<T>();
		if elem == 0 || count.checked_mul(elem)? != bytes.len() {
			return None;
		}
		// SAFETY: `count * size_of::<T>() == bytes.len()` was checked above, and the caller's contract
		// makes those bytes an aligned, initialized `T` array.
		Some(unsafe { slice::from_raw_parts(bytes.as_ptr() as *const T, count) })
	}

	pub fn iter_str(&self) -> impl Iterator<Item = &'a str> + 'a {
		let data = self.data_bytes();
		let offsets = self.offsets();
		let row_count = self.row_count();
		let offsets_len = offsets.len();
		(0..row_count).map(move |i| {
			if i + 1 >= offsets_len {
				return "";
			}
			let start = offsets[i] as usize;
			let end = offsets[i + 1] as usize;
			if end > data.len() {
				return "";
			}
			str::from_utf8(&data[start..end]).unwrap_or("")
		})
	}

	pub fn iter_bytes(&self) -> impl Iterator<Item = &'a [u8]> + 'a {
		let data = self.data_bytes();
		let offsets = self.offsets();
		let row_count = self.row_count();
		let offsets_len = offsets.len();
		(0..row_count).map(move |i| {
			if i + 1 >= offsets_len {
				return &[][..];
			}
			let start = offsets[i] as usize;
			let end = offsets[i + 1] as usize;
			if end > data.len() {
				return &[][..];
			}
			&data[start..end]
		})
	}

	#[inline]
	pub fn is_defined_at(&self, index: usize) -> bool {
		let bv = self.defined_bitvec();
		if bv.is_empty() {
			return true;
		}
		match bv.get(index / 8) {
			Some(b) => (b >> (index % 8)) & 1 == 1,
			None => false,
		}
	}

	#[inline]
	pub fn utf8_at(&self, index: usize) -> Option<&'a str> {
		if self.type_code() != ValueKind::Utf8 || !self.is_defined_at(index) {
			return None;
		}
		let offsets = self.offsets();
		if index + 1 >= offsets.len() {
			return None;
		}
		let start = offsets[index] as usize;
		let end = offsets[index + 1] as usize;
		let data = self.data_bytes();
		if end > data.len() || start > end {
			return None;
		}
		str::from_utf8(&data[start..end]).ok()
	}

	#[inline]
	pub fn blob_at(&self, index: usize) -> Option<&'a [u8]> {
		if self.type_code() != ValueKind::Blob || !self.is_defined_at(index) {
			return None;
		}
		let offsets = self.offsets();
		if index + 1 >= offsets.len() {
			return None;
		}
		let start = offsets[index] as usize;
		let end = offsets[index + 1] as usize;
		let data = self.data_bytes();
		if end > data.len() || start > end {
			return None;
		}
		Some(&data[start..end])
	}

	#[inline]
	pub fn bool_at(&self, index: usize) -> Option<bool> {
		if self.type_code() != ValueKind::Boolean || !self.is_defined_at(index) {
			return None;
		}
		let bytes = self.data_bytes();
		let byte = bytes.get(index / 8).copied()?;
		Some((byte >> (index % 8)) & 1 == 1)
	}

	#[inline]
	pub fn u64_at(&self, index: usize) -> Option<u64> {
		if !self.is_defined_at(index) {
			return None;
		}
		match self.type_code() {
			// SAFETY: type_code Uint8 means the buffer is a marshalled &[u64]: aligned, initialized.
			ValueKind::Uint8 => unsafe { self.as_slice::<u64>()?.get(index).copied() },
			// SAFETY: type_code Uint4 means the buffer is a marshalled &[u32]: aligned, initialized.
			ValueKind::Uint4 => unsafe { self.as_slice::<u32>()?.get(index).copied().map(u64::from) },
			// SAFETY: type_code Uint2 means the buffer is a marshalled &[u16]: aligned, initialized.
			ValueKind::Uint2 => unsafe { self.as_slice::<u16>()?.get(index).copied().map(u64::from) },
			// SAFETY: type_code Uint1 means the buffer is a marshalled &[u8]: aligned, initialized.
			ValueKind::Uint1 => unsafe { self.as_slice::<u8>()?.get(index).copied().map(u64::from) },
			_ => None,
		}
	}

	#[inline]
	pub fn u32_at(&self, index: usize) -> Option<u32> {
		if !self.is_defined_at(index) {
			return None;
		}
		match self.type_code() {
			// SAFETY: type_code Uint4 means the buffer is a marshalled &[u32]: aligned, initialized.
			ValueKind::Uint4 => unsafe { self.as_slice::<u32>()?.get(index).copied() },
			// SAFETY: type_code Uint2 means the buffer is a marshalled &[u16]: aligned, initialized.
			ValueKind::Uint2 => unsafe { self.as_slice::<u16>()?.get(index).copied().map(u32::from) },
			// SAFETY: type_code Uint1 means the buffer is a marshalled &[u8]: aligned, initialized.
			ValueKind::Uint1 => unsafe { self.as_slice::<u8>()?.get(index).copied().map(u32::from) },
			_ => None,
		}
	}

	#[inline]
	pub fn u16_at(&self, index: usize) -> Option<u16> {
		if !self.is_defined_at(index) {
			return None;
		}
		match self.type_code() {
			// SAFETY: type_code Uint2 means the buffer is a marshalled &[u16]: aligned, initialized.
			ValueKind::Uint2 => unsafe { self.as_slice::<u16>()?.get(index).copied() },
			// SAFETY: type_code Uint1 means the buffer is a marshalled &[u8]: aligned, initialized.
			ValueKind::Uint1 => unsafe { self.as_slice::<u8>()?.get(index).copied().map(u16::from) },
			_ => None,
		}
	}

	#[inline]
	pub fn u8_at(&self, index: usize) -> Option<u8> {
		if self.type_code() != ValueKind::Uint1 || !self.is_defined_at(index) {
			return None;
		}
		// SAFETY: the Uint1 check above means the buffer is a marshalled &[u8]: aligned, initialized.
		unsafe { self.as_slice::<u8>()?.get(index).copied() }
	}

	#[inline]
	pub fn i64_at(&self, index: usize) -> Option<i64> {
		if !self.is_defined_at(index) {
			return None;
		}
		match self.type_code() {
			// SAFETY: type_code Int8 means the buffer is a marshalled &[i64]: aligned, initialized.
			ValueKind::Int8 => unsafe { self.as_slice::<i64>()?.get(index).copied() },
			// SAFETY: type_code Int4 means the buffer is a marshalled &[i32]: aligned, initialized.
			ValueKind::Int4 => unsafe { self.as_slice::<i32>()?.get(index).copied().map(i64::from) },
			// SAFETY: type_code Int2 means the buffer is a marshalled &[i16]: aligned, initialized.
			ValueKind::Int2 => unsafe { self.as_slice::<i16>()?.get(index).copied().map(i64::from) },
			// SAFETY: type_code Int1 means the buffer is a marshalled &[i8]: aligned, initialized.
			ValueKind::Int1 => unsafe { self.as_slice::<i8>()?.get(index).copied().map(i64::from) },
			_ => None,
		}
	}

	#[inline]
	pub fn i32_at(&self, index: usize) -> Option<i32> {
		if !self.is_defined_at(index) {
			return None;
		}
		match self.type_code() {
			// SAFETY: type_code Int4 means the buffer is a marshalled &[i32]: aligned, initialized.
			ValueKind::Int4 => unsafe { self.as_slice::<i32>()?.get(index).copied() },
			// SAFETY: type_code Int2 means the buffer is a marshalled &[i16]: aligned, initialized.
			ValueKind::Int2 => unsafe { self.as_slice::<i16>()?.get(index).copied().map(i32::from) },
			// SAFETY: type_code Int1 means the buffer is a marshalled &[i8]: aligned, initialized.
			ValueKind::Int1 => unsafe { self.as_slice::<i8>()?.get(index).copied().map(i32::from) },
			_ => None,
		}
	}

	#[inline]
	pub fn i16_at(&self, index: usize) -> Option<i16> {
		if !self.is_defined_at(index) {
			return None;
		}
		match self.type_code() {
			// SAFETY: type_code Int2 means the buffer is a marshalled &[i16]: aligned, initialized.
			ValueKind::Int2 => unsafe { self.as_slice::<i16>()?.get(index).copied() },
			// SAFETY: type_code Int1 means the buffer is a marshalled &[i8]: aligned, initialized.
			ValueKind::Int1 => unsafe { self.as_slice::<i8>()?.get(index).copied().map(i16::from) },
			_ => None,
		}
	}

	#[inline]
	pub fn i8_at(&self, index: usize) -> Option<i8> {
		if self.type_code() != ValueKind::Int1 || !self.is_defined_at(index) {
			return None;
		}
		// SAFETY: the Int1 check above means the buffer is a marshalled &[i8]: aligned, initialized.
		unsafe { self.as_slice::<i8>()?.get(index).copied() }
	}

	#[inline]
	pub fn u128_at(&self, index: usize) -> Option<u128> {
		if self.type_code() != ValueKind::Uint16 || !self.is_defined_at(index) {
			return None;
		}
		// SAFETY: the Uint16 check above means the buffer is a marshalled &[u128], so it carries u128's
		// 16-byte alignment and is initialized.
		unsafe { self.as_slice::<u128>()?.get(index).copied() }
	}

	#[inline]
	pub fn i128_at(&self, index: usize) -> Option<i128> {
		if self.type_code() != ValueKind::Int16 || !self.is_defined_at(index) {
			return None;
		}
		// SAFETY: the Int16 check above means the buffer is a marshalled &[i128], so it carries i128's
		// 16-byte alignment and is initialized.
		unsafe { self.as_slice::<i128>()?.get(index).copied() }
	}

	#[inline]
	pub fn f64_at(&self, index: usize) -> Option<f64> {
		if !self.is_defined_at(index) {
			return None;
		}
		match self.type_code() {
			// SAFETY: type_code Float8 means the buffer is a marshalled &[f64]: aligned, initialized.
			ValueKind::Float8 => unsafe { self.as_slice::<f64>()?.get(index).copied() },
			// SAFETY: type_code Float4 means the buffer is a marshalled &[f32]: aligned, initialized.
			ValueKind::Float4 => unsafe { self.as_slice::<f32>()?.get(index).copied().map(f64::from) },
			_ => None,
		}
	}

	#[inline]
	pub fn f32_at(&self, index: usize) -> Option<f32> {
		if self.type_code() != ValueKind::Float4 || !self.is_defined_at(index) {
			return None;
		}
		// SAFETY: the Float4 check above means the buffer is a marshalled &[f32]: aligned, initialized.
		unsafe { self.as_slice::<f32>()?.get(index).copied() }
	}

	#[inline]
	pub fn date_at(&self, index: usize) -> Option<Date> {
		if self.type_code() != ValueKind::Date || !self.is_defined_at(index) {
			return None;
		}
		// SAFETY: the Date check above means the buffer is a marshalled &[Date]; `Date` is
		// `repr(transparent)` over `i32`, so it is aligned and every bit pattern is a valid value.
		unsafe { self.as_slice::<Date>()?.get(index).copied() }
	}

	#[inline]
	pub fn datetime_at(&self, index: usize) -> Option<DateTime> {
		if self.type_code() != ValueKind::DateTime || !self.is_defined_at(index) {
			return None;
		}
		// SAFETY: the DateTime check above means the buffer is a marshalled &[DateTime]; `DateTime` is
		// `repr(transparent)` over `u64`, so it is aligned and every bit pattern is a valid value.
		unsafe { self.as_slice::<DateTime>()?.get(index).copied() }
	}

	#[inline]
	pub fn time_at(&self, index: usize) -> Option<Time> {
		if self.type_code() != ValueKind::Time || !self.is_defined_at(index) {
			return None;
		}
		// SAFETY: the Time check above means the buffer is a marshalled &[Time]; `Time` is
		// `repr(transparent)` over `u64`, so it is aligned and every bit pattern is a valid value.
		unsafe { self.as_slice::<Time>()?.get(index).copied() }
	}

	#[inline]
	pub fn duration_at(&self, index: usize) -> Option<Duration> {
		if self.type_code() != ValueKind::Duration || !self.is_defined_at(index) {
			return None;
		}
		// SAFETY: the Duration check above means the buffer is a marshalled &[Duration]; `Duration` is
		// `repr(C)` over `i32`/`i32`/`i64`, so it is aligned and every bit pattern is a valid value.
		unsafe { self.as_slice::<Duration>()?.get(index).copied() }
	}
}

/// # Safety
///
/// `buf` must be a host-produced descriptor: either empty, or `buf.ptr` valid for `buf.len`
/// initialized bytes that stay live as long as `buf` itself is borrowed.
unsafe fn read_buffer(buf: &ExternCBuffer) -> &[u8] {
	if buf.ptr.is_null() || buf.len == 0 {
		&[]
	} else {
		// SAFETY: the branch above rules out a null pointer and a zero length; the caller's contract
		// makes the remaining `buf.len` bytes initialized and live for `'a`.
		unsafe { slice::from_raw_parts(buf.ptr, buf.len) }
	}
}

/// # Safety
///
/// Same contract as [`read_buffer`].
unsafe fn read_buffer_str(buf: &ExternCBuffer) -> &str {
	// SAFETY: forwarding this function's own contract, which is `read_buffer`'s.
	let bytes: &[u8] = unsafe { read_buffer(buf) };
	str::from_utf8(bytes).unwrap_or("")
}

pub type DiffKind = DiffType;