From ab3b5005a10dda508d8ea691ed5081e5ecfd3b1f Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Mon, 2 Nov 2026 18:44:19 +0100 Subject: [PATCH 1/2] feat: queue release catch-up for already-mapped credly templates The 2026-11-02 go-live mapped 216 templates before the release queue (#5166/#5167) existed, so none of them have a row in credlyTemplateReleases and their pre-go-live history is never recovered by the live poll. Adds --release-unreleased --integration-id to the resolve script: lists every mapped template missing a release and queues them all, lightest by badgeCount first, so the long tail drains ahead of the dozen high-volume ones in the single-release-at-a-time FIFO walker. Refs linuxfoundation/lfx-cdp-insights-ops#975 Signed-off-by: Umberto Sgueglia --- docs/knowledge/sys-cdp.md | 24 +++---- .../src/bin/resolve-credly-templates.ts | 71 ++++++++++++++++--- .../src/credlyTemplateResolve.test.ts | 40 +++++++++++ .../src/credlyTemplateResolve.ts | 21 +++++- .../src/credly/releases.test.ts | 57 +++++++++++++++ .../data-access-layer/src/credly/releases.ts | 19 +++++ 6 files changed, 208 insertions(+), 24 deletions(-) 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..3d12cedaa0 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,13 @@ 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 { findOrCreateCredlyIntegration, upsertCredlyTemplateDecision } from './index' import { completeCredlyTemplateRelease, + listMappedCredlyTemplatesWithoutRelease, nextPendingCredlyTemplateRelease, queueCredlyTemplateRelease, saveCredlyTemplateReleaseCursor, @@ -19,6 +22,30 @@ async function emptyQueue(qx: QueryExecutor): Promise { await qx.result(`DELETE FROM integration."credlyTemplateReleases"`) } +async function mappedTemplate( + qx: QueryExecutor, + badgeTemplateId: string, + badgeCount: number | null, +): Promise { + 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, + }) +} + async function queue(qx: QueryExecutor, badgeTemplateId: string, floor = FLOOR, top = TOP) { await emptyQueue(qx) await queueCredlyTemplateRelease(qx, { badgeTemplateId, floor, top }) @@ -94,3 +121,33 @@ 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([]) + }) +}) diff --git a/services/libs/data-access-layer/src/credly/releases.ts b/services/libs/data-access-layer/src/credly/releases.ts index d05950b64a..6833ad7b8b 100644 --- a/services/libs/data-access-layer/src/credly/releases.ts +++ b/services/libs/data-access-layer/src/credly/releases.ts @@ -65,6 +65,25 @@ export async function saveCredlyTemplateReleaseCursor( ) } +export interface IUnreleasedMappedCredlyTemplate { + badgeTemplateId: string + badgeCount: number | null +} + +// 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 + LEFT JOIN integration."credlyTemplateReleases" r ON r."badgeTemplateId" = m."badgeTemplateId" + WHERE m.status = 'mapped' AND r."badgeTemplateId" IS NULL + ORDER BY m."badgeCount" ASC NULLS FIRST, m."badgeTemplateId"`, + ) +} + export async function completeCredlyTemplateRelease( qx: QueryExecutor, release: Pick, From a39f4361650fb279dc7e9b785a259c9612a37a5e Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Mon, 2 Nov 2026 19:19:12 +0100 Subject: [PATCH 2/2] fix: exclude inactive or unroutable mappings from release catch-up listMappedCredlyTemplatesWithoutRelease only checked status = 'mapped', unlike listActiveCredlyTemplateMappings. A deactivated mapping, or one whose integration was deleted, got queued anyway; the release walker finds it unmapped in routing and completes it with nothing fetched, burning the one release a future re-mapping would have queued. Applies the same live-and-routable predicate used elsewhere: not deactivated, integration not deleted, segment matches. Found independently by Copilot and Cursor Bugbot on PR review. Refs linuxfoundation/lfx-cdp-insights-ops#975 Signed-off-by: Umberto Sgueglia --- .../src/credly/releases.test.ts | 30 +++++++++++++++++-- .../data-access-layer/src/credly/releases.ts | 14 ++++++++- 2 files changed, 41 insertions(+), 3 deletions(-) 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 3d12cedaa0..4b7e595d33 100644 --- a/services/libs/data-access-layer/src/credly/releases.test.ts +++ b/services/libs/data-access-layer/src/credly/releases.test.ts @@ -4,7 +4,11 @@ import { DEFAULT_TENANT_ID, generateUUIDv1 } from '@crowd/common' import type { QueryExecutor } from '@crowd/database' import { withQx } from '@crowd/test-kit/db' -import { findOrCreateCredlyIntegration, upsertCredlyTemplateDecision } from './index' +import { + deactivateCredlyTemplateMapping, + findOrCreateCredlyIntegration, + upsertCredlyTemplateDecision, +} from './index' import { completeCredlyTemplateRelease, listMappedCredlyTemplatesWithoutRelease, @@ -26,7 +30,7 @@ async function mappedTemplate( qx: QueryExecutor, badgeTemplateId: string, badgeCount: number | null, -): Promise { +): Promise<{ integrationId: string }> { const segmentId = generateUUIDv1() await qx.result( `INSERT INTO public.segments (id, slug, name, "tenantId", "createdAt", "updatedAt") @@ -44,6 +48,7 @@ async function mappedTemplate( rule: 'exact_sku', ruleVersion: 1, }) + return { integrationId } } async function queue(qx: QueryExecutor, badgeTemplateId: string, floor = FLOOR, top = TOP) { @@ -150,4 +155,25 @@ describe('listMappedCredlyTemplatesWithoutRelease', () => { 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 6833ad7b8b..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 { @@ -70,6 +72,9 @@ export interface IUnreleasedMappedCredlyTemplate { 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( @@ -78,9 +83,16 @@ export async function listMappedCredlyTemplatesWithoutRelease( 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 r."badgeTemplateId" IS NULL + 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 }, ) }