re_protos 0.35.0

use std::hash::Hasher as _;

use arrow::datatypes::Schema as ArrowSchema;
use arrow::error::ArrowError;
use itertools::Itertools as _;

use re_log_types::external::re_types_core::ComponentDescriptor;
use re_log_types::{RecordingId, TableId};

use crate::{TypeConversionError, invalid_field, missing_field};

/// Compression format used.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[repr(u8)]
pub enum Compression {
    Off = 0,

    /// Very fast compression and decompression, but not very good compression ratio.
    LZ4 = 1,
}

impl From<crate::common::v1alpha1::Compression> for Compression {
    fn from(value: crate::common::v1alpha1::Compression) -> Self {
        match value {
            crate::common::v1alpha1::Compression::Unspecified
            | crate::common::v1alpha1::Compression::None => Self::Off,
            crate::common::v1alpha1::Compression::Lz4 => Self::LZ4,
        }
    }
}

impl From<Compression> for crate::common::v1alpha1::Compression {
    fn from(value: Compression) -> Self {
        match value {
            Compression::Off => Self::None,
            Compression::LZ4 => Self::Lz4,
        }
    }
}

// --- Arrow ---

impl TryFrom<&crate::common::v1alpha1::Schema> for ArrowSchema {
    type Error = ArrowError;

    fn try_from(value: &crate::common::v1alpha1::Schema) -> Result<Self, Self::Error> {
        let schema_bytes = value
            .arrow_schema
            .as_ref()
            .ok_or(ArrowError::InvalidArgumentError(
                "missing schema bytes".to_owned(),
            ))?;
        Ok(Self::clone(
            re_sorbet::migrated_schema_from_ipc(schema_bytes)?.as_ref(),
        ))
    }
}

impl TryFrom<&ArrowSchema> for crate::common::v1alpha1::Schema {
    type Error = ArrowError;

    fn try_from(value: &ArrowSchema) -> Result<Self, Self::Error> {
        Ok(Self {
            arrow_schema: Some(re_sorbet::ipc_from_schema(value)?.into()),
        })
    }
}

impl TryFrom<crate::common::v1alpha1::Schema> for ArrowSchema {
    type Error = ArrowError;

    fn try_from(value: crate::common::v1alpha1::Schema) -> Result<Self, Self::Error> {
        (&value).try_into()
    }
}

// --- EntryId ---

impl From<re_log_types::EntryId> for crate::common::v1alpha1::EntryId {
    #[inline]
    fn from(value: re_log_types::EntryId) -> Self {
        Self {
            id: Some(value.id.into()),
        }
    }
}

impl TryFrom<crate::common::v1alpha1::EntryId> for re_log_types::EntryId {
    type Error = TypeConversionError;

    fn try_from(value: crate::common::v1alpha1::EntryId) -> Result<Self, Self::Error> {
        let id = value
            .id
            .ok_or(missing_field!(crate::common::v1alpha1::EntryId, "id"))?;
        Ok(Self { id: id.try_into()? })
    }
}

// shortcuts

impl From<re_tuid::Tuid> for crate::common::v1alpha1::EntryId {
    fn from(id: re_tuid::Tuid) -> Self {
        let id: re_log_types::EntryId = id.into();
        Self {
            id: Some(id.id.into()),
        }
    }
}

impl TryFrom<crate::common::v1alpha1::Tuid> for crate::common::v1alpha1::EntryId {
    type Error = TypeConversionError;

    fn try_from(id: crate::common::v1alpha1::Tuid) -> Result<Self, Self::Error> {
        let id: re_tuid::Tuid = id.try_into()?;
        Ok(Self {
            id: Some(id.into()),
        })
    }
}

// --- SegmentId ---

pub use re_types_core::SegmentId;

impl TryFrom<crate::common::v1alpha1::SegmentId> for SegmentId {
    type Error = TypeConversionError;

    fn try_from(value: crate::common::v1alpha1::SegmentId) -> Result<Self, Self::Error> {
        Ok(Self::from(value.id.ok_or(missing_field!(
            crate::common::v1alpha1::SegmentId,
            "id"
        ))?))
    }
}

impl From<SegmentId> for crate::common::v1alpha1::SegmentId {
    fn from(value: SegmentId) -> Self {
        Self {
            id: Some(value.into()),
        }
    }
}

// shortcuts

impl From<String> for crate::common::v1alpha1::SegmentId {
    fn from(id: String) -> Self {
        Self { id: Some(id) }
    }
}

impl From<&str> for crate::common::v1alpha1::SegmentId {
    fn from(id: &str) -> Self {
        Self {
            id: Some(id.to_owned()),
        }
    }
}

// --- DatasetKind ---

#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum DatasetKind {
    Recording,
    Blueprint,
    Asset,
}

impl DatasetKind {
    pub fn name(&self) -> &str {
        match self {
            DatasetKind::Recording => "dataset",
            DatasetKind::Blueprint => "blueprint dataset",
            DatasetKind::Asset => "asset dataset",
        }
    }

    /// Name of the thing this dataset contains.
    pub fn contained_name(&self) -> &str {
        match self {
            DatasetKind::Recording => "segment",
            DatasetKind::Blueprint => "blueprint",
            DatasetKind::Asset => "asset",
        }
    }

    pub fn store_kind(self) -> re_log_types::StoreKind {
        match self {
            Self::Recording | Self::Asset => re_log_types::StoreKind::Recording,
            Self::Blueprint => re_log_types::StoreKind::Blueprint,
        }
    }

