From 0e62c9fa1f70b6778290649bb20f028820cd159f Mon Sep 17 00:00:00 2001 From: Ankit Tiwari Date: Wed, 14 Feb 2024 19:35:03 +0530 Subject: [PATCH 01/10] feat: Add bulk import queries --- .../supertokens/storage/postgresql/Start.java | 13 +++- .../postgresql/config/PostgreSQLConfig.java | 4 ++ .../postgresql/queries/BulkImportQueries.java | 65 +++++++++++++++++++ .../postgresql/queries/GeneralQueries.java | 7 ++ 4 files changed, 88 insertions(+), 1 deletion(-) create mode 100644 src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java diff --git a/src/main/java/io/supertokens/storage/postgresql/Start.java b/src/main/java/io/supertokens/storage/postgresql/Start.java index 91940ee0..293ea409 100644 --- a/src/main/java/io/supertokens/storage/postgresql/Start.java +++ b/src/main/java/io/supertokens/storage/postgresql/Start.java @@ -25,6 +25,8 @@ import io.supertokens.pluginInterface.authRecipe.AuthRecipeUserInfo; import io.supertokens.pluginInterface.authRecipe.LoginMethod; import io.supertokens.pluginInterface.authRecipe.sqlStorage.AuthRecipeSQLStorage; +import io.supertokens.pluginInterface.bulkimport.BulkImportStorage; +import io.supertokens.pluginInterface.bulkimport.BulkImportUser; import io.supertokens.pluginInterface.dashboard.DashboardSearchTags; import io.supertokens.pluginInterface.dashboard.DashboardSessionInfo; import io.supertokens.pluginInterface.dashboard.DashboardUser; @@ -109,7 +111,7 @@ public class Start implements SessionSQLStorage, EmailPasswordSQLStorage, EmailVerificationSQLStorage, ThirdPartySQLStorage, JWTRecipeSQLStorage, PasswordlessSQLStorage, UserMetadataSQLStorage, UserRolesSQLStorage, UserIdMappingStorage, - UserIdMappingSQLStorage, MultitenancyStorage, MultitenancySQLStorage, DashboardSQLStorage, TOTPSQLStorage, ActiveUsersStorage, AuthRecipeSQLStorage { + UserIdMappingSQLStorage, MultitenancyStorage, MultitenancySQLStorage, DashboardSQLStorage, TOTPSQLStorage, ActiveUsersStorage, AuthRecipeSQLStorage, BulkImportStorage { // these configs are protected from being modified / viewed by the dev using the SuperTokens // SaaS. If the core is not running in SuperTokens SaaS, this array has no effect. @@ -2990,4 +2992,13 @@ public UserIdMapping[] getUserIdMapping_Transaction(TransactionConnection con, A throw new StorageQueryException(e); } } + + @Override + public void addBulkImportUsers(AppIdentifier appIdentifier, ArrayList users) throws StorageQueryException { + try { + BulkImportQueries.insertBulkImportUsers(this, users); + } catch (SQLException e) { + throw new StorageQueryException(e); + } + } } diff --git a/src/main/java/io/supertokens/storage/postgresql/config/PostgreSQLConfig.java b/src/main/java/io/supertokens/storage/postgresql/config/PostgreSQLConfig.java index cbfac0ab..21f0af08 100644 --- a/src/main/java/io/supertokens/storage/postgresql/config/PostgreSQLConfig.java +++ b/src/main/java/io/supertokens/storage/postgresql/config/PostgreSQLConfig.java @@ -302,6 +302,10 @@ public String getTotpUsedCodesTable() { return addSchemaAndPrefixToTableName("totp_used_codes"); } + public String getBulkImportUsersTable() { + return addSchemaAndPrefixToTableName("bulk_import_users"); + } + private String addSchemaAndPrefixToTableName(String tableName) { return addSchemaToTableName(postgresql_table_names_prefix + tableName); } diff --git a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java new file mode 100644 index 00000000..5995ac2d --- /dev/null +++ b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java @@ -0,0 +1,65 @@ +/* + * 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.storage.postgresql.queries; + +import static io.supertokens.storage.postgresql.QueryExecutorTemplate.update; +import static io.supertokens.storage.postgresql.PreparedStatementValueSetter.NO_OP_SETTER; + +import java.sql.SQLException; +import java.util.ArrayList; + +import io.supertokens.pluginInterface.bulkimport.BulkImportUser; +import io.supertokens.pluginInterface.exceptions.StorageQueryException; +import io.supertokens.storage.postgresql.Start; +import io.supertokens.storage.postgresql.config.Config; + +public class BulkImportQueries { + static String getQueryToCreateBulkImportUsersTable(Start start) { + return "CREATE TABLE IF NOT EXISTS " + Config.getConfig(start).getBulkImportUsersTable() + " (" + + "id CHAR(36) PRIMARY KEY," + + "raw_data TEXT NOT NULL," + + "status VARCHAR(128) DEFAULT 'NEW'," + + "error_msg TEXT," + + "created_at BIGINT DEFAULT EXTRACT(EPOCH FROM CURRENT_TIMESTAMP)," + + "updated_at BIGINT DEFAULT EXTRACT(EPOCH FROM CURRENT_TIMESTAMP)" + + " );"; + } + + public static String getQueryToCreateStatusUpdatedAtIndex(Start start) { + return "CREATE INDEX IF NOT EXISTS bulk_import_users_status_updated_at_index ON " + + Config.getConfig(start).getBulkImportUsersTable() + " (status, updated_at)"; + } + + public static void insertBulkImportUsers(Start start, ArrayList users) + throws SQLException, StorageQueryException { + StringBuilder queryBuilder = new StringBuilder( + "INSERT INTO " + Config.getConfig(start).getBulkImportUsersTable() + " (id, raw_data) VALUES "); + for (BulkImportUser user : users) { + queryBuilder.append("('") + .append(user.id) + .append("', '") + .append(user.toString()) + .append("')"); + + if (user != users.get(users.size() - 1)) { + queryBuilder.append(","); + } + } + queryBuilder.append(";"); + update(start, queryBuilder.toString(), NO_OP_SETTER); + } +} diff --git a/src/main/java/io/supertokens/storage/postgresql/queries/GeneralQueries.java b/src/main/java/io/supertokens/storage/postgresql/queries/GeneralQueries.java index 729b8a51..1b9c4844 100644 --- a/src/main/java/io/supertokens/storage/postgresql/queries/GeneralQueries.java +++ b/src/main/java/io/supertokens/storage/postgresql/queries/GeneralQueries.java @@ -522,6 +522,13 @@ public static void createTablesIfNotExists(Start start) throws SQLException, Sto update(start, TOTPQueries.getQueryToCreateTenantIdIndexForUsedCodesTable(start), NO_OP_SETTER); } + if (!doesTableExists(start, Config.getConfig(start).getBulkImportUsersTable())) { + getInstance(start).addState(CREATING_NEW_TABLE, null); + update(start, BulkImportQueries.getQueryToCreateBulkImportUsersTable(start), NO_OP_SETTER); + // index: + update(start, BulkImportQueries.getQueryToCreateStatusUpdatedAtIndex(start), NO_OP_SETTER); + } + } catch (Exception e) { if (e.getMessage().contains("schema") && e.getMessage().contains("does not exist") && numberOfRetries < 1) { From cbdf6a4ab2a9520c3365b394322a75d36fe3a3d7 Mon Sep 17 00:00:00 2001 From: Ankit Tiwari Date: Thu, 15 Feb 2024 23:30:41 +0530 Subject: [PATCH 02/10] fix: PR changes --- .../supertokens/storage/postgresql/Start.java | 22 +++++- .../postgresql/queries/BulkImportQueries.java | 78 ++++++++++++++++--- .../postgresql/queries/GeneralQueries.java | 8 +- 3 files changed, 93 insertions(+), 15 deletions(-) diff --git a/src/main/java/io/supertokens/storage/postgresql/Start.java b/src/main/java/io/supertokens/storage/postgresql/Start.java index 293ea409..cf3f45a3 100644 --- a/src/main/java/io/supertokens/storage/postgresql/Start.java +++ b/src/main/java/io/supertokens/storage/postgresql/Start.java @@ -2994,9 +2994,27 @@ public UserIdMapping[] getUserIdMapping_Transaction(TransactionConnection con, A } @Override - public void addBulkImportUsers(AppIdentifier appIdentifier, ArrayList users) throws StorageQueryException { + public void addBulkImportUsers(AppIdentifier appIdentifier, ArrayList users) + throws StorageQueryException, + io.supertokens.pluginInterface.bulkimport.exceptions.DuplicateUserIdException { + try { + BulkImportQueries.insertBulkImportUsers(this, users); + } catch (SQLException e) { + if (e instanceof PSQLException) { + ServerErrorMessage serverErrorMessage = ((PSQLException) e).getServerErrorMessage(); + if (isPrimaryKeyError(serverErrorMessage, Config.getConfig(this).getBulkImportUsersTable())) { + throw new io.supertokens.pluginInterface.bulkimport.exceptions.DuplicateUserIdException(); + } + } + throw new StorageQueryException(e); + } + } + + @Override + public JsonObject[] getBulkImportUsers(AppIdentifier appIdentifier, @Nonnull Integer limit, @Nullable String status, + @Nullable String bulkImportUserId) throws StorageQueryException { try { - BulkImportQueries.insertBulkImportUsers(this, users); + return BulkImportQueries.getBulkImportUsers(this, limit, status, bulkImportUserId); } catch (SQLException e) { throw new StorageQueryException(e); } diff --git a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java index 5995ac2d..97318158 100644 --- a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java +++ b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java @@ -17,10 +17,16 @@ package io.supertokens.storage.postgresql.queries; import static io.supertokens.storage.postgresql.QueryExecutorTemplate.update; -import static io.supertokens.storage.postgresql.PreparedStatementValueSetter.NO_OP_SETTER; +import static io.supertokens.storage.postgresql.QueryExecutorTemplate.execute; import java.sql.SQLException; import java.util.ArrayList; +import java.util.List; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + +import com.google.gson.JsonObject; import io.supertokens.pluginInterface.bulkimport.BulkImportUser; import io.supertokens.pluginInterface.exceptions.StorageQueryException; @@ -48,18 +54,68 @@ public static void insertBulkImportUsers(Start start, ArrayList throws SQLException, StorageQueryException { StringBuilder queryBuilder = new StringBuilder( "INSERT INTO " + Config.getConfig(start).getBulkImportUsersTable() + " (id, raw_data) VALUES "); - for (BulkImportUser user : users) { - queryBuilder.append("('") - .append(user.id) - .append("', '") - .append(user.toString()) - .append("')"); - - if (user != users.get(users.size() - 1)) { + + int userCount = users.size(); + + for (int i = 0; i < userCount; i++) { + queryBuilder.append(" (?, ?)"); + + if (i < userCount - 1) { queryBuilder.append(","); } } - queryBuilder.append(";"); - update(start, queryBuilder.toString(), NO_OP_SETTER); + + update(start, queryBuilder.toString(), pst -> { + int parameterIndex = 1; + for (BulkImportUser user : users) { + pst.setString(parameterIndex++, user.id); + pst.setString(parameterIndex++, user.toString()); + } + }); + } + + public static JsonObject[] getBulkImportUsers(Start start, @Nonnull Integer limit, @Nullable String status, + @Nullable String bulkImportUserId) + throws SQLException, StorageQueryException { + + ArrayList bulkImportUsers = new ArrayList<>(); + String baseQuery = "SELECT * FROM " + Config.getConfig(start).getBulkImportUsersTable(); + + StringBuilder queryBuilder = new StringBuilder(baseQuery); + List parameters = new ArrayList<>(); + + if (status != null) { + queryBuilder.append(" WHERE status = ?"); + parameters.add(status); + } + + if (bulkImportUserId != null) { + queryBuilder.append(status != null ? " AND" : " WHERE") + .append(" id >= ?"); + parameters.add(bulkImportUserId); + } + + queryBuilder.append(" LIMIT ?"); + parameters.add(limit); + + String query = queryBuilder.toString(); + + return execute(start, query, pst -> { + for (int i = 0; i < parameters.size(); i++) { + pst.setObject(i + 1, parameters.get(i)); + } + }, result -> { + while (result.next()) { + JsonObject user = new JsonObject(); + user.addProperty("id", result.getString("id")); + user.addProperty("raw_data", result.getString("raw_data")); + user.addProperty("status", result.getString("status")); + user.addProperty("error_msg", result.getString("error_msg")); + user.addProperty("created_at", result.getLong("created_at")); + user.addProperty("updated_at", result.getLong("updated_at")); + bulkImportUsers.add(user); + } + return bulkImportUsers.toArray(new JsonObject[0]); + }); } } diff --git a/src/main/java/io/supertokens/storage/postgresql/queries/GeneralQueries.java b/src/main/java/io/supertokens/storage/postgresql/queries/GeneralQueries.java index 1b9c4844..934787ce 100644 --- a/src/main/java/io/supertokens/storage/postgresql/queries/GeneralQueries.java +++ b/src/main/java/io/supertokens/storage/postgresql/queries/GeneralQueries.java @@ -564,7 +564,10 @@ public static void deleteAllTables(Start start) throws SQLException, StorageQuer String DROP_QUERY = "DROP INDEX IF EXISTS all_auth_recipe_users_pagination_index"; update(start, DROP_QUERY, NO_OP_SETTER); } - + { + String DROP_QUERY = "DROP INDEX IF EXISTS bulk_import_users_status_updated_at_index"; + update(start, DROP_QUERY, NO_OP_SETTER); + } { String DROP_QUERY = "DROP TABLE IF EXISTS " + getConfig(start).getAppsTable() + "," @@ -598,7 +601,8 @@ public static void deleteAllTables(Start start) throws SQLException, StorageQuer + getConfig(start).getDashboardUsersTable() + "," + getConfig(start).getDashboardSessionsTable() + "," + getConfig(start).getTotpUsedCodesTable() + "," + getConfig(start).getTotpUserDevicesTable() + "," - + getConfig(start).getTotpUsersTable(); + + getConfig(start).getTotpUsersTable() + "," + + getConfig(start).getBulkImportUsersTable(); update(start, DROP_QUERY, NO_OP_SETTER); } } From 5f20244a24c5bb589ba094cec0f11e5b6953def0 Mon Sep 17 00:00:00 2001 From: Ankit Tiwari Date: Mon, 19 Feb 2024 20:44:24 +0530 Subject: [PATCH 03/10] fix: PR changes --- .../supertokens/storage/postgresql/Start.java | 4 +- .../postgresql/queries/BulkImportQueries.java | 62 ++++++++++++++----- .../postgresql/queries/GeneralQueries.java | 1 + 3 files changed, 49 insertions(+), 18 deletions(-) diff --git a/src/main/java/io/supertokens/storage/postgresql/Start.java b/src/main/java/io/supertokens/storage/postgresql/Start.java index cf3f45a3..c6e8d509 100644 --- a/src/main/java/io/supertokens/storage/postgresql/Start.java +++ b/src/main/java/io/supertokens/storage/postgresql/Start.java @@ -2994,7 +2994,7 @@ public UserIdMapping[] getUserIdMapping_Transaction(TransactionConnection con, A } @Override - public void addBulkImportUsers(AppIdentifier appIdentifier, ArrayList users) + public void addBulkImportUsers(AppIdentifier appIdentifier, List users) throws StorageQueryException, io.supertokens.pluginInterface.bulkimport.exceptions.DuplicateUserIdException { try { @@ -3011,7 +3011,7 @@ public void addBulkImportUsers(AppIdentifier appIdentifier, ArrayList getBulkImportUsers(AppIdentifier appIdentifier, @Nonnull Integer limit, @Nullable String status, @Nullable String bulkImportUserId) throws StorageQueryException { try { return BulkImportQueries.getBulkImportUsers(this, limit, status, bulkImportUserId); diff --git a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java index 97318158..45b514e3 100644 --- a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java +++ b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java @@ -19,6 +19,7 @@ import static io.supertokens.storage.postgresql.QueryExecutorTemplate.update; import static io.supertokens.storage.postgresql.QueryExecutorTemplate.execute; +import java.sql.ResultSet; import java.sql.SQLException; import java.util.ArrayList; import java.util.List; @@ -28,20 +29,28 @@ import com.google.gson.JsonObject; +import io.supertokens.pluginInterface.RowMapper; import io.supertokens.pluginInterface.bulkimport.BulkImportUser; import io.supertokens.pluginInterface.exceptions.StorageQueryException; import io.supertokens.storage.postgresql.Start; import io.supertokens.storage.postgresql.config.Config; +import io.supertokens.storage.postgresql.utils.Utils; public class BulkImportQueries { static String getQueryToCreateBulkImportUsersTable(Start start) { - return "CREATE TABLE IF NOT EXISTS " + Config.getConfig(start).getBulkImportUsersTable() + " (" + String schema = Config.getConfig(start).getTableSchema(); + String tableName = Config.getConfig(start).getBulkImportUsersTable(); + return "CREATE TABLE IF NOT EXISTS " + tableName + " (" + "id CHAR(36) PRIMARY KEY," + + "app_id VARCHAR(64) NOT NULL DEFAULT 'public'," + "raw_data TEXT NOT NULL," + "status VARCHAR(128) DEFAULT 'NEW'," + "error_msg TEXT," + "created_at BIGINT DEFAULT EXTRACT(EPOCH FROM CURRENT_TIMESTAMP)," - + "updated_at BIGINT DEFAULT EXTRACT(EPOCH FROM CURRENT_TIMESTAMP)" + + "updated_at BIGINT DEFAULT EXTRACT(EPOCH FROM CURRENT_TIMESTAMP)," + + "CONSTRAINT " + Utils.getConstraintName(schema, tableName, "app_id", "fkey") + " " + + "FOREIGN KEY(app_id) " + + "REFERENCES " + Config.getConfig(start).getAppsTable() + " (app_id) ON DELETE CASCADE" + " );"; } @@ -50,7 +59,12 @@ public static String getQueryToCreateStatusUpdatedAtIndex(Start start) { + Config.getConfig(start).getBulkImportUsersTable() + " (status, updated_at)"; } - public static void insertBulkImportUsers(Start start, ArrayList users) + public static String getQueryToCreateCreatedAtIndex(Start start) { + return "CREATE INDEX IF NOT EXISTS bulk_import_users_created_at_index ON " + + Config.getConfig(start).getBulkImportUsersTable() + " (created_at)"; + } + + public static void insertBulkImportUsers(Start start, List users) throws SQLException, StorageQueryException { StringBuilder queryBuilder = new StringBuilder( "INSERT INTO " + Config.getConfig(start).getBulkImportUsersTable() + " (id, raw_data) VALUES "); @@ -74,11 +88,10 @@ public static void insertBulkImportUsers(Start start, ArrayList }); } - public static JsonObject[] getBulkImportUsers(Start start, @Nonnull Integer limit, @Nullable String status, + public static List getBulkImportUsers(Start start, @Nonnull Integer limit, @Nullable String status, @Nullable String bulkImportUserId) throws SQLException, StorageQueryException { - ArrayList bulkImportUsers = new ArrayList<>(); String baseQuery = "SELECT * FROM " + Config.getConfig(start).getBulkImportUsersTable(); StringBuilder queryBuilder = new StringBuilder(baseQuery); @@ -91,11 +104,11 @@ public static JsonObject[] getBulkImportUsers(Start start, @Nonnull Integer limi if (bulkImportUserId != null) { queryBuilder.append(status != null ? " AND" : " WHERE") - .append(" id >= ?"); + .append(" id <= ?"); parameters.add(bulkImportUserId); } - queryBuilder.append(" LIMIT ?"); + queryBuilder.append(" ORDER BY created_at DESC LIMIT ?"); parameters.add(limit); String query = queryBuilder.toString(); @@ -105,17 +118,34 @@ public static JsonObject[] getBulkImportUsers(Start start, @Nonnull Integer limi pst.setObject(i + 1, parameters.get(i)); } }, result -> { + List bulkImportUsers = new ArrayList<>(); while (result.next()) { - JsonObject user = new JsonObject(); - user.addProperty("id", result.getString("id")); - user.addProperty("raw_data", result.getString("raw_data")); - user.addProperty("status", result.getString("status")); - user.addProperty("error_msg", result.getString("error_msg")); - user.addProperty("created_at", result.getLong("created_at")); - user.addProperty("updated_at", result.getLong("updated_at")); - bulkImportUsers.add(user); + bulkImportUsers.add(BulkImportUserJsonObjectRowMapper.getInstance().mapOrThrow(result)); } - return bulkImportUsers.toArray(new JsonObject[0]); + return bulkImportUsers; }); } + + private static class BulkImportUserJsonObjectRowMapper implements RowMapper { + private static final BulkImportUserJsonObjectRowMapper INSTANCE = new BulkImportUserJsonObjectRowMapper(); + + private BulkImportUserJsonObjectRowMapper() { + } + + private static BulkImportUserJsonObjectRowMapper getInstance() { + return INSTANCE; + } + + @Override + public JsonObject map(ResultSet result) throws Exception { + JsonObject user = new JsonObject(); + user.addProperty("id", result.getString("id")); + user.addProperty("raw_data", result.getString("raw_data")); + user.addProperty("status", result.getString("status")); + user.addProperty("error_msg", result.getString("error_msg")); + user.addProperty("created_at", result.getLong("created_at")); + user.addProperty("updated_at", result.getLong("updated_at")); + return user; + } + } } diff --git a/src/main/java/io/supertokens/storage/postgresql/queries/GeneralQueries.java b/src/main/java/io/supertokens/storage/postgresql/queries/GeneralQueries.java index 934787ce..9bb1e708 100644 --- a/src/main/java/io/supertokens/storage/postgresql/queries/GeneralQueries.java +++ b/src/main/java/io/supertokens/storage/postgresql/queries/GeneralQueries.java @@ -527,6 +527,7 @@ public static void createTablesIfNotExists(Start start) throws SQLException, Sto update(start, BulkImportQueries.getQueryToCreateBulkImportUsersTable(start), NO_OP_SETTER); // index: update(start, BulkImportQueries.getQueryToCreateStatusUpdatedAtIndex(start), NO_OP_SETTER); + update(start, BulkImportQueries.getQueryToCreateCreatedAtIndex(start), NO_OP_SETTER); } } catch (Exception e) { From dc76529803eeb7b39c906ee55ac45b5aa013c8ad Mon Sep 17 00:00:00 2001 From: Ankit Tiwari Date: Tue, 20 Feb 2024 12:06:24 +0530 Subject: [PATCH 04/10] fix: PR changes --- .../supertokens/storage/postgresql/Start.java | 7 ++++- .../postgresql/queries/BulkImportQueries.java | 31 ++++++++----------- 2 files changed, 19 insertions(+), 19 deletions(-) diff --git a/src/main/java/io/supertokens/storage/postgresql/Start.java b/src/main/java/io/supertokens/storage/postgresql/Start.java index c6e8d509..db9da19c 100644 --- a/src/main/java/io/supertokens/storage/postgresql/Start.java +++ b/src/main/java/io/supertokens/storage/postgresql/Start.java @@ -27,6 +27,7 @@ import io.supertokens.pluginInterface.authRecipe.sqlStorage.AuthRecipeSQLStorage; import io.supertokens.pluginInterface.bulkimport.BulkImportStorage; import io.supertokens.pluginInterface.bulkimport.BulkImportUser; +import io.supertokens.pluginInterface.bulkimport.BulkImportUserInfo; import io.supertokens.pluginInterface.dashboard.DashboardSearchTags; import io.supertokens.pluginInterface.dashboard.DashboardSessionInfo; import io.supertokens.pluginInterface.dashboard.DashboardUser; @@ -2996,6 +2997,7 @@ public UserIdMapping[] getUserIdMapping_Transaction(TransactionConnection con, A @Override public void addBulkImportUsers(AppIdentifier appIdentifier, List users) throws StorageQueryException, + TenantOrAppNotFoundException, io.supertokens.pluginInterface.bulkimport.exceptions.DuplicateUserIdException { try { BulkImportQueries.insertBulkImportUsers(this, users); @@ -3005,13 +3007,16 @@ public void addBulkImportUsers(AppIdentifier appIdentifier, List if (isPrimaryKeyError(serverErrorMessage, Config.getConfig(this).getBulkImportUsersTable())) { throw new io.supertokens.pluginInterface.bulkimport.exceptions.DuplicateUserIdException(); } + if (isForeignKeyConstraintError(serverErrorMessage, Config.getConfig(this).getBulkImportUsersTable(), "app_id")) { + throw new TenantOrAppNotFoundException(appIdentifier); + } } throw new StorageQueryException(e); } } @Override - public List getBulkImportUsers(AppIdentifier appIdentifier, @Nonnull Integer limit, @Nullable String status, + public List getBulkImportUsers(AppIdentifier appIdentifier, @Nonnull Integer limit, @Nullable BulkImportUserStatus status, @Nullable String bulkImportUserId) throws StorageQueryException { try { return BulkImportQueries.getBulkImportUsers(this, limit, status, bulkImportUserId); diff --git a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java index 45b514e3..01cc45c4 100644 --- a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java +++ b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java @@ -27,10 +27,10 @@ import javax.annotation.Nonnull; import javax.annotation.Nullable; -import com.google.gson.JsonObject; - import io.supertokens.pluginInterface.RowMapper; +import io.supertokens.pluginInterface.bulkimport.BulkImportStorage.BulkImportUserStatus; import io.supertokens.pluginInterface.bulkimport.BulkImportUser; +import io.supertokens.pluginInterface.bulkimport.BulkImportUserInfo; import io.supertokens.pluginInterface.exceptions.StorageQueryException; import io.supertokens.storage.postgresql.Start; import io.supertokens.storage.postgresql.config.Config; @@ -88,7 +88,7 @@ public static void insertBulkImportUsers(Start start, List users }); } - public static List getBulkImportUsers(Start start, @Nonnull Integer limit, @Nullable String status, + public static List getBulkImportUsers(Start start, @Nonnull Integer limit, @Nullable BulkImportUserStatus status, @Nullable String bulkImportUserId) throws SQLException, StorageQueryException { @@ -118,34 +118,29 @@ public static List getBulkImportUsers(Start start, @Nonnull Integer pst.setObject(i + 1, parameters.get(i)); } }, result -> { - List bulkImportUsers = new ArrayList<>(); + List bulkImportUsers = new ArrayList<>(); while (result.next()) { - bulkImportUsers.add(BulkImportUserJsonObjectRowMapper.getInstance().mapOrThrow(result)); + bulkImportUsers.add(BulkImportUserInfoRowMapper.getInstance().mapOrThrow(result)); } return bulkImportUsers; }); } - private static class BulkImportUserJsonObjectRowMapper implements RowMapper { - private static final BulkImportUserJsonObjectRowMapper INSTANCE = new BulkImportUserJsonObjectRowMapper(); + private static class BulkImportUserInfoRowMapper implements RowMapper { + private static final BulkImportUserInfoRowMapper INSTANCE = new BulkImportUserInfoRowMapper(); - private BulkImportUserJsonObjectRowMapper() { + private BulkImportUserInfoRowMapper() { } - private static BulkImportUserJsonObjectRowMapper getInstance() { + private static BulkImportUserInfoRowMapper getInstance() { return INSTANCE; } @Override - public JsonObject map(ResultSet result) throws Exception { - JsonObject user = new JsonObject(); - user.addProperty("id", result.getString("id")); - user.addProperty("raw_data", result.getString("raw_data")); - user.addProperty("status", result.getString("status")); - user.addProperty("error_msg", result.getString("error_msg")); - user.addProperty("created_at", result.getLong("created_at")); - user.addProperty("updated_at", result.getLong("updated_at")); - return user; + public BulkImportUserInfo map(ResultSet result) throws Exception { + return new BulkImportUserInfo(result.getString("id"), result.getString("raw_data"), + BulkImportUserStatus.valueOf(result.getString("status")), + result.getLong("created_at"), result.getLong("updated_at")); } } } From 2417ef2f720ae7f460dfc9b9c246a976df47cfa9 Mon Sep 17 00:00:00 2001 From: Ankit Tiwari Date: Tue, 20 Feb 2024 17:50:00 +0530 Subject: [PATCH 05/10] fix: PR changes --- .../supertokens/storage/postgresql/Start.java | 6 ++-- .../postgresql/queries/BulkImportQueries.java | 32 ++++++++++++------- 2 files changed, 23 insertions(+), 15 deletions(-) diff --git a/src/main/java/io/supertokens/storage/postgresql/Start.java b/src/main/java/io/supertokens/storage/postgresql/Start.java index db9da19c..16d3f0e2 100644 --- a/src/main/java/io/supertokens/storage/postgresql/Start.java +++ b/src/main/java/io/supertokens/storage/postgresql/Start.java @@ -3000,7 +3000,7 @@ public void addBulkImportUsers(AppIdentifier appIdentifier, List TenantOrAppNotFoundException, io.supertokens.pluginInterface.bulkimport.exceptions.DuplicateUserIdException { try { - BulkImportQueries.insertBulkImportUsers(this, users); + BulkImportQueries.insertBulkImportUsers(this, appIdentifier, users); } catch (SQLException e) { if (e instanceof PSQLException) { ServerErrorMessage serverErrorMessage = ((PSQLException) e).getServerErrorMessage(); @@ -3017,9 +3017,9 @@ public void addBulkImportUsers(AppIdentifier appIdentifier, List @Override public List getBulkImportUsers(AppIdentifier appIdentifier, @Nonnull Integer limit, @Nullable BulkImportUserStatus status, - @Nullable String bulkImportUserId) throws StorageQueryException { + @Nullable String bulkImportUserId, @Nullable Long createdAt) throws StorageQueryException { try { - return BulkImportQueries.getBulkImportUsers(this, limit, status, bulkImportUserId); + return BulkImportQueries.getBulkImportUsers(this, appIdentifier, limit, status, bulkImportUserId, createdAt); } catch (SQLException e) { throw new StorageQueryException(e); } diff --git a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java index 01cc45c4..289514dd 100644 --- a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java +++ b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java @@ -32,6 +32,7 @@ import io.supertokens.pluginInterface.bulkimport.BulkImportUser; import io.supertokens.pluginInterface.bulkimport.BulkImportUserInfo; import io.supertokens.pluginInterface.exceptions.StorageQueryException; +import io.supertokens.pluginInterface.multitenancy.AppIdentifier; import io.supertokens.storage.postgresql.Start; import io.supertokens.storage.postgresql.config.Config; import io.supertokens.storage.postgresql.utils.Utils; @@ -41,13 +42,15 @@ static String getQueryToCreateBulkImportUsersTable(Start start) { String schema = Config.getConfig(start).getTableSchema(); String tableName = Config.getConfig(start).getBulkImportUsersTable(); return "CREATE TABLE IF NOT EXISTS " + tableName + " (" - + "id CHAR(36) PRIMARY KEY," + + "id CHAR(36)," + "app_id VARCHAR(64) NOT NULL DEFAULT 'public'," + "raw_data TEXT NOT NULL," + "status VARCHAR(128) DEFAULT 'NEW'," + "error_msg TEXT," + "created_at BIGINT DEFAULT EXTRACT(EPOCH FROM CURRENT_TIMESTAMP)," + "updated_at BIGINT DEFAULT EXTRACT(EPOCH FROM CURRENT_TIMESTAMP)," + + "CONSTRAINT " + Utils.getConstraintName(schema, tableName, null, "pkey") + + " PRIMARY KEY(app_id, id)," + "CONSTRAINT " + Utils.getConstraintName(schema, tableName, "app_id", "fkey") + " " + "FOREIGN KEY(app_id) " + "REFERENCES " + Config.getConfig(start).getAppsTable() + " (app_id) ON DELETE CASCADE" @@ -56,23 +59,23 @@ static String getQueryToCreateBulkImportUsersTable(Start start) { public static String getQueryToCreateStatusUpdatedAtIndex(Start start) { return "CREATE INDEX IF NOT EXISTS bulk_import_users_status_updated_at_index ON " - + Config.getConfig(start).getBulkImportUsersTable() + " (status, updated_at)"; + + Config.getConfig(start).getBulkImportUsersTable() + " (app_id, status, updated_at)"; } public static String getQueryToCreateCreatedAtIndex(Start start) { return "CREATE INDEX IF NOT EXISTS bulk_import_users_created_at_index ON " - + Config.getConfig(start).getBulkImportUsersTable() + " (created_at)"; + + Config.getConfig(start).getBulkImportUsersTable() + " (app_id, created_at)"; } - public static void insertBulkImportUsers(Start start, List users) + public static void insertBulkImportUsers(Start start, AppIdentifier appIdentifier, List users) throws SQLException, StorageQueryException { StringBuilder queryBuilder = new StringBuilder( - "INSERT INTO " + Config.getConfig(start).getBulkImportUsersTable() + " (id, raw_data) VALUES "); + "INSERT INTO " + Config.getConfig(start).getBulkImportUsersTable() + " (id, app_id, raw_data) VALUES "); int userCount = users.size(); for (int i = 0; i < userCount; i++) { - queryBuilder.append(" (?, ?)"); + queryBuilder.append(" (?, ?, ?)"); if (i < userCount - 1) { queryBuilder.append(","); @@ -83,13 +86,14 @@ public static void insertBulkImportUsers(Start start, List users int parameterIndex = 1; for (BulkImportUser user : users) { pst.setString(parameterIndex++, user.id); + pst.setString(parameterIndex++, appIdentifier.getAppId()); pst.setString(parameterIndex++, user.toString()); } }); } - public static List getBulkImportUsers(Start start, @Nonnull Integer limit, @Nullable BulkImportUserStatus status, - @Nullable String bulkImportUserId) + public static List getBulkImportUsers(Start start, AppIdentifier appIdentifier, @Nonnull Integer limit, @Nullable BulkImportUserStatus status, + @Nullable String bulkImportUserId, @Nullable Long createdAt) throws SQLException, StorageQueryException { String baseQuery = "SELECT * FROM " + Config.getConfig(start).getBulkImportUsersTable(); @@ -98,13 +102,17 @@ public static List getBulkImportUsers(Start start, @Nonnull List parameters = new ArrayList<>(); if (status != null) { - queryBuilder.append(" WHERE status = ?"); + queryBuilder.append(" WHERE app_id = ? status = ?"); + parameters.add(appIdentifier.getAppId()); parameters.add(status); } - if (bulkImportUserId != null) { - queryBuilder.append(status != null ? " AND" : " WHERE") - .append(" id <= ?"); + if (bulkImportUserId != null && createdAt != null) { + queryBuilder + .append(status != null ? " AND" : " WHERE") + .append(" created_at < ? OR (created_at = ? AND id <= ?)"); + parameters.add(createdAt); + parameters.add(createdAt); parameters.add(bulkImportUserId); } From fd1089fbf8dce18649f148dbc1a63949861945fc Mon Sep 17 00:00:00 2001 From: Ankit Tiwari Date: Tue, 20 Feb 2024 18:02:48 +0530 Subject: [PATCH 06/10] fix: PR changes --- .../storage/postgresql/queries/BulkImportQueries.java | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java index 289514dd..4f25a4ed 100644 --- a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java +++ b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java @@ -101,16 +101,17 @@ public static List getBulkImportUsers(Start start, AppIdenti StringBuilder queryBuilder = new StringBuilder(baseQuery); List parameters = new ArrayList<>(); + queryBuilder.append(" WHERE app_id = ?"); + parameters.add(appIdentifier.getAppId()); + if (status != null) { - queryBuilder.append(" WHERE app_id = ? status = ?"); - parameters.add(appIdentifier.getAppId()); - parameters.add(status); + queryBuilder.append(" AND status = ?"); + parameters.add(status.toString()); } if (bulkImportUserId != null && createdAt != null) { queryBuilder - .append(status != null ? " AND" : " WHERE") - .append(" created_at < ? OR (created_at = ? AND id <= ?)"); + .append(" AND created_at < ? OR (created_at = ? AND id <= ?)"); parameters.add(createdAt); parameters.add(createdAt); parameters.add(bulkImportUserId); From adc0df90b97a4e20b93f8b9c27bb8f589a8f9932 Mon Sep 17 00:00:00 2001 From: Ankit Tiwari Date: Thu, 22 Feb 2024 12:09:56 +0530 Subject: [PATCH 07/10] fix: PR changes --- .../storage/postgresql/queries/BulkImportQueries.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java index 4f25a4ed..51b67039 100644 --- a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java +++ b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java @@ -117,7 +117,7 @@ public static List getBulkImportUsers(Start start, AppIdenti parameters.add(bulkImportUserId); } - queryBuilder.append(" ORDER BY created_at DESC LIMIT ?"); + queryBuilder.append(" ORDER BY created_at DESC, id DESC LIMIT ?"); parameters.add(limit); String query = queryBuilder.toString(); From b41a916ea41a919847f41acf9d1df364c43f5f50 Mon Sep 17 00:00:00 2001 From: Ankit Tiwari Date: Fri, 23 Feb 2024 18:43:40 +0530 Subject: [PATCH 08/10] fix: PR changes --- .../supertokens/storage/postgresql/Start.java | 16 +++++++-- .../postgresql/queries/BulkImportQueries.java | 34 +++++++++++++++++++ 2 files changed, 48 insertions(+), 2 deletions(-) diff --git a/src/main/java/io/supertokens/storage/postgresql/Start.java b/src/main/java/io/supertokens/storage/postgresql/Start.java index 16d3f0e2..f638b851 100644 --- a/src/main/java/io/supertokens/storage/postgresql/Start.java +++ b/src/main/java/io/supertokens/storage/postgresql/Start.java @@ -25,9 +25,9 @@ import io.supertokens.pluginInterface.authRecipe.AuthRecipeUserInfo; import io.supertokens.pluginInterface.authRecipe.LoginMethod; import io.supertokens.pluginInterface.authRecipe.sqlStorage.AuthRecipeSQLStorage; -import io.supertokens.pluginInterface.bulkimport.BulkImportStorage; import io.supertokens.pluginInterface.bulkimport.BulkImportUser; import io.supertokens.pluginInterface.bulkimport.BulkImportUserInfo; +import io.supertokens.pluginInterface.bulkimport.sqlStorage.BulkImportSQLStorage; import io.supertokens.pluginInterface.dashboard.DashboardSearchTags; import io.supertokens.pluginInterface.dashboard.DashboardSessionInfo; import io.supertokens.pluginInterface.dashboard.DashboardUser; @@ -112,7 +112,7 @@ public class Start implements SessionSQLStorage, EmailPasswordSQLStorage, EmailVerificationSQLStorage, ThirdPartySQLStorage, JWTRecipeSQLStorage, PasswordlessSQLStorage, UserMetadataSQLStorage, UserRolesSQLStorage, UserIdMappingStorage, - UserIdMappingSQLStorage, MultitenancyStorage, MultitenancySQLStorage, DashboardSQLStorage, TOTPSQLStorage, ActiveUsersStorage, AuthRecipeSQLStorage, BulkImportStorage { + UserIdMappingSQLStorage, MultitenancyStorage, MultitenancySQLStorage, DashboardSQLStorage, TOTPSQLStorage, ActiveUsersStorage, AuthRecipeSQLStorage, BulkImportSQLStorage { // these configs are protected from being modified / viewed by the dev using the SuperTokens // SaaS. If the core is not running in SuperTokens SaaS, this array has no effect. @@ -3024,4 +3024,16 @@ public List getBulkImportUsers(AppIdentifier appIdentifier, throw new StorageQueryException(e); } } + + @Override + public void updateBulkImportUserStatus_Transaction(AppIdentifier appIdentifier, TransactionConnection con, @Nonnull String[] bulkImportUserIds, @Nonnull BulkImportUserStatus status) + throws StorageQueryException { + Connection sqlCon = (Connection) con.getConnection(); + try { + BulkImportQueries.updateBulkImportUserStatus_Transaction(this, sqlCon, appIdentifier, bulkImportUserIds, status); + } catch (SQLException e) { + throw new StorageQueryException(e); + } + } + } diff --git a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java index 51b67039..688983e4 100644 --- a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java +++ b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java @@ -19,6 +19,7 @@ import static io.supertokens.storage.postgresql.QueryExecutorTemplate.update; import static io.supertokens.storage.postgresql.QueryExecutorTemplate.execute; +import java.sql.Connection; import java.sql.ResultSet; import java.sql.SQLException; import java.util.ArrayList; @@ -92,6 +93,39 @@ public static void insertBulkImportUsers(Start start, AppIdentifier appIdentifie }); } + public static void updateBulkImportUserStatus_Transaction(Start start, Connection con, AppIdentifier appIdentifier, @Nonnull String[] bulkImportUserIds, @Nonnull BulkImportUserStatus status) + throws SQLException, StorageQueryException { + if (bulkImportUserIds.length == 0) { + return; + } + + String baseQuery = "UPDATE " + Config.getConfig(start).getBulkImportUsersTable() + " SET status = ?, updated_at = EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) WHERE app_id = ?"; + StringBuilder queryBuilder = new StringBuilder(baseQuery); + + List parameters = new ArrayList<>(); + + parameters.add(status.toString()); + parameters.add(appIdentifier.getAppId()); + + queryBuilder.append(" AND id IN ("); + for (int i = 0; i < bulkImportUserIds.length; i++) { + if (i != 0) { + queryBuilder.append(", "); + } + queryBuilder.append("?"); + parameters.add(bulkImportUserIds[i]); + } + queryBuilder.append(")"); + + String query = queryBuilder.toString(); + + update(con, query, pst -> { + for (int i = 0; i < parameters.size(); i++) { + pst.setObject(i + 1, parameters.get(i)); + } + }); + } + public static List getBulkImportUsers(Start start, AppIdentifier appIdentifier, @Nonnull Integer limit, @Nullable BulkImportUserStatus status, @Nullable String bulkImportUserId, @Nullable Long createdAt) throws SQLException, StorageQueryException { From 6a447b4253165dea56209f52e5700ecc49fc2df2 Mon Sep 17 00:00:00 2001 From: Ankit Tiwari Date: Tue, 27 Feb 2024 15:10:06 +0530 Subject: [PATCH 09/10] fix: PR changes --- .../storage/postgresql/queries/GeneralQueries.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/main/java/io/supertokens/storage/postgresql/queries/GeneralQueries.java b/src/main/java/io/supertokens/storage/postgresql/queries/GeneralQueries.java index aa0c5512..5e8574a2 100644 --- a/src/main/java/io/supertokens/storage/postgresql/queries/GeneralQueries.java +++ b/src/main/java/io/supertokens/storage/postgresql/queries/GeneralQueries.java @@ -588,6 +588,10 @@ public static void deleteAllTables(Start start) throws SQLException, StorageQuer String DROP_QUERY = "DROP INDEX IF EXISTS bulk_import_users_status_updated_at_index"; update(start, DROP_QUERY, NO_OP_SETTER); } + { + String DROP_QUERY = "DROP INDEX IF EXISTS bulk_import_users_created_at_index"; + update(start, DROP_QUERY, NO_OP_SETTER); + } { String DROP_QUERY = "DROP TABLE IF EXISTS " + getConfig(start).getAppsTable() + "," From 9d8b283df343451b888129b99290f2f325398e33 Mon Sep 17 00:00:00 2001 From: Ankit Tiwari Date: Wed, 28 Feb 2024 10:17:19 +0530 Subject: [PATCH 10/10] fix: PR changes --- .../supertokens/storage/postgresql/Start.java | 3 +-- .../postgresql/queries/BulkImportQueries.java | 19 +++++++++---------- 2 files changed, 10 insertions(+), 12 deletions(-) diff --git a/src/main/java/io/supertokens/storage/postgresql/Start.java b/src/main/java/io/supertokens/storage/postgresql/Start.java index 52d8fa28..ce0d27f2 100644 --- a/src/main/java/io/supertokens/storage/postgresql/Start.java +++ b/src/main/java/io/supertokens/storage/postgresql/Start.java @@ -26,7 +26,6 @@ import io.supertokens.pluginInterface.authRecipe.LoginMethod; import io.supertokens.pluginInterface.authRecipe.sqlStorage.AuthRecipeSQLStorage; import io.supertokens.pluginInterface.bulkimport.BulkImportUser; -import io.supertokens.pluginInterface.bulkimport.BulkImportUserInfo; import io.supertokens.pluginInterface.bulkimport.sqlStorage.BulkImportSQLStorage; import io.supertokens.pluginInterface.dashboard.DashboardSearchTags; import io.supertokens.pluginInterface.dashboard.DashboardSessionInfo; @@ -3065,7 +3064,7 @@ public void addBulkImportUsers(AppIdentifier appIdentifier, List } @Override - public List getBulkImportUsers(AppIdentifier appIdentifier, @Nonnull Integer limit, @Nullable BulkImportUserStatus status, + public List getBulkImportUsers(AppIdentifier appIdentifier, @Nonnull Integer limit, @Nullable BulkImportUserStatus status, @Nullable String bulkImportUserId, @Nullable Long createdAt) throws StorageQueryException { try { return BulkImportQueries.getBulkImportUsers(this, appIdentifier, limit, status, bulkImportUserId, createdAt); diff --git a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java index 688983e4..4c77b1a1 100644 --- a/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java +++ b/src/main/java/io/supertokens/storage/postgresql/queries/BulkImportQueries.java @@ -31,7 +31,6 @@ import io.supertokens.pluginInterface.RowMapper; import io.supertokens.pluginInterface.bulkimport.BulkImportStorage.BulkImportUserStatus; import io.supertokens.pluginInterface.bulkimport.BulkImportUser; -import io.supertokens.pluginInterface.bulkimport.BulkImportUserInfo; import io.supertokens.pluginInterface.exceptions.StorageQueryException; import io.supertokens.pluginInterface.multitenancy.AppIdentifier; import io.supertokens.storage.postgresql.Start; @@ -126,7 +125,7 @@ public static void updateBulkImportUserStatus_Transaction(Start start, Connectio }); } - public static List getBulkImportUsers(Start start, AppIdentifier appIdentifier, @Nonnull Integer limit, @Nullable BulkImportUserStatus status, + public static List getBulkImportUsers(Start start, AppIdentifier appIdentifier, @Nonnull Integer limit, @Nullable BulkImportUserStatus status, @Nullable String bulkImportUserId, @Nullable Long createdAt) throws SQLException, StorageQueryException { @@ -161,27 +160,27 @@ public static List getBulkImportUsers(Start start, AppIdenti pst.setObject(i + 1, parameters.get(i)); } }, result -> { - List bulkImportUsers = new ArrayList<>(); + List bulkImportUsers = new ArrayList<>(); while (result.next()) { - bulkImportUsers.add(BulkImportUserInfoRowMapper.getInstance().mapOrThrow(result)); + bulkImportUsers.add(BulkImportUserRowMapper.getInstance().mapOrThrow(result)); } return bulkImportUsers; }); } - private static class BulkImportUserInfoRowMapper implements RowMapper { - private static final BulkImportUserInfoRowMapper INSTANCE = new BulkImportUserInfoRowMapper(); + private static class BulkImportUserRowMapper implements RowMapper { + private static final BulkImportUserRowMapper INSTANCE = new BulkImportUserRowMapper(); - private BulkImportUserInfoRowMapper() { + private BulkImportUserRowMapper() { } - private static BulkImportUserInfoRowMapper getInstance() { + private static BulkImportUserRowMapper getInstance() { return INSTANCE; } @Override - public BulkImportUserInfo map(ResultSet result) throws Exception { - return new BulkImportUserInfo(result.getString("id"), result.getString("raw_data"), + public BulkImportUser map(ResultSet result) throws Exception { + return BulkImportUser.fromDBJson(result.getString("id"), result.getString("raw_data"), BulkImportUserStatus.valueOf(result.getString("status")), result.getLong("created_at"), result.getLong("updated_at")); }