1/**
2 * UsageLimitSources — quota from places this environment cannot run turns
3 * on, today a CLIProxyAPI hub pooling several subscription accounts.
4 *
5 * Each configured `settings.usageLimitSources` entry is polled on the
6 * provider health-check interval and on every settings change, then
7 * published as one snapshot per source over `subscribeServerConfig`. A source
8 * that fails keeps its row with `error` set so the user can see it is
9 * configured but unreachable. Nothing is persisted: like provider status,
10 * this is live state that re-derives on boot.
11 *
12 * @module usage/UsageLimitSources
13 */
14import {
15 DEFAULT_PROVIDER_HEALTH_REFRESH_INTERVAL,
16 UsageLimitSourceError,
17 type UsageLimitSourceConsumeResetCreditInput,
18 type ProviderConsumeResetCreditResult,
19 type ServerSettings,
20 type UsageLimitSourceConfig,
21 type UsageLimitSourceId,
22 type UsageLimitSourceSnapshot,
23} from "@t3tools/contracts";
24import { resolveServerBackgroundActivitySettings } from "@t3tools/shared/backgroundActivitySettings";
25import * as Context from "effect/Context";
26import * as DateTime from "effect/DateTime";
27import * as Duration from "effect/Duration";
28import * as Effect from "effect/Effect";
29import * as Equal from "effect/Equal";
30import * as Layer from "effect/Layer";
31import * as PubSub from "effect/PubSub";
32import * as Ref from "effect/Ref";
33import * as Semaphore from "effect/Semaphore";
34import * as Stream from "effect/Stream";
35
36import * as BackgroundPolicy from "../background/BackgroundPolicy.ts";
37import * as Settings from "../serverSettings.ts";
38import { makeCliproxyApi } from "./cliproxyApi.ts";
39
40export class UsageLimitSources extends Context.Service<
41 UsageLimitSources,
42 {
43 readonly current: Effect.Effect<ReadonlyArray<UsageLimitSourceSnapshot>>;
44 /** The current set followed by every change, with repeats dropped. */
45 readonly streamChanges: Stream.Stream<ReadonlyArray<UsageLimitSourceSnapshot>>;
46 /** Re-read every source now. Never fails; failures land on the snapshot. */
47 readonly refresh: Effect.Effect<void>;
48 readonly consumeResetCredit: (
49 input: UsageLimitSourceConsumeResetCreditInput,
50 ) => Effect.Effect<ProviderConsumeResetCreditResult, UsageLimitSourceError>;
51 }
52>()("t3/usage/UsageLimitSources") {}
53
54function sourceLabel(id: string, config: UsageLimitSourceConfig): string {
55 if (config.label) return config.label;
56 const fallback = id.trim() || "Usage limit source";
57 try {
58 return new URL(config.url).host || fallback;
59 } catch {
60 return fallback;
61 }
62}
63
64/** @public Service construction is part of the canonical Effect module API. */
65export const make = Effect.gen(function* () {
66 const api = yield* makeCliproxyApi;
67 const settingsService = yield* Settings.ServerSettingsService;
68 const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy;
69 const stateRef = yield* Ref.make<ReadonlyArray<UsageLimitSourceSnapshot>>([]);
70 const changes = yield* Effect.acquireRelease(
71 PubSub.unbounded<ReadonlyArray<UsageLimitSourceSnapshot>>(),
72 PubSub.shutdown,
73 );
74
75 const readSource = Effect.fn("UsageLimitSources.readSource")(function* (
76 id: UsageLimitSourceId,
77 config: UsageLimitSourceConfig,
78 ) {
79 const checkedAt = DateTime.formatIso(yield* DateTime.now);
80 const base = { id, kind: config.kind, label: sourceLabel(id, config), checkedAt } as const;
81 if (config.managementKey.length === 0) {
82 return { ...base, accounts: [], error: "No management key configured." };
83 }
84 const accounts = yield* api.readAccounts(config).pipe(Effect.result);
85 if (accounts._tag === "Failure") {
86 yield* Effect.logDebug("usage limit source read failed", { id, cause: accounts.failure });
87 return { ...base, accounts: [], error: accounts.failure.detail };
88 }
89 return { ...base, accounts: accounts.success };
90 });
91
92 const publish = (next: ReadonlyArray<UsageLimitSourceSnapshot>) =>
93 Effect.gen(function* () {
94 const changed = yield* Ref.modify(stateRef, (previous) =>
95 Equal.equals(previous, next) ? [false, previous] : [true, next],
96 );
97 if (changed) yield* PubSub.publish(changes, next);
98 });
99
100 // One refresh at a time: a slow hub read started before a settings change
101 // must not publish after the change's own refresh and resurrect a removed
102 // source. Callers queue behind the in-flight run and see current settings.
103 const refreshLock = yield* Semaphore.make(1);
104 const refresh = Effect.gen(function* () {
105 const settings = yield* settingsService.getSettings.pipe(
106 Effect.orElseSucceed((): ServerSettings | null => null),
107 );
108 const entries = Object.entries(settings?.usageLimitSources ?? {}).filter(
109 ([, config]) => config.enabled,
110 );
111 const snapshots = yield* Effect.forEach(
112 entries,
113 ([id, config]) => readSource(id as UsageLimitSourceId, config),
114 { concurrency: 4 },
115 );
116 yield* publish(snapshots);
117 }).pipe(refreshLock.withPermits(1), Effect.ignoreCause({ log: true }));
118
119 // Shares the refresh lock so a stale in-flight read cannot overwrite a redemption.
120 const consumeResetCredit = (input: UsageLimitSourceConsumeResetCreditInput) =>
121 Effect.gen(function* () {
122 const settings = yield* settingsService.getSettings.pipe(
123 Effect.mapError(
124 () => new UsageLimitSourceError({ detail: "Could not read hub settings." }),
125 ),
126 );
127 const config = settings.usageLimitSources[input.sourceId];
128 if (!config?.enabled || !config.managementKey) {
129 return yield* new UsageLimitSourceError({
130 detail: "The usage limit source is missing or disabled.",
131 });
132 }
133 const result = yield* api.consume(config, input.accountId, input.creditId);
134 const snapshot = yield* readSource(input.sourceId, config);
135 const previous = yield* Ref.get(stateRef);
136 yield* publish(previous.map((source) => (source.id === input.sourceId ? snapshot : source)));
137 return result;
138 }).pipe(refreshLock.withPermits(1));
139
140 // Settings edits re-read straight away so a new hub shows up without
141 // waiting for the interval, and a removed one leaves the list.
142 yield* settingsService.streamChanges.pipe(
143 Stream.map((settings) => settings.usageLimitSources),
144 Stream.changes,
145 Stream.runForEach(() => refresh),
146 Effect.forkScoped,
147 );
148
149 const interval = settingsService.getSettings.pipe(
150 Effect.map(
151 (settings) => resolveServerBackgroundActivitySettings(settings).providerHealthRefreshInterval,
152 ),
153 Effect.orElseSucceed(() => DEFAULT_PROVIDER_HEALTH_REFRESH_INTERVAL),
154 );
155 yield* Effect.forever(
156 interval.pipe(
157 Effect.flatMap((wait) =>
158 Effect.sleep(Duration.toMillis(Duration.fromInputUnsafe(wait)) <= 0 ? "60 seconds" : wait),
159 ),
160 Effect.andThen(backgroundPolicy.shouldRunScopeWork({ type: "provider-status" })),
161 Effect.flatMap((shouldRun) => (shouldRun ? refresh : Effect.void)),
162 Effect.ignoreCause({ log: true }),
163 ),
164 ).pipe(Effect.forkScoped);
165
166 yield* refresh.pipe(Effect.forkScoped);
167
168 return {
169 current: Ref.get(stateRef),
170 consumeResetCredit,
171 refresh,
172 get streamChanges() {
173 return Stream.unwrap(
174 Effect.gen(function* () {
175 const subscription = yield* PubSub.subscribe(changes);
176 const snapshot = yield* Ref.get(stateRef);
177 return Stream.concat(Stream.make(snapshot), Stream.fromSubscription(subscription)).pipe(
178 Stream.changes,
179 );
180 }),
181 );
182 },
183 } satisfies UsageLimitSources["Service"];
184});
185
186export const layer = Layer.effect(UsageLimitSources, make);