chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/spaces/memory-storage - in-memory SpaceStoragePort, and the blob side
// of the route context.
//
// Backs unit tests and gives handler tests a store with no database. Keyed
// exactly like the SQL adapters so the shared conformance suite exercises the
// same behaviour everywhere.
/**
* The blob store and blob link table createSpaceRoutes wants, in memory. A real
* deployment passes the account's own store and link table; here they are two
* Maps that answer the same way.
*
* @returns {{blobs: import('@pdsjs/core/ports').BlobPort, linkBlob: (blobCid: string, recordUri: string, recordTime: number|null) => Promise<void>, unlinkBlobs: (recordUri: string) => Promise<void>, links: Map<string, Set<string>>, put: (cid: string, data: Uint8Array, mimeType?: string) => void}}
*/
export function createMemoryBlobContext() {
/** @type {Map<string, {data: Uint8Array, mimeType: string}>} */
const stored = new Map();
/** @type {Map<string, Set<string>>} record uri -> blob cids */
const links = new Map();
return {
blobs: {
async get(_did, cid) {
const blob = stored.get(cid);
return blob ? { ...blob, data: new Uint8Array(blob.data) } : null;
},
async put(_did, cid, data, mimeType) {
stored.set(cid, { data: new Uint8Array(data), mimeType });
},
async delete(_did, cid) {
stored.delete(cid);
},
},
async linkBlob(blobCid, recordUri) {
let forRecord = links.get(recordUri);
if (!forRecord) {
forRecord = new Set();
links.set(recordUri, forRecord);
}
forRecord.add(blobCid);
},
async unlinkBlobs(recordUri) {
links.delete(recordUri);
},
links,
put(cid, data, mimeType = 'application/octet-stream') {
stored.set(cid, { data, mimeType });
},
};
}
/**
* A copy that shares no object with the stored row.
* @param {import('@pdsjs/core/ports').SpaceRow} row
* @returns {import('@pdsjs/core/ports').SpaceRow}
*/
function copySpace(row) {
return { ...row, policy: row.policy ? structuredClone(row.policy) : null };
}
/**
* Keyset pagination over a sorted key list.
* @template T
* @param {string[]} sortedKeys
* @param {string|null} cursor - exclusive lower bound
* @param {number} limit
* @param {(key: string) => T} map
* @returns {{items: T[], cursor: string|null}}
*/
function page(sortedKeys, cursor, limit, map) {
const start = cursor ? sortedKeys.findIndex((k) => k > cursor) : 0;
if (start === -1) return { items: [], cursor: null };
const slice = sortedKeys.slice(start, start + limit);
return {
items: slice.map(map),
// A full page always yields a cursor, even if it happened to be the last
// one. The next call then returns empty with a null cursor. This matches
// `WHERE key > ? LIMIT n` in the SQL adapters — do not "optimise" it to
// return null on the final full page, or the two diverge.
cursor: slice.length === limit ? slice[slice.length - 1] : null,
};
}
/**
* @returns {import('@pdsjs/core/ports').SpaceStoragePort}
*/
export function createMemorySpaceStorage() {
/** @type {Map<string, import('@pdsjs/core/ports').SpaceRow>} */
const spaces = new Map();
/** @type {Map<string, Map<string, import('@pdsjs/core/ports').SpaceMemberAccess>>} space uri -> did -> access */
const members = new Map();
/** @type {Map<string, Map<string, import('@pdsjs/core/ports').SpaceRecordRow>>} */
const records = new Map();
/** @type {Map<string, {setHash: Uint8Array|null, rev: string|null}>} */
const repos = new Map();
/** @type {Map<string, import('@pdsjs/core/ports').SpaceOpRow[]>} */
const oplog = new Map();
/** @type {Map<string, Map<string, import('@pdsjs/core/ports').SpaceWriterRow>>} */
const writers = new Map();
/** @type {Map<string, Map<string, import('@pdsjs/core/ports').SpaceRecipientRow>>} */
const recipients = new Map();
/** @type {Map<string, string>} `${space} ${jti}` -> expiresAt */
const revocations = new Map();
/** @type {Map<string, import('@pdsjs/core/ports').SpaceNotifyRow>} */
const outbox = new Map();
/** @type {Map<string, import('@pdsjs/core/ports').SpaceDeletedNotifyRow>} */
const deletedOutbox = new Map();
return {
async getSpace(uri) {
const row = spaces.get(uri);
return row ? copySpace(row) : null;
},
async putSpace(space) {
spaces.set(space.uri, copySpace(space));
},
async deleteSpace(uri, deletedAt) {
const row = spaces.get(uri);
if (row) spaces.set(uri, { ...row, deletedAt });
},
async listSpaces({
type = null,
did = null,
cursor = null,
limit = 50,
excludeDeleted = false,
}) {
let keys = [...spaces.keys()].sort();
if (type || did || excludeDeleted) {
keys = keys.filter((uri) => {
// Filter on the stored components, not by re-parsing the URI: the
// domain supplies them, and every adapter must agree on the result.
const row = /** @type {import('@pdsjs/core/ports').SpaceRow} */ (
spaces.get(uri)
);
if (type && row.spaceType !== type) return false;
if (did && row.spaceDid !== did) return false;
if (excludeDeleted && row.deletedAt) return false;
return true;
});
}
const { items, cursor: next } = page(keys, cursor, limit, (uri) => {
// uri comes from spaces.keys(), so this lookup is always present.
const row = /** @type {import('@pdsjs/core/ports').SpaceRow} */ (
spaces.get(uri)
);
return copySpace(row);
});
return { spaces: items, cursor: next };
},
async putMember(space, did, access) {
let forSpace = members.get(space);
if (!forSpace) {
forSpace = new Map();
members.set(space, forSpace);
}
forSpace.set(did, { read: access.read, write: access.write });
},
async removeMember(space, did) {
members.get(space)?.delete(did);
},
async listMembers(space, cursor, limit) {
const forSpace = members.get(space);
const keys = [...(forSpace?.keys() ?? [])].sort();
const { items, cursor: next } = page(keys, cursor, limit, (did) => ({
did,
.../** @type {import('@pdsjs/core/ports').SpaceMemberAccess} */ (
forSpace?.get(did)
),
}));
return { members: items, cursor: next };
},
async getMember(space, did) {
const access = members.get(space)?.get(did);
return access ? { ...access } : null;
},
async removeAllMembers(space) {
members.delete(space);
},
async getSpaceRecord(space, collection, rkey) {
const row = records.get(space)?.get(`${collection}/${rkey}`);
return row ? { ...row, value: new Uint8Array(row.value) } : null;
},
async getSpaceRecordCid(space, collection, rkey) {
return records.get(space)?.get(`${collection}/${rkey}`)?.cid ?? null;
},
async listSpaceRecords(
space,
{ collection, cursor, limit, reverse, excludeValues },
) {
const inSpace = records.get(space) ?? new Map();
let keys = [...inSpace.values()]
.filter((r) => !collection || r.collection === collection)
.map((r) => `${r.collection}/${r.rkey}`)
.sort();
// Descending unless the caller asks otherwise, matching the reference.
if (!reverse) keys.reverse();
if (cursor) {
keys = keys.filter((k) => (reverse ? k > cursor : k < cursor));
}
const items = keys.slice(0, limit).map((key) => {
const row = inSpace.get(key);
const { value, ...rest } = row;
return excludeValues
? { ...rest }
: { ...rest, value: new Uint8Array(value) };
});
return {
records: items,
cursor: items.length === limit ? keys[limit - 1] : null,
};
},
async listAllSpaceRecords(space) {
const inSpace = records.get(space) ?? new Map();
return [...inSpace.keys()].sort().map((key) => {
const row = inSpace.get(key);
return { collection: row.collection, rkey: row.rkey, cid: row.cid };
});
},
async commitSpaceWrite(space, { writes, setHash, rev, indexedAt }) {
// Validate everything before mutating: this method is atomic.
for (const w of writes) {
if (w.action !== 'delete' && !(w.value instanceof Uint8Array)) {
throw new Error(
`space write value must be Uint8Array: ${w.collection}/${w.rkey}`,
);
}
}
let inSpace = records.get(space);
if (!inSpace) {
inSpace = new Map();
records.set(space, inSpace);
}
const ops = oplog.get(space) ?? [];
writes.forEach((w, idx) => {
const key = `${w.collection}/${w.rkey}`;
if (w.action === 'delete') {
inSpace.delete(key);
} else {
inSpace.set(key, {
collection: w.collection,
rkey: w.rkey,
cid: /** @type {string} */ (w.cid),
value: new Uint8Array(/** @type {Uint8Array} */ (w.value)),
repoRev: rev,
indexedAt,
});
}
ops.push({
rev,
idx,
action: w.action,
collection: w.collection,
rkey: w.rkey,
cid: w.cid,
prev: w.prev,
});
});
oplog.set(space, ops);
repos.set(space, { setHash: new Uint8Array(setHash), rev });
},
async getSpaceRepo(space) {
const repo = repos.get(space);
if (!repo) return null;
return {
setHash: repo.setHash ? new Uint8Array(repo.setHash) : null,
rev: repo.rev,
};
},
async listSpaceOps(space, since, limit) {
return (oplog.get(space) ?? [])
.filter((op) => (since === null ? true : op.rev > since))
.sort((a, b) =>
a.rev === b.rev ? a.idx - b.idx : a.rev < b.rev ? -1 : 1,
)
.slice(0, limit)
.map((op) => ({ ...op }));
},
async getSpaceWriter(space, did) {
const row = writers.get(space)?.get(did);
return row ? { ...row, data: new Uint8Array(row.data) } : null;
},
async getSpaceRevHead(space) {
let head = null;
for (const row of (writers.get(space) ?? new Map()).values()) {
if (head === null || row.spaceRev > head) head = row.spaceRev;
}
return head;
},
async removeSpaceWriter(space, did) {
writers.get(space)?.delete(did);
},
async putSpaceWriter(space, did, repoRev, data, spaceRev) {
let inSpace = writers.get(space);
if (!inSpace) {
inSpace = new Map();
writers.set(space, inSpace);
}
inSpace.set(did, { did, repoRev, data: new Uint8Array(data), spaceRev });
},
async listSpaceWriters(space, cursor, limit) {
const rows = [...(writers.get(space) ?? new Map()).values()]
.filter((row) => row.spaceRev > (cursor ?? ''))
.sort((a, b) => (a.spaceRev < b.spaceRev ? -1 : 1))
.slice(0, limit)
.map((row) => ({ ...row, data: new Uint8Array(row.data) }));
return {
writers: rows,
cursor: rows.length ? rows[rows.length - 1].spaceRev : null,
};
},
async putCredentialRevocations(space, jtis, expiresAt) {
const now = new Date().toISOString();
for (const [key, at] of revocations) {
if (at <= now) revocations.delete(key);
}
for (const jti of jtis) {
const key = `${space} ${jti}`;
const held = revocations.get(key);
revocations.set(key, held && held > expiresAt ? held : expiresAt);
}
},
async isCredentialRevoked(space, jti) {
const at = revocations.get(`${space} ${jti}`);
return at !== undefined && at > new Date().toISOString();
},
async enqueueSpaceNotify(row) {
outbox.set(row.space, {
...row,
data: new Uint8Array(row.data),
attempts: 0,
});
},
async listDueSpaceNotify(now, limit) {
return [...outbox.values()]
.filter((row) => row.nextAttemptAt <= now)
.sort((a, b) => (a.nextAttemptAt < b.nextAttemptAt ? -1 : 1))
.slice(0, limit)
.map((row) => ({ ...row, data: new Uint8Array(row.data) }));
},
async rescheduleSpaceNotify(space, repoRev, attempts, nextAttemptAt) {
const row = outbox.get(space);
if (row?.repoRev === repoRev) {
outbox.set(space, { ...row, attempts, nextAttemptAt });
}
},
async deleteSpaceNotify(space, repoRev) {
if (outbox.get(space)?.repoRev === repoRev) outbox.delete(space);
},
async enqueueSpaceDeletedNotify(row) {
deletedOutbox.set(`${row.space} ${row.service}`, { ...row });
},
async listDueSpaceDeletedNotify(now, limit) {
return [...deletedOutbox.values()]
.filter((row) => row.nextAttemptAt <= now)
.sort((a, b) => (a.nextAttemptAt < b.nextAttemptAt ? -1 : 1))
.slice(0, limit)
.map((row) => ({ ...row }));
},
async rescheduleSpaceDeletedNotify(
space,
service,
attempts,
nextAttemptAt,
) {
const key = `${space} ${service}`;
const row = deletedOutbox.get(key);
if (row) deletedOutbox.set(key, { ...row, attempts, nextAttemptAt });
},
async deleteSpaceDeletedNotify(space, service) {
deletedOutbox.delete(`${space} ${service}`);
},
async nextSpaceNotifyAt() {
let next = null;
for (const row of [...outbox.values(), ...deletedOutbox.values()]) {
if (next === null || row.nextAttemptAt < next) next = row.nextAttemptAt;
}
return next;
},
async putCredentialRecipient(
space,
serviceDid,
serviceEndpoint,
expiresAt,
) {
let inSpace = recipients.get(space);
if (!inSpace) {
inSpace = new Map();
recipients.set(space, inSpace);
}
inSpace.set(serviceDid, { serviceDid, serviceEndpoint, expiresAt });
},
async deleteCredentialRecipient(space, serviceDid) {
recipients.get(space)?.delete(serviceDid);
},
async listCredentialRecipients(space) {
const inSpace = recipients.get(space) ?? new Map();
const now = new Date().toISOString();
return [...inSpace.keys()]
.sort()
.map((k) => ({ ...inSpace.get(k) }))
.filter((r) => r.expiresAt > now);
},
};
}