From 4c3a0df4b414ec47a983d865638f7b809c14ef7d Mon Sep 17 00:00:00 2001 From: Pierre Date: Sun, 8 Feb 2026 17:01:06 +0100 Subject: [PATCH] file event improvement using midhash --- docker/Dockerfile | 1 - docker/entrypoint.sh | 1 - docker/supervisord.conf | 1 - internal/config/config.go | 21 +------ internal/watcher/dispatcher.go | 4 +- internal/watcher/midhash.go | 107 +++++++++++++++++++++++++++++++++ internal/watcher/types.go | 12 ++-- internal/watcher/watcher.go | 37 ++---------- internal/watchers/manager.go | 57 +++++++++--------- 9 files changed, 154 insertions(+), 87 deletions(-) create mode 100644 internal/watcher/midhash.go diff --git a/docker/Dockerfile b/docker/Dockerfile index b3a97b4..bb6aafc 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -125,7 +125,6 @@ ENV META_CORE_PATH=/meta-core \ META_CORE_HTTP_PORT=9000 \ META_CORE_HTTP_HOST=0.0.0.0 \ REDIS_PORT=6379 \ - WATCH_FOLDER_LIST=/files/ \ WATCH_INTERVAL_MS=1000 # Expose ports diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh index bbda7c7..6c97864 100644 --- a/docker/entrypoint.sh +++ b/docker/entrypoint.sh @@ -30,7 +30,6 @@ echo "[entrypoint] meta-core starting..." echo "[entrypoint] META_CORE_PATH=${META_CORE_PATH:-/meta-core}" echo "[entrypoint] FILES_PATH=${FILES_PATH:-/files}" echo "[entrypoint] SERVICE_NAME=${SERVICE_NAME:-meta-core}" -echo "[entrypoint] WATCH_FOLDER_LIST=${WATCH_FOLDER_LIST:-/files/}" # Execute the main command (supervisord) exec "$@" diff --git a/docker/supervisord.conf b/docker/supervisord.conf index 20bac61..c25ab8b 100644 --- a/docker/supervisord.conf +++ b/docker/supervisord.conf @@ -36,7 +36,6 @@ environment= META_CORE_HTTP_PORT="%(ENV_META_CORE_HTTP_PORT)s", META_CORE_HTTP_HOST="%(ENV_META_CORE_HTTP_HOST)s", REDIS_PORT="%(ENV_REDIS_PORT)s", - WATCH_FOLDER_LIST="%(ENV_WATCH_FOLDER_LIST)s", WATCH_INTERVAL_MS="%(ENV_WATCH_INTERVAL_MS)s", API_PORT="80", BASE_URL="%(ENV_BASE_URL)s" diff --git a/internal/config/config.go b/internal/config/config.go index 38c3338..9b7e30f 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -33,7 +33,6 @@ type Config struct { DeadServiceThresholdMS int // Dead service threshold in ms (default: 600000) // File watcher configuration - WatchFolderList []string // List of folders to watch for file changes WatchIntervalMS int // Polling interval for network mounts (default: 1000) DebounceMS int // File change debounce time (default: 30000) EnableFileWatcher bool // Enable file watcher (default: true) @@ -64,10 +63,6 @@ func Load() *Config { EnableFileWatcher: getEnvBool("ENABLE_FILE_WATCHER", true), } - // Parse watch folder list (comma-separated) - watchFolders := getEnv("WATCH_FOLDER_LIST", "/files/") - cfg.WatchFolderList = parseCommaSeparated(watchFolders) - // Set mounts directory cfg.MountsDir = cfg.MetaCorePath + "/mounts" @@ -133,17 +128,7 @@ func getEnvBool(key string, defaultValue bool) bool { return defaultValue } -func parseCommaSeparated(s string) []string { - if s == "" { - return []string{} - } - parts := strings.Split(s, ",") - result := make([]string, 0, len(parts)) - for _, part := range parts { - trimmed := strings.TrimSpace(part) - if trimmed != "" { - result = append(result, trimmed) - } - } - return result +// DefaultWatchPath returns the default watch folder path +func (c *Config) DefaultWatchPath() string { + return c.FilesPath + "/watch" } diff --git a/internal/watcher/dispatcher.go b/internal/watcher/dispatcher.go index 761e4f0..8858aa2 100644 --- a/internal/watcher/dispatcher.go +++ b/internal/watcher/dispatcher.go @@ -60,8 +60,8 @@ func (d *Dispatcher) dispatchToStream(event FileEvent) { if event.Size > 0 { fields["size"] = fmt.Sprintf("%d", event.Size) } - if event.PartialHash != "" { - fields["partialHash"] = event.PartialHash + if event.MidHash256 != "" { + fields["midhash256"] = event.MidHash256 } if event.OldPath != "" { fields["oldPath"] = event.OldPath diff --git a/internal/watcher/midhash.go b/internal/watcher/midhash.go new file mode 100644 index 0000000..8154055 --- /dev/null +++ b/internal/watcher/midhash.go @@ -0,0 +1,107 @@ +package watcher + +import ( + "crypto/sha256" + "encoding/base32" + "encoding/binary" + "io" + "os" + "strings" +) + +const ( + // MidHashSampleSize is the size of the sample to hash (1MB) + MidHashSampleSize = 1024 * 1024 // 1MB + + // MidHash256Code is the custom multicodec for midhash256 + MidHash256Code = 0x1000 +) + +// ComputeMidHash256 computes a midhash256 for a file. +// +// Algorithm: +// - For files <= 1MB: Hashes entire file content + size prefix +// - For files > 1MB: Hashes middle 1MB + size prefix +// +// The size prefix (8-byte big-endian) ensures that files with identical +// middle content but different sizes produce different hashes. +// +// Returns: CID v1 string in base32lower format (prefixed with "b") +func ComputeMidHash256(filePath string) (string, error) { + // Get file info + fileInfo, err := os.Stat(filePath) + if err != nil { + return "", err + } + fileSize := fileInfo.Size() + + // Open file + file, err := os.Open(filePath) + if err != nil { + return "", err + } + defer file.Close() + + // Create size buffer (8-byte big-endian) + sizeBuffer := make([]byte, 8) + binary.BigEndian.PutUint64(sizeBuffer, uint64(fileSize)) + + // Read sample data + var sampleData []byte + if fileSize <= MidHashSampleSize { + // Small file: read entire content + sampleData = make([]byte, fileSize) + _, err = io.ReadFull(file, sampleData) + if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF { + return "", err + } + } else { + // Large file: read middle 1MB + middleOffset := (fileSize - MidHashSampleSize) / 2 + _, err = file.Seek(middleOffset, io.SeekStart) + if err != nil { + return "", err + } + sampleData = make([]byte, MidHashSampleSize) + _, err = io.ReadFull(file, sampleData) + if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF { + return "", err + } + } + + // Compute SHA-256 hash of [size + sample] + hasher := sha256.New() + hasher.Write(sizeBuffer) + hasher.Write(sampleData) + hashBytes := hasher.Sum(nil) + + // Build CID v1 bytes: + // [version(0x01)] + [varint(codec 0x1000)] + [multihash] + // multihash = [varint(hashCode 0x1000)] + [varint(length 32)] + [hash bytes] + cidBytes := make([]byte, 0, 1+2+2+1+32) + + // CID version 1 + cidBytes = append(cidBytes, 0x01) + + // Content codec as varint (0x1000 = 4096) + // varint encoding: 0x1000 = 0b0001_0000_0000_0000 + // Split into 7-bit chunks: 0b00_0000_0000 (high) and 0b010_0000 (low with continuation) + // Result: [0x80, 0x20] for 0x1000 + cidBytes = append(cidBytes, 0x80, 0x20) + + // Multihash: hash function code as varint (0x1000) + cidBytes = append(cidBytes, 0x80, 0x20) + + // Multihash: length as varint (32 = 0x20) + cidBytes = append(cidBytes, 0x20) + + // Multihash: hash bytes + cidBytes = append(cidBytes, hashBytes...) + + // Encode as base32lower (RFC 4648 without padding) + encoded := base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(cidBytes) + encoded = strings.ToLower(encoded) + + // Add multibase prefix "b" for base32lower + return "b" + encoded, nil +} diff --git a/internal/watcher/types.go b/internal/watcher/types.go index 4a211d5..0a0820f 100644 --- a/internal/watcher/types.go +++ b/internal/watcher/types.go @@ -14,12 +14,12 @@ const ( // FileEvent represents a file system event type FileEvent struct { - Type FileEventType `json:"type"` - Path string `json:"path"` // Relative to FILES_PATH - Size int64 `json:"size,omitempty"` - Timestamp int64 `json:"timestamp"` - PartialHash string `json:"partialHash,omitempty"` // Hash of first 64KB - OldPath string `json:"oldPath,omitempty"` // For rename events + Type FileEventType `json:"type"` + Path string `json:"path"` // Relative to FILES_PATH + Size int64 `json:"size,omitempty"` + Timestamp int64 `json:"timestamp"` + MidHash256 string `json:"midhash256,omitempty"` // midhash256 CID (SHA-256 of middle 1MB + file size) + OldPath string `json:"oldPath,omitempty"` // For rename events } // PendingEvent tracks a file event that's being debounced diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go index b5620a0..084388c 100644 --- a/internal/watcher/watcher.go +++ b/internal/watcher/watcher.go @@ -1,9 +1,6 @@ package watcher import ( - "crypto/sha256" - "encoding/hex" - "io" "log" "os" "path/filepath" @@ -13,10 +10,6 @@ import ( "github.com/metazla/meta-core/internal/config" ) -const ( - // PartialHashSize is the number of bytes to hash for file identification - PartialHashSize = 64 * 1024 // 64KB -) // Watcher provides file scanning and event dispatching type Watcher struct { @@ -46,11 +39,11 @@ func NewWatcher(cfg *config.Config, dispatcher *Dispatcher) (*Watcher, error) { // handleDebouncedEvent processes a debounced event func (w *Watcher) handleDebouncedEvent(event FileEvent) { - // Compute partial hash for add/change events + // Compute midhash256 for add/change events if event.Type == EventTypeAdd || event.Type == EventTypeChange { fullPath := filepath.Join(w.filesPath, event.Path) - if hash, err := computePartialHash(fullPath); err == nil { - event.PartialHash = hash + if hash, err := ComputeMidHash256(fullPath); err == nil { + event.MidHash256 = hash } } @@ -102,9 +95,9 @@ func (w *Watcher) ScanDirectory(root string) int { Timestamp: NowMS(), } - // Compute partial hash - if hash, err := computePartialHash(path); err == nil { - event.PartialHash = hash + // Compute midhash256 + if hash, err := ComputeMidHash256(path); err == nil { + event.MidHash256 = hash } // Dispatch directly @@ -178,21 +171,3 @@ func (w *Watcher) GetFilesPath() string { return w.filesPath } -// computePartialHash computes SHA-256 hash of first 64KB of a file -func computePartialHash(path string) (string, error) { - file, err := os.Open(path) - if err != nil { - return "", err - } - defer file.Close() - - hasher := sha256.New() - buffer := make([]byte, PartialHashSize) - n, err := file.Read(buffer) - if err != nil && err != io.EOF { - return "", err - } - - hasher.Write(buffer[:n]) - return hex.EncodeToString(hasher.Sum(nil)), nil -} diff --git a/internal/watchers/manager.go b/internal/watchers/manager.go index 2f4d327..f2c205a 100644 --- a/internal/watchers/manager.go +++ b/internal/watchers/manager.go @@ -40,8 +40,8 @@ func NewManager(cfg *config.Config) (*Manager, error) { return nil, fmt.Errorf("failed to create watchers directory: %w", err) } - // Load or migrate - if err := m.loadOrMigrate(); err != nil { + // Load existing or create default + if err := m.loadOrCreateDefault(); err != nil { return nil, err } @@ -55,43 +55,46 @@ func (m *Manager) SetOnChanged(cb func()) { m.onChanged = cb } -// loadOrMigrate loads existing config or migrates from ENV -func (m *Manager) loadOrMigrate() error { +// loadOrCreateDefault loads existing config or creates default watcher +func (m *Manager) loadOrCreateDefault() error { // Try to load existing file if _, err := os.Stat(m.watchersFile); err == nil { return m.load() } - // File doesn't exist, check for ENV migration - if len(m.config.WatchFolderList) > 0 { - log.Printf("[Watchers] Migrating from WATCH_FOLDER_LIST env var") - return m.migrateFromEnv() - } - - // No config, start empty - log.Printf("[Watchers] No existing config, starting with empty watchers list") - return nil + // File doesn't exist, create default watcher for /files/watch + log.Printf("[Watchers] No existing config, creating default watcher for %s", m.config.DefaultWatchPath()) + return m.createDefaultWatcher() } -// migrateFromEnv creates watchers from WATCH_FOLDER_LIST env var -func (m *Manager) migrateFromEnv() error { +// createDefaultWatcher creates the default /files/watch watcher and folder +func (m *Manager) createDefaultWatcher() error { m.mu.Lock() defer m.mu.Unlock() - for _, path := range m.config.WatchFolderList { - id := uuid.New().String()[:8] - watcher := &WatcherConfig{ - ID: id, - Path: path, - IntervalMs: DefaultIntervalMs, - Enabled: true, - CreatedAt: NowMS(), - } - m.watchers[id] = watcher - log.Printf("[Watchers] Migrated watcher: %s -> %s", id, path) + defaultPath := m.config.DefaultWatchPath() + + // Create the watch folder if it doesn't exist + if err := os.MkdirAll(defaultPath, 0755); err != nil { + log.Printf("[Watchers] Warning: failed to create default watch folder %s: %v", defaultPath, err) + // Continue anyway - folder might be created later or mounted + } else { + log.Printf("[Watchers] Created default watch folder: %s", defaultPath) } - // Save migrated config + // Create default watcher + id := uuid.New().String()[:8] + watcher := &WatcherConfig{ + ID: id, + Path: defaultPath, + IntervalMs: DefaultIntervalMs, + Enabled: true, + CreatedAt: NowMS(), + } + m.watchers[id] = watcher + log.Printf("[Watchers] Created default watcher: %s -> %s", id, defaultPath) + + // Save config return m.saveUnsafe() }