-- 018: WP3 publishing execution spine (fences, attempt journal, receipts).
-- Preserves 006: same drafts / scheduled_publications / publications tables,
-- same PKs, UNIQUE(draft_id) on schedules and global idempotency_key
-- uniqueness. Adds the scheduling principal, revision/approval proof,
-- credential generation/revocation, execution generation and the durable
-- attempt journal the scheduler claims, fences and finalizes against.
--
-- The current runner (db/migrate.ts) wraps each file in ONE transaction, so
-- this file is written to be transactional end to end: no CREATE INDEX
-- CONCURRENTLY, bounded by lock_timeout, preflight fails closed on corrupt
-- tenant references instead of silently reparenting them. Quiesce the
-- publishing worker for the upgrade (plan §2 "Existing data rollout").
SET LOCAL lock_timeout = '5s';

-- ---------------------------------------------------------------------------
-- Preflight: refuse to upgrade data that the tenant-composite FKs below would
-- reject anyway, with an operator-readable reason. Nothing is reparented.
-- ---------------------------------------------------------------------------
DO $$
DECLARE v_bad bigint;
BEGIN
  SELECT count(*) INTO v_bad FROM scheduled_publications sp
   WHERE NOT EXISTS (SELECT 1 FROM drafts d WHERE d.id = sp.draft_id AND d.workspace_id = sp.workspace_id)
      OR NOT EXISTS (SELECT 1 FROM connections c WHERE c.id = sp.connection_id AND c.workspace_id = sp.workspace_id);
  IF v_bad > 0 THEN
    RAISE EXCEPTION '018 preflight: % scheduled_publications rows reference a draft/connection in another workspace', v_bad;
  END IF;
  SELECT count(*) INTO v_bad FROM publications p
   WHERE NOT EXISTS (SELECT 1 FROM drafts d WHERE d.id = p.draft_id AND d.workspace_id = p.workspace_id)
      OR NOT EXISTS (SELECT 1 FROM connections c WHERE c.id = p.connection_id AND c.workspace_id = p.workspace_id);
  IF v_bad > 0 THEN
    RAISE EXCEPTION '018 preflight: % publications rows reference a draft/connection in another workspace', v_bad;
  END IF;
  SELECT count(*) INTO v_bad FROM credentials cr
   WHERE cr.connection_id IS NOT NULL
     AND NOT EXISTS (SELECT 1 FROM connections c WHERE c.id = cr.connection_id AND c.workspace_id = cr.workspace_id);
  IF v_bad > 0 THEN
    RAISE EXCEPTION '018 preflight: % credentials rows reference a connection in another workspace', v_bad;
  END IF;
  SELECT count(*) INTO v_bad FROM (
    SELECT 1 FROM credentials WHERE kind = 'connector-oauth' AND connection_id IS NOT NULL
     GROUP BY workspace_id, connection_id HAVING count(*) > 1
  ) dup;
  IF v_bad > 0 THEN
    RAISE EXCEPTION '018 preflight: % connections have more than one connector-oauth credential; reconcile before upgrading', v_bad;
  END IF;
END $$;

-- ---------------------------------------------------------------------------
-- Credentials: generation + revocation. key_version is NOT a token
-- generation (APV1 sealing writes version 1 on every rotation).
-- ---------------------------------------------------------------------------
ALTER TABLE credentials
  ADD COLUMN generation bigint NOT NULL DEFAULT 1 CHECK (generation > 0),
  ADD COLUMN revoked_at timestamptz;
ALTER TABLE credentials ADD CONSTRAINT credentials_id_workspace_key UNIQUE (id, workspace_id);
-- One live connector credential per connection; readers stop using LIMIT 1.
CREATE UNIQUE INDEX credentials_connector_oauth_unique
  ON credentials (workspace_id, connection_id) WHERE kind = 'connector-oauth';

-- generation is DB-owned: any secret-material or lifecycle change rotates it,
-- and callers cannot rewind it.
CREATE FUNCTION credentials_bump_generation() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
  IF NEW.ciphertext IS DISTINCT FROM OLD.ciphertext
     OR NEW.key_version IS DISTINCT FROM OLD.key_version
     OR NEW.rotated_at IS DISTINCT FROM OLD.rotated_at
     OR NEW.revoked_at IS DISTINCT FROM OLD.revoked_at THEN
    NEW.generation := OLD.generation + 1;
  ELSE
    NEW.generation := OLD.generation;
  END IF;
  RETURN NEW;
