// Peer delivery in a page (extensions/peer-delivery.md): the links, and the client over them.
//
// Everything that decides is in the `peer` bundle (src/peer.rs), the Rust a native sink runs:
// which holders to ask, whether a reply is the object its receipt names, what to set aside and
// what to sign. This file owns what only the page can do: it carries frames over a data channel
// (§7) or makes the HTTP requests of §4.5, and it never uses a byte a peer sent until the
// bundle has said `taken`.
//
// `w` below is the bundle's exports.

const LINK_FAILED = "peer_link_failed";
const UNEXPECTED = "peer_frame_unexpected";
const MALFORMED = "peer_frame_malformed";
const TOO_LARGE = "peer_object_too_large";
const RECORD_MAX = 1 << 20;

/** A refusal of §4.4 or §4.5; `reason` is its code. */
export class PeerError extends Error {
  constructor(reason) {
    super(reason);
    this.reason = reason;
  }
}

export const hex = (b) => Array.from(b, (x) => x.toString(16).padStart(2, "0")).join("");
export const unhex = (s) => Uint8Array.from(s.match(/../g) || [], (x) => parseInt(x, 16));

// ───────────────────────────── links ─────────────────────────────
//
// A link carries messages whole and in order: { send(bytes), recv(ms), close() }. `recv`
// rejects with `peer_link_failed` when no message comes within `ms` or the link has closed.

function inbox() {
  const waiting = [];
  const ready = [];
  let ended = false;
  return {
    push(m) {
      if (ended) return;
      const w = waiting.shift();
      if (w) w.resolve(m);
      else ready.push(m);
    },
    end() {
      ended = true;
      for (const w of waiting.splice(0)) w.reject(new PeerError(LINK_FAILED));
    },
    shift(ms) {
      if (ready.length) return Promise.resolve(ready.shift());
      if (ended) return Promise.reject(new PeerError(LINK_FAILED));
      return new Promise((resolve, reject) => {
        const entry = {
          resolve: (m) => {
            clearTimeout(timer);
            resolve(m);
          },
          reject: (e) => {
            clearTimeout(timer);
            reject(e);
          },
        };
        const timer = setTimeout(() => {
          const k = waiting.indexOf(entry);
          if (k >= 0) waiting.splice(k, 1);
          reject(new PeerError(LINK_FAILED));
        }, ms);
        waiting.push(entry);
      });
    },
  };
}

/** A link over an open, reliable and ordered data channel (§7). */
export function channelLink(channel) {
  channel.binaryType = "arraybuffer";
  const box = inbox();
  // A text message is no frame: it reads as an empty one, which is malformed.
  channel.addEventListener("message", (e) => box.push(typeof e.data === "string" ? new Uint8Array(0) : new Uint8Array(e.data)));
  channel.addEventListener("close", () => box.end());
  channel.addEventListener("error", () => box.end());
  const HIGH = 1 << 20;
  return {
    async send(bytes) {
      if (channel.readyState !== "open") throw new PeerError(LINK_FAILED);
      if (channel.bufferedAmount > HIGH) {
        channel.bufferedAmountLowThreshold = HIGH >> 1;
        await new Promise((ok) => channel.addEventListener("bufferedamountlow", ok, { once: true }));
      }
      try {
        channel.send(bytes);
      } catch {
        throw new PeerError(LINK_FAILED);
      }
    },
    recv: (ms) => box.shift(ms),
    close() {
      try {
        channel.close();
      } catch {
        // already closed
      }
      box.end();
    },
  };
}

/** Two ends of a link in one process, for a run with no browser. */
export function memoryLinks() {
  const [a, b] = [inbox(), inbox()];
  let open = true;
  const end = (mine, theirs) => ({
    async send(bytes) {
      if (!open) throw new PeerError(LINK_FAILED);
      theirs.push(Uint8Array.from(bytes));
    },
    recv: (ms) => mine.shift(ms),
    close() {
      open = false;
      a.end();
      b.end();
    },
  });
  return [end(a, b), end(b, a)];
}

// ───────────────────────────── the asker's side of one holder ─────────────────────────────
//
// A connection to one holder: { record(), want(group, index, max), ack(body), close() }.
// `want` resolves to { bytes } or { miss: code } and rejects with a PeerError for a refusal of
// §4.4; after one the connection is closed.

