Fix #2718: Update entity_thread schema to store activity feeds (#2726)

bootstrap/sql/mysql/v004__create_db_connection_info.sql

@@ -0,0 +1,17 @@
+DROP TABLE IF EXISTS thread_entity;
+
+CREATE TABLE IF NOT EXISTS thread_entity (
+    id VARCHAR(36) GENERATED ALWAYS AS (json ->> '$.id') STORED NOT NULL,
+    entityLink VARCHAR(256) GENERATED ALWAYS AS (json ->> '$.about') NOT NULL,
+    assignedTo VARCHAR(256) GENERATED ALWAYS AS (json ->> '$.addressedTo'),
+    json JSON NOT NULL,
+    createdAt BIGINT UNSIGNED GENERATED ALWAYS AS (json ->> '$.threadTs') STORED NOT NULL,
+    createdBy VARCHAR(256) GENERATED ALWAYS AS (json ->> '$.createdBy') STORED NOT NULL,
+    updatedAt BIGINT UNSIGNED GENERATED ALWAYS AS (json ->> '$.updatedAt') NOT NULL,
+    updatedBy VARCHAR(256) GENERATED ALWAYS AS (json ->> '$.updatedBy') NOT NULL,
+    resolved BOOLEAN GENERATED ALWAYS AS (JSON_EXTRACT(json, '$.resolved')),
+    PRIMARY KEY (id),
+    INDEX (updatedBy),
+    INDEX (updatedAt)
+);
+

catalog-rest-service/src/main/java/org/openmetadata/catalog/jdbi3/FeedRepository.java

@@ -198,7 +198,7 @@ public class FeedRepository {
     Set<String> uniqueValues = new HashSet<>();
     for (String t : threadIds) {
       // If an entity has multiple relationships (created, mentioned, repliedTo etc.) to the same thread
-      // Don't sent duplicated copies of the thread in response
+      // Don't send duplicated copies of the thread in response
       if (uniqueValues.add(t)) {
         threads.add(EntityUtil.validate(t, dao.feedDAO().findById(t), Thread.class));
       }

catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/feeds/FeedResource.java

@@ -35,6 +35,7 @@ import javax.ws.rs.QueryParam;
 import javax.ws.rs.core.Context;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
+import javax.ws.rs.core.SecurityContext;
 import javax.ws.rs.core.UriInfo;
 import org.openmetadata.catalog.api.feed.CreateThread;
 import org.openmetadata.catalog.entity.feed.Thread;
@@ -131,12 +132,10 @@ public class FeedResource {
             content = @Content(mediaType = "application/json", schema = @Schema(implementation = CreateThread.class))),
         @ApiResponse(responseCode = "400", description = "Bad request")
       })
-  public Response create(@Context UriInfo uriInfo, @Valid CreateThread cr) throws IOException, ParseException {
+  public Response create(@Context UriInfo uriInfo, @Context SecurityContext securityContext, @Valid CreateThread create)
-    Thread thread =
+      throws IOException, ParseException {
-        new Thread().withId(UUID.randomUUID()).withThreadTs(System.currentTimeMillis()).withAbout(cr.getAbout());
+    Thread thread = getThread(securityContext, create);
-    // For now redundantly storing everything in json (that includes fromEntity, addressedTo entity)
+    FeedUtil.addPost(thread, new Post().withMessage(create.getMessage()).withFrom(create.getFrom()));
-    // TODO - This needs cleanup later if this information is too much or inconsistent in relationship table
-    FeedUtil.addPost(thread, new Post().withMessage(cr.getMessage()).withFrom(cr.getFrom()));
     addHref(uriInfo, dao.create(thread));
     return Response.created(thread.getHref()).entity(thread).build();
   }
@@ -158,4 +157,15 @@ public class FeedResource {
     Thread thread = addHref(uriInfo, dao.addPostToThread(id, post));
     return Response.created(thread.getHref()).entity(thread).build();
   }
+
+  private Thread getThread(SecurityContext securityContext, CreateThread create) {
+    return new Thread()
+        .withId(UUID.randomUUID())
+        .withThreadTs(System.currentTimeMillis())
+        .withCreatedBy(securityContext.getUserPrincipal().getName())
+        .withAbout(create.getAbout())
+        .withAddressedTo(create.getAddressedTo())
+        .withUpdatedBy(securityContext.getUserPrincipal().getName())
+        .withUpdatedAt(System.currentTimeMillis());
+  }
 }

catalog-rest-service/src/main/resources/json/schema/api/feed/createThread.json

@@ -13,6 +13,10 @@
       "description": "ID of User (regular user or bot) posting the message",
       "$ref": "../../type/basic.json#/definitions/uuid"
     },
+    "addressedTo": {
+      "description": "User or team this thread is addressed to in format <#E/{entities}/{entityName}/{field}/{fieldValue}.",
+      "$ref": "../../type/basic.json#/definitions/entityLink"
+    },
     "about": {
       "description": "Data asset about which this thread is created for with format <#E/{entities}/{entityType}/{field}/{fieldValue}",
       "$ref": "../../type/basic.json#/definitions/entityLink"

catalog-rest-service/src/main/resources/json/schema/entity/feed/thread.json

@@ -48,6 +48,23 @@
       "description": "User or team this thread is addressed to in format <#E/{entities}/{entityName}/{field}/{fieldValue}.",
       "$ref": "../../type/basic.json#/definitions/entityLink"
     },
+    "createdBy": {
+      "description": "User who created the thread.",
+      "type": "string"
+    },
+    "updatedAt": {
+      "description": "Last update time corresponding to the new version of the entity in Unix epoch time milliseconds.",
+      "$ref": "../../type/basic.json#/definitions/timestamp"
+    },
+    "updatedBy": {
+      "description": "User who made the update.",
+      "type": "string"
+    },
+    "resolved": {
+      "description": "When `true` indicates the thread has been resolved",
+      "type": "boolean",
+      "default": false
+    },
     "posts": {
       "type": "array",
       "items": {