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 Context from "effect/Context";
6import * as Effect from "effect/Effect";
7import * as FileSystem from "effect/FileSystem";
8import * as Layer from "effect/Layer";
9import * as Option from "effect/Option";
10import * as Redacted from "effect/Redacted";
11import * as Schema from "effect/Schema";
12import * as Stream from "effect/Stream";
13import type * as Scope from "effect/Scope";
14import { HttpClient, HttpClientRequest, HttpClientResponse, FetchHttpClient } from "effect/http";
15import * as ArchiveConnection from "./ArchiveConnection.ts";
16import * as ArchiveCatalog from "./ArchiveCatalog.ts";
17import {
18 ArchiveCatalogEntry,
19 type ArchivePublication,
20 type EnvironmentId,
21 type ThreadId,
22} from "@t3tools/contracts";
23import * as ArchiveObjects from "./ArchiveObjects.ts";
24
25const decodeHash = Schema.decodeEffect(ArchiveObjectHash);
26const storageError = (cause: unknown) =>
27 new ArchiveError({ reason: "storage", cause: Redacted.make(cause) });
28const connectionError = (cause: unknown) =>
29 new ArchiveError({ reason: "unavailable", cause: Redacted.make(cause) });
30
31export class ArchiveClient extends Context.Service<
32 ArchiveClient,
33 {
34 readonly enabled: Effect.Effect<boolean, ArchiveError>;
35 readonly publish: (
36 input: ArchivePublication,
37 ) => Effect.Effect<ArchiveCatalogEntry, ArchiveError>;
38 readonly withdraw: (
39 environmentId: EnvironmentId,
40 threadId: ThreadId,
41 manifestHash: string,
42 ) => Effect.Effect<void, ArchiveError>;
43 readonly location: Effect.Effect<string, ArchiveError>;
44 readonly upload: (
45 file: string,
46 ) => Effect.Effect<typeof ArchiveObjectReceipt.Type, ArchiveError>;
47 readonly open: (
48 origin: string,
49 hash: string,
50 range?: string,
51 ) => Effect.Effect<
52 | { kind: "file"; path: string }
53 | { kind: "response"; response: HttpClientResponse.HttpClientResponse },
54 ArchiveError,
55 Scope.Scope
56 >;
57 readonly download: (
58 location: string,
59 hash: string,
60 destination: string,
61 ) => Effect.Effect<void, ArchiveError>;
62 }
63>()("t3/archive/ArchiveClient") {}
64
65export const layer = Layer.effect(
66 ArchiveClient,
67 Effect.gen(function* () {
68 const fs = yield* FileSystem.FileSystem;
69 const http = yield* HttpClient.HttpClient;
70 const connection = yield* ArchiveConnection.ArchiveConnection;
71 const catalog = yield* ArchiveCatalog.ArchiveCatalog;
72 const local = yield* ArchiveObjects.ArchiveObjects;
73 const enabled = connection.enabled;
74 const target = connection.target;
75 const location = target.pipe(
76 Effect.flatMap((connection) =>
77 Option.isSome(connection)
78 ? Effect.succeed(connection.value.url)
79 : Option.isSome(local.root)
80 ? Effect.succeed(`file:${local.root.value}`)
81 : Effect.fail(new ArchiveError({ reason: "unavailable" })),
82 ),
83 );
84 const upload = Effect.fn("ArchiveClient.upload")(function* (file: string) {
85 const digest = NodeCrypto.createHash("sha256");
86 yield* fs.stream(file).pipe(
87 Stream.runForEach((bytes) =>
88 Effect.sync(() => {
89 digest.update(bytes);
90 }),
91 ),
92 Effect.mapError(storageError),
93 );
94 const hash = digest.digest("hex");
95 const connection = yield* target;
96 if (Option.isNone(connection))
97 return yield* local.put(hash, fs.stream(file)).pipe(Effect.mapError(storageError));
98 const receipt = yield* HttpClientRequest.put(
99 `${connection.value.url}/api/archive/objects/${hash}`,
100 ).pipe(
101 HttpClientRequest.setHeader("authorization", connection.value.authorization),
102 HttpClientRequest.bodyStream(fs.stream(file), { contentType: "application/octet-stream" }),
103 http.execute,
104 Effect.provideService(FetchHttpClient.RequestInit, { redirect: "error" }),
105 Effect.flatMap(HttpClientResponse.filterStatusOk),
106 Effect.flatMap(HttpClientResponse.schemaBodyJson(ArchiveObjectReceipt)),
107 Effect.mapError(connectionError),
108 Effect.timeout("5 minutes"),
109 Effect.mapError(connectionError),
110 );
111 const stat = yield* fs.stat(file).pipe(Effect.mapError(storageError));
112 if (receipt.hash !== hash || BigInt(receipt.size) !== stat.size)
113 return yield* new ArchiveError({ reason: "integrity" });
114 return receipt;
115 });
116 const open = Effect.fn("ArchiveClient.open")(function* (
117 origin: string,
118 input: string,
119 range?: string,
120 ) {
121 const hash = yield* decodeHash(input).pipe(Effect.mapError(storageError));
122 if ((yield* location) !== origin) return yield* new ArchiveError({ reason: "unavailable" });
123 const connection = yield* target;
124 if (Option.isNone(connection))
125 return { kind: "file" as const, path: yield* local.file(hash) };
126 const response = yield* HttpClient.withScope(http)
127 .execute(
128 HttpClientRequest.get(`${connection.value.url}/api/archive/objects/${hash}`).pipe(
129 HttpClientRequest.setHeaders({
130 authorization: connection.value.authorization,
131 ...(range ? { range } : {}),
132 "accept-encoding": "identity",
133 }),
134 ),
135 )
136 .pipe(
137 Effect.provideService(FetchHttpClient.RequestInit, { redirect: "error" }),
138 Effect.timeout("10 seconds"),
139 Effect.mapError(connectionError),
140 );
141 if (![200, 206, 416].includes(response.status))
142 return yield* new ArchiveError({ reason: "unavailable" });
143 return { kind: "response" as const, response };
144 });
145 const download = Effect.fn("ArchiveClient.download")(function* (
146 origin: string,
147 input: string,
148 destination: string,
149 ) {
150 const hash = yield* decodeHash(input).pipe(Effect.mapError(storageError));
151 if ((yield* location) !== origin) return yield* new ArchiveError({ reason: "unavailable" });
152 const connection = yield* target;
153 const body: Stream.Stream<Uint8Array, unknown> = Option.isNone(connection)
154 ? fs.stream(yield* local.file(hash))
155 : (yield* HttpClientRequest.get(`${connection.value.url}/api/archive/objects/${hash}`).pipe(
156 HttpClientRequest.setHeader("authorization", connection.value.authorization),
157 http.execute,
158 Effect.provideService(FetchHttpClient.RequestInit, { redirect: "error" }),
159 Effect.timeout("10 seconds"),
160 Effect.flatMap(HttpClientResponse.filterStatusOk),
161 Effect.mapError(connectionError),
162 )).stream;
163 const digest = NodeCrypto.createHash("sha256");
164 yield* body.pipe(
165 Stream.timeoutOrElse({
166 duration: "10 seconds",
167 orElse: () => Stream.fail(new ArchiveError({ reason: "unavailable" })),
168 }),
169 Stream.tap((bytes) =>
170 Effect.sync(() => {
171 digest.update(bytes);
172 }),
173 ),
174 Stream.run(fs.sink(destination, { flag: "wx", mode: 0o600 })),
175 Effect.mapError(connectionError),
176 );
177 if (digest.digest("hex") !== hash) return yield* new ArchiveError({ reason: "integrity" });
178 });
179 const publish = (input: ArchivePublication) =>
180 Effect.gen(function* () {
181 const connection = yield* target;
182 if (Option.isNone(connection)) return yield* catalog.publish(input);
183 return yield* HttpClientRequest.post(`${connection.value.url}/api/archive/catalog`).pipe(
184 HttpClientRequest.setHeader("authorization", connection.value.authorization),
185 HttpClientRequest.bodyJson(input),
186 Effect.flatMap(http.execute),
187 Effect.provideService(FetchHttpClient.RequestInit, { redirect: "error" }),
188 Effect.flatMap(HttpClientResponse.filterStatusOk),
189 Effect.flatMap(
190 HttpClientResponse.schemaBodyJson(Schema.toCodecJson(ArchiveCatalogEntry)),
191 ),
192 Effect.timeout("5 minutes"),
193 Effect.mapError(connectionError),
194 );
195 });
196 const withdraw = (environmentId: EnvironmentId, threadId: ThreadId, manifestHash: string) =>
197 Effect.gen(function* () {
198 const connection = yield* target;
199 if (Option.isNone(connection))
200 return yield* catalog.withdraw(environmentId, threadId, manifestHash);
201 yield* HttpClientRequest.delete(
202 `${connection.value.url}/api/archive/catalog/${encodeURIComponent(environmentId)}/${encodeURIComponent(threadId)}`,
203 ).pipe(
204 HttpClientRequest.setHeader("authorization", connection.value.authorization),
205 HttpClientRequest.bodyJson({ manifestHash }),
206 Effect.flatMap(http.execute),
207 Effect.provideService(FetchHttpClient.RequestInit, { redirect: "error" }),
208 Effect.flatMap(HttpClientResponse.filterStatusOk),
209 Effect.mapError(connectionError),
210 Effect.timeout("10 seconds"),
211 Effect.mapError(connectionError),
212 );
213 });
214 return ArchiveClient.of({ enabled, location, upload, download, open, publish, withdraw });
215 }),
216);