Initial implementation of TriggerShell
A Python CLI (typer) that bootstraps Node/pnpm and launches a Next.js 16 web app for running configured shell scripts: YAML config validated by a shared Zod schema, dynamic per-script forms mapped to shadcn controls, argv-safe execa execution with live WebSocket streaming, SQLite/Drizzle run history, and optional argon2 session + API token auth. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 5
@@ -0,0 +1,73 @@
|
||||
import { eq } from "drizzle-orm";
|
||||
import { getConfig } from "../config/load";
|
||||
import { getDb } from "../db/client";
|
||||
import { apiTokens } from "../db/schema";
|
||||
import { getSession } from "./session";
|
||||
import { hashToken, verifyTokenHash } from "./tokens";
|
||||
|
||||
export interface AuthContext {
|
||||
authenticated: boolean;
|
||||
/** display identity: a username, `api-token:<name>`, or "anonymous" when auth is disabled */
|
||||
identity: string | null;
|
||||
via: "session" | "token" | "disabled" | null;
|
||||
}
|
||||
|
||||
const anonymous: AuthContext = {
|
||||
authenticated: true,
|
||||
identity: "anonymous",
|
||||
via: "disabled",
|
||||
};
|
||||
|
||||
function checkBearerToken(request: Request): AuthContext | null {
|
||||
const header = request.headers.get("authorization");
|
||||
if (!header?.startsWith("Bearer ")) return null;
|
||||
|
||||
const token = header.slice("Bearer ".length).trim();
|
||||
if (!token) return null;
|
||||
|
||||
const candidateHash = hashToken(token);
|
||||
const db = getDb();
|
||||
const match = db
|
||||
.select()
|
||||
.from(apiTokens)
|
||||
.all()
|
||||
.find((row) => verifyTokenHash(candidateHash, row.tokenHash));
|
||||
|
||||
if (!match) return null;
|
||||
|
||||
db.update(apiTokens)
|
||||
.set({ lastUsedAt: new Date() })
|
||||
.where(eq(apiTokens.id, match.id))
|
||||
.run();
|
||||
return {
|
||||
authenticated: true,
|
||||
identity: `api-token:${match.name}`,
|
||||
via: "token",
|
||||
};
|
||||
}
|
||||
|
||||
/** Called at the top of every Route Handler as defense-in-depth, independent of `proxy.ts`. */
|
||||
export async function requireAuth(request?: Request): Promise<AuthContext> {
|
||||
const { config } = getConfig();
|
||||
if (!config.auth.enabled) return anonymous;
|
||||
|
||||
if (request) {
|
||||
const tokenResult = checkBearerToken(request);
|
||||
if (tokenResult) return tokenResult;
|
||||
}
|
||||
|
||||
const session = await getSession();
|
||||
if (session.userId) {
|
||||
return {
|
||||
authenticated: true,
|
||||
identity: session.username ?? session.userId,
|
||||
via: "session",
|
||||
};
|
||||
}
|
||||
|
||||
return { authenticated: false, identity: null, via: null };
|
||||
}
|
||||
|
||||
export function unauthorizedResponse(): Response {
|
||||
return Response.json({ error: "Unauthorized" }, { status: 401 });
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
import argon2 from "argon2";
|
||||
|
||||
export async function hashPassword(password: string): Promise<string> {
|
||||
return argon2.hash(password, { type: argon2.argon2id });
|
||||
}
|
||||
|
||||
export async function verifyPassword(
|
||||
hash: string,
|
||||
password: string,
|
||||
): Promise<boolean> {
|
||||
try {
|
||||
return await argon2.verify(hash, password);
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
const attempts = new Map<string, { count: number; resetAt: number }>();
|
||||
|
||||
const MAX_ATTEMPTS = 10;
|
||||
const WINDOW_MS = 5 * 60 * 1000;
|
||||
|
||||
/** Simple in-memory per-key rate limit; adequate for a local/trusted tool, no external store needed. */
|
||||
export function isRateLimited(key: string): boolean {
|
||||
const entry = attempts.get(key);
|
||||
const now = Date.now();
|
||||
if (!entry || entry.resetAt < now) return false;
|
||||
return entry.count >= MAX_ATTEMPTS;
|
||||
}
|
||||
|
||||
export function recordFailedAttempt(key: string) {
|
||||
const now = Date.now();
|
||||
const entry = attempts.get(key);
|
||||
if (!entry || entry.resetAt < now) {
|
||||
attempts.set(key, { count: 1, resetAt: now + WINDOW_MS });
|
||||
return;
|
||||
}
|
||||
entry.count += 1;
|
||||
}
|
||||
|
||||
export function clearAttempts(key: string) {
|
||||
attempts.delete(key);
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
import { cookies } from "next/headers";
|
||||
import { getIronSession, unsealData, type SessionOptions } from "iron-session";
|
||||
import { getConfig } from "../config/load";
|
||||
|
||||
export interface SessionData {
|
||||
userId?: string;
|
||||
username?: string;
|
||||
}
|
||||
|
||||
export const SESSION_COOKIE_NAME = "triggershell_session";
|
||||
|
||||
export function getSessionOptions(): SessionOptions {
|
||||
const { config } = getConfig();
|
||||
return {
|
||||
cookieName: SESSION_COOKIE_NAME,
|
||||
password: config.auth.sessionSecret ?? "",
|
||||
ttl: Math.round(config.auth.sessionTtlHours * 3600),
|
||||
cookieOptions: {
|
||||
httpOnly: true,
|
||||
secure: process.env.NODE_ENV === "production",
|
||||
sameSite: "lax",
|
||||
path: "/",
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/** For use inside Route Handlers / Server Components / Server Actions only. */
|
||||
export async function getSession() {
|
||||
return getIronSession<SessionData>(await cookies(), getSessionOptions());
|
||||
}
|
||||
|
||||
/** For contexts outside Next's request pipeline - the raw WS `upgrade` handler in server.ts. */
|
||||
export async function verifySessionCookieValue(
|
||||
sealed: string | undefined,
|
||||
): Promise<SessionData | null> {
|
||||
if (!sealed) return null;
|
||||
try {
|
||||
const options = getSessionOptions();
|
||||
const data = await unsealData<SessionData>(sealed, {
|
||||
password: options.password,
|
||||
ttl: options.ttl,
|
||||
});
|
||||
return data.userId ? data : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export function extractCookieValue(
|
||||
cookieHeader: string | undefined,
|
||||
name: string,
|
||||
): string | undefined {
|
||||
if (!cookieHeader) return undefined;
|
||||
for (const part of cookieHeader.split(";")) {
|
||||
const separatorIndex = part.indexOf("=");
|
||||
if (separatorIndex === -1) continue;
|
||||
const key = part.slice(0, separatorIndex).trim();
|
||||
if (key === name)
|
||||
return decodeURIComponent(part.slice(separatorIndex + 1).trim());
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
import { eq } from "drizzle-orm";
|
||||
import { getDb } from "../db/client";
|
||||
import { users, apiTokens } from "../db/schema";
|
||||
import { getConfig } from "../config/load";
|
||||
|
||||
/** Config is the source of truth for who's allowed in; this materializes it into SQLite so the
|
||||
* runtime auth-check path is uniform and `lastLoginAt`/`lastUsedAt` can be tracked. Called on boot. */
|
||||
export function syncAuthFromConfig() {
|
||||
const { config } = getConfig();
|
||||
if (!config.auth.enabled) return;
|
||||
|
||||
const db = getDb();
|
||||
|
||||
for (const configUser of config.auth.users) {
|
||||
const existing = db
|
||||
.select()
|
||||
.from(users)
|
||||
.where(eq(users.username, configUser.username))
|
||||
.get();
|
||||
if (existing) {
|
||||
if (existing.passwordHash !== configUser.passwordHash) {
|
||||
db.update(users)
|
||||
.set({ passwordHash: configUser.passwordHash })
|
||||
.where(eq(users.id, existing.id))
|
||||
.run();
|
||||
}
|
||||
} else {
|
||||
db.insert(users)
|
||||
.values({
|
||||
username: configUser.username,
|
||||
passwordHash: configUser.passwordHash,
|
||||
})
|
||||
.run();
|
||||
}
|
||||
}
|
||||
|
||||
for (const configToken of config.auth.tokens) {
|
||||
const existing = db
|
||||
.select()
|
||||
.from(apiTokens)
|
||||
.where(eq(apiTokens.name, configToken.name))
|
||||
.get();
|
||||
if (existing) {
|
||||
if (existing.tokenHash !== configToken.tokenHash) {
|
||||
db.update(apiTokens)
|
||||
.set({ tokenHash: configToken.tokenHash })
|
||||
.where(eq(apiTokens.id, existing.id))
|
||||
.run();
|
||||
}
|
||||
} else {
|
||||
db.insert(apiTokens)
|
||||
.values({ name: configToken.name, tokenHash: configToken.tokenHash })
|
||||
.run();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
import crypto from "node:crypto";
|
||||
|
||||
/** Config/DB store `sha256:<hex>` - opaque tokens are never stored in reversible form. */
|
||||
export function hashToken(token: string): string {
|
||||
return `sha256:${crypto.createHash("sha256").update(token).digest("hex")}`;
|
||||
}
|
||||
|
||||
export function verifyTokenHash(
|
||||
candidateHash: string,
|
||||
storedHash: string,
|
||||
): boolean {
|
||||
const a = Buffer.from(candidateHash);
|
||||
const b = Buffer.from(storedHash);
|
||||
if (a.length !== b.length) return false;
|
||||
return crypto.timingSafeEqual(a, b);
|
||||
}
|
||||
|
||||
export function generateToken(): string {
|
||||
return crypto.randomBytes(32).toString("hex");
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import { parse as parseYaml } from "yaml";
|
||||
import { configSchema, type TriggerShellConfig } from "./schema";
|
||||
|
||||
export class ConfigError extends Error {
|
||||
issues: string[];
|
||||
|
||||
constructor(message: string, issues: string[] = []) {
|
||||
super(message);
|
||||
this.name = "ConfigError";
|
||||
this.issues = issues;
|
||||
}
|
||||
}
|
||||
|
||||
/** Resolves `${VAR}` / `${VAR:-default}` references against process.env. */
|
||||
function interpolateEnv(raw: string): string {
|
||||
return raw.replace(
|
||||
/\$\{([A-Z0-9_]+)(:-([^}]*))?\}/gi,
|
||||
(_match, name: string, _hasDefault, fallback: string) => {
|
||||
const value = process.env[name];
|
||||
if (value !== undefined && value !== "") return value;
|
||||
if (fallback !== undefined) return fallback;
|
||||
return "";
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
export interface LoadedConfig {
|
||||
config: TriggerShellConfig;
|
||||
configPath: string;
|
||||
configDir: string;
|
||||
dbPath: string;
|
||||
logsDir: string;
|
||||
}
|
||||
|
||||
export function resolveConfigPath(configPathArg?: string): string {
|
||||
const candidate =
|
||||
configPathArg ??
|
||||
process.env.TRIGGERSHELL_CONFIG_PATH ??
|
||||
"triggershell.config.yaml";
|
||||
return path.resolve(/*turbopackIgnore: true*/ candidate);
|
||||
}
|
||||
|
||||
export function loadConfig(configPathArg?: string): LoadedConfig {
|
||||
const configPath = resolveConfigPath(configPathArg);
|
||||
|
||||
// This path is resolved at runtime from a user-supplied config location, never known at build
|
||||
// time - see the `--skip-build`/tracing note in docs/ARCHITECTURE.md.
|
||||
if (!fs.existsSync(/*turbopackIgnore: true*/ configPath)) {
|
||||
throw new ConfigError(`Config file not found at ${configPath}`);
|
||||
}
|
||||
|
||||
const raw = fs.readFileSync(/*turbopackIgnore: true*/ configPath, "utf-8");
|
||||
const interpolated = interpolateEnv(raw);
|
||||
|
||||
let parsedYaml: unknown;
|
||||
try {
|
||||
parsedYaml = parseYaml(interpolated);
|
||||
} catch (error) {
|
||||
throw new ConfigError(`Failed to parse YAML: ${(error as Error).message}`);
|
||||
}
|
||||
|
||||
const result = configSchema.safeParse(parsedYaml);
|
||||
if (!result.success) {
|
||||
const issues = result.error.issues.map(
|
||||
(issue) => `${issue.path.join(".") || "(root)"}: ${issue.message}`,
|
||||
);
|
||||
throw new ConfigError("Config validation failed", issues);
|
||||
}
|
||||
|
||||
const configDir = path.dirname(configPath);
|
||||
const config = result.data;
|
||||
|
||||
return {
|
||||
config,
|
||||
configPath,
|
||||
configDir,
|
||||
dbPath: path.resolve(configDir, config.database.path),
|
||||
logsDir: path.resolve(configDir, config.logs.dir),
|
||||
};
|
||||
}
|
||||
|
||||
declare global {
|
||||
var __triggershellConfig: LoadedConfig | undefined;
|
||||
}
|
||||
|
||||
// Anchored on `globalThis` - see the comment in `runner/events.ts` for why: Next compiles Route
|
||||
// Handlers through a separate module graph from what `server.ts` requires directly, so a plain
|
||||
// module-level singleton would reparse the config file a second time instead of reusing one.
|
||||
/** Loads once per process and caches the result; the custom server restarts the process on config edits. */
|
||||
export function getConfig(): LoadedConfig {
|
||||
if (!globalThis.__triggershellConfig) {
|
||||
globalThis.__triggershellConfig = loadConfig();
|
||||
}
|
||||
return globalThis.__triggershellConfig;
|
||||
}
|
||||
|
||||
export function getScript(
|
||||
scriptId: string,
|
||||
): TriggerShellConfig["scripts"][number] | undefined {
|
||||
return getConfig().config.scripts.find((script) => script.id === scriptId);
|
||||
}
|
||||
@@ -0,0 +1,271 @@
|
||||
import { z } from "zod";
|
||||
|
||||
const identifier = z
|
||||
.string()
|
||||
.min(1)
|
||||
.regex(/^[a-zA-Z0-9][a-zA-Z0-9_-]*$/, {
|
||||
message:
|
||||
"must start with an alphanumeric character and contain only letters, numbers, - and _",
|
||||
});
|
||||
|
||||
const controlSchema = z.enum([
|
||||
"text",
|
||||
"textarea",
|
||||
"password",
|
||||
"number",
|
||||
"slider",
|
||||
"checkbox",
|
||||
"switch",
|
||||
"select",
|
||||
"radio",
|
||||
"multiselect",
|
||||
"checkboxGroup",
|
||||
]);
|
||||
|
||||
export type ControlType = z.infer<typeof controlSchema>;
|
||||
|
||||
const baseVariable = z.object({
|
||||
name: identifier,
|
||||
label: z.string().min(1).optional(),
|
||||
description: z.string().optional(),
|
||||
required: z.boolean().default(false),
|
||||
secret: z.boolean().default(false),
|
||||
control: controlSchema.optional(),
|
||||
passAs: z.enum(["arg", "flag", "env", "stdin"]).default("arg"),
|
||||
argName: z.string().optional(),
|
||||
envName: z.string().optional(),
|
||||
joinWith: z.string().default(","),
|
||||
});
|
||||
|
||||
const stringVariable = baseVariable.extend({
|
||||
type: z.literal("string"),
|
||||
default: z.string().optional(),
|
||||
pattern: z.string().optional(),
|
||||
minLength: z.number().int().nonnegative().optional(),
|
||||
maxLength: z.number().int().nonnegative().optional(),
|
||||
multiline: z.boolean().default(false),
|
||||
});
|
||||
|
||||
const numberVariable = baseVariable.extend({
|
||||
type: z.literal("number"),
|
||||
default: z.number().optional(),
|
||||
min: z.number().optional(),
|
||||
max: z.number().optional(),
|
||||
step: z.number().positive().optional(),
|
||||
});
|
||||
|
||||
const booleanVariable = baseVariable.extend({
|
||||
type: z.literal("boolean"),
|
||||
default: z.boolean().default(false),
|
||||
});
|
||||
|
||||
const enumVariable = baseVariable.extend({
|
||||
type: z.literal("enum"),
|
||||
default: z.string().optional(),
|
||||
choices: z.array(z.string().min(1)).min(1),
|
||||
});
|
||||
|
||||
const multiselectVariable = baseVariable.extend({
|
||||
type: z.literal("multiselect"),
|
||||
default: z.array(z.string()).default([]),
|
||||
choices: z.array(z.string().min(1)).min(1),
|
||||
});
|
||||
|
||||
export const variableSchema = z.discriminatedUnion("type", [
|
||||
stringVariable,
|
||||
numberVariable,
|
||||
booleanVariable,
|
||||
enumVariable,
|
||||
multiselectVariable,
|
||||
]);
|
||||
|
||||
export type VariableConfig = z.infer<typeof variableSchema>;
|
||||
|
||||
const variableWithChecks = variableSchema.superRefine((variable, ctx) => {
|
||||
if (variable.type === "string" && variable.pattern) {
|
||||
try {
|
||||
new RegExp(variable.pattern);
|
||||
} catch {
|
||||
ctx.addIssue({
|
||||
code: "custom",
|
||||
message: `invalid regular expression: ${variable.pattern}`,
|
||||
path: ["pattern"],
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
if (
|
||||
variable.type === "number" &&
|
||||
variable.min !== undefined &&
|
||||
variable.max !== undefined
|
||||
) {
|
||||
if (variable.min > variable.max) {
|
||||
ctx.addIssue({
|
||||
code: "custom",
|
||||
message: "min must be <= max",
|
||||
path: ["min"],
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
if (variable.control === "slider" && variable.type !== "number") {
|
||||
ctx.addIssue({
|
||||
code: "custom",
|
||||
message: "control 'slider' requires type 'number'",
|
||||
path: ["control"],
|
||||
});
|
||||
}
|
||||
if (
|
||||
variable.control === "slider" &&
|
||||
variable.type === "number" &&
|
||||
(variable.min === undefined || variable.max === undefined)
|
||||
) {
|
||||
ctx.addIssue({
|
||||
code: "custom",
|
||||
message: "control 'slider' requires both min and max",
|
||||
path: ["control"],
|
||||
});
|
||||
}
|
||||
|
||||
if (variable.passAs === "arg" && !variable.argName) {
|
||||
ctx.addIssue({
|
||||
code: "custom",
|
||||
message: "passAs 'arg' requires argName",
|
||||
path: ["argName"],
|
||||
});
|
||||
}
|
||||
if (variable.passAs === "flag" && !variable.argName) {
|
||||
ctx.addIssue({
|
||||
code: "custom",
|
||||
message: "passAs 'flag' requires argName",
|
||||
path: ["argName"],
|
||||
});
|
||||
}
|
||||
if (variable.passAs === "flag" && variable.type !== "boolean") {
|
||||
ctx.addIssue({
|
||||
code: "custom",
|
||||
message: "passAs 'flag' requires type 'boolean'",
|
||||
path: ["passAs"],
|
||||
});
|
||||
}
|
||||
if (variable.passAs === "env" && !variable.envName) {
|
||||
ctx.addIssue({
|
||||
code: "custom",
|
||||
message: "passAs 'env' requires envName",
|
||||
path: ["envName"],
|
||||
});
|
||||
}
|
||||
|
||||
if (variable.secret && variable.type !== "string") {
|
||||
ctx.addIssue({
|
||||
code: "custom",
|
||||
message: "secret variables must be of type 'string'",
|
||||
path: ["secret"],
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
const scriptSchema = z
|
||||
.object({
|
||||
id: identifier,
|
||||
name: z.string().min(1),
|
||||
description: z.string().optional(),
|
||||
command: z.string().min(1),
|
||||
args: z.array(z.string()).default([]),
|
||||
cwd: z.string().default("./"),
|
||||
shell: z.boolean().default(false),
|
||||
timeoutSeconds: z.number().int().positive().max(86400).default(1800),
|
||||
variables: z.array(variableWithChecks).default([]),
|
||||
})
|
||||
.superRefine((script, ctx) => {
|
||||
const seen = new Set<string>();
|
||||
for (const [index, variable] of script.variables.entries()) {
|
||||
if (seen.has(variable.name)) {
|
||||
ctx.addIssue({
|
||||
code: "custom",
|
||||
message: `duplicate variable name '${variable.name}' in script '${script.id}'`,
|
||||
path: ["variables", index, "name"],
|
||||
});
|
||||
}
|
||||
seen.add(variable.name);
|
||||
}
|
||||
});
|
||||
|
||||
export type ScriptConfig = z.infer<typeof scriptSchema>;
|
||||
|
||||
const userSchema = z.object({
|
||||
username: identifier,
|
||||
passwordHash: z.string().min(1),
|
||||
});
|
||||
|
||||
const tokenSchema = z.object({
|
||||
name: identifier,
|
||||
tokenHash: z.string().min(1),
|
||||
});
|
||||
|
||||
const authSchema = z
|
||||
.object({
|
||||
enabled: z.boolean().default(true),
|
||||
sessionSecret: z.string().optional(),
|
||||
sessionTtlHours: z.number().positive().default(12),
|
||||
users: z.array(userSchema).default([]),
|
||||
tokens: z.array(tokenSchema).default([]),
|
||||
})
|
||||
.superRefine((auth, ctx) => {
|
||||
if (auth.enabled) {
|
||||
if (!auth.sessionSecret || auth.sessionSecret.length < 32) {
|
||||
ctx.addIssue({
|
||||
code: "custom",
|
||||
message:
|
||||
"auth.sessionSecret must be set and at least 32 characters when auth is enabled",
|
||||
path: ["sessionSecret"],
|
||||
});
|
||||
}
|
||||
if (auth.users.length === 0 && auth.tokens.length === 0) {
|
||||
ctx.addIssue({
|
||||
code: "custom",
|
||||
message: "auth.enabled is true but no users or tokens are configured",
|
||||
path: ["users"],
|
||||
});
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
const serverSchema = z.object({
|
||||
host: z.string().default("127.0.0.1"),
|
||||
port: z.number().int().positive().max(65535).default(4173),
|
||||
basePath: z.string().default(""),
|
||||
});
|
||||
|
||||
const databaseSchema = z.object({
|
||||
path: z.string().default(".triggershell/triggershell.db"),
|
||||
});
|
||||
|
||||
const logsSchema = z.object({
|
||||
dir: z.string().default(".triggershell/logs"),
|
||||
retentionDays: z.number().int().positive().default(30),
|
||||
});
|
||||
|
||||
export const configSchema = z
|
||||
.object({
|
||||
server: serverSchema.prefault({}),
|
||||
auth: authSchema.prefault({}),
|
||||
database: databaseSchema.prefault({}),
|
||||
logs: logsSchema.prefault({}),
|
||||
scripts: z.array(scriptSchema).default([]),
|
||||
})
|
||||
.superRefine((config, ctx) => {
|
||||
const seen = new Set<string>();
|
||||
for (const [index, script] of config.scripts.entries()) {
|
||||
if (seen.has(script.id)) {
|
||||
ctx.addIssue({
|
||||
code: "custom",
|
||||
message: `duplicate script id '${script.id}'`,
|
||||
path: ["scripts", index, "id"],
|
||||
});
|
||||
}
|
||||
seen.add(script.id);
|
||||
}
|
||||
});
|
||||
|
||||
export type TriggerShellConfig = z.infer<typeof configSchema>;
|
||||
@@ -0,0 +1,32 @@
|
||||
import type { ScriptConfig } from "./schema";
|
||||
import { resolveControl } from "./ui-control-map";
|
||||
|
||||
/** Client-safe view of a script: strips defaults for `secret` variables so nothing sensitive
|
||||
* ever reaches the browser, and resolves the effective UI control for each variable. */
|
||||
export function serializeScriptForClient(script: ScriptConfig) {
|
||||
return {
|
||||
id: script.id,
|
||||
name: script.name,
|
||||
description: script.description ?? null,
|
||||
variables: script.variables.map((variable) => {
|
||||
const control = resolveControl(variable);
|
||||
// `secret` is only ever true for the string variant (enforced by the config schema), so this
|
||||
// narrows `variable` and keeps `default` type-correct for every other variant.
|
||||
if (variable.type === "string" && variable.secret) {
|
||||
return { ...variable, default: undefined, control };
|
||||
}
|
||||
return { ...variable, control };
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
export type ClientScript = ReturnType<typeof serializeScriptForClient>;
|
||||
export type ClientVariable = ClientScript["variables"][number];
|
||||
|
||||
export function serializeScriptSummary(script: ScriptConfig) {
|
||||
return {
|
||||
id: script.id,
|
||||
name: script.name,
|
||||
description: script.description ?? null,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
import type { ControlType, VariableConfig } from "./schema";
|
||||
|
||||
/** Resolves the effective shadcn control for a variable: explicit `control` wins, otherwise a type-based default. */
|
||||
export function resolveControl(variable: VariableConfig): ControlType {
|
||||
if (variable.control) return variable.control;
|
||||
|
||||
switch (variable.type) {
|
||||
case "string":
|
||||
if (variable.secret) return "password";
|
||||
if (variable.multiline) return "textarea";
|
||||
return "text";
|
||||
case "number":
|
||||
return "number";
|
||||
case "boolean":
|
||||
return "checkbox";
|
||||
case "enum":
|
||||
return "select";
|
||||
case "multiselect":
|
||||
return "multiselect";
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import Database from "better-sqlite3";
|
||||
import { drizzle } from "drizzle-orm/better-sqlite3";
|
||||
import { migrate } from "drizzle-orm/better-sqlite3/migrator";
|
||||
import * as schema from "./schema";
|
||||
import { getConfig } from "../config/load";
|
||||
|
||||
type Db = ReturnType<typeof drizzle<typeof schema>>;
|
||||
|
||||
declare global {
|
||||
var __triggershellDb: Db | undefined;
|
||||
}
|
||||
|
||||
// Anchored on `globalThis` - see the comment in `runner/events.ts` for why: Next compiles Route
|
||||
// Handlers through a separate module graph from what `server.ts` requires directly, so a plain
|
||||
// module-level singleton would open a second, wasteful SQLite connection instead of reusing one.
|
||||
export function getDb(): Db {
|
||||
if (globalThis.__triggershellDb) return globalThis.__triggershellDb;
|
||||
|
||||
const { dbPath } = getConfig();
|
||||
fs.mkdirSync(path.dirname(dbPath), { recursive: true });
|
||||
|
||||
const sqlite = new Database(dbPath);
|
||||
sqlite.pragma("journal_mode = WAL");
|
||||
sqlite.pragma("foreign_keys = ON");
|
||||
|
||||
globalThis.__triggershellDb = drizzle(sqlite, { schema });
|
||||
return globalThis.__triggershellDb;
|
||||
}
|
||||
|
||||
/** Applies committed migrations; safe to call on every boot. */
|
||||
export function migrateOnBoot() {
|
||||
const db = getDb();
|
||||
const migrationsFolder = path.resolve(
|
||||
import.meta.dirname,
|
||||
"../../../drizzle",
|
||||
);
|
||||
migrate(db, { migrationsFolder });
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
import { sqliteTable, text, integer, index } from "drizzle-orm/sqlite-core";
|
||||
|
||||
export const users = sqliteTable("users", {
|
||||
id: text("id")
|
||||
.primaryKey()
|
||||
.$defaultFn(() => crypto.randomUUID()),
|
||||
username: text("username").notNull().unique(),
|
||||
passwordHash: text("password_hash").notNull(),
|
||||
createdAt: integer("created_at", { mode: "timestamp" })
|
||||
.notNull()
|
||||
.$defaultFn(() => new Date()),
|
||||
lastLoginAt: integer("last_login_at", { mode: "timestamp" }),
|
||||
});
|
||||
|
||||
export const runStatusValues = [
|
||||
"queued",
|
||||
"running",
|
||||
"succeeded",
|
||||
"failed",
|
||||
"cancelled",
|
||||
"timed_out",
|
||||
"interrupted",
|
||||
] as const;
|
||||
export type RunStatus = (typeof runStatusValues)[number];
|
||||
|
||||
export const runs = sqliteTable(
|
||||
"runs",
|
||||
{
|
||||
id: text("id")
|
||||
.primaryKey()
|
||||
.$defaultFn(() => crypto.randomUUID()),
|
||||
scriptId: text("script_id").notNull(),
|
||||
scriptName: text("script_name").notNull(),
|
||||
status: text("status", { enum: runStatusValues })
|
||||
.notNull()
|
||||
.default("queued"),
|
||||
variables: text("variables", { mode: "json" })
|
||||
.notNull()
|
||||
.$type<Record<string, unknown>>(),
|
||||
resolvedCommand: text("resolved_command").notNull(),
|
||||
pid: integer("pid"),
|
||||
exitCode: integer("exit_code"),
|
||||
startedAt: integer("started_at", { mode: "timestamp" }),
|
||||
endedAt: integer("ended_at", { mode: "timestamp" }),
|
||||
timeoutSeconds: integer("timeout_seconds"),
|
||||
triggeredBy: text("triggered_by").notNull(),
|
||||
logFilePath: text("log_file_path").notNull(),
|
||||
errorMessage: text("error_message"),
|
||||
createdAt: integer("created_at", { mode: "timestamp" })
|
||||
.notNull()
|
||||
.$defaultFn(() => new Date()),
|
||||
},
|
||||
(table) => [
|
||||
index("runs_script_id_idx").on(table.scriptId),
|
||||
index("runs_status_idx").on(table.status),
|
||||
],
|
||||
);
|
||||
|
||||
export const apiTokens = sqliteTable("api_tokens", {
|
||||
id: text("id")
|
||||
.primaryKey()
|
||||
.$defaultFn(() => crypto.randomUUID()),
|
||||
name: text("name").notNull().unique(),
|
||||
tokenHash: text("token_hash").notNull(),
|
||||
createdAt: integer("created_at", { mode: "timestamp" })
|
||||
.notNull()
|
||||
.$defaultFn(() => new Date()),
|
||||
lastUsedAt: integer("last_used_at", { mode: "timestamp" }),
|
||||
});
|
||||
|
||||
export type Run = typeof runs.$inferSelect;
|
||||
export type NewRun = typeof runs.$inferInsert;
|
||||
export type User = typeof users.$inferSelect;
|
||||
export type ApiToken = typeof apiTokens.$inferSelect;
|
||||
@@ -0,0 +1,80 @@
|
||||
import type { ScriptConfig } from "../config/schema";
|
||||
|
||||
export interface Invocation {
|
||||
argv: string[];
|
||||
env: Record<string, string>;
|
||||
stdin?: string;
|
||||
/** variable values with `secret: true` fields replaced, safe to persist/display */
|
||||
redactedVariables: Record<string, unknown>;
|
||||
/** human-readable command line with secrets redacted, safe to persist/display */
|
||||
redactedCommandLine: string;
|
||||
}
|
||||
|
||||
const REDACTED = "***";
|
||||
|
||||
function stringifyValue(value: unknown, joinWith: string): string {
|
||||
if (Array.isArray(value)) return value.join(joinWith);
|
||||
if (typeof value === "boolean") return String(value);
|
||||
return String(value);
|
||||
}
|
||||
|
||||
/** Builds an argv-array invocation from validated variable values. Never produces a shell string. */
|
||||
export function buildInvocation(
|
||||
script: ScriptConfig,
|
||||
values: Record<string, unknown>,
|
||||
): Invocation {
|
||||
const argv = [...script.args];
|
||||
const env: Record<string, string> = {};
|
||||
const redactedArgv = [...script.args];
|
||||
const redactedVariables: Record<string, unknown> = {};
|
||||
|
||||
for (const variable of script.variables) {
|
||||
const raw = values[variable.name] ?? variable.default;
|
||||
if (raw === undefined || raw === null || raw === "") {
|
||||
redactedVariables[variable.name] = raw;
|
||||
continue;
|
||||
}
|
||||
|
||||
redactedVariables[variable.name] = variable.secret ? REDACTED : raw;
|
||||
|
||||
switch (variable.passAs) {
|
||||
case "arg": {
|
||||
const argName = variable.argName!;
|
||||
const value = stringifyValue(raw, variable.joinWith);
|
||||
argv.push(argName, value);
|
||||
redactedArgv.push(argName, variable.secret ? REDACTED : value);
|
||||
break;
|
||||
}
|
||||
case "flag": {
|
||||
if (raw === true) {
|
||||
argv.push(variable.argName!);
|
||||
redactedArgv.push(variable.argName!);
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "env": {
|
||||
const value = stringifyValue(raw, variable.joinWith);
|
||||
env[variable.envName!] = value;
|
||||
break;
|
||||
}
|
||||
case "stdin": {
|
||||
// handled by caller via the returned `stdin` field; only one stdin variable is meaningful
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const stdinVariable = script.variables.find(
|
||||
(variable) => variable.passAs === "stdin",
|
||||
);
|
||||
const stdin = stdinVariable
|
||||
? stringifyValue(
|
||||
values[stdinVariable.name] ?? stdinVariable.default,
|
||||
stdinVariable.joinWith,
|
||||
)
|
||||
: undefined;
|
||||
|
||||
const redactedCommandLine = [script.command, ...redactedArgv].join(" ");
|
||||
|
||||
return { argv, env, stdin, redactedVariables, redactedCommandLine };
|
||||
}
|
||||
@@ -0,0 +1,172 @@
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import { execa } from "execa";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { getDb } from "../db/client";
|
||||
import { runs, type RunStatus } from "../db/schema";
|
||||
import { getConfig, getScript } from "../config/load";
|
||||
import type { ScriptConfig } from "../config/schema";
|
||||
import { buildInvocation, type Invocation } from "./build-args";
|
||||
import { registerRun, unregisterRun } from "./registry";
|
||||
import { emitRunMessage } from "./events";
|
||||
|
||||
export class ScriptNotFoundError extends Error {}
|
||||
|
||||
interface StartRunOptions {
|
||||
scriptId: string;
|
||||
variables: Record<string, unknown>;
|
||||
triggeredBy: string;
|
||||
}
|
||||
|
||||
export async function startRun({
|
||||
scriptId,
|
||||
variables,
|
||||
triggeredBy,
|
||||
}: StartRunOptions): Promise<string> {
|
||||
const script = getScript(scriptId);
|
||||
if (!script) throw new ScriptNotFoundError(`Unknown script '${scriptId}'`);
|
||||
|
||||
const { logsDir } = getConfig();
|
||||
fs.mkdirSync(logsDir, { recursive: true });
|
||||
|
||||
const invocation = buildInvocation(script, variables);
|
||||
const db = getDb();
|
||||
|
||||
const runId = crypto.randomUUID();
|
||||
const logFilePath = path.join(logsDir, `${runId}.log`);
|
||||
|
||||
db.insert(runs)
|
||||
.values({
|
||||
id: runId,
|
||||
scriptId: script.id,
|
||||
scriptName: script.name,
|
||||
status: "queued",
|
||||
variables: invocation.redactedVariables,
|
||||
resolvedCommand: invocation.redactedCommandLine,
|
||||
timeoutSeconds: script.timeoutSeconds,
|
||||
triggeredBy,
|
||||
logFilePath,
|
||||
})
|
||||
.run();
|
||||
|
||||
// Fire and forget - the caller gets the runId immediately, progress streams over WS/polling.
|
||||
void executeRun(runId, script, invocation, logFilePath).catch((error) => {
|
||||
console.error(`[runner] unhandled error executing run ${runId}:`, error);
|
||||
});
|
||||
|
||||
return runId;
|
||||
}
|
||||
|
||||
async function executeRun(
|
||||
runId: string,
|
||||
script: ScriptConfig,
|
||||
invocation: Invocation,
|
||||
logFilePath: string,
|
||||
) {
|
||||
const db = getDb();
|
||||
const logStream = fs.createWriteStream(logFilePath, { flags: "a" });
|
||||
|
||||
const setStatus = (
|
||||
status: RunStatus,
|
||||
extra: Partial<typeof runs.$inferInsert> = {},
|
||||
) => {
|
||||
db.update(runs)
|
||||
.set({ status, ...extra })
|
||||
.where(eq(runs.id, runId))
|
||||
.run();
|
||||
emitRunMessage({
|
||||
type: "status",
|
||||
runId,
|
||||
status,
|
||||
exitCode: (extra.exitCode as number | null | undefined) ?? null,
|
||||
ts: Date.now(),
|
||||
});
|
||||
};
|
||||
|
||||
const { configDir } = getConfig();
|
||||
const cwd = path.resolve(configDir, script.cwd);
|
||||
const controller = new AbortController();
|
||||
|
||||
let seq = 0;
|
||||
const onChunk = (stream: "stdout" | "stderr") => (data: Buffer) => {
|
||||
const chunk = data.toString("utf-8");
|
||||
logStream.write(chunk);
|
||||
emitRunMessage({
|
||||
type: "output",
|
||||
runId,
|
||||
stream,
|
||||
chunk,
|
||||
seq: seq++,
|
||||
ts: Date.now(),
|
||||
});
|
||||
};
|
||||
|
||||
registerRun({ runId, scriptId: script.id, controller });
|
||||
setStatus("running", { startedAt: new Date() });
|
||||
|
||||
try {
|
||||
const subprocess = execa(script.command, invocation.argv, {
|
||||
cwd,
|
||||
env: { ...process.env, ...invocation.env },
|
||||
timeout: script.timeoutSeconds * 1000,
|
||||
cancelSignal: controller.signal,
|
||||
reject: false,
|
||||
shell: script.shell,
|
||||
input: invocation.stdin,
|
||||
buffer: false,
|
||||
});
|
||||
|
||||
subprocess.stdout?.on("data", onChunk("stdout"));
|
||||
subprocess.stderr?.on("data", onChunk("stderr"));
|
||||
|
||||
const result = await subprocess;
|
||||
|
||||
let status: RunStatus;
|
||||
if (result.isCanceled) status = "cancelled";
|
||||
else if (result.timedOut) status = "timed_out";
|
||||
else if (result.failed) status = "failed";
|
||||
else status = "succeeded";
|
||||
|
||||
setStatus(status, {
|
||||
exitCode: result.exitCode ?? null,
|
||||
endedAt: new Date(),
|
||||
errorMessage:
|
||||
status === "failed" || status === "timed_out"
|
||||
? (result.shortMessage ?? null)
|
||||
: null,
|
||||
});
|
||||
} catch (error) {
|
||||
setStatus("failed", {
|
||||
endedAt: new Date(),
|
||||
errorMessage: (error as Error).message,
|
||||
});
|
||||
} finally {
|
||||
logStream.end();
|
||||
unregisterRun(runId);
|
||||
}
|
||||
}
|
||||
|
||||
/** On boot, any DB row still `running`/`queued` has no live handle in this process - mark it interrupted
|
||||
* rather than pretending we can resume streaming its output. */
|
||||
export function reconcileOrphanedRuns() {
|
||||
const db = getDb();
|
||||
const now = new Date();
|
||||
const orphaned = db
|
||||
.update(runs)
|
||||
.set({
|
||||
status: "interrupted",
|
||||
endedAt: now,
|
||||
errorMessage: "Server restarted while this run was in progress.",
|
||||
})
|
||||
.where(eq(runs.status, "running"))
|
||||
.run();
|
||||
db.update(runs)
|
||||
.set({
|
||||
status: "interrupted",
|
||||
endedAt: now,
|
||||
errorMessage: "Server restarted before this run could start.",
|
||||
})
|
||||
.where(eq(runs.status, "queued"))
|
||||
.run();
|
||||
return orphaned;
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
import { EventEmitter } from "node:events";
|
||||
import type { ServerMessage } from "../ws/protocol";
|
||||
|
||||
declare global {
|
||||
var __triggershellRunEvents: EventEmitter | undefined;
|
||||
}
|
||||
|
||||
/** Decouples the run engine from the WS transport: the engine emits, ws/server.ts broadcasts.
|
||||
* Anchored on `globalThis` because Next compiles Route Handlers through its own module graph,
|
||||
* separate from the modules `server.ts` requires directly via tsx - a plain module-level
|
||||
* singleton would silently end up duplicated (one copy per graph) instead of shared. */
|
||||
export const runEvents: EventEmitter = globalThis.__triggershellRunEvents ?? new EventEmitter();
|
||||
globalThis.__triggershellRunEvents = runEvents;
|
||||
runEvents.setMaxListeners(0);
|
||||
|
||||
export function emitRunMessage(message: ServerMessage) {
|
||||
runEvents.emit("message", message);
|
||||
}
|
||||
@@ -0,0 +1,47 @@
|
||||
interface RunHandle {
|
||||
runId: string;
|
||||
scriptId: string;
|
||||
controller: AbortController;
|
||||
pid?: number;
|
||||
}
|
||||
|
||||
declare global {
|
||||
var __triggershellRunHandles: Map<string, RunHandle> | undefined;
|
||||
}
|
||||
|
||||
/** In-memory only - live run handles cannot survive a process restart; see `reconcileOrphanedRuns`.
|
||||
* Anchored on `globalThis` - see the comment in `runner/events.ts` for why a plain module-level
|
||||
* singleton isn't safe here (Next compiles Route Handlers through a separate module graph from
|
||||
* what `server.ts` requires directly). */
|
||||
const handles: Map<string, RunHandle> = globalThis.__triggershellRunHandles ?? new Map();
|
||||
globalThis.__triggershellRunHandles = handles;
|
||||
|
||||
export function registerRun(handle: RunHandle) {
|
||||
handles.set(handle.runId, handle);
|
||||
}
|
||||
|
||||
export function unregisterRun(runId: string) {
|
||||
handles.delete(runId);
|
||||
}
|
||||
|
||||
export function getRunHandle(runId: string): RunHandle | undefined {
|
||||
return handles.get(runId);
|
||||
}
|
||||
|
||||
export function isRunLive(runId: string): boolean {
|
||||
return handles.has(runId);
|
||||
}
|
||||
|
||||
/** Requests cancellation; returns false if the run isn't tracked (already finished or not this process). */
|
||||
export function cancelRun(runId: string): boolean {
|
||||
const handle = handles.get(runId);
|
||||
if (!handle) return false;
|
||||
handle.controller.abort();
|
||||
return true;
|
||||
}
|
||||
|
||||
export function killAllRuns() {
|
||||
for (const handle of handles.values()) {
|
||||
handle.controller.abort();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
import { clsx, type ClassValue } from "clsx";
|
||||
import { twMerge } from "tailwind-merge";
|
||||
|
||||
export function cn(...inputs: ClassValue[]) {
|
||||
return twMerge(clsx(inputs));
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
import { z } from "zod";
|
||||
import type { ScriptConfig, VariableConfig } from "../config/schema";
|
||||
|
||||
function fieldSchema(variable: VariableConfig): z.ZodTypeAny {
|
||||
let field: z.ZodTypeAny;
|
||||
|
||||
switch (variable.type) {
|
||||
case "string": {
|
||||
let s = z.string();
|
||||
if (variable.minLength !== undefined) s = s.min(variable.minLength);
|
||||
if (variable.maxLength !== undefined) s = s.max(variable.maxLength);
|
||||
if (variable.pattern) s = s.regex(new RegExp(variable.pattern));
|
||||
field = s;
|
||||
break;
|
||||
}
|
||||
case "number": {
|
||||
let n = z.number();
|
||||
if (variable.min !== undefined) n = n.min(variable.min);
|
||||
if (variable.max !== undefined) n = n.max(variable.max);
|
||||
field = n;
|
||||
break;
|
||||
}
|
||||
case "boolean":
|
||||
field = z.boolean();
|
||||
break;
|
||||
case "enum":
|
||||
field = z.enum(variable.choices as [string, ...string[]]);
|
||||
break;
|
||||
case "multiselect":
|
||||
field = z.array(z.enum(variable.choices as [string, ...string[]]));
|
||||
break;
|
||||
}
|
||||
|
||||
if (!variable.required) {
|
||||
field = field.optional().or(z.literal(""));
|
||||
}
|
||||
|
||||
return field;
|
||||
}
|
||||
|
||||
/** Builds one Zod object schema from a list of variable definitions - the single source of truth
|
||||
* imported by both the client form resolver and the server-side run-creation handler. */
|
||||
export function buildVariableSchemaFromList(
|
||||
variables: readonly VariableConfig[],
|
||||
) {
|
||||
const shape: Record<string, z.ZodTypeAny> = {};
|
||||
for (const variable of variables) {
|
||||
shape[variable.name] = fieldSchema(variable);
|
||||
}
|
||||
return z.object(shape);
|
||||
}
|
||||
|
||||
export function buildVariableSchema(script: ScriptConfig) {
|
||||
return buildVariableSchemaFromList(script.variables);
|
||||
}
|
||||
|
||||
export type VariableValues = Record<string, unknown>;
|
||||
@@ -0,0 +1,35 @@
|
||||
import type { RunStatus } from "../db/schema";
|
||||
|
||||
export type ClientMessage =
|
||||
| { type: "subscribe"; runId: string }
|
||||
| { type: "unsubscribe"; runId: string }
|
||||
| { type: "cancel"; runId: string };
|
||||
|
||||
export type ServerMessage =
|
||||
| {
|
||||
type: "output";
|
||||
runId: string;
|
||||
stream: "stdout" | "stderr";
|
||||
chunk: string;
|
||||
seq: number;
|
||||
ts: number;
|
||||
}
|
||||
| {
|
||||
type: "status";
|
||||
runId: string;
|
||||
status: RunStatus;
|
||||
exitCode?: number | null;
|
||||
ts: number;
|
||||
}
|
||||
| { type: "error"; runId: string; message: string };
|
||||
|
||||
export function isClientMessage(value: unknown): value is ClientMessage {
|
||||
if (typeof value !== "object" || value === null) return false;
|
||||
const message = value as Record<string, unknown>;
|
||||
return (
|
||||
(message.type === "subscribe" ||
|
||||
message.type === "unsubscribe" ||
|
||||
message.type === "cancel") &&
|
||||
typeof message.runId === "string"
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,96 @@
|
||||
import type { IncomingMessage } from "node:http";
|
||||
import { WebSocketServer, WebSocket } from "ws";
|
||||
import { getConfig } from "../config/load";
|
||||
import {
|
||||
extractCookieValue,
|
||||
verifySessionCookieValue,
|
||||
SESSION_COOKIE_NAME,
|
||||
} from "../auth/session";
|
||||
import { hashToken, verifyTokenHash } from "../auth/tokens";
|
||||
import { getDb } from "../db/client";
|
||||
import { apiTokens } from "../db/schema";
|
||||
import { runEvents } from "../runner/events";
|
||||
import { cancelRun } from "../runner/registry";
|
||||
import { isClientMessage, type ServerMessage } from "./protocol";
|
||||
|
||||
const subscriptions = new Map<string, Set<WebSocket>>();
|
||||
|
||||
function subscribe(runId: string, ws: WebSocket) {
|
||||
let set = subscriptions.get(runId);
|
||||
if (!set) {
|
||||
set = new Set();
|
||||
subscriptions.set(runId, set);
|
||||
}
|
||||
set.add(ws);
|
||||
}
|
||||
|
||||
function unsubscribe(runId: string, ws: WebSocket) {
|
||||
subscriptions.get(runId)?.delete(ws);
|
||||
}
|
||||
|
||||
function unsubscribeAll(ws: WebSocket) {
|
||||
for (const set of subscriptions.values()) set.delete(ws);
|
||||
}
|
||||
|
||||
runEvents.on("message", (message: ServerMessage) => {
|
||||
const set = subscriptions.get(message.runId);
|
||||
if (!set || set.size === 0) return;
|
||||
const payload = JSON.stringify(message);
|
||||
for (const ws of set) {
|
||||
if (ws.readyState === WebSocket.OPEN) ws.send(payload);
|
||||
}
|
||||
});
|
||||
|
||||
/** Authenticates the WS upgrade outside Next's normal request pipeline: session cookie first,
|
||||
* falling back to a `?token=` bearer-style query param for non-browser clients. */
|
||||
export async function authenticateUpgrade(
|
||||
req: IncomingMessage,
|
||||
): Promise<boolean> {
|
||||
const { config } = getConfig();
|
||||
if (!config.auth.enabled) return true;
|
||||
|
||||
const url = new URL(req.url ?? "/", "http://internal");
|
||||
const tokenParam = url.searchParams.get("token");
|
||||
if (tokenParam) {
|
||||
const candidateHash = hashToken(tokenParam);
|
||||
const db = getDb();
|
||||
const match = db
|
||||
.select()
|
||||
.from(apiTokens)
|
||||
.all()
|
||||
.some((row) => verifyTokenHash(candidateHash, row.tokenHash));
|
||||
if (match) return true;
|
||||
}
|
||||
|
||||
const sealed = extractCookieValue(req.headers.cookie, SESSION_COOKIE_NAME);
|
||||
const session = await verifySessionCookieValue(sealed);
|
||||
return Boolean(session?.userId);
|
||||
}
|
||||
|
||||
export function attachWsServer(wss: WebSocketServer) {
|
||||
wss.on("connection", (ws: WebSocket) => {
|
||||
ws.on("message", (raw) => {
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(raw.toString());
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
if (!isClientMessage(parsed)) return;
|
||||
|
||||
switch (parsed.type) {
|
||||
case "subscribe":
|
||||
subscribe(parsed.runId, ws);
|
||||
break;
|
||||
case "unsubscribe":
|
||||
unsubscribe(parsed.runId, ws);
|
||||
break;
|
||||
case "cancel":
|
||||
cancelRun(parsed.runId);
|
||||
break;
|
||||
}
|
||||
});
|
||||
|
||||
ws.on("close", () => unsubscribeAll(ws));
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user