diff --git a/backend/migrations/20220802211304_capture.down.sql b/backend/migrations/20220802211304_capture.down.sql new file mode 100644 index 0000000000..4cc134749c --- /dev/null +++ b/backend/migrations/20220802211304_capture.down.sql @@ -0,0 +1 @@ +DROP TABLE capture; diff --git a/backend/migrations/20220802211304_capture.up.sql b/backend/migrations/20220802211304_capture.up.sql new file mode 100644 index 0000000000..12360a03f3 --- /dev/null +++ b/backend/migrations/20220802211304_capture.up.sql @@ -0,0 +1,21 @@ +CREATE TABLE capture ( + workspace_id VARCHAR(50) NOT NULL, + path VARCHAR(255) NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + created_by VARCHAR(50) NOT NULL, + payload JSONB NOT NULL DEFAULT 'null'::jsonb + CHECK (length(payload::text) < 10 * 1024), + + PRIMARY KEY (workspace_id, path), + FOREIGN KEY (workspace_id) REFERENCES workspace(id) +); + +ALTER TABLE capture ENABLE ROW LEVEL SECURITY; + +CREATE POLICY see_own ON capture FOR ALL +USING ( SPLIT_PART(capture.path, '/', 1) = 'u' + AND SPLIT_PART(capture.path, '/', 2) = current_setting('session.user')); + +CREATE POLICY see_member ON capture FOR ALL +USING ( SPLIT_PART(capture.path, '/', 1) = 'g' + AND SPLIT_PART(capture.path, '/', 2) = any(regexp_split_to_array(current_setting('session.groups'), ',')::text[])); diff --git a/backend/openapi.yaml b/backend/openapi.yaml index de5b29b67e..bd25d68d78 100644 --- a/backend/openapi.yaml +++ b/backend/openapi.yaml @@ -3268,6 +3268,47 @@ paths: text/plain: schema: type: string + + /w/{workspace}/capture/{path}: + put: + summary: create flow preview capture + operationId: createCapture + tags: + - capture + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/Path" + responses: + "201": + description: flow preview capture created + post: + summary: update flow preview capture + operationId: updateCapture + tags: + - capture + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/Path" + responses: + "204": + description: flow preview captured + get: + summary: get flow preview capture + operationId: getCapture + tags: + - capture + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/Path" + responses: + "200": + description: captured flow preview + content: + application/json: + schema: {} + "404": + description: capture does not exist for this flow + components: securitySchemes: bearerAuth: diff --git a/backend/sqlx-data.json b/backend/sqlx-data.json index 9d9e167936..8cfdeea5d0 100644 --- a/backend/sqlx-data.json +++ b/backend/sqlx-data.json @@ -181,6 +181,20 @@ }, "query": "INSERT INTO workspace_invite\n (workspace_id, email, is_admin)\n VALUES ($1, $2, $3)" }, + "0a9a191273c735c41d56ea46a39ffca075a0550eada87df7162c5037164ad6bf": { + "describe": { + "columns": [], + "nullable": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Varchar" + ] + } + }, + "query": "\n INSERT INTO capture\n (workspace_id, path, created_by)\n VALUES ($1, $2, $3)\n ON CONFLICT (workspace_id, path)\n DO UPDATE SET created_at = now()\n " + }, "0ae9160591ae00117d20a616cfe07e38f0c32953c7e881e916c389255190b72d": { "describe": { "columns": [], @@ -2714,6 +2728,20 @@ }, "query": "SELECT (flow_status->'step')::integer FROM queue WHERE id = $1" }, + "b9468b9e16f55db11b33d8e9793e6e3ae6c5add6ca02414140adb724120a6800": { + "describe": { + "columns": [], + "nullable": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Jsonb" + ] + } + }, + "query": "\n UPDATE capture\n SET payload = $3\n WHERE workspace_id = $1\n AND path = $2\n " + }, "bb56e61c7cfb09c0a28fb3226dfe91704c70d9fe15eda18e6889adfd7496f80b": { "describe": { "columns": [ @@ -3211,6 +3239,20 @@ }, "query": "SELECT EXISTS(SELECT 1 FROM workspace WHERE workspace.id = $1)" }, + "e02b99525cb1f8737acfec86809f59c8cff67bb1ec3926680fb691cc3573738a": { + "describe": { + "columns": [], + "nullable": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Int8" + ] + } + }, + "query": "\n DELETE FROM capture\n WHERE workspace_id = $1\n AND created_by = $2\n AND created_at <=\n ( SELECT created_at\n FROM capture\n WHERE workspace_id = $1\n AND created_by = $2\n ORDER BY created_at DESC\n OFFSET $3\n LIMIT 1 )\n " + }, "e262f83b672a558092dc959b28a919f18e44b2a0d03b27b0ddba99896940c6d3": { "describe": { "columns": [ @@ -3387,6 +3429,27 @@ }, "query": "DELETE FROM usr_to_group WHERE usr = $1 AND group_ = $2 AND workspace_id = $3" }, + "f4960efbb4189b595541608c9ab5cbbd56eb42059e961b6338b685105626a5c7": { + "describe": { + "columns": [ + { + "name": "payload", + "ordinal": 0, + "type_info": "Jsonb" + } + ], + "nullable": [ + false + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + } + }, + "query": "\n SELECT payload\n FROM capture\n WHERE workspace_id = $1\n AND path = $2\n " + }, "f4a1bb4dfeafa3f28385c19d10730f2f49b4ebe5bba631cee46c760cee2e6949": { "describe": { "columns": [ diff --git a/backend/src/capture.rs b/backend/src/capture.rs new file mode 100644 index 0000000000..19705860c8 --- /dev/null +++ b/backend/src/capture.rs @@ -0,0 +1,125 @@ +use axum::{ + extract::{Extension, Path}, + routing::{get, post, put}, + Json, Router, +}; +use hyper::StatusCode; + +use crate::{ + db::{UserDB, DB}, + error::{JsonResult, Result}, + users::Authed, + utils::{not_found_if_none, StripPath}, +}; + +const KEEP_LAST: i64 = 8; + +pub fn workspaced_service() -> Router { + Router::new() + .route("/*path", put(new_payload)) + .route("/*path", get(get_payload)) +} + +pub fn global_service() -> Router { + Router::new().route("/*path", post(update_payload)) +} + +pub async fn new_payload( + authed: Authed, + Extension(user_db): Extension, + Path((w_id, path)): Path<(String, StripPath)>, +) -> Result { + let mut tx = user_db.begin(&authed).await?; + + sqlx::query!( + " + INSERT INTO capture + (workspace_id, path, created_by) + VALUES ($1, $2, $3) + ON CONFLICT (workspace_id, path) + DO UPDATE SET created_at = now() + ", + &w_id, + &path.to_path(), + &authed.username, + ) + .execute(&mut tx) + .await?; + + /* Retain only KEEP_LAST most recent captures by this user in this workspace. */ + sqlx::query!( + " + DELETE FROM capture + WHERE workspace_id = $1 + AND created_by = $2 + AND created_at <= + ( SELECT created_at + FROM capture + WHERE workspace_id = $1 + AND created_by = $2 + ORDER BY created_at DESC + OFFSET $3 + LIMIT 1 ) + ", + &w_id, + &authed.username, + KEEP_LAST, + ) + .execute(&mut tx) + .await?; + + tx.commit().await?; + + Ok(StatusCode::CREATED) +} + +pub async fn update_payload( + Extension(db): Extension, + Path((w_id, path)): Path<(String, StripPath)>, + Json(payload): Json, +) -> Result { + let mut tx = db.begin().await?; + + sqlx::query!( + " + UPDATE capture + SET payload = $3 + WHERE workspace_id = $1 + AND path = $2 + ", + &w_id, + &path.to_path(), + &payload, + ) + .execute(&mut tx) + .await?; + + tx.commit().await?; + + Ok(StatusCode::NO_CONTENT) +} + +pub async fn get_payload( + authed: Authed, + Extension(user_db): Extension, + Path((w_id, path)): Path<(String, StripPath)>, +) -> JsonResult { + let mut tx = user_db.begin(&authed).await?; + + let payload = sqlx::query_scalar!( + " + SELECT payload + FROM capture + WHERE workspace_id = $1 + AND path = $2 + ", + &w_id, + &path.to_path(), + ) + .fetch_optional(&mut tx) + .await?; + + tx.commit().await?; + + not_found_if_none(payload, "capture", path.to_path()).map(axum::Json) +} diff --git a/backend/src/lib.rs b/backend/src/lib.rs index 91a0962875..97c4463368 100644 --- a/backend/src/lib.rs +++ b/backend/src/lib.rs @@ -22,6 +22,7 @@ extern crate magic_crypt; extern crate dotenv; mod audit; +mod capture; mod client; mod db; mod error; @@ -155,7 +156,8 @@ pub async fn run_server( .nest("/audit", audit::workspaced_service()) .nest("/acls", granular_acls::workspaced_service()) .nest("/workspaces", workspaces::workspaced_service()) - .nest("/flows", flows::workspaced_service()), + .nest("/flows", flows::workspaced_service()) + .nest("/capture", capture::workspaced_service()), ) .nest("/workspaces", workspaces::global_service()) .nest( @@ -169,6 +171,7 @@ pub async fn run_server( .route_layer(from_extractor::()) .route_layer(from_extractor::()) .nest("/w/:workspace_id/jobs", jobs::global_service()) + .nest("/w/:workspace_id/capture", capture::global_service()) .nest( "/auth", users::make_unauthed_service().layer(Extension(argon2)),