feat: multithreaded bulk import (#1077) (#1079)

* feat: Add BulkImport APIs and cron

* chore: update pull request template

* fix: Use the correct tenant config to create the proxy storage

* fix: PR changes

* fix: PR changes

* fix: PR changes

* fix: PR changes

* fix: PR changes

* fix: PR changes

* fix: Update version

* fix: PR changes

* fix: PR changes

* fix: Rename DeleteBulkImportUser API path

* fix: disable bulk import for in-memory db

* fix: a bug with createTotpDevices

* fix: PR changes

* feat: Add an api to import user in sync

* feat: Add an api to get count of bulk import users

* fix: PR changes

* fix: Add error codes and plainTextPassword import

* fix: PR changes

* feat: multithreaded bulk import

* fix: changelog update

* fix: add new test

* fix: fixing unreliable mutithreaded bulk import with mysql

* fix: review fixes

* fix: fixing failing tests

* feat: bulkimport flow tests

* feat: bulk import cron starter api

* fix: tweaking params for faster import

* fix: tests

* checkpoint

* fix: remove vacuuming

* fix: minor tweaks

* feat: bulk inserting the bulk migration data

* fix: fast as a lightning

* fix: restoring lost method

* fix: reworked error handling to comform previous approach with messages

* fix: fixing tests

* fix: fixing failing tests, changing version

* chore: update changelog

* fix: fixing issues and failing tests

* fix: review changes

* fix: review fixes, reworking cron start/stop

---------

Co-authored-by: Tamas Soltesz <[email protected]>
Co-authored-by: Ankit Tiwari <[email protected]>

Author: 3 people

.github/PULL_REQUEST_TEMPLATE.md

@@ -37,6 +37,7 @@ highlighting the necessary changes)
     - If no such branch exists, then create one from the latest released branch.
 - [ ] If added a foreign key constraint on `app_id_to_user_id` table, make sure to delete from this table when deleting
   the user as well if `deleteUserIdMappingToo` is false.
+- [ ] If added a new recipe, then make sure to update the bulk import API to include the new recipe.
 
 ## Remaining TODOs for this PR
 

CHANGELOG.md

@@ -7,6 +7,46 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
 
+## [9.4.0]
+
+### Added
+- Adds property `bulk_migration_parallelism` for fine-tuning the worker threads number
+- Adds APIs to bulk import users
+    - GET `/bulk-import/users`
+    - POST `/bulk-import/users`
+    - GET `/bulk-import/users/count`
+    - POST `/bulk-import/users/remove`
+    - POST `/bulk-import/users/import`
+    - POST `/bulk-import/backgroundjob`
+    - GET `/bulk-import/backgroundjob`
+- Adds `ProcessBulkImportUsers` cron job to process bulk import users
+- Adds multithreaded worker support for the `ProcessBulkImportUsers` cron job for faster bulk imports
+- Adds support for lazy importing users
+
+### Migrations
+
+```sql
+"CREATE TABLE IF NOT EXISTS bulk_import_users (
+                id CHAR(36),
+                app_id VARCHAR(64) NOT NULL DEFAULT 'public',
+                primary_user_id VARCHAR(36),
+                raw_data TEXT NOT NULL,
+                status VARCHAR(128) DEFAULT 'NEW',
+                error_msg TEXT,
+                created_at BIGINT NOT NULL, 
+                updated_at BIGINT NOT NULL, 
+                CONSTRAINT bulk_import_users_pkey PRIMARY KEY(app_id, id),
+                CONSTRAINT bulk_import_users__app_id_fkey FOREIGN KEY(app_id) REFERENCES apps(app_id) ON DELETE CASCADE
+);
+
+CREATE INDEX IF NOT EXISTS bulk_import_users_status_updated_at_index ON bulk_import_users (app_id, status, updated_at);
+
+CREATE INDEX IF NOT EXISTS bulk_import_users_pagination_index1 ON bulk_import_users (app_id, status, created_at DESC,
+ id DESC);
+ 
+CREATE INDEX IF NOT EXISTS bulk_import_users_pagination_index2 ON bulk_import_users (app_id, created_at DESC, id DESC);
+```
+
 ## [9.3.1]
 
 - Includes exception class name in 500 error message

build.gradle

