1import { EnvironmentRegistry } from "@t3tools/client-runtime/connection";
2import {
3 EnvironmentRpcSubscriptionObserver,
4 request,
5 type EnvironmentRpcSubscriptionObservation,
6} from "@t3tools/client-runtime/rpc";
7import {
8 type BackgroundScope,
9 type ClientActivityReportInput,
10 type EnvironmentId,
11 WS_METHODS,
12} from "@t3tools/contracts";
13import * as Clock from "effect/Clock";
14import * as DateTime from "effect/DateTime";
15import * as Effect from "effect/Effect";
16import * as Layer from "effect/Layer";
17import * as Queue from "effect/Queue";
18import * as Schedule from "effect/Schedule";
19import * as Stream from "effect/Stream";
20import * as SubscriptionRef from "effect/SubscriptionRef";
21
22import { randomUUID } from "./utils";
23
24const CLIENT_ID_STORAGE_KEY = "t3.backgroundActivity.clientId";
25const REPORT_INTERVAL_MS = 25_000;
26// Shorter than the interval, so a report that never settles cannot hold up the next one.
27const REPORT_TIMEOUT_MS = 15_000;
28const LEASE_TTL_MS = 45_000;
29const RECENT_INTERACTION_WINDOW_MS = LEASE_TTL_MS;
30const BASELINE_SCOPES: ReadonlyArray<BackgroundScope> = [{ type: "provider-status" }];
31
32interface RetainedScope {
33 readonly environmentId: EnvironmentId;
34 readonly scope: BackgroundScope;
35 refCount: number;
36}
37
38const retainedScopes = new Map<string, RetainedScope>();
39const retainedScopeListeners = new Set<() => void>();
40
41function notifyRetainedScopesChanged(): void {
42 for (const listener of retainedScopeListeners) {
43 try {
44 listener();
45 } catch {}
46 }
47}
48
49function stableScopeKey(environmentId: EnvironmentId, scope: BackgroundScope): string {
50 switch (scope.type) {
51 case "server-config":
52 case "diagnostics":
53 return JSON.stringify([environmentId, scope.type]);
54 case "provider-status":
55 return JSON.stringify([environmentId, scope.type, scope.instanceId ?? null]);
56 case "vcs-status":
57 case "git-refs":
58 return JSON.stringify([environmentId, scope.type, scope.cwd]);
59 case "thread":
60 return JSON.stringify([environmentId, scope.type, scope.threadId]);
61 }
62}
63
64function getClientId(): string {
65 try {
66 const existing = window.localStorage.getItem(CLIENT_ID_STORAGE_KEY);
67 if (existing) return existing;
68 const next = randomUUID();
69 window.localStorage.setItem(CLIENT_ID_STORAGE_KEY, next);
70 return next;
71 } catch {
72 return "ephemeral-browser-client";
73 }
74}
75
76function resolveClientKind(): ClientActivityReportInput["clientKind"] {
77 return window.desktopBridge ? "desktop-renderer" : "web";
78}
79
80function wasRecentlyInteracted(lastInteractionAtMs: number, observedAtMs: number): boolean {
81 return (
82 lastInteractionAtMs <= observedAtMs &&
83 observedAtMs - lastInteractionAtMs <= RECENT_INTERACTION_WINDOW_MS
84 );
85}
86
87function createActivityReport(
88 environmentId: EnvironmentId,
89 lastInteractionAtMs: number,
90 observedAtMs: number,
91): ClientActivityReportInput {
92 const scopes = [...BASELINE_SCOPES];
93 for (const entry of retainedScopes.values()) {
94 if (entry.environmentId === environmentId) {
95 scopes.push(entry.scope);
96 }
97 }
98 return {
99 environmentId,
100 clientId: getClientId(),
101 clientKind: resolveClientKind(),
102 visible: document.visibilityState === "visible",
103 focused: document.hasFocus(),
104 recentlyInteracted: wasRecentlyInteracted(lastInteractionAtMs, observedAtMs),
105 appState: document.visibilityState === "visible" ? "active" : "background",
106 scopes,
107 ttlMs: LEASE_TTL_MS,
108 observedAt: DateTime.makeUnsafe(observedAtMs),
109 };
110}
111
112function scopeForSubscription(
113 observation: EnvironmentRpcSubscriptionObservation,
114): BackgroundScope | null {
115 if (observation.method === WS_METHODS.subscribeResourceTelemetry) {
116 return { type: "diagnostics" };
117 }
118 if (observation.method !== WS_METHODS.subscribeVcsStatus) {
119 return null;
120 }
121 const input = observation.input as { readonly cwd?: unknown; readonly includeRemote?: unknown };
122 if (input.includeRemote === false) return null;
123 return typeof input.cwd === "string" ? { type: "vcs-status", cwd: input.cwd } : null;
124}
125
126function retainBackgroundScope(environmentId: EnvironmentId, scope: BackgroundScope): () => void {
127 const key = stableScopeKey(environmentId, scope);
128 const existing = retainedScopes.get(key);
129 if (existing) {
130 existing.refCount += 1;
131 } else {
132 retainedScopes.set(key, { environmentId, scope, refCount: 1 });
133 notifyRetainedScopesChanged();
134 }
135
136 return () => {
137 const current = retainedScopes.get(key);
138 if (!current) return;
139 current.refCount -= 1;
140 if (current.refCount <= 0) {
141 retainedScopes.delete(key);
142 notifyRetainedScopesChanged();
143 }
144 };
145}
146
147export function observeBackgroundActivitySubscription(
148 observation: EnvironmentRpcSubscriptionObservation,
149): Effect.Effect<Effect.Effect<void>> {
150 const scope = scopeForSubscription(observation);
151 if (scope === null) {
152 return Effect.succeed(Effect.void);
153 }
154 return Effect.sync(() => {
155 const release = retainBackgroundScope(observation.environmentId as EnvironmentId, scope);
156 return Effect.sync(release);
157 });
158}
159
160export function retainedBackgroundScopes(
161 environmentId: EnvironmentId,
162): ReadonlyArray<BackgroundScope> {
163 return Array.from(retainedScopes.values(), (entry) =>
164 entry.environmentId === environmentId ? entry.scope : null,
165 ).filter((scope): scope is BackgroundScope => scope !== null);
166}
167
168export const layerObserver = Layer.succeed(
169 EnvironmentRpcSubscriptionObserver,
170 EnvironmentRpcSubscriptionObserver.of({
171 observe: observeBackgroundActivitySubscription,
172 }),
173);
174
175export const layer = Layer.effectDiscard(
176 Effect.gen(function* () {
177 if (typeof window === "undefined" || typeof document === "undefined") {
178 return;
179 }
180
181 const registry = yield* EnvironmentRegistry.EnvironmentRegistry;
182 const clock = yield* Clock.Clock;
183 const reportRequests = yield* Queue.sliding<void>(1);
184 const requestReport = () => Queue.offerUnsafe(reportRequests, undefined);
185 let lastInteractionAtMs = clock.currentTimeMillisUnsafe();
186 const recordInteraction = () => {
187 const observedAtMs = clock.currentTimeMillisUnsafe();
188 const wasRecent = wasRecentlyInteracted(lastInteractionAtMs, observedAtMs);
189 lastInteractionAtMs = observedAtMs;
190 if (!wasRecent) {
191 requestReport();
192 }
193 };
194 const passiveListenerOptions = { passive: true } as const;
195
196 const report = Effect.gen(function* () {
197 const observedAtMs = yield* Clock.currentTimeMillis;
198 const entries = yield* SubscriptionRef.get(registry.entries);
199 yield* Effect.forEach(
200 entries.keys(),
201 (environmentId) =>
202 registry
203 .run(
204 environmentId,
205 request(
206 WS_METHODS.serverReportClientActivity,
207 createActivityReport(environmentId, lastInteractionAtMs, observedAtMs),
208 ),
209 )
210 // Not `ignore`: tearing down an environment mid-request interrupts its report,
211 // and an interrupt leaking out here would end the reporting loop for good.
212 // The timeout covers a live connection that never answers, which would stall it.
213 .pipe(Effect.timeout(`${REPORT_TIMEOUT_MS} millis`), Effect.ignoreCause),
214 { concurrency: "unbounded", discard: true },
215 );
216 }).pipe(Effect.withSpan("web.backgroundActivity.report"));
217
218 yield* Effect.acquireRelease(
219 Effect.sync(() => {
220 retainedScopeListeners.add(requestReport);
221 document.addEventListener("visibilitychange", requestReport);
222 window.addEventListener("focus", requestReport);
223 window.addEventListener("blur", requestReport);
224 window.addEventListener("online", requestReport);
225 window.addEventListener("pointermove", recordInteraction);
226 window.addEventListener("keydown", recordInteraction);
227 window.addEventListener("wheel", recordInteraction, passiveListenerOptions);
228 window.addEventListener("touchstart", recordInteraction, passiveListenerOptions);
229 }),
230 () =>
231 Effect.sync(() => {
232 retainedScopeListeners.delete(requestReport);
233 document.removeEventListener("visibilitychange", requestReport);
234 window.removeEventListener("focus", requestReport);
235 window.removeEventListener("blur", requestReport);
236 window.removeEventListener("online", requestReport);
237 window.removeEventListener("pointermove", recordInteraction);
238 window.removeEventListener("keydown", recordInteraction);
239 window.removeEventListener("wheel", recordInteraction);
240 window.removeEventListener("touchstart", recordInteraction);
241 }),
242 );
243
244 yield* SubscriptionRef.changes(registry.entries).pipe(
245 Stream.runForEach(() => Effect.sync(requestReport)),
246 Effect.forkScoped,
247 );
248 yield* Stream.fromQueue(reportRequests).pipe(
249 Stream.debounce("250 millis"),
250 Stream.runForEach(() => report),
251 Effect.forkScoped,
252 );
253 yield* Effect.sync(requestReport).pipe(
254 Effect.repeat(Schedule.spaced(`${REPORT_INTERVAL_MS} millis`)),
255 Effect.forkScoped,
256 );
257 }),
258);