HttpRemote
tcmts.cli.sync.HttpRemote · #lrhz5gnn · class
import type { LockEntry } from "#codebase/tcmts/cli/codebase";
import type { ContributionView } from "#codebase/tcmts/cli/contributions";
import type { ConflictRecord } from "#codebase/tcmts/cli/merge";
import { ConflictError, leanRecord, normalizeFingerprint } from "#codebase/tcmts/cli/sync";
import type {
BranchState, Caller, NamespaceNode, NewContribution, Output, Remote, SharedLint, SharedResult,
} from "#codebase/tcmts/cli/sync";
import type { DefRecord } from "#codebase/tcmts/core/types";
import type { Causal, Contribution, Release } from "#codebase/tcmts/store";
import http from "node:http";
import https from "node:https";
import type { Readable } from "node:stream";
import type { TLSSocket } from "node:tls";
import { createGunzip, gzipSync } from "node:zlib";
export class HttpRemote implements Remote {
readonly label: string;
private url: URL;
private token: string | undefined;
private fingerprint: string | undefined;
constructor(url: string, token?: string, fingerprint?: string) {
this.label = url;
this.url = new URL(url.endsWith("/") ? url : url + "/");
this.token = token;
this.fingerprint = fingerprint && normalizeFingerprint(fingerprint);
if (this.fingerprint && this.url.protocol !== "https:")
throw new Error(`a certificate fingerprint needs an https:// URL (${url})`);
}
/**
* POST and resolve with the response stream. With a pinned fingerprint the server's certificate is checked
* as soon as TLS is up, before any of the request (including the token) is written.
*/
/** whether the server takes compressed bodies: its replies say so (RFC 7694); an older one's don't */
private gzips = false;
/** whether it takes lean records ({@link leanRecord}): its replies say so, as an older one's don't */
private lean = false;
private post(route: string, body: unknown, retried = false): Promise<Readable> {
const target = new URL(route, this.url);
const json = JSON.stringify(body);
// a big body (definitions, mostly) goes compressed, to a server that takes it: their JSON packs about eightfold
const zipped = this.gzips && json.length > 64 * 1024;
const payload = zipped ? gzipSync(json) : json;
const headers = {
"content-type": "application/json",
"content-length": Buffer.byteLength(payload),
"accept-encoding": "gzip",
// (and that this side takes lean records: what it gets, it derives the rest of itself)
"x-tcm-records": "lean",
...(zipped ? { "content-encoding": "gzip" } : {}),
...(this.token ? { authorization: `Bearer ${this.token}` } : {}),
};
return new Promise((resolve, reject) => {
const tls = target.protocol === "https:";
const req = (tls ? https : http).request(
target,
{
method: "POST",
headers,
...(tls && this.fingerprint ? { rejectUnauthorized: false } : {}),
},
(raw) => {
if (/\bgzip\b/.test(String(raw.headers["accept-encoding"] ?? ""))) this.gzips = true;
if (raw.headers["x-tcm-records"] === "lean") this.lean = true;
const res: Readable =
raw.headers["content-encoding"] === "gzip" ? raw.pipe(createGunzip()) : raw;
if ((raw.statusCode ?? 500) < 300) return resolve(res);
let text = "";
res
.setEncoding("utf8")
.on("data", (d: string) => (text += d))
.on("end", () => {
const msg = ((): string => {
try {
return (
(JSON.parse(text) as { error?: string }).error ?? raw.statusMessage ?? "error"
);
} catch {
return raw.statusMessage ?? "error";
}
})();
reject(
raw.statusCode === 409
? new ConflictError(msg)
: new Error(`${this.label}: ${msg}`),
);
});
},
);
req.on("error", (e: NodeJS.ErrnoException) => {
// a kept-alive connection the server closed while this side was busy (verifying what it received):
// the request never reached it, so it's sent again, once, on a new one
if (req.reusedSocket && e.code === "ECONNRESET" && !retried)
return void this.post(route, body, true).then(resolve, reject);
reject(
e instanceof ConflictError ? e : new Error(`cannot reach ${this.label}: ${e.message}`),
);
});
// a server that accepts and then stalls must not hang the client; `run` may legitimately be quiet for long, and
// `put` is as long as checking a batch of definitions takes there
const quiet = route === "put" ? 600 : 60;
if (route !== "run")
req.setTimeout(quiet * 1000, () => req.destroy(new Error(`no response for ${quiet}s`)));
const send = () => req.end(payload);
if (!tls || !this.fingerprint) return send();
req.on("socket", (socket) => {
const check = () => {
const got = (socket as TLSSocket).getPeerCertificate().fingerprint256 ?? "";
if (normalizeFingerprint(got) === this.fingerprint) send();
else
req.destroy(
new Error(
`certificate fingerprint mismatch (got ${got}); refusing to send the token`,
),
);
};
if (
(socket as TLSSocket).encrypted &&
!(socket as TLSSocket).connecting &&
(socket as TLSSocket).getPeerCertificate().fingerprint256
)
check();
else socket.once("secureConnect", check);
});
});
}
private async call<T>(route: string, body: unknown): Promise<T> {
const res = await this.post(route, body);
let text = "";
for await (const chunk of res) text += chunk;
return JSON.parse(text) as T;
}
async missing(components: string[]) {
return (await this.call<{ missing: string[] }>("missing", { components })).missing;
}
async get(hashes: string[]) {
return (await this.call<{ records: DefRecord[] }>("get", { hashes })).records;
}
put(records: DefRecord[]) {
return this.call<{ stored: number }>("put", {
records: this.lean ? records.map(leanRecord) : records,
});
}
async packages() {
return (await this.call<{ packages: LockEntry[] }>("packages", {})).packages;
}
async putPackages(entries: LockEntry[]) {
await this.call("put-packages", { packages: entries });
}
async branch(name: string) {
return (await this.call<{ branch: BranchState | null }>("branch", { name })).branch;
}
async setBranch(
name: string,
head: string,
expect: string | null,
command: string,
conflicts?: Record<string, ConflictRecord>,
) {
await this.call("set-branch", { name, head, expect, command, conflicts });
}
async ancestors(head: string) {
return (await this.call<{ causals: string[] }>("ancestors", { head })).causals;
}
async missingCausals(hashes: string[]) {
return (await this.call<{ missing: string[] }>("missing-causals", { hashes })).missing;
}
async getCausals(hashes: string[]) {
return (await this.call<{ causals: Causal[] }>("get-causals", { hashes })).causals;
}
async putCausals(causals: Causal[]) {
await this.call("put-causals", { causals });
}
async missingNamespaces(hashes: string[]) {
return (await this.call<{ missing: string[] }>("missing-namespaces", { hashes })).missing;
}
async getNamespaces(hashes: string[]) {
return (await this.call<{ nodes: NamespaceNode[] }>("get-namespaces", { hashes })).nodes;
}
async putNamespaces(nodes: NamespaceNode[]) {
await this.call("put-namespaces", { nodes });
}
async releases() {
return (await this.call<{ releases: Release[] }>("releases", {})).releases;
}
async resultKeys(hashes: string[]) {
return (await this.call<{ keys: [string, string][] }>("result-keys", { hashes })).keys;
}
async results(pairs: [string, string][]) {
return (await this.call<{ results: SharedResult[] }>("results", { pairs })).results;
}
async putResults(results: SharedResult[]) {
await this.call("put-results", { results });
}
async lintKeys(hashes: string[]) {
return (await this.call<{ keys: [string, string][] }>("lint-keys", { hashes })).keys;
}
async lints(pairs: [string, string][]) {
return (await this.call<{ lints: SharedLint[] }>("lints", { pairs })).lints;
}
async putLints(lints: SharedLint[]) {
await this.call("put-lints", { lints });
}
async putReleases(releases: Release[]) {
await this.call("put-releases", { releases });
}
async contributions(all: boolean) {
return (await this.call<{ contributions: Contribution[] }>("contributions", { all }))
.contributions;
}
contribution(id: number) {
return this.call<ContributionView>("contribution", { id });
}
openContribution(c: NewContribution) {
return this.call<{ id: number; created: boolean }>("open-contribution", c);
}
async comment(id: number, c: { author: string; body: string; about?: string | null }) {
await this.call("comment", { id, ...c });
}
mergeContribution(id: number, author: string) {
return this.call<{ mergedAs: string; todo: number }>("merge-contribution", { id, author });
}
async closeContribution(id: number, author: string, reopen = false) {
await this.call(reopen ? "reopen-contribution" : "close-contribution", { id, author });
}
async whoami(): Promise<Caller> {
try {
return await this.call<Caller>("whoami", {});
} catch (e) {
// a server from before contributors had one token, which is its maintainer's
if (/unknown route 'whoami'/.test((e as Error).message)) return { role: "maintainer" };
throw e;
}
}
async users() {
return (await this.call<{ users: { handle: string; added: number }[] }>("users", {})).users;
}
async addUser(handle: string) {
return (await this.call<{ token: string }>("add-user", { handle })).token;
}
async removeUser(handle: string) {
return (await this.call<{ removed: boolean }>("remove-user", { handle })).removed;
}
/** Output streams back as NDJSON: `{"o":"stdout","d":"..."}` lines, then `{"exit":0}`. */
async run(hash: string, args: string[], out: Output) {
const res = await this.post("run", { hash, args });
let buf = "";
let exit = 1;
const handle = (line: string) => {
if (!line) return;
const msg = JSON.parse(line) as {
o?: "stdout" | "stderr";
d?: string;
exit?: number;
error?: string;
};
if (msg.o) out(msg.o, msg.d ?? "");
if (msg.error) out("stderr", `${msg.error}\n`);
if (msg.exit !== undefined) exit = msg.exit;
};
res.setEncoding("utf8");
for await (const chunk of res) {
buf += chunk as string;
const lines = buf.split("\n");
buf = lines.pop()!;
lines.forEach(handle);
}
handle(buf);
return exit;
}
}
uses
used by