412 lines
11 KiB
TypeScript
412 lines
11 KiB
TypeScript
import { BusEvent } from "@/bus/bus-event"
|
|
import { Bus } from "@/bus"
|
|
import { type IPty } from "node-pty"
|
|
import z from "zod"
|
|
import { Identifier } from "../id/id"
|
|
import { Log } from "../util/log"
|
|
import { Instance } from "../project/instance"
|
|
import { Shell } from "@/shell/shell"
|
|
import { Plugin } from "@/plugin"
|
|
|
|
export namespace Pty {
|
|
const log = Log.create({ service: "pty" })
|
|
|
|
const BUFFER_LIMIT = 1024 * 1024 * 2
|
|
const BUFFER_CHUNK = 64 * 1024
|
|
const encoder = new TextEncoder()
|
|
const decoder = new TextDecoder()
|
|
const FRAME_META = 0
|
|
const FRAME_OUTPUT = 1
|
|
const FRAME_INPUT = 2
|
|
const MAX_CONNECTION = 200
|
|
|
|
type Socket = {
|
|
readyState: number
|
|
send: (data: string | Uint8Array | ArrayBuffer) => void
|
|
close: (code?: number, reason?: string) => void
|
|
}
|
|
|
|
type Subscriber = {
|
|
id: number
|
|
connection: string
|
|
}
|
|
|
|
const sockets = new WeakMap<object, number>()
|
|
const owners = new WeakMap<object, string>()
|
|
let socketCounter = 0
|
|
let connectionCounter = 0
|
|
|
|
const tagSocket = (ws: Socket) => {
|
|
if (!ws || typeof ws !== "object") return
|
|
const next = (socketCounter = (socketCounter + 1) % Number.MAX_SAFE_INTEGER)
|
|
sockets.set(ws, next)
|
|
return next
|
|
}
|
|
|
|
const connection = () => {
|
|
connectionCounter = (connectionCounter + 1) % Number.MAX_SAFE_INTEGER
|
|
return `${Date.now().toString(36)}-${connectionCounter.toString(36)}`
|
|
}
|
|
|
|
const normalizeConnection = (value?: string) => {
|
|
const next = typeof value === "string" ? value.trim() : ""
|
|
if (!next) return connection()
|
|
if (next.length > MAX_CONNECTION) return connection()
|
|
if (encoder.encode(next).length > 255) return connection()
|
|
return next
|
|
}
|
|
|
|
const output = (connection: string, data: string) => {
|
|
const channel = encoder.encode(connection)
|
|
const chunk = encoder.encode(data)
|
|
const out = new Uint8Array(2 + channel.length + chunk.length)
|
|
out[0] = FRAME_OUTPUT
|
|
out[1] = channel.length
|
|
out.set(channel, 2)
|
|
out.set(chunk, 2 + channel.length)
|
|
return out
|
|
}
|
|
|
|
const input = (message: string | Uint8Array | ArrayBuffer) => {
|
|
if (typeof message === "string") {
|
|
return { data: message }
|
|
}
|
|
|
|
const bytes = message instanceof Uint8Array ? message : new Uint8Array(message)
|
|
if (bytes[0] !== FRAME_INPUT) return
|
|
const size = bytes[1]
|
|
if (!Number.isSafeInteger(size) || size < 0) return
|
|
if (bytes.length < 2 + size) return
|
|
return {
|
|
connection: decoder.decode(bytes.subarray(2, 2 + size)),
|
|
data: decoder.decode(bytes.subarray(2 + size)),
|
|
}
|
|
}
|
|
|
|
// WebSocket control frame: 0x00 + UTF-8 JSON ({ cursor, connection }).
|
|
const meta = (cursor: number, connection: string) => {
|
|
const json = JSON.stringify({ cursor, connection })
|
|
const bytes = encoder.encode(json)
|
|
const out = new Uint8Array(bytes.length + 1)
|
|
out[0] = FRAME_META
|
|
out.set(bytes, 1)
|
|
return out
|
|
}
|
|
|
|
type Spawn = (file: string, args: string | string[], options: unknown) => IPty
|
|
let override: Spawn | undefined
|
|
let spawn: Spawn | undefined
|
|
|
|
const pty = async (): Promise<Spawn> => {
|
|
if (override) return override
|
|
if (spawn) return spawn
|
|
const mod = await import("node-pty")
|
|
const next = mod.spawn as Spawn
|
|
spawn = next
|
|
return next
|
|
}
|
|
|
|
export function setSpawn(input?: Spawn) {
|
|
override = input
|
|
if (input) return
|
|
spawn = undefined
|
|
}
|
|
|
|
export const Info = z
|
|
.object({
|
|
id: Identifier.schema("pty"),
|
|
title: z.string(),
|
|
command: z.string(),
|
|
args: z.array(z.string()),
|
|
cwd: z.string(),
|
|
status: z.enum(["running", "exited"]),
|
|
pid: z.number(),
|
|
})
|
|
.meta({ ref: "Pty" })
|
|
|
|
export type Info = z.infer<typeof Info>
|
|
|
|
export const CreateInput = z.object({
|
|
command: z.string().optional(),
|
|
args: z.array(z.string()).optional(),
|
|
cwd: z.string().optional(),
|
|
title: z.string().optional(),
|
|
env: z.record(z.string(), z.string()).optional(),
|
|
})
|
|
|
|
export type CreateInput = z.infer<typeof CreateInput>
|
|
|
|
export const UpdateInput = z.object({
|
|
title: z.string().optional(),
|
|
size: z
|
|
.object({
|
|
rows: z.number(),
|
|
cols: z.number(),
|
|
})
|
|
.optional(),
|
|
})
|
|
|
|
export type UpdateInput = z.infer<typeof UpdateInput>
|
|
|
|
export const Event = {
|
|
Created: BusEvent.define("pty.created", z.object({ info: Info })),
|
|
Updated: BusEvent.define("pty.updated", z.object({ info: Info })),
|
|
Exited: BusEvent.define("pty.exited", z.object({ id: Identifier.schema("pty"), exitCode: z.number() })),
|
|
Deleted: BusEvent.define("pty.deleted", z.object({ id: Identifier.schema("pty") })),
|
|
}
|
|
|
|
interface ActiveSession {
|
|
info: Info
|
|
process: IPty
|
|
buffer: string
|
|
bufferCursor: number
|
|
cursor: number
|
|
subscribers: Map<Socket, Subscriber>
|
|
}
|
|
|
|
const state = Instance.state(
|
|
() => new Map<string, ActiveSession>(),
|
|
async (sessions) => {
|
|
for (const session of sessions.values()) {
|
|
try {
|
|
session.process.kill()
|
|
} catch {}
|
|
for (const ws of session.subscribers.keys()) {
|
|
try {
|
|
ws.close()
|
|
} catch {
|
|
// ignore
|
|
}
|
|
}
|
|
}
|
|
sessions.clear()
|
|
},
|
|
)
|
|
|
|
export function list() {
|
|
return Array.from(state().values()).map((s) => s.info)
|
|
}
|
|
|
|
export function get(id: string) {
|
|
return state().get(id)?.info
|
|
}
|
|
|
|
export async function create(input: CreateInput) {
|
|
const id = Identifier.create("pty", false)
|
|
const command = input.command || Shell.preferred()
|
|
const args = input.args || []
|
|
if (command.endsWith("sh")) {
|
|
args.push("-l")
|
|
}
|
|
|
|
const cwd = input.cwd || Instance.directory
|
|
const shellEnv = await Plugin.trigger("shell.env", { cwd }, { env: {} })
|
|
const env = {
|
|
...process.env,
|
|
...input.env,
|
|
...shellEnv.env,
|
|
TERM: "xterm-256color",
|
|
OPENCODE_TERMINAL: "1",
|
|
} as Record<string, string>
|
|
|
|
if (process.platform === "win32") {
|
|
env.LC_ALL = "C.UTF-8"
|
|
env.LC_CTYPE = "C.UTF-8"
|
|
env.LANG = "C.UTF-8"
|
|
}
|
|
log.info("creating session", { id, cmd: command, args, cwd })
|
|
|
|
const spawn = await pty()
|
|
const ptyProcess = spawn(command, args, {
|
|
name: "xterm-256color",
|
|
cwd,
|
|
env,
|
|
})
|
|
|
|
const info = {
|
|
id,
|
|
title: input.title || `Terminal ${id.slice(-4)}`,
|
|
command,
|
|
args,
|
|
cwd,
|
|
status: "running",
|
|
pid: ptyProcess.pid,
|
|
} as const
|
|
const session: ActiveSession = {
|
|
info,
|
|
process: ptyProcess,
|
|
buffer: "",
|
|
bufferCursor: 0,
|
|
cursor: 0,
|
|
subscribers: new Map(),
|
|
}
|
|
state().set(id, session)
|
|
ptyProcess.onData((chunk) => {
|
|
session.cursor += chunk.length
|
|
|
|
for (const [ws, sub] of session.subscribers) {
|
|
if (ws.readyState !== 1) {
|
|
session.subscribers.delete(ws)
|
|
continue
|
|
}
|
|
|
|
if (typeof ws === "object" && sockets.get(ws) !== sub.id) {
|
|
session.subscribers.delete(ws)
|
|
continue
|
|
}
|
|
|
|
try {
|
|
ws.send(output(sub.connection, chunk))
|
|
} catch {
|
|
session.subscribers.delete(ws)
|
|
}
|
|
}
|
|
|
|
session.buffer += chunk
|
|
if (session.buffer.length <= BUFFER_LIMIT) return
|
|
const excess = session.buffer.length - BUFFER_LIMIT
|
|
session.buffer = session.buffer.slice(excess)
|
|
session.bufferCursor += excess
|
|
})
|
|
ptyProcess.onExit(({ exitCode }) => {
|
|
log.info("session exited", { id, exitCode })
|
|
session.info.status = "exited"
|
|
for (const ws of session.subscribers.keys()) {
|
|
try {
|
|
ws.close()
|
|
} catch {
|
|
// ignore
|
|
}
|
|
}
|
|
session.subscribers.clear()
|
|
Bus.publish(Event.Exited, { id, exitCode })
|
|
state().delete(id)
|
|
})
|
|
Bus.publish(Event.Created, { info })
|
|
return info
|
|
}
|
|
|
|
export async function update(id: string, input: UpdateInput) {
|
|
const session = state().get(id)
|
|
if (!session) return
|
|
if (input.title) {
|
|
session.info.title = input.title
|
|
}
|
|
if (input.size) {
|
|
session.process.resize(input.size.cols, input.size.rows)
|
|
}
|
|
Bus.publish(Event.Updated, { info: session.info })
|
|
return session.info
|
|
}
|
|
|
|
export async function remove(id: string) {
|
|
const session = state().get(id)
|
|
if (!session) return
|
|
log.info("removing session", { id })
|
|
try {
|
|
session.process.kill()
|
|
} catch {}
|
|
for (const ws of session.subscribers.keys()) {
|
|
try {
|
|
ws.close()
|
|
} catch {
|
|
// ignore
|
|
}
|
|
}
|
|
session.subscribers.clear()
|
|
state().delete(id)
|
|
Bus.publish(Event.Deleted, { id })
|
|
}
|
|
|
|
export function resize(id: string, cols: number, rows: number) {
|
|
const session = state().get(id)
|
|
if (session && session.info.status === "running") {
|
|
session.process.resize(cols, rows)
|
|
}
|
|
}
|
|
|
|
export function write(id: string, data: string) {
|
|
const session = state().get(id)
|
|
if (session && session.info.status === "running") {
|
|
session.process.write(data)
|
|
}
|
|
}
|
|
|
|
export function connect(id: string, ws: Socket, cursor?: number, connectionID?: string) {
|
|
const session = state().get(id)
|
|
if (!session) {
|
|
ws.close()
|
|
return
|
|
}
|
|
log.info("client connected to session", { id })
|
|
|
|
const socketId = tagSocket(ws)
|
|
if (socketId === undefined) {
|
|
ws.close()
|
|
return
|
|
}
|
|
|
|
const previous = owners.get(ws)
|
|
if (previous && previous !== id) {
|
|
state().get(previous)?.subscribers.delete(ws)
|
|
}
|
|
|
|
owners.set(ws, id)
|
|
const sub = {
|
|
id: socketId,
|
|
connection: normalizeConnection(connectionID),
|
|
}
|
|
session.subscribers.set(ws, sub)
|
|
|
|
const cleanup = () => {
|
|
session.subscribers.delete(ws)
|
|
if (owners.get(ws) === id) owners.delete(ws)
|
|
}
|
|
|
|
const start = session.bufferCursor
|
|
const end = session.cursor
|
|
|
|
const from =
|
|
cursor === -1 ? end : typeof cursor === "number" && Number.isSafeInteger(cursor) ? Math.max(0, cursor) : 0
|
|
|
|
const data = (() => {
|
|
if (!session.buffer) return ""
|
|
if (from >= end) return ""
|
|
const offset = Math.max(0, from - start)
|
|
if (offset >= session.buffer.length) return ""
|
|
return session.buffer.slice(offset)
|
|
})()
|
|
|
|
if (data) {
|
|
try {
|
|
for (let i = 0; i < data.length; i += BUFFER_CHUNK) {
|
|
ws.send(output(sub.connection, data.slice(i, i + BUFFER_CHUNK)))
|
|
}
|
|
} catch {
|
|
cleanup()
|
|
ws.close()
|
|
return
|
|
}
|
|
}
|
|
|
|
try {
|
|
ws.send(meta(end, sub.connection))
|
|
} catch {
|
|
cleanup()
|
|
ws.close()
|
|
return
|
|
}
|
|
return {
|
|
onMessage: (message: string | Uint8Array | ArrayBuffer) => {
|
|
const next = input(message)
|
|
if (!next?.data) return
|
|
if (next.connection && next.connection !== sub.connection) return
|
|
session.process.write(next.data)
|
|
},
|
|
onClose: () => {
|
|
log.info("client disconnected from session", { id })
|
|
cleanup()
|
|
},
|
|
}
|
|
}
|
|
}
|