Refactor storeMessages to improve batch handling and error recovery, add per-message fallback, and enhance logging.
This commit is contained in:
@@ -40,35 +40,124 @@ public class ChatLogQueries {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
public static @NotNull CompletableFuture<Boolean> storeMessages(HashMap<UUID, List<ChatLog>> chatMessages) {
|
public static @NotNull CompletableFuture<Boolean> storeMessages(
|
||||||
String insertQuery = "INSERT INTO chat_log (uuid, time_stamp, server, type, channel, receiver, chat_message, mini_message, blocked) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)";
|
HashMap<UUID, List<ChatLog>> chatMessages
|
||||||
return CompletableFuture.supplyAsync(() -> {
|
) {
|
||||||
try (Connection connection = DatabaseConnection.createTransactionConnection()) {
|
String insertQuery = """
|
||||||
PreparedStatement preparedStatement = connection.prepareStatement(insertQuery);
|
INSERT INTO chat_log
|
||||||
for (List<ChatLog> chatLogList : chatMessages.values()) {
|
(uuid, time_stamp, server, type, channel, receiver,
|
||||||
for (ChatLog chatLog : chatLogList) {
|
chat_message, mini_message, blocked)
|
||||||
chatLog.prepareStatement(preparedStatement);
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
preparedStatement.addBatch();
|
""";
|
||||||
}
|
|
||||||
}
|
return CompletableFuture.supplyAsync(() -> {
|
||||||
int[] updatedRowsCount = preparedStatement.executeBatch();
|
try (Connection connection = DatabaseConnection.createTransactionConnection();
|
||||||
boolean isSuccess = Arrays.stream(updatedRowsCount).allMatch(i -> i >= 0);
|
PreparedStatement preparedStatement = connection.prepareStatement(insertQuery)) {
|
||||||
|
|
||||||
|
try {
|
||||||
|
addMessagesToBatch(chatMessages, preparedStatement);
|
||||||
|
|
||||||
|
int[] updatedRowsCount = preparedStatement.executeBatch();
|
||||||
|
|
||||||
|
boolean isSuccess = Arrays.stream(updatedRowsCount)
|
||||||
|
.allMatch(i -> i >= 0 || i == Statement.SUCCESS_NO_INFO);
|
||||||
|
|
||||||
|
if (isSuccess) {
|
||||||
|
connection.commit();
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
if (isSuccess) {
|
|
||||||
connection.commit();
|
|
||||||
return true;
|
|
||||||
} else {
|
|
||||||
connection.rollback();
|
connection.rollback();
|
||||||
ALogger.warn("Failed to store messages");
|
} catch (BatchUpdateException exception) {
|
||||||
return false;
|
connection.rollback();
|
||||||
|
|
||||||
|
ALogger.error(
|
||||||
|
"Failed to store chat batch. Trying messages individually.",
|
||||||
|
exception
|
||||||
|
);
|
||||||
}
|
}
|
||||||
} catch (SQLException sqlException) {
|
|
||||||
ALogger.error("Failed to store chat messages", sqlException);
|
return storeMessagesIndividually(
|
||||||
throw new CompletionException("Failed to store chat messages", sqlException);
|
connection,
|
||||||
|
preparedStatement,
|
||||||
|
chatMessages
|
||||||
|
);
|
||||||
|
|
||||||
|
} catch (SQLException exception) {
|
||||||
|
ALogger.error("Failed to store chat messages", exception);
|
||||||
|
throw new CompletionException(
|
||||||
|
"Failed to store chat messages",
|
||||||
|
exception
|
||||||
|
);
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private static void addMessagesToBatch(
|
||||||
|
HashMap<UUID, List<ChatLog>> chatMessages,
|
||||||
|
PreparedStatement preparedStatement
|
||||||
|
) throws SQLException {
|
||||||
|
for (List<ChatLog> chatLogList : chatMessages.values()) {
|
||||||
|
for (ChatLog chatLog : chatLogList) {
|
||||||
|
chatLog.prepareStatement(preparedStatement);
|
||||||
|
preparedStatement.addBatch();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static boolean storeMessagesIndividually(
|
||||||
|
Connection connection,
|
||||||
|
PreparedStatement preparedStatement,
|
||||||
|
HashMap<UUID, List<ChatLog>> chatMessages
|
||||||
|
) throws SQLException {
|
||||||
|
boolean storedAny = false;
|
||||||
|
|
||||||
|
for (List<ChatLog> chatLogList : chatMessages.values()) {
|
||||||
|
Iterator<ChatLog> iterator = chatLogList.iterator();
|
||||||
|
|
||||||
|
while (iterator.hasNext()) {
|
||||||
|
ChatLog chatLog = iterator.next();
|
||||||
|
|
||||||
|
if (storeMessage(preparedStatement, chatLog)) {
|
||||||
|
storedAny = true;
|
||||||
|
} else {
|
||||||
|
iterator.remove();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
connection.commit();
|
||||||
|
|
||||||
|
return storedAny;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static boolean storeMessage(
|
||||||
|
PreparedStatement preparedStatement,
|
||||||
|
ChatLog chatLog
|
||||||
|
) {
|
||||||
|
try {
|
||||||
|
preparedStatement.clearParameters();
|
||||||
|
|
||||||
|
chatLog.prepareStatement(preparedStatement);
|
||||||
|
preparedStatement.executeUpdate();
|
||||||
|
|
||||||
|
return true;
|
||||||
|
} catch (SQLException exception) {
|
||||||
|
logInvalidMessage(chatLog, exception);
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void logInvalidMessage(
|
||||||
|
ChatLog chatLog,
|
||||||
|
SQLException exception
|
||||||
|
) {
|
||||||
|
ALogger.error(
|
||||||
|
"Failed to store chat message. Removing offending message: " + chatLog,
|
||||||
|
exception
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
public static @NotNull CompletableFuture<List<ChatLog>> retrieveMessages(ChatLogHandler chatLogHandler, UUID uuid, Duration duration, String server) {
|
public static @NotNull CompletableFuture<List<ChatLog>> retrieveMessages(ChatLogHandler chatLogHandler, UUID uuid, Duration duration, String server) {
|
||||||
String query = "SELECT * FROM chat_log WHERE uuid = ? AND time_stamp > ? AND server = ? AND type = 'public'";
|
String query = "SELECT * FROM chat_log WHERE uuid = ? AND time_stamp > ? AND server = ? AND type = 'public'";
|
||||||
return CompletableFuture.supplyAsync(() -> {
|
return CompletableFuture.supplyAsync(() -> {
|
||||||
|
|||||||
@@ -43,4 +43,19 @@ public class ChatLog implements BatchInsertable {
|
|||||||
);
|
);
|
||||||
preparedStatement.setInt(9, blocked ? 1 : 0);
|
preparedStatement.setInt(9, blocked ? 1 : 0);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String toString() {
|
||||||
|
return "ChatLog{" +
|
||||||
|
"uuid=" + uuid +
|
||||||
|
", timestamp=" + timestamp +
|
||||||
|
", server='" + server + '\'' +
|
||||||
|
", type=" + type.name() +
|
||||||
|
", channel='" + channel + '\'' +
|
||||||
|
", receiver='" + receiver + '\'' +
|
||||||
|
", message='" + message + '\'' +
|
||||||
|
", miniMessage=" + (miniMessage == null ? null : GsonComponentSerializer.gson().serialize(miniMessage)) +
|
||||||
|
", blocked=" + blocked +
|
||||||
|
'}';
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user