Merge pull request #401 from lmnr-ai/dev

Backoff, cache, optimizations, options for browser requests

Author: dinmukhamedm

.github/workflows/build-app-server-binary.yml

@@ -0,0 +1,88 @@
+name: Build & publish app-server binary
+on:
+  release:
+    types:
+      - published
+
+jobs:
+  build-and-upload:
+    name: Build and Upload
+    # This is used for HF Spaces, where we use
+    # debian:bookworm, that has GLIBC 2.36, but
+    # ubuntu-latest has GLIBC 2.38,
+    runs-on: ubuntu-22.04
+    permissions:
+      contents: write
+      packages: write
+      attestations: write
+      id-token: write
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Update Rust toolchain
+        run: rustup update
+
+      - name: Install additional dependencies
+        run: sudo apt-get update && sudo apt-get install -y protobuf-compiler libssl-dev
+
+      - name: Build Release Binary
+        run: cargo build --release
+        working-directory: ./app-server
+
+      - name: verify binary path
+        run: |
+          if [[ -f app-server/target/release/app-server ]]; then
+            echo "Binary found at app-server/target/release/app-server"
+          else
+            echo "Binary not found at app-server/target/release/app-server"
+            exit 1
+          fi
+
+      - name: Get latest release
+        id: get_latest_release
+        uses: actions/github-script@v7
+        with:
+          script: |
+            const response = await github.rest.repos.listReleases({
+              owner: context.repo.owner,
+              repo: context.repo.repo
+            });
+            if (response.data.length === 0) {
+              core.setFailed('No releases found');
+              return;
+            }
+            const latest_release = response.data.sort((a, b) => new Date(b.created_at) - new Date(a.created_at))[0];
+            core.setOutput('upload_url', latest_release.upload_url);
+            core.setOutput('release_id', latest_release.id);
+
+      - name: Delete existing assets
+        uses: actions/github-script@v7
+        with:
+          script: |
+            const release_id = '${{ steps.get_latest_release.outputs.release_id }}';
+            const response = await github.rest.repos.listReleaseAssets({
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              release_id: release_id
+            });
+
+            for (const asset of response.data) {
+              if (asset.name === 'app-server') {
+                await github.rest.repos.deleteReleaseAsset({
+                  owner: context.repo.owner,
+                  repo: context.repo.repo,
+                  asset_id: asset.id
+                });
+              }
+            }
+
+      - name: Upload to latest release
+        uses: actions/upload-release-asset@v1
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        with:
+          upload_url: ${{ steps.get_latest_release.outputs.upload_url }}
+          asset_path: ./app-server/target/release/app-server
+          asset_name: app-server
+          asset_content_type: application/octet-stream

app-server/.env.example

@@ -15,3 +15,6 @@ CODE_EXECUTOR_URL=http://localhost:8811
 # must be exactly 32 bytes (64 hex characters)
 AEAD_SECRET_KEY=0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef
 ENVIRONMENT=FULL
+
+# Optional, if you want to use Redis for caching. Create a Redis/Valkey instance and set the URL here.
+# REDIS_URL=redis://localhost:6379

app-server/Cargo.lock

@@ -24,6 +24,7 @@ aws-config = "1.5.10"
 aws-credential-types = "1.2.1"
 aws-sdk-bedrockruntime = "1.61.0"
 aws-sdk-s3 = "1.63.0"
+backoff = {version = "0.4.0", features = ["tokio"]}
 base64 = "0.22.1"
 bimap = "0.6.3"
 bytes = "1.7.1"
@@ -49,6 +50,7 @@ num_cpus = "1.16.0"
 prost = "0.13"
 rand = "0.8.5"
 rayon = "1.10"
+redis = {version = "0.28.2", features = ["tokio-comp"]}
 regex = "1.10.3"
 reqwest = {version = "0.12", default-features = false, features = ["rustls-tls", "json", "stream", "multipart"]}
 reqwest-eventsource = "0.6.0"

app-server/Cargo.toml

@@ -3,13 +3,12 @@ use std::sync::Arc;
 use sqlx::PgPool;
 use uuid::Uuid;
 
