[sources.oathkeeper_file]
  type = "file"
  include = ["/var/log/oathkeeper/access.log"]
  read_from = "beginning"

[transforms.oathkeeper_parse]
  type = "remap"
  inputs = ["oathkeeper_file"]
  source = '''
    raw = to_string(.message) ?? ""
    parsed = parse_json(raw) ?? {}
    request_method = to_string(get(parsed, ["request", "method"]) ?? "") ?? ""
    request_path = to_string(get(parsed, ["request", "path"]) ?? "") ?? ""
    request_url = to_string(get(parsed, ["request", "url"]) ?? "") ?? ""
    request_host = to_string(get(parsed, ["request", "host"]) ?? "") ?? ""
    request_scheme = to_string(get(parsed, ["request", "scheme"]) ?? "") ?? ""
    request_query = to_string(get(parsed, ["request", "query"]) ?? "") ?? ""
    response_status = get(parsed, ["response", "status"]) ?? 0
    identity_id = to_string(get(parsed, ["identity", "id"]) ?? "") ?? ""
    headers = get(parsed, ["headers"]) ?? {}
    user_agent = to_string(get(headers, ["User-Agent"]) ?? "") ?? ""
    referer = to_string(get(headers, ["Referer"]) ?? "") ?? ""
    rule_id = to_string(get(parsed, ["rule", "id"]) ?? "") ?? ""
    upstream_url = to_string(get(parsed, ["upstream", "url"]) ?? "") ?? ""
    client_id = to_string(get(parsed, ["client", "id"]) ?? "") ?? ""
    parent_session_id = to_string(get(parsed, ["extra", "parent_session_id"]) ?? "") ?? ""
    parsed_url = parse_url(request_url) ?? {}
    query_params = get(parsed_url, ["query"]) ?? {}
    event_path = to_string(parsed.path) ?? to_string(parsed.http_path) ?? ""
    if event_path == "" { event_path = request_path }
    if event_path == "" { event_path = request_url }
    event_client_id = to_string(parsed.client_id) ?? ""
    if event_client_id == "" { event_client_id = client_id }
    if event_client_id == "" { event_client_id = to_string(get(query_params, ["client_id"]) ?? "") ?? "" }
    if event_client_id == "" { event_client_id = to_string(get(query_params, ["clientId"]) ?? "") ?? "" }

    . = {
      "request_id": to_string(parsed.request_id) ?? to_string(parsed.req_id) ?? "",
      "method": to_string(parsed.method) ?? to_string(parsed.http_method) ?? request_method,
      "path": event_path,
      "status": to_int(parsed.status) ?? to_int(parsed.status_code) ?? to_int(response_status) ?? 0,
      "latency_ms": to_int(parsed.latency_ms) ?? to_int(parsed.duration_ms) ?? to_int(parsed.took) ?? 0,
      "client_id": event_client_id,
      "rp": to_string(parsed.rp) ?? "",
      "action": to_string(parsed.action) ?? "",
      "target": to_string(parsed.target) ?? "",
      "rule_id": to_string(parsed.rule_id) ?? rule_id,
      "host": to_string(parsed.host) ?? request_host,
      "scheme": to_string(parsed.scheme) ?? request_scheme,
      "query": to_string(parsed.query) ?? request_query,
      "upstream_url": to_string(parsed.upstream_url) ?? upstream_url,
      "subject": to_string(parsed.subject) ?? identity_id,
      "parent_session_id": to_string(parsed.parent_session_id) ?? parent_session_id,
      "client_ip": to_string(parsed.client_ip) ?? to_string(parsed.remote_ip) ?? to_string(parsed.ip) ?? "",
      "user_agent": to_string(parsed.user_agent) ?? user_agent,
      "referer": referer,
      "decision": to_string(parsed.decision) ?? to_string(parsed.result) ?? "",
      "bytes_in": to_int(parsed.bytes_in) ?? to_int(parsed.request_bytes) ?? 0,
      "bytes_out": to_int(parsed.bytes_out) ?? to_int(parsed.response_bytes) ?? 0,
      "trace_id": to_string(parsed.trace_id) ?? "",
      "span_id": to_string(parsed.span_id) ?? "",
      "raw": raw
    }
  '''

[sinks.clickhouse]
  type = "clickhouse"
  inputs = ["oathkeeper_parse"]
  endpoint = "http://ory_clickhouse:8123"
  database = "ory"
  table = "oathkeeper_access_logs"
  compression = "gzip"
  auth.strategy = "basic"
  auth.user = "${ORY_CLICKHOUSE_USER}"
  auth.password = "${ORY_CLICKHOUSE_PASSWORD}"