Implementing persistence using spaces

Cargo.toml

@@ -188,6 +188,7 @@ triomphe = "0.1.13"
 tui-term = { version = "=0.1.9", default-features = false }
 url = "2.2.2"
 urlencoding = "2.1.2"
+uuid = { version = "1.5.0", features = ["v4"] }
 webbrowser = "0.8.7"
 which = "4.4.0"
 unicode-segmentation = "1.10.1"

crates/turborepo-analytics/Cargo.toml

@@ -16,4 +16,4 @@ tokio = { workspace = true, features = ["full", "time"] }
 tracing = { workspace = true }
 turborepo-api-client = { workspace = true }
 turborepo-vercel-api = { workspace = true }
-uuid = { version = "1.5.0", features = ["v4"] }
+uuid = { workspace = true, features = ["v4"] }

crates/turborepo-api-client/src/spaces.rs

@@ -13,6 +13,15 @@ pub enum RunStatus {
     Completed,
 }
 
+impl AsRef<str> for RunStatus {
+    fn as_ref(&self) -> &str {
+        match self {
+            RunStatus::Running => "RUNNING",
+            RunStatus::Completed => "COMPLETED",
+        }
+    }
+}
+
 #[derive(Serialize)]
 pub struct SpaceClientSummary {
     pub id: &'static str,
@@ -144,13 +153,13 @@ impl APIClient {
         &self,
         space_id: &str,
         api_auth: &APIAuth,
-        payload: CreateSpaceRunPayload,
+        payload: &CreateSpaceRunPayload,
     ) -> Result<SpaceRun, Error> {
         let url = format!("/v0/spaces/{}/runs", space_id);
         let request_builder = self
             .create_request_builder(&url, api_auth, Method::POST)
             .await?
-            .json(&payload);
+            .json(payload);
 
         let response =
             retry::make_retryable_request(request_builder, retry::RetryStrategy::Timeout)
@@ -167,7 +176,7 @@ impl APIClient {
         space_id: &str,
         run_id: &str,
         api_auth: &APIAuth,
-        task: SpaceTaskSummary,
+        task: &SpaceTaskSummary,
     ) -> Result<(), Error> {
         let request_builder = self
             .create_request_builder(
@@ -176,7 +185,7 @@ impl APIClient {
                 Method::POST,
             )
             .await?
-            .json(&task);
+            .json(task);
 
         retry::make_retryable_request(request_builder, retry::RetryStrategy::Timeout)
             .await?

crates/turborepo-db/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "turborepo-db"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+camino = { workspace = true }
+serde_json = { workspace = true }
+sqlx = { version = "0.8.1", features = ["runtime-tokio", "sqlite"] }
+thiserror = { workspace = true }
+turbopath = { workspace = true }
+turborepo-api-client = { workspace = true }
+uuid = { workspace = true, features = ["v4"] }
+
+[lints]
+workspace = true

crates/turborepo-db/migrations/20240828183512_initial_tables.sql

@@ -0,0 +1,35 @@
+CREATE TABLE IF NOT EXISTS runs (
+    id TEXT PRIMARY KEY, -- primary key should be uuid
+    start_time INTEGER NOT NULL,
+    end_time INTEGER,
+    exit_code INTEGER,
+    status TEXT NOT NULL,
+    command TEXT NOT NULL,
+    package_inference_root TEXT,
+    context TEXT NOT NULL,
+    git_branch TEXT,
+    git_sha TEXT,
+    origination_user TEXT NOT NULL,
+    client_id TEXT NOT NULL,
+    client_name TEXT NOT NULL,
+    client_version TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS tasks (
+    id TEXT PRIMARY KEY,
+    run_id TEXT NOT NULL,
+    name TEXT NOT NULL,
+    package TEXT NOT NULL,
+    hash TEXT NOT NULL,
+    start_time INTEGER NOT NULL,
+    end_time INTEGER NOT NULL,
+    cache_status TEXT NOT NULL,
+    exit_code INTEGER,
+    logs TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS task_dependencies (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    task_id TEXT NOT NULL,
+    dependency_id TEXT NOT NULL
+);

crates/turborepo-db/src/lib.rs

@@ -0,0 +1,113 @@
+use camino::Utf8Path;
+use sqlx::{Pool, Sqlite, SqlitePool};
+use thiserror::Error;
+use turbopath::{AbsoluteSystemPath, AbsoluteSystemPathBuf};
+use turborepo_api_client::spaces::{CreateSpaceRunPayload, RunStatus, SpaceTaskSummary};
+use uuid::Uuid;
+
+#[derive(Debug, Error)]
+pub enum Error {
+    #[error("failed to connect to database: {0}")]
+    Sqlx(#[from] sqlx::Error),
+    #[error("failed to migrate database: {0}")]
+    Migrate(#[from] sqlx::migrate::MigrateError),
+    #[error("failed to serialize")]
+    Serialize(#[from] serde_json::Error),
+}
+
+#[derive(Clone)]
+pub struct DatabaseHandle {
+    pool: Pool<Sqlite>,
+}
+
+impl DatabaseHandle {
+    pub async fn new(cache_dir: &Utf8Path, repo_root: &AbsoluteSystemPath) -> Result<Self, Error> {
+        let cache_dir = AbsoluteSystemPathBuf::from_unknown(&repo_root, &cache_dir);
+        let pool = SqlitePool::connect(&format!(
+            "sqlite://{}?mode=rwc",
+            cache_dir.join_component("turbo.db")
+        ))
+        .await?;
+
+        sqlx::migrate!().run(&pool).await?;
+
+        Ok(Self { pool })
+    }
+
+    pub async fn create_run(&self, payload: &CreateSpaceRunPayload) -> Result<Uuid, Error> {
+        let id = Uuid::new_v4();
+        sqlx::query(
+            "INSERT INTO runs (
+              id,
+              start_time,
+              status,
+              command,
+              package_inference_root,
+              context,
+              git_branch,
+              git_sha,
+              origination_user,
+              client_id,
+              client_name,
+              client_version
+             ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)",
+        )
+        .bind(id.to_string())
+        .bind(payload.start_time)
+        .bind(payload.status.as_ref())
+        .bind(&payload.command)
+        .bind(&payload.package_inference_root)
+        .bind(payload.run_context)
+        .bind(&payload.git_branch)
+        .bind(&payload.git_sha)
+        .bind(&payload.user)
+        .bind(payload.client.id)
+        .bind(payload.client.name)
+        .bind(&payload.client.version)
+        .execute(&self.pool)
+        .await?;
+
+        Ok(id)
+    }
+
+    pub async fn finish_run(&self, id: Uuid, end_time: i64, exit_code: i32) -> Result<(), Error> {
+        sqlx::query("UPDATE runs SET status = $1, end_time = $2, exit_code = $3 WHERE id = $4")
+            .bind(RunStatus::Completed.as_ref())
+            .bind(end_time)
+            .bind(exit_code)
+            .bind(id.to_string())
+            .execute(&self.pool)
+            .await?;
+
+        Ok(())
+    }
+
+    pub async fn finish_task(&self, id: Uuid, summary: &SpaceTaskSummary) -> Result<(), Error> {
+        sqlx::query(
+            "INSERT INTO tasks (
+              run_id,
+              name,
+              package,
+              hash,
+              start_time,
+              end_time,
+              cache_status,
+              exit_code,
+              logs
+            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)",
+        )
+        .bind(id.to_string())
+        .bind(&summary.name)
+        .bind(&summary.workspace)
+        .bind(&summary.hash)
+        .bind(summary.start_time)
+        .bind(summary.end_time)
+        .bind(serde_json::to_string(&summary.cache)?)
+        .bind(summary.exit_code)
+        .bind(&summary.logs)
+        .execute(&self.pool)
+        .await?;
+
+        Ok(())
+    }
+}

crates/turborepo-lib/Cargo.toml

@@ -118,6 +118,7 @@ turborepo-api-client = { workspace = true }
 turborepo-auth = { path = "../turborepo-auth" }
 turborepo-cache = { workspace = true }
 turborepo-ci = { workspace = true }
+turborepo-db = { path = "../turborepo-db" }
 turborepo-dirs = { path = "../turborepo-dirs" }
 turborepo-env = { workspace = true }
 turborepo-errors = { workspace = true }
@@ -133,6 +134,7 @@ turborepo-unescape = { workspace = true }
 turborepo-vercel-api = { path = "../turborepo-vercel-api" }
 twox-hash = "1.6.3"
 uds_windows = "1.0.2"
+uuid = { version = "1.5.0", features = ["v4"] }
 wax = { workspace = true }
 webbrowser = { workspace = true }
 which = { workspace = true }

crates/turborepo-lib/src/run/mod.rs

@@ -411,6 +411,7 @@ impl Run {
             self.api_auth.clone(),
             Vendor::get_user(),
             &self.scm,
+            self.opts.cache_opts.cache_dir.clone(),
         );
 
         let mut visitor = Visitor::new(

crates/turborepo-lib/src/run/summary/mod.rs

@@ -13,6 +13,7 @@ mod task;
 mod task_factory;
 use std::{collections::HashSet, io, io::Write};
 
+use camino::Utf8PathBuf;
 use chrono::{DateTime, Local};
 pub use duration::TurboDuration;
 pub use execution::{TaskExecutionSummary, TaskTracker};
@@ -69,6 +70,8 @@ pub enum Error {
     Env(#[source] turborepo_env::Error),
     #[error("failed to construct task summary: {0}")]
     TaskSummary(#[from] task_factory::Error),
+    #[error(transparent)]
+    Database(#[from] turborepo_db::Error),
 }
 
 // NOTE: When changing this, please ensure that the server side is updated to
@@ -136,24 +139,28 @@ impl RunTracker {
         api_auth: Option<APIAuth>,
         user: String,
         scm: &SCM,
+        cache_dir: Utf8PathBuf,
     ) -> Self {
         let scm = SCMState::get(env_at_execution_start, scm, repo_root);
-
-        let spaces_client_handle =
-            SpacesClient::new(spaces_id.clone(), spaces_api_client, api_auth).and_then(
-                |spaces_client| {
-                    let payload = CreateSpaceRunPayload::new(
-                        started_at,
-                        synthesized_command.clone(),
-                        package_inference_root,
-                        scm.branch.clone(),
-                        scm.sha.clone(),
-                        version.to_string(),
-                        user.clone(),
-                    );
-                    spaces_client.start(payload).ok()
-                },
+        let spaces_client_handle = SpacesClient::new(
+            spaces_id.clone(),
+            spaces_api_client,
+            api_auth,
+            repo_root.to_owned(),
+            cache_dir,
+        )
+        .and_then(|spaces_client| {
+            let payload = CreateSpaceRunPayload::new(
+                started_at,
+                synthesized_command.clone(),
+                package_inference_root,
+                scm.branch.clone(),
+                scm.sha.clone(),
+                version.to_string(),
+                user.clone(),
             );
+            spaces_client.start(payload).ok()
+        });
 
         RunTracker {
             scm,