chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/storage-sqlite/space - SQLite SpaceStoragePort (proposal 0016).
//
// The schema is defined once in @pdsjs/core/space-schema and shared with the
// Cloudflare adapter; only the driver differs between them.
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: Uint8Array|Buffer, repo_rev: string, indexed_at: string}} SpaceRecordDbRow */
/** @typedef {Omit<SpaceRecordDbRow, 'value'> & {value?: Uint8Array|Buffer}} 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: Uint8Array|Buffer, space_rev: string}} SpaceWriterDbRow */
/** @typedef {{space: string, repo_rev: string, data: Uint8Array|Buffer, 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 */
/**
* @param {Uint8Array|Buffer|null} value
* @returns {Uint8Array|null}
*/
function toBytes(value) {
if (!value) return null;
return value instanceof Uint8Array && value.constructor === Uint8Array
? value
: 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} */ (toBytes(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} */ (toBytes(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} */ (toBytes(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} */ (toBytes(row.value)) }),
repoRev: row.repo_rev,
indexedAt: row.indexed_at,
};
}
/**
* @param {import('./driver.js').SQLiteDatabase} db
* @returns {import('@pdsjs/core/ports').SpaceStoragePort}
*/
export function createSpaceStorage(db) {
db.exec(SPACE_SCHEMA_SQL);
for (const statement of SPACE_MIGRATIONS) {
try {
db.exec(statement);
} catch {
// Already applied.
}
}
const stmts = {
getSpace: db.prepare('SELECT * FROM space WHERE uri = ?'),
putSpace: db.prepare(
`INSERT OR REPLACE INTO space
(uri, space_did, space_type, is_owner, policy_json, created_at, deleted_at)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
),
deleteSpace: db.prepare('UPDATE space SET deleted_at = ? WHERE uri = ?'),
putMember: db.prepare(
`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`,
),
removeMember: db.prepare(
'DELETE FROM space_member WHERE space = ? AND did = ?',
),
listMembers: db.prepare(
'SELECT did, can_read, can_write FROM space_member WHERE space = ? AND did > ? ORDER BY did LIMIT ?',
),
getMember: db.prepare(
'SELECT can_read, can_write FROM space_member WHERE space = ? AND did = ?',
),
removeAllMembers: db.prepare('DELETE FROM space_member WHERE space = ?'),
removeWriter: db.prepare(
'DELETE FROM space_writer WHERE space = ? AND did = ?',
),
getRecord: db.prepare(
'SELECT collection, rkey, cid, value, repo_rev, indexed_at FROM space_record WHERE space = ? AND collection = ? AND rkey = ?',
),
getRecordCid: db.prepare(
'SELECT cid FROM space_record WHERE space = ? AND collection = ? AND rkey = ?',
),
listAllRecords: db.prepare(
"SELECT collection, rkey, cid FROM space_record WHERE space = ? ORDER BY (collection || '/' || rkey)",
),
putRecord: db.prepare(
`INSERT OR REPLACE INTO space_record
(space, collection, rkey, cid, value, repo_rev, indexed_at)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
),
deleteRecord: db.prepare(
'DELETE FROM space_record WHERE space = ? AND collection = ? AND rkey = ?',
),
putOp: db.prepare(
`INSERT INTO space_record_oplog
(space, rev, idx, action, collection, rkey, cid, prev)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
),
listOps: db.prepare(
'SELECT rev, idx, action, collection, rkey, cid, prev FROM space_record_oplog WHERE space = ? AND rev > ? ORDER BY rev, idx LIMIT ?',
),
putRepo: db.prepare(
'INSERT OR REPLACE INTO space_repo (space, set_hash, rev) VALUES (?, ?, ?)',
),
getRepo: db.prepare('SELECT set_hash, rev FROM space_repo WHERE space = ?'),
getWriter: db.prepare(
'SELECT did, rev, data, space_rev FROM space_writer WHERE space = ? AND did = ?',
),
getSpaceRevHead: db.prepare(
'SELECT MAX(space_rev) AS head FROM space_writer WHERE space = ?',
),
putWriter: db.prepare(
'INSERT OR REPLACE INTO space_writer (space, did, rev, data, space_rev) VALUES (?, ?, ?, ?, ?)',
),
listWriters: db.prepare(
'SELECT did, rev, data, space_rev FROM space_writer WHERE space = ? AND space_rev > ? ORDER BY space_rev LIMIT ?',
),
putRevocation: db.prepare(
`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)`,
),
deleteExpiredRevocations: db.prepare(
'DELETE FROM space_credential_revocation WHERE expires_at <= ?',
),
getRevocation: db.prepare(
'SELECT 1 FROM space_credential_revocation WHERE space = ? AND jti = ? AND expires_at > ?',
),
enqueueNotify: db.prepare(
`INSERT OR REPLACE INTO space_notify_outbox
(space, repo_rev, data, attempts, next_attempt_at, deadline_at)
VALUES (?, ?, ?, 0, ?, ?)`,
),
listDueNotify: db.prepare(
'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 ?',
),
rescheduleNotify: db.prepare(
'UPDATE space_notify_outbox SET attempts = ?, next_attempt_at = ? WHERE space = ? AND repo_rev = ?',
),
deleteNotify: db.prepare(
'DELETE FROM space_notify_outbox WHERE space = ? AND repo_rev = ?',
),
nextNotifyAt: db.prepare(
`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
)`,
),
enqueueDeletedNotify: db.prepare(
`INSERT OR REPLACE INTO space_deleted_notify_outbox
(space, service, service_endpoint, attempts, next_attempt_at, deadline_at)
VALUES (?, ?, ?, ?, ?, ?)`,
),
listDueDeletedNotify: db.prepare(
'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 ?',
),
rescheduleDeletedNotify: db.prepare(
'UPDATE space_deleted_notify_outbox SET attempts = ?, next_attempt_at = ? WHERE space = ? AND service = ?',
),
deleteDeletedNotify: db.prepare(
'DELETE FROM space_deleted_notify_outbox WHERE space = ? AND service = ?',
),
putRecipient: db.prepare(
`INSERT OR REPLACE INTO space_credential_recipient
(space, service_did, service_endpoint, expires_at)
VALUES (?, ?, ?, ?)`,
),
deleteRecipient: db.prepare(
'DELETE FROM space_credential_recipient WHERE space = ? AND service_did = ?',
),
listRecipients: db.prepare(
'SELECT service_did, service_endpoint, expires_at FROM space_credential_recipient WHERE space = ? AND expires_at > ? ORDER BY service_did',
),
};
return {
async getSpace(uri) {
const row = /** @type {SpaceDbRow|undefined} */ (stmts.getSpace.get(uri));
return row ? toSpaceRow(row) : null;
},
async putSpace(space) {
stmts.putSpace.run(
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) {
stmts.deleteSpace.run(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[]} */ (
db
.prepare(
`SELECT * FROM space WHERE ${where.join(' AND ')} ORDER BY uri LIMIT ?`,
)
.all(...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) {
stmts.putMember.run(
space,
did,
access.read ? 1 : 0,
access.write ? 1 : 0,
);
},
async removeMember(space, did) {
stmts.removeMember.run(space, did);
},
async listMembers(space, cursor, limit) {
const rows = /** @type {SpaceMemberDbRow[]} */ (
stmts.listMembers.all(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|undefined} */ (
stmts.getMember.get(space, did)
);
return row
? { read: row.can_read === 1, write: row.can_write === 1 }
: null;
},
async removeAllMembers(space) {
stmts.removeAllMembers.run(space);
},
async getSpaceRecord(space, collection, rkey) {
const row = /** @type {SpaceRecordDbRow|undefined} */ (
stmts.getRecord.get(space, collection, rkey)
);
return row ? toRecordRow(row) : null;
},
async getSpaceRecordCid(space, collection, rkey) {
const row = /** @type {{cid: string}|undefined} */ (
stmts.getRecordCid.get(space, collection, rkey)
);
return row ? row.cid : null;
},
async listSpaceRecords(space, opts) {
const { sql, params } = spaceRecordsQuery(space, opts);
const rows = /** @type {SpaceRecordListDbRow[]} */ (
db.prepare(sql).all(...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}>} */ (
stmts.listAllRecords.all(space)
);
},
async commitSpaceWrite(space, { writes, setHash, rev, indexedAt }) {
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}`,
);
}
}
db.exec('BEGIN');
try {
writes.forEach((w, idx) => {
if (w.action === 'delete') {
stmts.deleteRecord.run(space, w.collection, w.rkey);
} else {
stmts.putRecord.run(
space,
w.collection,
w.rkey,
w.cid,
w.value,
rev,
indexedAt,
);
}
stmts.putOp.run(
space,
rev,
idx,
w.action,
w.collection,
w.rkey,
w.cid,
w.prev,
);
});
stmts.putRepo.run(space, setHash, rev);
db.exec('COMMIT');
} catch (err) {
// Never let a rollback failure replace the real error.
try {
db.exec('ROLLBACK');
} catch {
// transaction already unwound by SQLite
}
throw err;
}
},
async getSpaceRepo(space) {
const row =
/** @type {{set_hash: Uint8Array|Buffer|null, rev: string|null}|undefined} */ (
stmts.getRepo.get(space)
);
if (!row) return null;
return { setHash: toBytes(row.set_hash), rev: row.rev };
},
async listSpaceOps(space, since, limit) {
const rows = /** @type {SpaceOpDbRow[]} */ (
stmts.listOps.all(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|undefined} */ (
stmts.getWriter.get(space, did)
);
return row ? toWriterRow(row) : null;
},
async getSpaceRevHead(space) {
const row = /** @type {{head: string|null}|undefined} */ (
stmts.getSpaceRevHead.get(space)
);
return row?.head ?? null;
},
async putSpaceWriter(space, did, repoRev, data, spaceRev) {
stmts.putWriter.run(space, did, repoRev, data, spaceRev);
},
async removeSpaceWriter(space, did) {
stmts.removeWriter.run(space, did);
},
async listSpaceWriters(space, cursor, limit) {
const rows = /** @type {SpaceWriterDbRow[]} */ (
stmts.listWriters.all(space, cursor ?? '', limit)
);
const writers = rows.map(toWriterRow);
return {
cursor: writers.length ? writers[writers.length - 1].spaceRev : null,
writers,
};
},
async putCredentialRevocations(space, jtis, expiresAt) {
stmts.deleteExpiredRevocations.run(new Date().toISOString());
for (const jti of jtis) stmts.putRevocation.run(space, jti, expiresAt);
},
async isCredentialRevoked(space, jti) {
return Boolean(
stmts.getRevocation.get(space, jti, new Date().toISOString()),
);
},
async enqueueSpaceNotify(row) {
stmts.enqueueNotify.run(
row.space,
row.repoRev,
row.data,
row.nextAttemptAt,
row.deadlineAt,
);
},
async listDueSpaceNotify(now, limit) {
const rows = /** @type {SpaceNotifyDbRow[]} */ (
stmts.listDueNotify.all(now, limit)
);
return rows.map(toNotifyRow);
},
async rescheduleSpaceNotify(space, repoRev, attempts, nextAttemptAt) {
stmts.rescheduleNotify.run(attempts, nextAttemptAt, space, repoRev);
},
async deleteSpaceNotify(space, repoRev) {
stmts.deleteNotify.run(space, repoRev);
},
async enqueueSpaceDeletedNotify(row) {
stmts.enqueueDeletedNotify.run(
row.space,
row.service,
row.serviceEndpoint,
row.attempts,
row.nextAttemptAt,
row.deadlineAt,
);
},
async listDueSpaceDeletedNotify(now, limit) {
const rows = /** @type {SpaceDeletedNotifyDbRow[]} */ (
stmts.listDueDeletedNotify.all(now, limit)
);
return rows.map(toDeletedNotifyRow);
},
async rescheduleSpaceDeletedNotify(
space,
service,
attempts,
nextAttemptAt,
) {
stmts.rescheduleDeletedNotify.run(
attempts,
nextAttemptAt,
space,
service,
);
},
async deleteSpaceDeletedNotify(space, service) {
stmts.deleteDeletedNotify.run(space, service);
},
async nextSpaceNotifyAt() {
const row = /** @type {{next: string|null}|undefined} */ (
stmts.nextNotifyAt.get()
);
return row?.next ?? null;
},
async putCredentialRecipient(
space,
serviceDid,
serviceEndpoint,
expiresAt,
) {
stmts.putRecipient.run(space, serviceDid, serviceEndpoint, expiresAt);
},
async deleteCredentialRecipient(space, serviceDid) {
stmts.deleteRecipient.run(space, serviceDid);
},
async listCredentialRecipients(space) {
const rows = /** @type {SpaceRecipientDbRow[]} */ (
stmts.listRecipients.all(space, new Date().toISOString())
);
return rows.map((r) => ({
serviceDid: r.service_did,
serviceEndpoint: r.service_endpoint,
expiresAt: r.expires_at,
}));
},
};
}