// The player integration of WAI's stream audit (wai/extensions/stream-audit.md): what a player
// signs for one playback session of a stream another service delivers over HTTP, and the bundle
// it sends to the owner's collector (§9).
//
//   AuditSession    one session: its key, its objects, its spans, its ads, its log, its record
//   acknowledge     settlement §3 acknowledgements over plain SHA-256 (§4, settlement §14)
//   ContentSpans    observations to the record's spans (§5.2)
//   AdSpans         observations to a claim's spans over an ad's media time (ad-presentation §10)
//   parsePlaylist   an HLS media playlist: its segments, program times and marked ads (§6.1)
//   attach          a media element's observations, every 200 ms and on its events
//
// Everything is the platform's: WebCrypto's SHA-256 and Ed25519, IntersectionObserver, the media
// element; BLAKE3 for jwp-receipts §2.2's tree root is ./blake3.js. No dependency.

import { blake3 } from "./blake3.js";

const enc = new TextEncoder();
const subtle = globalThis.crypto.subtle;

export const SESSION_MAGIC = enc.encode("WAS1");
export const SESSION_DOMAIN = enc.encode("wai:audit-session\x01");
export const CHANNEL_DOMAIN = enc.encode("wai:audit-channel\x01");
export const MARKER_DOMAIN = enc.encode("wai:audit-marker\x01");
export const ORIGIN_DOMAIN = enc.encode("wai:settle-origin\x01");
export const ACK_MAGIC = enc.encode("WDA1");
export const ACK_DOMAIN = enc.encode("wai:settle-ack\x01");
export const ITEMS_DOMAIN = enc.encode("wai:settle-items\x01");
export const LEAF_DOMAIN = enc.encode("moq-jwp:merkle-leaf\x01");
export const NODE_DOMAIN = enc.encode("moq-jwp:merkle-node\x01");
export const KEY_SESSION_DOMAIN = enc.encode("wai:ad-presentation-session\x01");
export const CLAIM_MAGIC = enc.encode("WPC1");
export const CLAIM_DOMAIN = enc.encode("wai:ad-presentation-claim\x01");
export const CREATIVE_DOMAIN = enc.encode("wai:ad-creative\x01");
export const LIVE = 1;
export const ON_DEMAND = 2;
export const ENDED = { ended: 1, user: 2, hidden: 3, idle: 4, error: 5 };
export const ROLE = { ORIGIN: 1, DELIVERY: 2, SINK_DECODE: 3, DISPLAY: 4 };
export const AUDIBLE = 1;
export const AD = 2;
export const RESUMED = 2;
export const GEOMETRY_PLAYER = 3;
export const NOT_EXPOSED = 0xffffffff;
export const NO_STARTUP = 0xffffffff;
export const DEFAULT_WINDOW_US = 10_000_000;

/** energy-measurement §2.7's class bytes: the unmetered marker, `ModelBased`, `Estimator`. */
export const CLASS = { UNMETERED: Uint8Array.of(0), MODEL_BASED: Uint8Array.of(2), ESTIMATOR: Uint8Array.of(3) };

/** An `OnChipCounter` class: its declared uncertainty and window, and its filtering (0 unknown, 1 off, 2 on). */
export function onChipCounter(relativePpm, windowUs, filtering) {
  const b = new Uint8Array(14);
  const dv = new DataView(b.buffer);
  b[0] = 4;
  dv.setUint32(1, relativePpm, false);
  dv.setBigUint64(5, BigInt(windowUs), false);
  b[13] = filtering;
  return b;
}

// ---------------------------------------------------------------- bytes

export const concat = (...parts) => {
  const out = new Uint8Array(parts.reduce((n, p) => n + p.length, 0));
  let at = 0;
  for (const p of parts) {
    out.set(p, at);
    at += p.length;
  }
  return out;
};
const u8 = (v) => Uint8Array.of(v & 0xff);
const u16 = (v) => Uint8Array.of(v & 0xff, (v >>> 8) & 0xff);
const u32 = (v) => {
  const b = new Uint8Array(4);
  new DataView(b.buffer).setUint32(0, v >>> 0, true);
  return b;
};
const u64 = (v) => {
  const b = new Uint8Array(8);
  new DataView(b.buffer).setBigUint64(0, BigInt(v), true);
  return b;
};
export const hex = (b) => Array.from(b, (x) => x.toString(16).padStart(2, "0")).join("");
export const unhex = (s) => Uint8Array.from(s.match(/../g) ?? [], (h) => parseInt(h, 16));
const cmp = (a, b) => {
  for (let i = 0; i < Math.min(a.length, b.length); i++) if (a[i] !== b[i]) return a[i] - b[i];
  return a.length - b.length;
};
const ZERO = new Uint8Array(32);