    /// The limits enforced when registering segments into a dataset of this kind.
    pub fn limits(self) -> DatasetLimits {
        match self {
            Self::Recording => DatasetLimits::UNLIMITED,

            // Blueprint datasets aren't expected to have huge segments.
            //
            // The size limit of 25MB, comes from a blueprint with 100 empty
            // views being ~555 KiB. So some good head-room on top of that.
            Self::Blueprint => {
                DatasetLimits {
                    static_chunks_only: false,
                    max_segment_size_bytes: Some(25 * 1024 * 1024), // 25 MB
                    max_segment_count: None,
                }
            }

            // Asset datasets hold a small set of static blobs shared across a
            // dataset's segments, so we keep them deliberately small.
            //
            // The exact numbers here aren't important, and are there as a starting
            // point.
            Self::Asset => DatasetLimits {
                static_chunks_only: true,
                max_segment_size_bytes: Some(300 * 1024 * 1024), // 300 MiB
                max_segment_count: Some(12),
            },
        }
    }
}

/// Limits enforced when registering segments into a dataset.
///
/// Which limits apply depends on the dataset's [`DatasetKind`]; see [`DatasetKind::limits`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DatasetLimits {
    /// Reject any segment that contains temporal chunks.
    pub static_chunks_only: bool,

    /// Reject any segment whose total encoded size exceeds this many bytes.
    ///
    /// Measured as the sum of each chunk's `chunk_byte_len`, i.e. its IPC byte
    /// length as stored in the recording. This is the on-disk size, not the
    /// uncompressed in-memory size.
    pub max_segment_size_bytes: Option<u64>,

    /// Reject registration once the dataset already holds this many segments.
    pub max_segment_count: Option<u64>,
}

impl DatasetLimits {
    /// No limits: a dataset may hold any number of segments of any size and kind.
    pub const UNLIMITED: Self = Self {
        static_chunks_only: false,
        max_segment_size_bytes: None,
        max_segment_count: None,
    };
}

impl From<DatasetKind> for crate::common::v1alpha1::DatasetKind {
    fn from(value: DatasetKind) -> Self {
        match value {
            DatasetKind::Recording => Self::Recording,
            DatasetKind::Blueprint => Self::Blueprint,
            DatasetKind::Asset => Self::Asset,
        }
    }
}

impl From<crate::common::v1alpha1::DatasetKind> for DatasetKind {
    fn from(value: crate::common::v1alpha1::DatasetKind) -> Self {
        match value {
            crate::common::v1alpha1::DatasetKind::Unspecified
            | crate::common::v1alpha1::DatasetKind::Recording => Self::Recording,
            crate::common::v1alpha1::DatasetKind::Blueprint => Self::Blueprint,
            crate::common::v1alpha1::DatasetKind::Asset => Self::Asset,
        }
    }
}

impl From<super::rerun_cloud_v1alpha1::EntryKind> for DatasetKind {
    fn from(value: super::rerun_cloud_v1alpha1::EntryKind) -> Self {
        match value {
            super::rerun_cloud_v1alpha1::EntryKind::BlueprintDataset => DatasetKind::Blueprint,
            super::rerun_cloud_v1alpha1::EntryKind::AssetDataset => DatasetKind::Asset,
            super::rerun_cloud_v1alpha1::EntryKind::Unspecified
            | super::rerun_cloud_v1alpha1::EntryKind::Dataset
            | super::rerun_cloud_v1alpha1::EntryKind::DatasetView
            | super::rerun_cloud_v1alpha1::EntryKind::Table
            | super::rerun_cloud_v1alpha1::EntryKind::TableView => DatasetKind::Recording,
        }
    }
}

impl std::fmt::Display for DatasetKind {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Recording => f.write_str("Recording"),
            Self::Blueprint => f.write_str("Blueprint"),
            Self::Asset => f.write_str("Asset"),
        }
    }
}

impl std::str::FromStr for DatasetKind {
    type Err = String;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "Recording" => Ok(Self::Recording),
            "Blueprint" => Ok(Self::Blueprint),
            "Asset" => Ok(Self::Asset),
            other => Err(format!("unknown DatasetKind: {other:?}")),
        }
    }
}

// --- DatasetHandle ---

#[derive(Debug, Clone)]
pub struct DatasetHandle {
    pub id: Option<re_log_types::EntryId>,
    pub dataset_kind: DatasetKind,
    pub url: url::Url,
}

impl DatasetHandle {
    /// Create a new dataset handle
    pub fn new(url: url::Url, dataset_kind: DatasetKind) -> Self {
        Self {
            id: None,
            dataset_kind,
            url,
        }
    }

    pub fn with_id(mut self, id: re_log_types::EntryId) -> Self {
        self.id = Some(id);
        self
    }
}

impl TryFrom<crate::common::v1alpha1::DatasetHandle> for DatasetHandle {
    type Error = TypeConversionError;

