1import type { ServerLifecycleStreamEvent } from "@t3tools/contracts";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as PubSub from "effect/PubSub";
6import * as Ref from "effect/Ref";
7import * as Stream from "effect/Stream";
8
9type LifecycleEventInput =
10 | Omit<Extract<ServerLifecycleStreamEvent, { type: "welcome" }>, "sequence">
11 | Omit<Extract<ServerLifecycleStreamEvent, { type: "ready" }>, "sequence">
12 | Omit<Extract<ServerLifecycleStreamEvent, { type: "legacyThreadMigration" }>, "sequence">;
13
14interface SnapshotState {
15 readonly sequence: number;
16 readonly events: ReadonlyArray<ServerLifecycleStreamEvent>;
17}
18
19export class ServerLifecycleEvents extends Context.Service<
20 ServerLifecycleEvents,
21 {
22 readonly publish: (event: LifecycleEventInput) => Effect.Effect<ServerLifecycleStreamEvent>;
23 readonly snapshot: Effect.Effect<SnapshotState>;
24 readonly stream: Stream.Stream<ServerLifecycleStreamEvent>;
25 }
26>()("t3/serverLifecycleEvents") {}
27
28const make = Effect.gen(function* () {
29 const pubsub = yield* PubSub.unbounded<ServerLifecycleStreamEvent>();
30 const state = yield* Ref.make<SnapshotState>({
31 sequence: 0,
32 events: [],
33 });
34
35 return {
36 publish: (event) =>
37 Ref.modify(state, (current) => {
38 const nextSequence = current.sequence + 1;
39 const nextEvent = {
40 ...event,
41 sequence: nextSequence,
42 } satisfies ServerLifecycleStreamEvent;
43 const nextEvents = [
44 nextEvent,
45 ...current.events.filter((entry) => entry.type !== nextEvent.type),
46 ];
47 return [nextEvent, { sequence: nextSequence, events: nextEvents }] as const;
48 }).pipe(Effect.tap((event) => PubSub.publish(pubsub, event))),
49 snapshot: Ref.get(state),
50 get stream() {
51 return Stream.fromPubSub(pubsub);
52 },
53 } satisfies ServerLifecycleEvents["Service"];
54});
55
56export const layer = Layer.effect(ServerLifecycleEvents, make);