| 1 | import { ByteWriter, FLUSH, DELIM, ZERO_OID, fromUtf8, planFetch, parseTag, pkt, pktText, readPkts, sideband, packStream, utf8, type ObjectReader } from "@diffinery/git-core"; |
| 2 | import type { RawObject } from "../repo/objects.js"; |
| 3 | |
| 4 | export const AGENT = "diffinery/0.1"; |
| 5 | |
| 6 | export type RefEntry = { |
| 7 | name: string; |
| 8 | oid: string; |
| 9 | }; |
| 10 | |
| 11 | export interface UploadContext { |
| 12 | /** `readRawAll`: the same objects as readRaw, in order, read in batches (the repository's store has it). */ |
| 13 | store: ObjectReader & { readRaw(oid: string): Promise<RawObject | null>; readRawAll?(oids: readonly string[]): AsyncIterable<RawObject>; has(oid: string): boolean }; |
| 14 | refs: RefEntry[]; |
| 15 | /** e.g. refs/heads/main — HEAD points here even when the branch does not exist yet. */ |
| 16 | headTarget: string; |
| 17 | } |
| 18 | |
| 19 | // ------------------------------------------------------------------ advertisements |
| 20 | |
| 21 | export function advertiseUploadPackV2(): Uint8Array { |
| 22 | const w = new ByteWriter(); |
| 23 | w.write(pkt("version 2\n")); |
| 24 | w.write(pkt(`agent=${AGENT}\n`)); |
| 25 | w.write(pkt("ls-refs=unborn\n")); |
| 26 | w.write(pkt("fetch=shallow wait-for-done\n")); |
| 27 | w.write(pkt("server-option\n")); |
| 28 | w.write(pkt("object-format=sha1\n")); |
| 29 | w.write(FLUSH); |
| 30 | return w.bytes(); |
| 31 | } |
| 32 | |
| 33 | export function advertiseReceivePack(refs: RefEntry[]): Uint8Array { |
| 34 | const w = new ByteWriter(); |
| 35 | w.write(pkt("# service=git-receive-pack\n")); |
| 36 | w.write(FLUSH); |
| 37 | const caps = `report-status report-status-v2 delete-refs side-band-64k quiet atomic ofs-delta push-options object-format=sha1 agent=${AGENT}`; |
| 38 | if (refs.length === 0) { |
| 39 | w.write(pkt(`${ZERO_OID} capabilities^{}\0${caps}\n`)); |
| 40 | } else { |
| 41 | refs.forEach((r, i) => w.write(pkt(i === 0 ? `${r.oid} ${r.name}\0${caps}\n` : `${r.oid} ${r.name}\n`))); |
| 42 | } |
| 43 | w.write(FLUSH); |
| 44 | return w.bytes(); |
| 45 | } |
| 46 | |
| 47 | // ------------------------------------------------------------------ protocol v2: upload-pack |
| 48 | |
| 49 | interface V2Request { |
| 50 | command: string; |
| 51 | capabilities: string[]; |
| 52 | args: string[]; |
| 53 | } |
| 54 | |
| 55 | function parseV2(body: Uint8Array): V2Request { |
| 56 | const { pkts } = readPkts(body); |
| 57 | const req: V2Request = { command: "", capabilities: [], args: [] }; |
| 58 | let section: "caps" | "args" = "caps"; |
| 59 | for (const p of pkts) { |
| 60 | if (p.type === "delim") { |
| 61 | section = "args"; |
| 62 | continue; |
| 63 | } |
| 64 | if (p.type === "flush") break; |
| 65 | const line = pktText(p); |
| 66 | if (section === "caps") { |
| 67 | if (line.startsWith("command=")) req.command = line.slice(8); |
| 68 | else req.capabilities.push(line); |
| 69 | } else req.args.push(line); |
| 70 | } |
| 71 | return req; |
| 72 | } |
| 73 | |
| 74 | /** Where a response goes, piece by piece; each write resolves once the client has taken it. */ |
| 75 | export type Sink = (bytes: Uint8Array) => Promise<void>; |
| 76 | |
| 77 | export async function handleUploadPackV2(body: Uint8Array, ctx: UploadContext, sink: Sink): Promise<void> { |
| 78 | const req = parseV2(body); |
| 79 | if (req.command === "ls-refs") return sink(await lsRefs(req, ctx)); |
| 80 | if (req.command === "fetch") return fetchCommand(req, ctx, sink); |
| 81 | throw new Error(`unsupported command: ${req.command || "(none)"}`); |
| 82 | } |
| 83 | |
| 84 | async function lsRefs(req: V2Request, ctx: UploadContext): Promise<Uint8Array> { |
| 85 | const prefixes = req.args.filter((a) => a.startsWith("ref-prefix ")).map((a) => a.slice(11)); |
| 86 | const symrefs = req.args.includes("symrefs"); |
| 87 | const peel = req.args.includes("peel"); |
| 88 | const unborn = req.args.includes("unborn"); |
| 89 | const match = (name: string) => prefixes.length === 0 || prefixes.some((p) => name.startsWith(p)); |
| 90 | const w = new ByteWriter(); |
| 91 | const head = ctx.refs.find((r) => r.name === ctx.headTarget); |
| 92 | if (match("HEAD")) { |
| 93 | if (head) w.write(pkt(`${head.oid} HEAD${symrefs ? ` symref-target:${ctx.headTarget}` : ""}\n`)); |
| 94 | else if (unborn) w.write(pkt(`unborn HEAD${symrefs ? ` symref-target:${ctx.headTarget}` : ""}\n`)); |
| 95 | } |
| 96 | for (const r of ctx.refs) { |
| 97 | if (!match(r.name)) continue; |
| 98 | let line = `${r.oid} ${r.name}`; |
| 99 | if (peel && r.name.startsWith("refs/tags/")) { |
| 100 | const o = await ctx.store.read(r.oid); |
| 101 | if (o?.type === "tag") line += ` peeled:${parseTag(o.content).object}`; |
| 102 | } |
| 103 | w.write(pkt(`${line}\n`)); |
| 104 | } |
| 105 | w.write(FLUSH); |
| 106 | return w.bytes(); |
| 107 | } |
| 108 | |
| 109 | async function fetchCommand(req: V2Request, ctx: UploadContext, sink: Sink): Promise<void> { |
| 110 | const wants: string[] = []; |
| 111 | const haves: string[] = []; |
| 112 | let done = false; |
| 113 | let waitForDone = false; |
| 114 | let noProgress = false; |
| 115 | let depth: number | undefined; |
| 116 | let since: number | undefined; |
| 117 | const clientShallow = new Set<string>(); |
| 118 | for (const a of req.args) { |
| 119 | if (a.startsWith("want ")) wants.push(a.slice(5)); |
| 120 | else if (a.startsWith("want-ref ")) { |
| 121 | const ref = ctx.refs.find((r) => r.name === a.slice(9)); |
| 122 | if (ref) wants.push(ref.oid); |
| 123 | } else if (a.startsWith("have ")) haves.push(a.slice(5)); |
| 124 | else if (a === "done") done = true; |
| 125 | else if (a === "wait-for-done") waitForDone = true; |
| 126 | else if (a === "no-progress") noProgress = true; |
| 127 | else if (a.startsWith("shallow ")) clientShallow.add(a.slice(8)); |
| 128 | else if (a.startsWith("deepen ")) depth = Math.max(1, Number.parseInt(a.slice(7), 10) || 1); |
| 129 | else if (a.startsWith("deepen-since ")) since = Number.parseInt(a.slice(13), 10); |
| 130 | else if (a === "deepen-relative") { |
| 131 | // Approximation: treat relative deepening as absolute depth from the wants. |
| 132 | } else if (a.startsWith("deepen-not ")) throw new Error("--shallow-exclude is not supported; use --depth or --shallow-since"); |
| 133 | else if (a.startsWith("filter ")) throw new Error("partial clones (--filter) are not supported yet"); |
| 134 | } |
| 135 | for (const w of wants) if (!ctx.store.has(w)) throw new Error(`not our ref ${w}`); |
| 136 | const common = haves.filter((h) => ctx.store.has(h)); |
| 137 | |
| 138 | const out = new ByteWriter(); |
| 139 | if (!done) { |
| 140 | out.write(pkt("acknowledgments\n")); |
| 141 | if (common.length === 0) out.write(pkt("NAK\n")); |
| 142 | for (const c of common) out.write(pkt(`ACK ${c}\n`)); |
| 143 | if (waitForDone) { |
| 144 | // Negotiation only (git push with push.negotiate): no "ready" and no pack until "done". |
| 145 | out.write(FLUSH); |
| 146 | return sink(out.bytes()); |
| 147 | } |
| 148 | out.write(pkt("ready\n")); |
| 149 | out.write(DELIM); |
| 150 | } |
| 151 | const plan = await planFetch(ctx.store, wants, common, { depth, since, clientShallow }); |
| 152 | if (depth !== undefined || since !== undefined || clientShallow.size) { |
| 153 | out.write(pkt("shallow-info\n")); |
| 154 | for (const oid of plan.shallow) out.write(pkt(`shallow ${oid}\n`)); |
| 155 | for (const oid of plan.unshallow) out.write(pkt(`unshallow ${oid}\n`)); |
| 156 | out.write(DELIM); |
| 157 | } |
| 158 | out.write(pkt("packfile\n")); |
| 159 | const oids = plan.oids; |
| 160 | if (!noProgress) sideband(2, utf8(`Enumerating objects: ${oids.length}, done.\n`), out); |
| 161 | await sink(out.bytes()); |
| 162 | |
| 163 | // The pack goes out as it is read, in side-band packets, so memory holds one object at a time and |
| 164 | // a clone of a large repository can't exhaust the Durable Object. |
| 165 | async function* entries() { |
| 166 | if (ctx.store.readRawAll) { |
| 167 | for await (const raw of ctx.store.readRawAll(oids)) yield { type: raw.type, size: raw.size, deflated: raw.deflated }; |
| 168 | return; |
| 169 | } |
| 170 | for (const oid of oids) { |
| 171 | const raw = await ctx.store.readRaw(oid); |
| 172 | if (!raw) throw new Error(`missing object ${oid}`); |
| 173 | yield { type: raw.type, size: raw.size, deflated: raw.deflated }; |
| 174 | } |
| 175 | } |
| 176 | const band = new SidebandBuffer(sink); |
| 177 | try { |
| 178 | for await (const part of packStream(oids.length, entries())) await band.write(part); |
| 179 | await band.flush(); |
| 180 | } catch (e) { |
| 181 | // Mid-pack, the only way to tell the client is the error band. |
| 182 | const w = new ByteWriter(); |
| 183 | sideband(3, utf8(`${(e as Error).message}\n`), w); |
| 184 | await sink(w.bytes()); |
| 185 | throw e; |
| 186 | } |
| 187 | const end = new ByteWriter(); |
| 188 | if (!noProgress) sideband(2, utf8(`Total ${oids.length} (delta 0), reused ${oids.length} (delta 0)\n`), end); |
| 189 | end.write(FLUSH); |
| 190 | await sink(end.bytes()); |
| 191 | } |
| 192 | |
| 193 | /** |
| 194 | * Gathers pack bytes into full side-band-64k packets (65515 bytes of data each), written in place |
| 195 | * after room for the packet's length and band: a clone passes its whole pack through here, so each |
| 196 | * byte is copied once. Every packet gets a buffer of its own, since the sink may still hold the last. |
| 197 | */ |
| 198 | class SidebandBuffer { |
| 199 | private static readonly MAX = 65515; |
| 200 | private static readonly HEAD = 5; |
| 201 | private buf = new Uint8Array(SidebandBuffer.HEAD + SidebandBuffer.MAX); |
| 202 | private used = 0; |
| 203 | constructor(private sink: Sink) {} |
| 204 | |
| 205 | async write(data: Uint8Array): Promise<void> { |
| 206 | let i = 0; |
| 207 | while (i < data.length) { |
| 208 | const take = Math.min(SidebandBuffer.MAX - this.used, data.length - i); |
| 209 | this.buf.set(data.subarray(i, i + take), SidebandBuffer.HEAD + this.used); |
| 210 | this.used += take; |
| 211 | i += take; |
| 212 | if (this.used === SidebandBuffer.MAX) await this.flush(); |
| 213 | } |
| 214 | } |
| 215 | |
| 216 | async flush(): Promise<void> { |
| 217 | if (!this.used) return; |
| 218 | const packet = this.buf.subarray(0, SidebandBuffer.HEAD + this.used); |
| 219 | packet.set(utf8(packet.length.toString(16).padStart(4, "0")), 0); |
| 220 | packet[4] = 1; |
| 221 | this.buf = new Uint8Array(SidebandBuffer.HEAD + SidebandBuffer.MAX); |
| 222 | this.used = 0; |
| 223 | await this.sink(packet); |
| 224 | } |
| 225 | } |
| 226 | |
| 227 | // ------------------------------------------------------------------ receive-pack |
| 228 | |
| 229 | export interface RefCommand { |
| 230 | oldOid: string; |
| 231 | newOid: string; |
| 232 | ref: string; |
| 233 | } |
| 234 | |
| 235 | export interface ReceiveRequest { |
| 236 | commands: RefCommand[]; |
| 237 | capabilities: Set<string>; |
| 238 | pushOptions: string[]; |
| 239 | pack: Uint8Array | null; |
| 240 | } |
| 241 | |
| 242 | export function parseReceivePack(body: Uint8Array): ReceiveRequest { |
| 243 | const { pkts, offset } = readPkts(body, 0, true); |
| 244 | const commands: RefCommand[] = []; |
| 245 | let capabilities = new Set<string>(); |
| 246 | for (const p of pkts) { |
| 247 | if (p.type !== "data") continue; |
| 248 | let line = fromUtf8(p.data).replace(/\n$/, ""); |
| 249 | const nul = line.indexOf("\0"); |
| 250 | if (nul >= 0) { |
| 251 | capabilities = new Set(line.slice(nul + 1).trim().split(" ").filter(Boolean)); |
| 252 | line = line.slice(0, nul); |
| 253 | } |
| 254 | if (line.startsWith("shallow ")) continue; |
| 255 | const [oldOid, newOid, ref] = line.split(" "); |
| 256 | if (!oldOid || !newOid || !ref) throw new Error(`bad command: ${line}`); |
| 257 | commands.push({ oldOid, newOid, ref }); |
| 258 | } |
| 259 | let rest = offset; |
| 260 | const pushOptions: string[] = []; |
| 261 | if (capabilities.has("push-options")) { |
| 262 | const opts = readPkts(body, rest, true); |
| 263 | for (const p of opts.pkts) if (p.type === "data") pushOptions.push(pktText(p)); |
| 264 | rest = opts.offset; |
| 265 | } |
| 266 | const pack = rest < body.length ? body.subarray(rest) : null; |
| 267 | return { commands, capabilities, pushOptions, pack }; |
| 268 | } |
| 269 | |
| 270 | /** |
| 271 | * `parseReceivePack` for the start of a push body read on its own (a large push staged in R2): the |
| 272 | * commands and push options, and the offset where the pack starts, or null when `head` ends before |
| 273 | * them, so the caller reads more. |
| 274 | */ |
| 275 | export function parseReceivePackHead(head: Uint8Array): { req: ReceiveRequest; packOffset: number } | null { |
| 276 | const complete = (start: number) => { |
| 277 | try { |
| 278 | const r = readPkts(head, start, true); |
| 279 | return r.pkts.at(-1)?.type === "flush" ? r.offset : null; |
| 280 | } catch (e) { |
| 281 | if (/truncated pkt-line/.test((e as Error).message)) return null; |
| 282 | throw e; |
| 283 | } |
| 284 | }; |
| 285 | const commandsEnd = complete(0); |
| 286 | if (commandsEnd === null) return null; |
| 287 | const req = parseReceivePack(head.subarray(0, commandsEnd)); |
| 288 | let packOffset = commandsEnd; |
| 289 | if (req.capabilities.has("push-options")) { |
| 290 | const optionsEnd = complete(commandsEnd); |
| 291 | if (optionsEnd === null) return null; |
| 292 | for (const p of readPkts(head, commandsEnd, true).pkts) if (p.type === "data") req.pushOptions.push(pktText(p)); |
| 293 | packOffset = optionsEnd; |
| 294 | } |
| 295 | return { req, packOffset }; |
| 296 | } |
| 297 | |
| 298 | export interface RefResult { |
| 299 | ref: string; |
| 300 | ok: boolean; |
| 301 | reason?: string; |
| 302 | } |
| 303 | |
| 304 | /** The platform's error for a Durable Object stopped by a new version of its code. */ |
| 305 | export const resetByDeploy = (e: unknown) => e instanceof Error && /durable object reset because its code was updated/i.test(e.message); |
| 306 | |
| 307 | /** |
| 308 | * The unpack status for a push that couldn't be unpacked or stored. Running out of memory says to |
| 309 | * split the push, the wording clients (and the GitHub importer, which then pushes smaller chunks) |
| 310 | * already treat as "too big". |
| 311 | */ |
| 312 | export function unpackFailure(e: unknown): string { |
| 313 | const message = e instanceof Error ? e.message : String(e); |
| 314 | if (e instanceof RangeError || /allocation failed|out of memory/i.test(message)) { |
| 315 | return `the push needs more memory than one push may use; split it into smaller pushes (${message})`; |
| 316 | } |
| 317 | return message; |
| 318 | } |
| 319 | |
| 320 | export function reportStatus(req: ReceiveRequest, unpackError: string | null, results: RefResult[], messages: string[] = []): Uint8Array { |
| 321 | const report = new ByteWriter(); |
| 322 | report.write(pkt(unpackError ? `unpack ${unpackError}\n` : "unpack ok\n")); |
| 323 | for (const r of results) report.write(pkt(r.ok ? `ok ${r.ref}\n` : `ng ${r.ref} ${r.reason ?? "rejected"}\n`)); |
| 324 | report.write(FLUSH); |
| 325 | if (!req.capabilities.has("side-band-64k")) return report.bytes(); |
| 326 | const out = new ByteWriter(); |
| 327 | for (const m of messages) sideband(2, utf8(`${m}\n`), out); |
| 328 | sideband(1, report.bytes(), out); |
| 329 | out.write(FLUSH); |
| 330 | return out.bytes(); |
| 331 | } |
| 332 | |
| 333 | export function errorPacket(message: string): Uint8Array { |
| 334 | const w = new ByteWriter(); |
| 335 | w.write(pkt(`ERR ${message}\n`)); |
| 336 | return w.bytes(); |
| 337 | } |