From 3a08e2761408009421f4b130defaf3351488a967 Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Mon, 27 Oct 2025 11:38:05 +0300 Subject: [PATCH] feat: Add Session --- packages/core/src/examples.ts | 39 + packages/core/src/session/index.ts | 205 ++++- packages/core/src/session/json.ts | 118 --- packages/core/src/session/schema.ts | 152 ---- packages/core/src/session/types.ts | 1076 --------------------------- packages/sdk/src/index.ts | 238 +----- 6 files changed, 274 insertions(+), 1554 deletions(-) delete mode 100644 packages/core/src/session/json.ts delete mode 100644 packages/core/src/session/schema.ts delete mode 100644 packages/core/src/session/types.ts diff --git a/packages/core/src/examples.ts b/packages/core/src/examples.ts index 988bf589..6327dbda 100644 --- a/packages/core/src/examples.ts +++ b/packages/core/src/examples.ts @@ -44,6 +44,45 @@ export namespace Examples { export const Session = { id: Id("session"), + description: "A gaming session for streaming gameplay", + retry: { + maxAttempts: 3, + factor: 2, + minTimeoutInMs: 1000, + maxTimeoutInMs: 30000, + randomize: true, + }, + queue: { + name: "gaming-queue", + concurrencyLimit: 5, + }, + machine: "ns3-large" as const, + maxDuration: 3600, + relayUrl: "https://relay.nestri.io", + resolution: "1920x1080" as const, + framerate: 60, + room: "game-room-123", + gpuVendor: "nvidia", + gpuName: "RTX 4090", + gpuIndex: 0, + gpuCardPath: "/dev/dri/renderD128", + videoCodec: "h264", + videoEncoder: "nvenc", + videoRateControl: "cbr" as const, + videoCqp: 26, + videoBitrate: 6000, + videoBitrateMax: 8000, + videoEncoderType: "hardware" as const, + videoBitDepth: 8 as const, + audioCaptureMethod: "pipewire" as const, + audioCodec: "opus", + audioEncoder: "opusenc", + audioRateControl: "cbr" as const, + audioBitrate: 128, + audioBitrateMax: 192, + zeroCopy: false, + verbose: false, + run: "async (payload, { ctx, signal }) => { /* session logic */ }", }; export const App = { diff --git a/packages/core/src/session/index.ts b/packages/core/src/session/index.ts index 535c0c74..a14c397c 100644 --- a/packages/core/src/session/index.ts +++ b/packages/core/src/session/index.ts @@ -1,3 +1,202 @@ -export * from "./json.js"; -export * from "./types.js"; -export * from "./schema.js"; +import { z } from "zod/v4"; +import { Identifier } from "../id.js"; +import { Examples } from "../examples.js"; + +// Basic types we need +export const MachinePresetName = z.enum([ + "ns3-micro", + "ns3-small", + "ns3-medium", + "ns3-large", +]); + +export const ScreenResolutionPresetName = z.enum([ + "1280x720", + "1080x1920", + "1440x2560", + "2160x3840", +]); + +export const VideoCodecName = z.enum(["h264", "h265", "vp8", "vp9", "av1"]); + +export const AudioCodecName = z.enum(["opus", "aac", "mp3", "pcm"]); + +export const RetryOptions = z.object({ + maxAttempts: z.number().int().optional(), + factor: z.number().optional(), + minTimeoutInMs: z.number().int().optional(), + maxTimeoutInMs: z.number().int().optional(), + randomize: z.boolean().optional(), + outOfMemory: z + .object({ + machine: MachinePresetName.optional(), + }) + .optional(), +}); + +export namespace Session { + export const Info = z + .object({ + id: Identifier.schema("session").meta({ + description: "The unique identifier of the session", + example: Examples.Session.id, + }), + description: z.string().optional().meta({ + description: "Description of what the session does", + example: Examples.Session.description, + }), + retry: RetryOptions.optional().meta({ + description: "Retry configuration for failed session attempts", + example: Examples.Session.retry, + }), + queue: z + .object({ + name: z.string().optional(), + concurrencyLimit: z.number().optional(), + }) + .optional() + .meta({ + description: "Queue configuration for concurrent session execution", + example: Examples.Session.queue, + }), + machine: MachinePresetName.optional().meta({ + description: "Machine preset for the session", + example: Examples.Session.machine, + }), + maxDuration: z.number().optional().meta({ + description: "Maximum duration in seconds the session can run", + example: Examples.Session.maxDuration, + }), + relayUrl: z.url().optional().meta({ + description: "Nestri relay URL for streaming and signaling", + example: Examples.Session.relayUrl, + }), + resolution: ScreenResolutionPresetName.optional().meta({ + description: "Display or stream resolution in 'WxH' format", + example: Examples.Session.resolution, + }), + framerate: z.number().optional().meta({ + description: "Display or stream framerate", + example: Examples.Session.framerate, + }), + room: z.string().optional().meta({ + description: "Room name or identifier for multiplayer sessions", + example: Examples.Session.room, + }), + gpuVendor: z.string().optional().meta({ + description: "Preferred GPU vendor", + example: Examples.Session.gpuVendor, + }), + gpuName: z.string().optional().meta({ + description: "Preferred GPU name", + example: Examples.Session.gpuName, + }), + gpuIndex: z.number().optional().meta({ + description: "Specific GPU index to use", + example: Examples.Session.gpuIndex, + }), + gpuCardPath: z.string().optional().meta({ + description: "Force specific GPU card by /dev/dri path", + example: Examples.Session.gpuCardPath, + }), + videoCodec: VideoCodecName.optional().meta({ + description: "Preferred video codec for streaming", + example: Examples.Session.videoCodec, + }), + videoEncoder: z.string().optional().meta({ + description: "Override video encoder implementation", + example: Examples.Session.videoEncoder, + }), + videoRateControl: z + .enum(["cbr", "vbr", "cq", "lossless"]) + .optional() + .meta({ + description: "Video rate control method", + example: Examples.Session.videoRateControl, + }), + videoCqp: z.number().min(1).max(51).optional().meta({ + description: "Constant Quantization Parameter for quality control", + example: Examples.Session.videoCqp, + }), + videoBitrate: z.number().min(1).optional().meta({ + description: "Target video bitrate in kbps", + example: Examples.Session.videoBitrate, + }), + videoBitrateMax: z.number().min(1).optional().meta({ + description: "Maximum video bitrate in kbps", + example: Examples.Session.videoBitrateMax, + }), + videoEncoderType: z.enum(["hardware", "software"]).optional().meta({ + description: "Encoder type: hardware or software", + example: Examples.Session.videoEncoderType, + }), + videoBitDepth: z + .union([z.literal(8), z.literal(10)]) + .optional() + .meta({ + description: "Video bit depth (8 or 10)", + example: Examples.Session.videoBitDepth, + }), + audioCaptureMethod: z + .enum(["pipewire", "pulse", "alsa"]) + .optional() + .meta({ + description: "Audio capture method", + example: Examples.Session.audioCaptureMethod, + }), + audioCodec: AudioCodecName.optional().meta({ + description: "Preferred audio codec for streaming", + example: Examples.Session.audioCodec, + }), + audioEncoder: z.string().optional().meta({ + description: "Override audio encoder binary", + example: Examples.Session.audioEncoder, + }), + audioRateControl: z.enum(["cbr", "vbr"]).optional().meta({ + description: "Audio rate control method", + example: Examples.Session.audioRateControl, + }), + audioBitrate: z.number().min(1).optional().meta({ + description: "Target audio bitrate in kbps", + example: Examples.Session.audioBitrate, + }), + audioBitrateMax: z.number().optional().meta({ + description: "Maximum audio bitrate in kbps", + example: Examples.Session.audioBitrateMax, + }), + zeroCopy: z.boolean().optional().meta({ + description: "Whether to use DMA-BUF for zero-copy video pipelines", + example: Examples.Session.zeroCopy, + }), + verbose: z.boolean().optional().meta({ + description: "Enable verbose output for debugging", + example: Examples.Session.verbose, + }), + }) + .meta({ + ref: "Session", + description: + "Represents a session configuration with streaming and execution options", + example: Examples.Session, + }); + + export type Info = z.infer; + + // Session interface that matches the SDK session object + export interface Session { + /** Optional description of the session */ + description?: string; + + /** Start a session without waiting for the result */ + start: (options?: any, requestOptions?: any) => Promise; + + /** Start a session and wait for the result */ + startAndWait: (options?: any) => any; + } + + // Session configuration type that includes the run function + export type Config = Omit & { + /** The function that executes when the session is started */ + run: (ctx: any) => Promise; + }; +} diff --git a/packages/core/src/session/json.ts b/packages/core/src/session/json.ts deleted file mode 100644 index 2681b742..00000000 --- a/packages/core/src/session/json.ts +++ /dev/null @@ -1,118 +0,0 @@ -import { z } from "zod/v4"; - -const LiteralSchema = z.union([z.string(), z.number(), z.boolean(), z.null()]); -type Literal = z.infer; - -export type DeserializedJson = - | Literal - | { [key: string]: DeserializedJson } - | DeserializedJson[]; - -export const DeserializedJsonSchema: z.ZodType = z.lazy(() => - z.union([ - LiteralSchema, - z.array(DeserializedJsonSchema), - z.record(z.string(), DeserializedJsonSchema), - ]), -); - -const SerializableSchema = z.union([ - z.string(), - z.number(), - z.boolean(), - z.null(), - z.date(), - z.undefined(), - z.symbol(), -]); -type Serializable = z.infer; - -export type SerializableJson = - | Serializable - | { [key: string]: SerializableJson } - | SerializableJson[]; - -export const SerializableJsonSchema: z.ZodType = z.lazy(() => - z.union([ - SerializableSchema, - z.array(SerializableJsonSchema), - z.record(z.string(), SerializableJsonSchema), - ]), -); - -/** - * JSON Schema type definition - compatible with JSON Schema Draft 7 - * Based on the JSONSchema7 type from @types/json-schema but defined inline to avoid import issues - */ -export interface JSONSchema { - $id?: string; - $ref?: string; - $schema?: string; - $comment?: string; - - type?: JSONSchemaType | JSONSchemaType[]; - enum?: any[]; - const?: any; - - // Number/Integer validations - multipleOf?: number; - maximum?: number; - exclusiveMaximum?: number; - minimum?: number; - exclusiveMinimum?: number; - - // String validations - maxLength?: number; - minLength?: number; - pattern?: string; - format?: string; - - // Array validations - items?: JSONSchema | JSONSchema[]; - additionalItems?: JSONSchema | boolean; - maxItems?: number; - minItems?: number; - uniqueItems?: boolean; - contains?: JSONSchema; - - // Object validations - maxProperties?: number; - minProperties?: number; - required?: string[]; - properties?: Record; - patternProperties?: Record; - additionalProperties?: JSONSchema | boolean; - dependencies?: Record; - propertyNames?: JSONSchema; - - // Conditional schemas - if?: JSONSchema; - then?: JSONSchema; - else?: JSONSchema; - - // Boolean logic - allOf?: JSONSchema[]; - anyOf?: JSONSchema[]; - oneOf?: JSONSchema[]; - not?: JSONSchema; - - // Metadata - title?: string; - description?: string; - default?: any; - readOnly?: boolean; - writeOnly?: boolean; - examples?: any[]; - - // Additional properties for extensibility - [key: string]: any; -} - -export type JSONSchemaType = - | "string" - | "number" - | "integer" - | "boolean" - | "object" - | "array" - | "null"; diff --git a/packages/core/src/session/schema.ts b/packages/core/src/session/schema.ts deleted file mode 100644 index 28146ecf..00000000 --- a/packages/core/src/session/schema.ts +++ /dev/null @@ -1,152 +0,0 @@ -export type SchemaZodEsque = { - _input: TInput; - _output: TParsedInput; -}; - -export function isSchemaZodEsque( - schema: Schema, -): schema is SchemaZodEsque { - return ( - typeof schema === "object" && - "_def" in schema && - "parse" in schema && - "parseAsync" in schema && - "safeParse" in schema - ); -} - -export type SchemaValibotEsque = { - schema: { - _types?: { - input: TInput; - output: TParsedInput; - }; - }; -}; - -export function isSchemaValibotEsque( - schema: Schema, -): schema is SchemaValibotEsque { - return typeof schema === "object" && "_types" in schema; -} - -export type SchemaArkTypeEsque = { - inferIn: TInput; - infer: TParsedInput; -}; - -export function isSchemaArkTypeEsque( - schema: Schema, -): schema is SchemaArkTypeEsque { - return ( - typeof schema === "object" && "_inferIn" in schema && "_infer" in schema - ); -} - -export type SchemaMyZodEsque = { - parse: (input: any) => TInput; -}; - -export type SchemaSuperstructEsque = { - create: (input: unknown) => TInput; -}; - -export type SchemaCustomValidatorEsque = ( - input: unknown, -) => Promise | TInput; - -export type SchemaYupEsque = { - validateSync: (input: unknown) => TInput; -}; - -export type SchemaScaleEsque = { - assert(value: unknown): asserts value is TInput; -}; - -export type SchemaWithoutInput = - | SchemaCustomValidatorEsque - | SchemaMyZodEsque - | SchemaScaleEsque - | SchemaSuperstructEsque - | SchemaYupEsque; - -export type SchemaWithInputOutput = - | SchemaZodEsque - | SchemaValibotEsque - | SchemaArkTypeEsque; - -export type Schema = SchemaWithInputOutput | SchemaWithoutInput; - -export type inferSchema = - TSchema extends SchemaWithInputOutput - ? { - in: $TIn; - out: $TOut; - } - : TSchema extends SchemaWithoutInput - ? { - in: $InOut; - out: $InOut; - } - : never; - -export type inferSchemaIn< - TSchema extends Schema | undefined, - TDefault = unknown, -> = TSchema extends Schema ? inferSchema["in"] : TDefault; - -export type inferSchemaOut< - TSchema extends Schema | undefined, - TDefault = unknown, -> = TSchema extends Schema ? inferSchema["out"] : TDefault; - -export type SchemaParseFn = (value: unknown) => Promise | TType; -export type AnySchemaParseFn = SchemaParseFn; - -export function getSchemaParseFn( - procedureParser: Schema, -): SchemaParseFn { - const parser = procedureParser as any; - - if (typeof parser === "function" && typeof parser.assert === "function") { - // ParserArkTypeEsque - arktype schemas shouldn't be called as a function because they return a union type instead of throwing - return parser.assert.bind(parser); - } - - if (typeof parser === "function") { - // ParserValibotEsque (>= v0.31.0) - // ParserCustomValidatorEsque - return parser; - } - - if (typeof parser.parseAsync === "function") { - // ParserZodEsque - return parser.parseAsync.bind(parser); - } - - if (typeof parser.parse === "function") { - // ParserZodEsque - // ParserValibotEsque (< v0.13.0) - return parser.parse.bind(parser); - } - - if (typeof parser.validateSync === "function") { - // ParserYupEsque - return parser.validateSync.bind(parser); - } - - if (typeof parser.create === "function") { - // ParserSuperstructEsque - return parser.create.bind(parser); - } - - if (typeof parser.assert === "function") { - // ParserScaleEsque - return (value) => { - parser.assert(value); - return value as TType; - }; - } - - throw new Error("Could not find a validator fn"); -} diff --git a/packages/core/src/session/types.ts b/packages/core/src/session/types.ts deleted file mode 100644 index fa3add3a..00000000 --- a/packages/core/src/session/types.ts +++ /dev/null @@ -1,1076 +0,0 @@ -import { z } from "zod/v4"; -import type { Prettify, IdempotencyKey } from "@/utils/types.js"; -import type { - Schema, - inferSchemaOut, - inferSchemaIn, -} from "@/session/schema.js"; -import { - DeserializedJsonSchema, - JSONSchema, - SerializableJson, -} from "./json.js"; - -export type InitOutput = Record | void | undefined; -export type SessionSchema = Schema | undefined; - -export const SessionExecutionAttempt = z.object({ - number: z.number(), - startedAt: z.coerce.date(), -}); - -export type SessionExecutionAttempt = z.infer; - -export const SessionRun = z.object({ - id: z.string(), - payload: z.string(), - payloadType: z.string(), - tags: z.array(z.string()), - isTest: z.boolean().default(false), - createdAt: z.coerce.date(), - startedAt: z.coerce.date().default(() => new Date()), - idempotencyKey: z.string().optional(), - maxAttempts: z.number().optional(), - version: z.string().optional(), - metadata: z.record(z.string(), DeserializedJsonSchema).optional(), - maxDuration: z.number().optional(), - /** The priority of the sesison. Wih a value of 10 it will be dequeued before runs that were started 9 seconds before it (assuming they had no priority set). */ - priority: z.number().optional(), - baseCostInCents: z.number().optional(), - - parentSessionRunId: z.string().optional(), - rootSessionRunId: z.string().optional(), - - // These are only used during execution, not in run.ctx - durationMs: z.number().optional(), - costInCents: z.number().optional(), - - region: z.string().optional(), -}); - -export type SessionRun = z.infer; - -export const SessionRunContext = z.object({ - // Passthrough needed for backwards compatibility - attempt: SessionExecutionAttempt.loose(), - run: SessionRun.omit({ - payload: true, - payloadType: true, - metadata: true, - durationMs: true, - costInCents: true, - }), -}); - -export type SessionRunContext = z.infer; - -export type RunFnParams = Prettify<{ - /** Metadata about the session, run and attempt. */ - ctx: Context; - /** If you use the `init` function, this will be whatever you returned. */ - init?: TInitOutput; - /** Abort signal that is aborted when a session exceeds it's maxDuration or if the session is cancelled. Can be used to automatically cancel downstream requests */ - signal: AbortSignal; -}>; - -export type MiddlewareFnParams = Prettify<{ - ctx: Context; - next: () => Promise; - /** Abort signal that is aborted when a session exceeds it's maxDuration or if the session is cancelled. Can be used to automatically cancel downstream requests */ - signal: AbortSignal; -}>; - -export type InitFnParams = Prettify<{ - ctx: Context; - /** Abort signal that is aborted when a session exceeds it's maxDuration or if the session is cancelled. Can be used to automatically cancel downstream requests */ - signal: AbortSignal; -}>; - -export type StartFnParams = Prettify<{ - ctx: Context; - init?: InitOutput; - /** Abort signal that is aborted when a session exceeds it's maxDuration or if the session is cancelled. Can be used to automatically cancel downstream requests */ - signal: AbortSignal; -}>; - -export type CancelFnParams = Prettify<{ - ctx: Context; - /** Abort signal that is aborted when a session exceeds it's maxDuration or if the session is cancelled. Can be used to automatically cancel downstream requests */ - signal: AbortSignal; - runPromise: Promise; - init?: InitOutput; -}>; - -export type Context = SessionRunContext; - -// Default is gpu-small -export const MachinePresetName = z.enum([ - "cpu-micro", - "gpu-micro", - "cpu-small", - "gpu-small", - "cpu-medium", - "gpu-medium", - "cpu-large", - "gpu-large", -]); - -export type MachinePresetName = z.infer; - -export const ScreenResolutionPresetName = z.enum([ - "1280x720", - "1080x1920", - "1440x2560", - "2160x3840", -]); - -export type ScreenResolutionPresetName = z.infer< - typeof ScreenResolutionPresetName ->; - -export const RetryOptions = z.object({ - /** The number of attempts before giving up */ - maxAttempts: z.number().int().optional(), - /** The exponential factor to use when calculating the next retry time. - * - * Each subsequent retry will be calculated as `previousTimeout * factor` - */ - factor: z.number().optional(), - /** The minimum time to wait before retrying */ - minTimeoutInMs: z.number().int().optional(), - /** The maximum time to wait before retrying */ - maxTimeoutInMs: z.number().int().optional(), - /** Randomize the timeout between retries. - * - * This can be useful to prevent the thundering herd problem where all retries happen at the same time. - */ - randomize: z.boolean().optional(), - /** If a session fails with an Out Of Memory (OOM) error and you have this set, it will retry with the machine you specify. - * Note: it will not default to this machinefor new runs, only for failures caused by OOM errors. - * So if you frequently have attempts failing with OOM errors, you should set the default machine to be higher. - */ - outOfMemory: z - .object({ - machine: MachinePresetName.optional(), - }) - .optional(), -}); - -export type RetryOptions = z.infer; - -type CommonSessionOptions< - TIdentifier extends string, - TPayload = void, - TOutput = unknown, - TInitOutput extends InitOutput = any, -> = { - /** An id for your session. This must be unique inside your account and not change between versions. */ - id: TIdentifier; - - description?: string; - - /** The retry settings when an uncaught error is thrown. - * - * If omitted it will use the values in your `nestri.config.ts` file. - * - * @example - * - * ``` - * export const sessionWithRetries = session({ - retry: { - maxAttempts: 10, - factor: 1.8, - minTimeoutInMs: 500, - maxTimeoutInMs: 30_000, - randomize: false, - }, - run: async ({ payload, ctx }) => { - //... - }, - }); - * ``` - * */ - retry?: RetryOptions; - - /** Used to configure what should happen when more than one session is ran at the same time. - * - * @example - * one at a time execution - * - * ```ts - * export const oneAtATime = session({ - id: "one-at-a-time", - queue: { - concurrencyLimit: 1, - }, - run: async ({ payload, ctx }) => { - //... - }, - }); - * ``` - */ - queue?: { - name?: string; - concurrencyLimit?: number; - }; - /** Configure the spec of the machine you want your session to run on. - * - * @example - * - * ```ts - * export const demandingGame = session({ - id: "cyber-punk-game", - machine: "gpu-large", - run: async ({ payload, ctx }) => { - //... - }, - }); - * ``` - */ - machine?: MachinePresetName; - - /** - * The maximum duration in compute-time seconds that a session is allowed to run. If the session exceeds this duration, it will be stopped. - * - * Minimum value is 5 seconds - */ - maxDuration?: number; - - /** This gets called when a session is started. It's where you put the code you want to execute. - * - * @param payload - The payload that is passed to your session when it's started. This must be JSON serializable. - * @param params - Metadata about the run. - */ - run: ( - payload: TPayload, - params: RunFnParams, - ) => Promise; - - /** - * catchError is called when the run function throws an error. It can be used to modify the error or return new retry options. - */ - catchError?: OnCatchErrorHookFunction; - - onResume?: OnResumeHookFunction; - onWait?: OnWaitHookFunction; - onComplete?: OnCompleteHookFunction; - onCancel?: OnCancelHookFunction; - - /** - * middleware allows you to run code "around" the run function. This can be useful for logging, metrics, or other cross-cutting concerns. - * - * When writing middleware, you should always call `next()` to continue the execution of the session: - * - * ```ts - * export const middlewareSession = session({ - * id: "session-with-middleware", - * middleware: async (payload, { ctx, next }) => { - * console.log("Before run"); - * await next(); - * console.log("After run"); - * }, - * run: async (payload, { ctx }) => {} - * }); - * ``` - */ - middleware?: OnMiddlewareHookFunction; - - /** - * onStart is called the first time a session is started in a run (not before every retry) - */ - onStart?: OnStartHookFunction; - - /** - * onSuccess is called after the run function has successfully completed. - */ - onSuccess?: OnSuccessHookFunction; - - /** - * onFailure is called after a session has failed (meaning the run function threw an error and won't be retried anymore) - */ - onFailure?: OnFailureHookFunction; - - /** - * JSON Schema for the session payload. This will be synced to the server during indexing. - * Should be a valid JSON Schema Draft 7 object. - */ - jsonSchema?: JSONSchema; - - /** - * Enable verbose output for debugging and detailed logs. - */ - verbose?: boolean; - - /** - * The Nestri relay URL used to connect the session for streaming and signaling. - */ - relayUrl?: string; - - /** - * Display or stream resolution in 'WxH' format. - * - * @example "1920x1080" - */ - resolution?: ScreenResolutionPresetName; - - /** - * Display or stream framerate. - * - * @default 60 - */ - framerate?: number; - - /** - * Optional room name or identifier for multiplayer or collaborative sessions. - * If this is ommited, a random room id will be generated for you. - */ - room?: string; - - /** - * Preferred GPU vendor for this session (e.g. 'nvidia', 'amd', 'intel'). - */ - gpuVendor?: string; - - /** - * Preferred GPU name (e.g. 'RTX 4090', 'A6000'). - */ - gpuName?: string; - - /** - * Specific GPU index to use (e.g. 0 for the first GPU). - */ - gpuIndex?: number; - - /** - * Force a specific GPU card or render device by its /dev/dri path. - * - * @example "/dev/dri/renderD128" - */ - gpuCardPath?: string; - - /** - * Preferred video codec for streaming. - * - * @default "h264" - */ - videoCodec?: string; - - /** - * Override the video encoder implementation (e.g. 'nvenc', 'vaapi', 'x264'). - */ - videoEncoder?: string; - - /** - * Video rate control method. - * - * @default "cbr" - */ - videoRateControl?: "cbr" | "vbr" | "cq" | "lossless"; - - /** - * Constant Quantization Parameter (CQP) for quality control. - * Range: 1–51 (lower = higher quality). - * - * @default 26 - */ - videoCqp?: number; - - /** - * Target video bitrate in kbps. - * - * @default 6000 - */ - videoBitrate?: number; - - /** - * Maximum video bitrate in kbps. - * - * @default 8000 - */ - videoBitrateMax?: number; - - /** - * Encoder type to use for video: hardware or software. - * - * @default "hardware" - */ - videoEncoderType?: "hardware" | "software"; - - /** - * Video bit depth (8 or 10), only relevant with DMA-BUF and non-H264 codecs. - * - * @default 8 - */ - videoBitDepth?: 8 | 10; - - /** - * Audio capture method (e.g. 'pipewire', 'pulse', 'alsa'). - * - * @default "pipewire" - */ - audioCaptureMethod?: "pipewire" | "pulse" | "alsa"; - - /** - * Preferred audio codec for streaming (e.g. 'opus', 'aac'). - * - * @default "opus" - */ - audioCodec?: string; - - /** - * Override audio encoder binary (e.g. 'opusenc', 'ffmpeg'). - */ - audioEncoder?: string; - - /** - * Audio rate control method. - * - * @default "cbr" - */ - audioRateControl?: "cbr" | "vbr"; - - /** - * Target audio bitrate in kbps. - * - * @default 128 - */ - audioBitrate?: number; - - /** - * Maximum audio bitrate in kbps. - * - * @default 192 - */ - audioBitrateMax?: number; - - /** - * Whether to use DMA-BUF for zero-copy video pipelines. - * - * @default false - */ - dmaBuf?: boolean; -}; - -export type SessionOptions< - TIdentifier extends string, - TPayload = void, - TOutput = unknown, - TInitOutput extends InitOutput = any, -> = CommonSessionOptions; - -// Session options when payloadSchema is provided - payload should be any -export type SessionOptionsWithSchema< - TIdentifier extends string, - TOutput = unknown, - TInitOutput extends InitOutput = any, -> = CommonSessionOptions & { - jsonSchema: JSONSchema; -}; - -export type SessionWithSchemaOptions< - TIdentifier extends string, - TSchema extends SessionSchema | undefined = undefined, - TOutput = unknown, - TInitOutput extends InitOutput = any, -> = CommonSessionOptions< - TIdentifier, - inferSchemaOut, - TOutput, - TInitOutput -> & { - schema?: TSchema; -}; - -export interface SessionWithSchema< - TIdentifier extends string, - TSchema extends SessionSchema | undefined = undefined, - TOutput = any, -> extends Session, TOutput> { - schema?: TSchema; -} - -export type SessionInitOutput = Record | void | undefined; - -export type SessionWait = - | { - type: "duration"; - date: Date; - } - | { - type: "token"; - token: string; - } - | { - type: "session"; - runId: string; - }; - -export type SessionResumeHookParams< - TPayload = unknown, - TInitOutput extends SessionInitOutput = SessionInitOutput, -> = { - ctx: SessionRunContext; - wait: SessionWait; - payload: TPayload; - session: string; - signal: AbortSignal; - init?: TInitOutput; -}; - -export type OnResumeHookFunction< - TPayload, - TInitOutput extends SessionInitOutput = SessionInitOutput, -> = ( - params: SessionResumeHookParams, -) => undefined | void | Promise; - -export type SessionSuccessHookParams< - TPayload = unknown, - TOutput = unknown, - TInitOutput extends SessionInitOutput = SessionInitOutput, -> = { - ctx: SessionRunContext; - payload: TPayload; - session: string; - output: TOutput; - signal: AbortSignal; - init?: TInitOutput; -}; - -export type OnSuccessHookFunction< - TPayload, - TOutput, - TInitOutput extends SessionInitOutput = SessionInitOutput, -> = ( - params: SessionSuccessHookParams, -) => undefined | void | Promise; - -export type SessionCompleteSuccessResult = { - ok: true; - data: TOutput; -}; - -export type SessionCompleteErrorResult = { - ok: false; - error: unknown; -}; - -export type SessionCompleteResult = - | SessionCompleteSuccessResult - | SessionCompleteErrorResult; - -export type SessionCompleteHookParams< - TPayload = unknown, - TOutput = unknown, - TInitOutput extends SessionInitOutput = SessionInitOutput, -> = { - ctx: SessionRunContext; - payload: TPayload; - session: string; - result: SessionCompleteResult; - signal: AbortSignal; - init?: TInitOutput; -}; - -export type OnCompleteHookFunction< - TPayload, - TOutput, - TInitOutput extends SessionInitOutput = SessionInitOutput, -> = ( - params: SessionCompleteHookParams, -) => undefined | void | Promise; - -export type SessionCatchErrorHookParams< - TPayload = unknown, - TInitOutput extends SessionInitOutput = SessionInitOutput, -> = { - ctx: SessionRunContext; - payload: TPayload; - session: string; - error: unknown; - retry?: RetryOptions; - retryAt?: Date; - retryDelayInMs?: number; - signal: AbortSignal; - init?: TInitOutput; -}; - -export type HandleErrorModificationOptions = { - skipRetrying?: boolean | undefined; - retryAt?: Date | undefined; - retryDelayInMs?: number | undefined; - retry?: RetryOptions | undefined; - error?: unknown; -}; - -export type HandleErrorResult = - | undefined - | void - | HandleErrorModificationOptions - | Promise; - -export type OnCatchErrorHookFunction< - TPayload, - TInitOutput extends SessionInitOutput = SessionInitOutput, -> = ( - params: SessionCatchErrorHookParams, -) => HandleErrorResult; - -export type SessionMiddlewareHookParams = { - ctx: SessionRunContext; - payload: TPayload; - session: string; - signal: AbortSignal; - next: () => Promise; -}; - -export type OnMiddlewareHookFunction = ( - params: SessionMiddlewareHookParams, -) => Promise; - -export type SessionCancelHookParams< - TPayload = unknown, - TRunOutput = any, - TInitOutput extends SessionInitOutput = SessionInitOutput, -> = { - ctx: SessionRunContext; - payload: TPayload; - session: string; - runPromise: Promise; - init?: TInitOutput; - signal: AbortSignal; -}; - -export type OnCancelHookFunction< - TPayload, - TRunOutput = any, - TInitOutput extends SessionInitOutput = SessionInitOutput, -> = ( - params: SessionCancelHookParams, -) => undefined | void | Promise; - -export type SessionWaitHookParams< - TPayload = unknown, - TInitOutput extends SessionInitOutput = SessionInitOutput, -> = { - wait: SessionWait; - ctx: SessionRunContext; - payload: TPayload; - session: string; - signal: AbortSignal; - init?: TInitOutput; -}; - -export type OnWaitHookFunction< - TPayload, - TInitOutput extends SessionInitOutput = SessionInitOutput, -> = ( - params: SessionWaitHookParams, -) => undefined | void | Promise; - -export type SessionStartHookParams< - TPayload = unknown, - TInitOutput extends SessionInitOutput = SessionInitOutput, -> = { - ctx: SessionRunContext; - payload: TPayload; - session: string; - signal: AbortSignal; - init?: TInitOutput; -}; - -export type OnStartHookFunction< - TPayload, - TInitOutput extends SessionInitOutput = SessionInitOutput, -> = ( - params: SessionStartHookParams, -) => undefined | void | Promise; - -export type SessionFailureHookParams< - TPayload = unknown, - TInitOutput extends SessionInitOutput = SessionInitOutput, -> = { - ctx: SessionRunContext; - payload: TPayload; - session: string; - error: unknown; - signal: AbortSignal; - init?: TInitOutput; -}; - -export type OnFailureHookFunction< - TPayload, - TInitOutput extends SessionInitOutput = SessionInitOutput, -> = ( - params: SessionFailureHookParams, -) => undefined | void | Promise; - -export interface Session< - TIdentifier extends string, - TInput = void, - TOutput = any, -> { - /** - * The id of the session. - */ - id: TIdentifier; - - description?: string; - - jsonSchema?: JSONSchema; - - /** - * Start a session with the given payload, and continue without waiting for the result. If you want to wait for the result, use `sessionAndWait`. Returns the id of the started session. - * @param payload - * @param options - * @returns RunHandle - * - `id` - The id of the started session run. - */ - start: ( - payload: TInput, - options?: StartOptions, - requestOptions?: any, //TriggerApiRequestOptions, - ) => Promise>; - - /** - * Start a session with the given payload, and wait for the result. Returns the result of the session - * @param payload - * @param options - Options for the session - * @returns SessionRunResult - * @example - * ``` - * const result = await session.startAndWait({ foo: "bar" }); - * - * if (result.ok) { - * console.log(result.output); - * } else { - * console.error(result.error); - * } - * ``` - */ - startAndWait: ( - payload: TInput, - options?: StartAndWaitOptions, - ) => SessionRunPromise; -} - -//an array of 1, 2, or 3 strings -const RunTag = z.string().max(128, "Tags must be less than 128 characters"); -export const RunTags = z.union([RunTag, RunTag.array()]); - -export type RunTags = z.infer; - -export type StartOptions = { - /** - * A unique key that can be used to ensure that a session is only started once per key. - * - * You can use `idempotencyKeys.create` to create an idempotency key first, and then pass it to the session options. - * - * @example - * - * ```typescript - * import { idempotencyKeys, sessions } from "@nestri/sdk"; - * - * export const myGameSession = session({ - * id: "my-game-session", - * run: async (payload: any) => { - * // scoped to the session run by default - * const idempotencyKey = await idempotencyKeys.create("my-game-session-key"); - * - * // Use the idempotency key when starting child sessions - * await session.startAndWait(payload, { idempotencyKey }); - * - * // scoped globally, does not include the session run ID - * const globalIdempotencyKey = await idempotencyKeys.create("my-session-key", { scope: "global" }); - * - * await childSession.startAndWait(payload, { idempotencyKey: globalIdempotencyKey }); - * - * // You can also pass a string directly, which is the same as a global idempotency key - * await childSession.startAndWait(payload, { idempotencyKey: "my-very-unique-key" }); - * } - * }); - * ``` - * - * When starting a session inside another session, we automatically inject the run ID into the key material. - * - * If you are starting a session from your backend, ensure you include some sufficiently unique key material to prevent collisions. - * - * @example - * - * ```typescript - * import { idempotencyKeys, sessions } from "@nestri/sdk"; - * - * // Somewhere in your backend - * const idempotencyKey = await idempotenceKeys.create(["my-session-start", "user-123"]); - * await session.start("my-session", { foo: "bar" }, { idempotencyKey }); - * ``` - * - */ - idempotencyKey?: IdempotencyKey | string | string[]; - - /** - * The time-to-live for the idempotency key. Once the TTL has passed, the key can be used again. - * - * Specify a duration string like "1h", "10s", "30m", etc. - */ - idempotencyKeyTTL?: string; - - /** - * The maximum number of retry attempts for the session if it fails to start. - * If not specified, it will use the nestri or the default retry policy from your nestri.config file. - */ - maxAttempts?: number; - - /** - * You can override the queue for the session. If a queue doesn't exist for the given name, the run will be in the PENDING_VERSION state until the queue is created.. - */ - queue?: string; - - /** - * The `concurrencyKey` creates a copy of the queue for every unique value of the key. - * For example, if the queue (set when starting or on the session) has a concurrency limit of 10, - * and you set the concurrency key to `userId`, then each user will have their own queue with a concurrency limit of 10. - */ - concurrencyKey?: string; - - /** - * The delay before the session is ran. This can be a string like "1h" or a Date object. - * - * @example - * "1h" - 1 hour - * "30d" - 30 days - * "15m" - 15 minutes - * "2w" - 2 weeks - * "60s" - 60 seconds - * new Date("2025-01-01T00:00:00Z") - */ - delay?: string | Date; - - /** - * Set a time-to-live for this run. If the run is not executed within this time, it will be removed from the queue and never execute. - * - * @example - * - * ```ts - * await gameSession.start({ foo: "bar" }, { ttl: "1h" }); - * await gameSession.start({ foo: "bar" }, { ttl: 60 * 60 }); // 1 hour - * ``` - * - * The minimum value is 1 second. Setting the `ttl` to `0` will disable the TTL and the run will never expire. - * - * **Note:** Runs in development have a default `ttl` of 10 minutes. You can override this by setting the `ttl` option. - */ - ttl?: string | number; - - /** - * If started at the same time, a higher priority run will be executed first. - * - The value is a time offset in seconds that determines the order of dequeuing. - * If you start two sessions 9 seconds apart but the second one has `priority: 10`, it will be executed before the first one. - * - * @example - * ```ts - // no priority = 0 - await gameSession.start({ foo: "bar" }); - - //... imagine 9s pass by - - // this run will start before the run above that was started 9s ago (with no priority) - await gameSession.start({ foo: "bar" }, { priority: 10 }); - ``` - * - */ - priority?: number; - - /** - * Tags to attach to the run. Tags can be used to filter sessions in the Nestri Studio and using the SDK. - * - * You can set up to 10 tags per run, they must be less than 128 characters each. - * - * We recommend prefixing tags with a namespace using an underscore or colon, like `user_1234567` or `org:9876543`. - * - * @example - * - * ```ts - * await myGameSession.start({ foo: "bar" }, { tags: ["user:1234567", "org:9876543"] }); - * ``` - */ - tags?: RunTags; - - /** - * Metadata to attach to the run. Metadata can be used to store additional information about the run. Limited to 256KB. - */ - metadata?: Record; - - /** - * The maximum duration in compute-time seconds that a session is allowed to run. If the session exceeds this duration, it will be stopped. - * - * This will override the session's maxDuration. - * - * Minimum value is 5 seconds - */ - maxDuration?: number; - - /** - * The machine preset to use for this run. This will override the session's machine preset and any defaults. - */ - machine?: MachinePresetName; - - /** - * Specify the version of the deployed session to run. By default the "current" version is used at the time of execution, - * but you can specify a specific version to run here. - * - * @example - * - * ```ts - * await mySession.start({ foo: "bar" }, { version: "20250208.1" }); - * ``` - * - * Note that this option is only available for `start` and NOT `startAndWait`. The "wait" versions will always be locked - * to the same version as the parent session that is starting the child sessions. - */ - version?: string; - - /** - * Specify the region to run the session in. This overrides the default region set for your team member in the Nestri Studio. - * - * Check the Regions page in the Studio/Docs for regions that are close to you. - * - * In BYOG this won't do anything, so it's fine to set it in your code. - * - * @example - * - * ```ts - * await mySession.start({ foo: "bar" }, { region: "fi-dc-central" }); - * ``` - */ - region?: string; -}; - -export type StartAndWaitOptions = Omit; - -declare const __output: unique symbol; -declare const __payload: unique symbol; -type BrandRun = { [__output]: O; [__payload]: P }; -export type BrandedRun = T & BrandRun; - -export type RunHandle< - TSessionIdentifier extends string, - TPayload, - TOutput, -> = BrandedRun< - { - id: string; - /** - * An auto-generated JWT that can be used to access the run - */ - publicAccessToken: string; - sessionIdentifier: TSessionIdentifier; - }, - TPayload, - TOutput ->; - -export type SessionRunResult = - | { - ok: true; - id: string; - sessionIdentifier: TIdentifier; - output: TOutput; - } - | { - ok: false; - id: string; - sessionIdentifier: TIdentifier; - error: unknown; - }; - -export class SubsessionUnwrapError extends Error { - public readonly sessionID: string; - public readonly runId: string; - public readonly cause?: unknown; - - constructor(sessionID: string, runId: string, subsessionError: unknown) { - if (subsessionError instanceof Error) { - super(`Error in ${sessionID}: ${subsessionError.message}`); - this.cause = subsessionError; - this.name = "SubsessionUnwrapError"; - } else { - super(`Error in ${sessionID}`); - this.name = "SubsessionUnwrapError"; - this.cause = subsessionError; - } - - this.sessionID = sessionID; - this.runId = runId; - } -} - -export class SessionRunPromise< - TIdentifier extends string, - TOutput, -> extends Promise> { - constructor( - executor: ( - resolve: ( - value: - | SessionRunResult - | PromiseLike>, - ) => void, - reject: (reason?: any) => void, - ) => void, - private readonly sessionID: TIdentifier, - ) { - super(executor); - } - - async unwrap(): Promise { - return this.then((result) => { - if (result.ok) { - return result.output; - } else { - throw new SubsessionUnwrapError( - this.sessionID, - result.id, - result.error, - ); - } - }); - } -} - -export type RunTypes = { - output: TOutput; - payload: TPayload; - sessionIdentifier: TSessionIdentifier; -}; - -export type AnyRunTypes = RunTypes; - -export type InferRunTypes = - T extends RunHandle - ? RunTypes - : T extends Session - ? RunTypes - : AnyRunTypes; - -export type RunHandleFromTypes = RunHandle< - TRunTypes["sessionIdentifier"], - TRunTypes["payload"], - TRunTypes["output"] ->; - -export type BatchedRunHandle< - TSessionIdentifier extends string, - TPayload, - TOutput, -> = BrandedRun< - { - id: string; - sessionIdentifier: TSessionIdentifier; - isCached: boolean; - idempotencyKey?: string; - }, - TPayload, - TOutput ->; diff --git a/packages/sdk/src/index.ts b/packages/sdk/src/index.ts index 388441dd..15f3c4cd 100644 --- a/packages/sdk/src/index.ts +++ b/packages/sdk/src/index.ts @@ -1,217 +1,45 @@ -import { - type Session, - type RunTypes, - type InitOutput, - type StartOptions, - type inferSchemaIn, - type SchemaParseFn, - type SessionSchema, - type SessionOptions, - type SessionWithSchema, - type SessionWithSchemaOptions, - type RunHandleFromTypes, - type AnyRunTypes, - type SessionOptionsWithSchema, - getSchemaParseFn, - SessionRunPromise, -} from "@nestri/core/session/index"; +import { Identifier } from "@nestri/core/id"; +import { Session } from "@nestri/core/session/index"; -// Overload: when payloadSchema is provided, payload type should be any -export function createRun< - TIdentifier extends string, - TOutput = unknown, - TInitOutput extends InitOutput = any, ->( - params: SessionOptionsWithSchema, -): Session; - -// Overload: normal case without payloadSchema -export function createRun< - TIdentifier extends string, - TInput = void, - TOutput = unknown, - TInitOutput extends InitOutput = any, ->( - params: SessionOptions, -): Session; - -export function createRun< - TIdentifier extends string, - TInput = void, - TOutput = unknown, - TInitOutput extends InitOutput = any, ->( - params: - | SessionOptions - | SessionOptionsWithSchema, -): Session | Session { - const session: Session = { - id: params.id, - description: params.description, - jsonSchema: params.jsonSchema, - start: async (payload, options) => { - return await run_internal>( - "run()", - params.id, +// Simple session creation function using our clean Session types +export function createSession(config: Session.Config): Session.Session { + const session: Session.Session = { + description: config.description, + start: async (payload: any) => { + console.log( + `Starting session ${config.description} with payload:`, payload, - undefined, - { - queue: params.queue?.name, - ...options, - }, ); + // TODO: Implement actual session start logic + return { + id: `run_${Date.now()}` as `run_${string}`, + sessionIdentifier: config.description, + publicAccessToken: "mock_token", + }; }, - startAndWait: (payload, options) => { - return new SessionRunPromise((resolve, reject) => { - triggerAndWait_internal( - "triggerAndWait()", - params.id, - payload, - undefined, - { - queue: params.queue?.name, - ...options, - }, - ) - .then((result) => { - resolve(result); - }) - .catch((error) => { - reject(error); + + startAndWait: (payload: any) => { + console.log( + `Starting session ${config.description} and waiting with payload:`, + payload, + ); + // TODO: Implement actual session start and wait logic + return new Promise((resolve) => { + setTimeout(() => { + resolve({ + ok: true, + id: `run_${Date.now()}` as any, + sessionIdentifier: config.description, }); - }, params.id); + }, 1000); + }); }, }; - registerTaskLifecycleHooks(params.id, params); - - resourceCatalog.registerTaskMetadata({ - id: params.id, - description: params.description, - queue: params.queue, - retry: params.retry - ? { ...defaultRetryOptions, ...params.retry } - : undefined, - machine: - typeof params.machine === "string" - ? { preset: params.machine } - : params.machine, - maxDuration: params.maxDuration, - payloadSchema: params.jsonSchema, - fns: { - run: params.run, - }, - }); - - const queue = params.queue; - - if (queue && typeof queue.name === "string") { - resourceCatalog.registerQueueMetadata({ - name: queue.name, - concurrencyLimit: queue.concurrencyLimit, - }); - } - - // @ts-expect-error - session[Symbol.for("nestri/session")] = true; + // Store the run function for later execution + (session as any)._runFunction = config.run; return session; } -export function createSchemaTask< - TIdentifier extends string, - TSchema extends SessionSchema | undefined = undefined, - TOutput = unknown, - TInitOutput extends InitOutput = any, ->( - params: SessionWithSchemaOptions, -): SessionWithSchema { - const parsePayload = params.schema - ? getSchemaParseFn>(params.schema) - : undefined; - - const session: SessionWithSchema = { - id: params.id, - description: params.description, - schema: params.schema, - start: async (payload, options, requestOptions) => { - return await run_internal< - RunTypes, TOutput> - >( - "run()", - params.id, - payload, - parsePayload, - { - queue: params.queue?.name, - ...options, - }, - requestOptions, - ); - }, - startAndWait: (payload, options) => { - return new SessionRunPromise((resolve, reject) => { - triggerAndWait_internal, TOutput>( - "triggerAndWait()", - params.id, - payload, - parsePayload, - { - queue: params.queue?.name, - ...options, - }, - ) - .then((result) => { - resolve(result); - }) - .catch((error) => { - reject(error); - }); - }, params.id); - }, - }; - - registerTaskLifecycleHooks(params.id, params); - - resourceCatalog.registerTaskMetadata({ - id: params.id, - description: params.description, - queue: params.queue, - retry: params.retry - ? { ...defaultRetryOptions, ...params.retry } - : undefined, - machine: { preset: params.machine }, - maxDuration: params.maxDuration, - fns: { - run: params.run, - parsePayload, - }, - schema: params.schema, - }); - - const queue = params.queue; - - if (queue && typeof queue.name === "string") { - resourceCatalog.registerQueueMetadata({ - name: queue.name, - concurrencyLimit: queue.concurrencyLimit, - }); - } - - // @ts-expect-error - session[Symbol.for("nestri/session")] = true; - - return session; -} - -export const session = createRun; -export const schemaSession = createSchemaTask; - -async function run_internal( - name: string, - id: TRunTypes["sessionIdentifier"], - payload: TRunTypes["payload"], - parsePayload?: SchemaParseFn, - options?: StartOptions, - requestOptions?: any, //TriggerApiRequestOptions, -): Promise> {} +export const session = createSession;