Add workers commands with versioned deployments

- workers list/get/create/delete
- workers deploy with versioning (v1, v2, ...)
- Backend trait abstracts API vs DB
- Supports .js, .ts, .wasm files

Author: max-lt

Cargo.lock

@@ -16,5 +16,9 @@ dirs = "6"
 thiserror = "2"
 colored = "3"
 tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
-sqlx = { version = "0.8", features = ["runtime-tokio", "postgres", "chrono"] }
+sqlx = { version = "0.8", features = ["runtime-tokio", "postgres", "chrono", "uuid"] }
 chrono = { version = "0.4", features = ["serde"] }
+uuid = { version = "1", features = ["serde"] }
+reqwest = { version = "0.12", features = ["json"] }
+sha2 = "0.10"
+hex = "0.4"

Cargo.toml

@@ -58,15 +58,41 @@ ow infra db status
 ow infra db baseline
 ```
 
+### Workers
+
+Works with both `api` and `db` aliases.
+
+```bash
+# List workers
+ow workers list
+ow workers ls
+
+# Create a worker
+ow workers create my-api -d "My API worker"
+
+# Get worker details
+ow workers get my-api
+
+# Deploy code to a worker
+ow workers deploy my-api ./src/index.ts -m "Initial deploy"
+ow workers deploy my-api ./src/index.ts -m "Fix bug"  # v2, v3, ...
+
+# Delete a worker
+ow workers delete my-api
+ow workers rm my-api
+```
+
+Supported file types: `.js`, `.ts`, `.wasm`
+
 ### Using Aliases
 
 ```bash
 # Use default alias
-ow db status
+ow workers list
 
 # Use specific alias (mc-style, alias as first argument)
+ow prod workers list
 ow infra db migrate
