diff --git a/docs/knowledge/sys-cdp.md b/docs/knowledge/sys-cdp.md index ace1a7f27e..fa453e5e95 100644 --- a/docs/knowledge/sys-cdp.md +++ b/docs/knowledge/sys-cdp.md @@ -26,18 +26,18 @@ split them into a directory (`packages_worker`, `pcc_sync_worker`, `snowflake_co How source data gets in. [`data-flow.md`](data-flow.md) shows how these connect. -| Service | Kind | Role | -| --------------------------- | -------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -| `webhook_api` | HTTP | Receives platform webhooks and hands them to `integration_stream_worker` | -| `integration_run_worker` | Kafka consumer | Runs an integration sync and splits it into streams | -| `integration_stream_worker` | Kafka consumer | Fetches and processes each stream, emits results to `data_sink_worker` | -| `data_sink_worker` | Kafka consumer | Resolves members and organizations, writes `activityRelations` to Postgres, sends activity content to the `activities` Kafka topic for Tinybird, and queues an OpenSearch sync | -| `connectors_worker` | Temporal | Connector framework v2 runtime: dispatches and runs syncs for the connectors registered in `src/main.ts`, currently GitHub and Octolens, plus a dummy connector in development. `services/libs/connectors` also holds Credly code that is not registered yet | -| `nango_worker` | Temporal | GitHub sync through Nango: repo syncs, webhook processing, connection cleanup | -| `nango_webhook_api` | HTTP | Receives Nango webhooks and starts `nango_worker` workflows | -| `snowflake_connectors` | Temporal | Exports source data from Snowflake through S3, transforms it, emits to `data_sink_worker` | -| `git_integration` | Python | Clones repositories, extracts commits and maintainers, sends them to Kafka | -| `mailing_list_integration` | Python | Ingests mailing list archives (`crowdmail`) | +| Service | Kind | Role | +| --------------------------- | -------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `webhook_api` | HTTP | Receives platform webhooks and hands them to `integration_stream_worker` | +| `integration_run_worker` | Kafka consumer | Runs an integration sync and splits it into streams | +| `integration_stream_worker` | Kafka consumer | Fetches and processes each stream, emits results to `data_sink_worker` | +| `data_sink_worker` | Kafka consumer | Resolves members and organizations, writes `activityRelations` to Postgres, sends activity content to the `activities` Kafka topic for Tinybird, and queues an OpenSearch sync | +| `connectors_worker` | Temporal | Connector framework v2 runtime: dispatches and runs syncs for the connectors registered in `src/main.ts` — GitHub, Octolens and Credly (live in prod since 2026-11-02), plus a dummy connector in development | +| `nango_worker` | Temporal | GitHub sync through Nango: repo syncs, webhook processing, connection cleanup | +| `nango_webhook_api` | HTTP | Receives Nango webhooks and starts `nango_worker` workflows | +| `snowflake_connectors` | Temporal | Exports source data from Snowflake through S3, transforms it, emits to `data_sink_worker` | +| `git_integration` | Python | Clones repositories, extracts commits and maintainers, sends them to Kafka | +| `mailing_list_integration` | Python | Ingests mailing list archives (`crowdmail`) | ## Members, organizations and identity diff --git a/services/apps/connectors_worker/src/bin/resolve-credly-templates.ts b/services/apps/connectors_worker/src/bin/resolve-credly-templates.ts index a17e3ae9e1..15fc104451 100644 --- a/services/apps/connectors_worker/src/bin/resolve-credly-templates.ts +++ b/services/apps/connectors_worker/src/bin/resolve-credly-templates.ts @@ -14,7 +14,10 @@ import { listCredlyTemplateMappings, listHeldCredlyTemplateMappings, } from '@crowd/data-access-layer/src/credly' -import { queueCredlyTemplateRelease } from '@crowd/data-access-layer/src/credly/releases' +import { + listMappedCredlyTemplatesWithoutRelease, + queueCredlyTemplateRelease, +} from '@crowd/data-access-layer/src/credly/releases' import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database' import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor' import { getServiceLogger } from '@crowd/logging' @@ -29,6 +32,7 @@ import { loadCredlyTemplates, loadSubprojectSegments, releasesToQueue, + unreleasedTemplatesToQueue, } from '../credlyTemplateResolve' const log = getServiceLogger() @@ -36,11 +40,20 @@ const HELD_QUEUE_PRINTED = 20 function usage(): never { log.error( - 'Usage: resolve-credly-templates --integration-id [--apply [--notify | --digest]] [--out ] | --release ', + 'Usage: resolve-credly-templates --integration-id [--apply [--notify | --digest]] [--out ] | --release | --release-unreleased --integration-id ', ) process.exit(1) } +function credlyHttpClient(token: string): ReturnType { + return createHttpClient({ + acquireToken: async () => ({ id: CREDLY_POOL_ENTRY_ID, value: token }), + parkToken: async () => undefined, + invalidateToken: async () => undefined, + log, + }) +} + function takeFlag(argv: string[], flag: string): string | undefined { const flagIndex = argv.indexOf(flag) if (flagIndex === -1) return undefined @@ -59,6 +72,7 @@ function takeSwitch(argv: string[], flag: string): boolean { interface Args { release?: string + releaseUnreleased: boolean integrationId: string apply: boolean notify: 'changes' | 'digest' | null @@ -70,7 +84,20 @@ function parseArgs(rawArgv: string[]): Args { const release = takeFlag(argv, '--release') if (release) { if (argv.length > 0) usage() - return { release, integrationId: '', apply: false, notify: null, outFile: '' } + return { + release, + releaseUnreleased: false, + integrationId: '', + apply: false, + notify: null, + outFile: '', + } + } + const releaseUnreleased = takeSwitch(argv, '--release-unreleased') + if (releaseUnreleased) { + const integrationId = takeFlag(argv, '--integration-id') + if (!integrationId || argv.length > 0) usage() + return { releaseUnreleased, integrationId, apply: false, notify: null, outFile: '' } } const integrationId = takeFlag(argv, '--integration-id') const apply = takeSwitch(argv, '--apply') @@ -81,11 +108,13 @@ function parseArgs(rawArgv: string[]): Args { takeFlag(argv, '--out') ?? join(tmpdir(), `credly-resolve-${new Date().toISOString().replace(/[:.]/g, '-')}.json`) if (!integrationId || argv.length > 0) usage() - return { integrationId, apply, notify, outFile } + return { releaseUnreleased: false, integrationId, apply, notify, outFile } } async function main(): Promise { - const { release, integrationId, apply, notify, outFile } = parseArgs(process.argv.slice(2)) + const { release, releaseUnreleased, integrationId, apply, notify, outFile } = parseArgs( + process.argv.slice(2), + ) const db = await getDbConnection(WRITE_DB_CONFIG()) const qx = pgpQx(db) @@ -107,15 +136,35 @@ async function main(): Promise { return } + if (releaseUnreleased) { + const credential = await getCredential(qx, integrationId) + assertCredentialKind(credential, 'credly-oauth-client') + const { token } = await mintCredlyAccessToken(credential) + const http = credlyHttpClient(token) + const [listing, unreleased] = await Promise.all([ + loadCredlyTemplates(http, credential.data.organizationId, log), + listMappedCredlyTemplatesWithoutRelease(qx), + ]) + if (!listing.complete) { + log.warn( + 'the credly listing was not complete; templates missing from it fall back to the full history floor', + ) + } + const releases = unreleasedTemplatesToQueue(listing.templates, unreleased, new Date()) + for (const queued of releases) { + await queueCredlyTemplateRelease(qx, queued) + } + log.info( + { queued: releases.length, lightest: releases[0], heaviest: releases[releases.length - 1] }, + 'credly mapped templates without a release were queued, lightest first', + ) + return + } + const credential = await getCredential(qx, integrationId) assertCredentialKind(credential, 'credly-oauth-client') const { token } = await mintCredlyAccessToken(credential) - const http = createHttpClient({ - acquireToken: async () => ({ id: CREDLY_POOL_ENTRY_ID, value: token }), - parkToken: async () => undefined, - invalidateToken: async () => undefined, - log, - }) + const http = credlyHttpClient(token) const snowflake = SnowflakeClient.fromEnv({ parentLog: log, maxConnections: 3 }) try { diff --git a/services/apps/connectors_worker/src/credlyTemplateResolve.test.ts b/services/apps/connectors_worker/src/credlyTemplateResolve.test.ts index 0be27b0db1..3761691c55 100644 --- a/services/apps/connectors_worker/src/credlyTemplateResolve.test.ts +++ b/services/apps/connectors_worker/src/credlyTemplateResolve.test.ts @@ -11,6 +11,7 @@ import { loadCredlyTemplates, loadSubprojectSegments, releasesToQueue, + unreleasedTemplatesToQueue, } from './credlyTemplateResolve' function decision(id: string, overrides: Partial = {}): ResolvedTemplate { @@ -319,3 +320,42 @@ describe('releasesToQueue', () => { expect(releases[0].floor).toBe(CREDLY_HISTORY_START) }) }) + +describe('unreleasedTemplatesToQueue', () => { + const now = new Date('2026-11-02T12:00:00.000Z') + + it('queues one release per unreleased template, floored at its own creation date', () => { + const releases = unreleasedTemplatesToQueue( + [ + { id: 'ckad', name: 'CKAD', createdAt: '2019-01-01T00:00:00.000Z' }, + { id: 'fresh', name: 'Fresh', createdAt: '2026-11-01T09:00:00.000Z' }, + ], + [ + { badgeTemplateId: 'fresh', badgeCount: 10 }, + { badgeTemplateId: 'ckad', badgeCount: 58312 }, + ], + now, + ) + + expect(releases).toEqual([ + { + badgeTemplateId: 'fresh', + floor: '2026-11-01T09:00:00.000Z', + top: '2026-11-02T12:00:00.000Z', + }, + { badgeTemplateId: 'ckad', floor: CREDLY_HISTORY_START, top: '2026-11-02T12:00:00.000Z' }, + ]) + }) + + it('falls back to the history start for a template missing from the listing', () => { + const releases = unreleasedTemplatesToQueue( + [], + [{ badgeTemplateId: 'gone', badgeCount: 3 }], + now, + ) + + expect(releases).toEqual([ + { badgeTemplateId: 'gone', floor: CREDLY_HISTORY_START, top: '2026-11-02T12:00:00.000Z' }, + ]) + }) +}) diff --git a/services/apps/connectors_worker/src/credlyTemplateResolve.ts b/services/apps/connectors_worker/src/credlyTemplateResolve.ts index 4d57dc4f15..0c6a07e80b 100644 --- a/services/apps/connectors_worker/src/credlyTemplateResolve.ts +++ b/services/apps/connectors_worker/src/credlyTemplateResolve.ts @@ -15,7 +15,10 @@ import { findOrCreateCredlyIntegration, upsertCredlyTemplateDecision, } from '@crowd/data-access-layer/src/credly' -import type { ICredlyTemplateReleaseInput } from '@crowd/data-access-layer/src/credly/releases' +import type { + ICredlyTemplateReleaseInput, + IUnreleasedMappedCredlyTemplate, +} from '@crowd/data-access-layer/src/credly/releases' import { queueCredlyTemplateRelease } from '@crowd/data-access-layer/src/credly/releases' import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' import { listActiveSubprojectSegments } from '@crowd/data-access-layer/src/segments' @@ -219,6 +222,22 @@ export function releasesToQueue( })) } +// One-time catch-up for templates that were mapped before the release queue existed: same floor +// rule as a normal release, ordered lightest-first by the caller so they drain ahead of the few +// high-volume ones a single release walker works through serially. +export function unreleasedTemplatesToQueue( + templates: TemplateToResolve[], + unreleased: IUnreleasedMappedCredlyTemplate[], + now: Date, +): ICredlyTemplateReleaseInput[] { + const createdAtById = new Map(templates.map((template) => [template.id, template.createdAt])) + return unreleased.map((entry) => ({ + badgeTemplateId: entry.badgeTemplateId, + floor: releaseFloor(createdAtById.get(entry.badgeTemplateId), now), + top: now.toISOString(), + })) +} + // A release is queued in the same transaction as the decision that frees the template: a run that // dies in between leaves the template held, and the next run decides it again. export async function applyDecisions( diff --git a/services/libs/data-access-layer/src/credly/releases.test.ts b/services/libs/data-access-layer/src/credly/releases.test.ts index bb73d01ec1..4b7e595d33 100644 --- a/services/libs/data-access-layer/src/credly/releases.test.ts +++ b/services/libs/data-access-layer/src/credly/releases.test.ts @@ -1,10 +1,17 @@ import { test as base, describe, expect } from 'vitest' +import { DEFAULT_TENANT_ID, generateUUIDv1 } from '@crowd/common' import type { QueryExecutor } from '@crowd/database' import { withQx } from '@crowd/test-kit/db' +import { + deactivateCredlyTemplateMapping, + findOrCreateCredlyIntegration, + upsertCredlyTemplateDecision, +} from './index' import { completeCredlyTemplateRelease, + listMappedCredlyTemplatesWithoutRelease, nextPendingCredlyTemplateRelease, queueCredlyTemplateRelease, saveCredlyTemplateReleaseCursor, @@ -19,6 +26,31 @@ async function emptyQueue(qx: QueryExecutor): Promise { await qx.result(`DELETE FROM integration."credlyTemplateReleases"`) } +async function mappedTemplate( + qx: QueryExecutor, + badgeTemplateId: string, + badgeCount: number | null, +): Promise<{ integrationId: string }> { + const segmentId = generateUUIDv1() + await qx.result( + `INSERT INTO public.segments (id, slug, name, "tenantId", "createdAt", "updatedAt") + VALUES ($(id), $(id), $(id), $(tenantId), NOW(), NOW())`, + { id: segmentId, tenantId: DEFAULT_TENANT_ID }, + ) + const { id: integrationId } = await findOrCreateCredlyIntegration(qx, segmentId) + await upsertCredlyTemplateDecision(qx, { + badgeTemplateId, + badgeTemplateName: badgeTemplateId, + badgeCount, + status: 'mapped', + segmentId, + integrationId, + rule: 'exact_sku', + ruleVersion: 1, + }) + return { integrationId } +} + async function queue(qx: QueryExecutor, badgeTemplateId: string, floor = FLOOR, top = TOP) { await emptyQueue(qx) await queueCredlyTemplateRelease(qx, { badgeTemplateId, floor, top }) @@ -94,3 +126,54 @@ describe('credly template releases', () => { ).rejects.toThrow() }) }) + +describe('listMappedCredlyTemplatesWithoutRelease', () => { + test('lists mapped templates missing a release, lightest first', async ({ qx }) => { + await emptyQueue(qx) + await mappedTemplate(qx, 'tpl-heavy', 58312) + await mappedTemplate(qx, 'tpl-light', 10) + await mappedTemplate(qx, 'tpl-no-count', null) + await queueCredlyTemplateRelease(qx, { badgeTemplateId: 'tpl-heavy', floor: FLOOR, top: TOP }) + + const unreleased = await listMappedCredlyTemplatesWithoutRelease(qx) + + expect(unreleased).toEqual([ + { badgeTemplateId: 'tpl-no-count', badgeCount: null }, + { badgeTemplateId: 'tpl-light', badgeCount: 10 }, + ]) + }) + + test('drops a template once its release is queued', async ({ qx }) => { + await emptyQueue(qx) + await mappedTemplate(qx, 'tpl-x', 5) + + expect(await listMappedCredlyTemplatesWithoutRelease(qx)).toEqual([ + { badgeTemplateId: 'tpl-x', badgeCount: 5 }, + ]) + + await queueCredlyTemplateRelease(qx, { badgeTemplateId: 'tpl-x', floor: FLOOR, top: TOP }) + + expect(await listMappedCredlyTemplatesWithoutRelease(qx)).toEqual([]) + }) + + test('excludes a deactivated mapping', async ({ qx }) => { + await emptyQueue(qx) + await mappedTemplate(qx, 'tpl-deactivated', 1) + await deactivateCredlyTemplateMapping(qx, 'tpl-deactivated') + + expect(await listMappedCredlyTemplatesWithoutRelease(qx)).toEqual([]) + }) + + test('excludes a mapping whose integration was deleted', async ({ qx }) => { + await emptyQueue(qx) + const { integrationId } = await mappedTemplate(qx, 'tpl-orphaned', 1) + await qx.result( + `UPDATE public.integrations SET "deletedAt" = now() WHERE id = $(integrationId)`, + { + integrationId, + }, + ) + + expect(await listMappedCredlyTemplatesWithoutRelease(qx)).toEqual([]) + }) +}) diff --git a/services/libs/data-access-layer/src/credly/releases.ts b/services/libs/data-access-layer/src/credly/releases.ts index d05950b64a..53d0f452d7 100644 --- a/services/libs/data-access-layer/src/credly/releases.ts +++ b/services/libs/data-access-layer/src/credly/releases.ts @@ -1,3 +1,5 @@ +import { PlatformType } from '@crowd/types' + import { QueryExecutor } from '../queryExecutor' export interface ICredlyTemplateReleaseInput { @@ -65,6 +67,35 @@ export async function saveCredlyTemplateReleaseCursor( ) } +export interface IUnreleasedMappedCredlyTemplate { + badgeTemplateId: string + badgeCount: number | null +} + +// Same "live and routable" predicate as listActiveCredlyTemplateMappings: an inactive or +// unroutable mapping would have its release completed with nothing fetched (the walker +// finds it unmapped in routing), burning the one release a future re-mapping could queue. +// Lightest templates first: they drain fast and don't sit behind the handful of +// high-volume ones in the FIFO queue a single release walker works through. +export async function listMappedCredlyTemplatesWithoutRelease( + qx: QueryExecutor, +): Promise { + return qx.select( + `SELECT m."badgeTemplateId", m."badgeCount" + FROM integration."credlyTemplateMappings" m + JOIN public.integrations i ON i.id = m."integrationId" + LEFT JOIN integration."credlyTemplateReleases" r ON r."badgeTemplateId" = m."badgeTemplateId" + WHERE m.status = 'mapped' + AND m."deactivatedAt" IS NULL + AND i.platform = $(platform) + AND i."deletedAt" IS NULL + AND i."segmentId" = m."segmentId" + AND r."badgeTemplateId" IS NULL + ORDER BY m."badgeCount" ASC NULLS FIRST, m."badgeTemplateId"`, + { platform: PlatformType.CREDLY }, + ) +} + export async function completeCredlyTemplateRelease( qx: QueryExecutor, release: Pick,