feat(sync): add RowPush wire message for server-originated writes

Origin needs to fan out row post-images for writes with no
client-authored CRDT delta (SQL DML, DDL-managed system rows). Reusing
DeltaPush was ambiguous since it carries Loro update bytes rather than
a MessagePack post-image, leaving the receiver unable to distinguish
the two encodings or the upsert/delete operation. Add a dedicated
RowPush message (0x15) with an explicit RowOp and wire it into the
sync session loop.

Author: farhan-syah

Cargo.lock

@@ -113,3 +113,59 @@ pub struct CollectionPurgedMsg {
     /// Origin WAL LSN at which the hard-delete was committed.
     pub purge_lsn: u64,
 }
+
+/// Row-level operation carried by a [`RowPushMsg`].
+#[derive(
+    Debug,
+    Clone,
+    Copy,
+    PartialEq,
+    Eq,
+    Default,
+    Serialize,
+    Deserialize,
+    zerompk::ToMessagePack,
+    zerompk::FromMessagePack,
+)]
+pub enum RowOp {
+    /// The row was created or updated; `payload` is its post-image.
+    #[default]
+    Upsert,
+    /// The row was removed; `payload` is empty.
+    Delete,
+}
+
+/// Row post-image push (server → client, 0x15).
+///
+/// Origin sends this for writes that originated on the server — SQL DML, or
+/// DDL-managed system rows such as retention policies and alerts — where there
+/// is no client-authored CRDT operation to replicate. The peer applies it as a
+/// row upsert or delete against its local state.
+///
+/// Deliberately NOT [`DeltaPushMsg`]: that message is client → server and its
+/// `delta` field is Loro update bytes. Carrying a MessagePack post-image in it
+/// would leave the receiver unable to tell the two encodings apart, and the
+/// operation (upsert vs delete) would have to be inferred from an empty
+/// payload rather than stated.
+#[derive(
+    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
+)]
+pub struct RowPushMsg {
+    /// Collection the row belongs to.
+    pub collection: String,
+    /// Row / document ID.
+    pub document_id: String,
+    /// MessagePack-encoded row post-image. Empty for [`RowOp::Delete`].
+    pub payload: Vec<u8>,
+    /// Whether the row was written or removed.
+    pub op: RowOp,
+    /// Origin-assigned WAL LSN for the write, when known (`0` otherwise).
+    #[serde(default)]
+    pub lsn: u64,
+    /// Originating peer / node ID.
+    #[serde(default)]
+    pub peer_id: u64,
+    /// Per-collection monotonic sequence, for ordering diagnostics.
+    #[serde(default)]
+    pub sequence: u64,
+}

nodedb-types/src/sync/wire/delta.rs

@@ -36,6 +36,14 @@ pub enum SyncMessageType {
     /// must drop local Loro state and remove the collection's redb
     /// record; future deltas for the collection are not sync-eligible.
     CollectionPurged = 0x14,
+    /// Row post-image push (server → client, 0x15).
+    ///
+    /// Carries a row's full post-image as MessagePack, NOT Loro update
+    /// bytes — distinct from [`Self::DeltaPush`], which is client → server
+    /// and carries a CRDT delta. Origin fans these out for writes that
+    /// originated on the server (SQL DML, DDL-managed system rows), where no
+    /// client-authored CRDT operation exists to replicate.
+    RowPush = 0x15,
     ShapeSubscribe = 0x20,
     ShapeSnapshot = 0x21,
     ShapeDelta = 0x22,
@@ -121,6 +129,7 @@ impl SyncMessageType {
             0x12 => Some(Self::DeltaReject),
             0x13 => Some(Self::CollectionSchema),
             0x14 => Some(Self::CollectionPurged),
+            0x15 => Some(Self::RowPush),
             0x20 => Some(Self::ShapeSubscribe),
             0x21 => Some(Self::ShapeSnapshot),
             0x22 => Some(Self::ShapeDelta),

nodedb-types/src/sync/wire/frame.rs

@@ -80,7 +80,9 @@ pub use array::{
 };
 pub use collection_schema::{CollectionDescriptor, CollectionSchemaSyncMsg};
 pub use columnar::{ColumnarInsertAckMsg, ColumnarInsertMsg};
-pub use delta::{CollectionPurgedMsg, DeltaAckMsg, DeltaPushMsg, DeltaRejectMsg};
+pub use delta::{
+    CollectionPurgedMsg, DeltaAckMsg, DeltaPushMsg, DeltaRejectMsg, RowOp, RowPushMsg,
+};
 pub use frame::{SyncFrame, SyncMessageType};
 pub use fts::{FtsDeleteAckMsg, FtsDeleteMsg, FtsIndexAckMsg, FtsIndexMsg};
 pub use presence::{PeerPresence, PresenceBroadcastMsg, PresenceLeaveMsg, PresenceUpdateMsg};

nodedb-types/src/sync/wire/mod.rs

@@ -439,20 +439,28 @@ pub(in crate::control::server::sync) async fn handle_sync_session(
                         .insert(delta.collection.clone());
                 }
 
-                let push_msg = nodedb_types::sync::wire::DeltaPushMsg {
+                // A row post-image, not a Loro delta — so this goes out as
+                // `RowPush`, never `DeltaPush`. The two carry different
+                // encodings, and the peer cannot tell them apart from the
+                // bytes alone.
+                let push_msg = nodedb_types::sync::wire::RowPushMsg {
                     collection: delta.collection,
                     document_id: delta.document_id,
-                    delta: delta.payload,
+                    payload: delta.payload,
+                    op: match delta.op {
+                        crate::event::crdt_sync::types::DeltaOp::Upsert => {
+                            nodedb_types::sync::wire::RowOp::Upsert
+                        }
+                        crate::event::crdt_sync::types::DeltaOp::Delete => {
+                            nodedb_types::sync::wire::RowOp::Delete
+                        }
+                    },
+                    lsn: delta.lsn,
                     peer_id: delta.peer_id,
-                    mutation_id: delta.sequence,
-                    checksum: 0,
-                    device_valid_time_ms: None,
-                    producer_id: 0,
-                    epoch: 0,
-                    seq: 0,
+                    sequence: delta.sequence,
                 };
                 if let Some(frame) = nodedb_types::sync::wire::SyncFrame::new_msgpack(
-                    nodedb_types::sync::wire::SyncMessageType::DeltaPush,
+                    nodedb_types::sync::wire::SyncMessageType::RowPush,
                     &push_msg,
                 ) && ws
                     .send(Message::Binary(frame.to_bytes().into()))