ENG-1791 Obsidian sync does not recover from Concept upsert failure (#1070)

Author: maparent

apps/obsidian/src/utils/publishNode.ts

@@ -17,6 +17,7 @@ import {
   syncPublishedNodeAssets,
 } from "./syncDgNodesToSupabase";
 import { isProvisionalSchema } from "./typeUtils";
+import { intersection, difference } from "@repo/utils/setOperations";
 
 import type { DiscourseNodeInVault } from "./getDiscourseNodes";
 import type { SupabaseContext } from "./supabaseContext";
@@ -68,31 +69,6 @@ const publishSchema = async ({
   }
 };
 
-const intersection = <T>(set1: Set<T>, set2: Set<T>): Set<T> => {
-  // @ts-expect-error - Set.intersection is ES2025 feature
-  if (set1.intersection) return set1.intersection(set2); // eslint-disable-line
-  const r: Set<T> = new Set();
-  for (const x of set1) {
-    if (set2.has(x)) r.add(x);
-  }
-  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,

apps/obsidian/src/utils/syncDgNodesToSupabase.ts

@@ -28,6 +28,8 @@ import {
 } from "./getDiscourseNodes";
 import { isAcceptedSchema } from "./typeUtils";
 import { getTemplatePluginInfo } from "./templates";
+import { difference } from "@repo/utils/setOperations";
+import { getAllPages } from "@repo/database/lib/pagination";
 
 const DEFAULT_TIME = "1970-01-01";
 export type ChangeType = "title" | "content";
@@ -226,6 +228,7 @@ type BuildChangedNodesOptions = {
   supabaseClient: DGSupabaseClient;
   context: SupabaseContext;
   changeTypesByPath?: Map<string, ChangeType[]>;
+  fullSync?: boolean;
 };
 
 const mergeChangeTypes = (
@@ -322,6 +325,7 @@ const buildChangedNodesFromNodes = async ({
   supabaseClient,
   context,
   changeTypesByPath,
+  fullSync = false,
 }: BuildChangedNodesOptions): Promise<ObsidianDiscourseNodeData[]> => {
   if (nodes.length === 0) {
     return [];
@@ -339,6 +343,34 @@ const buildChangedNodesFromNodes = async ({
     context.spaceId,
   );
   const changedNodes: ObsidianDiscourseNodeData[] = [];
+  let missingConcepts: Set<string> | undefined;
+  if (fullSync) {
+    const existingConceptIds = await getAllPages(
+      supabaseClient
+        .from("my_concepts")
+        .select("source_local_id")
+        .eq("space_id", context.spaceId)
+        .eq("arity", 0)
+        .eq("is_schema", false)
+        .order("id"),
+      1000,
+    );
+    if (Array.isArray(existingConceptIds)) {
+      // Here, compensating for concepts that never got upserted
+      // Probably due to non-atomicity of upsert of concept and content
+      // TODO try to see if there are other cases
+      // In particular, using timing when concepts get reordered by dependency
+      // may be error-prone
+      // fail silently otherwise, there will be other opportunities
+      const nodeIds = new Set(nodes.map((n) => n.nodeInstanceId));
+      const dbConceptIds = new Set(
+        existingConceptIds
+          .map((d) => d.source_local_id)
+          .filter((id) => id !== null),
+      );
+      missingConcepts = difference(nodeIds, dbConceptIds);
+    }
+  }
 
   for (const node of nodes) {
     if (node.frontmatter.importedFromRid) continue;
@@ -355,7 +387,10 @@ const buildChangedNodesFromNodes = async ({
         : detectedChangeTypes;
     const finalChangeTypes = mergedChangeTypes;
 
-    if (finalChangeTypes.length === 0) {
+    if (
+      finalChangeTypes.length === 0 &&
+      !missingConcepts?.has(node.nodeInstanceId)
+    ) {
       continue;
     }
 
@@ -397,6 +432,7 @@ export const syncAllNodesAndRelations = async (
           nodes: allNodes,
           supabaseClient,
           context,
+          fullSync: true,
         });
 
     const accountLocalId = plugin.settings.accountLocalId;

packages/database/src/lib/pagination.ts

@@ -0,0 +1,36 @@
+import { PostgrestError, PostgrestFilterBuilder } from "@supabase/supabase-js";
+import type { Database, Tables } from "@repo/database/dbTypes";
+
+type TableName =
+  | keyof Database["public"]["Tables"]
+  | keyof Database["public"]["Views"];
+
+type PGQuery<Name extends TableName, Result> = PostgrestFilterBuilder<
+  { PostgrestVersion: "12" },
+  Database["public"],
+  Tables<Name>,
+  Result[],
+  Name
+>;
+
+export const getAllPages = async <Name extends TableName, Result>(
+  query: PGQuery<Name, Result>,
+  limit: number = 200,
+): Promise<Result[] | PostgrestError> => {
+  // note: Bypassing protections
+  // eslint-disable-next-line
+  if ((query as any).url.search.indexOf("order") < 0)
+    throw new Error("Missing order clause on paginated query");
+  let offset = 0;
+  const rows: Result[] = [];
+  // eslint-disable-next-line no-constant-condition
+  while (true) {
+    const result = await query.range(offset, offset + limit - 1);
+    const { data, error } = result;
+    if (error) return error;
+    rows.push(...data);
+    if (data.length < limit) break;
+    offset += data.length;
+  }
+  return rows;
+};

packages/utils/src/setOperations.ts

@@ -0,0 +1,24 @@
+export const intersection = <T>(set1: Set<T>, set2: Set<T>): Set<T> => {
+  // @ts-expect-error - Set.intersection is ES2025 feature
+  if (set1.intersection) return set1.intersection(set2); // eslint-disable-line
+  const r: Set<T> = new Set();
+  for (const x of set1) {
+    if (set2.has(x)) r.add(x);
+  }
+  return r;
+};
+
+export 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;
+};