nodedb-lite 0.1.0

//! Spatial geometry insert / delete push.

use std::ops::ControlFlow;
use std::sync::Arc;

use futures::SinkExt;
use tokio::sync::Mutex;
use tokio_tungstenite::tungstenite::Message;

use nodedb_types::sync::wire::{EngineKind, SyncFrame, SyncMessageType, stream_id_for};

use super::send::send_binary;
use crate::sync::client::SyncClient;
use crate::sync::delegate::SyncDelegate;

pub(super) async fn push<S>(
    client: &Arc<SyncClient>,
    delegate: &Arc<dyn SyncDelegate>,
    sink: &Arc<Mutex<S>>,
) -> ControlFlow<()>
where
    S: SinkExt<Message> + Unpin,
    S::Error: std::fmt::Display,
{
    // The instance's own identity, not its Loro peer id: Origin dedups these
    // entries by `lite_id`, and a peer id both names a different thing and
    // changes under a collision rotation, which would split one producer's
    // stream in two.
    let lite_id = delegate.sync_identity().lite_id;
    let producer_id = client.producer_id().await;
    let epoch = client.accepted_epoch().await;

    for (durable_key, mut entry) in delegate.pending_spatial_inserts().await {
        // Announce the collection's schema before its first data frame so a
        // lite-only collection is registered on Origin before its geometries land.
        if super::control::ensure_collection_announced(client, delegate, sink, &entry.collection)
            .await
            .is_break()
        {
            return ControlFlow::Break(());
        }
        let stream_id = stream_id_for(EngineKind::Spatial, &entry.collection);
        if entry.seq == 0 {
            entry.seq = delegate.next_stream_seq(stream_id).await;
            if let Err(e) = delegate
                .persist_spatial_insert_seq(&durable_key, &entry)
                .await
            {
                tracing::warn!(error = %e, "failed to persist stream seq into durable entry; retaining for retry");
                continue;
            }
        }
        let seq = entry.seq;
        let msg = nodedb_types::sync::wire::SpatialInsertMsg {
            lite_id: lite_id.clone(),
            collection: entry.collection.clone(),
            field: entry.field.clone(),
            doc_id: entry.doc_id.clone(),
            geometry_bytes: entry.geometry_bytes.clone(),
            batch_id: entry.batch_id,
            producer_id,
            epoch,
            seq,
        };
        let Some(frame) = SyncFrame::try_encode(SyncMessageType::SpatialInsert, &msg) else {
            tracing::error!(
                collection = %entry.collection, field = %entry.field, doc_id = %entry.doc_id, batch_id = entry.batch_id,
                "failed to encode SpatialInsert frame; dropping entry"
            );
            delegate.acknowledge_spatial_insert(durable_key).await;
            continue;
        };
        if let Err(e) = send_binary(sink, frame).await {
            tracing::warn!(
                collection = %entry.collection, field = %entry.field, doc_id = %entry.doc_id,
                batch_id = entry.batch_id, error = %e,
                "SpatialInsert send failed; durable entry retained for re-send on reconnect"
            );
            return ControlFlow::Break(());
        }
        // Mark in-flight: durable entry survives until Origin ack.
        delegate
            .mark_spatial_insert_in_flight(entry.batch_id, durable_key)
            .await;
        tracing::debug!(
            collection = %entry.collection, field = %entry.field, doc_id = %entry.doc_id, batch_id = entry.batch_id,
            "sent SpatialInsert to Origin; awaiting ack before deleting durable entry"
        );
    }

    for (durable_key, mut entry) in delegate.pending_spatial_deletes().await {
        if super::control::ensure_collection_announced(client, delegate, sink, &entry.collection)
            .await
            .is_break()
        {
            return ControlFlow::Break(());
        }
        let stream_id = stream_id_for(EngineKind::Spatial, &entry.collection);
        if entry.seq == 0 {
            entry.seq = delegate.next_stream_seq(stream_id).await;
            if let Err(e) = delegate
                .persist_spatial_delete_seq(&durable_key, &entry)
                .await
            {
                tracing::warn!(error = %e, "failed to persist stream seq into durable entry; retaining for retry");
                continue;
            }
        }
        let seq = entry.seq;
        let msg = nodedb_types::sync::wire::SpatialDeleteMsg {
            lite_id: lite_id.clone(),
            collection: entry.collection.clone(),
            field: entry.field.clone(),
            doc_id: entry.doc_id.clone(),
            batch_id: entry.batch_id,
            producer_id,
            epoch,
            seq,
        };
        let Some(frame) = SyncFrame::try_encode(SyncMessageType::SpatialDelete, &msg) else {
            tracing::error!(
                collection = %entry.collection, field = %entry.field, doc_id = %entry.doc_id, batch_id = entry.batch_id,
                "failed to encode SpatialDelete frame; dropping entry"
            );
            delegate.acknowledge_spatial_delete(durable_key).await;
            continue;
        };
        if let Err(e) = send_binary(sink, frame).await {
            tracing::warn!(
                collection = %entry.collection, field = %entry.field, doc_id = %entry.doc_id,
                batch_id = entry.batch_id, error = %e,
                "SpatialDelete send failed; durable entry retained for re-send on reconnect"
            );
            return ControlFlow::Break(());
        }
        // Mark in-flight: durable entry survives until Origin ack.
        delegate
            .mark_spatial_delete_in_flight(entry.batch_id, durable_key)
            .await;
        tracing::debug!(
            collection = %entry.collection, field = %entry.field, doc_id = %entry.doc_id, batch_id = entry.batch_id,
            "sent SpatialDelete to Origin; awaiting ack before deleting durable entry"
        );
    }
    ControlFlow::Continue(())
}