// Copyright 2018 The Bazel Authors. All rights reserved.
//
// 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.google.devtools.build.lib.remote.util;

import static com.google.common.base.Strings.isNullOrEmpty;
import static com.google.common.base.Throwables.getStackTraceAsString;
import static java.util.stream.Collectors.joining;

import build.bazel.remote.execution.v2.Action;
import build.bazel.remote.execution.v2.ActionResult;
import build.bazel.remote.execution.v2.Digest;
import build.bazel.remote.execution.v2.Platform;
import com.google.common.base.Ascii;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.AsyncCallable;
import com.google.common.util.concurrent.FluentFuture;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.devtools.build.lib.actions.ActionInput;
import com.google.devtools.build.lib.actions.ExecutionRequirements;
import com.google.devtools.build.lib.actions.Spawn;
import com.google.devtools.build.lib.actions.SpawnMetrics;
import com.google.devtools.build.lib.actions.SpawnResult;
import com.google.devtools.build.lib.actions.Spawns;
import com.google.devtools.build.lib.authandtls.CallCredentialsProvider;
import com.google.devtools.build.lib.authandtls.credentialhelper.CredentialHelperException;
import com.google.devtools.build.lib.remote.ExecutionStatusException;
import com.google.devtools.build.lib.remote.common.BulkTransferException;
import com.google.devtools.build.lib.remote.common.CacheNotFoundException;
import com.google.devtools.build.lib.remote.common.OutputDigestMismatchException;
import com.google.devtools.build.lib.remote.common.RemoteCacheClient.ActionKey;
import com.google.devtools.build.lib.remote.options.RemoteOptions;
import com.google.devtools.build.lib.server.FailureDetails;
import com.google.devtools.build.lib.server.FailureDetails.FailureDetail;
import com.google.devtools.build.lib.vfs.PathFragment;
import com.google.protobuf.Any;
import com.google.protobuf.ByteString;
import com.google.protobuf.Duration;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.Timestamp;
import com.google.protobuf.util.Durations;
import com.google.rpc.BadRequest;
import com.google.rpc.Code;
import com.google.rpc.DebugInfo;
import com.google.rpc.Help;
import com.google.rpc.LocalizedMessage;
import com.google.rpc.PreconditionFailure;
import com.google.rpc.QuotaFailure;
import com.google.rpc.RequestInfo;
import com.google.rpc.ResourceInfo;
import com.google.rpc.RetryInfo;
import com.google.rpc.Status;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.text.DecimalFormat;
import java.text.DecimalFormatSymbols;
import java.time.Instant;
import java.util.Arrays;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.function.BiFunction;
import javax.annotation.Nullable;

/** Utility methods for the remote package. * */
public final class Utils {

  private Utils() {}

  /**
   * Returns the result of a {@link ListenableFuture} if successful, or throws any checked {@link
   * Exception} directly if it's an {@link IOException} or else wraps it in an {@link IOException}.
   *
   * <p>Cancel the future on {@link InterruptedException}
   */
  public static <T> T getFromFuture(ListenableFuture<T> f)
      throws IOException, InterruptedException {
    return getFromFuture(f, /* cancelOnInterrupt */ true);
  }

  /**
   * Returns the result of a {@link ListenableFuture} if successful, or throws any checked {@link
   * Exception} directly if it's an {@link IOException} or else wraps it in an {@link IOException}.
   *
   * @param cancelOnInterrupt cancel the future on {@link InterruptedException} if {@code true}.
   */
  public static <T> T getFromFuture(ListenableFuture<T> f, boolean cancelOnInterrupt)
      throws IOException, InterruptedException {
    try {
      return f.get();
    } catch (CancellationException e) {
      throw new InterruptedException();
    } catch (ExecutionException e) {
      Throwable cause = e.getCause();
      if (cause instanceof InterruptedException) {
        throw (InterruptedException) cause;
      }
      if (cause instanceof IOException) {
        throw (IOException) cause;
      }
      if (cause instanceof RuntimeException) {
        throw (RuntimeException) cause;
      }
      throw new IOException(cause);
    } catch (InterruptedException e) {
      if (cancelOnInterrupt) {
        f.cancel(true);
      }
      throw e;
    }
  }

