|
| 1 | +import { Agent as SecretStreamAgent } from 'secret-stream-http' |
| 2 | +import z32 from 'z32' |
| 3 | + |
| 4 | +import { TypedEventTarget } from '../lib/event-target.js' |
| 5 | +import type { DownloadCreateRequest } from '../routes/downloads.js' |
| 6 | +import { type DownloadStateUpdate } from '../types.js' |
| 7 | +import { StatusError } from './errors.js' |
| 8 | +import { errors, jsonError } from './errors.js' |
| 9 | +import { secretStreamFetch } from './secret-stream-fetch.js' |
| 10 | +import { StateUpdateEvent } from './state-update-event.js' |
| 11 | +import { addTrailingSlash, generateId, getErrorCode, noop } from './utils.js' |
| 12 | + |
| 13 | +type DownloadRequestState = DownloadStateUpdate & |
| 14 | + Omit<DownloadCreateRequest, 'mapShareUrls'> & { downloadId: string } |
| 15 | + |
| 16 | +export class DownloadRequest extends TypedEventTarget< |
| 17 | + InstanceType<typeof StateUpdateEvent<DownloadStateUpdate>> |
| 18 | +> { |
| 19 | + #state: DownloadRequestState |
| 20 | + #abortController = new AbortController() |
| 21 | + #transform = new TransformStream({ |
| 22 | + transform: (chunk, controller) => { |
| 23 | + if (this.#state.status !== 'downloading') { |
| 24 | + throw new Error('Download has been cancelled or encountered an error') |
| 25 | + } |
| 26 | + this.#updateState({ |
| 27 | + status: 'downloading', |
| 28 | + bytesDownloaded: this.#state.bytesDownloaded + chunk.byteLength, |
| 29 | + }) |
| 30 | + controller.enqueue(chunk) |
| 31 | + }, |
| 32 | + }) |
| 33 | + #dispatcher: SecretStreamAgent |
| 34 | + |
| 35 | + constructor( |
| 36 | + stream: WritableStream<Uint8Array>, |
| 37 | + { mapShareUrls, ...rest }: DownloadCreateRequest, |
| 38 | + keyPair: { publicKey: Uint8Array; secretKey: Uint8Array }, |
| 39 | + ) { |
| 40 | + super() |
| 41 | + this.#state = { |
| 42 | + ...rest, |
| 43 | + status: 'downloading', |
| 44 | + bytesDownloaded: 0, |
| 45 | + downloadId: generateId(), |
| 46 | + } |
| 47 | + let remotePublicKey: Uint8Array |
| 48 | + try { |
| 49 | + remotePublicKey = z32.decode(this.#state.senderDeviceId) |
| 50 | + } catch { |
| 51 | + throw new errors.INVALID_SENDER_DEVICE_ID( |
| 52 | + `Invalid sender device ID: ${this.#state.senderDeviceId}`, |
| 53 | + ) |
| 54 | + } |
| 55 | + if (remotePublicKey.length !== 32) { |
| 56 | + throw new errors.INVALID_SENDER_DEVICE_ID( |
| 57 | + `Invalid sender device ID: ${this.#state.senderDeviceId}`, |
| 58 | + ) |
| 59 | + } |
| 60 | + this.#dispatcher = new SecretStreamAgent({ remotePublicKey, keyPair }) |
| 61 | + this.#start({ mapShareUrls, stream, remotePublicKey, keyPair }).catch( |
| 62 | + async (error) => { |
| 63 | + // In case the error happens before we pipe to the stream, we need to abort the stream |
| 64 | + stream.abort().catch(noop) |
| 65 | + if (error.name === 'AbortError') { |
| 66 | + this.#updateState({ status: 'aborted' }) |
| 67 | + } else if (getErrorCode(error) === 'DOWNLOAD_SHARE_CANCELED') { |
| 68 | + this.#updateState({ status: 'canceled' }) |
| 69 | + } else if (getErrorCode(error)) { |
| 70 | + // Specific known error from the server |
| 71 | + this.#updateState({ status: 'error', error }) |
| 72 | + } else { |
| 73 | + // Once the download has started, the sender can only close the |
| 74 | + // connection to cancel the download, which we only see as an |
| 75 | + // ECONNRESET error here, which could happen for multiple reasons. |
| 76 | + // Rather than immediately updating the state to error, we first check |
| 77 | + // with the sender to see if we can access the status of the share, |
| 78 | + // namely whether it was canceled, or if a different error occurred on |
| 79 | + // the server side. |
| 80 | + try { |
| 81 | + const response = await secretStreamFetch(mapShareUrls, { |
| 82 | + dispatcher: this.#dispatcher, |
| 83 | + signal: AbortSignal.timeout(2000), |
| 84 | + }) |
| 85 | + const json = await response.json() |
| 86 | + if (json.status) { |
| 87 | + this.#updateState({ status: json.status, error: json.error }) |
| 88 | + return |
| 89 | + } |
| 90 | + } catch (err) { |
| 91 | + // Ignore errors from checking the status and update state with original error |
| 92 | + } |
| 93 | + this.#updateState({ status: 'error', error: jsonError(error) }) |
| 94 | + } |
| 95 | + }, |
| 96 | + ) |
| 97 | + } |
| 98 | + |
| 99 | + async #start({ |
| 100 | + mapShareUrls, |
| 101 | + stream, |
| 102 | + remotePublicKey, |
| 103 | + keyPair, |
| 104 | + }: { |
| 105 | + mapShareUrls: string[] |
| 106 | + stream: WritableStream<Uint8Array> |
| 107 | + remotePublicKey: Uint8Array |
| 108 | + keyPair: { publicKey: Uint8Array; secretKey: Uint8Array } |
| 109 | + }) { |
| 110 | + const downloadUrls = mapShareUrls.map( |
| 111 | + (baseUrl) => new URL('download', addTrailingSlash(baseUrl)), |
| 112 | + ) |
| 113 | + const response = await secretStreamFetch(downloadUrls, { |
| 114 | + dispatcher: this.#dispatcher, |
| 115 | + }) |
| 116 | + if (!response.body) { |
| 117 | + throw new errors.DOWNLOAD_ERROR('Could not connect to map share sender') |
| 118 | + } |
| 119 | + if (!response.ok) { |
| 120 | + throw new StatusError(response.status, await response.json()) |
| 121 | + } |
| 122 | + if (this.#abortController.signal.aborted) { |
| 123 | + response.body.cancel().catch(noop) |
| 124 | + throw new DOMException('Download aborted', 'AbortError') |
| 125 | + } |
| 126 | + await response.body.pipeThrough(this.#transform).pipeTo(stream, { |
| 127 | + signal: this.#abortController.signal, |
| 128 | + }) |
| 129 | + this.#updateState({ status: 'completed' }) |
| 130 | + } |
| 131 | + |
| 132 | + get state() { |
| 133 | + return this.#state |
| 134 | + } |
| 135 | + |
| 136 | + cancel() { |
| 137 | + this.#abortController.abort() |
| 138 | + } |
| 139 | + |
| 140 | + #updateState(update: DownloadStateUpdate) { |
| 141 | + this.#state = { ...this.#state, ...update } |
| 142 | + this.dispatchEvent(new StateUpdateEvent(update)) |
| 143 | + } |
| 144 | +} |
0 commit comments