Open History to see the last commit
1// @effect-diagnostics nodeBuiltinImport:off -- the isolated parser process is an external runtime boundary.
2import * as NodeChildProcess from "node:child_process";
3import * as NodeURL from "node:url";
4import * as Cache from "effect/Cache";
5import * as Clock from "effect/Clock";
6import * as Context from "effect/Context";
7import * as Effect from "effect/Effect";
8import * as FileSystem from "effect/FileSystem";
9import * as Layer from "effect/Layer";
10import * as Schema from "effect/Schema";
11import * as Semaphore from "effect/Semaphore";
12import * as HostProcess from "@t3tools/shared/HostProcess";
13
14const extensions = new Set(["docx", "xlsx", "pptx", "odt", "ods", "odp", "rtf", "epub", "ipynb"]);
15
16class DocumentTextError extends Schema.TaggedError<DocumentTextError>()("DocumentTextError", {
17 operation: Schema.Literal("extract"),
18 path: Schema.String,
19 extension: Schema.String,
20 cause: Schema.Defect(),
21}) {
22 override get message(): string {
23 return `Failed to ${this.operation} ${this.extension} document text.`;
24 }
25}
26
27export const resolveExtractedTextPath = Effect.fn("DocumentText.resolveExtractedTextPath")(
28 function* (path: string) {
29 const extension = path.split(".").at(-1)?.toLowerCase();
30 if (!extension || !extensions.has(extension)) return;
31 const fs = yield* FileSystem.FileSystem;
32 return yield* fs.exists(path + ".txt").pipe(
33 Effect.map((exists) => (exists ? path + ".txt" : undefined)),
34 Effect.orElseSucceed(() => undefined),
35 );
36 },
37);
38
39export class DocumentText extends Context.Service<
40 DocumentText,
41 {
42 readonly ensure: (path: string) => Effect.Effect<string | undefined>;
43 }
44>()("t3/provider/DocumentText") {}
45
46const make = Effect.gen(function* () {
47 const fs = yield* FileSystem.FileSystem;
48 const executable = yield* HostProcess.ExecutablePath;
49 const isExecutable = yield* HostProcess.IsExecutable;
50 const environment = yield* HostProcess.Environment;
51 const clock = yield* Clock.Clock;
52 // A large DOCX worker peaks near 270 MiB RSS, so two slots bound parser memory near 540 MiB.
53 const lock = yield* Semaphore.make(2);
54 const failed = new Map<string, number>();
55 // A cancelled extraction can still be publishing while its retry runs in the other slot.
56 let attempt = 0;
57 // In a bundle import.meta.url is bin.mjs; development and tests use this source module.
58 const entry = NodeURL.fileURLToPath(
59 new URL(import.meta.url.endsWith(".ts") ? "../bin.ts" : "./bin.mjs", import.meta.url),
60 );
61 const extract = Effect.fn("DocumentText.extract")(function* (path: string) {
62 const extension = path.split(".").at(-1)!.toLowerCase();
63 const sidecar = `${path}.txt`;
64 if (yield* fs.exists(sidecar)) return sidecar;
65 if ((failed.get(path) ?? 0) > clock.currentTimeMillisUnsafe()) return;
66 const info = yield* fs.stat(path);
67 if (Number(info.size) > 16 * 1024 * 1024) return;
68 const content = yield* Effect.tryPromise({
69 try: (signal) =>
70 new Promise<string>((resolve, reject) => {
71 NodeChildProcess.execFile(
72 executable,
73 [...(isExecutable ? [] : [entry]), "__document-text", path, `.${extension}`],
74 {
75 signal,
76 timeout: 15_000,
77 maxBuffer: 4 * 1024 * 1024 + 1024,
78 encoding: "utf8",
79 windowsHide: true,
80 env: {
81 ...environment,
82 ELECTRON_RUN_AS_NODE: "1",
83 NODE_OPTIONS: `${environment.NODE_OPTIONS ?? ""} --max-old-space-size=192`,
84 },
85 },
86 (error, stdout) => (error ? reject(error) : resolve(stdout)),
87 );
88 }),
89 catch: (cause) => new DocumentTextError({ operation: "extract", path, extension, cause }),
90 }).pipe(
91 Effect.tapError(({ cause }) => {
92 if (failed.size >= 1024) failed.delete(failed.keys().next().value!);
93 failed.set(
94 path,
95 cause instanceof Error && "code" in cause && cause.code === 2
96 ? Infinity
97 : clock.currentTimeMillisUnsafe() + 1_000,
98 );
99 return Effect.logWarning("Document text extraction failed.", { extension });
100 }),
101 );
102 const pending = `${sidecar}.${attempt++}.part`;
103 yield* fs
104 .writeFileString(pending, content)
105 .pipe(
106 Effect.andThen(fs.rename(pending, sidecar)),
107 Effect.ensuring(fs.remove(pending, { force: true }).pipe(Effect.ignore)),
108 Effect.uninterruptible,
109 );
110 return sidecar;
111 }, lock.withPermits(1));
112 const extracting = yield* Cache.make({ capacity: Infinity, timeToLive: 0, lookup: extract });
113 const ensure = Effect.fn("DocumentText.ensure")(
114 function* (path: string) {
115 const extension = path.split(".").at(-1)?.toLowerCase();
116 if (!extension || !extensions.has(extension)) return;
117 const sidecar = `${path}.txt`;
118 if (yield* fs.exists(sidecar)) return sidecar;
119 return yield* Cache.get(extracting, path);
120 },
121 Effect.orElseSucceed(() => undefined),
122 );
123 return DocumentText.of({ ensure });
124});
125
126export const layer = Layer.effect(DocumentText, make);