  /**
   * Returns the (exec root relative) path of a spawn output that should be made available via
   * {@link SpawnResult#getInMemoryOutput(ActionInput)}.
   */
  @Nullable
  public static PathFragment getInMemoryOutputPath(Spawn spawn) {
    String outputPath =
        spawn.getExecutionInfo().get(ExecutionRequirements.REMOTE_EXECUTION_INLINE_OUTPUTS);
    if (outputPath != null) {
      return PathFragment.create(outputPath);
    }
    return null;
  }

  /** Constructs a {@link SpawnResult}. */
  public static SpawnResult createSpawnResult(
      DigestUtil digestUtil,
      ActionKey actionKey,
      int exitCode,
      boolean cacheHit,
      String runnerName,
      @Nullable InMemoryOutput inMemoryOutput,
      Timestamp executionStartTimestamp,
      Timestamp executionCompletedTimestamp,
      SpawnMetrics spawnMetrics,
      String mnemonic) {
    SpawnResult.Builder builder =
        new SpawnResult.Builder()
            .setStatus(
                exitCode == 0 ? SpawnResult.Status.SUCCESS : SpawnResult.Status.NON_ZERO_EXIT)
            .setExitCode(exitCode)
            .setRunnerName(cacheHit ? runnerName + " cache hit" : runnerName)
            .setCacheHit(cacheHit)
            .setStartTime(timestampToInstant(executionStartTimestamp))
            .setWallTimeInMs(
                (int)
                    java.time.Duration.between(
                            timestampToInstant(executionStartTimestamp),
                            timestampToInstant(executionCompletedTimestamp))
                        .toMillis())
            .setSpawnMetrics(spawnMetrics)
            .setRemote(true)
            .setDigest(digestUtil.asSpawnLogProto(actionKey));
    if (exitCode != 0) {
      builder.setFailureDetail(
          FailureDetail.newBuilder()
              .setMessage(mnemonic + " returned a non-zero exit code when running remotely")
              .setSpawn(
                  FailureDetails.Spawn.newBuilder()
                      .setCode(FailureDetails.Spawn.Code.NON_ZERO_EXIT))
              .build());
    }
    if (inMemoryOutput != null) {
      builder.setInMemoryOutput(inMemoryOutput.getOutput(), inMemoryOutput.getContents());
    }
    return builder.build();
  }

  public static Instant timestampToInstant(Timestamp timestamp) {
    return Instant.ofEpochSecond(timestamp.getSeconds(), timestamp.getNanos());
  }

  private static String statusName(int code) {
    // 'convert_underscores' to 'Convert Underscores'
    String name = Code.forNumber(code).getValueDescriptor().getName();
    return Arrays.stream(name.split("_"))
        .map(word -> Ascii.toUpperCase(word.substring(0, 1)) + Ascii.toLowerCase(word.substring(1)))
        .collect(joining(" "));
  }

  private static String errorDetailsMessage(Iterable<Any> details)
      throws InvalidProtocolBufferException {
    String messages = "";
    for (Any detail : details) {
      messages += "  " + errorDetailMessage(detail) + "\n";
    }
    return messages;
  }

  private static String durationMessage(Duration duration) {
    // this will give us seconds, might want to consider something nicer (graduating ms, s, m, h, d,
    // w?)
    return Durations.toString(duration);
  }

  private static String retryInfoMessage(RetryInfo retryInfo) {
    return "Retry delay recommendation of " + durationMessage(retryInfo.getRetryDelay());
  }

  private static String debugInfoMessage(DebugInfo debugInfo) {
    String message = "";
    if (debugInfo.getStackEntriesCount() > 0) {
      message +=
          "Debug Stack Information:\n  " + String.join("\n  ", debugInfo.getStackEntriesList());
    }
    if (!debugInfo.getDetail().isEmpty()) {
      if (debugInfo.getStackEntriesCount() > 0) {
        message += "\n";
      }
      message += "Debug Details: " + debugInfo.getDetail();
    }
    return message;
  }

  private static String quotaFailureMessage(QuotaFailure quotaFailure) {
    String message = "Quota Failure";
    if (quotaFailure.getViolationsCount() > 0) {
      message += ":";
    }
    for (QuotaFailure.Violation violation : quotaFailure.getViolationsList()) {
      message += "\n    " + violation.getSubject() + ": " + violation.getDescription();
    }
    return message;
  }

