feat: add enrichment loop and github-repos-enricher entry point

Signed-off-by: Mouad BANI <mouad-mb@outlook.com>

Author: mbani01

services/apps/packages_worker/src/bin/github-repos-enricher.ts

@@ -0,0 +1,69 @@
+import fs from 'fs'
+import path from 'path'
+
+import { getServiceLogger } from '@crowd/logging'
+
+import { getEnricherConfig } from '../config'
+import { getPackagesDb } from '../db'
+import { runEnrichmentLoop } from '../enricher/runEnrichmentLoop'
+
+const log = getServiceLogger()
+
+const liveFilePath = path.join(__dirname, '../tmp/github-repos-enricher-live.tmp')
+const readyFilePath = path.join(__dirname, '../tmp/github-repos-enricher-ready.tmp')
+
+let shuttingDown = false
+
+const shutdown = async () => {
+  if (shuttingDown) return
+  shuttingDown = true
+  log.info('Shutting down github-repos-enricher...')
+}
+
+process.on('SIGINT', shutdown)
+process.on('SIGTERM', shutdown)
+
+const main = async () => {
+  log.info('github-repos-enricher starting...')
+
+  const config = getEnricherConfig()
+
+  if (config.tokens.length === 0) {
+    log.error('ENRICHER_GITHUB_TOKENS is required (comma-separated PATs)')
+    process.exit(1)
+  }
+
+  const qx = await getPackagesDb()
+  await qx.selectOne('SELECT 1')
+  log.info('Connected to packages-db.')
+
+  fs.mkdirSync(path.dirname(liveFilePath), { recursive: true })
+
+  const healthInterval = setInterval(async () => {
+    if (shuttingDown) return
+    try {
+      await Promise.all([
+        fs.promises.open(liveFilePath, 'a').then((f) => f.close()),
+        fs.promises.open(readyFilePath, 'a').then((f) => f.close()),
+      ])
+    } catch (err) {
+      log.warn({ err }, 'Failed to write health probe files')
+    }
+  }, 5000)
+
+  log.info(
+    { tokens: config.tokens.length, pageSize: config.pageSize, batchSize: config.batchSize },
+    'Starting enrichment loop',
+  )
+
+  await runEnrichmentLoop(qx, config, () => shuttingDown)
+
+  clearInterval(healthInterval)
+  log.info('github-repos-enricher stopped.')
+  process.exit(0)
+}
+
+main().catch((err) => {
+  log.error({ err }, 'github-repos-enricher fatal error')
+  process.exit(1)
+})

services/apps/packages_worker/src/enricher/runEnrichmentLoop.ts

