FAI-14708 Mock data feed should reset models BEFORE writing new recor…

…ds (#1913)
faros-ai · Feb 12, 2025 · fc9f84f · fc9f84f
1 parent 99b5199
commit fc9f84f
Showing 1 changed file with 14 additions and 0 deletions.
diff --git a/destinations/airbyte-faros-destination/src/destination.ts b/destinations/airbyte-faros-destination/src/destination.ts
@@ -816,6 +816,13 @@ export class FarosDestination extends AirbyteDestination<DestinationConfig> {
                 msg.sourceType === 'faros-feeds'
                   ? msg.sourceMode
                   : msg.sourceType;
+
+              if (sourceModeOrType === 'mock-data-feed') {
+                this.logger.info('Running a mock data feed sync. Resetting all models before writing records.');
+                ctx.markAllStreamsForReset();
+                await resetData?.(isResetSync);
+              }
+
               await updateLocalAccount?.(msg);
               if (sourceVersion) sourceVersion.version = msg.sourceVersion;
               ctx.setSourceConfig(msg.redactedConfig);
@@ -968,6 +975,7 @@ export class FarosDestination extends AirbyteDestination<DestinationConfig> {
           isBackfillSync,
           isFarosSource: sourceConfigReceived,
           sourceSucceeded,
+          sourceModeOrType,
         })
       ) {
         if (!streamStatusReceived) {
@@ -1010,6 +1018,7 @@ export class FarosDestination extends AirbyteDestination<DestinationConfig> {
     isBackfillSync: boolean;
     isFarosSource: boolean;
     sourceSucceeded: boolean;
+    sourceModeOrType: string;
   }): boolean {
     const {
       syncErrors,
@@ -1019,13 +1028,18 @@ export class FarosDestination extends AirbyteDestination<DestinationConfig> {
       isBackfillSync,
       isFarosSource,
       sourceSucceeded,
+      sourceModeOrType,
     } = syncInfo;
     if (isBackfillSync) {
       this.logger.info(
         'Running a backfill sync. Skipping reset of non-incremental models.'
       );
       return false;
     }
+    if (sourceModeOrType === 'mock-data-feed') {
+      // Reset is performed BEFORE writing records, so we don't need to reset again
+      return false;
+    }
     if (streamStatusReceived) {
       if (syncErrors.dst.length) {
         this.logger.warn(