export async function sha256(...parts) {
  return new Uint8Array(await subtle.digest("SHA-256", concat(...parts)));
}

// ---------------------------------------------------------------- §3, settlement §14

/** The channel id of a name (§3.1). */
export async function channelId(name) {
  if (!/^[\x21-\x7e]{1,255}$/.test(name)) throw new RangeError("a channel name is 1 to 255 bytes of printable ASCII");
  return sha256(CHANNEL_DOMAIN, enc.encode(name));
}

/** The session of a session key in a channel (§3.2). */
export async function sessionOf(channel, key) {
  return (await sha256(KEY_SESSION_DOMAIN, channel, key)).subarray(0, 16);
}

/** The upstream an origin names (settlement §14.2). */
export async function originUpstream(origin) {
  return sha256(ORIGIN_DOMAIN, enc.encode(origin));
}

/** The marker digest of a marker's text (§6.1). */
export async function markerDigest(marker) {
  return sha256(MARKER_DOMAIN, enc.encode(marker));
}

/** jwp-receipts §2.2's tree root in its first form, over 32-byte content hashes. */
export function treeRoot(hashes) {
  if (hashes.length === 0) return ZERO;
  let level = hashes.map((h) => blake3(concat(LEAF_DOMAIN, h)));
  while (level.length > 1) {
    if (level.length % 2) level.push(level[level.length - 1]);
    const next = [];
    for (let i = 0; i < level.length; i += 2) next.push(blake3(concat(NODE_DOMAIN, level[i], level[i + 1])));
    level = next;
  }
  return level[0];
}

/** An item list (settlement §3.4): whole objects, `[{ hash, length }]`. */
export function itemList(items) {
  return concat(...items.flatMap((i) => [i.hash, u64(0), u64(i.length)]));
}

// ---------------------------------------------------------------- keys

/** A session key from 32 bytes of seed (the corpus's keys; a player makes a fresh one). */
export async function keyFromSeed(seed) {
  const pkcs8 = concat(Uint8Array.of(0x30, 0x2e, 0x02, 0x01, 0x00, 0x30, 0x05, 0x06, 0x03, 0x2b, 0x65, 0x70, 0x04, 0x22, 0x04, 0x20), seed);
  const privateKey = await subtle.importKey("pkcs8", pkcs8, { name: "Ed25519" }, true, ["sign"]);
  const jwk = await subtle.exportKey("jwk", privateKey);
  const b64 = jwk.x.replace(/-/g, "+").replace(/_/g, "/");
  const publicKey = Uint8Array.from(atob(b64 + "=".repeat((4 - (b64.length % 4)) % 4)), (c) => c.charCodeAt(0));
  return { privateKey, publicKey };
}

/** A fresh session key (§3.2): the platform's random generator, used for this session alone. */
export async function newKey() {
  const k = await subtle.generateKey({ name: "Ed25519" }, false, ["sign", "verify"]);
  return { privateKey: k.privateKey, publicKey: new Uint8Array(await subtle.exportKey("raw", k.publicKey)) };
}

async function sign(privateKey, domain, body) {
  return new Uint8Array(await subtle.sign({ name: "Ed25519" }, privateKey, concat(domain, body)));
}

// ---------------------------------------------------------------- §4 acknowledgements

/**
 * §4.1, §4.2: the acknowledgements of the objects received, `[{ at, origin, media, length, hash }]`
 * in the order received, in windows of `windowUs` from `start`, the last ending at `end`: one per
 * origin and window with an object, chained per origin, ordered by origin then window; a repeat
 * within a window counted and not listed. Gives `{ acks: [{ record, items, digest }], repeatObjects,
 * repeatBytes }`.
 */
