ENG-1865 Publish Roam stored relations for shared nodes

apps/roam/src/utils/createReifiedBlock.ts

@@ -102,6 +102,36 @@ export const countReifiedRelations = async (): Promise<number> => {
   return (r[0] || [0])[0] as number;
 };
 
+export type ReifiedRelationData = {
+  sourceUid: string;
+  destinationUid: string;
+  hasSchema: string;
+  importedFromRid?: string;
+};
+
+export type ReifiedRelationDataWithRelId = ReifiedRelationData & {
+  relationId: string;
+};
+
+export const getReifiedRelations = async (): Promise<
+  ReifiedRelationDataWithRelId[]
+> => {
+  const pageUid = getExistingRelationPageUid();
+  if (pageUid === undefined) return [];
+  const r = await window.roamAlphaAPI.data.async.q(
+    `[:find ?ruid ?rdata :where
+      [?p :block/uid "${pageUid}"]
+      [?p :block/children ?c]
+      [?c :block/uid ?ruid]
+      [?c :block/props ?pr]
+      [(get ?pr :${DISCOURSE_GRAPH_PROP_NAME}) ?rdata] ]`,
+  );
+  return r.map((x) => ({
+    relationId: x[0] as string,
+    ...(x[1] as ReifiedRelationData),
+  }));
+};
+
 export const createReifiedRelation = async ({
   sourceUid,
   relationBlockUid,

apps/roam/src/utils/publishNodesToGroups.ts

@@ -1,5 +1,22 @@
 import type { DGSupabaseClient } from "@repo/database/lib/client";
 import { getAvailableGroupIds } from "@repo/database/lib/groups";
+import getDiscourseRelations from "./getDiscourseRelations";
+import getDiscourseNodes from "./getDiscourseNodes";
+import {
+  getReifiedRelations,
+  type ReifiedRelationDataWithRelId,
+} from "./createReifiedBlock";
+import type {
+  CrossAppNodeSchema,
+  CrossAppRelation,
+  CrossAppRelationTripleSchema,
+} from "@repo/database/crossAppContracts";
+import type { DiscourseRelation } from "./getDiscourseRelations";
+import {
+  crossAppRelationToDbConcept,
+  crossAppRelationTripleSchemaToDbConcept,
+} from "@repo/database/lib/crossAppConverters";
+import { spaceUriAndLocalIdToRid } from "@repo/database/lib/rid";
 
 export type PublishNode = {
   uid: string;
@@ -13,6 +30,309 @@ type PublishNodesResult = {
   failedGroupIds: string[];
 };
 
+const getAllPublishedIds = async (
+  client: DGSupabaseClient,
+  spaceId: number,
+  groupId: string,
+): Promise<string[]> => {
+  const response = await client
+    .from("ResourceAccess")
+    .select("source_local_id")
+    .eq("space_id", spaceId)
+    .eq("account_uid", groupId);
+  if (response.error) throw response.error;
+  return response.data.map((r) => r.source_local_id);
+};
+
+const getGroupSpaceIdAndUrls = async (
+  client: DGSupabaseClient,
+  groupId: string,
+): Promise<Record<number, string>> => {
+  const response = await client
+    .from("SpaceAccess")
+    .select("space_id")
+    .eq("account_uid", groupId);
+  if (response.error) throw response.error;
+  const spaceIds = response.data.map((r) => r.space_id);
+  const response2 = await client
+    .from("Space")
+    .select("id, url")
+    .in("id", spaceIds);
+  if (response2.error) throw response2.error;
+  return Object.fromEntries(response2.data.map(({ id, url }) => [id, url]));
+};
+
+// Use readImportedSourceIdentity when it's integrated.
+// eslint-disable-next-line @typescript-eslint/no-unused-vars
+const getSpaceIdOf = (nodeId: string): number | undefined => undefined;
+
+const getSyncedRelationLocalIds = async (
+  client: DGSupabaseClient,
+  spaceId: number,
+): Promise<string[]> => {
+  const response = await client
+    .from("my_concepts")
+    .select("source_local_id")
+    .eq("space_id", spaceId)
+    .eq("arity", 2); // intentionally include both schemas and instances
+  if (response.error) throw response.error;
+  return response.data
+    .map((r) => r.source_local_id)
+    .filter((id) => id !== null);
+};
+
+const ensureRelationsSynced = async ({
+  client,
+  spaceId,
+  relations,
+  existing,
+  isImportedFrom,
+  spaceUids,
+}: {
+  client: DGSupabaseClient;
+  spaceId: number;
+  relations: ReifiedRelationDataWithRelId[];
+  existing: Set<string>;
+  isImportedFrom: (id: string) => number | undefined;
+  spaceUids: Record<number, string>;
+}): Promise<void> => {
+  const missing = relations.filter((r) => !existing.has(r.relationId));
+  const asCrossApp: CrossAppRelation[] = missing
+    .map((r) => {
+      const sourceSpaceUid = spaceUids[isImportedFrom(r.sourceUid) || 0];
+      const destinationSpaceUid =
+        spaceUids[isImportedFrom(r.destinationUid) || 0];
+      const sourceId =
+        sourceSpaceUid === undefined
+          ? r.sourceUid
+          : spaceUriAndLocalIdToRid(sourceSpaceUid, r.sourceUid);
+      const destinationId =
+        destinationSpaceUid === undefined
+          ? r.destinationUid
+          : spaceUriAndLocalIdToRid(destinationSpaceUid, r.destinationUid);
+      const relData = window.roamAlphaAPI.pull(
+        "[:create/time :edit/time {:create/user [:user/uid]}]",
+        `[:block/uid "${r.relationId}"]`,
+      ) as unknown as {
+        ":create/time": number;
+        ":edit/time": number;
+        ":create/user": { ":user/uid": string };
+      };
+      if (relData == undefined || !relData[":create/user"]) return null;
+      const userUid = relData[":create/user"][":user/uid"];
+
+      return {
+        localId: r.relationId,
+        relationType: r.hasSchema,
+        source: sourceId,
+        destination: destinationId,
+        authorId: userUid,
+        createdAt: new Date(relData[":create/time"]),
+        modifiedAt: new Date(relData[":edit/time"]),
+      };
+    })
+    .filter((s) => s !== null);
+  if (asCrossApp.length == 0) return;
+  const asDbData = asCrossApp.map((r) => crossAppRelationToDbConcept(r));
+  const response = await client.rpc("upsert_concepts", {
+    v_space_id: spaceId,
+    data: asDbData,
+  });
+  if (response.error) throw response.error;
+  asCrossApp.forEach((r) => {
+    existing.add(r.localId);
+  });
+};
+
+const ensureRelationSchemasSynced = async ({
+  client,
+  spaceId,
+  relationSchemas,
+  existing,
+  nodeSchemaById,
+}: {
+  client: DGSupabaseClient;
+  spaceId: number;
+  relationSchemas: DiscourseRelation[];
+  existing: Set<string>;
+  nodeSchemaById: Record<string, CrossAppNodeSchema>;
+}): Promise<void> => {
+  const missing = relationSchemas.filter((r) => !existing.has(r.id));
+  const asCrossApp: CrossAppRelationTripleSchema[] = missing
+    .map((r) => {
+      const relData = window.roamAlphaAPI.pull(
+        "[:create/time :edit/time {:create/user [:user/uid]}]",
+        `[:block/uid "${r.id}"]`,
+      ) as unknown as {
+        ":create/time": number;
+        ":edit/time": number;
+        ":create/user": { ":user/uid": string };
+      };
+      if (!relData) return null;
+      const userUid = relData[":create/user"][":user/uid"];
+
+      return {
+        localId: r.id,
+        sourceType: r.source,
+        destinationType: r.destination,
+        label: r.label,
+        complement: r.complement,
+        authorId: userUid,
+        createdAt: new Date(relData[":create/time"]),
+        modifiedAt: new Date(relData[":edit/time"]),
+      };
+    })
+    .filter((x) => x !== null);
+  if (asCrossApp.length == 0) return;
+  const asDbData = asCrossApp
+    .map((r) => {
+      const sourceSchema = nodeSchemaById[r.sourceType];
+      const destinationSchema = nodeSchemaById[r.destinationType];
+      if (!sourceSchema || !destinationSchema) return undefined;
+      return crossAppRelationTripleSchemaToDbConcept({
+        node: r,
+        sourceNodeSchema: sourceSchema,
+        destinationNodeSchema: destinationSchema,
+      });
+    })
+    .filter((d) => d !== undefined);
+  const response = await client.rpc("upsert_concepts", {
+    v_space_id: spaceId,
+    data: asDbData,
+  });
+  if (response.error) throw response.error;
+  asDbData.map((d) => {
+    if (d.source_local_id) existing.add(d.source_local_id);
+  });
+};
+
+const getCrossAppNodeSchemaById = (): Record<string, CrossAppNodeSchema> => {
+  const allNodeSchemas = getDiscourseNodes();
+  const result: CrossAppNodeSchema[] = allNodeSchemas
+    .map((s) => {
+      const relData = window.roamAlphaAPI.pull(
+        "[:create/time :edit/time {:create/user [:user/uid]}]",
+        `[:block/uid "${s.type}"]`,
+      ) as unknown as {
+        ":create/time": number;
+        ":edit/time": number;
+        ":create/user": { ":user/uid": string };
+      };
+      if (!relData) return null;
+      const userUid = relData[":create/user"][":user/uid"];
+      return {
+        localId: s.type,
+        label: s.text,
+        authorId: userUid,
+        createdAt: new Date(relData[":create/time"]),
+      };
+    })
+    .filter((s) => s !== null);
+  return Object.fromEntries(result.map((r) => [r.localId, r]));
+};
+
+export const publishCorrespondingRelations = async ({
+  client,
+  spaceId,
+  groupIds,
+  forNodeIds,
+}: {
+  client: DGSupabaseClient;
+  spaceId: number;
+  groupIds: string[];
+  forNodeIds?: Set<string>;
+}): Promise<string[] | null> => {
+  const allRelationsSchemas = getDiscourseRelations();
+  const allRelationSchemasById = Object.fromEntries(
+    allRelationsSchemas.map((s) => [s.id, s]),
+  );
+  const nodeSchemaById = getCrossAppNodeSchemaById();
+  // Should we even handle non-reified relations?
+  // I need a way to know if a relation is imported.
+  const allRelations = await getReifiedRelations();
+  const failedGroupIds: string[] = [];
+  const sourceIdOfNodes: Record<string, number> = {};
+  const syncedRelationIds = new Set(
+    await getSyncedRelationLocalIds(client, spaceId),
+  );
+  const isImportedFrom = (nodeLocalId: string): number => {
+    let cached = sourceIdOfNodes[nodeLocalId];
+    if (cached === undefined) {
+      cached = sourceIdOfNodes[nodeLocalId] =
+        getSpaceIdOf(nodeLocalId) || spaceId;
+    }
+    return cached === spaceId ? 0 : cached;
+  };
+  const relations =
+    forNodeIds !== undefined
+      ? allRelations.filter(
+          (r) =>
+            r.importedFromRid === undefined &&
+            (forNodeIds.has(r.sourceUid) || forNodeIds.has(r.destinationUid)),
+        )
+      : allRelations.filter((r) => r.importedFromRid === undefined);
+  for (const groupId of groupIds) {
+    const groupSpaces = await getGroupSpaceIdAndUrls(client, groupId);
+    const publishedIds = new Set(
+      await getAllPublishedIds(client, spaceId, groupId),
+    );
+    const relsBetweenPublishedAll = relations.filter(
+      (r) =>
+        (publishedIds.has(r.sourceUid) ||
+          (isImportedFrom(r.sourceUid) || 0) in groupSpaces) &&
+        (publishedIds.has(r.destinationUid) ||
+          (isImportedFrom(r.destinationUid) || 0) in groupSpaces),
+    );
+    const relationSchemaIds = new Set(
+      relsBetweenPublishedAll
+        .map((r) => r.hasSchema)
+        // filter out deleted schemas
+        .filter((id) => id in allRelationSchemasById),
+    );
+    const relationSchemaTriples = allRelationsSchemas.filter((r) =>
+      relationSchemaIds.has(r.id),
+    );
+    const relsBetweenPublished = relsBetweenPublishedAll.filter((r) =>
+      relationSchemaIds.has(r.hasSchema),
+    );
+    await ensureRelationSchemasSynced({
+      client,
+      spaceId,
+      relationSchemas: relationSchemaTriples,
+      existing: syncedRelationIds,
+      nodeSchemaById,
+    });
+    await ensureRelationsSynced({
+      client,
+      spaceId,
+      relations: relsBetweenPublished,
+      existing: syncedRelationIds,
+      isImportedFrom,
+      spaceUids: groupSpaces,
+    });
+    const missingIds = [
+      ...relsBetweenPublished
+        .map((r) => r.relationId)
+        .filter((id) => !publishedIds.has(id)),
+      ...relationSchemaTriples
+        .map((r) => r.id)
+        .filter((id) => !publishedIds.has(id)),
+    ];
+    const grantRes = await client.from("ResourceAccess").upsert(
+      missingIds.map((sourceLocalId) => ({
+        account_uid: groupId,
+        source_local_id: sourceLocalId,
+        space_id: spaceId,
+      })),
+      { ignoreDuplicates: true },
+    );
+    if (!isIgnorableUpsertError(grantRes.error)) {
+      failedGroupIds.push(groupId);
+    }
+  }
+  return failedGroupIds.length ? failedGroupIds : null;
+};
+
 // 23505 = unique_violation: the grant already exists, which counts as success.
 const isIgnorableUpsertError = (error: { code?: string } | null): boolean =>
   !error || error.code === "23505";
@@ -121,6 +441,12 @@ export const publishNodesToGroups = async ({
 
     result.okGroupIds.push(groupId);
   }
+  await publishCorrespondingRelations({
+    client,
+    spaceId,
+    groupIds: targetGroupIds,
+    forNodeIds: new Set(syncedNodeUids),
+  });
 
   result.publishedNodeUids = result.okGroupIds.length > 0 ? syncedNodeUids : [];
   return result;

packages/database/src/inputTypes.ts

@@ -6,7 +6,7 @@ export type LocalAccountDataInput = Partial<
 export type LocalDocumentDataInput = Partial<
   Omit<
     Database["public"]["CompositeTypes"]["document_local_input"],
-    "author_inline"
+    "author_inline" | "contents"
   > & { author_inline: LocalAccountDataInput }
 >;
 export type LocalContentDataInput = Partial<