feat: add HyperDB replication to file.drop plugin
- Add database watcher to monitor @filedrop/files collection changes - Implement robust database initialization with retry logic (30 attempts) - Add replication status checking and logging on startup - Add peer connection handlers for active replication updates - Enable real-time database synchronization across P2NS network
This commit is contained in:
@@ -470,7 +470,7 @@ function createPluginChannelsForPeer(peerId, conn, mux) {
|
||||
return;
|
||||
}
|
||||
setRecreating(pluginDomain, protocol, peerId);
|
||||
peerChannel.lastReopenAttempt = now;
|
||||
peerChannel.lastReopenAttempt = Date.now();
|
||||
logDebug('ChannelManager', `Channel ${fullProtocol} fully closed, recreating for peer ${peerId.substring(0, 16)}...`);
|
||||
// Remove old channel reference
|
||||
channelInfo.peerChannels.delete(peerId);
|
||||
|
||||
@@ -14,6 +14,7 @@
|
||||
|
||||
const sdk = require('../../includes/plugins/sdk');
|
||||
const crypto = require('crypto');
|
||||
const { setupDatabaseWatcher } = require('./watcher');
|
||||
|
||||
const DRIVE_NAME = 'uploads';
|
||||
const EXPIRY_MS = 24 * 60 * 60 * 1000; // 24 hours
|
||||
@@ -425,9 +426,81 @@ async function onInit() {
|
||||
localDriveKey = driveInfo.key;
|
||||
sdk.log.info('file.drop', `Local drive key: ${localDriveKey.slice(0, 16)}...`);
|
||||
|
||||
// Wait for database
|
||||
// Wait for database to be available with retries
|
||||
let dbReady = false;
|
||||
for (let i = 0; i < 30; i++) {
|
||||
try {
|
||||
await sdk.db.ready();
|
||||
if (!sdk.db.closed) {
|
||||
dbReady = true;
|
||||
sdk.log.info('file.drop', 'HyperDB ready');
|
||||
break;
|
||||
}
|
||||
} catch (err) {
|
||||
if (err.message && err.message.includes('Database not initialized')) {
|
||||
sdk.log.debug('file.drop', `Database not ready yet, attempt ${i + 1}/30`);
|
||||
} else {
|
||||
sdk.log.warn('file.drop', `Database error on attempt ${i + 1}/30: ${err.message}`);
|
||||
}
|
||||
}
|
||||
await new Promise(resolve => setTimeout(resolve, 1000)); // Wait 1 second
|
||||
}
|
||||
|
||||
if (!dbReady) {
|
||||
sdk.log.warn('file.drop', 'Database not available after 30 seconds - plugin will continue but database operations may fail');
|
||||
}
|
||||
|
||||
// Verify replication is enabled
|
||||
try {
|
||||
const dbReplication = sdk.db.replication;
|
||||
if (dbReplication) {
|
||||
const dbStatus = dbReplication.getStatus();
|
||||
if (dbStatus && dbStatus.active) {
|
||||
sdk.log.info('file.drop', `Database replication active (${dbStatus.peers} peers)`);
|
||||
} else {
|
||||
sdk.log.info('file.drop', 'Database replication will be enabled automatically');
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
sdk.log.debug('file.drop', `Could not check database replication status: ${err.message}`);
|
||||
}
|
||||
|
||||
// Setup database watcher (async, but don't wait - it will handle errors internally)
|
||||
setupDatabaseWatcher().catch(err => {
|
||||
sdk.log.error('file.drop', `Error in database watcher setup: ${err.message}`);
|
||||
});
|
||||
|
||||
// Watch for peer connections/disconnections to trigger active replication updates
|
||||
sdk.events.on('peer-connected', async (data) => {
|
||||
// Trigger active replication update when peer connects (backup mechanism)
|
||||
// The replication manager will also handle this, but this ensures updates happen
|
||||
try {
|
||||
// Give replication a moment to establish
|
||||
await new Promise(resolve => setTimeout(resolve, 1000));
|
||||
|
||||
// Check if database replication is active and trigger update
|
||||
const dbReplication = sdk.db.replication;
|
||||
if (dbReplication && dbReplication.isActive()) {
|
||||
const dbManager = require('../../includes/plugins/db-manager');
|
||||
const core = dbManager.getPluginCore('file.drop');
|
||||
if (core && !core.closed) {
|
||||
const oldLength = core.length;
|
||||
await core.update({ wait: true });
|
||||
const newLength = core.length;
|
||||
if (newLength > oldLength) {
|
||||
sdk.log.info('file.drop', `Database updated after peer ${data.peerId.slice(0, 16)} connection, length: ${oldLength} -> ${newLength}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
sdk.log.debug('file.drop', `Error triggering replication update after peer connection: ${err.message}`);
|
||||
}
|
||||
});
|
||||
|
||||
sdk.events.on('peer-disconnected', async (data) => {
|
||||
// Peer disconnection handling - minimal implementation for consistency
|
||||
sdk.log.debug('file.drop', `Peer disconnected: ${data.peerId.slice(0, 16)}...`);
|
||||
});
|
||||
|
||||
// Run initial cleanup
|
||||
await cleanupExpiredFiles();
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
/**
|
||||
* File Drop Plugin - Database Watcher
|
||||
*
|
||||
* Handles database change watching for file metadata updates
|
||||
*/
|
||||
|
||||
const sdk = require('../../includes/plugins/sdk');
|
||||
|
||||
/**
|
||||
* Watch for database changes and log updates
|
||||
*/
|
||||
async function setupDatabaseWatcher() {
|
||||
// Wait for database to be available with retries
|
||||
let retries = 30;
|
||||
while (retries > 0) {
|
||||
try {
|
||||
await sdk.db.ready();
|
||||
if (!sdk.db.closed) {
|
||||
sdk.db.watch(async (...args) => {
|
||||
try {
|
||||
const update = args[0];
|
||||
|
||||
// Check if we have a valid update for file changes
|
||||
if (update && typeof update === 'object' && update.collection === '@filedrop/files') {
|
||||
sdk.log.info('file.drop', `Database watcher detected file change: ${JSON.stringify(update)}`);
|
||||
|
||||
// For now, just log the change. Future enhancements could include:
|
||||
// - Broadcasting file updates via WebSocket
|
||||
// - Notifying other peers of file changes
|
||||
// - Updating UI with real-time file status
|
||||
|
||||
const fileId = update.key?.id;
|
||||
if (fileId) {
|
||||
sdk.log.info('file.drop', `File metadata updated: ${fileId}`);
|
||||
} else {
|
||||
sdk.log.debug('file.drop', 'File change detected but no file ID available');
|
||||
}
|
||||
} else if (update) {
|
||||
sdk.log.debug('file.drop', `Database watcher received update for different collection: ${update.collection || 'unknown'}`);
|
||||
}
|
||||
// Ignore other update types or setup calls
|
||||
} catch (err) {
|
||||
sdk.log.error('file.drop', `Error in database watcher: ${err.message}`);
|
||||
}
|
||||
});
|
||||
sdk.log.info('file.drop', 'Database watcher set up');
|
||||
return;
|
||||
}
|
||||
} catch (err) {
|
||||
// Database not ready yet, continue waiting
|
||||
if (err.message && err.message.includes('Database not initialized')) {
|
||||
retries--;
|
||||
await new Promise(resolve => setTimeout(resolve, 1000)); // Wait 1 second
|
||||
continue;
|
||||
}
|
||||
// Other error - log and return
|
||||
sdk.log.error('file.drop', `Error setting up database watcher: ${err.message}`);
|
||||
return;
|
||||
}
|
||||
break;
|
||||
}
|
||||
if (retries === 0) {
|
||||
sdk.log.warn('file.drop', 'Database watcher setup timed out - database not available after 30 seconds');
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
setupDatabaseWatcher
|
||||
};
|
||||
Reference in New Issue
Block a user