@@ -0,0 +1,183 @@
+import { getServiceChildLogger } from '@crowd/logging'
+import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
+
+import { getEnricherConfig } from '../config'
+import { fetchLightRepo, parseGithubUrl } from './fetchLightRepo'
+import { FetchError, LightRepoResult } from './types'
+import { updateEnrichedRepos } from './updateEnrichedRepos'
+
+const log = getServiceChildLogger('github-repos-enricher')
+
+async function fetchWithRetries(
+  url: string,
+  token: string,
+  maxRetries: number,
+): Promise<LightRepoResult | null> {
+  for (let attempt = 0; attempt <= maxRetries; attempt++) {
+    try {
+      return await fetchLightRepo(url, token)
+    } catch (err) {
+      if (!(err instanceof FetchError)) throw err
+
+      if (['NOT_FOUND', 'AUTH', 'MALFORMED'].includes(err.kind)) {
+        log.warn({ url, kind: err.kind }, err.message)
+        return null
+      }
+
+      if (err.kind === 'RATE_LIMIT') throw err
+
+      if (attempt < maxRetries) {
+        const backoffMs = 1000 * 2 ** attempt
+        log.warn({ url, attempt, backoffMs }, `Transient error, retrying: ${err.message}`)
+        await new Promise((r) => setTimeout(r, backoffMs))
+      } else {
+        log.error({ url }, `Gave up after ${maxRetries} retries: ${err.message}`)
+        return null
+      }
+    }
+  }
+  return null
+}
+
+async function fetchPage(
+  qx: QueryExecutor,
+  cursor: string | null,
+  pageSize: number,
+  updateIntervalHours: number,
+): Promise<{ rows: Array<{ id: string; url: string }>; urls: string[] }> {
+  const rows = await qx.select(
+    `
+    SELECT id, url
+    FROM repos
+    WHERE host = 'github'
+      AND (last_synced_at IS NULL OR last_synced_at < NOW() - INTERVAL '$(updateIntervalHours) hours')
+      AND ($(cursor) IS NULL OR id > $(cursor))
+    ORDER BY id
+    LIMIT $(pageSize)
+    `,
+    { cursor, pageSize, updateIntervalHours },
+  )
+  return {
+    rows,
+    urls: rows.map((r: { url: string }) => r.url),
+  }
+}
+
+async function processPage(
+  urls: string[],
+  tokens: string[],
+  parkedUntil: Map<string, number>,
+  config: ReturnType<typeof getEnricherConfig>,
+  qx: QueryExecutor,
+): Promise<{ fetched: number; failed: number; flushed: number }> {
+  const validUrls: string[] = []
+  let skipped = 0
+  for (const url of urls) {
+    try {
+      parseGithubUrl(url)
+      validUrls.push(url)
+    } catch {
+      skipped++
+    }
+  }
+  if (skipped > 0) log.warn(`Skipped ${skipped} non-GitHub URLs`)
+
+  const buffer: LightRepoResult[] = []
+  const failures: Array<{ url: string; reason: string }> = []
+  let failed = 0
+  let flushed = 0
+  let nextIdx = 0
+
+  await Promise.all(
+    tokens.map(async (token, tokenIdx) => {
+      // Respect any park set during a previous page of this run
+      const initialPark = (parkedUntil.get(token) ?? 0) - Date.now()
+      if (initialPark > 0) {
+        log.warn(`token#${tokenIdx} still parked, waiting ${Math.round(initialPark / 1000)}s`)
+        await new Promise((r) => setTimeout(r, initialPark))
+      }
+
+      while (true) {
+        const idx = nextIdx++
+        if (idx >= validUrls.length) break
+        const url = validUrls[idx]
+
+        try {
+          const result = await fetchWithRetries(url, token, config.maxRetries)
+          if (result) {
+            buffer.push(result)
+            if (buffer.length >= config.batchSize) {
+              const batch = buffer.splice(0)
+              await updateEnrichedRepos(qx, batch)
+              flushed += batch.length
+            }
+          } else {
+            failures.push({ url, reason: 'see warn log above' })
+            failed++
+          }
+        } catch (err) {
+          if (err instanceof FetchError && err.kind === 'RATE_LIMIT') {
+            const resetAt = err.resetAt ?? Date.now() + 60_000
+            const waitMs = Math.max(1_000, resetAt - Date.now())
+            parkedUntil.set(token, resetAt)
+            log.warn(
+              { tokenIdx, parkedUntil: new Date(resetAt).toISOString() },
+              `token#${tokenIdx} rate limited — parking for ${Math.round(waitMs / 1000)}s`,
+            )
+            await new Promise((r) => setTimeout(r, waitMs))
+            failures.push({ url, reason: 'rate-limit' })
+            failed++
+          } else {
+            log.error({ url, err }, 'Unexpected error')
+            failures.push({ url, reason: (err as Error).message })
+            failed++
+          }
+        }
+      }
+    }),
+  )
+
+  if (buffer.length > 0) {
+    await updateEnrichedRepos(qx, buffer)
+    flushed += buffer.length
+  }
+
+  if (failures.length > 0) {
+    log.warn({ failures }, `${failures.length} repo(s) failed this page`)
+  }
+
+  return { fetched: validUrls.length - failed, failed, flushed }
+}
+
+export async function runEnrichmentLoop(
+  qx: QueryExecutor,
+  config: ReturnType<typeof getEnricherConfig>,
+  isShuttingDown: () => boolean,
+): Promise<void> {
+  const parkedUntil = new Map<string, number>()
+  let cursor: string | null = null
+  let pageNum = 0
+
+  while (!isShuttingDown()) {
+    pageNum++
+
+    const { rows, urls } = await fetchPage(qx, cursor, config.pageSize, config.updateIntervalHours)
+
+    if (urls.length === 0) {
+      log.info('No more repos to process, sleeping')
+      await new Promise((r) => setTimeout(r, config.idleSleepSec * 1000))
+      cursor = null
+      continue
+    }
+
+    const { fetched, failed, flushed } = await processPage(urls, config.tokens, parkedUntil, config, qx)
+
+    log.info(
+      `Page ${pageNum}: read=${urls.length} fetched=${fetched} failed=${failed} flushed=${flushed}`,
+    )
+
+    if (rows.length > 0) {
+      cursor = rows[rows.length - 1].id
+    }
+  }
+}

services/apps/script_executor_worker/package.json