END $$;

CREATE TRIGGER trg_cred_generation BEFORE UPDATE ON credentials
  FOR EACH ROW EXECUTE FUNCTION credentials_bump_generation();
-- Credentials are an execution-authorization input: changes bump the epoch
-- exactly like memberships (reuses bump_authz_epoch, 001/005).
CREATE TRIGGER trg_cred_bump_epoch
  AFTER INSERT OR DELETE OR UPDATE OF ciphertext, key_version, rotated_at, revoked_at ON credentials
  FOR EACH ROW EXECUTE FUNCTION bump_authz_epoch();

-- ---------------------------------------------------------------------------
-- Connections: tombstones keep journal evidence intact after removal.
-- ---------------------------------------------------------------------------
ALTER TABLE connections ADD COLUMN disconnected_at timestamptz;
CREATE TRIGGER trg_conn_disconnect_bump_epoch AFTER UPDATE OF disconnected_at ON connections
  FOR EACH ROW WHEN (NEW.disconnected_at IS DISTINCT FROM OLD.disconnected_at)
  EXECUTE FUNCTION bump_authz_epoch();

-- Group sharing gates userCanUseConnection, so it bumps the epoch too (008
-- only covered group_members).
CREATE TRIGGER trg_group_connection_bump_epoch AFTER INSERT OR UPDATE OR DELETE ON group_connections
  FOR EACH ROW EXECUTE FUNCTION bump_authz_epoch();

-- role_permissions is global. Every workspace's Owner/Admin/... roles come
-- from it, so a change bumps all workspaces once per statement, locked in
-- UUID order to avoid lock-order inversions between concurrent bumps.
CREATE FUNCTION bump_all_authz_epochs() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
  PERFORM 1 FROM workspaces ORDER BY id FOR UPDATE;
  UPDATE workspaces SET authz_epoch = authz_epoch + 1;
  RETURN NULL;
END $$;
CREATE TRIGGER trg_role_permissions_bump_epoch AFTER INSERT OR UPDATE OR DELETE ON role_permissions
  FOR EACH STATEMENT EXECUTE FUNCTION bump_all_authz_epochs();

-- ---------------------------------------------------------------------------
-- Drafts: revision + approval proof + business attempt counter.
-- ---------------------------------------------------------------------------
ALTER TABLE drafts
  ADD COLUMN revision bigint NOT NULL DEFAULT 1 CHECK (revision > 0),
  ADD COLUMN approved_revision bigint,
  ADD COLUMN approved_hash text,
  ADD COLUMN approved_at timestamptz,
  ADD COLUMN review_reason text,
  ADD COLUMN publication_attempt bigint NOT NULL DEFAULT 0 CHECK (publication_attempt >= 0);
-- drafts_id_workspace_key (id, workspace_id) already exists from 015.

-- ---------------------------------------------------------------------------
-- Scheduled publications: principal/fence snapshot + execution generation.
-- ---------------------------------------------------------------------------
ALTER TABLE scheduled_publications
  ADD COLUMN created_at timestamptz NOT NULL DEFAULT now(),
  ADD COLUMN updated_at timestamptz NOT NULL DEFAULT now(),
  ADD COLUMN scheduled_by_user_id uuid REFERENCES users(id) ON DELETE NO ACTION ON UPDATE RESTRICT,
  ADD COLUMN required_permission text
    CHECK (required_permission IN ('publication.schedule', 'publication.publish')),
  ADD COLUMN authorized_epoch bigint,
  ADD COLUMN approved_revision bigint,
  ADD COLUMN post_hash text,
  ADD COLUMN credential_id uuid,
  ADD COLUMN credential_generation bigint,
  ADD COLUMN generation bigint NOT NULL DEFAULT 1 CHECK (generation > 0),
  ADD COLUMN attempt_no bigint NOT NULL DEFAULT 1 CHECK (attempt_no > 0),
  ADD COLUMN execution_version smallint NOT NULL DEFAULT 0 CHECK (execution_version IN (0, 1)),
  ADD COLUMN timezone text NOT NULL DEFAULT 'UTC',
  ADD COLUMN local_publish_at timestamp,
  ADD COLUMN tzdata_version text,
  ADD COLUMN next_attempt_at timestamptz,
  ADD COLUMN retry_count int NOT NULL DEFAULT 0 CHECK (retry_count >= 0),
  ADD COLUMN error_code text;

