1import { ArchiveError, ArchiveObjectHash, ArchiveObjectReceipt } from "@t3tools/contracts";
2// Incremental hashes keep media uploads bounded; Effect Crypto hashes whole buffers.
3// @effect-diagnostics nodeBuiltinImport:off
4import * as NodeCrypto from "node:crypto";
5import * as HostProcess from "@t3tools/shared/HostProcess";
6import * as Config from "effect/Config";
7import * as Context from "effect/Context";
8import * as Effect from "effect/Effect";
9import * as FileSystem from "effect/FileSystem";
10import * as Layer from "effect/Layer";
11import * as Option from "effect/Option";
12import * as Path from "effect/Path";
13import * as Redacted from "effect/Redacted";
14import * as Schema from "effect/Schema";
15import * as Stream from "effect/Stream";
16
17const decodeHash = Schema.decodeEffect(ArchiveObjectHash);
18const isArchiveError = Schema.is(ArchiveError);
19const storageError = (cause: unknown) =>
20 new ArchiveError({ reason: "storage", cause: Redacted.make(cause) });
21
22export class ArchiveObjects extends Context.Service<
23 ArchiveObjects,
24 {
25 readonly verifyDurableStore: Effect.Effect<void, ArchiveError>;
26 readonly directory: Effect.Effect<string, ArchiveError>;
27 readonly root: Option.Option<string>;
28 readonly file: (hash: string) => Effect.Effect<string, ArchiveError>;
29 readonly put: <E, R>(
30 hash: string,
31 body: Stream.Stream<Uint8Array, E, R>,
32 ) => Effect.Effect<typeof ArchiveObjectReceipt.Type, ArchiveError | E, R>;
33 }
34>()("t3/archive/ArchiveObjects") {}
35
36export const layer = Layer.effect(
37 ArchiveObjects,
38 Effect.gen(function* () {
39 const fs = yield* FileSystem.FileSystem;
40 const path = yield* Path.Path;
41 const root = yield* Config.option(Config.String("T3CODE_ARCHIVE_DIR")).pipe(Effect.orDie);
42 const expectedStoreId = yield* Config.option(Config.String("T3CODE_ARCHIVE_STORE_ID")).pipe(
43 Effect.orDie,
44 );
45 const directory = Effect.gen(function* () {
46 if (Option.isNone(root)) return yield* new ArchiveError({ reason: "unavailable" });
47 if (Option.isSome(expectedStoreId)) {
48 const actual = yield* fs
49 .readFileString(path.join(root.value, ".knife-archive-store-id"))
50 .pipe(Effect.mapError(storageError));
51 if (actual.trim() !== expectedStoreId.value)
52 return yield* new ArchiveError({ reason: "storage" });
53 }
54 const stat = yield* fs.stat(root.value).pipe(Effect.mapError(storageError));
55 if (stat.type !== "Directory") return yield* new ArchiveError({ reason: "storage" });
56 const dir = path.join(root.value, "objects");
57 yield* fs
58 .makeDirectory(dir, { recursive: true, mode: 0o700 })
59 .pipe(Effect.mapError(storageError));
60 return dir;
61 });
62 const file = Effect.fn("ArchiveObjects.file")(function* (input: string) {
63 const hash = yield* decodeHash(input).pipe(Effect.mapError(storageError));
64 const name = path.join(yield* directory, hash);
65 const stat = yield* fs.stat(name).pipe(Effect.mapError(storageError));
66 if (stat.type !== "File") return yield* new ArchiveError({ reason: "storage" });
67 return name;
68 });
69 const put = <E, R>(input: string, body: Stream.Stream<Uint8Array, E, R>) =>
70 Effect.scoped(
71 Effect.gen(function* () {
72 const hash = yield* decodeHash(input).pipe(Effect.mapError(storageError));
73 const dir = yield* directory;
74 const temporary = yield* fs
75 .makeTempDirectoryScoped({ directory: dir, prefix: ".upload-" })
76 .pipe(Effect.mapError(storageError));
77 const staged = path.join(temporary, "object");
78 const digest = NodeCrypto.createHash("sha256");
79 let size = 0;
80 yield* body.pipe(
81 Stream.tap((bytes) =>
82 Effect.sync(() => {
83 digest.update(bytes);
84 size += bytes.length;
85 }),
86 ),
87 Stream.run(fs.sink(staged, { flag: "wx", mode: 0o600 })),
88 Effect.mapError((cause) => (isArchiveError(cause) ? cause : storageError(cause))),
89 );
90 if (digest.digest("hex") !== hash)
91 return yield* new ArchiveError({ reason: "integrity" });
92 const verified = NodeCrypto.createHash("sha256");
93 yield* fs.stream(staged).pipe(
94 Stream.runForEach((bytes) =>
95 Effect.sync(() => {
96 verified.update(bytes);
97 }),
98 ),
99 Effect.mapError(storageError),
100 );
101 if (verified.digest("hex") !== hash)
102 return yield* new ArchiveError({ reason: "integrity" });
103 yield* fs.open(staged, { flag: "r+" }).pipe(
104 Effect.flatMap((file) => file.sync),
105 Effect.mapError(storageError),
106 );
107 yield* fs.rename(staged, path.join(dir, hash)).pipe(Effect.mapError(storageError));
108 if (!(yield* HostProcess.isWindows))
109 yield* fs.open(dir).pipe(
110 Effect.flatMap((file) => file.sync),
111 Effect.mapError(storageError),
112 );
113 return { hash, size };
114 }),
115 );
116 return ArchiveObjects.of({
117 root,
118 directory,
119 file,
120 put,
121 verifyDurableStore: Effect.gen(function* () {
122 if (Option.isNone(expectedStoreId)) return yield* new ArchiveError({ reason: "storage" });
123 yield* directory;
124 }),
125 });
126 }),
127);