Source

// The receiving device: joins the share from its link, connects to the sender, asks for files one
// at a time, keeps them in this browser's storage as they arrive, checks each one with SHA-256 if
// asked, and hands it to the browser's download. It reconnects on its own and picks up where it was.
import { get, set, subscribe, serverTimestamp } from "@bounded-sh/client";
import { sha256 } from "@noble/hashes/sha2";
import { bytesToHex } from "@noble/hashes/utils";
import { me } from "./lib/bounded";
import { gathered, iceConfig, randomId, readRoute, type Route } from "./lib/rtc";
import { parse, type FileInfo, type FromReceiver, type FromSender } from "./lib/protocol";
import { openSink, type Sink } from "./lib/sink";
import { Rate } from "./lib/format";
import { deviceName, isIOS } from "./lib/device";

export type LinkState =
  | "joining"
  | "offline"      // could not reach Drop's server to join
  | "connecting"
  | "live"
  | "reconnecting" // the connection dropped; trying again
  | "noanswer"     // the sender's page is closed, asleep or offline
  | "blocked"      // the sender answers, but the two devices cannot reach each other
  | "ended"        // the sender stopped sharing
  | "missing"      // the link is wrong or no longer open
  | "self";        // this is the sending device

export type FileState = "offered" | "queued" | "receiving" | "checking" | "done" | "failed";

export interface FileView {
  info: FileInfo;
  state: FileState;
  received: number;
  check: "ok" | null;
  error: string | null;
  url: string | null;
  file: File | null;
  saved: boolean;
}

export interface ReceiveSnapshot {
  state: LinkState;
  from: string;
  files: FileView[];
  got: number;
  total: number;
  done: number;
  count: number;
  rate: number;
  route: Route | null;
  verify: boolean;
  /** Some file is being kept in memory, which limits very large files. */
  inMemory: boolean;
}

interface Incoming extends FileView {
  sink: Sink | null;
  writes: Promise<void>;
  pending: ArrayBuffer[];
  pendingBytes: number;
  hasher: ReturnType<typeof sha256.create> | null;
}

const RETRY_MS = [1000, 2000, 4000, 8000, 15000, 30000];
const NOANSWER_RETRY_MS = [5000, 10000, 20000, 30000, 60000];
/** After this many tries in a row the receiver stops on its own and offers Try again. */
const MAX_TRIES = 10;
const FLUSH = 1024 * 1024;
const VERIFY_KEY = "drop:verify";

/**
 * The type the sender claims is kept only for photos, videos and audio (so Share can save them to
 * the gallery). Anything else is plain bytes: a browser that opens the file instead of downloading
 * it must never run a sender's HTML or SVG on Drop's own address.
 */
function safeType(type: unknown): string {
  const t = String(type || "").toLowerCase();
  return /^(image|video|audio)\/[a-z0-9.-]+$/.test(t) && !/svg|xml/.test(t) ? t : "application/octet-stream";
}

export class Receiver {
  private me = "";
  private pc: RTCPeerConnection | null = null;
  private dc: RTCDataChannel | null = null;
  private session: string | null = null;
  private offerId = "";
  private answered = false;
  private attempt = 0;
  private failures = 0;
  private timers: number[] = [];
  private ticker: number | undefined;
  private stops: (() => Promise<void>)[] = [];
  private listeners = new Set<() => void>();
  private stopped = false;
  private state: LinkState = "joining";
  private from = "";
  private files = new Map<number, Incoming>();
  /** The file whose bytes are coming in on the current connection. */
  private streaming: number | null = null;
  private sums = new Map<number, string>();
  private sumWaiters = new Map<number, (hex: string) => void>();
  private bytesIn = 0;
  private rate = new Rate();
  private speed = 0;
  private route: Route | null = null;
  private verify: boolean;
  private dirty = false;
  private snap: ReceiveSnapshot;

  constructor(readonly dropId: string) {
    let v = true;
    try { v = localStorage.getItem(VERIFY_KEY) !== "off"; } catch { /* default on */ }
    this.verify = v;
    this.snap = this.build();
  }