ALTER TABLE scheduled_publications DROP CONSTRAINT scheduled_publications_state_check;
ALTER TABLE scheduled_publications ADD CONSTRAINT scheduled_publications_state_check
  CHECK (state IN ('pending', 'in_flight', 'done', 'failed', 'needs_reauthorization', 'cancelled'));

-- Legacy rows: timestamps from the draft, due time from publish_at, UTC zone.
UPDATE scheduled_publications sp
   SET created_at = d.created_at, updated_at = d.updated_at, next_attempt_at = sp.publish_at
  FROM drafts d
 WHERE d.id = sp.draft_id AND d.workspace_id = sp.workspace_id;

-- Existing publications are authoritative: their schedule is done, draft published.
UPDATE scheduled_publications sp SET state = 'done', error = NULL, error_code = NULL
 WHERE EXISTS (SELECT 1 FROM publications p
                WHERE p.workspace_id = sp.workspace_id
                  AND (p.idempotency_key = sp.idempotency_key OR p.draft_id = sp.draft_id));
UPDATE drafts d SET state = 'published'
 WHERE d.state <> 'published'
   AND EXISTS (SELECT 1 FROM publications p WHERE p.workspace_id = d.workspace_id AND p.draft_id = d.id);
-- No recorded principal: never infer one from author/audit history.
UPDATE scheduled_publications SET state = 'needs_reauthorization', error_code = 'LEGACY_AUTHORIZATION_MISSING'
 WHERE state = 'pending';
-- In flight without a publication: the send may have happened.
UPDATE scheduled_publications SET state = 'failed', error_code = 'PUBLISH_OUTCOME_UNKNOWN'
 WHERE state = 'in_flight';

-- Upgraded (execution_version = 1) rows must carry every execution fence
-- while executable or done. Legacy rows (0) are readable, never executable:
-- every claim requires execution_version = 1.
ALTER TABLE scheduled_publications ADD CONSTRAINT scheduled_publications_fences_check CHECK (
  execution_version = 0
  OR state IN ('failed', 'needs_reauthorization', 'cancelled')
  OR (scheduled_by_user_id IS NOT NULL AND required_permission IS NOT NULL
      AND authorized_epoch IS NOT NULL AND approved_revision IS NOT NULL
      AND post_hash IS NOT NULL AND credential_id IS NOT NULL
      AND credential_generation IS NOT NULL AND next_attempt_at IS NOT NULL)
);

ALTER TABLE scheduled_publications
  ADD CONSTRAINT scheduled_publications_id_workspace_key UNIQUE (id, workspace_id);

-- Tenant-composite replacements for 006's single-column FKs (same ON DELETE
-- semantics; journal tables below add NO ACTION guards).
ALTER TABLE scheduled_publications
  DROP CONSTRAINT scheduled_publications_draft_id_fkey,
  DROP CONSTRAINT scheduled_publications_connection_id_fkey,
  ADD CONSTRAINT scheduled_publications_draft_ws_fkey FOREIGN KEY (draft_id, workspace_id)
    REFERENCES drafts (id, workspace_id) ON DELETE CASCADE ON UPDATE RESTRICT,
  ADD CONSTRAINT scheduled_publications_connection_ws_fkey FOREIGN KEY (connection_id, workspace_id)
    REFERENCES connections (id, workspace_id) ON DELETE NO ACTION ON UPDATE RESTRICT,
  ADD CONSTRAINT scheduled_publications_credential_ws_fkey FOREIGN KEY (credential_id, workspace_id)
    REFERENCES credentials (id, workspace_id) ON DELETE NO ACTION ON UPDATE RESTRICT;

