import { ok } from '@atcute/client';
import type { Did, 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,
FEED_COMMENT_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<ShTangledRepoPull.Main> {
number: number;
state: 'open' | 'closed' | 'merged';
}
export interface PullComment extends HydratedRecord<ShTangledRepoPullComment.Main> {}
export interface PullDetail {
pull: PullSummary;
comments: PullComment[];
}
export interface CreatePullInput {
title: string;
body: string;
targetBranch: string;
sourceBranch: string;
patch: string;
sourceRepoDid?: Did;
}
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<HydratedRecord<ShTangledRepoPull.Main>> => {
const record = await appviewJson<AppviewRecordView<ShTangledRepoPull.Main>>(
'sh.tangled.repo.getPull',
{ pull: pullUri },
);
return appviewRecordToHydrated(record);
};
const hydratePullBacklinks = (refs: BacklinkRecordRef[]): Promise<Array<HydratedRecord<ShTangledRepoPull.Main>>> =>
hydrateBacklinksFromAppview<ShTangledRepoPull.Main>(
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<ShTangledRepoPullStatus.Main>(latestBacklinkRefByRkey(refs)!);
return normalizePullStatus(status.status);
};
const getPullUpdatedAt = async (
pull: HydratedRecord<ShTangledRepoPull.Main>,
stateUpdatedAt?: string,
): Promise<number> =>
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<PullSummary[]> => {
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<PaginatedResult<PullSummary>> =>
listBacklinkedRecordsPage<ShTangledRepoPull.Main, 'open' | 'closed' | 'merged'>(
repo.repoDid,
'sh.tangled.repo.pull:target.repo',
options,
getLatestPullStatus,
(pull) => getPullUpdatedAt(pull),
hydratePullBacklinks,
);
const getPullCountFromBacklinks = async (repo: RepoContext): Promise<number> =>
(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<PullDetail> => {
const pulls = await listPullsFromBacklinks(repo);
const pull = findNumberedRecord(pulls, pullRef);
if (!pull) {
throw new Error(`Pull request not found`);
}
const [newCommentRefs, legacyCommentRefs] = await Promise.all([
getBacklinks(pull.uri, 'sh.tangled.feed.comment:subject'),
getBacklinks(pull.uri, 'sh.tangled.repo.pull.comment:pull'),
]);
const commentRefs = [...newCommentRefs];
const seenUris = new Set(commentRefs.map(r => `at://${r.did}/${r.collection}/${r.rkey}`));
for (const ref of legacyCommentRefs) {
const refUri = `at://${ref.did}/${ref.collection}/${ref.rkey}`;
if (!seenUris.has(refUri)) {
commentRefs.push(ref);
seenUris.add(refUri);
}
}
const hydrated = await hydrateBacklinks<any>(commentRefs);
const comments = hydrated.map((record) => normalizePullComment(record, pull.uri));
comments.sort(sortByCreatedAt);
return {
pull,
comments,
};
};
const normalizePullComment = (hydrated: HydratedRecord<any>, pullUri: ResourceUri): PullComment => {
const val = hydrated.value;
if (!val || typeof val !== 'object') {
return hydrated as PullComment;
}
const bodyText = (val.body && typeof val.body === 'object' && 'text' in val.body)
? val.body.text
: (typeof val.body === 'string' ? val.body : '');
const pullRoundIdx = val.pullRoundIdx !== undefined ? val.pullRoundIdx : val.round;
const replyTo = val.replyTo?.uri ?? val.replyTo;
const mappedValue: ShTangledRepoPullComment.Main = {
$type: 'sh.tangled.repo.pull.comment',
body: bodyText,
createdAt: val.createdAt || new Date().toISOString(),
pull: val.pull ?? val.subject?.uri ?? pullUri,
references: val.references ?? (replyTo ? [replyTo] : []),
};
return {
...hydrated,
value: {
...mappedValue,
replyTo,
pullRoundIdx,
} as any,
};
};
const listPullCommentsFromAppview = async (pullUri: ResourceUri): Promise<PullComment[]> => {
const records = await listAllAppviewRecords<any>(
'sh.tangled.feed.listComments',
pullUri,
);
const comments = await Promise.all(records.map(async (record) => {
const hydrated = await appviewRecordToHydrated<any>(record);
return normalizePullComment(hydrated, pullUri);
}));
comments.sort(sortByCreatedAt);
return comments;
};
const listPullsFromAppview = async (repo: RepoContext): Promise<PullSummary[]> => {
const records = await listAllAppviewRecords<ShTangledRepoPull.Main>(
'sh.tangled.repo.listPulls',
repo.repoDid,
) as Array<AppviewStatefulRecordView<ShTangledRepoPull.Main>>;
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<PaginatedResult<PullSummary>> => {
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<ShTangledRepoPull.Main, 'open' | 'closed' | 'merged'>(
'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<ShTangledRepoPull.Main, 'open' | 'closed' | 'merged'>(
'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<PullDetail> => {
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<PullSummary[]> =>
appviewFallback(
() => listPullsFromAppview(repo),
() => listPullsFromBacklinks(repo),
);
export const listPullsPage = async (
repo: RepoContext,
options: StatefulPageOptions<'open' | 'closed' | 'merged'>,
): Promise<PaginatedResult<PullSummary>> =>
appviewFallback(
() => listPullsPageFromAppview(repo, options),
() => listPullsPageFromBacklinks(repo, options),
);
export const getRepoPullCount = async (repo: RepoContext): Promise<number> =>
appviewFallback(
() => getAppviewRecordCount('sh.tangled.repo.countPulls', repo.repoDid),
() => getPullCountFromBacklinks(repo),
);
export const getPull = async (repo: RepoContext, pullRef: string): Promise<PullDetail> =>
appviewFallback(
() => getPullFromAppview(repo, pullRef),
() => getPullFromBacklinks(repo, pullRef),
);
export const fetchPullRoundPatch = async (
pull: PullSummary,
roundIndex: number,
): Promise<string> => {
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<RecordCreationResult> => {
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,
repo: input.sourceRepoDid,
},
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,
pullCid: string,
body: string,
pullRoundIdx: number,
): Promise<RecordCreationResult> => {
const rpc = createAuthRpc(agent);
const record = {
$type: 'sh.tangled.feed.comment',
subject: {
uri: pullUri,
cid: pullCid,
},
body: {
$type: 'sh.tangled.markup.markdown',
text: body.trim(),
},
createdAt: new Date().toISOString(),
pullRoundIdx,
};
const result = await ok(
rpc.post('com.atproto.repo.createRecord', {
input: {
repo: agent.sub,
collection: FEED_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<RecordCreationResult> => {
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<RecordCreationResult> => {
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');
};