Add local event log storage (#390)

* Add event data structures

This adds the data structures required to start syncing events, rather
than syncing history directly.

Adjust event

Fix

Add event data structure to client

* Add server event table sql

* Add client event table migration

Adjust migration

* Insert into event table from client

* Add event merge function

Right now this just ensures we have the right amount of events given the
history we have

BUT it will also be used to merge CREATE/DELETE events, resulting in
history being deleted :)

* Make CI happy

* Adjust

* we don't limit history length any more

* Update atuin-client/src/database.rs

Co-authored-by: Conrad Ludgate <conradludgate@gmail.com>

* fix usage

* Fix typo

* New Rust, new clippy stuff

Co-authored-by: Conrad Ludgate <conradludgate@gmail.com>

1 files changed, 72 insertions, 1 deletions

diff --git a/atuin-client/src/database.rs b/atuin-client/src/database.rs
index ba28daf30..b0d05dbac 100644
--- a/atuin-client/src/database.rs
+++ b/atuin-client/src/database.rs
@@ -13,6 +13,7 @@ use sqlx::{
 };
 
 use super::{
+    event::{Event, EventType},
     history::History,
     ordering,
     settings::{FilterMode, SearchMode},
@@ -61,6 +62,8 @@ pub trait Database: Send + Sync {
 
     async fn update(&self, h: &History) -> Result<()>;
     async fn history_count(&self) -> Result<i64>;
+    async fn event_count(&self) -> Result<i64>;
+    async fn merge_events(&self) -> Result<i64>;
 
     async fn first(&self) -> Result<History>;
     async fn last(&self) -> Result<History>;
@@ -115,6 +118,27 @@ impl Sqlite {
         Ok(())
     }
 
+    async fn save_event(tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, e: &Event) -> Result<()> {
+        let event_type = match e.event_type {
+            EventType::Create => "create",
+            EventType::Delete => "delete",
+        };
+
+        sqlx::query(
+            "insert or ignore into events(id, timestamp, hostname, event_type, history_id)
+                values(?1, ?2, ?3, ?4, ?5)",
+        )
+        .bind(e.id.as_str())
+        .bind(e.timestamp.timestamp_nanos())
+        .bind(e.hostname.as_str())
+        .bind(event_type)
+        .bind(e.history_id.as_str())
+        .execute(tx)
+        .await?;
+
+        Ok(())
+    }
+
     async fn save_raw(tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, h: &History) -> Result<()> {
         sqlx::query(
             "insert or ignore into history(id, timestamp, duration, exit, command, cwd, session, hostname)
@@ -152,9 +176,11 @@ impl Sqlite {
 impl Database for Sqlite {
     async fn save(&mut self, h: &History) -> Result<()> {
         debug!("saving history to sqlite");
+        let event = Event::new_create(h);
 
         let mut tx = self.pool.begin().await?;
         Self::save_raw(&mut tx, h).await?;
+        Self::save_event(&mut tx, &event).await?;
         tx.commit().await?;
 
         Ok(())
@@ -166,7 +192,9 @@ impl Database for Sqlite {
         let mut tx = self.pool.begin().await?;
 
         for i in h {
-            Self::save_raw(&mut tx, i).await?
+            let event = Event::new_create(i);
+            Self::save_raw(&mut tx, i).await?;
+            Self::save_event(&mut tx, &event).await?;
         }
 
         tx.commit().await?;
@@ -302,6 +330,49 @@ impl Database for Sqlite {
         Ok(res)
     }
 
+    async fn event_count(&self) -> Result<i64> {
+        let res: i64 = sqlx::query_scalar("select count(1) from events")
+            .fetch_one(&self.pool)
+            .await?;
+
+        Ok(res)
+    }
+
+    // Ensure that we have correctly merged the event log
+    async fn merge_events(&self) -> Result<i64> {
+        // Ensure that we do not have more history locally than we do events.
+        // We can think of history as the merged log of events. There should never be more history than
+        // events, and the only time this could happen is if someone is upgrading from an old Atuin version
+        // from before we stored events.
+        let history_count = self.history_count().await?;
+        let event_count = self.event_count().await?;
+
+        if history_count > event_count {
+            // pass an empty context, because with global listing we don't care
+            let no_context = Context {
+                cwd: String::from(""),
+                session: String::from(""),
+                hostname: String::from(""),
+            };
+
+            // We're just gonna load everything into memory here. That sucks, I know, sorry.
+            // But also even if you have a LOT of history that should be fine, and we're only going to be doing this once EVER.
+            let all_the_history = self
+                .list(FilterMode::Global, &no_context, None, false)
+                .await?;
+
+            let mut tx = self.pool.begin().await?;
+            for i in all_the_history.iter() {
+                // A CREATE for every single history item is to be expected.
+                let event = Event::new_create(i);
+                Self::save_event(&mut tx, &event).await?;
+            }
+            tx.commit().await?;
+        }
+
+        Ok(0)
+    }
+
     async fn history_count(&self) -> Result<i64> {
         let res: (i64,) = sqlx::query_as("select count(1) from history")
             .fetch_one(&self.pool)