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/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/linkblog.ts

@@ -10,12 +10,15 @@ import { isValidRkey, invalidRkeyResponse } from '../utils/validation';
 import {
   deleteLinkblogShare,
   getPublicationMeta,
+  getLinkblogTarget,
   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 +109,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,
   });
 }
 
@@ -215,6 +218,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 +248,124 @@ 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;
+
+  // 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 = ?, 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(previousSiteUri, 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 = ?, atmosphere_synced = NULL
+       WHERE source_type = 'atproto.documents' AND subject_did = ? AND feed_url = ?`
+    ).bind(nextSiteUri, previousSiteUri, 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);
+  return json({
+    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 === publicationUri(session.did),
+    })),
+  });
+}
+
+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' },
+    }
+  );
+}

backend/src/services/atmosphere-subscription-sync.ts

@@ -33,6 +33,7 @@ import { resolvePdsUrl } from '../utils/did-resolver';
 import { generateTid } from '../utils/tid';
 import {
   SUBSCRIPTION_COLLECTION,
+  deleteAtmosphereSubscription,
   isPublicationUri,
   writeAtmosphereSubscription,
 } from './atmosphere-subscription';
@@ -75,6 +76,7 @@ interface LocalDocSub {
   record_uri: string;
   feed_url: string;
   atmosphere_synced: number | null;
+  atmosphere_previous_feed_url: string | null;
 }
 
 interface PublicationMeta {
@@ -184,16 +186,20 @@ export async function reconcileAtmosphereSubscriptions(
 
     // Step 2: local atproto.documents subs, keyed by publication URI (feedUrl).
     const localResult = await env.DB.prepare(
-      `SELECT record_uri, feed_url, atmosphere_synced
+      `SELECT record_uri, feed_url, atmosphere_synced, atmosphere_previous_feed_url
          FROM subscriptions_cache
         WHERE user_did = ? AND source_type = 'atproto.documents'`
     )
       .bind(session.did)
       .all<LocalDocSub>();
     const localSubs = localResult.results || [];
     const localByPub = new Map<string, LocalDocSub>();
+    const supersededGraphPubs = new Set<string>();
     for (const sub of localSubs) {
       if (isPublicationUri(sub.feed_url)) localByPub.set(sub.feed_url, sub);
+      if (isPublicationUri(sub.atmosphere_previous_feed_url)) {
+        supersededGraphPubs.add(sub.atmosphere_previous_feed_url);
+      }
     }
 
     // Tier-aware headroom for imports — counts ACTIVE subs only, since the limit
@@ -225,6 +231,10 @@ export async function reconcileAtmosphereSubscriptions(
     let droppedOverCap = 0;
     for (const pubUri of graphPubs) {
       if (localByPub.has(pubUri)) continue;
+      // A publication switch deliberately superseded this edge. Its destination
+      // row below owns deleting it; importing it here would resurrect the old
+      // follower scope and prevent the migration from converging.
+      if (supersededGraphPubs.has(pubUri)) continue;
       if (ops >= MAX_OPS) {
         result.hasMore = true;
         break;
@@ -323,6 +333,51 @@ export async function reconcileAtmosphereSubscriptions(
 
     // Step 4: reconcile each local pub-sub against the graph.
     for (const [pubUri, sub] of localByPub) {
+      if (sub.atmosphere_previous_feed_url) {
+        const previousPubUri = sub.atmosphere_previous_feed_url;
+        if (graphPubs.has(previousPubUri)) {
+          if (ops >= MAX_OPS) {
+            result.hasMore = true;
+            break;
+          }
+          const removeOld = await deleteAtmosphereSubscription(session, previousPubUri);
+          if (!removeOld.success) {
+            result.warnings.push(
+              `Failed to replace old publication follow ${previousPubUri}: ${removeOld.error}`
+            );
+            continue;
+          }
+          graphPubs.delete(previousPubUri);
+          ops++;
+        }
+
+        if (!graphPubs.has(pubUri)) {
+          if (ops >= MAX_OPS) {
+            result.hasMore = true;
+            break;
+          }
+          const writeNew = await writeAtmosphereSubscription(session, pubUri);
+          if (!writeNew.success) {
+            result.warnings.push(
+              `Failed to push replacement edge for ${pubUri}: ${writeNew.error}`
+            );
+            continue;
+          }
+          graphPubs.add(pubUri);
+          result.pushed++;
+          ops++;
+        }
+
+        await env.DB.prepare(
+          `UPDATE subscriptions_cache
+           SET atmosphere_synced = unixepoch(), atmosphere_previous_feed_url = NULL
+           WHERE record_uri = ?`
+        )
+          .bind(sub.record_uri)
+          .run();
+        continue;
+      }
+
       if (graphPubs.has(pubUri)) {
         // Present both places — claim it if not yet marked (e.g. an in-app follow).
         if (sub.atmosphere_synced === null) {