content: /*
   *  Copyright 1999-2018 Alibaba Group Holding Ltd.
   *
   *  Licensed 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 com.alibaba.nacos.naming.core.v2.service.impl;
  
  import com.alibaba.nacos.api.exception.NacosException;
  import com.alibaba.nacos.api.exception.runtime.NacosRuntimeException;
  import com.alibaba.nacos.api.naming.pojo.Instance;
  import com.alibaba.nacos.common.notify.NotifyCenter;
  import com.alibaba.nacos.common.utils.ClassUtils;
  import com.alibaba.nacos.common.utils.CollectionUtils;
  import com.alibaba.nacos.consistency.DataOperation;
  import com.alibaba.nacos.consistency.SerializeFactory;
  import com.alibaba.nacos.consistency.Serializer;
  import com.alibaba.nacos.consistency.cp.CPProtocol;
  import com.alibaba.nacos.consistency.cp.RequestProcessor4CP;
  import com.alibaba.nacos.consistency.entity.ReadRequest;
  import com.alibaba.nacos.consistency.entity.Response;
  import com.alibaba.nacos.consistency.entity.WriteRequest;
  import com.alibaba.nacos.consistency.snapshot.LocalFileMeta;
  import com.alibaba.nacos.consistency.snapshot.Reader;
  import com.alibaba.nacos.consistency.snapshot.SnapshotOperation;
  import com.alibaba.nacos.consistency.snapshot.Writer;
  import com.alibaba.nacos.core.distributed.ProtocolManager;
  import com.alibaba.nacos.core.utils.Loggers;
  import com.alibaba.nacos.naming.consistency.persistent.impl.AbstractSnapshotOperation;
  import com.alibaba.nacos.naming.constants.Constants;
  import com.alibaba.nacos.naming.core.v2.ServiceManager;
  import com.alibaba.nacos.naming.core.v2.client.Client;
  import com.alibaba.nacos.naming.core.v2.client.ClientAttributes;
  import com.alibaba.nacos.naming.core.v2.client.ClientSyncData;
  import com.alibaba.nacos.naming.core.v2.client.impl.IpPortBasedClient;
  import com.alibaba.nacos.naming.core.v2.client.manager.impl.PersistentIpPortClientManager;
  import com.alibaba.nacos.naming.core.v2.event.client.ClientOperationEvent;
  import com.alibaba.nacos.naming.core.v2.event.metadata.MetadataEvent;
  import com.alibaba.nacos.naming.core.v2.pojo.InstancePublishInfo;
  import com.alibaba.nacos.naming.core.v2.pojo.Service;
  import com.alibaba.nacos.naming.core.v2.service.ClientOperationService;
  import com.alibaba.nacos.naming.pojo.Subscriber;
  import com.alibaba.nacos.sys.utils.ApplicationUtils;
  import com.alibaba.nacos.sys.utils.DiskUtils;
  import com.alipay.sofa.jraft.util.CRC64;
  import com.google.protobuf.ByteString;
  import org.springframework.stereotype.Component;
  
  import java.io.ByteArrayInputStream;
  import java.io.IOException;
  import java.io.InputStream;
  import java.io.Serializable;
  import java.nio.file.Paths;
  import java.util.ArrayList;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Set;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReentrantReadWriteLock;
  import java.util.zip.Checksum;
  
  /**
   * Operation service for persistent clients and services. only for v2 For persistent instances, clientId must be in the
   * format of host:port.
   *
   * @author <a href="mailto:liaochuntao@live.com">liaochuntao</a>
   * @author xiweng.yy
   */
  @Component("persistentClientOperationServiceImpl")
  public class PersistentClientOperationServiceImpl extends RequestProcessor4CP implements ClientOperationService {
      
      private final PersistentIpPortClientManager clientManager;
      
      private final Serializer serializer = SerializeFactory.getDefault();
      
      private final CPProtocol protocol;
      
      private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
      
      private final ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
      
      private static final int INITIAL_CAPACITY = 128;
      
      public PersistentClientOperationServiceImpl(final PersistentIpPortClientManager clientManager) {
          this.clientManager = clientManager;
          this.protocol = ApplicationUtils.getBean(ProtocolManager.class).getCpProtocol();
          this.protocol.addRequestProcessors(Collections.singletonList(this));
      }
      
      @Override
      public void registerInstance(Service service, Instance instance, String clientId) {
          Service singleton = ServiceManager.getInstance().getSingleton(service);
          if (singleton.isEphemeral()) {
              throw new NacosRuntimeException(NacosException.INVALID_PARAM,
                      String.format("Current service %s is ephemeral service, can't register persistent instance.",
                              singleton.getGroupedServiceName()));
          }
          final InstanceStoreRequest request = new InstanceStoreRequest();
          request.setService(service);
          request.setInstance(instance);
          request.setClientId(clientId);
          final WriteRequest writeRequest = WriteRequest.newBuilder().setGroup(group())
                  .setData(ByteString.copyFrom(serializer.serialize(request))).setOperation(DataOperation.ADD.name())
                  .build();
          
          try {
              protocol.write(writeRequest);
              Loggers.RAFT.info("Client registered. service={}, clientId={}, instance={}", service, clientId, instance);
          } catch (Exception e) {
              throw new NacosRuntimeException(NacosException.SERVER_ERROR, e);
          }
      }
      
      /**
       * update persistent instance.
       */
      public void updateInstance(Service service, Instance instance, String clientId) {
          Service singleton = ServiceManager.getInstance().getSingleton(service);
          if (singleton.isEphemeral()) {
              throw new NacosRuntimeException(NacosException.INVALID_PARAM,
                      String.format("Current service %s is ephemeral service, can't update persistent instance.",
                              singleton.getGroupedServiceName()));
          }
          final PersistentClientOperationServiceImpl.InstanceStoreRequest request = new PersistentClientOperationServiceImpl.InstanceStoreRequest();
          request.setService(service);
          request.setInstance(instance);
          request.setClientId(clientId);
          final WriteRequest writeRequest = WriteRequest.newBuilder().setGroup(group())
                  .setData(ByteString.copyFrom(serializer.serialize(request))).setOperation(DataOperation.CHANGE.name())
                  .build();
          try {
              protocol.write(writeRequest);
          } catch (Exception e) {
              throw new NacosRuntimeException(NacosException.SERVER_ERROR, e);
          }
      }
      
      @Override
      public void batchRegisterInstance(Service service, List<Instance> instances, String clientId) {
          //TODO PersistentClientOperationServiceImpl Nacos batchRegister
      }
      
      @Override
      public void deregisterInstance(Service service, Instance instance, String clientId) {
          final InstanceStoreRequest request = new InstanceStoreRequest();
          request.setService(service);
          request.setInstance(instance);
          request.setClientId(clientId);
          final WriteRequest writeRequest = WriteRequest.newBuilder().setGroup(group())
                  .setData(ByteString.copyFrom(serializer.serialize(request))).setOperation(DataOperation.DELETE.name())
                  .build();
          
          try {
              protocol.write(writeRequest);
              Loggers.RAFT.info("Client unregistered. service={}, clientId={}, instance={}", service, clientId, instance);
          } catch (Exception e) {
              throw new NacosRuntimeException(NacosException.SERVER_ERROR, e);
          }
      }
      
      @Override
      public void subscribeService(Service service, Subscriber subscriber, String clientId) {
          throw new UnsupportedOperationException("No persistent subscribers");
      }
      
      @Override
      public void unsubscribeService(Service service, Subscriber subscriber, String clientId) {
          throw new UnsupportedOperationException("No persistent subscribers");
      }
      
      @Override
      public Response onRequest(ReadRequest request) {
          throw new UnsupportedOperationException("Temporary does not support");
      }
      
      @Override
      public Response onApply(WriteRequest request) {
          final Lock lock = readLock;
          lock.lock();
          try {
              final InstanceStoreRequest instanceRequest = serializer.deserialize(request.getData().toByteArray());
              final DataOperation operation = DataOperation.valueOf(request.getOperation());
              switch (operation) {
                  case ADD:
                      onInstanceRegister(instanceRequest.service, instanceRequest.instance,
                              instanceRequest.getClientId());
                      break;
                  case DELETE:
                      onInstanceDeregister(instanceRequest.service, instanceRequest.getClientId());
                      break;
                  case CHANGE:
                      if (instanceAndServiceExist(instanceRequest)) {
                          onInstanceRegister(instanceRequest.service, instanceRequest.instance,
                                  instanceRequest.getClientId());
                      }
                      break;
                  default:
                      return Response.newBuilder().setSuccess(false).setErrMsg("unsupport operation : " + operation)
                              .build();
              }
              return Response.newBuilder().setSuccess(true).build();
          } catch (Exception e) {
              Loggers.RAFT.warn("Persistent client operation failed. ", e);
              return Response.newBuilder().setSuccess(false)
                      .setErrMsg("Persistent client operation failed. " + e.getMessage()).build();
          } finally {
              lock.unlock();
          }
      }
      
      private boolean instanceAndServiceExist(InstanceStoreRequest instanceRequest) {
          return clientManager.contains(instanceRequest.getClientId()) && clientManager.getClient(
                  instanceRequest.getClientId()).getAllPublishedService().contains(instanceRequest.service);
      }
      
      private void onInstanceRegister(Service service, Instance instance, String clientId) {
          Service singleton = ServiceManager.getInstance().getSingleton(service);
          if (!clientManager.contains(clientId)) {
              clientManager.clientConnected(clientId, new ClientAttributes());
          }
          Client client = clientManager.getClient(clientId);
          InstancePublishInfo instancePublishInfo = getPublishInfo(instance);
          client.addServiceInstance(singleton, instancePublishInfo);
          client.setLastUpdatedTime();
          NotifyCenter.publishEvent(new ClientOperationEvent.ClientRegisterServiceEvent(singleton, clientId));
          NotifyCenter.publishEvent(
                  new MetadataEvent.InstanceMetadataEvent(singleton, instancePublishInfo.getMetadataId(), false));
      }
      
      private void onInstanceDeregister(Service service, String clientId) {
          Service singleton = ServiceManager.getInstance().getSingleton(service);
          Client client = clientManager.getClient(clientId);
          if (client == null) {
              Loggers.RAFT.warn("client not exist onInstanceDeregister, clientId : {} ", clientId);
              return;
          }
          final InstancePublishInfo removedInstance = client.removeServiceInstance(singleton);
          client.setLastUpdatedTime();
          if (client.getAllPublishedService().isEmpty()) {
              clientManager.clientDisconnected(clientId);
          }
          NotifyCenter.publishEvent(new ClientOperationEvent.ClientDeregisterServiceEvent(singleton, clientId));
          if (null != removedInstance) {
              NotifyCenter.publishEvent(
                      new MetadataEvent.InstanceMetadataEvent(singleton, removedInstance.getMetadataId(), true));
          }
      }
      
      @Override
      public List<SnapshotOperation> loadSnapshotOperate() {
          return Collections.singletonList(new PersistentInstanceSnapshotOperation(lock));
      }
      
      @Override
      public String group() {
          return Constants.NAMING_PERSISTENT_SERVICE_GROUP_V2;
      }
      
      public static class InstanceStoreRequest implements Serializable {
          
          private static final long serialVersionUID = -9077205657156890549L;
          
          private Service service;
          
          private Instance instance;
          
          private String clientId;
          
          public Service getService() {
              return service;
          }
          
          public void setService(Service service) {
              this.service = service;
          }
          
          public Instance getInstance() {
              return instance;
          }
          
          public void setInstance(Instance instance) {
              this.instance = instance;
          }
          
          public String getClientId() {
              return clientId;
          }
          
          public void setClientId(String clientId) {
              this.clientId = clientId;
          }
          
      }
      
      private class PersistentInstanceSnapshotOperation extends AbstractSnapshotOperation {
          
          private final String snapshotSaveTag = ClassUtils.getSimpleName(getClass()) + ".SAVE";
          
          private final String snapshotLoadTag = ClassUtils.getSimpleName(getClass()) + ".LOAD";
          
          private static final String SNAPSHOT_ARCHIVE = "persistent_instance.zip";
          
          public PersistentInstanceSnapshotOperation(ReentrantReadWriteLock lock) {
              super(lock);
          }
          
          @Override
          protected boolean writeSnapshot(Writer writer) throws IOException {
              final String writePath = writer.getPath();
              final String outputFile = Paths.get(writePath, SNAPSHOT_ARCHIVE).toString();
              final Checksum checksum = new CRC64();
              try (InputStream inputStream = dumpSnapshot()) {
                  DiskUtils.compressIntoZipFile("instance", inputStream, outputFile, checksum);
              }
              final LocalFileMeta meta = new LocalFileMeta();
              meta.append(CHECK_SUM_KEY, Long.toHexString(checksum.getValue()));
              return writer.addFile(SNAPSHOT_ARCHIVE, meta);
          }
          
          @Override
          protected boolean readSnapshot(Reader reader) throws Exception {
              final String readerPath = reader.getPath();
              Loggers.RAFT.info("snapshot start to load from : {}", readerPath);
              final String sourceFile = Paths.get(readerPath, SNAPSHOT_ARCHIVE).toString();
              final Checksum checksum = new CRC64();
              byte[] snapshotBytes = DiskUtils.decompress(sourceFile, checksum);
              LocalFileMeta fileMeta = reader.getFileMeta(SNAPSHOT_ARCHIVE);
              if (fileMeta.getFileMeta().containsKey(CHECK_SUM_KEY) && !Objects.equals(
                      Long.toHexString(checksum.getValue()), fileMeta.get(CHECK_SUM_KEY))) {
                  throw new IllegalArgumentException("Snapshot checksum failed");
              }
              loadSnapshot(snapshotBytes);
              Loggers.RAFT.info("snapshot success to load from : {}", readerPath);
              return true;
          }
          
          protected InputStream dumpSnapshot() {
              Map<String, IpPortBasedClient> clientMap = clientManager.showClients();
              ConcurrentHashMap<String, ClientSyncData> clone = new ConcurrentHashMap<>(INITIAL_CAPACITY);
              clientMap.forEach((clientId, client) -> clone.put(clientId, client.generateSyncData()));
              return new ByteArrayInputStream(serializer.serialize(clone));
          }
          
          protected void loadSnapshot(byte[] snapshotBytes) {
              ConcurrentHashMap<String, ClientSyncData> newData = serializer.deserialize(snapshotBytes);
              Collection<String> oldClientIds = clientManager.allClientId();
              // add or update
              for (Map.Entry<String, ClientSyncData> entry : newData.entrySet()) {
                  if (oldClientIds.contains(entry.getKey())) {
                      // update alive client
                      updateSyncDataToClient(entry, (IpPortBasedClient) clientManager.getClient(entry.getKey()));
                  } else {
                      // add new client
                      IpPortBasedClient snapshotClient = new IpPortBasedClient(entry.getKey(), false);
                      snapshotClient.setAttributes(entry.getValue().getAttributes());
                      snapshotClient.init();
                      addSyncDataToClient(entry, snapshotClient);
                  }
              }
              // remove dead client
              removeDeadClient(newData.keySet(), oldClientIds);
          }
          
          /**
           * update instance info for client.
           *
           * @param entry entry
           * @param client client
           */
          private void updateSyncDataToClient(Map.Entry<String, ClientSyncData> entry, IpPortBasedClient client) {
              ClientSyncData data = entry.getValue();
              List<String> namespaces = data.getNamespaces();
              List<String> groupNames = data.getGroupNames();
              List<String> serviceNames = data.getServiceNames();
              List<InstancePublishInfo> instances = data.getInstancePublishInfos();
              // alive instance data: Service = InstancePublishInfo
              Map<Service, InstancePublishInfo> newInstanceInfoMap = new HashMap<>(instances.size());
              for (int i = 0; i < namespaces.size(); i++) {
                  Service service = Service.newService(namespaces.get(i), groupNames.get(i), serviceNames.get(i), false);
                  newInstanceInfoMap.put(service, instances.get(i));
              }
              // old instance data
              Collection<Service> oldPublishedService = client.getAllPublishedService();
              Set<Service> aliveInstanceServices = newInstanceInfoMap.keySet();
              // add or update existed service
              for (Service service : aliveInstanceServices) {
                  Service singleton = ServiceManager.getInstance().getSingleton(service);
                  InstancePublishInfo newInstanceInfo = newInstanceInfoMap.get(singleton);
                  if (oldPublishedService.contains(singleton)) {
                      // update if necessary
                      InstancePublishInfo oldInstanceInfo = client.getInstancePublishInfo(singleton);
                      if (oldInstanceInfo != null && !newInstanceInfo.equals(oldInstanceInfo)) {
                          client.putServiceInstance(singleton, newInstanceInfo);
                          NotifyCenter.publishEvent(
                                  new ClientOperationEvent.ClientRegisterServiceEvent(singleton, client.getClientId()));
                          Loggers.RAFT.info("[SNAPSHOT-DATA-UPDATE] service={}, instance={}", service, newInstanceInfo);
                      }
                  } else {
                      // add
                      client.putServiceInstance(singleton, newInstanceInfo);
                      NotifyCenter.publishEvent(
                              new ClientOperationEvent.ClientRegisterServiceEvent(singleton, client.getClientId()));
                      Loggers.RAFT.info("[SNAPSHOT-DATA-ADD] service={}, instance={}", service, newInstanceInfo);
                  }
              }
              // remove dead instance
              for (Service service : oldPublishedService) {
                  if (!aliveInstanceServices.contains(service)) {
                      InstancePublishInfo oldInfo = client.getInstancePublishInfo(service);
                      // metric ip count decrement
                      client.removeServiceInstance(service);
                      NotifyCenter.publishEvent(
                              new ClientOperationEvent.ClientDeregisterServiceEvent(service, client.getClientId()));
                      Loggers.RAFT.info("[SNAPSHOT-DATA-REMOVE] service={}, instance={}", service, oldInfo);
                  }
              }
          }
          
          /**
           * remove certain client which has dead.
           *
           * @param aliveClientIds new client ids
           * @param oldClientIds old client ids
           */
          private void removeDeadClient(Collection<String> aliveClientIds, Collection<String> oldClientIds) {
              // return if empty
              if (CollectionUtils.isEmpty(oldClientIds)) {
                  return;
              }
              for (String oldClientId : oldClientIds) {
                  // no contains if discaonnect
                  if (!aliveClientIds.contains(oldClientId)) {
                      Client client = clientManager.getClient(oldClientId);
                      // remove all publishedService
                      if (client != null) {
                          if (CollectionUtils.isNotEmpty(client.getAllPublishedService())) {
                              for (Service service : client.getAllPublishedService()) {
                                  Service singleton = ServiceManager.getInstance().getSingleton(service);
                                  InstancePublishInfo oldInfo = client.getInstancePublishInfo(service);
                                  // metric ip count decrement
                                  client.removeServiceInstance(service);
                                  NotifyCenter.publishEvent(
                                          new ClientOperationEvent.ClientDeregisterServiceEvent(singleton,
                                                  client.getClientId()));
                                  Loggers.RAFT.info("[SNAPSHOT-DATA-REMOVE] service={}, instance={}", singleton, oldInfo);
                              }
                          }
                          // remove client
                          clientManager.removeAndRelease(client.getClientId());
                          Loggers.RAFT.info("[SNAPSHOT-DATA-REMOVE] client={}", client);
                      }
                  }
              }
          }
          
          private void addSyncDataToClient(Map.Entry<String, ClientSyncData> entry, IpPortBasedClient client) {
              ClientSyncData data = entry.getValue();
              List<String> namespaces = data.getNamespaces();
              List<String> groupNames = data.getGroupNames();
              List<String> serviceNames = data.getServiceNames();
              List<InstancePublishInfo> instances = data.getInstancePublishInfos();
              List<ClientOperationEvent.ClientRegisterServiceEvent> waitPublishEvents = new ArrayList<>();
              for (int i = 0; i < namespaces.size(); i++) {
                  Service service = Service.newService(namespaces.get(i), groupNames.get(i), serviceNames.get(i), false);
                  Service singleton = ServiceManager.getInstance().getSingleton(service);
                  client.putServiceInstance(singleton, instances.get(i));
                  Loggers.RAFT.info("[SNAPSHOT-DATA-ADD] service={}, instance={}", service, instances.get(i));
                  waitPublishEvents.add(
                          new ClientOperationEvent.ClientRegisterServiceEvent(singleton, client.getClientId()));
              }
              clientManager.addSyncClient(client);
              for (ClientOperationEvent.ClientRegisterServiceEvent waitPublishEvent : waitPublishEvents) {
                  NotifyCenter.publishEvent(waitPublishEvent);
              }
          }
          
          @Override
          protected String getSnapshotSaveTag() {
              return snapshotSaveTag;
          }
          
          @Override
          protected String getSnapshotLoadTag() {
              return snapshotLoadTag;
          }
      }
      
  }
  