feat(dgw): run AI analysis of a recording as a provisioner task

crates/job-queue-libsql/Cargo.toml

@@ -19,3 +19,6 @@ tracing = "0.1"
 typed-builder = "0.21"
 ulid = { version = "1.2", features = ["uuid"] }
 uuid = "1.23"
+
+[dev-dependencies]
+tokio = { version = "1.52", features = ["macros", "rt"] }

crates/job-queue-libsql/src/lib.rs

@@ -402,6 +402,27 @@ impl JobQueue for LibSqlJobQueue {
 
         Ok(Some(scheduled_for))
     }
+
+    async fn job_definitions(&self, name: &str) -> anyhow::Result<Vec<String>> {
+        let sql_query = "SELECT json(def) FROM job_queue WHERE name = :name AND failed_attempts < :max_attempts";
+
+        let params = ((":name", name), (":max_attempts", self.max_attempts));
+
+        trace!(%sql_query, ?params, "Listing job definitions");
+
+        let mut rows = self
+            .conn
+            .query(sql_query, params)
+            .await
+            .context("failed to execute SQL query")?;
+
+        let mut definitions = Vec::new();
+        while let Some(row) = rows.next().await.context("failed to read the row")? {
+            definitions.push(row.get::<String>(0).context("failed to read the job definition")?);
+        }
+
+        Ok(definitions)
+    }
 }
 
 // Typically, migrations should not be modified once released, and we should only be appending to this list.
@@ -425,3 +446,55 @@ const MIGRATIONS: &[&str] = &[
 
     CREATE INDEX idx_scheduled_for ON job_queue(scheduled_for);",
 ];
+
+#[cfg(test)]
+mod tests {
+    use job_queue::{DynJob, Job, JobQueue as _, RunnerWaker};
+
+    use super::*;
+
+    struct Named {
+        name: &'static str,
+        def: &'static str,
+    }
+
+    #[async_trait]
+    impl Job for Named {
+        fn name(&self) -> &str {
+            self.name
+        }
+
+        fn write_json(&self) -> anyhow::Result<String> {
+            Ok(self.def.to_owned())
+        }
+
+        async fn run(&mut self) -> anyhow::Result<()> {
+            Ok(())
+        }
+    }
+
+    #[tokio::test]
+    async fn job_definitions_lists_the_stored_jobs_of_one_name() {
+        let conn = libsql::Builder::new_local(":memory:")
+            .build()
+            .await
+            .expect("build database")
+            .connect()
+            .expect("connect");
+        let queue = LibSqlJobQueue::builder()
+            .runner_waker(RunnerWaker::new(|| {}))
+            .conn(conn)
+            .build();
+        queue.setup().await.expect("setup");
+
+        for (name, def) in [("a", r#"{"n":1}"#), ("b", r#"{"n":2}"#), ("a", r#"{"n":3}"#)] {
+            let job: DynJob = Box::new(Named { name, def });
+            queue.push_job(&job, None).await.expect("push");
+        }
+
+        let mut definitions = queue.job_definitions("a").await.expect("list");
+        definitions.sort();
+
+        assert_eq!(definitions, [r#"{"n":1}"#, r#"{"n":3}"#]);
+    }
+}

crates/job-queue/src/lib.rs

@@ -62,6 +62,9 @@ pub trait JobQueue: Send + Sync {
 
     /// Retrieves the closest future scheduled date
     async fn next_scheduled_date(&self) -> anyhow::Result<Option<OffsetDateTime>>;
+
+    /// Lists the definitions of the stored jobs named `name` that may still run, in no particular order.
+    async fn job_definitions(&self, name: &str) -> anyhow::Result<Vec<String>>;
 }
 
 pub struct JobCtx {

devolutions-gateway/Cargo.toml

@@ -24,6 +24,7 @@ transport.path = "../crates/transport"
 jmux-proxy.path = "../crates/jmux-proxy"
 devolutions-agent-shared.path = "../crates/devolutions-agent-shared"
 devolutions-gateway-task.path = "../crates/devolutions-gateway-task"
+devolutions-gateway-ai.path = "../crates/devolutions-gateway-ai"
 devolutions-log.path = "../crates/devolutions-log"
 job-queue.path = "../crates/job-queue"
 job-queue-libsql.path = "../crates/job-queue-libsql"
@@ -38,6 +39,8 @@ sysevent-codes.path = "../crates/sysevent-codes"
 agent-tunnel = { path = "../crates/agent-tunnel" }
 agent-tunnel-libsql.path = "../crates/agent-tunnel-libsql"
 gateway-db.path = "../crates/gateway-db"
+provisioner-task.path = "../crates/provisioner-task"
+provisioner-task-libsql.path = "../crates/provisioner-task-libsql"
 ironrdp-pdu = { version = "0.9", features = ["std"] }
 ironrdp-core = { version = "0.2", features = ["std"] }
 ironrdp-rdcleanpath = "0.2"
@@ -91,6 +94,7 @@ tracing = "0.1"
 
 # Async, futures…
 tokio = { version = "1.52", features = ["signal", "net", "io-util", "time", "rt", "rt-multi-thread", "sync", "macros", "parking_lot", "fs"] }
+tokio-util = "0.7"
 tokio-rustls = { version = "0.26", default-features = false, features = ["logging", "tls12", "ring"] }
 reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-native-roots", "json", "stream", "http2", "socks"] }
 http-client-proxy = { path = "../crates/http-client-proxy" }

devolutions-gateway/src/api/jrec.intent.md

@@ -0,0 +1,10 @@
+# Background
+
+Jrec is the endpoint to push, streaming, read recordings and its associated files.
+
+
+# AI analysis
+
+An AI analysis is a task initiated by the provisioner.
+It requires the scope token with access `gateway.tasks.recording-ai-analysis.start` to access the url `POST /jet/jrec/{session_id}/ai-analysis`; Naturally in order to read the task's state, it requires the scope token with access `gateway.tasks.recording-ai-analysis.read`.
+The analyzed output belongs to the recording reading scope, however, the associated metadata, including but not limited to error messages, token usages etc. belong to the task reading scope.