-- porch/middleware/keypool.wo — the key pool: an actor per shard, picked by
-- hash of the key, that serializes rate-limit counting (this file, kind 1)
-- and idempotency begin (Task 4, kind 2) against the @table rows in
-- store.wo. This is the only file that knows a pool exists — the
-- middlewares call through it and never touch RateLimitCounter
-- themselves. IdempotencyKey is the one exception: the response has to
-- travel through that table (a Resp cannot ride the mailbox — see
-- below), so idempotent.wo reads the row a begin call already committed.
--
-- `call`'s reply crosses the actor boundary as a single copyable scalar
-- (WO-E226 — no class, no Text can ride it). The exact count is decided
-- atomically inside `receive`; `pool_count` packs it with the window's
-- remaining time into one Int and unpacks that into the `Verdict` callers
-- actually read, so the packing never leaks outside this file. kind 2
-- (Task 4) reuses the exact same pool_pack scheme for its outcome code —
-- WO-E226 forces every `receive` in the program to agree on one return
-- type, so a second encoding is not an option.

use time

-- To a pool actor. kind 1 = count (this file); kind 2 = begin (Task 4
-- fills in the arm — the fields below are already shaped for it: the
-- bare idempotency key travels in `key`, the body digest in `digest`,
-- and the actor runs `handler` against `req` itself so a duplicate waits
-- in the mailbox rather than needing a held reply).
-- To a pool actor. `kind` is kept even though only one kind exists today:
-- idempotency's `kind: 2` arm was built, reviewed and then REVERTED (see
-- archive/porch-idempotency), and it will come back. Adding a second kind is
-- a field and an `if`, not a redesign.
class PoolMsg {
  kind:   Int
  key:    Text
  limit:  Int     -- count: max requests per window
  window: Int     -- count: window size, µs
}

-- What the limiter reads back from a count. `allowed` and `limit` are
-- filled in by `pool_count` — the caller already knows `limit`, it is the
-- one it sent. `count` and `reset_at` come from the actor.
class Verdict {
  allowed:  Bool
  count:    Int
  limit:    Int
  reset_at: Int -- wall-clock ms (time.now()) when this key's window resets
}

