feat(plugins): cache Git marketplace sources

This commit is contained in:
Neil
2026-07-13 03:33:44 -07:00
parent 4558e5d46c
commit dcd01a5b62
7 changed files with 876 additions and 33 deletions
+55
View File
@@ -0,0 +1,55 @@
import { execFile } from 'node:child_process'
import { promisify } from 'node:util'
import { isAllowedPluginGitUrl } from '../../shared/plugins/plugin-install-lockfile'
const execFileAsync = promisify(execFile)
const PLUGIN_GIT_TIMEOUT_MS = 120_000
const COMMIT_PATTERN = /^(?:[0-9a-f]{40}|[0-9a-f]{64})$/
/** Runs system Git with argv-only invocation so credential helpers and SSH
* remotes work without exposing an executable remote-helper surface. */
export async function runPluginGit(args: string[], cwd: string): Promise<string> {
const { stdout } = await execFileAsync('git', args, {
cwd,
timeout: PLUGIN_GIT_TIMEOUT_MS,
windowsHide: true,
env: {
...process.env,
// Existing non-interactive helpers and SSH agents still work, but a
// background marketplace refresh can never hang on a terminal prompt.
GIT_TERMINAL_PROMPT: '0'
}
})
return stdout.trim()
}
/** Checks out one Git ref into an empty destination and returns exact HEAD. */
export async function checkoutPluginGitSource(input: {
url: string
ref: string
destination: string
workingDirectory: string
}): Promise<string> {
if (!isAllowedPluginGitUrl(input.url)) {
throw new Error('plugin Git URL must use HTTPS or SSH')
}
const ref = input.ref.trim()
if (COMMIT_PATTERN.test(ref)) {
await runPluginGit(['init', '--quiet', input.destination], input.workingDirectory)
await runPluginGit(['remote', 'add', 'origin', input.url], input.destination)
await runPluginGit(['fetch', '--quiet', '--depth', '1', 'origin', ref], input.destination)
await runPluginGit(['checkout', '--quiet', 'FETCH_HEAD'], input.destination)
} else {
const args = ['clone', '--quiet', '--depth', '1']
if (ref.length > 0) {
args.push('--branch', ref)
}
args.push('--', input.url, input.destination)
await runPluginGit(args, input.workingDirectory)
}
const resolvedCommit = await runPluginGit(['rev-parse', 'HEAD'], input.destination)
if (!COMMIT_PATTERN.test(resolvedCommit)) {
throw new Error('Git resolved an invalid commit identity')
}
return resolvedCommit
}
+7 -33
View File
@@ -1,9 +1,7 @@
import { execFile } from 'node:child_process'
import { mkdtemp, realpath, rm } from 'node:fs/promises'
import { existsSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { isAbsolute, join, relative, resolve, sep } from 'node:path'
import { promisify } from 'node:util'
import {
PLUGIN_MANIFEST_FILENAME,
isQualifiedPluginKey
@@ -14,6 +12,7 @@ import {
} from '../../shared/plugins/plugin-install-lockfile'
import { readPluginLockfile, writePluginLockfile } from './plugin-install-lockfile-store'
import { installStagedPluginTree, type PluginInstallResult } from './plugin-install-staging'
import { checkoutPluginGitSource } from './plugin-git-repository'
export type { PluginInstallResult } from './plugin-install-staging'
@@ -34,8 +33,6 @@ export {
* swap; the previous version dir is kept for one-step rollback.
*/
const execFileAsync = promisify(execFile)
const GIT_TIMEOUT_MS = 120_000
const pluginMutationChains = new Map<string, Promise<void>>()
async function serializePluginMutation<T>(
@@ -77,20 +74,6 @@ export async function installPluginFromLocalPath(input: {
})
}
async function runGit(args: string[], cwd: string): Promise<string> {
const { stdout } = await execFileAsync('git', args, {
cwd,
timeout: GIT_TIMEOUT_MS,
windowsHide: true,
env: {
...process.env,
// Never block an install on an interactive credential/host prompt.
GIT_TERMINAL_PROMPT: '0'
}
})
return stdout.trim()
}
export async function installPluginFromGit(input: {
pluginsDir: string
url: string
@@ -105,21 +88,12 @@ export async function installPluginFromGit(input: {
const stagingDir = await mkdtemp(join(tmpdir(), 'orca-plugin-install-'))
try {
const ref = input.ref.trim()
if (/^(?:[0-9a-f]{40}|[0-9a-f]{64})$/.test(ref)) {
// Exact commit: shallow-fetch just that object.
await runGit(['init', '--quiet', stagingDir], tmpdir())
await runGit(['remote', 'add', 'origin', input.url], stagingDir)
await runGit(['fetch', '--quiet', '--depth', '1', 'origin', ref], stagingDir)
await runGit(['checkout', '--quiet', 'FETCH_HEAD'], stagingDir)
} else {
const args = ['clone', '--quiet', '--depth', '1']
if (ref.length > 0) {
args.push('--branch', ref)
}
args.push('--', input.url, stagingDir)
await runGit(args, tmpdir())
}
const resolvedCommit = await runGit(['rev-parse', 'HEAD'], stagingDir)
const resolvedCommit = await checkoutPluginGitSource({
url: input.url,
ref,
destination: stagingDir,
workingDirectory: tmpdir()
})
return await installStagedPluginTree({
pluginsDir: input.pluginsDir,
stagingDir,
@@ -0,0 +1,63 @@
import { createReadStream } from 'node:fs'
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import {
PLUGIN_MARKETPLACE_FILENAME,
pluginMarketplaceSchema,
type PluginMarketplace
} from '../../shared/plugins/plugin-marketplace'
import { checkoutPluginGitSource } from './plugin-git-repository'
import type { PluginMarketplaceRegisteredSource } from './plugin-marketplace-store'
const MARKETPLACE_INDEX_MAX_BYTES = 16 * 1024 * 1024
export type PluginMarketplaceFetchResult = {
marketplaceCommit: string
marketplace: PluginMarketplace
}
/** Fetches a marketplace through system Git so private repositories use the
* same SSH agent and credential helpers as every other Orca Git operation. */
export async function fetchPluginMarketplace(
source: PluginMarketplaceRegisteredSource
): Promise<PluginMarketplaceFetchResult> {
const stagingDirectory = await mkdtemp(join(tmpdir(), 'orca-plugin-marketplace-'))
try {
const marketplaceCommit = await checkoutPluginGitSource({
url: source.source.url,
ref: source.source.ref,
destination: stagingDirectory,
workingDirectory: tmpdir()
})
const marketplace = await readPluginMarketplaceIndex(stagingDirectory)
return { marketplaceCommit, marketplace }
} finally {
await rm(stagingDirectory, { recursive: true, force: true })
}
}
export async function readPluginMarketplaceIndex(
rootDirectory: string
): Promise<PluginMarketplace> {
const path = join(rootDirectory, PLUGIN_MARKETPLACE_FILENAME)
const chunks: Buffer[] = []
let totalBytes = 0
for await (const chunk of createReadStream(path)) {
const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)
totalBytes += bytes.byteLength
if (totalBytes > MARKETPLACE_INDEX_MAX_BYTES) {
throw new Error(`${PLUGIN_MARKETPLACE_FILENAME} exceeds its size limit`)
}
chunks.push(bytes)
}
try {
return pluginMarketplaceSchema.parse(
JSON.parse(Buffer.concat(chunks, totalBytes).toString('utf8'))
)
} catch (error) {
throw new Error(
`invalid ${PLUGIN_MARKETPLACE_FILENAME}: ${error instanceof Error ? error.message : String(error)}`
)
}
}
@@ -0,0 +1,183 @@
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, describe, expect, it, vi } from 'vitest'
import type {
PluginMarketplace,
PluginMarketplaceGitSource
} from '../../shared/plugins/plugin-marketplace'
import type { PluginMarketplaceFetchResult } from './plugin-marketplace-fetch'
import { PluginMarketplaceService } from './plugin-marketplace-service'
import type { PluginMarketplaceRegisteredSource } from './plugin-marketplace-store'
const roots: string[] = []
async function tempRoot(): Promise<string> {
const root = await mkdtemp(join(tmpdir(), 'orca-marketplace-service-'))
roots.push(root)
return root
}
function source(url = 'https://github.com/community/plugins.git'): PluginMarketplaceGitSource {
return { kind: 'git', url, ref: 'main' }
}
function marketplace(
name = 'Community',
pluginKey = 'community.theme',
pluginUrl = 'https://github.com/community/theme.git'
): PluginMarketplace {
return {
name,
owner: name.toLowerCase(),
plugins: [
{
id: pluginKey,
source: { kind: 'git', url: pluginUrl, ref: 'v1' },
description: 'A theme',
categories: ['themes']
}
]
}
}
function fetched(value = marketplace(), commit = 'a'.repeat(40)): PluginMarketplaceFetchResult {
return { marketplaceCommit: commit, marketplace: value }
}
afterEach(async () => {
await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })))
})
describe('PluginMarketplaceService', () => {
it('fetches before registration, then serves browse data from the local snapshot', async () => {
const fetcher = vi
.fn<
(registration: PluginMarketplaceRegisteredSource) => Promise<PluginMarketplaceFetchResult>
>()
.mockResolvedValue(fetched())
const service = new PluginMarketplaceService({ pluginsDataDir: await tempRoot(), fetcher })
const added = await service.addSource(source())
expect(added).toMatchObject({ marketplace: { name: 'Community' }, stale: false })
expect(fetcher).toHaveBeenCalledTimes(1)
await expect(service.listPlugins()).resolves.toEqual([
expect.objectContaining({
marketplaceSourceId: added.id,
pluginKey: 'community.theme',
official: false,
bundled: false
})
])
await service.listSources()
expect(fetcher).toHaveBeenCalledTimes(1)
})
it('keeps and labels the last valid snapshot when a refresh is offline', async () => {
const fetcher = vi
.fn<
(registration: PluginMarketplaceRegisteredSource) => Promise<PluginMarketplaceFetchResult>
>()
.mockResolvedValueOnce(fetched())
.mockRejectedValueOnce(new Error('offline'))
const service = new PluginMarketplaceService({ pluginsDataDir: await tempRoot(), fetcher })
const added = await service.addSource(source())
await expect(service.refreshSource(added.id)).resolves.toMatchObject({
marketplace: { name: 'Community', resolvedCommit: 'a'.repeat(40) },
stale: true,
error: 'offline'
})
await expect(service.listPlugins()).resolves.toEqual([
expect.objectContaining({ pluginKey: 'community.theme' })
])
})
it('atomically replaces the cache only after a valid refreshed index', async () => {
const fetcher = vi
.fn<
(registration: PluginMarketplaceRegisteredSource) => Promise<PluginMarketplaceFetchResult>
>()
.mockResolvedValueOnce(fetched())
.mockResolvedValueOnce(fetched(marketplace('Updated'), 'b'.repeat(40)))
const root = await tempRoot()
const service = new PluginMarketplaceService({ pluginsDataDir: root, fetcher })
const added = await service.addSource(source())
await expect(service.refreshSource(added.id)).resolves.toMatchObject({
marketplace: { name: 'Updated', resolvedCommit: 'b'.repeat(40) },
stale: false
})
await expect(
new PluginMarketplaceService({ pluginsDataDir: root, fetcher }).listPlugins()
).resolves.toEqual([expect.objectContaining({ marketplaceName: 'Updated' })])
})
it('does not persist a source whose first fetch fails', async () => {
const service = new PluginMarketplaceService({
pluginsDataDir: await tempRoot(),
fetcher: async () => {
throw new Error('authentication failed')
}
})
await expect(service.addSource(source())).rejects.toThrow('authentication failed')
await expect(service.listSources()).resolves.toEqual([])
})
it('rejects reserved identities outside the official organization', async () => {
const service = new PluginMarketplaceService({
pluginsDataDir: await tempRoot(),
fetcher: async () =>
fetched(
marketplace('Attack', 'community.orca-secrets', 'https://github.com/attacker/x.git')
)
})
await expect(service.addSource(source())).rejects.toThrow(
'reserved plugin identity community.orca-secrets'
)
await expect(service.listSources()).resolves.toEqual([])
})
it('derives the Official badge only from the canonical marketplace and source organization', async () => {
const officialMarketplace: PluginMarketplace = {
name: 'Orca Plugins',
owner: 'stablyai',
plugins: [
{
id: 'stablyai.orca-skills',
source: {
kind: 'git',
url: 'git@github.com:stablyai/orca-skills.git',
ref: 'main'
},
categories: ['skills']
}
]
}
const service = new PluginMarketplaceService({
pluginsDataDir: await tempRoot(),
fetcher: async () => fetched(officialMarketplace)
})
await service.addSource(source('https://github.com/stablyai/orca-plugins.git'))
await expect(service.listPlugins()).resolves.toEqual([
expect.objectContaining({ pluginKey: 'stablyai.orca-skills', official: true })
])
})
it('removes source metadata and browse listings together', async () => {
const service = new PluginMarketplaceService({
pluginsDataDir: await tempRoot(),
fetcher: async () => fetched()
})
const added = await service.addSource(source())
await expect(service.removeSource(added.id)).resolves.toBe(true)
await expect(service.listSources()).resolves.toEqual([])
await expect(service.listPlugins()).resolves.toEqual([])
})
})
@@ -0,0 +1,267 @@
import {
OFFICIAL_MARKETPLACE_OWNER,
isOfficialMarketplaceGitSource,
isOfficialOrganizationGitSource,
isOfficialPluginIdentity,
isReservedPluginIdentity,
pluginMarketplaceGitSourceSchema,
type PluginMarketplaceEntry,
type PluginMarketplaceGitSource
} from '../../shared/plugins/plugin-marketplace'
import {
fetchPluginMarketplace,
type PluginMarketplaceFetchResult
} from './plugin-marketplace-fetch'
import {
marketplaceSourceId,
PluginMarketplaceStore,
type PluginMarketplaceCachedSnapshot,
type PluginMarketplaceRegisteredSource
} from './plugin-marketplace-store'
export type PluginMarketplaceSourceState = {
id: string
source: PluginMarketplaceGitSource
addedAt: number
marketplace: {
name: string
owner: string
resolvedCommit: string
fetchedAt: number
} | null
stale: boolean
error?: string
}
export type PluginMarketplaceListing = {
marketplaceSourceId: string
marketplaceName: string
marketplaceOwner: string
marketplaceCommit: string
pluginKey: string
source: PluginMarketplaceEntry['source']
description?: string
categories: string[]
official: boolean
bundled: boolean
}
type MarketplaceFetcher = (
source: PluginMarketplaceRegisteredSource
) => Promise<PluginMarketplaceFetchResult>
export class PluginMarketplaceService {
private readonly store: PluginMarketplaceStore
private readonly fetcher: MarketplaceFetcher
private readonly refreshChains = new Map<string, Promise<PluginMarketplaceSourceState>>()
constructor(options: {
pluginsDataDir: string
fetcher?: MarketplaceFetcher
store?: PluginMarketplaceStore
}) {
this.store = options.store ?? new PluginMarketplaceStore(options.pluginsDataDir)
this.fetcher = options.fetcher ?? fetchPluginMarketplace
}
async listSources(): Promise<PluginMarketplaceSourceState[]> {
const sources = await this.store.listSources()
return Promise.all(
sources.map(async (source) => {
try {
return this.stateFromSnapshot(source, await this.store.readSnapshot(source.id), false)
} catch (error) {
return this.stateFromSnapshot(source, null, true, errorMessage(error))
}
})
)
}
async addSource(source: PluginMarketplaceGitSource): Promise<PluginMarketplaceSourceState> {
const parsedSource = pluginMarketplaceGitSourceSchema.parse(source)
const sourceId = marketplaceSourceId(parsedSource)
const existing = (await this.store.listSources()).find((candidate) => candidate.id === sourceId)
const candidate: PluginMarketplaceRegisteredSource = existing ?? {
id: sourceId,
source: parsedSource,
addedAt: Date.now()
}
const fetched = await this.fetchAndValidate(candidate)
const registered = existing ?? (await this.store.addSource(parsedSource, candidate.addedAt))
try {
const snapshot = await this.store.writeSnapshot({ source: registered, ...fetched })
return this.stateFromSnapshot(registered, snapshot, false)
} catch (error) {
if (!existing) {
await this.store.removeSource(registered.id).catch(() => undefined)
}
throw error
}
}
async removeSource(sourceId: string): Promise<boolean> {
return this.store.removeSource(sourceId)
}
async refreshSource(sourceId: string): Promise<PluginMarketplaceSourceState> {
const previous = this.refreshChains.get(sourceId) ?? Promise.resolve(null)
const refresh = previous.catch(() => null).then(() => this.performRefresh(sourceId))
this.refreshChains.set(sourceId, refresh)
try {
return await refresh
} finally {
if (this.refreshChains.get(sourceId) === refresh) {
this.refreshChains.delete(sourceId)
}
}
}
async refreshAll(): Promise<PluginMarketplaceSourceState[]> {
const sources = await this.store.listSources()
return Promise.all(sources.map((source) => this.refreshSource(source.id)))
}
async listPlugins(): Promise<PluginMarketplaceListing[]> {
const states = await this.listSnapshots()
return states
.flatMap(({ source, snapshot }) =>
snapshot.marketplace.plugins.map((entry) => this.listingFromEntry(source, snapshot, entry))
)
.sort((left, right) =>
`${left.pluginKey}\0${left.marketplaceSourceId}`.localeCompare(
`${right.pluginKey}\0${right.marketplaceSourceId}`
)
)
}
async findPlugin(
marketplaceSourceId: string,
pluginKey: string
): Promise<PluginMarketplaceListing | null> {
const source = (await this.store.listSources()).find(
(candidate) => candidate.id === marketplaceSourceId
)
if (!source) {
return null
}
const snapshot = await this.store.readSnapshot(source.id)
const entry = snapshot?.marketplace.plugins.find((plugin) => plugin.id === pluginKey)
return snapshot && entry ? this.listingFromEntry(source, snapshot, entry) : null
}
private async performRefresh(sourceId: string): Promise<PluginMarketplaceSourceState> {
const source = (await this.store.listSources()).find((candidate) => candidate.id === sourceId)
if (!source) {
throw new Error(`unknown marketplace source: ${sourceId}`)
}
try {
const fetched = await this.fetchAndValidate(source)
const snapshot = await this.store.writeSnapshot({ source, ...fetched })
return this.stateFromSnapshot(source, snapshot, false)
} catch (error) {
const cached = await this.store.readSnapshot(source.id).catch(() => null)
if (!cached) {
throw error
}
return this.stateFromSnapshot(source, cached, true, errorMessage(error))
}
}
private async fetchAndValidate(
source: PluginMarketplaceRegisteredSource
): Promise<PluginMarketplaceFetchResult> {
const fetched = await this.fetcher(source)
validateMarketplaceProvenance(source, fetched)
return fetched
}
private async listSnapshots(): Promise<
{ source: PluginMarketplaceRegisteredSource; snapshot: PluginMarketplaceCachedSnapshot }[]
> {
const sources = await this.store.listSources()
const snapshots = await Promise.all(
sources.map(async (source) => ({
source,
snapshot: await this.store.readSnapshot(source.id)
}))
)
return snapshots.filter(
(
candidate
): candidate is {
source: PluginMarketplaceRegisteredSource
snapshot: PluginMarketplaceCachedSnapshot
} => candidate.snapshot !== null
)
}
private listingFromEntry(
source: PluginMarketplaceRegisteredSource,
snapshot: PluginMarketplaceCachedSnapshot,
entry: PluginMarketplaceEntry
): PluginMarketplaceListing {
const official =
isOfficialMarketplaceGitSource(source.source.url) &&
snapshot.marketplace.owner.toLowerCase() === OFFICIAL_MARKETPLACE_OWNER &&
isOfficialPluginIdentity(entry.id) &&
isOfficialOrganizationGitSource(entry.source.url)
return {
marketplaceSourceId: source.id,
marketplaceName: snapshot.marketplace.name,
marketplaceOwner: snapshot.marketplace.owner,
marketplaceCommit: snapshot.marketplaceCommit,
pluginKey: entry.id,
source: entry.source,
...(entry.description ? { description: entry.description } : {}),
categories: entry.categories,
official,
bundled: false
}
}
private stateFromSnapshot(
source: PluginMarketplaceRegisteredSource,
snapshot: PluginMarketplaceCachedSnapshot | null,
stale: boolean,
error?: string
): PluginMarketplaceSourceState {
return {
id: source.id,
source: source.source,
addedAt: source.addedAt,
marketplace: snapshot
? {
name: snapshot.marketplace.name,
owner: snapshot.marketplace.owner,
resolvedCommit: snapshot.marketplaceCommit,
fetchedAt: snapshot.fetchedAt
}
: null,
stale,
...(error ? { error } : {})
}
}
}
function validateMarketplaceProvenance(
source: PluginMarketplaceRegisteredSource,
fetched: PluginMarketplaceFetchResult
): void {
if (
isOfficialMarketplaceGitSource(source.source.url) &&
fetched.marketplace.owner.toLowerCase() !== OFFICIAL_MARKETPLACE_OWNER
) {
throw new Error('official marketplace metadata has an unexpected owner')
}
for (const entry of fetched.marketplace.plugins) {
if (isReservedPluginIdentity(entry.id) && !isOfficialOrganizationGitSource(entry.source.url)) {
throw new Error(
`reserved plugin identity ${entry.id} must resolve to the stablyai organization`
)
}
}
}
function errorMessage(error: unknown): string {
return error instanceof Error ? error.message : String(error)
}
@@ -0,0 +1,85 @@
import { mkdtemp, readdir, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, describe, expect, it } from 'vitest'
import type { PluginMarketplaceGitSource } from '../../shared/plugins/plugin-marketplace'
import { marketplaceSourceId, PluginMarketplaceStore } from './plugin-marketplace-store'
const roots: string[] = []
async function tempRoot(): Promise<string> {
const root = await mkdtemp(join(tmpdir(), 'orca-marketplace-store-'))
roots.push(root)
return root
}
function source(ref = 'main'): PluginMarketplaceGitSource {
return { kind: 'git', url: 'https://github.com/community/plugins.git', ref }
}
afterEach(async () => {
await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })))
})
describe('PluginMarketplaceStore', () => {
it('persists bounded source registrations under a deterministic opaque id', async () => {
const root = await tempRoot()
const first = new PluginMarketplaceStore(root)
const registered = await first.addSource(source(), 123)
expect(registered).toEqual({
id: marketplaceSourceId(source()),
source: source(),
addedAt: 123
})
await expect(new PluginMarketplaceStore(root).listSources()).resolves.toEqual([registered])
await expect(first.addSource(source(), 999)).resolves.toEqual(registered)
})
it('atomically publishes a strict cached snapshot and removes it with its source', async () => {
const root = await tempRoot()
const store = new PluginMarketplaceStore(root)
const registered = await store.addSource(source(), 123)
const snapshot = await store.writeSnapshot({
source: registered,
marketplaceCommit: 'a'.repeat(40),
fetchedAt: 456,
marketplace: {
name: 'Community',
owner: 'community',
plugins: [
{
id: 'community.theme',
source: {
kind: 'git',
url: 'https://github.com/community/theme.git',
ref: 'v1'
},
categories: ['themes']
}
]
}
})
await expect(new PluginMarketplaceStore(root).readSnapshot(registered.id)).resolves.toEqual(
snapshot
)
expect(
(await readdir(join(root, 'marketplaces', 'snapshots'))).filter((entry) =>
entry.endsWith('.tmp')
)
).toEqual([])
await expect(store.removeSource(registered.id)).resolves.toBe(true)
await expect(store.readSnapshot(registered.id)).resolves.toBeNull()
})
it('keeps distinct refs as distinct marketplace sources', () => {
expect(marketplaceSourceId(source('main'))).not.toBe(marketplaceSourceId(source('stable')))
})
it('rejects path-like source ids before reading or deleting cache files', async () => {
const store = new PluginMarketplaceStore(await tempRoot())
await expect(store.readSnapshot('../outside')).rejects.toThrow()
await expect(store.removeSource('../outside')).rejects.toThrow()
})
})
@@ -0,0 +1,216 @@
import { createHash, randomUUID } from 'node:crypto'
import { createReadStream } from 'node:fs'
import { mkdir, rename, rm, writeFile } from 'node:fs/promises'
import { dirname, join } from 'node:path'
import { z } from 'zod'
import {
pluginMarketplaceGitSourceSchema,
pluginMarketplaceSchema,
type PluginMarketplace,
type PluginMarketplaceGitSource
} from '../../shared/plugins/plugin-marketplace'
export const PLUGIN_MARKETPLACE_SOURCE_LIMIT = 64
export const PLUGIN_MARKETPLACE_SOURCE_ID_PATTERN = /^[0-9a-f]{32}$/
const sourceIdSchema = z.string().regex(PLUGIN_MARKETPLACE_SOURCE_ID_PATTERN)
const registeredSourceSchema = z.strictObject({
id: sourceIdSchema,
source: pluginMarketplaceGitSourceSchema,
addedAt: z.number().finite().nonnegative()
})
const sourceFileSchema = z.strictObject({
schemaVersion: z.literal(1),
sources: z.array(registeredSourceSchema).max(PLUGIN_MARKETPLACE_SOURCE_LIMIT)
})
const cachedSnapshotSchema = z.strictObject({
schemaVersion: z.literal(1),
sourceId: sourceIdSchema,
source: pluginMarketplaceGitSourceSchema,
marketplaceCommit: z.string().regex(/^(?:[0-9a-f]{40}|[0-9a-f]{64})$/),
fetchedAt: z.number().finite().nonnegative(),
marketplace: pluginMarketplaceSchema
})
export type PluginMarketplaceRegisteredSource = z.infer<typeof registeredSourceSchema>
export type PluginMarketplaceCachedSnapshot = z.infer<typeof cachedSnapshotSchema>
const SOURCE_FILE_MAX_BYTES = 2 * 1024 * 1024
const SNAPSHOT_FILE_MAX_BYTES = 16 * 1024 * 1024
export function marketplaceSourceId(source: PluginMarketplaceGitSource): string {
const parsed = pluginMarketplaceGitSourceSchema.parse(source)
return createHash('sha256')
.update(`orca-plugin-marketplace-source-v1\0${parsed.url}\0${parsed.ref}`)
.digest('hex')
.slice(0, 32)
}
export class PluginMarketplaceStore {
private readonly sourcesPath: string
private readonly snapshotDirectory: string
private sources: PluginMarketplaceRegisteredSource[] | null = null
private writeChain: Promise<void> = Promise.resolve()
constructor(pluginsDataDir: string) {
const root = join(pluginsDataDir, 'marketplaces')
this.sourcesPath = join(root, 'sources.json')
this.snapshotDirectory = join(root, 'snapshots')
}
async listSources(): Promise<readonly PluginMarketplaceRegisteredSource[]> {
await this.loadSources()
return this.sources!
}
async addSource(
source: PluginMarketplaceGitSource,
addedAt = Date.now()
): Promise<PluginMarketplaceRegisteredSource> {
const parsedSource = pluginMarketplaceGitSourceSchema.parse(source)
const registration = registeredSourceSchema.parse({
id: marketplaceSourceId(parsedSource),
source: parsedSource,
addedAt
})
await this.mutateSources((sources) => {
if (sources.some((candidate) => candidate.id === registration.id)) {
return [...sources]
}
if (sources.length >= PLUGIN_MARKETPLACE_SOURCE_LIMIT) {
throw new Error(`marketplace source limit (${PLUGIN_MARKETPLACE_SOURCE_LIMIT}) reached`)
}
return [...sources, registration]
})
return this.sources!.find((candidate) => candidate.id === registration.id)!
}
async removeSource(sourceId: string): Promise<boolean> {
const parsedId = sourceIdSchema.parse(sourceId)
let removed = false
await this.mutateSources((sources) => {
const next = sources.filter((source) => source.id !== parsedId)
removed = next.length !== sources.length
return next
})
if (removed) {
await rm(this.snapshotPath(parsedId), { force: true })
}
return removed
}
async writeSnapshot(input: {
source: PluginMarketplaceRegisteredSource
marketplaceCommit: string
fetchedAt?: number
marketplace: PluginMarketplace
}): Promise<PluginMarketplaceCachedSnapshot> {
const snapshot = cachedSnapshotSchema.parse({
schemaVersion: 1,
sourceId: input.source.id,
source: input.source.source,
marketplaceCommit: input.marketplaceCommit,
fetchedAt: input.fetchedAt ?? Date.now(),
marketplace: input.marketplace
})
await writeAtomicJson(this.snapshotPath(input.source.id), snapshot)
return snapshot
}
async readSnapshot(sourceId: string): Promise<PluginMarketplaceCachedSnapshot | null> {
const parsedId = sourceIdSchema.parse(sourceId)
try {
const raw = JSON.parse(
await readBoundedText(this.snapshotPath(parsedId), SNAPSHOT_FILE_MAX_BYTES)
)
const parsed = cachedSnapshotSchema.parse(raw)
if (parsed.sourceId !== parsedId) {
throw new Error('marketplace snapshot source identity does not match its cache path')
}
return parsed
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
return null
}
throw new Error(
`marketplace snapshot is invalid: ${error instanceof Error ? error.message : String(error)}`
)
}
}
private async loadSources(): Promise<void> {
if (this.sources !== null) {
return
}
try {
const parsed = sourceFileSchema.parse(
JSON.parse(await readBoundedText(this.sourcesPath, SOURCE_FILE_MAX_BYTES))
)
const ids = new Set<string>()
for (const source of parsed.sources) {
if (source.id !== marketplaceSourceId(source.source) || ids.has(source.id)) {
throw new Error('marketplace source identity is inconsistent or duplicated')
}
ids.add(source.id)
}
this.sources = parsed.sources
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
this.sources = []
return
}
throw new Error(
`marketplace sources are invalid: ${error instanceof Error ? error.message : String(error)}`
)
}
}
private async mutateSources(
mutation: (
sources: readonly PluginMarketplaceRegisteredSource[]
) => PluginMarketplaceRegisteredSource[]
): Promise<void> {
const update = this.writeChain
.catch(() => undefined)
.then(async () => {
await this.loadSources()
const next = sourceFileSchema.parse({
schemaVersion: 1,
sources: mutation(this.sources!)
}).sources
await writeAtomicJson(this.sourcesPath, { schemaVersion: 1, sources: next })
this.sources = next
})
this.writeChain = update
await update
}
private snapshotPath(sourceId: string): string {
return join(this.snapshotDirectory, `${sourceId}.json`)
}
}
async function readBoundedText(path: string, limit: number): Promise<string> {
const chunks: Buffer[] = []
let totalBytes = 0
for await (const chunk of createReadStream(path)) {
const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)
totalBytes += bytes.byteLength
if (totalBytes > limit) {
throw new Error(`file exceeds its ${limit}-byte limit`)
}
chunks.push(bytes)
}
return Buffer.concat(chunks, totalBytes).toString('utf8')
}
async function writeAtomicJson(path: string, value: unknown): Promise<void> {
await mkdir(dirname(path), { recursive: true })
const temporary = `${path}.${randomUUID()}.tmp`
try {
await writeFile(temporary, `${JSON.stringify(value, null, 2)}\n`, 'utf8')
await rename(temporary, path)
} finally {
await rm(temporary, { force: true }).catch(() => undefined)
}
}