ALTER TABLE publications
  DROP CONSTRAINT publications_draft_id_fkey,
  DROP CONSTRAINT publications_connection_id_fkey,
  ADD CONSTRAINT publications_draft_ws_fkey FOREIGN KEY (draft_id, workspace_id)
    REFERENCES drafts (id, workspace_id) ON DELETE CASCADE ON UPDATE RESTRICT,
  ADD CONSTRAINT publications_connection_ws_fkey FOREIGN KEY (connection_id, workspace_id)
    REFERENCES connections (id, workspace_id) ON DELETE NO ACTION ON UPDATE RESTRICT;

-- Schedules: DB-owned updated_at; generation and execution_version only move forward.
CREATE FUNCTION scheduled_publications_guard() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
  IF NEW.generation < OLD.generation THEN
    RAISE EXCEPTION 'schedule generation cannot decrease' USING ERRCODE = '23514';
  END IF;
  IF NEW.execution_version < OLD.execution_version THEN
    RAISE EXCEPTION 'schedule execution_version cannot decrease' USING ERRCODE = '23514';
  END IF;
  NEW.updated_at := now();
  RETURN NEW;
END $$;
CREATE TRIGGER trg_scheduled_pub_guard BEFORE UPDATE ON scheduled_publications
  FOR EACH ROW EXECUTE FUNCTION scheduled_publications_guard();

-- ---------------------------------------------------------------------------
-- Execution journal: one row per business attempt. Snapshot columns are
-- immutable; mutable columns carry ownership (owner_token/claim_epoch/lease)
-- and the dispatch marker.
-- ---------------------------------------------------------------------------
CREATE TABLE publishing_attempts (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE ON UPDATE RESTRICT,
  schedule_id uuid NOT NULL,
  draft_id uuid NOT NULL,
  connection_id uuid NOT NULL,
  attempt_no bigint NOT NULL CHECK (attempt_no > 0),
  idempotency_key text NOT NULL CHECK (btrim(idempotency_key) <> ''),
  schedule_generation bigint NOT NULL CHECK (schedule_generation > 0),
  post_snapshot jsonb NOT NULL CHECK (jsonb_typeof(post_snapshot) = 'object'),
  post_hash text NOT NULL CHECK (btrim(post_hash) <> ''),
  scheduled_by_user_id uuid NOT NULL REFERENCES users(id) ON DELETE NO ACTION ON UPDATE RESTRICT,
  required_permission text NOT NULL
    CHECK (required_permission IN ('publication.schedule', 'publication.publish')),
  approved_revision bigint NOT NULL CHECK (approved_revision > 0),
  reviewer_user_id uuid NOT NULL REFERENCES users(id) ON DELETE NO ACTION ON UPDATE RESTRICT,
  credential_id uuid NOT NULL,
  credential_generation bigint NOT NULL CHECK (credential_generation > 0),
  authorized_epoch bigint NOT NULL,
  dispatch_count int NOT NULL DEFAULT 0 CHECK (dispatch_count >= 0 AND dispatch_count <= 5),
  status text NOT NULL DEFAULT 'ready'
    CHECK (status IN ('ready', 'in_progress', 'completed', 'failed', 'unknown', 'abandoned')),
  claim_epoch bigint NOT NULL DEFAULT 0 CHECK (claim_epoch >= 0),
  owner_token uuid,
  lease_until timestamptz,
  dispatch_started_at timestamptz,
  provider_retry_deadline timestamptz,
  error_code text,
  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now(),
  -- Ownership exists exactly while in progress.
  CHECK ((status = 'in_progress') = (owner_token IS NOT NULL AND lease_until IS NOT NULL)),
  CHECK (status = 'in_progress' OR (owner_token IS NULL AND lease_until IS NULL)),
  -- A dispatch marker implies at least one counted dispatch.
  CHECK (dispatch_started_at IS NULL OR dispatch_count > 0),
  UNIQUE (workspace_id, id),
  UNIQUE (workspace_id, draft_id, attempt_no),
  UNIQUE (workspace_id, idempotency_key),
  FOREIGN KEY (schedule_id, workspace_id) REFERENCES scheduled_publications (id, workspace_id)
    ON DELETE NO ACTION ON UPDATE RESTRICT,
  FOREIGN KEY (draft_id, workspace_id) REFERENCES drafts (id, workspace_id)
    ON DELETE NO ACTION ON UPDATE RESTRICT,
  FOREIGN KEY (connection_id, workspace_id) REFERENCES connections (id, workspace_id)
    ON DELETE NO ACTION ON UPDATE RESTRICT,
  FOREIGN KEY (credential_id, workspace_id) REFERENCES credentials (id, workspace_id)
    ON DELETE NO ACTION ON UPDATE RESTRICT
);

