c/backend: shard_table to be able to notify subscribers

src/v/cluster/shard_table.h

@@ -13,6 +13,7 @@
 
 #include "base/seastarx.h"
 #include "cluster/logger.h"
+#include "cluster/notification.h"
 #include "container/chunked_hash_map.h"
 #include "model/fundamental.h"
 #include "model/ktp.h"
@@ -74,6 +75,10 @@ class shard_table final {
           log_rev);
         _ntp_idx.insert_or_assign(ntp, shard_revision{shard, log_rev});
         _group_idx.insert_or_assign(g, shard_revision{shard, log_rev});
+
+        for (auto& cb : _notifications) {
+            cb.second({ntp, g, shard});
+        }
     }
 
     void
@@ -104,6 +109,31 @@ class shard_table final {
           log_rev);
         _ntp_idx.erase(ntp);
         _group_idx.erase(g);
+
+        for (auto& cb : _notifications) {
+            cb.second({ntp, g, std::nullopt});
+        }
+    }
+
+    struct change_t {
+        model::ntp ntp;
+        raft::group_id g;
+        std::optional<ss::shard_id> shard;
+    };
+    using change_cb_t = ss::noncopyable_function<void(change_t)>;
+
+    notification_id_type register_notification(change_cb_t cb) {
+        auto id = _notification_id++;
+        _notifications.emplace_back(id, std::move(cb));
+        return id;
+    }
+
+    void unregister_delta_notification(cluster::notification_id_type id) {
+        std::erase_if(
+          _notifications,
+          [id](const std::pair<cluster::notification_id_type, change_cb_t>& n) {
+              return n.first == id;
+          });
     }
 
 private:
@@ -131,5 +161,9 @@ class shard_table final {
       _ntp_idx;
     // raft index
     chunked_hash_map<raft::group_id, shard_revision> _group_idx;
+
+    cluster::notification_id_type _notification_id{0};
+    std::vector<std::pair<cluster::notification_id_type, change_cb_t>>
+      _notifications;
 };
 } // namespace cluster