Skip to content

Commit f8908f6

Browse files
chriswritescode-devchriswritescode-dev
committed
feat: add bundle-based OCM mirror sync
1 parent 71ba737 commit f8908f6

7 files changed

Lines changed: 681 additions & 11 deletions

File tree

‎backend/src/routes/internal/repo-mirror.ts‎

Lines changed: 227 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
import { Hono } from 'hono'
22
import type { Database } from 'bun:sqlite'
33
import { spawn } from 'child_process'
4-
import { createWriteStream } from 'fs'
4+
import { copyFileSync, createReadStream, createWriteStream, existsSync } from 'fs'
55
import { mkdirSync, mkdtempSync, writeFileSync } from 'fs'
66
import { Readable } from 'stream'
77
import { pipeline } from 'stream/promises'
@@ -42,8 +42,94 @@ interface CommitBody {
4242
gzip?: boolean
4343
}
4444

45+
interface PatchBody {
46+
baseHead?: string | null
47+
patch?: string
48+
force?: boolean
49+
}
50+
4551
const LEGACY_UPGRADE_MESSAGE = 'this ocm CLI is too old for this server; upgrade to ocm-cli >= 0.1.2 (the mirror upload protocol changed to chunked uploads)'
4652

53+
function gitRaw(repoPath: string, args: string[], env: NodeJS.ProcessEnv = process.env, input?: string): Promise<string> {
54+
return new Promise((resolve, reject) => {
55+
const child = spawn('git', args, { cwd: repoPath, env })
56+
let stdout = ''
57+
let stderr = ''
58+
child.stdout.on('data', (chunk: Buffer) => { stdout += chunk.toString() })
59+
child.stderr.on('data', (chunk: Buffer) => { stderr += chunk.toString() })
60+
child.on('error', reject)
61+
child.on('close', (code) => {
62+
if (code === 0) resolve(stdout)
63+
else reject(new Error(stderr.trim() || `git exited with code ${code}`))
64+
})
65+
if (input !== undefined) child.stdin.end(input)
66+
})
67+
}
68+
69+
async function createMirrorPatch(fullPath: string): Promise<string> {
70+
const untracked = (await gitRaw(fullPath, ['ls-files', '--others', '--exclude-standard', '-z']).catch(() => ''))
71+
.split('\0')
72+
.filter(Boolean)
73+
if (untracked.length === 0) return gitRaw(fullPath, ['diff', '--binary', 'HEAD', '--'])
74+
75+
const indexPath = (await safeGitOut(fullPath, ['rev-parse', '--git-path', 'index']))?.trim()
76+
const tempIndexDir = mkdtempSync(join(getReposPath(), '.ocm-index-'))
77+
const tempIndex = join(tempIndexDir, 'index')
78+
const env = { ...process.env, GIT_INDEX_FILE: tempIndex }
79+
80+
try {
81+
if (indexPath && existsSync(join(fullPath, indexPath))) {
82+
copyFileSync(join(fullPath, indexPath), tempIndex)
83+
}
84+
await gitRaw(fullPath, ['add', '-N', '--', ...untracked], env)
85+
return gitRaw(fullPath, ['diff', '--binary', 'HEAD', '--'], env)
86+
} finally {
87+
await fsp.rm(tempIndexDir, { recursive: true, force: true }).catch(() => {})
88+
}
89+
}
90+
91+
async function applyMirrorPatch(fullPath: string, patch: string): Promise<void> {
92+
if (!patch) return
93+
await gitRaw(fullPath, ['apply', '--binary', '--whitespace=nowarn', '-'], process.env, patch)
94+
}
95+
96+
async function importBundle(fullPath: string, bundlePath: string, branch: string | null): Promise<void> {
97+
await gitRaw(fullPath, ['fetch', bundlePath, '+refs/heads/*:refs/remotes/ocm-sync/*', '+refs/tags/*:refs/tags/*'])
98+
const refs = await gitRaw(fullPath, ['for-each-ref', '--format=%(refname:strip=3) %(objectname)', 'refs/remotes/ocm-sync'])
99+
for (const line of refs.split('\n')) {
100+
const trimmed = line.trim()
101+
if (!trimmed) continue
102+
const firstSpace = trimmed.indexOf(' ')
103+
if (firstSpace === -1) continue
104+
const name = trimmed.slice(0, firstSpace)
105+
const sha = trimmed.slice(firstSpace + 1)
106+
await gitRaw(fullPath, ['update-ref', `refs/heads/${name}`, sha])
107+
}
108+
109+
if (branch) {
110+
await gitRaw(fullPath, ['checkout', branch])
111+
const head = (await gitRaw(fullPath, ['rev-parse', `refs/remotes/ocm-sync/${branch}`])).trim()
112+
if (head) await gitRaw(fullPath, ['reset', '--hard', head])
113+
}
114+
115+
await gitRaw(fullPath, ['for-each-ref', '--format=%(refname)', 'refs/remotes/ocm-sync'])
116+
.then(async (out) => {
117+
for (const ref of out.split('\n').map((line) => line.trim()).filter(Boolean)) {
118+
await gitRaw(fullPath, ['update-ref', '-d', ref])
119+
}
120+
})
121+
.catch(() => {})
122+
}
123+
124+
async function createBundle(fullPath: string): Promise<string> {
125+
const stagingRoot = join(getReposPath(), '.ocm-staging')
126+
mkdirSync(stagingRoot, { recursive: true })
127+
const bundleDir = mkdtempSync(join(stagingRoot, 'bundle-'))
128+
const bundlePath = join(bundleDir, 'repo.bundle')
129+
await gitRaw(fullPath, ['bundle', 'create', bundlePath, '--all'])
130+
return bundlePath
131+
}
132+
47133
export function createInternalRepoMirrorRoutes(db: Database) {
48134
const app = new Hono()
49135

@@ -214,6 +300,146 @@ export function createInternalRepoMirrorRoutes(db: Database) {
214300
return c.json({ ok: true })
215301
})
216302

