fix(billing): reserve quota before spending, measure listening time, serialize lines

Addresses review feedback on #90.

Quota was non-atomic (P1)
  `read → +1 → put` after the provider call meant every request arriving while
  another was in flight read the same pre-spend count, passed the cap, and
  spent. Now every metered route reserves a slot BEFORE calling a provider and
  refunds it if the provider never ran, and read-modify-write is serialized per
  key. KV has no compare-and-set, so the per-key chain only covers one isolate;
  that collapses the dominant case (one learner's own concurrent calls) but
  cross-colo interleaving still needs a Durable Object — filed as follow-up.
  Applies to coach, coach/stream, pick-word, extract-words, tts, and notebook
  summaries.

Realtime listening was billed by sampling, not measuring (P1)
  The heartbeat charged a flat five minutes based only on whether playback
  happened to be paused when the timer fired: playing 4:59 then pausing cost
  nothing, resuming just before a tick cost the full five. Elapsed unpaused
  time is now accumulated, banked the moment playback flips, and flushed on
  pause, stop, and mode change so partial intervals aren't discarded. A failed
  report puts the time back rather than giving it away. The backend accepts
  fractional minutes (usage is already stored as a float) instead of rounding
  every partial minute away.

Subtitle lines could overlap (P1)
  Processing now makes background round-trips before deciding anything, so
  several lines could each clear `overlay.isOpen()` while the others awaited,
  each spend quota, then each replace the previous card. One line is processed
  at a time; while one runs only the newest arrival is held. Eligibility is
  rechecked after the awaited pick, and the card is awaited so the gate covers
  its whole lifetime rather than only up to mount.

Stale realtime socket on a cache-key flap (P2)
  A key arriving while `getWsKey()` was in flight left a stray connection whose
  later errors could stop a healthy cached session. Sessions now carry a mode
  generation that async work rechecks after every await; the reconnect backoff
  checks it too.

UI correctness (P2)
  - The limit sheet showed AI and Listening meters even when the *auto* bucket
    was what ran out, reporting balances for two features that still worked and
    none for the one that stopped. The exhausted meter now leads.
  - A failed AI-usage fetch became a real-looking 0/0 meter, so the sheet
    claimed the learner had used up "0" messages. Missing meters stay null and
    fall back to plan-neutral copy; the popup says usage is unavailable rather
    than drawing a bar from nothing.

Docs (P2)
  web/.env.example still pinned 40/150/600, overriding the new defaults for
  anyone copying it, and backend/README.md still documented free as 480 min.

Tests: reservation semantics including two concurrency cases (verified
non-vacuous by removing the lock), and the listening-billing pins extended to
cover measured time, flush-on-transition, failure rollback, and the mode-race
guard.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: Atharva-Kanherkar

backend/README.md

@@ -9,8 +9,14 @@ Pro ($8/mo), and Max ($16/mo) tiers. It does three small jobs:
    the Worker enforces per-tier caps without calling Clerk per request. Billing
    itself (Dodo → Clerk metadata) is handled in the web app, not here.
 2. **Metering** — counts listening minutes per user per calendar month in
-   Workers KV and enforces the per-tier cap (`CAP_MINUTES` free = 480/8 h,
+   Workers KV and enforces the per-tier cap (`CAP_MINUTES` free = 600/10 h,
    `PRO_CAP_MINUTES` = 1200/20 h, `MAX_CAP_MINUTES` = 3600/60 h; see `plan.ts`).
+   A session is billed by exactly one path: the shared-cache path charges the
+   real audio duration of each chunk it transcribes, while the realtime path —
+   where audio goes straight to OpenAI and never reaches this Worker — reports
+   measured unpaused playback to `POST /v1/usage/heartbeat` (fractional minutes
+   accepted, clamped to 10 per report). Running both double-charged the same
+   playback. `GET /v1/usage` returns the current listening balance.
 3. **Shared transcript cache** — users share a per-episode transcript cache.
    Cache hits return stored segments with no audio upload; cache misses are
    transcribed server-side once via Whisper and stored in Workers KV.

backend/src/index.ts

@@ -129,7 +129,12 @@ export default {
         if (!auth.ok) return auth.response;
 
         const body = (await req.json().catch(() => ({}))) as { minutes?: number };
-        const minutes = Math.max(0, Math.min(10, Math.round(Number(body.minutes) || 0)));
+        // Fractional minutes are accepted (usage is stored as a float): the
+        // client measures real unpaused playback, and rounding here threw away
+        // every partial minute — or rounded a 31-second stretch up to a whole
+        // one. Still clamped to [0, 10] per report so a bad client can't
+        // charge an arbitrary amount in one call.
+        const minutes = Math.max(0, Math.min(10, Number(body.minutes) || 0));
         const cap = capMinutesForPlan(env, effectivePlanFromProfile(auth.profile));
         let used: number;
         try {

extension/background.js

@@ -404,9 +404,12 @@
 
   // src/lib/usage-client.ts
   function meter(used, limit) {
-    const u = Math.max(0, Math.floor(Number(used) || 0));
-    const l = Math.max(0, Math.floor(Number(limit) || 0));
-    return { used: u, limit: l, left: Math.max(0, l - u) };
+    const u = Number(used);
+    const l = Number(limit);
+    if (!Number.isFinite(u) || !Number.isFinite(l)) return null;
+    const usedN = Math.max(0, Math.floor(u));
+    const limitN = Math.max(0, Math.floor(l));
+    return { used: usedN, limit: limitN, left: Math.max(0, limitN - usedN) };
   }
   async function fetchUsage() {
     const token = await getSyncToken();
@@ -424,8 +427,10 @@
     return {
       plan: raw.plan || listen.plan || "free",
       unlimited: !!raw.unlimited,
-      ai: meter(raw.ai?.used, raw.ai?.limit),
-      auto: meter(raw.auto?.used, raw.auto?.limit),
+      // Each half is independent: the AI endpoint can fail while the listening
+      // one answers (and vice versa). Whatever is missing stays null.
+      ai: aiData ? meter(raw.ai?.used, raw.ai?.limit) : null,
+      auto: aiData ? meter(raw.auto?.used, raw.auto?.limit) : null,
       listening: listenData ? meter(listen.usedMinutes, listen.capMinutes) : null,
       tiers: raw.tiers || null
     };

extension/content.js

@@ -2602,11 +2602,14 @@
     card.appendChild(title);
     card.appendChild(body);
     if (usage) {
-      const meters = [
-        ["AI messages", usage.ai.limit > 0 ? usage.ai : null, "calls"],
-        ["Listening Mode", usage.listening, "minutes"]
-      ];
-      for (const [label, m, unit] of meters) {
+      const all = {
+        ai: ["AI messages", usage.ai, "calls"],
+        auto: ["Word picking & audio", usage.auto, "calls"],
+        listening: ["Listening Mode", usage.listening, "minutes"]
+      };
+      const order = [kind, ...["ai", "auto", "listening"].filter((k) => k !== kind)];
+      for (const k of order) {
+        const [label, m, unit] = all[k];
         if (m && m.limit > 0) card.appendChild(buildMeter(label, m.used, m.limit, unit));
       }
     }
@@ -3246,6 +3249,8 @@
     let cacheKey2 = "";
     let listeningActive = false;
     let hourlyCapNotified = false;
+    let lineInFlight = false;
+    let queuedLine = null;
     let cachePollTimer = null;
     let lastCacheCueKey = "";
     let playbackRelayTimer = null;
@@ -3433,6 +3438,21 @@
     }
     async function onLine(text, context) {
       if (pipelineDisabled) return;
+      if (lineInFlight) {
+        queuedLine = { text, context };
+        return;
+      }
+      lineInFlight = true;
+      try {
+        await processLine(text, context);
+      } finally {
+        lineInFlight = false;
+      }
+      const next = queuedLine;
+      queuedLine = null;
+      if (next) await onLine(next.text, next.context);
+    }
+    async function processLine(text, context) {
       settings = await getSettings();
       if (settings.pauseMode === "off") return;
       const siteKey = adapter ? adapter.name : "generic";
@@ -3493,6 +3513,10 @@
         log("no target word in:", normalized);
         return;
       }
+      if (isOpen()) {
+        log("skipped line (card opened while picking):", normalized.slice(0, 40));
+        return;
+      }
       const stats = await getStats();
       const now = Date.now();
       const cardTimestamps = stats.cardTimestamps || [];
@@ -3519,7 +3543,7 @@
         }
       }
       log("showing card for:", target.token.base);
-      void handleCard(target, normalized, tokens, context).catch((err) => {
+      await handleCard(target, normalized, tokens, context).catch((err) => {
         warn("handleCard failed:", err);
         dismissAgent();
       });

extension/offscreen/offscreen.js

@@ -29,28 +29,59 @@
       this.code = code;
     }
   };
+  var MIN_FLUSH_MS = 3e4;
+  var HEARTBEAT_INTERVAL_MS = 5 * 60 * 1e3;
+  function accrueListening(session, now = Date.now()) {
+    if (session.billedFromMs == null) return;
+    const delta = now - session.billedFromMs;
+    if (delta > 0) session.pendingBillMs += delta;
+    session.billedFromMs = now;
+  }
+  function setListeningClock(session, playing) {
+    if (session.auth.kind !== "cloud" || session.useCache) {
+      session.billedFromMs = null;
+      return;
+    }
+    if (playing) {
+      if (session.billedFromMs == null) session.billedFromMs = Date.now();
+      return;
+    }
+    accrueListening(session);
+    session.billedFromMs = null;
+  }
+  async function flushListening(session, final = false) {
+    if (session.auth.kind !== "cloud") return;
+    accrueListening(session);
+    if (session.pendingBillMs <= 0) return;
+    if (!final && session.pendingBillMs < MIN_FLUSH_MS) return;
+    const { backendUrl, syncToken } = session.auth;
+    const minutes = Math.min(10, session.pendingBillMs / 6e4);
+    const sentMs = minutes * 6e4;
+    session.pendingBillMs = Math.max(0, session.pendingBillMs - sentMs);
+    try {
+      const res = await fetch(backendUrl + "/v1/usage/heartbeat", {
+        method: "POST",
+        headers: { "Content-Type": "application/json", Authorization: "Bearer " + syncToken },
+        body: JSON.stringify({ minutes })
+      });
+      if (res.status === 429) {
+        olog("monthly cap reached \u2014 stopping");
+        report(session.tabId, "quota-exceeded", "monthly listening hours used up");
+        stop(session.tabId);
+      }
+    } catch (err) {
+      session.pendingBillMs += sentMs;
+      olog("heartbeat failed (will retry):", String(err));
+    }
+  }
   function startHeartbeat(session) {
     if (session.auth.kind !== "cloud") return;
     if (session.useCache) return;
     if (session.heartbeat) return;
-    const { backendUrl, syncToken } = session.auth;
-    session.heartbeat = setInterval(async () => {
-      if (!session.active || session.playbackPaused) return;
-      try {
-        const res = await fetch(backendUrl + "/v1/usage/heartbeat", {
-          method: "POST",
-          headers: { "Content-Type": "application/json", Authorization: "Bearer " + syncToken },
-          body: JSON.stringify({ minutes: 5 })
-        });
-        if (res.status === 429) {
-          olog("monthly cap reached \u2014 stopping");
-          report(session.tabId, "quota-exceeded", "monthly listening hours used up");
-          stop(session.tabId);
-        }
-      } catch (err) {
-        olog("heartbeat failed (will retry):", String(err));
-      }
-    }, 5 * 60 * 1e3);
+    setListeningClock(session, session.active && !session.playbackPaused);
+    session.heartbeat = setInterval(() => {
+      void flushListening(session);
+    }, HEARTBEAT_INTERVAL_MS);
   }
   function closeRealtimeSocket(session) {
     const ws = session.ws;
@@ -67,11 +98,14 @@
     }
   }
   function applyCacheMode(session) {
+    session.modeGeneration += 1;
     if (session.useCache) {
       if (session.heartbeat) {
         clearInterval(session.heartbeat);
         session.heartbeat = null;
       }
+      void flushListening(session, true);
+      session.billedFromMs = null;
       closeRealtimeSocket(session);
       if (!session.chunkTimer) {
         session.chunkTimer = setInterval(() => {
@@ -106,8 +140,10 @@
   }
   function onPlaybackUpdate(session, time, paused) {
     const prev = session.playbackTime;
+    const wasPaused = session.playbackPaused;
     session.playbackTime = time;
     session.playbackPaused = paused;
+    if (paused !== wasPaused) setListeningClock(session, !paused);
     if (Math.abs(time - prev) > SEEK_THRESHOLD_SEC) {
       olog("playback seek detected", prev, "\u2192", time, "\u2014 resetting audio buffer");
       resetAudioBuffer(session, time);
@@ -285,7 +321,10 @@
       chunkStarted: false,
       transcribing: false,
       chunkTimer: null,
-      useCache
+      useCache,
+      billedFromMs: null,
+      pendingBillMs: 0,
+      modeGeneration: 0
     };
     sessions[tabId] = session;
     proc.onaudioprocess = (e) => {
@@ -315,7 +354,13 @@
     applyCacheMode(session);
   }
   async function connectWS(session) {
+    const generation = session.modeGeneration;
+    const stale = () => !session.active || session.useCache || session.modeGeneration !== generation;
     const wsKey = await getWsKey(session);
+    if (stale()) {
+      olog("dropping realtime connect \u2014 session switched modes while authorizing");
+      return;
+    }
     let ws;
     try {
       ws = new WebSocket(RT_URL, ["realtime", "openai-insecure-api-key." + wsKey]);
@@ -376,7 +421,7 @@
       if (ev.code !== 4001 && session.reconnects < 3) {
         session.reconnects += 1;
         setTimeout(() => {
-          if (session.active) {
+          if (session.active && !session.useCache) {
             connectWS(session).catch((err) => {
               const code = err instanceof CodedError ? err.code : "capture-failed";
               report(session.tabId, code, String(err && err.message || err));
@@ -394,6 +439,8 @@
     const session = sessions[tabId];
     if (!session) return;
     session.active = false;
+    setListeningClock(session, false);
+    void flushListening(session, true);
     if (session.heartbeat) clearInterval(session.heartbeat);
     if (session.chunkTimer) clearInterval(session.chunkTimer);
     if (session.useCache && session.pcmBuffer.length && !session.playbackPaused) {

extension/popup/popup.js

@@ -413,6 +413,9 @@
     const fmt = (n) => unit === "minutes" ? `${Number.isInteger(n / 60) ? n / 60 : (n / 60).toFixed(1)}h` : n.toLocaleString();
     return `<div class="av-meter"><div class="av-meter-row"><span>${esc(label)}</span><span class="av-meter-val">${esc(fmt(Math.min(used, limit)))} / ${esc(fmt(limit))}</span></div><div class="av-meter-track"><div class="${cls}" style="width:${pct}%"></div></div></div>`;
   }
+  function meterLow(m) {
+    return !!m && m.limit > 0 && m.used / m.limit >= 0.8;
+  }
   async function renderUsage() {
     const el = byId("usage");
     const token = await getSyncToken();
@@ -431,9 +434,10 @@
       return;
     }
     const planName = usage.plan === "max" ? "Max" : usage.plan === "pro" ? "Pro" : "Free";
-    const meters = usage.unlimited ? `<p class="av-usage-note">No caps on this account.</p>` : meterMarkup("AI messages", usage.ai.used, usage.ai.limit, "calls") + (usage.listening ? meterMarkup("Listening Mode", usage.listening.used, usage.listening.limit, "minutes") : "");
-    const aiLow = !usage.unlimited && usage.ai.limit > 0 && usage.ai.used / usage.ai.limit >= 0.8;
-    const listenLow = !usage.unlimited && !!usage.listening && usage.listening.limit > 0 && usage.listening.used / usage.listening.limit >= 0.8;
+    const bars = usage.unlimited ? "" : (usage.ai ? meterMarkup("AI messages", usage.ai.used, usage.ai.limit, "calls") : "") + (usage.listening ? meterMarkup("Listening Mode", usage.listening.used, usage.listening.limit, "minutes") : "");
+    const meters = usage.unlimited ? `<p class="av-usage-note">No caps on this account.</p>` : bars || `<p class="av-usage-note">Usage is unavailable right now.</p>`;
+    const aiLow = !usage.unlimited && meterLow(usage.ai);
+    const listenLow = !usage.unlimited && meterLow(usage.listening);
     const offer = usage.plan === "free" ? usage.tiers?.pro : usage.plan === "pro" ? usage.tiers?.max : null;
     const cta = (aiLow || listenLow) && offer?.checkoutUrl ? `<button id="usage-upgrade" class="av-btn av-btn-primary av-btn-block av-usage-cta" type="button">Upgrade to ${esc(offer.name)} \u2014 ${esc(offer.priceLabel)}</button>` : "";
     el.innerHTML = `<div class="av-usage-head"><span class="av-usage-title">This month</span><span class="av-usage-plan">${esc(planName)}</span></div>` + meters + cta;

src/entries/content.ts

@@ -49,6 +49,9 @@ declare global {
   let listeningActive = false;
   /** Rate-limits the "hourly card cap" notice to once per rolling window. */
   let hourlyCapNotified = false;
+  /** One subtitle line processed at a time; see onLine. */
+  let lineInFlight = false;
+  let queuedLine: { text: string; context?: LineContext } | null = null;
   let cachePollTimer: ReturnType<typeof setInterval> | null = null;
   let lastCacheCueKey = "";
   let playbackRelayTimer: ReturnType<typeof setInterval> | null = null;
@@ -269,8 +272,35 @@ declare global {
     return candidate;
   }
 
+  /**
+   * Subtitle lines arrive faster than a line takes to process, and processing
+   * now makes background round-trips (word extraction, word picking) before it
+   * decides anything. Left unserialized, several lines could each clear the
+   * `overlay.isOpen()` check while the others were awaiting, each spend quota,
+   * and then each replace the previous card — `presentWord()` dismisses whatever
+   * is up. So: one line in flight at a time, and while one is running only the
+   * newest arrival is held. Older queued lines are dropped on purpose; their
+   * moment on screen has passed, and showing a card for them would be wrong
+   * even if it were free.
+   */
   async function onLine(text: string, context?: LineContext): Promise<void> {
     if (pipelineDisabled) return;
+    if (lineInFlight) {
+      queuedLine = { text, context };
+      return;
+    }
+    lineInFlight = true;
+    try {
+      await processLine(text, context);
+    } finally {
+      lineInFlight = false;
+    }
+    const next = queuedLine;
+    queuedLine = null;
+    if (next) await onLine(next.text, next.context);
+  }
+
+  async function processLine(text: string, context?: LineContext): Promise<void> {
 
     settings = await storage.getSettings();
 
@@ -340,6 +370,14 @@ declare global {
     );
     if (!target) { log("no target word in:", normalized); return; }
 
+    // pickTargetSmart just awaited a network round-trip; a card may have opened
+    // in the meantime (a review can be triggered from the panel). Re-check
+    // rather than trusting the read from before the await.
+    if (overlay.isOpen()) {
+      log("skipped line (card opened while picking):", normalized.slice(0, 40));
+      return;
+    }
+
     const stats = await storage.getStats();
     const now = Date.now();
     const cardTimestamps = stats.cardTimestamps || [];
@@ -371,7 +409,11 @@ declare global {
     }
     log("showing card for:", target.token.base);
 
-    void handleCard(target, normalized, tokens, context).catch((err) => {
+    // Awaited, so the in-flight gate covers the card's whole lifetime — not just
+    // up to the moment it mounts. Previously this was fire-and-forget, leaving a
+    // window where the next line was already picking a word before the card had
+    // rendered and `overlay.isOpen()` could see it.
+    await handleCard(target, normalized, tokens, context).catch((err) => {
       warn("handleCard failed:", err);
       overlay.dismissAgent();
     });