harmos-storage 0.1.6

/* ***
 *
 * Harmos
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * ***
 */

// Stored Entry Kinds Shared By Scans And Live Subscriptions

/* === Modules === */

mod sealed {
    pub trait Sealed {}
}

/* === Re-Exports === */

/* === Imports === */

// Standard Library Imports
use std::time::SystemTime;

// Third Party Imports
use serde::Serialize;

// Local Imports
use super::backend::{BackendSnapshot, ScanPage, StorageBackend};
use super::manager::Operation;
use super::query::{FileSelection, ObjectSelection, RecordSelection, SchemaSelection, StreamSelection};
use super::{FileReference, ObjectEntry, StorageCursor, StorageFuture, StorageSchema, StoredRecord, StoredRow};
use crate::Identifier;
use harmos_domain::model::storage::{IdentitySelection, StoredIdentity};

/* === Constants === */

/* === Macros === */

/* === Type Aliases === */

/* === Structs === */

pub struct Queued<'a>(&'a Operation);

/* === Enums === */

/* === Traits === */

pub trait Entry: Clone + Serialize + Send + Sync + 'static + sealed::Sealed {
    type Selection: Send + Sync;

    fn position(&self) -> &StorageCursor;
    fn matches(&self, selection: &Self::Selection) -> bool;

    fn start(_selection: &Self::Selection) -> u64 {
        0
    }

    fn pending(queued: Queued<'_>) -> &[Self];
    fn scan<'a>(
        backend: &'a dyn StorageBackend, snapshot: &'a BackendSnapshot, selection: &'a Self::Selection, page: ScanPage,
        memory: crate::arrow::RetainedMemory,
    ) -> StorageFuture<'a, Vec<Self>>;
}

/* === Implementations === */

impl<'a> Queued<'a> {
    pub(super) fn new(operation: &'a Operation) -> Self {
        Self(operation)
    }
}

impl sealed::Sealed for ObjectEntry {}
impl sealed::Sealed for StoredRecord {}
impl sealed::Sealed for StoredRow {}
impl sealed::Sealed for StorageSchema {}
impl sealed::Sealed for FileReference {}
impl sealed::Sealed for StoredIdentity {}

impl Entry for ObjectEntry {
    type Selection = ObjectSelection;

    fn position(&self) -> &StorageCursor {
        &self.position
    }

    // Latest selection is resolved by the backend; matching checks identity only.
    fn matches(&self, selection: &ObjectSelection) -> bool {
        self.version.check_position(&self.position).is_ok()
            && selection.schema.as_ref().is_none_or(|key| key == &self.value.schema)
            && selection.key.as_ref().is_none_or(|key| key == &self.value.item)
    }

    fn pending(queued: Queued<'_>) -> &[Self] {
        match queued.0 {
            Operation::ObjectGroup { objects, .. } => objects.as_slice(),
            _ => &[],
        }
    }

    fn scan<'a>(
        backend: &'a dyn StorageBackend, snapshot: &'a BackendSnapshot, selection: &'a ObjectSelection, page: ScanPage,
        memory: crate::arrow::RetainedMemory,
    ) -> StorageFuture<'a, Vec<Self>> {
        let future = backend.read_objects(snapshot, selection, page, memory);
        let span = tracing::trace_span!("storage.backend.scan", method = "primary.read_objects");
        Box::pin(tracing::Instrument::instrument(future, span))
    }
}

impl Entry for StoredRecord {
    type Selection = RecordSelection;

    fn position(&self) -> &StorageCursor {
        &self.position
    }

    fn matches(&self, selection: &RecordSelection) -> bool {
        self.schema == selection.schema
            && optional(&selection.scope, self.scope.as_ref())
            && optional(&selection.item, Some(&self.item))
            && within(selection.source_start, selection.source_end, self.source_time)
    }

    fn pending(queued: Queued<'_>) -> &[Self] {
        match queued.0 {
            Operation::Records(values) => values.as_slice(),
            _ => &[],
        }
    }

    fn scan<'a>(
        backend: &'a dyn StorageBackend, snapshot: &'a BackendSnapshot, selection: &'a RecordSelection, page: ScanPage,
        memory: crate::arrow::RetainedMemory,
    ) -> StorageFuture<'a, Vec<Self>> {
        let future = backend.read_records(snapshot, selection, page, memory);
        let span = tracing::trace_span!("storage.backend.scan", method = "primary.read_records");
        Box::pin(tracing::Instrument::instrument(future, span))
    }
}

impl Entry for StoredRow {
    type Selection = StreamSelection;

    fn position(&self) -> &StorageCursor {
        &self.record.position
    }

