Playback flow and paywall

/** Stable per viewer AND device, meaningless outside your systems (no raw user id, no personal data). */
export function viewerRef(userId: string, installId: string, secret: string): string {
  const h = (msg: string, n: number) => createHmac('sha256', secret).update(msg).digest('base64url').slice(0, n);
  return `${h(`user:${userId}`, 22)}:${h(`device:${installId}`, 11)}`;
}

/** Free when episode_no <= free_until_episode (null = no free episodes); otherwise your paywall decides. */
export async function mayWatch(d: PlayDeps, userId: string, filmId: string, episodeNo: number): Promise<'ok' | 'locked' | 'not_playable'> {
  const film = await d.film(filmId);
  const ep = await d.episode(filmId, episodeNo);
  // Placeholder rows (id null), episodes not stream_ready and withdrawn rows never reach Wandura.
  if (!film || !ep || !ep.id || !ep.stream_ready || !ep.available) return 'not_playable';
  const free = film.free_until_episode !== null && episodeNo <= film.free_until_episode;
  return free || (await d.hasUnlocked(userId, filmId, episodeNo)) ? 'ok' : 'locked';
}

export async function handlePlay(r: PlayRequest, d: PlayDeps): Promise<PlayResponse> {
  // 1. Your paywall. A locked episode never reaches Wandura.
  const may = await mayWatch(d, r.userId, r.filmId, r.episodeNo);
  if (may !== 'ok') return { ok: false, outcome: may === 'locked' ? 'locked' : 'not_ready' };

  // 2. Up to two following episodes the viewer may also watch. Stop at the first one that is not.
  const following: number[] = [];
  for (let k = 1; k <= 2 && (await mayWatch(d, r.userId, r.filmId, r.episodeNo + k)) === 'ok'; k++) following.push(r.episodeNo + k);

  // 3. One batch: item 0 is the play, the rest are prefetches. Always send the play — it is what counts the view —
  //    even if the app already started the episode with a prefetched ticket.
  const viewer_ref = viewerRef(r.userId, r.installId, d.viewerRefSecret);
  const body = {
    episodes: [
      { film_ref: r.filmId, episode_no: r.episodeNo, intent: 'play' },
      ...following.map((n) => ({ film_ref: r.filmId, episode_no: n, intent: 'prefetch' })),
    ],
    union_info: r.unionInfo,
    drm_kek: r.drmKek,
    viewer_ref, // stable per viewer + device: a retry gets the SAME ticket back, not a second one
  };
  let items: BatchItem[];
  try {
    items = await requestBatch(body, d.call);
  } catch (e) {
    if (e instanceof WanduraApiError) return { ok: false, outcome: handleTicketError(e).outcome };
    throw e;
  }

  const [anchor, ...rest] = items;
  if (!anchor || anchor.status !== 200) {
    const err = new WanduraApiError({ message: anchor?.error ?? 'no item', status: anchor?.status ?? null, errorCode: anchor?.error_code, reason: anchor?.reason });
    const h = handleTicketError(err);
    log('ticket_refused', { status: err.status, error_code: err.errorCode, reason: err.reason, viewer_ref, alert: h.alert });
    return { ok: false, outcome: h.outcome };
  }
  // 4. Only `private-encrypted` may be sold. Anything else: do not play it, tell Wandura.
  const current = asTicket(anchor);
  if (!current || current.protection !== 'private-encrypted') {
    log('ticket_unexpected_protection', { protection: anchor.protection, viewer_ref });
    return { ok: false, outcome: 'unavailable' };
  }
  // A refused or failed prefetch is not an error: the app asks again when the viewer gets there.
  const next = rest.flatMap((it, i) => {
    const t = it.status === 200 ? asTicket(it) : null;
    const n = following[i];
    return t && n !== undefined && t.protection === 'private-encrypted' ? [{ episode_no: n, ticket: t }] : [];
  });
  return { ok: true, current, next };
}

/** POST /playback. If only item 0 failed with a retryable status, retry once after its own `retry_after`. */
async function requestBatch(body: object, o: CallOptions = {}): Promise<BatchItem[]> {
  for (let attempt = 0; ; attempt++) {
    const { data } = await call<{ items: BatchItem[] }>('POST', '/api/delivery/v1/playback', { policy: INTERACTIVE, ...o, body });
    const a = data.items[0];
    if (!a || a.status === 200 || attempt >= 1 || !isRetryable(a.status, a.error_code)) return data.items;
    const wait = (a.retry_after ?? 1) * 1_000 + Math.random() * 1_000;
    await (o.sleep ?? ((ms: number) => new Promise<void>((res) => setTimeout(res, ms))))(wait);
  }
}

function asTicket(it: BatchItem): Ticket | null {
  if (!it.vid || !it.play_auth_token || !it.expires_at || !it.protection) return null;
  return { vid: it.vid, play_auth_token: it.play_auth_token, expires_at: it.expires_at, protection: it.protection, subtitles: it.subtitles ?? [] };
}

