Use last row read for training sheet

Author: Lan2u

src/init-dependencies/google/cache-sheet-data.ts

@@ -1,8 +1,7 @@
 import {Client} from '@libsql/client/.';
 import {Logger} from 'pino';
-import * as O from 'fp-ts/Option';
 import {dbExecute} from '../../util';
-import {LastGoogleSheetRowRead} from '../../dependencies';
+import {LastGoogleSheetRowRead} from '../../read-models/shared-state/return-types';
 
 // This would be more efficient with a simple key-value store.
 export const cacheSheetData =
@@ -32,12 +31,7 @@ export const cacheSheetData =
             last_row_read = excluded.last_row_read,
             cached_data = excluded.cached_data;
         `,
-        [
-          cacheTimestamp,
-          sheetId,
-          O.getOrElse<null | number>(() => null)(last_row_read),
-          cachedData,
-        ]
+        [cacheTimestamp, sheetId, JSON.stringify(last_row_read), cachedData]
       );
     } catch (e) {
       logger.error(e, 'Failed to insert cache data, failing silently...');

src/read-models/external-event-sources/google/training-sheet.ts

@@ -33,6 +33,8 @@ import * as RA from 'fp-ts/ReadonlyArray';
 import {v4} from 'uuid';
 import {lookup} from 'fp-ts/ReadonlyArray';
 import {array} from 'fp-ts';
+import {LastGoogleSheetRowRead} from '../../shared-state/return-types';
+import * as R from 'fp-ts/Record';
 
 const ROW_BATCH_SIZE = 50;
 const FORM_RESPONSES_SHEET_REGEX = /^Form Responses [0-9]*/i;
@@ -178,19 +180,22 @@ export const columnBoundsRequired = (
   return [Math.min(...colIndexes), Math.max(...colIndexes)];
 };
 
+type LastRowRead = number;
+
 const pullNewEquipmentQuizResultsForSheet = async (
   logger: Logger,
   googleHelpers: GoogleHelpers,
   equipmentId: UUID,
   trainingSheetId: string,
   sheet: GoogleSheetMetadata,
   timezone: string,
+  prevLastRowRead: O.Option<LastRowRead>,
   updateState: (event: EventOfType<'EquipmentTrainingQuizResult'>) => void
-): Promise<void> => {
+): Promise<LastRowRead> => {
   logger = logger.child({sheet_name: sheet.name});
   logger.info('Processing sheet');
   for (const [rowStart, rowEnd] of getChunkIndexes(
-    2, // 1-indexed and first row is headers.
+    O.getOrElse(() => 1)(prevLastRowRead) + 1, // 1-indexed and first row is headers.
     sheet.rowCount,
     ROW_BATCH_SIZE
   )) {
@@ -214,7 +219,7 @@ const pullNewEquipmentQuizResultsForSheet = async (
         rowStart,
         rowEnd
       );
-      return;
+      return rowStart - 1;
     }
     logger.info('Pulled data from google, extracting...');
     const result = extractGoogleSheetData(
@@ -232,19 +237,21 @@ const pullNewEquipmentQuizResultsForSheet = async (
     }
   }
   logger.info('Finished processing sheet');
+  return sheet.rowCount - 1;
 };
 
 export const pullNewEquipmentQuizResults = async (
   logger: Logger,
   googleHelpers: GoogleHelpers,
   equipmentId: UUID,
   trainingSheetId: string,
+  prevLastRowsRead: Readonly<LastGoogleSheetRowRead>,
   updateState: (
     event:
       | EventOfType<'EquipmentTrainingQuizSync'>
       | EventOfType<'EquipmentTrainingQuizResult'>
   ) => void
-): Promise<void> => {
+): Promise<Readonly<LastGoogleSheetRowRead>> => {
   logger.info('Scanning training sheet. Pulling google sheet data...');
 
   const initialMeta = await googleHelpers.pullGoogleSheetDataMetadata(
@@ -253,7 +260,7 @@ export const pullNewEquipmentQuizResults = async (
   )();
   if (E.isLeft(initialMeta)) {
     logger.warn(initialMeta.left);
-    return;
+    return prevLastRowsRead;
   }
 
   logger.info('Got meta data for sheet...');
@@ -298,27 +305,43 @@ export const pullNewEquipmentQuizResults = async (
     sheets.push(meta.value);
   }
 
+  const newLastRowRead: LastGoogleSheetRowRead = JSON.parse(
+    JSON.stringify(prevLastRowsRead)
+  ) as LastGoogleSheetRowRead;
+
   for (const sheet of sheets) {
-    await pullNewEquipmentQuizResultsForSheet(
+    const prevLastRowReadForSheet = pipe(
+      prevLastRowsRead,
+      R.lookup(trainingSheetId),
+      O.flatMap(R.lookup(trainingSheetId))
+    );
+    const lastRowRead = await pullNewEquipmentQuizResultsForSheet(
       logger,
       googleHelpers,
       equipmentId,
       trainingSheetId,
       sheet,
       initialMeta.right.properties.timeZone,
+      prevLastRowReadForSheet,
       updateState
     );
+    if (!newLastRowRead[trainingSheetId]) {
+      newLastRowRead[trainingSheetId] = {};
+    }
+    newLastRowRead[trainingSheetId][sheet.name] = lastRowRead;
   }
 
   logger.info(
-    'Finished pulling equipment quiz results for all sheets, generating quiz sync event...'
+    'Finished pulling equipment quiz results for all sheets %o, generating quiz sync event...',
+    newLastRowRead
   );
 
   updateState(
     constructEvent('EquipmentTrainingQuizSync')({
       equipmentId,
     })
   );
+  return newLastRowRead;
 };
 
 export async function asyncApplyTrainingSheetEvents(
@@ -371,11 +394,12 @@ export async function asyncApplyTrainingSheetEvents(
           updateState(event);
         };
 
-        await pullNewEquipmentQuizResults(
+        const lastRowRead = await pullNewEquipmentQuizResults(
           equipmentLogger,
           googleHelpers,
           equipment.id,
           equipmentTrainingSheetId,
+          equipment.lastRowsRead,
           collectEvents
         );
         equipmentLogger.info(
@@ -386,6 +410,7 @@ export async function asyncApplyTrainingSheetEvents(
           new Date(),
           equipmentTrainingSheetId,
           equipmentLogger,
+          lastRowRead,
           events
         );
       }

src/read-models/shared-state/return-types.ts

@@ -44,8 +44,12 @@ export type TrainerInfo = Pick<
   trainerSince: Date;
 };
 
+export type GoogleSheetId = string;
 export type SheetName = string;
-export type LastGoogleSheetRowRead = Record<SheetName, number>;
+export type LastGoogleSheetRowRead = Record<
+  GoogleSheetId,
+  Record<SheetName, number>
+>;
 
 export type MinimalEquipment = {
   id: UUID;