| 1 | import type { Env } from "../env.js"; |
| 2 | import { newId, now } from "../lib/util.js"; |
| 3 | |
| 4 | /** A container image name below an owner: diffinery.com/<owner_login>/<name>. */ |
| 5 | export interface PackageRow { |
| 6 | id: string; |
| 7 | owner_id: string; |
| 8 | owner_login: string; |
| 9 | owner_kind: "user" | "org" | "agent"; |
| 10 | name: string; |
| 11 | repo_id: string | null; |
| 12 | repo_name: string | null; |
| 13 | visibility: "private" | "public"; |
| 14 | description: string | null; |
| 15 | created_by: string | null; |
| 16 | created_at: string; |
| 17 | updated_at: string; |
| 18 | } |
| 19 | |
| 20 | export interface ManifestRow { |
| 21 | /** rowid, used as the numeric version id in the REST API. */ |
| 22 | id?: number; |
| 23 | package_id: string; |
| 24 | digest: string; |
| 25 | media_type: string; |
| 26 | artifact_type: string | null; |
| 27 | subject: string | null; |
| 28 | annotations: string | null; |
| 29 | size: number; |
| 30 | total_size: number; |
| 31 | platforms: string | null; |
| 32 | pushed_by: string | null; |
| 33 | created_at: string; |
| 34 | } |
| 35 | |
| 36 | export interface UploadRow { |
| 37 | id: string; |
| 38 | package_id: string; |
| 39 | r2_key: string; |
| 40 | multipart_id: string | null; |
| 41 | parts: string; |
| 42 | size: number; |
| 43 | tail: number; |
| 44 | digest: string | null; |
| 45 | /** SHA-256 progress over the bytes so far (lib/sha256.ts), null once finished or for older uploads. */ |
| 46 | hash_state: string | null; |
| 47 | created_at: string; |
| 48 | updated_at: string; |
| 49 | } |
| 50 | |
| 51 | export interface BlobRow { |
| 52 | digest: string; |
| 53 | size: number; |
| 54 | r2_key: string; |
| 55 | created_at: string; |
| 56 | } |
| 57 | |
| 58 | export const DIGEST_RE = /^sha256:[a-f0-9]{64}$/; |
| 59 | export const TAG_RE = /^[A-Za-z0-9_][A-Za-z0-9._-]{0,127}$/; |
| 60 | /** OCI repository path components below the owner, lowercase only. */ |
| 61 | export const NAME_RE = /^[a-z0-9]+(?:(?:\.|_|__|-+)[a-z0-9]+)*(?:\/[a-z0-9]+(?:(?:\.|_|__|-+)[a-z0-9]+)*)*$/; |
| 62 | |
| 63 | export const manifestKey = (digest: string) => `registry/manifests/${digest.replace(":", "/")}`; |
| 64 | export const blobKey = (uploadId: string) => `registry/blobs/${uploadId}`; |
| 65 | export const tailKey = (uploadId: string) => `registry/uploads/${uploadId}/tail`; |
| 66 | |
| 67 | const PKG_SELECT = "SELECT p.*, a.login AS owner_login, a.kind AS owner_kind, r.name AS repo_name FROM packages p JOIN accounts a ON a.id = p.owner_id LEFT JOIN repos r ON r.id = p.repo_id"; |
| 68 | |
| 69 | /** The registry's index in D1. Bytes live in R2 under registry/. */ |
| 70 | export class RegistryStore { |
| 71 | constructor(private env: Env) {} |
| 72 | |
| 73 | private get db() { |
| 74 | return this.env.DB; |
| 75 | } |
| 76 | |
| 77 | // ---------------------------------------------------------------- packages |
| 78 | |
| 79 | async package(ownerId: string, name: string): Promise<PackageRow | null> { |
| 80 | return this.db.prepare(`${PKG_SELECT} WHERE p.owner_id = ? AND p.name = ?`).bind(ownerId, name).first<PackageRow>(); |
| 81 | } |
| 82 | |
| 83 | async packageById(id: string): Promise<PackageRow | null> { |
| 84 | return this.db.prepare(`${PKG_SELECT} WHERE p.id = ?`).bind(id).first<PackageRow>(); |
| 85 | } |
| 86 | |
| 87 | /** A package that moved to another owner with its repository, by its old <owner>/<name> (see migrations/0011_redirects.sql). */ |
| 88 | async movedPackage(ownerLogin: string, name: string): Promise<PackageRow | null> { |
| 89 | const r = await this.db.prepare("SELECT package_id FROM package_redirects WHERE owner_login = ? AND name = ?").bind(ownerLogin, name).first<{ package_id: string }>(); |
| 90 | return r ? this.packageById(r.package_id) : null; |
| 91 | } |
| 92 | |
| 93 | /** Creates the package unless a concurrent push just did; returns it either way. */ |
| 94 | async createPackage(ownerId: string, name: string, repoId: string | null, createdBy: string): Promise<PackageRow> { |
| 95 | const t = now(); |
| 96 | await this.db |
| 97 | .prepare("INSERT INTO packages (id, owner_id, name, repo_id, visibility, created_by, created_at, updated_at) VALUES (?, ?, ?, ?, 'private', ?, ?, ?) ON CONFLICT(owner_id, name) DO NOTHING") |
| 98 | .bind(newId(), ownerId, name, repoId, createdBy, t, t) |
| 99 | .run(); |
| 100 | return (await this.package(ownerId, name))!; |
| 101 | } |
| 102 | |
| 103 | async packages(filter: { ownerIds?: string[]; repoIds?: string[]; repoId?: string; all?: boolean; limit?: number }): Promise<Array<PackageRow & { tag_count: number; version_count: number }>> { |
| 104 | const where: string[] = []; |
| 105 | const params: unknown[] = []; |
| 106 | if (filter.repoId) { |
| 107 | where.push("p.repo_id = ?"); |
| 108 | params.push(filter.repoId); |
| 109 | } else if (!filter.all) { |
| 110 | const ors: string[] = []; |
| 111 | if (filter.ownerIds?.length) { |
| 112 | ors.push(`p.owner_id IN (${filter.ownerIds.map(() => "?").join(",")})`); |
| 113 | params.push(...filter.ownerIds); |
| 114 | } |
| 115 | if (filter.repoIds?.length) { |
| 116 | ors.push(`p.repo_id IN (${filter.repoIds.map(() => "?").join(",")})`); |
| 117 | params.push(...filter.repoIds); |
| 118 | } |
| 119 | ors.push("p.visibility = 'public'"); |
| 120 | where.push(`(${ors.join(" OR ")})`); |
| 121 | } |
| 122 | const q = `SELECT p.*, a.login AS owner_login, a.kind AS owner_kind, r.name AS repo_name, |
| 123 | (SELECT count(*) FROM package_tags t WHERE t.package_id = p.id) AS tag_count, |
| 124 | (SELECT count(*) FROM package_manifests m WHERE m.package_id = p.id) AS version_count |
| 125 | FROM packages p JOIN accounts a ON a.id = p.owner_id LEFT JOIN repos r ON r.id = p.repo_id |
| 126 | ${where.length ? `WHERE ${where.join(" AND ")}` : ""} ORDER BY p.updated_at DESC LIMIT ?`; |
| 127 | params.push(filter.limit ?? 500); |
| 128 | return (await this.db.prepare(q).bind(...params).all<PackageRow & { tag_count: number; version_count: number }>()).results; |
| 129 | } |
| 130 | |
| 131 | async updatePackage(id: string, patch: { visibility?: PackageRow["visibility"]; repoId?: string | null; description?: string | null }): Promise<void> { |
| 132 | const sets: string[] = ["updated_at = ?"]; |
| 133 | const params: unknown[] = [now()]; |
| 134 | if (patch.visibility) { |
| 135 | sets.push("visibility = ?"); |
| 136 | params.push(patch.visibility); |
| 137 | } |
| 138 | if (patch.repoId !== undefined) { |
| 139 | sets.push("repo_id = ?"); |
| 140 | params.push(patch.repoId); |
| 141 | } |
| 142 | if (patch.description !== undefined) { |
| 143 | sets.push("description = ?"); |
| 144 | params.push(patch.description); |
| 145 | } |
| 146 | await this.db.prepare(`UPDATE packages SET ${sets.join(", ")} WHERE id = ?`).bind(...params, id).run(); |
| 147 | } |
| 148 | |
| 149 | async touchPackage(id: string): Promise<void> { |
| 150 | await this.db.prepare("UPDATE packages SET updated_at = ? WHERE id = ?").bind(now(), id).run(); |
| 151 | } |
| 152 | |
| 153 | /** Deletes a package and everything it indexes; unreferenced bytes go at the next cleanup. */ |
| 154 | async deletePackage(id: string): Promise<void> { |
| 155 | await this.db.batch([ |
| 156 | this.db.prepare("DELETE FROM package_tags WHERE package_id = ?").bind(id), |
| 157 | this.db.prepare("DELETE FROM package_manifests WHERE package_id = ?").bind(id), |
| 158 | this.db.prepare("DELETE FROM package_manifest_blobs WHERE package_id = ?").bind(id), |
| 159 | this.db.prepare("DELETE FROM package_blobs WHERE package_id = ?").bind(id), |
| 160 | this.db.prepare("DELETE FROM registry_uploads WHERE package_id = ?").bind(id), |
| 161 | this.db.prepare("DELETE FROM packages WHERE id = ?").bind(id), |
| 162 | ]); |
| 163 | } |
| 164 | |
| 165 | // ---------------------------------------------------------------- blobs |
| 166 | |
| 167 | async blob(digest: string): Promise<BlobRow | null> { |
| 168 | return this.db.prepare("SELECT * FROM registry_blobs WHERE digest = ?").bind(digest).first<BlobRow>(); |
| 169 | } |
| 170 | |
| 171 | /** The blob, if this package holds it. */ |
| 172 | async packageBlob(packageId: string, digest: string): Promise<BlobRow | null> { |
| 173 | return this.db.prepare("SELECT b.* FROM package_blobs pb JOIN registry_blobs b ON b.digest = pb.digest WHERE pb.package_id = ? AND pb.digest = ?").bind(packageId, digest).first<BlobRow>(); |
| 174 | } |
| 175 | |
| 176 | async packageBlobs(packageId: string, digests: string[]): Promise<Set<string>> { |
| 177 | if (!digests.length) return new Set(); |
| 178 | const found = new Set<string>(); |
| 179 | for (let i = 0; i < digests.length; i += 90) { |
| 180 | const chunk = digests.slice(i, i + 90); |
| 181 | const r = await this.db.prepare(`SELECT digest FROM package_blobs WHERE package_id = ? AND digest IN (${chunk.map(() => "?").join(",")})`).bind(packageId, ...chunk).all<{ digest: string }>(); |
| 182 | for (const x of r.results) found.add(x.digest); |
| 183 | } |
| 184 | return found; |
| 185 | } |
| 186 | |
| 187 | /** Records a verified blob (keeping an existing copy if there is one) and returns the stored row. */ |
| 188 | async addBlob(digest: string, size: number, r2Key: string): Promise<BlobRow> { |
| 189 | await this.db.prepare("INSERT INTO registry_blobs (digest, size, r2_key, created_at) VALUES (?, ?, ?, ?) ON CONFLICT(digest) DO NOTHING").bind(digest, size, r2Key, now()).run(); |
| 190 | return (await this.blob(digest))!; |
| 191 | } |
| 192 | |
| 193 | async linkBlob(packageId: string, digest: string): Promise<void> { |
| 194 | await this.db.prepare("INSERT INTO package_blobs (package_id, digest, created_at) VALUES (?, ?, ?) ON CONFLICT DO NOTHING").bind(packageId, digest, now()).run(); |
| 195 | } |
| 196 | |
| 197 | async unlinkBlob(packageId: string, digest: string): Promise<void> { |
| 198 | await this.db.prepare("DELETE FROM package_blobs WHERE package_id = ? AND digest = ?").bind(packageId, digest).run(); |
| 199 | } |
| 200 | |
| 201 | // ---------------------------------------------------------------- uploads |
| 202 | |
| 203 | async createUpload(packageId: string): Promise<UploadRow> { |
| 204 | const id = newId(); |
| 205 | const t = now(); |
| 206 | const row: UploadRow = { id, package_id: packageId, r2_key: blobKey(id), multipart_id: null, parts: "[]", size: 0, tail: 0, digest: null, hash_state: null, created_at: t, updated_at: t }; |
| 207 | await this.db.prepare("INSERT INTO registry_uploads (id, package_id, r2_key, parts, size, tail, created_at, updated_at) VALUES (?, ?, ?, '[]', 0, 0, ?, ?)").bind(id, packageId, row.r2_key, t, t).run(); |
| 208 | return row; |
| 209 | } |
| 210 | |
| 211 | async upload(id: string): Promise<UploadRow | null> { |
| 212 | return this.db.prepare("SELECT * FROM registry_uploads WHERE id = ?").bind(id).first<UploadRow>(); |
| 213 | } |
| 214 | |
| 215 | async saveUpload(u: UploadRow): Promise<void> { |
| 216 | await this.db |
| 217 | .prepare("UPDATE registry_uploads SET multipart_id = ?, parts = ?, size = ?, tail = ?, digest = ?, hash_state = ?, updated_at = ? WHERE id = ?") |
| 218 | .bind(u.multipart_id, u.parts, u.size, u.tail, u.digest, u.hash_state, now(), u.id) |
| 219 | .run(); |
| 220 | } |
| 221 | |
| 222 | async deleteUpload(id: string): Promise<void> { |
| 223 | await this.db.prepare("DELETE FROM registry_uploads WHERE id = ?").bind(id).run(); |
| 224 | } |
| 225 | |
| 226 | // ---------------------------------------------------------------- manifests and tags |
| 227 | |
| 228 | async manifest(packageId: string, digest: string): Promise<ManifestRow | null> { |
| 229 | return this.db.prepare("SELECT * FROM package_manifests WHERE package_id = ? AND digest = ?").bind(packageId, digest).first<ManifestRow>(); |
| 230 | } |
| 231 | |
| 232 | async tagDigest(packageId: string, tag: string): Promise<string | null> { |
| 233 | return (await this.db.prepare("SELECT digest FROM package_tags WHERE package_id = ? AND tag = ?").bind(packageId, tag).first<{ digest: string }>())?.digest ?? null; |
| 234 | } |
| 235 | |
| 236 | /** Which of these manifests the package holds, with the size of each one's content. */ |
| 237 | async manifestSizes(packageId: string, digests: string[]): Promise<Map<string, number>> { |
| 238 | const found = new Map<string, number>(); |
| 239 | for (let i = 0; i < digests.length; i += 90) { |
| 240 | const chunk = digests.slice(i, i + 90); |
| 241 | const r = await this.db.prepare(`SELECT digest, total_size FROM package_manifests WHERE package_id = ? AND digest IN (${chunk.map(() => "?").join(",")})`).bind(packageId, ...chunk).all<{ digest: string; total_size: number }>(); |
| 242 | for (const x of r.results) found.set(x.digest, x.total_size); |
| 243 | } |
| 244 | return found; |
| 245 | } |
| 246 | |
| 247 | async putManifest(m: ManifestRow, tag: string | null, blobs: string[]): Promise<void> { |
| 248 | const t = now(); |
| 249 | const stmts = [ |
| 250 | ...blobs.map((d) => this.db.prepare("INSERT INTO package_manifest_blobs (package_id, manifest, digest) VALUES (?, ?, ?) ON CONFLICT DO NOTHING").bind(m.package_id, m.digest, d)), |
| 251 | this.db |
| 252 | .prepare("INSERT INTO package_manifests (package_id, digest, media_type, artifact_type, subject, annotations, size, total_size, platforms, pushed_by, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(package_id, digest) DO NOTHING") |
| 253 | .bind(m.package_id, m.digest, m.media_type, m.artifact_type, m.subject, m.annotations, m.size, m.total_size, m.platforms, m.pushed_by, m.created_at), |
| 254 | this.db.prepare("UPDATE packages SET updated_at = ? WHERE id = ?").bind(t, m.package_id), |
| 255 | ]; |
| 256 | if (tag) stmts.push(this.db.prepare("INSERT INTO package_tags (package_id, tag, digest, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(package_id, tag) DO UPDATE SET digest = excluded.digest, updated_at = excluded.updated_at").bind(m.package_id, tag, m.digest, t)); |
| 257 | await this.db.batch(stmts); |
| 258 | } |
| 259 | |
| 260 | async tags(packageId: string, opts: { n?: number; last?: string } = {}): Promise<string[]> { |
| 261 | const r = await this.db |
| 262 | .prepare("SELECT tag FROM package_tags WHERE package_id = ? AND tag > ? ORDER BY tag LIMIT ?") |
| 263 | .bind(packageId, opts.last ?? "", Math.min(Math.max(opts.n ?? 10000, 0), 10000)) |
| 264 | .all<{ tag: string }>(); |
| 265 | return r.results.map((x) => x.tag); |
| 266 | } |
| 267 | |
| 268 | async deleteTag(packageId: string, tag: string): Promise<boolean> { |
| 269 | const r = await this.db.prepare("DELETE FROM package_tags WHERE package_id = ? AND tag = ?").bind(packageId, tag).run(); |
| 270 | return (r.meta.changes ?? 0) > 0; |
| 271 | } |
| 272 | |
| 273 | /** Deletes a manifest and the tags pointing at it. */ |
| 274 | async deleteManifest(packageId: string, digest: string): Promise<boolean> { |
| 275 | const [, r] = await this.db.batch([ |
| 276 | this.db.prepare("DELETE FROM package_tags WHERE package_id = ? AND digest = ?").bind(packageId, digest), |
| 277 | this.db.prepare("DELETE FROM package_manifests WHERE package_id = ? AND digest = ?").bind(packageId, digest), |
| 278 | this.db.prepare("DELETE FROM package_manifest_blobs WHERE package_id = ? AND manifest = ?").bind(packageId, digest), |
| 279 | ]); |
| 280 | return (r!.meta.changes ?? 0) > 0; |
| 281 | } |
| 282 | |
| 283 | async versionById(packageId: string, id: number): Promise<(ManifestRow & { tags: string[] }) | null> { |
| 284 | const m = await this.db.prepare("SELECT rowid AS id, * FROM package_manifests WHERE package_id = ? AND rowid = ?").bind(packageId, id).first<ManifestRow>(); |
| 285 | if (!m) return null; |
| 286 | const tags = (await this.db.prepare("SELECT tag FROM package_tags WHERE package_id = ? AND digest = ? ORDER BY tag").bind(packageId, m.digest).all<{ tag: string }>()).results.map((t) => t.tag); |
| 287 | return { ...m, tags }; |
| 288 | } |
| 289 | |
| 290 | async referrers(packageId: string, subject: string): Promise<ManifestRow[]> { |
| 291 | return (await this.db.prepare("SELECT * FROM package_manifests WHERE package_id = ? AND subject = ? ORDER BY created_at").bind(packageId, subject).all<ManifestRow>()).results; |
| 292 | } |
| 293 | |
| 294 | /** Versions (manifests) with their tags, newest first. Children of multi-platform images are left out. */ |
| 295 | async versions(packageId: string, opts: { limit?: number; all?: boolean } = {}): Promise<Array<ManifestRow & { tags: string[] }>> { |
| 296 | const manifests = (await this.db.prepare("SELECT rowid AS id, * FROM package_manifests WHERE package_id = ? ORDER BY created_at DESC LIMIT ?").bind(packageId, opts.limit ?? 1000).all<ManifestRow>()).results; |
| 297 | const tags = (await this.db.prepare("SELECT tag, digest FROM package_tags WHERE package_id = ? ORDER BY updated_at DESC").bind(packageId).all<{ tag: string; digest: string }>()).results; |
| 298 | const byDigest = new Map<string, string[]>(); |
| 299 | for (const t of tags) byDigest.set(t.digest, [...(byDigest.get(t.digest) ?? []), t.tag]); |
| 300 | const children = new Set<string>(); |
| 301 | if (!opts.all) { |
| 302 | for (const m of manifests) { |
| 303 | for (const d of (m.platforms ? (JSON.parse(m.platforms) as Array<{ digest?: string }>) : []).map((p) => p.digest)) if (d) children.add(d); |
| 304 | if (m.subject) children.add(m.digest); |
| 305 | } |
| 306 | } |
| 307 | return manifests.filter((m) => byDigest.has(m.digest) || !children.has(m.digest)).map((m) => ({ ...m, tags: byDigest.get(m.digest) ?? [] })); |
| 308 | } |
| 309 | |
| 310 | // ---------------------------------------------------------------- cleanup |
| 311 | |
| 312 | /** |
| 313 | * Removes abandoned uploads and bytes nothing references any more. Runs from the |
| 314 | * cron trigger; a grace period keeps blobs that a push in progress is about to use. |
| 315 | */ |
| 316 | async cleanup(limit = 200): Promise<{ uploads: number; blobs: number; manifests: number }> { |
| 317 | const day = new Date(Date.now() - 86400_000).toISOString(); |
| 318 | const stale = (await this.db.prepare("SELECT * FROM registry_uploads WHERE updated_at < ? LIMIT ?").bind(day, limit).all<UploadRow>()).results; |
| 319 | for (const u of stale) { |
| 320 | if (u.multipart_id) await this.env.BUCKET.resumeMultipartUpload(u.r2_key, u.multipart_id).abort().catch(() => {}); |
| 321 | await this.env.BUCKET.delete([u.r2_key, tailKey(u.id)]); |
| 322 | await this.deleteUpload(u.id); |
| 323 | } |
| 324 | // Blobs a package holds that none of its manifests use any more (deleted versions, abandoned pushes). |
| 325 | await this.db |
| 326 | .prepare("DELETE FROM package_blobs WHERE rowid IN (SELECT pb.rowid FROM package_blobs pb WHERE pb.created_at < ? AND NOT EXISTS (SELECT 1 FROM package_manifest_blobs mb WHERE mb.package_id = pb.package_id AND mb.digest = pb.digest) LIMIT ?)") |
| 327 | .bind(day, limit * 5) |
| 328 | .run(); |
| 329 | // Blobs no package holds, past the grace period. |
| 330 | const blobs = (await this.db.prepare("SELECT b.* FROM registry_blobs b WHERE b.created_at < ? AND NOT EXISTS (SELECT 1 FROM package_blobs pb WHERE pb.digest = b.digest) LIMIT ?").bind(day, limit).all<BlobRow>()).results; |
| 331 | for (const b of blobs) { |
| 332 | await this.db.prepare("DELETE FROM registry_blobs WHERE digest = ? AND NOT EXISTS (SELECT 1 FROM package_blobs pb WHERE pb.digest = ?)").bind(b.digest, b.digest).run(); |
| 333 | if (!(await this.blob(b.digest))) await this.env.BUCKET.delete(b.r2_key); |
| 334 | } |
| 335 | // Manifest bytes no package lists. They are tiny, so only the R2 listing costs anything; do a page per run. |
| 336 | let manifests = 0; |
| 337 | // A random shard each run, so every manifest gets looked at over time. |
| 338 | const shard = [...crypto.getRandomValues(new Uint8Array(1))].map((b) => b.toString(16).padStart(2, "0")).join(""); |
| 339 | const listed = await this.env.BUCKET.list({ prefix: `registry/manifests/sha256/${shard}`, limit: 500 }); |
| 340 | const old = listed.objects.filter((o) => o.uploaded.toISOString() < day); |
| 341 | if (old.length) { |
| 342 | const digests = old.map((o) => o.key.slice("registry/manifests/".length).replace("/", ":")); |
| 343 | const inUse = new Set<string>(); |
| 344 | for (let i = 0; i < digests.length; i += 90) { |
| 345 | const chunk = digests.slice(i, i + 90); |
| 346 | const r = await this.db.prepare(`SELECT DISTINCT digest FROM package_manifests WHERE digest IN (${chunk.map(() => "?").join(",")})`).bind(...chunk).all<{ digest: string }>(); |
| 347 | for (const x of r.results) inUse.add(x.digest); |
| 348 | } |
| 349 | const unused = old.filter((_, i) => !inUse.has(digests[i]!)).map((o) => o.key); |
| 350 | if (unused.length) await this.env.BUCKET.delete(unused); |
| 351 | manifests = unused.length; |
| 352 | } |
| 353 | return { uploads: stale.length, blobs: blobs.length, manifests }; |
| 354 | } |
| 355 | } |