  async start() {
    this.setState("joining");
    try {
      this.me = await me();
      const joinPath = `drops/${this.dropId}/joins/${this.me}`;
      const joined = await get(joinPath);
      if (joined?.role === "sender") return this.setState("self");
      if (!joined) {
        try {
          await set(joinPath, { role: "receiver", name: deviceName(), at: serverTimestamp() });
        } catch {
          return this.setState("missing");
        }
      }
      if (this.stopped) return;
      this.stops.push(await subscribe(`drops/${this.dropId}/links/${this.me}`, { onData: (d: any) => void this.onLink(d) }));
      this.stops.push(await subscribe(`drops/${this.dropId}`, { onData: (d: any) => this.onDrop(d) }));
    } catch {
      for (const s of this.stops.splice(0)) await s();
      if (this.stopped) return;
      this.setState("offline");
      window.addEventListener("online", () => this.retryNow(), { once: true });
      return;
    }
    const wake = () => { if (document.visibilityState === "visible" && ["reconnecting", "noanswer", "blocked"].includes(this.state)) this.retryNow(); };
    document.addEventListener("visibilitychange", wake);
    window.addEventListener("online", wake);
    this.stops.push(async () => { document.removeEventListener("visibilitychange", wake); window.removeEventListener("online", wake); });
    this.ticker = window.setInterval(() => this.tick(), 400);
  }

  async stop() {
    this.stopped = true;
    this.clearTimers();
    clearInterval(this.ticker);
    this.closePc();
    this.dropLink();
    for (const s of this.stops) await s();
    for (const f of this.files.values()) {
      if (f.url) URL.revokeObjectURL(f.url);
      if (f.state !== "done") void f.sink?.discard();
    }
  }

  subscribe = (fn: () => void) => {
    this.listeners.add(fn);
    return () => { this.listeners.delete(fn); };
  };
  snapshot = () => this.snap;

  retryNow() {
    if (this.stopped) return;
    if (this.state === "offline") return void this.start();
    if (["ended", "missing", "self", "joining"].includes(this.state)) return;
    this.attempt = 0;
    void this.connect();
  }

  /** Receive one file (it joins the queue). */
  want(i: number) {
    const f = this.files.get(i);
    if (!f || (f.state !== "offered" && f.state !== "failed")) return;
    this.reset(f);
    f.state = "queued";
    this.changed();
    this.pump();
  }

  wantAll() {
    for (const f of this.files.values()) if (f.state === "offered") f.state = "queued";
    this.changed();
    this.pump();
  }

  setVerify(on: boolean) {
    this.verify = on;
    try { localStorage.setItem(VERIFY_KEY, on ? "on" : "off"); } catch { /* this visit only */ }
    this.changed();
  }

  /** Hands a received file to the browser's download. */
  save(i: number) {
    const f = this.files.get(i);
    if (!f?.url) return;
    const a = document.createElement("a");
    a.href = f.url;
    a.download = f.info.name;
    a.rel = "noopener";
    document.body.appendChild(a);
    a.click();
    a.remove();
    f.saved = true;
    this.changed();
  }

  // Connection (the same steps as Monitor's parent device) -------------------------------------

  private onDrop(doc: any) {
    if (!doc || this.stopped) return;
    if (doc.open === false) return this.ended();
    if (doc.session !== this.session || this.state === "noanswer") {
      this.session = String(doc.session || "");
      this.attempt = 0;
      this.failures = 0;
      void this.connect();
    }
  }

  private async onLink(doc: any) {
    const pc = this.pc;
    if (!doc || !pc || this.answered || doc.answerId !== this.offerId || pc.signalingState !== "have-local-offer") return;
    this.answered = true;
    this.clearTimers();
    try {
      await pc.setRemoteDescription(JSON.parse(String(doc.answer)));
    } catch {
      if (this.pc === pc) this.lost(false);
      return;
    }
    // The sender answered; if the devices still cannot reach each other, ICE fails or stalls.
    this.later(() => { if (this.pc === pc && pc.connectionState !== "connected") this.lost(true); }, 20000);
  }

