/*
 *
 * Copyright 2017, Google Inc.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are
 * met:
 *
 *     * Redistributions of source code must retain the above copyright
 * notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above
 * copyright notice, this list of conditions and the following disclaimer
 * in the documentation and/or other materials provided with the
 * distribution.
 *     * Neither the name of Google Inc. nor the names of its
 * contributors may be used to endorse or promote products derived from
 * this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 */

#include "src/core/ext/filters/max_age/max_age_filter.h"

#include <limits.h>
#include <string.h>

#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/channel/channel_stack_builder.h"
#include "src/core/lib/iomgr/timer.h"
#include "src/core/lib/surface/channel_init.h"
#include "src/core/lib/transport/http2_errors.h"
#include "src/core/lib/transport/service_config.h"

#define DEFAULT_MAX_CONNECTION_AGE_MS INT_MAX
#define DEFAULT_MAX_CONNECTION_AGE_GRACE_MS INT_MAX
#define DEFAULT_MAX_CONNECTION_IDLE_MS INT_MAX

typedef struct channel_data {
  /* We take a reference to the channel stack for the timer callback */
  grpc_channel_stack* channel_stack;
  /* Guards access to max_age_timer, max_age_timer_pending, max_age_grace_timer
     and max_age_grace_timer_pending */
  gpr_mu max_age_timer_mu;
  /* True if the max_age timer callback is currently pending */
  bool max_age_timer_pending;
  /* True if the max_age_grace timer callback is currently pending */
  bool max_age_grace_timer_pending;
  /* The timer for checking if the channel has reached its max age */
  grpc_timer max_age_timer;
  /* The timer for checking if the max-aged channel has uesed up the grace
     period */
  grpc_timer max_age_grace_timer;
  /* The timer for checking if the channel's idle duration reaches
     max_connection_idle */
  grpc_timer max_idle_timer;
  /* Allowed max time a channel may have no outstanding rpcs */
  gpr_timespec max_connection_idle;
  /* Allowed max time a channel may exist */
  gpr_timespec max_connection_age;
  /* Allowed grace period after the channel reaches its max age */
  gpr_timespec max_connection_age_grace;
  /* Closure to run when the channel's idle duration reaches max_connection_idle
     and should be closed gracefully */
  grpc_closure close_max_idle_channel;
  /* Closure to run when the channel reaches its max age and should be closed
     gracefully */
  grpc_closure close_max_age_channel;
  /* Closure to run the channel uses up its max age grace time and should be
     closed forcibly */
  grpc_closure force_close_max_age_channel;
  /* Closure to run when the init fo channel stack is done and the max_idle
     timer should be started */
  grpc_closure start_max_idle_timer_after_init;
  /* Closure to run when the init fo channel stack is done and the max_age timer
     should be started */
  grpc_closure start_max_age_timer_after_init;
  /* Closure to run when the goaway op is finished and the max_age_timer */
  grpc_closure start_max_age_grace_timer_after_goaway_op;
  /* Closure to run when the channel connectivity state changes */
  grpc_closure channel_connectivity_changed;
  /* Records the current connectivity state */
  grpc_connectivity_state connectivity_state;
  /* Number of active calls */
  gpr_atm call_count;
} channel_data;

/* Increase the nubmer of active calls. Before the increasement, if there are no
   calls, the max_idle_timer should be cancelled. */
static void increase_call_count(grpc_exec_ctx* exec_ctx, channel_data* chand) {
  if (gpr_atm_full_fetch_add(&chand->call_count, 1) == 0) {
    grpc_timer_cancel(exec_ctx, &chand->max_idle_timer);
  }
}

/* Decrease the nubmer of active calls. After the decrement, if there are no
   calls, the max_idle_timer should be started. */
