Connect linkblogs to existing publications

backend/migrations/0060_linkblog_publication.sql

@@ -0,0 +1,4 @@
+-- Optional existing standard.site publication used for new linkblog posts.
+-- NULL means Skyreader's managed skyreader-links publication.
+ALTER TABLE user_settings ADD COLUMN linkblog_publication TEXT;
+ALTER TABLE user_settings ADD COLUMN linkblog_content_format TEXT;

backend/migrations/0061_subscription_previous_feed_url.sql

@@ -0,0 +1,4 @@
+-- Preserve the prior publication URI while a linkblog follower is being moved.
+-- Atmospheric reconciliation uses it to delete the old portable graph edge
+-- before confirming the replacement edge.
+ALTER TABLE subscriptions_cache ADD COLUMN atmosphere_previous_feed_url TEXT DEFAULT NULL;

backend/migrations/0062_subscription_site_url.sql

@@ -0,0 +1,9 @@
+-- Persist a subscription's siteUrl (the human-facing home page of the source).
+--
+-- It was only ever mirrored to the PDS record, so it existed on the device that
+-- created the subscription and nowhere else. Linkblogs made that load-bearing: a
+-- linkblog connected to an existing publication has an arbitrary rkey, so the
+-- publication URI alone can't say "this is a linkblog" — the author's public
+-- linkblog page, stored here, is the tell. Without it a followed linkblog reads
+-- as a generic "Blog" on any device that didn't create the follow.
+ALTER TABLE subscriptions_cache ADD COLUMN site_url TEXT DEFAULT NULL;

backend/src/index.ts

