Add audit_stream() for db audit provider.

Author: tmpfs

Cargo.lock

@@ -109,10 +109,13 @@ tracing = "0.1"
 tracing-subscriber = { version = "0.3.16", features = ["env-filter", "json"] }
 secrecy = { version = "0.10", features = ["serde"] }
 serde = { version = "1", features = ["derive"] }
+
 tokio = { version = "1", features = ["rt", "macros", "time", "sync"]}
+tokio-util = { version = "0.7", default-features = false, features = ["io", "compat"] }
+tokio-stream = "0.1"
+
 serde_json = "1"
 serde_with = { version = "3", features = ["base64"] }
-tokio-util = { version = "0.7", default-features = false, features = ["io", "compat"] }
 async-trait = "0.1"
 async-recursion = "1"
 typeshare = "1"

Cargo.toml

@@ -85,7 +85,9 @@ args = [
   "--workspace",
   "--all-features",
   "--open",
-  "--no-deps"
+  "--no-deps",
+  "--exclude",
+  "sos-vfs",
 ]
 dependencies = ["clean-doc"]
 

Makefile.toml

@@ -48,6 +48,7 @@ sha2.workspace = true
 serde.workspace = true
 tokio.workspace = true
 url.workspace = true
+tokio-stream.workspace = true
 
 # sqlite
 async-sqlite = { workspace = true, optional = true }

crates/database/Cargo.toml

@@ -1,9 +1,13 @@
 //! Database audit log provider.
-use crate::{db::AuditEntity, Error};
+use crate::{
+    db::{AuditEntity, AuditRecord, AuditRow},
+    Error,
+};
 use async_sqlite::Client;
 use async_trait::async_trait;
 use futures::stream::BoxStream;
 use sos_audit::{AuditEvent, AuditStreamSink};
+use tokio_stream::wrappers::ReceiverStream;
 
 /// Audit provider that appends to a database table.
 pub struct AuditDatabaseProvider<E>
@@ -76,6 +80,36 @@ where
         BoxStream<'static, std::result::Result<AuditEvent, Self::Error>>,
         Self::Error,
     > {
-        todo!();
+        let (tx, rx) = tokio::sync::mpsc::channel::<
+            std::result::Result<AuditEvent, Self::Error>,
+        >(16);
+
+        self.client
+            .conn_and_then(move |conn| {
+                let mut stmt = if reverse {
+                    conn.prepare("SELECT * FROM audit_logs ORDER DESC")?
+                } else {
+                    conn.prepare("SELECT * FROM audit_logs ORDER ASC")?
+                };
+                let mut rows = stmt.query([])?;
+
+                while let Some(row) = rows.next()? {
+                    let row: AuditRow = row.try_into()?;
+                    let record: AuditRecord = row.try_into()?;
+                    let inner_tx = tx.clone();
+                    futures::executor::block_on(async move {
+                        if let Err(e) = inner_tx.send(Ok(record.event)).await
+                        {
+                            tracing::error!(error = %e);
+                        }
+                    });
+                }
+
+                Ok::<_, Error>(())
+            })
+            .await
+            .map_err(Error::from)?;
+
+        Ok(Box::pin(ReceiverStream::new(rx)))
     }
 }

crates/database/src/audit_provider.rs

@@ -13,7 +13,8 @@ mod server;
 mod system_message;
 
 pub use account::{AccountEntity, AccountRecord};
-pub use audit::AuditEntity;
+pub(crate) use audit::AuditRow;
+pub use audit::{AuditEntity, AuditRecord};
 pub use event::{CommitRecord, EventEntity, EventTable};
 #[cfg(feature = "files")]
 pub use file::FileEntity;