/**
* Wirecore-APDS: APDS API implementation with Hypercore backend
*
* Drop-in replacement for APDS that uses Hypercore/Hyperbee instead of cachekv.
* Maintains full API compatibility with existing APDS clients.
*/
import Corestore from 'corestore'
import Hyperbee from 'hyperbee'
import Hyperswarm from 'hyperswarm'
import crypto from 'node:crypto'
import https from 'node:https'
import hypercore_crypto from 'hypercore-crypto'
import nacl from 'tweetnacl'
// Configuration (can be adjusted at runtime)
export const config = {
maxConcurrentFetches: parseInt(process.env.MAX_CONCURRENT_FETCHES) || 5,
maxCachedMessages: parseInt(process.env.MAX_CACHED_MESSAGES) || 10000,
lazyLoadLimit: parseInt(process.env.LAZY_LOAD_LIMIT) || 20,
fetchTimeout: parseInt(process.env.FETCH_TIMEOUT) || 5000,
remoteServer: process.env.REMOTE_SERVER || 'https://pub.wiredove.net'
}
// Apply connection limit
https.globalAgent.maxSockets = config.maxConcurrentFetches
let store = null
let metaBee = null // For key-value metadata
let blobBee = null // For content-addressable blobs
let messageCore = null // For append-only message log
let swarm = null
// In-memory cache for performance (like original APDS)
let hashLog = []
let openedLog = []
let hashSet = new Set() // O(1) duplicate checking
const inFlightFetches = new Map() // Track pending blob fetches
export const apds = {}
/**
* Initialize wirecore-apds with Hypercore backend
*/
apds.start = async (appId) => {
if (store) {
console.warn('APDS already started, ignoring duplicate start() call')
return
}
// Create Corestore
store = new Corestore(`./${appId}-wirecore`)
await store.ready()
// Create Hyperbees for key-value storage
metaBee = new Hyperbee(store.get({ name: '__metadata' }), {
keyEncoding: 'utf-8',
valueEncoding: 'json'
})
await metaBee.ready()
blobBee = new Hyperbee(store.get({ name: '__blobs' }), {
keyEncoding: 'utf-8',
valueEncoding: 'utf-8'
})
await blobBee.ready()
// Create message log Hypercore
messageCore = store.get({ name: '__messages' })
await messageCore.ready()
// Initialize Hyperswarm for P2P replication
swarm = new Hyperswarm()
swarm.on('connection', (socket) => {
store.replicate(socket)
})
// Announce all cores
swarm.join(metaBee.feed.discoveryKey)
swarm.join(blobBee.feed.discoveryKey)
swarm.join(messageCore.discoveryKey)
// Load existing message log into memory
await rebuildCache()
console.log('Wirecore-APDS started')
console.log(' Metadata core:', metaBee.feed.key.toString('hex'))
console.log(' Blobs core:', blobBee.feed.key.toString('hex'))
console.log(' Messages core:', messageCore.key.toString('hex'))
}
/**
* Rebuild in-memory cache from Hypercore
*/
async function rebuildCache() {
hashLog = []
openedLog = []
hashSet = new Set()
// Read all messages from Hypercore (up to max cache size)
const length = messageCore.length
const startIndex = Math.max(0, length - config.maxCachedMessages)
for (let i = startIndex; i < length; i++) {
try {
const block = await messageCore.get(i)
const msg = JSON.parse(block.toString('utf-8'))
hashLog.push(msg.hash)
openedLog.push(msg)
hashSet.add(msg.hash)
} catch (err) {
console.error(`Error loading message at index ${i}:`, err.message)
// Skip corrupted message, continue loading
}
}
console.log(`Loaded ${hashLog.length} messages from Hypercore (last ${config.maxCachedMessages})`)
}
/**
* Generate keypair (same as APDS)
*/
apds.generate = async () => {
const keyPair = hypercore_crypto.keyPair()
const pubkey = keyPair.publicKey.toString('base64')
const secretKey = keyPair.secretKey.toString('base64')
return pubkey + secretKey
}
/**
* Get stored keypair
*/
apds.keypair = async () => {
const node = await metaBee.get('keypair')
return node ? node.value : null
}
apds.pubkey = async () => {
const keypair = await apds.keypair()
return keypair ? keypair.substring(0, 44) : null
}
apds.privkey = async () => {
const keypair = await apds.keypair()
return keypair ? keypair.substring(44) : null
}
/**
* Hash data using crypto (ANProto compatible)
*/
apds.hash = async (data) => {
const hash = crypto.createHash('sha256').update(data, 'utf-8').digest()
return Buffer.from(hash).toString('base64')
}
/**
* Store blob by hash (content-addressable)
*/
apds.make = async (data) => {
const hash = await apds.hash(data)
// Store in Hyperbee
await blobBee.put(hash, data)
return hash
}
/**
* Retrieve blob by hash (with lazy loading from wiredove.net)
*/
apds.get = async (hash) => {
// Try local storage first
const node = await blobBee.get(hash)
if (node) return node.value
// Check if already fetching this blob
if (inFlightFetches.has(hash)) {
return await inFlightFetches.get(hash)
}
// Start new fetch and track it
const fetchPromise = (async () => {
try {
const blob = await new Promise((resolve, reject) => {
const req = https.get(`${config.remoteServer}/${hash}`, { timeout: config.fetchTimeout }, (res) => {
if (res.statusCode === 404) {
resolve(null)
return
}
if (res.statusCode !== 200) {
resolve(null)
return
}
let data = ''
res.on('data', chunk => data += chunk)
res.on('end', () => resolve(data))
res.on('error', () => resolve(null))
})
req.on('error', () => resolve(null))
req.on('timeout', () => {
req.destroy()
resolve(null)
})
})
if (blob) {
// Cache it locally for future use
await blobBee.put(hash, blob)
console.log(`📥 Fetched blob from wiredove.net: ${hash.substring(0, 16)}...`)
return blob
}
} catch (err) {
// Silently fail and return null
} finally {
// Remove from in-flight tracking
inFlightFetches.delete(hash)
}
return null
})()
inFlightFetches.set(hash, fetchPromise)
return await fetchPromise
}
/**
* Store key-value pair
*/
apds.put = async (key, value) => {
await metaBee.put(key, value)
}
/**
* Remove key
*/
apds.rm = async (key) => {
await metaBee.del(key)
}
/**
* Open/verify ANProto signature
* Compatible with wiredove's nacl.sign format
*/
apds.open = async (msg) => {
try {
if (!msg || msg.length <= 44) return null
// Extract pubkey (first 44 chars) and signed data (rest)
const publicKey = msg.substring(0, 44)
const signedData = msg.substring(44)
// Decode from base64
const publicKeyBytes = Buffer.from(publicKey, 'base64')
const signedBytes = Buffer.from(signedData, 'base64')
// Verify and open signature
const opened = nacl.sign.open(signedBytes, publicKeyBytes)
if (!opened) return null
// Return opened message (timestamp+hash)
return Buffer.from(opened).toString('utf8')
} catch (err) {
return null
}
}
/**
* Add message to append-only log
*/
apds.add = async (msg) => {
const opened = await apds.open(msg)
if (!opened) return false
const hash = await apds.make(msg)
// Check if already in log (O(1) with Set)
if (hashSet.has(hash)) return false
// Create message object
const obj = {
hash,
sig: msg,
author: msg.substring(0, 44),
opened: opened,
text: opened && opened.length >= 13 ? await apds.get(opened.substring(13)) : null,
ts: opened && opened.length >= 13 ? opened.substring(0, 13) : null
}
// Append to Hypercore
await messageCore.append(Buffer.from(JSON.stringify(obj)))
// Update cache with memory limit (LRU-style)
if (hashLog.length >= config.maxCachedMessages) {
const oldHash = hashLog.shift()
openedLog.shift()
hashSet.delete(oldHash)
}
hashLog.push(hash)
openedLog.push(obj)
hashSet.add(hash)
return true
}
/**
* Query messages (with parallel lazy text loading for first 20 results)
*/
apds.query = async (query) => {
let results = query ?
(query.startsWith('?') ?
openedLog.filter(msg => msg.text && msg.text.toUpperCase().includes(query.substring(1).replace(/%20/g, ' ').toUpperCase())) :
openedLog.filter(msg => msg.author === query || msg.hash === query)
) :
openedLog
// Lazy-load missing text for first N messages in parallel
const limit = Math.min(config.lazyLoadLimit, results.length)
const fetchPromises = []
for (let i = 0; i < limit; i++) {
const msg = results[i]
if (msg.text === null && msg.opened && msg.opened.length >= 13) {
const contentHash = msg.opened.substring(13)
// Fetch in parallel
fetchPromises.push(
apds.get(contentHash).then(text => {
if (text) msg.text = text // Update in-memory cache
})
)
}
}
// Wait for all fetches to complete in parallel
await Promise.all(fetchPromises)
return results
}
/**
* Get all unique pubkeys
*/
apds.getPubkeys = async () => {
const pubkeys = new Set()
for (const msg of openedLog) {
pubkeys.add(msg.author)
}
return Array.from(pubkeys)
}
/**
* Get latest message from pubkey
*/
apds.getLatest = async (pubkey) => {
const messages = openedLog.filter(msg => msg.author === pubkey)
return messages.length > 0 ? messages[messages.length - 1] : null
}
/**
* Get message logs
*/
apds.getHashLog = async () => hashLog
apds.getOpenedLog = async () => openedLog
/**
* Cleanup and close
*/
apds.close = async () => {
if (swarm) await swarm.destroy()
if (store) await store.close()
console.log('Wirecore-APDS closed')
}
/**
* Export Hypercore info for debugging
*/
apds.getCoreInfo = () => ({
metadataKey: metaBee.feed.key.toString('hex'),
blobsKey: blobBee.feed.key.toString('hex'),
messagesKey: messageCore.key.toString('hex'),
messageCount: messageCore.length
})