  private async connect() {
    const config = await iceConfig();
    if (this.stopped || ["ended", "missing", "self"].includes(this.state)) return;
    this.clearTimers();
    this.closePc();
    if (this.state !== "noanswer" && this.state !== "blocked") this.setState(this.attempt ? "reconnecting" : "connecting");
    const pc = new RTCPeerConnection(config);
    this.pc = pc;
    const dc = pc.createDataChannel("drop", { ordered: true });
    dc.binaryType = "arraybuffer";
    this.dc = dc;
    dc.onmessage = (m) => this.onMessage(dc, m.data);
    dc.onclose = () => { if (this.pc === pc) this.lost(false); };
    pc.onconnectionstatechange = () => {
      if (this.pc !== pc) return;
      const s = pc.connectionState;
      if (s === "connected") {
        this.attempt = 0;
        this.failures = 0;
        this.clearTimers();
        this.dropLink();
        setTimeout(() => void readRoute(pc).then((r) => { if (this.pc === pc) { this.route = r; this.changed(); } }).catch(() => null), 1000);
      } else if (s === "failed") {
        this.lost(true);
      } else if (s === "disconnected") {
        this.later(() => { if (this.pc === pc && pc.connectionState !== "connected") this.lost(false); }, 5000);
      }
    };
    try {
      await pc.setLocalDescription(await pc.createOffer());
      await gathered(pc);
      if (this.pc !== pc) return;
      this.offerId = randomId(12);
      this.answered = false;
      await set(`drops/${this.dropId}/links/${this.me}`, {
        offer: JSON.stringify(pc.localDescription), offerId: this.offerId, session: this.session || "", answer: "", answerId: "", at: serverTimestamp(),
      });
    } catch {
      if (this.pc === pc) this.lost(false);
      return;
    }
    this.later(() => { if (this.pc === pc && !this.answered) this.noAnswer(); }, 12000);
  }

  private lost(iceFailed: boolean) {
    if (this.stopped || ["ended", "missing", "self"].includes(this.state)) return;
    const wasAnswered = this.answered;
    this.clearTimers();
    this.closePc();
    this.dropLink();
    if (iceFailed && wasAnswered) this.failures++;
    this.attempt++;
    if (this.attempt > MAX_TRIES) return this.setState(this.failures >= 2 ? "blocked" : "noanswer");
    this.setState(this.failures >= 2 ? "blocked" : "reconnecting");
    this.later(() => void this.connect(), this.failures >= 2 ? 30000 : RETRY_MS[Math.min(this.attempt - 1, RETRY_MS.length - 1)]);
  }

  private noAnswer() {
    this.clearTimers();
    this.closePc();
    this.dropLink();
    this.attempt++;
    this.setState("noanswer");
    if (this.attempt <= MAX_TRIES) this.later(() => void this.connect(), NOANSWER_RETRY_MS[Math.min(this.attempt - 1, NOANSWER_RETRY_MS.length - 1)]);
  }

  private ended() {
    this.clearTimers();
    this.closePc();
    this.dropLink();
    for (const f of this.files.values()) {
      if (f.state === "receiving" || f.state === "queued" || (f.state === "checking" && !this.sums.has(f.info.i) && f.hasher)) {
        this.fail(f, "The sender stopped sharing before this file arrived.");
      }
    }
    this.setState("ended");
  }

  /**
   * The connection setup lists this device's network addresses, so it is kept only while a try to
   * connect is under way: deleted once connected, when a try fails, and when the share ends or this
   * screen closes. (A delete started as the page unloads is cancelled by the browser.)
   */
  private dropLink() {
    if (this.me) void set(`drops/${this.dropId}/links/${this.me}`, null).catch(() => null);
  }

