Release/0.26.0

.github/ISSUE_TEMPLATE/---bug-report.md

@@ -0,0 +1,32 @@
+---
+name: "\U0001F41E Bug report"
+about: Report a bug in Airy. If this doesn’t look right, [choose a different type](https://github.com/airyhq/airy/issues/new/choose).
+title: ""
+labels: bug
+assignees: ""
+---
+
+**Describe the bug**
+A clear and concise description of what the bug is.
+
+**To Reproduce**
+Steps to reproduce the behavior:
+
+1. I installed Airy using '...'
+2. When I ran command / did this in the UI
+3. See error
+
+**Expected behavior**
+A clear and concise description of what you expected to happen.
+
+**Screenshots**
+If applicable, add screenshots to help explain your problem.
+
+**Desktop (please complete the following information):**
+
+- OS: [e.g. iOS]
+- Browser [e.g. Chrome, Safari]
+- Version [e.g. 22]
+
+**Additional context**
+Add any other context about the problem here.

.github/ISSUE_TEMPLATE/---feature-request.md

@@ -0,0 +1,19 @@
+---
+name: "\U0001F680 Feature request"
+about: Suggest an idea for Airy. If this doesn’t look right, [choose a different type](https://github.com/airyhq/airy/issues/new/choose).
+title: ""
+labels: feature, needs discussion
+assignees: ""
+---
+
+**Is your feature request related to a problem? Please describe.**
+A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]
+
+**Describe the solution you'd like**
+A clear and concise description of what you want to happen.
+
+**Describe alternatives you've considered**
+A clear and concise description of any alternative solutions or features you've considered.
+
+**Additional context**
+Add any other context or screenshots about the feature request here.

.github/dependabot.yml

@@ -3,4 +3,4 @@ updates:
   - package-ecosystem: "npm"
     directory: "/"
     schedule:
-      interval: "daily"
+      interval: "weekly"

.github/workflows/main.yml

@@ -57,10 +57,12 @@ jobs:
         run: |
           echo ${{ secrets.PAT }} | docker login ghcr.io -u airydevci --password-stdin
           ./scripts/push-images.sh
+
       - name: Install aws cli
         uses: chrislennon/action-aws-cli@v1.1
         env:
           ACTIONS_ALLOW_UNSECURE_COMMANDS: 'true'
+
       - name: Upload airy binary to S3
         if: ${{ github.ref == 'refs/heads/develop' || startsWith(github.ref, 'refs/heads/release') || github.ref == 'refs/heads/main' }}
         run: |
@@ -69,6 +71,7 @@ jobs:
           AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
           AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
           GITHUB_BRANCH: ${{ github.ref }}
+
       - name: Publish http-client library to npm
         if: ${{ startsWith(github.ref, 'refs/heads/main') }}
         run: |
@@ -78,6 +81,7 @@ jobs:
           DEPLOY_NPM_USERNAME: ${{ secrets.DEPLOY_NPM_USERNAME }}
           DEPLOY_NPM_PASSWORD: ${{ secrets.DEPLOY_NPM_PASSWORD }}
           DEPLOY_NPM_EMAIL: ${{ secrets.DEPLOY_NPM_EMAIL }}
+
       - name: Publish chat-plugin library to npm
         if: ${{ startsWith(github.ref, 'refs/heads/main') }}
         run: |
@@ -87,4 +91,3 @@ jobs:
           DEPLOY_NPM_USERNAME: ${{ secrets.DEPLOY_NPM_USERNAME }}
           DEPLOY_NPM_PASSWORD: ${{ secrets.DEPLOY_NPM_PASSWORD }}
           DEPLOY_NPM_EMAIL: ${{ secrets.DEPLOY_NPM_EMAIL }}
-          GITHUB_BRANCH: ${{ github.ref }}

VERSION

@@ -1 +1 @@
-0.25.0
+0.26.0

WORKSPACE

@@ -9,9 +9,9 @@ load("@bazel_tools//tools/build_defs/repo:git.bzl", "git_repository")
 # Airy Bazel tools
 git_repository(
     name = "com_github_airyhq_bazel_tools",
-    commit = "b4f52101d80c342ff5e4734fb4c4385af921ffec",
+    commit = "2b34bc2b2c6fed64e4bb223d4624e06eda390e4d",
     remote = "https://github.com/airyhq/bazel-tools.git",
-    shallow_since = "1624959927 +0200",
+    shallow_since = "1625065439 +0200",
 )
 
 load("@com_github_airyhq_bazel_tools//:repositories.bzl", "airy_bazel_tools_dependencies", "airy_jvm_deps")

backend/api/communication/src/main/java/co/airy/core/api/communication/Stores.java

@@ -28,6 +28,7 @@
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.kstream.KGroupedStream;
+import org.apache.kafka.streams.kstream.KGroupedTable;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.KTable;
 import org.apache.kafka.streams.kstream.Materialized;
@@ -103,8 +104,8 @@ private void startStream() {
                 .mapValues(readReceipt -> CountAction.reset(readReceipt.getReadDate()));
 
         // produce unread count metadata
-        messageStream.selectKey((messageId, message) -> message.getConversationId())
-                .filter((conversationId, message) -> message.getIsFromContact())
+        messageStream.filter((conversationId, message) -> message != null && message.getIsFromContact())
+                .selectKey((messageId, message) -> message.getConversationId())
                 .mapValues(message -> CountAction.increment(message.getSentAt()))
                 .merge(resetStream)
                 .groupByKey()
@@ -127,25 +128,26 @@ private void startStream() {
                 })
                 .to(applicationCommunicationMetadata);
 
-        final KGroupedStream<String, MessageContainer> messageGroupedStream = messageStream.toTable()
+        final KGroupedTable<String, MessageContainer> messageGroupedTable = messageStream.toTable()
                 .leftJoin(metadataTable, (message, metadataMap) -> MessageContainer.builder()
                         .message(message)
                         .metadataMap(Optional.ofNullable(metadataMap).orElse(new MetadataMap()))
                         .build(), Materialized.as(messagesByIdStore))
-                .toStream()
-                .filter((messageId, messageContainer) -> messageContainer != null)
-                .groupBy((messageId, messageContainer) -> messageContainer.getMessage().getConversationId());
+                .groupBy((messageId, messageContainer) -> KeyValue.pair(messageContainer.getMessage().getConversationId(), messageContainer));
 
 
         // messages store
-        messageGroupedStream.aggregate(MessagesTreeSet::new,
-                ((key, value, aggregate) -> {
+        messageGroupedTable.aggregate(MessagesTreeSet::new,
+                (key, value, aggregate) -> {
                     aggregate.update(value);
                     return aggregate;
-                }), Materialized.as(messagesStore));
+                }, (key, value, aggregate) -> {
+                    aggregate.remove(value);
+                    return aggregate;
+                }, Materialized.as(messagesStore));
 
         // Conversation stores
-        messageGroupedStream
+        messageGroupedTable
                 .aggregate(Conversation::new,
                         (conversationId, container, aggregate) -> {
                             if (aggregate.getLastMessageContainer() == null) {
@@ -164,6 +166,10 @@ private void startStream() {
                                 aggregate.setSourceConversationId(container.getMessage().getSenderId());
                             }
 
+                            return aggregate;
+                        }, (conversationId, container, aggregate) -> {
+                            // If the deleted message was the last message we have no way of replacing it
+                            // so we have no choice but to keep it
                             return aggregate;
                         })
                 .join(channelTable, Conversation::getChannelId,

backend/api/communication/src/test/java/co/airy/core/api/communication/MessagesTest.java

@@ -10,6 +10,8 @@
 import co.airy.kafka.test.junit.SharedKafkaTestResource;
 import co.airy.spring.core.AirySpringBootApplication;
 import co.airy.spring.test.WebTestHelper;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.avro.specific.SpecificRecordBase;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.AfterAll;
@@ -102,6 +104,33 @@ void canFetchMessages() throws Exception {
                 "/messages.list endpoint error");
     }
 
+    @Test
+    void canDeleteMessages() throws Exception {
+        final String conversationId = UUID.randomUUID().toString();
+
+        int messageCount = 2;
+        final List<ProducerRecord<String, SpecificRecordBase>> records = TestConversation.generateRecords(conversationId, channel, 2);
+        kafkaTestHelper.produceRecords(records);
+
+        final String payload = "{\"conversation_id\":\"" + conversationId + "\"}";
+        retryOnException(
+                () -> {
+                    final String content = webTestHelper.post("/messages.list", payload)
+                            .andExpect(status().isOk())
+                            .andExpect(jsonPath("$.data", hasSize(messageCount)))
+                            .andReturn().getResponse().getContentAsString();
+
+                    final JsonNode jsonNode = new ObjectMapper().readTree(content);
+                    final String messageId = jsonNode.get("data").get(0).get("id").textValue();
+
+                    kafkaTestHelper.produceRecord(new ProducerRecord<>(applicationCommunicationMessages.name(), messageId, null));
+
+                    webTestHelper.post("/messages.list", payload)
+                            .andExpect(status().isOk())
+                            .andExpect(jsonPath("$.data", hasSize(messageCount - 1)));
+                }, "message was not deleted");
+    }
+
     @Test
     void canReturnMetadata() throws Exception {
         final String conversationId = UUID.randomUUID().toString();
@@ -173,7 +202,6 @@ void canReturnTwilioMessagesUnparsed() throws Exception {
         final String conversationId = UUID.randomUUID().toString();
         final String messageId = UUID.randomUUID().toString();
         final String sourceConversationId = "+491234567";
-        final String text = "Hello World";
         final String sourceChannelId = "+497654321";
         final String token = "token";
 

backend/api/communication/src/test/java/co/airy/core/api/communication/SendMessageControllerTest.java

@@ -31,10 +31,13 @@
 import static co.airy.core.api.communication.util.Topics.applicationCommunicationChannels;
 import static co.airy.core.api.communication.util.Topics.applicationCommunicationMessages;
 import static co.airy.core.api.communication.util.Topics.getTopics;
+import static co.airy.test.Timing.retryOnException;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
+import static org.hamcrest.core.Is.is;
 import static org.junit.jupiter.api.Assertions.fail;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 
 @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, classes = AirySpringBootApplication.class)
@@ -86,6 +89,13 @@ void canSendTextMessages() throws Exception {
                         "\"message\":%s}",
                 conversationId, messagePayload);
 
+        retryOnException(
+                () -> webTestHelper.post("/conversations.info",
+                        "{\"conversation_id\":\"" + conversationId + "\"}")
+                        .andExpect(status().isOk()),
+                "Could not find conversation"
+        );
+
         final String response = webTestHelper.post("/messages.send", requestPayload)
                 .andExpect(status().isOk())
                 .andReturn().getResponse().getContentAsString();

backend/api/websocket/src/main/java/co/airy/core/api/websocket/WebSocketController.java

@@ -23,7 +23,9 @@ public class WebSocketController {
     }
 
     public void onMessage(Message message) {
-        messagingTemplate.convertAndSend(QUEUE_EVENTS, MessageEvent.fromMessage(message));
+        if (message != null) {
+            messagingTemplate.convertAndSend(QUEUE_EVENTS, MessageEvent.fromMessage(message));
+        }
     }
 
     public void onChannel(Channel channel) {

backend/media/BUILD

@@ -7,11 +7,9 @@ app_deps = [
     "//backend:base_app",
     "//backend/model/message",
     "//backend/model/metadata",
-    "//lib/java/mapping",
+    "//lib/java/uuid",
     "//lib/java/url",
-    "//lib/java/spring/kafka/core:spring-kafka-core",
-    "//lib/java/spring/kafka/streams:spring-kafka-streams",
-    "@maven//:javax_xml_bind_jaxb_api",
+    "//lib/java/spring/web:spring-web",
     "@maven//:org_springframework_retry_spring_retry",
     "@maven//:org_aspectj_aspectjweaver",
     "@maven//:com_amazonaws_aws_java_sdk_core",

backend/media/src/main/java/co/airy/core/media/MediaController.java

@@ -0,0 +1,61 @@
+package co.airy.core.media;
+
+import co.airy.core.media.services.MediaUpload;
+import co.airy.log.AiryLoggerFactory;
+import co.airy.spring.web.payload.RequestErrorResponsePayload;
+import co.airy.uuid.UUIDv5;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.slf4j.Logger;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.multipart.MultipartFile;
+
+import java.io.InputStream;
+
+@RestController
+public class MediaController {
+    private static final Logger log = AiryLoggerFactory.getLogger(MediaController.class);
+    private final MediaUpload mediaUpload;
+
+    public MediaController(MediaUpload mediaUpload) {
+        this.mediaUpload = mediaUpload;
+    }
+
+
+    @PostMapping(value = "/media.uploadFile", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
+    public ResponseEntity<?> mediaUpload(@RequestParam("file") MultipartFile multipartFile) {
+        final String originalFileName = multipartFile.getOriginalFilename();
+
+        if (originalFileName == null) {
+            return ResponseEntity.unprocessableEntity().body(new RequestErrorResponsePayload("Request is missing original file name"));
+        }
+
+        try {
+            final InputStream is = multipartFile.getInputStream();
+            String fileName = UUIDv5.fromFile(is).toString();
+
+            final String originalFileExtension = originalFileName.contains(".") ? originalFileName.substring(originalFileName.lastIndexOf(".")) : "";
+            fileName = fileName.concat(originalFileExtension);
+
+            return ResponseEntity.ok(new MediaUploadResponsePayload(mediaUpload.uploadMedia(is, fileName)));
+        } catch (Exception e) {
+            log.error("Media upload failed:", e);
+            return ResponseEntity.badRequest().body(new RequestErrorResponsePayload(String.format("Media Upload failed with error: %s", e.getMessage())));
+        }
+    }
+}
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+class MediaUploadResponsePayload {
+    private String mediaUrl;
+}
+