// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.vespa.hosted.controller.restapi.systemflags;

import com.yahoo.concurrent.DaemonThreadFactory;
import com.yahoo.vespa.athenz.identity.ServiceIdentityProvider;
import com.yahoo.vespa.flags.FlagId;
import com.yahoo.vespa.flags.json.FlagData;
import com.yahoo.vespa.hosted.controller.api.systemflags.v1.FlagsTarget;
import com.yahoo.vespa.hosted.controller.api.systemflags.v1.SystemFlagsDataArchive;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.function.Function;
import java.util.stream.Collectors;

import static com.yahoo.vespa.hosted.controller.restapi.systemflags.SystemFlagsDeployResult.FlagDataChange;

/**
 * Deploy a flags data archive to all targets in a given system
 *
 * @author bjorncs
 */
class SystemFlagsDeployer  {

    private final FlagsClient client;
    private final Set<FlagsTarget> targets;
    private final ExecutorCompletionService<SystemFlagsDeployResult> completionService =
            new ExecutorCompletionService<>(Executors.newCachedThreadPool(new DaemonThreadFactory("system-flags-deployer-")));


    SystemFlagsDeployer(ServiceIdentityProvider identityProvider, Set<FlagsTarget> targets) {
        this(new FlagsClient(identityProvider, targets), targets);
    }

    SystemFlagsDeployer(FlagsClient client, Set<FlagsTarget> targets) {
        this.client = client;
        this.targets = targets;
    }

    SystemFlagsDeployResult deployFlags(SystemFlagsDataArchive archive, boolean dryRun) {
        for (FlagsTarget target : targets) {
            completionService.submit(() -> deployFlags(target, archive.flagData(target), dryRun));
        }
        List<SystemFlagsDeployResult> results = new ArrayList<>();
        Future<SystemFlagsDeployResult> future;
        try {
            while (results.size() < targets.size() && (future = completionService.take()) != null) {
                try {
                    results.add(future.get());
                } catch (ExecutionException e) {
                    // TODO Handle errors
                    throw new RuntimeException(e);
                }
            }
            return SystemFlagsDeployResult.merge(results);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    // TODO Handle http status code 4xx/5xx (e.g for unknown flag id)
    private SystemFlagsDeployResult deployFlags(FlagsTarget target, Set<FlagData> flagData, boolean dryRun) {
        Map<FlagId, FlagData> wantedFlagData = lookupTable(flagData);
        Map<FlagId, FlagData> currentFlagData = lookupTable(client.listFlagData(target));

        List<FlagDataChange> result = new ArrayList<>();

        wantedFlagData.forEach((id, data) -> {
            if (currentFlagData.containsKey(id)) {
                FlagData currentData = currentFlagData.get(id);
                if (currentData.toJsonNode().equals(data.toJsonNode())) {
                    return; // noop
                }
                result.add(FlagDataChange.updated(id, Set.of(target), data, currentData));
            } else {
                result.add(FlagDataChange.created(id, Set.of(target), data));
            }
            if (!dryRun) {
                client.putFlagData(target, data);
            }
        });

        currentFlagData.forEach((id, data) -> {
            if (!wantedFlagData.containsKey(id)) {
                if (!dryRun) {
                    client.deleteFlagData(target, id);
                }
                result.add(FlagDataChange.deleted(id, Set.of(target)));
            }
        });

        return new SystemFlagsDeployResult(result);
    }

    private static Map<FlagId, FlagData> lookupTable(Collection<FlagData> data) {
        return data.stream().collect(Collectors.toMap(FlagData::id, Function.identity()));
    }

}

@@ -624,16 +624,42 @@
 il.Map;%0A
+import java.util.Objects;%0A
 import j
@@ -3094,63 +3094,8 @@
 %3E %7B%0A
-            if (currentFlagData.containsKey(id)) %7B%0A    
     
@@ -3162,23 +3162,57 @@
         
-    if 
+if (currentData != null && Objects.equals
 (current
@@ -3228,24 +3228,18 @@
 onNode()
-.equals(
+, 
 data.toJ
@@ -3264,28 +3264,24 @@
             
-    
 return; // n
@@ -3284,105 +3284,191 @@
 // n
-oop%0A                %7D%0A                result.add(FlagDataChange.updated(id, Set.of
+ew flag data identical to existing%0A            %7D%0A            if (!dryRun) %7B%0A                client.putFlagData
 (target
-)
 , data
-,
+);%0A            %7D%0A            result.add(%0A                   
  cur
@@ -3475,19 +3475,24 @@
 rentData
-));
+ != null
 %0A       
@@ -3500,17 +3500,8 @@
     
-%7D else %7B%0A
     
@@ -3508,35 +3508,26 @@
             
-result.add(
+? 
 FlagDataChan
@@ -3533,11 +3533,11 @@
 nge.
-cre
+upd
 ated
@@ -3565,116 +3565,103 @@
 data
-));%0A            %7D%0A            if (!dryRun) %7B%0A                client.putFlagData
+, currentData)%0A                            : FlagDataChange.created(id, Set.of
 (target
+)
 , data)
+)
 ;
-%0A            %7D
 %0A   

Add result to list only if put operation succeeds