  private static String preconditionFailureMessage(PreconditionFailure preconditionFailure) {
    String message = "Precondition Failure";
    if (preconditionFailure.getViolationsCount() > 0) {
      message += ":";
    }
    for (PreconditionFailure.Violation violation : preconditionFailure.getViolationsList()) {
      message +=
          "\n    ("
              + violation.getType()
              + ") "
              + violation.getSubject()
              + ": "
              + violation.getDescription();
    }
    return message;
  }

  private static String badRequestMessage(BadRequest badRequest) {
    String message = "Bad Request";
    if (badRequest.getFieldViolationsCount() > 0) {
      message += ":";
    }
    for (BadRequest.FieldViolation fieldViolation : badRequest.getFieldViolationsList()) {
      message += "\n    " + fieldViolation.getField() + ": " + fieldViolation.getDescription();
    }
    return message;
  }

  private static String requestInfoMessage(RequestInfo requestInfo) {
    return "Request Info: " + requestInfo.getRequestId() + " => " + requestInfo.getServingData();
  }

  private static String resourceInfoMessage(ResourceInfo resourceInfo) {
    String message =
        "Resource Info: "
            + resourceInfo.getResourceType()
            + ": name='"
            + resourceInfo.getResourceName()
            + "', owner='"
            + resourceInfo.getOwner()
            + "'";
    if (!resourceInfo.getDescription().isEmpty()) {
      message += ", description: " + resourceInfo.getDescription();
    }
    return message;
  }

  private static String helpMessage(Help help) {
    String message = "Help";
    if (help.getLinksCount() > 0) {
      message += ":";
    }
    for (Help.Link link : help.getLinksList()) {
      message += "\n    " + link.getDescription() + ": " + link.getUrl();
    }
    return message;
  }

  private static String errorDetailMessage(Any detail) throws InvalidProtocolBufferException {
    if (detail.is(RetryInfo.class)) {
      return retryInfoMessage(detail.unpack(RetryInfo.class));
    }
    if (detail.is(DebugInfo.class)) {
      return debugInfoMessage(detail.unpack(DebugInfo.class));
    }
    if (detail.is(QuotaFailure.class)) {
      return quotaFailureMessage(detail.unpack(QuotaFailure.class));
    }
    if (detail.is(PreconditionFailure.class)) {
      return preconditionFailureMessage(detail.unpack(PreconditionFailure.class));
    }
    if (detail.is(BadRequest.class)) {
      return badRequestMessage(detail.unpack(BadRequest.class));
    }
    if (detail.is(RequestInfo.class)) {
      return requestInfoMessage(detail.unpack(RequestInfo.class));
    }
    if (detail.is(ResourceInfo.class)) {
      return resourceInfoMessage(detail.unpack(ResourceInfo.class));
    }
    if (detail.is(Help.class)) {
      return helpMessage(detail.unpack(Help.class));
    }
    return "Unrecognized error detail: " + detail;
  }

  private static String localizedStatusMessage(Status status)
      throws InvalidProtocolBufferException {
    String languageTag = Locale.getDefault().toLanguageTag();
    for (Any detail : status.getDetailsList()) {
      if (detail.is(LocalizedMessage.class)) {
        LocalizedMessage message = detail.unpack(LocalizedMessage.class);
        if (message.getLocale().equals(languageTag)) {
          return message.getMessage();
        }
      }
    }
    return status.getMessage();
  }

  private static String executionStatusExceptionErrorMessage(ExecutionStatusException e)
      throws InvalidProtocolBufferException {
    Status status = e.getOriginalStatus();
    return statusName(status.getCode())
        + ": "
        + localizedStatusMessage(status)
        + "\n"
        + errorDetailsMessage(status.getDetailsList());
  }

  private static String grpcAwareErrorMessage(IOException e) {
    io.grpc.Status errStatus = io.grpc.Status.fromThrowable(e);
    if (e.getCause() instanceof ExecutionStatusException) {
      // Display error message returned by the remote service.
      try {
        return "Remote Execution Failure:\n"
            + executionStatusExceptionErrorMessage((ExecutionStatusException) e.getCause());
      } catch (InvalidProtocolBufferException protoEx) {
        return "Error occurred attempting to format an error message for "
            + errStatus
            + ": "
            + Throwables.getStackTraceAsString(protoEx);
      }
    }
    if (!errStatus.getCode().equals(io.grpc.Status.UNKNOWN.getCode())) {
      // Display error message returned by the gRPC library, prefixed by the status code.
      StringBuilder sb = new StringBuilder();
      sb.append(errStatus.getCode().name());
      sb.append(": ");
      sb.append(errStatus.getDescription());
      // If the error originated from a credential helper, print additional debugging information.
      for (Throwable t = errStatus.getCause(); t != null; t = t.getCause()) {
        if (t instanceof CredentialHelperException) {
          sb.append(": ");
          sb.append(t.getMessage());
          break;
        }
      }
      return sb.toString();
    }
    return e.getMessage();
  }

