1import type { EnvironmentId, ThreadId } from "@t3tools/contracts";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Ref from "effect/Ref";
6
7import { threadKey } from "./entities.ts";
8
9export type ThreadHistoryLoadEarlierResult =
10 | { readonly _tag: "loaded" }
11 | { readonly _tag: "noop" }
12 | { readonly _tag: "busy" }
13 | { readonly _tag: "error"; readonly message: string };
14
15export type ThreadHistoryHandler = {
16 readonly loadEarlier: (
17 throughEntryId?: string,
18 workAnchorId?: string,
19 ) => Effect.Effect<ThreadHistoryLoadEarlierResult>;
20 readonly reconcilePendingWork: () => Effect.Effect<void>;
21};
22
23export type ThreadHistoryRegistration = {
24 readonly environmentId: EnvironmentId;
25 readonly threadId: ThreadId;
26 readonly handler: ThreadHistoryHandler;
27};
28
29export class ThreadHistoryController extends Context.Service<
30 ThreadHistoryController,
31 {
32 readonly register: (
33 environmentId: EnvironmentId,
34 threadId: ThreadId,
35 handler: ThreadHistoryHandler,
36 ) => Effect.Effect<ThreadHistoryRegistration>;
37 readonly unregister: (registration: ThreadHistoryRegistration) => Effect.Effect<void>;
38 readonly loadEarlier: (
39 environmentId: EnvironmentId,
40 threadId: ThreadId,
41 throughEntryId?: string,
42 workAnchorId?: string,
43 ) => Effect.Effect<ThreadHistoryLoadEarlierResult>;
44 readonly reconcilePendingWork: (
45 environmentId: EnvironmentId,
46 threadId: ThreadId,
47 ) => Effect.Effect<void>;
48 }
49>()("@t3tools/client-runtime/state/threadHistoryController") {}
50
51export const layer: Layer.Layer<ThreadHistoryController> = Layer.effect(
52 ThreadHistoryController,
53 Effect.gen(function* () {
54 const handlers = yield* Ref.make(new Map<string, ThreadHistoryHandler>());
55 return ThreadHistoryController.of({
56 register: Effect.fn("ThreadHistoryController.register")(
57 function (environmentId, threadId, handler) {
58 return Ref.update(handlers, (current) => {
59 const next = new Map(current);
60 next.set(threadKey({ environmentId, threadId }), handler);
61 return next;
62 }).pipe(
63 Effect.as({
64 environmentId,
65 threadId,
66 handler,
67 } satisfies ThreadHistoryRegistration),
68 );
69 },
70 ),
71 unregister: Effect.fn("ThreadHistoryController.unregister")(function (registration) {
72 return Ref.update(handlers, (current) => {
73 const key = threadKey({
74 environmentId: registration.environmentId,
75 threadId: registration.threadId,
76 });
77 // Only remove when this registration still owns the slot. An older
78 // fiber finalizer must not delete a newer handler for the same key.
79 if (current.get(key) !== registration.handler) {
80 return current;
81 }
82 const next = new Map(current);
83 next.delete(key);
84 return next;
85 });
86 }),
87 loadEarlier: Effect.fn("ThreadHistoryController.loadEarlier")(
88 function (environmentId, threadId, throughEntryId, workAnchorId) {
89 return Ref.get(handlers).pipe(
90 Effect.flatMap((current) => {
91 const handler = current.get(threadKey({ environmentId, threadId }));
92 if (handler === undefined) {
93 return Effect.succeed({ _tag: "noop" } satisfies ThreadHistoryLoadEarlierResult);
94 }
95 return handler.loadEarlier(throughEntryId, workAnchorId);
96 }),
97 );
98 },
99 ),
100 reconcilePendingWork: Effect.fn("ThreadHistoryController.reconcilePendingWork")(
101 function (environmentId, threadId) {
102 return Ref.get(handlers).pipe(
103 Effect.flatMap(
104 (current) =>
105 current.get(threadKey({ environmentId, threadId }))?.reconcilePendingWork() ??
106 Effect.void,
107 ),
108 );
109 },
110 ),
111 });
112 }),
113);