Merge branch 'release/0.5.0' into main

.github/workflows/main.yml

@@ -42,45 +42,14 @@ jobs:
 
       - name: Test
         run: |
-          "${GITHUB_WORKSPACE}/bin/bazel" test --test_tag_filters=-lint //...
+          bazel test --test_tag_filters=-lint //...
 
       - name: Build all artifacts
         run: |
-          "${GITHUB_WORKSPACE}/bin/bazel" build //...
+          bazel build //...
 
-      - name: Configure AWS credentials
-        uses: aws-actions/configure-aws-credentials@v1
-        with:
-          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
-          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
-          aws-region: us-east-1
-
-      - name: Push beta app images
-        if: ${{github.ref == 'refs/heads/develop'}}
-        run: |
-          echo ${{ secrets.PAT }} | docker login ghcr.io -u airydevci --password-stdin
-          release_targets=$(bazel query "filter("beta$", //...)" --output label)
-          for target in $release_targets; do
-            echo "Deploying $target"
-            bazel run $target
-          done
-
-      - name: Push release app images
-        if: startsWith(github.ref, 'refs/heads/release')
-        run: |
-          echo ${{ secrets.PAT }} | docker login ghcr.io -u airydevci --password-stdin
-          release_targets=$(bazel query "filter("release$", //...)" --output label)
-          for target in $release_targets; do
-            echo "Deploying $target"
-            bazel run $target
-          done
-
-      - name: Push release app images
-        if: ${{github.ref == 'refs/heads/main'}}
+      - name: Push images
+        if: startsWith(github.ref, 'refs/heads/release') || startsWith(github.ref, 'refs/heads/main') || startsWith(github.ref, 'refs/heads/develop')
         run: |
           echo ${{ secrets.PAT }} | docker login ghcr.io -u airydevci --password-stdin
-          release_targets=$(bazel query "filter("latest$", //...)" --output label)
-          for target in $release_targets; do
-            echo "Deploying $target"
-            bazel run $target
-          done
+          ./scripts/push-images.sh ${{ github.ref }}

BUILD

@@ -2,6 +2,7 @@ load("@rules_java//java:defs.bzl", "java_library", "java_plugin")
 load("@io_bazel_rules_docker//container:container.bzl", "container_image")
 load("@com_github_atlassian_bazel_tools//multirun:def.bzl", "multirun")
 load("@bazel_gazelle//:def.bzl", "gazelle")
+load("@io_bazel_rules_go//go:def.bzl", "TOOLS_NOGO", "nogo")
 
 package(default_visibility = ["//visibility:public"])
 
