simplified jetstream

package-lock.json

···
       "version": "0.0.1",
       "dependencies": {
         "@astrojs/check": "^0.9.4",
+        "@atcute/jetstream": "^1.0.2",
         "@atproto/api": "^0.16.2",
         "@atproto/xrpc": "^0.7.1",
         "@nulfrost/leaflet-loader-astro": "^1.1.0",
···
       "dependencies": {
         "@atcute/lexicons": "^1.0.4",
         "@badrap/valita": "^0.4.5"
+      }
+    },
+    "node_modules/@atcute/jetstream": {
+      "version": "1.0.2",
+      "resolved": "https://registry.npmjs.org/@atcute/jetstream/-/jetstream-1.0.2.tgz",
+      "integrity": "sha512-ZtdNNxl4zq9cgUpXSL9F+AsXUZt0Zuyj0V7974D7LxdMxfTItPnMZ9dRG8GoFkkGz3+pszdsG888Ix8C0F2+mA==",
+      "license": "MIT",
+      "dependencies": {
+        "@atcute/lexicons": "^1.0.2",
+        "@badrap/valita": "^0.4.2",
+        "@mary-ext/event-iterator": "^1.0.0",
+        "@mary-ext/simple-event-emitter": "^1.0.0",
+        "partysocket": "^1.1.4",
+        "type-fest": "^4.41.0",
+        "yocto-queue": "^1.2.1"
       }
     },
     "node_modules/@atcute/lexicons": {
···
         "@jridgewell/sourcemap-codec": "^1.4.14"
       }
     },