    fn try_from(value: crate::common::v1alpha1::DatasetHandle) -> Result<Self, Self::Error> {
        let dataset_kind = crate::common::v1alpha1::DatasetKind::try_from(value.dataset_kind);
        // TODO(RR-5123): Remove once new client -> old server isn't needed anymore.
        #[expect(deprecated)]
        let store_kind = crate::common::v1alpha1::StoreKind::try_from(value.store_kind);

        let dataset_kind = if let Ok(store_kind) = store_kind
            && !matches!(store_kind, crate::common::v1alpha1::StoreKind::Unspecified)
            && matches!(
                dataset_kind,
                Ok(crate::common::v1alpha1::DatasetKind::Unspecified) | Err(_)
            ) {
            Ok(match store_kind {
                crate::common::v1alpha1::StoreKind::Unspecified => {
                    crate::common::v1alpha1::DatasetKind::Unspecified
                }
                crate::common::v1alpha1::StoreKind::Recording => {
                    crate::common::v1alpha1::DatasetKind::Recording
                }
                crate::common::v1alpha1::StoreKind::Blueprint => {
                    crate::common::v1alpha1::DatasetKind::Blueprint
                }
            })
        } else {
            dataset_kind
        };

        Ok(Self {
            id: value.entry_id.map(|id| id.try_into()).transpose()?,
            dataset_kind: dataset_kind?.into(),
            url: value
                .dataset_url
                .ok_or(missing_field!(
                    crate::common::v1alpha1::DatasetHandle,
                    "dataset_url"
                ))?
                .parse()
                .map_err(|err| {
                    invalid_field!(crate::common::v1alpha1::DatasetHandle, "dataset_url", err)
                })?,
        })
    }
}

impl From<DatasetHandle> for crate::common::v1alpha1::DatasetHandle {
    // The deprecated `store_kind` field is still populated for old servers,
    // see the TODO below.
    #[allow(deprecated)]
    fn from(value: DatasetHandle) -> Self {
        Self {
            entry_id: value.id.map(Into::into),
            // TODO(RR-5123): Remove once new client -> old server isn't needed anymore.
            store_kind: crate::common::v1alpha1::StoreKind::from(value.dataset_kind.store_kind())
                as i32,
            dataset_kind: crate::common::v1alpha1::DatasetKind::from(value.dataset_kind) as i32,
            dataset_url: Some(value.url.to_string()),
        }
    }
}

// --- TaskId ---

//TODO(ab): we should migrate the full `TaskId` wrapper from `redap_tasks`
impl crate::common::v1alpha1::TaskId {
    pub fn new() -> Self {
        Self::from_hashable(re_tuid::Tuid::new())
    }

    pub fn into_string(self) -> String {
        self.id
    }

    pub fn from_hashable<H: std::hash::Hash>(hashable: H) -> Self {
        let mut hasher = std::hash::DefaultHasher::new();
        hashable.hash(&mut hasher);
        let id = hasher.finish();

        Self {
            id: format!("task_{id:016x}"),
        }
    }
}

impl From<String> for crate::common::v1alpha1::TaskId {
    fn from(id: String) -> Self {
        Self { id }
    }
}

impl From<crate::common::v1alpha1::TaskId> for String {
    fn from(task_id: crate::common::v1alpha1::TaskId) -> Self {
        task_id.id
    }
}

// Make `quiver::Column<TaskId>` work (backed by a `Utf8` column):
quiver::newtype_datatype!(crate::common::v1alpha1::TaskId, quiver::Utf8);

// ---

impl TryFrom<crate::common::v1alpha1::Tuid> for re_tuid::Tuid {
    type Error = TypeConversionError;

    fn try_from(value: crate::common::v1alpha1::Tuid) -> Result<Self, Self::Error> {
        let time_ns = value
            .time_ns
            .ok_or(missing_field!(crate::common::v1alpha1::Tuid, "time_ns"))?;
        let inc = value
            .inc
            .ok_or(missing_field!(crate::common::v1alpha1::Tuid, "inc"))?;

        Ok(Self::from_nanos_and_inc(time_ns, inc))
    }
}

impl From<re_tuid::Tuid> for crate::common::v1alpha1::Tuid {
    fn from(value: re_tuid::Tuid) -> Self {
        Self {
            time_ns: Some(value.nanos_since_epoch()),
            inc: Some(value.inc()),
        }
    }
}

impl From<re_log_types::EntityPath> for crate::common::v1alpha1::EntityPath {
    fn from(value: re_log_types::EntityPath) -> Self {
        Self {
            path: value.to_string(),
        }
    }
}

impl TryFrom<crate::common::v1alpha1::EntityPath> for re_log_types::EntityPath {
    type Error = TypeConversionError;

    fn try_from(value: crate::common::v1alpha1::EntityPath) -> Result<Self, Self::Error> {
        Self::parse_strict(&value.path)
            .map_err(|err| invalid_field!(crate::common::v1alpha1::EntityPath, "path", err))
    }
}

impl From<re_log_types::AbsoluteTimeRange> for crate::common::v1alpha1::TimeRange {
    fn from(value: re_log_types::AbsoluteTimeRange) -> Self {
        Self {
            start: value.min().as_i64(),
            end: value.max().as_i64(),
        }
    }
}

impl From<crate::common::v1alpha1::TimeRange> for re_log_types::AbsoluteTimeRange {
    fn from(value: crate::common::v1alpha1::TimeRange) -> Self {
        Self::new(
            re_log_types::TimeInt::new_temporal(value.start),
            re_log_types::TimeInt::new_temporal(value.end),
        )
    }
}

impl From<re_log_types::AbsoluteTimeRange> for crate::common::v1alpha1::IndexRange {
    fn from(value: re_log_types::AbsoluteTimeRange) -> Self {
        Self {
            time_range: Some(value.into()),
        }
    }
}

impl TryFrom<crate::common::v1alpha1::IndexRange> for re_log_types::AbsoluteTimeRange {
    type Error = TypeConversionError;

