chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/cloudflare/space - Durable Object SQLite SpaceStoragePort (proposal 0016).
//
// Shares its schema with the storage-sqlite adapter via @pdsjs/core/space-schema;
// only the driver differs. Behaviour is pinned by the shared conformance suite
// in test/helpers/space-storage-conformance.js, which both adapters run.
import {
SPACE_MIGRATIONS,
SPACE_SCHEMA_SQL,
spaceRecordCursor,
spaceRecordsQuery,
} from '@pdsjs/core/space-schema';
/** @typedef {{uri: string, space_did: string, space_type: string, is_owner: number, policy_json: string|null, created_at: string, deleted_at: string|null}} SpaceDbRow */
/** @typedef {{did: string, can_read: number, can_write: number}} SpaceMemberDbRow */
/** @typedef {{collection: string, rkey: string, cid: string, value: ArrayBuffer|Uint8Array, repo_rev: string, indexed_at: string}} SpaceRecordDbRow */
/** @typedef {Omit<SpaceRecordDbRow, 'value'> & {value?: ArrayBuffer|Uint8Array}} SpaceRecordListDbRow */
/** @typedef {{rev: string, idx: number, action: string, collection: string, rkey: string, cid: string|null, prev: string|null}} SpaceOpDbRow */
/** @typedef {{did: string, rev: string, data: ArrayBuffer|Uint8Array, space_rev: string}} SpaceWriterDbRow */
/** @typedef {{space: string, repo_rev: string, data: ArrayBuffer|Uint8Array, attempts: number, next_attempt_at: string, deadline_at: string}} SpaceNotifyDbRow */
/** @typedef {{space: string, service: string, service_endpoint: string, attempts: number, next_attempt_at: string, deadline_at: string}} SpaceDeletedNotifyDbRow */
/** @typedef {{service_did: string, service_endpoint: string, expires_at: string}} SpaceRecipientDbRow */
/**
* Durable Object SQLite returns ArrayBuffer for BLOB columns.
* @param {ArrayBuffer|Uint8Array|null|undefined} value
* @returns {Uint8Array|null}
*/
function normalizeBlob(value) {
if (!value) return null;
if (value instanceof Uint8Array) return new Uint8Array(value);
return new Uint8Array(value);
}
/**
* @param {SpaceWriterDbRow} row
* @returns {import('@pdsjs/core/ports').SpaceWriterRow}
*/
function toWriterRow(row) {
return {
did: row.did,
repoRev: row.rev,
data: /** @type {Uint8Array} */ (normalizeBlob(row.data)),
spaceRev: row.space_rev,
};
}
/**
* @param {SpaceNotifyDbRow} row
* @returns {import('@pdsjs/core/ports').SpaceNotifyRow}
*/
function toNotifyRow(row) {
return {
space: row.space,
repoRev: row.repo_rev,
data: /** @type {Uint8Array} */ (normalizeBlob(row.data)),
attempts: row.attempts,
nextAttemptAt: row.next_attempt_at,
deadlineAt: row.deadline_at,
};
}
/**
* @param {SpaceDeletedNotifyDbRow} row
* @returns {import('@pdsjs/core/ports').SpaceDeletedNotifyRow}
*/
function toDeletedNotifyRow(row) {
return {
space: row.space,
service: row.service,
serviceEndpoint: row.service_endpoint,
attempts: row.attempts,
nextAttemptAt: row.next_attempt_at,
deadlineAt: row.deadline_at,
};
}
/**
* @param {SpaceDbRow} row
* @returns {import('@pdsjs/core/ports').SpaceRow}
*/
function toSpaceRow(row) {
return {
uri: row.uri,
spaceDid: row.space_did,
spaceType: row.space_type,
isOwner: row.is_owner === 1,
policy: row.policy_json ? JSON.parse(row.policy_json) : null,
createdAt: row.created_at,
deletedAt: row.deleted_at,
};
}
/**
* @param {SpaceMemberDbRow} row
* @returns {import('@pdsjs/core/ports').SpaceMemberRow}
*/
function toMemberRow(row) {
return {
did: row.did,
read: row.can_read === 1,
write: row.can_write === 1,
};
}
/**
* @param {SpaceRecordDbRow} row
* @returns {import('@pdsjs/core/ports').SpaceRecordRow}
*/
function toRecordRow(row) {
return {
collection: row.collection,
rkey: row.rkey,
cid: row.cid,
value: /** @type {Uint8Array} */ (normalizeBlob(row.value)),
repoRev: row.repo_rev,
indexedAt: row.indexed_at,
};
}
/**
* The value column is absent from a metadata-only listing.
* @param {SpaceRecordListDbRow} row
* @returns {import('@pdsjs/core/ports').SpaceRecordListRow}
*/
function toRecordListRow(row) {
return {
collection: row.collection,
rkey: row.rkey,
cid: row.cid,
...(row.value === undefined
? {}
: { value: /** @type {Uint8Array} */ (normalizeBlob(row.value)) }),
repoRev: row.repo_rev,
indexedAt: row.indexed_at,
};
}
/**
* @param {any} sql - D1SqlInterface, from state.storage.sql
* @param {{transactionSync: (fn: () => any) => any}} storage - state.storage
* @returns {import('@pdsjs/core/ports').SpaceStoragePort}
*/
export function createSpaceStorage(sql, storage) {
sql.exec(SPACE_SCHEMA_SQL);
for (const statement of SPACE_MIGRATIONS) {
try {
sql.exec(statement);
} catch {
// Already applied.
}
}
/**
* Single-row read. Do NOT use cursor.one() — it throws on an empty result,
* which is why the actor-store adapter in index.js carries its own safeOne.
* @param {string} query
* @param {...any} params
* @returns {any|null}
*/
function one(query, ...params) {
const rows = sql.exec(query, ...params).toArray();
return rows.length > 0 ? rows[0] : null;
}
/**
* @param {string} query
* @param {...any} params
* @returns {any[]}
*/
function all(query, ...params) {
return sql.exec(query, ...params).toArray();
}
return {
async getSpace(uri) {
const row = one('SELECT * FROM space WHERE uri = ?', uri);
return row ? toSpaceRow(row) : null;
},
async putSpace(space) {
sql.exec(
`INSERT OR REPLACE INTO space
(uri, space_did, space_type, is_owner, policy_json, created_at, deleted_at)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
space.uri,
space.spaceDid,
space.spaceType,
space.isOwner ? 1 : 0,
space.policy ? JSON.stringify(space.policy) : null,
space.createdAt,
space.deletedAt,
);
},
async deleteSpace(uri, deletedAt) {
sql.exec('UPDATE space SET deleted_at = ? WHERE uri = ?', deletedAt, uri);
},
async listSpaces({
type = null,
did = null,
cursor = null,
limit = 50,
excludeDeleted = false,
}) {
// Filtering is a plain WHERE clause because the domain stores the URI's
// authority and type as columns; the adapter never parses space URIs.
// That also means LIMIT applies to matches, so no scan-and-refill loop.
const where = ['uri > ?'];
const params = /** @type {any[]} */ ([cursor ?? '']);
if (did) {
where.push('space_did = ?');
params.push(did);
}
if (type) {
where.push('space_type = ?');
params.push(type);
}
if (excludeDeleted) where.push('deleted_at IS NULL');
params.push(limit);
const rows = /** @type {SpaceDbRow[]} */ (
all(
`SELECT * FROM space WHERE ${where.join(' AND ')} ORDER BY uri LIMIT ?`,
...params,
)
);
const spaces = rows.map(toSpaceRow);
return {
spaces,
// A full page always yields a cursor, matching every other paginated
// method here and the memory adapter's `page` helper.
cursor: spaces.length === limit ? spaces[spaces.length - 1].uri : null,
};
},
async putMember(space, did, access) {
sql.exec(
`INSERT INTO space_member (space, did, can_read, can_write) VALUES (?, ?, ?, ?)
ON CONFLICT (space, did) DO UPDATE SET can_read = excluded.can_read, can_write = excluded.can_write`,
space,
did,
access.read ? 1 : 0,
access.write ? 1 : 0,
);
},
async removeMember(space, did) {
sql.exec(
'DELETE FROM space_member WHERE space = ? AND did = ?',
space,
did,
);
},
async listMembers(space, cursor, limit) {
const rows = /** @type {SpaceMemberDbRow[]} */ (
all(
'SELECT did, can_read, can_write FROM space_member WHERE space = ? AND did > ? ORDER BY did LIMIT ?',
space,
cursor ?? '',
limit,
)
);
const members = rows.map(toMemberRow);
return {
members,
cursor:
members.length === limit ? members[members.length - 1].did : null,
};
},
async getMember(space, did) {
const row = /** @type {SpaceMemberDbRow|null} */ (
one(
'SELECT can_read, can_write FROM space_member WHERE space = ? AND did = ?',
space,
did,
)
);
return row
? { read: row.can_read === 1, write: row.can_write === 1 }
: null;
},
async removeAllMembers(space) {
sql.exec('DELETE FROM space_member WHERE space = ?', space);
},
async getSpaceRecord(space, collection, rkey) {
const row = one(
'SELECT collection, rkey, cid, value, repo_rev, indexed_at FROM space_record WHERE space = ? AND collection = ? AND rkey = ?',
space,
collection,
rkey,
);
return row ? toRecordRow(row) : null;
},
async getSpaceRecordCid(space, collection, rkey) {
const row = one(
'SELECT cid FROM space_record WHERE space = ? AND collection = ? AND rkey = ?',
space,
collection,
rkey,
);
return row ? row.cid : null;
},
async listSpaceRecords(space, opts) {
const { sql: query, params } = spaceRecordsQuery(space, opts);
const rows = /** @type {SpaceRecordListDbRow[]} */ (
all(query, ...params)
);
const records = rows.map(toRecordListRow);
return {
records,
cursor:
records.length === opts.limit
? spaceRecordCursor(records[records.length - 1])
: null,
};
},
async listAllSpaceRecords(space) {
return /** @type {Array<{collection: string, rkey: string, cid: string}>} */ (
all(
"SELECT collection, rkey, cid FROM space_record WHERE space = ? ORDER BY (collection || '/' || rkey)",
space,
)
);
},
async commitSpaceWrite(space, { writes, setHash, rev, indexedAt }) {
// Validate everything before opening the transaction, so a bad batch
// cannot leave a partially applied commit behind.
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}`,
);
}
}
// Durable Object SQL rejects explicit BEGIN/COMMIT statements;
// transactionSync is the supported way to make several statements atomic.
storage.transactionSync(() => {
writes.forEach((w, idx) => {
if (w.action === 'delete') {
sql.exec(
'DELETE FROM space_record WHERE space = ? AND collection = ? AND rkey = ?',
space,
w.collection,
w.rkey,
);
} else {
sql.exec(
`INSERT OR REPLACE INTO space_record
(space, collection, rkey, cid, value, repo_rev, indexed_at)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
space,
w.collection,
w.rkey,
w.cid,
w.value,
rev,
indexedAt,
);
}
sql.exec(
`INSERT INTO space_record_oplog
(space, rev, idx, action, collection, rkey, cid, prev)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
space,
rev,
idx,
w.action,
w.collection,
w.rkey,
w.cid,
w.prev,
);
});
sql.exec(
'INSERT OR REPLACE INTO space_repo (space, set_hash, rev) VALUES (?, ?, ?)',
space,
setHash,
rev,
);
});
},
async getSpaceRepo(space) {
const row = one(
'SELECT set_hash, rev FROM space_repo WHERE space = ?',
space,
);
if (!row) return null;
return { setHash: normalizeBlob(row.set_hash), rev: row.rev };
},
async listSpaceOps(space, since, limit) {
const rows = /** @type {SpaceOpDbRow[]} */ (
all(
'SELECT rev, idx, action, collection, rkey, cid, prev FROM space_record_oplog WHERE space = ? AND rev > ? ORDER BY rev, idx LIMIT ?',
space,
since ?? '',
limit,
)
);
return rows.map((r) => ({
rev: r.rev,
idx: r.idx,
action: /** @type {'create'|'update'|'delete'} */ (r.action),
collection: r.collection,
rkey: r.rkey,
cid: r.cid,
prev: r.prev,
}));
},
async getSpaceWriter(space, did) {
const row = /** @type {SpaceWriterDbRow|null} */ (
one(
'SELECT did, rev, data, space_rev FROM space_writer WHERE space = ? AND did = ?',
space,
did,
)
);
return row ? toWriterRow(row) : null;
},
async getSpaceRevHead(space) {
const row = /** @type {{head: string|null}|null} */ (
one(
'SELECT MAX(space_rev) AS head FROM space_writer WHERE space = ?',
space,
)
);
return row?.head ?? null;
},
async removeSpaceWriter(space, did) {
sql.exec(
'DELETE FROM space_writer WHERE space = ? AND did = ?',
space,
did,
);
},
async putSpaceWriter(space, did, repoRev, data, spaceRev) {
sql.exec(
'INSERT OR REPLACE INTO space_writer (space, did, rev, data, space_rev) VALUES (?, ?, ?, ?, ?)',
space,
did,
repoRev,
data,
spaceRev,
);
},
async listSpaceWriters(space, cursor, limit) {
const rows = /** @type {SpaceWriterDbRow[]} */ (
all(
'SELECT did, rev, data, space_rev FROM space_writer WHERE space = ? AND space_rev > ? ORDER BY space_rev LIMIT ?',
space,
cursor ?? '',
limit,
)
);
const writers = rows.map(toWriterRow);
return {
writers,
cursor: writers.length ? writers[writers.length - 1].spaceRev : null,
};
},
async putCredentialRevocations(space, jtis, expiresAt) {
sql.exec(
'DELETE FROM space_credential_revocation WHERE expires_at <= ?',
new Date().toISOString(),
);
for (const jti of jtis) {
sql.exec(
`INSERT INTO space_credential_revocation (space, jti, expires_at)
VALUES (?, ?, ?)
ON CONFLICT (space, jti) DO UPDATE SET expires_at = MAX(expires_at, excluded.expires_at)`,
space,
jti,
expiresAt,
);
}
},
async isCredentialRevoked(space, jti) {
return Boolean(
one(
'SELECT 1 AS revoked FROM space_credential_revocation WHERE space = ? AND jti = ? AND expires_at > ?',
space,
jti,
new Date().toISOString(),
),
);
},
async enqueueSpaceNotify(row) {
sql.exec(
`INSERT OR REPLACE INTO space_notify_outbox
(space, repo_rev, data, attempts, next_attempt_at, deadline_at)
VALUES (?, ?, ?, 0, ?, ?)`,
row.space,
row.repoRev,
row.data,
row.nextAttemptAt,
row.deadlineAt,
);
},
async listDueSpaceNotify(now, limit) {
const rows = /** @type {SpaceNotifyDbRow[]} */ (
all(
'SELECT space, repo_rev, data, attempts, next_attempt_at, deadline_at FROM space_notify_outbox WHERE next_attempt_at <= ? ORDER BY next_attempt_at LIMIT ?',
now,
limit,
)
);
return rows.map(toNotifyRow);
},
async rescheduleSpaceNotify(space, repoRev, attempts, nextAttemptAt) {
sql.exec(
'UPDATE space_notify_outbox SET attempts = ?, next_attempt_at = ? WHERE space = ? AND repo_rev = ?',
attempts,
nextAttemptAt,
space,
repoRev,
);
},
async deleteSpaceNotify(space, repoRev) {
sql.exec(
'DELETE FROM space_notify_outbox WHERE space = ? AND repo_rev = ?',
space,
repoRev,
);
},
async enqueueSpaceDeletedNotify(row) {
sql.exec(
`INSERT OR REPLACE INTO space_deleted_notify_outbox
(space, service, service_endpoint, attempts, next_attempt_at, deadline_at)
VALUES (?, ?, ?, ?, ?, ?)`,
row.space,
row.service,
row.serviceEndpoint,
row.attempts,
row.nextAttemptAt,
row.deadlineAt,
);
},
async listDueSpaceDeletedNotify(now, limit) {
const rows = /** @type {SpaceDeletedNotifyDbRow[]} */ (
all(
'SELECT space, service, service_endpoint, attempts, next_attempt_at, deadline_at FROM space_deleted_notify_outbox WHERE next_attempt_at <= ? ORDER BY next_attempt_at LIMIT ?',
now,
limit,
)
);
return rows.map(toDeletedNotifyRow);
},
async rescheduleSpaceDeletedNotify(
space,
service,
attempts,
nextAttemptAt,
) {
sql.exec(
'UPDATE space_deleted_notify_outbox SET attempts = ?, next_attempt_at = ? WHERE space = ? AND service = ?',
attempts,
nextAttemptAt,
space,
service,
);
},
async deleteSpaceDeletedNotify(space, service) {
sql.exec(
'DELETE FROM space_deleted_notify_outbox WHERE space = ? AND service = ?',
space,
service,
);
},
async nextSpaceNotifyAt() {
const row = /** @type {{next: string|null}|null} */ (
one(
`SELECT MIN(next) AS next FROM (
SELECT MIN(next_attempt_at) AS next FROM space_notify_outbox
UNION ALL
SELECT MIN(next_attempt_at) AS next FROM space_deleted_notify_outbox
)`,
)
);
return row?.next ?? null;
},
async putCredentialRecipient(
space,
serviceDid,
serviceEndpoint,
expiresAt,
) {
sql.exec(
`INSERT OR REPLACE INTO space_credential_recipient
(space, service_did, service_endpoint, expires_at)
VALUES (?, ?, ?, ?)`,
space,
serviceDid,
serviceEndpoint,
expiresAt,
);
},
async deleteCredentialRecipient(space, serviceDid) {
sql.exec(
'DELETE FROM space_credential_recipient WHERE space = ? AND service_did = ?',
space,
serviceDid,
);
},
async listCredentialRecipients(space) {
const rows = /** @type {SpaceRecipientDbRow[]} */ (
all(
'SELECT service_did, service_endpoint, expires_at FROM space_credential_recipient WHERE space = ? AND expires_at > ? ORDER BY service_did',
space,
new Date().toISOString(),
)
);
return rows.map((r) => ({
serviceDid: r.service_did,
serviceEndpoint: r.service_endpoint,
expiresAt: r.expires_at,
}));
},
};
}