Standardised communication ion the coin indexer to be simulat to the other workers

Signed-off-by: Robert Gogete <gogeterobert@yahoo.com>

Author: gogeterobert

src/application/workers/CoinIndexer/CoinIndexer.ts

@@ -3,6 +3,7 @@ import { spawn, Worker, Thread } from 'threads';
 import { IWorker } from '../IWorker';
 import {
   CoinIndexerEventNames,
+  CoinIndexerEvents,
   CoinStateUpdatedEvent,
 } from './CoinIndexerEvents';
 
@@ -16,7 +17,7 @@ interface ICoinIndexer extends IWorker {
 }
 
 export class CoinIndexer
-  extends EventEmitter
+  extends (EventEmitter as { new (): CoinIndexerEvents })
   implements ICoinIndexer
 {
   private worker: import('threads').ModuleThread<CoinIndexerWorkerApi> | null = null;
@@ -54,8 +55,10 @@ export class CoinIndexer
       )) as import('threads').ModuleThread<CoinIndexerWorkerApi>;
     }
 
-    this.worker.onCoinStateUpdated().subscribe((coinState: CoinStateUpdatedEvent) => {
-      this.emit(CoinIndexerEventNames.CoinStateUpdated, coinState);
+    this.worker.onCoinStateUpdated().subscribe({
+      next: (coinState: CoinStateUpdatedEvent) => {
+        this.emit(CoinIndexerEventNames.CoinStateUpdated, coinState);
+      },
     });
 
     try {

src/application/workers/CoinIndexer/CoinIndexer.worker.logic.ts

@@ -1,5 +1,6 @@
 import Database from 'better-sqlite3';
-import { CoinIndexerEvents, CoinStateUpdatedEvent } from './CoinIndexerEvents';
+import { Observable } from 'observable-fns';
+import { CoinStateUpdatedEvent } from './CoinIndexerEvents';
 import { CoinRepository, CoinRow } from '../../repositories/CoinRepository';
 import { WalletRepository, WalletRow } from '../../repositories/WalletRepository';
 import { IBlockchainService } from '../../interfaces/IBlockChainService';
@@ -11,11 +12,13 @@ import { ChiaBlockchainService } from '../../../infrastructure/BlockchainService
 let db: Database.Database | null = null;
 let coinRepo: CoinRepository | null = null;
 let walletRepo: WalletRepository | null = null;
-let events: CoinIndexerEvents | null = null;
 let started = false;
 let intervalId: NodeJS.Timeout | null = null;
 let blockchainService: IBlockchainService | null = null;
 
+let coinStateObservable: Observable<CoinStateUpdatedEvent> | null = null;
+let coinStateObserver: ((event: CoinStateUpdatedEvent) => void) | null = null;
+
 function mapUnspentCoinToDbFields(
   coin: Coin,
   wallet_id: string,
@@ -34,7 +37,7 @@ function mapUnspentCoinToDbFields(
 }
 
 async function sync() {
-  if (!coinRepo || !walletRepo || !events || !blockchainService) return;
+  if (!coinRepo || !walletRepo || !blockchainService) return;
   const wallets: WalletRow[] = walletRepo.getWallets();
 
   let peer: Peer | null = null; // TODO Update this by getting a value from PeerCluster from datalayer driver
@@ -54,12 +57,14 @@ async function sync() {
         wallet.synced_to_height || 0,
       );
       coinRepo.upsertCoin(wallet.address, mapped);
-      events.emitCoinStateUpdated({
-        wallet_id: wallet.address,
-        coinId: mapped.coinId,
-        status: 'unspent',
-        synced_height: wallet.synced_to_height || 0,
-      });
+      if (coinStateObserver) {
+        coinStateObserver({
+          wallet_id: wallet.address,
+          coinId: mapped.coinId,
+          status: 'unspent',
+          synced_height: wallet.synced_to_height || 0,
+        });
+      }
     }
   }
   // Check pending coins
@@ -73,23 +78,24 @@ async function sync() {
     );
     if (!spendable) {
       coinRepo.updateCoinStatus(coin.wallet_id, coin.coinId, 'spent', coin.synced_height);
-      events.emitCoinStateUpdated({
-        wallet_id: coin.wallet_id,
-        coinId: coin.coinId,
-        status: 'spent',
-        synced_height: coin.synced_height,
-      });
+      if (coinStateObserver) {
+        coinStateObserver({
+          wallet_id: coin.wallet_id,
+          coinId: coin.coinId,
+          status: 'spent',
+          synced_height: coin.synced_height,
+        });
+      }
     }
   }
 }
 
 export const api = {
-  async start(_blockchainType: string, dbPath: string = './coin_indexer.sqlite') {
+  async start(_blockchainType: BlockChainType, dbPath: string = './coin_indexer.sqlite') {
     if (started) return;
     db = new Database(dbPath);
     coinRepo = new CoinRepository(db);
     walletRepo = new WalletRepository(db);
-    events = new CoinIndexerEvents();
 
     switch (_blockchainType) {
       case BlockChainType.Test:
@@ -101,7 +107,6 @@ export const api = {
         break;
     }
 
-    // peer = ... (should be set externally or passed in)
     started = true;
     intervalId = setInterval(sync, 1000);
   },
@@ -111,20 +116,31 @@ export const api = {
     db = null;
     coinRepo = null;
     walletRepo = null;
-    events = null;
     blockchainService = null;
+    coinStateObservable = null;
+    coinStateObserver = null;
   },
-  onCoinStateUpdated(listener: (event: CoinStateUpdatedEvent) => void) {
-    if (!events) throw new Error('CoinIndexerEvents not initialized');
-    events.onCoinStateUpdated(listener);
+  onCoinStateUpdated() {
+    if (!coinStateObservable) {
+      coinStateObservable = new Observable<CoinStateUpdatedEvent>((observer) => {
+        coinStateObserver = (event: CoinStateUpdatedEvent) => {
+          observer.next(event);
+        };
+        return () => {
+          coinStateObserver = null;
+        };
+      });
+    }
+    return coinStateObservable;
   },
   __reset() {
     if (intervalId) clearInterval(intervalId);
     db = null;
     coinRepo = null;
     walletRepo = null;
-    events = null;
     blockchainService = null;
     started = false;
+    coinStateObservable = null;
+    coinStateObserver = null;
   },
 };

src/application/workers/CoinIndexer/CoinIndexerEvents.ts

@@ -1,7 +1,5 @@
-import { EventEmitter } from 'events';
-
 export enum CoinIndexerEventNames {
-  CoinStateUpdated = 'CoinStateUpdated',
+  CoinStateUpdated = 'coinStateUpdated',
 }
 
 export interface CoinStateUpdatedEvent {
@@ -11,11 +9,7 @@ export interface CoinStateUpdatedEvent {
   synced_height: number;
 }
 
-export class CoinIndexerEvents extends EventEmitter {
-  emitCoinStateUpdated(event: CoinStateUpdatedEvent) {
-    this.emit(CoinIndexerEventNames.CoinStateUpdated, event);
-  }
-  onCoinStateUpdated(listener: (event: CoinStateUpdatedEvent) => void) {
-    this.on(CoinIndexerEventNames.CoinStateUpdated, listener);
-  }
+export interface CoinIndexerEvents {
+  on(event: CoinIndexerEventNames.CoinStateUpdated, listener: (event: CoinStateUpdatedEvent) => void): this;
+  emit(event: CoinIndexerEventNames.CoinStateUpdated, eventData: CoinStateUpdatedEvent): boolean;
 }

test/application/workers/BlockIndexer/BlockIndexer.worker.logic.test.ts

@@ -3,20 +3,6 @@ import Database from 'better-sqlite3';
 import { BlockChainType } from '../../../../src/application/types/BlockChain';
 import fs from 'fs';
 
-// Mock BlockchainService for unit tests
-class MockBlockchainService {
-  private blocks: any[] = [];
-  constructor(blocks: any[] = []) {
-    this.blocks = blocks;
-  }
-  async getCurrentBlockchainHeight() {
-    return this.blocks.length;
-  }
-  async getBlockchainBlockByHeight(h: number) {
-    return this.blocks[h - 1] || null;
-  }
-}
-
 describe('BlockIndexer.worker.logic api', () => {
   const dbPath = 'test_blockindexer_worker_logic.sqlite';
 

test/application/workers/CoinIndexer/CoinIndexer.worker.logic.test.ts

@@ -1,62 +1,46 @@
 import Database from 'better-sqlite3';
 import { api as coinIndexerApi } from '../../../../src/application/workers/CoinIndexer/CoinIndexer.worker.logic';
-import { CoinStateUpdatedEvent } from '../../../../src/application/workers/CoinIndexer/CoinIndexerEvents';
-import { CoinRepository } from '../../../../src/application/repositories/CoinRepository';
-import { WalletRepository } from '../../../../src/application/repositories/WalletRepository';
-import type { IBlockchainService } from '../../../../src/application/interfaces/IBlockChainService';
-import type { Peer, Coin } from '@dignetwork/datalayer-driver';
+import { BlockChainType } from '../../../../src/application/types/BlockChain';
+import { existsSync, unlinkSync } from 'fs';
 
-describe('CoinIndexer.worker.logic', () => {
-  const dbPath = ':memory:';
-  let db: Database.Database;
-  let coinRepo: CoinRepository;
-  let walletRepo: WalletRepository;
-  let mockService: IBlockchainService;
-  let mockPeer: Peer;
+const dbPath = 'test_coinindexer_worker_logic.sqlite';
 
-  beforeEach(async () => {
-    coinIndexerApi.__reset();
-    db = new Database(dbPath);
-    coinRepo = new CoinRepository(db);
-    walletRepo = new WalletRepository(db);
-    // Add a wallet for tests
-    walletRepo.addWallet('xch1234');
-    // Mock blockchain service and peer
-    mockService = {
-      listUnspentCoins: jest.fn().mockResolvedValue({ coins: [] }),
-      isCoinSpendable: jest.fn().mockResolvedValue(true),
-      // ...other methods can be no-ops
-    } as any;
-    mockPeer = {} as Peer;
-    // If needed, inject mockService and mockPeer into the worker here
-    await coinIndexerApi.start('Test', dbPath);
+describe('CoinIndexer.worker.logic api', () => {
+  beforeAll(() => {
+    if (existsSync(dbPath)) unlinkSync(dbPath);
   });
-
+  
   afterEach(() => {
-    coinIndexerApi.stop();
+    try { new Database(dbPath).close(); } catch {}
+  });
+
+  it('should create the database file after start', async () => {
     coinIndexerApi.__reset();
+    if (existsSync(dbPath)) unlinkSync(dbPath);
+    await coinIndexerApi.start(BlockChainType.Test, dbPath);
+    expect(existsSync(dbPath)).toBe(true);
+    // Check table exists (should be named 'coin' not 'coins')
+    const db = new Database(dbPath);
+    const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='coin'").get();
+    expect(tables).toBeDefined();
     db.close();
+    coinIndexerApi.stop();
+  });
+
+  it('should not start twice', async () => {
+    coinIndexerApi.__reset();
+    await coinIndexerApi.start(BlockChainType.Test, dbPath);
+    await coinIndexerApi.start(BlockChainType.Test, dbPath); // should not throw
+    coinIndexerApi.stop();
   });
 
-  it('should emit CoinStateUpdated event on sync', async () => {
-    // Arrange: mock listUnspentCoins to return a coin
-    const coin: Coin = {
-      coin_id: Buffer.from('aabbcc', 'hex'),
-      parentCoinInfo: Buffer.from('ddeeff', 'hex'),
-      puzzleHash: Buffer.from('112233', 'hex'),
-      amount: BigInt(1000),
-    } as any;
-    (mockService.listUnspentCoins as jest.Mock).mockResolvedValue({ coins: [coin] });
-    walletRepo.updateWalletSync('xch1234', 10, 'abc');
-    const eventPromise = new Promise<void>((resolve) => {
-      coinIndexerApi.onCoinStateUpdated((event: CoinStateUpdatedEvent) => {
-        expect(event.wallet_id).toBe('xch1234');
-        expect(event.status).toBe('unspent');
-        resolve();
-      });
-    });
-    // Wait for the interval to trigger sync (simulate passage of time)
-    await new Promise((r) => setTimeout(r, 1100));
-    await eventPromise;
+  it('should stop and reset', async () => {
+    coinIndexerApi.__reset();
+    await coinIndexerApi.start(BlockChainType.Test, dbPath);
+    coinIndexerApi.stop();
+    coinIndexerApi.__reset();
+    // Should be able to start again
+    await coinIndexerApi.start(BlockChainType.Test, dbPath);
+    coinIndexerApi.stop();
   });
 });