export async function acknowledge(key, presentation, start, end, windowUs, received) {
  const mediaClass = presentation === LIVE ? 1 : 2;
  const windows = new Map(); // "origin\u0000window" -> { origin, k, classes: Map(class -> items) }
  const seen = new Set();
  let repeatObjects = 0;
  let repeatBytes = 0;
  for (const r of received) {
    const k = Math.floor((r.at - start) / windowUs);
    const id = `${r.origin}\u0000${k}\u0000${hex(r.hash)}`;
    if (seen.has(id)) {
      repeatObjects += 1;
      repeatBytes += r.length;
      continue;
    }
    seen.add(id);
    const w = `${r.origin}\u0000${k}`;
    if (!windows.has(w)) windows.set(w, { origin: r.origin, k, classes: new Map() });
    const cls = r.media ? mediaClass : 4;
    const c = windows.get(w).classes;
    if (!c.has(cls)) c.set(cls, []);
    c.get(cls).push({ hash: r.hash, length: r.length });
  }
  const order = [...windows.values()].sort((a, b) => (a.origin < b.origin ? -1 : a.origin > b.origin ? 1 : a.k - b.k));
  const prev = new Map();
  const acks = [];
  for (const w of order) {
    const ws = start + w.k * windowUs;
    const we = Math.min(ws + windowUs, end);
    const classes = [...w.classes.entries()].sort((a, b) => a[0] - b[0]);
    const lists = classes.map(([, items]) => itemList(items));
    const entries = await Promise.all(
      classes.map(async ([cls, items], i) =>
        concat(u8(cls), ZERO, ZERO, u32(items.length), u64(items.reduce((n, x) => n + x.length, 0)), treeRoot(items.map((x) => x.hash)), await sha256(ITEMS_DOMAIN, lists[i])),
      ),
    );
    const body = concat(ACK_MAGIC, u8(1), u8(1), u8(2), await originUpstream(w.origin), key.publicKey, u64(ws), u64(we), prev.get(w.origin) ?? ZERO, u16(entries.length), ...entries);
    const digest = await sha256(body);
    prev.set(w.origin, digest);
    acks.push({ record: concat(body, await sign(key.privateKey, ACK_DOMAIN, body)), items: lists, digest });
  }
  return { acks, repeatObjects, repeatBytes };
}

// ---------------------------------------------------------------- §5.2 spans

/**
 * The record's spans (§5.2): observations `{ at (µs), playing, visibleBp, audible, ad }`; a span
 * begins at each change of playback, of the visible fraction's step (`quantumBp`), of audibility
 * or of being within an ad; its fraction is the least observed over it.
 */
export class ContentSpans {
  constructor(start, { quantumBp = 100 } = {}) {
    this.start = start;
    this.quantum = quantumBp;
    this.done = [];
    this.cur = null;
  }

  ms(at) {
    return Math.max(0, Math.floor((at - this.start) / 1000));
  }

  close(at) {
    if (this.cur) {
      const end = this.ms(at);
      if (end > this.cur.startMs) {
        const flags = (this.cur.audible ? AUDIBLE : 0) | (this.cur.ad ? AD : 0);
        this.done.push({ startMs: this.cur.startMs, lengthMs: end - this.cur.startMs, visibleBp: this.cur.visibleBp, flags });
      }
    }
    this.cur = null;
  }

  observe({ at, playing, visibleBp, audible, ad = false }) {
    if (!playing) return this.close(at);
    const vis = Math.max(0, Math.min(10_000, Math.round(visibleBp)));
    const cls = Math.floor(vis / this.quantum);
    if (this.cur && (cls !== this.cur.cls || !!audible !== this.cur.audible || !!ad !== this.cur.ad)) this.close(at);
    if (this.cur) {
      this.cur.visibleBp = Math.min(this.cur.visibleBp, vis);
      return;
    }
    this.cur = { startMs: this.ms(at), visibleBp: vis, cls, audible: !!audible, ad: !!ad };
  }

  spans(at) {
    this.close(at);
    return this.done.slice();
  }
}

/**
 * A claim's spans over an ad's media time, by ad-presentation §10's rules: `{ positionMs, playing,
 * visibleBp, audible, seeked }`; media time already stated is never stated again, and a span after
 * a pause, a stall or a seek is a resumption.
 */
export class AdSpans {
  constructor({ quantumBp = 100 } = {}) {
    this.quantum = quantumBp;
    this.done = [];
    this.cur = null;
    this.maxEnd = 0;
    this.resumeNext = false;
  }

  close() {
    if (this.cur && this.cur.endMs > this.cur.startMs) {
      const flags = (this.cur.audible ? AUDIBLE : 0) | (this.cur.resumed ? RESUMED : 0);
      this.done.push({ startMs: this.cur.startMs, lengthMs: this.cur.endMs - this.cur.startMs, visibleBp: this.cur.visibleBp, flags });
      this.maxEnd = Math.max(this.maxEnd, this.cur.endMs);
    }
    this.cur = null;
  }

  observe({ positionMs, playing, visibleBp, audible, seeked = false }) {
    const pos = Math.max(0, Math.floor(positionMs));
    if (!playing || seeked) {
      this.close();
      if (this.done.length > 0) this.resumeNext = true;
      if (!playing) return;
    }
    const vis = Math.max(0, Math.min(10_000, Math.round(visibleBp)));
    const cls = Math.floor(vis / this.quantum);
    if (this.cur) {
      if (pos < this.cur.endMs) return;
      if (cls !== this.cur.cls || !!audible !== this.cur.audible) {
        this.cur.endMs = pos;
        this.close();
      } else {
        this.cur.endMs = pos;
        this.cur.visibleBp = Math.min(this.cur.visibleBp, vis);
        return;
      }
    }
    if (pos < this.maxEnd) return;
    const continues = !this.resumeNext && this.done.length > 0 && pos === this.maxEnd;
    const resumed = this.done.length > 0 && !continues;
    this.cur = { startMs: pos, endMs: pos, visibleBp: vis, cls, audible: !!audible, resumed };
    this.resumeNext = false;
  }

