-- pg_trickle 0.83.0 -> 0.84.0 upgrade migration
--
-- v0.84.0 — bootstrap catalog parity repair and manifest-tooling baseline

SELECT set_config('pg_trickle.enabled', 'off', true);

LOCK TABLE pgtrickle.pgt_stream_tables
    IN SHARE ROW EXCLUSIVE MODE;

ALTER TABLE pgtrickle.pgt_stream_tables
    ADD COLUMN IF NOT EXISTS tentative_frontier JSONB,
    ADD COLUMN IF NOT EXISTS in_shadow_build BOOLEAN NOT NULL DEFAULT FALSE,
    ADD COLUMN IF NOT EXISTS shadow_table_name TEXT;

CREATE TABLE IF NOT EXISTS pgtrickle.pgt_snapshots (
    snapshot_id      BIGSERIAL PRIMARY KEY,
    pgt_id           BIGINT NOT NULL
                     REFERENCES pgtrickle.pgt_stream_tables(pgt_id) ON DELETE CASCADE,
    snapshot_schema  TEXT NOT NULL,
    snapshot_table   TEXT NOT NULL,
    snapshot_version TEXT NOT NULL,
    frontier         JSONB,
    created_at       TIMESTAMPTZ NOT NULL DEFAULT now(),
    CONSTRAINT uq_snapshot_table UNIQUE (snapshot_schema, snapshot_table)
);

CREATE INDEX IF NOT EXISTS idx_pgt_snapshots_pgt_id
    ON pgtrickle.pgt_snapshots (pgt_id);

ALTER TABLE pgtrickle.pgt_snapshots
    ADD COLUMN IF NOT EXISTS snapshot_relid OID,
    ADD COLUMN IF NOT EXISTS snapshot_provenance_token TEXT,
    ADD COLUMN IF NOT EXISTS created_by_role_oid OID;

CREATE UNIQUE INDEX IF NOT EXISTS idx_pgt_snapshots_snapshot_relid
    ON pgtrickle.pgt_snapshots (snapshot_relid);

DO $$
BEGIN
    IF EXISTS (
        SELECT 1
        FROM pgtrickle.pgt_stream_tables
        WHERE max_differential_joins < 0
           OR max_delta_fraction < 0.0
           OR max_delta_fraction > 1.0
    ) THEN
        RAISE EXCEPTION
            'pg_trickle 0.84.0 migration found invalid persisted numeric configuration; repair rows before upgrading';
    END IF;

    IF NOT EXISTS (
        SELECT 1
        FROM pg_catalog.pg_constraint
        WHERE conrelid = 'pgtrickle.pgt_stream_tables'::regclass
          AND conname = 'pgt_stream_tables_max_differential_joins_check'
    ) THEN
        ALTER TABLE pgtrickle.pgt_stream_tables
            ADD CONSTRAINT pgt_stream_tables_max_differential_joins_check
            CHECK (max_differential_joins IS NULL OR max_differential_joins >= 0);
    END IF;

    IF NOT EXISTS (
        SELECT 1
        FROM pg_catalog.pg_constraint
        WHERE conrelid = 'pgtrickle.pgt_stream_tables'::regclass
          AND conname = 'pgt_stream_tables_max_delta_fraction_check'
    ) THEN
        ALTER TABLE pgtrickle.pgt_stream_tables
            ADD CONSTRAINT pgt_stream_tables_max_delta_fraction_check
            CHECK (max_delta_fraction IS NULL OR
                   (max_delta_fraction >= 0.0 AND max_delta_fraction <= 1.0));
    END IF;
END
$$;

CREATE TABLE IF NOT EXISTS pgtrickle.pgt_subscriptions (
    stream_table TEXT NOT NULL,
    channel      TEXT NOT NULL,
    created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (stream_table, channel)
);

CREATE INDEX IF NOT EXISTS idx_hist_pgt_start
    ON pgtrickle.pgt_refresh_history (pgt_id, start_time);

DROP INDEX IF EXISTS pgtrickle.pgt_refresh_history_start_time_idx;

DO $$
BEGIN
    IF to_regclass('pgtrickle.pgt_snapshots') IS NOT NULL
       AND NOT EXISTS (
           SELECT 1
           FROM pg_catalog.pg_extension ext
           JOIN LATERAL unnest(ext.extconfig) AS cfg(relid) ON TRUE
           WHERE ext.extname = 'pg_trickle'
             AND cfg.relid = 'pgtrickle.pgt_snapshots'::regclass
       ) THEN
        PERFORM pg_catalog.pg_extension_config_dump('pgtrickle.pgt_snapshots', '');
    END IF;

    IF to_regclass('pgtrickle.pgt_subscriptions') IS NOT NULL
       AND NOT EXISTS (
           SELECT 1
           FROM pg_catalog.pg_extension ext
           JOIN LATERAL unnest(ext.extconfig) AS cfg(relid) ON TRUE
           WHERE ext.extname = 'pg_trickle'
             AND cfg.relid = 'pgtrickle.pgt_subscriptions'::regclass
       ) THEN
        PERFORM pg_catalog.pg_extension_config_dump('pgtrickle.pgt_subscriptions', '');
    END IF;
END
$$;


-- Generated by scripts/check_sql_api_policy.py emit-acl-sql
REVOKE EXECUTE ON ALL FUNCTIONS IN SCHEMA pgtrickle FROM PUBLIC;

-- Explicit overload policy follows.

