1import {
2 ProviderSetupError,
3 type ChatGptReconnectProfile,
4 type ChatGptTransferredProfile,
5 type ProviderAuthRespondInput,
6 type ProviderAuthStartInput,
7 type ProviderAuthState,
8 type ProviderInstanceId,
9 type ProviderSessionId,
10} from "@t3tools/contracts";
11import * as Context from "effect/Context";
12import * as Effect from "effect/Effect";
13import * as Layer from "effect/Layer";
14import type * as Scope from "effect/Scope";
15import * as Stream from "effect/Stream";
16import * as Semaphore from "effect/Semaphore";
17
18import * as ProviderSessionManager from "../orchestration-v2/ProviderSessionManager.ts";
19import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
20import * as ProviderInstanceRegistry from "./ProviderInstanceRegistry.ts";
21import type { ProviderAuthController } from "@t3tools/provider-core/server/auth";
22
23export type { ProviderAuthController } from "@t3tools/provider-core/server/auth";
24
25interface ProviderAuthTarget {
26 readonly instanceId: ProviderInstanceId;
27}
28
29export class ProviderAuthService extends Context.Service<
30 ProviderAuthService,
31 {
32 readonly reconnectProfile: (
33 input: ProviderAuthTarget & { methodId: string },
34 ) => Effect.Effect<ChatGptReconnectProfile | null, ProviderSetupError>;
35 readonly importProfile: (
36 input: ProviderAuthTarget & { profile: ChatGptTransferredProfile },
37 ) => Effect.Effect<ProviderAuthState, ProviderSetupError>;
38 readonly start: (
39 input: ProviderAuthStartInput,
40 ownerSessionId: string,
41 ) => Effect.Effect<ProviderAuthState, ProviderSetupError>;
42 readonly complete: (
43 input: ProviderAuthTarget & { readonly flowId: string; readonly callbackUrl: string },
44 ownerSessionId: string,
45 ) => Effect.Effect<ProviderAuthState, ProviderSetupError>;
46 readonly respond: (
47 input: ProviderAuthRespondInput,
48 ownerSessionId: string,
49 ) => Effect.Effect<ProviderAuthState, ProviderSetupError>;
50 readonly cancel: (
51 input: ProviderAuthTarget & { readonly flowId: string },
52 ownerSessionId: string,
53 ) => Effect.Effect<ProviderAuthState, ProviderSetupError>;
54 readonly logout: (
55 input: ProviderAuthTarget,
56 ) => Effect.Effect<ProviderAuthState, ProviderSetupError>;
57 readonly subscribe: (
58 input: ProviderAuthTarget,
59 ownerSessionId: string,
60 ) => Stream.Stream<ProviderAuthState, ProviderSetupError>;
61 readonly tryHandlePromptCommand: (
62 input: ProviderAuthTarget & { readonly text: string; readonly hasAttachments: boolean },
63 ) => Effect.Effect<boolean, ProviderSetupError>;
64 }
65>()("t3/provider/ProviderAuthService") {}
66
67export const makeProviderAuthService = Effect.gen(function* () {
68 const registry = yield* ProviderInstanceRegistry.ProviderInstanceRegistry;
69 const projections = yield* ProjectionStore.ProjectionStoreV2;
70 const providerSessions = yield* ProviderSessionManager.ProviderSessionManagerV2;
71 const credentialChanges = yield* Semaphore.make(1);
72
73 const getController = Effect.fn("ProviderAuthService.getController")(function* (
74 instanceId: ProviderInstanceId,
75 operation: string,
76 ) {
77 const instance = yield* registry.getInstance(instanceId);
78 if (!instance?.auth) {
79 return yield* new ProviderSetupError({
80 instanceId,
81 operation,
82 detail: instance
83 ? "This provider does not support sign-in in T3 Code."
84 : "This provider instance is no longer available.",
85 });
86 }
87 return instance.auth;
88 });
89
90 // Native sessions may still belong to the previous provider after the
91 // selected model changes. Read session bindings, not the selected model,
92 // when invalidating credentials for sign-in or sign-out.
93 const stopSessions = Effect.fn("ProviderAuthService.stopSessions")(function* (
94 instanceId: ProviderInstanceId,
95 binding: ProviderAuthController["credentialBinding"],
96 ) {
97 const failure = (detail: string) =>
98 new ProviderSetupError({ instanceId, operation: "stopSessions", detail });
99 const affectedIds = new Set([
100 instanceId,
101 ...(binding === undefined
102 ? []
103 : (yield* registry.listInstances)
104 .filter(
105 (instance) =>
106 instance.auth?.credentialBinding?.key === binding.key &&
107 instance.auth.credentialBinding.owner === binding.owner,
108 )
109 .map((instance) => instance.instanceId)),
110 ]);
111 const threadIds = yield* projections
112 .getRecoveryThreadIds("runtime")
113 .pipe(
114 Effect.mapError(() => failure("Could not read the provider's active sessions. Try again.")),
115 );
116 const released = new Set<ProviderSessionId>();
117 yield* Effect.forEach(
118 threadIds,
119 (threadId) =>
120 projections.getThreadRecords(threadId, ["providerSessions"]).pipe(
121 Effect.flatMap((projection) =>
122 Effect.forEach(
123 projection.providerSessions.filter(
124 (session) =>
125 affectedIds.has(session.providerInstanceId) &&
126 session.status !== "stopped" &&
127 session.status !== "error" &&
128 !released.has(session.id),
129 ),
130 (session) =>
131 Effect.gen(function* () {
132 if (session.providerInstanceId !== instanceId) {
133 const current = yield* registry.getInstance(session.providerInstanceId);
134 if (
135 !binding ||
136 current?.auth?.credentialBinding?.key !== binding.key ||
137 current.auth.credentialBinding.owner !== binding.owner
138 )
139 return;
140 }
141 yield* providerSessions
142 .release({
143 providerSessionId: session.id,
144 reason: "manual_shutdown",
145 detail: "Provider sign-in changed.",
146 })
147 .pipe(Effect.tap(() => Effect.sync(() => released.add(session.id))));
148 }),
149 { discard: true },
150 ),
151 ),
152 Effect.mapError(() =>
153 failure("Could not stop all sessions for this provider. Try again."),
154 ),
155 ),
156 { discard: true },
157 );
158 if (binding) {
159 yield* Effect.forEach(
160 (yield* registry.listInstances).filter(
161 (instance) =>
162 instance.instanceId !== instanceId &&
163 affectedIds.has(instance.instanceId) &&
164 instance.auth?.credentialBinding?.key === binding.key &&
165 instance.auth.credentialBinding.owner === binding.owner,
166 ),
167 (instance) => instance.auth?.invalidate ?? Effect.void,
168 { discard: true },
169 );
170 }
171 });
172
173 const checkSharedBinding = Effect.fnUntraced(function* (
174 instanceId: ProviderInstanceId,
175 operation: "start" | "logout",
176 auth: ProviderAuthController,
177 ) {
178 const binding = auth.credentialBinding;
179 if (!binding) return;
180 const instances = yield* registry.listInstances;
181 for (const instance of instances) {
182 if (
183 instance.instanceId !== instanceId &&
184 instance.auth?.credentialBinding?.key === binding.key &&
185 instance.auth.credentialBinding.owner === binding.owner &&
186 instance.auth.isChangingCredentials &&
187 (yield* instance.auth.isChangingCredentials)
188 ) {
189 return yield* new ProviderSetupError({
190 instanceId,
191 operation,
192 detail:
193 "Another provider instance is changing this shared sign-in. Finish or cancel it first.",
194 });
195 }
196 }
197 });
198
199 return ProviderAuthService.of({
200 reconnectProfile: Effect.fnUntraced(function* (input) {
201 const auth = yield* getController(input.instanceId, "export");
202 if (!auth.reconnectProfile)
203 return yield* new ProviderSetupError({
204 instanceId: input.instanceId,
205 operation: "export",
206 detail: "This provider does not support ChatGPT profile transfer.",
207 });
208 return yield* auth.reconnectProfile(input.methodId);
209 }),
210 importProfile: (input) =>
211 credentialChanges.withPermit(
212 Effect.gen(function* () {
213 const auth = yield* getController(input.instanceId, "import");
214 yield* checkSharedBinding(input.instanceId, "start", auth);
215 if (!auth.importProfile)
216 return yield* new ProviderSetupError({
217 instanceId: input.instanceId,
218 operation: "import",
219 detail: "This provider does not support ChatGPT profile transfer.",
220 });
221 return yield* auth.importProfile(
222 input.profile,
223 stopSessions(input.instanceId, auth.credentialBinding),
224 );
225 }),
226 ),
227 start: Effect.fn("ProviderAuthService.start")(function* (input, ownerSessionId) {
228 return yield* credentialChanges.withPermit(
229 Effect.gen(function* () {
230 const auth = yield* getController(input.instanceId, "start");
231 yield* checkSharedBinding(input.instanceId, "start", auth);
232 return yield* auth.start(
233 ownerSessionId,
234 stopSessions(input.instanceId, auth.credentialBinding),
235 input.methodId,
236 input.returnUrl,
237 input.callbackMode,
238 );
239 }),
240 );
241 }),
242 respond: Effect.fn("ProviderAuthService.respond")(function* (input, ownerSessionId) {
243 const auth = yield* getController(input.instanceId, "respond");
244 if (!auth.respond) {
245 return yield* new ProviderSetupError({
246 instanceId: input.instanceId,
247 operation: "respond",
248 detail: "This provider does not accept this sign-in interaction.",
249 });
250 }
251 return yield* auth.respond(ownerSessionId, input);
252 }),
253 complete: Effect.fn("ProviderAuthService.complete")(function* (input, ownerSessionId) {
254 const auth = yield* getController(input.instanceId, "complete");
255 return yield* auth.complete(ownerSessionId, input);
256 }),
257 cancel: Effect.fn("ProviderAuthService.cancel")(function* (input, ownerSessionId) {
258 const auth = yield* getController(input.instanceId, "cancel");
259 return yield* auth.cancel(ownerSessionId, input.flowId);
260 }),
261 logout: Effect.fn("ProviderAuthService.logout")(function* (input) {
262 return yield* credentialChanges.withPermit(
263 Effect.gen(function* () {
264 const auth = yield* getController(input.instanceId, "logout");
265 yield* checkSharedBinding(input.instanceId, "logout", auth);
266 return yield* auth.logout(stopSessions(input.instanceId, auth.credentialBinding));
267 }),
268 );
269 }),
270 subscribe: (input, ownerSessionId) =>
271 Effect.gen(function* () {
272 const changes = yield* registry.subscribeChanges;
273 const initial = yield* getController(input.instanceId, "subscribe");
274 return Stream.concat(
275 Stream.succeed(initial),
276 Stream.fromSubscription(changes).pipe(
277 Stream.mapEffect(() => getController(input.instanceId, "subscribe")),
278 ),
279 ).pipe(
280 Stream.changesWith((previous, next) => previous === next),
281 Stream.switchMap((auth) => auth.subscribe(ownerSessionId)),
282 );
283 }).pipe(Stream.unwrap),
284 tryHandlePromptCommand: Effect.fn("ProviderAuthService.tryHandlePromptCommand")(
285 function* (input) {
286 const instance = yield* registry.getInstance(input.instanceId);
287 if (!instance?.auth?.isLogoutPrompt?.(input.text, input.hasAttachments)) {
288 return false;
289 }
290 return yield* credentialChanges.withPermit(
291 Effect.gen(function* () {
292 const auth = yield* getController(input.instanceId, "logout");
293 if (!auth.isLogoutPrompt?.(input.text, input.hasAttachments)) return false;
294 yield* checkSharedBinding(input.instanceId, "logout", auth);
295 yield* auth.logout(stopSessions(input.instanceId, auth.credentialBinding));
296 return true;
297 }),
298 );
299 },
300 ),
301 });
302});
303
304export const layer = Layer.effect(ProviderAuthService, makeProviderAuthService);