1import type { RelayAgentActivityState } from "@t3tools/contracts/relay";
2import { RelayAgentActivityState as RelayAgentActivityStateSchema } from "@t3tools/contracts/relay";
3import * as Context from "effect/Context";
4import * as DateTime from "effect/DateTime";
5import * as Effect from "effect/Effect";
6import * as Function from "effect/Function";
7import * as Layer from "effect/Layer";
8import * as Option from "effect/Option";
9import * as Schema from "effect/Schema";
10import { and, desc, eq, isNull, lt, sql } from "drizzle-orm";
11
12import * as RelayDb from "../db.ts";
13import { relayAgentActivityRows, relayEnvironmentLinks } from "../persistence/schema.ts";
14
15export class AgentActivityRowUpsertPersistenceError extends Schema.TaggedError<AgentActivityRowUpsertPersistenceError>()(
16 "AgentActivityRowUpsertPersistenceError",
17 {
18 environmentId: Schema.String,
19 threadId: Schema.String,
20 cause: Schema.Defect(),
21 },
22) {
23 override get message(): string {
24 return `Failed to persist agent activity state for environment ${this.environmentId}, thread ${this.threadId}.`;
25 }
26}
27
28export class AgentActivityRowDeletePersistenceError extends Schema.TaggedError<AgentActivityRowDeletePersistenceError>()(
29 "AgentActivityRowDeletePersistenceError",
30 {
31 environmentId: Schema.String,
32 threadId: Schema.String,
33 cause: Schema.Defect(),
34 },
35) {
36 override get message(): string {
37 return `Failed to delete agent activity state for environment ${this.environmentId}, thread ${this.threadId}.`;
38 }
39}
40
41export class AgentActivityRowPruneTerminalPersistenceError extends Schema.TaggedError<AgentActivityRowPruneTerminalPersistenceError>()(
42 "AgentActivityRowPruneTerminalPersistenceError",
43 {
44 updatedBefore: Schema.String,
45 cause: Schema.Defect(),
46 },
47) {
48 override get message(): string {
49 return `Failed to prune terminal agent activity rows updated before ${this.updatedBefore}.`;
50 }
51}
52
53export class AgentActivityRowListPersistenceError extends Schema.TaggedError<AgentActivityRowListPersistenceError>()(
54 "AgentActivityRowListPersistenceError",
55 {
56 userId: Schema.String,
57 cause: Schema.Defect(),
58 },
59) {
60 override get message(): string {
61 return `Failed to list agent activity state for user ${this.userId}.`;
62 }
63}
64
65export class AgentActivityRows extends Context.Service<
66 AgentActivityRows,
67 {
68 readonly upsert: (input: {
69 readonly environmentPublicKey: string;
70 readonly state: RelayAgentActivityState;
71 }) => Effect.Effect<void, AgentActivityRowUpsertPersistenceError>;
72 readonly pruneTerminal: (input: {
73 readonly updatedBefore: string;
74 }) => Effect.Effect<void, AgentActivityRowPruneTerminalPersistenceError>;
75 readonly remove: (input: {
76 readonly environmentId: string;
77 readonly environmentPublicKey: string;
78 readonly threadId: string;
79 }) => Effect.Effect<void, AgentActivityRowDeletePersistenceError>;
80 readonly listForUser: (input: {
81 readonly userId: string;
82 }) => Effect.Effect<
83 ReadonlyArray<RelayAgentActivityState>,
84 AgentActivityRowListPersistenceError
85 >;
86 readonly getForUserThread: (input: {
87 readonly userId: string;
88 readonly environmentId: string;
89 readonly threadId: string;
90 }) => Effect.Effect<RelayAgentActivityState | null, AgentActivityRowListPersistenceError>;
91 }
92>()("t3code-relay/agentActivity/AgentActivityRows") {}
93
94const decodeJsonString = Schema.decodeEffect(Schema.fromJsonString(Schema.Unknown));
95
96const encodeRelayAgentActivityStateJson = Schema.encodeEffect(
97 Schema.fromJsonString(RelayAgentActivityStateSchema),
98);
99
100const decodeRelayAgentActivityState = Schema.decodeUnknownOption(RelayAgentActivityStateSchema);
101
102export const make = Effect.gen(function* () {
103 const db = yield* RelayDb.RelayDb;
104
105 return AgentActivityRows.of({
106 upsert: Effect.fn("relay.agent_activity_rows.upsert")(function* (input) {
107 yield* Effect.annotateCurrentSpan({
108 "relay.environment_id": input.state.environmentId,
109 "relay.thread_id": input.state.threadId,
110 });
111 const now = yield* DateTime.now;
112 const stateJson = yield* encodeRelayAgentActivityStateJson(input.state).pipe(
113 Effect.flatMap(decodeJsonString),
114 Effect.map(Function.cast<unknown, RelayAgentActivityState>),
115 Effect.mapError(
116 (cause) =>
117 new AgentActivityRowUpsertPersistenceError({
118 environmentId: input.state.environmentId,
119 threadId: input.state.threadId,
120 cause,
121 }),
122 ),
123 );
124 yield* db
125 .insert(relayAgentActivityRows)
126 .values({
127 environmentId: input.state.environmentId,
128 environmentPublicKey: input.environmentPublicKey,
129 threadId: input.state.threadId,
130 stateJson,
131 updatedAt: input.state.updatedAt,
132 createdAt: DateTime.formatIso(now),
133 })
134 .onConflictDoUpdate({
135 target: [
136 relayAgentActivityRows.environmentId,
137 relayAgentActivityRows.environmentPublicKey,
138 relayAgentActivityRows.threadId,
139 ],
140 set: {
141 stateJson,
142 updatedAt: input.state.updatedAt,
143 },
144 })
145 .pipe(
146 Effect.mapError(
147 (cause) =>
148 new AgentActivityRowUpsertPersistenceError({
149 environmentId: input.state.environmentId,
150 threadId: input.state.threadId,
151 cause,
152 }),
153 ),
154 );
155 }),
156
157 remove: Effect.fn("relay.agent_activity_rows.remove")(function* (input) {
158 yield* Effect.annotateCurrentSpan({
159 "relay.environment_id": input.environmentId,
160 "relay.thread_id": input.threadId,
161 });
162 yield* db
163 .delete(relayAgentActivityRows)
164 .where(
165 and(
166 eq(relayAgentActivityRows.environmentId, input.environmentId),
167 eq(relayAgentActivityRows.environmentPublicKey, input.environmentPublicKey),
168 eq(relayAgentActivityRows.threadId, input.threadId),
169 ),
170 )
171 .pipe(
172 Effect.mapError(
173 (cause) =>
174 new AgentActivityRowDeletePersistenceError({
175 environmentId: input.environmentId,
176 threadId: input.threadId,
177 cause,
178 }),
179 ),
180 );
181 }),
182
183 pruneTerminal: Effect.fn("relay.agent_activity_rows.prune_terminal")(function* (input) {
184 yield* Effect.annotateCurrentSpan({
185 "relay.agent_activity_prune.before": input.updatedBefore,
186 });
187 yield* db
188 .delete(relayAgentActivityRows)
189 .where(
190 and(
191 sql`${relayAgentActivityRows.stateJson} ->> 'phase' IN ('completed', 'failed')`,
192 lt(relayAgentActivityRows.updatedAt, input.updatedBefore),
193 ),
194 )
195 .pipe(
196 Effect.mapError(
197 (cause) =>
198 new AgentActivityRowPruneTerminalPersistenceError({
199 updatedBefore: input.updatedBefore,
200 cause,
201 }),
202 ),
203 );
204 }),
205
206 listForUser: Effect.fn("relay.agent_activity_rows.list_for_user")(function* (input) {
207 return yield* db
208 .select({ stateJson: relayAgentActivityRows.stateJson })
209 .from(relayAgentActivityRows)
210 .innerJoin(
211 relayEnvironmentLinks,
212 and(
213 eq(relayEnvironmentLinks.environmentId, relayAgentActivityRows.environmentId),
214 eq(
215 relayEnvironmentLinks.environmentPublicKey,
216 relayAgentActivityRows.environmentPublicKey,
217 ),
218 ),
219 )
220 .where(
221 and(
222 eq(relayEnvironmentLinks.userId, input.userId),
223 isNull(relayEnvironmentLinks.revokedAt),
224 eq(relayEnvironmentLinks.liveActivitiesEnabled, true),
225 ),
226 )
227 .orderBy(desc(relayAgentActivityRows.updatedAt))
228 .pipe(
229 Effect.map((rows) =>
230 rows.flatMap((row) => Option.toArray(decodeRelayAgentActivityState(row.stateJson))),
231 ),
232 Effect.mapError(
233 (cause) =>
234 new AgentActivityRowListPersistenceError({
235 userId: input.userId,
236 cause,
237 }),
238 ),
239 );
240 }),
241
242 getForUserThread: Effect.fn("relay.agent_activity_rows.get_for_user_thread")(function* (input) {
243 return yield* db
244 .select({ stateJson: relayAgentActivityRows.stateJson })
245 .from(relayAgentActivityRows)
246 .innerJoin(
247 relayEnvironmentLinks,
248 and(
249 eq(relayEnvironmentLinks.environmentId, relayAgentActivityRows.environmentId),
250 eq(
251 relayEnvironmentLinks.environmentPublicKey,
252 relayAgentActivityRows.environmentPublicKey,
253 ),
254 ),
255 )
256 .where(
257 and(
258 eq(relayEnvironmentLinks.userId, input.userId),
259 isNull(relayEnvironmentLinks.revokedAt),
260 eq(relayAgentActivityRows.environmentId, input.environmentId),
261 eq(relayAgentActivityRows.threadId, input.threadId),
262 ),
263 )
264 .orderBy(desc(relayAgentActivityRows.updatedAt))
265 .pipe(
266 Effect.map((rows) => {
267 for (const row of rows) {
268 const decoded = decodeRelayAgentActivityState(row.stateJson);
269 if (Option.isSome(decoded)) {
270 return decoded.value;
271 }
272 }
273 return null;
274 }),
275 Effect.mapError(
276 (cause) =>
277 new AgentActivityRowListPersistenceError({
278 userId: input.userId,
279 cause,
280 }),
281 ),
282 );
283 }),
284 });
285});
286
287export const layer = Layer.effect(AgentActivityRows, make);