mary.my.id
streaming RAR extractor
jsr.io/@mary/rar
jsr
1import { crc32 } from '@mary/crc32';
2
3import { ChecksumMismatchError } from '../errors.ts';
4import type { Reader } from '../reader/types.ts';
5
6import { read } from './buffer.ts';
7
8/**
9 * @module
10 *
11 * shared helpers for the RAR5 and RAR3 extractors: a `DataSegment` description
12 * of one volume-local slice of an entry's data area, a stream that pulls
13 * multiple segments in order, and a lazy-open passthrough used by entry
14 * `body` getters whose underlying data is materialized on demand.
15 */
16
17/** a slice of an entry's data area within a single volume. */
18export interface DataSegment {
19 reader: Reader;
20 offset: number;
21 size: number;
22}
23
24/**
25 * reads exactly `length` bytes from `reader` at `offset`.
26 *
27 * @param reader the volume reader
28 * @param offset starting byte position within the reader
29 * @param length number of bytes to read
30 * @returns the requested bytes as a fresh `Uint8Array`
31 * @throws if the reader returns fewer than `length` bytes
32 */
33export const readBytes = async (
34 reader: Reader,
35 offset: number,
36 length: number,
37): Promise<Uint8Array<ArrayBuffer>> => {
38 return await read(await reader.read(offset, length), length);
39};
40
41/**
42 * adapts a lazily-opened stream into one that can be returned synchronously.
43 * `open` runs on the first pull, not at construction, so a stream that is
44 * never consumed never triggers the decode or I/O work behind it. the
45 * cancellation flag handles the race where cancel arrives before the `open`
46 * callback resolves: without it, the freshly-opened inner stream would be
47 * assigned onto an already-cancelled outer stream and leak its reader.
48 *
49 * @param open invoked once on first pull to produce the inner stream
50 * @returns a ReadableStream that delegates to `open()`'s result
51 */
52export const deferStream = (
53 open: () => Promise<ReadableStream<Uint8Array>>,
54): ReadableStream<Uint8Array> => {
55 let inner: ReadableStreamDefaultReader<Uint8Array> | undefined;
56 let opening: Promise<void> | undefined;
57 let cancelled = false;
58 let cancelReason: unknown;
59
60 const ensureOpen = (): Promise<void> => {
61 if (opening !== undefined) {
62 return opening;
63 }
64 opening = (async () => {
65 const stream = await open();
66 if (cancelled) {
67 await stream.cancel(cancelReason);
68 return;
69 }
70 inner = stream.getReader();
71 })();
72 return opening;
73 };
74
75 return new ReadableStream({
76 async pull(controller) {
77 try {
78 await ensureOpen();
79 } catch (error) {
80 controller.error(error);
81 return;
82 }
83 if (inner === undefined) {
84 controller.close();
85 return;
86 }
87 const { done, value } = await inner.read();
88 if (done) {
89 controller.close();
90 } else {
91 controller.enqueue(value);
92 }
93 },
94 async cancel(reason) {
95 cancelled = true;
96 cancelReason = reason;
97 await inner?.cancel(reason);
98 },
99 });
100};
101
102/**
103 * pulls the bytes of every segment through one ReadableStream, in order, so
104 * the decoder (or AES-CBC decrypt stream) can read an entry's data area
105 * without ever buffering the whole payload in memory.
106 *
107 * zero-length segments are skipped because some readers (e.g. fromFetch)
108 * cannot represent a zero-byte read as a valid HTTP Range header.
109 *
110 * @param segments ordered slices of the entry's data area
111 * @returns a stream emitting every segment's bytes back-to-back
112 */
113export const streamSegments = (
114 segments: readonly DataSegment[],
115): ReadableStream<Uint8Array<ArrayBuffer>> => {
116 let index = 0;
117 let current: ReadableStreamDefaultReader<Uint8Array<ArrayBuffer>> | undefined;
118 let cancelled = false;
119 let cancelReason: unknown;
120
121 return new ReadableStream({
122 async pull(controller) {
123 for (;;) {
124 if (cancelled) {
125 controller.close();
126 return;
127 }
128 if (current === undefined) {
129 // zero-length segments cannot be opened through some readers
130 // (e.g. fromFetch builds a malformed `bytes=N--1` Range
131 // header when length is 0); skip them entirely.
132 while (index < segments.length && segments[index].size === 0) {
133 index++;
134 }
135 if (index === segments.length) {
136 controller.close();
137 return;
138 }
139
140 const segment = segments[index++];
141 const stream = await segment.reader.read(segment.offset, segment.size);
142 if (cancelled) {
143 // cancel arrived while we were awaiting the segment; release
144 // the freshly-opened stream rather than orphaning it.
145 await stream.cancel(cancelReason);
146 controller.close();
147 return;
148 }
149 current = stream.getReader();
150 }
151
152 const { done, value } = await current.read();
153 if (done) {
154 current = undefined;
155 continue;
156 }
157
158 controller.enqueue(value);
159 return;
160 }
161 },
162 async cancel(reason) {
163 cancelled = true;
164 cancelReason = reason;
165 await current?.cancel(reason);
166 },
167 });
168};
169
170/**
171 * an incremental integrity check fed one chunk at a time. {@link finalize}
172 * runs the comparison after the last byte is seen and throws a
173 * {@link ChecksumMismatchError} on mismatch. used by {@link verifyingPassthrough}
174 * to validate the stored fast-path body stream of an entry, and by entry
175 * decode paths to validate the chunked output before returning it.
176 */
177export interface IntegrityVerifier {
178 update(chunk: Uint8Array<ArrayBuffer>): void;
179 /**
180 * called once after the last chunk; compares the accumulated state and
181 * the received byte count against the entry's recorded values.
182 *
183 * @param receivedSize the total number of bytes fed via {@link update}
184 * @throws ChecksumMismatchError when size or hash disagree
185 */
186 finalize(receivedSize: number): void;
187}
188
189/**
190 * wraps `source` so every emitted chunk is fed to `verifier`, and the
191 * verifier's `finalize` is called when the source closes. used by entry
192 * `body` getters whose fast path streams archive bytes verbatim, so a
193 * streamed consumer gets the same integrity guarantee as one going through
194 * a buffered decode + finalize path.
195 *
196 * when `verifier` is `undefined` — e.g. an entry that records neither a
197 * CRC32 nor a hash — the source is returned unchanged.
198 *
199 * @param source the byte stream to verify
200 * @param verifier an incremental verifier, or `undefined` to skip
201 * @returns a stream that pipes `source` and validates on close
202 */
203export const verifyingPassthrough = (
204 source: ReadableStream<Uint8Array<ArrayBuffer>>,
205 verifier: IntegrityVerifier | undefined,
206): ReadableStream<Uint8Array<ArrayBuffer>> => {
207 if (verifier === undefined) {
208 return source;
209 }
210
211 let reader: ReadableStreamDefaultReader<Uint8Array<ArrayBuffer>> | undefined;
212 let received = 0;
213 let cancelled = false;
214
215 return new ReadableStream({
216 async pull(controller) {
217 if (reader === undefined) {
218 if (cancelled) {
219 controller.close();
220 return;
221 }
222 reader = source.getReader();
223 }
224
225 const { done, value } = await reader.read();
226 if (done) {
227 try {
228 verifier.finalize(received);
229 } catch (error) {
230 controller.error(error);
231 return;
232 }
233 controller.close();
234 return;
235 }
236
237 verifier.update(value);
238 received += value.length;
239 controller.enqueue(value);
240 },
241 async cancel(reason) {
242 cancelled = true;
243 await reader?.cancel(reason);
244 },
245 });
246};
247
248/**
249 * builds an {@link IntegrityVerifier} that accumulates a running CRC32 and
250 * checks both the expected size and the expected CRC at finalization.
251 *
252 * @param entryName entry name surfaced in error messages
253 * @param expectedSize the total byte count the stream should deliver
254 * @param expectedCrc the stored CRC32 to compare against
255 * @returns a verifier that throws on size or CRC mismatch
256 */
257export const crc32Verifier = (
258 entryName: string,
259 expectedSize: number,
260 expectedCrc: number,
261): IntegrityVerifier => {
262 let computed = 0;
263 return {
264 update(chunk) {
265 computed = crc32(chunk, computed);
266 },
267 finalize(received) {
268 if (received !== expectedSize) {
269 throw new ChecksumMismatchError(
270 `stored data size mismatch: got ${received} bytes, expected ${expectedSize}` +
271 ` (entry ${entryName})`,
272 );
273 }
274 if (computed !== expectedCrc) {
275 throw new ChecksumMismatchError(
276 `CRC32 mismatch: ` +
277 `stored 0x${expectedCrc.toString(16).padStart(8, '0')}, ` +
278 `computed 0x${computed.toString(16).padStart(8, '0')}` +
279 ` (entry ${entryName})`,
280 );
281 }
282 },
283 };
284};