Source

// The sending device: holds the picked files, hands out a link, a QR code and a 6-digit code,
// answers each receiving device's connection request, and streams the files to it over WebRTC.
// Bounded only carries the connection setup and the code; file names and bytes go device to device.
import { get, set, setMany, subscribe, serverTimestamp } from "@bounded-sh/client";
import { me } from "./lib/bounded";
import { gathered, iceConfig, randomCode, randomId, readRoute, type Route } from "./lib/rtc";
import { WINDOW, parse, type FileInfo, type FromReceiver, type FromSender } from "./lib/protocol";
import { read, sumOf, toOutgoing, type Outgoing, type Pick } from "./lib/items";
import { Rate } from "./lib/format";
import { deviceName } from "./lib/device";

const CODE_MS = 10 * 60 * 1000;
const HIGH_WATER = 4 * 1024 * 1024;
const LEFTOVER = "drop:sending";

interface Link { id: string; offer: string; offerId: string; session: string; answerId: string }

interface Peer {
  id: string;
  pc: RTCPeerConnection;
  offerId: string;
  dc: RTCDataChannel | null;
  connected: boolean;
  /** Bumped for every file request, so an older stream stops. */
  serving: number;
  servingIndex: number;
  sent: number;
  acked: number;
  wake: (() => void) | null;
}

export interface ReceiverView {
  id: string;
  name: string;
  online: boolean;
  got: number;
  total: number;
  done: number;
  count: number;
  rate: number;
  route: Route | null;
}

export interface SenderSnapshot {
  state: "starting" | "sharing" | "stopped" | "failed";
  error: string | null;
  link: string | null;
  code: string | null;
  files: FileInfo[];
  receivers: ReceiverView[];
}

const rowId = (r: any): string => String(r.id ?? String(r._id ?? "").split("/").pop());

/** Deletes the connection setup (each receiver's network addresses) left in a share this device owns. Run once it is closed, so no new one appears. */
async function clearLinks(id: string) {
  const rows: any[] = (await get(`drops/${id}/links`, { limit: 1000 }).catch(() => null))?.data ?? [];
  await Promise.all(rows.map((r) => set(`drops/${id}/links/${rowId(r)}`, null).catch(() => null)));
}

export class Sender {
  readonly session = randomId(12);
  readonly id = randomId(20);
  private me = "";
  private items: Outgoing[];
  private peers = new Map<string, Peer>();
  private views = new Map<string, ReceiverView>();
  private rates = new Map<string, Rate>();
  private stops: (() => Promise<void> | void)[] = [];
  private listeners = new Set<() => void>();
  private state: SenderSnapshot["state"] = "starting";
  private error: string | null = null;
  private code: string | null = null;
  private codeTimer: number | undefined;
  private snap: SenderSnapshot;

  constructor(readonly picks: Pick[]) {
    this.items = toOutgoing(picks, 0);
    this.snap = this.build();
  }

  get link() { return `${location.origin}/r#${this.id}`; }

  async start() {
    try {
      this.me = await me();
      await setMany([
        { path: `drops/${this.id}`, document: { owner: this.me, session: this.session, open: true, createdAt: serverTimestamp() } },
        { path: `drops/${this.id}/joins/${this.me}`, document: { role: "sender", name: deviceName(), at: serverTimestamp() } },
      ]);
      this.remember();
      this.stops.push(await subscribe(`drops/${this.id}/links`, { limit: 200, onData: (rows: any[]) => this.onLinks(rows || []) }));
      this.stops.push(await subscribe(`drops/${this.id}/joins`, { limit: 200, onData: (rows: any[]) => this.onJoins(rows || []) }));
      this.state = "sharing";
      this.changed();
      await this.renewCode();
    } catch (err) {
      if (this.state === "stopped") return;
      this.state = "failed";
      this.error = /offline|network|fetch/i.test(String((err as Error)?.message)) ? "You seem to be offline. Connect to the internet and try again." : "Could not start sharing. Check the connection and try again.";
      this.changed();
    }
  }

  /** Ends the share: the link and code stop working and every receiver is told. */
  async stop() {
    if (this.state === "stopped") return;
    this.state = "stopped";
    clearTimeout(this.codeTimer);
    for (const p of this.peers.values()) this.send(p, { t: "bye" });
    for (const s of this.stops) await s();
    for (const id of [...this.peers.keys()]) this.closePeer(id);
    for (const v of this.views.values()) v.online = false;
    this.changed();
    const code = this.code;
    this.code = null;
    this.forget();
    await set(`drops/${this.id}`, { open: false }).catch(() => null);
    if (code) await set(`codes/${code}`, null).catch(() => null);
    await clearLinks(this.id);
  }

