-- es_saga_shard_leases — shard ownership for the worker fleet (plan §P4.2).
--
-- A worker must hold the `(saga_name, shard_id)` lease to OWN that shard's
-- scans (timeouts, signal drain, recovery). Ownership is *work distribution
-- only* — it decides which worker scans which partition of instances. It is
-- NOT a correctness fence: the per-instance lease (es_saga_leases) still
-- guarantees effects never run concurrently, and OCC still fences the log.
-- So a shard owned by two workers (during a rebalance) is safe — just
-- duplicated scan effort — and a shard owned by nobody for a tick just
-- delays its scans, never drops them. Hence: no fence token here (unlike
-- mire's es_projection_leases), unlike the instance lease this is best-effort
-- coordination, kept deliberately boring.
--
-- Acquisition is the same single-statement steal-if-expired pattern as
-- es_saga_leases: INSERT ... ON CONFLICT DO UPDATE ... WHERE leased_until
-- has lapsed (or it is already ours), so racing acquirers serialise on the
-- PK row lock and the loser sees the winner's fresh deadline.
CREATE TABLE IF NOT EXISTS es_saga_shard_leases (
    saga_name    TEXT        NOT NULL,
    shard_id     INTEGER     NOT NULL,
    worker_id    TEXT        NOT NULL,
    leased_until TIMESTAMPTZ NOT NULL,
    heartbeat_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (saga_name, shard_id)
);

-- Free/expired-shard scan (acquisition candidates) filters on
-- (saga_name, leased_until).
CREATE INDEX IF NOT EXISTS idx_saga_shard_leases_live
    ON es_saga_shard_leases (saga_name, leased_until);

-- es_saga_worker_heartbeats — fleet membership, decoupled from assignment.
--
-- The shard-share denominator is "how many workers are alive?", NOT "how
-- many shards are currently held". If it were the latter, a brand-new
-- worker that holds zero shards would be invisible to incumbents, who would
-- compute target = num_shards (they look like the sole holder) and never
-- shed — the newcomer could never get in (it can't steal a live shard
-- lease). So each worker heartbeats its presence here every tick regardless
-- of how many shards it holds; `target = ceil(num_shards / live_workers)`
-- uses this count, so incumbents make room the moment a new worker appears.
-- A worker that stops heartbeating ages out of the count within the
-- liveness window and its shards (also lapsing) are reabsorbed.
CREATE TABLE IF NOT EXISTS es_saga_worker_heartbeats (
    saga_name    TEXT        NOT NULL,
    worker_id    TEXT        NOT NULL,
    heartbeat_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (saga_name, worker_id)
);

-- Live-worker count: WHERE saga_name = $1 AND heartbeat_at > now() - window.
CREATE INDEX IF NOT EXISTS idx_saga_worker_heartbeats_live
    ON es_saga_worker_heartbeats (saga_name, heartbeat_at);
