automatic db migrations on version bump
This commit is contained in:
Raven Scott
@@ -632,13 +632,14 @@ See [test-scripts/README.md](../test-scripts/README.md) for more information abo
|
||||
|
||||
5. **Check status**: Use `sdk.db.replication.getStatus()` to monitor replication health.
|
||||
|
||||
6. **Version bumping**: Bump `version` in config.json when making incompatible schema changes (e.g., removing indexes, changing key fields). This creates a new database topic.
|
||||
6. **Version bumping**: Bump `version` in config.json when making incompatible schema changes (e.g., removing indexes, changing key fields). This creates a new replicated core topic; P2NS migrates local data from the previous version on startup.
|
||||
|
||||
7. **Schema migrations**: When bumping version for schema changes:
|
||||
- Update `version` in `config.json`
|
||||
- Delete local database: `rm -rf plugin-sites/{domain}/db`
|
||||
- Restart P2NS
|
||||
- Coordinate version update with all peers
|
||||
7. **Schema migrations**: When bumping `version` in `config.json`, P2NS automatically copies all collection data from the previous version's Hypercore into the new one on startup. A manifest at `plugin-sites/{domain}/db/.p2ns-db-version.json` tracks the last migrated version.
|
||||
|
||||
- Update `version` in `config.json` (and schema in `hyperdb` if needed)
|
||||
- Restart P2NS — local data is migrated automatically
|
||||
- Coordinate the same version bump with all peers so replication uses the same core topic
|
||||
- If upgrading from before auto-migration and data appears missing, add the old version to `hyperdb.versionHistory` in `config.json` once (e.g. `"versionHistory": ["1.2.1"]`) and restart
|
||||
|
||||
## Reference
|
||||
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
/**
|
||||
* Deterministic Hypercore keypairs per plugin domain + config version.
|
||||
*/
|
||||
|
||||
const crypto = require('crypto');
|
||||
const sodium = require('sodium-native');
|
||||
|
||||
/**
|
||||
* @param {string} pluginDomain
|
||||
* @param {string} pluginVersion
|
||||
* @returns {Buffer}
|
||||
*/
|
||||
function getDatabaseSeed(pluginDomain, pluginVersion) {
|
||||
return crypto.createHash('sha256')
|
||||
.update(`plugin-db-${pluginDomain}-v${pluginVersion}`)
|
||||
.digest();
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {string} pluginDomain
|
||||
* @param {string} pluginVersion
|
||||
* @returns {{ publicKey: Buffer, secretKey: Buffer }}
|
||||
*/
|
||||
function getVersionedKeypair(pluginDomain, pluginVersion) {
|
||||
const seed = getDatabaseSeed(pluginDomain, pluginVersion);
|
||||
const publicKey = Buffer.allocUnsafe(32);
|
||||
const secretKey = Buffer.allocUnsafe(64);
|
||||
sodium.crypto_sign_seed_keypair(publicKey, secretKey, seed);
|
||||
return { publicKey, secretKey };
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {import('corestore')} store
|
||||
* @param {string} pluginDomain
|
||||
* @param {string} pluginVersion
|
||||
* @returns {import('hypercore')}
|
||||
*/
|
||||
function getVersionedCore(store, pluginDomain, pluginVersion) {
|
||||
const keypair = getVersionedKeypair(pluginDomain, pluginVersion);
|
||||
try {
|
||||
return store.get({ key: keypair.publicKey, keyPair: keypair });
|
||||
} catch (err) {
|
||||
const { logWarn } = require('../infrastructure/logger');
|
||||
logWarn('DBCoreKeys', `Keypair-based core creation failed for ${pluginDomain} v${pluginVersion}, using name-based: ${err.message}`);
|
||||
return store.get({ name: `plugin-db-${pluginDomain}` });
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
getDatabaseSeed,
|
||||
getVersionedKeypair,
|
||||
getVersionedCore
|
||||
};
|
||||
@@ -0,0 +1,103 @@
|
||||
/**
|
||||
* Load and validate generated HyperDB definition modules.
|
||||
*/
|
||||
|
||||
const path = require('path');
|
||||
const fs = require('fs').promises;
|
||||
const { logDebug, logError, logWarn } = require('../infrastructure/logger');
|
||||
|
||||
/**
|
||||
* @param {Object} def
|
||||
* @param {string} defPath
|
||||
*/
|
||||
function validateDefinition(def, defPath) {
|
||||
if (!def || typeof def !== 'object') {
|
||||
throw new Error(`Database definition is not an object. Expected object with collections, indexes, and resolve functions. Path: ${defPath}`);
|
||||
}
|
||||
|
||||
const requiredProperties = ['collections', 'indexes', 'resolveCollection', 'resolveIndex'];
|
||||
const missingProperties = requiredProperties.filter((prop) => !(prop in def));
|
||||
|
||||
if (missingProperties.length > 0) {
|
||||
throw new Error(`Database definition missing required properties: ${missingProperties.join(', ')}. Path: ${defPath}. Available properties: ${Object.keys(def).join(', ')}`);
|
||||
}
|
||||
|
||||
if (!Array.isArray(def.collections)) {
|
||||
throw new Error(`Database definition 'collections' must be an array. Got: ${typeof def.collections}. Path: ${defPath}`);
|
||||
}
|
||||
|
||||
if (!Array.isArray(def.indexes)) {
|
||||
throw new Error(`Database definition 'indexes' must be an array. Got: ${typeof def.indexes}. Path: ${defPath}`);
|
||||
}
|
||||
|
||||
if (typeof def.resolveCollection !== 'function') {
|
||||
throw new Error(`Database definition 'resolveCollection' must be a function. Got: ${typeof def.resolveCollection}. Path: ${defPath}`);
|
||||
}
|
||||
|
||||
if (typeof def.resolveIndex !== 'function') {
|
||||
throw new Error(`Database definition 'resolveIndex' must be a function. Got: ${typeof def.resolveIndex}. Path: ${defPath}`);
|
||||
}
|
||||
|
||||
if (def.version !== undefined && typeof def.version !== 'number') {
|
||||
logWarn('DBDefinition', `Database definition 'version' should be a number. Got: ${typeof def.version}. Path: ${defPath}`);
|
||||
}
|
||||
|
||||
logDebug('DBDefinition', `Database definition validated: ${def.collections.length} collections, ${def.indexes.length} indexes`);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {string} dbDir
|
||||
* @returns {Promise<Object>}
|
||||
*/
|
||||
async function loadDatabaseDefinition(dbDir) {
|
||||
const defPath = path.join(dbDir, 'index.js');
|
||||
|
||||
try {
|
||||
await fs.access(defPath);
|
||||
} catch (accessErr) {
|
||||
logError('DBDefinition', `Database definition file does not exist: ${defPath}`);
|
||||
throw new Error(`Database definition file not found at ${defPath}. HyperDB schemas need to be built.`);
|
||||
}
|
||||
|
||||
const projectRoot = path.resolve(__dirname, '../..');
|
||||
const originalNodePath = process.env.NODE_PATH || '';
|
||||
const nodeModulesPath = path.join(projectRoot, 'node_modules');
|
||||
process.env.NODE_PATH = nodeModulesPath + (originalNodePath ? path.delimiter + originalNodePath : '');
|
||||
|
||||
const Module = require('module');
|
||||
const originalResolveFilename = Module._resolveFilename;
|
||||
|
||||
let def;
|
||||
try {
|
||||
const resolvedPath = require.resolve(defPath);
|
||||
delete require.cache[resolvedPath];
|
||||
|
||||
Module._resolveFilename = function (request, parent, isMain, options) {
|
||||
if (request.startsWith('hyperdb/') && !path.isAbsolute(request)) {
|
||||
try {
|
||||
return require.resolve(request, { paths: [nodeModulesPath] });
|
||||
} catch (e) {
|
||||
// fall through
|
||||
}
|
||||
}
|
||||
return originalResolveFilename.call(this, request, parent, isMain, options);
|
||||
};
|
||||
|
||||
def = require(defPath);
|
||||
} finally {
|
||||
Module._resolveFilename = originalResolveFilename;
|
||||
process.env.NODE_PATH = originalNodePath;
|
||||
}
|
||||
|
||||
if (def && typeof def === 'object' && def.default) {
|
||||
def = def.default;
|
||||
}
|
||||
|
||||
validateDefinition(def, defPath);
|
||||
return def;
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
validateDefinition,
|
||||
loadDatabaseDefinition
|
||||
};
|
||||
+12
-164
@@ -8,9 +8,10 @@ const HyperDB = require('hyperdb');
|
||||
const Corestore = require('corestore');
|
||||
const path = require('path');
|
||||
const fs = require('fs').promises;
|
||||
const crypto = require('crypto');
|
||||
const { logDebug, logError, logInfo, logWarn } = require('../infrastructure/logger');
|
||||
const { getDatabaseWithRecovery } = require('./db-recovery');
|
||||
const { loadDatabaseDefinition } = require('./db-definition');
|
||||
const { getVersionedCore } = require('./db-core-keys');
|
||||
|
||||
// Per-plugin corestore instances
|
||||
const pluginStores = new Map();
|
||||
@@ -18,58 +19,6 @@ const databaseInstances = new Map();
|
||||
const pluginCoreKeys = new Map(); // Store core keys for replication
|
||||
const pluginCores = new Map(); // Store cores for replication
|
||||
|
||||
/**
|
||||
* Validate database definition structure
|
||||
* @param {Object} def - Database definition object
|
||||
* @param {string} defPath - Path to definition file (for error messages)
|
||||
* @throws {Error} If definition is invalid
|
||||
*/
|
||||
function validateDefinition(def, defPath) {
|
||||
if (!def || typeof def !== 'object') {
|
||||
throw new Error(`Database definition is not an object. Expected object with collections, indexes, and resolve functions. Path: ${defPath}`);
|
||||
}
|
||||
|
||||
// Check for required properties
|
||||
const requiredProperties = ['collections', 'indexes', 'resolveCollection', 'resolveIndex'];
|
||||
const missingProperties = [];
|
||||
|
||||
for (const prop of requiredProperties) {
|
||||
if (!(prop in def)) {
|
||||
missingProperties.push(prop);
|
||||
}
|
||||
}
|
||||
|
||||
if (missingProperties.length > 0) {
|
||||
throw new Error(`Database definition missing required properties: ${missingProperties.join(', ')}. Path: ${defPath}. Available properties: ${Object.keys(def).join(', ')}`);
|
||||
}
|
||||
|
||||
// Validate collections
|
||||
if (!Array.isArray(def.collections)) {
|
||||
throw new Error(`Database definition 'collections' must be an array. Got: ${typeof def.collections}. Path: ${defPath}`);
|
||||
}
|
||||
|
||||
// Validate indexes
|
||||
if (!Array.isArray(def.indexes)) {
|
||||
throw new Error(`Database definition 'indexes' must be an array. Got: ${typeof def.indexes}. Path: ${defPath}`);
|
||||
}
|
||||
|
||||
// Validate resolve functions
|
||||
if (typeof def.resolveCollection !== 'function') {
|
||||
throw new Error(`Database definition 'resolveCollection' must be a function. Got: ${typeof def.resolveCollection}. Path: ${defPath}`);
|
||||
}
|
||||
|
||||
if (typeof def.resolveIndex !== 'function') {
|
||||
throw new Error(`Database definition 'resolveIndex' must be a function. Got: ${typeof def.resolveIndex}. Path: ${defPath}`);
|
||||
}
|
||||
|
||||
// Validate version (optional but recommended)
|
||||
if (def.version !== undefined && typeof def.version !== 'number') {
|
||||
logWarn('DBManager', `Database definition 'version' should be a number. Got: ${typeof def.version}. Path: ${defPath}`);
|
||||
}
|
||||
|
||||
logDebug('DBManager', `Database definition validated: ${def.collections.length} collections, ${def.indexes.length} indexes`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize the plugin database store for a specific plugin
|
||||
* @param {string} pluginDomain - Plugin domain name
|
||||
@@ -142,86 +91,19 @@ async function getDatabase(pluginDomain, pluginDir, dbDir) {
|
||||
throw new Error(`Failed to initialize corestore for ${pluginDomain}`);
|
||||
}
|
||||
|
||||
// Load database definition
|
||||
const defPath = path.join(dbDir, 'index.js');
|
||||
logDebug('DBManager', `Loading database definition from: ${defPath}`);
|
||||
|
||||
// Check if definition file exists before trying to require it
|
||||
try {
|
||||
await fs.access(defPath);
|
||||
logDebug('DBManager', `Definition file exists: ${defPath}`);
|
||||
} catch (accessErr) {
|
||||
logError('DBManager', `Database definition file does not exist: ${defPath}`);
|
||||
logError('DBManager', `This means HyperDB schemas were not built successfully.`);
|
||||
logError('DBManager', `Solutions:`);
|
||||
logError('DBManager', `1. Delete plugin-sites/${pluginDomain}/spec/ or cache/plugin-spec/${pluginDomain}/ and restart P2NS`);
|
||||
logError('DBManager', `2. Check plugin config.json has valid hyperdb configuration`);
|
||||
logError('DBManager', `3. Verify schema builder has required dependencies installed`);
|
||||
throw new Error(`Database definition file not found at ${defPath}. HyperDB schemas need to be built. Try: rm -rf plugin-sites/${pluginDomain}/spec/ cache/plugin-spec/${pluginDomain}/ && restart P2NS`);
|
||||
}
|
||||
|
||||
|
||||
let def;
|
||||
try {
|
||||
// Clear require cache to allow hot-reloading during development
|
||||
const resolvedPath = require.resolve(defPath);
|
||||
delete require.cache[resolvedPath];
|
||||
|
||||
// For generated files in temp directories, we need to ensure module resolution works
|
||||
// The generated files require 'hyperdb/runtime' which must resolve from project root
|
||||
// We'll use a custom require that resolves modules from project root
|
||||
const projectRoot = path.resolve(__dirname, '../..');
|
||||
const originalNodePath = process.env.NODE_PATH || '';
|
||||
|
||||
// Add project's node_modules to NODE_PATH temporarily
|
||||
const nodeModulesPath = path.join(projectRoot, 'node_modules');
|
||||
process.env.NODE_PATH = nodeModulesPath + (originalNodePath ? path.delimiter + originalNodePath : '');
|
||||
|
||||
// Clear module cache for require to pick up NODE_PATH change
|
||||
const Module = require('module');
|
||||
const originalResolveFilename = Module._resolveFilename;
|
||||
|
||||
try {
|
||||
// Temporarily override module resolution to look in project root
|
||||
Module._resolveFilename = function(request, parent, isMain, options) {
|
||||
// For package subpaths like 'hyperdb/runtime', ensure we resolve from project root
|
||||
if (request.startsWith('hyperdb/') && !path.isAbsolute(request)) {
|
||||
try {
|
||||
return require.resolve(request, { paths: [nodeModulesPath] });
|
||||
} catch (e) {
|
||||
// Fall through to default resolution
|
||||
}
|
||||
}
|
||||
return originalResolveFilename.call(this, request, parent, isMain, options);
|
||||
};
|
||||
|
||||
def = require(defPath);
|
||||
} finally {
|
||||
// Restore original module resolution
|
||||
Module._resolveFilename = originalResolveFilename;
|
||||
process.env.NODE_PATH = originalNodePath;
|
||||
}
|
||||
|
||||
// Handle default export if present (ESM compatibility)
|
||||
if (def && typeof def === 'object' && def.default) {
|
||||
def = def.default;
|
||||
}
|
||||
|
||||
def = await loadDatabaseDefinition(dbDir);
|
||||
logDebug('DBManager', `Database definition loaded successfully from ${defPath}`);
|
||||
} catch (err) {
|
||||
logError('DBManager', `Could not load database definition from ${defPath}: ${err.message}`);
|
||||
if (err.code === 'MODULE_NOT_FOUND') {
|
||||
throw new Error(`Database definition file not found at ${defPath}. File exists but require() failed. Check file permissions and syntax.`);
|
||||
}
|
||||
throw new Error(`Failed to load database definition: ${err.message}. Path: ${defPath}`);
|
||||
}
|
||||
|
||||
// Validate definition structure
|
||||
logDebug('DBManager', `Validating database definition structure...`);
|
||||
try {
|
||||
validateDefinition(def, defPath);
|
||||
} catch (err) {
|
||||
logError('DBManager', `Database definition validation failed: ${err.message}`);
|
||||
throw new Error(`Invalid database definition: ${err.message}`);
|
||||
logError('DBManager', `Solutions:`);
|
||||
logError('DBManager', `1. Delete plugin-sites/${pluginDomain}/spec/ or cache/plugin-spec/${pluginDomain}/ and restart P2NS`);
|
||||
logError('DBManager', `2. Check plugin config.json has valid hyperdb configuration`);
|
||||
throw err;
|
||||
}
|
||||
|
||||
let core;
|
||||
@@ -231,14 +113,6 @@ async function getDatabase(pluginDomain, pluginDir, dbDir) {
|
||||
core = existingCore;
|
||||
logDebug('DBManager', `Reusing existing core for ${pluginDomain} with key ${core.key.toString('hex').slice(0, 16)}...`);
|
||||
} else {
|
||||
// Generate a deterministic keypair for this plugin's database
|
||||
// All peers using the same plugin domain AND version will use the same keypair
|
||||
// This ensures they share the same database regardless of storage location
|
||||
// Changing the version creates a new database topic (useful for schema migrations)
|
||||
const crypto = require('crypto');
|
||||
const sodium = require('sodium-native');
|
||||
|
||||
// Read plugin config to get version for seed
|
||||
let pluginVersion = '1.0.0';
|
||||
try {
|
||||
const configPath = path.join(pluginDir, 'config.json');
|
||||
@@ -251,37 +125,9 @@ async function getDatabase(pluginDomain, pluginDir, dbDir) {
|
||||
logWarn('DBManager', `Could not read plugin version, using default: ${err.message}`);
|
||||
}
|
||||
|
||||
// Generate a deterministic seed from the plugin domain AND version
|
||||
// This allows schema migrations by bumping version
|
||||
const seed = crypto.createHash('sha256')
|
||||
.update(`plugin-db-${pluginDomain}-v${pluginVersion}`)
|
||||
.digest();
|
||||
|
||||
logInfo('DBManager', `Database seed: plugin-db-${pluginDomain}-v${pluginVersion}`);
|
||||
|
||||
// Generate a deterministic Ed25519 keypair from the seed
|
||||
// This ensures all peers use the same keypair
|
||||
const publicKey = Buffer.allocUnsafe(32);
|
||||
const secretKey = Buffer.allocUnsafe(64);
|
||||
sodium.crypto_sign_seed_keypair(publicKey, secretKey, seed);
|
||||
|
||||
const keypair = { publicKey, secretKey };
|
||||
|
||||
logDebug('DBManager', `Creating hypercore for ${pluginDomain} with deterministic keypair`);
|
||||
logDebug('DBManager', `Core key (hex): ${publicKey.toString('hex')}`);
|
||||
|
||||
// Use keypair-based core creation
|
||||
// Corestore's get() with a keypair should create a writable core
|
||||
// If this doesn't work, we'll fall back to name-based (which is less deterministic)
|
||||
try {
|
||||
// Try using keypair directly - this should create a writable core
|
||||
core = store.get({ key: publicKey, keyPair: keypair });
|
||||
} catch (err) {
|
||||
// Fallback to name-based if keypair approach doesn't work
|
||||
logWarn('DBManager', `Keypair-based core creation failed, using name-based: ${err.message}`);
|
||||
const coreName = `plugin-db-${pluginDomain}`;
|
||||
core = store.get({ name: coreName });
|
||||
}
|
||||
core = getVersionedCore(store, pluginDomain, pluginVersion);
|
||||
logDebug('DBManager', `Core key (hex): ${core.key.toString('hex')}`);
|
||||
}
|
||||
|
||||
// Ensure core is fully ready before proceeding
|
||||
@@ -499,6 +345,8 @@ function getPluginCore(pluginDomain) {
|
||||
* Get or create database with incompatible-data recovery (sanity scan + local wipe).
|
||||
*/
|
||||
async function getDatabaseSafe(pluginDomain, pluginDir, dbDir) {
|
||||
const { ensurePluginDatabaseVersionMigration } = require('./db-version-migration');
|
||||
await ensurePluginDatabaseVersionMigration(pluginDomain, pluginDir, dbDir);
|
||||
return getDatabaseWithRecovery(pluginDomain, pluginDir, dbDir, getDatabase);
|
||||
}
|
||||
|
||||
|
||||
@@ -132,7 +132,7 @@ async function getDatabaseWithRecovery(pluginDomain, pluginDir, specDbDir, getDa
|
||||
);
|
||||
logWarn(
|
||||
'DBRecovery',
|
||||
`If this persists after reset, bump version in plugin config.json to rotate the replicated core.`
|
||||
`If this persists after reset, bump version in plugin config.json (data migrates automatically on restart).`
|
||||
);
|
||||
await resetPluginDatabaseStorage(pluginDomain, pluginDir, { reason: err.message });
|
||||
continue;
|
||||
|
||||
@@ -0,0 +1,566 @@
|
||||
/**
|
||||
* Migrate plugin HyperDB data when config.json version changes (new deterministic core key).
|
||||
*/
|
||||
|
||||
const HyperDB = require('hyperdb');
|
||||
const Corestore = require('corestore');
|
||||
const path = require('path');
|
||||
const fs = require('fs').promises;
|
||||
const { logDebug, logError, logInfo, logWarn } = require('../infrastructure/logger');
|
||||
const { loadDatabaseDefinition } = require('./db-definition');
|
||||
const { getVersionedCore } = require('./db-core-keys');
|
||||
|
||||
const MANIFEST_FILE = '.p2ns-db-version.json';
|
||||
const VERSION_HISTORY_FILE = '.p2ns-version-history.json';
|
||||
|
||||
/**
|
||||
* @param {string} pluginDir
|
||||
* @returns {string}
|
||||
*/
|
||||
function getPluginDbDir(pluginDir) {
|
||||
return path.join(pluginDir, 'db');
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {string} pluginDir
|
||||
* @returns {string}
|
||||
*/
|
||||
function getManifestPath(pluginDir) {
|
||||
return path.join(getPluginDbDir(pluginDir), MANIFEST_FILE);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {string} pluginDir
|
||||
* @returns {Promise<{ version: string, migratedAt?: string }|null>}
|
||||
*/
|
||||
async function readVersionManifest(pluginDir) {
|
||||
try {
|
||||
const raw = await fs.readFile(getManifestPath(pluginDir), 'utf8');
|
||||
const parsed = JSON.parse(raw);
|
||||
if (parsed && typeof parsed.version === 'string') {
|
||||
return parsed;
|
||||
}
|
||||
} catch (err) {
|
||||
if (err.code !== 'ENOENT') {
|
||||
logWarn('DBVersionMigration', `Could not read version manifest: ${err.message}`);
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {string} pluginDir
|
||||
* @param {string} version
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
async function writeVersionManifest(pluginDir, version) {
|
||||
const manifest = {
|
||||
version,
|
||||
migratedAt: new Date().toISOString()
|
||||
};
|
||||
await fs.mkdir(getPluginDbDir(pluginDir), { recursive: true });
|
||||
await fs.writeFile(getManifestPath(pluginDir), `${JSON.stringify(manifest, null, 2)}\n`, 'utf8');
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {string} pluginDir
|
||||
* @returns {Promise<string[]>}
|
||||
*/
|
||||
async function readVersionHistory(pluginDir) {
|
||||
try {
|
||||
const raw = await fs.readFile(path.join(getPluginDbDir(pluginDir), VERSION_HISTORY_FILE), 'utf8');
|
||||
const parsed = JSON.parse(raw);
|
||||
if (Array.isArray(parsed)) {
|
||||
return parsed.filter((v) => typeof v === 'string');
|
||||
}
|
||||
} catch (err) {
|
||||
if (err.code !== 'ENOENT') {
|
||||
logWarn('DBVersionMigration', `Could not read version history: ${err.message}`);
|
||||
}
|
||||
}
|
||||
return [];
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {string} pluginDir
|
||||
* @param {string} version
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
async function appendVersionHistory(pluginDir, version) {
|
||||
const history = await readVersionHistory(pluginDir);
|
||||
if (!history.includes(version)) {
|
||||
history.push(version);
|
||||
}
|
||||
await fs.mkdir(getPluginDbDir(pluginDir), { recursive: true });
|
||||
await fs.writeFile(
|
||||
path.join(getPluginDbDir(pluginDir), VERSION_HISTORY_FILE),
|
||||
`${JSON.stringify(history, null, 2)}\n`,
|
||||
'utf8'
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {string} pluginDir
|
||||
* @returns {Promise<string>}
|
||||
*/
|
||||
async function readPluginConfigVersion(pluginDir) {
|
||||
try {
|
||||
const configPath = path.join(pluginDir, 'config.json');
|
||||
const config = JSON.parse(await fs.readFile(configPath, 'utf8'));
|
||||
if (config.version && typeof config.version === 'string') {
|
||||
return config.version;
|
||||
}
|
||||
} catch (err) {
|
||||
logWarn('DBVersionMigration', `Could not read plugin version from config: ${err.message}`);
|
||||
}
|
||||
return '1.0.0';
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {Object} hyperdbConfig
|
||||
* @returns {string[]}
|
||||
*/
|
||||
function getCollectionIds(hyperdbConfig) {
|
||||
if (!hyperdbConfig || !hyperdbConfig.schemas || !hyperdbConfig.schemas.namespace) {
|
||||
return [];
|
||||
}
|
||||
const namespace = hyperdbConfig.schemas.namespace;
|
||||
const collections = hyperdbConfig.collections || [];
|
||||
return collections
|
||||
.filter((c) => c && c.name)
|
||||
.map((c) => `@${namespace}/${c.name}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {Object} def
|
||||
* @returns {string[]}
|
||||
*/
|
||||
function getCollectionIdsFromDefinition(def) {
|
||||
if (!def || !Array.isArray(def.collections)) {
|
||||
return [];
|
||||
}
|
||||
return def.collections
|
||||
.map((c) => {
|
||||
if (typeof c === 'string') return c;
|
||||
if (c && typeof c.name === 'string') return c.name;
|
||||
if (c && typeof c.id === 'string') return c.id;
|
||||
return null;
|
||||
})
|
||||
.filter(Boolean);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {Object} db
|
||||
* @param {string} collectionId
|
||||
* @returns {Promise<Object[]>}
|
||||
*/
|
||||
async function exportCollection(db, collectionId) {
|
||||
try {
|
||||
const stream = db.find(collectionId, {}, {});
|
||||
if (stream && typeof stream.toArray === 'function') {
|
||||
return await stream.toArray();
|
||||
}
|
||||
const docs = [];
|
||||
for await (const doc of stream) {
|
||||
docs.push(doc);
|
||||
}
|
||||
return docs;
|
||||
} catch (err) {
|
||||
logWarn('DBVersionMigration', `Export failed for ${collectionId}: ${err.message}`);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {import('corestore')} store
|
||||
* @param {string} pluginDomain
|
||||
* @param {string} pluginVersion
|
||||
* @param {string} dbDir
|
||||
* @returns {Promise<{ db: Object, core: Object }|null>}
|
||||
*/
|
||||
async function openVersionedDatabase(store, pluginDomain, pluginVersion, dbDir) {
|
||||
const core = getVersionedCore(store, pluginDomain, pluginVersion);
|
||||
await core.ready();
|
||||
|
||||
if (core.closed) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const def = await loadDatabaseDefinition(dbDir);
|
||||
const db = HyperDB.bee(core, def, {
|
||||
autoUpdate: true,
|
||||
writable: true
|
||||
});
|
||||
await db.ready();
|
||||
|
||||
return { db, core, def };
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {Object} db
|
||||
* @param {string[]} collectionIds
|
||||
* @returns {Promise<boolean>}
|
||||
*/
|
||||
async function databaseHasRecords(db, collectionIds) {
|
||||
for (const collectionId of collectionIds) {
|
||||
try {
|
||||
const stream = db.find(collectionId, {}, { limit: 1 });
|
||||
const rows = stream && typeof stream.toArray === 'function'
|
||||
? await stream.toArray()
|
||||
: [];
|
||||
if (rows.length > 0) {
|
||||
return true;
|
||||
}
|
||||
} catch (err) {
|
||||
logDebug('DBVersionMigration', `hasRecords check for ${collectionId}: ${err.message}`);
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {import('corestore')} store
|
||||
* @param {string} pluginDomain
|
||||
* @param {string} pluginVersion
|
||||
* @param {string} dbDir
|
||||
* @param {string[]} collectionIds
|
||||
* @returns {Promise<boolean>}
|
||||
*/
|
||||
async function versionCoreHasRecords(store, pluginDomain, pluginVersion, dbDir, collectionIds) {
|
||||
let opened = null;
|
||||
try {
|
||||
opened = await openVersionedDatabase(store, pluginDomain, pluginVersion, dbDir);
|
||||
if (!opened) return false;
|
||||
return await databaseHasRecords(opened.db, collectionIds);
|
||||
} catch (err) {
|
||||
logDebug('DBVersionMigration', `versionCoreHasRecords(${pluginVersion}): ${err.message}`);
|
||||
return false;
|
||||
} finally {
|
||||
if (opened) {
|
||||
try {
|
||||
if (!opened.db.closed) await opened.db.close();
|
||||
} catch (err) {
|
||||
logDebug('DBVersionMigration', `Error closing db during probe: ${err.message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {Object} sourceDb
|
||||
* @param {Object} targetDb
|
||||
* @param {string[]} collectionIds
|
||||
* @returns {Promise<{ migrated: number, collections: number }>}
|
||||
*/
|
||||
async function copyCollections(sourceDb, targetDb, collectionIds) {
|
||||
let migrated = 0;
|
||||
|
||||
for (const collectionId of collectionIds) {
|
||||
const docs = await exportCollection(sourceDb, collectionId);
|
||||
if (docs.length === 0) continue;
|
||||
|
||||
for (const doc of docs) {
|
||||
try {
|
||||
await targetDb.insert(collectionId, doc);
|
||||
migrated += 1;
|
||||
} catch (err) {
|
||||
logWarn('DBVersionMigration', `Skipped record in ${collectionId}: ${err.message}`);
|
||||
}
|
||||
}
|
||||
logInfo('DBVersionMigration', `Copied ${docs.length} record(s) from ${collectionId}`);
|
||||
}
|
||||
|
||||
if (migrated > 0 && typeof targetDb.flush === 'function') {
|
||||
await targetDb.flush();
|
||||
}
|
||||
|
||||
return { migrated, collections: collectionIds.length };
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {string} pluginDomain
|
||||
* @param {string} pluginDir
|
||||
* @param {string} dbDir
|
||||
* @param {string} fromVersion
|
||||
* @param {string} toVersion
|
||||
* @param {string[]} collectionIds
|
||||
* @returns {Promise<{ migrated: number }>}
|
||||
*/
|
||||
async function migrateBetweenVersions(pluginDomain, pluginDir, dbDir, fromVersion, toVersion, collectionIds) {
|
||||
const pluginStoreDir = getPluginDbDir(pluginDir);
|
||||
await fs.mkdir(pluginStoreDir, { recursive: true });
|
||||
|
||||
const store = new Corestore(pluginStoreDir);
|
||||
await store.ready();
|
||||
|
||||
let source = null;
|
||||
let target = null;
|
||||
|
||||
try {
|
||||
logInfo(
|
||||
'DBVersionMigration',
|
||||
`Migrating ${pluginDomain} database from v${fromVersion} to v${toVersion} (${collectionIds.length} collection(s))`
|
||||
);
|
||||
|
||||
source = await openVersionedDatabase(store, pluginDomain, fromVersion, dbDir);
|
||||
target = await openVersionedDatabase(store, pluginDomain, toVersion, dbDir);
|
||||
|
||||
const { migrated } = await copyCollections(source.db, target.db, collectionIds);
|
||||
|
||||
logInfo(
|
||||
'DBVersionMigration',
|
||||
`Migration complete for ${pluginDomain}: ${migrated} record(s) moved to v${toVersion}`
|
||||
);
|
||||
|
||||
return { migrated };
|
||||
} finally {
|
||||
if (source && !source.db.closed) {
|
||||
try {
|
||||
await source.db.close();
|
||||
} catch (err) {
|
||||
logDebug('DBVersionMigration', `Error closing source db: ${err.message}`);
|
||||
}
|
||||
}
|
||||
if (target && !target.db.closed) {
|
||||
try {
|
||||
await target.db.close();
|
||||
} catch (err) {
|
||||
logDebug('DBVersionMigration', `Error closing target db: ${err.message}`);
|
||||
}
|
||||
}
|
||||
try {
|
||||
await store.close();
|
||||
} catch (err) {
|
||||
logDebug('DBVersionMigration', `Error closing migration store: ${err.message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {string} pluginDir
|
||||
* @returns {Promise<Object|null>}
|
||||
*/
|
||||
async function readPluginHyperdbConfig(pluginDir) {
|
||||
try {
|
||||
const config = JSON.parse(await fs.readFile(path.join(pluginDir, 'config.json'), 'utf8'));
|
||||
return config.hyperdb || null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve source version when manifest is missing but an older core may still hold data.
|
||||
* @param {string} pluginDomain
|
||||
* @param {string} pluginDir
|
||||
* @param {string} dbDir
|
||||
* @param {string} targetVersion
|
||||
* @param {string[]} collectionIds
|
||||
* @param {Object|null} hyperdbConfig
|
||||
* @returns {Promise<string|null>}
|
||||
*/
|
||||
async function discoverLegacySourceVersion(pluginDomain, pluginDir, dbDir, targetVersion, collectionIds, hyperdbConfig) {
|
||||
const candidates = new Set();
|
||||
|
||||
const history = await readVersionHistory(pluginDir);
|
||||
for (const v of history) {
|
||||
if (v !== targetVersion) candidates.add(v);
|
||||
}
|
||||
|
||||
const configHistory = hyperdbConfig && Array.isArray(hyperdbConfig.versionHistory)
|
||||
? hyperdbConfig.versionHistory
|
||||
: [];
|
||||
for (const v of configHistory) {
|
||||
if (typeof v === 'string' && v !== targetVersion) {
|
||||
candidates.add(v);
|
||||
}
|
||||
}
|
||||
|
||||
if (hyperdbConfig && typeof hyperdbConfig.previousVersion === 'string' && hyperdbConfig.previousVersion !== targetVersion) {
|
||||
candidates.add(hyperdbConfig.previousVersion);
|
||||
}
|
||||
|
||||
const pluginStoreDir = getPluginDbDir(pluginDir);
|
||||
const store = new Corestore(pluginStoreDir);
|
||||
await store.ready();
|
||||
|
||||
try {
|
||||
for (const version of candidates) {
|
||||
const hasData = await versionCoreHasRecords(store, pluginDomain, version, dbDir, collectionIds);
|
||||
if (hasData) {
|
||||
logInfo('DBVersionMigration', `Discovered legacy data for ${pluginDomain} at v${version}`);
|
||||
return version;
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
try {
|
||||
await store.close();
|
||||
} catch (err) {
|
||||
logDebug('DBVersionMigration', `Error closing discovery store: ${err.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Run version migration before opening the active plugin database.
|
||||
* @param {string} pluginDomain
|
||||
* @param {string} pluginDir
|
||||
* @param {string} dbDir
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
async function ensurePluginDatabaseVersionMigration(pluginDomain, pluginDir, dbDir) {
|
||||
const targetVersion = await readPluginConfigVersion(pluginDir);
|
||||
const hyperdbConfig = await readPluginHyperdbConfig(pluginDir);
|
||||
|
||||
let collectionIds = getCollectionIds(hyperdbConfig);
|
||||
if (collectionIds.length === 0) {
|
||||
try {
|
||||
const def = await loadDatabaseDefinition(dbDir);
|
||||
collectionIds = getCollectionIdsFromDefinition(def);
|
||||
} catch (err) {
|
||||
logWarn('DBVersionMigration', `Could not resolve collections for ${pluginDomain}: ${err.message}`);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
const manifest = await readVersionManifest(pluginDir);
|
||||
|
||||
if (manifest && manifest.version === targetVersion) {
|
||||
await appendVersionHistory(pluginDir, targetVersion);
|
||||
return;
|
||||
}
|
||||
|
||||
let sourceVersion = manifest ? manifest.version : null;
|
||||
|
||||
if (!sourceVersion) {
|
||||
const pluginStoreDir = getPluginDbDir(pluginDir);
|
||||
const store = new Corestore(pluginStoreDir);
|
||||
await store.ready();
|
||||
try {
|
||||
const currentHasData = await versionCoreHasRecords(
|
||||
store,
|
||||
pluginDomain,
|
||||
targetVersion,
|
||||
dbDir,
|
||||
collectionIds
|
||||
);
|
||||
if (currentHasData) {
|
||||
logInfo(
|
||||
'DBVersionMigration',
|
||||
`No manifest for ${pluginDomain}; data already present on v${targetVersion}, recording manifest`
|
||||
);
|
||||
await writeVersionManifest(pluginDir, targetVersion);
|
||||
await appendVersionHistory(pluginDir, targetVersion);
|
||||
return;
|
||||
}
|
||||
} finally {
|
||||
try {
|
||||
await store.close();
|
||||
} catch (err) {
|
||||
logDebug('DBVersionMigration', `Error closing probe store: ${err.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
sourceVersion = await discoverLegacySourceVersion(
|
||||
pluginDomain,
|
||||
pluginDir,
|
||||
dbDir,
|
||||
targetVersion,
|
||||
collectionIds,
|
||||
hyperdbConfig
|
||||
);
|
||||
|
||||
if (!sourceVersion) {
|
||||
logDebug(
|
||||
'DBVersionMigration',
|
||||
`No prior database version to migrate for ${pluginDomain} (target v${targetVersion})`
|
||||
);
|
||||
await writeVersionManifest(pluginDir, targetVersion);
|
||||
await appendVersionHistory(pluginDir, targetVersion);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
if (sourceVersion === targetVersion) {
|
||||
await writeVersionManifest(pluginDir, targetVersion);
|
||||
await appendVersionHistory(pluginDir, targetVersion);
|
||||
return;
|
||||
}
|
||||
|
||||
const pluginStoreDir = getPluginDbDir(pluginDir);
|
||||
const probeStore = new Corestore(pluginStoreDir);
|
||||
await probeStore.ready();
|
||||
try {
|
||||
if (await versionCoreHasRecords(probeStore, pluginDomain, targetVersion, dbDir, collectionIds)) {
|
||||
logInfo(
|
||||
'DBVersionMigration',
|
||||
`Target v${targetVersion} for ${pluginDomain} already has data; recording manifest without copy`
|
||||
);
|
||||
await writeVersionManifest(pluginDir, targetVersion);
|
||||
await appendVersionHistory(pluginDir, targetVersion);
|
||||
await appendVersionHistory(pluginDir, sourceVersion);
|
||||
return;
|
||||
}
|
||||
} finally {
|
||||
try {
|
||||
await probeStore.close();
|
||||
} catch (err) {
|
||||
logDebug('DBVersionMigration', `Error closing target probe store: ${err.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
const { migrated } = await migrateBetweenVersions(
|
||||
pluginDomain,
|
||||
pluginDir,
|
||||
dbDir,
|
||||
sourceVersion,
|
||||
targetVersion,
|
||||
collectionIds
|
||||
);
|
||||
|
||||
if (migrated === 0) {
|
||||
const hadLegacy = await discoverLegacySourceVersion(
|
||||
pluginDomain,
|
||||
pluginDir,
|
||||
dbDir,
|
||||
targetVersion,
|
||||
collectionIds,
|
||||
hyperdbConfig
|
||||
);
|
||||
if (hadLegacy && hadLegacy !== sourceVersion) {
|
||||
logWarn(
|
||||
'DBVersionMigration',
|
||||
`Manifest pointed at v${sourceVersion} but data was found on v${hadLegacy}; retrying migration`
|
||||
);
|
||||
await migrateBetweenVersions(
|
||||
pluginDomain,
|
||||
pluginDir,
|
||||
dbDir,
|
||||
hadLegacy,
|
||||
targetVersion,
|
||||
collectionIds
|
||||
);
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
logError(
|
||||
'DBVersionMigration',
|
||||
`Failed to migrate ${pluginDomain} from v${sourceVersion} to v${targetVersion}: ${err.message}`
|
||||
);
|
||||
throw err;
|
||||
}
|
||||
|
||||
await writeVersionManifest(pluginDir, targetVersion);
|
||||
await appendVersionHistory(pluginDir, targetVersion);
|
||||
await appendVersionHistory(pluginDir, sourceVersion);
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
ensurePluginDatabaseVersionMigration,
|
||||
readVersionManifest,
|
||||
writeVersionManifest,
|
||||
getCollectionIds
|
||||
};
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "Global Profile",
|
||||
"version": "1.2.2",
|
||||
"version": "1.2.3",
|
||||
"domain": "global.profile",
|
||||
"enabled": true,
|
||||
"description": "Universal user identity system for all P2NS plugins with P2P profile replication",
|
||||
@@ -11,6 +11,7 @@
|
||||
"dependencies": {},
|
||||
"www": "www",
|
||||
"hyperdb": {
|
||||
"previousVersion": "1.2.2",
|
||||
"schemas": {
|
||||
"namespace": "profile",
|
||||
"structs": [
|
||||
|
||||
Reference in New Issue
Block a user