import { EnvironmentId, ORCHESTRATION_WS_METHODS, WS_METHODS } from "@t3tools/contracts";import * as Cause from "effect/Cause";import * as Context from "effect/Context";import type * as Duration from "effect/Duration";import * as Effect from "effect/Effect";import * as Option from "effect/Option";import * as Schema from "effect/Schema";import * as Stream from "effect/Stream";import * as SubscriptionRef from "effect/SubscriptionRef";import { RpcClientError } from "effect/unstable/rpc"; import { EnvironmentSupervisor } from "../connection/supervisor.ts";import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; export class EnvironmentRpcUnavailableError extends Schema.TaggedErrorClass<EnvironmentRpcUnavailableError>()( "EnvironmentRpcUnavailableError", { environmentId: EnvironmentId, environmentLabel: Schema.String, method: Schema.optionalKey(Schema.String), },) { override get message(): string { const method = this.method === undefined ? "" : ` for RPC method ${this.method}`; return `${this.environmentLabel} is not connected${method}.`; }} export const isEnvironmentRpcUnavailableError = Schema.is(EnvironmentRpcUnavailableError); export interface EnvironmentRpcRequestObservation { readonly environmentId: string; readonly method: string;} export class EnvironmentRpcRequestObserver extends Context.Reference<{ readonly observe: ( request: EnvironmentRpcRequestObservation, ) => Effect.Effect<Effect.Effect<void>>;}>("@t3tools/client-runtime/rpc/EnvironmentRpcRequestObserver", { defaultValue: () => ({ observe: () => Effect.succeed(Effect.void), }),}) {} export type EnvironmentRpcTag = keyof WsRpcProtocolClient & string;type RpcMethod<TTag extends EnvironmentRpcTag> = WsRpcProtocolClient[TTag]; export type EnvironmentSubscriptionRpcTag = | typeof ORCHESTRATION_WS_METHODS.subscribeShell | typeof ORCHESTRATION_WS_METHODS.subscribeThread | typeof WS_METHODS.subscribeAuthAccess | typeof WS_METHODS.subscribeServerConfig | typeof WS_METHODS.subscribeServerLifecycle | typeof WS_METHODS.subscribeTerminalEvents | typeof WS_METHODS.subscribeTerminalMetadata | typeof WS_METHODS.subscribePreviewEvents | typeof WS_METHODS.subscribeDiscoveredLocalServers | typeof WS_METHODS.previewAutomationConnect | typeof WS_METHODS.subscribeVcsStatus | typeof WS_METHODS.terminalAttach; export type EnvironmentStreamCommandRpcTag = | typeof WS_METHODS.cloudInstallRelayClient | typeof WS_METHODS.gitRunStackedAction; export type EnvironmentStreamRpcTag = | EnvironmentSubscriptionRpcTag | EnvironmentStreamCommandRpcTag; export type EnvironmentUnaryRpcTag = Exclude<EnvironmentRpcTag, EnvironmentStreamRpcTag>;const isRpcClientError = Schema.is(RpcClientError.RpcClientError); export type EnvironmentRpcInput<TTag extends EnvironmentRpcTag> = Parameters<RpcMethod<TTag>>[0]; export type EnvironmentRpcSuccess<TTag extends EnvironmentUnaryRpcTag> = RpcMethod<TTag> extends (input: any, options?: any) => Effect.Effect<infer A, any, any> ? A : never; export type EnvironmentRpcFailure<TTag extends EnvironmentUnaryRpcTag> = RpcMethod<TTag> extends (input: any, options?: any) => Effect.Effect<any, infer E, any> ? E : never; export type EnvironmentRpcStreamValue<TTag extends EnvironmentStreamRpcTag> = RpcMethod<TTag> extends (input: any, options?: any) => Stream.Stream<infer A, any, any> ? A : never; export type EnvironmentRpcStreamFailure<TTag extends EnvironmentStreamRpcTag> = RpcMethod<TTag> extends (input: any, options?: any) => Stream.Stream<any, infer E, any> ? E : never; const currentSession = Effect.fn("EnvironmentRpc.currentSession")(function* (method?: string) { const supervisor = yield* EnvironmentSupervisor; return yield* SubscriptionRef.get(supervisor.session).pipe( Effect.flatMap( Option.match({ onNone: () => Effect.fail( new EnvironmentRpcUnavailableError({ environmentId: supervisor.target.environmentId, environmentLabel: supervisor.target.label, ...(method === undefined ? {} : { method }), }), ), onSome: Effect.succeed, }), ), );}); export const request = Effect.fn("EnvironmentRpc.request")(function* < TTag extends EnvironmentUnaryRpcTag,>(tag: TTag, input: EnvironmentRpcInput<TTag>) { const supervisor = yield* EnvironmentSupervisor; yield* Effect.annotateCurrentSpan({ "environment.id": supervisor.target.environmentId, "rpc.method": tag, }); const session = yield* currentSession(tag); const observer = yield* EnvironmentRpcRequestObserver; const method = session.client[tag] as ( input: EnvironmentRpcInput<TTag>, ) => Effect.Effect<EnvironmentRpcSuccess<TTag>, EnvironmentRpcFailure<TTag>>; const completeObservation = yield* observer.observe({ environmentId: supervisor.target.environmentId, method: tag, }); return yield* method(input).pipe(Effect.ensuring(completeObservation));}); export function runStream<TTag extends EnvironmentStreamCommandRpcTag>( tag: TTag, input: EnvironmentRpcInput<TTag>,): Stream.Stream< EnvironmentRpcStreamValue<TTag>, EnvironmentRpcStreamFailure<TTag> | EnvironmentRpcUnavailableError, EnvironmentSupervisor> { return Stream.unwrap( currentSession(tag).pipe( Effect.map((session) => { const method = session.client[tag] as ( input: EnvironmentRpcInput<TTag>, ) => Stream.Stream<EnvironmentRpcStreamValue<TTag>, EnvironmentRpcStreamFailure<TTag>>; return method(input); }), ), ).pipe( Stream.withSpan("EnvironmentRpc.runStream", { attributes: { "rpc.method": tag }, }), );} export function subscribe<TTag extends EnvironmentSubscriptionRpcTag>( tag: TTag, input: EnvironmentRpcInput<TTag>, options?: { readonly onExpectedFailure?: ( cause: Cause.Cause<EnvironmentRpcStreamFailure<TTag>>, ) => Effect.Effect<void, never, never>; readonly retryExpectedFailureAfter?: Duration.Input; },): Stream.Stream< EnvironmentRpcStreamValue<TTag>, EnvironmentRpcStreamFailure<TTag>, EnvironmentSupervisor> { return Stream.unwrap( EnvironmentSupervisor.pipe( Effect.map((supervisor) => SubscriptionRef.changes(supervisor.session).pipe( Stream.switchMap( Option.match({ onNone: () => Stream.empty, onSome: (session) => { const method = session.client[tag] as ( input: EnvironmentRpcInput<TTag>, ) => Stream.Stream< EnvironmentRpcStreamValue<TTag>, EnvironmentRpcStreamFailure<TTag> >; const subscribeToSession = (): Stream.Stream< EnvironmentRpcStreamValue<TTag>, EnvironmentRpcStreamFailure<TTag> > => Stream.suspend(() => method(input).pipe( Stream.catchCause((cause) => { const hasOnlyExpectedFailures = cause.reasons.length > 0 && cause.reasons.every((reason) => reason._tag === "Fail"); const isTransportFailure = hasOnlyExpectedFailures && cause.reasons.every( (reason) => reason._tag === "Fail" && isRpcClientError(reason.error), ); if (isTransportFailure) { return Stream.fromEffect( Effect.logWarning( "Durable RPC subscription lost its transport; waiting for the next session.", { errorTag: "RpcClientError", reasonCount: cause.reasons.length, method: tag, environmentId: supervisor.target.environmentId, }, ), ).pipe(Stream.drain); } if (hasOnlyExpectedFailures && options?.onExpectedFailure !== undefined) { const handled = Stream.fromEffect(options.onExpectedFailure(cause)).pipe( Stream.drain, ); if (options.retryExpectedFailureAfter === undefined) { return handled; } return handled.pipe( Stream.concat( Stream.fromEffect( Effect.sleep(options.retryExpectedFailureAfter), ).pipe(Stream.drain), ), Stream.concat(subscribeToSession()), ); } return Stream.failCause(cause); }), ), ); return subscribeToSession(); }, }), ), ), ), ), ).pipe( Stream.withSpan("EnvironmentRpc.subscribe", { attributes: { "rpc.method": tag }, }), );} export const config = Effect.gen(function* () { const session = yield* currentSession(); return yield* session.initialConfig;}).pipe(Effect.withSpan("EnvironmentRpc.config"));
import { EnvironmentId, ORCHESTRATION_WS_METHODS, WS_METHODS } from "@t3tools/contracts";
import * as Cause from "effect/Cause";
import * as Context from "effect/Context";
import type * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as Option from "effect/Option";
import * as Schema from "effect/Schema";
import * as Stream from "effect/Stream";
import * as SubscriptionRef from "effect/SubscriptionRef";
import { RpcClientError } from "effect/unstable/rpc";
import { EnvironmentSupervisor } from "../connection/supervisor.ts";
import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
export class EnvironmentRpcUnavailableError extends Schema.TaggedErrorClass<EnvironmentRpcUnavailableError>()(
"EnvironmentRpcUnavailableError",
{
environmentId: EnvironmentId,
environmentLabel: Schema.String,
method: Schema.optionalKey(Schema.String),
},
) {
override get message(): string {
const method = this.method === undefined ? "" : ` for RPC method ${this.method}`;
return `${this.environmentLabel} is not connected${method}.`;
}
}
export const isEnvironmentRpcUnavailableError = Schema.is(EnvironmentRpcUnavailableError);
export interface EnvironmentRpcRequestObservation {
readonly environmentId: string;
readonly method: string;
}
export class EnvironmentRpcRequestObserver extends Context.Reference<{
readonly observe: (
request: EnvironmentRpcRequestObservation,
) => Effect.Effect<Effect.Effect<void>>;
}>("@t3tools/client-runtime/rpc/EnvironmentRpcRequestObserver", {
defaultValue: () => ({
observe: () => Effect.succeed(Effect.void),
}),
}) {}
export type EnvironmentRpcTag = keyof WsRpcProtocolClient & string;
type RpcMethod<TTag extends EnvironmentRpcTag> = WsRpcProtocolClient[TTag];
export type EnvironmentSubscriptionRpcTag =
| typeof ORCHESTRATION_WS_METHODS.subscribeShell
| typeof ORCHESTRATION_WS_METHODS.subscribeThread
| typeof WS_METHODS.subscribeAuthAccess
| typeof WS_METHODS.subscribeServerConfig
| typeof WS_METHODS.subscribeServerLifecycle
| typeof WS_METHODS.subscribeTerminalEvents
| typeof WS_METHODS.subscribeTerminalMetadata
| typeof WS_METHODS.subscribePreviewEvents
| typeof WS_METHODS.subscribeDiscoveredLocalServers
| typeof WS_METHODS.previewAutomationConnect
| typeof WS_METHODS.subscribeVcsStatus
| typeof WS_METHODS.terminalAttach;
export type EnvironmentStreamCommandRpcTag =
| typeof WS_METHODS.cloudInstallRelayClient
| typeof WS_METHODS.gitRunStackedAction;
export type EnvironmentStreamRpcTag =
| EnvironmentSubscriptionRpcTag
| EnvironmentStreamCommandRpcTag;
export type EnvironmentUnaryRpcTag = Exclude<EnvironmentRpcTag, EnvironmentStreamRpcTag>;
const isRpcClientError = Schema.is(RpcClientError.RpcClientError);
export type EnvironmentRpcInput<TTag extends EnvironmentRpcTag> = Parameters<RpcMethod<TTag>>[0];
export type EnvironmentRpcSuccess<TTag extends EnvironmentUnaryRpcTag> =
RpcMethod<TTag> extends (input: any, options?: any) => Effect.Effect<infer A, any, any>
? A
: never;
export type EnvironmentRpcFailure<TTag extends EnvironmentUnaryRpcTag> =
RpcMethod<TTag> extends (input: any, options?: any) => Effect.Effect<any, infer E, any>
? E
: never;
export type EnvironmentRpcStreamValue<TTag extends EnvironmentStreamRpcTag> =
RpcMethod<TTag> extends (input: any, options?: any) => Stream.Stream<infer A, any, any>
? A
: never;
export type EnvironmentRpcStreamFailure<TTag extends EnvironmentStreamRpcTag> =
RpcMethod<TTag> extends (input: any, options?: any) => Stream.Stream<any, infer E, any>
? E
: never;
const currentSession = Effect.fn("EnvironmentRpc.currentSession")(function* (method?: string) {
const supervisor = yield* EnvironmentSupervisor;
return yield* SubscriptionRef.get(supervisor.session).pipe(
Effect.flatMap(
Option.match({
onNone: () =>
Effect.fail(
new EnvironmentRpcUnavailableError({
environmentId: supervisor.target.environmentId,
environmentLabel: supervisor.target.label,
...(method === undefined ? {} : { method }),
}),
),
onSome: Effect.succeed,
}),
),
);
});
export const request = Effect.fn("EnvironmentRpc.request")(function* <
TTag extends EnvironmentUnaryRpcTag,
>(tag: TTag, input: EnvironmentRpcInput<TTag>) {
const supervisor = yield* EnvironmentSupervisor;
yield* Effect.annotateCurrentSpan({
"environment.id": supervisor.target.environmentId,
"rpc.method": tag,
});
const session = yield* currentSession(tag);
const observer = yield* EnvironmentRpcRequestObserver;
const method = session.client[tag] as (
input: EnvironmentRpcInput<TTag>,
) => Effect.Effect<EnvironmentRpcSuccess<TTag>, EnvironmentRpcFailure<TTag>>;
const completeObservation = yield* observer.observe({
environmentId: supervisor.target.environmentId,
method: tag,
});
return yield* method(input).pipe(Effect.ensuring(completeObservation));
});
export function runStream<TTag extends EnvironmentStreamCommandRpcTag>(
tag: TTag,
input: EnvironmentRpcInput<TTag>,
): Stream.Stream<
EnvironmentRpcStreamValue<TTag>,
EnvironmentRpcStreamFailure<TTag> | EnvironmentRpcUnavailableError,
EnvironmentSupervisor
> {
return Stream.unwrap(
currentSession(tag).pipe(
Effect.map((session) => {
const method = session.client[tag] as (
input: EnvironmentRpcInput<TTag>,
) => Stream.Stream<EnvironmentRpcStreamValue<TTag>, EnvironmentRpcStreamFailure<TTag>>;
return method(input);
}),
),
).pipe(
Stream.withSpan("EnvironmentRpc.runStream", {
attributes: { "rpc.method": tag },
}),
);
}
export function subscribe<TTag extends EnvironmentSubscriptionRpcTag>(
tag: TTag,
input: EnvironmentRpcInput<TTag>,
options?: {
readonly onExpectedFailure?: (
cause: Cause.Cause<EnvironmentRpcStreamFailure<TTag>>,
) => Effect.Effect<void, never, never>;
readonly retryExpectedFailureAfter?: Duration.Input;
},
): Stream.Stream<
EnvironmentRpcStreamValue<TTag>,
EnvironmentRpcStreamFailure<TTag>,
EnvironmentSupervisor
> {
return Stream.unwrap(
EnvironmentSupervisor.pipe(
Effect.map((supervisor) =>
SubscriptionRef.changes(supervisor.session).pipe(
Stream.switchMap(
Option.match({
onNone: () => Stream.empty,
onSome: (session) => {
const method = session.client[tag] as (
input: EnvironmentRpcInput<TTag>,
) => Stream.Stream<
EnvironmentRpcStreamValue<TTag>,
EnvironmentRpcStreamFailure<TTag>
>;
const subscribeToSession = (): Stream.Stream<
EnvironmentRpcStreamValue<TTag>,
EnvironmentRpcStreamFailure<TTag>
> =>
Stream.suspend(() =>
method(input).pipe(
Stream.catchCause((cause) => {
const hasOnlyExpectedFailures =
cause.reasons.length > 0 &&
cause.reasons.every((reason) => reason._tag === "Fail");
const isTransportFailure =
hasOnlyExpectedFailures &&
cause.reasons.every(
(reason) => reason._tag === "Fail" && isRpcClientError(reason.error),
);
if (isTransportFailure) {
return Stream.fromEffect(
Effect.logWarning(
"Durable RPC subscription lost its transport; waiting for the next session.",
{
errorTag: "RpcClientError",
reasonCount: cause.reasons.length,
method: tag,
environmentId: supervisor.target.environmentId,
},
),
).pipe(Stream.drain);
}
if (hasOnlyExpectedFailures && options?.onExpectedFailure !== undefined) {
const handled = Stream.fromEffect(options.onExpectedFailure(cause)).pipe(
Stream.drain,
);
if (options.retryExpectedFailureAfter === undefined) {
return handled;
}
return handled.pipe(
Stream.concat(
Stream.fromEffect(
Effect.sleep(options.retryExpectedFailureAfter),
).pipe(Stream.drain),
),
Stream.concat(subscribeToSession()),
);
}
return Stream.failCause(cause);
}),
),
);
return subscribeToSession();
},
}),
),
),
),
),
).pipe(
Stream.withSpan("EnvironmentRpc.subscribe", {
attributes: { "rpc.method": tag },
}),
);
}
export const config = Effect.gen(function* () {
const session = yield* currentSession();
return yield* session.initialConfig;
}).pipe(Effect.withSpan("EnvironmentRpc.config"));