feat: Add ProcessBulkImportUsers cron job

supertokens · Mar 20, 2024 · ca2472a · ca2472a
1 parent e2930b3
commit ca2472a
Show file tree

Hide file tree

Showing 5 changed files with 44 additions and 10 deletions.
diff --git a/src/main/java/io/supertokens/pluginInterface/Storage.java b/src/main/java/io/supertokens/pluginInterface/Storage.java
@@ -32,6 +32,8 @@ public interface Storage {
     // if silent is true, do not log anything out on the console
     void constructor(String processId, boolean silent, boolean isTesting);
 
+    Storage createBulkImportProxyStorageInstance();
+
     void loadConfig(JsonObject jsonConfig, Set<LOG_LEVEL> logLevels, TenantIdentifier tenantIdentifier) throws InvalidConfigException;
 
     // this returns a unique ID based on the db's connection URI and table prefix such that

diff --git a/src/main/java/io/supertokens/pluginInterface/StorageUtils.java b/src/main/java/io/supertokens/pluginInterface/StorageUtils.java
@@ -16,8 +16,8 @@
 
 package io.supertokens.pluginInterface;
 
-import io.supertokens.pluginInterface.authRecipe.AuthRecipeStorage;
 import io.supertokens.pluginInterface.authRecipe.sqlStorage.AuthRecipeSQLStorage;
+import io.supertokens.pluginInterface.bulkimport.sqlStorage.BulkImportSQLStorage;
 import io.supertokens.pluginInterface.dashboard.sqlStorage.DashboardSQLStorage;
 import io.supertokens.pluginInterface.emailpassword.sqlStorage.EmailPasswordSQLStorage;
 import io.supertokens.pluginInterface.emailverification.sqlStorage.EmailVerificationSQLStorage;
@@ -133,4 +133,12 @@ public static MultitenancyStorage getMultitenancyStorage(Storage storage) {
         }
         return (MultitenancyStorage) storage;
     }
+
+    public static BulkImportSQLStorage getBulkImportStorage(Storage storage) {
+        if (storage.getType() != STORAGE_TYPE.SQL) {
+            // we only support SQL for now
+            throw new UnsupportedOperationException("");
+        }
+        return (BulkImportSQLStorage) storage;
+    }
 }
