added graceful shutdown for db

Author: rsercano

src/arby.ts

@@ -21,7 +21,7 @@ import { getNewTrade$, GetTradeParams } from './trade/trade';
 import { getStartShutdown$ } from './utils';
 import { Dictionary, Market } from 'ccxt';
 import { verifyMarkets } from './centralized/verify-markets';
-import { initDB$, InitDBparams, InitDBResponse } from './db/db';
+import { initDB$, InitDBparams, InitDBResponse, closeDB$ } from './db/db';
 import { OrderRepository } from './db/order-repository';
 
 type StartArbyParams = {
@@ -135,6 +135,7 @@ export const startArby = ({
                   loggers,
                   removeOpenDEXorders$,
                   removeCEXorders$,
+                  closeDB$,
                   CEX,
                 })
               ).pipe(
@@ -148,6 +149,7 @@ export const startArby = ({
                     removeOpenDEXorders$,
                     removeCEXorders$,
                     CEX,
+                    closeDB$,
                   });
                 })
               );

src/db/db.ts

@@ -35,11 +35,13 @@ const createModels = (sequelize: Sequelize): InitDBResponse => {
   return models;
 };
 
+let sequelize: Sequelize;
+
 const initDB$ = ({
   logger,
   dataDir,
 }: InitDBparams): Observable<InitDBResponse> => {
-  const sequelize = new Sequelize({
+  sequelize = new Sequelize({
     storage: dataDir ? `${dataDir}/arby.db` : undefined,
     logging: logger.trace,
     dialect: 'sqlite',
@@ -60,4 +62,8 @@ const initDB$ = ({
   );
 };
 
-export { initDB$, InitDBparams, InitDBResponse };
+const closeDB$ = (): Observable<void> => {
+  return from(sequelize.close());
+};
+
+export { initDB$, InitDBparams, InitDBResponse, closeDB$ };

src/opendex/catch-error.ts

@@ -10,6 +10,7 @@ import { Logger, Loggers } from '../logger';
 import { errorCodes, errors } from '../opendex/errors';
 import { GetCleanupParams } from '../trade/cleanup';
 import { removeOpenDEXorders$ } from './remove-orders';
+import { closeDB$ } from '../db/db';
 
 const catchOpenDEXerror = (
   loggers: Loggers,
@@ -73,6 +74,7 @@ const catchOpenDEXerror = (
                   loggers,
                   removeOpenDEXorders$,
                   removeCEXorders$,
+                  closeDB$,
                   CEX,
                 }).pipe(ignoreElements()),
                 timer(RETRY_INTERVAL)

src/trade/cleanup.ts

@@ -47,6 +47,7 @@ type GetCleanupParams = {
       orderId: string
     ) => Observable<Order>
   ) => Observable<unknown>;
+  closeDB$: () => Observable<void>;
   CEX: Exchange;
 };
 
@@ -55,6 +56,7 @@ const getCleanup$ = ({
   loggers,
   removeOpenDEXorders$,
   removeCEXorders$,
+  closeDB$,
   CEX,
 }: GetCleanupParams): Observable<unknown> => {
   const retryOnError = (logger: Logger, source: Observable<any>) => {
@@ -77,6 +79,11 @@ const getCleanup$ = ({
   const retryOnErrorOpenDEX = curriedRetryOnError(loggers.opendex);
   const retryonErrorCEX = curriedRetryOnError(loggers.centralized);
   return combineLatest(
+    closeDB$().pipe(
+      tap(() => {
+        loggers.db.info('DB has been closed');
+      })
+    ),
     removeOpenDEXorders$({
       config,
       getXudClient$,