wip DI implemented, Queue helpers

.gitignore

@@ -6,3 +6,4 @@ node_modules/
 kovan.*
 dump.rdb
 .idea
+build

package.json

@@ -31,7 +31,9 @@
     "ioredis": "^4.14.1",
     "json-schema-to-ts": "^2.2.0",
     "node-fetch": "^2.6.7",
+    "reflect-metadata": "^0.1.13",
     "torn-token": "link:../torn-token",
+    "tsyringe": "^4.6.0",
     "tx-manager": "link:../tx-manager",
     "uuid": "^8.3.0",
     "web3": "^1.3.0",

src/app/index.ts

@@ -1,3 +1,4 @@
+import 'reflect-metadata';
 import createServer from './server';
 import { utils } from 'ethers';
 import { port, rewardAccount } from '../config';

src/app/routes.ts

@@ -4,12 +4,13 @@ import { FromSchema } from 'json-schema-to-ts';
 import { rewardAccount, tornadoServiceFee } from '../config';
 import { version } from '../../package.json';
 import { configService, getJobService, getPriceService } from '../services';
-import { JobType } from '../types';
+import { RelayerJobType } from '../types';
 
-const priceService = getPriceService();
-const jobService = getJobService();
 
 export function mainHandler(server: FastifyInstance, options, next) {
+  const jobService = getJobService();
+  const priceService = getPriceService();
+
   server.get('/',
     async (req, res) => {
       res.type('text/html')
@@ -42,6 +43,7 @@ export function mainHandler(server: FastifyInstance, options, next) {
 }
 
 export function relayerHandler(server: FastifyInstance, options, next) {
+  const jobService = getJobService();
   server.get<{ Params: { id: string } }>('/jobs/:id',
     { schema: jobsSchema },
     async (req, res) => {
@@ -54,7 +56,7 @@ export function relayerHandler(server: FastifyInstance, options, next) {
     { schema: withdrawSchema },
     async (req, res) => {
       console.log(req.body);
-      const id = await jobService.postJob(JobType.TORNADO_WITHDRAW, req.body);
+      const id = await jobService.postJob(RelayerJobType.TORNADO_WITHDRAW, req.body);
       res.send({ id });
     });
   next();

src/config.ts

@@ -1,31 +1,28 @@
-import { JobType } from './types';
+import { RelayerJobType } from './types';
 import tornConfig, { availableIds } from 'torn-token';
 
 require('dotenv').config();
 
 export const netId = <availableIds>Number(process.env.NET_ID || 1);
 export const redisUrl = process.env.REDIS_URL || 'redis://127.0.0.1:6379';
-export const httpRpcUrl = process.env.HTTP_RPC_URL;
-export const wsRpcUrl = process.env.WS_RPC_URL;
+export const rpcUrl = process.env.RPC_URL;
 export const oracleRpcUrl = process.env.ORACLE_RPC_URL || 'https://mainnet.infura.io/';
 export const offchainOracleAddress = '0x07D91f5fb9Bf7798734C3f606dB065549F6893bb';
 export const multiCallAddress = '0xda3c19c6fe954576707fa24695efb830d9cca1ca';
 export const aggregatorAddress = process.env.AGGREGATOR;
-export const minerMerkleTreeHeight = 20;
 export const privateKey = process.env.PRIVATE_KEY;
 export const instances = tornConfig.instances;
 export const torn = tornConfig;
 export const port = process.env.APP_PORT || 8000;
 export const tornadoServiceFee = Number(process.env.REGULAR_TORNADO_WITHDRAW_FEE);
-export const miningServiceFee = Number(process.env.MINING_SERVICE_FEE);
 export const rewardAccount = process.env.REWARD_ACCOUNT;
 export const governanceAddress = '0x5efda50f22d34F262c29268506C5Fa42cB56A1Ce';
 export const tornadoGoerliProxy = '0x454d870a72e29d5E5697f635128D18077BD04C60';
 export const gasLimits = {
-  [JobType.TORNADO_WITHDRAW]: 390000,
+  [RelayerJobType.TORNADO_WITHDRAW]: 390000,
   WITHDRAW_WITH_EXTRA: 700000,
-  [JobType.MINING_REWARD]: 455000,
-  [JobType.MINING_WITHDRAW]: 400000,
+  [RelayerJobType.MINING_REWARD]: 455000,
+  [RelayerJobType.MINING_WITHDRAW]: 400000,
 };
 export const minimumBalance = '1000000000000000000';
 export const baseFeeReserve = Number(process.env.BASE_FEE_RESERVE_PERCENTAGE);

src/modules/contracts.ts

@@ -5,11 +5,12 @@ import {
   TornadoProxyABI__factory,
 } from '../../contracts';
 import { providers } from 'ethers';
-import { httpRpcUrl, multiCallAddress, netId, offchainOracleAddress, oracleRpcUrl } from '../config';
+import { rpcUrl, multiCallAddress, netId, offchainOracleAddress, oracleRpcUrl } from '../config';
+
+export function getProvider(isStatic = true, customRpcUrl?: string) {
+  if (isStatic) return new providers.StaticJsonRpcProvider(customRpcUrl || rpcUrl, netId);
+  else return new providers.JsonRpcProvider(customRpcUrl || rpcUrl, netId);
 
-export function getProvider(isStatic = true, rpcUrl?: string) {
-  if (isStatic) return new providers.StaticJsonRpcProvider(rpcUrl || httpRpcUrl, netId);
-  else return new providers.JsonRpcProvider(rpcUrl || httpRpcUrl, netId);
 }
 
 export const getTornadoProxyContract = (proxyAddress: string) => {

src/modules/index.ts

@@ -1,2 +1,2 @@
-export { default as redis } from './redis';
+export { default as redisStore } from './redis';
 export { resolve } from './ensResolver';

src/modules/redis.ts

@@ -1,12 +1,25 @@
 import Redis from 'ioredis';
 import { redisUrl } from '../config';
+import { container, singleton } from 'tsyringe';
 
-const redisClient = new Redis(redisUrl, { maxRetriesPerRequest: null });
-const redisSubscriber = new Redis(redisUrl, { maxRetriesPerRequest: null });
+@singleton()
+export class RedisStore {
+  get client(): Redis.Redis {
+    return this._client;
+  }
 
-export const getClient = () => redisClient.on('error', (error) => {
-  throw error;
-});
-export const getSubscriber = () => redisSubscriber;
+  get subscriber(): Redis.Redis {
+    return this._subscriber;
+  }
 
-export default { getClient, getSubscriber };
+  private readonly _subscriber: Redis.Redis;
+  private readonly _client: Redis.Redis;
+
+  constructor() {
+    this._client = new Redis(redisUrl, { maxRetriesPerRequest: null });
+    this._subscriber = new Redis(redisUrl, { maxRetriesPerRequest: null });
+    console.log('RedisStore new instance');
+  }
+}
+
+export default () => container.resolve(RedisStore);

src/queue/index.ts

@@ -1,48 +1,105 @@
-import { Job, Processor, Queue, Worker } from 'bullmq';
-import { redis } from '../modules';
-import { JobStatus, JobType, Token } from '../types';
-import { relayerProcessor } from './relayerProcessor';
-import { WithdrawalData } from '../services/TxService';
-import { schedulerProcessor } from './schedulerProcessor';
-import { configService } from '../services';
+import { Processor, Queue, QueueScheduler, Worker } from 'bullmq';
+import { JobStatus, RelayerJobType, Token } from '../types';
+import { WithdrawalData } from '../services/tx.service';
 import { BigNumber } from 'ethers';
+import { priceProcessor } from './price.processor';
+import { autoInjectable } from 'tsyringe';
+import { RedisStore } from '../modules/redis';
+import { ConfigService } from '../services/config.service';
+import { relayerProcessor } from './relayer.processor';
 
-const connection = redis.getClient();
+type PriceJobData = Token[]
+type PriceJobReturn = number
+type HealthJobReturn = { balance: BigNumber, isEnought: boolean }
 
-export type SchedulerJobProcessors = {
-  updatePrices: Processor,
-  checkBalance: Processor
+type RelayerJobData = WithdrawalData & { id: string, status: JobStatus, type: RelayerJobType }
+
+type RelayerJobReturn = any
+
+export type RelayerProcessor = Processor<RelayerJobData, RelayerJobReturn, RelayerJobType>
+export type PriceProcessor = Processor<PriceJobData, PriceJobReturn, 'updatePrice'>
+
+@autoInjectable()
+export class PriceQueueHelper {
+  _queue: Queue<PriceJobData, PriceJobReturn, 'updatePrice'>;
+  _worker: Worker<PriceJobData, PriceJobReturn, 'updatePrice'>;
+  _scheduler: QueueScheduler;
+
+  constructor(private store?: RedisStore) {
+  }
+
+  get queue() {
+    if (!this._queue) {
+      this._queue = new Queue<Token[], PriceJobReturn, 'updatePrice'>('price', {
+        connection: this.store.client,
+        defaultJobOptions: {
+          removeOnFail: 10,
+          removeOnComplete: 10,
+        },
+      });
+    }
+    return this._queue;
+  }
+
+  get worker() {
+    if (!this._worker) {
+      this._worker = new Worker<PriceJobData, PriceJobReturn, 'updatePrice'>('price', priceProcessor, {
+        connection: this.store.client,
+        concurrency: 1,
+      });
+    }
+    return this._worker;
+  }
+
+  get scheduler() {
+    if (!this._scheduler) {
+      this._scheduler = new QueueScheduler('price', { connection: this.store.client });
+    }
+    return this._scheduler;
+  }
+
+  async addRepeatable(tokens: PriceJobData) {
+    await this.queue.add('updatePrice', tokens, {
+      repeat: {
+        every: 30000,
+        immediately: true,
+      },
+    });
+  }
 }
 
-type SchedulerJobName = keyof SchedulerJobProcessors
-type SchedulerJobData = Token[] | null
-type SchedulerJobReturn = Record<string, string> | { balance: BigNumber, isEnought: boolean }
-type RelayerJobData = WithdrawalData & { id: string, status: JobStatus, type: JobType }
-type RelayerJobReturn = void
 
-// export interface SchedulerProcessor {
-//   <U extends SchedulerJobName>(job: Job<SchedulerJobData, SchedulerJobReturn, U>): SchedulerJobProcessors[U];
-//
-// }
+@autoInjectable()
+export class RelayerQueueHelper {
+  private _queue: Queue<RelayerJobData, RelayerJobReturn, RelayerJobType>;
+  private _worker: Worker<RelayerJobData, RelayerJobReturn, RelayerJobType>;
+  private _scheduler: QueueScheduler;
+
+  constructor(private store?: RedisStore, private config?: ConfigService) {
+  }
+
+  get queue() {
+    if (!this._queue) {
+      this._queue = new Queue<RelayerJobData, RelayerJobReturn, RelayerJobType>(this.config.queueName, { connection: this.store.client });
+    }
+    return this._queue;
+  }
+
+  get worker() {
+    if (!this._worker) {
+      this._worker = new Worker<RelayerJobData, RelayerJobReturn, RelayerJobType>(this.config.queueName, relayerProcessor, { connection: this.store.client });
+    }
+    return this._worker;
+  }
+
+  get scheduler() {
+    if (!this._scheduler) {
+      this._scheduler = new QueueScheduler(this.config.queueName, { connection: this.store.client });
+    }
+    return this._scheduler;
+  }
 
 
-export interface RelayerProcessor {
-  (job: Job<RelayerJobData, RelayerJobReturn, JobType>): Promise<void>;
 }
 
-export const schedulerQueue = new Queue<Token[], any, SchedulerJobName>('scheduler', {
-  connection,
-  defaultJobOptions: {
-    removeOnFail: 10,
-    removeOnComplete: 10,
-  },
-});
-export const getSchedulerWorker = () => new Worker<SchedulerJobData, SchedulerJobReturn, SchedulerJobName>(schedulerQueue.name, (job) => schedulerProcessor(job), {
-  connection,
-  concurrency: 3,
-});
-
-export const relayerQueue = new Queue<RelayerJobData, RelayerJobReturn, JobType>(configService.queueName, { connection });
-export const getRelayerWorker = () => new Worker<RelayerJobData, RelayerJobReturn, JobType>(relayerQueue.name, relayerProcessor, { connection });
-
 

src/queue/price.processor.ts

@@ -0,0 +1,9 @@
+import { getPriceService } from '../services';
+import { PriceProcessor } from './index';
+
+export const priceProcessor: PriceProcessor = async (job) => {
+  const priceService = getPriceService();
+  const result = await priceService.fetchPrices(job.data);
+  console.log('priceProcessor', result);
+  return await priceService.savePrices(result);
+};

src/queue/relayer.processor.ts

@@ -0,0 +1,16 @@
+import { RelayerProcessor } from './index';
+import { getTxService } from '../services';
+import { JobStatus } from '../types';
+
+export const relayerProcessor: RelayerProcessor = async (job) => {
+  await job.update({ ...job.data, status: JobStatus.ACCEPTED });
+  console.log(`Start processing a new ${job.data.type} job ${job.id}`);
+
+  const txService = getTxService();
+  const withdrawalData = job.data;
+  await txService.checkTornadoFee(withdrawalData);
+  const txData = await txService.prepareTxData(withdrawalData);
+  const receipt = await txService.sendTx(txData);
+  console.log(receipt);
+  return receipt;
+};

src/queue/relayerProcessor.ts

@@ -1,6 +0,0 @@
-import { RelayerProcessor } from './index';
-
-export const relayerProcessor: RelayerProcessor = async (job) => {
-
-  console.log(job.data);
-};

src/queue/scheduler.processor.ts

@@ -0,0 +1,6 @@
+import { configService } from '../services';
+import { Processor } from 'bullmq';
+
+export const checkBalance: Processor = async (job) => {
+  return await configService.getBalance();
+};

src/queue/schedulerProcessor.ts

@@ -1,15 +0,0 @@
-import { configService, getPriceService } from '../services';
-import { Processor } from 'bullmq';
-
-export const schedulerProcessor: Processor = async (job) => {
-  switch (job.name) {
-  case 'updatePrices': {
-    const result = await getPriceService().fetchPrices(job.data);
-    return result;
-  }
-  case 'checkBalance': {
-    console.log(job.data);
-    return await configService.getBalance();
-  }
-  }
-};

src/queue/worker.ts

@@ -1,23 +1,24 @@
-import { getRelayerWorker, getSchedulerWorker } from './';
-import { configService, getPriceService } from '../services';
+import 'reflect-metadata';
+import { PriceQueueHelper, RelayerQueueHelper } from './';
+import { configService } from '../services';
 
 
 export const schedulerWorker = async () => {
   await configService.init();
-  const priceService = getPriceService();
-  const schedulerWorkerWorker = getSchedulerWorker();
-  console.log('price worker');
-  schedulerWorkerWorker.on('active', () => console.log('worker active'));
-  schedulerWorkerWorker.on('completed', async (job, result) => {
-    if (job.name === 'updatePrices') {
-      // await priceService.savePrices(result);
-    }
+  const price = new PriceQueueHelper();
+  console.log('price worker', price.queue.name);
+  price.worker.on('active', () => console.log('worker active'));
+  price.worker.on('completed', async (job, result) => {
+    console.log(`Job ${job.id} completed with result: ${result}`);
   });
-  schedulerWorkerWorker.on('failed', (job, error) => console.log(error));
+  price.worker.on('failed', (job, error) => console.log(error));
 };
 
 export const relayerWorker = async () => {
-  const relayerWorker = getRelayerWorker();
-  relayerWorker.on('completed', (job, result) => console.log(result));
-  relayerWorker.on('failed', (job, error) => console.log(error));
+  await configService.init();
+  const relayer = new RelayerQueueHelper();
+  relayer.worker.on('completed', (job, result) => {
+    console.log(`Job ${job.id} completed with result: ${result}`);
+  });
+  relayer.worker.on('failed', (job, error) => console.log(error));
 };

src/services/JobService.ts

@@ -1,70 +0,0 @@
-import { v4 } from 'uuid';
-import { JobStatus, JobType } from '../types';
-import { relayerQueue, schedulerQueue } from '../queue';
-import { WithdrawalData } from './TxService';
-import { getClient } from '../modules/redis';
-import { Job } from 'bullmq';
-import { configService } from './index';
-
-export class JobService {
-  store: ReturnType<typeof getClient>;
-
-  constructor() {
-    this.store = getClient();
-  }
-
-  async postJob(type: JobType, data: WithdrawalData) {
-    const id = v4();
-
-    const job = await relayerQueue.add(
-      type,
-      {
-        id,
-        type,
-        status: JobStatus.QUEUED,
-        ...data,
-      },
-      {},
-    );
-    this.save(job);
-    return id;
-  }
-
-  save(job: Job) {
-    return this.store.set(`job:${job.data.id}`, job.id);
-  }
-
-  async getJob(id: string) {
-    const key = 'job:' + id;
-    console.log(key);
-    const jobId = await this.store.get(key);
-    return await relayerQueue.getJob(jobId);
-  }
-
-  async getQueueCount() {
-    return await relayerQueue.getJobCountByTypes('active', 'waiting', 'delayed');
-  }
-
-  private async _clearSchedulerJobs() {
-    const jobs = await schedulerQueue.getJobs();
-    await Promise.all(jobs.map(job => schedulerQueue.remove(job.id)));
-  }
-
-  async setupRepeatableJobs() {
-    await this._clearSchedulerJobs();
-    await schedulerQueue.add('updatePrices', configService.tokens, {
-      repeat: {
-        every: 30000,
-        immediately: true,
-      },
-    });
-    await schedulerQueue.add('checkBalance', null, {
-      repeat: {
-        every: 30000,
-        immediately: true,
-      },
-    });
-  }
-}
-
-export default () => new JobService();

src/services/config.service.ts

@@ -1,13 +1,4 @@
-import {
-  httpRpcUrl,
-  instances,
-  minimumBalance,
-  netId,
-  privateKey,
-  torn,
-  tornadoGoerliProxy,
-  tornToken,
-} from '../config';
+import { instances, minimumBalance, netId, privateKey, rpcUrl, torn, tornadoGoerliProxy, tornToken } from '../config';
 import { Token } from '../types';
 import { getProvider, getTornadoProxyContract, getTornadoProxyLightContract } from '../modules/contracts';
 import { resolve } from '../modules';
@@ -15,35 +6,36 @@ import { ProxyLightABI, TornadoProxyABI } from '../../contracts';
 import { availableIds, netIds, NetInstances } from '../../../torn-token';
 import { getAddress } from 'ethers/lib/utils';
 import { providers, Wallet } from 'ethers';
+import { container, singleton } from 'tsyringe';
 
 type relayerQueueName = `relayer_${availableIds}`
 
+@singleton()
 export class ConfigService {
   static instance: ConfigService;
-  netId: availableIds;
   netIdKey: netIds;
   queueName: relayerQueueName;
   tokens: Token[];
-  privateKey: string;
-  rpcUrl: string;
   private _proxyAddress: string;
   private _proxyContract: TornadoProxyABI | ProxyLightABI;
   addressMap = new Map<string, InstanceProps>();
   isLightMode: boolean;
   instances: NetInstances;
-  provider: providers.StaticJsonRpcProvider;
+  provider: providers.JsonRpcProvider;
   wallet: Wallet;
+  public readonly netId: availableIds = netId;
+  public readonly privateKey = privateKey;
+  public readonly rpcUrl = rpcUrl;
+  isInit: boolean;
 
   constructor() {
-    this.netId = netId;
     this.netIdKey = `netId${this.netId}`;
     this.queueName = `relayer_${this.netId}`;
     this.isLightMode = ![1, 5].includes(netId);
-    this.privateKey = privateKey;
-    this.rpcUrl = httpRpcUrl;
     this.instances = instances[this.netIdKey];
     this.provider = getProvider(false);
     this.wallet = new Wallet(this.privateKey, this.provider);
+    console.log(this.wallet.address);
     this._fillInstanceMap();
   }
 
@@ -57,6 +49,7 @@ export class ConfigService {
 
   private _fillInstanceMap() {
     if (!this.instances) throw new Error('config mismatch, check your environment variables');
+    // TODO
     for (const [currency, { instanceAddress, symbol, decimals }] of Object.entries(this.instances)) {
       Object.entries(instanceAddress).forEach(([amount, address]) => {
           if (address) {
@@ -76,12 +69,13 @@ export class ConfigService {
     try {
       await this.provider.getNetwork();
     } catch (e) {
-      throw new Error(`Could not detect network, check your rpc url: ${this.rpcUrl}`);
+      throw new Error(`Could not detect network, check your rpc url: ${this.rpcUrl}. ` + e.message);
     }
   }
 
   async init() {
     try {
+      if (this.isInit) return;
       await this._checkNetwork();
       if (this.isLightMode) {
         this._proxyAddress = await resolve(torn.tornadoProxyLight.address);
@@ -92,14 +86,18 @@ export class ConfigService {
           this._proxyAddress = await resolve(torn.tornadoRouter.address);
         }
         this._proxyContract = getTornadoProxyContract(this._proxyAddress);
-        this.tokens = [tornToken, ...Object.values(torn.instances['netId1'])]
-          .map<Token>(el => (el.tokenAddress && {
-            address: getAddress(el.tokenAddress),
-            ...el,
-          })).filter(Boolean);
-        console.log(
-          `Configuration completed\n-- netId: ${this.netId}\n-- rpcUrl: ${this.rpcUrl}`);
       }
+      // TODO get instances from registry
+
+      this.tokens = [tornToken, ...Object.values(torn.instances['netId1'])]
+        .map<Token>(el => (el.tokenAddress && {
+          address: getAddress(el.tokenAddress),
+          decimals: el.decimals,
+          symbol: el.symbol,
+        })).filter(Boolean);
+      console.log(
+        `Configuration completed\n-- netId: ${this.netId}\n-- rpcUrl: ${this.rpcUrl}`);
+      this.isInit = true;
     } catch (e) {
       console.error(`${this.constructor.name} Error:`, e.message);
     }
@@ -115,12 +113,6 @@ export class ConfigService {
     return { balance, isEnougth };
   }
 
-  public static getServiceInstance() {
-    if (!ConfigService.instance) {
-      ConfigService.instance = new ConfigService();
-    }
-    return ConfigService.instance;
-  }
 }
 
 type InstanceProps = {
@@ -130,4 +122,4 @@ type InstanceProps = {
   decimals: number,
 }
 
-export default ConfigService.getServiceInstance();
+export default container.resolve(ConfigService);

src/services/index.ts

@@ -1,4 +1,4 @@
-export { default as configService } from './ConfigService';
-export { default as getPriceService } from './PriceService';
-export { default as getJobService } from './JobService';
-export { default as txService } from './TxService';
+export { default as configService } from './config.service';
+export { default as getPriceService } from './price.service';
+export { default as getJobService } from './job.service';
+export { default as getTxService } from './tx.service';

src/services/job.service.ts

@@ -0,0 +1,54 @@
+import { v4 } from 'uuid';
+import { JobStatus, RelayerJobType } from '../types';
+import { PriceQueueHelper, RelayerQueueHelper } from '../queue';
+import { WithdrawalData } from './tx.service';
+import { container, injectable } from 'tsyringe';
+import { ConfigService } from './config.service';
+
+@injectable()
+export class JobService {
+  constructor(private price?: PriceQueueHelper, private relayer?: RelayerQueueHelper, public config?: ConfigService) {
+  }
+
+  async postJob(type: RelayerJobType, data: WithdrawalData) {
+    const id = v4();
+
+    const job = await this.relayer.queue.add(
+      type,
+      {
+        id,
+        type,
+        status: JobStatus.QUEUED,
+        ...data,
+      },
+      { jobId: id },
+    );
+    return job.id;
+  }
+
+  async getJob(jobId: string) {
+    return await this.relayer.queue.getJob(jobId);
+  }
+
+  async getQueueCount() {
+    return await this.relayer.queue.getJobCountByTypes('active', 'waiting', 'delayed');
+  }
+
+  private async _clearSchedulerJobs() {
+    const jobs = await this.price.queue.getJobs();
+    await Promise.all(jobs.map(job => job.remove()));
+  }
+
+  async setupRepeatableJobs() {
+    await this._clearSchedulerJobs();
+    await this.price.addRepeatable(this.config.tokens);
+    // await this.schedulerQ.add('checkBalance', null, {
+    //   repeat: {
+    //     every: 30000,
+    //     immediately: true,
+    //   },
+    // });
+  }
+}
+
+export default () => container.resolve(JobService);

src/services/price.service.ts

@@ -4,15 +4,15 @@ import { MultiCall } from '../../contracts/MulticallAbi';
 import { BigNumber } from 'ethers';
 import { defaultAbiCoder } from 'ethers/lib/utils';
 import { Token } from '../types';
-import { redis } from '../modules';
-
-const redisClient = redis.getClient();
+import { container, injectable } from 'tsyringe';
+import { RedisStore } from '../modules/redis';
 
+@injectable()
 export class PriceService {
-  oracle: OffchainOracleAbi;
-  multiCall: MulticallAbi;
+  private oracle: OffchainOracleAbi;
+  private multiCall: MulticallAbi;
 
-  constructor() {
+  constructor(private store: RedisStore) {
     this.oracle = getOffchainOracleContract();
     this.multiCall = getMultiCallContract();
   }
@@ -28,12 +28,12 @@ export class PriceService {
 
   async fetchPrices(tokens: Token[]) {
     const names = tokens.reduce((p, c) => {
-      p[c.address] = c.symbol;
+      p[c.address] = c.symbol.toLowerCase();
       return p;
     }, {});
     const callData = this.prepareCallData(tokens);
     const { results, success } = await this.multiCall.multicall(callData);
-    const prices: { [p: string]: string } = {};
+    const prices: Record<string, string> = {};
     for (let i = 0; i < results.length; i++) {
       if (!success[i]) {
         continue;
@@ -48,16 +48,16 @@ export class PriceService {
   }
 
   async getPrice(currency: string) {
-    return await redisClient.hget('prices', currency);
+    return await this.store.client.hget('prices', currency);
   }
 
   async getPrices() {
-    return await redisClient.hgetall('prices');
+    return await this.store.client.hgetall('prices');
   }
 
   async savePrices(prices: Record<string, string>) {
-    await redisClient.hset('prices', prices);
+    return await this.store.client.hset('prices', prices);
   }
 }
 
-export default () => new PriceService();
+export default () => container.resolve(PriceService);

src/services/tx.service.ts

@@ -1,12 +1,13 @@
-import { TxManager } from 'tx-manager';
-import { configService } from './index';
-import { ProxyLightABI, TornadoProxyABI } from '../../contracts';
-import { formatEther, parseEther, parseUnits } from 'ethers/lib/utils';
-import { gasLimits, httpRpcUrl, tornadoServiceFee } from '../config';
-import { BigNumber, BigNumberish, BytesLike } from 'ethers';
-import { JobType } from '../types';
-import getPriceService from './PriceService';
+import { TransactionData, TxManager } from 'tx-manager';
 import { GasPriceOracle } from 'gas-price-oracle';
+import { Provider } from '@ethersproject/providers';
+import { formatEther, parseUnits } from 'ethers/lib/utils';
+import { BigNumber, BigNumberish, BytesLike } from 'ethers';
+import { configService, getPriceService } from './index';
+import { ProxyLightABI, TornadoProxyABI } from '../../contracts';
+import { gasLimits, tornadoServiceFee } from '../config';
+import { RelayerJobType } from '../types';
+import { PriceService } from './price.service';
 
 export type WithdrawalData = {
   contract: string,
@@ -24,37 +25,33 @@ export type WithdrawalData = {
 export class TxService {
   txManager: TxManager;
   tornadoProxy: TornadoProxyABI | ProxyLightABI;
-  priceService: ReturnType<typeof getPriceService>;
   oracle: GasPriceOracle;
+  provider: Provider;
+  priceService: PriceService;
 
   constructor() {
     const { privateKey, rpcUrl, netId } = configService;
     this.txManager = new TxManager({ privateKey, rpcUrl });
     this.tornadoProxy = configService.proxyContract;
-    this.oracle = new GasPriceOracle({ defaultRpc: httpRpcUrl, chainId: netId });
+    this.provider = this.tornadoProxy.provider;
+    this.oracle = new GasPriceOracle({ defaultRpc: rpcUrl, chainId: netId });
     this.priceService = getPriceService();
   }
 
-  async init() {
-    const currentTx = this.txManager.createTx({
-      nonce: 123,
-      to: '0x2f04c418e91585222a7042FFF4aB7281D34FdfCC',
-      value: parseEther('1'),
-    });
+  async sendTx(tx: TransactionData) {
+    const currentTx = this.txManager.createTx(tx);
 
-    const receipt = await currentTx.send()
+    return await currentTx.send()
       .on('transactionHash', txHash => console.log({ txHash }))
       .on('mined', receipt => console.log('Mined in block', receipt.blockNumber))
       .on('confirmations', confirmations => console.log({ confirmations }));
-
-    return receipt;
   }
 
-  private async prepareCallData(data: WithdrawalData) {
+  async prepareTxData(data: WithdrawalData): Promise<TransactionData> {
     const { contract, proof, args } = data;
     const calldata = this.tornadoProxy.interface.encodeFunctionData('withdraw', [contract, proof, ...args]);
     return {
-      value: data.args[5],
+      value: args[5],
       to: this.tornadoProxy.address,
       data: calldata,
       gasLimit: gasLimits['WITHDRAW_WITH_EXTRA'],
@@ -62,21 +59,26 @@ export class TxService {
   }
 
   async checkTornadoFee({ args, contract }: WithdrawalData) {
-    const { currency, amount, decimals } = configService.getInstance(contract);
+    const instance = configService.getInstance(contract);
+    if (!instance) throw new Error('Instance not found');
+    const { currency, amount, decimals } = instance;
     const [fee, refund] = [args[4], args[5]].map(BigNumber.from);
     const gasPrice = await this.getGasPrice();
-    const ethPrice = await this.priceService.getPrice(currency);
-    const operationCost = gasPrice.mul((gasLimits[JobType.TORNADO_WITHDRAW]));
+    // TODO check refund value
+    const operationCost = gasPrice.mul((gasLimits[RelayerJobType.TORNADO_WITHDRAW]));
 
     const serviceFee = parseUnits(amount, decimals)
-      .mul(tornadoServiceFee * 1e10)
-      .div(100 * 1e10);
+      .mul(`${tornadoServiceFee * 1e10}`)
+      .div(`${100 * 1e10}`);
 
     let desiredFee = operationCost.add(serviceFee);
-    if (currency !== 'eth') {
+
+    if (!configService.isLightMode && currency !== 'eth') {
+      const ethPrice = await this.priceService.getPrice(currency);
+      const numerator = BigNumber.from(10).pow(decimals);
       desiredFee = operationCost
         .add(refund)
-        .mul(10 ** decimals)
+        .mul(numerator)
         .div(ethPrice)
         .add(serviceFee);
     }
@@ -90,17 +92,15 @@ export class TxService {
     if (fee.lt(desiredFee)) {
       throw new Error('Provided fee is not enough. Probably it is a Gas Price spike, try to resubmit.');
     }
-
   }
 
   async getGasPrice(): Promise<BigNumber> {
-    const { baseFeePerGas = 0 } = await this.tornadoProxy.provider.getBlock('latest');
-    // const gasPrice = await this.tornadoProxy.provider.getGasPrice();
-    if (baseFeePerGas) return baseFeePerGas;
+    // TODO eip https://eips.ethereum.org/EIPS/eip-1559
+    const { baseFeePerGas = 0 } = await this.provider.getBlock('latest');
+    if (baseFeePerGas) return await this.provider.getGasPrice();
     const { fast = 0 } = await this.oracle.gasPrices();
     return parseUnits(String(fast), 'gwei');
   }
-
 }
 
 export default () => new TxService();

src/types.ts

@@ -1,4 +1,4 @@
-export enum JobType {
+export enum RelayerJobType {
   TORNADO_WITHDRAW = 'TORNADO_WITHDRAW',
   MINING_REWARD = 'MINING_REWARD',
   MINING_WITHDRAW = 'MINING_WITHDRAW',

src/worker.ts

@@ -1,3 +1,3 @@
-import { relayerWorker, schedulerWorker } from './queue/worker';
+import { schedulerWorker } from './queue/worker';
 
 schedulerWorker();

yarn.lock

@@ -4411,6 +4411,11 @@ reduce-flatten@^2.0.0:
   resolved "https://registry.yarnpkg.com/reduce-flatten/-/reduce-flatten-2.0.0.tgz#734fd84e65f375d7ca4465c69798c25c9d10ae27"
   integrity sha512-EJ4UNY/U1t2P/2k6oqotuX2Cc3T6nxJwsM0N0asT7dhrtH1ltUxDn4NalSYmPE2rCkVpcf/X6R0wDwcFpzhd4w==
 
+reflect-metadata@^0.1.13:
+  version "0.1.13"
+  resolved "https://registry.yarnpkg.com/reflect-metadata/-/reflect-metadata-0.1.13.tgz#67ae3ca57c972a2aa1642b10fe363fe32d49dc08"
+  integrity sha512-Ts1Y/anZELhSsjMcU605fU9RE4Oi3p5ORujwbIKXfWa+0Zxs510Qrmrce5/Jowq3cHSZSJqBjypxmHarc+vEWg==
+
 regexpp@^3.2.0:
   version "3.2.0"
   resolved "https://registry.yarnpkg.com/regexpp/-/regexpp-3.2.0.tgz#0425a2768d8f23bad70ca4b90461fa2f1213e1b2"
@@ -4979,7 +4984,7 @@ ts-toolbelt@^9.6.0:
   resolved "https://registry.yarnpkg.com/ts-toolbelt/-/ts-toolbelt-9.6.0.tgz#50a25426cfed500d4a09bd1b3afb6f28879edfd5"
   integrity sha512-nsZd8ZeNUzukXPlJmTBwUAuABDe/9qtVDelJeT/qW0ow3ZS3BsQJtNkan1802aM9Uf68/Y8ljw86Hu0h5IUW3w==
 
-tslib@^1.14.1, tslib@^1.8.1:
+tslib@^1.14.1, tslib@^1.8.1, tslib@^1.9.3:
   version "1.14.1"
   resolved "https://registry.yarnpkg.com/tslib/-/tslib-1.14.1.tgz#cf2d38bdc34a134bcaf1091c41f6619e2f672d00"
   integrity sha512-Xni35NKzjgMrwevysHTCArtLDpPvye8zV/0E4EyYn43P7/7qvQwPh9BGkHewbMulVntbigmcT7rdX3BNo9wRJg==
@@ -4996,6 +5001,13 @@ tsutils@^3.21.0:
   dependencies:
     tslib "^1.8.1"
 
+tsyringe@^4.6.0:
+  version "4.6.0"
+  resolved "https://registry.yarnpkg.com/tsyringe/-/tsyringe-4.6.0.tgz#14915d3d7f0db35e1cf7269bdbf7c440713c8d07"
+  integrity sha512-BMQAZamSfEmIQzH8WJeRu1yZGQbPSDuI9g+yEiKZFIcO46GPZuMOC2d0b52cVBdw1d++06JnDSIIZvEnogMdAw==
+  dependencies:
+    tslib "^1.9.3"
+
 tunnel-agent@^0.6.0:
   version "0.6.0"
   resolved "https://registry.yarnpkg.com/tunnel-agent/-/tunnel-agent-0.6.0.tgz#27a5dea06b36b04a0a9966774b290868f0fc40fd"