@@ -170,3 +171,10 @@ exports_files(
 # gazelle:build_file_name BUILD
 # gazelle:prefix
 gazelle(name = "gazelle")
+
+nogo(
+    name = "airy_nogo",
+    config = "//tools/build:nogo_config.json",
+    visibility = ["//visibility:public"],
+    deps = TOOLS_NOGO,
+)

VERSION

@@ -1 +1 @@
-0.4.0
+0.5.0

WORKSPACE

@@ -129,7 +129,7 @@ load("@io_bazel_rules_go//go:deps.bzl", "go_register_toolchains", "go_rules_depe
 
 go_rules_dependencies()
 
-go_register_toolchains()
+go_register_toolchains(nogo = "@//:airy_nogo")  # my_nogo is in the top-level BUILD file of this workspace
 
 load("@bazel_gazelle//:deps.bzl", "gazelle_dependencies")
 

backend/api/admin/BUILD

@@ -5,7 +5,7 @@ load("//tools/build:container_push.bzl", "container_push")
 app_deps = [
     "//backend:base_app",
     "//:springboot_actuator",
-    "//backend/model/channel:channel",
+    "//backend/model/channel",
     "//backend:tag",
     "//backend:webhook",
     "//lib/java/uuid",

backend/api/communication/BUILD

@@ -7,9 +7,9 @@ app_deps = [
     "//backend:base_app",
     "//:springboot_actuator",
     "//:springboot_websocket",
-    "//backend/model/message:message",
-    "//backend/model/channel:channel",
-    "//backend/model/metadata:metadata",
+    "//backend/model/message",
+    "//backend/model/channel",
+    "//backend/model/metadata",
     "//backend:read-receipt",
     "//lib/java/mapping",
     "//lib/java/date",

backend/media/BUILD

@@ -0,0 +1,41 @@
+load("//tools/build:springboot.bzl", "springboot")
+load("//tools/build:junit5.bzl", "junit5")
+load("//tools/build:container_push.bzl", "container_push")
+
+app_deps = [
+    "//backend:base_app",
+    "//backend/model/message",
+    "//backend/model/metadata",
+    "//lib/java/uuid",
+    "//lib/java/mapping",
+    "//lib/java/spring/kafka/core:spring-kafka-core",
+    "//lib/java/spring/kafka/streams:spring-kafka-streams",
+    "@maven//:io_lettuce_lettuce_core",
+    "@maven//:org_springframework_data_spring_data_redis",
+]
+
+springboot(
+    name = "media-resolver",
+    srcs = glob(["src/main/java/**/*.java"]),
+    main_class = "co.airy.spring.core.AirySpringBootApplication",
+    deps = app_deps,
+)
+
+[
+    junit5(
+        size = "medium",
+        file = file,
+        resources = glob(["src/test/resources/**/*"]),
+        deps = [
+            ":app",
+            "//backend:base_test",
+            "//lib/java/kafka/test:kafka-test",
+        ] + app_deps,
+    )
+    for file in glob(["src/test/java/**/*Test.java"])
+]
+
+container_push(
+    registry = "ghcr.io/airyhq/media",
+    repository = "resolver",
+)

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

@@ -0,0 +1,43 @@
+package co.airy.core.media;
+
+import co.airy.kafka.streams.KafkaStreamsWrapper;
+import co.airy.log.AiryLoggerFactory;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.slf4j.Logger;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.boot.context.event.ApplicationStartedEvent;
+import org.springframework.context.ApplicationListener;
+import org.springframework.stereotype.Component;
+
+@Component
+public class Resolver implements ApplicationListener<ApplicationStartedEvent>, DisposableBean {
+    private final Logger log = AiryLoggerFactory.getLogger(Resolver.class);
+
+    private static final String appId = "media.Resolver";
+    private final KafkaStreamsWrapper streams;
+
+    public Resolver(KafkaStreamsWrapper streams) {
+        this.streams = streams;
+    }
+
+    @Override
+    public void onApplicationEvent(ApplicationStartedEvent event) {
+        final StreamsBuilder builder = new StreamsBuilder();
+
+        streams.start(builder.build(), appId);
+    }
+
+    @Override
+    public void destroy() {
+        if (streams != null) {
+            streams.close();
+        }
+    }
+
+
+    // visible for testing
+    KafkaStreams.State getStreamState() {
+        return streams.state();
+    }
+}

backend/media/src/main/resources/application.properties

@@ -0,0 +1,4 @@
+kafka.brokers=${KAFKA_BROKERS}
+kafka.schema-registry-url=${KAFKA_SCHEMA_REGISTRY_URL}
+
+storage.host=${STORAGE_HOST}

backend/sources/chat-plugin/BUILD

@@ -8,8 +8,8 @@ app_deps = [
     "//:springboot_actuator",
     "//:springboot_websocket",
     "//:springboot_security",
-    "//backend/model/channel:channel",
-    "//backend/model/message:message",
+    "//backend/model/channel",
+    "//backend/model/message",
     "//lib/java/uuid",
     "//lib/java/date",
     "//lib/java/mapping",

backend/sources/facebook/connector/BUILD

@@ -4,9 +4,9 @@ load("//tools/build:container_push.bzl", "container_push")
 
 app_deps = [
     "//backend:base_app",
-    "//backend/model/channel:channel",
-    "//backend/model/message:message",
-    "//backend/model/metadata:metadata",
+    "//backend/model/channel",
+    "//backend/model/message",
+    "//backend/model/metadata",
     "//lib/java/uuid",
     "//lib/java/log",
     "//lib/java/spring/web:spring-web",

backend/sources/facebook/events-router/BUILD

@@ -4,8 +4,8 @@ load("//tools/build:container_push.bzl", "container_push")
 
 app_deps = [
     "//backend:base_app",
-    "//backend/model/channel:channel",
-    "//backend/model/message:message",
+    "//backend/model/channel",
+    "//backend/model/message",
     "//lib/java/uuid",
     "//lib/java/log",
     "//lib/java/kafka/schema:source-facebook-events",

backend/sources/facebook/events-router/src/main/java/co/airy/core/sources/facebook/MessageParser.java

@@ -53,6 +53,9 @@ public Message.Builder parse(final String payload) throws Exception {
         } else if (appId != null && !appId.equals(this.facebookAppId)) {
             senderType = SenderType.SOURCE_USER;
             senderId = appId;
+        } else if(isEcho && appId == null) {
+            senderType = SenderType.SOURCE_USER;
+            senderId = getSourceConversationId(webhookMessaging);
         } else {
             // Filter out echoes coming from this app
             throw new NotAMessageException();

backend/sources/facebook/events-router/src/test/java/co/airy/core/sources/facebook/EventsRouterTest.java

@@ -3,6 +3,7 @@
 import co.airy.avro.communication.Channel;
 import co.airy.avro.communication.ChannelConnectionState;
 import co.airy.avro.communication.Message;
+import co.airy.avro.communication.SenderType;
 import co.airy.kafka.schema.Topic;
 import co.airy.kafka.schema.application.ApplicationCommunicationChannels;
 import co.airy.kafka.schema.application.ApplicationCommunicationMessages;
@@ -34,6 +35,7 @@
 import java.util.concurrent.TimeUnit;
 
 import static co.airy.test.Timing.retryOnException;
+import static org.hamcrest.CoreMatchers.is;
 import static org.apache.kafka.streams.KafkaStreams.State.RUNNING;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
@@ -136,4 +138,33 @@ void joinsAndCountsMessagesCorrectly() throws Exception {
                 assertEquals(messages.stream().filter(m -> m.getConversationId().equals(conversationId)).count(), expectedCount.longValue())
         );
     }
+
+    @Test
+    void parsesPageMessagesCorrectly() throws Exception {
+        final String channelId = "channel-id";
+        final String pageId = "page-id";
+
+        final String payload = "{\"object\":\"page\",\"entry\":[{\"id\":\"%s\",\"time\":1609250136582," +
+                "\"messaging\":[{\"sender\":{\"id\":\"%s\"},\"recipient\":{\"id\":\"1912214878880084\"},\"timestamp\":1609250136503,\"message\":" +
+                "{\"mid\":\"<message_id>\",\"is_echo\":true,\"text\":\"text of the message\"}}]}]}";
+
+        kafkaTestHelper.produceRecord(new ProducerRecord<>(applicationCommunicationChannels.name(), channelId, Channel.newBuilder()
+                .setId(channelId)
+                .setConnectionState(ChannelConnectionState.CONNECTED)
+                .setSourceChannelId(pageId)
+                .setName("fb-page-a")
+                .setSource("facebook")
+                .setToken("")
+                .build()));
+
+        final String webhookPayload = String.format(payload, pageId, pageId);
+        kafkaTestHelper.produceRecord(new ProducerRecord<>(sourceFacebookEvents.name(), UUID.randomUUID().toString(), webhookPayload));
+
+        TimeUnit.SECONDS.sleep(5);
+        List<Message> messages = kafkaTestHelper.consumeValues(1, applicationCommunicationMessages.name());
+        assertThat(messages, hasSize(1));
+
+        Message message = messages.get(0);
+        assertThat(message.getSenderType(), is(SenderType.SOURCE_USER));
+    }
 }

backend/sources/google/connector/BUILD

@@ -4,8 +4,8 @@ load("//tools/build:container_push.bzl", "container_push")
 
 app_deps = [
     "//backend:base_app",
-    "//backend/model/channel:channel",
-    "//backend/model/message:message",
+    "//backend/model/channel",
+    "//backend/model/message",
     "//lib/java/spring/kafka/core:spring-kafka-core",
     "//lib/java/spring/kafka/streams:spring-kafka-streams",
     "//lib/java/uuid",

backend/sources/google/events-router/BUILD

@@ -4,9 +4,9 @@ load("//tools/build:container_push.bzl", "container_push")
 
 app_deps = [
     "//backend:base_app",
-    "//backend/model/channel:channel",
-    "//backend/model/message:message",
-    "//backend/model/metadata:metadata",
+    "//backend/model/channel",
+    "//backend/model/message",
+    "//backend/model/metadata",
     "//lib/java/uuid",
     "//lib/java/kafka/schema:source-google-events",
     "//lib/java/spring/kafka/core:spring-kafka-core",

backend/sources/twilio/connector/BUILD

@@ -4,8 +4,8 @@ load("//tools/build:container_push.bzl", "container_push")
 
 app_deps = [
     "//backend:base_app",
-    "//backend/model/channel:channel",
-    "//backend/model/message:message",
+    "//backend/model/channel",
+    "//backend/model/message",
     "//lib/java/log",
     "//lib/java/mapping",
     "//lib/java/uuid",

backend/sources/twilio/events-router/BUILD

@@ -4,8 +4,8 @@ load("//tools/build:container_push.bzl", "container_push")
 
 app_deps = [
     "//backend:base_app",
-    "//backend/model/channel:channel",
-    "//backend/model/message:message",
+    "//backend/model/channel",
+    "//backend/model/message",
     "//lib/java/uuid",
     "//lib/java/log",
     "//lib/java/kafka/schema:source-twilio-events",

backend/webhook/publisher/BUILD

@@ -4,7 +4,7 @@ load("//tools/build:container_push.bzl", "container_push")
 
 app_deps = [
     "//backend:base_app",
-    "//backend/model/message:message",
+    "//backend/model/message",
     "//backend:webhook",
     "//lib/java/uuid",
     "//lib/java/mapping",