feat: drive the credly badge poll from the badge state table

backend/src/database/migrations/V1793181595__credly-badges-template.sql

@@ -0,0 +1,6 @@
+ALTER TABLE integration."credlyBadges"
+    ADD COLUMN "badgeTemplateName" TEXT NOT NULL DEFAULT '',
+    ADD COLUMN "badgeTemplateUrl"  TEXT NULL;
+
+ALTER TABLE integration."credlyBadges"
+    ALTER COLUMN "badgeTemplateName" DROP DEFAULT;

services/apps/connectors_worker/src/bin/dry-run-credly-badge-poll.ts

@@ -121,6 +121,8 @@ async function main(): Promise<void> {
   const deps: CredlyDeps = {
     loadTemplateRouting: async () => options.routing,
     lockTemplate: async () => true,
+    findBadges: async () => [],
+    saveBadges: async () => undefined,
   }
 
   const outcome = await runBadgePoll({

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

@@ -5,6 +5,7 @@ import {
   listStaleCredlyTemplateMappings,
   lockCredlyTemplateMapping,
 } from '@crowd/data-access-layer/src/credly'
+import { findCredlyBadges, upsertCredlyBadges } from '@crowd/data-access-layer/src/credly/badges'
 import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
 
 import { createCredlyDeps } from './deps'
@@ -14,6 +15,10 @@ vi.mock('@crowd/data-access-layer/src/credly', () => ({
   listStaleCredlyTemplateMappings: vi.fn(),
   lockCredlyTemplateMapping: vi.fn(),
 }))
+vi.mock('@crowd/data-access-layer/src/credly/badges', () => ({
+  findCredlyBadges: vi.fn(),
+  upsertCredlyBadges: vi.fn(),
+}))
 
 const mockedActive = vi.mocked(listActiveCredlyTemplateMappings)
 const mockedStale = vi.mocked(listStaleCredlyTemplateMappings)
@@ -91,4 +96,15 @@ describe('createCredlyDeps', () => {
     expect(await deps.lockTemplate('tpl-cka', target)).toBe(true)
     expect(lockCredlyTemplateMapping).toHaveBeenCalledWith(qx, 'tpl-cka', target)
   })
+
+  it('reads and writes the badge state through the table', async () => {
+    vi.mocked(findCredlyBadges).mockResolvedValueOnce([])
+    const deps = createCredlyDeps({ qx, integrationId: 'anchor' })
+
+    expect(await deps.findBadges(['b-1'])).toEqual([])
+    await deps.saveBadges([])
+
+    expect(findCredlyBadges).toHaveBeenCalledWith(qx, ['b-1'])
+    expect(upsertCredlyBadges).toHaveBeenCalledWith(qx, [])
+  })
 })

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

@@ -3,6 +3,8 @@ import {
   listStaleCredlyTemplateMappings,
   lockCredlyTemplateMapping,
 } from '@crowd/data-access-layer/src/credly'
+import type { ICredlyBadgeInput, ICredlyBadgeRow } from '@crowd/data-access-layer/src/credly/badges'
+import { findCredlyBadges, upsertCredlyBadges } from '@crowd/data-access-layer/src/credly/badges'
 
 import type { DepsFactoryInput } from '../../types'
 
@@ -20,6 +22,8 @@ export interface CredlyTemplateRouting {
 export interface CredlyDeps {
   loadTemplateRouting(): Promise<CredlyTemplateRouting>
   lockTemplate(badgeTemplateId: string, target: CredlyTemplateTarget): Promise<boolean>
+  findBadges(badgeIds: string[]): Promise<ICredlyBadgeRow[]>
+  saveBadges(badges: ICredlyBadgeInput[]): Promise<void>
 }
 
 export function createCredlyDeps({ qx }: DepsFactoryInput): CredlyDeps {
@@ -44,5 +48,11 @@ export function createCredlyDeps({ qx }: DepsFactoryInput): CredlyDeps {
     lockTemplate(badgeTemplateId, target) {
       return lockCredlyTemplateMapping(qx, badgeTemplateId, target)
     },
+    findBadges(badgeIds) {
+      return findCredlyBadges(qx, badgeIds)
+    },
+    saveBadges(badges) {
+      return upsertCredlyBadges(qx, badges)
+    },
   }
 }

services/libs/connectors/src/connectors/credly/syncs/badgePoll.test.ts

@@ -1,5 +1,6 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
+import type { ICredlyBadgeInput, ICredlyBadgeRow } from '@crowd/data-access-layer/src/credly/badges'
 import { CredlyActivityType } from '@crowd/integrations/src/integrations/credly/types'
 import type { Logger } from '@crowd/logging'
 
@@ -65,18 +66,47 @@ const DEFAULT_ROUTING: CredlyTemplateRouting = {
   staleTemplateIds: [],
 }
 
+function storedBadge(overrides: Partial<ICredlyBadgeRow> = {}): ICredlyBadgeRow {
+  return {
+    badgeId: 'badge-1',
+    badgeTemplateId: 'template-1',
+    badgeTemplateName: 'CKA: Certified Kubernetes Administrator',
+    badgeTemplateUrl: null,
+    state: 'accepted',
+    stateUpdatedAt: '2026-10-22T10:00:00.000Z',
+    issuedAt: '2026-10-22T00:00:00.000Z',
+    expiresAt: null,
+    replacementBadgeId: null,
+    replacesBadgeId: null,
+    segmentId: 'seg-stored',
+    integrationId: 'int-stored',
+    recipientEmail: 'maria.rossi@example.org',
+    issuedTo: 'Maria Rossi',
+    expiredEmittedAt: null,
+    firstSeenAt: '2026-10-22T12:00:00.000Z',
+    lastSeenAt: '2026-10-22T12:00:00.000Z',
+    ...overrides,
+  }
+}
+
 function makeCtx(
   watermark: Record<string, unknown> | null = null,
   hasRunBudget = () => true,
   routing: CredlyTemplateRouting = DEFAULT_ROUTING,
   emitEnabled = true,
+  stored: ICredlyBadgeRow[] = [],
 ) {
   const emitted: unknown[][] = []
   const emits: { records: unknown[]; segmentId?: string; integrationId?: string }[] = []
   const watermarks: Record<string, unknown>[] = []
+  const saved: ICredlyBadgeInput[] = []
   const deps: CredlyDeps = {
     loadTemplateRouting: vi.fn(async () => routing),
     lockTemplate: vi.fn(async () => true),
+    findBadges: vi.fn(async (ids: string[]) => stored.filter((row) => ids.includes(row.badgeId))),
+    saveBadges: vi.fn(async (rows: ICredlyBadgeInput[]) => {
+      saved.push(...rows)
+    }),
   }
   const ctx: SyncContext<CredlyDeps> = {
     channel: { channelId: ORGANIZATION_ID, channelName: ORGANIZATION_ID },
@@ -101,6 +131,7 @@ function makeCtx(
     emitted,
     emits,
     deps,
+    saved,
     watermarks,
     lastWatermark: () => watermarks[watermarks.length - 1],
   }
@@ -346,29 +377,223 @@ describe('runBadgePoll', () => {
     })
   })
 
-  it('emits expired-badge only for accepted badges whose expiry has already passed', async () => {
+  it('stores a new badge with the routing target and leaves expiries to the expiry sync', async () => {
     mockedFetchBadgesPage.mockResolvedValueOnce(
       page([
         badge({ id: 'expired', expires_at: '2026-10-01T00:00:00.000Z' }),
-        badge({ id: 'live', expires_at: '2028-10-01T00:00:00.000Z' }),
         badge({ id: 'pending', state: 'pending', state_updated_at: null }),
         badge({ id: 'revoked', state: 'revoked', expires_at: '2026-10-01T00:00:00.000Z' }),
       ]),
     )
-    const { ctx, emitted } = makeCtx()
+    const { ctx, emitted, saved, deps } = makeCtx()
 
     await runBadgePoll(ctx)
 
     expect(types(emitted[0])).toEqual([
       CredlyActivityType.ISSUED_BADGE,
       CredlyActivityType.ACCEPTED_BADGE,
-      CredlyActivityType.EXPIRED_BADGE,
+      CredlyActivityType.ISSUED_BADGE,
+      CredlyActivityType.REVOKED_BADGE,
+    ])
+    expect(deps.findBadges).toHaveBeenCalledWith(['expired', 'pending', 'revoked'])
+    expect(saved.map((row) => [row.badgeId, row.state, row.segmentId, row.expiresAt])).toEqual([
+      ['expired', 'accepted', 'seg-default', '2026-10-01T00:00:00.000Z'],
+      ['pending', 'pending', 'seg-default', null],
+      ['revoked', 'revoked', 'seg-default', '2026-10-01T00:00:00.000Z'],
+    ])
+    expect(saved[0]).toMatchObject({
+      badgeTemplateName: 'CKA: Certified Kubernetes Administrator',
+      recipientEmail: 'maria.rossi@example.org',
+      issuedTo: 'Maria Rossi',
+    })
+    expect(LOG.info).toHaveBeenCalledWith(
+      expect.objectContaining({ routed: 2, pendingStored: 1, stateChanges: 0, unchanged: 0 }),
+      expect.any(String),
+    )
+  })
+
+  it('emits nothing for a known badge whose state did not change, but marks it seen', async () => {
+    mockedFetchBadgesPage.mockResolvedValueOnce(
+      page([badge({ state_updated_at: '2026-10-27T10:00:00.000Z' })]),
+    )
+    const { ctx, emits, saved, deps } = makeCtx(null, () => true, DEFAULT_ROUTING, true, [
+      storedBadge(),
+    ])
+
+    await runBadgePoll(ctx)
+
+    expect(emits).toEqual([])
+    expect(deps.lockTemplate).toHaveBeenCalledWith('template-1', {
+      segmentId: 'seg-stored',
+      integrationId: 'int-stored',
+      locked: true,
+    })
+    expect(saved.map((row) => [row.badgeId, row.segmentId, row.stateUpdatedAt])).toEqual([
+      ['badge-1', 'seg-stored', '2026-10-27T10:00:00.000Z'],
+    ])
+    expect(LOG.info).toHaveBeenCalledWith(
+      expect.objectContaining({ unchanged: 1, routed: 0 }),
+      expect.any(String),
+    )
+  })
+
+  it('emits only the state activity on the stored segment when a known badge changes state', async () => {
+    mockedFetchBadgesPage.mockResolvedValueOnce(
+      page([
+        badge({
+          state: 'replaced',
+          state_updated_at: '2026-10-27T10:00:00.000Z',
+          replacement_badge_id: 'badge-2',
+        }),
+        badge({ id: 'badge-2', replaces_badge_id: 'badge-1' }),
+      ]),
+    )
+    const { ctx, emits, saved, deps } = makeCtx(null, () => true, DEFAULT_ROUTING, true, [
+      storedBadge(),
+    ])
+
+    await runBadgePoll(ctx)
+
+    expect(
+      emits.map(({ records, segmentId, integrationId }) => ({
+        types: types(records),
+        segmentId,
+        integrationId,
+      })),
+    ).toEqual([
+      {
+        types: [CredlyActivityType.REPLACED_BADGE],
+        segmentId: 'seg-stored',
+        integrationId: 'int-stored',
+      },
+      {
+        types: [CredlyActivityType.ISSUED_BADGE, CredlyActivityType.ACCEPTED_BADGE],
+        segmentId: 'seg-default',
+        integrationId: 'int-default',
+      },
+    ])
+    expect(deps.lockTemplate).toHaveBeenCalledTimes(1)
+    expect(deps.lockTemplate).toHaveBeenCalledWith('template-1', {
+      segmentId: 'seg-stored',
+      integrationId: 'int-stored',
+      locked: true,
+    })
+    expect(
+      saved.map((row) => [row.badgeId, row.state, row.segmentId, row.replacementBadgeId]),
+    ).toEqual([
+      ['badge-1', 'replaced', 'seg-stored', 'badge-2'],
+      ['badge-2', 'accepted', 'seg-default', null],
+    ])
+    expect(LOG.info).toHaveBeenCalledWith(
+      expect.objectContaining({ stateChanges: 1, routed: 1 }),
+      expect.any(String),
+    )
+  })
+
+  it('emits issued and accepted for a stored pending badge once it is accepted', async () => {
+    mockedFetchBadgesPage.mockResolvedValueOnce(page([badge()]))
+    const { ctx, emitted, emits } = makeCtx(null, () => true, DEFAULT_ROUTING, true, [
+      storedBadge({ state: 'pending', stateUpdatedAt: null }),
+    ])
+
+    await runBadgePoll(ctx)
+
+    expect(emitted).toHaveLength(1)
+    expect(types(emitted[0])).toEqual([
       CredlyActivityType.ISSUED_BADGE,
       CredlyActivityType.ACCEPTED_BADGE,
+    ])
+    expect(emits[0]).toMatchObject({ segmentId: 'seg-stored', integrationId: 'int-stored' })
+  })
+
+  it('locks the template of a new pending badge even though nothing is emitted yet', async () => {
+    mockedFetchBadgesPage.mockResolvedValueOnce(
+      page([badge({ state: 'pending', state_updated_at: null })]),
+    )
+    const { ctx, emits, saved, deps } = makeCtx()
+
+    await runBadgePoll(ctx)
+
+    expect(emits).toEqual([])
+    expect(deps.lockTemplate).toHaveBeenCalledWith('template-1', DEFAULT_TARGET)
+    expect(saved.map((row) => [row.badgeId, row.state])).toEqual([['badge-1', 'pending']])
+  })
+
+  it('builds a state activity with the member of the first sight, not the live one', async () => {
+    mockedFetchBadgesPage.mockResolvedValueOnce(
+      page([
+        badge({
+          state: 'revoked',
+          state_updated_at: '2026-10-27T10:00:00.000Z',
+          recipient_email: 'new.address@example.org',
+          issued_to: 'M. Rossi',
+        }),
+      ]),
+    )
+    const { ctx, emitted, saved } = makeCtx(null, () => true, DEFAULT_ROUTING, true, [
+      storedBadge({ recipientEmail: 'maria.rossi@example.org', issuedTo: 'Maria Rossi' }),
+    ])
+
+    await runBadgePoll(ctx)
+
+    const [activity] = emitted[0] as { member: { identities: { value: string }[] } }[]
+    expect(activity.member.identities.map((identity) => identity.value)).toEqual([
+      'maria.rossi@example.org',
+      'maria.rossi@example.org',
+    ])
+    expect(saved[0]).toMatchObject({ recipientEmail: 'maria.rossi@example.org', state: 'revoked' })
+  })
+
+  it('skips a badge whose stored dates do not parse and keeps the rest of the page', async () => {
+    mockedFetchBadgesPage.mockResolvedValueOnce(
+      page([
+        badge({ id: 'bad-expiry', expires_at: 'not a date' }),
+        badge({ id: 'bad-pending', state: 'pending', state_updated_at: null, issued_at: '' }),
+        badge({ id: 'good' }),
+      ]),
+    )
+    const { ctx, emitted, saved } = makeCtx()
+
+    await runBadgePoll(ctx)
+
+    expect(emitted).toHaveLength(1)
+    expect(saved.map((row) => row.badgeId)).toEqual(['good'])
+    expect(LOG.info).toHaveBeenCalledWith(
+      expect.objectContaining({ skipped: 2, routed: 1 }),
+      expect.any(String),
+    )
+  })
+
+  it('emits a badge listed twice on one page only once, with its last version', async () => {
+    mockedFetchBadgesPage.mockResolvedValueOnce(
+      page([badge({ state: 'pending', state_updated_at: null }), badge()]),
+    )
+    const { ctx, emitted, saved } = makeCtx()
+
+    await runBadgePoll(ctx)
+
+    expect(emitted).toHaveLength(1)
+    expect(types(emitted[0])).toEqual([
       CredlyActivityType.ISSUED_BADGE,
+      CredlyActivityType.ACCEPTED_BADGE,
+    ])
+    expect(saved.map((row) => [row.badgeId, row.state])).toEqual([['badge-1', 'accepted']])
+  })
+
+  it('neither reads nor writes the state table during a shadow run', async () => {
+    mockedFetchBadgesPage.mockResolvedValueOnce(page([badge()]))
+    const { ctx, emitted, deps } = makeCtx(null, () => true, DEFAULT_ROUTING, false, [
+      storedBadge(),
+    ])
+
+    await runBadgePoll(ctx)
+
+    expect(types(emitted[0])).toEqual([
       CredlyActivityType.ISSUED_BADGE,
-      CredlyActivityType.REVOKED_BADGE,
+      CredlyActivityType.ACCEPTED_BADGE,
     ])
+    expect(deps.findBadges).not.toHaveBeenCalled()
+    expect(deps.saveBadges).not.toHaveBeenCalled()
   })
 
   it('skips a badge the mapper cannot process and keeps the rest of the page', async () => {