/** The exchange of §4 over a link. */
export function overLink(w, link, { timeout = 10000 } = {}) {
  const read = (m) => {
    try {
      return JSON.parse(w.peerFrame(m));
    } catch {
      throw new PeerError(MALFORMED);
    }
  };
  // The bytes a head announced, from the `data` frames that follow it (§4.4).
  const reply = async (len) => {
    const t = new w.PeerTransfer(len);
    try {
      for (;;) {
        const m = await link.recv(timeout);
        if (read(m).frame !== "data") throw new PeerError(UNEXPECTED);
        let whole;
        try {
          whole = t.push(m.subarray(1));
        } catch {
          throw new PeerError(UNEXPECTED);
        }
        if (whole) return t.bytes();
      }
    } finally {
      t.free();
    }
  };
  const guarded = (f) => async (...a) => {
    try {
      return await f(...a);
    } catch (e) {
      link.close();
      throw e instanceof PeerError ? e : new PeerError(LINK_FAILED);
    }
  };
  return {
    kind: "link",
    record: guarded(async () => {
      await link.send(w.peerRecordWant());
      const f = read(await link.recv(timeout));
      if (f.frame !== "record-head") throw new PeerError(UNEXPECTED);
      return reply(f.len);
    }),
    want: guarded(async (group, index, max) => {
      await link.send(w.peerWant(group, index));
      const f = read(await link.recv(timeout));
      if (f.frame === "miss") {
        if (f.group !== hex(group) || f.index !== index) throw new PeerError(UNEXPECTED);
        return { miss: f.code };
      }
      if (f.frame !== "head") throw new PeerError(UNEXPECTED);
      const why = w.peerCheckHead(group, index, unhex(f.group), f.index, f.len, max);
      if (why) throw new PeerError(why);
      return { bytes: await reply(f.len) };
    }),
    ack: guarded(async (body) => {
      const frame = new Uint8Array(body.length + 1);
      frame[0] = 7;
      frame.set(body, 1);
      await link.send(frame);
      const f = read(await link.recv(timeout));
      if (f.frame !== "ack-result") throw new PeerError(UNEXPECTED);
      return { taken: f.taken, digest: f.digest };
    }),
    close: () => link.close(),
  };
}

/** The requests of §4.5 to a holder at `base`, a URL ending in `/`. */
export function overHttp(w, base, { timeout = 10000, fetch: fetchIt = globalThis.fetch } = {}) {
  // No credentials, no cookie, no referrer, and the browser's HTTP cache bypassed (§7).
  const request = (path, extra = {}) =>
    fetchIt(new URL(path, base), { credentials: "omit", referrerPolicy: "no-referrer", cache: "no-store", mode: "cors", redirect: "error", signal: AbortSignal.timeout(timeout), ...extra }).catch(() => {
      throw new PeerError(LINK_FAILED);
    });
  // A response's body, read to at most `max` bytes and held to its stated length.
  const body = async (res, max) => {
    const stated = res.headers.get("content-length");
    if (stated !== null && Number(stated) > max) {
      await res.body?.cancel().catch(() => {});
      throw new PeerError(TOO_LARGE);
    }
    const reader = res.body.getReader();
    const parts = [];
    let n = 0;
    for (;;) {
      const { done, value } = await reader.read().catch(() => {
        throw new PeerError(LINK_FAILED);
      });
      if (done) break;
      n += value.length;
      if (n > max || (stated !== null && n > Number(stated))) {
        await reader.cancel().catch(() => {});
        throw new PeerError(stated === null ? TOO_LARGE : LINK_FAILED);
      }
      parts.push(value);
    }
    if (stated !== null && n !== Number(stated)) throw new PeerError(LINK_FAILED);
    const out = new Uint8Array(n);
    let at = 0;
    for (const p of parts) {
      out.set(p, at);
      at += p.length;
    }
    return out;
  };
  return {
    kind: "http",
    async record() {
      const res = await request("holdings");
      if (res.status !== 200) throw new PeerError(LINK_FAILED);
      return body(res, RECORD_MAX);
    },
    async want(group, index, max) {
      const res = await request(`object/${hex(group)}/${index}`);
      if (res.status === 404) return { miss: 1 };
      if (res.status === 503) return { miss: 3 };
      if (res.status !== 200) throw new PeerError(LINK_FAILED);
      return { bytes: await body(res, max) };
    },
    async ack(ackBody) {
      const res = await request("ack", { method: "POST", headers: { "content-type": "application/octet-stream" }, body: ackBody });
      if (res.status === 422) return { taken: false, digest: null };
      if (res.status !== 200) throw new PeerError(LINK_FAILED);
      return { taken: true, digest: hex(await body(res, 32)) };
    },
    close() {},
  };
}

// ───────────────────────────── the asker ─────────────────────────────