    fn try_from(value: crate::common::v1alpha1::IndexRange) -> Result<Self, Self::Error> {
        value
            .time_range
            .ok_or(missing_field!(
                crate::common::v1alpha1::IndexRange,
                "time_range"
            ))
            .map(|time_range| Self::new(time_range.start, time_range.end))
    }
}

impl TryFrom<crate::common::v1alpha1::Timeline> for re_log_types::TimelineName {
    type Error = TypeConversionError;

    fn try_from(value: crate::common::v1alpha1::Timeline) -> Result<Self, Self::Error> {
        Ok(Self::try_new(&value.name)?)
    }
}

impl From<re_log_types::TimelineName> for crate::common::v1alpha1::Timeline {
    fn from(value: re_log_types::TimelineName) -> Self {
        Self {
            name: value.to_string(),
        }
    }
}

impl From<re_log_types::Timeline> for crate::common::v1alpha1::Timeline {
    fn from(value: re_log_types::Timeline) -> Self {
        Self {
            name: value.name().to_string(),
        }
    }
}

impl From<re_log_types::TimeType> for crate::common::v1alpha1::TimeType {
    fn from(value: re_log_types::TimeType) -> Self {
        match value {
            re_log_types::TimeType::Sequence => Self::Sequence,
            re_log_types::TimeType::DurationNs => Self::DurationNs,
            re_log_types::TimeType::TimestampNs => Self::TimestampNs,
        }
    }
}

impl std::fmt::Display for crate::common::v1alpha1::TimeType {
    /// Short, human-readable name, matching `re_log_types::TimeType`'s `Display`
    /// (`"sequence"`, `"duration"`, `"timestamp"`, or `"unknown"`).
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(match self {
            Self::Unspecified => "unknown",
            Self::Sequence => "sequence",
            Self::DurationNs => "duration",
            Self::TimestampNs => "timestamp",
        })
    }
}

impl From<i64> for crate::common::v1alpha1::TimelineTime {
    fn from(time: i64) -> Self {
        Self { time }
    }
}

impl TryFrom<crate::common::v1alpha1::IndexColumnSelector> for re_log_types::TimelineName {
    type Error = TypeConversionError;

    fn try_from(value: crate::common::v1alpha1::IndexColumnSelector) -> Result<Self, Self::Error> {
        let timeline = value.timeline.ok_or(missing_field!(
            crate::common::v1alpha1::IndexColumnSelector,
            "timeline"
        ))?;
        timeline.try_into()
    }
}

impl From<re_log_types::TimelineName> for crate::common::v1alpha1::IndexColumnSelector {
    fn from(value: re_log_types::TimelineName) -> Self {
        Self {
            timeline: Some(value.into()),
        }
    }
}

impl From<crate::common::v1alpha1::ApplicationId> for re_log_types::ApplicationId {
    #[inline]
    fn from(value: crate::common::v1alpha1::ApplicationId) -> Self {
        Self::from(value.id)
    }
}

impl From<re_log_types::ApplicationId> for crate::common::v1alpha1::ApplicationId {
    #[inline]
    fn from(value: re_log_types::ApplicationId) -> Self {
        Self {
            id: value.to_string(),
        }
    }
}

impl From<crate::common::v1alpha1::StoreKind> for re_log_types::StoreKind {
    #[inline]
    fn from(value: crate::common::v1alpha1::StoreKind) -> Self {
        match value {
            crate::common::v1alpha1::StoreKind::Unspecified
            | crate::common::v1alpha1::StoreKind::Recording => Self::Recording,
            crate::common::v1alpha1::StoreKind::Blueprint => Self::Blueprint,
        }
    }
}

impl From<re_log_types::StoreKind> for crate::common::v1alpha1::StoreKind {
    #[inline]
    fn from(value: re_log_types::StoreKind) -> Self {
        match value {
            re_log_types::StoreKind::Recording => Self::Recording,
            re_log_types::StoreKind::Blueprint => Self::Blueprint,
        }
    }
}

/// The store id failed to deserialize due to missing application id.
///
/// This may happen when migrating older RRD (before application ID was moved to `StoreId`). This
/// error can be recovered from if the application ID is known.
//TODO(#10730): this is specifically for 0.24 back compat. Switch to `TypeConversionError` when cleaning up.
#[derive(Debug, Clone)]
pub struct StoreIdMissingApplicationIdError {
    pub store_kind: re_log_types::StoreKind,
    pub recording_id: RecordingId,
}

impl StoreIdMissingApplicationIdError {
    /// Recover the store ID by providing the application ID.
    pub fn recover(self, application_id: re_log_types::ApplicationId) -> re_log_types::StoreId {
        re_log_types::StoreId::new(self.store_kind, application_id, self.recording_id)
    }

    #[inline]
    pub fn into_type_conversion_error(self, msg: impl Into<String>) -> TypeConversionError {
        TypeConversionError::LegacyStoreIdError(format!(
            "{} (kind: {}, recording_id: {})",
            msg.into(),
            self.store_kind,
            self.recording_id
        ))
    }
}

/// Convert a store id
impl TryFrom<crate::common::v1alpha1::StoreId> for re_log_types::StoreId {
    type Error = StoreIdMissingApplicationIdError;

