364 lines
13 KiB
TypeScript
364 lines
13 KiB
TypeScript
// One T3 Code server. Keeps a shell stream open, holds every project and thread shell in memory,
|
|
// and reconnects with backoff. Writes go over the same socket, and thread pages go over HTTP.
|
|
|
|
import { RpcError, RpcSocket } from "./rpc.ts";
|
|
import {
|
|
ORCHESTRATION_PROTOCOL_VERSION,
|
|
type BoundedThread,
|
|
type Command,
|
|
type EnvironmentDescriptor,
|
|
type LaunchThreadInput,
|
|
type Project,
|
|
type ServerConfig,
|
|
type ShellStreamItem,
|
|
type ThreadShell,
|
|
} from "./types.ts";
|
|
|
|
export type HostStatus =
|
|
| "connecting"
|
|
| "online"
|
|
| "offline"
|
|
/** The token expired or was revoked. A new one has to be paired in. */
|
|
| "auth_invalid"
|
|
/** The server speaks another protocol version or is a different environment. */
|
|
| "incompatible";
|
|
|
|
export interface HostConfig {
|
|
id: string;
|
|
label: string;
|
|
/**
|
|
* The server's HTTP origins, preferred first, such as a LAN route then a VPN route. Every one
|
|
* must reach the same T3 server, since they share the environment ID and the token.
|
|
*/
|
|
urls: string[];
|
|
token: string;
|
|
environmentId: string;
|
|
}
|
|
|
|
/** What happened the last time the gateway tried one route. */
|
|
export interface RouteAttempt {
|
|
ok: boolean;
|
|
error: string | null;
|
|
at: string;
|
|
}
|
|
|
|
export interface ThreadChange {
|
|
host: T3Host;
|
|
previous: ThreadShell | undefined;
|
|
next: ThreadShell | undefined;
|
|
/** True while loading the first snapshot after start, when nothing should notify. */
|
|
initial: boolean;
|
|
}
|
|
|
|
export interface HostOptions {
|
|
onThreadChange?: (change: ThreadChange) => void;
|
|
onStatus?: (host: T3Host) => void;
|
|
log?: (message: string) => void;
|
|
}
|
|
|
|
const BACKOFF_MIN_MS = 1_000;
|
|
const BACKOFF_MAX_MS = 5 * 60_000;
|
|
const AUTH_RETRY_MS = 10 * 60_000;
|
|
/** How long one route gets to answer before the next is tried. */
|
|
const ROUTE_TIMEOUT_MS = 5_000;
|
|
|
|
export class HostError extends Error {
|
|
constructor(
|
|
message: string,
|
|
readonly status: number,
|
|
) {
|
|
super(message);
|
|
this.name = "HostError";
|
|
}
|
|
}
|
|
|
|
export class T3Host {
|
|
status: HostStatus = "connecting";
|
|
lastError: string | null = null;
|
|
readonly projects = new Map<string, Project>();
|
|
readonly threads = new Map<string, ThreadShell>();
|
|
serverConfig: ServerConfig | null = null;
|
|
serverVersion: string | null = null;
|
|
/** Results of the latest connect, by route. Routes after the one that answered were not tried. */
|
|
readonly routeAttempts = new Map<string, RouteAttempt>();
|
|
/** When the current connection came up. Null when not online. */
|
|
onlineSince: string | null = null;
|
|
/** When the last connection ended. Null if it never connected or is online now. */
|
|
lastOnlineAt: string | null = null;
|
|
/** When the next connect is scheduled, while waiting to retry. */
|
|
nextRetryAt: string | null = null;
|
|
|
|
private socket: RpcSocket | null = null;
|
|
private sequence: number | null = null;
|
|
private hasSnapshot = false;
|
|
private stopped = false;
|
|
private attempt = 0;
|
|
private retryTimer: ReturnType<typeof setTimeout> | null = null;
|
|
/** Set by reconnect() so a deliberate close connects again at once, with no backoff. */
|
|
private reconnecting = false;
|
|
/** The route this connection uses. Chosen afresh, preferred first, on every connect. */
|
|
private activeUrl: string;
|
|
|
|
constructor(
|
|
readonly config: HostConfig,
|
|
private readonly options: HostOptions = {},
|
|
) {
|
|
const [first] = config.urls;
|
|
if (!first) throw new Error(`${config.label} has no URL`);
|
|
this.activeUrl = first;
|
|
}
|
|
|
|
get id() {
|
|
return this.config.id;
|
|
}
|
|
|
|
get label() {
|
|
return this.config.label;
|
|
}
|
|
|
|
/** The route in use, or the one that will be tried first. */
|
|
get url() {
|
|
return this.activeUrl;
|
|
}
|
|
|
|
start() {
|
|
this.stopped = false;
|
|
void this.connect();
|
|
}
|
|
|
|
stop() {
|
|
this.stopped = true;
|
|
if (this.retryTimer) clearTimeout(this.retryTimer);
|
|
this.socket?.close();
|
|
}
|
|
|
|
/**
|
|
* Connects again now, starting from the preferred route. For when a better route may be back,
|
|
* such as the LAN after getting home, or to skip a backoff wait. Does nothing mid-connect.
|
|
*/
|
|
reconnect() {
|
|
if (this.stopped || this.status === "connecting") return;
|
|
this.attempt = 0;
|
|
if (this.socket) {
|
|
this.reconnecting = true;
|
|
this.socket.close();
|
|
return;
|
|
}
|
|
if (this.retryTimer) clearTimeout(this.retryTimer);
|
|
void this.connect();
|
|
}
|
|
|
|
/** Sends one orchestration command and resolves with its sequence. */
|
|
async dispatch(command: Command): Promise<{ sequence: number }> {
|
|
return this.requireSocket().call("orchestration.dispatchCommand", command);
|
|
}
|
|
|
|
async launchThread(input: LaunchThreadInput): Promise<{ threadId: string }> {
|
|
return this.requireSocket().call("orchestration.launchThread", input);
|
|
}
|
|
|
|
/** The recent window of one thread: its messages and turn items. */
|
|
bounded(threadId: string): Promise<BoundedThread> {
|
|
return this.http<BoundedThread>(`/api/orchestration/threads/${encodeURIComponent(threadId)}/bounded`);
|
|
}
|
|
|
|
/** The session's expiry, from `/api/auth/session`. */
|
|
async sessionExpiresAt(): Promise<string | null> {
|
|
const session = await this.http<{ expiresAt?: string }>("/api/auth/session");
|
|
return session.expiresAt ?? null;
|
|
}
|
|
|
|
supports(capability: string): boolean {
|
|
return this.serverConfig?.environment?.capabilities?.[capability] === true;
|
|
}
|
|
|
|
private requireSocket(): RpcSocket {
|
|
if (this.socket === null || this.status !== "online") {
|
|
throw new HostError(`${this.label} is ${this.status}`, 503);
|
|
}
|
|
return this.socket;
|
|
}
|
|
|
|
private async http<T>(path: string, init: RequestInit = {}): Promise<T> {
|
|
const response = await fetch(new URL(path, this.activeUrl), {
|
|
...init,
|
|
headers: {
|
|
authorization: `Bearer ${this.config.token}`,
|
|
"x-t3-orchestration-protocol": String(ORCHESTRATION_PROTOCOL_VERSION),
|
|
...init.headers,
|
|
},
|
|
});
|
|
if (!response.ok) {
|
|
const body = await response.text();
|
|
throw new HostError(`${this.label} ${path} returned ${response.status}: ${body.slice(0, 200)}`, response.status);
|
|
}
|
|
return response.json() as Promise<T>;
|
|
}
|
|
|
|
private setStatus(status: HostStatus, error: string | null = null) {
|
|
if (status === "online") this.onlineSince = new Date().toISOString();
|
|
else if (this.status === "online") {
|
|
this.lastOnlineAt = new Date().toISOString();
|
|
this.onlineSince = null;
|
|
}
|
|
this.status = status;
|
|
this.lastError = error;
|
|
this.options.onStatus?.(this);
|
|
}
|
|
|
|
private async connect() {
|
|
if (this.stopped) return;
|
|
this.nextRetryAt = null;
|
|
this.setStatus("connecting", this.lastError);
|
|
try {
|
|
const descriptor = await this.pickRoute();
|
|
const protocol = descriptor.orchestrationProtocolVersion ?? 1;
|
|
if (protocol !== ORCHESTRATION_PROTOCOL_VERSION) {
|
|
throw new HostError(`speaks orchestration protocol ${protocol}, the gateway speaks ${ORCHESTRATION_PROTOCOL_VERSION}`, 426);
|
|
}
|
|
if (descriptor.environmentId !== this.config.environmentId) {
|
|
throw new HostError(`is environment ${descriptor.environmentId}, expected ${this.config.environmentId}`, 409);
|
|
}
|
|
const { ticket } = await this.http<{ ticket: string }>("/api/auth/websocket-ticket/rpc", { method: "POST" });
|
|
const wsUrl = new URL("/ws", this.activeUrl);
|
|
wsUrl.protocol = wsUrl.protocol === "https:" ? "wss:" : "ws:";
|
|
wsUrl.searchParams.set("wsTicket", ticket);
|
|
wsUrl.searchParams.set("orchestrationProtocol", String(ORCHESTRATION_PROTOCOL_VERSION));
|
|
const socket = await RpcSocket.connect(wsUrl.href);
|
|
this.socket = socket;
|
|
this.serverConfig = await socket.call<ServerConfig>("server.getConfig", {});
|
|
this.subscribe(socket);
|
|
this.attempt = 0;
|
|
this.setStatus("online");
|
|
await socket.closed;
|
|
this.socket = null;
|
|
if (this.reconnecting) {
|
|
this.reconnecting = false;
|
|
void this.connect();
|
|
return;
|
|
}
|
|
this.scheduleRetry("offline", "connection closed");
|
|
} catch (error) {
|
|
this.socket?.close();
|
|
this.socket = null;
|
|
const message = error instanceof Error ? error.message : String(error);
|
|
if (error instanceof HostError && error.status === 401) this.scheduleRetry("auth_invalid", message);
|
|
else if (error instanceof HostError && (error.status === 426 || error.status === 409)) this.scheduleRetry("incompatible", message);
|
|
else this.scheduleRetry("offline", message);
|
|
}
|
|
}
|
|
|
|
/** Uses the first route whose well-known descriptor answers, and returns that descriptor. */
|
|
private async pickRoute(): Promise<EnvironmentDescriptor> {
|
|
const failures: string[] = [];
|
|
this.routeAttempts.clear();
|
|
for (const url of this.config.urls) {
|
|
try {
|
|
const response = await fetch(new URL("/.well-known/t3/environment", url), {
|
|
signal: AbortSignal.timeout(ROUTE_TIMEOUT_MS),
|
|
});
|
|
if (!response.ok) throw new Error(`returned ${response.status}`);
|
|
const descriptor = (await response.json()) as EnvironmentDescriptor;
|
|
if (url !== this.activeUrl) this.options.log?.(`${this.label}: using ${url}`);
|
|
this.activeUrl = url;
|
|
this.serverVersion = descriptor.serverVersion ?? null;
|
|
this.routeAttempts.set(url, { ok: true, error: null, at: new Date().toISOString() });
|
|
return descriptor;
|
|
} catch (error) {
|
|
const message = routeError(error);
|
|
this.routeAttempts.set(url, { ok: false, error: message, at: new Date().toISOString() });
|
|
failures.push(`${new URL(url).host} ${message}`);
|
|
}
|
|
}
|
|
throw new HostError(`no route answered (${failures.join(", ")})`, 503);
|
|
}
|
|
|
|
private subscribe(socket: RpcSocket) {
|
|
const input = this.hasSnapshot && this.sequence !== null ? { afterSequence: this.sequence } : {};
|
|
const { done } = socket.stream("orchestration.subscribeShell", input, (value) =>
|
|
this.apply(value as ShellStreamItem),
|
|
);
|
|
// The stream only ends when the socket does, or on a server error. Either way, start over.
|
|
done.then(
|
|
() => socket.close(),
|
|
(error: RpcError) => {
|
|
this.options.log?.(`${this.label}: shell stream ended: ${error.message}`);
|
|
socket.close();
|
|
},
|
|
);
|
|
}
|
|
|
|
private scheduleRetry(status: HostStatus, error: string) {
|
|
if (this.stopped) return;
|
|
this.setStatus(status, error);
|
|
this.options.log?.(`${this.label}: ${status}: ${error}`);
|
|
const delay =
|
|
status === "auth_invalid" || status === "incompatible"
|
|
? AUTH_RETRY_MS
|
|
: Math.min(BACKOFF_MAX_MS, BACKOFF_MIN_MS * 2 ** this.attempt++) * (0.5 + Math.random() / 2);
|
|
this.nextRetryAt = new Date(Date.now() + delay).toISOString();
|
|
this.retryTimer = setTimeout(() => void this.connect(), delay);
|
|
}
|
|
|
|
/** Applies one shell stream item to the in-memory state, reporting each thread that changed. */
|
|
apply(item: ShellStreamItem) {
|
|
switch (item.kind) {
|
|
case "snapshot": {
|
|
const { snapshot } = item;
|
|
for (const project of snapshot.projects) this.projects.set(project.id, project);
|
|
const known = new Set(snapshot.projects.map((project) => project.id));
|
|
for (const id of this.projects.keys()) if (!known.has(id)) this.projects.delete(id);
|
|
this.sequence = Math.max(this.sequence ?? 0, snapshot.snapshotSequence);
|
|
// The server also sends project-only snapshots with no threads: one first on every resume,
|
|
// and more as project metadata fills in. T3's own client keeps its threads for those, and
|
|
// so does the gateway.
|
|
if (snapshot.threads.length === 0 && this.hasSnapshot) return;
|
|
const initial = !this.hasSnapshot;
|
|
this.hasSnapshot = true;
|
|
const next = new Map(snapshot.threads.map((thread) => [thread.id, thread]));
|
|
for (const [id, previous] of this.threads) {
|
|
if (!next.has(id)) this.changeThread(previous, undefined, initial);
|
|
}
|
|
for (const thread of next.values()) this.changeThread(this.threads.get(thread.id), thread, initial);
|
|
return;
|
|
}
|
|
case "project.updated":
|
|
this.projects.set(item.project.id, item.project);
|
|
this.sequence = item.sequence;
|
|
return;
|
|
case "project.removed":
|
|
this.projects.delete(item.projectId);
|
|
this.sequence = item.sequence;
|
|
return;
|
|
case "thread.updated":
|
|
this.sequence = item.sequence;
|
|
if (item.location === "active") this.changeThread(this.threads.get(item.thread.id), item.thread, false);
|
|
else this.changeThread(this.threads.get(item.thread.id), undefined, false);
|
|
return;
|
|
case "thread.removed": {
|
|
this.sequence = item.sequence;
|
|
const previous = this.threads.get(item.threadId);
|
|
if (previous) this.changeThread(previous, undefined, false);
|
|
return;
|
|
}
|
|
case "synchronized":
|
|
return;
|
|
}
|
|
}
|
|
|
|
private changeThread(previous: ThreadShell | undefined, next: ThreadShell | undefined, initial: boolean) {
|
|
if (next) this.threads.set(next.id, next);
|
|
else if (previous) this.threads.delete(previous.id);
|
|
this.options.onThreadChange?.({ host: this, previous, next, initial });
|
|
}
|
|
}
|
|
|
|
/** A short reason a route failed, worded for a watch screen. */
|
|
function routeError(error: unknown): string {
|
|
if (error instanceof DOMException && error.name === "TimeoutError") return `no answer in ${ROUTE_TIMEOUT_MS / 1000} s`;
|
|
const message = error instanceof Error ? error.message : String(error);
|
|
if (/ECONNREFUSED|Unable to connect|refused/i.test(message)) return "refused";
|
|
if (/ENOTFOUND|getaddrinfo|DNS/i.test(message)) return "name not found";
|
|
return message;
|
|
}
|