backend for resource sync

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

@@ -1,4 +1,5 @@
 use anyhow::Context;
+use mongo_indexed::doc;
 use monitor_client::{
   api::execute::RunSync,
   entities::{
@@ -7,6 +8,7 @@ use monitor_client::{
     build::Build,
     builder::Builder,
     deployment::Deployment,
+    monitor_timestamp,
     permission::PermissionLevel,
     procedure::Procedure,
     repo::Repo,
@@ -16,6 +18,7 @@ use monitor_client::{
     user::User,
   },
 };
+use mungos::by_id::update_one_by_id;
 use resolver_api::Resolve;
 
 use crate::{
@@ -27,7 +30,7 @@ use crate::{
     update::update_update,
   },
   resource,
-  state::State,
+  state::{db_client, State},
 };
 
 impl Resolve<RunSync, (User, Update)> for State {
@@ -41,7 +44,7 @@ impl Resolve<RunSync, (User, Update)> for State {
     >(&sync, &user, PermissionLevel::Execute)
     .await?;
 
-    let (res, logs) =
+    let (res, logs, hash, message) =
       crate::helpers::sync::remote::get_remote_resources(&sync)
         .await
         .context("failed to get remote resources")?;
@@ -233,6 +236,26 @@ impl Resolve<RunSync, (User, Update)> for State {
       .await,
     );
 
+    if let Err(e) = update_one_by_id(
+      &db_client().await.resource_syncs,
+      &sync.id,
+      doc! {
+        "$set": {
+          "info.last_sync_ts": monitor_timestamp(),
+          "info.last_sync_hash": hash,
+          "info.last_sync_message": message,
+        }
+      },
+      None,
+    )
+    .await
+    {
+      warn!(
+        "failed to update resource sync {} info after sync | {e:#}",
+        sync.name
+      )
+    }
+
     update.finalize();
     update_update(update.clone()).await?;
 

bin/core/src/api/write/mod.rs

@@ -111,6 +111,7 @@ enum WriteRequest {
   CopyResourceSync(CopyResourceSync),
   DeleteResourceSync(DeleteResourceSync),
   UpdateResourceSync(UpdateResourceSync),
+  RefreshResourceSyncPending(RefreshResourceSyncPending),
 
   // ==== TAG ====
   CreateTag(CreateTag),

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

@@ -1,12 +1,35 @@
+use anyhow::Context;
 use monitor_client::{
   api::write::*,
   entities::{
-    permission::PermissionLevel, sync::ResourceSync, user::User,
+    self,
+    alerter::Alerter,
+    build::Build,
+    builder::Builder,
+    deployment::Deployment,
+    permission::PermissionLevel,
+    procedure::Procedure,
+    repo::Repo,
+    server::Server,
+    server_template::ServerTemplate,
+    sync::{PendingUpdates, ResourceSync},
+    user::User,
   },
 };
+use mungos::{
+  by_id::update_one_by_id,
+  mongodb::bson::{doc, to_document},
+};
 use resolver_api::Resolve;
 
-use crate::{resource, state::State};
+use crate::{
+  helpers::{
+    query::get_id_to_tags,
+    sync::resource::{get_updates_for_view, AllResourcesById},
+  },
+  resource,
+  state::{db_client, State},
+};
 
 impl Resolve<CreateResourceSync, User> for State {
   #[instrument(name = "CreateResourceSync", skip(self, user))]
@@ -59,3 +82,120 @@ impl Resolve<UpdateResourceSync, User> for State {
     resource::update::<ResourceSync>(&id, config, &user).await
   }
 }
+
+impl Resolve<RefreshResourceSyncPending, User> for State {
+  async fn resolve(
+    &self,
+    RefreshResourceSyncPending { sync }: RefreshResourceSyncPending,
+    user: User,
+  ) -> anyhow::Result<ResourceSync> {
+    // Even though this is a write request, this doesn't change any config. Anyone that can execute the
+    // sync should be able to do this.
+    let sync = resource::get_check_permissions::<
+      entities::sync::ResourceSync,
+    >(&sync, &user, PermissionLevel::Execute)
+    .await?;
+
+    let (res, _, hash, message) =
+      crate::helpers::sync::remote::get_remote_resources(&sync)
+        .await
+        .context("failed to get remote resources")?;
+    let resources = res?;
+
+    let all_resources = AllResourcesById::load().await?;
+    let id_to_tags = get_id_to_tags(None).await?;
+
+    let pending = PendingUpdates {
+      hash,
+      message,
+      server_updates: get_updates_for_view::<Server>(
+        resources.servers,
+        sync.config.delete,
+        &all_resources,
+        &id_to_tags,
+      )
+      .await
+      .context("failed to get server updates")?,
+      deployment_updates: get_updates_for_view::<Deployment>(
+        resources.deployments,
+        sync.config.delete,
+        &all_resources,
+        &id_to_tags,
+      )
+      .await
+      .context("failed to get deployment updates")?,
+      build_updates: get_updates_for_view::<Build>(
+        resources.builds,
+        sync.config.delete,
+        &all_resources,
+        &id_to_tags,
+      )
+      .await
+      .context("failed to get build updates")?,
+      repo_updates: get_updates_for_view::<Repo>(
+        resources.repos,
+        sync.config.delete,
+        &all_resources,
+        &id_to_tags,
+      )
+      .await
+      .context("failed to get repo updates")?,
+      procedure_updates: get_updates_for_view::<Procedure>(
+        resources.procedures,
+        sync.config.delete,
+        &all_resources,
+        &id_to_tags,
+      )
+      .await
+      .context("failed to get procedure updates")?,
+      alerter_updates: get_updates_for_view::<Alerter>(
+        resources.alerters,
+        sync.config.delete,
+        &all_resources,
+        &id_to_tags,
+      )
+      .await
+      .context("failed to get alerter updates")?,
+      builder_updates: get_updates_for_view::<Builder>(
+        resources.builders,
+        sync.config.delete,
+        &all_resources,
+        &id_to_tags,
+      )
+      .await
+      .context("failed to get builder updates")?,
+      server_template_updates:
+        get_updates_for_view::<ServerTemplate>(
+          resources.server_templates,
+          sync.config.delete,
+          &all_resources,
+          &id_to_tags,
+        )
+        .await
+        .context("failed to get server template updates")?,
+      resource_sync_updates: get_updates_for_view::<
+        entities::sync::ResourceSync,
+      >(
+        resources.syncs,
+        sync.config.delete,
+        &all_resources,
+        &id_to_tags,
+      )
+      .await
+      .context("failed to get resource sync updates")?,
+    };
+
+    let pending = to_document(&pending)
+      .context("failed to serialize pending to document")?;
+
+    update_one_by_id(
+      &db_client().await.resource_syncs,
+      &sync.id,
+      doc! { "$set": { "info.pending": pending } },
+      None,
+    )
+    .await?;
+
+    crate::resource::get::<ResourceSync>(&sync.id).await
+  }
+}

bin/core/src/helpers/sync/remote.rs

@@ -1,13 +1,22 @@
 use anyhow::{anyhow, Context};
 use monitor_client::entities::{
-  sync::ResourceSync, toml::ResourcesToml, update::Log, CloneArgs,
+  sync::ResourceSync, to_monitor_name, toml::ResourcesToml,
+  update::Log, CloneArgs, LatestCommit,
 };
 
 use crate::config::core_config;
 
 pub async fn get_remote_resources(
   sync: &ResourceSync,
-) -> anyhow::Result<(anyhow::Result<ResourcesToml>, Vec<Log>)> {
+) -> anyhow::Result<(
+  anyhow::Result<ResourcesToml>,
+  Vec<Log>,
+  // commit short hash
+  String,
+  // commit message
+  String,
+)> {
+  let name = to_monitor_name(&sync.name);
   let clone_args: CloneArgs = sync.into();
 
   let config = core_config();
@@ -28,6 +37,12 @@ pub async fn get_remote_resources(
       .await
       .context("failed to clone resource repo")?;
 
+  let repo_dir = config.sync_directory.join(&name);
+  let LatestCommit { hash, message } =
+    git::get_commit_hash_info(&repo_dir)
+      .await
+      .context("failed to get commit hash info")?;
+
   let repo_path = config.sync_directory.join(&sync.name);
   let resource_path = repo_path.join(&sync.config.resource_path);
 
@@ -42,5 +57,5 @@ pub async fn get_remote_resources(
     warn!("failed to remove sync repo directory | {e:?}")
   }
 
-  Ok((res, logs))
+  Ok((res, logs, hash, message))
 }

bin/core/src/helpers/sync/resource.rs

@@ -20,7 +20,7 @@ use monitor_client::{
   },
 };
 use mungos::find::find_collect;
-use partial_derive2::MaybeNone;
+use partial_derive2::{Diff, FieldDiff, MaybeNone};
 use resolver_api::Resolve;
 
 use crate::{resource::MonitorResource, state::State};
@@ -191,6 +191,139 @@ pub trait ResourceSync: MonitorResource + Sized {
   }
 }
 
+/// Turns all the diffs into a readable string
+pub async fn get_updates_for_view<Resource: ResourceSync>(
+  resources: Vec<ResourceToml<Resource::PartialConfig>>,
+  delete: bool,
+  all_resources: &AllResourcesById,
+  id_to_tags: &HashMap<String, Tag>,
+) -> anyhow::Result<Option<String>> {
+  let map = find_collect(Resource::coll().await, None, None)
+    .await
+    .context("failed to get resources from db")?
+    .into_iter()
+    .map(|r| (r.name.clone(), r))
+    .collect::<HashMap<_, _>>();
+
+  let mut any_change = false;
+
+  let mut to_delete = Vec::<String>::new();
+  if delete {
+    for resource in map.values() {
+      if !resources.iter().any(|r| r.name == resource.name) {
+        any_change = true;
+        to_delete.push(resource.name.clone())
+      }
+    }
+  }
+
+  let mut log = format!("{} Updates", Resource::resource_type());
+
+  for mut resource in resources {
+    match map.get(&resource.name) {
+      Some(original) => {
+        // First merge toml resource config (partial) onto default resource config.
+        // Makes sure things that aren't defined in toml (come through as None) actually get removed.
+        let config: Resource::Config = resource.config.into();
+        resource.config = config.into();
+
+        let diff = Resource::get_diff(
+          original.config.clone(),
+          resource.config,
+          all_resources,
+        )?;
+
+        let original_tags = original
+          .tags
+          .iter()
+          .filter_map(|id| id_to_tags.get(id).map(|t| t.name.clone()))
+          .collect::<Vec<_>>();
+
+        // Only proceed if there are any fields to update,
+        // or a change to tags / description
+        if diff.is_none()
+          && resource.description == original.description
+          && resource.tags == original_tags
+        {
+          continue;
+        }
+
+        any_change = true;
+
+        log.push_str(&format!(
+          "\n{}: {}: '{}'\n-------------------",
+          colored("UPDATE", "blue"),
+          Resource::resource_type(),
+          bold(&resource.name)
+        ));
+
+        let mut lines = Vec::<String>::new();
+        if resource.description != original.description {
+          lines.push(format!(
+            "{}: 'description'\n{}:  {}\n{}:    {}",
+            muted("field"),
+            muted("from"),
+            colored(&original.description, "red"),
+            muted("to"),
+            colored(&resource.description, "green")
+          ));
+        }
+        if resource.tags != original_tags {
+          let from = colored(&format!("{:?}", original_tags), "red");
+          let to = colored(&format!("{:?}", resource.tags), "green");
+          lines.push(format!(
+            "{}: 'tags'\n{}:  {from}\n{}:    {to}",
+            muted("field"),
+            muted("from"),
+            muted("to"),
+          ));
+        }
+        lines.extend(diff.iter_field_diffs().map(
+          |FieldDiff { field, from, to }| {
+            format!(
+              "{}: '{field}'\n{}:  {}\n{}:    {}",
+              muted("field"),
+              muted("from"),
+              colored(&from, "red"),
+              muted("to"),
+              colored(&to, "green")
+            )
+          },
+        ));
+        log.push('\n');
+        log.push_str(&lines.join("\n-------------------\n"));
+      }
+      None => {
+        any_change = true;
+        log.push_str(&format!(
+          "\n{}: {}: {}\n{}: {}\n{}: {:?}\n{}: {}",
+          colored("CREATE", "green"),
+          Resource::resource_type(),
+          bold(&resource.name),
+          muted("description"),
+          resource.description,
+          muted("tags"),
+          resource.tags,
+          muted("config"),
+          serde_json::to_string_pretty(&resource.config)
+            .context("failed to serialize config to json")?
+        ))
+      }
+    }
+  }
+
+  for name in to_delete {
+    log.push_str(&format!(
+      "\n{}: {}: '{}'\n-------------------",
+      colored("DELETE", "red"),
+      Resource::resource_type(),
+      bold(&name)
+    ));
+  }
+
+  Ok(any_change.then_some(log))
+}
+
 /// Gets all the resources to update. For use in sync execution.
 pub async fn get_updates_for_execution<Resource: ResourceSync>(
   resources: Vec<ResourceToml<Resource::PartialConfig>>,
@@ -318,15 +451,6 @@ pub async fn get_updates_for_execution<Resource: ResourceSync>(
     }
   }
 
-  for name in &to_delete {
-    // println!(
-    //   "\n{}: {}: '{}'\n-------------------",
-    //   "DELETE".red(),
-    //   Resource::display(),
-    //   name.bold(),
-    // );
-  }
-
   Ok((to_create, to_update, to_delete))
 }
 

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

@@ -82,3 +82,17 @@ pub struct UpdateResourceSync {
   /// The partial config update to apply.
   pub config: _PartialResourceSyncConfig,
 }
+
+//
+
+/// Trigger a refresh of the computed diff logs for view.
+#[typeshare]
+#[derive(
+  Serialize, Deserialize, Debug, Clone, Request, EmptyTraits,
+)]
+#[empty_traits(MonitorWriteRequest)]
+#[response(ResourceSync)]
+pub struct RefreshResourceSyncPending {
+  /// Id or name
+  pub sync: String,
+}

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

@@ -55,12 +55,41 @@ pub type ResourceSync =
 #[typeshare]
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct ResourceSyncInfo {
-  /// Unix timestamp of last sync
+  /// Unix timestamp of last applied sync
   pub last_sync_ts: I64,
-  /// Short commit hash of last sync
+  /// Short commit hash of last applied sync
   pub last_sync_hash: String,
-  /// Commit message of last sync
+  /// Commit message of last applied sync
   pub last_sync_message: String,
+  /// Readable logs of pending updates
+  pub pending: PendingUpdates,
+}
+
+#[typeshare]
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct PendingUpdates {
+  /// The commit hash which produced these pending updates
+  pub hash: String,
+  /// The commit message which produced these pending updates
+  pub message: String,
+  /// Readable log of any pending server updates
+  pub server_updates: Option<String>,
+  /// Readable log of any pending deployment updates
+  pub deployment_updates: Option<String>,
+  /// Readable log of any pending build updates
+  pub build_updates: Option<String>,
+  /// Readable log of any pending repo updates
+  pub repo_updates: Option<String>,
+  /// Readable log of any pending procedure updates
+  pub procedure_updates: Option<String>,
+  /// Readable log of any pending alerter updates
+  pub alerter_updates: Option<String>,
+  /// Readable log of any pending builder updates
+  pub builder_updates: Option<String>,
+  /// Readable log of any pending server template updates
+  pub server_template_updates: Option<String>,
+  /// Readable log of any pending resource sync updates
+  pub resource_sync_updates: Option<String>,
 }
 
 #[typeshare(serialized_as = "Partial<ResourceSyncConfig>")]

client/core/ts/src/responses.ts

@@ -218,6 +218,7 @@ export type WriteResponses = {
   CopyResourceSync: Types.ResourceSync;
   DeleteResourceSync: Types.ResourceSync;
   UpdateResourceSync: Types.ResourceSync;
+  RefreshResourceSyncPending: Types.ResourceSync;
 
   // ==== TAG ====
   CreateTag: Types.Tag;

client/core/ts/src/types.ts

@@ -1174,13 +1174,40 @@ export interface ResourceSyncConfig {
 	webhook_enabled: boolean;
 }
 
+export interface PendingUpdates {
+	/** The commit hash which produced these pending updates */
+	hash: string;
+	/** The commit message which produced these pending updates */
+	message: string;
+	/** Readable log of any pending server updates */
+	server_updates?: string;
+	/** Readable log of any pending deployment updates */
+	deployment_updates?: string;
+	/** Readable log of any pending build updates */
+	build_updates?: string;
+	/** Readable log of any pending repo updates */
+	repo_updates?: string;
+	/** Readable log of any pending procedure updates */
+	procedure_updates?: string;
+	/** Readable log of any pending alerter updates */
+	alerter_updates?: string;
+	/** Readable log of any pending builder updates */
+	builder_updates?: string;
+	/** Readable log of any pending server template updates */
+	server_template_updates?: string;
+	/** Readable log of any pending resource sync updates */
+	resource_sync_updates?: string;
+}
+
 export interface ResourceSyncInfo {
-	/** Unix timestamp of last sync */
+	/** Unix timestamp of last applied sync */
 	last_sync_ts: I64;
-	/** Short commit hash of last sync */
+	/** Short commit hash of last applied sync */
 	last_sync_hash: string;
-	/** Commit message of last sync */
+	/** Commit message of last applied sync */
 	last_sync_message: string;
+	/** Readable logs of pending updates */
+	pending: PendingUpdates;
 }
 
 export type ResourceSync = Resource<ResourceSyncConfig, ResourceSyncInfo>;
@@ -3386,6 +3413,12 @@ export interface UpdateResourceSync {
 	config: _PartialResourceSyncConfig;
 }
 
+/** Trigger a refresh of the computed diff logs for view. */
+export interface RefreshResourceSyncPending {
+	/** Id or name */
+	sync: string;
+}
+
 /** Create a tag. Response: [Tag]. */
 export interface CreateTag {
 	/** The name of the tag. */
@@ -3921,6 +3954,7 @@ export type WriteRequest =
 	| { type: "CopyResourceSync", params: CopyResourceSync }
 	| { type: "DeleteResourceSync", params: DeleteResourceSync }
 	| { type: "UpdateResourceSync", params: UpdateResourceSync }
+	| { type: "RefreshResourceSyncPending", params: RefreshResourceSyncPending }
 	| { type: "CreateTag", params: CreateTag }
 	| { type: "DeleteTag", params: DeleteTag }
 	| { type: "RenameTag", params: RenameTag }