-- Inactive poll-history V1-to-V2 retirement. Application owner authenticates
-- both ciphertext versions; SQL independently fences identity and lifetime.
BEGIN;
LOCK TABLE public.server_poll_archives,public.server_poll_storage_admissions IN ACCESS EXCLUSIVE MODE;
ALTER TABLE public.server_poll_archives DROP CONSTRAINT server_poll_archives_version_check;
ALTER TABLE public.server_poll_archives ADD CONSTRAINT server_poll_archives_version_check
 CHECK(version=1 OR (version=2 AND decoded_bytes<=24576 AND compressed_bytes<=24576 AND ciphertext_bytes<=33000));

CREATE OR REPLACE FUNCTION public.server_poll_archive_immutable() RETURNS trigger
LANGUAGE plpgsql SET TimeZone='UTC' SET search_path=pg_catalog,public AS $$
BEGIN
 IF TG_OP='DELETE' AND public.server_poll_storage_gc_allowed(OLD.receipt_id) THEN RETURN OLD; END IF;
 IF TG_OP='UPDATE' AND OLD.version=1 AND NEW.version=2
   AND current_setting('mm.poll_retire_receipt',true)=OLD.receipt_id::text
   AND NEW.ciphertext_bytes<=OLD.ciphertext_bytes
   AND NEW.ciphertext ~ '^[A-Za-z0-9+/]+={0,2}$'
   AND replace(encode(decode(NEW.ciphertext,'base64'),'base64'),E'\n','')=NEW.ciphertext
   AND octet_length(decode(NEW.ciphertext,'base64'))=NEW.compressed_bytes+28
   AND (to_jsonb(OLD)-ARRAY['version','source_digest','source_sha256','decoded_bytes','compressed_bytes','ciphertext','ciphertext_bytes','archive_proof'])=
       (to_jsonb(NEW)-ARRAY['version','source_digest','source_sha256','decoded_bytes','compressed_bytes','ciphertext','ciphertext_bytes','archive_proof'])
   AND EXISTS(
    SELECT 1 FROM public.server_poll_storage_admissions a JOIN public.mcp_operations o ON o.id=a.execution_id AND o.connection_id=a.connection_id
    JOIN public.mcp_plans p ON p.id=o.plan_id AND p.connection_id=o.connection_id
    JOIN public.mcp_operation_steps s ON s.operation_id=o.id AND s.step_index=0
    JOIN public.mcp_operation_replay_tombstones t ON t.operation_id=o.id AND t.plan_id=p.id AND t.connection_id=o.connection_id
    JOIN public.mcp_connections c ON c.id=o.connection_id AND c.owner_id=a.owner_id
    WHERE a.receipt_id=OLD.receipt_id AND a.identity_kind='mcp' AND a.storage_state='compacted'
     AND a.execution_id=OLD.execution_id AND a.connection_id=OLD.connection_id AND a.owner_id=OLD.owner_id
     AND a.binding_proof=OLD.binding_proof AND a.context_proof=OLD.context_proof AND a.archive_bytes=OLD.ciphertext_bytes
     AND a.original_activity_at=OLD.original_activity_at AND a.authority_expires_at=OLD.original_expires_at
     AND o.state='applied' AND o.verification='passed' AND o.error_code IS NULL AND isfinite(o.finished_at)
     AND o.finished_at<transaction_timestamp()-interval '180 days'
     AND p.state='completed' AND p.operation_id='servers.poll' AND p.operation_version=1 AND p.approved_by IS NOT NULL
     AND s.state='verified' AND s.finished_at IS NOT NULL AND t.action_id=p.operation_id
     AND t.target_global AND cardinality(t.target_server_ids)>0
   ) THEN RETURN NEW; END IF;
 RAISE EXCEPTION 'poll archive immutable outside exact retirement or original lifetime GC';
END $$;

