diff --git a/CMakeLists.txt b/CMakeLists.txt
index 4ed6be3..2281ae5 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -120,11 +120,15 @@ set(source_files
   src/binsrv/event/unknown_post_header.hpp
   src/binsrv/event/unknown_post_header.cpp
 
-  # billog files
+  # binlog files
   src/binsrv/basic_logger_fwd.hpp
   src/binsrv/basic_logger.hpp
   src/binsrv/basic_logger.cpp
 
+  src/binsrv/basic_storage_backend_fwd.hpp
+  src/binsrv/basic_storage_backend.hpp
+  src/binsrv/basic_storage_backend.cpp
+
   src/binsrv/cout_logger.hpp
   src/binsrv/cout_logger.cpp
 
@@ -134,9 +138,8 @@ set(source_files
   src/binsrv/file_logger.hpp
   src/binsrv/file_logger.cpp
 
-  src/binsrv/filesystem_storage_fwd.hpp
-  src/binsrv/filesystem_storage.hpp
-  src/binsrv/filesystem_storage.cpp
+  src/binsrv/filesystem_storage_backend.hpp
+  src/binsrv/filesystem_storage_backend.cpp
 
   src/binsrv/log_severity_fwd.hpp
   src/binsrv/log_severity.hpp
@@ -147,9 +150,16 @@ set(source_files
   src/binsrv/logger_factory.hpp
   src/binsrv/logger_factory.cpp
 
-  src/binsrv/master_config_fwd.hpp
-  src/binsrv/master_config.hpp
-  src/binsrv/master_config.cpp
+  src/binsrv/main_config_fwd.hpp
+  src/binsrv/main_config.hpp
+  src/binsrv/main_config.cpp
+
+  src/binsrv/storage_fwd.hpp
+  src/binsrv/storage.hpp
+  src/binsrv/storage.cpp
+
+  src/binsrv/storage_backend_factory.hpp
+  src/binsrv/storage_backend_factory.cpp
 
   src/binsrv/storage_config_fwd.hpp
   src/binsrv/storage_config.hpp
diff --git a/master_config.json b/main_config.json
similarity index 92%
rename from master_config.json
rename to main_config.json
index 03030ee..6b68c93 100644
--- a/master_config.json
+++ b/main_config.json
@@ -10,6 +10,7 @@
     "password": ""
   },
   "storage": {
+    "type": "fs",
     "path": "./storage"
   }
 }
diff --git a/mtr/binlog_streaming/r/binsrv.result b/mtr/binlog_streaming/r/binsrv.result
index d54adb9..e2a200a 100644
--- a/mtr/binlog_streaming/r/binsrv.result
+++ b/mtr/binlog_streaming/r/binsrv.result
@@ -38,6 +38,7 @@ SET @binsrv_config_json = JSON_OBJECT(
 'password', ''
   ),
 'storage', JSON_OBJECT(
+'type', 'fs',
 'path', @storage_path
 )
 );
diff --git a/mtr/binlog_streaming/t/binsrv.test b/mtr/binlog_streaming/t/binsrv.test
index 41b5d3a..a7cdaee 100644
--- a/mtr/binlog_streaming/t/binsrv.test
+++ b/mtr/binlog_streaming/t/binsrv.test
@@ -43,7 +43,7 @@ INSERT INTO t1 VALUES(DEFAULT);
 --replace_result $binsrv_storage_path <BINSRV_STORAGE_PATH>
 eval SET @storage_path = '$binsrv_storage_path';
 
---let $binsrv_log_path = $MYSQL_TMP_DIR/binsrv.log
+--let $binsrv_log_path = $MYSQL_TMP_DIR/binsrv_utility.log
 --replace_result $binsrv_log_path <BINSRV_LOG_PATH>
 eval SET @log_path = '$binsrv_log_path';
 
@@ -64,6 +64,7 @@ eval SET @binsrv_config_json = JSON_OBJECT(
      'password', ''
   ),
   'storage', JSON_OBJECT(
+     'type', 'fs',
      'path', @storage_path
   )
 );
diff --git a/src/app.cpp b/src/app.cpp
index 5209c34..06fe13c 100644
--- a/src/app.cpp
+++ b/src/app.cpp
@@ -11,15 +11,18 @@
 #include <stdexcept>
 #include <string>
 #include <string_view>
+#include <utility>
 
 #include <boost/lexical_cast.hpp>
 
 #include "binsrv/basic_logger.hpp"
+#include "binsrv/basic_storage_backend.hpp"
 #include "binsrv/exception_handling_helpers.hpp"
-#include "binsrv/filesystem_storage.hpp"
 #include "binsrv/log_severity.hpp"
 #include "binsrv/logger_factory.hpp"
-#include "binsrv/master_config.hpp"
+#include "binsrv/main_config.hpp"
+#include "binsrv/storage.hpp"
+#include "binsrv/storage_backend_factory.hpp"
 
 #include "binsrv/event/code_type.hpp"
 #include "binsrv/event/event.hpp"
