chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
/**
* A read-only git object store built from packfiles, no git binary needed:
* entries are walked sequentially (pako's total_in reveals each zlib
* stream's length, which is the only way to find entry boundaries), deltas
* are resolved against their bases, and objects are indexed by SHA-1 id.
* Enough to serve a file browser: resolve a ref to a commit, walk trees,
* read blobs.
*
* Workers-clean: pako is pure JS.
*/
import pako from 'pako';
import { splitPack } from './pack.js';
import { Sha1 } from './sha1.js';
const OBJ_COMMIT = 1;
const OBJ_TREE = 2;
const OBJ_BLOB = 3;
const OBJ_TAG = 4;
const OBJ_OFS_DELTA = 6;
const OBJ_REF_DELTA = 7;
/** @type {Record<number, string>} */
const TYPE_NAMES = {
[OBJ_COMMIT]: 'commit',
[OBJ_TREE]: 'tree',
[OBJ_BLOB]: 'blob',
[OBJ_TAG]: 'tag',
};
/** How many objects indexPacks names in one digest batch. */
const HASH_BATCH = 512;
/**
* How many bytes of materialized objects indexPacks keeps as delta bases
* when it is not retaining the whole graph. A base evicted below this line
* is built again from the pack, so the number trades work for memory.
*/
const CACHE_BYTES = 64 * 1024 * 1024;
/**
* Whether an entry is part of the commit graph, which is what a caller
* walking history keeps.
* @param {PackEntryMeta} meta
* @returns {boolean}
*/
function isHistory(meta) {
return meta.type === 'commit' || meta.type === 'tag';
}
const decoder = new TextDecoder();
/**
* Inflate one zlib stream that starts at `offset`, returning the data and
* where the stream ended.
* @param {Uint8Array} bytes
* @param {number} offset
* @returns {{data: Uint8Array, end: number}}
*/
export function inflateAt(bytes, offset) {
const inflator = new pako.Inflate();
inflator.push(bytes.subarray(offset), true);
if (inflator.err) {
throw new Error(`pack inflate failed at ${offset}: ${inflator.msg}`);
}
// total_in comes from pako's underlying zlib stream state, which the
// typings do not declare
const stream = /** @type {{strm: {total_in: number}}} */ (
/** @type {unknown} */ (inflator)
).strm;
return {
data: /** @type {Uint8Array} */ (inflator.result),
end: offset + stream.total_in,
};
}
/**
* Apply a git delta (copy/insert instruction stream) to its base.
* @param {Uint8Array} delta
* @param {Uint8Array} base
* @returns {Uint8Array}
*/
export function applyDelta(delta, base) {
let pos = 0;
const readVarint = () => {
let value = 0;
let shift = 0;
let byte;
do {
byte = delta[pos++];
value |= (byte & 0x7f) << shift;
shift += 7;
} while (byte & 0x80);
return value;
};
const baseSize = readVarint();
if (baseSize !== base.length) {
throw new Error('delta base size mismatch');
}
const targetSize = readVarint();
const out = new Uint8Array(targetSize);
let outPos = 0;
while (pos < delta.length) {
const op = delta[pos++];
if (op & 0x80) {
let copyOffset = 0;
let copySize = 0;
for (let i = 0; i < 4; i++) {
if (op & (1 << i)) copyOffset |= delta[pos++] << (i * 8);
}
for (let i = 0; i < 3; i++) {
if (op & (0x10 << i)) copySize |= delta[pos++] << (i * 8);
}
if (copySize === 0) copySize = 0x10000;
out.set(base.subarray(copyOffset, copyOffset + copySize), outPos);
outPos += copySize;
} else if (op > 0) {
out.set(delta.subarray(pos, pos + op), outPos);
outPos += op;
pos += op;
} else {
throw new Error('delta opcode 0 is reserved');
}
}
if (outPos !== targetSize) throw new Error('delta produced wrong size');
return out;
}
/**
* @param {string} type
* @param {Uint8Array} content
* @returns {string} the object id
*/
export function objectId(type, content) {
const header = new TextEncoder().encode(`${type} ${content.length}\0`);
const digest = new Sha1().update(header).update(content).digest();
return Array.from(digest)
.map((b) => b.toString(16).padStart(2, '0'))
.join('');
}
/** @typedef {{type: string, data: Uint8Array}} GitObject */
const encoder = new TextEncoder();
/** SHA-1 of the empty input, for proving the platform's digest works. */
const EMPTY_SHA1 = 'da39a3ee5e6b4b0d3255bfef95601890afd80709';
/** @param {Uint8Array} bytes */
function toHex(bytes) {
return Array.from(bytes)
.map((b) => b.toString(16).padStart(2, '0'))
.join('');
}
/**
* The platform's SHA-1, or null where there is none to be had.
*
* Naming every object in a pack means hashing every byte of it, and a
* repository's blobs are almost all of those bytes. The built-in digest does
* that an order of magnitude faster than the one in sha1.js, which is why it
* is worth an async index. Proven against a known digest rather than assumed
* from the presence of the API, since a platform may carry the interface and
* refuse the algorithm.
* @type {Promise<((data: ArrayBuffer) => Promise<string>)|null>|null}
*/
let nativeDigest = null;
function platformDigest() {
if (nativeDigest) return nativeDigest;
nativeDigest = (async () => {
const subtle = globalThis.crypto?.subtle;
if (!subtle) return null;
try {
const proof = await subtle.digest('SHA-1', new Uint8Array(0));
if (toHex(new Uint8Array(proof)) !== EMPTY_SHA1) return null;
return async (/** @type {ArrayBuffer} */ data) =>
toHex(new Uint8Array(await subtle.digest('SHA-1', data)));
} catch {
return null;
}
})();
return nativeDigest;
}
/**
* Parse one pack entry's header from the entry's own bytes. A delta entry
* reads as 'ofs-delta' or 'ref-delta'; its object type is its base's. The
* compressed data begins at dataStart.
* @param {Uint8Array} bytes - the entry's bytes, header first
* @returns {{type: string, size: number, dataStart: number, baseRel: number|null, baseSha: string|null}}
*/
export function readEntryHeader(bytes) {
let pos = 0;
let byte = bytes[pos++];
const typeCode = (byte >> 4) & 0x07;
let size = byte & 0x0f;
let shift = 4;
while (byte & 0x80) {
byte = bytes[pos++];
size += (byte & 0x7f) * 2 ** shift;
shift += 7;
}
/** @type {number|null} */
let baseRel = null;
/** @type {string|null} */
let baseSha = null;
/** @type {string|undefined} */
let type = TYPE_NAMES[typeCode];
if (typeCode === OBJ_OFS_DELTA) {
byte = bytes[pos++];
let rel = byte & 0x7f;
while (byte & 0x80) {
byte = bytes[pos++];
rel = ((rel + 1) << 7) | (byte & 0x7f);
}
baseRel = rel;
type = 'ofs-delta';
} else if (typeCode === OBJ_REF_DELTA) {
baseSha = toHex(bytes.subarray(pos, pos + 20));
pos += 20;
type = 'ref-delta';
}
if (!type) throw new Error(`unknown pack entry type ${typeCode}`);
return { type, size, dataStart: pos, baseRel, baseSha };
}
/**
* Where a pack entry sits and what it resolved to, reported to the
* indexPacks collector. Offsets are relative to the entry's own pack.
* @typedef {Object} PackEntryMeta
* @property {number} pack - index into the packs argument
* @property {number} offset - the entry's first byte
* @property {number} size - materialized object size in bytes
* @property {string} type - resolved object type
* @property {number|null} baseOffset - an ofs-delta entry's base entry offset
* @property {string|null} baseSha - a ref-delta entry's base object id
*/
/**
* Index every object in the given packs. Later packs may ref-delta against
* objects from earlier ones, so pass chained bundle packs oldest first.
* @param {Uint8Array[]} packs - complete packfiles
* @param {(sha: string, meta: PackEntryMeta) => void} [onObject] - called for
* every entry once its id is known, in entry order within each pack
* @param {(sha: string) => Promise<GitObject|null>} [resolveBase] - supplies
* a ref-delta base absent from the given packs. An incremental bundle may
* delta against its prerequisites, which a caller holding the full
* repository can materialize.
* @param {{retain?: 'all'|'commits'|'none', cacheBytes?: number}} [options] -
* `retain` is which objects the returned map holds. A caller that only
* walks history asks for 'commits'; one that only wants the collector asks
* for 'none'. Both cost a bounded amount of memory over the pack itself,
* where 'all' costs the whole materialized object graph. `cacheBytes`
* bounds the delta bases those two modes keep, for a runtime whose memory
* is tighter than the default.
* @returns {Promise<Map<string, GitObject>>} the retained objects by SHA-1 id
*/
export async function indexPacks(packs, onObject, resolveBase, options) {
const retain = options?.retain ?? 'all';
/** Objects the caller asked to keep. */
/** @type {Map<string, GitObject>} */
const bySha = new Map();
/**
* Where each named object sits, which is what lets an evicted delta base
* be built again from the pack bytes.
* @type {Map<string, {pack: number, offset: number}>}
*/
const placeBySha = new Map();
const digest = await platformDigest();
/**
* Objects walked but not yet named. Hashing them together lets the
* platform digest run over the whole batch at once. The batch is bounded:
* flush copies every object it holds into a header-prefixed buffer, so a
* batch the size of a pack costs a second copy of that pack.
* @type {Array<{object: GitObject, meta: PackEntryMeta}>}
*/
let pending = [];
/**
* @param {string} sha
* @param {{object: GitObject, meta: PackEntryMeta}} entry
*/
function emit(sha, entry) {
if (retain === 'all' || (retain === 'commits' && isHistory(entry.meta))) {
bySha.set(sha, entry.object);
}
placeBySha.set(sha, { pack: entry.meta.pack, offset: entry.meta.offset });
onObject?.(sha, entry.meta);
}
async function flush() {
if (pending.length === 0) return;
const batch = pending;
pending = [];
if (!digest) {
for (const entry of batch) {
emit(objectId(entry.object.type, entry.object.data), entry);
}
return;
}
const ids = await Promise.all(
batch.map(({ object }) => {
const header = encoder.encode(`${object.type} ${object.data.length}\0`);
// Sized exactly, so its ArrayBuffer goes to the digest with no
// further copy.
const buffer = new Uint8Array(header.length + object.data.length);
buffer.set(header);
buffer.set(object.data, header.length);
return digest(buffer.buffer);
}),
);
for (let i = 0; i < batch.length; i++) emit(ids[i], batch[i]);
}
/**
* Materialized entries, holding what the deltas still to come need.
* Retaining every object costs the whole inflated graph, so 'commits' and
* 'none' bound this and build an evicted base again from the pack bytes.
* A Map iterates in insertion order, and a hit reinserts, so the first key
* is the least recently used one.
* @type {Map<number, GitObject>}
*/
const cache = new Map();
const cacheLimit =
retain === 'all'
? Number.POSITIVE_INFINITY
: (options?.cacheBytes ?? CACHE_BYTES);
let cacheBytes = 0;
/**
* One key across every pack. A pack offset is below 2^32.
* @param {number} p
* @param {number} offset
*/
const cacheKey = (p, offset) => p * 2 ** 32 + offset;
/**
* @param {number} p
* @param {number} offset
* @param {GitObject} object
*/
function cachePut(p, offset, object) {
const key = cacheKey(p, offset);
cache.set(key, object);
cacheBytes += object.data.length;
while (cacheBytes > cacheLimit && cache.size > 1) {
const oldest = cache.keys().next().value;
if (oldest === undefined || oldest === key) break;
cacheBytes -= /** @type {GitObject} */ (cache.get(oldest)).data.length;
cache.delete(oldest);
}
}
/**
* @param {number} p
* @param {number} offset
* @returns {GitObject|undefined}
*/
function cacheGet(p, offset) {
const key = cacheKey(p, offset);
const object = cache.get(key);
if (object === undefined) return undefined;
cache.delete(key);
cache.set(key, object);
return object;
}
/**
* The entry at this offset, inflated and with its deltas applied. A base
* still cached costs nothing; an evicted one is built again from the pack,
* down its whole delta chain.
* @param {number} p
* @param {number} offset
* @returns {Promise<GitObject>}
*/
async function materialize(p, offset) {
const cached = cacheGet(p, offset);
if (cached) return cached;
/** The chain from the wanted entry down to its deepest missing base. */
/** @type {Array<{pack: number, offset: number}>} */
const chain = [];
/** @type {GitObject|undefined} */
let base;
let at = { pack: p, offset };
for (;;) {
chain.push(at);
const header = readEntryHeader(packs[at.pack].subarray(at.offset));
if (header.baseRel !== null) {
const next = { pack: at.pack, offset: at.offset - header.baseRel };
const hit = cacheGet(next.pack, next.offset);
if (hit) {
base = hit;
break;
}
at = next;
continue;
}
if (header.baseSha !== null) {
base = await baseBySha(header.baseSha);
break;
}
break;
}
for (let i = chain.length - 1; i >= 0; i--) {
const { pack: cp, offset: co } = chain[i];
const header = readEntryHeader(packs[cp].subarray(co));
const { data } = inflateAt(packs[cp], co + header.dataStart);
if (header.baseRel === null && header.baseSha === null) {
base = { type: header.type, data };
} else {
if (!base) throw new Error(`delta base for entry ${co} not found`);
base = { type: base.type, data: applyDelta(data, base.data) };
}
cachePut(cp, co, base);
}
return /** @type {GitObject} */ (base);
}
/**
* @param {string} sha
* @returns {Promise<GitObject|undefined>}
*/
async function baseBySha(sha) {
let place = placeBySha.get(sha);
if (!place) {
// The base is usually an object from an earlier pack, already named.
// A pack that ref-deltas against itself needs this one named first.
await flush();
place = placeBySha.get(sha);
}
if (place) return await materialize(place.pack, place.offset);
if (resolveBase) return (await resolveBase(sha)) ?? undefined;
return undefined;
}
for (let packIndex = 0; packIndex < packs.length; packIndex++) {
const pack = packs[packIndex];
const { count } = splitPack(pack);
let offset = 12;
for (let i = 0; i < count; i++) {
const entryStart = offset;
const header = readEntryHeader(pack.subarray(entryStart));
const { data, end } = inflateAt(pack, entryStart + header.dataStart);
offset = end;
/** @type {GitObject} */
let object;
/** @type {number|null} */
let baseOffset = null;
/** @type {string|null} */
let baseSha = null;
if (header.baseRel !== null) {
baseOffset = entryStart - header.baseRel;
const base =
cacheGet(packIndex, baseOffset) ??
(await materialize(packIndex, baseOffset));
object = { type: base.type, data: applyDelta(data, base.data) };
} else if (header.baseSha !== null) {
baseSha = header.baseSha;
const base = await baseBySha(baseSha);
if (!base) throw new Error(`ref-delta base ${baseSha} not found`);
object = { type: base.type, data: applyDelta(data, base.data) };
} else {
object = { type: header.type, data };
}
cachePut(packIndex, entryStart, object);
if (pending.length >= HASH_BATCH) await flush();
pending.push({
object,
meta: {
pack: packIndex,
offset: entryStart,
size: object.data.length,
type: object.type,
baseOffset,
baseSha,
},
});
}
// Each pack is named before the next one walks, so a ref-delta there
// finds its base.
await flush();
}
return bySha;
}
/** @typedef {{mode: string, name: string, sha: string, type: 'dir'|'file'|'symlink'|'submodule'}} TreeEntry */
/**
* @param {Uint8Array} data - a tree object's content
* @returns {TreeEntry[]}
*/
export function parseTree(data) {
/** @type {TreeEntry[]} */
const entries = [];
let pos = 0;
while (pos < data.length) {
const space = data.indexOf(0x20, pos);
const nul = data.indexOf(0x00, space);
const mode = decoder.decode(data.subarray(pos, space));
const name = decoder.decode(data.subarray(space + 1, nul));
const sha = Array.from(data.subarray(nul + 1, nul + 21))
.map((b) => b.toString(16).padStart(2, '0'))
.join('');
/** @type {TreeEntry['type']} */
let type = 'file';
if (mode === '40000' || mode === '040000') type = 'dir';
else if (mode === '120000') type = 'symlink';
else if (mode === '160000') type = 'submodule';
entries.push({ mode, name, sha, type });
pos = nul + 21;
}
return entries;
}
/**
* The tree id a commit points at.
* @param {Uint8Array} data - a commit object's content
* @returns {string}
*/
export function commitTree(data) {
const text = decoder.decode(data);
const match = /^tree ([0-9a-f]{40})/m.exec(text);
if (!match) throw new Error('commit has no tree line');
return match[1];
}
/**
* @typedef {Object} CommitInfo
* @property {string} tree
* @property {string[]} parents
* @property {string} author - "Name <email>"
* @property {number} authoredAt - epoch ms
* @property {string} committer - "Name <email>"
* @property {number} committedAt - epoch ms
* @property {string} message
*/
/**
* An ident line's name and time. "Name <email> 1700000000 +0000": the epoch
* and zone are the last two tokens, the ident is everything before them.
* @param {string} value
* @returns {{ident: string, at: number}}
*/
function parseIdent(value) {
const match = /^(.*) (\d+) [+-]\d{4}$/.exec(value);
if (!match) return { ident: value, at: 0 };
return { ident: match[1], at: Number(match[2]) * 1000 };
}
/**
* Parse a commit object's headers and message.
* @param {Uint8Array} data - a commit object's content
* @returns {CommitInfo}
*/
export function parseCommit(data) {
const text = decoder.decode(data);
const blank = text.indexOf('\n\n');
const header = blank === -1 ? text : text.slice(0, blank);
const message = blank === -1 ? '' : text.slice(blank + 2);
let tree = '';
/** @type {string[]} */
const parents = [];
let author = { ident: '', at: 0 };
let committer = { ident: '', at: 0 };
for (const line of header.split('\n')) {
// A continuation line (a gpgsig body) belongs to the previous header.
if (line.startsWith(' ')) continue;
const space = line.indexOf(' ');
if (space === -1) continue;
const key = line.slice(0, space);
const value = line.slice(space + 1);
if (key === 'tree') tree = value;
else if (key === 'parent') parents.push(value);
else if (key === 'author') author = parseIdent(value);
else if (key === 'committer') committer = parseIdent(value);
}
return {
tree,
parents,
author: author.ident,
authoredAt: author.at,
committer: committer.ident,
committedAt: committer.at,
message,
};
}
/**
* The object an annotated tag points at.
* @param {Uint8Array} data - a tag object's content
* @returns {string}
*/
export function tagTarget(data) {
const text = decoder.decode(data);
const match = /^object ([0-9a-f]{40})/m.exec(text);
if (!match) throw new Error('tag has no object line');
return match[1];
}
/**
* Resolve any ref target down to a commit id: peel annotated tags.
* @param {Map<string, GitObject>} objects
* @param {string} sha
* @returns {string} a commit object id
*/
export function resolveToCommit(objects, sha) {
let current = sha;
for (let hops = 0; hops < 10; hops++) {
const object = objects.get(current);
if (!object) throw new Error(`object ${current} not found in packs`);
if (object.type === 'tag') {
current = tagTarget(object.data);
continue;
}
if (object.type === 'commit') return current;
throw new Error(`ref points at a ${object.type}, not a commit`);
}
throw new Error('tag chain too deep');
}
/**
* Resolve any ref target down to the tree it names: peel annotated tags,
* then take the commit's tree.
* @param {Map<string, GitObject>} objects
* @param {string} sha
* @returns {string} a tree object id
*/
export function resolveToTree(objects, sha) {
let current = sha;
for (let hops = 0; hops < 10; hops++) {
const object = objects.get(current);
if (!object) throw new Error(`object ${current} not found in packs`);
if (object.type === 'tag') {
current = tagTarget(object.data);
} else if (object.type === 'commit') {
return commitTree(object.data);
} else if (object.type === 'tree') {
return current;
} else {
throw new Error(`ref points at a ${object.type}, not a commit`);
}
}
throw new Error('tag chain too deep');
}