finished ceo

apps/stock/data-ingestion/src/handlers/ceo/actions/get-posts.action.ts

@@ -6,7 +6,7 @@ export async function getPosts(
   payload: any,
   _context: any
 ): Promise<unknown> {
-  const { ceoId, symbol, timestamp, untilTimestamp } = payload;
+  const { ceoId, symbol, timestamp, untilTimestamp, finished } = payload;
   const proxy = this.proxy?.getProxy();
   if (!proxy) {
     this.logger.warn('No proxy available for processing individual CEO symbol');
@@ -41,9 +41,9 @@ export async function getPosts(
     if (spielCount === 0) {
       this.logger.warn(`No spiels found for ceoId ${ceoId}`);
       await this.mongodb.updateMany(
-        'ceoChannels',
+        'ceoSymbols',
         { ceoId },
-        { $set: { lastSpielTime: timestamp, finished: true } }
+        { $set: { lastSpielTime: timestamp || Date.now(), finished: true } }
       );
       return null; // No data to process
     }
@@ -83,7 +83,7 @@ export async function getPosts(
 
     await this.mongodb.batchUpsert('ceoPosts', posts, ['spielId']);
     await this.mongodb.updateMany(
-      'ceoChannels',
+      'ceoSymbols',
       { ceoId },
       { $set: { lastSpielTime: latestSpielTime } }
     );
@@ -91,12 +91,14 @@ export async function getPosts(
 
     // If untilTimestamp is not provider keep going to the end
     // Otherwise keep going until the lastSpiel is before the untilTimestamp
-    if( untilTimestamp === undefined || (untilTimestamp && latestSpielTime <= untilTimestamp)) {
+    if( !finished || (finished && untilTimestamp && latestSpielTime > untilTimestamp)) {
       await this.scheduleOperation(
         'get-posts',
         {
           ceoId: ceoId,
           timestamp: latestSpielTime,
+          finished: finished,
+          untilTimestamp: untilTimestamp,
         },
         { priority: 0 }
       );

apps/stock/data-ingestion/src/handlers/ceo/actions/update-unique-symbols.action.ts

@@ -36,6 +36,8 @@ export async function updateUniqueSymbols(
         {
           ceoId: symbol.ceoId,
           symbol: symbol.symbol,
+          finished: symbol.finished || false,
+          untilTimestamp: symbol.lastSpielTime || null,
         },
         { priority: 10 }
       );