  private closePc() {
    const pc = this.pc;
    this.pc = null;
    this.dc = null;
    this.answered = false;
    this.streaming = null;
    this.route = null;
    pc?.close();
  }

  // Files ------------------------------------------------------------------------------------

  private send(m: FromReceiver) {
    if (this.dc?.readyState === "open") this.dc.send(JSON.stringify(m));
  }

  private onMessage(dc: RTCDataChannel, raw: unknown) {
    if (dc !== this.dc) return;
    if (raw instanceof ArrayBuffer) return this.onBytes(raw);
    const m = parse<FromSender>(raw);
    if (!m) return;
    if (m.t === "files") {
      this.from = String(m.from || "");
      for (const info of m.files) {
        const f = this.files.get(info.i);
        if (f) f.info = info;
        else this.files.set(info.i, { info, state: "offered", received: 0, check: null, error: null, url: null, file: null, saved: false, sink: null, writes: Promise.resolve(), pending: [], pendingBytes: 0, hasher: null });
      }
      if (this.state !== "live") this.setState("live");
      // After a reconnect: ask again for checksums still owed.
      for (const f of this.files.values()) if (f.state === "checking" && f.hasher && !this.sums.has(f.info.i)) this.send({ t: "sum", i: f.info.i });
      this.changed();
      this.pump();
    } else if (m.t === "end") {
      const f = this.files.get(m.i);
      if (!f || this.streaming !== m.i) return;
      this.streaming = null;
      void this.finish(f);
      this.pump();
    } else if (m.t === "sum") {
      this.sums.set(m.i, m.hex);
      this.sumWaiters.get(m.i)?.(m.hex);
    } else if (m.t === "error") {
      const f = this.files.get(m.i);
      if (!f) return;
      if (this.streaming === m.i) this.streaming = null;
      this.fail(f, m.message);
      this.pump();
    } else if (m.t === "bye") {
      this.ended();
    }
  }

  /** Starts (or resumes) the next wanted file, one at a time. */
  private pump() {
    if (this.streaming !== null || this.dc?.readyState !== "open") return;
    const list = [...this.files.values()].sort((a, b) => a.info.i - b.info.i);
    const next = list.find((f) => f.state === "receiving") || list.find((f) => f.state === "queued");
    if (next) void this.request(next);
  }

  private async request(f: Incoming) {
    const dc = this.dc;
    const i = f.info.i;
    this.streaming = i;
    if (!f.sink) {
      f.sink = await openSink(f.info.size, String(i));
      f.received = 0;
      f.hasher = this.verify ? sha256.create() : null;
    }
    if (dc !== this.dc || this.streaming !== i) return;
    f.state = "receiving";
    this.send({ t: "get", i, from: f.received });
    if (f.hasher && !this.sums.has(i)) this.send({ t: "sum", i });
    this.changed();
  }

  private onBytes(buf: ArrayBuffer) {
    const f = this.streaming === null ? null : this.files.get(this.streaming);
    if (!f || f.state !== "receiving") return;
    if (f.received + buf.byteLength > f.info.size) {
      // Start a clean connection, so no leftover bytes of this file are taken for the next one.
      this.fail(f, "More data arrived than this file holds. Receive it again.");
      return this.lost(false);
    }
    f.received += buf.byteLength;
    this.bytesIn += buf.byteLength;
    f.hasher?.update(new Uint8Array(buf));
    f.pending.push(buf);
    f.pendingBytes += buf.byteLength;
    if (f.pendingBytes >= FLUSH) this.flush(f);
    this.dirty = true;
  }

  /** Writes what is held in memory to storage, then tells the sender how far this device got. */
  private flush(f: Incoming) {
    const sink = f.sink;
    if (!f.pendingBytes || !sink) return;
    const part = new Blob(f.pending);
    const at = f.received;
    f.pending = [];
    f.pendingBytes = 0;
    f.writes = f.writes.then(() => sink.write(part)).then(
      () => { if (this.streaming === f.info.i) this.send({ t: "ack", i: f.info.i, at }); },
      () => {
        if (f.sink !== sink || f.state === "failed") return;
        const streaming = this.streaming === f.info.i;
        this.fail(f, "This device ran out of space for the file. Free up some space and receive it again.");
        // The sender is still sending this file: start a clean connection rather than mix its bytes into the next one.
        if (streaming) this.lost(false);
      },
    );
  }

