
CREATE SCHEMA IF NOT EXISTS ops;
CREATE SCHEMA IF NOT EXISTS public_legacy;
CREATE SCHEMA IF NOT EXISTS backup_canonical_20260219;

CREATE TABLE IF NOT EXISTS ops.canonical_table_map (
    table_name text PRIMARY KEY,
    canonical_schema text NOT NULL,
    notes text
);

INSERT INTO ops.canonical_table_map (table_name, canonical_schema, notes)
VALUES
    ('analisis_conversacion', 'ai', 'AI domain source of truth'),
    ('auth_remember_tokens', 'auth', 'Auth domain source of truth'),
    ('categorias_combinables_v2', 'catalog', 'Catalog domain source of truth'),
    ('category_rules_v2', 'catalog', 'Catalog domain source of truth'),
    ('citas', 'commerce', 'Commerce domain source of truth'),
    ('clientes', 'crm', 'CRM domain source of truth'),
    ('codigos_activacion', 'auth', 'Auth domain source of truth'),
    ('comprobantes_transacciones', 'commerce', 'Commerce domain source of truth'),
    ('configuraciones', 'platform', 'Platform domain source of truth'),
    ('configuraciones_planes_pago', 'commerce', 'Commerce domain source of truth'),
    ('conocimientos_empleado_digital', 'ai', 'AI domain source of truth'),
    ('disponibilidad_usuarios', 'workforce', 'Workforce domain source of truth'),
    ('empleado_digital', 'ai', 'AI domain source of truth'),
    ('entity_category_links', 'catalog', 'Catalog domain source of truth'),
    ('error_log', 'ops', 'Ops domain source of truth'),
    ('estadisticas', 'platform', 'Platform domain source of truth'),
    ('estadisticas_costos_ia', 'ai', 'AI domain source of truth'),
    ('excepciones_vacaciones', 'workforce', 'Workforce domain source of truth'),
    ('inventario', 'catalog', 'Catalog domain source of truth'),
    ('inventario_variantes_categoria', 'catalog', 'Catalog domain source of truth'),
    ('inventario_variantes_resolucion_log', 'catalog', 'Catalog domain source of truth'),
    ('login', 'auth', 'Auth domain source of truth'),
    ('login_attempt_tracker', 'auth', 'Auth domain source of truth'),
    ('modelos_planes_pago', 'commerce', 'Commerce domain source of truth'),
    ('notas_internas', 'platform', 'Platform domain source of truth'),
    ('numeros_whatsapp_business', 'messaging', 'Messaging domain source of truth'),
    ('opi_prompts', 'platform', 'Platform domain source of truth'),
    ('pagos_planes', 'commerce', 'Commerce domain source of truth'),
    ('planes_pago', 'commerce', 'Commerce domain source of truth'),
    ('pos_idempotency_requests', 'ops', 'Ops domain source of truth'),
    ('preguntas_clasificacion_multi', 'crm', 'CRM domain source of truth'),
    ('reportes', 'crm', 'CRM domain source of truth'),
    ('transacciones', 'commerce', 'Commerce domain source of truth'),
    ('usuarios', 'workforce', 'Workforce domain source of truth'),
    ('ventas', 'commerce', 'Commerce domain source of truth'),
    ('webhook_ingress_dlq', 'ops', 'Ops domain source of truth'),
    ('webhook_ingress_events', 'ops', 'Ops domain source of truth'),
    ('whatsapp_templates', 'messaging', 'Messaging domain source of truth'),
    ('whatsapp_templates_for_staff', 'messaging', 'Messaging domain source of truth')
ON CONFLICT (table_name) DO UPDATE
SET canonical_schema = EXCLUDED.canonical_schema,
    notes = EXCLUDED.notes;

