feat(api): launches under a session (#422)

Phase 1 of decision 0085, the api piece (see .nestri contract
window-streams).

A box run (session) now holds launches: one game or app each. Billing,
allowance and the free weekend still count sessions; nothing about
limits changes. session_box_active_unique stays.

- Migration 0027: launch table (id lch_..., session_id, game_id, state
requested|running|ended|failed, first, time_started, time_stopped,
time_stop_requested, end_reason, error_message). Backfills a first
launch for every run that has not stopped.
- POST /session: same request; the response data gains launches (the
first launch).
- POST /session/{id}/launch {gameId}: 201 with the launch. Same library
check as POST /session. 409 unless the run is starting or live and not
being stopped.
- POST /session/{id}/launch/{launchId}/stop: marks the launch; host gets
launch.stop until it reports it over. A launch in a run no host has
claimed ends at once.
- POST /session/{id}/launch/{launchId}/state (host, run's claimToken):
running | ended | failed, optional endReason/errorMessage.
- When the last launch is over, the run is stopped through the same path
as POST /session/{id}/stop. A run ending ends every launch in it.
- GET /session/{id} gains launches.
- GET /machine/jobs: session.start gains launchId (the first launch);
new kinds launch.start and launch.stop. Agents skip unknown kinds, so
old hosts behave as before.

Tests: apps/api 203 pass (new test/launch.test.ts), packages/core 256
pass, against the local test Postgres.
This commit is contained in:
Wanjohi
2026-10-17 13:16:50 +03:00
committed by GitHub
parent 5d49066189
commit 7209e910d9
17 changed files with 9502 additions and 54 deletions
@@ -0,0 +1,29 @@
CREATE TYPE "public"."launch_state" AS ENUM('requested', 'running', 'ended', 'failed');--> statement-breakpoint
CREATE TABLE "launch" (
"id" char(30) PRIMARY KEY NOT NULL,
"time_created" timestamp with time zone DEFAULT now() NOT NULL,
"time_updated" timestamp with time zone DEFAULT now() NOT NULL,
"time_deleted" timestamp with time zone,
"session_id" char(30) NOT NULL,
"game_id" char(30) NOT NULL,
"state" "launch_state" DEFAULT 'requested' NOT NULL,
"first" boolean DEFAULT false NOT NULL,
"time_started" timestamp with time zone,
"time_stopped" timestamp with time zone,
"time_stop_requested" timestamp with time zone,
"error_message" text,
"end_reason" text
);
--> statement-breakpoint
ALTER TABLE "launch" ADD CONSTRAINT "launch_session_id_session_id_fk" FOREIGN KEY ("session_id") REFERENCES "public"."session"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "launch" ADD CONSTRAINT "launch_game_id_game_id_fk" FOREIGN KEY ("game_id") REFERENCES "public"."game"("id") ON DELETE restrict ON UPDATE no action;--> statement-breakpoint
CREATE INDEX "launch_session_idx" ON "launch" USING btree ("session_id");--> statement-breakpoint
CREATE INDEX "launch_state_idx" ON "launch" USING btree ("state");--> statement-breakpoint
-- Every run that has not stopped gets its first launch, so the rule "the last
-- launch over stops the run" holds for runs that began before launches did.
INSERT INTO "launch" ("id", "session_id", "game_id", "state", "first", "time_started")
SELECT 'lch_' || substr(md5(random()::text || s."id"), 1, 26), s."id", s."game_id",
(CASE WHEN s."state" = 'live' THEN 'running' ELSE 'requested' END)::"launch_state",
true, s."time_started"
FROM "session" s
WHERE s."time_stopped" IS NULL AND s."time_deleted" IS NULL;
@@ -0,0 +1 @@
ALTER TABLE "machine" ADD COLUMN "capabilities" text[];
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
@@ -190,6 +190,20 @@
"when": 1792095717042,
"tag": "0026_session_end_reason",
"breakpoints": true
},
{
"idx": 27,
"version": "7",
"when": 1792231242669,
"tag": "0027_launch_under_session",
"breakpoints": true
},
{
"idx": 28,
"version": "7",
"when": 1792231939946,
"tag": "0028_machine_capabilities",
"breakpoints": true
}
]
}
+2 -1
View File
@@ -47,7 +47,8 @@ export const ErrorCodes = {
INVALID_PARAMETER: 'invalid_parameter',
INVALID_FORMAT: 'invalid_format',
INVALID_STATE: 'invalid_state',
IN_USE: 'resource_in_use'
IN_USE: 'resource_in_use',
HOST_TOO_OLD: 'host_too_old'
},
Authentication: {
+6
View File
@@ -184,6 +184,12 @@ export namespace Examples {
endReason: null
};
export const Launch = {
id: Id('launch'),
gameId: Id('game'),
state: 'running' as const
};
export const GameDownload = {
id: Id('gameDownload'),
hostId: Id('machine'),
+1
View File
@@ -18,6 +18,7 @@ export namespace Identifier {
steamLinkRequest: 'slr',
box: 'box',
session: 'ses',
launch: 'lch',
accessToken: 'pat',
game: 'gam',
userLibrary: 'ulb',
+8 -3
View File
@@ -54,7 +54,10 @@ export namespace Machine {
pciId: z
.string()
.regex(/^[0-9a-f]{4}:[0-9a-f]{4}$/)
.meta({ description: 'vendor:device, the ids the model was looked up by', example: '1002:7590' })
.meta({
description: 'vendor:device, the ids the model was looked up by',
example: '1002:7590'
})
})
.meta({ ref: 'Gpu' });
@@ -388,7 +391,8 @@ export namespace Machine {
export const touchLastSeen = fn(
Info.pick({ id: true }).extend({
endpointId: EndpointId.optional(),
slots: z.number().int().min(0).optional()
slots: z.number().int().min(0).optional(),
capabilities: z.array(z.string()).optional()
}),
async (input) => {
try {
@@ -402,7 +406,8 @@ export namespace Machine {
// clearing the column would deregister a working host
// from every route that reads it.
...(input.endpointId ? { endpointId: input.endpointId } : {}),
...(input.slots !== undefined ? { slots: input.slots } : {})
...(input.slots !== undefined ? { slots: input.slots } : {}),
...(input.capabilities !== undefined ? { capabilities: input.capabilities } : {})
})
.where(eq(MachineTable.id, input.id))
.returning({ lastSeen: MachineTable.lastSeen })
+5 -1
View File
@@ -105,7 +105,11 @@ export const MachineTable = pgTable(
// How many runs this host will hold at once, as its last heartbeat said.
// Null for an agent that does not say, which is handed every waiting
// run as before; a number is what makes the rest wait their turn.
slots: integer('slots')
slots: integer('slots'),
// What this host's agent says it can do, as its last heartbeat said.
// Null for an agent that does not say, which is treated as able to do
// nothing new. Gated on this rather than on a version. ref(d-0085)
capabilities: text('capabilities').array()
},
(t) => [
uniqueIndex('machine_secret_hash_unique').on(t.secretHash),
+280 -42
View File
@@ -10,8 +10,11 @@ import { ErrorCodes, VisibleError } from '../error.js';
import { Examples } from '../examples.js';
import { fn } from '../fn.js';
import { GameTable } from '../game/game.sql.js';
import { Identifier } from '../id.js';
import { MachineTable } from '../machine/machine.sql.js';
import { Trial } from '../trial/index.js';
import { Launch } from './launch.js';
import { LaunchTable } from './launch.sql.js';
import { SessionState, SessionTable } from './session.sql.js';
/**
@@ -149,6 +152,10 @@ export namespace Session {
'Why this run is over, once it is. Null while it runs, and for a run whose host did not say',
example: Examples.Session.endReason
}),
launches: Launch.Info.array().optional().meta({
description:
'What runs in the box, oldest first: the first is the launch the run was asked for with. The run is stopped when the last of them is over'
}),
queue: Queue.nullable().optional().meta({
description:
'Where this run is in line for its host, while every slot there is taken. Null once it is not waiting'
@@ -167,8 +174,11 @@ export namespace Session {
trial: z.boolean().optional()
}),
async (input) => {
return Database.use(async (tx) => {
return tx
// The run and its first launch are one write: a run with nothing to
// launch in it would never be stopped by its last launch ending.
// ref(d-0085)
return Database.transaction(async (tx) => {
const session = await tx
.insert(SessionTable)
.values({
id: input.id,
@@ -180,6 +190,13 @@ export namespace Session {
})
.returning()
.then((rows) => serialize(rows[0]!));
const first = await Launch.insert({
id: Identifier.ascending('launch'),
sessionId: session.id,
gameId: session.gameId,
first: true
});
return { ...session, launches: [first] };
});
}
);
@@ -309,7 +326,7 @@ export namespace Session {
async (input) => {
const now = sql`now()`;
return Database.use(async (tx) => {
return tx
const row = await tx
.update(SessionTable)
.set({
state: input.state,
@@ -336,10 +353,23 @@ export namespace Session {
const row = rows.at(0);
return row ? serialize(row) : null;
});
if (row) await launchesFollow(row);
return row;
});
}
);
/**
* What a run's state means for the launches in it: the first is running
* once the box is live, and none outlives the run. ref(d-0085)
*/
async function launchesFollow(run: Info) {
if (run.state === 'live') await Launch.firstStarted(run.id);
if (run.state === 'ended' || run.state === 'failed') {
await Launch.endAllOf([run.id], run.endReason ?? null);
}
}
/**
* The states an agent is allowed to move a run into.
*
@@ -425,9 +455,13 @@ export namespace Session {
example: Examples.Session.linkedAccountId
}),
userId: z.string().meta({
description:
'Whose run it is: the box owner, whose Steam sign-in on this host plays it',
description: 'Whose run it is: the box owner, whose Steam sign-in on this host plays it',
example: Examples.User.id
}),
launchId: z.string().nullable().meta({
description:
'The run’s first launch: the game above. Report its exit against it, and give it to the launched process as `NESTRI_LAUNCH_ID`. Null only for a run older than launches',
example: Examples.Launch.id
})
})
.meta({
@@ -465,7 +499,12 @@ export namespace Session {
description: 'One run to end now, as handed to the agent running it'
});
export const Job = z.discriminatedUnion('kind', [StartJob, StopJob]);
export const Job = z.discriminatedUnion('kind', [
StartJob,
StopJob,
Launch.StartJob,
Launch.StopJob
]);
export type Job = z.infer<typeof Job>;
@@ -506,39 +545,54 @@ export namespace Session {
await leaveAbandoned(machineId);
const free = await freeSlots(machineId);
const starts = await Database.use(async (tx) => {
return tx
.select({ session: SessionTable, box: BoxTable, game: GameTable })
.from(SessionTable)
.innerJoin(BoxTable, eq(SessionTable.boxId, BoxTable.id))
.innerJoin(GameTable, eq(SessionTable.gameId, GameTable.id))
.where(
and(
eq(BoxTable.machineId, machineId),
eq(SessionTable.state, 'requested'),
isNull(SessionTable.timeDeleted),
isNull(BoxTable.timeDeleted)
return (
tx
.select({
session: SessionTable,
box: BoxTable,
game: GameTable,
launchId: LaunchTable.id
})
.from(SessionTable)
.innerJoin(BoxTable, eq(SessionTable.boxId, BoxTable.id))
.innerJoin(GameTable, eq(SessionTable.gameId, GameTable.id))
.leftJoin(
LaunchTable,
and(eq(LaunchTable.sessionId, SessionTable.id), eq(LaunchTable.first, true))
)
)
// Paid ahead of free, then first come first served. See {@link Queue}.
.orderBy(SessionTable.trial, SessionTable.timeCreated)
.limit(free)
.then((rows) =>
rows.map(
(row): Job => ({
kind: 'session.start',
sessionId: row.session.id,
boxId: row.box.id,
// The run's own size when it was given one.
boxTier: (row.session.tier ?? row.box.tier) as z.infer<typeof StartJob>['boxTier'],
gameId: row.game.id,
steamAppId: row.game.steamAppId,
linkedAccountId: row.session.linkedAccountId,
userId: row.box.userId
})
.where(
and(
eq(BoxTable.machineId, machineId),
eq(SessionTable.state, 'requested'),
isNull(SessionTable.timeDeleted),
isNull(BoxTable.timeDeleted)
)
)
);
// Paid ahead of free, then first come first served. See {@link Queue}.
.orderBy(SessionTable.trial, SessionTable.timeCreated)
.limit(free)
.then((rows) =>
rows.map(
(row): Job => ({
kind: 'session.start',
sessionId: row.session.id,
boxId: row.box.id,
// The run's own size when it was given one.
boxTier: (row.session.tier ?? row.box.tier) as z.infer<typeof StartJob>['boxTier'],
gameId: row.game.id,
steamAppId: row.game.steamAppId,
linkedAccountId: row.session.linkedAccountId,
userId: row.box.userId,
launchId: row.launchId
})
)
)
);
});
return [...stops, ...starts];
// Launch work after run work: ending a run ends its launches, and a
// launch is only added to a box that is already up. ref(d-0085)
const launches = await Launch.jobsForMachine(machineId);
return [...stops, ...launches.stops, ...starts, ...launches.starts];
});
/** The runs a host is holding: claimed and not yet over. */
@@ -584,7 +638,7 @@ export namespace Session {
* is ended by its host, never from here.
*/
export const leaveAbandoned = fn(z.string(), async (machineId) => {
await Database.use((tx) =>
const left = await Database.use((tx) =>
tx
.update(SessionTable)
.set({
@@ -604,6 +658,11 @@ export namespace Session {
sql`coalesce(${SessionTable.timeSeen}, ${SessionTable.timeCreated}) < now() - make_interval(secs => ${ABANDONED_AFTER_SECONDS})`
)
)
.returning({ id: SessionTable.id })
);
await Launch.endAllOf(
left.map((r) => r.id),
'left_queue'
);
});
@@ -747,14 +806,30 @@ export namespace Session {
*
* Null when the run is not there or not this person's.
*/
export const requestStop = fn(z.object({ id: Info.shape.id, userId: z.string() }), async (input) => {
export const requestStop = fn(
z.object({ id: Info.shape.id, userId: z.string() }),
async (input) => {
return stopRun(input.id, input.userId);
}
);
/**
* The one way a run is asked to end, whoever asks: its owner, or the last
* of its launches being over. Scoped to an owner when one is given.
*/
async function stopRun(id: string, userId: string | null) {
return Database.transaction(async (tx) => {
const mine = tx
.select({ id: BoxTable.id })
.from(BoxTable)
.where(and(eq(BoxTable.userId, input.userId), isNull(BoxTable.timeDeleted)));
.where(
and(
userId === null ? undefined : eq(BoxTable.userId, userId),
isNull(BoxTable.timeDeleted)
)
);
const scope = and(
eq(SessionTable.id, input.id),
eq(SessionTable.id, id),
inArray(SessionTable.boxId, mine),
isNull(SessionTable.timeDeleted)
);
@@ -770,7 +845,10 @@ export namespace Session {
})
.where(and(scope, eq(SessionTable.state, 'requested')))
.returning();
if (cancelled) return serialize(cancelled);
if (cancelled) {
await Launch.endAllOf([cancelled.id], 'user');
return serialize(cancelled);
}
const [marked] = await tx
.update(SessionTable)
.set({ timeStopRequested: sql`coalesce(${SessionTable.timeStopRequested}, now())` })
@@ -780,7 +858,7 @@ export namespace Session {
const [over] = await tx.select().from(SessionTable).where(scope);
return over ? serialize(over) : null;
});
});
}
/** The boxes one host is responsible for, as a subquery to scope a write. */
function boxesOn(tx: Parameters<Parameters<typeof Database.use>[0]>[0], machineId: string) {
@@ -1010,6 +1088,7 @@ export namespace Session {
if (box) {
await Box.setState({ id: moved.boxId, ...box });
}
await launchesFollow(moved);
// Burn follows the run's state, in the same transaction that
// moved it. A session that went live without its meter starting
@@ -1140,6 +1219,165 @@ export namespace Session {
}
);
/**
* The states a run takes another launch in: claimed by its host and not
* being stopped. A run still waiting for a host has its first launch and
* nothing more yet. ref(d-0085)
*/
const TAKES_LAUNCHES = ['starting', 'live'] as const;
export type AddLaunchResult =
| { outcome: 'notFound' }
| { outcome: 'notRunning'; session: Info }
| { outcome: 'hostTooOld' }
| { outcome: 'created'; launch: Launch.Info };
/**
* Add a launch to a run that is going. The library and account checks are
* the caller's, as they are for the run itself; billing is not asked again,
* because a launch is not a run and costs nothing of its own.
*/
export const addLaunch = fn(
z.object({ sessionId: z.string(), userId: z.string(), gameId: z.string() }),
async (input): Promise<AddLaunchResult> => {
return Database.transaction(async (tx) => {
// Locked, so a stop or an end racing this either lands first and
// the launch is refused, or lands after and ends it with the run.
const [row] = await tx
.select({ session: SessionTable })
.from(SessionTable)
.innerJoin(BoxTable, eq(SessionTable.boxId, BoxTable.id))
.where(
and(
eq(SessionTable.id, input.sessionId),
eq(BoxTable.userId, input.userId),
isNull(SessionTable.timeDeleted),
isNull(BoxTable.timeDeleted)
)
)
.for('update', { of: SessionTable });
if (!row) return { outcome: 'notFound' };
const run = row.session;
if (
!TAKES_LAUNCHES.includes(run.state as (typeof TAKES_LAUNCHES)[number]) ||
run.timeStopped !== null ||
run.timeStopRequested !== null
) {
return { outcome: 'notRunning', session: serialize(run) };
}
if (!(await Launch.hostTakesLaunches(run.id))) return { outcome: 'hostTooOld' };
const launch = await Launch.insert({
id: Identifier.ascending('launch'),
sessionId: run.id,
gameId: input.gameId,
first: false
});
return { outcome: 'created', launch };
});
}
);
/**
* Stop the run when nothing is left running in it: the last window closed
* is the person saying they are done. The same stop its owner asks for.
* ref(d-0085)
*/
async function stopIfEmpty(sessionId: string) {
// One transaction holding the run's row, the same lock `addLaunch`
// takes: a launch added while this decides either lands first and is
// seen as open, or waits and finds the run stopping.
return Database.transaction(async (tx) => {
await tx
.select({ id: SessionTable.id })
.from(SessionTable)
.where(eq(SessionTable.id, sessionId))
.for('update');
if (await Launch.anyOpen(sessionId)) return false;
await stopRun(sessionId, null);
return true;
});
}
/**
* The owner closes one launch. A launch in a run no host holds yet ends at
* once; otherwise it is marked, and the host is handed `launch.stop` until
* it reports the launch over. Null when the run or launch is not theirs.
*/
export const stopLaunch = fn(
z.object({ sessionId: z.string(), launchId: z.string(), userId: z.string() }),
async (input): Promise<Launch.Info | null | 'hostTooOld'> => {
const run = await forOwner({ id: input.sessionId, userId: input.userId });
if (!run) return null;
const row = await Launch.row(run.id, input.launchId);
if (!row) return null;
if (!Launch.OPEN.includes(row.state as (typeof Launch.OPEN)[number])) {
return Launch.serialize(row);
}
if (run.state === 'requested') {
const ended = await Launch.compareAndSet({
id: row.id,
from: row.state as Launch.Info['state'],
to: 'ended',
endReason: 'user'
});
await stopIfEmpty(run.id);
return ended ?? Launch.serialize((await Launch.row(run.id, row.id))!);
}
// A host that does not handle launches would never end this one.
if (!(await Launch.hostTakesLaunches(run.id))) return 'hostTooOld';
return (
(await Launch.markStop(row.id)) ?? Launch.serialize((await Launch.row(run.id, row.id))!)
);
}
);
export interface LaunchReportResult {
outcome: Exclude<TransitionOutcome, never>;
launch: Launch.Info | null;
}
/**
* The host reports where one launch has got to. Held to the run's claim, as
* every host report on a run is. When the report ends the last launch, the
* run is stopped. ref(d-0085)
*/
export const reportLaunch = fn(
z.object({
sessionId: z.string(),
launchId: z.string(),
machineId: z.string(),
claimToken: ClaimToken,
state: Launch.ReportableState,
endReason: EndReason.nullable().optional(),
errorMessage: z.string().nullable().optional()
}),
async (input): Promise<LaunchReportResult> => {
const run = await rowForMachine(input.sessionId, input.machineId);
if (!run) return { outcome: 'forbidden', launch: null };
const row = await Launch.row(run.id, input.launchId);
if (!row) return { outcome: 'forbidden', launch: null };
const current = Launch.serialize(row);
const same = current.state === input.state;
if (!same && !Launch.NEXT_STATES[current.state].includes(input.state)) {
return { outcome: 'illegal', launch: current };
}
if (run.claimToken === null || run.claimToken !== input.claimToken) {
return { outcome: 'notHolder', launch: current };
}
if (same) return { outcome: 'unchanged', launch: current };
const moved = await Launch.compareAndSet({
id: row.id,
from: current.state,
to: input.state,
endReason: input.endReason ?? null,
errorMessage: input.errorMessage ?? null
});
if (!moved) return { outcome: 'lost', launch: current };
if (moved.state === 'ended' || moved.state === 'failed') await stopIfEmpty(run.id);
return { outcome: 'moved', launch: moved };
}
);
export function serialize(input: typeof SessionTable.$inferSelect): z.infer<typeof Info> {
return {
id: input.id,
+51
View File
@@ -0,0 +1,51 @@
import { boolean, index, pgEnum, pgTable, text } from 'drizzle-orm/pg-core';
import { id, timestamps, ulid, utc } from '../db/types.js';
import { GameTable } from '../game/game.sql.js';
import { SessionTable } from './session.sql.js';
/**
* Where one launch inside a run is.
*
* `requested` is the work order, as it is for a run: the host is handed it and
* reports `running` once the game is started in the box. `ended` and `failed`
* are terminal. A launch never outlives its run: when the run stops, every
* launch in it that is not over is ended with it. ref(d-0085)
*/
export const LaunchState = pgEnum('launch_state', ['requested', 'running', 'ended', 'failed']);
/**
* One game or app started inside a running box.
*
* The run is still what is billed and counted; a launch costs nothing of its
* own. A run starts with one launch, its first, which the host starts as part
* of starting the box; any later one is handed to the host as a job of its own
* while the box runs. When the last launch of a run is over, the run is
* stopped. ref(d-0085)
*/
export const LaunchTable = pgTable(
'launch',
{
...id,
...timestamps,
sessionId: ulid('session_id')
.notNull()
.references(() => SessionTable.id, { onDelete: 'cascade' }),
gameId: ulid('game_id')
.notNull()
.references(() => GameTable.id, { onDelete: 'restrict' }),
state: LaunchState('state').notNull().default('requested'),
/**
* The launch the run was asked for with. The host starts it with the box
* (`session.start`), so it is never handed out as a `launch.start`.
*/
first: boolean('first').notNull().default(false),
timeStarted: utc('time_started'),
timeStopped: utc('time_stopped'),
/** When the owner asked for this launch to end, or null if nobody has. */
timeStopRequested: utc('time_stop_requested'),
errorMessage: text('error_message'),
endReason: text('end_reason')
},
(t) => [index('launch_session_idx').on(t.sessionId), index('launch_state_idx').on(t.state)]
);
+382
View File
@@ -0,0 +1,382 @@
import { and, asc, eq, inArray, isNotNull, isNull, sql } from 'drizzle-orm';
import z from 'zod';
import { BoxTable } from '../box/box.sql.js';
import { Database } from '../db/index.js';
import { Examples } from '../examples.js';
import { fn } from '../fn.js';
import { GameTable } from '../game/game.sql.js';
import { MachineTable } from '../machine/machine.sql.js';
import { LaunchState, LaunchTable } from './launch.sql.js';
import { SessionTable } from './session.sql.js';
/**
* One game or app inside a running box. ref(d-0085)
*
* The run (`Session`) is the box and the billing unit; a launch is what runs in
* it. This module knows launches and nothing about stopping runs, so it can be
* imported by `Session` without a cycle: everything here that can end the last
* launch of a run says so in its result, and `Session` acts on it.
*/
export namespace Launch {
export const Info = z
.object({
id: z
.string()
.meta({ description: 'Unique identifier for this launch', example: Examples.Launch.id }),
gameId: z
.string()
.meta({ description: 'The game this launch runs', example: Examples.Launch.gameId }),
state: z.enum(LaunchState.enumValues).meta({
description:
'`requested`: waiting for the host to start it in the box. `running`: started. `ended` and `failed` are over',
example: Examples.Launch.state
}),
timeStarted: z.string().nullable().meta({
description: 'When the host started it in the box',
example: '2026-07-28T12:00:00.000Z'
}),
timeStopped: z.string().nullable().meta({ description: 'When it ended', example: null }),
timeStopRequested: z.string().nullable().meta({
description:
'When its owner asked for it to end. It stays `running` until its host has stopped it',
example: null
}),
endReason: z.string().nullable().meta({
description: 'Why it is over, in the words of `Session.endReason`, once it is',
example: null
}),
errorMessage: z
.string()
.nullable()
.meta({ description: 'Why it failed, when it did', example: null })
})
.meta({ ref: 'Launch', description: 'One game or app started inside a run of a box' });
export type Info = z.infer<typeof Info>;
export const ReportableState = z.enum(['running', 'ended', 'failed']);
export type ReportableState = z.infer<typeof ReportableState>;
/** Where a launch may go next. `requested` is written once, at creation. */
export const NEXT_STATES: Record<Info['state'], readonly Info['state'][]> = {
requested: ['running', 'ended', 'failed'],
running: ['ended', 'failed'],
ended: [],
failed: []
};
export const OPEN = ['requested', 'running'] as const;
/**
* What a host's heartbeat lists when its agent handles `launch.start` and
* `launch.stop`. A host without it is never handed either, and a launch it
* would have to act on is refused instead. ref(d-0085)
*/
export const CAPABILITY = 'launches';
async function machineTakes(where: ReturnType<typeof eq>) {
return Database.use((tx) =>
tx
.select({ capabilities: MachineTable.capabilities })
.from(MachineTable)
.where(where)
.then((rows) => rows.at(0)?.capabilities?.includes(CAPABILITY) ?? false)
);
}
/** Whether the host a run's box is placed on handles launches. */
export async function hostTakesLaunches(sessionId: string): Promise<boolean> {
const machineId = await Database.use((tx) =>
tx
.select({ machineId: BoxTable.machineId })
.from(SessionTable)
.innerJoin(BoxTable, eq(SessionTable.boxId, BoxTable.id))
.where(eq(SessionTable.id, sessionId))
.then((rows) => rows.at(0)?.machineId ?? null)
);
return machineId ? machineTakes(eq(MachineTable.id, machineId)) : false;
}
export function serialize(row: typeof LaunchTable.$inferSelect): Info {
return {
id: row.id,
gameId: row.gameId,
state: row.state as Info['state'],
timeStarted: row.timeStarted?.toISOString() ?? null,
timeStopped: row.timeStopped?.toISOString() ?? null,
timeStopRequested: row.timeStopRequested?.toISOString() ?? null,
endReason: row.endReason,
errorMessage: row.errorMessage
};
}
/** Every launch of one run, oldest first. */
export const listForSession = fn(z.string(), async (sessionId) =>
Database.use((tx) =>
tx
.select()
.from(LaunchTable)
.where(and(eq(LaunchTable.sessionId, sessionId), isNull(LaunchTable.timeDeleted)))
.orderBy(asc(LaunchTable.timeCreated), asc(LaunchTable.id))
.then((rows) => rows.map(serialize))
)
);
/** Whether a run still has a launch that is not over. */
export async function anyOpen(sessionId: string): Promise<boolean> {
return Database.use((tx) =>
tx
.select({ id: LaunchTable.id })
.from(LaunchTable)
.where(
and(
eq(LaunchTable.sessionId, sessionId),
inArray(LaunchTable.state, [...OPEN]),
isNull(LaunchTable.timeDeleted)
)
)
.limit(1)
.then((rows) => rows.length > 0)
);
}
/**
* End every launch of these runs that is not over. A launch never outlives
* its run, so this goes with every write that stops one.
*/
export async function endAllOf(sessionIds: string[], endReason: string | null) {
if (sessionIds.length === 0) return;
await Database.use((tx) =>
tx
.update(LaunchTable)
.set({
state: 'ended',
timeStopped: sql`coalesce(${LaunchTable.timeStopped}, now())`,
endReason: sql`coalesce(${LaunchTable.endReason}, ${endReason})`
})
.where(
and(inArray(LaunchTable.sessionId, sessionIds), inArray(LaunchTable.state, [...OPEN]))
)
);
}
/** The run's first launch is running once the box is live: the host started it with the box. */
export async function firstStarted(sessionId: string) {
await Database.use((tx) =>
tx
.update(LaunchTable)
.set({ state: 'running', timeStarted: sql`coalesce(${LaunchTable.timeStarted}, now())` })
.where(
and(
eq(LaunchTable.sessionId, sessionId),
eq(LaunchTable.first, true),
eq(LaunchTable.state, 'requested')
)
)
);
}
export async function insert(input: {
id: string;
sessionId: string;
gameId: string;
first: boolean;
}) {
return Database.use((tx) =>
tx
.insert(LaunchTable)
.values(input)
.returning()
.then((rows) => serialize(rows[0]!))
);
}
/** One launch of one run, as the row. */
export async function row(sessionId: string, launchId: string) {
return Database.use((tx) =>
tx
.select()
.from(LaunchTable)
.where(
and(
eq(LaunchTable.id, launchId),
eq(LaunchTable.sessionId, sessionId),
isNull(LaunchTable.timeDeleted)
)
)
.then((rows) => rows.at(0) ?? null)
);
}
/**
* Move a launch from one exact state to another, or do nothing. The state is
* in the `where`, so two writers racing get one winner.
*/
export async function compareAndSet(input: {
id: string;
from: Info['state'];
to: Info['state'];
endReason?: string | null;
errorMessage?: string | null;
}) {
const over = input.to === 'ended' || input.to === 'failed';
return Database.use((tx) =>
tx
.update(LaunchTable)
.set({
state: input.to,
...(input.to === 'running'
? { timeStarted: sql`coalesce(${LaunchTable.timeStarted}, now())` }
: {}),
...(over
? {
timeStopped: sql`coalesce(${LaunchTable.timeStopped}, now())`,
endReason: input.endReason ?? null,
errorMessage: input.to === 'failed' ? (input.errorMessage ?? null) : null
}
: {})
})
.where(and(eq(LaunchTable.id, input.id), eq(LaunchTable.state, input.from)))
.returning()
.then((rows) => (rows.at(0) ? serialize(rows[0]!) : null))
);
}
export async function markStop(id: string) {
return Database.use((tx) =>
tx
.update(LaunchTable)
.set({ timeStopRequested: sql`coalesce(${LaunchTable.timeStopRequested}, now())` })
.where(and(eq(LaunchTable.id, id), inArray(LaunchTable.state, [...OPEN])))
.returning()
.then((rows) => (rows.at(0) ? serialize(rows[0]!) : null))
);
}
/**
* A launch to start in a box that is already running. The run's first
* launch is never one of these: the host starts it with the box.
*/
export const StartJob = z
.object({
kind: z
.literal('launch.start')
.meta({ description: 'What the agent is being asked to do', example: 'launch.start' }),
sessionId: z
.string()
.meta({ description: 'The run the box is in', example: Examples.Session.id }),
launchId: z.string().meta({
description:
'The launch to start. Report against it, and give it to the launched process as `NESTRI_LAUNCH_ID`',
example: Examples.Launch.id
}),
boxId: z
.string()
.meta({ description: 'The running box to start it in', example: Examples.Session.boxId }),
gameId: z
.string()
.meta({ description: 'The game to launch', example: Examples.Session.gameId }),
steamAppId: z.number().int().meta({
description: 'The same game, in the id the store knows it by',
example: Examples.Game.steamAppId
}),
linkedAccountId: z.string().meta({
description: 'Which linked account is playing',
example: Examples.Session.linkedAccountId
}),
userId: z.string().meta({ description: 'Whose run it is', example: Examples.User.id })
})
.meta({
ref: 'LaunchStartJob',
description:
'One launch to add to a running box. Handed out on every poll until the host reports the launch `running`, `ended` or `failed`'
});
/** A launch its owner asked to end, repeated until the host reports it over. */
export const StopJob = z
.object({
kind: z
.literal('launch.stop')
.meta({ description: 'What the agent is being asked to do', example: 'launch.stop' }),
sessionId: z
.string()
.meta({ description: 'The run the box is in', example: Examples.Session.id }),
launchId: z.string().meta({ description: 'The launch to end', example: Examples.Launch.id }),
boxId: z.string().meta({ description: 'The box it runs in', example: Examples.Session.boxId })
})
.meta({
ref: 'LaunchStopJob',
description:
'One launch to end now, because its owner closed it. Handed out on every poll until the host reports it `ended` or `failed`, including for a launch the host never started'
});
type StartJob = z.infer<typeof StartJob>;
type StopJob = z.infer<typeof StopJob>;
export async function jobsForMachine(
machineId: string
): Promise<{ stops: StopJob[]; starts: StartJob[] }> {
// A host that has not said it handles launches would skip these jobs
// and leave them repeating forever.
if (!(await machineTakes(eq(MachineTable.id, machineId)))) return { stops: [], starts: [] };
const onHost = and(
eq(BoxTable.machineId, machineId),
isNull(LaunchTable.timeDeleted),
isNull(SessionTable.timeDeleted),
isNull(BoxTable.timeDeleted)
);
const stops = await Database.use((tx) =>
tx
.select({ sessionId: SessionTable.id, launchId: LaunchTable.id, boxId: BoxTable.id })
.from(LaunchTable)
.innerJoin(SessionTable, eq(LaunchTable.sessionId, SessionTable.id))
.innerJoin(BoxTable, eq(SessionTable.boxId, BoxTable.id))
.where(
and(
onHost,
inArray(SessionTable.state, ['starting', 'live']),
inArray(LaunchTable.state, [...OPEN]),
isNotNull(LaunchTable.timeStopRequested)
)
)
.orderBy(LaunchTable.timeStopRequested)
.then((rows) => rows.map((r): StopJob => ({ kind: 'launch.stop', ...r })))
);
const starts = await Database.use((tx) =>
tx
.select({ launch: LaunchTable, session: SessionTable, box: BoxTable, game: GameTable })
.from(LaunchTable)
.innerJoin(SessionTable, eq(LaunchTable.sessionId, SessionTable.id))
.innerJoin(BoxTable, eq(SessionTable.boxId, BoxTable.id))
.innerJoin(GameTable, eq(LaunchTable.gameId, GameTable.id))
.where(
and(
onHost,
// Only into a box that is up, and not one being stopped.
eq(SessionTable.state, 'live'),
isNull(SessionTable.timeStopRequested),
eq(LaunchTable.state, 'requested'),
eq(LaunchTable.first, false),
isNull(LaunchTable.timeStopRequested)
)
)
.orderBy(LaunchTable.timeCreated)
.then((rows) =>
rows.map(
(r): StartJob => ({
kind: 'launch.start',
sessionId: r.session.id,
launchId: r.launch.id,
boxId: r.box.id,
gameId: r.game.id,
steamAppId: r.game.steamAppId,
linkedAccountId: r.session.linkedAccountId,
userId: r.box.userId
})
)
)
);
return { stops, starts };
}
}