Enhance Store struct with session management capabilities

- Added `session_defaults` and `session_advisory_locks` fields to the `Store` struct for improved session configuration.
- Updated the `connect` method to initialize these new fields.
- Refactored `session_builder` to utilize the new session defaults.
- Introduced `set_session_defaults` and `with_session_defaults` methods for flexible session configuration.
- Updated tests to validate the new session management features, ensuring correct behavior during session operations.

Author: jordcodes

src/store.rs

@@ -11,12 +11,18 @@ use std::time::Duration;
 #[derive(Clone)]
 pub struct Store {
     pool: PgPool,
+    session_defaults: AppendOptions,
+    session_advisory_locks: bool,
 }
 
 impl Store {
     pub async fn connect(url: &str) -> Result<Self> {
         let pool = PgPool::connect(url).await?;
-        Ok(Self { pool })
+        Ok(Self {
+            pool,
+            session_defaults: AppendOptions::default(),
+            session_advisory_locks: false,
+        })
     }
 
     pub fn builder(url: impl Into<String>) -> StoreBuilder {
@@ -30,17 +36,31 @@ impl Store {
     }
 
     pub fn document_session(&self) -> DocumentSession {
-        self.docs().session()
+        self.session_builder().build()
     }
 
     pub fn session_builder(&self) -> SessionBuilder {
         SessionBuilder {
             store: self.clone(),
-            defaults: AppendOptions::default(),
-            use_advisory_lock: false,
+            defaults: self.session_defaults.clone(),
+            use_advisory_lock: self.session_advisory_locks,
         }
     }
 
+    pub fn session_defaults(&self) -> (&AppendOptions, bool) {
+        (&self.session_defaults, self.session_advisory_locks)
+    }
+
+    pub fn set_session_defaults(&mut self, defaults: AppendOptions, advisory_locks: bool) {
+        self.session_defaults = defaults;
+        self.session_advisory_locks = advisory_locks;
+    }
+
+    pub fn with_session_defaults(mut self, defaults: AppendOptions, advisory_locks: bool) -> Self {
+        self.set_session_defaults(defaults, advisory_locks);
+        self
+    }
+
     pub fn events(&self) -> Events {
         Events {
             pool: self.pool.clone(),
@@ -132,7 +152,11 @@ impl StoreBuilder {
             opts = opts.acquire_timeout(t);
         }
         let pool = opts.connect(&self.url).await?;
-        Ok(Store { pool })
+        Ok(Store {
+            pool,
+            session_defaults: AppendOptions::default(),
+            session_advisory_locks: false,
+        })
     }
 }
 
@@ -194,14 +218,18 @@ impl SessionBuilder {
 
         let mut session = DocumentSession::new(store.pool.clone(), events);
         if let Some(headers) = defaults.headers.clone() {
-            let headers_clone = headers.clone();
-            session.merge_event_headers(headers_clone);
-            if let Some(key) = headers.get("idempotency_key").and_then(|v| v.as_str()) {
-                session.set_event_idempotency_key(key);
-            }
+            session.merge_event_headers(headers);
         }
         session.set_event_causation_id(defaults.causation_id);
         session.set_event_correlation_id(defaults.correlation_id);
+        if let Some(key) = defaults
+            .headers
+            .as_ref()
+            .and_then(|h| h.get("idempotency_key"))
+            .and_then(|v| v.as_str())
+        {
+            session.set_event_idempotency_key(key);
+        }
         session
     }
 }

tests/documents_repo.rs

@@ -1,4 +1,5 @@
 use anyhow::Result;
+use rillflow::events::AppendOptions;
 use rillflow::{Error, Event, Expected, Store};
 use serde::{Deserialize, Serialize};
 use serde_json::json;
@@ -109,7 +110,7 @@ async fn session_load_store_delete_roundtrip() -> Result<()> {
     let port = container.get_host_port_ipv4(5432).await?;
     let url = format!("postgres://postgres:postgres@{host}:{port}/postgres?sslmode=disable");
 
-    let store = Store::connect(&url).await?;
+    let mut store = Store::connect(&url).await?;
     rillflow::testing::migrate_core_schema(store.pool()).await?;
 
     #[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
@@ -119,11 +120,15 @@ async fn session_load_store_delete_roundtrip() -> Result<()> {
     }
 
     let id = Uuid::new_v4();
-    let mut session = store
-        .session_builder()
-        .merge_headers(json!({"source": "test"}))
-        .advisory_locks(true)
-        .build();
+    store.set_session_defaults(
+        AppendOptions {
+            headers: Some(json!({"source": "test"})),
+            causation_id: None,
+            correlation_id: None,
+        },
+        true,
+    );
+    let mut session = store.document_session();
 
     // create new doc via session store + save
     session.store(