ct/l0: Add object path utility for random prefix

Signed-off-by: Oren Leiman <oren.leiman@redpanda.com>

Author: oleiman

src/v/cloud_topics/object_utils.cc

@@ -83,4 +83,33 @@ object_path_factory::level_zero_path_to_epoch(std::string_view key) {
     return cluster_epoch(epoch);
 }
 
+std::expected<decltype(object_id::prefix), std::string>
+object_path_factory::level_zero_path_to_prefix(std::string_view key) {
+    // find the level zero prefix and chop it off
+    auto name = key;
+    auto it = name.find(level_zero_data_dir_str);
+    if (it == std::string_view::npos) {
+        return std::unexpected(
+          fmt::format("L0 object name missing prefix: {}", key));
+    }
+    name.remove_prefix(it + std::strlen(level_zero_data_dir_str));
+
+    if (name.size() < prefix_digits + 1) {
+        return std::unexpected(
+          fmt::format("L0 object name is too short: {}", key));
+    }
+    name = name.substr(0, prefix_digits);
+    name.remove_suffix(name.size() - prefix_digits);
+
+    // parse the prefix into a uint16_t
+    decltype(object_id::prefix) pfx{0};
+    auto res = std::from_chars(name.data(), name.data() + name.size(), pfx);
+    if (res.ptr != name.data() + name.size() || res.ec != std::errc{}) {
+        return std::unexpected(
+          fmt::format("L0 object name has invalid prefix: {}", key));
+    }
+
+    return pfx;
+}
+
 } // namespace cloud_topics

src/v/cloud_topics/object_utils.h

@@ -36,6 +36,9 @@ class object_path_factory {
      */
     static std::expected<cluster_epoch, std::string>
       level_zero_path_to_epoch(std::string_view);
+
+    static std::expected<decltype(object_id::prefix), std::string>
+      level_zero_path_to_prefix(std::string_view);
 };
 
 } // namespace cloud_topics

src/v/cloud_topics/tests/object_utils_test.cc

@@ -70,3 +70,35 @@ TEST(ObjectPathFactory, LevelZeroParseEpoch) {
       "L0 object name has invalid epoch: "
       "level_zero/data/000/00000X0000000010042/asdfasdf");
 }
+
+TEST(ObjectPathFactory, LevelZeroParsePrefix) {
+    EXPECT_EQ(
+      cloud_topics::object_path_factory::level_zero_path_to_prefix(
+        "level_zero/data/123/"),
+      123);
+
+    EXPECT_EQ(
+      cloud_topics::object_path_factory::level_zero_path_to_prefix(
+        "level_zero/data/123/000000000000010042/"),
+      123);
+
+    EXPECT_EQ(
+      cloud_topics::object_path_factory::level_zero_path_to_prefix(
+        "level_asdf_zero/data/000/000000000000010042/asdfasdf")
+        .error(),
+      "L0 object name missing prefix: "
+      "level_asdf_zero/data/000/000000000000010042/asdfasdf");
+
+    EXPECT_EQ(
+      cloud_topics::object_path_factory::level_zero_path_to_prefix(
+        "level_zero/data/00/")
+        .error(),
+      "L0 object name is too short: level_zero/data/00/");
+
+    EXPECT_EQ(
+      cloud_topics::object_path_factory::level_zero_path_to_prefix(
+        "level_zero/data/0X0/0000000000000010042/asdfasdf")
+        .error(),
+      "L0 object name has invalid prefix: "
+      "level_zero/data/0X0/0000000000000010042/asdfasdf");
+}