1import * as Cloudflare from "alchemy/Cloudflare";
2import * as SQL from "alchemy/SQL/D1";
3import { sha256 } from "alchemy/Util/sha256";
4import * as Clock from "effect/Clock";
5import * as Duration from "effect/Duration";
6import * as Effect from "effect/Effect";
7import { Hex } from "effect/encoding";
8import * as Layer from "effect/Layer";
9import * as Path from "effect/Path";
10import * as Result from "effect/Result";
11import * as Schema from "effect/Schema";
12import * as Etag from "effect/http/Etag";
13import * as FetchHttpClient from "effect/http/FetchHttpClient";
14import * as HttpPlatform from "effect/http/HttpPlatform";
15import * as HttpRouter from "effect/http/HttpRouter";
16import { HttpServerRequest } from "effect/http/HttpServerRequest";
17import * as HttpServerResponse from "effect/http/HttpServerResponse";
18import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
19import { ManifestJson, type ManifestPackage } from "../Manifest.ts";
20import {
21 BadRequest,
22 Forbidden,
23 manifestArtifactName,
24 MissingTarballs,
25 PackageTooLarge,
26 PkgApi,
27 RunNotInProgress,
28 RunRef,
29 Upstream,
30} from "../Protocol.ts";
31import {
32 Bucket,
33 Cache,
34 COMMENT_MARKER,
35 Database,
36 RegistryConfig,
37 tarballKey,
38} from "./Bindings.ts";
39import * as GitHub from "./GitHub.ts";
40import * as KVCache from "./KVCache.ts";
41import { sweep } from "./Sweep.ts";
42import * as Tags from "./Tags.ts";
43
44/** What GitHub says about a run. */
45const Run = Schema.Struct({
46 ...RunRef.fields,
47 headSha: Schema.String,
48 headBranch: Schema.NullOr(Schema.String),
49 headRepo: Schema.String,
50 pr: Schema.NullOr(Schema.Number),
51});
52type Run = typeof Run.Type;
53
54const bindings = Layer.mergeAll(
55 Cloudflare.R2.ReadWriteBucketBinding,
56 Cloudflare.D1.QueryDatabaseBinding,
57 Cloudflare.KV.ReadWriteNamespaceBinding,
58 Cloudflare.Workers.CronEventSourceLive,
59 FetchHttpClient.layer,
60);
61
62const SHORT = 7;
63
64/** How long a resolved run is reused for uploads before asking GitHub again. */
65const RUN_CACHE = Duration.minutes(1);
66
67const CHECK_NAME = "Preview packages";
68
69/** The public origin a request arrived on. */
70const origin = Effect.map(
71 HttpServerRequest,
72 (request) => `https://${request.headers.host ?? "localhost"}`,
73);
74
75/**
76 * A run from a fork: its head commit belongs to another repository, so its
77 * workflow file, and therefore everything it packs, is the fork's.
78 */
79const isFork = (run: Run) => run.headRepo !== run.repo;
80
81/**
82 * Tags every package in a publication receives, all derived from the run.
83 * Runs on the repository's own commits get the head commit, the short
84 * commit, `branch:<name>`, and `pr:N` plus `pr:N:<short-sha>` for pull requests.
85 * A fork's run gets only `pr:N:<short-sha>`: commit tags are shared by every publisher, and a fork can
86 * run any commit it likes, including one already published from the
87 * repository, so letting it write them would let it repoint them.
88 */
89export const tagsFor = (run: Run): string[] => {
90 if (isFork(run)) {
91 return run.pr === null ? [] : [installTag(run)];
92 }
93 const tags = [run.headSha, run.headSha.slice(0, SHORT)];
94 if (run.pr !== null) tags.push(`pr:${run.pr}`, installTag(run));
95 if (run.headBranch) tags.push(`branch:${run.headBranch}`);
96 return tags;
97};
98
99/** The tag install commands are written against. */
100export const installTag = (run: Run) =>
101 run.pr !== null
102 ? `pr:${run.pr}:${run.headSha.slice(0, SHORT)}`
103 : run.headSha.slice(0, SHORT);
104
105const upstream = (e: { readonly _tag: string; readonly message?: string }) =>
106 new Upstream({ message: GitHub.describe(e) });
107
108/**
109 * Storage failures are not part of the protocol: they are defects, which
110 * the router answers with a 500 after logging the cause.
111 */
112const storageFailures = ["R2Error", "SqlError", "SchemaError"] as const;
113
114/**
115 * Resolve the run a request names through GitHub; nothing about it is
116 * trusted from the client. Only allowed repositories are looked up at all,
117 * and the run must be in progress, since requests come from inside it.
118 */
119const lookupRun = Effect.fn("lookupRun")(function* (ref: RunRef) {
120 const { policy } = yield* RegistryConfig;
121 const github = yield* GitHub.GitHubApp;
122
123 if (!policy.repos.includes(ref.repo)) {
124 return yield* new Forbidden({ message: `${ref.repo} may not publish` });
125 }
126 const data = yield* github
127 .getRun(ref.repo, ref.runId)
128 .pipe(Effect.mapError(upstream));
129 if (
130 data.status !== "in_progress" ||
131 (data.run_attempt !== undefined && data.run_attempt !== ref.attempt)
132 ) {
133 return yield* new RunNotInProgress({ message: "run is not in progress" });
134 }
135 if (data.event !== "push" && data.event !== "pull_request") {
136 return yield* new BadRequest({
137 message: `unsupported event ${data.event}`,
138 });
139 }
140 const headRepo = data.head_repository?.full_name ?? data.repository.full_name;
141 let pr: number | null = null;
142 if (data.event === "pull_request") {
143 if (data.head_branch === null) {
144 return yield* new BadRequest({
145 message: `pull request run ${ref.runId} has no head branch`,
146 });
147 }
148 // The pull request whose head is this run's head, from the repository
149 // the run's head lives in. The same commit can head several pull
150 // requests, including one opened from a fork against a commit the
151 // repository already published; matching the head repository keeps a
152 // run's publication on its own pull request.
153 const pulls = yield* github
154 .pullRequestsForCommit(
155 ref.repo,
156 headRepo,
157 data.head_branch,
158 data.head_sha,
159 )
160 .pipe(Effect.mapError(upstream));
161 const first = pulls[0];
162 if (first === undefined) {
163 return yield* new BadRequest({
164 message: `no open pull request from ${headRepo} has head ${data.head_sha}`,
165 });
166 }
167 pr = first.number;
168 } else if (headRepo !== ref.repo) {
169 return yield* new BadRequest({
170 message: `push run ${ref.runId} is for ${headRepo}, not ${ref.repo}`,
171 });
172 }
173 return {
174 ...ref,
175 headSha: data.head_sha,
176 headBranch: data.head_branch,
177 headRepo,
178 pr,
179 } satisfies Run;
180});
181
182/**
183 * The proof. The job uploaded the manifest as an artifact of its run,
184 * named by the manifest's hash. Only the job can add artifacts to the
185 * run, and GitHub reports the run's artifacts to the App, so a matching
186 * name means this run vouched for exactly this manifest text.
187 */
188const requireVouched = Effect.fn("requireVouched")(function* (
189 run: Run,
190 manifestText: string,
191) {
192 const github = yield* GitHub.GitHubApp;
193 const expected = manifestArtifactName(yield* sha256(manifestText));
194 const artifacts = yield* github
195 .listRunArtifacts(run.repo, run.runId, expected)
196 .pipe(Effect.mapError(upstream));
197 if (!artifacts.some((a) => a.name === expected && !a.expired)) {
198 return yield* new Forbidden({
199 message: `run ${run.runId} has not vouched for this manifest (no artifact ${expected})`,
200 });
201 }
202});
203
204const publish = Effect.fn("publish")(function* (
205 run: Run,
206 manifestText: string,
207) {
208 const { policy } = yield* RegistryConfig;
209 const r2 = yield* Cloudflare.R2.ReadWriteBucket(Bucket);
210 const github = yield* GitHub.GitHubApp;
211 const base = yield* origin;
212
213 yield* requireVouched(run, manifestText);
214 const manifest = yield* Schema.decodeUnknownEffect(ManifestJson)(
215 manifestText,
216 ).pipe(
217 Effect.mapError(
218 (e) => new BadRequest({ message: `invalid manifest: ${String(e)}` }),
219 ),
220 );
221 // The run vouched for the manifest, but the tags come from the run's
222 // head, so the packed checkout has to be that commit: a `pull_request`
223 // job that packs the synthetic merge commit would otherwise publish bytes
224 // built from a commit that exists nowhere under the commit's name.
225 if (manifest.head !== run.headSha) {
226 return yield* new BadRequest({
227 message: `manifest was packed at ${manifest.head}, but the run's head is ${run.headSha}`,
228 });
229 }
230 const packages = manifest.packages;
231 const maxSize = policy.maxPackageSize;
232 const tooLarge =
233 maxSize === undefined
234 ? undefined
235 : packages.find((pkg) => BigInt(pkg.size) > maxSize);
236 if (tooLarge !== undefined) {
237 return yield* new PackageTooLarge({
238 message: `${tooLarge.name} exceeds ${maxSize} bytes`,
239 });
240 }
241
242 const missing = yield* Effect.filter(
243 packages,
244 (pkg) =>
245 r2
246 .head(tarballKey(pkg.name, pkg.sha256))
247 .pipe(Effect.map((object) => object === null)),
248 { concurrency: 8 },
249 ).pipe(
250 Effect.map((packages) =>
251 packages.map(({ name, sha256 }) => ({ name, sha256 })),
252 ),
253 );
254 if (missing.length > 0) {
255 return yield* new MissingTarballs({ missing });
256 }
257
258 const now = yield* Clock.currentTimeMillis;
259 const expiresAt = now + Duration.toMillis(policy.ttl);
260 const prs = run.pr !== null ? [`${run.repo}#${run.pr}`] : [];
261 const tags = tagsFor(run);
262 const tag = installTag(run);
263 const published = yield* Effect.forEach(
264 packages,
265 Effect.fn(function* (pkg: ManifestPackage) {
266 yield* Effect.forEach(
267 tags,
268 (tag) =>
269 Tags.upsert({
270 package: pkg.name,
271 tag,
272 sha256: pkg.sha256,
273 expiresAt,
274 prs,
275 }),
276 { discard: true },
277 );
278 return {
279 name: pkg.name,
280 group: pkg.group,
281 url: `${base}/${pkg.name}/${tag}`,
282 tags,
283 };
284 }),
285 { concurrency: 4 },
286 );
287
288 // A check on the commit itself, so the install lines are visible on
289 // pushes and on fork pull requests alike.
290 yield* github
291 .createCheckRun(run.repo, {
292 headSha: run.headSha,
293 name: CHECK_NAME,
294 title: `${packages.length} package(s) published`,
295 summary: GitHub.renderInstalls(published, manifest.groups),
296 detailsUrl: base,
297 })
298 .pipe(
299 Effect.catch((e) =>
300 Effect.logWarning(
301 `check run on ${run.repo}@${run.headSha} failed: ${GitHub.describe(e)}`,
302 ),
303 ),
304 );
305 if (run.pr !== null) {
306 const body = GitHub.renderComment(published, manifest.groups, {
307 publishedAt: now,
308 expiresAt,
309 });
310 yield* github
311 .upsertComment(run.repo, run.pr, COMMENT_MARKER, body)
312 .pipe(
313 Effect.catch((e) =>
314 Effect.logWarning(
315 `comment on ${run.repo}#${run.pr} failed: ${GitHub.describe(e)}`,
316 ),
317 ),
318 );
319 }
320 return { packages: published };
321});
322
323/**
324 * Content-addressed upload, streamed straight into R2 with the hash
325 * verified by R2 itself. Untagged objects are swept after a day.
326 */
327const uploadTarball = Effect.fn("uploadTarball")(function* (
328 request: HttpServerRequest,
329 name: string,
330 sha256: string,
331) {
332 const { policy } = yield* RegistryConfig;
333 const r2 = yield* Cloudflare.R2.ReadWriteBucket(Bucket);
334
335 const contentLength = Number(request.headers["content-length"] ?? 0);
336 if (!Number.isSafeInteger(contentLength) || contentLength <= 0) {
337 return yield* new BadRequest({ message: "Content-Length is required" });
338 }
339 const maxSize = policy.maxPackageSize;
340 if (maxSize !== undefined && BigInt(contentLength) > maxSize) {
341 return yield* new PackageTooLarge({
342 message: `tarball exceeds ${maxSize} bytes`,
343 });
344 }
345 const key = tarballKey(name, sha256);
346 const existing = yield* r2.head(key);
347 if (existing !== null) {
348 return { name, sha256, size: existing.size, uploaded: false };
349 }
350 yield* r2
351 .put(key, request.stream, {
352 contentLength,
353 // The params schema already constrained this to 64 hex chars.
354 sha256: Result.getOrThrow(Hex.decode(sha256)),
355 })
356 .pipe(
357 Effect.mapError(
358 (e) => new BadRequest({ message: `upload rejected: ${String(e)}` }),
359 ),
360 );
361 return { name, sha256, size: contentLength, uploaded: true };
362});
363
364/**
365 * Parse `/<name>/<tag>` and `/<name>/-/<sha256>.tgz`. Scoped names take
366 * two segments; the tag is everything after the name and may itself contain
367 * `/` and `:`.
368 */
369export const parseInstallPath = (
370 pathname: string,
371 scope: string | undefined,
372):
373 | { kind: "tag"; name: string; tag: string }
374 | { kind: "tarball"; name: string; sha256: string }
375 | undefined => {
376 let path: string;
377 try {
378 path = decodeURIComponent(pathname.replace(/^\/+/, ""));
379 } catch {
380 // Malformed percent-encoding is a path nothing can be published under.
381 return undefined;
382 }
383 const tarball = path.match(
384 /^(@[^/]+\/[^/@]+|[^/@]+)\/-\/([a-f0-9]{64})\.tgz$/,
385 );
386 if (tarball) {
387 return {
388 kind: "tarball",
389 name: qualify(tarball[1]!, scope),
390 sha256: tarball[2]!,
391 };
392 }
393 const segments = path.split("/");
394 const nameLength = path.startsWith("@") ? 2 : 1;
395 const name = segments.slice(0, nameLength).join("/");
396 const tag = segments.slice(nameLength).join("/");
397 if (segments.length <= nameLength || name === "" || tag === "") {
398 return undefined;
399 }
400 return { kind: "tag", name: qualify(name, scope), tag };
401};
402
403const qualify = (name: string, scope: string | undefined) =>
404 scope !== undefined && !name.startsWith("@") ? `${scope}/${name}` : name;
405
406const notFound = (message: string) =>
407 HttpServerResponse.json({ error: message }, { status: 404 });
408
409/**
410 * Install URLs: `/<name>/<tag>` redirects to the immutable tarball URL,
411 * `/<name>/-/<sha256>.tgz` streams it from R2. Everything the API routes
412 * do not claim lands here.
413 */
414const install = Effect.gen(function* () {
415 const request = yield* HttpServerRequest;
416 const { aliases } = yield* RegistryConfig;
417 const r2 = yield* Cloudflare.R2.ReadWriteBucket(Bucket);
418
419 if (request.method !== "GET" && request.method !== "HEAD") {
420 return yield* HttpServerResponse.json(
421 { error: "method not allowed" },
422 { status: 405 },
423 );
424 }
425 const host = request.headers.host ?? "localhost";
426 const target = parseInstallPath(
427 new URL(request.url, `https://${host}`).pathname,
428 aliases[host],
429 );
430 if (target === undefined) {
431 return yield* notFound("not found");
432 }
433 if (target.kind === "tag") {
434 const sha256 = yield* Tags.resolve(target.name, target.tag);
435 if (sha256 === undefined) {
436 return yield* notFound(`${target.name}@${target.tag} not found`);
437 }
438 return HttpServerResponse.redirect(`/${target.name}/-/${sha256}.tgz`, {
439 status: 302,
440 headers: { "cache-control": "no-store" },
441 });
442 }
443 const object = yield* r2.get(tarballKey(target.name, target.sha256));
444 if (object === null) {
445 return yield* notFound("tarball not found");
446 }
447 const headers = {
448 "content-type": "application/gzip",
449 "content-length": String(object.size),
450 "cache-control": "public, max-age=31536000, immutable",
451 };
452 return request.method === "HEAD"
453 ? HttpServerResponse.empty({ status: 200, headers })
454 : HttpServerResponse.stream(object.body, { status: 200, headers });
455});
456
457const banner = Effect.map(origin, (base) =>
458 HttpServerResponse.text(
459 `Preview package registry. Install with: bun add ${base}/<package>/<commit|branch:name|pr:N|pr:N:short-sha>\n`,
460 ),
461);
462
463/**
464 * The registry never serves files, so the HTTP platform's file surface is
465 * stubbed rather than pulling a file system into the Worker.
466 */
467const HttpPlatformStub = Layer.succeed(HttpPlatform.HttpPlatform, {
468 platform: "web",
469 compression: {
470 algorithms: new Set<HttpPlatform.CompressionAlgorithm>(),
471 compressResponse: (response) => Effect.succeed(response),
472 },
473 fileResponse: () => Effect.die("fileResponse is not supported"),
474 fileWebResponse: () => Effect.die("fileWebResponse is not supported"),
475});
476
477/**
478 * The registry Worker's Init phase. Runs at plan time to register bindings
479 * and the cron trigger, and once per isolate at runtime to build the
480 * router. Everything it needs comes from {@link RegistryConfig}.
481 */
482export const handler = Effect.gen(function* () {
483 const config = yield* RegistryConfig;
484 // Binding the capabilities here is what registers the R2, D1, and KV
485 // bindings on the Worker at plan time. Handlers bind them again per event,
486 // which at runtime is just a lookup in the Worker's environment.
487 yield* Cloudflare.R2.ReadWriteBucket(Bucket);
488 yield* Cloudflare.KV.ReadWriteNamespace(Cache);
489 const d1 = yield* Cloudflare.D1.QueryDatabase(Database);
490
491 // Built once per isolate. The D1 client memoizes its connection on each
492 // event's scope, so its layer is safe to share.
493 const github = yield* GitHub.GitHubApp.pipe(
494 Effect.provide(GitHub.GitHubAppLive),
495 );
496 const services = Layer.mergeAll(
497 Cloudflare.R2.ReadWriteBucketBinding,
498 Cloudflare.KV.ReadWriteNamespaceBinding,
499 SQL.D1Layer(d1),
500 Layer.succeed(GitHub.GitHubApp, github),
501 Layer.succeed(RegistryConfig, config),
502 );
503
504 // Recently resolved runs are reused so the uploads that follow a publish
505 // do not repeat the lookup. Failures are not kept.
506 const runs = KVCache.make(
507 (ref: RunRef) => `run:${ref.repo}#${ref.runId}:${ref.attempt}`,
508 Run,
509 (ref) => Effect.map(lookupRun(ref), (value) => ({ value, ttl: RUN_CACHE })),
510 );
511
512 yield* Cloudflare.Workers.cron(config.cron, () =>
513 sweep.pipe(Effect.provide(services)),
514 );
515
516 const registry = HttpApiBuilder.group(PkgApi, "Registry", (handlers) =>
517 handlers
518 .handle("publish", ({ payload }) =>
519 Effect.flatMap(runs(payload.run), (run) =>
520 publish(run, payload.manifest),
521 ).pipe(Effect.catchTag(storageFailures, (e) => Effect.die(e))),
522 )
523 // Raw so the body streams into R2 instead of being buffered.
524 .handleRaw("uploadTarball", ({ query, params, request }) =>
525 Effect.andThen(
526 runs(query),
527 uploadTarball(request, params.name, params.sha256),
528 ).pipe(Effect.catchTag("R2Error", (e) => Effect.die(e))),
529 ),
530 );
531
532 const healthGroup = HttpApiBuilder.group(PkgApi, "Health", (handlers) =>
533 handlers.handle("health", () => Effect.succeed({ ok: true })),
534 );
535
536 // Unlike `HttpApiBuilder.group`, the router defers a route's requirements
537 // to request time rather than the layer build, so the services are
538 // provided on the route effect itself.
539 const installRoutes = HttpRouter.use((router) =>
540 Effect.all([
541 router.add("GET", "/", banner),
542 router.add(
543 "*",
544 "/*",
545 install.pipe(
546 Effect.catchTag(storageFailures, (e) => Effect.die(e)),
547 Effect.provide(services),
548 ),
549 ),
550 ]),
551 );
552
553 return {
554 fetch: Layer.mergeAll(HttpApiBuilder.layer(PkgApi), installRoutes).pipe(
555 Layer.provide([registry, healthGroup]),
556 Layer.provide([Etag.layer, HttpPlatformStub, Path.layer]),
557 Layer.provide(services),
558 HttpRouter.toHttpEffect,
559 ),
560 };
561}).pipe(Effect.provide(bindings));