-- pg_lease 0.1.0 — prototype semantics (Architecture A).
--
-- Pure SQL: one heap table, single-statement conditional transitions,
-- clock_timestamp() time authority, lazy expiry, bounded-poll blocking
-- acquire. No C, no workers, no advisory locks.
--
-- Expiry-evaluation model: LAZY — state is a function of (stored row,
-- clock_timestamp()); every operation evaluates expiry before acting.
-- Same-owner re-acquire: idempotent success returning the current epoch,
-- expiry unchanged (spec clarification, Issue 3).
-- Renewal check instant: clock_timestamp() at the atomic transition
-- (spec clarification, Issue 2).

CREATE SCHEMA lease;

CREATE TABLE lease.leases (
    key        text        NOT NULL,
    owner      text,                       -- NULL = FREE (released; epoch history preserved)
    epoch      bigint      NOT NULL,       -- monotonic fencing token
    expires_at timestamptz,                -- NULL when owner IS NULL
    CONSTRAINT leases_key_pkey PRIMARY KEY (key),
    CONSTRAINT leases_held_shape_check CHECK (
        (owner IS NULL AND expires_at IS NULL)
        OR (owner IS NOT NULL AND expires_at IS NOT NULL)
    )
);

-- Acquire a lease.
--
-- Non-blocking (wait = 0, default): returns immediately, acquired = false
-- if the key is currently HELD and unexpired.
-- Blocking (wait > 0): polls until acquirable or the wait bound passes.
-- Idempotent: acquire by the current holder returns acquired = true with
-- the current epoch and expiry unchanged.
CREATE OR REPLACE FUNCTION lease.acquire(
    p_key   text,
    p_owner text,
    p_ttl   interval,
    p_wait  interval DEFAULT interval '0'
)
RETURNS TABLE (acquired boolean, epoch bigint, expires_at timestamptz)
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, lease
AS $fn$
DECLARE
    r           lease.leases%ROWTYPE;
    deadline    timestamptz;
    v_epoch     bigint;
    v_expires   timestamptz;
BEGIN
    IF p_key IS NULL OR p_key = '' OR p_owner IS NULL OR p_owner = ''
       OR p_ttl IS NULL OR p_ttl <= interval '0'
       OR p_wait IS NULL OR p_wait < interval '0' THEN
        RAISE EXCEPTION 'invalid_input'
            USING ERRCODE = '22023',
                  DETAIL = 'a key/owner argument is NULL or empty, ttl is not positive, or wait is negative (acquire only)';
    END IF;

    deadline := clock_timestamp() + p_wait;
    LOOP
        -- Lock-free read: a waiting acquirer must never hold the key row,
        -- or it would block release/renew by others while it polls (which
        -- would make a lease freed mid-wait unobtainable, violating S4.1).
        SELECT * INTO r FROM lease.leases
         WHERE key = p_key;

        IF NOT FOUND THEN
            -- FREE (never leased). Insert; a concurrent creator re-loops.
            INSERT INTO lease.leases AS l (key, owner, epoch, expires_at)
            VALUES (p_key, p_owner, 1, clock_timestamp() + p_ttl)
            ON CONFLICT (key) DO NOTHING;

            IF FOUND THEN
                RETURN QUERY SELECT true, 1::bigint,
                                    clock_timestamp() + p_ttl;
                RETURN;
            END IF;
            -- Lost the insert race; a holder exists. Deny or re-loop.
            IF p_wait = interval '0' THEN
                RETURN QUERY SELECT false, NULL::bigint, NULL::timestamptz;
                RETURN;
            END IF;
        ELSIF r.owner = p_owner AND r.expires_at > clock_timestamp() THEN
            -- Same-owner re-acquire of a HELD lease: idempotent, no state
            -- change (I10). A LAPSED row owned by us falls through to the
            -- takeover branch below (lazy expiry: state is a function of
            -- row + clock).
            RETURN QUERY SELECT true, r.epoch, r.expires_at;
            RETURN;
        ELSE
            -- Takeover attempt of FREE/LAPSED as a single guarded atomic
            -- statement. Concurrent acquirers are serialized by the row
            -- lock inside this UPDATE; the WHERE clause is re-evaluated
            -- against the committed row after any lock wait, so exactly
            -- one winner (I1, I2, I6). The WHERE sees the CURRENT row, so
            -- state that moved between the read and the UPDATE is honored;
            -- the epoch is derived from the current row (strictly
            -- increasing even if it changed under us). Lock is held only
            -- for this statement, never across the sleep below.
            UPDATE lease.leases
               SET owner = p_owner, epoch = leases.epoch + 1,
                   expires_at = clock_timestamp() + p_ttl
             WHERE key = p_key
               AND (lease.leases.owner IS NULL
                    OR lease.leases.expires_at <= clock_timestamp())
             RETURNING lease.leases.epoch, lease.leases.expires_at
                INTO v_epoch, v_expires;

            IF FOUND THEN
                RETURN QUERY SELECT true, v_epoch, v_expires;
                RETURN;
            END IF;

            -- HELD by another owner, unexpired (state moved under us or
            -- was already held). Non-blocking callers get a denial; a
            -- blocking caller that has exhausted its wait bound raises
            -- `timeout` (SQLSTATE 57014) so the two causes remain
            -- distinguishable per spec §7 (API review finding 1).
            IF p_wait = interval '0' THEN
                RETURN QUERY SELECT false, NULL::bigint, NULL::timestamptz;
                RETURN;
            ELSIF clock_timestamp() >= deadline THEN
                RAISE EXCEPTION 'timeout'
                    USING ERRCODE = '57014',
                          DETAIL = 'blocking acquire exceeded the wait bound';
            END IF;
        END IF;

        PERFORM pg_sleep(extract(epoch from
            least(interval '50ms', deadline - clock_timestamp()))::float8);
    END LOOP;
