Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664import { test } from "@playwright/test";
import { serviceDo, getOk, refsOf } from "./api";import { attach, fresh, note, within } from "./record";import { checkGit } from "./cobgit";import { config, type Account } from "./config";import { lsRemote } from "./git";import { knotFirehose, pdsFirehose, type Firehose } from "./firehose";import type { PocketPreferences } from "./lexicons";import { get, partsOf, recordUrl, until, xrpc, type Reply, type Written } from "./net";import { checkStream, streamCarries, watcher } from "./stream";import type { Held } from "./tabs";import { checkVersions, isCob, versionsBefore, type Version } from "./versions";
// read while the write is still held or a hop cannot tell what it changedexport interface Before { versions?: Version[]; mark?: number; knotMark?: number; pdsMark?: number; pdsExisted?: boolean; released?: number;}
export interface Settling { how: Held["method"] | "settled"; before?: Before; // whose notices get checked. with nobody here an "absent" check can't fail, so it proves nothing people?: Account[];}
export interface Hop { name: string; present: (written: Written, who: Account, settling?: Settling) => Promise<unknown>; absent: (written: Pick<Written, "uri">, who?: Account, settling?: Settling) => Promise<unknown>; look?: (uri: string | undefined) => Promise<Before>;}
// creates have no rkey yet, the knot mints it, so there is no uri to readexport const lookBefore = async (hops: Hop[], { body }: Held): Promise<Before> => { const uri = body.rkey ? `at://${body.repo}/${body.collection}/${body.rkey}` : undefined; return Object.assign({}, ...(await Promise.all(hops.map((hop) => hop.look?.(uri)))));};
// one step per check so the report shows how long after the write each service had it// the tidy replay runs outside a test, so test.info throws thereconst step = <T>(title: string, check: () => Promise<T>): Promise<T> => { try { test.info(); } catch { return check(); } return test.step(title, async (info) => { const record = fresh(); try { return await within(record, check); } finally { await attach(info, record); } });};
// one hop is one row, so a failing sub-check is named on it// the hop's own check keeps its plain messageconst alongside = async <T>( own: Promise<T>, checks: Record<string, Promise<unknown> | undefined>) => { const named = Object.entries(checks).filter(([, check]) => check); const [first, ...rest] = await Promise.allSettled([own, ...named.map(([, check]) => check)]); const failed = [first, ...rest].flatMap((check, at) => check.status === "rejected" ? [`${at ? `${named[at - 1][0]}: ` : ""}${(check.reason as Error).message}`] : [] ); if (failed.length) throw new Error(failed.join("\n ")); return (first as PromiseFulfilledResult<T>).value;};
const recordHop = (name: string, base: string) => { const has = ({ uri, cid }: Written) => until(`${name} should have ${uri} at ${cid}`, 10_000, async () => { const { status, body } = await get(recordUrl(base, uri)); const saw = status === 200 ? `it has ${body.cid}` : `it answered ${status} ${JSON.stringify(body)}`; return [status === 200 && body.cid === cid, saw, body.value]; }); const lacks = ({ uri }: Pick<Written, "uri">) => until(`${name} should have dropped ${uri}`, 10_000, async () => { const { status, body } = await get(recordUrl(base, uri)); // down or erroring is not the same as gone, only a 404 counts const gone = status === 404 || body?.error === "RecordNotFound"; return [ gone, status === 200 ? "it still has it" : `it answered ${status} ${JSON.stringify(body)}` ]; }); return { name, host: new URL(base).host, has, lacks, present: has, absent: lacks };};
const firehosed = ( watcher: Firehose, written: { uri: string; cid?: string }, settling?: Settling, since?: number, existed = true) => step("firehose", async () => { const how = settling?.how; const action = how === "put" ? (existed ? "update" : "create") : how === "settled" ? undefined : how; const frame = await watcher.check(written, action, since); const released = settling?.before?.released; if (released) note("arrived", `${frame.at - released} ms after release`); });
const pdsRecords = recordHop("pds", config.pdsUrl);export const pds = { ...pdsRecords, present: (written: Written, _who?: Account, settling?: Settling) => alongside(pdsRecords.has(written), { firehose: firehosed( pdsFirehose, written, settling, settling?.before?.pdsMark, settling?.before?.pdsExisted ) }), absent: ({ uri }: Pick<Written, "uri">, _who?: Account, settling?: Settling) => alongside(pdsRecords.lacks({ uri }), { firehose: firehosed(pdsFirehose, { uri }, settling, settling?.before?.pdsMark) }), look: async (uri: string | undefined): Promise<Before> => { const pdsMark = pdsFirehose.mark(); if (!uri) return { pdsMark }; // a pds put can create too, so its method alone cannot tell the action const { status, body } = await get(recordUrl(config.pdsUrl, uri)); if (status !== 200 && status !== 404 && body?.error !== "RecordNotFound") throw new Error( `pds should say whether ${uri} exists before the write but answered ${status} ${JSON.stringify(body)}` ); return { pdsMark, pdsExisted: status === 200 }; }};
const hydrantDb = recordHop("hydrant db", config.hydrantUrl);const streamed = ({ uri, cid }: { uri: string; cid?: string }, settling?: Settling) => streamCarries(uri) ? step("stream", async () => { const { mark, released } = settling?.before ?? {}; const frame = await checkStream({ uri, cid }, settling?.how ?? "settled", mark); // checkStream only notices a frame on its next wake, so the arrival time comes from the watcher if (released && frame) note("arrived", `${frame.at - released} ms after release`); }) : undefined;export const hydrant = { ...hydrantDb, name: "hydrant", present: (written: Written, _who?: Account, settling?: Settling) => alongside( step("db", () => hydrantDb.has(written)), { stream: streamed(written, settling) } ), absent: ({ uri }: Pick<Written, "uri">, _who?: Account, settling?: Settling) => alongside( step("db", () => hydrantDb.lacks({ uri })), { stream: streamed({ uri }, settling) } ), look: async () => ({ mark: watcher.mark() })};
const knotRecords = recordHop("knot", config.knotUrl);// a cob's uri authority is the repo, so the author lives in the record bodyconst editorIs = async (written: Written, did: string) => { const value = await knotRecords.has(written); if (value["x-tngl-editor"] !== did) { throw new Error( `knot should show ${did} as the editor of ${written.uri} but x-tngl-editor is ${value["x-tngl-editor"]} (use knotCopy for a record the knot wrote itself)` ); }};
// one listCobVersions read feeds both checks, no cid means the record was deletedconst cobChecks = ( written: { uri: string; cid?: string }, editor: string | undefined, settling?: Settling) => { const firehose = isCob(written.uri) || partsOf(written.uri).collection === "org.tangled.git.refs" ? firehosed(knotFirehose, written, settling, settling?.before?.knotMark) : undefined; if (!isCob(written.uri)) return { firehose }; const chain = step("versions", () => checkVersions(written, { editor, had: settling?.before?.versions }) ); return { firehose, versions: chain, // a chain that never settles fails as versions already, git would only repeat it // git's step starts once the chain is in, so its bar is only git's time git: chain.then( (versions) => step("git", () => checkGit(written, versions)), () => undefined ) };};const versionsLook = async (uri: string | undefined) => ({ knotMark: knotFirehose.mark(), versions: await versionsBefore(uri)});
const knotHop = (checksEditor: boolean): Hop => ({ name: "knot", present: (written, who, settling) => { const editor = checksEditor ? who.did : undefined; return alongside( step("record", () => (editor ? editorIs(written, editor) : knotRecords.has(written))), cobChecks(written, editor, settling) ); }, absent: ({ uri }, who, settling) => alongside( step("record", () => knotRecords.lacks({ uri })), cobChecks({ uri }, checksEditor ? who?.did : undefined, settling) ), look: versionsLook});
// the knot wrote this one itself, so there is no x-tngl-editor to checkexport const knotCopy = knotHop(false);
export const knot = { ...knotRecords, ...knotHop(true) };
// the knot rewrites the refs record a moment after a push, so wait for itexport const refsNaming = ( uri: string, what: string, wanted: (named: [sha: string, ref: string][]) => boolean) => until(what, 10_000, async () => { const now = await refsOf(uri); if (!now) return [false, "it has no refs record"]; const said = now.named.map(([sha, ref]) => `${ref} ${sha}`).join(", "); return [wanted(now.named), `it names ${said}`, now]; });
export const git: Hop = { name: "git ls-remote", present: (written, who) => until(`git should list exactly the refs ${written.uri} names`, 10_000, async () => { const stored = await refsOf(written.uri); if (stored?.cid !== written.cid) return [false, `the knot has ${stored?.cid ?? "no record"}`]; const named = Object.fromEntries(stored.named.map(([sha, ref]) => [ref, sha])); if (Object.keys(named).length === 0) return [false, "the record names no refs"]; // git lists HEAD, the refs record never does const listed = await lsRemote(who, partsOf(written.uri).repo, "refs/"); return [ JSON.stringify(Object.entries(named).sort()) === JSON.stringify(Object.entries(listed).sort()), `git lists ${JSON.stringify(listed)}, the record names ${JSON.stringify(named)}` ]; }), absent: async () => { throw new Error("git has no refs record to lose"); }};
const awaitRecord = async ({ uri, cid }: { uri: string; cid?: string }) => ( await get( xrpc(config.bobbinUrl, "org.tangled.bobbin.awaitRecord", { uri, ...(cid ? { cid } : {}) }) ) ).body;
const said = (body: { status?: string; reason?: string; detail?: string }) => body.status ? `it said ${body.status} ${body.reason ?? ""} ${body.detail ?? ""}` : JSON.stringify(body);
// parallel workers can push pds to hydrant to bobbin past 15 s, hence 30 sconst indexed = (written: Written) => until( `bobbin should have indexed ${written.uri} at ${written.cid}`, 30_000, async () => { const body = await awaitRecord(written); // a rejected record never gets indexed, so throw instead of polling if (body.status === "rejected") throw new Error(`bobbin should have indexed ${written.uri} but ${said(body)}`); return [body.status === "indexed" && body.cid === written.cid, said(body)]; }, { longPoll: true } );
const deleted = ({ uri }: Pick<Written, "uri">) => until( `bobbin should have deleted ${uri}`, 15_000, async () => { const body = await awaitRecord({ uri }); return [body.status === "deleted", said(body)]; }, { longPoll: true } );
const mirrorList = async (kind: "Branches" | "Tags", repoDid: string) => { const { status, body } = await get( xrpc(config.bobbinUrl, `org.tangled.temp.git.list${kind}`, { repo: repoDid, limit: "100" }) ); const names: string[] = (body?.[kind.toLowerCase()] ?? []).map((one: any) => one.name); return { status, names };};
export const bobbin = { name: "bobbin", indexed, list: mirrorList, present: indexed, absent: deleted, // a stall here means knot2 is rate limiting hydrant's backfill repoByName: (owner: string, name: string) => get(xrpc(config.bobbinUrl, "org.tangled.temp.repo.getRepoByName", { owner, name })), hasRepo: (ownerDid: string, name: string) => until( `bobbin should know the repo ${name} (hydrant may be backing off knot2)`, 30_000, async () => { const { status } = await bobbin.repoByName(ownerDid, name); return [status === 200, `getRepoByName answered ${status}`]; } ), // the mirror 404s an unsynced repo, even an empty one, and the web reads that as a knot needing an upgrade knowsRepo: (repoDid: string) => until(`bobbin should know ${repoDid}`, 60_000, async () => { const { status } = await mirrorList("Branches", repoDid); return [status === 200, `listBranches answered ${status}`]; }), // the mirror picks a push up a moment after the knot, so pages that read git wait for this first hasBranch: (repoDid: string, branch: string, sha: string) => until(`bobbin should have ${branch} at ${sha}`, 60_000, async () => { const { status, body } = await get( xrpc(config.bobbinUrl, "org.tangled.temp.git.getRef", { repo: repoDid, ref: `refs/heads/${branch}` }) ); return [ status === 200 && body.oid === sha, `getRef answered ${status} ${JSON.stringify(body)}` ]; }), countStars: async (subject: string) => { const body = await getOk( xrpc(config.bobbinUrl, "org.tangled.feed.countStars", { subject }) ); return body.count as number; }, listForks: (subject: string) => getOk(xrpc(config.bobbinUrl, "org.tangled.repo.listForks", { subject, limit: "50" })), countForks: async (subject: string) => { const body = await getOk( xrpc(config.bobbinUrl, "org.tangled.repo.countForks", { subject }) ); return body.count as number; }};
interface Workflow { name?: string; id: string; status: string; error?: string;}const FINISHED = ["success", "failed", "cancelled", "timeout"];
const finishedRun = (repoDid: string, sha: string) => until<{ pipeline: string; workflow: Workflow }>( "spindle should finish the workflow", 120_000, async () => { const { status, body } = await get( xrpc(config.spindleUrl, "org.tangled.ci.queryPipelines", { repo: repoDid, commits: sha }) ); const runs: { pipeline: string; workflow: Workflow }[] = ( body?.pipelines ?? [] ).flatMap((pipeline: { id: string; workflows?: Workflow[] }) => (pipeline.workflows ?? []).map((workflow) => ({ pipeline: pipeline.id, workflow })) ); const done = runs.find((one) => FINISHED.includes(one.workflow.status)); return [ Boolean(done), `it answered ${status} with ${JSON.stringify(runs.map((one) => one.workflow))}`, done ]; } );
// triggers look like org.tangled.event.push, so match on the suffixexport const spindle = { hasPipeline: (repoDid: string, sha: string, event: "push" | "pullRequest", budget = 30_000) => until( `spindle should have a ${event} pipeline for ${repoDid} at ${sha}`, budget, async () => { const { status, body } = await get( xrpc(config.spindleUrl, "org.tangled.ci.queryPipelines", { repo: repoDid, commits: sha }) ); const pipelines: { commit: string; trigger?: { $type?: string } }[] = body.pipelines ?? []; const found = pipelines.find( (p) => p.commit === sha && p.trigger?.$type?.endsWith(`.event.${event}`) ); const saw = status === 200 ? `it listed ${pipelines.length} pipelines for the commit` : `it answered ${status}`; return [Boolean(found), saw, found]; } ), finishedRun, finishedWorkflow: async (repoDid: string, sha: string) => (await finishedRun(repoDid, sha)).workflow};
export const settle = async ( written: Written, hops: Hop[], who: Account, nsid: string, how: Settling["how"], before?: Before, people?: Account[]) => { const settling = { how, before, people }; const checks = await Promise.allSettled( hops.map((hop) => step(hop.name, () => how === "delete" ? hop.absent(written, who, settling) : hop.present(written, who, settling) ) ) ); if (checks.every((check) => check.status === "fulfilled")) return; const table = checks.map((check, at) => check.status === "fulfilled" ? ` ✓ ${hops[at].name}` : ` ✗ ${hops[at].name} ${(check.reason as Error).message}` ); throw new Error([`${nsid} ${written.uri} ${written.cid} (${how})`, ...table].join("\n"));};
// pocket keys the saved lists by the token's audconst POCKET_AUD = "did:web:tangled.org";const KNOT_PREF = "org.tangled.temp.actor.defs#savedKnotPref";const SPINDLE_PREF = "org.tangled.temp.actor.defs#savedSpindlePref";const RECENTS_PREF = "org.tangled.temp.actor.defs#recentsPref";
export interface Saved { knots: string[]; spindles: string[];}
type Stored = PocketPreferences["preferences"];
const storedBy = async (who: Account): Promise<Stored> => { const lxm = "com.bad-example.pocket.getPreferences"; const reply = await serviceDo(who, config.bobbinUrl, POCKET_AUD, lxm); // a 400 NotFound means the account never saved anything if (reply.status === 400 && reply.body?.error === "NotFound") return []; if (reply.status !== 200) throw new Error(`pocket answered ${JSON.stringify(reply.body)}`); return reply.body.preferences;};
export const savedBy = async (who: Account): Promise<Saved> => { const stored = await storedBy(who); const hosts = (items: { value: string }[]) => items.map((item) => item.value).sort(); return { knots: hosts(stored.flatMap((pref) => (pref.$type === KNOT_PREF ? pref.items : []))), spindles: hosts(stored.flatMap((pref) => (pref.$type === SPINDLE_PREF ? pref.items : []))) };};
export const recentsIn = async (who: Account) => (await storedBy(who)).flatMap((pref) => (pref.$type === RECENTS_PREF ? pref.items : []));
// a single try, saveFor is the one that retriesexport const putSaved = async (who: Account, saved: Saved): Promise<Reply> => { const lxm = "com.bad-example.pocket.putPreferences"; const items = (hosts: string[]) => hosts.map((host) => ({ id: host, value: host })); return serviceDo(who, config.bobbinUrl, POCKET_AUD, lxm, { json: { preferences: [ { $type: KNOT_PREF, items: items(saved.knots) }, { $type: SPINDLE_PREF, items: items(saved.spindles) } ] } satisfies PocketPreferences });};
// pocket can't resolve a new account's did doc for a momentexport const saveFor = async (who: Account, saved: Saved) => until(`pocket should take the save for ${who.handle}`, 15_000, async () => { const reply = await putSaved(who, saved); return [reply.status === 200, `it answered ${reply.status} ${JSON.stringify(reply.body)}`]; });
// deliberi only answers the recipient, not other accountsconst deliberiGet = async (who: Account, lxm: string, params?: Record<string, string>) => { const aud = `did:web:${new URL(config.deliberiUrl).host}`; const reply = await serviceDo(who, config.deliberiUrl, aud, lxm, { params }); if (reply.status !== 200) throw new Error(`deliberi answered ${JSON.stringify(reply.body)}`); return reply.body;};
export type Toggles = Record<string, boolean>;
export const notificationsOf = (who: Account): Promise<Toggles> => deliberiGet(who, "org.tangled.temp.notification.getPreferences");
export interface Notice { uri: string; type: string; read: boolean; actorDid: string; repoDid?: string; issueAt?: string; pullAt?: string;}
export const noticesOf = async (who: Account): Promise<Notice[]> => (await deliberiGet(who, "org.tangled.temp.notification.listNotifications", { limit: "50" })) .notifications;
export const aboutOf = (notice: Notice) => notice.issueAt ?? notice.pullAt;
// a state change notice is the ticket uri with #rev on the endconst madeFrom = (notice: Notice, uri: string) => notice.uri === uri || notice.uri.startsWith(`${uri}#`);
// deliberi only shows each person their own notices, so ask as each oneexport const noticesMadeFrom = async (uri: string, people: Account[]) => ( await Promise.all( people.map(async (who) => (await noticesOf(who)) .filter((n) => madeFrom(n, uri)) .map((notice) => ({ who, notice })) ) ) ).flat();
// deliberi saves the rev with its notices, so once _ingested shows the rev the notices are writtenconst ingested = async (written: Written) => { // an edit keeps the uri, so only a rev at least as new as the write proves it landed const watcher = isCob(written.uri) ? knotFirehose : pdsFirehose; const { rev } = await watcher.check(written); return until( `deliberi should have ingested ${written.uri} at ${written.cid}`, 20_000, async () => { const { status, body } = await get( `${config.deliberiUrl}/xrpc/_ingested?${new URLSearchParams({ uri: written.uri })}` ); // a json 404 is just unseen, a plain text 404 means the route is missing if (status === 404 && typeof body === "string") throw new Error( `deliberi has no /xrpc/_ingested route, run it with DELIBERI_DEV=true, nothing else can say it processed ${written.uri}` ); // only tickets carry a cid, so a missing one doesn't count against it const has = status === 200 && body.rev >= rev && (body.cid === undefined || body.cid === written.cid); return [ has, `/_ingested answered ${status} ${JSON.stringify(body)}, the write is at rev ${rev}` ]; } );};
const told = (found: { who: Account; notice: Notice }[]) => found.map(({ who, notice }) => `${notice.type} to ${who.name}`);
export const deliberi: Hop = { name: "deliberi", present: async (written, _who, settling) => { await step("ingested", () => ingested(written)); const found = await noticesMadeFrom(written.uri, settling?.people ?? []); note("deliberi made", told(found).join(", ") || "no notice"); }, absent: async ({ uri }, _who, settling) => { await until(`deliberi should have no notice made from ${uri}`, 15_000, async () => { const found = await noticesMadeFrom(uri, settling?.people ?? []); return [found.length === 0, `it still has ${told(found).join(", ")}`]; }); }};
// the opener is the editor of the first version, read from the knot's gitexport const knotmirror = { // the mirror lags the knot a moment, so this polls openerOf: (uri: string) => until(`knotmirror should list who opened ${uri}`, 30_000, async () => { const { repo, collection, rkey } = partsOf(uri); const { status, body } = await get( xrpc(config.knotmirrorUrl, "org.tangled.temp.repo.listRecordOrigins", { repo, collection, limit: "1000" }) ); const found = status === 200 ? body.origins?.find((o: { rkey: string }) => o.rkey === rkey) : undefined; return [ Boolean(found), `listRecordOrigins answered ${status} ${JSON.stringify(body)}`, found?.author as string ]; })};
export const openers = { view: async (uri: string): Promise<string> => (await getOk(xrpc(config.bobbinUrl, "org.tangled.track.getTicketView", { uri }))).author .did, listed: async (repoDid: string, list: "listIssues" | "listPulls", uri: string) => { const body = await getOk( xrpc(config.bobbinUrl, `org.tangled.repo.${list}`, { subject: repoDid, limit: "100" }) ); return body.items.find((item: { uri: string }) => item.uri === uri)?.author as string | undefined; }, searched: async (repoDid: string, kind: "issue" | "pull", query: string) => { const body = await getOk( xrpc(config.bobbinUrl, "org.tangled.search.query", { repo: repoDid, nsid: "org.tangled.track.ticket", kind, ...(query ? { q: query } : {}), limit: "100" }) ); return body.hits as { uri: string; author?: string }[]; }};