adding PDS level backfill for future relay banned user bypass

.claude/settings.local.json

···
       "Bash(npm:*)",
       "Bash(docker-compose restart:*)",
       "Bash(echo \"# Bluesky Branding Removal Checklist\n\n## 1. App Icons & Images\n- assets/app-icons/*.png - App icons (replace with Aurora Prism branding)\n- assets/favicon.png - Browser favicon\n- assets/icon-android-*.png - Android icons\n- assets/default-avatar.png - Default avatar image\n\n## 2. App Metadata\n- app.json - App name, slug, description\n- package.json - App name and description\n\n## 3. Text References (276 occurrences)\n- Onboarding screens (src/screens/Onboarding/)\n- Signup screens (src/screens/Signup/)\n- Settings/About screens\n- Terms of Service / Privacy Policy references\n- Help text and tooltips\n- Error messages mentioning Bluesky\n\n## 4. URLs\n- bsky.app references (feed URLs, profile URLs)\n- bsky.social references\n- Links to Bluesky support/help\n\n## 5. Service Names\n- Bluesky Moderation Service references\n- Default feed generator names\n\nTotal: 276 text references found\")",
-      "Bash(psql \"$DATABASE_URL\" -c \"SELECT \n  (SELECT COUNT(*) FROM users) as users,\n  (SELECT COUNT(*) FROM posts) as posts,\n  (SELECT COUNT(*) FROM likes) as likes,\n  (SELECT COUNT(*) FROM reposts) as reposts,\n  (SELECT COUNT(*) FROM follows) as follows,\n  (SELECT COUNT(*) FROM blocks) as blocks;\")"
+      "Bash(psql \"$DATABASE_URL\" -c \"SELECT \n  (SELECT COUNT(*) FROM users) as users,\n  (SELECT COUNT(*) FROM posts) as posts,\n  (SELECT COUNT(*) FROM likes) as likes,\n  (SELECT COUNT(*) FROM reposts) as reposts,\n  (SELECT COUNT(*) FROM follows) as follows,\n  (SELECT COUNT(*) FROM blocks) as blocks;\")",
+      "Bash(dig +short TXT _atproto.spacelawshitpost.me)",
+      "Bash(python3 -m json.tool)"
     ],
     "deny": [],
     "ask": []

.env.example

···
 # AppView Configuration
 # Copy this to .env and customize as needed
 
+# Relay/Firehose Configuration
+# Primary relay source (usually Bluesky's main relay)
+# RELAY_URL=wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos
+
+# Additional relay sources (comma-separated)
+# Use this to crawl independent PDS instances that aren't in Bluesky's relay
+# Example: atproto.africa hosts blacksky.app and other independent instances
+# ADDITIONAL_RELAY_URLS=wss://atproto.africa/xrpc/com.atproto.sync.subscribeRepos
+
 # Constellation Integration (enhanced stats from microcosm.blue)
 # Set to false to disable and avoid timeout errors
 CONSTELLATION_ENABLED=false

ON_DEMAND_PDS_BACKFILL.md

