| 1 | // This file is part of HFS - Copyright 2021-2023, Massimo Melina <a@rejetto.com> - License https://www.gnu.org/licenses/gpl-3.0.txt |
| 2 | |
| 3 | import { Readable } from 'stream' |
| 4 | import assert from 'assert' |
| 5 | import { buf as crc32 } from 'crc-32' |
| 6 | |
| 7 | const ZIP64_SIZE_LIMIT = 0xffffffff |
| 8 | const ZIP64_NUMBER_LIMIT = 0xffff |
| 9 | |
| 10 | const FLAGS = 0x0808 // bit3 = no crc in local header + bit11 = utf8 |
| 11 | |
| 12 | interface ZipSource { |
| 13 | path: string |
| 14 | sourcePath?: string |
| 15 | getData?: () => Readable // deferred stream, so that we don't keep many open files because of calculateSize() |
| 16 | size?: number |
| 17 | ts?: Date |
| 18 | mode?: number |
| 19 | } |
| 20 | interface QuickZipEntry { |
| 21 | size: number, |
| 22 | crc: number, |
| 23 | ts: Date, |
| 24 | pathAsBuffer: Buffer, |
| 25 | offset: number, |
| 26 | version: number, |
| 27 | extAttr: number, |
| 28 | } |
| 29 | // the point of this class is the method applyRange, which allows seeking forward in the zip file quickly (useful to resume downloads) |
| 30 | export class QuickZipStream extends Readable { |
| 31 | private workingFile: Readable | undefined |
| 32 | private finished = false |
| 33 | private readonly entries: QuickZipEntry[] = [] |
| 34 | private dataWritten = 0 |
| 35 | private consumedCalculating: ZipSource[] = [] |
| 36 | private skip: number = 0 |
| 37 | private limit?: number |
| 38 | private now = new Date() |
| 39 | |
| 40 | constructor(private readonly walker: AsyncIterableIterator<ZipSource>) { |
| 41 | super({}) |
| 42 | } |
| 43 | |
| 44 | getArchiveEntries() { |
| 45 | return this.entries.map(x => String(x.pathAsBuffer)) |
| 46 | } |
| 47 | |
| 48 | closeStream() { |
| 49 | this.finished = true |
| 50 | this.push(null) // EOF |
| 51 | } |
| 52 | |
| 53 | continuePiping() { |
| 54 | setImmediate(() => this.push('')) // stimulate the pipe. In some situations this needs to be done at next tick; not sure when, but maybe it's when we are inside a _read call that writes nothing |
| 55 | } |
| 56 | |
| 57 | applyRange(start: number, end: number) { |
| 58 | if (end < start) |
| 59 | return this.closeStream() |
| 60 | this.skip = start |
| 61 | this.limit = end - start + 1 |
| 62 | } |
| 63 | |
| 64 | controlledPush(chunk: number[] | Buffer) { |
| 65 | if (this.finished) return |
| 66 | if (Array.isArray(chunk)) |
| 67 | chunk = buffer(chunk) |
| 68 | this.dataWritten += chunk.length |
| 69 | if (this.skip) { |
| 70 | if (this.skip >= chunk.length) { |
| 71 | this.skip -= chunk.length |
| 72 | return true |
| 73 | } |
| 74 | chunk = chunk.subarray(this.skip) |
| 75 | this.skip = 0 |
| 76 | } |
| 77 | const lastBit = this.limit! < chunk.length |
| 78 | if (lastBit) |
| 79 | chunk = chunk.subarray(0, this.limit) |
| 80 | |
| 81 | const ret = this.push(chunk) |
| 82 | if (lastBit) |
| 83 | this.closeStream() |
| 84 | return ret |
| 85 | } |
| 86 | |
| 87 | async calculateSize(howLong:number = 1000) { |
| 88 | const endBy = Date.now() + howLong |
| 89 | for await (const value of this.walker) { // getting the entries is the slow part |
| 90 | if (Date.now() >= endBy) |
| 91 | return NaN |
| 92 | this.consumedCalculating.push(value) // keep the same shape of the generator, so |
| 93 | } |
| 94 | // if we reach here, then we were able to consume all entries of the walker (in time) |
| 95 | let offset = 0 |
| 96 | let centralDirSize = 0 |
| 97 | for (const file of this.consumedCalculating) { |
| 98 | const pathSize = Buffer.from(file.path, 'utf8').length |
| 99 | const { size=0, getData } = file |
| 100 | const extraLength = (size > ZIP64_SIZE_LIMIT ? 2 : 0) + (offset > ZIP64_SIZE_LIMIT ? 1 : 0) |
| 101 | const extraDataSize = extraLength && (2+2 + extraLength*8) |
| 102 | offset += 4+2+2+2+ 4+4+4+4+ 2+2+ pathSize + size |
| 103 | if (getData) |
| 104 | offset += 4+4+2*(size > ZIP64_SIZE_LIMIT ? 8 : 4) |
| 105 | centralDirSize += 4+2+2+2+2+ 4+4+4+4+ 2+2+2+2+2+ 4+4 + pathSize + extraDataSize |
| 106 | } |
| 107 | const n = this.consumedCalculating.length |
| 108 | const centralDirOffset = offset |
| 109 | if (n >= ZIP64_NUMBER_LIMIT |
| 110 | || centralDirOffset >= ZIP64_SIZE_LIMIT |
| 111 | || centralDirSize >= ZIP64_SIZE_LIMIT) |
| 112 | centralDirSize += 4+8+2+2+4+4+8+8+8+8+4+4+8+4 |
| 113 | centralDirSize += 4+4+2+2+4+4+2 |
| 114 | return offset + centralDirSize |
| 115 | } |
| 116 | |
| 117 | async _read() { |
| 118 | if (this.finished || this.destroyed) return |
| 119 | if (this.workingFile) |
| 120 | return this.workingFile.resume() |
| 121 | const file = this.consumedCalculating.shift() |
| 122 | || (await this.walker.next()).value as ZipSource |
| 123 | if (!file) |
| 124 | return this.closeArchive() |
| 125 | let { path, sourcePath, getData, size=0, ts=this.now, mode=0o40775 } = file |
| 126 | const pathAsBuffer = Buffer.from(path, 'utf8') |
| 127 | const offset = this.dataWritten |
| 128 | const version = 20 |
| 129 | this.controlledPush([ |
| 130 | 4, 0x04034b50, |
| 131 | 2, version, |
| 132 | 2, FLAGS, |
| 133 | 2, 0, // compression = store |
| 134 | ...ts2buf(ts || this.now), |
| 135 | // in our mode, crc and sizes are zero in local file header, and written in the data-descriptor after the file data, and in the central directory |
| 136 | 4, 0, // crc |
| 137 | 4, 0, // compressed size |
| 138 | 4, 0, // uncompressed size |
| 139 | 2, pathAsBuffer.length, |
| 140 | 2, 0, // length of the extra field |
| 141 | ]) |
| 142 | this.controlledPush(pathAsBuffer) |
| 143 | if (this.finished) return |
| 144 | |
| 145 | const cache = sourcePath ? crcCache[sourcePath] : undefined |
| 146 | const cacheHit = Number(cache?.ts) === Number(ts) |
| 147 | let crc = cacheHit ? cache!.crc : getData ? crc32([]) : 0 |
| 148 | const extAttr = !mode ? 0 : (mode | 0x8000) * 0x10000 // it's like <<16 but doesn't overflow so easily |
| 149 | const entry = { size, crc, pathAsBuffer, ts, offset, version, extAttr } |
| 150 | if (!getData) { |
| 151 | this.entries.push(entry) |
| 152 | return this.continuePiping() |
| 153 | } |
| 154 | if (this.skip >= size && cacheHit) { |
| 155 | this.skip -= size |
| 156 | this.dataWritten += size |
| 157 | this.entries.push(entry) |
| 158 | } |
| 159 | else await new Promise<void>(resolve => { |
| 160 | const data = this.workingFile = getData() |
| 161 | data.on('error', (err) => { |
| 162 | if ((err as any)?.code !== 'EACCES') |
| 163 | console.error('Zipping:', String(err)) |
| 164 | data.destroy(err) |
| 165 | resolve() |
| 166 | }) |
| 167 | data.on('end', ()=>{ |
| 168 | entry.crc = crc |
| 169 | if (sourcePath) |
| 170 | crcCache[sourcePath] = { ts, crc } |
| 171 | this.entries.push(entry) |
| 172 | resolve() |
| 173 | }) |
| 174 | data.on('data', chunk => { |
| 175 | if (this.destroyed) |
| 176 | return data.destroy() |
| 177 | if (!this.controlledPush(chunk)) // destination buffer full |
| 178 | data.pause() // slow down |
| 179 | if (!cacheHit) |
| 180 | crc = crc32(chunk, crc) |
| 181 | if (this.finished) |
| 182 | return data.destroy() |
| 183 | }) |
| 184 | }) |
| 185 | this.workingFile = undefined |
| 186 | const sizeForSize = size > ZIP64_SIZE_LIMIT ? 8 : 4 |
| 187 | if (this.controlledPush([ |
| 188 | 4, 0x08074b50, |
| 189 | 4, entry.crc, |
| 190 | sizeForSize, size, |
| 191 | sizeForSize, size, |
| 192 | ])) |
| 193 | this.continuePiping() |
| 194 | } |
| 195 | |
| 196 | closeArchive() { |
| 197 | let centralDirOffset = this.dataWritten |
| 198 | for (let { size, ts, crc, offset, pathAsBuffer, version, extAttr } of this.entries) { |
| 199 | const extra = [] |
| 200 | if (size > ZIP64_SIZE_LIMIT) { |
| 201 | extra.push(size, size) |
| 202 | size = ZIP64_SIZE_LIMIT |
| 203 | } |
| 204 | if (offset > ZIP64_SIZE_LIMIT) { |
| 205 | extra.push(offset) |
| 206 | offset = ZIP64_SIZE_LIMIT |
| 207 | } |
| 208 | const extraData = buffer(!extra.length ? [] |
| 209 | : [ 2,1, 2,8*extra.length, ...extra.flatMap(x=> [8,x]) ]) |
| 210 | if (extraData.length && version < 45) |
| 211 | version = 45 |
| 212 | this.controlledPush([ |
| 213 | 4, 0x02014b50, // central dir signature |
| 214 | 2, version, |
| 215 | 2, version, |
| 216 | 2, FLAGS, |
| 217 | 2, 0, // compression method = store |
| 218 | ...ts2buf(ts), |
| 219 | 4, crc, |
| 220 | 4, size, // compressed |
| 221 | 4, size, |
| 222 | 2, pathAsBuffer.length, |
| 223 | 2, extraData.length, |
| 224 | 2, 0, //comment length |
| 225 | 2, 0, // disk |
| 226 | 2, 0, // attr |
| 227 | 4, extAttr, |
| 228 | 4, offset, |
| 229 | ]) |
| 230 | this.controlledPush(pathAsBuffer) |
| 231 | this.controlledPush(extraData) |
| 232 | } |
| 233 | const after = this.dataWritten |
| 234 | let centralDirSize = after - centralDirOffset |
| 235 | let n = this.entries.length |
| 236 | if (n >= ZIP64_NUMBER_LIMIT |
| 237 | || centralDirOffset >= ZIP64_SIZE_LIMIT |
| 238 | || centralDirSize >= ZIP64_SIZE_LIMIT) { |
| 239 | this.controlledPush([ |
| 240 | 4, 0x06064b50, // end of central dir zip64 |
| 241 | 8, 44, |
| 242 | 2, 45, |
| 243 | 2, 45, |
| 244 | 4, 0, |
| 245 | 4, 0, |
| 246 | 8, n, |
| 247 | 8, n, |
| 248 | 8, centralDirSize, |
| 249 | 8, centralDirOffset, |
| 250 | ]) |
| 251 | this.controlledPush([ |
| 252 | 4, 0x07064b50, |
| 253 | 4, 0, |
| 254 | 8, after, |
| 255 | 4, 1, |
| 256 | ]) |
| 257 | centralDirOffset = ZIP64_SIZE_LIMIT |
| 258 | centralDirSize = ZIP64_SIZE_LIMIT |
| 259 | n = ZIP64_NUMBER_LIMIT |
| 260 | } |
| 261 | this.controlledPush([ |
| 262 | 4, 0x06054b50, // end of central directory signature |
| 263 | 4, 0, // disk-related stuff |
| 264 | 2, n, |
| 265 | 2, n, |
| 266 | 4, centralDirSize, |
| 267 | 4, centralDirOffset, |
| 268 | 2, 0, // comment length |
| 269 | ]) |
| 270 | this.closeStream() |
| 271 | } |
| 272 | } |
| 273 | |
| 274 | function buffer(pairs: number[]) { |
| 275 | assert(pairs.length % 2 === 0) |
| 276 | let total = 0 |
| 277 | for (let i=0; i < pairs.length; i+=2) |
| 278 | total += pairs[i]! |
| 279 | const ret = Buffer.alloc(total, 0) |
| 280 | let offset = 0 |
| 281 | let i = 0 |
| 282 | while (i < pairs.length) { |
| 283 | const size = pairs[i++] |
| 284 | const data = pairs[i++]! |
| 285 | if (size === 1) |
| 286 | ret.writeUInt8(data, offset) |
| 287 | else if (size === 2) |
| 288 | ret.writeUInt16LE(data, offset) |
| 289 | else if (size === 4) |
| 290 | if (data < 0) // needed for crc32 |
| 291 | ret.writeInt32LE(data, offset) |
| 292 | else |
| 293 | ret.writeUInt32LE(data, offset) |
| 294 | else if (size === 8) |
| 295 | ret.writeBigUInt64LE(BigInt(data), offset) |
| 296 | else |
| 297 | throw 'unsupported' |
| 298 | offset += size |
| 299 | } |
| 300 | return ret |
| 301 | } |
| 302 | |
| 303 | function ts2buf(ts:Date) { |
| 304 | const date = ((ts.getFullYear() - 1980) & 0x7F) << 9 | (ts.getMonth() + 1) << 5 | ts.getDate() |
| 305 | const time = ts.getHours() << 11 | ts.getMinutes() << 5 | (ts.getSeconds() / 2) & 0x0F |
| 306 | return [ |
| 307 | 2, time, |
| 308 | 2, date, |
| 309 | ] |
| 310 | } |
| 311 | |
| 312 | interface CrcCacheEntry { ts: Date, crc: number } |
| 313 | const crcCache: Record<string, CrcCacheEntry> = {} |