    #[inline]
    fn try_from(value: crate::common::v1alpha1::StoreId) -> Result<Self, Self::Error> {
        let store_kind = value.kind().into();
        let recording_id = RecordingId::from(value.recording_id);

        //TODO(#10730): switch to `TypeConversionError` when cleaning up 0.24 back compat
        match value.application_id {
            None => Err(StoreIdMissingApplicationIdError {
                store_kind,
                recording_id,
            }),
            Some(application_id) => Ok(re_log_types::StoreId::new(
                store_kind,
                application_id,
                recording_id,
            )),
        }
    }
}

impl From<re_log_types::StoreId> for crate::common::v1alpha1::StoreId {
    #[inline]
    fn from(value: re_log_types::StoreId) -> Self {
        let kind: crate::common::v1alpha1::StoreKind = value.kind().into();
        Self {
            kind: kind as i32,
            recording_id: value.recording_id().as_str().to_owned(),
            application_id: Some(value.application_id().clone().into()),
        }
    }
}

impl From<re_log_types::TableId> for crate::common::v1alpha1::TableId {
    #[inline]
    fn from(value: re_log_types::TableId) -> Self {
        Self {
            id: value.as_str().to_owned(),
        }
    }
}

impl From<crate::common::v1alpha1::TableId> for re_log_types::TableId {
    #[inline]
    fn from(value: crate::common::v1alpha1::TableId) -> Self {
        TableId::from(value.id)
    }
}

// --- Scanning & Querying ---

#[derive(Debug, Default, Clone)]
pub struct ScanParameters {
    pub columns: Vec<String>,
    pub on_missing_columns: IfMissingBehavior,
    pub filter: Option<String>,
    pub limit_offset: Option<i64>,
    pub limit_len: Option<i64>,
    pub order_by: Vec<ScanParametersOrderClause>,
    pub explain_plan: bool,
    pub explain_filter: bool,
}

impl TryFrom<crate::common::v1alpha1::ScanParameters> for ScanParameters {
    type Error = TypeConversionError;

    fn try_from(value: crate::common::v1alpha1::ScanParameters) -> Result<Self, Self::Error> {
        Ok(Self {
            columns: value.columns,
            on_missing_columns: crate::common::v1alpha1::IfMissingBehavior::try_from(
                value.on_missing_columns,
            )?
            .into(),
            filter: value.filter,
            limit_offset: value.limit_offset,
            limit_len: value.limit_len,
            order_by: value
                .order_by
                .into_iter()
                .map(|ob| ob.try_into())
                .try_collect()?,
            explain_plan: value.explain_plan,
            explain_filter: value.explain_filter,
        })
    }
}

impl From<ScanParameters> for crate::common::v1alpha1::ScanParameters {
    fn from(value: ScanParameters) -> Self {
        Self {
            columns: value.columns,
            on_missing_columns: crate::common::v1alpha1::IfMissingBehavior::from(
                value.on_missing_columns,
            ) as _,
            filter: value.filter,
            limit_offset: value.limit_offset,
            limit_len: value.limit_len,
            order_by: value.order_by.into_iter().map(|ob| ob.into()).collect(),
            explain_plan: value.explain_plan,
            explain_filter: value.explain_filter,
        }
    }
}

#[derive(Debug, Default, Clone)]
pub struct ScanParametersOrderClause {
    pub descending: bool,
    pub nulls_last: bool,
    pub column_name: String,
}

impl TryFrom<crate::common::v1alpha1::ScanParametersOrderClause> for ScanParametersOrderClause {
    type Error = TypeConversionError;

    fn try_from(
        value: crate::common::v1alpha1::ScanParametersOrderClause,
    ) -> Result<Self, Self::Error> {
        Ok(Self {
            descending: value.descending,
            nulls_last: value.nulls_last,
            column_name: value.column_name.ok_or(missing_field!(
                crate::common::v1alpha1::ScanParametersOrderClause,
                "column_name"
            ))?,
        })
    }
}

