feat: allow restoring and removing archive via env var (#130)

integrationos-archiver/Cargo.toml

@@ -20,6 +20,7 @@ reqwest-retry = "0.6.1"
 reqwest-tracing = "0.5.3"
 serde.workspace = true
 serde_json.workspace = true
+strum = { workspace = true, features = ["derive"] }
 tempfile = "3.12.0"
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
 tokio-util = "0.7.11"

integrationos-archiver/src/config.rs

@@ -1,6 +1,15 @@
 use envconfig::Envconfig;
 use integrationos_domain::database::DatabaseConfig;
 use std::fmt::{Display, Formatter};
+use strum::{AsRefStr, EnumString};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, EnumString, AsRefStr)]
+#[strum(serialize_all = "kebab-case")]
+pub enum Mode {
+    Dump,
+    DumpDelete,
+    Restore,
+}
 
 #[derive(Envconfig, Clone)]
 pub struct ArchiverConfig {
@@ -19,6 +28,8 @@ pub struct ArchiverConfig {
     pub read_buffer_size: usize,
     #[envconfig(from = "PROCESSING_CHUNK_TIMEOUT_SECS", default = "30")]
     pub processing_chunk_timeout_secs: u64,
+    #[envconfig(from = "MODE", default = "dump")]
+    pub mode: Mode,
 }
 
 impl Display for ArchiverConfig {
@@ -28,12 +39,13 @@ impl Display for ArchiverConfig {
         writeln!(f, "GS_STORAGE_BUCKET: {}", self.gs_storage_bucket)?;
         writeln!(f, "GS_STORAGE_URI: {}", self.gs_storage_uri)?;
         writeln!(f, "MAX_RETRIES: {}", self.max_retries)?;
-        write!(
+        writeln!(
             f,
             "PROCESSING_CHUNK_TIMEOUT_SECS: {}",
             self.processing_chunk_timeout_secs
         )?;
         writeln!(f, "READ_BUFFER_SIZE_BYTES: {}", self.read_buffer_size)?;
+        writeln!(f, "MODE: {}", self.mode.as_ref())?;
         write!(f, "{}", self.db_config)
     }
 }

integrationos-archiver/src/event/completed.rs

@@ -1,5 +1,5 @@
 use super::EventMetadata;
-use chrono::{DateTime, Utc};
+use chrono::{DateTime, NaiveDate, Utc};
 use integrationos_domain::Id;
 use serde::{Deserialize, Serialize};
 
@@ -19,6 +19,16 @@ impl Completed {
             completed_at: Utc::now(),
         }
     }
+
+    pub fn date(&self) -> NaiveDate {
+        self.completed_at.date_naive()
+    }
+
+    #[cfg(test)]
+    pub fn with_date(mut self, date: DateTime<Utc>) -> Self {
+        self.completed_at = date;
+        self
+    }
 }
 
 impl EventMetadata for Completed {

integrationos-archiver/src/event/dumped.rs

@@ -1,5 +1,5 @@
 use super::EventMetadata;
-use chrono::{DateTime, Utc};
+use chrono::{DateTime, NaiveDate, Utc};
 use integrationos_domain::Id;
 use serde::{Deserialize, Serialize};
 
@@ -17,6 +17,10 @@ impl Dumped {
             dumped_at: Utc::now(),
         }
     }
+
+    pub fn date(&self) -> NaiveDate {
+        self.dumped_at.date_naive()
+    }
 }
 
 impl EventMetadata for Dumped {

integrationos-archiver/src/event/failed.rs

@@ -1,5 +1,5 @@
 use super::EventMetadata;
-use chrono::{DateTime, Utc};
+use chrono::{DateTime, NaiveDate, Utc};
 use integrationos_domain::Id;
 use serde::{Deserialize, Serialize};
 
@@ -19,6 +19,10 @@ impl Failed {
             failed_at: Utc::now(),
         }
     }
+
+    pub fn date(&self) -> NaiveDate {
+        self.failed_at.date_naive()
+    }
 }
 
 impl EventMetadata for Failed {

integrationos-archiver/src/event/mod.rs

@@ -2,13 +2,16 @@ pub mod completed;
 pub mod dumped;
 pub mod failed;
 pub mod started;
+pub mod uploaded;
 
+use chrono::NaiveDate;
 use completed::Completed;
 use dumped::Dumped;
 use failed::Failed;
 use integrationos_domain::Id;
 use serde::{Deserialize, Serialize};
 use started::Started;
+use uploaded::Uploaded;
 
 pub trait EventMetadata {
     fn reference(&self) -> Id;
@@ -20,5 +23,18 @@ pub enum Event {
     Started(Started),
     Dumped(Dumped),
     Failed(Failed),
+    Uploaded(Uploaded),
     Completed(Completed),
 }
+
+impl Event {
+    pub fn date(&self) -> NaiveDate {
+        match self {
+            Event::Started(e) => e.date(),
+            Event::Dumped(e) => e.date(),
+            Event::Failed(e) => e.date(),
+            Event::Uploaded(e) => e.date(),
+            Event::Completed(e) => e.date(),
+        }
+    }
+}

integrationos-archiver/src/event/started.rs

@@ -1,6 +1,9 @@
+use std::str::FromStr;
+
 use super::EventMetadata;
-use chrono::{DateTime, Utc};
-use integrationos_domain::{prefix::IdPrefix, Id};
+use anyhow::Result;
+use chrono::{DateTime, NaiveDate, Utc};
+use integrationos_domain::{prefix::IdPrefix, Id, Store};
 use serde::{Deserialize, Serialize};
 
 #[derive(Debug, Serialize, Deserialize, Clone)]
@@ -9,14 +12,25 @@ pub struct Started {
     #[serde(rename = "_id")]
     id: Id,
     started_at: DateTime<Utc>,
+    collection: Store,
 }
 
 impl Started {
-    pub fn new() -> Self {
-        Self {
+    pub fn new(collection: String) -> Result<Self> {
+        let store = Store::from_str(&collection).map_err(|e| anyhow::anyhow!(e))?;
+        Ok(Self {
             id: Id::now(IdPrefix::Archive),
             started_at: Utc::now(),
-        }
+            collection: store,
+        })
+    }
+
+    pub fn collection(&self) -> &Store {
+        &self.collection
+    }
+
+    pub fn date(&self) -> NaiveDate {
+        self.started_at.date_naive()
     }
 }
 

integrationos-archiver/src/event/uploaded.rs

@@ -0,0 +1,30 @@
+use super::EventMetadata;
+use chrono::{DateTime, NaiveDate, Utc};
+use integrationos_domain::Id;
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+#[serde(rename_all = "camelCase")]
+pub struct Uploaded {
+    id: Id,
+    uploaded_at: DateTime<Utc>,
+}
+
+impl Uploaded {
+    pub fn new(id: Id) -> Self {
+        Self {
+            id,
+            uploaded_at: Utc::now(),
+        }
+    }
+
+    pub fn date(&self) -> NaiveDate {
+        self.uploaded_at.date_naive()
+    }
+}
+
+impl EventMetadata for Uploaded {
+    fn reference(&self) -> Id {
+        self.id
+    }
+}