Fix #13342: The activity feed does not reflect changes to the objects that the user is subscribed to (#13359)

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

@@ -1174,6 +1174,36 @@ public interface CollectionDAO {
         @Bind("relation") int relation,
         @Define("condition") String condition);
 
+    @SqlQuery(
+        "SELECT json FROM thread_entity <condition> AND "
+            // Entity for which the thread is about is owned by the user or his teams
+            + "(entityId in (SELECT toId FROM entity_relationship WHERE "
+            + "((fromEntity='user' AND fromId= :userId) OR "
+            + "(fromEntity='team' AND fromId IN (<teamIds>))) AND relation=8) OR "
+            + "id in (SELECT toId FROM entity_relationship WHERE (fromEntity='user' AND fromId= :userId AND toEntity='THREAD' AND relation IN (1,2))) "
+            + " OR id in (SELECT toId FROM entity_relationship WHERE ((fromEntity='user' AND fromId= :userId) OR "
+            + "(fromEntity='team' AND fromId IN (<teamIds>))) AND relation=11)) "
+            + "ORDER BY createdAt DESC "
+            + "LIMIT :limit")
+    List<String> listThreadsByOwnerOrFollows(
+        @BindUUID("userId") UUID userId,
+        @BindList("teamIds") List<String> teamIds,
+        @Bind("limit") int limit,
+        @Define("condition") String condition);
+
+    @SqlQuery(
+        "SELECT count(id) FROM thread_entity <condition> AND "
+            + "(entityId in (SELECT toId FROM entity_relationship WHERE "
+            + "((fromEntity='user' AND fromId= :userId) OR "
+            + "(fromEntity='team' AND fromId IN (<teamIds>))) AND relation=8) OR "
+            + "id in (SELECT toId FROM entity_relationship WHERE (fromEntity='user' AND fromId= :userId AND toEntity='THREAD' AND relation IN (1,2))) "
+            + " OR id in (SELECT toId FROM entity_relationship WHERE ((fromEntity='user' AND fromId= :userId) OR "
+            + "(fromEntity='team' AND fromId IN (<teamIds>))) AND relation=11))")
+    int listCountThreadsByOwnerOrFollows(
+        @BindUUID("userId") UUID userId,
+        @BindList("teamIds") List<String> teamIds,
+        @Define("condition") String condition);
+
     @SqlQuery(
         "SELECT json FROM thread_entity <condition> AND "
             + "MD5(id) in ("

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

@@ -115,7 +115,8 @@ public class FeedRepository {
     MENTIONS,
     FOLLOWS,
     ASSIGNED_TO,
-    ASSIGNED_BY
+    ASSIGNED_BY,
+    OWNER_OR_FOLLOWS
   }
 
   public enum PaginationType {
@@ -577,6 +578,8 @@ public class FeedRepository {
             filteredThreads = getThreadsByFollows(filter, userId, limit + 1);
           } else if (FilterType.MENTIONS.equals(filter.getFilterType())) {
             filteredThreads = getThreadsByMentions(filter, userId, limit + 1);
+          } else if (FilterType.OWNER_OR_FOLLOWS.equals(filter.getFilterType())) {
+            filteredThreads = getThreadsByOwnerOrFollows(filter, userId, limit + 1);
           } else {
             filteredThreads = getThreadsByOwner(filter, userId, limit + 1);
           }
@@ -969,6 +972,14 @@ public class FeedRepository {
     return new FilteredThreads(threads, totalCount);
   }
 
+  private FilteredThreads getThreadsByOwnerOrFollows(FeedFilter filter, UUID userId, int limit) {
+    List<String> teamIds = getTeamIds(userId);
+    List<String> jsons = dao.feedDAO().listThreadsByOwnerOrFollows(userId, teamIds, limit, filter.getCondition());
+    List<Thread> threads = JsonUtils.readObjects(jsons, Thread.class);
+    int totalCount = dao.feedDAO().listCountThreadsByOwnerOrFollows(userId, teamIds, filter.getCondition());
+    return new FilteredThreads(threads, totalCount);
+  }
+
   /** Get a list of team names that the given user is a part of. */
   private List<String> getTeamNames(User user) {
     List<String> teamNames = null;

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

@@ -996,6 +996,50 @@ public class FeedResourceTest extends OpenMetadataApplicationTest {
     assertEquals(totalThreadCount + 1, threads.getPaging().getTotal());
   }
 
+  @Test
+  void list_threadsWithOwnerOrFollowerFilter() throws HttpResponseException {
+    int totalThreadCount = listThreads(null, null, ADMIN_AUTH_HEADERS).getPaging().getTotal();
+    String user1 = USER1.getId().toString(); // user1 is the owner of TABLE
+    // Get thread counts for user1 and user2
+    int user1ThreadCount = listThreadsWithFilter(user1, FilterType.OWNER, USER_AUTH_HEADERS).getPaging().getTotal();
+
+    // create another thread on an entity with team2 as owner
+    String team2 = TABLE2.getOwner().getId().toString();
+    assertNotEquals(user1, team2);
+    createAndCheck(
+        create().withAbout(String.format("<#E::table::%s>", TABLE2.getFullyQualifiedName())).withFrom(ADMIN_USER_NAME),
+        ADMIN_AUTH_HEADERS);
+
+    // user1 thread count remains the same as the newly created thread belongs to team2 and user1 is not part of it
+    ThreadList threads = listThreadsWithFilter(user1, FilterType.OWNER, USER_AUTH_HEADERS);
+    assertEquals(user1ThreadCount, threads.getPaging().getTotal());
+
+    String entityLink = String.format("<#E::table::%s>", TABLE2.getFullyQualifiedName());
+    int initialThreadCount = listThreads(entityLink, null, USER_AUTH_HEADERS).getPaging().getTotal();
+
+    // Create threads
+    createAndCheck(create().withMessage("Message 1").withAbout(entityLink), ADMIN_AUTH_HEADERS);
+
+    createAndCheck(create().withMessage("Message 2").withAbout(entityLink), ADMIN_AUTH_HEADERS);
+
+    // Make the USER follow TABLE2
+    followTable(TABLE2.getId(), USER1.getId(), USER_AUTH_HEADERS);
+    with()
+        .pollInterval(ONE_SECOND)
+        .await("Threads With Follows")
+        .until(
+            () -> {
+              ThreadList followThreads =
+                  listThreadsWithFilter(USER.getId().toString(), FilterType.FOLLOWS, USER_AUTH_HEADERS);
+              return followThreads.getPaging().getTotal().equals(initialThreadCount + 3);
+            });
+
+    // filter by OWNER_OR_FOLLOWS we should list both team owned listing and followed table threads.
+    ThreadList ownerOrFollowTreads =
+        listThreadsWithFilter(USER.getId().toString(), FilterType.OWNER_OR_FOLLOWS, USER_AUTH_HEADERS);
+    assertEquals(threads.getPaging().getTotal() + 3, ownerOrFollowTreads.getPaging().getTotal());
+  }
+
   @Test
   void list_threadsWithMentionsFilter() throws HttpResponseException {
     // Create a thread with user mention