-- Snapshot immutability, terminal statuses, and DB-owned claim epoch: every
-- new owner token advances claim_epoch by exactly one, so a deposed owner's
-- (owner_token, claim_epoch) pair can never match again.
CREATE FUNCTION publishing_attempts_guard() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
  IF (NEW.schedule_id, NEW.draft_id, NEW.connection_id, NEW.attempt_no, NEW.idempotency_key,
      NEW.schedule_generation, NEW.post_snapshot, NEW.post_hash, NEW.scheduled_by_user_id,
      NEW.required_permission, NEW.approved_revision, NEW.reviewer_user_id,
      NEW.credential_id, NEW.credential_generation, NEW.created_at)
     IS DISTINCT FROM
     (OLD.schedule_id, OLD.draft_id, OLD.connection_id, OLD.attempt_no, OLD.idempotency_key,
      OLD.schedule_generation, OLD.post_snapshot, OLD.post_hash, OLD.scheduled_by_user_id,
      OLD.required_permission, OLD.approved_revision, OLD.reviewer_user_id,
      OLD.credential_id, OLD.credential_generation, OLD.created_at) THEN
    RAISE EXCEPTION 'publishing attempt snapshot is immutable' USING ERRCODE = '23514';
  END IF;
  IF OLD.status IN ('completed', 'abandoned') AND NEW.status IS DISTINCT FROM OLD.status THEN
    RAISE EXCEPTION 'publishing attempt % is terminal', OLD.status USING ERRCODE = '23514';
  END IF;
  IF NEW.dispatch_count < OLD.dispatch_count THEN
    RAISE EXCEPTION 'dispatch_count cannot decrease' USING ERRCODE = '23514';
  END IF;
  IF NEW.provider_retry_deadline IS DISTINCT FROM OLD.provider_retry_deadline
     AND OLD.provider_retry_deadline IS NOT NULL THEN
    RAISE EXCEPTION 'provider_retry_deadline is fixed at first dispatch' USING ERRCODE = '23514';
  END IF;
  IF NEW.owner_token IS NOT NULL AND NEW.owner_token IS DISTINCT FROM OLD.owner_token THEN
    NEW.claim_epoch := OLD.claim_epoch + 1;
  ELSE
    NEW.claim_epoch := OLD.claim_epoch;
  END IF;
  NEW.updated_at := now();
  RETURN NEW;
END $$;

CREATE TRIGGER trg_publishing_attempt_ws_immutable BEFORE UPDATE ON publishing_attempts
  FOR EACH ROW EXECUTE FUNCTION reject_workspace_change();
CREATE TRIGGER trg_publishing_attempt_guard BEFORE UPDATE ON publishing_attempts
  FOR EACH ROW EXECUTE FUNCTION publishing_attempts_guard();

-- Durable provider evidence, separate from the attempt so it survives activity
-- timeouts and stale-owner completion refusal. Append-only.
CREATE TABLE publishing_receipts (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE ON UPDATE RESTRICT,
  attempt_id uuid NOT NULL,
  remote_id text NOT NULL CHECK (btrim(remote_id) <> '' AND length(remote_id) <= 512),
  remote_url text CHECK (remote_url IS NULL OR length(remote_url) <= 2048),
  published_at timestamptz,
  provider_metadata jsonb NOT NULL DEFAULT '{}' CHECK (jsonb_typeof(provider_metadata) = 'object'),
  received_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (workspace_id, attempt_id),
  FOREIGN KEY (attempt_id, workspace_id) REFERENCES publishing_attempts (id, workspace_id)
    ON DELETE NO ACTION ON UPDATE RESTRICT
);

