derive.subscribeFinalizedBlocks w/ events & author (#5064)

* derive.subscribeFinalizedBlocks w/ events & author

* Apply createdAtHash (some)

* Pass through events registry

* Throw on use

packages/api-derive/src/chain/getBlock.ts

@@ -5,7 +5,7 @@ import type { Observable } from 'rxjs';
 import type { SignedBlockExtended } from '../type/types';
 import type { DeriveApi } from '../types';
 
-import { catchError, combineLatest, map, of, switchMap } from 'rxjs';
+import { combineLatest, map, of, switchMap } from 'rxjs';
 
 import { createSignedBlockExtended } from '../type';
 import { memo } from '../util';
@@ -23,8 +23,8 @@ import { memo } from '../util';
  * console.log(`block #${block.header.number} was authored by ${author}`);
  * ```
  */
-export function getBlock (instanceId: string, api: DeriveApi): (hash: Uint8Array | string) => Observable<SignedBlockExtended | undefined> {
-  return memo(instanceId, (blockHash: Uint8Array | string): Observable<SignedBlockExtended | undefined> =>
+export function getBlock (instanceId: string, api: DeriveApi): (hash: Uint8Array | string) => Observable<SignedBlockExtended> {
+  return memo(instanceId, (blockHash: Uint8Array | string): Observable<SignedBlockExtended> =>
     combineLatest([
       api.rpc.chain.getBlock(blockHash),
       api.queryAt(blockHash).pipe(
@@ -39,13 +39,8 @@ export function getBlock (instanceId: string, api: DeriveApi): (hash: Uint8Array
       )
     ]).pipe(
       map(([signedBlock, [events, validators]]) =>
-        createSignedBlockExtended(api.registry, signedBlock, events, validators)
-      ),
-      catchError((): Observable<undefined> =>
-        // where rpc.chain.getHeader throws, we will land here - it can happen that
-        // we supplied an invalid hash. (Due to defaults, storage will have an
-        // empty value, so only the RPC is affected). So return undefined
-        of()
+        createSignedBlockExtended(events.registry, signedBlock, events, validators)
       )
-    ));
+    )
+  );
 }

packages/api-derive/src/chain/getBlockByNumber.ts

@@ -10,8 +10,8 @@ import { switchMap } from 'rxjs';
 
 import { memo } from '../util';
 
-export function getBlockByNumber (instanceId: string, api: DeriveApi): (blockNumber: AnyNumber) => Observable<SignedBlockExtended | undefined> {
-  return memo(instanceId, (blockNumber: AnyNumber): Observable<SignedBlockExtended | undefined> =>
+export function getBlockByNumber (instanceId: string, api: DeriveApi): (blockNumber: AnyNumber) => Observable<SignedBlockExtended> {
+  return memo(instanceId, (blockNumber: AnyNumber): Observable<SignedBlockExtended> =>
     api.rpc.chain.getBlockHash(blockNumber).pipe(
       switchMap((h) => api.derive.chain.getBlock(h))
     )

packages/api-derive/src/chain/getHeader.ts

@@ -6,7 +6,7 @@ import type { AccountId } from '@polkadot/types/interfaces';
 import type { HeaderExtended } from '../type/types';
 import type { DeriveApi } from '../types';
 
-import { catchError, combineLatest, map, of, switchMap } from 'rxjs';
+import { combineLatest, map, of, switchMap } from 'rxjs';
 
 import { createHeaderExtended } from '../type';
 import { memo } from '../util';
@@ -25,8 +25,8 @@ import { memo } from '../util';
  * console.log(`block #${number} was authored by ${author}`);
  * ```
  */
-export function getHeader (instanceId: string, api: DeriveApi): (blockHash: Uint8Array | string) => Observable<HeaderExtended | undefined> {
-  return memo(instanceId, (blockHash: Uint8Array | string): Observable<HeaderExtended | undefined> =>
+export function getHeader (instanceId: string, api: DeriveApi): (blockHash: Uint8Array | string) => Observable<HeaderExtended> {
+  return memo(instanceId, (blockHash: Uint8Array | string): Observable<HeaderExtended> =>
     combineLatest([
       api.rpc.chain.getHeader(blockHash),
       api.queryAt(blockHash).pipe(
@@ -39,12 +39,6 @@ export function getHeader (instanceId: string, api: DeriveApi): (blockHash: Uint
     ]).pipe(
       map(([header, validators]) =>
         createHeaderExtended(header.registry, header, validators)
-      ),
-      catchError((): Observable<undefined> =>
-        // where rpc.chain.getHeader throws, we will land here - it can happen that
-        // we supplied an invalid hash. (Due to defaults, storeage will have an
-        // empty value, so only the RPC is affected). So return undefined
-        of()
       )
     ));
 }

packages/api-derive/src/chain/subscribeFinalizedBlocks.ts

@@ -0,0 +1,24 @@
+// Copyright 2017-2022 @polkadot/api-derive authors & contributors
+// SPDX-License-Identifier: Apache-2.0
+
+import type { Observable } from 'rxjs';
+import type { SignedBlockExtended } from '../type/types';
+import type { DeriveApi } from '../types';
+
+import { switchMap } from 'rxjs';
+
+import { memo } from '../util';
+
+/**
+ * @name subscribeFinalizedBlocks
+ * @returns The finalized block & events for that block
+ */
+export function subscribeFinalizedBlocks (instanceId: string, api: DeriveApi): () => Observable<SignedBlockExtended> {
+  return memo(instanceId, (): Observable<SignedBlockExtended> =>
+    api.derive.chain.subscribeFinalizedHeads().pipe(
+      switchMap((header) =>
+        api.derive.chain.getBlock(header.createdAtHash || header.hash)
+      )
+    )
+  );
+}

packages/api-derive/src/chain/subscribeFinalizedHeads.ts

@@ -9,6 +9,9 @@ import { from, of, switchMap } from 'rxjs';
 
 import { memo } from '../util';
 
+/**
+ * Returns a header range from startHash to to (not including) endHash, i.e. lastBlock.parentHash === endHash
+ */
 export function _getHeaderRange (instanceId: string, api: DeriveApi): (startHash: Hash, endHash: Hash, prev?: Header[]) => Observable<Header[]> {
   return memo(instanceId, (startHash: Hash, endHash: Hash, prev: Header[] = []): Observable<Header[]> =>
     api.rpc.chain.getHeader(startHash).pipe(
@@ -39,6 +42,7 @@ export function subscribeFinalizedHeads (instanceId: string, api: DeriveApi): ()
         const startHash = header.parentHash;
 
         prevHash = header.hash;
+        header.createdAtHash = prevHash;
 
         return endHash === null || startHash.eq(endHash)
           ? of(header)

packages/api-derive/src/chain/subscribeNewBlocks.ts

@@ -32,7 +32,7 @@ export function subscribeNewBlocks (instanceId: string, api: DeriveApi): () => O
         ]);
       }),
       map(([header, block, events]) =>
-        createSignedBlockExtended(block.registry, block, events, header.validators)
+        createSignedBlockExtended(events.registry, block, events, header.validators)
       )
     )
   );