From 6ce0020e192eb041791936f4726afd5cf092dec9 Mon Sep 17 00:00:00 2001 From: Steve Date: Tue, 13 Jan 2026 23:15:34 -0500 Subject: [PATCH] chore: added verification to data flow --- packages/server/schema.sql | 4 +- packages/server/src/index.ts | 4 +- packages/server/src/routes/admin.ts | 110 +++++++++++----------- packages/server/src/routes/feed.ts | 7 +- packages/server/src/routes/index.ts | 2 +- packages/server/src/types/index.ts | 1 + packages/server/src/utils/document.ts | 16 ++-- packages/server/src/utils/index.ts | 1 + packages/server/src/utils/verification.ts | 99 +++++++++++++++++++ packages/server/tables.csv | 51 ---------- 10 files changed, 176 insertions(+), 119 deletions(-) create mode 100644 packages/server/src/utils/verification.ts delete mode 100644 packages/server/tables.csv diff --git a/packages/server/schema.sql b/packages/server/schema.sql index b17aa64..9977fae 100644 --- a/packages/server/schema.sql +++ b/packages/server/schema.sql @@ -55,9 +55,11 @@ CREATE TABLE IF NOT EXISTS resolved_documents ( view_url TEXT, -- Constructed canonical URL (pub_url + path) pds_endpoint TEXT, -- Cached PDS endpoint for this DID resolved_at TEXT DEFAULT (datetime('now')), - stale_at TEXT -- When this record should be re-resolved + stale_at TEXT, -- When this record should be re-resolved + verified INTEGER DEFAULT 0 -- Whether the record has been verified via .well-known or link tag ); CREATE INDEX IF NOT EXISTS idx_resolved_documents_rkey ON resolved_documents(rkey DESC); CREATE INDEX IF NOT EXISTS idx_resolved_documents_stale ON resolved_documents(stale_at); CREATE INDEX IF NOT EXISTS idx_resolved_documents_pub_url ON resolved_documents(pub_url); +CREATE INDEX IF NOT EXISTS idx_resolved_documents_verified ON resolved_documents(verified); diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts index 649fce6..958dfaa 100644 --- a/packages/server/src/index.ts +++ b/packages/server/src/index.ts @@ -1,7 +1,7 @@ import { Hono } from "hono"; import { cors } from "hono/cors"; import type { Bindings } from "./types"; -import { health, webhook, feed, stats, records } from "./routes"; +import { health, webhook, feed, stats, records, admin } from "./routes"; import { processDocument } from "./utils"; const app = new Hono<{ Bindings: Bindings }>(); @@ -15,7 +15,7 @@ app.route("/webhook", webhook); app.route("/feed", feed); app.route("/stats", stats); app.route("/records", records); -//app.route("/admin", admin); +app.route("/admin", admin); // Legacy alias: /feed-raw -> /feed/raw app.get("/feed-raw", async (c) => { diff --git a/packages/server/src/routes/admin.ts b/packages/server/src/routes/admin.ts index 84990d0..05e656c 100644 --- a/packages/server/src/routes/admin.ts +++ b/packages/server/src/routes/admin.ts @@ -5,73 +5,73 @@ const admin = new Hono<{ Bindings: Bindings }>(); // Queue all documents for re-processing admin.post("/resolve-all", async (c) => { - try { - const db = c.env.DB; - const queue = c.env.RESOLUTION_QUEUE; + try { + const db = c.env.DB; + const queue = c.env.RESOLUTION_QUEUE; - // Get all records from repo_records - const { results } = await db - .prepare( - `SELECT did, rkey FROM repo_records - WHERE collection = 'site.standard.document'` - ) - .all<{ did: string; rkey: string }>(); + // Get all records from repo_records + const { results } = await db + .prepare( + `SELECT did, rkey FROM repo_records + WHERE collection = 'site.standard.document'`, + ) + .all<{ did: string; rkey: string }>(); - if (!results || results.length === 0) { - return c.json({ message: "No documents to process", queued: 0 }); - } + if (!results || results.length === 0) { + return c.json({ message: "No documents to process", queued: 0 }); + } - // Queue in batches of 100 (Cloudflare Queue limit) - const batchSize = 100; - let queued = 0; + // Queue in batches of 100 (Cloudflare Queue limit) + const batchSize = 100; + let queued = 0; - for (let i = 0; i < results.length; i += batchSize) { - const batch = results.slice(i, i + batchSize); - const messages = batch.map((row) => ({ - body: { - did: row.did, - collection: "site.standard.document", - rkey: row.rkey, - }, - })); + for (let i = 0; i < results.length; i += batchSize) { + const batch = results.slice(i, i + batchSize); + const messages = batch.map((row) => ({ + body: { + did: row.did, + collection: "site.standard.document", + rkey: row.rkey, + }, + })); - await queue.sendBatch(messages); - queued += messages.length; - } + await queue.sendBatch(messages); + queued += messages.length; + } - return c.json({ - message: "Documents queued for re-processing", - queued, - }); - } catch (error) { - return c.json( - { error: "Failed to queue documents", details: String(error) }, - 500 - ); - } + return c.json({ + message: "Documents queued for re-processing", + queued, + }); + } catch (error) { + return c.json( + { error: "Failed to queue documents", details: String(error) }, + 500, + ); + } }); // Mark all documents as stale (alternative - lets cron handle it) admin.post("/mark-stale", async (c) => { - try { - const db = c.env.DB; + try { + const db = c.env.DB; - const result = await db - .prepare( - `UPDATE resolved_documents SET stale_at = datetime('now', '-1 hour')` - ) - .run(); + const result = await db + .prepare( + `UPDATE resolved_documents SET stale_at = datetime('now', '-1 hour')`, + ) + .run(); - return c.json({ - message: "All documents marked as stale", - affected: result.meta.changes, - }); - } catch (error) { - return c.json( - { error: "Failed to mark documents as stale", details: String(error) }, - 500 - ); - } + return c.json({ + message: "All documents marked as stale", + affected: result.meta.changes, + }); + } catch (error) { + return c.json( + { error: "Failed to mark documents as stale", details: String(error) }, + 500, + ); + } }); export default admin; diff --git a/packages/server/src/routes/feed.ts b/packages/server/src/routes/feed.ts index dd2000e..a89c39d 100644 --- a/packages/server/src/routes/feed.ts +++ b/packages/server/src/routes/feed.ts @@ -83,7 +83,7 @@ feed.get("/raw", async (c) => { .prepare( `SELECT did, rkey FROM repo_records WHERE collection = 'site.standard.document' - ORDER BY rkey DESC + ORDER BY published_at DESC LIMIT ? OFFSET ?` ) .bind(limit, offset) @@ -116,9 +116,10 @@ feed.get("/", async (c) => { cover_image_cid, cover_image_url, bsky_post_ref, tags, published_at, updated_at, pub_url, pub_name, pub_description, pub_icon_cid, pub_icon_url, view_url, pds_endpoint, - resolved_at, stale_at + resolved_at, stale_at, verified FROM resolved_documents - ORDER BY rkey DESC + WHERE verified = 1 + ORDER BY published_at DESC LIMIT ? OFFSET ?` ) .bind(limit, offset) diff --git a/packages/server/src/routes/index.ts b/packages/server/src/routes/index.ts index a567fa7..5fbf8c4 100644 --- a/packages/server/src/routes/index.ts +++ b/packages/server/src/routes/index.ts @@ -3,4 +3,4 @@ export { default as webhook } from "./webhook"; export { default as feed } from "./feed"; export { default as stats } from "./stats"; export { default as records } from "./records"; -//export { default as admin } from "./admin"; +export { default as admin } from "./admin"; diff --git a/packages/server/src/types/index.ts b/packages/server/src/types/index.ts index 060bcd2..a712ada 100644 --- a/packages/server/src/types/index.ts +++ b/packages/server/src/types/index.ts @@ -98,4 +98,5 @@ export interface ResolvedDocumentRow { pds_endpoint: string | null; resolved_at: string | null; stale_at: string | null; + verified: number | null; } diff --git a/packages/server/src/utils/document.ts b/packages/server/src/utils/document.ts index 3c6ee8c..2fcba7a 100644 --- a/packages/server/src/utils/document.ts +++ b/packages/server/src/utils/document.ts @@ -1,6 +1,7 @@ import { resolvePds } from "./resolver"; import { parseAtUri } from "./at-uri"; import { buildBlobUrl, extractBlobCid } from "./blob"; +import { verifyDocumentRecord } from "./verification"; // Raw document record from PDS interface DocumentRecord { @@ -198,8 +199,11 @@ export async function processDocument( } } - // 6. Insert/update resolved_documents + // 6. Verify the document const uri = `at://${did}/${collection}/${rkey}`; + const verified = await verifyDocumentRecord(pubUrl, site, viewUrl, uri); + + // 7. Insert/update resolved_documents const STALE_OFFSET_HOURS = 12; await db @@ -209,26 +213,26 @@ export async function processDocument( cover_image_cid, cover_image_url, bsky_post_ref, tags, published_at, updated_at, pub_url, pub_name, pub_description, pub_icon_cid, pub_icon_url, view_url, pds_endpoint, - resolved_at, stale_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'), datetime('now', '+${STALE_OFFSET_HOURS} hours')) + resolved_at, stale_at, verified + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'), datetime('now', '+${STALE_OFFSET_HOURS} hours'), ?) ON CONFLICT(uri) DO UPDATE SET title = ?, description = ?, path = ?, site = ?, content = ?, text_content = ?, cover_image_cid = ?, cover_image_url = ?, bsky_post_ref = ?, tags = ?, published_at = ?, updated_at = ?, pub_url = ?, pub_name = ?, pub_description = ?, pub_icon_cid = ?, pub_icon_url = ?, view_url = ?, pds_endpoint = ?, - resolved_at = datetime('now'), stale_at = datetime('now', '+${STALE_OFFSET_HOURS} hours')` + resolved_at = datetime('now'), stale_at = datetime('now', '+${STALE_OFFSET_HOURS} hours'), verified = ?` ) .bind( // INSERT values uri, did, rkey, title, description, path, site, content, textContent, coverImageCid, coverImageUrl, bskyPostRef, tags, publishedAt, updatedAt, pubUrl, pubName, pubDescription, - pubIconCid, pubIconUrl, viewUrl, pds, + pubIconCid, pubIconUrl, viewUrl, pds, verified ? 1 : 0, // UPDATE values title, description, path, site, content, textContent, coverImageCid, coverImageUrl, bskyPostRef, tags, publishedAt, updatedAt, pubUrl, pubName, pubDescription, - pubIconCid, pubIconUrl, viewUrl, pds + pubIconCid, pubIconUrl, viewUrl, pds, verified ? 1 : 0 ) .run(); diff --git a/packages/server/src/utils/index.ts b/packages/server/src/utils/index.ts index aa181bb..1558992 100644 --- a/packages/server/src/utils/index.ts +++ b/packages/server/src/utils/index.ts @@ -2,3 +2,4 @@ export { parseAtUri, buildAtUri, type AtUriComponents } from "./at-uri"; export { resolvePds } from "./resolver"; export { resolveViewUrl, processDocument } from "./document"; export { buildBlobUrl, extractBlobCid } from "./blob"; +export { verifyPublication, verifyDocument, verifyDocumentRecord } from "./verification"; diff --git a/packages/server/src/utils/verification.ts b/packages/server/src/utils/verification.ts new file mode 100644 index 0000000..204b464 --- /dev/null +++ b/packages/server/src/utils/verification.ts @@ -0,0 +1,99 @@ +/** + * Verification utilities for standard.site records. + * + * Publications are verified via /.well-known/site.standard.publication + * Documents are verified via in HTML + */ + +/** + * Verifies a publication by checking /.well-known/site.standard.publication + * @param pubUrl The publication's base URL (e.g., "https://example.com") + * @param siteUri The expected AT-URI of the publication (e.g., "at://did:plc:abc/site.standard.publication/rkey") + * @returns true if the .well-known endpoint returns the matching AT-URI + */ +export async function verifyPublication( + pubUrl: string, + siteUri: string +): Promise { + try { + const baseUrl = pubUrl.startsWith("http") ? pubUrl : `https://${pubUrl}`; + const wellKnownUrl = `${baseUrl.replace(/\/$/, "")}/.well-known/site.standard.publication`; + + const response = await fetch(wellKnownUrl, { + headers: { Accept: "text/plain" }, + }); + + if (!response.ok) return false; + + const body = await response.text(); + return body.trim() === siteUri.trim(); + } catch { + return false; + } +} + +/** + * Verifies a document by checking for a matching tag + * @param viewUrl The document's canonical URL (e.g., "https://example.com/blog/post") + * @param documentUri The expected AT-URI of the document (e.g., "at://did:plc:abc/site.standard.document/rkey") + * @returns true if the HTML contains a matching link tag + */ +export async function verifyDocument( + viewUrl: string, + documentUri: string +): Promise { + try { + const response = await fetch(viewUrl, { + headers: { Accept: "text/html" }, + }); + + if (!response.ok) return false; + + const html = await response.text(); + + // Look for + // Using regex to avoid heavy HTML parser dependency + const linkPattern = + /]+rel=["']site\.standard\.document["'][^>]+href=["']([^"']+)["'][^>]*>/i; + const altPattern = + /]+href=["']([^"']+)["'][^>]+rel=["']site\.standard\.document["'][^>]*>/i; + + const match = html.match(linkPattern) || html.match(altPattern); + if (!match) return false; + + return match[1].trim() === documentUri.trim(); + } catch { + return false; + } +} + +/** + * Combined verification for a document record. + * Checks publication verification first (if applicable), then document verification. + * + * @param pubUrl The publication's base URL + * @param siteUri The AT-URI of the publication (from document's site field) + * @param viewUrl The document's canonical URL + * @param documentUri The AT-URI of the document + * @returns true if either publication or document verification passes + */ +export async function verifyDocumentRecord( + pubUrl: string | null, + siteUri: string | null, + viewUrl: string | null, + documentUri: string +): Promise { + // Try publication verification first (if we have a publication AT-URI) + if (pubUrl && siteUri && siteUri.startsWith("at://")) { + const pubVerified = await verifyPublication(pubUrl, siteUri); + if (pubVerified) return true; + } + + // Fall back to document verification (if we have a view URL) + if (viewUrl) { + const docVerified = await verifyDocument(viewUrl, documentUri); + if (docVerified) return true; + } + + return false; +} diff --git a/packages/server/tables.csv b/packages/server/tables.csv deleted file mode 100644 index 0813b62..0000000 --- a/packages/server/tables.csv +++ /dev/null @@ -1,51 +0,0 @@ -name,sql -_cf_KV,"CREATE TABLE _cf_KV ( - key TEXT PRIMARY KEY, - value BLOB - ) WITHOUT ROWID" -repo_records,"CREATE TABLE repo_records ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - did TEXT NOT NULL, - rkey TEXT NOT NULL, - collection TEXT NOT NULL, - cid TEXT, - synced_at TEXT DEFAULT (datetime('now')), - UNIQUE(did, collection, rkey) -)" -pds_cache,"CREATE TABLE pds_cache ( - did TEXT PRIMARY KEY, - pds_endpoint TEXT NOT NULL, - cached_at TEXT DEFAULT (datetime('now')) -)" -record_cache,"CREATE TABLE record_cache ( - uri TEXT PRIMARY KEY, - did TEXT NOT NULL, - collection TEXT NOT NULL, - rkey TEXT NOT NULL, - record_data TEXT NOT NULL, -- JSON blob - cached_at TEXT DEFAULT (datetime('now')) -)" -publication_cache,"CREATE TABLE publication_cache ( - at_uri TEXT PRIMARY KEY, - base_url TEXT NOT NULL, - cached_at TEXT DEFAULT (datetime('now')) -)" -sync_metadata,"CREATE TABLE sync_metadata ( - key TEXT PRIMARY KEY, - value TEXT NOT NULL, - updated_at TEXT DEFAULT (datetime('now')) -)" -resolved_documents,"CREATE TABLE resolved_documents ( - uri TEXT PRIMARY KEY, - did TEXT NOT NULL, - rkey TEXT NOT NULL, - title TEXT, - path TEXT, - site TEXT, - content TEXT, -- JSON blob - text_content TEXT, - published_at TEXT, - view_url TEXT, - resolved_at TEXT DEFAULT (datetime('now')), - stale_at TEXT -- When this record should be re-resolved -)" -- 2.51.2