chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
/**
* git-remote-helper protocol implementation for atproto:// remotes.
*
* Speaks the line protocol git feeds a remote helper on stdin/stdout:
* capabilities, list, fetch and push. Fetch downloads the repo record's
* bundle chain from PDS blob storage and unbundles into the local object
* database. Push creates an incremental bundle, uploads it as chunked
* blobs, and advances the record with a compare-and-swap write.
*/
import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { createInterface } from 'node:readline';
import {
credentialApprove,
credentialFill,
credentialReject,
} from './credential.js';
import {
bundleCreate,
bundleUnbundle,
catObject,
deleteRef,
git,
independentHeads,
isAncestor,
objectExists,
objectTypes,
revParse,
updateRef,
} from './git.js';
import { resolveDidInfo, resolveRepoLocation } from './identity.js';
import { GIT_REPO_COLLECTION, GIT_SPACE_TYPE } from './lexicon.js';
import {
buildBundleIndex,
encodePackIndex,
PACK_INDEX_MIME,
} from './pack-index.js';
import {
compactionStart,
concatChunks,
createRepoKey,
parseRepoRecord,
repoNameOf,
rkeyOf,
splitIntoChunks,
} from './record.js';
import { findSessionByDid, loadSession, saveSession } from './session-store.js';
import { generateSpaceSigKey } from './space-signature.js';
import { parseRemotes, upstreamRemote } from './upstream.js';
import { parseAtprotoUrl } from './url.js';
import { XrpcClient, XrpcError } from './xrpc.js';
/**
* Under the 50 MiB a pds.js server accepts. A server with a smaller limit,
* the reference PDS at 5 MB say, refuses the first chunk with 413 and the
* push halves the size until it takes one.
*/
const DEFAULT_CHUNK_SIZE = 8 * 1024 * 1024;
/** A chunk size a push halves no further; a 413 there is the answer. */
const MIN_CHUNK_SIZE = 64 * 1024;
const DEFAULT_REPACK_THRESHOLD = 16;
const DEFAULT_COMPACT_FACTOR = 2;
/**
* @typedef {Object} PushSpec
* @property {boolean} force
* @property {string} src - empty string for a deletion
* @property {string} dst
*/
/**
* @typedef {{record: import('./record.js').GitRepoRecord, cid: string}} StoredRecord
*/
/**
* A space member's copy of the repository: their record, read from their own
* PDS. The name is the ref component their branches are listed under.
* @typedef {Object} MemberSource
* @property {string} did
* @property {string} name
* @property {XrpcClient} client
* @property {import('./record.js').GitRepoRecord} record
*/
/**
* A member's handle or DID as one git ref component. A DID's colons are the
* only expected illegal characters; anything else unexpected maps to `-` too.
* @param {string} name
* @returns {string}
*/
function refComponent(name) {
return name.replace(/[^A-Za-z0-9._-]/g, '-');
}
export class AtprotoRemoteHelper {
/**
* @param {Object} options
* @param {string} options.url - atproto://<authority>/<repo> remote URL
* @param {NodeJS.ReadableStream} options.input - protocol command stream
* @param {(text: string) => void} options.write - protocol response sink
* @param {(text: string) => void} options.log - progress/diagnostics sink
* @param {Record<string, string|undefined>} [options.env]
*/
constructor(options) {
const { authority, repoName } = parseAtprotoUrl(options.url);
this.authority = authority;
this.repoName = repoName;
/**
* The space holding the repository, once found: null for a public one,
* undefined until ensureSpace has asked.
* @type {{spaceType: string, skey: string}|null|undefined}
*/
this.space = undefined;
/**
* The record key the repository is at, which ensureSpace settles beside
* the space. A public repository carries a minted key; one in a space
* takes the name, because the space is keyed by the name.
* @type {string}
*/
this.rkey = '';
this.input = options.input;
this.write = options.write;
this.log = options.log;
this.env = options.env ?? process.env;
/** @type {import('./identity.js').RepoLocation|null} */
this.location = null;
/** @type {XrpcClient|null} */
this.client = null;
/** @type {StoredRecord|null|undefined} */
this.cached = undefined;
/** @type {import('./credential.js').Credential|null} */
this.usedCredential = null;
/** The chunk size this push uploads at, halved on each 413. */
/** @type {number|null} */
this.chunk = null;
this.readAuthReady = false;
/** @type {{client: XrpcClient, did: string}|null} */
this.pushTarget = null;
/** @type {import('./xrpc.js').SpaceReadAuth|null} */
this.sharedCredential = null;
/** @type {MemberSource[]|null} */
this.memberSources = null;
/** @type {Map<string, Promise<string>>} */
this.kemKeys = new Map();
}
/**
* An account's `#atproto_kem` key, from its DID document, looked up once.
* @param {string} did
* @returns {Promise<string>}
*/
kemKeyOf(did) {
let key = this.kemKeys.get(did);
if (!key) {
key = resolveDidInfo(did, this.env).then((info) => {
if (!info.kemKey) throw new Error(`${did} publishes no KEM key`);
return info.kemKey;
});
this.kemKeys.set(did, key);
}
return key;
}
/** @returns {Promise<{location: import('./identity.js').RepoLocation, client: XrpcClient}>} */
async ensureClient() {
if (!this.location) {
this.location = await resolveRepoLocation(this.authority, this.env);
}
if (!this.client) {
this.client = new XrpcClient(this.location.service);
this.client.readsRepo = this.location.did;
this.client.resolveKemKey = (did) => this.kemKeyOf(did);
}
return { location: this.location, client: this.client };
}
/**
* Where the name on the address leads: a public record, a private one, or
* nothing yet. The account's own listing is read first, anonymously, and
* the name is matched against each record's own name, because a public
* record carries a minted key. Where there is none, the private form is a
* space of the git type under the repository's name, and a space that
* answers is the repository's. A name with neither is a public repository
* about to take its first push, and its key is minted here.
* @returns {Promise<{spaceType: string, skey: string}|null>}
*/
async ensureSpace() {
if (this.space !== undefined) return this.space;
const { location, client } = await this.ensureClient();
const found = (
await client.listRecords(location.did, GIT_REPO_COLLECTION)
).find((row) => repoNameOf(row.value, rkeyOf(row.uri)) === this.repoName);
if (found) {
this.space = null;
this.rkey = rkeyOf(found.uri);
this.cached = { record: parseRepoRecord(found.value), cid: found.cid };
return null;
}
const space = { spaceType: GIT_SPACE_TYPE, skey: this.repoName };
const uri = `at://${location.did}/space/${space.spaceType}/${space.skey}`;
this.space = (await client.spaceExists(uri)) ? space : null;
this.rkey = this.space ? this.repoName : createRepoKey();
return this.space;
}
/**
* The AT-URI of the space holding this repository. The remote URL's
* authority is the space authority, so the URI anchors on its DID.
* @returns {string}
*/
spaceUri() {
if (!this.space || !this.location) {
throw new Error('not a space remote, or the authority is unresolved');
}
return `at://${this.location.did}/space/${this.space.spaceType}/${this.space.skey}`;
}
/**
* The identifier and password to authenticate with, from the environment or
* the git credential store.
* @returns {{identifier: string, password: string, cred: import('./credential.js').Credential|null}}
*/
gatherCredentials() {
let password = this.env.ATPROTO_GIT_PASSWORD;
/** @type {import('./credential.js').Credential|null} */
let cred = null;
if (!password) {
cred = credentialFill(this.authority);
password = cred.password;
}
if (!password) {
throw new Error(
'no credentials. Set ATPROTO_GIT_PASSWORD or store a password with git credential (protocol=atproto)',
);
}
const identifier =
this.env.ATPROTO_GIT_IDENTIFIER ?? cred?.username ?? this.authority;
return { identifier, password, cred };
}
/**
* The PDS hosting the caller's own account, which for a space read may not
* be the repo host. ATPROTO_GIT_USER_SERVICE short-circuits discovery the
* way ATPROTO_GIT_SERVICE does for the remote authority.
* @param {string} identifier - the caller's handle or DID
* @returns {Promise<import('./identity.js').RepoLocation>}
*/
async resolveCallerLocation(identifier) {
const override = this.env.ATPROTO_GIT_USER_SERVICE;
if (override) {
if (identifier.startsWith('did:')) {
return { did: identifier, service: override };
}
const client = new XrpcClient(override);
return { did: await client.resolveHandle(identifier), service: override };
}
if (identifier === this.authority) {
return /** @type {import('./identity.js').RepoLocation} */ (
this.location
);
}
return resolveRepoLocation(identifier, this.env);
}
/**
* The stored OAuth session to use, if any. Tried under the configured
* identifier and the remote authority, then by the repo owner's DID for a
* login stored under a name this remote does not use.
* @param {string} did - the repo owner's DID
* @returns {{identifier: string, session: import('./oauth.js').OauthSession}|null}
*/
findStoredSession(did) {
for (const candidate of [this.env.ATPROTO_GIT_IDENTIFIER, this.authority]) {
if (!candidate) continue;
const session = loadSession(this.env, candidate);
if (session) return { identifier: candidate, session };
}
const session = findSessionByDid(this.env, did);
if (session) {
return { identifier: session.identifier ?? session.did, session };
}
return null;
}
/**
* Bind a stored OAuth session to a client, persisting each rotated token
* pair. A refresh rotates the refresh token too, so a pair that is used and
* not written back is a session lost.
* @param {XrpcClient} client
* @param {string} identifier
* @param {import('./oauth.js').OauthSession} session
* @returns {Promise<void>}
*/
async bindOauthSession(client, identifier, session) {
await client.setOauthSession(session, (updated) =>
saveSession(this.env, identifier, updated),
);
}
/**
* Authenticate space reads. The repo host admits its own account's session,
* so the owner authenticates directly. Anyone else authenticates to their
* own PDS for a delegation token, exchanges it at the space authority for a
* credential bound to a fresh key, and presents that credential with a DPoP
* proof per request.
*
* A stored OAuth session is preferred; the password flow is the fallback.
* @returns {Promise<void>}
*/
async ensureReadAuth() {
if (!this.space || this.readAuthReady) return;
const { location, client } = await this.ensureClient();
if (client.accessJwt || client.oauth) {
client.spaceAuth = { kind: 'session' };
this.readAuthReady = true;
return;
}
const stored = this.findStoredSession(location.did);
if (stored) {
const { identifier, session } = stored;
if (session.did === location.did) {
await this.bindOauthSession(client, identifier, session);
client.spaceAuth = { kind: 'session' };
} else {
const callerClient = new XrpcClient(session.service);
await this.bindOauthSession(callerClient, identifier, session);
await this.credentialReadAuth(client, callerClient);
}
this.readAuthReady = true;
return;
}
const { identifier, password, cred } = this.gatherCredentials();
const caller = await this.resolveCallerLocation(identifier);
const callerClient =
caller.service === location.service
? client
: new XrpcClient(caller.service);
try {
const session = await callerClient.createSession(identifier, password);
if (session.did === location.did) {
client.accessJwt = session.accessJwt;
client.did = session.did;
client.spaceAuth = { kind: 'session' };
} else {
await this.credentialReadAuth(client, callerClient);
}
if (cred) {
this.usedCredential = cred;
credentialApprove(this.authority, cred);
}
this.readAuthReady = true;
} catch (err) {
if (cred && err instanceof XrpcError && err.status === 401) {
credentialReject(this.authority, cred);
}
throw err;
}
}
/**
* The member half of the read chain: a delegation token from the member's
* own PDS, exchanged at the space authority for a credential bound to a
* fresh ephemeral key.
* @param {XrpcClient} client - the repo host client that will read
* @param {XrpcClient} callerClient - authenticated against the member's PDS
* @returns {Promise<void>}
*/
async credentialReadAuth(client, callerClient) {
const delegation = await callerClient.getDelegationToken(this.spaceUri());
const key = await generateSpaceSigKey();
const credential = await client.getSpaceCredential(
this.spaceUri(),
delegation,
key,
);
client.spaceAuth = { kind: 'credential', credential, key };
}
/**
* A space credential for reading member repos on their own hosts. A member
* reader already holds one; the space authority reading their own space
* mints one here, through their own delegation, because a foreign host
* accepts nothing else.
* @returns {Promise<import('./xrpc.js').SpaceReadAuth>}
*/
async ensureCredential() {
const { client } = await this.ensureClient();
await this.ensureReadAuth();
if (client.spaceAuth?.kind === 'credential') return client.spaceAuth;
if (this.sharedCredential) return this.sharedCredential;
const delegation = await client.getDelegationToken(this.spaceUri());
const key = await generateSpaceSigKey();
const credential = await client.getSpaceCredential(
this.spaceUri(),
delegation,
key,
);
this.sharedCredential = { kind: 'credential', credential, key };
return this.sharedCredential;
}
/**
* Every other member's copy of this repository. The writer set comes from
* the authority; each member's record comes from their own PDS, resolved
* through their DID document and read with the space credential. A member
* that cannot be reached costs their branches, not the fetch.
* @returns {Promise<MemberSource[]>}
*/
async loadMemberSources() {
if (this.memberSources) return this.memberSources;
const { location, client } = await this.ensureClient();
await this.ensureReadAuth();
/** @type {MemberSource[]} */
const sources = [];
// The writer set admits no session, not even the space authority's own, so
// the credential the member reads need is minted before it is asked for.
client.spaceAuth = await this.ensureCredential();
const writers = await client.listSpaceRepos(this.spaceUri());
for (const writer of writers) {
if (writer.did === location.did) continue;
try {
const info = await resolveDidInfo(writer.did, this.env);
const memberClient = new XrpcClient(info.service);
memberClient.readsRepo = writer.did;
memberClient.resolveKemKey = async () => {
if (!info.kemKey)
throw new Error(`${writer.did} publishes no KEM key`);
return info.kemKey;
};
memberClient.spaceAuth = await this.ensureCredential();
const result = await memberClient.getSpaceRecord(
this.spaceUri(),
GIT_REPO_COLLECTION,
this.repoName,
);
if (!result) continue;
sources.push({
did: writer.did,
name: refComponent(info.handle ?? writer.did),
client: memberClient,
record: parseRepoRecord(result.value),
});
} catch (err) {
this.log(
`warning: skipping member ${writer.did}: ${err instanceof Error ? err.message : err}\n`,
);
}
}
this.memberSources = sources;
return sources;
}
/**
* @param {boolean} fresh - bypass the cached copy (pushes need current CAS state)
* @returns {Promise<StoredRecord|null>}
*/
async loadRecord(fresh) {
await this.ensureSpace();
if (!fresh && this.cached !== undefined) return this.cached;
const { location, client } = await this.ensureClient();
/** @type {{cid: string, value: unknown}|null} */
let result;
if (this.space) {
await this.ensureReadAuth();
result = await client.getSpaceRecord(
this.spaceUri(),
GIT_REPO_COLLECTION,
this.rkey,
);
} else {
result = await client.getRecord(
location.did,
GIT_REPO_COLLECTION,
this.rkey,
);
}
this.cached = result
? { record: parseRepoRecord(result.value), cid: result.cid }
: null;
return this.cached;
}
/** @returns {number} */
chunkSize() {
const raw = this.env.ATPROTO_GIT_CHUNK_SIZE;
const parsed = raw ? Number.parseInt(raw, 10) : NaN;
return Number.isFinite(parsed) && parsed > 0 ? parsed : DEFAULT_CHUNK_SIZE;
}
/** @returns {number} */
repackThreshold() {
const raw = this.env.ATPROTO_GIT_REPACK_THRESHOLD;
const parsed = raw ? Number.parseInt(raw, 10) : NaN;
return Number.isFinite(parsed) && parsed > 1
? parsed
: DEFAULT_REPACK_THRESHOLD;
}
/** @returns {number} the geometric compaction factor; 0 disables merging */
compactFactor() {
const raw = this.env.ATPROTO_GIT_COMPACT_FACTOR;
const parsed = raw ? Number.parseFloat(raw) : NaN;
return Number.isFinite(parsed) && parsed >= 0
? parsed
: DEFAULT_COMPACT_FACTOR;
}
/** Run the protocol loop until git closes stdin. */
async run() {
const rl = createInterface({ input: this.input });
/** @type {string[]} */
const fetchBatch = [];
/** @type {string[]} */
const pushBatch = [];
for await (const line of rl) {
if (line === '') {
if (fetchBatch.length > 0) {
await this.doFetch(fetchBatch.splice(0));
this.write('\n');
} else if (pushBatch.length > 0) {
await this.doPush(pushBatch.splice(0));
}
continue;
}
if (line === 'capabilities') {
this.write('fetch\npush\noption\n\n');
} else if (line === 'list' || line === 'list for-push') {
await this.doList(line === 'list for-push');
} else if (line.startsWith('option ')) {
const name = line.split(' ')[1];
this.write(
name === 'verbosity' || name === 'progress'
? 'ok\n'
: 'unsupported\n',
);
} else if (line.startsWith('fetch ')) {
fetchBatch.push(line.slice('fetch '.length));
} else if (line.startsWith('push ')) {
pushBatch.push(line.slice('push '.length));
} else {
throw new Error(`unknown remote helper command: ${line}`);
}
}
}
/**
* The repository record this copy was taken from, where the checkout says
* so: an atproto remote naming the same repository at another account. A
* push to a fork carries it, which is what makes the fork findable. The
* remote's authority is resolved to its DID, since the record names a DID
* and a handle can move.
* @returns {Promise<string|undefined>}
*/
async upstreamUri() {
const listing = git(['config', '--get-regexp', '^remote\\..*\\.url$']);
if (listing.status !== 0) return undefined;
const remote = upstreamRemote(parseRemotes(listing.stdout), {
authority: this.authority,
repoName: this.repoName,
space: this.space,
});
if (!remote) return undefined;
try {
const [from, to] = await Promise.all([
resolveRepoLocation(remote.authority, this.env),
resolveRepoLocation(this.authority, this.env),
]);
if (from.did === to.did) return undefined;
// The upstream record is the upstream account's own and carries their
// key, so the name on the remote URL is matched against the listing.
const found = (
await new XrpcClient(from.service).listRecords(
from.did,
GIT_REPO_COLLECTION,
)
).find(
(row) => repoNameOf(row.value, rkeyOf(row.uri)) === remote.repoName,
);
return found?.uri;
} catch {
// An upstream that does not resolve is left unnamed; the push stands.
return undefined;
}
}
/** @param {boolean} forPush */
async doList(forPush) {
await this.ensureSpace();
// A space push targets the pusher's own record, so the for-push listing
// shows that record's refs: the bases git computes fast-forwards against.
if (this.space && forPush) {
const target = await this.ensurePushTarget();
const stored = await this.loadTargetRecord(target);
for (const ref of stored?.record.refs ?? []) {
this.write(`${ref.sha} ${ref.name}\n`);
}
this.write('\n');
return;
}
const stored = await this.loadRecord(forPush);
if (stored) {
for (const ref of stored.record.refs) {
this.write(`${ref.sha} ${ref.name}\n`);
}
// HEAD comes from the space authority's record: their default branch is
// the repository's.
const head = stored.record.defaultBranch;
if (
!forPush &&
head &&
stored.record.refs.some((r) => r.name === `refs/heads/${head}`)
) {
this.write(`@refs/heads/${head} HEAD\n`);
}
} else if (!forPush) {
this.log(
`note: no git repo record "${this.repoName}" on ${this.authority} yet\n`,
);
}
if (this.space) {
for (const source of await this.loadMemberSources()) {
for (const ref of source.record.refs) {
if (!ref.name.startsWith('refs/heads/')) continue;
this.write(
`${ref.sha} refs/heads/${source.name}/${ref.name.slice('refs/heads/'.length)}\n`,
);
}
}
}
this.write('\n');
}
/**
* Make the requested objects available locally by replaying the bundle
* chain. Bundles whose heads are all present are skipped.
* @param {string[]} batch - lines of the form "<sha> <name>"
*/
async doFetch(batch) {
const wanted = batch.map((line) => line.split(' ')[0]);
const stored = await this.loadRecord(false);
const sources = this.space ? await this.loadMemberSources() : [];
if (!stored && sources.length === 0) {
throw new Error(`no git repo record "${this.repoName}" to fetch from`);
}
await this.replayChains();
for (const sha of wanted) {
if (!objectExists(sha)) {
throw new Error(
`object ${sha} not present after replaying the bundle chains`,
);
}
}
}
/**
* Replay every chain's bundles into the local object database. Bundles
* whose heads are all present are skipped.
*/
async replayChains() {
const stored = await this.loadRecord(false);
const sources = this.space ? await this.loadMemberSources() : [];
const { location, client } = await this.ensureClient();
const tmp = mkdtempSync(join(tmpdir(), 'atproto-git-'));
try {
// The authority's chain first: a chain is self-contained, but replaying
// the canonical history before member branches keeps failure messages
// pointing at the member whose chain actually broke.
/** @type {Array<{label: string, bundles: import('./record.js').BundleEntry[], download: (cid: string) => Promise<Uint8Array>}>} */
const chains = [];
if (stored) {
chains.push({
label: this.authority,
bundles: stored.record.bundles,
download: (cid) =>
this.space
? client.getSpaceBlob(this.spaceUri(), cid)
: client.getBlob(location.did, cid),
});
}
for (const source of sources) {
chains.push({
label: source.name,
bundles: source.record.bundles,
download: (cid) => source.client.getSpaceBlob(this.spaceUri(), cid),
});
}
let file = 0;
for (const chain of chains) {
let index = 0;
for (const bundle of chain.bundles) {
index += 1;
if (
bundle.heads.length > 0 &&
bundle.heads.every((sha) => objectExists(sha))
) {
continue;
}
this.log(
`Downloading bundle ${index}/${chain.bundles.length} from ${chain.label} (${bundle.parts.length} chunks)\n`,
);
/** @type {Uint8Array[]} */
const chunks = [];
for (const part of bundle.parts) {
chunks.push(await chain.download(part.ref.$link));
}
file += 1;
const path = join(tmp, `bundle-${file}.bundle`);
writeFileSync(path, concatChunks(chunks));
bundleUnbundle(path);
}
}
} finally {
rmSync(tmp, { recursive: true, force: true });
}
}
/**
* Where a space push writes: the pusher's own record on their own PDS. The
* space authority's record is the canonical one; any other member's pushes
* advance their own copy, which fetch lists under their name.
* @returns {Promise<{client: XrpcClient, did: string}>}
*/
async ensurePushTarget() {
if (this.pushTarget) return this.pushTarget;
const { location, client } = await this.ensureClient();
const stored = this.findStoredSession(location.did);
if (stored) {
const { identifier, session } = stored;
if (session.did === location.did) {
await this.bindOauthSession(client, identifier, session);
await this.ensureReadAuth();
this.pushTarget = { client, did: location.did };
} else {
const pushClient = new XrpcClient(session.service);
await this.bindOauthSession(pushClient, identifier, session);
this.pushTarget = { client: pushClient, did: session.did };
}
return this.pushTarget;
}
const { identifier, password, cred } = this.gatherCredentials();
const caller = await this.resolveCallerLocation(identifier);
if (caller.service === location.service) {
// The owner's host: login enforces that the session is the owner's, so
// a member misconfigured to their authority's PDS fails loudly here.
await this.login(location, client);
await this.ensureReadAuth();
this.pushTarget = { client, did: location.did };
return this.pushTarget;
}
const pushClient = new XrpcClient(caller.service);
try {
const session = await pushClient.createSession(identifier, password);
if (cred) {
this.usedCredential = cred;
credentialApprove(this.authority, cred);
}
this.pushTarget = { client: pushClient, did: session.did };
return this.pushTarget;
} catch (err) {
if (cred && err instanceof XrpcError && err.status === 401) {
credentialReject(this.authority, cred);
}
throw err;
}
}
/**
* The push target's own record, read fresh: a push compares and swaps
* against it, so a cached copy would turn a concurrent push into a lost
* one.
* @param {{client: XrpcClient, did: string}} target
* @returns {Promise<StoredRecord|null>}
*/
async loadTargetRecord(target) {
const result = await target.client.getSpaceRecord(
this.spaceUri(),
GIT_REPO_COLLECTION,
this.rkey,
);
return result
? { record: parseRepoRecord(result.value), cid: result.cid }
: null;
}
/**
* Authenticate for a push. Password comes from ATPROTO_GIT_PASSWORD or the
* git credential store; the identifier defaults to the remote authority.
* @param {import('./identity.js').RepoLocation} location
* @param {XrpcClient} client
*/
async login(location, client) {
if (client.accessJwt || client.oauth) return;
// A stored OAuth session for the repo owner beats the password flow. A
// session for anyone else cannot push (a repo has one writer), so it
// falls through rather than failing.
const stored = this.findStoredSession(location.did);
if (stored && stored.session.did === location.did) {
await this.bindOauthSession(client, stored.identifier, stored.session);
return;
}
const { identifier, password, cred } = this.gatherCredentials();
try {
const session = await client.createSession(identifier, password);
if (session.did !== location.did) {
throw new Error(
`authenticated as ${session.did} but the remote is owned by ${location.did}`,
);
}
if (cred) {
this.usedCredential = cred;
credentialApprove(this.authority, cred);
}
} catch (err) {
if (cred && err instanceof XrpcError && err.status === 401) {
credentialReject(this.authority, cred);
}
throw err;
}
}
/** @param {string} line @returns {PushSpec} */
parsePushSpec(line) {
const force = line.startsWith('+');
const spec = force ? line.slice(1) : line;
const colon = spec.indexOf(':');
if (colon === -1) {
throw new Error(`malformed push refspec: ${line}`);
}
return { force, src: spec.slice(0, colon), dst: spec.slice(colon + 1) };
}
/** @param {string[]} batch - refspec lines, "+" prefix marks a force push */
async doPush(batch) {
const specs = batch.map((line) => this.parsePushSpec(line));
/** @type {Map<string, string>} */
const results = new Map();
try {
await this.pushSpecs(specs, results);
} catch (err) {
const reason =
err instanceof Error ? err.message.replace(/\s+/g, ' ') : String(err);
for (const spec of specs) {
if (!results.has(spec.dst)) results.set(spec.dst, `error ${reason}`);
}
}
for (const spec of specs) {
const result = results.get(spec.dst) ?? 'error unknown failure';
this.write(
result === 'ok'
? `ok ${spec.dst}\n`
: `error ${spec.dst} ${result.slice('error '.length)}\n`,
);
}
this.write('\n');
}
/**
* @param {PushSpec[]} specs
* @param {Map<string, string>} results - dst to "ok" or "error <reason>"
*/
async pushSpecs(specs, results) {
/** @type {{client: XrpcClient, did: string}} */
let target;
/** @type {StoredRecord|null} */
let stored;
await this.ensureSpace();
if (this.space) {
target = await this.ensurePushTarget();
stored = await this.loadTargetRecord(target);
} else {
const { location, client } = await this.ensureClient();
await this.login(location, client);
target = { client, did: location.did };
stored = await this.loadRecord(true);
}
const client = target.client;
/** @type {Map<string, string>} */
const remoteRefs = new Map(
(stored?.record.refs ?? []).map((r) => [r.name, r.sha]),
);
/** @type {PushSpec[]} */
const updates = [];
/** @type {PushSpec[]} */
const deletions = [];
for (const spec of specs) {
if (spec.src === '') {
if (remoteRefs.delete(spec.dst)) {
deletions.push(spec);
} else {
results.set(spec.dst, 'error remote ref does not exist');
}
continue;
}
const srcSha = revParse(spec.src);
if (!srcSha) {
results.set(spec.dst, `error cannot resolve ${spec.src}`);
continue;
}
const oldSha = remoteRefs.get(spec.dst);
if (oldSha === srcSha) {
results.set(spec.dst, 'ok');
continue;
}
if (oldSha && !spec.force) {
if (!objectExists(oldSha)) {
results.set(
spec.dst,
'error remote ref is not known locally, fetch first',
);
continue;
}
if (!isAncestor(oldSha, srcSha)) {
results.set(spec.dst, 'error non-fast-forward, fetch or force push');
continue;
}
}
updates.push(spec);
}
if (updates.length === 0 && deletions.length === 0) return;
/** @type {import('./record.js').BundleEntry[]} */
let bundles = stored ? [...stored.record.bundles] : [];
// The basis must cover every object already in the chain; a repeated
// object makes the smart HTTP merge (pack.js) serve a pack a clone
// rejects. Tips alone miss history that only a superseded head
// reaches. Each bundle's objects are reachable from its recorded
// heads, so the tips plus all recorded heads cover the chain. A chain
// replay supplies any basis object missing locally.
let basis = [
...new Set([
...remoteRefs.values(),
...bundles.flatMap((bundle) => bundle.heads),
]),
];
// basis is deduplicated, so the map size tells whether every id exists.
let basisComplete = true;
if (updates.length > 0) {
let present = objectTypes(basis);
if (present.size < basis.length) {
await this.replayChains();
present = objectTypes(basis);
}
if (present.size < basis.length) {
this.log(
`warning: ${basis.length - present.size} basis object(s) unavailable; the new bundle may repeat chain objects until a repack\n`,
);
basis = basis.filter((sha) => present.has(sha));
basisComplete = false;
}
}
if (updates.length > 0) {
const heads = updates.map(
(spec) => /** @type {string} */ (revParse(spec.src)),
);
const bytes = this.buildBundle(
updates.map((spec) => spec.src),
basis,
);
if (bytes) {
bundles = await this.appendCompacted(
client,
bundles,
bytes,
heads,
basisComplete,
);
}
}
for (const spec of updates) {
remoteRefs.set(spec.dst, /** @type {string} */ (revParse(spec.src)));
}
/** @type {import('./record.js').RefEntry[]} */
const refs = [...remoteRefs.entries()]
.map(([name, sha]) => ({ name, sha }))
.sort((a, b) => (a.name < b.name ? -1 : 1));
if (
bundles.length >= this.repackThreshold() &&
refs.every((ref) => objectExists(ref.sha))
) {
this.log('Repacking bundle chain into a single full bundle\n');
const repacked = await this.repack(client, refs);
if (repacked) bundles = [repacked];
}
const now = new Date().toISOString();
const branches = refs.filter((r) => r.name.startsWith('refs/heads/'));
const defaultBranch =
stored?.record.defaultBranch ??
(branches.some((r) => r.name === 'refs/heads/main')
? 'main'
: branches[0]?.name.slice('refs/heads/'.length));
const upstream = (await this.upstreamUri()) ?? stored?.record.upstream;
/** @type {import('./record.js').GitRepoRecord} */
const record = {
$type: GIT_REPO_COLLECTION,
name: this.repoName,
description: stored?.record.description,
defaultBranch,
...(upstream ? { upstream } : {}),
refs,
bundles,
createdAt: stored?.record.createdAt ?? now,
updatedAt: now,
};
try {
if (this.space) {
await client.putSpaceRecord(
this.spaceUri(),
target.did,
GIT_REPO_COLLECTION,
this.rkey,
record,
stored?.cid ?? null,
);
} else {
await client.putRecord(
target.did,
GIT_REPO_COLLECTION,
this.rkey,
record,
stored?.cid ?? null,
);
}
} catch (err) {
if (err instanceof XrpcError && err.error === 'InvalidSwap') {
throw new Error('remote was updated concurrently, fetch and retry');
}
if (
this.space &&
err instanceof XrpcError &&
err.error === 'InvalidToken' &&
err.status === 403
) {
throw new Error(
'this session cannot write to the space. Use the account password, or log in with a space: scope that covers this repo; an app password has no space access',
);
}
throw err;
}
this.cached = undefined;
this.memberSources = null;
for (const spec of [...updates, ...deletions]) {
results.set(spec.dst, 'ok');
}
}
/**
* Write a bundle holding the positives minus the basis and return its
* bytes. Returns null when the bundle would be empty (refs-only update).
* @param {string[]} positives - local ref names to include
* @param {string[]} basis - object ids the receiver already has
* @returns {Uint8Array|null}
*/
buildBundle(positives, basis) {
const tmp = mkdtempSync(join(tmpdir(), 'atproto-git-'));
try {
const file = join(tmp, 'push.bundle');
if (!bundleCreate(file, positives, basis)) return null;
return new Uint8Array(readFileSync(file));
} finally {
rmSync(tmp, { recursive: true, force: true });
}
}
/**
* Upload bundle bytes as chunked blobs and return the chain entry.
* @param {XrpcClient} client
* @param {Uint8Array} bytes
* @param {string[]} heads - object ids the bundle's refs point at
* @returns {Promise<import('./record.js').BundleEntry>}
*/
async uploadBundle(client, bytes, heads) {
this.log(`Uploading bundle (${bytes.length} bytes)\n`);
const parts = await this.uploadChunked(client, bytes);
const index = await this.uploadBundleIndex(client, bytes);
// One blob keeps the single-blob field, which is what a reader written
// before the chunked form looks for.
return {
heads: [...new Set(heads)],
parts,
...(index?.length === 1 ? { index: index[0] } : {}),
...(index && index.length > 1 ? { indexParts: index } : {}),
};
}
/**
* Upload bytes as blobs of the push's chunk size. A server that refuses a
* chunk as too large gets the whole payload again at half the size, down
* to MIN_CHUNK_SIZE; the blobs it already took go unreferenced and the
* server's orphan sweep reaps them.
* @param {XrpcClient} client
* @param {Uint8Array} bytes
* @param {string} [mimeType]
* @returns {Promise<import('./record.js').BlobObject[]>}
*/
async uploadChunked(client, bytes, mimeType) {
this.chunk ??= this.chunkSize();
for (;;) {
const chunks = splitIntoChunks(bytes, this.chunk);
/** @type {import('./record.js').BlobObject[]} */
const parts = [];
try {
for (const chunk of chunks) {
parts.push(await client.uploadBlob(chunk, mimeType));
}
return parts;
} catch (err) {
const refused =
err instanceof XrpcError &&
(err.status === 413 || err.error === 'PayloadTooLarge');
if (!refused || this.chunk <= MIN_CHUNK_SIZE) throw err;
this.chunk = Math.max(MIN_CHUNK_SIZE, Math.floor(this.chunk / 2));
this.log(
`Server refused a ${chunks[0].length}-byte blob; retrying in ${this.chunk}-byte chunks\n`,
);
}
}
}
/**
* Build the bundle's object index and upload it. The index lets a reader
* fetch single objects with ranged blob reads. It is an optional
* companion: a repository whose bundle defeats the builder (an object
* form it cannot walk, a bundle past the index format's reach) still
* pushes, and readers download whole bundles as they do for records
* written before the index existed.
* @param {XrpcClient} client
* @param {Uint8Array} bytes - the complete bundle file
* @returns {Promise<import('./record.js').BlobObject[]|null>} the index
* blobs in order, or null if the index was skipped
*/
async uploadBundleIndex(client, bytes) {
try {
// An incremental bundle may delta against its prerequisites, which
// the local repository holds.
const index = await buildBundleIndex(bytes, async (sha) =>
catObject(sha),
);
// A repository large enough indexes past one blob, so the index is
// chunked the way the bundle is.
return await this.uploadChunked(
client,
encodePackIndex(index),
PACK_INDEX_MIME,
);
} catch (err) {
this.log(
`warning: bundle index skipped: ${err instanceof Error ? err.message : err}\n`,
);
return null;
}
}
/**
* Build and upload one bundle whose refs are the given heads, named
* through temporary local refs so bundle create has names to record.
* @param {XrpcClient} client
* @param {string[]} heads - object ids, deduplicated
* @param {string[]} basis - object ids the receiver already has
* @returns {Promise<import('./record.js').BundleEntry|null>}
*/
async bundleFromHeads(client, heads, basis) {
/** @type {string[]} */
const tempRefs = [];
try {
heads.forEach((sha, i) => {
const name = `refs/atproto-repack/${i}`;
updateRef(name, sha);
tempRefs.push(name);
});
const bytes = this.buildBundle(tempRefs, basis);
return bytes ? await this.uploadBundle(client, bytes, heads) : null;
} finally {
for (const name of tempRefs) deleteRef(name);
}
}
/**
* Append the push bundle to the chain, first merging the tail that
* compactionStart selects into one bundle. The merged bundle's refs are
* the independent subset of the tail's heads plus the push's, so history
* a superseded head reaches stays covered without the head list growing
* by one per push. Its basis is the heads of the bundles it keeps in
* place, which keeps the chain disjoint. The merge needs every chain
* head present locally and every bundle to name its heads; otherwise the
* push bundle appends unmerged.
* @param {XrpcClient} client
* @param {import('./record.js').BundleEntry[]} bundles
* @param {Uint8Array} bytes - the built push bundle
* @param {string[]} heads - object ids the pushed refs point at
* @param {boolean} canMerge - every chain head exists locally
* @returns {Promise<import('./record.js').BundleEntry[]>}
*/
async appendCompacted(client, bundles, bytes, heads, canMerge) {
const sizes = [
...bundles.map((b) => b.parts.reduce((sum, p) => sum + p.size, 0)),
bytes.length,
];
const start = compactionStart(sizes, this.compactFactor());
if (
canMerge &&
start < bundles.length &&
bundles.every((b) => b.heads.length > 0)
) {
const merged = independentHeads([
...new Set([...bundles.slice(start).flatMap((b) => b.heads), ...heads]),
]);
const kept = independentHeads([
...new Set(bundles.slice(0, start).flatMap((b) => b.heads)),
]);
this.log(`Compacting ${bundles.length - start + 1} bundles into one\n`);
const entry = await this.bundleFromHeads(client, merged, kept);
if (entry) return [...bundles.slice(0, start), entry];
}
return [...bundles, await this.uploadBundle(client, bytes, heads)];
}
/**
* Build one full bundle covering every remote ref.
* @param {XrpcClient} client
* @param {import('./record.js').RefEntry[]} refs
* @returns {Promise<import('./record.js').BundleEntry|null>}
*/
repack(client, refs) {
return this.bundleFromHeads(
client,
[...new Set(refs.map((r) => r.sha))],
[],
);
}
}