diff --git a/backend/src/database/migrations/V1793617305__credly-template-releases.sql b/backend/src/database/migrations/V1793617305__credly-template-releases.sql new file mode 100644 index 0000000000..826b50b536 --- /dev/null +++ b/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; diff --git a/services/libs/connectors/src/connectors/credly/apiClient.test.ts b/services/libs/connectors/src/connectors/credly/apiClient.test.ts index d372f6bc36..67ad37c056 100644 --- a/services/libs/connectors/src/connectors/credly/apiClient.test.ts +++ b/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, diff --git a/services/libs/connectors/src/connectors/credly/apiClient.ts b/services/libs/connectors/src/connectors/credly/apiClient.ts index 5a955447f8..d9d12d5b0d 100644 --- a/services/libs/connectors/src/connectors/credly/apiClient.ts +++ b/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, diff --git a/services/libs/data-access-layer/src/credly/releases.test.ts b/services/libs/data-access-layer/src/credly/releases.test.ts new file mode 100644 index 0000000000..bb73d01ec1 --- /dev/null +++ b/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 { + 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() + }) +}) diff --git a/services/libs/data-access-layer/src/credly/releases.ts b/services/libs/data-access-layer/src/credly/releases.ts new file mode 100644 index 0000000000..d05950b64a --- /dev/null +++ b/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 { + 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 { + 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, + cursor: ICredlyTemplateReleaseCursor, +): Promise { + 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, +): Promise { + 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, + ) +}