-- admin_global
REVOKE EXECUTE ON FUNCTION pgtrickle.advance_watermark(text, timestamp with time zone) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.clear_caches() FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.convert_buffers_to_unlogged() FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.create_refresh_group(text, text[], text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.create_watermark_group(text, text[], double precision) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.drain(integer) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.drop_refresh_group(text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.drop_watermark_group(text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.gate_source(text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.migrate() FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.pause_all() FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.pause_scheduler(text[]) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.rebuild_cdc_triggers() FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.restore_stream_tables() FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.resume_all() FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.resume_scheduler(text[]) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.setup_self_monitoring() FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.teardown_self_monitoring() FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.ungate_source(text) FROM PUBLIC;

-- arbitrary_sql
REVOKE EXECUTE ON FUNCTION pgtrickle.write_and_refresh(text, text) FROM PUBLIC;

-- internal
REVOKE EXECUTE ON FUNCTION pgtrickle._signal_launcher_rescan() FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.handle_vp_promoted(text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.pgt_ivm_apply_delta(bigint, integer, boolean, boolean) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.pgt_ivm_apply_delta_enr(bigint, integer, boolean, boolean) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.pgt_ivm_handle_truncate(bigint) FROM PUBLIC;

-- owner_lifecycle
REVOKE EXECUTE ON FUNCTION pgtrickle.alter_stream_table(text, text, text, text, text, text, text, text, boolean, boolean, text, text, bigint, integer, text, integer, double precision, text, double precision) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.attach_embedding_outbox(text, text, integer, integer) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.attach_outbox(text, integer, integer) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.bulk_alter_stream_tables(text[], json) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.bulk_create(jsonb) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.bulk_drop_stream_tables(text[]) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.canary_begin(text, text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.canary_diff(text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.canary_promote(text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.create_or_replace_stream_table(text, text, text, text, boolean, text, text, text, boolean, boolean, text, integer, double precision, text, boolean, text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.create_stream_table(text, text, text, text, boolean, text, text, text, boolean, boolean, text, integer, double precision, text, boolean, text, integer) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.create_stream_table_batch(text, text, text, boolean, text, integer, double precision) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.create_stream_table_cost_optimized(text, text, text, boolean, text, integer, double precision) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.create_stream_table_fast_append_only(text, text, text, text, text, integer, double precision) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.create_stream_table_if_not_exists(text, text, text, text, boolean, text, text, text, boolean, boolean, text, integer, double precision, text, boolean, text, integer) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.create_stream_table_realtime(text, text, text, boolean, text, integer, double precision) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.detach_outbox(text, boolean) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.drop_snapshot(text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.drop_stream_table(text, boolean) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.drop_stream_table_publication(text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.embedding_stream_table(text, text, text, text, text, text, boolean) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.exec_stream_ddl(text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.pause_stream_table(text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.refresh_efficiency() FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.refresh_groups() FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.refresh_if_stale(text, interval) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.refresh_stream_table(text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.refresh_timeline(integer) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.repair_stream_table(text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.reset_fuse(text, text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.restore_from_snapshot(text, text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.resume_stream_table(text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.set_stream_table_refresh_policy(text, text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.set_stream_table_sla(text, interval) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.set_stream_table_storage_policy(text, boolean, text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.snapshot_stream_table(text, text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.stream_table_to_publication(text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.subscribe(text, text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.subscribe_distance(text, text, text, text, text, double precision) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.unsubscribe(text, text) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle.unsubscribe_distance(text, text) FROM PUBLIC;

-- public_read
GRANT EXECUTE ON FUNCTION pgtrickle.bootstrap_gate_status() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.cache_stats() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.cdc_pause_status() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.change_buffer_sizes() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.check_cdc_health() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.cluster_worker_summary() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.commit_latency_stats() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.dedup_stats() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.dependency_tree() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.diagnose_errors(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.diamond_groups() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.explain_dag(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.explain_delta(text, text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.explain_diff_sql(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.explain_query_rewrite(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.explain_refresh_mode(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.explain_st(text, boolean) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.explain_stream_table(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.export_definition(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.fuse_status() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.get_refresh_history(text, integer) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.get_staleness(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.health_check() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.health_summary() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.history_prune_status() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.is_drained() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.list_auxiliary_columns(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.list_distance_subscriptions(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.list_snapshots(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.list_sources(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.list_subscriptions() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.metrics_summary() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.parallel_job_status(integer) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.parse_duration_seconds(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.pg_trickle_hash(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.pg_trickle_hash_multi(text[]) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.pgt_scc_status() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.pgt_status() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.pgtrickle_refresh_stats() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.preflight() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.preview_stream_table(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.recommend_refresh_mode(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.recommend_schedule(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.reliability_counters() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.schedule_recommendations() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.scheduler_overhead() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.self_monitoring_status() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.shared_buffer_stats() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.sla_summary() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.slot_health() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.source_gates() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.source_stable_name(oid) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.st_auto_threshold(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.st_refresh_stats() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.stream_table_definition(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.stream_table_lineage(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.stream_table_spec(oid) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.stream_table_spec(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.trigger_inventory() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.tune_recommendations() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.validate_query(text) TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.vector_status() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.version() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.version_check() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.view_evolution_status() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.wal_source_status() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.watermark_groups() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.watermark_status() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.watermarks() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.worker_allocation_status() TO PUBLIC;
GRANT EXECUTE ON FUNCTION pgtrickle.worker_pool_status() TO PUBLIC;

-- trigger_entry
REVOKE EXECUTE ON FUNCTION pgtrickle._on_ddl_end() FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION pgtrickle._on_sql_drop() FROM PUBLIC;


INSERT INTO pgtrickle.pgt_schema_version (version, description)
VALUES (
    '0.84.0',
    'Bootstrap catalog parity repair and manifest tooling baseline'
)
ON CONFLICT (version) DO NOTHING;