1/* oxlint-disable t3code/no-global-process-runtime -- This bootstrap runs before dependencies are installed. */
2// @effect-diagnostics globalDate:off globalFetch:off - bootstrap has no Effect dependencies.
3// @effect-diagnostics nodeBuiltinImport:off - standalone collector installed outside the checkout.
4import * as NodeCrypto from "node:crypto";
5import * as NodeFS from "node:fs";
6import * as NodeOS from "node:os";
7import * as NodePath from "node:path";
8
9export interface SummaryEvent {
10 id: string;
11 recordedAt: number;
12 host: string;
13 platform: string;
14 arch: string;
15 ci: boolean;
16 runId: string;
17 tasks: {
18 package: string;
19 task: string;
20 source: "local" | "remote" | "unspecified" | "executed" | "failed";
21 savedMs: number;
22 }[];
23}
24
25export function summarize(value: unknown): SummaryEvent["tasks"] {
26 if (!value || typeof value !== "object" || !("tasks" in value) || !Array.isArray(value.tasks))
27 return [];
28 return value.tasks.flatMap((task: unknown) => {
29 if (
30 !task ||
31 typeof task !== "object" ||
32 !("package_name" in task) ||
33 !("task_name" in task) ||
34 !("result" in task)
35 )
36 return [];
37 if (
38 typeof task.package_name !== "string" ||
39 typeof task.task_name !== "string" ||
40 !task.result ||
41 typeof task.result !== "object"
42 )
43 return [];
44 let source: SummaryEvent["tasks"][number]["source"] = "executed";
45 let savedMs = 0;
46 if ("CacheHit" in task.result) {
47 const hit = task.result.CacheHit;
48 if (
49 !hit ||
50 typeof hit !== "object" ||
51 !("saved_duration_ms" in hit) ||
52 typeof hit.saved_duration_ms !== "number" ||
53 !Number.isSafeInteger(hit.saved_duration_ms) ||
54 hit.saved_duration_ms < 0
55 )
56 return [];
57 if ("source" in hit && hit.source !== "Local" && hit.source !== "Remote") return [];
58 source = !("source" in hit) ? "unspecified" : hit.source === "Remote" ? "remote" : "local";
59 savedMs = hit.saved_duration_ms;
60 } else if ("RestoreFailed" in task.result) source = "failed";
61 else if (!("Spawned" in task.result)) return [];
62 else if (
63 task.result.Spawned &&
64 typeof task.result.Spawned === "object" &&
65 "cache_status" in task.result.Spawned &&
66 task.result.Spawned.cache_status === "Disabled"
67 )
68 return [];
69 return [
70 {
71 package: task.package_name.slice(0, 256),
72 task: task.task_name.slice(0, 256),
73 source,
74 savedMs,
75 },
76 ];
77 });
78}
79
80export function observeSummaries(
81 root: string,
82 endpoint: string,
83 env: NodeJS.ProcessEnv,
84 home = NodeOS.homedir(),
85) {
86 const cache = env.VITE_CACHE_PATH || NodePath.join(root, "node_modules/.vite/task-cache");
87 const queue = NodePath.join(home, ".cache/t3-task-cache/telemetry");
88 let host =
89 env.T3_TASK_CACHE_HOST ||
90 (env.GITHUB_ACTIONS === "true"
91 ? `${env.RUNNER_ENVIRONMENT || "github"}/${env.RUNNER_OS || NodeOS.platform()}/${env.RUNNER_ARCH || NodeOS.arch()}`
92 : NodeOS.hostname());
93 try {
94 host =
95 env.T3_TASK_CACHE_HOST ||
96 NodeFS.readFileSync(NodePath.join(home, ".agents/MACHINE"), "utf8").trim() ||
97 host;
98 } catch {
99 /* Machine labels are optional outside the fleet. */
100 }
101 const seen = new Set<string>();
102 const snapshot = (file: string, enqueue: boolean) => {
103 try {
104 const before = NodeFS.statSync(file, { bigint: true });
105 const contents = NodeFS.readFileSync(file, "utf8");
106 const after = NodeFS.statSync(file, { bigint: true });
107 if (before.ino !== after.ino || before.mtimeNs !== after.mtimeNs) return;
108 // Nested/concurrent launchers see the same summary; stable ids prevent double counting.
109 const id = NodeCrypto.createHash("sha256")
110 .update(`${host}\0${root}\0${before.ino}\0${before.mtimeNs}\0${contents}`)
111 .digest("hex");
112 if (seen.has(id)) return;
113 seen.add(id);
114 if (!enqueue) return;
115 const tasks = summarize(JSON.parse(contents));
116 if (!tasks.length) return;
117 const event: SummaryEvent = {
118 id,
119 recordedAt: Number(before.mtimeNs / 1_000_000n),
120 host: host.slice(0, 256),
121 platform: NodeOS.platform(),
122 arch: NodeOS.arch(),
123 ci: env.GITHUB_ACTIONS === "true",
124 runId: env.GITHUB_RUN_ID || "",
125 tasks,
126 };
127 NodeFS.mkdirSync(queue, { recursive: true, mode: 0o700 });
128 const temporary = NodePath.join(queue, `${id}.${process.pid}.tmp`);
129 NodeFS.writeFileSync(temporary, JSON.stringify({ endpoint, event }), { mode: 0o600 });
130 NodeFS.renameSync(temporary, NodePath.join(queue, `${id}.json`));
131 } catch {
132 /* Telemetry must never prevent task execution. */
133 }
134 };
135 const scan = (enqueue: boolean) => {
136 for (const entry of NodeFS.readdirSync(cache, { withFileTypes: true })) {
137 if (entry.isDirectory())
138 snapshot(NodePath.join(cache, entry.name, "last-summary.json"), enqueue);
139 }
140 };
141 try {
142 NodeFS.mkdirSync(cache, { recursive: true });
143 scan(false);
144 // Capture nested runs before the parent's atomic summary replaces them.
145 const watcher = NodeFS.watch(cache, { recursive: true }, (_event, file) => {
146 if (file?.endsWith("last-summary.json")) snapshot(NodePath.join(cache, file), true);
147 });
148 watcher.on("error", () => watcher.close());
149 return () => {
150 watcher.close();
151 try {
152 scan(true);
153 } catch {
154 /* A task may remove its cache directory. */
155 }
156 };
157 } catch {
158 return () => {};
159 }
160}
161
162export async function flushTelemetry(home = NodeOS.homedir(), env = process.env) {
163 const queue = NodePath.join(home, ".cache/t3-task-cache/telemetry");
164 const deadline = Date.now() + 15_000;
165 const skipped = new Set<string>();
166 // A flusher that finds the lock taken exits and leaves its event to the holder. Re-check after
167 // releasing, or an event queued after the holder's last scan waits for the next command.
168 try {
169 while (
170 (await drainTelemetry(queue, home, env, deadline, skipped)) &&
171 Date.now() < deadline &&
172 NodeFS.readdirSync(queue).some((entry) => entry.endsWith(".json") && !skipped.has(entry))
173 );
174 } catch {
175 /* Leave unsent events for a later invocation. */
176 }
177}
178
179async function drainTelemetry(
180 queue: string,
181 home: string,
182 env: NodeJS.ProcessEnv,
183 deadline: number,
184 skipped: Set<string>,
185) {
186 const lock = NodePath.join(queue, "flush.lock");
187 try {
188 NodeFS.mkdirSync(queue, { recursive: true, mode: 0o700 });
189 try {
190 if (Date.now() - NodeFS.statSync(lock).mtimeMs > 60_000) NodeFS.rmdirSync(lock);
191 } catch {
192 /* No previous flusher. */
193 }
194 NodeFS.mkdirSync(lock);
195 } catch {
196 return false;
197 }
198 try {
199 const files = NodeFS.readdirSync(queue)
200 .filter((file) => file.endsWith(".json"))
201 .sort(
202 (a, b) =>
203 NodeFS.statSync(NodePath.join(queue, a)).mtimeMs -
204 NodeFS.statSync(NodePath.join(queue, b)).mtimeMs,
205 );
206 // Bound disk use while offline. Unsent events retry on the next invocation.
207 for (const file of files.slice(0, Math.max(0, files.length - 2000)))
208 NodeFS.rmSync(NodePath.join(queue, file), { force: true });
209 const credentialsFile = NodePath.join(home, ".config/t3-task-cache/credentials.json");
210 const credentials = NodeFS.existsSync(credentialsFile)
211 ? JSON.parse(NodeFS.readFileSync(credentialsFile, "utf8"))
212 : {};
213 // Re-scan between sends so a concurrent command's final summary joins this flush.
214 for (let sent = 0; sent < 100 && Date.now() < deadline;) {
215 const file = NodeFS.readdirSync(queue)
216 .filter((entry) => entry.endsWith(".json") && !skipped.has(entry))
217 .sort()[0];
218 if (!file) return true;
219 const path = NodePath.join(queue, file);
220 const { endpoint, event } = JSON.parse(NodeFS.readFileSync(path, "utf8"));
221 const url = new URL(endpoint);
222 const configured = env.VP_REMOTE_CACHE_URL ? new URL(env.VP_REMOTE_CACHE_URL) : undefined;
223 const configuredEndpoint = configured && new URL(configured.href);
224 if (configuredEndpoint) {
225 configuredEndpoint.username = "";
226 configuredEndpoint.password = "";
227 }
228 const credential =
229 configured?.username && configuredEndpoint?.href.replace(/\/$/, "") === endpoint
230 ? {
231 name: decodeURIComponent(configured.username),
232 token: decodeURIComponent(configured.password),
233 }
234 : credentials[endpoint];
235 let authorization: string;
236 if (
237 credential &&
238 !(
239 env.GITHUB_ACTIONS === "true" &&
240 env.ACTIONS_ID_TOKEN_REQUEST_URL &&
241 env.ACTIONS_ID_TOKEN_REQUEST_TOKEN
242 )
243 )
244 authorization = `Basic ${Buffer.from(`${credential.name}:${credential.token}`).toString("base64")}`;
245 else if (env.ACTIONS_ID_TOKEN_REQUEST_URL && env.ACTIONS_ID_TOKEN_REQUEST_TOKEN) {
246 const tokenUrl = new URL(env.ACTIONS_ID_TOKEN_REQUEST_URL);
247 tokenUrl.searchParams.set("audience", endpoint);
248 const response = await fetch(tokenUrl, {
249 headers: { Authorization: `Bearer ${env.ACTIONS_ID_TOKEN_REQUEST_TOKEN}` },
250 signal: AbortSignal.timeout(3000),
251 });
252 if (!response.ok) return false;
253 const token = (await response.json()) as { value: string };
254 authorization = `Bearer ${token.value}`;
255 } else {
256 skipped.add(file);
257 continue;
258 }
259 sent++;
260 url.pathname = `${url.pathname.replace(/\/$/, "")}/telemetry`;
261 const response = await fetch(url, {
262 method: "POST",
263 headers: { Authorization: authorization, "Content-Type": "application/json" },
264 body: JSON.stringify(event),
265 signal: AbortSignal.timeout(3000),
266 });
267 await response.body?.cancel();
268 if (response.ok || [400, 401, 403, 413].includes(response.status)) NodeFS.unlinkSync(path);
269 else return false;
270 }
271 } catch {
272 /* Leave unsent events for a later invocation. */
273 } finally {
274 NodeFS.rmdirSync(lock);
275 }
276}