path: root/packages/meshbay-hub/src/meshbay_hub/static/transfers.js

/**
 * Transfers that outlive the page that started them.
 *
 * Downloads and uploads used to be state inside GroupPage, which meant leaving
 * a group killed them — the component unmounted, its effect closed the
 * DataChannel, and a half-written file was all you had. They live here instead:
 * a module-level store that nothing unmounts, with the group page as one of
 * several possible views onto it.
 *
 * Two consequences worth stating, because they are the reason this exists:
 *
 *   - The transport cannot be closed just because a page went away. A group
 *     page hands its transport over with `releaseWhenIdle()`, and the last
 *     transfer using it closes it.
 *   - Signing out is different from navigating. It cancels everything and
 *     closes what it was using, because the tokens those transfers are running
 *     on are about to stop being ours.
 *
 * No browser globals: exercised under Node by
 * packages/meshbay-hub/tests/test_transfers.py.
 */

const SPEED_WINDOW_MS = 5000;

/** Not finished: still preparing, waiting for a slot, or transferring. One
 *  definition, because six places ask and they were drifting apart. */
function _live(status) {
  return status === 'preparing' || status === 'queued' || status === 'running'
    || status === 'paused';
}

/** Raised by `run` when it stopped because the transfer was paused. */
function _pausedError() {
  const err = new Error('Paused');
  err.name = 'PausedError';
  return err;
}

function _abortError() {
  const err = new Error('Cancelled');
  err.name = 'AbortError';
  return err;
}

let _nextId = 1;

export class TransferStore {
  constructor(now = () => Date.now()) {
    this._now = now;
    this._items = [];
    this._subs = new Set();
    this._releasing = new Set();
  }

  subscribe(fn) {
    this._subs.add(fn);
    return () => this._subs.delete(fn);
  }

  _emit() {
    for (const fn of this._subs) fn(this.list());
  }

  /**
   * What a view needs to render, as plain data — never the internals, so a
   * render cannot accidentally hold a transport alive.
   */
  list() {
    return this._items.map(it => ({
      id: it.id,
      kind: it.kind,
      name: it.name,
      total: it.total,
      done: it.done,
      status: it.status,
      // How many are in front of this one, and whose limit is holding it up:
      // "your own two slots are busy" and "the node is full" are different
      // situations and the person can act on only one of them.
      ahead: it.ahead || 0,
      queuedByOwnLimit: Boolean(
        it.lease && it.lease.cap && it.lease.used >= it.lease.cap),
      error: it.error || '',
      speed: this._speed(it),
      // The ETA is drawn only once the window holds a few seconds of real
      // measurement -- see etaSeconds.
      settled: it.samples.length > 2
        && (it.samples[it.samples.length - 1].t - it.samples[0].t) >= 3000,
      percent: it.total ? Math.min(100, Math.round(it.done / it.total * 100)) : 0,
      // Only for a file written into a folder the browser granted us: that is
      // the one case where the page can read its own download back.
      canOpen: it.status === 'done' && typeof it.open === 'function',
      // Whether the target can be stopped and continued. False is the honest
      // answer for a service-worker stream, and the button is not drawn.
      pausable: Boolean(it.pausable),
    }));
  }

  get active() {
    return this._items.filter(it => it.status === 'running').length;
  }

  /** Running or waiting for a slot — what the nav badge counts. */
  get pending() {
    return this._items.filter(it => _live(it.status)).length;
  }

  _speed(it) {
    // Over a window rather than since the start: a transfer that stalls should
    // read as slow immediately, not as its own historical average.
    const s = it.samples;
    if (s.length < 2) return 0;
    const dt = (s[s.length - 1].t - s[0].t) / 1000;
    if (dt <= 0) return 0;
    return (s[s.length - 1].done - s[0].done) / dt;
  }

