package com.lbg.kafka.opa;

import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.gson.Gson;
import kafka.security.auth.Acl;
import kafka.security.auth.Authorizer;
import kafka.security.auth.Operation;
import kafka.security.auth.Resource;
import lombok.Cleanup;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.security.auth.KafkaPrincipal;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import static kafka.network.RequestChannel.Session;

@Slf4j
public class OpaAuthorizer implements Authorizer {

  private final static String OPA_AUTHORIZER_URL_CONFIG = "opa.authorizer.url";
  private final static String OPA_AUTHORIZER_DENY_ON_ERROR_CONFIG = "opa.authorizer.allow.on.error";
  private final static String OPA_AUTHORIZER_CACHE_INITIAL_CAPACITY_CONFIG = "opa.authorizer.cache.initial.capacity";
  private final static String OPA_AUTHORIZER_CACHE_MAXIMUM_SIZE_CONFIG = "opa.authorizer.cache.maximum.size";
  private final static String OPA_AUTHORIZER_CACHE_EXPIRE_AFTER_MS_CONFIG = "opa.authorizer.cache.expire.after.ms";

  private String opaUrl;
  private boolean allowOnError;
  private int initialCapacity;
  private int maximumSize;
  private long expireAfterMs;

  private final Gson gson = new Gson();

  private final Map<String, Object> configs = new HashMap<>();

  private LoadingCache<Msg.Input, Boolean> cache = CacheBuilder.newBuilder()
    .initialCapacity(initialCapacity)
    .maximumSize(maximumSize)
    .expireAfterWrite(expireAfterMs, TimeUnit.MILLISECONDS)
    .build(
      new CacheLoader<Msg.Input, Boolean>() {
        @Override
        public Boolean load(Msg.Input input) {
          return allow(input);
        }
      }
    );

  private boolean allow(Msg.Input input) {
    try {
      HttpURLConnection conn = (HttpURLConnection) new URL(opaUrl).openConnection();

      conn.setDoOutput(true);
      conn.setRequestMethod("POST");
      conn.setRequestProperty("Content-Type", "application/json");

      String data = gson.toJson(new Msg(input));
      OutputStream os = conn.getOutputStream();
      os.write(data.getBytes());
      os.flush();

      if (log.isTraceEnabled()) {
        log.trace("Response code: {}, Request data: {}", conn.getResponseCode(), data);
      }

      @Cleanup BufferedReader br = new BufferedReader(new InputStreamReader((conn.getInputStream())));
      return (boolean) gson.fromJson(br.readLine(), Map.class).get("result");
    } catch (IOException e) {
      return allowOnError;
    }
  }

  public boolean authorize(Session session, Operation operation, Resource resource) {
    try {
      return cache.get(new Msg.Input(operation, resource, session));
    } catch (ExecutionException e) {
      return allowOnError;
    }
  }

  public void configure(Map<String, ?> configs) {
    this.configs.putAll(configs);
    if (log.isTraceEnabled()) {
      log.trace("CONFIGS: {}", this.configs);
    }
    opaUrl = (String) configs.get(OPA_AUTHORIZER_URL_CONFIG);
    allowOnError = Boolean.valueOf((String) configs.get(OPA_AUTHORIZER_DENY_ON_ERROR_CONFIG));
    initialCapacity = Integer.parseInt((String) configs.get(OPA_AUTHORIZER_CACHE_INITIAL_CAPACITY_CONFIG));
    maximumSize = Integer.parseInt((String) configs.get(OPA_AUTHORIZER_CACHE_MAXIMUM_SIZE_CONFIG));
    expireAfterMs = Long.parseLong((String) configs.get(OPA_AUTHORIZER_CACHE_EXPIRE_AFTER_MS_CONFIG));
  }

  public void addAcls(scala.collection.immutable.Set<Acl> acls, Resource resource) {
  }

  public boolean removeAcls(scala.collection.immutable.Set<Acl> acls, Resource resource) {
    return false;
  }

  public boolean removeAcls(Resource resource) {
    return false;
  }

  public scala.collection.immutable.Set<Acl> getAcls(Resource resource) {
    return null;
  }

  public scala.collection.immutable.Map<Resource, scala.collection.immutable.Set<Acl>> getAcls(KafkaPrincipal principal) {
    return null;
  }

  public scala.collection.immutable.Map<Resource, scala.collection.immutable.Set<Acl>> getAcls() {
    return null;
  }

  public void close() {
  }

  @Data
  static class Msg {
    private final Input input;

    @Data
    static class Input {
      private final Operation operation;
      private final Resource resource;
      private final Session session;
    }
  }
}

@@ -691,24 +691,51 @@
 a.util.Map;%0A
+import java.util.Optional;%0A
 import java.
@@ -2878,32 +2878,196 @@
 ror;%0A    %7D%0A  %7D%0A%0A
+  public String getValueOrDefault(String property, String defaultValue) %7B%0A    return Optional.ofNullable((String) configs.get(property)).orElse(defaultValue);%0A  %7D%0A%0A
   public boolean
@@ -3132,24 +3132,24 @@
 resource) %7B%0A
-
     try %7B%0A  
@@ -3476,34 +3476,40 @@
  = (String) 
-configs.ge
+getValueOrDefaul
 t(OPA_AUTHOR
@@ -3523,16 +3523,41 @@
 L_CONFIG
+, %22http://localhost:8181%22
 );%0A    a
@@ -3591,34 +3591,40 @@
 Of((String) 
-configs.ge
+getValueOrDefaul
 t(OPA_AUTHOR
@@ -3644,24 +3644,33 @@
 ERROR_CONFIG
+, %22false%22
 ));%0A    init
@@ -3705,34 +3705,40 @@
 nt((String) 
-configs.ge
+getValueOrDefaul
 t(OPA_AUTHOR
@@ -3767,24 +3767,31 @@
 ACITY_CONFIG
+, %22100%22
 ));%0A    maxi
@@ -3822,34 +3822,40 @@
 nt((String) 
-configs.ge
+getValueOrDefaul
 t(OPA_AUTHOR
@@ -3880,24 +3880,31 @@
 _SIZE_CONFIG
+, %22100%22
 ));%0A    expi
@@ -3939,26 +3939,32 @@
 String) 
-configs.ge
+getValueOrDefaul
 t(OPA_AU
@@ -4000,16 +4000,26 @@
 S_CONFIG
+, %22600000%22
 ));%0A  %7D%0A

make parameters optional to avoid null pointers