1import {
2 type AuthSessionId,
3 type BackgroundPolicySnapshot,
4 type BackgroundScope,
5 type ClientActivityLease,
6 type ClientActivityReportInput,
7 type HostPowerSnapshot,
8 type RpcClientId,
9} from "@t3tools/contracts";
10import {
11 getBackgroundActivityPresetSettings,
12 resolveServerBackgroundActivitySettings,
13 type ResolvedBackgroundActivitySettings,
14} from "@t3tools/shared/backgroundActivitySettings";
15import * as DateTime from "effect/DateTime";
16import * as Context from "effect/Context";
17import * as Effect from "effect/Effect";
18import * as Layer from "effect/Layer";
19import * as PubSub from "effect/PubSub";
20import * as Ref from "effect/Ref";
21import * as Scope from "effect/Scope";
22import * as Semaphore from "effect/Semaphore";
23import * as Stream from "effect/Stream";
24
25import * as ServerSettings from "../serverSettings.ts";
26import { subscribeBeforeSnapshot } from "../utils/subscribeBeforeSnapshot.ts";
27import * as HostPowerMonitor from "./HostPowerMonitor.ts";
28
29export class BackgroundPolicy extends Context.Service<
30 BackgroundPolicy,
31 {
32 readonly reportClientActivity: (
33 sessionId: AuthSessionId,
34 rpcClientId: RpcClientId,
35 input: ClientActivityReportInput,
36 ) => Effect.Effect<void>;
37 readonly removeRpcClient: (
38 sessionId: AuthSessionId,
39 rpcClientId: RpcClientId,
40 ) => Effect.Effect<void>;
41 readonly reportHostPowerState: (snapshot: HostPowerSnapshot) => Effect.Effect<void>;
42 readonly snapshot: Effect.Effect<BackgroundPolicySnapshot>;
43 readonly streamChanges: Stream.Stream<BackgroundPolicySnapshot>;
44 readonly subscribe: Effect.Effect<
45 {
46 readonly latest: BackgroundPolicySnapshot;
47 readonly changes: Stream.Stream<BackgroundPolicySnapshot>;
48 },
49 never,
50 Scope.Scope
51 >;
52 readonly hasDemand: (scope: BackgroundScope) => Effect.Effect<boolean>;
53 readonly shouldRunScopeWork: (scope: BackgroundScope) => Effect.Effect<boolean>;
54 readonly shouldRunOpportunisticWork: Effect.Effect<boolean>;
55 }
56>()("t3/background/BackgroundPolicy") {}
57
58const DEFAULT_LEASE_TTL_MS = 45_000;
59const MAX_LEASE_TTL_MS = 120_000;
60export const MAX_CLIENT_ACTIVITY_LEASES_PER_RPC_CLIENT = 16;
61
62function scopeKey(scope: BackgroundScope): string {
63 switch (scope.type) {
64 case "server-config":
65 case "diagnostics":
66 return scope.type;
67 case "provider-status":
68 return scope.instanceId ? `${scope.type}:${scope.instanceId}` : scope.type;
69 case "vcs-status":
70 case "git-refs":
71 return `${scope.type}:${scope.cwd}`;
72 case "thread":
73 return `${scope.type}:${scope.threadId}`;
74 }
75}
76
77function isLeaseActive(lease: ClientActivityLease, now: DateTime.Utc): boolean {
78 return DateTime.isGreaterThan(lease.expiresAt, now);
79}
80
81function leaseKey(lease: Pick<ClientActivityLease, "sessionId" | "rpcClientId" | "clientId">) {
82 return JSON.stringify([lease.sessionId, lease.rpcClientId, lease.clientId]);
83}
84
85function upsertClientActivityLease(
86 leases: ReadonlyMap<string, ClientActivityLease>,
87 lease: ClientActivityLease,
88 now: DateTime.Utc,
89): Map<string, ClientActivityLease> {
90 const next = new Map(leases);
91 for (const [key, current] of next) {
92 if (!isLeaseActive(current, now)) {
93 next.delete(key);
94 }
95 }
96
97 const key = leaseKey(lease);
98 if (!next.has(key)) {
99 let connectionLeaseCount = 0;
100 let oldestConnectionLease:
101 | {
102 readonly key: string;
103 readonly updatedAtMs: number;
104 }
105 | undefined;
106 for (const [currentKey, current] of next) {
107 if (current.sessionId !== lease.sessionId || current.rpcClientId !== lease.rpcClientId) {
108 continue;
109 }
110 connectionLeaseCount += 1;
111 const updatedAtMs = DateTime.toEpochMillis(current.updatedAt);
112 if (oldestConnectionLease === undefined || updatedAtMs < oldestConnectionLease.updatedAtMs) {
113 oldestConnectionLease = { key: currentKey, updatedAtMs };
114 }
115 }
116 if (
117 connectionLeaseCount >= MAX_CLIENT_ACTIVITY_LEASES_PER_RPC_CLIENT &&
118 oldestConnectionLease !== undefined
119 ) {
120 next.delete(oldestConnectionLease.key);
121 }
122 }
123
124 next.set(key, lease);
125 return next;
126}
127
128function isForegroundLease(lease: ClientActivityLease, now: DateTime.Utc): boolean {
129 return isLeaseActive(lease, now) && lease.visible && (lease.focused || lease.recentlyInteracted);
130}
131
132function leaseHasScope(lease: ClientActivityLease, scope: BackgroundScope): boolean {
133 const key = scopeKey(scope);
134 return lease.scopes.some((leaseScope) => scopeKey(leaseScope) === key);
135}
136
137function hasThermalPressure(hostPower: HostPowerSnapshot): boolean {
138 return hostPower.thermalState === "serious" || hostPower.thermalState === "critical";
139}
140
141function isHostConstrained(
142 hostPower: HostPowerSnapshot,
143 settings: ResolvedBackgroundActivitySettings,
144): boolean {
145 if (hostPower.stale) return false;
146 if (
147 hostPower.suspended ||
148 (settings.pauseWhenHostLocked && hostPower.locked === "true") ||
149 hasThermalPressure(hostPower)
150 ) {
151 return true;
152 }
153 if (settings.pauseWhenHostLowPower && hostPower.lowPowerMode === "true") return true;
154 return settings.pauseWhenOnBattery && hostPower.onBattery === "true";
155}
156
157function isClientConstrained(
158 lease: ClientActivityLease,
159 settings: ResolvedBackgroundActivitySettings,
160): boolean {
161 if (settings.pauseWhenClientLowPower && lease.lowPowerMode === "true") return true;
162 return settings.pauseWhenOnBattery && lease.batteryState === "unplugged";
163}
164
165function leaseMayRunScopedWork(
166 lease: ClientActivityLease,
167 scope: BackgroundScope,
168 now: DateTime.Utc,
169 settings: ResolvedBackgroundActivitySettings,
170): boolean {
171 const activeWithScope = isLeaseActive(lease, now) && leaseHasScope(lease, scope);
172 if (!activeWithScope || isClientConstrained(lease, settings)) {
173 return false;
174 }
175 if (settings.profile === "performance") {
176 return true;
177 }
178 return isForegroundLease(lease, now);
179}
180
181function computeSnapshot(input: {
182 readonly hostPower: HostPowerSnapshot;
183 readonly leases: ReadonlyMap<string, ClientActivityLease>;
184 readonly now: DateTime.Utc;
185 readonly settings: ResolvedBackgroundActivitySettings;
186 readonly updatedAt: DateTime.Utc;
187}): BackgroundPolicySnapshot {
188 const activeLeases = [...input.leases.values()].filter((lease) =>
189 isLeaseActive(lease, input.now),
190 );
191 const foregroundLeases = activeLeases.filter((lease) => isForegroundLease(lease, input.now));
192 const activeScopeKeys = new Set<string>();
193 for (const lease of activeLeases) {
194 for (const scope of lease.scopes) {
195 activeScopeKeys.add(scopeKey(scope));
196 }
197 }
198
199 return {
200 hostPower: input.hostPower,
201 leases: activeLeases,
202 activeForegroundLeaseCount: foregroundLeases.length,
203 activeScopeKeys: [...activeScopeKeys].toSorted(),
204 shouldRunOpportunisticWork:
205 foregroundLeases.some((lease) => !isClientConstrained(lease, input.settings)) &&
206 !isHostConstrained(input.hostPower, input.settings),
207 updatedAt: input.updatedAt,
208 };
209}
210
211/** @public Service construction is part of the canonical Effect module API. */
212export const make = Effect.fn("background.policy.make")(function* () {
213 const hostPowerMonitor = yield* HostPowerMonitor.HostPowerMonitor;
214 const serverSettings = yield* ServerSettings.ServerSettingsService;
215 const leasesRef = yield* Ref.make(new Map<string, ClientActivityLease>());
216 const changes = yield* PubSub.sliding<BackgroundPolicySnapshot>(1);
217 const publishMutex = yield* Semaphore.make(1);
218
219 const backgroundActivitySettings = serverSettings.getSettings.pipe(
220 Effect.map(resolveServerBackgroundActivitySettings),
221 Effect.orElseSucceed(() => getBackgroundActivityPresetSettings("balanced")),
222 );
223
224 const snapshot = Effect.gen(function* () {
225 const [hostPower, leases, now, settings] = yield* Effect.all([
226 hostPowerMonitor.snapshot,
227 Ref.get(leasesRef),
228 DateTime.now,
229 backgroundActivitySettings,
230 ]);
231 return computeSnapshot({ hostPower, leases, now, settings, updatedAt: now });
232 });
233
234 const publishSnapshotUnlocked = snapshot.pipe(
235 Effect.flatMap((next) => PubSub.publish(changes, next)),
236 );
237 const publishSnapshot = publishMutex.withPermits(1)(publishSnapshotUnlocked);
238
239 const reportClientActivity: BackgroundPolicy["Service"]["reportClientActivity"] = (
240 sessionId,
241 rpcClientId,
242 input,
243 ) =>
244 publishMutex.withPermits(1)(
245 Effect.gen(function* () {
246 const ttlMs = Math.min(
247 Math.max(input.ttlMs ?? DEFAULT_LEASE_TTL_MS, 1_000),
248 MAX_LEASE_TTL_MS,
249 );
250 const now = yield* DateTime.now;
251 const expiresAt = DateTime.add(now, { milliseconds: ttlMs });
252 const lease: ClientActivityLease = {
253 sessionId,
254 rpcClientId,
255 clientId: input.clientId,
256 clientKind: input.clientKind,
257 visible: input.visible,
258 focused: input.focused,
259 recentlyInteracted: input.recentlyInteracted,
260 ...(input.appState !== undefined ? { appState: input.appState } : {}),
261 ...(input.lowPowerMode !== undefined ? { lowPowerMode: input.lowPowerMode } : {}),
262 ...(input.batteryState !== undefined ? { batteryState: input.batteryState } : {}),
263 ...(input.networkType !== undefined ? { networkType: input.networkType } : {}),
264 scopes: input.scopes,
265 updatedAt: now,
266 expiresAt,
267 };
268 yield* Ref.update(leasesRef, (leases) => upsertClientActivityLease(leases, lease, now));
269 yield* publishSnapshotUnlocked;
270 }),
271 );
272
273 const removeRpcClient: BackgroundPolicy["Service"]["removeRpcClient"] = (
274 sessionId,
275 rpcClientId,
276 ) =>
277 publishMutex.withPermits(1)(
278 Ref.update(leasesRef, (leases) => {
279 const next = new Map(leases);
280 for (const [key, lease] of next) {
281 if (lease.sessionId === sessionId && lease.rpcClientId === rpcClientId) {
282 next.delete(key);
283 }
284 }
285 return next;
286 }).pipe(Effect.andThen(publishSnapshotUnlocked), Effect.asVoid),
287 );
288
289 const hasDemand: BackgroundPolicy["Service"]["hasDemand"] = (scope) =>
290 Effect.map(snapshot, (current) => current.activeScopeKeys.includes(scopeKey(scope)));
291
292 const shouldRunScopeWork: BackgroundPolicy["Service"]["shouldRunScopeWork"] = (scope) =>
293 Effect.gen(function* () {
294 const [current, settings] = yield* Effect.all([snapshot, backgroundActivitySettings]);
295 if (isHostConstrained(current.hostPower, settings)) {
296 return false;
297 }
298 return current.leases.some((lease) =>
299 leaseMayRunScopedWork(lease, scope, current.updatedAt, settings),
300 );
301 });
302
303 const shouldRunOpportunisticWork = Effect.map(
304 snapshot,
305 (current) => current.shouldRunOpportunisticWork,
306 );
307
308 yield* Stream.runForEach(hostPowerMonitor.streamChanges, () => publishSnapshot).pipe(
309 Effect.forkScoped,
310 );
311 yield* Stream.runForEach(serverSettings.streamChanges, () => publishSnapshot).pipe(
312 Effect.forkScoped,
313 );
314
315 yield* Effect.forever(
316 Effect.sleep("15 seconds").pipe(
317 Effect.andThen(
318 publishMutex.withPermits(1)(
319 Effect.gen(function* () {
320 const now = yield* DateTime.now;
321 yield* Ref.update(leasesRef, (leases) => {
322 const next = new Map(leases);
323 for (const [key, lease] of next) {
324 if (!isLeaseActive(lease, now)) {
325 next.delete(key);
326 }
327 }
328 return next;
329 });
330 yield* publishSnapshotUnlocked;
331 }),
332 ),
333 ),
334 ),
335 ).pipe(Effect.forkScoped);
336
337 return BackgroundPolicy.of({
338 reportClientActivity,
339 removeRpcClient,
340 reportHostPowerState: hostPowerMonitor.report,
341 snapshot,
342 streamChanges: Stream.fromPubSub(changes),
343 subscribe: subscribeBeforeSnapshot(changes, snapshot, publishMutex),
344 hasDemand,
345 shouldRunScopeWork,
346 shouldRunOpportunisticWork,
347 });
348});
349
350export const layer = Layer.effect(BackgroundPolicy, make());