feat: let native triggers act through per-user or shared connections

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
hugocasa
2026-10-15 18:04:02 +02:00
co-authored by Claude Opus 5.5
parent 245d03b254
commit 74c13f83f6
59 changed files with 1914 additions and 984 deletions
+31 -15
View File
@@ -84,7 +84,7 @@ pub trait External: Send + Sync + 'static {
async fn prepare_webhook(&self, db, w_id, headers, body, script_path, is_flow) -> Result<PushArgsOwned>;
fn service_config_from_create_response(&self, data, resp) -> Option<serde_json::Value>;
fn additional_routes(&self) -> axum::Router;
async fn http_client_request<T, B>(&self, url, method, workspace_id, tx, db, headers, body) -> Result<T>;
async fn http_client_request<T, B>(&self, url, method, workspace_id, connection_path, db, headers, body) -> Result<T>;
}
```
@@ -115,28 +115,44 @@ Use this when the external_id is known before the create call (e.g., Google gene
Use this when the external_id is assigned by the remote service and the webhook URL needs to be corrected after creation.
### OAuth Token Storage (Three-Table Pattern)
### OAuth App and Connections
OAuth tokens are stored across three tables, NOT in `workspace_integrations.oauth_data` directly:
The OAuth **app** (client) is configured per workspace and service; **connections** (connected
accounts) are many per service. Each trigger acts through one connection, named by
`native_trigger.connection_path`:
| Table | What's Stored |
|-------|---------------|
| `workspace_integrations` | `oauth_data` JSON with `base_url`, `client_id`, `client_secret`, `instance_shared` flag; `resource_path` pointing to the variable |
| `variable` | Encrypted `access_token` (at the path stored in `resource_path`), linked to `account` via `account` column |
| `account` | `refresh_token`, keyed by `workspace_id` + `client` (service name) + `is_workspace_integration = true` |
| `workspace_integrations` | `oauth_data` JSON with `base_url`, `client_id`, `client_secret`, `instance_shared` flag. One row per (workspace, service); it must exist for every connection, since token refresh reads the client from it |
| `variable` | Encrypted `access_token` at the connection path, linked to `account` via its `account` column |
| `account` | `refresh_token`, with `client` = service name and `is_workspace_integration = true` |
| `resource` | The connection's resource at the same path, `{ "token": "$var:<path>" }` |
The `decrypt_oauth_data()` function in `lib.rs` assembles these into a unified struct:
A user may use a connection when they can read its variable (checked with a user transaction,
so the variable's row-level security decides): `resolve_usable_connection` /
`list_usable_connections` in `lib.rs`. Background work (sync, renewal, rename re-registration,
webhook handling) uses the stored `connection_path` directly.
`decrypt_oauth_data(db, w_id, service, connection_path)` assembles these into one struct, and
every service's `OAuthData` carries `connection_path` so `http_client_request` can refresh and
write back the right connection:
```rust
pub struct OAuthConfig {
pub base_url: String,
pub access_token: String, // decrypted from variable
pub refresh_token: Option<String>, // from account table
pub access_token: String, // decrypted from the connection's variable
pub refresh_token: Option<String>, // from its account
pub client_id: String, // from oauth_data or instance settings
pub client_secret: String, // from oauth_data or instance settings
pub connection_path: String,
}
```
Instance-level sharing: when `oauth_data.instance_shared == true`, `client_id` and `client_secret` are read from global settings instead of workspace_integrations.
A refreshed token is written back by connection path and account id, never by service: providers
like Nextcloud rotate refresh tokens, so writing one onto another account breaks its next refresh.
Instance-level sharing: when the workspace has no OAuth app of its own and the instance admin
shares one (`share_with_workspaces` in the `oauths` setting; not available for Nextcloud), the
connect flow uses it and records `{"instance_shared": true}` in `workspace_integrations`.
### URL Resolution
@@ -311,7 +327,7 @@ impl External for NewService {
});
let response: CreateTriggerResponse = self
.http_client_request(&url, Method::POST, w_id, tx, db, None, Some(&payload))
.http_client_request(&url, Method::POST, w_id, &oauth_data.connection_path, db, None, Some(&payload))
.await?;
Ok(response)
@@ -343,7 +359,7 @@ impl External for NewService {
});
let _: serde_json::Value = self
.http_client_request(&url, Method::PUT, w_id, tx, db, None, Some(&payload))
.http_client_request(&url, Method::PUT, w_id, &oauth_data.connection_path, db, None, Some(&payload))
.await?;
// Fetch back the updated state to get the resolved config
@@ -361,7 +377,7 @@ impl External for NewService {
tx: &mut PgConnection,
) -> Result<Self::TriggerData> {
let url = format!("{}/api/webhooks/{}", oauth_data.base_url, external_id);
self.http_client_request::<_, ()>(&url, Method::GET, w_id, tx, db, None, None).await
self.http_client_request::<_, ()>(&url, Method::GET, w_id, &oauth_data.connection_path, db, None, None).await
}
async fn delete(
@@ -374,7 +390,7 @@ impl External for NewService {
) -> Result<()> {
let url = format!("{}/api/webhooks/{}", oauth_data.base_url, external_id);
let _: serde_json::Value = self
.http_client_request::<_, ()>(&url, Method::DELETE, w_id, tx, db, None, None)
.http_client_request::<_, ()>(&url, Method::DELETE, w_id, &oauth_data.connection_path, db, None, None)
.await
.or_else(|e| match &e {
Error::InternalErr(msg) if msg.contains("404") => Ok(serde_json::Value::Null),
@@ -749,7 +765,7 @@ pub async fn prepare_native_trigger_args(service_name, db, w_id, headers, body)
When `workspace_integrations.oauth_data.instance_shared == true`, `decrypt_oauth_data()` reads `client_id` and `client_secret` from instance-level global settings instead of workspace-level. This allows admins to share OAuth app credentials across workspaces.
The frontend handles this via the `generate_instance_connect_url` endpoint in `workspace_integrations.rs`.
`generate_connect_url` picks the client (`resolve_oauth_client`): the workspace's own app when configured, the instance's otherwise. A GitHub OAuth app only redirects below its single registered callback, so connections made with the instance GitHub app land on `/oauth/callback/github/native_trigger` instead of the workspace settings page.
---
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM variable WHERE workspace_id = $1 AND account = ANY($2) RETURNING path",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "path",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Int4Array"
]
},
"nullable": [
false
]
},
"hash": "00b32f6cd59154d9e529c8577072f000ae6cf37928508a82f51d3b3a325f3d1a"
}
@@ -0,0 +1,24 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE variable SET value = $1 WHERE workspace_id = $2 AND path = $3 RETURNING account",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "account",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": [
true
]
},
"hash": "17a29b1db3b9aea8ce4158ed1e475a4e26c766a15b0ec209504a24b2460575be"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT external_id, webhook_token_hash FROM native_trigger WHERE workspace_id = $1 AND service_name = $2",
"query": "SELECT external_id, connection_path FROM native_trigger\n WHERE workspace_id = $1 AND service_name = $2 AND ($3::text IS NULL OR connection_path = $3)",
"describe": {
"columns": [
{
@@ -10,7 +10,7 @@
},
{
"ordinal": 1,
"name": "webhook_token_hash",
"name": "connection_path",
"type_info": "Varchar"
}
],
@@ -28,13 +28,14 @@
]
}
}
}
},
"Text"
]
},
"nullable": [
false,
false
true
]
},
"hash": "6c75c89fb215c646f54f2036c40a86a4f4ea8880cc5d0b511aa70b6fd50072c5"
"hash": "1ecd50aa13f5d4babdfaaba6ff56940ac9e6d7a28a27e576b5dda4d7e709ad1a"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO workspace_integrations (\n workspace_id,\n service_name,\n oauth_data,\n resource_path,\n created_by,\n created_at,\n updated_at\n ) VALUES (\n $1, $2, $3, $4, $5, now(), now()\n )\n ON CONFLICT (workspace_id, service_name)\n DO UPDATE SET\n oauth_data = $3,\n resource_path = $4,\n updated_at = now()\n ",
"query": "\n INSERT INTO workspace_integrations (\n workspace_id,\n service_name,\n oauth_data,\n created_by,\n created_at,\n updated_at\n ) VALUES (\n $1, $2, $3, $4, now(), now()\n )\n ON CONFLICT (workspace_id, service_name)\n DO UPDATE SET\n oauth_data = $3,\n updated_at = now()\n ",
"describe": {
"columns": [],
"parameters": {
@@ -19,11 +19,10 @@
}
},
"Jsonb",
"Text",
"Varchar"
]
},
"nullable": []
},
"hash": "a588f4caa014008b50eccd09122b09f8a098e58791893bacaaf2ff67a30c031c"
"hash": "2602402bedcdbc45cdc0d64a76ce075c6ae51404037b0a7d4d33faf6a7d6a6d8"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n oauth_data as \"oauth_data: sqlx::types::Json<WorkspaceOAuthConfig>\",\n service_name as \"service_name!: ServiceName\",\n resource_path\n FROM\n workspace_integrations\n WHERE\n workspace_id = $1\n ",
"query": "\n SELECT\n oauth_data as \"oauth_data: sqlx::types::Json<WorkspaceOAuthConfig>\",\n service_name as \"service_name!: ServiceName\"\n FROM\n workspace_integrations\n WHERE\n workspace_id = $1\n ",
"describe": {
"columns": [
{
@@ -23,11 +23,6 @@
}
}
}
},
{
"ordinal": 2,
"name": "resource_path",
"type_info": "Text"
}
],
"parameters": {
@@ -37,9 +32,8 @@
},
"nullable": [
true,
false,
true
false
]
},
"hash": "0010ef26da16facd1c2c832601ac687c4c27de46a90f45496b8446af1a9d0578"
"hash": "30b1021260b5ec83a04be1f3dbea08cc81ee85cf98696d9e29a34557587eaf86"
}
@@ -0,0 +1,28 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO workspace_integrations (workspace_id, service_name, oauth_data, created_by)\n VALUES ($1, $2, $3, $4)\n ON CONFLICT (workspace_id, service_name) DO UPDATE SET oauth_data = EXCLUDED.oauth_data\n WHERE workspace_integrations.oauth_data IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
{
"Custom": {
"name": "native_trigger_service",
"kind": {
"Enum": [
"nextcloud",
"google",
"github"
]
}
}
},
"Jsonb",
"Varchar"
]
},
"nullable": []
},
"hash": "3419911d3acc46540ad818015390d4f366724e0d4040faeccf443ef9ce30665c"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n workspace_id,\n service_name AS \"service_name!: ServiceName\",\n oauth_data,\n resource_path,\n created_at,\n updated_at,\n created_by\n FROM\n workspace_integrations\n WHERE\n workspace_id = $1\n AND service_name = $2\n ",
"query": "\n SELECT\n workspace_id,\n service_name AS \"service_name!: ServiceName\",\n oauth_data,\n created_at,\n updated_at,\n created_by\n FROM\n workspace_integrations\n WHERE\n workspace_id = $1\n AND service_name = $2\n ",
"describe": {
"columns": [
{
@@ -31,21 +31,16 @@
},
{
"ordinal": 3,
"name": "resource_path",
"type_info": "Text"
},
{
"ordinal": 4,
"name": "created_at",
"type_info": "Timestamptz"
},
{
"ordinal": 5,
"ordinal": 4,
"name": "updated_at",
"type_info": "Timestamptz"
},
{
"ordinal": 6,
"ordinal": 5,
"name": "created_by",
"type_info": "Varchar"
}
@@ -71,11 +66,10 @@
false,
false,
true,
true,
false,
false,
false
]
},
"hash": "7443ba1f922e190bb9a0ca313847f8d27c35a6ee3aff20157d6285e73aa923ef"
"hash": "3e5bdc2e071fc2f1e3c7971736272f20bb5a0aa921a614bd02898d3f162660c2"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE workspace_integrations SET resource_path = REGEXP_REPLACE(resource_path, 'u/' || $2 || '/(.*)', $1 || '/\\1') WHERE resource_path LIKE ('u/' || $2 || '/%') AND workspace_id = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "3f77282b7e0611ef7e6e8de654381208165ba7414fc592ca8dfffc9838ac715e"
}
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM resource WHERE workspace_id = $1 AND path = ANY($2)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"TextArray"
]
},
"nullable": []
},
"hash": "4d48a34016cc642fe3b3b24220615a59165009048a7c93942ee2ecd78524fb5a"
}
@@ -0,0 +1,30 @@
{
"db_name": "PostgreSQL",
"query": "SELECT v.value, a.refresh_token\n FROM variable v\n JOIN account a ON a.workspace_id = v.workspace_id AND a.id = v.account\n WHERE v.workspace_id = $1 AND v.path = $2\n AND a.client = $3 AND a.is_workspace_integration = true",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "value",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "refresh_token",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": [
false,
false
]
},
"hash": "5cb209fe91eeccdc4c208db7655b6349376f09dad74380302cb815fc2626c69c"
}
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE account SET\n refresh_token = COALESCE($1, refresh_token),\n expires_at = now() + interval '1 hour',\n refresh_error = NULL\n WHERE workspace_id = $2 AND id = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text",
"Int4"
]
},
"nullable": []
},
"hash": "61c4b826e90badc1307eadfe0c117efc6c8c87ba5542bce9c5f85b1da07e9a4b"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM native_trigger WHERE workspace_id = $1 AND service_name = $2 RETURNING webhook_token_hash",
"query": "DELETE FROM native_trigger\n WHERE workspace_id = $1 AND service_name = $2 AND ($3::text IS NULL OR connection_path = $3)\n RETURNING webhook_token_hash",
"describe": {
"columns": [
{
@@ -23,12 +23,13 @@
]
}
}
}
},
"Text"
]
},
"nullable": [
false
]
},
"hash": "b9686c115e83ad0725503e2c076960914f825eaa2b44281a4e147cea6e886e7a"
"hash": "6e1231d0db1d3736b05a41aa0a3f125cb2f383a53645d7c48cbd0478c3364663"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE account SET\n refresh_token = $1,\n expires_at = now() + interval '1 hour',\n refresh_error = NULL\n WHERE workspace_id = $2 AND client = $3 AND is_workspace_integration = true",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "6e83478011a8f65bf294ad7886139369e386f6a552c6626be48ecaa0e5ab78a7"
}
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM account WHERE workspace_id = $1 AND id = $2 AND is_workspace_integration = true",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Int4"
]
},
"nullable": []
},
"hash": "7316c6eef49c12d959329ee611f69ed368469c40f48bf25243fd96002ac4c7c5"
}
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE account SET\n expires_at = now() + interval '1 hour',\n refresh_error = NULL\n WHERE workspace_id = $1 AND client = $2 AND is_workspace_integration = true",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": []
},
"hash": "7b1d29170c9c4ad4e3a15c4e8acbeb6769dc6fc97269beee607a5625a495a121"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE workspace_integrations SET resource_path = REGEXP_REPLACE(resource_path, 'u/' || $2 || '/(.*)', 'u/' || $1 || '/\\1') WHERE resource_path LIKE ('u/' || $2 || '/%') AND workspace_id = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "89791ad1a0862b6475ebfdeb54b0101e124fcf9d12e93d84b44457c72c7604a5"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO native_trigger (\n external_id,\n workspace_id,\n service_name,\n script_path,\n is_flow,\n webhook_token_hash,\n service_config,\n summary,\n enabled\n ) VALUES (\n $1, $2, $3, $4, $5, $6, $7, $8, $9\n )\n ON CONFLICT (external_id, workspace_id, service_name)\n DO UPDATE SET script_path = $4, is_flow = $5, webhook_token_hash = $6, service_config = $7, summary = $8, error = NULL, updated_at = NOW()\n ",
"query": "\n INSERT INTO native_trigger (\n external_id,\n workspace_id,\n service_name,\n script_path,\n is_flow,\n webhook_token_hash,\n service_config,\n summary,\n enabled,\n connection_path\n ) VALUES (\n $1, $2, $3, $4, $5, $6, $7, $8, $9, $10\n )\n ON CONFLICT (external_id, workspace_id, service_name)\n DO UPDATE SET script_path = $4, is_flow = $5, webhook_token_hash = $6, service_config = $7, summary = $8, connection_path = $10, error = NULL, updated_at = NOW()\n ",
"describe": {
"columns": [],
"parameters": {
@@ -24,10 +24,11 @@
"Varchar",
"Jsonb",
"Varchar",
"Bool"
"Bool",
"Varchar"
]
},
"nullable": []
},
"hash": "64dce306efc0d9989542dba5bf003dc94b9337a8d3b69805308258d1fa145e63"
"hash": "8ec380ed48a8366e8869c6d1aa06ba774a6ded50ab047d0b0302c7430cee5b70"
}
@@ -1,34 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT resource_path FROM workspace_integrations WHERE workspace_id = $1 AND service_name = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "resource_path",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text",
{
"Custom": {
"name": "native_trigger_service",
"kind": {
"Enum": [
"nextcloud",
"google",
"github"
]
}
}
}
]
},
"nullable": [
true
]
},
"hash": "993e514229a1d508a44ef07b4399f9697cdefaa1f45ca665f72bc6fcf2797c7e"
}
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM variable WHERE workspace_id = $1 AND path = $2 RETURNING account",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "account",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
true
]
},
"hash": "9d7416a5458bb3d4a145cc1d2d35fa8e032a4ccdd23d7ed9a6c58d791b7bbd08"
}
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE native_trigger SET connection_path = REGEXP_REPLACE(connection_path, 'u/' || $2 || '/(.*)', $1 || '/\\1') WHERE connection_path LIKE ('u/' || $2 || '/%') AND workspace_id = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "c17ac5cf890831ede0811263a0f35e8cb2fe4df224e2ab633a1e69c4bcfa6cd1"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n nt.external_id,\n nt.workspace_id,\n nt.service_name AS \"service_name!: ServiceName\",\n nt.script_path,\n nt.is_flow,\n nt.webhook_token_hash,\n nt.service_config,\n nt.error,\n nt.created_at,\n nt.updated_at,\n nt.summary,\n nt.enabled\n FROM\n native_trigger nt\n WHERE\n nt.workspace_id = $1 AND\n nt.service_name = $2 AND\n ($5::text IS NULL OR nt.script_path = $5) AND\n ($6::bool IS NULL OR nt.is_flow = $6) AND\n (\n (nt.is_flow = false AND EXISTS (\n SELECT 1 FROM script s\n WHERE s.workspace_id = nt.workspace_id\n AND s.path = nt.script_path\n ))\n OR\n (nt.is_flow = true AND EXISTS (\n SELECT 1 FROM flow f\n WHERE f.workspace_id = nt.workspace_id\n AND f.path = nt.script_path\n ))\n ) AND\n ( $7\n OR nt.script_path = ANY($8)\n OR EXISTS ( SELECT 1 FROM unnest($9::text[]) AS pfx\n WHERE nt.script_path = pfx\n OR left(nt.script_path, length(pfx) + 1) = pfx || '/' ) )\n LIMIT $3\n OFFSET $4\n ",
"query": "\n SELECT\n nt.external_id,\n nt.workspace_id,\n nt.service_name AS \"service_name!: ServiceName\",\n nt.script_path,\n nt.is_flow,\n nt.webhook_token_hash,\n nt.service_config,\n nt.error,\n nt.created_at,\n nt.updated_at,\n nt.summary,\n nt.enabled,\n nt.connection_path\n FROM\n native_trigger nt\n WHERE\n nt.workspace_id = $1 AND\n nt.service_name = $2 AND\n ($5::text IS NULL OR nt.script_path = $5) AND\n ($6::bool IS NULL OR nt.is_flow = $6) AND\n (\n (nt.is_flow = false AND EXISTS (\n SELECT 1 FROM script s\n WHERE s.workspace_id = nt.workspace_id\n AND s.path = nt.script_path\n ))\n OR\n (nt.is_flow = true AND EXISTS (\n SELECT 1 FROM flow f\n WHERE f.workspace_id = nt.workspace_id\n AND f.path = nt.script_path\n ))\n ) AND\n ( $7\n OR nt.script_path = ANY($8)\n OR EXISTS ( SELECT 1 FROM unnest($9::text[]) AS pfx\n WHERE nt.script_path = pfx\n OR left(nt.script_path, length(pfx) + 1) = pfx || '/' ) )\n LIMIT $3\n OFFSET $4\n ",
"describe": {
"columns": [
{
@@ -73,6 +73,11 @@
"ordinal": 11,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 12,
"name": "connection_path",
"type_info": "Varchar"
}
],
"parameters": {
@@ -111,8 +116,9 @@
false,
false,
true,
false
false,
true
]
},
"hash": "f3cd54017710536747b8aca797a8b983774ee2127dceb790181ed91b987f44ff"
"hash": "c5a7f56bed61174904d3bb1b7be704a4d1a3a4da8c995319f3cc3ce96d8ef5cc"
}
@@ -0,0 +1,29 @@
{
"db_name": "PostgreSQL",
"query": "SELECT v.path, r.created_by AS \"owner?\"\n FROM variable v\n JOIN account a ON a.workspace_id = v.workspace_id AND a.id = v.account\n LEFT JOIN resource r ON r.workspace_id = v.workspace_id AND r.path = v.path\n WHERE v.workspace_id = $1 AND a.client = $2 AND a.is_workspace_integration = true\n ORDER BY v.path",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "owner?",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false,
true
]
},
"hash": "c93af685f45ed641a9f2bb3beed5298eef6f75ec3852e0f007b98f71bfe56d53"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n external_id,\n workspace_id,\n service_name AS \"service_name!: ServiceName\",\n script_path,\n is_flow,\n webhook_token_hash,\n service_config,\n error,\n created_at,\n updated_at,\n summary,\n enabled\n FROM\n native_trigger\n WHERE\n workspace_id = $1\n AND service_name = $2\n AND script_path = $3\n AND is_flow = $4\n LIMIT 1\n ",
"query": "\n SELECT\n external_id,\n workspace_id,\n service_name AS \"service_name!: ServiceName\",\n script_path,\n is_flow,\n webhook_token_hash,\n service_config,\n error,\n created_at,\n updated_at,\n summary,\n enabled,\n connection_path\n FROM\n native_trigger\n WHERE\n workspace_id = $1\n AND service_name = $2\n AND script_path = $3\n AND is_flow = $4\n LIMIT 1\n ",
"describe": {
"columns": [
{
@@ -73,6 +73,11 @@
"ordinal": 11,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 12,
"name": "connection_path",
"type_info": "Varchar"
}
],
"parameters": {
@@ -106,8 +111,9 @@
false,
false,
true,
false
false,
true
]
},
"hash": "0987f164fe3d64bf0a6a4e9699c2d1f339670da3ed1c08203d54d45798d022dd"
"hash": "d976c78db15dac2c70edcea6d7d9ceed91c54f592d0e1f5139d5e111241ca451"
}
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE native_trigger SET connection_path = REGEXP_REPLACE(connection_path, 'u/' || $2 || '/(.*)', 'u/' || $1 || '/\\1') WHERE connection_path LIKE ('u/' || $2 || '/%') AND workspace_id = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "daaaa027338cd77d6400658667351a4d3bbe6d1309cd8205e853097c9f81a471"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n external_id,\n workspace_id,\n service_name AS \"service_name!: ServiceName\",\n script_path,\n is_flow,\n webhook_token_hash,\n service_config,\n error,\n created_at,\n updated_at,\n summary,\n enabled\n FROM\n native_trigger\n WHERE\n workspace_id = $1\n AND service_name = $2\n AND external_id = $3\n ",
"query": "\n SELECT\n external_id,\n workspace_id,\n service_name AS \"service_name!: ServiceName\",\n script_path,\n is_flow,\n webhook_token_hash,\n service_config,\n error,\n created_at,\n updated_at,\n summary,\n enabled,\n connection_path\n FROM\n native_trigger\n WHERE\n workspace_id = $1\n AND service_name = $2\n AND external_id = $3\n ",
"describe": {
"columns": [
{
@@ -73,6 +73,11 @@
"ordinal": 11,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 12,
"name": "connection_path",
"type_info": "Varchar"
}
],
"parameters": {
@@ -105,8 +110,9 @@
false,
false,
true,
false
false,
true
]
},
"hash": "c7766afaea3e187824698cae2b090af9f49ded98cd1824f1ac91cc5dbe709bc1"
"hash": "efae221cbce1c9d4b96f9990a012b3efce1740b7037cac9983529bc052cf2dd9"
}
@@ -0,0 +1 @@
ALTER TABLE native_trigger DROP COLUMN IF EXISTS connection_path;
@@ -0,0 +1,10 @@
-- Each native trigger names the connection (OAuth resource) it acts through, instead of every
-- trigger of a service sharing the single workspace connection.
ALTER TABLE native_trigger ADD COLUMN connection_path VARCHAR(255);
UPDATE native_trigger nt
SET connection_path = wi.resource_path
FROM workspace_integrations wi
WHERE wi.workspace_id = nt.workspace_id
AND wi.service_name = nt.service_name
AND wi.resource_path IS NOT NULL;
@@ -26,8 +26,8 @@ use windmill_native_triggers::{
native_trigger_is_enabled,
nextcloud::NextCloud,
require_native_integration_use, set_native_trigger_enabled, store_native_trigger,
store_workspace_integration, update_native_trigger, External, ExternalReadFailure,
HttpRequestError, NativeTriggerConfig, OAuthConfig, ServiceName,
store_workspace_integration, update_native_trigger, update_oauth_token_resource, External,
ExternalReadFailure, HttpRequestError, NativeTriggerConfig, OAuthConfig, ServiceName,
};
// ============================================================================
@@ -133,7 +133,6 @@ async fn setup_oauth_integration(
"test-workspace",
service_name,
oauth_data,
Some(resource_path),
)
.await?;
tx.commit().await?;
@@ -149,73 +148,70 @@ fn now_ms() -> i64 {
}
// ============================================================================
// 1. Resource Path Change
// 1. Several connections of one service
// ============================================================================
/// Each connection keeps its own tokens: a refresh written back for one, or disconnecting one,
/// must not touch another. Providers that rotate refresh tokens invalidate the old one, so a
/// refresh token copied onto the wrong account breaks that account's next refresh.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_resource_path_change(db: Pool<Postgres>) -> anyhow::Result<()> {
let path_a = "u/test-user/native_gworkspace";
setup_oauth_integration(
async fn test_connections_are_isolated(db: Pool<Postgres>) -> anyhow::Result<()> {
let path_a = "u/test-user/native_nextcloud";
let path_b = "u/other-user/native_nextcloud";
for (path, token) in [(path_a, "a"), (path_b, "b")] {
setup_oauth_integration(
&db,
ServiceName::Nextcloud,
path,
&format!("access-{token}"),
&format!("refresh-{token}"),
None,
)
.await?;
}
update_oauth_token_resource(
&db,
ServiceName::Google,
path_a,
"token-a",
"refresh-a",
None,
)
.await?;
// Verify decrypt works at path A
let config: OAuthConfig =
decrypt_oauth_data(&db, "test-workspace", ServiceName::Google).await?;
assert_eq!(config.access_token, "token-a");
// Cleanup old path
let mut tx = db.begin().await?;
windmill_native_triggers::workspace_integrations::cleanup_oauth_resource(
&mut *tx,
"test-workspace",
ServiceName::Google,
path_a,
"access-a2",
Some("refresh-a2"),
)
.await;
tx.commit().await?;
// Recreate at path B
let path_b = "u/test-user/native_gworkspace_v2";
setup_oauth_integration(
&db,
ServiceName::Google,
path_b,
"token-b",
"refresh-b",
None,
let a: OAuthConfig =
decrypt_oauth_data(&db, "test-workspace", ServiceName::Nextcloud, Some(path_a)).await?;
let b: OAuthConfig =
decrypt_oauth_data(&db, "test-workspace", ServiceName::Nextcloud, Some(path_b)).await?;
assert_eq!(
(a.access_token.as_str(), a.refresh_token.as_deref()),
("access-a2", Some("refresh-a2"))
);
assert_eq!(
(b.access_token.as_str(), b.refresh_token.as_deref()),
("access-b", Some("refresh-b"))
);
let mut tx = db.begin().await?;
windmill_native_triggers::workspace_integrations::cleanup_connection(
&mut *tx,
"test-workspace",
path_a,
)
.await?;
tx.commit().await?;
// Path A resources should be gone
let var_count: i64 = sqlx::query_scalar!(
"SELECT count(*) FROM variable WHERE workspace_id = 'test-workspace' AND path = $1",
path_a,
assert!(decrypt_oauth_data::<OAuthConfig>(
&db,
"test-workspace",
ServiceName::Nextcloud,
Some(path_a)
)
.fetch_one(&db)
.await?
.unwrap_or(0);
assert_eq!(var_count, 0, "variable at old path should be deleted");
let res_count: i64 = sqlx::query_scalar!(
"SELECT count(*) FROM resource WHERE workspace_id = 'test-workspace' AND path = $1",
path_a,
)
.fetch_one(&db)
.await?
.unwrap_or(0);
assert_eq!(res_count, 0, "resource at old path should be deleted");
// Path B should work
let config: OAuthConfig =
decrypt_oauth_data(&db, "test-workspace", ServiceName::Google).await?;
assert_eq!(config.access_token, "token-b");
assert_eq!(config.refresh_token.as_deref(), Some("refresh-b"));
.await
.is_err());
let b: OAuthConfig =
decrypt_oauth_data(&db, "test-workspace", ServiceName::Nextcloud, Some(path_b)).await?;
assert_eq!(b.refresh_token.as_deref(), Some("refresh-b"));
Ok(())
}
@@ -237,8 +233,13 @@ async fn test_decrypt_workspace_level(db: Pool<Postgres>) -> anyhow::Result<()>
)
.await?;
let config: OAuthConfig =
decrypt_oauth_data(&db, "test-workspace", ServiceName::Google).await?;
let config: OAuthConfig = decrypt_oauth_data(
&db,
"test-workspace",
ServiceName::Google,
Some(resource_path),
)
.await?;
assert_eq!(config.access_token, "ws-access-token");
assert_eq!(config.refresh_token.as_deref(), Some("ws-refresh-token"));
@@ -282,8 +283,13 @@ async fn test_decrypt_instance_level(db: Pool<Postgres>) -> anyhow::Result<()> {
)
.await?;
let config: OAuthConfig =
decrypt_oauth_data(&db, "test-workspace", ServiceName::Google).await?;
let config: OAuthConfig = decrypt_oauth_data(
&db,
"test-workspace",
ServiceName::Google,
Some(resource_path),
)
.await?;
assert_eq!(config.client_id, "instance-client-id");
assert_eq!(config.client_secret, "instance-client-secret");
@@ -306,8 +312,13 @@ async fn test_token_update_persists(db: Pool<Postgres>) -> anyhow::Result<()> {
.await?;
// Verify old tokens
let config: OAuthConfig =
decrypt_oauth_data(&db, "test-workspace", ServiceName::Google).await?;
let config: OAuthConfig = decrypt_oauth_data(
&db,
"test-workspace",
ServiceName::Google,
Some(resource_path),
)
.await?;
assert_eq!(config.access_token, "old-access-token");
// Simulate token refresh: update variable + account
@@ -330,8 +341,13 @@ async fn test_token_update_persists(db: Pool<Postgres>) -> anyhow::Result<()> {
.await?;
// Verify new tokens
let config: OAuthConfig =
decrypt_oauth_data(&db, "test-workspace", ServiceName::Google).await?;
let config: OAuthConfig = decrypt_oauth_data(
&db,
"test-workspace",
ServiceName::Google,
Some(resource_path),
)
.await?;
assert_eq!(config.access_token, "new-access-token");
assert_eq!(config.refresh_token.as_deref(), Some("new-refresh-token"));
@@ -458,6 +474,7 @@ async fn test_delete_integration_full_cascade(db: Pool<Postgres>) -> anyhow::Res
json!({"triggerType": "drive"}),
None,
true,
"u/test-user/native_conn",
)
.await?;
@@ -468,12 +485,12 @@ async fn test_delete_integration_full_cascade(db: Pool<Postgres>) -> anyhow::Res
// Step 2: Cleanup OAuth resources
let mut tx = db.begin().await?;
windmill_native_triggers::workspace_integrations::cleanup_oauth_resource(
windmill_native_triggers::workspace_integrations::cleanup_service_connections(
&mut *tx,
"test-workspace",
ServiceName::Google,
)
.await;
.await?;
tx.commit().await?;
// Step 3: Delete workspace integration
@@ -549,17 +566,18 @@ async fn test_cleanup_preserves_triggers(db: Pool<Postgres>) -> anyhow::Result<(
json!({"triggerType": "drive"}),
None,
true,
"u/test-user/native_conn",
)
.await?;
// Cleanup OAuth only — should NOT remove the trigger
let mut tx = db.begin().await?;
windmill_native_triggers::workspace_integrations::cleanup_oauth_resource(
windmill_native_triggers::workspace_integrations::cleanup_connection(
&mut *tx,
"test-workspace",
ServiceName::Google,
resource_path,
)
.await;
.await?;
tx.commit().await?;
// OAuth resources gone
@@ -607,6 +625,7 @@ async fn test_rename_moves_native_trigger(db: Pool<Postgres>) -> anyhow::Result<
json!({"event": "OCP\\Files\\Events\\Node\\NodeCreatedEvent"}),
None,
true,
"u/test-user/native_conn",
)
.await?;
// An unrelated trigger already sitting on the target path must not be reported as moved.
@@ -624,6 +643,7 @@ async fn test_rename_moves_native_trigger(db: Pool<Postgres>) -> anyhow::Result<
json!({"event": "OCP\\Files\\Events\\Node\\NodeCreatedEvent"}),
None,
true,
"u/test-user/native_conn",
)
.await?;
@@ -696,6 +716,7 @@ async fn test_list_native_triggers_scope_filter(db: Pool<Postgres>) -> anyhow::R
json!({"event": "OCP\\Files\\Events\\Node\\NodeCreatedEvent"}),
None,
true,
"u/test-user/native_conn",
)
.await?;
}
@@ -931,6 +952,7 @@ async fn test_native_trigger_enabled_toggle(db: Pool<Postgres>) -> anyhow::Resul
json!({"event": "OCA\\Files\\Event\\LoadAdditionalScriptsEvent"}),
None,
true,
"u/test-user/native_conn",
)
.await?;
@@ -980,6 +1002,7 @@ async fn test_native_trigger_enabled_toggle(db: Pool<Postgres>) -> anyhow::Resul
json!({}),
None,
false,
"u/test-user/native_conn",
)
.await?;
assert!(
@@ -997,6 +1020,7 @@ async fn test_native_trigger_enabled_toggle(db: Pool<Postgres>) -> anyhow::Resul
json!({}),
None,
true,
"u/test-user/native_conn",
)
.await?;
assert!(
+106 -30
View File
@@ -20638,12 +20638,22 @@ paths:
schema:
$ref: "#/components/schemas/NativeServiceName"
requestBody:
description: redirect_uri
description: >-
redirect_uri, and where to save the connection (defaults to
u/{caller}/native_{resource_type}). Uses the workspace OAuth client, or the instance
one when the instance shares it.
required: true
content:
application/json:
schema:
$ref: "#/components/schemas/RedirectUri"
type: object
properties:
redirect_uri:
type: string
resource_path:
type: string
required:
- redirect_uri
responses:
"200":
description: native trigger service connect url
@@ -20673,10 +20683,10 @@ paths:
schema:
type: boolean
/w/{workspace}/native_triggers/integrations/{service_name}/generate_instance_connect_url:
post:
summary: generate connect url using instance-level credentials
operationId: generateInstanceConnectUrl
/w/{workspace}/native_triggers/integrations/{service_name}/connections:
get:
summary: list the native trigger connections of a service the caller can use
operationId: listNativeTriggerConnections
tags:
- workspace_integration
parameters:
@@ -20686,16 +20696,37 @@ paths:
required: true
schema:
$ref: "#/components/schemas/NativeServiceName"
requestBody:
description: redirect_uri
required: true
content:
application/json:
schema:
$ref: "#/components/schemas/RedirectUri"
responses:
"200":
description: authorization URL using instance credentials
description: connections
content:
application/json:
schema:
type: array
items:
$ref: "#/components/schemas/NativeTriggerConnection"
/w/{workspace}/native_triggers/integrations/{service_name}/connections/delete:
delete:
summary: disconnect a native trigger connection and delete the triggers using it
operationId: deleteNativeTriggerConnection
tags:
- workspace_integration
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- name: service_name
in: path
required: true
schema:
$ref: "#/components/schemas/NativeServiceName"
- name: path
in: query
required: true
schema:
type: string
responses:
"200":
description: connection deleted
content:
application/json:
schema:
@@ -20749,17 +20780,15 @@ paths:
type: string
redirect_uri:
type: string
resource_path:
type: string
required:
- code
- state
- redirect_uri
responses:
"200":
description: native trigger service oauth completed
description: path of the saved connection
content:
text/plain:
application/json:
schema:
type: string
@@ -21034,6 +21063,12 @@ paths:
required: true
schema:
type: string
- name: connection_path
in: query
description: Connection to browse through; defaults to the caller's only usable one
required: false
schema:
type: string
responses:
"200":
description: list of available NextCloud events
@@ -21056,6 +21091,12 @@ paths:
required: true
schema:
type: string
- name: connection_path
in: query
description: Connection to browse through; defaults to the caller's only usable one
required: false
schema:
type: string
responses:
"200":
description: list of Google Calendars
@@ -21078,6 +21119,12 @@ paths:
required: true
schema:
type: string
- name: connection_path
in: query
description: Connection to browse through; defaults to the caller's only usable one
required: false
schema:
type: string
- name: q
in: query
description: search query to filter files by name
@@ -21119,6 +21166,12 @@ paths:
required: true
schema:
type: string
- name: connection_path
in: query
description: Connection to browse through; defaults to the caller's only usable one
required: false
schema:
type: string
responses:
"200":
description: list of shared drives
@@ -21141,6 +21194,12 @@ paths:
required: true
schema:
type: string
- name: connection_path
in: query
description: Connection to browse through; defaults to the caller's only usable one
required: false
schema:
type: string
responses:
"200":
description: list of GitHub repositories
@@ -39081,6 +39140,10 @@ components:
enabled:
type: boolean
description: Whether the trigger starts a job when it fires
connection_path:
type: string
nullable: true
description: Path of the connection the trigger acts through on the external service
required:
- external_id
- workspace_id
@@ -39123,6 +39186,10 @@ components:
enabled:
type: boolean
description: Whether the trigger starts a job when it fires
connection_path:
type: string
nullable: true
description: Path of the connection the trigger acts through on the external service
external_data:
type: object
nullable: true
@@ -39154,10 +39221,6 @@ components:
oauth_data:
nullable: true
$ref: "#/components/schemas/WorkspaceOAuthConfig"
resource_path:
type: string
nullable: true
description: Path to the resource storing the OAuth token
required:
- service_name
@@ -39203,14 +39266,6 @@ components:
- async
- sync
RedirectUri:
type: object
properties:
redirect_uri:
type: string
required:
- redirect_uri
NativeTriggerData:
type: object
description: Data for creating or updating a native trigger
@@ -39235,11 +39290,32 @@ components:
Whether the trigger starts a job when it fires. Honoured on create only, so a
trigger can be registered already paused; an update ignores it and setenabled is
the only way to change an existing trigger's state. Defaults to true.
connection_path:
type: string
nullable: true
description: >-
Path of the connection the trigger acts through. Honoured on create only; an update
keeps the trigger's connection. When omitted on create, the caller's only usable
connection for the service is used.
required:
- script_path
- is_flow
- service_config
NativeTriggerConnection:
type: object
description: An account connected for native triggers, stored as an OAuth resource
properties:
path:
type: string
description: Path of the resource and variable holding the connection
owner:
type: string
nullable: true
description: Username of whoever connected the account, when the caller can read its resource
required:
- path
CreateTriggerResponse:
type: object
description: Response returned when a native trigger is created
+1 -1
View File
@@ -1188,7 +1188,7 @@ async fn offboard_user_from_workspace<'c>(
// ---- Related path tables ----
sqlx::query!(
r#"UPDATE workspace_integrations SET resource_path = REGEXP_REPLACE(resource_path, 'u/' || $2 || '/(.*)', $1 || '/\1') WHERE resource_path LIKE ('u/' || $2 || '/%') AND workspace_id = $3"#,
r#"UPDATE native_trigger SET connection_path = REGEXP_REPLACE(connection_path, 'u/' || $2 || '/(.*)', $1 || '/\1') WHERE connection_path LIKE ('u/' || $2 || '/%') AND workspace_id = $3"#,
&new_prefix, username, w_id
).execute(&mut **tx).await?;
+2 -3
View File
@@ -297,8 +297,7 @@ async fn rename_user(
// keeps a like-named member of some other workspace out of it.
if let Some(old_username) = old_instance_username.filter(|u| *u != ru.new_username) {
let old_principal = windmill_common::users::username_to_permissioned_as(&old_username);
let new_principal =
windmill_common::users::username_to_permissioned_as(&ru.new_username);
let new_principal = windmill_common::users::username_to_permissioned_as(&ru.new_username);
sqlx::query!(
"UPDATE app SET policy = jsonb_set(policy, ARRAY['on_behalf_of'], to_jsonb($1::text))
WHERE policy->>'on_behalf_of' = $2 AND policy->>'on_behalf_of_email' = $3",
@@ -618,7 +617,7 @@ async fn update_username_in_workpsace<'c>(
.await?;
sqlx::query!(
r#"UPDATE workspace_integrations SET resource_path = REGEXP_REPLACE(resource_path, 'u/' || $2 || '/(.*)', 'u/' || $1 || '/\1') WHERE resource_path LIKE ('u/' || $2 || '/%') AND workspace_id = $3"#,
r#"UPDATE native_trigger SET connection_path = REGEXP_REPLACE(connection_path, 'u/' || $2 || '/(.*)', 'u/' || $1 || '/\1') WHERE connection_path LIKE ('u/' || $2 || '/%') AND workspace_id = $3"#,
new_username,
old_username,
w_id
@@ -924,6 +924,7 @@ pub fn workspace_integration_oauth_key(client_name: &str) -> &str {
pub fn workspace_integration_token_endpoint(client_name: &str, base_url: &str) -> String {
match client_name {
"google" => "https://oauth2.googleapis.com/token".to_string(),
"github" => "https://github.com/login/oauth/access_token".to_string(),
_ => format!("{}/apps/oauth2/api/v1/token", base_url),
}
}
@@ -932,6 +933,7 @@ pub fn workspace_integration_token_endpoint(client_name: &str, base_url: &str) -
pub fn workspace_integration_auth_endpoint(client_name: &str, base_url: &str) -> String {
match client_name {
"google" => "https://accounts.google.com/o/oauth2/v2/auth".to_string(),
"github" => "https://github.com/login/oauth/authorize".to_string(),
_ => format!("{}/apps/oauth2/authorize", base_url),
}
}
@@ -33,7 +33,7 @@ impl External for GitHub {
async fn create(
&self,
w_id: &str,
_oauth_data: &Self::OAuthData,
oauth_data: &Self::OAuthData,
webhook_token: &str,
data: &NativeTriggerData<Self::ServiceConfig>,
db: &DB,
@@ -69,7 +69,15 @@ impl External for GitHub {
});
let response: CreateWebhookResponse = self
.http_client_request(&url, Method::POST, w_id, db, None, Some(&payload))
.http_client_request(
&url,
Method::POST,
w_id,
&oauth_data.connection_path,
db,
None,
Some(&payload),
)
.await?;
Ok(response)
@@ -78,7 +86,7 @@ impl External for GitHub {
async fn update(
&self,
w_id: &str,
_oauth_data: &Self::OAuthData,
oauth_data: &Self::OAuthData,
external_id: &str,
webhook_token: &str,
data: &NativeTriggerData<Self::ServiceConfig>,
@@ -114,7 +122,15 @@ impl External for GitHub {
});
let _: serde_json::Value = self
.http_client_request(&url, Method::PATCH, w_id, db, None, Some(&payload))
.http_client_request(
&url,
Method::PATCH,
w_id,
&oauth_data.connection_path,
db,
None,
Some(&payload),
)
.await?;
// Return the resolved service_config
@@ -125,7 +141,7 @@ impl External for GitHub {
async fn get(
&self,
w_id: &str,
_oauth_data: &Self::OAuthData,
oauth_data: &Self::OAuthData,
external_id: &str,
db: &DB,
_tx: &mut PgConnection,
@@ -137,7 +153,14 @@ impl External for GitHub {
};
let data = self
.get_webhook(w_id, &owner, &repo, external_id, db)
.get_webhook(
w_id,
&oauth_data.connection_path,
&owner,
&repo,
external_id,
db,
)
.await?;
Ok(Some(data))
}
@@ -145,7 +168,7 @@ impl External for GitHub {
async fn delete(
&self,
w_id: &str,
_oauth_data: &Self::OAuthData,
oauth_data: &Self::OAuthData,
external_id: &str,
db: &DB,
_tx: &mut PgConnection,
@@ -165,7 +188,15 @@ impl External for GitHub {
// Swallow 404 only (webhook may already be deleted on GitHub); propagate
// other errors so callers don't think cleanup succeeded when it didn't.
let result: std::result::Result<serde_json::Value, _> = self
.http_client_request::<_, ()>(&url, Method::DELETE, w_id, db, None, None)
.http_client_request::<_, ()>(
&url,
Method::DELETE,
w_id,
&oauth_data.connection_path,
db,
None,
None,
)
.await;
match result {
@@ -180,7 +211,7 @@ impl External for GitHub {
db: &DB,
workspace_id: &str,
triggers: &[NativeTrigger],
_oauth_data: &Self::OAuthData,
oauth_data: &Self::OAuthData,
synced: &mut Vec<crate::sync::TriggerSyncInfo>,
errors: &mut Vec<crate::sync::SyncError>,
) {
@@ -208,7 +239,15 @@ impl External for GitHub {
);
let result: std::result::Result<GithubWebhookApiResponse, _> = self
.http_client_request::<_, ()>(&url, Method::GET, workspace_id, db, None, None)
.http_client_request::<_, ()>(
&url,
Method::GET,
workspace_id,
&oauth_data.connection_path,
db,
None,
None,
)
.await;
match result {
@@ -363,6 +402,7 @@ impl GitHub {
async fn get_webhook(
&self,
w_id: &str,
connection_path: &str,
owner: &str,
repo: &str,
external_id: &str,
@@ -374,7 +414,7 @@ impl GitHub {
);
let response: GithubWebhookApiResponse = self
.http_client_request::<_, ()>(&url, Method::GET, w_id, db, None, None)
.http_client_request::<_, ()>(&url, Method::GET, w_id, connection_path, db, None, None)
.await?;
Ok(GithubTriggerData {
@@ -17,6 +17,7 @@ pub struct GithubOAuthData {
pub access_token: String,
pub refresh_token: Option<String>,
pub token_expires_at: Option<chrono::DateTime<chrono::Utc>>,
pub connection_path: String,
}
/// User-provided configuration for a GitHub webhook trigger.
@@ -1,14 +1,15 @@
use std::sync::Arc;
use axum::{extract::Path, routing::get, Extension, Json, Router};
use axum::{
extract::{Path, Query},
routing::get,
Extension, Json, Router,
};
use http::Method;
use windmill_api_auth::ApiAuthed;
use windmill_common::{error::JsonResult, DB};
use windmill_common::{db::UserDB, error::JsonResult, DB};
use crate::{
get_workspace_integration, map_external_error, require_native_integration_use, External,
ServiceName,
};
use crate::{map_external_error, picker_connection, ConnectionQuery, External, ServiceName};
use super::{GitHub, GithubApiRepoResponse, GithubRepoEntry};
@@ -19,10 +20,18 @@ async fn list_repos(
authed: ApiAuthed,
Extension(handler): Extension<Arc<GitHub>>,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Path(workspace_id): Path<String>,
Query(query): Query<ConnectionQuery>,
) -> JsonResult<Vec<GithubRepoEntry>> {
require_native_integration_use(&authed)?;
get_workspace_integration(&db, &workspace_id, ServiceName::Github).await?;
let connection_path = picker_connection(
&authed,
user_db,
&workspace_id,
ServiceName::Github,
query.connection_path.as_deref(),
)
.await?;
let mut all_entries = Vec::new();
@@ -35,7 +44,15 @@ async fn list_repos(
);
let repos: Vec<GithubApiRepoResponse> = handler
.http_client_request::<_, ()>(&url, Method::GET, &workspace_id, &db, None, None)
.http_client_request::<_, ()>(
&url,
Method::GET,
&workspace_id,
&connection_path,
&db,
None,
None,
)
.await
.map_err(map_external_error)?;
@@ -42,7 +42,7 @@ impl External for Google {
async fn create(
&self,
w_id: &str,
_oauth_data: &Self::OAuthData,
oauth_data: &Self::OAuthData,
webhook_token: &str,
data: &NativeTriggerData<Self::ServiceConfig>,
db: &DB,
@@ -51,8 +51,16 @@ impl External for Google {
// At creation time, channel_id also becomes the trigger's external_id
// (see external_id_and_metadata_from_response).
let channel_id = uuid::Uuid::new_v4().to_string();
self.create_watch_channel(w_id, &channel_id, &channel_id, webhook_token, data, db)
.await
self.create_watch_channel(
w_id,
&oauth_data.connection_path,
&channel_id,
&channel_id,
webhook_token,
data,
db,
)
.await
}
async fn update(
@@ -72,7 +80,15 @@ impl External for Google {
// external_id stays the stable routing key used in the webhook URL.
let channel_id = uuid::Uuid::new_v4().to_string();
let resp = self
.create_watch_channel(w_id, external_id, &channel_id, webhook_token, data, db)
.create_watch_channel(
w_id,
&oauth_data.connection_path,
external_id,
&channel_id,
webhook_token,
data,
db,
)
.await?;
self.service_config_from_create_response(data, &resp)
@@ -86,7 +102,7 @@ impl External for Google {
async fn delete(
&self,
w_id: &str,
_oauth_data: &Self::OAuthData,
oauth_data: &Self::OAuthData,
external_id: &str,
db: &DB,
tx: &mut PgConnection,
@@ -122,7 +138,15 @@ impl External for Google {
// Stop the channel (ignore errors - channel may have already expired)
let result: std::result::Result<serde_json::Value, _> = self
.http_client_request(&url, Method::POST, w_id, db, None, Some(&stop_request))
.http_client_request(
&url,
Method::POST,
w_id,
&oauth_data.connection_path,
db,
None,
Some(&stop_request),
)
.await;
if let Err(e) = result {
@@ -231,6 +255,7 @@ impl Google {
async fn create_watch_channel(
&self,
w_id: &str,
connection_path: &str,
external_id: &str,
channel_id: &str,
webhook_token: &str,
@@ -263,12 +288,24 @@ impl Google {
match data.service_config.trigger_type {
GoogleTriggerType::Drive => {
self.create_drive_watch(w_id, &data.service_config, &watch_request, db)
.await
self.create_drive_watch(
w_id,
connection_path,
&data.service_config,
&watch_request,
db,
)
.await
}
GoogleTriggerType::Calendar => {
self.create_calendar_watch(w_id, &data.service_config, &watch_request, db)
.await
self.create_calendar_watch(
w_id,
connection_path,
&data.service_config,
&watch_request,
db,
)
.await
}
}
}
@@ -276,6 +313,7 @@ impl Google {
async fn create_drive_watch(
&self,
w_id: &str,
connection_path: &str,
config: &GoogleServiceConfig,
watch_request: &WatchRequest,
db: &DB,
@@ -285,14 +323,30 @@ impl Google {
// Specific file: use files.watch
let url = format!("{}/files/{}/watch", endpoints::DRIVE_API_BASE, resource_id);
self.http_client_request(&url, Method::POST, w_id, db, None, Some(watch_request))
.await
self.http_client_request(
&url,
Method::POST,
w_id,
connection_path,
db,
None,
Some(watch_request),
)
.await
}
None => {
// All changes: use changes.watch
let token_url = format!("{}/changes/startPageToken", endpoints::DRIVE_API_BASE);
let token_response: serde_json::Value = self
.http_client_request::<_, ()>(&token_url, Method::GET, w_id, db, None, None)
.http_client_request::<_, ()>(
&token_url,
Method::GET,
w_id,
connection_path,
db,
None,
None,
)
.await?;
let start_page_token = token_response
@@ -313,6 +367,7 @@ impl Google {
&watch_url,
Method::POST,
w_id,
connection_path,
db,
None,
Some(&watch_body),
@@ -325,6 +380,7 @@ impl Google {
async fn create_calendar_watch(
&self,
w_id: &str,
connection_path: &str,
config: &GoogleServiceConfig,
watch_request: &WatchRequest,
db: &DB,
@@ -339,8 +395,16 @@ impl Google {
urlencoding::encode(calendar_id)
);
self.http_client_request(&url, Method::POST, w_id, db, None, Some(watch_request))
.await
self.http_client_request(
&url,
Method::POST,
w_id,
connection_path,
db,
None,
Some(watch_request),
)
.await
}
/// Renew an expiring Google watch channel.
@@ -362,6 +426,12 @@ impl Google {
.map(|v| serde_json::from_value(v.clone()))
.transpose()?
.ok_or_else(|| Error::InternalErr("Missing service config".to_string()))?;
let connection_path = trigger.connection_path.as_deref().ok_or_else(|| {
Error::BadRequest(format!(
"Cannot renew channel {}: the trigger has no Google connection",
trigger.external_id
))
})?;
let rotated = match crate::rotate_webhook_token(
db,
@@ -430,7 +500,15 @@ impl Google {
}
};
let result: std::result::Result<serde_json::Value, _> = self
.http_client_request(&url, Method::POST, w_id, db, None, Some(&stop_request))
.http_client_request(
&url,
Method::POST,
w_id,
connection_path,
db,
None,
Some(&stop_request),
)
.await;
if let Err(e) = result {
tracing::warn!(
@@ -444,11 +522,11 @@ impl Google {
// Create new watch channel with a fresh channel ID
let resp = match config.trigger_type {
GoogleTriggerType::Drive => {
self.create_drive_watch(w_id, &config, &watch_request, db)
self.create_drive_watch(w_id, connection_path, &config, &watch_request, db)
.await?
}
GoogleTriggerType::Calendar => {
self.create_calendar_watch(w_id, &config, &watch_request, db)
self.create_calendar_watch(w_id, connection_path, &config, &watch_request, db)
.await?
}
};
@@ -134,6 +134,7 @@ pub struct GoogleOAuthData {
pub refresh_token: Option<String>,
/// When the access token expires
pub token_expires_at: Option<chrono::DateTime<chrono::Utc>>,
pub connection_path: String,
}
/// Google API endpoints
@@ -10,10 +10,9 @@ use serde::{Deserialize, Serialize};
use windmill_api_auth::ApiAuthed;
use windmill_common::{error::JsonResult, DB};
use crate::{
get_workspace_integration, map_external_error, require_native_integration_use, External,
ServiceName,
};
use windmill_common::db::UserDB;
use crate::{map_external_error, picker_connection, ConnectionQuery, External, ServiceName};
use super::Google;
@@ -85,16 +84,25 @@ pub struct DriveFilesQuery {
pub page_token: Option<String>,
#[serde(default)]
pub shared_with_me: bool,
pub connection_path: Option<String>,
}
async fn list_calendars(
authed: ApiAuthed,
Extension(handler): Extension<Arc<Google>>,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Path(workspace_id): Path<String>,
Query(query): Query<ConnectionQuery>,
) -> JsonResult<Vec<GoogleCalendarEntry>> {
require_native_integration_use(&authed)?;
get_workspace_integration(&db, &workspace_id, ServiceName::Google).await?;
let connection_path = picker_connection(
&authed,
user_db,
&workspace_id,
ServiceName::Google,
query.connection_path.as_deref(),
)
.await?;
let url = format!(
"{}/users/me/calendarList",
@@ -102,7 +110,15 @@ async fn list_calendars(
);
let response: GoogleCalendarListResponse = handler
.http_client_request::<_, ()>(&url, Method::GET, &workspace_id, &db, None, None)
.http_client_request::<_, ()>(
&url,
Method::GET,
&workspace_id,
&connection_path,
&db,
None,
None,
)
.await
.map_err(map_external_error)?;
@@ -123,11 +139,18 @@ async fn list_drive_files(
authed: ApiAuthed,
Extension(handler): Extension<Arc<Google>>,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Path(workspace_id): Path<String>,
Query(query): Query<DriveFilesQuery>,
) -> JsonResult<GoogleDriveFilesResponse> {
require_native_integration_use(&authed)?;
get_workspace_integration(&db, &workspace_id, ServiceName::Google).await?;
let connection_path = picker_connection(
&authed,
user_db,
&workspace_id,
ServiceName::Google,
query.connection_path.as_deref(),
)
.await?;
let drive_query = if query.shared_with_me {
"sharedWithMe = true and trashed = false".to_string()
@@ -156,7 +179,15 @@ async fn list_drive_files(
}
let response: DriveApiResponse = handler
.http_client_request::<_, ()>(&url, Method::GET, &workspace_id, &db, None, None)
.http_client_request::<_, ()>(
&url,
Method::GET,
&workspace_id,
&connection_path,
&db,
None,
None,
)
.await
.map_err(map_external_error)?;
@@ -199,10 +230,18 @@ async fn list_shared_drives(
authed: ApiAuthed,
Extension(handler): Extension<Arc<Google>>,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Path(workspace_id): Path<String>,
Query(query): Query<ConnectionQuery>,
) -> JsonResult<Vec<SharedDriveEntry>> {
require_native_integration_use(&authed)?;
get_workspace_integration(&db, &workspace_id, ServiceName::Google).await?;
let connection_path = picker_connection(
&authed,
user_db,
&workspace_id,
ServiceName::Google,
query.connection_path.as_deref(),
)
.await?;
let url = format!(
"{}/drives?pageSize=100&fields=drives(id,name)",
@@ -210,7 +249,15 @@ async fn list_shared_drives(
);
let response: SharedDrivesApiResponse = handler
.http_client_request::<_, ()>(&url, Method::GET, &workspace_id, &db, None, None)
.http_client_request::<_, ()>(
&url,
Method::GET,
&workspace_id,
&connection_path,
&db,
None,
None,
)
.await
.map_err(map_external_error)?;
+70 -32
View File
@@ -1,11 +1,11 @@
use crate::{
classify_read_failure, decrypt_oauth_data, delete_native_trigger, delete_token_by_hash,
get_native_trigger, list_native_triggers, lock::TriggerLock, map_external_error,
map_external_error_with, rotate_webhook_token, set_native_trigger_enabled,
store_native_trigger, sync::EXTERNAL_TRIGGER_MISSING_ERROR, update_native_trigger_error,
update_native_trigger_if_runnable_unchanged, webhook_token_label, webhook_token_scopes,
External, ExternalReadFailure, NativeTrigger, NativeTriggerConfig, NativeTriggerData,
ServiceName,
map_external_error_with, resolve_usable_connection, rotate_webhook_token,
set_native_trigger_enabled, store_native_trigger, sync::EXTERNAL_TRIGGER_MISSING_ERROR,
update_native_trigger_error, update_native_trigger_if_runnable_unchanged, webhook_token_label,
webhook_token_scopes, External, ExternalReadFailure, NativeTrigger, NativeTriggerConfig,
NativeTriggerData, ServiceName,
};
use axum::{
extract::{Path, Query},
@@ -187,9 +187,20 @@ async fn create_native_trigger<T: External>(
)
.await?;
let integration_service = service_name.integration_service();
let oauth_data: T::OAuthData =
decrypt_oauth_data(&db, &workspace_id, integration_service).await?;
let connection_path = resolve_usable_connection(
&mut tx,
&workspace_id,
service_name.integration_service(),
data.connection_path.as_deref(),
)
.await?;
let oauth_data: T::OAuthData = decrypt_oauth_data(
&db,
&workspace_id,
service_name.integration_service(),
Some(&connection_path),
)
.await?;
let resp = handler
.create(
@@ -241,6 +252,7 @@ async fn create_native_trigger<T: External>(
service_config,
data.summary.as_deref(),
data.enabled,
&connection_path,
)
.await?;
@@ -282,10 +294,6 @@ async fn update_native_trigger_handler<T: External>(
.await?;
require_runnable_exists(&db, &workspace_id, &data.script_path, data.is_flow).await?;
let integration_service = service_name.integration_service();
let oauth_data: T::OAuthData =
decrypt_oauth_data(&db, &workspace_id, integration_service).await?;
let lock = TriggerLock::acquire(&db, &workspace_id, service_name, &external_id).await?;
let mut tx = user_db.clone().begin(&authed).await?;
@@ -294,6 +302,20 @@ async fn update_native_trigger_handler<T: External>(
.await?
.ok_or_else(|| Error::NotFound(format!("Native trigger not found: {}", external_id)))?;
// Saving re-registers the webhook as the connection's account, so the editor must be allowed
// to act as it, not just to write the runnable.
let integration_service = service_name.integration_service();
if let Some(path) = existing.connection_path.as_deref() {
resolve_usable_connection(&mut tx, &workspace_id, integration_service, Some(path)).await?;
}
let oauth_data: T::OAuthData = decrypt_oauth_data(
&db,
&workspace_id,
integration_service,
existing.connection_path.as_deref(),
)
.await?;
let runnable_changed =
existing.script_path != data.script_path || existing.is_flow != data.is_flow;
@@ -446,9 +468,13 @@ async fn get_native_trigger_handler<T: External>(
)
.await?;
let integration_service = service_name.integration_service();
let oauth_data: T::OAuthData =
decrypt_oauth_data(&db, &workspace_id, integration_service).await?;
let oauth_data: T::OAuthData = decrypt_oauth_data(
&db,
&workspace_id,
service_name.integration_service(),
windmill_trigger.connection_path.as_deref(),
)
.await?;
let native_trigger = handler
.get(&workspace_id, &oauth_data, &external_id, &db, &mut tx)
@@ -554,23 +580,35 @@ async fn delete_native_trigger_handler<T: External>(
)
.await?;
let integration_service = service_name.integration_service();
let oauth_data: T::OAuthData =
decrypt_oauth_data(&db, &workspace_id, integration_service).await?;
handler
.delete(&workspace_id, &oauth_data, &external_id, &db, &mut tx)
.await
.map_err(|e| {
map_external_error_with(e, |m| {
let end = if m.ends_with(['.', '!', '?']) {
""
} else {
"."
};
format!("{m}{end} The trigger was kept in Windmill, so it can still fire.")
})
})?;
// A connection can be deleted from under its triggers. Without it the registration cannot be
// removed, but deleting the row still revokes the webhook token, so the service's deliveries
// stop starting jobs; refusing would leave a trigger nobody can remove.
match decrypt_oauth_data::<T::OAuthData>(
&db,
&workspace_id,
service_name.integration_service(),
existing.connection_path.as_deref(),
)
.await
{
Ok(oauth_data) => handler
.delete(&workspace_id, &oauth_data, &external_id, &db, &mut tx)
.await
.map_err(|e| {
map_external_error_with(e, |m| {
let end = if m.ends_with(['.', '!', '?']) {
""
} else {
"."
};
format!("{m}{end} The trigger was kept in Windmill, so it can still fire.")
})
})?,
Err(e @ (Error::NotFound(_) | Error::BadRequest(_))) => tracing::warn!(
"Deleting {service_name} trigger {external_id} without removing its registration: {e}"
),
Err(e) => return Err(e),
}
let deleted =
delete_native_trigger(&mut *tx, &workspace_id, service_name, &external_id).await?;
+169 -85
View File
@@ -45,6 +45,7 @@ use std::{collections::HashMap, fmt::Debug};
use strum::{EnumIter, IntoEnumIterator};
use tokio::task;
use windmill_common::{
db::UserDB,
error::{to_anyhow, Error, Result},
triggers::TriggerKind,
utils::HTTP_CLIENT,
@@ -230,6 +231,9 @@ pub struct NativeTrigger {
/// its initial value and `setenabled` is its only mutator afterwards, so saving a
/// configuration can never silently re-enable a trigger someone paused.
pub enabled: bool,
/// Path of the connection (the OAuth resource created by the connect flow) the trigger acts
/// through on the external service.
pub connection_path: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -253,6 +257,11 @@ pub struct NativeTriggerData<C> {
/// An update ignores it: `setenabled` is the only way to change an existing trigger's state.
#[serde(default = "default_true")]
pub enabled: bool,
/// Honoured on create only: the webhook is registered under this connection's account, so
/// switching it would leave the registration owned by an account the trigger no longer uses.
/// When omitted, the caller's only readable connection for the service is used.
#[serde(default)]
pub connection_path: Option<String>,
}
#[derive(Debug, Clone, FromRow, Serialize, Deserialize)]
@@ -260,7 +269,6 @@ pub struct WorkspaceIntegration {
pub workspace_id: String,
pub service_name: ServiceName,
pub oauth_data: serde_json::Value,
pub resource_path: Option<String>,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
pub created_by: String,
@@ -399,12 +407,18 @@ pub trait External: Send + Sync + 'static {
url: &str,
method: Method,
workspace_id: &str,
connection_path: &str,
db: &DB,
headers: Option<HashMap<String, String>>,
body: Option<&B>,
) -> Result<T> {
let oauth_config: OAuthConfig =
decrypt_oauth_data(db, workspace_id, Self::SERVICE_NAME).await?;
let oauth_config: OAuthConfig = decrypt_oauth_data(
db,
workspace_id,
Self::SERVICE_NAME.integration_service(),
Some(connection_path),
)
.await?;
let result = make_http_request(
url,
@@ -446,14 +460,14 @@ pub trait External: Send + Sync + 'static {
task::spawn({
let db_clone = db.clone();
let workspace_id_clone = workspace_id.to_string();
let service_name = Self::SERVICE_NAME;
let connection_path = connection_path.to_string();
let new_access_token = refreshed_oauth_config.access_token.clone();
let new_refresh_token = refreshed_oauth_config.refresh_token.clone();
async move {
update_oauth_token_resource(
&db_clone,
&workspace_id_clone,
service_name,
&connection_path,
&new_access_token,
new_refresh_token.as_deref(),
)
@@ -594,6 +608,7 @@ pub struct OAuthConfig {
pub refresh_token: Option<String>,
pub client_id: String,
pub client_secret: String,
pub connection_path: String,
}
pub async fn make_http_request<T: DeserializeOwned + Send, B: Serialize>(
@@ -776,51 +791,50 @@ async fn get_instance_oauth_credentials(
.await
}
/// Assemble the credentials of one connection: its token from the variable and account at
/// `connection_path`, and the OAuth client from the workspace (or instance) configuration.
///
/// Only an account made by the connect flow is accepted, since token refresh goes through this
/// service's OAuth client and any other account would be refreshed against the wrong one.
pub async fn decrypt_oauth_data<T: DeserializeOwned>(
db: &DB,
workspace_id: &str,
service_name: ServiceName,
connection_path: Option<&str>,
) -> Result<T> {
let integration = get_workspace_integration(db, workspace_id, service_name).await?;
let oauth_data = integration.oauth_data;
let resource_path = integration.resource_path.as_deref().ok_or_else(|| {
Error::InternalErr(format!(
"No resource_path in {} integration config. Please reconnect the integration.",
service_name
let connection_path = connection_path.ok_or_else(|| {
Error::BadRequest(format!(
"This trigger has no {service_name} connection. Delete it and create it again with a connection."
))
})?;
let integration = get_workspace_integration(db, workspace_id, service_name).await?;
let oauth_data = integration.oauth_data;
let mc = build_crypt(db, workspace_id).await?;
let var_row = sqlx::query!(
"SELECT value, account FROM variable WHERE workspace_id = $1 AND path = $2",
let row = sqlx::query!(
"SELECT v.value, a.refresh_token
FROM variable v
JOIN account a ON a.workspace_id = v.workspace_id AND a.id = v.account
WHERE v.workspace_id = $1 AND v.path = $2
AND a.client = $3 AND a.is_workspace_integration = true",
workspace_id,
resource_path,
connection_path,
service_name.as_str(),
)
.fetch_optional(db)
.await?
.ok_or_else(|| {
Error::InternalErr(format!(
"Variable at {} not found for {} integration",
resource_path, service_name
Error::NotFound(format!(
"No {service_name} connection at {connection_path}. Connect an account there or pick another connection."
))
})?;
let access_token = decrypt(&mc, var_row.value)
let access_token = decrypt(&mc, row.value)
.map_err(|e| Error::InternalErr(format!("Failed to decrypt access token: {}", e)))?;
let refresh_token = if let Some(account_id) = var_row.account {
sqlx::query_scalar!(
"SELECT refresh_token FROM account WHERE workspace_id = $1 AND id = $2",
workspace_id,
account_id,
)
.fetch_optional(db)
.await?
} else {
None
};
let refresh_token = Some(row.refresh_token);
let (client_id, client_secret) = if oauth_data
.get("instance_shared")
@@ -853,6 +867,7 @@ pub async fn decrypt_oauth_data<T: DeserializeOwned>(
"refresh_token": refresh_token,
"client_id": client_id,
"client_secret": client_secret,
"connection_path": connection_path,
});
serde_json::from_value(assembled)
@@ -978,6 +993,7 @@ pub async fn refresh_oauth_tokens(
.or_else(|| oauth_config.refresh_token.clone()),
client_id: oauth_config.client_id.clone(),
client_secret: oauth_config.client_secret.clone(),
connection_path: oauth_config.connection_path.clone(),
})
}
@@ -994,70 +1010,58 @@ pub async fn refresh_oauth_tokens(
})
}
async fn update_oauth_token_resource(
/// Write a refreshed token back to the one connection it was refreshed for. The account is found
/// through that connection's variable, never by service: several connections share a service, and
/// a provider that rotates refresh tokens invalidates the old one, so writing it onto another
/// account would break that account's next refresh.
pub async fn update_oauth_token_resource(
db: &DB,
workspace_id: &str,
service_name: ServiceName,
connection_path: &str,
new_access_token: &str,
new_refresh_token: Option<&str>,
) {
let result = async {
let integration = get_workspace_integration(db, workspace_id, service_name).await?;
let resource_path = integration.resource_path.ok_or_else(|| {
Error::InternalErr(format!(
"No resource_path in {} integration config",
service_name
))
})?;
let mc = build_crypt(db, workspace_id).await?;
let encrypted_token = encrypt(&mc, new_access_token);
sqlx::query!(
"UPDATE variable SET value = $1 WHERE workspace_id = $2 AND path = $3",
let account_id = sqlx::query_scalar!(
"UPDATE variable SET value = $1 WHERE workspace_id = $2 AND path = $3 RETURNING account",
encrypted_token,
workspace_id,
resource_path,
connection_path,
)
.fetch_optional(db)
.await?
.flatten();
let Some(account_id) = account_id else {
return Ok(());
};
// Even without a new refresh token, expires_at moves so the background refresh does not
// re-refresh immediately.
sqlx::query!(
"UPDATE account SET
refresh_token = COALESCE($1, refresh_token),
expires_at = now() + interval '1 hour',
refresh_error = NULL
WHERE workspace_id = $2 AND id = $3",
new_refresh_token,
workspace_id,
account_id,
)
.execute(db)
.await?;
if let Some(refresh_token) = new_refresh_token {
sqlx::query!(
"UPDATE account SET
refresh_token = $1,
expires_at = now() + interval '1 hour',
refresh_error = NULL
WHERE workspace_id = $2 AND client = $3 AND is_workspace_integration = true",
refresh_token,
workspace_id,
service_name.as_str(),
)
.execute(db)
.await?;
} else {
// Even without a new refresh token, update expires_at to prevent
// the background refresh from re-refreshing immediately
sqlx::query!(
"UPDATE account SET
expires_at = now() + interval '1 hour',
refresh_error = NULL
WHERE workspace_id = $1 AND client = $2 AND is_workspace_integration = true",
workspace_id,
service_name.as_str(),
)
.execute(db)
.await?;
}
Ok::<(), Error>(())
}
.await;
if let Err(e) = result {
tracing::error!(
"Failed to update OAuth tokens for {} in workspace {}: {}",
service_name,
"Failed to update OAuth tokens of connection {} in workspace {}: {}",
connection_path,
workspace_id,
e
);
@@ -1192,6 +1196,7 @@ pub async fn store_native_trigger<'c, E: sqlx::Executor<'c, Database = Postgres>
service_config: C,
summary: Option<&str>,
enabled: bool,
connection_path: &str,
) -> Result<()> {
use windmill_common::auth::hash_token;
@@ -1211,12 +1216,13 @@ pub async fn store_native_trigger<'c, E: sqlx::Executor<'c, Database = Postgres>
webhook_token_hash,
service_config,
summary,
enabled
enabled,
connection_path
) VALUES (
$1, $2, $3, $4, $5, $6, $7, $8, $9
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10
)
ON CONFLICT (external_id, workspace_id, service_name)
DO UPDATE SET script_path = $4, is_flow = $5, webhook_token_hash = $6, service_config = $7, summary = $8, error = NULL, updated_at = NOW()
DO UPDATE SET script_path = $4, is_flow = $5, webhook_token_hash = $6, service_config = $7, summary = $8, connection_path = $10, error = NULL, updated_at = NOW()
"#,
external_id,
workspace_id,
@@ -1227,6 +1233,7 @@ pub async fn store_native_trigger<'c, E: sqlx::Executor<'c, Database = Postgres>
sqlx::types::Json(service_config) as _,
summary,
enabled,
connection_path,
)
.execute(db)
.await?;
@@ -1424,7 +1431,8 @@ pub async fn get_native_trigger<'c, E: sqlx::Executor<'c, Database = Postgres>>(
created_at,
updated_at,
summary,
enabled
enabled,
connection_path
FROM
native_trigger
WHERE
@@ -1464,7 +1472,8 @@ pub async fn get_native_trigger_by_script<'c, E: sqlx::Executor<'c, Database = P
created_at,
updated_at,
summary,
enabled
enabled,
connection_path
FROM
native_trigger
WHERE
@@ -1520,7 +1529,8 @@ pub async fn list_native_triggers<'c, E: sqlx::Executor<'c, Database = Postgres>
nt.created_at,
nt.updated_at,
nt.summary,
nt.enabled
nt.enabled,
nt.connection_path
FROM
native_trigger nt
WHERE
@@ -1699,7 +1709,6 @@ pub async fn store_workspace_integration(
workspace_id: &str,
service_name: ServiceName,
oauth_data: serde_json::Value,
resource_path: Option<&str>,
) -> Result<()> {
sqlx::query!(
r#"
@@ -1707,23 +1716,20 @@ pub async fn store_workspace_integration(
workspace_id,
service_name,
oauth_data,
resource_path,
created_by,
created_at,
updated_at
) VALUES (
$1, $2, $3, $4, $5, now(), now()
$1, $2, $3, $4, now(), now()
)
ON CONFLICT (workspace_id, service_name)
DO UPDATE SET
oauth_data = $3,
resource_path = $4,
updated_at = now()
"#,
workspace_id,
service_name as ServiceName,
oauth_data,
resource_path,
authed.username,
)
.execute(&mut *tx)
@@ -1746,6 +1752,85 @@ pub fn require_native_integration_use(authed: &ApiAuthed) -> Result<()> {
Ok(())
}
#[derive(Debug, Serialize)]
pub struct Connection {
pub path: String,
/// Username of whoever connected the account, when the caller can read its resource.
pub owner: Option<String>,
}
/// The connections of a service the caller may use. `tx` must be a user transaction: using a
/// connection means acting as the account behind it, so it is gated on reading its token
/// variable, which the variable's row-level security decides.
pub async fn list_usable_connections(
tx: &mut PgConnection,
workspace_id: &str,
service_name: ServiceName,
) -> Result<Vec<Connection>> {
let connections = sqlx::query_as!(
Connection,
r#"SELECT v.path, r.created_by AS "owner?"
FROM variable v
JOIN account a ON a.workspace_id = v.workspace_id AND a.id = v.account
LEFT JOIN resource r ON r.workspace_id = v.workspace_id AND r.path = v.path
WHERE v.workspace_id = $1 AND a.client = $2 AND a.is_workspace_integration = true
ORDER BY v.path"#,
workspace_id,
service_name.as_str(),
)
.fetch_all(&mut *tx)
.await?;
Ok(connections)
}
/// Pick the connection a new trigger acts through: the one asked for if the caller may use it,
/// otherwise the caller's only usable connection. `tx` must be a user transaction.
pub async fn resolve_usable_connection(
tx: &mut PgConnection,
workspace_id: &str,
service_name: ServiceName,
requested: Option<&str>,
) -> Result<String> {
let usable = list_usable_connections(tx, workspace_id, service_name).await?;
match requested {
Some(path) if usable.iter().any(|c| c.path == path) => Ok(path.to_string()),
Some(path) => Err(Error::NotAuthorized(format!(
"No {service_name} connection you can use at {path}. Connect your own account or ask \
for read access to that connection."
))),
None => match usable.as_slice() {
[only] => Ok(only.path.clone()),
[] => Err(Error::BadRequest(format!(
"You have no {service_name} connection. Connect your account first."
))),
_ => Err(Error::BadRequest(format!(
"You can use several {service_name} connections; pick one with connection_path."
))),
},
}
}
#[derive(Debug, Deserialize)]
pub struct ConnectionQuery {
pub connection_path: Option<String>,
}
/// Gate for the pickers that browse the service while configuring a trigger: returns the
/// connection to browse through, which the caller must be allowed to use.
pub async fn picker_connection(
authed: &ApiAuthed,
user_db: UserDB,
workspace_id: &str,
service_name: ServiceName,
requested: Option<&str>,
) -> Result<String> {
require_native_integration_use(authed)?;
let mut tx = user_db.begin(authed).await?;
let path = resolve_usable_connection(&mut tx, workspace_id, service_name, requested).await?;
tx.commit().await?;
Ok(path)
}
pub async fn get_workspace_integration<'c, E: sqlx::Executor<'c, Database = Postgres>>(
db: E,
workspace_id: &str,
@@ -1758,7 +1843,6 @@ pub async fn get_workspace_integration<'c, E: sqlx::Executor<'c, Database = Post
workspace_id,
service_name AS "service_name!: ServiceName",
oauth_data,
resource_path,
created_at,
updated_at,
created_by
@@ -131,6 +131,7 @@ impl External for NextCloud {
&url,
Method::POST,
w_id,
&oauth_data.connection_path,
db,
Some(headers),
Some(&full_nextcloud_payload),
@@ -167,6 +168,7 @@ impl External for NextCloud {
&url,
Method::POST,
w_id,
&oauth_data.connection_path,
db,
Some(headers),
Some(&full_nextcloud_payload),
@@ -214,7 +216,15 @@ impl External for NextCloud {
headers.insert("OCS-APIRequest".to_string(), "true".to_string());
let ocs_response: OcsResponse<NextCloudTriggerData> = self
.http_client_request::<_, ()>(&url, Method::GET, w_id, db, Some(headers), None)
.http_client_request::<_, ()>(
&url,
Method::GET,
w_id,
&oauth_data.connection_path,
db,
Some(headers),
None,
)
.await?;
Ok(Some(ocs_response.ocs.data))
@@ -239,7 +249,15 @@ impl External for NextCloud {
// A webhook already removed in Nextcloud is the outcome this call wants, so only 404 is
// swallowed; anything else would leave a live webhook behind while Windmill forgets it.
let _: serde_json::Value = self
.http_client_request::<_, ()>(&url, Method::DELETE, w_id, db, Some(headers), None)
.http_client_request::<_, ()>(
&url,
Method::DELETE,
w_id,
&oauth_data.connection_path,
db,
Some(headers),
None,
)
.await
.or_else(|e| match http_error_status(&e) {
Some(StatusCode::NOT_FOUND) => Ok(serde_json::Value::Null),
@@ -353,6 +371,7 @@ impl NextCloud {
&url,
Method::GET,
w_id,
&oauth_data.connection_path,
db,
Some(headers),
None,
@@ -9,6 +9,7 @@ pub struct NextCloudOAuthData {
pub access_token: String,
pub refresh_token: Option<String>,
pub token_expires_at: Option<chrono::DateTime<chrono::Utc>>,
pub connection_path: String,
}
#[derive(Debug, Serialize, Deserialize)]
@@ -1,8 +1,13 @@
use std::{collections::HashMap, sync::Arc};
use axum::{extract::Path, routing::get, Extension, Json, Router};
use axum::{
extract::{Path, Query},
routing::get,
Extension, Json, Router,
};
use http::Method;
use windmill_common::{
db::UserDB,
error::{Error, JsonResult},
DB,
};
@@ -12,16 +17,25 @@ use windmill_api_auth::ApiAuthed;
use crate::{
get_workspace_integration, map_external_error,
nextcloud::{NextCloudEventType, OcsResponse},
require_native_integration_use, External, ServiceName,
picker_connection, ConnectionQuery, External, ServiceName,
};
async fn list_available_events<T: External>(
authed: ApiAuthed,
Extension(handler): Extension<Arc<T>>,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Path(workspace_id): Path<String>,
Query(query): Query<ConnectionQuery>,
) -> JsonResult<Vec<NextCloudEventType>> {
require_native_integration_use(&authed)?;
let connection_path = picker_connection(
&authed,
user_db,
&workspace_id,
ServiceName::Nextcloud,
query.connection_path.as_deref(),
)
.await?;
let integration = get_workspace_integration(&db, &workspace_id, ServiceName::Nextcloud).await?;
let base_url = integration
@@ -43,6 +57,7 @@ async fn list_available_events<T: External>(
&url,
Method::GET,
&workspace_id,
&connection_path,
&db,
Some(headers),
None,
+11 -14
View File
@@ -69,19 +69,8 @@ async fn reregister_service<T: External>(
moved: &[&MovedNativeTrigger],
handler: T,
) {
let oauth_data: T::OAuthData =
match decrypt_oauth_data(db, w_id, T::SERVICE_NAME.integration_service()).await {
Ok(oauth_data) => oauth_data,
Err(e) => {
for trigger in moved {
record_failure::<T>(db, w_id, &trigger.external_id, &e).await;
}
return;
}
};
for trigger in moved {
if let Err(e) = reregister_one(db, authed, w_id, &handler, &oauth_data, trigger).await {
if let Err(e) = reregister_one(db, authed, w_id, &handler, trigger).await {
record_failure::<T>(db, w_id, &trigger.external_id, &e).await;
}
}
@@ -99,7 +88,6 @@ async fn reregister_one<T: External>(
authed: &ApiAuthed,
w_id: &str,
handler: &T,
oauth_data: &T::OAuthData,
moved: &MovedNativeTrigger,
) -> Result<()> {
let external_id = moved.external_id.as_str();
@@ -136,8 +124,17 @@ async fn reregister_one<T: External>(
service_config,
summary: trigger.summary.clone(),
enabled: trigger.enabled,
connection_path: trigger.connection_path.clone(),
};
let oauth_data: T::OAuthData = decrypt_oauth_data(
db,
w_id,
T::SERVICE_NAME.integration_service(),
trigger.connection_path.as_deref(),
)
.await?;
// The token is scoped to the runnable path and only its hash is kept, so pointing the webhook
// at the new path means minting a replacement rather than reusing the old one. Commit it
// before handing it out: a service may call back the moment it accepts the new URL.
@@ -157,7 +154,7 @@ async fn reregister_one<T: External>(
let updated = handler
.update(
w_id,
oauth_data,
&oauth_data,
external_id,
&webhook_token,
&data,
+35 -20
View File
@@ -6,6 +6,9 @@ use serde::Serialize;
use crate::ServiceName;
#[cfg(feature = "native_trigger")]
use itertools::Itertools;
#[cfg(feature = "native_trigger")]
use crate::{
decrypt_oauth_data, list_native_triggers, update_native_trigger_error,
@@ -238,38 +241,50 @@ pub async fn sync_workspace_triggers<T: External>(
let mut synced = Vec::new();
let mut errors = Vec::new();
// Use the integration service for OAuth lookup (e.g., GoogleDrive/GoogleCalendar -> Google)
let integration_service = T::SERVICE_NAME.integration_service();
// Maintenance runs per connection: each one's triggers are registered under its account, and
// Nextcloud reconciles against that account's own list of webhooks.
let by_connection = windmill_triggers
.into_iter()
.into_group_map_by(|t| t.connection_path.clone());
let oauth_data = {
match decrypt_oauth_data(db, workspace_id, integration_service).await {
for (connection_path, triggers) in by_connection {
let oauth_data = match decrypt_oauth_data(
db,
workspace_id,
T::SERVICE_NAME.integration_service(),
connection_path.as_deref(),
)
.await
{
Ok(oauth_data) => oauth_data,
Err(e) => {
tracing::error!(
"Failed to get workspace integration OAuth data for {}: {}",
"Failed to get the OAuth data of connection {:?} in {}: {}",
connection_path,
workspace_id,
e
);
errors.push(SyncError {
resource_path: format!("workspace:{}", workspace_id),
error_message: format!("Failed to get workspace integration OAuth data: {}", e),
resource_path: connection_path
.unwrap_or_else(|| format!("workspace:{}", workspace_id)),
error_message: format!("Failed to get connection OAuth data: {}", e),
error_type: "oauth_error".to_string(),
});
return Ok((Vec::new(), errors));
continue;
}
}
};
};
handler
.maintain_triggers(
db,
workspace_id,
&windmill_triggers,
&oauth_data,
&mut synced,
&mut errors,
)
.await;
handler
.maintain_triggers(
db,
workspace_id,
&triggers,
&oauth_data,
&mut synced,
&mut errors,
)
.await;
}
tracing::info!(
"Sync completed for {} in workspace '{}'. Updated: {}, Errors: {}",
@@ -2,7 +2,7 @@ use axum::Router;
#[cfg(feature = "native_trigger")]
use axum::{
extract::Path,
extract::{Path, Query},
routing::{delete, get, post},
Extension, Json,
};
@@ -30,14 +30,18 @@ use windmill_common::{
};
#[cfg(feature = "native_trigger")]
use windmill_api_auth::ApiAuthed;
use windmill_api_auth::{require_is_writer, ApiAuthed};
#[cfg(feature = "native_trigger")]
use crate::{
decrypt_oauth_data, delete_token_by_hash, delete_workspace_integration, nextcloud::OcsResponse,
resolve_endpoint, store_workspace_integration, ServiceName,
decrypt_oauth_data, delete_token_by_hash, delete_workspace_integration,
list_usable_connections, nextcloud::OcsResponse, require_native_integration_use,
resolve_endpoint, Connection, ServiceName,
};
#[cfg(feature = "native_trigger")]
use std::collections::HashMap;
#[cfg(feature = "native_trigger")]
use windmill_oauth::{OClient, Url, OAUTH_HTTP_CLIENT};
@@ -57,24 +61,27 @@ type HmacSha256 = Hmac<Sha256>;
const STATE_EXPIRATION_SECONDS: i64 = 600; // 10 minutes
/// Generate a signed OAuth state that is cluster-safe.
/// The state contains: workspace_id, service_name, timestamp, and nonce.
/// The state contains: workspace_id, service_name, timestamp, nonce, and the path the connection
/// will be saved at, so the callback needs nothing from the page the provider redirects to.
/// It's signed with HMAC-SHA256 using the workspace key.
#[cfg(feature = "native_trigger")]
async fn generate_signed_state(
db: &DB,
workspace_id: &str,
service_name: ServiceName,
connection_path: &str,
) -> Result<String> {
use windmill_common::variables::get_workspace_key;
let nonce = uuid::Uuid::new_v4().to_string();
let timestamp = chrono::Utc::now().timestamp();
let payload = format!(
"{}:{}:{}:{}",
"{}:{}:{}:{}:{}",
workspace_id,
service_name.as_str(),
timestamp,
nonce
nonce,
URL_SAFE_NO_PAD.encode(connection_path)
);
// Get workspace key for signing
@@ -91,65 +98,58 @@ async fn generate_signed_state(
Ok(format!("{}:{}", encoded_payload, encoded_signature))
}
/// Validate a signed OAuth state.
/// Returns true if the state is valid (correct signature and not expired).
/// Validate a signed OAuth state (correct signature, workspace and service, not expired) and
/// return the connection path it was issued for.
#[cfg(feature = "native_trigger")]
async fn validate_signed_state(db: &DB, state: &str, workspace_id: &str) -> Result<bool> {
async fn validate_signed_state(
db: &DB,
state: &str,
workspace_id: &str,
service_name: ServiceName,
) -> Result<String> {
use windmill_common::variables::get_workspace_key;
let parts: Vec<&str> = state.split(':').collect();
if parts.len() != 2 {
return Ok(false);
let invalid = || Error::BadRequest("Invalid or expired state parameter".to_string());
let (encoded_payload, encoded_signature) = state.split_once(':').ok_or_else(invalid)?;
let payload = URL_SAFE_NO_PAD
.decode(encoded_payload)
.ok()
.and_then(|bytes| String::from_utf8(bytes).ok())
.ok_or_else(invalid)?;
// Parse payload: workspace_id:service_name:timestamp:nonce:connection_path
let [state_workspace_id, state_service, timestamp, _nonce, encoded_path] = payload
.split(':')
.collect::<Vec<_>>()
.try_into()
.map_err(|_| invalid())?;
if state_workspace_id != workspace_id || state_service != service_name.as_str() {
return Err(invalid());
}
let encoded_payload = parts[0];
let encoded_signature = parts[1];
// Decode payload
let payload_bytes = match URL_SAFE_NO_PAD.decode(encoded_payload) {
Ok(bytes) => bytes,
Err(_) => return Ok(false),
};
let payload = match String::from_utf8(payload_bytes) {
Ok(s) => s,
Err(_) => return Ok(false),
};
// Parse payload: workspace_id:service_name:timestamp:nonce
let payload_parts: Vec<&str> = payload.split(':').collect();
if payload_parts.len() != 4 {
return Ok(false);
let timestamp: i64 = timestamp.parse().map_err(|_| invalid())?;
if chrono::Utc::now().timestamp() - timestamp > STATE_EXPIRATION_SECONDS {
return Err(invalid());
}
let state_workspace_id = payload_parts[0];
let timestamp: i64 = match payload_parts[2].parse() {
Ok(ts) => ts,
Err(_) => return Ok(false),
};
// Verify workspace_id matches
if state_workspace_id != workspace_id {
return Ok(false);
}
// Check expiration
let now = chrono::Utc::now().timestamp();
if now - timestamp > STATE_EXPIRATION_SECONDS {
return Ok(false);
}
// Verify signature
let key = get_workspace_key(workspace_id, db).await?;
let mut mac = HmacSha256::new_from_slice(key.as_bytes())
.map_err(|e| Error::InternalErr(e.to_string()))?;
mac.update(payload.as_bytes());
let received_signature = URL_SAFE_NO_PAD
.decode(encoded_signature)
.map_err(|_| invalid())?;
mac.verify_slice(&received_signature)
.map_err(|_| invalid())?;
let received_signature = match URL_SAFE_NO_PAD.decode(encoded_signature) {
Ok(bytes) => bytes,
Err(_) => return Ok(false),
};
Ok(mac.verify_slice(&received_signature).is_ok())
URL_SAFE_NO_PAD
.decode(encoded_path)
.ok()
.and_then(|bytes| String::from_utf8(bytes).ok())
.ok_or_else(invalid)
}
#[cfg(feature = "native_trigger")]
@@ -194,20 +194,65 @@ pub struct OAuthConfigResponse {
pub redirect_uri: Option<String>,
}
/// The OAuth client a connection is made with: the workspace's own when an admin configured one,
/// otherwise the instance's when the instance admin shares it. The flag says which.
#[cfg(feature = "native_trigger")]
async fn resolve_oauth_client(
db: &DB,
workspace_id: &str,
service_name: ServiceName,
) -> Result<(WorkspaceOAuthConfig, bool)> {
let workspace_config =
get_workspace_oauth_config::<WorkspaceOAuthConfig>(db, workspace_id, service_name)
.await
.ok()
.filter(|c| !c.instance_shared && !c.client_id.is_empty());
match workspace_config {
Some(config) => Ok((config, false)),
None => Ok((get_instance_oauth_config(db, service_name).await?, true)),
}
}
#[cfg(feature = "native_trigger")]
fn default_connection_path(authed: &ApiAuthed, service_name: ServiceName) -> String {
format!(
"u/{}/native_{}",
authed.username,
service_name.resource_type()
)
}
#[cfg(feature = "native_trigger")]
async fn generate_connect_url(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Path((workspace_id, service_name)): Path<(String, ServiceName)>,
Json(RedirectUri { redirect_uri }): Json<RedirectUri>,
Json(body): Json<ConnectUrlBody>,
) -> JsonResult<String> {
require_admin(authed.is_admin, &workspace_id)?;
require_native_integration_use(&authed)?;
let oauth_config =
get_workspace_oauth_config_as_oauth_config(&db, &workspace_id, service_name).await?;
let (oauth_config, is_instance_shared) =
resolve_oauth_client(&db, &workspace_id, service_name).await?;
let connection_path = body
.resource_path
.filter(|p| !p.is_empty())
.unwrap_or_else(|| default_connection_path(&authed, service_name));
// A GitHub OAuth app only redirects under its one registered callback, and the instance app's
// is the resource connect callback, so its connections land on a page below that one.
let redirect_uri = if is_instance_shared && service_name == ServiceName::Github {
let origin = Url::parse(&body.redirect_uri)
.map_err(|e| Error::BadRequest(format!("Invalid redirect URI: {e}")))?
.origin()
.ascii_serialization();
format!("{origin}/oauth/callback/github/native_trigger")
} else {
body.redirect_uri
};
// Generate a signed state that is cluster-safe
let state = generate_signed_state(&db, &workspace_id, service_name).await?;
let state = generate_signed_state(&db, &workspace_id, service_name, &connection_path).await?;
let auth_url = build_authorization_url(&oauth_config, service_name, &state, &redirect_uri);
Ok(Json(auth_url))
}
@@ -287,14 +332,22 @@ async fn fetch_nextcloud_user_id(base_url: &str, access_token: &str) -> anyhow::
Ok(ocs.ocs.data.id)
}
/// Delete all native triggers for a workspace+service, including remote webhook cleanup.
/// Delete the native triggers of a workspace+service — only those using `connection_path` when
/// given — including remote webhook cleanup.
/// This is best-effort: errors during remote cleanup or token deletion are logged but ignored.
#[cfg(feature = "native_trigger")]
async fn delete_triggers_for_service(db: &DB, workspace_id: &str, service_name: ServiceName) {
async fn delete_triggers_for_service(
db: &DB,
workspace_id: &str,
service_name: ServiceName,
connection_path: Option<&str>,
) {
let triggers = sqlx::query!(
"SELECT external_id, webhook_token_hash FROM native_trigger WHERE workspace_id = $1 AND service_name = $2",
"SELECT external_id, connection_path FROM native_trigger
WHERE workspace_id = $1 AND service_name = $2 AND ($3::text IS NULL OR connection_path = $3)",
workspace_id,
service_name as ServiceName
service_name as ServiceName,
connection_path,
)
.fetch_all(db)
.await;
@@ -311,27 +364,33 @@ async fn delete_triggers_for_service(db: &DB, workspace_id: &str, service_name:
return;
}
// For Nextcloud: try to delete webhooks on the remote instance (best-effort)
if service_name == ServiceName::Nextcloud {
if let Ok(oauth_data) =
decrypt_oauth_data::<BasicOAuthData>(db, workspace_id, service_name).await
{
for trigger in &triggers {
// Nextcloud and GitHub: try to delete the webhooks remotely (best-effort).
// Google: skip remote cleanup (watch channels expire naturally).
if service_name != ServiceName::Google {
let mut credentials: HashMap<Option<String>, Option<BasicOAuthData>> = HashMap::new();
for trigger in &triggers {
if !credentials.contains_key(&trigger.connection_path) {
let oauth_data = decrypt_oauth_data::<BasicOAuthData>(
db,
workspace_id,
service_name,
trigger.connection_path.as_deref(),
)
.await
.ok();
credentials.insert(trigger.connection_path.clone(), oauth_data);
}
let Some(Some(oauth_data)) = credentials.get(&trigger.connection_path) else {
continue;
};
if service_name == ServiceName::Nextcloud {
try_delete_nextcloud_webhook(
&oauth_data.base_url,
&oauth_data.access_token,
&trigger.external_id,
)
.await;
}
}
}
// For GitHub: try to delete webhooks on GitHub (best-effort)
if service_name == ServiceName::Github {
if let Ok(oauth_data) =
decrypt_oauth_data::<BasicOAuthData>(db, workspace_id, service_name).await
{
for trigger in &triggers {
} else {
try_delete_github_webhook(
db,
workspace_id,
@@ -342,15 +401,17 @@ async fn delete_triggers_for_service(db: &DB, workspace_id: &str, service_name:
}
}
}
// For Google: skip remote cleanup (watch channels expire naturally)
// Revoke the tokens the deleted rows actually named, not the ones listed above: a
// re-registration running concurrently mints a replacement, and leaving that alive would let a
// disconnected integration keep starting jobs.
let deleted_token_hashes = sqlx::query_scalar!(
"DELETE FROM native_trigger WHERE workspace_id = $1 AND service_name = $2 RETURNING webhook_token_hash",
"DELETE FROM native_trigger
WHERE workspace_id = $1 AND service_name = $2 AND ($3::text IS NULL OR connection_path = $3)
RETURNING webhook_token_hash",
workspace_id,
service_name as ServiceName
service_name as ServiceName,
connection_path,
)
.fetch_all(db)
.await
@@ -378,13 +439,12 @@ async fn delete_integration(
) -> JsonResult<String> {
require_admin(authed.is_admin, &workspace_id)?;
// Delete triggers first (needs OAuth data that cleanup_oauth_resource will remove)
delete_triggers_for_service(&db, &workspace_id, service_name).await;
// Delete triggers first: removing their registrations needs the connections cleaned up below
delete_triggers_for_service(&db, &workspace_id, service_name, None).await;
let mut tx = user_db.begin(&authed).await?;
// Clean up account+variable+resource
cleanup_oauth_resource(&mut *tx, &workspace_id, service_name).await;
cleanup_service_connections(&mut *tx, &workspace_id, service_name).await?;
let deleted = delete_workspace_integration(&mut *tx, &workspace_id, service_name).await?;
@@ -419,7 +479,6 @@ async fn delete_integration(
struct WorkspaceIntegrations {
service_name: ServiceName,
oauth_data: Option<sqlx::types::Json<WorkspaceOAuthConfig>>,
resource_path: Option<String>,
}
#[cfg(feature = "native_trigger")]
@@ -436,8 +495,7 @@ async fn list_integrations(
r#"
SELECT
oauth_data as "oauth_data: sqlx::types::Json<WorkspaceOAuthConfig>",
service_name as "service_name!: ServiceName",
resource_path
service_name as "service_name!: ServiceName"
FROM
workspace_integrations
WHERE
@@ -450,23 +508,14 @@ async fn list_integrations(
let key_value = integrations
.into_iter()
.map(|integration| {
(
integration.service_name,
(integration.oauth_data, integration.resource_path),
)
})
.map(|integration| (integration.service_name, integration.oauth_data))
.collect::<std::collections::HashMap<_, _>>();
use strum::IntoEnumIterator;
let integrations = ServiceName::iter()
.map(|service_name| {
let (oauth_data, resource_path) = key_value
.get(&service_name)
.cloned()
.map(|(od, rp)| (od, rp))
.unwrap_or((None, None));
WorkspaceIntegrations { service_name, oauth_data, resource_path }
let oauth_data = key_value.get(&service_name).cloned().flatten();
WorkspaceIntegrations { service_name, oauth_data }
})
.collect::<Vec<_>>();
@@ -475,12 +524,18 @@ async fn list_integrations(
Ok(Json(integrations))
}
/// Whether members can connect an account for this service: the workspace has an OAuth client,
/// or the instance shares one.
#[cfg(feature = "native_trigger")]
async fn integration_exist(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Path((workspace_id, service_name)): Path<(String, ServiceName)>,
) -> JsonResult<bool> {
if is_instance_sharing_enabled(&db, service_name).await? {
return Ok(Json(true));
}
let mut tx = user_db.begin(&authed).await?;
let exists = sqlx::query_scalar!(
r#"
@@ -504,8 +559,10 @@ async fn integration_exist(
#[cfg(feature = "native_trigger")]
#[derive(Debug, Deserialize)]
struct RedirectUri {
struct ConnectUrlBody {
redirect_uri: String,
/// Where to save the connection; defaults to `u/<caller>/native_<resource type>`.
resource_path: Option<String>,
}
#[cfg(feature = "native_trigger")]
@@ -514,7 +571,6 @@ struct OAuthCallbackBody {
redirect_uri: String,
code: String,
state: String,
resource_path: Option<String>,
}
#[cfg(feature = "native_trigger")]
@@ -531,60 +587,26 @@ async fn oauth_callback(
Path((workspace_id, service_name)): Path<(String, ServiceName)>,
Json(body): Json<OAuthCallbackBody>,
) -> JsonResult<String> {
require_admin(authed.is_admin, &workspace_id)?;
require_native_integration_use(&authed)?;
let state_was_valid = validate_signed_state(&db, &body.state, &workspace_id).await?;
let resource_path =
validate_signed_state(&db, &body.state, &workspace_id, service_name).await?;
if !state_was_valid {
return Err(Error::BadRequest(
"Invalid or expired state parameter".to_string(),
));
}
// Check if this integration uses instance-shared credentials
let existing_oauth_data = sqlx::query_scalar!(
r#"SELECT oauth_data FROM workspace_integrations
WHERE workspace_id = $1 AND service_name = $2"#,
workspace_id,
service_name as ServiceName
)
.fetch_optional(&db)
.await?
.flatten();
let is_instance_shared = existing_oauth_data
.as_ref()
.and_then(|v| v.get("instance_shared"))
.and_then(|v| v.as_bool())
.unwrap_or(false);
let oauth_config = if is_instance_shared {
get_instance_oauth_config(&db, service_name).await?
} else {
get_workspace_oauth_config::<WorkspaceOAuthConfig>(&db, &workspace_id, service_name).await?
};
let (oauth_config, is_instance_shared) =
resolve_oauth_client(&db, &workspace_id, service_name).await?;
let token_response =
exchange_code_for_token(&oauth_config, service_name, &body.code, &body.redirect_uri)
.await?;
let resource_path = body
.resource_path
.filter(|p| !p.is_empty())
.unwrap_or_else(|| {
format!(
"u/{}/native_{}",
authed.username,
service_name.resource_type()
)
});
let expires_in = token_response.expires_in.unwrap_or(3600);
// A user transaction, so the variable and resource policies decide whether the caller may
// write a connection at this path.
let mut tx = user_db.begin(&authed).await?;
// Clean up any previous account+variable+resource for this integration
cleanup_oauth_resource(&mut *tx, &workspace_id, service_name).await;
// Reconnecting at a path replaces the connection there; triggers using it keep working.
cleanup_connection(&mut *tx, &workspace_id, &resource_path).await?;
// 1. Create account record for token refresh
let account_id = sqlx::query_scalar!(
@@ -659,25 +681,22 @@ async fn oauth_callback(
.await
.map_err(|e| Error::InternalErr(format!("Failed to create resource: {}", e)))?;
// 4. Store config + resource_path in workspace_integrations (no tokens).
// For instance-shared integrations, store the flag instead of credentials.
let stored_data = if is_instance_shared {
json!({
"instance_shared": true,
"base_url": "",
})
} else {
to_value(&oauth_config).unwrap()
};
store_workspace_integration(
&mut *tx,
&authed,
&workspace_id,
service_name,
stored_data,
Some(&resource_path),
)
.await?;
// 4. The first connection made with the instance's client records that in the workspace.
// Token refresh reads the client from this row, so it must exist for every connection.
if is_instance_shared {
sqlx::query!(
"INSERT INTO workspace_integrations (workspace_id, service_name, oauth_data, created_by)
VALUES ($1, $2, $3, $4)
ON CONFLICT (workspace_id, service_name) DO UPDATE SET oauth_data = EXCLUDED.oauth_data
WHERE workspace_integrations.oauth_data IS NULL",
workspace_id,
service_name as ServiceName,
json!({ "instance_shared": true, "base_url": "" }),
authed.username,
)
.execute(&mut *tx)
.await?;
}
audit_log(
&mut *tx,
@@ -685,17 +704,14 @@ async fn oauth_callback(
&format!("workspace_integrations.{}.connect", service_name),
ActionKind::Create,
&workspace_id,
Some(&format!("Connected {} integration via OAuth", service_name)),
Some(&resource_path),
None,
)
.await?;
tx.commit().await?;
Ok(Json(format!(
"{} integration connected successfully via OAuth",
service_name
)))
Ok(Json(resource_path))
}
/// Token response from OAuth token exchange
@@ -807,7 +823,6 @@ pub async fn create_workspace_integration(
&workspace_id,
service_name,
to_value(oauth_data).unwrap(),
None,
)
.await?;
@@ -816,16 +831,6 @@ pub async fn create_workspace_integration(
Ok(())
}
#[cfg(feature = "native_trigger")]
#[inline]
async fn get_workspace_oauth_config_as_oauth_config(
db: &DB,
workspace_id: &str,
service_name: ServiceName,
) -> Result<WorkspaceOAuthConfig> {
get_workspace_oauth_config::<WorkspaceOAuthConfig>(db, workspace_id, service_name).await
}
#[cfg(feature = "native_trigger")]
fn build_authorization_url(
config: &WorkspaceOAuthConfig,
@@ -856,73 +861,82 @@ fn build_authorization_url(
format!("{}?{}", base_auth_url, query_string)
}
/// Remove the connection at `path`: its variable, the account behind it, and its resource.
#[cfg(feature = "native_trigger")]
pub async fn cleanup_oauth_resource(
pub async fn cleanup_connection(
tx: &mut sqlx::PgConnection,
workspace_id: &str,
path: &str,
) -> Result<()> {
let account_id = sqlx::query_scalar!(
"DELETE FROM variable WHERE workspace_id = $1 AND path = $2 RETURNING account",
workspace_id,
path,
)
.fetch_optional(&mut *tx)
.await?
.flatten();
if let Some(account_id) = account_id {
sqlx::query!(
"DELETE FROM account WHERE workspace_id = $1 AND id = $2 AND is_workspace_integration = true",
workspace_id,
account_id,
)
.execute(&mut *tx)
.await?;
}
sqlx::query!(
"DELETE FROM resource WHERE workspace_id = $1 AND path = $2",
workspace_id,
path,
)
.execute(&mut *tx)
.await?;
Ok(())
}
/// Remove every connection of a service in the workspace.
#[cfg(feature = "native_trigger")]
pub async fn cleanup_service_connections(
tx: &mut sqlx::PgConnection,
workspace_id: &str,
service_name: ServiceName,
) {
// Look up the stored resource_path from workspace_integrations
let stored_resource_path: Option<String> = sqlx::query_scalar!(
r#"SELECT resource_path FROM workspace_integrations WHERE workspace_id = $1 AND service_name = $2"#,
workspace_id,
service_name as ServiceName,
)
.fetch_optional(&mut *tx)
.await
.ok()
.flatten()
.flatten();
// Find and delete any existing account+variable+resource for this integration
let account_ids: Vec<i32> = sqlx::query_scalar!(
) -> Result<()> {
let account_ids = sqlx::query_scalar!(
"DELETE FROM account WHERE workspace_id = $1 AND client = $2 AND is_workspace_integration = true RETURNING id",
workspace_id,
service_name.as_str(),
)
.fetch_all(&mut *tx)
.await
.unwrap_or_default();
.await?;
if !account_ids.is_empty() {
// Delete variables linked to these accounts
let _ = sqlx::query!(
"DELETE FROM variable WHERE workspace_id = $1 AND account = ANY($2)",
workspace_id,
&account_ids,
)
.execute(&mut *tx)
.await;
}
let paths = sqlx::query_scalar!(
"DELETE FROM variable WHERE workspace_id = $1 AND account = ANY($2) RETURNING path",
workspace_id,
&account_ids,
)
.fetch_all(&mut *tx)
.await?;
// Delete resource by exact stored path, or fall back to legacy pattern
let resource_type = service_name.resource_type();
if let Some(ref path) = stored_resource_path {
let _ = sqlx::query!(
"DELETE FROM resource WHERE workspace_id = $1 AND path = $2",
workspace_id,
path,
)
.execute(&mut *tx)
.await;
} else {
// Legacy fallback for integrations created before user-chosen paths
let _ = sqlx::query!(
"DELETE FROM resource WHERE workspace_id = $1 AND resource_type = $2 AND path LIKE 'u/%/native_%'",
workspace_id,
resource_type,
)
.execute(&mut *tx)
.await;
}
sqlx::query!(
"DELETE FROM resource WHERE workspace_id = $1 AND path = ANY($2)",
workspace_id,
&paths,
)
.execute(&mut *tx)
.await?;
Ok(())
}
/// Check if the instance admin has enabled sharing of OAuth credentials for a given service.
/// Currently only supported for Google (gworkspace).
#[cfg(feature = "native_trigger")]
async fn is_instance_sharing_enabled(db: &DB, service_name: ServiceName) -> Result<bool> {
// Only Google supports instance sharing for now
if service_name != ServiceName::Google {
// Each Nextcloud client belongs to one Nextcloud server, which is a workspace choice.
if service_name == ServiceName::Nextcloud {
return Ok(false);
}
@@ -931,7 +945,7 @@ async fn is_instance_sharing_enabled(db: &DB, service_name: ServiceName) -> Resu
None => return Ok(false),
};
let key = service_name.resource_type(); // "gworkspace"
let key = service_name.resource_type();
let entry = match oauths_value.get(key) {
Some(v) => v,
None => return Ok(false),
@@ -986,36 +1000,73 @@ async fn check_instance_sharing_available(
}
#[cfg(feature = "native_trigger")]
async fn generate_instance_connect_url(
async fn list_connections(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Path((workspace_id, service_name)): Path<(String, ServiceName)>,
) -> JsonResult<Vec<Connection>> {
let mut tx = user_db.begin(&authed).await?;
let connections = list_usable_connections(&mut tx, &workspace_id, service_name).await?;
tx.commit().await?;
Ok(Json(connections))
}
#[cfg(feature = "native_trigger")]
#[derive(Debug, Deserialize)]
struct ConnectionPathQuery {
path: String,
}
/// Disconnect one account: its triggers are deleted with it, since nothing else can act on
/// their registrations.
#[cfg(feature = "native_trigger")]
async fn delete_connection(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Path((workspace_id, service_name)): Path<(String, ServiceName)>,
Json(RedirectUri { redirect_uri }): Json<RedirectUri>,
Query(ConnectionPathQuery { path }): Query<ConnectionPathQuery>,
) -> JsonResult<String> {
require_admin(authed.is_admin, &workspace_id)?;
let instance_config = get_instance_oauth_config(&db, service_name).await?;
// Store a marker in workspace_integrations — NOT the actual credentials.
// The callback and token refresh will read credentials from global settings
// when they see instance_shared=true.
let mut tx = user_db.begin(&authed).await?;
crate::store_workspace_integration(
&mut tx,
require_is_writer(
&authed,
&path,
&workspace_id,
service_name,
json!({ "instance_shared": true, "base_url": "" }),
db.clone(),
"SELECT extra_perms FROM variable WHERE path = $1 AND workspace_id = $2",
"variable",
)
.await?;
let mut tx = user_db.begin(&authed).await?;
if !list_usable_connections(&mut tx, &workspace_id, service_name)
.await?
.iter()
.any(|c| c.path == path)
{
return Err(Error::NotFound(format!(
"No {service_name} connection at {path}"
)));
}
// Before the connection goes: removing the registrations needs its token.
delete_triggers_for_service(&db, &workspace_id, service_name, Some(&path)).await;
cleanup_connection(&mut *tx, &workspace_id, &path).await?;
audit_log(
&mut *tx,
&authed,
&format!("workspace_integrations.{}.disconnect", service_name),
ActionKind::Delete,
&workspace_id,
Some(&path),
None,
)
.await?;
tx.commit().await?;
// Generate signed state and build authorization URL
let state = generate_signed_state(&db, &workspace_id, service_name).await?;
let auth_url = build_authorization_url(&instance_config, service_name, &state, &redirect_uri);
Ok(Json(auth_url))
Ok(Json(format!("Disconnected {path}")))
}
#[cfg(feature = "native_trigger")]
@@ -1032,9 +1083,10 @@ pub fn workspaced_service() -> Router {
"/{service_name}/instance_sharing_available",
get(check_instance_sharing_available),
)
.route("/{service_name}/connections", get(list_connections))
.route(
"/{service_name}/generate_instance_connect_url",
post(generate_instance_connect_url),
"/{service_name}/connections/delete",
delete(delete_connection),
)
.route("/{service_name}/delete", delete(delete_integration))
.route("/{service_name}/callback", post(oauth_callback));
+2
View File
@@ -281,6 +281,8 @@ export async function pushNativeTrigger(
is_flow: localTrigger.is_flow,
service_config: localTrigger.service_config,
summary: localTrigger.summary,
// Used on create only; an update keeps the connection the webhook was registered under.
connection_path: localTrigger.connection_path,
};
if (remoteTrigger) {
@@ -0,0 +1,110 @@
<script lang="ts">
import { WorkspaceIntegrationService } from '$lib/gen'
import type { NativeServiceName, NativeTriggerConnection } from '$lib/gen/types.gen'
import { Button } from '$lib/components/common'
import Alert from '$lib/components/common/alert/Alert.svelte'
import Select from '$lib/components/select/Select.svelte'
import { sendUserToast } from '$lib/toast'
import { Plug } from 'lucide-svelte'
import { useOperatingWorkspace } from '$lib/components/operatingWorkspace.svelte'
import { connectNativeAccount, getServiceConfig } from './utils'
interface Props {
service: NativeServiceName
value: string | undefined
/** An existing trigger keeps the connection its webhook was registered under. */
locked?: boolean
disabled?: boolean
}
let { service, value = $bindable(), locked = false, disabled = false }: Props = $props()
const operatingWorkspace = useOperatingWorkspace()
const displayName = $derived(getServiceConfig(service)?.serviceDisplayName ?? service)
let connections = $state<NativeTriggerConnection[] | undefined>(undefined)
let connecting = $state(false)
async function load() {
try {
connections = await WorkspaceIntegrationService.listNativeTriggerConnections({
workspace: $operatingWorkspace!,
serviceName: service
})
} catch (err: any) {
connections = []
sendUserToast(`Failed to load ${displayName} connections: ${err.body ?? err.message}`, true)
}
if (!locked && value === undefined && connections.length > 0) {
value = connections[0].path
}
}
$effect(() => {
service
$operatingWorkspace
load()
})
async function connect() {
connecting = true
try {
const path = await connectNativeAccount($operatingWorkspace!, service)
await load()
value = path
sendUserToast(`${displayName} account connected`)
} catch (err: any) {
sendUserToast(`Could not connect ${displayName}: ${err.body ?? err.message}`, true)
} finally {
connecting = false
}
}
const items = $derived.by(() => {
const list = (connections ?? []).map((c) => ({ label: c.path, value: c.path }))
if (value && !list.some((i) => i.value === value)) {
list.push({ label: value, value })
}
return list
})
const unusable = $derived(
connections !== undefined && !!value && !connections.some((c) => c.path === value)
)
</script>
<div class="flex flex-col gap-2">
<div class="flex flex-row gap-2 items-center">
<Select
class="grow"
{items}
bind:value
disabled={disabled || locked}
loading={connections === undefined}
placeholder="Connect an account to create this trigger"
noItemsMsg="No {displayName} account connected yet"
/>
{#if !locked}
<Button
unifiedSize="md"
variant="default"
startIcon={{ icon: Plug }}
onclick={connect}
disabled={disabled || connecting}
loading={connecting}
>
Connect account
</Button>
{/if}
</div>
{#if locked}
<p class="text-2xs text-secondary">
The webhook is registered under this account. To use another one, create a new trigger.
</p>
{/if}
{#if unusable}
<Alert type="warning" title="You cannot use this connection" size="xs">
Saving acts as the account at {value}. Ask its owner for read access to it, or create a new
trigger with your own account.
</Alert>
{/if}
</div>
@@ -0,0 +1,69 @@
<script lang="ts">
import { onMount } from 'svelte'
import { page } from '$app/state'
import { WorkspaceIntegrationService } from '$lib/gen'
import type { NativeServiceName } from '$lib/gen/types.gen'
import Alert from '$lib/components/common/alert/Alert.svelte'
import { Loader2 } from 'lucide-svelte'
import { postNativeOAuthResult } from './utils'
interface Props {
/** The exact redirect URI the authorization request was made with. */
redirectUri: string
}
let { redirectUri }: Props = $props()
let outcome = $state<{ ok: true; path: string } | { ok: false; error: string } | undefined>()
// The signed state carries the workspace and service it was issued for, so this page works
// whatever workspace is selected and wherever the provider sends the user back.
function decodeState(state: string): { workspace: string; service: NativeServiceName } {
const payload = atob(state.split(':')[0].replace(/-/g, '+').replace(/_/g, '/'))
const [workspace, service] = payload.split(':')
return { workspace, service: service as NativeServiceName }
}
async function finish(): Promise<{ ok: true; path: string } | { ok: false; error: string }> {
const error = page.url.searchParams.get('error')
const code = page.url.searchParams.get('code')
const state = page.url.searchParams.get('state')
if (error || !code || !state) {
return { ok: false, error: error ?? 'The provider returned no authorization code' }
}
try {
const { workspace, service } = decodeState(state)
const path = await WorkspaceIntegrationService.nativeTriggerServiceCallback({
workspace,
serviceName: service,
requestBody: { code, state, redirect_uri: redirectUri }
})
return { ok: true, path }
} catch (err: any) {
return { ok: false, error: err.body ?? err.message ?? String(err) }
}
}
onMount(async () => {
outcome = await finish()
postNativeOAuthResult(outcome)
if (window.opener) {
window.close()
}
})
</script>
<div class="max-w-lg mx-auto mt-16 px-4">
{#if outcome === undefined}
<div class="flex items-center gap-2 text-sm text-secondary">
<Loader2 class="animate-spin" size={16} />
Finishing the connection...
</div>
{:else if outcome.ok}
<Alert type="success" title="Account connected">
Saved as {outcome.path}. You can close this window.
</Alert>
{:else}
<Alert type="error" title="Could not connect the account">{outcome.error}</Alert>
{/if}
</div>
@@ -6,8 +6,10 @@
validateCommonFields,
getServiceConfig,
getTemplatePath,
saveNativeTriggerFromCfg
saveNativeTriggerFromCfg,
setNativeConnectionContext
} from './utils'
import NativeConnectionPicker from './NativeConnectionPicker.svelte'
import { usedTriggerKinds } from '$lib/stores'
import { emptyString, sendUserToast } from '$lib/utils'
import { useEditRights, useTriggerLock } from '$lib/operatorWriteRights'
@@ -130,6 +132,8 @@
let externalError = $state<string | undefined>(undefined)
let retryEdit = $state<(() => void) | undefined>(undefined)
let enabled = $state(true)
let connectionPath = $state<string | undefined>(undefined)
setNativeConnectionContext(() => connectionPath)
export function openNew(
nis_flow?: boolean,
@@ -161,6 +165,7 @@
externalError = undefined
retryEdit = undefined
enabled = true
connectionPath = undefined
}
export function openRecreate(nativeTrigger: ExtendedNativeTrigger) {
@@ -190,6 +195,7 @@
externalError = undefined
retryEdit = undefined
enabled = nativeTrigger.enabled
connectionPath = nativeTrigger.connection_path ?? undefined
}
export async function openEdit(
@@ -224,6 +230,7 @@
initialScriptPath = ''
summary = ''
enabled = true
connectionPath = undefined
try {
const fullTrigger = await NativeTriggerService.getNativeTrigger({
@@ -241,6 +248,7 @@
externalData = fullTrigger.external_data
externalError = fullTrigger.external_error ?? undefined
enabled = fullTrigger.enabled
connectionPath = fullTrigger.connection_path ?? undefined
// Apply default values if provided (for draft triggers)
if (defaultValues) {
@@ -268,7 +276,8 @@
script_path: scriptPath,
is_flow: isFlow,
service_config: serviceConfig,
summary: summary !== '' ? summary : undefined
summary: summary !== '' ? summary : undefined,
connection_path: connectionPath
}
}
@@ -313,7 +322,8 @@
loadingForm ||
!can_write ||
!hasChanged ||
loadError !== undefined
loadError !== undefined ||
(isNew && !connectionPath)
)
const saveCfg = $derived.by(getSaveCfg)
@@ -562,6 +572,17 @@
</Section>
{/if}
{#if !loadingConfig}
<Section label="{serviceInfo?.serviceDisplayName} account">
<NativeConnectionPicker
{service}
bind:value={connectionPath}
locked={!isNew}
disabled={loading || !can_write}
/>
</Section>
{/if}
{#if loadingConfig}
<Section label="{serviceInfo?.serviceDisplayName} configuration">
<div class="flex items-center gap-2 text-secondary text-xs">
@@ -569,20 +590,25 @@
Loading configuration from {serviceInfo?.serviceDisplayName}...
</div>
</Section>
{:else if isNew && !connectionPath}
<!-- The pickers browse the service as the chosen account, so there is nothing to
configure until one is chosen. -->
{:else if ServiceFormComponent}
<ServiceFormComponent
bind:loading={loadingForm}
bind:this={serviceFormRef}
bind:serviceConfig
bind:errors
{externalData}
disabled={loading || loadingConfig || !can_write}
path={scriptPath}
{isFlow}
token=""
triggerTokens={undefined}
scopes={[]}
/>
{#key connectionPath}
<ServiceFormComponent
bind:loading={loadingForm}
bind:this={serviceFormRef}
bind:serviceConfig
bind:errors
{externalData}
disabled={loading || loadingConfig || !can_write}
path={scriptPath}
{isFlow}
token=""
triggerTokens={undefined}
scopes={[]}
/>
{/key}
{:else}
<Section label="{serviceInfo?.serviceDisplayName} configuration">
<div class="text-red-500 text-xs space-y-2">
@@ -7,8 +7,10 @@
import { sendUserToast } from '$lib/toast'
import { Loader2 } from 'lucide-svelte'
import { useOperatingWorkspace } from '$lib/components/operatingWorkspace.svelte'
import { useNativeConnection } from '../../utils'
const operatingWorkspace = useOperatingWorkspace()
const connectionPath = useNativeConnection()
interface Props {
serviceConfig: Record<string, any>
@@ -66,7 +68,8 @@
loading = true
try {
repos = await NativeTriggerService.listGithubRepos({
workspace: $operatingWorkspace
workspace: $operatingWorkspace,
connectionPath: connectionPath()
})
} catch (err: any) {
console.error('Failed to load GitHub repositories:', err)
@@ -6,8 +6,10 @@
import { Loader2, X, RefreshCw } from 'lucide-svelte'
import GoogleCalendarIcon from '$lib/components/icons/GoogleCalendarIcon.svelte'
import { useOperatingWorkspace } from '$lib/components/operatingWorkspace.svelte'
import { useNativeConnection } from '../../utils'
const operatingWorkspace = useOperatingWorkspace()
const connectionPath = useNativeConnection()
interface Props {
calendarId: string
@@ -26,7 +28,8 @@
loading = true
try {
calendars = await NativeTriggerService.listGoogleCalendars({
workspace: $operatingWorkspace
workspace: $operatingWorkspace,
connectionPath: connectionPath()
})
if (calendarId && !calendarName) {
const found = calendars.find((c) => c.id === calendarId)
@@ -9,8 +9,10 @@
import { Debounced, watch } from 'runed'
import { untrack } from 'svelte'
import { useOperatingWorkspace } from '$lib/components/operatingWorkspace.svelte'
import { useNativeConnection } from '../../utils'
const operatingWorkspace = useOperatingWorkspace()
const connectionPath = useNativeConnection()
interface Props {
resourceId: string
@@ -53,6 +55,7 @@
try {
const params: Parameters<typeof NativeTriggerService.listGoogleDriveFiles>[0] = {
workspace: $operatingWorkspace,
connectionPath: connectionPath(),
pageToken
}
@@ -98,7 +101,8 @@
try {
sharedDrives = await NativeTriggerService.listGoogleSharedDrives({
workspace: $operatingWorkspace
workspace: $operatingWorkspace,
connectionPath: connectionPath()
})
} catch (err: any) {
sendUserToast(`Failed to load shared drives: ${err.body || err.message}`, true)
@@ -6,10 +6,11 @@
import SchemaForm from '$lib/components/SchemaForm.svelte'
import Section from '$lib/components/Section.svelte'
import { Loader2 } from 'lucide-svelte'
import { getNextcloudSchema } from '../../utils'
import { getNextcloudSchema, useNativeConnection } from '../../utils'
import { useOperatingWorkspace } from '$lib/components/operatingWorkspace.svelte'
const operatingWorkspace = useOperatingWorkspace()
const connectionPath = useNativeConnection()
interface Props {
serviceConfig: Record<string, any>
@@ -49,7 +50,8 @@
eventsError = undefined
try {
const events = await NativeTriggerService.listNextCloudEvents({
workspace: $operatingWorkspace!
workspace: $operatingWorkspace!,
connectionPath: connectionPath()
})
availableEvents = events
serviceSchema = getNextcloudSchema(events)
@@ -2,8 +2,84 @@ import type { NativeServiceName, NativeTrigger, NativeTriggerData } from '$lib/g
import { isCloudHosted } from '$lib/cloud'
import { NativeTriggerService, WorkspaceIntegrationService } from '$lib/gen'
import { get, type Writable } from 'svelte/store'
import { getContext, setContext } from 'svelte'
import { sendUserToast } from '$lib/toast'
const CONNECTION_CONTEXT = Symbol('nativeTriggerConnection')
/** Makes the connection a trigger editor acts through available to the pickers of its form. */
export function setNativeConnectionContext(connectionPath: () => string | undefined) {
setContext(CONNECTION_CONTEXT, connectionPath)
}
/** The connection the service pickers browse through; undefined lets the server pick. */
export function useNativeConnection(): () => string | undefined {
return getContext<(() => string | undefined) | undefined>(CONNECTION_CONTEXT) ?? (() => undefined)
}
const CONNECT_MESSAGE = 'native_trigger_oauth'
export type NativeOAuthMessage =
| { type: typeof CONNECT_MESSAGE; ok: true; path: string }
| { type: typeof CONNECT_MESSAGE; ok: false; error: string }
/** The page the provider redirects to, as registered on the workspace's OAuth apps. */
export function nativeOAuthRedirectUri(service: NativeServiceName): string {
return `${window.location.origin}/workspace_settings?tab=native_triggers&service=${service}`
}
/** Hands the outcome of a connect popup back to the page that opened it. */
export function postNativeOAuthResult(
result: { ok: true; path: string } | { ok: false; error: string }
) {
const message = { type: CONNECT_MESSAGE, ...result } as NativeOAuthMessage
window.opener?.postMessage(message, window.location.origin)
}
/**
* Connect an account in a popup and resolve with the path of the saved connection. The
* connection is saved at `resourcePath`, or at the caller's default path when omitted.
*/
export async function connectNativeAccount(
workspace: string,
service: NativeServiceName,
resourcePath?: string
): Promise<string> {
const authUrl = await WorkspaceIntegrationService.generateNativeTriggerServiceConnectUrl({
workspace,
serviceName: service,
requestBody: { redirect_uri: nativeOAuthRedirectUri(service), resource_path: resourcePath }
})
const popup = window.open(authUrl, '_blank', 'popup,width=600,height=750')
if (!popup) {
throw new Error('The connection window was blocked. Allow popups for this site and retry.')
}
return new Promise((resolve, reject) => {
const onMessage = (event: MessageEvent<NativeOAuthMessage>) => {
if (event.origin !== window.location.origin || event.data?.type !== CONNECT_MESSAGE) return
cleanup()
if (event.data.ok) resolve(event.data.path)
else reject(new Error(event.data.error))
}
// The callback page posts its result and then closes itself, so a closed window may still
// have a message in flight: give it a moment before treating the close as a cancel.
const closedPoll = setInterval(() => {
if (popup.closed) {
clearInterval(closedPoll)
setTimeout(() => {
cleanup()
reject(new Error('The connection window was closed before finishing'))
}, 1000)
}
}, 500)
function cleanup() {
window.removeEventListener('message', onMessage)
clearInterval(closedPoll)
}
window.addEventListener('message', onMessage)
})
}
export interface NativeTriggerConfig {
readonly serviceDisplayName: string
readonly serviceKey: NativeServiceName
@@ -216,8 +292,13 @@ export async function saveNativeTriggerFromCfg(
service_config: triggerCfg.service_config,
summary: triggerCfg.summary
}
// Only a create can set it: an update ignores the field, and `setenabled` owns it thereafter.
const createBody: NativeTriggerData = { ...requestBody, enabled: triggerCfg.enabled ?? true }
// Only a create can set these: an update ignores both, `setenabled` owns `enabled` thereafter,
// and the connection stays the one the webhook was registered under.
const createBody: NativeTriggerData = {
...requestBody,
enabled: triggerCfg.enabled ?? true,
connection_path: triggerCfg.connection_path
}
const serviceName = NATIVE_TRIGGER_SERVICES[service].serviceDisplayName
@@ -4,22 +4,27 @@
import { Button, Alert } from '$lib/components/common'
import Skeleton from '$lib/components/common/skeleton/Skeleton.svelte'
import SettingsPageHeader from '$lib/components/settings/SettingsPageHeader.svelte'
import { Check, X, ExternalLink, Cog, Plug } from 'lucide-svelte'
import { X, ExternalLink, Cog, Plug } from 'lucide-svelte'
import NextcloudIcon from '$lib/components/icons/NextcloudIcon.svelte'
import GithubIcon from '$lib/components/icons/GithubIcon.svelte'
import GoogleIcon from '$lib/components/icons/GoogleIcon.svelte'
import { WorkspaceIntegrationService, type NativeServiceName } from '$lib/gen'
import {
WorkspaceIntegrationService,
type NativeServiceName,
type NativeTriggerConnection
} from '$lib/gen'
import ClipboardPanel from '$lib/components/details/ClipboardPanel.svelte'
import OAuthClientConfig from './OAuthClientConfig.svelte'
import ConfirmationModal from '../common/confirmationModal/ConfirmationModal.svelte'
import { createAsyncConfirmationModal } from '../common/confirmationModal/asyncConfirmationModal.svelte'
import Path from '$lib/components/Path.svelte'
import { page } from '$app/state'
import { goto } from '$app/navigation'
import {
connectNativeAccount,
nativeOAuthRedirectUri
} from '$lib/components/triggers/native/utils'
interface WorkspaceIntegration {
service_name: string
resource_path?: string
oauth_data: {
client_id: string
client_secret: string
@@ -89,16 +94,13 @@
}
let integrations = $state<WorkspaceIntegration[]>([])
let connections = $state<Record<string, NativeTriggerConnection[]>>({})
let loading = $state(false)
let connecting = $state<string | null>(null)
let showingConfig = $state<string | null>(null)
let instanceSharingAvailable = $state<Record<string, boolean>>({})
let pendingCallback = $state<{
serviceName: NativeServiceName
code: string
state: string
workspace: string
} | null>(null)
// The service whose "where to save the connection" form is open.
let choosingPathFor = $state<string | null>(null)
let resourcePath = $state<string | undefined>(undefined)
let pathError = $state<string | undefined>(undefined)
let confirmationModal = createAsyncConfirmationModal()
@@ -113,9 +115,9 @@
})
integrations = response.map((item) => ({
service_name: item.service_name,
resource_path: item.resource_path ?? undefined,
oauth_data: item.oauth_data || null
}))
await Promise.all(Object.keys(supportedServices).map(loadConnections))
} catch (err: any) {
console.error('Failed to load workspace integrations:', err)
sendUserToast(`Failed to load integrations: ${err.message}`, true)
@@ -124,48 +126,84 @@
}
}
async function loadConnections(serviceName: string) {
if (!$workspaceStore) return
try {
connections[serviceName] = await WorkspaceIntegrationService.listNativeTriggerConnections({
workspace: $workspaceStore,
serviceName: serviceName as NativeServiceName
})
} catch {
connections[serviceName] = []
}
}
async function deleteIntegration(serviceName: string) {
if (!$workspaceStore) return
const displayName = supportedServices[serviceName]?.displayName ?? serviceName
const confirmed = await confirmationModal.ask({
title: `Disconnect ${displayName}?`,
confirmationText: 'Disconnect',
children: `This will delete all ${displayName} triggers associated with this integration. This action cannot be undone.`
title: `Remove the ${displayName} integration?`,
confirmationText: 'Remove',
children: `This removes the OAuth app configuration, every connected ${displayName} account and all ${displayName} triggers. This action cannot be undone.`
})
if (!confirmed) return
try {
await WorkspaceIntegrationService.deleteNativeTriggerService({
workspace: $workspaceStore,
serviceName: serviceName as any
serviceName: serviceName as NativeServiceName
})
sendUserToast(`${displayName} disconnected successfully`)
sendUserToast(`${displayName} integration removed`)
loadIntegrations()
} catch (err: any) {
sendUserToast(`Failed to disconnect ${displayName}: ${err.message}`, true)
sendUserToast(`Failed to remove ${displayName}: ${err.body ?? err.message}`, true)
}
}
async function connectService(serviceName: string, redirectUri: string) {
async function disconnect(serviceName: string, path: string) {
if (!$workspaceStore) return
const displayName = supportedServices[serviceName]?.displayName ?? serviceName
const confirmed = await confirmationModal.ask({
title: `Disconnect ${path}?`,
confirmationText: 'Disconnect',
children: `This deletes the ${displayName} triggers created with this account. This action cannot be undone.`
})
if (!confirmed) return
try {
await WorkspaceIntegrationService.deleteNativeTriggerConnection({
workspace: $workspaceStore,
serviceName: serviceName as NativeServiceName,
path
})
sendUserToast(`${path} disconnected`)
loadConnections(serviceName)
} catch (err: any) {
sendUserToast(`Failed to disconnect ${path}: ${err.body ?? err.message}`, true)
}
}
async function connect(serviceName: string) {
if (!$workspaceStore) return
connecting = serviceName
try {
const auth_url = await WorkspaceIntegrationService.generateNativeTriggerServiceConnectUrl({
workspace: $workspaceStore,
serviceName: serviceName as any,
requestBody: { redirect_uri: redirectUri }
})
if (auth_url) {
window.location.href = auth_url
}
const path = await connectNativeAccount(
$workspaceStore,
serviceName as NativeServiceName,
resourcePath
)
sendUserToast(`${supportedServices[serviceName]?.displayName} account connected as ${path}`)
choosingPathFor = null
await loadConnections(serviceName)
} catch (err: any) {
sendUserToast(
`Failed to connect ${supportedServices[serviceName]?.displayName}: ${err.message}`,
`Failed to connect ${supportedServices[serviceName]?.displayName}: ${err.body ?? err.message}`,
true
)
} finally {
connecting = null
}
}
@@ -176,7 +214,7 @@
try {
await WorkspaceIntegrationService.createNativeTriggerService({
workspace: $workspaceStore,
serviceName: serviceName as any,
serviceName: serviceName as NativeServiceName,
requestBody: oauthData
})
sendUserToast(
@@ -207,33 +245,9 @@
}
}
async function connectWithInstanceCredentials(serviceName: string) {
if (!$workspaceStore) return
connecting = serviceName
try {
const redirectUri = getRedirectUri(serviceName)
const auth_url = await WorkspaceIntegrationService.generateInstanceConnectUrl({
workspace: $workspaceStore,
serviceName: serviceName as NativeServiceName,
requestBody: { redirect_uri: redirectUri }
})
if (auth_url) {
window.location.href = auth_url
}
} catch (err: any) {
sendUserToast(
`Failed to connect ${supportedServices[serviceName]?.displayName}: ${err.message}`,
true
)
connecting = null
}
}
function isConfigured(integration: WorkspaceIntegration): boolean {
if (integration.oauth_data === null) return false
if (integration.oauth_data.instance_shared) return true
/** Whether the workspace has its own OAuth app for the service. */
function hasWorkspaceApp(integration: WorkspaceIntegration | null): boolean {
if (!integration?.oauth_data || integration.oauth_data.instance_shared) return false
const serviceConfig = supportedServices[integration.service_name]
const needsBaseUrl = serviceConfig?.requiresBaseUrl !== false
return (
@@ -243,82 +257,10 @@
)
}
function isConnected(integration: WorkspaceIntegration): boolean {
if (integration.oauth_data?.instance_shared) {
return !!integration.resource_path
}
return isConfigured(integration) && !!integration.resource_path
}
function getIntegrationByService(serviceName: string): WorkspaceIntegration | null {
return integrations.find((integration) => integration.service_name === serviceName) || null
}
function handleOAuthCallback(
workspace: string,
serviceName: NativeServiceName,
code: string,
state: string
) {
// Phase 1: store callback params and clean URL — don't call backend yet
pendingCallback = { serviceName, code, state, workspace }
// Pre-populate resource path from existing integration (for reconnect)
const integration = getIntegrationByService(serviceName)
resourcePath = integration?.resource_path ?? undefined
const url = new URL(page.url)
url.searchParams.delete('code')
url.searchParams.delete('state')
url.searchParams.delete('service')
goto(url.toString(), { replaceState: true, noScroll: true, keepFocus: true })
}
async function finalizePendingCallback() {
if (!pendingCallback) return
const { serviceName, code, state, workspace } = pendingCallback
try {
const redirectUri = getRedirectUri(serviceName)
await WorkspaceIntegrationService.nativeTriggerServiceCallback({
serviceName,
workspace,
requestBody: {
code,
state,
redirect_uri: redirectUri,
resource_path: resourcePath
}
})
sendUserToast(`${supportedServices[serviceName]?.displayName} connected successfully!`)
await loadIntegrations()
} catch (err: any) {
sendUserToast(`Failed to complete OAuth connection: ${err.message}`, true)
} finally {
pendingCallback = null
resourcePath = undefined
pathError = undefined
}
}
$effect(() => {
if (
page.url.searchParams.has('code') &&
page.url.searchParams.has('state') &&
page.url.searchParams.has('service') &&
$workspaceStore
) {
const service = page.url.searchParams.get('service')! as NativeServiceName
const code = page.url.searchParams.get('code')!
const state = page.url.searchParams.get('state')!
handleOAuthCallback($workspaceStore, service, code, state)
}
})
function getRedirectUri(serviceName: string): string {
return `${window.location.origin}/workspace_settings?tab=native_triggers&service=${serviceName}`
}
$effect(() => {
if ($workspaceStore) {
loadIntegrations()
@@ -330,48 +272,11 @@
<div class="flex flex-col">
<SettingsPageHeader
title="Native Triggers"
description="Connect your workspace to external services for native triggers and enhanced functionality. These connections are shared across all workspace members and are required for native triggers to work."
description="Configure the OAuth app of each service. Members then connect their own account, or one shared through a folder, and each trigger acts as the account it was created with."
link="https://www.windmill.dev/docs/core_concepts/native_triggers"
/>
{#if pendingCallback}
{@const serviceName = pendingCallback.serviceName}
{@const config = supportedServices[serviceName]}
<div class="border border-gray-200 dark:border-gray-700 rounded-md p-4 bg-surface-tertiary">
<div class="text-sm font-semibold text-emphasis mb-2">
Save {config?.displayName ?? serviceName} credentials as resource
</div>
<div class="text-xs text-secondary mb-3">
Choose where to save the OAuth resource. This resource will store the access token for the
integration.
</div>
<Path
kind="resource"
initialPath={resourcePath ?? ''}
namePlaceholder={'native_' + serviceName}
bind:path={resourcePath}
bind:error={pathError}
/>
<div class="flex gap-2 mt-3">
<Button
variant="accent"
disabled={!resourcePath || !!pathError}
onclick={finalizePendingCallback}
>
Save
</Button>
<Button
onclick={() => {
pendingCallback = null
resourcePath = undefined
pathError = undefined
}}
>
Cancel
</Button>
</div>
</div>
{:else if loading}
{#if loading}
<div class="space-y-4">
{#each new Array(3) as _}
<Skeleton layout={[[6], 0.4]} />
@@ -382,12 +287,12 @@
{#each Object.entries(supportedServices) as [serviceName, config]}
{@const integration = getIntegrationByService(serviceName)}
{@const isConnecting = connecting === serviceName}
{@const isOAuthConfigured = integration && isConfigured(integration)}
{@const isServiceConnected = integration && isConnected(integration)}
{@const isShowingConfig = showingConfig === serviceName}
{@const workspaceApp = hasWorkspaceApp(integration)}
{@const usesInstanceApp = !workspaceApp && instanceSharingAvailable[serviceName]}
{@const serviceConnections = connections[serviceName] ?? []}
<div class="border border-gray-200 dark:border-gray-700 rounded-md p-4 bg-surface-tertiary">
<div class="flex items-center justify-between">
<div class="flex items-center justify-between gap-4">
<div class="flex items-center gap-3">
<div class="w-8 h-8 flex items-center justify-center">
<config.icon class="w-6 h-6" />
@@ -398,65 +303,37 @@
</div>
</div>
<div class="flex items-center gap-2">
{#if isServiceConnected}
<div class="flex items-center gap-1 text-green-600 text-xs">
<Check size={16} />
<span class="font-semibold">Connected</span>
</div>
<div class="flex flex-wrap items-center justify-end gap-2">
{#if workspaceApp || usesInstanceApp}
<Button
onclick={() =>
integration?.oauth_data?.instance_shared
? connectWithInstanceCredentials(serviceName)
: connectService(serviceName, getRedirectUri(serviceName))}
variant="default"
onclick={() => {
choosingPathFor = serviceName
resourcePath = undefined
}}
disabled={isConnecting}
startIcon={{ icon: Plug }}
>
{isConnecting ? 'Reconnecting...' : 'Reconnect'}
Connect account
</Button>
{/if}
<Button
variant="default"
onclick={() => (showingConfig = showingConfig === serviceName ? null : serviceName)}
startIcon={{ icon: Cog }}
>
Configure OAuth
</Button>
{#if integration?.oauth_data}
<Button
variant="default"
destructive
onclick={() => deleteIntegration(serviceName)}
startIcon={{ icon: X }}
>
Delete
</Button>
{:else if isOAuthConfigured}
<Button
variant="accent"
onclick={() => connectService(serviceName, getRedirectUri(serviceName))}
disabled={isConnecting}
startIcon={{ icon: Plug }}
>
{isConnecting ? 'Connecting...' : 'Connect'}
</Button>
<Button
destructive
onclick={() => deleteIntegration(serviceName)}
startIcon={{ icon: X }}
>
Delete
</Button>
{:else if instanceSharingAvailable[serviceName]}
<Button
variant="accent"
onclick={() => connectWithInstanceCredentials(serviceName)}
disabled={isConnecting}
startIcon={{ icon: Plug }}
>
{isConnecting ? 'Connecting...' : 'Connect'}
</Button>
{:else}
<Button
variant="accent"
onclick={() =>
(showingConfig = showingConfig === serviceName ? null : serviceName)}
startIcon={{ icon: Cog }}
>
Configure OAuth
Remove
</Button>
{/if}
{#if config.docsUrl}
<Button href={config.docsUrl} target="_blank" startIcon={{ icon: ExternalLink }}>
Docs
@@ -465,25 +342,85 @@
</div>
</div>
{#if instanceSharingAvailable[serviceName] && !isOAuthConfigured}
{#if usesInstanceApp}
<div class="mt-3 pt-3 border-t border-gray-200 dark:border-gray-700">
<Alert type="info" title="Redirect URI required">
<p class="text-sm mb-2">
Your instance admin has configured Google OAuth for native triggers. Before
connecting, ensure the following redirect URI has been added to the
<a
href="https://console.cloud.google.com/apis/credentials"
target="_blank"
rel="noopener noreferrer"
class="underline">Google Cloud Console</a
> by the instance admin:
</p>
<ClipboardPanel content={getRedirectUri(serviceName)} size="sm" />
</Alert>
{#if serviceName === 'google'}
<Alert type="info" title="Uses the instance OAuth app">
<p class="text-sm mb-2">
Your instance admin shares a Google OAuth app. Before connecting, ensure the
following redirect URI has been added to it in the
<a
href="https://console.cloud.google.com/apis/credentials"
target="_blank"
rel="noopener noreferrer"
class="underline">Google Cloud Console</a
>:
</p>
<ClipboardPanel content={nativeOAuthRedirectUri('google')} size="sm" />
</Alert>
{:else}
<Alert type="info" title="Uses the instance OAuth app">
Your instance admin shares a {config.displayName} OAuth app, so members can connect
their accounts without further setup.
</Alert>
{/if}
</div>
{/if}
{#if isShowingConfig}
{#if choosingPathFor === serviceName}
<div class="mt-3 pt-3 border-t border-gray-200 dark:border-gray-700">
<div class="text-xs text-secondary mb-2">
Choose where to save the connection. Save it in a folder to let the folder's members
create triggers with this account.
</div>
<Path
kind="resource"
initialPath=""
namePlaceholder={'native_' + serviceName}
bind:path={resourcePath}
bind:error={pathError}
/>
<div class="flex gap-2 mt-3">
<Button
variant="accent"
disabled={!resourcePath || !!pathError || isConnecting}
loading={isConnecting}
onclick={() => connect(serviceName)}
>
Connect
</Button>
<Button variant="default" onclick={() => (choosingPathFor = null)}>Cancel</Button>
</div>
</div>
{/if}
{#if serviceConnections.length > 0}
<div
class="mt-3 pt-3 border-t border-gray-200 dark:border-gray-700 flex flex-col gap-1"
>
<div class="text-xs font-semibold text-emphasis mb-1">Connected accounts</div>
{#each serviceConnections as connection (connection.path)}
<div class="flex items-center justify-between gap-2 text-xs">
<span class="font-mono truncate">{connection.path}</span>
<div class="flex items-center gap-2 shrink-0">
{#if connection.owner}
<span class="text-secondary">connected by {connection.owner}</span>
{/if}
<Button
unifiedSize="xs"
variant="subtle"
destructive
onclick={() => disconnect(serviceName, connection.path)}
>
Disconnect
</Button>
</div>
</div>
{/each}
</div>
{/if}
{#if showingConfig === serviceName}
<div class="mt-3 pt-3 border-t border-gray-200 dark:border-gray-700">
{#if serviceName === 'nextcloud'}
<Alert type="info" title="Requirements" class="mb-4">
@@ -516,9 +453,9 @@
{/if}
<OAuthClientConfig
{serviceName}
redirectUri={getRedirectUri(serviceName)}
redirectUri={nativeOAuthRedirectUri(serviceName as NativeServiceName)}
serviceDisplayName={config.displayName}
existingConfig={integration?.oauth_data}
existingConfig={workspaceApp ? integration?.oauth_data : null}
requiresBaseUrl={config.requiresBaseUrl !== false}
clientIdPlaceholder={config.clientIdPlaceholder}
clientSecretPlaceholder={config.clientSecretPlaceholder}
@@ -533,12 +470,6 @@
</div>
{/each}
</div>
{#if integrations.length === 0}
<Alert type="warning" title="No Integrations Connected">
Connect to external services above to enable native triggers for your workspace.
</Alert>
{/if}
{/if}
</div>
@@ -70,6 +70,9 @@
import GitSyncSection from '$lib/components/git_sync/GitSyncSection.svelte'
import Trashbin from '$lib/components/settings/Trashbin.svelte'
import { untrack } from 'svelte'
import NativeOAuthCallback from '$lib/components/triggers/native/NativeOAuthCallback.svelte'
import { nativeOAuthRedirectUri } from '$lib/components/triggers/native/utils'
import type { NativeServiceName } from '$lib/gen'
import DbtSettings, {
convertDbtSettingsFromBackend,
type DbtSettingsType
@@ -385,6 +388,13 @@
// The deploy filters configure what may be promoted into the parent, so they only mean
// something for a fork. A root workspace deploys nowhere by lineage.
const showDeployToTab = $derived(Boolean(currentWorkspace?.parent_workspace_id))
// The page native trigger OAuth providers redirect to. Anyone can connect an account, so the
// callback is handled before the admin gate.
const nativeOAuthService = $derived(
$page.url.searchParams.get('tab') === 'native_triggers' && $page.url.searchParams.has('state')
? ($page.url.searchParams.get('service') as NativeServiceName | null)
: null
)
const canAdmin = $derived(($userStore?.is_admin ?? false) || Boolean($superadmin))
let yamlMode = $state(false)
// A workspace where the user is not an admin hides the toggle, so it must not stay in YAML mode.
@@ -1532,7 +1542,9 @@
</script>
<CenteredPage wrapperClasses="pb-0 h-full" handleOverflow={false} class="flex flex-col h-full">
{#if canAdmin || isForkOwner}
{#if nativeOAuthService}
<NativeOAuthCallback redirectUri={nativeOAuthRedirectUri(nativeOAuthService)} />
{:else if canAdmin || isForkOwner}
<!-- No title of its own: the band already says which workspace's settings these are, and the
two controls that sat beside the title are the page's actions. The panes below scroll
inside themselves, so the bar keeps its edge at rest rather than only under a scroll. -->
@@ -0,0 +1,8 @@
<script lang="ts">
import { page } from '$app/state'
import NativeOAuthCallback from '$lib/components/triggers/native/NativeOAuthCallback.svelte'
</script>
<!-- Where native trigger connections made with the instance GitHub OAuth app land: that app only
redirects below its registered callback, /oauth/callback/github. -->
<NativeOAuthCallback redirectUri="{page.url.origin}/oauth/callback/github/native_trigger" />