  private async finish(f: Incoming) {
    if (f.received !== f.info.size) return this.fail(f, "Part of this file did not arrive. Receive it again.");
    f.state = "checking";
    this.changed();
    this.flush(f);
    const sink = f.sink!;
    let blob: Blob;
    try {
      await f.writes;
      if (f.sink !== sink || f.state !== "checking") return;
      blob = await sink.finish();
    } catch {
      return this.fail(f, "This device could not keep the file. Free up some space and receive it again.");
    }
    if (f.hasher) {
      const mine = bytesToHex(f.hasher.digest());
      const theirs = this.sums.get(f.info.i) ?? await new Promise<string>((resolve) => this.sumWaiters.set(f.info.i, resolve));
      this.sumWaiters.delete(f.info.i);
      if (f.state !== "checking") return;
      if (mine !== theirs) {
        await sink.discard();
        return this.fail(f, "This file did not arrive exactly as sent. Receive it again.");
      }
      f.check = "ok";
    }
    f.file = new File([blob], f.info.name, { type: safeType(f.info.type) });
    f.url = URL.createObjectURL(f.file);
    f.state = "done";
    this.changed();
    // iPhone and iPad ask before every download, so there each file waits for a tap on Save.
    if (!isIOS) this.save(f.info.i);
  }

  private fail(f: Incoming, message: string) {
    if (f.state === "failed") return;
    void f.sink?.discard();
    this.reset(f);
    f.state = "failed";
    f.error = message;
    this.changed();
  }

  private reset(f: Incoming) {
    if (f.url) URL.revokeObjectURL(f.url);
    Object.assign(f, { sink: null, writes: Promise.resolve(), pending: [], pendingBytes: 0, hasher: null, received: 0, check: null, error: null, url: null, file: null, saved: false });
    this.sums.delete(f.info.i);
  }

  // Snapshot -----------------------------------------------------------------------------------

  private tick() {
    const busy = [...this.files.values()].some((f) => f.state === "receiving" || f.state === "queued");
    this.speed = busy && this.state === "live" ? this.rate.push(this.bytesIn) : (this.rate.reset(), 0);
    if (this.dirty || busy) this.changed();
    const s = this.snap;
    if (this.state === "live") this.send({ t: "progress", got: s.got, total: s.total, done: s.done, count: s.count });
  }

  private later(fn: () => void, ms: number) {
    this.timers.push(window.setTimeout(fn, ms));
  }

  private clearTimers() {
    for (const t of this.timers) clearTimeout(t);
    this.timers = [];
  }

  private setState(s: LinkState) {
    this.state = s;
    this.changed();
  }

  private changed() {
    this.dirty = false;
    this.snap = this.build();
    for (const fn of this.listeners) fn();
  }

  private build(): ReceiveSnapshot {
    const files = [...this.files.values()].sort((a, b) => a.info.i - b.info.i);
    const wanted = files.filter((f) => f.state !== "offered" && f.state !== "failed");
    return {
      state: this.state,
      from: this.from,
      files: files.map(({ info, state, received, check, error, url, file, saved }) => ({ info, state, received: state === "done" ? info.size : received, check, error, url, file, saved })),
      got: wanted.reduce((n, f) => n + (f.state === "done" ? f.info.size : Math.min(f.received, f.info.size)), 0),
      total: wanted.reduce((n, f) => n + f.info.size, 0),
      done: wanted.filter((f) => f.state === "done").length,
      count: wanted.length,
      rate: this.speed,
      route: this.route,
      verify: this.verify,
      inMemory: files.some((f) => f.sink?.kind === "memory" && (f.state === "receiving" || f.state === "checking")),
    };
  }
}