feat: added atomicity to the standard site sync

Author: Kai-ros

app/components/BlogPost.server.vue

@@ -15,6 +15,7 @@ const props = defineProps<{ title: string; htmlContent: string; date: string }>(
       class="text-xs text-fg-subtle"
     />
 
+    <!-- INFO: v-html can only be used in ssr on html elements, not components -->
     <div class="" v-html="htmlContent" />
   </main>
 </template>

modules/standard-site-sync.ts

@@ -5,124 +5,133 @@ import { safeParse } from 'valibot'
 import * as site from '../shared/types/lexicons/site'
 import { BlogPostSchema } from '../shared/schemas/blog'
 import { NPMX_SITE } from '../shared/utils/constants'
+import { parseBasicFrontmatter } from '../shared/utils/parse-basic-frontmatter'
+import { TID } from '@atproto/common'
+import { Client } from '@atproto/lex'
 
 const syncedDocuments = new Map<string, string>()
+const CLOCK_ID_THREE = 3
+const DATE_TO_MICROSECONDS = 1000
 
+// TODO: Currently logging quite a lot, can remove some later if we want
 export default defineNuxtModule({
   meta: { name: 'standard-site-sync' },
   async setup() {
     const nuxt = useNuxt()
     const { resolve } = createResolver(import.meta.url)
     const contentDir = resolve('../app/pages/blog')
 
+    // Authentication with PDS using an app password
+    const pdsUrl = process.env.NPMX_PDS_URL
+    if (!pdsUrl) {
+      console.warn('[standard-site-sync] NPMX_PDS_URL not set, skipping sync')
+      return
+    }
+    // Instantiate a single new client instance that is reused for every file
+    const client = new Client(pdsUrl)
+
     if (nuxt.options._prepare) return
 
     nuxt.hook('build:before', async () => {
-      const glob = await import('fast-glob').then(m => m.default)
-      const files = await glob(`${contentDir}/**/*.md`)
-
-      for (const file of files) {
-        await syncFile(file, NPMX_SITE)
+      const { globby } = await import('globby')
+      const files: string[] = await globby(`${contentDir}/**/*.md`)
+
+      // INFO: Arbitrarily chosen concurrency limit, can be changed if needed
+      const concurrencyLimit = 5
+      for (let i = 0; i < files.length; i += concurrencyLimit) {
+        const batch = files.slice(i, i + concurrencyLimit)
+        // Process files in parallel
+        await Promise.all(
+          batch.map(file =>
+            syncFile(file, NPMX_SITE, client).catch(error =>
+              console.error(`[standard-site-sync] Error in ${file}:` + error),
+            ),
+          ),
+        )
       }
     })
 
-    nuxt.hook('builder:watch', async (_event, path) => {
-      if (path.endsWith('.md')) {
-        await syncFile(resolve(nuxt.options.rootDir, path), NPMX_SITE)
+    nuxt.hook('builder:watch', async (event, path) => {
+      if (!path.endsWith('.md')) return
+
+      // Ignore deleted files
+      if (event === 'unlink') {
+        console.log(`[standard-site-sync] File deleted: ${path}`)
+        return
       }
+
+      // Process add/change events only
+      await syncFile(resolve(nuxt.options.rootDir, path), NPMX_SITE, client).catch(err =>
+        console.error(`[standard-site-sync] Failed ${path}:`, err),
+      )
     })
   },
 })
 
-// TODO: Placeholder algo, can likely be simplified
-function parseBasicFrontmatter(fileContent: string): Record<string, any> {
-  const match = fileContent.match(/^---\r?\n([\s\S]+?)\r?\n---/)
-  if (!match) return {}
-
-  const obj: Record<string, any> = {}
-  const lines = match[1]?.split('\n')
-
-  if (!lines) return {}
-
-  for (const line of lines) {
-    const [key, ...valParts] = line.split(':')
-    if (key && valParts.length) {
-      let value = valParts.join(':').trim()
-
-      // Remove surrounding quotes
-      value = value.replace(/^["']|["']$/g, '')
-
-      // Handle Booleans
-      if (value === 'true') {
-        obj[key.trim()] = true
-        continue
-      }
-      if (value === 'false') {
-        obj[key.trim()] = false
-        continue
-      }
+/*
+ * INFO: Loads record to atproto and ensures uniqueness by checking the date the article is published
+ * publishedAt is an id that does not change
+ * Atomicity is enforced with upsert using publishedAt so we always update existing records instead of creating new ones
+ * Clock id(3) provides a deterministic ID
+ * WARN: DOES NOT CATCH ERRORS, THIS MUST BE HANDLED
+ */
+const syncFile = async (filePath: string, siteUrl: string, client: Client) => {
+  const fileContent = readFileSync(filePath, 'utf-8')
+  const frontmatter = parseBasicFrontmatter(fileContent)
+
+  // Schema expects 'path' & frontmatter provides 'slug'
+  const normalizedFrontmatter = {
+    ...frontmatter,
+    path: typeof frontmatter.slug === 'string' ? `/blog/${frontmatter.slug}` : frontmatter.path,
+  }
 
-      // Handle basic array [tag1, tag2]
-      if (value.startsWith('[') && value.endsWith(']')) {
-        obj[key.trim()] = value
-          .slice(1, -1)
-          .split(',')
-          .map(s => s.trim().replace(/^["']|["']$/g, ''))
-      } else {
-        obj[key.trim()] = value
-      }
-    }
+  const result = safeParse(BlogPostSchema, normalizedFrontmatter)
+  if (!result.success) {
+    console.warn(`[standard-site-sync] Validation failed for ${filePath}`, result.issues)
+    return
   }
-  return obj
-}
 
-const syncFile = async (filePath: string, siteUrl: string) => {
-  try {
-    const fileContent = readFileSync(filePath, 'utf-8')
-    const frontmatter = parseBasicFrontmatter(fileContent)
+  const data = result.output
 
-    // Schema expects 'path' & frontmatter provides 'slug'
-    if (frontmatter.slug) {
-      frontmatter.path = `/blog/${frontmatter.slug}`
+  // filter drafts
+  if (data.draft) {
+    if (process.env.DEBUG === 'true') {
+      console.debug(`[standard-site-sync] Skipping draft: ${data.path}`)
     }
+    return
+  }
 
-    const result = safeParse(BlogPostSchema, frontmatter)
-    if (!result.success) {
-      console.warn(`[standard-site-sync] Validation failed for ${filePath}`, result.issues)
-      return
-    }
+  // Keys are sorted to provide a more stable hash
+  const hash = createHash('sha256')
+    .update(JSON.stringify(data, Object.keys(data).sort()))
+    .digest('hex')
 
-    const data = result.output
+  if (syncedDocuments.get(data.path) === hash) {
+    return
+  }
 
-    // filter drafts
-    if (data.draft) {
-      if (process.env.DEBUG === 'true') {
-        console.debug(`[standard-site-sync] Skipping draft: ${data.path}`)
-      }
-      return
-    }
+  const document = site.standard.document.$build({
+    site: siteUrl as `${string}:${string}`,
+    path: data.path,
+    title: data.title,
+    description: data.description ?? data.excerpt,
+    tags: data.tags,
+    // This can be extended to update the site.standard.document .updatedAt if it is changed and use the posts date here
+    publishedAt: new Date(data.date).toISOString(),
+  })
 
-    const hash = createHash('sha1').update(JSON.stringify(data)).digest('hex')
+  const dateInMicroSeconds = new Date(result.output.date).getTime() * DATE_TO_MICROSECONDS
 
-    if (syncedDocuments.get(data.path) === hash) {
-      return
-    }
+  // Clock id(3) needs to be the same everytime to get the same TID from a timestamp
+  const tid = TID.fromTime(dateInMicroSeconds, CLOCK_ID_THREE)
 
-    // TODO: Review later
-    const document = site.standard.document.$build({
-      site: siteUrl as `${string}:${string}`,
-      path: data.path,
-      title: data.title,
-      description: data.description ?? data.excerpt,
-      tags: data.tags,
-      publishedAt: new Date(data.date).toISOString(),
-    })
+  // client.put is async and needs to be awaited
+  await client.put(site.standard.document, document, {
+    rkey: tid.str,
+  })
 
-    console.log('[standard-site-sync] Pushing:', JSON.stringify(document, null, 2))
-    // TODO: Real PDS push
+  // TODO: Replace with real PDS push
+  console.log('[standard-site-sync] Pushing:', JSON.stringify(document, null, 2))
 
-    syncedDocuments.set(data.path, hash)
-  } catch (error) {
-    console.error(`[standard-site-sync] Error in ${filePath}:`, error)
-  }
+  syncedDocuments.set(data.path, hash)
 }

package.json

@@ -36,8 +36,9 @@
     "start:playwright:webserver": "NODE_ENV=test pnpm build && pnpm preview --port 5678"
   },
   "dependencies": {
-    "@atproto/api": "^0.18.17",
     "@atcute/bluesky-richtext-segmenter": "3.0.0",
+    "@atproto/api": "^0.18.17",
+    "@atproto/common": "0.5.10",
     "@atproto/lex": "0.0.13",
     "@atproto/oauth-client-node": "^0.3.15",
     "@deno/doc": "jsr:^0.189.1",
@@ -71,6 +72,7 @@
     "@vueuse/nuxt": "14.2.0",
     "@vueuse/router": "^14.2.0",
     "focus-trap": "^7.8.0",
+    "globby": "16.1.0",
     "marked": "17.0.1",
     "module-replacements": "2.11.0",
     "nuxt": "4.3.0",
@@ -83,6 +85,7 @@
     "simple-git": "3.30.0",
     "spdx-license-list": "6.11.0",
     "std-env": "3.10.0",
+    "tinyglobby": "0.2.15",
     "ufo": "1.6.3",
     "unocss": "66.6.0",
     "unplugin-vue-router": "0.19.2",

pnpm-lock.yaml

@@ -0,0 +1,34 @@
+export function parseBasicFrontmatter(fileContent: string): Record<string, unknown> {
+  const match = fileContent.match(/^---\s*\n([\s\S]*?)\n---\s*(?:\n|$)/)
+  if (!match?.[1]) return {}
+
+  return match[1].split('\n').reduce(
+    (acc, line) => {
+      const idx = line.indexOf(':')
+      if (idx === -1) return acc
+
+      const key = line.slice(0, idx).trim()
+
+      // Remove surrounding quotes
+      let value = line
+        .slice(idx + 1)
+        .trim()
+        .replace(/^["']|["']$/g, '')
+
+      // Type coercion (handles 123, 45.6, boolean, arrays)
+      if (value === 'true') acc[key] = true
+      else if (value === 'false') acc[key] = false
+      else if (/^-?\d+$/.test(value)) acc[key] = parseInt(value, 10)
+      else if (/^-?\d+\.\d+$/.test(value)) acc[key] = parseFloat(value)
+      else if (value.startsWith('[') && value.endsWith(']')) {
+        acc[key] = value
+          .slice(1, -1)
+          .split(',')
+          .map(s => s.trim().replace(/^["']|["']$/g, ''))
+      } else acc[key] = value
+
+      return acc
+    },
+    {} as Record<string, unknown>,
+  )
+}