@@ -25,6 +25,9 @@ import {
   handleUpdatePublication,
   handleDiscover,
   handleDiscoverFriends,
+  handleListPublications,
+  handleConnectPublication,
+  handleResolvePublication,
 } from './routes/linkblog';
 import { handleAtmosphereSubscription } from './routes/atmosphere';
 import {
@@ -289,6 +292,17 @@ export default {
             response = await handleUpdatePublication(request, env);
           }
           break;
+        case url.pathname === '/api/linkblog/publications':
+          if (!session) return unauthorizedResponse(headers);
+          response = await handleListPublications(request, env);
+          break;
+        case url.pathname === '/api/linkblog/publication/connect':
+          if (!session) return unauthorizedResponse(headers);
+          response = await handleConnectPublication(request, env);
+          break;
+        case url.pathname.startsWith('/api/linkblog/resolve/'):
+          response = await handleResolvePublication(request, env);
+          break;
 
         // Subscribe via the Atmosphere — writes the portable
         // site.standard.graph.subscription record, and (for a signed-in user)

backend/src/routes/feeds-v2.ts

@@ -8,6 +8,10 @@ import type {
 } from '../services/feed-proxy-client';
 import { resolveStandardSite } from '../utils/canonical-url';
 import { getReadKeys } from './reading';
+import {
+  getLinkblogTargets,
+  publicationUri as linkblogPublicationUri,
+} from '../services/linkblog-sync';
 
 interface V2FeedResponse {
   title: string;
@@ -317,6 +321,92 @@ interface V2BatchDocumentResponse {
   readCursor?: number;
 }
 
+interface DocumentScopeRequest {
+  did: string;
+  siteUri?: string;
+  since_digest?: string;
+}
+
+// The publication scopes this user's own subscription rows point at, per author.
+// That's the migrated truth: when an author connects (or disconnects) a
+// publication, migrateLinkblogFollowers rewrites these rows immediately, while a
+// follower's device keeps requesting whatever scope it cached.
+async function subscribedDocumentScopes(
+  env: Env,
+  userDid: string,
+  dids: string[]
+): Promise<Map<string, Set<string>>> {
+  const out = new Map<string, Set<string>>();
+  if (dids.length === 0) return out;
+  const placeholders = dids.map(() => '?').join(',');
+  const rows = await env.DB.prepare(
+    `SELECT subject_did, feed_url FROM subscriptions_cache
+     WHERE user_did = ? AND source_type = 'atproto.documents' AND subject_did IN (${placeholders})`
+  )
+    .bind(userDid, ...dids)
+    .all<{ subject_did: string; feed_url: string | null }>();
+  for (const row of rows.results ?? []) {
+    if (!row.feed_url) continue;
+    const scopes = out.get(row.subject_did) ?? new Set<string>();
+    scopes.add(row.feed_url);
+    out.set(row.subject_did, scopes);
+  }
+  return out;
+}
+
+/**
+ * Re-point requests that name a linkblog publication its author has moved off.
+ *
+ * A follower's scope lives on their device (the local subscription's feedUrl) and
+ * in their PDS record, neither of which we can rewrite synchronously when an
+ * author connects an existing publication. Left alone, that client asks for the
+ * abandoned publication forever: the proxy filters by site URI, so the feed just
+ * stops updating — no error, no signal. Correcting it here means every client,
+ * on every device, self-heals on its next poll.
+ *
+ * Deliberately narrow. A scope is only stale when it names the author's own
+ * Skyreader publication (`skyreader-links`, unambiguously a linkblog follow) or
+ * when this user's subscription row has already been migrated to the author's
+ * current target. An ordinary publication subscription is never touched.
+ *
+ * Returns the requests to forward plus a `did\ncorrectedScope → requestedScope`
+ * map, so the response still echoes the scope the client asked for and its
+ * per-scope digests/reconciliation keys stay stable.
+ */
+async function correctLinkblogScopes(
+  env: Env,
+  userDid: string,
+  entries: DocumentScopeRequest[]
+): Promise<{ requests: DocumentScopeRequest[]; restore: Map<string, string> }> {
+  const restore = new Map<string, string>();
+  // Own-linkblog pulls resolve their own target client-side, and an unscoped
+  // request already gets everything the author wrote.
+  const scoped = entries.filter((e) => e.siteUri && e.did !== userDid);
+  if (scoped.length === 0) return { requests: entries, restore };
+
+  const dids = [...new Set(scoped.map((e) => e.did))];
+  const [targets, subscribed] = await Promise.all([
+    getLinkblogTargets(env, dids),
+    subscribedDocumentScopes(env, userDid, dids),
+  ]);
+  const requested = new Set(entries.map((e) => `${e.did}\n${e.siteUri ?? ''}`));
+
+  const requests = entries.map((entry) => {
+    if (!entry.siteUri || entry.did === userDid) return entry;
+    const target = targets.get(entry.did);
+    if (!target || target.siteUri === entry.siteUri) return entry;
+    const rows = subscribed.get(entry.did);
+    const stale =
+      entry.siteUri === linkblogPublicationUri(entry.did) ||
+      (!!rows && !rows.has(entry.siteUri) && rows.has(target.siteUri));
+    // Don't collapse two requested scopes onto one — the client asked for both.
+    if (!stale || requested.has(`${entry.did}\n${target.siteUri}`)) return entry;
+    restore.set(`${entry.did}\n${target.siteUri}`, entry.siteUri);
+    return { ...entry, siteUri: target.siteUri };
+  });
+  return { requests, restore };
+}
+
 /**
  * POST /api/v2/documents/batch
  *
@@ -406,8 +496,23 @@ export async function handleV2BatchDocumentFetch(
   }
 
   try {
+    // Re-point any scope whose author has since moved their linkblog, echoing the
+    // requested scope back so the client's keys are untouched. Best-effort: a D1
+    // hiccup here must not cost the user their documents.
+    let requests = valid;
+    let restore = new Map<string, string>();
+    try {
+      ({ requests, restore } = await correctLinkblogScopes(env, session.did, valid));
+    } catch (e) {
+      console.error('Linkblog scope correction failed; forwarding scopes as-is:', e);
+    }
+
     const client = new FeedProxyClient(env);
-    const proxyEntries = await client.fetchDocumentsBatch(valid);
+    const proxyEntries = await client.fetchDocumentsBatch(requests);
+    for (const entry of proxyEntries) {
+      const requestedScope = entry.siteUri && restore.get(`${entry.did}\n${entry.siteUri}`);
+      if (requestedScope) entry.siteUri = requestedScope;
+    }
     authors.push(...proxyEntries);
 
     // Inline read annotation, identical to the feed path but keyed by recordUri

backend/src/routes/linkblog.ts

@@ -9,13 +9,18 @@ import { hasRequiredScopes, insufficientScopesResponse, LINKBLOG_SCOPES } from '
 import { isValidRkey, invalidRkeyResponse } from '../utils/validation';
 import {
   deleteLinkblogShare,
+  FOREIGN_RECORD_ERROR,
   getPublicationMeta,
+  getLinkblogTarget,
+  linkblogBaseUrl,
   publicationUri,
   updateLinkblogShareNote,
   updatePublication,
   writeLinkblogShare,
   type LinkblogShareInput,
+  type ContentFormat,
 } from '../services/linkblog-sync';
+import { createPDSClient } from '../services/pds-client';
 import { getLinkblogDiscover, getLinkblogFriends } from '../services/linkblog-discovery';
 
 function json(body: unknown, status = 200): Response {
@@ -106,7 +111,7 @@ export async function handleCreateLinkblogShare(request: Request, env: Env): Pro
     uri: result.data.uri,
     cid: result.data.cid,
     rkey,
-    publication: publicationUri(session.did),
+    publication: (await getLinkblogTarget(env, session.did)).siteUri,
   });
 }
 
@@ -142,6 +147,8 @@ export async function handleUpdateLinkblogShare(request: Request, env: Env): Pro
   const result = await updateLinkblogShareNote(session, rkey, body.note);
   if (!result.success) {
     if (isScopeError(result.error)) return insufficientScopesResponse();
+    // Not a PDS failure: the record isn't ours to rewrite.
+    if (result.error === FOREIGN_RECORD_ERROR) return json({ error: result.error }, 409);
     return json({ error: result.error }, result.retryable ? 503 : 502);
   }
   return json({ uri: result.data.uri, cid: result.data.cid, rkey });
@@ -168,6 +175,8 @@ export async function handleDeleteLinkblogShare(request: Request, env: Env): Pro
   const result = await deleteLinkblogShare(session, rkey);
   if (!result.success) {
     if (isScopeError(result.error)) return insufficientScopesResponse();
+    // Not a PDS failure: the record isn't ours to delete.
+    if (result.error === FOREIGN_RECORD_ERROR) return json({ error: result.error }, 409);
     return json({ error: result.error }, result.retryable ? 503 : 502);
   }
   return json({ success: true });
@@ -215,6 +224,9 @@ export async function handleUpdatePublication(request: Request, env: Env): Promi
   if (!hasRequiredScopes(session.grantedScopes, LINKBLOG_SCOPES)) {
     return insufficientScopesResponse();
   }
+  if ((await getLinkblogTarget(env, session.did)).external) {
+    return json({ error: 'This publication is managed by its home app' }, 409);
+  }
 
   let body: { name?: string; description?: string };
   try {
@@ -242,3 +254,166 @@ export async function handleUpdatePublication(request: Request, env: Env): Promi
   const meta = await getPublicationMeta(session, env);
   return json(meta);
 }
+
+const FORMATS = new Set<ContentFormat>(['leaflet', 'pckt', 'offprint', 'markpub']);
+
+export async function migrateLinkblogFollowers(
+  env: Env,
+  subjectDid: string,
+  previousSiteUri: string,
+  nextSiteUri: string
+): Promise<void> {
+  if (previousSiteUri === nextSiteUri) return;
+
+  // Heal the follower's siteUrl while we're here. The author's public linkblog
+  // page is what tells the reader this publication is a linkblog once its rkey is
+  // no longer `skyreader-links` (see sourceDisplay); rows created before we
+  // persisted it have none. COALESCE so a user-set value is never overwritten.
+  const linkblogPage = linkblogBaseUrl(env, subjectDid);
+
+  // A follower may already subscribe to the destination publication. Keep that
+  // row and use it to reconcile the old graph edge, then remove the redundant
+  // source row. Otherwise move the source row in place.
+  await env.DB.batch([
+    env.DB.prepare(
+      `UPDATE subscriptions_cache AS destination
+       SET atmosphere_previous_feed_url = COALESCE(
+             (SELECT source.atmosphere_previous_feed_url
+              FROM subscriptions_cache AS source
+              WHERE source.user_did = destination.user_did
+                AND source.source_type = 'atproto.documents'
+                AND source.subject_did = ? AND source.feed_url = ?),
+             destination.atmosphere_previous_feed_url,
+             ?
+           ),
+           site_url = COALESCE(site_url, ?),
+           atmosphere_synced = NULL
+       WHERE source_type = 'atproto.documents' AND subject_did = ? AND feed_url = ?
+         AND EXISTS (
+           SELECT 1 FROM subscriptions_cache AS source
+           WHERE source.user_did = destination.user_did
+             AND source.source_type = 'atproto.documents'
+             AND source.subject_did = ? AND source.feed_url = ?
+         )`
+    ).bind(
+      subjectDid,
+      previousSiteUri,
+      previousSiteUri,
+      linkblogPage,
+      subjectDid,
+      nextSiteUri,
+      subjectDid,
+      previousSiteUri
+    ),
+    env.DB.prepare(
+      `DELETE FROM subscriptions_cache AS source
+       WHERE source_type = 'atproto.documents' AND subject_did = ? AND feed_url = ?
+         AND EXISTS (
+           SELECT 1 FROM subscriptions_cache AS destination
+           WHERE destination.user_did = source.user_did
+             AND destination.source_type = 'atproto.documents'
+             AND destination.subject_did = ? AND destination.feed_url = ?
+         )`
+    ).bind(subjectDid, previousSiteUri, subjectDid, nextSiteUri),
+    env.DB.prepare(
+      `UPDATE subscriptions_cache
+       SET feed_url = ?,
+           atmosphere_previous_feed_url = COALESCE(atmosphere_previous_feed_url, ?),
+           site_url = COALESCE(site_url, ?),
+           atmosphere_synced = NULL
+       WHERE source_type = 'atproto.documents' AND subject_did = ? AND feed_url = ?`
+    ).bind(nextSiteUri, previousSiteUri, linkblogPage, subjectDid, previousSiteUri),
+  ]);
+}
+
+export async function handleListPublications(request: Request, env: Env): Promise<Response> {
+  if (request.method !== 'GET') return json({ error: 'Method not allowed' }, 405);
+  const session = await getSessionFromRequest(request, env);
+  if (!session) return json({ error: 'Unauthorized' }, 401);
+  const result = await createPDSClient(session).listAllRecords<{ name?: string; url?: string }>(
+    'site.standard.publication',
+    { maxPages: 5, maxRecords: 200 }
+  );
+  if (!result.success) return json({ error: result.error }, 502);
+
+  const defaultUri = publicationUri(session.did);
+  const publications = result.data.map((record) => ({
+    uri: record.uri,
+    rkey: record.uri.split('/').pop(),
+    name: record.value.name || 'Untitled publication',
+    url: record.value.url,
+    isDefault: record.uri === defaultUri,
+  }));
+  // The Skyreader linkblog is always offered, even before its record exists — it's
+  // created lazily on first share, and a user who connected an external
+  // publication without ever sharing would otherwise have no way back. Choosing it
+  // hits the disconnect path, which works whether or not the record is there.
+  if (!publications.some((p) => p.isDefault)) {
+    publications.unshift({
+      uri: defaultUri,
+      rkey: defaultUri.split('/').pop(),
+      name: 'Skyreader linkblog',
+      url: undefined,
+      isDefault: true,
+    });
+  }
+  return json({ publications });
+}
+
+export async function handleConnectPublication(request: Request, env: Env): Promise<Response> {
+  const session = await getSessionFromRequest(request, env);
+  if (!session) return json({ error: 'Unauthorized' }, 401);
+  const now = Math.floor(Date.now() / 1000);
+  if (request.method === 'DELETE') {
+    const previousTarget = await getLinkblogTarget(env, session.did);
+    const nextSiteUri = publicationUri(session.did);
+    await env.DB.prepare(
+      'UPDATE user_settings SET linkblog_publication = NULL, linkblog_content_format = NULL, updated_at = ? WHERE user_did = ?'
+    )
+      .bind(now, session.did)
+      .run();
+    await migrateLinkblogFollowers(env, session.did, previousTarget.siteUri, nextSiteUri);
+    return json(await getPublicationMeta(session, env));
+  }
+  if (request.method !== 'PUT') return json({ error: 'Method not allowed' }, 405);
+  let body: { publicationUri?: string; format?: ContentFormat };
+  try {
+    body = await request.json();
+  } catch {
+    return json({ error: 'Invalid JSON body' }, 400);
+  }
+  const match = body.publicationUri?.match(
+    /^at:\/\/([^/]+)\/site\.standard\.publication\/([^/]+)$/
+  );
+  if (!match || match[1] !== session.did)
+    return json({ error: 'Choose a publication from your own Atmosphere account' }, 400);
+  const selectedPublicationUri = body.publicationUri!;
+  if (body.format && !FORMATS.has(body.format))
+    return json({ error: 'Unsupported content format' }, 400);
+  const exists = await createPDSClient(session).getRecord('site.standard.publication', match[2]);
+  if (!exists.success) return json({ error: 'Publication not found' }, 404);
+  const format = body.format || 'leaflet';
+  const previousTarget = await getLinkblogTarget(env, session.did);
+  await env.DB.prepare(
+    `INSERT INTO user_settings (user_did, linkblog_publication, linkblog_content_format, created_at, updated_at)
+    VALUES (?, ?, ?, ?, ?) ON CONFLICT(user_did) DO UPDATE SET linkblog_publication=excluded.linkblog_publication,
+    linkblog_content_format=excluded.linkblog_content_format, updated_at=excluded.updated_at`
+  )
+    .bind(session.did, selectedPublicationUri, format, now, now)
+    .run();
+  await migrateLinkblogFollowers(env, session.did, previousTarget.siteUri, selectedPublicationUri);
+  return json(await getPublicationMeta(session, env));
+}
+
+export async function handleResolvePublication(request: Request, env: Env): Promise<Response> {
+  if (request.method !== 'GET') return json({ error: 'Method not allowed' }, 405);
+  const did = decodeURIComponent(new URL(request.url).pathname.split('/').pop() || '');
+  if (!did.startsWith('did:')) return json({ error: 'Invalid DID' }, 400);
+  const target = await getLinkblogTarget(env, did);
+  return new Response(
+    JSON.stringify({ siteUri: target.siteUri, defaultSiteUri: publicationUri(did) }),
+    {
+      headers: { 'Content-Type': 'application/json', 'Cache-Control': 'public, max-age=60' },
+    }
+  );
+}