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