starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-26 08:01:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add sync_drafts and list_users_with_draft_on_path endpoints

Browse Source
This commit is contained in:
Diego Imbert
2026-05-28 00:26:42 +02:00
parent 39b9d8f9b8
commit 690741a5b2
8 changed files with 592 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0b8689175467c4720293faedc819a00016d0ece9ca8dcc62a23e639e44af872e.json
Generated
+43
View File
@@ -0,0 +1,43 @@
{
"db_name": "PostgreSQL",
"query": "SELECT value as \"value!: sqlx::types::Json<Box<serde_json::value::RawValue>>\", created_at\n FROM draft\n WHERE workspace_id = $1\n AND username = $2\n AND path = $3\n AND typ = $4\n AND created_at > $5",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "value!: sqlx::types::Json<Box<serde_json::value::RawValue>>",
"type_info": "Json"
},
{
"ordinal": 1,
"name": "created_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
{
"Custom": {
"name": "draft_type",
"kind": {
"Enum": [
"script",
"flow",
"app"
]
}
}
},
"Timestamptz"
]
},
"nullable": [
false,
false
]
},
"hash": "0b8689175467c4720293faedc819a00016d0ece9ca8dcc62a23e639e44af872e"
}
backend/.sqlx/query-3695b65f9308035d81088bf3f42c3ba4beba134018b50c93038c237cc573d285.json
Generated
+53
View File
@@ -0,0 +1,53 @@
{
"db_name": "PostgreSQL",
"query": "SELECT path, typ as \"typ: DraftType\", value as \"value!: sqlx::types::Json<Box<serde_json::value::RawValue>>\", created_at\n FROM draft\n WHERE workspace_id = $1\n AND username = $2\n AND created_at > $3",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "typ: DraftType",
"type_info": {
"Custom": {
"name": "draft_type",
"kind": {
"Enum": [
"script",
"flow",
"app"
]
}
}
}
},
{
"ordinal": 2,
"name": "value!: sqlx::types::Json<Box<serde_json::value::RawValue>>",
"type_info": "Json"
},
{
"ordinal": 3,
"name": "created_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Timestamptz"
]
},
"nullable": [
false,
false,
false,
false
]
},
"hash": "3695b65f9308035d81088bf3f42c3ba4beba134018b50c93038c237cc573d285"
}
backend/.sqlx/query-692d7e8b450fb97a2f2f194b1ba8d46908c21b52d75a6aa796bc50a01c8fb01d.json
Generated
+37
View File
@@ -0,0 +1,37 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO draft (workspace_id, username, path, typ, value, created_at)\n VALUES ($1, $2, $3, $4, $5::text::json, now())\n ON CONFLICT (workspace_id, path, typ, username) WHERE username IS NOT NULL\n DO UPDATE SET value = EXCLUDED.value, created_at = now()\n RETURNING created_at",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "created_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
{
"Custom": {
"name": "draft_type",
"kind": {
"Enum": [
"script",
"flow",
"app"
]
}
}
},
"Text"
]
},
"nullable": [
false
]
},
"hash": "692d7e8b450fb97a2f2f194b1ba8d46908c21b52d75a6aa796bc50a01c8fb01d"
}
backend/.sqlx/query-dce6d995445a30ff6d2ead7545c67bbde86e00dc4db3020e9905d1893908bd71.json
Generated
+52
View File
@@ -0,0 +1,52 @@
{
"db_name": "PostgreSQL",
"query": "SELECT path, typ as \"typ: DraftType\", value as \"value!: sqlx::types::Json<Box<serde_json::value::RawValue>>\", created_at\n FROM draft\n WHERE workspace_id = $1\n AND username = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "typ: DraftType",
"type_info": {
"Custom": {
"name": "draft_type",
"kind": {
"Enum": [
"script",
"flow",
"app"
]
}
}
}
},
{
"ordinal": 2,
"name": "value!: sqlx::types::Json<Box<serde_json::value::RawValue>>",
"type_info": "Json"
},
{
"ordinal": 3,
"name": "created_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false,
false,
false,
false
]
},
"hash": "dce6d995445a30ff6d2ead7545c67bbde86e00dc4db3020e9905d1893908bd71"
}
backend/.sqlx/query-f3565a507f21e047d92eb21315767f8e3c39a055ba0fecbce3986d8f371fed51.json
Generated
+41
View File
@@ -0,0 +1,41 @@
{
"db_name": "PostgreSQL",
"query": "SELECT username, created_at\n FROM draft\n WHERE workspace_id = $1\n AND path = $2\n AND typ = $3\n ORDER BY username NULLS LAST",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "username",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "created_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Text",
"Text",
{
"Custom": {
"name": "draft_type",
"kind": {
"Enum": [
"script",
"flow",
"app"
]
}
}
}
]
},
"nullable": [
true,
false
]
},
"hash": "f3565a507f21e047d92eb21315767f8e3c39a055ba0fecbce3986d8f371fed51"
}
backend/windmill-api/openapi.yaml
+136
View File
@@ -7802,6 +7802,142 @@ paths:
items:
type: string
/w/{workspace}/drafts/sync:
post:
summary: bi-directional user-draft sync
operationId: syncDrafts
tags:
- draft
parameters:
- $ref: "#/components/parameters/WorkspaceId"
requestBody:
required: true
content:
application/json:
schema:
type: object
properties:
last_sync:
type: string
format: date-time
description: Server timestamp of the last successful sync. Omit on first sync.
drafts:
type: array
items:
type: object
properties:
path:
type: string
typ:
type: string
enum: [script, flow, app]
value: {}
required: [path, typ, value]
force:
type: boolean
description: Skip the conflict check and overwrite the server copy.
required: [drafts]
responses:
"200":
description: sync result
content:
application/json:
schema:
type: object
properties:
missed_drafts:
type: array
items:
type: object
properties:
path:
type: string
typ:
type: string
enum: [script, flow, app]
value: {}
created_at:
type: string
format: date-time
required: [path, typ, value, created_at]
statuses:
type: array
items:
oneOf:
- type: object
properties:
status:
type: string
enum: [saved]
path:
type: string
typ:
type: string
enum: [script, flow, app]
created_at:
type: string
format: date-time
required: [status, path, typ, created_at]
- type: object
properties:
status:
type: string
enum: [rejected]
path:
type: string
typ:
type: string
enum: [script, flow, app]
server_value: {}
server_created_at:
type: string
format: date-time
incoming_value: {}
required:
- status
- path
- typ
- server_value
- server_created_at
- incoming_value
current_timestamp:
type: string
format: date-time
required: [missed_drafts, statuses, current_timestamp]
/w/{workspace}/drafts/users_with_draft/{kind}/{path}:
get:
summary: list users with a saved draft on a path
operationId: listUsersWithDraftOnPath
tags:
- draft
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- name: kind
in: path
required: true
schema:
type: string
enum: [script, flow, app]
- $ref: "#/components/parameters/ScriptPath"
responses:
"200":
description: list of usernames with a draft; `null` represents a legacy workspace-level draft
content:
application/json:
schema:
type: array
items:
type: object
properties:
username:
type: string
nullable: true
created_at:
type: string
format: date-time
required: [created_at]
/w/{workspace}/scripts/create:
post:
summary: create script
backend/windmill-api/src/drafts.rs
+228
View File
@@ -0,0 +1,228 @@
/*
* Author: Ruben Fiszel
* Copyright: Windmill Labs, Inc 2024
* This file and its contents are licensed under the AGPLv3 License.
* Please see the included NOTICE for copyright information and
* LICENSE-AGPL for a copy of the license.
*/
use crate::db::{ApiAuthed, DB};
use axum::{
extract::{Extension, Path},
routing::{get, post},
Json, Router,
};
use serde::{Deserialize, Serialize};
use windmill_common::error::Result;
pub fn workspaced_service() -> Router {
Router::new().route("/sync", post(sync_drafts)).route(
"/users_with_draft/{kind}/{*path}",
get(list_users_with_draft_on_path),
)
}
#[derive(sqlx::Type, Serialize, Deserialize, Debug, PartialEq, Clone, Copy)]
#[sqlx(type_name = "DRAFT_TYPE", rename_all = "lowercase")]
#[serde(rename_all(serialize = "lowercase", deserialize = "lowercase"))]
pub enum DraftType {
Script,
Flow,
App,
}
#[derive(Deserialize, Debug, Clone)]
pub struct IncomingDraft {
pub path: String,
pub typ: DraftType,
pub value: sqlx::types::Json<Box<serde_json::value::RawValue>>,
}
#[derive(Deserialize, Debug)]
pub struct SyncDraftsRequest {
/// Server timestamp of the client's last successful sync. Used both to
/// stream back drafts written by other sessions since then
/// (`missed_drafts`) and to detect conflicts when the client tries to
/// push a draft whose server copy moved forward (`status: rejected`).
pub last_sync: Option<chrono::DateTime<chrono::Utc>>,
pub drafts: Vec<IncomingDraft>,
/// When true, skip the conflict check and always overwrite the server copy.
#[serde(default)]
pub force: bool,
}
#[derive(Serialize, Debug)]
pub struct MissedDraft {
pub path: String,
pub typ: DraftType,
pub value: sqlx::types::Json<Box<serde_json::value::RawValue>>,
pub created_at: chrono::DateTime<chrono::Utc>,
}
#[derive(Serialize, Debug)]
#[serde(tag = "status", rename_all = "lowercase")]
pub enum DraftSyncStatus {
Saved {
path: String,
typ: DraftType,
created_at: chrono::DateTime<chrono::Utc>,
},
Rejected {
path: String,
typ: DraftType,
/// Current server copy at conflict-detection time.
server_value: sqlx::types::Json<Box<serde_json::value::RawValue>>,
server_created_at: chrono::DateTime<chrono::Utc>,
/// The value the client tried to push. Echoed back so the client can
/// retry with `force = true` without re-reading its local state.
incoming_value: sqlx::types::Json<Box<serde_json::value::RawValue>>,
},
}
#[derive(Serialize, Debug)]
pub struct SyncDraftsResponse {
pub missed_drafts: Vec<MissedDraft>,
pub statuses: Vec<DraftSyncStatus>,
pub current_timestamp: chrono::DateTime<chrono::Utc>,
}
async fn sync_drafts(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
Json(req): Json<SyncDraftsRequest>,
) -> Result<Json<SyncDraftsResponse>> {
let username = &authed.username;
let current_timestamp = sqlx::query_scalar!("SELECT now()")
.fetch_one(&db)
.await?
.expect("now() is never null");
let missed_drafts = if let Some(last_sync) = req.last_sync {
sqlx::query_as!(
MissedDraft,
r#"SELECT path, typ as "typ: DraftType", value as "value!: sqlx::types::Json<Box<serde_json::value::RawValue>>", created_at
FROM draft
WHERE workspace_id = $1
AND username = $2
AND created_at > $3"#,
&w_id,
username,
last_sync,
)
.fetch_all(&db)
.await?
} else {
// Initial sync — return everything the user has on the server.
sqlx::query_as!(
MissedDraft,
r#"SELECT path, typ as "typ: DraftType", value as "value!: sqlx::types::Json<Box<serde_json::value::RawValue>>", created_at
FROM draft
WHERE workspace_id = $1
AND username = $2"#,
&w_id,
username,
)
.fetch_all(&db)
.await?
};
let mut statuses = Vec::with_capacity(req.drafts.len());
for incoming in &req.drafts {
if !req.force {
if let Some(last_sync) = req.last_sync {
let conflict = sqlx::query!(
r#"SELECT value as "value!: sqlx::types::Json<Box<serde_json::value::RawValue>>", created_at
FROM draft
WHERE workspace_id = $1
AND username = $2
AND path = $3
AND typ = $4
AND created_at > $5"#,
&w_id,
username,
incoming.path,
incoming.typ as DraftType,
last_sync,
)
.fetch_optional(&db)
.await?;
if let Some(row) = conflict {
statuses.push(DraftSyncStatus::Rejected {
path: incoming.path.clone(),
typ: incoming.typ,
server_value: row.value,
server_created_at: row.created_at,
incoming_value: sqlx::types::Json(
serde_json::value::RawValue::from_string(
incoming.value.0.get().to_string(),
)
.expect("RawValue round-trip"),
),
});
continue;
}
}
}
let row = sqlx::query!(
r#"INSERT INTO draft (workspace_id, username, path, typ, value, created_at)
VALUES ($1, $2, $3, $4, $5::text::json, now())
ON CONFLICT (workspace_id, path, typ, username) WHERE username IS NOT NULL
DO UPDATE SET value = EXCLUDED.value, created_at = now()
RETURNING created_at"#,
&w_id,
username,
incoming.path,
incoming.typ as DraftType,
serde_json::to_string(&incoming.value).unwrap(),
)
.fetch_one(&db)
.await?;
statuses.push(DraftSyncStatus::Saved {
path: incoming.path.clone(),
typ: incoming.typ,
created_at: row.created_at,
});
}
Ok(Json(SyncDraftsResponse {
missed_drafts,
statuses,
current_timestamp,
}))
}
#[derive(Serialize, Debug)]
pub struct UserWithDraft {
/// `None` represents a legacy workspace-level draft (no owner).
pub username: Option<String>,
pub created_at: chrono::DateTime<chrono::Utc>,
}
async fn list_users_with_draft_on_path(
_authed: ApiAuthed,
Extension(db): Extension<DB>,
Path((w_id, kind, path)): Path<(String, DraftType, windmill_common::utils::StripPath)>,
) -> Result<Json<Vec<UserWithDraft>>> {
let rows = sqlx::query_as!(
UserWithDraft,
r#"SELECT username, created_at
FROM draft
WHERE workspace_id = $1
AND path = $2
AND typ = $3
ORDER BY username NULLS LAST"#,
&w_id,
path.to_path(),
kind as DraftType,
)
.fetch_all(&db)
.await?;
Ok(Json(rows))
}
backend/windmill-api/src/lib.rs
+2
View File
@@ -77,6 +77,7 @@ mod capture;
mod concurrency_groups;
mod db;
mod db_health;
mod drafts;
#[cfg(feature = "private")]
pub mod ee;
@@ -552,6 +553,7 @@ pub async fn run_server(
"/concurrency_groups",
concurrency_groups::workspaced_service(),
)
.nest("/drafts", drafts::workspaced_service())
.nest("/embeddings", embeddings::workspaced_service())
.nest("/favorites", favorite::workspaced_service())
.nest("/flows", flows::workspaced_service())