> ## Documentation Index
> Fetch the complete documentation index at: https://docs.pipeshub.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Connector manager
# Connector Manager API
The Connector Manager Service provides centralized management of authentication tokens and connector configurations for third-party integrations. This service handles OAuth flows, token lifecycle management, and connector configurations for various external services like Google Workspace, OneDrive, SharePoint, Confluence, and more.
## Base URL
All endpoints are prefixed with `/api/v1/connectors`
## Authentication
All endpoints require authentication via Bearer token:
```http theme={null}
Authorization: Bearer YOUR_TOKEN
```
**Internal endpoints** use scoped token authentication for service-to-service communication with specific scopes:
* `FETCH_CONFIG` - For configuration updates and token refresh operations
## Architecture Overview
The Connector Manager Service is built on a Node.js backend with MongoDB for data persistence, Redis for caching, and ETCD for secure configuration storage. The service consists of several key components:
* **Token Management** - Handles OAuth token lifecycle including creation, refresh, and revocation
* **Connector Configuration** - Manages connector settings and credentials
* **Event Broadcasting** - Kafka-based events for system integration
* **Health Monitoring** - Service health checks and dependency monitoring
The service integrates with these components:
* **Configuration Manager** - Manages encrypted configuration storage
* **IAM Service** - Handles user authentication and authorization
* **Kafka** - Event streaming for token and connector state changes
* **ETCD** - Encrypted key-value storage for sensitive configurations
* **Backend Connector Service** - Python service that handles actual connector implementations
## Data Models
### Connectors
Connector configurations that define enabled/disabled state and metadata for third-party integrations.
### Token Events
Events broadcast when tokens are created, refreshed, revoked, or expired.
### Entity Events
Events broadcast when connectors are enabled or disabled, triggering synchronization workflows.
### Configuration
Encrypted storage of OAuth credentials, API keys, and service endpoints.
## API Endpoints
Manage connector configurations including retrieval, updates, OAuth flows, and filtering options.
Retrieve all connectors for the organization (admin access required).
**Endpoint:** `GET /api/v1/connectors/`
**Access Control:** Admin privileges required
**Authentication:** Bearer token required
**Status:** `200 OK`
**Note:** Response format depends on the backend connector service implementation.
```json theme={null}
{
"connectors": [
{
"name": "google_workspace",
"isEnabled": true,
"lastUpdated": "2025-04-27T12:00:00.000Z"
}
]
}
```
Retrieve all currently active/enabled connectors.
**Endpoint:** `GET /api/v1/connectors/active`
**Authentication:** Bearer token required
**Status:** `200 OK`
**Note:** Response format depends on the backend connector service implementation.
```json theme={null}
{
"success": true,
"connectors": [
{
"name": "OneDrive",
"appGroup": "Microsoft 365",
"authType": "OAUTH_ADMIN_CONSENT",
"appDescription": "Sync files and folders from OneDrive",
"appCategories": [
"Storage"
],
"iconPath": "/assets/icons/connectors/onedrive.svg",
"supportsRealtime": false,
"supportsSync": true,
"config": {
"iconPath": "/assets/icons/connectors/onedrive.svg",
"supportsRealtime": false,
"supportsSync": true,
"documentationLinks": [
{
"title": "Azure AD App Registration Setup",
"url": "https://docs.microsoft.com/en-us/azure/active-directory/develop/quickstart-register-app",
"type": "setup"
}
],
"auth": {
"type": "OAUTH",
"displayRedirectUri": false,
"redirectUri": "http://localhost:3000/OneDrive/oauth/callback",
"schema": {
"fields": [
{
"name": "clientId",
"displayName": "Application (Client) ID",
"placeholder": "Enter your Azure AD Application ID",
"description": "The Application (Client) ID from Azure AD App Registration",
"fieldType": "TEXT",
"required": true,
"defaultValue": "",
"validation": {
"minLength": 10,
"maxLength": 1000
},
"isSecret": false
},
{
"name": "clientSecret",
"displayName": "Client Secret",
"placeholder": "Enter your Azure AD Client Secret",
"description": "The Client Secret from Azure AD App Registration",
"fieldType": "PASSWORD",
"required": true,
"defaultValue": "",
"validation": {
"minLength": 10,
"maxLength": 1000
},
"isSecret": true
},
{
"name": "tenantId",
"displayName": "Directory (Tenant) ID",
"placeholder": "Enter your Azure AD Tenant ID",
"description": "The Directory (Tenant) ID from Azure AD",
"fieldType": "TEXT",
"required": true,
"defaultValue": "",
"validation": {
"minLength": 10,
"maxLength": 1000
},
"isSecret": false
},
{
"name": "hasAdminConsent",
"displayName": "Has Admin Consent",
"placeholder": "",
"description": "Check if admin consent has been granted for the application",
"fieldType": "CHECKBOX",
"required": true,
"defaultValue": false,
"validation": {
"minLength": 10,
"maxLength": 1000
},
"isSecret": false
},
{
"name": "redirectUri",
"displayName": "Redirect URI",
"placeholder": "http://localhost:3000/OneDrive/oauth/callback",
"description": "The redirect URI for OAuth authentication",
"fieldType": "URL",
"required": false,
"defaultValue": "",
"validation": {
"minLength": 10,
"maxLength": 2000
},
"isSecret": false
}
]
},
"values": {},
"customFields": [],
"customValues": {},
"conditionalDisplay": {
"redirectUri": {
"showWhen": {
"field": "hasAdminConsent",
"operator": "equals",
"value": false
}
}
}
},
"sync": {
"supportedStrategies": [
"SCHEDULED",
"MANUAL"
],
"selectedStrategy": "MANUAL",
"webhookConfig": {
"supported": false,
"webhookUrl": "",
"events": [],
"verificationToken": "",
"secretKey": ""
},
"scheduledConfig": {
"intervalMinutes": 60,
"cronExpression": "",
"timezone": "UTC",
"startTime": 0,
"nextTime": 0,
"endTime": 0,
"maxRepetitions": 0,
"repetitionCount": 0
},
"realtimeConfig": {
"supported": false,
"connectionType": "WEBSOCKET"
},
"customFields": [],
"customValues": {},
"values": {}
},
"filters": {
"schema": {
"fields": [
{
"name": "fileTypes",
"displayName": "File Types",
"description": "Select the types of files to sync",
"fieldType": "MULTISELECT",
"required": false,
"defaultValue": [],
"options": [
"document",
"spreadsheet",
"presentation",
"pdf",
"image",
"video"
],
"operators": [
"IN",
"NOT_IN"
]
},
{
"name": "folders",
"displayName": "Folders",
"description": "Select folders to sync from",
"fieldType": "MULTISELECT",
"required": false,
"defaultValue": [],
"options": [],
"operators": [
"IN",
"NOT_IN"
]
}
]
},
"values": {},
"customFields": [],
"customValues": {},
"endpoints": {
"fileTypes": "static",
"folders": "https://graph.microsoft.com/v1.0/me/drive/root/children"
}
}
},
"isActive": true,
"isConfigured": true,
"createdAtTimestamp": 1759184704366,
"updatedAtTimestamp": 1759184917316
}
]
}
```
Retrieve all currently inactive/disabled connectors.
**Endpoint:** `GET /api/v1/connectors/inactive`
**Authentication:** Bearer token required
**Status:** `200 OK`
**Note:** Response format depends on the backend connector service implementation.
```json theme={null}
{
"success": true,
"connectors": [
{
"name": "Linear",
"appGroup": "Linear",
"authType": "OAUTH_ADMIN_CONSENT",
"appDescription": "Modern issue tracking and project management",
"appCategories": [
"Ticket Management"
],
"iconPath": "/assets/icons/connectors/linear.svg",
"supportsRealtime": false,
"supportsSync": true,
"config": {
"iconPath": "/assets/icons/connectors/linear.svg",
"supportsRealtime": false,
"supportsSync": true,
"documentationLinks": [
{
"title": "Linear Registration Setup",
"url": "https://linear.app/developers",
"type": "setup"
}
],
"auth": {
"type": "OAUTH",
"displayRedirectUri": false,
"redirectUri": "http://localhost:3000/linear/oauth/callback",
"schema": {
"fields": [
// fields for auth
]
},
"values": {},
"customFields": [],
"customValues": {},
"conditionalDisplay": {
"redirectUri": {
"showWhen": {
"field": "hasAdminConsent",
"operator": "equals",
"value": false
}
}
}
},
"sync": {
"supportedStrategies": [
"SCHEDULED",
"MANUAL"
],
"selectedStrategy": "MANUAL",
"webhookConfig": {
"supported": false,
"webhookUrl": "",
"events": [],
"verificationToken": "",
"secretKey": ""
},
"scheduledConfig": {
"intervalMinutes": 60,
"cronExpression": "",
"timezone": "UTC",
"startTime": 0,
"nextTime": 0,
"endTime": 0,
"maxRepetitions": 0,
"repetitionCount": 0
},
"realtimeConfig": {
"supported": false,
"connectionType": "WEBSOCKET"
},
"customFields": [],
"customValues": {},
"values": {}
},
"filters": {
"schema": {
"fields": [
{
// types of documents to fetch
}
]
},
"values": {},
"customFields": [],
"customValues": {},
"endpoints": {
"fileTypes": "static",
"folders": "https://api.linear.app/graphql"
}
}
},
"isActive": false,
"isConfigured": true,
"createdAtTimestamp": 1759184704366,
"updatedAtTimestamp": 1759184917316
}
]
}
```
Retrieve configuration for a specific connector.
**Endpoint:** `GET /api/v1/connectors/config/:connectorName`
**Path Parameters:**
* `connectorName`: Connector name (string, required)
**Access Control:** Admin privileges required
```http theme={null}
GET /api/v1/connectors/config/OneDrive
```
**Status:** `200 OK`
**Note:** Response format depends on the backend connector service implementation.
```json theme={null}
{
"success": true,
"config": {
"name": "OneDrive",
"appGroupId": null,
"appGroup": "Microsoft 365",
"authType": "OAUTH_ADMIN_CONSENT",
"appDescription": "Sync files and folders from OneDrive",
"appCategories": [
"Storage"
],
"supportsRealtime": false,
"supportsSync": true,
"iconPath": "/assets/icons/connectors/onedrive.svg",
"config": {
"auth": {},
"sync": {},
"filters": {}
},
"isActive": false,
"isConfigured": false
}
}
```
Update configuration for a specific connector.
**Endpoint:** `PUT /api/v1/connectors/config/:connectorName`
**Path Parameters:**
* `connectorName`: Connector name (string, required)
**Access Control:** Admin privileges required
**Request Body Parameters:**
| Parameter | Type | Required | Description |
| --------- | ------ | -------- | ---------------------------- |
| auth | object | No | Authentication configuration |
| sync | object | No | Synchronization settings |
| filters | object | No | Filter configurations |
| baseUrl | string | Yes | Base URL for the connector |
```json theme={null}
{
"auth": {
"clientId": "xxxxxxx",
"clientSecret": "xxxxxxxx",
"tenantId": "xxxxxxx",
"hasAdminConsent": true,
"redirectUri": ""
},
"sync": {
"selectedStrategy": "SCHEDULED",
"scheduledConfig": {}
},
"filters": {}
}
```
**Status:** `200 OK`
**Note:** Response format depends on the backend connector service implementation.
```json theme={null}
{
"success": true,
"config": {
"auth": {
"clientId": ".....",
"clientSecret": "your-oauth-secret",
"hasAdminConsent": true,
"redirectUri": ""
},
"sync": {
"selectedStrategy": "SCHEDULED",
"scheduledConfig": {}
},
"filters": {},
"credentials": null,
"oauth": null
}
}
```
Retrieve the configuration schema for a specific connector type.
**Endpoint:** `GET /api/v1/connectors/schema/:connectorName`
**Path Parameters:**
* `connectorName`: Connector name (string, required)
**Access Control:** Admin privileges required
**Status:** `200 OK`
**Note:** Response format depends on the backend connector service implementation.
```json theme={null}
{
"success": true,
"schema": {
"iconPath": "/assets/icons/connectors/onedrive.svg",
"supportsRealtime": false,
"supportsSync": true,
"documentationLinks": [
{
"title": "Azure AD App Registration Setup",
"url": "https://docs.microsoft.com/en-us/azure/active-directory/develop/quickstart-register-app",
"type": "setup"
}
],
"auth": {
"type": "OAUTH",
"displayRedirectUri": false,
"redirectUri": "http://localhost:3000/OneDrive/oauth/callback",
"schema": {
"fields": [
{
"name": "clientId",
"displayName": "Application (Client) ID",
"placeholder": "Enter your Azure AD Application ID",
"description": "The Application (Client) ID from Azure AD App Registration",
"fieldType": "TEXT",
"required": true,
"defaultValue": "",
"validation": {
"minLength": 10,
"maxLength": 1000
},
"isSecret": false
},
{
"name": "clientSecret",
"displayName": "Client Secret",
"placeholder": "Enter your Azure AD Client Secret",
"description": "The Client Secret from Azure AD App Registration",
"fieldType": "PASSWORD",
"required": true,
"defaultValue": "",
"validation": {
"minLength": 10,
"maxLength": 1000
},
"isSecret": true
},
{
"name": "tenantId",
"displayName": "Directory (Tenant) ID",
"placeholder": "Enter your Azure AD Tenant ID",
"description": "The Directory (Tenant) ID from Azure AD",
"fieldType": "TEXT",
"required": true,
"defaultValue": "",
"validation": {
"minLength": 10,
"maxLength": 1000
},
"isSecret": false
},
{
"name": "hasAdminConsent",
"displayName": "Has Admin Consent",
"placeholder": "",
"description": "Check if admin consent has been granted for the application",
"fieldType": "CHECKBOX",
"required": true,
"defaultValue": false,
"validation": {
"minLength": 10,
"maxLength": 1000
},
"isSecret": false
},
{
"name": "redirectUri",
"displayName": "Redirect URI",
"placeholder": "http://localhost:3000/OneDrive/oauth/callback",
"description": "The redirect URI for OAuth authentication",
"fieldType": "URL",
"required": false,
"defaultValue": "",
"validation": {
"minLength": 10,
"maxLength": 2000
},
"isSecret": false
}
]
},
"values": {},
"customFields": [],
"customValues": {},
"conditionalDisplay": {
"redirectUri": {
"showWhen": {
"field": "hasAdminConsent",
"operator": "equals",
"value": false
}
}
}
},
"sync": {
"supportedStrategies": [
"SCHEDULED",
"MANUAL"
],
"selectedStrategy": "MANUAL",
"webhookConfig": {
"supported": false,
"webhookUrl": "",
"events": [],
"verificationToken": "",
"secretKey": ""
},
"scheduledConfig": {
"intervalMinutes": 60,
"cronExpression": "",
"timezone": "UTC",
"startTime": 0,
"nextTime": 0,
"endTime": 0,
"maxRepetitions": 0,
"repetitionCount": 0
},
"realtimeConfig": {
"supported": false,
"connectionType": "WEBSOCKET"
},
"customFields": [],
"customValues": {},
"values": {}
},
"filters": {
"schema": {
"fields": [
{
"name": "fileTypes",
"displayName": "File Types",
"description": "Select the types of files to sync",
"fieldType": "MULTISELECT",
"required": false,
"defaultValue": [],
"options": [
"document",
"spreadsheet",
"presentation",
"pdf",
"image",
"video"
],
"operators": [
"IN",
"NOT_IN"
]
},
{
"name": "folders",
"displayName": "Folders",
"description": "Select folders to sync from",
"fieldType": "MULTISELECT",
"required": false,
"defaultValue": [],
"options": [],
"operators": [
"IN",
"NOT_IN"
]
}
]
},
"values": {},
"customFields": [],
"customValues": {},
"endpoints": {
"fileTypes": "static",
"folders": "https://graph.microsoft.com/v1.0/me/drive/root/children"
}
}
}
}
```
Enable or disable a specific connector.
**Endpoint:** `POST /api/v1/connectors/toggle/:connectorName`
**Path Parameters:**
* `connectorName`: Connector name (string, required)
**Access Control:** Admin privileges required
**Status:** `200 OK`
**Note:** Response format depends on the backend connector service implementation.
```json theme={null}
{
"success":true,
"message":"Connector OneDrive toggled successfully"
}
```
Handle OAuth authorization flows for connector authentication.
Generate OAuth authorization URL for a connector.
**Endpoint:** `GET /api/v1/connectors/:connectorName/oauth/authorize`
**Path Parameters:**
* `connectorName`: Connector name (string, required)
**Query Parameters:**
| Parameter | Type | Required | Description |
| --------- | ------ | -------- | ---------------------------------- |
| baseUrl | string | Yes | Base URL for the connector service |
**Access Control:** Admin privileges required
```http theme={null}
GET /api/v1/connectors/google_workspace/oauth/authorize?baseUrl=https://api.example.com
```
**Status:** `200 OK`
**Note:** Response format depends on the backend connector service implementation.
```json theme={null}
{
"authorizationUrl": "https://accounts.google.com/oauth/authorize?client_id=CLIENT_ID&redirect_uri=REDIRECT_URI&scope=SCOPES&response_type=code&state=STATE"
}
```
Handle OAuth callback after user authorization.
**Endpoint:** `GET /api/v1/connectors/:connectorName/oauth/callback`
**Path Parameters:**
* `connectorName`: Connector name (string, required)
**Query Parameters:**
| Parameter | Type | Required | Description |
| --------- | ------ | -------- | -------------------------------------- |
| code | string | No | Authorization code from OAuth provider |
| state | string | No | State parameter for CSRF protection |
| error | string | No | Error code if authorization failed |
| baseUrl | string | Yes | Base URL for the connector service |
**Access Control:** Admin privileges required
```http theme={null}
GET /api/v1/connectors/google_workspace/oauth/callback?code=AUTH_CODE&state=STATE&baseUrl=https://api.example.com
```
**Status:** `200 OK`
**Success Response:**
```json theme={null}
{
"redirectUrl": "https://app.example.com/connectors/success?connector=google_workspace"
}
```
**Error Response:**
```json theme={null}
{
"redirectUrl": "https://app.example.com/connectors/error?error=access_denied"
}
```
**Note:** The backend service may return a 302 redirect, which is converted to a JSON response with `redirectUrl` to avoid CORS issues.
Exchange OAuth authorization code for access and refresh tokens (Google Workspace specific).
**Endpoint:** `POST /api/v1/connectors/getTokenFromCode`
**Access Control:** Admin privileges required
**Request Body Parameters:**
| Parameter | Type | Required | Description |
| --------- | ------ | -------- | ------------------------------------ |
| tempCode | string | Yes | OAuth authorization code from Google |
```json theme={null}
{
"tempCode": "4/0AX4XfWjYX3Z9X8Y7W6V5U4T3S2R1Q0P"
}
```
**Process:**
1. Retrieves Google Workspace configuration from Configuration Manager
2. Exchanges authorization code for access and refresh tokens via Google's token endpoint
3. Verifies the ID token matches the authenticated user's email
4. Stores credentials securely via Configuration Manager
5. Creates or updates connector in MongoDB
6. Publishes connector enabled event to Kafka
7. Determines enabled apps based on OAuth scopes received
**Status:** `200 OK` (Existing Connector)
```json theme={null}
{
"message": "Connector is now enabled",
"connector": {
"_id": "60d21b4667d0d8992e610c85",
"name": "Google Workspace",
"isEnabled": true,
"lastUpdatedBy": "user456",
"orgId": "org123",
"updatedAt": "2025-04-27T12:25:00.000Z"
}
}
```
**Status:** `201 Created` (New Connector)
```json theme={null}
{
"message": "Connector Google Workspace created and enabled",
"connector": {
"_id": "60d21b4667d0d8992e610c86",
"name": "Google Workspace",
"isEnabled": true,
"lastUpdatedBy": "user456",
"orgId": "org123",
"createdAt": "2025-04-27T12:25:00.000Z"
}
}
```
Manage connector-specific filtering options and configurations.
Retrieve filter options for a specific connector.
**Endpoint:** `GET /api/v1/connectors/filters/:connectorName`
**Path Parameters:**
* `connectorName`: Connector name (string, required)
**Access Control:** Admin privileges required
**Status:** `200 OK`
**Note:** Response format depends on the backend connector service implementation.
```json theme={null}
{
"connectorName": "google_workspace",
"filters": {
"fileTypes": ["pdf", "docx", "txt"],
"folders": {
"includeSharedDrives": true,
"excludedFolders": ["/Archive"]
}
}
}
```
Save or update filter options for a specific connector.
**Endpoint:** `POST /api/v1/connectors/filters/:connectorName`
**Path Parameters:**
* `connectorName`: Connector name (string, required)
**Access Control:** Admin privileges required
**Request Body Parameters:**
| Parameter | Type | Required | Description |
| ------------- | ------ | -------- | --------------------------- |
| filterOptions | object | Yes | Filter configuration object |
```json theme={null}
{
"filterOptions": {
"fileTypes": ["pdf", "docx"],
"folders": {
"includeSharedDrives": false,
"excludedFolders": ["/Archive", "/Personal"]
},
"dateRange": {
"enabled": true,
"from": "2024-01-01",
"to": "2024-12-31"
}
}
}
```
**Status:** `200 OK`
**Note:** Response format depends on the backend connector service implementation.
```json theme={null}
{
"message": "Filter options saved successfully",
"connectorName": "google_workspace"
}
```
Internal endpoints for service-to-service communication and system maintenance.
Refresh an expired access token using a refresh token (Google Workspace specific).
**Endpoint:** `POST /api/v1/connectors/internal/refreshIndividualConnectorToken`
**Authentication:** Requires scoped token with `FETCH_CONFIG` scope
**Process:**
1. Retrieves refresh token from Configuration Manager
2. Exchanges refresh token for new access token via Google's token endpoint
3. Implements retry logic with exponential backoff (3 attempts)
4. Updates stored credentials via Configuration Manager
5. Maintains existing refresh token and expiry times
**Note:** This endpoint automatically retrieves the refresh token from secure storage and exchanges it for a new access token.
**Status:** `200 OK`
```json theme={null}
{
"message": "accesstoken updated Successfully"
}
```
Update the application configuration for the connector manager service.
**Endpoint:** `POST /api/v1/connectors/updateAppConfig`
**Authentication:** Requires scoped token with `FETCH_CONFIG` scope
**Note:** This endpoint reloads the application configuration from the configuration manager and updates the dependency injection container.
**Status:** `200 OK`
```json theme={null}
{
"message": "Connectors configuration updated successfully",
"config": {
"jwtSecret": "[REDACTED]",
"scopedJwtSecret": "[REDACTED]",
"frontendUrl": "https://app.example.com",
"connectorBackend": "https://connectors.example.com"
}
}
```
**Note:** Sensitive configuration values are redacted in the response.
Monitor the health and status of the Connector Manager Service and its dependencies.
Retrieve the health status of the service and all its dependencies.
**Endpoint:** `GET /api/v1/health`
**Authentication:** No authentication required for health checks
**Status:** `200 OK`
**Healthy Response:**
```json theme={null}
{
"status": "healthy",
"timestamp": "2025-04-27T12:00:00.000Z",
"services": {
"redis": "healthy",
"kafka": "healthy",
"mongodb": "healthy",
"arangodb": "healthy",
"KVStoreservice": "healthy"
}
}
```
**Unhealthy Response:**
```json theme={null}
{
"status": "unhealthy",
"timestamp": "2025-04-27T12:00:00.000Z",
"services": {
"redis": "healthy",
"kafka": "unhealthy",
"mongodb": "healthy",
"arangodb": "healthy",
"KVStoreservice": "unhealthy"
}
}
```
**Service Status Values:**
* `healthy` - Service is operational
* `unhealthy` - Service is down or experiencing issues
* `unknown` - Service status could not be determined
**Dependencies Checked:**
* **Redis** - Cache and session storage
* **Kafka** - Event streaming for token and connector events
* **MongoDB** - Connector configuration and metadata storage
* **ArangoDB** - Knowledge base service dependency
* **KVStoreservice** - ETCD key-value store for encrypted configuration
## Event System
The Connector Manager Service broadcasts events through Kafka to notify other services about token and connector state changes. These events trigger actions like connector synchronization, indexing updates, and audit logging.
### Event Topics
| Topic | Description |
| --------------- | ------------------------------------------------------------------ |
| `token-events` | Token lifecycle events (creation, refresh, revocation, expiration) |
| `entity-events` | Connector state change events (enabled, disabled) |
### Event Types
| Event Type | Description |
| ----------------- | -------------------------------------- |
| `TOKEN_CREATED` | Triggered when a new token is acquired |
| `TOKEN_REFRESHED` | Triggered when a token is refreshed |
| `TOKEN_REVOKED` | Triggered when a token is revoked |
| `TOKEN_EXPIRED` | Triggered when a token expires |
| `appEnabled` | Triggered when a connector is enabled |
| `appDisabled` | Triggered when a connector is disabled |
### Event Payload Structures
```json theme={null}
{
"eventId": "event_123456",
"eventType": "TOKEN_REFRESHED",
"tokenReferenceId": "token_ref_789",
"serviceType": "GOOGLE_DRIVE",
"accountId": "user@example.com",
"timestamp": 1714208400000
}
```
**Service Types:**
* `ONEDRIVE`
* `GOOGLE_DRIVE`
* `CONFLUENCE`
* `JIRA`
```json theme={null}
{
"eventType": "appEnabled",
"timestamp": 1714208400000,
"payload": {
"orgId": "org123456",
"appGroup": "Google Workspace",
"appGroupId": "connector_123",
"credentialsRoute": "https://cm.example.com/api/v1/configurationManager/internal/connectors/googleWorkspaceCredentials",
"refreshTokenRoute": "https://cm.example.com/api/v1/connectors/internal/refreshIndividualConnectorToken",
"apps": ["DRIVE", "GMAIL", "CALENDAR"],
"syncAction": "immediate"
}
}
```
**Sync Actions:**
* `none` - No synchronization required
* `immediate` - Start synchronization immediately
* `scheduled` - Schedule synchronization for later
```json theme={null}
{
"eventType": "appDisabled",
"timestamp": 1714208400000,
"payload": {
"orgId": "org123456",
"appGroup": "Google Workspace",
"appGroupId": "connector_123",
"apps": ["DRIVE", "GMAIL", "CALENDAR"]
}
}
```
## Schema Definitions
```typescript theme={null}
interface AppConfig {
// Authentication secrets
jwtSecret: string;
scopedJwtSecret: string;
cookieSecret: string;
rsAvailable: string;
// Service endpoints
communicationBackend: string;
frontendUrl: string;
iamBackend: string;
authBackend: string;
cmBackend: string;
kbBackend: string;
esBackend: string;
storageBackend: string;
tokenBackend: string;
aiBackend: string;
connectorBackend: string;
connectorPublicUrl: string;
indexingBackend: string;
// Infrastructure configuration
kafka: {
brokers: string[];
sasl?: {
mechanism: 'plain' | 'scram-sha-256' | 'scram-sha-512';
username: string;
password: string;
};
};
redis: {
host: string;
port: number;
password?: string;
db?: number;
};
mongo: {
uri: string;
db: string;
};
qdrant: {
port: number;
apiKey: string;
host: string;
grpcPort: number;
};
arango: {
url: string;
db: string;
username: string;
password: string;
};
etcd: {
host: string;
port: number;
dialTimeout: number;
};
smtp: {
host: string;
port: number;
username?: string;
password?: string;
fromEmail: string;
} | null;
storage: {
storageType: string;
endpoint: string;
};
}
```
```typescript theme={null}
enum ServiceType {
ONEDRIVE = 'ONEDRIVE',
GOOGLE_DRIVE = 'GOOGLE_DRIVE',
CONFLUENCE = 'CONFLUENCE',
JIRA = 'JIRA',
}
enum TokenEventType {
TOKEN_CREATED = 'TOKEN_CREATED',
TOKEN_REFRESHED = 'TOKEN_REFRESHED',
TOKEN_REVOKED = 'TOKEN_REVOKED',
TOKEN_EXPIRED = 'TOKEN_EXPIRED',
}
interface ITokenEvent {
eventId: string;
eventType: TokenEventType;
tokenReferenceId: string;
serviceType: ServiceType;
accountId: string;
timestamp: Number;
}
```
```typescript theme={null}
enum SyncAction {
None = 'none',
Immediate = 'immediate',
Scheduled = 'scheduled',
}
enum EventType {
AppEnabledEvent = 'appEnabled',
AppDisabledEvent = 'appDisabled',
}
interface Event {
eventType: EventType;
timestamp: number;
payload: AppEnabledEvent | AppDisabledEvent;
}
interface AppEnabledEvent {
orgId: string;
appGroup: string;
appGroupId: string;
credentialsRoute?: string;
refreshTokenRoute?: string;
apps: string[];
syncAction: SyncAction;
}
interface AppDisabledEvent {
orgId: string;
appGroup: string;
appGroupId: string;
apps: string[];
}
```
```typescript theme={null}
enum GoogleWorkspaceApp {
Drive = 'DRIVE',
Gmail = 'GMAIL',
Calendar = 'CALENDAR',
}
interface GoogleWorkspaceApps {
apps: GoogleWorkspaceApp[];
}
const scopeToAppMap: { [key: string]: GoogleWorkspaceApp } = {
'https://www.googleapis.com/auth/gmail.readonly': GoogleWorkspaceApp.Gmail,
'https://www.googleapis.com/auth/calendar.readonly': GoogleWorkspaceApp.Calendar,
'https://www.googleapis.com/auth/drive.readonly': GoogleWorkspaceApp.Drive,
};
```
```typescript theme={null}
interface HealthStatus {
status: 'healthy' | 'unhealthy';
timestamp: string;
services: {
redis: string;
kafka: string;
mongodb: string;
arangodb: string;
KVStoreservice: string;
};
}
```
```typescript theme={null}
// Update connector config validation
const updateConnectorConfigSchema = z.object({
body: z.object({
auth: z.any(),
sync: z.any(),
filters: z.any(),
baseUrl: z.string(),
}),
params: z.object({
connectorName: z.string(),
}),
});
// OAuth authorization URL validation
const getOAuthAuthorizationUrlSchema = z.object({
params: z.object({
connectorName: z.string(),
}),
query: z.object({
baseUrl: z.string(),
}),
});
// OAuth callback validation
const handleOAuthCallbackSchema = z.object({
params: z.object({
connectorName: z.string(),
}),
query: z.object({
baseUrl: z.string(),
code: z.string().optional(),
state: z.string().optional(),
error: z.string().optional(),
}),
});
```
## Configuration Management
The Connector Manager Service uses ETCD as a key-value store for managing configurations securely. All sensitive information, such as OAuth client IDs/secrets and access tokens, are encrypted using AES-256 before storage.
### Configuration Types
The service manages several types of configurations:
1. **Service Connection Information**
* Database connection strings (MongoDB, ArangoDB)
* Message broker details (Kafka brokers, authentication)
* Cache settings (Redis host, port, credentials)
2. **Service Discovery**
* Backend service endpoints (IAM, Auth, Configuration Manager, etc.)
* Frontend URL for OAuth redirects
3. **Authentication Secrets**
* JWT signing keys
* Scoped JWT secrets for service-to-service communication
* Cookie encryption secrets
4. **Connector Credentials**
* OAuth client IDs and secrets
* Access tokens and refresh tokens
* Token expiration timestamps
### Security Measures
* All sensitive configuration data is encrypted using AES-256 before storage in ETCD
* Encryption keys are securely managed and not stored in plaintext
* Scoped JWT tokens are used for internal service communication
* Token refresh operations include retry logic with exponential backoff
## Error Handling
All endpoints return structured error responses with specific HTTP status codes:
```json theme={null}
{
"error": {
"code": "VALIDATION_ERROR",
"message": "Invalid request parameters",
"details": "Connector name is required"
},
"meta": {
"requestId": "req-error-123",
"timestamp": "2025-04-27T13:20:00.000Z"
}
}
```
**Common HTTP Status Codes:**
* `200` - Success
* `201` - Created (for new resources)
* `400` - Bad Request (validation errors, missing parameters)
* `401` - Unauthorized (missing or invalid authentication)
* `403` - Forbidden (insufficient privileges, admin required)
* `404` - Not Found (connector or resource not found)
* `500` - Internal Server Error (backend service failures)
**Backend Service Error Handling:**
* Connection refused → "Connector Service is currently unavailable. Please check your network connection or try again later."
* Service-specific errors → Mapped to appropriate HTTP status codes with detailed error messages
* Automatic retry logic with exponential backoff for failed requests (3 retries)
## Integration Architecture
### Backend Connector Service Integration
The Connector Manager Service acts as a proxy to a Python-based Backend Connector Service that handles the actual connector implementations. Most endpoints forward requests to this backend service and return the responses.
**Request Flow:**
1. Client sends request to Connector Manager Service
2. Connector Manager validates authentication and authorization
3. Request is forwarded to Backend Connector Service
4. Backend Connector Service processes the request
5. Response is returned through Connector Manager to client
### Configuration Manager Integration
The Connector Manager Service integrates closely with the Configuration Manager for:
1. **Secure Credential Storage** - OAuth client IDs, secrets, and tokens
2. **Service Discovery** - Endpoint URLs for backend services
3. **Configuration Management** - Connector settings and preferences
4. **Encrypted Data Handling** - All sensitive data is encrypted before storage
### Google Workspace OAuth Flow
For Google Workspace connectors, the service implements a complete OAuth 2.0 flow:
1. **Authorization Request** - Generate authorization URL with appropriate scopes
2. **Code Exchange** - Exchange authorization code for access and refresh tokens
3. **Token Verification** - Verify ID token matches authenticated user
4. **Credential Storage** - Securely store tokens via Configuration Manager
5. **Connector Activation** - Enable connector and publish events
6. **Scope Mapping** - Map OAuth scopes to enabled applications (Drive, Gmail, Calendar)
### Token Lifecycle Management
The service provides automatic token management with the following features:
1. **Token Acquisition** - OAuth flow handling and token exchange
2. **Token Refresh** - Automatic refresh before expiration with retry logic
3. **Token Revocation** - Secure token deletion and cleanup
4. **Event Broadcasting** - Kafka events for token state changes
5. **Error Handling** - Retry logic and fallback mechanisms