resource sync state

bin/core/src/api/execute/sync.rs

@@ -18,7 +18,7 @@ use monitor_client::{
     user::{sync_user, User},
   },
 };
-use mungos::by_id::update_one_by_id;
+use mungos::{by_id::update_one_by_id, mongodb::bson::to_document};
 use resolver_api::Resolve;
 use serror::serialize_error_pretty;
 
@@ -33,7 +33,7 @@ use crate::{
     },
     update::update_update,
   },
-  resource,
+  resource::{self, refresh_resource_sync_state_cache},
   state::{db_client, State},
 };
 
@@ -319,8 +319,10 @@ impl Resolve<RunSync, (User, Update)> for State {
       .await,
     );
 
+    let db = db_client().await;
+
     if let Err(e) = update_one_by_id(
-      &db_client().await.resource_syncs,
+      &db.resource_syncs,
       &sync.id,
       doc! {
         "$set": {
@@ -357,6 +359,21 @@ impl Resolve<RunSync, (User, Update)> for State {
     }
 
     update.finalize();
+
+    // Need to manually update the update before cache refresh,
+    // and before broadcast with add_update.
+    // The Err case of to_document should be unreachable,
+    // but will fail to update cache in that case.
+    if let Ok(update_doc) = to_document(&update) {
+      let _ = update_one_by_id(
+        &db.updates,
+        &update.id,
+        mungos::update::Update::Set(update_doc),
+        None,
+      )
+      .await;
+      refresh_resource_sync_state_cache().await;
+    }
     update_update(update.clone()).await?;
 
     Ok(update)

bin/core/src/api/read/sync.rs

@@ -4,8 +4,8 @@ use monitor_client::{
   entities::{
     permission::PermissionLevel,
     sync::{
-      ResourceSync, ResourceSyncActionState, ResourceSyncListItem,
-      ResourceSyncState,
+      PendingSyncUpdatesData, ResourceSync, ResourceSyncActionState,
+      ResourceSyncListItem, ResourceSyncState,
     },
     user::User,
   },
@@ -96,6 +96,19 @@ impl Resolve<GetResourceSyncsSummary, User> for State {
     for resource_sync in resource_syncs {
       res.total += 1;
 
+      match resource_sync.info.pending.data {
+        PendingSyncUpdatesData::Ok(data) => {
+          if !data.no_updates() {
+            res.pending += 1;
+            continue;
+          }
+        }
+        PendingSyncUpdatesData::Err(_) => {
+          res.failed += 1;
+          continue;
+        }
+      }
+
       match (
         cache.get(&resource_sync.id).await.unwrap_or_default(),
         action_states
@@ -115,6 +128,9 @@ impl Resolve<GetResourceSyncsSummary, User> for State {
         (ResourceSyncState::Syncing, _) => {
           unreachable!()
         }
+        (ResourceSyncState::Pending, _) => {
+          unreachable!()
+        }
       }
     }
 

bin/core/src/main.rs

@@ -38,6 +38,7 @@ async fn app() -> anyhow::Result<()> {
   resource::spawn_build_state_refresh_loop();
   resource::spawn_repo_state_refresh_loop();
   resource::spawn_procedure_state_refresh_loop();
+  resource::spawn_resource_sync_state_refresh_loop();
 
   // Setup static frontend services
   let frontend_path = frontend_path();

bin/core/src/resource/mod.rs

@@ -59,6 +59,10 @@ pub use procedure::{
 pub use repo::{
   refresh_repo_state_cache, spawn_repo_state_refresh_loop,
 };
+pub use sync::{
+  refresh_resource_sync_state_cache,
+  spawn_resource_sync_state_refresh_loop,
+};
 
 /// Implement on each monitor resource for common methods
 pub trait MonitorResource {

bin/core/src/resource/sync.rs

@@ -1,18 +1,27 @@
+use std::time::Duration;
+
+use anyhow::Context;
+use mongo_indexed::doc;
 use monitor_client::entities::{
   resource::Resource,
   sync::{
-    PartialResourceSyncConfig, ResourceSync, ResourceSyncConfig,
-    ResourceSyncConfigDiff, ResourceSyncInfo, ResourceSyncListItem,
-    ResourceSyncListItemInfo, ResourceSyncQuerySpecifics,
-    ResourceSyncState,
+    PartialResourceSyncConfig, PendingSyncUpdatesData, ResourceSync,
+    ResourceSyncConfig, ResourceSyncConfigDiff, ResourceSyncInfo,
+    ResourceSyncListItem, ResourceSyncListItemInfo,
+    ResourceSyncQuerySpecifics, ResourceSyncState,
   },
   update::{ResourceTargetVariant, Update},
   user::User,
   Operation,
 };
-use mungos::mongodb::Collection;
+use mungos::{
+  find::find_collect,
+  mongodb::{options::FindOneOptions, Collection},
+};
 
-use crate::state::{action_states, db_client};
+use crate::state::{
+  action_states, db_client, resource_sync_state_cache,
+};
 
 impl super::MonitorResource for ResourceSync {
   type Config = ResourceSyncConfig;
@@ -34,6 +43,11 @@ impl super::MonitorResource for ResourceSync {
   async fn to_list_item(
     resource_sync: Resource<Self::Config, Self::Info>,
   ) -> Self::ListItem {
+    let state = get_resource_sync_state(
+      &resource_sync.id,
+      &resource_sync.info.pending.data,
+    )
+    .await;
     ResourceSyncListItem {
       id: resource_sync.id,
       name: resource_sync.name,
@@ -42,10 +56,10 @@ impl super::MonitorResource for ResourceSync {
       info: ResourceSyncListItemInfo {
         repo: resource_sync.config.repo,
         branch: resource_sync.config.branch,
-        last_sync_ts: 0,
-        last_sync_hash: String::new(),
-        last_sync_message: String::new(),
-        state: ResourceSyncState::Unknown,
+        last_sync_ts: resource_sync.info.last_sync_ts,
+        last_sync_hash: resource_sync.info.last_sync_hash,
+        last_sync_message: resource_sync.info.last_sync_message,
+        state,
       },
     }
   }
@@ -124,3 +138,107 @@ impl super::MonitorResource for ResourceSync {
     Ok(())
   }
 }
+
+pub fn spawn_resource_sync_state_refresh_loop() {
+  tokio::spawn(async move {
+    loop {
+      refresh_resource_sync_state_cache().await;
+      tokio::time::sleep(Duration::from_secs(60)).await;
+    }
+  });
+}
+
+pub async fn refresh_resource_sync_state_cache() {
+  let _ = async {
+    let resource_syncs =
+      find_collect(&db_client().await.resource_syncs, None, None)
+        .await
+        .context("failed to get resource_syncs from db")?;
+    let cache = resource_sync_state_cache();
+    for resource_sync in resource_syncs {
+      let state =
+        get_resource_sync_state_from_db(&resource_sync.id).await;
+      cache.insert(resource_sync.id, state).await;
+    }
+    anyhow::Ok(())
+  }
+  .await
+  .inspect_err(|e| {
+    error!("failed to refresh resource_sync state cache | {e:#}")
+  });
+}
+
+async fn get_resource_sync_state(
+  id: &String,
+  data: &PendingSyncUpdatesData,
+) -> ResourceSyncState {
+  if let Some(state) = action_states()
+    .resource_sync
+    .get(id)
+    .await
+    .and_then(|s| {
+      s.get()
+        .map(|s| {
+          if s.syncing {
+            Some(ResourceSyncState::Syncing)
+          } else {
+            None
+          }
+        })
+        .ok()
+    })
+    .flatten()
+  {
+    return state;
+  }
+  let data = match data {
+    PendingSyncUpdatesData::Err(_) => {
+      return ResourceSyncState::Failed
+    }
+    PendingSyncUpdatesData::Ok(data) => data,
+  };
+  if !data.no_updates() {
+    return ResourceSyncState::Pending;
+  }
+  resource_sync_state_cache()
+    .get(id)
+    .await
+    .unwrap_or_default()
+}
+
+async fn get_resource_sync_state_from_db(
+  id: &str,
+) -> ResourceSyncState {
+  async {
+    let state = db_client()
+      .await
+      .updates
+      .find_one(
+        doc! {
+          "target.type": "ResourceSync",
+          "target.id": id,
+          "operation": "RunSync"
+        },
+        FindOneOptions::builder()
+          .sort(doc! { "start_ts": -1 })
+          .build(),
+      )
+      .await?
+      .map(|u| {
+        if u.success {
+          ResourceSyncState::Ok
+        } else {
+          ResourceSyncState::Failed
+        }
+      })
+      .unwrap_or(ResourceSyncState::Ok);
+    anyhow::Ok(state)
+  }
+  .await
+  .inspect_err(|e| {
+    warn!(
+      "failed to get resource sync state from db for {id} | {e:#}"
+    )
+  })
+  .unwrap_or(ResourceSyncState::Unknown)
+}

client/core/rs/src/api/read/sync.rs

@@ -104,6 +104,8 @@ pub struct GetResourceSyncsSummaryResponse {
   pub ok: u32,
   /// The number of syncs currently syncing.
   pub syncing: u32,
+  /// The number of syncs with pending updates
+  pub pending: u32,
   /// The number of syncs with failed state.
   pub failed: u32,
   /// The number of syncs with unknown state.

client/core/rs/src/entities/sync.rs

@@ -37,12 +37,14 @@ pub struct ResourceSyncListItemInfo {
   Debug, Clone, Copy, Default, Serialize, Deserialize, Display,
 )]
 pub enum ResourceSyncState {
-  /// Last sync successful (or never synced)
+  /// Last sync successful (or never synced). No Changes pending
   Ok,
   /// Last sync failed
   Failed,
   /// Currently syncing
   Syncing,
+  /// Updates pending
+  Pending,
   /// Other case
   #[default]
   Unknown,

client/core/ts/src/types.ts

@@ -1210,12 +1210,14 @@ export type ResourceSync = Resource<ResourceSyncConfig, ResourceSyncInfo>;
 export type GetResourceSyncResponse = ResourceSync;
 
 export enum ResourceSyncState {
-	/** Last sync successful (or never synced) */
+	/** Last sync successful (or never synced). No Changes pending */
 	Ok = "Ok",
 	/** Last sync failed */
 	Failed = "Failed",
 	/** Currently syncing */
 	Syncing = "Syncing",
+	/** Updates pending */
+	Pending = "Pending",
 	/** Other case */
 	Unknown = "Unknown",
 }
@@ -2656,6 +2658,8 @@ export interface GetResourceSyncsSummaryResponse {
 	ok: number;
 	/** The number of syncs currently syncing. */
 	syncing: number;
+	/** The number of syncs with pending updates */
+	pending: number;
 	/** The number of syncs with failed state. */
 	failed: number;
 	/** The number of syncs with unknown state. */