/**
 * An asker's session (§5, §6). `origin(group, index)` fetches an object's bytes from the
 * service the sink takes delivery from; `policy` is the operator's (§3.1) and `setting` the
 * viewer's (§3.2).
 */
export class PeerClient {
  constructor(w, { seed, policy, setting, origin, now = () => Date.now() }) {
    this.w = w;
    this.asker = new w.PeerAsker(seed, JSON.stringify(policy || {}), JSON.stringify(setting || {}));
    this.origin = origin;
    this.now = now;
    this.conns = new Map();
    this.groups = new Map();
    this.acks = [];
  }

  /** §3.3: whether this sink may ask peers. While it may not, no connection is opened. */
  get mayAsk() {
    return this.asker.permitted();
  }

  get key() {
    return hex(this.asker.key());
  }

  /** Hold a group receipt and its objects' receipts, as JSON values. Returns the link hash. */
  holdGroup(receipt, objectReceipts) {
    const text = JSON.stringify(receipt);
    const link = this.asker.holdGroup(text);
    const objects = new Map();
    for (const o of objectReceipts) {
      const t = JSON.stringify(o);
      objects.set(this.asker.holdObject(link, t), t);
    }
    this.groups.set(hex(link), { receipt: text, objects });
    return hex(link);
  }

  /**
   * A neighbour, after those already added in the asker's order. `open` makes the connection
   * and is not called when the sink may not ask (§3.3). Returns the holder's key, or null.
   */
  async addNeighbour(open) {
    if (!this.mayAsk) return null;
    let conn;
    try {
      conn = await open();
      const record = await conn.record();
      // holdings §4.1: the holder's key follows the magic and the version.
      const holder = hex(record.subarray(5, 37));
      if (record.length < 37 || this.conns.has(holder)) {
        conn.close();
        return null;
      }
      this.asker.offerRecord(record);
      this.conns.set(holder, conn);
      return holder;
    } catch {
      conn?.close();
      return null;
    }
  }

  /** §5.1: the records' labels now. */
  records() {
    return JSON.parse(this.asker.records(this.now()));
  }

  /**
   * §5.2, §5.3: object `index` of the group with link hash `group` (hexadecimal): from the
   * holders of its route in order, then the origin. Resolves to { bytes, from, route,
   * attempts, verified }: `bytes` is null unless they verified under the object's receipt.
   */
  async get(group, index) {
    const g = unhex(group);
    const route = JSON.parse(this.asker.route(this.now(), g, index));
    const attempts = [];
    for (const holder of route.holders) {
      const conn = this.conns.get(holder);
      if (!conn) continue;
      const key = unhex(holder);
      this.asker.asked(key);
      let outcome;
      try {
        const r = await conn.want(g, index, this.asker.maxObjectBytes());
        if (r.miss !== undefined) {
          outcome = `miss reason=${this.asker.missed(key, g, index, r.miss)}`;
        } else {
          outcome = this.asker.take(key, g, index, r.bytes);
          if (outcome === "taken") {
            attempts.push({ holder, outcome });
            return { bytes: r.bytes, from: holder, route: route.label, attempts, verified: true };
          }
        }
      } catch (e) {
        const reason = e instanceof PeerError ? e.reason : LINK_FAILED;
        outcome = `${reason === TOO_LARGE ? "refused" : "failed"} reason=${reason}`;
        // §4.4: the asker asks that holder nothing more over a link that failed it.
        conn.close();
        this.conns.delete(holder);
      }
      attempts.push({ holder, outcome });
    }
    const bytes = await this.origin(group, index);
    const held = this.groups.get(group);
    const receipt = held?.objects.get(index);
    const refusal = receipt ? this.w.peerVerifyObject(held.receipt, receipt, bytes) : "refused reason=peer_receipt_not_held";
    return { bytes: refusal === "" ? bytes : null, from: "origin", route: route.label, attempts, verified: refusal === "", refusal };
  }

  /**
   * §6.1: sign and send the acknowledgement of what each holder served and this asker verified
   * in the window from `start` to `end` (microseconds). Resolves to one entry a holder:
   * { upstream, record, items, sent, taken }.
   */
  async acknowledge(start, end) {
    const out = [];
    for (const holder of JSON.parse(this.asker.owed())) {
      const body = this.asker.acknowledge(unhex(holder), start, end);
      if (!body) continue;
      const entry = { upstream: holder, ...JSON.parse(this.w.peerAckParts(body)), sent: false, taken: false };
      const conn = this.conns.get(holder);
      if (conn) {
        try {
          const r = await conn.ack(body);
          entry.sent = true;
          entry.taken = r.taken;
        } catch {
          // The record is kept for the party this asker settles with (§6.1).
        }
      }
      out.push(entry);
      this.acks.push(entry);
    }
    return out;
  }

