Fix #6814 Backend: Feed API response doesn't include all tasks of user (#7284)

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

@@ -777,7 +777,7 @@ public interface CollectionDAO {
         connectionType = POSTGRES)
     @ConnectionAwareSqlQuery(
         value =
-            "SELECT count(id) FROM thread_entity WHERE "
+            "SELECT count(id) FROM thread_entity WHERE type='Task' AND "
                 + "(:status IS NULL OR taskStatus = :status) AND "
                 + "JSON_OVERLAPS(taskAssignees, :userTeamJsonMysql) ",
         connectionType = MYSQL)
@@ -786,6 +786,70 @@ public interface CollectionDAO {
         @Bind("userTeamJsonMysql") String userTeamJsonMysql,
         @Bind("status") TaskStatus status);
 
+    @ConnectionAwareSqlQuery(
+        value =
+            "SELECT json FROM thread_entity WHERE type='Task' AND updatedAt > :before AND taskStatus = :status AND "
+                + "AND (taskAssignees @> ANY (ARRAY[<userTeamJsonPostgres>]::jsonb[]) OR createdBy = :username) "
+                + "ORDER BY createdAt DESC "
+                + "LIMIT :limit",
+        connectionType = POSTGRES)
+    @ConnectionAwareSqlQuery(
+        value =
+            "SELECT json FROM thread_entity WHERE type='Task' AND updatedAt > :before AND taskStatus = :status AND "
+                + "AND (JSON_OVERLAPS(taskAssignees, :userTeamJsonMysql) OR createdBy = :username) "
+                + "ORDER BY createdAt DESC "
+                + "LIMIT :limit",
+        connectionType = MYSQL)
+    List<String> listTasksOfUserBefore(
+        @BindList("userTeamJsonPostgres") List<String> userTeamJsonPostgres,
+        @Bind("userTeamJsonMysql") String userTeamJsonMysql,
+        @Bind("username") String username,
+        @Bind("limit") int limit,
+        @Bind("before") long before,
+        @Bind("status") TaskStatus status);
+
+    @ConnectionAwareSqlQuery(
+        value =
+            "SELECT json FROM thread_entity WHERE type='Task' AND updatedAt < :after "
+                + "AND (:status IS NULL OR taskStatus = :status) "
+                + "AND (taskAssignees @> ANY (ARRAY[<userTeamJsonPostgres>]::jsonb[]) OR createdBy = :username) "
+                + "ORDER BY createdAt DESC "
+                + "LIMIT :limit",
+        connectionType = POSTGRES)
+    @ConnectionAwareSqlQuery(
+        value =
+            "SELECT json FROM thread_entity WHERE type='Task' AND updatedAt < :after "
+                + "AND (:status IS NULL OR taskStatus = :status) "
+                + "AND (JSON_OVERLAPS(taskAssignees, :userTeamJsonMysql) OR createdBy = :username) "
+                + "ORDER BY createdAt DESC "
+                + "LIMIT :limit",
+        connectionType = MYSQL)
+    List<String> listTasksOfUserAfter(
+        @BindList("userTeamJsonPostgres") List<String> userTeamJsonPostgres,
+        @Bind("userTeamJsonMysql") String userTeamJsonMysql,
+        @Bind("username") String username,
+        @Bind("limit") int limit,
+        @Bind("after") long after,
+        @Bind("status") TaskStatus status);
+
+    @ConnectionAwareSqlQuery(
+        value =
+            "SELECT count(id) FROM thread_entity WHERE type='Task' "
+                + "AND (:status IS NULL OR taskStatus = :status) "
+                + "AND (taskAssignees @> ANY (ARRAY[<userTeamJsonPostgres>]::jsonb[]) OR createdBy = :username) ",
+        connectionType = POSTGRES)
+    @ConnectionAwareSqlQuery(
+        value =
+            "SELECT count(id) FROM thread_entity WHERE type='Task' "
+                + "AND (:status IS NULL OR taskStatus = :status) "
+                + "AND (JSON_OVERLAPS(taskAssignees, :userTeamJsonMysql) OR createdBy = :username) ",
+        connectionType = MYSQL)
+    int listCountTasksOfUser(
+        @BindList("userTeamJsonPostgres") List<String> userTeamJsonPostgres,
+        @Bind("userTeamJsonMysql") String userTeamJsonMysql,
+        @Bind("username") String username,
+        @Bind("status") TaskStatus status);
+
     @SqlQuery(
         "SELECT json FROM thread_entity WHERE type='Task' AND updatedAt > :before "
             + "AND (:status IS NULL OR taskStatus = :status) "

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

@@ -703,9 +703,11 @@ public class FeedRepository {
           // Only two filter types are supported for tasks -> ASSIGNED_TO, ASSIGNED_BY
           if (filterType == FilterType.ASSIGNED_BY) {
             filteredThreads = getTasksAssignedBy(userId, limit + 1, time, taskStatus, paginationType);
-          } else {
+          } else if (filterType == FilterType.ASSIGNED_TO) {
-            // make ASSIGNED_TO a default filter
             filteredThreads = getTasksAssignedTo(userId, limit + 1, time, taskStatus, paginationType);
+          } else {
+            // Get all the tasks assigned to or created by the user
+            filteredThreads = getTasksOfUser(userId, limit + 1, time, taskStatus, paginationType);
           }
         } else {
           if (filterType == FilterType.FOLLOWS) {
@@ -999,6 +1001,28 @@ public class FeedRepository {
     return thread;
   }
 
+  /** Return the tasks created by or assigned to the user. */
+  private FilteredThreads getTasksOfUser(
+      String userId, int limit, long time, TaskStatus status, PaginationType paginationType) throws IOException {
+    User user = dao.userDAO().findEntityById(UUID.fromString(userId));
+    String username = user.getName();
+    List<String> teamIds = getTeamIds(userId);
+    List<String> userTeamJsonPostgres = getUserTeamJsonPostgres(userId, teamIds);
+    String userTeamJsonMysql = getUserTeamJsonMysql(userId, teamIds);
+    List<String> jsons;
+    if (paginationType == PaginationType.BEFORE) {
+      jsons =
+          dao.feedDAO().listTasksOfUserBefore(userTeamJsonPostgres, userTeamJsonMysql, username, limit, time, status);
+    } else {
+      jsons =
+          dao.feedDAO().listTasksOfUserAfter(userTeamJsonPostgres, userTeamJsonMysql, username, limit, time, status);
+    }
+    List<Thread> threads = JsonUtils.readObjects(jsons, Thread.class);
+    int totalCount = dao.feedDAO().listCountTasksOfUser(userTeamJsonPostgres, userTeamJsonMysql, username, status);
+    sortPostsInThreads(threads);
+    return new FilteredThreads(threads, totalCount);
+  }
+
   /** Return the tasks created by the user. */
   private FilteredThreads getTasksAssignedBy(
       String userId, int limit, long time, TaskStatus status, PaginationType paginationType) throws IOException {

catalog-rest-service/src/test/java/org/openmetadata/catalog/resources/feeds/FeedResourceTest.java

@@ -401,6 +401,10 @@ public class FeedResourceTest extends CatalogApplicationTest {
     assertEquals(assignedToCount + 1, tasks.getPaging().getTotal());
     assertEquals(assignedToCount + 1, tasks.getData().size());
 
+    tasks = listTasks(null, USER.getId().toString(), null, null, null, ADMIN_AUTH_HEADERS);
+    assertEquals(assignedToCount + assignedByCount + 2, tasks.getPaging().getTotal());
+    assertEquals(assignedToCount + assignedByCount + 2, tasks.getData().size());
+
     ThreadCount count = listTasksCount(null, TaskStatus.Open, userAuthHeaders);
     int totalOpenTaskCount = count.getTotalCount();
     count = listTasksCount(null, TaskStatus.Closed, userAuthHeaders);