cloud_topics/reconciler: Use L1 io and object primitives

This alters the reconciler so it builds and uploads proper L1 objects,
using the l1::io and l1::object_builder abstractions.

A follow-up will change the commit and LRO offset portions of the
reconciler to use the metastore and the ctp_stm.

Author: wdberkeley

src/v/cloud_topics/BUILD

@@ -95,11 +95,13 @@ redpanda_cc_library(
         ":data_plane_api",
         "//src/v/base",
         "//src/v/cloud_topics:state_accessors",
+        "//src/v/cloud_topics/level_one/common:file_io",
         "//src/v/cloud_topics/level_one/domain:domain_supervisor",
         "//src/v/cloud_topics/level_one/metastore:frontend",
         "//src/v/cloud_topics/level_zero/common:extent_meta",
         "//src/v/cloud_topics/reconciler",
         "//src/v/cluster",
+        "//src/v/config",
         "//src/v/container:chunked_vector",
         "//src/v/model",
         "//src/v/ssx:sharded_service_container",

src/v/cloud_topics/app.cc

@@ -16,6 +16,7 @@
 #include "cloud_topics/manager/manager.h"
 #include "cluster/cluster_epoch_service.h"
 #include "cluster/controller.h"
+#include "config/node_config.h"
 #include "ssx/sharded_service_container.h"
 
 #include <seastar/core/coroutine.hh>
@@ -50,13 +51,26 @@ ss::future<> app::construct(
 
     co_await construct_service(state, data_plane.get());
 
+    // Touch the L1 staging directory before L1 i/o starts.
+    co_await ss::recursive_touch_directory(
+      config::node().l1_staging_path().string());
+
+    co_await construct_service(
+      l1_io,
+      config::node().l1_staging_path(),
+      ss::sharded_parameter([&remote] { return &remote->local(); }),
+      bucket,
+      ss::sharded_parameter([&cloud_cache] { return &cloud_cache->local(); }));
+
     co_await construct_service(
       reconciler,
       ss::sharded_parameter(
         [&partition_mgr] { return &partition_mgr->local(); }),
-      remote,
       data_plane.get(),
-      bucket);
+      ss::sharded_parameter([this] { return &l1_io.local(); }),
+      ss::sharded_parameter(
+        [&metadata_cache] { return &metadata_cache->local(); }));
+
     co_await construct_service(domain_supervisor, controller);
     co_await construct_service(
       l1_metastore_fe,

src/v/cloud_topics/app.h

@@ -10,6 +10,7 @@
 
 #pragma once
 
+#include "cloud_topics/level_one/common/file_io.h"
 #include "cloud_topics/level_one/domain/domain_supervisor.h"
 #include "cloud_topics/level_one/metastore/frontend.h"
 #include "cloud_topics/reconciler/reconciler.h"
@@ -72,6 +73,7 @@ class app : public ssx::sharded_service_container {
     ss::sstring _logger_name;
     std::unique_ptr<data_plane_api> data_plane;
     ss::sharded<state_accessors> state;
+    ss::sharded<l1::file_io> l1_io;
     ss::sharded<reconciler::reconciler> reconciler;
     ss::sharded<l1::domain_supervisor> domain_supervisor;
     ss::sharded<l1::frontend> l1_metastore_fe;

src/v/cloud_topics/level_one/common/BUILD

@@ -4,13 +4,21 @@ package(default_visibility = ["//src/v/cloud_topics/level_one:__subpackages__"])
 
 redpanda_cc_library(
     name = "object_id",
+    srcs = [
+        "object_id.cc",
+    ],
     hdrs = [
         "object_id.h",
     ],
+    visibility = [
+        "//src/v/cloud_topics/level_one:__subpackages__",
+        "//src/v/cloud_topics/reconciler:__subpackages__",
+    ],
     deps = [
         "//src/v/base",
         "//src/v/utils:named_type",
         "//src/v/utils:uuid",
+        "@fmt",
     ],
 )
 
@@ -67,6 +75,10 @@ redpanda_cc_library(
     name = "abstract_io",
     srcs = ["abstract_io.cc"],
     hdrs = ["abstract_io.h"],
+    visibility = [
+        "//src/v/cloud_topics/level_one:__subpackages__",
+        "//src/v/cloud_topics/reconciler:__subpackages__",
+    ],
     deps = [
         ":object_id",
         "//src/v/container:chunked_vector",
@@ -79,6 +91,11 @@ redpanda_cc_library(
     name = "file_io",
     srcs = ["file_io.cc"],
     hdrs = ["file_io.h"],
+    visibility = [
+        "//src/v/cloud_topics:__pkg__",
+        "//src/v/cloud_topics/level_one:__subpackages__",
+        "//src/v/cloud_topics/reconciler:__pkg__",
+    ],
     deps = [
         ":abstract_io",
         ":object_id",

src/v/cloud_topics/reconciler/BUILD

@@ -50,13 +50,14 @@ redpanda_cc_library(
     ],
     visibility = ["//visibility:public"],
     deps = [
-        ":range_batch_consumer",
+        ":reconciliation_consumer",
         "//src/v/base",
-        "//src/v/cloud_io:remote",
-        "//src/v/cloud_storage",
         "//src/v/cloud_storage_clients",
         "//src/v/cloud_topics:object_utils",
         "//src/v/cloud_topics:types",
+        "//src/v/cloud_topics/level_one/common:abstract_io",
+        "//src/v/cloud_topics/level_one/common:object",
+        "//src/v/cloud_topics/level_one/common:object_id",
         "//src/v/cloud_topics/level_one/common:object_utils",
         "//src/v/cloud_topics/level_zero/stm:ctp_stm_api",
         "//src/v/cluster",