chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
import { validateRecord } from '@bigmoves/lexicon';
import { domainToLexiconDns, nsidToDomains } from './authority.js';
import { resolveDid } from './did.js';
import { lookupTxt, parseDidFromTxt } from './dns.js';
import { coreSchemas } from './schemas/com-atproto.js';
import { knownRecords } from './schemas/known-records.js';
import { publishedSchemas } from './schemas/published.js';
/**
* @typedef {import('@pdsjs/core/ports').ValidationStatus} ValidationStatus
* @typedef {import('@bigmoves/lexicon').Lexicon} Lexicon
*/
/**
* Lexicon resolver with full AT Protocol spec compliance.
* Resolves schemas via DNS authority lookup, DID resolution, and PDS fetch.
* Recursively resolves all referenced schemas for proper validation.
* Validates records using @bigmoves/lexicon.
*
* Implements LexiconResolverPort.
*/
export class LexiconResolver {
/**
* @param {Object} [opts]
* @param {number} [opts.ttl=3600000] - Cache TTL in ms (default 1 hour)
* @param {number} [opts.negativeTtl=60000] - TTL for cached "not published"
* answers (default 1 minute), kept short so a transient resolution failure
* isn't pinned for the full TTL
* @param {typeof fetch} [opts.fetch] - Fetch implementation
* @param {string} [opts.dohUrl] - DNS-over-HTTPS endpoint
* @param {string} [opts.plcUrl] - PLC directory URL
* @param {Lexicon[]} [opts.schemas] - Pre-loaded static schemas
* @param {string[]} [opts.enforce] - Collections whose validation failures
* reject the write even when the lexicon came off the network. Defaults to
* the set the reference PDS validates strictly; pre-loaded schemas are
* always included.
*/
constructor(opts = {}) {
/** @type {Map<string, Lexicon>} */
this.schemas = new Map();
/** @type {Map<string, {fetchedAt: number, ttl: number}>} */
this.cache = new Map();
/** @type {Map<string, Promise<boolean>>} */
this.pendingFetches = new Map();
// Every NSID under one authority shares a DNS record and a DID document.
// Resolving a record type pulls in a dozen schemas from the same authority,
// so these two caches carry most of the saving.
/** @type {Map<string, {value: string|null, fetchedAt: number, ttl: number}>} */
this.authorities = new Map();
/** @type {Map<string, Promise<string|null>>} */
this.pendingAuthorities = new Map();
/** @type {Map<string, {value: string|null, fetchedAt: number, ttl: number}>} */
this.endpoints = new Map();
/** @type {Map<string, Promise<string|null>>} */
this.pendingEndpoints = new Map();
this.ttl = opts.ttl ?? 3600000;
this.negativeTtl = opts.negativeTtl ?? 60000;
this.fetch = opts.fetch ?? globalThis.fetch;
this.dohUrl = opts.dohUrl;
this.plcUrl = opts.plcUrl;
/**
* Collections a validation failure may reject the write for. See assertValid.
* @type {Set<string>}
*/
this.enforced = new Set(opts.enforce ?? knownRecords);
// Pre-load core com.atproto schemas (not published via DNS)
for (const schema of coreSchemas) {
this.schemas.set(schema.id, schema);
this.cache.set(schema.id, { fetchedAt: Date.now(), ttl: this.ttl });
this.enforced.add(schema.id);
}
// Pre-load the lexicons for the enforced record types, so the common write
// path resolves nothing over the network. They carry the same TTL as a
// fetched schema, so the published document replaces each one as it ages.
for (const schema of publishedSchemas) {
this.schemas.set(schema.id, schema);
this.cache.set(schema.id, { fetchedAt: Date.now(), ttl: this.ttl });
}
// Pre-load user-provided static schemas
for (const schema of opts.schemas ?? []) {
this.schemas.set(schema.id, schema);
this.cache.set(schema.id, { fetchedAt: Date.now(), ttl: this.ttl });
this.enforced.add(schema.id);
}
}
/**
* Validate a record against its schema.
*
* Validation covers the enforced collections — the ones this PDS ships, or
* the `enforce` list, which defaults to the set the reference PDS validates
* strictly. `com.atproto.repo.createRecord` defines the default as "validate
* only for known Lexicons", so any other collection reports 'unknown'
* without resolving anything. That is what the reference PDS answers for a
* lexicon it does not hold, and resolving one costs the network chain to
* change nothing: a collection this server does not enforce cannot fail the
* write, and a failure there reports 'unknown' too.
*
* Pass `strict` — the caller's `validate: true` — to require validation and
* enforce whatever resolves, which is the one path that reaches the network.
*
* @param {string} collection - The collection/NSID (e.g., 'app.bsky.feed.post')
* @param {object} record - The record to validate
* @param {Object} [opts]
* @param {boolean} [opts.strict=false] - Require validation and enforce
* network-resolved schemas
* @returns {Promise<ValidationStatus>}
*/
async assertValid(collection, record, opts = {}) {
const strict = opts.strict === true;
if (!strict && !this.enforced.has(collection)) {
return 'unknown';
}
/** @type {Set<string>} */
const unresolved = new Set();
const hasSchema = await this.ensureSchemaWithRefs(collection, unresolved);
if (!hasSchema) {
if (strict) {
throw new Error(`no published lexicon for ${collection}`);
}
return 'unknown';
}
// Pass ALL loaded schemas to validateRecord for ref resolution
const allSchemas = Array.from(this.schemas.values());
const error = validateRecord(allSchemas, collection, record);
if (!error) {
return 'valid';
}
if (!strict) {
// An error naming a schema we failed to fetch is a resolution failure,
// not a bad record — a DNS or directory hiccup, or a ref whose authority
// never published it.
for (const nsid of unresolved) {
if (error.message.includes(nsid)) {
return 'unknown';
}
}
}
const path = error.path ? `${error.path}: ` : '';
throw new Error(`${path}${error.message}`);
}
/**
* Whether an NSID's lexicon is published — i.e. resolvable via the protocol
* (DNS authority → DID → com.atproto.lexicon.schema record). Bundled schemas
* count as published. ensureSchema caches both answers, so repeated checks
* don't re-hit DNS for an NSID that has no published lexicon.
* @param {string} nsid
* @returns {Promise<boolean>}
*/
async isPublished(nsid) {
return this.ensureSchema(nsid);
}
/**
* The resolved lexicon document for an NSID, or null if it isn't published /
* resolvable. Uses the same resolution and cache as validation, so a schema
* already loaded is returned without a network round-trip.
* @param {string} nsid
* @returns {Promise<Lexicon|null>}
*/
async getLexicon(nsid) {
const ok = await this.ensureSchema(nsid);
return ok ? (this.schemas.get(nsid) ?? null) : null;
}
/**
* Ensure a schema and all its refs are loaded. Refs that can't be resolved
* are collected into `unresolved` rather than failing the call, so a caller
* can tell an incomplete schema set apart from an unpublished collection.
* @param {string} nsid
* @param {Set<string>} [unresolved] - Collects NSIDs that failed to resolve
* @param {Set<string>} [seen] - Guards against ref cycles
* @returns {Promise<boolean>} - true if the schema itself is available
*/
async ensureSchemaWithRefs(nsid, unresolved = new Set(), seen = new Set()) {
if (seen.has(nsid)) {
return true;
}
seen.add(nsid);
const hasMain = await this.ensureSchema(nsid);
const schema = hasMain ? this.schemas.get(nsid) : undefined;
if (!schema) {
unresolved.add(nsid);
return false;
}
// ensureSchema serves fresh schemas from cache, so walking every ref costs
// nothing beyond the schemas that actually need refreshing. Siblings
// resolve together: `seen` is marked before the first await in each call,
// so a shared ref is still fetched one time.
await Promise.all(
this.extractRefs(schema).map((ref) =>
this.ensureSchemaWithRefs(ref, unresolved, seen),
),
);
return true;
}
/**
* Ensure a single schema is loaded (without refs).
* @param {string} nsid
* @returns {Promise<boolean>} - true if schema is available
*/
async ensureSchema(nsid) {
// Check if already loaded and not stale
const cached = this.cache.get(nsid);
const isFresh = cached && Date.now() - cached.fetchedAt < cached.ttl;
if (isFresh) {
return this.schemas.has(nsid);
}
// Concurrent callers for the same NSID share one in-flight resolution, so
// every caller gets the fetched answer rather than racing it.
const pending = this.pendingFetches.get(nsid);
if (pending) {
return pending;
}
// Stale or not cached - try to fetch/refresh
const fetching = (async () => {
try {
const lexicon = await this.fetchLexicon(nsid);
if (lexicon) {
this.schemas.set(nsid, lexicon);
this.cache.set(nsid, { fetchedAt: Date.now(), ttl: this.ttl });
return true;
}
} catch {
// Resolution failed - fall back to existing schema if available
}
// A resolution that produced nothing is cached on the shorter
// negativeTtl. Without that entry every write to an unpublished
// collection repeats the whole chain, and so does every write once a
// preloaded schema ages out and the authority is unreachable.
this.cache.set(nsid, { fetchedAt: Date.now(), ttl: this.negativeTtl });
// If fetch failed but we have an existing (stale) schema, use it
return this.schemas.has(nsid);
})();
this.pendingFetches.set(nsid, fetching);
try {
return await fetching;
} finally {
this.pendingFetches.delete(nsid);
}
}
/**
* Extract all ref NSIDs from a lexicon schema.
* @param {Lexicon} schema
* @returns {string[]} - Array of referenced NSIDs
*/
extractRefs(schema) {
/** @type {Set<string>} */
const refs = new Set();
/**
* Recursively walk object to find refs
* @param {unknown} obj
*/
const walk = (obj) => {
if (!obj || typeof obj !== 'object') return;
if (Array.isArray(obj)) {
for (const item of obj) {
walk(item);
}
return;
}
const record = /** @type {Record<string, unknown>} */ (obj);
// Check for ref type
if (record.type === 'ref' && typeof record.ref === 'string') {
const ref = record.ref;
// Extract NSID from ref (could be "com.atproto.repo.strongRef" or "lex:com.atproto.repo.strongRef#main")
const nsid = this.refToNsid(ref);
if (nsid) {
refs.add(nsid);
}
}
// Check for union refs
if (record.type === 'union' && Array.isArray(record.refs)) {
for (const ref of record.refs) {
if (typeof ref === 'string') {
const nsid = this.refToNsid(ref);
if (nsid) {
refs.add(nsid);
}
}
}
}
// Recurse into nested objects
for (const value of Object.values(record)) {
walk(value);
}
};
walk(schema);
return Array.from(refs);
}
/**
* Convert a ref string to an NSID.
* @param {string} ref - e.g., "com.atproto.repo.strongRef" or "lex:app.bsky.feed.post#main"
* @returns {string|null} - The NSID or null if invalid
*/
refToNsid(ref) {
// Remove lex: prefix if present
let nsid = ref.startsWith('lex:') ? ref.slice(4) : ref;
// Remove fragment (e.g., #main, #strongRef)
const hashIndex = nsid.indexOf('#');
if (hashIndex !== -1) {
nsid = nsid.slice(0, hashIndex);
}
// Validate it looks like an NSID (has at least 3 segments)
const parts = nsid.split('.');
if (parts.length >= 3) {
return nsid;
}
return null;
}
/**
* Resolve a key one time. Concurrent callers share the in-flight answer, and
* the result is held for a TTL — the short negativeTtl when nothing resolved.
* @template T
* @param {Map<string, {value: T|null, fetchedAt: number, ttl: number}>} cache
* @param {Map<string, Promise<T|null>>} pending
* @param {string} key
* @param {() => Promise<T|null>} resolve
* @returns {Promise<T|null>}
*/
async resolveOnce(cache, pending, key, resolve) {
const cached = cache.get(key);
if (cached && Date.now() - cached.fetchedAt < cached.ttl) {
return cached.value;
}
const inFlight = pending.get(key);
if (inFlight) return inFlight;
const fetching = (async () => {
/** @type {T|null} */
let value = null;
try {
value = await resolve();
} catch {
// A lookup that threw caches as an absent one, on the short TTL.
}
cache.set(key, {
value,
fetchedAt: Date.now(),
ttl: value ? this.ttl : this.negativeTtl,
});
return value;
})();
pending.set(key, fetching);
try {
return await fetching;
} finally {
pending.delete(key);
}
}
/**
* The DID an authority domain publishes in its `_lexicon` TXT record.
* @param {string} domain - e.g. "feed.bsky.app"
* @returns {Promise<string|null>}
*/
async resolveAuthority(domain) {
return this.resolveOnce(
this.authorities,
this.pendingAuthorities,
domain,
async () => {
const txtRecords = await lookupTxt(domainToLexiconDns(domain), {
dohUrl: this.dohUrl,
fetch: this.fetch,
});
for (const txt of txtRecords) {
const did = parseDidFromTxt(txt);
if (did) return did;
}
return null;
},
);
}
/**
* The PDS endpoint in an authority's DID document.
* @param {string} did
* @returns {Promise<string|null>}
*/
async resolveEndpoint(did) {
return this.resolveOnce(this.endpoints, this.pendingEndpoints, did, () =>
resolveDid(did, { plcUrl: this.plcUrl, fetch: this.fetch }),
);
}
/**
* Fetch a lexicon via the AT Protocol resolution chain.
* @param {string} nsid
* @returns {Promise<Lexicon|null>}
*/
async fetchLexicon(nsid) {
// 1-3. Look up the authority's DID, trying the spec-form domain before the
// two-segment fallback.
let did = null;
for (const domain of nsidToDomains(nsid)) {
did = await this.resolveAuthority(domain);
if (did) break;
}
if (!did) return null;
// 4. Resolve DID to PDS endpoint
const pdsEndpoint = await this.resolveEndpoint(did);
if (!pdsEndpoint) return null;
// 5. Fetch lexicon from PDS
const url = `${pdsEndpoint}/xrpc/com.atproto.repo.getRecord?repo=${encodeURIComponent(did)}&collection=com.atproto.lexicon.schema&rkey=${encodeURIComponent(nsid)}`;
const response = await this.fetch(url);
if (!response.ok) return null;
const data = await response.json();
return data.value ?? null;
}
}