diff --git a/api/org_tangled/cbor_gen.go b/api/org_tangled/cbor_gen.go index f68fd7c4c..5d7db298a 100644 --- a/api/org_tangled/cbor_gen.go +++ b/api/org_tangled/cbor_gen.go @@ -8738,136 +8738,6 @@ func (t *ReviewPatch_Source) UnmarshalCBOR(r io.Reader) (err error) { return nil } -func (t *ShTombstone) MarshalCBOR(w io.Writer) error { - if t == nil { - _, err := w.Write(cbg.CborNull) - return err - } - - cw := cbg.NewCborWriter(w) - - if _, err := cw.Write([]byte{162}); err != nil { - return err - } - - // t.LexiconTypeID (string) (string) - if len("$type") > 1000000 { - return xerrors.Errorf("Value in field \"$type\" was too long") - } - - if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("$type"))); err != nil { - return err - } - if _, err := cw.WriteString(string("$type")); err != nil { - return err - } - - if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("org.tangled.shTombstone"))); err != nil { - return err - } - if _, err := cw.WriteString(string("org.tangled.shTombstone")); err != nil { - return err - } - - // t.CreatedAt (string) (string) - if len("createdAt") > 1000000 { - return xerrors.Errorf("Value in field \"createdAt\" was too long") - } - - if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("createdAt"))); err != nil { - return err - } - if _, err := cw.WriteString(string("createdAt")); err != nil { - return err - } - - if len(t.CreatedAt) > 1000000 { - return xerrors.Errorf("Value in field t.CreatedAt was too long") - } - - if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.CreatedAt))); err != nil { - return err - } - if _, err := cw.WriteString(string(t.CreatedAt)); err != nil { - return err - } - return nil -} - -func (t *ShTombstone) UnmarshalCBOR(r io.Reader) (err error) { - *t = ShTombstone{} - - cr := cbg.NewCborReader(r) - - maj, extra, err := cr.ReadHeader() - if err != nil { - return err - } - defer func() { - if err == io.EOF { - err = io.ErrUnexpectedEOF - } - }() - - if maj != cbg.MajMap { - return fmt.Errorf("cbor input should be of type map") - } - - if extra > cbg.MaxLength { - return fmt.Errorf("ShTombstone: map struct too large (%d)", extra) - } - - n := extra - - nameBuf := make([]byte, 9) - for i := uint64(0); i < n; i++ { - nameLen, ok, err := cbg.ReadFullStringIntoBuf(cr, nameBuf, 1000000) - if err != nil { - return err - } - - if !ok { - // Field doesn't exist on this type, so ignore it - if err := cbg.ScanForLinks(cr, func(cid.Cid) {}); err != nil { - return err - } - continue - } - - switch string(nameBuf[:nameLen]) { - // t.LexiconTypeID (string) (string) - case "$type": - - { - sval, err := cbg.ReadStringWithMax(cr, 1000000) - if err != nil { - return err - } - - t.LexiconTypeID = string(sval) - } - // t.CreatedAt (string) (string) - case "createdAt": - - { - sval, err := cbg.ReadStringWithMax(cr, 1000000) - if err != nil { - return err - } - - t.CreatedAt = string(sval) - } - - default: - // Field doesn't exist on this type, so ignore it - if err := cbg.ScanForLinks(r, func(cid.Cid) {}); err != nil { - return err - } - } - } - - return nil -} func (t *SnippetString) MarshalCBOR(w io.Writer) error { if t == nil { _, err := w.Write(cbg.CborNull) diff --git a/api/org_tangled/shTombstone.go b/api/org_tangled/shTombstone.go deleted file mode 100644 index 243a6a100..000000000 --- a/api/org_tangled/shTombstone.go +++ /dev/null @@ -1,22 +0,0 @@ -// Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT. - -package org_tangled - -// schema: org.tangled.shTombstone - -import ( - "github.com/bluesky-social/indigo/lex/util" -) - -const ( - ShTombstoneNSID = "org.tangled.shTombstone" -) - -func init() { - util.RegisterType("org.tangled.shTombstone", &ShTombstone{}) -} // -// RECORDTYPE: ShTombstone -type ShTombstone struct { - LexiconTypeID string `json:"$type,const=org.tangled.shTombstone" cborgen:"$type,const=org.tangled.shTombstone"` - CreatedAt string `json:"createdAt" cborgen:"createdAt"` -} diff --git a/bobbin/crates/bobbin/src/main.rs b/bobbin/crates/bobbin/src/main.rs index 8ddedc987..6e0d2e8e6 100644 --- a/bobbin/crates/bobbin/src/main.rs +++ b/bobbin/crates/bobbin/src/main.rs @@ -431,7 +431,7 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { entropy, stream_health: stream_health.clone(), ws: ws.clone(), - hydrant: Some(hydrant), + hydrant: Some(hydrant.clone()), cancel: cancel.clone(), disconnects: None, warming_shadow: None, @@ -502,6 +502,7 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { .with_invites(invites) .with_claims(claims) .with_ticket_authors(ticket_authors) + .with_hydrant(Some(hydrant)) .with_stream_health(stream_health) .with_moderation(moderation); let app = router(state).layer(axum::middleware::from_fn(metrics::xrpc)); diff --git a/bobbin/crates/resolver/src/hydrant.rs b/bobbin/crates/resolver/src/hydrant.rs index a601b64aa..54b688d00 100644 --- a/bobbin/crates/resolver/src/hydrant.rs +++ b/bobbin/crates/resolver/src/hydrant.rs @@ -164,6 +164,10 @@ impl HydrantClient { Ok(url) } + pub fn xrpc_url(&self, nsid: &str) -> Result { + self.endpoint(["xrpc", nsid]) + } + fn repo_url(&self, did: &Did) -> Result { self.endpoint(["repos", did.as_str()]) } diff --git a/bobbin/crates/xrpc/src/lib.rs b/bobbin/crates/xrpc/src/lib.rs index 562205cf0..f20d9159d 100644 --- a/bobbin/crates/xrpc/src/lib.rs +++ b/bobbin/crates/xrpc/src/lib.rs @@ -29,7 +29,7 @@ use bobbin_edge_index::{ }; use bobbin_knot_proxy::{FillConfig, KnotProxyError, MirrorProxy, ProxyResponse, TicketAuthors}; use bobbin_record_lru::RecordStore; -use bobbin_resolver::{IdentityResolveError, IdentityResolver, RepoIdResolver}; +use bobbin_resolver::{HydrantClient, IdentityResolveError, IdentityResolver, RepoIdResolver}; use bobbin_runtime::{ReqwestHttp, UnixMicros}; use bobbin_search::{ActorIndex, ActorIndexError, SearchReader}; use bobbin_slingshot_client::{SlingshotClient, SlingshotError}; @@ -219,6 +219,7 @@ pub struct AppState { /// the ingest stream's liveness, absent when nothing publishes it pub stream_health: Option>, pub moderation: Option, + pub hydrant: Option, cors: Cors, service_auth_config: service_auth::ServiceAuthConfig>, enrich_router: Arc>, @@ -270,6 +271,7 @@ impl AppState { ), stream_health: None, moderation: None, + hydrant: None, cors: Cors::Off, service_auth_config: ServiceAuthConfig::new( Did::new_static("did:web:localhost").unwrap(), @@ -346,6 +348,11 @@ impl AppState { self } + pub fn with_hydrant(mut self, hydrant: Option) -> Self { + self.hydrant = hydrant; + self + } + pub fn with_stream_health(mut self, stream_health: Arc) -> Self { self.stream_health = Some(stream_health); self @@ -616,6 +623,10 @@ pub fn router(state: AppState) -> Router { &format!("/xrpc/{}", pocket::PUT_PREFERENCES), axum::routing::post(pocket::put_preferences), ) + .route( + &format!("/xrpc/{UNMARK_LIAR_ACCOUNT}"), + axum::routing::post(unmark_liar_account), + ) .merge(mirror_routes()) .layer( TraceLayer::new_for_http() @@ -3643,6 +3654,46 @@ async fn proxy_mirror_handler( .map_err(map_proxy_error) } +const UNMARK_LIAR_ACCOUNT: &str = "org.tangled.foster.unmarkLiarAccount"; + +// hydrant checks the service auth: bobbin resolves dids through hydrant, which +// serves liar accounts keyless docs +async fn unmark_liar_account( + State(state): State, + headers: HeaderMap, + body: Bytes, +) -> Result { + let hydrant = state + .hydrant + .as_ref() + .ok_or_else(|| XrpcError::UpstreamUnavailable("hydrant is unset".into()))?; + let url = hydrant + .xrpc_url(UNMARK_LIAR_ACCOUNT) + .map_err(|e| XrpcError::UpstreamUnavailable(format!("hydrant: {e}")))?; + let mut request = reqwest::Client::new().post(url).body(body); + for name in [AUTHORIZATION, CONTENT_TYPE] { + if let Some(value) = headers.get(&name) { + request = request.header(name, value); + } + } + + let upstream = request + .send() + .await + .map_err(|e| XrpcError::UpstreamUnavailable(format!("hydrant: {e}")))?; + + let status = upstream.status(); + let content_type = upstream.headers().get(CONTENT_TYPE).cloned(); + let mut response = Response::builder() + .status(status) + .body(Body::from_stream(upstream.bytes_stream())) + .expect("response body construction must succeed"); + if let Some(value) = content_type { + response.headers_mut().insert(CONTENT_TYPE, value); + } + Ok(response) +} + async fn proxy_mirror_write_handler( state: AppState, headers: HeaderMap, diff --git a/cmd/cborgen/cborgen.go b/cmd/cborgen/cborgen.go index 1ca6ddd54..92cda5a75 100644 --- a/cmd/cborgen/cborgen.go +++ b/cmd/cborgen/cborgen.go @@ -127,7 +127,6 @@ func main() { org_tangled.RepoReviewUnspeccedRound{}, org_tangled.ReviewPatch{}, org_tangled.ReviewPatch_Source{}, - org_tangled.ShTombstone{}, org_tangled.SnippetString{}, org_tangled.SnippetString_TextFile{}, org_tangled.TrackTicket{}, diff --git a/lexicons/org/tangled/shTombstone.json b/lexicons/org/tangled/shTombstone.json deleted file mode 100644 index f8cb7a44c..000000000 --- a/lexicons/org/tangled/shTombstone.json +++ /dev/null @@ -1,21 +0,0 @@ -{ - "lexicon": 1, - "id": "org.tangled.shTombstone", - "defs": { - "main": { - "type": "record", - "description": "Written to an account's own repo once its org.tangled records have been copied over from the liar pds.", - "key": "literal:self", - "record": { - "type": "object", - "required": ["createdAt"], - "properties": { - "createdAt": { - "type": "string", - "format": "datetime" - } - } - } - } - } -} diff --git a/web/src/lib/api/lexicons/index.ts b/web/src/lib/api/lexicons/index.ts index abc84b032..3db669e2c 100644 --- a/web/src/lib/api/lexicons/index.ts +++ b/web/src/lib/api/lexicons/index.ts @@ -128,7 +128,6 @@ export * as OrgTangledSearchQuery from "./types/org/tangled/search/query.js"; export * as OrgTangledSecretAddSecret from "./types/org/tangled/secret/addSecret.js"; export * as OrgTangledSecretListSecrets from "./types/org/tangled/secret/listSecrets.js"; export * as OrgTangledSecretRemoveSecret from "./types/org/tangled/secret/removeSecret.js"; -export * as OrgTangledShTombstone from "./types/org/tangled/shTombstone.js"; export * as OrgTangledSnippetString from "./types/org/tangled/snippet/string.js"; export * as OrgTangledStringCountStrings from "./types/org/tangled/string/countStrings.js"; export * as OrgTangledStringGetString from "./types/org/tangled/string/getString.js"; diff --git a/web/src/lib/api/lexicons/types/org/tangled/shTombstone.ts b/web/src/lib/api/lexicons/types/org/tangled/shTombstone.ts deleted file mode 100644 index a4d47bdad..000000000 --- a/web/src/lib/api/lexicons/types/org/tangled/shTombstone.ts +++ /dev/null @@ -1,25 +0,0 @@ -import type {} from "@atcute/lexicons"; -import * as v from "@atcute/lexicons/validations"; -import type {} from "@atcute/lexicons/ambient"; - -const _mainSchema = /*#__PURE__*/ v.record( - /*#__PURE__*/ v.literal("self"), - /*#__PURE__*/ v.object({ - $type: /*#__PURE__*/ v.literal("org.tangled.shTombstone"), - createdAt: /*#__PURE__*/ v.datetimeString(), - }), -); - -type main$schematype = typeof _mainSchema; - -export interface mainSchema extends main$schematype {} - -export const mainSchema = _mainSchema as mainSchema; - -export interface Main extends v.InferInput {} - -declare module "@atcute/lexicons/ambient" { - interface Records { - "org.tangled.shTombstone": mainSchema; - } -} diff --git a/web/src/lib/components/migration/MigrationLanding.svelte b/web/src/lib/components/migration/MigrationLanding.svelte index 82e9fb3e6..f1605ecee 100644 --- a/web/src/lib/components/migration/MigrationLanding.svelte +++ b/web/src/lib/components/migration/MigrationLanding.svelte @@ -10,6 +10,7 @@ import { importFromLiar, MigrationScopeError, + unmarkLiarAccount, type ImportProgress } from "#lib/migration/import.ts"; import { waitForMigration, type MigrationConfig } from "#lib/migration/status.ts"; @@ -26,8 +27,9 @@ let progress = $state(null); let waiting = $state(false); let needsGrant = $state(false); - // a retry after the records are in skips the import + // a retry skips whatever already went through let imported = false; + let unmarked = false; const migrate = createAction(async () => { const agent = auth.agent; @@ -35,17 +37,23 @@ needsGrant = false; try { + const fetch = config.liarMock + ? (await import("#lib/migration/mock.ts")).mockLiarFetch + : undefined; if (!imported) { - const fetch = config.liarMock - ? (await import("#lib/migration/mock.ts")).mockLiarFetch - : undefined; await importFromLiar(agent, { liarPdsUrl: config.liarPdsUrl, + bobbinUrl: config.bobbinUrl, fetch, onProgress: (next) => (progress = next) }); imported = true; } + if (!unmarked) { + progress = { phase: "finalizing", done: 0, total: 0 }; + await unmarkLiarAccount(agent, config.bobbinUrl, { fetch }); + unmarked = true; + } progress = null; waiting = true; diff --git a/web/src/lib/migration/import.test.ts b/web/src/lib/migration/import.test.ts index 2b14b409b..0fd8add6f 100644 --- a/web/src/lib/migration/import.test.ts +++ b/web/src/lib/migration/import.test.ts @@ -7,11 +7,12 @@ import { MigrationScopeError } from "#lib/migration/import.ts"; import { buildRepoCar } from "#lib/migration/repoCar.ts"; -import { TOMBSTONE_COLLECTION } from "#lib/migration/tombstone.ts"; import type { Nsid, Rkey } from "#lib/api/syntax.ts"; const did = "did:plc:alice"; const liarPdsUrl = "https://liar.test"; +const bobbinUrl = "https://bobbin.test"; +const unmark = "rpc:org.tangled.foster.unmarkLiarAccount?aud=*"; const CID = "bafyreihltcnuuyqp2jm24aqydpnlj7b6w3ogwrplomrjtg5rifv44mmjey"; const createdAt = "2026-01-01T00:00:00.000Z"; @@ -27,7 +28,7 @@ const liarFetchFor = ( ) => vi.fn(async () => new Response(await buildRepoCar(did, records, options))); const fakePds = ({ - scope = "atproto repo:org.tangled.feed.star repo:org.tangled.actor.profile repo:org.tangled.shTombstone", + scope = `atproto repo:org.tangled.feed.star repo:org.tangled.actor.profile ${unmark}`, aud = "https://pds.test", failApplyOnCall }: { scope?: string; aud?: string; failApplyOnCall?: number } = {}) => { @@ -60,12 +61,6 @@ const fakePds = ({ })) }); } - case "/xrpc/com.atproto.repo.getRecord": { - const key = `${url.searchParams.get("collection")}/${url.searchParams.get("rkey")}`; - return records.has(key) - ? json({ uri: `at://${did}/${key}`, cid: CID, value: records.get(key) }) - : json({ error: "RecordNotFound" }, 400); - } case "/xrpc/com.atproto.repo.applyWrites": { applyCalls++; if (applyCalls === failApplyOnCall) @@ -75,13 +70,6 @@ const fakePds = ({ records.set(`${write.collection}/${write.rkey}`, write.value); return json({}); } - case "/xrpc/com.atproto.repo.createRecord": { - records.set(`${body.collection}/${body.rkey}`, body.record); - return json({ - uri: `at://${did}/${body.collection}/${body.rkey}`, - cid: CID - }); - } } return json({ error: "MethodNotImplemented" }, 501); } @@ -90,7 +78,7 @@ const fakePds = ({ }; describe("importFromLiar", () => { - it("writes every org.tangled record in batches, then the tombstone, and skips the rest", async () => { + it("writes every org.tangled record in batches and skips the rest", async () => { const pds = fakePds(); const phases: string[] = []; const fetch = liarFetchFor([ @@ -109,6 +97,7 @@ describe("importFromLiar", () => { const result = await importFromLiar(pds.agent, { liarPdsUrl, + bobbinUrl, fetch, onProgress: ({ phase }) => void (phases.at(-1) === phase || phases.push(phase)) }); @@ -120,8 +109,7 @@ describe("importFromLiar", () => { description: "hi" }); expect([...pds.records.keys()].some((key) => key.startsWith("app.bsky"))).toBe(false); - expect(pds.records.has(`${TOMBSTONE_COLLECTION}/self`)).toBe(true); - expect(phases).toEqual(["reading", "importing", "finalizing"]); + expect(phases).toEqual(["reading", "importing"]); const requested = new URL(String(fetch.mock.calls[0]?.[0])); expect(requested.origin + requested.pathname).toBe( `${liarPdsUrl}/xrpc/com.atproto.sync.getRepo` @@ -129,23 +117,21 @@ describe("importFromLiar", () => { expect(requested.searchParams.get("did")).toBe(did); }); - it("resumes after a failed batch without rewriting records or tombstoning early", async () => { + it("resumes after a failed batch without rewriting records", async () => { const pds = fakePds({ failApplyOnCall: 2 }); const records = Array.from({ length: 150 }, (_, index) => star(index)); const fetch = liarFetchFor(records); - await expect(importFromLiar(pds.agent, { liarPdsUrl, fetch })).rejects.toThrow(); + await expect(importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch })).rejects.toThrow(); expect(pds.records.size).toBe(100); - expect(pds.records.has(`${TOMBSTONE_COLLECTION}/self`)).toBe(false); pds.records.set(`org.tangled.feed.star/${records[0]!.rkey}`, { edited: true }); - const result = await importFromLiar(pds.agent, { liarPdsUrl, fetch }); + const result = await importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch }); expect(result).toEqual({ written: 50, skipped: 100 }); expect(pds.records.get(`org.tangled.feed.star/${records[0]!.rkey}`)).toEqual({ edited: true }); - expect(pds.records.has(`${TOMBSTONE_COLLECTION}/self`)).toBe(true); }); it("stops before writing anything when the grant misses a collection", async () => { @@ -159,21 +145,25 @@ describe("importFromLiar", () => { } ]); - const failure = await importFromLiar(pds.agent, { liarPdsUrl, fetch }).catch( + const failure = await importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch }).catch( (cause: unknown) => cause ); expect(failure).toBeInstanceOf(MigrationScopeError); expect((failure as MigrationScopeError).missing).toEqual([ { resource: "repo", collection: "org.tangled.actor.profile", actions: ["create"] }, - { resource: "repo", collection: TOMBSTONE_COLLECTION, actions: ["create"] } + { + resource: "rpc", + lxm: "org.tangled.foster.unmarkLiarAccount", + aud: "did:web:bobbin.test" + } ]); expect(pds.records.size).toBe(0); }); it("doesn't ask for scopes over collections that are already on the pds", async () => { const pds = fakePds({ - scope: "atproto repo:org.tangled.feed.star repo:org.tangled.shTombstone" + scope: `atproto repo:org.tangled.feed.star ${unmark}` }); const profile = { $type: "org.tangled.actor.profile", description: "hi" }; pds.records.set("org.tangled.actor.profile/self", profile); @@ -182,17 +172,17 @@ describe("importFromLiar", () => { { collection: "org.tangled.actor.profile", rkey: "self", value: profile } ]); - await expect(importFromLiar(pds.agent, { liarPdsUrl, fetch })).resolves.toEqual({ + await expect(importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch })).resolves.toEqual({ written: 1, skipped: 1 }); }); - it("refuses an incomplete repo instead of tombstoning over missing data", async () => { + it("refuses an incomplete repo instead of unmarking over missing data", async () => { const pds = fakePds(); const fetch = liarFetchFor([star(0)], { dropRecordBlocks: true }); - await expect(importFromLiar(pds.agent, { liarPdsUrl, fetch })).rejects.toThrow( + await expect(importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch })).rejects.toThrow( "incomplete" ); expect(pds.records.size).toBe(0); @@ -202,7 +192,7 @@ describe("importFromLiar", () => { const pds = fakePds({ aud: `${liarPdsUrl}/` }); const fetch = liarFetchFor([star(0)]); - await expect(importFromLiar(pds.agent, { liarPdsUrl, fetch })).rejects.toThrow( + await expect(importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch })).rejects.toThrow( "nowhere to move" ); expect(fetch).not.toHaveBeenCalled(); @@ -234,11 +224,11 @@ describe("batchWrites", () => { }); describe("missingImportScopes", () => { - it("always asks for the tombstone collection too", () => { + it("always asks for the unmark rpc too", () => { expect( - missingImportScopes("atproto", ["org.tangled.feed.star" as Nsid]).map( - (p) => p.resource === "repo" && p.collection + missingImportScopes("atproto", ["org.tangled.feed.star" as Nsid], bobbinUrl).map((p) => + p.resource === "repo" ? p.collection : p.lxm ) - ).toEqual(["org.tangled.feed.star", TOMBSTONE_COLLECTION]); + ).toEqual(["org.tangled.feed.star", "org.tangled.foster.unmarkLiarAccount"]); }); }); diff --git a/web/src/lib/migration/import.ts b/web/src/lib/migration/import.ts index ece36b5ad..c956c8be3 100644 --- a/web/src/lib/migration/import.ts +++ b/web/src/lib/migration/import.ts @@ -3,18 +3,19 @@ import { mainSchema as applyWritesSchema } from "@atcute/atproto/types/repo/appl import { mainSchema as listRecordsSchema } from "@atcute/atproto/types/repo/listRecords"; import type { OAuthUserAgent } from "@atcute/oauth-browser-client"; import { clear } from "#lib/api/cache.ts"; -import { createRecord, readRecord } from "#lib/api/write.ts"; -import type { Nsid, Rkey } from "#lib/api/syntax.ts"; -import { createClient } from "#lib/auth/agent.ts"; +import { createBobbinClient } from "#lib/api/client.ts"; +import { jsonPost } from "#lib/api/_request.ts"; +import { nsidOf, type Nsid, type Rkey } from "#lib/api/syntax.ts"; +import { createClient, mintServiceAuth, serviceDidForHost } from "#lib/auth/agent.ts"; import { missingPermissions, type Permission } from "#lib/auth/scopes.ts"; import { fetchLiarRepo, readLiarRecords, type LiarRecord } from "#lib/migration/liar.ts"; import { sameOrigin } from "#lib/migration/status.ts"; -import { TOMBSTONE_COLLECTION, TOMBSTONE_RKEY, tombstoneRecord } from "#lib/migration/tombstone.ts"; // xrpc-server's default json body limit is 100 kb const MAX_WRITES = 100; const MAX_BATCH_BYTES = 90 * 1024; const LIST_PAGE = 100; +const UNMARK = nsidOf("org.tangled.foster.unmarkLiarAccount"); export interface ImportProgress { phase: "reading" | "importing" | "finalizing"; @@ -40,16 +41,17 @@ interface CreateWrite { export const missingImportScopes = ( scope: string, - collections: Iterable + collections: Iterable, + bobbinUrl: string ): readonly Permission[] => - missingPermissions( - scope, - [...new Set([...collections, TOMBSTONE_COLLECTION])].map((collection): Permission => ({ + missingPermissions(scope, [ + ...[...new Set(collections)].map((collection): Permission => ({ resource: "repo", collection, actions: ["create"] - })) - ); + })), + { resource: "rpc", lxm: UNMARK, aud: serviceDidForHost(bobbinUrl) } + ]); export const batchWrites = (writes: readonly CreateWrite[]): CreateWrite[][] => { const encoder = new TextEncoder(); @@ -120,21 +122,34 @@ const writesToMake = async ( })); }; -// goes last, hydrant drops the user from liar as soon as it sees this -const writeTombstone = async (agent: OAuthUserAgent): Promise => { - if (await readRecord(agent, TOMBSTONE_COLLECTION, TOMBSTONE_RKEY)) return; - await createRecord(agent, TOMBSTONE_COLLECTION, tombstoneRecord(), TOMBSTONE_RKEY); +// goes after the import, hydrant drops the user from liar as soon as this lands +export const unmarkLiarAccount = async ( + agent: OAuthUserAgent, + bobbinUrl: string, + { fetch, signal }: { fetch?: typeof globalThis.fetch; signal?: AbortSignal } = {} +): Promise => { + const token = await mintServiceAuth(agent, { + aud: serviceDidForHost(bobbinUrl), + lxm: UNMARK, + signal + }); + await jsonPost(createBobbinClient({ serviceUrl: bobbinUrl, fetch }), UNMARK, undefined, { + headers: { authorization: `Bearer ${token}` }, + signal + }); }; export const importFromLiar = async ( agent: OAuthUserAgent, { liarPdsUrl, + bobbinUrl, fetch, signal, onProgress }: { liarPdsUrl: string; + bobbinUrl: string; fetch?: typeof globalThis.fetch; signal?: AbortSignal; onProgress?: (progress: ImportProgress) => void; @@ -155,7 +170,8 @@ export const importFromLiar = async ( const missing = missingImportScopes( agent.session.token.scope, - writes.map((write) => write.collection) + writes.map((write) => write.collection), + bobbinUrl ); if (missing.length > 0) throw new MigrationScopeError(missing); @@ -171,9 +187,6 @@ export const importFromLiar = async ( onProgress?.({ phase: "importing", done, total: writes.length }); } - signal?.throwIfAborted(); - onProgress?.({ phase: "finalizing", done, total: writes.length }); - await writeTombstone(agent); clear(); return { written: writes.length, skipped: records.length - writes.length }; }; diff --git a/web/src/lib/migration/liar.test.ts b/web/src/lib/migration/liar.test.ts index 4986446ad..14b90972b 100644 --- a/web/src/lib/migration/liar.test.ts +++ b/web/src/lib/migration/liar.test.ts @@ -12,11 +12,6 @@ describe("readLiarRecords", () => { it("keeps org.tangled records and drops everything else", async () => { const car = await buildRepoCar(did, [ { collection: "org.tangled.feed.star", rkey: "3aaaaaaaaaaaa", value: star }, - { - collection: "org.tangled.shTombstone", - rkey: "self", - value: { $type: "org.tangled.shTombstone", createdAt: star.createdAt } - }, { collection: "org.tangled.actor.profile", rkey: "self", diff --git a/web/src/lib/migration/liar.ts b/web/src/lib/migration/liar.ts index a47c19423..e9ca1511b 100644 --- a/web/src/lib/migration/liar.ts +++ b/web/src/lib/migration/liar.ts @@ -2,7 +2,6 @@ import { fromStream } from "@atcute/repo"; import { createBobbinClient } from "#lib/api/client.ts"; import { rawGet } from "#lib/api/_request.ts"; import { nsidOf, recordKeyOf, type Nsid, type Rkey } from "#lib/api/syntax.ts"; -import { TOMBSTONE_COLLECTION } from "#lib/migration/tombstone.ts"; const IMPORT_PREFIX = "org.tangled."; @@ -30,12 +29,7 @@ export const readLiarRecords = async ( const records: LiarRecord[] = []; try { for await (const entry of repo) { - if ( - !entry.collection.startsWith(IMPORT_PREFIX) || - entry.collection === TOMBSTONE_COLLECTION - ) { - continue; - } + if (!entry.collection.startsWith(IMPORT_PREFIX)) continue; records.push({ collection: nsidOf(entry.collection), rkey: recordKeyOf(entry.rkey), @@ -45,7 +39,7 @@ export const readLiarRecords = async ( } finally { await repo.dispose(); } - // without this a truncated car ends in a tombstone over records we never wrote + // without this a truncated car ends in an unmark over records we never wrote if (repo.missingBlocks.length > 0) { throw new Error( `The repo from the liar PDS is incomplete (${repo.missingBlocks.length} blocks missing).` diff --git a/web/src/lib/migration/mock.ts b/web/src/lib/migration/mock.ts index 1b7bd89b8..831780542 100644 --- a/web/src/lib/migration/mock.ts +++ b/web/src/lib/migration/mock.ts @@ -1,9 +1,9 @@ import type { MiniDoc } from "#lib/api/identity.ts"; import { buildRepoCar } from "#lib/migration/repoCar.ts"; -import { TOMBSTONE_COLLECTION, TOMBSTONE_RKEY } from "#lib/migration/tombstone.ts"; const STARS = 230; const CREATED_AT = "2026-01-01T00:00:00.000Z"; +const starRkey = (index: number) => `3mock${String(index).padStart(8, "0")}`; // more than one applyWrites batch, so progress and batching get exercised const mockRecords = (did: string) => [ @@ -24,7 +24,7 @@ const mockRecords = (did: string) => [ }, ...Array.from({ length: STARS }, (_, index) => ({ collection: "org.tangled.feed.star", - rkey: `3mock${String(index).padStart(8, "0")}`, + rkey: starRkey(index), value: { $type: "org.tangled.feed.star", subject: { $type: "org.tangled.feed.star#repo", identity: `did:plc:mockrepo${index}` }, @@ -35,6 +35,9 @@ const mockRecords = (did: string) => [ export const mockLiarFetch: typeof globalThis.fetch = async (input) => { const request = new URL(input instanceof Request ? input.url : String(input)); + if (request.pathname === "/xrpc/org.tangled.foster.unmarkLiarAccount") { + return new Response(null, { status: 200 }); + } const did = request.searchParams.get("did"); if (request.pathname !== "/xrpc/com.atproto.sync.getRepo" || !did) { return new Response("{}", { status: 404 }); @@ -44,6 +47,7 @@ export const mockLiarFetch: typeof globalThis.fetch = async (input) => { }); }; +// the mock records landing on the real pds counts as migrated export const mockHydrantDoc = async ( doc: MiniDoc, liarPdsUrl: string, @@ -52,11 +56,11 @@ export const mockHydrantDoc = async ( if (!doc.pds) return doc; const url = new URL("/xrpc/com.atproto.repo.getRecord", doc.pds); url.searchParams.set("repo", doc.did); - url.searchParams.set("collection", TOMBSTONE_COLLECTION); - url.searchParams.set("rkey", TOMBSTONE_RKEY); - const tombstoned = await fetch(url).then( + url.searchParams.set("collection", "org.tangled.feed.star"); + url.searchParams.set("rkey", starRkey(STARS - 1)); + const imported = await fetch(url).then( (response) => response.ok, () => false ); - return tombstoned ? doc : { ...doc, pds: liarPdsUrl as MiniDoc["pds"] }; + return imported ? doc : { ...doc, pds: liarPdsUrl as MiniDoc["pds"] }; }; diff --git a/web/src/lib/migration/status.test.ts b/web/src/lib/migration/status.test.ts index ab3b14942..7211a4087 100644 --- a/web/src/lib/migration/status.test.ts +++ b/web/src/lib/migration/status.test.ts @@ -113,8 +113,8 @@ describe("waitForMigration", () => { }); describe("migrationPending with the mock hydrant", () => { - it("stays pending until the user's pds has the tombstone", async () => { - let tombstoned = false; + it("stays pending until the user's pds has the mock records", async () => { + let imported = false; const fetch = vi.fn(async (input) => { const url = new URL(String(input)); if (url.pathname.endsWith("resolveMiniDoc")) { @@ -125,25 +125,25 @@ describe("migrationPending with the mock hydrant", () => { } ); } - return new Response("{}", { status: tombstoned ? 200 : 400 }); + return new Response("{}", { status: imported ? 200 : 400 }); }); const ctx = { serviceUrl: "https://bobbin.test", fetch } as unknown as BobbinContext; const config = { liarPdsUrl, liarMock: true }; await expect(migrationPending(ctx, did, config)).resolves.toBe(true); - tombstoned = true; + imported = true; await expect(migrationPending(ctx, did, config)).resolves.toBe(false); }); }); describe("mock hydrant", () => { const realDoc = { did, handle: "alice.test", pds: "https://pds.test" } as unknown as MiniDoc; - const pdsHolding = (tombstone: boolean) => + const pdsHolding = (imported: boolean) => vi.fn( - async () => new Response("{}", { status: tombstone ? 200 : 400 }) + async () => new Response("{}", { status: imported ? 200 : 400 }) ); - it("shows the liar pds until the user's pds holds the tombstone", async () => { + it("shows the liar pds until the user's pds holds the mock records", async () => { const before = pdsHolding(false); expect((await mockHydrantDoc(realDoc, liarPdsUrl, before)).pds).toBe(liarPdsUrl); @@ -151,7 +151,7 @@ describe("mock hydrant", () => { expect((await mockHydrantDoc(realDoc, liarPdsUrl, after)).pds).toBe("https://pds.test"); const asked = new URL(String(after.mock.calls[0]?.[0])); expect(asked.origin).toBe("https://pds.test"); - expect(asked.searchParams.get("collection")).toBe("org.tangled.shTombstone"); + expect(asked.searchParams.get("collection")).toBe("org.tangled.feed.star"); }); it("serves a repo the real reader can walk", async () => { diff --git a/web/src/lib/migration/status.ts b/web/src/lib/migration/status.ts index baedf06ba..e43fd01bf 100644 --- a/web/src/lib/migration/status.ts +++ b/web/src/lib/migration/status.ts @@ -19,7 +19,7 @@ export const sameOrigin = (candidate: string | undefined, expectedUrl: string): return expected !== null && candidate !== undefined && originOf(candidate) === expected; }; -// not cached, it has to flip as soon as hydrant sees the tombstone +// not cached, it has to flip as soon as hydrant unmarks the account export const migrationPending = async ( ctx: BobbinContext, did: string, @@ -48,7 +48,7 @@ const sleep = (ms: number, signal?: AbortSignal) => signal?.addEventListener("abort", onAbort, { once: true }); }); -// hydrant reads the tombstone off the firehose, so the doc lags the write +// hydrant may lag the unmark, so this polls the doc export const waitForMigration = async ( ctx: BobbinContext, did: string, diff --git a/web/src/lib/migration/tombstone.test.ts b/web/src/lib/migration/tombstone.test.ts deleted file mode 100644 index 0e065c004..000000000 --- a/web/src/lib/migration/tombstone.test.ts +++ /dev/null @@ -1,13 +0,0 @@ -import { is } from "@atcute/lexicons/validations"; -import { describe, expect, it } from "vitest"; -import { mainSchema } from "#lib/api/lexicons/types/org/tangled/shTombstone.ts"; -import { TOMBSTONE_COLLECTION, tombstoneRecord } from "#lib/migration/tombstone.ts"; - -describe("tombstoneRecord", () => { - it("is a valid org.tangled.shTombstone record", () => { - const record = tombstoneRecord(new Date("2026-01-01T00:00:00.000Z")); - - expect(is(mainSchema, record)).toBe(true); - expect(record.$type).toBe(TOMBSTONE_COLLECTION); - }); -}); diff --git a/web/src/lib/migration/tombstone.ts b/web/src/lib/migration/tombstone.ts deleted file mode 100644 index 15b397fed..000000000 --- a/web/src/lib/migration/tombstone.ts +++ /dev/null @@ -1,11 +0,0 @@ -import type * as OrgTangledShTombstone from "#lib/api/lexicons/types/org/tangled/shTombstone.ts"; -import { datetimeOf, nsidOf, recordKeyOf } from "#lib/api/syntax.ts"; - -// hydrant has to watch this collection -export const TOMBSTONE_COLLECTION = nsidOf("org.tangled.shTombstone"); -export const TOMBSTONE_RKEY = recordKeyOf("self"); - -export const tombstoneRecord = (now: Date = new Date()): OrgTangledShTombstone.Main => ({ - $type: "org.tangled.shTombstone", - createdAt: datetimeOf(now.toISOString()) -}); diff --git a/web/src/lib/oauth-client-metadata.ts b/web/src/lib/oauth-client-metadata.ts index 63fa5521f..0fb4de64d 100644 --- a/web/src/lib/oauth-client-metadata.ts +++ b/web/src/lib/oauth-client-metadata.ts @@ -23,7 +23,6 @@ const scopes = [ "repo:org.tangled.repo.manifest", "repo:org.tangled.repo.review.unspecced.round", "repo:org.tangled.review.patch", - "repo:org.tangled.shTombstone", "repo:org.tangled.track.ticket", "repo:org.tangled.track.unspecced.milestone", "repo:sh.tangled.repo.issue", @@ -79,6 +78,7 @@ const scopes = [ "rpc:farm.tranquil.delegation.createAccount?aud=*", "rpc:farm.tranquil.delegation.listControlledAccounts?aud=*", "rpc:org.tangled.feed.getTimeline?aud=*", + "rpc:org.tangled.foster.unmarkLiarAccount?aud=*", "rpc:org.tangled.graph.listNetworkVouches?aud=*", "rpc:org.tangled.repo.createRepo?aud=*", "rpc:org.tangled.repo.deleteRepo?aud=*", diff --git a/web/src/lib/server/config.ts b/web/src/lib/server/config.ts index 7a96d0b25..3179f0df5 100644 --- a/web/src/lib/server/config.ts +++ b/web/src/lib/server/config.ts @@ -69,7 +69,7 @@ export type WebConfig = { liarPdsUrl: string; /** empty turns foster read routing off */ fosterKnotUrl: KnotOrigin | ""; - /** fakes the liar pds and hydrant's tombstone pickup */ + /** fakes the liar pds and hydrant's unmark */ liarMock: boolean; };