  /**
   * Start a transfer.
   *
   * `run` receives `{ signal, onProgress }`. It must poll `signal.aborted` — a
   * cancel that only sets a flag nobody reads is a button that lies.
   */
  /**
   * Start a transfer.
   *
   * `run` receives `{ signal, onProgress, lease }`. It must poll
   * `signal.aborted` — a cancel that only sets a flag nobody reads is a button
   * that lies.
   *
   * `prepare` is optional and runs before anything else, with the row already
   * on screen. It is where a download opens its target, which can take tens of
   * seconds — the streamed path waits for the worker, twice, and a Save As
   * dialog waits for a person. Doing that *before* creating the row meant three
   * clicks produced no panel at all, not even the icon, and then several rows
   * at once. Returning `false` drops the row again, which is what a dismissed
   * dialog should look like: nothing, rather than a cancelled transfer nobody
   * started.
   *
   * `makeLease` is called after `prepare` succeeds, never before. A granted
   * slot must be taken up within the node's deadline, so it is asked for once
   * there is somewhere to write — see file-utils.js's downloadEntry.
   */
  start({ kind, name, total = 0, transport = null, run, open = null,
          lease = null, prepare = null, makeLease = null, pausable = false }) {
    const item = {
      id: _nextId++,
      kind, name, total, transport, open, lease,
      done: 0,
      // A transfer that has to wait for a slot starts as 'queued', not
      // 'running'. Two different things are true of it — nothing is moving, and
      // nothing is wrong — and a status that conflates them is what makes a
      // queue look like a hang.
      status: prepare ? 'preparing'
        : (lease && lease.state !== 'granted' ? 'queued' : 'running'),
      ahead: (lease && lease.ahead) || 0,
      error: '',
      samples: [{ t: this._now(), done: 0 }],
      signal: { aborted: false, paused: false },
      // Whether this transfer can be stopped and continued. A download learns
      // it from `prepare`, because only its target knows; an upload says so
      // outright, because a `File` is always seekable and the node keeps the
      // position (see uploads.py).
      pausable: Boolean(pausable),
      // Where a resumed run picks up, in chunks. Zero until something pauses.
      resumeFrom: 0,
      // Resolved by resume(); awaited by the run loop while paused.
      resumed: null,
      _wake: null,
      // Pausing gives the slot back, so resuming has to be able to ask for
      // another one. A transfer handed a lease directly cannot, and must not
      // be offered a button that would drop its slot for good.
      _canRelease: Boolean(makeLease),
    };
    this._items.push(item);
    this._emit();

    const onProgress = (done, total) => {
      item.done = done;
      if (total) item.total = total;
      const t = this._now();
      item.samples.push({ t, done });
      while (item.samples.length > 2 && t - item.samples[0].t > SPEED_WINDOW_MS) {
        item.samples.shift();
      }
      this._emit();
    };

    const finish = (status, error = '') => {
      item.status = status;
      item.error = error;
      this._emit();
      this._maybeRelease(item.transport);
    };

    // The slot is given back in a `finally` around everything, so it survives
    // a throw, a cancel and a return alike. A slot not returned is a member who
    // cannot start another transfer until the node times it out.
    const finished = () => {
      if (item.lease) item.lease.release(
        item.signal.aborted ? 'cancelled' : 'done');
    };

    // Installed here and not inside the promise chain below. A state push that
    // arrived before the first microtask ran was simply dropped, so a transfer
    // could sit at the position it was given when it was created and never
    // appear to move — the widget showing "3 ahead" for ever while the node
    // quietly worked through the queue. Nothing about that looks wrong from
    // either side, which is why it needs a test rather than a reading.
    if (item.lease) this._watchLease(item);

    const promise = Promise.resolve()
      .then(async () => {
        if (prepare) {
          const ready = await prepare();
          if (item.signal.aborted) throw _abortError();
          if (ready === false) {
            // Dismissed. Not a failure and not a cancellation: nothing was ever
            // started, so nothing should be left on screen to explain.
            this._drop(item.id);
            return undefined;
          }
          if (ready && ready.name) item.name = ready.name;
          // Only the target knows. A service-worker stream is already an HTTP
          // response the browser is writing to its own download folder: not
          // writing to it stalls that download outside our control, and an idle
          // worker is terminated within seconds, taking the stream with it. So
          // the button is offered where it works and nowhere else — a pause
          // that silently restarts from zero is worse than no pause.
          if (ready && ready.pausable) item.pausable = true;
          item.status = 'running';
          this._emit();
        }
        // Run, and be prepared to be stopped and started again.
        //
        // A paused transfer holds **nothing**: its slot goes back to the node
        // and resuming rejoins the queue at the tail. Anything else lets one
        // member close a node by pausing four downloads and going to lunch.
        // So the lease is taken inside this loop, not before it.
        for (;;) {
          if (makeLease && !item.lease) {
            item.lease = makeLease();
            this._watchLease(item);
            if (item.lease.state !== 'granted') {
              item.status = 'queued';
              item.ahead = item.lease.ahead || 0;
              this._emit();
            }
          }
          if (item.lease) {
            await item.lease.acquire();
            if (item.signal.aborted) throw _abortError();
            item.status = 'running';
            this._emit();
          }
          try {
            return await run({ signal: item.signal, onProgress,
                               lease: item.lease, from: item.resumeFrom || 0 });
          } catch (err) {
            if (err.name !== 'PausedError') throw err;
          }
          // Where to pick up. `run` records it on the signal rather than
          // returning it, because it has to survive being thrown past.
          item.resumeFrom = item.signal.resumeFrom || 0;
          if (item.lease) {
            item.lease.release('paused');
            item.lease = null;
          }
          item.status = 'paused';
          item.ahead = 0;
          this._emit();
          this._maybeRelease(item.transport);
          await item.resumed;
          if (item.signal.aborted) throw _abortError();
        }
      })
      .then(() => {
        if (item.signal.aborted) finish('cancelled');
        else {
          if (item.total) item.done = item.total;
          finish('done');
        }
      })
      .catch(err => {
        if (item.signal.aborted || err.name === 'AbortError') finish('cancelled');
        else finish('failed', err.message || String(err));
      })
      .finally(finished);

    item.promise = promise;
    return item.id;
  }

  _watchLease(item) {
    item.lease._onState = (lease) => {
      if (item.status !== 'queued' && item.status !== 'running') return;
      item.ahead = lease.ahead;
      item.status = lease.state === 'granted' ? 'running' : 'queued';
      this._emit();
    };
  }