@@ -12,6 +12,7 @@
     "recalculate-enrichment-affiliations": "npx tsx src/bin/recalculate-enrichment-affiliations.ts",
     "recalculate-all-affiliations": "npx tsx src/bin/recalculate-all-affiliations.ts",
     "add-lf-projects-to-collection": "npx tsx src/bin/add-lf-projects-to-collection.ts",
+    "sync-light-repos": "npx tsx src/bin/sync-light-repos/index.ts",
     "lint": "npx eslint --ext .ts src --max-warnings=0",
     "format": "npx prettier --write \"src/**/*.ts\"",
     "format-check": "npx prettier --check .",

services/apps/script_executor_worker/src/bin/sync-light-repos/fetchLightRepo.ts

@@ -0,0 +1,96 @@
+import { FetchError, LightRepoResult } from './types'
+
+const GRAPHQL_URL = 'https://api.github.com/graphql'
+
+const REPO_QUERY = `
+  query($owner: String!, $name: String!) {
+    repository(owner: $owner, name: $name) {
+      description
+      primaryLanguage { name }
+      repositoryTopics(first: 25) { nodes { topic { name } } }
+      stargazerCount
+      forkCount
+      watchers { totalCount }
+      issues(states: OPEN) { totalCount }
+      pushedAt
+      isArchived
+      isDisabled
+      isFork
+      createdAt
+    }
+  }
+`
+
+export function parseGithubUrl(url: string): { owner: string; name: string } {
+  const match = url.match(/https?:\/\/github\.com\/([^/]+)\/([^/]+?)(?:\.git)?\/?$/)
+  if (!match) throw new FetchError('MALFORMED', `Cannot parse GitHub URL: ${url}`)
+  return { owner: match[1], name: match[2] }
+}
+
+export async function fetchLightRepo(url: string, token: string): Promise<LightRepoResult> {
+  const { owner, name } = parseGithubUrl(url)
+
+  let response: Response
+  try {
+    response = await fetch(GRAPHQL_URL, {
+      method: 'POST',
+      headers: {
+        Authorization: `bearer ${token}`,
+        'Content-Type': 'application/json',
+      },
+      body: JSON.stringify({ query: REPO_QUERY, variables: { owner, name } }),
+    })
+  } catch (err) {
+    throw new FetchError('TRANSIENT', `Network error for ${url}: ${(err as Error).message}`)
+  }
+
+  const resetSec = parseInt(response.headers.get('x-ratelimit-reset') ?? '0', 10)
+  const resetMs = resetSec ? resetSec * 1000 + 5_000 : Date.now() + 65_000
+
+  if (response.status === 401) {
+    throw new FetchError('AUTH', `401 Unauthorized for ${url}`)
+  }
+
+  if (response.status === 403) {
+    const body = await response.text()
+    if (body.toLowerCase().includes('rate limit')) {
+      throw new FetchError('RATE_LIMIT', `Rate limited on ${url}`, resetMs)
+    }
+    throw new FetchError('AUTH', `403 Forbidden for ${url}`)
+  }
+
+  if (response.status === 404) throw new FetchError('NOT_FOUND', `404 for ${url}`)
+  if (response.status >= 500) throw new FetchError('TRANSIENT', `${response.status} for ${url}`)
+
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  const json = (await response.json()) as any
+
+  if (json.errors?.length) {
+    const err = json.errors[0]
+    if (err.type === 'RATE_LIMITED') throw new FetchError('RATE_LIMIT', `RATE_LIMITED for ${url}`, resetMs)
+    if (err.type === 'NOT_FOUND') throw new FetchError('NOT_FOUND', `NOT_FOUND for ${url}`)
+    throw new FetchError('TRANSIENT', `GraphQL error for ${url}: ${err.message ?? err.type}`)
+  }
+
+  const repo = json.data?.repository
+  if (!repo) throw new FetchError('NOT_FOUND', `No repository data for ${url}`)
+
+  return {
+    url,
+    host: 'github',
+    owner,
+    name,
+    description: repo.description ?? null,
+    primaryLanguage: repo.primaryLanguage?.name ?? null,
+    topics: (repo.repositoryTopics?.nodes ?? []).map((n: { topic: { name: string } }) => n.topic.name),
+    stars: repo.stargazerCount ?? 0,
+    forks: repo.forkCount ?? 0,
+    watchers: repo.watchers?.totalCount ?? 0,
+    openIssues: repo.issues?.totalCount ?? 0,
+    lastCommitAt: repo.pushedAt ?? null,
+    archived: repo.isArchived ?? false,
+    disabled: repo.isDisabled ?? false,
+    isFork: repo.isFork ?? false,
+    createdAt: repo.createdAt ?? null,
+  }
+}