1 files changed, 17 insertions, 16 deletions
diff --git a/packages/pipeline/src/scripts/pull_radar_relay_orders.ts b/packages/pipeline/src/scripts/pull_radar_relay_orders.ts
index c4d3f7095..b3a4d887e 100644
--- a/packages/pipeline/src/scripts/pull_radar_relay_orders.ts
+++ b/packages/pipeline/src/scripts/pull_radar_relay_orders.ts
@@ -2,15 +2,14 @@
 import { HttpClient } from '@0x/connect';
 import * as R from 'ramda';
 import 'reflect-metadata';
-import { Connection, ConnectionOptions, createConnection } from 'typeorm';
+import { Connection, ConnectionOptions, createConnection, EntityManager } from 'typeorm';
 
-import { SraOrder } from '../entities';
+import { createObservedTimestampForOrder, SraOrder } from '../entities';
 import * as ormConfig from '../ormconfig';
 import { parseSraOrders } from '../parsers/sra_orders';
 import { handleError } from '../utils';
 
 const RADAR_RELAY_URL = 'https://api.radarrelay.com/0x/v2';
-const BATCH_SAVE_SIZE = 1000; // Number of orders to save at once.
 const ORDERS_PER_PAGE = 10000; // Number of orders to get per request.
 
 let connection: Connection;
@@ -29,24 +28,26 @@ async function getOrderbook(): Promise<void> {
     });
     console.log(`Got ${rawOrders.records.length} orders.`);
     console.log('Parsing orders...');
+    // Parse the sra orders, then add source url to each.
     const orders = R.pipe(parseSraOrders, R.map(setSourceUrl(RADAR_RELAY_URL)))(rawOrders);
-    const ordersRepository = connection.getRepository(SraOrder);
-    // TODO(albrow): Move batch saving to a utility function to reduce
-    // duplicated code.
-    for (const ordersBatch of R.splitEvery(BATCH_SAVE_SIZE, orders)) {
-        await ordersRepository.save(ordersBatch);
-    }
+    // Save all the orders and update the observed time stamps in a single
+    // transaction.
+    console.log('Saving orders and updating timestamps...');
+    await connection.transaction(async (manager: EntityManager): Promise<void> => {
+        for (const order of orders) {
+            await manager.save(SraOrder, order);
+            const observedTimestamp = createObservedTimestampForOrder(order);
+            await manager.save(observedTimestamp);
+        }
+    });
 }
 
 const sourceUrlProp = R.lensProp('sourceUrl');
 
+/**
+ * Sets the source url for a single order. Returns a new order instead of
+ * mutating the given one.
+ */
 const setSourceUrl = R.curry((sourceURL: string, order: SraOrder): SraOrder => {
     return R.set(sourceUrlProp, sourceURL, order);
 });
-
-const firstSeenProp = R.lensProp('firstSeenTimestamp');
-const lastUpdatedProp = R.lensProp('lastUpdatedTimestamp');
-
-const setFirstSeen = R.curry((sourceURL: string, order: SraOrder): SraOrder => {
-    return R.set(firstSeenTimestampProp, sourceURL, order);
-});