events: Fetch metadata at arbitrary blocks (#727)

* subxt/rpc: Fetch metadata at arbitrary blocks

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* subxt/events: Fetch metadata for events at arbitrary blocks

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* Revert "subxt/events: Fetch metadata for events at arbitrary blocks"

This reverts commit 381409b7a8916611d7c44dc6ad58a90993b6c297.

* subxt/events: Custom constructor with metadata for events

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* Update subxt/src/events/events_type.rs

Co-authored-by: James Wilson <james@jsdw.me>

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>
Co-authored-by: James Wilson <james@jsdw.me>

subxt/src/client/online_client.rs

@@ -93,7 +93,7 @@ impl<T: Config> OnlineClient<T> {
         let (genesis_hash, runtime_version, metadata) = future::join3(
             rpc.genesis_hash(),
             rpc.runtime_version(None),
-            rpc.metadata(),
+            rpc.metadata(None),
         )
         .await;
 
@@ -310,7 +310,7 @@ impl<T: Config> RuntimeUpdaterStream<T> {
             Err(err) => return Some(Err(err)),
         };
 
-        let metadata = match self.client.rpc().metadata().await {
+        let metadata = match self.client.rpc().metadata(None).await {
             Ok(metadata) => metadata,
             Err(err) => return Some(Err(err)),
         };

subxt/src/events/events_client.rs

@@ -66,13 +66,7 @@ where
                 }
             };
 
-            let event_bytes = client
-                .rpc()
-                .storage(&system_events_key().0, Some(block_hash))
-                .await?
-                .map(|e| e.0)
-                .unwrap_or_else(Vec::new);
-
+            let event_bytes = get_event_bytes(&client, Some(block_hash)).await?;
             Ok(Events::new(client.metadata(), block_hash, event_bytes))
         }
     }
@@ -84,3 +78,20 @@ fn system_events_key() -> StorageKey {
     storage_key.extend(twox_128(b"Events").to_vec());
     StorageKey(storage_key)
 }
+
+// Get the event bytes from the provided client, at the provided block hash.
+pub(crate) async fn get_event_bytes<T, Client>(
+    client: &Client,
+    block_hash: Option<T::Hash>,
+) -> Result<Vec<u8>, Error>
+where
+    T: Config,
+    Client: OnlineClientT<T>,
+{
+    Ok(client
+        .rpc()
+        .storage(&system_events_key().0, block_hash)
+        .await?
+        .map(|e| e.0)
+        .unwrap_or_else(Vec::new))
+}

subxt/src/events/events_type.rs

@@ -9,7 +9,9 @@ use super::{
     StaticEvent,
 };
 use crate::{
+    client::OnlineClientT,
     error::Error,
+    events::events_client::get_event_bytes,
     metadata::EventMetadata,
     Config,
     Metadata,
@@ -64,6 +66,50 @@ impl<T: Config> Events<T> {
         }
     }
 
+    /// Obtain the events from a block hash given custom metadata and a client.
+    ///
+    /// This method gives users the ability to inspect the events of older blocks,
+    /// where the metadata changed. For those cases, the user is responsible for
+    /// providing a valid metadata.
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    /// # #[tokio::main]
+    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    ///  use subxt::{ OnlineClient, PolkadotConfig, events::Events };
+    ///
+    ///  let client = OnlineClient::<PolkadotConfig>::new().await.unwrap();
+    ///
+    ///  // Get the hash of an older block.
+    ///  let block_hash = client
+    ///     .rpc()
+    ///     .block_hash(Some(1u32.into()))
+    ///     .await?
+    ///     .expect("didn't pass a block number; qed");
+    ///  // Fetch the metadata of the given block.
+    ///  let metadata = client.rpc().metadata(Some(block_hash)).await?;
+    ///  // Fetch the events from the client.
+    ///  let events = Events::new_from_client(metadata, block_hash, client);
+    /// # Ok(())
+    /// # }
+    /// ```
+    ///
+    /// # Note
+    ///
+    /// Prefer to use [`crate::events::EventsClient::at`] to obtain the events.
+    pub async fn new_from_client<Client>(
+        metadata: Metadata,
+        block_hash: T::Hash,
+        client: Client,
+    ) -> Result<Self, Error>
+    where
+        Client: OnlineClientT<T>,
+    {
+        let event_bytes = get_event_bytes(&client, Some(block_hash)).await?;
+        Ok(Events::new(metadata, block_hash, event_bytes))
+    }
+
     /// The number of events.
     pub fn len(&self) -> u32 {
         self.num_events

subxt/src/rpc/rpc.rs

@@ -443,10 +443,10 @@ impl<T: Config> Rpc<T> {
     }
 
     /// Fetch the metadata
-    pub async fn metadata(&self) -> Result<Metadata, Error> {
+    pub async fn metadata(&self, at: Option<T::Hash>) -> Result<Metadata, Error> {
         let bytes: Bytes = self
             .client
-            .request("state_getMetadata", rpc_params![])
+            .request("state_getMetadata", rpc_params![at])
             .await?;
         let meta: RuntimeMetadataPrefixed = Decode::decode(&mut &bytes[..])?;
         let metadata: Metadata = meta.try_into()?;