From ad939d48cc36fa85f9324da0b6618e9326affb34 Mon Sep 17 00:00:00 2001 From: sid597 Date: Mon, 22 Jun 2026 21:40:26 +0530 Subject: [PATCH 1/3] [ENG-1852] Keep Roam shared content fresh --- ...ertRoamNodeToFullContent.simple.example.ts | 47 +++ .../src/utils/convertRoamNodeToFullContent.ts | 14 +- apps/roam/src/utils/syncDgNodesToSupabase.ts | 338 ++++++++++++++++-- 3 files changed, 359 insertions(+), 40 deletions(-) create mode 100644 apps/roam/src/utils/convertRoamNodeToFullContent.simple.example.ts diff --git a/apps/roam/src/utils/convertRoamNodeToFullContent.simple.example.ts b/apps/roam/src/utils/convertRoamNodeToFullContent.simple.example.ts new file mode 100644 index 000000000..c2e942927 --- /dev/null +++ b/apps/roam/src/utils/convertRoamNodeToFullContent.simple.example.ts @@ -0,0 +1,47 @@ +import type { TreeNode } from "roamjs-components/types"; +import type { CrossAppNode } from "@repo/database/crossAppNodeContract"; +import { buildFullMarkdown } from "./convertRoamNodeToFullContent"; +import { contentTypes } from "@repo/content-model"; + +/** + * Compact example of a Roam page tree rendered as shared `full` markdown + * content. Typechecking this file keeps the example aligned with the cross-app + * content contract. + */ + +const block = (text: string, children: TreeNode[] = []): TreeNode => ({ + text, + children, + order: 0, + parents: [], + uid: "", + heading: 0, + open: true, + viewType: "bullet", + blockViewType: "outline", + editTime: new Date(0), + textAlign: "left", + props: { imageResize: {}, iframe: {} }, +}); + +const title = "Sleep improves memory consolidation"; + +const blocks: TreeNode[] = [ + block( + "Multiple studies show that sleep after learning strengthens memory traces.", + ), + block("Supporting evidence:", [block("[[EVD]] - Rasch & Born 2013")]), +]; + +export const roamClaimFullMarkdownSimpleExample: { + title: string; + blocks: TreeNode[]; + full: CrossAppNode["content"]["full"]; +} = { + title, + blocks, + full: { + format: contentTypes.markdown, + value: buildFullMarkdown({ title, blocks }), + }, +}; diff --git a/apps/roam/src/utils/convertRoamNodeToFullContent.ts b/apps/roam/src/utils/convertRoamNodeToFullContent.ts index 88369bb98..3654eb2cd 100644 --- a/apps/roam/src/utils/convertRoamNodeToFullContent.ts +++ b/apps/roam/src/utils/convertRoamNodeToFullContent.ts @@ -1,10 +1,19 @@ import { toMarkdown } from "./pageToMarkdown"; -import { type RoamDiscourseNodeData } from "./getAllDiscourseNodesSince"; import { type DiscourseNode } from "./getDiscourseNodes"; import getFullTreeByParentUid from "roamjs-components/queries/getFullTreeByParentUid"; import getPageViewType from "roamjs-components/queries/getPageViewType"; import type { TreeNode, ViewType } from "roamjs-components/types"; import type { LocalContentDataInput } from "@repo/database/inputTypes"; +import { contentTypes } from "@repo/content-model"; + +export type RoamFullContentNode = { + author_local_id: string; + source_local_id: string; + created: string | number; + last_modified: string | number; + text: string; + node_title?: string; +}; const FULL_MARKDOWN_OPTS = { refs: true, @@ -38,7 +47,7 @@ export const buildFullMarkdown = ({ export const convertRoamNodeToFullContent = ({ nodes, }: { - nodes: RoamDiscourseNodeData[]; + nodes: RoamFullContentNode[]; }): LocalContentDataInput[] => nodes.flatMap((node) => { try { @@ -55,6 +64,7 @@ export const convertRoamNodeToFullContent = ({ ).toISOString(), text: buildFullMarkdown({ title, blocks, viewType }), variant: "full", + content_type: contentTypes.roamMarkdown, scale: "document", }, ]; diff --git a/apps/roam/src/utils/syncDgNodesToSupabase.ts b/apps/roam/src/utils/syncDgNodesToSupabase.ts index aaa95db71..04ba896ce 100644 --- a/apps/roam/src/utils/syncDgNodesToSupabase.ts +++ b/apps/roam/src/utils/syncDgNodesToSupabase.ts @@ -3,6 +3,7 @@ import { getAllDiscourseNodesSince, nodeTypeSince, } from "./getAllDiscourseNodesSince"; +import getDiscourseNodeFormatExpression from "./getDiscourseNodeFormatExpression"; import { cleanupOrphanedNodes } from "./cleanupOrphanedNodes"; import { getLoggedInClient, @@ -18,7 +19,10 @@ import { } from "./conceptConversion"; import { fetchEmbeddingsForNodes } from "./upsertNodesAsContentWithEmbeddings"; import { convertRoamNodeToLocalContent } from "./upsertNodesAsContentWithEmbeddings"; -import { convertRoamNodeToFullContent } from "./convertRoamNodeToFullContent"; +import { + convertRoamNodeToFullContent, + type RoamFullContentNode, +} from "./convertRoamNodeToFullContent"; import type { DGSupabaseClient } from "@repo/database/lib/client"; import { intersection } from "@repo/utils/setOperations"; import type { Json, Enums } from "@repo/database/dbTypes"; @@ -42,6 +46,7 @@ const SYNC_TIMEOUT = "60s"; // must be less than half the SYNC_INTERVAL. const BATCH_SIZE = 200; const CONCEPT_BATCH_SIZE = 200; const END_SYNC_TASK_RESULT_VERSION = 1; +const DEFAULT_SYNC_TIME = new Date("1970-01-01").getTime(); type SyncPhaseDurations = Record; @@ -556,16 +561,29 @@ export const convertDgToSupabaseConcepts = async ({ nodesSince, since, allNodeTypes, + sharedNodeTypeIds = new Set(), supabaseClient, context, }: { nodesSince: RoamDiscourseNodeData[]; since: number | undefined; allNodeTypes: DiscourseNode[]; + sharedNodeTypeIds?: ReadonlySet; supabaseClient: DGSupabaseClient; context: SupabaseContext; }) => { - const nodeTypes = await nodeTypeSince(since, allNodeTypes); + const changedNodeTypes = await nodeTypeSince(since, allNodeTypes); + const nodeTypesByUid = new Map( + changedNodeTypes.map((nodeType) => [nodeType.type, nodeType]), + ); + + allNodeTypes.forEach((nodeType) => { + if (sharedNodeTypeIds.has(nodeType.type)) { + nodeTypesByUid.set(nodeType.type, nodeType); + } + }); + const nodeTypes = Array.from(nodeTypesByUid.values()); + await upsertNodeSchemaToContent({ nodeTypesUids: nodeTypes.map((node) => node.type), spaceId: context.spaceId, @@ -606,15 +624,44 @@ export const convertDgToSupabaseConcepts = async ({ }); }; +const uploadContentBatches = async ({ + content, + supabaseClient, + context, +}: { + content: LocalContentDataInput[]; + supabaseClient: DGSupabaseClient; + context: SupabaseContext; +}): Promise => { + if (content.length === 0) { + return; + } + + const batches = chunk(content, BATCH_SIZE); + + for (let idx = 0; idx < batches.length; idx++) { + const batch = batches[idx]; + + const { error } = await supabaseClient.rpc("upsert_content", { + data: batch as Json, + v_space_id: context.spaceId, + v_creator_id: context.userId, + content_as_document: true, + }); + + if (error) { + throw new Error(`upsert_content failed for batch ${idx + 1}`, { + cause: error, + }); + } + } +}; + export const upsertNodesToSupabaseAsContentWithEmbeddings = async ( roamNodes: RoamDiscourseNodeData[], supabaseClient: DGSupabaseClient, context: SupabaseContext, - options: { includeFullContent?: boolean } = {}, ): Promise => { - const { userId } = context; - const { includeFullContent = false } = options; - if (roamNodes.length === 0) { return; } @@ -622,34 +669,6 @@ export const upsertNodesToSupabaseAsContentWithEmbeddings = async ( nodes: roamNodes, }); - const uploadBatches = async ( - batches: LocalContentDataInput[][], - ): Promise => { - for (let idx = 0; idx < batches.length; idx++) { - const batch = batches[idx]; - - const { error } = await supabaseClient.rpc("upsert_content", { - data: batch as Json, - v_space_id: context.spaceId, - v_creator_id: userId, - content_as_document: true, - }); - - if (error) { - throw new Error(`upsert_content failed for batch ${idx + 1}`, { - cause: error, - }); - } - } - }; - - if (includeFullContent) { - const fullContent = convertRoamNodeToFullContent({ - nodes: roamNodes, - }); - await uploadBatches(chunk(fullContent, BATCH_SIZE)); - } - let nodesWithEmbeddings: LocalContentDataInput[]; try { nodesWithEmbeddings = await fetchEmbeddingsForNodes( @@ -672,7 +691,32 @@ export const upsertNodesToSupabaseAsContentWithEmbeddings = async ( ); } - await uploadBatches(chunk(nodesWithEmbeddings, BATCH_SIZE)); + await uploadContentBatches({ + content: nodesWithEmbeddings, + supabaseClient, + context, + }); +}; + +const upsertRoamNodesToSupabaseAsFullContent = async ({ + nodes, + supabaseClient, + context, +}: { + nodes: RoamFullContentNode[]; + supabaseClient: DGSupabaseClient; + context: SupabaseContext; +}): Promise => { + if (nodes.length === 0) { + return; + } + + const fullContent = convertRoamNodeToFullContent({ nodes }); + await uploadContentBatches({ + content: fullContent, + supabaseClient, + context, + }); }; const getAllUsers = async (): Promise => { @@ -782,6 +826,163 @@ const getAllMissingOrNewDiscourseNodes = async ({ ]; }; +const getSharedNodeInstanceSourceLocalIds = async ({ + supabaseClient, + spaceId, +}: { + supabaseClient: DGSupabaseClient; + spaceId: number; +}): Promise> => { + const sharedResources = await getAllPages( + supabaseClient + .from("ResourceAccess") + .select("source_local_id") + .eq("space_id", spaceId) + .order("source_local_id") + .order("account_uid"), + 1000, + ); + + if (!Array.isArray(sharedResources)) throw sharedResources; + + const sharedSourceLocalIds = new Set( + sharedResources.map((resource) => resource.source_local_id), + ); + const syncedInstanceConcepts = await getAllPages( + supabaseClient + .from("my_concepts") + .select("source_local_id") + .eq("space_id", spaceId) + .eq("is_schema", false) + .order("source_local_id"), + 1000, + ); + + if (!Array.isArray(syncedInstanceConcepts)) throw syncedInstanceConcepts; + + return new Set( + syncedInstanceConcepts + .map((concept) => concept.source_local_id) + .filter( + (id): id is string => id !== null && sharedSourceLocalIds.has(id), + ), + ); +}; + +const getSharedSourceLocalIdsMissingFullContent = async ({ + supabaseClient, + spaceId, + sharedSourceLocalIds, +}: { + supabaseClient: DGSupabaseClient; + spaceId: number; + sharedSourceLocalIds: ReadonlySet; +}): Promise> => { + const fullContentRows = await getAllPages( + supabaseClient + .from("Content") + .select("source_local_id") + .eq("space_id", spaceId) + .eq("variant", "full") + .order("source_local_id"), + 1000, + ); + + if (!Array.isArray(fullContentRows)) throw fullContentRows; + + const sourceLocalIdsWithFullContent = new Set( + fullContentRows + .map((row) => row.source_local_id) + .filter((id): id is string => id !== null), + ); + + return new Set( + [...sharedSourceLocalIds].filter( + (id) => !sourceLocalIdsWithFullContent.has(id), + ), + ); +}; + +type SharedFullContentUpdateRow = { + author_local_id: string; + source_local_id: string; + created: number; + node_edit_time: number; + page_edit_time: number; + text: string; +}; + +type SharedFullContentUpdate = { + fullContentNode: RoamFullContentNode; + nodeTypeId: string; +}; + +const getSharedRoamNodesWithFullContentUpdatesSince = async ({ + sourceLocalIds, + since, + nodeTypes, +}: { + sourceLocalIds: ReadonlySet; + since: number | undefined; + nodeTypes: DiscourseNode[]; +}): Promise => { + const sharedSourceLocalIds = Array.from(sourceLocalIds); + if (sharedSourceLocalIds.length === 0 || nodeTypes.length === 0) { + return []; + } + + const sinceMs = since ?? DEFAULT_SYNC_TIME; + const query = `[ + :find ?node-title ?uid ?nodeCreateTime ?nodeEditTime ?pageEditTime ?author_local_id + :keys text source_local_id created node_edit_time page_edit_time author_local_id + :in $ [?sharedUid ...] ?since + :where + [?node :block/uid ?sharedUid] + [?node :node/title ?node-title] + [?node :block/uid ?uid] + [?node :create/time ?nodeCreateTime] + [?node :create/user ?user-eid] + [?user-eid :user/uid ?author_local_id] + [(get-else $ ?node :edit/time ?nodeCreateTime) ?nodeEditTime] + [(get-else $ ?node :page/edit-time ?nodeEditTime) ?pageEditTime] + [or + [(> ?nodeEditTime ?since)] + [(> ?pageEditTime ?since)]] + ]`; + + const rows = (await window.roamAlphaAPI.data.backend.q( + query, + sharedSourceLocalIds, + sinceMs, + )) as unknown[] as SharedFullContentUpdateRow[]; + const typeMatchers = nodeTypes.map((node) => ({ + node, + regex: getDiscourseNodeFormatExpression(node.format), + })); + + return rows.flatMap((row) => { + const matchingNodeType = typeMatchers.find(({ regex }) => + regex.test(row.text), + )?.node; + if (matchingNodeType === undefined) { + return []; + } + + return [ + { + fullContentNode: { + author_local_id: row.author_local_id, + source_local_id: row.source_local_id, + created: row.created, + last_modified: Math.max(row.node_edit_time, row.page_edit_time), + text: row.text, + }, + nodeTypeId: matchingNodeType.type, + }, + ]; + }); +}; + export const createOrUpdateDiscourseEmbedding = async ( showToast = false, ): Promise => { @@ -899,7 +1100,7 @@ export const createOrUpdateDiscourseEmbedding = async ( (n) => n.backedBy === "user", ); - const allNodeInstances = await measureSyncPhase({ + const changedNodeInstances = await measureSyncPhase({ phase: isInitialSync ? "getAllMissingOrNewDiscourseNodes" : "getAllDiscourseNodesSince", @@ -914,6 +1115,56 @@ export const createOrUpdateDiscourseEmbedding = async ( }) : getAllDiscourseNodesSince(sinceTime, allDgNodeTypes), }); + const sharedSourceLocalIds = await measureSyncPhase({ + phase: "getSharedNodeInstanceSourceLocalIds", + phases, + operation: () => + getSharedNodeInstanceSourceLocalIds({ + supabaseClient: activeSupabaseClient, + spaceId: activeContext.spaceId, + }), + }); + const sharedSourceLocalIdsToBackfill = await measureSyncPhase({ + phase: "getSharedSourceLocalIdsMissingFullContent", + phases, + operation: () => + getSharedSourceLocalIdsMissingFullContent({ + supabaseClient: activeSupabaseClient, + spaceId: activeContext.spaceId, + sharedSourceLocalIds, + }), + }); + const sharedSourceLocalIdsToRefresh = new Set( + [...sharedSourceLocalIds].filter( + (id) => !sharedSourceLocalIdsToBackfill.has(id), + ), + ); + const sharedFullContentUpdates = await measureSyncPhase({ + phase: "getSharedFullContentUpdates", + phases, + operation: async () => { + const [refreshed, backfilled] = await Promise.all([ + getSharedRoamNodesWithFullContentUpdatesSince({ + sourceLocalIds: sharedSourceLocalIdsToRefresh, + since: sinceTime, + nodeTypes: allDgNodeTypes, + }), + getSharedRoamNodesWithFullContentUpdatesSince({ + sourceLocalIds: sharedSourceLocalIdsToBackfill, + since: DEFAULT_SYNC_TIME, + nodeTypes: allDgNodeTypes, + }), + ]); + return [...refreshed, ...backfilled]; + }, + }); + const sharedFullContentNodes = sharedFullContentUpdates.map( + (update) => update.fullContentNode, + ); + const sharedNodeTypeIds = new Set( + sharedFullContentUpdates.map((update) => update.nodeTypeId), + ); + await measureSyncPhase({ phase: "upsertUsers", phases, @@ -925,19 +1176,30 @@ export const createOrUpdateDiscourseEmbedding = async ( phases, operation: () => upsertNodesToSupabaseAsContentWithEmbeddings( - allNodeInstances, + changedNodeInstances, activeSupabaseClient, activeContext, ), }); + await measureSyncPhase({ + phase: "upsertFullContent", + phases, + operation: () => + upsertRoamNodesToSupabaseAsFullContent({ + nodes: sharedFullContentNodes, + supabaseClient: activeSupabaseClient, + context: activeContext, + }), + }); await measureSyncPhase({ phase: "convertConcepts", phases, operation: () => convertDgToSupabaseConcepts({ - nodesSince: allNodeInstances, + nodesSince: changedNodeInstances, since: sinceTime, allNodeTypes: allDgNodeTypes, + sharedNodeTypeIds, supabaseClient: activeSupabaseClient, context: activeContext, }), From 4d2396119f356ce58f518b9521356dcd3c84a9c9 Mon Sep 17 00:00:00 2001 From: sid597 Date: Thu, 9 Jul 2026 12:13:36 +0530 Subject: [PATCH 2/3] ENG-1852 Address shared full content review feedback --- ...ertRoamNodeToFullContent.simple.example.ts | 10 +---- .../src/utils/convertRoamNodeToFullContent.ts | 38 ++++++++++++------- apps/roam/src/utils/syncDgNodesToSupabase.ts | 10 +++-- 3 files changed, 32 insertions(+), 26 deletions(-) diff --git a/apps/roam/src/utils/convertRoamNodeToFullContent.simple.example.ts b/apps/roam/src/utils/convertRoamNodeToFullContent.simple.example.ts index c2e942927..0773fae26 100644 --- a/apps/roam/src/utils/convertRoamNodeToFullContent.simple.example.ts +++ b/apps/roam/src/utils/convertRoamNodeToFullContent.simple.example.ts @@ -1,14 +1,8 @@ import type { TreeNode } from "roamjs-components/types"; -import type { CrossAppNode } from "@repo/database/crossAppNodeContract"; +import type { CrossAppNode } from "@repo/database/crossAppContracts"; import { buildFullMarkdown } from "./convertRoamNodeToFullContent"; import { contentTypes } from "@repo/content-model"; -/** - * Compact example of a Roam page tree rendered as shared `full` markdown - * content. Typechecking this file keeps the example aligned with the cross-app - * content contract. - */ - const block = (text: string, children: TreeNode[] = []): TreeNode => ({ text, children, @@ -41,7 +35,7 @@ export const roamClaimFullMarkdownSimpleExample: { title, blocks, full: { - format: contentTypes.markdown, + contentType: contentTypes.markdown, value: buildFullMarkdown({ title, blocks }), }, }; diff --git a/apps/roam/src/utils/convertRoamNodeToFullContent.ts b/apps/roam/src/utils/convertRoamNodeToFullContent.ts index 3654eb2cd..2a31825c8 100644 --- a/apps/roam/src/utils/convertRoamNodeToFullContent.ts +++ b/apps/roam/src/utils/convertRoamNodeToFullContent.ts @@ -3,8 +3,10 @@ import { type DiscourseNode } from "./getDiscourseNodes"; import getFullTreeByParentUid from "roamjs-components/queries/getFullTreeByParentUid"; import getPageViewType from "roamjs-components/queries/getPageViewType"; import type { TreeNode, ViewType } from "roamjs-components/types"; -import type { LocalContentDataInput } from "@repo/database/inputTypes"; import { contentTypes } from "@repo/content-model"; +import type { CrossAppNode } from "@repo/database/crossAppContracts"; +import { crossAppNodeToDbContent } from "@repo/database/lib/crossAppConverters"; +import type { LocalContentDataInput } from "@repo/database/inputTypes"; export type RoamFullContentNode = { author_local_id: string; @@ -12,6 +14,7 @@ export type RoamFullContentNode = { created: string | number; last_modified: string | number; text: string; + node_type_id: string; node_title?: string; }; @@ -54,20 +57,27 @@ export const convertRoamNodeToFullContent = ({ const title = node.node_title ?? node.text; const blocks = getFullTreeByParentUid(node.source_local_id).children; const viewType = getPageViewType(title) || "bullet"; - return [ - { - author_local_id: node.author_local_id, - source_local_id: node.source_local_id, - created: new Date(node.created || Date.now()).toISOString(), - last_modified: new Date( - node.last_modified || Date.now(), - ).toISOString(), - text: buildFullMarkdown({ title, blocks, viewType }), - variant: "full", - content_type: contentTypes.roamMarkdown, - scale: "document", + const crossAppNode: CrossAppNode = { + author: { localId: node.author_local_id }, + localId: node.source_local_id, + createdAt: new Date(node.created || Date.now()), + modifiedAt: new Date(node.last_modified || Date.now()), + nodeType: { localId: node.node_type_id }, + content: { + direct: { + localId: node.source_local_id, + value: title, + }, + full: { + localId: node.source_local_id, + value: buildFullMarkdown({ title, blocks, viewType }), + contentType: contentTypes.roamMarkdown, + scale: "document", + }, }, - ]; + }; + const fullContent = crossAppNodeToDbContent(crossAppNode, "full"); + return fullContent === undefined ? [] : [fullContent]; } catch (error) { console.error( `convertRoamNodeToFullContent: failed to build full markdown for ${node.source_local_id}:`, diff --git a/apps/roam/src/utils/syncDgNodesToSupabase.ts b/apps/roam/src/utils/syncDgNodesToSupabase.ts index 04ba896ce..bece2dcda 100644 --- a/apps/roam/src/utils/syncDgNodesToSupabase.ts +++ b/apps/roam/src/utils/syncDgNodesToSupabase.ts @@ -854,19 +854,20 @@ const getSharedNodeInstanceSourceLocalIds = async ({ .select("source_local_id") .eq("space_id", spaceId) .eq("is_schema", false) + .eq("arity", 0) .order("source_local_id"), 1000, ); if (!Array.isArray(syncedInstanceConcepts)) throw syncedInstanceConcepts; - return new Set( + const syncedInstanceSourceLocalIds = new Set( syncedInstanceConcepts .map((concept) => concept.source_local_id) - .filter( - (id): id is string => id !== null && sharedSourceLocalIds.has(id), - ), + .filter((id): id is string => id !== null), ); + + return intersection(syncedInstanceSourceLocalIds, sharedSourceLocalIds); }; const getSharedSourceLocalIdsMissingFullContent = async ({ @@ -976,6 +977,7 @@ const getSharedRoamNodesWithFullContentUpdatesSince = async ({ created: row.created, last_modified: Math.max(row.node_edit_time, row.page_edit_time), text: row.text, + node_type_id: matchingNodeType.type, }, nodeTypeId: matchingNodeType.type, }, From fddb012cc7c7ea1bb40772f57480644f9c222eff Mon Sep 17 00:00:00 2001 From: sid597 Date: Thu, 9 Jul 2026 12:21:15 +0530 Subject: [PATCH 3/3] ENG-1852 Align full content example type --- .../src/utils/convertRoamNodeToFullContent.simple.example.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/roam/src/utils/convertRoamNodeToFullContent.simple.example.ts b/apps/roam/src/utils/convertRoamNodeToFullContent.simple.example.ts index 0773fae26..cef2e3160 100644 --- a/apps/roam/src/utils/convertRoamNodeToFullContent.simple.example.ts +++ b/apps/roam/src/utils/convertRoamNodeToFullContent.simple.example.ts @@ -35,7 +35,7 @@ export const roamClaimFullMarkdownSimpleExample: { title, blocks, full: { - contentType: contentTypes.markdown, + contentType: contentTypes.roamMarkdown, value: buildFullMarkdown({ title, blocks }), }, };