Additional derive cleanups (#4307)

* Additional derive cleanups

* Adjust

* Multi cache set

* Adjust mapping

packages/api-derive/src/accounts/identity.ts

@@ -13,7 +13,7 @@ import { combineLatest, map, of, switchMap } from 'rxjs';
 
 import { isHex, u8aToString } from '@polkadot/util';
 
-import { memo } from '../util';
+import { firstObservable, memo } from '../util';
 
 type IdentityInfoAdditional = PalletIdentityIdentityInfo['additional'][0];
 
@@ -106,9 +106,7 @@ export function identity (instanceId: string, api: ApiInterfaceRx): (accountId?:
 
 export function hasIdentity (instanceId: string, api: ApiInterfaceRx): (accountId: AccountId | Uint8Array | string) => Observable<DeriveHasIdentity> {
   return memo(instanceId, (accountId: AccountId | Uint8Array | string): Observable<DeriveHasIdentity> =>
-    api.derive.accounts.hasIdentityMulti([accountId]).pipe(
-      map(([first]) => first)
-    )
+    firstObservable(api.derive.accounts.hasIdentityMulti([accountId]))
   );
 }
 

packages/api-derive/src/collective/helpers.ts

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 import type { ApiInterfaceRx } from '@polkadot/api/types';
+import type { Collective } from './types';
 
 export function getInstance (api: ApiInterfaceRx, section: string): string {
   const instances = api.registry.getModuleInstances(api.runtimeVersion.specName.toString(), section);
@@ -10,3 +11,8 @@ export function getInstance (api: ApiInterfaceRx, section: string): string {
     ? instances[0]
     : section;
 }
+
+export function withSection <T> (_section: Collective, fn: (section: string, instanceId: string, api: ApiInterfaceRx) => T): (instanceId: string, api: ApiInterfaceRx) => T {
+  return (instanceId: string, api: ApiInterfaceRx) =>
+    fn(getInstance(api, _section), instanceId, api);
+}

packages/api-derive/src/collective/members.ts

@@ -11,16 +11,14 @@ import { of } from 'rxjs';
 import { isFunction } from '@polkadot/util';
 
 import { memo } from '../util';
-import { getInstance } from './getInstance';
+import { withSection } from './helpers';
 
 export function members (_section: Collective): (instanceId: string, api: ApiInterfaceRx) => () => Observable<AccountId[]> {
-  return (instanceId: string, api: ApiInterfaceRx) => {
-    const section = getInstance(api, _section);
-
-    return memo(instanceId, (): Observable<AccountId[]> =>
+  return withSection(_section, (section, instanceId, api) =>
+    memo(instanceId, (): Observable<AccountId[]> =>
       isFunction(api.query[section]?.members)
         ? api.query[section as 'council'].members()
         : of([])
-    );
-  };
+    )
+  );
 }

packages/api-derive/src/collective/prime.ts

@@ -5,19 +5,18 @@ import type { Observable } from 'rxjs';
 import type { ApiInterfaceRx } from '@polkadot/api/types';
 import type { Option } from '@polkadot/types';
 import type { AccountId } from '@polkadot/types/interfaces';
+import type { Collective } from './types';
 
 import { map, of } from 'rxjs';
 
 import { isFunction } from '@polkadot/util';
 
 import { memo } from '../util';
-import { getInstance } from './getInstance';
+import { withSection } from './helpers';
 
-export function prime (_section: string): (instanceId: string, api: ApiInterfaceRx) => () => Observable<AccountId | null> {
-  return (instanceId: string, api: ApiInterfaceRx) => {
-    const section = getInstance(api, _section);
-
-    return memo(instanceId, (): Observable<AccountId | null> =>
+export function prime (_section: Collective): (instanceId: string, api: ApiInterfaceRx) => () => Observable<AccountId | null> {
+  return withSection(_section, (section, instanceId, api) =>
+    memo(instanceId, (): Observable<AccountId | null> =>
       isFunction(api.query[section as 'council']?.prime)
         ? api.query[section as 'council'].prime<Option<AccountId>>().pipe(
           map((optPrime): AccountId | null =>
@@ -25,6 +24,6 @@ export function prime (_section: string): (instanceId: string, api: ApiInterface
           )
         )
         : of(null)
-    );
-  };
+    )
+  );
 }

packages/api-derive/src/collective/proposals.ts

@@ -13,17 +13,17 @@ import { catchError, combineLatest, map, of, switchMap } from 'rxjs';
 import { isFunction } from '@polkadot/util';
 
 import { memo } from '../util';
-import { getInstance } from './getInstance';
+import { withSection } from './helpers';
 
 type Result = [(Hash | Uint8Array | string)[], (Option<Proposal> | null)[], Option<Votes>[]];
 
 function parse (api: ApiInterfaceRx, [hashes, proposals, votes]: Result): DeriveCollectiveProposal[] {
   return proposals
-    .map((proposalOpt, index): DeriveCollectiveProposal | null =>
-      proposalOpt && proposalOpt.isSome
+    .map((o, index): DeriveCollectiveProposal | null =>
+      o && o.isSome
         ? {
           hash: api.registry.createType('Hash', hashes[index]),
-          proposal: proposalOpt.unwrap(),
+          proposal: o.unwrap(),
           votes: votes[index].unwrapOr(null)
         }
         : null
@@ -31,7 +31,7 @@ function parse (api: ApiInterfaceRx, [hashes, proposals, votes]: Result): Derive
     .filter((proposal): proposal is DeriveCollectiveProposal => !!proposal);
 }
 
-function _proposalsFrom (instanceId: string, api: ApiInterfaceRx, section: string): (hashes: (Hash | Uint8Array | string)[]) => Observable<DeriveCollectiveProposal[]> {
+function _proposalsFrom (section: string, instanceId: string, api: ApiInterfaceRx): (hashes: (Hash | Uint8Array | string)[]) => Observable<DeriveCollectiveProposal[]> {
   return memo(instanceId, (hashes: (Hash | Uint8Array | string)[]): Observable<DeriveCollectiveProposal[]> =>
     (isFunction(api.query[section]?.proposals) && hashes.length
       ? combineLatest([
@@ -39,8 +39,8 @@ function _proposalsFrom (instanceId: string, api: ApiInterfaceRx, section: strin
         // this should simply be api.query[section].proposalOf.multi<Option<Proposal>>(hashes),
         // however we have had cases on Edgeware where the indices have moved around after an
         // upgrade, which results in invalid on-chain data
-        combineLatest(hashes.map((hash) =>
-          api.query[section].proposalOf<Option<Proposal>>(hash).pipe(
+        combineLatest(hashes.map((h) =>
+          api.query[section].proposalOf<Option<Proposal>>(h).pipe(
             catchError(() => of(null))
           )
         )),
@@ -48,69 +48,63 @@ function _proposalsFrom (instanceId: string, api: ApiInterfaceRx, section: strin
       ])
       : of<Result>([[], [], []])
     ).pipe(
-      map((result) => parse(api, result))
+      map((r) => parse(api, r))
     )
   );
 }
 
-export function hasProposals (_section: Collective): (instanceId: string, api: ApiInterfaceRx) => () => Observable<boolean> {
-  return (instanceId: string, api: ApiInterfaceRx) => {
-    const section = getInstance(api, _section);
+function withProposals <T> (_section: Collective, fn: (section: string, instanceId: string, api: ApiInterfaceRx, proposalsFrom: (hashes: (string | Hash | Uint8Array)[]) => Observable<DeriveCollectiveProposal[]>) => T): (instanceId: string, api: ApiInterfaceRx) => T {
+  return withSection(_section, (section, instanceId, api) =>
+    fn(section, instanceId, api, _proposalsFrom(section, instanceId, api))
+  );
+}
 
-    return memo(instanceId, (): Observable<boolean> =>
+export function hasProposals (_section: Collective): (instanceId: string, api: ApiInterfaceRx) => () => Observable<boolean> {
+  return withSection(_section, (section, instanceId, api) =>
+    memo(instanceId, (): Observable<boolean> =>
       of(isFunction(api.query[section]?.proposals))
-    );
-  };
+    )
+  );
 }
 
 export function proposalCount (_section: Collective): (instanceId: string, api: ApiInterfaceRx) => () => Observable<u32 | null> {
-  return (instanceId: string, api: ApiInterfaceRx) => {
-    const section = getInstance(api, _section);
-
-    return memo(instanceId, (): Observable<u32 | null> =>
+  return withSection(_section, (section, instanceId, api) =>
+    memo(instanceId, (): Observable<u32 | null> =>
       isFunction(api.query[section].proposalCount)
         ? api.query[section as 'council'].proposalCount()
         : of(null)
-    );
-  };
+    )
+  );
 }
 
 export function proposalHashes (_section: Collective): (instanceId: string, api: ApiInterfaceRx) => () => Observable<Hash[]> {
-  return (instanceId: string, api: ApiInterfaceRx) => {
-    const section = getInstance(api, _section);
-
-    return memo(instanceId, (): Observable<Hash[]> =>
+  return withSection(_section, (section, instanceId, api) =>
+    memo(instanceId, (): Observable<Hash[]> =>
       isFunction(api.query[section]?.proposals)
         ? api.query[section as 'council'].proposals()
         : of([])
-    );
-  };
+    )
+  );
 }
 
 export function proposals (_section: Collective): (instanceId: string, api: ApiInterfaceRx) => () => Observable<DeriveCollectiveProposal[]> {
-  return (instanceId: string, api: ApiInterfaceRx) => {
-    const section = getInstance(api, _section);
-    const proposalsFrom = _proposalsFrom(instanceId, api, section);
-
-    return memo(instanceId, (): Observable<DeriveCollectiveProposal[]> =>
+  return withProposals(_section, (section, instanceId, api, proposalsFrom) =>
+    memo(instanceId, (): Observable<DeriveCollectiveProposal[]> =>
       api.derive[section as 'council'].proposalHashes().pipe(
         switchMap(proposalsFrom)
       )
-    );
-  };
+    )
+  );
 }
 
 export function proposal (_section: Collective): (instanceId: string, api: ApiInterfaceRx) => (hash: Hash | Uint8Array | string) => Observable<DeriveCollectiveProposal | null> {
-  return (instanceId: string, api: ApiInterfaceRx) => {
-    const section = getInstance(api, _section);
-    const proposalsFrom = _proposalsFrom(instanceId, api, section);
-
-    return memo(instanceId, (hash: Hash | Uint8Array | string): Observable<DeriveCollectiveProposal | null> =>
+  return withProposals(_section, (section, instanceId, api, proposalsFrom) =>
+    memo(instanceId, (hash: Hash | Uint8Array | string): Observable<DeriveCollectiveProposal | null> =>
       isFunction(api.query[section]?.proposals)
         ? proposalsFrom([hash]).pipe(
           map(([proposal]) => proposal)
         )
         : of(null)
-    );
-  };
+    )
+  );
 }

packages/api-derive/src/staking/account.ts

@@ -11,7 +11,7 @@ import { combineLatest, map, switchMap } from 'rxjs';
 
 import { BN, BN_ZERO } from '@polkadot/util';
 
-import { memo } from '../util';
+import { firstObservable, memo } from '../util';
 
 const QUERY_OPTS = {
   withDestination: true,
@@ -85,8 +85,6 @@ export function accounts (instanceId: string, api: ApiInterfaceRx): (accountIds:
  */
 export function account (instanceId: string, api: ApiInterfaceRx): (accountId: Uint8Array | string) => Observable<DeriveStakingAccount> {
   return memo(instanceId, (accountId: Uint8Array | string): Observable<DeriveStakingAccount> =>
-    api.derive.staking.accounts([accountId]).pipe(
-      map(([first]) => first)
-    )
+    firstObservable(api.derive.staking.accounts([accountId]))
   );
 }

packages/api-derive/src/staking/cache.ts

@@ -25,3 +25,16 @@ export function getEraMultiCache <T> (CACHE_KEY: string, eras: EraIndex[], withA
 
   return cached;
 }
+
+export function setEraMultiCache <T extends { era: EraIndex }> (CACHE_KEY: string, withActive: boolean, values: T[]): T[] {
+  !withActive && values.forEach((v) => deriveCache.set(`${CACHE_KEY}-${v.era.toString()}`, v));
+
+  return values;
+}
+
+export function filterCachedEras <T extends { era: EraIndex }> (eras: EraIndex[], cached: T[], query: T[]): T[] {
+  return eras.map((e) =>
+    cached.find(({ era }) => e.eq(era)) ||
+    query.find(({ era }) => e.eq(era)) as T
+  );
+}

packages/api-derive/src/staking/erasExposure.ts

@@ -8,11 +8,11 @@ import type { AccountId, EraIndex } from '@polkadot/types/interfaces';
 import type { PalletStakingExposure } from '@polkadot/types/lookup';
 import type { DeriveEraExposure, DeriveEraNominatorExposure, DeriveEraValidatorExposure } from '../types';
 
-import { combineLatest, map, of } from 'rxjs';
+import { map, of } from 'rxjs';
 
 import { deriveCache, memo } from '../util';
 import { getEraCache } from './cache';
-import { erasHistoricApply } from './util';
+import { combineEras, erasHistoricApply } from './util';
 
 type KeysAndExposures = [StorageKey<[EraIndex, AccountId]>, PalletStakingExposure][];
 
@@ -62,12 +62,5 @@ export function eraExposure (instanceId: string, api: ApiInterfaceRx): (era: Era
   );
 }
 
-export function _erasExposure (instanceId: string, api: ApiInterfaceRx): (eras: EraIndex[], withActive: boolean) => Observable<DeriveEraExposure[]> {
-  return memo(instanceId, (eras: EraIndex[], withActive: boolean): Observable<DeriveEraExposure[]> =>
-    eras.length
-      ? combineLatest(eras.map((era) => api.derive.staking._eraExposure(era, withActive)))
-      : of([])
-  );
-}
-
+export const _erasExposure = combineEras('_eraExposure');
 export const erasExposure = erasHistoricApply('_erasExposure');

packages/api-derive/src/staking/erasPoints.ts

@@ -11,8 +11,8 @@ import { map, of } from 'rxjs';
 
 import { BN_ZERO } from '@polkadot/util';
 
-import { deriveCache, memo } from '../util';
-import { getEraMultiCache } from './cache';
+import { memo } from '../util';
+import { filterCachedEras, getEraMultiCache, setEraMultiCache } from './cache';
 import { erasHistoricApply, filterEras } from './util';
 
 const CACHE_KEY = 'eraPoints';
@@ -47,16 +47,7 @@ export function _erasPoints (instanceId: string, api: ApiInterfaceRx): (eras: Er
     return !remaining.length
       ? of(cached)
       : api.query.staking.erasRewardPoints.multi(remaining).pipe(
-        map((points): DeriveEraPoints[] => {
-          const query = mapPoints(remaining, points);
-
-          !withActive && query.forEach((q) => deriveCache.set(`${CACHE_KEY}-${q.era.toString()}`, q));
-
-          return eras.map((era): DeriveEraPoints =>
-            cached.find((cached) => era.eq(cached.era)) ||
-            query.find((query) => era.eq(query.era)) as DeriveEraPoints
-          );
-        })
+        map((p) => filterCachedEras(eras, cached, setEraMultiCache(CACHE_KEY, withActive, mapPoints(remaining, p))))
       );
   });
 }

packages/api-derive/src/staking/erasPrefs.ts

@@ -8,11 +8,11 @@ import type { EraIndex } from '@polkadot/types/interfaces';
 import type { PalletStakingValidatorPrefs } from '@polkadot/types/lookup';
 import type { DeriveEraPrefs, DeriveEraValPrefs } from '../types';
 
-import { combineLatest, map, of } from 'rxjs';
+import { map, of } from 'rxjs';
 
 import { deriveCache, memo } from '../util';
 import { getEraCache } from './cache';
-import { erasHistoricApply } from './util';
+import { combineEras, erasHistoricApply } from './util';
 
 const CACHE_KEY = 'eraPrefs';
 
@@ -50,12 +50,5 @@ export function eraPrefs (instanceId: string, api: ApiInterfaceRx): (era: EraInd
   );
 }
 
-export function _erasPrefs (instanceId: string, api: ApiInterfaceRx): (eras: EraIndex[], withActive: boolean) => Observable<DeriveEraPrefs[]> {
-  return memo(instanceId, (eras: EraIndex[], withActive: boolean): Observable<DeriveEraPrefs[]> =>
-    eras.length
-      ? combineLatest(eras.map((era) => api.derive.staking._eraPrefs(era, withActive)))
-      : of([])
-  );
-}
-
+export const _erasPrefs = combineEras('_eraPrefs');
 export const erasPrefs = erasHistoricApply('_erasPrefs');

packages/api-derive/src/staking/erasRewards.ts

@@ -9,8 +9,8 @@ import type { DeriveEraRewards } from '../types';
 
 import { map, of } from 'rxjs';
 
-import { deriveCache, memo } from '../util';
-import { getEraMultiCache } from './cache';
+import { memo } from '../util';
+import { filterCachedEras, getEraMultiCache, setEraMultiCache } from './cache';
 import { erasHistoricApply, filterEras } from './util';
 
 const CACHE_KEY = 'eraRewards';
@@ -36,16 +36,7 @@ export function _erasRewards (instanceId: string, api: ApiInterfaceRx): (eras: E
     }
 
     return api.query.staking.erasValidatorReward.multi(remaining).pipe(
-      map((optRewards) => {
-        const query = mapRewards(remaining, optRewards);
-
-        !withActive && query.forEach((q) => deriveCache.set(`${CACHE_KEY}-${q.era.toString()}`, q));
-
-        return eras.map((era): DeriveEraRewards =>
-          cached.find((cached) => era.eq(cached.era)) ||
-          query.find((query) => era.eq(query.era)) as DeriveEraRewards
-        );
-      })
+      map((r) => filterCachedEras(eras, cached, setEraMultiCache(CACHE_KEY, withActive, mapRewards(remaining, r))))
     );
   });
 }

packages/api-derive/src/staking/erasSlashes.ts

@@ -12,7 +12,7 @@ import { combineLatest, map, of } from 'rxjs';
 
 import { deriveCache, memo } from '../util';
 import { getEraCache } from './cache';
-import { erasHistoricApply } from './util';
+import { combineEras, erasHistoricApply } from './util';
 
 const CACHE_KEY = 'eraSlashes';
 
@@ -58,14 +58,5 @@ export function eraSlashes (instanceId: string, api: ApiInterfaceRx): (era: EraI
   );
 }
 
-export function _erasSlashes (instanceId: string, api: ApiInterfaceRx): (eras: EraIndex[], withActive: boolean) => Observable<DeriveEraSlashes[]> {
-  return memo(instanceId, (eras: EraIndex[], withActive: boolean): Observable<DeriveEraSlashes[]> =>
-    eras.length
-      ? combineLatest(
-        eras.map((era) => api.derive.staking._eraSlashes(era, withActive))
-      )
-      : of([])
-  );
-}
-
+export const _erasSlashes = combineEras('_eraSlashes');
 export const erasSlashes = erasHistoricApply('_erasSlashes');

packages/api-derive/src/staking/keys.ts

@@ -11,7 +11,7 @@ import type { DeriveStakingKeys } from './types';
 
 import { combineLatest, map, of, switchMap } from 'rxjs';
 
-import { memo } from '../util';
+import { firstObservable, memo } from '../util';
 
 function extractsIds (stashId: Uint8Array | string, queuedKeys: [AccountId, NodeRuntimeSessionKeys | AccountId[]][], nextKeys: Option<NodeRuntimeSessionKeys>): DeriveStakingKeys {
   const sessionIds = (queuedKeys.find(([currentId]) => currentId.eq(stashId)) || [undefined, [] as AccountId[]])[1];
@@ -29,9 +29,7 @@ function extractsIds (stashId: Uint8Array | string, queuedKeys: [AccountId, Node
 
 export function keys (instanceId: string, api: ApiInterfaceRx): (stashId: Uint8Array | string) => Observable<DeriveStakingKeys> {
   return memo(instanceId, (stashId: Uint8Array | string): Observable<DeriveStakingKeys> =>
-    api.derive.staking.keysMulti([stashId]).pipe(
-      map(([first]) => first)
-    )
+    firstObservable(api.derive.staking.keysMulti([stashId]))
   );
 }
 

packages/api-derive/src/staking/ownExposure.ts

@@ -9,7 +9,7 @@ import type { DeriveOwnExposure } from '../types';
 
 import { map, of } from 'rxjs';
 
-import { memo } from '../util';
+import { firstObservable, memo } from '../util';
 import { erasHistoricApplyAccount } from './util';
 
 export function _ownExposures (instanceId: string, api: ApiInterfaceRx): (accountId: Uint8Array | string, eras: EraIndex[], withActive: boolean) => Observable<DeriveOwnExposure[]> {
@@ -17,8 +17,8 @@ export function _ownExposures (instanceId: string, api: ApiInterfaceRx): (accoun
   return memo(instanceId, (accountId: Uint8Array | string, eras: EraIndex[], _withActive: boolean): Observable<DeriveOwnExposure[]> =>
     eras.length
       ? api.queryMulti<PalletStakingExposure[]>([
-        ...eras.map((era): QueryableStorageMultiArg<'rxjs'> => [api.query.staking.erasStakersClipped, [era, accountId]]),
-        ...eras.map((era): QueryableStorageMultiArg<'rxjs'> => [api.query.staking.erasStakers, [era, accountId]])
+        ...eras.map((e): QueryableStorageMultiArg<'rxjs'> => [api.query.staking.erasStakersClipped, [e, accountId]]),
+        ...eras.map((e): QueryableStorageMultiArg<'rxjs'> => [api.query.staking.erasStakers, [e, accountId]])
       ]).pipe(
         map((all): DeriveOwnExposure[] =>
           eras.map((era, index) => ({ clipped: all[index], era, exposure: all[eras.length + index] }))
@@ -30,9 +30,7 @@ export function _ownExposures (instanceId: string, api: ApiInterfaceRx): (accoun
 
 export function ownExposure (instanceId: string, api: ApiInterfaceRx): (accountId: Uint8Array | string, era: EraIndex) => Observable<DeriveOwnExposure> {
   return memo(instanceId, (accountId: Uint8Array | string, era: EraIndex): Observable<DeriveOwnExposure> =>
-    api.derive.staking._ownExposures(accountId, [era], true).pipe(
-      map(([first]) => first)
-    )
+    firstObservable(api.derive.staking._ownExposures(accountId, [era], true))
   );
 }
 

packages/api-derive/src/staking/ownSlashes.ts

@@ -10,7 +10,7 @@ import type { DeriveStakerSlashes } from '../types';
 
 import { map, of } from 'rxjs';
 
-import { memo } from '../util';
+import { firstObservable, memo } from '../util';
 import { erasHistoricApplyAccount } from './util';
 
 export function _ownSlashes (instanceId: string, api: ApiInterfaceRx): (accountId: Uint8Array | string, eras: EraIndex[], withActive: boolean) => Observable<DeriveStakerSlashes[]> {
@@ -18,8 +18,8 @@ export function _ownSlashes (instanceId: string, api: ApiInterfaceRx): (accountI
   return memo(instanceId, (accountId: Uint8Array | string, eras: EraIndex[], _withActive: boolean): Observable<DeriveStakerSlashes[]> =>
     eras.length
       ? api.queryMulti<(Option<ITuple<[Perbill, BalanceOf]>> | Option<BalanceOf>)[]>([
-        ...eras.map((era): QueryableStorageMultiArg<'rxjs'> => [api.query.staking.validatorSlashInEra, [era, accountId]]),
-        ...eras.map((era): QueryableStorageMultiArg<'rxjs'> => [api.query.staking.nominatorSlashInEra, [era, accountId]])
+        ...eras.map((e): QueryableStorageMultiArg<'rxjs'> => [api.query.staking.validatorSlashInEra, [e, accountId]]),
+        ...eras.map((e): QueryableStorageMultiArg<'rxjs'> => [api.query.staking.nominatorSlashInEra, [e, accountId]])
       ]).pipe(
         map((values): DeriveStakerSlashes[] =>
           eras.map((era, index) => ({
@@ -36,9 +36,7 @@ export function _ownSlashes (instanceId: string, api: ApiInterfaceRx): (accountI
 
 export function ownSlash (instanceId: string, api: ApiInterfaceRx): (accountId: Uint8Array | string, era: EraIndex) => Observable<DeriveStakerSlashes> {
   return memo(instanceId, (accountId: Uint8Array | string, era: EraIndex): Observable<DeriveStakerSlashes> =>
-    api.derive.staking._ownSlashes(accountId, [era], true).pipe(
-      map(([first]) => first)
-    )
+    firstObservable(api.derive.staking._ownSlashes(accountId, [era], true))
   );
 }
 

packages/api-derive/src/staking/query.ts

@@ -10,7 +10,7 @@ import type { DeriveStakingQuery, StakingQueryFlags } from '../types';
 
 import { combineLatest, map, of, switchMap } from 'rxjs';
 
-import { memo } from '../util';
+import { firstObservable, memo } from '../util';
 
 function parseDetails (stashId: AccountId, controllerIdOpt: Option<AccountId> | null, nominatorsOpt: Option<PalletStakingNominations>, rewardDestination: PalletStakingRewardDestination, validatorPrefs: PalletStakingValidatorPrefs, exposure: PalletStakingExposure, stakingLedgerOpt: Option<PalletStakingStakingLedger>): DeriveStakingQuery {
   return {
@@ -95,9 +95,7 @@ function getBatch (api: ApiInterfaceRx, activeEra: EraIndex, stashIds: AccountId
  */
 export function query (instanceId: string, api: ApiInterfaceRx): (accountId: Uint8Array | string, flags: StakingQueryFlags) => Observable<DeriveStakingQuery> {
   return memo(instanceId, (accountId: Uint8Array | string, flags: StakingQueryFlags): Observable<DeriveStakingQuery> =>
-    api.derive.staking.queryMulti([accountId], flags).pipe(
-      map(([first]) => first)
-    )
+    firstObservable(api.derive.staking.queryMulti([accountId], flags))
   );
 }
 

packages/api-derive/src/staking/stakerExposure.ts

@@ -8,7 +8,7 @@ import type { DeriveEraValidatorExposure, DeriveStakerExposure } from '../types'
 
 import { map, switchMap } from 'rxjs';
 
-import { memo } from '../util';
+import { firstObservable, memo } from '../util';
 
 export function _stakerExposures (instanceId: string, api: ApiInterfaceRx): (accountIds: (Uint8Array | string)[], eras: EraIndex[], withActive: boolean) => Observable<DeriveStakerExposure[][]> {
   return memo(instanceId, (accountIds: (Uint8Array | string)[], eras: EraIndex[], withActive: boolean): Observable<DeriveStakerExposure[][]> => {
@@ -48,8 +48,6 @@ export function stakerExposures (instanceId: string, api: ApiInterfaceRx): (acco
 
 export function stakerExposure (instanceId: string, api: ApiInterfaceRx): (accountId: Uint8Array | string, withActive?: boolean) => Observable<DeriveStakerExposure[]> {
   return memo(instanceId, (accountId: Uint8Array | string, withActive = false): Observable<DeriveStakerExposure[]> =>
-    api.derive.staking.stakerExposures([accountId], withActive).pipe(
-      map(([first]) => first)
-    )
+    firstObservable(api.derive.staking.stakerExposures([accountId], withActive))
   );
 }

packages/api-derive/src/staking/stakerPrefs.ts

@@ -14,7 +14,7 @@ import { erasHistoricApplyAccount } from './util';
 export function _stakerPrefs (instanceId: string, api: ApiInterfaceRx): (accountId: Uint8Array | string, eras: EraIndex[], withActive: boolean) => Observable<DeriveStakerPrefs[]> {
   // eslint-disable-next-line @typescript-eslint/no-unused-vars
   return memo(instanceId, (accountId: Uint8Array | string, eras: EraIndex[], _withActive: boolean): Observable<DeriveStakerPrefs[]> =>
-    api.query.staking.erasValidatorPrefs.multi(eras.map((era) => [era, accountId])).pipe(
+    api.query.staking.erasValidatorPrefs.multi(eras.map((e) => [e, accountId])).pipe(
       map((all): DeriveStakerPrefs[] =>
         all.map((validatorPrefs, index): DeriveStakerPrefs => ({
           era: eras[index],

packages/api-derive/src/staking/stakerRewards.ts

@@ -13,7 +13,7 @@ import { combineLatest, map, of, switchMap } from 'rxjs';
 
 import { BN_BILLION, BN_ZERO } from '@polkadot/util';
 
-import { memo } from '../util';
+import { firstObservable, memo } from '../util';
 
 type ErasResult = [DeriveEraPoints[], DeriveEraPrefs[], DeriveEraRewards[]];
 
@@ -98,7 +98,7 @@ function removeClaimed (validators: string[], queryValidators: DeriveStakingQuer
     if (index !== -1) {
       const valLedger = queryValidators[index].stakingLedger;
 
-      if (valLedger?.claimedRewards.some((era) => reward.era.eq(era))) {
+      if (valLedger?.claimedRewards.some((e) => reward.era.eq(e))) {
         rm.push(validatorId);
       }
     }
@@ -110,14 +110,14 @@ function removeClaimed (validators: string[], queryValidators: DeriveStakingQuer
 }
 
 function filterRewards (eras: EraIndex[], valInfo: [string, DeriveStakingQuery][], { rewards, stakingLedger }: { rewards: DeriveStakerReward[]; stakingLedger: PalletStakingStakingLedger }): DeriveStakerReward[] {
-  const filter = eras.filter((era) => !stakingLedger.claimedRewards.some((e) => e.eq(era)));
+  const filter = eras.filter((e) => !stakingLedger.claimedRewards.some((s) => s.eq(e)));
   const validators = valInfo.map(([v]) => v);
   const queryValidators = valInfo.map(([, q]) => q);
 
   return rewards
     .filter(({ isEmpty }) => !isEmpty)
     .filter((reward): boolean => {
-      if (!filter.some((filter) => reward.era.eq(filter))) {
+      if (!filter.some((e) => reward.era.eq(e))) {
         return false;
       }
 
@@ -186,10 +186,9 @@ export function _stakerRewards (instanceId: string, api: ApiInterfaceRx): (accou
 
 export function stakerRewards (instanceId: string, api: ApiInterfaceRx): (accountId: Uint8Array | string, withActive?: boolean) => Observable<DeriveStakerReward[]> {
   return memo(instanceId, (accountId: Uint8Array | string, withActive = false): Observable<DeriveStakerReward[]> =>
-    api.derive.staking.erasHistoric(withActive).pipe(
-      switchMap((eras) => api.derive.staking._stakerRewards([accountId], eras, withActive)),
-      map(([first]) => first)
-    )
+    firstObservable(api.derive.staking.erasHistoric(withActive).pipe(
+      switchMap((eras) => api.derive.staking._stakerRewards([accountId], eras, withActive))
+    ))
   );
 }
 

packages/api-derive/src/staking/util.ts

@@ -1,18 +1,19 @@
 // Copyright 2017-2021 @polkadot/api-derive authors & contributors
 // SPDX-License-Identifier: Apache-2.0
 
-import type { ApiInterfaceRx } from '@polkadot/api/types';
+import type { Observable } from 'rxjs';
+import type { ApiInterfaceRx, ObsInnerType } from '@polkadot/api/types';
 import type { EraIndex } from '@polkadot/types/interfaces';
 import type { ExactDerive } from '../bundle';
 
-import { switchMap } from 'rxjs';
+import { combineLatest, of, switchMap } from 'rxjs';
 
 import { memo } from '../util';
 
 type ApplyReturn<T extends keyof ExactDerive['staking']> = ReturnType<ExactDerive['staking'][T]>;
 
 export function filterEras <T extends { era: EraIndex }> (eras: EraIndex[], list: T[]): EraIndex[] {
-  return eras.filter((era) => !list.some((entry) => era.eq(entry.era)));
+  return eras.filter((e) => !list.some(({ era }) => e.eq(era)));
 }
 
 export function erasHistoricApply <F extends '_erasExposure' | '_erasPoints' | '_erasPrefs' | '_erasRewards' | '_erasSlashes'> (fn: F): (instanceId: string, api: ApiInterfaceRx) => (withActive?: boolean) => ApplyReturn<F> {
@@ -36,3 +37,15 @@ export function erasHistoricApplyAccount <F extends '_ownExposures' | '_ownSlash
       )
     ) as any;
 }
+
+export function combineEras <F extends '_eraExposure' | '_eraPrefs' | '_eraSlashes'> (fn: F): (instanceId: string, api: ApiInterfaceRx) => (eras: EraIndex[], withActive: boolean) => Observable<ObsInnerType<ApplyReturn<F>>[]> {
+  return (instanceId: string, api: ApiInterfaceRx) =>
+    // Cannot quite get the typing right, but it is right in the code
+    // eslint-disable-next-line @typescript-eslint/no-unsafe-return
+    memo(instanceId, (eras: EraIndex[], withActive: boolean) =>
+      // eslint-disable-next-line @typescript-eslint/no-unsafe-return
+      eras.length
+        ? combineLatest(eras.map((e) => api.derive.staking[fn](e, withActive)))
+        : of([])
+    ) as any;
+}

packages/api-derive/src/util/index.ts

@@ -1,8 +1,16 @@
 // Copyright 2017-2021 @polkadot/api-derive authors & contributors
 // SPDX-License-Identifier: Apache-2.0
 
+import type { Observable } from 'rxjs';
+
+import { map } from 'rxjs';
+
 export { drr, memo } from '@polkadot/rpc-core';
 
 export * from './approvalFlagsToBools';
 export * from './cache';
 export * from './cacheImpl';
+
+export function firstObservable <T> (obs: Observable<T[]>): Observable<T> {
+  return obs.pipe(map(([a]) => a));
+}