/**
* Wirecore Bridge - Local relay for P2P ↔ Relay sync
*
* Responsibilities:
* 1. Read from local Corestore (shared with pear-app)
* 2. Watch for new posts in own feed
* 3. Push new posts to relay (YAML + signature)
* 4. Pull relay posts and cache locally
* 5. Maintain sync state
*/
import Corestore from 'corestore'
import Hyperbee from 'hyperbee'
import crypto from 'hypercore-crypto'
import b4a from 'b4a'
import { RelayClient } from '../lib/relay-client.js'
import { ANProto } from '../lib/anproto.js'
export class WirecoreBridge {
constructor(relayUrl, existingStore = null, storagePath = null) {
this.relayUrl = relayUrl
this.storagePath = storagePath
this.store = existingStore // Reuse existing Corestore if provided
this.metaBee = null
this.contentCache = null // Phase 3: content-by-hash storage
this.syncBee = null // Track sync state
this.relayClient = null
this.myFeed = null
this.anprotoKeypair = null
this._pullInterval = null // For cleanup
this._appendHandler = null // For cleanup
this._ownStore = false // Track if we created the store
}
/**
* Initialize bridge (read-only access to pear-app storage)
*/
async init() {
console.log('🌉 Initializing bridge...')
console.log(' Relay:', this.relayUrl)
// Open Corestore if not provided
if (!this.store) {
if (!this.storagePath) {
throw new Error('Either existingStore or storagePath must be provided')
}
console.log(' Opening storage:', this.storagePath)
this.store = new Corestore(this.storagePath)
await this.store.ready()
this._ownStore = true
} else {
console.log(' Using existing Corestore')
}
// Load metadata (keypair, etc.)
this.metaBee = new Hyperbee(this.store.get({ name: 'metadata' }), {
keyEncoding: 'utf-8',
valueEncoding: 'json'
})
await this.metaBee.ready()
// Phase 3: Load content cache (shared with pear-app)
this.contentCache = new Hyperbee(this.store.get({ name: 'content-cache' }), {
keyEncoding: 'utf-8',
valueEncoding: 'utf-8'
})
await this.contentCache.ready()
// Create sync state tracker
this.syncBee = new Hyperbee(this.store.get({ name: 'bridge-sync' }), {
keyEncoding: 'utf-8',
valueEncoding: 'json'
})
await this.syncBee.ready()
// Load ANProto keypair
const keypairNode = await this.metaBee.get('anproto-keypair')
if (!keypairNode) {
throw new Error('No identity found. Run: pear run . cli init')
}
const { publicKey, secretKey } = ANProto.loadKeypair(keypairNode.value)
this.anprotoKeypair = { publicKey, secretKey }
// Open own feed
this.myFeed = this.store.get({
publicKey: this.anprotoKeypair.publicKey,
secretKey: this.anprotoKeypair.secretKey
})
await this.myFeed.ready()
// Initialize relay client
this.relayClient = new RelayClient(this.relayUrl)
console.log('✅ Bridge initialized')
console.log(' Pubkey:', b4a.toString(this.anprotoKeypair.publicKey, 'base64'))
console.log(' Feed length:', this.myFeed.length)
}
/**
* Get last synced index for own feed
*/
async getLastSyncedIndex() {
const node = await this.syncBee.get('last-synced-index')
return node ? node.value : 0
}
/**
* Set last synced index
*/
async setLastSyncedIndex(index) {
await this.syncBee.put('last-synced-index', index)
}
/**
* Get sync statistics
*/
async getStats() {
const node = await this.syncBee.get('stats')
return node ? node.value : {
totalPushed: 0,
totalPulled: 0,
lastSync: null,
lastError: null
}
}
/**
* Update sync statistics
*/
async updateStats(updates) {
const stats = await this.getStats()
const newStats = { ...stats, ...updates, lastSync: Date.now() }
await this.syncBee.put('stats', newStats)
return newStats
}
/**
* Push all unsent posts to relay
*/
async push() {
const lastSynced = await this.getLastSyncedIndex()
const currentLength = this.myFeed.length
if (lastSynced >= currentLength) {
console.log('✓ Already up to date (no new posts)')
return { pushed: 0, errors: 0 }
}
console.log(`📤 Pushing ${currentLength - lastSynced} posts to relay...`)
const pubkey = b4a.toString(this.anprotoKeypair.publicKey, 'base64')
let pushed = 0
let errors = 0
for (let i = lastSynced; i < currentLength; i++) {
try {
const block = await this.myFeed.get(i)
const entry = JSON.parse(block.toString())
// Skip if no YAML (old format posts)
if (!entry.yaml) {
console.log(` ⏭ Skipping post ${i} (old format, no YAML)`)
await this.setLastSyncedIndex(i + 1)
continue
}
// Extract YAML content
const yaml = entry.yaml
// Extract image data if present (should be data URL format)
let imageData = null
if (entry.imageData) {
// entry.imageData is stored as data URL: data:image/jpeg;base64,...
// Send as-is - wiredove protocol expects data URL strings
imageData = entry.imageData
}
// Hash YAML content (base64, not base64url - ANProto uses base64)
const contentHash = ANProto.hash(yaml)
// Sign the content hash (ANProto.sign adds timestamp internally)
const signature = ANProto.sign(contentHash, this.anprotoKeypair)
// Push to relay (with image data if present)
await this.relayClient.push(pubkey, yaml, signature, imageData)
console.log(` ✓ Pushed post ${i}${imageData ? ' (with image)' : ''}`)
pushed++
// Update sync state
await this.setLastSyncedIndex(i + 1)
} catch (err) {
console.error(` ✗ Failed to push post ${i}:`, err.message)
errors++
// Continue with next post
}
}
console.log(`✅ Push complete: ${pushed} pushed, ${errors} errors`)
// Update stats
const stats = await this.getStats()
await this.updateStats({
totalPushed: stats.totalPushed + pushed
})
return { pushed, errors }
}
/**
* Pull posts from relay and cache locally
*/
async pull() {
console.log('📥 Pulling from relay...')
try {
const posts = await this.relayClient.pull()
const count = Array.isArray(posts) ? posts.length : 0
console.log(` Received ${count} posts`)
// Phase 3: Store posts in contentCache by hash
let stored = 0
if (Array.isArray(posts)) {
for (const post of posts) {
if (post.text) {
// Hash the content
const contentHash = ANProto.hash(post.text)
// Store in contentCache (only if not already there)
const existing = await this.contentCache.get(contentHash)
if (!existing) {
await this.contentCache.put(contentHash, post.text)
stored++
}
}
}
console.log(` Stored ${stored} new posts in contentCache`)
// Log recent posts
const recent = posts.slice(0, 5)
console.log(` Showing ${recent.length} most recent:`)
recent.forEach(post => {
const author = post.author ? post.author.substring(0, 16) + '...' : 'unknown'
const preview = post.text ? post.text.substring(0, 50) : ''
console.log(` • ${author}: ${preview}`)
})
}
console.log('✅ Pull complete')
// Update stats
const stats = await this.getStats()
await this.updateStats({
totalPulled: stats.totalPulled + count
})
return { pulled: count }
} catch (err) {
console.error('❌ Pull failed:', err.message)
await this.updateStats({ lastError: err.message })
throw err
}
}
/**
* Watch for new posts and auto-push
*/
async watch() {
console.log('👁️ Watching for new posts...')
console.log(' Press Ctrl+C to stop\n')
// Set up feed listener
this.myFeed.on('append', async () => {
console.log('🔔 New post detected!')
try {
await this.push()
} catch (err) {
console.error('Failed to push:', err.message)
}
})
// Keep process alive
await new Promise(() => {})
}
/**
* Daemon mode: watch + periodic pull
*/
async start(pullInterval = 5 * 60 * 1000) {
console.log('🚀 Starting bridge daemon...')
console.log(` Pull interval: ${pullInterval / 1000}s\n`)
// Initial sync
try {
await this.push()
} catch (err) {
console.error('Initial push failed:', err.message)
}
try {
await this.pull()
} catch (err) {
console.error('Initial pull failed:', err.message)
}
// Watch for new posts
this._appendHandler = async () => {
console.log('🔔 New post detected!')
try {
await this.push()
} catch (err) {
console.error('Failed to push:', err.message)
}
}
this.myFeed.on('append', this._appendHandler)
// Periodic pull from relay
this._pullInterval = setInterval(async () => {
console.log('⏰ Periodic pull from relay...')
try {
await this.pull()
} catch (err) {
console.error('Periodic pull failed:', err.message)
}
}, pullInterval)
console.log('✅ Bridge daemon running\n')
}
/**
* Stop daemon mode (cleanup intervals and listeners)
*/
async stop() {
console.log('🛑 Stopping bridge daemon...')
// Clear periodic pull interval
if (this._pullInterval) {
clearInterval(this._pullInterval)
this._pullInterval = null
}
// Remove append event listener
if (this._appendHandler) {
this.myFeed.removeListener('append', this._appendHandler)
this._appendHandler = null
}
console.log('✅ Bridge daemon stopped')
}
/**
* One-time sync (push + pull)
*/
async sync() {
console.log('🔄 Syncing with relay...\n')
let pushResult = { pushed: 0, errors: 0 }
let pullResult = { pulled: 0 }
try {
pushResult = await this.push()
} catch (err) {
console.error('Push failed:', err.message)
}
try {
pullResult = await this.pull()
} catch (err) {
console.error('Pull failed:', err.message)
}
console.log('\n✅ Sync complete')
console.log(` Pushed: ${pushResult.pushed}`)
console.log(` Pulled: ${pullResult.pulled}`)
return { ...pushResult, ...pullResult }
}
/**
* Status check
*/
async status() {
const lastSynced = await this.getLastSyncedIndex()
const currentLength = this.myFeed.length
const pending = currentLength - lastSynced
const stats = await this.getStats()
// Check relay health
const health = await this.relayClient.health()
const status = {
relay: {
url: this.relayUrl,
online: health.online,
error: health.error || null
},
storage: {
path: this.storagePath,
feedLength: currentLength
},
sync: {
lastSyncedIndex: lastSynced,
pendingPosts: pending,
lastSync: stats.lastSync,
lastError: stats.lastError
},
stats: {
totalPushed: stats.totalPushed,
totalPulled: stats.totalPulled
}
}
console.log('\nBridge Status:')
console.log('━━━━━━━━━━━━━━')
console.log('Relay:')
console.log(` URL: ${status.relay.url}`)
console.log(` Status: ${status.relay.online ? '🟢 Online' : '🔴 Offline'}`)
if (status.relay.error) {
console.log(` Error: ${status.relay.error}`)
}
console.log('\nStorage:')
console.log(` Path: ${status.storage.path}`)
console.log(` Feed: ${status.storage.feedLength} posts`)
console.log('\nSync:')
console.log(` Last index: ${status.sync.lastSyncedIndex}`)
console.log(` Pending: ${status.sync.pendingPosts} posts`)
if (status.sync.lastSync) {
console.log(` Last sync: ${new Date(status.sync.lastSync).toLocaleString()}`)
}
if (status.sync.lastError) {
console.log(` Last error: ${status.sync.lastError}`)
}
console.log('\nStatistics:')
console.log(` Pushed: ${status.stats.totalPushed}`)
console.log(` Pulled: ${status.stats.totalPulled}`)
return status
}
/**
* Cleanup resources
*/
async close() {
console.log('Closing bridge...')
// Only close the store if we created it (not shared with web server)
if (this.store && this._ownStore) {
await this.store.close()
}
console.log('✅ Bridge closed')
}
}