feat: add the credly template release queue and the badge template filter

backend/src/database/migrations/V1793617305__credly-template-releases.sql

@@ -0,0 +1,17 @@
+CREATE TABLE IF NOT EXISTS integration."credlyTemplateReleases" (
+    "badgeTemplateId" TEXT PRIMARY KEY,
+    "floor"           TIMESTAMPTZ NOT NULL,
+    "top"             TIMESTAMPTZ NOT NULL,
+    "since"           TIMESTAMPTZ NULL,
+    "until"           TIMESTAMPTZ NULL,
+    "page"            INTEGER NOT NULL DEFAULT 1,
+    "queuedAt"        TIMESTAMPTZ NOT NULL DEFAULT now(),
+    "completedAt"     TIMESTAMPTZ NULL,
+    "createdAt"       TIMESTAMPTZ NOT NULL DEFAULT now(),
+    "updatedAt"       TIMESTAMPTZ NOT NULL DEFAULT now(),
+    CHECK ("floor" < "top")
+);
+
+CREATE INDEX IF NOT EXISTS "ix_credlyTemplateReleases_pending"
+    ON integration."credlyTemplateReleases" ("queuedAt")
+    WHERE "completedAt" IS NULL;

services/libs/connectors/src/connectors/credly/apiClient.test.ts

@@ -98,7 +98,7 @@ describe('fetchBadgesPage', () => {
     )
   })
 
