strimserve/src/lib/streamCache.svelte.ts

import { browser } from '$app/environment';
import { SvelteSet, SvelteMap } from 'svelte/reactivity';
import { readStored, writeStored } from './utils.ts';
import type { StreamSummary, PartialDownload } from './types.ts';

const STORAGE_KEY = 'cachedStreams';
const PARTIALS_KEY = 'partialStreams';
const PARTS_DIR = 'parts';

// MAX_ATTEMPTS is consecutive attempts that acquired zero chunks, so the exponential
// backoff below tolerates roughly 8 minutes of outage before giving up
const MAX_ATTEMPTS = 8;

const CHUNK_SIZE = 4 * 1024 * 1024;

// 60s as the limit for downloading one chunk is a reasonable guess
// as the intent is to catch socket breakage during cellular network changes for example
const CHUNK_TIMEOUT = 60_000;

class QuotaError extends Error {
    constructor(readonly missing: number) {
        super();
    }
}

// total bytes used by our OPFS origin
export const usage = $state<{ bytes?: number }>({});

export const cached = new SvelteSet<string>(readStored<string[]>(STORAGE_KEY, []));
export const downloading = new SvelteMap<string, number>();

// bytes already written to OPFS for each incomplete download
export const partials = new SvelteMap<string, PartialDownload>(
    Object.entries(readStored<Record<string, PartialDownload>>(PARTIALS_KEY, {}))
);

if (browser) {
    refreshUsage();
    $effect.root(() => {
        $effect(() => {
            writeStored(STORAGE_KEY, [...cached]);
        });
        $effect(() => {
            writeStored(PARTIALS_KEY, Object.fromEntries(partials));
        });
    });
}

export async function download(stream: StreamSummary) {
    if (downloading.has(stream.id) || cached.has(stream.id)) return;

    const resumed = partials.get(stream.id);
    downloading.set(stream.id, percent(resumed?.received ?? 0, resumed?.total ?? 0));

    try {
        let attempt = 0;
        while (true) {
            const before = partials.get(stream.id)?.received ?? 0;
            try {
                await attemptDownload(stream);
                partials.delete(stream.id);
                cached.add(stream.id);
                return;
            } catch (err) {
                if (err instanceof QuotaError) {
                    alert(
                        `not enough disk space ;_;, need ${Math.ceil(err.missing / 1_000_000)} MB more.`
                    );
                    return;
                }
                console.error(`[streamCache] ${stream.id} attempt failed`, err);
                // reset attempts if download of at least one chunk successful
                attempt = (partials.get(stream.id)?.received ?? 0) > before ? 0 : attempt + 1;
                if (attempt === MAX_ATTEMPTS) return;
            }
            await waitForRetry(attempt);
        }
    } finally {
        downloading.delete(stream.id);
    }
}

export async function remove(streamId: string) {
    cached.delete(streamId);
    await discard(streamId);
}

export async function getUrl(streamId: string): Promise<string | null> {
    if (!cached.has(streamId)) return null;
    try {
        const root = await navigator.storage.getDirectory();
        const handle = await root.getFileHandle(streamId);
        const file = await handle.getFile();
        return URL.createObjectURL(file);
    } catch {
        cached.delete(streamId);
        return null;
    }
}

async function attemptDownload(stream: StreamSummary) {
    let state = partials.get(stream.id);
    let checkedSpace = false;
    while (!state || state.received < state.total) {
        state = await fetchChunk(stream, state);
        partials.set(stream.id, state);
        downloading.set(stream.id, percent(state.received, state.total));

        // first chunk's HTTP response carries the total stream size
        if (!checkedSpace) {
            await checkSpace(stream.id, state);
            checkedSpace = true;
        }
    }
    console.log(`[streamCache] ${stream.id} assembling ${state.total} bytes`);
    await assemble(stream.id, state.total);
    console.log(`[streamCache] ${stream.id} assembled`);
}

/** check available space below what's needed to download the stream */
async function checkSpace(streamId: string, state: PartialDownload) {
    const { usage: used, quota } = await navigator.storage.estimate();
    if (used === undefined || quota === undefined) return;

    // extra CHUNK_SIZE is due to the delta in `assemble()` merging one chunk in (+1) then deleting it
    const peak = used + (state.total - state.received) + CHUNK_SIZE;
    if (peak > quota) throw new QuotaError(peak - quota);
}

