Real-time Sync: Add push logic and run method

lib/core/src/sync/mod.rs

@@ -1,17 +1,21 @@
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::time::Duration;
 
 use anyhow::{anyhow, Result};
+use futures_util::TryFutureExt;
 
+use crate::sync::model::sync::{Record, SetRecordRequest, SetRecordStatus};
+use crate::utils;
 use crate::{persist::Persister, prelude::Signer};
 
 use self::client::SyncerClient;
-use self::model::DecryptedRecord;
 use self::model::{
     data::{ChainSyncData, ReceiveSyncData, SendSyncData, SyncData},
     sync::ListChangesRequest,
     RecordType, SyncState,
 };
+use self::model::{DecryptedRecord, SyncOutgoingDetails};
 
 pub(crate) mod client;
 pub(crate) mod model;
@@ -50,6 +54,21 @@ impl SyncService {
         }
     }
 
+    async fn run(self: Arc<Self>) -> Result<()> {
+        let cloned = self.clone();
+
+        tokio::spawn(async move {
+            loop {
+                if let Err(err) = cloned.pull().and_then(|_| cloned.push()).await {
+                    log::debug!("Could not run sync event loop: {err:?}");
+                }
+                tokio::time::sleep(Duration::from_secs(30)).await;
+            }
+        });
+
+        Ok(())
+    }
+
     fn commit_record(
         &self,
         decrypted_record: &DecryptedRecord,
@@ -208,6 +227,124 @@ impl SyncService {
 
         Ok(())
     }
+
+    async fn push(&self) -> Result<()> {
+        let outgoing_details = self.persister.get_sync_outgoing_details()?;
+
+        for SyncOutgoingDetails {
+            record_id,
+            record_type,
+            ..
+        } in outgoing_details
+        {
+            // Step 1: Get the sync state, ensure it exists
+            let Some(sync_state) = unwrap_or_continue!(
+                self.persister.get_sync_state_by_record_id(&record_id),
+                "Could not retrieve sync state: {}"
+            ) else {
+                log::debug!("Cannot sync record without any sync state. Skipping.");
+                continue;
+            };
+
+            // Step 2: Fetch the sync data
+            let sync_data = unwrap_or_continue!(
+                self.load_sync_data(&sync_state.data_id, record_type),
+                "Could not load sync data: {}"
+            );
+
+            // Step 3: Create the record to push outwards
+            let record = unwrap_or_continue!(
+                Record::new(
+                    sync_state.record_id.clone(),
+                    sync_data,
+                    sync_state.record_revision,
+                    self.signer.clone()
+                ),
+                "Could not create new record: {}"
+            );
+
+            // Step 5: Push the record
+            let req = unwrap_or_continue!(
+                SetRecordRequest::new(record, utils::now(), self.signer.clone()),
+                "Could not create SetRecordRequest: {}"
+            );
+            let reply =
+                unwrap_or_continue!(self.client.push(req).await, "Could not push record: {}");
+
+            // Step 6: Check for conflict. If present, skip and retry on the next call
+            if reply.status() == SetRecordStatus::Conflict {
+                continue;
+            }
+
+            // Step 7: Update state revision
+            unwrap_or_continue!(
+                self.persister.set_sync_state(SyncState {
+                    data_id: sync_state.data_id,
+                    record_id: sync_state.record_id,
+                    record_revision: reply.new_revision,
+                    is_local: sync_state.is_local,
+                }),
+                "Could not set final sync state after push: {}"
+            );
+        }
+
+        Ok(())
+    }
+
+    pub(crate) async fn enqueue_record(
+        &self,
+        data_id: &str,
+        record_type: RecordType,
+        new_updated_fields: Option<Vec<String>>,
+    ) -> Result<()> {
+        let maybe_previous_sync_state = self.persister.get_sync_state_by_data_id(data_id)?;
+
+        let record_revision = maybe_previous_sync_state
+            .as_ref()
+            .map(|sync_state| sync_state.record_revision)
+            .unwrap_or(0);
+        let is_local = maybe_previous_sync_state
+            .as_ref()
+            .map(|sync_state| sync_state.is_local)
+            .unwrap_or(true);
+        let record_id = maybe_previous_sync_state
+            .map(|sync_state| sync_state.record_id)
+            .unwrap_or(uuid::Uuid::new_v4().to_string());
+
+        // Try to merge previous updated fields with new ones, if either is present
+        let maybe_previous_updated_fields = self
+            .persister
+            .get_sync_outgoing_details_by_id(&record_id)?
+            .map(|details| details.updated_fields)
+            .flatten();
+
+        let updated_fields = match (maybe_previous_updated_fields, new_updated_fields) {
+            (None, None) => None,
+            (Some(previous_fields), None) => Some(previous_fields),
+            (None, Some(new_fields)) => Some(new_fields),
+            (Some(previous_fields), Some(new_fields)) => {
+                Some([previous_fields, new_fields].concat())
+            }
+        };
+
+        let outgoing_details = SyncOutgoingDetails {
+            record_id: record_id.clone(),
+            record_type,
+            updated_fields,
+            commit_time: utils::now(),
+        };
+        let new_state = SyncState {
+            data_id: data_id.to_string(),
+            record_id,
+            record_revision,
+            is_local,
+        };
+
+        self.persister
+            .set_sync_outgoing_details_and_state(outgoing_details, new_state)?;
+
+        Ok(())
+    }
 }
 
 #[cfg(test)]

lib/core/src/sync/model/client.rs

@@ -1,4 +1,5 @@
 use anyhow::Result;
+use lwk_wollet::hashes::hex::DisplayHex as _;
 use openssl::sha::sha256;
 use std::sync::Arc;
 
@@ -7,7 +8,10 @@ use crate::{
     utils,
 };
 
-use super::{sync::ListChangesRequest, MESSAGE_PREFIX};
+use super::{
+    sync::{ListChangesRequest, Record, SetRecordRequest},
+    CURRENT_SCHEMA_VERSION, MESSAGE_PREFIX,
+};
 
 fn sign_message(msg: &[u8], signer: Arc<Box<dyn Signer>>) -> Result<String, SignerError> {
     let msg = [MESSAGE_PREFIX, msg].concat();
@@ -29,3 +33,25 @@ impl ListChangesRequest {
         })
     }
 }
+impl SetRecordRequest {
+    pub(crate) fn new(
+        record: Record,
+        request_time: u32,
+        signer: Arc<Box<dyn Signer>>,
+    ) -> Result<Self, SignerError> {
+        let msg = format!(
+            "{}-{}-{}-{}-{}",
+            record.id,
+            record.data.to_lower_hex_string(),
+            record.revision,
+            CURRENT_SCHEMA_VERSION.to_string(),
+            request_time,
+        );
+        let signature = sign_message(msg.as_bytes(), signer)?;
+        Ok(Self {
+            record: Some(record),
+            request_time,
+            signature,
+        })
+    }
+}