  spans() {
    this.close();
    return this.done.slice();
  }
}

// ---------------------------------------------------------------- §6 claims

/** The creative digest of `[{ hash, durationMs }]` (ad-presentation §4). */
export async function creativeDigest(objects) {
  return sha256(CREATIVE_DOMAIN, u32(objects.length), ...objects.flatMap((o) => [o.hash, u32(o.durationMs)]));
}

/** The content hashes of the objects the spans touch, in order (ad-presentation §9.2 step 6). */
export function touched(objects, spans) {
  const out = [];
  let s = 0;
  for (const o of objects) {
    const e = s + o.durationMs;
    if (spans.some((x) => x.startMs < e && x.startMs + x.lengthMs > s)) out.push(o.hash);
    s = e;
  }
  return out;
}

/** A presentation claim (ad-presentation §5) for an ad of the channel (§6.3), signed. */
export async function signClaim(key, c) {
  const spans = c.spans.flatMap((s) => [u32(s.startMs), u32(s.lengthMs), u16(s.visibleBp), u8(s.flags)]);
  const body = concat(
    CLAIM_MAGIC, u8(1), c.channel, await markerDigest(c.marker), c.session, u8(c.slot ?? 0), await creativeDigest(c.creative), key.publicKey,
    u8(c.geometry ?? GEOMETRY_PLAYER), u32(c.frames ?? NOT_EXPOSED), u16(c.objects.length), ...c.objects, u16(c.spans.length), ...spans, u16(0),
  );
  return concat(body, await sign(key.privateKey, CLAIM_DOMAIN, body));
}

// ---------------------------------------------------------------- §7 the log and CMCD

const OBJECT_NULLS = ["ot", "url", "content_hash", "capability", "rendition", "layer", "layers_total", "layer_cap", "counted", "transferred", "d_ms", "dl_ms", "ttfb_ms", "ttlb_ms", "rc", "path", "receipt"];

/** A delivery-telemetry observation (§2.1), every member in its order. */
export function observation(sid, sn, at, event, st, { contentHash = null, rc = null, present = null } = {}) {
  let object = null;
  if (event === "response") {
    object = Object.fromEntries(OBJECT_NULLS.map((k) => [k, null]));
    object.content_hash = contentHash;
    object.rc = rc;
  }
  return {
    v: 1, sid, sn, ts_ms: Math.floor(at / 1000), event, cid: null, st, sf: "h", object, present,
    switch: null, layer_change: null, capable: null, cohort: null, metrics: null, claim: null,
  };
}

/** delivery-telemetry §7.2's start time and stalls of a log. */
export function sessionMetrics(log) {
  let start = null;
  let present = null;
  let stall = null;
  let stalls = 0;
  let stallMs = 0;
  for (const o of log) {
    const t = o.ts_ms;
    if (o.event === "start" && start === null) start = t;
    else if (o.event === "present" && o.present.first) present = t;
    else if (o.event === "stall-start" && present !== null && stall === null) stall = t;
    else if (o.event === "stall-end") {
      if (stall !== null) {
        stalls += 1;
        stallMs += t - stall;
      }
      stall = null;
    } else if ((o.event === "end" || o.event === "abandon") && stall !== null) {
      stalls += 1;
      stallMs += t - stall;
      stall = null;
    }
  }
  const vst = present !== null && start !== null ? present - Math.min(start, present) : null;
  return { vst, stalls, stallMs };
}

const PREFIX = "science.transaction.wai-";
const sfString = (s) => `"${s.replace(/\\/g, "\\\\").replace(/"/g, '\\"')}"`;

/** The Event-mode CMCD report of an observation (delivery-telemetry §3, §4), or null. */
export function cmcdReport(o) {
  if (o.event === "present" && !o.present.first) return null;
  const m = new Map();
  m.set("v", "2");
  m.set("sid", sfString(o.sid));
  m.set("sn", String(o.sn));
  if (o.cid !== null) m.set("cid", sfString(o.cid));
  m.set("st", o.st);
  m.set("sf", o.sf);
  if (o.object) {
    if (o.object.content_hash !== null) m.set(PREFIX + "ch", sfString(o.object.content_hash));
    if (o.object.rc !== null) m.set("rc", String(o.object.rc));
  }
  m.set("ts", String(o.ts_ms));
  const [e, sta] = { response: ["rr", null], start: ["ps", "s"], present: ["ps", "p"], "stall-end": ["ps", "p"], "stall-start": ["ps", "r"], abandon: ["ps", "q"], end: ["ps", "e"] }[o.event];
  m.set("e", e);
  if (sta) m.set("sta", sta);
  if (o.present) {
    m.set(PREFIX + "u", sfString(String(o.present.unit)));
    m.set(PREFIX + "pl", sfString(String(o.present.layers)));
  }
  return [...m.keys()].sort().map((k) => `${k}=${m.get(k)}`).join(",");
}

