diff --git a/connectors/google/src/connector.rs b/connectors/google/src/connector.rs index f560739c9..829cea6ed 100644 --- a/connectors/google/src/connector.rs +++ b/connectors/google/src/connector.rs @@ -8,7 +8,10 @@ use crate::auth::{ }; use crate::drive::DriveClient; use crate::gmail::{MessageFormat, MessagePart}; -use crate::models::{GoogleDirectoryUser, GoogleSyncCheckpoint, SearchUsersResponse}; +use crate::models::{ + DriveFolderDiscoveryEntry, DriveFolderDiscoveryResponse, GoogleDirectoryUser, + GoogleSyncCheckpoint, SearchUsersResponse, +}; use crate::sync::SyncManager; use anyhow::{anyhow, Context, Result}; use async_trait::async_trait; @@ -808,6 +811,79 @@ impl GoogleConnector { Ok(ActionResponse::success(serde_json::to_value(result)?).into_response()) } + + async fn execute_discover_folders( + &self, + _params: JsonValue, + creds: &ServiceCredential, + ) -> Result { + // Only JWT secrets are supported for shared-drive discovery. + if creds.auth_type != AuthType::Jwt { + return Ok(ActionResponse::failure( + "discover_folders requires JWT credentials".to_string(), + ) + .into_response()); + } + + let principal_email = creds + .principal_email + .as_deref() + .ok_or_else(|| anyhow!("Missing principal_email in credentials"))?; + + let auth = crate::auth::create_service_auth(creds, SourceType::GoogleDrive)?; + + let google_auth = crate::auth::GoogleAuth::ServiceAccount(auth); + + // 1. List all shared drives + let drives_response = self + .sync_manager + .drive_client() + .list_drives(&google_auth, principal_email) + .await?; + + let mut items: Vec = Vec::new(); + + // 2. Add each shared drive as a selectable root + for drive in &drives_response.drives { + items.push(DriveFolderDiscoveryEntry { + id: drive.id.clone(), + name: drive.name.clone(), + path: format!("/{} (Shared Drive)", drive.name), + drive_id: drive.id.clone(), + kind: "shared_drive_root".to_string(), + }); + } + + // 3. For each shared drive, list top-level folder children. + // Any failure propagates so the admin gets an actionable error. + for drive in &drives_response.drives { + let folders = self + .sync_manager + .drive_client() + .list_folder_children(&google_auth, principal_email, &drive.id, &drive.id) + .await + .with_context(|| { + format!( + "Failed to list folder children for shared drive '{}' ({})", + drive.name, drive.id + ) + })?; + + for folder in &folders.files { + items.push(DriveFolderDiscoveryEntry { + id: folder.id.clone(), + name: folder.name.clone(), + path: format!("/{}/{}", drive.name, folder.name), + drive_id: drive.id.clone(), + kind: "folder".to_string(), + }); + } + } + + let result = DriveFolderDiscoveryResponse { items }; + + Ok(ActionResponse::success(serde_json::to_value(result)?).into_response()) + } } #[async_trait] @@ -907,6 +983,21 @@ impl Connector for GoogleConnector { admin_only: false, hidden: false, }, + ActionDefinition { + name: "discover_folders".to_string(), + description: + "List accessible shared drives and their top-level folders for folder-path filter selection." + .to_string(), + mode: omni_connector_sdk::ActionMode::Read, + input_schema: json!({ + "type": "object", + "properties": {}, + "required": [] + }), + source_types: vec![SourceType::GoogleDrive], + admin_only: true, + hidden: true, + }, ActionDefinition { name: "google_workspace_call".to_string(), description: "Call a Google Workspace API through the installed gws CLI" @@ -1156,6 +1247,7 @@ impl Connector for GoogleConnector { match action { "fetch_file" => self.execute_fetch_file(params, &creds).await, "search_users" => self.execute_search_users(params, &creds).await, + "discover_folders" => self.execute_discover_folders(params, &creds).await, "google_workspace_schema" => self.execute_gws_schema(params).await, "google_workspace_call" => self.execute_gws_call(params, &creds).await, _ => { diff --git a/connectors/google/src/drive.rs b/connectors/google/src/drive.rs index 51e613b46..8d37cff96 100644 --- a/connectors/google/src/drive.rs +++ b/connectors/google/src/drive.rs @@ -1042,19 +1042,79 @@ impl DriveClient { &self, token: &str, page_token: &str, + ) -> Result { + self.list_changes_internal(token, page_token, None).await + } + + /// Like `list_changes` but scoped to a single shared drive. + pub async fn list_changes_for_drive( + &self, + token: &str, + page_token: &str, + drive_id: &str, + ) -> Result { + self.list_changes_internal(token, page_token, Some(drive_id)) + .await + } + + /// Drive-scoped start page token. Required for per-drive incremental sync. + pub async fn get_start_page_token_for_drive( + &self, + token: &str, + drive_id: &str, + ) -> Result { + let url = format!("{}/changes/startPageToken", drive_api_base().as_str()); + let drive_id_owned = drive_id.to_string(); + + let params = vec![("supportsAllDrives", "true"), ("driveId", &drive_id_owned)]; + + let response = self + .client + .get(&url) + .bearer_auth(token) + .query(¶ms) + .send() + .await?; + + if !response.status().is_success() { + let error_text = response.text().await?; + return Err(anyhow!( + "Failed to get start page token for drive {}: {}", + drive_id, + error_text + )); + } + + let response_json: serde_json::Value = response.json().await?; + let start_page_token = response_json["startPageToken"] + .as_str() + .ok_or_else(|| anyhow!("Missing startPageToken in drive-scoped response"))?; + + Ok(start_page_token.to_string()) + } + + /// Internal helper that optionally scopes to a drive. + async fn list_changes_internal( + &self, + token: &str, + page_token: &str, + drive_id: Option<&str>, ) -> Result { let url = format!("{}/changes", drive_api_base().as_str()); - let params = vec![ + let mut params: Vec<(&str, &str)> = vec![ ("pageToken", page_token), ("includeItemsFromAllDrives", "true"), ("supportsAllDrives", "true"), ("includeRemoved", "true"), ( "fields", - "nextPageToken,changes(changeType,removed,file(id,name,mimeType,webViewLink,createdTime,modifiedTime,size,parents,shared,permissions(id,type,emailAddress,role),owners(emailAddress)),fileId,time)", + "nextPageToken,newStartPageToken,changes(changeType,removed,file(id,name,mimeType,webViewLink,createdTime,modifiedTime,size,parents,shared,permissions(id,type,emailAddress,domain,role,allowFileDiscovery,permissionDetails),owners(emailAddress)),fileId,time)", ), ]; + if let Some(did) = drive_id { + params.push(("driveId", did)); + } let response = self .client @@ -1080,6 +1140,339 @@ impl DriveClient { ) }) } + + /// List shared drives visible to the delegated principal's credentials. + /// + /// Paginates through all pages and returns the full list, sorted by drive name. + /// Only shared drives that the authenticated principal can see are returned; + /// does NOT use `useDomainAdminAccess` so drives the principal explicitly has + /// access to (e.g. shared drives they are members of) are surfaced. + pub async fn list_drives( + &self, + auth: &GoogleAuth, + user_email: &str, + ) -> Result { + let user_email_owned = user_email.to_string(); + let rate_limiter = self.rate_limiter.clone(); + let client = self.client.clone(); + + let mut all_drives: Vec = Vec::new(); + let mut page_token: Option = None; + + loop { + let page: DrivesListResponse = + execute_with_auth_retry(auth, &user_email_owned, rate_limiter.clone(), |token| { + let page_token = page_token.clone(); + let client = client.clone(); + async move { + let url = format!("{}/drives", drive_api_base().as_str()); + let mut params: Vec<(&str, &str)> = vec![ + ("pageSize", "100"), + ("fields", "nextPageToken,drives(id,name)"), + ]; + if let Some(ref pt) = page_token { + params.push(("pageToken", pt)); + } + + let response = client + .get(&url) + .bearer_auth(&token) + .query(¶ms) + .send() + .await?; + + let status = response.status(); + if !status.is_success() { + return classify_google_api_error( + response, + "Failed to list shared drives".to_string(), + ) + .await; + } + + let response_text = response.text().await?; + let parsed: DrivesListResponse = serde_json::from_str(&response_text) + .map_err(|e| { + anyhow!( + "Failed to parse drives list response: {}. Raw: {}", + e, + response_text + ) + })?; + + Ok(ApiResult::Success(parsed)) + } + }) + .await?; + + all_drives.extend(page.drives); + + if let Some(next) = page.next_page_token { + page_token = Some(next); + } else { + break; + } + } + + // Deterministic sort by name for stable UI ordering. + all_drives.sort_by(|a, b| a.name.cmp(&b.name)); + + Ok(DrivesListResponse { + drives: all_drives, + next_page_token: None, + }) + } + + /// List all files (not just folders) in a single shared drive with the given cut-off. + /// Uses `corpora=drive` and `driveId` to keep the query scoped. + /// Paginates fully and returns all files. + pub async fn list_files_in_drive( + &self, + auth: &GoogleAuth, + user_email: &str, + drive_id: &str, + page_token: Option<&str>, + modified_after: Option<&str>, + ) -> Result { + let drive_id_owned = drive_id.to_string(); + let page_token = page_token.map(|s| s.to_string()); + let modified_after = modified_after.map(|s| s.to_string()); + + execute_with_auth_retry(auth, user_email, self.rate_limiter.clone(), |token| { + let drive_id = drive_id_owned.clone(); + let page_token = page_token.clone(); + let modified_after = modified_after.clone(); + async move { + let url = format!("{}/files", drive_api_base().as_str()); + let query = build_files_query(modified_after.as_deref()); + + let mut params: Vec<(&str, &str)> = vec![ + ("pageSize", "100"), + ("fields", "nextPageToken,files(id,name,mimeType,webViewLink,createdTime,modifiedTime,size,parents,shared,permissions(id,type,emailAddress,domain,role,allowFileDiscovery,permissionDetails),owners(emailAddress))"), + ("q", query.as_str()), + ("orderBy", "modifiedTime desc"), + ("supportsAllDrives", "true"), + ("includeItemsFromAllDrives", "true"), + ("corpora", "drive"), + ("driveId", &drive_id), + ]; + if let Some(ref pt) = page_token { + params.push(("pageToken", pt)); + } + + debug!("[GOOGLE API CALL] list_files_in_drive drive={}, page_token={:?}", drive_id, page_token); + let response = self + .client + .get(&url) + .bearer_auth(&token) + .query(¶ms) + .send() + .await + .with_context(|| format!("Failed to list files in drive {}", drive_id))?; + + let status = response.status(); + if !status.is_success() { + return classify_google_api_error(response, "Failed to list files in drive").await; + } + + let response_text = response.text().await?; + let parsed: FilesListResponse = serde_json::from_str(&response_text).map_err(|e| { + anyhow!( + "Failed to parse drive-scoped file list response: {}. Raw: {}", + e, response_text + ) + })?; + + Ok(ApiResult::Success(parsed)) + } + }).await + } + + /// List all files under a given folder (non-recursive, paginated). + /// + /// Lists ALL immediate children (files and sub-folders) regardless of + /// modification time so that old folders containing new files are never + /// missed. Date-based filtering is applied client-side in the caller. + /// Used by folder-subtree traversal in scoped filtered sync. + pub async fn list_files_in_folder( + &self, + auth: &GoogleAuth, + user_email: &str, + folder_id: &str, + drive_id: &str, + page_token: Option<&str>, + ) -> Result { + let folder_id_owned = folder_id.to_string(); + let drive_id_owned = drive_id.to_string(); + let page_token = page_token.map(|s| s.to_string()); + + execute_with_auth_retry(auth, user_email, self.rate_limiter.clone(), |token| { + let folder_id = folder_id_owned.clone(); + let drive_id = drive_id_owned.clone(); + let page_token = page_token.clone(); + async move { + let url = format!("{}/files", drive_api_base().as_str()); + + // No date-cutoff in the query — we must discover old folders + // that may contain new files. The caller applies the cutoff + // client-side to non-folder file types only. + let query = format!("trashed=false and '{}' in parents", folder_id); + + let mut params: Vec<(&str, &str)> = vec![ + ("pageSize", "100"), + ("fields", "nextPageToken,files(id,name,mimeType,webViewLink,createdTime,modifiedTime,size,parents,shared,permissions(id,type,emailAddress,domain,role,allowFileDiscovery,permissionDetails),owners(emailAddress))"), + ("q", query.as_str()), + ("supportsAllDrives", "true"), + ("includeItemsFromAllDrives", "true"), + ("corpora", "drive"), + ("driveId", &drive_id), + ]; + if let Some(ref pt) = page_token { + params.push(("pageToken", pt)); + } + + debug!("[GOOGLE API CALL] list_files_in_folder folder={}, drive={}, page_token={:?}", folder_id, drive_id, page_token); + let response = self + .client + .get(&url) + .bearer_auth(&token) + .query(¶ms) + .send() + .await + .with_context(|| format!("Failed to list files in folder {}", folder_id))?; + + let status = response.status(); + if !status.is_success() { + return classify_google_api_error(response, "Failed to list files in folder").await; + } + + let response_text = response.text().await?; + let parsed: FilesListResponse = serde_json::from_str(&response_text).map_err(|e| { + anyhow!( + "Failed to parse folder-scoped file list response: {}. Raw: {}", + e, response_text + ) + })?; + + Ok(ApiResult::Success(parsed)) + } + }).await + } + + /// List immediate children (sub-folders only) of a given folder in any drive. + /// + /// Paginates through all pages and returns the full list, sorted by name. + /// Uses `corpora=drive` and `driveId` scoping for shared-drive children. + pub async fn list_folder_children( + &self, + auth: &GoogleAuth, + user_email: &str, + folder_id: &str, + drive_id: &str, + ) -> Result { + let folder_id_owned = folder_id.to_string(); + let drive_id_owned = drive_id.to_string(); + let user_email_owned = user_email.to_string(); + let rate_limiter = self.rate_limiter.clone(); + let client = self.client.clone(); + + let mut all_files: Vec = Vec::new(); + let mut page_token: Option = None; + + loop { + let page: FilesListResponse = execute_with_auth_retry( + auth, + &user_email_owned, + rate_limiter.clone(), + |token| { + let folder_id = folder_id_owned.clone(); + let drive_id = drive_id_owned.clone(); + let page_token = page_token.clone(); + let client = client.clone(); + async move { + let url = format!("{}/files", drive_api_base().as_str()); + + let query = format!( + "trashed=false and mimeType='application/vnd.google-apps.folder' and '{}' in parents", + folder_id + ); + + let mut params: Vec<(&str, &str)> = vec![ + ("pageSize", "100"), + ("fields", "nextPageToken,files(id,name,mimeType,parents,createdTime)"), + ("q", query.as_str()), + ("supportsAllDrives", "true"), + ("includeItemsFromAllDrives", "true"), + ("corpora", "drive"), + ("driveId", &drive_id), + ]; + if let Some(ref pt) = page_token { + params.push(("pageToken", pt)); + } + + let response = client + .get(&url) + .bearer_auth(&token) + .query(¶ms) + .send() + .await?; + + let status = response.status(); + if !status.is_success() { + return classify_google_api_error( + response, + format!("Failed to list folder children for {folder_id}"), + ) + .await; + } + + let response_text = response.text().await?; + let parsed: FilesListResponse = + serde_json::from_str(&response_text).map_err(|e| { + anyhow!( + "Failed to parse folder children response: {}. Raw: {}", + e, + response_text + ) + })?; + + Ok(ApiResult::Success(parsed)) + } + }, + ) + .await?; + + all_files.extend(page.files); + + if let Some(next) = page.next_page_token { + page_token = Some(next); + } else { + break; + } + } + + // Deterministic sort by name for stable UI ordering. + all_files.sort_by(|a, b| a.name.cmp(&b.name)); + + Ok(FilesListResponse { + files: all_files, + next_page_token: None, + }) + } +} + +/// Response from the Drive API v3 `/drives` endpoint. +#[derive(Debug, Deserialize)] +pub struct DrivesListResponse { + pub drives: Vec, + #[serde(rename = "nextPageToken")] + pub next_page_token: Option, +} + +#[derive(Debug, Deserialize, Clone)] +pub struct DriveMetadata { + pub id: String, + pub name: String, } #[derive(Debug, Deserialize)] diff --git a/connectors/google/src/models.rs b/connectors/google/src/models.rs index 6205a36c3..6df206c6f 100644 --- a/connectors/google/src/models.rs +++ b/connectors/google/src/models.rs @@ -23,6 +23,121 @@ pub struct GoogleDirectoryUser { pub is_admin: bool, } +/// A single entry in the persisted folder-path filter list in source config. +/// Stored under `sources.config.folder_path_filters`. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct FolderPathFilterEntry { + pub id: String, + pub name: String, + pub path: String, + #[serde(rename = "driveId")] + pub drive_id: String, + pub kind: FolderPathFilterKind, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum FolderPathFilterKind { + SharedDriveRoot, + Folder, +} + +/// Result entry returned by the `discover_folders` connector action. +#[derive(Debug, Clone, Serialize)] +pub struct DriveFolderDiscoveryEntry { + pub id: String, + pub name: String, + pub path: String, + #[serde(rename = "driveId")] + pub drive_id: String, + pub kind: String, +} + +/// Response from the `discover_folders` connector action. +#[derive(Debug, Clone, Serialize)] +pub struct DriveFolderDiscoveryResponse { + pub items: Vec, +} + +/// Helper to parse folder_path_filters from a source config. +/// +/// Returns: +/// - `Ok(Some(filters))` — a valid non-empty, deduplicated filter list was found +/// - `Ok(None)` — the key is absent or explicitly an empty array (index all) +/// - `Err(msg)` — the key is present but malformed (fail closed) +pub fn parse_folder_path_filters( + config: &serde_json::Value, +) -> Result>, String> { + let raw = match config.get("folder_path_filters") { + Some(v) => v, + None => return Ok(None), + }; + + // Null is malformed — fail closed. + if raw.is_null() { + return Err( + "folder_path_filters must not be null; omit the key or use [] to index all." + .to_string(), + ); + } + + let arr = raw.as_array().ok_or_else(|| { + format!( + "folder_path_filters must be an array, got {}", + serde_json::to_string(raw).unwrap_or_default() + ) + })?; + + // Empty array → index all. + if arr.is_empty() { + return Ok(None); + } + + // Parse each entry, rejecting null and validating non-empty required fields. + let mut entries: Vec = Vec::new(); + let mut seen_ids: std::collections::HashSet = std::collections::HashSet::new(); + + for (i, item) in arr.iter().enumerate() { + let entry: FolderPathFilterEntry = serde_json::from_value(item.clone()).map_err(|e| { + format!( + "folder_path_filters[{}] is invalid: {} (value={})", + i, + e, + serde_json::to_string(item).unwrap_or_default() + ) + })?; + + // Validate non-empty required fields. + if entry.id.is_empty() { + return Err(format!("folder_path_filters[{}].id must not be empty", i)); + } + if entry.name.is_empty() { + return Err(format!("folder_path_filters[{}].name must not be empty", i)); + } + if entry.path.is_empty() { + return Err(format!("folder_path_filters[{}].path must not be empty", i)); + } + if entry.drive_id.is_empty() { + return Err(format!( + "folder_path_filters[{}].drive_id must not be empty", + i + )); + } + + // Deduplicate by stable ID — first occurrence wins. + if seen_ids.insert(entry.id.clone()) { + entries.push(entry); + } + } + + if entries.is_empty() { + Ok(None) + } else { + Ok(Some(entries)) + } +} + #[derive(Debug, Clone, Serialize)] pub struct SearchUsersResponse { pub users: Vec, @@ -39,10 +154,54 @@ pub struct GoogleConnectorState { pub webhook_expires_at: Option, } +/// Normalised scope fingerprint for shared-drive/folder filtered mode. +/// `"all"` means no folder filters (index everything). +/// Otherwise it is a deterministic representation of the effective drive/folder scope. +pub fn compute_scope_fingerprint(filters: Option<&[FolderPathFilterEntry]>) -> String { + match filters { + None | Some([]) => "all".to_string(), + Some(list) => { + let entire_drives: HashSet<&str> = list + .iter() + .filter(|filter| matches!(filter.kind, FolderPathFilterKind::SharedDriveRoot)) + .map(|filter| filter.drive_id.as_str()) + .collect(); + let mut parts: Vec = list + .iter() + .filter(|filter| { + matches!(filter.kind, FolderPathFilterKind::SharedDriveRoot) + || !entire_drives.contains(filter.drive_id.as_str()) + }) + .map(|filter| format!("{}:{}:{}", filter.drive_id, filter.kind.as_str(), filter.id)) + .collect(); + parts.sort(); + parts.dedup(); + parts.join(";") + } + } +} + +impl FolderPathFilterKind { + pub fn as_str(&self) -> &'static str { + match self { + FolderPathFilterKind::SharedDriveRoot => "shared_drive_root", + FolderPathFilterKind::Folder => "folder", + } + } +} + #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct GoogleSyncCheckpoint { pub gmail_history_ids: Option>, pub drive_page_tokens: Option>, + /// Per-drive change tokens used in scoped (filtered) incremental sync. + /// Keyed by shared-drive ID; absent for unfiltered mode. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub drive_change_tokens: Option>, + /// Scope fingerprint for detecting filter transitions. + /// `"all"` or a sorted `;`-joined list of selected drive/kind/ID tuples. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub drive_scope_fingerprint: Option, pub chat: Option, } @@ -420,6 +579,10 @@ impl WebhookNotification { pub struct DriveChangesResponse { #[serde(rename = "nextPageToken")] pub next_page_token: Option, + /// The starting page token for future incremental changes. + /// Returned on the last page of a changes.list response. + #[serde(rename = "newStartPageToken", default)] + pub new_start_page_token: Option, pub changes: Vec, } @@ -1411,11 +1574,9 @@ mod tests { match event { ConnectorEvent::DocumentCreated { permissions, .. } => { assert!(permissions.users.contains(&"owner@example.com".to_string())); - assert!( - permissions - .users - .contains(&"viewer@example.com".to_string()) - ); + assert!(permissions + .users + .contains(&"viewer@example.com".to_string())); assert_eq!(permissions.users.len(), 2); } _ => panic!("Expected DocumentCreated event"), @@ -1588,11 +1749,9 @@ mod tests { ConnectorEvent::DocumentCreated { permissions, .. } => { assert!(permissions.users.contains(&"alice@example.com".to_string())); assert!(permissions.users.contains(&"owner@example.com".to_string())); - assert!( - permissions - .users - .contains(&"sender@example.com".to_string()) - ); + assert!(permissions + .users + .contains(&"sender@example.com".to_string())); } _ => panic!("Expected DocumentCreated event"), } @@ -1673,4 +1832,158 @@ mod tests { _ => panic!("Expected DocumentCreated event"), } } + + // ======================================================================== + // Folder path filter parse tests + // + // Focus on behaviour rather than Serde mechanics: + // 1. absent / null / empty — the three filter knob states + // 2. successful parse + first-wins deduplication + // 3. a single table of malformed inputs that must all fail closed + // ======================================================================== + + #[test] + fn test_folder_filter_absent_or_empty_is_none() { + // Absent key. + let r = parse_folder_path_filters(&json!({"domain": "x"})); + assert!(r.is_ok() && r.unwrap().is_none()); + + // Empty array (explicit “no filter”). + let r = parse_folder_path_filters(&json!({"folder_path_filters": []})); + assert!(r.is_ok() && r.unwrap().is_none()); + } + + #[test] + fn test_folder_filter_valid_parse_with_dedup() { + let config = json!({ + "folder_path_filters": [ + { + "id": "0ADRI000001", + "name": "Engineering", + "path": "/Engineering (Shared Drive)", + "driveId": "0ADRI000001", + "kind": "shared_drive_root" + }, + { + "id": "1FOLDER0001", + "name": "Docs", + "path": "/Engineering/Docs", + "driveId": "0ADRI000001", + "kind": "folder" + }, + // duplicate of the first – must be silently dropped + { + "id": "0ADRI000001", + "name": "Duplicate", + "path": "/Engineering (Shared Drive)", + "driveId": "0ADRI000001", + "kind": "shared_drive_root" + } + ] + }); + let r = parse_folder_path_filters(&config).unwrap().unwrap(); + assert_eq!(r.len(), 2, "duplicate id deduped away"); + assert_eq!(r[0].name, "Engineering", "first occurrence wins"); + assert_eq!(r[1].name, "Docs"); + assert!(matches!(r[0].kind, FolderPathFilterKind::SharedDriveRoot)); + assert!(matches!(r[1].kind, FolderPathFilterKind::Folder)); + } + + #[test] + fn test_folder_filter_malformed_rejected() { + // Table of invalid configurations — every row must produce Err. + let cases: Vec<(&str, serde_json::Value)> = vec![ + ("null", serde_json::Value::Null), + ("not an array", json!("not-an-array")), + ("array of primitives", json!(["invalid"])), + ("missing required fields", json!([{"id": "only-id"}])), + ( + "unknown kind", + json!([{ + "id":"1", "name":"T", "path":"/T", "driveId":"1", "kind":"nope" + }]), + ), + ( + "unknown extra field", + json!([{ + "id":"1", "name":"T", "path":"/T", "driveId":"1", "kind":"folder", + "extra":"x" + }]), + ), + ( + "empty id", + json!([{ + "id":"", "name":"T", "path":"/T", "driveId":"1", "kind":"folder" + }]), + ), + ( + "empty name", + json!([{ + "id":"1", "name":"", "path":"/T", "driveId":"1", "kind":"folder" + }]), + ), + ( + "empty driveId", + json!([{ + "id":"1", "name":"T", "path":"/T", "driveId":"", "kind":"folder" + }]), + ), + ]; + for (label, val) in &cases { + let config = json!({"folder_path_filters": val}); + assert!( + parse_folder_path_filters(&config).is_err(), + "should reject: {}", + label + ); + } + } + + // ======================================================================== + // Scope fingerprint tests + // ======================================================================== + + #[test] + fn scope_fingerprint_none_returns_all() { + assert_eq!(compute_scope_fingerprint(None), "all"); + assert_eq!(compute_scope_fingerprint(Some(&[])), "all"); + } + + #[test] + fn scope_fingerprint_deterministic_sorted() { + let filters = vec![ + FolderPathFilterEntry { + id: "drive-b".to_string(), + name: "B".to_string(), + path: "/B".to_string(), + drive_id: "b".to_string(), + kind: FolderPathFilterKind::SharedDriveRoot, + }, + FolderPathFilterEntry { + id: "drive-a".to_string(), + name: "A".to_string(), + path: "/A".to_string(), + drive_id: "a".to_string(), + kind: FolderPathFilterKind::SharedDriveRoot, + }, + FolderPathFilterEntry { + id: "redundant-folder".to_string(), + name: "Folder".to_string(), + path: "/A/Folder".to_string(), + drive_id: "a".to_string(), + kind: FolderPathFilterKind::Folder, + }, + ]; + let fp = compute_scope_fingerprint(Some(&filters)); + // Fingerprint reflects effective scope: the selected drive root supersedes + // its redundant folder selection, and remaining entries are sorted. + assert_eq!( + fp, + "a:shared_drive_root:drive-a;b:shared_drive_root:drive-b" + ); + + // Reverse input order should produce same result. + let filters_rev: Vec = filters.into_iter().rev().collect(); + assert_eq!(compute_scope_fingerprint(Some(&filters_rev)), fp); + } } diff --git a/connectors/google/src/sync.rs b/connectors/google/src/sync.rs index 098c536a7..026a1e722 100644 --- a/connectors/google/src/sync.rs +++ b/connectors/google/src/sync.rs @@ -1,11 +1,12 @@ -use anyhow::{Context, Result, anyhow}; +use anyhow::{anyhow, Context, Result}; use dashmap::DashMap; -use futures::{StreamExt, stream}; +use futures::{stream, StreamExt}; use omni_connector_sdk::SyncContext; -use std::collections::{HashMap, HashSet}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::future::Future; -use std::sync::Arc; +use std::pin::Pin; use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; use std::time::{Duration, Instant}; use time::{self, OffsetDateTime}; use tokio::sync::{Mutex, Notify, OwnedSemaphorePermit, Semaphore}; @@ -84,6 +85,14 @@ fn google_drive_max_download_bytes() -> usize { .unwrap_or(DEFAULT_GOOGLE_DRIVE_MAX_DOWNLOAD_BYTES) } +fn drive_scope_requires_full_sync( + sync_type: SyncType, + stored_fingerprint: Option<&str>, + current_fingerprint: &str, +) -> bool { + sync_type != SyncType::Incremental || stored_fingerprint != Some(current_fingerprint) +} + fn google_webhook_debounce_duration_ms() -> u64 { std::env::var("GOOGLE_WEBHOOK_DEBOUNCE_SECONDS") .ok() @@ -266,7 +275,7 @@ async fn emit_metadata_only_drive_event( } use crate::admin::AdminClient; -use crate::auth::{GoogleAuth, GoogleOAuthCredentials, OAuthAuth, google_max_retries}; +use crate::auth::{google_max_retries, GoogleAuth, GoogleOAuthCredentials, OAuthAuth}; use crate::cache::LruFolderCache; use crate::chat::{ ChatClient, GoogleChatAttachmentSource, GoogleChatMessage, GoogleChatSpace, @@ -276,9 +285,9 @@ use crate::connector::build_attachment_doc_id; use crate::drive::{DriveClient, FileContent}; use crate::gmail::{BatchThreadResult, ExtractedAttachment, GmailClient, MessageFormat}; use crate::models::{ - AttachmentPointer, GmailThread, GoogleChatSegmentCheckpoint, GoogleChatSpaceCheckpoint, - GoogleConnectorState, GoogleSyncCheckpoint, UserFile, WebhookChannel, WebhookChannelResponse, - WebhookNotification, mime_type_to_content_type, + mime_type_to_content_type, AttachmentPointer, GmailThread, GoogleChatSegmentCheckpoint, + GoogleChatSpaceCheckpoint, GoogleConnectorState, GoogleSyncCheckpoint, UserFile, + WebhookChannel, WebhookChannelResponse, WebhookNotification, }; use omni_connector_sdk::RateLimiter; use omni_connector_sdk::SdkClient; @@ -444,12 +453,11 @@ impl GoogleChatSegment { extra.insert("message_count".to_string(), json!(self.messages.len())); extra.insert( "message_names".to_string(), - json!( - self.messages - .iter() - .map(|m| m.name.clone()) - .collect::>() - ), + json!(self + .messages + .iter() + .map(|m| m.name.clone()) + .collect::>()), ); extra.insert("thread_names".to_string(), json!(self.thread_names())); extra.insert( @@ -499,11 +507,10 @@ impl GoogleChatSegment { let mut attrs = HashMap::new(); attrs.insert( "space".to_string(), - json!( - self.space_display_name - .as_deref() - .unwrap_or(&self.space_name) - ), + json!(self + .space_display_name + .as_deref() + .unwrap_or(&self.space_name)), ); attrs.insert("space_id".to_string(), json!(self.space_name)); attrs.insert("threads".to_string(), json!(self.thread_names())); @@ -1059,6 +1066,69 @@ impl SyncManager { Ok((drive_format, gmail_format)) } + // TODO: When folder-path filters are narrowed or changed, documents that were + // previously indexed but are no longer in scope are NOT automatically pruned from + // the search index. A future reconciliation step should delete documents whose + // parent ancestry no longer intersects the configured filter set. + + /// Check whether a file should be included based on configured folder-path filters. + /// Uses `self.folder_cache` (the same LRU cache used by `build_full_path`) for + /// folder-metadata lookups during ancestry resolution. + /// + /// Returns `Ok(true)` if no filters are configured (include-everything mode). + /// Returns `Ok(false)` if the file's ancestry is outside the configured scope. + /// Returns `Err(...)` if ancestry metadata could not be determined due to a + /// transient API error — the caller should fail/retry the sync rather than + /// silently including or excluding the file. + async fn is_file_in_allowed_folder( + &self, + auth: &GoogleAuth, + user_email: &str, + file: &crate::models::GoogleDriveFile, + allowed_folder_ids: &HashSet, + ) -> Result { + if allowed_folder_ids.is_empty() { + return Ok(true); + } + + let folder_cache = self.folder_cache.clone(); + let drive_client = self.drive_client.clone(); + let google_auth = auth.clone(); + let user_email_owned = user_email.to_string(); + + let lookup = move |folder_id: String| -> Pin< + Box)>> + Send>, + > { + let folder_cache = folder_cache.clone(); + let drive_client = drive_client.clone(); + let google_auth = google_auth.clone(); + let user_email_owned = user_email_owned.clone(); + Box::pin(async move { + // Try the LRU folder cache first (same cache used by build_full_path). + if let Some(cached) = folder_cache.get(&folder_id) { + let parent = cached.parents.as_ref().and_then(|p| p.first()).cloned(); + return Ok((cached.name, parent)); + } + // Cache miss: fetch from API and populate cache. + let meta = drive_client + .get_folder_metadata(&google_auth, &user_email_owned, &folder_id) + .await + .with_context(|| { + format!( + "Failed to fetch folder metadata for {} while evaluating folder-path filter", + folder_id + ) + })?; + let name = meta.name.clone(); + let parent = meta.parents.as_ref().and_then(|p| p.first()).cloned(); + folder_cache.insert(folder_id.clone(), meta.into()); + Ok((name, parent)) + }) + }; + + check_ancestry_static(&file.parents, allowed_folder_ids, &lookup).await + } + async fn sync_drive_for_user( &self, user_email: &str, @@ -1068,6 +1138,7 @@ impl SyncManager { ctx: &SyncContext, created_after: Option<&str>, content_cache: Arc, + folder_filter_ids: Option>>, ) -> Result<(usize, usize)> { info!("Processing Drive files for user: {}", user_email); @@ -1109,6 +1180,25 @@ impl SyncManager { // we always emit and let the indexer skip unchanged docs. for file in response.files { if self.should_index_file(&file) { + // Apply folder-path filter if configured. + // Errors propagate upward so the sync fails and can be retried. + if let Some(ref allowed_ids) = folder_filter_ids { + if !self + .is_file_in_allowed_folder( + &service_auth, + user_email, + &file, + allowed_ids, + ) + .await? + { + debug!( + "Skipping file {} ({}) — outside configured folder scope", + file.name, file.id + ); + continue; + } + } file_batch.push(UserFile { user_email: Arc::new(user_email.to_string()), file, @@ -1182,6 +1272,7 @@ impl SyncManager { ctx: &SyncContext, start_page_token: &str, content_cache: Arc, + folder_filter_ids: Option>>, ) -> Result<(usize, usize)> { info!( "Processing incremental Drive sync for user {} from pageToken {}", @@ -1245,6 +1336,21 @@ impl SyncManager { continue; } + // Apply folder-path filter if configured. + // Errors propagate upward so the sync fails and can be retried. + if let Some(ref allowed_ids) = folder_filter_ids { + if !self + .is_file_in_allowed_folder(&service_auth, user_email, &file, allowed_ids) + .await? + { + debug!( + "Skipping change for file {} ({}) — outside configured folder scope", + file.name, file.id + ); + continue; + } + } + file_batch.push(UserFile { user_email: Arc::new(user_email.to_string()), file, @@ -1668,6 +1774,512 @@ impl SyncManager { Ok((scanned, updated)) } + /// Build a scope-aware Drive checkpoint that preserves Gmail and Chat state. + fn build_drive_checkpoint( + &self, + existing: &GoogleSyncCheckpoint, + drive_page_tokens: Option>, + drive_change_tokens: Option>, + drive_scope_fingerprint: Option, + ) -> GoogleSyncCheckpoint { + GoogleSyncCheckpoint { + gmail_history_ids: existing.gmail_history_ids.clone(), + drive_page_tokens, + drive_change_tokens, + drive_scope_fingerprint, + chat: existing.chat.clone(), + } + } + + /// Helper: apply the date cutoff to a single Google Drive file. + /// Returns `true` if the file passes the cutoff (or no cutoff is set). + /// Folders are never filtered by the cutoff — they must be discovered + /// regardless of age so their contents can be indexed. + fn pass_cutoff(file: &crate::models::GoogleDriveFile, cutoff: OffsetDateTime) -> bool { + if file.mime_type == "application/vnd.google-apps.folder" { + return true; + } + file.modified_time + .as_deref() + .and_then(|value| parse_google_time(Some(value))) + .is_some_and(|modified| modified > cutoff) + || file + .created_time + .as_deref() + .and_then(|value| parse_google_time(Some(value))) + .is_some_and(|created| created > cutoff) + } + + /// Flush a batch of user files through `process_file_batch` and update totals. + async fn flush_batch( + &self, + batch: &mut Vec, + source_id: &str, + sync_run_id: &str, + ctx: &SyncContext, + service_auth: Arc, + content_cache: Arc, + total_scanned: &mut usize, + total_updated: &mut usize, + ) -> Result<()> { + if batch.is_empty() { + return Ok(()); + } + let (s, u) = self + .process_file_batch( + std::mem::take(batch), + source_id, + sync_run_id, + ctx, + service_auth, + content_cache, + ) + .await?; + *total_scanned += s; + *total_updated += u; + Ok(()) + } + + /// Scoped full sync: crawl selected shared drives and/or folder subtrees only, + /// using the configured DWD principal (no user impersonation/iteration). + async fn sync_drive_scoped( + &self, + source: &Source, + service_creds: &ServiceCredential, + sync_type: SyncType, + existing_state: &GoogleSyncCheckpoint, + ctx: &SyncContext, + ) -> Result { + let sync_run_id = ctx.sync_run_id(); + let service_auth = Arc::new(self.create_auth(service_creds, source.source_type).await?); + let (drive_cutoff_date, _gmail_cutoff_date) = self.get_cutoff_date()?; + let drive_cutoff = parse_google_time(Some(&drive_cutoff_date)) + .ok_or_else(|| anyhow!("Invalid Drive cutoff time: {}", drive_cutoff_date))?; + + // Parse filters; will error on malformed config. + let filter_entries = match crate::models::parse_folder_path_filters(&source.config) { + Ok(Some(f)) => f, + Ok(None) => { + // Should not happen — caller guards this branch — but be safe. + return Err(anyhow!("sync_drive_scoped called without filters")); + } + Err(e) => { + return Err(anyhow!("Invalid folder path filter configuration: {}", e)); + } + }; + + let current_fingerprint = crate::models::compute_scope_fingerprint(Some(&filter_entries)); + + // Detect scope transition. If the fingerprint differs from the + // checkpoint, force a full sync regardless of requested mode. + let stored_fingerprint = existing_state.drive_scope_fingerprint.as_deref(); + let scope_changed = stored_fingerprint != Some(¤t_fingerprint); + + if scope_changed { + info!( + "Drive scope transition detected: {:?} -> {} — forcing full sync", + stored_fingerprint, current_fingerprint + ); + } + let effective_full = + drive_scope_requires_full_sync(sync_type, stored_fingerprint, ¤t_fingerprint); + + // Build drive groups: driveId → (selected folder IDs, whether entire drive is selected) + struct DriveGroup { + drive_id: String, + entire_drive: bool, + folder_ids: Vec, + } + + let mut drive_groups: HashMap = HashMap::new(); + for entry in &filter_entries { + let group = drive_groups + .entry(entry.drive_id.clone()) + .or_insert(DriveGroup { + drive_id: entry.drive_id.clone(), + entire_drive: false, + folder_ids: Vec::new(), + }); + match entry.kind { + crate::models::FolderPathFilterKind::SharedDriveRoot => { + group.entire_drive = true; + group.folder_ids.clear(); // entire-drive supersedes per-folder. + } + crate::models::FolderPathFilterKind::Folder => { + if !group.entire_drive { + group.folder_ids.push(entry.id.clone()); + } + } + } + } + + let drives: Vec = drive_groups.into_values().collect(); + if drives.is_empty() { + info!("No valid drive scopes found — indexing nothing"); + return Ok(self.build_drive_checkpoint( + existing_state, + None, + None, + Some(current_fingerprint), + )); + } + + // Resolve the delegated principal email. + let principal_email = service_creds + .principal_email + .as_deref() + .ok_or_else(|| anyhow!("Missing principal_email for scoped Drive sync"))? + .to_string(); + + // Shared-drive operations need a single access token (no impersonation per user). + let access_token = service_auth.get_access_token(&principal_email).await?; + + let content_cache = Arc::new(DriveContentCache::default()); + let mut total_scanned = 0; + let mut total_updated = 0; + + let mut new_change_tokens: HashMap = HashMap::new(); + let old_change_tokens = existing_state + .drive_change_tokens + .clone() + .unwrap_or_default(); + const BATCH_SIZE: usize = 200; + + for drive_group in &drives { + if ctx.is_cancelled() { + info!("Sync {} cancelled during drive-scoped sync", sync_run_id); + break; + } + + info!( + "Processing drive {} (entire={}, folders={:?})", + drive_group.drive_id, drive_group.entire_drive, drive_group.folder_ids + ); + + let stored_token = old_change_tokens.get(&drive_group.drive_id); + let use_incremental = !effective_full && stored_token.is_some(); + + if use_incremental { + // Incremental per-drive changes path. Use newStartPageToken + // from the API response to advance the token. + let start_token = stored_token.unwrap(); + let mut current_token = start_token.clone(); + + loop { + let response = self + .drive_client + .list_changes_for_drive( + &access_token, + ¤t_token, + &drive_group.drive_id, + ) + .await?; + + if ctx.is_cancelled() { + break; + } + + let mut file_batch: Vec = Vec::new(); + + for change in &response.changes { + let is_removed = change.removed.unwrap_or(false); + + if is_removed { + if let Some(file_id) = &change.file_id { + info!( + "File {} was removed (scoped incremental), publishing deletion", + file_id + ); + self.publish_deletion_event(ctx, file_id).await?; + } + continue; + } + + if let Some(file) = &change.file { + if !self.should_index_file(file) { + continue; + } + + // For folder-only selections, ancestry-check the change. + if !drive_group.entire_drive { + let allowed: HashSet = + drive_group.folder_ids.iter().cloned().collect(); + if !self + .is_file_in_allowed_folder( + &service_auth, + &principal_email, + file, + &allowed, + ) + .await? + { + debug!( + "Skipping changed file {} ({}) — outside folder scope", + file.name, file.id + ); + continue; + } + } + + file_batch.push(UserFile { + user_email: Arc::new(principal_email.clone()), + file: file.clone(), + }); + + if file_batch.len() >= BATCH_SIZE { + self.flush_batch( + &mut file_batch, + &source.id, + &sync_run_id, + ctx, + service_auth.clone(), + content_cache.clone(), + &mut total_scanned, + &mut total_updated, + ) + .await?; + } + } + } + + // Flush remaining changes for this page. + self.flush_batch( + &mut file_batch, + &source.id, + &sync_run_id, + ctx, + service_auth.clone(), + content_cache.clone(), + &mut total_scanned, + &mut total_updated, + ) + .await?; + + // Use newStartPageToken from response if present. + if let Some(new_token) = &response.new_start_page_token { + new_change_tokens.insert(drive_group.drive_id.clone(), new_token.clone()); + } + + match response.next_page_token { + Some(token) => current_token = token, + None => break, + } + } + + // Fallback: if no newStartPageToken was received, fetch one. + if !new_change_tokens.contains_key(&drive_group.drive_id) { + match self + .drive_client + .get_start_page_token_for_drive(&access_token, &drive_group.drive_id) + .await + { + Ok(token) => { + new_change_tokens.insert(drive_group.drive_id.clone(), token); + } + Err(e) => { + warn!( + "Failed to get start page token for drive {}: {}", + drive_group.drive_id, e + ); + } + } + } + } else { + // Capture the drive-scoped token before the full crawl so changes + // made while pagination is in progress are replayed next time. + let next_change_token = match self + .drive_client + .get_start_page_token_for_drive(&access_token, &drive_group.drive_id) + .await + { + Ok(token) => Some(token), + Err(error) => { + warn!( + "Failed to get start page token for drive {}: {}", + drive_group.drive_id, error + ); + None + } + }; + + // Full scoped sync for this drive. Stream directly into batches. + let mut file_batch: Vec = Vec::new(); + + if drive_group.entire_drive { + // Entire-drive selection: list all files with corpora=drive + let mut page_token: Option = None; + loop { + let response = self + .drive_client + .list_files_in_drive( + &service_auth, + &principal_email, + &drive_group.drive_id, + page_token.as_deref(), + Some(&drive_cutoff_date), + ) + .await?; + + for file in response.files { + if self.should_index_file(&file) { + file_batch.push(UserFile { + user_email: Arc::new(principal_email.clone()), + file, + }); + if file_batch.len() >= BATCH_SIZE { + self.flush_batch( + &mut file_batch, + &source.id, + &sync_run_id, + ctx, + service_auth.clone(), + content_cache.clone(), + &mut total_scanned, + &mut total_updated, + ) + .await?; + } + } + } + + if ctx.is_cancelled() { + break; + } + + page_token = response.next_page_token; + if page_token.is_none() { + break; + } + } + } else { + // Folder-level selection: traverse each selected subtree. + let mut seen_file_ids: HashSet = HashSet::new(); + let mut visited_folders: HashSet = HashSet::new(); + + for folder_id in &drive_group.folder_ids { + if ctx.is_cancelled() { + break; + } + if !visited_folders.insert(folder_id.clone()) { + continue; // already traversed via overlapping selection + } + + let mut queue = VecDeque::from([folder_id.clone()]); + while let Some(current_folder_id) = queue.pop_front() { + if ctx.is_cancelled() { + break; + } + + // Mark visited to avoid re-traversal via + // overlapping folder selections. + visited_folders.insert(current_folder_id.clone()); + + let mut page_token: Option = None; + loop { + // List ALL children — no date cutoff so old + // folders containing new files are never missed. + let response = self + .drive_client + .list_files_in_folder( + &service_auth, + &principal_email, + ¤t_folder_id, + &drive_group.drive_id, + page_token.as_deref(), + ) + .await?; + + for file in response.files { + if seen_file_ids.contains(&file.id) { + continue; + } + seen_file_ids.insert(file.id.clone()); + + // Folders are enqueued unconditionally. + if file.mime_type == "application/vnd.google-apps.folder" + && !visited_folders.contains(&file.id) + { + queue.push_back(file.id.clone()); + } + + // Non-folder files: apply cutoff locally. + if !self.should_index_file(&file) { + continue; + } + if !Self::pass_cutoff(&file, drive_cutoff) { + continue; + } + + file_batch.push(UserFile { + user_email: Arc::new(principal_email.clone()), + file, + }); + if file_batch.len() >= BATCH_SIZE { + self.flush_batch( + &mut file_batch, + &source.id, + &sync_run_id, + ctx, + service_auth.clone(), + content_cache.clone(), + &mut total_scanned, + &mut total_updated, + ) + .await?; + } + } + + if ctx.is_cancelled() { + break; + } + + page_token = response.next_page_token; + if page_token.is_none() { + break; + } + } + } + } + } + + // Flush any remaining files for this drive. + self.flush_batch( + &mut file_batch, + &source.id, + &sync_run_id, + ctx, + service_auth.clone(), + content_cache.clone(), + &mut total_scanned, + &mut total_updated, + ) + .await?; + + if ctx.is_cancelled() { + break; + } + + if let Some(token) = next_change_token { + new_change_tokens.insert(drive_group.drive_id.clone(), token); + } + } + } + + info!( + "Scoped Drive sync completed for source {}: {} scanned, {} updated", + source.id, total_scanned, total_updated + ); + + self.folder_cache.clear(); + + Ok(self.build_drive_checkpoint( + existing_state, + None, // No per-user page tokens in scoped mode. + Some(new_change_tokens), + Some(current_fingerprint), + )) + } + async fn sync_drive_source_internal( &self, source: &Source, @@ -1677,10 +2289,46 @@ impl SyncManager { ctx: &SyncContext, ) -> Result { let sync_run_id = ctx.sync_run_id(); - let service_auth = Arc::new(self.create_auth(service_creds, source.source_type).await?); - // Calculate cutoff date for filtering + // Parse folder-path filters. Malformed config is a sync error. + let parsed_filters = match crate::models::parse_folder_path_filters(&source.config) { + Ok(f) => f, + Err(e) => { + return Err(anyhow!("Invalid folder path filter configuration: {}", e)); + } + }; + + // Scoped mode (filters present, non-OAuth): delegate to sync_drive_scoped. + if parsed_filters.is_some() && !service_auth.is_oauth() { + return self + .sync_drive_scoped(source, service_creds, sync_type, &existing_state, ctx) + .await; + } + + // === Unfiltered / OAuth path: existing all-user listing === + + // Compute fingerprint. For OAuth sources, filters are ignored — + // folder filtering is DWD-only — so the fingerprint is always "all". + let current_fingerprint = if service_auth.is_oauth() { + "all".to_string() + } else { + match &parsed_filters { + Some(f) => crate::models::compute_scope_fingerprint(Some(f)), + None => "all".to_string(), + } + }; + + // Detect scope transition and force full sync if needed. + let stored_fingerprint = existing_state.drive_scope_fingerprint.as_deref(); + let scope_changed = stored_fingerprint != Some(¤t_fingerprint); + if scope_changed { + info!( + "Drive scope transition detected: {:?} -> {} — forcing full sync", + stored_fingerprint, current_fingerprint + ); + } + let (drive_cutoff_date, _gmail_cutoff_date) = self.get_cutoff_date()?; info!("Using Drive cutoff date: {}", drive_cutoff_date); @@ -1716,12 +2364,14 @@ impl SyncManager { filtered }; - let is_incremental = matches!(sync_type, SyncType::Incremental); + // Determine effective incremental mode — scope transitions force full sync. + let is_incremental = + !drive_scope_requires_full_sync(sync_type, stored_fingerprint, ¤t_fingerprint); let gmail_history_ids = existing_state.gmail_history_ids.clone(); let chat_checkpoint = existing_state.chat.clone(); let old_page_tokens = existing_state.drive_page_tokens.unwrap_or_default(); - let can_resume_full = sync_type == SyncType::Full && ctx.is_resume(); + let can_resume_full = sync_type == SyncType::Full && ctx.is_resume() && !scope_changed; let mut new_page_tokens: HashMap = if can_resume_full { old_page_tokens.clone() } else { @@ -1739,6 +2389,13 @@ impl SyncManager { let mut successful_users = 0; let mut errors = 0; let mut last_error: Option = None; + let folder_filter_ids: Option>> = if service_auth.is_oauth() { + None + } else { + parsed_filters + .as_ref() + .map(|filters| Arc::new(filters.iter().map(|filter| filter.id.clone()).collect())) + }; let content_cache = Arc::new(DriveContentCache::default()); let parallel_users = google_drive_parallel_users(); info!("Processing Drive users with concurrency {}", parallel_users); @@ -1751,6 +2408,7 @@ impl SyncManager { let ctx = ctx.clone(); let content_cache = content_cache.clone(); let stored_page_token = old_page_tokens.get(cur_user_email.as_str()).cloned(); + let folder_filter_ids = folder_filter_ids.clone(); async move { if can_resume_full && stored_page_token.is_some() { @@ -1801,6 +2459,7 @@ impl SyncManager { &ctx, start_token, content_cache.clone(), + folder_filter_ids.clone(), ) .await { @@ -1828,6 +2487,7 @@ impl SyncManager { &ctx, Some(&drive_cutoff_date), content_cache.clone(), + folder_filter_ids.clone(), ) .await }; @@ -1878,6 +2538,8 @@ impl SyncManager { } else { Some(new_page_tokens.clone()) }, + drive_change_tokens: None, + drive_scope_fingerprint: Some(current_fingerprint.clone()), chat: chat_checkpoint.clone(), }; ctx.save_checkpoint(serde_json::to_value(&checkpoint_state)?) @@ -1933,6 +2595,8 @@ impl SyncManager { } else { Some(new_page_tokens) }, + drive_change_tokens: None, + drive_scope_fingerprint: Some(current_fingerprint), chat: chat_checkpoint, }) } @@ -2137,6 +2801,8 @@ impl SyncManager { Some(new_history_ids.clone()) }, drive_page_tokens: drive_page_tokens.clone(), + drive_change_tokens: existing_state.drive_change_tokens.clone(), + drive_scope_fingerprint: existing_state.drive_scope_fingerprint.clone(), chat: chat_checkpoint.clone(), }; ctx.save_checkpoint(serde_json::to_value(&checkpoint_state)?) @@ -2185,6 +2851,8 @@ impl SyncManager { Some(new_history_ids) }, drive_page_tokens, + drive_change_tokens: existing_state.drive_change_tokens.clone(), + drive_scope_fingerprint: existing_state.drive_scope_fingerprint.clone(), chat: chat_checkpoint, }) } @@ -2340,6 +3008,8 @@ impl SyncManager { let checkpoint_state = GoogleSyncCheckpoint { gmail_history_ids: existing_state.gmail_history_ids.clone(), drive_page_tokens: existing_state.drive_page_tokens.clone(), + drive_change_tokens: existing_state.drive_change_tokens.clone(), + drive_scope_fingerprint: existing_state.drive_scope_fingerprint.clone(), chat: Some(chat_checkpoint.clone()), }; ctx.save_checkpoint(serde_json::to_value(&checkpoint_state)?) @@ -2349,6 +3019,8 @@ impl SyncManager { Ok(GoogleSyncCheckpoint { gmail_history_ids: existing_state.gmail_history_ids, drive_page_tokens: existing_state.drive_page_tokens, + drive_change_tokens: existing_state.drive_change_tokens, + drive_scope_fingerprint: existing_state.drive_scope_fingerprint, chat: Some(chat_checkpoint), }) } @@ -3481,6 +4153,11 @@ impl SyncManager { Ok(webhook_response) } + /// Public accessor for the Drive client (used by connector actions). + pub fn drive_client(&self) -> &DriveClient { + &self.drive_client + } + pub async fn stop_webhook_for_source( &self, source_id: &str, @@ -3542,24 +4219,30 @@ impl SyncManager { let mut path_components = vec![file_name.to_string()]; let mut current_folder_id = folder_id.to_string(); - // Build path by traversing up the folder hierarchy - let mut depth = 0; + // Build path by traversing up the folder hierarchy. + // Uses cycle detection via visited set; the iteration cap is only a safety net. + let mut visited = HashSet::new(); + let mut _iteration: usize = 0; + const MAX_ITERATIONS: usize = 10_000; loop { - depth += 1; - debug!( - "Path building depth: {}, current folder: {}", - depth, current_folder_id - ); - - // TODO: Remove this - if depth > 50 { + _iteration += 1; + if _iteration >= MAX_ITERATIONS { warn!( - "Path building depth exceeded 50 levels for file: {}, folder: {}", + "Path building exceeded maximum iterations for file: {}, folder: {}", file_name, folder_id ); break; } + // Check for cycles before processing. + if !visited.insert(current_folder_id.clone()) { + warn!( + "Cycle detected in folder ancestry for file: {}, folder: {}", + file_name, current_folder_id + ); + break; + } + let cached_folder = self.folder_cache.get(¤t_folder_id); let parent_folder_id: Option = match cached_folder { @@ -3580,7 +4263,7 @@ impl SyncManager { ); let folder_metadata = self .drive_client - .get_folder_metadata(&auth, &user_email, &folder_id) + .get_folder_metadata(&auth, &user_email, ¤t_folder_id) .await; match folder_metadata { @@ -4347,6 +5030,60 @@ impl SyncManager { } } +/// Core ancestry-check logic — the production walker used by `is_file_in_allowed_folder` +/// and tested directly with mock lookups. +/// +/// Given a file's parent IDs, an allowed set, and a function to fetch folder +/// metadata by ID, walks up the parent chain and returns true if any ancestor +/// is in the allowed set. +/// +/// `lookup` returns `(name, Option)` for a folder ID. +/// Returns Err on error, Ok(true) if found in allowed set, Ok(false) if outside. +/// Has a safety cap of 10_000 iterations to prevent infinite loops. +pub(crate) async fn check_ancestry_static( + file_parents: &Option>, + allowed_ids: &HashSet, + lookup: &(dyn Fn(String) -> Pin)>> + Send>> + + Send + + Sync), +) -> Result { + if allowed_ids.is_empty() { + return Ok(true); + } + + let mut stack: Vec = Vec::new(); + if let Some(parents) = file_parents { + for p in parents { + stack.push(p.clone()); + } + } + + let mut visited = HashSet::new(); + let mut iterations: usize = 0; + const MAX_ITERATIONS: usize = 10_000; + + while let Some(cur) = stack.pop() { + iterations += 1; + if iterations > MAX_ITERATIONS { + // Safety cap — treat as outside rather than infinite loop. + return Ok(false); + } + if !visited.insert(cur.clone()) { + // Cycle detected — skip, don't include. + continue; + } + if allowed_ids.contains(&cur) { + return Ok(true); + } + let (_name, parent) = lookup(cur).await?; + if let Some(p) = parent { + stack.push(p); + } + } + + Ok(false) +} + #[cfg(test)] mod tests { use super::*; @@ -4523,4 +5260,139 @@ mod tests { assert!(semaphore.try_acquire_many_owned(2).is_ok()); } + + // ======================================================================== + // Folder path filter logic + // ======================================================================== + + #[tokio::test] + async fn folder_filter_matches_direct_and_nested_ancestors() { + let allowed: HashSet = ["allowed".to_string()].into_iter().collect(); + let lookup = + |id: String| -> Pin)>> + Send>> { + Box::pin(async move { + let parent = match id.as_str() { + "child" => Some("allowed".to_string()), + "outside" => Some("root".to_string()), + "cycle-a" => Some("cycle-b".to_string()), + "cycle-b" => Some("cycle-a".to_string()), + _ => None, + }; + Ok((id, parent)) + }) + }; + + assert!( + check_ancestry_static(&Some(vec!["allowed".to_string()]), &allowed, &lookup) + .await + .unwrap() + ); + assert!( + check_ancestry_static(&Some(vec!["child".to_string()]), &allowed, &lookup) + .await + .unwrap() + ); + assert!( + !check_ancestry_static(&Some(vec!["outside".to_string()]), &allowed, &lookup) + .await + .unwrap() + ); + assert!( + !check_ancestry_static(&Some(vec!["cycle-a".to_string()]), &allowed, &lookup) + .await + .unwrap() + ); + } + + #[tokio::test] + async fn folder_filter_propagates_lookup_errors() { + let allowed: HashSet = ["allowed".to_string()].into_iter().collect(); + let lookup = + |id: String| -> Pin)>> + Send>> { + Box::pin(async move { Err(anyhow!("API error for {id}")) }) + }; + + assert!( + check_ancestry_static(&Some(vec!["broken".to_string()]), &allowed, &lookup) + .await + .is_err() + ); + } + + #[test] + fn drive_scope_changes_force_full_sync() { + assert!(drive_scope_requires_full_sync( + SyncType::Full, + Some("all"), + "all" + )); + assert!(drive_scope_requires_full_sync( + SyncType::Incremental, + None, + "all" + )); + assert!(drive_scope_requires_full_sync( + SyncType::Incremental, + Some("all"), + "drive:folder:id", + )); + assert!(!drive_scope_requires_full_sync( + SyncType::Incremental, + Some("drive:folder:id"), + "drive:folder:id", + )); + } + + #[test] + fn folder_scope_cutoff_uses_parsed_created_or_modified_times() { + let cutoff = parse_google_time(Some("2025-01-01T00:00:00Z")).unwrap(); + let file = |mime_type: &str, created_time: &str, modified_time: &str| { + crate::models::GoogleDriveFile { + id: "file".to_string(), + name: "File".to_string(), + mime_type: mime_type.to_string(), + web_view_link: None, + created_time: Some(created_time.to_string()), + modified_time: Some(modified_time.to_string()), + size: None, + parents: None, + shared: None, + permissions: None, + owners: None, + } + }; + + assert!(SyncManager::pass_cutoff( + &file( + "application/pdf", + "2024-01-01T00:00:00Z", + "2025-01-02T00:00:00Z" + ), + cutoff, + )); + assert!(SyncManager::pass_cutoff( + &file( + "application/pdf", + "2025-01-02T01:00:00+01:00", + "2024-01-01T00:00:00Z" + ), + cutoff, + )); + assert!(!SyncManager::pass_cutoff( + &file( + "application/pdf", + "2024-01-01T00:00:00Z", + "2024-12-31T23:59:59Z" + ), + cutoff, + )); + assert!(SyncManager::pass_cutoff( + &file( + "application/vnd.google-apps.folder", + "2020-01-01T00:00:00Z", + "2020-01-01T00:00:00Z", + ), + cutoff, + )); + } } diff --git a/services/connector-manager/src/handlers.rs b/services/connector-manager/src/handlers.rs index 546e4e16f..754e76ff4 100644 --- a/services/connector-manager/src/handlers.rs +++ b/services/connector-manager/src/handlers.rs @@ -358,159 +358,282 @@ pub async fn execute_action( _headers: HeaderMap, Json(request): Json, ) -> Result { - info!( - "Executing action '{}' for source {} (user {:?}, params keys: {:?})", - request.action, - request.source_id, - request.user_id, - request - .params - .as_object() - .map(|m| m.keys().cloned().collect::>()) - .unwrap_or_default() - ); - - let source_repo = SourceRepository::new(state.db_pool.pool()); - let source = source_repo - .find_by_id(request.source_id.clone()) - .await - .map_err(|e| ApiError::Internal(e.to_string()))? - .ok_or_else(|| ApiError::NotFound(format!("Source not found: {}", request.source_id)))?; - - // Look up the connector manifest to get connector_url and action metadata + let is_transient = request.transient_credentials.is_some(); + let source_type: SourceType; + let source: Option; + let creds: shared::models::ServiceCredential; + let mut params = request.params.clone(); + let mut transient_actor_email = None; let manifests = get_registered_manifests(&state.redis_client).await; - let manifest = manifests - .iter() - .find(|m| m.source_types.contains(&source.source_type)); - let connector_url = manifest - .as_ref() - .map(|m| m.connector_url.clone()) - .ok_or_else(|| { - ApiError::NotFound(format!( - "Connector not registered for type: {:?}", - source.source_type - )) + let (connector_url, action_admin_only) = if is_transient { + // ======== TRANSIENT MODE ======== + // Transient mode: source_type + transient_credentials required, no source/credential DB. + let tc = request.transient_credentials.as_ref().unwrap(); + source_type = request.source_type.ok_or_else(|| { + ApiError::BadRequest( + "source_type is required when transient_credentials are provided".to_string(), + ) })?; + if request.source_id.is_some() { + return Err(ApiError::BadRequest( + "source_id must not be set when transient_credentials are provided".to_string(), + )); + } - let action_def = manifest.and_then(|m| m.actions.iter().find(|a| a.name == request.action)); - let action_mode = action_def.map(|a| a.mode).unwrap_or_default(); - // Reject unknown action names — they must exist in the manifest - let action_def = action_def.ok_or_else(|| { - ApiError::BadRequest(format!( - "Unknown action '{}' for source type {:?}", - request.action, source.source_type - )) - })?; - let action_admin_only = action_def.admin_only; - - // Generic read_only enforcement — the source config is the authority. - let source_read_only = source - .config - .get("read_only") - .and_then(|v| v.as_bool()) - .unwrap_or(false); - if let Some(m) = manifest.as_ref() { - if (m.read_only || source_read_only) && action_mode == ActionMode::Write { + // Resolve user/admin from user_id (required in transient mode). + let user_id = request.user_id.as_ref().ok_or_else(|| { + ApiError::BadRequest("user_id is required in transient mode".to_string()) + })?; + + // Look up the connector manifest by source_type. + let manifest = manifests + .iter() + .find(|m| m.source_types.contains(&source_type)) + .ok_or_else(|| { + ApiError::NotFound(format!( + "Connector not registered for type: {:?}", + source_type + )) + })?; + + let connector_url = manifest.connector_url.clone(); + let action_def = manifest + .actions + .iter() + .find(|a| a.name == request.action) + .ok_or_else(|| { + ApiError::BadRequest(format!( + "Unknown action '{}' for source type {:?}", + request.action, source_type + )) + })?; + if !action_def.source_types.contains(&source_type) { return Err(ApiError::BadRequest(format!( - "Action '{}' is not allowed: source is read-only", - request.action + "Action '{}' does not support source type {:?}", + request.action, source_type ))); } - } + let action_admin_only = action_def.admin_only; + let action_mode = action_def.mode; - let creds_repo = ServiceCredentialsRepo::new(state.db_pool.pool().clone()) - .map_err(|e| ApiError::Internal(e.to_string()))?; - let creds = match resolve_credentials( - &creds_repo, - &request.source_id, - request.user_id.as_deref(), - action_admin_only, - ) - .await? - { - CredentialResolution::Resolved(c) => c, - CredentialResolution::NeedsUserAuth { provider } => { - return Ok(needs_user_auth_response( - &request.source_id, - source.source_type, - provider, - )?); - } - CredentialResolution::NoCredentials => { - return Err(ApiError::NotFound(format!( - "Credentials not found for source: {}", - request.source_id + if action_mode != ActionMode::Read { + return Err(ApiError::BadRequest(format!( + "Transient action '{}' must be read-only", + request.action ))); } - }; - - // Resolve Omni document ID -> source external_id. - // TODO: replace hard-coded param names with a connector-declared resolve_params list. - let mut params = request.params.clone(); - let doc_id = params - .get("document_id") - .or_else(|| params.get("file_id")) - .and_then(|v| v.as_str()) - .map(|s| s.to_string()); - if let Some(doc_id) = doc_id { - let doc_repo = DocumentRepository::new(state.db_pool.pool()); - if let Ok(Some(doc)) = doc_repo.find_by_id(&doc_id).await { - info!( - "Resolved document/file ID {} -> external_id {}", - doc_id, doc.external_id - ); - if let Some(obj) = params.as_object_mut() { - obj.remove("document_id"); - obj.remove("file_id"); - obj.insert( - "file_id".to_string(), - serde_json::Value::String(doc.external_id), - ); - } - } - // If not found, assume the ID is already a source-native ID and pass through - } - - // Merge source config into params for legacy connector compatibility. - // Connectors that use the typed context (e.g. Darwinbox) ignore these. - if params.is_null() { - params = serde_json::Value::Object(serde_json::Map::new()); - } - if let (Some(src_obj), Some(params_obj)) = (source.config.as_object(), params.as_object_mut()) { - for (k, v) in src_obj { - params_obj.entry(k.clone()).or_insert_with(|| v.clone()); - } - } - // Resolve actor email for the execution context - let actor_email = if let Some(uid) = request.user_id.as_ref() { let user_repo = UserRepository::new(state.db_pool.pool()); let user = user_repo - .find_by_id(uid.clone()) + .find_by_id(user_id.clone()) .await .map_err(|e| ApiError::Internal(e.to_string()))? - .ok_or_else(|| ApiError::NotFound(format!("User not found: {uid}")))?; - // Enforce admin_only action authorization + .ok_or_else(|| ApiError::NotFound(format!("User not found: {user_id}")))?; if action_admin_only && user.role != shared::models::UserRole::Admin { return Err(ApiError::BadRequest(format!( "Action '{}' requires admin privileges", request.action ))); } - Some(user.email) + transient_actor_email = Some(user.email); + + // Adapt the transient payload to the connector SDK's credential model. + let tc = tc.clone(); + creds = shared::models::ServiceCredential { + id: "transient".to_string(), + source_id: "transient".to_string(), + user_id: Some(user_id.clone()), + provider: tc.provider, + auth_type: tc.auth_type, + principal_email: tc.principal_email, + credentials: tc.credentials, + config: tc.config, + expires_at: None, + last_validated_at: None, + created_at: time::OffsetDateTime::now_utc(), + updated_at: time::OffsetDateTime::now_utc(), + }; + source = None; + + info!( + "Executing transient action '{}' for source_type {:?} (user {:?})", + request.action, source_type, request.user_id + ); + + (connector_url, action_admin_only) + } else { + // ======== PERSISTED MODE (unchanged) ======== + if request.source_type.is_some() { + return Err(ApiError::BadRequest( + "source_type is only valid with transient_credentials".to_string(), + )); + } + let source_id = request + .source_id + .as_deref() + .filter(|id| !id.is_empty()) + .ok_or_else(|| ApiError::BadRequest("source_id is required".to_string()))? + .to_string(); + + let source_repo = SourceRepository::new(state.db_pool.pool()); + let db_source = source_repo + .find_by_id(source_id.clone()) + .await + .map_err(|e| ApiError::Internal(e.to_string()))? + .ok_or_else(|| ApiError::NotFound(format!("Source not found: {source_id}")))?; + + source_type = db_source.source_type; + + let manifest = manifests + .iter() + .find(|m| m.source_types.contains(&source_type)) + .ok_or_else(|| { + ApiError::NotFound(format!( + "Connector not registered for type: {:?}", + source_type + )) + })?; + let connector_url = manifest.connector_url.clone(); + + let action_def = manifest + .actions + .iter() + .find(|a| a.name == request.action) + .ok_or_else(|| { + ApiError::BadRequest(format!( + "Unknown action '{}' for source type {:?}", + request.action, source_type + )) + })?; + if !action_def.source_types.contains(&source_type) { + return Err(ApiError::BadRequest(format!( + "Action '{}' does not support source type {:?}", + request.action, source_type + ))); + } + let action_admin_only = action_def.admin_only; + let action_mode = action_def.mode; + + // Generic read_only enforcement — the source config is the authority. + let source_read_only = db_source + .config + .get("read_only") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + if (manifest.read_only || source_read_only) && action_mode == ActionMode::Write { + return Err(ApiError::BadRequest(format!( + "Action '{}' is not allowed: source is read-only", + request.action + ))); + } + + let creds_repo = ServiceCredentialsRepo::new(state.db_pool.pool().clone()) + .map_err(|e| ApiError::Internal(e.to_string()))?; + creds = match resolve_credentials( + &creds_repo, + &source_id, + request.user_id.as_deref(), + action_admin_only, + ) + .await? + { + CredentialResolution::Resolved(c) => c, + CredentialResolution::NeedsUserAuth { provider } => { + return Ok(needs_user_auth_response(&source_id, source_type, provider)?); + } + CredentialResolution::NoCredentials => { + return Err(ApiError::NotFound(format!( + "Credentials not found for source: {source_id}" + ))); + } + }; + + // Resolve Omni document ID -> source external_id (persisted mode only). + let doc_id = params + .get("document_id") + .or_else(|| params.get("file_id")) + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + if let Some(doc_id) = doc_id { + let doc_repo = DocumentRepository::new(state.db_pool.pool()); + if let Ok(Some(doc)) = doc_repo.find_by_id(&doc_id).await { + info!( + "Resolved document/file ID {} -> external_id {}", + doc_id, doc.external_id + ); + if let Some(obj) = params.as_object_mut() { + obj.remove("document_id"); + obj.remove("file_id"); + obj.insert( + "file_id".to_string(), + serde_json::Value::String(doc.external_id), + ); + } + } + } + + // Merge source config into params for legacy connector compatibility. + if params.is_null() { + params = serde_json::Value::Object(serde_json::Map::new()); + } + if let (Some(src_obj), Some(params_obj)) = + (db_source.config.as_object(), params.as_object_mut()) + { + for (k, v) in src_obj { + params_obj.entry(k.clone()).or_insert_with(|| v.clone()); + } + } + + source = Some(db_source); + + info!( + "Executing action '{}' for source {} (user {:?}, params keys: {:?})", + request.action, + source_id, + request.user_id, + request + .params + .as_object() + .map(|m| m.keys().cloned().collect::>()) + .unwrap_or_default() + ); + + (connector_url, action_admin_only) + }; + + // ==== Common path for both modes ==== + // Resolve the actor email; transient mode already loaded the actor above. + let actor_email = if !is_transient { + if let Some(uid) = request.user_id.as_ref() { + let user_repo = UserRepository::new(state.db_pool.pool()); + let user = user_repo + .find_by_id(uid.clone()) + .await + .map_err(|e| ApiError::Internal(e.to_string()))? + .ok_or_else(|| ApiError::NotFound(format!("User not found: {uid}")))?; + // Enforce admin_only action authorization (user already validated for transient mode above). + if action_admin_only && user.role != shared::models::UserRole::Admin { + return Err(ApiError::BadRequest(format!( + "Action '{}' requires admin privileges", + request.action + ))); + } + Some(user.email) + } else { + None + } } else { - None + transient_actor_email }; + if params.is_null() { + params = serde_json::Value::Object(serde_json::Map::new()); + } + info!( - "Dispatching action '{}' to connector {} with credential {} (provider={:?}, auth_type={:?}, principal={:?})", - request.action, - connector_url, - creds.id, - creds.provider, - creds.auth_type, - creds.principal_email, + "Dispatching action '{}' to connector {} (provider={:?}, auth_type={:?}, principal={:?})", + request.action, connector_url, creds.provider, creds.auth_type, creds.principal_email, ); let client = ConnectorClient::new(); @@ -518,7 +641,7 @@ pub async fn execute_action( action: request.action, params, credentials: Some(creds), - source: Some(source), + source, actor_email, }; @@ -531,7 +654,6 @@ pub async fn execute_action( let status = response.status(); let mut builder = axum::response::Response::builder().status(status); - // Forward all headers except hop-by-hop connection headers. let hop_by_hop = [ "connection", "keep-alive", diff --git a/services/connector-manager/src/models.rs b/services/connector-manager/src/models.rs index 04e65e733..0f4cac67d 100644 --- a/services/connector-manager/src/models.rs +++ b/services/connector-manager/src/models.rs @@ -91,9 +91,26 @@ pub struct TriggerSyncResponse { pub status: String, } +/// Transient credentials accepted during setup/discovery before a source or +/// credential row exists in the database. Used by the setup UI to discover +/// connector capabilities (e.g. list shared drives) with an inline credential +/// that has not yet been persisted. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct TransientCredentials { + pub provider: shared::models::ServiceProvider, + pub auth_type: shared::models::AuthType, + pub principal_email: Option, + pub credentials: JsonValue, + #[serde(default)] + pub config: JsonValue, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ExecuteActionRequest { - pub source_id: String, + /// Source ID, required when persisted credentials should be resolved. + #[serde(default)] + pub source_id: Option, /// Acting user. `None` for org-level / system-initiated calls (sync, /// future org-level agents); `Some` for chat tool dispatch and other /// user-context invocations. @@ -102,6 +119,14 @@ pub struct ExecuteActionRequest { pub action: String, #[serde(default)] pub params: JsonValue, + /// Source type hint, required when `transient_credentials` are provided. + #[serde(default)] + pub source_type: Option, + /// Inline credentials for setup/discovery actions. When present, the request + /// is handled in **transient mode**: `source_id` and DB credentials are + /// not used, and `source_type` must identify the connector. + #[serde(default)] + pub transient_credentials: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/web/src/lib/components/google-drive-folder-selector.svelte b/web/src/lib/components/google-drive-folder-selector.svelte new file mode 100644 index 000000000..4106fc93f --- /dev/null +++ b/web/src/lib/components/google-drive-folder-selector.svelte @@ -0,0 +1,425 @@ + + +
+ +

{description}

+ + + {#if isLoading && !hasLoaded} +
+ + Discovering shared drives and folders... +
+ {/if} + + + {#if errorMessage && !isLoading} + + + Discovery Failed + +

{errorMessage}

+ + + + {/if} + + + {#if hasLoaded && allItems.length === 0 && !isLoading && !errorMessage} +

+ No shared drives found. Make sure the service account has access to shared drives in + this domain. +

+ {/if} + + + +
+ + + {#if isLoading} + + {/if} +
+ + {#if showDropdown && filteredItems.length > 0} + event.preventDefault()} + onCloseAutoFocus={(event) => event.preventDefault()}> + {#each filteredItems as item (item.id)} + + {/each} + + {/if} + + + + {#if selected.length > 0} +
+ {#each selected as item (item.id)} +
+ {item.name} + {#if !disabled} + + {/if} +
+ {/each} +
+ {:else if !isLoading && hasLoaded} +

+ No folders selected — all accessible files will be indexed. +

+ {/if} +
diff --git a/web/src/lib/components/google-workspace-setup.svelte b/web/src/lib/components/google-workspace-setup.svelte index ad59045b4..29f52a48d 100644 --- a/web/src/lib/components/google-workspace-setup.svelte +++ b/web/src/lib/components/google-workspace-setup.svelte @@ -3,13 +3,16 @@ import { Button } from '$lib/components/ui/button' import { Label } from '$lib/components/ui/label' import { Checkbox } from '$lib/components/ui/checkbox' + import * as Card from '$lib/components/ui/card' import { AuthType } from '$lib/types' + import type { FolderPathFilter } from '$lib/types' import { toast } from 'svelte-sonner' import { goto } from '$app/navigation' import googleDriveLogo from '$lib/images/icons/google-drive.svg' import gmailLogo from '$lib/images/icons/gmail.svg' import googleChatLogo from '$lib/images/icons/google-chat.svg' import GoogleServiceAccountForm from '$lib/components/google-service-account-form.svelte' + import GoogleDriveFolderSelector from '$lib/components/google-drive-folder-selector.svelte' interface Props { open: boolean @@ -27,6 +30,30 @@ let connectGmail = $state(true) let connectChat = $state(false) let isSubmitting = $state(false) + let driveFolderFilters = $state([]) + + // Credential-context guard: clear selected folder filters whenever the + // credential inputs change after initialization so filters from one set + // of credentials cannot be silently submitted with another. + let prevCredContext = $state('') + let credInitialized = $state(false) + + $effect(() => { + const token = JSON.stringify({ + sa: serviceAccountJson, + pe: principalEmail, + dm: domain, + }) + if (!credInitialized) { + credInitialized = true + prevCredContext = token + return + } + if (token !== prevCredContext) { + prevCredContext = token + driveFolderFilters = [] + } + }) async function handleSubmit() { isSubmitting = true @@ -55,9 +82,20 @@ } const credentials = { service_account_key: serviceAccountJson } - const config = { + + // Base config shared by all Google services (domain only). + const baseConfig: Record = { domain: domain || null, } + + // Drive-specific config with optional folder filters. + const driveConfig: Record = { + domain: domain || null, + } + if (driveFolderFilters.length > 0) { + driveConfig.folder_path_filters = driveFolderFilters + } + const authType = AuthType.JWT const provider = 'google' @@ -69,7 +107,7 @@ scope: 'org', name: 'Google Drive', sourceType: 'google_drive', - config, + config: driveConfig, }), }) @@ -88,7 +126,7 @@ authType: authType, principalEmail: principalEmail || null, credentials, - config, + config: driveConfig, }), }) @@ -105,7 +143,7 @@ scope: 'org', name: 'Gmail', sourceType: 'gmail', - config, + config: baseConfig, }), }) @@ -124,7 +162,7 @@ authType: authType, principalEmail: principalEmail || null, credentials: credentials, - config, + config: baseConfig, }), }) @@ -141,7 +179,7 @@ scope: 'org', name: 'Google Chat', sourceType: 'google_chat', - config, + config: baseConfig, }), }) @@ -160,7 +198,7 @@ authType: authType, principalEmail: principalEmail || null, credentials: credentials, - config, + config: baseConfig, }), }) @@ -240,6 +278,24 @@ + + {#if connectDrive} + + + Drive Folder Filters + + Optionally restrict indexing to specific shared drives and folders. + + + + + + + {/if} diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 3371ae6c8..2d75d4376 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -74,8 +74,17 @@ export interface ClickUpSourceConfig { space_filters?: string[] } +export interface FolderPathFilter { + id: string + name: string + path: string + driveId: string + kind: 'shared_drive_root' | 'folder' +} + export interface GoogleDriveSourceConfig { - // Future: shared_drive_filters, mime_type_filters, folder_path_filters, etc. + folder_path_filters?: FolderPathFilter[] + // Future: shared_drive_filters, mime_type_filters, etc. } export interface GmailSourceConfig { diff --git a/web/src/lib/types/search.ts b/web/src/lib/types/search.ts index 685d73f08..643276f01 100644 --- a/web/src/lib/types/search.ts +++ b/web/src/lib/types/search.ts @@ -158,6 +158,18 @@ export interface SearchUsersResponse { hasMore: boolean } +export interface DriveFolderDiscoveryEntry { + id: string + name: string + path: string + driveId: string + kind: 'shared_drive_root' | 'folder' +} + +export interface DriveFolderDiscoveryResponse { + items: DriveFolderDiscoveryEntry[] +} + export interface ConnectorActionResponse { status: string result?: T diff --git a/web/src/routes/(admin)/admin/settings/integrations/drive/[sourceId]/+page.server.ts b/web/src/routes/(admin)/admin/settings/integrations/drive/[sourceId]/+page.server.ts index 892ef3f9c..2bea3ba4c 100644 --- a/web/src/routes/(admin)/admin/settings/integrations/drive/[sourceId]/+page.server.ts +++ b/web/src/routes/(admin)/admin/settings/integrations/drive/[sourceId]/+page.server.ts @@ -7,6 +7,7 @@ import { serviceCredentialsRepository } from '$lib/server/repositories/service-c import { userRepository } from '$lib/server/db/users' import { getConfig } from '$lib/server/config' import { AuthType, SourceType } from '$lib/types' +import type { FolderPathFilter } from '$lib/types' export const load: PageServerLoad = async ({ params, locals }) => { requireAdmin(locals) @@ -29,7 +30,12 @@ export const load: PageServerLoad = async ({ params, locals }) => { const creds = await serviceCredentialsRepository.getOrgCredsBySourceId(source.id) const credsConfig = (creds?.config as { domain?: string } | null) ?? {} - const sourceConfig = (source.config as { domain?: string } | null) ?? {} + const sourceConfig = + (source.config as { domain?: string; folder_path_filters?: unknown } | null) ?? {} + + const folderPathFilters = Array.isArray(sourceConfig?.folder_path_filters) + ? sourceConfig.folder_path_filters + : [] const gmailSibling = await sourcesRepository.findActiveByTypeAndCreator( SourceType.GMAIL, @@ -43,9 +49,74 @@ export const load: PageServerLoad = async ({ params, locals }) => { principalEmail: creds?.principalEmail ?? '', domain: credsConfig.domain ?? sourceConfig.domain ?? '', gmailSiblingId: gmailSibling?.id ?? null, + folderPathFilters: folderPathFilters as FolderPathFilter[], } } +function parseFolderPathFilters(formData: FormData): Record[] { + const raw = formData.get('folder_path_filters') as string | null + if (!raw) return [] + let parsed: unknown + try { + parsed = JSON.parse(raw) + } catch { + throw error(400, 'folder_path_filters is not valid JSON') + } + if (!Array.isArray(parsed)) { + throw error(400, 'folder_path_filters must be a JSON array') + } + // Each entry must have all required fields with correct types and no unknown fields. + const allowedEntryKeys = ['id', 'name', 'path', 'driveId', 'kind'] + const seenIds = new Set() + for (const entry of parsed) { + if (!entry || typeof entry !== 'object') { + throw error(400, 'Each folder filter entry must be a non-null object') + } + const e = entry as Record + // Reject unknown fields. + for (const key of Object.keys(e)) { + if (!allowedEntryKeys.includes(key)) { + throw error(400, `Unknown field '${key}' in folder filter entry`) + } + } + // Validate required string fields. + if (typeof e.id !== 'string' || !e.id) { + throw error(400, 'Each folder filter entry must have a non-empty id') + } + if (typeof e.name !== 'string' || !e.name) { + throw error(400, 'Each folder filter entry must have a non-empty name') + } + if (typeof e.path !== 'string' || !e.path) { + throw error(400, 'Each folder filter entry must have a non-empty path') + } + if (typeof e.driveId !== 'string' || !e.driveId) { + throw error(400, 'Each folder filter entry must have a non-empty driveId') + } + if (e.kind !== 'shared_drive_root' && e.kind !== 'folder') { + throw error( + 400, + "Each folder filter entry must have kind 'shared_drive_root' or 'folder'", + ) + } + // Deduplicate by stable ID (first-wins). + if (seenIds.has(e.id)) { + continue + } + seenIds.add(e.id) + } + // Return deduplicated array — first-wins. + const deduplicated: Record[] = [] + seenIds.clear() + for (const entry of parsed) { + const e = entry as Record + const id = e.id as string + if (seenIds.has(id)) continue + seenIds.add(id) + deduplicated.push(e) + } + return deduplicated +} + export const actions: Actions = { default: async ({ request, params, locals, fetch }) => { const user = locals.user @@ -109,9 +180,30 @@ export const actions: Actions = { } } + const folderPathFilters = parseFolderPathFilters(formData) + + // Merge folder_path_filters into config while PRESERVING all existing source config keys. + const existingConfig: Record = + source.config && + typeof source.config === 'object' && + !Array.isArray(source.config) + ? (source.config as Record) + : {} + + const mergedConfig: Record = { ...existingConfig, domain } + mergedConfig.folder_path_filters = folderPathFilters + + // Preserve existing credential config as well. + const existingCredConfig: Record = + existingCreds?.config && + typeof existingCreds.config === 'object' && + !Array.isArray(existingCreds.config) + ? (existingCreds.config as Record) + : {} + await serviceCredentialsRepository.updateBySourceId(source.id, { principalEmail, - config: { domain }, + config: { ...existingCredConfig, domain }, credentials: serviceAccountJson ? { service_account_key: serviceAccountJson } : null, @@ -122,11 +214,24 @@ export const actions: Actions = { userFilterMode, userWhitelist, userBlacklist, - config: { domain }, + config: mergedConfig, }) } else { // OAuth or other auth types — admin can only toggle enabled. - await updateSourceById(source.id, { isActive }) + // Still merge any config changes without destroying existing keys. + const existingConfig: Record = + source.config && + typeof source.config === 'object' && + !Array.isArray(source.config) + ? (source.config as Record) + : {} + const folderPathFilters = parseFolderPathFilters(formData) + const mergedConfig: Record = { ...existingConfig } + mergedConfig.folder_path_filters = folderPathFilters + await updateSourceById(source.id, { + isActive, + config: mergedConfig, + }) } if (isActive) { diff --git a/web/src/routes/(admin)/admin/settings/integrations/drive/[sourceId]/+page.svelte b/web/src/routes/(admin)/admin/settings/integrations/drive/[sourceId]/+page.svelte index aa8d23f7b..51a04c76a 100644 --- a/web/src/routes/(admin)/admin/settings/integrations/drive/[sourceId]/+page.svelte +++ b/web/src/routes/(admin)/admin/settings/integrations/drive/[sourceId]/+page.svelte @@ -10,6 +10,7 @@ import { Badge } from '$lib/components/ui/badge' import { Search, X, AlertCircle, Info, Loader2 } from '@lucide/svelte' import GoogleServiceAccountForm from '$lib/components/google-service-account-form.svelte' + import GoogleDriveFolderSelector from '$lib/components/google-drive-folder-selector.svelte' import { onMount } from 'svelte' import { beforeNavigate } from '$app/navigation' import type { PageProps } from './$types' @@ -19,6 +20,7 @@ ConnectorActionResponse, } from '$lib/types/search' import { AuthType } from '$lib/types' + import type { FolderPathFilter } from '$lib/types' import googleDriveLogo from '$lib/images/icons/google-drive.svg' let { data }: PageProps = $props() @@ -33,6 +35,10 @@ let principalEmail = $state(data.principalEmail) let domain = $state(data.domain) + let folderFilters = $state( + (Array.isArray(data.folderPathFilters) ? data.folderPathFilters : []) as FolderPathFilter[], + ) + let searchQuery = $state('') let searchResults = $state([]) let isSearching = $state(false) @@ -50,6 +56,7 @@ let originalSelectedUsers: string[] = [] let originalPrincipalEmail = data.principalEmail let originalDomain = data.domain + let originalFolderFilters = $state([...folderFilters]) async function searchUsers() { if (searchQuery.trim().length < 2) { @@ -196,10 +203,15 @@ const usersChanged = JSON.stringify(selectedUsers.sort()) !== JSON.stringify(originalSelectedUsers.sort()) + const foldersChanged = + JSON.stringify(folderFilters.map((f) => f.id).sort()) !== + JSON.stringify(originalFolderFilters.map((f) => f.id).sort()) + hasUnsavedChanges = enabled !== originalEnabled || userFilterMode !== originalUserFilterMode || usersChanged || + foldersChanged || principalEmail !== originalPrincipalEmail || domain !== originalDomain || serviceAccountJson.trim().length > 0 @@ -286,6 +298,15 @@ Integrations settings. + + + Folder-level filtering unavailable + + To select specific folders or shared drives for indexing, switch to a + domain-wide delegation (service account) connection. OAuth-connected + sources always index all accessible files. + + {:else if isJwt}
@@ -314,6 +335,29 @@ {/if}
+ +
+
+

Drive Folder Filters

+

+ Optionally restrict indexing to specific shared drives and top-level + folders. All sub-folders within a selected item are included. +

+
+ + + +
+

User Access Control

diff --git a/web/src/routes/api/connectors/[sourceType]/action/+server.ts b/web/src/routes/api/connectors/[sourceType]/action/+server.ts new file mode 100644 index 000000000..386e61101 --- /dev/null +++ b/web/src/routes/api/connectors/[sourceType]/action/+server.ts @@ -0,0 +1,193 @@ +import { json, error } from '@sveltejs/kit' +import type { RequestHandler } from './$types' +import { getConfig } from '$lib/server/config' +import { logger } from '$lib/server/logger' +import { SourceType } from '$lib/types' + +/** + * Invokes an action directly on a connector before a source exists. + * + * The transient credential is forwarded to connector-manager's normal /action + * endpoint and is never stored or returned. + * + * SECURITY: Strictly allowlisted — currently only google_drive's + * discover_folders action with JWT credentials is accepted. + */ +export const POST: RequestHandler = async ({ params: routeParams, request, locals }) => { + // Admin-only + if (!locals.user) { + throw error(401, 'Unauthorized') + } + if (locals.user.role !== 'admin') { + throw error(403, 'Admin access required') + } + + // Enforce body size limit BEFORE reading — check Content-Length header first. + const contentLength = request.headers.get('content-length') + if (contentLength && parseInt(contentLength, 10) > 64 * 1024) { + throw error(413, 'Request body too large') + } + // Read body via streaming reader; stop and cancel once 64KiB is exceeded. + // Never call arrayBuffer/text first which would buffer the full body. + let bodyBytes = new Uint8Array(0) + const reader = request.body?.getReader() + const MAX_BYTES = 64 * 1024 + if (reader) { + try { + while (true) { + const { done, value } = await reader.read() + if (done) break + if (bodyBytes.length + value.length > MAX_BYTES) { + await reader.cancel() + throw error(413, 'Request body too large') + } + const newBody = new Uint8Array(bodyBytes.length + value.length) + newBody.set(bodyBytes, 0) + newBody.set(value, bodyBytes.length) + bodyBytes = newBody + } + } catch (err) { + // If we already threw a 413, re-throw it directly rather than wrapping. + if (err && typeof err === 'object' && 'status' in err) { + throw err + } + // Otherwise wrap as parse error — empty/null body will land here. + throw error(400, 'Failed to read request body') + } + } + const text = new TextDecoder().decode(bodyBytes) + + let body: Record + try { + const parsed = JSON.parse(text) + if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) { + throw error(400, 'Request body must be a JSON object') + } + body = parsed as Record + } catch (err) { + if (err && typeof err === 'object' && 'status' in err) { + throw err + } + throw error(400, 'Invalid JSON body') + } + + // ======== STRICT ALLOWLIST ======== + // Reject unknown top-level fields. + const allowedFields = ['action', 'params', 'serviceAccountJson', 'principalEmail', 'domain'] + for (const key of Object.keys(body)) { + if (!allowedFields.includes(key)) { + throw error(400, `Unknown field: '${key}'`) + } + } + + const sourceType = routeParams.sourceType + const action = body.action as string | undefined + const params = body.params as Record | undefined + const serviceAccountJson = body.serviceAccountJson as string | undefined + const principalEmail = body.principalEmail as string | undefined + const domain = body.domain as string | undefined + + if (!action) { + throw error(400, 'Action is required') + } + if (action !== 'discover_folders') { + throw error(400, `Connector action '${action}' is not supported`) + } + if (sourceType !== SourceType.GOOGLE_DRIVE) { + throw error(400, `Connector actions are not supported for source type '${sourceType}'`) + } + + // Only allow empty params object for discover_folders. + if (params !== undefined) { + if (typeof params !== 'object' || params === null || Array.isArray(params)) { + throw error(400, 'params must be a JSON object') + } + if (Object.keys(params).length > 0) { + throw error(400, 'discover_folders does not accept any params') + } + } + + // Transient credentials must be provided for preview (not optional). + if (!serviceAccountJson || !principalEmail || !domain) { + throw error( + 400, + 'serviceAccountJson, principalEmail, and domain are required for connector actions', + ) + } + + // Validate types are strings. + if (typeof serviceAccountJson !== 'string') { + throw error(400, 'serviceAccountJson must be a string') + } + if (typeof principalEmail !== 'string') { + throw error(400, 'principalEmail must be a string') + } + if (typeof domain !== 'string') { + throw error(400, 'domain must be a string') + } + + // Validate service account JSON + try { + JSON.parse(serviceAccountJson) + } catch { + throw error(400, 'Invalid service account JSON') + } + + try { + const config = getConfig() + const connectorManagerUrl = config.services.connectorManagerUrl + + // Forward the setup credential through connector-manager's generic transient action mode. + const response = await fetch(`${connectorManagerUrl}/action`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + source_type: sourceType, + user_id: locals.user.id, + action: 'discover_folders', + params: params || {}, + transient_credentials: { + provider: 'google', + auth_type: 'jwt', + principal_email: principalEmail, + credentials: { + service_account_key: serviceAccountJson, + }, + config: { + domain: domain, + }, + }, + }), + signal: AbortSignal.timeout(30_000), + }) + + if (!response.ok) { + let errorMessage = 'Failed to discover folders' + const contentType = response.headers.get('content-type') || '' + if (contentType.includes('application/json')) { + try { + const errorBody = await response.json() + errorMessage = errorBody.error || errorBody.message || errorMessage + } catch { + // ignore parse errors + } + } else { + errorMessage = (await response.text()) || errorMessage + } + logger.error(`Connector action failed`, { + status: response.status, + error: errorMessage, + }) + throw error(response.status, errorMessage) + } + + const result = await response.json() + return json(result) + } catch (err) { + if (err && typeof err === 'object' && 'status' in err) { + throw err + } + logger.error('Error executing connector action:', err) + throw error(500, 'Internal server error') + } +} diff --git a/web/src/routes/api/connectors/[sourceType]/action/action.test.ts b/web/src/routes/api/connectors/[sourceType]/action/action.test.ts new file mode 100644 index 000000000..46e99e921 --- /dev/null +++ b/web/src/routes/api/connectors/[sourceType]/action/action.test.ts @@ -0,0 +1,138 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' + +// Mock server dependencies before importing the handler. +vi.mock('$lib/server/config', () => ({ + getConfig: vi.fn(() => ({ + database: { url: 'postgresql://test:test@localhost:5432/test' }, + redis: { url: 'redis://localhost:6379' }, + services: { + searcherUrl: 'http://searcher.test', + indexerUrl: 'http://indexer.test', + aiServiceUrl: 'http://ai.test', + connectorManagerUrl: 'http://cm.test', + }, + session: { secret: 'test-secret', cookieName: 'test-cookie', durationDays: 30 }, + app: { publicUrl: 'http://localhost:3000' }, + })), +})) + +vi.mock('$lib/server/logger', () => ({ + logger: { error: vi.fn() }, +})) + +const { POST } = await import('./+server') + +function mockRequest(body: unknown, headers: Record = {}): Request { + return new Request('http://localhost/api/connectors/google_drive/action', { + method: 'POST', + headers: { 'Content-Type': 'application/json', ...headers }, + body: JSON.stringify(body), + }) +} + +function mockLocals(): Record { + return { user: { id: 'admin-1', role: 'admin' } } +} + +/** + * Helper that calls POST and converts both success and thrown HttpError + * into a consistent { status, body? } shape so tests don't need .catch(). + */ +async function callPost( + body: unknown, + locals?: Record, + sourceType = 'google_drive', +): Promise<{ status: number; body?: unknown }> { + try { + const response = await POST({ + params: { sourceType }, + request: mockRequest(body), + locals: locals ?? mockLocals(), + } as unknown as Parameters[0]) + return { status: response.status, body: await response.json().catch(() => undefined) } + } catch (err: unknown) { + // SvelteKit error() throws HttpError { status, body } + const httpError = err as { status?: number; body?: unknown } + if (httpError.status !== undefined) { + return { status: httpError.status, body: httpError.body } + } + throw err + } +} + +describe('POST /api/connectors/[sourceType]/action', () => { + afterEach(() => { + vi.unstubAllGlobals() + }) + + it('requires an admin', async () => { + const request = { + action: 'discover_folders', + serviceAccountJson: '{}', + principalEmail: 'a@b.com', + domain: 'b.com', + } + + expect((await callPost(request, { user: null })).status).toBe(401) + expect((await callPost(request, { user: { id: 'u1', role: 'member' } })).status).toBe(403) + }) + + it('rejects connector source types without an allowlisted transient action', async () => { + const response = await callPost( + { + action: 'discover_folders', + serviceAccountJson: '{}', + principalEmail: 'a@b.com', + domain: 'b.com', + }, + undefined, + 'slack', + ) + + expect(response.status).toBe(400) + }) + + it('forwards validated setup credentials through the normal action endpoint', async () => { + const connectorResponse = { + status: 'success', + result: { items: [] }, + } + const connectorFetch = vi.fn().mockResolvedValue( + new Response(JSON.stringify(connectorResponse), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }), + ) + vi.stubGlobal('fetch', connectorFetch) + + const response = await callPost({ + action: 'discover_folders', + serviceAccountJson: '{"client_email":"service@example.com"}', + principalEmail: 'admin@example.com', + domain: 'example.com', + params: {}, + }) + + expect(response).toEqual({ status: 200, body: connectorResponse }) + expect(connectorFetch).toHaveBeenCalledOnce() + const [url, init] = connectorFetch.mock.calls[0] as [string, RequestInit] + expect(url).toBe('http://cm.test/action') + const forwarded = JSON.parse(String(init.body)) as Record + expect(forwarded).not.toHaveProperty('source_id') + expect(forwarded).toMatchObject({ + source_type: 'google_drive', + user_id: 'admin-1', + action: 'discover_folders', + params: {}, + transient_credentials: { + provider: 'google', + auth_type: 'jwt', + principal_email: 'admin@example.com', + credentials: { + service_account_key: '{"client_email":"service@example.com"}', + }, + config: { domain: 'example.com' }, + }, + }) + }) +}) diff --git a/web/src/routes/api/sources/+server.ts b/web/src/routes/api/sources/+server.ts index 7fb15629b..4078996d8 100644 --- a/web/src/routes/api/sources/+server.ts +++ b/web/src/routes/api/sources/+server.ts @@ -7,7 +7,6 @@ import { ulid } from 'ulid' import { logger } from '$lib/server/logger' import { SourceType, DEFAULT_SYNC_INTERVAL_SECONDS } from '$lib/types' import { getSourcesByType } from '$lib/server/db/sources' -import { env } from '$env/dynamic/private' export const GET: RequestHandler = async ({ locals }) => { if (!locals.user) { @@ -77,7 +76,8 @@ export const GET: RequestHandler = async ({ locals }) => { } export const POST: RequestHandler = async ({ request, locals }) => { - if (!locals.user) { + const user = locals.user + if (!user) { throw error(401, 'Unauthorized') } @@ -92,7 +92,7 @@ export const POST: RequestHandler = async ({ request, locals }) => { throw error(400, 'Name and sourceType are required') } - if (scope === 'org' && locals.user.role !== 'admin') { + if (scope === 'org' && user.role !== 'admin') { throw error(403, 'Only admins can create org-wide sources') } const sourcesOfType = await getSourcesByType(sourceType) @@ -104,7 +104,7 @@ export const POST: RequestHandler = async ({ request, locals }) => { const uniqueSourceTypes: string[] = [SourceType.GOOGLE_DRIVE, SourceType.GMAIL] if (uniqueSourceTypes.includes(sourceType)) { const existingForUser = sourcesOfType.find( - (s) => s.scope === 'user' && s.createdBy === locals.user.id, + (s) => s.scope === 'user' && s.createdBy === user.id, ) if (existingForUser) { throw error(409, `A ${sourceType} source already exists`) @@ -119,8 +119,8 @@ export const POST: RequestHandler = async ({ request, locals }) => { name, sourceType, scope, - config: validatedConfig || {}, - createdBy: locals.user.id, + config: config || {}, + createdBy: user.id, isActive: isActive ?? false, syncIntervalSeconds: DEFAULT_SYNC_INTERVAL_SECONDS[sourceType as SourceType], })