danielroe.dev
mirror your GitHub repos to tangled.org automatically
synchub.to
mirror
sync
tangled
github
git
7.0 kB
187 lines
1import type {
2 CreateEvent,
3 DeleteEvent,
4 InstallationEvent,
5 InstallationRepositoriesEvent,
6 PushEvent,
7 RepositoryEvent,
8} from '@octokit/webhooks-types'
9import { verify } from '@octokit/webhooks-methods'
10import { sql } from 'drizzle-orm'
11import { installation, job, webhookEvent } from '#server/db/schema'
12import { kickWorker } from '#server/utils/kick-worker'
13import { enqueue } from '#server/utils/queue'
14import { revokeKeysForInstallation } from '#server/utils/tangled-pubkey'
15
16const RECOGNISED_EVENTS = new Set([
17 'push',
18 'create',
19 'delete',
20 'repository',
21 'installation',
22 'installation_repositories',
23])
24
25export default defineEventHandler(async event => {
26 const { githubWebhookSecret } = useRuntimeConfig()
27 if (!githubWebhookSecret) {
28 throw createError({ statusCode: 500, statusMessage: 'webhook secret not configured' })
29 }
30
31 const signature = getRequestHeader(event, 'x-hub-signature-256')
32 const deliveryHeader = getRequestHeader(event, 'x-github-delivery')
33 const eventName = getRequestHeader(event, 'x-github-event')
34
35 if (!signature || !deliveryHeader || !eventName) {
36 throw createError({ statusCode: 400, statusMessage: 'missing required headers' })
37 }
38
39 const rawBody = await readRawBody(event, 'utf8')
40 if (!rawBody) {
41 throw createError({ statusCode: 400, statusMessage: 'empty body' })
42 }
43
44 const valid = await verify(githubWebhookSecret, rawBody, signature)
45 if (!valid) {
46 throw createError({ statusCode: 401, statusMessage: 'invalid signature' })
47 }
48
49 const deliveryId = deliveryHeader.toLowerCase()
50
51 // Idempotent insert. If this delivery has already been seen, the row is not
52 // re-inserted and `inserted` is empty; we 200 and skip downstream work.
53 const db = useDb()
54 const inserted = await db
55 .insert(webhookEvent)
56 .values({
57 deliveryId,
58 source: 'github',
59 event: eventName,
60 })
61 .onConflictDoNothing({ target: webhookEvent.deliveryId })
62 .returning({ deliveryId: webhookEvent.deliveryId })
63
64 if (inserted.length === 0) {
65 return { ok: true, duplicate: true }
66 }
67
68 // Bookkeeping for installation lifecycle events. Sync work itself is enqueued
69 // by later commits; for now we just keep the `installation` table in step so
70 // those commits can FK-reference rows that already exist.
71 if (eventName === 'installation') {
72 const body = await readBody<InstallationEvent>(event)
73 const action = body.action
74
75 if (action === 'created') {
76 const account = body.installation.account
77 // GitHub's `installation.account` is `User | Enterprise | null`, but the
78 // ones we accept are user/org accounts; bail loudly on anything else.
79 if (!account || !('login' in account) || !('type' in account)) {
80 throw createError({ statusCode: 400, statusMessage: 'unsupported installation account' })
81 }
82 const accountType = account.type === 'Organization' ? 'Organization' : 'User'
83 await db.insert(installation).values({
84 id: body.installation.id,
85 accountLogin: account.login,
86 accountId: account.id,
87 accountType,
88 }).onConflictDoNothing({ target: installation.id })
89 }
90 else if (action === 'deleted') {
91 // Revoke the user's sh.tangled.publicKey PDS records first; the cascade
92 // below drops the local ssh_key rows that hold the rkeys we need.
93 await revokeKeysForInstallation(body.installation.id)
94 // Cancel any still-pending work for this install. The `job` table has no
95 // FK to `installation` (payload carries the id as JSON), so nothing
96 // cascades; left alone, these jobs retry to exhaustion failing on the
97 // now-missing FK target (e.g. the ssh_key insert). Drop only unstarted
98 // work; a running job finishes and no-ops on its own guards.
99 await db.delete(job).where(sql`
100 ${job.status} in ('queued', 'failed')
101 AND (${job.payload}->>'installationId')::bigint = ${body.installation.id}
102 `)
103 // installation row deletion cascades to user_identity, ssh_key, repo_mapping.
104 await db.delete(installation).where(sql`${installation.id} = ${body.installation.id}`)
105 }
106 else if (action === 'suspend') {
107 await db.update(installation)
108 .set({ suspendedAt: new Date() })
109 .where(sql`${installation.id} = ${body.installation.id}`)
110 }
111 else if (action === 'unsuspend') {
112 await db.update(installation)
113 .set({ suspendedAt: null })
114 .where(sql`${installation.id} = ${body.installation.id}`)
115 }
116 }
117
118 // Enqueue work for events we care about. Envelope shape is the minimum the
119 // handler needs to re-fetch fresh data via the GitHub API; we never persist
120 // the raw webhook body. See PLAN.md "Deferred / follow-ups".
121 if (RECOGNISED_EVENTS.has(eventName)) {
122 await enqueueForEvent(event, eventName, deliveryId)
123 // Nudge the worker to drain the just-enqueued job now rather than on the
124 // next cron tick. Fire-and-forget via waitUntil; cron is the safety net.
125 kickWorker(event)
126 }
127
128 return { ok: true, deliveryId }
129})
130
131async function enqueueForEvent(event: Parameters<typeof readBody>[0], eventName: string, deliveryId: string) {
132 if (eventName === 'push') {
133 const body = await readBody<PushEvent>(event)
134 if (!body.installation) return
135 await enqueue('github.push', {
136 deliveryId,
137 installationId: body.installation.id,
138 githubRepoId: body.repository.id,
139 ref: body.ref,
140 before: body.before,
141 after: body.after,
142 })
143 }
144 else if (eventName === 'create') {
145 const body = await readBody<CreateEvent>(event)
146 if (!body.installation) return
147 await enqueue('github.create', {
148 deliveryId,
149 installationId: body.installation.id,
150 githubRepoId: body.repository.id,
151 refType: body.ref_type,
152 ref: body.ref,
153 })
154 }
155 else if (eventName === 'delete') {
156 const body = await readBody<DeleteEvent>(event)
157 if (!body.installation) return
158 await enqueue('github.delete', {
159 deliveryId,
160 installationId: body.installation.id,
161 githubRepoId: body.repository.id,
162 refType: body.ref_type,
163 ref: body.ref,
164 })
165 }
166 else if (eventName === 'repository') {
167 const body = await readBody<RepositoryEvent>(event)
168 if (!body.installation) return
169 await enqueue('github.repository', {
170 deliveryId,
171 installationId: body.installation.id,
172 githubRepoId: body.repository.id,
173 action: body.action,
174 })
175 }
176 else if (eventName === 'installation_repositories') {
177 const body = await readBody<InstallationRepositoriesEvent>(event)
178 await enqueue('github.installation_repositories', {
179 deliveryId,
180 installationId: body.installation.id,
181 action: body.action,
182 addedRepoIds: 'repositories_added' in body ? body.repositories_added.map(r => r.id) : [],
183 removedRepoIds: 'repositories_removed' in body ? body.repositories_removed.map(r => r.id) : [],
184 })
185 }
186 // 'installation' is handled inline above for bookkeeping; no job enqueued.
187}