  /** Adds more files; devices already connected see them at once. */
  add(picks: Pick[]) {
    this.picks.push(...picks);
    this.items.push(...toOutgoing(picks, this.items.length));
    for (const p of this.peers.values()) this.sendFiles(p);
    this.changed();
  }

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

  // The 6-digit code ---------------------------------------------------------------------------

  /** Writes a fresh code that points to this share for 10 minutes, and replaces it before it runs out. */
  private async renewCode() {
    clearTimeout(this.codeTimer);
    if (this.state !== "sharing") return;
    const old = this.code;
    let fresh: string | null = null;
    for (let i = 0; i < 6 && !fresh; i++) {
      const code = randomCode();
      try {
        await set(`codes/${code}`, { drop: this.id, by: this.me, at: serverTimestamp() });
        fresh = code;
      } catch {
        // Another sender holds this code right now; pick another.
      }
    }
    if (this.state !== "sharing") {
      if (fresh) await set(`codes/${fresh}`, null).catch(() => null);
      return;
    }
    if (fresh) {
      this.code = fresh;
      this.remember();
      this.changed();
      if (old) await set(`codes/${old}`, null).catch(() => null);
    }
    this.codeTimer = window.setTimeout(() => void this.renewCode(), fresh ? CODE_MS - 20_000 : 30_000);
  }

  private remember() {
    try { sessionStorage.setItem(LEFTOVER, JSON.stringify({ id: this.id, code: this.code })); } catch { /* best effort */ }
  }

  private forget() {
    try { sessionStorage.removeItem(LEFTOVER); } catch { /* best effort */ }
  }

  /** A share left open by a reload of this tab is closed: its files are gone with the page. Returns whether there was one. */
  static async closeLeftover(): Promise<boolean> {
    let left: { id?: string; code?: string } | null = null;
    try { left = JSON.parse(sessionStorage.getItem(LEFTOVER) || "null"); sessionStorage.removeItem(LEFTOVER); } catch { /* none */ }
    if (!left?.id) return false;
    await me().catch(() => null);
    await set(`drops/${left.id}`, { open: false }).catch(() => null);
    if (left.code) await set(`codes/${left.code}`, null).catch(() => null);
    await clearLinks(left.id);
    return true;
  }

  // Signaling ----------------------------------------------------------------------------------

  private onJoins(rows: any[]) {
    for (const r of rows) {
      const id = rowId(r);
      if (r.role !== "receiver") continue;
      const v = this.views.get(id);
      if (v) v.name = String(r.name || v.name);
      else this.views.set(id, { id, name: String(r.name || "A device"), online: this.peers.get(id)?.dc?.readyState === "open", got: 0, total: 0, done: 0, count: 0, rate: 0, route: null });
    }
    this.changed();
  }

  private onLinks(rows: any[]) {
    for (const r of rows) {
      const link: Link = { id: rowId(r), offer: String(r.offer || ""), offerId: String(r.offerId || ""), session: String(r.session || ""), answerId: String(r.answerId || "") };
      if (!link.offer || link.session !== this.session || link.answerId === link.offerId) continue;
      if (this.peers.get(link.id)?.offerId === link.offerId) continue;
      void this.answer(link);
    }
  }

  private async answer(link: Link) {
    const config = await iceConfig();
    if (this.state !== "sharing" || this.peers.get(link.id)?.offerId === link.offerId) return;
    this.closePeer(link.id);
    const pc = new RTCPeerConnection(config);
    const peer: Peer = { id: link.id, pc, offerId: link.offerId, dc: null, connected: false, serving: 0, servingIndex: -1, sent: 0, acked: 0, wake: null };
    this.peers.set(link.id, peer);
    const current = () => this.peers.get(link.id) === peer;

    pc.ondatachannel = (e) => {
      const dc = e.channel;
      dc.binaryType = "arraybuffer";
      dc.bufferedAmountLowThreshold = 1024 * 1024;
      peer.dc = dc;
      const opened = () => { this.sendFiles(peer); this.setOnline(peer.id, true); };
      dc.onopen = opened;
      dc.onmessage = (m) => this.onMessage(peer, m.data);
      dc.onbufferedamountlow = () => this.wake(peer);
      dc.onclose = () => { if (current()) this.closePeer(peer.id); };
      if (dc.readyState === "open") opened();
    };
    pc.onconnectionstatechange = () => {
      if (!current()) return;
      if (pc.connectionState === "connected") {
        peer.connected = true;
        setTimeout(() => void readRoute(pc).then((route) => {
          const v = this.views.get(peer.id);
          if (v && current()) { v.route = route; this.changed(); }
        }).catch(() => null), 1000);
      }
      if (pc.connectionState === "failed" || pc.connectionState === "closed") this.closePeer(peer.id);
    };

    try {
      await pc.setRemoteDescription(JSON.parse(link.offer));
      await pc.setLocalDescription(await pc.createAnswer());
      await gathered(pc);
      if (!current() || this.state !== "sharing") return;
      await set(`drops/${this.id}/links/${link.id}`, { answer: JSON.stringify(pc.localDescription), answerId: link.offerId });
    } catch {
      // The receiver sent a newer offer meanwhile (the answer write is refused), or the offer was
      // unusable. Either way the receiver retries with a fresh offer.
      if (current()) this.closePeer(link.id);
    }
    // A receiver that never connects is dropped, so a stale offer does not hold a connection open.
    // Its connection setup goes too, unless it has sent a newer one, in case its page closed before it could delete it.
    setTimeout(() => {
      if (peer.connected || (this.peers.get(link.id) ?? peer) !== peer) return;
      this.closePeer(peer.id);
      void set(`drops/${this.id}/links/${link.id}`, null).catch(() => null);
    }, 30000);
  }

