feat(export): date range for the JSON exports

aw-datastore/examples/export_memory.rs

@@ -95,7 +95,8 @@ fn main() {
             );
         }
         "stream" => {
-            ds.write_export(&conn, None, std::io::sink()).unwrap();
+            ds.write_export(&conn, None, None, None, std::io::sink())
+                .unwrap();
             println!("{mode}: {count} synthetic events, {:?}", start.elapsed());
         }
         "materialized" => {

aw-datastore/src/datastore.rs

@@ -356,7 +356,7 @@ pub struct DatastoreInstance {
 /// express in nanoseconds (before 1677 or after 2262), so such a bound means "no limit"
 /// instead of panicking the datastore worker. Bounds that can't match anything are
 /// caught first by [`filter_excludes_everything`].
-fn filter_nanos(dt: DateTime<Utc>) -> i64 {
+pub(crate) fn filter_nanos(dt: DateTime<Utc>) -> i64 {
     dt.timestamp_nanos_opt().unwrap_or(if dt.timestamp() < 0 {
         i64::MIN
     } else {
@@ -854,17 +854,22 @@ impl DatastoreInstance {
 
     /// Serialize an export one event at a time using the caller's connection.
     /// Returns the bucket ID for a single-bucket export (for download naming).
+    /// Events are limited to those overlapping `[start, end]`, cut to it.
     pub fn write_export(
         &self,
         conn: &Connection,
         bucket_id: Option<&str>,
+        start: Option<DateTime<Utc>>,
+        end: Option<DateTime<Utc>>,
         writer: impl std::io::Write,
     ) -> Result<Option<String>, DatastoreError> {
         crate::export::write_export(
             conn,
             self.db_version,
             &self.buckets_cache,
             bucket_id,
+            start,
+            end,
             writer,
         )?;
         Ok(bucket_id.map(str::to_owned).or_else(|| {

aw-datastore/src/export.rs

@@ -9,37 +9,42 @@ use serde::{
     Serialize, Serializer,
 };
 
-use crate::datastore::{events_source, parse_event_row, prefer_endtime_index};
+use crate::datastore::{events_source, filter_nanos, parse_event_row, prefer_endtime_index};
 use crate::DatastoreError;
 
 struct EventRows<'a> {
     conn: &'a Connection,
     db_version: i32,
     bucket: &'a Bucket,
+    range: (i64, i64),
 }
 
 impl Serialize for EventRows<'_> {
     fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        let (start, end) = self.range;
+        let mut seq = serializer.serialize_seq(None)?;
+        if start > end {
+            // An empty range, like get_events: no events rather than an error.
+            return seq.end();
+        }
         let sql = format!(
             "SELECT id, starttime, endtime, data
              FROM {}
              WHERE bucketrow = ?1 AND endtime >= ?2 AND starttime <= ?3
              ORDER BY starttime DESC, endtime ASC, id ASC",
-            events_source(self.db_version, false)
+            events_source(
+                self.db_version,
+                prefer_endtime_index(self.bucket, start, end, None)
+            )
         );
         let mut stmt = self.conn.prepare_cached(&sql).map_err(S::Error::custom)?;
         let mut rows = stmt
-            .query(rusqlite::params![
-                self.bucket.bid.unwrap(),
-                i64::MIN,
-                i64::MAX
-            ])
+            .query(rusqlite::params![self.bucket.bid.unwrap(), start, end])
             .map_err(S::Error::custom)?;
-        let mut seq = serializer.serialize_seq(None)?;
         while let Some(row) = rows.next().map_err(S::Error::custom)? {
-            // Match get_events' default range, clipping and corrupt-row policy.
-            // i64::MIN so None starttime includes pre-epoch events, like get_events.
-            match crate::datastore::parse_event_row(row, Some((i64::MIN, i64::MAX))) {
+            // Match get_events' range, clipping and corrupt-row policy: events
+            // overlapping the range are cut to it.
+            match crate::datastore::parse_event_row(row, Some((start, end))) {
                 Ok(event) => seq.serialize_element(&event)?,
                 Err(err) => warn!("Corrupt event in bucket {}: {}", self.bucket.id, err),
             }
@@ -52,6 +57,7 @@ struct ExportBucket<'a> {
     conn: &'a Connection,
     db_version: i32,
     bucket: &'a Bucket,
+    range: (i64, i64),
 }
 
 impl Serialize for ExportBucket<'_> {
@@ -74,6 +80,7 @@ impl Serialize for ExportBucket<'_> {
                 conn: self.conn,
                 db_version: self.db_version,
                 bucket: self.bucket,
+                range: self.range,
             },
         )?;
         map.end()
@@ -85,6 +92,7 @@ struct ExportBuckets<'a> {
     db_version: i32,
     buckets: &'a HashMap<String, Bucket>,
     selected: Option<&'a str>,
+    range: (i64, i64),
 }
 
 impl Serialize for ExportBuckets<'_> {
@@ -98,6 +106,7 @@ impl Serialize for ExportBuckets<'_> {
                         conn: self.conn,
                         db_version: self.db_version,
                         bucket,
+                        range: self.range,
                     },
                 )?;
             }
@@ -106,13 +115,22 @@ impl Serialize for ExportBuckets<'_> {
     }
 }
 
+/// Serialize buckets with their events overlapping `[start, end]`, cut to it;
+/// `None` leaves that side open, so no range exports everything.
 pub(crate) fn write_export(
     conn: &Connection,
     db_version: i32,
     buckets: &HashMap<String, Bucket>,
     selected: Option<&str>,
+    start: Option<DateTime<Utc>>,
+    end: Option<DateTime<Utc>>,
     writer: impl Write,
 ) -> Result<(), DatastoreError> {
+    // i64::MIN so an open start includes pre-epoch events, like get_events.
+    let range = (
+        start.map_or(i64::MIN, filter_nanos),
+        end.map_or(i64::MAX, filter_nanos),
+    );
     if let Some(id) = selected {
         if !buckets.contains_key(id) {
             return Err(DatastoreError::NoSuchBucket(id.to_owned()));
@@ -130,6 +148,7 @@ pub(crate) fn write_export(
                 db_version,
                 buckets,
                 selected,
+                range,
             },
         },
     )
@@ -488,20 +507,88 @@ mod tests {
             }
             let expected = serde_json::to_value(BucketsExport { buckets }).unwrap();
             let mut output = Vec::new();
-            ds.write_export(&conn, selected, &mut output).unwrap();
+            ds.write_export(&conn, selected, None, None, &mut output)
+                .unwrap();
             assert_eq!(
                 serde_json::from_slice::<serde_json::Value>(&output).unwrap(),
                 expected
             );
         }
     }
 
+    #[test]
+    fn ranged_json_export_matches_get_events_for_the_same_range() {
+        let (conn, mut ds) = setup();
+        let start = DateTime::from_timestamp(10, 0).unwrap();
+        let end = DateTime::from_timestamp(25, 0).unwrap();
+        for selected in [None, Some("populated")] {
+            let mut buckets = ds.get_buckets();
+            buckets.retain(|id, _| selected.is_none() || selected == Some(id.as_str()));
+            for (id, bucket) in &mut buckets {
+                bucket.events = Some(TryVec::new(
+                    ds.get_events(&conn, id, Some(start), Some(end), None)
+                        .unwrap(),
+                ));
+            }
+            let expected = serde_json::to_value(BucketsExport { buckets }).unwrap();
+            let mut output = Vec::new();
+            ds.write_export(&conn, selected, Some(start), Some(end), &mut output)
+                .unwrap();
+            let exported = serde_json::from_slice::<serde_json::Value>(&output).unwrap();
+            assert_eq!(exported, expected);
+        }
+        // Cut to the range: the event from -10s to 10s keeps only its last instant,
+        // the one at 30s is left out.
+        let mut output = Vec::new();
+        ds.write_export(
+            &conn,
+            Some("populated"),
+            Some(start),
+            Some(end),
+            &mut output,
+        )
+        .unwrap();
+        let exported = serde_json::from_slice::<serde_json::Value>(&output).unwrap();
+        let events = exported["buckets"]["populated"]["events"]
+            .as_array()
+            .unwrap();
+        assert_eq!(events.len(), 4);
+        for event in events {
+            let timestamp = DateTime::parse_from_rfc3339(event["timestamp"].as_str().unwrap())
+                .unwrap()
+                .with_timezone(&Utc);
+            let end_of_event = timestamp
+                + Duration::nanoseconds((event["duration"].as_f64().unwrap() * 1e9) as i64);
+            assert!(
+                timestamp >= start && end_of_event <= end,
+                "{event} is outside the range"
+            );
+        }
+    }
+
+    #[test]
+    fn json_export_with_an_empty_range_has_no_events() {
+        let (conn, ds) = setup();
+        let (start, end) = (
+            DateTime::from_timestamp(25, 0).unwrap(),
+            DateTime::from_timestamp(10, 0).unwrap(),
+        );
+        let mut output = Vec::new();
+        ds.write_export(&conn, None, Some(start), Some(end), &mut output)
+            .unwrap();
+        let exported = serde_json::from_slice::<serde_json::Value>(&output).unwrap();
+        for id in ["populated", "empty"] {
+            assert_eq!(exported["buckets"][id]["events"], serde_json::json!([]));
+            assert_eq!(exported["buckets"][id]["id"], id); // the bucket itself is still there
+        }
+    }
+
     #[test]
     fn missing_bucket_and_writer_failures_propagate() {
         let (conn, ds) = setup();
         let mut output = Vec::new();
         assert!(matches!(
-            ds.write_export(&conn, Some("missing"), &mut output),
+            ds.write_export(&conn, Some("missing"), None, None, &mut output),
             Err(DatastoreError::NoSuchBucket(_))
         ));
         assert!(output.is_empty());
@@ -520,7 +607,7 @@ mod tests {
             }
         }
         assert!(matches!(
-            ds.write_export(&conn, None, FailingWriter(500)),
+            ds.write_export(&conn, None, None, None, FailingWriter(500)),
             Err(DatastoreError::InternalError(_))
         ));
     }

aw-datastore/src/worker.rs

@@ -222,7 +222,12 @@ pub enum Response {
 #[allow(clippy::large_enum_variant)]
 #[derive(Debug)]
 pub enum Command {
-    Export(Option<String>, File),
+    Export(
+        Option<String>,
+        Option<DateTime<Utc>>,
+        Option<DateTime<Utc>>,
+        File,
+    ),
     ExportCsv(
         String,
         Option<DateTime<Utc>>,
@@ -633,9 +638,9 @@ impl DatastoreWorker {
                 Ok(()) => Ok(Response::Empty()),
                 Err(e) => Err(e),
             },
-            Command::Export(bucket_id, mut file) => {
+            Command::Export(bucket_id, start, end, mut file) => {
                 let mut writer = BufWriter::new(&mut file);
-                let name = ds.write_export(tx, bucket_id.as_deref(), &mut writer)?;
+                let name = ds.write_export(tx, bucket_id.as_deref(), start, end, &mut writer)?;
                 writer.flush().map_err(|err| {
                     DatastoreError::InternalError(format!("Failed to flush export: {err}"))
                 })?;
@@ -1071,12 +1076,20 @@ impl Datastore {
     /// Write a consistent export to an empty temporary file. The file is
     /// returned only after serialization and flushing succeed. Event rows are
     /// serialized individually, including this worker's uncommitted writes.
+    /// `start`/`end` limit the events to those overlapping that range, cut to it.
     pub fn export_to_file(
         &self,
         bucket_id: Option<&str>,
+        start: Option<DateTime<Utc>>,
+        end: Option<DateTime<Utc>>,
         file: File,
     ) -> Result<(File, Option<String>), DatastoreError> {
-        match self.request(Command::Export(bucket_id.map(str::to_owned), file))? {
+        match self.request(Command::Export(
+            bucket_id.map(str::to_owned),
+            start,
+            end,
+            file,
+        ))? {
             Response::Export(file, name) => Ok((file, name)),
             _ => panic!("Invalid response"),
         }

aw-datastore/tests/datastore.rs

@@ -1658,7 +1658,7 @@ mod datastore_tests {
             assert_eq!(total_count, 4, "v{version} total count");
 
             // Export paths share the index selection.
-            ds.export_to_file(Some("testid"), tempfile::tempfile().unwrap())
+            ds.export_to_file(Some("testid"), None, None, tempfile::tempfile().unwrap())
                 .unwrap_or_else(|e| panic!("v{version} export: {e:?}"));
             ds.export_csv_to_file(
                 "testid",

aw-server/src/endpoints/bucket.rs

@@ -107,7 +107,7 @@ pub fn bucket_new(
 }
 
 /// Parse an optional RFC 3339 query parameter, answering 400 when it doesn't parse.
-fn parse_time_param(
+pub(crate) fn parse_time_param(
     name: &str,
     value: Option<String>,
 ) -> Result<Option<DateTime<Utc>>, HttpErrorJson> {
@@ -369,12 +369,21 @@ pub fn bucket_events_delete_many(
     }
 }
 
-#[get("/<bucket_id>/export")]
+/// Export a bucket as JSON, with all its events, or with `start`/`end` (RFC 3339)
+/// those overlapping that range, cut to it like `/events`.
+#[get("/<bucket_id>/export?<start>&<end>")]
 pub fn bucket_export(
     bucket_id: &str,
+    start: Option<String>,
+    end: Option<String>,
     state: &State<ServerState>,
 ) -> Result<BucketsExportRocket, HttpErrorJson> {
-    BucketsExportRocket::new(&state.datastore, Some(bucket_id))
+    BucketsExportRocket::new(
+        &state.datastore,
+        Some(bucket_id),
+        parse_time_param("start", start)?,
+        parse_time_param("end", end)?,
+    )
 }
 
 /// Stream events for a single bucket as a CSV file.

aw-server/src/endpoints/export.rs

@@ -1,9 +1,21 @@
 use rocket::State;
 
+use crate::endpoints::bucket::parse_time_param;
 use crate::endpoints::util::BucketsExportRocket;
 use crate::endpoints::{HttpErrorJson, ServerState};
 
-#[get("/")]
-pub fn buckets_export(state: &State<ServerState>) -> Result<BucketsExportRocket, HttpErrorJson> {
-    BucketsExportRocket::new(&state.datastore, None)
+/// Export every bucket as JSON, with all events, or with `start`/`end` (RFC 3339)
+/// those overlapping that range, cut to it like `/events`.
+#[get("/?<start>&<end>")]
+pub fn buckets_export(
+    start: Option<String>,
+    end: Option<String>,
+    state: &State<ServerState>,
+) -> Result<BucketsExportRocket, HttpErrorJson> {
+    BucketsExportRocket::new(
+        &state.datastore,
+        None,
+        parse_time_param("start", start)?,
+        parse_time_param("end", end)?,
+    )
 }