MOSIP-45087 - Updated the regex for fetching OTP from SMTP and added retry mechanism for smtp websocket connection

Signed-off-by: Mohanachandran S <mohanachandran.s@technoforte.co.in>

Author: mohanachandran-s

apitest-commons/src/main/java/io/mosip/testrig/apirig/testrunner/OTPListener.java

@@ -4,8 +4,13 @@
 import java.net.http.HttpClient;
 import java.net.http.WebSocket;
 import java.net.http.WebSocket.Listener;
+import java.nio.ByteBuffer;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
@@ -25,6 +30,10 @@ public class OTPListener {
 
 	public static volatile boolean bTerminate = false;
 
+	private static final AtomicBoolean reconnecting = new AtomicBoolean(false);
+
+	public static final ConcurrentLinkedQueue<String> reconnectEvents = new ConcurrentLinkedQueue<>();
+
 	public OTPListener() {
 		if (ConfigManager.IsDebugEnabled()) {
 			logger.setLevel(Level.ALL);
@@ -75,13 +84,36 @@ private static class WebSocketClient implements WebSocket.Listener {
 
 		private final ObjectMapper objectMapper = new ObjectMapper();
 		private final StringBuilder messageBuffer = new StringBuilder();
+		private ScheduledExecutorService pingScheduler;
 
 		@Override
 		public void onOpen(WebSocket webSocket) {
 			logger.info("OTP WebSocket connection opened.");
+			startPing(webSocket);
 			Listener.super.onOpen(webSocket);
 		}
 
+		private void startPing(WebSocket webSocket) {
+			pingScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+				Thread t = new Thread(r, "ws-ping");
+				t.setDaemon(true);
+				return t;
+			});
+			pingScheduler.scheduleAtFixedRate(() -> {
+				if (bTerminate) {
+					pingScheduler.shutdown();
+					return;
+				}
+				webSocket.sendPing(ByteBuffer.wrap("ping".getBytes()));
+			}, 30, 30, TimeUnit.SECONDS);
+		}
+
+		private void stopPing() {
+			if (pingScheduler != null && !pingScheduler.isShutdown()) {
+				pingScheduler.shutdownNow();
+			}
+		}
+
 		@Override
 		public synchronized CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
 
@@ -172,13 +204,70 @@ public synchronized CompletionStage<?> onText(WebSocket webSocket, CharSequence
 		@Override
 		public void onError(WebSocket webSocket, Throwable error) {
 			logger.error("WebSocket error occurred", error);
+			stopPing();
+			scheduleReconnect("error");
 			Listener.super.onError(webSocket, error);
 		}
 
 		@Override
 		public CompletionStage<?> onClose(WebSocket webSocket, int statusCode, String reason) {
 			logger.warn("WebSocket closed. Status: " + statusCode + " Reason: " + reason);
+			stopPing();
+			if (statusCode != 1000) {
+				scheduleReconnect("abnormal close (status=" + statusCode + ")");
+			}
 			return Listener.super.onClose(webSocket, statusCode, reason);
 		}
+
+		private void scheduleReconnect(String trigger) {
+			if (bTerminate) return;
+			if (!reconnecting.compareAndSet(false, true)) {
+				logger.info("Reconnect already in progress, ignoring trigger: " + trigger);
+				return;
+			}
+			logger.info("Scheduling WebSocket reconnect after " + trigger);
+			Thread t = new Thread(() -> {
+				try {
+					int attempt = 0;
+					while (!bTerminate) {
+						attempt++;
+						long delayMs = Math.min(3000L * attempt, 60000L);
+						try {
+							Thread.sleep(delayMs);
+						} catch (InterruptedException e) {
+							Thread.currentThread().interrupt();
+							return;
+						}
+						if (bTerminate) return;
+						try {
+							URI iamUri = URI.create(ConfigManager.getIAMUrl());
+							String host = iamUri.getHost();
+							int firstDot = host.indexOf('.');
+							if (firstDot < 0 || firstDot == host.length() - 1) return;
+							String domain = host.substring(firstDot + 1);
+							String websocketUrl = "wss://smtp." + domain + "/mocksmtp/websocket";
+							logger.info("Reconnecting WebSocket (attempt " + attempt + "): " + websocketUrl);
+							HTTP_CLIENT.newWebSocketBuilder()
+									.buildAsync(URI.create(websocketUrl), new WebSocketClient())
+									.get(30, TimeUnit.SECONDS);
+							logger.info("WebSocket reconnected successfully on attempt " + attempt);
+							reconnectEvents.add("WebSocket reconnected successfully on attempt "
+									+ attempt + " at " + java.time.LocalDateTime.now());
+							return;
+						} catch (Exception e) {
+							String failMsg = "WebSocket reconnect attempt " + attempt + " failed: "
+									+ e.getMessage() + " at " + java.time.LocalDateTime.now();
+							logger.warn(failMsg + ". Retrying in "
+									+ Math.min(3000L * (attempt + 1), 60000L) + "ms...");
+							reconnectEvents.add(failMsg);
+						}
+					}
+				} finally {
+					reconnecting.set(false);
+				}
+			}, "ws-reconnect");
+			t.setDaemon(true);
+			t.start();
+		}
 	}
 }

apitest-commons/src/main/java/io/mosip/testrig/apirig/utils/NotificationListener.java

@@ -18,7 +18,8 @@ public final class NotificationListener {
 
 	private static final int MAX_QUEUE_SIZE = 50;
 	private static final long INACTIVE_EXPIRY_MS = Long.parseLong(ConfigManager.getproperty("otp_queue_inactive_expiry_time")) * 60 * 1000; // 15 mins
-	private static final Pattern OTP_PATTERN = Pattern.compile("(?i)\\bis\\s+(\\d{6})\\s+and\\s+is\\s+valid");
+	private static final Pattern OTP_PATTERN = Pattern.compile(
+			"(?i)(?:\\bis\\s+(\\d{6})\\s+and\\s+is\\s+valid|\\buse\\s+(?:otp\\s+)?(\\d{6})\\b)");
 	private static final ConcurrentHashMap<String, EmailQueue> otpQueues = new ConcurrentHashMap<>();
 
 	private static final ConcurrentHashMap<String, EmailQueue> notificationQueues = new ConcurrentHashMap<>();
@@ -193,12 +194,12 @@ public static String parseOtp(String message) {
 		}
 		Matcher matcher = OTP_PATTERN.matcher(message);
 		if (matcher.find()) {
-
-			String otp = matcher.group(1);
-
-			logger.info("Extracted OTP=" + otp + " from message");
-
-			return otp;
+			for (int i = 1; i <= matcher.groupCount(); i++) {
+				if (matcher.group(i) != null) {
+					logger.info("Extracted OTP=" + matcher.group(i) + " from message");
+					return matcher.group(i);
+				}
+			}
 		}
 
 		logger.info("OTP not found in message");

apitest-commons/src/main/java/io/mosip/testrig/apirig/utils/ReportUtil.java

@@ -6,6 +6,7 @@
 
 import io.mosip.testrig.apirig.dto.OutputValidationDto;
 import io.mosip.testrig.apirig.testrunner.JsonPrecondtion;
+import io.mosip.testrig.apirig.testrunner.OTPListener;
 
 /**
  * Class to show the result in table and text area format in testng report
@@ -38,6 +39,13 @@ public static String getTextAreaForHeaders(String headers) {
 	 * @return html table
 	 */
 	public static String getOutputValidationReport(Map<String, List<OutputValidationDto>> outputresult) {
+		StringBuilder reconnectWarnings = new StringBuilder();
+		String reconnectEvent;
+		while ((reconnectEvent = OTPListener.reconnectEvents.poll()) != null) {
+			reconnectWarnings.append("<div style='background-color:#FFF3CD; border-left:4px solid #FFA500; padding:6px; margin:4px 0;'>")
+					.append("<b>&#9888; WebSocket Reconnect:</b> ").append(reconnectEvent).append("</div>");
+		}
+
 		String htmlforReport = "<table width='100%' charset='UTF8'>\r\n" + "  <tr style='background-color: #d3d3d3;'>\r\n" + "    <th>FieldName</th>\r\n"
 				+ "    <th>Expected Value</th> \r\n" + "    <th>Actual Value</th>\r\n" + "    <th>Status</th>\r\n"
 				+ "  </tr>\r\n";
@@ -70,10 +78,10 @@ public static String getOutputValidationReport(Map<String, List<OutputValidation
 			}
 		}
 		if (!outputValidationDone) {
-		    return "<b style=\"background-color: #0A0;\">Marking test case as passed. As Output validation not performed and no errors in the response</b><br>\n";
+		    return reconnectWarnings + "<b style=\"background-color: #0A0;\">Marking test case as passed. As Output validation not performed and no errors in the response</b><br>\n";
 		}
-			
-		htmlforReport = temp + htmlforReport + "</table>";
+
+		htmlforReport = reconnectWarnings + temp + htmlforReport + "</table>";
 		return htmlforReport;
 	}