Add witness-related filters and providers

src/chain-observers/classifiers/index.ts

@@ -6,3 +6,4 @@ export { RcAccountClassifier } from "./rc-account-classifier";
 export { ImpactedAccountClassifier } from "./impacted-account-classifier";
 export { OperationClassifier } from "./operation-classifier";
 export { FeedPriceClassifier } from "./feed-price-classifier";
+export { WitnessClassifier } from "./witness-classifier";

src/chain-observers/classifiers/witness-classifier.ts

+import { CollectorClassifierBase } from "./collector-classifier-base";
+
+export interface IWitness {
+  owner: string;
+  runningVersion: string;
+  totalMissedBlocks: number;
+  lastConfirmedBlockNum: number;
+}
+
+export interface IWitnessData {
+  witnesses: Record<string, IWitness>;
+}
+
+export class WitnessClassifier extends CollectorClassifierBase {
+  public type!: IWitnessData;
+}

src/chain-observers/collectors/jsonrpc/witness-collector.ts

+import { IWitness } from "../../classifiers/witness-classifier";
+import { DataEvaluationContext } from "../../factories/data-evaluation-context";
+import { CollectorBase, TAvailableClassifiers } from "../collector-base";
+
+export interface IWitnessCollectorOptions {
+  witness: string;
+}
+
+const MAX_WITNESS_GET_LIMIT = 100;
+
+export class WitnessCollector extends CollectorBase {
+  private readonly witnesses: Record<string, number> = {};
+
+  protected pushOptions(data: IWitnessCollectorOptions): void {
+    this.witnesses[data.witness] = (this.witnesses[data.witness] || 0) + 1;
+  }
+
+  protected popOptions(data: IWitnessCollectorOptions): void {
+    this.witnesses[data.witness] = (this.witnesses[data.witness] || 1) - 1;
+
+    if (this.witnesses[data.witness] === 0)
+      delete this.witnesses[data.witness];
+  }
+
+  public async fetchData(_: DataEvaluationContext) {
+    const witnesses: Record<string, IWitness> = {};
+
+    const witnessNames = Object.keys(this.witnesses);
+    for (let i = 0; i < witnessNames.length; i += MAX_WITNESS_GET_LIMIT) {
+      const chunk = witnessNames.slice(i, i + MAX_WITNESS_GET_LIMIT);
+
+      const { witnesses: owners } = await this.worker.chain!.api.database_api.find_witnesses({ owners: chunk });
+
+      for(const account of owners)
+        witnessNames[account.owner] = {
+          name: account.owner,
+          runningVersion: account.running_version,
+          totalMissedBlocks: account.total_missed,
+          lastConfirmedBlockNum: account.last_confirmed_block_num
+        };
+    }
+
+    return {
+      WitnessClassifier: {
+        witnesses
+      }
+    } satisfies Partial<TAvailableClassifiers>;
+  };
+}

src/chain-observers/factories/jsonrpc/factory-data.ts

 import type { WorkerBee } from "../../../bot";
 import {
   AccountClassifier, BlockClassifier, BlockHeaderClassifier,
-  DynamicGlobalPropertiesClassifier, FeedPriceClassifier, ImpactedAccountClassifier, OperationClassifier, RcAccountClassifier
+  DynamicGlobalPropertiesClassifier, FeedPriceClassifier, ImpactedAccountClassifier, OperationClassifier, RcAccountClassifier,
+  WitnessClassifier
 } from "../../classifiers";
 import { IEvaluationContextClass } from "../../classifiers/collector-classifier-base";
 import { CollectorBase } from "../../collectors/collector-base";
@@ -13,6 +14,7 @@ import { BlockCollector } from "../../collectors/jsonrpc/block-collector";
 import { DynamicGlobalPropertiesCollector } from "../../collectors/jsonrpc/dynamic-global-properties-collector";
 import { FeedPriceCollector } from "../../collectors/jsonrpc/feed-price-collector";
 import { RcAccountCollector } from "../../collectors/jsonrpc/rc-account-collector";