303+
app.get('/:repoId/mirror/bundle', async (c) => {
304+
const repoIdRaw = c.req.param('repoId')
305+
const repoId = Number(repoIdRaw)
306+
if (!Number.isFinite(repoId)) return c.json({ error: 'invalid repoId' }, 400)
307+
const repo = getRepoById(db, repoId)
308+
if (!repo) return c.json({ error: 'repo not found' }, 404)
309+
310+
let bundlePath: string | undefined
311+
try {
312+
bundlePath = await createBundle(repo.fullPath)
313+
const stream = createReadStream(bundlePath)
314+
stream.on('close', () => {
315+
if (bundlePath) fsp.rm(join(bundlePath, '..'), { recursive: true, force: true }).catch(() => {})
316+
})
317+
return new Response(Readable.toWeb(stream) as ReadableStream, {
318+
headers: { 'Content-Type': 'application/octet-stream' },
319+
})
320+
} catch (error) {
321+
logger.error('mirror bundle download failed:', error)
322+
if (bundlePath) await fsp.rm(join(bundlePath, '..'), { recursive: true, force: true }).catch(() => {})
323+
return c.json({ error: getErrorMessage(error) }, 500)
324+
}
325+
})
326+
327+
app.post('/:repoId/mirror/bundle', async (c) => {
328+
const repoIdRaw = c.req.param('repoId')
329+
const repoId = Number(repoIdRaw)
330+
if (!Number.isFinite(repoId)) return c.json({ error: 'invalid repoId' }, 400)
331+
const repo = getRepoById(db, repoId)
332+
if (!repo) return c.json({ error: 'repo not found' }, 404)
333+
if (isRepoInUse(db, repoId) && c.req.query('force') !== '1') {
334+
return c.json({ error: 'repo_in_use', message: 'open OpenCode sessions are using this repo; rerun with force=1' }, 409)
335+
}
336+
337+
const rawBody = c.req.raw.body
338+
if (!rawBody) return c.json({ error: 'no body provided' }, 400)
339+
340+
const stagingRoot = join(getReposPath(), '.ocm-staging')
341+
mkdirSync(stagingRoot, { recursive: true })
342+
const bundleDir = mkdtempSync(join(stagingRoot, 'bundle-upload-'))
343+
const bundlePath = join(bundleDir, 'repo.bundle')
344+
const branch = c.req.header('x-ocm-branch')?.trim() || null
345+
346+
try {
347+
const body = Readable.fromWeb(rawBody as unknown as Parameters<typeof Readable.fromWeb>[0])
348+
await pipeline(body, createWriteStream(bundlePath))
349+
await importBundle(repo.fullPath, bundlePath, branch)
350+
351+
const branchName = await safeGitOut(repo.fullPath, ['rev-parse', '--abbrev-ref', 'HEAD'])
352+
const head = await safeGitOut(repo.fullPath, ['rev-parse', 'HEAD'])
353+
if (branchName) updateRepoBranch(db, repoId, branchName.trim())
354+
updateLastPulled(db, repoId)
355+
356+
return c.json({
357+
repoId,
358+
fullPath: repo.fullPath,
359+
branch: branchName?.trim() || null,
360+
head: head?.trim() || null,
361+
created: false,
362+
})
363+
} catch (error) {
364+
logger.error('mirror bundle upload failed:', error)
365+
return c.json({ error: getErrorMessage(error) }, 409)
366+
} finally {
367+
await fsp.rm(bundleDir, { recursive: true, force: true }).catch(() => {})
368+
}
369+
})
370+
371+
app.get('/:repoId/mirror/patch', async (c) => {
372+
const repoIdRaw = c.req.param('repoId')
373+
const repoId = Number(repoIdRaw)
374+
if (!Number.isFinite(repoId)) return c.json({ error: 'invalid repoId' }, 400)
375+
const repo = getRepoById(db, repoId)
376+
if (!repo) return c.json({ error: 'repo not found' }, 404)
377+
378+
try {
379+
const branchName = await safeGitOut(repo.fullPath, ['rev-parse', '--abbrev-ref', 'HEAD'])
380+
const head = await safeGitOut(repo.fullPath, ['rev-parse', 'HEAD'])
381+
const patch = await createMirrorPatch(repo.fullPath)
382+
return c.json({
383+
repoId: repo.id,
384+
branch: branchName?.trim() || null,
385+
head: head?.trim() || null,
386+
patch,
387+
})
388+
} catch (error) {
389+
logger.error('mirror patch snapshot failed:', error)
390+
return c.json({ error: getErrorMessage(error) }, 500)
391+
}
392+
})
393+
394+
app.post('/:repoId/mirror/patch', async (c) => {
395+
const repoIdRaw = c.req.param('repoId')
396+
const repoId = Number(repoIdRaw)
397+
if (!Number.isFinite(repoId)) return c.json({ error: 'invalid repoId' }, 400)
398+
399+
let body: PatchBody
400+
try {
401+
body = (await c.req.json()) as PatchBody
402+
} catch {
403+
return c.json({ error: 'invalid json body' }, 400)
404+
}
405+
406+
const repo = getRepoById(db, repoId)
407+
if (!repo) return c.json({ error: 'repo not found' }, 404)
408+
if (!body.patch && body.patch !== '') return c.json({ error: 'patch required' }, 400)
409+
if (body.force !== true && isRepoInUse(db, repoId)) {
410+
return c.json({ error: 'repo_in_use', message: 'open OpenCode sessions are using this repo; rerun with force=1' }, 409)
411+
}
412+
413+
try {
414+
const currentHead = await safeGitOut(repo.fullPath, ['rev-parse', 'HEAD'])
415+
const currentHeadTrimmed = currentHead?.trim() || null
416+
const baseHead = body.baseHead?.trim() || null
417+
if (baseHead && currentHeadTrimmed && baseHead !== currentHeadTrimmed) {
418+
return c.json({ error: 'head_mismatch', message: 'Manager repo HEAD differs from patch base' }, 409)
419+
}
420+
421+
await applyMirrorPatch(repo.fullPath, body.patch)
422+
423+
const branchName = await safeGitOut(repo.fullPath, ['rev-parse', '--abbrev-ref', 'HEAD'])
424+
const head = await safeGitOut(repo.fullPath, ['rev-parse', 'HEAD'])
425+
426+
if (branchName) updateRepoBranch(db, repoId, branchName.trim())
427+
updateLastPulled(db, repoId)
428+
429+
return c.json({
430+
repoId,
431+
fullPath: repo.fullPath,
432+
branch: branchName?.trim() || null,
433+
head: head?.trim() || null,
434+
created: false,
435+
applied: true,
436+
})
437+
} catch (error) {
438+
logger.error('mirror patch failed:', error)
439+
return c.json({ error: getErrorMessage(error) }, 409)
440+
}
441+
})
442+
217443
app.get('/:repoId/mirror', async (c) => {
218444
const repoIdRaw = c.req.param('repoId')
219445
const repoId = Number(repoIdRaw)

‎backend/test/routes/internal/repo-mirror.test.ts‎

Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -238,6 +238,133 @@ describe('internal-repo-mirror routes', () => {
238238
})
239239
})
240240

