feat: queue release catch-up for already-mapped credly templates

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
 

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,18 +32,28 @@ import {
   loadCredlyTemplates,
   loadSubprojectSegments,
   releasesToQueue,
+  unreleasedTemplatesToQueue,
 } from '../credlyTemplateResolve'
 
 const log = getServiceLogger()
 const HELD_QUEUE_PRINTED = 20
 
 function usage(): never {
   log.error(
-    'Usage: resolve-credly-templates --integration-id <uuid> [--apply [--notify | --digest]] [--out <file>] | --release <badge template id>',
+    'Usage: resolve-credly-templates --integration-id <uuid> [--apply [--notify | --digest]] [--out <file>] | --release <badge template id> | --release-unreleased --integration-id <uuid>',
   )
   process.exit(1)
 }
 
+function credlyHttpClient(token: string): ReturnType<typeof createHttpClient> {
+  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<void> {
-  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<void> {
     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 {

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> = {}): 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' },
+    ])
+  })
+})

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(

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<void> {
   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([])
+  })
+})