package com.octaldata.session.kafka

import com.octaldata.domain.DeploymentConnectionConfig
import com.octaldata.session.AccessControlResult
import com.octaldata.session.Acl
import com.octaldata.session.AclOps
import org.apache.kafka.common.acl.AccessControlEntry
import org.apache.kafka.common.acl.AccessControlEntryFilter
import org.apache.kafka.common.acl.AclBinding
import org.apache.kafka.common.acl.AclBindingFilter
import org.apache.kafka.common.acl.AclOperation
import org.apache.kafka.common.acl.AclPermissionType
import org.apache.kafka.common.errors.SecurityDisabledException
import org.apache.kafka.common.resource.PatternType
import org.apache.kafka.common.resource.ResourcePattern
import org.apache.kafka.common.resource.ResourcePatternFilter
import org.apache.kafka.common.resource.ResourceType

class KafkaAclOps(private val config: DeploymentConnectionConfig.KafkaCompatible) : AclOps {

   override suspend fun acls(): Result<AccessControlResult> = runCatching {
      config.withAdminClient { admin ->
         try {
            val acls = admin.describeAcls(AclBindingFilter.ANY).values().suspend().map {
               Acl(
                  principal = it.entry().principal(), // eg User:paul
                  resourceType = it.pattern().resourceType().name,
                  patternType = it.pattern().patternType().name, // literal,match
                  resourceName = it.pattern().name(), // eg topic-123
                  operation = it.entry().operation().name, // WRITE,DESCRIBE etc
                  permissionType = it.entry().permissionType().name, // ANY,DENY,ALLOW
                  host = it.entry().host(), // host or *
               )
            }
            AccessControlResult(true, acls)
         } catch (e: SecurityDisabledException) {
            AccessControlResult(false, emptyList())
         }
      }
   }

   override suspend fun deleteAcl(acls: List<Acl>): Result<Boolean> = runCatching {
      config.withAdminClient { admin ->
         admin.deleteAcls(acls.map {
            AclBindingFilter(
               ResourcePatternFilter(
                  ResourceType.valueOf(it.resourceType),
                  it.resourceName,
                  PatternType.valueOf(it.patternType)
               ),
               AccessControlEntryFilter(
                  it.principal,
                  it.host,
                  AclOperation.valueOf(it.operation),
                  AclPermissionType.valueOf(it.permissionType)
               )
            )
         }).all().suspend()
         true
      }
   }

   override suspend fun createAcls(acls: List<Acl>): Result<Boolean> = runCatching {
      config.withAdminClient { admin ->
         admin.createAcls(acls.map {
            AclBinding(
               ResourcePattern(
                  ResourceType.valueOf(it.resourceType),
                  it.resourceName,
                  PatternType.valueOf(it.patternType)
               ),
               AccessControlEntry(
                  it.principal,
                  it.host,
                  AclOperation.valueOf(it.operation),
                  AclPermissionType.valueOf(it.permissionType)
               )
            )
         }).all().suspend()
         true
      }
   }
}

// KafkaPrincipal,ResourceType,PatternType,ResourceName,Operation,PermissionType,Host
// User:john,TOPIC,LITERAL,*,WRITE,ALLOW,*
// User:john,TOPIC,LITERAL,*,CREATE,ALLOW,*
// User:john,TOPIC,LITERAL,*,DESCRIBE,ALLOW,*
// User:john,CLUSTER,LITERAL,kafka-cluster,IDEMPOTENT_WRITE,ALLOW,*

