diff --git a/apps/docs/content/docs/en/cli/knowledge.mdx b/apps/docs/content/docs/en/cli/knowledge.mdx index a2fb532b6d8..d09f27e1dad 100644 --- a/apps/docs/content/docs/en/cli/knowledge.mdx +++ b/apps/docs/content/docs/en/cli/knowledge.mdx @@ -216,7 +216,7 @@ sim knowledge create [options] ## Create knowledge connector ```bash -sim knowledge connectors create [options] +sim knowledge connectors create [options] ``` **Arguments** @@ -225,7 +225,7 @@ sim knowledge connectors create [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Unique knowledge base identifier. | +| `knowledgeBaseId` | Yes | Unique knowledge base identifier. | @@ -246,7 +246,7 @@ sim knowledge connectors create [options] ## Delete knowledge connector ```bash -sim knowledge connectors delete [options] +sim knowledge connectors delete [options] ``` **Arguments** @@ -255,7 +255,7 @@ sim knowledge connectors delete [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -275,7 +275,7 @@ sim knowledge connectors delete [options] ## Get knowledge connector ```bash -sim knowledge connectors get +sim knowledge connectors get ``` **Arguments** @@ -284,7 +284,7 @@ sim knowledge connectors get | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -292,7 +292,7 @@ sim knowledge connectors get ## List knowledge connector documents ```bash -sim knowledge connectors documents list [options] +sim knowledge connectors documents list [options] ``` **Arguments** @@ -301,7 +301,7 @@ sim knowledge connectors documents list [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -318,10 +318,38 @@ sim knowledge connectors documents list [options] +## Update knowledge connector documents + +```bash +sim knowledge connectors documents update [options] +``` + +**Arguments** + + + +| Argument | Required | Description | +| --- | --- | --- | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | +| `connectorId` | Yes | Connector selected for the operation. | + + + +**Options** + + + +| Option | Required | Description | +| --- | --- | --- | +| `--operation ` | Yes | Whether to restore or exclude the selected documents. Accepted values: `restore`, `exclude`. | +| `--document ` | Yes | Connector document identifiers to update. (space-separated, or @path / @- with one value per line). | + + + ## List knowledge connectors ```bash -sim knowledge connectors list [options] +sim knowledge connectors list [options] ``` **Arguments** @@ -330,7 +358,7 @@ sim knowledge connectors list [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Unique knowledge base identifier. | +| `knowledgeBaseId` | Yes | Unique knowledge base identifier. | @@ -346,10 +374,10 @@ sim knowledge connectors list [options] -## Update knowledge connector +## Queue a knowledge connector synchronization ```bash -sim knowledge connectors update [options] +sim knowledge connectors sync [options] ``` **Arguments** @@ -358,7 +386,7 @@ sim knowledge connectors update [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -369,16 +397,15 @@ sim knowledge connectors update [options] | Option | Required | Description | | --- | --- | --- | -| `--source-config ` | No | Replacement source selection and filtering configuration. (JSON, or @path / @- to read a file or stdin). | -| `--sync-interval-minutes ` | No | New scheduled synchronization interval in minutes. | -| `--status ` | No | New connector state. Accepted values: `active`, `paused`. | +| `--rehydrate` | No | Re-fetch and re-index every existing connector document. | +| `--no-rehydrate` | No | Send --rehydrate as false. | -## Update knowledge connector documents +## Update knowledge connector ```bash -sim knowledge connectors documents update [options] +sim knowledge connectors update [options] ``` **Arguments** @@ -387,7 +414,7 @@ sim knowledge connectors documents update [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -398,8 +425,9 @@ sim knowledge connectors documents update [options] | Option | Required | Description | | --- | --- | --- | -| `--operation ` | Yes | Whether to restore or exclude the selected documents. Accepted values: `restore`, `exclude`. | -| `--document ` | Yes | Connector document identifiers to update. (space-separated, or @path / @- with one value per line). | +| `--source-config ` | No | Replacement source selection and filtering configuration. Updating a runnable connector queues synchronization; paused connectors remain paused. (JSON, or @path / @- to read a file or stdin). | +| `--sync-interval-minutes ` | No | New scheduled synchronization interval in minutes. | +| `--status ` | No | New connector state. Accepted values: `active`, `paused`. | @@ -588,34 +616,6 @@ sim knowledge search [options] -## Sync knowledge connector - -```bash -sim knowledge sync create [options] -``` - -**Arguments** - - - -| Argument | Required | Description | -| --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | -| `connectorId` | Yes | Connector selected for the operation. | - - - -**Options** - - - -| Option | Required | Description | -| --- | --- | --- | -| `--rehydrate` | No | Re-fetch and re-index every existing connector document. | -| `--no-rehydrate` | No | Send --rehydrate as false. | - - - ## Update knowledge base ```bash diff --git a/apps/docs/content/docs/en/cli/reference.mdx b/apps/docs/content/docs/en/cli/reference.mdx index 4447a60c831..de5a6bd8f34 100644 --- a/apps/docs/content/docs/en/cli/reference.mdx +++ b/apps/docs/content/docs/en/cli/reference.mdx @@ -1236,7 +1236,7 @@ sim knowledge create [options] Create Knowledge Connector ```bash -sim knowledge connectors create [options] +sim knowledge connectors create [options] ``` **Arguments** @@ -1245,7 +1245,7 @@ sim knowledge connectors create [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Unique knowledge base identifier. | +| `knowledgeBaseId` | Yes | Unique knowledge base identifier. | @@ -1268,7 +1268,7 @@ sim knowledge connectors create [options] Delete Knowledge Connector ```bash -sim knowledge connectors delete [options] +sim knowledge connectors delete [options] ``` **Arguments** @@ -1277,7 +1277,7 @@ sim knowledge connectors delete [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -1299,7 +1299,7 @@ sim knowledge connectors delete [options] Get Knowledge Connector ```bash -sim knowledge connectors get +sim knowledge connectors get ``` **Arguments** @@ -1308,7 +1308,7 @@ sim knowledge connectors get | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -1318,7 +1318,7 @@ sim knowledge connectors get List Knowledge Connector Documents ```bash -sim knowledge connectors documents list [options] +sim knowledge connectors documents list [options] ``` **Arguments** @@ -1327,7 +1327,7 @@ sim knowledge connectors documents list [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -1344,12 +1344,42 @@ sim knowledge connectors documents list [options] +### sim knowledge connectors documents update + +Update Knowledge Connector Documents + +```bash +sim knowledge connectors documents update [options] +``` + +**Arguments** + + + +| Argument | Required | Description | +| --- | --- | --- | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | +| `connectorId` | Yes | Connector selected for the operation. | + + + +**Options** + + + +| Option | Required | Description | +| --- | --- | --- | +| `--operation ` | Yes | Whether to restore or exclude the selected documents. Accepted values: `restore`, `exclude`. | +| `--document ` | Yes | Connector document identifiers to update. (space-separated, or @path / @- with one value per line). | + + + ### sim knowledge connectors list List Knowledge Connectors ```bash -sim knowledge connectors list [options] +sim knowledge connectors list [options] ``` **Arguments** @@ -1358,7 +1388,7 @@ sim knowledge connectors list [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Unique knowledge base identifier. | +| `knowledgeBaseId` | Yes | Unique knowledge base identifier. | @@ -1374,12 +1404,12 @@ sim knowledge connectors list [options] -### sim knowledge connectors update +### sim knowledge connectors sync -Update Knowledge Connector +Queue a knowledge connector synchronization ```bash -sim knowledge connectors update [options] +sim knowledge connectors sync [options] ``` **Arguments** @@ -1388,7 +1418,7 @@ sim knowledge connectors update [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -1399,18 +1429,17 @@ sim knowledge connectors update [options] | Option | Required | Description | | --- | --- | --- | -| `--source-config ` | No | Replacement source selection and filtering configuration. (JSON, or @path / @- to read a file or stdin). | -| `--sync-interval-minutes ` | No | New scheduled synchronization interval in minutes. | -| `--status ` | No | New connector state. Accepted values: `active`, `paused`. | +| `--rehydrate` | No | Re-fetch and re-index every existing connector document. | +| `--no-rehydrate` | No | Send --rehydrate as false. | -### sim knowledge connectors documents update +### sim knowledge connectors update -Update Knowledge Connector Documents +Update Knowledge Connector ```bash -sim knowledge connectors documents update [options] +sim knowledge connectors update [options] ``` **Arguments** @@ -1419,7 +1448,7 @@ sim knowledge connectors documents update [options] | Argument | Required | Description | | --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | +| `knowledgeBaseId` | Yes | Knowledge base that owns the connector. | | `connectorId` | Yes | Connector selected for the operation. | @@ -1430,8 +1459,9 @@ sim knowledge connectors documents update [options] | Option | Required | Description | | --- | --- | --- | -| `--operation ` | Yes | Whether to restore or exclude the selected documents. Accepted values: `restore`, `exclude`. | -| `--document ` | Yes | Connector document identifiers to update. (space-separated, or @path / @- with one value per line). | +| `--source-config ` | No | Replacement source selection and filtering configuration. Updating a runnable connector queues synchronization; paused connectors remain paused. (JSON, or @path / @- to read a file or stdin). | +| `--sync-interval-minutes ` | No | New scheduled synchronization interval in minutes. | +| `--status ` | No | New connector state. Accepted values: `active`, `paused`. | @@ -1638,36 +1668,6 @@ sim knowledge search [options] -### sim knowledge sync create - -Sync Knowledge Connector - -```bash -sim knowledge sync create [options] -``` - -**Arguments** - - - -| Argument | Required | Description | -| --- | --- | --- | -| `id` | Yes | Knowledge base that owns the connector. | -| `connectorId` | Yes | Connector selected for the operation. | - - - -**Options** - - - -| Option | Required | Description | -| --- | --- | --- | -| `--rehydrate` | No | Re-fetch and re-index every existing connector document. | -| `--no-rehydrate` | No | Send --rehydrate as false. | - - - ### sim knowledge update Update Knowledge Base diff --git a/apps/docs/openapi-v2-knowledge.json b/apps/docs/openapi-v2-knowledge.json index 40fc4e450f6..2a3ee6dd098 100644 --- a/apps/docs/openapi-v2-knowledge.json +++ b/apps/docs/openapi-v2-knowledge.json @@ -766,7 +766,7 @@ "patch": { "operationId": "updateKnowledgeConnector", "summary": "Update Knowledge Connector", - "description": "Update connector source configuration, schedule, or active state. Authentication material cannot be changed through this operation. A workspace API key is rejected with `403`; use a personal API key.", + "description": "Update connector source configuration, schedule, or active state. Replacing source configuration on a runnable connector queues an immediate synchronization; paused connectors retain the change without synchronizing until resumed. Source configuration cannot be replaced while synchronization is already in progress. Authentication material cannot be changed through this operation. A workspace API key is rejected with `403`; use a personal API key.", "tags": ["Knowledge Bases"], "parameters": [ { @@ -3988,7 +3988,7 @@ "description": "Workspace that owns the knowledge base." }, "sourceConfig": { - "description": "Replacement source selection and filtering configuration.", + "description": "Replacement source selection and filtering configuration. Updating a runnable connector queues synchronization; paused connectors remain paused.", "type": "object", "propertyNames": { "type": "string" diff --git a/apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts b/apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts index bfa2f27c880..969c620e32a 100644 --- a/apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts +++ b/apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts @@ -6,6 +6,7 @@ import { import { defineInternalJsonRoute, internalRateLimits } from '@/lib/api/server/routes' import { internalKnowledgeAnalytics, + resolveInternalKnowledgeBillingAttribution, toInternalKnowledgeConnector, toInternalKnowledgeConnectorDetail, } from '@/lib/knowledge/api/internal-route' @@ -47,10 +48,12 @@ export const PATCH = defineInternalJsonRoute({ reason: 'Preserve existing internal connector-update behavior', }), errorPolicy: internalKnowledgeErrorPolicies.connectors, - mapInput: ({ params, body }) => ({ + mapInput: ({ params, body }, { principal, request }) => ({ connectorId: params.connectorId, knowledgeBaseId: params.id, updates: body, + resolveBillingAttribution: (workspaceId: string) => + resolveInternalKnowledgeBillingAttribution(request, principal, workspaceId), source: 'ui' as const, }), useCase: updateKnowledgeConnector, diff --git a/apps/sim/app/api/knowledge/connectors/sync/route.ts b/apps/sim/app/api/knowledge/connectors/sync/route.ts index f58f8f83bd0..4c98bbaf197 100644 --- a/apps/sim/app/api/knowledge/connectors/sync/route.ts +++ b/apps/sim/app/api/knowledge/connectors/sync/route.ts @@ -296,6 +296,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => { const dueConnectors = await db .select({ id: knowledgeConnector.id, + nextSyncAt: knowledgeConnector.nextSyncAt, workspaceId: knowledgeBase.workspaceId, }) .from(knowledgeConnector) @@ -328,7 +329,12 @@ export const GET = withRouteHandler(async (request: NextRequest) => { throw new Error(`Connector ${connector.id} is missing workspace billing context`) } const billingAttribution = await resolveSystemBillingAttribution(connector.workspaceId) - await dispatchSync(connector.id, { billingAttribution, requestId }) + await dispatchSync(connector.id, { + billingAttribution, + expectedNextSyncAt: connector.nextSyncAt ?? undefined, + requestId, + requireRunnable: true, + }) } catch (error) { logger.error(`[${requestId}] Failed to dispatch sync for connector ${connector.id}`, error) } diff --git a/apps/sim/app/api/v2/knowledge/[id]/connectors/route.test.ts b/apps/sim/app/api/v2/knowledge/[id]/connectors/route.test.ts index 685a477429a..d2e58cc9b38 100644 --- a/apps/sim/app/api/v2/knowledge/[id]/connectors/route.test.ts +++ b/apps/sim/app/api/v2/knowledge/[id]/connectors/route.test.ts @@ -264,6 +264,31 @@ describe('v2 knowledge connector routes', () => { expect(mocks.connectorRemoved).toHaveBeenCalledOnce() }) + it('passes source changes to application billing without an adapter resolver', async () => { + const response = await updateConnector( + request(`/api/v2/knowledge/${KNOWLEDGE_BASE_ID}/connectors/${CONNECTOR_ID}`, 'PATCH', { + workspaceId: WORKSPACE_ID, + sourceConfig: { pageIds: ['page-2'] }, + }), + connectorContext + ) + + expect(response.status).toBe(200) + expect(mocks.update).toHaveBeenCalledWith( + expect.objectContaining({ + input: expect.objectContaining({ + assertedWorkspaceId: WORKSPACE_ID, + updates: expect.objectContaining({ sourceConfig: { pageIds: ['page-2'] } }), + }), + }) + ) + expect(mocks.update).toHaveBeenCalledWith( + expect.objectContaining({ + input: expect.not.objectContaining({ resolveBillingAttribution: expect.anything() }), + }) + ) + }) + it('queues connector synchronization without an adapter billing resolver', async () => { const response = await syncConnector( request(`/api/v2/knowledge/${KNOWLEDGE_BASE_ID}/connectors/${CONNECTOR_ID}/sync`, 'POST', { diff --git a/apps/sim/background/knowledge-connector-sync.test.ts b/apps/sim/background/knowledge-connector-sync.test.ts index 39a2d8f4ca6..b817ea2d636 100644 --- a/apps/sim/background/knowledge-connector-sync.test.ts +++ b/apps/sim/background/knowledge-connector-sync.test.ts @@ -60,6 +60,8 @@ describe('knowledge connector sync worker', () => { connectorId: 'connector-1', requestId: 'request-1', fullSync: true, + requireRunnable: true, + dispatchToken: 'dispatch-1', billingAttribution: BILLING_ATTRIBUTION, }) @@ -72,7 +74,9 @@ describe('knowledge connector sync worker', () => { expect(mockExecuteSync).toHaveBeenCalledWith('connector-1', { billingAttribution: BILLING_ATTRIBUTION, fullSync: true, + requireRunnable: true, rehydrate: undefined, + dispatchToken: 'dispatch-1', }) }) @@ -94,7 +98,9 @@ describe('knowledge connector sync worker', () => { expect(mockExecuteSync).toHaveBeenCalledWith('connector-1', { billingAttribution: BILLING_ATTRIBUTION, fullSync: undefined, + requireRunnable: undefined, rehydrate: true, + dispatchToken: undefined, }) }) }) diff --git a/apps/sim/background/knowledge-connector-sync.ts b/apps/sim/background/knowledge-connector-sync.ts index 1076103324c..c4805bef9fd 100644 --- a/apps/sim/background/knowledge-connector-sync.ts +++ b/apps/sim/background/knowledge-connector-sync.ts @@ -10,8 +10,15 @@ import { CONNECTOR_SYNC_MAX_DURATION_SECONDS } from '@/lib/knowledge/connectors/ const logger = createLogger('TriggerKnowledgeConnectorSync') export async function executeConnectorSyncJob(payload: unknown) { - const { connectorId, fullSync, rehydrate, requestId, billingAttribution, dispatchToken } = - assertConnectorSyncPayload(payload) + const { + connectorId, + fullSync, + requireRunnable, + rehydrate, + requestId, + billingAttribution, + dispatchToken, + } = assertConnectorSyncPayload(payload) logger.info(`[${requestId}] Starting connector sync: ${connectorId}`) @@ -19,6 +26,7 @@ export async function executeConnectorSyncJob(payload: unknown) { const result = await executeSync(connectorId, { billingAttribution, fullSync, + requireRunnable, rehydrate, dispatchToken, }) diff --git a/apps/sim/connectors/linear/linear.test.ts b/apps/sim/connectors/linear/linear.test.ts new file mode 100644 index 00000000000..873612db586 --- /dev/null +++ b/apps/sim/connectors/linear/linear.test.ts @@ -0,0 +1,37 @@ +/** + * @vitest-environment node + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { mockFetchWithRetry } = vi.hoisted(() => ({ + mockFetchWithRetry: vi.fn(), +})) + +vi.mock('@/lib/knowledge/documents/utils', () => ({ + fetchWithRetry: mockFetchWithRetry, + VALIDATE_RETRY_OPTIONS: { maxRetries: 0 }, +})) + +import { linearConnector } from '@/connectors/linear/linear' + +describe('linearConnector authentication', () => { + beforeEach(() => { + vi.clearAllMocks() + mockFetchWithRetry.mockResolvedValue( + new Response(JSON.stringify({ data: { teams: { nodes: [{ id: 'team-1' }] } } }), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }) + ) + }) + + it.each([ + ['personal API key', 'lin_api_example', 'lin_api_example'], + ['OAuth access token', 'opaque-oauth-token', 'Bearer opaque-oauth-token'], + ])('sends a %s with the documented authorization shape', async (_kind, token, expected) => { + await expect(linearConnector.validateConfig(token, {})).resolves.toEqual({ valid: true }) + + const request = mockFetchWithRetry.mock.calls[0]?.[1] as RequestInit | undefined + expect(new Headers(request?.headers).get('Authorization')).toBe(expected) + }) +}) diff --git a/apps/sim/connectors/linear/linear.ts b/apps/sim/connectors/linear/linear.ts index 3a1639b311f..93aab253e64 100644 --- a/apps/sim/connectors/linear/linear.ts +++ b/apps/sim/connectors/linear/linear.ts @@ -7,6 +7,7 @@ import { fetchWithRetry, VALIDATE_RETRY_OPTIONS } from '@/lib/knowledge/document import { linearConnectorMeta } from '@/connectors/linear/meta' import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types' import { joinTagArray, parseMultiValue, parseTagDate } from '@/connectors/utils' +import { linearAuthorizationHeader } from '@/tools/linear/utils' const logger = createLogger('LinearConnector') @@ -88,7 +89,7 @@ async function linearGraphQL( method: 'POST', headers: { 'Content-Type': 'application/json', - Authorization: `Bearer ${accessToken}`, + Authorization: linearAuthorizationHeader(accessToken), }, body: JSON.stringify({ query, variables }), }, diff --git a/apps/sim/lib/api/contracts/v2/knowledge.ts b/apps/sim/lib/api/contracts/v2/knowledge.ts index 7c290fd1ca1..c0a12a40ac0 100644 --- a/apps/sim/lib/api/contracts/v2/knowledge.ts +++ b/apps/sim/lib/api/contracts/v2/knowledge.ts @@ -1652,7 +1652,9 @@ export const v2UpdateKnowledgeConnectorBodySchema = z sourceConfig: z .record(z.string(), z.unknown().describe('Connector-specific source configuration value.')) .optional() - .describe('Replacement source selection and filtering configuration.'), + .describe( + 'Replacement source selection and filtering configuration. Updating a runnable connector queues synchronization; paused connectors remain paused.' + ), syncIntervalMinutes: z .number() .int() diff --git a/apps/sim/lib/api/contracts/v2/openapi/knowledge.ts b/apps/sim/lib/api/contracts/v2/openapi/knowledge.ts index 3ac09dd84ef..978199a5eeb 100644 --- a/apps/sim/lib/api/contracts/v2/openapi/knowledge.ts +++ b/apps/sim/lib/api/contracts/v2/openapi/knowledge.ts @@ -354,7 +354,7 @@ const declaredRoutes = [ knowledgeOperation({ operationId: 'updateKnowledgeConnector', summary: 'Update Knowledge Connector', - description: `Update connector source configuration, schedule, or active state. Authentication material cannot be changed through this operation. ${WORKSPACE_API_KEY_DENIED}`, + description: `Update connector source configuration, schedule, or active state. Replacing source configuration on a runnable connector queues an immediate synchronization; paused connectors retain the change without synchronizing until resumed. Source configuration cannot be replaced while synchronization is already in progress. Authentication material cannot be changed through this operation. ${WORKSPACE_API_KEY_DENIED}`, errors: RESOURCE_CONFLICT_ERRORS, success: { description: 'The updated connector.' }, }), diff --git a/apps/sim/lib/copilot/tools/server/knowledge/knowledge-base.test.ts b/apps/sim/lib/copilot/tools/server/knowledge/knowledge-base.test.ts index 95afa3a9066..8c3757a4cb2 100644 --- a/apps/sim/lib/copilot/tools/server/knowledge/knowledge-base.test.ts +++ b/apps/sim/lib/copilot/tools/server/knowledge/knowledge-base.test.ts @@ -631,6 +631,41 @@ describe('manage_knowledge_base trusted application delegation', () => { expect(result.message).not.toContain('private-db') }) + it('passes trusted billing attribution for a source-change sync', async () => { + mockUpdateKnowledgeConnector.mockResolvedValueOnce({ + connector: { + id: 'connector-1', + knowledgeBaseId: KNOWLEDGE_BASE.id, + connectorType: 'notion', + sourceConfig: { pageIds: ['page-2'] }, + status: 'active', + }, + }) + + const result = await knowledgeBaseServerTool.execute( + { + operation: 'update_connector', + args: { connectorId: 'connector-1', sourceConfig: { pageIds: ['page-2'] } }, + }, + BILLED_CONTEXT + ) + + expect(result).toMatchObject({ + success: true, + message: 'Connector updated successfully. Synchronization was queued for the source change.', + }) + const call = mockUpdateKnowledgeConnector.mock.calls[0]?.[0] as { + input?: { resolveBillingAttribution?: (workspaceId: string) => Promise } + } + expectDelegatedPrincipal(call) + if (!call.input?.resolveBillingAttribution) { + throw new Error('Copilot connector update did not provide billing attribution') + } + await expect(call.input.resolveBillingAttribution('workspace-paid')).resolves.toEqual( + BILLED_CONTEXT.billingAttribution + ) + }) + it.each([ [ 'update_document', diff --git a/apps/sim/lib/copilot/tools/server/knowledge/knowledge-base.ts b/apps/sim/lib/copilot/tools/server/knowledge/knowledge-base.ts index 20d88e2840f..c65ad755b66 100644 --- a/apps/sim/lib/copilot/tools/server/knowledge/knowledge-base.ts +++ b/apps/sim/lib/copilot/tools/server/knowledge/knowledge-base.ts @@ -1013,16 +1013,27 @@ export const knowledgeBaseServerTool: BaseServerTool + requireKnowledgeBillingAttribution(context, canonicalWorkspaceId), + source: 'agent', + } + ) return { success: true, - message: 'Connector updated successfully', + message: + updates.sourceConfig === undefined + ? 'Connector updated successfully' + : connector.status === 'paused' || connector.status === 'disabled' + ? 'Connector updated successfully. The source change will synchronize when the connector is resumed.' + : 'Connector updated successfully. Synchronization was queued for the source change.', data: { id: args.connectorId, ...(updates.sourceConfig !== undefined && { sourceConfig: updates.sourceConfig }), diff --git a/apps/sim/lib/knowledge/application/connectors.test.ts b/apps/sim/lib/knowledge/application/connectors.test.ts index 8947d91829e..cc17ed8ba0f 100644 --- a/apps/sim/lib/knowledge/application/connectors.test.ts +++ b/apps/sim/lib/knowledge/application/connectors.test.ts @@ -117,6 +117,8 @@ const delegatedPrincipal = { resourceScope: {}, } +const BILLING = { actorUserId: 'shared-user', workspaceId: 'workspace-a' } as never + describe('knowledge connector application use cases', () => { beforeEach(() => { vi.clearAllMocks() @@ -138,6 +140,7 @@ describe('knowledge connector application use cases', () => { mocks.resolveTokenIdentity.mockResolvedValue({ kind: 'oauth', userId: 'credential-owner' }) mocks.refreshToken.mockResolvedValue('access-token') mocks.validateConnectorConfig.mockResolvedValue({ valid: true }) + mocks.resolveBilling.mockResolvedValue(BILLING) }) afterAll(resetDbChainMock) @@ -281,11 +284,13 @@ describe('knowledge connector application use cases', () => { connectorId: 'connector-b', assertedWorkspaceId: 'workspace-a', updates: { sourceConfig: { space: 'ENG' } }, + resolveBillingAttribution: mocks.resolveBilling, source: 'agent', }, }) const orchestrationInput = mocks.updateConnector.mock.calls[0]?.[0] as { + resolveBillingAttribution?: () => Promise validateSourceConfig?: ( connector: { connectorType: string @@ -298,6 +303,10 @@ describe('knowledge connector application use cases', () => { if (!orchestrationInput.validateSourceConfig) { throw new Error('Application command did not provide source-config validation') } + if (!orchestrationInput.resolveBillingAttribution) { + throw new Error('Application command did not provide sync billing attribution') + } + await expect(orchestrationInput.resolveBillingAttribution()).resolves.toBe(BILLING) await expect( orchestrationInput.validateSourceConfig( { @@ -319,6 +328,7 @@ describe('knowledge connector application use cases', () => { expect.any(String) ) expect(mocks.validateConnectorConfig).toHaveBeenCalledWith('access-token', { space: 'ENG' }) + expect(mocks.resolveBilling).toHaveBeenCalledWith('workspace-a') }) it('rejects connector creation when the writer cannot use the workspace credential', async () => { diff --git a/apps/sim/lib/knowledge/application/connectors.ts b/apps/sim/lib/knowledge/application/connectors.ts index 9a591df3f5b..1510526af86 100644 --- a/apps/sim/lib/knowledge/application/connectors.ts +++ b/apps/sim/lib/knowledge/application/connectors.ts @@ -77,6 +77,7 @@ export interface UpdateKnowledgeConnectorInput extends KnowledgeConnectorApplica syncIntervalMinutes?: number status?: 'active' | 'paused' } + resolveBillingAttribution?(workspaceId: string): Promise } export interface DeleteKnowledgeConnectorInput extends KnowledgeConnectorApplicationInput { @@ -347,6 +348,13 @@ export const updateKnowledgeConnector = defineAuthorizedKnowledgeUseCase({ knowledgeBase: connectorTarget(context), connectorId: context.connectorId, updates: input.updates, + resolveBillingAttribution: () => { + const workspaceId = requireConnectorWorkspaceId(context) + return ( + input.resolveBillingAttribution?.(workspaceId) ?? + resolveKnowledgeBillingAttribution(principal, context) + ) + }, validateSourceConfig: (connector, sourceConfig) => { const workspaceId = requireConnectorWorkspaceId(context) return validateConnectorSourceConfig({ diff --git a/apps/sim/lib/knowledge/connectors/queue.test.ts b/apps/sim/lib/knowledge/connectors/queue.test.ts index 3d1ae6f89fe..74a32286291 100644 --- a/apps/sim/lib/knowledge/connectors/queue.test.ts +++ b/apps/sim/lib/knowledge/connectors/queue.test.ts @@ -12,15 +12,24 @@ import { } from '@sim/testing' import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest' -const { mockExecuteSync, mockIsTriggerAvailable, mockResolveTriggerRegion, mockTrigger } = - vi.hoisted(() => ({ - mockExecuteSync: vi.fn(), - mockIsTriggerAvailable: vi.fn(), - mockResolveTriggerRegion: vi.fn(), - mockTrigger: vi.fn(), - })) - -vi.mock('@trigger.dev/sdk', () => ({ tasks: { trigger: mockTrigger } })) +const { + mockCreateIdempotencyKey, + mockExecuteSync, + mockIsTriggerAvailable, + mockResolveTriggerRegion, + mockTrigger, +} = vi.hoisted(() => ({ + mockCreateIdempotencyKey: vi.fn(), + mockExecuteSync: vi.fn(), + mockIsTriggerAvailable: vi.fn(), + mockResolveTriggerRegion: vi.fn(), + mockTrigger: vi.fn(), +})) + +vi.mock('@trigger.dev/sdk', () => ({ + idempotencyKeys: { create: mockCreateIdempotencyKey }, + tasks: { trigger: mockTrigger }, +})) vi.mock('@/lib/core/async-jobs/region', () => ({ resolveTriggerRegion: mockResolveTriggerRegion, })) @@ -29,6 +38,7 @@ vi.mock('@/lib/knowledge/documents/service', () => ({ })) vi.mock('@/lib/knowledge/connectors/sync-engine', () => ({ executeSync: mockExecuteSync, + isConnectorRunnableStatus: (status: string) => status === 'active' || status === 'error', connectorIsLive: () => ({ type: 'connectorIsLive' }), LOCKABLE_CONNECTOR_STATUSES: ['active', 'error', 'pending'], })) @@ -59,6 +69,7 @@ const BILLING_ATTRIBUTION = { periodEnd: '2026-08-01T00:00:00.000Z', }, } +const NEXT_SYNC_AT = new Date('2026-07-15T12:00:00.000Z') describe('connector sync queue', () => { beforeEach(() => { @@ -67,8 +78,10 @@ describe('connector sync queue', () => { queueTableRows(schemaMock.knowledgeConnector, [ { knowledgeBaseId: 'knowledge-base-1', + connectorStatus: 'active', connectorArchivedAt: null, connectorDeletedAt: null, + connectorNextSyncAt: NEXT_SYNC_AT, workspaceId: 'workspace-paid', kbDeletedAt: null, }, @@ -76,6 +89,7 @@ describe('connector sync queue', () => { /** `markSyncPending` now reports whether it actually took the queue entry. */ dbChainMockFns.returning.mockResolvedValue([{ id: 'connector-1' }]) mockIsTriggerAvailable.mockReturnValue(true) + mockCreateIdempotencyKey.mockResolvedValue('idempotency-key') mockResolveTriggerRegion.mockResolvedValue('us-east-1') mockTrigger.mockResolvedValue({ id: 'run-1' }) }) @@ -96,6 +110,7 @@ describe('connector sync queue', () => { { connectorId: 'connector-1', fullSync: true, + requireRunnable: undefined, rehydrate: undefined, requestId: 'request-1', billingAttribution: BILLING_ATTRIBUTION, @@ -128,6 +143,75 @@ describe('connector sync queue', () => { ) }) + it('carries the runnable requirement into the queued payload', async () => { + await dispatchSync('connector-1', { + billingAttribution: BILLING_ATTRIBUTION, + expectedNextSyncAt: NEXT_SYNC_AT, + requireRunnable: true, + requestId: 'request-1', + }) + + expect(mockTrigger).toHaveBeenCalledWith( + 'knowledge-connector-sync', + expect.objectContaining({ connectorId: 'connector-1', requireRunnable: true }), + expect.objectContaining({ idempotencyKey: 'idempotency-key' }) + ) + expect(mockCreateIdempotencyKey).toHaveBeenCalledWith( + `knowledge-connector-sync:connector-1:${NEXT_SYNC_AT.toISOString()}`, + { scope: 'global' } + ) + }) + + it('skips an automatic dispatch after the due time changes', async () => { + await dispatchSync('connector-1', { + billingAttribution: BILLING_ATTRIBUTION, + expectedNextSyncAt: new Date('2026-07-15T11:00:00.000Z'), + requireRunnable: true, + requestId: 'request-1', + }) + + expect(mockCreateIdempotencyKey).not.toHaveBeenCalled() + expect(mockTrigger).not.toHaveBeenCalled() + expect(mockExecuteSync).not.toHaveBeenCalled() + }) + + it('rejects automatic dispatch without an expected due time', async () => { + await expect( + dispatchSync('connector-1', { + billingAttribution: BILLING_ATTRIBUTION, + requireRunnable: true, + requestId: 'request-1', + }) + ).rejects.toThrow('Automatic connector sync dispatch requires the expected next sync time') + + expect(mockTrigger).not.toHaveBeenCalled() + expect(mockExecuteSync).not.toHaveBeenCalled() + }) + + it('skips automatic dispatch when the connector was paused concurrently', async () => { + resetDbChainMock() + queueTableRows(schemaMock.knowledgeConnector, [ + { + knowledgeBaseId: 'knowledge-base-1', + connectorStatus: 'paused', + connectorArchivedAt: null, + connectorDeletedAt: null, + workspaceId: 'workspace-paid', + kbDeletedAt: null, + }, + ]) + + await dispatchSync('connector-1', { + billingAttribution: BILLING_ATTRIBUTION, + expectedNextSyncAt: NEXT_SYNC_AT, + requireRunnable: true, + requestId: 'request-1', + }) + + expect(mockTrigger).not.toHaveBeenCalled() + expect(mockExecuteSync).not.toHaveBeenCalled() + }) + it('releases the lock when it errors a connector whose knowledge base is gone', async () => { resetDbChainMock() queueTableRows(schemaMock.knowledgeConnector, [ @@ -220,6 +304,15 @@ describe('connector sync queue', () => { /** The create path is born `pending`, so queueing must still take that row. */ expect(lockable).toContain('pending') + /** An existing queue owner must not have its token replaced by a duplicate dispatch. */ + expect( + hasMockCondition( + queueWhere, + (node: MockCondition) => + node.type === 'isNull' && node.column === schemaMock.knowledgeConnector.syncLockToken + ) + ).toBe(true) + /** * A live run owns its row, and a paused or disabled connector must not be * pulled back into a queued sync by a dispatch that raced the status change. diff --git a/apps/sim/lib/knowledge/connectors/queue.ts b/apps/sim/lib/knowledge/connectors/queue.ts index a1c408f26e9..1bdfa5f5670 100644 --- a/apps/sim/lib/knowledge/connectors/queue.ts +++ b/apps/sim/lib/knowledge/connectors/queue.ts @@ -4,8 +4,8 @@ import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' import { generateId } from '@sim/utils/id' import { isRecordLike } from '@sim/utils/object' -import { tasks } from '@trigger.dev/sdk' -import { and, eq, inArray } from 'drizzle-orm' +import { idempotencyKeys, tasks } from '@trigger.dev/sdk' +import { and, eq, inArray, isNull } from 'drizzle-orm' import { assertBillingAttributionSnapshot, type BillingAttributionSnapshot, @@ -14,6 +14,7 @@ import { resolveTriggerRegion } from '@/lib/core/async-jobs/region' import { connectorIsLive, executeSync, + isConnectorRunnableStatus, LOCKABLE_CONNECTOR_STATUSES, } from '@/lib/knowledge/connectors/sync-engine' import { isTriggerAvailable } from '@/lib/knowledge/documents/service' @@ -23,6 +24,8 @@ const logger = createLogger('ConnectorSyncQueue') export interface ConnectorSyncPayload { connectorId: string fullSync?: boolean + /** Skip automatic work if the connector is paused or disabled before execution starts. */ + requireRunnable?: boolean /** * Force re-hydration + re-indexing of already-synced documents for connectors * whose rendered content can drift without a hash change (see @@ -46,7 +49,9 @@ export interface ConnectorSyncPayload { export interface DispatchSyncOptions { billingAttribution: BillingAttributionSnapshot + expectedNextSyncAt?: Date fullSync?: boolean + requireRunnable?: boolean rehydrate?: boolean requestId?: string } @@ -68,6 +73,9 @@ export function assertConnectorSyncPayload(value: unknown): ConnectorSyncPayload if (value.fullSync !== undefined && typeof value.fullSync !== 'boolean') { throw new Error('Connector sync payload fullSync must be a boolean when provided') } + if (value.requireRunnable !== undefined && typeof value.requireRunnable !== 'boolean') { + throw new Error('Connector sync payload requireRunnable must be a boolean when provided') + } if (value.rehydrate !== undefined && typeof value.rehydrate !== 'boolean') { throw new Error('Connector sync payload rehydrate must be a boolean when provided') } @@ -81,6 +89,7 @@ export function assertConnectorSyncPayload(value: unknown): ConnectorSyncPayload return { connectorId: value.connectorId, fullSync: value.fullSync as boolean | undefined, + requireRunnable: value.requireRunnable as boolean | undefined, rehydrate: value.rehydrate as boolean | undefined, requestId: value.requestId, billingAttribution: assertBillingAttributionSnapshot(value.billingAttribution), @@ -108,13 +117,12 @@ export const SYNC_DISPATCH_FAILED_ERROR = 'Sync could not be queued' * the recovery it is meant to trigger. The token is what makes the release * below provably this dispatch's own. * - * Deliberately still writes a row already `pending`. The create path is born - * `pending` in its INSERT but carries no lease and no token, so skipping it as - * a redundant write would leave every new connector ageing against `updatedAt` - * and holding a token this dispatch cannot match — defeating both guards above - * on exactly the path where a failed hand-off is most visible. The cost is one - * extra UPDATE per connector creation, which is rare; the scheduler's own - * dispatches only ever see `active`/`error` rows and are unaffected. + * Deliberately still writes an unowned row already `pending`. The create path is + * born `pending` in its INSERT but carries no lease and no token, so skipping it + * as a redundant write would leave every new connector ageing against + * `updatedAt`. A pending row that already has a token is not taken again: that + * token proves a hand-off already owns it, and replacing it would make the + * queued task unable to acquire the lock. * * Takes the entry only from a status a run may start from — the same * {@link LOCKABLE_CONNECTOR_STATUSES} the lock acquisition uses, so queueing and @@ -141,6 +149,7 @@ async function markSyncPending(connectorId: string): Promise { and( eq(knowledgeConnector.id, connectorId), inArray(knowledgeConnector.status, LOCKABLE_CONNECTOR_STATUSES), + isNull(knowledgeConnector.syncLockToken), connectorIsLive() ) ) @@ -212,11 +221,19 @@ export async function dispatchSync( if (!isNonEmptyString(connectorId)) { throw new Error('Connector sync dispatch requires a connector ID') } + if ( + options.requireRunnable && + (!(options.expectedNextSyncAt instanceof Date) || + Number.isNaN(options.expectedNextSyncAt.getTime())) + ) { + throw new Error('Automatic connector sync dispatch requires the expected next sync time') + } const requestId = options?.requestId ?? generateId() const payload = assertConnectorSyncPayload({ connectorId, fullSync: options?.fullSync, + requireRunnable: options?.requireRunnable, rehydrate: options?.rehydrate, requestId, billingAttribution: options?.billingAttribution, @@ -225,8 +242,10 @@ export async function dispatchSync( const connectorRows = await db .select({ knowledgeBaseId: knowledgeConnector.knowledgeBaseId, + connectorStatus: knowledgeConnector.status, connectorArchivedAt: knowledgeConnector.archivedAt, connectorDeletedAt: knowledgeConnector.deletedAt, + connectorNextSyncAt: knowledgeConnector.nextSyncAt, workspaceId: knowledgeBase.workspaceId, kbDeletedAt: knowledgeBase.deletedAt, }) @@ -277,6 +296,24 @@ export async function dispatchSync( }) return } + if (payload.requireRunnable && !isConnectorRunnableStatus(row.connectorStatus)) { + logger.info('Skipping automatic sync dispatch: connector is not runnable', { + connectorId, + status: row.connectorStatus, + requestId, + }) + return + } + if ( + options.expectedNextSyncAt && + row.connectorNextSyncAt?.getTime() !== options.expectedNextSyncAt.getTime() + ) { + logger.info('Skipping stale automatic sync dispatch: next sync time changed', { + connectorId, + requestId, + }) + return + } if (!row.workspaceId) { throw new Error(`Connector ${connectorId} is missing workspace billing context`) } @@ -311,10 +348,20 @@ export async function dispatchSync( * `pending` until the reaper's TTL. */ try { + const idempotencyKey = options.expectedNextSyncAt + ? await idempotencyKeys.create( + `knowledge-connector-sync:${connectorId}:${options.expectedNextSyncAt.toISOString()}`, + { scope: 'global' } + ) + : undefined await tasks.trigger( 'knowledge-connector-sync', { ...payload, dispatchToken }, - { tags, region: await resolveTriggerRegion() } + { + ...(idempotencyKey ? { idempotencyKey } : {}), + tags, + region: await resolveTriggerRegion(), + } ) } catch (error) { await releaseFailedDispatch(connectorId, dispatchToken, error) @@ -335,6 +382,7 @@ export async function dispatchSync( executeSync(connectorId, { fullSync: payload.fullSync, + requireRunnable: payload.requireRunnable, rehydrate: payload.rehydrate, billingAttribution: payload.billingAttribution, dispatchToken, diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts index f41f5027e4b..1d9cbc63ff1 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts @@ -17,6 +17,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { classifySuspectListing, evaluateListingSafety, + isConnectorRunnableStatus, isStuckDocumentSweepEligible, mergeHydratedDocument, type PreviousListingObservation, @@ -60,6 +61,16 @@ vi.mock('@/connectors/registry.server', () => ({ }, })) +describe('isConnectorRunnableStatus', () => { + it.each(['active', 'error'])('allows automatic sync from %s', (status) => { + expect(isConnectorRunnableStatus(status)).toBe(true) + }) + + it.each(['paused', 'disabled', 'syncing'])('blocks automatic sync from %s', (status) => { + expect(isConnectorRunnableStatus(status)).toBe(false) + }) +}) + describe('shouldReconcileDeletions', () => { it('runs on a clean full listing', async () => { const { shouldReconcileDeletions } = await import('@/lib/knowledge/connectors/sync-engine') diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.ts b/apps/sim/lib/knowledge/connectors/sync-engine.ts index 3aafd238c3e..f90e588ee9e 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.ts @@ -200,6 +200,12 @@ const STALE_PROCESSING_MINUTES = resolveStaleProcessingMinutes( envNumber(env.KB_CONFIG_MAX_ATTEMPTS, 3) ) const RETRY_WINDOW_DAYS = 7 +const RUNNABLE_CONNECTOR_STATUSES = ['active', 'error'] as const + +/** Whether an automatic connector sync may begin from this persisted state. */ +export function isConnectorRunnableStatus(status: string): boolean { + return RUNNABLE_CONNECTOR_STATUSES.some((runnableStatus) => runnableStatus === status) +} /** * Processing states the stuck-document sweep may reclaim from. @@ -1479,6 +1485,7 @@ export async function executeSync( options: { billingAttribution: BillingAttributionSnapshot fullSync?: boolean + requireRunnable?: boolean rehydrate?: boolean /** * The queue entry this run is allowed to consume. Absent only for tasks @@ -1513,22 +1520,27 @@ export async function executeSync( return { ...result, error: 'connector_unavailable' } } - const connector = connectorRows[0] + const connectorBeforeLock = connectorRows[0] - const connectorConfig = CONNECTOR_REGISTRY[connector.connectorType] + const connectorConfig = CONNECTOR_REGISTRY[connectorBeforeLock.connectorType] if (!connectorConfig) { - throw new Error(`Unknown connector type: ${connector.connectorType}`) + throw new Error(`Unknown connector type: ${connectorBeforeLock.connectorType}`) } const kbRows = await db .select({ userId: knowledgeBase.userId, workspaceId: knowledgeBase.workspaceId }) .from(knowledgeBase) - .where(and(eq(knowledgeBase.id, connector.knowledgeBaseId), isNull(knowledgeBase.deletedAt))) + .where( + and( + eq(knowledgeBase.id, connectorBeforeLock.knowledgeBaseId), + isNull(knowledgeBase.deletedAt) + ) + ) .limit(1) if (kbRows.length === 0) { logger.warn( - `Skipping sync: knowledge base ${connector.knowledgeBaseId} is deleted (connector ${connectorId})` + `Skipping sync: knowledge base ${connectorBeforeLock.knowledgeBaseId} is deleted (connector ${connectorId})` ) await db .update(knowledgeConnector) @@ -1561,7 +1573,7 @@ export async function executeSync( const kbOwner: KnowledgeBaseOwner = { workspaceId: kbRows[0].workspaceId, userId } if (!kbOwner.workspaceId) { throw new Error( - `Knowledge base ${connector.knowledgeBaseId} is missing workspace billing context` + `Knowledge base ${connectorBeforeLock.knowledgeBaseId} is missing workspace billing context` ) } if (billingAttribution.workspaceId !== kbOwner.workspaceId) { @@ -1569,8 +1581,6 @@ export async function executeSync( `Connector sync billing attribution does not match knowledge base workspace ${kbOwner.workspaceId}` ) } - const sourceConfig = connector.sourceConfig as Record - /** * Identifies this run for the terminal writes. Generated before the CAS and * written by it, so ownership is established atomically with the lock — and @@ -1604,7 +1614,7 @@ export async function executeSync( isNull(knowledgeConnector.deletedAt) ) ) - .returning({ id: knowledgeConnector.id }) + .returning() if (lockResult.length === 0) { /** @@ -1647,6 +1657,13 @@ export async function executeSync( return { ...result, error: 'sync_in_progress' } } + /** + * The row returned by the lock is the authoritative sync snapshot. A source update + * committed before the lock is included here; one attempted after it sees `syncing` + * and conflicts instead of letting this worker process stale configuration. + */ + const connector = lockResult[0] + const sourceConfig = connector.sourceConfig as Record const syncStartedAt = new Date() /** Seeded at lock acquisition, which wrote `updatedAt` itself. */ let lastHeartbeatAtMs = Date.now() diff --git a/apps/sim/lib/knowledge/orchestration/connectors.test.ts b/apps/sim/lib/knowledge/orchestration/connectors.test.ts index 34a88f54a84..3438779727f 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.test.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.test.ts @@ -143,6 +143,8 @@ describe('performUpdateKnowledgeConnector', () => { beforeEach(() => { vi.clearAllMocks() resetDbChainMock() + mockDispatchSync.mockResolvedValue(undefined) + resolveBillingAttribution.mockResolvedValue(BILLING) }) afterAll(resetDbChainMock) @@ -153,6 +155,7 @@ describe('performUpdateKnowledgeConnector', () => { knowledgeBase: KB, connectorId: 'conn-1', updates: {}, + resolveBillingAttribution, }) expect(outcome).toMatchObject({ success: false, errorCode: 'validation' }) @@ -168,6 +171,7 @@ describe('performUpdateKnowledgeConnector', () => { knowledgeBase: KB, connectorId: 'conn-1', updates: { syncIntervalMinutes: 5 }, + resolveBillingAttribution, }) expect(outcome).toMatchObject({ success: false, errorCode: 'forbidden' }) @@ -182,6 +186,7 @@ describe('performUpdateKnowledgeConnector', () => { knowledgeBase: KB, connectorId: 'conn-1', updates: { sourceConfig: { database: 'gone' } }, + resolveBillingAttribution, validateSourceConfig: async () => ({ message: 'Database not found', errorCode: 'validation' as const, @@ -205,6 +210,7 @@ describe('performUpdateKnowledgeConnector', () => { knowledgeBase: KB, connectorId: 'conn-1', updates: { sourceConfig: { database: 'x' } }, + resolveBillingAttribution, validateSourceConfig: async () => ({ message: 'Failed to refresh access token. Please reconnect your account.', errorCode: 'unauthorized' as const, @@ -225,6 +231,7 @@ describe('performUpdateKnowledgeConnector', () => { knowledgeBase: KB, connectorId: 'conn-1', updates: { status: 'active' }, + resolveBillingAttribution, }) expect(outcome).toMatchObject({ success: true }) @@ -290,6 +297,7 @@ describe('performUpdateKnowledgeConnector', () => { knowledgeBase: KB, connectorId: 'conn-1', updates: { status: 'paused' }, + resolveBillingAttribution, recordSemanticAudit: false, }) @@ -297,10 +305,319 @@ describe('performUpdateKnowledgeConnector', () => { expect(mockRecordAudit).not.toHaveBeenCalled() }) - it('refuses an update whose status moved after the guards ran', async () => { + it('queues synchronization after replacing an active connector source', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { + id: 'conn-1', + connectorType: 'notion', + status: 'active', + syncIntervalMinutes: 0, + }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([ + { + id: 'conn-1', + connectorType: 'notion', + status: 'active', + syncIntervalMinutes: 0, + }, + ]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { sourceConfig: { database: 'next' } }, + resolveBillingAttribution, + validateSourceConfig: async () => null, + }) + + expect(outcome).toMatchObject({ success: true }) + expect(resolveBillingAttribution).toHaveBeenCalledOnce() + expect(mockDispatchSync).toHaveBeenCalledWith('conn-1', { + billingAttribution: BILLING, + expectedNextSyncAt: expect.any(Date), + requestId: 'req-1', + requireRunnable: true, + }) + }) + + it('reports a queue failure and leaves the source sync due for retry', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { + id: 'conn-1', + connectorType: 'notion', + status: 'active', + syncIntervalMinutes: 0, + }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([ + { + id: 'conn-1', + connectorType: 'notion', + status: 'active', + syncIntervalMinutes: 0, + }, + ]) + mockDispatchSync.mockRejectedValueOnce(new Error('queue unavailable')) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { sourceConfig: { database: 'next' } }, + resolveBillingAttribution, + validateSourceConfig: async () => null, + }) + + expect(outcome).toMatchObject({ + success: false, + errorCode: 'internal', + error: 'queue unavailable', + }) + expect(dbChainMockFns.update).toHaveBeenCalledOnce() + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ + sourceConfig: { database: 'next' }, + nextSyncAt: expect.any(Date), + }) + ) + expect(mockDispatchSync).toHaveBeenCalledOnce() + }) + + it('saves a paused connector source without synchronizing it', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'paused' }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'paused' }, + ]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { sourceConfig: { database: 'next' } }, + resolveBillingAttribution, + validateSourceConfig: async () => null, + }) + + expect(outcome).toMatchObject({ success: true }) + expect(resolveBillingAttribution).not.toHaveBeenCalled() + expect(mockDispatchSync).not.toHaveBeenCalled() + }) + + it('does not synchronize a schedule-only update', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'active' }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'active' }, + ]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { syncIntervalMinutes: 60 }, + resolveBillingAttribution, + }) + + expect(outcome).toMatchObject({ success: true }) + expect(resolveBillingAttribution).not.toHaveBeenCalled() + expect(mockDispatchSync).not.toHaveBeenCalled() + }) + + it('preserves an already-due source sync when scheduled sync is disabled', async () => { + const pendingSourceSyncAt = new Date(0) dbChainMockFns.limit.mockResolvedValueOnce([ - { id: 'conn-1', connectorType: 'notion', status: 'pending' }, + { + id: 'conn-1', + connectorType: 'notion', + nextSyncAt: pendingSourceSyncAt, + status: 'active', + }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([ + { + id: 'conn-1', + connectorType: 'notion', + nextSyncAt: pendingSourceSyncAt, + status: 'active', + }, ]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { syncIntervalMinutes: 0 }, + resolveBillingAttribution, + }) + + expect(outcome).toMatchObject({ success: true }) + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ nextSyncAt: pendingSourceSyncAt, syncIntervalMinutes: 0 }) + ) + expect(mockDispatchSync).not.toHaveBeenCalled() + }) + + it('rejects an interval update that races with a source-sync due marker', async () => { + dbChainMockFns.limit + .mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', nextSyncAt: null, status: 'active' }, + ]) + .mockResolvedValueOnce([ + { + id: 'conn-1', + connectorType: 'notion', + nextSyncAt: new Date(), + status: 'active', + }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { syncIntervalMinutes: 0 }, + resolveBillingAttribution, + }) + + expect(outcome).toMatchObject({ success: false, errorCode: 'conflict' }) + expect(mockDispatchSync).not.toHaveBeenCalled() + }) + + it('rejects a source replacement while synchronization is in progress', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'syncing' }, + ]) + const validateSourceConfig = vi.fn() + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { sourceConfig: { database: 'next' } }, + resolveBillingAttribution, + validateSourceConfig, + }) + + expect(outcome).toMatchObject({ success: false, errorCode: 'conflict' }) + expect(validateSourceConfig).not.toHaveBeenCalled() + expect(resolveBillingAttribution).not.toHaveBeenCalled() + expect(dbChainMockFns.update).not.toHaveBeenCalled() + }) + + it('rejects a schedule change while synchronization is in progress', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'syncing' }, + ]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { syncIntervalMinutes: 0 }, + resolveBillingAttribution, + }) + + expect(outcome).toMatchObject({ + success: false, + error: 'Sync already in progress', + errorCode: 'conflict', + }) + expect(dbChainMockFns.update).not.toHaveBeenCalled() + }) + + it('rejects a schedule change that races with synchronization startup', async () => { + dbChainMockFns.limit + .mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', nextSyncAt: null, status: 'active' }, + ]) + .mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', nextSyncAt: null, status: 'syncing' }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { syncIntervalMinutes: 0 }, + resolveBillingAttribution, + }) + + expect(outcome).toMatchObject({ + success: false, + error: 'Sync already in progress', + errorCode: 'conflict', + }) + expect(mockDispatchSync).not.toHaveBeenCalled() + }) + + it('fails before persisting when sync billing attribution cannot be resolved', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'active' }, + ]) + const rejectsBilling = vi.fn().mockRejectedValue(new Error('billing unavailable')) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { sourceConfig: { database: 'next' } }, + resolveBillingAttribution: rejectsBilling, + validateSourceConfig: async () => null, + }) + + expect(outcome).toMatchObject({ success: false, errorCode: 'internal' }) + expect(dbChainMockFns.update).not.toHaveBeenCalled() + expect(mockDispatchSync).not.toHaveBeenCalled() + }) + + it('rejects a source replacement that races with a pause', async () => { + dbChainMockFns.limit + .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'active' }]) + .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'paused' }]) + dbChainMockFns.returning.mockResolvedValueOnce([]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { sourceConfig: { database: 'next' } }, + resolveBillingAttribution, + validateSourceConfig: async () => null, + }) + + expect(outcome).toMatchObject({ success: false, errorCode: 'conflict' }) + expect(mockDispatchSync).not.toHaveBeenCalled() + }) + + it('rejects a pause that races with synchronization startup', async () => { + dbChainMockFns.limit + .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'active' }]) + .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'syncing' }]) + dbChainMockFns.returning.mockResolvedValueOnce([]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { status: 'paused' }, + resolveBillingAttribution, + }) + + expect(outcome).toMatchObject({ success: false, errorCode: 'conflict' }) + expect(mockDispatchSync).not.toHaveBeenCalled() + }) + + it('refuses an update whose status moved after the guards ran', async () => { + dbChainMockFns.limit + .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'pending' }]) + .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'syncing' }]) /** The CAS matches nothing because a worker took the lock in between. */ dbChainMockFns.returning.mockResolvedValueOnce([]) @@ -325,7 +642,7 @@ describe('performUpdateKnowledgeConnector', () => { */ expect( hasMockCondition( - dbChainMockFns.where.mock.calls.at(-1)?.[0], + dbChainMockFns.where.mock.calls.at(-2)?.[0], (node: MockCondition) => node.type === 'eq' && node.left === schemaMock.knowledgeConnector.status && diff --git a/apps/sim/lib/knowledge/orchestration/connectors.ts b/apps/sim/lib/knowledge/orchestration/connectors.ts index 1e7c8a175db..a7223ef6253 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.ts @@ -376,6 +376,8 @@ export interface PerformUpdateKnowledgeConnectorParams extends KnowledgeOperatio syncIntervalMinutes?: number status?: 'active' | 'paused' } + /** Resolves the payer only when a source change will queue synchronization. */ + resolveBillingAttribution: () => Promise /** * Validates a replacement `sourceConfig` against the live source. Supplied by * the caller because resolving the connector's token needs the requesting @@ -418,7 +420,15 @@ export async function getKnowledgeConnector( export async function performUpdateKnowledgeConnector( params: PerformUpdateKnowledgeConnectorParams ): Promise { - const { knowledgeBase: kb, connectorId, updates, validateSourceConfig, request, source } = params + const { + knowledgeBase: kb, + connectorId, + updates, + resolveBillingAttribution, + validateSourceConfig, + request, + source, + } = params const requestId = params.requestId ?? generateRequestId() const updatedFields = Object.keys(updates).filter( @@ -492,16 +502,37 @@ export async function performUpdateKnowledgeConnector( } } - const values: Partial = { updatedAt: new Date() } + const resultingStatus = updates.status ?? existing.status + const shouldDispatchSourceSync = + updates.sourceConfig !== undefined && + resultingStatus !== 'paused' && + resultingStatus !== 'disabled' + let billingAttribution: BillingAttributionSnapshot | undefined + let dispatchSourceSync: Awaited> | undefined + if (shouldDispatchSourceSync) { + try { + billingAttribution = await resolveBillingAttribution() + dispatchSourceSync = await loadDispatchSync() + } catch (error) { + return classifyKnowledgeFailure(error, requestId, `Update connector ${connectorId}`) + } + } + + const updateTimestamp = new Date() + const values: Partial = { + updatedAt: updateTimestamp, + } if (updates.sourceConfig !== undefined) { values.sourceConfig = updates.sourceConfig } if (updates.syncIntervalMinutes !== undefined) { values.syncIntervalMinutes = updates.syncIntervalMinutes values.nextSyncAt = - updates.syncIntervalMinutes > 0 - ? new Date(Date.now() + updates.syncIntervalMinutes * 60 * 1000) - : null + existing.nextSyncAt && existing.nextSyncAt <= updateTimestamp + ? existing.nextSyncAt + : updates.syncIntervalMinutes > 0 + ? new Date(updateTimestamp.getTime() + updates.syncIntervalMinutes * 60 * 1000) + : null } if (updates.status !== undefined) { values.status = updates.status @@ -524,41 +555,42 @@ export async function performUpdateKnowledgeConnector( } } } + if (shouldDispatchSourceSync) { + values.nextSyncAt = updateTimestamp + } let updated: ConnectorRow try { + const updateConditions = [ + eq(knowledgeConnector.id, connectorId), + eq(knowledgeConnector.knowledgeBaseId, kb.id), + isNull(knowledgeConnector.archivedAt), + isNull(knowledgeConnector.deletedAt), + ] + updateConditions.push(eq(knowledgeConnector.status, existing.status)) + if (values.nextSyncAt !== undefined) { + updateConditions.push( + existing.nextSyncAt + ? eq(knowledgeConnector.nextSyncAt, existing.nextSyncAt) + : isNull(knowledgeConnector.nextSyncAt) + ) + } + const [row] = await db .update(knowledgeConnector) .set(values) - .where( - and( - eq(knowledgeConnector.id, connectorId), - eq(knowledgeConnector.knowledgeBaseId, kb.id), - /** - * Compare-and-set on the status this request was authorized against. - * - * The guards above ran on a row read moments earlier, and a worker can - * take the lock in between. Without this the write lands on a row that - * is now `syncing`: it would overwrite the run's status and — because - * leaving `pending` also clears the lock columns — wipe the token its - * heartbeat and terminal write match on, stranding a sync that had - * already started. - */ - eq(knowledgeConnector.status, existing.status), - isNull(knowledgeConnector.archivedAt), - isNull(knowledgeConnector.deletedAt) - ) - ) + .where(and(...updateConditions)) .returning() if (!row) { - /** - * Either the connector went away or its status moved under us. Both are - * conflicts rather than "not found": the caller's decision was made - * against a state that no longer holds, and re-reading to tell them apart - * would race the same way. - */ - return fail('Connector changed while the update was being applied', 'conflict') + const current = await getKnowledgeConnector(kb.id, connectorId) + if (current?.status === 'syncing') { + return fail('Sync already in progress', 'conflict') + } + if (current) { + return fail('Connector changed during the update; retry the request', 'conflict') + } + return fail('Connector not found', 'not_found') } updated = row } catch (error) { @@ -589,6 +621,23 @@ export async function performUpdateKnowledgeConnector( }) } + if (dispatchSourceSync && billingAttribution) { + try { + await dispatchSourceSync(connectorId, { + billingAttribution, + expectedNextSyncAt: updateTimestamp, + requestId, + requireRunnable: true, + }) + } catch (error) { + return classifyKnowledgeFailure( + error, + requestId, + `Dispatch source-change sync for connector ${connectorId}` + ) + } + } + return { success: true, connector: withoutSecret(updated) } } diff --git a/packages/sim-cli/src/contract/commands.test.ts b/packages/sim-cli/src/contract/commands.test.ts index da0ac01386b..42e11f84a19 100644 --- a/packages/sim-cli/src/contract/commands.test.ts +++ b/packages/sim-cli/src/contract/commands.test.ts @@ -70,6 +70,27 @@ describe('the command tree', () => { } }) + it('places connector synchronization with the other connector commands', () => { + const all = leafPaths({ includeHidden: true }) + + expect(all).toContain('knowledge connectors sync') + expect(all).not.toContain('knowledge sync create') + for (const path of [ + ['create'], + ['delete'], + ['get'], + ['list'], + ['sync'], + ['update'], + ['documents', 'list'], + ['documents', 'update'], + ]) { + expect(commandAt('knowledge', 'connectors', ...path).helpInformation()).toContain( + '' + ) + } + }) + it('spells one concept with one flag name across the contract', () => { // `predicate` was `--filter` on two row commands and `--predicate` on the // third, and the same idea was `--q` here and `--query` on knowledge search. diff --git a/packages/sim-cli/src/contract/commands.ts b/packages/sim-cli/src/contract/commands.ts index 78d1b767657..de4ee806fad 100644 --- a/packages/sim-cli/src/contract/commands.ts +++ b/packages/sim-cli/src/contract/commands.ts @@ -144,15 +144,34 @@ export const CLI_CONTRACT: CliContract = { selectAll: { boolean: true, describe: 'Apply to every document in the knowledge base' }, }, }, + createKnowledgeConnector: { + pathArgumentNames: KNOWLEDGE_BASE_PATH_ARGUMENT, + }, + listKnowledgeConnectors: { + pathArgumentNames: KNOWLEDGE_BASE_PATH_ARGUMENT, + }, + getKnowledgeConnector: { + pathArgumentNames: KNOWLEDGE_BASE_PATH_ARGUMENT, + }, listKnowledgeConnectorDocuments: { command: 'knowledge connectors documents list', + pathArgumentNames: KNOWLEDGE_BASE_PATH_ARGUMENT, + }, + updateKnowledgeConnector: { + pathArgumentNames: KNOWLEDGE_BASE_PATH_ARGUMENT, }, updateKnowledgeConnectorDocuments: { command: 'knowledge connectors documents update', + pathArgumentNames: KNOWLEDGE_BASE_PATH_ARGUMENT, flags: { documentIds: { name: 'document', list: true }, }, }, + syncKnowledgeConnector: { + command: 'knowledge connectors sync', + pathArgumentNames: KNOWLEDGE_BASE_PATH_ARGUMENT, + describe: 'Queue a knowledge connector synchronization', + }, // `DELETE /workflows/[id]/deploy` is an undeploy, not a delete. undeployWorkflow: { command: 'workflows undeploy', @@ -182,6 +201,7 @@ export const CLI_CONTRACT: CliContract = { confirm: 'This deletes the document and its embeddings.', }, deleteKnowledgeConnector: { + pathArgumentNames: KNOWLEDGE_BASE_PATH_ARGUMENT, confirm: 'This deletes the connector; --delete-documents also deletes its synchronized documents.', }, diff --git a/packages/sim-cli/src/generated/v2-api.ts b/packages/sim-cli/src/generated/v2-api.ts index b3dde8bd6bc..21b95b5735d 100644 --- a/packages/sim-cli/src/generated/v2-api.ts +++ b/packages/sim-cli/src/generated/v2-api.ts @@ -9367,7 +9367,8 @@ export const V2_OPERATIONS = { }, sourceConfig: { kind: 'object', - describe: 'Replacement source selection and filtering configuration.', + describe: + 'Replacement source selection and filtering configuration. Updating a runnable connector queues synchronization; paused connectors remain paused.', }, syncIntervalMinutes: { kind: 'integer', diff --git a/packages/sim-cli/src/runtime/build.ts b/packages/sim-cli/src/runtime/build.ts index 958ccce154e..df6cfcd98d8 100644 --- a/packages/sim-cli/src/runtime/build.ts +++ b/packages/sim-cli/src/runtime/build.ts @@ -235,8 +235,11 @@ function addLeafCommand( const group = groupFor(groups, groupName) if (rest.length > 1) { - const [subName, ...tail] = rest - nestedGroup(group, subName).addCommand(buildLeaf(operation, spec, tail.join(' '))) + let parent = group + for (const segment of rest.slice(0, -1)) { + parent = nestedGroup(parent, segment) + } + parent.addCommand(buildLeaf(operation, spec, rest[rest.length - 1])) return }