/** The `application/cmcd` batch of observations (live-metering §6.3). */
export function cmcdBatch(log) {
  return log.map(cmcdReport).filter((x) => x !== null).join("\n");
}

// ---------------------------------------------------------------- §6.1 playlists

/** The attributes of a tag's attribute list (`KEY=value,KEY="quoted, value"`). */
export function attributes(list) {
  const out = {};
  const re = /([A-Z0-9-]+)=("[^"]*"|[^,]*)/g;
  let m;
  while ((m = re.exec(list))) out[m[1]] = m[2].startsWith('"') ? m[2].slice(1, -1) : m[2];
  return out;
}

/**
 * An HLS media playlist: its segments `{ uri, durationMs, programMs }` (absolute URIs; program
 * times from `EXT-X-PROGRAM-DATE-TIME`), whether it is on demand (`EXT-X-ENDLIST`), the key URIs
 * it names, and its marked ads `{ marker, startMs, durationMs }` (§6.1): `EXT-X-DATERANGE` tags
 * with `SCTE35-OUT` or `SCTE35-CMD`, or a `CLASS` in `adClasses`.
 */
export function parsePlaylist(text, base, { adClasses = [] } = {}) {
  const segments = [];
  const ads = [];
  const keys = [];
  let duration = null;
  let program = null;
  let map = null;
  let onDemand = false;
  for (const raw of text.split(/\r?\n/)) {
    const line = raw.trimEnd();
    if (line.startsWith("#EXTINF:")) duration = Math.max(1, Math.round(parseFloat(line.slice(8)) * 1000));
    else if (line.startsWith("#EXT-X-PROGRAM-DATE-TIME:")) program = Date.parse(line.slice(25));
    else if (line.startsWith("#EXT-X-ENDLIST")) onDemand = true;
    else if (line.startsWith("#EXT-X-MAP:")) map = new URL(attributes(line.slice(11)).URI, base).href;
    else if (line.startsWith("#EXT-X-KEY:") || line.startsWith("#EXT-X-SESSION-KEY:")) {
      const a = attributes(line.slice(line.indexOf(":") + 1));
      if (a.URI) keys.push(new URL(a.URI, base).href);
    } else if (line.startsWith("#EXT-X-DATERANGE:")) {
      const a = attributes(line.slice(17));
      const isAd = "SCTE35-OUT" in a || "SCTE35-CMD" in a || adClasses.includes(a.CLASS);
      const d = parseFloat(a.DURATION ?? a["PLANNED-DURATION"] ?? "NaN");
      if (isAd && Number.isFinite(d)) ads.push({ marker: line, id: a.ID, startMs: Date.parse(a["START-DATE"]), durationMs: Math.round(d * 1000) });
    } else if (line && !line.startsWith("#")) {
      segments.push({ uri: new URL(line, base).href, durationMs: duration ?? 1, programMs: program });
      if (program !== null && duration !== null) program += duration;
      duration = null;
    }
  }
  return { segments, ads, keys, map, onDemand };
}

// ---------------------------------------------------------------- the session

/**
 * One playback session of a channel (§3.2): `await AuditSession.start({ channel, owner, collector })`,
 * then `object()` for every response the player fetched, `playlist()` for each media playlist's
 * text, `observe()` as it plays, `event()` for the log's start, first presentation, stalls and end,
 * `energy()` for a role's figure, and `close(ended)` for the bundle, which `send()` posts.
 */
export class AuditSession {
  static async start({ channel, owner, collector = null, presentation = null, windowUs = DEFAULT_WINDOW_US, key = null, now = () => Date.now() * 1000, adClasses = [], cmcdEveryMs = 10_000 }) {
    const s = new AuditSession();
    s.now = now;
    s.start = now();
    s.channelName = channel;
    s.channel = await channelId(channel);
    s.owner = typeof owner === "string" ? unhex(owner) : owner;
    s.collector = collector;
    s.presentation = presentation;
    s.windowUs = Math.max(1_000_000, windowUs);
    s.key = key ?? (await newKey());
    s.session = await sessionOf(s.channel, s.key.publicKey);
    s.sid = hex(s.session);
    s.adClasses = adClasses;
    s.received = [];
    s.pending = [];
    s.byUrl = new Map();
    s.notes = [];
    s.log = [];
    s.sentCmcd = 0;
    s.spans = new ContentSpans(s.start);
    s.segments = new Map();
    s.ads = new Map(); // marker -> { marker, startMs, durationMs, recorder }
    s.keyUris = new Set();
    s.manualClaims = [];
    s.energyEntries = new Map();
    s.attachments = [];
    s.closed = null;
    if (collector && cmcdEveryMs > 0 && typeof setInterval === "function") s.timer = setInterval(() => s.flushCmcd(), cmcdEveryMs);
    return s;
  }

