feat(drive-abci): preliminary support for event subscriptions in Platform

Cargo.toml

@@ -36,6 +36,7 @@ members = [
     "packages/rs-sdk-ffi",
     "packages/wasm-drive-verify",
     "packages/dash-platform-balance-checker",
+    "packages/rs-dash-event-bus",
     "packages/rs-platform-wallet",
     "packages/wasm-sdk",
 ]

Dockerfile

@@ -373,6 +373,7 @@ COPY --parents \
     packages/rs-platform-versioning \
     packages/rs-platform-value-convertible \
     packages/rs-drive-abci \
+    packages/rs-dash-event-bus \
     packages/dashpay-contract \
     packages/withdrawals-contract \
     packages/masternode-reward-shares-contract \
@@ -451,6 +452,7 @@ COPY --parents \
     .cargo \
     packages/dapi-grpc \
     packages/rs-dapi-grpc-macros \
+    packages/rs-dash-event-bus \
     packages/rs-dpp \
     packages/rs-drive \
     packages/rs-platform-value \
@@ -553,6 +555,7 @@ COPY --parents \
     Cargo.toml \
     rust-toolchain.toml \
     .cargo \
+    packages/rs-dash-event-bus \
     packages/rs-dpp \
     packages/rs-platform-value \
     packages/rs-platform-serialization \

packages/dapi-grpc/protos/platform/v0/platform.proto

@@ -6,6 +6,94 @@ package org.dash.platform.dapi.v0;
 
 import "google/protobuf/timestamp.proto";
 
+// Platform events streaming (v0)
+message PlatformEventsCommand {
+  message PlatformEventsCommandV0 {
+    oneof command {
+      AddSubscriptionV0 add = 1;
+      RemoveSubscriptionV0 remove = 2;
+      PingV0 ping = 3;
+    }
+  }
+  oneof version { PlatformEventsCommandV0 v0 = 1; }
+}
+
+message PlatformEventsResponse {
+  message PlatformEventsResponseV0 {
+    oneof response {
+      PlatformEventMessageV0 event = 1;
+      AckV0 ack = 2;
+      PlatformErrorV0 error = 3;
+    }
+  }
+  oneof version { PlatformEventsResponseV0 v0 = 1; }
+}
+
+message AddSubscriptionV0 {
+  string client_subscription_id = 1;
+  PlatformFilterV0 filter = 2;
+}
+
+message RemoveSubscriptionV0 {
+  string client_subscription_id = 1;
+}
+
+message PingV0 { uint64 nonce = 1; }
+
+message AckV0 {
+  string client_subscription_id = 1;
+  string op = 2; // "add" | "remove"
+}
+
+message PlatformErrorV0 {
+  string client_subscription_id = 1;
+  uint32 code = 2;
+  string message = 3;
+}
+
+message PlatformEventMessageV0 {
+  string client_subscription_id = 1;
+  PlatformEventV0 event = 2;
+}
+
+// Initial placeholder filter and event to be refined during integration
+// Filter for StateTransitionResult events
+message StateTransitionResultFilter {
+  // When set, only match StateTransitionResult events for this tx hash.
+  optional bytes tx_hash = 1;
+}
+
+message PlatformFilterV0 {
+  oneof kind {
+    bool all = 1; // subscribe to all platform events
+    bool block_committed = 2; // subscribe to BlockCommitted events only
+    StateTransitionResultFilter state_transition_result = 3; // subscribe to StateTransitionResult events (optionally filtered by tx_hash)
+  }
+}
+
+message PlatformEventV0 {
+  message BlockMetadata {
+    uint64 height = 1 [ jstype = JS_STRING ];
+    uint64 time_ms = 2 [ jstype = JS_STRING ];
+    bytes block_id_hash = 3;
+  }
+
+  message BlockCommitted {
+    BlockMetadata meta = 1;
+    uint32 tx_count = 2;
+  }
+
+  message StateTransitionFinalized {
+    BlockMetadata meta = 1;
+    bytes tx_hash = 2;
+  }
+
+  oneof event {
+    BlockCommitted block_committed = 1;
+    StateTransitionFinalized state_transition_finalized = 2;
+  }
+}
+
 service Platform {
   rpc broadcastStateTransition(BroadcastStateTransitionRequest)
       returns (BroadcastStateTransitionResponse);
@@ -102,6 +190,10 @@ service Platform {
   rpc getGroupActions(GetGroupActionsRequest) returns (GetGroupActionsResponse);
   rpc getGroupActionSigners(GetGroupActionSignersRequest)
       returns (GetGroupActionSignersResponse);
+
+  // Bi-directional stream for multiplexed platform events subscriptions
+  rpc subscribePlatformEvents(stream PlatformEventsCommand)
+      returns (stream PlatformEventsResponse);
 }
 
 // Proof message includes cryptographic proofs for validating responses

packages/rs-dash-event-bus/Cargo.toml

@@ -0,0 +1,35 @@
+[package]
+name = "rs-dash-event-bus"
+version = "2.1.0-dev.7"
+edition = "2024"
+license = "MIT"
+description = "Shared event bus and Platform events multiplexer for Dash Platform (rs-dapi, rs-drive-abci, rs-sdk)"
+
+[lib]
+name = "dash_event_bus"
+path = "src/lib.rs"
+
+[features]
+default = []
+metrics = ["dep:metrics"]
+
+[dependencies]
+tokio = { version = "1", features = ["rt", "macros", "sync", "time"] }
+tokio-stream = { version = "0.1", features = ["sync"] }
+tokio-util = { version = "0.7", features = ["rt"] }
+tracing = "0.1"
+futures = "0.3"
+
+# Internal workspace crates
+dapi-grpc = { path = "../dapi-grpc" }
+
+# Optional metrics
+metrics = { version = "0.24.2", optional = true }
+
+[dev-dependencies]
+tokio = { version = "1", features = [
+    "rt-multi-thread",
+    "macros",
+    "sync",
+    "time",
+] }