1import * as Cloudflare from "alchemy/Cloudflare";
2import * as Clock from "effect/Clock";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Option from "effect/Option";
6import * as Stream from "effect/Stream";
7import {
8 Bucket,
9 ORPHAN_GRACE,
10 RegistryConfig,
11 SWEEP_LOOKAHEAD,
12 tarballKey,
13} from "./Bindings.ts";
14import * as GitHub from "./GitHub.ts";
15import * as Tags from "./Tags.ts";
16
17/**
18 * Expiry sweep. Tags tied to pull requests are extended while any of those
19 * pull requests is open and pinned to the latest close time plus TTL once
20 * they all close. Expired tags go away, and tarballs nothing points at are
21 * deleted from R2.
22 */
23export const sweep = Effect.gen(function* () {
24 const { policy } = yield* RegistryConfig;
25 const r2 = yield* Cloudflare.R2.ReadWriteBucket(Bucket);
26 const github = yield* GitHub.GitHubApp;
27
28 const now = yield* Clock.currentTimeMillis;
29 const ttl = Duration.toMillis(policy.ttl);
30 const due = yield* Tags.dueLinked(now + Duration.toMillis(SWEEP_LOOKAHEAD));
31
32 // One lookup per pull request across every due row. `undefined` means the
33 // pull request is still open; a failed lookup is recorded as such.
34 const lookups = yield* Effect.forEach(
35 new Set(due.flatMap((row) => row.linked_prs)),
36 (ref) => {
37 const [repo, number] = ref.split("#");
38 return github.getPullRequest(repo!, Number(number)).pipe(
39 Effect.map((pr) =>
40 pr.state === "open"
41 ? undefined
42 : Date.parse(pr.merged_at ?? pr.closed_at ?? "") || now,
43 ),
44 Effect.map((closedAt) => [ref, { closedAt }] as const),
45 Effect.catch((e) =>
46 Effect.logWarning(
47 `pull request ${ref} lookup failed: ${GitHub.describe(e)}`,
48 ).pipe(Effect.as([ref, undefined] as const)),
49 ),
50 );
51 },
52 { concurrency: 4 },
53 );
54 const results = new Map(lookups);
55 const extended = yield* Effect.forEach(due, (row) => {
56 const known = row.linked_prs.flatMap((ref) => {
57 const result = results.get(ref);
58 return result === undefined ? [] : [result.closedAt];
59 });
60 // Open pull requests dominate; otherwise the latest close wins. A row
61 // whose pull requests could not all be checked is kept past the next
62 // sweep rather than left to expire: it is re-checked, never deleted,
63 // on GitHub's bad day.
64 const closes = known.filter((closedAt) => closedAt !== undefined);
65 const expiresAt =
66 closes.length < known.length
67 ? now + ttl
68 : known.length < row.linked_prs.length
69 ? Math.max(row.expires_at, now + Duration.toMillis(SWEEP_LOOKAHEAD))
70 : Math.max(...closes) + ttl;
71 return expiresAt === row.expires_at
72 ? Effect.succeed(false)
73 : Tags.setExpiry(row.package, row.tag, expiresAt).pipe(Effect.as(true));
74 }).pipe(Effect.map((changed) => changed.filter(Boolean).length));
75
76 const removed = yield* Tags.deleteExpired(now);
77 const referenced = yield* Tags.referencedTarballs();
78 const unreferenced = removed.filter(
79 ({ package: pkg, sha256 }) => !referenced.has(`${pkg}/${sha256}`),
80 );
81 yield* Effect.forEach(
82 unreferenced,
83 ({ package: pkg, sha256 }) => r2.delete(tarballKey(pkg, sha256)),
84 { discard: true },
85 );
86
87 // Uploads that never got tagged.
88 const orphans = yield* Stream.paginate(
89 undefined as string | undefined,
90 (cursor) =>
91 r2
92 .list({ cursor, limit: 500 })
93 .pipe(
94 Effect.map(
95 (page) =>
96 [
97 page.objects,
98 page.truncated ? Option.some(page.cursor) : Option.none(),
99 ] as const,
100 ),
101 ),
102 ).pipe(
103 Stream.filter((object) => {
104 const match = object.key.match(/^(.+)\/([a-f0-9]{64})\.tgz$/);
105 return (
106 match !== null &&
107 !referenced.has(`${decodeURIComponent(match[1]!)}/${match[2]}`) &&
108 object.uploaded.getTime() < now - Duration.toMillis(ORPHAN_GRACE)
109 );
110 }),
111 Stream.mapEffect((object) => r2.delete(object.key)),
112 Stream.runCount,
113 );
114 yield* Effect.logInfo(
115 `sweep: ${due.length} due, ${extended} extended, ${removed.length} tags expired, ${unreferenced.length + orphans} tarballs deleted`,
116 );
117});