SCRAPE-7: Implement insertPrices() bulk operation (#12)

Co-authored-by: Stephen Minakian <stephenminakian@gmail.com>
Co-committed-by: Stephen Minakian <stephenminakian@gmail.com>

__tests__/scraper/insertPrices.test.js

@@ -0,0 +1,582 @@
+/**
+ * Tests for insertPrices() bulk operation
+ *
+ * Covers:
+ * - Only units with non-null prices are inserted
+ * - bulkWrite uses updateOne with upsert for idempotency
+ * - Filter uses unit_code + date_checked composite key
+ * - Price record includes unit_code, date_checked, price, last_updated, data_source
+ * - Empty price records logs warning and returns early
+ * - Error handling when bulkWrite fails
+ * - Logging of inserted and updated counts
+ * - Re-running on same day updates existing records (idempotent)
+ */
+
+const { MongoClient } = require('mongodb');
+const { MongoMemoryServer } = require('mongodb-memory-server');
+
+// Will require insertPrices after implementation
+let insertPrices;
+
+let mongoServer;
+let client;
+let db;
+
+const PRICES_COLLECTION = 'unit_prices_migration_test';
+
+// Mock logger for capturing log calls
+function createMockLogger() {
+  return {
+    info: jest.fn(),
+    warn: jest.fn(),
+    error: jest.fn()
+  };
+}
+
+beforeAll(async () => {
+  mongoServer = await MongoMemoryServer.create();
+  const uri = mongoServer.getUri();
+  client = new MongoClient(uri);
+  await client.connect();
+  db = client.db('test_apartments');
+
+  // Dynamically require to pick up implementation
+  ({ insertPrices } = require('../../services/scraperService'));
+});
+
+afterAll(async () => {
+  if (client) await client.close();
+  if (mongoServer) await mongoServer.stop();
+});
+
+beforeEach(async () => {
+  // Clean the prices collection before each test
+  const collections = await db.listCollections().toArray();
+  for (const col of collections) {
+    await db.collection(col.name).deleteMany({});
+  }
+});
+
+describe('insertPrices', () => {
+  const TODAY = '2026-02-05';
+
+  // ---------------------------------------------------------------
+  // 1. Only units with non-null prices are inserted
+  // ---------------------------------------------------------------
+  describe('filtering units with null prices', () => {
+    it('should only insert records for units with non-null prices', async () => {
+      const logger = createMockLogger();
+      const units = [
+        { unit_code: 'A101', price: 1200 },
+        { unit_code: 'A102', price: null },
+        { unit_code: 'A103', price: 1500 },
+        { unit_code: 'A104', price: null }
+      ];
+
+      await insertPrices(db, units, TODAY, logger);
+
+      const docs = await db.collection(PRICES_COLLECTION).find({}).toArray();
+      expect(docs).toHaveLength(2);
+
+      const unitCodes = docs.map(d => d.unit_code).sort();
+      expect(unitCodes).toEqual(['A101', 'A103']);
+    });
+
+    it('should skip units with price of null and not create any record for them', async () => {
+      const logger = createMockLogger();
+      const units = [
+        { unit_code: 'B101', price: null },
+        { unit_code: 'B102', price: null }
+      ];
+
+      const result = await insertPrices(db, units, TODAY, logger);
+
+      expect(result.insertedCount).toBe(0);
+
+      const docs = await db.collection(PRICES_COLLECTION).find({}).toArray();
+      expect(docs).toHaveLength(0);
+    });
+
+    it('should handle a mix of priced and null-priced units', async () => {
+      const logger = createMockLogger();
+      const units = [
+        { unit_code: 'C101', price: 950 },
+        { unit_code: 'C102', price: null },
+        { unit_code: 'C103', price: 1100 },
+        { unit_code: 'C104', price: null },
+        { unit_code: 'C105', price: 1300 }
+      ];
+
+      await insertPrices(db, units, TODAY, logger);
+
+      const count = await db.collection(PRICES_COLLECTION).countDocuments();
+      expect(count).toBe(3);
+    });
+  });
+
+  // ---------------------------------------------------------------
+  // 2. bulkWrite uses updateOne with upsert for idempotency
+  // ---------------------------------------------------------------
+  describe('bulkWrite with upsert operations', () => {
+    it('should create updateOne operations for each priced unit', async () => {
+      const logger = createMockLogger();
+      const units = [
+        { unit_code: 'D101', price: 1200 },
+        { unit_code: 'D102', price: 1500 }
+      ];
+
+      const result = await insertPrices(db, units, TODAY, logger);
+
+      // Both units should be upserted (new inserts)
+      expect(result.insertedCount).toBeGreaterThan(0);
+
+      const docs = await db.collection(PRICES_COLLECTION).find({}).toArray();
+      expect(docs).toHaveLength(2);
+    });
+
+    it('should use upsert so re-running does not create duplicates', async () => {
+      const logger = createMockLogger();
+      const units = [{ unit_code: 'D201', price: 1200 }];
+
+      // First run
+      await insertPrices(db, units, TODAY, logger);
+
+      // Second run on same day
+      await insertPrices(db, units, TODAY, logger);
+
+      // Should still have only 1 document (not 2)
+      const docs = await db.collection(PRICES_COLLECTION).find({}).toArray();
+      expect(docs).toHaveLength(1);
+    });
+  });
+
+  // ---------------------------------------------------------------
+  // 3. Filter uses unit_code + date_checked composite key
+  // ---------------------------------------------------------------
+  describe('composite key: unit_code + date_checked', () => {
+    it('should use unit_code and date_checked as the filter for upsert', async () => {
+      const logger = createMockLogger();
+      const units = [{ unit_code: 'E101', price: 1200 }];
+
+      await insertPrices(db, units, TODAY, logger);
+
+      const doc = await db.collection(PRICES_COLLECTION).findOne({
+        unit_code: 'E101',
+        date_checked: TODAY
+      });
+
+      expect(doc).not.toBeNull();
+      expect(doc.unit_code).toBe('E101');
+      expect(doc.date_checked).toBe(TODAY);
+    });
+
+    it('should create separate records for the same unit on different days', async () => {
+      const logger = createMockLogger();
+      const units = [{ unit_code: 'E201', price: 1200 }];
+
+      await insertPrices(db, units, '2026-02-04', logger);
+      await insertPrices(db, units, '2026-02-05', logger);
+
+      const docs = await db.collection(PRICES_COLLECTION)
+        .find({ unit_code: 'E201' })
+        .sort({ date_checked: 1 })
+        .toArray();
+
+      expect(docs).toHaveLength(2);
+      expect(docs[0].date_checked).toBe('2026-02-04');
+      expect(docs[1].date_checked).toBe('2026-02-05');
+    });
+
+    it('should create separate records for different units on the same day', async () => {
+      const logger = createMockLogger();
+      const units = [
+        { unit_code: 'E301', price: 1000 },
+        { unit_code: 'E302', price: 1100 }
+      ];
+
+      await insertPrices(db, units, TODAY, logger);
+
+      const docs = await db.collection(PRICES_COLLECTION)
+        .find({ date_checked: TODAY })
+        .toArray();
+
+      expect(docs).toHaveLength(2);
+    });
+  });
+
+  // ---------------------------------------------------------------
+  // 4. Price record includes required fields
+  // ---------------------------------------------------------------
+  describe('price record fields', () => {
+    it('should include unit_code in the price record', async () => {
+      const logger = createMockLogger();
+      const units = [{ unit_code: 'F101', price: 1400 }];
+
+      await insertPrices(db, units, TODAY, logger);
+
+      const doc = await db.collection(PRICES_COLLECTION).findOne({});
+      expect(doc.unit_code).toBe('F101');
+    });
+
+    it('should include date_checked in the price record', async () => {
+      const logger = createMockLogger();
+      const units = [{ unit_code: 'F201', price: 1400 }];
+
+      await insertPrices(db, units, TODAY, logger);
+
+      const doc = await db.collection(PRICES_COLLECTION).findOne({});
+      expect(doc.date_checked).toBe(TODAY);
+    });
+
+    it('should include price as a number in the price record', async () => {
+      const logger = createMockLogger();
+      const units = [{ unit_code: 'F301', price: 1550 }];
+
+      await insertPrices(db, units, TODAY, logger);
+
+      const doc = await db.collection(PRICES_COLLECTION).findOne({});
+      expect(doc.price).toBe(1550);
+      expect(typeof doc.price).toBe('number');
+    });
+
+    it('should include last_updated as an ISO timestamp string', async () => {
+      const logger = createMockLogger();
+      const beforeTime = new Date().toISOString();
+
+      const units = [{ unit_code: 'F401', price: 1600 }];
+      await insertPrices(db, units, TODAY, logger);
+
+      const afterTime = new Date().toISOString();
+
+      const doc = await db.collection(PRICES_COLLECTION).findOne({});
+      expect(doc.last_updated).toBeDefined();
+      expect(typeof doc.last_updated).toBe('string');
+      expect(doc.last_updated >= beforeTime).toBe(true);
+      expect(doc.last_updated <= afterTime).toBe(true);
+    });
+
+    it('should include data_source set to "web_scraper"', async () => {
+      const logger = createMockLogger();
+      const units = [{ unit_code: 'F501', price: 1700 }];
+
+      await insertPrices(db, units, TODAY, logger);
+
+      const doc = await db.collection(PRICES_COLLECTION).findOne({});
+      expect(doc.data_source).toBe('web_scraper');
+    });
+
+    it('should include all five required fields in every price record', async () => {
+      const logger = createMockLogger();
+      const units = [{ unit_code: 'F601', price: 1800 }];
+
+      await insertPrices(db, units, TODAY, logger);
+
+      const doc = await db.collection(PRICES_COLLECTION).findOne({});
+      expect(doc).toHaveProperty('unit_code');
+      expect(doc).toHaveProperty('date_checked');
+      expect(doc).toHaveProperty('price');
+      expect(doc).toHaveProperty('last_updated');
+      expect(doc).toHaveProperty('data_source');
+    });
+  });
+
+  // ---------------------------------------------------------------
+  // 5. Empty price records logs warning and returns early
+  // ---------------------------------------------------------------
+  describe('empty price records', () => {
+    it('should log a warning when no units have prices', async () => {
+      const logger = createMockLogger();
+      const units = [
+        { unit_code: 'G101', price: null },
+        { unit_code: 'G102', price: null }
+      ];
+
+      await insertPrices(db, units, TODAY, logger);
+
+      expect(logger.warn).toHaveBeenCalledWith('No price records to insert');
+    });
+
+    it('should return insertedCount of 0 for all null-priced units', async () => {
+      const logger = createMockLogger();
+      const units = [{ unit_code: 'G201', price: null }];
+
+      const result = await insertPrices(db, units, TODAY, logger);
+
+      expect(result.insertedCount).toBe(0);
+    });
+
+    it('should log a warning when units array is empty', async () => {
+      const logger = createMockLogger();
+
+      await insertPrices(db, [], TODAY, logger);
+
+      expect(logger.warn).toHaveBeenCalledWith('No price records to insert');
+    });
+
+    it('should not perform any database writes for empty/null-priced arrays', async () => {
+      const logger = createMockLogger();
+
+      await insertPrices(db, [], TODAY, logger);
+
+      const count = await db.collection(PRICES_COLLECTION).countDocuments();
+      expect(count).toBe(0);
+    });
+  });
+
+  // ---------------------------------------------------------------
+  // 6. Error handling when bulkWrite fails
+  // ---------------------------------------------------------------
+  describe('error handling', () => {
+    it('should throw error when bulkWrite fails', async () => {
+      const logger = createMockLogger();
+      const units = [{ unit_code: 'H101', price: 1000 }];
+
+      const mockCollection = {
+        bulkWrite: jest.fn().mockRejectedValue(new Error('Connection lost'))
+      };
+      const mockDb = {
+        collection: jest.fn().mockReturnValue(mockCollection)
+      };
+
+      await expect(insertPrices(mockDb, units, TODAY, logger)).rejects.toThrow('Connection lost');
+    });
+
+    it('should log error details when bulkWrite fails', async () => {
+      const logger = createMockLogger();
+      const units = [{ unit_code: 'H201', price: 1000 }];
+
+      const mockCollection = {
+        bulkWrite: jest.fn().mockRejectedValue(new Error('Write concern timeout'))
+      };
+      const mockDb = {
+        collection: jest.fn().mockReturnValue(mockCollection)
+      };
+
+      try {
+        await insertPrices(mockDb, units, TODAY, logger);
+      } catch (e) {
+        // Expected to throw
+      }
+
+      expect(logger.error).toHaveBeenCalledWith(
+        'Failed to insert prices',
+        expect.objectContaining({
+          errorType: 'Error',
+          errorMessage: 'Write concern timeout'
+        })
+      );
+    });
+
+    it('should re-throw the original error', async () => {
+      const logger = createMockLogger();
+      const units = [{ unit_code: 'H301', price: 1000 }];
+
+      const originalError = new TypeError('Invalid operation');
+      const mockCollection = {
+        bulkWrite: jest.fn().mockRejectedValue(originalError)
+      };
+      const mockDb = {
+        collection: jest.fn().mockReturnValue(mockCollection)
+      };
+
+      await expect(insertPrices(mockDb, units, TODAY, logger)).rejects.toBe(originalError);
+    });
+  });
+
+  // ---------------------------------------------------------------
+  // 7. Logging of inserted and updated counts
+  // ---------------------------------------------------------------
+  describe('logging of result counts', () => {
+    it('should log inserted and updated counts on success', async () => {
+      const logger = createMockLogger();
+      const units = [
+        { unit_code: 'I101', price: 1000 },
+        { unit_code: 'I102', price: 1100 }
+      ];
+
+      await insertPrices(db, units, TODAY, logger);
+
+      expect(logger.info).toHaveBeenCalledWith(
+        'Prices inserted',
+        expect.objectContaining({
+          inserted: expect.any(Number),
+          updated: expect.any(Number)
+        })
+      );
+    });
+
+    it('should log correct counts for new inserts', async () => {
+      const logger = createMockLogger();
+      const units = [
+        { unit_code: 'I201', price: 1000 },
+        { unit_code: 'I202', price: 1100 }
+      ];
+
+      await insertPrices(db, units, TODAY, logger);
+
+      // For new inserts: inserted=2, updated=0
+      expect(logger.info).toHaveBeenCalledWith(
+        'Prices inserted',
+        expect.objectContaining({
+          inserted: 2,
+          updated: 0
+        })
+      );
+    });
+
+    it('should log correct counts when updating existing records', async () => {
+      const logger = createMockLogger();
+      const units = [{ unit_code: 'I301', price: 1000 }];
+
+      // First insert
+      await insertPrices(db, units, TODAY, logger);
+
+      // Reset mock
+      logger.info.mockClear();
+
+      // Second run updates existing
+      const updatedUnits = [{ unit_code: 'I301', price: 1050 }];
+      await insertPrices(db, updatedUnits, TODAY, logger);
+
+      expect(logger.info).toHaveBeenCalledWith(
+        'Prices inserted',
+        expect.objectContaining({
+          inserted: 0,
+          updated: 1
+        })
+      );
+    });
+  });
+
+  // ---------------------------------------------------------------
+  // 8. Re-running on same day updates existing records (idempotent)
+  // ---------------------------------------------------------------
+  describe('idempotent re-runs on same day', () => {
+    it('should update the price when re-running on the same day', async () => {
+      const logger = createMockLogger();
+
+      // First run with original price
+      await insertPrices(db, [{ unit_code: 'J101', price: 1200 }], TODAY, logger);
+
+      const doc1 = await db.collection(PRICES_COLLECTION).findOne({
+        unit_code: 'J101',
+        date_checked: TODAY
+      });
+      expect(doc1.price).toBe(1200);
+
+      // Re-run with updated price (same day)
+      await insertPrices(db, [{ unit_code: 'J101', price: 1250 }], TODAY, logger);
+
+      const doc2 = await db.collection(PRICES_COLLECTION).findOne({
+        unit_code: 'J101',
+        date_checked: TODAY
+      });
+      expect(doc2.price).toBe(1250);
+
+      // Should still have only 1 record
+      const count = await db.collection(PRICES_COLLECTION).countDocuments({
+        unit_code: 'J101',
+        date_checked: TODAY
+      });
+      expect(count).toBe(1);
+    });
+
+    it('should update last_updated when re-running on the same day', async () => {
+      const logger = createMockLogger();
+
+      // First run
+      await insertPrices(db, [{ unit_code: 'J201', price: 1300 }], TODAY, logger);
+      const doc1 = await db.collection(PRICES_COLLECTION).findOne({ unit_code: 'J201' });
+      const firstUpdated = doc1.last_updated;
+
+      // Small delay to ensure different timestamp
+      await new Promise(resolve => setTimeout(resolve, 10));
+
+      // Re-run on same day
+      await insertPrices(db, [{ unit_code: 'J201', price: 1350 }], TODAY, logger);
+      const doc2 = await db.collection(PRICES_COLLECTION).findOne({ unit_code: 'J201' });
+      const secondUpdated = doc2.last_updated;
+
+      expect(secondUpdated > firstUpdated).toBe(true);
+    });
+
+    it('should return combined insert/update count reflecting the operation', async () => {
+      const logger = createMockLogger();
+      const units = [
+        { unit_code: 'J301', price: 1000 },
+        { unit_code: 'J302', price: 1100 }
+      ];
+
+      // First run - inserts
+      const result1 = await insertPrices(db, units, TODAY, logger);
+      expect(result1.insertedCount).toBe(2);
+
+      // Re-run - updates
+      const result2 = await insertPrices(db, units, TODAY, logger);
+      expect(result2.insertedCount).toBe(2); // combined upserted + modified
+
+      // Still only 2 total records
+      const count = await db.collection(PRICES_COLLECTION).countDocuments();
+      expect(count).toBe(2);
+    });
+
+    it('should handle a mix of new and existing records on re-run', async () => {
+      const logger = createMockLogger();
+
+      // First run with unit A
+      await insertPrices(db, [{ unit_code: 'J401', price: 1000 }], TODAY, logger);
+
+      // Re-run with unit A (existing) and unit B (new)
+      const mixedUnits = [
+        { unit_code: 'J401', price: 1050 },
+        { unit_code: 'J402', price: 1200 }
+      ];
+      await insertPrices(db, mixedUnits, TODAY, logger);
+
+      // Should have 2 total records
+      const docs = await db.collection(PRICES_COLLECTION).find({}).toArray();
+      expect(docs).toHaveLength(2);
+
+      // Verify updated price
+      const unitA = await db.collection(PRICES_COLLECTION).findOne({ unit_code: 'J401' });
+      expect(unitA.price).toBe(1050);
+    });
+  });
+
+  // ---------------------------------------------------------------
+  // 9. Uses ordered: false for performance
+  // ---------------------------------------------------------------
+  describe('ordered: false for performance', () => {
+    it('should successfully process all priced units in a batch', async () => {
+      const logger = createMockLogger();
+      const units = [
+        { unit_code: 'K101', price: 900 },
+        { unit_code: 'K102', price: 1000 },
+        { unit_code: 'K103', price: 1100 },
+        { unit_code: 'K104', price: 1200 },
+        { unit_code: 'K105', price: 1300 }
+      ];
+
+      await insertPrices(db, units, TODAY, logger);
+
+      const count = await db.collection(PRICES_COLLECTION).countDocuments();
+      expect(count).toBe(5);
+    });
+  });
+
+  // ---------------------------------------------------------------
+  // 10. Uses correct collection
+  // ---------------------------------------------------------------
+  describe('collection usage', () => {
+    it('should write to the configured PRICES collection', async () => {
+      const logger = createMockLogger();
+      const units = [{ unit_code: 'L101', price: 1400 }];
+
+      await insertPrices(db, units, TODAY, logger);
+
+      // Verify data is in the correct collection
+      const doc = await db.collection(PRICES_COLLECTION).findOne({ unit_code: 'L101' });
+      expect(doc).not.toBeNull();
+    });
+  });
+});

services/scraperService.js

@@ -430,11 +430,75 @@ async function upsertUnits(db, units, logger) {
   }
 }
 
