import type { ServerProvider } from "@t3tools/contracts";import { Duration, Effect, PubSub, Ref, Scope, Stream } from "effect";import * as Semaphore from "effect/Semaphore"; import type { ServerProviderShape } from "./Services/ServerProvider"; export function makeManagedServerProvider<Settings>(input: { readonly getSettings: Effect.Effect<Settings>; readonly streamSettings: Stream.Stream<Settings>; readonly haveSettingsChanged: (previous: Settings, next: Settings) => boolean; readonly checkProvider: Effect.Effect<ServerProvider>; readonly refreshInterval?: Duration.Input;}): Effect.Effect<ServerProviderShape, never, Scope.Scope> { return Effect.gen(function* () { const refreshSemaphore = yield* Semaphore.make(1); const changesPubSub = yield* Effect.acquireRelease( PubSub.unbounded<ServerProvider>(), PubSub.shutdown, ); const initialSettings = yield* input.getSettings; const initialSnapshot = yield* input.checkProvider; const snapshotRef = yield* Ref.make(initialSnapshot); const settingsRef = yield* Ref.make(initialSettings); const refreshSnapshot = refreshSemaphore.withPermits(1)( Effect.gen(function* () { const nextSettings = yield* input.getSettings; const nextSnapshot = yield* input.checkProvider; yield* Ref.set(settingsRef, nextSettings); yield* Ref.set(snapshotRef, nextSnapshot); yield* PubSub.publish(changesPubSub, nextSnapshot); return nextSnapshot; }), ); const reconcileSnapshot = (nextSettings: Settings) => Effect.gen(function* () { const previousSettings = yield* Ref.get(settingsRef); if (!input.haveSettingsChanged(previousSettings, nextSettings)) { yield* Ref.set(settingsRef, nextSettings); return yield* Ref.get(snapshotRef); } return yield* refreshSnapshot; }); yield* Stream.runForEach(input.streamSettings, (nextSettings) => Effect.asVoid(reconcileSnapshot(nextSettings)), ).pipe(Effect.forkScoped); yield* Effect.forever( Effect.sleep(input.refreshInterval ?? "60 seconds").pipe( Effect.flatMap(() => refreshSnapshot), Effect.ignoreCause({ log: true }), ), ).pipe(Effect.forkScoped); return { getSnapshot: input.getSettings.pipe( Effect.flatMap(reconcileSnapshot), Effect.tapError(Effect.logError), Effect.orDie, ), refresh: refreshSnapshot.pipe(Effect.tapError(Effect.logError), Effect.orDie), get streamChanges() { return Stream.fromPubSub(changesPubSub); }, } satisfies ServerProviderShape; });}
import type { ServerProvider } from "@t3tools/contracts";
import { Duration, Effect, PubSub, Ref, Scope, Stream } from "effect";
import * as Semaphore from "effect/Semaphore";
import type { ServerProviderShape } from "./Services/ServerProvider";
export function makeManagedServerProvider<Settings>(input: {
readonly getSettings: Effect.Effect<Settings>;
readonly streamSettings: Stream.Stream<Settings>;
readonly haveSettingsChanged: (previous: Settings, next: Settings) => boolean;
readonly checkProvider: Effect.Effect<ServerProvider>;
readonly refreshInterval?: Duration.Input;
}): Effect.Effect<ServerProviderShape, never, Scope.Scope> {
return Effect.gen(function* () {
const refreshSemaphore = yield* Semaphore.make(1);
const changesPubSub = yield* Effect.acquireRelease(
PubSub.unbounded<ServerProvider>(),
PubSub.shutdown,
);
const initialSettings = yield* input.getSettings;
const initialSnapshot = yield* input.checkProvider;
const snapshotRef = yield* Ref.make(initialSnapshot);
const settingsRef = yield* Ref.make(initialSettings);
const refreshSnapshot = refreshSemaphore.withPermits(1)(
Effect.gen(function* () {
const nextSettings = yield* input.getSettings;
const nextSnapshot = yield* input.checkProvider;
yield* Ref.set(settingsRef, nextSettings);
yield* Ref.set(snapshotRef, nextSnapshot);
yield* PubSub.publish(changesPubSub, nextSnapshot);
return nextSnapshot;
}),
);
const reconcileSnapshot = (nextSettings: Settings) =>
Effect.gen(function* () {
const previousSettings = yield* Ref.get(settingsRef);
if (!input.haveSettingsChanged(previousSettings, nextSettings)) {
yield* Ref.set(settingsRef, nextSettings);
return yield* Ref.get(snapshotRef);
}
return yield* refreshSnapshot;
});
yield* Stream.runForEach(input.streamSettings, (nextSettings) =>
Effect.asVoid(reconcileSnapshot(nextSettings)),
).pipe(Effect.forkScoped);
yield* Effect.forever(
Effect.sleep(input.refreshInterval ?? "60 seconds").pipe(
Effect.flatMap(() => refreshSnapshot),
Effect.ignoreCause({ log: true }),
),
).pipe(Effect.forkScoped);
return {
getSnapshot: input.getSettings.pipe(
Effect.flatMap(reconcileSnapshot),
Effect.tapError(Effect.logError),
Effect.orDie,
),
refresh: refreshSnapshot.pipe(Effect.tapError(Effect.logError), Effect.orDie),
get streamChanges() {
return Stream.fromPubSub(changesPubSub);
},
} satisfies ServerProviderShape;
});
}