  public static String grpcAwareErrorMessage(Throwable error, boolean verboseFailures) {
    String errorMessage;
    if (error instanceof IOException) {
      errorMessage = grpcAwareErrorMessage((IOException) error);
    } else {
      errorMessage = error.getMessage();
    }

    if (isNullOrEmpty(errorMessage)) {
      errorMessage = error.getClass().getSimpleName();
    }

    if (verboseFailures) {
      // On --verbose_failures print the whole stack trace
      errorMessage += "\n" + getStackTraceAsString(error);
    }

    return errorMessage;
  }

  @SuppressWarnings("ProtoParseWithRegistry")
  public static ListenableFuture<ActionResult> downloadAsActionResult(
      ActionKey actionDigest,
      BiFunction<Digest, OutputStream, ListenableFuture<Void>> downloadFunction) {
    ByteArrayOutputStream data = new ByteArrayOutputStream(/* size= */ 1024);
    ListenableFuture<Void> download = downloadFunction.apply(actionDigest.getDigest(), data);
    return FluentFuture.from(download)
        .transformAsync(
            (v) -> {
              try {
                return Futures.immediateFuture(ActionResult.parseFrom(data.toByteArray()));
              } catch (InvalidProtocolBufferException e) {
                return Futures.immediateFailedFuture(e);
              }
            },
            MoreExecutors.directExecutor())
        .catching(CacheNotFoundException.class, (e) -> null, MoreExecutors.directExecutor());
  }

  public static void verifyBlobContents(Digest expected, Digest actual) throws IOException {
    if (!expected.equals(actual)) {
      throw new OutputDigestMismatchException(expected, actual);
    }
  }

  public static Action buildAction(
      Digest command,
      Digest inputRoot,
      @Nullable Platform platform,
      java.time.Duration timeout,
      boolean cacheable,
      @Nullable ByteString salt) {
    Action.Builder action = Action.newBuilder();
    action.setCommandDigest(command);
    action.setInputRootDigest(inputRoot);
    if (!timeout.isZero()) {
      action.setTimeout(Duration.newBuilder().setSeconds(timeout.getSeconds()));
    }
    if (!cacheable) {
      action.setDoNotCache(true);
    }
    if (platform != null) {
      action.setPlatform(platform);
    }
    if (salt != null) {
      action.setSalt(salt);
    }
    return action.build();
  }

  /** An in-memory output file. */
  public static final class InMemoryOutput {
    private final ActionInput output;
    private final ByteString contents;

    public InMemoryOutput(ActionInput output, ByteString contents) {
      this.output = output;
      this.contents = contents;
    }

    public ActionInput getOutput() {
      return output;
    }

    public ByteString getContents() {
      return contents;
    }
  }

  /**
   * Call an asynchronous code block. If the block throws unauthenticated error, refresh the
   * credentials using {@link CallCredentialsProvider} and call it again.
   *
   * <p>If any other exception thrown by the code block, it will be caught and wrapped in the
   * returned {@link ListenableFuture}.
   */
  public static <V> ListenableFuture<V> refreshIfUnauthenticatedAsync(
      AsyncCallable<V> call, CallCredentialsProvider callCredentialsProvider) {
    Preconditions.checkNotNull(call);
    Preconditions.checkNotNull(callCredentialsProvider);

    try {
      return Futures.catchingAsync(
          call.call(),
          Throwable.class,
          (e) -> refreshIfUnauthenticatedAsyncOnException(e, call, callCredentialsProvider),
          MoreExecutors.directExecutor());
    } catch (Throwable t) {
      return refreshIfUnauthenticatedAsyncOnException(t, call, callCredentialsProvider);
    }
  }