static void decrease_call_count(grpc_exec_ctx* exec_ctx, channel_data* chand) {
  if (gpr_atm_full_fetch_add(&chand->call_count, -1) == 1) {
    GRPC_CHANNEL_STACK_REF(chand->channel_stack, "max_age max_idle_timer");
    grpc_timer_init(
        exec_ctx, &chand->max_idle_timer,
        gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), chand->max_connection_idle),
        &chand->close_max_idle_channel, gpr_now(GPR_CLOCK_MONOTONIC));
  }
}

static void start_max_idle_timer_after_init(grpc_exec_ctx* exec_ctx, void* arg,
                                            grpc_error* error) {
  channel_data* chand = arg;
  /* Decrease call_count. If there are no active calls at this time,
     max_idle_timer will start here. If the number of active calls is not 0,
     max_idle_timer will start after all the active calls end. */
  decrease_call_count(exec_ctx, chand);
  GRPC_CHANNEL_STACK_UNREF(exec_ctx, chand->channel_stack,
                           "max_age start_max_idle_timer_after_init");
}

static void start_max_age_timer_after_init(grpc_exec_ctx* exec_ctx, void* arg,
                                           grpc_error* error) {
  channel_data* chand = arg;
  gpr_mu_lock(&chand->max_age_timer_mu);
  chand->max_age_timer_pending = true;
  GRPC_CHANNEL_STACK_REF(chand->channel_stack, "max_age max_age_timer");
  grpc_timer_init(
      exec_ctx, &chand->max_age_timer,
      gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), chand->max_connection_age),
      &chand->close_max_age_channel, gpr_now(GPR_CLOCK_MONOTONIC));
  gpr_mu_unlock(&chand->max_age_timer_mu);
  grpc_transport_op* op = grpc_make_transport_op(NULL);
  op->on_connectivity_state_change = &chand->channel_connectivity_changed,
  op->connectivity_state = &chand->connectivity_state;
  grpc_channel_next_op(exec_ctx,
                       grpc_channel_stack_element(chand->channel_stack, 0), op);
  GRPC_CHANNEL_STACK_UNREF(exec_ctx, chand->channel_stack,
                           "max_age start_max_age_timer_after_init");
}

static void start_max_age_grace_timer_after_goaway_op(grpc_exec_ctx* exec_ctx,
                                                      void* arg,
                                                      grpc_error* error) {
  channel_data* chand = arg;
  gpr_mu_lock(&chand->max_age_timer_mu);
  chand->max_age_grace_timer_pending = true;
  GRPC_CHANNEL_STACK_REF(chand->channel_stack, "max_age max_age_grace_timer");
  grpc_timer_init(exec_ctx, &chand->max_age_grace_timer,
                  gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
                               chand->max_connection_age_grace),
                  &chand->force_close_max_age_channel,
                  gpr_now(GPR_CLOCK_MONOTONIC));
  gpr_mu_unlock(&chand->max_age_timer_mu);
  GRPC_CHANNEL_STACK_UNREF(exec_ctx, chand->channel_stack,
                           "max_age start_max_age_grace_timer_after_goaway_op");
}

static void close_max_idle_channel(grpc_exec_ctx* exec_ctx, void* arg,
                                   grpc_error* error) {
  channel_data* chand = arg;
  gpr_atm_no_barrier_fetch_add(&chand->call_count, 1);
  if (error == GRPC_ERROR_NONE) {
    grpc_transport_op* op = grpc_make_transport_op(NULL);
    op->goaway_error =
        grpc_error_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING("max_idle"),
                           GRPC_ERROR_INT_HTTP2_ERROR, GRPC_HTTP2_NO_ERROR);
    grpc_channel_element* elem =
        grpc_channel_stack_element(chand->channel_stack, 0);
    elem->filter->start_transport_op(exec_ctx, elem, op);
  } else if (error != GRPC_ERROR_CANCELLED) {
    GRPC_LOG_IF_ERROR("close_max_idle_channel", error);
  }
  GRPC_CHANNEL_STACK_UNREF(exec_ctx, chand->channel_stack,
                           "max_age max_idle_timer");
}

