Feature/collab topic thread context

src/prerna/collaboration/BrainFixtureMessageSource.java

@@ -31,11 +31,13 @@
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Base64;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -89,6 +91,58 @@ public Map<String, Object> fetch(User user, String source, String conversationId
 		return "email".equals(source) && graphId != null ? byId.get(graphId) : null;
 	}
 
+	// the same over mail.json: every message of these conversations, newest first in each
+	@Override
+	public List<Map<String, Object>> conversationMessages(User user, java.util.Collection<String> conversationIds,
+			int perConversation, java.util.concurrent.atomic.AtomicInteger unread) {
+		java.util.Set<String> wanted = new java.util.HashSet<>(conversationIds);
+		Map<String, List<Map<String, Object>>> byConversation = new LinkedHashMap<>();
+		for (Map<String, Object> message : byId.values()) {
+			String conversation = String.valueOf(message.get("conversationId"));
+			if (wanted.contains(conversation)) {
+				byConversation.computeIfAbsent(conversation, k -> new ArrayList<>()).add(message);
+			}
+		}
+		List<Map<String, Object>> out = new ArrayList<>();
+		for (List<Map<String, Object>> messages : byConversation.values()) {
+			messages.sort((a, b) -> String.valueOf(b.get("receivedDateTime"))
+					.compareTo(String.valueOf(a.get("receivedDateTime"))));
+			out.addAll(messages.size() > perConversation ? messages.subList(0, perConversation) : messages);
+		}
+		return out;
+	}
+
+	// the same search over mail.json: the terms are looked for in each message's subject and body text
+	@Override
+	public List<Map<String, Object>> search(User user, List<String> terms, boolean all, Instant from, Instant to,
+			int max) {
+		List<Map<String, Object>> out = new ArrayList<>();
+		for (Map<String, Object> message : byId.values()) {
+			Object at = message.get("receivedDateTime");
+			if (at == null || (from != null && at.toString().compareTo(from.toString()) < 0)
+					|| (to != null && at.toString().compareTo(to.toString()) > 0)) {
+				continue;
+			}
+			Object body = message.get("body") instanceof Map<?, ?> b ? b.get("content") : null;
+			String text = (message.get("subject") + " " + body).replaceAll("<[^>]*>", " ").toLowerCase(Locale.ROOT);
+			boolean hit = all ? terms.stream().allMatch(t -> text.contains(t.toLowerCase(Locale.ROOT)))
+					: terms.stream().anyMatch(t -> text.contains(t.toLowerCase(Locale.ROOT)));
+			if (hit) {
+				Map<String, Object> found = new LinkedHashMap<>();
+				for (String key : List.of("id", "conversationId", "subject", "from", "receivedDateTime",
+						"parentFolderId")) {
+					found.put(key, message.get(key));
+				}
+				// Graph's bodyPreview: the start of the body, so hits can be ranked before they are read
+				String preview = String.valueOf(body).replaceAll("<[^>]*>", " ").replaceAll("\\s+", " ").trim();
+				found.put("bodyPreview", preview.length() > 255 ? preview.substring(0, 255) : preview);
+				out.add(found);
+			}
+		}
+		out.sort((a, b) -> String.valueOf(b.get("receivedDateTime")).compareTo(String.valueOf(a.get("receivedDateTime"))));
+		return out.size() > max ? new ArrayList<>(out.subList(0, max)) : out;
+	}
+
 	// a mail.json message may carry "attachments" in Graph shape, with base64 contentBytes
 	@Override
 	public Map<String, List<Map<String, Object>>> attachments(User user, String source, List<String> graphIds) {

src/prerna/collaboration/BrainGraphMessageSource.java

@@ -31,12 +31,16 @@
 import java.net.URLEncoder;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
+import java.time.Instant;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 
 import org.apache.hc.core5.http.ContentType;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import prerna.auth.User;
 import prerna.io.connector.ms.MicrosoftLoginUtils;
@@ -47,8 +51,18 @@
 // reads one message from Graph with the owner's delegated token, only after the rules let it through
 final class BrainGraphMessageSource implements BrainMessageSource {
 
+	private static final Logger classLogger = LogManager.getLogger(BrainGraphMessageSource.class);
+
 	private static final String BASE = MicrosoftTokenFiller.MS_GRAPH_BASE_API + "/v1.0";
 	private static final int BATCH = 20;
+	private static final int SEARCH_PAGE = 50;
+	private static final int CONVERSATION_BATCH = 10;
+	private static final int CONVERSATION_PAGE = 25;
+	// what the rules need of a message, and no body
+	private static final String CONVERSATION_SELECT = "id,conversationId,subject,from,receivedDateTime,parentFolderId,"
+			+ "isDraft";
+	private static final String SEARCH_SELECT = "id,conversationId,subject,from,receivedDateTime,parentFolderId,"
+			+ "isDraft,bodyPreview";
 	private static final int RETRIES = 3;
 	private static final long MAX_WAIT_MS = 10000;
 	private static final String MAIL_SELECT = "subject,from,toRecipients,ccRecipients,body,uniqueBody,receivedDateTime,"
@@ -68,6 +82,124 @@ public Map<String, Object> fetch(User user, String source, String conversationId
 		return "teams".equals(source) ? fromChat(message) : message;
 	}
 
+	// Graph's own index over subject and body: one search, followed page by page up to max. $search cannot be combined
+	// with $filter, so the dates ride in the query; a source that rejects them is asked again without
+	@Override
+	@SuppressWarnings("unchecked")
+	public List<Map<String, Object>> search(User user, List<String> terms, boolean all, Instant from, Instant to,
+			int max) throws Exception {
+		String token = MicrosoftLoginUtils.getValidAccessToken(user);
+		String words = String.join(all ? " AND " : " OR ", terms.stream().map(t -> t.replace("\"", " ")).toList());
+		String range = from == null || to == null ? ""
+				: " AND received:" + from.toString().substring(0, 10) + ".." + to.toString().substring(0, 10);
+		for (String query : range.isEmpty() ? List.of(words) : List.of("(" + words + ")" + range, words)) {
+			List<Map<String, Object>> out = new ArrayList<>();
+			String url = BASE + "/me/messages?$search=" + encode("\"" + query + "\"") + "&$top=" + SEARCH_PAGE
+					+ "&$select=" + SEARCH_SELECT;
+			try {
+				while (url != null && out.size() < max) {
+					Map<String, Object> page = CollaborationDbUtils.parseMap(
+							HttpHelperUtility.getRequest(url, MicrosoftLoginUtils.getBearerHeader(token), null, null, null));
+					if (page.get("value") instanceof List<?> values) {
+						for (Object value : values) {
+							out.add((Map<String, Object>) value);
+						}
+					}
+					url = page.get("@odata.nextLink") instanceof String next ? next : null;
+				}
+			} catch (SemossPixelException e) {
+				throw e;
+			} catch (Exception e) {
+				if (query.equals(words)) {
+					throw e;
+				}
+				continue;
+			}
+			out.sort((a, b) -> String.valueOf(b.get("receivedDateTime")).compareTo(String.valueOf(a.get("receivedDateTime"))));
+			return out.size() > max ? new ArrayList<>(out.subList(0, max)) : out;
+		}
+		return List.of();
+	}
+
+	// Graph $batch of conversation listings, 10 a call: the messages of each conversation without their bodies, so a
+	// topic's mail can be listed whole whether or not it was imported, and the rules decide before a body is read.
+	// A conversation is followed page by page until perConversation messages are listed or it has no more. A
+	// throttled (429) or busy (5xx) listing is sent again; one that is refused, or still failing after the retries,
+	// counts in unread.
+	@Override
+	@SuppressWarnings("unchecked")
+	public List<Map<String, Object>> conversationMessages(User user, java.util.Collection<String> conversationIds,
+			int perConversation, java.util.concurrent.atomic.AtomicInteger unread) throws Exception {
+		String token = MicrosoftLoginUtils.getValidAccessToken(user);
+		List<String> ids = new ArrayList<>(conversationIds);
+		List<Map<String, Object>> out = new ArrayList<>();
+		int pageSize = Math.min(perConversation, CONVERSATION_PAGE);
+		for (int start = 0; start < ids.size(); start += CONVERSATION_BATCH) {
+			Map<String, Map<String, Object>> pending = new LinkedHashMap<>();
+			for (int i = start; i < Math.min(ids.size(), start + CONVERSATION_BATCH); i++) {
+				String filter = "conversationId eq '" + ids.get(i).replace("'", "''") + "'";
+				pending.put(String.valueOf(i), Map.of("id", String.valueOf(i), "method", "GET", "url",
+						"/me/messages?$filter=" + encode(filter) + "&$top=" + pageSize + "&$select="
+								+ CONVERSATION_SELECT));
+			}
+			Map<String, Integer> collected = new HashMap<>();
+			Map<String, Integer> retries = new HashMap<>();
+			while (!pending.isEmpty()) {
+				Map<String, Object> reply = CollaborationDbUtils.parseMap(HttpHelperUtility.postRequestStringBody(
+						BASE + "/$batch", MicrosoftLoginUtils.getBearerHeader(token),
+						CollaborationDbUtils.toJson(Map.of("requests", new ArrayList<>(pending.values()))),
+						ContentType.APPLICATION_JSON, null, null, null, BrainGraphHeaderSource.TIMEOUT));
+				Map<String, Map<String, Object>> answered = new HashMap<>();
+				if (reply.get("responses") instanceof List<?> list) {
+					for (Object r : list) {
+						Map<String, Object> response = (Map<String, Object>) r;
+						answered.put(String.valueOf(response.get("id")), response);
+					}
+				}
+				Map<String, Map<String, Object>> next = new LinkedHashMap<>();
+				long wait = 0;
+				for (Map.Entry<String, Map<String, Object>> entry : pending.entrySet()) {
+					String id = entry.getKey();
+					Map<String, Object> response = answered.get(id);
+					int status = response != null && response.get("status") instanceof Number n ? n.intValue() : 0;
+					if (response != null && (status == 429 || status >= 500) && retries.getOrDefault(id, 0) < RETRIES) {
+						int attempt = retries.merge(id, 1, Integer::sum) - 1;
+						next.put(id, entry.getValue());
+						wait = Math.max(wait, retryAfter(response, attempt));
+					} else if (status == 200 && response.get("body") instanceof Map<?, ?> body
+							&& body.get("value") instanceof List<?> values) {
+						int have = collected.getOrDefault(id, 0);
+						for (Object value : values) {
+							if (have < perConversation) {
+								out.add((Map<String, Object>) value);
+								have++;
+							}
+						}
+						collected.put(id, have);
+						// the rest of a long conversation, from the link Graph gives for the next page
+						if (have < perConversation && body.get("@odata.nextLink") instanceof String link
+								&& link.startsWith(BASE)) {
+							next.put(id, Map.of("id", id, "method", "GET", "url", link.substring(BASE.length())));
+						}
+					} else {
+						classLogger.warn("Graph could not list a conversation of the topic: status {}", status);
+						unread.incrementAndGet();
+					}
+				}
+				pending = next;
+				if (wait > 0 && !pending.isEmpty()) {
+					try {
+						Thread.sleep(wait);
+					} catch (InterruptedException e) {
+						Thread.currentThread().interrupt();
+						break;
+					}
+				}
+			}
+		}
+		return out;
+	}
+
 	// Graph $batch, 20 a call: one round trip instead of one per message
 	@Override
 	@SuppressWarnings("unchecked")

src/prerna/collaboration/BrainMessageSource.java

@@ -28,6 +28,7 @@
 package prerna.collaboration;
 
 import java.nio.file.Path;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -68,6 +69,24 @@ default List<Fetched> fetchAll(User user, String source, String conversationId,
 		return out;
 	}
 
+	// email whose subject or body holds the terms (every term when all is set, else any), newest first and at most max,
+	// each in Graph shape without the body: id, conversationId, subject, from, receivedDateTime, parentFolderId,
+	// isDraft. A source that cannot search returns null; a login problem (SemossPixelException) is thrown
+	default List<Map<String, Object>> search(User user, List<String> terms, boolean all, Instant from, Instant to,
+			int max) throws Exception {
+		return null;
+	}
+
+	// the messages of these conversations without their bodies, in Graph shape (id, conversationId, subject, from,
+	// receivedDateTime, parentFolderId, isDraft), at most perConversation from each, in no particular order; the
+	// bodies of the ones the rules let through are read with fetchAll. A source that cannot list a conversation
+	// returns null; a login problem (SemossPixelException) is thrown. unread counts the conversations that could
+	// not be listed, so the caller can say the search was partial.
+	default List<Map<String, Object>> conversationMessages(User user, java.util.Collection<String> conversationIds,
+			int perConversation, java.util.concurrent.atomic.AtomicInteger unread) throws Exception {
+		return null;
+	}
+
 	// what is attached to messages already known to have attachments, by message id, as
 	// BrainAttachments.describe maps (no bytes, inline images left out); a source that
 	// cannot tell returns nothing