1import {
2 AuthAdministrativeScopes,
3 AuthArchiveReadScope,
4 AuthOrchestrationOperateScope,
5 AuthOrchestrationReadScope,
6 AuthSessionId,
7 CommandId,
8 HubError,
9 HubInboxEnvironment,
10 HubOrganizationCommand,
11 HUB_THREAD_ORGANIZATION_FIELDS,
12 ORCHESTRATION_PROTOCOL_HEADER,
13 ORCHESTRATION_PROTOCOL_VERSION_TEXT,
14 OrchestrationV2ShellSnapshot,
15 OrchestrationV2ThreadShell,
16 type EnvironmentId,
17 type HubInbox as HubInboxResult,
18 type HubInboxCommand,
19 type ThreadId,
20} from "@t3tools/contracts";
21import * as Clock from "effect/Clock";
22import * as Context from "effect/Context";
23import * as DateTime from "effect/DateTime";
24import * as Effect from "effect/Effect";
25import * as Layer from "effect/Layer";
26import * as Redacted from "effect/Redacted";
27import * as Schedule from "effect/Schedule";
28import * as Schema from "effect/Schema";
29import * as Semaphore from "effect/Semaphore";
30import * as SqlClient from "effect/sql/SqlClient";
31import * as Struct from "effect/Struct";
32import {
33 FetchHttpClient,
34 HttpBody,
35 HttpClient,
36 HttpClientRequest,
37 HttpClientResponse,
38} from "effect/http";
39import type { AuthenticatedSession } from "../auth/EnvironmentAuth.ts";
40import * as Hub from "./Hub.ts";
41import * as ArchiveCatalog from "../archive/ArchiveCatalog.ts";
42import { ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION } from "../orchestration-v2/ProjectionStore.ts";
43
44const Organization = Schema.Struct(
45 Struct.pick(OrchestrationV2ThreadShell.fields, HUB_THREAD_ORGANIZATION_FIELDS),
46);
47type Organization = typeof Organization.Type;
48const jsonOrganization = Schema.fromJsonString(Schema.toCodecJson(Organization));
49const jsonSnapshot = Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ShellSnapshot));
50const SourceActivity = Schema.Struct({
51 userMessageAt: Schema.NullOr(Schema.Number),
52 completedAt: Schema.NullOr(Schema.Number),
53 pendingRequestId: Schema.NullOr(Schema.String),
54});
55const jsonSourceActivity = Schema.fromJsonString(SourceActivity);
56const encodeCommand = Schema.encodeEffect(Schema.toCodecJson(HubOrganizationCommand));
57const isHubError = Schema.is(HubError);
58const sourceActivity = (thread: OrchestrationV2ThreadShell): typeof SourceActivity.Type => {
59 const authoredAt = thread.latestUserAuthoredMessageAt ?? thread.latestUserMessageAt;
60 return {
61 userMessageAt: authoredAt == null ? null : DateTime.toEpochMillis(authoredAt),
62 completedAt:
63 thread.latestRunCompletedAt == null
64 ? null
65 : DateTime.toEpochMillis(thread.latestRunCompletedAt),
66 pendingRequestId: thread.pendingRuntimeRequest?.id ?? null,
67 };
68};
69
70const internalSession: AuthenticatedSession = {
71 sessionId: AuthSessionId.make("hub-inbox"),
72 subject: "hub-inbox",
73 method: "bearer-access-token",
74 scopes: AuthAdministrativeScopes,
75};
76
77export class HubInboxError extends Schema.TaggedError<HubInboxError>()("HubInboxError", {
78 reason: Schema.Literals(["thread-missing", "thread-state", "storage"]),
79 cause: Schema.optional(Schema.Redacted(Schema.Defect())),
80}) {
81 override get message(): string {
82 switch (this.reason) {
83 case "thread-missing":
84 return "This chat is unavailable. Reconnect its computer to refresh the chat list.";
85 case "thread-state":
86 return "The chat changed. Refresh its status and try again.";
87 case "storage":
88 return "The server could not save the chat. Check its storage and try again.";
89 }
90 }
91}
92
93export class HubInbox extends Context.Service<
94 HubInbox,
95 {
96 readonly read: (
97 session: AuthenticatedSession,
98 ) => Effect.Effect<HubInboxResult, HubError | HubInboxError>;
99 readonly organize: (
100 input: HubInboxCommand,
101 session: AuthenticatedSession,
102 ) => Effect.Effect<HubInboxEnvironment, HubError | HubInboxError>;
103 readonly refresh: Effect.Effect<void, HubError | HubInboxError>;
104 }
105>()("t3/environment/HubInbox") {}
106
107const applyCommand = (
108 value: Organization,
109 command: HubOrganizationCommand,
110 now: DateTime.Utc,
111): Organization => {
112 switch (command.type) {
113 case "thread.settle":
114 return {
115 ...value,
116 settledOverride: "settled",
117 settledAt:
118 value.settledOverride === "settled" ? value.settledAt : (command.settledAt ?? now),
119 unsettledAt: null,
120 pinnedAt: null,
121 pinOrderKey: null,
122 activeOrderKey: null,
123 snoozedUntil: null,
124 snoozedAt: null,
125 };
126 case "thread.unsettle":
127 return {
128 ...value,
129 settledOverride: "active",
130 settledAt: null,
131 unsettledAt: value.settledOverride === "active" ? value.unsettledAt : now,
132 };
133 case "thread.snooze":
134 return { ...value, snoozedUntil: DateTime.makeUnsafe(command.snoozedUntil), snoozedAt: now };
135 case "thread.unsnooze":
136 return { ...value, snoozedUntil: null, snoozedAt: null };
137 case "thread.pin":
138 return {
139 ...value,
140 pinnedAt: value.pinnedAt ?? now,
141 pinOrderKey:
142 value.pinnedAt == null ? (command.orderKey ?? value.pinOrderKey) : value.pinOrderKey,
143 settledOverride: value.settledOverride === "settled" ? "active" : value.settledOverride,
144 settledAt: value.settledOverride === "settled" ? null : value.settledAt,
145 snoozedUntil: null,
146 snoozedAt: null,
147 };
148 case "thread.unpin":
149 return { ...value, pinnedAt: null, pinOrderKey: null };
150 case "thread.pin.reorder":
151 return { ...value, pinOrderKey: command.orderKey };
152 case "thread.active.reorder":
153 return { ...value, activeOrderKey: command.orderKey };
154 case "thread.auto-settle.set":
155 return {
156 ...value,
157 autoSettleDisabledAt: command.enabled ? null : (value.autoSettleDisabledAt ?? now),
158 };
159 }
160};
161
162const wakeForSourceActivity = (
163 desired: Organization,
164 baseline: typeof SourceActivity.Type,
165 thread: OrchestrationV2ThreadShell,
166 now: DateTime.Utc,
167): Organization => {
168 const current = sourceActivity(thread);
169 const wakeSnooze =
170 desired.snoozedUntil != null &&
171 (DateTime.toEpochMillis(desired.snoozedUntil) <= DateTime.toEpochMillis(now) ||
172 (current.pendingRequestId !== null &&
173 current.pendingRequestId !== baseline.pendingRequestId) ||
174 (current.completedAt !== null &&
175 current.completedAt !== baseline.completedAt &&
176 (thread.status === "completed" || thread.status === "failed")));
177 const wakeSettlement =
178 desired.settledOverride === "settled" &&
179 current.userMessageAt !== null &&
180 current.userMessageAt !== baseline.userMessageAt;
181 if (!wakeSnooze && !wakeSettlement) return desired;
182 return {
183 ...desired,
184 ...(wakeSnooze ? { snoozedUntil: null, snoozedAt: null } : {}),
185 ...(wakeSettlement
186 ? ({ settledOverride: "active", settledAt: null, unsettledAt: now } as const)
187 : {}),
188 };
189};
190
191const sourceCommands = (
192 thread: OrchestrationV2ThreadShell,
193 desired: Organization,
194 revision: string,
195): ReadonlyArray<HubOrganizationCommand> => {
196 const commands: Array<HubOrganizationCommand> = [];
197 const base = { threadId: thread.id, commandId: CommandId.make(`${revision}:settle`) };
198 if (desired.settledOverride !== thread.settledOverride) {
199 if (desired.settledOverride === "settled")
200 commands.push({
201 ...base,
202 type: "thread.settle",
203 ...(desired.settledAt == null ? {} : { settledAt: desired.settledAt }),
204 });
205 if (desired.settledOverride === "active")
206 commands.push({ ...base, type: "thread.unsettle", reason: "user" });
207 }
208 if ((desired.pinnedAt != null) !== (thread.pinnedAt != null))
209 commands.push(
210 desired.pinnedAt != null
211 ? {
212 ...base,
213 commandId: CommandId.make(`${revision}:pin`),
214 type: "thread.pin",
215 ...(desired.pinOrderKey == null ? {} : { orderKey: desired.pinOrderKey }),
216 }
217 : { ...base, commandId: CommandId.make(`${revision}:pin`), type: "thread.unpin" },
218 );
219 if (
220 desired.pinnedAt != null &&
221 thread.pinnedAt != null &&
222 desired.pinOrderKey != null &&
223 desired.pinOrderKey !== thread.pinOrderKey
224 )
225 commands.push({
226 ...base,
227 commandId: CommandId.make(`${revision}:pin-order`),
228 type: "thread.pin.reorder",
229 orderKey: desired.pinOrderKey,
230 });
231 if (
232 desired.pinnedAt == null &&
233 desired.settledOverride !== "settled" &&
234 desired.activeOrderKey != null &&
235 desired.activeOrderKey !== thread.activeOrderKey
236 )
237 commands.push({
238 ...base,
239 commandId: CommandId.make(`${revision}:active-order`),
240 type: "thread.active.reorder",
241 orderKey: desired.activeOrderKey,
242 });
243 if ((desired.autoSettleDisabledAt != null) !== (thread.autoSettleDisabledAt != null))
244 commands.push({
245 ...base,
246 commandId: CommandId.make(`${revision}:auto-settle`),
247 type: "thread.auto-settle.set",
248 enabled: desired.autoSettleDisabledAt == null,
249 });
250 if (
251 (desired.snoozedUntil == null ? null : DateTime.toEpochMillis(desired.snoozedUntil)) !==
252 (thread.snoozedUntil == null ? null : DateTime.toEpochMillis(thread.snoozedUntil))
253 )
254 commands.push(
255 desired.snoozedUntil == null
256 ? {
257 ...base,
258 commandId: CommandId.make(`${revision}:unsnooze`),
259 type: "thread.unsnooze",
260 reason: "user",
261 }
262 : {
263 ...base,
264 commandId: CommandId.make(`${revision}:snooze`),
265 type: "thread.snooze",
266 snoozedUntil: DateTime.formatIso(desired.snoozedUntil),
267 },
268 );
269 return commands;
270};
271
272const storageError = (cause: unknown) =>
273 new HubInboxError({ reason: "storage", cause: Redacted.make(cause) });
274
275const make = Effect.gen(function* () {
276 const hub = yield* Hub.Hub;
277 const archive = yield* ArchiveCatalog.ArchiveCatalog;
278 const sql = yield* SqlClient.SqlClient;
279 const client = yield* HttpClient.HttpClient;
280 const lock = yield* Semaphore.make(1);
281 const refreshLock = yield* Semaphore.make(1);
282
283 const expireSnoozes = Effect.fnUntraced(function* (id: EnvironmentId) {
284 const now = yield* Clock.currentTimeMillis;
285 const rows = yield* sql<{
286 thread_id: string;
287 command_id: string;
288 organization_json: string;
289 }>`SELECT thread_id, command_id, organization_json FROM hub_inbox_pending WHERE environment_id = ${id}`.pipe(
290 Effect.mapError(storageError),
291 );
292 for (const row of rows) {
293 const value = yield* Schema.decodeEffect(jsonOrganization)(row.organization_json).pipe(
294 Effect.mapError(storageError),
295 );
296 if (value.snoozedUntil == null || DateTime.toEpochMillis(value.snoozedUntil) > now) continue;
297 const json = yield* Schema.encodeEffect(jsonOrganization)({
298 ...value,
299 snoozedUntil: null,
300 snoozedAt: null,
301 }).pipe(Effect.mapError(storageError));
302 yield* sql`UPDATE hub_inbox_pending SET organization_json = ${json}, command_id = ${`${row.command_id}:expired`} WHERE environment_id = ${id} AND thread_id = ${row.thread_id} AND command_id = ${row.command_id}`.pipe(
303 Effect.mapError(storageError),
304 );
305 }
306 });
307
308 const readEnvironment = Effect.fnUntraced(function* (id: EnvironmentId) {
309 yield* expireSnoozes(id);
310 const { rows, pending } = yield* sql
311 .withTransaction(
312 Effect.all({
313 rows: sql<{
314 snapshot_json: string;
315 }>`SELECT snapshot_json FROM hub_inbox WHERE environment_id = ${id}`,
316 pending: sql<{
317 thread_id: ThreadId;
318 organization_json: string;
319 }>`SELECT thread_id, organization_json FROM hub_inbox_pending WHERE environment_id = ${id}`,
320 }),
321 )
322 .pipe(Effect.mapError(storageError));
323 const row = rows[0];
324 if (!row) return undefined;
325 const snapshot = yield* Schema.decodeEffect(jsonSnapshot)(row.snapshot_json).pipe(
326 Effect.mapError(storageError),
327 );
328 const overrides = new Map(
329 yield* Effect.forEach(pending, (value) =>
330 Schema.decodeEffect(jsonOrganization)(value.organization_json).pipe(
331 Effect.map((state) => [value.thread_id, state] as const),
332 Effect.mapError(storageError),
333 ),
334 ),
335 );
336 return HubInboxEnvironment.make({
337 environmentId: id,
338 snapshot: {
339 ...snapshot,
340 threads: snapshot.threads.map((thread) => ({ ...thread, ...overrides.get(thread.id) })),
341 },
342 pendingThreadIds: pending.map((value) => value.thread_id),
343 });
344 });
345
346 const request = Effect.fnUntraced(function* <A>(
347 id: EnvironmentId,
348 path: string,
349 decoder: Schema.Decoder<A>,
350 command?: HubOrganizationCommand,
351 ) {
352 const upstream = yield* hub.connect(id, internalSession);
353 let outgoing = HttpClientRequest.make(command ? "POST" : "GET")(
354 new URL(path, upstream.url).href,
355 ).pipe(
356 HttpClientRequest.bearerToken(upstream.token),
357 HttpClientRequest.setHeader(
358 ORCHESTRATION_PROTOCOL_HEADER,
359 ORCHESTRATION_PROTOCOL_VERSION_TEXT,
360 ),
361 );
362 if (command)
363 outgoing = HttpClientRequest.setBody(
364 outgoing,
365 HttpBody.jsonUnsafe(yield* encodeCommand(command).pipe(Effect.mapError(storageError))),
366 );
367 return yield* Effect.raceFirst(
368 client.execute(outgoing).pipe(
369 Effect.provideService(FetchHttpClient.RequestInit, { redirect: "manual" }),
370 Effect.flatMap(
371 Effect.fnUntraced(function* (response) {
372 if (response.status !== 200)
373 return yield* new HubError({
374 reason:
375 response.status === 401 || response.status === 403 ? "credential" : "upstream",
376 });
377 return yield* HttpClientResponse.schemaBodyJson(decoder)(response);
378 }),
379 ),
380 Effect.timeout("10 seconds"),
381 Effect.mapError((cause) =>
382 isHubError(cause)
383 ? cause
384 : new HubError({ reason: "upstream", cause: Redacted.make(cause) }),
385 ),
386 ),
387 upstream.invalidated.pipe(
388 Effect.andThen(Effect.fail(new HubError({ reason: "credential" }))),
389 ),
390 ).pipe(Effect.scoped);
391 });
392
393 const refreshEnvironment = Effect.fnUntraced(function* (id: EnvironmentId) {
394 yield* expireSnoozes(id);
395 let snapshot = yield* request(id, "api/orchestration/shell", OrchestrationV2ShellSnapshot);
396 const pending = yield* sql<{
397 thread_id: ThreadId;
398 command_id: string;
399 organization_json: string;
400 source_activity_json: string;
401 }>`SELECT thread_id, command_id, organization_json, source_activity_json FROM hub_inbox_pending WHERE environment_id = ${id}`.pipe(
402 Effect.mapError(storageError),
403 );
404 const confirmedRevisions: Array<{ threadId: ThreadId; commandId: string; sequence: number }> =
405 [];
406 for (const row of pending) {
407 const thread = snapshot.threads.find((value) => value.id === row.thread_id);
408 if (!thread || thread.archivedAt !== null || thread.deletedAt !== null) {
409 confirmedRevisions.push({
410 threadId: row.thread_id,
411 commandId: row.command_id,
412 sequence: snapshot.snapshotSequence,
413 });
414 continue;
415 }
416 let desired = yield* Schema.decodeEffect(jsonOrganization)(row.organization_json).pipe(
417 Effect.mapError(storageError),
418 );
419 const baseline = yield* Schema.decodeEffect(jsonSourceActivity)(
420 row.source_activity_json,
421 ).pipe(Effect.mapError(storageError));
422 const updated = wakeForSourceActivity(desired, baseline, thread, yield* DateTime.now);
423 if (updated !== desired) {
424 desired = updated;
425 const json = yield* Schema.encodeEffect(jsonOrganization)(desired).pipe(
426 Effect.mapError(storageError),
427 );
428 const revision = `${row.command_id}:activity`;
429 yield* sql`UPDATE hub_inbox_pending SET organization_json = ${json}, command_id = ${revision} WHERE environment_id = ${id} AND thread_id = ${row.thread_id} AND command_id = ${row.command_id}`.pipe(
430 Effect.mapError(storageError),
431 );
432 row.command_id = revision;
433 row.organization_json = json;
434 }
435 const commands = sourceCommands(thread, desired, row.command_id);
436 let dispatched = true;
437 let sequence = snapshot.snapshotSequence;
438 for (const command of commands) {
439 const current = yield* sql<{
440 command_id: string;
441 }>`SELECT command_id FROM hub_inbox_pending WHERE environment_id = ${id} AND thread_id = ${row.thread_id}`.pipe(
442 Effect.mapError(storageError),
443 );
444 if (current[0]?.command_id !== row.command_id) {
445 dispatched = false;
446 break;
447 }
448 const result = yield* request(
449 id,
450 "api/hub/organization",
451 Schema.Struct({ sequence: Schema.Number }),
452 command,
453 ).pipe(Effect.result);
454 if (result._tag === "Failure") {
455 dispatched = false;
456 break;
457 }
458 sequence = Math.max(sequence, result.success.sequence);
459 }
460 if (!dispatched) continue;
461 confirmedRevisions.push({ threadId: row.thread_id, commandId: row.command_id, sequence });
462 }
463 if (confirmedRevisions.some((value) => value.sequence > snapshot.snapshotSequence))
464 snapshot = yield* request(id, "api/orchestration/shell", OrchestrationV2ShellSnapshot);
465 const json = yield* Schema.encodeEffect(jsonSnapshot)(snapshot).pipe(
466 Effect.mapError(storageError),
467 );
468 yield* sql
469 .withTransaction(
470 Effect.gen(function* () {
471 yield* sql`INSERT INTO hub_inbox(environment_id, snapshot_json) VALUES (${id}, ${json}) ON CONFLICT(environment_id) DO UPDATE SET snapshot_json = excluded.snapshot_json`;
472 for (const revision of confirmedRevisions.filter(
473 (value) => value.sequence <= snapshot.snapshotSequence,
474 )) {
475 const row = pending.find(
476 (value) =>
477 value.thread_id === revision.threadId && value.command_id === revision.commandId,
478 );
479 const thread = snapshot.threads.find((value) => value.id === revision.threadId);
480 if (row && thread && thread.archivedAt === null && thread.deletedAt === null) {
481 const desired = yield* Schema.decodeEffect(jsonOrganization)(
482 row.organization_json,
483 ).pipe(Effect.mapError(storageError));
484 const baseline = yield* Schema.decodeEffect(jsonSourceActivity)(
485 row.source_activity_json,
486 ).pipe(Effect.mapError(storageError));
487 const updated = wakeForSourceActivity(desired, baseline, thread, yield* DateTime.now);
488 if (updated !== desired) {
489 const json = yield* Schema.encodeEffect(jsonOrganization)(updated).pipe(
490 Effect.mapError(storageError),
491 );
492 yield* sql`UPDATE hub_inbox_pending SET organization_json = ${json}, command_id = ${`${revision.commandId}:activity`} WHERE environment_id = ${id} AND thread_id = ${revision.threadId} AND command_id = ${revision.commandId}`;
493 continue;
494 }
495 }
496 yield* sql`DELETE FROM hub_inbox_pending WHERE environment_id = ${id} AND thread_id = ${revision.threadId} AND command_id = ${revision.commandId}`;
497 }
498 }),
499 )
500 .pipe(Effect.mapError(storageError));
501 });
502
503 const refresh = refreshLock.withPermit(
504 Effect.gen(function* () {
505 const directory = yield* hub.directory(internalSession);
506 yield* Effect.forEach(
507 directory.environments.filter((value) => value.scopes.includes(AuthOrchestrationReadScope)),
508 (value) =>
509 refreshEnvironment(value.environmentId).pipe(
510 Effect.catchTags({
511 HubError: () => Effect.void,
512 }),
513 ),
514 { concurrency: 4, discard: true },
515 );
516 }),
517 );
518
519 yield* refresh.pipe(
520 Effect.catchTags({
521 HubError: () => Effect.void,
522 HubInboxError: () => Effect.logWarning("Hub inbox storage is unavailable"),
523 }),
524 Effect.repeat(Schedule.spaced("15 seconds")),
525 Effect.forkScoped,
526 );
527
528 const authorize = Effect.fnUntraced(function* (id: EnvironmentId, session: AuthenticatedSession) {
529 const scopes = yield* hub.authorize(id, session);
530 if (
531 !scopes.includes(AuthOrchestrationReadScope) ||
532 !scopes.includes(AuthOrchestrationOperateScope)
533 )
534 return yield* new HubError({ reason: "credential" });
535 });
536
537 return HubInbox.of({
538 refresh,
539 read: (session) =>
540 Effect.gen(function* () {
541 const directory = yield* hub.directory(session, { refresh: false });
542 const archived = session.scopes.includes(AuthArchiveReadScope)
543 ? (yield* archive.list.pipe(Effect.orElseSucceed(() => ({ entries: [] })))).entries
544 : [];
545 const values = yield* Effect.forEach(
546 directory.environments.filter((value) =>
547 value.scopes.includes(AuthOrchestrationReadScope),
548 ),
549 (value) =>
550 Effect.gen(function* () {
551 const cached = yield* readEnvironment(value.environmentId);
552 const entries = archived.filter(
553 (entry) => entry.environmentId === value.environmentId,
554 );
555 if (!entries.length) return cached;
556 const snapshot = cached?.snapshot ?? {
557 schemaVersion: ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION,
558 snapshotSequence: 0,
559 projects: [],
560 threads: [],
561 archivedThreads: [],
562 };
563 const fresh = entries.filter(
564 (entry) =>
565 entry.throughSequence > snapshot.snapshotSequence ||
566 ![...snapshot.threads, ...snapshot.archivedThreads].some(
567 (thread) => thread.id === entry.thread.id,
568 ),
569 );
570 const replaced = new Set(fresh.map((entry) => entry.thread.id));
571 const projects = new Map(snapshot.projects.map((project) => [project.id, project]));
572 for (const entry of fresh)
573 if (!projects.has(entry.project.id)) projects.set(entry.project.id, entry.project);
574 return HubInboxEnvironment.make({
575 environmentId: value.environmentId,
576 pendingThreadIds: cached?.pendingThreadIds ?? [],
577 snapshot: {
578 ...snapshot,
579 snapshotSequence: Math.max(
580 snapshot.snapshotSequence,
581 ...fresh.map((entry) => entry.throughSequence),
582 ),
583 projects: [...projects.values()],
584 threads: snapshot.threads.filter((thread) => !replaced.has(thread.id)),
585 archivedThreads: [
586 ...snapshot.archivedThreads.filter((thread) => !replaced.has(thread.id)),
587 ...fresh.map((entry) => entry.thread),
588 ],
589 },
590 });
591 }),
592 );
593 return { environments: values.filter((value) => value !== undefined) };
594 }),
595 organize: (input, session) =>
596 Effect.gen(function* () {
597 yield* authorize(input.environmentId, session);
598 const cached = yield* readEnvironment(input.environmentId);
599 if (!cached?.snapshot.threads.some((value) => value.id === input.command.threadId))
600 yield* refreshLock.withPermit(refreshEnvironment(input.environmentId));
601 return yield* lock.withPermit(
602 Effect.gen(function* () {
603 yield* authorize(input.environmentId, session);
604 const entry = yield* readEnvironment(input.environmentId);
605 const receipt =
606 yield* sql`SELECT 1 FROM hub_inbox_commands WHERE environment_id = ${input.environmentId} AND command_id = ${input.command.commandId}`.pipe(
607 Effect.mapError(storageError),
608 );
609 if (entry && receipt.length > 0) return entry;
610 const thread = entry?.snapshot.threads.find(
611 (value) => value.id === input.command.threadId,
612 );
613 if (!entry || !thread || thread.deletedAt !== null)
614 return yield* new HubInboxError({ reason: "thread-missing" });
615 const command = input.command;
616 const now = yield* DateTime.now;
617 if (
618 thread.archivedAt !== null ||
619 (command.type === "thread.pin.reorder" && thread.pinnedAt == null) ||
620 (command.type === "thread.active.reorder" &&
621 (thread.pinnedAt != null || thread.settledOverride === "settled")) ||
622 (command.type === "thread.snooze" &&
623 Date.parse(command.snoozedUntil) <= DateTime.toEpochMillis(now))
624 )
625 return yield* new HubInboxError({ reason: "thread-state" });
626 const desired = applyCommand(
627 Struct.pick(thread, HUB_THREAD_ORGANIZATION_FIELDS),
628 command,
629 now,
630 );
631 const json = yield* Schema.encodeEffect(jsonOrganization)(desired).pipe(
632 Effect.mapError(storageError),
633 );
634 const baseline = yield* Schema.encodeEffect(jsonSourceActivity)(
635 sourceActivity(thread),
636 ).pipe(Effect.mapError(storageError));
637 yield* sql
638 .withTransaction(
639 Effect.gen(function* () {
640 yield* sql`INSERT INTO hub_inbox_pending(environment_id, thread_id, command_id, organization_json, source_activity_json) VALUES (${input.environmentId}, ${command.threadId}, ${command.commandId}, ${json}, ${baseline}) ON CONFLICT(environment_id, thread_id) DO UPDATE SET command_id = excluded.command_id, organization_json = excluded.organization_json, source_activity_json = excluded.source_activity_json`;
641 yield* sql`INSERT INTO hub_inbox_commands(environment_id, command_id) VALUES (${input.environmentId}, ${command.commandId})`;
642 }),
643 )
644 .pipe(Effect.mapError(storageError));
645 return {
646 ...entry,
647 pendingThreadIds: [...new Set([...entry.pendingThreadIds, thread.id])],
648 snapshot: {
649 ...entry.snapshot,
650 threads: entry.snapshot.threads.map((value) =>
651 value.id === thread.id ? { ...value, ...desired } : value,
652 ),
653 },
654 };
655 }),
656 );
657 }),
658 });
659});
660
661export const layer = Layer.effect(HubInbox, make);