  st() {
    return (this.presentation ?? LIVE) === LIVE ? "l" : "v";
  }

  /** A log event: `start`, `present` (the first presentation), `stall-start`, `stall-end`, `end`, `abandon`. */
  event(name, at = this.now()) {
    if (name === "present" && this.log.some((o) => o.event === "present")) return;
    const present = name === "present" ? { unit: 0, layers: 1, first: true, late_ms: null, decode_us: null } : null;
    this.log.push(observation(this.sid, this.log.length, at, name, this.st(), { present }));
  }

  /** Whether a URL is one the session must not acknowledge: a key a playlist names (§4.1). */
  isKey(url) {
    return this.keyUris.has(url);
  }

  /**
   * A response the player received: its URL (after redirects), its bytes with content codings
   * removed, whether it is media, its status, and the CMSD fields it exposed, if any.
   */
  object({ url, bytes, media, at = this.now(), status = 200, cmsdStatic = null, cmsdDynamic = null }) {
    if (this.closed || this.isKey(url)) return Promise.resolve(null);
    // A copy, now: a player may hand the buffer on (and so empty it) as soon as this returns. The
    // object takes its place in the session in the order it is given, whenever its hash is done.
    const data = Uint8Array.from(bytes);
    const entry = { at, origin: new URL(url).origin, media, length: data.length, hash: null };
    this.received.push(entry);
    const o = observation(this.sid, this.log.length, at, "response", this.st(), { contentHash: null, rc: status });
    this.log.push(o);
    const note = cmsdStatic !== null || cmsdDynamic !== null ? { content_hash: null, cmsd_static: cmsdStatic, cmsd_dynamic: cmsdDynamic } : null;
    if (note) this.notes.push(note);
    const p = sha256(data).then((hash) => {
      entry.hash = hash;
      o.object.content_hash = hex(hash);
      if (note) note.content_hash = hex(hash);
      this.byUrl.set(url, hash);
      return hash;
    });
    this.pending.push(p);
    return p;
  }

  /** A media playlist's text: its segments and marked ads join the session's (§6.1). */
  playlist(text, url) {
    const p = parsePlaylist(text, url, { adClasses: this.adClasses });
    if (this.presentation === null) this.presentation = p.onDemand ? ON_DEMAND : LIVE;
    for (const k of p.keys) this.keyUris.add(k);
    for (const s of p.segments) this.segments.set(s.uri, s);
    for (const a of p.ads) if (!this.ads.has(a.marker)) this.ads.set(a.marker, { ...a, recorder: new AdSpans() });
    return p;
  }

  /**
   * The program time of media time `seconds`, for an on-demand presentation whose first segment
   * starts at media time 0: the first segment's program time plus it; null with no program time.
   * A player that knows the program time of its position (a live one) gives it instead.
   */
  programAt(seconds) {
    let first = null;
    for (const s of this.segments.values()) if (s.programMs !== null && (first === null || s.programMs < first)) first = s.programMs;
    return first === null ? null : first + Math.round(seconds * 1000);
  }

  /** The marked ad a program time is within, or null. */
  adAt(programMs) {
    if (programMs === null || programMs === undefined) return null;
    for (const a of this.ads.values()) if (programMs >= a.startMs && programMs < a.startMs + a.durationMs) return a;
    return null;
  }

  /** The presented playlist's segments a marked ad's span meets, in order (§6.2). */
  adSegments(ad) {
    const segs = [...this.segments.values()].filter((s) => s.programMs !== null && s.programMs < ad.startMs + ad.durationMs && s.programMs + s.durationMs > ad.startMs);
    return segs.sort((a, b) => a.programMs - b.programMs);
  }

  /** The creative of a marked ad: its segments, each by the hash of what was fetched; null while one is not (§6.2). */
  creativeOf(ad) {
    const segs = this.adSegments(ad);
    if (segs.length === 0 || segs.some((s) => !this.byUrl.has(s.uri))) return null;
    return { objects: segs.map((s) => ({ hash: this.byUrl.get(s.uri), durationMs: s.durationMs })), mediaStart: segs[0].programMs };
  }