  asks() {
    return JSON.parse(this.asker.asks());
  }

  close() {
    for (const c of this.conns.values()) c.close();
    this.conns.clear();
    this.asker.free();
  }
}

// ───────────────────────────── the holder ─────────────────────────────

/** A store of objects by group and position: in memory. A page uses `cacheStore`. */
export function memoryStore() {
  const m = new Map();
  return {
    get: async (group, index) => m.get(`${group}/${index}`) || null,
    put: async (group, index, bytes) => void m.set(`${group}/${index}`, bytes),
    delete: async (group, index) => m.delete(`${group}/${index}`),
  };
}

/** A store in a cache of the Cache API under the page's own origin (§7). */
export async function cacheStore(name = "wai-peer-1") {
  const cache = await caches.open(name);
  const url = (group, index) => `/wai-peer/${group}/${index}`;
  return {
    async get(group, index) {
      const res = await cache.match(url(group, index));
      return res ? new Uint8Array(await res.arrayBuffer()) : null;
    },
    put: (group, index, bytes) => cache.put(url(group, index), new Response(bytes, { headers: { "content-type": "application/octet-stream" } })),
    delete: (group, index) => cache.delete(url(group, index)),
  };
}

/**
 * A holder (§4.3, §6.3): it signs a record of what its store holds, serves it over links, and
 * checks the acknowledgements it is sent. A sink that may not serve signs no record and serves
 * no link.
 */
export class PeerServer {
  constructor(w, { seed, policy, setting, store }) {
    this.w = w;
    this.holder = new w.PeerHolder(seed, JSON.stringify(policy || {}), JSON.stringify(setting || {}));
    this.store = store;
    this.links = 0;
    this.stats = { wants: 0, served: 0, misses: 0, acks: [] };
  }

  get mayServe() {
    return this.holder.permitted();
  }

  get key() {
    return hex(this.holder.key());
  }

  /** Sign the record: `entries` is [{ group, leaf_count, held: [position, …] }, …]. */
  publish(asOfMs, expiresMs, entries) {
    return this.holder.seal(asOfMs, expiresMs, JSON.stringify(entries));
  }

  record() {
    return this.holder.record() || null;
  }

  /** §4.3: the answer to a `want` on link `id`: { bytes } or { miss: code }. */
  async object(id, group, index) {
    this.stats.wants++;
    const bytes = await this.store.get(hex(group), index);
    const code = this.holder.answer(group, index, !!bytes);
    if (code) {
      this.stats.misses++;
      return { miss: code };
    }
    this.holder.served(id, bytes);
    this.stats.served++;
    return { bytes };
  }

  /** §6.3: an `ack` frame's body received on link `id`. */
  takeAck(id, body) {
    const r = JSON.parse(this.holder.takeAck(id, body));
    this.stats.acks.push(r);
    return r;
  }

  /** Serve one link until it closes or sends what §4.3 does not admit. */
  async serve(link, { idle = 120000 } = {}) {
    if (!this.mayServe) {
      link.close();
      return;
    }
    const w = this.w;
    const id = `link-${++this.links}`;
    const sendAll = async (bytes) => {
      const max = w.peerDataMax();
      for (let at = 0; at < bytes.length; at += max) await link.send(w.peerData(bytes.subarray(at, at + max)));
    };
    for (;;) {
      let m;
      let f;
      try {
        m = await link.recv(idle);
        f = JSON.parse(w.peerFrame(m));
      } catch {
        break;
      }
      try {
        if (f.frame === "record-want") {
          const r = this.record();
          if (!r) break;
          await link.send(w.peerRecordHead(r.length));
          await sendAll(r);
        } else if (f.frame === "want") {
          const group = unhex(f.group);
          const a = await this.object(id, group, f.index);
          if (a.miss) {
            await link.send(w.peerMiss(group, f.index, a.miss));
          } else {
            await link.send(w.peerHead(group, f.index, a.bytes.length));
            await sendAll(a.bytes);
          }
        } else if (f.frame === "ack") {
          const r = this.takeAck(id, m.subarray(1));
          await link.send(w.peerAckResult(unhex(r.digest), r.outcome === "taken"));
        } else {
          // §4.3: any other frame is peer_frame_unexpected.
          break;
        }
      } catch {
        break;
      }
    }
    link.close();
  }
}