fix(email-inbound-connector): implement a reconnect and reopen mechan…

…ism (#3506)

* fix(email-inbound-connector): implement a reconnect and reopen mechanism

* fix(email-connectors): add equalsIgnoreCase for INBOX

* fix(email-inbound-connector): add integration tests, modify the split for folder creation, and add unit tests

* fix(email-inbound-connector): add integration tests, modify the split for folder creation, and add unit tests 2

connectors-e2e-test/connectors-e2e-test-mail/src/test/java/io/camunda/connector/e2e/BaseEmailTest.java

@@ -38,7 +38,6 @@ public class BaseEmailTest {
   private static final GreenMail greenMail = new GreenMail();
   @TempDir File tempDir;
   private GreenMailUser greenMailUser = greenMail.setUser("[email protected]", "password");
-  ;
 
   @BeforeAll
   static void setup() {

connectors-e2e-test/connectors-e2e-test-mail/src/test/java/io/camunda/connector/e2e/InboundEmailTest.java

@@ -0,0 +1,252 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH
+ * under one or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information regarding copyright
+ * ownership. Camunda licenses this file to you under the Apache License,
+ * Version 2.0; 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.camunda.connector.e2e;
+
+import static io.camunda.connector.e2e.BpmnFile.replace;
+import static org.mockito.Mockito.when;
+
+import io.camunda.connector.e2e.app.TestConnectorRuntimeApplication;
+import io.camunda.connector.runtime.inbound.state.ProcessImportResult;
+import io.camunda.connector.runtime.inbound.state.ProcessStateStore;
+import io.camunda.operate.CamundaOperateClient;
+import io.camunda.operate.exception.OperateException;
+import io.camunda.operate.model.ProcessDefinition;
+import io.camunda.zeebe.client.ZeebeClient;
+import io.camunda.zeebe.model.bpmn.BpmnModelInstance;
+import io.camunda.zeebe.model.bpmn.instance.Process;
+import io.camunda.zeebe.process.test.assertions.BpmnAssert;
+import io.camunda.zeebe.spring.test.ZeebeSpringTest;
+import jakarta.mail.Flags;
+import jakarta.mail.MessagingException;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.awaitility.core.ConditionTimeoutException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+
+@SpringBootTest(
+    classes = {TestConnectorRuntimeApplication.class},
+    properties = {
+      "spring.main.allow-bean-definition-overriding=true",
+    },
+    webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+@ZeebeSpringTest
+@ExtendWith(MockitoExtension.class)
+public class InboundEmailTest extends BaseEmailTest {
+
+  private static final ScheduledExecutorService scheduler =
+      Executors.newSingleThreadScheduledExecutor();
+  private static AtomicLong counter = new AtomicLong(1);
+  @Autowired ProcessStateStore processStateStore;
+  @Autowired CamundaOperateClient camundaOperateClient;
+  @Mock private ProcessDefinition processDef;
+  @Autowired private ZeebeClient zeebeClient;
+
+  @BeforeEach
+  public void beforeEach() {
+    super.reset();
+  }
+
+  @Test
+  public void shouldReceiveEmailAndSetAsSeen() throws OperateException, MessagingException {
+    var model =
+        replace(
+            "email-inbound-connector-intermediate_unseen_read.bpmn",
+            BpmnFile.Replace.replace("55555", super.getUnsecureImapPort()));
+
+    mockProcessDefinition(model);
+
+    scheduler.schedule(
+        () -> super.sendEmail("[email protected]", "test", "hey"), 2, TimeUnit.SECONDS);
+
+    processStateStore.update(
+        new ProcessImportResult(
+            Map.of(
+                new ProcessImportResult.ProcessDefinitionIdentifier(
+                    processDef.getBpmnProcessId(), processDef.getTenantId()),
+                new ProcessImportResult.ProcessDefinitionVersion(
+                    processDef.getKey(), processDef.getVersion().intValue()))));
+    var bpmnTest = ZeebeTest.with(zeebeClient).deploy(model).createInstance();
+
+    bpmnTest = bpmnTest.waitForProcessCompletion();
+
+    Assertions.assertTrue(
+        Arrays.stream(super.getLastReceivedEmails())
+            .findFirst()
+            .get()
+            .getFlags()
+            .contains(Flags.Flag.SEEN));
+    BpmnAssert.assertThat(bpmnTest.getProcessInstanceEvent())
+        .hasVariableWithValue("subject", "test");
+    BpmnAssert.assertThat(bpmnTest.getProcessInstanceEvent())
+        .hasVariableWithValue("plainTextBody", "hey");
+  }
+
+  @Test
+  public void shouldThrowWhenAllMessageAreSeen() throws OperateException, MessagingException {
+    var model =
+        replace(
+            "email-inbound-connector-intermediate_unseen_read.bpmn",
+            BpmnFile.Replace.replace("55555", super.getUnsecureImapPort()));
+
+    mockProcessDefinition(model);
+
+    super.sendEmail("[email protected]", "test", "hey");
+
+    Arrays.stream(getLastReceivedEmails()).findFirst().get().setFlag(Flags.Flag.SEEN, true);
+
+    processStateStore.update(
+        new ProcessImportResult(
+            Map.of(
+                new ProcessImportResult.ProcessDefinitionIdentifier(
+                    processDef.getBpmnProcessId(), processDef.getTenantId()),
+                new ProcessImportResult.ProcessDefinitionVersion(
+                    processDef.getKey(), processDef.getVersion().intValue()))));
+    var bpmnTest = ZeebeTest.with(zeebeClient).deploy(model).createInstance();
+
+    Assertions.assertThrows(ConditionTimeoutException.class, bpmnTest::waitForProcessCompletion);
+  }
+
+  @Test
+  public void shouldReceiveEmailAndDelete() throws OperateException, MessagingException {
+    var model =
+        replace(
+            "email-inbound-connector-intermediate_unseen_delete.bpmn",
+            BpmnFile.Replace.replace("55555", super.getUnsecureImapPort()));
+
+    mockProcessDefinition(model);
+
+    scheduler.schedule(
+        () -> super.sendEmail("[email protected]", "test", "hey"), 2, TimeUnit.SECONDS);
+
+    processStateStore.update(
+        new ProcessImportResult(
+            Map.of(
+                new ProcessImportResult.ProcessDefinitionIdentifier(
+                    processDef.getBpmnProcessId(), processDef.getTenantId()),
+                new ProcessImportResult.ProcessDefinitionVersion(
+                    processDef.getKey(), processDef.getVersion().intValue()))));
+
+    var bpmnTest = ZeebeTest.with(zeebeClient).deploy(model).createInstance();
+
+    bpmnTest = bpmnTest.waitForProcessCompletion();
+
+    Assertions.assertTrue(
+        Arrays.stream(super.getLastReceivedEmails())
+            .findFirst()
+            .get()
+            .getFlags()
+            .contains(Flags.Flag.DELETED));
+    BpmnAssert.assertThat(bpmnTest.getProcessInstanceEvent())
+        .hasVariableWithValue("subject", "test");
+    BpmnAssert.assertThat(bpmnTest.getProcessInstanceEvent())
+        .hasVariableWithValue("plainTextBody", "hey");
+  }
+
+  @Test
+  public void shouldReceiveEmailAndMove() throws OperateException, MessagingException {
+    var model =
+        replace(
+            "email-inbound-connector-intermediate_unseen_move.bpmn",
+            BpmnFile.Replace.replace("55555", super.getUnsecureImapPort()));
+
+    mockProcessDefinition(model);
+
+    scheduler.schedule(
+        () -> super.sendEmail("[email protected]", "test", "hey"), 2, TimeUnit.SECONDS);
+
+    processStateStore.update(
+        new ProcessImportResult(
+            Map.of(
+                new ProcessImportResult.ProcessDefinitionIdentifier(
+                    processDef.getBpmnProcessId(), processDef.getTenantId()),
+                new ProcessImportResult.ProcessDefinitionVersion(
+                    processDef.getKey(), processDef.getVersion().intValue()))));
+
+    var bpmnTest = ZeebeTest.with(zeebeClient).deploy(model).createInstance();
+
+    bpmnTest = bpmnTest.waitForProcessCompletion();
+
+    Assertions.assertEquals(2, getLastReceivedEmails().length);
+    Assertions.assertTrue(
+        Arrays.stream(getLastReceivedEmails())
+            .findFirst()
+            .get()
+            .getFlags()
+            .contains(Flags.Flag.DELETED));
+    BpmnAssert.assertThat(bpmnTest.getProcessInstanceEvent())
+        .hasVariableWithValue("subject", "test");
+    BpmnAssert.assertThat(bpmnTest.getProcessInstanceEvent())
+        .hasVariableWithValue("plainTextBody", "hey");
+  }
+
+  private void mockProcessDefinition(BpmnModelInstance model) throws OperateException {
+    when(camundaOperateClient.getProcessDefinitionModel(1L)).thenReturn(model);
+    when(processDef.getKey()).thenReturn(1L);
+    when(processDef.getTenantId()).thenReturn(zeebeClient.getConfiguration().getDefaultTenantId());
+    when(processDef.getBpmnProcessId())
+        .thenReturn(model.getModelElementsByType(Process.class).stream().findFirst().get().getId());
+    when(processDef.getVersion()).thenReturn(counter.getAndIncrement());
+  }
+
+  @Test
+  public void shouldPollEmailAndMove() throws OperateException, MessagingException {
+    var model =
+        replace(
+            "email-inbound-connector-intermediate_all_delete.bpmn",
+            BpmnFile.Replace.replace("55555", super.getUnsecureImapPort()));
+
+    mockProcessDefinition(model);
+
+    super.sendEmail("[email protected]", "test", "hey");
+
+    Arrays.stream(super.getLastReceivedEmails()).findFirst().get().setFlag(Flags.Flag.SEEN, true);
+
+    processStateStore.update(
+        new ProcessImportResult(
+            Map.of(
+                new ProcessImportResult.ProcessDefinitionIdentifier(
+                    processDef.getBpmnProcessId(), processDef.getTenantId()),
+                new ProcessImportResult.ProcessDefinitionVersion(
+                    processDef.getKey(), processDef.getVersion().intValue()))));
+
+    var bpmnTest = ZeebeTest.with(zeebeClient).deploy(model).createInstance();
+
+    bpmnTest = bpmnTest.waitForProcessCompletion();
+
+    Assertions.assertTrue(
+        Arrays.stream(super.getLastReceivedEmails())
+            .findFirst()
+            .get()
+            .getFlags()
+            .contains(Flags.Flag.DELETED));
+    BpmnAssert.assertThat(bpmnTest.getProcessInstanceEvent())
+        .hasVariableWithValue("subject", "test");
+    BpmnAssert.assertThat(bpmnTest.getProcessInstanceEvent())
+        .hasVariableWithValue("plainTextBody", "hey");
+  }
+}

connectors-e2e-test/connectors-e2e-test-mail/src/test/java/io/camunda/connector/e2e/EmailTests.java → connectors-e2e-test/connectors-e2e-test-mail/src/test/java/io/camunda/connector/e2e/OutboundEmailTests.java

@@ -47,7 +47,7 @@
     webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
 @ZeebeSpringTest
 @ExtendWith(MockitoExtension.class)
-public class EmailTests extends BaseEmailTest {
+public class OutboundEmailTests extends BaseEmailTest {
 
   private static final ScheduledExecutorService scheduler =
       Executors.newSingleThreadScheduledExecutor();

connectors-e2e-test/connectors-e2e-test-mail/src/test/resources/email-inbound-connector-intermediate_all_delete.bpmn

@@ -0,0 +1,64 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<bpmn:definitions xmlns:bpmn="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlns:dc="http://www.omg.org/spec/DD/20100524/DC" xmlns:zeebe="http://camunda.org/schema/zeebe/1.0" xmlns:di="http://www.omg.org/spec/DD/20100524/DI" xmlns:modeler="http://camunda.org/schema/modeler/1.0" id="Definitions_1ry9nn1" targetNamespace="http://bpmn.io/schema/bpmn" exporter="Camunda Modeler" exporterVersion="5.24.0" modeler:executionPlatform="Camunda Cloud" modeler:executionPlatformVersion="8.5.0">
+  <bpmn:process id="Process_1crecrd" isExecutable="true">
+    <bpmn:endEvent id="Event_1aq97te">
+      <bpmn:incoming>Flow_1vp6srd</bpmn:incoming>
+    </bpmn:endEvent>
+    <bpmn:startEvent id="Event_0xs76az">
+      <bpmn:outgoing>Flow_0lofu08</bpmn:outgoing>
+    </bpmn:startEvent>
+    <bpmn:sequenceFlow id="Flow_0lofu08" sourceRef="Event_0xs76az" targetRef="Event_1y1ygyn" />
+    <bpmn:intermediateCatchEvent id="Event_1y1ygyn" zeebe:modelerTemplate="io.camunda.connectors.inbound.EmailIntermediate.v1" zeebe:modelerTemplateVersion="1" zeebe:modelerTemplateIcon="data:image/svg+xml;base64,PHN2ZyB3aWR0aD0iMTYiIGhlaWdodD0iMTYiIHZpZXdCb3g9IjAgMCAxNiAxNiIgZmlsbD0ibm9uZSIgeG1sbnM9Imh0dHA6Ly93d3cudzMub3JnLzIwMDAvc3ZnIj4KPGcgY2xpcC1wYXRoPSJ1cmwoI2NsaXAwXzkwXzI0MjApIj4KPHBhdGggZD0iTTguMzM4MzUgOS45NTM2NUwxMC4zODk0IDEyLjAxMDRMOC4zMzI2MiAxNC4wNjcyTDkuMTQ2MTYgMTQuODc1TDEyLjAxMDcgMTIuMDEwNEw5LjE0NjE2IDkuMTQ1ODNMOC4zMzgzNSA5Ljk1MzY1WiIgZmlsbD0iYmxhY2siLz4KPHBhdGggZD0iTTEyLjM0ODggOS45NTM2NUwxNC4zOTk4IDEyLjAxMDRMMTIuMzQzIDE0LjA2NzJMMTMuMTU2NiAxNC44NzVMMTYuMDIxMiAxMi4wMTA0TDEzLjE1NjYgOS4xNDU4M0wxMi4zNDg4IDkuOTUzNjVaIiBmaWxsPSJibGFjayIvPgo8cGF0aCBkPSJNMy45NzIgMTEuNDM3NUgxLjEyNTMzVjIuNzkyMTlMNy42NzM3NiA3LjMyMzk2QzcuNzY5NjcgNy4zOTA0OSA3Ljg4MzYgNy40MjYxNCA4LjAwMDMyIDcuNDI2MTRDOC4xMTcwNSA3LjQyNjE0IDguMjMwOTggNy4zOTA0OSA4LjMyNjg5IDcuMzIzOTZMMTQuODc1MyAyLjc5MjE5VjhIMTYuMDIxMlYyLjI3MDgzQzE2LjAyMTIgMS45NjY5NCAxNS45MDA0IDEuNjc1NDkgMTUuNjg1NiAxLjQ2MDYxQzE1LjQ3MDcgMS4yNDU3MiAxNS4xNzkyIDEuMTI1IDE0Ljg3NTMgMS4xMjVIMS4xMjUzM0MwLjgyMTQzMiAxLjEyNSAwLjUyOTk4NCAxLjI0NTcyIDAuMzE1MDk5IDEuNDYwNjFDMC4xMDAyMTQgMS42NzU0OSAtMC4wMjA1MDc4IDEuOTY2OTQgLTAuMDIwNTA3OCAyLjI3MDgzVjExLjQzNzVDLTAuMDIwNTA3OCAxMS43NDE0IDAuMTAwMjE0IDEyLjAzMjggMC4zMTUwOTkgMTIuMjQ3N0MwLjUyOTk4NCAxMi40NjI2IDAuODIxNDMyIDEyLjU4MzMgMS4xMjUzMyAxMi41ODMzSDMuOTcyVjExLjQzNzVaTTEzLjYxNDkgMi4yNzA4M0w4LjAwMDMyIDYuMTU1MjFMMi4zODU3NCAyLjI3MDgzSDEzLjYxNDlaIiBmaWxsPSIjRkM1RDBEIi8+CjxwYXRoIGQ9Ik00LjI4MjEgOS45NTM2NUw2LjMzMzE0IDEyLjAxMDRMNC4yNzYzNyAxNC4wNjcyTDUuMDg5OTEgMTQuODc1TDcuOTU0NDkgMTIuMDEwNEw1LjA4OTkxIDkuMTQ1ODNMNC4yODIxIDkuOTUzNjVaIiBmaWxsPSJibGFjayIvPgo8L2c+CjxkZWZzPgo8Y2xpcFBhdGggaWQ9ImNsaXAwXzkwXzI0MjAiPgo8cmVjdCB3aWR0aD0iMTYiIGhlaWdodD0iMTYiIGZpbGw9IndoaXRlIi8+CjwvY2xpcFBhdGg+CjwvZGVmcz4KPC9zdmc+Cg==">
+      <bpmn:extensionElements>
+        <zeebe:properties>
+          <zeebe:property name="inbound.type" value="io.camunda:connector-email-inbound:1" />
+          <zeebe:property name="authentication.type" value="simple" />
+          <zeebe:property name="authentication.username" value="[email protected]" />
+          <zeebe:property name="authentication.password" value="password" />
+          <zeebe:property name="data.imapConfig.imapHost" value="localhost" />
+          <zeebe:property name="data.imapConfig.imapPort" value="55555" />
+          <zeebe:property name="data.imapConfig.imapCryptographicProtocol" value="NONE" />
+          <zeebe:property name="data.pollingWaitTime" value="PT2S" />
+          <zeebe:property name="data.pollingConfigDiscriminator" value="allPollingConfig" />
+          <zeebe:property name="data.pollingConfig.handlingStrategy" value="DELETE" />
+          <zeebe:property name="consumeUnmatchedEvents" value="true" />
+          <zeebe:property name="correlationKeyExpression" value="=&#34;ok&#34;" />
+          <zeebe:property name="deduplicationModeManualFlag" value="false" />
+          <zeebe:property name="deduplicationMode" value="AUTO" />
+          <zeebe:property name="resultVariable" />
+          <zeebe:property name="resultExpression" value="={fromAddress : fromAddress, messageId : messageId, subject: subject, size: size, plainTextBody : plainTextBody }" />
+        </zeebe:properties>
+      </bpmn:extensionElements>
+      <bpmn:incoming>Flow_0lofu08</bpmn:incoming>
+      <bpmn:outgoing>Flow_1vp6srd</bpmn:outgoing>
+      <bpmn:messageEventDefinition id="MessageEventDefinition_1om0jbh" messageRef="Message_1rm9467" />
+    </bpmn:intermediateCatchEvent>
+    <bpmn:sequenceFlow id="Flow_1vp6srd" sourceRef="Event_1y1ygyn" targetRef="Event_1aq97te" />
+  </bpmn:process>
+  <bpmn:message id="Message_1rm9467" name="0514aee7-2eab-4cf7-9f6c-acae87549711" zeebe:modelerTemplate="io.camunda.connectors.inbound.EmailIntermediate.v1">
+    <bpmn:extensionElements>
+      <zeebe:subscription correlationKey="=&#34;ok&#34;" />
+    </bpmn:extensionElements>
+  </bpmn:message>
+  <bpmndi:BPMNDiagram id="BPMNDiagram_1">
+    <bpmndi:BPMNPlane id="BPMNPlane_1" bpmnElement="Process_1crecrd">
+      <bpmndi:BPMNShape id="Event_1aq97te_di" bpmnElement="Event_1aq97te">
+        <dc:Bounds x="782" y="72" width="36" height="36" />
+      </bpmndi:BPMNShape>
+      <bpmndi:BPMNShape id="Event_0xs76az_di" bpmnElement="Event_0xs76az">
+        <dc:Bounds x="192" y="72" width="36" height="36" />
+      </bpmndi:BPMNShape>
+      <bpmndi:BPMNShape id="Event_1hvfzo1_di" bpmnElement="Event_1y1ygyn">
+        <dc:Bounds x="432" y="72" width="36" height="36" />
+      </bpmndi:BPMNShape>
+      <bpmndi:BPMNEdge id="Flow_0lofu08_di" bpmnElement="Flow_0lofu08">
+        <di:waypoint x="228" y="90" />
+        <di:waypoint x="432" y="90" />
+      </bpmndi:BPMNEdge>
+      <bpmndi:BPMNEdge id="Flow_1vp6srd_di" bpmnElement="Flow_1vp6srd">
+        <di:waypoint x="468" y="90" />
+        <di:waypoint x="782" y="90" />
+      </bpmndi:BPMNEdge>
+    </bpmndi:BPMNPlane>
+  </bpmndi:BPMNDiagram>
+</bpmn:definitions>