  /** An observation of playback: `{ at, playing, visibleBp, audible, programMs, seeked }`. */
  observe({ at = this.now(), playing, visibleBp, audible, programMs = null, seeked = false }) {
    const ad = this.adAt(programMs);
    this.spans.observe({ at, playing, visibleBp, audible, ad: ad !== null });
    for (const a of this.ads.values()) {
      const segs = this.adSegments(a);
      const base = segs.length ? segs[0].programMs : a.startMs;
      if (a === ad) {
        // Played into the ad from before it: playback was continuous across its start, so its
        // first span starts at media time 0 (§6.3), as leaving it runs a span to where it left.
        const last = this.last;
        if (!a.recorder.cur && a.recorder.done.length === 0 && playing && !seeked && last?.playing && last.programMs !== null && last.programMs < base) {
          a.recorder.observe({ positionMs: 0, playing: true, visibleBp, audible });
        }
        a.recorder.observe({ positionMs: programMs - base, playing, visibleBp, audible, seeked });
      } else if (a.recorder.cur) {
        // Playback left the ad: its span runs to where it left, at most the creative's end.
        if (playing && programMs !== null && !seeked) {
          const d = segs.length ? segs.reduce((n, o) => n + o.durationMs, 0) : a.durationMs;
          a.recorder.observe({ positionMs: Math.min(programMs - base, d), playing: true, visibleBp: a.recorder.cur.visibleBp, audible: a.recorder.cur.audible });
        }
        a.recorder.observe({ positionMs: 0, playing: false, visibleBp: 0, audible: false });
      }
    }
    this.last = { playing, programMs };
  }

  /** A claim with its parts given directly: `{ marker, creative: [{ hash, durationMs }], spans, geometry }`. */
  claim(c) {
    this.manualClaims.push(c);
  }

  /** A role's figure over the session: its microjoules and its class bytes (§5.3). */
  energy(role, joulesMicro, classBytes) {
    this.energyEntries.set(role, { role, joules: joulesMicro, cls: classBytes });
  }

  /** A record to attach (§8.3): its registered name and its bytes. */
  attach(name, bytes) {
    this.attachments.push({ name, bytes });
  }

  /** Send the Event-mode reports not yet sent (§7.2). */
  async flushCmcd() {
    if (!this.collector) return;
    await Promise.all(this.pending);
    const fresh = this.log.slice(this.sentCmcd);
    const body = cmcdBatch(fresh);
    this.sentCmcd = this.log.length;
    if (!body) return;
    try {
      await fetch(new URL("v1/cmcd", this.collector), { method: "POST", headers: { "content-type": "application/cmcd" }, body, keepalive: true });
    } catch {
      // Telemetry is advisory; the record binds the log (§7.2).
    }
  }

  /** The session ends (§8.2): its records signed, and the bundle (§9). `end` defaults to now. */
  async close(ended, end = this.now()) {
    if (this.closed) return this.closed;
    if (this.timer) clearInterval(this.timer);
    await Promise.all(this.pending);
    const presentation = this.presentation ?? LIVE;
    const spans = this.spans.spans(end);
    const acked = await acknowledge(this.key, presentation, this.start, end, this.windowUs, this.received);
    const claims = [];
    for (const a of this.ads.values()) {
      const sp = a.recorder.spans();
      const c = this.creativeOf(a);
      if (sp.length > 0 && c) claims.push({ marker: a.marker, creative: c.objects, spans: sp });
    }
    claims.push(...this.manualClaims);
    const signed = [];
    for (const c of claims) {
      const objects = touched(c.creative, c.spans);
      const bytes = await signClaim(this.key, { channel: this.channel, session: this.session, marker: c.marker, creative: c.creative, objects, spans: c.spans, geometry: c.geometry, slot: c.slot });
      signed.push({ bytes, digest: await sha256(bytes.subarray(0, bytes.length - 64)), marker: c.marker, creative: c.creative });
    }
    const log = JSON.stringify([this.log]);
    const notes = this.notes.length ? JSON.stringify(this.notes) : null;
    const m = sessionMetrics(this.log);
    const origins = [...new Set(this.received.map((r) => r.origin))].sort();
    const atts = await Promise.all(this.attachments.map(async (a) => ({ ...a, digest: await sha256(a.bytes) })));
    const sorted = (ds) => ds.slice().sort(cmp);
    const energy = [...this.energyEntries.values()].sort((a, b) => a.role - b.role);
    const body = concat(
      SESSION_MAGIC, u8(1), this.channel, this.owner, this.key.publicKey, this.session, u8(presentation), u64(this.start), u64(end), u8(ENDED[ended] ?? ended),
      u8(origins.length), ...origins.flatMap((o) => [u16(o.length), enc.encode(o)]),
      u16(acked.acks.length), ...sorted(acked.acks.map((a) => a.digest)),
      u16(signed.length), ...sorted(signed.map((c) => c.digest)),
      u32(spans.length), ...spans.flatMap((s) => [u32(s.startMs), u32(s.lengthMs), u16(s.visibleBp), u8(s.flags)]),
      u32(spans.length ? m.vst ?? NO_STARTUP : NO_STARTUP), u32(m.stalls), u64(m.stallMs), u32(acked.repeatObjects), u64(acked.repeatBytes),
      await sha256(enc.encode(log)), notes ? await sha256(enc.encode(notes)) : ZERO,
      u8(energy.length), ...energy.flatMap((e) => [u8(e.role), u64(e.joules), e.cls]),
      u16(atts.length), ...sorted(atts.map((a) => a.digest)),
    );
    const record = concat(body, await sign(this.key.privateKey, SESSION_DOMAIN, body));
    this.closed = {
      v: 1,
      session: hex(record),
      acks: acked.acks.map((a) => ({ record: hex(a.record), items: a.items.map(hex) })),
      claims: signed.map((c) => ({ claim: hex(c.bytes), marker: c.marker, creative: c.creative.map((o) => ({ content_hash: hex(o.hash), duration_ms: o.durationMs })) })),
      log,
      notes,
      attachments: atts.map((a) => ({ name: a.name, record: hex(a.bytes) })),
    };
    return this.closed;
  }