  /** Remove a row entirely. Only for a transfer that never started. */
  _drop(id) {
    this._items = this._items.filter(it => it.id !== id);
    this._emit();
  }

  /**
   * Hand a finished download to the browser to display.
   *
   * As close to "open it" as a web page gets: the bytes go to a new tab and the
   * browser decides what to do with them. A page cannot start a desktop
   * application, and cannot show a file manager — there is no API for either,
   * in any browser, by design.
   */
  open(id) {
    const item = this._items.find(it => it.id === id);
    if (item && typeof item.open === 'function') return item.open();
  }

  /**
   * Stop a running transfer, keeping what it has already written.
   *
   * Only while running: a queued transfer is already stopped and holds no slot,
   * and pausing it would only cost it its place. Only where the target can do
   * it — see the note in `start`.
   *
   * The slot goes back to the node at once (§6.2 of the plan): a paused
   * transfer holds nothing, and resuming rejoins the queue at the tail.
   */
  pause(id) {
    const item = this._items.find(it => it.id === id);
    if (!item || !item.pausable || !item._canRelease
        || item.status !== 'running') return;
    item.signal.paused = true;
    // Created here rather than in resume(): the run loop awaits it the moment
    // `run` throws, which can be sooner than the next call into this store.
    item.resumed = new Promise((resolve) => { item._wake = resolve; });
    this._emit();
  }

  /** Start it again, from where it stopped, behind whatever is waiting now. */
  resume(id) {
    const item = this._items.find(it => it.id === id);
    if (!item || item.status !== 'paused') return;
    item.signal.paused = false;
    item.status = 'queued';
    this._emit();
    if (item._wake) { item._wake(); item._wake = null; }
  }

  cancel(id) {
    const item = this._items.find(it => it.id === id);
    // 'queued' too: a transfer waiting for a slot is exactly the one somebody
    // is most likely to give up on, and its queue entry has to go with it or
    // the node grants a slot to a transfer that will never use it.
    if (!item || !_live(item.status)) return;
    item.signal.aborted = true;
    if (item.lease) item.lease.release('cancelled');
    // A paused run is parked on `item.resumed`. Without this it stays parked
    // for the life of the page, holding its target open, and the row says
    // "cancelled" over a download that never stopped.
    if (item._wake) { item._wake(); item._wake = null; }
    // Marked at once. The work stops when it next looks, but a cancelled
    // transfer should not keep reporting progress in the meantime.
    item.status = 'cancelled';
    this._emit();
    this._maybeRelease(item.transport);
  }

  cancelAll() {
    for (const it of this._items) {
      if (_live(it.status)) this.cancel(it.id);
    }
  }

  /** Drop everything finished, keeping what is still running or waiting. */
  clearFinished() {
    this._items = this._items.filter(it => _live(it.status));
    this._emit();
  }

  _busy(transport) {
    // Queued counts as busy: a transport closed while a transfer waits for a
    // slot can never be granted one, and the transfer would sit at "waiting"
    // for ever with nothing left to answer it.
    return this._items.some(
      it => it.transport === transport && _live(it.status));
  }

  /**
   * The group page is going away. Close its transport once nothing is using it,
   * which may be now or may be in twenty minutes.
   */
  releaseWhenIdle(transport) {
    if (!transport) return;
    this._releasing.add(transport);
    this._maybeRelease(transport);
  }

  _maybeRelease(transport) {
    if (!transport || !this._releasing.has(transport)) return;
    if (this._busy(transport)) return;
    this._releasing.delete(transport);
    try {
      transport.onIndexSync = null;
      transport.close();
    } catch { /* already gone */ }
  }

  /** Signing out: stop everything and let go of every transport. */
  reset() {
    this.cancelAll();
    for (const transport of [...this._releasing]) {
      this._releasing.delete(transport);
      try { transport.close(); } catch { /* already gone */ }
    }
    for (const it of this._items) {
      if (it.transport) {
        try { it.transport.close(); } catch { /* already gone */ }
      }
    }
    this._items = [];
    this._emit();
  }
}

export const transfers = new TransferStore();

/**
 * Seconds left, or null when saying nothing is the honest answer.
 *
 * Withheld until the speed window has real samples in it: a figure computed
 * from the first two chunks of a transfer swings between "4 seconds" and "an
 * hour" and back, and a number that behaves like that is worse than a blank —
 * people read the first one they see and plan around it.
 */
export function etaSeconds(item) {
  if (item.status !== 'running' || !item.total || !item.speed) return null;
  const left = item.total - item.done;
  if (left <= 0) return null;
  const secs = left / item.speed;
  return Number.isFinite(secs) ? secs : null;
}

/** Human-readable rate, for a widget that updates several times a second. */
export function formatSpeed(bytesPerSecond) {
  if (!bytesPerSecond || bytesPerSecond < 1) return '';
  if (bytesPerSecond < 1024 * 1024) return `${Math.round(bytesPerSecond / 1024)} KB/s`;
  return `${(bytesPerSecond / (1024 * 1024)).toFixed(1)} MB/s`;
}