pipecircuit 0.0.2

use crate::storage::warp_table::inconsistency::invariant_violation;
use crate::storage::warp_table::secondary_index::{
    RangeIndexMembership, SlotSet, StableSlotResolveOutcome, StableSlotResolver,
};
use crate::storage::warp_table::{RangeIndexBounds, WarpTable, WarpTableRow};

#[doc(hidden)]
pub struct DetachedSelection {
    slots: Option<SlotSet>,
}

#[doc(hidden)]
#[must_use]
pub enum DetachedSecondaryIndexLookupOutcome<SecondaryKey> {
    Present {
        secondary_key: SecondaryKey,
        selection: DetachedSelection,
    },
    Absent {
        secondary_key: SecondaryKey,
        selection: DetachedSelection,
    },
}

#[doc(hidden)]
#[must_use]
pub enum DetachedRangeIndexLookupOutcome<Key> {
    Present {
        bounds: RangeIndexBounds<Key>,
        selection: DetachedSelection,
    },
    Absent {
        bounds: RangeIndexBounds<Key>,
        selection: DetachedSelection,
    },
}

macro_rules! impl_detached_lookup {
    ($outcome:ident) => {
        impl<Key> $outcome<Key> {
            fn into_selection(self) -> DetachedSelection {
                match self {
                    Self::Present { selection, .. } | Self::Absent { selection, .. } => selection,
                }
            }
        }

        impl<Key> From<$outcome<Key>> for DetachedSelection {
            fn from(outcome: $outcome<Key>) -> Self {
                outcome.into_selection()
            }
        }
    };
}

impl_detached_lookup!(DetachedSecondaryIndexLookupOutcome);
impl_detached_lookup!(DetachedRangeIndexLookupOutcome);

#[must_use]
pub struct MutableSelection<'table, Row: WarpTableRow, Brand = Row> {
    table: &'table mut WarpTable<Row, Brand>,
    slots: Option<SlotSet>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[must_use]
pub enum MutableSelectionForEachOutcome {
    Updated { rows: u64 },
}

macro_rules! mutable_algebra_methods {
    ($receiver:ident, $selection:expr) => {
        pub fn and<Operand>(
            $receiver,
            find: impl for<'indexes> FnOnce(Row::DetachedQuery<'indexes>) -> Operand,
        ) -> Operand::Output
        where
            Operand: MutableSelectionOperand<'table, Row, Brand>,
        {
            let selection = $selection;
            find(Row::detached_query(selection.table.__indexes())).and_with(selection)
        }

        pub fn or<Operand>(
            $receiver,
            find: impl for<'indexes> FnOnce(Row::DetachedQuery<'indexes>) -> Operand,
        ) -> Operand::Output
        where
            Operand: MutableSelectionOperand<'table, Row, Brand>,
        {
            let selection = $selection;
            find(Row::detached_query(selection.table.__indexes())).or_with(selection)
        }

        pub fn and_not<Operand>(
            $receiver,
            find: impl for<'indexes> FnOnce(Row::DetachedQuery<'indexes>) -> Operand,
        ) -> Operand::Output
        where
            Operand: MutableSelectionOperand<'table, Row, Brand>,
        {
            let selection = $selection;
            find(Row::detached_query(selection.table.__indexes())).and_not_with(selection)
        }

        pub fn xor<Operand>(
            $receiver,
            find: impl for<'indexes> FnOnce(Row::DetachedQuery<'indexes>) -> Operand,
        ) -> Operand::Output
        where
            Operand: MutableSelectionOperand<'table, Row, Brand>,
        {
            let selection = $selection;
            find(Row::detached_query(selection.table.__indexes())).xor_with(selection)
        }
    };
}

impl<'table, Row: WarpTableRow, Brand> MutableSelection<'table, Row, Brand> {
    fn new(table: &'table mut WarpTable<Row, Brand>, slots: Option<SlotSet>) -> Self {
        Self { table, slots }
    }

    pub fn len(&self) -> u64 {
        self.slots.as_ref().map_or(0, SlotSet::len)
    }

