UI and plugin fix

This commit is contained in:
Pierre
2026-01-18 16:48:06 +01:00
parent 30d156494c
commit 2a7fba7a0b
10 changed files with 1039 additions and 15 deletions
@@ -1867,6 +1867,64 @@ export class UnifiedAPIServer {
});
}
});
// Clear all metadata and trigger rescan
this.app.post('/api/metadata/clear', async (request, reply) => {
if (!this.kvClient) {
return reply.status(503).send({
status: 'error',
message: 'KV client not available'
});
}
try {
console.log('Clearing all metadata via API');
// Get all file keys and delete them
const hashIds = await this.kvClient.getAllHashIds();
let deletedCount = 0;
for (const hashId of hashIds) {
try {
await this.kvClient.delete(`/file/${hashId}`);
deletedCount++;
} catch (err) {
console.warn(`Failed to delete key /file/${hashId}:`, err);
}
}
// Also delete the index key
try {
await this.kvClient.delete('/file/__index__');
} catch (err) {
// Index might not exist
}
console.log(`Cleared ${deletedCount} file metadata entries`);
// Trigger rescan if callback is available
if (this.triggerScanCallback) {
console.log('Triggering rescan after metadata clear');
this.triggerScanCallback().catch(error => {
console.error('Error during post-clear rescan:', error);
});
}
return {
status: 'ok',
message: `Cleared ${deletedCount} files`,
deletedCount,
rescanTriggered: !!this.triggerScanCallback
};
} catch (error: any) {
console.error('Error clearing metadata:', error);
return reply.status(500).send({
status: 'error',
message: 'Failed to clear metadata',
details: error.message
});
}
});
}
/**
@@ -2083,7 +2141,7 @@ export class UnifiedAPIServer {
}
});
// Clear plugin cache
// Clear plugin cache for a specific plugin
this.app.post<{ Params: { pluginId: string } }>('/api/plugins/:pluginId/clear-cache', async (request, reply) => {
const pluginManager = this.getPluginManager?.();
if (!pluginManager) {
@@ -2099,6 +2157,55 @@ export class UnifiedAPIServer {
};
});
// Clear all plugin caches
this.app.post('/api/plugins/clear-cache', async (request, reply) => {
try {
const fs = await import('fs/promises');
const fsSync = await import('fs');
const path = await import('path');
const cacheDir = path.join(config.CACHE_FOLDER_PATH, 'plugin-cache');
let deletedFiles = 0;
let deletedDirs = 0;
// Clear plugin cache directory
if (fsSync.existsSync(cacheDir)) {
const entries = await fs.readdir(cacheDir, { withFileTypes: true });
for (const entry of entries) {
const entryPath = path.join(cacheDir, entry.name);
try {
if (entry.isDirectory()) {
await fs.rm(entryPath, { recursive: true, force: true });
deletedDirs++;
} else {
await fs.unlink(entryPath);
deletedFiles++;
}
} catch (err) {
console.warn(`Failed to delete ${entryPath}:`, err);
}
}
}
console.log(`Cleared plugin cache: ${deletedDirs} directories, ${deletedFiles} files`);
return {
status: 'ok',
message: `Cleared plugin cache`,
deletedDirs,
deletedFiles
};
} catch (error: any) {
console.error('Error clearing plugin cache:', error);
return reply.status(500).send({
status: 'error',
message: 'Failed to clear plugin cache',
details: error.message
});
}
});
// Rescan plugins (reload from container manager)
this.app.post('/api/plugins/rescan', async (request, reply) => {
const pluginManager = this.getPluginManager?.();
@@ -2378,6 +2485,134 @@ export class UnifiedAPIServer {
message: 'Use docker logs command directly for container logs'
});
});
// Add new container plugin
this.app.post<{
Body: {
pluginId: string;
image: string;
instances?: number;
resources?: { memory?: string; cpus?: number };
config?: Record<string, unknown>;
network?: boolean;
defaultQueue?: 'fast' | 'background';
}
}>('/api/plugins/containers', async (request, reply) => {
if (!this.containerManager) {
return reply.status(503).send({ error: 'Container manager not initialized' });
}
const { pluginId, image, instances, resources, config: pluginConfig, network, defaultQueue } = request.body;
if (!pluginId) {
return reply.status(400).send({ error: 'pluginId is required' });
}
if (!image) {
return reply.status(400).send({ error: 'image is required' });
}
try {
await this.containerManager.addPlugin(pluginId, image, {
instances,
resources,
config: pluginConfig,
network,
defaultQueue,
});
return {
success: true,
message: `Plugin '${pluginId}' added successfully`
};
} catch (error: any) {
return reply.status(400).send({
error: 'Failed to add plugin',
details: error.message
});
}
});
// Remove container plugin
this.app.delete<{
Params: { pluginId: string }
}>('/api/plugins/containers/:pluginId', async (request, reply) => {
if (!this.containerManager) {
return reply.status(503).send({ error: 'Container manager not initialized' });
}
const { pluginId } = request.params;
try {
await this.containerManager.removePlugin(pluginId);
return {
success: true,
message: `Plugin '${pluginId}' removed successfully`
};
} catch (error: any) {
return reply.status(500).send({
error: 'Failed to remove plugin',
details: error.message
});
}
});
// Update container plugin
this.app.put<{
Params: { pluginId: string };
Body: {
image?: string;
instances?: number;
resources?: { memory?: string; cpus?: number };
config?: Record<string, unknown>;
network?: boolean;
defaultQueue?: 'fast' | 'background';
}
}>('/api/plugins/containers/:pluginId', async (request, reply) => {
if (!this.containerManager) {
return reply.status(503).send({ error: 'Container manager not initialized' });
}
const { pluginId } = request.params;
const updates = request.body;
if (!updates || Object.keys(updates).length === 0) {
return reply.status(400).send({ error: 'No updates provided' });
}
try {
await this.containerManager.updatePluginConfig(pluginId, updates);
return {
success: true,
message: `Plugin '${pluginId}' updated successfully`
};
} catch (error: any) {
return reply.status(500).send({
error: 'Failed to update plugin',
details: error.message
});
}
});
// Restart all container plugins
this.app.post('/api/plugins/containers/restart-all', async (request, reply) => {
if (!this.containerManager) {
return reply.status(503).send({ error: 'Container manager not initialized' });
}
try {
await this.containerManager.restartAllPlugins();
return {
success: true,
message: 'All plugins restarted successfully'
};
} catch (error: any) {
return reply.status(500).send({
error: 'Failed to restart plugins',
details: error.message
});
}
});
}
/**
@@ -115,6 +115,17 @@ interface EnvConfig {
/** Docker Compose project name for grouping containers in Docker Desktop (optional). When set, plugin containers appear grouped as a stack. */
PLUGIN_STACK_NAME?: string;
/**
* Host path for plugin cache folders.
* This is the actual host filesystem path that plugin containers will mount.
* Required for persistent plugin caches across container restarts.
* Example: "/d/workspace/MetaMesh/meta-root-v2/dev/cache/meta-sort/plugin-cache"
*/
PLUGIN_CACHE_HOST_PATH?: string;
// NOTE: Plugin output files are written via WebDAV (WEBDAV_URL/plugin/<pluginId>/)
// No PLUGIN_OUTPUT_HOST_PATH needed - plugins use HTTP PUT to write files
}
// Parse POLLING_PATHS from JSON string
@@ -151,15 +162,15 @@ function parsePollingPaths(): PollingPathConfig[] | undefined {
export const config: EnvConfig = {
WATCH_FOLDER_LIST: process.env.WATCH_FOLDER_LIST,
INDEX_FOLDER_PATH: process.env.INDEX_FOLDER_PATH || '/data/cache/hash',//default /data/cache/hash
INDEX_FOLDER_PATH: process.env.INDEX_FOLDER_PATH || '/data/cache/hash-index',//default /data/cache/hash-index
UPDATE_INTERVAL_MS: parseInt(process.env.UPDATE_INTERVAL_MS || "30000", 10),//default 30000
WATCH_FOLDER_POOLING_INTERVAL_MS: parseInt(process.env.WATCH_FOLDER_POOLING_INTERVAL_MS || "0", 10),//default 0
CHMOD_FOR_NEW_FOLDER: process.env.CHMOD_FOR_NEW_FOLDER && parseInt(process.env.CHMOD_FOR_NEW_FOLDER, 8),//default 0o777 (this is the default for mkdir)
CACHE_FOLDER_PATH: process.env.CACHE_FOLDER_PATH || '/data/cache/plugins',//default /data/cache/plugins
CACHE_FOLDER_PATH: process.env.CACHE_FOLDER_PATH || '/data/cache',//default /data/cache
JELLYFIN_ENDPOINT: process.env.JELLYFIN_ENDPOINT!,
JELLYFIN_API_KEY: process.env.JELLYFIN_API_KEY!,
FUSE_API_PORT: parseInt(process.env.FUSE_API_PORT || "3000", 10),//default 3000
FUSE_API_HOST: process.env.FUSE_API_HOST || 'localhost',//default localhost
FUSE_API_HOST: process.env.FUSE_API_HOST || '0.0.0.0',//default 0.0.0.0
METADATA_FORMATS: (process.env.METADATA_FORMATS || 'meta').split(',').map(f => f.trim()).filter(f => f.length > 0),//default 'meta'
MAX_WORKER_THREADS: process.env.MAX_WORKER_THREADS ? parseInt(process.env.MAX_WORKER_THREADS, 10) : undefined,
POLLING_PATHS: parsePollingPaths(),
@@ -186,6 +197,7 @@ export const config: EnvConfig = {
CONTAINER_NETWORK: process.env.CONTAINER_NETWORK || 'meta-network',
PLUGIN_WEBDAV_URL: process.env.PLUGIN_WEBDAV_URL,
PLUGIN_STACK_NAME: process.env.PLUGIN_STACK_NAME,
PLUGIN_CACHE_HOST_PATH: process.env.PLUGIN_CACHE_HOST_PATH,
};
if(process.env.TEST!=='true') {
@@ -370,3 +370,75 @@ export function getEnabledPlugins(
export function getDefaultConfigPath(): string {
return process.env.CONTAINER_PLUGINS_CONFIG || '/app/plugins.yml';
}
/**
* Convert ContainerPluginConfig back to YAML-friendly format
*/
function configToYamlObject(config: ContainerPluginConfig): Record<string, unknown> {
const obj: Record<string, unknown> = {
enabled: config.enabled,
image: config.image,
instances: config.instances,
};
if (config.resources) {
obj.resources = config.resources;
}
if (config.config && Object.keys(config.config).length > 0) {
obj.config = config.config;
}
if (config.network) {
obj.network = config.network;
}
if (config.mounts && config.mounts.length > 0) {
obj.mounts = config.mounts;
}
if (config.healthCheck) {
obj.healthCheck = config.healthCheck;
}
if (config.defaultQueue) {
obj.defaultQueue = config.defaultQueue;
}
return obj;
}
/**
* Save container plugins configuration to YAML file
*/
export async function saveConfig(
configPath: string,
config: ContainerPluginsConfig
): Promise<void> {
// Convert config to YAML-friendly format
const yamlConfig: Record<string, unknown> = {
version: config.version || '1.0',
plugins: {} as Record<string, unknown>,
};
for (const [pluginId, pluginConfig] of Object.entries(config.plugins)) {
(yamlConfig.plugins as Record<string, unknown>)[pluginId] = configToYamlObject(pluginConfig);
}
// Stringify to YAML
const yamlContent = YAML.stringify(yamlConfig, {
indent: 2,
lineWidth: 0, // Disable line wrapping
});
// Write to file
try {
await fs.writeFile(configPath, yamlContent, 'utf-8');
} catch (error) {
throw new ConfigParseError(
'Failed to write configuration file',
configPath,
error instanceof Error ? error : undefined
);
}
}
@@ -6,10 +6,12 @@
*/
import { EventEmitter } from 'events';
import { promises as fs } from 'fs';
import { DockerClient } from './DockerClient.js';
import {
loadConfig,
getEnabledPlugins,
saveConfig,
} from './ConfigParser.js';
import type {
ContainerPluginsConfig,
@@ -298,9 +300,41 @@ export class ContainerManager extends EventEmitter {
* Sees all mounts including SMB via rclone
*
* Note: No volume mounts for /files - plugins use HTTP/WebDAV client
* Cache: Each plugin gets a dedicated cache folder mounted at /cache
*/
const mounts: Array<{ source: string; target: string; readonly: boolean }> = [];
// Add cache mount for plugin persistence (survives container restarts)
// Use PLUGIN_CACHE_HOST_PATH if set (host filesystem path for Docker bind mounts)
// Otherwise fall back to internal path (only works if container shares filesystem)
if (config.PLUGIN_CACHE_HOST_PATH) {
const pluginCacheDir = `${config.PLUGIN_CACHE_HOST_PATH}/${pluginId}`;
// Create directory on internal path (which maps to host path)
const internalCacheDir = `${config.CACHE_FOLDER_PATH}/plugin-cache/${pluginId}`;
await fs.mkdir(internalCacheDir, { recursive: true });
mounts.push({
source: pluginCacheDir,
target: '/cache',
readonly: false,
});
} else {
console.warn(`[ContainerManager] PLUGIN_CACHE_HOST_PATH not set. Plugin '${pluginId}' cache will not persist across restarts.`);
}
// NOTE: Plugin output files are written via WebDAV (WEBDAV_URL/plugin/<pluginId>/)
// No output mount needed - plugins use HTTP PUT to write files
// Add any additional mounts from plugin config
if (pluginConfig.mounts) {
for (const mount of pluginConfig.mounts) {
mounts.push({
source: mount.source,
target: mount.target,
readonly: mount.readonly ?? true,
});
}
}
// No file mounts needed - plugins access files via WebDAV
if (!this.webdavUrl) {
console.warn(`[ContainerManager] WARNING: WebDAV not configured. Plugin '${pluginId}' won't have access to files. Set PLUGIN_WEBDAV_URL.`);
@@ -687,6 +721,199 @@ export class ContainerManager extends EventEmitter {
instance.tasksFailed++;
}
}
/**
* Add a new plugin dynamically
*/
async addPlugin(
pluginId: string,
image: string,
pluginConfig?: Partial<ContainerPluginConfig>
): Promise<void> {
// Validate plugin ID format
if (!/^[a-z][a-z0-9-]*$/.test(pluginId)) {
throw new Error(
`Invalid plugin ID '${pluginId}': must be lowercase alphanumeric with hyphens, starting with a letter`
);
}
// Check if plugin already exists
if (this.instances.has(pluginId)) {
throw new Error(`Plugin '${pluginId}' already exists`);
}
console.log(`[ContainerManager] Adding new plugin '${pluginId}' with image '${image}'`);
// Build full plugin config
const fullConfig: ContainerPluginConfig = {
enabled: pluginConfig?.enabled !== false,
image,
instances: pluginConfig?.instances ?? 1,
resources: pluginConfig?.resources,
config: pluginConfig?.config,
network: pluginConfig?.network ?? false,
mounts: pluginConfig?.mounts,
healthCheck: pluginConfig?.healthCheck,
defaultQueue: pluginConfig?.defaultQueue,
};
// Update in-memory config
if (!this.pluginsConfig) {
this.pluginsConfig = { version: '1.0', plugins: {} };
}
this.pluginsConfig.plugins[pluginId] = fullConfig;
// Save to YAML file
await saveConfig(this.configPath, this.pluginsConfig);
// Spawn the plugin containers
if (fullConfig.enabled) {
await this.spawnPlugin(pluginId, fullConfig);
}
console.log(`[ContainerManager] Plugin '${pluginId}' added successfully`);
}
/**
* Remove a plugin
*/
async removePlugin(pluginId: string): Promise<void> {
console.log(`[ContainerManager] Removing plugin '${pluginId}'...`);
const instances = this.instances.get(pluginId);
if (instances) {
// Stop health check
const interval = this.healthCheckIntervals.get(pluginId);
if (interval) {
clearInterval(interval);
this.healthCheckIntervals.delete(pluginId);
}
// Stop all instances
for (const instance of instances) {
await this.stopInstance(instance);
}
this.instances.delete(pluginId);
}
// Remove from config
if (this.pluginsConfig && this.pluginsConfig.plugins[pluginId]) {
delete this.pluginsConfig.plugins[pluginId];
// Save to YAML file
await saveConfig(this.configPath, this.pluginsConfig);
}
// Reset round robin counter
this.roundRobinCounters.delete(pluginId);
console.log(`[ContainerManager] Plugin '${pluginId}' removed successfully`);
}
/**
* Restart all plugins
*/
async restartAllPlugins(): Promise<void> {
console.log('[ContainerManager] Restarting all plugins...');
const pluginIds = Array.from(this.instances.keys());
for (const pluginId of pluginIds) {
try {
await this.restartPlugin(pluginId);
} catch (error) {
console.error(`[ContainerManager] Failed to restart plugin '${pluginId}':`, error);
}
}
console.log('[ContainerManager] All plugins restarted');
}
/**
* Update plugin image and restart
*/
async updatePluginImage(pluginId: string, newImage: string): Promise<void> {
console.log(`[ContainerManager] Updating plugin '${pluginId}' to image '${newImage}'...`);
const pluginConfig = this.pluginsConfig?.plugins[pluginId];
if (!pluginConfig) {
throw new Error(`Plugin '${pluginId}' not found`);
}
// Update image in config
pluginConfig.image = newImage;
// Save to YAML file
if (this.pluginsConfig) {
await saveConfig(this.configPath, this.pluginsConfig);
}
// Restart the plugin to use the new image
await this.restartPlugin(pluginId);
console.log(`[ContainerManager] Plugin '${pluginId}' updated successfully`);
}
/**
* Update plugin configuration
*/
async updatePluginConfig(
pluginId: string,
updates: Partial<ContainerPluginConfig>
): Promise<void> {
console.log(`[ContainerManager] Updating configuration for plugin '${pluginId}'...`);
const pluginConfig = this.pluginsConfig?.plugins[pluginId];
if (!pluginConfig) {
throw new Error(`Plugin '${pluginId}' not found`);
}
// Apply updates
if (updates.image !== undefined) {
pluginConfig.image = updates.image;
}
if (updates.instances !== undefined) {
pluginConfig.instances = updates.instances;
}
if (updates.resources !== undefined) {
pluginConfig.resources = updates.resources;
}
if (updates.config !== undefined) {
pluginConfig.config = updates.config;
}
if (updates.enabled !== undefined) {
pluginConfig.enabled = updates.enabled;
}
if (updates.network !== undefined) {
pluginConfig.network = updates.network;
}
if (updates.defaultQueue !== undefined) {
pluginConfig.defaultQueue = updates.defaultQueue;
}
// Save to YAML file
if (this.pluginsConfig) {
await saveConfig(this.configPath, this.pluginsConfig);
}
// Restart the plugin to apply changes
if (this.instances.has(pluginId)) {
await this.restartPlugin(pluginId);
}
console.log(`[ContainerManager] Plugin '${pluginId}' configuration updated`);
}
/**
* Get the current plugins configuration
*/
getPluginsConfig(): ContainerPluginsConfig | null {
return this.pluginsConfig;
}
}
/**
@@ -329,6 +329,14 @@ export class WatchedFileProcessor implements FileProcessorInterface {
midHash256 = await computeMidHash256(filePath);
performanceMetrics.recordHashComputation('cid_midhash256', Math.ceil(performance.now() - computeStart));
performanceMetrics.recordCacheMiss('midhash256');
// Store in cache for future lookups
(this.fileProcessor as any).indexManager?.addFileCid(
filePath,
fileStats.size,
fileStats.mtime.toISOString(),
{ cid_midhash256: midHash256 }
);
}
// STEP 2: Initialize metadata with midhash256
@@ -181,6 +181,9 @@ export class FileProcessorPiscina implements FileAnalyzerInterface{
// Record midhash256 computation time (excludes cache hits)
performanceMetrics.recordHashComputation('cid_midhash256', computeTime);
performanceMetrics.recordCacheMiss('midhash256');
// Store in cache for future lookups
this.indexManager.addFileCid(filePath, stats.size, stats.mtime.toISOString(), { cid_midhash256: midHash256 });
}
// Initialize metadata with midhash256
@@ -358,6 +361,9 @@ export class FileProcessorPiscina implements FileAnalyzerInterface{
// Record midhash256 computation time (excludes cache hits)
performanceMetrics.recordHashComputation('cid_midhash256', computeTime);
performanceMetrics.recordCacheMiss('midhash256');
// Store in cache for future lookups
this.indexManager.addFileCid(filePath, stats.size, stats.mtime.toISOString(), { cid_midhash256: midHash256 });
}
// Initialize metadata with midhash256
@@ -169,6 +169,19 @@ export class PluginManager extends EventEmitter {
// Save state
await this.saveState();
// Send config to running container plugin
if (this.containerManager) {
const instance = this.containerManager.getHealthyInstance(pluginId);
if (instance) {
try {
await this.containerManager.sendConfig(instance, config);
console.log(`[PluginManager] Sent config update to container plugin '${pluginId}'`);
} catch (error) {
console.warn(`[PluginManager] Failed to send config to plugin '${pluginId}':`, error);
}
}
}
// Emit event
this.emit('plugin:config-changed', { pluginId, config });
}
@@ -210,6 +223,9 @@ export class PluginManager extends EventEmitter {
// Load state for plugin active status
const state = await this.loadState();
// Track plugins that need config sent
const pluginsWithPersistedConfig: Array<{ pluginId: string; config: Record<string, unknown> }> = [];
for (const pluginStatus of status.plugins) {
const manifest = this.containerManager.getPluginManifest(pluginStatus.pluginId);
if (!manifest) continue;
@@ -248,9 +264,14 @@ export class PluginManager extends EventEmitter {
this.activePluginIds.add(pluginId);
}
// Store config
const config = stateEntry?.config || {};
this.configs.set(pluginId, config);
// Store config from persisted state
const persistedConfig = stateEntry?.config || {};
this.configs.set(pluginId, persistedConfig);
// If there's persisted config, queue it for sending to the container
if (Object.keys(persistedConfig).length > 0) {
pluginsWithPersistedConfig.push({ pluginId, config: persistedConfig });
}
console.log(`[PluginManager] Loaded plugin: ${pluginId}`);
}
@@ -258,6 +279,20 @@ export class PluginManager extends EventEmitter {
// Rebuild execution order
this.rebuildExecutionOrder();
await this.saveState();
// Send persisted configs to running containers
// This ensures configs saved via API are applied after container restart
for (const { pluginId, config } of pluginsWithPersistedConfig) {
const instance = this.containerManager.getHealthyInstance(pluginId);
if (instance) {
try {
await this.containerManager.sendConfig(instance, config);
console.log(`[PluginManager] Sent persisted config to plugin '${pluginId}'`);
} catch (error) {
console.warn(`[PluginManager] Failed to send persisted config to plugin '${pluginId}':`, error);
}
}
}
}
/**