  /** Post the bundle to the collector (§12); its answer. `keepalive` for a page that is going away. */
  async send({ keepalive = false } = {}) {
    if (!this.closed || !this.collector) return null;
    await this.flushCmcd();
    const res = await fetch(new URL("v1/sessions", this.collector), { method: "POST", headers: { "content-type": "text/plain;charset=UTF-8" }, body: JSON.stringify(this.closed), keepalive });
    return { status: res.status, body: await res.json().catch(() => null) };
  }
}

/**
 * Observe a media element for a session: every `intervalMs` (at most 200, §5.2) and on its events;
 * the log's first presentation, stalls and end from its `playing`, `waiting` and `ended` events;
 * a fraction of 0 while the page is hidden or out of focus. `programMs()` gives the program time of
 * the position (the playlist's `EXT-X-PROGRAM-DATE-TIME` timeline), or null. On `ended` the session
 * closes and its bundle is sent; on `pagehide` it closes as `user` and is sent with `keepalive`.
 */
export function attach(video, session, { programMs = () => null, intervalMs = 200, onBundle = null } = {}) {
  let ratio = 0;
  let seeked = false;
  let presented = false;
  let stalled = false;
  const io = new IntersectionObserver((entries) => {
    for (const e of entries) ratio = e.intersectionRatio;
  }, { threshold: Array.from({ length: 101 }, (_, i) => i / 100) });
  io.observe(video);
  const tick = () => {
    const shown = document.visibilityState === "visible" && document.hasFocus();
    session.observe({
      playing: !video.paused && !video.ended && video.readyState >= 3 && video.playbackRate === 1,
      visibleBp: shown ? ratio * 10_000 : 0,
      audible: !video.muted && video.volume > 0,
      programMs: programMs(),
      seeked,
    });
    seeked = false;
  };
  const onPlaying = () => {
    if (!presented) {
      presented = true;
      session.event("present");
    } else if (stalled) {
      stalled = false;
      session.event("stall-end");
    }
    tick();
  };
  const onWaiting = () => {
    if (presented && !stalled) {
      stalled = true;
      session.event("stall-start");
    }
    tick();
  };
  const finish = async (ended, keepalive) => {
    tick();
    stop();
    session.event("end");
    await session.close(ended);
    const answer = await session.send({ keepalive }).catch((e) => ({ status: 0, body: String(e) }));
    if (onBundle) onBundle(session.closed, answer);
  };
  const onEnded = () => finish("ended", false);
  const onHide = () => finish("user", true);
  const onSeek = () => {
    seeked = true;
    tick();
  };
  const events = ["play", "pause", "volumechange", "ratechange"];
  for (const e of events) video.addEventListener(e, tick);
  video.addEventListener("playing", onPlaying);
  video.addEventListener("waiting", onWaiting);
  video.addEventListener("seeking", onSeek);
  video.addEventListener("ended", onEnded);
  document.addEventListener("visibilitychange", tick);
  window.addEventListener("pagehide", onHide);
  const timer = setInterval(tick, Math.min(intervalMs, 200));
  session.event("start");
  function stop() {
    clearInterval(timer);
    io.disconnect();
    for (const e of events) video.removeEventListener(e, tick);
    video.removeEventListener("playing", onPlaying);
    video.removeEventListener("waiting", onWaiting);
    video.removeEventListener("seeking", onSeek);
    video.removeEventListener("ended", onEnded);
    document.removeEventListener("visibilitychange", tick);
    window.removeEventListener("pagehide", onHide);
  }
  return { stop, finish };
}