CREATE TABLE IF NOT EXISTS ops.canonical_reconciliation_runs (
    id bigserial PRIMARY KEY,
    migration_tag text NOT NULL,
    table_name text NOT NULL,
    canonical_schema text NOT NULL,
    ts_column text,
    pre_public_count bigint NOT NULL,
    pre_canonical_count bigint NOT NULL,
    post_public_count bigint NOT NULL,
    post_canonical_count bigint NOT NULL,
    inserted_from_public bigint NOT NULL,
    updated_from_public bigint NOT NULL,
    deleted_from_public bigint NOT NULL,
    recorded_conflicts bigint NOT NULL,
    executed_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS ops.canonical_reconciliation_conflicts (
    id bigserial PRIMARY KEY,
    migration_tag text NOT NULL,
    table_name text NOT NULL,
    canonical_schema text NOT NULL,
    pk jsonb,
    reason text NOT NULL,
    public_row jsonb,
    canonical_row jsonb,
    created_at timestamptz NOT NULL DEFAULT now()
);

CREATE OR REPLACE FUNCTION ops.reconcile_public_to_canonical(
    p_migration_tag text,
    p_canonical_schema text,
    p_table_name text
)
RETURNS void
LANGUAGE plpgsql
AS $$
DECLARE
    v_pk_cols text[];
    v_common_cols text[];
    v_non_pk_cols text[];
    v_ts_col text;

    v_pk_cols_sql text;
    v_common_cols_sql text;
    v_update_assign_sql text;
    v_pk_join_sql text;
    v_pk_json_p_sql text;
    v_pk_json_c_sql text;
    v_diff_expr_sql text;

    v_inserted bigint := 0;
    v_updated bigint := 0;
    v_deleted bigint := 0;
    v_conflicts bigint := 0;
    v_rowcount bigint := 0;

    v_pre_public bigint := 0;
    v_pre_canonical bigint := 0;
    v_post_public bigint := 0;
    v_post_canonical bigint := 0;

    v_reason_expr_sql text;
    v_canonical_identity_override text := '';
    v_public_identity_override text := '';
BEGIN
    PERFORM 1
    FROM pg_class c
    JOIN pg_namespace n ON n.oid = c.relnamespace
    WHERE n.nspname = 'public'
      AND c.relname = p_table_name
      AND c.relkind = 'r';

    IF NOT FOUND THEN
        RAISE EXCEPTION 'public.% no es tabla base o no existe', p_table_name;
    END IF;

    PERFORM 1
    FROM pg_class c
    JOIN pg_namespace n ON n.oid = c.relnamespace
    WHERE n.nspname = p_canonical_schema
      AND c.relname = p_table_name
      AND c.relkind = 'r';

    IF NOT FOUND THEN
        RAISE EXCEPTION '%.% no es tabla base o no existe', p_canonical_schema, p_table_name;
    END IF;

    SELECT array_agg(c.column_name ORDER BY c.ordinal_position)
      INTO v_common_cols
    FROM information_schema.columns c
    JOIN information_schema.columns p
      ON p.column_name = c.column_name
     AND p.table_schema = 'public'
     AND p.table_name = p_table_name
    WHERE c.table_schema = p_canonical_schema
      AND c.table_name = p_table_name;

    IF COALESCE(array_length(v_common_cols, 1), 0) = 0 THEN
        RAISE EXCEPTION 'No hay columnas en comun entre public.% y %.%', p_table_name, p_canonical_schema, p_table_name;
    END IF;

    SELECT array_agg(a.attname ORDER BY array_position(i.indkey, a.attnum))
      INTO v_pk_cols
    FROM pg_index i
    JOIN pg_class c ON c.oid = i.indrelid
    JOIN pg_namespace n ON n.oid = c.relnamespace
    JOIN pg_attribute a
      ON a.attrelid = c.oid
     AND a.attnum = ANY(i.indkey)
    WHERE i.indisprimary
      AND n.nspname = p_canonical_schema
      AND c.relname = p_table_name;

    SELECT col
      INTO v_ts_col
    FROM unnest(ARRAY['updated_at', 'updated_on', 'modified_at', 'last_updated_at']) AS col
    WHERE col = ANY(v_common_cols)
    LIMIT 1;

    SELECT string_agg(format('%I', col), ', ')
      INTO v_common_cols_sql
    FROM unnest(v_common_cols) AS col;

    IF EXISTS (
        SELECT 1
        FROM information_schema.columns
        WHERE table_schema = p_canonical_schema
          AND table_name = p_table_name
          AND is_identity = 'YES'
          AND identity_generation = 'ALWAYS'
          AND column_name = ANY(v_common_cols)
    ) THEN
        v_canonical_identity_override := 'OVERRIDING SYSTEM VALUE';
    END IF;

    IF EXISTS (
        SELECT 1
        FROM information_schema.columns
        WHERE table_schema = 'public'
          AND table_name = p_table_name
          AND is_identity = 'YES'
          AND identity_generation = 'ALWAYS'
          AND column_name = ANY(v_common_cols)
    ) THEN
        v_public_identity_override := 'OVERRIDING SYSTEM VALUE';
    END IF;

    EXECUTE format('SELECT COUNT(*) FROM public.%I', p_table_name)
       INTO v_pre_public;

    EXECUTE format('SELECT COUNT(*) FROM %I.%I', p_canonical_schema, p_table_name)
       INTO v_pre_canonical;

    IF COALESCE(array_length(v_pk_cols, 1), 0) = 0 THEN
        IF p_table_name <> 'configuraciones'
           OR NOT ('business_id' = ANY(v_common_cols) AND 'role_id' = ANY(v_common_cols)) THEN
            RAISE EXCEPTION 'Tabla % sin PK no soportada para reconciliacion automatica', p_table_name;
        END IF;

        EXECUTE format(
            'INSERT INTO ops.canonical_reconciliation_conflicts
                 (migration_tag, table_name, canonical_schema, pk, reason, public_row, canonical_row)
             SELECT $1, $2, $3,
                    jsonb_build_object(''business_id'', p.business_id, ''role_id'', p.role_id),
                    ''only_public'', to_jsonb(p), NULL
             FROM public.%I p
             WHERE NOT EXISTS (
                 SELECT 1
                 FROM %I.%I c
                 WHERE c.business_id = p.business_id
                   AND c.role_id = p.role_id
             )',
            p_table_name,
            p_canonical_schema,
            p_table_name
        )
        USING p_migration_tag, p_table_name, p_canonical_schema;
        GET DIAGNOSTICS v_rowcount = ROW_COUNT;
        v_conflicts := v_conflicts + v_rowcount;

        EXECUTE format(
            'INSERT INTO ops.canonical_reconciliation_conflicts
                 (migration_tag, table_name, canonical_schema, pk, reason, public_row, canonical_row)
             SELECT $1, $2, $3,
                    jsonb_build_object(''business_id'', c.business_id, ''role_id'', c.role_id),
                    ''only_canonical'', NULL, to_jsonb(c)
             FROM %I.%I c
             WHERE NOT EXISTS (
                 SELECT 1
                 FROM public.%I p
                 WHERE p.business_id = c.business_id
                   AND p.role_id = c.role_id
             )',
            p_canonical_schema,
            p_table_name,
            p_table_name
        )
        USING p_migration_tag, p_table_name, p_canonical_schema;
        GET DIAGNOSTICS v_rowcount = ROW_COUNT;
        v_conflicts := v_conflicts + v_rowcount;

        EXECUTE format(
            'INSERT INTO %I.%I (%s) %s
             SELECT %s
             FROM public.%I p
             WHERE NOT EXISTS (
                 SELECT 1
                 FROM %I.%I c
                 WHERE c.business_id = p.business_id
                   AND c.role_id = p.role_id
             )',
            p_canonical_schema,
            p_table_name,
            v_common_cols_sql,
            v_canonical_identity_override,
            v_common_cols_sql,
            p_table_name,
            p_canonical_schema,
            p_table_name
        );
        GET DIAGNOSTICS v_inserted = ROW_COUNT;

    ELSE
        SELECT string_agg(format('%I', col), ', ')
          INTO v_pk_cols_sql
        FROM unnest(v_pk_cols) AS col;

        SELECT array_agg(col)
          INTO v_non_pk_cols
        FROM unnest(v_common_cols) AS col
        WHERE NOT (col = ANY(v_pk_cols));

        SELECT string_agg(format('p.%1$I = c.%1$I', col), ' AND ')
          INTO v_pk_join_sql
        FROM unnest(v_pk_cols) AS col;

        SELECT string_agg(format('%L, p.%I', col, col), ', ')
          INTO v_pk_json_p_sql
        FROM unnest(v_pk_cols) AS col;

        SELECT string_agg(format('%L, c.%I', col, col), ', ')
          INTO v_pk_json_c_sql
        FROM unnest(v_pk_cols) AS col;

        SELECT 'md5(to_jsonb(ROW(' || string_agg(format('p.%I', col), ', ') || '))::text) <> md5(to_jsonb(ROW(' || string_agg(format('c.%I', col), ', ') || '))::text)'
          INTO v_diff_expr_sql
        FROM unnest(v_common_cols) AS col;

        EXECUTE format(
            'INSERT INTO ops.canonical_reconciliation_conflicts
                 (migration_tag, table_name, canonical_schema, pk, reason, public_row, canonical_row)
             SELECT $1, $2, $3,
                    jsonb_build_object(%s),
                    ''only_public'', to_jsonb(p), NULL
             FROM public.%I p
             WHERE NOT EXISTS (
                 SELECT 1
                 FROM %I.%I c
                 WHERE %s
             )',
            v_pk_json_p_sql,
            p_table_name,
            p_canonical_schema,
            p_table_name,
            v_pk_join_sql
        )
        USING p_migration_tag, p_table_name, p_canonical_schema;
        GET DIAGNOSTICS v_rowcount = ROW_COUNT;
        v_conflicts := v_conflicts + v_rowcount;

        EXECUTE format(
            'INSERT INTO ops.canonical_reconciliation_conflicts
                 (migration_tag, table_name, canonical_schema, pk, reason, public_row, canonical_row)
             SELECT $1, $2, $3,
                    jsonb_build_object(%s),
                    ''only_canonical'', NULL, to_jsonb(c)
             FROM %I.%I c
             WHERE NOT EXISTS (
                 SELECT 1
                 FROM public.%I p
                 WHERE %s
             )',
            v_pk_json_c_sql,
            p_canonical_schema,
            p_table_name,
            p_table_name,
            v_pk_join_sql
        )
        USING p_migration_tag, p_table_name, p_canonical_schema;
        GET DIAGNOSTICS v_rowcount = ROW_COUNT;
        v_conflicts := v_conflicts + v_rowcount;

        IF v_ts_col IS NOT NULL THEN
            v_reason_expr_sql := format(
                'CASE
                    WHEN p.%1$I IS NOT NULL AND c.%1$I IS NOT NULL AND p.%1$I > c.%1$I THEN %2$L
                    WHEN p.%1$I IS NOT NULL AND c.%1$I IS NOT NULL AND p.%1$I < c.%1$I THEN %3$L
                    ELSE %4$L
                 END',
                v_ts_col,
                'public_newer',
                'canonical_newer',
                'timestamp_tie_or_null'
            );
        ELSE
            v_reason_expr_sql := quote_literal('no_timestamp_conflict');
        END IF;

        EXECUTE format(
            'INSERT INTO ops.canonical_reconciliation_conflicts
                 (migration_tag, table_name, canonical_schema, pk, reason, public_row, canonical_row)
             SELECT $1, $2, $3,
                    jsonb_build_object(%s),
                    %s,
                    to_jsonb(p),
                    to_jsonb(c)
             FROM public.%I p
             JOIN %I.%I c
               ON %s
             WHERE %s',
            v_pk_json_p_sql,
            v_reason_expr_sql,
            p_table_name,
            p_canonical_schema,
            p_table_name,
            v_pk_join_sql,
            v_diff_expr_sql
        )
        USING p_migration_tag, p_table_name, p_canonical_schema;
        GET DIAGNOSTICS v_rowcount = ROW_COUNT;
        v_conflicts := v_conflicts + v_rowcount;

        EXECUTE format(
            'INSERT INTO %I.%I (%s) %s
             SELECT %s
             FROM public.%I p
             WHERE NOT EXISTS (
                 SELECT 1
                 FROM %I.%I c
                 WHERE %s
             )',
            p_canonical_schema,
            p_table_name,
            v_common_cols_sql,
            v_canonical_identity_override,
            v_common_cols_sql,
            p_table_name,
            p_canonical_schema,
            p_table_name,
            v_pk_join_sql
        );
        GET DIAGNOSTICS v_inserted = ROW_COUNT;

        IF v_ts_col IS NOT NULL AND COALESCE(array_length(v_non_pk_cols, 1), 0) > 0 THEN
            SELECT string_agg(format('%1$I = p.%1$I', col), ', ')
              INTO v_update_assign_sql
            FROM unnest(v_non_pk_cols) AS col;

            EXECUTE format(
                'UPDATE %I.%I c
                 SET %s
                 FROM public.%I p
                 WHERE %s
                   AND p.%I IS NOT NULL
                   AND c.%I IS NOT NULL
                   AND p.%I > c.%I
                   AND %s',
                p_canonical_schema,
                p_table_name,
                v_update_assign_sql,
                p_table_name,
                v_pk_join_sql,
                v_ts_col,
                v_ts_col,
                v_ts_col,
                v_ts_col,
                v_diff_expr_sql
            );
            GET DIAGNOSTICS v_updated = ROW_COUNT;
        END IF;

    END IF;

    EXECUTE format('SELECT COUNT(*) FROM public.%I', p_table_name)
       INTO v_post_public;

    EXECUTE format('SELECT COUNT(*) FROM %I.%I', p_canonical_schema, p_table_name)
       INTO v_post_canonical;

    INSERT INTO ops.canonical_reconciliation_runs (
        migration_tag,
        table_name,
        canonical_schema,
        ts_column,
        pre_public_count,
        pre_canonical_count,
        post_public_count,
        post_canonical_count,
        inserted_from_public,
        updated_from_public,
        deleted_from_public,
        recorded_conflicts
    ) VALUES (
        p_migration_tag,
        p_table_name,
        p_canonical_schema,
        v_ts_col,
        v_pre_public,
        v_pre_canonical,
        v_post_public,
        v_post_canonical,
        v_inserted,
        v_updated,
        v_deleted,
        v_conflicts
    );
END;
$$;

DO $$
DECLARE
    rec record;
BEGIN
    FOR rec IN
        SELECT m.table_name, m.canonical_schema
        FROM ops.canonical_table_map AS m
        ORDER BY m.table_name
    LOOP
        EXECUTE format('LOCK TABLE public.%I IN ACCESS EXCLUSIVE MODE', rec.table_name);
        EXECUTE format('LOCK TABLE %I.%I IN ACCESS EXCLUSIVE MODE', rec.canonical_schema, rec.table_name);
    END LOOP;
END;
$$;

DO $$
DECLARE
    rec record;
BEGIN
    FOR rec IN
        SELECT table_name, canonical_schema
        FROM ops.canonical_table_map
        ORDER BY table_name
    LOOP
        EXECUTE format('DROP TABLE IF EXISTS backup_canonical_20260219.%I__%I', rec.canonical_schema, rec.table_name);
        EXECUTE format('CREATE TABLE backup_canonical_20260219.%I__%I AS TABLE %I.%I', rec.canonical_schema, rec.table_name, rec.canonical_schema, rec.table_name);

        EXECUTE format('DROP TABLE IF EXISTS backup_canonical_20260219.public__%I', rec.table_name);
        EXECUTE format('CREATE TABLE backup_canonical_20260219.public__%I AS TABLE public.%I', rec.table_name, rec.table_name);
    END LOOP;
END;
$$;

DROP TRIGGER IF EXISTS trg_sync_mirror_platform_estadisticas ON public.estadisticas;
DROP TRIGGER IF EXISTS trg_sync_mirror_platform_estadisticas ON platform.estadisticas;
DROP TRIGGER IF EXISTS trg_sync_mirror_ai_analisis_conversacion ON public.analisis_conversacion;
DROP TRIGGER IF EXISTS trg_sync_mirror_ai_analisis_conversacion ON ai.analisis_conversacion;
DROP TRIGGER IF EXISTS trg_sync_mirror_ai_estadisticas_costos_ia ON public.estadisticas_costos_ia;
DROP TRIGGER IF EXISTS trg_sync_mirror_ai_estadisticas_costos_ia ON ai.estadisticas_costos_ia;
DROP TRIGGER IF EXISTS trg_sync_mirror_crm_reportes ON public.reportes;
DROP TRIGGER IF EXISTS trg_sync_mirror_crm_reportes ON crm.reportes;
DROP TRIGGER IF EXISTS trg_sync_mirror_webhook_ingress_events ON public.webhook_ingress_events;
DROP TRIGGER IF EXISTS trg_sync_mirror_webhook_ingress_events ON ops.webhook_ingress_events;
DROP TRIGGER IF EXISTS trg_sync_mirror_webhook_ingress_dlq ON public.webhook_ingress_dlq;
DROP TRIGGER IF EXISTS trg_sync_mirror_webhook_ingress_dlq ON ops.webhook_ingress_dlq;
DROP TRIGGER IF EXISTS trg_sync_mirror_inventario_variantes_resolucion_log ON public.inventario_variantes_resolucion_log;
DROP TRIGGER IF EXISTS trg_sync_mirror_inventario_variantes_resolucion_log ON catalog.inventario_variantes_resolucion_log;

DROP FUNCTION IF EXISTS platform.sync_estadisticas_mirror();
DROP FUNCTION IF EXISTS ai.sync_analisis_conversacion_mirror();
DROP FUNCTION IF EXISTS ai.sync_estadisticas_costos_ia_mirror();
DROP FUNCTION IF EXISTS crm.sync_reportes_mirror();
DROP FUNCTION IF EXISTS ops.sync_webhook_ingress_events_mirror();
DROP FUNCTION IF EXISTS ops.sync_webhook_ingress_dlq_mirror();
DROP FUNCTION IF EXISTS catalog.sync_inventario_variantes_resolucion_log_mirror();

DO $$
DECLARE
    rec record;
    v_tag constant text := '2026-02-19-canonicalization-v1';
BEGIN
    FOR rec IN
        SELECT table_name, canonical_schema
        FROM ops.canonical_table_map
        ORDER BY table_name
    LOOP
        PERFORM ops.reconcile_public_to_canonical(v_tag, rec.canonical_schema, rec.table_name);
    END LOOP;
END;
$$;

DO $$
DECLARE
    rec record;
    v_relkind char;
    v_owner text := current_user;
BEGIN
    FOR rec IN
        SELECT table_name, canonical_schema
        FROM ops.canonical_table_map
        ORDER BY table_name
    LOOP
        SELECT c.relkind
          INTO v_relkind
        FROM pg_class c
        JOIN pg_namespace n ON n.oid = c.relnamespace
        WHERE n.nspname = 'public'
          AND c.relname = rec.table_name;

        IF v_relkind = 'r' THEN
            IF EXISTS (
                SELECT 1
                FROM pg_class c
                JOIN pg_namespace n ON n.oid = c.relnamespace
                WHERE n.nspname = 'public_legacy'
                  AND c.relname = rec.table_name
                  AND c.relkind = 'r'
            ) THEN
                RAISE EXCEPTION 'public_legacy.% ya existe; abortando para evitar sobreescritura de respaldo', rec.table_name;
            END IF;

            EXECUTE format('ALTER TABLE public.%I SET SCHEMA public_legacy', rec.table_name);
        END IF;

        EXECUTE format(
            'CREATE OR REPLACE VIEW public.%I AS SELECT * FROM %I.%I',
            rec.table_name,
            rec.canonical_schema,
            rec.table_name
        );

        EXECUTE format('ALTER VIEW public.%I OWNER TO %I', rec.table_name, v_owner);
    END LOOP;
END;
$$;

DO $$
BEGIN
    EXECUTE format(
        'ALTER ROLE %I IN DATABASE %I SET search_path = ai, auth, catalog, commerce, crm, messaging, ops, platform, workforce, public',
        current_user,
        current_database()
    );
END;
$$;

CREATE OR REPLACE FUNCTION ops.audit_domain_public_divergence()
RETURNS TABLE (
    table_name text,
    canonical_schema text,
    public_object_kind text,
    legacy_table_exists boolean,
    canonical_row_count bigint,
    public_row_count bigint,
    divergent boolean
)
LANGUAGE plpgsql
AS $$
DECLARE
    rec record;
    v_public_kind char;
    v_legacy_exists boolean;
    v_canonical_count bigint;
    v_public_count bigint;
BEGIN
    FOR rec IN
        SELECT table_name, canonical_schema
        FROM ops.canonical_table_map
        ORDER BY table_name
    LOOP
        SELECT c.relkind
          INTO v_public_kind
        FROM pg_class c
        JOIN pg_namespace n ON n.oid = c.relnamespace
        WHERE n.nspname = 'public'
          AND c.relname = rec.table_name;

        SELECT EXISTS (
            SELECT 1
            FROM pg_class c
            JOIN pg_namespace n ON n.oid = c.relnamespace
            WHERE n.nspname = 'public_legacy'
              AND c.relname = rec.table_name
              AND c.relkind = 'r'
        )
          INTO v_legacy_exists;

        EXECUTE format('SELECT COUNT(*) FROM %I.%I', rec.canonical_schema, rec.table_name)
           INTO v_canonical_count;

        EXECUTE format('SELECT COUNT(*) FROM public.%I', rec.table_name)
           INTO v_public_count;

        table_name := rec.table_name;
        canonical_schema := rec.canonical_schema;
        public_object_kind := CASE v_public_kind
            WHEN 'v' THEN 'view'
            WHEN 'r' THEN 'table'
            WHEN 'm' THEN 'materialized_view'
            WHEN NULL THEN 'missing'
            ELSE v_public_kind::text
        END;
        legacy_table_exists := v_legacy_exists;
        canonical_row_count := v_canonical_count;
        public_row_count := v_public_count;
        divergent := (v_public_kind IS DISTINCT FROM 'v') OR (v_canonical_count <> v_public_count);

        RETURN NEXT;
    END LOOP;
END;
$$;

CREATE OR REPLACE VIEW ops.canonicalization_status AS
SELECT *
FROM ops.audit_domain_public_divergence();
