/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.kafka.metadata.authorizer;

import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.server.immutable.ImmutableMap;
import org.apache.kafka.server.immutable.ImmutableNavigableSet;

import java.util.ArrayList;
import java.util.List;

/**
 * An immutable class that stores the ACLs in KRaft-based clusters.
 */
public class AclCache {
    /**
     * Contains all of the current ACLs sorted by (resource type, resource name).
     */
    private final ImmutableNavigableSet<StandardAcl> aclsByResource;

    /**
     * Contains all of the current ACLs indexed by UUID.
     */
    private final ImmutableMap<Uuid, StandardAcl> aclsById;

    AclCache() {
        this(ImmutableNavigableSet.empty(), ImmutableMap.empty());
    }

    private AclCache(final ImmutableNavigableSet<StandardAcl> aclsByResource, final ImmutableMap<Uuid, StandardAcl> aclsById) {
        this.aclsByResource = aclsByResource;
        this.aclsById = aclsById;
    }

    public ImmutableNavigableSet<StandardAcl> aclsByResource() {
        return aclsByResource;
    }

    Iterable<AclBinding> acls(AclBindingFilter filter) {
        List<AclBinding> aclBindingList = new ArrayList<>();
        aclsByResource.forEach(acl -> {
            AclBinding aclBinding = acl.toBinding();
            if (filter.matches(aclBinding)) {
                aclBindingList.add(aclBinding);
            }
        });
        return aclBindingList;
    }

    int count() {
        return aclsById.size();
    }

    StandardAcl getAcl(Uuid id) {
        return aclsById.get(id);
    }

    AclCache addAcl(Uuid id, StandardAcl acl) {
        StandardAcl prevAcl = this.aclsById.get(id);
        if (prevAcl != null) {
            throw new RuntimeException("An ACL with ID " + id + " already exists.");
        }

        ImmutableMap<Uuid, StandardAcl> aclsById = this.aclsById.updated(id, acl);

        if (this.aclsByResource.contains(acl)) {
            throw new RuntimeException("Unable to add the ACL with ID " + id +
                    " to aclsByResource");
        }

        ImmutableNavigableSet<StandardAcl> aclsByResource = this.aclsByResource.added(acl);
        return new AclCache(aclsByResource, aclsById);
    }

    AclCache removeAcl(Uuid id) {
        StandardAcl acl = this.aclsById.get(id);
        if (acl == null) {
            throw new RuntimeException("ID " + id + " not found in aclsById.");
        }
        ImmutableMap<Uuid, StandardAcl> aclsById = this.aclsById.removed(id);

        if (!this.aclsByResource.contains(acl)) {
            throw new RuntimeException("Unable to remove the ACL with ID " + id +
                    " from aclsByResource");
        }

        ImmutableNavigableSet<StandardAcl> aclsByResource = this.aclsByResource.removed(acl);
        return new AclCache(aclsByResource, aclsById);
    }
}