-  it('adds state and expires_at_max only as extra filters inside the window', async () => {
+  it('adds state, expires_at_max and badge_template_id only as extra filters inside the window', async () => {
     const http = httpReturning(pageOf([]))
 
     await fetchBadgesPage(
@@ -110,6 +110,7 @@ describe('fetchBadgesPage', () => {
         per: 1,
         states: ['accepted', 'replaced'],
         expiresAtMax: '2026-10-22T00:00:00.000Z',
+        badgeTemplateId: 'tpl-lfc131',
       },
       log,
     )
@@ -119,6 +120,7 @@ describe('fetchBadgesPage', () => {
       'filter[state_updated_at_max]': '2026-10-21T00:00:00Z',
       'filter[state]': 'accepted,replaced',
       'filter[expires_at_max]': '2026-10-22T00:00:00.000Z',
+      'filter[badge_template_id]': 'tpl-lfc131',
       sort: 'state_updated_at',
       per: 1,
       page: 1,

services/libs/connectors/src/connectors/credly/apiClient.ts

@@ -28,6 +28,7 @@ export interface BadgesPageParams {
   per?: number
   states?: CredlyBadgeState[]
   expiresAtMax?: string
+  badgeTemplateId?: string
 }
 
 export interface BadgeTemplatesPageParams {
@@ -101,6 +102,7 @@ export async function fetchBadgesPage(
       'filter[state_updated_at_max]': window.max,
       ...(params.states?.length && { 'filter[state]': params.states.join(',') }),
       ...(params.expiresAtMax && { 'filter[expires_at_max]': params.expiresAtMax }),
+      ...(params.badgeTemplateId && { 'filter[badge_template_id]': params.badgeTemplateId }),
       sort: 'state_updated_at',
       per: params.per ?? CREDLY_PAGE_SIZE,
       page: params.page,

services/libs/data-access-layer/src/credly/releases.test.ts

@@ -0,0 +1,96 @@
+import { test as base, describe, expect } from 'vitest'
+
+import type { QueryExecutor } from '@crowd/database'
+import { withQx } from '@crowd/test-kit/db'
+
+import {
+  completeCredlyTemplateRelease,
+  nextPendingCredlyTemplateRelease,
+  queueCredlyTemplateRelease,
+  saveCredlyTemplateReleaseCursor,
+} from './releases'
+
+const test = withQx(base)
+
+const FLOOR = '2022-10-25T00:00:00.000Z'
+const TOP = '2026-11-02T10:00:00.000Z'
+
+async function emptyQueue(qx: QueryExecutor): Promise<void> {
+  await qx.result(`DELETE FROM integration."credlyTemplateReleases"`)
+}
+
+async function queue(qx: QueryExecutor, badgeTemplateId: string, floor = FLOOR, top = TOP) {
+  await emptyQueue(qx)
+  await queueCredlyTemplateRelease(qx, { badgeTemplateId, floor, top })
+  const release = await nextPendingCredlyTemplateRelease(qx)
+  if (release?.badgeTemplateId !== badgeTemplateId) throw new Error('queue is not empty')
+  return release
+}
+
+describe('credly template releases', () => {
+  test('serves pending releases in queue order and skips completed ones', async ({ qx }) => {
+    const first = await queue(qx, 'tpl-first')
+    await queueCredlyTemplateRelease(qx, { badgeTemplateId: 'tpl-second', floor: FLOOR, top: TOP })
+
+    expect((await nextPendingCredlyTemplateRelease(qx))?.badgeTemplateId).toBe('tpl-first')
+    expect(await completeCredlyTemplateRelease(qx, first)).toBe(1)
+    const second = await nextPendingCredlyTemplateRelease(qx)
+    expect(second?.badgeTemplateId).toBe('tpl-second')
+    expect(await completeCredlyTemplateRelease(qx, second)).toBe(1)
+    expect(await nextPendingCredlyTemplateRelease(qx)).toBeNull()
+    expect(await completeCredlyTemplateRelease(qx, second)).toBe(0)
+  })
+
+  test('keeps the window cursor of a release in progress', async ({ qx }) => {
+    const release = await queue(qx, 'tpl')
+
+    expect(
+      await saveCredlyTemplateReleaseCursor(qx, release, {
+        top: '2026-10-26T10:00:00.000Z',
+        since: '2026-10-20T00:00:00.000Z',
+        until: '2026-10-26T10:00:00.000Z',
+        page: 3,
+      }),
+    ).toBe(1)
+
+    const stored = await nextPendingCredlyTemplateRelease(qx)
+    expect(stored).toMatchObject({ badgeTemplateId: 'tpl', page: 3 })
+    expect(new Date(stored.top).toISOString()).toBe('2026-10-26T10:00:00.000Z')
+    expect(new Date(stored.since).toISOString()).toBe('2026-10-20T00:00:00.000Z')
+    expect(new Date(stored.until).toISOString()).toBe('2026-10-26T10:00:00.000Z')
+  })
+
+  test('queuing a template again restarts it, and the walk in flight loses its writes', async ({
+    qx,
+  }) => {
+    const stale = await queue(qx, 'tpl')
+    await completeCredlyTemplateRelease(qx, stale)
+
+    await queueCredlyTemplateRelease(qx, {
+      badgeTemplateId: 'tpl',
+      floor: '2024-01-01T00:00:00.000Z',
+      top: '2026-11-03T10:00:00.000Z',
+    })
+
+    const fresh = await nextPendingCredlyTemplateRelease(qx)
+    expect(fresh).toMatchObject({ badgeTemplateId: 'tpl', page: 1, since: null, until: null })
+    expect(new Date(fresh.floor).toISOString()).toBe('2024-01-01T00:00:00.000Z')
+    expect(fresh.completedAt).toBeNull()
+    expect(
+      await saveCredlyTemplateReleaseCursor(qx, stale, {
+        top: TOP,
+        since: null,
+        until: null,
+        page: 2,
+      }),
+    ).toBe(0)
+    expect(await completeCredlyTemplateRelease(qx, stale)).toBe(0)
+  })
+
+  test('rejects a floor that does not sit below the top', async ({ qx }) => {
+    await emptyQueue(qx)
+    await expect(
+      queueCredlyTemplateRelease(qx, { badgeTemplateId: 'tpl', floor: TOP, top: FLOOR }),
+    ).rejects.toThrow()
+  })
+})

services/libs/data-access-layer/src/credly/releases.ts

@@ -0,0 +1,79 @@
+import { QueryExecutor } from '../queryExecutor'
+
+export interface ICredlyTemplateReleaseInput {
+  badgeTemplateId: string
+  floor: string
+  top: string
+}
+
+export interface ICredlyTemplateReleaseCursor {
+  top: string
+  since: string | null
+  until: string | null
+  page: number
+}
+
+export interface ICredlyTemplateRelease extends ICredlyTemplateReleaseInput {
+  since: string | null
+  until: string | null
+  page: number
+  queuedAt: string
+  completedAt: string | null
+}
+
+const COLUMNS = `"badgeTemplateId", floor, top, since, until, page, "queuedAt", "completedAt"`
+
+export async function queueCredlyTemplateRelease(
+  qx: QueryExecutor,
+  release: ICredlyTemplateReleaseInput,
+): Promise<void> {
+  await qx.result(
+    `INSERT INTO integration."credlyTemplateReleases" ("badgeTemplateId", floor, top)
+     VALUES ($(badgeTemplateId), $(floor), $(top))
+     ON CONFLICT ("badgeTemplateId") DO UPDATE
+       SET floor = EXCLUDED.floor, top = EXCLUDED.top,
+           since = NULL, until = NULL, page = 1,
+           "queuedAt" = now(), "completedAt" = NULL, "updatedAt" = now()`,
+    release,
+  )
+}
+
+export async function nextPendingCredlyTemplateRelease(
+  qx: QueryExecutor,
+): Promise<ICredlyTemplateRelease | null> {
+  return qx.selectOneOrNone(
+    `SELECT ${COLUMNS} FROM integration."credlyTemplateReleases"
+     WHERE "completedAt" IS NULL
+     ORDER BY "queuedAt", "badgeTemplateId"
+     LIMIT 1`,
+  )
+}
+
+// Writes carry the queuedAt they read: a release queued again meanwhile is a new walk, and the
+// old walker must get 0 rows instead of putting its cursor on the new bounds.
+export async function saveCredlyTemplateReleaseCursor(
+  qx: QueryExecutor,
+  release: Pick<ICredlyTemplateRelease, 'badgeTemplateId' | 'queuedAt'>,
+  cursor: ICredlyTemplateReleaseCursor,
+): Promise<number> {
+  return qx.result(
+    `UPDATE integration."credlyTemplateReleases"
+     SET top = $(top), since = $(since), until = $(until), page = $(page), "updatedAt" = now()
+     WHERE "badgeTemplateId" = $(badgeTemplateId) AND "queuedAt" = $(queuedAt)
+       AND "completedAt" IS NULL`,
+    { badgeTemplateId: release.badgeTemplateId, queuedAt: release.queuedAt, ...cursor },
+  )
+}
+
+export async function completeCredlyTemplateRelease(
+  qx: QueryExecutor,
+  release: Pick<ICredlyTemplateRelease, 'badgeTemplateId' | 'queuedAt'>,
+): Promise<number> {
+  return qx.result(
+    `UPDATE integration."credlyTemplateReleases"
+     SET "completedAt" = now(), since = NULL, until = NULL, page = 1, "updatedAt" = now()
+     WHERE "badgeTemplateId" = $(badgeTemplateId) AND "queuedAt" = $(queuedAt)
+       AND "completedAt" IS NULL`,
+    release,
+  )
+}