|
| 1 | +/** |
| 2 | + * sync-light-repos |
| 3 | + * |
| 4 | + * Fetches GitHub repo metadata via GraphQL and upserts into the `repos` table. |
| 5 | + * Runs one async worker per token — each worker claims URLs by index so no two |
| 6 | + * requests ever share a token concurrently. |
| 7 | + * |
| 8 | + * Success tracking: a successful fetch updates repos.last_synced_at to NOW(). |
| 9 | + * Failed repos keep a stale/null last_synced_at and are picked up on the next run. |
| 10 | + * TODO: fetchPage will later filter by last_synced_at < NOW() - update_interval |
| 11 | + * so this script becomes a continuous sync with no extra failure tracking needed. |
| 12 | + * |
| 13 | + * Usage: |
| 14 | + * pnpm run sync-light-repos -- [options] |
| 15 | + * |
| 16 | + * Options: |
| 17 | + * --page-size <n> Repos fetched from source per cursor page (default: 200) |
| 18 | + * --batch-size <n> Upsert batch size (default: 50) |
| 19 | + * --max-retries <n> Per-repo transient retry cap (default: 3) |
| 20 | + * --start-after <id> Resume from cursor id (printed after each page) |
| 21 | + * --limit <n> Stop after N repos total (for testing) |
| 22 | + * --dry-run Fetch but skip DB writes |
| 23 | + * |
| 24 | + * Environment: |
| 25 | + * GITHUB_TOKENS Comma-separated GitHub PATs (required) |
| 26 | + * CROWD_DB_WRITE_HOST/PORT/USERNAME/PASSWORD/DATABASE |
| 27 | + * SERVICE |
| 28 | + */ |
| 29 | + |
| 30 | +import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database' |
| 31 | +import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor' |
| 32 | +import { getServiceChildLogger } from '@crowd/logging' |
| 33 | + |
| 34 | +import { fetchLightRepo, parseGithubUrl } from './fetchLightRepo' |
| 35 | +import { FetchError, LightRepoResult } from './types' |
| 36 | +import { upsertLightRepos } from './upsertLightRepos' |
| 37 | + |
| 38 | +const log = getServiceChildLogger('sync-light-repos') |
| 39 | + |
| 40 | +function parseArgs() { |
| 41 | + const args = process.argv.slice(2) |
| 42 | + const getArg = (flag: string) => { |
| 43 | + const idx = args.indexOf(flag) |
| 44 | + return idx !== -1 && idx + 1 < args.length ? args[idx + 1] : undefined |
| 45 | + } |
| 46 | + |
| 47 | + const pageSize = parseInt(getArg('--page-size') ?? '200', 10) |
| 48 | + const batchSize = parseInt(getArg('--batch-size') ?? '50', 10) |
| 49 | + const maxRetries = parseInt(getArg('--max-retries') ?? '3', 10) |
| 50 | + const startAfter = getArg('--start-after') ?? null |
| 51 | + const limitRaw = getArg('--limit') |
| 52 | + const limit = limitRaw !== undefined ? parseInt(limitRaw, 10) : null |
| 53 | + const dryRun = args.includes('--dry-run') |
| 54 | + |
| 55 | + if (isNaN(pageSize) || pageSize <= 0) { log.error('--page-size must be a positive integer'); process.exit(1) } |
| 56 | + if (isNaN(batchSize) || batchSize <= 0) { log.error('--batch-size must be a positive integer'); process.exit(1) } |
| 57 | + if (isNaN(maxRetries) || maxRetries < 0) { log.error('--max-retries must be a non-negative integer'); process.exit(1) } |
| 58 | + if (limit !== null && (isNaN(limit) || limit <= 0)) { log.error('--limit must be a positive integer'); process.exit(1) } |
| 59 | + |
| 60 | + return { pageSize, batchSize, maxRetries, startAfter, limit, dryRun } |
| 61 | +} |
| 62 | + |
| 63 | +// TODO: add LEFT JOIN repos r ON r.url = pr.url and filter |
| 64 | +// WHERE (r.last_synced_at IS NULL OR r.last_synced_at < NOW() - INTERVAL '$(updateIntervalHours) hours') |
| 65 | +// once the update interval logic is scoped in. |
| 66 | +async function fetchPage( |
| 67 | + qx: ReturnType<typeof pgpQx>, |
| 68 | + cursor: string | null, |
| 69 | + pageSize: number, |
| 70 | +): Promise<{ urls: string[]; nextCursor: string | null }> { |
| 71 | + const rows = await qx.select( |
| 72 | + ` |
| 73 | + SELECT id, url |
| 74 | + FROM public.repositories |
| 75 | + WHERE url LIKE 'https://github.com/%' |
| 76 | + AND "deletedAt" IS NULL |
| 77 | + ${cursor ? 'AND id > $(cursor)' : ''} |
| 78 | + ORDER BY id |
| 79 | + LIMIT $(pageSize) |
| 80 | + `, |
| 81 | + { cursor, pageSize }, |
| 82 | + ) |
| 83 | + return { |
| 84 | + urls: rows.map((r: { url: string }) => r.url), |
| 85 | + nextCursor: rows.length > 0 ? (rows[rows.length - 1] as { id: string }).id : null, |
| 86 | + } |
| 87 | +} |
| 88 | + |
| 89 | +async function fetchWithRetries( |
| 90 | + url: string, |
| 91 | + token: string, |
| 92 | + maxRetries: number, |
| 93 | +): Promise<LightRepoResult | null> { |
| 94 | + for (let attempt = 0; attempt <= maxRetries; attempt++) { |
| 95 | + try { |
| 96 | + return await fetchLightRepo(url, token) |
| 97 | + } catch (err) { |
| 98 | + if (!(err instanceof FetchError)) throw err |
| 99 | + |
| 100 | + if (['NOT_FOUND', 'AUTH', 'MALFORMED'].includes(err.kind)) { |
| 101 | + log.warn({ url, kind: err.kind }, err.message) |
| 102 | + return null |
| 103 | + } |
| 104 | + |
| 105 | + if (err.kind === 'RATE_LIMIT') throw err |
| 106 | + |
| 107 | + if (attempt < maxRetries) { |
| 108 | + const backoffMs = 1000 * 2 ** attempt |
| 109 | + log.warn({ url, attempt, backoffMs }, `Transient error, retrying: ${err.message}`) |
| 110 | + await new Promise((r) => setTimeout(r, backoffMs)) |
| 111 | + } else { |
| 112 | + log.error({ url }, `Gave up after ${maxRetries} retries: ${err.message}`) |
| 113 | + return null |
| 114 | + } |
| 115 | + } |
| 116 | + } |
| 117 | + return null |
| 118 | +} |
| 119 | + |
| 120 | +async function processPage( |
| 121 | + urls: string[], |
| 122 | + tokens: string[], |
| 123 | + parkedUntil: Map<string, number>, |
| 124 | + opts: ReturnType<typeof parseArgs>, |
| 125 | + qx: ReturnType<typeof pgpQx>, |
| 126 | +): Promise<{ fetched: number; failed: number; flushed: number }> { |
| 127 | + const validUrls: string[] = [] |
| 128 | + let skipped = 0 |
| 129 | + for (const url of urls) { |
| 130 | + try { parseGithubUrl(url); validUrls.push(url) } catch { skipped++ } |
| 131 | + } |
| 132 | + if (skipped > 0) log.warn(`Skipped ${skipped} non-GitHub URLs`) |
| 133 | + |
| 134 | + const buffer: LightRepoResult[] = [] |
| 135 | + const failures: Array<{ url: string; reason: string }> = [] |
| 136 | + let failed = 0 |
| 137 | + let flushed = 0 |
| 138 | + let nextIdx = 0 |
| 139 | + |
| 140 | + await Promise.all( |
| 141 | + tokens.map(async (token, tokenIdx) => { |
| 142 | + // Respect any park set during a previous page of this run |
| 143 | + const initialPark = (parkedUntil.get(token) ?? 0) - Date.now() |
| 144 | + if (initialPark > 0) { |
| 145 | + log.warn(`token#${tokenIdx} still parked, waiting ${Math.round(initialPark / 1000)}s`) |
| 146 | + await new Promise((r) => setTimeout(r, initialPark)) |
| 147 | + } |
| 148 | + |
| 149 | + while (true) { |
| 150 | + const idx = nextIdx++ |
| 151 | + if (idx >= validUrls.length) break |
| 152 | + const url = validUrls[idx] |
| 153 | + |
| 154 | + try { |
| 155 | + const result = await fetchWithRetries(url, token, opts.maxRetries) |
| 156 | + if (result) { |
| 157 | + buffer.push(result) |
| 158 | + if (!opts.dryRun && buffer.length >= opts.batchSize) { |
| 159 | + const batch = buffer.splice(0) |
| 160 | + await upsertLightRepos(qx, batch) |
| 161 | + flushed += batch.length |
| 162 | + } |
| 163 | + } else { |
| 164 | + failures.push({ url, reason: 'see warn log above' }) |
| 165 | + failed++ |
| 166 | + } |
| 167 | + } catch (err) { |
| 168 | + if (err instanceof FetchError && err.kind === 'RATE_LIMIT') { |
| 169 | + const resetAt = err.resetAt ?? Date.now() + 60_000 |
| 170 | + const waitMs = Math.max(1_000, resetAt - Date.now()) |
| 171 | + parkedUntil.set(token, resetAt) |
| 172 | + log.warn( |
| 173 | + { tokenIdx, parkedUntil: new Date(resetAt).toISOString() }, |
| 174 | + `token#${tokenIdx} rate limited — parking for ${Math.round(waitMs / 1000)}s`, |
| 175 | + ) |
| 176 | + await new Promise((r) => setTimeout(r, waitMs)) |
| 177 | + failures.push({ url, reason: 'rate-limit' }) |
| 178 | + failed++ |
| 179 | + } else { |
| 180 | + log.error({ url, err }, 'Unexpected error') |
| 181 | + failures.push({ url, reason: (err as Error).message }) |
| 182 | + failed++ |
| 183 | + } |
| 184 | + } |
| 185 | + } |
| 186 | + }), |
| 187 | + ) |
| 188 | + |
| 189 | + if (!opts.dryRun && buffer.length > 0) { |
| 190 | + await upsertLightRepos(qx, buffer) |
| 191 | + flushed += buffer.length |
| 192 | + } |
| 193 | + |
| 194 | + if (failures.length > 0) { |
| 195 | + log.warn({ failures }, `${failures.length} repo(s) failed this page`) |
| 196 | + } |
| 197 | + |
| 198 | + return { fetched: validUrls.length - failed, failed, flushed } |
| 199 | +} |
| 200 | + |
| 201 | +async function main() { |
| 202 | + const opts = parseArgs() |
| 203 | + |
| 204 | + const tokens = (process.env.GITHUB_TOKENS ?? '') |
| 205 | + .split(',') |
| 206 | + .map((t) => t.trim()) |
| 207 | + .filter(Boolean) |
| 208 | + |
| 209 | + if (tokens.length === 0) { |
| 210 | + log.error('GITHUB_TOKENS is required (comma-separated PATs)') |
| 211 | + process.exit(1) |
| 212 | + } |
| 213 | + |
| 214 | + // TODO: when connecting the real DB, replace with a connection pool and add keepalive / |
| 215 | + // reconnect-on-error handling. A single long-lived connection will be dropped by the server |
| 216 | + // during multi-hour runs (TCP timeout, idle reaper), crashing the script. Completed work |
| 217 | + // is safe via last_synced_at, but the run stops and must be manually resumed. |
| 218 | + const dbConnection = await getDbConnection(WRITE_DB_CONFIG()) |
| 219 | + const qx = pgpQx(dbConnection) |
| 220 | + |
| 221 | + log.info('='.repeat(60)) |
| 222 | + log.info('sync-light-repos') |
| 223 | + log.info(`tokens=${tokens.length} page-size=${opts.pageSize} batch-size=${opts.batchSize}`) |
| 224 | + log.info(`max-retries=${opts.maxRetries} dry-run=${opts.dryRun} limit=${opts.limit ?? 'none'}`) |
| 225 | + log.info(`start-after=${opts.startAfter ?? '(beginning)'}`) |
| 226 | + log.info('='.repeat(60)) |
| 227 | + |
| 228 | + const parkedUntil = new Map<string, number>() |
| 229 | + let cursor = opts.startAfter |
| 230 | + let pageNum = 0 |
| 231 | + let totalProcessed = 0 |
| 232 | + let totalFailed = 0 |
| 233 | + let totalFlushed = 0 |
| 234 | + |
| 235 | + while (true) { |
| 236 | + pageNum++ |
| 237 | + |
| 238 | + const remaining = opts.limit !== null ? opts.limit - totalProcessed : opts.pageSize |
| 239 | + if (remaining <= 0) break |
| 240 | + |
| 241 | + const { urls, nextCursor } = await fetchPage(qx, cursor, Math.min(opts.pageSize, remaining)) |
| 242 | + |
| 243 | + if (urls.length === 0) { |
| 244 | + log.info('No more repos to process') |
| 245 | + break |
| 246 | + } |
| 247 | + |
| 248 | + const { fetched, failed, flushed } = await processPage(urls, tokens, parkedUntil, opts, qx) |
| 249 | + |
| 250 | + totalProcessed += urls.length |
| 251 | + totalFailed += failed |
| 252 | + totalFlushed += flushed |
| 253 | + |
| 254 | + log.info( |
| 255 | + `Page ${pageNum}: read=${urls.length} fetched=${fetched} failed=${failed}${opts.dryRun ? ' [dry-run]' : ` flushed=${flushed}`}`, |
| 256 | + ) |
| 257 | + |
| 258 | + if (nextCursor) { |
| 259 | + log.info(`Resume with: --start-after ${nextCursor}`) |
| 260 | + cursor = nextCursor |
| 261 | + } |
| 262 | + |
| 263 | + if (urls.length < Math.min(opts.pageSize, remaining)) break |
| 264 | + } |
| 265 | + |
| 266 | + log.info('='.repeat(60)) |
| 267 | + log.info(`Summary: pages=${pageNum} processed=${totalProcessed} failed=${totalFailed} flushed=${totalFlushed}`) |
| 268 | + log.info('='.repeat(60)) |
| 269 | + |
| 270 | + process.exit(totalFailed > 0 ? 1 : 0) |
| 271 | +} |
| 272 | + |
| 273 | +main().catch((err) => { |
| 274 | + log.error({ err }, 'Unexpected error') |
| 275 | + process.exit(1) |
| 276 | +}) |
0 commit comments