client.ts · Source

client.tspackages/storage/src/client.tsraw ← src15,939 B← Back to the app
1/**
2 * @factory/storage — thin S3 client over an injectable fetch (PURE logic +
3 * one I/O seam).
4 *
5 * No SDK. Every request is signed by {@link signS3Request} (presign.ts) and
6 * executed through `fetchLike` — global `fetch` in production, a fake in
7 * tests. The S3/MinIO endpoint is never hardcoded; it comes from
8 * {@link S3Config} (environment).
9 *
10 * Error mapping (ARCHITECTURE.md error model):
11 *  - 404            → StorageError, code STORAGE_ERROR, retryable=false
12 *  - other 4xx      → StorageError, code STORAGE_ERROR, retryable=false
13 *  - 5xx            → StorageError, code STORAGE_ERROR, retryable=true
14 *  - network/timeout → StorageError, code STORAGE_TIMEOUT, retryable=true
15 */
16 
17import {
18  sha256Hex,
19  signS3Request,
20  EMPTY_BODY_SHA256,
21  buildUrl,
22  canonicalQuery,
23} from "./presign.js";
24import { objectRequestParts } from "./presign.js";
25import type { S3Config, FetchLike, FetchResponse, FetchInit } from "./types.js";
26import { StorageError } from "./types.js";
27 
28export interface S3ClientOptions {
29  config: S3Config;
30  /** Injectable fetch — defaults to global fetch. Inject a fake in tests. */
31  fetchLike?: FetchLike;
32  /** Observability hook — called before every request and after every response. */
33  onRequest?: (info: {
34    method: string;
35    url: string;
36    headers?: Record<string, string>;
37    status?: number;
38    ok?: boolean;
39    durationMs?: number;
40  }) => void;
41  /** Injectable clock (defaults to `new Date()`) — determinism in tests. */
42  now?: () => Date;
43}
44 
45/** Parsed <Contents><Key> entries of a ListBucketResult response. */
46export interface ListObjectsResult {
47  keys: string[];
48  truncated: boolean;
49  nextToken?: string;
50}
51 
52/** Minimal PURE XML parser for S3 ListBucketResult — extracts <Key> values
53 *  (and <IsTruncated>/<NextContinuationToken>/<NextMarker>) without a DOM.
54 *  Exposed separately so it is testable against a fixture. */
55export function parseListBucketXml(xml: string): ListObjectsResult {
56  const keys: string[] = [];
57  const inKey = /<Key>([\s\S]*?)<\/Key>/g;
58  let m: RegExpExecArray | null;
59  while ((m = inKey.exec(xml)) !== null) {
60    keys.push(decodeXmlEntities(m[1]!.trim()));
61  }
62  const trunc = /<IsTruncated>(true|false)<\/IsTruncated>/.exec(xml);
63  const truncated = trunc?.[1] === "true";
64  const next =
65    /<NextContinuationToken>([\s\S]*?)<\/NextContinuationToken>/.exec(xml)?.[1] ??
66    /<NextMarker>([\s\S]*?)<\/NextMarker>/.exec(xml)?.[1];
67  const out: ListObjectsResult = { keys, truncated };
68  if (next !== undefined && next.trim() !== "") out.nextToken = next.trim();
69  return out;
70}
71 
72function decodeXmlEntities(s: string): string {
73  return s
74    .replace(/&lt;/g, "<")
75    .replace(/&gt;/g, ">")
76    .replace(/&quot;/g, '"')
77    .replace(/&apos;/g, "'")
78    .replace(/&amp;/g, "&");
79}
80 
81function decodeXmlEntitiesSafe(s: string | undefined): string {
82  return s === undefined ? "" : decodeXmlEntities(s);
83}
84 
85/** Virtual-host style bucket URL (bucket becomes a subdomain of the host). */
86function buildVirtualHostUrl(
87  origin: string,
88  bucket: string,
89  query: Record<string, string>,
90): string {
91  const m = /^([a-z][a-z0-9+.-]*:\/\/)([^/?#]+)(.*)$/i.exec(origin);
92  if (!m) return `${origin}/${bucket}/?${canonicalQuery(query)}`;
93  return `${m[1]}${bucket}.${m[2]!}${m[3] ?? ""}/?${canonicalQuery(query)}`;
94}
95 
96function parseHeadResponse(resp: FetchResponse): {
97  size: number;
98  contentType: string;
99  etag: string;
100  lastModified: string;
101} {
102  const h = resp.headers;
103  const lower: Record<string, string> = {};
104  for (const [k, v] of Object.entries(h)) lower[k.toLowerCase()] = v;
105  const size = Number.parseInt(lower["content-length"] ?? "0", 10);
106  return {
107    size: Number.isFinite(size) ? size : 0,
108    contentType: lower["content-type"] ?? "application/octet-stream",
109    etag: lower["etag"] ?? "",
110    lastModified: lower["last-modified"] ?? "",
111  };
112}
113 
114/** Extract the ETag from a CreateMultipartUpload response. */
115export function parseUploadIdXml(xml: string): string {
116  return decodeXmlEntitiesSafe(/<UploadId>([\s\S]*?)<\/UploadId>/.exec(xml)?.[1] ?? "").trim();
117}
118 
119/** Build the CompleteMultipartUpload request XML (PURE). */
120export function buildCompleteMultipartXml(
121  parts: readonly { partNumber: number; etag: string }[],
122): string {
123  const entries = parts
124    .slice()
125    .sort((a, b) => a.partNumber - b.partNumber)
126    .map(
127      (p) =>
128        `  <Part>\n    <PartNumber>${p.partNumber}</PartNumber>\n    <ETag>&quot;${p.etag}&quot;</ETag>\n  </Part>`,
129    )
130    .join("\n");
131  return `<?xml version="1.0" encoding="UTF-8"?>\n<CompleteMultipartUpload>\n${entries}\n</CompleteMultipartUpload>`;
132}
133 
134/**
135 * Adapt the global `fetch` to {@link FetchLike}: lower-cased header map,
136 * the body fully read into a `Uint8Array`, and `json()`/`text()` over that
137 * buffer (a `Response` stream can only be consumed once, so the adapter
138 * reads it eagerly — object sizes here are bounded by the upload limits).
139 */
140async function globalFetchLike(input: string, init?: FetchInit): Promise<FetchResponse> {
141  const res = await fetch(input, init ?? undefined);
142  const headers: Record<string, string> = {};
143  res.headers.forEach((value, key) => {
144    headers[key.toLowerCase()] = value;
145  });
146  const bodyBytes: Uint8Array = new Uint8Array(await res.arrayBuffer());
147  return {
148    ok: res.ok,
149    status: res.status,
150    headers,
151    body: bodyBytes.length > 0 ? bodyBytes : null,
152    json<T>(): Promise<T> {
153      return Promise.resolve(JSON.parse(new TextDecoder().decode(bodyBytes)) as T);
154    },
155    text(): Promise<string> {
156      return Promise.resolve(new TextDecoder().decode(bodyBytes));
157    },
158  };
159}
160 
161export class S3Client {
162  private readonly config: S3Config;
163  private readonly fetchLike: FetchLike;
164  private readonly onRequest?: S3ClientOptions["onRequest"];
165  private readonly now: () => Date;
166 
167  constructor(opts: S3ClientOptions) {
168    this.config = opts.config;
169    this.fetchLike = opts.fetchLike ?? globalFetchLike;
170    this.onRequest = opts.onRequest;
171    this.now = opts.now ?? (() => new Date());
172  }
173 
174  /** PUT an object. Signs for the exact payload hash; the body is the raw bytes. */
175  async putObject(key: string, body: Uint8Array | string, contentType: string): Promise<void> {
176    const bytes = typeof body === "string" ? new TextEncoder().encode(body) : body;
177    const payloadHash = sha256Hex(bytes);
178    const { path } = objectRequestParts(this.config, key);
179    const signed = this.sign("PUT", path, { "content-type": contentType }, payloadHash);
180    const url = buildUrl(this.config, path);
181    await this.send(url, {
182      method: "PUT",
183      headers: signed.headers,
184      body: bytes,
185    });
186  }
187 
188  /** GET an object (full body). */
189  async getObject(key: string): Promise<{ body: Uint8Array; contentType: string; size: number }> {
190    const { path } = objectRequestParts(this.config, key);
191    const signed = this.sign("GET", path, {}, EMPTY_BODY_SHA256);
192    const url = buildUrl(this.config, path);
193    const resp = await this.send(url, { method: "GET", headers: signed.headers, body: null });
194    const h: Record<string, string> = {};
195    for (const [k, v] of Object.entries(resp.headers)) h[k.toLowerCase()] = v;
196    const size = Number.parseInt(h["content-length"] ?? "0", 10);
197    return {
198      body: resp.body ?? new Uint8Array(0),
199      contentType: h["content-type"] ?? "application/octet-stream",
200      size: Number.isFinite(size) ? size : (resp.body?.length ?? 0),
201    };
202  }
203 
204  /**
205   * Read the FIRST `n` bytes of an object (HTTP range GET) — the MIME
206   * magic-byte sniff path (TASK-0101 §4): the server downloads only the
207   * first 8 KiB of an uploaded object to run `sniffMagicBytes`.
208   * Returns the partial body plus the total object size from
209   * `Content-Range` (so the caller can still verify the declared size).
210   */
211  async readPrefix(key: string, n: number): Promise<{ prefix: Uint8Array; totalSize: number }> {
212    const { path } = objectRequestParts(this.config, key);
213    const signed = this.sign("GET", path, { range: `bytes=0-${n - 1}` }, EMPTY_BODY_SHA256);
214    const url = buildUrl(this.config, path);
215    const resp = await this.send(url, { method: "GET", headers: signed.headers, body: null });
216    const h: Record<string, string> = {};
217    for (const [k, v] of Object.entries(resp.headers)) h[k.toLowerCase()] = v;
218    // Content-Range: bytes 0-7/482113 (total may be "*" if unknown).
219    const m = /bytes\s+\d+-\d+\/(\d+|\*)/.exec(h["content-range"] ?? "");
220    const total = m?.[1] !== undefined && m[1] !== "*" ? Number.parseInt(m[1], 10) : NaN;
221    return {
222      prefix: resp.body ?? new Uint8Array(0),
223      totalSize: Number.isFinite(total) ? total : (resp.body?.length ?? 0),
224    };
225  }
226 
227  /** HEAD an object (metadata only). */
228  async headObject(key: string): Promise<{
229    size: number;
230    contentType: string;
231    etag: string;
232    lastModified: string;
233  }> {
234    const { path } = objectRequestParts(this.config, key);
235    const signed = this.sign("HEAD", path, {}, EMPTY_BODY_SHA256);
236    const url = buildUrl(this.config, path);
237    const resp = await this.send(url, { method: "HEAD", headers: signed.headers, body: null });
238    return parseHeadResponse(resp);
239  }
240 
241  /** DELETE an object (idempotent — S3 deletes missing objects with 204). */
242  async deleteObject(key: string): Promise<void> {
243    const { path } = objectRequestParts(this.config, key);
244    const signed = this.sign("DELETE", path, {}, EMPTY_BODY_SHA256);
245    const url = buildUrl(this.config, path);
246    await this.send(url, { method: "DELETE", headers: signed.headers, body: null });
247  }
248 
249  /** List objects under a prefix (GET bucket?list-type=2&prefix=…).
250   *
251   *  Pagination: pass the previous page's `nextToken` as `continuationToken`
252   *  (TASK-0104 orphan sweeper — a full `files/` sweep must page past the
253   *  1000-key cap, not silently truncate). */
254  async listObjects(prefix: string, maxKeys?: number, continuationToken?: string): Promise<ListObjectsResult> {
255    const query: Record<string, string> = { "list-type": "2", prefix };
256    if (maxKeys !== undefined) query["max-keys"] = String(maxKeys);
257    if (continuationToken !== undefined) query["continuation-token"] = continuationToken;
258    // List is a bucket-level operation. Path-style puts the bucket in the
259    // path; virtual-host puts it in the host and the canonical path is "/".
260    const origin = this.config.endpoint.replace(/\/+$/, "");
261    const vh = this.config.forcePathStyle !== true;
262    const path = vh ? "/" : `/${this.config.bucket}`;
263    const signed = this.sign("GET", path, {}, EMPTY_BODY_SHA256, query);
264    const url = vh
265      ? buildVirtualHostUrl(origin, this.config.bucket, query)
266      : buildUrl(this.config, path, query);
267    const resp = await this.send(url, { method: "GET", headers: signed.headers, body: null });
268    const xml = await resp.text();
269    return parseListBucketXml(xml);
270  }
271 
272  /** Start a multipart upload → UploadId. (Large uploads use multipart.) */
273  async createMultipartUpload(key: string): Promise<{ uploadId: string }> {
274    const { path } = objectRequestParts(this.config, key);
275    const query: Record<string, string> = { uploads: "" };
276    const signed = this.sign("POST", path, {}, EMPTY_BODY_SHA256, query);
277    const url = buildUrl(this.config, path, query);
278    const resp = await this.send(url, { method: "POST", headers: signed.headers, body: null });
279    const xml = await resp.text();
280    return { uploadId: parseUploadIdXml(xml) };
281  }
282 
283  /** Upload one part of a multipart upload → ETag. */
284  async uploadPart(
285    key: string,
286    uploadId: string,
287    partNumber: number,
288    body: Uint8Array | string,
289  ): Promise<{ partNumber: number; etag: string }> {
290    const bytes = typeof body === "string" ? new TextEncoder().encode(body) : body;
291    const payloadHash = sha256Hex(bytes);
292    const { path } = objectRequestParts(this.config, key);
293    const query: Record<string, string> = { partNumber: String(partNumber), uploadId };
294    const signed = this.sign("PUT", path, {}, payloadHash, query);
295    const url = buildUrl(this.config, path, query);
296    const resp = await this.send(url, {
297      method: "PUT",
298      headers: signed.headers,
299      body: bytes,
300    });
301    const h: Record<string, string> = {};
302    for (const [k, v] of Object.entries(resp.headers)) h[k.toLowerCase()] = v;
303    const etag = (h["etag"] ?? "").replace(/^"|"$/g, "");
304    return { partNumber, etag };
305  }
306 
307  /** Complete a multipart upload (assembles the parts server-side). */
308  async completeMultipartUpload(
309    key: string,
310    uploadId: string,
311    parts: readonly { partNumber: number; etag: string }[],
312  ): Promise<void> {
313    const xml = buildCompleteMultipartXml(parts);
314    const bytes = new TextEncoder().encode(xml);
315    const payloadHash = sha256Hex(bytes);
316    const { path } = objectRequestParts(this.config, key);
317    const query: Record<string, string> = { uploadId };
318    // content-type is part of the signed headers — sign it, don't add it after.
319    const signed = this.sign(
320      "POST",
321      path,
322      { "content-type": "application/xml" },
323      payloadHash,
324      query,
325    );
326    const url = buildUrl(this.config, path, query);
327    await this.send(url, {
328      method: "POST",
329      headers: signed.headers,
330      body: bytes,
331    });
332  }
333 
334  // ── internals ──────────────────────────────────────────────────────────
335 
336  private sign(
337    method: string,
338    path: string,
339    headers: Record<string, string>,
340    payloadHash: string,
341    query?: Record<string, string>,
342  ): ReturnType<typeof signS3Request> {
343    return signS3Request({
344      method,
345      path,
346      query: query ?? {},
347      headers,
348      payloadHash,
349      config: this.config,
350      now: this.now(),
351      // Authenticated header-signed requests MUST carry x-amz-date +
352      // x-amz-content-sha256 in the signed set (AWS SigV4 requirement).
353      alwaysSignContentSha: true,
354    });
355  }
356 
357  private async send(url: string, init: FetchInit): Promise<FetchResponse> {
358    const t0 = typeof performance !== "undefined" ? performance.now() : Date.now();
359    this.onRequest?.({ method: init.method ?? "GET", url });
360    let resp: FetchResponse;
361    try {
362      resp = await this.fetchLike(url, init);
363    } catch (err) {
364      const durationMs = typeof performance !== "undefined" ? performance.now() - t0 : 0;
365      this.onRequest?.({
366        method: init.method ?? "GET",
367        url,
368        durationMs,
369        ok: false,
370        status: 0,
371      });
372      throw new StorageError({
373        message: `storage request failed: ${(err as Error)?.message ?? "network error"}`,
374        code: "STORAGE_TIMEOUT",
375        retryable: true,
376        cause: err,
377      });
378    }
379    const durationMs = typeof performance !== "undefined" ? performance.now() - t0 : 0;
380    this.onRequest?.({
381      method: init.method ?? "GET",
382      url,
383      status: resp.status,
384      ok: resp.ok,
385      durationMs,
386    });
387    if (!resp.ok) {
388      const message = await safeErrorBody(resp, url);
389      const retryable = resp.status >= 500 || resp.status === 429;
390      throw new StorageError({
391        message: `storage ${init.method ?? "GET"} ${url} → ${resp.status}: ${message}`,
392        code: retryable ? "STORAGE_ERROR" : "STORAGE_ERROR",
393        status: resp.status,
394        retryable,
395      });
396    }
397    return resp;
398  }
399}
400 
401async function safeErrorBody(resp: FetchResponse, url: string): Promise<string> {
402  try {
403    if (resp.body && resp.body.length > 0) {
404      const text = new TextDecoder("utf-8", { fatal: false }).decode(resp.body);
405      const code = /<Code>([\s\S]*?)<\/Code>/.exec(text)?.[1] ?? text.slice(0, 200);
406      return code.trim() || `HTTP ${resp.status}`;
407    }
408  } catch {
409    // fall through
410  }
411  return `HTTP ${resp.status} on ${url.split("/").slice(-1)[0] ?? ""}`;
412}
413