faster hash plugin version
This commit is contained in:
@@ -5,3 +5,4 @@ dist/
|
|||||||
.DS_Store
|
.DS_Store
|
||||||
CLAUDE.md
|
CLAUDE.md
|
||||||
.claude/settings.local.json
|
.claude/settings.local.json
|
||||||
|
target/
|
||||||
|
|||||||
Generated
+2084
File diff suppressed because it is too large
Load Diff
@@ -29,7 +29,6 @@ bendy = "0.3"
|
|||||||
# Serialization
|
# Serialization
|
||||||
serde = { version = "1", features = ["derive"] }
|
serde = { version = "1", features = ["derive"] }
|
||||||
serde_json = "1"
|
serde_json = "1"
|
||||||
bincode = "1.3"
|
|
||||||
|
|
||||||
# HTTP client
|
# HTTP client
|
||||||
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
|
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
|
||||||
|
|||||||
+30
-12
@@ -1,19 +1,37 @@
|
|||||||
# MetaMesh Plugin: full-hash
|
# MetaMesh Plugin: full-hash (Rust)
|
||||||
# Computes SHA-256 hash of the entire file with CSV caching
|
# High-performance file hashing with multi-algorithm support
|
||||||
|
|
||||||
FROM node:20-slim AS builder
|
FROM rust:1.83-slim AS builder
|
||||||
WORKDIR /app
|
WORKDIR /app
|
||||||
COPY package.json tsconfig.json ./
|
|
||||||
RUN npm install
|
# Install build dependencies
|
||||||
|
RUN apt-get update && apt-get install -y pkg-config libssl-dev && rm -rf /var/lib/apt/lists/*
|
||||||
|
|
||||||
|
# Copy manifests and build dependencies first (for caching)
|
||||||
|
COPY Cargo.toml ./
|
||||||
|
RUN mkdir src && echo "fn main() {}" > src/main.rs && cargo build --release && rm -rf src
|
||||||
|
|
||||||
|
# Copy source and build
|
||||||
COPY src/ ./src/
|
COPY src/ ./src/
|
||||||
RUN npm run build
|
RUN touch src/main.rs && cargo build --release
|
||||||
|
|
||||||
FROM node:20-slim
|
FROM debian:bookworm-slim
|
||||||
WORKDIR /app
|
WORKDIR /app
|
||||||
COPY package.json ./
|
|
||||||
RUN npm install --omit=dev
|
# Install runtime dependencies (curl for healthcheck, ca-certificates for HTTPS)
|
||||||
COPY --from=builder /app/dist ./dist
|
RUN apt-get update && apt-get install -y ca-certificates curl && rm -rf /var/lib/apt/lists/*
|
||||||
|
|
||||||
|
COPY --from=builder /app/target/release/metamesh-plugin-fast-full-hash /app/plugin
|
||||||
|
|
||||||
EXPOSE 8080
|
EXPOSE 8080
|
||||||
HEALTHCHECK --interval=30s --timeout=5s --start-period=5s --retries=3 \
|
HEALTHCHECK --interval=30s --timeout=5s --start-period=5s --retries=3 \
|
||||||
CMD node -e "require('http').get('http://localhost:8080/health', (r) => process.exit(r.statusCode === 200 ? 0 : 1))"
|
CMD curl -f http://localhost:8080/health || exit 1
|
||||||
CMD ["node", "dist/index.js"]
|
|
||||||
|
ENV RUST_LOG=info
|
||||||
|
ENV CACHE_PATH=/cache
|
||||||
|
|
||||||
|
# WebDAV URL for file access (set by container manager)
|
||||||
|
# Example: http://meta-sort-dev/webdav
|
||||||
|
ENV WEBDAV_URL=
|
||||||
|
|
||||||
|
CMD ["/app/plugin"]
|
||||||
|
|||||||
@@ -1,26 +0,0 @@
|
|||||||
{
|
|
||||||
"name": "@metazla/plugin-full-hash",
|
|
||||||
"version": "1.0.0",
|
|
||||||
"description": "MetaMesh plugin: Computes SHA-256 hash of the entire file",
|
|
||||||
"type": "module",
|
|
||||||
"main": "dist/index.js",
|
|
||||||
"scripts": {
|
|
||||||
"build": "tsc",
|
|
||||||
"start": "node dist/index.js",
|
|
||||||
"dev": "tsx watch src/index.ts"
|
|
||||||
},
|
|
||||||
"dependencies": {
|
|
||||||
"crc-32": "^1.2.2",
|
|
||||||
"csv-parse": "^5.5.5",
|
|
||||||
"csv-stringify": "^6.4.6",
|
|
||||||
"fastify": "^5.0.0"
|
|
||||||
},
|
|
||||||
"devDependencies": {
|
|
||||||
"@types/node": "^20.0.0",
|
|
||||||
"tsx": "^4.0.0",
|
|
||||||
"typescript": "^5.3.0"
|
|
||||||
},
|
|
||||||
"engines": {
|
|
||||||
"node": ">=20.0.0"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
+2
-2
@@ -21,8 +21,8 @@ pub struct ManifestResponse {
|
|||||||
|
|
||||||
pub async fn manifest() -> Json<ManifestResponse> {
|
pub async fn manifest() -> Json<ManifestResponse> {
|
||||||
Json(ManifestResponse {
|
Json(ManifestResponse {
|
||||||
id: "fast-full-hash",
|
id: "full-hash",
|
||||||
name: "Fast Full File Hash (Rust)",
|
name: "Full File Hash",
|
||||||
version: env!("CARGO_PKG_VERSION"),
|
version: env!("CARGO_PKG_VERSION"),
|
||||||
description: "High-performance parallel hashing with optimized I/O",
|
description: "High-performance parallel hashing with optimized I/O",
|
||||||
author: "MetaMesh",
|
author: "MetaMesh",
|
||||||
|
|||||||
+77
-26
@@ -8,7 +8,8 @@ use std::time::Instant;
|
|||||||
use crate::cache::CacheManager;
|
use crate::cache::CacheManager;
|
||||||
use crate::client::MetaCoreClient;
|
use crate::client::MetaCoreClient;
|
||||||
use crate::config::{HashAlgorithm, SharedConfig};
|
use crate::config::{HashAlgorithm, SharedConfig};
|
||||||
use crate::hasher::hash_file;
|
use crate::hasher::{hash_bytes, hash_file};
|
||||||
|
use crate::webdav::WebDavClient;
|
||||||
|
|
||||||
#[derive(Deserialize)]
|
#[derive(Deserialize)]
|
||||||
#[serde(rename_all = "camelCase")]
|
#[serde(rename_all = "camelCase")]
|
||||||
@@ -114,33 +115,74 @@ enum ProcessResult {
|
|||||||
Skipped(String),
|
Skipped(String),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// File source - either local filesystem or WebDAV
|
||||||
|
enum FileSource {
|
||||||
|
Local {
|
||||||
|
path: std::path::PathBuf,
|
||||||
|
size: u64,
|
||||||
|
mtime: i64,
|
||||||
|
},
|
||||||
|
WebDav {
|
||||||
|
data: Vec<u8>,
|
||||||
|
size: u64,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
async fn do_process_file(
|
async fn do_process_file(
|
||||||
state: Arc<ProcessState>,
|
state: Arc<ProcessState>,
|
||||||
req: ProcessRequest,
|
req: ProcessRequest,
|
||||||
) -> anyhow::Result<ProcessResult> {
|
) -> anyhow::Result<ProcessResult> {
|
||||||
let path = Path::new(&req.file_path);
|
let path = Path::new(&req.file_path);
|
||||||
|
|
||||||
// Check if file exists
|
// Extract filename from path
|
||||||
if !path.exists() {
|
|
||||||
anyhow::bail!("File not found: {}", req.file_path);
|
|
||||||
}
|
|
||||||
|
|
||||||
// Get file metadata
|
|
||||||
let metadata = std::fs::metadata(path)?;
|
|
||||||
let size = metadata.len();
|
|
||||||
let mtime = metadata
|
|
||||||
.modified()
|
|
||||||
.ok()
|
|
||||||
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
|
|
||||||
.map(|d| d.as_secs() as i64)
|
|
||||||
.unwrap_or(0);
|
|
||||||
|
|
||||||
let filename = path
|
let filename = path
|
||||||
.file_name()
|
.file_name()
|
||||||
.and_then(|n| n.to_str())
|
.and_then(|n| n.to_str())
|
||||||
.unwrap_or("unknown")
|
.unwrap_or("unknown")
|
||||||
.to_string();
|
.to_string();
|
||||||
|
|
||||||
|
// Try to get file - either from local filesystem or WebDAV
|
||||||
|
let source = if path.exists() {
|
||||||
|
// Local file available
|
||||||
|
let metadata = std::fs::metadata(path)?;
|
||||||
|
let size = metadata.len();
|
||||||
|
let mtime = metadata
|
||||||
|
.modified()
|
||||||
|
.ok()
|
||||||
|
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
|
||||||
|
.map(|d| d.as_secs() as i64)
|
||||||
|
.unwrap_or(0);
|
||||||
|
|
||||||
|
tracing::debug!("Using local file: {}", req.file_path);
|
||||||
|
FileSource::Local {
|
||||||
|
path: path.to_path_buf(),
|
||||||
|
size,
|
||||||
|
mtime,
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
// Try WebDAV
|
||||||
|
let webdav_url = std::env::var("WEBDAV_URL")
|
||||||
|
.map_err(|_| anyhow::anyhow!("File not found locally and WEBDAV_URL not set: {}", req.file_path))?;
|
||||||
|
|
||||||
|
tracing::info!("Fetching via WebDAV: {}", req.file_path);
|
||||||
|
let client = WebDavClient::new(&webdav_url);
|
||||||
|
let reader = client.get_file(&req.file_path).await?;
|
||||||
|
let size = reader.size().unwrap_or(0);
|
||||||
|
let data = reader.bytes().await?;
|
||||||
|
|
||||||
|
tracing::info!("WebDAV fetch complete: {} bytes", data.len());
|
||||||
|
FileSource::WebDav {
|
||||||
|
data,
|
||||||
|
size,
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
// Get file size and mtime for cache lookup
|
||||||
|
let (size, mtime) = match &source {
|
||||||
|
FileSource::Local { size, mtime, .. } => (*size, *mtime),
|
||||||
|
FileSource::WebDav { size, .. } => (*size, 0), // No mtime for WebDAV
|
||||||
|
};
|
||||||
|
|
||||||
// Get config
|
// Get config
|
||||||
let config = state.config.read().await.clone();
|
let config = state.config.read().await.clone();
|
||||||
|
|
||||||
@@ -194,12 +236,20 @@ async fn do_process_file(
|
|||||||
);
|
);
|
||||||
|
|
||||||
let enabled_set = to_compute.iter().copied().collect();
|
let enabled_set = to_compute.iter().copied().collect();
|
||||||
let computed = tokio::task::spawn_blocking({
|
let computed = match source {
|
||||||
let path = path.to_path_buf();
|
FileSource::Local { path, .. } => {
|
||||||
let buffer_size = config.buffer_size;
|
// Use file-based hashing for local files
|
||||||
move || hash_file(&path, &enabled_set, buffer_size)
|
let buffer_size = config.buffer_size;
|
||||||
})
|
tokio::task::spawn_blocking(move || hash_file(&path, &enabled_set, buffer_size))
|
||||||
.await??;
|
.await??
|
||||||
|
}
|
||||||
|
FileSource::WebDav { data, .. } => {
|
||||||
|
// Use bytes-based hashing for WebDAV data
|
||||||
|
let filename_clone = filename.clone();
|
||||||
|
tokio::task::spawn_blocking(move || hash_bytes(&data, &filename_clone, &enabled_set))
|
||||||
|
.await??
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
// Merge computed hashes into results
|
// Merge computed hashes into results
|
||||||
for (key, value) in &computed {
|
for (key, value) in &computed {
|
||||||
@@ -214,10 +264,11 @@ async fn do_process_file(
|
|||||||
.insert(filename.clone(), size, mtime, all_hashes)
|
.insert(filename.clone(), size, mtime, all_hashes)
|
||||||
.await;
|
.await;
|
||||||
|
|
||||||
// Save cache periodically (could be optimized with debouncing)
|
// CRITICAL: Save cache to disk BEFORE reporting done
|
||||||
if let Err(e) = state.cache.save().await {
|
// This ensures the cache file is written and synced before the callback
|
||||||
tracing::warn!("Failed to save cache: {}", e);
|
state.cache.save().await?;
|
||||||
}
|
|
||||||
|
tracing::info!("Cache saved for {} ({} hashes)", filename, results.len());
|
||||||
}
|
}
|
||||||
|
|
||||||
// Store results to meta-core
|
// Store results to meta-core
|
||||||
|
|||||||
Vendored
+259
-126
@@ -1,172 +1,256 @@
|
|||||||
use anyhow::{Context, Result};
|
use anyhow::{Context, Result};
|
||||||
use serde::{Deserialize, Serialize};
|
use std::collections::{HashMap, HashSet};
|
||||||
use std::collections::HashMap;
|
use std::fs::{self, File, OpenOptions};
|
||||||
use std::fs::{self, File};
|
use std::io::{BufRead, BufReader, Write};
|
||||||
use std::io::{BufReader, BufWriter};
|
|
||||||
use std::path::{Path, PathBuf};
|
use std::path::{Path, PathBuf};
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use tokio::sync::RwLock;
|
use tokio::sync::RwLock;
|
||||||
|
|
||||||
const CACHE_VERSION: u32 = 1;
|
use crate::config::HashAlgorithm;
|
||||||
const CACHE_FILENAME: &str = "hash_cache.bin";
|
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
/// Cache entry for a single file
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
pub struct CacheEntry {
|
pub struct CacheEntry {
|
||||||
pub filename: String,
|
pub filename: String,
|
||||||
pub size: u64,
|
pub size: u64,
|
||||||
pub mtime: i64,
|
pub mtime: i64,
|
||||||
pub hashes: HashMap<String, String>,
|
pub hash: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
/// Per-algorithm CSV index manager
|
||||||
pub struct CacheIndex {
|
///
|
||||||
pub version: u32,
|
/// Each algorithm has its own CSV file: index-{algorithm}.csv
|
||||||
pub entries: HashMap<String, CacheEntry>,
|
/// Format: path,size,mtime,hash
|
||||||
|
///
|
||||||
|
/// Uses append-only writes for efficiency (like meta-hash does)
|
||||||
|
struct AlgorithmIndex {
|
||||||
|
file_path: PathBuf,
|
||||||
|
/// In-memory cache: key = "filename|size|mtime", value = hash
|
||||||
|
entries: HashMap<String, String>,
|
||||||
|
/// Track which entries are new and need to be appended
|
||||||
|
pending_writes: Vec<CacheEntry>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Default for CacheIndex {
|
impl AlgorithmIndex {
|
||||||
fn default() -> Self {
|
fn new(cache_dir: &Path, algorithm: &str) -> Self {
|
||||||
|
let file_path = cache_dir.join(format!("index-{}.csv", algorithm));
|
||||||
Self {
|
Self {
|
||||||
version: CACHE_VERSION,
|
file_path,
|
||||||
entries: HashMap::new(),
|
entries: HashMap::new(),
|
||||||
|
pending_writes: Vec::new(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
impl CacheIndex {
|
fn make_key(filename: &str, size: u64, mtime: i64) -> String {
|
||||||
/// Generate cache key from file metadata
|
format!("{}|{}|{}", filename, size, mtime)
|
||||||
pub fn make_key(filename: &str, size: u64, mtime: i64) -> String {
|
|
||||||
format!("{}-{}-{}", filename, size, mtime)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Get cached entry if it exists and matches file metadata
|
/// Load existing entries from CSV file
|
||||||
pub fn get(&self, filename: &str, size: u64, mtime: i64) -> Option<&CacheEntry> {
|
fn load(&mut self) -> Result<()> {
|
||||||
|
if !self.file_path.exists() {
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
|
||||||
|
let file = File::open(&self.file_path)
|
||||||
|
.with_context(|| format!("Failed to open index file: {:?}", self.file_path))?;
|
||||||
|
let reader = BufReader::new(file);
|
||||||
|
|
||||||
|
let mut first_line = true;
|
||||||
|
for line in reader.lines() {
|
||||||
|
let line = line?;
|
||||||
|
|
||||||
|
// Skip header
|
||||||
|
if first_line {
|
||||||
|
first_line = false;
|
||||||
|
if line.starts_with("path,") {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Parse CSV line: path,size,mtime,hash
|
||||||
|
let parts: Vec<&str> = line.splitn(4, ',').collect();
|
||||||
|
if parts.len() == 4 {
|
||||||
|
let filename = parts[0].to_string();
|
||||||
|
let size: u64 = parts[1].parse().unwrap_or(0);
|
||||||
|
let mtime: i64 = parts[2].parse().unwrap_or(0);
|
||||||
|
let hash = parts[3].to_string();
|
||||||
|
|
||||||
|
let key = Self::make_key(&filename, size, mtime);
|
||||||
|
self.entries.insert(key, hash);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get cached hash for a file
|
||||||
|
fn get(&self, filename: &str, size: u64, mtime: i64) -> Option<&String> {
|
||||||
let key = Self::make_key(filename, size, mtime);
|
let key = Self::make_key(filename, size, mtime);
|
||||||
self.entries.get(&key)
|
self.entries.get(&key)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Insert or update a cache entry
|
/// Add a hash to the cache (queues for append)
|
||||||
pub fn insert(&mut self, filename: String, size: u64, mtime: i64, hashes: HashMap<String, String>) {
|
fn insert(&mut self, filename: String, size: u64, mtime: i64, hash: String) {
|
||||||
let key = Self::make_key(&filename, size, mtime);
|
let key = Self::make_key(&filename, size, mtime);
|
||||||
self.entries.insert(
|
|
||||||
key,
|
// Only queue for write if not already in cache
|
||||||
CacheEntry {
|
if !self.entries.contains_key(&key) {
|
||||||
filename,
|
self.pending_writes.push(CacheEntry {
|
||||||
|
filename: filename.clone(),
|
||||||
size,
|
size,
|
||||||
mtime,
|
mtime,
|
||||||
hashes,
|
hash: hash.clone(),
|
||||||
},
|
});
|
||||||
);
|
self.entries.insert(key, hash);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Append pending entries to CSV file
|
||||||
|
fn flush(&mut self) -> Result<()> {
|
||||||
|
if self.pending_writes.is_empty() {
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
|
||||||
|
// Ensure parent directory exists
|
||||||
|
if let Some(parent) = self.file_path.parent() {
|
||||||
|
fs::create_dir_all(parent)?;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check if file exists to determine if we need a header
|
||||||
|
let needs_header = !self.file_path.exists() || fs::metadata(&self.file_path)?.len() == 0;
|
||||||
|
|
||||||
|
// Open file in append mode
|
||||||
|
let mut file = OpenOptions::new()
|
||||||
|
.create(true)
|
||||||
|
.append(true)
|
||||||
|
.open(&self.file_path)?;
|
||||||
|
|
||||||
|
// Write header if needed
|
||||||
|
if needs_header {
|
||||||
|
writeln!(file, "path,size,mtime,hash")?;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Append new entries
|
||||||
|
for entry in &self.pending_writes {
|
||||||
|
writeln!(file, "{},{},{},{}", entry.filename, entry.size, entry.mtime, entry.hash)?;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Ensure data is flushed to disk
|
||||||
|
file.flush()?;
|
||||||
|
file.sync_all()?;
|
||||||
|
|
||||||
|
let count = self.pending_writes.len();
|
||||||
|
self.pending_writes.clear();
|
||||||
|
|
||||||
|
tracing::debug!("Appended {} entries to {:?}", count, self.file_path);
|
||||||
|
|
||||||
|
Ok(())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Cache manager that handles multiple algorithm indexes
|
||||||
pub struct CacheManager {
|
pub struct CacheManager {
|
||||||
cache_path: PathBuf,
|
cache_dir: PathBuf,
|
||||||
index: Arc<RwLock<CacheIndex>>,
|
/// Per-algorithm indexes
|
||||||
|
indexes: Arc<RwLock<HashMap<String, AlgorithmIndex>>>,
|
||||||
|
/// Track which algorithms have been loaded
|
||||||
|
loaded: Arc<RwLock<HashSet<String>>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl CacheManager {
|
impl CacheManager {
|
||||||
pub fn new(cache_dir: &str) -> Self {
|
pub fn new(cache_dir: &str) -> Self {
|
||||||
let cache_path = PathBuf::from(cache_dir).join(CACHE_FILENAME);
|
|
||||||
Self {
|
Self {
|
||||||
cache_path,
|
cache_dir: PathBuf::from(cache_dir),
|
||||||
index: Arc::new(RwLock::new(CacheIndex::default())),
|
indexes: Arc::new(RwLock::new(HashMap::new())),
|
||||||
|
loaded: Arc::new(RwLock::new(HashSet::new())),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Load cache from disk
|
/// Load all indexes (called at startup)
|
||||||
pub async fn load(&self) -> Result<()> {
|
pub async fn load(&self) -> Result<()> {
|
||||||
if !self.cache_path.exists() {
|
// Load indexes for all known algorithms
|
||||||
tracing::info!("No existing cache found at {:?}", self.cache_path);
|
for algo in HashAlgorithm::all() {
|
||||||
|
self.ensure_loaded(algo.key()).await?;
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Ensure a specific algorithm's index is loaded
|
||||||
|
async fn ensure_loaded(&self, algorithm: &str) -> Result<()> {
|
||||||
|
let mut loaded = self.loaded.write().await;
|
||||||
|
if loaded.contains(algorithm) {
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
|
|
||||||
let path = self.cache_path.clone();
|
let mut indexes = self.indexes.write().await;
|
||||||
let index = tokio::task::spawn_blocking(move || -> Result<CacheIndex> {
|
let mut index = AlgorithmIndex::new(&self.cache_dir, algorithm);
|
||||||
let file = File::open(&path).context("Failed to open cache file")?;
|
|
||||||
let reader = BufReader::new(file);
|
|
||||||
let index: CacheIndex =
|
|
||||||
bincode::deserialize_from(reader).context("Failed to deserialize cache")?;
|
|
||||||
|
|
||||||
// Version check
|
if let Err(e) = index.load() {
|
||||||
if index.version != CACHE_VERSION {
|
tracing::warn!("Failed to load index for {}: {}", algorithm, e);
|
||||||
tracing::warn!(
|
} else {
|
||||||
"Cache version mismatch (got {}, expected {}), starting fresh",
|
tracing::info!("Loaded {} entries for {}", index.entries.len(), algorithm);
|
||||||
index.version,
|
}
|
||||||
CACHE_VERSION
|
|
||||||
);
|
|
||||||
return Ok(CacheIndex::default());
|
|
||||||
}
|
|
||||||
|
|
||||||
Ok(index)
|
indexes.insert(algorithm.to_string(), index);
|
||||||
})
|
loaded.insert(algorithm.to_string());
|
||||||
.await??;
|
|
||||||
|
|
||||||
tracing::info!("Loaded {} cache entries", index.entries.len());
|
|
||||||
*self.index.write().await = index;
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Save cache to disk atomically
|
|
||||||
pub async fn save(&self) -> Result<()> {
|
|
||||||
let index = self.index.read().await.clone();
|
|
||||||
let cache_path = self.cache_path.clone();
|
|
||||||
|
|
||||||
tokio::task::spawn_blocking(move || -> Result<()> {
|
|
||||||
// Ensure parent directory exists
|
|
||||||
if let Some(parent) = cache_path.parent() {
|
|
||||||
fs::create_dir_all(parent)?;
|
|
||||||
}
|
|
||||||
|
|
||||||
// Write to temp file first
|
|
||||||
let temp_path = cache_path.with_extension("tmp");
|
|
||||||
{
|
|
||||||
let file = File::create(&temp_path)?;
|
|
||||||
let writer = BufWriter::new(file);
|
|
||||||
bincode::serialize_into(writer, &index)?;
|
|
||||||
}
|
|
||||||
|
|
||||||
// Atomic rename
|
|
||||||
fs::rename(&temp_path, &cache_path)?;
|
|
||||||
Ok(())
|
|
||||||
})
|
|
||||||
.await??;
|
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Get cached hashes for a file
|
/// Get cached hashes for a file (returns all available hashes)
|
||||||
pub async fn get(&self, filename: &str, size: u64, mtime: i64) -> Option<HashMap<String, String>> {
|
pub async fn get(&self, filename: &str, size: u64, mtime: i64) -> Option<HashMap<String, String>> {
|
||||||
let index = self.index.read().await;
|
let indexes = self.indexes.read().await;
|
||||||
index.get(filename, size, mtime).map(|e| e.hashes.clone())
|
let mut results = HashMap::new();
|
||||||
|
|
||||||
|
for (algo, index) in indexes.iter() {
|
||||||
|
if let Some(hash) = index.get(filename, size, mtime) {
|
||||||
|
results.insert(algo.clone(), hash.clone());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if results.is_empty() {
|
||||||
|
None
|
||||||
|
} else {
|
||||||
|
Some(results)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Store hashes in cache
|
/// Store hashes in cache (queues for append)
|
||||||
pub async fn insert(&self, filename: String, size: u64, mtime: i64, hashes: HashMap<String, String>) {
|
pub async fn insert(&self, filename: String, size: u64, mtime: i64, hashes: HashMap<String, String>) {
|
||||||
let mut index = self.index.write().await;
|
let mut indexes = self.indexes.write().await;
|
||||||
index.insert(filename, size, mtime, hashes);
|
|
||||||
|
for (algo, hash) in hashes {
|
||||||
|
// Create index if it doesn't exist
|
||||||
|
if !indexes.contains_key(&algo) {
|
||||||
|
let mut index = AlgorithmIndex::new(&self.cache_dir, &algo);
|
||||||
|
let _ = index.load(); // Ignore errors, will create new file
|
||||||
|
indexes.insert(algo.clone(), index);
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Some(index) = indexes.get_mut(&algo) {
|
||||||
|
index.insert(filename.clone(), size, mtime, hash);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Get missing hash algorithms for a file (not in cache or not computed)
|
/// Flush all pending writes to disk (MUST be called before reporting done)
|
||||||
pub async fn get_missing_algos(
|
pub async fn save(&self) -> Result<()> {
|
||||||
&self,
|
let mut indexes = self.indexes.write().await;
|
||||||
filename: &str,
|
let mut total_written = 0;
|
||||||
size: u64,
|
|
||||||
mtime: i64,
|
|
||||||
required: &[&str],
|
|
||||||
) -> Vec<String> {
|
|
||||||
let index = self.index.read().await;
|
|
||||||
let cached = index.get(filename, size, mtime);
|
|
||||||
|
|
||||||
required
|
for (algo, index) in indexes.iter_mut() {
|
||||||
.iter()
|
let pending = index.pending_writes.len();
|
||||||
.filter(|algo| {
|
if pending > 0 {
|
||||||
cached
|
index.flush().with_context(|| format!("Failed to flush index for {}", algo))?;
|
||||||
.map(|e| !e.hashes.contains_key(**algo))
|
total_written += pending;
|
||||||
.unwrap_or(true)
|
}
|
||||||
})
|
}
|
||||||
.map(|s| s.to_string())
|
|
||||||
.collect()
|
if total_written > 0 {
|
||||||
|
tracing::info!("Flushed {} total cache entries to disk", total_written);
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -176,25 +260,74 @@ mod tests {
|
|||||||
use tempfile::TempDir;
|
use tempfile::TempDir;
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn test_cache_roundtrip() {
|
async fn test_cache_append_mode() {
|
||||||
let temp_dir = TempDir::new().unwrap();
|
let temp_dir = TempDir::new().unwrap();
|
||||||
let cache = CacheManager::new(temp_dir.path().to_str().unwrap());
|
let cache = CacheManager::new(temp_dir.path().to_str().unwrap());
|
||||||
|
|
||||||
// Insert some data
|
// Insert first entry
|
||||||
let mut hashes = HashMap::new();
|
let mut hashes1 = HashMap::new();
|
||||||
hashes.insert("cid_sha2-256".to_string(), "baejbeitest".to_string());
|
hashes1.insert("cid_sha2-256".to_string(), "baejbeitest1".to_string());
|
||||||
cache
|
cache.insert("test1.mkv".to_string(), 1000, 12345, hashes1).await;
|
||||||
.insert("test.mkv".to_string(), 1000, 12345, hashes.clone())
|
|
||||||
.await;
|
|
||||||
|
|
||||||
// Save and reload
|
|
||||||
cache.save().await.unwrap();
|
cache.save().await.unwrap();
|
||||||
|
|
||||||
let cache2 = CacheManager::new(temp_dir.path().to_str().unwrap());
|
// Insert second entry
|
||||||
cache2.load().await.unwrap();
|
let mut hashes2 = HashMap::new();
|
||||||
|
hashes2.insert("cid_sha2-256".to_string(), "baejbeitest2".to_string());
|
||||||
|
cache.insert("test2.mkv".to_string(), 2000, 12346, hashes2).await;
|
||||||
|
cache.save().await.unwrap();
|
||||||
|
|
||||||
let result = cache2.get("test.mkv", 1000, 12345).await;
|
// Verify file contents (should have header + 2 entries)
|
||||||
assert!(result.is_some());
|
let csv_path = temp_dir.path().join("index-cid_sha2-256.csv");
|
||||||
assert_eq!(result.unwrap().get("cid_sha2-256").unwrap(), "baejbeitest");
|
let content = fs::read_to_string(&csv_path).unwrap();
|
||||||
|
let lines: Vec<&str> = content.lines().collect();
|
||||||
|
|
||||||
|
assert_eq!(lines.len(), 3); // header + 2 entries
|
||||||
|
assert!(lines[0].starts_with("path,"));
|
||||||
|
assert!(lines[1].contains("test1.mkv"));
|
||||||
|
assert!(lines[2].contains("test2.mkv"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_cache_reload() {
|
||||||
|
let temp_dir = TempDir::new().unwrap();
|
||||||
|
|
||||||
|
// First cache instance
|
||||||
|
{
|
||||||
|
let cache = CacheManager::new(temp_dir.path().to_str().unwrap());
|
||||||
|
let mut hashes = HashMap::new();
|
||||||
|
hashes.insert("cid_sha2-256".to_string(), "baejbeitest".to_string());
|
||||||
|
cache.insert("test.mkv".to_string(), 1000, 12345, hashes).await;
|
||||||
|
cache.save().await.unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
// Second cache instance (reload)
|
||||||
|
{
|
||||||
|
let cache = CacheManager::new(temp_dir.path().to_str().unwrap());
|
||||||
|
cache.load().await.unwrap();
|
||||||
|
|
||||||
|
let result = cache.get("test.mkv", 1000, 12345).await;
|
||||||
|
assert!(result.is_some());
|
||||||
|
assert_eq!(result.unwrap().get("cid_sha2-256").unwrap(), "baejbeitest");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_no_duplicate_writes() {
|
||||||
|
let temp_dir = TempDir::new().unwrap();
|
||||||
|
let cache = CacheManager::new(temp_dir.path().to_str().unwrap());
|
||||||
|
|
||||||
|
// Insert same entry twice
|
||||||
|
let mut hashes = HashMap::new();
|
||||||
|
hashes.insert("cid_sha2-256".to_string(), "baejbeitest".to_string());
|
||||||
|
cache.insert("test.mkv".to_string(), 1000, 12345, hashes.clone()).await;
|
||||||
|
cache.insert("test.mkv".to_string(), 1000, 12345, hashes).await;
|
||||||
|
cache.save().await.unwrap();
|
||||||
|
|
||||||
|
// Verify only one entry in file
|
||||||
|
let csv_path = temp_dir.path().join("index-cid_sha2-256.csv");
|
||||||
|
let content = fs::read_to_string(&csv_path).unwrap();
|
||||||
|
let lines: Vec<&str> = content.lines().collect();
|
||||||
|
|
||||||
|
assert_eq!(lines.len(), 2); // header + 1 entry
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -257,6 +257,137 @@ fn hash_file_single_pass(
|
|||||||
Ok((results, bt_leaf_hashes, file_size))
|
Ok((results, bt_leaf_hashes, file_size))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Hash data from bytes (for WebDAV streaming)
|
||||||
|
///
|
||||||
|
/// Similar to hash_file but works with in-memory data.
|
||||||
|
/// Note: Midhash256 requires knowing file size and reading middle section,
|
||||||
|
/// so we compute it from the bytes directly.
|
||||||
|
pub fn hash_bytes(
|
||||||
|
data: &[u8],
|
||||||
|
filename: &str,
|
||||||
|
enabled: &HashSet<HashAlgorithm>,
|
||||||
|
) -> Result<HashMap<String, String>> {
|
||||||
|
let mut results = HashMap::new();
|
||||||
|
let file_size = data.len() as u64;
|
||||||
|
|
||||||
|
// Check what we need to compute
|
||||||
|
let needs_bt = enabled.contains(&HashAlgorithm::BtPiecesRoot)
|
||||||
|
|| enabled.contains(&HashAlgorithm::BtInfoHash);
|
||||||
|
|
||||||
|
// Initialize standard hashers
|
||||||
|
let mut sha256 = enabled.contains(&HashAlgorithm::Sha256).then(Sha256::new);
|
||||||
|
let mut sha1 = enabled.contains(&HashAlgorithm::Sha1).then(Sha1::new);
|
||||||
|
let mut md5 = enabled.contains(&HashAlgorithm::Md5).then(Md5::new);
|
||||||
|
let mut crc32 = enabled.contains(&HashAlgorithm::Crc32).then(crc32fast::Hasher::new);
|
||||||
|
let mut sha3_256 = enabled.contains(&HashAlgorithm::Sha3_256).then(Sha3_256::new);
|
||||||
|
let mut sha3_384 = enabled.contains(&HashAlgorithm::Sha3_384).then(Sha3_384::new);
|
||||||
|
|
||||||
|
// Process data in chunks for BT leaf hashes
|
||||||
|
let mut bt_leaf_hashes: Vec<[u8; 32]> = Vec::new();
|
||||||
|
|
||||||
|
// Update all hashers with full data
|
||||||
|
if let Some(ref mut h) = sha256 { h.update(data); }
|
||||||
|
if let Some(ref mut h) = sha1 { h.update(data); }
|
||||||
|
if let Some(ref mut h) = md5 { h.update(data); }
|
||||||
|
if let Some(ref mut h) = crc32 { h.update(data); }
|
||||||
|
if let Some(ref mut h) = sha3_256 { h.update(data); }
|
||||||
|
if let Some(ref mut h) = sha3_384 { h.update(data); }
|
||||||
|
|
||||||
|
// Compute BT leaf hashes
|
||||||
|
if needs_bt {
|
||||||
|
if file_size == 0 {
|
||||||
|
bt_leaf_hashes.push([0u8; 32]);
|
||||||
|
} else {
|
||||||
|
for chunk in data.chunks(BT_BLOCK_SIZE) {
|
||||||
|
let mut block = [0u8; BT_BLOCK_SIZE];
|
||||||
|
block[..chunk.len()].copy_from_slice(chunk);
|
||||||
|
// Pad with zeros if needed (last block)
|
||||||
|
let mut hasher = Sha256::new();
|
||||||
|
hasher.update(&block);
|
||||||
|
bt_leaf_hashes.push(hasher.finalize().into());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let pieces_root = compute_merkle_root(&bt_leaf_hashes);
|
||||||
|
|
||||||
|
if enabled.contains(&HashAlgorithm::BtPiecesRoot) {
|
||||||
|
results.insert(
|
||||||
|
HashAlgorithm::BtPiecesRoot.key().to_string(),
|
||||||
|
to_cid(&pieces_root, HashAlgorithm::BtPiecesRoot),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
if enabled.contains(&HashAlgorithm::BtInfoHash) {
|
||||||
|
match compute_info_hash(filename, file_size, &pieces_root) {
|
||||||
|
Ok(info_hash) => {
|
||||||
|
results.insert(
|
||||||
|
HashAlgorithm::BtInfoHash.key().to_string(),
|
||||||
|
to_cid(&info_hash, HashAlgorithm::BtInfoHash),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
tracing::warn!("Failed to compute BT info hash: {}", e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Midhash256 from bytes
|
||||||
|
if enabled.contains(&HashAlgorithm::Midhash256) {
|
||||||
|
let hash = compute_midhash256_from_bytes(data);
|
||||||
|
results.insert(
|
||||||
|
HashAlgorithm::Midhash256.key().to_string(),
|
||||||
|
to_cid(&hash, HashAlgorithm::Midhash256),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Finalize standard hashes
|
||||||
|
if let Some(h) = sha256 {
|
||||||
|
results.insert(HashAlgorithm::Sha256.key().to_string(), to_cid(&h.finalize(), HashAlgorithm::Sha256));
|
||||||
|
}
|
||||||
|
if let Some(h) = sha1 {
|
||||||
|
results.insert(HashAlgorithm::Sha1.key().to_string(), to_cid(&h.finalize(), HashAlgorithm::Sha1));
|
||||||
|
}
|
||||||
|
if let Some(h) = md5 {
|
||||||
|
results.insert(HashAlgorithm::Md5.key().to_string(), to_cid(&h.finalize(), HashAlgorithm::Md5));
|
||||||
|
}
|
||||||
|
if let Some(h) = crc32 {
|
||||||
|
results.insert(HashAlgorithm::Crc32.key().to_string(), to_cid(&h.finalize().to_be_bytes(), HashAlgorithm::Crc32));
|
||||||
|
}
|
||||||
|
if let Some(h) = sha3_256 {
|
||||||
|
results.insert(HashAlgorithm::Sha3_256.key().to_string(), to_cid(&h.finalize(), HashAlgorithm::Sha3_256));
|
||||||
|
}
|
||||||
|
if let Some(h) = sha3_384 {
|
||||||
|
results.insert(HashAlgorithm::Sha3_384.key().to_string(), to_cid(&h.finalize(), HashAlgorithm::Sha3_384));
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(results)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Compute midhash256 from bytes (same algorithm as file version)
|
||||||
|
fn compute_midhash256_from_bytes(data: &[u8]) -> [u8; 32] {
|
||||||
|
use sha2::Digest;
|
||||||
|
|
||||||
|
const SAMPLE_SIZE: usize = 1024 * 1024; // 1MB
|
||||||
|
let file_size = data.len();
|
||||||
|
|
||||||
|
let mut hasher = Sha256::new();
|
||||||
|
|
||||||
|
// Always include file size
|
||||||
|
hasher.update(file_size.to_le_bytes());
|
||||||
|
|
||||||
|
if file_size <= SAMPLE_SIZE {
|
||||||
|
// Small file: hash entire content
|
||||||
|
hasher.update(data);
|
||||||
|
} else {
|
||||||
|
// Large file: hash middle 1MB
|
||||||
|
let middle_start = (file_size - SAMPLE_SIZE) / 2;
|
||||||
|
hasher.update(&data[middle_start..middle_start + SAMPLE_SIZE]);
|
||||||
|
}
|
||||||
|
|
||||||
|
hasher.finalize().into()
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|||||||
+1
-1
@@ -3,7 +3,7 @@ pub mod btih_v2;
|
|||||||
mod cid;
|
mod cid;
|
||||||
pub mod midhash256;
|
pub mod midhash256;
|
||||||
|
|
||||||
pub use algorithms::hash_file;
|
pub use algorithms::{hash_bytes, hash_file};
|
||||||
pub use btih_v2::{
|
pub use btih_v2::{
|
||||||
compute_btv2_hashes, compute_info_hash, compute_merkle_root, format_magnet_link,
|
compute_btv2_hashes, compute_info_hash, compute_merkle_root, format_magnet_link,
|
||||||
format_magnet_urn, BtV2Hashes, BT_BLOCK_SIZE,
|
format_magnet_urn, BtV2Hashes, BT_BLOCK_SIZE,
|
||||||
|
|||||||
@@ -1,72 +0,0 @@
|
|||||||
/**
|
|
||||||
* MetaMesh Plugin: full-hash
|
|
||||||
*
|
|
||||||
* ============================================================================
|
|
||||||
* PLUGIN MOUNT ARCHITECTURE - DO NOT MODIFY WITHOUT AUTHORIZATION
|
|
||||||
* ============================================================================
|
|
||||||
*
|
|
||||||
* Each plugin container has exactly 3 mounts (2 for plugins without output):
|
|
||||||
*
|
|
||||||
* 1. /files (READ-ONLY) - Shared media files, read access only
|
|
||||||
* 2. /cache (READ-WRITE) - Plugin-specific cache folder
|
|
||||||
* 3. /files/plugin/<id> (READ-WRITE) - Plugin output folder (if needed)
|
|
||||||
*
|
|
||||||
* This plugin (full-hash) only requires mounts 1 and 2:
|
|
||||||
* - /files (RO) - to read entire files for hashing
|
|
||||||
* - /cache (RW) - to cache computed hashes in CSV files
|
|
||||||
*
|
|
||||||
* Cache Structure (one CSV per hash type):
|
|
||||||
* - /cache/index-cid_sha2-256.csv
|
|
||||||
* - /cache/index-cid_sha1.csv
|
|
||||||
* - /cache/index-cid_md5.csv
|
|
||||||
* - /cache/index-cid_crc32.csv
|
|
||||||
* - /cache/index-cid_sha3-256.csv
|
|
||||||
* - /cache/index-cid_sha3-384.csv
|
|
||||||
*
|
|
||||||
* Cache key format: filename-size-mtime (recomputes if file changes)
|
|
||||||
*
|
|
||||||
* Configuration:
|
|
||||||
* - enabledHashes: comma-separated list of hash types to compute
|
|
||||||
* - Default: all available hashes
|
|
||||||
*
|
|
||||||
* SECURITY: Plugins must NEVER write to /files directly.
|
|
||||||
* All write operations go to /cache or /files/plugin/<id> only.
|
|
||||||
*
|
|
||||||
* ============================================================================
|
|
||||||
*/
|
|
||||||
|
|
||||||
import Fastify from 'fastify';
|
|
||||||
import type { HealthResponse, ProcessRequest, ProcessResponse, CallbackPayload, ConfigureRequest, ConfigureResponse } from './types.js';
|
|
||||||
import { manifest, process as processFile, configure } from './plugin.js';
|
|
||||||
|
|
||||||
const app = Fastify({ logger: true });
|
|
||||||
let ready = false;
|
|
||||||
|
|
||||||
app.get('/health', async (): Promise<HealthResponse> => ({ status: 'healthy', ready, version: manifest.version }));
|
|
||||||
app.get('/manifest', async () => manifest);
|
|
||||||
app.post<{ Body: ConfigureRequest }>('/configure', async (request): Promise<ConfigureResponse> => {
|
|
||||||
try {
|
|
||||||
configure(request.body.config);
|
|
||||||
console.log(`[${manifest.id}] Configuration updated`);
|
|
||||||
return { status: 'ok' };
|
|
||||||
} catch (error) {
|
|
||||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
|
||||||
console.error(`[${manifest.id}] Configuration error:`, errorMessage);
|
|
||||||
return { status: 'error', error: errorMessage };
|
|
||||||
}
|
|
||||||
});
|
|
||||||
app.post<{ Body: ProcessRequest }>('/process', async (request, reply) => {
|
|
||||||
const { taskId, cid, filePath, callbackUrl, metaCoreUrl } = request.body;
|
|
||||||
if (!taskId || !cid || !filePath || !callbackUrl || !metaCoreUrl) {
|
|
||||||
return reply.send({ status: 'rejected', error: 'Missing required fields' } as ProcessResponse);
|
|
||||||
}
|
|
||||||
processFile(request.body, async (payload: CallbackPayload) => {
|
|
||||||
try { await fetch(callbackUrl, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(payload) }); }
|
|
||||||
catch (error) { console.error(`[${manifest.id}] Callback error:`, error); }
|
|
||||||
}).catch(console.error);
|
|
||||||
return reply.send({ status: 'accepted' } as ProcessResponse);
|
|
||||||
});
|
|
||||||
|
|
||||||
const port = parseInt(process.env.PORT || '8080', 10);
|
|
||||||
app.listen({ port, host: '0.0.0.0' }).then(() => { ready = true; console.log(`[${manifest.id}] Listening on port ${port}`); });
|
|
||||||
process.on('SIGTERM', async () => { ready = false; await app.close(); process.exit(0); });
|
|
||||||
@@ -7,3 +7,4 @@ pub mod cache;
|
|||||||
pub mod client;
|
pub mod client;
|
||||||
pub mod config;
|
pub mod config;
|
||||||
pub mod hasher;
|
pub mod hasher;
|
||||||
|
pub mod webdav;
|
||||||
|
|||||||
@@ -1,73 +0,0 @@
|
|||||||
/**
|
|
||||||
* meta-core API Client for writing metadata
|
|
||||||
* Fails gracefully if meta-core is unavailable (logs warning, returns without throwing)
|
|
||||||
*/
|
|
||||||
|
|
||||||
export class MetaCoreClient {
|
|
||||||
constructor(private baseUrl: string) {}
|
|
||||||
|
|
||||||
private async safeFetch(url: string, options: RequestInit): Promise<Response | null> {
|
|
||||||
try {
|
|
||||||
return await fetch(url, { ...options, signal: AbortSignal.timeout(5000) });
|
|
||||||
} catch (error) {
|
|
||||||
console.warn(`[MetaCoreClient] Warning: meta-core unavailable at ${this.baseUrl}`);
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
async setProperty(hashId: string, key: string, value: string): Promise<void> {
|
|
||||||
const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}/${key}`, {
|
|
||||||
method: 'PUT',
|
|
||||||
headers: { 'Content-Type': 'application/json' },
|
|
||||||
body: JSON.stringify({ value }),
|
|
||||||
});
|
|
||||||
if (response && !response.ok) {
|
|
||||||
console.warn(`[MetaCoreClient] Failed to set property: ${response.status}`);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
async getProperty(hashId: string, key: string): Promise<string | null> {
|
|
||||||
const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}/${key}`, { method: 'GET' });
|
|
||||||
if (!response || response.status === 404) return null;
|
|
||||||
if (!response.ok) return null;
|
|
||||||
const data = await response.json() as { value?: string };
|
|
||||||
return data.value ?? null;
|
|
||||||
}
|
|
||||||
|
|
||||||
async mergeMetadata(hashId: string, metadata: Record<string, string>): Promise<void> {
|
|
||||||
const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}`, {
|
|
||||||
method: 'PATCH',
|
|
||||||
headers: { 'Content-Type': 'application/json' },
|
|
||||||
body: JSON.stringify(metadata),
|
|
||||||
});
|
|
||||||
if (response && !response.ok) {
|
|
||||||
console.warn(`[MetaCoreClient] Failed to merge metadata: ${response.status}`);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
async deleteProperty(hashId: string, key: string): Promise<void> {
|
|
||||||
const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}/${key}`, { method: 'DELETE' });
|
|
||||||
if (response && !response.ok && response.status !== 404) {
|
|
||||||
console.warn(`[MetaCoreClient] Failed to delete property: ${response.status}`);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
async addToSet(hashId: string, key: string, value: string): Promise<void> {
|
|
||||||
const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}/_add/${key}`, {
|
|
||||||
method: 'POST',
|
|
||||||
headers: { 'Content-Type': 'application/json' },
|
|
||||||
body: JSON.stringify({ value }),
|
|
||||||
});
|
|
||||||
if (response && !response.ok) {
|
|
||||||
console.warn(`[MetaCoreClient] Failed to add to set: ${response.status}`);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
async getMetadata(hashId: string): Promise<Record<string, string>> {
|
|
||||||
const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}`, { method: 'GET' });
|
|
||||||
if (!response || response.status === 404) return {};
|
|
||||||
if (!response.ok) return {};
|
|
||||||
const data = await response.json() as { metadata?: Record<string, string> };
|
|
||||||
return data.metadata ?? {};
|
|
||||||
}
|
|
||||||
}
|
|
||||||
-510
@@ -1,510 +0,0 @@
|
|||||||
/**
|
|
||||||
* Full Hash Plugin
|
|
||||||
*
|
|
||||||
* Computes multiple hashes of the entire file with CSV-based caching.
|
|
||||||
* Uses the same CID format as the meta-hash library.
|
|
||||||
*
|
|
||||||
* Available hashes (all enabled by default):
|
|
||||||
* - cid_sha2-256: SHA-256 hash
|
|
||||||
* - cid_sha1: SHA-1 hash
|
|
||||||
* - cid_md5: MD5 hash
|
|
||||||
* - cid_crc32: CRC32 checksum
|
|
||||||
* - cid_sha3-256: SHA3-256 hash
|
|
||||||
* - cid_sha3-384: SHA3-384 hash
|
|
||||||
*
|
|
||||||
* Note: midhash256 is NOT computed here - it's computed in-process for fast file ID.
|
|
||||||
* Note: btih_v2 requires special handling (BitTorrent v2 merkle tree) - not implemented yet.
|
|
||||||
*
|
|
||||||
* Cache:
|
|
||||||
* - Stores computed hashes in CSV files: /cache/index-{hashType}.csv
|
|
||||||
* - Cache key is: filename-size-mtime (recomputes if file changes)
|
|
||||||
*/
|
|
||||||
|
|
||||||
import { createReadStream, promises as fs } from 'fs';
|
|
||||||
import { createHash, Hash } from 'crypto';
|
|
||||||
import { stat } from 'fs/promises';
|
|
||||||
import path from 'path';
|
|
||||||
import { parse } from 'csv-parse';
|
|
||||||
import { stringify } from 'csv-stringify/sync';
|
|
||||||
import CRC32 from 'crc-32';
|
|
||||||
import type { PluginManifest, ProcessRequest, CallbackPayload } from './types.js';
|
|
||||||
import { MetaCoreClient } from './meta-core-client.js';
|
|
||||||
import { createWebDAVClient, WebDAVClient } from './webdav-client.js';
|
|
||||||
|
|
||||||
// Initialize WebDAV client if WEBDAV_URL is set
|
|
||||||
const webdavClient = createWebDAVClient();
|
|
||||||
if (webdavClient) {
|
|
||||||
console.log('[full-hash] Using WebDAV for file access');
|
|
||||||
} else {
|
|
||||||
console.log('[full-hash] Using direct filesystem access');
|
|
||||||
}
|
|
||||||
|
|
||||||
// Cache folder path (mounted as /cache in container)
|
|
||||||
const CACHE_FOLDER_PATH = globalThis.process.env.CACHE_PATH || '/cache';
|
|
||||||
|
|
||||||
// Hash algorithm definitions
|
|
||||||
const HASH_ALGORITHMS = {
|
|
||||||
'cid_sha2-256': { crypto: 'sha256', prefix: 'baejbei' },
|
|
||||||
'cid_sha1': { crypto: 'sha1', prefix: 'baeir' },
|
|
||||||
'cid_md5': { crypto: 'md5', prefix: 'bahkq' },
|
|
||||||
'cid_crc32': { crypto: null, prefix: 'bagzafmq' }, // Special handling
|
|
||||||
'cid_sha3-256': { crypto: 'sha3-256', prefix: 'baelb' },
|
|
||||||
'cid_sha3-384': { crypto: 'sha3-384', prefix: 'baekr' },
|
|
||||||
} as const;
|
|
||||||
|
|
||||||
type HashAlgorithm = keyof typeof HASH_ALGORITHMS;
|
|
||||||
|
|
||||||
// All available hash types (default: all enabled)
|
|
||||||
const ALL_HASH_TYPES: HashAlgorithm[] = Object.keys(HASH_ALGORITHMS) as HashAlgorithm[];
|
|
||||||
|
|
||||||
// Plugin configuration (set via /configure endpoint)
|
|
||||||
let enabledHashes: HashAlgorithm[] = [...ALL_HASH_TYPES];
|
|
||||||
|
|
||||||
// Per-hash-type caches
|
|
||||||
const hashCaches: Map<HashAlgorithm, Map<string, string>> = new Map();
|
|
||||||
const cacheLoaded: Set<HashAlgorithm> = new Set();
|
|
||||||
const cacheChanges: Set<HashAlgorithm> = new Set();
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Get index file path for a hash type
|
|
||||||
*/
|
|
||||||
function getIndexPath(hashType: HashAlgorithm): string {
|
|
||||||
return path.join(CACHE_FOLDER_PATH, `index-${hashType}.csv`);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Load cache for a specific hash type
|
|
||||||
*/
|
|
||||||
async function loadHashCache(hashType: HashAlgorithm): Promise<void> {
|
|
||||||
if (cacheLoaded.has(hashType)) return;
|
|
||||||
|
|
||||||
try {
|
|
||||||
await fs.mkdir(CACHE_FOLDER_PATH, { recursive: true });
|
|
||||||
|
|
||||||
const indexPath = getIndexPath(hashType);
|
|
||||||
const exists = await fs.access(indexPath).then(() => true).catch(() => false);
|
|
||||||
|
|
||||||
if (!exists) {
|
|
||||||
hashCaches.set(hashType, new Map());
|
|
||||||
cacheLoaded.add(hashType);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
const cache = new Map<string, string>();
|
|
||||||
const records: Array<{ path: string; size: string; mtime: string; hash: string }> = await new Promise((resolve, reject) => {
|
|
||||||
const results: Array<{ path: string; size: string; mtime: string; hash: string }> = [];
|
|
||||||
createReadStream(indexPath)
|
|
||||||
.pipe(parse({ columns: true, skip_empty_lines: true }))
|
|
||||||
.on('data', (record) => results.push(record))
|
|
||||||
.on('end', () => resolve(results))
|
|
||||||
.on('error', reject);
|
|
||||||
});
|
|
||||||
|
|
||||||
for (const record of records) {
|
|
||||||
const cacheKey = `${record.path}-${record.size}-${record.mtime}`;
|
|
||||||
cache.set(cacheKey, record.hash);
|
|
||||||
}
|
|
||||||
|
|
||||||
hashCaches.set(hashType, cache);
|
|
||||||
cacheLoaded.add(hashType);
|
|
||||||
console.log(`[full-hash] Loaded ${cache.size} entries for ${hashType}`);
|
|
||||||
} catch (error) {
|
|
||||||
console.error(`[full-hash] Error loading cache for ${hashType}:`, error);
|
|
||||||
hashCaches.set(hashType, new Map());
|
|
||||||
cacheLoaded.add(hashType);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Save cache for a specific hash type
|
|
||||||
*/
|
|
||||||
async function saveHashCache(hashType: HashAlgorithm): Promise<void> {
|
|
||||||
if (!cacheChanges.has(hashType)) return;
|
|
||||||
|
|
||||||
try {
|
|
||||||
const cache = hashCaches.get(hashType);
|
|
||||||
if (!cache || cache.size === 0) return;
|
|
||||||
|
|
||||||
const entries: Array<{ path: string; size: string; mtime: string; hash: string }> = [];
|
|
||||||
for (const [key, hash] of cache.entries()) {
|
|
||||||
const [filename, size, mtime] = key.split('-');
|
|
||||||
// Handle filenames that might contain dashes
|
|
||||||
const parts = key.split('-');
|
|
||||||
const mtimeIdx = parts.findIndex(p => p.includes('T') && p.includes('Z'));
|
|
||||||
if (mtimeIdx >= 2) {
|
|
||||||
const filename = parts.slice(0, mtimeIdx - 1).join('-');
|
|
||||||
const size = parts[mtimeIdx - 1];
|
|
||||||
const mtime = parts.slice(mtimeIdx).join('-');
|
|
||||||
entries.push({ path: filename, size, mtime, hash });
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
const csvString = stringify(entries, {
|
|
||||||
header: true,
|
|
||||||
columns: ['path', 'size', 'mtime', 'hash'],
|
|
||||||
});
|
|
||||||
|
|
||||||
await fs.writeFile(getIndexPath(hashType), csvString);
|
|
||||||
cacheChanges.delete(hashType);
|
|
||||||
console.log(`[full-hash] Saved ${entries.length} entries for ${hashType}`);
|
|
||||||
} catch (error) {
|
|
||||||
console.error(`[full-hash] Error saving cache for ${hashType}:`, error);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Get cached hash value
|
|
||||||
*/
|
|
||||||
function getCachedHash(hashType: HashAlgorithm, filename: string, size: number, mtime: string): string | null {
|
|
||||||
const cache = hashCaches.get(hashType);
|
|
||||||
if (!cache) return null;
|
|
||||||
const cacheKey = `${filename}-${size}-${mtime}`;
|
|
||||||
return cache.get(cacheKey) || null;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Add hash to cache
|
|
||||||
*/
|
|
||||||
function addToCache(hashType: HashAlgorithm, filename: string, size: number, mtime: string, hash: string): void {
|
|
||||||
let cache = hashCaches.get(hashType);
|
|
||||||
if (!cache) {
|
|
||||||
cache = new Map();
|
|
||||||
hashCaches.set(hashType, cache);
|
|
||||||
}
|
|
||||||
const cacheKey = `${filename}-${size}-${mtime}`;
|
|
||||||
cache.set(cacheKey, hash);
|
|
||||||
cacheChanges.add(hashType);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Encode bytes to base32 (RFC 4648, lowercase, no padding)
|
|
||||||
*/
|
|
||||||
function base32Encode(buffer: Buffer): string {
|
|
||||||
const alphabet = 'abcdefghijklmnopqrstuvwxyz234567';
|
|
||||||
let result = '';
|
|
||||||
let bits = 0;
|
|
||||||
let value = 0;
|
|
||||||
|
|
||||||
for (const byte of buffer) {
|
|
||||||
value = (value << 8) | byte;
|
|
||||||
bits += 8;
|
|
||||||
while (bits >= 5) {
|
|
||||||
bits -= 5;
|
|
||||||
result += alphabet[(value >> bits) & 0x1f];
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if (bits > 0) {
|
|
||||||
result += alphabet[(value << (5 - bits)) & 0x1f];
|
|
||||||
}
|
|
||||||
|
|
||||||
return result;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Format hash as CID with appropriate prefix
|
|
||||||
*/
|
|
||||||
function formatAsCid(hashType: HashAlgorithm, hashHex: string): string {
|
|
||||||
const { prefix } = HASH_ALGORITHMS[hashType];
|
|
||||||
const hashBuffer = Buffer.from(hashHex, 'hex');
|
|
||||||
const base32Hash = base32Encode(hashBuffer);
|
|
||||||
return `${prefix}${base32Hash}`;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Compute all hashes for a file in a single pass (filesystem mode)
|
|
||||||
*/
|
|
||||||
async function computeHashesFromFilesystem(filePath: string, hashTypes: HashAlgorithm[]): Promise<Map<HashAlgorithm, string>> {
|
|
||||||
return new Promise((resolve, reject) => {
|
|
||||||
const results = new Map<HashAlgorithm, string>();
|
|
||||||
const hashers: Map<HashAlgorithm, Hash> = new Map();
|
|
||||||
let crc32Value = 0;
|
|
||||||
const needsCrc32 = hashTypes.includes('cid_crc32');
|
|
||||||
|
|
||||||
// Initialize crypto hashers
|
|
||||||
for (const hashType of hashTypes) {
|
|
||||||
const algo = HASH_ALGORITHMS[hashType];
|
|
||||||
if (algo.crypto) {
|
|
||||||
try {
|
|
||||||
hashers.set(hashType, createHash(algo.crypto));
|
|
||||||
} catch (e) {
|
|
||||||
console.warn(`[full-hash] Hash algorithm ${algo.crypto} not available, skipping ${hashType}`);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
const stream = createReadStream(filePath);
|
|
||||||
|
|
||||||
stream.on('data', (chunk: Buffer | string) => {
|
|
||||||
const data = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
|
|
||||||
// Update all crypto hashers
|
|
||||||
for (const [, hasher] of hashers) {
|
|
||||||
hasher.update(data);
|
|
||||||
}
|
|
||||||
// Update CRC32
|
|
||||||
if (needsCrc32) {
|
|
||||||
crc32Value = CRC32.buf(data, crc32Value);
|
|
||||||
}
|
|
||||||
});
|
|
||||||
|
|
||||||
stream.on('end', () => {
|
|
||||||
// Finalize crypto hashes
|
|
||||||
for (const [hashType, hasher] of hashers) {
|
|
||||||
const hexHash = hasher.digest('hex');
|
|
||||||
const cid = formatAsCid(hashType, hexHash);
|
|
||||||
results.set(hashType, cid);
|
|
||||||
}
|
|
||||||
// Finalize CRC32
|
|
||||||
if (needsCrc32) {
|
|
||||||
// Convert to unsigned 32-bit and then to hex
|
|
||||||
const unsigned = crc32Value >>> 0;
|
|
||||||
const hexHash = unsigned.toString(16).padStart(8, '0');
|
|
||||||
const cid = formatAsCid('cid_crc32', hexHash);
|
|
||||||
results.set('cid_crc32', cid);
|
|
||||||
}
|
|
||||||
resolve(results);
|
|
||||||
});
|
|
||||||
|
|
||||||
stream.on('error', reject);
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Compute all hashes for a file via WebDAV streaming
|
|
||||||
*/
|
|
||||||
async function computeHashesFromWebDAV(client: WebDAVClient, filePath: string, hashTypes: HashAlgorithm[]): Promise<Map<HashAlgorithm, string>> {
|
|
||||||
const results = new Map<HashAlgorithm, string>();
|
|
||||||
const hashers: Map<HashAlgorithm, Hash> = new Map();
|
|
||||||
let crc32Value = 0;
|
|
||||||
const needsCrc32 = hashTypes.includes('cid_crc32');
|
|
||||||
|
|
||||||
// Initialize crypto hashers
|
|
||||||
for (const hashType of hashTypes) {
|
|
||||||
const algo = HASH_ALGORITHMS[hashType];
|
|
||||||
if (algo.crypto) {
|
|
||||||
try {
|
|
||||||
hashers.set(hashType, createHash(algo.crypto));
|
|
||||||
} catch (e) {
|
|
||||||
console.warn(`[full-hash] Hash algorithm ${algo.crypto} not available, skipping ${hashType}`);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Stream file from WebDAV
|
|
||||||
for await (const chunk of client.streamFile(filePath)) {
|
|
||||||
// Update all crypto hashers
|
|
||||||
for (const [, hasher] of hashers) {
|
|
||||||
hasher.update(chunk);
|
|
||||||
}
|
|
||||||
// Update CRC32
|
|
||||||
if (needsCrc32) {
|
|
||||||
crc32Value = CRC32.buf(chunk, crc32Value);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Finalize crypto hashes
|
|
||||||
for (const [hashType, hasher] of hashers) {
|
|
||||||
const hexHash = hasher.digest('hex');
|
|
||||||
const cid = formatAsCid(hashType, hexHash);
|
|
||||||
results.set(hashType, cid);
|
|
||||||
}
|
|
||||||
|
|
||||||
// Finalize CRC32
|
|
||||||
if (needsCrc32) {
|
|
||||||
const unsigned = crc32Value >>> 0;
|
|
||||||
const hexHash = unsigned.toString(16).padStart(8, '0');
|
|
||||||
const cid = formatAsCid('cid_crc32', hexHash);
|
|
||||||
results.set('cid_crc32', cid);
|
|
||||||
}
|
|
||||||
|
|
||||||
return results;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Compute all hashes for a file in a single pass
|
|
||||||
*/
|
|
||||||
async function computeHashes(filePath: string, hashTypes: HashAlgorithm[]): Promise<Map<HashAlgorithm, string>> {
|
|
||||||
if (webdavClient) {
|
|
||||||
return computeHashesFromWebDAV(webdavClient, filePath, hashTypes);
|
|
||||||
}
|
|
||||||
return computeHashesFromFilesystem(filePath, hashTypes);
|
|
||||||
}
|
|
||||||
|
|
||||||
export const manifest: PluginManifest = {
|
|
||||||
id: 'full-hash',
|
|
||||||
name: 'Full File Hash',
|
|
||||||
version: '1.0.0',
|
|
||||||
description: 'Computes multiple hashes (SHA-256, SHA-1, MD5, CRC32, SHA3) with CSV caching',
|
|
||||||
author: 'MetaMesh',
|
|
||||||
dependencies: [],
|
|
||||||
priority: 100,
|
|
||||||
color: '#9C27B0',
|
|
||||||
defaultQueue: 'background',
|
|
||||||
timeout: 300000, // 5 minutes for large files
|
|
||||||
schema: {
|
|
||||||
'cid_sha2-256': {
|
|
||||||
label: 'SHA-256 Hash',
|
|
||||||
type: 'cid',
|
|
||||||
readonly: true,
|
|
||||||
hint: 'SHA-256 hash in CID format',
|
|
||||||
},
|
|
||||||
'cid_sha1': {
|
|
||||||
label: 'SHA-1 Hash',
|
|
||||||
type: 'cid',
|
|
||||||
readonly: true,
|
|
||||||
hint: 'SHA-1 hash in CID format',
|
|
||||||
},
|
|
||||||
'cid_md5': {
|
|
||||||
label: 'MD5 Hash',
|
|
||||||
type: 'cid',
|
|
||||||
readonly: true,
|
|
||||||
hint: 'MD5 hash in CID format',
|
|
||||||
},
|
|
||||||
'cid_crc32': {
|
|
||||||
label: 'CRC32 Checksum',
|
|
||||||
type: 'cid',
|
|
||||||
readonly: true,
|
|
||||||
hint: 'CRC32 checksum in CID format',
|
|
||||||
},
|
|
||||||
'cid_sha3-256': {
|
|
||||||
label: 'SHA3-256 Hash',
|
|
||||||
type: 'cid',
|
|
||||||
readonly: true,
|
|
||||||
hint: 'SHA3-256 hash in CID format',
|
|
||||||
},
|
|
||||||
'cid_sha3-384': {
|
|
||||||
label: 'SHA3-384 Hash',
|
|
||||||
type: 'cid',
|
|
||||||
readonly: true,
|
|
||||||
hint: 'SHA3-384 hash in CID format',
|
|
||||||
},
|
|
||||||
},
|
|
||||||
config: {
|
|
||||||
enabledHashes: {
|
|
||||||
type: 'string',
|
|
||||||
label: 'Enabled hash algorithms (comma-separated)',
|
|
||||||
default: 'cid_sha2-256,cid_sha1,cid_md5,cid_crc32,cid_sha3-256,cid_sha3-384',
|
|
||||||
required: false,
|
|
||||||
},
|
|
||||||
},
|
|
||||||
};
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Configure which hashes to compute
|
|
||||||
*/
|
|
||||||
export function configure(config: Record<string, unknown>): void {
|
|
||||||
if (config.enabledHashes && typeof config.enabledHashes === 'string') {
|
|
||||||
const requested = config.enabledHashes.split(',').map(s => s.trim()) as HashAlgorithm[];
|
|
||||||
enabledHashes = requested.filter(h => h in HASH_ALGORITHMS);
|
|
||||||
console.log(`[full-hash] Enabled hashes: ${enabledHashes.join(', ')}`);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
export async function process(
|
|
||||||
request: ProcessRequest,
|
|
||||||
sendCallback: (payload: CallbackPayload) => Promise<void>
|
|
||||||
): Promise<void> {
|
|
||||||
const startTime = Date.now();
|
|
||||||
const metaCore = new MetaCoreClient(request.metaCoreUrl);
|
|
||||||
|
|
||||||
try {
|
|
||||||
const { cid, filePath, existingMeta } = request;
|
|
||||||
|
|
||||||
// Determine which hashes need to be computed
|
|
||||||
const hashesToCompute: HashAlgorithm[] = [];
|
|
||||||
const hashesFromCache: Map<HashAlgorithm, string> = new Map();
|
|
||||||
|
|
||||||
// Get file stats for cache lookup
|
|
||||||
let fileSize: number;
|
|
||||||
let mtime: string;
|
|
||||||
const filename = path.basename(filePath);
|
|
||||||
|
|
||||||
if (webdavClient) {
|
|
||||||
const stats = await webdavClient.stat(filePath);
|
|
||||||
fileSize = stats.size;
|
|
||||||
mtime = stats.mtime?.toISOString() || new Date().toISOString();
|
|
||||||
} else {
|
|
||||||
const stats = await stat(filePath);
|
|
||||||
fileSize = stats.size;
|
|
||||||
mtime = stats.mtime.toISOString();
|
|
||||||
}
|
|
||||||
|
|
||||||
// Load caches and check what's needed
|
|
||||||
for (const hashType of enabledHashes) {
|
|
||||||
// Skip if already in metadata
|
|
||||||
if (existingMeta?.[hashType]) {
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
|
|
||||||
// Load cache for this hash type
|
|
||||||
await loadHashCache(hashType);
|
|
||||||
|
|
||||||
// Check cache
|
|
||||||
const cachedHash = getCachedHash(hashType, filename, fileSize, mtime);
|
|
||||||
if (cachedHash) {
|
|
||||||
hashesFromCache.set(hashType, cachedHash);
|
|
||||||
} else {
|
|
||||||
hashesToCompute.push(hashType);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Skip if nothing to do
|
|
||||||
if (hashesToCompute.length === 0 && hashesFromCache.size === 0) {
|
|
||||||
console.log(`[full-hash] All hashes already computed for ${filePath}, skipping`);
|
|
||||||
await sendCallback({
|
|
||||||
taskId: request.taskId,
|
|
||||||
status: 'skipped',
|
|
||||||
duration: Date.now() - startTime,
|
|
||||||
reason: 'All hashes already computed',
|
|
||||||
});
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
// Compute missing hashes
|
|
||||||
let computedHashes = new Map<HashAlgorithm, string>();
|
|
||||||
if (hashesToCompute.length > 0) {
|
|
||||||
console.log(`[full-hash] Computing ${hashesToCompute.length} hashes for ${filePath}`);
|
|
||||||
computedHashes = await computeHashes(filePath, hashesToCompute);
|
|
||||||
|
|
||||||
// Add computed hashes to cache
|
|
||||||
for (const [hashType, hash] of computedHashes) {
|
|
||||||
addToCache(hashType, filename, fileSize, mtime, hash);
|
|
||||||
}
|
|
||||||
|
|
||||||
// Save updated caches
|
|
||||||
for (const hashType of hashesToCompute) {
|
|
||||||
await saveHashCache(hashType);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Merge cached and computed hashes
|
|
||||||
const allHashes = new Map([...hashesFromCache, ...computedHashes]);
|
|
||||||
|
|
||||||
// Store all hashes to meta-core
|
|
||||||
for (const [hashType, hash] of allHashes) {
|
|
||||||
await metaCore.setProperty(cid, hashType, hash);
|
|
||||||
}
|
|
||||||
|
|
||||||
const cacheHits = hashesFromCache.size;
|
|
||||||
const computed = computedHashes.size;
|
|
||||||
const mode = webdavClient ? 'WebDAV' : 'filesystem';
|
|
||||||
console.log(`[full-hash] Stored ${allHashes.size} hashes (${cacheHits} cached, ${computed} computed, ${mode})`);
|
|
||||||
|
|
||||||
await sendCallback({
|
|
||||||
taskId: request.taskId,
|
|
||||||
status: 'completed',
|
|
||||||
duration: Date.now() - startTime,
|
|
||||||
});
|
|
||||||
} catch (error) {
|
|
||||||
const duration = Date.now() - startTime;
|
|
||||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
|
||||||
console.error(`[full-hash] Error computing hashes for ${request.filePath}:`, errorMessage);
|
|
||||||
|
|
||||||
await sendCallback({
|
|
||||||
taskId: request.taskId,
|
|
||||||
status: 'failed',
|
|
||||||
duration,
|
|
||||||
error: errorMessage,
|
|
||||||
});
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,70 +0,0 @@
|
|||||||
/**
|
|
||||||
* MetaMesh Plugin Types
|
|
||||||
*/
|
|
||||||
|
|
||||||
export interface PluginManifest {
|
|
||||||
id: string;
|
|
||||||
name: string;
|
|
||||||
version: string;
|
|
||||||
description?: string;
|
|
||||||
author?: string;
|
|
||||||
dependencies?: string[];
|
|
||||||
schema?: Record<string, SchemaField>;
|
|
||||||
config?: Record<string, ConfigField>;
|
|
||||||
priority?: number;
|
|
||||||
color?: string;
|
|
||||||
defaultQueue?: 'fast' | 'background';
|
|
||||||
timeout?: number;
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface SchemaField {
|
|
||||||
label: string;
|
|
||||||
type?: 'string' | 'number' | 'boolean' | 'select' | 'text' | 'array' | 'json' | 'cid';
|
|
||||||
readonly?: boolean;
|
|
||||||
hint?: string;
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface ConfigField {
|
|
||||||
type: 'string' | 'number' | 'boolean' | 'select';
|
|
||||||
label?: string;
|
|
||||||
required?: boolean;
|
|
||||||
default?: unknown;
|
|
||||||
secret?: boolean;
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface HealthResponse {
|
|
||||||
status: 'healthy' | 'unhealthy';
|
|
||||||
ready: boolean;
|
|
||||||
version: string;
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface ProcessRequest {
|
|
||||||
taskId: string;
|
|
||||||
cid: string;
|
|
||||||
filePath: string;
|
|
||||||
callbackUrl: string;
|
|
||||||
metaCoreUrl: string;
|
|
||||||
existingMeta?: Record<string, string>;
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface ProcessResponse {
|
|
||||||
status: 'accepted' | 'rejected';
|
|
||||||
error?: string;
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface CallbackPayload {
|
|
||||||
taskId: string;
|
|
||||||
status: 'completed' | 'failed' | 'skipped';
|
|
||||||
duration: number;
|
|
||||||
error?: string;
|
|
||||||
reason?: string;
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface ConfigureRequest {
|
|
||||||
config: Record<string, unknown>;
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface ConfigureResponse {
|
|
||||||
status: 'ok' | 'error';
|
|
||||||
error?: string;
|
|
||||||
}
|
|
||||||
@@ -1,178 +0,0 @@
|
|||||||
/**
|
|
||||||
* WebDAV Client for Plugin File Access
|
|
||||||
*
|
|
||||||
* Provides HTTP-based file access to meta-sort's WebDAV endpoint.
|
|
||||||
* Replaces direct filesystem access for containerized plugins.
|
|
||||||
*/
|
|
||||||
|
|
||||||
export interface FileStats {
|
|
||||||
size: number;
|
|
||||||
mtime?: Date;
|
|
||||||
}
|
|
||||||
|
|
||||||
export class WebDAVClient {
|
|
||||||
private baseUrl: string;
|
|
||||||
|
|
||||||
constructor(baseUrl: string) {
|
|
||||||
// Remove trailing slash if present
|
|
||||||
this.baseUrl = baseUrl.replace(/\/$/, '');
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Convert absolute file path to WebDAV URL
|
|
||||||
* Strips /files prefix and appends to base URL
|
|
||||||
*/
|
|
||||||
toWebDAVUrl(filePath: string): string {
|
|
||||||
// filePath is like "/files/watch/movie.mkv"
|
|
||||||
// We need to convert to "http://meta-sort-dev/webdav/watch/movie.mkv"
|
|
||||||
let relativePath = filePath;
|
|
||||||
|
|
||||||
// Strip /files prefix if present
|
|
||||||
if (relativePath.startsWith('/files/')) {
|
|
||||||
relativePath = relativePath.substring(6); // Remove "/files"
|
|
||||||
} else if (relativePath.startsWith('/files')) {
|
|
||||||
relativePath = relativePath.substring(6);
|
|
||||||
}
|
|
||||||
|
|
||||||
// Ensure path starts with /
|
|
||||||
if (!relativePath.startsWith('/')) {
|
|
||||||
relativePath = '/' + relativePath;
|
|
||||||
}
|
|
||||||
|
|
||||||
return this.baseUrl + relativePath;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Get file stats (size, mtime) via HTTP HEAD request
|
|
||||||
*/
|
|
||||||
async stat(filePath: string): Promise<FileStats> {
|
|
||||||
const url = this.toWebDAVUrl(filePath);
|
|
||||||
|
|
||||||
const response = await fetch(url, { method: 'HEAD' });
|
|
||||||
|
|
||||||
if (!response.ok) {
|
|
||||||
throw new Error(`WebDAV HEAD failed for ${filePath}: ${response.status} ${response.statusText}`);
|
|
||||||
}
|
|
||||||
|
|
||||||
const contentLength = response.headers.get('content-length');
|
|
||||||
const lastModified = response.headers.get('last-modified');
|
|
||||||
|
|
||||||
return {
|
|
||||||
size: contentLength ? parseInt(contentLength, 10) : 0,
|
|
||||||
mtime: lastModified ? new Date(lastModified) : undefined,
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Stream file content for hashing (returns async iterator of chunks)
|
|
||||||
*/
|
|
||||||
async *streamFile(filePath: string): AsyncGenerator<Buffer> {
|
|
||||||
const url = this.toWebDAVUrl(filePath);
|
|
||||||
|
|
||||||
const response = await fetch(url, { method: 'GET' });
|
|
||||||
|
|
||||||
if (!response.ok) {
|
|
||||||
throw new Error(`WebDAV GET failed for ${filePath}: ${response.status} ${response.statusText}`);
|
|
||||||
}
|
|
||||||
|
|
||||||
if (!response.body) {
|
|
||||||
throw new Error(`WebDAV GET returned no body for ${filePath}`);
|
|
||||||
}
|
|
||||||
|
|
||||||
const reader = response.body.getReader();
|
|
||||||
try {
|
|
||||||
while (true) {
|
|
||||||
const { done, value } = await reader.read();
|
|
||||||
if (done) break;
|
|
||||||
yield Buffer.from(value);
|
|
||||||
}
|
|
||||||
} finally {
|
|
||||||
reader.releaseLock();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Read first N bytes of a file (for magic byte detection)
|
|
||||||
* Uses HTTP Range request
|
|
||||||
*/
|
|
||||||
async readBytes(filePath: string, start: number, end: number): Promise<Buffer> {
|
|
||||||
const url = this.toWebDAVUrl(filePath);
|
|
||||||
|
|
||||||
const response = await fetch(url, {
|
|
||||||
method: 'GET',
|
|
||||||
headers: {
|
|
||||||
'Range': `bytes=${start}-${end}`,
|
|
||||||
},
|
|
||||||
});
|
|
||||||
|
|
||||||
if (!response.ok && response.status !== 206) {
|
|
||||||
throw new Error(`WebDAV Range GET failed for ${filePath}: ${response.status} ${response.statusText}`);
|
|
||||||
}
|
|
||||||
|
|
||||||
const arrayBuffer = await response.arrayBuffer();
|
|
||||||
return Buffer.from(arrayBuffer);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Read entire file as buffer (for small files only)
|
|
||||||
*/
|
|
||||||
async readFile(filePath: string): Promise<Buffer> {
|
|
||||||
const url = this.toWebDAVUrl(filePath);
|
|
||||||
|
|
||||||
const response = await fetch(url, { method: 'GET' });
|
|
||||||
|
|
||||||
if (!response.ok) {
|
|
||||||
throw new Error(`WebDAV GET failed for ${filePath}: ${response.status} ${response.statusText}`);
|
|
||||||
}
|
|
||||||
|
|
||||||
const arrayBuffer = await response.arrayBuffer();
|
|
||||||
return Buffer.from(arrayBuffer);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Write file via HTTP PUT
|
|
||||||
*/
|
|
||||||
async writeFile(filePath: string, data: Buffer | string): Promise<void> {
|
|
||||||
const url = this.toWebDAVUrl(filePath);
|
|
||||||
|
|
||||||
const response = await fetch(url, {
|
|
||||||
method: 'PUT',
|
|
||||||
body: data,
|
|
||||||
headers: {
|
|
||||||
'Content-Type': 'application/octet-stream',
|
|
||||||
},
|
|
||||||
});
|
|
||||||
|
|
||||||
if (!response.ok) {
|
|
||||||
throw new Error(`WebDAV PUT failed for ${filePath}: ${response.status} ${response.statusText}`);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Check if file exists via HTTP HEAD
|
|
||||||
*/
|
|
||||||
async exists(filePath: string): Promise<boolean> {
|
|
||||||
const url = this.toWebDAVUrl(filePath);
|
|
||||||
|
|
||||||
try {
|
|
||||||
const response = await fetch(url, { method: 'HEAD' });
|
|
||||||
return response.ok;
|
|
||||||
} catch {
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Create a WebDAV client from environment variables
|
|
||||||
*/
|
|
||||||
export function createWebDAVClient(): WebDAVClient | null {
|
|
||||||
const webdavUrl = process.env.WEBDAV_URL;
|
|
||||||
|
|
||||||
if (!webdavUrl) {
|
|
||||||
console.warn('[webdav-client] WEBDAV_URL not set, WebDAV client unavailable');
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
|
|
||||||
return new WebDAVClient(webdavUrl);
|
|
||||||
}
|
|
||||||
+114
@@ -0,0 +1,114 @@
|
|||||||
|
use anyhow::{Context, Result};
|
||||||
|
use reqwest::Client;
|
||||||
|
|
||||||
|
/// WebDAV client for streaming file access
|
||||||
|
pub struct WebDavClient {
|
||||||
|
client: Client,
|
||||||
|
base_url: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl WebDavClient {
|
||||||
|
pub fn new(base_url: &str) -> Self {
|
||||||
|
Self {
|
||||||
|
client: Client::new(),
|
||||||
|
base_url: base_url.trim_end_matches('/').to_string(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Build full URL for a file path
|
||||||
|
/// The input path is a filesystem path like /files/watch/movie.mkv
|
||||||
|
/// The WebDAV server exposes /files as /webdav, so we strip /files prefix
|
||||||
|
fn file_url(&self, path: &str) -> String {
|
||||||
|
// Strip /files prefix since WebDAV mounts /files at the base URL
|
||||||
|
let clean_path = path
|
||||||
|
.trim_start_matches('/')
|
||||||
|
.strip_prefix("files/")
|
||||||
|
.unwrap_or(path.trim_start_matches('/'));
|
||||||
|
format!("{}/{}", self.base_url, clean_path)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Stream file contents via HTTP GET
|
||||||
|
pub async fn get_file(&self, path: &str) -> Result<WebDavReader> {
|
||||||
|
let url = self.file_url(path);
|
||||||
|
|
||||||
|
let response = self.client
|
||||||
|
.get(&url)
|
||||||
|
.send()
|
||||||
|
.await
|
||||||
|
.with_context(|| format!("Failed to GET {}", url))?;
|
||||||
|
|
||||||
|
if !response.status().is_success() {
|
||||||
|
anyhow::bail!("WebDAV GET failed: {} for {}", response.status(), url);
|
||||||
|
}
|
||||||
|
|
||||||
|
let content_length = response.content_length();
|
||||||
|
|
||||||
|
Ok(WebDavReader {
|
||||||
|
inner: response,
|
||||||
|
size: content_length,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get file size without downloading content (HEAD request)
|
||||||
|
pub async fn get_size(&self, path: &str) -> Result<u64> {
|
||||||
|
let url = self.file_url(path);
|
||||||
|
|
||||||
|
let response = self.client
|
||||||
|
.head(&url)
|
||||||
|
.send()
|
||||||
|
.await
|
||||||
|
.with_context(|| format!("Failed to HEAD {}", url))?;
|
||||||
|
|
||||||
|
if !response.status().is_success() {
|
||||||
|
anyhow::bail!("WebDAV HEAD failed: {} for {}", response.status(), url);
|
||||||
|
}
|
||||||
|
|
||||||
|
response
|
||||||
|
.content_length()
|
||||||
|
.ok_or_else(|| anyhow::anyhow!("No Content-Length header for {}", url))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Async reader wrapper for WebDAV response
|
||||||
|
pub struct WebDavReader {
|
||||||
|
inner: reqwest::Response,
|
||||||
|
size: Option<u64>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl WebDavReader {
|
||||||
|
pub fn size(&self) -> Option<u64> {
|
||||||
|
self.size
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Read all bytes (for smaller files or when you need the whole thing)
|
||||||
|
pub async fn bytes(self) -> Result<Vec<u8>> {
|
||||||
|
self.inner.bytes().await
|
||||||
|
.map(|b| b.to_vec())
|
||||||
|
.context("Failed to read response bytes")
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get the underlying response for streaming
|
||||||
|
pub fn into_response(self) -> reqwest::Response {
|
||||||
|
self.inner
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_file_url() {
|
||||||
|
let client = WebDavClient::new("http://localhost/webdav");
|
||||||
|
// /files prefix should be stripped
|
||||||
|
assert_eq!(client.file_url("/files/watch/test.mkv"), "http://localhost/webdav/watch/test.mkv");
|
||||||
|
assert_eq!(client.file_url("files/watch/test.mkv"), "http://localhost/webdav/watch/test.mkv");
|
||||||
|
|
||||||
|
// Trailing slash in base URL should be handled
|
||||||
|
let client2 = WebDavClient::new("http://localhost/webdav/");
|
||||||
|
assert_eq!(client2.file_url("/files/watch/test.mkv"), "http://localhost/webdav/watch/test.mkv");
|
||||||
|
|
||||||
|
// Paths without /files prefix should work too
|
||||||
|
assert_eq!(client.file_url("watch/test.mkv"), "http://localhost/webdav/watch/test.mkv");
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,18 +0,0 @@
|
|||||||
{
|
|
||||||
"compilerOptions": {
|
|
||||||
"target": "ES2022",
|
|
||||||
"module": "NodeNext",
|
|
||||||
"moduleResolution": "NodeNext",
|
|
||||||
"lib": ["ES2022"],
|
|
||||||
"outDir": "./dist",
|
|
||||||
"rootDir": "./src",
|
|
||||||
"strict": true,
|
|
||||||
"esModuleInterop": true,
|
|
||||||
"skipLibCheck": true,
|
|
||||||
"forceConsistentCasingInFileNames": true,
|
|
||||||
"declaration": true,
|
|
||||||
"sourceMap": true
|
|
||||||
},
|
|
||||||
"include": ["src/**/*"],
|
|
||||||
"exclude": ["node_modules", "dist"]
|
|
||||||
}
|
|
||||||
Reference in New Issue
Block a user