// Copyright 2021 The Pigweed Authors
//
// 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.

#pragma once

#include <array>
#include <cstdint>

#include "pw_assert/assert.h"
#include "pw_bytes/span.h"
#include "pw_log/proto/log.pwpb.h"
#include "pw_log_rpc/log_filter.h"
#include "pw_multisink/multisink.h"
#include "pw_protobuf/serialized_size.h"
#include "pw_result/result.h"
#include "pw_rpc/raw/server_reader_writer.h"
#include "pw_status/status.h"
#include "pw_sync/lock_annotations.h"
#include "pw_sync/mutex.h"

namespace pw::log_rpc {

// RpcLogDrain matches a MultiSink::Drain with with an RPC channel's writer. A
// RPC channel ID identifies this drain. The user must attach this drain
// to a MultiSink that returns a log::LogEntry, and provide a buffer large
// enough to hold the largest log::LogEntry transmittable. The user must call
// Flush(), which, on every call, packs as many log::LogEntry items as possible
// into a log::LogEntries message, writes the message to the provided writer,
// then repeats the process until there are no more entries in the MultiSink or
// the writer failed to write the outgoing package and error_handling is set to
// `kCloseStreamOnWriterError`. When error_handling is `kIgnoreWriterErrors` the
// drain will continue to retrieve log entries out of the MultiSink and attempt
// to send them out ignoring the writer errors without sending a drop count.
// Note: the error handling and drop count reporting might change in the future.
// Log filtering is done using the rules of the Filter provided if any.
class RpcLogDrain : public multisink::MultiSink::Drain {
 public:
  // Dictates how to handle server writer errors.
  enum class LogDrainErrorHandling {
    kIgnoreWriterErrors,
    kCloseStreamOnWriterError,
  };

  // The minimum buffer size, without the message payload or module sizes,
  // needed to retrieve a log::LogEntry from the attached MultiSink. The user
  // must account for the max message size to avoid log entry drops. The dropped
  // field is not accounted since a dropped message has all other fields unset.
  static constexpr size_t kMinEntrySizeWithoutPayload =
      // message
      protobuf::SizeOfFieldKey(1) +
      1  // Assume minimum varint length, skip the payload bytes.
      // line_level
      + protobuf::SizeOfFieldKey(2) +
      protobuf::kMaxSizeBytesUint32
      // flags
      + protobuf::SizeOfFieldKey(3) +
      protobuf::kMaxSizeBytesUint32
      // timestamp or time_since_last_entry
      + protobuf::SizeOfFieldKey(4) +
      protobuf::kMaxSizeBytesInt64
      // Module
      + protobuf::SizeOfFieldKey(7) +
      1;  // Assume minimum varint length, skip the module bytes.
  // The smallest buffer size must be able to fit a typical token size: 4 bytes.
  static constexpr size_t kMinEntryBufferSize = kMinEntrySizeWithoutPayload + 4;

  // When encoding LogEntry in LogEntries, there are kLogEntryEncodeFrameSize
  // bytes added to the encoded LogEntry. This constant and kMinEntryBufferSize
  // can be used to calculate the minimum RPC ChannelOutput buffer size.
  static constexpr size_t kLogEntryEncodeFrameSize =
      protobuf::SizeOfFieldKey(1)  // LogEntry
      + protobuf::kMaxSizeOfLength;

  // Creates a closed log stream with a writer that can be set at a later time.
  // The provided buffer must be large enough to hold the largest transmittable
  // log::LogEntry or a drop count message at the very least. The user can
  // choose to provide a unique mutex for the drain, or share it to save RAM as
  // long as they are aware of contengency issues.
  RpcLogDrain(const uint32_t channel_id,
              ByteSpan log_entry_buffer,
              sync::Mutex& mutex,
              LogDrainErrorHandling error_handling,
              Filter* filter = nullptr)
      : channel_id_(channel_id),
        error_handling_(error_handling),
        server_writer_(),
        log_entry_buffer_(log_entry_buffer),
        committed_entry_drop_count_(0),
        mutex_(mutex),
        filter_(filter) {
    PW_ASSERT(log_entry_buffer.size_bytes() >= kMinEntryBufferSize);
  }

  // Not copyable.
  RpcLogDrain(const RpcLogDrain&) = delete;
  RpcLogDrain& operator=(const RpcLogDrain&) = delete;

  // Configures the drain with a new open server writer if the current one is
  // not open.
  //
  // Return values:
  // OK - Successfully set the new open writer.
  // FAILED_PRECONDITION - The given writer is not open.
  // ALREADY_EXISTS - an open writer is already set.
  Status Open(rpc::RawServerWriter& writer) PW_LOCKS_EXCLUDED(mutex_);

  // Accesses log entries and sends them via the writer. Expected to be called
  // frequently to avoid log drops. If the writer fails to send a packet with
  // multiple log entries, the entries are dropped and a drop message with the
  // count is sent. When error_handling is kCloseStreamOnWriterError, the stream
  // will automatically be closed and Flush will return the writer error.
  //
  // Precondition: the drain must be attached to a MultiSink.
  //
  // Return values:
  // OK - all entries were consumed.
  // ABORTED - there was an error writing the packet, and error_handling equals
  // `kCloseStreamOnWriterError`.
  Status Flush() PW_LOCKS_EXCLUDED(mutex_);

  // Ends RPC log stream without flushing.
  //
  // Return values:
  // OK - successfully closed the server writer.
  // FAILED_PRECONDITION - The given writer is not open.
  // Errors from the underlying writer send packet.
  Status Close() PW_LOCKS_EXCLUDED(mutex_);

  uint32_t channel_id() const { return channel_id_; }

 private:
  enum class LogDrainState {
    kCaughtUp,
    kMoreEntriesRemaining,
  };

  // Fills the outgoing buffer with as many entries as possible.
  LogDrainState EncodeOutgoingPacket(log::LogEntries::MemoryEncoder& encoder,
                                     uint32_t& packed_entry_count_out)
      PW_EXCLUSIVE_LOCKS_REQUIRED(mutex_);

  const uint32_t channel_id_;
  const LogDrainErrorHandling error_handling_;
  rpc::RawServerWriter server_writer_ PW_GUARDED_BY(mutex_);
  const ByteSpan log_entry_buffer_ PW_GUARDED_BY(mutex_);
  uint32_t committed_entry_drop_count_ PW_GUARDED_BY(mutex_);
  sync::Mutex& mutex_;
  Filter* filter_;
};

}  // namespace pw::log_rpc