storage_write.cpp

// Unless explicitly stated otherwise all files in this repository are licensed
// under the Apache License Version 2.0.
//
// This product includes software developed at Datadog (https://www.datadoghq.com/).
// Copyright 2025-Present Datadog, Inc.

#include "datadog/impl/core/storage_write.hpp"

#include <algorithm>
#include <array>
#include <cctype>
#include <charconv>
#include <chrono>
#include <iostream>
#include <sstream>
#include <string>
#include <string_view>
#include <vector>

#include "datadog/impl/core/core.hpp"
#include "datadog/impl/core/storage/filesystem_wrapper.hpp"
#include "datadog/impl/core/storage/util.hpp"
#include "datadog/impl/core/storage_queue.hpp"
#include "datadog/impl/core/tlv.hpp"
#include "datadog/impl/core/types.hpp"
#include "datadog/impl/core/util/assert.hpp"

// Global version number applied to all event data stored persistently; may be bumped in
// the event of breaking changes in order to abandon previously-written events on disk.
// This versioning scheme applies to the storage implementation as a whole: individual
// features should implement their own versioning schemes internally if needed.
#define DATADOG_EVENT_STORAGE_VERSION "1"  // NOLINT(cppcoreguidelines-macro-usage)

namespace datadog::impl {

// Maximum possible base-10 digits in a uint64_t, without null terminator
static const size_t MAX_UINT64_DECIMAL_DIGITS = 20;

static std::string_view _ms_to_string(
    uint64_t timestamp_ms, std::array<char, MAX_UINT64_DECIMAL_DIGITS>& buffer
) {
  // Populate the provided buffer with the string representation of our uint64_t, using
  // std::to_chars, which does NOT write a null terminator
  char* begin = buffer.data();
  char* end = begin + buffer.size();
  const auto result = std::to_chars(begin, end, timestamp_ms);

  // We require a fixed-size buffer large enough to fit any uint64_t, so conversion
  // should always succeed
  DATADOG_ASSERT(result.ec == std::errc{}, "uint64 to string conversion failed");

  // Return a std::string_view, which does NOT require a null terminator, constructed
  // from our buffer
  const size_t len = result.ptr - begin;
  return std::string_view{begin, len};
}

static Timestamp _ms_to_timestamp(uint64_t timestamp_ms) {
  const int64_t count = static_cast<int64_t>(timestamp_ms);
  if (count < 0) {
    return Timestamp{};
  }
  return Timestamp{std::chrono::nanoseconds(count * 1000000)};
}

static uint64_t _timestamp_to_ms(Timestamp timestamp) {
  // Use raw milliseconds within the storage implementation, since we encode file
  // creation time in the filename with millisecond precision
  auto elapsed = timestamp.time_since_epoch();
  if (elapsed.count() < 0) {
    return 0;
  }
  return std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count();
}

BatchWriterConfig BatchWriterConfig::FromBatchSize(BatchSize batch_size) {
  const Duration max_file_age = BatchSize_ToMaxFileAgeForWrite(batch_size);
  return BatchWriterConfig(max_file_age);
}

BatchWriter::BatchWriter(
    const DiagnosticLogger& in_diagnostic_logger,
    IFilesystem& in_fs,
    FeatureEventStorage& in_storage,
    const platform::IClock& in_clock,
    BatchWriterConfig in_config
)
    : _diagnostic_logger(in_diagnostic_logger),
      _consent(TrackingConsent::Pending),
      _fs(in_fs),
      _storage(in_storage),
      _clock(in_clock),
      _config(in_config) {}

bool BatchWriter::SetTrackingConsent(TrackingConsent value) {
  // Ignore spurious change events
  if (_consent == value) {
    // Value change accepted successfully as a no-op
    return true;
  }

  // If we're transitioning out of the consent-pending state, any files in the pending
  // directory are about to be deleted or moved: ensure that we clear any active-file
  // details to ensure we stop writing to those files
  if (_consent == TrackingConsent::Pending) {
    _active_pending_file.Clear();
  }

  // Store the new value
  _consent = value;

  // If tracking consent has been revoked, delete all pending data
  if (value == TrackingConsent::NotGranted) {
    // Allow data that was collected while consent was granted to be drained and
    // uploaded; no new data will be fed in
    return _storage.DeletePendingBatches();
  }

  // If tracking consent has been granted, migrate all event data from the pending
  // directory to the granted directory, so the upload thread will be able to read it
  if (value == TrackingConsent::Granted) {
    return _storage.MigratePendingBatchesToGranted();
  }

  // If tracking consent has been changed back to pending, do nothing: the storage
  // thread will write future events to the pending directory, and the upload thread
  // will drain the granted directory
  return true;
}

bool BatchWriter::HandleWrite(
    Block event, Block event_metadata, bool bypass_tracking_consent
) {
  DATADOG_ASSERT(!event.empty(), "HandleWrite received empty event");

  if (bypass_tracking_consent) {
    return FlushEvent(event, event_metadata, TrackingConsent::Granted);
  }

  switch (_consent) {
    case TrackingConsent::Granted:
    case TrackingConsent::Pending:
      return FlushEvent(event, event_metadata, _consent);
    case TrackingConsent::NotGranted:
      return true;  // Event successfully handled as a no-op
  }
  DATADOG_ASSERT(false, "unhandled TrackingConsent enum value");
  return false;
}

bool BatchWriter::FlushEvent(
    Block event, Block event_metadata, TrackingConsent consent
) {
  // If we don't permit at least 1 write per file, reject all writes
  if (_config.max_writes_per_file <= 0) {
    return false;
  }

  // Compute the number of bytes needed to serialize our event block (tagged with a TLV
  // header) and, if applicable, the accompanying metadata block
  size_t num_bytes = TLVBlockHeader::SIZE + event.size();
  if (!event_metadata.empty()) {
    num_bytes += TLVBlockHeader::SIZE + event_metadata.size();
  }

  // Reject the event if it exceeds either the per-event cap or the batch file size
  // limit. In the normal production config max_event_size < max_file_size, so the event
  // cap is the binding constraint; taking the minimum preserves the file-size guard for
  // configurations where max_file_size may be set smaller than max_event_size.
  const size_t max_write_size = std::min(_config.max_event_size, _config.max_file_size);
  if (num_bytes > max_write_size) {
    _diagnostic_logger.Error(
        "Event dropped; size of single event exceeds max write size",
        {{"event_size", num_bytes},
         {"max_event_size", _config.max_event_size},
         {"max_file_size", _config.max_file_size}}
    );
    return false;
  }

  // Determine which file we should write to, and abort if we were unable to resolve an
  // appropriate writable file
  FileDetails* active_file = PrepareFileForNextWrite(consent, event, event_metadata);
  if (!active_file) {
    // If this error occurs, it's likely due to an underlying I/O error, or else we're
    // flooding the storage thread with 100+ batches worth of event data in a very small
    // time span, such that there are no available timestamps left to use as filenames
    _diagnostic_logger.Error("Event dropped; could not prepare batch file for write");
    return false;
  }

  // Open the file for write: we re-open batch files on each write, closing after each
  // event written, in order to ensure that events are reliably flushed as soon as
  // they're handled by the storage thread
  const bool append = true;
  const bool hold_advisory_lock = false;
  auto open_res = FilesystemWrapper(_fs).OpenForWrite(
      active_file->path.CStr(), append, hold_advisory_lock
  );
  if (open_res.value != FilesystemResult::OK) {
    _diagnostic_logger.Error(
        "Event dropped; could not open batch file for write",
        {{"path", active_file->path.Get()},
         {"error", FilesystemResultStr(open_res.value)}}
    );
    return false;
  }
  File& file = open_res.file;

  // We maintain a reusable buffer to concatenate all this data into a single contiguous
  // region, so we can write it to file atomically: to avoid excessive allocations,
  // round up to a reasonable threshold
  const size_t buffer_capacity = QuantizeBufferSize(num_bytes);
  _write_buffer.reserve(buffer_capacity);
  _write_buffer.resize(num_bytes);

  // Write our data to the intermediate buffer in TLV format
  char* write_addr = _write_buffer.data();
  char* write_buffer_end = write_addr + _write_buffer.size();

  // If we have a metadata block, prepend it
  if (!event_metadata.empty()) {
    const size_t metadata_tlv_size = EncodeTLVBlock(
        write_addr,
        write_buffer_end - write_addr,
        TLVBlockType::Metadata,
        event_metadata
    );
    DATADOG_ASSERT(
        metadata_tlv_size > 0, "Failed to write TLV metadata block to buffer"
    );
    write_addr += metadata_tlv_size;
  }

  // Encode the event block into our write buffer
  const size_t event_tlv_size = EncodeTLVBlock(
      write_addr, write_buffer_end - write_addr, TLVBlockType::Event, event
  );
  DATADOG_ASSERT(event_tlv_size > 0, "Failed to write TLV event block to buffer");
  write_addr += event_tlv_size;  // NOLINT(clang-analyzer-deadcode.DeadStores)
  DATADOG_ASSERT(write_addr == write_buffer_end, "Unexpected number of bytes encoded");

  // Perform a single atomic write to ensure that header + data (and metadata + event,
  // if applicable) are written together, all-or-nothing
  const auto write_res = file.Write(_write_buffer.data(), _write_buffer.size());
  if (write_res.value != FilesystemResult::OK) {
    // Write failed: log an error, drop the event, and carry on attempting to write
    // future events
    _diagnostic_logger.Error(
        "Event dropped; write to batch file failed",
        {{"path", active_file->path.Get()},
         {"error", FilesystemResultStr(write_res.value)}}
    );
    return false;
  }

  // File::~File() will close the file handle automatically in the event of failure, but
  // close it explicitly after a successful write
  const auto close_res = file.Close();
  if (close_res != FilesystemResult::OK) {
    _diagnostic_logger.Warning(
        "Failed to close batch file after event write",
        {{"path", active_file->path.Get()}, {"error", FilesystemResultStr(close_res)}}
    );
  }

  // Write successful; update our current-file state
  active_file->num_writes++;
  active_file->num_bytes_written += write_res.bytes_written;
  return true;
}

BatchWriter::FileDetails* BatchWriter::PrepareFileForNextWrite(
    TrackingConsent consent, Block event, Block event_metadata
) {
  // We should never be this deep in the actually-writing-something-to-a-file code path
  // when consent is NotGranted
  DATADOG_ASSERT(
      consent == TrackingConsent::Pending || consent == TrackingConsent::Granted,
      "Invalid TrackingConsent value for PrepareFileForNextWrite"
  );

  // Select the appropriate FileDetails member and directory path based on the consent
  // value associated with the event we want to write
  FileDetails& active_file =
      consent == TrackingConsent::Granted ? _active_granted_file : _active_pending_file;
  const StoragePath& consent_dir_path = consent == TrackingConsent::Granted
                                            ? _storage.GetGrantedPath()
                                            : _storage.GetPendingPath();

  // Check our last-used file's age, size, etc. to see if we can reuse it
  const Timestamp current_time = _clock.Now();
  if (CanReuseFileForNextWrite(active_file, current_time, event, event_metadata)) {
    return &active_file;
  }

  // If we can't reuse the last file, prepare to write a new file. List the directory
  // contents once here so that both the quota check and the filename search can share
  // the same listing.
  if (!CacheKnownFilenames(consent_dir_path)) {
    active_file.Clear();
    return nullptr;
  }

  // Enforce the directory size quota, using and updating the cached listing
  PurgeDirectoryIfNeeded(consent_dir_path);

  // Figure out what to name the new file, using the post-purge cached listing
  const auto next = GetFilenameForNextWrite(current_time);
  if (!next) {
    // All potential filenames are in use; can't proceed with file creation
    active_file.Clear();
    return nullptr;
  }
  const uint64_t next_filename_ms = next->first;
  const std::string& next_filename = next->second;

  // Reset our state to reflect that we have a new file, then return a non-owning
  // pointer to the buffer that holds the path to that file
  if (!active_file.Reset(consent_dir_path, next_filename, next_filename_ms)) {
    return nullptr;
  }
  return &active_file;
}

bool BatchWriter::CanReuseFileForNextWrite(
    const FileDetails& active_file,
    Timestamp current_time,
    Block event,
    Block event_metadata
) const {
  // If we have no current file, we need a new one
  if (active_file.path.Get().empty()) {
    return false;
  }

  // If the current file is older than our maximum age for a writable file, leave it
  // alone and start a new file
  const Timestamp presumed_creation_time = _ms_to_timestamp(active_file.filename_ms);
  const Duration presumed_age = current_time - presumed_creation_time;
  if (presumed_age > _config.max_file_age) {
    return false;
  }

  // If we've reached our hard limit on the number of events recorded in a single file,
  // it's time to start a new batch
  if (active_file.num_writes >= _config.max_writes_per_file) {
    return false;
  }

  // Compute the number of bytes we'll want to append to the current file, but take
  // caution not to use sizeof(TLVBlockHeader), which includes struct padding etc.;
  // always use TLVBlockHeader::SIZE when computing serialized size
  size_t num_bytes_to_write = TLVBlockHeader::SIZE + event.size();
  if (!event_metadata.empty()) {
    num_bytes_to_write += TLVBlockHeader::SIZE + event_metadata.size();
  }

  // If the file would exceed our hard limit on file size after write, it's time to call
  // it quits on that file and start a new one
  const size_t expected_size_after_write =
      active_file.num_bytes_written + num_bytes_to_write;
  if (expected_size_after_write > _config.max_file_size) {
    return false;
  }

  // We have a previously-used file that's still young enough and small enough to
  // continue writing to; allow it to be reused
  return true;
}

std::optional<std::pair<uint64_t, std::string>> BatchWriter::GetFilenameForNextWrite(
    Timestamp current_time
) const {
  // Our new file will use a filename reflecting the current timestamp. If a file with
  // that name already exists (per the pre-populated _last_known_filenames), we
  // increment by 1ms at a time to find a free slot, up to a limit of 100 attempts.

  // Use stack memory to convert uint64 to string for candidate filenames
  std::array<char, MAX_UINT64_DECIMAL_DIGITS> buffer{0};

  // Run an initial binary search to find the position in our sorted filenames array
  // where this entry would theoretically be inserted
  uint64_t filename_ms = _timestamp_to_ms(current_time);
  std::string_view filename = _ms_to_string(filename_ms, buffer);
  std::vector<std::string>::const_iterator it = std::lower_bound(
      _last_known_filenames.cbegin(), _last_known_filenames.cend(), filename
  );

  // If we didn't find an exact match, there's no existing file with this name, so we're
  // good to use it
  if (it == _last_known_filenames.cend() || *it != filename) {
    // Return a copy of the filename string
    return std::make_pair(filename_ms, std::string(filename));
  }

  // If the target filename is already in use, loop over all possible filenames that we
  // might use for the next ~100ms, until we land on one for which there is no collision
  // (or exhaust the search)
  for (uint64_t offset = 1; offset < 100; offset++) {
    // Compute the next filename in the series, incrementing by 1 millisecond
    filename_ms++;
    filename = _ms_to_string(filename_ms, buffer);

    // Perform a linear search from our current iterator position to the end of the
    // filename vector, to see if this next filename is also in use
    it = std::find_if(it, _last_known_filenames.cend(), [=](const std::string& s) {
      return s == filename;
    });

    // If we got no match, we're good to use the current filename
    if (it == _last_known_filenames.cend()) {
      // Return a copy of the filename string
      return std::make_pair(filename_ms, std::string(filename));
    }

    // Otherwise, our iterator has advanced, so the next loop will test the next
    // candidate filename in a smaller linear search space
  }

  // We somehow had 100 sequentially-named files in the target directory; give up on
  // file creation
  return std::nullopt;
}

bool BatchWriter::CacheKnownFilenames(const StoragePath& consent_dir_path) const {
  // Ensure that our vector is empty before we attempt to populate it
  _last_known_filenames.clear();

  // Retrieve a directory listing (regular files only), caching the results
  FilesystemWrapper fsw(_fs);
  const auto res = fsw.ListFiles(consent_dir_path.CStr(), _last_known_filenames);
  if (res != FilesystemResult::OK) {
    _diagnostic_logger.Warning(
        "Failed to examine existing batch files on event write: unable to list files",
        {{"path", consent_dir_path.Get()}, {"error", FilesystemResultStr(res)}}
    );
    return false;
  }

  // Sort filenames for deterministic iteration in timestamp-name-order, then strip any
  // non-numeric names (files not written by the SDK) so that both
  // PurgeDirectoryIfNeeded and GetFilenameForNextWrite operate only on batch files we
  // own.
  std::sort(_last_known_filenames.begin(), _last_known_filenames.end());
  _last_known_filenames.erase(
      std::remove_if(
          _last_known_filenames.begin(),
          _last_known_filenames.end(),
          [](const std::string& name) {
            return !std::all_of(name.begin(), name.end(), [](unsigned char c) {
              return std::isdigit(c);
            });
          }
      ),
      _last_known_filenames.end()
  );
  return true;
}

void BatchWriter::PurgeDirectoryIfNeeded(const StoragePath& dir) {
  // _last_known_filenames is pre-populated, sorted, and already filtered to
  // numeric-named (SDK-written) batch files by CacheKnownFilenames.

  // Accumulate sizes for each file using a reusable path buffer. Files whose size
  // cannot be determined are treated as zero: they don't push the running total toward
  // the quota threshold, but remain eligible for deletion if the purge loop runs and
  // reaches them (without contributing to total reduction when deleted).
  FilesystemWrapper fsw(_fs);
  StoragePath path_buf;
  std::vector<size_t> sizes;
  sizes.reserve(_last_known_filenames.size());
  size_t total = 0;
  for (const std::string& name : _last_known_filenames) {
    path_buf.MustSet(dir);
    if (!path_buf.Append(name)) {
      sizes.push_back(0);
      continue;
    }
    const auto res = fsw.GetFileSize(path_buf.CStr());
    const size_t file_size = res.value == FilesystemResult::OK ? res.size : 0;
    sizes.push_back(file_size);
    total += file_size;
  }

  // Note: this check covers only the files already on disk; it does not account for the
  // new batch file that PrepareFileForNextWrite is about to create. The directory may
  // therefore exceed max_directory_size by up to max_file_size after each rotation.
  // This matches the iOS SDK's check-on-create design and the overshoot is bounded and
  // transient, so it is accepted as a known limitation.
  if (total <= _config.max_directory_size) {
    return;
  }

  // Delete oldest files until we're within quota, erasing each from
  // _last_known_filenames and the parallel sizes vector in lock-step so
  // GetFilenameForNextWrite sees the post-purge state. Iterate by index rather than
  // iterator so we can erase in-place: on a successful deletion we don't advance i (the
  // next element slides into position); on failure we skip the file and increment.
  size_t i = 0;
  while (i < _last_known_filenames.size() && total > _config.max_directory_size) {
    path_buf.MustSet(dir);
    if (!path_buf.Append(_last_known_filenames[i])) {
      ++i;
      continue;
    }
    const auto delete_res = fsw.Delete(path_buf.CStr());
    // DoesNotExist means the upload thread already removed this file concurrently;
    // treat it as a successful removal so total and _last_known_filenames stay
    // consistent and we don't over-delete younger files to compensate.
    if (delete_res == FilesystemResult::OK ||
        delete_res == FilesystemResult::DoesNotExist) {
      total -= sizes[i];
      if (delete_res == FilesystemResult::OK) {
        _diagnostic_logger.Debug(
            "Deleted batch file to enforce directory size quota",
            {{"path", path_buf.Get()}}
        );
      }
      const auto idx = static_cast<std::ptrdiff_t>(i);
      _last_known_filenames.erase(_last_known_filenames.begin() + idx);
      sizes.erase(sizes.begin() + idx);
    } else {
      _diagnostic_logger.Warning(
          "Failed to delete batch file during directory quota enforcement",
          {{"path", path_buf.Get()}, {"error", FilesystemResultStr(delete_res)}}
      );
      ++i;
    }
  }
}

}  // namespace datadog::impl