| 1 | import { type Database, schema } from '@ghost/db'; |
| 2 | import { |
| 3 | Inject, |
| 4 | Injectable, |
| 5 | Logger, |
| 6 | type OnApplicationBootstrap, |
| 7 | type OnApplicationShutdown, |
| 8 | } from '@nestjs/common'; |
| 9 | import { ConfigService } |
| 10 | import { AuthService } |
| 11 | import { and, eq |
| 12 | |
| 13 | import { DATABASE } |
| 14 | import type { Auth |
| 15 | import { repositoryFullNameOf } |
| 16 | import { |
| 17 | dispatchToImporter, |
| 18 | githubImportConfig, |
| 19 | retryDelayMs, |
| 20 | } from '../../lib/imports/importer.js'; |
| 21 | import { githubAccessToken } |
| 22 | import { StaleImportAttemptError } |
| 23 | |
| 24 | // The importer calls back at least every 30 seconds while it works; four missed beats means it is gone. |
| 25 | const LEASE_MS = 2 |
| 26 | const TICK_MS = 30_000 |
| 27 | const CLAIM_BATCH = 10 |
| 28 | export const MAX_IMPORT_ATTEMPTS = |
| 29 | // Covers the slowest import seen end to end; an attempt that outlives it is retried by lease long before the key matters. |
| 30 | const PUSH_KEY_TTL_SECONDS = 24 |
| 31 | |
| 32 | type Claimed |
| 33 | id: string |
| 34 | repositoryId: string |
| 35 | requestedById: string |
| 36 | source: string |
| 37 | claimToken: string |
| 38 | attempts: number |
| 39 | apiKeyId: string |
| 40 | }; |
| 41 | |
| 42 | export type AttemptOutcome |
| 43 | | { succeeded |
| 44 | | { succeeded |
| 45 | |
| 46 | /** Hands each pending import to the importer and retries it when the importer refuses it, reports a retryable failure, or stops calling back. */ |
| 47 | @Injectable() |
| 48 | export class ImportDispatcherService |
| 49 | implements OnApplicationBootstrap |
| 50 | { |
| 51 | private readonly logger |
| 52 | private timer?: |
| 53 | private ticking?: |
| 54 | private rerun = |
| 55 | private stopped = |
| 56 | |
| 57 | constructor( |
| 58 | @Inject(DATABASE) |
| 59 | private readonly config |
| 60 | private readonly auth |
| 61 | ) {} |
| 62 | |
| 63 | onApplicationBootstrap() { |
| 64 | if (githubImportConfig(this |
| 65 | } |
| 66 | |
| 67 | async onApplicationShutdown() { |
| 68 | this.stopped = |
| 69 | clearTimeout(this.timer |
| 70 | await this.ticking |
| 71 | } |
| 72 | |
| 73 | /** Polls only while an import is unfinished, so an idle instance makes no database traffic and can sleep. */ |
| 74 | wake() { |
| 75 | if (this.stopped |
| 76 | if (this.ticking |
| 77 | this.rerun = |
| 78 | return; |
| 79 | } |
| 80 | clearTimeout(this.timer |
| 81 | this.ticking = |
| 82 | .catch((error: |
| 83 | this.logger.error |
| 84 | `Dispatching imports failed: ${error instanceof |
| 85 | ); |
| 86 | return true; |
| 87 | }) |
| 88 | .then((active) |
| 89 | this.ticking = |
| 90 | if (this.rerun |
| 91 | this.rerun = |
| 92 | this.wake(); |
| 93 | } else if (active |
| 94 | this.timer = |
| 95 | } |
| 96 | }); |
| 97 | } |
| 98 | |
| 99 | /** Returns whether any import is still pending or running. */ |
| 100 | async tick(): |
| 101 | await this.abandonSilent |
| 102 | const claimed = |
| 103 | await Promise.all |
| 104 | |
| 105 | const [active] |
| 106 | .select({ id: |
| 107 | .from(schema.repositoryImport |
| 108 | .where(inArray(schema |
| 109 | .limit(1); |
| 110 | return Boolean(active |
| 111 | } |
| 112 | |
| 113 | /** Every importer callback goes through here: a stale attempt is refused, a current one keeps its lease. */ |
| 114 | async renewLease(importId |
| 115 | const [row] |
| 116 | .update(schema.repositoryImport |
| 117 | .set({ leaseUntil: |
| 118 | .where(this.current |
| 119 | .returning({ repositoryId: |
| 120 | if (!row) |
| 121 | return row.repositoryId |
| 122 | } |
| 123 | |
| 124 | async settle(importId |
| 125 | const [row] |
| 126 | .select({ |
| 127 | repositoryId: schema.repositoryImport |
| 128 | attempts: schema.repositoryImport |
| 129 | apiKeyId: schema.repositoryImport |
| 130 | }) |
| 131 | .from(schema.repositoryImport |
| 132 | .where(this.current |
| 133 | if (!row) |
| 134 | |
| 135 | if (outcome.succeeded |
| 136 | await this.db |
| 137 | await tx |
| 138 | .update(schema.repositoryImport |
| 139 | .set({ status: |
| 140 | .where(this.current |
| 141 | if (outcome.defaultBranch |
| 142 | await tx |
| 143 | .update(schema.repository |
| 144 | .set({ defaultBranch: |
| 145 | .where(eq(schema |
| 146 | } |
| 147 | }); |
| 148 | } else { |
| 149 | const retry = |
| 150 | outcome.retryable && |
| 151 | ? retryDelayMs(row. |
| 152 | : null; |
| 153 | await this.db |
| 154 | .update(schema.repositoryImport |
| 155 | .set({ |
| 156 | status: retry === |
| 157 | nextAttemptAt: sql`now() + ${ |
| 158 | lastError: outcome.error |
| 159 | apiKeyId: null, |
| 160 | }) |
| 161 | .where(this.current |
| 162 | if (retry !== |
| 163 | } |
| 164 | await this.revokeKey |
| 165 | } |
| 166 | |
| 167 | private current(importId |
| 168 | return and( |
| 169 | eq(schema.repositoryImport |
| 170 | eq(schema.repositoryImport |
| 171 | eq(schema.repositoryImport |
| 172 | ); |
| 173 | } |
| 174 | |
| 175 | /** A silent attempt that used its last try fails here; one with tries left is picked up again by `claim`. */ |
| 176 | private async abandonSilent |
| 177 | // Updating from a locked subquery lets `returning` read the key id the update clears. |
| 178 | const { rows |
| 179 | update ${schema.repositoryImport} set |
| 180 | status = 'failed', |
| 181 | last_error = 'The importer stopped responding', |
| 182 | api_key_id = null, |
| 183 | updated_at = now() |
| 184 | from ( |
| 185 | select id, api_key_id from ${schema.repositoryImport} |
| 186 | where status = 'running' and lease_until < now() and attempts >= ${MAX_IMPORT_ATTEMPTS} |
| 187 | for update skip locked |
| 188 | ) as old |
| 189 | where ${schema.repositoryImport.id |
| 190 | returning old.api_key_id as "apiKeyId" |
| 191 | `); |
| 192 | for (const { |
| 193 | } |
| 194 | |
| 195 | private async claim |
| 196 | const due = |
| 197 | (${schema.repositoryImport.status |
| 198 | or (${schema.repositoryImport.status |
| 199 | ) and ${schema.repositoryImport.attempts |
| 200 | |
| 201 | const { rows |
| 202 | update ${schema.repositoryImport} set |
| 203 | status = 'running', |
| 204 | attempts = attempts + 1, |
| 205 | claim_token = gen_random_uuid(), |
| 206 | lease_until = now() + ${LEASE_MS} * interval '1 millisecond', |
| 207 | last_error = case when status = 'running' then 'The importer stopped responding' else last_error end, |
| 208 | updated_at = now() |
| 209 | where id in ( |
| 210 | select id from ${schema.repositoryImport} where |
| 211 | order by next_attempt_at limit ${CLAIM_BATCH} for update skip locked |
| 212 | ) |
| 213 | returning id, repository_id as "repositoryId", requested_by_id as "requestedById", source, claim_token as "claimToken", attempts, api_key_id as "apiKeyId" |
| 214 | `); |
| 215 | return rows; |
| 216 | } |
| 217 | |
| 218 | private async dispatch |
| 219 | const config = |
| 220 | if (!config) |
| 221 | |
| 222 | // A silent attempt's key is still live; the new attempt gets its own. |
| 223 | await this.revokeKey |
| 224 | |
| 225 | try { |
| 226 | const githubToken = |
| 227 | this.db, |
| 228 | this.auth, |
| 229 | row.requestedById, |
| 230 | ); |
| 231 | if (!githubToken) |
| 232 | await this.settle |
| 233 | succeeded: false, |
| 234 | error: |
| 235 | 'The GitHub account that started this import is no longer connected. Connect it again, then retry.', |
| 236 | retryable: false, |
| 237 | }); |
| 238 | return; |
| 239 | } |
| 240 | |
| 241 | const key = |
| 242 | body: { |
| 243 | userId: row.requestedById |
| 244 | name: 'GitHub import', |
| 245 | expiresIn: PUSH_KEY_TTL_SECONDS, |
| 246 | }, |
| 247 | }); |
| 248 | await this.db |
| 249 | .update(schema.repositoryImport |
| 250 | .set({ apiKeyId: |
| 251 | .where(this.current |
| 252 | |
| 253 | await dispatchToImporter(config |
| 254 | importId: row.id |
| 255 | attempt: row.claimToken |
| 256 | source: row.source |
| 257 | destination: await this |
| 258 | githubToken, |
| 259 | ghostToken: key.key |
| 260 | }); |
| 261 | } catch (error) |
| 262 | const reason = |
| 263 | error instanceof Error |
| 264 | ? error.cause |
| 265 | : error; |
| 266 | this.logger.warn |
| 267 | `Import ${row.id |
| 268 | ); |
| 269 | await this.settle |
| 270 | succeeded: false, |
| 271 | error: 'The import could not be started', |
| 272 | retryable: true, |
| 273 | }).catch((settleError |
| 274 | if (!(settleError |
| 275 | throw settleError; |
| 276 | }); |
| 277 | } |
| 278 | } |
| 279 | |
| 280 | private async destinationOf |
| 281 | const [row] |
| 282 | .select({ |
| 283 | fullName: repositoryFullNameOf( |
| 284 | schema.user, |
| 285 | schema.organization, |
| 286 | schema.repository, |
| 287 | ), |
| 288 | }) |
| 289 | .from(schema.repository |
| 290 | .innerJoin(schema.user |
| 291 | .leftJoin( |
| 292 | schema.organization, |
| 293 | eq(schema.organization |
| 294 | ) |
| 295 | .where(eq(schema |
| 296 | return row.fullName |
| 297 | } |
| 298 | |
| 299 | private async revokeKey |
| 300 | if (!apiKeyId) |
| 301 | await this.db |
| 302 | } |
| 303 | } |