Checkpoint: batch handlers with cursor

src/handlers/handler.rs

@@ -1,3 +1,6 @@
+use std::str::FromStr;
+use std::time::Duration;
+
 use nu_engine::eval_block_with_early_return;
 use nu_protocol::debugger::WithoutDebug;
 use nu_protocol::engine::Stack;
@@ -10,19 +13,31 @@ use scru128::Scru128Id;
 use crate::error::Error;
 use crate::nu;
 use crate::nu::frame_to_value;
+use crate::store::{FollowOption, Frame, ReadOptions, Store};
 use crate::thread_pool::ThreadPool;
+use crate::ttl::TTL;
 
-#[derive(Clone, Debug, serde::Deserialize)]
-#[serde(untagged)]
-pub enum StartDefinition {
-    Head { head: String },
-}
-
-#[derive(Clone, Debug, serde::Deserialize, Default)]
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, Default)]
 pub struct Meta {
     pub initial_state: Option<serde_json::Value>,
     pub pulse: Option<u64>,
-    pub start: Option<StartDefinition>,
+    pub mode: Mode,
+}
+
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, Default)]
+#[serde(rename_all = "snake_case")]
+pub enum Mode {
+    #[default]
+    Batch,
+    Online(StartMode),
+}
+
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, Default)]
+#[serde(rename_all = "snake_case")]
+pub enum StartMode {
+    #[default]
+    Tail,
+    Head(String),
 }
 
 #[derive(Clone)]
@@ -124,4 +139,67 @@ impl Handler {
             })?
             .into_value(Span::unknown())?)
     }
+
+    pub async fn get_cursor(&self, store: &Store) -> Option<Scru128Id> {
+        store
+            .head(&format!("{}.cursor", self.topic))
+            .and_then(|frame| {
+                frame.meta.and_then(|meta| {
+                    meta.get("frame_id").and_then(|v| {
+                        v.as_str()
+                            .and_then(|id| scru128::Scru128Id::from_str(id).ok())
+                    })
+                })
+            })
+    }
+
+    pub async fn save_cursor(&self, store: &Store, frame_id: Scru128Id) {
+        let _ = store
+            .append(
+                Frame::with_topic(format!("{}.cursor", self.topic))
+                    .meta(serde_json::json!({
+                        "handler_id": self.id.to_string(),
+                        "frame_id": frame_id.to_string(),
+                    }))
+                    .ttl(TTL::Head(1))
+                    .build(),
+            )
+            .await;
+    }
+
+    pub async fn configure_read_options(&self, store: &Store) -> ReadOptions {
+        let last_id: Option<Scru128Id> = match self.meta.mode {
+            Mode::Batch => store
+                .head(&format!("{}.cursor", self.topic))
+                .and_then(|frame| {
+                    frame.meta.and_then(|meta| {
+                        meta.get("frame_id").and_then(|v| {
+                            v.as_str()
+                                .and_then(|id| scru128::Scru128Id::from_str(id).ok())
+                        })
+                    })
+                }),
+            Mode::Online(ref start_mode) => match start_mode {
+                StartMode::Tail => None,
+                StartMode::Head(ref head) => store.head(head).map(|frame| frame.id),
+            },
+        };
+
+        eprintln!("LAST_ID: {:?}", last_id.map(|id| id.to_string()));
+
+        let follow_option = self
+            .meta
+            .pulse
+            .map(|pulse| FollowOption::WithHeartbeat(Duration::from_millis(pulse)))
+            .unwrap_or(FollowOption::On);
+
+        let is_tail = matches!(self.meta.mode, Mode::Online(_)) && last_id.is_none();
+        eprintln!("Tail: {}", is_tail);
+
+        ReadOptions::builder()
+            .follow(follow_option)
+            .tail(is_tail)
+            .maybe_last_id(last_id)
+            .build()
+    }
 }

src/handlers/mod.rs

@@ -1,5 +1,5 @@
 mod handler;
 mod serve;
 
-pub use handler::{Handler, Meta, StartDefinition};
+pub use handler::{Handler, Meta, Mode};
 pub use serve::serve;