impl From<ScanParametersOrderClause> for crate::common::v1alpha1::ScanParametersOrderClause {
    fn from(value: ScanParametersOrderClause) -> Self {
        Self {
            descending: value.descending,
            nulls_last: value.nulls_last,
            column_name: Some(value.column_name),
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum IfMissingBehavior {
    Skip,
    Error,
}

impl Default for IfMissingBehavior {
    fn default() -> Self {
        Self::Skip
    }
}

impl From<crate::common::v1alpha1::IfMissingBehavior> for IfMissingBehavior {
    fn from(value: crate::common::v1alpha1::IfMissingBehavior) -> Self {
        use crate::common::v1alpha1 as common;
        match value {
            common::IfMissingBehavior::Unspecified | common::IfMissingBehavior::Skip => Self::Skip,
            common::IfMissingBehavior::Error => Self::Error,
        }
    }
}

impl From<IfMissingBehavior> for crate::common::v1alpha1::IfMissingBehavior {
    fn from(value: IfMissingBehavior) -> Self {
        match value {
            IfMissingBehavior::Skip => Self::Skip,
            IfMissingBehavior::Error => Self::Error,
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum IfDuplicateBehavior {
    Overwrite,
    Skip,
    Error,
}

impl Default for IfDuplicateBehavior {
    fn default() -> Self {
        Self::Skip
    }
}

impl TryFrom<i32> for IfDuplicateBehavior {
    type Error = TypeConversionError;

    fn try_from(value: i32) -> Result<Self, TypeConversionError> {
        let proto_value = crate::common::v1alpha1::IfDuplicateBehavior::try_from(value)?;
        Ok(Self::from(proto_value))
    }
}

impl From<crate::common::v1alpha1::IfDuplicateBehavior> for IfDuplicateBehavior {
    fn from(value: crate::common::v1alpha1::IfDuplicateBehavior) -> Self {
        use crate::common::v1alpha1 as common;
        match value {
            common::IfDuplicateBehavior::Unspecified | common::IfDuplicateBehavior::Skip => {
                Self::Skip
            }
            common::IfDuplicateBehavior::Overwrite => Self::Overwrite,
            common::IfDuplicateBehavior::Error => Self::Error,
        }
    }
}

impl From<IfDuplicateBehavior> for crate::common::v1alpha1::IfDuplicateBehavior {
    fn from(value: IfDuplicateBehavior) -> Self {
        match value {
            IfDuplicateBehavior::Overwrite => Self::Overwrite,
            IfDuplicateBehavior::Skip => Self::Skip,
            IfDuplicateBehavior::Error => Self::Error,
        }
    }
}

// ---

impl From<ComponentDescriptor> for crate::common::v1alpha1::ComponentDescriptor {
    fn from(value: ComponentDescriptor) -> Self {
        Self {
            archetype: value.archetype.map(|n| n.full_name().to_owned()),
            component: Some(value.component.to_string()),
            component_type: value.component_type.map(|c| c.full_name().to_owned()),
        }
    }
}

impl TryFrom<crate::common::v1alpha1::ComponentDescriptor> for ComponentDescriptor {
    type Error = TypeConversionError;

    fn try_from(value: crate::common::v1alpha1::ComponentDescriptor) -> Result<Self, Self::Error> {
        let crate::common::v1alpha1::ComponentDescriptor {
            archetype,
            component,
            component_type,
        } = value;

        let component = component.ok_or(missing_field!(
            crate::common::v1alpha1::ComponentDescriptor,
            "component"
        ))?;

        Ok(ComponentDescriptor {
            archetype: archetype.and_then(|s| re_types_core::ArchetypeName::try_new(s).ok()),
            component: re_types_core::ComponentIdentifier::try_new(component).map_err(|err| {
                invalid_field!(
                    crate::common::v1alpha1::ComponentDescriptor,
                    "component",
                    err
                )
            })?,
            component_type: component_type
                .and_then(|s| re_types_core::ComponentType::try_new(s).ok()),
        })
    }
}

// ---

impl From<re_build_info::BuildInfo> for crate::common::v1alpha1::BuildInfo {
    fn from(build_info: re_build_info::BuildInfo) -> Self {
        Self {
            crate_name: Some(build_info.crate_name.to_string()),
            features: Some(build_info.features.to_string()),
            version: Some(build_info.version.into()),
            rustc_version: Some(build_info.rustc_version.to_string()),
            llvm_version: Some(build_info.llvm_version.to_string()),
            git_hash: Some(build_info.git_hash.to_string()),
            git_branch: Some(build_info.git_branch.to_string()),
            target_triple: Some(build_info.target_triple.to_string()),
            build_time: Some(build_info.datetime.to_string()),
            is_debug_build: Some(build_info.is_debug_build),
        }
    }
}

impl From<crate::common::v1alpha1::BuildInfo> for re_build_info::BuildInfo {
    fn from(build_info: crate::common::v1alpha1::BuildInfo) -> Self {
        Self {
            crate_name: build_info.crate_name().to_owned().into(),
            features: build_info.features().to_owned().into(),
            version: build_info.version.clone().unwrap_or_default().into(),
            rustc_version: build_info.rustc_version().to_owned().into(),
            llvm_version: build_info.llvm_version().to_owned().into(),
            git_hash: build_info.git_hash().to_owned().into(),
            git_branch: build_info.git_branch().to_owned().into(),
            is_in_rerun_workspace: false,
            target_triple: build_info.target_triple().to_owned().into(),
            datetime: build_info.build_time().to_owned().into(),
            is_debug_build: build_info.is_debug_build(),
        }
    }
}

impl From<re_build_info::CrateVersion> for crate::common::v1alpha1::SemanticVersion {
    fn from(version: re_build_info::CrateVersion) -> Self {
        crate::common::v1alpha1::SemanticVersion {
            major: Some(version.major.into()),
            minor: Some(version.minor.into()),
            patch: Some(version.patch.into()),
            meta: version.meta.map(Into::into),
        }
    }
}

impl From<crate::common::v1alpha1::SemanticVersion> for re_build_info::CrateVersion {
    fn from(version: crate::common::v1alpha1::SemanticVersion) -> Self {
        Self {
            major: version.major() as u8,
            minor: version.minor() as u8,
            patch: version.patch() as u8,
            meta: version.meta.map(Into::into),
        }
    }
}

impl From<re_build_info::Meta> for crate::common::v1alpha1::semantic_version::Meta {
    fn from(version_meta: re_build_info::Meta) -> Self {
        match version_meta {
            re_build_info::Meta::Rc(v) => Self::Rc(v.into()),

            re_build_info::Meta::Alpha(v) => Self::Alpha(v.into()),

            re_build_info::Meta::DevAlpha { alpha, commit } => {
                Self::DevAlpha(crate::common::v1alpha1::DevAlpha {
                    alpha: Some(alpha.into()),
                    commit: commit.map(|s| String::from_utf8_lossy(s).to_string()),
                })
            }
        }
    }
}

impl From<crate::common::v1alpha1::semantic_version::Meta> for re_build_info::Meta {
    fn from(version_meta: crate::common::v1alpha1::semantic_version::Meta) -> Self {
        match version_meta {
            crate::common::v1alpha1::semantic_version::Meta::Rc(v) => Self::Rc(v as _),

            crate::common::v1alpha1::semantic_version::Meta::Alpha(v) => Self::Alpha(v as _),

            crate::common::v1alpha1::semantic_version::Meta::DevAlpha(dev_alpha) => {
                Self::DevAlpha {
                    alpha: dev_alpha.alpha() as u8,
                    // TODO(cmc): support this, but that means DevAlpha is not-const
                    // anymore, which trigger a chain reaction of changes that I really
                    // don't want to get in right now.
                    commit: None,
                }
            }
        }
    }
}

// --- DataframePart / RerunChunk <-> RecordBatch ---

impl TryFrom<crate::common::v1alpha1::RerunChunk> for arrow::array::RecordBatch {
    type Error = TypeConversionError;

    fn try_from(value: crate::common::v1alpha1::RerunChunk) -> Result<Self, Self::Error> {
        Self::try_from(&value)
    }
}

impl TryFrom<&crate::common::v1alpha1::RerunChunk> for arrow::array::RecordBatch {
    type Error = TypeConversionError;

    fn try_from(value: &crate::common::v1alpha1::RerunChunk) -> Result<Self, Self::Error> {
        match value.encoder_version() {
            crate::common::v1alpha1::EncoderVersion::Unspecified => {
                return Err(missing_field!(
                    crate::common::v1alpha1::RerunChunk,
                    "encoder_version"
                ));
            }

            crate::common::v1alpha1::EncoderVersion::V0 => {
                let Some(bytes) = value.payload.as_ref() else {
                    return Err(missing_field!(
                        crate::common::v1alpha1::DataframePart,
                        "payload"
                    ));
                };
                let Some(batch) =
                    record_batch_from_ipc_bytes(bytes, Compression::Off, bytes.len() as u64)?
                else {
                    return Err(invalid_field!(
                        crate::common::v1alpha1::RerunChunk,
                        "payload",
                        "empty"
                    ));
                };
                Ok(batch)
            }
        }
    }
}

impl From<arrow::array::RecordBatch> for crate::common::v1alpha1::RerunChunk {
    fn from(value: arrow::array::RecordBatch) -> Self {
        Self::from(&value)
    }
}

impl From<&arrow::array::RecordBatch> for crate::common::v1alpha1::RerunChunk {
    fn from(value: &arrow::array::RecordBatch) -> Self {
        let version = crate::common::v1alpha1::EncoderVersion::V0;
        Self {
            encoder_version: version as i32,
            payload: Some(record_batch_to_ipc_bytes(value, Compression::Off).0.into()),
        }
    }
}

impl TryFrom<crate::common::v1alpha1::DataframePart> for arrow::array::RecordBatch {
    type Error = TypeConversionError;

    fn try_from(value: crate::common::v1alpha1::DataframePart) -> Result<Self, Self::Error> {
        Self::try_from(&value)
    }
}

impl TryFrom<&crate::common::v1alpha1::DataframePart> for arrow::array::RecordBatch {
    type Error = TypeConversionError;

    fn try_from(value: &crate::common::v1alpha1::DataframePart) -> Result<Self, Self::Error> {
        match value.encoder_version() {
            crate::common::v1alpha1::EncoderVersion::Unspecified => {
                return Err(missing_field!(
                    crate::common::v1alpha1::RerunChunk,
                    "encoder_version"
                ));
            }

            crate::common::v1alpha1::EncoderVersion::V0 => {
                let Some(bytes) = value.payload.as_ref() else {
                    return Err(missing_field!(
                        crate::common::v1alpha1::DataframePart,
                        "payload"
                    ));
                };

                let compression =
                    crate::common::v1alpha1::Compression::try_from(value.compression)?;
                let compression = Compression::from(compression);

                let Some(batch) =
                    record_batch_from_ipc_bytes(bytes, compression, value.uncompressed_size)?
                else {
                    return Err(invalid_field!(
                        crate::common::v1alpha1::RerunChunk,
                        "payload",
                        "empty"
                    ));
                };
                Ok(batch)
            }
        }
    }
}

impl From<arrow::array::RecordBatch> for crate::common::v1alpha1::DataframePart {
    fn from(value: arrow::array::RecordBatch) -> Self {
        Self::from(&value)
    }
}

impl From<&arrow::array::RecordBatch> for crate::common::v1alpha1::DataframePart {
    fn from(value: &arrow::array::RecordBatch) -> Self {
        let version = crate::common::v1alpha1::EncoderVersion::V0;
        let compression = crate::common::v1alpha1::Compression::Lz4;
        let (payload, uncompressed_size) = record_batch_to_ipc_bytes(value, compression.into());
        Self {
            encoder_version: version as i32, // make sure `version` is using the transport type!!
            payload: Some(payload.into()),
            compression: compression as i32, // make sure `compression` is using the transport type!!
            uncompressed_size,
        }
    }
}

/// `RecordBatch` to IPC bytes. No I/O, no failures.
#[tracing::instrument(level = "debug", skip_all)]
fn record_batch_to_ipc_bytes(
    batch: &arrow::array::RecordBatch,
    compression: Compression,
) -> (Vec<u8>, u64) {
    let schema = batch.schema_ref().as_ref();

    let mut uncompressed = Vec::new();

    {
        let mut sw = {
            let _span = tracing::trace_span!("schema").entered();
            arrow::ipc::writer::StreamWriter::try_new(&mut uncompressed, schema)
                .expect("encoding the schema of a valid RecordBatch as IPC bytes into a growable in-memory buffer cannot possibly fail")
        };

        {
            let _span = tracing::trace_span!("data").entered();
            sw.write(batch)
                .expect("encoding the data of a valid RecordBatch as IPC bytes into a growable in-memory buffer cannot possibly fail");
        }

        sw.finish()
            .expect("encoding a valid RecordBatch as IPC bytes into a growable in-memory buffer cannot possibly fail");
    }

    let uncompressed_size = uncompressed.len() as u64;

    let data = match compression {
        Compression::Off => uncompressed,
        Compression::LZ4 => {
            re_tracing::profile_scope!("lz4::compress");
            let _span = tracing::trace_span!("lz4::compress").entered();
            lz4_flex::block::compress(&uncompressed)
        }
    };

    (data, uncompressed_size)
}

/// IPC bytes to `RecordBatch`. `Ok(None)` if there's no data.
#[tracing::instrument(level = "debug", skip_all)]
fn record_batch_from_ipc_bytes(
    bytes: &[u8],
    compression: Compression,
    uncompressed_size: u64,
) -> Result<Option<arrow::array::RecordBatch>, ArrowError> {
    let mut uncompressed = Vec::new();
    let bytes = match compression {
        Compression::Off => bytes,
        Compression::LZ4 => {
            re_tracing::profile_scope!("LZ4-decompress");
            let _span = tracing::trace_span!("lz4::decompress").entered();
            uncompressed.resize(uncompressed_size as usize, 0);
            lz4_flex::block::decompress_into(bytes, &mut uncompressed).map_err(|err| {
                ArrowError::ParseError(format!("LZ4 decompression failure: {err:#}"))
            })?;
            uncompressed.as_slice()
        }
    };

    let mut stream = {
        let _span = tracing::trace_span!("schema").entered();
        arrow::ipc::reader::StreamReader::try_new(bytes, None)?
    };

    let _span = tracing::trace_span!("data").entered();
    stream.next().transpose()
}

// ---

#[cfg(test)]
mod tests {

    #[test]
    fn entity_path_conversion() {
        let entity_path = re_log_types::EntityPath::parse_strict("a/b/c").unwrap();
        let proto_entity_path: crate::common::v1alpha1::EntityPath = entity_path.clone().into();
        let entity_path2: re_log_types::EntityPath = proto_entity_path.try_into().unwrap();
        assert_eq!(entity_path, entity_path2);
    }

    #[test]
    fn time_range_conversion() {
        let time_range = re_log_types::AbsoluteTimeRange::new(
            re_log_types::TimeInt::new_temporal(123456789),
            re_log_types::TimeInt::new_temporal(987654321),
        );
        let proto_time_range: crate::common::v1alpha1::TimeRange = time_range.into();
        let time_range2: re_log_types::AbsoluteTimeRange = proto_time_range.into();
        assert_eq!(time_range, time_range2);
    }

    #[test]
    fn index_range_conversion() {
        let time_range = re_log_types::AbsoluteTimeRange::new(
            re_log_types::TimeInt::new_temporal(123456789),
            re_log_types::TimeInt::new_temporal(987654321),
        );
        let proto_index_range: crate::common::v1alpha1::IndexRange = time_range.into();
        let time_range2: re_log_types::AbsoluteTimeRange = proto_index_range.try_into().unwrap();
        assert_eq!(time_range, time_range2);
    }

    #[test]
    fn index_column_selector_conversion() {
        let timeline = re_log_types::TimelineName::log_time();
        let proto_index_column_selector: crate::common::v1alpha1::IndexColumnSelector =
            crate::common::v1alpha1::IndexColumnSelector {
                timeline: Some(timeline.into()),
            };
        let timeline2: re_log_types::TimelineName = proto_index_column_selector.try_into().unwrap();
        assert_eq!(timeline, timeline2);
    }

    #[test]
    fn application_id_conversion() {
        let application_id = re_log_types::ApplicationId::from("test");
        let proto_application_id: crate::common::v1alpha1::ApplicationId =
            application_id.clone().into();
        let application_id2: re_log_types::ApplicationId = proto_application_id.into();
        assert_eq!(application_id, application_id2);
    }

    #[test]
    fn store_kind_conversion() {
        let store_kind = re_log_types::StoreKind::Recording;
        let proto_store_kind: crate::common::v1alpha1::StoreKind = store_kind.into();
        let store_kind2: re_log_types::StoreKind = proto_store_kind.into();
        assert_eq!(store_kind, store_kind2);
    }

    #[test]
    fn store_id_conversion() {
        let store_id = re_log_types::StoreId::new(
            re_log_types::StoreKind::Recording,
            "test_app_id",
            "test_recording",
        );
        let proto_store_id: crate::common::v1alpha1::StoreId = store_id.clone().into();
        let store_id2: re_log_types::StoreId = proto_store_id.try_into().unwrap();
        assert_eq!(store_id, store_id2);
    }

    #[test]
    fn test_tuid_conversion() {
        let tuid = re_tuid::Tuid::new();
        let proto_tuid: crate::common::v1alpha1::Tuid = tuid.into();
        let tuid2: re_tuid::Tuid = proto_tuid.try_into().unwrap();
        assert_eq!(tuid, tuid2);
    }
}