[#3534] Whatsapp: Receive messages (#3614)

backend/model/metadata/src/main/java/co/airy/model/metadata/MetadataKeys.java

@@ -51,6 +51,7 @@ public static class MessageKeys {
 
         public static class Source {
             public static final String ID = "source.id";
+            // one of: delivered, seen
             public static final String DELIVERY_STATE = "source.delivery_state";
             public static final String ERROR = "source.error";
         }

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

@@ -89,7 +89,7 @@ void beforeEach() throws InterruptedException {
 
     // This tests simulates multiple users sending messages via multiple facebook pages
     // It ensures that we create the correct number of conversations and messages
-    //@Test
+    @Test
     void joinsAndCountsMessagesCorrectly() throws Exception {
         Random rand = new Random();
         List<String> pageIds = Arrays.asList("p1", "p2", "p3", "p4", "p5");

backend/sources/whatsapp/connector/src/main/java/co/airy/core/sources/whatsapp/ChannelsController.java

@@ -38,7 +38,7 @@ public ChannelsController(Api api, Stores stores) {
     @PostMapping("/channels.whatsapp.connect")
     ResponseEntity<?> connectWhatsapp(@RequestBody @Valid ConnectChannelRequestPayload payload) {
         final String token = payload.getUserToken();
-        final String phoneNumber = payload.getPhoneNumber();
+        final String phoneNumber = payload.getPhoneNumberId();
 
         final String channelId = UUIDv5.fromNamespaceAndName("whatsapp", phoneNumber).toString();
 

backend/sources/whatsapp/connector/src/main/java/co/airy/core/sources/whatsapp/payload/ConnectChannelRequestPayload.java

@@ -13,7 +13,7 @@
 @AllArgsConstructor
 public class ConnectChannelRequestPayload {
     @NotNull
-    private String phoneNumber;
+    private String phoneNumberId;
     @NotNull
     private String userToken;
     @NotNull

backend/sources/whatsapp/connector/src/test/java/co/airy/core/sources/whatsapp/ChannelsControllerTest.java

@@ -98,7 +98,7 @@ void canConnectWhatsappChannel() throws Exception {
         final String channelName = "My customer support phone number";
 
         final ConnectChannelRequestPayload connectPayload = ConnectChannelRequestPayload.builder()
-                .phoneNumber(phoneNumber)
+                .phoneNumberId(phoneNumber)
                 .name(channelName)
                 .userToken("user token")
                 .build();

backend/sources/whatsapp/events-router/BUILD

@@ -0,0 +1,45 @@
+load("@com_github_airyhq_bazel_tools//lint:buildifier.bzl", "check_pkg")
+load("//tools/build:springboot.bzl", "springboot")
+load("//tools/build:junit5.bzl", "junit5")
+load("//tools/build:container_release.bzl", "container_release")
+
+app_deps = [
+    "//backend:base_app",
+    "//:springboot_actuator",
+    "//backend/model/channel",
+    "//backend/model/message",
+    "//backend/model/metadata",
+    "//lib/java/uuid",
+    "//lib/java/log",
+    "//lib/java/kafka/schema:source-whatsapp-events",
+    "//lib/java/spring/kafka/core:spring-kafka-core",
+    "//lib/java/spring/kafka/streams:spring-kafka-streams",
+]
+
+springboot(
+    name = "events-router",
+    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_release(
+    registry = "ghcr.io/airyhq/sources",
+    repository = "whatsapp-events-router",
+)
+
+check_pkg(name = "buildifier")

backend/sources/whatsapp/events-router/src/main/java/co/airy/core/sources/whatsapp/EventsRouter.java

@@ -0,0 +1,156 @@
+package co.airy.core.sources.whatsapp;
+
+import co.airy.avro.communication.Channel;
+import co.airy.avro.communication.ChannelConnectionState;
+import co.airy.avro.communication.Message;
+import co.airy.avro.communication.Metadata;
+import co.airy.core.sources.whatsapp.dto.Event;
+import co.airy.core.sources.whatsapp.model.WebhookEntry.Change;
+import co.airy.core.sources.whatsapp.model.WebhookEvent;
+import co.airy.kafka.schema.application.ApplicationCommunicationChannels;
+import co.airy.kafka.schema.application.ApplicationCommunicationMessages;
+import co.airy.kafka.schema.application.ApplicationCommunicationMetadata;
+import co.airy.kafka.schema.source.SourceWhatsappEvents;
+import co.airy.kafka.streams.KafkaStreamsWrapper;
+import co.airy.log.AiryLoggerFactory;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.avro.specific.SpecificRecordBase;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.kstream.KTable;
+import org.slf4j.Logger;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.boot.actuate.health.Health;
+import org.springframework.boot.actuate.health.HealthIndicator;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.ApplicationListener;
+import org.springframework.stereotype.Component;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Stream;
+
+import static java.util.stream.Collectors.toList;
+
+@Component
+public class EventsRouter implements HealthIndicator, DisposableBean, ApplicationListener<ApplicationReadyEvent> {
+    private static final Logger log = AiryLoggerFactory.getLogger(EventsRouter.class);
+
+    private final String metadataStore = "metadata-store";
+    private final KafkaStreamsWrapper streams;
+    private final ObjectMapper objectMapper;
+    private final MessageMapper messageMapper;
+    private final KafkaProducer<String, SpecificRecordBase> kafkaProducer;
+
+    EventsRouter(KafkaStreamsWrapper streams, ObjectMapper objectMapper, MessageMapper messageMapper, KafkaProducer<String, SpecificRecordBase> kafkaProducer) {
+        this.streams = streams;
+        this.objectMapper = objectMapper;
+        this.messageMapper = messageMapper;
+        this.kafkaProducer = kafkaProducer;
+    }
+
+    private static final String appId = "sources.whatsapp.EventsRouter";
+
+    private final String applicationCommunicationMetadata = new ApplicationCommunicationMetadata().name();
+    private final String applicationCommunicationMessages = new ApplicationCommunicationMessages().name();
+
+
+    public void startStream() {
+        final StreamsBuilder builder = new StreamsBuilder();
+
+        // Channels table
+        KTable<String, Channel> channelsTable = builder.<String, Channel>stream(new ApplicationCommunicationChannels().name())
+                .groupBy((k, v) -> v.getSourceChannelId())
+                .reduce((aggValue, newValue) -> newValue)
+                .filter((sourceChannelId, channel) -> "whatsapp".equals(channel.getSource())
+                        && channel.getConnectionState().equals(ChannelConnectionState.CONNECTED));
+
+        builder.<String, String>stream(new SourceWhatsappEvents().name())
+                .flatMap((key, event) -> {
+                    WebhookEvent webhookEvent;
+                    try {
+                        webhookEvent = objectMapper.readValue(event, WebhookEvent.class);
+                        if (webhookEvent.getEntries() == null) {
+                            log.warn("empty entries. key={} event={}", key, event);
+                            return Collections.emptyList();
+                        }
+                    } catch (Exception e) {
+                        log.warn("error in record. key={} event={} e={}", key, event, e.toString());
+                        return Collections.emptyList();
+                    }
+
+                    return webhookEvent.getEntries()
+                            .stream()
+                            .flatMap(entry -> {
+                                final List<Change> changes = entry.getChanges();
+
+                                if (changes == null) {
+                                    return Stream.empty();
+                                }
+
+                                return changes.stream().map(change -> {
+                                    try {
+                                        final String sourceChannelId = change.getValue().getMetadata().getPhoneNumberId();
+                                        return KeyValue.pair(sourceChannelId, Event.builder().payload(change).build()
+                                        );
+                                    } catch (Exception e) {
+                                        log.warn("Skipping whatsapp error for record " + entry, e);
+                                        return null;
+                                    }
+                                });
+                            })
+                            .filter(Objects::nonNull)
+                            .collect(toList());
+                })
+                .join(channelsTable, (event, channel) -> event.toBuilder().channel(channel).build())
+                .flatMap((sourceChannelId, event) -> {
+                    try {
+                        return messageMapper.getRecords(event);
+                    }  catch (Exception e) {
+                        log.warn("skip whatsapp record for error: " + event.toString(), e);
+                        return List.of();
+                    }
+                })
+                .to((recordId, record, context) -> {
+                    if (record instanceof Metadata) {
+                        return applicationCommunicationMetadata;
+                    }
+                    if (record instanceof Message) {
+                        return applicationCommunicationMessages;
+                    }
+
+                    throw new IllegalStateException("Unknown type for record " + record);
+                });
+
+
+        streams.start(builder.build(), appId);
+    }
+
+    @Override
+    public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
+        startStream();
+    }
+
+    @Override
+    public void destroy() {
+        if (streams != null) {
+            streams.close();
+        }
+    }
+
+    @Override
+    public Health health() {
+        if (streams == null || !streams.state().isRunningOrRebalancing()) {
+            return Health.down().build();
+        }
+        return Health.up().build();
+    }
+
+    // visible for testing
+    KafkaStreams.State getStreamState() {
+        return streams.state();
+    }
+}

backend/sources/whatsapp/events-router/src/main/java/co/airy/core/sources/whatsapp/MessageMapper.java

@@ -0,0 +1,77 @@
+package co.airy.core.sources.whatsapp;
+
+import co.airy.avro.communication.Channel;
+import co.airy.avro.communication.DeliveryState;
+import co.airy.avro.communication.Message;
+import co.airy.avro.communication.Metadata;
+import co.airy.core.sources.whatsapp.dto.Event;
+import co.airy.core.sources.whatsapp.model.Value;
+import co.airy.core.sources.whatsapp.model.WebhookEntry;
+import co.airy.log.AiryLoggerFactory;
+import co.airy.model.metadata.MetadataKeys;
+import co.airy.uuid.UUIDv5;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.avro.specific.SpecificRecordBase;
+import org.apache.kafka.streams.KeyValue;
+import org.slf4j.Logger;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static co.airy.model.metadata.MetadataRepository.getId;
+import static co.airy.model.metadata.MetadataRepository.newConversationMetadata;
+
+@Component
+public class MessageMapper {
+    private static final Logger log = AiryLoggerFactory.getLogger(MessageMapper.class);
+
+    public List<KeyValue<String, SpecificRecordBase>> getRecords(Event event) {
+        final WebhookEntry.Change change = event.getPayload();
+        if(!"messages".equals(change.getField())) {
+            // TODO implement remaining fields
+            return List.of();
+        }
+
+        final Value value = change.getValue();
+
+        List<KeyValue<String, SpecificRecordBase>> results = new ArrayList<>();
+        for (Value.Contact contact : value.getContacts()) {
+            final String conversationId = getConversationId(event.getChannel(), contact.getWaId());
+            final Metadata metadata = newConversationMetadata(conversationId, MetadataKeys.ConversationKeys.Contact.DISPLAY_NAME, contact.getProfile().getName());
+            results.add(KeyValue.pair(getId(metadata).toString(), metadata));
+        }
+
+        for (JsonNode message : value.getMessages()) {
+            try {
+                final String sourceConversationId = message.get("from").textValue();
+                final String conversationId = getConversationId(event.getChannel(), sourceConversationId);
+                final String id = message.get("id").textValue();
+                final long timestamp = message.get("timestamp").asLong() * 1000;
+
+                final Message airyMessage = Message.newBuilder()
+                        .setChannelId(event.getChannel().getId())
+                        .setContent(message.toString())
+                        .setHeaders(Map.of())
+                        .setId(UUIDv5.fromName(id).toString())
+                        .setConversationId(conversationId)
+                        .setIsFromContact(true)
+                        .setDeliveryState(DeliveryState.DELIVERED)
+                        .setSenderId(sourceConversationId)
+                        .setSource("whatsapp")
+                        .setSentAt(timestamp)
+                        .build();
+                results.add(KeyValue.pair(airyMessage.getId(), airyMessage));
+            } catch (Exception e) {
+                log.error("Error mapping message", e);
+            }
+        }
+
+        return results;
+    }
+
+    private String getConversationId(Channel channel, String sourceConversationId) {
+        return UUIDv5.fromNamespaceAndName(channel.getId(), sourceConversationId).toString();
+    }
+}

backend/sources/whatsapp/events-router/src/main/java/co/airy/core/sources/whatsapp/dto/Event.java

@@ -0,0 +1,19 @@
+package co.airy.core.sources.whatsapp.dto;
+
+import co.airy.avro.communication.Channel;
+import co.airy.core.sources.whatsapp.model.WebhookEntry;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.io.Serializable;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@Builder(toBuilder = true)
+public class Event implements Serializable {
+    private WebhookEntry.Change payload;
+    private Channel channel;
+}

backend/sources/whatsapp/events-router/src/main/java/co/airy/core/sources/whatsapp/model/Value.java

@@ -0,0 +1,44 @@
+package co.airy.core.sources.whatsapp.model;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+// See https://developers.facebook.com/docs/whatsapp/cloud-api/webhooks/payload-examples
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class Value {
+    private String messagingProduct;
+    private Metadata metadata;
+    private List<Contact> contacts;
+    private List<JsonNode> messages;
+
+    private JsonNode statuses;
+
+    @Data
+    @NoArgsConstructor
+    @AllArgsConstructor
+    public static class Metadata {
+        private String displayPhoneNumber;
+        private String phoneNumberId;
+    }
+
+    @Data
+    @NoArgsConstructor
+    @AllArgsConstructor
+    public static class Contact {
+        private Profile profile;
+        private String waId;
+
+        @Data
+        @NoArgsConstructor
+        @AllArgsConstructor
+        public static class Profile {
+            private String name;
+        }
+    }
+}