/*
 * 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.skywalking.oap.server.configuration.etcd;

import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.ClientBuilder;
import io.etcd.jetcd.KV;
import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.kv.GetResponse;
import io.etcd.jetcd.options.GetOption;
import java.nio.charset.Charset;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.library.util.StringUtil;
import org.apache.skywalking.oap.server.configuration.api.ConfigTable;
import org.apache.skywalking.oap.server.configuration.api.ConfigWatcherRegister;
import org.apache.skywalking.oap.server.configuration.api.GroupConfigTable;

@Slf4j
public class EtcdConfigWatcherRegister extends ConfigWatcherRegister {

    private final KV client;

    public EtcdConfigWatcherRegister(EtcdServerSettings setting) {
        super(setting.getPeriod());
        ClientBuilder builder = Client.builder()
                                      .authority(setting.getAuthority())
                                      .endpoints(setting.getEndpointArray());

        if (StringUtil.isNotEmpty(setting.getNamespace())) {
            builder.namespace(ByteSequence.from(setting.getNamespace(), Charset.defaultCharset()));
        }
        if (setting.isAuthentication()) {
            builder.user(ByteSequence.from(setting.getUser(), Charset.defaultCharset()))
                   .password(ByteSequence.from(setting.getPassword(), Charset.defaultCharset()));
        }
        client = builder.build().getKVClient();
    }

    @Override
    public Optional<ConfigTable> readConfig(final Set<String> keys) {
        ConfigTable table = new ConfigTable();
        keys.forEach(e -> {
            try {
                GetResponse response = client.get(ByteSequence.from(e, Charset.defaultCharset())).get();

                if (0 == response.getCount()) {
                    table.add(new ConfigTable.ConfigItem(e, null));
                } else {
                    response.getKvs().forEach(kv -> table.add(new ConfigTable.ConfigItem(
                                                                  kv.getKey().toString(Charset.defaultCharset()),
                                                                  kv.getValue().toString(Charset.defaultCharset())
                                                              )
                    ));
                }
            } catch (Exception exp) {
                throw new EtcdConfigException("Failed to read configuration", exp);
            }
        });
        return Optional.of(table);
    }

    @Override
    public Optional<GroupConfigTable> readGroupConfig(final Set<String> keys) {
        GroupConfigTable groupConfigTable = new GroupConfigTable();
        keys.forEach(key -> {
            GroupConfigTable.GroupConfigItems groupConfigItems = new GroupConfigTable.GroupConfigItems(key);
            groupConfigTable.addGroupConfigItems(groupConfigItems);
            String groupKey = key + "/";

            GetOption option = GetOption.newBuilder()
                                        .withPrefix(ByteSequence.from(groupKey, Charset.defaultCharset()))
                                        .build();
            try {
                GetResponse response = client.get(ByteSequence.from(groupKey, Charset.defaultCharset()), option).get();
                if (0 != response.getCount()) {
                    List<KeyValue> groupItemKeys = response.getKvs();
                    if (groupItemKeys != null) {
                        groupItemKeys.forEach(groupItem -> {
                            String groupItemKey = groupItem.getKey().toString(Charset.defaultCharset());
                            if (!groupKey.equals(groupItemKey)) {
                                String itemValue = groupItem.getValue().toString(Charset.defaultCharset());
                                String itemName = groupItemKey.substring(groupKey.length());
                                groupConfigItems.add(
                                    new ConfigTable.ConfigItem(itemName, itemValue));
                            }
                        });
                    }
                }
            } catch (Exception exp) {
                throw new EtcdConfigException("Failed to read configuration", exp);
            }

        });
        return Optional.of(groupConfigTable);
    }

}
