eng-1485 rescan unpublished relations

Author: maparent

apps/obsidian/src/utils/publishNode.ts

@@ -15,6 +15,9 @@ import {
 import type { RelationInstance } from "~/types";
 import { getAvailableGroupIds } from "./importNodes";
 import { syncAllNodesAndRelations } from "./syncDgNodesToSupabase";
+import type { DiscourseNodeInVault } from "./getDiscourseNodes";
+import type { SupabaseContext } from "./supabaseContext";
+import type { TablesInsert } from "@repo/database/dbTypes";
 
 const publishSchema = async ({
   client,
@@ -74,6 +77,21 @@ const intersection = <T>(set1: Set<T>, set2: Set<T>): Set<T> => {
   return r;
 };
 
+const difference = <T>(set1: Set<T>, set2: Set<T>): Set<T> => {
+  // @ts-expect-error - Set.difference is ES2025 feature
+  if (set1.difference) return set1.difference(set2); // eslint-disable-line
+  const result = new Set(set1);
+  if (set1.size <= set2.size)
+    for (const e of set1) {
+      if (set2.has(e)) result.delete(e);
+    }
+  else
+    for (const e of set2) {
+      if (result.has(e)) result.delete(e);
+    }
+  return result;
+};
+
 export const publishNewRelation = async (
   plugin: DiscourseGraphPlugin,
   relation: RelationInstance,
@@ -249,6 +267,116 @@ export const publishNode = async ({
   return await publishNodeToGroup({ plugin, file, frontmatter, myGroup });
 };
 
+export const ensurePublishedRelationsAccuracy = async ({
+  client,
+  context,
+  allNodesById,
+  relationInstances,
+}: {
+  client: DGSupabaseClient;
+  context: SupabaseContext;
+  allNodesById: Record<string, DiscourseNodeInVault>;
+  relationInstances: RelationInstance[];
+}): Promise<void> => {
+  const myGroups = await getAvailableGroupIds(client);
+  const syncedRelationIdsResult = await client
+    .from("Concept")
+    .select("source_local_id")
+    .eq("space_id", context.spaceId)
+    .eq("is_schema", false)
+    .gt("arity", 0);
+  if (syncedRelationIdsResult.error) {
+    console.error(
+      "Could not get synced relation ids",
+      syncedRelationIdsResult.error,
+    );
+    return;
+  }
+  const syncedRelationIds = new Set(
+    (syncedRelationIdsResult.data || []).map((x) => x.source_local_id!),
+  );
+  // Also a good time to look at orphan relations
+  const existingRelationIds = new Set(relationInstances.map((r) => r.id));
+  const orphanRelationIds = difference(syncedRelationIds, existingRelationIds);
+  if (orphanRelationIds.size) {
+    const r = await client
+      .from("Concept")
+      .delete()
+      .eq("space_id", context.spaceId)
+      .in("source_local_id", [...orphanRelationIds]);
+    if (!r.error) {
+      for (const id of orphanRelationIds) {
+        syncedRelationIds.delete(id);
+      }
+    }
+  }
+  const missingPublishRecords: TablesInsert<"ResourceAccess">[] = [];
+  for (const group of myGroups) {
+    const publishableRelations = relationInstances.filter(
+      (r) =>
+        !r.importedFromRid &&
+        (
+          (allNodesById[r.source]?.frontmatter?.publishedToGroups as
+            | string[]
+            | undefined) || []
+        ).indexOf(group) >= 0 &&
+        (
+          (allNodesById[r.destination]?.frontmatter?.publishedToGroups as
+            | string[]
+            | undefined) || []
+        ).indexOf(group) >= 0,
+    );
+    const publishableRelationIds = new Set(
+      publishableRelations.map((x) => x.id),
+    );
+    const publishedIds = await client
+      .from("ResourceAccess")
+      .select("source_local_id")
+      .eq("account_uid", group)
+      .eq("space_id", context.spaceId);
+    if (publishedIds.error) {
+      console.error("Could not get synced relation ids", publishedIds.error);
+      continue;
+    }
+    const publishedRelationIds = intersection(
+      syncedRelationIds,
+      new Set((publishedIds.data || []).map((x) => x.source_local_id)),
+    );
+    const missingPublishableIds = difference(
+      publishableRelationIds,
+      publishedRelationIds,
+    );
+    if (missingPublishableIds.size > 0) {
+      missingPublishRecords.push(
+        /* eslint-disable @typescript-eslint/naming-convention */
+        ...[...missingPublishableIds].map((source_local_id) => ({
+          source_local_id,
+          space_id: context.spaceId,
+          account_uid: group,
+        })),
+        /* eslint-enable @typescript-eslint/naming-convention */
+      );
+    }
+    const extraPublishableIds = difference(
+      publishedRelationIds,
+      publishableRelationIds,
+    );
+    if (extraPublishableIds.size > 0) {
+      const r = await client
+        .from("ResourceAccess")
+        .delete()
+        .eq("account_uid", group)
+        .eq("space_id", context.spaceId)
+        .in("source_local_id", [...extraPublishableIds]);
+      if (r.error) console.error(r.error);
+    }
+  }
+  if (missingPublishRecords.length > 0) {
+    const r = await client.from("ResourceAccess").upsert(missingPublishRecords);
+    if (r.error) console.error(r.error);
+  }
+};
+
 export const publishNodeToGroup = async ({
   plugin,
   file,

apps/obsidian/src/utils/syncDgNodesToSupabase.ts

@@ -9,7 +9,7 @@ import {
   type SupabaseContext,
 } from "./supabaseContext";
 import { default as DiscourseGraphPlugin } from "~/index";
-import { publishNode } from "./publishNode";
+import { publishNode, ensurePublishedRelationsAccuracy } from "./publishNode";
 import { upsertNodesToSupabaseAsContentWithEmbeddings } from "./upsertNodesAsContentWithEmbeddings";
 import {
   orderConceptsByDependency,
@@ -416,6 +416,7 @@ export const syncAllNodesAndRelations = async (
       accountLocalId,
       plugin,
       allNodes,
+      startupRun: true,
     });
 
     // When synced nodes are already published, ensure non-text assets are in storage.
@@ -433,13 +434,15 @@ const convertDgToSupabaseConcepts = async ({
   accountLocalId,
   plugin,
   allNodes,
+  startupRun,
 }: {
   nodesSince: ObsidianDiscourseNodeData[];
   supabaseClient: DGSupabaseClient;
   context: SupabaseContext;
   accountLocalId: string;
   plugin: DiscourseGraphPlugin;
   allNodes?: DiscourseNodeInVault[];
+  startupRun?: boolean;
 }): Promise<void> => {
   const lastNodeSchemaSync = (
     await getLastNodeSchemaSyncTime(supabaseClient, context.spaceId)
@@ -562,6 +565,14 @@ const convertDgToSupabaseConcepts = async ({
             ? error
             : JSON.stringify(error, null, 2);
       throw new Error(`upsert_concepts failed: ${errorMessage}`);
+    } else if (startupRun === true) {
+      // occasional extra work: Make sure relations that should be published are.
+      await ensurePublishedRelationsAccuracy({
+        client: supabaseClient,
+        context,
+        allNodesById,
+        relationInstances: Object.values(relationInstancesData.relations),
+      });
     }
   }
 };