CREATE FUNCTION public.server_poll_retirement_final_guard() RETURNS trigger
LANGUAGE plpgsql SET search_path=pg_catalog,public AS $$
BEGIN
 IF NOT EXISTS(SELECT 1 FROM public.mcp_operation_replay_tombstones t
    WHERE t.operation_id=NEW.execution_id AND t.connection_id=NEW.connection_id AND t.action_id='servers.poll'
     AND t.target_global AND NOT EXISTS(SELECT 1 FROM public.mcp_plans p WHERE p.id=t.plan_id))
  OR EXISTS(SELECT 1 FROM public.mcp_operations o WHERE o.id=NEW.execution_id)
  OR NOT public.server_poll_storage_accounting_valid()
 THEN RAISE EXCEPTION 'poll retirement is incomplete'; END IF;
 RETURN NULL;
END $$;
CREATE CONSTRAINT TRIGGER server_poll_retirement_final_guard AFTER UPDATE ON public.server_poll_archives
 DEFERRABLE INITIALLY DEFERRED FOR EACH ROW WHEN(OLD.version=1 AND NEW.version=2)
 EXECUTE FUNCTION public.server_poll_retirement_final_guard();

CREATE OR REPLACE FUNCTION server_poll_storage_admission_guard() RETURNS trigger LANGUAGE plpgsql SET TimeZone='UTC' SET search_path=pg_catalog,public AS $$
DECLARE retention integer; charged integer; r public.server_poll_archives%ROWTYPE;
BEGIN
 PERFORM pg_advisory_xact_lock(hashtextextended('server-poll-storage-v1',0));
 IF TG_OP='DELETE' THEN
  IF NOT public.server_poll_storage_gc_allowed(OLD.receipt_id) OR EXISTS(SELECT 1 FROM public.server_poll_executions WHERE receipt_id=OLD.receipt_id)
   OR EXISTS(SELECT 1 FROM public.server_poll_intents WHERE receipt_id=OLD.receipt_id) OR EXISTS(SELECT 1 FROM public.server_poll_acks WHERE receipt_id=OLD.receipt_id)
   OR EXISTS(SELECT 1 FROM public.server_poll_receipts WHERE receipt_id=OLD.receipt_id) OR EXISTS(SELECT 1 FROM public.server_poll_quarantines WHERE receipt_id=OLD.receipt_id)
   OR EXISTS(SELECT 1 FROM public.server_poll_archives WHERE receipt_id=OLD.receipt_id) THEN RAISE EXCEPTION 'poll lifetime or children prevent admission deletion'; END IF;
  UPDATE public.server_poll_storage_usage SET admissions=admissions-CASE WHEN OLD.storage_state='full' THEN 1 ELSE 0 END,
   compacted_admissions=compacted_admissions-CASE WHEN OLD.storage_state='compacted' THEN 1 ELSE 0 END,
   archive_bytes=archive_bytes-COALESCE(OLD.archive_bytes,0),
   reserved_rows=reserved_rows-CASE WHEN OLD.storage_state='full' THEN 4019 ELSE 2 END,
   reserved_bytes=reserved_bytes-CASE WHEN OLD.storage_state='full' THEN 8421376 ELSE 32768+OLD.archive_bytes END
   WHERE scope IN('global','owner:'||OLD.owner_id);
  GET DIAGNOSTICS charged=ROW_COUNT; IF charged<>2 THEN RAISE EXCEPTION 'poll quota refund mismatch'; END IF; RETURN OLD;
 END IF;
 IF TG_OP='UPDATE' THEN
  IF OLD.storage_state='compacted' AND NEW.storage_state='compacted'
   AND current_setting('mm.poll_retire_receipt',true)=OLD.receipt_id::text
   AND (to_jsonb(OLD)-'archive_bytes')=(to_jsonb(NEW)-'archive_bytes') THEN
   SELECT * INTO r FROM public.server_poll_archives WHERE receipt_id=OLD.receipt_id AND version=2;
   IF NOT FOUND OR r.ciphertext_bytes<>NEW.archive_bytes OR NEW.archive_bytes>OLD.archive_bytes
    OR r.execution_id<>OLD.execution_id OR r.connection_id<>OLD.connection_id OR r.owner_id<>OLD.owner_id
   THEN RAISE EXCEPTION 'retired poll charge source mismatch'; END IF;
   UPDATE public.server_poll_storage_usage
    SET archive_bytes=archive_bytes-OLD.archive_bytes+NEW.archive_bytes,
        reserved_bytes=reserved_bytes-OLD.archive_bytes+NEW.archive_bytes
    WHERE scope IN ('global','owner:'||OLD.owner_id) AND compacted_admissions>=1 AND archive_bytes>=OLD.archive_bytes;
   GET DIAGNOSTICS charged=ROW_COUNT;
   IF charged<>2 THEN RAISE EXCEPTION 'retired poll charge adjustment mismatch'; END IF;
   RETURN NEW;
  END IF;
  -- This durable one-shot marker is written and ACKed in its OWN transaction
  -- before archive construction. A committed/unknown claim never resets and
  -- the worker never retries destructive work against an already claimed full
  -- source. Claim time is maintenance metadata, not renewed execution activity.
  IF OLD.storage_state='full' AND NEW.storage_state='full' AND OLD.compaction_claim IS NULL AND OLD.compaction_claimed_at IS NULL AND OLD.compaction_claim_xid IS NULL
   AND NEW.compaction_claim IS NOT NULL AND NEW.compaction_claim_xid IS NULL
   AND(to_jsonb(OLD)-ARRAY['compaction_claim','compaction_claimed_at','compaction_claim_xid'])=(to_jsonb(NEW)-ARRAY['compaction_claim','compaction_claimed_at','compaction_claim_xid'])
   AND public.server_poll_storage_compact_allowed(OLD.receipt_id)
  THEN NEW.compaction_claimed_at:=clock_timestamp(); NEW.compaction_claim_xid:=pg_current_xact_id(); RETURN NEW; END IF;
  IF(to_jsonb(OLD)-ARRAY['claim_token','claim_attempted_at','used_rows','used_bytes','storage_state','compact_at','archive_bytes','original_activity_at']) IS DISTINCT FROM
    (to_jsonb(NEW)-ARRAY['claim_token','claim_attempted_at','used_rows','used_bytes','storage_state','compact_at','archive_bytes','original_activity_at']) THEN RAISE EXCEPTION 'poll admission identity is immutable'; END IF;
  IF OLD.storage_state<>'full' THEN RAISE EXCEPTION 'compacted poll identity cannot reopen or change'; END IF;
  IF NEW.storage_state='compacted' THEN
   SELECT * INTO r FROM public.server_poll_archives WHERE receipt_id=OLD.receipt_id AND compaction_xid=pg_current_xact_id();
   IF NOT FOUND OR NEW.used_rows<>OLD.used_rows OR NEW.used_bytes<>OLD.used_bytes OR NEW.claim_token IS DISTINCT FROM OLD.claim_token
    OR NEW.claim_attempted_at IS DISTINCT FROM OLD.claim_attempted_at OR NEW.archive_bytes<>r.ciphertext_bytes OR NEW.original_activity_at<>r.original_activity_at
    OR EXISTS(SELECT 1 FROM public.server_poll_executions WHERE receipt_id=OLD.receipt_id)
    OR EXISTS(SELECT 1 FROM public.server_poll_intents WHERE receipt_id=OLD.receipt_id) OR EXISTS(SELECT 1 FROM public.server_poll_acks WHERE receipt_id=OLD.receipt_id)
    OR EXISTS(SELECT 1 FROM public.server_poll_receipts WHERE receipt_id=OLD.receipt_id) OR EXISTS(SELECT 1 FROM public.server_poll_quarantines WHERE receipt_id=OLD.receipt_id)
   THEN RAISE EXCEPTION 'poll compaction requires the complete current-transaction archive'; END IF;
   NEW.compact_at:=r.created_at;
   UPDATE public.server_poll_storage_usage SET admissions=admissions-1,compacted_admissions=compacted_admissions+1,
    archive_bytes=archive_bytes+r.ciphertext_bytes,reserved_rows=reserved_rows-4019+2,reserved_bytes=reserved_bytes-8421376+32768+r.ciphertext_bytes
    WHERE scope IN('global','owner:'||OLD.owner_id) AND admissions>=1;
   GET DIAGNOSTICS charged=ROW_COUNT; IF charged<>2 THEN RAISE EXCEPTION 'poll compaction charge mismatch'; END IF;
   RETURN NEW;
  END IF;
  IF NEW.compact_at IS NOT NULL OR NEW.archive_bytes IS NOT NULL OR NEW.original_activity_at IS NOT NULL THEN RAISE EXCEPTION 'full poll cannot carry archive metadata'; END IF;
  IF OLD.claim_token IS NOT NULL AND NEW.claim_token IS NULL AND OLD.claim_attempted_at IS NULL AND NEW.claim_attempted_at IS NOT NULL
   AND NEW.used_rows=OLD.used_rows AND NEW.used_bytes=OLD.used_bytes AND OLD.authority_expires_at>clock_timestamp()
  THEN NEW.claim_attempted_at:=clock_timestamp(); RETURN NEW; END IF;
  IF OLD.claim_token IS NOT DISTINCT FROM NEW.claim_token AND OLD.claim_attempted_at IS NOT DISTINCT FROM NEW.claim_attempted_at
   AND NEW.used_rows=OLD.used_rows+1 AND NEW.used_bytes>OLD.used_bytes THEN RETURN NEW; END IF;
  RAISE EXCEPTION 'poll admission cannot refresh, reopen or refund itself';
 END IF;
 IF NEW.storage_state<>'full' OR NEW.compact_at IS NOT NULL OR NEW.archive_bytes IS NOT NULL OR NEW.original_activity_at IS NOT NULL
  OR NEW.compaction_claim IS NOT NULL OR NEW.compaction_claimed_at IS NOT NULL OR NEW.compaction_claim_xid IS NOT NULL
  OR NOT public.server_poll_storage_accounting_valid() THEN RAISE EXCEPTION 'new poll requires full reservation and exact ledger'; END IF;
 SELECT requests_retention_days INTO retention FROM public.mcp_settings WHERE id=1;
 IF retention IS NULL OR retention NOT BETWEEN 1 AND 365 OR NEW.authority_issued_at>clock_timestamp() OR NEW.authority_expires_at<=clock_timestamp()
  OR NEW.claim_token IS NULL OR NEW.claim_attempted_at IS NOT NULL OR NEW.used_rows<>0 OR NEW.used_bytes<>0 THEN RAISE EXCEPTION 'new poll admission needs original live authority'; END IF;
 IF EXISTS(SELECT 1 FROM public.server_poll_storage_admissions WHERE receipt_id=NEW.receipt_id OR(identity_kind=NEW.identity_kind AND execution_id=NEW.execution_id)) THEN RAISE EXCEPTION 'poll identity already spent'; END IF;
 NEW.created_at:=clock_timestamp(); NEW.retention_days:=retention; NEW.safety_ms:=60000;
 NEW.gc_not_before:=NEW.authority_expires_at+make_interval(days=>retention)+interval '60 seconds';
 INSERT INTO public.server_poll_storage_usage(scope) VALUES('owner:'||NEW.owner_id) ON CONFLICT DO NOTHING;
 IF(SELECT count(*) FROM public.server_poll_storage_usage WHERE scope IN('global','owner:'||NEW.owner_id)
  AND admissions<CASE WHEN scope='global' THEN 128 ELSE 8 END
  AND reserved_rows<=CASE WHEN scope='global' THEN 514432 ELSE 32152 END-4019
  AND reserved_bytes<=CASE WHEN scope='global' THEN 1077936128 ELSE 67371008 END-8421376)<>2 THEN RAISE EXCEPTION 'poll admission quota exhausted'; END IF;
 UPDATE public.server_poll_storage_usage SET admissions=admissions+1,reserved_rows=reserved_rows+4019,reserved_bytes=reserved_bytes+8421376 WHERE scope IN('global','owner:'||NEW.owner_id);
 GET DIAGNOSTICS charged=ROW_COUNT; IF charged<>2 THEN RAISE EXCEPTION 'poll admission quota unavailable'; END IF;
 RETURN NEW;