-- Receipts are never rewritten. Deletion is only the workspace purge cascade
-- (the parent workspace row is already gone when the cascade reaches here).
CREATE FUNCTION publishing_receipts_append_only() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
  IF TG_OP = 'DELETE' THEN
    IF EXISTS (SELECT 1 FROM workspaces w WHERE w.id = OLD.workspace_id) THEN
      RAISE EXCEPTION 'publishing receipts are append-only' USING ERRCODE = '23514';
    END IF;
    RETURN OLD;
  END IF;
  RAISE EXCEPTION 'publishing receipts are append-only' USING ERRCODE = '23514';
END $$;
CREATE TRIGGER trg_publishing_receipt_append_only BEFORE UPDATE OR DELETE ON publishing_receipts
  FOR EACH ROW EXECUTE FUNCTION publishing_receipts_append_only();

-- Draft-creation request idempotency, committed atomically with the draft.
CREATE TABLE publishing_requests (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE ON UPDATE RESTRICT,
  user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE ON UPDATE RESTRICT,
  route text NOT NULL CHECK (btrim(route) <> ''),
  request_key uuid NOT NULL,
  request_hash text NOT NULL CHECK (btrim(request_hash) <> ''),
  response jsonb NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (workspace_id, user_id, route, request_key)
);
CREATE TRIGGER trg_publishing_request_ws_immutable BEFORE UPDATE ON publishing_requests
  FOR EACH ROW EXECUTE FUNCTION reject_workspace_change();

-- ---------------------------------------------------------------------------
-- Draft fence: content changes rotate the revision and void approval; they
-- are refused while a dispatch is active or its outcome is unknown.
-- ---------------------------------------------------------------------------
CREATE FUNCTION drafts_revision_fence() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
  IF NEW.publication_attempt < OLD.publication_attempt THEN
    RAISE EXCEPTION 'publication_attempt cannot decrease' USING ERRCODE = '23514';
  END IF;
  IF NEW.post IS DISTINCT FROM OLD.post THEN
    IF EXISTS (SELECT 1 FROM publishing_attempts a
                WHERE a.workspace_id = OLD.workspace_id AND a.draft_id = OLD.id
                  AND (a.status = 'unknown'
                       OR (a.status = 'in_progress' AND a.dispatch_started_at IS NOT NULL))) THEN
      RAISE EXCEPTION 'draft content is locked while a dispatch is unsettled' USING ERRCODE = '55000';
    END IF;
    NEW.revision := OLD.revision + 1;
    NEW.approved_revision := NULL;
    NEW.approved_hash := NULL;
    NEW.approved_at := NULL;
    NEW.reviewer_user_id := NULL;
  ELSE
    NEW.revision := OLD.revision;
  END IF;
  RETURN NEW;
END $$;
CREATE TRIGGER trg_draft_revision_fence BEFORE UPDATE ON drafts
  FOR EACH ROW EXECUTE FUNCTION drafts_revision_fence();

-- ---------------------------------------------------------------------------
-- RLS + grants for the new tenant tables (001/006/017 style).
-- ---------------------------------------------------------------------------
ALTER TABLE publishing_attempts ENABLE ROW LEVEL SECURITY;
ALTER TABLE publishing_receipts ENABLE ROW LEVEL SECURITY;
ALTER TABLE publishing_requests ENABLE ROW LEVEL SECURITY;
ALTER TABLE publishing_attempts FORCE ROW LEVEL SECURITY;
ALTER TABLE publishing_receipts FORCE ROW LEVEL SECURITY;
ALTER TABLE publishing_requests FORCE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation ON publishing_attempts
  USING (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid)
  WITH CHECK (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid);
CREATE POLICY tenant_isolation ON publishing_receipts
  USING (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid)
  WITH CHECK (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid);
CREATE POLICY tenant_isolation ON publishing_requests
  USING (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid)
  WITH CHECK (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid);

-- 003's default privileges grant all four verbs; narrow them explicitly:
-- the journal is never deleted by the app, receipts are append-only.
GRANT SELECT, INSERT, UPDATE ON publishing_attempts TO sp_app;
REVOKE DELETE ON publishing_attempts FROM sp_app;
GRANT SELECT, INSERT ON publishing_receipts TO sp_app;
REVOKE UPDATE, DELETE ON publishing_receipts FROM sp_app;
GRANT SELECT, INSERT, UPDATE, DELETE ON publishing_requests TO sp_app;