-- PoolMsg requires `req`/`handler` on every construction (an actor
-- NOTE: this file used to carry NullHandler, dummy_req() and fresh_req().
-- They existed ONLY because PoolMsg had to carry a Req and a Handler for
-- idempotency's kind-2 arm, which meant every rate-limit count allocated a
-- throwaway Req (four maps) it never read. With that arm reverted the
-- placeholders go too, and counting stops paying for a feature it never
-- used. They are preserved with the arm in archive/porch-idempotency.

-- Packs (count, remaining-ms-in-window) into one Int: count * 1e9 +
-- remaining_ms, remaining_ms clamped to stay under 1e9 (~11.5 days —
-- far past any realistic rate-limit window). That clamp only blurs the
-- advisory reset header on an absurdly long window; it never touches the
-- count, which is the correctness-critical half.
fn pool_pack(count: Int, remaining_ms: Int) -> Int {
  let r = remaining_ms;
  if r < 0 { r = 0; }
  if r >= 1_000_000_000 { r = 999_999_999; }
  return count * 1_000_000_000 + r;
}

-- One actor per shard. Reads the row for the key, decides, and writes the
-- new count by assigning to the row's field — that writes through and
-- maintains indexes; never delete-then-insert as an update of the SAME
-- row (the window prune below IS a delete-then-insert, but of a fresh
-- row for the new window — the stale row is retired, not mutated).
class KeyActor {
  fn receive(msg: PoolMsg) -> Int {

    -- Only kind 1 exists today. The `kind: 2` arm — idempotency, where the
    -- actor ran the route handler inside this receive so a duplicate waited
    -- in the mailbox — was built, reviewed and then REVERTED. It is whole in
    -- the tag archive/porch-idempotency, which doubles as the reproduction
    -- harness for the C-runtime crash that caused the revert: a SIGSEGV in
    -- wo_arena_alloc / wo_str_new under concurrent call()-parked callers.
    -- That arm allocated 5x what this one does inside receive and moved a
    -- whole Req plus a Handler through the mailbox; over ten gate runs every
    -- failure was one of its legs, and none were this one's.
    -- kind 1: count.
    let now = time.ticks();
    let hits = from c in RateLimitCounter where c.key == msg.key take 1 select c;

    if len(hits) == 0 {
      insert RateLimitCounter { key: msg.key, count: 1, window: now };
      return pool_pack(1, msg.window / 1000);
    }

    let row = hits[0];
    if now - row.window > msg.window {
      -- the window fully elapsed: prune the stale row rather than reset it
      -- in place — resetting keeps one row forever for every key ever
      -- seen, an unbounded leak for IP-keyed limiting. There is no
      -- sweeper; this lazy expiry on access is it.
      delete row;
      insert RateLimitCounter { key: msg.key, count: 1, window: now };
      return pool_pack(1, msg.window / 1000);
    }

    row.count = row.count + 1;
    let remaining_us = row.window + msg.window - now;
    if remaining_us < 0 { remaining_us = 0; }
    return pool_pack(row.count, remaining_us / 1000);
  }
}

class PoolSlot {
  a: actor PoolMsg
}

class Pool {
  actors: multi PoolSlot
}

-- Spawns n identical actors and returns the pool. n is a capacity knob:
-- too small and a hot key's mailbox saturates under load (a `call` trap,
-- answered 503 by the middleware — never a silent bypass). n < 1 is a
-- caller misconfiguration, not a capacity choice, and guarding it HERE
-- (not in pool_select's division) is what matters: every pool_select call
-- runs inside the middleware's own `try ... catch (e) { print_err(...);
-- nil }`, so a mod-by-zero trap there would be misreported as ordinary
-- 503 saturation forever (though now at least logged, not silently
-- swallowed), never surfacing the real bug on its own.
pub fn make_pool(n: Int) -> Pool {
  let count = n;
  if count < 1 { count = 1; }
  let actors: multi PoolSlot = [];
  let i = 0;
  while i < count {
    push(actors, PoolSlot { a: spawn KeyActor {} });
    i = i + 1;
  }
  return Pool { actors: actors };
}

-- Pool itself is demand-promoted to traced (WO-E222) the moment an app
-- aliases it — e.g. Limiter/Idempotent's own `pool: Pool` field, read on
-- every request without being consumed — so it can never live in an
-- actor's state or a message. PoolSlot is not: WO-E222's contains_traced
-- check only recurses into a field typed as a class name (or a `multi`/
-- `map` of one); `a: actor PoolMsg` is an actor handle, a different case
-- entirely, so it never pulls PoolSlot (or `multi PoolSlot`) into the
-- traced set the way wrapping it in Pool does. An actor CAN hold `multi
-- PoolSlot` directly in its own state — the exact shape chat/main.wo's
-- `Room { members: multi Mem }` already uses for a multi of actor
-- handles — which is what makes real per-connection sharding possible:
-- call make_pool(n) ONCE at process start, hand pool_slots(pool) to every
-- connection actor's spawn, and each one rebuilds a transient Pool via
-- pool_of(self.slots) wherever Limiter/Idempotent needs one. Calling
-- make_pool per connection instead (the natural misreading of this pair
-- sitting right after a capacity-sizing knob) gives every connection its
-- own actors and silently restores the lost-increment race this whole
-- design exists to prevent.
--
-- Both functions copy field-by-field, the same trick fresh_req uses above:
-- an actor handle is a plain, freely-copyable scalar (not traced), so
-- rebuilding each PoolSlot by value produces a list with no lingering
-- alias into the traced Pool (pool_slots) or the caller's own copy
-- (pool_of) — never a value some other reader could still be holding.
pub fn pool_slots(p: Pool) -> multi PoolSlot {
  let out: multi PoolSlot = [];
  for s in p.actors { push(out, PoolSlot { a: s.a }); }
  return out;
}

pub fn pool_of(s: multi PoolSlot) -> Pool {
  let out: multi PoolSlot = [];
  for x in s { push(out, PoolSlot { a: x.a }); }
  return Pool { actors: out };
}

-- Hashes a key to one of the pool's actors — sum of bytes modulo n, a
-- shard selector, not a security hash. The same key always selects the
-- same actor, which is the entire per-key serialization mechanism.
pub fn pool_select(pool: Pool, key: Text) -> actor PoolMsg {
  let sum = 0;
  let i = 0;
  while i < len(key) {
    sum = sum + byte_at(key, i);
    i = i + 1;
  }
  let idx = sum % len(pool.actors);
  return pool.actors[idx].a;
}

-- The count accessor every later task's limiter calls. Unpacks the
-- actor's scalar reply into the Verdict the limiter reads.
pub fn pool_count(pool: Pool, key: Text, limit: Int, window: Int) -> Verdict {
  let a = pool_select(pool, key);
  let raw = call(a, PoolMsg { kind: 1, key: key, limit: limit, window: window });
  let count = raw / 1_000_000_000;
  let remaining_ms = raw % 1_000_000_000;
  return Verdict {
    allowed:  count <= limit,
    count:    count,
    limit:    limit,
    reset_at: time.now() + remaining_ms
  };
}

-- NOTE: pool_begin() lived here — the accessor idempotent.wo called to run a
-- request through the actor. Reverted with the kind-2 arm; whole in the tag
-- archive/porch-idempotency.