-use crate::db::pipelines::PipelineVersion;
 use crate::pipeline::utils::get_target_pipeline_version_cache_key;
 use crate::routes::api_keys::hash_api_key;
 use crate::routes::error;
 use crate::{
-    cache::Cache,
-    db::{self, DB},
+    cache::{keys::PROJECT_API_KEY_CACHE_KEY, Cache, CacheTrait},
+    db::{self, pipelines::PipelineVersion, project_api_keys::ProjectApiKey, DB},
 };
 
 pub async fn query_target_pipeline_version(
@@ -32,7 +31,7 @@ pub async fn query_target_pipeline_version(
                 .await?;
             if let Some(pipeline_version) = &pipeline_version {
                 let _ = cache
-                    .insert::<PipelineVersion>(cache_key, pipeline_version)
+                    .insert::<PipelineVersion>(&cache_key, pipeline_version.clone())
                     .await;
             }
             Ok(pipeline_version)
@@ -44,17 +43,18 @@ pub async fn get_api_key_from_raw_value(
     pool: &PgPool,
     cache: Arc<Cache>,
     raw_api_key: String,
-) -> anyhow::Result<db::project_api_keys::ProjectApiKey> {
+) -> anyhow::Result<ProjectApiKey> {
     let api_key_hash = hash_api_key(&raw_api_key);
+    let cache_key = format!("{PROJECT_API_KEY_CACHE_KEY}:{api_key_hash}");
     let cache_res = cache
-        .get::<db::project_api_keys::ProjectApiKey>(&api_key_hash)
+        .get::<db::project_api_keys::ProjectApiKey>(&cache_key)
         .await;
     match cache_res {
         Ok(Some(api_key)) => Ok(api_key),
         Ok(None) | Err(_) => {
             let api_key = db::project_api_keys::get_api_key(pool, &api_key_hash).await?;
             let _ = cache
-                .insert::<db::project_api_keys::ProjectApiKey>(api_key_hash, &api_key)
+                .insert::<db::project_api_keys::ProjectApiKey>(&cache_key, api_key.clone())
                 .await;
 
             Ok(api_key)

app-server/src/api/utils.rs

@@ -1,86 +1,75 @@
+use std::sync::Arc;
+
 use actix_web::{options, post, web, HttpResponse};
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
-use crate::{db::project_api_keys::ProjectApiKey, routes::types::ResponseResult};
+use crate::{db::project_api_keys::ProjectApiKey, mq, routes::types::ResponseResult};
+
+pub const BROWSER_SESSIONS_QUEUE: &str = "browser_sessions_queue";
+pub const BROWSER_SESSIONS_EXCHANGE: &str = "browser_sessions_exchange";
+pub const BROWSER_SESSIONS_ROUTING_KEY: &str = "browser_sessions_routing_key";
 
-#[derive(Debug, Serialize, Deserialize)]
-struct RRWebEvent {
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct RRWebEvent {
     #[serde(rename = "type")]
-    event_type: i32,
-    timestamp: i64,
-    data: serde_json::Value,
+    pub event_type: i32,
+    pub timestamp: i64,
+    pub data: serde_json::Value,
 }
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, Clone)]
 #[serde(rename_all = "camelCase")]
-struct EventBatch {
-    events: Vec<RRWebEvent>,
-    session_id: Uuid,
-    trace_id: Uuid,
+pub struct EventBatch {
+    pub events: Vec<RRWebEvent>,
+    pub session_id: Uuid,
+    pub trace_id: Uuid,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct QueueBrowserEventMessage {
+    pub batch: EventBatch,
+    pub project_id: Uuid,
 }
 
 #[options("events")]
 async fn options_handler() -> ResponseResult {
-    // TODO: use cors middleware from actix_cors crate
     Ok(HttpResponse::Ok()
         .insert_header(("Access-Control-Allow-Origin", "*"))
         .insert_header(("Access-Control-Allow-Methods", "POST, OPTIONS"))
         .insert_header((
             "Access-Control-Allow-Headers",
-            "Authorization, Content-Type",
+            "Authorization, Content-Type, Content-Encoding, Accept",
         ))
         .insert_header(("Access-Control-Max-Age", "86400"))
         .finish())
 }
 
 #[post("events")]
 async fn create_session_event(
-    clickhouse: web::Data<clickhouse::Client>,
     batch: web::Json<EventBatch>,
     project_api_key: ProjectApiKey,
+    queue: web::Data<Arc<dyn mq::MessageQueue<QueueBrowserEventMessage>>>,
 ) -> ResponseResult {
-    // Skip if there are no events
-    if batch.events.is_empty() {
-        return Ok(HttpResponse::Ok().finish());
-    }
+    let filtered_batch = batch.into_inner();
 
-    // Prepare batch data
-    let mut query = String::from(
-        "
-        INSERT INTO browser_session_events (
-            event_id, session_id, trace_id, timestamp,
-            event_type, data, project_id
-        )
-        VALUES ",
-    );
-
-    let mut values = Vec::new();
-
-    for (i, event) in batch.events.iter().enumerate() {
-        if i > 0 {
-            query.push_str(", ");
-        }
-        query.push_str("(?, ?, ?, ?, ?, ?, ?)");
-
-        // Add each value individually
-        values.extend_from_slice(&[
-            Uuid::new_v4().to_string(),
-            batch.session_id.to_string(),
-            batch.trace_id.to_string(),
-            event.timestamp.to_string(),
-            event.event_type.to_string(),
-            event.data.to_string(),
-            project_api_key.project_id.to_string(),
-        ]);
+    // Return 400 Bad Request if trace_id is null (00000000-0000-0000-0000-000000000000)
+    if filtered_batch.trace_id == Uuid::nil() {
+        return Ok(HttpResponse::BadRequest().json(serde_json::json!({
+            "error": "Invalid trace_id: must not be null (00000000-0000-0000-0000-000000000000)"
+        })));
     }
 
-    // Execute batch insert with individual bindings
-    let mut query_with_bindings = clickhouse.query(&query);
-    for value in values {
-        query_with_bindings = query_with_bindings.bind(value);
-    }
-    query_with_bindings.execute().await?;
+    queue
+        .publish(
+            &QueueBrowserEventMessage {
+                batch: filtered_batch,
+                project_id: project_api_key.project_id,
+            },
+            BROWSER_SESSIONS_EXCHANGE,
+            BROWSER_SESSIONS_ROUTING_KEY,
+        )
+        .await?;
 
     Ok(HttpResponse::Ok().finish())
 }