feat: adding onboarding worker as standard pattern, and onboarding error reason on db

Signed-off-by: Umberto Sgueglia <usgueglia@contractor.linuxfoundation.org>

Author: ulemons

backend/src/database/migrations/V1787916364__onboarding-error-to-project-catalog.sql

@@ -0,0 +1,2 @@
+ALTER TABLE "projectCatalog"
+  ADD COLUMN IF NOT EXISTS "onboardingError" TEXT;

services/apps/automatic_onboarding_worker/src/activities.ts

@@ -1 +1 @@
-export {}
+export * from './activities/activities'

services/apps/automatic_onboarding_worker/src/activities/activities.ts

@@ -0,0 +1,79 @@
+import {
+  findProjectCatalogById,
+  findProjectCatalogPendingOnboarding,
+  updateProjectCatalog,
+} from '@crowd/data-access-layer'
+import { IDbProjectCatalog } from '@crowd/data-access-layer/src/project-catalog/types'
+import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
+import { getServiceLogger } from '@crowd/logging'
+
+import { svc } from '../main'
+import { onboardProject } from '../onboarder/onboarder'
+
+const log = getServiceLogger()
+
+export async function fetchProjectsPendingOnboarding(
+  batchSize: number,
+): Promise<IDbProjectCatalog[]> {
+  const qx = pgpQx(svc.postgres.reader.connection())
+
+  const projects = await findProjectCatalogPendingOnboarding(qx, { limit: batchSize })
+
+  log.info({ count: projects.length, batchSize }, 'Fetched projects pending onboarding.')
+
+  return projects
+}
+
+export async function onboardAndUpdateProject(project: IDbProjectCatalog): Promise<void> {
+  const qx = pgpQx(svc.postgres.writer.connection())
+  const startTime = Date.now()
+
+  // Guard: fetch fresh state to ensure the API is called at most once per project.
+  // Uses the writer connection to avoid replica lag missing a just-written onboardedAt.
+  const fresh = await findProjectCatalogById(qx, project.id)
+  if (fresh?.onboardedAt) {
+    log.info(
+      { id: project.id, repoUrl: project.repoUrl, onboardedAt: fresh.onboardedAt },
+      'Project already onboarded, skipping API call.',
+    )
+    return
+  }
+
+  log.info({ id: project.id, repoUrl: project.repoUrl }, 'Starting onboarding.')
+
+  const result = await onboardProject({
+    id: project.id,
+    repoUrl: project.repoUrl,
+    repoName: project.repoName,
+    projectSlug: project.projectSlug,
+  })
+
+  if (result.outcome === 'error') {
+    throw new Error(result.error ?? 'Unknown onboarding error')
+  }
+
+  await updateProjectCatalog(qx, project.id, {
+    onboardedAt: new Date().toISOString(),
+  })
+
+  const elapsedSeconds = ((Date.now() - startTime) / 1000).toFixed(1)
+
+  log.info(
+    { id: project.id, repoUrl: project.repoUrl, segmentId: result.segmentId, elapsedSeconds },
+    'Onboarding complete.',
+  )
+}
+
+export async function markProjectOnboardingFailed(
+  projectId: string,
+  reason: string,
+): Promise<void> {
+  const qx = pgpQx(svc.postgres.writer.connection())
+
+  await updateProjectCatalog(qx, projectId, {
+    action: 'error',
+    onboardingError: reason,
+  })
+
+  log.error({ id: projectId, reason }, 'Onboarding permanently failed, marked as error.')
+}

services/apps/automatic_onboarding_worker/src/main.ts

@@ -1,6 +1,8 @@
 import { Config } from '@crowd/archetype-standard'
 import { Options, ServiceWorker } from '@crowd/archetype-worker'
 
