Importer V3 (#395)

* start of importer refactor

* fish

* resh

* zsh

1 files changed, 58 insertions, 77 deletions

diff --git a/atuin-client/src/import/resh.rs b/atuin-client/src/import/resh.rs
index 3eea84d7..75487fee 100644
--- a/atuin-client/src/import/resh.rs
+++ b/atuin-client/src/import/resh.rs
@@ -1,9 +1,6 @@
-use std::{
-    fs::File,
-    io::{BufRead, BufReader},
-    path::{Path, PathBuf},
-};
+use std::{fs::File, io::Read, path::PathBuf};
 
+use async_trait::async_trait;
 use chrono::{TimeZone, Utc};
 use directories::UserDirs;
 use eyre::{eyre, Result};
@@ -11,7 +8,7 @@ use serde::Deserialize;
 
 use atuin_common::utils::uuid_v4;
 
-use super::{count_lines, Importer};
+use super::{get_histpath, unix_byte_lines, Importer, Loader};
 use crate::history::History;
 
 #[derive(Deserialize, Debug)]
@@ -72,88 +69,72 @@ pub struct ReshEntry {
 
 #[derive(Debug)]
 pub struct Resh {
-    file: BufReader<File>,
-    strbuf: String,
-    loc: usize,
+    bytes: Vec<u8>,
 }
 
+fn default_histpath() -> Result<PathBuf> {
+    let user_dirs = UserDirs::new().ok_or_else(|| eyre!("could not find user directories"))?;
+    let home_dir = user_dirs.home_dir();
+
+    Ok(home_dir.join(".resh_history.json"))
+}
+
+#[async_trait]
 impl Importer for Resh {
     const NAME: &'static str = "resh";
 
-    fn histpath() -> Result<PathBuf> {
-        let user_dirs = UserDirs::new().unwrap();
-        let home_dir = user_dirs.home_dir();
-
-        Ok(home_dir.join(".resh_history.json"))
+    async fn new() -> Result<Self> {
+        let mut bytes = Vec::new();
+        let path = get_histpath(default_histpath)?;
+        let mut f = File::open(path)?;
+        f.read_to_end(&mut bytes)?;
+        Ok(Self { bytes })
     }
 
-    fn parse(path: impl AsRef<Path>) -> Result<Self> {
-        let file = File::open(path)?;
-        let mut buf = BufReader::new(file);
-        let loc = count_lines(&mut buf)?;
-
-        Ok(Self {
-            file: buf,
-            strbuf: String::new(),
-            loc,
-        })
+    async fn entries(&mut self) -> Result<usize> {
+        Ok(super::count_lines(&self.bytes))
     }
-}
 
-impl Iterator for Resh {
-    type Item = Result<History>;
+    async fn load(self, h: &mut impl Loader) -> Result<()> {
+        for b in unix_byte_lines(&self.bytes) {
+            let s = match std::str::from_utf8(b) {
+                Ok(s) => s,
+                Err(_) => continue, // we can skip past things like invalid utf8
+            };
+            let entry = match serde_json::from_str::<ReshEntry>(s) {
+                Ok(e) => e,
+                Err(_) => continue, // skip invalid json :shrug:
+            };
 
-    fn next(&mut self) -> Option<Self::Item> {
-        self.strbuf.clear();
-        match self.file.read_line(&mut self.strbuf) {
-            Ok(0) => return None,
-            Ok(_) => (),
-            Err(e) => return Some(Err(eyre!("failed to read line: {}", e))), // we can skip past things like invalid utf8
-        }
-
-        // .resh_history.json lies about being a json. It is actually a file containing valid json
-        // on every line. This means that the last line will throw an error, as it is just an EOF.
-        // Without the special case here, that will crash the importer.
-        let entry = match serde_json::from_str::<ReshEntry>(&self.strbuf) {
-            Ok(e) => e,
-            Err(e) if e.is_eof() => return None,
-            Err(e) => {
-                return Some(Err(eyre!(
-                    "Invalid entry found in resh_history file: {}",
-                    e
-                )))
-            }
-        };
-
-        #[allow(clippy::cast_possible_truncation)]
-        #[allow(clippy::cast_sign_loss)]
-        let timestamp = {
-            let secs = entry.realtime_before.floor() as i64;
-            let nanosecs = (entry.realtime_before.fract() * 1_000_000_000_f64).round() as u32;
-            Utc.timestamp(secs, nanosecs)
-        };
-        #[allow(clippy::cast_possible_truncation)]
-        #[allow(clippy::cast_sign_loss)]
-        let duration = {
-            let secs = entry.realtime_after.floor() as i64;
-            let nanosecs = (entry.realtime_after.fract() * 1_000_000_000_f64).round() as u32;
-            let difference = Utc.timestamp(secs, nanosecs) - timestamp;
-            difference.num_nanoseconds().unwrap_or(0)
-        };
+            #[allow(clippy::cast_possible_truncation)]
+            #[allow(clippy::cast_sign_loss)]
+            let timestamp = {
+                let secs = entry.realtime_before.floor() as i64;
+                let nanosecs = (entry.realtime_before.fract() * 1_000_000_000_f64).round() as u32;
+                Utc.timestamp(secs, nanosecs)
+            };
+            #[allow(clippy::cast_possible_truncation)]
+            #[allow(clippy::cast_sign_loss)]
+            let duration = {
+                let secs = entry.realtime_after.floor() as i64;
+                let nanosecs = (entry.realtime_after.fract() * 1_000_000_000_f64).round() as u32;
+                let difference = Utc.timestamp(secs, nanosecs) - timestamp;
+                difference.num_nanoseconds().unwrap_or(0)
+            };
 
-        Some(Ok(History {
-            id: uuid_v4(),
-            timestamp,
-            duration,
-            exit: entry.exit_code,
-            command: entry.cmd_line,
-            cwd: entry.pwd,
-            session: uuid_v4(),
-            hostname: entry.host,
-        }))
-    }
+            h.push(History {
+                id: uuid_v4(),
+                timestamp,
+                duration,
+                exit: entry.exit_code,
+                command: entry.cmd_line,
+                cwd: entry.pwd,
+                session: uuid_v4(),
+                hostname: entry.host,
+            })
+            .await?;
+        }
 
-    fn size_hint(&self) -> (usize, Option<usize>) {
-        (self.loc, Some(self.loc))
+        Ok(())
     }
 }