chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
/**
* The index store: every watched record, the links it names, and the cursors
* that keep the stream resumable, in one SQLite database.
*
* The `appview_record` table holds each record whole as JSON. The
* `appview_link` table is the reverse index: one row per at-uri a record
* names, keyed by the JSON path it names it at, which is the whole question
* a discovery port asks. Records and their links commit together, so a read
* never sees half a write.
*
* The driver is the SQLite interface `@pdsjs/storage-sqlite/driver` names,
* which better-sqlite3 and node:sqlite both implement natively.
*/
/** @typedef {import('@pdsjs/storage-sqlite/driver').SQLiteDatabase} SQLiteDatabase */
/** @typedef {import('@pdsjs/storage-sqlite/driver').SQLiteStatement} SQLiteStatement */
/**
* @param {SQLiteDatabase} db
*/
export function createAppViewStore(db) {
db.exec(`
CREATE TABLE IF NOT EXISTS appview_record (
uri TEXT PRIMARY KEY,
did TEXT NOT NULL,
collection TEXT NOT NULL,
rkey TEXT NOT NULL,
cid TEXT,
value TEXT NOT NULL,
indexed_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_appview_record_collection
ON appview_record (collection, indexed_at);
CREATE TABLE IF NOT EXISTS appview_link (
uri TEXT NOT NULL,
path TEXT NOT NULL,
subject TEXT NOT NULL,
PRIMARY KEY (uri, path, subject)
);
CREATE INDEX IF NOT EXISTS idx_appview_link_subject
ON appview_link (subject);
CREATE TABLE IF NOT EXISTS appview_identity (
did TEXT PRIMARY KEY,
handle TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS appview_cursor (
source TEXT PRIMARY KEY,
value TEXT NOT NULL
);
`);
const now = () => new Date().toISOString();
/**
* @param {string} source
* @returns {string|undefined}
*/
const readCursor = (source) => {
const row = /** @type {{value: string}|undefined} */ (
db
.prepare('SELECT value FROM appview_cursor WHERE source = ?')
.get(source)
);
return row ? row.value : undefined;
};
/**
* @param {string} source
* @param {string} value
*/
const writeCursor = (source, value) => {
db.prepare(
'INSERT OR REPLACE INTO appview_cursor (source, value) VALUES (?, ?)',
).run(source, value);
};
return {
/**
* @param {RecordWrite} write
*/
putRecord(write) {
const links = linksOf(write.value);
transaction(db, () => {
db.prepare(
'INSERT OR REPLACE INTO appview_record (uri, did, collection, rkey, cid, value, indexed_at) VALUES (?, ?, ?, ?, ?, ?, ?)',
).run(
write.uri,
write.did,
write.collection,
write.rkey,
write.cid,
JSON.stringify(write.value),
now(),
);
db.prepare('DELETE FROM appview_link WHERE uri = ?').run(write.uri);
for (const link of links) {
db.prepare(
'INSERT OR REPLACE INTO appview_link (uri, path, subject) VALUES (?, ?, ?)',
).run(write.uri, link.path, link.subject);
}
});
},
/**
* @param {string} uri
*/
deleteRecord(uri) {
transaction(db, () => {
db.prepare('DELETE FROM appview_record WHERE uri = ?').run(uri);
db.prepare('DELETE FROM appview_link WHERE uri = ?').run(uri);
});
},
/**
* @param {string} uri
* @returns {IndexedRecord|null}
*/
getRecord(uri) {
const row = /** @type {any|undefined} */ (
db
.prepare(
'SELECT uri, did, collection, rkey, cid, value, indexed_at, rowid FROM appview_record WHERE uri = ?',
)
.get(uri)
);
return row ? toRecord(row) : null;
},
/**
* Newest inserted first. `before` pages: only records inserted before
* the rowid named. The order is the store's own insertion order, so a
* page never revisits a row and never skips one.
* @param {string} collection
* @param {number} [limit]
* @param {number} [before]
* @returns {IndexedRecord[]}
*/
listRecords(collection, limit = 100, before) {
const rows = /** @type {any[]} */ (
before
? db
.prepare(
'SELECT uri, did, collection, rkey, cid, value, indexed_at, rowid FROM appview_record WHERE collection = ? AND rowid < ? ORDER BY rowid DESC LIMIT ?',
)
.all(collection, before, limit)
: db
.prepare(
'SELECT uri, did, collection, rkey, cid, value, indexed_at, rowid FROM appview_record WHERE collection = ? ORDER BY rowid DESC LIMIT ?',
)
.all(collection, limit)
);
return rows.map(toRecord);
},
/**
* Every record of one collection naming a subject at a path. This is
* the discovery port's whole question; the answer is each record's
* address, read from its author when it matters.
* @param {string} subjectUri
* @param {string} collection
* @param {string} [path]
* @param {number} [limit]
* @returns {IndexedRecord[]}
*/
about(subjectUri, collection, path = 'subject.uri', limit = 1000) {
const rows = /** @type {any[]} */ (
db
.prepare(
`SELECT r.uri, r.did, r.collection, r.rkey, r.cid, r.value, r.indexed_at
FROM appview_link l JOIN appview_record r ON r.uri = l.uri
WHERE l.subject = ? AND l.path = ? AND r.collection = ?
ORDER BY r.rowid DESC LIMIT ?`,
)
.all(subjectUri, path, collection, limit)
);
return rows.map(toRecord);
},
/**
* @param {string} did
* @param {string} handle
*/
setIdentity(did, handle) {
db.prepare(
'INSERT OR REPLACE INTO appview_identity (did, handle, updated_at) VALUES (?, ?, ?)',
).run(did, handle, now());
},
/**
* @param {string} did
* @returns {string|null}
*/
handleOf(did) {
const row = /** @type {{handle: string}|undefined} */ (
db.prepare('SELECT handle FROM appview_identity WHERE did = ?').get(did)
);
return row ? row.handle : null;
},
/**
* An account deletion removes the account's records and their links.
* @param {string} did
*/
clearAccount(did) {
transaction(db, () => {
const uris = /** @type {any[]} */ (
db.prepare('SELECT uri FROM appview_record WHERE did = ?').all(did)
).map((row) => String(row.uri));
for (const uri of uris) {
db.prepare('DELETE FROM appview_link WHERE uri = ?').run(uri);
}
db.prepare('DELETE FROM appview_record WHERE did = ?').run(did);
});
},
/**
* @param {string} source
* @returns {number|undefined}
*/
cursor(source) {
const value = readCursor(source);
if (value === undefined) return undefined;
const seq = Number(value);
return Number.isFinite(seq) ? seq : undefined;
},
/**
* @param {string} source
* @param {number} seq
*/
setCursor(source, seq) {
writeCursor(source, String(seq));
},
/**
* @param {string} source
* @returns {string|undefined}
*/
textCursor(source) {
return readCursor(source);
},
/**
* @param {string} source
* @param {string} value
*/
setTextCursor(source, value) {
writeCursor(source, value);
},
};
}
/** @typedef {ReturnType<typeof createAppViewStore>} AppViewStore */
/**
* A record as the store holds it. `rowid` is the store's insertion order,
* which is what `listRecords` pages by.
* @typedef {Object} IndexedRecord
* @property {string} uri
* @property {string} did
* @property {string} collection
* @property {string} rkey
* @property {string|null} cid
* @property {Record<string, unknown>} value
* @property {string} indexedAt
* @property {number} rowid
*/
/**
* A record write from any source: jetstream, the native firehose, or a
* backfill pass. The cid is null for a row the archive replayed, because
* the on-disk format does not store it.
* @typedef {Object} RecordWrite
* @property {string} did
* @property {string} collection
* @property {string} rkey
* @property {string|null} cid
* @property {string} uri
* @property {number} seq
* @property {Record<string, *>} value
*/
/**
* A withdrawal: the record is gone from its author's repo.
* @typedef {Object} RecordDelete
* @property {string} did
* @property {string} collection
* @property {string} rkey
* @property {string} uri
* @property {number} seq
* @property {true} deleted
*/
/**
* The at-uri references one record names, as dotted JSON paths with array
* positions folded away: `{subject: {uri: ...}}` indexes under
* `subject.uri`, and a list of ref strings indexes under the list field's
* name, one row per entry.
* @param {unknown} value
* @param {string} [path]
* @returns {Array<{path: string, subject: string}>}
*/
export function linksOf(value, path = '') {
if (typeof value === 'string') {
return isRecordUri(value) ? [{ path, subject: value }] : [];
}
if (Array.isArray(value)) {
/** @type {Array<{path: string, subject: string}>} */
const found = [];
for (const entry of value) found.push(...linksOf(entry, path));
return found;
}
if (!value || typeof value !== 'object') return [];
/** @type {Array<{path: string, subject: string}>} */
const found = [];
for (const [key, entry] of Object.entries(value)) {
found.push(...linksOf(entry, path ? `${path}.${key}` : key));
}
return found;
}
/**
* An at-uri that names a record: a DID, a collection and a record key.
* A bare `at://did` names an account and indexes nothing.
* @param {string} value
* @returns {boolean}
*/
function isRecordUri(value) {
if (!value.startsWith('at://')) return false;
const segments = value.slice('at://'.length).split('/');
return segments.length >= 3 && segments.every((part) => part.length > 0);
}
/**
* @param {any} row
* @returns {IndexedRecord}
*/
function toRecord(row) {
return {
uri: String(row.uri),
did: String(row.did),
collection: String(row.collection),
rkey: String(row.rkey),
cid: row.cid == null ? null : String(row.cid),
value: JSON.parse(String(row.value)),
indexedAt: String(row.indexed_at),
rowid: Number(row.rowid),
};
}
/**
* @param {SQLiteDatabase} db
* @param {() => void} work
*/
function transaction(db, work) {
db.exec('BEGIN');
try {
work();
db.exec('COMMIT');
} catch (err) {
db.exec('ROLLBACK');
throw err;
}
}