-- ---------------------------------------------------------------------------
-- Indexes: keyset lists (no OFFSET), due/recovery discovery, FK coverage.
-- ---------------------------------------------------------------------------
CREATE INDEX drafts_ws_updated ON drafts (workspace_id, updated_at DESC, id DESC);
CREATE INDEX drafts_ws_state_updated ON drafts (workspace_id, state, updated_at DESC, id DESC);

CREATE INDEX scheduled_publications_ws_publish ON scheduled_publications (workspace_id, publish_at DESC, id DESC);
CREATE INDEX scheduled_publications_ws_conn_publish
  ON scheduled_publications (workspace_id, connection_id, publish_at DESC, id DESC);
CREATE INDEX scheduled_publications_next_due ON scheduled_publications (next_attempt_at, id)
  WHERE state = 'pending';
CREATE INDEX scheduled_publications_ws_credential ON scheduled_publications (workspace_id, credential_id)
  WHERE credential_id IS NOT NULL;

CREATE INDEX publications_ws_published ON publications (workspace_id, published_at DESC, id DESC);
CREATE INDEX publications_ws_conn_published
  ON publications (workspace_id, connection_id, published_at DESC, id DESC);

CREATE INDEX publishing_attempts_ws_schedule ON publishing_attempts (workspace_id, schedule_id, attempt_no DESC);
CREATE INDEX publishing_attempts_ws_connection ON publishing_attempts (workspace_id, connection_id);
CREATE INDEX publishing_attempts_ws_credential ON publishing_attempts (workspace_id, credential_id);
CREATE INDEX publishing_attempts_lease_recovery ON publishing_attempts (lease_until, id)
  WHERE status = 'in_progress';

-- ---------------------------------------------------------------------------
-- Discovery (replaces 006). Identifiers/times only, bounded, fixed
-- search_path, PUBLIC revoked. Callers enter withTenant(workspace_id) and
-- re-verify key/generation ownership under RLS.
-- ---------------------------------------------------------------------------
DROP FUNCTION publishing_due_schedules();

CREATE FUNCTION publishing_due_schedules(p_limit integer DEFAULT 100)
RETURNS TABLE(workspace_id uuid, schedule_id uuid, idempotency_key text, generation bigint, publish_at timestamptz)
LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$
  SELECT due.workspace_id, due.schedule_id, due.idempotency_key, due.generation, due.publish_at
  FROM (
    SELECT sp.workspace_id, sp.id AS schedule_id, sp.idempotency_key, sp.generation,
           sp.publish_at, sp.next_attempt_at AS due_at
      FROM scheduled_publications sp
     WHERE sp.state = 'pending' AND sp.execution_version = 1 AND sp.next_attempt_at <= now()
    UNION ALL
    SELECT a.workspace_id, a.schedule_id, a.idempotency_key, a.schedule_generation,
           sp.publish_at, a.lease_until AS due_at
      FROM publishing_attempts a
      JOIN scheduled_publications sp ON sp.id = a.schedule_id AND sp.workspace_id = a.workspace_id
     WHERE a.status = 'in_progress' AND a.lease_until < now()
  ) due
  ORDER BY due.due_at, due.schedule_id
  LIMIT LEAST(GREATEST(COALESCE(p_limit, 100), 1), 100)
$$;

-- Attempt journal first (every executable key), then legacy schedules/publications.
CREATE OR REPLACE FUNCTION publishing_schedule_workspace(p_idempotency_key text) RETURNS uuid
LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$
  SELECT COALESCE(
    (SELECT a.workspace_id FROM publishing_attempts a WHERE a.idempotency_key = p_idempotency_key LIMIT 1),
    (SELECT sp.workspace_id FROM scheduled_publications sp WHERE sp.idempotency_key = p_idempotency_key),
    (SELECT p.workspace_id FROM publications p WHERE p.idempotency_key = p_idempotency_key)
  )
$$;

REVOKE ALL ON FUNCTION publishing_due_schedules(integer) FROM PUBLIC;
REVOKE ALL ON FUNCTION publishing_schedule_workspace(text) FROM PUBLIC;
GRANT EXECUTE ON FUNCTION publishing_due_schedules(integer), publishing_schedule_workspace(text) TO sp_app;