| 1 | import { type Database, schema } from '@ghost/db'; |
| 2 | import { Inject, Injectable |
| 3 | import { eq } |
| 4 | |
| 5 | import { DATABASE } |
| 6 | import { WalStoreService } |
| 7 | |
| 8 | // Three columns per row, well under Postgres' 65535 bind parameters. |
| 9 | const INSERT_BATCH = 5000 |
| 10 | |
| 11 | export interface LogEntryRow |
| 12 | repositoryId: string; |
| 13 | ulid: string; |
| 14 | size: number; |
| 15 | } |
| 16 | |
| 17 | /** Bills log entries written before pushes were recorded (0035). */ |
| 18 | @Injectable() |
| 19 | export class RepositoryLogBackfillService |
| 20 | constructor( |
| 21 | @Inject(DATABASE) |
| 22 | private readonly wal |
| 23 | ) {} |
| 24 | |
| 25 | /** Entries of the repository's log with no billing row. Entries Ghost wrote for pull requests are billed through their own rows (0033), so they are left out. */ |
| 26 | async missingEntries(repositoryId |
| 27 | // Read before the rows: a pull ref write records its intent before its entry reaches the index, so every such entry this index holds is already in the rows below. |
| 28 | const stored = |
| 29 | if (!stored) |
| 30 | |
| 31 | const [billed, |
| 32 | this.db |
| 33 | .select({ ulid: |
| 34 | .from(schema.repositoryLogEntry |
| 35 | .where(eq(schema |
| 36 | this.db |
| 37 | .select({ id: |
| 38 | .from(schema.pullRequestRefWrite |
| 39 | .innerJoin( |
| 40 | schema.pullRequest, |
| 41 | eq(schema.pullRequest |
| 42 | ) |
| 43 | .where(eq(schema |
| 44 | this.db |
| 45 | .select({ id: |
| 46 | .from(schema.pullRequestRefWritePending |
| 47 | .innerJoin( |
| 48 | schema.pullRequest, |
| 49 | eq( |
| 50 | schema.pullRequest.id |
| 51 | schema.pullRequestRefWritePending.pullRequestId |
| 52 | ), |
| 53 | ) |
| 54 | .where(eq(schema |
| 55 | ]); |
| 56 | const skip = |
| 57 | ...billed.map( |
| 58 | ...pullRefWrites.map( |
| 59 | ...pendingPullRefWrites.map |
| 60 | ]); |
| 61 | |
| 62 | return stored.index |
| 63 | .filter(({ ulid |
| 64 | .map(({ ulid |
| 65 | } |
| 66 | |
| 67 | /** Bytes actually recorded: a row a push recorded meanwhile is skipped and not counted. */ |
| 68 | async record(rows |
| 69 | let recorded = |
| 70 | for (let start |
| 71 | const inserted = |
| 72 | .insert(schema.repositoryLogEntry |
| 73 | .values(rows.slice |
| 74 | .onConflictDoNothing() |
| 75 | .returning({ size: |
| 76 | recorded += inserted. |
| 77 | } |
| 78 | return recorded; |
| 79 | } |
| 80 | } |