@@ -19,8 +19,7 @@ compileTestJava { options.encoding = "UTF-8" }
 //    }
 //}
 
-version = "9.3.1"
-
+version = "9.4.0"
 
 repositories {
     mavenCentral()

config.yaml

@@ -170,3 +170,7 @@ core_config_version: 0
 
 # (Optional | Default: null) string value. The encryption key used for saving OAuth client secret on the database.
 # oauth_client_secret_encryption_key:
+
+# (DIFFERENT_ACROSS_APPS | OPTIONAL | Default: number of available processor cores) int value. If specified,
+# the supertokens core will use the specified number of threads to complete the migration of users.
+# bulk_migration_parallelism:

devConfig.yaml

@@ -170,3 +170,8 @@ disable_telemetry: true
 
 # (Optional | Default: null) string value. The encryption key used for saving OAuth client secret on the database.
 # oauth_client_secret_encryption_key:
+
+# (DIFFERENT_ACROSS_APPS | OPTIONAL | Default: number of available processor cores) int value. If specified,
+# the supertokens core will use the specified number of threads to complete the migration of users.
+# bulk_migration_parallelism:
+

src/main/java/io/supertokens/Main.java

@@ -20,6 +20,7 @@
 import io.supertokens.config.Config;
 import io.supertokens.config.CoreConfig;
 import io.supertokens.cronjobs.Cronjobs;
+import io.supertokens.cronjobs.bulkimport.ProcessBulkImportUsers;
 import io.supertokens.cronjobs.cleanupOAuthSessionsAndChallenges.CleanupOAuthSessionsAndChallenges;
 import io.supertokens.cronjobs.deleteExpiredAccessTokenSigningKeys.DeleteExpiredAccessTokenSigningKeys;
 import io.supertokens.cronjobs.deleteExpiredDashboardSessions.DeleteExpiredDashboardSessions;
@@ -61,6 +62,8 @@ public class Main {
 
     // this is a special variable that will be set to true by TestingProcessManager
     public static boolean isTesting = false;
+    // this flag is used in ProcessBulkImportUsersCronJobTest to skip the user validation
+    public static boolean isTesting_skipBulkImportUserValidationInCronJob = false;
 
     // this is a special variable that will be set to true by TestingProcessManager
     public static boolean makeConsolePrintSilent = false;
@@ -257,6 +260,9 @@ private void init() throws IOException, StorageQueryException {
         // starts DeleteExpiredAccessTokenSigningKeys cronjob if the access token signing keys can change
         Cronjobs.addCronjob(this, DeleteExpiredAccessTokenSigningKeys.init(this, uniqueUserPoolIdsTenants));
 
+        // initializes ProcessBulkImportUsers cronjob to process bulk import users
+        Cronjobs.addCronjob(this, ProcessBulkImportUsers.init(this, uniqueUserPoolIdsTenants));
+
         Cronjobs.addCronjob(this, CleanupOAuthSessionsAndChallenges.init(this, uniqueUserPoolIdsTenants));
 
         // this is to ensure tenantInfos are in sync for the new cron job as well

src/main/java/io/supertokens/StorageAndUserIdMappingForBulkImport.java

@@ -0,0 +1,31 @@
+/*
+ *    Copyright (c) 2024, VRAI Labs and/or its affiliates. All rights reserved.
+ *
+ *    This software is licensed under the Apache License, Version 2.0 (the
+ *    "License") as published by the Apache Software Foundation.
+ *
+ *    You may not use this file except in compliance with the License. You may
+ *    obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *    Unless required by applicable law or agreed to in writing, software
+ *    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ *    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ *    License for the specific language governing permissions and limitations
+ *    under the License.
+ */
+
+package io.supertokens;
+
+import io.supertokens.pluginInterface.Storage;
+import io.supertokens.pluginInterface.useridmapping.UserIdMapping;
+
+public class StorageAndUserIdMappingForBulkImport extends StorageAndUserIdMapping {
+
+    public String userIdInQuestion;
+
+    public StorageAndUserIdMappingForBulkImport(Storage storage,
+                                                UserIdMapping userIdMapping, String userIdInQuestion) {
+        super(storage, userIdMapping);
+        this.userIdInQuestion = userIdInQuestion;
+    }
+}