1import { ThreadId } from "@t3tools/contracts";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as FileSystem from "effect/FileSystem";
5import * as Layer from "effect/Layer";
6import * as Schema from "effect/Schema";
7
8import { resolveAttachmentRelativePath } from "../attachmentPaths.ts";
9import { parseAttachmentFileExtension, resolveAttachmentPathById } from "../attachmentStore.ts";
10import * as ServerConfig from "../config.ts";
11import * as DeviceService from "../device/DeviceService.ts";
12import * as PreviewManager from "../preview/Manager.ts";
13import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts";
14import * as TerminalManager from "../terminal/Manager.ts";
15
16export class ResourceCleanupError extends Schema.TaggedError<ResourceCleanupError>()(
17 "ResourceCleanupError",
18 {
19 operation: Schema.Literals(["terminal", "preview", "device", "attachment", "scheduled-tasks"]),
20 threadId: Schema.optional(Schema.String),
21 attachmentId: Schema.optional(Schema.String),
22 cause: Schema.Defect(),
23 },
24) {}
25
26export class ResourceCleanupService extends Context.Reference<{
27 readonly cleanupScheduledTasks: (threadId: ThreadId) => Effect.Effect<void, ResourceCleanupError>;
28 readonly cleanupTerminals: (threadId: string) => Effect.Effect<void, ResourceCleanupError>;
29 readonly cleanupPreviews: (threadId: string) => Effect.Effect<void, ResourceCleanupError>;
30 /** Releases the thread's device claims, as `device_close` does; the devices stay booted. */
31 readonly cleanupDevices: (threadId: ThreadId) => Effect.Effect<void, ResourceCleanupError>;
32 readonly cleanupAttachments: (
33 attachmentIds: ReadonlyArray<string>,
34 ) => Effect.Effect<void, ResourceCleanupError>;
35}>("t3/orchestration-v2/ResourceCleanupService", {
36 defaultValue: () => ({
37 cleanupScheduledTasks: () => Effect.void,
38 cleanupTerminals: () => Effect.void,
39 cleanupPreviews: () => Effect.void,
40 cleanupDevices: () => Effect.void,
41 cleanupAttachments: () => Effect.void,
42 }),
43}) {}
44
45export const layer = Layer.effect(
46 ResourceCleanupService,
47 Effect.gen(function* () {
48 const terminals = yield* TerminalManager.TerminalManager;
49 const previews = yield* PreviewManager.PreviewManager;
50 const devices = yield* DeviceService.DeviceService;
51 const fileSystem = yield* FileSystem.FileSystem;
52 const config = yield* ServerConfig.ServerConfig;
53 const scheduledTasks = yield* ScheduledTaskService.ScheduledTaskService;
54 return {
55 cleanupScheduledTasks: Effect.fn("ResourceCleanupService.cleanupScheduledTasks")(function (
56 threadId: ThreadId,
57 ) {
58 return scheduledTasks
59 .deleteForThread(threadId)
60 .pipe(
61 Effect.mapError(
62 (cause) =>
63 new ResourceCleanupError({ operation: "scheduled-tasks", threadId, cause }),
64 ),
65 );
66 }),
67 cleanupTerminals: Effect.fn("ResourceCleanupService.cleanupTerminals")(function (
68 threadId: string,
69 ) {
70 return terminals
71 .close({ threadId, deleteHistory: true })
72 .pipe(
73 Effect.mapError(
74 (cause) => new ResourceCleanupError({ operation: "terminal", threadId, cause }),
75 ),
76 );
77 }),
78 cleanupPreviews: Effect.fn("ResourceCleanupService.cleanupPreviews")(function (
79 threadId: string,
80 ) {
81 return previews
82 .close({ threadId: ThreadId.make(threadId) })
83 .pipe(
84 Effect.mapError(
85 (cause) => new ResourceCleanupError({ operation: "preview", threadId, cause }),
86 ),
87 );
88 }),
89 cleanupDevices: Effect.fn("ResourceCleanupService.cleanupDevices")(function* (
90 threadId: ThreadId,
91 ) {
92 // One close per session, so a reconnecting host does not keep the other hosts' claims.
93 const sessions = yield* devices.sessionsForThread(threadId);
94 const [, failures] = yield* Effect.partition(
95 sessions,
96 (session) =>
97 devices.close({ threadId, hostId: session.hostId, deviceId: session.deviceId }),
98 { concurrency: 1 },
99 );
100 const [failure] = failures;
101 if (failure !== undefined)
102 return yield* new ResourceCleanupError({ operation: "device", threadId, cause: failure });
103 }),
104 cleanupAttachments: Effect.fn("ResourceCleanupService.cleanupAttachments")(function (
105 attachmentIds: ReadonlyArray<string>,
106 ) {
107 return Effect.forEach(
108 attachmentIds,
109 (attachmentId) => {
110 const extension = parseAttachmentFileExtension(attachmentId);
111 const path =
112 resolveAttachmentPathById({
113 attachmentsDir: config.attachmentsDir,
114 attachmentId,
115 }) ??
116 (extension
117 ? resolveAttachmentRelativePath({
118 attachmentsDir: config.attachmentsDir,
119 relativePath: `${attachmentId}.${extension}`,
120 })
121 : null);
122 return path === null
123 ? Effect.void
124 : fileSystem.remove(path, { force: true }).pipe(
125 Effect.andThen(fileSystem.remove(`${path}.txt`, { force: true })),
126 Effect.mapError(
127 (cause) =>
128 new ResourceCleanupError({ operation: "attachment", attachmentId, cause }),
129 ),
130 );
131 },
132 { discard: true, concurrency: 4 },
133 );
134 }),
135 };
136 }),
137);