+import { scheduleProjectsOnboarding } from './schedules/scheduleProjectsOnboarding'
+
 const config: Config = {
   envvars: [
     'CROWD_API_SERVICE_URL',
@@ -34,5 +36,9 @@ setImmediate(async () => {
 
   svc.log.info('Automatic onboarding worker starting up.')
 
+  await scheduleProjectsOnboarding()
+
+  svc.log.info('Automatic onboarding worker running — schedule registered, waiting for Temporal.')
+
   await svc.start()
 })

services/apps/automatic_onboarding_worker/src/schedules/scheduleProjectsOnboarding.ts

@@ -0,0 +1,46 @@
+import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/client'
+
+import { svc } from '../main'
+import { IOnboardProjectsInput, onboardProjects } from '../workflows'
+
+const ONBOARDING_ARGS: IOnboardProjectsInput = {
+  batchSize: 50,
+}
+
+export const scheduleProjectsOnboarding = async () => {
+  svc.log.info('Scheduling projects onboarding')
+
+  try {
+    await svc.temporal.schedule.create({
+      scheduleId: 'projectsOnboarding',
+      spec: {
+        // Daily: catches up on whatever landed in 'onboard' state, independent of the evaluation schedule's timing.
+        cronExpressions: ['0 8 * * *'],
+      },
+      policies: {
+        overlap: ScheduleOverlapPolicy.SKIP,
+        catchupWindow: '1 minute',
+      },
+      action: {
+        type: 'startWorkflow',
+        workflowType: onboardProjects,
+        taskQueue: 'automatic-onboarding',
+        args: [ONBOARDING_ARGS],
+        // 50 projects × up to ~2min per attempt, up to 2 attempts each = ~3.3h worst case; set ceiling with margin.
+        workflowExecutionTimeout: '6 hours',
+        retry: {
+          initialInterval: '30 seconds',
+          backoffCoefficient: 2,
+          maximumAttempts: 3,
+        },
+      },
+    })
+  } catch (err) {
+    if (err instanceof ScheduleAlreadyRunning) {
+      svc.log.info('Schedule already registered in Temporal.')
+      svc.log.info('Configuration may have changed since. Please make sure they are in sync.')
+    } else {
+      throw new Error(err)
+    }
+  }
+}

services/apps/automatic_onboarding_worker/src/types.ts

@@ -0,0 +1,3 @@
+export interface IOnboardProjectsInput {
+  batchSize?: number
+}

services/apps/automatic_onboarding_worker/src/workflows.ts

@@ -0,0 +1,5 @@
+import type { IOnboardProjectsInput } from './types'
+import { onboardProjects } from './workflows/onboardProjects'
+
+export { onboardProjects }
+export type { IOnboardProjectsInput }

services/apps/automatic_onboarding_worker/src/workflows/index.ts

@@ -0,0 +1,63 @@
+import { log, proxyActivities, rootCause } from '@temporalio/workflow'
+
+import type * as activities from '../activities'
+import type { IOnboardProjectsInput } from '../types'
+
+// Short timeout: just a DB read.
+const fetchActivities = proxyActivities<typeof activities>({
+  startToCloseTimeout: '2 minutes',
+  retry: { maximumAttempts: 3 },
+})
+
+// Each onboarding call chains a segment create/query plus GitHub enrichment and integration calls; give generous headroom per project.
+const onboardActivities = proxyActivities<typeof activities>({
+  startToCloseTimeout: '2 minutes',
+  retry: { maximumAttempts: 2 },
+})
+
+export async function onboardProjects(input: IOnboardProjectsInput = {}): Promise<void> {
+  const { batchSize = 50 } = input
+
+  log.info('onboardProjects workflow started.')
+
+  const projects = await fetchActivities.fetchProjectsPendingOnboarding(batchSize)
+
+  if (projects.length === 0) {
+    log.info('No projects pending onboarding. Nothing to do.')
+    return
+  }
+
+  log.info(`Onboarding ${projects.length} project(s) (batch size: ${batchSize}).`)
+
+  let succeeded = 0
+  let failed = 0
+
+  for (let i = 0; i < projects.length; i++) {
+    const project = projects[i]
+    log.info(`[${i + 1}/${projects.length}] Onboarding: ${project.repoUrl}`)
+
+    try {
+      await onboardActivities.onboardAndUpdateProject(project)
+      succeeded++
+    } catch (err) {
+      // Activity-level retries are already exhausted at this point — mark as a
+      // terminal error so the daily schedule stops retrying this project forever.
+      failed++
+      const reason = rootCause(err) ?? String(err)
+      log.error(
+        `Onboarding failed for project id=${project.id} repoUrl=${project.repoUrl}: ${reason}`,
+      )
+
+      try {
+        await onboardActivities.markProjectOnboardingFailed(project.id, reason)
+      } catch (markErr) {
+        // Don't let a failure to record the error state abort the rest of the batch.
+        log.error(`Failed to mark project id=${project.id} as errored: ${String(markErr)}`)
+      }
+    }
+  }
+
+  log.info(
+    `Batch onboarding complete. total=${projects.length} succeeded=${succeeded} failed=${failed}`,
+  )
+}

services/apps/automatic_onboarding_worker/src/workflows/onboardProjects.ts

@@ -20,6 +20,7 @@ const PROJECT_CATALOG_COLUMNS = [
   'evaluationReason',
   'evaluatedAt',
   'onboardedAt',
+  'onboardingError',
   'syncedAt',
   'createdAt',
   'updatedAt',
@@ -472,6 +473,10 @@ export async function updateProjectCatalog(
     setClauses.push('"onboardedAt" = $(onboardedAt)')
     params.onboardedAt = data.onboardedAt
   }
+  if (data.onboardingError !== undefined) {
+    setClauses.push('"onboardingError" = $(onboardingError)')
+    params.onboardingError = data.onboardingError
+  }
 
   if (setClauses.length === 0) {
     return findProjectCatalogById(qx, id)