/** YOUR codes for your app. Never pass Wandura's `error` text to viewers. */
export type AppOutcome =
  | 'unavailable' // licence not active — "This title is no longer available"
  | 'not_ready' // not playable yet — "Not available yet"
  | 'busy' // retried already — "Busy, try again in a moment"
  | 'broken'; // a bug on your side — "Couldn't load this episode"

/** `alert` = page your engineers. `confirm: 'catalog'` = re-check with the catalog sync before an irreversible step. */
export type Handling = { outcome: AppOutcome; alert: boolean; confirm?: 'catalog' };

export function handleTicketError(e: WanduraApiError): Handling {
  switch (e.errorCode) {
    case 'not_entitled': return { outcome: 'unavailable', alert: false, confirm: 'catalog' };
    case 'not_ready': return { outcome: 'not_ready', alert: false };
    case 'rate_limited':
    case 'service_unavailable': return { outcome: 'busy', alert: false };
    case 'sdk_required':
    case 'invalid_device_info':
    case 'invalid_request':
    case 'unsupported_version': return { outcome: 'broken', alert: true };
  }
  // Unknown or missing `error_code`: decide by status class.
  const s = e.status;
  if (s === null || s === 429 || s >= 500) return { outcome: 'busy', alert: s !== null && s >= 500 };
  return { outcome: 'broken', alert: true };
}

def handle_play(user_id: str, install_id: str, film: dict, episodes_by_no: Dict[int, dict], episode_no: int,
                union_info: str, drm_kek: str, unlocked: Callable[[int], bool], secret: bytes, **call_opts: Any) -> Dict[str, Any]:
    decision = may_watch(film, episodes_by_no.get(episode_no), lambda: unlocked(episode_no))
    if decision != "ok":
        return {"ok": False, "outcome": "locked" if decision == "locked" else "not_ready"}
    following: List[int] = []
    for n in (episode_no + 1, episode_no + 2):
        if may_watch(film, episodes_by_no.get(n), lambda n=n: unlocked(n)) != "ok":
            break
        following.append(n)

    ref = viewer_ref(user_id, install_id, secret)
    body = {  # item 0 = the play (counts the view), the rest = prefetches
        "episodes": [{"film_ref": film["id"], "episode_no": episode_no, "intent": "play"}]
        + [{"film_ref": film["id"], "episode_no": n, "intent": "prefetch"} for n in following],
        "union_info": union_info,
        "drm_kek": drm_kek,
        "viewer_ref": ref,  # stable per viewer + device: a retry gets the same ticket back
    }
    try:
        data, _ = call("POST", "/api/delivery/v1/playback", body=body, policy=INTERACTIVE, **call_opts)
    except WanduraApiError as e:
        return {"ok": False, **handle_ticket_error(e.status, e.error_code)}

    items = data.get("items") or []
    anchor = items[0] if items else {}
    if anchor.get("status") != 200:
        log("ticket_refused", status=anchor.get("status"), error_code=anchor.get("error_code"), reason=anchor.get("reason"), viewer_ref=ref)
        return {"ok": False, **handle_ticket_error(anchor.get("status"), anchor.get("error_code"))}
    if anchor.get("protection") != "private-encrypted":  # only private-encrypted may be sold
        log("ticket_unexpected_protection", protection=anchor.get("protection"), viewer_ref=ref)
        return {"ok": False, "outcome": "unavailable", "alert": True}
    keep = ("vid", "play_auth_token", "expires_at", "subtitles")
    nxt = [{"episode_no": n, "ticket": {k: it.get(k) for k in keep}}
           for n, it in zip(following, items[1:])
           if it.get("status") == 200 and it.get("protection") == "private-encrypted"]  # refused prefetch = not an error
    return {"ok": True, "current": {k: anchor.get(k) for k in keep}, "next": nxt}

_OUTCOMES = {
    "not_entitled": ("unavailable", False),  # confirm with the catalog sync before you stop selling
    "not_ready": ("not_ready", False),
    "rate_limited": ("busy", False),
    "service_unavailable": ("busy", False),
    "sdk_required": ("broken", True),
    "invalid_device_info": ("broken", True),
    "invalid_request": ("broken", True),
    "unsupported_version": ("broken", True),
}


def handle_ticket_error(status: Optional[int], error_code: Optional[str]) -> Dict[str, Any]:
    """Your app's outcome for a failed ticket. Never show Wandura's `error` text to viewers."""
    if error_code in _OUTCOMES:
        outcome, alert = _OUTCOMES[error_code]
    elif status is None or status == 429 or status >= 500:  # unknown code: decide by status class
        outcome, alert = "busy", status is not None and status >= 500
    else:
        outcome, alert = "broken", True
    return {"outcome": outcome, "alert": alert}