/** fetch chunk starting from the last valid one, store in OPFS, and return updated state */
async function fetchChunk(stream: StreamSummary, state?: PartialDownload) {
    const start = state?.received ?? 0;
    const headers = new Headers({ Range: `bytes=${start}-${start + CHUNK_SIZE - 1}` });
    // attach existing ETag to the request
    if (state?.etag) headers.set('If-Range', state.etag);

    const res = await fetch(`/media/tracks/${stream.filename}`, {
        headers,
        signal: AbortSignal.timeout(CHUNK_TIMEOUT)
    });

    // 200 means the range was ignored or the file changed (different ETag)
    // 416 that our start offset is past its end
    // anything else can be transient, so the chunks already on disk are kept
    if (res.status !== 206) {
        if (res.status === 200 || res.status === 416) await discard(stream.id);
        throw new Error(`Download failed: ${res.status}`);
    }

    const total = Number(res.headers.get('content-range')?.split('/')[1]);
    if (!total) throw new Error('Malformed Content-Range');

    const etag = res.headers.get('etag') ?? res.headers.get('last-modified');
    if (state && state.total !== total) {
        await discard(stream.id);
        throw new Error('File changed while downloading');
    }

    const bytes = await res.arrayBuffer();
    const dir = await partsDir(stream.id, true);
    const handle = await dir.getFileHandle(String(start), { create: true });
    const writable = await handle.createWritable();
    await writable.write(bytes);
    await writable.close();

    return { received: start + bytes.byteLength, total, etag };
}

/** concatenate stream file once all the parts are downloaded */
async function assemble(streamId: string, total: number) {
    const root = await navigator.storage.getDirectory();
    const dir = await partsDir(streamId, true);
    const handle = await root.getFileHandle(streamId, { create: true });

    // track size from previous assembling attempt if there was one
    const done = (await handle.getFile()).size;

    const writable = await handle.createWritable({ keepExistingData: done > 0 });
    await writable.seek(done);

    // check at each step the chunk names (ranges) are coherent
    let written = done;
    while (written < total) {
        const name = String(written);
        const part = await dir.getFileHandle(name).catch(() => null);
        if (!part) break;
        const file = await part.getFile();
        if (!file.size) break;
        await writable.write(file);
        written += file.size;
        await dir.removeEntry(name).catch(() => {});
    }
    await writable.close();

    partials.delete(streamId);
    await sweep(streamId);
    // chunks went missing, shit's fucked...
    if (written !== total) throw new Error(`Assembled ${written} of ${total} bytes`);
}

/** returns directory handle dedicated to partial downloads of given stream id */
async function partsDir(streamId: string, create = false) {
    const root = await navigator.storage.getDirectory();
    const parts = await root.getDirectoryHandle(PARTS_DIR, { create });
    return parts.getDirectoryHandle(streamId, { create });
}

/** discard everything on disk for a stream that is not fully downloaded */
async function discard(streamId: string) {
    partials.delete(streamId);
    const root = await navigator.storage.getDirectory();
    await root.removeEntry(streamId).catch(() => {});
    await sweep(streamId);
}

/** remove unreferenced chunk directories (that aren't still downloading)
 *  `force` is there for `discard()` to use as an escape hatch, otherwise it'd be caught in
 *  the guard condition due to still being in `downloading` by then
 */
async function sweep(force?: string) {
    const root = await navigator.storage.getDirectory();
    const parts = await root.getDirectoryHandle(PARTS_DIR, { create: true });
    const orphans: string[] = [];
    for await (const name of parts.keys()) {
        if (partials.has(name)) continue;
        if (downloading.has(name) && name !== force) continue;
        orphans.push(name);
    }
    for (const name of orphans) await parts.removeEntry(name, { recursive: true }).catch(() => {});

    await refreshUsage();
}

async function refreshUsage() {
    usage.bytes = (await navigator.storage.estimate()).usage;
}

/** exponential backoff, cut short if the connection comes back before it elapses */
function waitForRetry(attempt: number) {
    return new Promise<void>((resolve) => {
        const done = () => {
            removeEventListener('online', done);
            clearTimeout(timer);
            resolve();
        };
        const timer = setTimeout(done, 2 ** attempt * 1000);
        addEventListener('online', done);
    });
}

function percent(received: number, total: number) {
    return total ? Math.round((received / total) * 100) : 0;
}