// Copyright 2020 Google LLC
//
// 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
//
//     https://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.

#ifndef ANALYTICS_ANALYTICS_RECORDER_H_
#define ANALYTICS_ANALYTICS_RECORDER_H_

#include <string>

#include "absl/container/btree_map.h"
#include "absl/time/time.h"
#include "core/event_logger.h"
#include "core/payload.h"
#include "core/strategy.h"
#include "platform/public/mutex.h"
#include "platform/public/single_thread_executor.h"
#include "proto/analytics/connections_log.proto.h"
#include "proto/connections_enums.proto.h"

namespace location {
namespace nearby {
namespace analytics {

class AnalyticsRecorder {
 public:
  static constexpr absl::string_view kVersion = "v1.0.0";

  explicit AnalyticsRecorder(EventLogger *event_logger);
  virtual ~AnalyticsRecorder();

  // Advertising phase
  void OnStartAdvertising(
      connections::Strategy strategy,
      const std::vector<location::nearby::proto::connections::Medium> &mediums)
      ABSL_LOCKS_EXCLUDED(mutex_);
  void OnStopAdvertising() ABSL_LOCKS_EXCLUDED(mutex_);

  // Discovery phase
  void OnStartDiscovery(
      connections::Strategy strategy,
      const std::vector<location::nearby::proto::connections::Medium> &mediums)
      ABSL_LOCKS_EXCLUDED(mutex_);
  void OnStopDiscovery() ABSL_LOCKS_EXCLUDED(mutex_);
  void OnEndpointFound(location::nearby::proto::connections::Medium medium)
      ABSL_LOCKS_EXCLUDED(mutex_);

  // Connection request
  void OnConnectionRequestReceived(const std::string &remote_endpoint_id)
      ABSL_LOCKS_EXCLUDED(mutex_);
  void OnConnectionRequestSent(const std::string &remote_endpoint_id)
      ABSL_LOCKS_EXCLUDED(mutex_);
  void OnRemoteEndpointAccepted(const std::string &remote_endpoint_id)
      ABSL_LOCKS_EXCLUDED(mutex_);
  void OnLocalEndpointAccepted(const std::string &remote_endpoint_id)
      ABSL_LOCKS_EXCLUDED(mutex_);
  void OnRemoteEndpointRejected(const std::string &remote_endpoint_id)
      ABSL_LOCKS_EXCLUDED(mutex_);
  void OnLocalEndpointRejected(const std::string &remote_endpoint_id)
      ABSL_LOCKS_EXCLUDED(mutex_);

  // Connection attempt
  void OnIncomingConnectionAttempt(
      location::nearby::proto::connections::ConnectionAttemptType type,
      location::nearby::proto::connections::Medium medium,
      location::nearby::proto::connections::ConnectionAttemptResult result,
      absl::Duration duration, const std::string &connection_token)
      ABSL_LOCKS_EXCLUDED(mutex_);
  void OnOutgoingConnectionAttempt(
      const std::string &remote_endpoint_id,
      location::nearby::proto::connections::ConnectionAttemptType type,
      location::nearby::proto::connections::Medium medium,
      location::nearby::proto::connections::ConnectionAttemptResult result,
      absl::Duration duration, const std::string &connection_token)
      ABSL_LOCKS_EXCLUDED(mutex_);

  // Connection established
  void OnConnectionEstablished(
      const std::string &endpoint_id,
      location::nearby::proto::connections::Medium medium,
      const std::string &connection_token) ABSL_LOCKS_EXCLUDED(mutex_);
  void OnConnectionClosed(
      const std::string &endpoint_id,
      location::nearby::proto::connections::Medium medium,
      location::nearby::proto::connections ::DisconnectionReason reason)
      ABSL_LOCKS_EXCLUDED(mutex_);

  // Invokes event_logger_.Log() at the end of life of client. Log action is
  // called in a separate thread to allow synchronous potentially lengthy
  // execution.
  void LogSession() ABSL_LOCKS_EXCLUDED(mutex_);

