refactor: replace perspective.addListener('link-*') with SPARQL subscriptions

app/src/composables/useCommunityService.ts

@@ -141,8 +141,8 @@
   // Community query is perspective-scoped (typically one per perspective — low cost).
   // Use ChannelSummary — lightweight model without @HasMany relations.
   // Property getters run by default (deepQuery=true) via batched VALUES queries.
   const { data: communities, loading: communitiesLoading, error: communitiesError } = useLiveQuery(Community, perspective);
   const { data: allChannels, loading: channelsLoading, error: channelsError } = useLiveQuery(ChannelSummary, perspective);
 
   // Cache for conversation instances — populated during data fetching, looked up in computeds.
   // Plain Map (not reactive) is sufficient: updates always precede the ref changes that trigger re-computation.
@@ -163,8 +163,8 @@
 
   const isAuthor = computed(() => communities.value[0]?.author === me.value.did);
   const community = computed<Community>(() => communities.value[0]);
   const pinnedChannels = computed(() => allChannels.value.filter((channel) => channel.isPinned));
   const conversationChannels = computed(() => allChannels.value.filter((channel) => channel.isConversation));
   const spaceChannels = computed(() => allChannels.value.filter((channel) => !channel.isConversation));
   const pinnedConversationsWithAgents = computed((): ChannelDataWithAgents[] => {
     return pinnedConversations.value.map((data) => ({
@@ -485,41 +485,80 @@
   perspective.addSyncStateChangeListener(syncStateListener);
 
   // Track channel participants automatically.
-  // Uses addListener because participant tracking needs link.author metadata,
-  // which isn't available from SPARQL subscription results.
-  // Local dedup set avoids redundant addLinks RPCs — ChannelSummary doesn't
-  // carry participants, so the old channel.participants.includes() check was lost.
+  //
+  // Previously this used `perspective.addListener('link-added', ...)`, which
+  // is a perspective-scoped firehose — every link added anywhere in the
+  // perspective wakes every listener.  We now drive the same logic from a
+  // targeted SPARQL subscription that only fires when the set of distinct
+  // `(channel, author)` pairs over the `flux://has_channel` predicate
+  // changes.  The reified link's author is exposed via
+  // `?_r ad4m:author ?author`, so the subscription observes exactly the
+  // signal we used to read off `link.author`.
+  //
+  // The dedup set keeps `addLinks` RPCs idempotent (the addLink itself is
+  // also idempotent on the executor side, but the dedup avoids the round
+  // trip).  Each subscription fire triggers a single `LinkQuery` refetch so
+  // the handler does not depend on the SPARQL binding shape — robust
+  // against any future result-shape changes.
   const knownParticipants = new Set<string>();
-  function handleParticipantTracking(link: any) {
-    if (link.data.predicate !== CHANNEL) return null;
-    if (!link.author) return null;
-
-    const channelId = link.data.source;
-    const channel = allChannels.value.find((c) => c.id === channelId);
-    if (!channel) return null;
-
-    const key = `${channelId}::${link.author}`;
-    if (knownParticipants.has(key)) return null;
-    knownParticipants.add(key);
-
-    perspective
-      .addLinks([{ source: channelId, predicate: 'flux://has_participant', target: link.author }])
-      .catch((error) => {
-        knownParticipants.delete(key);
-        console.error('Failed to add participant to channel:', {
-          channelId,
-          author: link.author,
-          error,
-        });
-      });
+  let channelLinksSub: { dispose: () => void } | null = null;
+  let participantTrackingCancelled = false;
 
-    return null;
+  async function refreshParticipantsFromChannelLinks() {
+    try {
+      const links = await perspective.get(new LinkQuery({ predicate: CHANNEL }));
+      for (const link of links) {
+        if (!link.author) continue;
+        const channelId = link.data.source;
+        const channel = allChannels.value.find((c) => c.id === channelId);
+        if (!channel) continue;
+
+        const key = `${channelId}::${link.author}`;
+        if (knownParticipants.has(key)) continue;
+        knownParticipants.add(key);
+
+        perspective
+          .addLinks([{ source: channelId, predicate: 'flux://has_participant', target: link.author }])
+          .catch((error) => {
+            knownParticipants.delete(key);
+            console.error('Failed to add participant to channel:', {
+              channelId,
+              author: link.author,
+              error,
+            });
+          });
+      }
+    } catch (error) {
+      console.error('Error refreshing channel participants:', error);
+    }
   }
-  perspective.addListener('link-added', handleParticipantTracking);
 
-  // Cleanup function to remove all listeners
+  (async () => {
+    try {
+      // SELECT every distinct `(source, author)` over the `flux://has_channel`
+      // predicate.  The reifier metadata is queried via the RDF 1.2 `reifies`
+      // pattern that the rest of the model query pipeline uses.
+      const sub = await perspective.subscribeQuery(`
+        SELECT DISTINCT ?source ?author WHERE {
+          ?_r <http://www.w3.org/1999/02/22-rdf-syntax-ns#reifies> <<( ?source <${CHANNEL}> ?_target )>> .
+          ?_r <ad4m://ontology/author> ?author .
+        }
+      `);
+      if (participantTrackingCancelled) {
+        sub.dispose();
+        return;
+      }
+      channelLinksSub = sub;
+      sub.onResult(() => refreshParticipantsFromChannelLinks());
+    } catch (error) {
+      console.error('Failed to subscribe to channel-link author pairs:', error);
+    }
+  })();
+
+  // Cleanup function to tear down active subscriptions
   function cleanup() {
-    perspective.removeListener('link-added', handleParticipantTracking);
+    participantTrackingCancelled = true;
+    channelLinksSub?.dispose();
   }
 
   getMembers();

packages/api/src/index.ts

@@ -20,7 +20,6 @@ import Message from './message';
 import MessageSummary from './message/MessageSummary';
 import Post from './post';
 import SemanticRelationship from './semantic-relationship';
-import subscribeToLinks from './subscribeToLinks';
 import subscribeToSyncState from './subscribeToSyncState';
 import Topic from './topic';
 import Task from './task';
@@ -52,7 +51,6 @@ export {
   MessageSummary,
   Post,
   SemanticRelationship,
-  subscribeToLinks,
   subscribeToSyncState,
   Topic,
   Task,

packages/create/templates/preact/src/components/TodoView.tsx

@@ -28,12 +28,31 @@ export default function TodoView({ perspective, source }: Props) {
 
   useEffect(() => {
     loadTodos();
-    const handler = (link: any) => {
-      if (link.data?.source === source && link.data?.predicate === 'ad4m://has_child') loadTodos();
-      return null;
+    // Targeted SPARQL subscription: fires only when the set of
+    // `ad4m://has_child` links from THIS source changes, instead of waking
+    // on every link event in the perspective.  Mirrors the pattern used
+    // across the Flux app and views.
+    let sub: { dispose: () => void } | null = null;
+    let unmounted = false;
+    (async () => {
+      try {
+        const s = await perspective.subscribeQuery(
+          `SELECT ?id WHERE { <${source}> <ad4m://has_child> ?id . }`,
+        );
+        if (unmounted) {
+          s.dispose();
+          return;
+        }
+        sub = s;
+        s.onResult(() => loadTodos());
+      } catch (error) {
+        console.error('Failed to subscribe to todo links:', error);
+      }
+    })();
+    return () => {
+      unmounted = true;
+      sub?.dispose();
     };
-    perspective.addListener('link-added', handler);
-    return () => perspective.removeListener('link-added', handler);
   }, [source]);
 
   async function createTodo(event: React.KeyboardEvent<Element>) {

packages/create/templates/vue/src/components/TodoView.vue

@@ -70,21 +70,35 @@ async function loadTodos() {
   todos.value = items;
 }
 
+// Targeted SPARQL subscription: fires only when the set of `ad4m://has_child`
+// links from THIS source changes, instead of waking on every link event in
+// the perspective.  Mirrors the pattern used across the Flux app and views.
+let todoLinksSub: { dispose: () => void } | null = null;
+let unmounted = false;
+
 onMounted(async () => {
   await perspective.ensureSDNASubjectClass(Todo);
   loadTodos();
-  perspective.addListener('link-added', handleLinkAdded);
+  try {
+    const sub = await perspective.subscribeQuery(
+      `SELECT ?id WHERE { <${source}> <ad4m://has_child> ?id . }`,
+    );
+    if (unmounted) {
+      sub.dispose();
+      return;
+    }
+    todoLinksSub = sub;
+    sub.onResult(() => loadTodos());
+  } catch (error) {
+    console.error('Failed to subscribe to todo links:', error);
+  }
 });
 
 onUnmounted(() => {
-  perspective.removeListener('link-added', handleLinkAdded);
+  unmounted = true;
+  todoLinksSub?.dispose();
 });
 
-function handleLinkAdded(link: any) {
-  if (link.data?.source === source && link.data?.predicate === 'ad4m://has_child') loadTodos();
-  return null;
-}
-
 const createTodo = async () => {
   const todo = await Todo.create(perspective, { title: title.value });
   await perspective.add(new Link({ source, predicate: 'ad4m://has_child', target: todo.id }));