···
+# On-Demand PDS Backfill
+
+This feature allows your AppView to automatically fetch users from independent PDS instances that aren't federated to Bluesky's relay.
+
+## How It Works
+
+### Automatic Backfill
+When a user tries to view a profile that doesn't exist in your AppView:
+1. The system detects the 404
+2. Resolves the user's DID to find their PDS (via plc.directory)
+3. Fetches their profile and recent content directly from their PDS
+4. Indexes it into your AppView
+5. Future requests will return the cached data
+
+### Manual Backfill (Admin Panel)
+You can also manually trigger backfills via the admin API:
+
+```bash
+# Trigger backfill for a specific DID
+curl -X POST https://your-appview.com/api/admin/backfill/pds \
+  -H "Content-Type: application/json" \
+  -d '{"did": "did:plc:63hvnyjvqi2nzzcsjgnry5we"}'
+
+# Check backfill status
+curl https://your-appview.com/api/admin/backfill/pds/status
+```
+
+## Example: Backfilling spacelawshitpost.me
+
+The user `spacelawshitpost.me` is on blacksky.app PDS and isn't in Bluesky's relay.
+
+### Automatic Method
+Just try to view their profile in your AppView:
+```
+GET /xrpc/app.bsky.actor.getProfile?actor=did:plc:63hvnyjvqi2nzzcsjgnry5we
+```
+
+First request: Returns 404 with message "Attempting to fetch from their PDS"
+Wait 5-10 seconds, then retry
+Second request: Returns full profile!
+
+### Manual Method
+```bash
+curl -X POST https://appview.dollspace.gay/api/admin/backfill/pds \
+  -H "Content-Type: application/json" \
+  -d '{"did": "did:plc:63hvnyjvqi2nzzcsjgnry5we"}'
+```
+
+## Rate Limiting
+
+- **Cooldown**: 5 minutes per DID (won't re-backfill the same user more frequently)
+- **Record Limit**: Maximum 1000 records per collection (prevents abuse)
+
+## Collections Backfilled
+
+When a user is backfilled, the system fetches:
+- `app.bsky.actor.profile` - Profile info
+- `app.bsky.feed.post` - Posts
+- `app.bsky.feed.like` - Likes
+- `app.bsky.feed.repost` - Reposts
+- `app.bsky.graph.follow` - Follows
+- `app.bsky.graph.block` - Blocks
+- And any other collections the PDS returns
+
+## Monitoring
+
+Check the server logs for backfill progress:
+```
+[ON_DEMAND_BACKFILL] Starting backfill for did:plc:...
+[ON_DEMAND_BACKFILL] did:plc:... is on PDS: blacksky.app
+[ON_DEMAND_BACKFILL] Backfilling spacelawshitpost.me from blacksky.app
+[ON_DEMAND_BACKFILL] Collections: app.bsky.actor.profile, app.bsky.feed.post, ...
+[ON_DEMAND_BACKFILL] Backfilled 42 records from app.bsky.feed.post
+[ON_DEMAND_BACKFILL] Completed backfill for did:plc:...
+```
+
+## Why Is This Needed?
+
+Some users are on independent PDS instances that:
+1. Aren't federated to Bluesky's main relay
+2. Require authentication for firehose access
+3. Are on relays that are currently offline (like atproto.africa)
+
+This on-demand system ensures your AppView can still serve these users when requested, without continuously polling or maintaining permanent connections to every independent PDS.

server/routes.ts

···
       `[FIREHOSE] Worker ${workerId}/${totalWorkers} - TypeScript firehose writer (firehose → Redis)`
     );
     firehoseClient.connect(workerId, totalWorkers);
+
+    // Connect to additional relay sources (blacksky.app, etc.)
+    const { initializeAdditionalRelays, additionalRelays } = require('./services/firehose');
+    initializeAdditionalRelays();
+
+    if (additionalRelays.length > 0) {
+      console.log(`[FIREHOSE] Connecting to ${additionalRelays.length} additional relay sources...`);
+      additionalRelays.forEach((relay: any, index: number) => {
+        relay.connect(workerId, totalWorkers);
+        console.log(`[FIREHOSE] Additional relay ${index + 1} connected`);
+      });
+    }
   } else if (!firehoseEnabled) {
     console.log(
       `[FIREHOSE] TypeScript firehose disabled (using Python firehose)`
···
         'TypeScript backfill has been disabled. Please use the Python backfill service instead.',
       info: 'Set BACKFILL_DAYS environment variable and run the Python unified worker.',
     });
+  });
+
+  // On-demand PDS backfill endpoint
+  app.post('/api/admin/backfill/pds', async (req, res) => {
+    try {
+      const { did } = req.body;
+
+      if (!did || !did.startsWith('did:')) {
+        res.status(400).json({ error: 'Invalid DID provided' });
+        return;
+      }
+
+      const { onDemandBackfill } = require('./services/on-demand-backfill');
+
+      // Trigger backfill (non-blocking)
+      onDemandBackfill.backfillUser(did).catch((error: any) => {
+        console.error(`[API] On-demand backfill failed for ${did}:`, error);
+      });
+
+      res.json({
+        success: true,
+        message: `Backfill started for ${did}. Check logs for progress.`,
+      });
+    } catch (error) {
+      console.error('[API] Error triggering on-demand backfill:', error);
+      res.status(500).json({ error: 'Failed to trigger backfill' });
+    }
+  });
+
+  // Get on-demand backfill status
+  app.get('/api/admin/backfill/pds/status', async (_req, res) => {
+    try {
+      const { onDemandBackfill } = require('./services/on-demand-backfill');
+      const status = onDemandBackfill.getStatus();
+      res.json(status);
+    } catch (error) {
+      console.error('[API] Error getting backfill status:', error);
+      res.status(500).json({ error: 'Failed to get status' });
+    }
   });
 
   // XRPC API Endpoints

server/services/firehose.ts

···
 }
 
 export const firehoseClient = new FirehoseClient();
+
+// Additional relay sources for multi-relay support
+export const additionalRelays: FirehoseClient[] = [];
+
+// Initialize additional relay clients from environment variable
+export function initializeAdditionalRelays() {
+  const additionalRelayUrls = process.env.ADDITIONAL_RELAY_URLS?.split(',').map(url => url.trim()).filter(Boolean) || [];
+
+  if (additionalRelayUrls.length > 0) {
+    console.log(`[FIREHOSE] Initializing ${additionalRelayUrls.length} additional relay sources:`, additionalRelayUrls);
+
+    for (const relayUrl of additionalRelayUrls) {
+      const client = new FirehoseClient(relayUrl);
+      additionalRelays.push(client);
+      console.log(`[FIREHOSE] Additional relay registered: ${relayUrl}`);
+    }
+  }
+
+  return additionalRelays;
+}

