moved db close operation to start subscribe and fixed saveOrder

Author: rsercano

src/arby.ts

@@ -34,6 +34,8 @@ type StartArbyParams = {
     getCentralizedExchangeOrder$,
     getOpenDEXcomplete$,
     shutdown$,
+    models,
+    saveOrder$,
   }: GetTradeParams) => Observable<boolean>;
   cleanup$: ({
     config,
@@ -135,7 +137,6 @@ export const startArby = ({
                   loggers,
                   removeOpenDEXorders$,
                   removeCEXorders$,
-                  closeDB$,
                   CEX,
                 })
               ).pipe(
@@ -149,7 +150,6 @@ export const startArby = ({
                     removeOpenDEXorders$,
                     removeCEXorders$,
                     CEX,
-                    closeDB$,
                   });
                 })
               );
@@ -182,8 +182,18 @@ if (!module.parent) {
       } else {
         console.log(error);
       }
-      process.exit(1);
+      closeDB$().subscribe({
+        complete: () => {
+          process.exit(1);
+        },
+      });
+    },
+    complete: () => {
+      closeDB$().subscribe({
+        complete: () => {
+          console.log('Shutdown complete. Goodbye, Arby.');
+        },
+      });
     },
-    complete: () => console.log('Shutdown complete. Goodbye, Arby.'),
   });
 }

src/centralized/execute-order.ts

@@ -61,7 +61,7 @@ const executeCEXorder$ = ({
         logger.info(
           `Centralized exchange order finished: ${JSON.stringify(order)}`
         );
-        saveOrder$({ order, logger, models });
+        saveOrder$({ order, logger, models }).subscribe();
       }),
       catchError((e, caught) => {
         logger.warn(`Failed to execute CEX order: ${e}. Retrying in 1000ms`);

src/opendex/catch-error.ts

@@ -10,7 +10,6 @@ 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,
@@ -74,7 +73,6 @@ const catchOpenDEXerror = (
                   loggers,
                   removeOpenDEXorders$,
                   removeCEXorders$,
-                  closeDB$,
                   CEX,
                 }).pipe(ignoreElements()),
                 timer(RETRY_INTERVAL)

src/trade/cleanup.ts

@@ -47,7 +47,6 @@ type GetCleanupParams = {
       orderId: string
     ) => Observable<Order>
   ) => Observable<unknown>;
-  closeDB$: () => Observable<void>;
   CEX: Exchange;
 };
 
@@ -56,7 +55,6 @@ const getCleanup$ = ({
   loggers,
   removeOpenDEXorders$,
   removeCEXorders$,
-  closeDB$,
   CEX,
 }: GetCleanupParams): Observable<unknown> => {
   const retryOnError = (logger: Logger, source: Observable<any>) => {
@@ -79,11 +77,6 @@ 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$,