    pub fn is_empty(&self) -> bool {
        self.slots.as_ref().is_none_or(SlotSet::is_empty)
    }

    mutable_algebra_methods!(self, self);

    pub fn for_each(self, mut update: impl FnMut(Row::Mut<'_>)) -> MutableSelectionForEachOutcome
    where
        Row::Indexes: super::rows::StableSlotResolver,
    {
        let Some(slots) = self.slots else {
            return MutableSelectionForEachOutcome::Updated { rows: 0 };
        };
        let rows = slots.len();
        for slot in slots.into_slots() {
            let StableSlotResolveOutcome::Resolved { dense_row, .. } = self
                .table
                .__indexes()
                .resolve_stable_slot(slot)
                .unwrap_or_else(|reason| invariant_violation(reason));
            self.table.update_at(dense_row, |row| update(row));
        }
        MutableSelectionForEachOutcome::Updated { rows }
    }

    fn union(self, right: DetachedSelection) -> Self {
        let slots = match (self.slots, right.slots) {
            (None, right) => right,
            (left, None) => left,
            (Some(left), Some(right)) => Some(left.union(&right)),
        };
        Self { slots, ..self }
    }

    fn intersection(self, right: DetachedSelection) -> Self {
        let slots = match (self.slots, right.slots) {
            (Some(left), Some(right)) => Some(left.intersection(&right)),
            _ => None,
        };
        Self { slots, ..self }
    }

    fn excluding(self, right: DetachedSelection) -> Self {
        let slots = match (self.slots, right.slots) {
            (left, None) => left,
            (Some(left), Some(right)) => Some(left.excluding(&right)),
            (None, Some(_)) => None,
        };
        Self { slots, ..self }
    }

    fn union_excluding_overlap(self, right: DetachedSelection) -> Self {
        let slots = match (self.slots, right.slots) {
            (None, right) => right,
            (left, None) => left,
            (Some(left), Some(right)) => Some(left.union_excluding_overlap(&right)),
        };
        Self { slots, ..self }
    }
}

#[doc(hidden)]
pub trait MutableSelectionOperand<'table, Row: WarpTableRow, Brand> {
    type Output;

    fn and_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output;
    fn or_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output;
    fn and_not_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output;
    fn xor_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output;
}

impl<'table, Row: WarpTableRow + 'table, Brand: 'table, Operand>
    MutableSelectionOperand<'table, Row, Brand> for Operand
where
    Operand: Into<DetachedSelection>,
{
    type Output = MutableSelection<'table, Row, Brand>;

    fn and_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output {
        left.intersection(self.into())
    }

    fn or_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output {
        left.union(self.into())
    }

    fn and_not_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output {
        left.excluding(self.into())
    }

    fn xor_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output {
        left.union_excluding_overlap(self.into())
    }
}

impl<'table, Row, Brand, Operand, Error> MutableSelectionOperand<'table, Row, Brand>
    for Result<Operand, Error>
where
    Row: WarpTableRow + 'table,
    Brand: 'table,
    Operand: Into<DetachedSelection>,
{
    type Output = Result<MutableSelection<'table, Row, Brand>, Error>;

    fn and_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output {
        self.map(|right| left.intersection(right.into()))
    }

    fn or_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output {
        self.map(|right| left.union(right.into()))
    }

    fn and_not_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output {
        self.map(|right| left.excluding(right.into()))
    }

    fn xor_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output {
        self.map(|right| left.union_excluding_overlap(right.into()))
    }
}

#[must_use]
pub enum MutableSecondaryIndexLookupOutcome<'table, Row: WarpTableRow, Brand, SecondaryKey> {
    Present {
        secondary_key: SecondaryKey,
        selection: MutableSelection<'table, Row, Brand>,
    },
    Absent {
        secondary_key: SecondaryKey,
        selection: MutableSelection<'table, Row, Brand>,
    },
}