END;
$fn$;

-- Renew a held lease.
--
-- Statuses: 'ok', 'not_owner' (never held / free / other owner),
-- 'lapsed' (caller was the holder but the lease expired), 'epoch_mismatch'.
CREATE OR REPLACE FUNCTION lease.renew(
    p_key   text,
    p_owner text,
    p_epoch bigint,
    p_ttl   interval
)
RETURNS TABLE (status text, expires_at timestamptz)
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, lease
AS $fn$
DECLARE
    r lease.leases%ROWTYPE;
BEGIN
    IF p_key IS NULL OR p_key = '' OR p_owner IS NULL OR p_owner = ''
       OR p_epoch IS NULL
       OR p_ttl IS NULL OR p_ttl <= interval '0' THEN
        RAISE EXCEPTION 'invalid_input'
            USING ERRCODE = '22023',
                  DETAIL = 'a key/owner argument is NULL or empty, ttl is not positive, or wait is negative (acquire only)';
    END IF;

    SELECT * INTO r FROM lease.leases
     WHERE key = p_key
     FOR UPDATE;

    -- Precedence (documented): holder check, then lapse check, then epoch
    -- check. `lapsed` lets a former holder distinguish its own expiry (must
    -- re-acquire) from never-held/free (not_owner) per spec §7.
    IF NOT FOUND OR r.owner IS NULL OR r.owner <> p_owner THEN
        RETURN QUERY SELECT 'not_owner'::text, NULL::timestamptz;
        RETURN;
    ELSIF r.expires_at <= clock_timestamp() THEN
        RETURN QUERY SELECT 'lapsed'::text, NULL::timestamptz;
        RETURN;
    END IF;
    IF r.epoch <> p_epoch THEN
        RETURN QUERY SELECT 'epoch_mismatch'::text, NULL::timestamptz;
        RETURN;
    END IF;

    UPDATE lease.leases
       SET expires_at = clock_timestamp() + p_ttl
     WHERE key = p_key;

    RETURN QUERY SELECT 'ok'::text, clock_timestamp() + p_ttl;
END;
$fn$;

-- Release a held lease.
--
-- Statuses: 'ok', 'not_owner' (never held / free / other owner),
-- 'lapsed' (holder's own lease expired), 'epoch_mismatch'.
CREATE OR REPLACE FUNCTION lease.release(
    p_key   text,
    p_owner text,
    p_epoch bigint
)
RETURNS TABLE (status text)
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, lease
AS $fn$
DECLARE
    r lease.leases%ROWTYPE;
BEGIN
    IF p_key IS NULL OR p_key = '' OR p_owner IS NULL OR p_owner = ''
       OR p_epoch IS NULL THEN
        RAISE EXCEPTION 'invalid_input'
            USING ERRCODE = '22023',
                  DETAIL = 'a key/owner argument is NULL or empty, ttl is not positive, or wait is negative (acquire only)';
    END IF;

    SELECT * INTO r FROM lease.leases
     WHERE key = p_key
     FOR UPDATE;

    -- Same precedence and `lapsed` distinguishability as renew (review
    -- finding 2); a lapsed holder's release is rejected with `lapsed`.
    IF NOT FOUND OR r.owner IS NULL OR r.owner <> p_owner THEN
        RETURN QUERY SELECT 'not_owner'::text;
        RETURN;
    ELSIF r.expires_at <= clock_timestamp() THEN
        RETURN QUERY SELECT 'lapsed'::text;
        RETURN;
    END IF;
    IF r.epoch <> p_epoch THEN
        RETURN QUERY SELECT 'epoch_mismatch'::text;
        RETURN;
    END IF;

    -- FREE, preserving the epoch counter (I2: monotonic across releases).
    UPDATE lease.leases
       SET owner = NULL, epoch = r.epoch + 1, expires_at = NULL
     WHERE key = p_key;

    RETURN QUERY SELECT 'ok'::text;
END;
$fn$;

-- Inspect a lease (time-derived: a HELD row past expiry reports as free).
CREATE OR REPLACE FUNCTION lease.inspect(
    p_key text
)
RETURNS TABLE (held boolean, owner text, epoch bigint, expires_at timestamptz)
LANGUAGE sql
STABLE
AS $fn$
    SELECT owner IS NOT NULL AND expires_at > clock_timestamp(),
           CASE WHEN owner IS NOT NULL
                 AND expires_at > clock_timestamp()
                THEN owner END,
           epoch,
           CASE WHEN owner IS NOT NULL
                 AND expires_at > clock_timestamp()
                THEN expires_at END
      FROM lease.leases
     WHERE key = p_key;
$fn$;

-- The lease table is extension-internal; the functions above are the
-- only supported API (review finding: invoker-privilege gap).
REVOKE ALL ON lease.leases FROM PUBLIC;
GRANT USAGE ON SCHEMA lease TO PUBLIC;