plcbundle-bun / src / worker.ts

/// <reference lib="webworker" />

import { search as jmespathSearch } from '@jmespath-community/jmespath';

export interface WorkerTask {
  dir: string;
  start: number;
  end: number;
  modulePath?: string;
  expression?: string;
  useSimple?: boolean;
  silent?: boolean;
  flush?: boolean;
  mode?: 'detect' | 'process' | 'query';
}

export interface WorkerProgress {
  type: 'progress';
  currentBundle: number;
  totalOps: number;
  totalBytes: number;
  matchCount: number;
}

export interface WorkerMatchBatch {
  type: 'match-batch';
  matches: Array<{ result: any }>;
}

export interface WorkerResult {
  type: 'result';
  totalOps: number;
  totalBytes: number;
  matchCount: number;
  data?: any;
}

self.onmessage = async (event: MessageEvent<WorkerTask>) => {
  const { 
    dir, 
    start, 
    end, 
    modulePath, 
    expression,
    useSimple,
    silent, 
    flush, 
    mode = 'detect' 
  } = event.data;
  
  if (silent) {
    globalThis.console = {
      log: () => {},
      error: () => {},
      warn: () => {},
      info: () => {},
      debug: () => {},
      trace: () => {},
    } as any;
  }
  
  let userFn: any;
  let queryFn: ((op: any) => any) | null = null;
  
  if (mode === 'query') {
    // Query mode
    if (useSimple) {
      const compiled = compileSimplePath(expression!);
      
      // Ultra-fast path for single property
      if (compiled.segments.length === 1 && compiled.segments[0].type === 'property') {
        const prop = compiled.segments[0].value as string;
        queryFn = (op) => op[prop];
      } else {
        queryFn = (op) => querySimplePath(op, compiled);
      }
    } else {
      queryFn = (op) => jmespathSearch(op, expression!);
    }
  } else {
    // Detect or process mode
    const mod = await import(modulePath!);
    userFn = mode === 'detect' ? (mod.detect || mod.default) : (mod.process || mod.default);
  }
  
  let totalOps = 0;
  let totalBytes = 0;
  let matchCount = 0;
  const BATCH_SIZE = 1000;
  let matchBatch: any[] = [];
  
  const flushBatch = () => {
    if (matchBatch.length > 0) {
      self.postMessage({ type: 'match-batch', matches: matchBatch } as WorkerMatchBatch);
      matchBatch = [];
    }
  };
  
  for (let bundleNum = start; bundleNum <= end; bundleNum++) {
    const bundlePath = `${dir}${bundleNum.toString().padStart(6, '0')}.jsonl.zst`;
    
    try {
      const compressed = await Bun.file(bundlePath).arrayBuffer();
      const decompressed = Bun.zstdDecompressSync(compressed);
      const text = new TextDecoder().decode(decompressed);
      const lines = text.split('\n');
      
      for (let position = 0; position < lines.length; position++) {
        const line = lines[position];
        if (!line.trim()) continue;
        
        totalOps++;
        totalBytes += line.length;
        const op = JSON.parse(line);
        
        if (mode === 'query') {
          try {
            const result = queryFn!(op);
            
            if (result !== null && result !== undefined) {
              matchCount++;
              matchBatch.push({ result });
              if (matchBatch.length >= BATCH_SIZE) flushBatch();
            }
          } catch (error) {}
        } else if (mode === 'detect') {
          const labels = userFn({ op });
          if (labels && labels.length > 0) {
            matchCount++;
            matchBatch.push({ result: { bundle: bundleNum, position, cid: op.cid.slice(-4), labels } });
            if (matchBatch.length >= BATCH_SIZE) flushBatch();
          }
        } else {
          // process mode
          userFn({ op, position, bundle: bundleNum, line });
        }
        
        if (totalOps % 10000 === 0) {
          self.postMessage({ type: 'progress', currentBundle: bundleNum, totalOps, totalBytes, matchCount } as WorkerProgress);
        }
      }
      
      self.postMessage({ type: 'progress', currentBundle: bundleNum, totalOps, totalBytes, matchCount } as WorkerProgress);
    } catch (error) {}
  }
  
  flushBatch();

  let prepareResult: any;
  if (mode !== 'query' && modulePath) {
    const mod = await import(modulePath);
    if (typeof mod.prepare === 'function') {
      try {
        prepareResult = await mod.prepare();
      } catch (error) {}
    }
  }
  
  self.postMessage({ type: 'result', totalOps, totalBytes, matchCount, data: prepareResult || null } as WorkerResult);
};

interface SimplePath {
  segments: Array<{ type: 'property' | 'index'; value: string | number }>;
}

function compileSimplePath(expression: string): SimplePath {
  const segments: Array<{ type: 'property' | 'index'; value: string | number }> = [];
  let current = '';
  let i = 0;
  
  while (i < expression.length) {
    const char = expression[i];
    if (char === '.') {
      if (current) segments.push({ type: 'property', value: current });
      current = '';
      i++;
    } else if (char === '[') {
      if (current) segments.push({ type: 'property', value: current });
      current = '';
      i++;
      let index = '';
      while (i < expression.length && expression[i] !== ']') {
        index += expression[i];
        i++;
      }
      segments.push({ type: 'index', value: parseInt(index) });
      i++;
    } else {
      current += char;
      i++;
    }
  }
  if (current) segments.push({ type: 'property', value: current });
  return { segments };
}

function querySimplePath(obj: any, compiled: SimplePath): any {
  let current = obj;
  for (const segment of compiled.segments) {
    if (current == null) return null;
    if (segment.type === 'property') {
      current = current[segment.value];
    } else {
      if (Array.isArray(current)) {
        current = current[segment.value as number];
      } else {
        return null;
      }
    }
  }
  return current;
}