/*
 * Copyright 2017 Google, Inc.
 *
 * 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.netflix.spinnaker.orca.clouddriver.tasks.manifest;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;
import javax.annotation.Nonnull;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableMap;
import com.netflix.spinnaker.kork.artifacts.model.Artifact;
import com.netflix.spinnaker.kork.core.RetrySupport;
import com.netflix.spinnaker.orca.ExecutionStatus;
import com.netflix.spinnaker.orca.Task;
import com.netflix.spinnaker.orca.TaskResult;
import com.netflix.spinnaker.orca.clouddriver.KatoService;
import com.netflix.spinnaker.orca.clouddriver.OortService;
import com.netflix.spinnaker.orca.clouddriver.model.TaskId;
import com.netflix.spinnaker.orca.clouddriver.tasks.AbstractCloudProviderAwareTask;
import com.netflix.spinnaker.orca.pipeline.model.Stage;
import com.netflix.spinnaker.orca.pipeline.util.ArtifactResolver;
import com.netflix.spinnaker.orca.pipeline.util.ContextParameterProcessor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.yaml.snakeyaml.Yaml;
import retrofit.client.Response;

@Component
@Slf4j
public class DeployManifestTask extends AbstractCloudProviderAwareTask implements Task {
  @Autowired
  KatoService kato;

  @Autowired
  OortService oort;

  @Autowired
  ArtifactResolver artifactResolver;

  @Autowired
  ObjectMapper objectMapper;

  Yaml yamlParser = new Yaml();

  @Autowired
  ContextParameterProcessor contextParameterProcessor;

  RetrySupport retrySupport = new RetrySupport();

  public static final String TASK_NAME = "deployManifest";

  @Nonnull
  @Override
  public TaskResult execute(@Nonnull Stage stage) {
    String credentials = getCredentials(stage);
    String cloudProvider = getCloudProvider(stage);

    List<Artifact> artifacts = artifactResolver.getArtifacts(stage);
    Map task = new HashMap(stage.getContext());
    String artifactSource = (String) task.get("source");
    if (StringUtils.isNotEmpty(artifactSource) && artifactSource.equals("artifact")) {
      Artifact manifestArtifact = artifactResolver.getBoundArtifactForId(stage, task.get("manifestArtifactId").toString());

      if (manifestArtifact == null) {
        throw new IllegalArgumentException("No artifact could be bound to '" + task.get("manifestArtifactId") + "'");
      }

      log.info("Using {} as the manifest to be deployed", manifestArtifact);

      manifestArtifact.setArtifactAccount((String) task.get("manifestArtifactAccount"));
      Response manifestText = retrySupport.retry(() -> oort.fetchArtifact(manifestArtifact.getArtifactAccount(),
          manifestArtifact.getType(),
          manifestArtifact.getReference()
      ), 5, 1000, true);

      try {
        Map manifest = objectMapper.convertValue(yamlParser.load(manifestText.getBody().in()), Map.class);
        Map<String, Object> manifestWrapper = new HashMap<>();
        manifestWrapper.put("manifest", manifest);

        manifestWrapper = contextParameterProcessor.process(
            manifestWrapper,
          contextParameterProcessor.buildExecutionContext(stage, true),
          true
        );

        if (manifestWrapper.containsKey("expressionEvaluationSummary")) {
          throw new IllegalStateException("Failure evaluating manifest expressions: " + manifestWrapper.get("expressionEvaluationSummary"));
        }

        task.put("manifest", manifestWrapper.get("manifest"));
        task.put("source", "text");
      } catch (IOException e) {
        throw new IllegalArgumentException("Failed to read manifest from '" + manifestArtifact + "' as '" + manifestText + "': " + e.getMessage(), e);
      }
    }

    List<String> requiredArtifactIds = (List<String>) task.get("requiredArtifactIds");
    requiredArtifactIds = requiredArtifactIds == null ? new ArrayList<>() : requiredArtifactIds;
    List<Artifact> requiredArtifacts = requiredArtifactIds.stream()
        .map(id -> artifactResolver.getBoundArtifactForId(stage, id))
        .collect(Collectors.toList());

    log.info("Deploying {} artifacts within the provided manifest", requiredArtifacts);

    task.put("requiredArtifacts", requiredArtifacts);
    task.put("optionalArtifacts", artifacts);
    Map<String, Map> operation = new ImmutableMap.Builder<String, Map>()
        .put(TASK_NAME, task)
        .build();

    TaskId taskId = kato.requestOperations(cloudProvider, Collections.singletonList(operation)).toBlocking().first();

    Map<String, Object> outputs = new ImmutableMap.Builder<String, Object>()
        .put("kato.result.expected", true)
        .put("kato.last.task.id", taskId)
        .put("deploy.account.name", credentials)
        .build();

    return new TaskResult(ExecutionStatus.SUCCEEDED, outputs);
  }
}

@@ -658,125 +658,8 @@
 t;%0A%0A
-import java.io.IOException;%0Aimport java.util.*;%0Aimport java.util.stream.Collectors;%0Aimport javax.annotation.Nonnull;%0A
 impo
@@ -702,24 +702,24 @@
 jectMapper;%0A
+
 import com.g
@@ -1718,16 +1718,282 @@
 ponse;%0A%0A
+import javax.annotation.Nonnull;%0Aimport java.io.IOException;%0Aimport java.util.ArrayList;%0Aimport java.util.Collections;%0Aimport java.util.HashMap;%0Aimport java.util.List;%0Aimport java.util.Map;%0Aimport java.util.stream.Collectors;%0Aimport java.util.stream.StreamSupport;%0A%0A
 @Compone
@@ -3603,49 +3603,40 @@
     
-Map m
+Iterable%3CObject%3E rawM
 anifest
+s
  = 
-objectMapper.convertValue(
 yaml
@@ -3646,16 +3646,19 @@
 ser.load
+All
 (manifes
@@ -3682,22 +3682,203 @@
 n())
-, Map.class
+;%0A        List%3CMap%3E manifests = StreamSupport.stream(rawManifests.spliterator(), false)%0A            .map(m -%3E objectMapper.convertValue(m, Map.class))%0A            .collect(Collectors.toList()
 );%0A
+%0A
     
@@ -3965,32 +3965,33 @@
 er.put(%22manifest
+s
 %22, manifest);%0A%0A 
@@ -3985,16 +3985,17 @@
 manifest
+s
 );%0A%0A    
@@ -4431,16 +4431,17 @@
 manifest
+s
 %22, manif
@@ -4464,16 +4464,17 @@
 manifest
+s
 %22));%0A   

support multiple docs in one artifact (#1955)