ct: add compaction_scheduler to app

Author: WillemKauf

src/v/cloud_topics/BUILD

@@ -99,6 +99,7 @@ redpanda_cc_library(
         "//src/v/base",
         "//src/v/cloud_topics:state_accessors",
         "//src/v/cloud_topics/level_one/common:file_io",
+        "//src/v/cloud_topics/level_one/compaction:scheduler",
         "//src/v/cloud_topics/level_one/domain:domain_supervisor",
         "//src/v/cloud_topics/level_one/metastore:leader_router",
         "//src/v/cloud_topics/level_one/metastore:replicated_metastore",

src/v/cloud_topics/app.cc

@@ -14,6 +14,7 @@
 #include "cloud_topics/data_plane_api.h"
 #include "cloud_topics/data_plane_impl.h"
 #include "cloud_topics/housekeeper/manager.h"
+#include "cloud_topics/level_one/compaction/scheduler.h"
 #include "cloud_topics/level_one/metastore/topic_purger.h"
 #include "cloud_topics/level_zero/gc/level_zero_gc.h"
 #include "cloud_topics/manager/manager.h"
@@ -113,6 +114,16 @@ ss::future<> app::construct(
                                    return &replicated_metastore.local();
                                }));
 
+    construct_single_service(
+      compaction_scheduler,
+      l1::compaction_cluster_state{
+        .self = self,
+        .leaders_table = leaders_table,
+        .topic_table = &controller->get_topics_state(),
+        .metadata_cache = metadata_cache},
+      &l1_io,
+      &replicated_metastore);
+
     // Must be last to register so it will be first to be stopped in
     // `app::stop`. This is to ensure that stopped services don't receive
     // callbacks.
@@ -129,6 +140,7 @@ ss::future<> app::start() {
     co_await domain_supervisor.invoke_on_all(
       [](auto& ds) { return ds.start(); });
     co_await housekeeper_manager.invoke_on_all(&housekeeper_manager::start);
+    co_await compaction_scheduler->start();
 
     // When start is called, we must have registered all the callbacks before
     // this as starting the manager will invoke callbacks for partitions already
@@ -228,4 +240,8 @@ ss::sharded<l1::replicated_metastore>* app::get_sharded_replicated_metastore() {
     return &replicated_metastore;
 }
 
+l1::compaction_scheduler* app::get_compaction_scheduler() {
+    return compaction_scheduler.get();
+}
+
 } // namespace cloud_topics

src/v/cloud_topics/app.h

@@ -11,6 +11,7 @@
 #pragma once
 
 #include "cloud_topics/level_one/common/file_io.h"
+#include "cloud_topics/level_one/compaction/scheduler.h"
 #include "cloud_topics/level_one/domain/domain_supervisor.h"
 #include "cloud_topics/level_one/metastore/leader_router.h"
 #include "cloud_topics/level_one/metastore/replicated_metastore.h"
@@ -74,6 +75,7 @@ class app : public ssx::sharded_service_container {
     ss::sharded<l1::domain_supervisor>* get_sharded_l1_domain_supervisor();
     ss::sharded<reconciler::reconciler>* get_reconciler();
     ss::sharded<l1::replicated_metastore>* get_sharded_replicated_metastore();
+    l1::compaction_scheduler* get_compaction_scheduler();
 
     // TODO: add 'get_control_plane_api' etc
 
@@ -92,6 +94,7 @@ class app : public ssx::sharded_service_container {
     ss::sharded<cloud_topics_manager> manager;
     ss::sharded<level_zero_gc> l0_gc;
     ss::sharded<housekeeper_manager> housekeeper_manager;
+    std::unique_ptr<l1::compaction_scheduler> compaction_scheduler;
 };
 
 } // namespace cloud_topics