1import {
2 SharedAccountAccess,
3 SharedAccountId,
4 SharedAccountError,
5 SharedAccountResponse,
6 SharedAccountCommit,
7 type SharedAccountRequest,
8} from "@t3tools/contracts";
9import * as KeyedLock from "@t3tools/shared/KeyedLock";
10import { decodeJwt } from "jose";
11import * as Clock from "effect/Clock";
12import * as Context from "effect/Context";
13import * as Crypto from "effect/Crypto";
14import * as Effect from "effect/Effect";
15import * as Redacted from "effect/Redacted";
16import * as Layer from "effect/Layer";
17import * as Option from "effect/Option";
18import * as Schema from "effect/Schema";
19import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http";
20import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
21import * as SharedAccounts from "./SharedAccounts.ts";
22import * as SharedServer from "../environment/SharedServer.ts";
23
24const Connection = Schema.Struct({ url: Schema.String, token: Schema.NonEmptyString });
25const CachedAccess = Schema.Struct({ revision: Schema.Int, credentials: SharedAccountAccess });
26const Tokens = Schema.Struct({
27 access_token: Schema.NonEmptyString,
28 refresh_token: Schema.optionalKey(Schema.NonEmptyString),
29 id_token: Schema.optionalKey(Schema.NonEmptyString),
30});
31
32const decodeAccess = Schema.decodeEffect(Schema.fromJsonString(CachedAccess));
33const encodeAccess = Schema.encodeEffect(Schema.fromJsonString(CachedAccess));
34const decodeCommit = Schema.decodeEffect(Schema.fromJsonString(SharedAccountCommit));
35const encodeCommit = Schema.encodeEffect(Schema.fromJsonString(SharedAccountCommit));
36const decodeId = Schema.decodeEffect(SharedAccountId);
37const isSharedAccountError = Schema.is(SharedAccountError);
38
39export class SharedAccountClient extends Context.Service<
40 SharedAccountClient,
41 {
42 readonly configure: (input: typeof Connection.Type) => Effect.Effect<void, SharedAccountError>;
43 readonly disconnect: Effect.Effect<void, SharedAccountError>;
44 readonly configured: Effect.Effect<boolean, SharedAccountError>;
45 readonly execute: (
46 request: SharedAccountRequest,
47 ) => Effect.Effect<SharedAccountResponse, SharedAccountError>;
48 readonly access: (
49 id: string,
50 forceRefresh?: boolean,
51 ) => Effect.Effect<SharedAccountAccess, SharedAccountError>;
52 }
53>()("t3/provider/SharedAccountClient") {}
54
55const make = Effect.gen(function* () {
56 const http = yield* HttpClient.HttpClient;
57 const crypto = yield* Crypto.Crypto;
58 const secrets = yield* ServerSecretStore.ServerSecretStore;
59 const local = yield* SharedAccounts.SharedAccounts;
60 const locks = yield* KeyedLock.make<string>();
61
62 const sharedServer = yield* SharedServer.SharedServer;
63 const connection = sharedServer.connection;
64
65 const requestRemote = (target: typeof Connection.Type, input: SharedAccountRequest) =>
66 HttpClientRequest.post(`${target.url}/api/shared/accounts`).pipe(
67 HttpClientRequest.setHeader("authorization", `Bearer ${target.token}`),
68 HttpClientRequest.bodyJson(input),
69 Effect.flatMap(http.execute),
70 Effect.flatMap((response) =>
71 response.status >= 200 && response.status < 300
72 ? HttpClientResponse.schemaBodyJson(SharedAccountResponse)(response)
73 : HttpClientResponse.schemaBodyJson(SharedAccountError)(response).pipe(
74 Effect.flatMap(Effect.fail),
75 ),
76 ),
77 Effect.mapError((cause) =>
78 isSharedAccountError(cause)
79 ? cause
80 : new SharedAccountError({ reason: "connection", cause: Redacted.make(cause) }),
81 ),
82 Effect.timeout("10 seconds"),
83 Effect.catchTags({
84 TimeoutError: (cause) =>
85 new SharedAccountError({ reason: "connection", cause: Redacted.make(cause) }),
86 }),
87 );
88
89 const execute = Effect.fn("SharedAccountClient.execute")(function* (input: SharedAccountRequest) {
90 const target = yield* connection;
91 return yield* Option.isNone(target) ? local.execute(input) : requestRemote(target.value, input);
92 });
93
94 const configure = Effect.fn("SharedAccountClient.configure")(function* (
95 input: typeof Connection.Type,
96 ) {
97 const url = yield* Effect.try({
98 try: () => {
99 const value = new URL(input.url);
100 if (
101 !["https:", "http:"].includes(value.protocol) ||
102 value.username ||
103 value.password ||
104 value.search ||
105 value.hash
106 )
107 throw new Error("Expected an HTTP server URL");
108 return value.toString().replace(/\/$/u, "");
109 },
110 catch: (cause) =>
111 new SharedAccountError({ reason: "connection", cause: Redacted.make(cause) }),
112 });
113 const target = { ...input, url };
114 yield* requestRemote(target, { type: "list" });
115 yield* sharedServer.save(target);
116 });
117
118 const access = Effect.fn("SharedAccountClient.access")(function* (
119 accountId: string,
120 forceRefresh = false,
121 ) {
122 const id = yield* decodeId(accountId).pipe(
123 Effect.mapError(
124 (cause) => new SharedAccountError({ reason: "missing", cause: Redacted.make(cause) }),
125 ),
126 );
127 const target = yield* connection;
128 const origin = Option.isSome(target) ? target.value.url : "local";
129 const digest = yield* crypto.digest("SHA-256", new TextEncoder().encode(origin)).pipe(
130 Effect.map((bytes) => Buffer.from(bytes).toString("hex")),
131 Effect.mapError(
132 (cause) => new SharedAccountError({ reason: "storage", cause: Redacted.make(cause) }),
133 ),
134 );
135 const cacheKey = `shared-account-access-${digest}-${id}`;
136 const pendingKey = `shared-account-refresh-${digest}-${id}`;
137 const pending = yield* secrets.get(pendingKey).pipe(
138 Effect.flatMap((bytes) =>
139 Option.isNone(bytes)
140 ? Effect.succeedNone
141 : decodeCommit(new TextDecoder().decode(bytes.value)).pipe(Effect.map(Option.some)),
142 ),
143 Effect.mapError(
144 (cause) => new SharedAccountError({ reason: "storage", cause: Redacted.make(cause) }),
145 ),
146 );
147 if (Option.isSome(pending)) {
148 yield* execute(pending.value).pipe(
149 Effect.catchTags({
150 SharedAccountError: (error) =>
151 error.reason === "stale" ? Effect.void : Effect.fail(error),
152 }),
153 );
154 yield* secrets
155 .remove(pendingKey)
156 .pipe(
157 Effect.mapError(
158 (cause) => new SharedAccountError({ reason: "storage", cause: Redacted.make(cause) }),
159 ),
160 );
161 }
162 const cached = yield* secrets.get(cacheKey).pipe(
163 Effect.flatMap((bytes) =>
164 Option.isNone(bytes)
165 ? Effect.succeedNone
166 : decodeAccess(new TextDecoder().decode(bytes.value)).pipe(Effect.map(Option.some)),
167 ),
168 Effect.mapError(
169 (cause) => new SharedAccountError({ reason: "storage", cause: Redacted.make(cause) }),
170 ),
171 );
172 const request: SharedAccountRequest = {
173 type: "acquire",
174 id,
175 ...(forceRefresh && Option.isSome(cached) ? { refreshRevision: cached.value.revision } : {}),
176 };
177 const now = yield* Clock.currentTimeMillis;
178 let response = yield* execute(request).pipe(
179 Effect.catchTags({
180 SharedAccountError: (error) => {
181 if (error.reason !== "connection" || forceRefresh || Option.isNone(cached))
182 return Effect.fail(error);
183 const value = cached.value;
184 if (value.credentials.driver === "codex" && value.credentials.expiresAt <= now)
185 return Effect.fail(error);
186 return Effect.succeed<SharedAccountResponse>({
187 type: "access",
188 account: { id, driver: value.credentials.driver, label: id, revision: value.revision },
189 credentials: value.credentials,
190 });
191 },
192 }),
193 );
194 while (response.type === "busy") {
195 if ((yield* Clock.currentTimeMillis) - now >= 8_000)
196 return yield* new SharedAccountError({ reason: "busy" });
197 yield* Effect.sleep("200 millis");
198 response = yield* execute(request);
199 }
200 if (response.type === "refresh") {
201 const claim = response;
202 response = yield* Effect.gen(function* () {
203 if ((yield* Clock.currentTimeMillis) + 8_000 >= claim.expiresAt)
204 return yield* new SharedAccountError({ reason: "stale" });
205 const tokens = yield* HttpClientRequest.post("https://auth.openai.com/oauth/token").pipe(
206 HttpClientRequest.bodyJson({
207 client_id: "app_EMoamEEZ73f0CkXaXp7hrann",
208 grant_type: "refresh_token",
209 refresh_token: claim.credentials.refreshToken,
210 }),
211 Effect.flatMap(http.execute),
212 Effect.flatMap(HttpClientResponse.filterStatusOk),
213 Effect.flatMap(HttpClientResponse.schemaBodyJson(Tokens)),
214 Effect.mapError(
215 (cause) => new SharedAccountError({ reason: "refresh", cause: Redacted.make(cause) }),
216 ),
217 Effect.timeout("8 seconds"),
218 Effect.catchTags({
219 TimeoutError: (cause) =>
220 new SharedAccountError({ reason: "refresh", cause: Redacted.make(cause) }),
221 }),
222 );
223 const expiresAt = yield* Effect.try({
224 try: () => {
225 const exp = decodeJwt(tokens.access_token).exp;
226 if (typeof exp !== "number") throw new Error("Missing token expiry");
227 return exp * 1000;
228 },
229 catch: (cause) =>
230 new SharedAccountError({ reason: "refresh", cause: Redacted.make(cause) }),
231 });
232 const commit: typeof SharedAccountCommit.Type = {
233 type: "commit",
234 id,
235 revision: claim.account.revision,
236 leaseToken: claim.leaseToken,
237 credentials: {
238 ...claim.credentials,
239 accessToken: tokens.access_token,
240 refreshToken: tokens.refresh_token ?? claim.credentials.refreshToken,
241 idToken: tokens.id_token ?? claim.credentials.idToken,
242 expiresAt,
243 },
244 };
245 const journal = yield* encodeCommit(commit).pipe(
246 Effect.mapError(
247 (cause) => new SharedAccountError({ reason: "storage", cause: Redacted.make(cause) }),
248 ),
249 );
250 yield* secrets
251 .set(pendingKey, new TextEncoder().encode(journal))
252 .pipe(
253 Effect.mapError(
254 (cause) => new SharedAccountError({ reason: "storage", cause: Redacted.make(cause) }),
255 ),
256 );
257 const committed = yield* execute(commit);
258 yield* secrets
259 .remove(pendingKey)
260 .pipe(
261 Effect.mapError(
262 (cause) => new SharedAccountError({ reason: "storage", cause: Redacted.make(cause) }),
263 ),
264 );
265 return committed;
266 }).pipe(Effect.uninterruptible);
267 }
268 if (response.type !== "access") return yield* new SharedAccountError({ reason: "connection" });
269 const json = yield* encodeAccess({
270 revision: response.account.revision,
271 credentials: response.credentials,
272 }).pipe(
273 Effect.mapError(
274 (cause) => new SharedAccountError({ reason: "storage", cause: Redacted.make(cause) }),
275 ),
276 );
277 yield* secrets
278 .set(cacheKey, new TextEncoder().encode(json))
279 .pipe(
280 Effect.mapError(
281 (cause) => new SharedAccountError({ reason: "storage", cause: Redacted.make(cause) }),
282 ),
283 );
284 return response.credentials;
285 });
286
287 return SharedAccountClient.of({
288 execute,
289 configure,
290 access: (id, force) => locks.withLock(id, access(id, force)),
291 configured: connection.pipe(Effect.map(Option.isSome)),
292 disconnect: sharedServer.disconnect,
293 });
294});
295
296export const layer = Layer.effect(SharedAccountClient, make).pipe(
297 Layer.provide(SharedServer.layer),
298);