mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-28 00:02:42 +00:00
feat: sync all userdraft kinds, switch draft owner to email FK, add id PK, scope draft list to readable paths
This commit is contained in:
+4
-15
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"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",
|
||||
"query": "SELECT path, typ, value as \"value!: sqlx::types::Json<Box<serde_json::value::RawValue>>\", created_at\n FROM draft\n WHERE workspace_id = $1\n AND email = $2",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
@@ -10,19 +10,8 @@
|
||||
},
|
||||
{
|
||||
"ordinal": 1,
|
||||
"name": "typ: DraftType",
|
||||
"type_info": {
|
||||
"Custom": {
|
||||
"name": "draft_type",
|
||||
"kind": {
|
||||
"Enum": [
|
||||
"script",
|
||||
"flow",
|
||||
"app"
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
"name": "typ",
|
||||
"type_info": "Varchar"
|
||||
},
|
||||
{
|
||||
"ordinal": 2,
|
||||
@@ -48,5 +37,5 @@
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "dce6d995445a30ff6d2ead7545c67bbde86e00dc4db3020e9905d1893908bd71"
|
||||
"hash": "05de4eda0c70e1fec7d1c1f996971ad550377ddca9ce98031a49eecb4a8e4c9d"
|
||||
}
|
||||
+3
-14
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"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",
|
||||
"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 email = $2\n AND path = $3\n AND typ = $4\n AND created_at > $5",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
@@ -19,18 +19,7 @@
|
||||
"Text",
|
||||
"Text",
|
||||
"Text",
|
||||
{
|
||||
"Custom": {
|
||||
"name": "draft_type",
|
||||
"kind": {
|
||||
"Enum": [
|
||||
"script",
|
||||
"flow",
|
||||
"app"
|
||||
]
|
||||
}
|
||||
}
|
||||
},
|
||||
"Text",
|
||||
"Timestamptz"
|
||||
]
|
||||
},
|
||||
@@ -39,5 +28,5 @@
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "0b8689175467c4720293faedc819a00016d0ece9ca8dcc62a23e639e44af872e"
|
||||
"hash": "2d8657ce809262cc261a381bcb19549970aa645f9e23c2525311fc1161fd21ff"
|
||||
}
|
||||
-53
@@ -1,53 +0,0 @@
|
||||
{
|
||||
"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"
|
||||
}
|
||||
-37
@@ -1,37 +0,0 @@
|
||||
{
|
||||
"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"
|
||||
}
|
||||
+30
@@ -0,0 +1,30 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT email, created_at\n FROM draft\n WHERE workspace_id = $1\n AND path = $2\n AND typ = $3\n ORDER BY email NULLS LAST",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "email",
|
||||
"type_info": "Varchar"
|
||||
},
|
||||
{
|
||||
"ordinal": 1,
|
||||
"name": "created_at",
|
||||
"type_info": "Timestamptz"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"Text",
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
true,
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "ca70f58d3c301c5b31749ea753dd1f63989c2fdae78e7e33053986862853dda5"
|
||||
}
|
||||
+26
@@ -0,0 +1,26 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "INSERT INTO draft (workspace_id, email, path, typ, value, created_at)\n VALUES ($1, $2, $3, $4, $5::text::json, now())\n ON CONFLICT (workspace_id, path, typ, email) WHERE email 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",
|
||||
"Varchar",
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "ddd58ed070d8c0bbbb060cd881ff618561958f183815f3dc339cf06c6f40946a"
|
||||
}
|
||||
-41
@@ -1,41 +0,0 @@
|
||||
{
|
||||
"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"
|
||||
}
|
||||
+42
@@ -0,0 +1,42 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT path, typ, value as \"value!: sqlx::types::Json<Box<serde_json::value::RawValue>>\", created_at\n FROM draft\n WHERE workspace_id = $1\n AND email = $2\n AND created_at > $3",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "path",
|
||||
"type_info": "Varchar"
|
||||
},
|
||||
{
|
||||
"ordinal": 1,
|
||||
"name": "typ",
|
||||
"type_info": "Varchar"
|
||||
},
|
||||
{
|
||||
"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": "f3afdf5898105679c9e032269a94e9d2dc904718e67a3583ae6bedd1c646a05c"
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
-- Add down migration script here
|
||||
ALTER TABLE draft DROP COLUMN id;
|
||||
|
||||
-- Recreate the enum. Drop any rows whose typ falls outside the original
|
||||
-- enum values; without this the type cast below would fail.
|
||||
DELETE FROM draft WHERE typ NOT IN ('script', 'flow', 'app');
|
||||
CREATE TYPE DRAFT_TYPE AS ENUM ('script', 'flow', 'app');
|
||||
ALTER TABLE draft ALTER COLUMN typ TYPE DRAFT_TYPE USING typ::DRAFT_TYPE;
|
||||
@@ -0,0 +1,14 @@
|
||||
-- Add up migration script here
|
||||
-- Loosen draft.typ from the DRAFT_TYPE enum to a plain VARCHAR so the
|
||||
-- bidirectional sync layer can store drafts for any UserDraft item kind
|
||||
-- (raw_app, resource, variable, trigger_*, ...) without an enum migration
|
||||
-- per kind.
|
||||
ALTER TABLE draft ALTER COLUMN typ TYPE VARCHAR(50) USING typ::text;
|
||||
DROP TYPE DRAFT_TYPE;
|
||||
|
||||
-- Give the draft table a proper synthetic primary key. The previous
|
||||
-- migration replaced the composite PK with two partial unique indexes
|
||||
-- (one for user-scoped rows, one for legacy NULL-username rows); having
|
||||
-- no PK at all breaks tools that assume one (pg_dump, replication, ORM
|
||||
-- drift detection).
|
||||
ALTER TABLE draft ADD COLUMN id BIGSERIAL PRIMARY KEY;
|
||||
@@ -0,0 +1,34 @@
|
||||
-- Add down migration script here
|
||||
ALTER TABLE draft DROP CONSTRAINT IF EXISTS draft_password_fkey;
|
||||
DROP INDEX IF EXISTS draft_pkey_with_user;
|
||||
|
||||
-- Best-effort: map back to a workspace username if the email matches a
|
||||
-- single `usr` row. Drafts whose email doesn't resolve to a usr row are
|
||||
-- dropped — the previous schema had no way to represent them.
|
||||
UPDATE draft
|
||||
SET email = usr.username
|
||||
FROM usr
|
||||
WHERE draft.workspace_id = usr.workspace_id
|
||||
AND draft.email = usr.email;
|
||||
|
||||
DELETE FROM draft
|
||||
WHERE email IS NOT NULL
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM usr
|
||||
WHERE usr.workspace_id = draft.workspace_id
|
||||
AND usr.username = draft.email
|
||||
);
|
||||
|
||||
ALTER TABLE draft RENAME COLUMN email TO username;
|
||||
ALTER TABLE draft ALTER COLUMN username TYPE VARCHAR(50);
|
||||
|
||||
ALTER TABLE draft
|
||||
ADD CONSTRAINT draft_usr_fkey
|
||||
FOREIGN KEY (workspace_id, username)
|
||||
REFERENCES usr(workspace_id, username)
|
||||
ON DELETE CASCADE
|
||||
ON UPDATE CASCADE;
|
||||
|
||||
CREATE UNIQUE INDEX draft_pkey_with_user
|
||||
ON draft (workspace_id, path, typ, username)
|
||||
WHERE username IS NOT NULL;
|
||||
@@ -0,0 +1,28 @@
|
||||
-- Add up migration script here
|
||||
-- Switch draft ownership from per-workspace `username` (FK to `usr`) to
|
||||
-- per-instance `email` (FK to `password`). Email is always populated on
|
||||
-- the authed session (including super-admins viewing foreign workspaces
|
||||
-- they aren't members of), so the FK no longer blocks legitimate writes.
|
||||
ALTER TABLE draft DROP CONSTRAINT IF EXISTS draft_usr_fkey;
|
||||
|
||||
DROP INDEX IF EXISTS draft_pkey_with_user;
|
||||
|
||||
ALTER TABLE draft RENAME COLUMN username TO email;
|
||||
ALTER TABLE draft ALTER COLUMN email TYPE VARCHAR(255);
|
||||
|
||||
UPDATE draft
|
||||
SET email = usr.email
|
||||
FROM usr
|
||||
WHERE draft.workspace_id = usr.workspace_id
|
||||
AND draft.email = usr.username;
|
||||
|
||||
ALTER TABLE draft
|
||||
ADD CONSTRAINT draft_password_fkey
|
||||
FOREIGN KEY (email)
|
||||
REFERENCES password(email)
|
||||
ON DELETE CASCADE
|
||||
ON UPDATE CASCADE;
|
||||
|
||||
CREATE UNIQUE INDEX draft_pkey_with_user
|
||||
ON draft (workspace_id, path, typ, email)
|
||||
WHERE email IS NOT NULL;
|
||||
@@ -7830,7 +7830,7 @@ paths:
|
||||
type: string
|
||||
typ:
|
||||
type: string
|
||||
enum: [script, flow, app]
|
||||
description: UserDraft item kind. Stored verbatim; the client owns the value set.
|
||||
value: {}
|
||||
force:
|
||||
type: boolean
|
||||
@@ -7854,7 +7854,6 @@ paths:
|
||||
type: string
|
||||
typ:
|
||||
type: string
|
||||
enum: [script, flow, app]
|
||||
value: {}
|
||||
created_at:
|
||||
type: string
|
||||
@@ -7873,7 +7872,6 @@ paths:
|
||||
type: string
|
||||
typ:
|
||||
type: string
|
||||
enum: [script, flow, app]
|
||||
created_at:
|
||||
type: string
|
||||
format: date-time
|
||||
@@ -7887,7 +7885,6 @@ paths:
|
||||
type: string
|
||||
typ:
|
||||
type: string
|
||||
enum: [script, flow, app]
|
||||
server_value: {}
|
||||
server_created_at:
|
||||
type: string
|
||||
@@ -7918,11 +7915,10 @@ paths:
|
||||
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
|
||||
description: list of users (by email) with a draft; `null` represents a legacy workspace-level draft
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
@@ -7930,7 +7926,7 @@ paths:
|
||||
items:
|
||||
type: object
|
||||
properties:
|
||||
username:
|
||||
email:
|
||||
type: string
|
||||
nullable: true
|
||||
created_at:
|
||||
|
||||
@@ -14,7 +14,10 @@ use axum::{
|
||||
Json, Router,
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use windmill_common::error::Result;
|
||||
use windmill_common::{
|
||||
db::UserDB,
|
||||
error::{Error, Result},
|
||||
};
|
||||
|
||||
pub fn workspaced_service() -> Router {
|
||||
Router::new().route("/sync", post(sync_drafts)).route(
|
||||
@@ -23,19 +26,14 @@ pub fn workspaced_service() -> Router {
|
||||
)
|
||||
}
|
||||
|
||||
#[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,
|
||||
/// Free-form item kind matching the frontend's `UserDraftItemKind`
|
||||
/// (`script`, `flow`, `app`, `raw_app`, `resource`, `variable`,
|
||||
/// `trigger_*`, ...). Stored verbatim — no server-side validation,
|
||||
/// since the set of kinds is owned by the client.
|
||||
pub typ: String,
|
||||
pub value: sqlx::types::Json<Box<serde_json::value::RawValue>>,
|
||||
/// When true, skip the conflict check for this entry and overwrite the
|
||||
/// server copy. Only the matching entry is forced — other entries in
|
||||
@@ -57,7 +55,7 @@ pub struct SyncDraftsRequest {
|
||||
#[derive(Serialize, Debug)]
|
||||
pub struct MissedDraft {
|
||||
pub path: String,
|
||||
pub typ: DraftType,
|
||||
pub typ: String,
|
||||
pub value: sqlx::types::Json<Box<serde_json::value::RawValue>>,
|
||||
pub created_at: chrono::DateTime<chrono::Utc>,
|
||||
}
|
||||
@@ -67,12 +65,12 @@ pub struct MissedDraft {
|
||||
pub enum DraftSyncStatus {
|
||||
Saved {
|
||||
path: String,
|
||||
typ: DraftType,
|
||||
typ: String,
|
||||
created_at: chrono::DateTime<chrono::Utc>,
|
||||
},
|
||||
Rejected {
|
||||
path: String,
|
||||
typ: DraftType,
|
||||
typ: String,
|
||||
/// Current server copy at conflict-detection time.
|
||||
server_value: sqlx::types::Json<Box<serde_json::value::RawValue>>,
|
||||
server_created_at: chrono::DateTime<chrono::Utc>,
|
||||
@@ -95,7 +93,7 @@ async fn sync_drafts(
|
||||
Path(w_id): Path<String>,
|
||||
Json(req): Json<SyncDraftsRequest>,
|
||||
) -> Result<Json<SyncDraftsResponse>> {
|
||||
let username = &authed.username;
|
||||
let email = &authed.email;
|
||||
let current_timestamp = sqlx::query_scalar!("SELECT now()")
|
||||
.fetch_one(&db)
|
||||
.await?
|
||||
@@ -104,13 +102,13 @@ async fn sync_drafts(
|
||||
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
|
||||
r#"SELECT path, typ, value as "value!: sqlx::types::Json<Box<serde_json::value::RawValue>>", created_at
|
||||
FROM draft
|
||||
WHERE workspace_id = $1
|
||||
AND username = $2
|
||||
AND email = $2
|
||||
AND created_at > $3"#,
|
||||
&w_id,
|
||||
username,
|
||||
email,
|
||||
last_sync,
|
||||
)
|
||||
.fetch_all(&db)
|
||||
@@ -119,12 +117,12 @@ async fn sync_drafts(
|
||||
// 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
|
||||
r#"SELECT path, typ, value as "value!: sqlx::types::Json<Box<serde_json::value::RawValue>>", created_at
|
||||
FROM draft
|
||||
WHERE workspace_id = $1
|
||||
AND username = $2"#,
|
||||
AND email = $2"#,
|
||||
&w_id,
|
||||
username,
|
||||
email,
|
||||
)
|
||||
.fetch_all(&db)
|
||||
.await?
|
||||
@@ -139,14 +137,14 @@ async fn sync_drafts(
|
||||
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 email = $2
|
||||
AND path = $3
|
||||
AND typ = $4
|
||||
AND created_at > $5"#,
|
||||
&w_id,
|
||||
username,
|
||||
email,
|
||||
incoming.path,
|
||||
incoming.typ as DraftType,
|
||||
incoming.typ,
|
||||
last_sync,
|
||||
)
|
||||
.fetch_optional(&db)
|
||||
@@ -155,7 +153,7 @@ async fn sync_drafts(
|
||||
if let Some(row) = conflict {
|
||||
statuses.push(DraftSyncStatus::Rejected {
|
||||
path: incoming.path.clone(),
|
||||
typ: incoming.typ,
|
||||
typ: incoming.typ.clone(),
|
||||
server_value: row.value,
|
||||
server_created_at: row.created_at,
|
||||
incoming_value: sqlx::types::Json(
|
||||
@@ -171,15 +169,15 @@ async fn sync_drafts(
|
||||
}
|
||||
|
||||
let row = sqlx::query!(
|
||||
r#"INSERT INTO draft (workspace_id, username, path, typ, value, created_at)
|
||||
r#"INSERT INTO draft (workspace_id, email, 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
|
||||
ON CONFLICT (workspace_id, path, typ, email) WHERE email IS NOT NULL
|
||||
DO UPDATE SET value = EXCLUDED.value, created_at = now()
|
||||
RETURNING created_at"#,
|
||||
&w_id,
|
||||
username,
|
||||
email,
|
||||
incoming.path,
|
||||
incoming.typ as DraftType,
|
||||
incoming.typ,
|
||||
serde_json::to_string(&incoming.value).unwrap(),
|
||||
)
|
||||
.fetch_one(&db)
|
||||
@@ -187,7 +185,7 @@ async fn sync_drafts(
|
||||
|
||||
statuses.push(DraftSyncStatus::Saved {
|
||||
path: incoming.path.clone(),
|
||||
typ: incoming.typ,
|
||||
typ: incoming.typ.clone(),
|
||||
created_at: row.created_at,
|
||||
});
|
||||
}
|
||||
@@ -202,29 +200,79 @@ async fn sync_drafts(
|
||||
#[derive(Serialize, Debug)]
|
||||
pub struct UserWithDraft {
|
||||
/// `None` represents a legacy workspace-level draft (no owner).
|
||||
pub username: Option<String>,
|
||||
pub email: Option<String>,
|
||||
pub created_at: chrono::DateTime<chrono::Utc>,
|
||||
}
|
||||
|
||||
async fn list_users_with_draft_on_path(
|
||||
_authed: ApiAuthed,
|
||||
authed: ApiAuthed,
|
||||
Extension(db): Extension<DB>,
|
||||
Path((w_id, kind, path)): Path<(String, DraftType, windmill_common::utils::StripPath)>,
|
||||
Extension(user_db): Extension<UserDB>,
|
||||
Path((w_id, kind, path)): Path<(String, String, windmill_common::utils::StripPath)>,
|
||||
) -> Result<Json<Vec<UserWithDraft>>> {
|
||||
let path = path.to_path();
|
||||
require_access_to_path(&authed, &user_db, &w_id, &kind, path).await?;
|
||||
|
||||
let rows = sqlx::query_as!(
|
||||
UserWithDraft,
|
||||
r#"SELECT username, created_at
|
||||
r#"SELECT email, created_at
|
||||
FROM draft
|
||||
WHERE workspace_id = $1
|
||||
AND path = $2
|
||||
AND typ = $3
|
||||
ORDER BY username NULLS LAST"#,
|
||||
ORDER BY email NULLS LAST"#,
|
||||
&w_id,
|
||||
path.to_path(),
|
||||
kind as DraftType,
|
||||
path,
|
||||
kind,
|
||||
)
|
||||
.fetch_all(&db)
|
||||
.await?;
|
||||
|
||||
Ok(Json(rows))
|
||||
}
|
||||
|
||||
/// Resolves to `Ok(())` only if `authed` can read the underlying item at
|
||||
/// `path`. Implemented by issuing a `SELECT 1` through `user_db` (which
|
||||
/// applies row-level extra_perms via `set_session_user`), so a user who
|
||||
/// can't see the script/flow/app gets back zero rows. We don't leak the
|
||||
/// set of paths that exist either way — both "not readable" and "doesn't
|
||||
/// exist" return the same 404.
|
||||
async fn require_access_to_path(
|
||||
authed: &ApiAuthed,
|
||||
user_db: &UserDB,
|
||||
w_id: &str,
|
||||
kind: &str,
|
||||
path: &str,
|
||||
) -> Result<()> {
|
||||
if authed.is_admin {
|
||||
return Ok(());
|
||||
}
|
||||
let table = match kind {
|
||||
"script" => "script",
|
||||
"flow" => "flow",
|
||||
"app" => "app",
|
||||
// Item kinds without a backing path-permission table (raw_app,
|
||||
// resource, variable, trigger_*, ...) fall back to the user's own
|
||||
// `u/{authed.username}/...` path namespace.
|
||||
_ => {
|
||||
let prefix = format!("u/{}/", authed.username);
|
||||
if path.starts_with(&prefix) {
|
||||
return Ok(());
|
||||
}
|
||||
return Err(Error::NotFound(format!("no draft visible at {path}")));
|
||||
}
|
||||
};
|
||||
let mut tx = user_db.clone().begin(authed).await?;
|
||||
let query =
|
||||
format!("SELECT 1 AS exists FROM {table} WHERE path = $1 AND workspace_id = $2 LIMIT 1");
|
||||
let row = sqlx::query_scalar::<_, i32>(&query)
|
||||
.bind(path)
|
||||
.bind(w_id)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
if row.is_none() {
|
||||
return Err(Error::NotFound(format!("no {kind} visible at {path}")));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -27,7 +27,7 @@
|
||||
try {
|
||||
await syncDrafts({
|
||||
workspace: conflict.workspace,
|
||||
user: conflict.user,
|
||||
email: conflict.email,
|
||||
drafts: [
|
||||
{
|
||||
itemKind: conflict.itemKind,
|
||||
|
||||
@@ -3,7 +3,7 @@ import { onDestroy, untrack } from 'svelte'
|
||||
import { deepEqual } from 'fast-equals'
|
||||
import { userStore, workspaceStore } from './stores'
|
||||
import { useLocalStorageValue } from './svelte5Utils.svelte'
|
||||
import { UserDraftDbSyncer, isSyncableKind } from './userDraftDbSyncer.svelte'
|
||||
import { UserDraftDbSyncer } from './userDraftDbSyncer.svelte'
|
||||
import { UserDraftConflictStore } from './userDraftConflictStore.svelte'
|
||||
|
||||
export const USER_DRAFT_ITEM_KINDS = [
|
||||
@@ -351,9 +351,9 @@ export function localDraftDiffers<V>(
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist a draft to localStorage and (when the kind has a backend `draft`
|
||||
* row) push it through the DbSyncer. Internal `_skipSync` opts pass-through
|
||||
* lets the syncer write missed drafts back to LS without re-syncing them.
|
||||
* Persist a draft via the DbSyncer queue. Internal `_skipSync` opts
|
||||
* pass-through lets the syncer write missed drafts back to LS without
|
||||
* re-syncing them.
|
||||
*/
|
||||
function pushToSyncer(
|
||||
itemKind: UserDraftItemKind,
|
||||
@@ -361,12 +361,11 @@ function pushToSyncer(
|
||||
value: unknown,
|
||||
workspace: string
|
||||
): void {
|
||||
if (!isSyncableKind(itemKind)) return
|
||||
const user = get(userStore)?.username
|
||||
if (!user) return
|
||||
const email = get(userStore)?.email
|
||||
if (!email) return
|
||||
UserDraftDbSyncer.pushDrafts({
|
||||
workspace,
|
||||
user,
|
||||
email,
|
||||
drafts: [{ itemKind, path, value }],
|
||||
onMissedDrafts: (drafts) => {
|
||||
for (const d of drafts) {
|
||||
@@ -380,8 +379,8 @@ function pushToSyncer(
|
||||
UserDraftConflictStore.enqueue(
|
||||
rejected.map((r) => ({
|
||||
workspace,
|
||||
user,
|
||||
itemKind: r.typ,
|
||||
email,
|
||||
itemKind: r.typ as UserDraftItemKind,
|
||||
rejected: r
|
||||
}))
|
||||
)
|
||||
|
||||
@@ -6,12 +6,13 @@
|
||||
* conflict; the single `UserDraftConflictModal` mounted in the root layout
|
||||
* consumes them one at a time so the user can decide per-conflict.
|
||||
*/
|
||||
import type { RejectedDraft, SyncableItemKind } from './userDraftDbSyncer.svelte'
|
||||
import type { UserDraftItemKind } from './userDraft.svelte'
|
||||
import type { RejectedDraft } from './userDraftDbSyncer.svelte'
|
||||
|
||||
export type ConflictEntry = {
|
||||
workspace: string
|
||||
user: string
|
||||
itemKind: SyncableItemKind
|
||||
email: string
|
||||
itemKind: UserDraftItemKind
|
||||
rejected: RejectedDraft
|
||||
}
|
||||
|
||||
|
||||
@@ -17,24 +17,11 @@ const LAST_SYNC_KEY = 'userdraft/lastSync'
|
||||
const DEBOUNCE_MS = 2_000
|
||||
const MAX_DEBOUNCE_MS = 10_000
|
||||
|
||||
export type SyncableItemKind = 'script' | 'flow' | 'app'
|
||||
|
||||
/**
|
||||
* UserDraft item kinds that have a matching backend `draft.typ` enum value.
|
||||
* Other kinds (resources, variables, individual triggers...) only exist in
|
||||
* localStorage for now and are filtered out of every push.
|
||||
*/
|
||||
const SYNCABLE_KINDS: ReadonlySet<UserDraftItemKind> = new Set(['script', 'flow', 'app'])
|
||||
|
||||
export function isSyncableKind(kind: UserDraftItemKind): kind is SyncableItemKind {
|
||||
return SYNCABLE_KINDS.has(kind)
|
||||
}
|
||||
|
||||
export type MissedDraft = SyncDraftsResponse['missed_drafts'][number]
|
||||
export type RejectedDraft = Extract<SyncDraftsResponse['statuses'][number], { status: 'rejected' }>
|
||||
|
||||
export type PendingDraft<V = unknown> = {
|
||||
itemKind: SyncableItemKind
|
||||
itemKind: UserDraftItemKind
|
||||
path: string
|
||||
value: V
|
||||
/**
|
||||
@@ -50,21 +37,26 @@ export type RejectedDraftsCallback = (rejected: RejectedDraft[]) => void
|
||||
|
||||
export type SyncOptions<V = unknown> = {
|
||||
workspace: string
|
||||
user: string
|
||||
email: string
|
||||
drafts: PendingDraft<V>[]
|
||||
onMissedDrafts?: MissedDraftCallback
|
||||
onDraftsRejected?: RejectedDraftsCallback
|
||||
}
|
||||
|
||||
type QueueKey = string
|
||||
function queueKey(workspace: string, user: string, kind: SyncableItemKind, path: string): QueueKey {
|
||||
return `${workspace}|${user}|${kind}|${path}`
|
||||
function queueKey(
|
||||
workspace: string,
|
||||
email: string,
|
||||
kind: UserDraftItemKind,
|
||||
path: string
|
||||
): QueueKey {
|
||||
return `${workspace}|${email}|${kind}|${path}`
|
||||
}
|
||||
|
||||
type QueuedEntry = {
|
||||
workspace: string
|
||||
user: string
|
||||
itemKind: SyncableItemKind
|
||||
email: string
|
||||
itemKind: UserDraftItemKind
|
||||
path: string
|
||||
value: unknown
|
||||
force: boolean
|
||||
@@ -113,13 +105,13 @@ async function flushQueue(): Promise<void> {
|
||||
const entries = Array.from(queue.values())
|
||||
queue.clear()
|
||||
|
||||
// Group entries by (workspace, user) — every sync call is scoped to a
|
||||
// single user, so we issue one request per distinct group. In practice
|
||||
// the queue is dominated by the active session's workspace+user, so
|
||||
// Group entries by (workspace, email) — every sync call is scoped to a
|
||||
// single email, so we issue one request per distinct group. In practice
|
||||
// the queue is dominated by the active session's workspace+email, so
|
||||
// there's almost always exactly one group.
|
||||
const groups = new Map<string, QueuedEntry[]>()
|
||||
for (const entry of entries) {
|
||||
const key = `${entry.workspace}|${entry.user}`
|
||||
const key = `${entry.workspace}|${entry.email}`
|
||||
const list = groups.get(key)
|
||||
if (list) list.push(entry)
|
||||
else groups.set(key, [entry])
|
||||
@@ -141,7 +133,7 @@ async function runSync(workspace: string, group: QueuedEntry[]): Promise<void> {
|
||||
}))
|
||||
await syncDrafts({
|
||||
workspace,
|
||||
user: group[0].user,
|
||||
email: group[0].email,
|
||||
drafts,
|
||||
onMissedDrafts,
|
||||
onDraftsRejected
|
||||
@@ -194,17 +186,17 @@ export const UserDraftDbSyncer = {
|
||||
|
||||
/**
|
||||
* Enqueue drafts for a batched sync. Repeated pushes for the same
|
||||
* (workspace, user, itemKind, path) coalesce — only the latest value /
|
||||
* (workspace, email, itemKind, path) coalesce — only the latest value /
|
||||
* callbacks survive. Returns the same Promise as the eventual
|
||||
* `syncDrafts` so callers can `await` flush completion.
|
||||
*/
|
||||
pushDrafts<V = unknown>(opts: SyncOptions<V>): void {
|
||||
const ws = opts.workspace
|
||||
const user = opts.user
|
||||
const email = opts.email
|
||||
for (const d of opts.drafts) {
|
||||
queue.set(queueKey(ws, user, d.itemKind, d.path), {
|
||||
queue.set(queueKey(ws, email, d.itemKind, d.path), {
|
||||
workspace: ws,
|
||||
user,
|
||||
email,
|
||||
itemKind: d.itemKind,
|
||||
path: d.path,
|
||||
value: d.value,
|
||||
|
||||
Reference in New Issue
Block a user