1/**
2 * CursorUsageAccounts - Cursor's account usage as a usage source, answered
3 * from a cache.
4 *
5 * Cursor's account API is slow, so a read answers from the cached account
6 * history and marks itself `refreshing` while a background refresh runs;
7 * `awaitRefresh` waits for that refresh instead. The caches persist in the
8 * server's state directory, so a restart only refetches the newest edge. See
9 * `accountCache`.
10 *
11 * @module provider-cursor/server/CursorUsageAccounts
12 */
13
14import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost";
15import type { ProviderUsageScan } from "@t3tools/provider-core/server/usage";
16import { writeFileStringAtomically } from "@t3tools/shared/atomicWrite";
17import * as HostProcess from "@t3tools/shared/HostProcess";
18import * as Clock from "effect/Clock";
19import * as Context from "effect/Context";
20import * as Deferred from "effect/Deferred";
21import * as Effect from "effect/Effect";
22import * as Fiber from "effect/Fiber";
23import * as FileSystem from "effect/FileSystem";
24import * as Layer from "effect/Layer";
25import * as Path from "effect/Path";
26import * as Schema from "effect/Schema";
27import * as Semaphore from "effect/Semaphore";
28
29import {
30 CURSOR_ACCOUNT_CACHE_FILE_NAME,
31 CURSOR_ACCOUNT_TTL_MS,
32 cursorFetchRange,
33 decodeCursorAccountCaches,
34 encodeCursorAccountCaches,
35 isCursorCacheFresh,
36 mergeCursorFetch,
37 type CursorAccountCache,
38 type CursorCredentialSource,
39} from "./accountCache.ts";
40import * as CursorAccountReader from "./CursorAccountReader.ts";
41
42const CURSOR_ACCOUNT_READ_ERROR = "Cursor account usage could not be read.";
43
44/** The cache file is narrowed by hand in `accountCache`, so JSON is enough here. */
45const decodeCacheFile = Schema.decodeUnknownEffect(
46 Schema.fromJsonString(Schema.Unknown as unknown as Schema.Codec<unknown>),
47);
48
49export class CursorUsageAccounts extends Context.Service<
50 CursorUsageAccounts,
51 {
52 /** This environment's Cursor account usage source, if it has one to report. */
53 readonly scan: (input: {
54 /** From the same settings snapshot as the rest of the usage read. */
55 readonly keychainUsageEnabled: boolean;
56 readonly windowStartMs: number;
57 readonly retentionCutoffMs: number;
58 readonly awaitRefresh: boolean;
59 }) => Effect.Effect<ReadonlyArray<ProviderUsageScan>>;
60 /** Waits for every cache write scheduled so far, as a restart would need. */
61 readonly awaitPersisted: Effect.Effect<void>;
62 }
63>()("@t3tools/provider-cursor/server/CursorUsageAccounts") {}
64
65const make = Effect.gen(function* () {
66 const fileSystem = yield* FileSystem.FileSystem;
67 const path = yield* Path.Path;
68 const host = yield* ProviderHost.ProviderHost;
69 const accountReader = yield* CursorAccountReader.CursorAccountReader;
70 const hostEnvironment = yield* HostProcess.Environment;
71 const platform = yield* HostProcess.Platform;
72
73 const cachePath = path.join(host.paths.stateDir, CURSOR_ACCOUNT_CACHE_FILE_NAME);
74 /** Account caches by credential source. */
75 const caches = new Map<string, CursorAccountCache>();
76 let cacheDirty = false;
77 /**
78 * The last failed refresh per credential source, standing for a TTL so a
79 * client refetching a broken login does not refetch Cursor each time. A
80 * `null` error means there is no login: no source to report.
81 */
82 const failures = new Map<string, { readonly atMs: number; readonly error: string | null }>();
83 /** The refresh in flight per credential source, which every read joins. */
84 const refreshes = new Map<string, Deferred.Deferred<void>>();
85
86 /** Loads the persisted caches once; concurrent first readers await the same load. */
87 const ensureLoaded = yield* Effect.cached(
88 fileSystem.readFileString(cachePath).pipe(
89 Effect.flatMap((raw) => decodeCacheFile(raw)),
90 Effect.catchCause(() => Effect.succeed(null)),
91 Effect.map((document) => {
92 for (const [key, cache] of decodeCursorAccountCaches(document)) caches.set(key, cache);
93 }),
94 ),
95 );
96
97 // Serialized so an older snapshot never lands after a newer one. The dirty
98 // flag is cleared before encoding, so a change while the write is in flight
99 // marks it dirty again; a failed write restores it so the next persist retries.
100 const persistLock = yield* Semaphore.make(1);
101 const persist = Effect.gen(function* () {
102 if (!cacheDirty) return;
103 cacheDirty = false;
104 yield* Effect.sync(() => encodeCursorAccountCaches(caches)).pipe(
105 Effect.flatMap((contents) => writeFileStringAtomically({ filePath: cachePath, contents })),
106 Effect.provideService(FileSystem.FileSystem, fileSystem),
107 Effect.provideService(Path.Path, path),
108 Effect.catchCause(() =>
109 Effect.sync(() => {
110 cacheDirty = true;
111 }),
112 ),
113 );
114 }).pipe(persistLock.withPermit, Effect.withSpan("CursorUsageAccounts.persist"));
115
116 const pendingPersists = new Set<Fiber.Fiber<void>>();
117 /** Writes the caches in the background, after the read that dirtied them answers. */
118 const schedulePersist = Effect.forkDetach(persist).pipe(
119 Effect.map((fiber) => {
120 pendingPersists.add(fiber);
121 fiber.addObserver(() => pendingPersists.delete(fiber));
122 }),
123 );
124 const awaitPersisted = Effect.suspend(() => Fiber.awaitAll([...pendingPersists])).pipe(
125 Effect.asVoid,
126 );
127 // A write still running at shutdown finishes first, so the next start keeps it.
128 yield* Effect.addFinalizer(() => awaitPersisted);
129
130 /** Fetches what one account cache is missing, then persists it if anything changed. */
131 const refreshAccount = Effect.fn("CursorUsageAccounts.refreshAccount")(function* (
132 credential: CursorCredentialSource,
133 credentialKey: string,
134 retentionStartMs: number,
135 ) {
136 const nowMs = yield* Clock.currentTimeMillis;
137 const fetchMissing = (cache: CursorAccountCache | undefined) => {
138 const range = cursorFetchRange(cache, retentionStartMs, nowMs);
139 return accountReader
140 .read(credential, range.sinceMs, range.untilMs)
141 .pipe(Effect.map((result) => ({ range, result })));
142 };
143 let base = caches.get(credentialKey);
144 let fetched = yield* fetchMissing(base);
145 // Another login's history replaces the cached account's, even if reading it fails.
146 if (
147 base !== undefined &&
148 fetched.result.accountKey !== null &&
149 fetched.result.accountKey !== base.accountKey
150 ) {
151 caches.delete(credentialKey);
152 cacheDirty = true;
153 base = undefined;
154 fetched = yield* fetchMissing(base);
155 }
156 const { range, result } = fetched;
157 if (result.missing || result.error !== null || result.accountKey === null) {
158 failures.set(credentialKey, {
159 atMs: nowMs,
160 error: result.missing || result.error !== null ? result.error : CURSOR_ACCOUNT_READ_ERROR,
161 });
162 yield* schedulePersist;
163 return;
164 }
165 const merged = mergeCursorFetch(
166 base,
167 result.accountKey,
168 range,
169 result.records,
170 nowMs,
171 retentionStartMs,
172 );
173 caches.set(credentialKey, merged.cache);
174 failures.delete(credentialKey);
175 // An unchanged edge is not worth rewriting the file for: after a restart
176 // the cache just refetches a slightly wider edge. Persisting anyway retries
177 // an earlier failed write; it is a no-op when nothing is dirty.
178 if (merged.changed) cacheDirty = true;
179 yield* schedulePersist;
180 });
181
182 /** Joins the refresh in flight, or starts one. */
183 const startRefresh = (
184 credential: CursorCredentialSource,
185 credentialKey: string,
186 retentionStartMs: number,
187 ) =>
188 // Enrollment and fork are atomic, so an interrupted caller cannot leave a
189 // registered refresh that nothing will finish.
190 Effect.uninterruptible(
191 Effect.gen(function* () {
192 const current = refreshes.get(credentialKey);
193 if (current !== undefined) return current;
194 const done = Deferred.makeUnsafe<void>();
195 refreshes.set(credentialKey, done);
196 // Detached: a departing client must not cancel a fetch later reads reuse.
197 yield* refreshAccount(credential, credentialKey, retentionStartMs).pipe(
198 Effect.catchCause(() =>
199 Clock.currentTimeMillis.pipe(
200 Effect.map((atMs) =>
201 failures.set(credentialKey, { atMs, error: CURSOR_ACCOUNT_READ_ERROR }),
202 ),
203 ),
204 ),
205 Effect.ensuring(
206 Effect.suspend(() => {
207 refreshes.delete(credentialKey);
208 return Deferred.succeed(done, undefined);
209 }),
210 ),
211 Effect.forkDetach,
212 );
213 return done;
214 }),
215 );
216
217 /**
218 * The account source. A cache inside its TTL, or a refresh that failed
219 * inside it, answers directly. Otherwise a refresh starts: `awaitRefresh`
220 * waits for it, and anything else answers from the cache marked `refreshing`.
221 */
222 const accountSource = Effect.fn("CursorUsageAccounts.accountSource")(function* (
223 credential: CursorCredentialSource,
224 authPath: string,
225 windowStartMs: number,
226 retentionStartMs: number,
227 awaitRefresh: boolean,
228 ) {
229 // No saved login means there is no account source to report, not a setup error.
230 if (
231 typeof credential === "string" &&
232 !(yield* fileSystem.exists(credential).pipe(Effect.orElseSucceed(() => true)))
233 ) {
234 return null;
235 }
236 yield* ensureLoaded;
237 const credentialKey = typeof credential === "string" ? credential : "keychain";
238 const nowMs = yield* Clock.currentTimeMillis;
239 const recentFailure = failures.get(credentialKey);
240 let refreshing = false;
241 if (
242 (recentFailure === undefined || nowMs - recentFailure.atMs >= CURSOR_ACCOUNT_TTL_MS) &&
243 !isCursorCacheFresh(caches.get(credentialKey), nowMs)
244 ) {
245 const refresh = yield* startRefresh(credential, credentialKey, retentionStartMs);
246 if (awaitRefresh) yield* Deferred.await(refresh);
247 else refreshing = true;
248 }
249
250 const cache = caches.get(credentialKey);
251 const failure = refreshing ? undefined : failures.get(credentialKey);
252 const failureMessage = failure === undefined ? undefined : failure.error;
253 if (failureMessage === null) return null;
254 if (cache === undefined) {
255 return {
256 dir: authPath,
257 // Never combine a local fallback with another server's account-wide history.
258 ...(refreshing
259 ? { files: [], refreshing: true }
260 : { files: null, message: failureMessage ?? CURSOR_ACCOUNT_READ_ERROR }),
261 } satisfies ProviderUsageScan;
262 }
263 // The same account includes CLI and desktop history from every machine.
264 // A stable remote fingerprint prevents connected environments counting it twice.
265 const source = `cursor-account:${cache.accountKey}`;
266 return {
267 dir: source,
268 hostId: "cursor.com",
269 volumeId: cache.accountKey,
270 files: [
271 {
272 path: source,
273 records: cache.records.filter((record) => record.timestampMs >= windowStartMs),
274 },
275 ],
276 ...(failureMessage === undefined
277 ? { status: "ok" }
278 : { status: "partial", message: failureMessage }),
279 ...(refreshing ? { refreshing: true } : {}),
280 } satisfies ProviderUsageScan;
281 });
282
283 const scan: CursorUsageAccounts["Service"]["scan"] = Effect.fn("CursorUsageAccounts.scan")(
284 function* ({ keychainUsageEnabled, windowStartMs, retentionCutoffMs, awaitRefresh }) {
285 const home = yield* HostProcess.HomeDirectory;
286 const userHome =
287 (platform === "win32" ? hostEnvironment["USERPROFILE"] : hostEnvironment["HOME"]) || home;
288 const configHome = hostEnvironment["XDG_CONFIG_HOME"]?.trim();
289 const cursorHome =
290 platform === "darwin"
291 ? path.join(userHome, "Library", "Application Support")
292 : platform === "win32"
293 ? hostEnvironment["APPDATA"] || path.join(userHome, "AppData", "Roaming")
294 : configHome && path.isAbsolute(configHome)
295 ? configHome
296 : path.join(userHome, ".config");
297 const authPath =
298 platform === "darwin"
299 ? path.join(userHome, ".cursor", "auth.json")
300 : path.join(cursorHome, platform === "win32" ? "Cursor" : "cursor", "auth.json");
301 const credentialStore = hostEnvironment["AGENT_CLI_CREDENTIAL_STORE"];
302 const loginUnavailable =
303 Boolean(hostEnvironment["CURSOR_AUTH_TOKEN"]?.trim()) ||
304 Boolean(hostEnvironment["CURSOR_API_KEY"]?.trim()) ||
305 credentialStore === "memory";
306 const useKeychain = platform === "darwin" && credentialStore !== "file";
307 if (useKeychain && !loginUnavailable && !keychainUsageEnabled) {
308 return [
309 {
310 dir: authPath,
311 volumeId: "",
312 files: null,
313 message: "Cursor account usage is off on this environment.",
314 action: "enableCursorKeychain",
315 },
316 ];
317 }
318 if (loginUnavailable) {
319 return [
320 {
321 dir: authPath,
322 files: null,
323 message: "Cursor account history needs a Cursor CLI login on this server.",
324 },
325 ];
326 }
327 const source = yield* accountSource(
328 useKeychain ? { kind: "keychain" } : authPath,
329 authPath,
330 windowStartMs,
331 retentionCutoffMs,
332 awaitRefresh,
333 );
334 return source === null ? [] : [source];
335 },
336 );
337
338 return CursorUsageAccounts.of({ scan, awaitPersisted });
339});
340
341export const layer = Layer.effect(CursorUsageAccounts, make);