+import { WitnessCollector } from "../../collectors/jsonrpc/witness-collector";
 
 export const JsonRpcFactoryData: (worker: WorkerBee) => Array<[IEvaluationContextClass, CollectorBase]> = (worker: WorkerBee) => ([
   [BlockHeaderClassifier, new BlockHeaderCollector(worker)],
@@ -22,5 +24,6 @@ export const JsonRpcFactoryData: (worker: WorkerBee) => Array<[IEvaluationContex
   [RcAccountClassifier, new RcAccountCollector(worker)],
   [ImpactedAccountClassifier, new ImpactedAccountCollector(worker)],
   [OperationClassifier, new OperationCollector(worker)],
-  [FeedPriceClassifier, new FeedPriceCollector(worker)]
+  [FeedPriceClassifier, new FeedPriceCollector(worker)],
+  [WitnessClassifier, new WitnessCollector(worker)]
 ]);

src/chain-observers/filters/witness-miss-block-filter.ts

+import type { WorkerBee } from "../../bot";
+import { WitnessClassifier } from "../classifiers";
+import type { TRegisterEvaluationContext } from "../classifiers/collector-classifier-base";
+import type { DataEvaluationContext } from "../factories/data-evaluation-context";
+import { FilterBase } from "./filter-base";
+
+export class WitnessMissedBlocksFilter extends FilterBase {
+  public constructor(
+    worker: WorkerBee,
+    private readonly witness: string,
+    private readonly missedBlocksCountMin: number
+  ) {
+    super(worker);
+  }
+
+  public usedContexts(): Array<TRegisterEvaluationContext> {
+    return [
+      WitnessClassifier.forOptions({ witness: this.witness })
+    ];
+  }
+
+  private initialMissedBlocksCount: number | undefined;
+  private previousLastBlockNumber: number | undefined;
+
+  public async match(data: DataEvaluationContext): Promise<boolean> {
+    const { witnesses } = await data.get(WitnessClassifier);
+
+    const witness = witnesses[this.witness];
+
+    if (this.previousLastBlockNumber === undefined) {
+      this.initialMissedBlocksCount = witness.totalMissedBlocks;
+      this.previousLastBlockNumber = witness.lastConfirmedBlockNum;
+
+      return false;
+    }
+
+    /*
+     * If witness missed more blocks than the minimum required and his last signed block number has not changed
+     * (he is not producing blocks)
+     */
+    if (this.previousLastBlockNumber === witness.lastConfirmedBlockNum
+      && this.initialMissedBlocksCount !== undefined
+      && witness.totalMissedBlocks > (this.initialMissedBlocksCount + this.missedBlocksCountMin)
+    ) {
+      // Reset missed blocks count to avoid multiple notifications for the same missed blocks streak
+      this.initialMissedBlocksCount = undefined;
+
+      return true;
+    }
+
+    // Update the initial missed blocks count if the last signed block number has changed - block missed streak reset
+    if (this.previousLastBlockNumber !== witness.lastConfirmedBlockNum)
+      this.initialMissedBlocksCount = witness.totalMissedBlocks;
+
+    this.previousLastBlockNumber = witness.lastConfirmedBlockNum;
+
+    return false;
+  }
+}

src/chain-observers/providers/witness-provider.ts

+import { TAccountName } from "@hiveio/wax";
+import { TRegisterEvaluationContext } from "../classifiers/collector-classifier-base";
+import { IWitness, WitnessClassifier } from "../classifiers/witness-classifier";
+import { DataEvaluationContext } from "../factories/data-evaluation-context";
+import { ProviderBase } from "./provider-base";
+
+export type TWitnessProvider<TAccounts extends Array<TAccountName>> = {
+  [K in TAccounts[number]]: IWitness;
+};
+
+export interface IWitnessProviderData<TAccounts extends Array<TAccountName>> {
+  witnesses: TWitnessProvider<TAccounts>;
+};
+
+export class WitnessProvider<TAccounts extends Array<TAccountName> = Array<TAccountName>> extends ProviderBase {
+  public constructor(
+    private readonly witnesses: TAccounts
+  ) {
+    super();
+  }
+
+  public usedContexts(): Array<TRegisterEvaluationContext> {
+    return this.witnesses.map(witness => WitnessClassifier.forOptions({
+      witness
+    }));
+  }
+
+  public async provide(data: DataEvaluationContext): Promise<IWitnessProviderData<TAccounts>> {
+    const result = {
+      witnesses: {}
+    };
+
+    const { witnesses } = await data.get(WitnessClassifier);
+    for(const witness of this.witnesses)
+      result.witnesses[witness] = witnesses[witness];
+
+    return result as IWitnessProviderData<TAccounts>;
+  }
+}

src/queen.ts

@@ -21,6 +21,7 @@ import { ReblogFilter } from "./chain-observers/filters/reblog-filter";
 import { TransactionIdFilter } from "./chain-observers/filters/transaction-id-filter";
 import { VoteFilter } from "./chain-observers/filters/vote-filter";
 import { WhaleAlertFilter } from "./chain-observers/filters/whale-alert-filter";
+import { WitnessMissedBlocksFilter } from "./chain-observers/filters/witness-miss-block-filter";
 import { AccountProvider } from "./chain-observers/providers/account-provider";
 import { BlockHeaderProvider } from "./chain-observers/providers/block-header-provider";
 import { BlockProvider } from "./chain-observers/providers/block-provider";
@@ -31,6 +32,7 @@ import { ProviderBase } from "./chain-observers/providers/provider-base";
 import { RcAccountProvider } from "./chain-observers/providers/rc-account-provider";
 import { TransactionByIdProvider } from "./chain-observers/providers/transaction-provider";
 import { WhaleAlertProvider } from "./chain-observers/providers/whale-alert-provider";
+import { WitnessProvider } from "./chain-observers/providers/witness-provider";
 import type { Observer, Unsubscribable } from "./types/subscribable";
 
 export class QueenBee<TPreviousSubscriberData extends object = {}> {
@@ -182,6 +184,12 @@ export class QueenBee<TPreviousSubscriberData extends object = {}> {
     return this;
   }
 
+  public onWitnessMissedBlocks(witness: TAccountName, missedBlocksMinCount: number): QueenBee<TPreviousSubscriberData> {
+    this.operands.push(new WitnessMissedBlocksFilter(this.worker, witness, missedBlocksMinCount));
+
+    return this;
+  }
+
   public onBlock(): QueenBee<TPreviousSubscriberData & Awaited<ReturnType<BlockHeaderProvider["provide"]>>> {
     this.operands.push(new BlockChangedFilter(this.worker));
     this.providers.push(new BlockHeaderProvider());
@@ -197,6 +205,14 @@ export class QueenBee<TPreviousSubscriberData extends object = {}> {
     return this;
   }
 
+  public provideWitnesses<
+    TAccounts extends Array<TAccountName>
+  >(...witnesses: TAccounts): QueenBee<TPreviousSubscriberData & Awaited<ReturnType<WitnessProvider<TAccounts>["provide"]>>> {
+    this.providers.push(new WitnessProvider<TAccounts>(witnesses));
+
+    return this;
+  }
+
   public provideRcAccounts<
     TAccounts extends Array<TAccountName>
   >(...accounts: TAccounts): QueenBee<TPreviousSubscriberData & Awaited<ReturnType<RcAccountProvider<TAccounts>["provide"]>>> {

src/wax/index.ts

@@ -11,7 +11,17 @@ export type WaxExtendTypes = {
         current_max_history: price;
         price_history: price[];
       }
-    }
+    };
+    find_witnesses: {
+      params: { owners: string[]; };
+      result: { witnesses: Array<{
+        owner: string;
+        total_missed: number;
+        running_version: string;
+        last_confirmed_block_num: number;
+        // ...
+      }>; };
+    };
   }
 };