+    "node_modules/@mary-ext/event-iterator": {
+      "version": "1.0.0",
+      "resolved": "https://registry.npmjs.org/@mary-ext/event-iterator/-/event-iterator-1.0.0.tgz",
+      "integrity": "sha512-l6gCPsWJ8aRCe/s7/oCmero70kDHgIK5m4uJvYgwEYTqVxoBOIXbKr5tnkLqUHEg6mNduB4IWvms3h70Hp9ADQ==",
+      "license": "BSD-3-Clause",
+      "dependencies": {
+        "yocto-queue": "^1.2.1"
+      }
+    },
+    "node_modules/@mary-ext/simple-event-emitter": {
+      "version": "1.0.0",
+      "resolved": "https://registry.npmjs.org/@mary-ext/simple-event-emitter/-/simple-event-emitter-1.0.0.tgz",
+      "integrity": "sha512-meA/zJZKIN1RVBNEYIbjufkUrW7/tRjHH60FjolpG1ixJKo76TB208qefQLNdOVDA7uIG0CGEDuhmMirtHKLAg==",
+      "license": "BSD-3-Clause"
+    },
     "node_modules/@nodelib/fs.scandir": {
       "version": "2.1.5",
       "resolved": "https://registry.npmjs.org/@nodelib/fs.scandir/-/fs.scandir-2.1.5.tgz",
···
       "dependencies": {
         "@types/estree": "^1.0.0"
       }
+    },
+    "node_modules/event-target-polyfill": {
+      "version": "0.0.4",
+      "resolved": "https://registry.npmjs.org/event-target-polyfill/-/event-target-polyfill-0.0.4.tgz",
+      "integrity": "sha512-Gs6RLjzlLRdT8X9ZipJdIZI/Y6/HhRLyq9RdDlCsnpxr/+Nn6bU2EFGuC94GjxqhM+Nmij2Vcq98yoHrU8uNFQ==",
+      "license": "MIT"
     },
     "node_modules/eventemitter3": {
       "version": "5.0.1",
···
       },
       "funding": {
         "url": "https://github.com/inikulin/parse5?sponsor=1"
+      }
+    },
+    "node_modules/partysocket": {
+      "version": "1.1.5",
+      "resolved": "https://registry.npmjs.org/partysocket/-/partysocket-1.1.5.tgz",
+      "integrity": "sha512-8uw9foq9bij4sKLCtTSHvyqMrMTQ5FJjrHc7BjoM2s95Vu7xYCN63ABpI7OZHC7ZMP5xaom/A+SsoFPXmTV6ZQ==",
+      "license": "MIT",
+      "dependencies": {
+        "event-target-polyfill": "^0.0.4"
       }
     },
     "node_modules/path-browserify": {

package.json

···
   },
   "dependencies": {
     "@astrojs/check": "^0.9.4",
+    "@atcute/jetstream": "^1.0.2",
     "@atproto/api": "^0.16.2",
     "@atproto/xrpc": "^0.7.1",
     "@nulfrost/leaflet-loader-astro": "^1.1.0",

src/components/content/ContentFeed.astro

···
 
       try {
         // Use shared jetstream instead of creating a new connection
-        const { startSharedStream, subscribeToPosts } = await import('../../lib/atproto/shared-jetstream');
+        const { startSharedStream, subscribeToPosts } = await import('../../lib/atproto/jetstream-client');
         
         // Start the shared stream
         await startSharedStream();
         
         // Subscribe to new posts
-        const unsubscribe = subscribeToPosts((record) => {
-          if (record.operation === 'create') {
-            const el = buildPostEl(record.value, record.did);
+        const unsubscribe = subscribeToPosts((event) => {
+          if (event.commit.operation === 'create') {
+            const el = buildPostEl(event.commit.record, event.did);
             // @ts-ignore
             container.insertBefore(el, container.firstChild);
             const posts = container.children;

src/components/content/StatusUpdate.astro

···
 </div>
 
 <script>
-  import { startSharedStream, subscribeToStatusUpdates } from '../../lib/atproto/shared-jetstream';
+  import { startSharedStream, subscribeToStatusUpdates } from '../../lib/atproto/jetstream-client';
   
   // Start the shared stream
   startSharedStream();
   
   // Subscribe to status updates
-  const unsubscribe = subscribeToStatusUpdates((record) => {
-    if (record.operation === 'create') {
-      updateStatusDisplay(record.value);
+  const unsubscribe = subscribeToStatusUpdates((event) => {
+    if (event.commit.operation === 'create') {
+      updateStatusDisplay(event.commit.record);
     }
   });
   

src/lib/atproto/jetstream-client.ts

···
-// Jetstream-based repository streaming with DID filtering (based on atptools)
+// Complete Jetstream implementation using documented @atcute/jetstream approach
+import { JetstreamSubscription, type CommitEvent } from '@atcute/jetstream';
 import { loadConfig } from '../config/site';
 
-export interface JetstreamRecord {
-  uri: string;
-  cid: string;
-  value: any;
-  indexedAt: string;
-  collection: string;
-  $type: string;
-  service: string;
-  did: string;
-  time_us: number;
-  operation: 'create' | 'update' | 'delete';
-}
-
 export interface JetstreamConfig {
   handle: string;
   did?: string;
···
 }
 
 export class JetstreamClient {
-  private ws: WebSocket | null = null;
+  private subscription: JetstreamSubscription | null = null;
   private config: JetstreamConfig;
-  private targetDid: string | null = null;
   private isStreaming = false;
   private listeners: {
-    onRecord?: (record: JetstreamRecord) => void;
+    onRecord?: (event: CommitEvent) => void;
     onError?: (error: Error) => void;
     onConnect?: () => void;
     onDisconnect?: () => void;
···
     this.config = {
       handle: config?.handle || siteConfig.atproto.handle,
       did: config?.did || siteConfig.atproto.did,
-      endpoint: config?.endpoint || 'wss://jetstream1.us-east.bsky.network/subscribe',
+      endpoint: config?.endpoint || 'wss://jetstream2.us-east.bsky.network',
       wantedCollections: config?.wantedCollections || [],
       wantedDids: config?.wantedDids || [],
       cursor: config?.cursor,
     };
-    this.targetDid = this.config.did || null;
     
-    console.log('🔧 JetstreamClient initialized with handle:', this.config.handle);
-    console.log('🎯 Target DID for filtering:', this.targetDid);
-    console.log('🌐 Endpoint:', this.config.endpoint);
+    console.log('🔧 JetstreamClient initialized');
   }
 
-  // Start streaming all repository activity
   async startStreaming(): Promise<void> {
     if (this.isStreaming) {
-      console.log('⚠️ Already streaming repository');
+      console.log('⚠️ Already streaming');
       return;
     }
 
-    console.log('🚀 Starting jetstream repository streaming...');
+    console.log('🚀 Starting jetstream streaming...');
     this.isStreaming = true;
 
     try {
-      // Resolve handle to DID if needed
-      if (!this.targetDid) {
-        this.targetDid = await this.resolveHandle(this.config.handle);
-        if (!this.targetDid) {
-          throw new Error(`Could not resolve handle: ${this.config.handle}`);
-        }
-        console.log('✅ Resolved DID:', this.targetDid);
+      // Add our DID to wanted DIDs if specified
+      const wantedDids = [...(this.config.wantedDids || [])];
+      if (this.config.did && !wantedDids.includes(this.config.did)) {
+        wantedDids.push(this.config.did);
       }
 
-      // Add target DID to wanted DIDs
-      if (this.targetDid && !this.config.wantedDids!.includes(this.targetDid)) {
-        this.config.wantedDids!.push(this.targetDid);
-      }
+      this.subscription = new JetstreamSubscription({
+        url: this.config.endpoint!,
+        wantedCollections: this.config.wantedCollections,
+        wantedDids: wantedDids as any,
+        cursor: this.config.cursor,
+        onConnectionOpen: () => {
+          console.log('✅ Connected to jetstream');
+          this.listeners.onConnect?.();
+        },
+        onConnectionClose: () => {
+          console.log('🔌 Disconnected from jetstream');
+          this.isStreaming = false;
+          this.listeners.onDisconnect?.();
+        },
+        onConnectionError: (error) => {
+          console.error('❌ Jetstream connection error:', error);
+          this.listeners.onError?.(new Error('Connection error'));
+        },
+      });
 
-      // Start WebSocket connection
-      this.connect();
+      // Process events using async iteration as documented
+      this.processEvents();
       
     } catch (error) {
       this.isStreaming = false;
···
     }
   }
 
-  // Stop streaming
-  stopStreaming(): void {
-    if (this.ws) {
-      this.ws.close();
-      this.ws = null;
-    }
-    this.isStreaming = false;
-    console.log('🛑 Stopped jetstream streaming');
-    this.listeners.onDisconnect?.();
-  }
+  private async processEvents(): Promise<void> {
+    if (!this.subscription) return;
 
-  // Connect to jetstream WebSocket
-  private connect(): void {
     try {
-      const url = new URL(this.config.endpoint!);
-      
-      // Add query parameters for filtering (using atptools' parameter names)
-      this.config.wantedCollections!.forEach((collection) => {
-        url.searchParams.append('wantedCollections', collection);
-      });
-      this.config.wantedDids!.forEach((did) => {
-        url.searchParams.append('wantedDids', did);
-      });
-      if (this.config.cursor) {
-        url.searchParams.set('cursor', this.config.cursor.toString());
+      // Use the documented async iteration approach
+      for await (const event of this.subscription) {
+        if (event.kind === 'commit') {
+          console.log('📝 New commit:', {
+            collection: event.commit.collection,
+            operation: event.commit.operation,
+            did: event.did,
+          });
+          
+          this.listeners.onRecord?.(event);
+        }
       }
-
-      console.log('🔌 Connecting to jetstream:', url.toString());
-      
-      this.ws = new WebSocket(url.toString());
-      
-      this.ws.onopen = () => {
-        console.log('✅ Connected to jetstream');
-        this.listeners.onConnect?.();
-      };
-
-      this.ws.onmessage = (event) => {
-        try {
-          const data = JSON.parse(event.data);
-          this.handleMessage(data);
-        } catch (error) {
-          console.error('Error parsing jetstream message:', error);
-        }
-      };
-
-      this.ws.onerror = (error) => {
-        console.error('❌ Jetstream WebSocket error:', error);
-        this.listeners.onError?.(new Error('WebSocket error'));
-      };
-
-      this.ws.onclose = () => {
-        console.log('🔌 Disconnected from jetstream');
-        this.isStreaming = false;
-        this.listeners.onDisconnect?.();
-      };
-
     } catch (error) {
-      console.error('Error connecting to jetstream:', error);
+      console.error('Error processing jetstream events:', error);
       this.listeners.onError?.(error as Error);
+    } finally {
+      this.isStreaming = false;
+      this.listeners.onDisconnect?.();
     }
   }
 
-  // Handle incoming jetstream messages
-  private handleMessage(data: any): void {
-    try {
-      // Handle different message types based on atptools' format
-      if (data.kind === 'commit') {
-        this.handleCommit(data);
-      } else if (data.kind === 'account') {
-        console.log('Account event:', data);
-      } else if (data.kind === 'identity') {
-        console.log('Identity event:', data);
-      } else {
-        console.log('Unknown message type:', data);
-      }
-    } catch (error) {
-      console.error('Error handling jetstream message:', error);
-    }
-  }
-
-  // Handle commit events (record changes)
-  private handleCommit(data: any): void {
-    try {
-      const commit = data.commit;
-      const event = data;
-
-      // Filter by DID if specified
-      if (this.targetDid && event.did !== this.targetDid) {
-        return;
-      }
-
-      const jetstreamRecord: JetstreamRecord = {
-        uri: `at://${event.did}/${commit.collection}/${commit.rkey}`,
-        cid: commit.cid || '',
-        value: commit.record || {},
-        indexedAt: new Date(event.time_us / 1000).toISOString(),
-        collection: commit.collection,
-        $type: (commit.record?.$type as string) || 'unknown',
-        service: this.inferService((commit.record?.$type as string) || '', commit.collection),
-        did: event.did,
-        time_us: event.time_us,
-        operation: commit.operation,
-      };
-
-      console.log('📝 New record from jetstream:', {
-        collection: jetstreamRecord.collection,
-        $type: jetstreamRecord.$type,
-        operation: jetstreamRecord.operation,
-        uri: jetstreamRecord.uri,
-        service: jetstreamRecord.service
-      });
-      
-      this.listeners.onRecord?.(jetstreamRecord);
-    } catch (error) {
-      console.error('Error handling commit:', error);
-    }
-  }
-
-  // Infer service from record type and collection
-  private inferService($type: string, collection: string): string {
-    if (collection.startsWith('grain.social')) return 'grain.social';
-    if (collection.startsWith('app.bsky')) return 'bsky.app';
-    if ($type.includes('grain')) return 'grain.social';
-    return 'unknown';
-  }
-
-  // Resolve handle to DID
-  private async resolveHandle(handle: string): Promise<string | null> {
-    try {
-      // For now, use the configured DID
-      // In a real implementation, you'd call the ATProto API
-      return this.config.did || null;
-    } catch (error) {
-      console.error('Error resolving handle:', error);
-      return null;
-    }
+  stopStreaming(): void {
+    this.subscription = null;
+    this.isStreaming = false;
+    console.log('🛑 Stopped jetstream streaming');
+    this.listeners.onDisconnect?.();
   }
 
   // Event listeners
-  onRecord(callback: (record: JetstreamRecord) => void): void {
+  onRecord(callback: (event: CommitEvent) => void): void {
     this.listeners.onRecord = callback;
   }
 
···
     this.listeners.onDisconnect = callback;
   }
 
-  // Get streaming status
   getStatus(): 'streaming' | 'stopped' {
     return this.isStreaming ? 'streaming' : 'stopped';
   }
+}
+
+// Shared Jetstream functionality
+let sharedJetstream: JetstreamClient | null = null;
+let connectionCount = 0;
+const listeners: Map<string, Set<(event: CommitEvent) => void>> = new Map();
+
+export function getSharedJetstream(): JetstreamClient {
+  if (!sharedJetstream) {
+    // Create a shared client with common collections
+    sharedJetstream = new JetstreamClient({
+      wantedCollections: [
+        'app.bsky.feed.post',
+        'a.status.update',
+        'social.grain.gallery',
+        'social.grain.gallery.item',
+        'social.grain.photo',
+        'com.whtwnd.blog.entry'
+      ]
+    });
+    
+    // Set up the main record handler that distributes to filtered listeners
+    sharedJetstream.onRecord((event) => {
+      // Distribute to all listeners that match the filter
+      listeners.forEach((listenerSet, filterKey) => {
+        if (matchesFilter(event, filterKey)) {
+          listenerSet.forEach(callback => callback(event));
+        }
+      });
+    });
+  }
+  return sharedJetstream;
+}
+
+// Start the shared stream (call once when first component needs it)
+export async function startSharedStream(): Promise<void> {
+  const jetstream = getSharedJetstream();
+  if (connectionCount === 0) {
+    await jetstream.startStreaming();
+  }
+  connectionCount++;
+}
+
+// Stop the shared stream (call when last component is done)
+export function stopSharedStream(): void {
+  connectionCount--;
+  if (connectionCount <= 0 && sharedJetstream) {
+    sharedJetstream.stopStreaming();
+    connectionCount = 0;
+  }
+}
+
+// Subscribe to filtered records
+export function subscribeToRecords(
+  filter: string | ((event: CommitEvent) => boolean),
+  callback: (event: CommitEvent) => void
+): () => void {
+  const filterKey = typeof filter === 'string' ? filter : filter.toString();
+  
+  if (!listeners.has(filterKey)) {
+    listeners.set(filterKey, new Set());
+  }
+  
+  const listenerSet = listeners.get(filterKey)!;
+  listenerSet.add(callback);
+  
+  // Return unsubscribe function
+  return () => {
+    const set = listeners.get(filterKey);
+    if (set) {
+      set.delete(callback);
+      if (set.size === 0) {
+        listeners.delete(filterKey);
+      }
+    }
+  };
+}
+
+// Helper to check if a record matches a filter
+function matchesFilter(event: CommitEvent, filterKey: string): boolean {
+  // Handle delete operations (no record property)
+  if (event.commit.operation === 'delete') {
+    // For delete operations, only support collection and operation matching
+    if (filterKey.startsWith('collection:')) {
+      const expectedCollection = filterKey.substring(11);
+      return event.commit.collection === expectedCollection;
+    }
+    if (filterKey.startsWith('operation:')) {
+      const expectedOperation = filterKey.substring(10);
+      return event.commit.operation === expectedOperation;
+    }
+    return false;
+  }
+  
+  // For create/update operations, we have record data
+  const record = event.commit.record;
+  const $type = record?.$type as string;
+  
+  // Support simple $type matching
+  if (filterKey.startsWith('$type:')) {
+    const expectedType = filterKey.substring(6);
+    return $type === expectedType;
+  }
+  
+  // Support collection matching
+  if (filterKey.startsWith('collection:')) {
+    const expectedCollection = filterKey.substring(11);
+    return event.commit.collection === expectedCollection;
+  }
+  
+  // Support operation matching
+  if (filterKey.startsWith('operation:')) {
+    const expectedOperation = filterKey.substring(10);
+    return event.commit.operation === expectedOperation;
+  }
+  
+  // Default to exact match
+  return $type === filterKey;
+}
+
+// Convenience functions for common filters
+export function subscribeToStatusUpdates(callback: (event: CommitEvent) => void): () => void {
+  return subscribeToRecords('$type:a.status.update', callback);
+}
+
+export function subscribeToPosts(callback: (event: CommitEvent) => void): () => void {
+  return subscribeToRecords('$type:app.bsky.feed.post', callback);
+}
+
+export function subscribeToGalleryUpdates(callback: (event: CommitEvent) => void): () => void {
+  return subscribeToRecords('collection:social.grain.gallery', callback);
 } 

src/lib/atproto/shared-jetstream.ts

···
-// Shared Jetstream instance for components to access and filter
-import { JetstreamClient, type JetstreamRecord } from './jetstream-client';
-
-let sharedJetstream: JetstreamClient | null = null;
-let connectionCount = 0;
-const listeners: Map<string, Set<(record: JetstreamRecord) => void>> = new Map();
-
-export function getSharedJetstream(): JetstreamClient {
-  if (!sharedJetstream) {
-    sharedJetstream = new JetstreamClient();
-    
-    // Set up the main record handler that distributes to filtered listeners
-    sharedJetstream.onRecord((record) => {
-      // Distribute to all listeners that match the filter
-      listeners.forEach((listenerSet, filterKey) => {
-        if (matchesFilter(record, filterKey)) {
-          listenerSet.forEach(callback => callback(record));
-        }
-      });
-    });
-  }
-  return sharedJetstream;
-}
-
-// Start the shared stream (call once when first component needs it)
-export async function startSharedStream(): Promise<void> {
-  const jetstream = getSharedJetstream();
-  if (connectionCount === 0) {
-    await jetstream.startStreaming();
-  }
-  connectionCount++;
-}
-
-// Stop the shared stream (call when last component is done)
-export function stopSharedStream(): void {
-  connectionCount--;
-  if (connectionCount <= 0 && sharedJetstream) {
-    sharedJetstream.stopStreaming();
-    connectionCount = 0;
-  }
-}
-
-// Subscribe to filtered records
-export function subscribeToRecords(
-  filter: string | ((record: JetstreamRecord) => boolean),
-  callback: (record: JetstreamRecord) => void
-): () => void {
-  const filterKey = typeof filter === 'string' ? filter : filter.toString();
-  
-  if (!listeners.has(filterKey)) {
-    listeners.set(filterKey, new Set());
-  }
-  
-  const listenerSet = listeners.get(filterKey)!;
-  listenerSet.add(callback);
-  
-  // Return unsubscribe function
-  return () => {
-    const set = listeners.get(filterKey);
-    if (set) {
-      set.delete(callback);
-      if (set.size === 0) {
-        listeners.delete(filterKey);
-      }
-    }
-  };
-}
-
-// Helper to check if a record matches a filter
-function matchesFilter(record: JetstreamRecord, filterKey: string): boolean {
-  // If filter is a function string, we can't easily evaluate it
-  // For now, support simple $type matching
-  if (filterKey.startsWith('$type:')) {
-    const expectedType = filterKey.substring(6);
-    return record.$type === expectedType;
-  }
-  
-  // Support collection matching
-  if (filterKey.startsWith('collection:')) {
-    const expectedCollection = filterKey.substring(11);
-    return record.collection === expectedCollection;
-  }
-  
-  // Support operation matching
-  if (filterKey.startsWith('operation:')) {
-    const expectedOperation = filterKey.substring(10);
-    return record.operation === expectedOperation;
-  }
-  
-  // Default to exact match
-  return filterKey === record.$type;
-}
-
-// Convenience functions for common filters
-export function subscribeToStatusUpdates(callback: (record: JetstreamRecord) => void): () => void {
-  return subscribeToRecords('$type:a.status.update', callback);
-}
-
-export function subscribeToPosts(callback: (record: JetstreamRecord) => void): () => void {
-  return subscribeToRecords('$type:app.bsky.feed.post', callback);
-}
-
-export function subscribeToGalleryUpdates(callback: (record: JetstreamRecord) => void): () => void {
-  return subscribeToRecords('collection:social.grain.gallery', callback);
-}

src/lib/services/content-system.ts

···
-import { AtprotoBrowser } from '../atproto/atproto-browser';
-import { JetstreamClient } from '../atproto/jetstream-client';
-import { GrainGalleryService } from './grain-gallery-service';
-import { loadConfig } from '../config/site';
-import type { AtprotoRecord } from '../atproto/atproto-browser';
-
-export interface ContentItem {
-  uri: string;
-  cid: string;
-  $type: string;
-  collection: string;
-  createdAt: string;
-  indexedAt: string;
-  value: any;
-  service: string;
-  operation?: 'create' | 'update' | 'delete';
-}
-
-export interface ContentFeed {
-  items: ContentItem[];
-  lastUpdated: string;
-  totalItems: number;
-  collections: string[];
-}
-
-export interface ContentSystemConfig {
-  enableStreaming?: boolean;
-  buildTimeOnly?: boolean;
-  collections?: string[];
-  maxItems?: number;
-}
-
-export class ContentSystem {
-  private browser: AtprotoBrowser;
-  private jetstream: JetstreamClient;
-  private grainGalleryService: GrainGalleryService;
-  private config: any;
-  private contentFeed: ContentFeed;
-  private isStreaming = false;
-
-  constructor() {
-    this.config = loadConfig();
-    this.browser = new AtprotoBrowser();
-    this.jetstream = new JetstreamClient();
-    this.grainGalleryService = new GrainGalleryService();
-    
-    this.contentFeed = {
-      items: [],
-      lastUpdated: new Date().toISOString(),
-      totalItems: 0,
-      collections: []
-    };
-  }
-
-  // Initialize content system (build-time)
-  async initialize(identifier: string, options: ContentSystemConfig = {}): Promise<ContentFeed> {
-    console.log('🚀 Initializing content system for:', identifier);
-    
-    try {
-      // Get repository info
-      const repoInfo = await this.browser.getRepoInfo(identifier);
-      if (!repoInfo) {
-        throw new Error(`Could not get repository info for: ${identifier}`);
-      }
-
-      console.log('📊 Repository info:', {
-        handle: repoInfo.handle,
-        did: repoInfo.did,
-        collections: repoInfo.collections.length,
-        recordCount: repoInfo.recordCount
-      });
-
-      // Gather all content from collections
-      const allItems: ContentItem[] = [];
-      const collections = options.collections || repoInfo.collections;
-
-      for (const collection of collections) {
-        console.log(`📦 Fetching from collection: ${collection}`);
-        const records = await this.browser.getCollectionRecords(identifier, collection, options.maxItems || 100);
-        
-        if (records && records.records) {
-          for (const record of records.records) {
-            const contentItem: ContentItem = {
-              uri: record.uri,
-              cid: record.cid,
-              $type: record.$type,
-              collection: record.collection,
-              createdAt: record.value?.createdAt || record.indexedAt,
-              indexedAt: record.indexedAt,
-              value: record.value,
-              service: this.inferService(record.$type, record.collection),
-              operation: 'create' // Build-time items are existing
-            };
-            
-            allItems.push(contentItem);
-          }
-        }
-      }
-
-      // Sort by creation date (newest first)
-      allItems.sort((a, b) => {
-        const dateA = new Date(a.createdAt);
-        const dateB = new Date(b.createdAt);
-        return dateB.getTime() - dateA.getTime();
-      });
-
-      this.contentFeed = {
-        items: allItems,
-        lastUpdated: new Date().toISOString(),
-        totalItems: allItems.length,
-        collections: collections
-      };
-
-      console.log(`✅ Content system initialized with ${allItems.length} items`);
-
-      // Start streaming if enabled
-      if (!options.buildTimeOnly && options.enableStreaming !== false) {
-        await this.startStreaming(identifier);
-      }
-
-      return this.contentFeed;
-    } catch (error) {
-      console.error('Error initializing content system:', error);
-      throw error;
-    }
-  }
-
-  // Start real-time streaming
-  async startStreaming(identifier: string): Promise<void> {
-    if (this.isStreaming) {
-      console.log('⚠️ Already streaming');
-      return;
-    }
-
-    console.log('🌊 Starting real-time content streaming...');
-    this.isStreaming = true;
-
-    // Set up jetstream event handlers
-    this.jetstream.onRecord((record) => {
-      this.handleNewContent(record);
-    });
-
-    this.jetstream.onError((error) => {
-      console.error('❌ Jetstream error:', error);
-    });
-
-    this.jetstream.onConnect(() => {
-      console.log('✅ Connected to real-time stream');
-    });
-
-    this.jetstream.onDisconnect(() => {
-      console.log('🔌 Disconnected from real-time stream');
-      this.isStreaming = false;
-    });
-
-    // Start streaming
-    await this.jetstream.startStreaming();
-  }
-
-  // Handle new content from streaming
-  private handleNewContent(jetstreamRecord: any): void {
-    const contentItem: ContentItem = {
-      uri: jetstreamRecord.uri,
-      cid: jetstreamRecord.cid,
-      $type: jetstreamRecord.$type,
-      collection: jetstreamRecord.collection,
-      createdAt: jetstreamRecord.value?.createdAt || jetstreamRecord.indexedAt,
-      indexedAt: jetstreamRecord.indexedAt,
-      value: jetstreamRecord.value,
-      service: jetstreamRecord.service,
-      operation: jetstreamRecord.operation
-    };
-
-    // Add to beginning of feed (newest first)
-    this.contentFeed.items.unshift(contentItem);
-    this.contentFeed.totalItems++;
-    this.contentFeed.lastUpdated = new Date().toISOString();
-
-    console.log('📝 New content added:', {
-      $type: contentItem.$type,
-      collection: contentItem.collection,
-      operation: contentItem.operation
-    });
-
-    // Emit event for UI updates
-    this.emitContentUpdate(contentItem);
-  }
-
-  // Get current content feed
-  getContentFeed(): ContentFeed {
-    return this.contentFeed;
-  }
-
-  // Get content by type
-  getContentByType($type: string): ContentItem[] {
-    return this.contentFeed.items.filter(item => item.$type === $type);
-  }
-
-  // Get content by collection
-  getContentByCollection(collection: string): ContentItem[] {
-    return this.contentFeed.items.filter(item => item.collection === collection);
-  }
-
-  // Get galleries (using specialized service)
-  async getGalleries(identifier: string): Promise<any[]> {
-    return await this.grainGalleryService.getGalleries(identifier);
-  }
-
-  // Filter content by function
-  filterContent(filterFn: (item: ContentItem) => boolean): ContentItem[] {
-    return this.contentFeed.items.filter(filterFn);
-  }
-
-  // Search content
-  searchContent(query: string): ContentItem[] {
-    const lowerQuery = query.toLowerCase();
-    return this.contentFeed.items.filter(item => {
-      const text = JSON.stringify(item.value).toLowerCase();
-      return text.includes(lowerQuery);
-    });
-  }
-
-  // Stop streaming
-  stopStreaming(): void {
-    if (this.isStreaming) {
-      this.jetstream.stopStreaming();
-      this.isStreaming = false;
-    }
-  }
-
-  // Infer service from record type and collection
-  private inferService($type: string, collection: string): string {
-    if (collection.startsWith('grain.social') || $type.includes('grain')) return 'grain.social';
-    if (collection.startsWith('app.bsky')) return 'bsky.app';
-    if (collection.startsWith('sh.tangled')) return 'sh.tangled';
-    return 'unknown';
-  }
-
-  // Event system for UI updates
-  private listeners: {
-    onContentUpdate?: (item: ContentItem) => void;
-    onContentAdd?: (item: ContentItem) => void;
-    onContentRemove?: (item: ContentItem) => void;
-  } = {};
-
-  onContentUpdate(callback: (item: ContentItem) => void): void {
-    this.listeners.onContentUpdate = callback;
-  }
-
-  onContentAdd(callback: (item: ContentItem) => void): void {
-    this.listeners.onContentAdd = callback;
-  }
-
-  onContentRemove(callback: (item: ContentItem) => void): void {
-    this.listeners.onContentRemove = callback;
-  }
-
-  private emitContentUpdate(item: ContentItem): void {
-    this.listeners.onContentUpdate?.(item);
-    if (item.operation === 'create') {
-      this.listeners.onContentAdd?.(item);
-    } else if (item.operation === 'delete') {
-      this.listeners.onContentRemove?.(item);
-    }
-  }
-
-  // Get streaming status
-  getStreamingStatus(): 'streaming' | 'stopped' {
-    return this.isStreaming ? 'streaming' : 'stopped';
-  }
-} 

src/pages/leaflets/[leaflet].astro

···
       <div class="text-sm text-gray-500 dark:text-gray-400 mb-6">
         {document.data.publishedAt && (
           <span>
-            Published: {new Date(document.data.publishedAt).toLocaleDateString('en-US', {
+            {new Date(document.data.publishedAt).toLocaleDateString('en-US', {
               year: 'numeric',
               month: 'long',
               day: 'numeric',