#[must_use]
pub enum MutableRangeIndexLookupOutcome<'table, Row: WarpTableRow, Brand, Key> {
    Present {
        bounds: RangeIndexBounds<Key>,
        selection: MutableSelection<'table, Row, Brand>,
    },
    Absent {
        bounds: RangeIndexBounds<Key>,
        selection: MutableSelection<'table, Row, Brand>,
    },
}

macro_rules! impl_mutable_lookup {
    ($outcome:ident) => {
        impl<'table, Row: WarpTableRow, Brand, Key> $outcome<'table, Row, Brand, Key> {
            pub fn len(&self) -> u64 {
                self.selection().len()
            }

            pub fn is_empty(&self) -> bool {
                self.selection().is_empty()
            }

            mutable_algebra_methods!(self, self.into_selection());

            pub fn for_each(
                self,
                update: impl FnMut(Row::Mut<'_>),
            ) -> MutableSelectionForEachOutcome
            where
                Row::Indexes: super::rows::StableSlotResolver,
            {
                self.into_selection().for_each(update)
            }

            fn selection(&self) -> &MutableSelection<'table, Row, Brand> {
                match self {
                    Self::Present { selection, .. } | Self::Absent { selection, .. } => selection,
                }
            }

            fn into_selection(self) -> MutableSelection<'table, Row, Brand> {
                match self {
                    Self::Present { selection, .. } | Self::Absent { selection, .. } => selection,
                }
            }
        }
    };
}

impl_mutable_lookup!(MutableSecondaryIndexLookupOutcome);
impl_mutable_lookup!(MutableRangeIndexLookupOutcome);

#[doc(hidden)]
pub fn bind_detached_secondary_lookup<SecondaryKey>(
    membership: (SecondaryKey, Option<SlotSet>),
) -> DetachedSecondaryIndexLookupOutcome<SecondaryKey> {
    let (secondary_key, slots) = membership;
    let present = slots.is_some();
    let selection = DetachedSelection { slots };
    if present {
        DetachedSecondaryIndexLookupOutcome::Present {
            secondary_key,
            selection,
        }
    } else {
        DetachedSecondaryIndexLookupOutcome::Absent {
            secondary_key,
            selection,
        }
    }
}

#[doc(hidden)]
pub fn bind_detached_range_lookup<Key>(
    membership: RangeIndexMembership<Key>,
) -> DetachedRangeIndexLookupOutcome<Key> {
    let (bounds, slots) = membership.into_parts();
    let present = slots.is_some();
    let selection = DetachedSelection { slots };
    if present {
        DetachedRangeIndexLookupOutcome::Present { bounds, selection }
    } else {
        DetachedRangeIndexLookupOutcome::Absent { bounds, selection }
    }
}

#[doc(hidden)]
pub fn bind_mutable_secondary_lookup<'table, Row, Brand, SecondaryKey>(
    table: &'table mut WarpTable<Row, Brand>,
    membership: (SecondaryKey, Option<SlotSet>),
) -> MutableSecondaryIndexLookupOutcome<'table, Row, Brand, SecondaryKey>
where
    Row: WarpTableRow,
{
    let (secondary_key, slots) = membership;
    let present = slots.is_some();
    let selection = MutableSelection::new(table, slots);
    if present {
        MutableSecondaryIndexLookupOutcome::Present {
            secondary_key,
            selection,
        }
    } else {
        MutableSecondaryIndexLookupOutcome::Absent {
            secondary_key,
            selection,
        }
    }
}

#[doc(hidden)]
pub fn bind_mutable_range_lookup<'table, Row, Brand, Key>(
    table: &'table mut WarpTable<Row, Brand>,
    membership: RangeIndexMembership<Key>,
) -> MutableRangeIndexLookupOutcome<'table, Row, Brand, Key>
where
    Row: WarpTableRow,
{
    let (bounds, slots) = membership.into_parts();
    let present = slots.is_some();
    let selection = MutableSelection::new(table, slots);
    if present {
        MutableRangeIndexLookupOutcome::Present { bounds, selection }
    } else {
        MutableRangeIndexLookupOutcome::Absent { bounds, selection }
    }
}