241+
describe('patch sync flow', () => {
242+
it('imports a git bundle into an existing manager repo', async () => {
243+
const sourceDir = join(getTmpRoot(), 'bundle-source')
244+
mkdirSync(sourceDir, { recursive: true })
245+
spawnSync('git', ['init', '-b', 'main'], { cwd: sourceDir, stdio: 'ignore' })
246+
spawnSync('git', ['config', 'user.email', 'test@test.com'], { cwd: sourceDir, stdio: 'ignore' })
247+
spawnSync('git', ['config', 'user.name', 'Test'], { cwd: sourceDir, stdio: 'ignore' })
248+
writeFileSync(join(sourceDir, 'tracked.txt'), 'from bundle\n')
249+
spawnSync('git', ['add', 'tracked.txt'], { cwd: sourceDir, stdio: 'ignore' })
250+
spawnSync('git', ['commit', '-m', 'source'], { cwd: sourceDir, stdio: 'ignore' })
251+
spawnSync('git', ['checkout', '-b', 'feature'], { cwd: sourceDir, stdio: 'ignore' })
252+
writeFileSync(join(sourceDir, 'feature.txt'), 'feature branch\n')
253+
spawnSync('git', ['add', 'feature.txt'], { cwd: sourceDir, stdio: 'ignore' })
254+
spawnSync('git', ['commit', '-m', 'feature'], { cwd: sourceDir, stdio: 'ignore' })
255+
spawnSync('git', ['checkout', 'main'], { cwd: sourceDir, stdio: 'ignore' })
256+
const bundlePath = join(getTmpRoot(), 'source.bundle')
257+
spawnSync('git', ['bundle', 'create', bundlePath, '--all'], { cwd: sourceDir, stdio: 'ignore' })
258+
259+
const targetDir = join(getTmpRoot(), 'bundle-target')
260+
mkdirSync(targetDir, { recursive: true })
261+
spawnSync('git', ['init', '-b', 'main'], { cwd: targetDir, stdio: 'ignore' })
262+
spawnSync('git', ['config', 'user.email', 'test@test.com'], { cwd: targetDir, stdio: 'ignore' })
263+
spawnSync('git', ['config', 'user.name', 'Test'], { cwd: targetDir, stdio: 'ignore' })
264+
writeFileSync(join(targetDir, 'old.txt'), 'old\n')
265+
spawnSync('git', ['add', 'old.txt'], { cwd: targetDir, stdio: 'ignore' })
266+
spawnSync('git', ['commit', '-m', 'target'], { cwd: targetDir, stdio: 'ignore' })
267+
268+
mockGetRepoById.mockReturnValue({ id: 1, fullPath: targetDir })
269+
mockSafeGitOut.mockImplementation(async (_repoPath: string, args: string[]) => {
270+
if (args[0] === 'rev-parse' && args[1] === '--abbrev-ref') return 'main'
271+
if (args[0] === 'rev-parse' && args[1] === 'HEAD') return 'abc'
272+
return null
273+
})
274+
275+
const res = await app.request('/api/internal/repos/1/mirror/bundle', {
276+
method: 'POST',
277+
body: readFileSync(bundlePath),
278+
headers: { 'content-type': 'application/octet-stream', 'x-ocm-branch': 'main' },
279+
})
280+
281+
expect(res.status).toBe(200)
282+
expect(readFileSync(join(targetDir, 'tracked.txt'), 'utf-8')).toBe('from bundle\n')
283+
const featureRef = spawnSync('git', ['rev-parse', '--verify', 'refs/heads/feature'], { cwd: targetDir, encoding: 'utf-8' })
284+
expect(featureRef.status).toBe(0)
285+
})
286+
287+
it('returns a git bundle for pull fast path', async () => {
288+
const repoDir = join(getTmpRoot(), 'bundle-download')
289+
mkdirSync(repoDir, { recursive: true })
290+
spawnSync('git', ['init', '-b', 'main'], { cwd: repoDir, stdio: 'ignore' })
291+
spawnSync('git', ['config', 'user.email', 'test@test.com'], { cwd: repoDir, stdio: 'ignore' })
292+
spawnSync('git', ['config', 'user.name', 'Test'], { cwd: repoDir, stdio: 'ignore' })
293+
writeFileSync(join(repoDir, 'tracked.txt'), 'bundle payload\n')
294+
spawnSync('git', ['add', 'tracked.txt'], { cwd: repoDir, stdio: 'ignore' })
295+
spawnSync('git', ['commit', '-m', 'bundle'], { cwd: repoDir, stdio: 'ignore' })
296+
mockGetRepoById.mockReturnValue({ id: 1, fullPath: repoDir })
297+
298+
const res = await app.request('/api/internal/repos/1/mirror/bundle')
299+
const body = Buffer.from(await res.arrayBuffer())
300+
301+
expect(res.status).toBe(200)
302+
expect(body.length).toBeGreaterThan(0)
303+
const verifyPath = join(getTmpRoot(), 'download.bundle')
304+
writeFileSync(verifyPath, body)
305+
const verify = spawnSync('git', ['bundle', 'verify', verifyPath], { cwd: repoDir, encoding: 'utf-8' })
306+
expect(verify.status).toBe(0)
307+
})
308+
309+
it('applies a patch to an existing manager repo', async () => {
310+
const repoDir = join(getTmpRoot(), 'patch-repo')
311+
mkdirSync(repoDir, { recursive: true })
312+
spawnSync('git', ['init'], { cwd: repoDir, stdio: 'ignore' })
313+
spawnSync('git', ['config', 'user.email', 'test@test.com'], { cwd: repoDir, stdio: 'ignore' })
314+
spawnSync('git', ['config', 'user.name', 'Test'], { cwd: repoDir, stdio: 'ignore' })
315+
writeFileSync(join(repoDir, 'tracked.txt'), 'before\n')
316+
spawnSync('git', ['add', 'tracked.txt'], { cwd: repoDir, stdio: 'ignore' })
317+
spawnSync('git', ['commit', '-m', 'initial'], { cwd: repoDir, stdio: 'ignore' })
318+
const head = spawnSync('git', ['rev-parse', 'HEAD'], { cwd: repoDir, encoding: 'utf-8' }).stdout.trim()
319+
320+
mockGetRepoById.mockReturnValue({ id: 1, fullPath: repoDir })
321+
mockSafeGitOut.mockImplementation(async (_repoPath: string, args: string[]) => {
322+
if (args[0] === 'rev-parse' && args[1] === 'HEAD') return head
323+
if (args[0] === 'rev-parse' && args[1] === '--abbrev-ref') return 'main'
324+
return null
325+
})
326+
327+
const patch = 'diff --git a/tracked.txt b/tracked.txt\nindex 6e58d95..7c6cae9 100644\n--- a/tracked.txt\n+++ b/tracked.txt\n@@ -1 +1 @@\n-before\n+after\n'
328+
const res = await app.request('/api/internal/repos/1/mirror/patch', {
329+
method: 'POST',
330+
body: JSON.stringify({ baseHead: head, patch }),
331+
headers: { 'content-type': 'application/json' },
332+
})
333+
334+
expect(res.status).toBe(200)
335+
expect(readFileSync(join(repoDir, 'tracked.txt'), 'utf-8')).toBe('after\n')
336+
expect(mockUpdateLastPulled).toHaveBeenCalledWith(expect.anything(), 1)
337+
})
338+
339+
it('returns a patch snapshot for pull fast path', async () => {
340+
const repoDir = join(getTmpRoot(), 'snapshot-repo')
341+
mkdirSync(repoDir, { recursive: true })
342+
spawnSync('git', ['init'], { cwd: repoDir, stdio: 'ignore' })
343+
spawnSync('git', ['config', 'user.email', 'test@test.com'], { cwd: repoDir, stdio: 'ignore' })
344+
spawnSync('git', ['config', 'user.name', 'Test'], { cwd: repoDir, stdio: 'ignore' })
345+
writeFileSync(join(repoDir, 'tracked.txt'), 'before\n')
346+
spawnSync('git', ['add', 'tracked.txt'], { cwd: repoDir, stdio: 'ignore' })
347+
spawnSync('git', ['commit', '-m', 'initial'], { cwd: repoDir, stdio: 'ignore' })
348+
writeFileSync(join(repoDir, 'tracked.txt'), 'after\n')
349+
350+
mockGetRepoById.mockReturnValue({ id: 1, fullPath: repoDir })
351+
mockSafeGitOut.mockImplementation(async (_repoPath: string, args: string[]) => {
352+
if (args[0] === 'rev-parse' && args[1] === 'HEAD') return 'abc'
353+
if (args[0] === 'rev-parse' && args[1] === '--abbrev-ref') return 'main'
354+
if (args[0] === 'rev-parse' && args[1] === '--git-path') return '.git/index'
355+
return null
356+
})
357+
358+
const res = await app.request('/api/internal/repos/1/mirror/patch')
359+
const json = await res.json() as { patch: string; head: string; branch: string }
360+
361+
expect(res.status).toBe(200)
362+
expect(json.head).toBe('abc')
363+
expect(json.branch).toBe('main')
364+
expect(json.patch).toContain('diff --git a/tracked.txt b/tracked.txt')
365+
})
366+
})
367+
241368
describe('chunked upload flow (begin/parts/commit)', () => {
242369
it('creates a repo and populates from chunked tarball', async () => {
243370
const targetPath = join(getTmpRoot(), 'test-repo')

0 commit comments

Comments
 (0)