import { ok } from '@atcute/client'; import type { ResourceUri } from '@atcute/lexicons/syntax'; import type { OAuthUserAgent } from '@atcute/oauth-browser-client'; import { ShTangledRepoMerge, ShTangledRepoPull, ShTangledRepoPullComment, ShTangledRepoPullStatus, } from '@atcute/tangled'; import { gunzipSync, gzipSync, strFromU8, strToU8 } from 'fflate'; import { appviewFallback, appviewJson, getAppviewRecordCount, listAllAppviewRecords, type AppviewRecordView, type AppviewStatefulRecordView, } from './appview'; import { getBacklinks, getBacklinksPage, getLatestBacklinkTimestamp, latestBacklinkRefByRkey, listBacklinkedRecordsPage, } from './backlinks'; import { PULL_COLLECTION, PULL_COMMENT_COLLECTION, PULL_STATUS_COLLECTION, } from './constants'; import { createServiceAuthRpc, getRpc } from './client'; import { appviewRecordToHydrated, fetchRecordValue, hydrateBacklinks, hydrateBacklinksFromAppview, } from './hydration'; import { createAuthRpc, resolveActor } from './identity'; import { extractAtUris, extractBlobCid, extractDids, findNumberedRecord, type HydratedRecord, maxTimestamp, optionalString, type PaginatedResult, parseAtUri, parseDirectRecordRef, sortByCreatedAt, timestampFromDate, timestampFromTid, toNumberMap, type RecordCreationResult, type BacklinkRecordRef, } from './records'; import type { RepoContext } from './repos'; import { getAppviewRecordSortTimestamp, listAppviewStatefulRecordsPage, listSearchedStatefulRecordsPage, sortStatefulRecordsByNewestRecord, sortStatefulRecordsByUpdatedAt, type StatefulPageOptions, } from './stateful'; export interface PullSummary extends HydratedRecord { number: number; state: 'open' | 'closed' | 'merged'; } export interface PullComment extends HydratedRecord {} export interface PullDetail { pull: PullSummary; comments: PullComment[]; } export interface CreatePullInput { title: string; body: string; targetBranch: string; sourceBranch: string; patch: string; } export interface MergePullInput { pull: PullSummary; patch: string; } const normalizePullStatus = (value?: string): 'open' | 'closed' | 'merged' => { if (value === 'merged' || value?.endsWith('.merged')) { return 'merged'; } if (value === 'closed' || value?.endsWith('.closed')) { return 'closed'; } return 'open'; }; export const getPullRecord = async (pullUri: ResourceUri): Promise> => { const record = await appviewJson>( 'sh.tangled.repo.getPull', { pull: pullUri }, ); return appviewRecordToHydrated(record); }; const hydratePullBacklinks = (refs: BacklinkRecordRef[]): Promise>> => hydrateBacklinksFromAppview( refs, 'sh.tangled.repo.getPulls', 'pulls', ); const getLatestPullStatus = async (pullUri: ResourceUri): Promise<'open' | 'closed' | 'merged'> => { const refs = await getBacklinks(pullUri, 'sh.tangled.repo.pull.status:pull'); if (refs.length === 0) { return 'open'; } const status = await fetchRecordValue(latestBacklinkRefByRkey(refs)!); return normalizePullStatus(status.status); }; const getPullUpdatedAt = async ( pull: HydratedRecord, stateUpdatedAt?: string, ): Promise => maxTimestamp( timestampFromDate(pull.value.createdAt), timestampFromTid(pull.rkey), ...pull.value.rounds.map((round) => timestampFromDate(round.createdAt)), timestampFromDate(stateUpdatedAt), await getLatestBacklinkTimestamp(pull.uri, [ `${PULL_COMMENT_COLLECTION}:pull`, `${PULL_STATUS_COLLECTION}:pull`, ]), ); const listPullsFromBacklinks = async (repo: RepoContext): Promise => { const refs = await getBacklinks(repo.repoDid, 'sh.tangled.repo.pull:target.repo'); const pulls = await hydratePullBacklinks(refs); const numbers = toNumberMap(pulls); const states = await Promise.all(pulls.map((pull) => getLatestPullStatus(pull.uri))); const summaries = pulls .map((pull, index) => ({ ...pull, number: numbers.get(pull.uri) ?? index + 1, state: states[index], })); return sortStatefulRecordsByUpdatedAt(summaries, (pull) => getPullUpdatedAt(pull)); }; const listPullsPageFromBacklinks = async ( repo: RepoContext, options: StatefulPageOptions<'open' | 'closed' | 'merged'>, ): Promise> => listBacklinkedRecordsPage( repo.repoDid, 'sh.tangled.repo.pull:target.repo', options, getLatestPullStatus, (pull) => getPullUpdatedAt(pull), hydratePullBacklinks, ); const getPullCountFromBacklinks = async (repo: RepoContext): Promise => (await getBacklinksPage(repo.repoDid, 'sh.tangled.repo.pull:target.repo', 1)).total ?? (await getBacklinks(repo.repoDid, 'sh.tangled.repo.pull:target.repo')).length; const getPullFromBacklinks = async (repo: RepoContext, pullRef: string): Promise => { const pulls = await listPullsFromBacklinks(repo); const pull = findNumberedRecord(pulls, pullRef); if (!pull) { throw new Error(`Pull request not found`); } const commentRefs = await getBacklinks(pull.uri, 'sh.tangled.repo.pull.comment:pull'); const comments = await hydrateBacklinks(commentRefs); comments.sort(sortByCreatedAt); return { pull, comments, }; }; const listPullCommentsFromAppview = async (pullUri: ResourceUri): Promise => { const records = await listAllAppviewRecords( 'sh.tangled.repo.pull.listComments', pullUri, ); const comments = await Promise.all(records.map((record) => appviewRecordToHydrated(record))); comments.sort(sortByCreatedAt); return comments; }; const listPullsFromAppview = async (repo: RepoContext): Promise => { const records = await listAllAppviewRecords( 'sh.tangled.repo.listPulls', repo.repoDid, ) as Array>; const pulls = await Promise.all(records.map((record) => appviewRecordToHydrated(record))); const numbers = toNumberMap(pulls); const summaries = pulls .map((pull, index) => ({ ...pull, number: numbers.get(pull.uri) ?? index + 1, state: normalizePullStatus(records[index].state), })); return sortStatefulRecordsByNewestRecord(summaries); }; const listPullsPageFromAppview = async ( repo: RepoContext, options: StatefulPageOptions<'open' | 'closed' | 'merged'>, ): Promise> => { if (options.query?.trim()) { // Appview search can scope to repo/author/nsid. State is still derived // from separate status records, so filter state after each search hit. return listSearchedStatefulRecordsPage( 'sh.tangled.repo.pull', repo, options, getLatestPullStatus, ); } // Temporary until Bobbin exposes newest-first/reverse repo pull pagination. // This loads all appview pull records, but sorts without per-item backlinks. return listAppviewStatefulRecordsPage( 'sh.tangled.repo.listPulls', repo.repoDid, options, normalizePullStatus, options.author ? { author: options.author } : {}, (pull) => Promise.resolve(getAppviewRecordSortTimestamp(pull)), ); }; const getPullFromAppview = async (repo: RepoContext, pullRef: string): Promise => { const directUri = parseDirectRecordRef(pullRef, PULL_COLLECTION); if (directUri) { const pullRecord = await getPullRecord(directUri); if (pullRecord.value.target.repo !== repo.repoDid) { throw new Error(`Pull request not found`); } const pull: PullSummary = { ...pullRecord, number: 0, state: await getLatestPullStatus(pullRecord.uri), }; return { pull, comments: await listPullCommentsFromAppview(pull.uri), }; } const pulls = await listPullsFromAppview(repo); const pull = findNumberedRecord(pulls, pullRef); if (!pull) { throw new Error(`Pull request not found`); } return { pull, comments: await listPullCommentsFromAppview(pull.uri), }; }; export const listPulls = async (repo: RepoContext): Promise => appviewFallback( () => listPullsFromAppview(repo), () => listPullsFromBacklinks(repo), ); export const listPullsPage = async ( repo: RepoContext, options: StatefulPageOptions<'open' | 'closed' | 'merged'>, ): Promise> => appviewFallback( () => listPullsPageFromAppview(repo, options), () => listPullsPageFromBacklinks(repo, options), ); export const getRepoPullCount = async (repo: RepoContext): Promise => appviewFallback( () => getAppviewRecordCount('sh.tangled.repo.countPulls', repo.repoDid), () => getPullCountFromBacklinks(repo), ); export const getPull = async (repo: RepoContext, pullRef: string): Promise => appviewFallback( () => getPullFromAppview(repo, pullRef), () => getPullFromBacklinks(repo, pullRef), ); export const fetchPullRoundPatch = async ( pull: PullSummary, roundIndex: number, ): Promise => { const round = pull.value.rounds[roundIndex]; if (!round) { throw new Error(`Round ${roundIndex + 1} not found`); } const cid = extractBlobCid(round.patchBlob); if (!cid) { throw new Error(`Missing patch blob CID`); } const author = pull.author.pds ? pull.author : await resolveActor(pull.author.did); const rpc = getRpc(author.pds); const bytes = await ok( rpc.get('com.atproto.sync.getBlob', { params: { did: author.did, cid, }, as: 'bytes', }), ); return strFromU8(gunzipSync(bytes)); }; export const createPull = async ( agent: OAuthUserAgent, repo: RepoContext, input: CreatePullInput, ): Promise => { const rpc = createAuthRpc(agent); const patchBytes = new Uint8Array(gzipSync(strToU8(input.patch))); const patchBlob = new Blob([patchBytes.buffer], { type: 'application/gzip', }); const uploaded = await ok( rpc.post('com.atproto.repo.uploadBlob', { input: patchBlob, }), ); const record: ShTangledRepoPull.Main = { $type: 'sh.tangled.repo.pull', title: input.title.trim(), body: optionalString(input.body), target: { repo: repo.repoDid, branch: input.targetBranch, }, source: { branch: input.sourceBranch, }, rounds: [ { createdAt: new Date().toISOString(), patchBlob: uploaded.blob, }, ], createdAt: new Date().toISOString(), mentions: extractDids(input.body), references: extractAtUris(input.body), }; const result = await ok( rpc.post('com.atproto.repo.createRecord', { input: { repo: agent.sub, collection: PULL_COLLECTION, record, }, }), ); return { uri: result.uri, rkey: parseAtUri(result.uri).rkey, }; }; export const createPullComment = async ( agent: OAuthUserAgent, pullUri: ResourceUri, body: string, ): Promise => { const rpc = createAuthRpc(agent); const record: ShTangledRepoPullComment.Main = { $type: 'sh.tangled.repo.pull.comment', pull: pullUri, body: body.trim(), createdAt: new Date().toISOString(), mentions: extractDids(body), references: extractAtUris(body), }; const result = await ok( rpc.post('com.atproto.repo.createRecord', { input: { repo: agent.sub, collection: PULL_COMMENT_COLLECTION, record, }, }), ); return { uri: result.uri, rkey: parseAtUri(result.uri).rkey, }; }; export const setPullStatus = async ( agent: OAuthUserAgent, pullUri: ResourceUri, status: 'open' | 'closed' | 'merged', ): Promise => { const rpc = createAuthRpc(agent); const record: ShTangledRepoPullStatus.Main = { $type: 'sh.tangled.repo.pull.status', pull: pullUri, status: status === 'merged' ? 'sh.tangled.repo.pull.status.merged' : status === 'closed' ? 'sh.tangled.repo.pull.status.closed' : 'sh.tangled.repo.pull.status.open', }; const result = await ok( rpc.post('com.atproto.repo.createRecord', { input: { repo: agent.sub, collection: PULL_STATUS_COLLECTION, record, }, }), ); return { uri: result.uri, rkey: parseAtUri(result.uri).rkey, }; }; export const mergePull = async ( agent: OAuthUserAgent, repo: RepoContext, input: MergePullInput, ): Promise => { const rpc = await createServiceAuthRpc(agent, repo.knot, 'sh.tangled.repo.merge'); const pull = input.pull; const mergeInput: ShTangledRepoMerge.$input = { did: repo.owner.did, name: repo.slug, branch: pull.value.target.branch, patch: input.patch, commitMessage: pull.value.title, authorName: pull.author.handle, ...(pull.value.body ? { commitBody: pull.value.body } : {}), }; await ok( rpc.post('sh.tangled.repo.merge', { input: mergeInput, as: null, }), ); return setPullStatus(agent, pull.uri, 'merged'); };