/**
 * 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))],
      [],
    );
  }
}