  private closePeer(id: string) {
    const p = this.peers.get(id);
    if (!p) return;
    this.peers.delete(id);
    p.serving++;
    this.wake(p);
    p.dc?.close();
    p.pc.close();
    this.setOnline(id, false);
  }

  // Sending files ----------------------------------------------------------------------------

  private send(peer: Peer, m: FromSender) {
    if (peer.dc?.readyState === "open") peer.dc.send(JSON.stringify(m));
  }

  private sendFiles(peer: Peer) {
    this.send(peer, { t: "files", from: deviceName(), files: this.items.map((it) => it.info) });
  }

  private onMessage(peer: Peer, raw: unknown) {
    const m = parse<FromReceiver>(raw);
    // A file index from the other device must be a number: "__proto__" would reach Array.prototype.
    if (!m || (m.t !== "progress" && !Number.isInteger(m.i))) return;
    if (m.t === "get") {
      void this.serve(peer, m.i, Math.max(0, Math.floor(m.from) || 0));
    } else if (m.t === "ack") {
      if (m.i === peer.servingIndex && m.at > peer.acked) { peer.acked = m.at; this.wake(peer); }
    } else if (m.t === "sum") {
      const item = this.items[m.i];
      if (!item) return;
      sumOf(item).then(
        (hex) => this.send(peer, { t: "sum", i: m.i, hex }),
        () => this.send(peer, { t: "error", i: m.i, message: "This device could not read the file to check it." }),
      );
    } else if (m.t === "progress") {
      const v = this.views.get(peer.id);
      if (!v) return;
      Object.assign(v, { got: m.got, total: m.total, done: m.done, count: m.count });
      let rate = this.rates.get(peer.id);
      if (!rate) this.rates.set(peer.id, (rate = new Rate()));
      v.rate = rate.push(m.got);
      this.changed();
    }
  }

  /** Streams item i from byte `from`, staying at most WINDOW bytes ahead of what the receiver saved. */
  private async serve(peer: Peer, i: number, from: number) {
    const token = ++peer.serving;
    this.wake(peer);
    const item = this.items[i];
    const live = () => peer.serving === token && peer.dc?.readyState === "open";
    if (!item) return this.send(peer, { t: "error", i, message: "That file is no longer offered." });
    peer.servingIndex = i;
    peer.sent = from;
    peer.acked = from;
    try {
      for await (const chunk of read(item, from)) {
        while (live() && (peer.sent - peer.acked > WINDOW || peer.dc!.bufferedAmount > HIGH_WATER)) await this.pause(peer);
        if (!live()) return;
        peer.dc!.send(chunk);
        peer.sent += chunk.byteLength;
      }
      if (live()) this.send(peer, { t: "end", i });
    } catch {
      if (live()) this.send(peer, { t: "error", i, message: "This device could not read the file. It may have been moved or changed." });
    }
  }

  private pause(peer: Peer): Promise<void> {
    return new Promise((resolve) => {
      const timer = setTimeout(() => { peer.wake = null; resolve(); }, 1000);
      peer.wake = () => { clearTimeout(timer); peer.wake = null; resolve(); };
    });
  }

  private wake(peer: Peer) { peer.wake?.(); }

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

  private setOnline(id: string, online: boolean) {
    const v = this.views.get(id);
    if (v) {
      v.online = online;
      if (!online) { v.rate = 0; this.rates.get(id)?.reset(); }
    }
    this.changed();
  }

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

  private build(): SenderSnapshot {
    return {
      state: this.state,
      error: this.error,
      link: this.state === "sharing" ? this.link : null,
      code: this.state === "sharing" ? this.code : null,
      files: this.items.map((it) => it.info),
      receivers: [...this.views.values()].map((v) => ({ ...v })),
    };
  }
}