 private:
  // Tracks the chunks and duration of a Payload on a particular medium.
  class PendingPayload {
   public:
    PendingPayload(location::nearby::proto::connections::PayloadType type,
                   std::int64_t total_size_bytes)
        : start_time_(SystemClock::ElapsedRealtime()),
          type_(type),
          total_size_bytes_(total_size_bytes),
          num_bytes_transferred_(0),
          num_chunks_(0) {}
    ~PendingPayload() = default;

    void AddChunk(std::int64_t chunk_size_bytes);

    proto::ConnectionsLog::Payload GetProtoPayload(
        location::nearby::proto::connections::PayloadStatus status);

    location::nearby::proto::connections::PayloadType type() const {
      return type_;
    }

    std::int64_t total_size_bytes() const { return total_size_bytes_; }

   private:
    absl::Time start_time_;
    location::nearby::proto::connections::PayloadType type_;
    std::int64_t total_size_bytes_;
    std::int64_t num_bytes_transferred_;
    int num_chunks_;
  };

  class LogicalConnection {
   public:
    LogicalConnection(
        location::nearby::proto::connections::Medium initial_medium,
        const std::string &connection_token) {
      PhysicalConnectionEstablished(initial_medium, connection_token);
    }
    LogicalConnection(const LogicalConnection &) = delete;
    LogicalConnection(LogicalConnection &&other)
        : current_medium_(std::move(other.current_medium_)),
          physical_connections_(std::move(other.physical_connections_)),
          incoming_payloads_(std::move(other.incoming_payloads_)),
          outgoing_payloads_(std::move(other.outgoing_payloads_)) {}
    LogicalConnection &operator=(const LogicalConnection &) = delete;
    LogicalConnection &&operator=(LogicalConnection &&) = delete;
    ~LogicalConnection() = default;

    void PhysicalConnectionEstablished(
        location::nearby::proto::connections::Medium medium,
        const std::string &connection_token);
    void PhysicalConnectionClosed(
        location::nearby::proto::connections::Medium medium,
        location::nearby::proto::connections::DisconnectionReason reason);
    void CloseAllPhysicalConnections();

    void IncomingPayloadStarted(
        std::int64_t payload_id,
        location::nearby::proto::connections::PayloadType type,
        std::int64_t total_size_bytes);
    void ChunkReceived(std::int64_t payload_id, std::int64_t size_bytes);
    void IncomingPayloadDone(
        std::int64_t payload_id,
        location::nearby::proto::connections::PayloadStatus status);
    void OutgoingPayloadStarted(
        std::int64_t payload_id,
        location::nearby::proto::connections::PayloadType type,
        std::int64_t total_size_bytes);
    void ChunkSent(std::int64_t payload_id, std::int64_t size_bytes);
    void OutgoingPayloadDone(
        std::int64_t payload_id,
        location::nearby::proto::connections::PayloadStatus status);

    std::vector<proto::ConnectionsLog::EstablishedConnection>
    GetEstablisedConnections();

   private:
    void FinishPhysicalConnection(
        proto::ConnectionsLog::EstablishedConnection *established_connection,
        location::nearby::proto::connections::DisconnectionReason reason);
    std::vector<proto::ConnectionsLog::Payload> ResolvePendingPayloads(
        absl::btree_map<std::int64_t, std::unique_ptr<PendingPayload>>
            &pending_payloads,
        location::nearby::proto::connections::DisconnectionReason reason);

    location::nearby::proto::connections::Medium current_medium_ =
        location::nearby::proto::connections::UNKNOWN_MEDIUM;
    absl::btree_map<
        location::nearby::proto::connections::Medium,
        std::unique_ptr<proto::ConnectionsLog::EstablishedConnection>>
        physical_connections_;
    absl::btree_map<std::int64_t, std::unique_ptr<PendingPayload>>
        incoming_payloads_;
    absl::btree_map<std::int64_t, std::unique_ptr<PendingPayload>>
        outgoing_payloads_;
  };

  bool CanRecordAnalyticsLocked(const std::string &method_name)
      ABSL_SHARED_LOCKS_REQUIRED(mutex_);