static void close_max_age_channel(grpc_exec_ctx* exec_ctx, void* arg,
                                  grpc_error* error) {
  channel_data* chand = arg;
  gpr_mu_lock(&chand->max_age_timer_mu);
  chand->max_age_timer_pending = false;
  gpr_mu_unlock(&chand->max_age_timer_mu);
  if (error == GRPC_ERROR_NONE) {
    GRPC_CHANNEL_STACK_REF(chand->channel_stack,
                           "max_age start_max_age_grace_timer_after_goaway_op");
    grpc_transport_op* op = grpc_make_transport_op(
        &chand->start_max_age_grace_timer_after_goaway_op);
    op->goaway_error =
        grpc_error_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING("max_age"),
                           GRPC_ERROR_INT_HTTP2_ERROR, GRPC_HTTP2_NO_ERROR);
    grpc_channel_element* elem =
        grpc_channel_stack_element(chand->channel_stack, 0);
    elem->filter->start_transport_op(exec_ctx, elem, op);
  } else if (error != GRPC_ERROR_CANCELLED) {
    GRPC_LOG_IF_ERROR("close_max_age_channel", error);
  }
  GRPC_CHANNEL_STACK_UNREF(exec_ctx, chand->channel_stack,
                           "max_age max_age_timer");
}

static void force_close_max_age_channel(grpc_exec_ctx* exec_ctx, void* arg,
                                        grpc_error* error) {
  channel_data* chand = arg;
  gpr_mu_lock(&chand->max_age_timer_mu);
  chand->max_age_grace_timer_pending = false;
  gpr_mu_unlock(&chand->max_age_timer_mu);
  if (error == GRPC_ERROR_NONE) {
    grpc_transport_op* op = grpc_make_transport_op(NULL);
    op->disconnect_with_error =
        GRPC_ERROR_CREATE_FROM_STATIC_STRING("Channel reaches max age");
    grpc_channel_element* elem =
        grpc_channel_stack_element(chand->channel_stack, 0);
    elem->filter->start_transport_op(exec_ctx, elem, op);
  } else if (error != GRPC_ERROR_CANCELLED) {
    GRPC_LOG_IF_ERROR("force_close_max_age_channel", error);
  }
  GRPC_CHANNEL_STACK_UNREF(exec_ctx, chand->channel_stack,
                           "max_age max_age_grace_timer");
}

static void channel_connectivity_changed(grpc_exec_ctx* exec_ctx, void* arg,
                                         grpc_error* error) {
  channel_data* chand = arg;
  if (chand->connectivity_state != GRPC_CHANNEL_SHUTDOWN) {
    grpc_transport_op* op = grpc_make_transport_op(NULL);
    op->on_connectivity_state_change = &chand->channel_connectivity_changed,
    op->connectivity_state = &chand->connectivity_state;
    grpc_channel_next_op(
        exec_ctx, grpc_channel_stack_element(chand->channel_stack, 0), op);
  } else {
    gpr_mu_lock(&chand->max_age_timer_mu);
    if (chand->max_age_timer_pending) {
      grpc_timer_cancel(exec_ctx, &chand->max_age_timer);
      chand->max_age_timer_pending = false;
    }
    if (chand->max_age_grace_timer_pending) {
      grpc_timer_cancel(exec_ctx, &chand->max_age_grace_timer);
      chand->max_age_grace_timer_pending = false;
    }
    gpr_mu_unlock(&chand->max_age_timer_mu);
    /* If there are no active calls, this increasement will cancel
       max_idle_timer, and prevent max_idle_timer from being started in the
       future. */
    increase_call_count(exec_ctx, chand);
  }
}

/* Constructor for call_data. */
static grpc_error* init_call_elem(grpc_exec_ctx* exec_ctx,
                                  grpc_call_element* elem,
                                  const grpc_call_element_args* args) {
  channel_data* chand = elem->channel_data;
  increase_call_count(exec_ctx, chand);
  return GRPC_ERROR_NONE;
}

