Something went wrong. Try again.
mirror your GitHub repos to tangled.org automatically forked from danielroe.dev/synchub.to
Something went wrong. Try again.
7.0 kB · 187 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188import type { CreateEvent, DeleteEvent, InstallationEvent, InstallationRepositoriesEvent, PushEvent, RepositoryEvent,} from '@octokit/webhooks-types'import { verify } from '@octokit/webhooks-methods'import { sql } from 'drizzle-orm'import { installation, job, webhookEvent } from '#server/db/schema'import { kickWorker } from '#server/utils/kick-worker'import { enqueue } from '#server/utils/queue'import { revokeKeysForInstallation } from '#server/utils/tangled-pubkey'
const RECOGNISED_EVENTS = new Set([ 'push', 'create', 'delete', 'repository', 'installation', 'installation_repositories',])
export default defineEventHandler(async event => { const { githubWebhookSecret } = useRuntimeConfig() if (!githubWebhookSecret) { throw createError({ statusCode: 500, statusMessage: 'webhook secret not configured' }) }
const signature = getRequestHeader(event, 'x-hub-signature-256') const deliveryHeader = getRequestHeader(event, 'x-github-delivery') const eventName = getRequestHeader(event, 'x-github-event')
if (!signature || !deliveryHeader || !eventName) { throw createError({ statusCode: 400, statusMessage: 'missing required headers' }) }
const rawBody = await readRawBody(event, 'utf8') if (!rawBody) { throw createError({ statusCode: 400, statusMessage: 'empty body' }) }
const valid = await verify(githubWebhookSecret, rawBody, signature) if (!valid) { throw createError({ statusCode: 401, statusMessage: 'invalid signature' }) }
const deliveryId = deliveryHeader.toLowerCase()
// Idempotent insert. If this delivery has already been seen, the row is not // re-inserted and `inserted` is empty; we 200 and skip downstream work. const db = useDb() const inserted = await db .insert(webhookEvent) .values({ deliveryId, source: 'github', event: eventName, }) .onConflictDoNothing({ target: webhookEvent.deliveryId }) .returning({ deliveryId: webhookEvent.deliveryId })
if (inserted.length === 0) { return { ok: true, duplicate: true } }
// Bookkeeping for installation lifecycle events. Sync work itself is enqueued // by later commits; for now we just keep the `installation` table in step so // those commits can FK-reference rows that already exist. if (eventName === 'installation') { const body = await readBody<InstallationEvent>(event) const action = body.action
if (action === 'created') { const account = body.installation.account // GitHub's `installation.account` is `User | Enterprise | null`, but the // ones we accept are user/org accounts; bail loudly on anything else. if (!account || !('login' in account) || !('type' in account)) { throw createError({ statusCode: 400, statusMessage: 'unsupported installation account' }) } const accountType = account.type === 'Organization' ? 'Organization' : 'User' await db.insert(installation).values({ id: body.installation.id, accountLogin: account.login, accountId: account.id, accountType, }).onConflictDoNothing({ target: installation.id }) } else if (action === 'deleted') { // Revoke the user's sh.tangled.publicKey PDS records first; the cascade // below drops the local ssh_key rows that hold the rkeys we need. await revokeKeysForInstallation(body.installation.id) // Cancel any still-pending work for this install. The `job` table has no // FK to `installation` (payload carries the id as JSON), so nothing // cascades; left alone, these jobs retry to exhaustion failing on the // now-missing FK target (e.g. the ssh_key insert). Drop only unstarted // work; a running job finishes and no-ops on its own guards. await db.delete(job).where(sql` ${job.status} in ('queued', 'failed') AND (${job.payload}->>'installationId')::bigint = ${body.installation.id} `) // installation row deletion cascades to user_identity, ssh_key, repo_mapping. await db.delete(installation).where(sql`${installation.id} = ${body.installation.id}`) } else if (action === 'suspend') { await db.update(installation) .set({ suspendedAt: new Date() }) .where(sql`${installation.id} = ${body.installation.id}`) } else if (action === 'unsuspend') { await db.update(installation) .set({ suspendedAt: null }) .where(sql`${installation.id} = ${body.installation.id}`) } }
// Enqueue work for events we care about. Envelope shape is the minimum the // handler needs to re-fetch fresh data via the GitHub API; we never persist // the raw webhook body. See PLAN.md "Deferred / follow-ups". if (RECOGNISED_EVENTS.has(eventName)) { await enqueueForEvent(event, eventName, deliveryId) // Nudge the worker to drain the just-enqueued job now rather than on the // next cron tick. Fire-and-forget via waitUntil; cron is the safety net. kickWorker(event) }
return { ok: true, deliveryId }})
async function enqueueForEvent(event: Parameters<typeof readBody>[0], eventName: string, deliveryId: string) { if (eventName === 'push') { const body = await readBody<PushEvent>(event) if (!body.installation) return await enqueue('github.push', { deliveryId, installationId: body.installation.id, githubRepoId: body.repository.id, ref: body.ref, before: body.before, after: body.after, }) } else if (eventName === 'create') { const body = await readBody<CreateEvent>(event) if (!body.installation) return await enqueue('github.create', { deliveryId, installationId: body.installation.id, githubRepoId: body.repository.id, refType: body.ref_type, ref: body.ref, }) } else if (eventName === 'delete') { const body = await readBody<DeleteEvent>(event) if (!body.installation) return await enqueue('github.delete', { deliveryId, installationId: body.installation.id, githubRepoId: body.repository.id, refType: body.ref_type, ref: body.ref, }) } else if (eventName === 'repository') { const body = await readBody<RepositoryEvent>(event) if (!body.installation) return await enqueue('github.repository', { deliveryId, installationId: body.installation.id, githubRepoId: body.repository.id, action: body.action, }) } else if (eventName === 'installation_repositories') { const body = await readBody<InstallationRepositoriesEvent>(event) await enqueue('github.installation_repositories', { deliveryId, installationId: body.installation.id, action: body.action, addedRepoIds: 'repositories_added' in body ? body.repositories_added.map(r => r.id) : [], removedRepoIds: 'repositories_removed' in body ? body.repositories_removed.map(r => r.id) : [], }) } // 'installation' is handled inline above for bookkeeping; no job enqueued.}