storage_write.hpp

// 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.

#pragma once

#include <cstdint>
#include <optional>
#include <string_view>
#include <utility>
#include <vector>

#include "datadog/core.hpp"

#include "datadog/impl/core/block.hpp"
#include "datadog/impl/core/platform/clock.hpp"
#include "datadog/impl/core/storage/feature_event.hpp"
#include "datadog/impl/core/storage/filesystem.hpp"
#include "datadog/impl/core/storage/path.hpp"
#include "datadog/impl/core/util/assert.hpp"
#include "datadog/impl/core/util/diagnostics.hpp"

namespace datadog::impl {

/**
 * Controls how often we create new batch files, based on both time and size limits.
 */
struct BatchWriterConfig {
  /**
   * Once a file exceeds this age, we will not write to it again.
   */
  Duration max_file_age;
  /**
   * If the next write to the current file would cause it to exceed this size (in
   * bytes), we will not write to that file and will instead create a new one.
   */
  size_t max_file_size{0x400000};  // 4 MB
  /**
   * If the encoded size of a single event write (metadata + event, with TLV headers)
   * exceeds this threshold, the event is dropped rather than written.
   */
  size_t max_event_size{512UL * 1024};  // 512 KB
  /**
   * Maximum total size (in bytes) of all batch files in a single consent-level
   * directory (pending or granted) for one feature. When a new batch file is about to
   * be created and the directory already exceeds this limit, the oldest files are
   * deleted until the total is within the quota.
   */
  size_t max_directory_size{512ULL * 1024 * 1024};  // 512 MB
  /**
   * Maximum number of events that we will write to a single file. A single write
   * operation may include both metadata and event blocks: i.e. if configured with a
   * maximum of 500 writes, a file will contain no more than 500 TLV Metadata blocks and
   * 500 TLV Event blocks.
   */
  int max_writes_per_file{500};

  explicit BatchWriterConfig(Duration in_max_file_age)
      : max_file_age(in_max_file_age) {}

  static BatchWriterConfig FromBatchSize(BatchSize batch_size);
};

/**
 * Implements the logic used in the storage thread to commit events to persistent
 * storage for a specific feature.
 *
 * Events will be written to "batch files", each of which contains a series of
 * TLV-encoded blocks of event data (optionally described with metadata blocks) which
 * have been generated by the associated feature within a contiguous span of time.
 *
 * Events are handled differently depending on the SDK's current TrackingConsent value.
 * When consent is pending, events are written to batches in the pending directory
 * (e.g. 'intermediate-v1/'), whereas events are written to the consent-granted
 * directory (e.g. 'v1/', which the upload thread reads from) when consent is granted.
 * `BatchWriter` uses a `FeatureEventStorage` interface to access these paths.
 *
 * When consent is explicitly revoked (NotGranted), events are not stored at all.
 *
 * The BatchWriter keeps track of the batch file that it most recently wrote to, and
 * on each write it makes decisions about when to close the file and start a new one,
 * what new files should be named, etc.
 */
class BatchWriter {
 private:
  // === Essential state and dependencies ===

  /**
   * Interface used to emit local log messages, mostly to aid in debugging the behavior
   * of the storage thread.
   */
  DiagnosticLogger _diagnostic_logger;

  /**
   * Current tracking consent value for the SDK, determining which consent-level
   * subdirectory (if any) new events will be written to.
   */
  TrackingConsent _consent;

  /**
   * Interface used for file I/O.
   */
  IFilesystem& _fs;

  /**
   * Reference to the event storage interface for the associated feature, owned by the
   * Core. Provides access to the directories where events are written, and handles
   * mass-deletion and migration of files between those directories.
   */
  FeatureEventStorage& _storage;

  /**
   * Clock used to determine file names and make decisions based on file age. MUST be
   * the same clock used by the upload thread.
   */
  const platform::IClock& _clock;

  /**
   * Specifies how large a batch file may get, how old it can get before we stop writing
   * to it, etc.
   */
  const BatchWriterConfig _config;

  // === Transient state used for batching decisions ===

  /**
   * Buffers the set of filenames retrieved in our last call to CacheKnownFilenames().
   * Contains the names of all SDK-written (numeric-named) batch files present in the
   * target directory for the current consent value, sorted in ascending order. Reset
   * when tracking consent changes.
   */
  mutable std::vector<std::string> _last_known_filenames;

  /**
   * Details of the last file we wrote an event to with a specific consent value.
   *
   * @note We never reopen existing files such that we would need to reinitialize this
   *  state from a non-empty file. If the SDK is initialized and batch files from a
   *  prior run are still present in the storage directory, the storage thread makes no
   *  attempt to reopen them. Similarly, when a file is moved due to a tracking consent
   *  change, it is closed and left as-is, and a brand new batch file is opened to
   *  contain new events.
   */
  struct FileDetails {
    StoragePath path;
    uint64_t filename_ms{0};
    int num_writes{0};
    size_t num_bytes_written{0};

