/*
 * 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.cassandra.schema;

import java.util.*;

import com.google.common.collect.ImmutableMap;

import org.apache.cassandra.config.Schema;

import static com.google.common.collect.Iterables.filter;

For backwards compatibility, in the first instance an IndexMetadata must have
TargetType.COLUMN and its Set of target columns must contain only a single
ColumnIdentifier. Hence, this is what is enforced by the public factory methods
on IndexMetadata.
These constraints, along with the internal datastructures here will be relaxed as
support is added for multiple target columns per-index and for indexes with
TargetType.ROW
/**
 * For backwards compatibility, in the first instance an IndexMetadata must have
 * TargetType.COLUMN and its Set of target columns must contain only a single
 * ColumnIdentifier. Hence, this is what is enforced by the public factory methods
 * on IndexMetadata.
 * These constraints, along with the internal datastructures here will be relaxed as
 * support is added for multiple target columns per-index and for indexes with
 * TargetType.ROW
 */
public class Indexes implements Iterable<IndexMetadata>
{
    private final ImmutableMap<String, IndexMetadata> indexesByName;
    private final ImmutableMap<UUID, IndexMetadata> indexesById;

    private Indexes(Builder builder)
    {
        indexesByName = builder.indexesByName.build();
        indexesById = builder.indexesById.build();
    }

    public static Builder builder()
    {
        return new Builder();
    }

    public static Indexes none()
    {
        return builder().build();
    }

    public Iterator<IndexMetadata> iterator()
    {
        return indexesByName.values().iterator();
    }

    public int size()
    {
        return indexesByName.size();
    }

    public boolean isEmpty()
    {
        return indexesByName.isEmpty();
    }

    
Get the index with the specified name
Params:
name – a non-qualified index name
Returns:
an empty Optional if the named index is not found; a non-empty optional of IndexMetadata otherwise/**
     * Get the index with the specified name
     *
     * @param name a non-qualified index name
     * @return an empty {@link Optional} if the named index is not found; a non-empty optional of {@link IndexMetadata} otherwise
     */
    public Optional<IndexMetadata> get(String name)
    {
        return Optional.ofNullable(indexesByName.get(name));
    }

    
Answer true if contains an index with the specified name.
Params:
name – a non-qualified index name.
Returns:
true if the named index is found; false otherwise/**
     * Answer true if contains an index with the specified name.
     * @param name a non-qualified index name.
     * @return true if the named index is found; false otherwise
     */
    public boolean has(String name)
    {
        return indexesByName.containsKey(name);
    }

    
Get the index with the specified id
Params:
id – a UUID which identifies an index
Returns:
an empty Optional if no index with the specified id is found; a non-empty optional of IndexMetadata otherwise/**
     * Get the index with the specified id
     *
     * @param id a UUID which identifies an index
     * @return an empty {@link Optional} if no index with the specified id is found; a non-empty optional of
     *         {@link IndexMetadata} otherwise
     */

    public Optional<IndexMetadata> get(UUID id)
    {
        return Optional.ofNullable(indexesById.get(id));
    }

    
Answer true if contains an index with the specified id.
Params:
id – a UUID which identifies an index.
Returns:
true if an index with the specified id is found; false otherwise/**
     * Answer true if contains an index with the specified id.
     * @param id a UUID which identifies an index.
     * @return true if an index with the specified id is found; false otherwise
     */
    public boolean has(UUID id)
    {
        return indexesById.containsKey(id);
    }

    
Create a SecondaryIndexes instance with the provided index added
/**
     * Create a SecondaryIndexes instance with the provided index added
     */
    public Indexes with(IndexMetadata index)
    {
        if (get(index.name).isPresent())
            throw new IllegalStateException(String.format("Index %s already exists", index.name));

        return builder().add(this).add(index).build();
    }

    
Creates a SecondaryIndexes instance with the index with the provided name removed
/**
     * Creates a SecondaryIndexes instance with the index with the provided name removed
     */
    public Indexes without(String name)
    {
        IndexMetadata index = get(name).orElseThrow(() -> new IllegalStateException(String.format("Index %s doesn't exist", name)));
        return builder().add(filter(this, v -> v != index)).build();
    }

    
Creates a SecondaryIndexes instance which contains an updated index definition
/**
     * Creates a SecondaryIndexes instance which contains an updated index definition
     */
    public Indexes replace(IndexMetadata index)
    {
        return without(index.name).with(index);
    }

    @Override
    public boolean equals(Object o)
    {
        return this == o || (o instanceof Indexes && indexesByName.equals(((Indexes) o).indexesByName));
    }

    @Override
    public int hashCode()
    {
        return indexesByName.hashCode();
    }

    @Override
    public String toString()
    {
        return indexesByName.values().toString();
    }

    public static String getAvailableIndexName(String ksName, String cfName, String indexNameRoot)
    {

        KeyspaceMetadata ksm = Schema.instance.getKSMetaData(ksName);
        Set<String> existingNames = ksm == null ? new HashSet<>() : ksm.existingIndexNames(null);
        String baseName = IndexMetadata.getDefaultIndexName(cfName, indexNameRoot);
        String acceptedName = baseName;
        int i = 0;
        while (existingNames.contains(acceptedName))
            acceptedName = baseName + '_' + (++i);

        return acceptedName;
    }

    public static final class Builder
    {
        final ImmutableMap.Builder<String, IndexMetadata> indexesByName = new ImmutableMap.Builder<>();
        final ImmutableMap.Builder<UUID, IndexMetadata> indexesById = new ImmutableMap.Builder<>();

        private Builder()
        {
        }

        public Indexes build()
        {
            return new Indexes(this);
        }

        public Builder add(IndexMetadata index)
        {
            indexesByName.put(index.name, index);
            indexesById.put(index.id, index);
            return this;
        }

        public Builder add(Iterable<IndexMetadata> indexes)
        {
            indexes.forEach(this::add);
            return this;
        }
    }
}