1// @effect-diagnostics nodeBuiltinImport:off globalFetch:off globalConsole:off globalTimers:off - Container bootstrap runs before the Effect server starts.
2import * as NodeCrypto from "node:crypto";
3import * as NodeFS from "node:fs";
4import * as NodeFSP from "node:fs/promises";
5import * as NodePath from "node:path";
6import * as NodeStreamPromises from "node:stream/promises";
7import * as NodeChildProcess from "node:child_process";
8import * as NodeUtil from "node:util";
9import * as NodeURL from "node:url";
10import * as Schema from "effect/Schema";
11import {
12 ForkReleaseManifest,
13 ForkReleaseListing,
14 latestForkRelease,
15} from "@t3tools/shared/forkUpdateDirectory";
16
17const run = NodeUtil.promisify(NodeChildProcess.execFile);
18const decodeManifest = Schema.decodeUnknownSync(ForkReleaseManifest);
19const decodeListing = Schema.decodeUnknownSync(ForkReleaseListing);
20
21class FeedUnavailableError extends Error {
22 constructor(message = "The release share is unavailable. Starting the installed runtime.") {
23 super(message);
24 }
25}
26
27function rethrowBodyError(error: unknown): never {
28 if (
29 error instanceof TypeError ||
30 (error instanceof DOMException && ["AbortError", "TimeoutError"].includes(error.name))
31 ) {
32 throw new FeedUnavailableError();
33 }
34 throw error;
35}
36
37async function request(url: URL, timeout: number) {
38 const response = await fetch(url, {
39 signal: AbortSignal.timeout(timeout),
40 cache: "no-store",
41 }).catch(() => {
42 throw new FeedUnavailableError();
43 });
44 if (response.status >= 500) {
45 throw new FeedUnavailableError();
46 }
47 if (!response.ok) throw new Error(`Release share returned HTTP ${response.status}.`);
48 return response;
49}
50
51export async function prepareRuntime(input: {
52 feedUrl: string;
53 baseDirectory: string;
54 cacheDirectory: string;
55 activate?: boolean;
56}): Promise<string> {
57 const baseUrl = new URL(input.feedUrl.endsWith("/") ? input.feedUrl : `${input.feedUrl}/`);
58 if (baseUrl.protocol !== "https:" && baseUrl.hostname !== "127.0.0.1") {
59 throw new Error("KNIFE_UPDATE_URL must use HTTPS.");
60 }
61
62 const listingUrl = new URL(baseUrl);
63 listingUrl.searchParams.set("ls", "");
64 const listing = await request(listingUrl, 30_000);
65 const version = latestForkRelease(decodeListing(await listing.json().catch(rethrowBodyError)));
66 if (version === undefined) {
67 throw new FeedUnavailableError("No releases are published. Starting the installed runtime.");
68 }
69 const response = await request(new URL(`${version}.json`, baseUrl), 30_000);
70 const release = decodeManifest(await response.json().catch(rethrowBodyError));
71 if (release.version !== version)
72 throw new Error("The published release version does not match its filename.");
73 const runtime = (
74 await NodeFSP.readFile(NodePath.join(input.baseDirectory, "knife-server-runtime"), "utf8")
75 ).trim();
76 if (runtime !== release.server.runtime) {
77 throw new Error(
78 "The release needs a newer container image. Rebuild the image before restarting.",
79 );
80 }
81
82 const artifacts = [
83 ["server", release.server],
84 ["assets", release.assets],
85 ["ui", release.ui],
86 ] as const;
87 const identity = NodeCrypto.createHash("sha256")
88 .update(artifacts.map(([, { sha256 }]) => sha256).join("\0"))
89 .digest("hex");
90 const blobs = NodePath.join(input.cacheDirectory, "artifacts");
91 await NodeFSP.mkdir(blobs, { recursive: true });
92 const directory = NodePath.join(input.cacheDirectory, `release-${identity}`);
93 const entrypoint = NodePath.join(directory, "apps/server/dist/bin.mjs");
94 const ready = await NodeFSP.readFile(NodePath.join(directory, "version"), "utf8").catch(
95 () => null,
96 );
97 if (ready === null) {
98 const staging = await NodeFSP.mkdtemp(NodePath.join(input.cacheDirectory, ".download-"));
99 try {
100 for (const [kind, artifact] of artifacts) {
101 const archive = NodePath.join(blobs, `${artifact.sha256}.tar.gz`);
102 const cached = await NodeFSP.stat(archive).then(
103 (stat) => stat.isFile() && stat.size === artifact.size,
104 () => false,
105 );
106 if (!cached) {
107 const partial = NodePath.join(staging, `${kind}.tar.gz`);
108 const downloaded = await request(new URL(artifact.file, baseUrl), 180_000);
109 const body = downloaded.body;
110 if (!body) {
111 throw new Error(`Could not download the ${kind} archive (HTTP ${downloaded.status}).`);
112 }
113 const hash = NodeCrypto.createHash("sha256");
114 let size = 0;
115 await NodeStreamPromises.pipeline(
116 async function* () {
117 try {
118 for await (const chunk of body) {
119 size += chunk.length;
120 if (size > artifact.size)
121 throw new Error(`The ${kind} archive exceeds its published size.`);
122 hash.update(chunk);
123 yield chunk;
124 }
125 } catch (error) {
126 rethrowBodyError(error);
127 }
128 },
129 NodeFS.createWriteStream(partial, { flags: "wx" }),
130 );
131 if (size !== artifact.size || hash.digest("hex") !== artifact.sha256) {
132 throw new Error(`The ${kind} archive failed checksum verification.`);
133 }
134
135 await NodeFSP.rename(partial, archive);
136 }
137
138 const { stdout } = await run("tar", ["-tzf", archive]);
139 const members = stdout.split(/\r?\n/).filter(Boolean);
140 if (
141 members.some((member) => {
142 const normalized = member.replace(/^\.\//, "").replace(/\/$/, "");
143 return (
144 normalized.startsWith("/") ||
145 normalized.split("/").includes("..") ||
146 (kind === "server" &&
147 !["", ".", "apps", "apps/server", "apps/server/dist"].includes(normalized) &&
148 !normalized.startsWith("apps/server/dist/"))
149 );
150 })
151 ) {
152 throw new Error(`The ${kind} archive contains an unexpected path.`);
153 }
154 const destination =
155 kind === "server" ? staging : NodePath.join(staging, "apps/server/dist/client");
156 await NodeFSP.mkdir(destination, { recursive: true });
157 await run("tar", [
158 "-xzf",
159 archive,
160 "-C",
161 destination,
162 "--no-same-owner",
163 "--no-same-permissions",
164 ]);
165 }
166
167 await NodeFSP.access(NodePath.join(staging, "apps/server/dist/bin.mjs"));
168 await NodeFSP.access(NodePath.join(staging, "apps/server/dist/client/index.html"));
169 await NodeFSP.symlink(
170 NodePath.join(input.baseDirectory, "apps/server/node_modules"),
171 NodePath.join(staging, "node_modules"),
172 "dir",
173 );
174 await run(
175 process.execPath,
176 [NodePath.join(staging, "apps/server/dist/bin.mjs"), "--version"],
177 {
178 env: { ...process.env, T3CODE_VERSION_OVERRIDE: release.server.version },
179 timeout: 30_000,
180 },
181 );
182 await NodeFSP.writeFile(
183 NodePath.join(staging, "knife-server-runtime"),
184 release.server.runtime,
185 );
186 await NodeFSP.writeFile(NodePath.join(staging, "version"), release.server.version);
187 await NodeFSP.rm(directory, { recursive: true, force: true });
188 await NodeFSP.rename(staging, directory);
189 } finally {
190 await NodeFSP.rm(staging, { recursive: true, force: true });
191 }
192 }
193
194 const keepDirectories = new Set([NodePath.basename(directory)]);
195 if (input.activate !== false) {
196 await NodeFSP.writeFile(NodePath.join(directory, "version"), release.server.version);
197 const pointer = NodePath.join(input.cacheDirectory, ".current");
198 await NodeFSP.rm(pointer, { force: true });
199 await NodeFSP.symlink(directory, pointer, "dir");
200 await NodeFSP.rename(pointer, NodePath.join(input.cacheDirectory, "current"));
201 } else {
202 const active = await NodeFSP.realpath(NodePath.join(input.cacheDirectory, "current")).catch(
203 () => null,
204 );
205 if (active !== null) keepDirectories.add(NodePath.basename(active));
206 }
207 for (const name of await NodeFSP.readdir(input.cacheDirectory)) {
208 if (/^release-[a-f0-9]{64}$/.test(name) && !keepDirectories.has(name)) {
209 await NodeFSP.rm(NodePath.join(input.cacheDirectory, name), { recursive: true, force: true });
210 }
211 }
212 const keep = new Set(artifacts.map(([, { sha256 }]) => `${sha256}.tar.gz`));
213 for (const name of await NodeFSP.readdir(blobs)) {
214 if (/^[a-f0-9]{64}\.tar\.gz$/.test(name) && !keep.has(name)) {
215 await NodeFSP.rm(NodePath.join(blobs, name));
216 }
217 }
218 return entrypoint;
219}
220
221if (
222 process.argv[1] &&
223 NodeURL.fileURLToPath(import.meta.url) === NodePath.resolve(process.argv[1])
224) {
225 const feedUrl = process.env.KNIFE_UPDATE_URL;
226 if (feedUrl) {
227 await prepareRuntime({
228 feedUrl,
229 baseDirectory: "/opt/knife",
230 cacheDirectory: NodePath.join(process.env.T3CODE_HOME ?? "/data", "runtime/knife-container"),
231 }).catch((error: unknown) => {
232 console.error(
233 error instanceof Error ? error.message : "Could not prepare the container update.",
234 );
235 if (!(error instanceof FeedUnavailableError)) process.exitCode = 1;
236 });
237 }
238} else if (process.argv[2] === "serve" && process.env.KNIFE_UPDATE_URL) {
239 const feedUrl = process.env.KNIFE_UPDATE_URL;
240 let downloading = false;
241 setInterval(() => {
242 if (downloading) return;
243 downloading = true;
244 void prepareRuntime({
245 feedUrl,
246 baseDirectory: "/opt/knife",
247 cacheDirectory: NodePath.join(process.env.T3CODE_HOME ?? "/data", "runtime/knife-container"),
248 activate: false,
249 })
250 .catch((error: unknown) => {
251 console.error(
252 error instanceof Error ? error.message : "Could not download the container update.",
253 );
254 })
255 .finally(() => {
256 downloading = false;
257 });
258 }, 4 * 60_000).unref();
259}