import {
type EnvironmentId,
type ServerConfig,
type ServerConfigStreamEvent,
type ServerLifecycleWelcomePayload,
type ServerLifecycleStreamReadyEvent,
type ServerSelfUpdateProgressEvent,
type ServerSelfUpdateResult,
WS_METHODS,
} from "@t3tools/contracts";
import * as Cause from "effect/Cause";
import * as Duration from "effect/Duration";
import * as Deferred from "effect/Deferred";
import * as Effect from "effect/Effect";
import * as Exit from "effect/Exit";
import * as Fiber from "effect/Fiber";
import * as Option from "effect/Option";
import * as Queue from "effect/Queue";
import * as Ref from "effect/Ref";
import * as Result from "effect/Result";
import * as Schema from "effect/Schema";
import * as Stream from "effect/Stream";
import * as SubscriptionRef from "effect/SubscriptionRef";
import { AsyncResult, Atom } from "effect/unstable/reactivity";
import {
createAtomCommandScheduler,
createEnvironmentRpcCommand,
createEnvironmentRpcQueryAtomFamily,
createEnvironmentRpcSubscriptionAtomFamily,
createRuntimeCommand,
scheduleAtomCommandEffect,
} from "./runtime.ts";
import { EnvironmentRegistry } from "../connection/registry.ts";
import { EnvironmentSupervisor } from "../connection/supervisor.ts";
import { safeErrorLogAttributes } from "../errors/safeLog.ts";
import { EnvironmentCacheStore } from "../platform/persistence.ts";
import {
isRpcClientError,
request,
runStream,
subscribe,
type EnvironmentRpcInput,
} from "../rpc/client.ts";
import { followStreamInEnvironment } from "./runtime.ts";
import {
applyServerConfigProjection,
type ServerConfigProjection,
withoutEnvironmentThemes,
} from "./serverConfigProjection.ts";
// Exported server state includes this type in its inferred public return type.
export type { ServerConfigProjection } from "./serverConfigProjection.ts";
export type ServerUpdateStage = "downloading" | "installing" | "resuming";
export type ServerUpdateState =
| { readonly status: "idle" }
| {
readonly status: "running";
readonly stage: ServerUpdateStage;
readonly fromVersion: string;
readonly targetVersion: string;
}
| {
readonly status: "failed";
readonly stage: ServerUpdateStage;
readonly fromVersion: string;
readonly targetVersion: string;
readonly message: string;
};
export interface ServerUpdateTarget {
readonly environmentId: EnvironmentId;
readonly input: EnvironmentRpcInput<typeof WS_METHODS.serverUpdateServer>;
}
const IDLE_SERVER_UPDATE_STATE: ServerUpdateState = { status: "idle" };
const EMPTY_SERVER_UPDATE_STATE_ATOM = Atom.make<ServerUpdateState>(IDLE_SERVER_UPDATE_STATE).pipe(
Atom.withLabel("environment-data:server:update-state:empty"),
);
const serverUpdateStateAtom = Atom.family((environmentId: EnvironmentId) =>
Atom.make<ServerUpdateState>(IDLE_SERVER_UPDATE_STATE).pipe(
Atom.withLabel(`environment-data:server:update-state:${environmentId}`),
),
);
export class ServerUpdateResumeTimeoutError extends Schema.TaggedErrorClass<ServerUpdateResumeTimeoutError>()(
"ServerUpdateResumeTimeoutError",
{
environmentId: Schema.String,
targetVersion: Schema.String,
},
) {
override get message(): string {
return `The server did not resume on t3@${this.targetVersion}.`;
}
}
export class ServerUpdateProgressIncompleteError extends Schema.TaggedErrorClass<ServerUpdateProgressIncompleteError>()(
"ServerUpdateProgressIncompleteError",
{
targetVersion: Schema.String,
},
) {
override get message(): string {
return `The t3@${this.targetVersion} update ended before the server accepted the restart.`;
}
}
export class ServerUpdateTerminalError extends Schema.TaggedErrorClass<ServerUpdateTerminalError>()(
"ServerUpdateTerminalError",
{
targetVersion: Schema.String,
status: Schema.Literals(["committed", "rolled-back", "failed"]),
reason: Schema.optional(Schema.String),
},
) {
override get message(): string {
return this.reason ?? `The t3@${this.targetVersion} update ${this.status}.`;
}
}
// Covers the 120-second trial deadline and a final restart of the previous
// version when the trial rolls back.
const SERVER_UPDATE_RESUME_TIMEOUT = Duration.minutes(4);
export function matchesServerUpdateReadyEvent(
result: ServerSelfUpdateResult,
event: ServerLifecycleStreamReadyEvent,
): boolean {
return result.updateId === undefined
? event.payload.environment.serverVersion === result.targetVersion
: event.payload.updateOutcome?.id === result.updateId;
}
export function matchesServerUpdateResumeEvent(
result: ServerSelfUpdateResult,
event: ServerLifecycleStreamReadyEvent,
): boolean {
return (
(result.method === "desktop-app" && result.desktopUpdateToken !== undefined) ||
matchesServerUpdateReadyEvent(result, event)
);
}
export function validateServerUpdateReadyEvent(
result: ServerSelfUpdateResult,
event: ServerLifecycleStreamReadyEvent,
): Effect.Effect<void, ServerUpdateTerminalError> {
if (result.updateId === undefined) return Effect.void;
const outcome = event.payload.updateOutcome;
if (
outcome?.id === result.updateId &&
outcome.status === "committed" &&
outcome.targetVersion === result.targetVersion &&
event.payload.environment.serverVersion === result.targetVersion
) {
return Effect.void;
}
return Effect.fail(
new ServerUpdateTerminalError({
targetVersion: result.targetVersion,
status: outcome?.status ?? "failed",
reason:
outcome?.reason ??
"The service launcher resumed without committing the requested server version.",
}),
);
}
/**
* Keeps reconnect attempts ~1s apart for the whole update restart.
*
* A restart takes the server down for ~15 seconds, but the supervisor's normal
* backoff ladder (1/2/4/8/16s) assumes an unexpected failure and lands attempts
* at ~3, 5, 9, 17 and 33 seconds — so a 15-second restart is observed as a
* 33-second "Resuming". Nudging on every backoff entry (not just the first)
* holds the retry cadence flat until the server answers again. The sleep before
* each nudge is the pacer: a connection that fails instantly re-enters backoff
* immediately and would otherwise spin a tight retry loop.
*
* A newly restarted server can also reject the first environment credential.
* Authentication blocks need the same paced retry during this known restart;
* permission and configuration failures remain blocked.
*
* Callers fork this as a child of the update command so it is interrupted as
* soon as the update settles, whether it succeeds, fails, or times out.
*/
export function nudgeReconnectDuringUpdateRestart(input: {
readonly stateChanges: Stream.Stream<
{
readonly phase: string;
readonly lastFailure?: { readonly reason: string } | null;
},
unknown
>;
readonly retryNow: Effect.Effect<void>;
readonly interval?: Duration.Duration;
}): Effect.Effect<void> {
return input.stateChanges.pipe(
Stream.filter(
(state) =>
state.phase === "backoff" ||
(state.phase === "blocked" && state.lastFailure?.reason === "authentication"),
),
Stream.runForEach(() =>
Effect.sleep(input.interval ?? Duration.seconds(1)).pipe(Effect.andThen(input.retryNow)),
),
Effect.timeoutOption(SERVER_UPDATE_RESUME_TIMEOUT),
Effect.ignore,
);
}
export function waitForNextEnvironmentReconnect<E>(
stateChanges: Stream.Stream<{ readonly phase: string }, E>,
): Effect.Effect<void, E> {
return stateChanges.pipe(
Stream.dropWhile((state) => state.phase === "connected"),
Stream.filter((state) => state.phase === "connected"),
Stream.runHead,
Effect.asVoid,
);
}
export const runDesktopCommitWithReconnectObserver = Effect.fn(
"runDesktopCommitWithReconnectObserver",
)(function* <EState, ECommit>(
stateChanges: Stream.Stream<{ readonly phase: string }, EState>,
commit: Effect.Effect<void, ECommit>,
) {
const armed = yield* Deferred.make<void>();
const reconnected = yield* Deferred.make<void>();
const observer = yield* stateChanges.pipe(
Stream.tap(() => Deferred.succeed(armed, undefined)),
waitForNextEnvironmentReconnect,
Effect.andThen(Deferred.succeed(reconnected, undefined)),
Effect.forkChild,
);
yield* Deferred.await(armed);
const commitExit = yield* commit.pipe(Effect.exit);
if (Exit.isSuccess(commitExit)) {
yield* Fiber.interrupt(observer);
return;
}
if (!isLegacyUpdateHandoffLoss(commitExit.cause)) {
yield* Fiber.interrupt(observer);
return yield* Effect.failCause(commitExit.cause);
}
yield* Deferred.await(reconnected).pipe(Effect.timeout(SERVER_UPDATE_RESUME_TIMEOUT));
return yield* Effect.failCause(commitExit.cause);
});
export const waitForDesktopUpdateTarget = Effect.fn("waitForDesktopUpdateTarget")(function* <
EReady,
ECommit,
>(
targetVersion: string,
nextReady: Effect.Effect<ServerLifecycleStreamReadyEvent, EReady>,
retryCommit: Effect.Effect<void, ECommit>,
maxCommitAttempts = 3,
): Effect.fn.Return<ServerLifecycleStreamReadyEvent, EReady | ECommit | ServerUpdateTerminalError> {
for (let attempt = 1; attempt <= maxCommitAttempts; attempt += 1) {
const ready = yield* nextReady;
if (ready.payload.environment.serverVersion === targetVersion) return ready;
if (attempt === maxCommitAttempts) break;
const retryExit = yield* retryCommit.pipe(Effect.exit);
if (Exit.isSuccess(retryExit)) break;
if (!isLegacyUpdateHandoffLoss(retryExit.cause)) {
return yield* Effect.failCause(retryExit.cause);
}
}
return yield* new ServerUpdateTerminalError({
targetVersion,
status: "failed",
reason: "The desktop app resumed without installing the prepared update.",
});
});
export function serverUpdateStateForProgressEvent(
fromVersion: string,
targetVersion: string,
event: ServerSelfUpdateProgressEvent,
): Extract<ServerUpdateState, { status: "running" }> {
return {
status: "running",
stage: event.type === "complete" ? "resuming" : event.stage,
fromVersion,
targetVersion,
};
}
export function serverUpdateStateForServerVersion(
state: ServerUpdateState,
serverVersion: string | null,
): ServerUpdateState {
return state.status === "idle" ||
state.status === "running" ||
serverVersion === null ||
state.fromVersion === serverVersion
? state
: IDLE_SERVER_UPDATE_STATE;
}
function serverUpdateFailureMessage(error: unknown): string {
return error instanceof Error ? error.message : "Server update failed.";
}
function isRpcSocketError(error: unknown): boolean {
if (!isRpcClientError(error)) {
return false;
}
switch (error.reason._tag) {
case "SocketReadError":
case "SocketWriteError":
case "SocketCloseError":
return true;
default:
return false;
}
}
export function isLegacyUpdateHandoffLoss(cause: Cause.Cause<unknown>): boolean {
if (Cause.hasInterruptsOnly(cause)) {
return true;
}
return (
cause.reasons.length > 0 &&
cause.reasons.every((reason) => Cause.isFailReason(reason) && isRpcSocketError(reason.error))
);
}
export function resolveServerUpdateProgressResult<E>(
targetVersion: string,
terminal: Option.Option<ServerSelfUpdateResult>,
streamExit: Exit.Exit<void, E>,
): Effect.Effect<ServerSelfUpdateResult, E | ServerUpdateProgressIncompleteError> {
if (
Option.isSome(terminal) &&
(Exit.isSuccess(streamExit) || isLegacyUpdateHandoffLoss(streamExit.cause))
) {
return Effect.succeed(terminal.value);
}
if (Exit.isFailure(streamExit)) {
return Effect.failCause(streamExit.cause);
}
return Effect.fail(new ServerUpdateProgressIncompleteError({ targetVersion }));
}
const cachedConfigSnapshotEvent = (config: ServerConfig): ServerConfigStreamEvent => ({
version: 1,
type: "snapshot",
config,
});
export interface ServerConfigSubscriptionOptions {
readonly environmentThemes?: boolean;
readonly usageLimitSources?: boolean;
}
export const makeEnvironmentServerConfigState = Effect.fn("EnvironmentServerConfigState.make")(
function* (subscription: ServerConfigSubscriptionOptions) {
const supervisor = yield* EnvironmentSupervisor;
const cache = yield* EnvironmentCacheStore;
const environmentId = supervisor.target.environmentId;
const cachedConfig = yield* cache.loadServerConfig(environmentId).pipe(
Effect.catch((error) =>
Effect.logWarning("Could not load cached server configuration.").pipe(
Effect.annotateLogs({
environmentId,
...safeErrorLogAttributes(error),
}),
Effect.as(Option.none<ServerConfig>()),
),
),
);
const state = yield* SubscriptionRef.make<Option.Option<ServerConfigProjection>>(
// Stripped on load as well as on save: a cache written by an earlier
// build can still carry published themes.
Option.map(cachedConfig, (cached) => ({
config: withoutEnvironmentThemes(cached),
latestEvent: cachedConfigSnapshotEvent(withoutEnvironmentThemes(cached)),
source: "cache" as const,
})),
);
const persistence = yield* Queue.sliding<ServerConfig>(1);
const pendingPersistence = yield* Ref.make<Option.Option<ServerConfig>>(Option.none());
const persist = Effect.fn("EnvironmentServerConfigState.persist")(function* (
config: ServerConfig,
) {
return yield* cache.saveServerConfig(environmentId, withoutEnvironmentThemes(config)).pipe(
Effect.as(true),
Effect.catch((error) =>
Effect.logWarning("Could not persist cached server configuration.").pipe(
Effect.annotateLogs({
environmentId,
...safeErrorLogAttributes(error),
}),
Effect.as(false),
),
),
);
});
const persistPending = Effect.fn("EnvironmentServerConfigState.persistPending")(function* (
config: ServerConfig,
) {
if (!(yield* persist(config))) {
return;
}
yield* Ref.update(pendingPersistence, (pending) =>
Option.isSome(pending) && pending.value === config ? Option.none() : pending,
);
});
yield* Stream.fromQueue(persistence).pipe(
Stream.debounce("500 millis"),
Stream.runForEach(persistPending),
Effect.forkScoped,
);
yield* subscribe(WS_METHODS.subscribeServerConfig, {
...(subscription.environmentThemes === true ? { environmentThemes: true } : {}),
...(subscription.usageLimitSources === true ? { usageLimitSources: true } : {}),
}).pipe(
Stream.runForEach((event) =>
Effect.gen(function* () {
const next = applyServerConfigProjection(yield* SubscriptionRef.get(state), event);
if (Option.isNone(next)) {
return;
}
yield* Ref.set(pendingPersistence, Option.some(next.value.config));
yield* SubscriptionRef.set(state, next);
yield* Queue.offer(persistence, next.value.config);
}),
),
Effect.forkScoped,
);
yield* Effect.addFinalizer(() =>
Ref.get(pendingPersistence).pipe(
Effect.flatMap(
Option.match({
onNone: () => Effect.void,
onSome: (config) => persist(config).pipe(Effect.asVoid),
}),
),
),
);
return state;
},
);
export function serverConfigStateChanges(
environmentId: EnvironmentId,
subscription: ServerConfigSubscriptionOptions,
) {
return followStreamInEnvironment(
environmentId,
Stream.unwrap(
makeEnvironmentServerConfigState(subscription).pipe(
Effect.map((state) =>
SubscriptionRef.changes(state).pipe(
Stream.filterMap((projection) =>
Option.match(projection, {
onNone: () => Result.failVoid,
onSome: (value) => Result.succeed(value),
}),
),
),
),
),
),
);
}
export function projectServerWelcome(
current: Option.Option<ServerLifecycleWelcomePayload>,
event: {
readonly type: "welcome" | "ready";
readonly payload: unknown;
},
): readonly [
Option.Option<ServerLifecycleWelcomePayload>,
ReadonlyArray<ServerLifecycleWelcomePayload>,
] {
if (event.type !== "welcome") {
return [current, []];
}
const welcome = event.payload as ServerLifecycleWelcomePayload;
return [Option.some(welcome), [welcome]];
}
export function resolveServerConfigValue(
projection: ServerConfigProjection | null,
initialConfig: ServerConfig | null,
): ServerConfig | null {
if (
projection?.source === "live" &&
(initialConfig === null ||
projection.config.environment.serverVersion === initialConfig.environment.serverVersion)
) {
return projection.config;
}
return initialConfig ?? projection?.config ?? null;
}
export function createServerEnvironmentAtoms<R, E>(
runtime: Atom.AtomRuntime<EnvironmentRegistry | EnvironmentCacheStore | R, E>,
options: {
readonly initialConfigValueAtom: (
environmentId: EnvironmentId,
) => Atom.Atom<ServerConfig | null>;
/**
* Whether this surface renders themes the environment publishes. Mobile
* keeps its own appearance settings, so it neither asks for the stream nor
* receives the payload.
*/
readonly environmentThemes?: boolean;
/** Whether this surface renders quota from configured usage-limit sources. */
readonly usageLimitSources?: boolean;
},
) {
const configScheduler = createAtomCommandScheduler();
// Updates stay serial end-to-end, but only their handoff phase occupies the config lane.
const updateScheduler = createAtomCommandScheduler();
const configConcurrency = {
mode: "serial" as const,
key: ({ environmentId }: { readonly environmentId: string }) => environmentId,
};
const configProjectionFamily = Atom.family((environmentId: EnvironmentId) =>
runtime
.atom(
serverConfigStateChanges(environmentId, {
...(options.environmentThemes === true ? { environmentThemes: true } : {}),
...(options.usageLimitSources === true ? { usageLimitSources: true } : {}),
}),
)
.pipe(
Atom.setIdleTTL(5 * 60_000),
Atom.withLabel(`environment-data:server:config-projection:${environmentId}`),
),
);
const configProjection = (target: {
readonly environmentId: EnvironmentId;
readonly input: EnvironmentRpcInput<typeof WS_METHODS.subscribeServerConfig>;
}) => configProjectionFamily(target.environmentId);
const emptyConfigAtom = Atom.make<ServerConfig | null>(null).pipe(
Atom.withLabel("environment-data:server:config:empty"),
);
const configValueAtom = Atom.family((environmentId: EnvironmentId | null) => {
if (environmentId === null) {
return emptyConfigAtom;
}
return Atom.make((get): ServerConfig | null => {
const projection = Option.getOrNull(
AsyncResult.value(get(configProjection({ environmentId, input: {} }))),
);
return resolveServerConfigValue(
projection,
get(options.initialConfigValueAtom(environmentId)),
);
}).pipe(Atom.withLabel(`environment-data:server:config:${environmentId}`));
});
const updateStateValueAtom = Atom.family((environmentId: EnvironmentId) =>
Atom.make((get) =>
serverUpdateStateForServerVersion(
get(serverUpdateStateAtom(environmentId)),
get(configValueAtom(environmentId))?.environment.serverVersion ?? null,
),
).pipe(Atom.withLabel(`environment-data:server:update-state-value:${environmentId}`)),
);
const updateStateAtom = (environmentId: EnvironmentId | null) =>
environmentId === null ? EMPTY_SERVER_UPDATE_STATE_ATOM : updateStateValueAtom(environmentId);
const updateServer = createRuntimeCommand<
EnvironmentRegistry | EnvironmentCacheStore | R,
E,
ServerUpdateTarget,
ServerSelfUpdateResult,
unknown
>(runtime, {
label: "environment-data:server:update-server",
scheduler: updateScheduler,
concurrency: configConcurrency,
execute: (target, atomRegistry) => {
const stateAtom = serverUpdateStateAtom(target.environmentId);
let targetVersion = target.input.targetVersion;
let fromVersion =
atomRegistry.get(configValueAtom(target.environmentId))?.environment.serverVersion ??
targetVersion;
let currentStage: ServerUpdateStage = "downloading";
let desktopCommitLostTransport = false;
atomRegistry.set(stateAtom, {
status: "running",
stage: currentStage,
fromVersion,
targetVersion,
});
return Effect.gen(function* () {
const environmentRegistry = yield* EnvironmentRegistry;
const desktopCommitStarting = yield* Deferred.make<void>();
const desktopReconnectObserverArmed = yield* Deferred.make<void>();
const desktopReconnected = yield* Deferred.make<void>();
yield* Deferred.await(desktopCommitStarting).pipe(
Effect.andThen(
environmentRegistry.stateChanges(target.environmentId).pipe(
Stream.tap(() => Deferred.succeed(desktopReconnectObserverArmed, undefined)),
Stream.dropWhile((state) => state.phase === "connected"),
Stream.filter((state) => state.phase === "connected"),
Stream.runHead,
),
),
Effect.andThen(Deferred.succeed(desktopReconnected, undefined)),
Effect.forkChild,
);
const result = yield* scheduleAtomCommandEffect(
atomRegistry,
configScheduler,
configConcurrency,
target,
Effect.gen(function* () {
const currentConfig = atomRegistry.get(configValueAtom(target.environmentId));
fromVersion = currentConfig?.environment.serverVersion ?? targetVersion;
atomRegistry.set(stateAtom, {
status: "running",
stage: currentStage,
fromVersion,
targetVersion,
});
const supportsProgress =
currentConfig?.environment.capabilities.serverSelfUpdateProgress === true;
const updateResult: ServerSelfUpdateResult = supportsProgress
? yield* Effect.gen(function* () {
const terminal = yield* Ref.make<Option.Option<ServerSelfUpdateResult>>(
Option.none(),
);
const streamExit = yield* environmentRegistry
.runStream(
target.environmentId,
runStream(WS_METHODS.serverUpdateServerWithProgress, target.input),
)
.pipe(
Stream.runForEach((event) =>
Effect.sync(() => {
currentStage = event.type === "complete" ? "resuming" : event.stage;
atomRegistry.set(
stateAtom,
serverUpdateStateForProgressEvent(fromVersion, targetVersion, event),
);
}).pipe(
Effect.andThen(
event.type === "complete"
? Ref.set(terminal, Option.some(event.result))
: Effect.void,
),
),
),
Effect.exit,
);
return yield* resolveServerUpdateProgressResult(
targetVersion,
yield* Ref.get(terminal),
streamExit,
);
})
: yield* Effect.gen(function* () {
const selfUpdateMethod = currentConfig?.environment.capabilities.serverSelfUpdate;
const exit = yield* environmentRegistry
.run(target.environmentId, request(WS_METHODS.serverUpdateServer, target.input))
.pipe(Effect.exit);
if (Exit.isSuccess(exit)) {
return exit.value;
}
if (
(selfUpdateMethod === "boot-service" || selfUpdateMethod === "respawn") &&
isLegacyUpdateHandoffLoss(exit.cause)
) {
// Older servers can tear down the transport before their
// unary acknowledgement arrives. Treat only that transport
// loss as a handoff, then prove it by waiting for target ready.
return { targetVersion, method: selfUpdateMethod };
}
return yield* Effect.failCause(exit.cause);
});
if (
updateResult.method === "desktop-app" &&
updateResult.desktopUpdateToken !== undefined
) {
yield* Deferred.succeed(desktopCommitStarting, undefined);
yield* Deferred.await(desktopReconnectObserverArmed);
const commitExit = yield* environmentRegistry
.run(
target.environmentId,
request(WS_METHODS.serverCommitDesktopUpdate, {
requestId: updateResult.desktopUpdateToken,
}),
)
.pipe(Effect.exit);
if (Exit.isFailure(commitExit) && !isLegacyUpdateHandoffLoss(commitExit.cause)) {
return yield* Effect.failCause(commitExit.cause);
}
desktopCommitLostTransport = Exit.isFailure(commitExit);
}
targetVersion = updateResult.targetVersion;
currentStage = "resuming";
atomRegistry.set(stateAtom, {
status: "running",
stage: currentStage,
fromVersion,
targetVersion,
});
return updateResult;
}),
);
// The update restart is intentional and the server stays unreachable
// for the whole restart, so hold the retry cadence flat instead of
// letting the supervisor climb its backoff ladder.
yield* nudgeReconnectDuringUpdateRestart({
stateChanges: environmentRegistry.stateChanges(target.environmentId),
retryNow: environmentRegistry.retryNow(target.environmentId),
}).pipe(Effect.forkChild);
if (result.method === "desktop-app" && desktopCommitLostTransport) {
yield* Deferred.await(desktopReconnected).pipe(
Effect.timeout(SERVER_UPDATE_RESUME_TIMEOUT),
);
}
const waitForReady = environmentRegistry
.followStream(target.environmentId, subscribe(WS_METHODS.subscribeServerLifecycle, {}))
.pipe(
Stream.filter(
(event): event is ServerLifecycleStreamReadyEvent =>
event.type === "ready" && matchesServerUpdateResumeEvent(result, event),
),
Stream.runHead,
Effect.timeoutOption(SERVER_UPDATE_RESUME_TIMEOUT),
Effect.map(Option.flatten),
);
const nextReady = waitForReady.pipe(
Effect.flatMap(
Option.match({
onNone: () =>
new ServerUpdateResumeTimeoutError({
environmentId: target.environmentId,
targetVersion,
}),
onSome: Effect.succeed,
}),
),
);
const desktopUpdateToken = result.desktopUpdateToken;
const resumed =
result.method === "desktop-app" && desktopUpdateToken !== undefined
? yield* waitForDesktopUpdateTarget(
result.targetVersion,
nextReady,
runDesktopCommitWithReconnectObserver(
environmentRegistry.stateChanges(target.environmentId),
environmentRegistry
.run(
target.environmentId,
request(WS_METHODS.serverCommitDesktopUpdate, {
requestId: desktopUpdateToken,
}),
)
.pipe(Effect.asVoid),
),
)
: yield* nextReady;
yield* validateServerUpdateReadyEvent(result, resumed);
atomRegistry.set(stateAtom, IDLE_SERVER_UPDATE_STATE);
return result;
}).pipe(
Effect.onExit((exit) =>
Effect.sync(() => {
if (Exit.isSuccess(exit)) {
return;
}
if (Cause.hasInterruptsOnly(exit.cause)) {
atomRegistry.set(stateAtom, IDLE_SERVER_UPDATE_STATE);
return;
}
atomRegistry.set(stateAtom, {
status: "failed",
stage: currentStage,
fromVersion,
targetVersion,
message: serverUpdateFailureMessage(Cause.squash(exit.cause)),
});
}),
),
);
},
});
const settingsValueAtom = Atom.family((environmentId: EnvironmentId) =>
Atom.make((get) => get(configValueAtom(environmentId))?.settings ?? null).pipe(
Atom.withLabel(`environment-data:server:settings:${environmentId}`),
),
);
const providersValueAtom = Atom.family((environmentId: EnvironmentId) =>
Atom.make((get) => get(configValueAtom(environmentId))?.providers ?? null).pipe(
Atom.withLabel(`environment-data:server:providers:${environmentId}`),
),
);
return {
configValueAtom,
updateStateAtom,
settingsValueAtom,
providersValueAtom,
providerAuthState: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
label: "environment-data:provider:auth-state",
tag: WS_METHODS.providerAuthSubscribe,
idleTtlMs: 0,
}),
startProviderAuth: createEnvironmentRpcCommand(runtime, {
label: "environment-data:provider:auth-start",
tag: WS_METHODS.providerAuthStart,
concurrency: {
mode: "singleFlight",
key: ({ environmentId, input }) => JSON.stringify([environmentId, input.instanceId]),
},
}),
completeProviderAuth: createEnvironmentRpcCommand(runtime, {
label: "environment-data:provider:auth-complete",
tag: WS_METHODS.providerAuthComplete,
}),
cancelProviderAuth: createEnvironmentRpcCommand(runtime, {
label: "environment-data:provider:auth-cancel",
tag: WS_METHODS.providerAuthCancel,
}),
logoutProviderAuth: createEnvironmentRpcCommand(runtime, {
label: "environment-data:provider:auth-logout",
tag: WS_METHODS.providerAuthLogout,
}),
providerInstallState: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
label: "environment-data:provider:install-state",
tag: WS_METHODS.providerInstallSubscribe,
idleTtlMs: 0,
}),
startProviderInstall: createEnvironmentRpcCommand(runtime, {
label: "environment-data:provider:install-start",
tag: WS_METHODS.providerInstallStart,
concurrency: {
mode: "singleFlight",
key: ({ environmentId }) => environmentId,
},
}),
cancelProviderInstall: createEnvironmentRpcCommand(runtime, {
label: "environment-data:provider:install-cancel",
tag: WS_METHODS.providerInstallCancel,
}),
removeProviderInstallation: createEnvironmentRpcCommand(runtime, {
label: "environment-data:provider:install-remove",
tag: WS_METHODS.providerInstallRemove,
}),
traceDiagnostics: createEnvironmentRpcQueryAtomFamily(runtime, {
label: "environment-data:server:trace-diagnostics",
tag: WS_METHODS.serverGetTraceDiagnostics,
}),
processDiagnostics: createEnvironmentRpcQueryAtomFamily(runtime, {
label: "environment-data:server:process-diagnostics",
tag: WS_METHODS.serverGetProcessDiagnostics,
}),
processResourceHistory: createEnvironmentRpcQueryAtomFamily(runtime, {
label: "environment-data:server:process-resource-history",
tag: WS_METHODS.serverGetProcessResourceHistory,
}),
resourceTelemetry: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
label: "environment-data:server:resource-telemetry",
tag: WS_METHODS.subscribeResourceTelemetry,
idleTtlMs: 0,
}),
resourceTelemetryHistory: createEnvironmentRpcQueryAtomFamily(runtime, {
label: "environment-data:server:resource-telemetry-history",
tag: WS_METHODS.serverGetResourceTelemetryHistory,
staleTimeMs: 5_000,
}),
// A cold transcript scan is measured in seconds, so keep the result around
// long enough that switching windows or re-rendering does not rescan.
usageSummary: createEnvironmentRpcQueryAtomFamily(runtime, {
label: "environment-data:server:usage-summary",
tag: WS_METHODS.serverGetUsageSummary,
staleTimeMs: 60_000,
}),
configProjection,
welcome: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
label: "environment-data:server:welcome",
tag: WS_METHODS.subscribeServerLifecycle,
transform: (stream) =>
stream.pipe(
Stream.mapAccum(Option.none<ServerLifecycleWelcomePayload>, projectServerWelcome),
),
}),
refreshProviders: createEnvironmentRpcCommand(runtime, {
label: "environment-data:server:refresh-providers",
tag: WS_METHODS.serverRefreshProviders,
concurrency: {
mode: "singleFlight",
key: ({ environmentId, input }) =>
JSON.stringify([
environmentId,
input.instanceId ?? null,
input.cwd ?? null,
input.refreshModels ?? false,
]),
},
}),
updateProvider: createEnvironmentRpcCommand(runtime, {
label: "environment-data:server:update-provider",
tag: WS_METHODS.serverUpdateProvider,
scheduler: configScheduler,
concurrency: configConcurrency,
}),
updateServer,
upsertKeybinding: createEnvironmentRpcCommand(runtime, {
label: "environment-data:server:upsert-keybinding",
tag: WS_METHODS.serverUpsertKeybinding,
scheduler: configScheduler,
concurrency: configConcurrency,
}),
removeKeybinding: createEnvironmentRpcCommand(runtime, {
label: "environment-data:server:remove-keybinding",
tag: WS_METHODS.serverRemoveKeybinding,
scheduler: configScheduler,
concurrency: configConcurrency,
}),
updateSettings: createEnvironmentRpcCommand(runtime, {
label: "environment-data:server:update-settings",
tag: WS_METHODS.serverUpdateSettings,
scheduler: configScheduler,
concurrency: configConcurrency,
}),
signalProcess: createEnvironmentRpcCommand(runtime, {
label: "environment-data:server:signal-process",
tag: WS_METHODS.serverSignalProcess,
}),
refreshUsageRates: createEnvironmentRpcCommand(runtime, {
label: "environment-data:server:refresh-usage-rates",
tag: WS_METHODS.serverRefreshUsageRates,
concurrency: {
mode: "singleFlight",
key: ({ environmentId }) => environmentId,
},
}),
retryResourceTelemetry: createEnvironmentRpcCommand(runtime, {
label: "environment-data:server:retry-resource-telemetry",
tag: WS_METHODS.serverRetryResourceTelemetry,
concurrency: {
mode: "singleFlight",
key: ({ environmentId }) => environmentId,
},
}),
};
}
