add endpoint to retrieve stream event history

Author: fa-sharp

.github/workflows/rust-client.yml

@@ -34,15 +34,17 @@ jobs:
           echo "version=$VERSION" >> $GITHUB_OUTPUT
 
       - name: Generate Rust client
+        id: generate
         run: |
-          cargo progenitor \
+          output=$(cargo progenitor \
             -i spec/openapi.json \
             -o clients/rust/ \
             -n tinistream-client \
             --interface builder \
             --tags separate \
             --license-name MIT \
-            --version ${{ steps.spec-version.outputs.version }} || true
+            --version ${{ steps.spec-version.outputs.version }}) || true
+          echo "output=$output" >> $GITHUB_OUTPUT
 
       - name: Check for changes
         id: changes
@@ -72,6 +74,11 @@ jobs:
             - Updated Rust client to match OpenAPI spec version `${{ steps.spec-version.outputs.version }}`
             - Generated using `cargo-progenitor`
 
+            **Output of `cargo-progenitor`**
+            ```
+            ${{ steps.generate.outputs.output }}
+            ```
+
             **Generated on:** ${{ steps.date.outputs.date }}
 
             Please review the changes and merge if they look correct.

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "tinistream"
-version = "0.1.2"
+version = "0.1.3"
 edition = "2021"
 publish = false
 
@@ -22,7 +22,7 @@ itertools = "0.14.0"
 rocket = { version = "0.5.1", features = ["json", "secrets"] }
 rocket_okapi = { version = "0.9.0", features = ["rapidoc", "rocket_ws"] }
 rocket_ws = { version = "0.1.1" }
-schemars = "0.8.22"
+schemars = { version = "0.8.22" }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 thiserror = "2.0"

api/Cargo.toml

@@ -2,14 +2,14 @@ use rocket::{futures::StreamExt, get, post, serde::json::Json, Route, State};
 use rocket_okapi::{okapi::openapi3::OpenApi, openapi, openapi_get_routes_spec};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
-use time::ext::NumericalDuration;
+use time::{ext::NumericalDuration, format_description::well_known, UtcDateTime};
 
 use crate::{
     auth::{create_client_token, ApiKeyAuth, Crypto},
     config::AppConfig,
     data::JsonStream,
     errors::ApiError,
-    redis::{stream_sse_url, RedisClient, StreamStatus, DATA_KEY, EVENT_KEY},
+    redis::*,
 };
 
 pub fn get_routes() -> (Vec<Route>, OpenApi) {
@@ -18,6 +18,7 @@ pub fn get_routes() -> (Vec<Route>, OpenApi) {
         create_stream,
         create_token,
         get_stream_info,
+        get_stream_events,
         add_events,
         add_events_json_stream,
         cancel_stream,
@@ -64,6 +65,35 @@ async fn get_stream_info(
     }))
 }
 
+/// # Get stream events
+/// Get all events so far in a stream
+#[openapi(tag = "Stream")]
+#[get("/events?<key>")]
+async fn get_stream_events(
+    _api_key: ApiKeyAuth,
+    key: &str,
+    reader: RedisReader,
+) -> Result<Json<Vec<StreamEvent>>, ApiError> {
+    let (prev_events, _, _) = reader.get_prev_events(key, None).await?;
+    let events = prev_events
+        .into_iter()
+        .filter_map(|(id, mut data)| {
+            let unix_millis: i64 = id.split('-').next().unwrap_or_default().parse().ok()?;
+            let date_time = UtcDateTime::from_unix_timestamp(unix_millis / 1000).ok()?;
+            let iso_time = date_time.format(&well_known::Rfc3339).ok()?;
+            let event = StreamEvent {
+                id: (*id).to_owned(),
+                time: iso_time,
+                event: data.remove(EVENT_KEY).as_deref().map(|e| e.to_owned())?,
+                data: data.remove(DATA_KEY).as_deref().map(|d| d.to_owned()),
+            };
+            Some(event)
+        })
+        .collect();
+
+    Ok(Json(events))
+}
+
 /// # Create stream
 /// Create a new stream, and get a client URL and token to connect to the stream
 #[openapi(tag = "Stream")]
@@ -217,6 +247,19 @@ pub struct StreamInfo {
     ttl: i64,
 }
 
+#[derive(JsonSchema, Serialize)]
+pub struct StreamEvent {
+    /// ID of the event
+    id: String,
+    /// Time of the event (ISO 8601 format)
+    time: String,
+    /// Name/type of the event
+    event: String,
+    /// Event data
+    #[serde(skip_serializing_if = "Option::is_none")]
+    data: Option<String>,
+}
+
 #[derive(JsonSchema, Deserialize)]
 struct StreamRequest {
     key: String,

api/src/api/stream.rs

@@ -81,7 +81,7 @@ impl RedisReader {
 
     /// Returns a tuple containing the previous events in the stream, the last event ID,
     /// and a boolean indicating if the stream has already ended.
-    async fn get_prev_events(
+    pub async fn get_prev_events(
         &self,
         key: &str,
         start_event_id: Option<&str>,

api/src/redis/reader.rs

@@ -2,7 +2,7 @@
   "openapi": "3.0.0",
   "info": {
     "title": "tinistream",
-    "version": "0.1.2"
+    "version": "0.1.3"
   },
   "paths": {
     "/api/health": {
@@ -494,6 +494,96 @@
         ]
       }
     },
+    "/api/stream/events": {
+      "get": {
+        "tags": [
+          "Stream"
+        ],
+        "summary": "Get stream events",
+        "description": "Get all events so far in a stream",
+        "operationId": "get_stream_events",
+        "parameters": [
+          {
+            "name": "key",
+            "in": "query",
+            "required": true,
+            "schema": {
+              "type": "string"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "type": "array",
+                  "items": {
+                    "$ref": "#/components/schemas/StreamEvent"
+                  }
+                }
+              }
+            }
+          },
+          "400": {
+            "description": "Bad request",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/ErrorMessage"
+                }
+              }
+            }
+          },
+          "401": {
+            "description": "Unauthorized",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/ErrorMessage"
+                }
+              }
+            }
+          },
+          "404": {
+            "description": "Not found",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/ErrorMessage"
+                }
+              }
+            }
+          },
+          "422": {
+            "description": "Incorrectly formatted",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/ErrorMessage"
+                }
+              }
+            }
+          },
+          "500": {
+            "description": "Internal server error",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/ErrorMessage"
+                }
+              }
+            }
+          }
+        },
+        "security": [
+          {
+            "ApiKey": []
+          }
+        ]
+      }
+    },
     "/api/stream/add": {
       "post": {
         "tags": [
@@ -985,6 +1075,33 @@
           }
         }
       },
+      "StreamEvent": {
+        "type": "object",
+        "required": [
+          "event",
+          "id",
+          "time"
+        ],
+        "properties": {
+          "id": {
+            "description": "ID of the event",
+            "type": "string"
+          },
+          "time": {
+            "description": "Time of the event (ISO 8601 format)",
+            "type": "string"
+          },
+          "event": {
+            "description": "Name/type of the event",
+            "type": "string"
+          },
+          "data": {
+            "description": "Event data",
+            "type": "string",
+            "nullable": true
+          }
+        }
+      },
       "AddEventsResponse": {
         "type": "object",
         "required": [