Something went wrong. Try again.
mirror your GitHub repos to tangled.org automatically synchub.to
mirror sync tangled github git
Something went wrong. Try again.
14 kB · 408 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409import type { OAuthSession } from '@atproto/oauth-client-node'import { and, eq, sql } from 'drizzle-orm'import { installation, repoMapping, userIdentity } from '../db/schema'import { restoreSessionOrNull } from './atproto-oauth'import { useDb } from './db'import { installationOctokit } from './github-app'import type { JobEnvelope } from './queue'import { enqueue } from './queue'import { type CreateRefPayload, type DeleteRefPayload, syncCreateRef, syncDeleteRef } from './sync-ref'import { syncPush, type PushPayload } from './sync-push'import { generateAndPublishKey, rotateKey } from './tangled-pubkey'import { enrollRepo, syncRepoMetadata } from './tangled-repo'
/** Job kinds `dispatch` understands; anything else throws as a job failure. */const KNOWN_KINDS = new Set([ 'github.push', 'github.create', 'github.delete', 'github.repository', 'github.installation_repositories', 'tangled.backfill-installation', 'tangled.create-repo', 'atproto.publish-pubkey',])
const BACKFILL_PAGE_SIZE = 100
interface PublishPubkeyPayload { did: string installationId: number /** * Dashboard "Rotate SSH key" sets this. Causes the handler to call * `rotateKey()` (delete old PDS record + DB row, then re-publish) rather * than the no-op-if-exists `generateAndPublishKey()` used at signup. */ force?: boolean}
interface CreateRepoPayload { installationId: number githubRepoId: number /** Dashboard "Resync now" sets this; see `enrollRepo` for semantics. */ force?: boolean}
interface InstallationRepositoriesPayload { installationId: number action: 'added' | 'removed' addedRepoIds: number[] removedRepoIds: number[]}
interface BackfillInstallationPayload { installationId: number page: number}
type RepositoryAction = | 'created' | 'edited' | 'renamed' | 'transferred' | 'deleted' | 'privatized' | 'publicized' | 'archived' | 'unarchived'
interface RepositoryPayload { installationId: number githubRepoId: number action: RepositoryAction}
function asObject(value: unknown): Record<string, unknown> { if (value === null || typeof value !== 'object') { throw new TypeError(`expected object payload, got ${typeof value}`) } return { ...value }}
function publishPubkeyPayload(value: unknown): PublishPubkeyPayload { const o = asObject(value) if (typeof o.did !== 'string' || typeof o.installationId !== 'number') { throw new TypeError('invalid atproto.publish-pubkey payload') } return { did: o.did, installationId: o.installationId, force: o.force === true }}
function createRepoPayload(value: unknown): CreateRepoPayload { const o = asObject(value) if (typeof o.installationId !== 'number' || typeof o.githubRepoId !== 'number') { throw new TypeError('invalid tangled.create-repo payload') } return { installationId: o.installationId, githubRepoId: o.githubRepoId, force: o.force === true, }}
function backfillInstallationPayload(value: unknown): BackfillInstallationPayload { const o = asObject(value) if (typeof o.installationId !== 'number' || typeof o.page !== 'number') { throw new TypeError('invalid tangled.backfill-installation payload') } return { installationId: o.installationId, page: o.page }}
function refPayload(kind: 'create' | 'delete', value: unknown): CreateRefPayload { const o = asObject(value) if ( typeof o.installationId !== 'number' || typeof o.githubRepoId !== 'number' || (o.refType !== 'branch' && o.refType !== 'tag') || typeof o.ref !== 'string' ) { throw new TypeError(`invalid github.${kind} payload`) } return { installationId: o.installationId, githubRepoId: o.githubRepoId, refType: o.refType, ref: o.ref, }}
const REPOSITORY_ACTIONS = new Set<string>([ 'created', 'edited', 'renamed', 'transferred', 'deleted', 'privatized', 'publicized', 'archived', 'unarchived',])
function isRepositoryAction(action: string): action is RepositoryAction { return REPOSITORY_ACTIONS.has(action)}
function repositoryPayload(value: unknown): RepositoryPayload { const o = asObject(value) if ( typeof o.installationId !== 'number' || typeof o.githubRepoId !== 'number' || typeof o.action !== 'string' || !isRepositoryAction(o.action) ) { throw new TypeError('invalid github.repository payload') } return { installationId: o.installationId, githubRepoId: o.githubRepoId, action: o.action, }}
function installationRepositoriesPayload(value: unknown): InstallationRepositoriesPayload { const o = asObject(value) if ( typeof o.installationId !== 'number' || (o.action !== 'added' && o.action !== 'removed') || !Array.isArray(o.addedRepoIds) || !Array.isArray(o.removedRepoIds) ) { throw new TypeError('invalid github.installation_repositories payload') } return { installationId: o.installationId, action: o.action, addedRepoIds: o.addedRepoIds.filter((id): id is number => typeof id === 'number'), removedRepoIds: o.removedRepoIds.filter((id): id is number => typeof id === 'number'), }}
export async function dispatch(envelope: JobEnvelope): Promise<void> { if (!KNOWN_KINDS.has(envelope.kind)) { throw new Error(`unknown job kind: ${envelope.kind}`) }
if (envelope.kind === 'github.push') { await syncPush(envelope.payload as PushPayload) return }
if (envelope.kind === 'github.create') { await syncCreateRef(refPayload('create', envelope.payload)) return }
if (envelope.kind === 'github.delete') { await syncDeleteRef(refPayload('delete', envelope.payload) as DeleteRefPayload) return }
if (envelope.kind === 'atproto.publish-pubkey') { const { did, installationId, force } = publishPubkeyPayload(envelope.payload) // The install may have been uninstalled after this job was enqueued; its // row (and any ssh_key FK target) is then gone, so the key insert would // fail forever on the foreign key. Drop cleanly instead of retrying, and // avoid publishing an orphan sh.tangled.publicKey record to the PDS. if (!(await installationExists(installationId))) return // The user may have revoked the app on their PDS after this job was // enqueued; the session is then gone and can't be restored. Drop cleanly // rather than throwing and retrying to exhaustion. const session = await restoreSessionOrNull(did) if (!session) return if (force) await rotateKey({ oauthSession: session, installationId }) else await generateAndPublishKey({ oauthSession: session, installationId }) return }
if (envelope.kind === 'tangled.create-repo') { const { installationId, githubRepoId, force } = createRepoPayload(envelope.payload) const session = await restoreSessionForInstallation(installationId) if (!session) return await enrollRepo({ oauthSession: session, installationId, githubRepoId, force }) return }
if (envelope.kind === 'tangled.backfill-installation') { const { installationId, page } = backfillInstallationPayload(envelope.payload) const octokit = await installationOctokit(installationId) const { data } = await octokit.request('GET /installation/repositories', { per_page: BACKFILL_PAGE_SIZE, page, })
// Fan out one tangled.create-repo job per repo on this page. for (const repo of data.repositories) { // eslint-disable-next-line no-await-in-loop -- enqueue is sequential by design await enqueue('tangled.create-repo', { installationId, githubRepoId: repo.id }) }
// If there are more pages, re-queue ourselves for the next one. This // keeps each tick small and bounded; an install with thousands of repos // walks through over many minutes rather than blocking one worker. const seenSoFar = (page - 1) * BACKFILL_PAGE_SIZE + data.repositories.length if (seenSoFar < data.total_count && data.repositories.length > 0) { await enqueue('tangled.backfill-installation', { installationId, page: page + 1 }) } return }
if (envelope.kind === 'github.installation_repositories') { const { installationId, action, addedRepoIds, removedRepoIds } = installationRepositoriesPayload(envelope.payload)
if (action === 'added') { // Fan out one tangled.create-repo job per added repo. The fan-out keeps // each unit small enough to fit comfortably in the per-job lease, lets // failures retry independently, and runs the OAuth precondition check // per repo (an install can outlive a tangled identity disconnection). for (const id of addedRepoIds) { // eslint-disable-next-line no-await-in-loop -- fan-out enqueue is sequential by design await enqueue('tangled.create-repo', { installationId, githubRepoId: id }) } return }
// 'removed': the install no longer has access to these repos. We can't // see them via the install token any more, so syncing has to stop. Leave // the tangled mirror in place (PLAN.md: "don't delete user data on our // say-so"). The user can manually re-add the repo on GitHub to re-enable. if (action === 'removed' && removedRepoIds.length > 0) { const db = useDb() await db.update(repoMapping) .set({ disabledAt: new Date(), updatedAt: new Date() }) .where(and( eq(repoMapping.installationId, installationId), sql`${repoMapping.githubRepoId} IN ${removedRepoIds}`, )) } return }
if (envelope.kind === 'github.repository') { await handleRepositoryEvent(repositoryPayload(envelope.payload)) return }}
async function handleRepositoryEvent(payload: RepositoryPayload): Promise<void> { const { installationId, githubRepoId, action } = payload const db = useDb()
// Most lifecycle actions only need to touch the local mapping, no PDS work. // `edited` and `publicized` (when we already have a mapping) go through the // OAuth-authed metadata sync helper. if (action === 'privatized' || action === 'transferred' || action === 'deleted') { await db.update(repoMapping) .set({ disabledAt: new Date(), updatedAt: new Date() }) .where(and( eq(repoMapping.installationId, installationId), eq(repoMapping.githubRepoId, githubRepoId), )) return }
if (action === 'renamed') { // Refetch the current full_name from GitHub; the webhook envelope is // intentionally tiny. We do NOT rename on the tangled side — there's no // procedure and a fresh-create-and-delete would lose stars/refs. Surface // the change in `lastError` with an `info:` prefix so the dashboard can // flag it without a schema change. const rows = await db.select({ id: repoMapping.id, githubFullName: repoMapping.githubFullName }) .from(repoMapping) .where(and( eq(repoMapping.installationId, installationId), eq(repoMapping.githubRepoId, githubRepoId), )) .limit(1) if (rows.length === 0) return const row = rows[0]!
const octokit = await installationOctokit(installationId) const { data: repo } = await octokit.request('GET /repositories/{repository_id}', { repository_id: githubRepoId, }) if (repo.full_name === row.githubFullName) return
await db.update(repoMapping) .set({ githubFullName: repo.full_name, lastError: `info: renamed on github from ${row.githubFullName} to ${repo.full_name}; tangled mirror name unchanged`, updatedAt: new Date(), }) .where(eq(repoMapping.id, row.id)) return }
if (action === 'publicized') { const rows = await db.select().from(repoMapping).where(and( eq(repoMapping.installationId, installationId), eq(repoMapping.githubRepoId, githubRepoId), )).limit(1) const row = rows[0]
if (!row) { // Repo flipped public without ever having been enrolled (the install // was added while it was private). Kick off a fresh enrolment. await enqueue('tangled.create-repo', { installationId, githubRepoId }) return }
await db.update(repoMapping) .set({ disabledAt: null, updatedAt: new Date() }) .where(eq(repoMapping.id, row.id))
if (!row.tangledRepoDid) { await enqueue('tangled.create-repo', { installationId, githubRepoId }) return }
// Already mirrored, just refresh metadata in case description/topics // changed while it was private. await runMetadataSync(installationId, githubRepoId) return }
if (action === 'edited') { await runMetadataSync(installationId, githubRepoId) return }
// 'created', 'archived', 'unarchived' — nothing for us to do. Repo creation // surfaces via `installation_repositories.added`; archive state isn't part // of the mirror surface in v1.}
async function runMetadataSync(installationId: number, githubRepoId: number): Promise<void> { const session = await restoreSessionForInstallation(installationId) if (!session) return await syncRepoMetadata({ oauthSession: session, installationId, githubRepoId })}
/** * Restore the OAuth session for the user identity bound to `installationId`, * or null if OAuth hasn't completed yet. The OAuth callback re-enqueues work * for all accessible repos on completion, so a null here is a benign drop: a * fresh trigger arrives once the identity exists. */async function restoreSessionForInstallation(installationId: number): Promise<OAuthSession | null> { const db = useDb() const identity = await db.select({ did: userIdentity.did }) .from(userIdentity) .where(sql`${userIdentity.installationId} = ${installationId}`) if (identity.length === 0) return null
return restoreSessionOrNull(identity[0]!.did)}
/** * Whether the `installation` row still exists. Jobs enqueued before an * uninstall can outlive the row (which cascade-deletes user_identity, ssh_key, * repo_mapping); handlers use this to drop such orphaned work cleanly rather * than failing on a foreign-key violation. */async function installationExists(installationId: number): Promise<boolean> { const db = useDb() const rows = await db.select({ id: installation.id }) .from(installation) .where(sql`${installation.id} = ${installationId}`) .limit(1) return rows.length > 0}