Liquid: Fix audit updating conditions

frontend/src/app/components/liquid-reserves-audit/federation-addresses-list/federation-addresses-list.component.ts

@@ -1,9 +1,9 @@
 import { Component, OnInit, ChangeDetectionStrategy, Input } from '@angular/core';
-import { Observable, concat } from 'rxjs';
+import { Observable, combineLatest, concat, of } from 'rxjs';
 import { delay, filter, map, share, skip, switchMap, tap, throttleTime } from 'rxjs/operators';
 import { ApiService } from '../../../services/api.service';
 import { Env, StateService } from '../../../services/state.service';
-import { AuditStatus, FederationAddress } from '../../../interfaces/node-api.interface';
+import { AuditStatus, CurrentPegs, FederationAddress } from '../../../interfaces/node-api.interface';
 import { WebsocketService } from '../../../services/websocket.service';
 
 @Component({
@@ -25,6 +25,9 @@ export class FederationAddressesListComponent implements OnInit {
   auditStatus$: Observable<AuditStatus>;
   auditUpdated$: Observable<boolean>;
   lastReservesBlockUpdate: number = 0;
+  currentPeg$: Observable<CurrentPegs>;
+  lastPegBlockUpdate: number = 0;
+  lastPegAmount: string = '';
 
   constructor(
     private apiService: ApiService,
@@ -40,7 +43,7 @@ export class FederationAddressesListComponent implements OnInit {
     if (!this.widget) {
       this.websocketService.want(['blocks']);
       this.auditStatus$ = concat(
-        this.apiService.federationAuditSynced$(),
+        this.apiService.federationAuditSynced$().pipe(share()),
         this.stateService.blocks$.pipe(
           skip(1),
           throttleTime(40000),
@@ -50,17 +53,41 @@ export class FederationAddressesListComponent implements OnInit {
         )
       );
 
-      this.auditUpdated$ = this.auditStatus$.pipe(
+      this.currentPeg$ = this.auditStatus$.pipe(
         filter(auditStatus => auditStatus.isAuditSynced === true),
-        map(auditStatus => {
+        switchMap(_ =>
-          const beforeLastBlockAudit = this.lastReservesBlockUpdate;
+          this.apiService.liquidPegs$().pipe(
-          this.lastReservesBlockUpdate = auditStatus.lastBlockAudit;
+            filter((currentPegs) => currentPegs.lastBlockUpdate >= this.lastPegBlockUpdate),
-          return auditStatus.lastBlockAudit > beforeLastBlockAudit ? true : false;
+            tap((currentPegs) => {
-        })
+              this.lastPegBlockUpdate = currentPegs.lastBlockUpdate;
+            })
+          )
+        ),
+        share()
+      );
+
+      this.auditUpdated$ = combineLatest([
+        this.auditStatus$,
+        this.currentPeg$
+      ]).pipe(
+        filter(([auditStatus, _]) => auditStatus.isAuditSynced === true),
+        map(([auditStatus, currentPeg]) => ({
+          lastBlockAudit: auditStatus.lastBlockAudit,
+          currentPegAmount: currentPeg.amount
+        })),
+        switchMap(({ lastBlockAudit, currentPegAmount }) => {
+          const blockAuditCheck = lastBlockAudit > this.lastReservesBlockUpdate;
+          const amountCheck = currentPegAmount !== this.lastPegAmount;
+          this.lastReservesBlockUpdate = lastBlockAudit;
+          this.lastPegAmount = currentPegAmount;
+          return of(blockAuditCheck || amountCheck);
+        }),
+        share()
       );
 
       this.federationAddresses$ = this.auditUpdated$.pipe(
         filter(auditUpdated => auditUpdated === true),
+        throttleTime(40000),
         switchMap(_ => this.apiService.federationAddresses$()),
         tap(_ => this.isLoading = false),
         share()

frontend/src/app/components/liquid-reserves-audit/federation-utxos-list/federation-utxos-list.component.ts

@@ -1,9 +1,9 @@
 import { Component, OnInit, ChangeDetectionStrategy, Input } from '@angular/core';
-import { Observable, concat } from 'rxjs';
+import { Observable, combineLatest, concat, of } from 'rxjs';
 import { delay, filter, map, share, skip, switchMap, tap, throttleTime } from 'rxjs/operators';
 import { ApiService } from '../../../services/api.service';
 import { Env, StateService } from '../../../services/state.service';
-import { AuditStatus, FederationUtxo } from '../../../interfaces/node-api.interface';
+import { AuditStatus, CurrentPegs, FederationUtxo } from '../../../interfaces/node-api.interface';
 import { WebsocketService } from '../../../services/websocket.service';
 
 @Component({
@@ -25,6 +25,9 @@ export class FederationUtxosListComponent implements OnInit {
   auditStatus$: Observable<AuditStatus>;
   auditUpdated$: Observable<boolean>;
   lastReservesBlockUpdate: number = 0;
+  currentPeg$: Observable<CurrentPegs>;
+  lastPegBlockUpdate: number = 0;
+  lastPegAmount: string = '';
 
   constructor(
     private apiService: ApiService,
@@ -40,7 +43,7 @@ export class FederationUtxosListComponent implements OnInit {
     if (!this.widget) {
       this.websocketService.want(['blocks']);
       this.auditStatus$ = concat(
-        this.apiService.federationAuditSynced$(),
+        this.apiService.federationAuditSynced$().pipe(share()),
         this.stateService.blocks$.pipe(
           skip(1),
           throttleTime(40000),
@@ -50,17 +53,41 @@ export class FederationUtxosListComponent implements OnInit {
         )
       );
 
-      this.auditUpdated$ = this.auditStatus$.pipe(
+      this.currentPeg$ = this.auditStatus$.pipe(
         filter(auditStatus => auditStatus.isAuditSynced === true),
-        map(auditStatus => {
+        switchMap(_ =>
-          const beforeLastBlockAudit = this.lastReservesBlockUpdate;
+          this.apiService.liquidPegs$().pipe(
-          this.lastReservesBlockUpdate = auditStatus.lastBlockAudit;
+            filter((currentPegs) => currentPegs.lastBlockUpdate >= this.lastPegBlockUpdate),
-          return auditStatus.lastBlockAudit > beforeLastBlockAudit ? true : false;
+            tap((currentPegs) => {
-        })
+              this.lastPegBlockUpdate = currentPegs.lastBlockUpdate;
+            })
+          )
+        ),
+        share()
+      );
+
+      this.auditUpdated$ = combineLatest([
+        this.auditStatus$,
+        this.currentPeg$
+      ]).pipe(
+        filter(([auditStatus, _]) => auditStatus.isAuditSynced === true),
+        map(([auditStatus, currentPeg]) => ({
+          lastBlockAudit: auditStatus.lastBlockAudit,
+          currentPegAmount: currentPeg.amount
+        })),
+        switchMap(({ lastBlockAudit, currentPegAmount }) => {
+          const blockAuditCheck = lastBlockAudit > this.lastReservesBlockUpdate;
+          const amountCheck = currentPegAmount !== this.lastPegAmount;
+          this.lastReservesBlockUpdate = lastBlockAudit;
+          this.lastPegAmount = currentPegAmount;
+          return of(blockAuditCheck || amountCheck);
+        }),
+        share()
       );
 
       this.federationUtxos$ = this.auditUpdated$.pipe(
         filter(auditUpdated => auditUpdated === true),
+        throttleTime(40000),
         switchMap(_ => this.apiService.federationUtxos$()),
         tap(_ => this.isLoading = false),
         share()

frontend/src/app/components/liquid-reserves-audit/reserves-audit-dashboard/reserves-audit-dashboard.component.ts

@@ -25,6 +25,7 @@ export class ReservesAuditDashboardComponent implements OnInit {
   liquidReservesMonth$: Observable<any>;
   fullHistory$: Observable<any>;
   private lastPegBlockUpdate: number = 0;
+  private lastPegAmount: string = '';
   private lastReservesBlockUpdate: number = 0;
 
 
@@ -51,32 +52,11 @@ export class ReservesAuditDashboardComponent implements OnInit {
       )
     );
 
-    this.auditUpdated$ = this.auditStatus$.pipe(
-      filter(auditStatus => auditStatus.isAuditSynced === true),
-      map(auditStatus => auditStatus.lastBlockAudit),
-      switchMap((lastBlockAudit) => {
-        return lastBlockAudit > this.lastReservesBlockUpdate ? of(true) : of(false);
-      }),
-    );
-
-    this.currentReserves$ = this.auditUpdated$.pipe(
-      filter(auditUpdated => auditUpdated === true),
-      switchMap(_ =>
-        this.apiService.liquidReserves$().pipe(
-          filter((currentReserves) => currentReserves.lastBlockUpdate > this.lastReservesBlockUpdate),
-          tap((currentReserves) => {
-            this.lastReservesBlockUpdate = currentReserves.lastBlockUpdate;
-          })
-        )
-      ),
-      share()
-    );
-
     this.currentPeg$ = this.auditStatus$.pipe(
       filter(auditStatus => auditStatus.isAuditSynced === true),
       switchMap(_ =>
         this.apiService.liquidPegs$().pipe(
-          filter((currentPegs) => currentPegs.lastBlockUpdate > this.lastPegBlockUpdate),
+          filter((currentPegs) => currentPegs.lastBlockUpdate >= this.lastPegBlockUpdate),
           tap((currentPegs) => {
             this.lastPegBlockUpdate = currentPegs.lastBlockUpdate;
           })
@@ -85,14 +65,48 @@ export class ReservesAuditDashboardComponent implements OnInit {
       share()
     );
 
+    this.auditUpdated$ = combineLatest([
+      this.auditStatus$,
+      this.currentPeg$
+    ]).pipe(
+      filter(([auditStatus, _]) => auditStatus.isAuditSynced === true),
+      map(([auditStatus, currentPeg]) => ({
+        lastBlockAudit: auditStatus.lastBlockAudit,
+        currentPegAmount: currentPeg.amount
+      })), 
+      switchMap(({ lastBlockAudit, currentPegAmount }) => {
+        const blockAuditCheck = lastBlockAudit > this.lastReservesBlockUpdate;
+        const amountCheck = currentPegAmount !== this.lastPegAmount;
+        this.lastPegAmount = currentPegAmount;
+        return of(blockAuditCheck || amountCheck);
+      }),
+      share()
+    );
+
+    this.currentReserves$ = this.auditUpdated$.pipe(
+      filter(auditUpdated => auditUpdated === true),
+      throttleTime(40000),
+      switchMap(_ =>
+        this.apiService.liquidReserves$().pipe(
+          filter((currentReserves) => currentReserves.lastBlockUpdate >= this.lastReservesBlockUpdate),
+          tap((currentReserves) => {
+            this.lastReservesBlockUpdate = currentReserves.lastBlockUpdate;
+          })
+        )
+      ),
+      share()
+    );
+
     this.federationUtxos$ = this.auditUpdated$.pipe(
       filter(auditUpdated => auditUpdated === true),
+      throttleTime(40000),
       switchMap(_ => this.apiService.federationUtxos$()),
       share()
     );
 
     this.federationAddresses$ = this.auditUpdated$.pipe(
       filter(auditUpdated => auditUpdated === true),
+      throttleTime(40000),
       switchMap(_ => this.apiService.federationAddresses$()),
       share()
     );

frontend/src/app/dashboard/dashboard.component.ts

@@ -56,6 +56,7 @@ export class DashboardComponent implements OnInit, OnDestroy, AfterViewInit {
   currencySubscription: Subscription;
   currency: string;
   private lastPegBlockUpdate: number = 0;
+  private lastPegAmount: string = '';
   private lastReservesBlockUpdate: number = 0;
 
   constructor(
@@ -239,9 +240,11 @@ export class DashboardComponent implements OnInit, OnDestroy, AfterViewInit {
         // Or when we receive a newer block, we wait 2 seconds so that the backend updates and we fetch the current peg
         this.stateService.blocks$
           .pipe(
+            skip(1),
+            throttleTime(40000),
             delay(2000),
             switchMap((_) => this.apiService.liquidPegs$()),
-            filter((currentPeg) => currentPeg.lastBlockUpdate > this.lastPegBlockUpdate),
+            filter((currentPeg) => currentPeg.lastBlockUpdate >= this.lastPegBlockUpdate),
             tap((currentPeg) => this.lastPegBlockUpdate = currentPeg.lastBlockUpdate)
           )
       ).pipe(
@@ -260,12 +263,23 @@ export class DashboardComponent implements OnInit, OnDestroy, AfterViewInit {
         )
       );
 
-      this.auditUpdated$ = this.auditStatus$.pipe(
+      this.auditUpdated$ = combineLatest([
-        filter(auditStatus => auditStatus.isAuditSynced === true),
+        this.auditStatus$,
-        map(auditStatus => auditStatus.lastBlockAudit),
+        this.currentPeg$
-        switchMap((lastBlockAudit) => {
+      ]).pipe(
-          return lastBlockAudit > this.lastReservesBlockUpdate ? of(true) : of(false);
+        filter(([auditStatus, _]) => auditStatus.isAuditSynced === true),
-        }),
+        map(([auditStatus, currentPeg]) => ({
+          lastBlockAudit: auditStatus.lastBlockAudit,
+          currentPegAmount: currentPeg.amount
+        })), 
+        switchMap(({ lastBlockAudit, currentPegAmount }) => {
+          console.log(lastBlockAudit, this.lastReservesBlockUpdate, currentPegAmount, this.lastPegAmount)
+          const blockAuditCheck = lastBlockAudit > this.lastReservesBlockUpdate;
+          const amountCheck = currentPegAmount !== this.lastPegAmount;
+          this.lastPegAmount = currentPegAmount;
+          console.log(blockAuditCheck || amountCheck)
+          return of(blockAuditCheck || amountCheck);
+        })
       );
 
       this.liquidReservesMonth$ = interval(60 * 60 * 1000).pipe(
@@ -287,9 +301,10 @@ export class DashboardComponent implements OnInit, OnDestroy, AfterViewInit {
 
       this.currentReserves$ = this.auditUpdated$.pipe(
         filter(auditUpdated => auditUpdated === true),
+        throttleTime(40000),
         switchMap(_ =>
           this.apiService.liquidReserves$().pipe(
-            filter((currentReserves) => currentReserves.lastBlockUpdate > this.lastReservesBlockUpdate),
+            filter((currentReserves) => currentReserves.lastBlockUpdate >= this.lastReservesBlockUpdate),
             tap((currentReserves) => {
               this.lastReservesBlockUpdate = currentReserves.lastBlockUpdate;
             })