package com.hermesworld.ais.galapagos.adminjobs.impl;

import com.hermesworld.ais.galapagos.kafka.KafkaCluster;
import com.hermesworld.ais.galapagos.kafka.KafkaClusters;
import org.apache.kafka.common.acl.AclBinding;
import org.json.JSONArray;
import org.json.JSONObject;
import org.springframework.boot.ApplicationArguments;
import org.springframework.stereotype.Component;

/**
 * Admin job to retrieve ALL ACL entries of a given Kafka Cluster, in JSON format. <br>
 * Note that the result is <b>not</b> limited to ACL entries generated by Galapagos. <br>
 * The job requires one parameter:
 * <ul>
 * <li><code>--kafka.environment=<i>&lt;id></i> - The ID of the Kafka Environment to return the ACL entries of, as
 * configured for Galapagos.</li>
 * </ul>
 *
 * @author AlbrechtFlo
 *
 */
@Component
public class ViewAclsJob extends SingleClusterAdminJob {

    public ViewAclsJob(KafkaClusters kafkaClusters) {
        super(kafkaClusters);
    }

    @Override
    public String getJobName() {
        return "view-acls";
    }

    @Override
    public void runOnCluster(KafkaCluster cluster, ApplicationArguments allArguments) throws Exception {
        JSONArray acls = new JSONArray();
        cluster.visitAcls(acl -> {
            acls.put(toJsonObject(acl));
            return true;
        }).get();

        System.out.println();
        System.out.println();
        System.out.println(acls.length() + " ACLs found:");
        System.out.println(acls);
        System.out.println();
        System.out.println();
    }

    private JSONObject toJsonObject(AclBinding aclBinding) {
        JSONObject pattern = new JSONObject();
        pattern.put("resourceType", aclBinding.pattern().resourceType().toString());
        pattern.put("patternType", aclBinding.pattern().patternType().toString());
        pattern.put("name", aclBinding.pattern().name());

        JSONObject entry = new JSONObject();
        entry.put("principal", aclBinding.entry().principal());
        entry.put("host", aclBinding.entry().host());
        entry.put("operation", aclBinding.entry().operation().toString());
        entry.put("permissionType", aclBinding.entry().permissionType().toString());

        JSONObject result = new JSONObject();
        result.put("pattern", pattern);
        result.put("entry", entry);

        return result;
    }

}