    fn matches(&self, selection: &StreamSelection) -> bool {
        let record = &self.record;
        record.schema == selection.schema
            && record.position.sequence > selection.after
            && optional(&selection.scope, record.scope.as_ref())
            && optional(&selection.item, Some(&record.item))
            && optional(&selection.stream, Some(&self.stream))
            && within(selection.source_start, selection.source_end, record.source_time)
    }

    fn start(selection: &StreamSelection) -> u64 {
        selection.after
    }

    fn pending(queued: Queued<'_>) -> &[Self] {
        match queued.0 {
            Operation::Stream(values) => values.as_slice(),
            _ => &[],
        }
    }

    fn scan<'a>(
        backend: &'a dyn StorageBackend, snapshot: &'a BackendSnapshot, selection: &'a StreamSelection, page: ScanPage,
        memory: crate::arrow::RetainedMemory,
    ) -> StorageFuture<'a, Vec<Self>> {
        let future = backend.read_stream(snapshot, selection, page, memory);
        let span = tracing::trace_span!("storage.backend.scan", method = "primary.read_stream");
        Box::pin(tracing::Instrument::instrument(future, span))
    }
}

impl Entry for StorageSchema {
    type Selection = SchemaSelection;

    fn position(&self) -> &StorageCursor {
        &self.position
    }

    fn matches(&self, selection: &SchemaSelection) -> bool {
        selection.schema.as_ref().is_none_or(|key| key == &self.key)
    }

    fn pending(queued: Queued<'_>) -> &[Self] {
        match queued.0 {
            Operation::Schema(value) => value.as_slice(),
            Operation::ObjectGroup { schemas, .. } => schemas.as_slice(),
            _ => &[],
        }
    }

    fn scan<'a>(
        backend: &'a dyn StorageBackend, snapshot: &'a BackendSnapshot, selection: &'a SchemaSelection, page: ScanPage,
        memory: crate::arrow::RetainedMemory,
    ) -> StorageFuture<'a, Vec<Self>> {
        let future = backend.read_schemas(snapshot, selection, page, memory);
        let span = tracing::trace_span!("storage.backend.scan", method = "primary.read_schemas");
        Box::pin(tracing::Instrument::instrument(future, span))
    }
}

impl Entry for FileReference {
    type Selection = FileSelection;

    fn position(&self) -> &StorageCursor {
        &self.position
    }

    fn matches(&self, selection: &FileSelection) -> bool {
        optional(&selection.scope, self.metadata.scope.as_ref()) && optional(&selection.item, Some(&self.metadata.id))
    }

    fn pending(queued: Queued<'_>) -> &[Self] {
        match queued.0 {
            Operation::File(value) => value.as_slice(),
            _ => &[],
        }
    }

    fn scan<'a>(
        backend: &'a dyn StorageBackend, snapshot: &'a BackendSnapshot, selection: &'a FileSelection, page: ScanPage,
        memory: crate::arrow::RetainedMemory,
    ) -> StorageFuture<'a, Vec<Self>> {
        let future = backend.read_files(snapshot, selection, page, memory);
        let span = tracing::trace_span!("storage.backend.scan", method = "primary.read_files");
        Box::pin(tracing::Instrument::instrument(future, span))
    }
}

// Identities Are Catalogued From Durable Storage Only; None Wait In The Queue.
impl Entry for StoredIdentity {
    type Selection = IdentitySelection;

    fn position(&self) -> &StorageCursor {
        self.first()
    }

    fn matches(&self, selection: &IdentitySelection) -> bool {
        selection.matches(self)
    }

    fn pending(_: Queued<'_>) -> &[Self] {
        &[]
    }

    fn scan<'a>(
        backend: &'a dyn StorageBackend, snapshot: &'a BackendSnapshot, selection: &'a IdentitySelection, page: ScanPage,
        memory: crate::arrow::RetainedMemory,
    ) -> StorageFuture<'a, Vec<Self>> {
        let future = backend.read_identities(snapshot, selection, page, memory);
        let span = tracing::trace_span!("storage.backend.scan", method = "primary.read_identities");
        Box::pin(tracing::Instrument::instrument(future, span))
    }
}

/* === Public Functions === */

/* === Private Functions === */

fn optional(expected: &Option<Identifier>, actual: Option<&Identifier>) -> bool {
    expected.as_ref().is_none_or(|expected| Some(expected) == actual)
}

// Half-Open Source Windows Select Only Records Carrying A Source Time.
fn within(start: Option<SystemTime>, end: Option<SystemTime>, time: Option<SystemTime>) -> bool {
    match (start, end) {
        (Some(start), Some(end)) => time.is_some_and(|time| time >= start && time < end),
        _ => true,
    }
}

/* === Tests === */