/* Destructor for call_data. */
static void destroy_call_elem(grpc_exec_ctx* exec_ctx, grpc_call_element* elem,
                              const grpc_call_final_info* final_info,
                              grpc_closure* ignored) {
  channel_data* chand = elem->channel_data;
  decrease_call_count(exec_ctx, chand);
}

/* Constructor for channel_data. */
static grpc_error* init_channel_elem(grpc_exec_ctx* exec_ctx,
                                     grpc_channel_element* elem,
                                     grpc_channel_element_args* args) {
  channel_data* chand = elem->channel_data;
  gpr_mu_init(&chand->max_age_timer_mu);
  chand->max_age_timer_pending = false;
  chand->max_age_grace_timer_pending = false;
  chand->channel_stack = args->channel_stack;
  chand->max_connection_age =
      DEFAULT_MAX_CONNECTION_AGE_MS == INT_MAX
          ? gpr_inf_future(GPR_TIMESPAN)
          : gpr_time_from_millis(DEFAULT_MAX_CONNECTION_AGE_MS, GPR_TIMESPAN);
  chand->max_connection_age_grace =
      DEFAULT_MAX_CONNECTION_AGE_GRACE_MS == INT_MAX
          ? gpr_inf_future(GPR_TIMESPAN)
          : gpr_time_from_millis(DEFAULT_MAX_CONNECTION_AGE_GRACE_MS,
                                 GPR_TIMESPAN);
  chand->max_connection_idle =
      DEFAULT_MAX_CONNECTION_IDLE_MS == INT_MAX
          ? gpr_inf_future(GPR_TIMESPAN)
          : gpr_time_from_millis(DEFAULT_MAX_CONNECTION_IDLE_MS, GPR_TIMESPAN);
  for (size_t i = 0; i < args->channel_args->num_args; ++i) {
    if (0 == strcmp(args->channel_args->args[i].key,
                    GRPC_ARG_MAX_CONNECTION_AGE_MS)) {
      const int value = grpc_channel_arg_get_integer(
          &args->channel_args->args[i],
          (grpc_integer_options){DEFAULT_MAX_CONNECTION_AGE_MS, 1, INT_MAX});
      chand->max_connection_age =
          value == INT_MAX ? gpr_inf_future(GPR_TIMESPAN)
                           : gpr_time_from_millis(value, GPR_TIMESPAN);
    } else if (0 == strcmp(args->channel_args->args[i].key,
                           GRPC_ARG_MAX_CONNECTION_AGE_GRACE_MS)) {
      const int value = grpc_channel_arg_get_integer(
          &args->channel_args->args[i],
          (grpc_integer_options){DEFAULT_MAX_CONNECTION_AGE_GRACE_MS, 0,
                                 INT_MAX});
      chand->max_connection_age_grace =
          value == INT_MAX ? gpr_inf_future(GPR_TIMESPAN)
                           : gpr_time_from_millis(value, GPR_TIMESPAN);
    } else if (0 == strcmp(args->channel_args->args[i].key,
                           GRPC_ARG_MAX_CONNECTION_IDLE_MS)) {
      const int value = grpc_channel_arg_get_integer(
          &args->channel_args->args[i],
          (grpc_integer_options){DEFAULT_MAX_CONNECTION_IDLE_MS, 1, INT_MAX});
      chand->max_connection_idle =
          value == INT_MAX ? gpr_inf_future(GPR_TIMESPAN)
                           : gpr_time_from_millis(value, GPR_TIMESPAN);
    }
  }
  grpc_closure_init(&chand->close_max_idle_channel, close_max_idle_channel,
                    chand, grpc_schedule_on_exec_ctx);
  grpc_closure_init(&chand->close_max_age_channel, close_max_age_channel, chand,
                    grpc_schedule_on_exec_ctx);
  grpc_closure_init(&chand->force_close_max_age_channel,
                    force_close_max_age_channel, chand,
                    grpc_schedule_on_exec_ctx);
  grpc_closure_init(&chand->start_max_idle_timer_after_init,
                    start_max_idle_timer_after_init, chand,
                    grpc_schedule_on_exec_ctx);
  grpc_closure_init(&chand->start_max_age_timer_after_init,
                    start_max_age_timer_after_init, chand,
                    grpc_schedule_on_exec_ctx);
  grpc_closure_init(&chand->start_max_age_grace_timer_after_goaway_op,
                    start_max_age_grace_timer_after_goaway_op, chand,
                    grpc_schedule_on_exec_ctx);
  grpc_closure_init(&chand->channel_connectivity_changed,
                    channel_connectivity_changed, chand,
                    grpc_schedule_on_exec_ctx);

  if (gpr_time_cmp(chand->max_connection_age, gpr_inf_future(GPR_TIMESPAN)) !=
      0) {
    /* When the channel reaches its max age, we send down an op with
       goaway_error set.  However, we can't send down any ops until after the
       channel stack is fully initialized.  If we start the timer here, we have
       no guarantee that the timer won't pop before channel stack initialization
       is finished.  To avoid that problem, we create a closure to start the
       timer, and we schedule that closure to be run after call stack
       initialization is done. */
    GRPC_CHANNEL_STACK_REF(chand->channel_stack,
                           "max_age start_max_age_timer_after_init");
    grpc_closure_sched(exec_ctx, &chand->start_max_age_timer_after_init,
                       GRPC_ERROR_NONE);
  }

  /* Initialize the number of calls as 1, so that the max_idle_timer will not
     start until start_max_idle_timer_after_init is invoked. */
  gpr_atm_rel_store(&chand->call_count, 1);
  if (gpr_time_cmp(chand->max_connection_idle, gpr_inf_future(GPR_TIMESPAN)) !=
      0) {
    GRPC_CHANNEL_STACK_REF(chand->channel_stack,
                           "max_age start_max_idle_timer_after_init");
    grpc_closure_sched(exec_ctx, &chand->start_max_idle_timer_after_init,
                       GRPC_ERROR_NONE);
  }
  return GRPC_ERROR_NONE;
}

