Implement ATJSON canonical content storage

Author: mdroidian

apps/obsidian/package.json

@@ -9,6 +9,7 @@
     "build": "tsx scripts/build.ts",
     "lint": "eslint .",
     "lint:fix": "eslint . --fix",
+    "test": "node --import tsx --test tests/atjsonContentWrite.test.ts",
     "publish": "tsx scripts/publish.ts --version 0.1.0",
     "check-types": "tsc --noEmit --skipLibCheck"
   },
@@ -39,6 +40,7 @@
   },
   "dependencies": {
     "@codemirror/view": "^6.38.8",
+    "@repo/content-model": "workspace:*",
     "@repo/database": "workspace:*",
     "@repo/utils": "workspace:*",
     "@supabase/supabase-js": "catalog:",

apps/obsidian/src/utils/importContentTypes.ts

@@ -0,0 +1,49 @@
+import {
+  TEXT_MARKDOWN_CONTENT_TYPE,
+  TEXT_PLAIN_CONTENT_TYPE,
+} from "@repo/content-model";
+
+export type ObsidianImportContentVariant = "direct" | "full";
+
+type ObsidianImportContentRow = {
+  variant: string | null;
+  // eslint-disable-next-line @typescript-eslint/naming-convention
+  content_type: string | null;
+};
+
+export const OBSIDIAN_IMPORT_CONTENT_TYPES = [
+  TEXT_PLAIN_CONTENT_TYPE,
+  TEXT_MARKDOWN_CONTENT_TYPE,
+] as const;
+
+export const getContentTypeForObsidianImportVariant = (
+  variant: ObsidianImportContentVariant,
+): (typeof OBSIDIAN_IMPORT_CONTENT_TYPES)[number] =>
+  variant === "full" ? TEXT_MARKDOWN_CONTENT_TYPE : TEXT_PLAIN_CONTENT_TYPE;
+
+export const isObsidianImportDirectRow = (
+  row: ObsidianImportContentRow,
+): boolean =>
+  row.variant === "direct" &&
+  (row.content_type ?? TEXT_PLAIN_CONTENT_TYPE) === TEXT_PLAIN_CONTENT_TYPE;
+
+export const isObsidianImportFullRow = (
+  row: ObsidianImportContentRow,
+): boolean =>
+  row.variant === "full" &&
+  (row.content_type ?? TEXT_MARKDOWN_CONTENT_TYPE) ===
+    TEXT_MARKDOWN_CONTENT_TYPE;
+
+export const selectObsidianImportContentRows = <
+  T extends ObsidianImportContentRow,
+>(
+  rows: T[],
+): {
+  direct: T | undefined;
+  full: T | undefined;
+} => ({
+  direct: rows.find(isObsidianImportDirectRow),
+  full: rows.find(isObsidianImportFullRow),
+});
+
+export { TEXT_MARKDOWN_CONTENT_TYPE, TEXT_PLAIN_CONTENT_TYPE };

apps/obsidian/src/utils/importNodes.ts

@@ -1,5 +1,11 @@
 /* eslint-disable @typescript-eslint/naming-convention */
 import type { Json } from "@repo/database/dbTypes";
+import {
+  OBSIDIAN_IMPORT_CONTENT_TYPES,
+  TEXT_PLAIN_CONTENT_TYPE,
+  getContentTypeForObsidianImportVariant,
+  selectObsidianImportContentRows,
+} from "./importContentTypes";
 import matter from "gray-matter";
 import { App, Notice, TFile } from "obsidian";
 import type { DGSupabaseClient } from "@repo/database/lib/client";
@@ -66,9 +72,10 @@ export const getPublishedNodesForGroups = async ({
   const { data, error } = await client
     .from("my_contents")
     .select(
-      "source_local_id, space_id, text, created, last_modified, variant, metadata, author_id",
+      "source_local_id, space_id, text, created, last_modified, variant, content_type, metadata, author_id",
     )
-    .neq("space_id", currentSpaceId);
+    .neq("space_id", currentSpaceId)
+    .in("content_type", OBSIDIAN_IMPORT_CONTENT_TYPES);
 
   if (error) {
     console.error("Error fetching published nodes:", error);
@@ -86,6 +93,7 @@ export const getPublishedNodesForGroups = async ({
     created: string | null;
     last_modified: string | null;
     variant: string | null;
+    content_type: string | null;
     author_id: number | null;
     metadata: Json;
   };
@@ -109,7 +117,7 @@ export const getPublishedNodesForGroups = async ({
     const latest = withDate.reduce((a, b) =>
       (a.last_modified ?? "") >= (b.last_modified ?? "") ? a : b,
     );
-    const direct = rows.find((r) => r.variant === "direct");
+    const { direct } = selectObsidianImportContentRows(rows);
     const text = direct?.text ?? latest.text ?? "";
     const createdAt = latest.created
       ? new Date(latest.created + "Z").valueOf()
@@ -267,6 +275,7 @@ export const fetchNodeContent = async ({
     .eq("source_local_id", nodeInstanceId)
     .eq("space_id", spaceId)
     .eq("variant", variant)
+    .eq("content_type", getContentTypeForObsidianImportVariant(variant))
     .maybeSingle();
 
   if (error || !data || data.text == null) {
@@ -301,6 +310,7 @@ export const fetchNodeContentWithMetadata = async ({
     .eq("source_local_id", nodeInstanceId)
     .eq("space_id", spaceId)
     .eq("variant", variant)
+    .eq("content_type", getContentTypeForObsidianImportVariant(variant))
     .maybeSingle();
 
   if (error || !data || data.text == null) {
@@ -342,10 +352,13 @@ const fetchNodeContentForImport = async ({
 } | null> => {
   const { data, error } = await client
     .from("my_contents")
-    .select("text, created, last_modified, variant, metadata, author_id")
+    .select(
+      "text, created, last_modified, variant, content_type, metadata, author_id",
+    )
     .eq("source_local_id", nodeInstanceId)
     .eq("space_id", spaceId)
-    .in("variant", ["direct", "full"]);
+    .in("variant", ["direct", "full"])
+    .in("content_type", OBSIDIAN_IMPORT_CONTENT_TYPES);
 
   if (error) {
     console.error("Error fetching node content for import:", error);
@@ -358,10 +371,10 @@ const fetchNodeContentForImport = async ({
     last_modified: string | null;
     author_id: number | null;
     variant: string | null;
+    content_type: string | null;
     metadata: Json;
   }>;
-  const direct = rows.find((r) => r.variant === "direct");
-  const full = rows.find((r) => r.variant === "full");
+  const { direct, full } = selectObsidianImportContentRows(rows);
   const authorId = full?.author_id ?? direct?.author_id ?? null;
 
   if (
@@ -412,6 +425,7 @@ export const getSourceContentDates = async ({
     .eq("source_local_id", nodeInstanceId)
     .eq("space_id", spaceId)
     .eq("variant", "direct")
+    .eq("content_type", TEXT_PLAIN_CONTENT_TYPE)
     .maybeSingle();
   if (error || !data) return null;
   return {
@@ -1542,6 +1556,7 @@ export const refreshImportedFile = async ({
     .eq("space_id", spaceId)
     .eq("source_local_id", frontmatter.nodeInstanceId)
     .eq("variant", "direct")
+    .eq("content_type", TEXT_PLAIN_CONTENT_TYPE)
     .maybeSingle();
   const metadata = metadataResp.data?.metadata;
   const filePath: string | undefined =

apps/obsidian/src/utils/publishNode.ts

@@ -21,6 +21,7 @@ import { isProvisionalSchema } from "./typeUtils";
 import type { DiscourseNodeInVault } from "./getDiscourseNodes";
 import type { SupabaseContext } from "./supabaseContext";
 import type { TablesInsert } from "@repo/database/dbTypes";
+import { TEXT_MARKDOWN_CONTENT_TYPE } from "@repo/content-model";
 
 const publishSchema = async ({
   client,
@@ -445,6 +446,7 @@ export const publishNodeToGroup = async ({
     .eq("source_local_id", nodeId)
     .eq("space_id", spaceId)
     .eq("variant", "full")
+    .eq("content_type", TEXT_MARKDOWN_CONTENT_TYPE)
     .maybeSingle();
   if (idResponse.error || !idResponse.data) {
     throw idResponse.error || new Error("no data while fetching node");

apps/obsidian/src/utils/syncDgNodesToSupabase.ts

@@ -23,6 +23,10 @@ import {
 } from "./conceptConversion";
 import { loadRelations } from "~/utils/relationsStore";
 import type { LocalConceptDataInput } from "@repo/database/inputTypes";
+import {
+  TEXT_MARKDOWN_CONTENT_TYPE,
+  TEXT_PLAIN_CONTENT_TYPE,
+} from "@repo/content-model";
 import {
   type DiscourseNodeInVault,
   collectDiscourseNodesFromVault,
@@ -59,6 +63,7 @@ const getAllNodeInstanceIdsFromSupabase = async (
       .select("source_local_id")
       .eq("space_id", spaceId)
       .eq("scale", "document")
+      .eq("content_type", TEXT_PLAIN_CONTENT_TYPE)
       .not("source_local_id", "is", null);
 
     if (error) {
@@ -168,6 +173,7 @@ const getLastContentSyncTime = async (
     .from("my_contents")
     .select("last_modified")
     .eq("space_id", spaceId)
+    .in("content_type", [TEXT_PLAIN_CONTENT_TYPE, TEXT_MARKDOWN_CONTENT_TYPE])
     .order("last_modified", { ascending: false })
     .limit(1)
     .maybeSingle();
@@ -273,6 +279,7 @@ const getExistingTitlesFromDatabase = async (
       .select("source_local_id, text")
       .eq("space_id", spaceId)
       .eq("variant", "direct")
+      .eq("content_type", TEXT_PLAIN_CONTENT_TYPE)
       .in("source_local_id", nodeInstanceIds);
 
   if (directError) {

apps/obsidian/src/utils/upsertNodesAsContentWithEmbeddings.ts

@@ -1,15 +1,21 @@
 /* eslint-disable @typescript-eslint/naming-convention */
 import { nextApiRoot } from "@repo/utils/execContext";
-import { DGSupabaseClient } from "@repo/database/lib/client";
-import { Json, CompositeTypes } from "@repo/database/dbTypes";
-import { SupabaseContext } from "./supabaseContext";
-import { ObsidianDiscourseNodeData, ChangeType } from "./syncDgNodesToSupabase";
-import { default as DiscourseGraphPlugin } from "~/index";
+import type { DGSupabaseClient } from "@repo/database/lib/client";
+import type { Json, CompositeTypes } from "@repo/database/dbTypes";
+import {
+  DG_ATJSON_CONTENT_TYPE,
+  TEXT_MARKDOWN_CONTENT_TYPE,
+  TEXT_PLAIN_CONTENT_TYPE,
+  createDgAtJsonMetadata,
+  derivePlainTextFromDgDocument,
+  obsidianMarkdownToDgDocument,
+} from "@repo/content-model";
+import type { SupabaseContext } from "./supabaseContext";
+import type { ObsidianDiscourseNodeData } from "./syncDgNodesToSupabase";
+import type DiscourseGraphPlugin from "~/index";
 
 type LocalContentDataInput = Partial<CompositeTypes<"content_local_input">>;
 
-type ContentVariant = "direct" | "full";
-
 const EMBEDDING_BATCH_SIZE = 200;
 const EMBEDDING_MODEL = "openai_text_embedding_3_small_1536";
 
@@ -19,31 +25,16 @@ type EmbeddingApiResponse = {
   }[];
 };
 
-/**
- * Determine which content variants to create based on change types
- */
-const getVariantsToCreate = (changeTypes: ChangeType[]): ContentVariant[] => {
-  const variants: ContentVariant[] = [];
-
-  if (changeTypes.includes("title")) {
-    variants.push("direct");
-  }
-
-  if (changeTypes.includes("content")) {
-    variants.push("full");
-  }
-
-  return variants;
-};
-
 const createNodeContentEntries = async (
   node: ObsidianDiscourseNodeData,
   accountLocalId: string,
   plugin: DiscourseGraphPlugin,
 ): Promise<LocalContentDataInput[]> => {
-  const variantsToCreate = getVariantsToCreate(node.changeTypes);
+  const shouldWriteDirect = node.changeTypes.includes("title");
+  const shouldWriteMarkdown = node.changeTypes.includes("content");
+  const shouldWriteAtJson = shouldWriteDirect || shouldWriteMarkdown;
 
-  if (variantsToCreate.length === 0) {
+  if (!shouldWriteDirect && !shouldWriteMarkdown && !shouldWriteAtJson) {
     return [];
   }
 
@@ -59,25 +50,45 @@ const createNodeContentEntries = async (
   const entries: LocalContentDataInput[] = [];
 
   // Create direct entry (title) if needed - will get embeddings
-  if (variantsToCreate.includes("direct")) {
+  if (shouldWriteDirect) {
     entries.push({
       ...baseEntry,
       text: node.file.basename,
       variant: "direct",
+      content_type: TEXT_PLAIN_CONTENT_TYPE,
       metadata: { filePath: node.file.path },
     });
   }
 
-  // Create full entry (content) if needed - no embeddings
-  if (variantsToCreate.includes("full")) {
+  if (shouldWriteMarkdown || shouldWriteAtJson) {
     try {
       const fullContent = await plugin.app.vault.read(node.file);
-      entries.push({
-        ...baseEntry,
-        text: fullContent,
-        variant: "full",
-        metadata: node.frontmatter as Json,
+      if (shouldWriteMarkdown) {
+        entries.push({
+          ...baseEntry,
+          text: fullContent,
+          variant: "full",
+          content_type: TEXT_MARKDOWN_CONTENT_TYPE,
+          metadata: node.frontmatter as Json,
+        });
+      }
+      const document = obsidianMarkdownToDgDocument({
+        title: node.file.basename,
+        markdown: fullContent,
+        metadata: {
+          filePath: node.file.path,
+          frontmatter: node.frontmatter as Json,
+        },
       });
+      if (shouldWriteAtJson) {
+        entries.push({
+          ...baseEntry,
+          text: derivePlainTextFromDgDocument(document),
+          variant: "full",
+          content_type: DG_ATJSON_CONTENT_TYPE,
+          metadata: createDgAtJsonMetadata({ document }) as unknown as Json,
+        });
+      }
     } catch (error) {
       console.error(`Error reading file content for ${node.file.path}:`, error);
     }
@@ -202,18 +213,24 @@ export const upsertNodesToSupabaseAsContentWithEmbeddings = async ({
     return;
   }
 
-  // Create two entries per node: one "direct" (title) and one "full" (content)
+  // Create representation rows based on the changed slice: title, Markdown body, and canonical ATJSON.
   const allContentEntries = await convertObsidianNodeToLocalContent({
     nodes: obsidianNodes,
     accountLocalId,
     plugin,
   });
 
   const directVariantEntries = allContentEntries.filter(
-    (entry) => entry.variant === "direct",
+    (entry) =>
+      entry.variant === "direct" &&
+      (entry.content_type ?? TEXT_PLAIN_CONTENT_TYPE) ===
+        TEXT_PLAIN_CONTENT_TYPE,
   );
   const fullVariantEntries = allContentEntries.filter(
-    (entry) => entry.variant === "full",
+    (entry) =>
+      entry.variant === "full" ||
+      (entry.content_type ?? TEXT_PLAIN_CONTENT_TYPE) !==
+        TEXT_PLAIN_CONTENT_TYPE,
   );
 
   let directEntriesWithEmbeddings: LocalContentDataInput[];
@@ -223,7 +240,7 @@ export const upsertNodesToSupabaseAsContentWithEmbeddings = async ({
   } catch (error: unknown) {
     const errorMessage = error instanceof Error ? error.message : String(error);
     console.error(
-      `upsertNodesToSupabaseAsContentWithEmbeddings: Embedding service failed – ${errorMessage}`,
+      `upsertNodesToSupabaseAsContentWithEmbeddings: Embedding service failed - ${errorMessage}`,
     );
     throw new Error(errorMessage);
   }