csvlua

local dquote = ('"'):byte()

local function extract_simple(record, start)
  local index = record:find(',', start)
  local stop_index
  local next_offset
  if index ~= nil then
    stop_index = index - 1
    next_offset = index + 1
  end
  return start, stop_index, next_offset
end

local function extract_quoted(record, start)
  start = start + 1
  local offset = start
  while true do
    local index = record:find('"', offset)
    local next_index = index + 1
    local next_byte = record:byte(next_index)
    if next_byte ~= dquote then
      -- found the end index of the field, return it
      return start, index - 1, index + 2
    end 
    offset = index + 2 -- advance both dquotes
  end
end

local function split_csv(record)
  local rv = {}
  local offset = 1
  while offset ~= nil do
    local start_idx
    local stop_idx
    if record:byte(offset) == dquote then
      start_idx, stop_idx, offset = extract_quoted(record, offset)
    else
      start_idx, stop_idx, offset = extract_simple(record, offset)
    end
    table.insert(rv, record:sub(start_idx, stop_idx))
  end
  return rv
end

local function parse_message(message, target)
  if not message then
    return
  end
  for  k, v in message:gmatch('(%S+)=(%S+),?') do
    target[k:gsub('^{', '')] = v:gsub('}$', ''):gsub(',$', '')
  end
end

local function parse_ms(record, key)
  if record[key] ~= nil then
    record[key] = tonumber(record[key]:gsub('ms$',''), 10)  
  end
end

local function normalize_urls(record)
  local key = 'BO-API-URL'
  if record[key] == nil then
    return
  end
  record[key] = record[key]
    :gsub('[0-9a-fA-F]+-[0-9a-fA-F]+-[0-9a-fA-F]+-[0-9a-fA-F]+-[0-9a-fA-F]+', '*****')
    :gsub('[0-9A-Z]+-[0-9A-Z]+-[0-9A-Z]+', '*****')
    :gsub('%d+', '*****')
    :gsub('instruments/[^/]+', '*****')
    :gsub('funding/([^/]+)/[^/]+', 'funding/%1/*****')
    :gsub('orders/[^/]+', 'orders/*****')
end

local function allow_keys(record, allowed)
  local rv = {}
  for key, value in pairs(record) do
    for _, allowed_key in ipairs(allowed) do
      if key == allowed_key then
        rv[key] = value
        break
      end
    end
  end
  return rv
end

-- declare a module variable to hold header
local headers = {}
function cb_filter(tag, timestamp, record)
  if headers[tag] == nil then
    headers[tag] = split_csv(record._raw)
    -- return -1 to drop the header
    return -1, timestamp, record
  end
  local data = {}
  local fields = split_csv(record._raw)
  for index, header in ipairs(headers[tag]) do
    data[header] = fields[index]
  end
  parse_message(data.message, data)
  normalize_urls(data)
  parse_ms(data, 'PROCESSING-TIME')
  timestamp =  parse_ms(data, 'messageTime')
  data = allow_keys(data, {'BO-API-URL', 'BO-API-HTTP-METHOD', 'RESPONSE-STATUS-CODE', 'API-PARTNER', 'PROCESSING-TIME' })
  data.timestamp = timestamp
  return 1, timestamp, data
end