fix: replay message queue dumping logic (#348)

* fix: replay message queue dumping logic * revert: previous logic to mark events as device mode processing done on App Update * test: add new db unit test case and fix old one
rudderlabs · Jul 31, 2023 · 307e730 · 307e730
1 parent 54d210c
commit 307e730
Show file tree

Hide file tree

Showing 10 changed files with 269 additions and 105 deletions.
diff --git a/Podfile.lock b/Podfile.lock
@@ -77,9 +77,9 @@ PODS:
   - Rudder-Braze (1.2.0):
     - Appboy-iOS-SDK (~> 4.5.4)
     - Rudder (~> 1.12)
-  - SDWebImage (5.16.0):
-    - SDWebImage/Core (= 5.16.0)
-  - SDWebImage/Core (5.16.0)
+  - SDWebImage (5.17.0):
+    - SDWebImage/Core (= 5.17.0)
+  - SDWebImage/Core (5.17.0)
 
 DEPENDENCIES:
   - Amplitude (~> 7.2.0)
@@ -124,7 +124,7 @@ SPEC CHECKSUMS:
   Rudder: 718bc5bf0de0e4d83850dedd792f08e5e2065a0a
   Rudder-Amplitude: f845cc125a1a58d4de6155391a2b0392815ae898
   Rudder-Braze: e42eb914a03cb418ed8b7a3cd90b724a91476631
-  SDWebImage: 2aea163b50bfcb569a2726b6a754c54a4506fcf6
+  SDWebImage: 750adf017a315a280c60fde706ab1e552a3ae4e9
 
 PODFILE CHECKSUM: 42bfa6ba9271b8d9518a669daca1fd9a7bbf9f6e
 

diff --git a/Sources/Classes/Headers/Public/RSDBMessage.h b/Sources/Classes/Headers/Public/RSDBMessage.h
@@ -14,6 +14,7 @@ NS_ASSUME_NONNULL_BEGIN
     @property (nonatomic) NSMutableArray<NSString *>* messages;
     @property (nonatomic) NSMutableArray<NSString *>* messageIds;
     @property (nonatomic) NSMutableArray<NSNumber *>* statusList;
+    @property (nonatomic) NSMutableArray<NSNumber *>* dmProcessed;
 @end
 
 NS_ASSUME_NONNULL_END
diff --git a/Sources/Classes/Headers/Public/RSDBPersistentManager.h b/Sources/Classes/Headers/Public/RSDBPersistentManager.h
@@ -24,8 +24,8 @@ NS_ASSUME_NONNULL_BEGIN
 -(void) createTables;
 -(void) createEventsTableWithVersion:(int) version;
 -(void) checkForMigrations;
--(BOOL) checkIfStatusColumnExists;
--(void) performMigration;
+-(BOOL) checkIfColumnExists:(NSString *) newColumn;
+-(void) performMigration:(NSString *) columnName;
 -(NSNumber*) saveEvent: (NSString*) message;
 - (void) clearOldEventsWithThreshold:(int)threshold;
 -(void) clearEventsFromDB: (NSMutableArray*) messageIds;
@@ -36,8 +36,10 @@ NS_ASSUME_NONNULL_BEGIN
 -(void) clearProcessedEventsFromDB;
 - (int) getDBRecordCountForMode:(MODES) mode;
 -(void) flushEventsFromDB;
--(void) updateDeviceModeEventsStatus;
-
+-(RSDBMessage*)fetchDeviceModeWithProcessedPendingEventsFromDb:(int) limit;
+-(int) getDeviceModeWithProcessedPendingEventsRecordCount;
+-(void) markDeviceModeTransformationAndProcessedDone:(NSNumber *) messageId;
+-(void) markDeviceModeProcessedDone:(NSNumber *) messageId;
 @end
 
 NS_ASSUME_NONNULL_END
diff --git a/Sources/Classes/Headers/Public/RSEnums.h b/Sources/Classes/Headers/Public/RSEnums.h
@@ -50,6 +50,11 @@ typedef enum {
     ALL = 3
 } MODES;
 
+typedef enum {
+    DM_PROCESSED_PENDING =0,
+    DM_PROCESSED_DONE =1,
+} DM_PROCESSED;
+
 typedef enum {
     ENABLED = YES,
     DISABLED = NO

diff --git a/Sources/Classes/Headers/Public/RSPreferenceManager.h b/Sources/Classes/Headers/Public/RSPreferenceManager.h
@@ -79,9 +79,6 @@ extern NSString *const RSSessionAutoTrackStatus;
 - (void) saveAutoTrackingStatus: (BOOL) autoTrackingStatus;
 - (BOOL) getAutoTrackingStatus;
 
-- (void) saveEventDeletionStatus;
-- (BOOL) getEventDeletionStatus;
-
 @end
 
 NS_ASSUME_NONNULL_END
diff --git a/Sources/Classes/RSDBPersistentManager.m b/Sources/Classes/RSDBPersistentManager.m
@@ -9,14 +9,15 @@
 #import "RSDBPersistentManager.h"
 #import "RSLogger.h"
 
-int const RS_DB_Version = 2;
+int const RS_DB_Version = 3;
 int const DEFAULT_STATUS_VALUE = 0;
 bool isReturnClauseSupported = NO;
 NSString* _Nonnull const TABLE_EVENTS = @"events";
 NSString* _Nonnull const COL_ID = @"id";
 NSString* _Nonnull const COL_MESSAGE = @"message";
 NSString* _Nonnull const COL_UPDATED = @"updated";
 NSString* _Nonnull const COL_STATUS = @"status";
+NSString* _Nonnull const COL_DM_PROCESSED = @"dm_processed";
 
 
 @implementation RSDBPersistentManager
@@ -47,44 +48,62 @@ - (void)createDB {
 // Migration is needed when an application is updated to the latest version of SDK from a version which doesn't has the status column in its events table
 - (void)checkForMigrations {
     [RSLogger logDebug:@"RSDBPersistentManager: checkForMigrations: checking if the event table has status column"];
-    if(![self checkIfStatusColumnExists]) {
+    bool isNewColumnAdded = NO;
+    if(![self checkIfColumnExists:COL_STATUS]) {
         [RSLogger logDebug:@"RSDBPersistentManager: checkForMigrations: events table doesn't has the status column performing migration"];
-        [self performMigration];
-        return;
+        [self performMigration:COL_STATUS];
+        isNewColumnAdded = YES;
+    }
+    if(![self checkIfColumnExists:COL_DM_PROCESSED]) {
+        [RSLogger logDebug:@"RSDBPersistentManager: checkForMigrations: events table doesn't has the dm_processed column performing migration"];
+        [self performMigration:COL_DM_PROCESSED];
+        isNewColumnAdded = YES;
+    }
+    if (!isNewColumnAdded) {
+        [RSLogger logDebug:@"RSDBPersistentManager: checkForMigrations: event table has status and dm_processed columns, no migration required"];
     }
-    [RSLogger logDebug:@"RSDBPersistentManager: checkForMigrations: event table has status column, no migration required"];
 }
 
-- (BOOL) checkIfStatusColumnExists {
-    NSString* checkIfStatusExistsSQLString = [[NSString alloc] initWithFormat:@"SELECT COUNT(*) from pragma_table_info(\"%@\") where name=\"%@\";", TABLE_EVENTS, COL_STATUS];
-    const char* statusCheckSQL = [checkIfStatusExistsSQLString UTF8String];
-    sqlite3_stmt *statusCheckStmt = nil;
-    BOOL statusColumnExists = NO;
-    if (sqlite3_prepare_v2(self->_database, statusCheckSQL, -1, &statusCheckStmt, nil) == SQLITE_OK) {
-        if(sqlite3_step(statusCheckStmt) == SQLITE_ROW) {
-            int count = sqlite3_column_int(statusCheckStmt, 0);
+- (BOOL) checkIfColumnExists:(NSString *) newColumn {
+    NSString* checkIfNewColumnExistsSQLString = [[NSString alloc] initWithFormat:@"SELECT COUNT(*) from pragma_table_info(\"%@\") where name=\"%@\";", TABLE_EVENTS, newColumn];
+    const char* newColumnCheckSQL = [checkIfNewColumnExistsSQLString UTF8String];
+    sqlite3_stmt *newColumnCheckStmt = nil;
+    BOOL newColumnExists = NO;
+    if (sqlite3_prepare_v2(self->_database, newColumnCheckSQL, -1, &newColumnCheckStmt, nil) == SQLITE_OK) {
+        if(sqlite3_step(newColumnCheckStmt) == SQLITE_ROW) {
+            int count = sqlite3_column_int(newColumnCheckStmt, 0);
             if(count > 0) {
-                statusColumnExists = YES;
+                newColumnExists = YES;
             }
         }
         else {
-            [RSLogger logWarn:[[NSString alloc] initWithFormat: @"RSDBPersistentManager: checkIfStatusColumnExists: SQLite Command Execution Failed: %@", checkIfStatusExistsSQLString]];
+            [RSLogger logWarn:[[NSString alloc] initWithFormat: @"RSDBPersistentManager: checkIfStatusColumnExists: SQLite Command Execution Failed: %@", checkIfNewColumnExistsSQLString]];
         }
     }
     else {
-        [RSLogger logError:[[NSString alloc] initWithFormat: @"RSDBPersistentManager: checkIfStatusColumnExists: SQLite Command Preparation Failed: %@", checkIfStatusExistsSQLString]];
+        [RSLogger logError:[[NSString alloc] initWithFormat: @"RSDBPersistentManager: checkIfStatusColumnExists: SQLite Command Preparation Failed: %@", checkIfNewColumnExistsSQLString]];
     }
-    sqlite3_finalize(statusCheckStmt);
-    return statusColumnExists;
+    sqlite3_finalize(newColumnCheckStmt);
+    return newColumnExists;
 }
 
-- (void) performMigration {
-    NSString* alterTableSQLString = [[NSString alloc] initWithFormat:@"ALTER TABLE %@ ADD COLUMN %@ INTEGER DEFAULT %d;", TABLE_EVENTS, COL_STATUS, DEFAULT_STATUS_VALUE];
-    NSString* updateTableSQLString = [[NSString alloc] initWithFormat:@"UPDATE %@ SET %@ = %d;", TABLE_EVENTS, COL_STATUS, DEVICE_MODE_PROCESSING_DONE];
-
-    if([self execSQL:alterTableSQLString] && [self execSQL:updateTableSQLString]) {
-        [RSLogger logDebug:@"RSDBPersistentManager: performMigration: events table migrated to add status column"];
-        return;
+- (void) performMigration:(NSString *) columnName {
+    if ([columnName isEqualToString:COL_STATUS]) {
+        NSString* alterTableSQLString = [[NSString alloc] initWithFormat:@"ALTER TABLE %@ ADD COLUMN %@ INTEGER DEFAULT %d;", TABLE_EVENTS, COL_STATUS, DEFAULT_STATUS_VALUE];
+        NSString* updateTableSQLString = [[NSString alloc] initWithFormat:@"UPDATE %@ SET %@ = %d;", TABLE_EVENTS, COL_STATUS, DEVICE_MODE_PROCESSING_DONE];
+
+        if([self execSQL:alterTableSQLString] && [self execSQL:updateTableSQLString]) {
+            [RSLogger logDebug:@"RSDBPersistentManager: performMigration: events table migrated to add status column"];
+            return;
+        }
+    } else if ([columnName isEqualToString:COL_DM_PROCESSED]) {
+        NSString* alterTableSQLString = [[NSString alloc] initWithFormat:@"ALTER TABLE %@ ADD COLUMN %@ INTEGER DEFAULT %d;", TABLE_EVENTS, COL_DM_PROCESSED, DM_PROCESSED_PENDING];
+        NSString* updateTableSQLString = [[NSString alloc] initWithFormat:@"UPDATE %@ SET %@ = (%@ | %d), %@ = %d;", TABLE_EVENTS, COL_STATUS, COL_STATUS, DEVICE_MODE_PROCESSING_DONE, COL_DM_PROCESSED, DM_PROCESSED_DONE];
+
+        if([self execSQL:alterTableSQLString] && [self execSQL:updateTableSQLString]) {
+            [RSLogger logDebug:@"RSDBPersistentManager: performMigration: events table migrated to add dm_processed column"];
+            return;
+        }
     }
     [RSLogger logError:@"RSDBPersistentManager: performMigration: events table migration failed"];
 }
@@ -99,6 +118,9 @@ -(void) createEventsTableWithVersion:(int) version {
         case 1:
             createTableSQLString = [[NSString alloc] initWithFormat:@"CREATE TABLE IF NOT EXISTS %@( %@ INTEGER PRIMARY KEY AUTOINCREMENT, %@ TEXT NOT NULL, %@ INTEGER NOT NULL);", TABLE_EVENTS, COL_ID, COL_MESSAGE, COL_UPDATED];
             break;
+        case 3:
+            createTableSQLString = [[NSString alloc] initWithFormat:@"CREATE TABLE IF NOT EXISTS %@( %@ INTEGER PRIMARY KEY AUTOINCREMENT, %@ TEXT NOT NULL, %@ INTEGER NOT NULL, %@ INTEGER DEFAULT %d, %@ INTEGER DEFAULT %d);", TABLE_EVENTS, COL_ID, COL_MESSAGE, COL_UPDATED, COL_STATUS, DEFAULT_STATUS_VALUE, COL_DM_PROCESSED, DM_PROCESSED_PENDING];
+            break;
         default:
             createTableSQLString = [[NSString alloc] initWithFormat:@"CREATE TABLE IF NOT EXISTS %@( %@ INTEGER PRIMARY KEY AUTOINCREMENT, %@ TEXT NOT NULL, %@ INTEGER NOT NULL, %@ INTEGER DEFAULT %d);", TABLE_EVENTS, COL_ID, COL_MESSAGE, COL_UPDATED, COL_STATUS, DEFAULT_STATUS_VALUE];
     }
@@ -174,17 +196,6 @@ - (void)clearEventsFromDB:(NSMutableArray<NSString *> *)messageIds {
     }
 }
 
--(void) updateDeviceModeEventsStatus {
-    NSString* updateEventStatusSQL = [[NSString alloc] initWithFormat:@"UPDATE %@ SET %@ = (%@ | %d) WHERE %@ IN (%d, %d);", TABLE_EVENTS, COL_STATUS, COL_STATUS, DEVICE_MODE_PROCESSING_DONE, COL_STATUS, NOT_PROCESSED, CLOUD_MODE_PROCESSING_DONE];
-    @synchronized (self) {
-        if([self execSQL:updateEventStatusSQL]) {
-            [RSLogger logDebug:[[NSString alloc] initWithFormat:@"RSDBPersistentManager: updateDeviceModeEventsStatus: Successfully updated the event status for unprocessed and cloud mode processing done events to device mode processing done."]];
-            return;
-        }
-        [RSLogger logError:[[NSString alloc] initWithFormat:@"RSDBPersistentManager: updateDeviceModeEventsStatuse: Failed to update the status for events."]];
-    }
-}
-
 -(RSDBMessage *)fetchEventsFromDB:(int)count ForMode:(MODES) mode {
     NSString* querySQLString = nil;
     switch(mode) {
@@ -221,6 +232,7 @@ - (RSDBMessage *) getEventsFromDB :(NSString*) querySQLString {
     NSMutableArray<NSString *> *messageIds = [[NSMutableArray alloc] init];
     NSMutableArray<NSString *> *messages = [[NSMutableArray alloc] init];
     NSMutableArray<NSNumber *>* statusList = [[NSMutableArray alloc] init];
+    NSMutableArray<NSNumber *>* dmProcessedList = [[NSMutableArray alloc] init];
 
     @synchronized (self) {
         sqlite3_stmt *queryStmt = nil;
@@ -231,9 +243,11 @@ - (RSDBMessage *) getEventsFromDB :(NSString*) querySQLString {
                 const unsigned char* queryResultCol1 = sqlite3_column_text(queryStmt, 1);
                 NSString *message = [[NSString alloc] initWithUTF8String:(char *)queryResultCol1];
                 int status = sqlite3_column_int(queryStmt,3);
+                int dmProcessed = sqlite3_column_int(queryStmt, 4);
                 [messageIds addObject:[[NSString alloc] initWithFormat:@"%d", messageId]];
                 [messages addObject:message];
                 [statusList addObject:[NSNumber numberWithInt:status]];
+                [dmProcessedList addObject:[NSNumber numberWithInt:dmProcessed]];
             }
         } else {
             [RSLogger logError:@"RSDBPersistentManager: getEventsFromDB: Failed to fetch events from DB"];
@@ -244,6 +258,7 @@ - (RSDBMessage *) getEventsFromDB :(NSString*) querySQLString {
     dbMessage.messageIds = messageIds;
     dbMessage.messages = messages;
     dbMessage.statusList = statusList;
+    dbMessage.dmProcessed = dmProcessedList;
     return dbMessage;
 }
 
@@ -261,6 +276,10 @@ - (int) getDBRecordCountForMode:(MODES) mode {
             countSQLString = [[NSString alloc] initWithFormat:@"SELECT COUNT(*) FROM %@", TABLE_EVENTS];
     }
     [RSLogger logDebug:[[NSString alloc] initWithFormat:@"RSDBPersistentManager: getDBRecordCount: countSQLString: %@", countSQLString]];
+    return [self getDBRecordCoun:countSQLString];
+}
+
+-(int) getDBRecordCoun:(NSString *) countSQLString {
     int count = 0;
     const char* countSQL = [countSQLString UTF8String];
     @synchronized (self) {
@@ -368,4 +387,37 @@ - (NSString *) getSQLiteVersion {
     }
     return sqliteVersion;
 }
+
+-(RSDBMessage*)fetchDeviceModeWithProcessedPendingEventsFromDb:(int) limit {
+    NSString* querySQLString = [[NSString alloc] initWithFormat:@"SELECT * FROM %@ WHERE %@ IN (%d,%d) AND %@ = %d ORDER BY %@ ASC LIMIT %d;", TABLE_EVENTS, COL_STATUS, NOT_PROCESSED, CLOUD_MODE_PROCESSING_DONE, COL_DM_PROCESSED, DM_PROCESSED_PENDING, COL_UPDATED, limit];
+    return [self getEventsFromDB:querySQLString];
+}
+
+-(int) getDeviceModeWithProcessedPendingEventsRecordCount {
+    NSString *countSQLString = [[NSString alloc] initWithFormat:@"SELECT COUNT(*) FROM %@ where %@ IN (%d,%d) AND %@ = %d", TABLE_EVENTS, COL_STATUS, NOT_PROCESSED, CLOUD_MODE_PROCESSING_DONE, COL_DM_PROCESSED, DM_PROCESSED_PENDING];
+    [RSLogger logDebug:[[NSString alloc] initWithFormat:@"RSDBPersistentManager: getDeviceModeWithProcessedPendingEventsRecordCount: countSQLString: %@", countSQLString]];
+    return [self getDBRecordCoun:countSQLString];
+}
+
+-(void) markDeviceModeTransformationAndProcessedDone:(NSNumber *) messageId {
+    NSString* updateEventStatusSQL = [[NSString alloc] initWithFormat:@"UPDATE %@ SET %@ = %@ | %d, %@ = %d WHERE %@ = %@;", TABLE_EVENTS, COL_STATUS, COL_STATUS, DEVICE_MODE_PROCESSING_DONE, COL_DM_PROCESSED, DM_PROCESSED_DONE, COL_ID, messageId];
+    @synchronized (self) {
+        if([self execSQL:updateEventStatusSQL]) {
+            [RSLogger logDebug:[[NSString alloc] initWithFormat:@"RSDBPersistentManager: markDeviceModeTransformationAndProcessedDone: Successfully updated the event status and dm_processed columns for events %@", messageId]];
+            return;
+        }
+        [RSLogger logDebug:[[NSString alloc] initWithFormat:@"RSDBPersistentManager: markDeviceModeTransformationAndProcessedDone: Failed to update the event status and dm_processed columns for events %@", messageId]];
+    }
+}
+
+-(void) markDeviceModeProcessedDone:(NSNumber *) messageId {
+    NSString* updateEventStatusSQL = [[NSString alloc] initWithFormat:@"UPDATE %@ SET %@ = %d WHERE %@ = %@;", TABLE_EVENTS, COL_DM_PROCESSED, DM_PROCESSED_DONE, COL_ID, messageId];
+    @synchronized (self) {
+        if([self execSQL:updateEventStatusSQL]) {
+            [RSLogger logDebug:[[NSString alloc] initWithFormat:@"RSDBPersistentManager: markDeviceModeProcessedDone: Successfully updated the event dm_processed for events %@", messageId]];
+            return;
+        }
+        [RSLogger logDebug:[[NSString alloc] initWithFormat:@"RSDBPersistentManager: markDeviceModeProcessedDone: Failed to update the dm_processed for events %@", messageId]];
+    }
+}
 @end