@@ -75,7 +78,7 @@ void log_span_dump(binsrv::basic_logger &logger,
 
 void receive_binlog_events(binsrv::basic_logger &logger,
                            easymysql::binlog &binlog,
-                           binsrv::filesystem_storage &storage) {
+                           binsrv::storage &storage) {
   // Network streams are requested with COM_BINLOG_DUMP and
   // each Binlog Event response is prepended with 00 OK-byte.
   static constexpr std::byte expected_event_packet_prefix{'\0'};
@@ -86,7 +89,7 @@ void receive_binlog_events(binsrv::basic_logger &logger,
 
   while (!(portion = binlog.fetch()).empty()) {
     if (portion[0] != expected_event_packet_prefix) {
-      util::exception_location().raise<std::invalid_argument>(
+      util::exception_location().raise<std::runtime_error>(
           "unexpected event prefix");
     }
     portion = portion.subspan(1U);
@@ -138,7 +141,7 @@ int main(int argc, char *argv[]) {
   const auto number_of_cmd_args = std::size(cmd_args);
   const auto executable_name = util::extract_executable_name(cmd_args);
 
-  if (number_of_cmd_args != binsrv::master_config::flattened_size + 1 &&
+  if (number_of_cmd_args != binsrv::main_config::flattened_size + 1 &&
       number_of_cmd_args != 2) {
     std::cerr
         << "usage: " << executable_name
@@ -162,17 +165,16 @@ int main(int argc, char *argv[]) {
     logger->log(binsrv::log_severity::delimiter,
                 util::get_readable_command_line_arguments(cmd_args));
 
-    binsrv::master_config_ptr config;
+    binsrv::main_config_ptr config;
     if (number_of_cmd_args == 2U) {
       logger->log(binsrv::log_severity::delimiter,
                   "Reading connection configuration from the JSON file.");
-      config = std::make_shared<binsrv::master_config>(cmd_args[1]);
-    } else if (number_of_cmd_args ==
-               binsrv::master_config::flattened_size + 1) {
+      config = std::make_shared<binsrv::main_config>(cmd_args[1]);
+    } else if (number_of_cmd_args == binsrv::main_config::flattened_size + 1) {
       logger->log(binsrv::log_severity::delimiter,
                   "Reading connection configuration from the command line "
                   "arguments.");
-      config = std::make_shared<binsrv::master_config>(cmd_args.subspan(1U));
+      config = std::make_shared<binsrv::main_config>(cmd_args.subspan(1U));
     } else {
       assert(false);
     }
@@ -197,22 +199,28 @@ int main(int argc, char *argv[]) {
 
     std::string msg;
     const auto &storage_config = config->root().get<"storage">();
-    binsrv::filesystem_storage storage(storage_config.get<"path">());
-    logger->log(binsrv::log_severity::info,
-                "created filesystem binlog storage");
-    msg = "filesystem binlog storage root path: ";
-    msg += storage.get_root_path();
+    auto storage_backend{
+        binsrv::storage_backend_factory::create(storage_config)};
+    logger->log(binsrv::log_severity::info, "created binlog storage backend");
+    msg = "type: ";
+    msg += storage_config.get<"type">();
+    msg += ", path: ";
+    msg += storage_config.get<"path">();
     logger->log(binsrv::log_severity::info, msg);
 
+    binsrv::storage storage{std::move(storage_backend)};
+    logger->log(binsrv::log_severity::info,
+                "created binlog storage with the provided backend");
+
     const auto last_binlog_name{storage.get_binlog_name()};
     // if storage position is detected to be 0 (empty data directory), we
     // start streaming from the position magic_binlog_offset (4)
     const auto last_binlog_position{
         std::max(binsrv::event::magic_binlog_offset, storage.get_position())};
     if (last_binlog_name.empty()) {
-      msg = "filesystem binlog storage initialized on an empty directory";
+      msg = "binlog storage initialized on an empty directory";
     } else {
-      msg = "filesystem binlog storage initialized at \"";
+      msg = "binlog storage initialized at \"";
       msg += last_binlog_name;
       msg += "\":";
       msg += std::to_string(last_binlog_position);
diff --git a/src/binsrv/basic_storage_backend.cpp b/src/binsrv/basic_storage_backend.cpp
new file mode 100644
index 0000000..119f205
--- /dev/null
+++ b/src/binsrv/basic_storage_backend.cpp
@@ -0,0 +1,54 @@
+#include "binsrv/basic_storage_backend.hpp"
+
+#include <stdexcept>
+#include <string>
+#include <string_view>
+
+#include "util/byte_span_fwd.hpp"
+#include "util/exception_location_helpers.hpp"
+
+namespace binsrv {
+
+[[nodiscard]] storage_object_name_container
+basic_storage_backend::list_objects() {
+  return do_list_objects();
+}
+
+[[nodiscard]] std::string
+basic_storage_backend::get_object(std::string_view name) {
+  return do_get_object(name);
+}
+
+void basic_storage_backend::put_object(std::string_view name,
+                                       util::const_byte_span content) {
+  return do_put_object(name, content);
+}
+
+void basic_storage_backend::open_stream(std::string_view name) {
+  if (stream_opened_) {
+    util::exception_location().raise<std::logic_error>(
+        "cannot open a new stream as the previous one has not been closed");
+  }
+
+  do_open_stream(name);
+  stream_opened_ = true;
+}
+
+void basic_storage_backend::write_data_to_stream(util::const_byte_span data) {
+  if (!stream_opened_) {
+    util::exception_location().raise<std::logic_error>(
+        "cannot write to the stream as it has not been opened");
+  }
+  do_write_data_to_stream(data);
+}
+
+void basic_storage_backend::close_stream() {
+  if (!stream_opened_) {
+    util::exception_location().raise<std::logic_error>(
+        "cannot close the stream as it has not been opened");
+  }
+  do_close_stream();
+  stream_opened_ = false;
+}
+
+} // namespace binsrv
diff --git a/src/binsrv/basic_storage_backend.hpp b/src/binsrv/basic_storage_backend.hpp
new file mode 100644
index 0000000..fb1d41c
--- /dev/null
+++ b/src/binsrv/basic_storage_backend.hpp
@@ -0,0 +1,46 @@
+#ifndef BINSRV_BASIC_STORAGE_BACKEND_HPP
+#define BINSRV_BASIC_STORAGE_BACKEND_HPP
+
+#include "binsrv/basic_storage_backend_fwd.hpp" // IWYU pragma: export
+
+#include <string>
+#include <string_view>
+
+#include "util/byte_span_fwd.hpp"
+
+namespace binsrv {
+
+class basic_storage_backend {
+public:
+  basic_storage_backend() = default;
+  basic_storage_backend(const basic_storage_backend &) = delete;
+  basic_storage_backend(basic_storage_backend &&) noexcept = delete;
+  basic_storage_backend &operator=(const basic_storage_backend &) = delete;
+  basic_storage_backend &operator=(basic_storage_backend &&) = delete;
+
+  virtual ~basic_storage_backend() = default;
+
+  [[nodiscard]] storage_object_name_container list_objects();
+  [[nodiscard]] std::string get_object(std::string_view name);
+  void put_object(std::string_view name, util::const_byte_span content);
+
+  void open_stream(std::string_view name);
+  void write_data_to_stream(util::const_byte_span data);
+  void close_stream();
+
+private:
+  bool stream_opened_{false};
+
+  [[nodiscard]] virtual storage_object_name_container do_list_objects() = 0;
+  [[nodiscard]] virtual std::string do_get_object(std::string_view name) = 0;
+  virtual void do_put_object(std::string_view name,
+                             util::const_byte_span content) = 0;
+
+  virtual void do_open_stream(std::string_view name) = 0;
+  virtual void do_write_data_to_stream(util::const_byte_span data) = 0;
+  virtual void do_close_stream() = 0;
+};
+
+} // namespace binsrv
+
+#endif // BINSRV_BASIC_STORAGE_BACKEND_HPP
diff --git a/src/binsrv/basic_storage_backend_fwd.hpp b/src/binsrv/basic_storage_backend_fwd.hpp
new file mode 100644
index 0000000..391d84f
--- /dev/null
+++ b/src/binsrv/basic_storage_backend_fwd.hpp
@@ -0,0 +1,38 @@
+#ifndef BINSRV_BASIC_STORAGE_BACKEND_FWD_HPP
+#define BINSRV_BASIC_STORAGE_BACKEND_FWD_HPP
+
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <unordered_map>
+
+namespace binsrv {
+
+class basic_storage_backend;
+
+using basic_storage_backend_ptr = std::unique_ptr<basic_storage_backend>;
+
+// the following hash calculation helper allows to look for keys represented
+// by either const char*, std::string_view of const std::string in unordered
+// containers with std::string key transparently without converting the value
+// being searched to std::string
+struct transparent_string_like_hash {
+  using is_transparent = void;
+  std::size_t operator()(const char *key) const noexcept {
+    return std::hash<std::string_view>{}(key);
+  }
+  std::size_t operator()(std::string_view key) const noexcept {
+    return std::hash<std::string_view>{}(key);
+  }
+  std::size_t operator()(const std::string &key) const noexcept {
+    return std::hash<std::string>{}(key);
+  }
+};
+// the container thatr uses transparent_string_like_hash also needs a
+// transparent version of KeyEqual template argument (std::equal_to<void>)
+using storage_object_name_container =
+    std::unordered_map<std::string, std::uint64_t, transparent_string_like_hash,
+                       std::equal_to<>>;
+} // namespace binsrv
+
+#endif // BINSRV_BASIC_STORAGE_BACKEND_FWD_HPP
diff --git a/src/binsrv/event/protocol_traits_fwd.hpp b/src/binsrv/event/protocol_traits_fwd.hpp
index 4ce9dd4..7e5a594 100644
--- a/src/binsrv/event/protocol_traits_fwd.hpp
+++ b/src/binsrv/event/protocol_traits_fwd.hpp
@@ -1,6 +1,7 @@
 #ifndef BINSRV_EVENT_PROTOCOL_TRAITS_FWD_HPP
 #define BINSRV_EVENT_PROTOCOL_TRAITS_FWD_HPP
 
+#include <array>
 #include <cstddef>
 #include <cstdint>
 #include <limits>
@@ -17,7 +18,8 @@ inline constexpr std::size_t unspecified_post_header_length{
 
 // https://github.com/mysql/mysql-server/blob/trunk/sql/log_event.h#L211
 // 4 bytes which all binlogs should begin with
-inline constexpr std::string_view magic_binlog_payload{"\xFE\x62\x69\x6E"};
+inline constexpr std::array<std::byte, 4> magic_binlog_payload{
+    std::byte{0xFE}, std::byte{0x62}, std::byte{0x69}, std::byte{0x6E}};
 inline constexpr std::uint64_t magic_binlog_offset{4ULL};
 
 } // namespace binsrv::event
diff --git a/src/binsrv/file_logger.cpp b/src/binsrv/file_logger.cpp
index 35d60e3..2b63119 100644
--- a/src/binsrv/file_logger.cpp
+++ b/src/binsrv/file_logger.cpp
@@ -14,7 +14,7 @@ namespace binsrv {
 file_logger::file_logger(log_severity min_level, std::string_view file_name)
     : basic_logger{min_level}, stream_{std::filesystem::path{file_name}} {
   if (!stream_.is_open()) {
-    util::exception_location().raise<std::invalid_argument>(
+    util::exception_location().raise<std::runtime_error>(
         "unable to create \"" + std::string(file_name) + "\" file for logging");
   }
 }
diff --git a/src/binsrv/filesystem_storage.cpp b/src/binsrv/filesystem_storage.cpp
deleted file mode 100644
index 7ef5c16..0000000
--- a/src/binsrv/filesystem_storage.cpp
+++ /dev/null
@@ -1,222 +0,0 @@
-#include "binsrv/filesystem_storage.hpp"
-
-#include <algorithm>
-#include <cstddef>
-#include <filesystem>
-#include <fstream>
-#include <ios>
-#include <stdexcept>
-#include <string>
-#include <string_view>
-#include <utility>
-
-#include "binsrv/event/protocol_traits_fwd.hpp"
-
-#include "util/byte_span.hpp"
-#include "util/exception_location_helpers.hpp"
-
-namespace binsrv {
-
-filesystem_storage::filesystem_storage(std::string_view root_path)
-    : root_path_{root_path}, binlog_names_{}, ofs_{} {
-  // TODO: switch to utf8 file names
-  if (!std::filesystem::exists(root_path_)) {
-    util::exception_location().raise<std::invalid_argument>(
-        "root path does not exist");
-  }
-  if (!std::filesystem::is_directory(root_path_)) {
-    util::exception_location().raise<std::invalid_argument>(
-        "root path is not a directory");
-  }
-
-  const auto index_path = get_index_path();
-  if (!std::filesystem::exists(index_path)) {
-    if (!std::filesystem::is_empty(root_path_)) {
-      util::exception_location().raise<std::invalid_argument>(
-          "root path directory does not contain the binlog index file and is "
-          "not empty");
-    }
-  } else {
-    load_binlog_index(index_path);
-  }
-  if (!binlog_names_.empty()) {
-    position_ = std::filesystem::file_size(root_path_ / binlog_names_.back());
-  }
-}
-
-[[nodiscard]] bool
-filesystem_storage::check_binlog_name(std::string_view binlog_name) noexcept {
-  // TODO: parse binlog name into "base name" and "rotation number"
-  //       e.g. "binlog.000001" -> ("binlog", 1)
-
-  // currently checking only that the name does not include a filesystem
-  // separator
-  return binlog_name.find(std::filesystem::path::preferred_separator) ==
-         std::string_view::npos;
-}
-
-void filesystem_storage::open_binlog(std::string_view binlog_name) {
-  if (ofs_.is_open()) {
-    util::exception_location().raise<std::logic_error>(
-        "cannot create a binlog as the previous one has noot been closed");
-  }
-  if (!check_binlog_name(binlog_name)) {
-    util::exception_location().raise<std::invalid_argument>(
-        "cannot create a binlog with invalid file name");
-  }
-
-  std::filesystem::path current_file_path{root_path_};
-  current_file_path /= binlog_name;
-
-  const bool is_appending{position_ != 0ULL};
-  // opening in binary mode with appending either appending or truncating
-  // depending on whether the storage was initialized on a non-empty
-  // directory or not
-  const auto mode{std::ios_base::binary |
-                  (is_appending ? std::ios_base::app : std::ios_base::trunc)};
-  ofs_.open(current_file_path, mode);
-  if (!ofs_.is_open()) {
-    util::exception_location().raise<std::invalid_argument>(
-        "cannot create a binlog file");
-  }
-  // writing the magic binlog footprint only if this is a newly
-  // created file
-  if (!is_appending) {
-    if (!ofs_.write(std::data(event::magic_binlog_payload),
-                    static_cast<std::streamoff>(
-                        std::size(event::magic_binlog_payload)))) {
-      util::exception_location().raise<std::invalid_argument>(
-          "cannot write magic payload to the binlog");
-    }
-  }
-
-  if (!is_appending) {
-    binlog_names_.emplace_back(binlog_name);
-    append_to_binlog_index(binlog_name);
-    position_ = event::magic_binlog_offset;
-  }
-}
-
-void filesystem_storage::write_event(util::const_byte_span event_data) {
-  if (!ofs_.is_open()) {
-    util::exception_location().raise<std::logic_error>(
-        "cannot write to the binlog file as it has not been opened");
-  }
-  const auto event_data_sv = util::as_string_view(event_data);
-  if (!ofs_.write(std::data(event_data_sv),
-                  static_cast<std::streamoff>(std::size(event_data_sv)))) {
-    util::exception_location().raise<std::invalid_argument>(
-        "cannot write event data to the binlog");
-  }
-  position_ += std::size(event_data);
-  // TODO: make sure that the data is properly written to the disk
-  //       use fsync() system call here
-}
-void filesystem_storage::close_binlog() {
-  if (!ofs_.is_open()) {
-    util::exception_location().raise<std::logic_error>(
-        "cannot close the binlog file as it has not been opened");
-  }
-  ofs_.close();
-  position_ = 0ULL;
-}
-
-[[nodiscard]] std::filesystem::path filesystem_storage::get_index_path() const {
-  auto result{root_path_};
-  result /= default_binlog_index_name;
-  return result;
-}
-
-void filesystem_storage::load_binlog_index(
-    const std::filesystem::path &index_path) {
-  if (!std::filesystem::is_regular_file(index_path)) {
-    util::exception_location().raise<std::invalid_argument>(
-        "the binlog index file is not a regular file");
-  }
-  // opening in text mode
-  std::ifstream index_ifs{index_path};
-  if (!index_ifs.is_open()) {
-    util::exception_location().raise<std::invalid_argument>(
-        "cannot open the binlog index file for reading");
-  }
-  std::string current_line;
-  while (std::getline(index_ifs, current_line)) {
-    if (current_line.empty()) {
-      continue;
-    }
-    const std::filesystem::path current_binlog_path{current_line};
-    if (current_binlog_path.parent_path() != default_binlog_index_entry_path) {
-      util::exception_location().raise<std::invalid_argument>(
-          "binlog index contains an entry that has an invalid path");
-    }
-    auto current_binlog_name{current_binlog_path.filename().string()};
-
-    if (current_binlog_name == default_binlog_index_name) {
-      util::exception_location().raise<std::invalid_argument>(
-          "binlog index contains a reference to the binlog index file name");
-    }
-    if (!check_binlog_name(current_binlog_name)) {
-      util::exception_location().raise<std::invalid_argument>(
-          "binlog index contains a reference a binlog file with invalid "
-          "name");
-    }
-    if (std::ranges::find(binlog_names_, current_binlog_name) !=
-        std::end(binlog_names_)) {
-      util::exception_location().raise<std::invalid_argument>(
-          "binlog index contains a duplicate entry");
-    }
-    binlog_names_.emplace_back(std::move(current_binlog_name));
-  }
-
-  std::size_t known_entries{0U};
-  for (auto const &dir_entry :
-       std::filesystem::directory_iterator{root_path_}) {
-    if (!dir_entry.is_regular_file()) {
-      util::exception_location().raise<std::invalid_argument>(
-          "filesystem storage directory contains an entry that is not a "
-          "regular file");
-    }
-    // TODO: check permissions here
-    const auto &dir_entry_path = dir_entry.path();
-    const auto dir_entry_filename = dir_entry_path.filename().string();
-
-    if (dir_entry_filename == default_binlog_index_name) {
-      continue;
-    }
-    if (std::ranges::find(binlog_names_, dir_entry_filename) ==
-        std::end(binlog_names_)) {
-      util::exception_location().raise<std::invalid_argument>(
-          "filesystem storage directory contains an entry that is not "
-          "referenced in the binlog index file");
-    }
-    ++known_entries;
-  }
-
-  if (known_entries != binlog_names_.size()) {
-    util::exception_location().raise<std::invalid_argument>(
-        "binlog index contains a reference to a non-existing file");
-  }
-
-  // TODO: add integrity checks (parsing + checksumming) for the binlog
-  //       files in the index
-}
-
-void filesystem_storage::append_to_binlog_index(std::string_view binlog_name) {
-  const auto index_path = get_index_path();
-  // opening in text mode with appending
-  std::ofstream index_ofs{index_path, std::ios_base::app};
-  if (!index_ofs.is_open()) {
-    util::exception_location().raise<std::invalid_argument>(
-        "cannot open binlog index file for updating");
-  }
-  std::filesystem::path binlog_path{default_binlog_index_entry_path};
-  binlog_path /= binlog_name;
-
-  index_ofs << binlog_path.generic_string() << '\n';
-  if (!index_ofs) {
-    util::exception_location().raise<std::invalid_argument>(
-        "cannot append to the binlog index file");
-  }
-}
-
-} // namespace binsrv
diff --git a/src/binsrv/filesystem_storage_backend.cpp b/src/binsrv/filesystem_storage_backend.cpp
new file mode 100644
index 0000000..9ab2f0c
--- /dev/null
+++ b/src/binsrv/filesystem_storage_backend.cpp
@@ -0,0 +1,127 @@
+#include "binsrv/filesystem_storage_backend.hpp"
+
+#include <cstddef>
+#include <filesystem>
+#include <fstream>
+#include <ios>
+#include <iterator>
+#include <stdexcept>
+#include <string>
+#include <string_view>
+
+// TODO: remove this include when switching to clang-18 where transitive
+//       IWYU 'export' pragmas are handled properly
+#include "binsrv/basic_storage_backend_fwd.hpp"
+
+#include "util/byte_span.hpp"
+#include "util/exception_location_helpers.hpp"
+
+namespace binsrv {
+
+filesystem_storage_backend::filesystem_storage_backend(
+    std::string_view root_path)
+    : root_path_{root_path}, ofs_{} {
+  // TODO: switch to utf8 file names
+  if (!std::filesystem::exists(root_path_)) {
+    util::exception_location().raise<std::invalid_argument>(
+        "root path does not exist");
+  }
+  if (!std::filesystem::is_directory(root_path_)) {
+    util::exception_location().raise<std::invalid_argument>(
+        "root path is not a directory");
+  }
+}
+
+[[nodiscard]] storage_object_name_container
+filesystem_storage_backend::do_list_objects() {
+  storage_object_name_container result;
+  for (auto const &dir_entry :
+       std::filesystem::directory_iterator{root_path_}) {
+    if (!dir_entry.is_regular_file()) {
+      util::exception_location().raise<std::logic_error>(
+          "filesystem storage directory contains an entry that is not a "
+          "regular file");
+    }
+    // TODO: check permissions here
+    result.emplace(dir_entry.path().filename().string(), dir_entry.file_size());
+  }
+  return result;
+}
+
+[[nodiscard]] std::string
+filesystem_storage_backend::do_get_object(std::string_view name) {
+  const auto index_path{get_object_path(name)};
+
+  static constexpr std::size_t max_file_size{1048576U};
+
+  // opening in binary mode
+  std::ifstream ifs{index_path, std::ios_base::binary};
+  if (!ifs.is_open()) {
+    util::exception_location().raise<std::runtime_error>(
+        "cannot open undellying object file");
+  }
+  auto file_size = std::filesystem::file_size(index_path);
+  if (file_size > max_file_size) {
+    util::exception_location().raise<std::out_of_range>(
+        "undellying object file is too large");
+  }
+
+  std::string file_content(file_size, 'x');
+  if (!ifs.read(std::data(file_content),
+                static_cast<std::streamoff>(file_size))) {
+    util::exception_location().raise<std::runtime_error>(
+        "cannot read undellying object file content");
+  }
+  return file_content;
+}
+
+void filesystem_storage_backend::do_put_object(std::string_view name,
+                                               util::const_byte_span content) {
+  const auto index_path = get_object_path(name);
+  // opening in binary mode with truncating
+  std::ofstream index_ofs{index_path, std::ios_base::trunc};
+  if (!index_ofs.is_open()) {
+    util::exception_location().raise<std::runtime_error>(
+        "cannot open undellying object file for writing");
+  }
+  const auto content_sv = util::as_string_view(content);
+  if (!index_ofs.write(std::data(content_sv),
+                       static_cast<std::streamoff>(std::size(content_sv)))) {
+    util::exception_location().raise<std::runtime_error>(
+        "cannot write date to undellying object file");
+  }
+}
+
+void filesystem_storage_backend::do_open_stream(std::string_view name) {
+  std::filesystem::path current_file_path{root_path_};
+  current_file_path /= name;
+
+  ofs_.open(current_file_path, std::ios_base::binary | std::ios_base::app);
+  if (!ofs_.is_open()) {
+    util::exception_location().raise<std::runtime_error>(
+        "cannot open underlying file for the stream");
+  }
+}
+
+void filesystem_storage_backend::do_write_data_to_stream(
+    util::const_byte_span data) {
+  const auto data_sv = util::as_string_view(data);
+  if (!ofs_.write(std::data(data_sv),
+                  static_cast<std::streamoff>(std::size(data_sv)))) {
+    util::exception_location().raise<std::runtime_error>(
+        "cannot write data to the underlying stream file");
+  }
+  // TODO: make sure that the data is properly written to the disk
+  //       use fsync() system call here
+}
+
+void filesystem_storage_backend::do_close_stream() { ofs_.close(); }
+
+[[nodiscard]] std::filesystem::path
+filesystem_storage_backend::get_object_path(std::string_view name) const {
+  auto result{root_path_};
+  result /= name;
+  return result;
+}
+
+} // namespace binsrv
diff --git a/src/binsrv/filesystem_storage_backend.hpp b/src/binsrv/filesystem_storage_backend.hpp
new file mode 100644
index 0000000..91d3189
--- /dev/null
+++ b/src/binsrv/filesystem_storage_backend.hpp
@@ -0,0 +1,40 @@
+#ifndef BINSRV_FILESYSTEM_STORAGE_BACKEND_HPP
+#define BINSRV_FILESYSTEM_STORAGE_BACKEND_HPP
+
+#include <filesystem>
+#include <fstream>
+#include <string_view>
+
+#include "binsrv/basic_storage_backend.hpp" // IWYU pragma: export
+
+namespace binsrv {
+
+class [[nodiscard]] filesystem_storage_backend : public basic_storage_backend {
+public:
+  explicit filesystem_storage_backend(std::string_view root_path);
+
+  [[nodiscard]] const std::filesystem::path &get_root_path() const noexcept {
+    return root_path_;
+  }
+
+private:
+  std::filesystem::path root_path_;
+  std::ofstream ofs_;
+
+  [[nodiscard]] storage_object_name_container do_list_objects() override;
+
+  [[nodiscard]] std::string do_get_object(std::string_view name) override;
+  void do_put_object(std::string_view name,
+                     util::const_byte_span content) override;
+
+  void do_open_stream(std::string_view name) override;
+  void do_write_data_to_stream(util::const_byte_span data) override;
+  void do_close_stream() override;
+
+  [[nodiscard]] std::filesystem::path
+  get_object_path(std::string_view name) const;
+};
+
+} // namespace binsrv
+
+#endif // BINSRV_FILESYSTEM_STORAGE_BACKEND_HPP
diff --git a/src/binsrv/filesystem_storage_fwd.hpp b/src/binsrv/filesystem_storage_fwd.hpp
deleted file mode 100644
index 5f48a86..0000000
--- a/src/binsrv/filesystem_storage_fwd.hpp
+++ /dev/null
@@ -1,10 +0,0 @@
-#ifndef BINSRV_FILESYSTEM_STORAGE_FWD_HPP
-#define BINSRV_FILESYSTEM_STORAGE_FWD_HPP
-
-namespace binsrv {
-
-class filesystem_storage;
-
-} // namespace binsrv
-
-#endif // BINSRV_FILESYSTEM_STORAGE_FWD_HPP
diff --git a/src/binsrv/master_config.cpp b/src/binsrv/main_config.cpp
similarity index 76%
rename from src/binsrv/master_config.cpp
rename to src/binsrv/main_config.cpp
index f22b4e9..222a1f4 100644
--- a/src/binsrv/master_config.cpp
+++ b/src/binsrv/main_config.cpp
@@ -1,4 +1,4 @@
-#include "binsrv/master_config.hpp"
+#include "binsrv/main_config.hpp"
 
 #include <cstddef>
 #include <filesystem>
@@ -21,33 +21,33 @@
 
 namespace binsrv {
 
-master_config::master_config(util::command_line_arg_view arguments) {
+main_config::main_config(util::command_line_arg_view arguments) {
   util::nv_tuple_from_command_line(arguments, impl_);
 }
 
-master_config::master_config(std::string_view file_name) {
+main_config::main_config(std::string_view file_name) {
   static constexpr std::size_t max_file_size{1048576U};
 
   const std::filesystem::path file_path{file_name};
   std::ifstream ifs{file_path};
   if (!ifs.is_open()) {
-    util::exception_location().raise<std::invalid_argument>(
+    util::exception_location().raise<std::runtime_error>(
         "cannot open configuration file");
   }
   auto file_size = std::filesystem::file_size(file_path);
   if (file_size == 0ULL) {
-    util::exception_location().raise<std::invalid_argument>(
+    util::exception_location().raise<std::out_of_range>(
         "configuration file is empty");
   }
   if (file_size > max_file_size) {
-    util::exception_location().raise<std::invalid_argument>(
+    util::exception_location().raise<std::out_of_range>(
         "configuration file is too large");
   }
 
   std::string file_content(file_size, 'x');
   if (!ifs.read(std::data(file_content),
                 static_cast<std::streamoff>(file_size))) {
-    util::exception_location().raise<std::invalid_argument>(
+    util::exception_location().raise<std::runtime_error>(
         "cannot read configuration file content");
   }
 
diff --git a/src/binsrv/master_config.hpp b/src/binsrv/main_config.hpp
similarity index 66%
rename from src/binsrv/master_config.hpp
rename to src/binsrv/main_config.hpp
index d7f7299..2b0a790 100644
--- a/src/binsrv/master_config.hpp
+++ b/src/binsrv/main_config.hpp
@@ -1,9 +1,9 @@
-#ifndef BINSRV_MASTER_CONFIG_HPP
-#define BINSRV_MASTER_CONFIG_HPP
+#ifndef BINSRV_MAIN_CONFIG_HPP
+#define BINSRV_MAIN_CONFIG_HPP
 
-#include "binsrv/master_config_fwd.hpp" // IWYU pragma: export
+#include "binsrv/main_config_fwd.hpp" // IWYU pragma: export
 
-#include "binsrv/logger_config.hpp" // IWYU pragma: export
+#include "binsrv/logger_config.hpp"  // IWYU pragma: export
 #include "binsrv/storage_config.hpp" // IWYU pragma: export
 
 #include "easymysql/connection_config.hpp" // IWYU pragma: export
@@ -13,7 +13,7 @@
 
 namespace binsrv {
 
-class [[nodiscard]] master_config {
+class [[nodiscard]] main_config {
 private:
   using impl_type = util::nv_tuple<
       // clang-format off
@@ -27,9 +27,9 @@ class [[nodiscard]] master_config {
   static constexpr std::size_t flattened_size{impl_type::flattened_size};
   static constexpr std::size_t depth{impl_type::depth};
 
-  explicit master_config(std::string_view file_name);
+  explicit main_config(std::string_view file_name);
 
-  explicit master_config(util::command_line_arg_view arguments);
+  explicit main_config(util::command_line_arg_view arguments);
 
   [[nodiscard]] const auto &root() const noexcept { return impl_; }
 
@@ -39,4 +39,4 @@ class [[nodiscard]] master_config {
 
 } // namespace binsrv
 
-#endif // BINSRV_MASTER_CONFIG_HPP
+#endif // BINSRV_MAIN_CONFIG_HPP
diff --git a/src/binsrv/main_config_fwd.hpp b/src/binsrv/main_config_fwd.hpp
new file mode 100644
index 0000000..c5d7a55
--- /dev/null
+++ b/src/binsrv/main_config_fwd.hpp
@@ -0,0 +1,14 @@
+#ifndef BINSRV_MAIN_CONFIG_FWD_HPP
+#define BINSRV_MAIN_CONFIG_FWD_HPP
+
+#include <memory>
+
+namespace binsrv {
+
+class main_config;
+
+using main_config_ptr = std::shared_ptr<main_config>;
+
+} // namespace binsrv
+
+#endif // BINSRV_MAIN_CONFIG_FWD_HPP
diff --git a/src/binsrv/master_config_fwd.hpp b/src/binsrv/master_config_fwd.hpp
deleted file mode 100644
index 4b04680..0000000
--- a/src/binsrv/master_config_fwd.hpp
+++ /dev/null
@@ -1,14 +0,0 @@
-#ifndef BINSRV_MASTER_CONFIG_FWD_HPP
-#define BINSRV_MASTER_CONFIG_FWD_HPP
-
-#include <memory>
-
-namespace binsrv {
-
-class master_config;
-
-using master_config_ptr = std::shared_ptr<master_config>;
-
-} // namespace binsrv
-
-#endif // BINSRV_MASTER_CONFIG_FWD_HPP
diff --git a/src/binsrv/storage.cpp b/src/binsrv/storage.cpp
new file mode 100644
index 0000000..aad66c8
--- /dev/null
+++ b/src/binsrv/storage.cpp
@@ -0,0 +1,155 @@
+#include "binsrv/storage.hpp"
+
+#include <algorithm>
+#include <cstddef>
+#include <filesystem>
+#include <sstream>
+#include <stdexcept>
+#include <string>
+#include <string_view>
+#include <utility>
+
+#include "binsrv/basic_storage_backend.hpp"
+
+#include "binsrv/event/protocol_traits_fwd.hpp"
+
+#include "util/byte_span.hpp"
+#include "util/exception_location_helpers.hpp"
+
+namespace binsrv {
+
+storage::storage(basic_storage_backend_ptr backend)
+    : backend_{std::move(backend)}, binlog_names_{} {
+  const auto storage_objects{backend_->list_objects()};
+  if (storage_objects.empty()) {
+    // initialized on a new / empty storage - no other actions required
+    return;
+  }
+
+  if (!storage_objects.contains(default_binlog_index_name)) {
+    util::exception_location().raise<std::logic_error>(
+        "storage is not empty but does not contain binlog index");
+  }
+
+  load_binlog_index();
+  validate_binlog_index(storage_objects);
+
+  if (!binlog_names_.empty()) {
+    // call to validate_binlog_index() guarantees that the name will be
+    // found here
+    position_ = storage_objects.at(binlog_names_.back());
+  }
+}
+
+[[nodiscard]] bool
+storage::check_binlog_name(std::string_view binlog_name) noexcept {
+  // TODO: parse binlog name into "base name" and "rotation number"
+  //       e.g. "binlog.000001" -> ("binlog", 1)
+
+  // currently checking only that the name does not include a filesystem
+  // separator
+  return binlog_name.find(std::filesystem::path::preferred_separator) ==
+         std::string_view::npos;
+}
+
+void storage::open_binlog(std::string_view binlog_name) {
+  if (!check_binlog_name(binlog_name)) {
+    util::exception_location().raise<std::logic_error>(
+        "cannot create a binlog with invalid name");
+  }
+
+  backend_->open_stream(binlog_name);
+
+  if (position_ == 0ULL) {
+    // writing the magic binlog footprint only if this is a newly
+    // created file
+    backend_->write_data_to_stream(event::magic_binlog_payload);
+
+    binlog_names_.emplace_back(binlog_name);
+    save_binlog_index();
+    position_ = event::magic_binlog_offset;
+  }
+}
+
+void storage::write_event(util::const_byte_span event_data) {
+  backend_->write_data_to_stream(event_data);
+  position_ += std::size(event_data);
+}
+
+void storage::close_binlog() {
+  backend_->close_stream();
+  position_ = 0ULL;
+}
+
+void storage::load_binlog_index() {
+  const auto index_content{backend_->get_object(default_binlog_index_name)};
+  // opening in text mode
+  std::istringstream index_iss{index_content};
+  std::string current_line;
+  while (std::getline(index_iss, current_line)) {
+    if (current_line.empty()) {
+      continue;
+    }
+    const std::filesystem::path current_binlog_path{current_line};
+    if (current_binlog_path.parent_path() != default_binlog_index_entry_path) {
+      util::exception_location().raise<std::logic_error>(
+          "binlog index contains an entry that has an invalid path");
+    }
+    auto current_binlog_name{current_binlog_path.filename().string()};
+
+    if (current_binlog_name == default_binlog_index_name) {
+      util::exception_location().raise<std::logic_error>(
+          "binlog index contains a reference to the binlog index name");
+    }
+    if (!check_binlog_name(current_binlog_name)) {
+      util::exception_location().raise<std::logic_error>(
+          "binlog index contains a reference to a binlog with invalid "
+          "name");
+    }
+    if (std::ranges::find(binlog_names_, current_binlog_name) !=
+        std::end(binlog_names_)) {
+      util::exception_location().raise<std::logic_error>(
+          "binlog index contains a duplicate entry");
+    }
+    binlog_names_.emplace_back(std::move(current_binlog_name));
+  }
+}
+
+void storage::validate_binlog_index(
+    const storage_object_name_container &object_names) {
+  std::size_t known_entries{0U};
+  for (auto const &[object_name, object_size] : object_names) {
+    if (object_name == default_binlog_index_name) {
+      continue;
+    }
+    if (std::ranges::find(binlog_names_, object_name) ==
+        std::end(binlog_names_)) {
+      util::exception_location().raise<std::logic_error>(
+          "storage contains an object that is not "
+          "referenced in the binlog index");
+    }
+    ++known_entries;
+  }
+
+  if (known_entries != std::size(binlog_names_)) {
+    util::exception_location().raise<std::logic_error>(
+        "binlog index contains a reference to a non-existing object");
+  }
+
+  // TODO: add integrity checks (parsing + checksumming) for the binlog
+  //       files in the index
+}
+
+void storage::save_binlog_index() {
+  std::ostringstream oss;
+  for (const auto &binlog_name : binlog_names_) {
+    std::filesystem::path binlog_path{default_binlog_index_entry_path};
+    binlog_path /= binlog_name;
+    oss << binlog_path.generic_string() << '\n';
+  }
+  const auto content{oss.str()};
+  backend_->put_object(default_binlog_index_name,
+                       util::as_const_byte_span(content));
+}
+
+} // namespace binsrv
diff --git a/src/binsrv/filesystem_storage.hpp b/src/binsrv/storage.hpp
similarity index 50%
rename from src/binsrv/filesystem_storage.hpp
rename to src/binsrv/storage.hpp
index 57cd602..a0ba71f 100644
--- a/src/binsrv/filesystem_storage.hpp
+++ b/src/binsrv/storage.hpp
@@ -1,36 +1,34 @@
-#ifndef BINSRV_FILESYSTEM_STORAGE_HPP
-#define BINSRV_FILESYSTEM_STORAGE_HPP
+#ifndef BINSRV_STORAGE_HPP
+#define BINSRV_STORAGE_HPP
 
-#include "binsrv/filesystem_storage_fwd.hpp" // IWYU pragma: export
+#include "binsrv/storage_fwd.hpp" // IWYU pragma: export
 
-#include <filesystem>
-#include <fstream>
 #include <string>
 #include <string_view>
 #include <vector>
 
+#include "binsrv/basic_storage_backend_fwd.hpp"
+
 #include "util/byte_span_fwd.hpp"
 
 namespace binsrv {
 
-class [[nodiscard]] filesystem_storage {
+class [[nodiscard]] storage {
 public:
   static constexpr std::string_view default_binlog_index_name{"binlog.index"};
   static constexpr std::string_view default_binlog_index_entry_path{"."};
 
-  explicit filesystem_storage(std::string_view root_path);
+  // passing by value as we are going to move from this unique_ptr
+  explicit storage(basic_storage_backend_ptr backend);
 
-  filesystem_storage(const filesystem_storage &) = delete;
-  filesystem_storage &operator=(const filesystem_storage &) = delete;
-  filesystem_storage(filesystem_storage &&) = delete;
-  filesystem_storage &operator=(filesystem_storage &&) = delete;
+  storage(const storage &) = delete;
+  storage &operator=(const storage &) = delete;
+  storage(storage &&) = delete;
+  storage &operator=(storage &&) = delete;
 
   // desctuctor is explicitly defined as default here to complete the rule of 5
-  ~filesystem_storage() = default;
+  ~storage() = default;
 
-  [[nodiscard]] const std::filesystem::path &get_root_path() const noexcept {
-    return root_path_;
-  }
   [[nodiscard]] std::string_view get_binlog_name() const noexcept {
     return binlog_names_.empty() ? std::string_view{} : binlog_names_.back();
   }
@@ -46,18 +44,17 @@ class [[nodiscard]] filesystem_storage {
   void close_binlog();
 
 private:
-  std::filesystem::path root_path_;
+  basic_storage_backend_ptr backend_;
 
   using binlog_name_container = std::vector<std::string>;
   binlog_name_container binlog_names_;
-  std::ofstream ofs_;
   std::uint64_t position_{0ULL};
 
-  [[nodiscard]] std::filesystem::path get_index_path() const;
-  void load_binlog_index(const std::filesystem::path &index_path);
-  void append_to_binlog_index(std::string_view binlog_name);
+  void load_binlog_index();
+  void validate_binlog_index(const storage_object_name_container &object_names);
+  void save_binlog_index();
 };
 
 } // namespace binsrv
 
-#endif // BINSRV_FILESYSTEM_STORAGE_HPP
+#endif // BINSRV_STORAGE_HPP
diff --git a/src/binsrv/storage_backend_factory.cpp b/src/binsrv/storage_backend_factory.cpp
new file mode 100644
index 0000000..94586d5
--- /dev/null
+++ b/src/binsrv/storage_backend_factory.cpp
@@ -0,0 +1,24 @@
+#include "binsrv/storage_backend_factory.hpp"
+
+#include <memory>
+#include <stdexcept>
+
+#include "binsrv/basic_storage_backend_fwd.hpp"
+#include "binsrv/filesystem_storage_backend.hpp"
+#include "binsrv/storage_config.hpp"
+
+#include "util/exception_location_helpers.hpp"
+
+namespace binsrv {
+
+basic_storage_backend_ptr
+storage_backend_factory::create(const storage_config &config) {
+  const auto &storage_backend_type = config.get<"type">();
+  if (storage_backend_type != "fs") {
+    util::exception_location().raise<std::invalid_argument>(
+        "unknown storage backend type");
+  }
+  return std::make_unique<filesystem_storage_backend>(config.get<"path">());
+}
+
+} // namespace binsrv
diff --git a/src/binsrv/storage_backend_factory.hpp b/src/binsrv/storage_backend_factory.hpp
new file mode 100644
index 0000000..d441377
--- /dev/null
+++ b/src/binsrv/storage_backend_factory.hpp
@@ -0,0 +1,16 @@
+#ifndef BINSRV_STORAGE_BACKEND_FACTORY_HPP
+#define BINSRV_STORAGE_BACKEND_FACTORY_HPP
+
+#include "binsrv/basic_storage_backend_fwd.hpp"
+#include "binsrv/storage_config_fwd.hpp"
+
+namespace binsrv {
+
+struct [[nodiscard]] storage_backend_factory {
+  [[nodiscard]] static basic_storage_backend_ptr
+  create(const storage_config &config);
+};
+
+} // namespace binsrv
+
+#endif // BINSRV_STORAGE_BACKEND_FACTORY_HPP
diff --git a/src/binsrv/storage_config.hpp b/src/binsrv/storage_config.hpp
index 26f7ffd..b063b3e 100644
--- a/src/binsrv/storage_config.hpp
+++ b/src/binsrv/storage_config.hpp
@@ -12,6 +12,7 @@ namespace binsrv {
 // clang-format off
 struct [[nodiscard]] storage_config
     : util::nv_tuple<
+          util::nv<"type", std::string>,
           util::nv<"path", std::string>
       > {};
 // clang-format on
diff --git a/src/binsrv/storage_fwd.hpp b/src/binsrv/storage_fwd.hpp
new file mode 100644
index 0000000..f633e4f
--- /dev/null
+++ b/src/binsrv/storage_fwd.hpp
@@ -0,0 +1,10 @@
+#ifndef BINSRV_STORAGE_FWD_HPP
+#define BINSRV_STORAGE_FWD_HPP
+
+namespace binsrv {
+
+class storage;
+
+} // namespace binsrv
+
+#endif // BINSRV_STORAGE_FWD_HPP
diff --git a/src/easymysql/binlog.cpp b/src/easymysql/binlog.cpp
index be0c7af..cbec59a 100644
--- a/src/easymysql/binlog.cpp
+++ b/src/easymysql/binlog.cpp
@@ -46,6 +46,8 @@ binlog::binlog(connection &conn, std::uint32_t server_id,
       }} {
   assert(!conn.is_empty());
 
+  // WL#2540: Replication event checksums
+  // https://dev.mysql.com/worklog/task/?id=2540
   static constexpr std::string_view crc_query{
       "SET @source_binlog_checksum = 'NONE', "
       "@master_binlog_checksum = 'NONE'"};
diff --git a/src/util/byte_span.hpp b/src/util/byte_span.hpp
index fdddab6..4680da7 100644
--- a/src/util/byte_span.hpp
+++ b/src/util/byte_span.hpp
@@ -18,6 +18,12 @@ inline std::string_view as_string_view(const_byte_span portion) noexcept {
           std::size(portion)};
 }
 
+inline const_byte_span as_const_byte_span(std::string_view portion) noexcept {
+  // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast)
+  return {reinterpret_cast<const std::byte *>(std::data(portion)),
+          std::size(portion)};
+}
+
 } // namespace util
 
 #endif // UTIL_BYTE_SPAN_HPP