// Copyright (c) 2011 The LevelDB Authors. All rights reserved.

// Use of this source code is governed by a BSD-style license that can be

// found in the LICENSE file. See the AUTHORS file for names of contributors.

#include "db/log_writer.h"

#include <cstdint>

#include "leveldb/env.h"

#include "util/coding.h"

#include "util/crc32c.h"

namespace leveldb {

namespace log {

static void InitTypeCrc(uint32_t* type_crc) {

  for (int i = 0; i <= kMaxRecordType; i++) {

    char t = static_cast<char>(i);

    type_crc[i] = crc32c::Value(&t, 1);

  }

}

Writer::Writer(WritableFile* dest) : dest_(dest), block_offset_(0) {

  InitTypeCrc(type_crc_);

}

Writer::Writer(WritableFile* dest, uint64_t dest_length)

    : dest_(dest), block_offset_(dest_length % kBlockSize) {

  InitTypeCrc(type_crc_);

}

Writer::~Writer() = default;

Status Writer::AddRecord(const Slice& slice) {

  const char* ptr = slice.data();

  size_t left = slice.size();

  // Fragment the record if necessary and emit it.  Note that if slice

  // is empty, we still want to iterate once to emit a single

  // zero-length record

  Status s;

  bool begin = true;

  do {

    const int leftover = kBlockSize - block_offset_;

    assert(leftover >= 0);

    if (leftover < kHeaderSize) {

      // Switch to a new block

      if (leftover > 0) {

        // Fill the trailer (literal below relies on kHeaderSize being 7)

        static_assert(kHeaderSize == 7, "");

        dest_->Append(Slice("\x00\x00\x00\x00\x00\x00", leftover));

      }

      block_offset_ = 0;

    }

    // Invariant: we never leave < kHeaderSize bytes in a block.

    assert(kBlockSize - block_offset_ - kHeaderSize >= 0);

    const size_t avail = kBlockSize - block_offset_ - kHeaderSize;

    const size_t fragment_length = (left < avail) ? left : avail;

    RecordType type;

    const bool end = (left == fragment_length);

    if (begin && end) {

      type = kFullType;

    } else if (begin) {

      type = kFirstType;

    } else if (end) {

      type = kLastType;

    } else {

      type = kMiddleType;

    }

    s = EmitPhysicalRecord(type, ptr, fragment_length);

    ptr += fragment_length;

    left -= fragment_length;

    begin = false;

  } while (s.ok() && left > 0);

  return s;

}

Status Writer::EmitPhysicalRecord(RecordType t, const char* ptr,

                                  size_t length) {

  assert(length <= 0xffff);  // Must fit in two bytes

  assert(block_offset_ + kHeaderSize + length <= kBlockSize);

  // Format the header

  char buf[kHeaderSize];

  buf[4] = static_cast<char>(length & 0xff);

  buf[5] = static_cast<char>(length >> 8);

  buf[6] = static_cast<char>(t);

  // Compute the crc of the record type and the payload.

  uint32_t crc = crc32c::Extend(type_crc_[t], ptr, length);

  crc = crc32c::Mask(crc);  // Adjust for storage

  EncodeFixed32(buf, crc);

  // Write the header and the payload

  Status s = dest_->Append(Slice(buf, kHeaderSize));

  if (s.ok()) {

    s = dest_->Append(Slice(ptr, length));

    if (s.ok()) {

      s = dest_->Flush();

    }

  }

  block_offset_ += kHeaderSize + length;

  return s;

}

}  // namespace log

}  // namespace leveldb