-ow prod db status
 ```
 
 ## Config File Format

README.md

@@ -0,0 +1,152 @@
+use super::{Backend, BackendError, CreateWorkerInput, DeployInput, Deployment, Worker};
+use reqwest::Client;
+
+pub struct ApiBackend {
+    client: Client,
+    base_url: String,
+    token: Option<String>,
+}
+
+impl ApiBackend {
+    pub fn new(base_url: String, token: Option<String>) -> Self {
+        Self {
+            client: Client::new(),
+            base_url,
+            token,
+        }
+    }
+
+    fn request(&self, method: reqwest::Method, path: &str) -> reqwest::RequestBuilder {
+        let url = format!("{}{}", self.base_url, path);
+        let mut req = self.client.request(method, &url);
+
+        if let Some(token) = &self.token {
+            req = req.bearer_auth(token);
+        }
+
+        req
+    }
+}
+
+impl Backend for ApiBackend {
+    async fn list_workers(&self) -> Result<Vec<Worker>, BackendError> {
+        let response = self
+            .request(reqwest::Method::GET, "/workers")
+            .send()
+            .await?;
+
+        if response.status() == reqwest::StatusCode::UNAUTHORIZED {
+            return Err(BackendError::Unauthorized);
+        }
+
+        if !response.status().is_success() {
+            let text = response.text().await.unwrap_or_default();
+            return Err(BackendError::Api(text));
+        }
+
+        let workers: Vec<Worker> = response.json().await?;
+        Ok(workers)
+    }
+
+    async fn get_worker(&self, name: &str) -> Result<Worker, BackendError> {
+        let response = self
+            .request(reqwest::Method::GET, &format!("/workers/{}", name))
+            .send()
+            .await?;
+
+        if response.status() == reqwest::StatusCode::NOT_FOUND {
+            return Err(BackendError::NotFound(format!(
+                "Worker '{}' not found",
+                name
+            )));
+        }
+
+        if response.status() == reqwest::StatusCode::UNAUTHORIZED {
+            return Err(BackendError::Unauthorized);
+        }
+
+        if !response.status().is_success() {
+            let text = response.text().await.unwrap_or_default();
+            return Err(BackendError::Api(text));
+        }
+
+        let worker: Worker = response.json().await?;
+        Ok(worker)
+    }
+
+    async fn create_worker(&self, input: CreateWorkerInput) -> Result<Worker, BackendError> {
+        let response = self
+            .request(reqwest::Method::POST, "/workers")
+            .json(&input)
+            .send()
+            .await?;
+
+        if response.status() == reqwest::StatusCode::UNAUTHORIZED {
+            return Err(BackendError::Unauthorized);
+        }
+
+        if !response.status().is_success() {
+            let text = response.text().await.unwrap_or_default();
+            return Err(BackendError::Api(text));
+        }
+
+        let worker: Worker = response.json().await?;
+        Ok(worker)
+    }
+
+    async fn delete_worker(&self, name: &str) -> Result<(), BackendError> {
+        let response = self
+            .request(reqwest::Method::DELETE, &format!("/workers/{}", name))
+            .send()
+            .await?;
+
+        if response.status() == reqwest::StatusCode::NOT_FOUND {
+            return Err(BackendError::NotFound(format!(
+                "Worker '{}' not found",
+                name
+            )));
+        }
+
+        if response.status() == reqwest::StatusCode::UNAUTHORIZED {
+            return Err(BackendError::Unauthorized);
+        }
+
+        if !response.status().is_success() {
+            let text = response.text().await.unwrap_or_default();
+            return Err(BackendError::Api(text));
+        }
+
+        Ok(())
+    }
+
+    async fn deploy_worker(
+        &self,
+        name: &str,
+        input: DeployInput,
+    ) -> Result<Deployment, BackendError> {
+        let response = self
+            .request(reqwest::Method::POST, &format!("/workers/{}/deploy", name))
+            .json(&input)
+            .send()
+            .await?;
+
+        if response.status() == reqwest::StatusCode::NOT_FOUND {
+            return Err(BackendError::NotFound(format!(
+                "Worker '{}' not found",
+                name
+            )));
+        }
+
+        if response.status() == reqwest::StatusCode::UNAUTHORIZED {
+            return Err(BackendError::Unauthorized);
+        }
+
+        if !response.status().is_success() {
+            let text = response.text().await.unwrap_or_default();
+            return Err(BackendError::Api(text));
+        }
+
+        let deployment: Deployment = response.json().await?;
+        Ok(deployment)
+    }
+}

src/backend/api.rs

@@ -0,0 +1,177 @@
+use super::{Backend, BackendError, CreateWorkerInput, DeployInput, Deployment, Worker};
+use sha2::{Digest, Sha256};
+use sqlx::{PgPool, Row};
+
+pub struct DbBackend {
+    pool: PgPool,
+}
+
+impl DbBackend {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+impl Backend for DbBackend {
+    async fn list_workers(&self) -> Result<Vec<Worker>, BackendError> {
+        let rows = sqlx::query(
+            r#"
+            SELECT id, name, "desc", current_version, created_at, updated_at
+            FROM workers
+            ORDER BY name
+            "#,
+        )
+        .fetch_all(&self.pool)
+        .await?;
+
+        let workers = rows
+            .iter()
+            .map(|row| Worker {
+                id: row.get::<uuid::Uuid, _>("id").to_string(),
+                name: row.get("name"),
+                description: row.get("desc"),
+                current_version: row.get("current_version"),
+                created_at: row.get("created_at"),
+                updated_at: row.get("updated_at"),
+            })
+            .collect();
+
+        Ok(workers)
+    }
+
+    async fn get_worker(&self, name: &str) -> Result<Worker, BackendError> {
+        let row = sqlx::query(
+            r#"
+            SELECT id, name, "desc", current_version, created_at, updated_at
+            FROM workers
+            WHERE name = $1
+            "#,
+        )
+        .bind(name)
+        .fetch_optional(&self.pool)
+        .await?
+        .ok_or_else(|| BackendError::NotFound(format!("Worker '{}' not found", name)))?;
+
+        Ok(Worker {
+            id: row.get::<uuid::Uuid, _>("id").to_string(),
+            name: row.get("name"),
+            description: row.get("desc"),
+            current_version: row.get("current_version"),
+            created_at: row.get("created_at"),
+            updated_at: row.get("updated_at"),
+        })
+    }
+
+    async fn create_worker(&self, input: CreateWorkerInput) -> Result<Worker, BackendError> {
+        // For CLI/admin mode, we need a user_id
+        // For now, get or create an "admin" user
+        let user_id: uuid::Uuid = sqlx::query_scalar(
+            r#"
+            INSERT INTO users (id, username, created_at, updated_at)
+            VALUES (gen_random_uuid(), 'cli-admin', now(), now())
+            ON CONFLICT (username) DO UPDATE SET username = users.username
+            RETURNING id
+            "#,
+        )
+        .fetch_one(&self.pool)
+        .await?;
+
+        let row = sqlx::query(
+            r#"
+            INSERT INTO workers (name, "desc", user_id)
+            VALUES ($1, $2, $3)
+            RETURNING id, name, "desc", current_version, created_at, updated_at
+            "#,
+        )
+        .bind(&input.name)
+        .bind(&input.description)
+        .bind(user_id)
+        .fetch_one(&self.pool)
+        .await?;
+
+        Ok(Worker {
+            id: row.get::<uuid::Uuid, _>("id").to_string(),
+            name: row.get("name"),
+            description: row.get("desc"),
+            current_version: row.get("current_version"),
+            created_at: row.get("created_at"),
+            updated_at: row.get("updated_at"),
+        })
+    }
+
+    async fn delete_worker(&self, name: &str) -> Result<(), BackendError> {
+        let result = sqlx::query("DELETE FROM workers WHERE name = $1")
+            .bind(name)
+            .execute(&self.pool)
+            .await?;
+
+        if result.rows_affected() == 0 {
+            return Err(BackendError::NotFound(format!(
+                "Worker '{}' not found",
+                name
+            )));
+        }
+
+        Ok(())
+    }
+
+    async fn deploy_worker(
+        &self,
+        name: &str,
+        input: DeployInput,
+    ) -> Result<Deployment, BackendError> {
+        // Get worker ID
+        let worker_id: uuid::Uuid = sqlx::query_scalar("SELECT id FROM workers WHERE name = $1")
+            .bind(name)
+            .fetch_optional(&self.pool)
+            .await?
+            .ok_or_else(|| BackendError::NotFound(format!("Worker '{}' not found", name)))?;
+
+        // Calculate hash
+        let mut hasher = Sha256::new();
+        hasher.update(&input.code);
+        let hash = hex::encode(hasher.finalize());
+
+        // Get next version
+        let current_version: Option<i32> =
+            sqlx::query_scalar("SELECT MAX(version) FROM worker_deployments WHERE worker_id = $1")
+                .bind(worker_id)
+                .fetch_one(&self.pool)
+                .await?;
+
+        let next_version = current_version.unwrap_or(0) + 1;
+
+        // Insert deployment
+        let row = sqlx::query(
+            r#"
+            INSERT INTO worker_deployments (worker_id, version, hash, code_type, code, message)
+            VALUES ($1, $2, $3, $4::enum_code_type, $5, $6)
+            RETURNING worker_id, version, hash, code_type::text, deployed_at, message
+            "#,
+        )
+        .bind(worker_id)
+        .bind(next_version)
+        .bind(&hash)
+        .bind(&input.code_type)
+        .bind(&input.code)
+        .bind(&input.message)
+        .fetch_one(&self.pool)
+        .await?;
+
+        // Update worker's current_version
+        sqlx::query("UPDATE workers SET current_version = $1 WHERE id = $2")
+            .bind(next_version)
+            .bind(worker_id)
+            .execute(&self.pool)
+            .await?;
+
+        Ok(Deployment {
+            worker_id: row.get::<uuid::Uuid, _>("worker_id").to_string(),
+            version: row.get("version"),
+            hash: row.get("hash"),
+            code_type: row.get("code_type"),
+            deployed_at: row.get("deployed_at"),
+            message: row.get("message"),
+        })
+    }
+}