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..5db023d983 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,21 @@ export function releasesToQueue( })) } +// One-time catch-up for templates mapped before the release queue existed: same floor rule as a +// normal release, lightest-first ordering left to the caller. +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..0776542d74 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,32 @@ export async function saveCredlyTemplateReleaseCursor( ) } +export interface IUnreleasedMappedCredlyTemplate { + badgeTemplateId: string + badgeCount: number | null +} + +// Same "live and routable" predicate as listActiveCredlyTemplateMappings, so an unroutable +// mapping can't burn its one release; lightest first drains ahead of heavy ones in the FIFO queue. +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,