Using the Data Cloud Ingestion to insert records with JWT connected app

datacloud-ingestion-insert.rb

require 'jwt'
require "faraday"
require "uri"

def token
  private_key = './keys/server.key'
  priv_key = OpenSSL::PKey::RSA.new(File.read(private_key))
  JWT.encode payload, priv_key, 'RS256'
end

def payload
  {
    "iss": "YOUR-CONNECTED-APP-CLIENT-ID",
    "sub": "YOUR-USERNAME",
    "aud": "https://login.salesforce.com",
    "exp": Time.now.to_i
  }  
end

def request_access_token
  params = {
    grant_type: 'urn:ietf:params:oauth:grant-type:jwt-bearer', 
    assertion: token
  }
  response = Faraday.post('https://login.salesforce.com/services/oauth2/token', URI.encode_www_form(params))
  body = JSON.parse(response.body)

  @access_token = body['access_token']
  @instance_url = body['instance_url']
end

def request_cdp_token
  params = {
    grant_type: 'urn:salesforce:grant-type:external:cdp', 
    subject_token_type: 'urn:ietf:params:oauth:token-type:access_token',
    subject_token: @access_token

  }
  response = Faraday.post("#{@instance_url}/services/a360/token", URI.encode_www_form(params))
  body = JSON.parse(response.body)

  @cdp_access_token = body['access_token']
  @cdp_instance_url = body['instance_url']
end

def insert_records
  ingestion_api_url = "https://#{@cdp_instance_url}/api/v1/ingest/sources/Anypoint_Platform/badge_rating"
  headers = {
    Authorization: "Bearer #{@cdp_access_token}", 
    'Content-Type': 'application/json'
  }
  body = {
      "data": [
          {
              "id": 1,
              "badge_id": "workshop-battle-station",
              "first_name": "Jeff",
              "last_name": "Douglas",
              "email": "jdouglas@salesforce.com",
              "rating": 5,
              "createddate": "2023-10-13 18:47:47.515304"
          }
      ]
  }

  response = Faraday.post(ingestion_api_url, body.to_json, headers)
  body = JSON.parse(response.body) 
  puts body 
end

request_access_token
request_cdp_token
insert_records