  private static <V> ListenableFuture<V> refreshIfUnauthenticatedAsyncOnException(
      Throwable t, AsyncCallable<V> call, CallCredentialsProvider callCredentialsProvider) {
    io.grpc.Status status = io.grpc.Status.fromThrowable(t);
    if (status != null
        && (status.getCode() == io.grpc.Status.Code.UNAUTHENTICATED
            || status.getCode() == io.grpc.Status.Code.PERMISSION_DENIED)) {
      try {
        callCredentialsProvider.refresh();
        return call.call();
      } catch (Throwable tt) {
        t.addSuppressed(tt);
      }
    }

    return Futures.immediateFailedFuture(t);
  }

  /** Same as {@link #refreshIfUnauthenticatedAsync} but calling a synchronous code block. */
  public static <V> V refreshIfUnauthenticated(
      Callable<V> call, CallCredentialsProvider callCredentialsProvider)
      throws IOException, InterruptedException {
    Preconditions.checkNotNull(call);
    Preconditions.checkNotNull(callCredentialsProvider);

    try {
      return call.call();
    } catch (Exception e) {
      io.grpc.Status status = io.grpc.Status.fromThrowable(e);
      if (status != null
          && (status.getCode() == io.grpc.Status.Code.UNAUTHENTICATED
              || status.getCode() == io.grpc.Status.Code.PERMISSION_DENIED)) {
        try {
          callCredentialsProvider.refresh();
          return call.call();
        } catch (Exception ex) {
          e.addSuppressed(ex);
        }
      }

      Throwables.throwIfInstanceOf(e, IOException.class);
      Throwables.throwIfInstanceOf(e, InterruptedException.class);
      Throwables.throwIfUnchecked(e);
      throw new AssertionError(e);
    }
  }

  private static final ImmutableList<String> UNITS = ImmutableList.of("KiB", "MiB", "GiB", "TiB");
  // Format as single digit decimal number.
  private static final DecimalFormat BYTE_COUNT_FORMAT =
      new DecimalFormat("0.0", new DecimalFormatSymbols(Locale.US));

  /**
   * Converts the number of bytes to a human readable string, e.g. 1024 -> 1 KiB.
   *
   * <p>Negative numbers are not allowed.
   */
  public static String bytesCountToDisplayString(long bytes) {
    Preconditions.checkArgument(bytes >= 0);

    if (bytes < 1024) {
      return bytes + " B";
    }

    int unitIndex = 0;
    long value = bytes;
    while ((unitIndex + 1) < UNITS.size() && value >= (1 << 20)) {
      value >>= 10;
      unitIndex++;
    }

    return String.format("%s %s", BYTE_COUNT_FORMAT.format(value / 1024.0), UNITS.get(unitIndex));
  }

  public static boolean shouldUploadLocalResultsToRemoteCache(
      RemoteOptions remoteOptions, Map<String, String> executionInfo) {
    return remoteOptions.remoteUploadLocalResults
        && Spawns.mayBeCachedRemotely(executionInfo)
        && !executionInfo.containsKey(ExecutionRequirements.NO_REMOTE_CACHE_UPLOAD);
  }

  public static void waitForBulkTransfer(
      Iterable<? extends ListenableFuture<?>> transfers, boolean cancelRemainingOnInterrupt)
      throws BulkTransferException, InterruptedException {
    BulkTransferException bulkTransferException = null;
    InterruptedException interruptedException = null;
    boolean interrupted = Thread.currentThread().isInterrupted();
    for (ListenableFuture<?> transfer : transfers) {
      try {
        if (interruptedException == null) {
          // Wait for all transfers to finish.
          getFromFuture(transfer, cancelRemainingOnInterrupt);
        } else {
          transfer.cancel(true);
        }
      } catch (IOException e) {
        if (bulkTransferException == null) {
          bulkTransferException = new BulkTransferException();
        }
        bulkTransferException.add(e);
      } catch (InterruptedException e) {
        interrupted = Thread.interrupted() || interrupted;
        interruptedException = e;
        if (!cancelRemainingOnInterrupt) {
          // leave the rest of the transfers alone
          break;
        }
      }
    }
    if (interrupted) {
      Thread.currentThread().interrupt();
    }
    if (interruptedException != null) {
      if (bulkTransferException != null) {
        interruptedException.addSuppressed(bulkTransferException);
      }
      throw interruptedException;
    }
    if (bulkTransferException != null) {
      throw bulkTransferException;
    }
  }
}