[#413] Add /metadata.set endpoint

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

@@ -0,0 +1,42 @@
+package co.airy.core.api.communication;
+
+import co.airy.avro.communication.MetadataAction;
+import co.airy.avro.communication.MetadataActionType;
+import co.airy.core.api.communication.payload.SetMetadataRequestPayload;
+import co.airy.payload.response.EmptyResponsePayload;
+import co.airy.payload.response.RequestErrorResponsePayload;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RestController;
+
+import javax.validation.Valid;
+import java.time.Instant;
+
+@RestController
+public class MetadataController {
+    private static String USER_NAMESPACE = "public";
+    private final Stores stores;
+
+    public MetadataController(Stores stores) {
+        this.stores = stores;
+    }
+
+    @PostMapping("/metadata.set")
+    ResponseEntity<?> setMetadata(@RequestBody @Valid SetMetadataRequestPayload setMetadataRequestPayload) {
+        final MetadataAction metadataAction = MetadataAction.newBuilder()
+                .setActionType(MetadataActionType.SET)
+                .setTimestamp(Instant.now().toEpochMilli())
+                .setConversationId(setMetadataRequestPayload.getConversationId())
+                .setValue(setMetadataRequestPayload.getValue())
+                .setKey(USER_NAMESPACE + "." + setMetadataRequestPayload.getKey())
+                .build();
+        try {
+            stores.storeMetadata(metadataAction);
+        } catch (Exception e) {
+            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(new RequestErrorResponsePayload(e.getMessage()));
+        }
+        return ResponseEntity.ok(new EmptyResponsePayload());
+    }
+}

backend/api/communication/src/main/java/co/airy/core/api/communication/payload/SetMetadataRequestPayload.java

@@ -0,0 +1,21 @@
+package co.airy.core.api.communication.payload;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import javax.validation.constraints.NotNull;
+
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@Data
+public class SetMetadataRequestPayload {
+    @NotNull
+    private String conversationId;
+    @NotNull
+    private String key;
+    @NotNull
+    private String value;
+}

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

@@ -0,0 +1,94 @@
+package co.airy.core.api.communication;
+
+import co.airy.avro.communication.Channel;
+import co.airy.avro.communication.ChannelConnectionState;
+import co.airy.core.api.communication.util.TestConversation;
+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.application.ApplicationCommunicationReadReceipts;
+import co.airy.kafka.test.KafkaTestHelper;
+import co.airy.kafka.test.junit.SharedKafkaTestResource;
+import co.airy.spring.core.AirySpringBootApplication;
+import co.airy.spring.test.WebTestHelper;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+import java.util.UUID;
+
+import static co.airy.test.Timing.retryOnException;
+import static org.hamcrest.core.Is.is;
+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)
+@TestPropertySource(value = "classpath:test.properties")
+@ExtendWith(SpringExtension.class)
+@AutoConfigureMockMvc
+public class MetadataControllerTest {
+    @RegisterExtension
+    public static final SharedKafkaTestResource sharedKafkaTestResource = new SharedKafkaTestResource();
+
+    private static KafkaTestHelper kafkaTestHelper;
+
+    @Autowired
+    private WebTestHelper webTestHelper;
+
+    private static final ApplicationCommunicationMessages applicationCommunicationMessages = new ApplicationCommunicationMessages();
+    private static final ApplicationCommunicationChannels applicationCommunicationChannels = new ApplicationCommunicationChannels();
+    private static final ApplicationCommunicationMetadata applicationCommunicationMetadata = new ApplicationCommunicationMetadata();
+    private static final ApplicationCommunicationReadReceipts applicationCommunicationReadReceipts = new ApplicationCommunicationReadReceipts();
+
+    @BeforeAll
+    static void beforeAll() throws Exception {
+        kafkaTestHelper = new KafkaTestHelper(sharedKafkaTestResource,
+                applicationCommunicationMessages,
+                applicationCommunicationChannels,
+                applicationCommunicationMetadata,
+                applicationCommunicationReadReceipts);
+
+        kafkaTestHelper.beforeAll();
+    }
+
+    @AfterAll
+    static void afterAll() throws Exception {
+        kafkaTestHelper.afterAll();
+    }
+
+    @BeforeEach
+    void beforeEach() throws Exception {
+        webTestHelper.waitUntilHealthy();
+    }
+    @Test
+    void canSetMetadata() throws Exception {
+        final Channel channel = Channel.newBuilder()
+                .setConnectionState(ChannelConnectionState.CONNECTED)
+                .setId("channel-id")
+                .setName("channel-name")
+                .setSource("facebook")
+                .setSourceChannelId("ps-id")
+                .build();
+        final String conversationId = UUID.randomUUID().toString();
+
+        kafkaTestHelper.produceRecord(new ProducerRecord<>(applicationCommunicationChannels.name(), channel.getId(), channel));
+        kafkaTestHelper.produceRecords(TestConversation.generateRecords(conversationId, channel, 1));
+
+        retryOnException(
+                () -> webTestHelper.post("/metadata.set",
+                        "{\"conversation_id\":\"" + conversationId + "\", \"key\": \"awesome.key\", \"value\": \"awesome-value\"}",
+                        "user-id")
+                        .andExpect(status().isOk()),
+                "Error setting metadata"
+        );
+    }
+}

docs/docs/api/http.md

@@ -643,3 +643,23 @@ The response comes in two parts:
   - `total`
 
     The total number of elements across all pages.
+
+### Metadata
+
+Please refer to our [metadata](glossary.md#metadata) definition for more
+information.
+
+### Setting metadata
+
+
+`POST /metadata.set`
+
+```json
+{
+  "conversation_id": "conversation-id",
+  "key": "ad.id",
+  "value": "Grace"
+}
+```
+
+This endpoint returns `200` if the operation was successful and `400` otherwise.

docs/docs/glossary.md

@@ -121,7 +121,18 @@ Header data contains information that is important for downstream processing. It
 also includes the message preview and tags that are useful for certain apps like
 automations.
 
-
 # User
 
 A user represents one authorized agent in the Airy Core Platform.
+
+# Metadata
+
+Metadata is data attached to a conversation consisting of a set of Key/Value pairs. A key can use the dot notation to represent namespaces.
+
+e.g.
+
+| Key                        	| Value   	|
+|----------------------------	|---------	|
+| "sender.id"                	| "123A"  	|
+| "sender.contact.first_name 	| "Grace" 	|
+