/**
* Content Sync Manager
*
* Transport-agnostic content-by-hash synchronization.
* Manages request/response lifecycle with deduplication,
* timeouts, and retries.
*
* Compatible with Wiredove's sendHash/sendBlob pattern.
*/
import { EventEmitter } from 'events'
import crypto from 'crypto'
export class ContentSyncManager extends EventEmitter {
constructor(contentStore, options = {}) {
super()
this.contentStore = contentStore
this.options = {
timeout: options.timeout || 30000, // 30s default
maxRetries: options.maxRetries || 2,
maxConcurrent: options.maxConcurrent || 10,
retryBackoff: options.retryBackoff || 5000 // 5s initial backoff
}
// Pending requests: hash -> { promise, resolve, reject, timeout, attempts }
this.pending = new Map()
// Transports: name -> { send: fn, isActive: fn }
this.transports = new Map()
// Negative cache: hash -> timestamp (for hashes we tried and failed)
this.negativeCache = new Map()
this.negativeCacheTTL = 5 * 60 * 1000 // 5 minutes
// Stats
this.stats = {
requests: 0,
hits: 0,
misses: 0,
timeouts: 0,
errors: 0
}
}
/**
* Register a transport for sending/receiving content
* @param {string} name - Transport name (e.g. 'hyperswarm', 'trystero')
* @param {Function} sendFn - async (hash, requestId) => void
* @param {Function} isActiveFn - () => boolean (has active connections?)
*/
registerTransport(name, sendFn, isActiveFn) {
this.transports.set(name, {
send: sendFn,
isActive: isActiveFn
})
console.log(`ContentSync: registered transport '${name}'`)
}
/**
* Request content by hash
* @param {string} hash - Base64 SHA256 hash
* @param {Object} options - { timeout, skipIfNoPeers }
* @returns {Promise<string>} YAML content or null
*/
async requestContent(hash, options = {}) {
this.stats.requests++
// Check local store first
const local = this.contentStore.get(hash)
if (local) {
this.stats.hits++
this.emit('hit', { hash })
return local
}
// Check negative cache
const negativeCached = this.negativeCache.get(hash)
if (negativeCached && (Date.now() - negativeCached < this.negativeCacheTTL)) {
this.emit('negative-cache-hit', { hash })
return null
}
// Check if we have active transports
const hasActiveTransport = Array.from(this.transports.values())
.some(t => t.isActive())
if (!hasActiveTransport) {
if (options.skipIfNoPeers) {
this.emit('no-peers', { hash })
return null
}
throw new Error('No active peers available')
}
// Deduplicate: if already requesting this hash, share the promise
if (this.pending.has(hash)) {
console.log(`ContentSync: deduplicating request for ${hash.substring(0, 8)}...`)
return this.pending.get(hash).promise
}
// Create new request
const requestId = crypto.randomUUID()
const timeout = options.timeout || this.options.timeout
// Create pending entry (will add promise after creation)
const pending = {
promise: null,
resolve: null,
reject: null,
timeoutId: null,
requestId,
attempts: 0,
startTime: Date.now()
}
// Create promise and store resolve/reject
const promise = new Promise((resolve, reject) => {
pending.resolve = resolve
pending.reject = reject
pending.timeoutId = setTimeout(() => {
this.handleTimeout(hash, requestId)
reject(new Error(`Content request timeout for ${hash}`))
}, timeout)
})
// Update pending with promise reference
pending.promise = promise
this.pending.set(hash, pending)
// Send request via all active transports
for (const [name, transport] of this.transports) {
if (transport.isActive()) {
try {
await transport.send(hash, requestId)
console.log(`ContentSync: sent request for ${hash.substring(0, 8)}... via ${name}`)
this.emit('request', { hash, transport: name, requestId })
} catch (err) {
console.error(`ContentSync: failed to send via ${name}:`, err.message)
}
}
}
this.stats.misses++
return promise
}
/**
* Handle incoming content request from peer
* @param {string} hash - Requested hash
* @param {string} requestId - Request ID for correlation
* @param {Function} respondFn - async (hash, content, requestId) => void
*/
async handleContentRequest(hash, requestId, respondFn) {
console.log(`ContentSync: received request for ${hash.substring(0, 8)}...`)
const content = this.contentStore.get(hash)
const found = !!content
try {
await respondFn(hash, content, requestId)
this.emit('served', { hash, found, requestId })
} catch (err) {
console.error(`ContentSync: failed to respond:`, err.message)
this.emit('serve-error', { hash, requestId, error: err })
}
}
/**
* Handle incoming content response from peer
* @param {string} hash - Content hash
* @param {string} content - YAML content (or null if not found)
* @param {string} requestId - Request ID for correlation
*/
async handleContentResponse(hash, content, requestId) {
const pending = this.pending.get(hash)
if (!pending) {
console.log(`ContentSync: received unsolicited response for ${hash.substring(0, 8)}...`)
return
}
// Verify hash matches content (integrity check)
if (content) {
const computed = this.contentStore.constructor.prototype.hash
? await this.contentStore.hash(content)
: hash // Skip verification if ContentStore doesn't expose hash fn
if (computed !== hash) {
console.error(`ContentSync: integrity violation! Hash mismatch for ${hash}`)
this.emit('integrity-violation', { hash, requestId })
// Don't store, don't resolve
return
}
// Store content
await this.contentStore.put(hash, content)
console.log(`ContentSync: received and stored ${hash.substring(0, 8)}...`)
}
// Clear timeout
clearTimeout(pending.timeoutId)
// Resolve promise
pending.resolve(content)
// Clean up
this.pending.delete(hash)
// Emit event
const latency = Date.now() - pending.startTime
this.emit('response', { hash, found: !!content, latency, requestId })
}
/**
* Handle request timeout
*/
handleTimeout(hash, requestId) {
const pending = this.pending.get(hash)
if (!pending) return
pending.attempts++
this.stats.timeouts++
console.log(`ContentSync: timeout for ${hash.substring(0, 8)}... (attempt ${pending.attempts})`)
// Retry logic
if (pending.attempts < this.options.maxRetries) {
const backoff = this.options.retryBackoff * Math.pow(2, pending.attempts - 1)
console.log(`ContentSync: retrying in ${backoff}ms...`)
setTimeout(() => {
this.retryRequest(hash, requestId)
}, backoff)
} else {
// Max retries reached, cache negative result
this.negativeCache.set(hash, Date.now())
this.emit('timeout', { hash, attempts: pending.attempts, requestId })
// Cleanup will happen when promise rejects
}
}
/**
* Retry a failed request
*/
async retryRequest(hash, requestId) {
console.log(`ContentSync: retrying ${hash.substring(0, 8)}...`)
// Send via all active transports again
for (const [name, transport] of this.transports) {
if (transport.isActive()) {
try {
await transport.send(hash, requestId)
this.emit('retry', { hash, transport: name, requestId })
} catch (err) {
console.error(`ContentSync: retry failed via ${name}:`, err.message)
}
}
}
}
/**
* Get statistics
*/
getStats() {
return {
...this.stats,
pending: this.pending.size,
transports: this.transports.size,
negativeCacheSize: this.negativeCache.size
}
}
/**
* Clear negative cache (useful for testing)
*/
clearNegativeCache() {
this.negativeCache.clear()
}
}
export default ContentSyncManager