diff --git a/src/main/java/io/supertokens/pluginInterface/bulkimport/BulkImportStorage.java b/src/main/java/io/supertokens/pluginInterface/bulkimport/BulkImportStorage.java
@@ -47,6 +47,11 @@ List<BulkImportUser> getBulkImportUsers(AppIdentifier appIdentifier, @Nonnull In
      */
     List<String> deleteBulkImportUsers(AppIdentifier appIdentifier, @Nonnull String[] bulkImportUserIds) throws StorageQueryException;
 
+    /**
+     * Returns the users from the bulk_import_users table for processing
+     */
+    List<BulkImportUser> getBulkImportUsersForProcessing(AppIdentifier appIdentifier, @Nonnull Integer limit) throws StorageQueryException;
+
     public enum BULK_IMPORT_USER_STATUS {
         NEW, PROCESSING, FAILED
     }

diff --git a/src/main/java/io/supertokens/pluginInterface/bulkimport/BulkImportUser.java b/src/main/java/io/supertokens/pluginInterface/bulkimport/BulkImportUser.java
@@ -27,16 +27,17 @@ public class BulkImportUser {
     public String id;
     public String externalUserId;
     public JsonObject userMetadata;
-    public List<String> userRoles;
+    public List<UserRole> userRoles;
     public List<TotpDevice> totpDevices;
     public List<LoginMethod> loginMethods;
 
     // Following fields come from the DB Record.
     public BULK_IMPORT_USER_STATUS status;
+    public String errorMessage;
     public Long createdAt;
     public Long updatedAt;
 
-    public BulkImportUser(String id, String externalUserId, JsonObject userMetadata, List<String> userRoles,
+    public BulkImportUser(String id, String externalUserId, JsonObject userMetadata, List<UserRole> userRoles,
             List<TotpDevice> totpDevices, List<LoginMethod> loginMethods) {
         this.id = id;
         this.externalUserId = externalUserId;
@@ -46,7 +47,7 @@ public BulkImportUser(String id, String externalUserId, JsonObject userMetadata,
         this.loginMethods = loginMethods;
     }
 
-    public static BulkImportUser fromTesting_fromJson(JsonObject jsonObject) {
+    public static BulkImportUser forTesting_fromJson(JsonObject jsonObject) {
         return new Gson().fromJson(jsonObject, BulkImportUser.class);
     }
 
@@ -60,10 +61,11 @@ public String toRawDataForDbStorage() {
         return jsonObject.toString();
     }
 
-    public static BulkImportUser fromRawDataFromDbStorage(String id, String rawData, BULK_IMPORT_USER_STATUS status, long createdAt, long updatedAt) {
+    public static BulkImportUser fromRawDataFromDbStorage(String id, String rawData, BULK_IMPORT_USER_STATUS status, String errorMessage, long createdAt, long updatedAt) {
         BulkImportUser user = new Gson().fromJson(rawData, BulkImportUser.class);
         user.id = id;
         user.status = status;
+        user.errorMessage = errorMessage;
         user.createdAt = createdAt;
         user.updatedAt = updatedAt;
         return user;
@@ -73,6 +75,16 @@ public JsonObject toJsonObject() {
         return new Gson().fromJson(new Gson().toJson(this), JsonObject.class);
     }
 
+    public static class UserRole {
+        public String role;
+        public List<String> tenantIds;
+
+        public UserRole(String role, List<String> tenantIds) {
+            this.role = role;
+            this.tenantIds = tenantIds;
+        }
+    }
+
     public static class TotpDevice {
         public String secretKey;
         public int period;
@@ -88,7 +100,7 @@ public TotpDevice(String secretKey, int period, int skew, String deviceName) {
     }
 
     public static class LoginMethod {
-        public String tenantId;
+        public List<String> tenantIds;
         public boolean isVerified;
         public boolean isPrimary;
         public long timeJoinedInMSSinceEpoch;
@@ -99,11 +111,12 @@ public static class LoginMethod {
         public String thirdPartyId;
         public String thirdPartyUserId;
         public String phoneNumber;
+        public String superTokensOrExternalUserId;
 
-        public LoginMethod(String tenantId, String recipeId, boolean isVerified, boolean isPrimary,
+        public LoginMethod(List<String> tenantIds, String recipeId, boolean isVerified, boolean isPrimary,
                 long timeJoinedInMSSinceEpoch, String email, String passwordHash, String hashingAlgorithm,
                 String thirdPartyId, String thirdPartyUserId, String phoneNumber) {
-            this.tenantId = tenantId;
+            this.tenantIds = tenantIds;
             this.recipeId = recipeId;
             this.isVerified = isVerified;
             this.isPrimary = isPrimary;
@@ -116,4 +129,4 @@ public LoginMethod(String tenantId, String recipeId, boolean isVerified, boolean
             this.phoneNumber = phoneNumber;
         }
     }
-}
+}
diff --git a/src/main/java/io/supertokens/pluginInterface/bulkimport/sqlStorage/BulkImportSQLStorage.java b/src/main/java/io/supertokens/pluginInterface/bulkimport/sqlStorage/BulkImportSQLStorage.java
@@ -17,6 +17,7 @@
 package io.supertokens.pluginInterface.bulkimport.sqlStorage;
 
 import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 
 import io.supertokens.pluginInterface.bulkimport.BulkImportStorage;
 import io.supertokens.pluginInterface.exceptions.StorageQueryException;
@@ -30,5 +31,10 @@ public interface BulkImportSQLStorage extends BulkImportStorage, SQLStorage {
      * Update the status of the users in the bulk_import_users table
      */
     void updateBulkImportUserStatus_Transaction(AppIdentifier appIdentifier,
-            TransactionConnection con, @Nonnull String[] bulkImportUserIds, @Nonnull BULK_IMPORT_USER_STATUS status) throws StorageQueryException;
+            TransactionConnection con, @Nonnull String[] bulkImportUserIds, @Nonnull BULK_IMPORT_USER_STATUS status, @Nullable String errorMessage) throws StorageQueryException;
+
+    /**
+     * Delete users by id from the bulk_import_users table
+     */
+    void deleteBulkImportUser_Transaction(AppIdentifier appIdentifier, TransactionConnection con, @Nonnull String bulkImportUserId) throws StorageQueryException;
 }