authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-04-01 09:03:08-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-04-01 16:03:08+00:00
logc6ff7c0d5b79e7dca8fcb675751e88187862b004
tree705d256bd033934d04a1df4e28e36ee096e38ba5
parenta9e5a84c03fbec2934777a0c718bcabec161863e
signature Signed by PGP key B5690EEEBB952194

[codex] migrate Effect.fn in apps/server/src/provider/makeManagedServerProvider.ts (#1640)

Co-authored-by: codex <codex@users.noreply.github.com>

1 files changed, 56 insertions(+), 54 deletions(-)

apps/server/src/provider/makeManagedServerProvider.ts+56-54
...@@ -5,68 +5,70 @@ import * as Semaphore from "effect/Semaphore";...@@ -5,68 +5,70 @@ import * as Semaphore from "effect/Semaphore";
5import type { ServerProviderShape } from "./Services/ServerProvider";5import type { ServerProviderShape } from "./Services/ServerProvider";
6import { ServerSettingsError } from "@t3tools/contracts";6import { ServerSettingsError } from "@t3tools/contracts";
77
8export function makeManagedServerProvider<Settings>(input: {8export const makeManagedServerProvider = Effect.fn("makeManagedServerProvider")(function* <
9 Settings,
10>(input: {
9 readonly getSettings: Effect.Effect<Settings>;11 readonly getSettings: Effect.Effect<Settings>;
10 readonly streamSettings: Stream.Stream<Settings>;12 readonly streamSettings: Stream.Stream<Settings>;
11 readonly haveSettingsChanged: (previous: Settings, next: Settings) => boolean;13 readonly haveSettingsChanged: (previous: Settings, next: Settings) => boolean;
12 readonly checkProvider: Effect.Effect<ServerProvider, ServerSettingsError>;14 readonly checkProvider: Effect.Effect<ServerProvider, ServerSettingsError>;
13 readonly refreshInterval?: Duration.Input;15 readonly refreshInterval?: Duration.Input;
14}): Effect.Effect<ServerProviderShape, ServerSettingsError, Scope.Scope> {16}): Effect.fn.Return<ServerProviderShape, ServerSettingsError, Scope.Scope> {
15 return Effect.gen(function* () {17 const refreshSemaphore = yield* Semaphore.make(1);
16 const refreshSemaphore = yield* Semaphore.make(1);18 const changesPubSub = yield* Effect.acquireRelease(
17 const changesPubSub = yield* Effect.acquireRelease(19 PubSub.unbounded<ServerProvider>(),
18 PubSub.unbounded<ServerProvider>(),20 PubSub.shutdown,
19 PubSub.shutdown,21 );
20 );22 const initialSettings = yield* input.getSettings;
21 const initialSettings = yield* input.getSettings;23 const initialSnapshot = yield* input.checkProvider;
22 const initialSnapshot = yield* input.checkProvider;24 const snapshotRef = yield* Ref.make(initialSnapshot);
23 const snapshotRef = yield* Ref.make(initialSnapshot);25 const settingsRef = yield* Ref.make(initialSettings);
24 const settingsRef = yield* Ref.make(initialSettings);
2526
26 const applySnapshot = (nextSettings: Settings, options?: { readonly forceRefresh?: boolean }) =>27 const applySnapshotBase = Effect.fn("applySnapshot")(function* (
27 refreshSemaphore.withPermits(1)(28 nextSettings: Settings,
28 Effect.gen(function* () {29 options?: { readonly forceRefresh?: boolean },
29 const forceRefresh = options?.forceRefresh === true;30 ) {
30 const previousSettings = yield* Ref.get(settingsRef);31 const forceRefresh = options?.forceRefresh === true;
31 if (!forceRefresh && !input.haveSettingsChanged(previousSettings, nextSettings)) {32 const previousSettings = yield* Ref.get(settingsRef);
32 yield* Ref.set(settingsRef, nextSettings);33 if (!forceRefresh && !input.haveSettingsChanged(previousSettings, nextSettings)) {
33 return yield* Ref.get(snapshotRef);34 yield* Ref.set(settingsRef, nextSettings);
34 }35 return yield* Ref.get(snapshotRef);
36 }
3537
36 const nextSnapshot = yield* input.checkProvider;38 const nextSnapshot = yield* input.checkProvider;
37 yield* Ref.set(settingsRef, nextSettings);39 yield* Ref.set(settingsRef, nextSettings);
38 yield* Ref.set(snapshotRef, nextSnapshot);40 yield* Ref.set(snapshotRef, nextSnapshot);
39 yield* PubSub.publish(changesPubSub, nextSnapshot);41 yield* PubSub.publish(changesPubSub, nextSnapshot);
40 return nextSnapshot;42 return nextSnapshot;
41 }),43 });
42 );44 const applySnapshot = (nextSettings: Settings, options?: { readonly forceRefresh?: boolean }) =>
45 refreshSemaphore.withPermits(1)(applySnapshotBase(nextSettings, options));
4346
44 const refreshSnapshot = Effect.gen(function* () {47 const refreshSnapshot = Effect.fn("refreshSnapshot")(function* () {
45 const nextSettings = yield* input.getSettings;48 const nextSettings = yield* input.getSettings;
46 return yield* applySnapshot(nextSettings, { forceRefresh: true });49 return yield* applySnapshot(nextSettings, { forceRefresh: true });
47 });50 });
4851
49 yield* Stream.runForEach(input.streamSettings, (nextSettings) =>52 yield* Stream.runForEach(input.streamSettings, (nextSettings) =>
50 Effect.asVoid(applySnapshot(nextSettings)),53 Effect.asVoid(applySnapshot(nextSettings)),
51 ).pipe(Effect.forkScoped);54 ).pipe(Effect.forkScoped);
5255
53 yield* Effect.forever(56 yield* Effect.forever(
54 Effect.sleep(input.refreshInterval ?? "60 seconds").pipe(57 Effect.sleep(input.refreshInterval ?? "60 seconds").pipe(
55 Effect.flatMap(() => refreshSnapshot),58 Effect.flatMap(() => refreshSnapshot()),
56 Effect.ignoreCause({ log: true }),59 Effect.ignoreCause({ log: true }),
57 ),60 ),
58 ).pipe(Effect.forkScoped);61 ).pipe(Effect.forkScoped);
5962
60 return {63 return {
61 getSnapshot: input.getSettings.pipe(64 getSnapshot: input.getSettings.pipe(
62 Effect.flatMap(applySnapshot),65 Effect.flatMap(applySnapshot),
63 Effect.tapError(Effect.logError),66 Effect.tapError(Effect.logError),
64 Effect.orDie,67 Effect.orDie,
65 ),68 ),
66 refresh: refreshSnapshot.pipe(Effect.tapError(Effect.logError), Effect.orDie),69 refresh: refreshSnapshot().pipe(Effect.tapError(Effect.logError), Effect.orDie),
67 get streamChanges() {70 get streamChanges() {
68 return Stream.fromPubSub(changesPubSub);71 return Stream.fromPubSub(changesPubSub);
69 },72 },
70 } satisfies ServerProviderShape;73 } satisfies ServerProviderShape;
71 });74});
72}