Implement DbJournal

2 files changed, 44 insertions, 1 deletions

diff --git a/server/src/main.rs b/server/src/main.rs
index 9b5adb7..6056dcd 100644
--- a/server/src/main.rs
+++ b/server/src/main.rs
@@ -1,4 +1,5 @@
 use std::{collections::HashMap, env, str::FromStr, sync::Arc};
+use serde_json::json;
 use uuid::Uuid;
 
 use auth::AuthenticatedSession;
@@ -18,7 +19,7 @@ mod auth;
 mod error;
 mod server;
 mod play;
-use crate::error::BridgeError;
+use crate::{error::BridgeError, play::{DbJournal, Journal}};
 use crate::{
     auth::{Authenticator, SessionId},
     server::ServerContext,
@@ -44,6 +45,9 @@ async fn main() {
         .await
         .expect("db connection");
 
+    let mut jnl = DbJournal::new(db_pool.clone(), Uuid::new_v4());
+    jnl.append(0, json!("starting server")).await.expect("new object");
+
     info!("Running db migrations");
     sqlx::migrate!().run(&db_pool).await.expect("db migration");
 
diff --git a/server/src/play.rs b/server/src/play.rs
index 8fe1872..242bdb8 100644
--- a/server/src/play.rs
+++ b/server/src/play.rs
@@ -1,4 +1,6 @@
 use async_trait::async_trait;
+use sqlx::{PgPool, query};
+use uuid::Uuid;
 
 use crate::error::BridgeError;
 
@@ -11,6 +13,43 @@ pub trait Journal {
     async fn replay(&mut self, seq: i64) -> Result<Vec<serde_json::Value>, BridgeError>;
 }
 
+pub struct DbJournal {
+    db: PgPool,
+    id: Uuid,
+}
+
+impl DbJournal {
+    pub fn new(db: PgPool, id: Uuid) -> Self {
+	Self { db, id }
+    }
+}
+
+#[async_trait]
+impl Journal for DbJournal {
+    async fn append(&mut self,seq:i64,payload:serde_json::Value) -> Result<(),BridgeError> {
+	query!(
+	    r#"
+              insert into object_journal (id, seq, payload)
+              values ($1, $2, $3)
+            "#,
+	    self.id, seq, payload,
+	).execute(&self.db).await?;
+	Ok(())
+    }
+
+    async fn replay(&mut self,seq:i64) -> Result<Vec<serde_json::Value>,BridgeError> {
+	let results = query!(
+	    r#"
+              select payload from object_journal
+              where id = $1 and seq >= $2
+              order by seq
+            "#,
+	    self.id, seq
+	    ).fetch_all(&self.db).await?;
+	Ok(results.into_iter().map(|v| v.payload).collect())
+    }
+}
+
 pub struct Table {}
 
 #[cfg(test)]