1import type { ChatGptHandoffInput, ChatGptHandoffState } from "@t3tools/contracts";
2import { ProviderSetupError } from "@t3tools/contracts";
3import * as Effect from "effect/Effect";
4import * as Option from "effect/Option";
5import * as Stream from "effect/Stream";
6import { FetchHttpClient } from "effect/http";
7import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
8import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
9import { makeCodexChatGptAuth } from "./CodexChatGptAuth.ts";
10
11// The primary owns OAuth for this stream; the destination owns the refresh session.
12export function subscribeChatGptHandoff(
13 input: ChatGptHandoffInput,
14 owner: string,
15 endpoints: { discoveryUrl: string; resource: string } | undefined = undefined,
16) {
17 return Stream.unwrap(
18 Effect.gen(function* () {
19 const bytes = new Map<string, Uint8Array>();
20 yield* Effect.addFinalizer(() => Effect.sync(() => bytes.clear()));
21 const store = ServerSecretStore.ServerSecretStore.of({
22 get: (key) => Effect.sync(() => Option.fromUndefinedOr(bytes.get(key))),
23 set: (key, value) =>
24 Effect.sync(() => {
25 bytes.set(key, value);
26 }),
27 remove: (key) =>
28 Effect.sync(() => {
29 bytes.delete(key);
30 }),
31 create: () => Effect.die("Handoff does not create persistent secrets."),
32 getOrCreateRandom: () => Effect.die("Handoff uses the destination environment identity."),
33 });
34 const auth = yield* makeCodexChatGptAuth({
35 ...endpoints,
36 instanceId: input.instanceId,
37 reconnectProfile: input.profile,
38 defaultReturnUrl: input.returnUrl,
39 }).pipe(
40 Effect.provideService(ServerSecretStore.ServerSecretStore, store),
41 Effect.provideService(
42 ServerEnvironment.ServerEnvironmentIdentity,
43 ServerEnvironment.ServerEnvironmentIdentity.of({
44 getEnvironmentId: Effect.succeed(input.environmentId),
45 }),
46 ),
47 Effect.provide(FetchHttpClient.layer),
48 );
49 yield* auth.controller.start(owner, Effect.void, "chatgpt", input.returnUrl, "server");
50 return auth.controller.subscribe(owner).pipe(
51 Stream.takeUntil((state) => ["succeeded", "failed", "cancelled"].includes(state.phase)),
52 Stream.mapEffect((state): Effect.Effect<ChatGptHandoffState, ProviderSetupError> =>
53 state.phase === "succeeded"
54 ? auth.exportProfile.pipe(
55 Effect.map((profile) => ({ phase: "finished" as const, profile })),
56 )
57 : Effect.succeed({ phase: "auth" as const, state }),
58 ),
59 );
60 }),
61 );
62}