+/**
+ * Insert price records for today using bulkWrite with upsert.
+ * Uses unit_code + date_checked as the composite key to prevent
+ * duplicate price records for the same unit on the same day.
+ * Re-running on the same day updates existing records (idempotent).
+ *
+ * @param {Db} db - MongoDB database instance
+ * @param {Array<Object>} units - Array of unit objects
+ * @param {string} date - Date in YYYY-MM-DD format
+ * @param {Object} logger - Logger instance
+ * @returns {Promise<Object>} Insert result with insertedCount
+ */
+async function insertPrices(db, units, date, logger) {
+  const collection = db.collection(config.COLLECTIONS.PRICES);
+  const now = new Date().toISOString();
+
+  // Filter out units without prices
+  const unitsWithPrices = units.filter(u => u.price !== null);
+
+  const priceRecords = unitsWithPrices.map(unit => ({
+    unit_code: unit.unit_code,
+    date_checked: date,
+    price: unit.price,
+    last_updated: now,
+    data_source: 'web_scraper'
+  }));
+
+  if (priceRecords.length === 0) {
+    logger.warn('No price records to insert');
+    return { insertedCount: 0 };
+  }
+
+  // Use updateOne with upsert to handle re-runs on same day
+  const operations = priceRecords.map(record => ({
+    updateOne: {
+      filter: {
+        unit_code: record.unit_code,
+        date_checked: record.date_checked
+      },
+      update: { $set: record },
+      upsert: true
+    }
+  }));
+
+  try {
+    const result = await collection.bulkWrite(operations, { ordered: false });
+
+    logger.info('Prices inserted', {
+      inserted: result.upsertedCount,
+      updated: result.modifiedCount
+    });
+
+    return { insertedCount: result.upsertedCount + result.modifiedCount };
+
+  } catch (error) {
+    logger.error('Failed to insert prices', {
+      errorType: error.name,
+      errorMessage: error.message
+    });
+    throw error;
+  }
+}
+
 module.exports = {
   fetchPage,
   parseUnits,
   convertDataTypes,
   upsertUnits,
+  insertPrices,
   // Export helpers for testing
   parseInteger,
   parsePositiveInteger,