server/services/on-demand-backfill.ts

···
+import { logCollector } from './log-collector';
+import { redisQueue } from './redis-queue';
+import { metricsService } from './metrics';
+
+interface BackfillJob {
+  did: string;
+  pdsUrl: string;
+  timestamp: number;
+}
+
+export class OnDemandBackfill {
+  private activeJobs: Map<string, BackfillJob> = new Map();
+  private recentlyBackfilled: Set<string> = new Set();
+  private readonly BACKFILL_COOLDOWN = 5 * 60 * 1000; // Don't re-backfill same DID for 5 minutes
+
+  /**
+   * Backfill a user from their PDS when they're not found in our AppView
+   */
+  async backfillUser(did: string): Promise<boolean> {
+    // Check if we're already backfilling this DID
+    if (this.activeJobs.has(did)) {
+      console.log(`[ON_DEMAND_BACKFILL] Already backfilling ${did}`);
+      return false;
+    }
+
+    // Check cooldown - don't spam backfill the same user
+    if (this.recentlyBackfilled.has(did)) {
+      console.log(`[ON_DEMAND_BACKFILL] ${did} recently backfilled, skipping`);
+      return false;
+    }
+
+    try {
+      console.log(`[ON_DEMAND_BACKFILL] Starting backfill for ${did}`);
+      logCollector.info(`On-demand backfill started for ${did}`);
+
+      // First, resolve the DID to find their PDS
+      const pdsUrl = await this.resolvePDS(did);
+
+      if (!pdsUrl) {
+        console.warn(`[ON_DEMAND_BACKFILL] Could not resolve PDS for ${did}`);
+        return false;
+      }
+
+      console.log(`[ON_DEMAND_BACKFILL] ${did} is on PDS: ${pdsUrl}`);
+
+      // Mark as active
+      this.activeJobs.set(did, {
+        did,
+        pdsUrl,
+        timestamp: Date.now(),
+      });
+
+      // Perform the backfill
+      await this.performBackfill(did, pdsUrl);
+
+      // Mark as recently backfilled (with cooldown)
+      this.recentlyBackfilled.add(did);
+      setTimeout(() => {
+        this.recentlyBackfilled.delete(did);
+      }, this.BACKFILL_COOLDOWN);
+
+      // Remove from active jobs
+      this.activeJobs.delete(did);
+
+      console.log(`[ON_DEMAND_BACKFILL] Completed backfill for ${did}`);
+      logCollector.success(`On-demand backfill completed for ${did}`);
+
+      return true;
+    } catch (error) {
+      console.error(`[ON_DEMAND_BACKFILL] Error backfilling ${did}:`, error);
+      logCollector.error(`On-demand backfill failed for ${did}`, { error });
+      this.activeJobs.delete(did);
+      metricsService.incrementError();
+      return false;
+    }
+  }
+
+  private async resolvePDS(did: string): Promise<string | null> {
+    try {
+      // Fetch DID document from PLC directory
+      const plcUrl = `https://plc.directory/${did}`;
+      const response = await fetch(plcUrl);
+
+      if (!response.ok) {
+        console.warn(`[ON_DEMAND_BACKFILL] Failed to resolve DID ${did}: ${response.status}`);
+        return null;
+      }
+
+      const didDoc = await response.json();
+
+      // Extract PDS service endpoint
+      const pdsService = didDoc.service?.find(
+        (s: any) => s.type === 'AtprotoPersonalDataServer'
+      );
+
+      if (!pdsService?.serviceEndpoint) {
+        console.warn(`[ON_DEMAND_BACKFILL] No PDS service found in DID document for ${did}`);
+        return null;
+      }
+
+      // Remove https:// prefix to get just the hostname
+      const pdsUrl = pdsService.serviceEndpoint.replace(/^https?:\/\//, '');
+
+      return pdsUrl;
+    } catch (error) {
+      console.error(`[ON_DEMAND_BACKFILL] Error resolving PDS for ${did}:`, error);
+      return null;
+    }
+  }
+
+  private async performBackfill(did: string, pdsUrl: string) {
+    try {
+      // First, get repo description
+      const describeUrl = `https://${pdsUrl}/xrpc/com.atproto.repo.describeRepo?repo=${did}`;
+      const describeResponse = await fetch(describeUrl);
+
+      if (!describeResponse.ok) {
+        throw new Error(`Failed to describe repo: ${describeResponse.status}`);
+      }
+
+      const describeData = await describeResponse.json();
+      const handle = describeData.handle;
+      const collections = describeData.collections || [];
+
+      console.log(`[ON_DEMAND_BACKFILL] Backfilling ${handle} (${did}) from ${pdsUrl}`);
+      console.log(`[ON_DEMAND_BACKFILL] Collections: ${collections.join(', ')}`);
+
+      // Process handle/identity first
+      await redisQueue.push({
+        type: 'identity',
+        data: {
+          did: did,
+          handle: handle,
+        },
+      });
+
+      metricsService.incrementEvent('#identity');
+
+      // Backfill each collection
+      for (const collection of collections) {
+        await this.backfillCollection(did, pdsUrl, collection);
+      }
+
+      console.log(`[ON_DEMAND_BACKFILL] Backfilled all collections for ${did}`);
+    } catch (error) {
+      console.error(`[ON_DEMAND_BACKFILL] Error during backfill:`, error);
+      throw error;
+    }
+  }
+
+  private async backfillCollection(did: string, pdsUrl: string, collection: string) {
+    try {
+      let cursor: string | undefined = undefined;
+      let totalRecords = 0;
+
+      do {
+        // Fetch records from this collection
+        const listUrl = `https://${pdsUrl}/xrpc/com.atproto.repo.listRecords?repo=${did}&collection=${collection}&limit=100${cursor ? `&cursor=${cursor}` : ''}`;
+        const response = await fetch(listUrl);
+
+        if (!response.ok) {
+          if (response.status === 400) {
+            // Collection doesn't exist, skip it
+            return;
+          }
+          throw new Error(`Failed to list ${collection}: ${response.status}`);
+        }
+
+        const data = await response.json();
+        const records = data.records || [];
+
+        if (records.length === 0) {
+          break;
+        }
+
+        // Process each record
+        for (const record of records) {
+          const path = record.uri.split('/').slice(-2).join('/'); // collection/rkey
+
+          const commit = {
+            repo: did,
+            ops: [
+              {
+                action: 'create',
+                path: path,
+                cid: record.cid,
+                record: record.value,
+              },
+            ],
+          };
+
+          // Push to Redis queue for processing
+          await redisQueue.push({
+            type: 'commit',
+            data: commit,
+            seq: undefined,
+          });
+
+          metricsService.incrementEvent('#commit');
+          totalRecords++;
+        }
+
+        cursor = data.cursor;
+
+        // Don't backfill more than 1000 records per collection (prevent abuse)
+        if (totalRecords >= 1000) {
+          console.log(`[ON_DEMAND_BACKFILL] Reached limit of 1000 records for ${collection}, stopping`);
+          break;
+        }
+
+      } while (cursor);
+
+      if (totalRecords > 0) {
+        console.log(`[ON_DEMAND_BACKFILL] Backfilled ${totalRecords} records from ${collection}`);
+      }
+
+    } catch (error) {
+      console.error(`[ON_DEMAND_BACKFILL] Error backfilling collection ${collection}:`, error);
+      throw error;
+    }
+  }
+
+  getStatus() {
+    return {
+      activeJobs: Array.from(this.activeJobs.values()),
+      recentlyBackfilled: this.recentlyBackfilled.size,
+      cooldownMs: this.BACKFILL_COOLDOWN,
+    };
+  }
+}
+
+export const onDemandBackfill = new OnDemandBackfill();

server/services/xrpc/services/actor-service.ts

···
   suggestedUsersUnspeccedSchema,
 } from '../schemas';
 import { xrpcApi } from '../../xrpc-api';
+import { onDemandBackfill } from '../../on-demand-backfill';
 
 /**
  * Get a single actor profile
···
     const profiles = await (xrpcApi as any)._getProfiles([params.actor], req);
 
     if (profiles.length === 0) {
+      // Profile not found - trigger on-demand backfill from their PDS
+      const actor = params.actor;
+
+      // Check if it's a DID (not a handle)
+      if (actor.startsWith('did:')) {
+        console.log(`[ON_DEMAND] Profile ${actor} not found, triggering backfill...`);
+
+        // Trigger backfill (non-blocking - don't wait for it)
+        onDemandBackfill.backfillUser(actor).catch(error => {
+          console.error(`[ON_DEMAND] Backfill failed for ${actor}:`, error);
+        });
+
+        // Return 404 with a helpful message
+        res.status(404).json({
+          error: 'ProfileNotFound',
+          message: 'Profile not found. Attempting to fetch from their PDS - try again in a few seconds.',
+        });
+        return;
+      }
+
+      // For handles, we'd need to resolve to DID first
       res.status(404).json({ error: 'Profile not found' });
       return;
     }