From e00a0cb02e736bba6b3b8cf012c7d0119ec1b9a5 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 11 Aug 2026 12:51:14 +0100 Subject: [PATCH 01/19] feat: add integration.sync_units control table Signed-off-by: Mouad BANI --- .../V1786442761__createSyncUnitsTable.sql | 25 +++++++++++++++++++ 1 file changed, 25 insertions(+) create mode 100644 backend/src/database/migrations/V1786442761__createSyncUnitsTable.sql diff --git a/backend/src/database/migrations/V1786442761__createSyncUnitsTable.sql b/backend/src/database/migrations/V1786442761__createSyncUnitsTable.sql new file mode 100644 index 0000000000..893657e826 --- /dev/null +++ b/backend/src/database/migrations/V1786442761__createSyncUnitsTable.sql @@ -0,0 +1,25 @@ +CREATE TABLE IF NOT EXISTS integration.sync_units ( + id UUID PRIMARY KEY DEFAULT uuid_generate_v4(), + "integrationId" UUID NOT NULL REFERENCES public.integrations(id), + platform TEXT NOT NULL, + "channelId" TEXT NOT NULL, + "channelName" TEXT NOT NULL, + "syncName" TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'active' + CHECK (status IN ('active','paused','dead_letter','decommissioned')), + "nextRunAt" TIMESTAMPTZ NOT NULL DEFAULT now(), + "lockedAt" TIMESTAMPTZ, + "lastRunAt" TIMESTAMPTZ, + "lastSuccessAt" TIMESTAMPTZ, + "consecutiveFailures" INT NOT NULL DEFAULT 0, + "lastErrorClass" TEXT, + watermark JSONB, + "emittedCount" INT, + "createdAt" TIMESTAMPTZ NOT NULL DEFAULT now(), + "updatedAt" TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE ("integrationId", "channelId", "syncName") +); + +CREATE INDEX IF NOT EXISTS "ix_sync_units_due" + ON integration.sync_units ("nextRunAt") + WHERE status = 'active'; From f5dafd74c31488d012e177abf28bbf3cb452019f Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 11 Aug 2026 12:52:04 +0100 Subject: [PATCH 02/19] feat: add sync units data access layer Signed-off-by: Mouad BANI --- .../src/integrationBuilder/syncUnits.ts | 115 ++++++++++++++++++ .../src/integrationBuilder/types.ts | 29 +++++ 2 files changed, 144 insertions(+) create mode 100644 services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts create mode 100644 services/libs/data-access-layer/src/integrationBuilder/types.ts diff --git a/services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts b/services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts new file mode 100644 index 0000000000..9d399e8c27 --- /dev/null +++ b/services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts @@ -0,0 +1,115 @@ +import type { QueryExecutor } from '../queryExecutor' + +import type { ISyncRunSuccess, ISyncUnit, SyncUnitUpsert } from './types' + +const MIN_INITIAL_DELAY_SECONDS = 10 +const MAX_INITIAL_DELAY_SECONDS = 900 +const CLAIM_LEASE_MINUTES = 5 + +export async function upsertSyncUnits(qx: QueryExecutor, units: SyncUnitUpsert[]): Promise { + if (units.length === 0) { + return 0 + } + + return qx.result( + `INSERT INTO integration.sync_units + ("integrationId", platform, "channelId", "channelName", "syncName", "nextRunAt") + SELECT u.*, now() + ($(minDelaySeconds) + random() * $(delaySpanSeconds)) * interval '1 second' + FROM unnest( + $(integrationIds)::uuid[], + $(platforms)::text[], + $(channelIds)::text[], + $(channelNames)::text[], + $(syncNames)::text[] + ) u + ON CONFLICT ("integrationId", "channelId", "syncName") + DO UPDATE SET "channelName" = EXCLUDED."channelName", "updatedAt" = now()`, + { + integrationIds: units.map((u) => u.integrationId), + platforms: units.map((u) => u.platform), + channelIds: units.map((u) => u.channelId), + channelNames: units.map((u) => u.channelName), + syncNames: units.map((u) => u.syncName), + minDelaySeconds: MIN_INITIAL_DELAY_SECONDS, + delaySpanSeconds: MAX_INITIAL_DELAY_SECONDS - MIN_INITIAL_DELAY_SECONDS, + }, + ) +} + +export async function claimDueUnits(qx: QueryExecutor, limit: number): Promise { + return qx.select( + `UPDATE integration.sync_units su + SET "lockedAt" = now(), "updatedAt" = now() + WHERE su.id IN ( + SELECT id + FROM integration.sync_units + WHERE status = 'active' + AND "nextRunAt" <= now() + AND ("lockedAt" IS NULL OR "lockedAt" < now() - $(leaseMinutes) * interval '1 minute') + ORDER BY "nextRunAt" + LIMIT $(limit) + FOR UPDATE SKIP LOCKED + ) + RETURNING su.*`, + { limit, leaseMinutes: CLAIM_LEASE_MINUTES }, + ) +} + +export async function rescheduleUnit( + qx: QueryExecutor, + id: string, + nextRunAt: Date, +): Promise { + await qx.result( + `UPDATE integration.sync_units + SET "nextRunAt" = $(nextRunAt), "lockedAt" = NULL, "updatedAt" = now() + WHERE id = $(id)`, + { id, nextRunAt }, + ) +} + +export async function recordRunSuccess( + qx: QueryExecutor, + id: string, + data: ISyncRunSuccess, +): Promise { + await qx.result( + `UPDATE integration.sync_units + SET watermark = $(watermark)::jsonb, + "emittedCount" = $(emittedCount), + "lastRunAt" = now(), + "lastSuccessAt" = now(), + "consecutiveFailures" = 0, + "updatedAt" = now() + WHERE id = $(id)`, + { id, watermark: JSON.stringify(data.watermark), emittedCount: data.emittedCount }, + ) +} + +export async function recordRunFailure( + qx: QueryExecutor, + id: string, + errorClass: string, + deadLetterAfter: number, +): Promise { + await qx.result( + `UPDATE integration.sync_units + SET "consecutiveFailures" = "consecutiveFailures" + 1, + "lastErrorClass" = $(errorClass), + "lastRunAt" = now(), + status = CASE WHEN "consecutiveFailures" + 1 >= $(deadLetterAfter) + THEN 'dead_letter' ELSE status END, + "updatedAt" = now() + WHERE id = $(id)`, + { id, errorClass, deadLetterAfter }, + ) +} + +export async function getUnitById(qx: QueryExecutor, id: string): Promise { + return qx.selectOneOrNone( + `SELECT * + FROM integration.sync_units + WHERE id = $(id)`, + { id }, + ) +} diff --git a/services/libs/data-access-layer/src/integrationBuilder/types.ts b/services/libs/data-access-layer/src/integrationBuilder/types.ts new file mode 100644 index 0000000000..925e2dba97 --- /dev/null +++ b/services/libs/data-access-layer/src/integrationBuilder/types.ts @@ -0,0 +1,29 @@ +export type SyncUnitStatus = 'active' | 'paused' | 'dead_letter' | 'decommissioned' + +export interface ISyncUnit { + id: string + integrationId: string + platform: string + channelId: string + channelName: string + syncName: string + status: SyncUnitStatus + nextRunAt: string + lockedAt: string | null + lastRunAt: string | null + lastSuccessAt: string | null + consecutiveFailures: number + lastErrorClass: string | null + watermark: Record | null + emittedCount: number | null +} + +export type SyncUnitUpsert = Pick< + ISyncUnit, + 'integrationId' | 'platform' | 'channelId' | 'channelName' | 'syncName' +> + +export interface ISyncRunSuccess { + watermark: Record + emittedCount: number +} From f41df42c29a553a7fc6e0afcbc3c2e8e669bf6bd Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 11 Aug 2026 13:00:55 +0100 Subject: [PATCH 03/19] feat: add integration-builder lib with connector registry Signed-off-by: Mouad BANI --- pnpm-lock.yaml | 47 +++++++++++++++---- .../libs/integration-builder/package.json | 21 +++++++++ .../libs/integration-builder/src/index.ts | 2 + .../libs/integration-builder/src/registry.ts | 23 +++++++++ .../libs/integration-builder/src/types.ts | 32 +++++++++++++ .../libs/integration-builder/tsconfig.json | 4 ++ 6 files changed, 119 insertions(+), 10 deletions(-) create mode 100644 services/libs/integration-builder/package.json create mode 100644 services/libs/integration-builder/src/index.ts create mode 100644 services/libs/integration-builder/src/registry.ts create mode 100644 services/libs/integration-builder/src/types.ts create mode 100644 services/libs/integration-builder/tsconfig.json diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 44e00af382..0329af257d 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2328,6 +2328,28 @@ importers: specifier: ^5.6.3 version: 5.6.3 + services/libs/integration-builder: + dependencies: + '@crowd/common': + specifier: workspace:* + version: link:../common + '@crowd/data-access-layer': + specifier: workspace:* + version: link:../data-access-layer + '@crowd/logging': + specifier: workspace:* + version: link:../logging + zod: + specifier: ^3.22.0 + version: 3.25.76 + devDependencies: + '@types/node': + specifier: ^20.8.2 + version: 20.12.7 + typescript: + specifier: ^5.6.3 + version: 5.6.3 + services/libs/integrations: dependencies: '@crowd/common': @@ -10810,6 +10832,9 @@ packages: peerDependencies: zod: ^3.25.28 || ^4 + zod@3.25.76: + resolution: {integrity: sha512-gzUt/qt81nXsFGKIFcC3YnfEAx5NkunCfnDlvuBSSFS02bcXu4Lmea0AFIUwbLWxWPx3d9p8S5QoaujKcNQxcQ==} + zod@4.3.6: resolution: {integrity: sha512-rftlrkhHZOcjDwkGlnUtZZkvaPHCsDATp4pGpuOOMDaTdDDXF91wuVDJoWoPsKX/3YPQ5fHuF3STjcYyKr+Qhg==} @@ -10969,8 +10994,8 @@ snapshots: dependencies: '@aws-crypto/sha256-browser': 3.0.0 '@aws-crypto/sha256-js': 3.0.0 - '@aws-sdk/client-sso-oidc': 3.572.0(@aws-sdk/client-sts@3.572.0) - '@aws-sdk/client-sts': 3.572.0 + '@aws-sdk/client-sso-oidc': 3.572.0 + '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) '@aws-sdk/core': 3.572.0 '@aws-sdk/credential-provider-node': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0) '@aws-sdk/middleware-host-header': 3.567.0 @@ -11164,11 +11189,11 @@ snapshots: transitivePeerDependencies: - aws-crt - '@aws-sdk/client-sso-oidc@3.572.0(@aws-sdk/client-sts@3.572.0)': + '@aws-sdk/client-sso-oidc@3.572.0': dependencies: '@aws-crypto/sha256-browser': 3.0.0 '@aws-crypto/sha256-js': 3.0.0 - '@aws-sdk/client-sts': 3.572.0 + '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) '@aws-sdk/core': 3.572.0 '@aws-sdk/credential-provider-node': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0) '@aws-sdk/middleware-host-header': 3.567.0 @@ -11207,7 +11232,6 @@ snapshots: '@smithy/util-utf8': 2.3.0 tslib: 2.6.2 transitivePeerDependencies: - - '@aws-sdk/client-sts' - aws-crt '@aws-sdk/client-sso@3.556.0': @@ -11383,11 +11407,11 @@ snapshots: transitivePeerDependencies: - aws-crt - '@aws-sdk/client-sts@3.572.0': + '@aws-sdk/client-sts@3.572.0(@aws-sdk/client-sso-oidc@3.572.0)': dependencies: '@aws-crypto/sha256-browser': 3.0.0 '@aws-crypto/sha256-js': 3.0.0 - '@aws-sdk/client-sso-oidc': 3.572.0(@aws-sdk/client-sts@3.572.0) + '@aws-sdk/client-sso-oidc': 3.572.0 '@aws-sdk/core': 3.572.0 '@aws-sdk/credential-provider-node': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0) '@aws-sdk/middleware-host-header': 3.567.0 @@ -11426,6 +11450,7 @@ snapshots: '@smithy/util-utf8': 2.3.0 tslib: 2.6.2 transitivePeerDependencies: + - '@aws-sdk/client-sso-oidc' - aws-crt '@aws-sdk/client-sts@3.985.0': @@ -11591,7 +11616,7 @@ snapshots: '@aws-sdk/credential-provider-ini@3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0)': dependencies: - '@aws-sdk/client-sts': 3.572.0 + '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) '@aws-sdk/credential-provider-env': 3.568.0 '@aws-sdk/credential-provider-process': 3.572.0 '@aws-sdk/credential-provider-sso': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) @@ -11768,7 +11793,7 @@ snapshots: '@aws-sdk/credential-provider-web-identity@3.568.0(@aws-sdk/client-sts@3.572.0)': dependencies: - '@aws-sdk/client-sts': 3.572.0 + '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) '@aws-sdk/types': 3.567.0 '@smithy/property-provider': 2.2.0 '@smithy/types': 2.12.0 @@ -12080,7 +12105,7 @@ snapshots: '@aws-sdk/token-providers@3.572.0(@aws-sdk/client-sso-oidc@3.572.0)': dependencies: - '@aws-sdk/client-sso-oidc': 3.572.0(@aws-sdk/client-sts@3.572.0) + '@aws-sdk/client-sso-oidc': 3.572.0 '@aws-sdk/types': 3.567.0 '@smithy/property-provider': 2.2.0 '@smithy/shared-ini-file-loader': 2.4.0 @@ -21191,4 +21216,6 @@ snapshots: dependencies: zod: 4.3.6 + zod@3.25.76: {} + zod@4.3.6: {} diff --git a/services/libs/integration-builder/package.json b/services/libs/integration-builder/package.json new file mode 100644 index 0000000000..eb1298b5b9 --- /dev/null +++ b/services/libs/integration-builder/package.json @@ -0,0 +1,21 @@ +{ + "name": "@crowd/integration-builder", + "private": true, + "main": "src/index.ts", + "scripts": { + "lint": "npx eslint --ext .ts src --max-warnings=0", + "format": "npx prettier --write \"src/**/*.ts\"", + "format-check": "npx prettier --check .", + "tsc-check": "tsc --noEmit" + }, + "devDependencies": { + "@types/node": "^20.8.2", + "typescript": "^5.6.3" + }, + "dependencies": { + "@crowd/common": "workspace:*", + "@crowd/data-access-layer": "workspace:*", + "@crowd/logging": "workspace:*", + "zod": "^3.22.0" + } +} diff --git a/services/libs/integration-builder/src/index.ts b/services/libs/integration-builder/src/index.ts new file mode 100644 index 0000000000..636ed8d91f --- /dev/null +++ b/services/libs/integration-builder/src/index.ts @@ -0,0 +1,2 @@ +export * from './registry' +export * from './types' diff --git a/services/libs/integration-builder/src/registry.ts b/services/libs/integration-builder/src/registry.ts new file mode 100644 index 0000000000..081751934b --- /dev/null +++ b/services/libs/integration-builder/src/registry.ts @@ -0,0 +1,23 @@ +import type { Manifest, SyncDefinition } from './types' + +const manifests = new Map() + +export function registerConnector(manifest: Manifest): void { + manifests.set(manifest.platform, manifest) +} + +export function getManifest(platform: string): Manifest { + const manifest = manifests.get(platform) + if (!manifest) { + throw new Error(`unknown platform ${platform}`) + } + return manifest +} + +export function getSync(platform: string, syncName: string): SyncDefinition { + const sync = getManifest(platform).syncs.find((s) => s.name === syncName) + if (!sync) { + throw new Error(`unknown sync ${platform}/${syncName}`) + } + return sync +} diff --git a/services/libs/integration-builder/src/types.ts b/services/libs/integration-builder/src/types.ts new file mode 100644 index 0000000000..88e6fef35a --- /dev/null +++ b/services/libs/integration-builder/src/types.ts @@ -0,0 +1,32 @@ +import type { Logger } from '@crowd/logging' + +export interface Channel { + channelId: string + channelName: string +} + +export interface Credential { + platform: string + kind: 'github-app' | 'token' + data: Record +} + +export interface SyncContext { + channel: Channel + watermark: Record | null + emit: (records: unknown[]) => Promise + commitWatermark: (watermark: Record) => Promise + log: Logger +} + +export interface SyncDefinition { + name: string + cadenceMinutes: number + run: (ctx: SyncContext) => Promise +} + +export interface Manifest { + platform: string + syncs: SyncDefinition[] + discover: (credential: Credential) => Promise +} diff --git a/services/libs/integration-builder/tsconfig.json b/services/libs/integration-builder/tsconfig.json new file mode 100644 index 0000000000..bf7f183850 --- /dev/null +++ b/services/libs/integration-builder/tsconfig.json @@ -0,0 +1,4 @@ +{ + "extends": "../../base.tsconfig.json", + "include": ["src/**/*"] +} From 5062058dcc39311dddcfb7eb28e461016324db55 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 11 Aug 2026 13:05:46 +0100 Subject: [PATCH 04/19] fix: skip sync units of soft-deleted integrations when claiming Signed-off-by: Mouad BANI --- .../src/integrationBuilder/syncUnits.ts | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts b/services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts index 9d399e8c27..3554b77935 100644 --- a/services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts +++ b/services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts @@ -41,12 +41,17 @@ export async function claimDueUnits(qx: QueryExecutor, limit: number): Promise Date: Tue, 11 Aug 2026 13:23:29 +0100 Subject: [PATCH 05/19] feat: add getCredential facade Signed-off-by: Mouad BANI --- .../integration-builder/src/credentials.ts | 51 +++++++++++++++++++ .../libs/integration-builder/src/index.ts | 1 + .../libs/integration-builder/src/types.ts | 11 +++- 3 files changed, 61 insertions(+), 2 deletions(-) create mode 100644 services/libs/integration-builder/src/credentials.ts diff --git a/services/libs/integration-builder/src/credentials.ts b/services/libs/integration-builder/src/credentials.ts new file mode 100644 index 0000000000..552e214991 --- /dev/null +++ b/services/libs/integration-builder/src/credentials.ts @@ -0,0 +1,51 @@ +import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' + +import type { Credential } from './types' + +export async function getCredential( + qx: QueryExecutor, + integrationId: string, +): Promise { + const integration: { platform: string } | null = await qx.selectOneOrNone( + `SELECT platform + FROM integrations + WHERE id = $(integrationId) AND "deletedAt" IS NULL`, + { integrationId }, + ) + + if (!integration) { + throw new Error(`integration ${integrationId} not found`) + } + + // POC scope: GitHub only; each migrated connector adds its platform case here + switch (integration.platform) { + case 'github': + case 'github-nango': + return githubAppCredential() + default: + throw new Error(`unsupported platform ${integration.platform}`) + } +} + +// POC only: secrets come from env; secret-manager adoption (OCI Vault) swaps this +// body without touching callers — getCredential stays the single entry point +function githubAppCredential(): Credential { + const appId = process.env.CROWD_GITHUB_APP_ID + const rawPrivateKey = process.env.CROWD_GITHUB_PRIVATE_KEY + + if (!appId || !rawPrivateKey) { + throw new Error( + 'missing CROWD_GITHUB_APP_ID or CROWD_GITHUB_PRIVATE_KEY environment variables', + ) + } + + const privateKey = rawPrivateKey.startsWith('-----') + ? rawPrivateKey + : Buffer.from(rawPrivateKey, 'base64').toString('ascii') + + return { + platform: 'github', + kind: 'github-app', + data: { appId, privateKey }, + } +} diff --git a/services/libs/integration-builder/src/index.ts b/services/libs/integration-builder/src/index.ts index 636ed8d91f..6e932df937 100644 --- a/services/libs/integration-builder/src/index.ts +++ b/services/libs/integration-builder/src/index.ts @@ -1,2 +1,3 @@ +export * from './credentials' export * from './registry' export * from './types' diff --git a/services/libs/integration-builder/src/types.ts b/services/libs/integration-builder/src/types.ts index 88e6fef35a..9e60f1c91c 100644 --- a/services/libs/integration-builder/src/types.ts +++ b/services/libs/integration-builder/src/types.ts @@ -5,10 +5,17 @@ export interface Channel { channelName: string } +export interface GithubAppCredentialData { + appId: string + privateKey: string +} + +// POC only: single variant; becomes a discriminated union (token, oauth2, ...) +// as more connectors land export interface Credential { platform: string - kind: 'github-app' | 'token' - data: Record + kind: 'github-app' + data: GithubAppCredentialData } export interface SyncContext { From 4daecd3dcaae460f56050b16842cd688b039cdd8 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 11 Aug 2026 13:28:55 +0100 Subject: [PATCH 06/19] fix: export integration builder DAL from package root Signed-off-by: Mouad BANI --- services/libs/data-access-layer/src/index.ts | 1 + services/libs/data-access-layer/src/integrationBuilder/index.ts | 2 ++ 2 files changed, 3 insertions(+) create mode 100644 services/libs/data-access-layer/src/integrationBuilder/index.ts diff --git a/services/libs/data-access-layer/src/index.ts b/services/libs/data-access-layer/src/index.ts index 9079622755..c96467b899 100644 --- a/services/libs/data-access-layer/src/index.ts +++ b/services/libs/data-access-layer/src/index.ts @@ -13,6 +13,7 @@ export * from './repositories' export * from './security_insights' export * from './segments' export * from './systemSettings' +export * from './integrationBuilder' export * from './integrations' export * from './auditLogs' export * from './maintainers' diff --git a/services/libs/data-access-layer/src/integrationBuilder/index.ts b/services/libs/data-access-layer/src/integrationBuilder/index.ts new file mode 100644 index 0000000000..ee1244f4c2 --- /dev/null +++ b/services/libs/data-access-layer/src/integrationBuilder/index.ts @@ -0,0 +1,2 @@ +export * from './syncUnits' +export * from './types' From 31d697f9e26bcb8ff48b4e0290286fe216ea2ed4 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 11 Aug 2026 13:41:09 +0100 Subject: [PATCH 07/19] style: fix prettier formatting in credentials facade Signed-off-by: Mouad BANI --- services/libs/integration-builder/src/credentials.ts | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/services/libs/integration-builder/src/credentials.ts b/services/libs/integration-builder/src/credentials.ts index 552e214991..d5daa68405 100644 --- a/services/libs/integration-builder/src/credentials.ts +++ b/services/libs/integration-builder/src/credentials.ts @@ -2,10 +2,7 @@ import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' import type { Credential } from './types' -export async function getCredential( - qx: QueryExecutor, - integrationId: string, -): Promise { +export async function getCredential(qx: QueryExecutor, integrationId: string): Promise { const integration: { platform: string } | null = await qx.selectOneOrNone( `SELECT platform FROM integrations @@ -34,9 +31,7 @@ function githubAppCredential(): Credential { const rawPrivateKey = process.env.CROWD_GITHUB_PRIVATE_KEY if (!appId || !rawPrivateKey) { - throw new Error( - 'missing CROWD_GITHUB_APP_ID or CROWD_GITHUB_PRIVATE_KEY environment variables', - ) + throw new Error('missing CROWD_GITHUB_APP_ID or CROWD_GITHUB_PRIVATE_KEY environment variables') } const privateKey = rawPrivateKey.startsWith('-----') From 94901f8231f30266442c3c5113db7a69716b6f60 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 11 Aug 2026 17:35:46 +0100 Subject: [PATCH 08/19] refactor: rename integration-builder to connectors Signed-off-by: Mouad BANI --- services/libs/{integration-builder => connectors}/package.json | 2 +- .../libs/{integration-builder => connectors}/src/credentials.ts | 0 services/libs/{integration-builder => connectors}/src/index.ts | 0 .../libs/{integration-builder => connectors}/src/registry.ts | 0 services/libs/{integration-builder => connectors}/src/types.ts | 0 services/libs/{integration-builder => connectors}/tsconfig.json | 0 .../src/{integrationBuilder => connectors}/index.ts | 0 .../src/{integrationBuilder => connectors}/syncUnits.ts | 0 .../src/{integrationBuilder => connectors}/types.ts | 0 services/libs/data-access-layer/src/index.ts | 2 +- 10 files changed, 2 insertions(+), 2 deletions(-) rename services/libs/{integration-builder => connectors}/package.json (92%) rename services/libs/{integration-builder => connectors}/src/credentials.ts (100%) rename services/libs/{integration-builder => connectors}/src/index.ts (100%) rename services/libs/{integration-builder => connectors}/src/registry.ts (100%) rename services/libs/{integration-builder => connectors}/src/types.ts (100%) rename services/libs/{integration-builder => connectors}/tsconfig.json (100%) rename services/libs/data-access-layer/src/{integrationBuilder => connectors}/index.ts (100%) rename services/libs/data-access-layer/src/{integrationBuilder => connectors}/syncUnits.ts (100%) rename services/libs/data-access-layer/src/{integrationBuilder => connectors}/types.ts (100%) diff --git a/services/libs/integration-builder/package.json b/services/libs/connectors/package.json similarity index 92% rename from services/libs/integration-builder/package.json rename to services/libs/connectors/package.json index eb1298b5b9..2e5d961cda 100644 --- a/services/libs/integration-builder/package.json +++ b/services/libs/connectors/package.json @@ -1,5 +1,5 @@ { - "name": "@crowd/integration-builder", + "name": "@crowd/connectors", "private": true, "main": "src/index.ts", "scripts": { diff --git a/services/libs/integration-builder/src/credentials.ts b/services/libs/connectors/src/credentials.ts similarity index 100% rename from services/libs/integration-builder/src/credentials.ts rename to services/libs/connectors/src/credentials.ts diff --git a/services/libs/integration-builder/src/index.ts b/services/libs/connectors/src/index.ts similarity index 100% rename from services/libs/integration-builder/src/index.ts rename to services/libs/connectors/src/index.ts diff --git a/services/libs/integration-builder/src/registry.ts b/services/libs/connectors/src/registry.ts similarity index 100% rename from services/libs/integration-builder/src/registry.ts rename to services/libs/connectors/src/registry.ts diff --git a/services/libs/integration-builder/src/types.ts b/services/libs/connectors/src/types.ts similarity index 100% rename from services/libs/integration-builder/src/types.ts rename to services/libs/connectors/src/types.ts diff --git a/services/libs/integration-builder/tsconfig.json b/services/libs/connectors/tsconfig.json similarity index 100% rename from services/libs/integration-builder/tsconfig.json rename to services/libs/connectors/tsconfig.json diff --git a/services/libs/data-access-layer/src/integrationBuilder/index.ts b/services/libs/data-access-layer/src/connectors/index.ts similarity index 100% rename from services/libs/data-access-layer/src/integrationBuilder/index.ts rename to services/libs/data-access-layer/src/connectors/index.ts diff --git a/services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts b/services/libs/data-access-layer/src/connectors/syncUnits.ts similarity index 100% rename from services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts rename to services/libs/data-access-layer/src/connectors/syncUnits.ts diff --git a/services/libs/data-access-layer/src/integrationBuilder/types.ts b/services/libs/data-access-layer/src/connectors/types.ts similarity index 100% rename from services/libs/data-access-layer/src/integrationBuilder/types.ts rename to services/libs/data-access-layer/src/connectors/types.ts diff --git a/services/libs/data-access-layer/src/index.ts b/services/libs/data-access-layer/src/index.ts index c96467b899..7424ddcdd4 100644 --- a/services/libs/data-access-layer/src/index.ts +++ b/services/libs/data-access-layer/src/index.ts @@ -13,7 +13,7 @@ export * from './repositories' export * from './security_insights' export * from './segments' export * from './systemSettings' -export * from './integrationBuilder' +export * from './connectors' export * from './integrations' export * from './auditLogs' export * from './maintainers' From d681c113506d2036ff3c5ba80391cefb39d90d11 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 11 Aug 2026 17:47:44 +0100 Subject: [PATCH 09/19] feat: add connectors worker with dispatcher workflow Signed-off-by: Mouad BANI --- pnpm-lock.yaml | 96 ++++++++++++++----- services/apps/connectors_worker/package.json | 35 +++++++ .../apps/connectors_worker/src/activities.ts | 3 + .../src/activities/dispatcherActivities.ts | 52 ++++++++++ services/apps/connectors_worker/src/main.ts | 36 +++++++ .../src/schedules/dispatcher.ts | 32 +++++++ services/apps/connectors_worker/src/types.ts | 1 + .../apps/connectors_worker/src/workflows.ts | 3 + .../src/workflows/dispatcher.ts | 21 ++++ services/apps/connectors_worker/tsconfig.json | 4 + 10 files changed, 261 insertions(+), 22 deletions(-) create mode 100644 services/apps/connectors_worker/package.json create mode 100644 services/apps/connectors_worker/src/activities.ts create mode 100644 services/apps/connectors_worker/src/activities/dispatcherActivities.ts create mode 100644 services/apps/connectors_worker/src/main.ts create mode 100644 services/apps/connectors_worker/src/schedules/dispatcher.ts create mode 100644 services/apps/connectors_worker/src/types.ts create mode 100644 services/apps/connectors_worker/src/workflows.ts create mode 100644 services/apps/connectors_worker/src/workflows/dispatcher.ts create mode 100644 services/apps/connectors_worker/tsconfig.json diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 0329af257d..63c18dadf8 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -608,6 +608,58 @@ importers: specifier: ^3.0.1 version: 3.1.0 + services/apps/connectors_worker: + dependencies: + '@crowd/archetype-standard': + specifier: workspace:* + version: link:../../archetypes/standard + '@crowd/archetype-worker': + specifier: workspace:* + version: link:../../archetypes/worker + '@crowd/common': + specifier: workspace:* + version: link:../../libs/common + '@crowd/connectors': + specifier: workspace:* + version: link:../../libs/connectors + '@crowd/data-access-layer': + specifier: workspace:* + version: link:../../libs/data-access-layer + '@crowd/logging': + specifier: workspace:* + version: link:../../libs/logging + '@crowd/redis': + specifier: workspace:* + version: link:../../libs/redis + '@crowd/temporal': + specifier: workspace:* + version: link:../../libs/temporal + '@crowd/types': + specifier: workspace:* + version: link:../../libs/types + '@temporalio/activity': + specifier: ~1.17.2 + version: 1.17.2 + '@temporalio/client': + specifier: ~1.17.2 + version: 1.17.2 + '@temporalio/workflow': + specifier: ~1.17.2 + version: 1.17.2 + tsx: + specifier: ^4.7.1 + version: 4.7.3 + typescript: + specifier: ^5.6.3 + version: 5.6.3 + devDependencies: + '@types/node': + specifier: ^20.8.2 + version: 20.12.7 + nodemon: + specifier: ^3.0.1 + version: 3.1.0 + services/apps/cron_service: dependencies: '@aws-sdk/client-s3': @@ -2227,6 +2279,28 @@ importers: specifier: ^5.6.3 version: 5.6.3 + services/libs/connectors: + dependencies: + '@crowd/common': + specifier: workspace:* + version: link:../common + '@crowd/data-access-layer': + specifier: workspace:* + version: link:../data-access-layer + '@crowd/logging': + specifier: workspace:* + version: link:../logging + zod: + specifier: ^3.22.0 + version: 3.25.76 + devDependencies: + '@types/node': + specifier: ^20.8.2 + version: 20.12.7 + typescript: + specifier: ^5.6.3 + version: 5.6.3 + services/libs/data-access-layer: dependencies: '@crowd/common': @@ -2328,28 +2402,6 @@ importers: specifier: ^5.6.3 version: 5.6.3 - services/libs/integration-builder: - dependencies: - '@crowd/common': - specifier: workspace:* - version: link:../common - '@crowd/data-access-layer': - specifier: workspace:* - version: link:../data-access-layer - '@crowd/logging': - specifier: workspace:* - version: link:../logging - zod: - specifier: ^3.22.0 - version: 3.25.76 - devDependencies: - '@types/node': - specifier: ^20.8.2 - version: 20.12.7 - typescript: - specifier: ^5.6.3 - version: 5.6.3 - services/libs/integrations: dependencies: '@crowd/common': diff --git a/services/apps/connectors_worker/package.json b/services/apps/connectors_worker/package.json new file mode 100644 index 0000000000..5cb359796b --- /dev/null +++ b/services/apps/connectors_worker/package.json @@ -0,0 +1,35 @@ +{ + "name": "@crowd/connectors-worker", + "private": true, + "scripts": { + "start": "CROWD_TEMPORAL_TASKQUEUE=connectors SERVICE=connectors-worker tsx src/main.ts", + "start:debug:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && CROWD_TEMPORAL_TASKQUEUE=connectors SERVICE=connectors-worker LOG_LEVEL=trace tsx --inspect=0.0.0.0:9243 src/main.ts", + "start:debug": "CROWD_TEMPORAL_TASKQUEUE=connectors SERVICE=connectors-worker LOG_LEVEL=trace tsx --inspect=0.0.0.0:9243 src/main.ts", + "dev:local": "nodemon --watch src --watch ../../libs --ext ts --exec pnpm run start:debug:local", + "dev": "nodemon --watch src --watch ../../libs --ext ts --exec pnpm run start:debug", + "lint": "npx eslint --ext .ts src --max-warnings=0", + "format": "npx prettier --write \"src/**/*.ts\"", + "format-check": "npx prettier --check .", + "tsc-check": "tsc --noEmit" + }, + "dependencies": { + "@crowd/archetype-standard": "workspace:*", + "@crowd/archetype-worker": "workspace:*", + "@crowd/common": "workspace:*", + "@crowd/data-access-layer": "workspace:*", + "@crowd/connectors": "workspace:*", + "@crowd/logging": "workspace:*", + "@crowd/redis": "workspace:*", + "@crowd/temporal": "workspace:*", + "@crowd/types": "workspace:*", + "@temporalio/activity": "~1.17.2", + "@temporalio/client": "~1.17.2", + "@temporalio/workflow": "~1.17.2", + "tsx": "^4.7.1", + "typescript": "^5.6.3" + }, + "devDependencies": { + "@types/node": "^20.8.2", + "nodemon": "^3.0.1" + } +} diff --git a/services/apps/connectors_worker/src/activities.ts b/services/apps/connectors_worker/src/activities.ts new file mode 100644 index 0000000000..c69b2985c9 --- /dev/null +++ b/services/apps/connectors_worker/src/activities.ts @@ -0,0 +1,3 @@ +import { claimDue, reschedule, startRun, touchHeartbeat } from './activities/dispatcherActivities' + +export { claimDue, reschedule, startRun, touchHeartbeat } diff --git a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts new file mode 100644 index 0000000000..c5c47a644d --- /dev/null +++ b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts @@ -0,0 +1,52 @@ +import { getSync } from '@crowd/connectors' +import { claimDueUnits, rescheduleUnit } from '@crowd/data-access-layer/src/connectors' +import type { ISyncUnit } from '@crowd/data-access-layer/src/connectors' +import { dbStoreQx } from '@crowd/data-access-layer/src/queryExecutor' +import { RedisCache } from '@crowd/redis' +import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@crowd/temporal' + +import { svc } from '../main' +import type { StartRunResult } from '../types' + +const TASK_QUEUE = 'connectors' +const HEARTBEAT_TTL_SECONDS = 300 +const CADENCE_JITTER_RATIO = 0.1 + +export async function claimDue(limit: number): Promise { + return claimDueUnits(dbStoreQx(svc.postgres.writer), limit) +} + +export async function startRun(unit: ISyncUnit): Promise { + try { + await svc.temporal.workflow.start('syncRun', { + taskQueue: TASK_QUEUE, + workflowId: `sync-run/${unit.id}`, + workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE, + workflowIdConflictPolicy: WorkflowIdConflictPolicy.FAIL, + args: [unit.id], + }) + return 'started' + } catch (err) { + if (err instanceof Error && err.name === 'WorkflowExecutionAlreadyStartedError') { + return 'alreadyRunning' + } + throw err + } +} + +export async function reschedule( + unitId: string, + platform: string, + syncName: string, +): Promise { + const { cadenceMinutes } = getSync(platform, syncName) + const jitterMinutes = cadenceMinutes * CADENCE_JITTER_RATIO * (Math.random() * 2 - 1) + const nextRunAt = new Date(Date.now() + (cadenceMinutes + jitterMinutes) * 60_000) + + await rescheduleUnit(dbStoreQx(svc.postgres.writer), unitId, nextRunAt) +} + +export async function touchHeartbeat(): Promise { + const cache = new RedisCache('connectors', svc.redis, svc.log) + await cache.set('dispatcherHeartbeat', new Date().toISOString(), HEARTBEAT_TTL_SECONDS) +} diff --git a/services/apps/connectors_worker/src/main.ts b/services/apps/connectors_worker/src/main.ts new file mode 100644 index 0000000000..7ebd785514 --- /dev/null +++ b/services/apps/connectors_worker/src/main.ts @@ -0,0 +1,36 @@ +import { Config } from '@crowd/archetype-standard' +import { Options, ServiceWorker } from '@crowd/archetype-worker' + +import { scheduleDispatcher } from './schedules/dispatcher' + +const config: Config = { + envvars: [], + producer: { + enabled: false, + }, + temporal: { + enabled: true, + }, + redis: { + enabled: true, + }, +} + +const options: Options = { + postgres: { + enabled: true, + }, + opensearch: { + enabled: false, + }, +} + +export const svc = new ServiceWorker(config, options) + +setImmediate(async () => { + await svc.init() + + await scheduleDispatcher() + + await svc.start() +}) diff --git a/services/apps/connectors_worker/src/schedules/dispatcher.ts b/services/apps/connectors_worker/src/schedules/dispatcher.ts new file mode 100644 index 0000000000..63ed04d218 --- /dev/null +++ b/services/apps/connectors_worker/src/schedules/dispatcher.ts @@ -0,0 +1,32 @@ +import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/client' + +import { svc } from '../main' +import { dispatcher } from '../workflows/dispatcher' + +export async function scheduleDispatcher(): Promise { + try { + await svc.temporal.schedule.create({ + scheduleId: 'connectors-dispatcher', + spec: { + intervals: [{ every: '30s' }], + }, + policies: { + overlap: ScheduleOverlapPolicy.SKIP, + catchupWindow: '1 minute', + }, + action: { + type: 'startWorkflow', + workflowType: dispatcher, + taskQueue: 'connectors', + args: [], + workflowExecutionTimeout: '5 minutes', + }, + }) + } catch (err) { + if (err instanceof ScheduleAlreadyRunning) { + svc.log.info('Dispatcher schedule already registered in Temporal.') + } else { + throw new Error(err) + } + } +} diff --git a/services/apps/connectors_worker/src/types.ts b/services/apps/connectors_worker/src/types.ts new file mode 100644 index 0000000000..ac915242b1 --- /dev/null +++ b/services/apps/connectors_worker/src/types.ts @@ -0,0 +1 @@ +export type StartRunResult = 'started' | 'alreadyRunning' diff --git a/services/apps/connectors_worker/src/workflows.ts b/services/apps/connectors_worker/src/workflows.ts new file mode 100644 index 0000000000..2044bd7bf3 --- /dev/null +++ b/services/apps/connectors_worker/src/workflows.ts @@ -0,0 +1,3 @@ +import { dispatcher } from './workflows/dispatcher' + +export { dispatcher } diff --git a/services/apps/connectors_worker/src/workflows/dispatcher.ts b/services/apps/connectors_worker/src/workflows/dispatcher.ts new file mode 100644 index 0000000000..44cbbae714 --- /dev/null +++ b/services/apps/connectors_worker/src/workflows/dispatcher.ts @@ -0,0 +1,21 @@ +import { proxyActivities } from '@temporalio/workflow' + +import type * as activities from '../activities/dispatcherActivities' + +const activity = proxyActivities({ + startToCloseTimeout: '1 minute', + retry: { maximumAttempts: 3, backoffCoefficient: 2 }, +}) + +const CLAIM_LIMIT = 100 + +export async function dispatcher(): Promise { + await activity.touchHeartbeat() + + const units = await activity.claimDue(CLAIM_LIMIT) + + for (const unit of units) { + await activity.startRun(unit) + await activity.reschedule(unit.id, unit.platform, unit.syncName) + } +} diff --git a/services/apps/connectors_worker/tsconfig.json b/services/apps/connectors_worker/tsconfig.json new file mode 100644 index 0000000000..bf7f183850 --- /dev/null +++ b/services/apps/connectors_worker/tsconfig.json @@ -0,0 +1,4 @@ +{ + "extends": "../../base.tsconfig.json", + "include": ["src/**/*"] +} From 8b22542601b1f9781ae99045fa7ed9829a1c5862 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 11 Aug 2026 18:20:22 +0100 Subject: [PATCH 10/19] feat: add sync-run workflow with dummy connector Signed-off-by: Mouad BANI --- .../apps/connectors_worker/src/activities.ts | 3 +- .../src/activities/syncRunActivities.ts | 70 +++++++++++++++++++ services/apps/connectors_worker/src/main.ts | 6 ++ .../apps/connectors_worker/src/workflows.ts | 3 +- .../src/workflows/dispatcher.ts | 10 ++- .../src/workflows/syncRun.ts | 13 ++++ .../connectors/src/testing/dummyConnector.ts | 18 +++++ 7 files changed, 118 insertions(+), 5 deletions(-) create mode 100644 services/apps/connectors_worker/src/activities/syncRunActivities.ts create mode 100644 services/apps/connectors_worker/src/workflows/syncRun.ts create mode 100644 services/libs/connectors/src/testing/dummyConnector.ts diff --git a/services/apps/connectors_worker/src/activities.ts b/services/apps/connectors_worker/src/activities.ts index c69b2985c9..585832d864 100644 --- a/services/apps/connectors_worker/src/activities.ts +++ b/services/apps/connectors_worker/src/activities.ts @@ -1,3 +1,4 @@ import { claimDue, reschedule, startRun, touchHeartbeat } from './activities/dispatcherActivities' +import { executeSync } from './activities/syncRunActivities' -export { claimDue, reschedule, startRun, touchHeartbeat } +export { claimDue, executeSync, reschedule, startRun, touchHeartbeat } diff --git a/services/apps/connectors_worker/src/activities/syncRunActivities.ts b/services/apps/connectors_worker/src/activities/syncRunActivities.ts new file mode 100644 index 0000000000..bab19f2844 --- /dev/null +++ b/services/apps/connectors_worker/src/activities/syncRunActivities.ts @@ -0,0 +1,70 @@ +import { Context } from '@temporalio/activity' + +import { getSync } from '@crowd/connectors' +import type { SyncContext } from '@crowd/connectors' +import { + getUnitById, + recordRunFailure, + recordRunSuccess, +} from '@crowd/data-access-layer/src/connectors' +import { dbStoreQx } from '@crowd/data-access-layer/src/queryExecutor' +import { getChildLogger } from '@crowd/logging' + +import { svc } from '../main' + +const DEAD_LETTER_AFTER = 5 +const HEARTBEAT_INTERVAL_MS = 10_000 + +export async function executeSync(unitId: string): Promise { + const qx = dbStoreQx(svc.postgres.writer) + + const unit = await getUnitById(qx, unitId) + if (!unit) { + throw new Error(`sync unit ${unitId} not found`) + } + + const activityContext = Context.current() + const log = getChildLogger('syncRun', svc.log, { + runId: activityContext.info.workflowExecution.runId, + unitId: unit.id, + platform: unit.platform, + syncName: unit.syncName, + channelName: unit.channelName, + }) + + let emittedCount = 0 + let committedWatermark = unit.watermark + + // POC only: emit collects counts in memory; the Kafka sink emitter arrives in M2 + const ctx: SyncContext = { + channel: { channelId: unit.channelId, channelName: unit.channelName }, + watermark: unit.watermark, + emit: async (records) => { + emittedCount += records.length + }, + commitWatermark: async (watermark) => { + committedWatermark = watermark + }, + log, + } + + const heartbeat = setInterval(() => activityContext.heartbeat(), HEARTBEAT_INTERVAL_MS) + + try { + const sync = getSync(unit.platform, unit.syncName) + await sync.run(ctx) + + await recordRunSuccess(qx, unitId, { + watermark: committedWatermark ?? {}, + emittedCount, + }) + log.info({ emittedCount }, 'sync run succeeded') + } catch (err) { + log.error(err, 'sync run failed') + // POC only: everything unclassified is framework.internal; the 7-class + // error taxonomy arrives with the M2 HTTP client + await recordRunFailure(qx, unitId, 'framework.internal', DEAD_LETTER_AFTER) + } finally { + clearInterval(heartbeat) + } +} diff --git a/services/apps/connectors_worker/src/main.ts b/services/apps/connectors_worker/src/main.ts index 7ebd785514..ec10cb39a6 100644 --- a/services/apps/connectors_worker/src/main.ts +++ b/services/apps/connectors_worker/src/main.ts @@ -1,5 +1,7 @@ import { Config } from '@crowd/archetype-standard' import { Options, ServiceWorker } from '@crowd/archetype-worker' +import { registerConnector } from '@crowd/connectors' +import { dummyConnector } from '@crowd/connectors/src/testing/dummyConnector' import { scheduleDispatcher } from './schedules/dispatcher' @@ -27,6 +29,10 @@ const options: Options = { export const svc = new ServiceWorker(config, options) +// POC only: dummy connector drives the control-plane end-to-end; real +// connectors register here starting with GitHub in M4 +registerConnector(dummyConnector) + setImmediate(async () => { await svc.init() diff --git a/services/apps/connectors_worker/src/workflows.ts b/services/apps/connectors_worker/src/workflows.ts index 2044bd7bf3..67cb17400b 100644 --- a/services/apps/connectors_worker/src/workflows.ts +++ b/services/apps/connectors_worker/src/workflows.ts @@ -1,3 +1,4 @@ import { dispatcher } from './workflows/dispatcher' +import { syncRun } from './workflows/syncRun' -export { dispatcher } +export { dispatcher, syncRun } diff --git a/services/apps/connectors_worker/src/workflows/dispatcher.ts b/services/apps/connectors_worker/src/workflows/dispatcher.ts index 44cbbae714..5908059622 100644 --- a/services/apps/connectors_worker/src/workflows/dispatcher.ts +++ b/services/apps/connectors_worker/src/workflows/dispatcher.ts @@ -1,4 +1,4 @@ -import { proxyActivities } from '@temporalio/workflow' +import { log, proxyActivities } from '@temporalio/workflow' import type * as activities from '../activities/dispatcherActivities' @@ -15,7 +15,11 @@ export async function dispatcher(): Promise { const units = await activity.claimDue(CLAIM_LIMIT) for (const unit of units) { - await activity.startRun(unit) - await activity.reschedule(unit.id, unit.platform, unit.syncName) + try { + await activity.startRun(unit) + await activity.reschedule(unit.id, unit.platform, unit.syncName) + } catch (err) { + log.error('failed to dispatch sync unit', { unitId: unit.id, err }) + } } } diff --git a/services/apps/connectors_worker/src/workflows/syncRun.ts b/services/apps/connectors_worker/src/workflows/syncRun.ts new file mode 100644 index 0000000000..40f2637e4e --- /dev/null +++ b/services/apps/connectors_worker/src/workflows/syncRun.ts @@ -0,0 +1,13 @@ +import { proxyActivities } from '@temporalio/workflow' + +import type * as activities from '../activities/syncRunActivities' + +const activity = proxyActivities({ + startToCloseTimeout: '30 minutes', + heartbeatTimeout: '1 minute', + retry: { maximumAttempts: 1 }, +}) + +export async function syncRun(unitId: string): Promise { + await activity.executeSync(unitId) +} diff --git a/services/libs/connectors/src/testing/dummyConnector.ts b/services/libs/connectors/src/testing/dummyConnector.ts new file mode 100644 index 0000000000..f63866084a --- /dev/null +++ b/services/libs/connectors/src/testing/dummyConnector.ts @@ -0,0 +1,18 @@ +import type { Manifest, SyncContext } from '../types' + +const TICK_COUNT = 3 + +export const dummyConnector: Manifest = { + platform: 'dummy', + syncs: [ + { + name: 'ticks', + cadenceMinutes: 60, + run: async (ctx: SyncContext) => { + await ctx.emit(Array.from({ length: TICK_COUNT }, (_, index) => ({ tick: index }))) + await ctx.commitWatermark({ since: new Date().toISOString() }) + }, + }, + ], + discover: async () => [{ channelId: 'dummy-channel', channelName: 'dummy/channel' }], +} From ff1248f3950cd198ed48e8b9296364db5d832d8b Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 12 Aug 2026 11:06:14 +0100 Subject: [PATCH 11/19] feat: add connectors worker docker setup Signed-off-by: Mouad BANI --- scripts/cli | 2 +- scripts/services/connectors-worker.yaml | 57 +++++++++++++++++++ .../docker/Dockerfile.connectors_worker | 23 ++++++++ 3 files changed, 81 insertions(+), 1 deletion(-) create mode 100644 scripts/services/connectors-worker.yaml create mode 100644 scripts/services/docker/Dockerfile.connectors_worker diff --git a/scripts/cli b/scripts/cli index 0b226a614e..6917ed0bc0 100755 --- a/scripts/cli +++ b/scripts/cli @@ -1189,7 +1189,7 @@ while test $# -gt 0; do exit ;; clean-start-fe-dev) - IGNORED_SERVICES=("frontend" "python-worker" "job-generator" "webhook-api" "profiles-worker" "organizations-enrichment-worker" "merge-suggestions-worker" "members-enrichment-worker" "exports-worker" "entity-merging-worker" "cache-worker" "categorization-worker" "cron-service" "data-sink-worker" "git-integration" "mailing-list-integration" "integration-run-worker" "integration-stream-worker" "nango-webhook-api" "nango-worker" "script-executor-worker" "search-sync-api" "search-sync-worker" "security-best-practices-worker" "snowflake-connectors-worker" "automatic-projects-discovery-worker" "pcc-sync-worker" "projects-evaluation-worker" "bq-dataset-ingest" "cargo-worker" "dockerhub-sync" "github-repos-enricher" "go-worker" "maven-worker" "npm-worker" "nuget-worker" "osv-worker" "packagist-worker" "pypi-worker" "rubygems-worker" "security-contacts-worker") + IGNORED_SERVICES=("frontend" "python-worker" "job-generator" "webhook-api" "profiles-worker" "organizations-enrichment-worker" "merge-suggestions-worker" "members-enrichment-worker" "exports-worker" "entity-merging-worker" "cache-worker" "categorization-worker" "cron-service" "data-sink-worker" "git-integration" "mailing-list-integration" "integration-run-worker" "integration-stream-worker" "nango-webhook-api" "nango-worker" "connectors-worker" "script-executor-worker" "search-sync-api" "search-sync-worker" "security-best-practices-worker" "snowflake-connectors-worker" "automatic-projects-discovery-worker" "pcc-sync-worker" "projects-evaluation-worker" "bq-dataset-ingest" "cargo-worker" "dockerhub-sync" "github-repos-enricher" "go-worker" "maven-worker" "npm-worker" "nuget-worker" "osv-worker" "packagist-worker" "pypi-worker" "rubygems-worker" "security-contacts-worker") CLEAN_START=1 DEV=1 start diff --git a/scripts/services/connectors-worker.yaml b/scripts/services/connectors-worker.yaml new file mode 100644 index 0000000000..140b27639e --- /dev/null +++ b/scripts/services/connectors-worker.yaml @@ -0,0 +1,57 @@ +version: '3.1' + +x-env-args: &env-args + DOCKER_BUILDKIT: 1 + NODE_ENV: docker + SERVICE: connectors-worker + CROWD_TEMPORAL_TASKQUEUE: connectors + SHELL: /bin/sh + +services: + connectors-worker: + build: + context: ../../ + dockerfile: ./scripts/services/docker/Dockerfile.connectors_worker + command: 'pnpm run start' + working_dir: /usr/crowd/app/services/apps/connectors_worker + env_file: + - ../../backend/.env.dist.local + - ../../backend/.env.dist.composed + - ../../backend/.env.override.local + - ../../backend/.env.override.composed + environment: + <<: *env-args + restart: always + networks: + - crowd-bridge + + connectors-worker-dev: + build: + context: ../../ + dockerfile: ./scripts/services/docker/Dockerfile.connectors_worker + command: 'pnpm run dev' + working_dir: /usr/crowd/app/services/apps/connectors_worker + env_file: + - ../../backend/.env.dist.local + - ../../backend/.env.dist.composed + - ../../backend/.env.override.local + - ../../backend/.env.override.composed + environment: + <<: *env-args + hostname: connectors-worker + networks: + - crowd-bridge + volumes: + - ../../services/libs/common/src:/usr/crowd/app/services/libs/common/src + - ../../services/libs/connectors/src:/usr/crowd/app/services/libs/connectors/src + - ../../services/libs/data-access-layer/src:/usr/crowd/app/services/libs/data-access-layer/src + - ../../services/libs/database/src:/usr/crowd/app/services/libs/database/src + - ../../services/libs/logging/src:/usr/crowd/app/services/libs/logging/src + - ../../services/libs/redis/src:/usr/crowd/app/services/libs/redis/src + - ../../services/libs/temporal/src:/usr/crowd/app/services/libs/temporal/src + - ../../services/libs/types/src:/usr/crowd/app/services/libs/types/src + - ../../services/apps/connectors_worker/src:/usr/crowd/app/services/apps/connectors_worker/src + +networks: + crowd-bridge: + external: true diff --git a/scripts/services/docker/Dockerfile.connectors_worker b/scripts/services/docker/Dockerfile.connectors_worker new file mode 100644 index 0000000000..8fb33c1aff --- /dev/null +++ b/scripts/services/docker/Dockerfile.connectors_worker @@ -0,0 +1,23 @@ +FROM node:20-alpine as builder + +RUN apk add --no-cache python3 make g++ + +WORKDIR /usr/crowd/app +RUN npm install -g corepack@latest && corepack enable pnpm && corepack prepare pnpm@9.15.0 --activate + +COPY ./pnpm-workspace.yaml ./pnpm-lock.yaml ./ +RUN pnpm fetch + +COPY ./services ./services +RUN pnpm i --frozen-lockfile + +FROM node:20-bookworm-slim as runner + +WORKDIR /usr/crowd/app +RUN npm install -g corepack@latest && corepack enable pnpm && corepack prepare pnpm@9.15.0 --activate && apt update && apt install -y ca-certificates --no-install-recommends && rm -rf /var/lib/apt/lists/* + +COPY --from=builder /usr/crowd/app/node_modules ./node_modules +COPY --from=builder /usr/crowd/app/services/base.tsconfig.json ./services/base.tsconfig.json +COPY --from=builder /usr/crowd/app/services/libs ./services/libs +COPY --from=builder /usr/crowd/app/services/archetypes/ ./services/archetypes +COPY --from=builder /usr/crowd/app/services/apps/connectors_worker/ ./services/apps/connectors_worker From ab10d0cc35786a044b7d819ebc919cff2128097b Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 12 Aug 2026 11:13:54 +0100 Subject: [PATCH 12/19] fix: rethrow sync run errors so temporal reflects failures Signed-off-by: Mouad BANI --- .../apps/connectors_worker/src/activities/syncRunActivities.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/services/apps/connectors_worker/src/activities/syncRunActivities.ts b/services/apps/connectors_worker/src/activities/syncRunActivities.ts index bab19f2844..391830360f 100644 --- a/services/apps/connectors_worker/src/activities/syncRunActivities.ts +++ b/services/apps/connectors_worker/src/activities/syncRunActivities.ts @@ -64,6 +64,7 @@ export async function executeSync(unitId: string): Promise { // POC only: everything unclassified is framework.internal; the 7-class // error taxonomy arrives with the M2 HTTP client await recordRunFailure(qx, unitId, 'framework.internal', DEAD_LETTER_AFTER) + throw err } finally { clearInterval(heartbeat) } From 9de976ffd31f32ca200197f60b0515f9ab0483c1 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 12 Aug 2026 12:37:18 +0100 Subject: [PATCH 13/19] feat: add connectors http core with error taxonomy Signed-off-by: Mouad BANI --- pnpm-lock.yaml | 65 +++---- services/libs/connectors/package.json | 1 + services/libs/connectors/src/http/client.ts | 184 ++++++++++++++++++++ services/libs/connectors/src/http/errors.ts | 84 +++++++++ services/libs/connectors/src/index.ts | 2 + 5 files changed, 296 insertions(+), 40 deletions(-) create mode 100644 services/libs/connectors/src/http/client.ts create mode 100644 services/libs/connectors/src/http/errors.ts diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 63c18dadf8..e7b06665b0 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2290,6 +2290,9 @@ importers: '@crowd/logging': specifier: workspace:* version: link:../logging + axios: + specifier: ^1.6.8 + version: 1.16.1 zod: specifier: ^3.22.0 version: 3.25.76 @@ -5719,9 +5722,6 @@ packages: axios@1.13.1: resolution: {integrity: sha512-hU4EGxxt+j7TQijx1oYdAjw4xuIp1wRQSsbMFwSthCWeBQur1eF+qJ5iQ5sN3Tw8YRzQNKb8jszgBdMDVqwJcw==} - axios@1.13.5: - resolution: {integrity: sha512-cz4ur7Vb0xS4/KUN0tPWe44eqxrIu31me+fbang3ijiNscE129POzipJJA6zniq2C/Z6sJCjMimjS8Lc/GAs8Q==} - axios@1.16.1: resolution: {integrity: sha512-caYkukvroVPO8KrzuJEb50Hm07KwfBZPEC3VeFHTsqWHvKTsy54hjJz9BS/cdaypROE2rH6xvm9mHX4fgWkr3A==} @@ -7129,15 +7129,6 @@ packages: fn.name@1.1.0: resolution: {integrity: sha512-GRnmB5gPyJpAhTQdSZTSp9uaPSvl09KoYcMQtsB9rQoOmzs9dH6ffeccH+Z+cv6P68Hu5bC6JjRh4Ah/mHSNRw==} - follow-redirects@1.15.11: - resolution: {integrity: sha512-deG2P0JfjrTxl50XGCDyfI97ZGVCxIpfKYmfyrQ54n5FO/0gfIES8C/Psl6kWVDolizcaaxZJnTS0QSMxvnsBQ==} - engines: {node: '>=4.0'} - peerDependencies: - debug: '*' - peerDependenciesMeta: - debug: - optional: true - follow-redirects@1.15.6: resolution: {integrity: sha512-wWN62YITEaOpSK584EZXJafH1AGpO8RVgElfkuXbTOrPX4fIfOyEpW/CsiNd8JdYrAoOvafRTOEnvsO++qCqFA==} engines: {node: '>=4.0'} @@ -7350,11 +7341,11 @@ packages: glob@6.0.4: resolution: {integrity: sha512-MKZeRNyYZAVVVG1oZeLaWie1uweH40m9AZwIwxyPbTSX4hHrVYSzLg0Ro5Z5R7XKkIX+Cc6oD1rqeDJnwsB8/A==} - deprecated: Glob versions prior to v9 are no longer supported + deprecated: Old versions of glob are not supported, and contain widely publicized security vulnerabilities, which have been fixed in the current version. Please update. Support for old versions may be purchased (at exorbitant rates) by contacting i@izs.me glob@7.2.3: resolution: {integrity: sha512-nFR0zLpU2YCaRxwoCJvL6UvCH2JFyFVIvwTLsIf21AuHlMskA1hhTdk+LlYJtOlYt9v6dvszD2BGRqBL+iQK9Q==} - deprecated: Glob versions prior to v9 are no longer supported + deprecated: Old versions of glob are not supported, and contain widely publicized security vulnerabilities, which have been fixed in the current version. Please update. Support for old versions may be purchased (at exorbitant rates) by contacting i@izs.me global-directory@4.0.1: resolution: {integrity: sha512-wHTUcDUoZ1H5/0iVqEudYW4/kAlN5cZ3j/bXn0Dpbizl9iaUVeWSHqiOjsgk6OW2bkLclbBjzewBz6weQ1zA2Q==} @@ -11046,8 +11037,8 @@ snapshots: dependencies: '@aws-crypto/sha256-browser': 3.0.0 '@aws-crypto/sha256-js': 3.0.0 - '@aws-sdk/client-sso-oidc': 3.572.0 - '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) + '@aws-sdk/client-sso-oidc': 3.572.0(@aws-sdk/client-sts@3.572.0) + '@aws-sdk/client-sts': 3.572.0 '@aws-sdk/core': 3.572.0 '@aws-sdk/credential-provider-node': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0) '@aws-sdk/middleware-host-header': 3.567.0 @@ -11241,11 +11232,11 @@ snapshots: transitivePeerDependencies: - aws-crt - '@aws-sdk/client-sso-oidc@3.572.0': + '@aws-sdk/client-sso-oidc@3.572.0(@aws-sdk/client-sts@3.572.0)': dependencies: '@aws-crypto/sha256-browser': 3.0.0 '@aws-crypto/sha256-js': 3.0.0 - '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) + '@aws-sdk/client-sts': 3.572.0 '@aws-sdk/core': 3.572.0 '@aws-sdk/credential-provider-node': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0) '@aws-sdk/middleware-host-header': 3.567.0 @@ -11284,6 +11275,7 @@ snapshots: '@smithy/util-utf8': 2.3.0 tslib: 2.6.2 transitivePeerDependencies: + - '@aws-sdk/client-sts' - aws-crt '@aws-sdk/client-sso@3.556.0': @@ -11459,11 +11451,11 @@ snapshots: transitivePeerDependencies: - aws-crt - '@aws-sdk/client-sts@3.572.0(@aws-sdk/client-sso-oidc@3.572.0)': + '@aws-sdk/client-sts@3.572.0': dependencies: '@aws-crypto/sha256-browser': 3.0.0 '@aws-crypto/sha256-js': 3.0.0 - '@aws-sdk/client-sso-oidc': 3.572.0 + '@aws-sdk/client-sso-oidc': 3.572.0(@aws-sdk/client-sts@3.572.0) '@aws-sdk/core': 3.572.0 '@aws-sdk/credential-provider-node': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0) '@aws-sdk/middleware-host-header': 3.567.0 @@ -11502,7 +11494,6 @@ snapshots: '@smithy/util-utf8': 2.3.0 tslib: 2.6.2 transitivePeerDependencies: - - '@aws-sdk/client-sso-oidc' - aws-crt '@aws-sdk/client-sts@3.985.0': @@ -11668,7 +11659,7 @@ snapshots: '@aws-sdk/credential-provider-ini@3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0)': dependencies: - '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) + '@aws-sdk/client-sts': 3.572.0 '@aws-sdk/credential-provider-env': 3.568.0 '@aws-sdk/credential-provider-process': 3.572.0 '@aws-sdk/credential-provider-sso': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) @@ -11845,7 +11836,7 @@ snapshots: '@aws-sdk/credential-provider-web-identity@3.568.0(@aws-sdk/client-sts@3.572.0)': dependencies: - '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) + '@aws-sdk/client-sts': 3.572.0 '@aws-sdk/types': 3.567.0 '@smithy/property-provider': 2.2.0 '@smithy/types': 2.12.0 @@ -12157,7 +12148,7 @@ snapshots: '@aws-sdk/token-providers@3.572.0(@aws-sdk/client-sso-oidc@3.572.0)': dependencies: - '@aws-sdk/client-sso-oidc': 3.572.0 + '@aws-sdk/client-sso-oidc': 3.572.0(@aws-sdk/client-sts@3.572.0) '@aws-sdk/types': 3.567.0 '@smithy/property-provider': 2.2.0 '@smithy/shared-ini-file-loader': 2.4.0 @@ -13660,9 +13651,10 @@ snapshots: '@sendgrid/client@8.1.3': dependencies: '@sendgrid/helpers': 8.0.0 - axios: 1.13.5 + axios: 1.16.1 transitivePeerDependencies: - debug + - supports-color '@sendgrid/helpers@8.0.0': dependencies: @@ -13674,6 +13666,7 @@ snapshots: '@sendgrid/helpers': 8.0.0 transitivePeerDependencies: - debug + - supports-color '@sindresorhus/is@0.14.0': {} @@ -13691,7 +13684,7 @@ snapshots: '@slack/types': 2.11.0 '@types/is-stream': 1.1.0 '@types/node': 20.12.7 - axios: 1.13.5 + axios: 1.16.1 eventemitter3: 3.1.2 form-data: 2.5.1 is-electron: 2.2.2 @@ -13700,6 +13693,7 @@ snapshots: p-retry: 4.6.2 transitivePeerDependencies: - debug + - supports-color '@slack/webhook@6.1.0': dependencies: @@ -15352,7 +15346,7 @@ snapshots: axios@0.21.4: dependencies: - follow-redirects: 1.15.11 + follow-redirects: 1.16.0 transitivePeerDependencies: - debug @@ -15373,7 +15367,7 @@ snapshots: axios@1.12.0: dependencies: - follow-redirects: 1.15.11 + follow-redirects: 1.16.0 form-data: 4.0.5 proxy-from-env: 1.1.0 transitivePeerDependencies: @@ -15387,14 +15381,6 @@ snapshots: transitivePeerDependencies: - debug - axios@1.13.5: - dependencies: - follow-redirects: 1.15.11 - form-data: 4.0.5 - proxy-from-env: 1.1.0 - transitivePeerDependencies: - - debug - axios@1.16.1: dependencies: follow-redirects: 1.16.0 @@ -17170,8 +17156,6 @@ snapshots: fn.name@1.1.0: {} - follow-redirects@1.15.11: {} - follow-redirects@1.15.6: {} follow-redirects@1.16.0: {} @@ -19177,10 +19161,11 @@ snapshots: peopledatalabs@6.1.5: dependencies: - axios: 1.13.5 + axios: 1.16.1 copy-anything: 3.0.5 transitivePeerDependencies: - debug + - supports-color pg-cloudflare@1.1.1: optional: true @@ -20085,7 +20070,7 @@ snapshots: asn1.js: 5.4.1 asn1.js-rfc2560: 5.0.1(asn1.js@5.4.1) asn1.js-rfc5280: 3.0.0 - axios: 1.13.5 + axios: 1.16.1 big-integer: 1.6.52 bignumber.js: 9.1.2 bn.js: 5.2.1 diff --git a/services/libs/connectors/package.json b/services/libs/connectors/package.json index 2e5d961cda..6346f02d6c 100644 --- a/services/libs/connectors/package.json +++ b/services/libs/connectors/package.json @@ -16,6 +16,7 @@ "@crowd/common": "workspace:*", "@crowd/data-access-layer": "workspace:*", "@crowd/logging": "workspace:*", + "axios": "^1.6.8", "zod": "^3.22.0" } } diff --git a/services/libs/connectors/src/http/client.ts b/services/libs/connectors/src/http/client.ts new file mode 100644 index 0000000000..6712c0b631 --- /dev/null +++ b/services/libs/connectors/src/http/client.ts @@ -0,0 +1,184 @@ +import axios, { AxiosHeaders, AxiosRequestConfig, AxiosResponse } from 'axios' + +import { timeout } from '@crowd/common' +import type { Logger } from '@crowd/logging' + +import { + ConnectorError, + ProviderUnavailableError, + RateLimitError, + errorFromHttpStatus, +} from './errors' + +export interface IPooledToken { + id: string + value: string +} + +export interface HttpResponse { + status: number + headers: Record + data: unknown +} + +export type TokenApplier = (config: AxiosRequestConfig, token: IPooledToken) => AxiosRequestConfig + +export type ResponseInterpreter = (response: HttpResponse) => ConnectorError | null + +export interface HttpClientDeps { + acquireToken: () => Promise + parkToken: (tokenId: string, resumeAt: Date) => Promise + quarantineToken: (tokenId: string) => Promise + correctBudget: (headers: Record) => Promise + log: Logger + applyToken?: TokenApplier + interpretResponse?: ResponseInterpreter +} + +export interface ConnectorHttp { + request(config: AxiosRequestConfig): Promise +} + +const MAX_ATTEMPTS = 3 +const BACKOFF_BASE_MS = 1000 +const RATE_LIMIT_FALLBACK_MS = 60_000 + +export function createHttpClient(deps: HttpClientDeps): ConnectorHttp { + return { + request: (config: AxiosRequestConfig) => requestWithRetry(deps, config), + } +} + +async function requestWithRetry(deps: HttpClientDeps, config: AxiosRequestConfig): Promise { + let lastError: ConnectorError = new ProviderUnavailableError() + for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) { + try { + return await attemptRequest(deps, config, true) + } catch (err) { + if (!(err instanceof ConnectorError) || err.errorClass !== 'provider.unavailable') { + throw err + } + lastError = err + if (attempt < MAX_ATTEMPTS) { + const delay = BACKOFF_BASE_MS * 2 ** (attempt - 1) + deps.log.warn({ attempt, delay, reason: err.message }, 'provider unavailable, backing off') + await timeout(delay) + } + } + } + throw lastError +} + +async function attemptRequest( + deps: HttpClientDeps, + config: AxiosRequestConfig, + allowTokenRotation: boolean, +): Promise { + const token = await deps.acquireToken() + const response = await send(deps, config, token) + const headers = normalizeHeaders(response.headers) + const error = classifyResponse(deps, response.status, headers, response.data) + + if (!error) { + await deps.correctBudget(headers) + return response.data + } + + if (error.errorClass === 'provider.rate_limit') { + const resumeAt = error.options?.resumeAt ?? computeResumeAt(headers) + await deps.parkToken(token.id, resumeAt) + if (allowTokenRotation) { + deps.log.info( + { tokenId: token.id, resumeAt }, + 'token rate limited, retrying with fresh token', + ) + return attemptRequest(deps, config, false) + } + throw new RateLimitError(error.message, { ...error.options, resumeAt }) + } + + if (error.errorClass === 'provider.auth') { + await deps.quarantineToken(token.id) + deps.log.warn( + { tokenId: token.id, status: response.status }, + 'token quarantined on auth failure', + ) + } + + throw error +} + +async function send( + deps: HttpClientDeps, + config: AxiosRequestConfig, + token: IPooledToken, +): Promise> { + const applyToken = deps.applyToken ?? applyBearerToken + try { + return await axios.request({ ...applyToken(config, token), validateStatus: () => true }) + } catch (err) { + throw new ProviderUnavailableError('no response from provider', { cause: err }) + } +} + +function applyBearerToken(config: AxiosRequestConfig, token: IPooledToken): AxiosRequestConfig { + const headers = AxiosHeaders.from(config.headers) + if (!headers.has('Authorization')) { + headers.set('Authorization', `Bearer ${token.value}`) + } + return { ...config, headers } +} + +function classifyResponse( + deps: HttpClientDeps, + status: number, + headers: Record, + data: unknown, +): ConnectorError | null { + const custom = deps.interpretResponse?.({ status, headers, data }) + if (custom) { + return custom + } + if (isRateLimited(status, headers)) { + return new RateLimitError(`provider rate limited (status ${status})`, { + status, + resumeAt: computeResumeAt(headers), + }) + } + if (status >= 400) { + return errorFromHttpStatus(status) + } + return null +} + +function isRateLimited(status: number, headers: Record): boolean { + if (status === 429) { + return true + } + if (status === 403 && headers['x-ratelimit-remaining'] === '0') { + return true + } + return 'retry-after' in headers +} + +function computeResumeAt(headers: Record): Date { + const retryAfterSeconds = Number(headers['retry-after']) + if (Number.isFinite(retryAfterSeconds) && retryAfterSeconds > 0) { + return new Date(Date.now() + retryAfterSeconds * 1000) + } + const resetEpochSeconds = Number(headers['x-ratelimit-reset']) + if (Number.isFinite(resetEpochSeconds) && resetEpochSeconds > 0) { + return new Date(resetEpochSeconds * 1000) + } + return new Date(Date.now() + RATE_LIMIT_FALLBACK_MS) +} + +function normalizeHeaders(raw: AxiosResponse['headers']): Record { + const headers: Record = {} + for (const [key, value] of Object.entries(raw)) { + if (value !== undefined && value !== null) { + headers[key.toLowerCase()] = Array.isArray(value) ? value.join(', ') : String(value) + } + } + return headers +} diff --git a/services/libs/connectors/src/http/errors.ts b/services/libs/connectors/src/http/errors.ts new file mode 100644 index 0000000000..7318cd12a6 --- /dev/null +++ b/services/libs/connectors/src/http/errors.ts @@ -0,0 +1,84 @@ +export type ErrorClass = + | 'provider.unavailable' + | 'provider.rate_limit' + | 'provider.auth' + | 'provider.contract' + | 'connector.code' + | 'sink.rejected' + | 'unknown' + +export interface ConnectorErrorOptions { + status?: number + resumeAt?: Date + cause?: unknown +} + +export class ConnectorError extends Error { + constructor( + readonly errorClass: ErrorClass, + message: string, + readonly options?: ConnectorErrorOptions, + ) { + super(message) + this.name = 'ConnectorError' + } +} + +export class ProviderUnavailableError extends ConnectorError { + constructor(message = 'provider unavailable', options?: ConnectorErrorOptions) { + super('provider.unavailable', message, options) + this.name = 'ProviderUnavailableError' + } +} + +export class RateLimitError extends ConnectorError { + constructor(message = 'rate limited by provider', options?: ConnectorErrorOptions) { + super('provider.rate_limit', message, options) + this.name = 'RateLimitError' + } +} + +export class ProviderAuthError extends ConnectorError { + constructor(message = 'provider authentication failed', options?: ConnectorErrorOptions) { + super('provider.auth', message, options) + this.name = 'ProviderAuthError' + } +} + +export class ProviderContractError extends ConnectorError { + constructor(message = 'unexpected provider response', options?: ConnectorErrorOptions) { + super('provider.contract', message, options) + this.name = 'ProviderContractError' + } +} + +export class ConnectorCodeError extends ConnectorError { + constructor(message = 'connector code error', options?: ConnectorErrorOptions) { + super('connector.code', message, options) + this.name = 'ConnectorCodeError' + } +} + +export function errorFromHttpStatus( + status: number | undefined, + message?: string, + options?: ConnectorErrorOptions, +): ConnectorError { + const opts = { ...options, status } + if (status === undefined) { + return new ProviderUnavailableError(message ?? 'no response from provider', opts) + } + if (status === 401 || status === 403) { + return new ProviderAuthError(message ?? `provider returned status ${status}`, opts) + } + if (status === 429) { + return new RateLimitError(message ?? `provider returned status ${status}`, opts) + } + if (status >= 500) { + return new ProviderUnavailableError(message ?? `provider returned status ${status}`, opts) + } + if (status >= 400) { + return new ProviderContractError(message ?? `provider returned status ${status}`, opts) + } + return new ConnectorError('unknown', message ?? `unexpected status ${status}`, opts) +} diff --git a/services/libs/connectors/src/index.ts b/services/libs/connectors/src/index.ts index 6e932df937..4e2d55400e 100644 --- a/services/libs/connectors/src/index.ts +++ b/services/libs/connectors/src/index.ts @@ -1,3 +1,5 @@ export * from './credentials' +export * from './http/client' +export * from './http/errors' export * from './registry' export * from './types' From f76efcc07c03236b5b0daf96fb351880d0bba17c Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 12 Aug 2026 12:40:01 +0100 Subject: [PATCH 14/19] fix: guard heartbeat callback and add connectors worker dockerignore Signed-off-by: Mouad BANI --- .../Dockerfile.connectors_worker.dockerignore | 18 ++++++++++++++++++ .../src/activities/syncRunActivities.ts | 8 +++++++- 2 files changed, 25 insertions(+), 1 deletion(-) create mode 100644 scripts/services/docker/Dockerfile.connectors_worker.dockerignore diff --git a/scripts/services/docker/Dockerfile.connectors_worker.dockerignore b/scripts/services/docker/Dockerfile.connectors_worker.dockerignore new file mode 100644 index 0000000000..4b74fc87af --- /dev/null +++ b/scripts/services/docker/Dockerfile.connectors_worker.dockerignore @@ -0,0 +1,18 @@ +**/.git +**/node_modules +**/venv* +**/.webpack +**/.serverless +**/.env +**/.env.* +**/.idea +**/.vscode +**/dist +.vscode/ +.github/ +frontend/ +scripts/ +.flake8 +*.md +Makefile +backend/ diff --git a/services/apps/connectors_worker/src/activities/syncRunActivities.ts b/services/apps/connectors_worker/src/activities/syncRunActivities.ts index 391830360f..fd8390a381 100644 --- a/services/apps/connectors_worker/src/activities/syncRunActivities.ts +++ b/services/apps/connectors_worker/src/activities/syncRunActivities.ts @@ -48,7 +48,13 @@ export async function executeSync(unitId: string): Promise { log, } - const heartbeat = setInterval(() => activityContext.heartbeat(), HEARTBEAT_INTERVAL_MS) + const heartbeat = setInterval(() => { + try { + activityContext.heartbeat() + } catch (err) { + log.warn({ errMsg: (err as Error).message }, 'heartbeat failed') + } + }, HEARTBEAT_INTERVAL_MS) try { const sync = getSync(unit.platform, unit.syncName) From 8a1391b5079d1fa08fe85624542d64aeec75a0a0 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 12 Aug 2026 12:47:09 +0100 Subject: [PATCH 15/19] fix: restrict rate limit detection to unambiguous signals Signed-off-by: Mouad BANI --- services/libs/connectors/src/http/client.ts | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/services/libs/connectors/src/http/client.ts b/services/libs/connectors/src/http/client.ts index 6712c0b631..cc07cce58a 100644 --- a/services/libs/connectors/src/http/client.ts +++ b/services/libs/connectors/src/http/client.ts @@ -155,10 +155,7 @@ function isRateLimited(status: number, headers: Record): boolean if (status === 429) { return true } - if (status === 403 && headers['x-ratelimit-remaining'] === '0') { - return true - } - return 'retry-after' in headers + return status === 403 && headers['x-ratelimit-remaining'] === '0' } function computeResumeAt(headers: Record): Date { From 5e4117a308e07119a94c604ed732ba9f7ba0762f Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 12 Aug 2026 17:51:33 +0100 Subject: [PATCH 16/19] fix: always apply pooled token in http client auth Signed-off-by: Mouad BANI --- services/libs/connectors/src/http/client.ts | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/services/libs/connectors/src/http/client.ts b/services/libs/connectors/src/http/client.ts index cc07cce58a..deb60509e4 100644 --- a/services/libs/connectors/src/http/client.ts +++ b/services/libs/connectors/src/http/client.ts @@ -123,9 +123,7 @@ async function send( function applyBearerToken(config: AxiosRequestConfig, token: IPooledToken): AxiosRequestConfig { const headers = AxiosHeaders.from(config.headers) - if (!headers.has('Authorization')) { - headers.set('Authorization', `Bearer ${token.value}`) - } + headers.set('Authorization', `Bearer ${token.value}`) return { ...config, headers } } From b5828760d9c62860670a88e6f813fad869382508 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 12 Aug 2026 18:14:16 +0100 Subject: [PATCH 17/19] fix: add default http timeout and narrow claimed unit payload Signed-off-by: Mouad BANI --- .../src/activities/dispatcherActivities.ts | 6 +++--- services/libs/connectors/src/http/client.ts | 7 ++++++- .../libs/data-access-layer/src/connectors/syncUnits.ts | 6 +++--- services/libs/data-access-layer/src/connectors/types.ts | 2 ++ 4 files changed, 14 insertions(+), 7 deletions(-) diff --git a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts index c5c47a644d..d1f21b1d61 100644 --- a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts +++ b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts @@ -1,6 +1,6 @@ import { getSync } from '@crowd/connectors' import { claimDueUnits, rescheduleUnit } from '@crowd/data-access-layer/src/connectors' -import type { ISyncUnit } from '@crowd/data-access-layer/src/connectors' +import type { IClaimedUnit } from '@crowd/data-access-layer/src/connectors' import { dbStoreQx } from '@crowd/data-access-layer/src/queryExecutor' import { RedisCache } from '@crowd/redis' import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@crowd/temporal' @@ -12,11 +12,11 @@ const TASK_QUEUE = 'connectors' const HEARTBEAT_TTL_SECONDS = 300 const CADENCE_JITTER_RATIO = 0.1 -export async function claimDue(limit: number): Promise { +export async function claimDue(limit: number): Promise { return claimDueUnits(dbStoreQx(svc.postgres.writer), limit) } -export async function startRun(unit: ISyncUnit): Promise { +export async function startRun(unit: IClaimedUnit): Promise { try { await svc.temporal.workflow.start('syncRun', { taskQueue: TASK_QUEUE, diff --git a/services/libs/connectors/src/http/client.ts b/services/libs/connectors/src/http/client.ts index deb60509e4..4daa5b7196 100644 --- a/services/libs/connectors/src/http/client.ts +++ b/services/libs/connectors/src/http/client.ts @@ -42,6 +42,7 @@ export interface ConnectorHttp { const MAX_ATTEMPTS = 3 const BACKOFF_BASE_MS = 1000 const RATE_LIMIT_FALLBACK_MS = 60_000 +const DEFAULT_TIMEOUT_MS = 60_000 export function createHttpClient(deps: HttpClientDeps): ConnectorHttp { return { @@ -115,7 +116,11 @@ async function send( ): Promise> { const applyToken = deps.applyToken ?? applyBearerToken try { - return await axios.request({ ...applyToken(config, token), validateStatus: () => true }) + return await axios.request({ + timeout: DEFAULT_TIMEOUT_MS, + ...applyToken(config, token), + validateStatus: () => true, + }) } catch (err) { throw new ProviderUnavailableError('no response from provider', { cause: err }) } diff --git a/services/libs/data-access-layer/src/connectors/syncUnits.ts b/services/libs/data-access-layer/src/connectors/syncUnits.ts index 3554b77935..df20b7e5a5 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.ts @@ -1,6 +1,6 @@ import type { QueryExecutor } from '../queryExecutor' -import type { ISyncRunSuccess, ISyncUnit, SyncUnitUpsert } from './types' +import type { IClaimedUnit, ISyncRunSuccess, ISyncUnit, SyncUnitUpsert } from './types' const MIN_INITIAL_DELAY_SECONDS = 10 const MAX_INITIAL_DELAY_SECONDS = 900 @@ -36,7 +36,7 @@ export async function upsertSyncUnits(qx: QueryExecutor, units: SyncUnitUpsert[] ) } -export async function claimDueUnits(qx: QueryExecutor, limit: number): Promise { +export async function claimDueUnits(qx: QueryExecutor, limit: number): Promise { return qx.select( `UPDATE integration.sync_units su SET "lockedAt" = now(), "updatedAt" = now() @@ -55,7 +55,7 @@ export async function claimDueUnits(qx: QueryExecutor, limit: number): Promise +export type IClaimedUnit = Pick + export interface ISyncRunSuccess { watermark: Record emittedCount: number From fac2c6289b2fd1b434dbce533303cb528847225b Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 13 Aug 2026 12:55:41 +0100 Subject: [PATCH 18/19] feat: add redis token pool for connectors Signed-off-by: Mouad BANI --- pnpm-lock.yaml | 23 ++-- services/libs/connectors/package.json | 1 + services/libs/connectors/src/index.ts | 1 + .../libs/connectors/src/pool/tokenPool.ts | 115 ++++++++++++++++++ 4 files changed, 130 insertions(+), 10 deletions(-) create mode 100644 services/libs/connectors/src/pool/tokenPool.ts diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index e7b06665b0..b0707d7c7c 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2290,6 +2290,9 @@ importers: '@crowd/logging': specifier: workspace:* version: link:../logging + '@crowd/redis': + specifier: workspace:* + version: link:../redis axios: specifier: ^1.6.8 version: 1.16.1 @@ -11037,8 +11040,8 @@ snapshots: dependencies: '@aws-crypto/sha256-browser': 3.0.0 '@aws-crypto/sha256-js': 3.0.0 - '@aws-sdk/client-sso-oidc': 3.572.0(@aws-sdk/client-sts@3.572.0) - '@aws-sdk/client-sts': 3.572.0 + '@aws-sdk/client-sso-oidc': 3.572.0 + '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) '@aws-sdk/core': 3.572.0 '@aws-sdk/credential-provider-node': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0) '@aws-sdk/middleware-host-header': 3.567.0 @@ -11232,11 +11235,11 @@ snapshots: transitivePeerDependencies: - aws-crt - '@aws-sdk/client-sso-oidc@3.572.0(@aws-sdk/client-sts@3.572.0)': + '@aws-sdk/client-sso-oidc@3.572.0': dependencies: '@aws-crypto/sha256-browser': 3.0.0 '@aws-crypto/sha256-js': 3.0.0 - '@aws-sdk/client-sts': 3.572.0 + '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) '@aws-sdk/core': 3.572.0 '@aws-sdk/credential-provider-node': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0) '@aws-sdk/middleware-host-header': 3.567.0 @@ -11275,7 +11278,6 @@ snapshots: '@smithy/util-utf8': 2.3.0 tslib: 2.6.2 transitivePeerDependencies: - - '@aws-sdk/client-sts' - aws-crt '@aws-sdk/client-sso@3.556.0': @@ -11451,11 +11453,11 @@ snapshots: transitivePeerDependencies: - aws-crt - '@aws-sdk/client-sts@3.572.0': + '@aws-sdk/client-sts@3.572.0(@aws-sdk/client-sso-oidc@3.572.0)': dependencies: '@aws-crypto/sha256-browser': 3.0.0 '@aws-crypto/sha256-js': 3.0.0 - '@aws-sdk/client-sso-oidc': 3.572.0(@aws-sdk/client-sts@3.572.0) + '@aws-sdk/client-sso-oidc': 3.572.0 '@aws-sdk/core': 3.572.0 '@aws-sdk/credential-provider-node': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0) '@aws-sdk/middleware-host-header': 3.567.0 @@ -11494,6 +11496,7 @@ snapshots: '@smithy/util-utf8': 2.3.0 tslib: 2.6.2 transitivePeerDependencies: + - '@aws-sdk/client-sso-oidc' - aws-crt '@aws-sdk/client-sts@3.985.0': @@ -11659,7 +11662,7 @@ snapshots: '@aws-sdk/credential-provider-ini@3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0)': dependencies: - '@aws-sdk/client-sts': 3.572.0 + '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) '@aws-sdk/credential-provider-env': 3.568.0 '@aws-sdk/credential-provider-process': 3.572.0 '@aws-sdk/credential-provider-sso': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) @@ -11836,7 +11839,7 @@ snapshots: '@aws-sdk/credential-provider-web-identity@3.568.0(@aws-sdk/client-sts@3.572.0)': dependencies: - '@aws-sdk/client-sts': 3.572.0 + '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) '@aws-sdk/types': 3.567.0 '@smithy/property-provider': 2.2.0 '@smithy/types': 2.12.0 @@ -12148,7 +12151,7 @@ snapshots: '@aws-sdk/token-providers@3.572.0(@aws-sdk/client-sso-oidc@3.572.0)': dependencies: - '@aws-sdk/client-sso-oidc': 3.572.0(@aws-sdk/client-sts@3.572.0) + '@aws-sdk/client-sso-oidc': 3.572.0 '@aws-sdk/types': 3.567.0 '@smithy/property-provider': 2.2.0 '@smithy/shared-ini-file-loader': 2.4.0 diff --git a/services/libs/connectors/package.json b/services/libs/connectors/package.json index 6346f02d6c..d6d839b833 100644 --- a/services/libs/connectors/package.json +++ b/services/libs/connectors/package.json @@ -16,6 +16,7 @@ "@crowd/common": "workspace:*", "@crowd/data-access-layer": "workspace:*", "@crowd/logging": "workspace:*", + "@crowd/redis": "workspace:*", "axios": "^1.6.8", "zod": "^3.22.0" } diff --git a/services/libs/connectors/src/index.ts b/services/libs/connectors/src/index.ts index 4e2d55400e..d5b10139c0 100644 --- a/services/libs/connectors/src/index.ts +++ b/services/libs/connectors/src/index.ts @@ -1,5 +1,6 @@ export * from './credentials' export * from './http/client' export * from './http/errors' +export * from './pool/tokenPool' export * from './registry' export * from './types' diff --git a/services/libs/connectors/src/pool/tokenPool.ts b/services/libs/connectors/src/pool/tokenPool.ts new file mode 100644 index 0000000000..b62685d0fa --- /dev/null +++ b/services/libs/connectors/src/pool/tokenPool.ts @@ -0,0 +1,115 @@ +import type { RedisClient } from '@crowd/redis' + +import type { IPooledToken } from '../http/client' +import { ProviderAuthError, RateLimitError } from '../http/errors' + +interface ITokenState { + value: string + parkedUntil?: string + quarantined?: boolean +} + +export interface TokenPool { + acquire(): Promise + park(tokenId: string, resumeAt: Date): Promise + quarantine(tokenId: string): Promise + seed(tokenId: string, value: string): Promise + earliestResumeAt(): Promise +} + +export function createTokenPool( + redis: RedisClient, + platform: string, + connectionId: string, +): TokenPool { + const tokensKey = `connectors:pool:${platform}:${connectionId}:tokens` + const lruKey = `connectors:pool:${platform}:${connectionId}:lru` + + async function readStates(): Promise> { + const raw = await redis.hGetAll(tokensKey) + const states = new Map() + for (const [id, json] of Object.entries(raw)) { + states.set(id, JSON.parse(json) as ITokenState) + } + return states + } + + function isHealthy(state: ITokenState, nowMs: number): boolean { + if (state.quarantined) { + return false + } + if (state.parkedUntil && new Date(state.parkedUntil).getTime() > nowMs) { + return false + } + return true + } + + function earliestParkedUntil(states: Map, nowMs: number): Date | null { + let earliest: Date | null = null + for (const state of states.values()) { + if (state.quarantined || !state.parkedUntil) { + continue + } + const parkedUntil = new Date(state.parkedUntil) + if (parkedUntil.getTime() <= nowMs) { + continue + } + if (!earliest || parkedUntil < earliest) { + earliest = parkedUntil + } + } + return earliest + } + + // POC only: read-modify-write can lose a concurrent park/quarantine on the same token + // within a ~ms window; accepted tradeoff — fix with atomic writes when productizing. + async function updateState(tokenId: string, update: Partial): Promise { + const json = await redis.hGet(tokensKey, tokenId) + if (!json) { + return + } + const state = JSON.parse(json) as ITokenState + await redis.hSet(tokensKey, tokenId, JSON.stringify({ ...state, ...update })) + } + + return { + async acquire(): Promise { + const nowMs = Date.now() + const states = await readStates() + const ordered = await redis.zRange(lruKey, 0, -1) + for (const id of ordered) { + const state = states.get(id) + if (state && isHealthy(state, nowMs)) { + await redis.zAdd(lruKey, { score: nowMs, value: id }) + return { id, value: state.value } + } + } + const resumeAt = earliestParkedUntil(states, nowMs) + if (resumeAt) { + throw new RateLimitError('token pool exhausted', { resumeAt }) + } + throw new ProviderAuthError('token pool empty') + }, + + async park(tokenId: string, resumeAt: Date): Promise { + await updateState(tokenId, { parkedUntil: resumeAt.toISOString() }) + }, + + // POC only: quarantined tokens are kept for inspection and never revived automatically + async quarantine(tokenId: string): Promise { + await updateState(tokenId, { quarantined: true }) + }, + + async seed(tokenId: string, value: string): Promise { + const json = await redis.hGet(tokensKey, tokenId) + const state = json ? (JSON.parse(json) as ITokenState) : {} + await redis.hSet(tokensKey, tokenId, JSON.stringify({ ...state, value })) + await redis.zAdd(lruKey, { score: 0, value: tokenId }, { NX: true }) + }, + + async earliestResumeAt(): Promise { + const states = await readStates() + return earliestParkedUntil(states, Date.now()) + }, + } +} From 032d4cdde29f22c4fdd0598ba0342bc62f6fc6c2 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 13 Aug 2026 17:26:09 +0100 Subject: [PATCH 19/19] feat: add token budgets with dispatcher admission Signed-off-by: Mouad BANI --- .../apps/connectors_worker/src/activities.ts | 11 +- .../src/activities/dispatcherActivities.ts | 26 +++- services/apps/connectors_worker/src/types.ts | 7 + .../src/workflows/dispatcher.ts | 8 +- .../libs/connectors/src/pool/tokenPool.ts | 139 +++++++++++++++++- 5 files changed, 178 insertions(+), 13 deletions(-) diff --git a/services/apps/connectors_worker/src/activities.ts b/services/apps/connectors_worker/src/activities.ts index 585832d864..054ba53dff 100644 --- a/services/apps/connectors_worker/src/activities.ts +++ b/services/apps/connectors_worker/src/activities.ts @@ -1,4 +1,11 @@ -import { claimDue, reschedule, startRun, touchHeartbeat } from './activities/dispatcherActivities' +import { + admitByBudget, + claimDue, + deferUnit, + reschedule, + startRun, + touchHeartbeat, +} from './activities/dispatcherActivities' import { executeSync } from './activities/syncRunActivities' -export { claimDue, executeSync, reschedule, startRun, touchHeartbeat } +export { admitByBudget, claimDue, deferUnit, executeSync, reschedule, startRun, touchHeartbeat } diff --git a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts index d1f21b1d61..e64e2ed876 100644 --- a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts +++ b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts @@ -1,4 +1,4 @@ -import { getSync } from '@crowd/connectors' +import { createTokenPool, getSync } from '@crowd/connectors' import { claimDueUnits, rescheduleUnit } from '@crowd/data-access-layer/src/connectors' import type { IClaimedUnit } from '@crowd/data-access-layer/src/connectors' import { dbStoreQx } from '@crowd/data-access-layer/src/queryExecutor' @@ -6,16 +6,38 @@ import { RedisCache } from '@crowd/redis' import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@crowd/temporal' import { svc } from '../main' -import type { StartRunResult } from '../types' +import type { IAdmissionResult, StartRunResult } from '../types' const TASK_QUEUE = 'connectors' const HEARTBEAT_TTL_SECONDS = 300 const CADENCE_JITTER_RATIO = 0.1 +const DEFAULT_RUN_ESTIMATE = 50 +const DEFER_MIN_MS = 30_000 +const DEFER_JITTER_MS = 60_000 export async function claimDue(limit: number): Promise { return claimDueUnits(dbStoreQx(svc.postgres.writer), limit) } +export async function admitByBudget(units: IClaimedUnit[]): Promise { + const admitted: IClaimedUnit[] = [] + const deferred: IClaimedUnit[] = [] + for (const unit of units) { + const pool = createTokenPool(svc.redis, unit.platform, unit.integrationId) + if (await pool.hasHeadroom(DEFAULT_RUN_ESTIMATE)) { + admitted.push(unit) + } else { + deferred.push(unit) + } + } + return { admitted, deferred } +} + +export async function deferUnit(unitId: string): Promise { + const delayMs = DEFER_MIN_MS + Math.random() * DEFER_JITTER_MS + await rescheduleUnit(dbStoreQx(svc.postgres.writer), unitId, new Date(Date.now() + delayMs)) +} + export async function startRun(unit: IClaimedUnit): Promise { try { await svc.temporal.workflow.start('syncRun', { diff --git a/services/apps/connectors_worker/src/types.ts b/services/apps/connectors_worker/src/types.ts index ac915242b1..31f1c992c0 100644 --- a/services/apps/connectors_worker/src/types.ts +++ b/services/apps/connectors_worker/src/types.ts @@ -1 +1,8 @@ +import type { IClaimedUnit } from '@crowd/data-access-layer/src/connectors' + export type StartRunResult = 'started' | 'alreadyRunning' + +export interface IAdmissionResult { + admitted: IClaimedUnit[] + deferred: IClaimedUnit[] +} diff --git a/services/apps/connectors_worker/src/workflows/dispatcher.ts b/services/apps/connectors_worker/src/workflows/dispatcher.ts index 5908059622..f3d800e79c 100644 --- a/services/apps/connectors_worker/src/workflows/dispatcher.ts +++ b/services/apps/connectors_worker/src/workflows/dispatcher.ts @@ -14,7 +14,9 @@ export async function dispatcher(): Promise { const units = await activity.claimDue(CLAIM_LIMIT) - for (const unit of units) { + const { admitted, deferred } = await activity.admitByBudget(units) + + for (const unit of admitted) { try { await activity.startRun(unit) await activity.reschedule(unit.id, unit.platform, unit.syncName) @@ -22,4 +24,8 @@ export async function dispatcher(): Promise { log.error('failed to dispatch sync unit', { unitId: unit.id, err }) } } + + for (const unit of deferred) { + await activity.deferUnit(unit.id) + } } diff --git a/services/libs/connectors/src/pool/tokenPool.ts b/services/libs/connectors/src/pool/tokenPool.ts index b62685d0fa..53d74ce529 100644 --- a/services/libs/connectors/src/pool/tokenPool.ts +++ b/services/libs/connectors/src/pool/tokenPool.ts @@ -3,27 +3,58 @@ import type { RedisClient } from '@crowd/redis' import type { IPooledToken } from '../http/client' import { ProviderAuthError, RateLimitError } from '../http/errors' -interface ITokenState { - value: string - parkedUntil?: string - quarantined?: boolean +const PROBE_STALENESS_MS = 90_000 + +export interface BudgetSnapshot { + limit: number + remaining: number + resetAt: Date +} + +export type BudgetProbe = ( + platform: string, + connectionId: string, + tokenId: string, +) => Promise + +// POC only: the probe is the single source of truth for budgets (github /rate_limit is free and +// limits are per installation token); budgets for other platforms are a later decision. +export interface TokenPoolOptions { + probeBudget?: BudgetProbe } export interface TokenPool { acquire(): Promise + hasHeadroom(estimate: number): Promise park(tokenId: string, resumeAt: Date): Promise quarantine(tokenId: string): Promise seed(tokenId: string, value: string): Promise earliestResumeAt(): Promise } +interface ITokenState { + value: string + parkedUntil?: string + quarantined?: boolean +} + +interface IBucket { + limit: number + remaining: number + resetAtMs: number + probedAtMs: number +} + export function createTokenPool( redis: RedisClient, platform: string, connectionId: string, + options?: TokenPoolOptions, ): TokenPool { const tokensKey = `connectors:pool:${platform}:${connectionId}:tokens` const lruKey = `connectors:pool:${platform}:${connectionId}:lru` + const bucketKey = (tokenId: string) => + `connectors:pool:${platform}:${connectionId}:budget:${tokenId}` async function readStates(): Promise> { const raw = await redis.hGetAll(tokensKey) @@ -72,25 +103,117 @@ export function createTokenPool( await redis.hSet(tokensKey, tokenId, JSON.stringify({ ...state, ...update })) } + async function readBucket(tokenId: string): Promise { + const raw = await redis.hGetAll(bucketKey(tokenId)) + if (!raw.probedAt) { + return null + } + return { + limit: Number(raw.limit), + remaining: Number(raw.remaining), + resetAtMs: Number(raw.resetAt), + probedAtMs: Number(raw.probedAt), + } + } + + function needsProbe(bucket: IBucket | null, nowMs: number): boolean { + return !bucket || nowMs - bucket.probedAtMs > PROBE_STALENESS_MS || nowMs >= bucket.resetAtMs + } + + async function loadBucket( + probe: BudgetProbe, + tokenId: string, + nowMs: number, + ): Promise { + const bucket = await readBucket(tokenId) + if (!needsProbe(bucket, nowMs)) { + return bucket + } + const snapshot = await probe(platform, connectionId, tokenId) + if (!snapshot) { + return null + } + const probed = { + limit: snapshot.limit, + remaining: snapshot.remaining, + resetAtMs: snapshot.resetAt.getTime(), + probedAtMs: nowMs, + } + await redis.hSet(bucketKey(tokenId), { + limit: String(probed.limit), + remaining: String(probed.remaining), + resetAt: String(probed.resetAtMs), + probedAt: String(probed.probedAtMs), + }) + return probed + } + return { async acquire(): Promise { const nowMs = Date.now() const states = await readStates() const ordered = await redis.zRange(lruKey, 0, -1) + const probe = options?.probeBudget + let earliestBudgetResetAt: Date | null = null for (const id of ordered) { const state = states.get(id) - if (state && isHealthy(state, nowMs)) { - await redis.zAdd(lruKey, { score: nowMs, value: id }) - return { id, value: state.value } + if (!state || !isHealthy(state, nowMs)) { + continue } + if (probe) { + const bucket = await loadBucket(probe, id, nowMs) + if (bucket && bucket.remaining <= 0) { + const resetAt = new Date(bucket.resetAtMs) + if (!earliestBudgetResetAt || resetAt < earliestBudgetResetAt) { + earliestBudgetResetAt = resetAt + } + continue + } + if (bucket) { + await redis.hIncrBy(bucketKey(id), 'remaining', -1) + } + } + await redis.zAdd(lruKey, { score: nowMs, value: id }) + return { id, value: state.value } } - const resumeAt = earliestParkedUntil(states, nowMs) + const parkedResumeAt = earliestParkedUntil(states, nowMs) + const resumeAt = + parkedResumeAt && earliestBudgetResetAt + ? new Date(Math.min(parkedResumeAt.getTime(), earliestBudgetResetAt.getTime())) + : (parkedResumeAt ?? earliestBudgetResetAt) if (resumeAt) { throw new RateLimitError('token pool exhausted', { resumeAt }) } throw new ProviderAuthError('token pool empty') }, + async hasHeadroom(estimate: number): Promise { + const probe = options?.probeBudget + if (!probe) { + return true + } + const nowMs = Date.now() + const states = await readStates() + if (states.size === 0) { + return true + } + let pooledRemaining = 0 + for (const [id, state] of states.entries()) { + if (!isHealthy(state, nowMs)) { + continue + } + const bucket = await loadBucket(probe, id, nowMs) + if (!bucket) { + return true + } + pooledRemaining += bucket.remaining + if (pooledRemaining >= estimate) { + return true + } + } + return false + }, + async park(tokenId: string, resumeAt: Date): Promise { await updateState(tokenId, { parkedUntil: resumeAt.toISOString() }) },