END $$;
CREATE OR REPLACE FUNCTION server_poll_storage_gc_allowed(id uuid) RETURNS boolean LANGUAGE plpgsql SET TimeZone='UTC' SET search_path=pg_catalog,public AS $$
DECLARE a public.server_poll_storage_admissions%ROWTYPE; retention integer; activity timestamptz; terminal_at timestamptz;
BEGIN
 IF current_setting('mm.poll_storage_gc',true) IS DISTINCT FROM 'on' THEN RETURN false; END IF;
 PERFORM pg_advisory_xact_lock(hashtextextended('server-poll-storage-v1',0));
 SELECT * INTO a FROM public.server_poll_storage_admissions WHERE receipt_id=id FOR UPDATE;
 IF NOT FOUND OR a.authority_expires_at>=clock_timestamp() OR a.gc_not_before>=clock_timestamp() THEN RETURN false; END IF;
 SELECT requests_retention_days INTO retention FROM public.mcp_settings WHERE mcp_settings.id=1 FOR SHARE;
 IF retention IS NULL OR retention NOT BETWEEN 1 AND 365 THEN RETURN false; END IF;
 terminal_at:=a.created_at;
 IF a.identity_kind='mcp' THEN
  IF NOT pg_try_advisory_xact_lock(hashtextextended('mcp-operation:'||a.execution_id::text,0)) THEN RETURN false; END IF;
  BEGIN
   SELECT GREATEST(o.finished_at,p.expires_at) INTO terminal_at FROM public.mcp_operations o
    JOIN public.mcp_plans p ON p.id=o.plan_id AND p.connection_id=o.connection_id JOIN public.mcp_connections c ON c.id=o.connection_id
    WHERE o.id=a.execution_id AND o.connection_id=a.connection_id AND c.owner_id=a.owner_id AND p.operation_id='servers.poll'
      AND o.state IN('applied','failed','unknown','rejected','partial','cancelled') AND o.finished_at IS NOT NULL FOR UPDATE OF o,p NOWAIT;
   IF NOT FOUND THEN
    IF a.storage_state<>'compacted' OR EXISTS(SELECT 1 FROM public.mcp_operations o WHERE o.id=a.execution_id) OR NOT EXISTS (
      SELECT 1 FROM public.mcp_operation_replay_tombstones t JOIN public.mcp_connections c ON c.id=t.connection_id
      WHERE t.operation_id=a.execution_id AND t.connection_id=a.connection_id AND c.owner_id=a.owner_id
       AND t.action_id='servers.poll' AND t.target_global AND cardinality(t.target_server_ids)>0
       AND NOT EXISTS(SELECT 1 FROM public.mcp_plans WHERE mcp_plans.id=t.plan_id)
    ) OR EXISTS(SELECT 1 FROM public.server_poll_archives r WHERE r.receipt_id=id AND
      (r.version<>2 OR r.execution_id<>a.execution_id OR r.connection_id<>a.connection_id OR r.owner_id<>a.owner_id
       OR r.original_activity_at IS DISTINCT FROM a.original_activity_at OR r.original_expires_at<>a.authority_expires_at
       OR r.binding_proof<>a.binding_proof OR r.context_proof<>a.context_proof OR r.ciphertext_bytes<>a.archive_bytes))
    THEN RETURN false; END IF;
    terminal_at:=a.original_activity_at;
   END IF;
   IF terminal_at IS NULL THEN RETURN false; END IF;
  EXCEPTION WHEN lock_not_available THEN RETURN false; END;
 END IF;
 IF a.storage_state='compacted' THEN activity:=GREATEST(a.original_activity_at,a.authority_expires_at,terminal_at);
 ELSE
  SELECT GREATEST(a.authority_expires_at,a.created_at,COALESCE(a.claim_attempted_at,a.created_at),terminal_at,
   COALESCE((SELECT created_at FROM public.server_poll_executions WHERE receipt_id=id),a.created_at),
   COALESCE((SELECT max(created_at) FROM public.server_poll_intents WHERE receipt_id=id),a.created_at),
   COALESCE((SELECT max(created_at) FROM public.server_poll_acks WHERE receipt_id=id),a.created_at),
   COALESCE((SELECT created_at FROM public.server_poll_receipts WHERE receipt_id=id),a.created_at),
   COALESCE((SELECT created_at FROM public.server_poll_quarantines WHERE receipt_id=id),a.created_at)) INTO activity;
 END IF;
 RETURN isfinite(activity) AND activity+make_interval(days=>GREATEST(a.retention_days,retention))+a.safety_ms*interval '1 millisecond'<clock_timestamp();
END $$;
INSERT INTO public.schema_migrations(id) VALUES ('0046_server_poll_operation_retirement') ON CONFLICT(id) DO NOTHING;
COMMIT;