heartbeat updates

src/ORM/client/client.service.ts

@@ -9,7 +9,7 @@ import { ClientEntity } from './client.entity';
 @Injectable()
 export class ClientService {
 
-
+    private heartbeatBulkUpdate: { [id: string]:{id: string, hashRate: number, updatedAt: Date}} = {};
 
     constructor(
         @InjectRepository(ClientEntity)
@@ -28,13 +28,61 @@ export class ClientService {
             .execute();
     }
 
-    public async heartbeat(id, hashRate: number, updatedAt: Date) {
+    //public async heartbeat(id, hashRate: number, updatedAt: Date) {
-        return await this.clientRepository.update({ id }, { hashRate, deletedAt: null, updatedAt });
+    //     return await this.clientRepository.update({ id }, { hashRate, deletedAt: null, updatedAt });
+    // }
+
+    public heartbeatBulkAsync(id, hashRate: number, updatedAt: Date){
+        if(this.heartbeatBulkUpdate[id] != null){
+            this.heartbeatBulkUpdate[id].hashRate = hashRate;
+            this.heartbeatBulkUpdate[id].updatedAt = updatedAt;
+            return;
+        }
+        this.heartbeatBulkUpdate[id] = {id, hashRate, updatedAt};
     }
 
-    // public async save(client: Partial<ClientEntity>) {
+    public async doBulkHeartbeatUpdate(){
-    //     return await this.clientRepository.save(client);
+        if(Object.keys(this.heartbeatBulkUpdate).length < 1){
-    // }
+            console.log('No heartbeats to update.')
+            return;
+        }
+
+        const values = Object.entries(this.heartbeatBulkUpdate).map(([key, value]) => {
+            return  `('${value.id}', ${value.hashRate}, '${value.updatedAt.toISOString()}')`
+        }).join(',');
+
+        const query = `
+            DO $$
+            BEGIN
+                CREATE TEMP TABLE temp_heartbeats (
+                    id UUID,
+                    "hashRate" DECIMAL,
+                    "updatedAt" TIMESTAMP
+                ) ON COMMIT DROP;
+
+                INSERT INTO temp_heartbeats (id, "hashRate", "updatedAt")
+                VALUES ${values};
+
+                UPDATE "client_entity" ce
+                SET "hashRate" = th."hashRate",
+                    "deletedAt" = NULL,
+                    "updatedAt" = th."updatedAt"
+                FROM temp_heartbeats th
+                WHERE ce.id = th.id;
+            END;
+            $$;
+        `;
+
+        try {
+            await this.clientRepository.query(query);
+            console.log(`Bulk updated ${this.heartbeatBulkUpdate.length} client heartbeats`);
+        } catch (error) {
+            console.error('Bulk heartbeat failed:', error.message, 'Query:', query);
+            throw error;
+        }
+
+        this.heartbeatBulkUpdate = {};
+    }
 
 
     public async insert(partialClient: Partial<ClientEntity>): Promise<ClientEntity> {

src/models/StratumV1Client.ts

@@ -545,10 +545,10 @@ export class StratumV1Client {
                 await this.statistics.addShares(this.clientEntity, this.sessionDifficulty);
                 const now = new Date();
                 // only update every minute
-                if (this.clientEntity.updatedAt == null || now.getTime() - this.clientEntity.updatedAt.getTime() > 1000 * 60) {
+                //if (this.clientEntity.updatedAt == null || now.getTime() - this.clientEntity.updatedAt.getTime() > 1000 * 60) {
-                    await this.clientService.heartbeat(this.clientEntity.id, this.hashRate, now);
+                await this.clientService.heartbeatBulkAsync(this.clientEntity.id, this.hashRate, now);
                 this.clientEntity.updatedAt = now;
-                }
+                //}
 
             } catch (e) {
                 console.log(e);

src/services/app.service.ts

@@ -55,6 +55,11 @@ export class AppService implements OnModuleInit {
             await this.clientStatisticsService.doBulkAsyncUpdate();
         }, 1000 * 60 * 1);
 
+        setInterval(async () => {
+            //console.log('Bulk update client stats');
+            await this.clientService.doBulkHeartbeatUpdate();
+        }, 1000 * 60 * 1);
+
 
     }