/* Destructor for channel_data. */
static void destroy_channel_elem(grpc_exec_ctx* exec_ctx,
                                 grpc_channel_element* elem) {}

const grpc_channel_filter grpc_max_age_filter = {
    grpc_call_next_op,
    grpc_channel_next_op,
    0, /* sizeof_call_data */
    init_call_elem,
    grpc_call_stack_ignore_set_pollset_or_pollset_set,
    destroy_call_elem,
    sizeof(channel_data),
    init_channel_elem,
    destroy_channel_elem,
    grpc_call_next_get_peer,
    grpc_channel_next_get_info,
    "max_age"};

static bool maybe_add_max_age_filter(grpc_exec_ctx* exec_ctx,
                                     grpc_channel_stack_builder* builder,
                                     void* arg) {
  const grpc_channel_args* channel_args =
      grpc_channel_stack_builder_get_channel_arguments(builder);
  const grpc_arg* a =
      grpc_channel_args_find(channel_args, GRPC_ARG_MAX_CONNECTION_AGE_MS);
  bool enable = false;
  if (a != NULL && a->type == GRPC_ARG_INTEGER && a->value.integer != INT_MAX) {
    enable = true;
  }
  a = grpc_channel_args_find(channel_args, GRPC_ARG_MAX_CONNECTION_IDLE_MS);
  if (a != NULL && a->type == GRPC_ARG_INTEGER && a->value.integer != INT_MAX) {
    enable = true;
  }
  if (enable) {
    return grpc_channel_stack_builder_prepend_filter(
        builder, &grpc_max_age_filter, NULL, NULL);
  } else {
    return true;
  }
}

void grpc_max_age_filter_init(void) {
  grpc_channel_init_register_stage(GRPC_SERVER_CHANNEL,
                                   GRPC_CHANNEL_INIT_BUILTIN_PRIORITY,
                                   maybe_add_max_age_filter, NULL);
}

void grpc_max_age_filter_shutdown(void) {}