flusso-kernel 0.17.0

use std::hash::{Hash, Hasher};

use serde::{Deserialize, Serialize};

use crate::{Field, common};

use super::Filter;

/// Folds rows from a related `table` into the document. The [`kind`](Self::kind)
/// names the relationship — which side carries the key, and whether one row or
/// many fold in; `filters`, `order_by`, and `limit` narrow and shape the rows
/// that come back.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Join {
    pub table: common::TableName,
    pub kind: JoinKind,
    pub primary_key: common::ColumnName,
    /// Whether the folded-in object may be absent. Only meaningful for a to-one
    /// join (`belongs_to`/`has_one`); a to-many join is always a non-null array.
    /// A to-one join defaults to nullable unless the schema marks it `required`.
    #[serde(default)]
    pub nullable: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub filters: Option<Vec<Filter>>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub order_by: Option<Vec<OrderBy>>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit: Option<u64>,
    pub fields: Vec<Field>,
    /// Extra column equalities on the hop, ANDed with the key. Never set on a
    /// [`ManyToMany`](JoinKind::ManyToMany) join.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub on: Vec<ColumnPair>,
    /// Whether a sink indexes the folded-in rows. `false` still builds them
    /// into the stored document, but maps the whole subtree as one disabled
    /// object: nothing under it is searchable, filterable or sortable, and it
    /// costs one mapping field.
    #[serde(
        default = "super::field::enabled",
        skip_serializing_if = "super::field::is_enabled"
    )]
    pub indexed: bool,
}

/// Hashes `indexed` only when turned off and `on` only when non-empty, so a
/// schema that never opts in keeps the content hash (and so the physical
/// index) it had before the fields existed.
impl Hash for Join {
    fn hash<H: Hasher>(&self, state: &mut H) {
        let Self {
            table,
            kind,
            primary_key,
            nullable,
            filters,
            order_by,
            limit,
            fields,
            on,
            indexed,
        } = self;
        table.hash(state);
        kind.hash(state);
        primary_key.hash(state);
        nullable.hash(state);
        filters.hash(state);
        order_by.hash(state);
        limit.hash(state);
        fields.hash(state);
        if !indexed {
            "indexed".hash(state);
        }
        hash_on("on", on, state);
    }
}

/// One extra equality on a relation hop: `local` is a column of the table
/// before the hop (the parent row), `related` a column of the hop's table.
/// SQL equality, so a null on either side matches nothing.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct ColumnPair {
    /// The column on the table before the hop.
    pub local: common::ColumnName,
    /// The column on the hop's own table.
    pub related: common::ColumnName,
}

/// Hashes a hop's `on` pairs only when there are any, so a relation without
/// them keeps its content hash. `tag` keeps a `via` hop's pairs apart from the
/// aggregate's own, which would otherwise hash to the same bytes.
pub(super) fn hash_on<H: Hasher>(tag: &str, on: &[ColumnPair], state: &mut H) {
    if !on.is_empty() {
        tag.hash(state);
        on.hash(state);
    }
}

/// The relationship a join expresses. The verb carries both the cardinality
/// and — the part that matters — **which table holds the key**:
///
/// - [`BelongsTo`](Self::BelongsTo): *this* row points at the related row
///   (`column` is on the parent table) → a single object.
/// - [`HasOne`](Self::HasOne) / [`HasMany`](Self::HasMany): the related rows
///   point back at this one (`foreign_key` is on the related table) → a single
///   object / a nested array.
/// - [`ManyToMany`](Self::ManyToMany): both sides are keyed through a junction
///   table → a nested array.
#[derive(Debug, Clone, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum JoinKind {
    BelongsTo { column: common::ColumnName },
    HasOne { foreign_key: common::ColumnName },
    HasMany { foreign_key: common::ColumnName },
    ManyToMany { through: Through },
}

impl JoinKind {
    /// Whether this join folds in many rows (a nested array) rather than one
    /// (an object).
    pub fn is_to_many(&self) -> bool {
        matches!(self, JoinKind::HasMany { .. } | JoinKind::ManyToMany { .. })
    }
}

/// How an aggregate's related rows tie back to the parent — a direct FK on the
/// aggregated table, or a junction table. (Joins carry their key inside
/// [`JoinKind`]; aggregates are inherently over-many, so `belongs_to` has no
/// aggregate counterpart.)
#[derive(Debug, Clone, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AggregateKey {
    Direct(common::ColumnName),
    Through(Through),
}

/// A junction table linking two sides of a many-to-many relation.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Through {
    pub table: common::TableName,
    pub left_key: common::ColumnName,
    pub right_key: common::ColumnName,
    /// Conditions on the junction rows themselves (e.g. skip cancelled order
    /// lines), as opposed to the relation's own `filters` on the target table.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub filters: Option<Vec<Filter>>,
}

/// Hashes `filters` only when non-empty, so a schema that never opts in keeps
/// the content hash (and so the physical index) it had before the field existed.
impl Hash for Through {
    fn hash<H: Hasher>(&self, state: &mut H) {
        let Self {
            table,
            left_key,
            right_key,
            filters,
        } = self;
        table.hash(state);
        left_key.hash(state);
        right_key.hash(state);
        if let Some(filters) = filters.as_deref().filter(|f| !f.is_empty()) {
            filters.hash(state);
        }
    }
}

#[derive(Debug, Clone, Hash, Serialize, Deserialize)]
pub struct OrderBy {
    pub column: common::ColumnName,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub direction: Option<Direction>,
}

#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Direction {
    Asc,
    Desc,
}