ENG-1634 Re-sync nodes if needed at the time the users publish the node

apps/obsidian/src/utils/publishNode.ts

@@ -1,9 +1,6 @@
 import type { FrontMatterCache, TFile } from "obsidian";
-import { Notice } from "obsidian";
 import type { default as DiscourseGraphPlugin } from "~/index";
 import { getLoggedInClient, getSupabaseContext } from "./supabaseContext";
-import { addFile } from "@repo/database/lib/files";
-import mime from "mime-types";
 import type { DGSupabaseClient } from "@repo/database/lib/client";
 import {
   getRelationsForNodeInstanceId,
@@ -15,7 +12,10 @@ import {
 } from "./relationsStore";
 import type { RelationInstance } from "~/types";
 import { getAvailableGroupIds } from "./importNodes";
-import { syncAllNodesAndRelations } from "./syncDgNodesToSupabase";
+import {
+  syncAllNodesAndRelations,
+  syncPublishedNodeAssets,
+} from "./syncDgNodesToSupabase";
 import type { DiscourseNodeInVault } from "./getDiscourseNodes";
 import type { SupabaseContext } from "./supabaseContext";
 import type { TablesInsert } from "@repo/database/dbTypes";
@@ -251,21 +251,30 @@ export const publishNode = async ({
   plugin,
   file,
   frontmatter,
+  republish,
 }: {
   plugin: DiscourseGraphPlugin;
   file: TFile;
   frontmatter: FrontMatterCache;
+  republish?: boolean;
 }): Promise<void> => {
   const client = await getLoggedInClient(plugin);
   if (!client) throw new Error("Cannot get client");
   const myGroups = new Set(await getAvailableGroupIds(client));
   if (myGroups.size === 0) throw new Error("Cannot get group");
   const existingPublish =
     (frontmatter.publishedToGroups as undefined | string[]) || [];
+  if (!republish) await syncAllNodesAndRelations(plugin);
   const commonGroups = existingPublish.filter((g) => myGroups.has(g));
   // temporary single-group assumption
   const myGroup = (commonGroups.length > 0 ? commonGroups : [...myGroups])[0]!;
-  return await publishNodeToGroup({ plugin, file, frontmatter, myGroup });
+  return await publishNodeToGroup({
+    plugin,
+    file,
+    frontmatter,
+    myGroup,
+    republish,
+  });
 };
 
 export const ensurePublishedRelationsAccuracy = async ({
@@ -413,11 +422,13 @@ export const publishNodeToGroup = async ({
   file,
   frontmatter,
   myGroup,
+  republish,
 }: {
   plugin: DiscourseGraphPlugin;
   file: TFile;
   frontmatter: FrontMatterCache;
   myGroup: string;
+  republish?: boolean;
 }): Promise<void> => {
   const nodeId = frontmatter.nodeInstanceId as string | undefined;
   if (!nodeId) throw new Error("Please sync the node first");
@@ -464,7 +475,8 @@ export const publishNodeToGroup = async ({
   );
 
   const skipPublishAccess =
-    existingPublish.includes(myGroup) && lastModified <= lastModifiedDb;
+    republish ||
+    (existingPublish.includes(myGroup) && lastModified <= lastModifiedDb);
 
   if (!skipPublishAccess) {
     const publishSpaceResponse = await client.from("SpaceAccess").upsert(
@@ -506,65 +518,15 @@ export const publishNodeToGroup = async ({
       });
     }
   }
-
-  // Always sync non-text assets when node is published to this group
-  const existingFiles: string[] = [];
-  const existingReferencesReq = await client
-    .from("my_file_references")
-    .select("*")
-    .eq("space_id", spaceId)
-    .eq("source_local_id", nodeId);
-  if (existingReferencesReq.error) {
-    console.error(existingReferencesReq.error);
-    return;
-  }
-  const existingReferencesByPath = Object.fromEntries(
-    existingReferencesReq.data.map((ref) => [ref.filepath, ref]),
-  );
-
-  for (const attachment of attachments) {
-    const mimetype = mime.lookup(attachment.path) || "application/octet-stream";
-    if (mimetype.startsWith("text/")) continue;
-    // Do not use standard upload for large files
-    if (attachment.stat.size >= 6 * 1024 * 1024) {
-      new Notice(
-        `Asset file ${attachment.path} is larger than 6Mb and will not be uploaded`,
-      );
-      continue;
-    }
-    existingFiles.push(attachment.path);
-    const existingRef = existingReferencesByPath[attachment.path];
-    if (
-      !existingRef ||
-      new Date(existingRef.last_modified + "Z").valueOf() <
-        attachment.stat.mtime
-    ) {
-      const content = await plugin.app.vault.readBinary(attachment);
-      await addFile({
-        client,
-        spaceId,
-        sourceLocalId: nodeId,
-        fname: attachment.path,
-        mimetype,
-        created: new Date(attachment.stat.ctime),
-        lastModified: new Date(attachment.stat.mtime),
-        content,
-      });
-    }
-  }
-  let cleanupCommand = client
-    .from("FileReference")
-    .delete()
-    .eq("space_id", spaceId)
-    .eq("source_local_id", nodeId);
-  if (existingFiles.length)
-    cleanupCommand = cleanupCommand.notIn("filepath", [
-      ...new Set(existingFiles),
-    ]);
-  const cleanupResult = await cleanupCommand;
-  // do not fail on cleanup
-  if (cleanupResult.error) console.error(cleanupResult.error);
-
+  if (!republish)
+    await syncPublishedNodeAssets({
+      plugin,
+      client,
+      nodeId,
+      spaceId,
+      file,
+      attachments,
+    });
   if (!existingPublish.includes(myGroup))
     await plugin.app.fileManager.processFrontMatter(
       file,

apps/obsidian/src/utils/syncDgNodesToSupabase.ts

@@ -1,5 +1,7 @@
 /* eslint-disable @typescript-eslint/naming-convention */
-import { FrontMatterCache, Notice, TFile } from "obsidian";
+import { Notice, TFile } from "obsidian";
+import { addFile } from "@repo/database/lib/files";
+import mime from "mime-types";
 import { ensureNodeInstanceId } from "~/utils/nodeInstanceId";
 import type { DGSupabaseClient } from "@repo/database/lib/client";
 import type { Json } from "@repo/database/dbTypes";
@@ -9,7 +11,7 @@ import {
   type SupabaseContext,
 } from "./supabaseContext";
 import { default as DiscourseGraphPlugin } from "~/index";
-import { publishNode, ensurePublishedRelationsAccuracy } from "./publishNode";
+import { ensurePublishedRelationsAccuracy } from "./publishNode";
 import { upsertNodesToSupabaseAsContentWithEmbeddings } from "./upsertNodesAsContentWithEmbeddings";
 import {
   orderConceptsByDependency,
@@ -579,6 +581,92 @@ const convertDgToSupabaseConcepts = async ({
   }
 };
 
+export const syncPublishedNodeAssets = async ({
+  plugin,
+  client,
+  nodeId,
+  spaceId,
+  file,
+  attachments,
+}: {
+  plugin: DiscourseGraphPlugin;
+  client: DGSupabaseClient;
+  nodeId: string;
+  spaceId: number;
+  file: TFile;
+  attachments?: TFile[];
+}) => {
+  if (attachments === undefined) {
+    const embeds = plugin.app.metadataCache.getFileCache(file)?.embeds ?? [];
+    attachments = embeds
+      .map(({ link }) => {
+        const attachment = plugin.app.metadataCache.getFirstLinkpathDest(
+          link,
+          file.path,
+        );
+        return attachment;
+      })
+      .filter((a) => !!a);
+  }
+  // Always sync non-text assets when node is published to this group
+  const existingFiles: string[] = [];
+  const existingReferencesReq = await client
+    .from("my_file_references")
+    .select("*")
+    .eq("space_id", spaceId)
+    .eq("source_local_id", nodeId);
+  if (existingReferencesReq.error) {
+    console.error(existingReferencesReq.error);
+    return;
+  }
+  const existingReferencesByPath = Object.fromEntries(
+    existingReferencesReq.data.map((ref) => [ref.filepath, ref]),
+  ) as Record<string, (typeof existingReferencesReq.data)[0]>;
+
+  for (const attachment of attachments) {
+    const mimetype = mime.lookup(attachment.path) || "application/octet-stream";
+    if (mimetype.startsWith("text/")) continue;
+    // Do not use standard upload for large files
+    if (attachment.stat.size >= 6 * 1024 * 1024) {
+      new Notice(
+        `Asset file ${attachment.path} is larger than 6Mb and will not be uploaded`,
+      );
+      continue;
+    }
+    existingFiles.push(attachment.path);
+    const existingRef = existingReferencesByPath[attachment.path];
+    if (
+      !existingRef ||
+      new Date(existingRef.last_modified + "Z").valueOf() <
+        attachment.stat.mtime
+    ) {
+      const content = await plugin.app.vault.readBinary(attachment);
+      await addFile({
+        client,
+        spaceId,
+        sourceLocalId: nodeId,
+        fname: attachment.path,
+        mimetype,
+        created: new Date(attachment.stat.ctime),
+        lastModified: new Date(attachment.stat.mtime),
+        content,
+      });
+    }
+  }
+  let cleanupCommand = client
+    .from("FileReference")
+    .delete()
+    .eq("space_id", spaceId)
+    .eq("source_local_id", nodeId);
+  if (existingFiles.length)
+    cleanupCommand = cleanupCommand.notIn("filepath", [
+      ...new Set(existingFiles),
+    ]);
+  const cleanupResult = await cleanupCommand;
+  // do not fail on cleanup
+  if (cleanupResult.error) console.error(cleanupResult.error);
+};
+
 /**
  * For nodes that are already published, ensure non-text assets are pushed to
  * storage. Called after content sync so new embeds (e.g. images) get uploaded.
@@ -587,17 +675,26 @@ const syncPublishedNodesAssets = async (
   plugin: DiscourseGraphPlugin,
   nodes: ObsidianDiscourseNodeData[],
 ): Promise<void> => {
+  const context = await getSupabaseContext(plugin);
+  if (!context) throw new Error("Cannot get context");
+  const spaceId = context.spaceId;
+  const client = await getLoggedInClient(plugin);
+  if (!client) throw new Error("Cannot get client");
   const published = nodes.filter(
     (n) =>
       ((n.frontmatter.publishedToGroups as string[] | undefined)?.length ?? 0) >
       0,
   );
   for (const node of published) {
     try {
-      await publishNode({
+      const nodeId = node.frontmatter.nodeInstanceId as string | undefined;
+      if (!nodeId) throw new Error("Please sync the node first");
+      await syncPublishedNodeAssets({
         plugin,
+        client,
+        nodeId,
+        spaceId,
         file: node.file,
-        frontmatter: node.frontmatter as FrontMatterCache,
       });
     } catch (error) {
       console.error(
@@ -651,7 +748,14 @@ const syncChangedNodesToSupabase = async ({
 
   // When file changes affect an already-published node, ensure new non-text
   // assets (e.g. images) are pushed to storage.
-  await syncPublishedNodesAssets(plugin, changedNodes);
+  try {
+    await syncPublishedNodesAssets(plugin, changedNodes);
+  } catch (error) {
+    // console.error(
+    //   `Failed to sync published node assets`,
+    //   error,
+    // );
+  }
 };
 
 /**