  // Callbacks the ConnectionsLog proto byte array data to the EventLogger with
  // ClientSession sub-proto.
  void LogClientSession();
  // Callbacks the ConnectionsLog proto byte array data to the EventLogger.
  void LogEvent(location::nearby::proto::connections::EventType event_type);

  void UpdateStrategySessionLocked(
      connections::Strategy strategy,
      location::nearby::proto::connections::SessionRole role)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
  void RecordAdvertisingPhaseDurationLocked() const
      ABSL_SHARED_LOCKS_REQUIRED(mutex_);
  void FinishAdvertisingPhaseLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
  void RecordDiscoveryPhaseDurationLocked() const
      ABSL_SHARED_LOCKS_REQUIRED(mutex_);
  void FinishDiscoveryPhaseLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
  bool UpdateAdvertiserConnectionRequestLocked(
      proto::ConnectionsLog::ConnectionRequest *request)
      ABSL_SHARED_LOCKS_REQUIRED(mutex_);
  bool UpdateDiscovererConnectionRequestLocked(
      proto::ConnectionsLog::ConnectionRequest *request)
      ABSL_SHARED_LOCKS_REQUIRED(mutex_);
  bool BothEndpointsRespondedLocked(
      proto::ConnectionsLog::ConnectionRequest *request)
      ABSL_SHARED_LOCKS_REQUIRED(mutex_);
  void LocalEndpointRespondedLocked(
      const std::string &remote_endpoint_id,
      location::nearby::proto::connections::ConnectionRequestResponse response)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
  void RemoteEndpointRespondedLocked(
      const std::string &remote_endpoint_id,
      location::nearby::proto::connections::ConnectionRequestResponse response)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
  void MarkConnectionRequestIgnoredLocked(
      proto::ConnectionsLog::ConnectionRequest *request)
      ABSL_SHARED_LOCKS_REQUIRED(mutex_);
  void FinishStrategySessionLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);

  location::nearby::proto::connections::ConnectionsStrategy
  StrategyToConnectionStrategy(connections::Strategy strategy);

  // Not owned by AnalyticsRecorder. Pointer must refer to a valid object
  // that outlives the one constructed.
  EventLogger *event_logger_;

  SingleThreadExecutor serial_executor_;
  // Protects all sub-protos reading and writing in ConnectionLog.
  Mutex mutex_;

  // ClientSession
  std::unique_ptr<proto::ConnectionsLog::ClientSession> client_session_ =
      std::make_unique<proto::ConnectionsLog::ClientSession>();
  absl::Time started_client_session_time_;
  bool session_was_logged_ ABSL_GUARDED_BY(mutex_) = false;

  // Current StrategySession
  connections::Strategy current_strategy_ ABSL_GUARDED_BY(mutex_) =
      connections::Strategy::kNone;
  std::unique_ptr<proto::ConnectionsLog::StrategySession>
      current_strategy_session_ ABSL_GUARDED_BY(mutex_);
  absl::Time started_strategy_session_time_ ABSL_GUARDED_BY(mutex_);

  // Current AdvertisingPhase
  std::unique_ptr<proto::ConnectionsLog::AdvertisingPhase>
      current_advertising_phase_;
  absl::Time started_advertising_phase_time_;

  // Current DiscoveryPhase
  std::unique_ptr<proto::ConnectionsLog::DiscoveryPhase>
      current_discovery_phase_;
  absl::Time started_discovery_phase_time_;

  absl::btree_map<std::string,
                  std::unique_ptr<proto::ConnectionsLog::ConnectionRequest>>
      incoming_connection_requests_ ABSL_GUARDED_BY(mutex_);
  absl::btree_map<std::string,
                  std::unique_ptr<proto::ConnectionsLog::ConnectionRequest>>
      outgoing_connection_requests_ ABSL_GUARDED_BY(mutex_);
  absl::btree_map<std::string, std::unique_ptr<LogicalConnection>>
      active_connections_ ABSL_GUARDED_BY(mutex_);
};

}  // namespace analytics
}  // namespace nearby
}  // namespace location

#endif  // ANALYTICS_ANALYTICS_RECORDER_H_