    void Clear() {
      const bool path_ok = path.Set("");
      DATADOG_ASSERT(path_ok, "Failed to write empty string into StoragePath");
      filename_ms = 0;
      num_writes = 0;
      num_bytes_written = 0;
    }

    bool Reset(
        const StoragePath& parent_dir,
        std::string_view filename,
        uint64_t in_filename_ms
    ) {
      path.MustSet(parent_dir);
      if (path.Append(filename)) {
        filename_ms = in_filename_ms;
        num_writes = 0;
        num_bytes_written = 0;
        return true;
      }
      Clear();
      return false;
    }
  };
  FileDetails _active_pending_file;  // Last file that we wrote to in pending dir
  FileDetails _active_granted_file;  // Last file that we wrote to in granted dir

  /**
   * Buffer used to concatenate TLV block data prior to writing, so that we can ensure
   * atomic writes: i.e. a TLV header will never be written without its corresponding
   * block data, and if a Metadata block is present, it will always be followed by the
   * corresponding Event block.
   *
   * We might be able to avoid allocations from this buffer if we loosened our
   * guarantees about batch file consistency and allowed partial writes.
   */
  std::vector<char> _write_buffer;

 public:
  explicit BatchWriter(
      const DiagnosticLogger& in_diagnostic_logger,
      IFilesystem& in_fs,
      FeatureEventStorage& in_storage,
      const platform::IClock& in_clock,
      BatchWriterConfig in_config
  );

  /**
   * Notifies the storage thread that the SDK's tracking consent value has changed.
   */
  bool SetTrackingConsent(TrackingConsent value);

  /**
   * Handles an event generated by a feature, writing it to the appropriate event
   * storage directory based on the current tracking consent.
   *
   * `event` is the binary payload to be written as a TLV 'Event' block; it must be
   * non-empty. `event_metadata` is optional metadata to be written as a TLV 'Metadata'
   * block immediately preceding the 'Event' block; pass an empty block to omit it.
   *
   * When `bypass_tracking_consent` is `true`, the event is always written to the
   * granted-consent storage directory, regardless of the current `TrackingConsent`
   * value. This is intended for crash reporting, which must honour the consent that was
   * active at crash time rather than the current live consent.
   *
   * Atomic writes are guaranteed: on success, all requested data is written to the
   * file; on failure, no data is written.
   */
  bool HandleWrite(Block event, Block event_metadata, bool bypass_tracking_consent);

 private:
  /**
   * Attempts to write the given event to a batch file within the directory associated
   * with the given `consent` value. Reuses the last file written to (either
   * `_active_pending_file` or `_active_granted_file`, depending on `consent`) if still
   * valid, otherwise creates a new file.
   */
  bool FlushEvent(Block event, Block event_metadata, TrackingConsent consent);

  /**
   * Resolves the appropriate file that we should use to write the next event to,
   * whether that's _active_(pending|granted)_file or a newly-created file. Populates
   * _active_pending_file (when `consent` is pending) or _active_granted_file (when
   * `consent` is granted).
   *
   * Returns a pointer to the active file struct that holds the details of the batch
   * file to be written to, which may or may not exist yet on disk. A return value of
   * nullptr indicates that no file is available and the event can not be stored.
   */
  FileDetails* PrepareFileForNextWrite(
      TrackingConsent consent, Block event, Block event_metadata
  );

  /**
   * Determines whether _active_file is still suitable for writing this event to.
   */
  bool CanReuseFileForNextWrite(
      const FileDetails& active_file,
      Timestamp current_time,
      Block event,
      Block event_metadata
  ) const;

  /**
   * Designates a name for a newly-created batch file, returning both the integer
   * millisecond count reflecting file creation time and the string-formatted value.
   *
   * Requires that `_last_known_filenames` has already been populated by
   * `CacheKnownFilenames` before this function is called.
   */
  std::optional<std::pair<uint64_t, std::string>> GetFilenameForNextWrite(
      Timestamp current_time
  ) const;

  /**
   * Populates `_last_known_filenames` with a sorted listing of all numeric-named
   * (SDK-written) regular files in `consent_dir_path`, excluding any foreign files
   * whose names contain non-digit characters. Must be called in
   * `PrepareFileForNextWrite` before invoking `PurgeDirectoryIfNeeded` or
   * `GetFilenameForNextWrite`.
   *
   * Returns true on success, false if the directory listing fails.
   */
  bool CacheKnownFilenames(const StoragePath& consent_dir_path) const;

  /**
   * Enforces the per-directory size quota for `dir` by deleting the oldest batch files
   * until the total size of remaining files falls within `_config.max_directory_size`.
   *
   * Reads and mutates `_last_known_filenames` (pre-populated and pre-filtered to
   * numeric-named files by `CacheKnownFilenames`), removing deleted entries so that
   * `GetFilenameForNextWrite` sees an accurate post-purge listing.
   *
   * Called only when a new batch file is about to be created, mirroring the iOS SDK's
   * approach of deferring the quota check to file-creation time. The check is
   * best-effort: errors from sizing individual files are tolerated.
   */
  void PurgeDirectoryIfNeeded(const StoragePath& dir);
};

}  // namespace datadog::impl