cartodb-4.42/app/services/carto/federated_tables_service.rb

module Carto
  class FederatedTablesService
    def initialize(user:)
      @user = user
      @superuser_db_connection = @user.in_database(as: :superuser)
      @user_db_connection = @user.in_database()
    end

    # Federated Servers

    def list_servers(order:, direction:, per_page:, page:, offset:)
      @user_db_connection[
        select_federated_servers_query(order: order, direction: direction, per_page: per_page, offset: offset)
      ].all
    end

    def count_servers
      @user_db_connection[count_federated_servers_query].first[:count]
    end

    def register_server(federated_server_name:, **attributes)
      @superuser_db_connection[
        register_federated_server_query(federated_server_name: federated_server_name, **attributes)
      ].first
      grant_access_to_federated_server(
        federated_server_name: federated_server_name,
        db_role: @user.database_username
      )
      get_server(federated_server_name: federated_server_name)
    end

    def grant_access_to_federated_server(federated_server_name:, db_role:)
      @superuser_db_connection[
        grant_access_to_federated_server_query(
          federated_server_name: federated_server_name,
          db_role: db_role
        )
      ].all
    end

    def get_server(federated_server_name:)
      @user_db_connection[get_federated_server_query(federated_server_name: federated_server_name)].first
    end

    def update_server(federated_server_name:, **attributes)
      unregister_server(federated_server_name: federated_server_name)
      register_server(federated_server_name: federated_server_name, **attributes)
      get_server(federated_server_name: federated_server_name)
    end

    def unregister_server(federated_server_name:)
      @superuser_db_connection[
        unregister_federated_server_query(federated_server_name: federated_server_name)
      ].first
    end

    def revoke_access_to_federated_server(federated_server_name:, db_role:)
      @superuser_db_connection[
        revoke_access_to_federated_server_query(
          federated_server_name: federated_server_name,
          db_role: db_role
        )
      ].all
    end

    # Remote Schemas

    def list_remote_schemas(federated_server_name:, order:, direction:, per_page:, page:, offset:)
      @user_db_connection[
        select_remote_schemas_query(
          federated_server_name: federated_server_name,
          order: order,
          direction: direction,
          per_page: per_page,
          offset: offset
        )
      ].all
    end

    def count_remote_schemas(federated_server_name:)
      @user_db_connection[
        count_remote_schemas_query(federated_server_name: federated_server_name)
      ].first[:count]
    end

    # Remote Tables

    def list_remote_tables(federated_server_name:, remote_schema_name:, order:, direction:, per_page:, page:, offset:)
      remote_tables = @user_db_connection[
        select_remote_tables_query(
          federated_server_name: federated_server_name,
          remote_schema_name: remote_schema_name,
          order: order,
          direction: direction,
          per_page: per_page,
          offset: offset
        )
      ].all
      remote_tables.map do |remote_table|
        remote_table.merge(columns: JSON.parse(remote_table[:columns], symbolize_names: true))
      end
    end

    def count_remote_tables(federated_server_name:, remote_schema_name:)
      @user_db_connection[
        count_remote_tables_query(federated_server_name: federated_server_name, remote_schema_name: remote_schema_name)
      ].first[:count]
    end

    def register_table(federated_server_name:, remote_schema_name:, remote_table_name:, **attributes)
      @user_db_connection[
        register_remote_table_query(
          federated_server_name: federated_server_name,
          remote_schema_name: remote_schema_name,
          remote_table_name: remote_table_name,
          **attributes
        )
      ].first
      get_remote_table(
        federated_server_name: federated_server_name,
        remote_schema_name: remote_schema_name,
        remote_table_name: remote_table_name
      )
    end

    def get_remote_table(federated_server_name:, remote_schema_name:, remote_table_name:)
      remote_table = @user_db_connection[
        get_remote_table_query(
          federated_server_name: federated_server_name,
          remote_schema_name: remote_schema_name,
          remote_table_name: remote_table_name
        )
      ].first
      return unless remote_table
      remote_table.merge(columns: JSON.parse(remote_table[:columns], symbolize_names: true))
    end

    def update_table(federated_server_name:, remote_schema_name:, remote_table_name:, **attributes)
      unregister_table(
        federated_server_name: federated_server_name,
        remote_schema_name: remote_schema_name,
        remote_table_name: remote_table_name
      )
      register_table(
        federated_server_name: federated_server_name,
        remote_schema_name: remote_schema_name,
        remote_table_name: remote_table_name,
        **attributes
      )
      get_remote_table(
        federated_server_name: federated_server_name,
        remote_schema_name: remote_schema_name,
        remote_table_name: remote_table_name
      )
    end

    def unregister_table(federated_server_name:, remote_schema_name:, remote_table_name:)
      @user_db_connection[
        unregister_remote_table_query(
          federated_server_name: federated_server_name,
          remote_schema_name: remote_schema_name,
          remote_table_name: remote_table_name
        )
      ].first
    end

    private

    def select_federated_servers_query(order:, direction:, per_page:, offset:)
      %{
        SELECT * FROM (#{select_servers_query}) AS federated_servers
        ORDER BY #{order} #{direction}
        LIMIT #{per_page}
        OFFSET #{offset}
      }.squish
    end

    def count_federated_servers_query
      "SELECT COUNT(*) FROM (#{select_servers_query}) AS federated_servers"
    end

    def select_servers_query
      %{
        SELECT
          name as federated_server_name,
          readmode as mode,
          dbname,
          host,
          port,
          username,
          '*****' as password
        FROM cartodb.CDB_Federated_Server_List_Servers()
      }.squish
    end

    def register_federated_server_query(
      federated_server_name:,
      host: nil,
      port: nil,
      dbname: nil,
      mode:,
      username: nil,
      password: nil
    )
      config_server = {}
      config_server["dbname"] = dbname if dbname.present?
      config_server["host"] = host if host.present?
      config_server["port"] = port if port.present?
      config_server["extensions"] = "postgis"
      config_server["updatable"] = mode != 'read-write' ? "false" : "write"
      config_server["use_remote_estimate"] = "true"
      config_server["fetch_size"] = "1000"
      config_credentials = {}
      config_credentials["username"] = username if username.present?
      config_credentials["password"] = password if password.present?

      config = { server: config_server, credentials: config_credentials }

      server_str = @user_db_connection.literal(federated_server_name)
      config_str = @user_db_connection.literal(config.to_json)

      %{
        SELECT cartodb.CDB_Federated_Server_Register_PG(server := #{server_str}::text, config := #{config_str}::jsonb)
      }.squish
    end

    def grant_access_to_federated_server_query(federated_server_name:, db_role:)
      server_str = @user_db_connection.literal(federated_server_name)
      db_role_str = @user_db_connection.literal(db_role)
      %{
        SELECT cartodb.CDB_Federated_Server_Grant_Access(server := #{server_str}, db_role := #{db_role_str}::name)
      }.squish
    end

    def revoke_access_to_federated_server_query(federated_server_name:, db_role:)
      server_str = @user_db_connection.literal(federated_server_name)
      db_role_str = @user_db_connection.literal(db_role)
      %{
        SELECT cartodb.CDB_Federated_Server_Revoke_Access(server := #{server_str}, db_role := #{db_role_str}::name)
      }.squish
    end

    def get_federated_server_query(federated_server_name:)
      server_str = @user_db_connection.literal(federated_server_name)
      %{
        SELECT
          name as federated_server_name,
          readmode as mode,
          dbname,
          host,
          port,
          username,
          '*****' as password
        FROM cartodb.CDB_Federated_Server_List_Servers(server := #{server_str}::text)
      }.squish
    end

    def unregister_federated_server_query(federated_server_name:)
      server_str = @user_db_connection.literal(federated_server_name)
      %{
        SELECT
          cartodb.CDB_Federated_Server_Unregister(server := #{server_str}::text)
      }.squish
    end

    # Remote Schemas

    def select_remote_schemas_query(federated_server_name:, order:, direction:, per_page:, offset:)
      %{
        SELECT * FROM (#{select_schemas_query(federated_server_name: federated_server_name)}) AS remote_schemas
        ORDER BY #{order} #{direction}
        LIMIT #{per_page}
        OFFSET #{offset}
      }.squish
    end

    def count_remote_schemas_query(federated_server_name:)
      "SELECT COUNT(*) FROM (#{select_schemas_query(federated_server_name: federated_server_name)}) AS remote_schemas"
    end

    def select_schemas_query(federated_server_name:)
      server_str = @user_db_connection.literal(federated_server_name)
      %{
        SELECT
          remote_schema as remote_schema_name
        FROM
          cartodb.CDB_Federated_Server_List_Remote_Schemas(server => #{server_str})
      }.squish
    end

    # Remote Tables

    def select_remote_tables_query(federated_server_name:, remote_schema_name:, order:, direction:, per_page:, offset:)
      %{
        SELECT * FROM (#{select_tables_query(federated_server_name: federated_server_name, remote_schema_name: remote_schema_name)}) AS remote_tables
        ORDER BY #{order} #{direction}
        LIMIT #{per_page}
        OFFSET #{offset}
      }.squish
    end

    def count_remote_tables_query(federated_server_name:, remote_schema_name:)
      "SELECT COUNT(*) FROM (#{select_tables_query(federated_server_name: federated_server_name, remote_schema_name: remote_schema_name)}) AS remote_tables"
    end

    def select_tables_query(federated_server_name:, remote_schema_name:)
      server_str = @user_db_connection.literal(federated_server_name)
      schema_str = @user_db_connection.literal(remote_schema_name)
      %{
        SELECT
          registered,
          local_qualified_name as qualified_name,
          remote_table as remote_table_name,
          '#{remote_schema_name}' as remote_schema_name,
          id_column_name,
          geom_column_name,
          webmercator_column_name,
          columns
        FROM
          cartodb.CDB_Federated_Server_List_Remote_Tables(server => #{server_str}, remote_schema => #{schema_str})
      }.squish
    end

    def literal_attribute_or_null(attribute)
      return "NULL" if attribute.blank?
      @user_db_connection.literal(attribute)
    end

    def register_remote_table_query(
      federated_server_name:,
      remote_schema_name:,
      remote_table_name:,
      local_table_name_override:,
      id_column_name: nil,
      geom_column_name: nil,
      webmercator_column_name: nil
    )
      %{
        SELECT cartodb.CDB_Federated_Table_Register(
          server => #{literal_attribute_or_null(federated_server_name)},
          remote_schema => #{literal_attribute_or_null(remote_schema_name)},
          remote_table => #{literal_attribute_or_null(remote_table_name)},
          id_column => #{literal_attribute_or_null(id_column_name)},
          geom_column => #{literal_attribute_or_null(geom_column_name)},
          webmercator_column => #{literal_attribute_or_null(webmercator_column_name)},
          local_name => #{literal_attribute_or_null(local_table_name_override)}
        )
      }.squish
    end

    def get_remote_table_query(federated_server_name:, remote_schema_name:, remote_table_name:)
      server_str = @user_db_connection.literal(federated_server_name)
      schema_str = @user_db_connection.literal(remote_schema_name)
      table_str = @user_db_connection.literal(remote_table_name)
      %{
        SELECT
          registered,
          local_qualified_name as qualified_name,
          remote_table as remote_table_name,
          #{schema_str} as remote_schema_name,
          id_column_name,
          geom_column_name,
          webmercator_column_name,
          columns
        FROM
          cartodb.CDB_Federated_Server_List_Remote_Tables(
            server => #{server_str},
            remote_schema => #{schema_str}
          )
        WHERE
          remote_table = #{table_str}
      }.squish
    end

    def unregister_remote_table_query(federated_server_name:, remote_schema_name:, remote_table_name:)
      server_str = @user_db_connection.literal(federated_server_name)
      schema_str = @user_db_connection.literal(remote_schema_name)
      table_str = @user_db_connection.literal(remote_table_name)
      %{
        SELECT
          cartodb.CDB_Federated_Table_Unregister(
            server => #{server_str},
            remote_schema => #{schema_str},
            remote_table => #{table_str}
          )
      }.squish
    end
  end
end