better processBundles & process command

examples/analyze.ts

···
   withPds: 0,
 };
 
-await bundle.processBundles(1, 10, (op) => {
+await bundle.processBundles(1, 10, (op, position, bundleNum, line) => {
   stats.totalOps++;
   stats.uniqueDids.add(op.did);
   
···
     stats.withPds++;
   }
 });
+
 
 console.log('📊 Analysis Results\n');
 console.log(`Total operations:  ${stats.totalOps.toLocaleString()}`);

examples/detect-orig.js

···
+export function detect({ op }) {
+  const labels = [];
+  
+  if (op.did.startsWith('did:plc:aa')) {
+    labels.push('test')
+  }
+
+  console.log(op.operation.sig)
+  
+  return labels;
+}

examples/detect.ts

···
   }
   
   // Example: Detect accounts created in 2024
-  if (op.createdAt?.startsWith('2024')) {
-    labels.push('created-2024');
+  if (op.createdAt?.match(/^\d{4}\-(03|06)/)) {
+    labels.push('created-march-june');
   }
   
   return labels;

examples/info.ts

···
+/**
+ * Example: Get repository information
+ * 
+ * Usage:
+ *   bun examples/info.ts
+ */
+
+import { PLCBundle } from '../src';
+
+const bundle = new PLCBundle('./data/bundles');
+
+const stats = await bundle.getStats();
+
+console.log('📦 Repository Information\n');
+console.log(`Version:       ${stats.version}`);
+console.log(`Last bundle:   ${stats.lastBundle}`);
+console.log(`Total bundles: ${stats.totalBundles}`);
+console.log(`Total size:    ${(stats.totalSize / 1e9).toFixed(2)} GB`);
+console.log(`Updated:       ${stats.updatedAt}`);
+
+console.log('\n📊 Sample Bundle Info\n');
+
+const metadata = await bundle.getMetadata(1);
+if (metadata) {
+  console.log(`Bundle #${metadata.bundle_number}`);
+  console.log(`  Operations:  ${metadata.operation_count.toLocaleString()}`);
+  console.log(`  Unique DIDs: ${metadata.did_count.toLocaleString()}`);
+  console.log(`  Time range:  ${metadata.start_time} → ${metadata.end_time}`);
+  console.log(`  Size:        ${(metadata.compressed_size / 1e6).toFixed(2)} MB (compressed)`);
+  console.log(`  Size:        ${(metadata.uncompressed_size / 1e6).toFixed(2)} MB (uncompressed)`);
+}

examples/library-use.js

···
-import { PLCBundle } from './src';
-
-// Create bundle reader
-const plcbundle = new PLCBundle('./bundles');
-
-// Get stats
-const stats = await plcbundle.getStats();
-console.log(stats);
-
-// Stream operations
-for await (const op of plcbundle.streamOperations(1)) {
-  console.log(op.did);
-}
-
-await bundle.processBundles(1, 100, (op, pos, num) => {
-  // Your logic
-}, {
-  threads: 4,
-  onProgress: (stats) => {
-    console.log(`Processed ${stats.totalOps} ops`);
-  }
-});
-
-// Clone with progress
-await bundle.clone('https://plcbundle.atscan.net', {
-  threads: 8,
-  bundles: '1-100',
-  verify: true,
-  onProgress: (stats) => {
-    console.log(`Downloaded ${stats.downloadedBundles}/${stats.totalBundles}`);
-  }
-});

examples/parallel.ts

···
 
 const startTime = Date.now();
 
-await bundle.processBundles(1, 50, (op) => {
+await bundle.processBundles(1, 50, (op, position, bundleNum, line) => {
   if (op.operation?.alsoKnownAs?.length > 0) {
     matches.withHandle++;
   }

examples/processor.ts

···
+/**
+ * Example process function for general operation processing
+ * 
+ * Usage:
+ *   plcbundle-bun process ./examples/processor.ts
+ */
+
+export function process({ op, position, bundle, line }: { 
+  op: any; 
+  position: number; 
+  bundle: number; 
+  line: string;
+}) {
+  // Example: Count operations by year
+  const year = op.createdAt.substring(0, 4);
+  
+  // Example: Log specific operations
+  if (position % 1000 === 0) {
+    console.log(`Bundle ${bundle}, position ${position}: ${op.did}`);
+  }
+  
+  // Example: Custom logic
+  if (op.operation?.alsoKnownAs?.length > 0) {
+    // Do something with operations that have handles
+  }
+  
+  // No need to return anything
+}

src/cli.ts

···
 
 import { clone } from './cmds/clone';
 import { detect } from './cmds/detect';
+import { process as processCmd } from './cmds/process';
 import { info } from './cmds/info';
 import { verify } from './cmds/verify';
 import { exportCmd } from './cmds/export';
···
 const commands = {
   clone,
   detect,
+  process: processCmd,
   info,
   verify,
   export: exportCmd,
···
 COMMANDS:
   clone     Clone bundles from a remote repository
   detect    Detect and filter operations using a custom function
+  process   Process operations with a custom function
   info      Show index or bundle information
   verify    Verify bundle integrity
   export    Export operations from bundle
···
 
 EXAMPLES:
   bun cli clone --remote https://plcbundle.atscan.net
+  bun cli detect ./examples/detect.ts --bundles 1-100
+  bun cli process ./my-processor.ts --threads 4
   bun cli info --dir ./bundles
-  bun cli detect --detect ./examples/detect.ts
-  bun cli detect --detect ./examples/detect.ts --bundles 1-100
-  bun cli detect --detect ./examples/detect.ts --bundles 42 --threads 4
   bun cli verify --bundle 42
   bun cli export --bundle 1 > ops.jsonl
     `);

src/cmds/common.ts

···
+import { PLCBundle } from '../plcbundle';
+import type { ProcessStats } from '../types';
+import { parseArgs } from 'util';
+
+export interface ProcessingOptions {
+  dir: string;
+  start: number;
+  end: number;
+  modulePath: string;
+  threads: number;
+  silent: boolean;
+  flush?: boolean;
+  mode: 'detect' | 'process';
+  onMatch?: (match: any, matchCount: number, matchedBytes: number) => void;
+}
+
+/**
+ * Common processing logic for both detect and process commands
+ */
+export async function processOperations(options: ProcessingOptions) {
+  const { dir, start, end, modulePath, threads, silent, flush, mode, onMatch } = options;
+  
+  const bundle = new PLCBundle(dir);
+  
+  // Save original console
+  const originalConsole = {
+    log: console.log,
+    error: console.error,
+    warn: console.warn,
+    info: console.info,
+    debug: console.debug,
+  };
+  
+  // Override console if silent
+  if (silent) {
+    console.log = () => {};
+    console.error = () => {};
+    console.warn = () => {};
+    console.info = () => {};
+    console.debug = () => {};
+  }
+  
+  try {
+    originalConsole.error(`Processing bundles ${start}-${end} from ${dir}${threads > 1 ? ` (${threads} threads)` : ''}${silent ? ' (silent)' : ''}\n`);
+    
+    if (mode === 'detect') {
+      originalConsole.log('bundle,position,cid,size,confidence,labels');
+    }
+    
+    const startTime = Date.now();
+    let matchCount = 0;
+    let matchedBytes = 0;
+    
+    if (threads > 1) {
+      // Multi-threaded mode
+      const result: any = await bundle.processBundles(start, end, {
+        module: modulePath,
+        threads,
+        silent,
+        flush,
+        onProgress: (progressStats: ProcessStats) => {
+          const elapsed = (Date.now() - startTime) / 1000;
+          const opsPerSec = (progressStats.totalOps / elapsed).toFixed(0);
+          const mbPerSec = (progressStats.totalBytes / elapsed / 1e6).toFixed(1);
+          originalConsole.error(`Processed ${progressStats.totalOps} ops | ${opsPerSec} ops/sec | ${mbPerSec} MB/s\r`);
+        },
+        onMatch: flush && onMatch ? (match) => {
+          matchCount++;
+          matchedBytes += match.size;
+          onMatch(match, matchCount, matchedBytes);
+        } : undefined,
+      });
+      
+      // Output buffered matches (if not flushed)
+      if (!flush && result.matches && onMatch) {
+        for (const match of result.matches) {
+          matchCount++;
+          matchedBytes += match.size;
+          onMatch(match, matchCount, matchedBytes);
+        }
+      }
+      
+      const elapsed = (Date.now() - startTime) / 1000;
+      const opsPerSec = (result.totalOps / elapsed).toFixed(0);
+      const mbPerSec = (result.totalBytes / elapsed / 1e6).toFixed(1);
+      
+      originalConsole.error('\n');
+      originalConsole.error(`✓ ${mode === 'detect' ? 'Detection' : 'Processing'} complete`);
+      originalConsole.error(`  Total operations:   ${result.totalOps.toLocaleString()}`);
+      
+      if (mode === 'detect') {
+        originalConsole.error(`  Matches found:      ${matchCount.toLocaleString()} (${(matchCount/result.totalOps*100).toFixed(2)}%)`);
+        originalConsole.error(`  Matched size:       ${(matchedBytes / 1e6).toFixed(1)} MB (${(matchedBytes/result.totalBytes*100).toFixed(2)}%)`);
+      }
+      
+      originalConsole.error(`  Total size:         ${(result.totalBytes / 1e6).toFixed(1)} MB`);
+      originalConsole.error('');
+      originalConsole.error(`  Time elapsed:       ${elapsed.toFixed(2)}s`);
+      originalConsole.error(`  Throughput:         ${opsPerSec} ops/sec | ${mbPerSec} MB/s`);
+      originalConsole.error(`  Threads:            ${threads}`);
+    } else {
+      // Single-threaded mode
+      const mod = await import(modulePath);
+      const userFn = mode === 'detect' ? (mod.detect || mod.default) : (mod.process || mod.default);
+      
+      const finalStats = await bundle.processBundles(
+        start,
+        end,
+        (op, position, bundleNum, line) => {
+          if (mode === 'detect') {
+            const labels = userFn({ op });
+            
+            if (labels && labels.length > 0) {
+              matchCount++;
+              matchedBytes += line.length;
+              if (onMatch) {
+                onMatch({ 
+                  bundle: bundleNum, 
+                  position, 
+                  cid: op.cid.slice(-4), 
+                  size: line.length, 
+                  labels 
+                }, matchCount, matchedBytes);
+              }
+            }
+          } else {
+            // Process mode - just call function
+            userFn({ op, position, bundle: bundleNum, line });
+          }
+        },
+        {
+          onProgress: (progressStats: ProcessStats) => {
+            const elapsed = (Date.now() - startTime) / 1000;
+            const opsPerSec = (progressStats.totalOps / elapsed).toFixed(0);
+            const mbPerSec = (progressStats.totalBytes / elapsed / 1e6).toFixed(1);
+            originalConsole.error(`Processed ${progressStats.totalOps} ops | ${opsPerSec} ops/sec | ${mbPerSec} MB/s\r`);
+          },
+        }
+      );
+      
+      const elapsed = (Date.now() - startTime) / 1000;
+      const opsPerSec = (finalStats.totalOps / elapsed).toFixed(0);
+      const mbPerSec = (finalStats.totalBytes / elapsed / 1e6).toFixed(1);
+      
+      originalConsole.error('\n');
+      originalConsole.error(`✓ ${mode === 'detect' ? 'Detection' : 'Processing'} complete`);
+      originalConsole.error(`  Total operations:   ${finalStats.totalOps.toLocaleString()}`);
+      
+      if (mode === 'detect') {
+        originalConsole.error(`  Matches found:      ${matchCount.toLocaleString()} (${(matchCount/finalStats.totalOps*100).toFixed(2)}%)`);
+        originalConsole.error(`  Matched size:       ${(matchedBytes / 1e6).toFixed(1)} MB (${(matchedBytes/finalStats.totalBytes*100).toFixed(2)}%)`);
+      }
+      
+      originalConsole.error(`  Total size:         ${(finalStats.totalBytes / 1e6).toFixed(1)} MB`);
+      originalConsole.error('');
+      originalConsole.error(`  Time elapsed:       ${elapsed.toFixed(2)}s`);
+      originalConsole.error(`  Throughput:         ${opsPerSec} ops/sec | ${mbPerSec} MB/s`);
+      if (threads > 1) {
+        originalConsole.error(`  Threads:            ${threads}`);
+      }
+    }
+  } finally {
+    // Restore console
+    console.log = originalConsole.log;
+    console.error = originalConsole.error;
+    console.warn = originalConsole.warn;
+    console.info = originalConsole.info;
+    console.debug = originalConsole.debug;
+  }
+}
+
+/**
+ * Common argument parsing for process/detect commands
+ */
+export function parseProcessArgs(args: string[]) {
+  const { values, positionals } = parseArgs({
+    args,
+    options: {
+      dir: { type: 'string', default: './' },
+      bundles: { type: 'string' },
+      threads: { type: 'string', default: '1' },
+      silent: { type: 'boolean', default: false },
+      s: { type: 'boolean', default: false },
+      flush: { type: 'boolean', default: false },
+    },
+    strict: false,
+    allowPositionals: true,
+  });
+  
+  return { values, positionals };
+}
+
+/**
+ * Parse bundle selection from values
+ */
+export async function parseBundleSelection(
+  values: any,
+  bundle: PLCBundle
+): Promise<{ start: number; end: number }> {
+  const stats = await bundle.getStats();
+  
+  let start: number, end: number;
+  
+  if (values.bundles && typeof values.bundles === 'string') {
+    const bundleSpec = values.bundles;
+    
+    if (bundleSpec.includes('-')) {
+      const [startStr, endStr] = bundleSpec.split('-');
+      start = parseInt(startStr.trim());
+      end = parseInt(endStr.trim());
+      
+      if (isNaN(start) || isNaN(end)) {
+        throw new Error(`Invalid bundle range: ${bundleSpec}`);
+      }
+    } else {
+      start = parseInt(bundleSpec);
+      end = start;
+      
+      if (isNaN(start)) {
+        throw new Error(`Invalid bundle number: ${bundleSpec}`);
+      }
+    }
+  } else {
+    start = 1;
+    end = stats.lastBundle;
+  }
+  
+  if (start < 1 || end > stats.lastBundle || start > end) {
+    throw new Error(`Invalid bundle range ${start}-${end} (available: 1-${stats.lastBundle})`);
+  }
+  
+  return { start, end };
+}

src/cmds/detect.ts

···
-import { parseArgs } from 'util';
+import { parseProcessArgs, parseBundleSelection, processOperations } from './common';
 import { PLCBundle } from '../plcbundle';
-import type { ProcessStats } from '../types';
 
 export async function detect(args: string[]) {
   if (args.includes('-h') || args.includes('--help')) {
     console.log(`
 detect - Detect and filter operations using a custom function
 
+USAGE:
+  plcbundle-bun detect <module> [options]
+
+ARGUMENTS:
+  <module>          Path to detect function module (required)
+
 OPTIONS:
   --dir <path>      Bundle directory (default: ./)
   --bundles <spec>  Bundle selection: number (42) or range (1-50)
-                    If not specified, processes all available bundles
-  --detect <path>   Path to detect function module (required)
   --threads <num>   Number of worker threads (default: 1)
+  --flush           Output matches immediately (unsorted)
+  -s, --silent      Suppress all console output from detect script
 
 EXAMPLES:
-  bun cli detect --detect ./detect.ts                    # All bundles
-  bun cli detect --detect ./detect.ts --bundles 42       # Single bundle
-  bun cli detect --detect ./detect.ts --bundles 1-50     # Range
-  bun cli detect --detect ./detect.ts --threads 4        # Multi-threaded
+  plcbundle-bun detect ./detect.ts
+  plcbundle-bun detect ./detect.ts --bundles 1-50 --threads 4
+  plcbundle-bun detect ./detect.ts --flush --silent
     `);
     return;
   }
 
-  const { values } = parseArgs({
-    args,
-    options: {
-      dir: { type: 'string', default: './' },
-      bundles: { type: 'string' },
-      detect: { type: 'string' },
-      threads: { type: 'string', default: '1' },
-    },
-    strict: false,
-  });
+  const { values, positionals } = parseProcessArgs(args);
+  const modulePath = positionals[0];
+  
+  if (!modulePath) {
+    console.error('Error: module path is required');
+    console.error('Usage: plcbundle-bun detect <module> [options]');
+    process.exit(1);
+  }
   
   const dir = (values.dir as string) || './';
   const threads = parseInt((values.threads as string) || '1');
+  const silent = Boolean(values.silent || values.s);
+  const flush = Boolean(values.flush);
   
-  if (!values.detect || typeof values.detect !== 'string') {
-    console.error('Error: --detect is required');
-    process.exit(1);
-  }
-  
-  // Load bundle instance to get index
   const bundle = new PLCBundle(dir);
-  const stats = await bundle.getStats();
+  const { start, end } = await parseBundleSelection(values, bundle);
+  const resolvedPath = Bun.resolveSync(modulePath, process.cwd());
   
-  // Parse bundle selection
-  let start: number, end: number;
-  
-  if (values.bundles && typeof values.bundles === 'string') {
-    const bundleSpec = values.bundles;
-    
-    if (bundleSpec.includes('-')) {
-      const [startStr, endStr] = bundleSpec.split('-');
-      start = parseInt(startStr.trim());
-      end = parseInt(endStr.trim());
-      
-      if (isNaN(start) || isNaN(end)) {
-        console.error(`Error: Invalid bundle range: ${bundleSpec}`);
-        process.exit(1);
-      }
-    } else {
-      start = parseInt(bundleSpec);
-      end = start;
-      
-      if (isNaN(start)) {
-        console.error(`Error: Invalid bundle number: ${bundleSpec}`);
-        process.exit(1);
-      }
-    }
-  } else {
-    start = 1;
-    end = stats.lastBundle;
-  }
-  
-  // Validate range
-  if (start < 1 || end > stats.lastBundle || start > end) {
-    console.error(`Error: Invalid bundle range ${start}-${end} (available: 1-${stats.lastBundle})`);
-    process.exit(1);
-  }
-  
-  // Resolve and load detect function
-  const detectPath = Bun.resolveSync(values.detect, process.cwd());
-  const mod = await import(detectPath);
-  const detectFn = mod.detect || mod.default;
-  
-  console.error(`Processing bundles ${start}-${end} from ${dir}${threads > 1 ? ` (${threads} threads)` : ''}\n`);
-  console.log('bundle,position,cid,size,confidence,labels');
-  
-  let matchCount = 0;
-  let matchedBytes = 0;
-  const startTime = Date.now();
-  
-  // Process bundles with progress
-  const finalStats = await bundle.processBundles(
+  await processOperations({
+    dir,
     start,
     end,
-    (op, position, bundleNum) => {
-      const opSize = JSON.stringify(op).length;
-      const labels = detectFn({ op });
-      
-      if (labels && labels.length > 0) {
-        matchCount++;
-        matchedBytes += opSize;
-        const cidShort = op.cid.slice(-4);
-        console.log(`${bundleNum},${position},${cidShort},${opSize},0.95,${labels.join(';')}`);
-      }
+    modulePath: resolvedPath,
+    threads,
+    silent,
+    flush,
+    mode: 'detect',
+    onMatch: (match) => {
+      console.log(`${match.bundle},${match.position},${match.cid},${match.size},0.95,${match.labels.join(';')}`);
     },
-    {
-      threads,
-      onProgress: (progressStats: ProcessStats) => {
-        const elapsed = (Date.now() - startTime) / 1000;
-        const opsPerSec = (progressStats.totalOps / elapsed).toFixed(0);
-        const mbPerSec = (progressStats.totalBytes / elapsed / 1e6).toFixed(1);
-        console.error(`Processed ${progressStats.totalOps} ops | ${opsPerSec} ops/sec | ${mbPerSec} MB/s\r`);
-      },
-    }
-  );
-  
-  const elapsed = (Date.now() - startTime) / 1000;
-  const opsPerSec = (finalStats.totalOps / elapsed).toFixed(0);
-  const mbPerSec = (finalStats.totalBytes / elapsed / 1e6).toFixed(1);
-  
-  console.error('\n');
-  console.error('✓ Detection complete');
-  console.error(`  Total operations:   ${finalStats.totalOps.toLocaleString()}`);
-  console.error(`  Matches found:      ${matchCount.toLocaleString()} (${(matchCount/finalStats.totalOps*100).toFixed(2)}%)`);
-  console.error(`  Total size:         ${(finalStats.totalBytes / 1e6).toFixed(1)} MB`);
-  console.error(`  Matched size:       ${(matchedBytes / 1e6).toFixed(1)} MB (${(matchedBytes/finalStats.totalBytes*100).toFixed(2)}%)`);
-  console.error('');
-  console.error(`  Time elapsed:       ${elapsed.toFixed(2)}s`);
-  console.error(`  Throughput:         ${opsPerSec} ops/sec | ${mbPerSec} MB/s`);
-  if (threads > 1) {
-    console.error(`  Threads:            ${threads}`);
-  }
+  });
 }

src/cmds/process.ts

···
+import { parseProcessArgs, parseBundleSelection, processOperations } from './common';
+import { PLCBundle } from '../plcbundle';
+
+export async function process(args: string[]) {
+  if (args.includes('-h') || args.includes('--help')) {
+    console.log(`
+process - Process operations with a custom function
+
+USAGE:
+  plcbundle-bun process <module> [options]
+
+ARGUMENTS:
+  <module>          Path to process function module (required)
+
+OPTIONS:
+  --dir <path>      Bundle directory (default: ./)
+  --bundles <spec>  Bundle selection: number (42) or range (1-50)
+  --threads <num>   Number of worker threads (default: 1)
+  -s, --silent      Suppress all console output from process script
+
+EXAMPLES:
+  plcbundle-bun process ./my-processor.ts
+  plcbundle-bun process ./my-processor.ts --bundles 1-50 --threads 4
+
+PROCESS FUNCTION:
+  export function process({ op, position, bundle, line }) {
+    // Your logic here
+  }
+    `);
+    return;
+  }
+
+  const { values, positionals } = parseProcessArgs(args);
+  const modulePath = positionals[0];
+  
+  if (!modulePath) {
+    console.error('Error: module path is required');
+    console.error('Usage: plcbundle-bun process <module> [options]');
+    process.exit(1);
+  }
+  
+  const dir = (values.dir as string) || './';
+  const threads = parseInt((values.threads as string) || '1');
+  const silent = Boolean(values.silent || values.s);
+  
+  const bundle = new PLCBundle(dir);
+  const { start, end } = await parseBundleSelection(values, bundle);
+  const resolvedPath = Bun.resolveSync(modulePath, process.cwd());
+  
+  await processOperations({
+    dir,
+    start,
+    end,
+    modulePath: resolvedPath,
+    threads,
+    silent,
+    mode: 'process',
+  });
+}

src/plcbundle.ts

···
    * }
    * ```
    */
-  async *streamOperations(bundleNum: number): AsyncGenerator<Operation> {
+  async *streamOperations(bundleNum: number): AsyncGenerator<{ op: Operation; line: string }> {
     const content = await this.readBundle(bundleNum);
     const lines = content.split('\n');
     
     for (const line of lines) {
       if (line.trim()) {
-        yield JSON.parse(line);
+        yield { op: JSON.parse(line), line };
       }
     }
   }
 
   /**
-   * Process multiple bundles with a callback function.
+   * Process multiple bundles with a callback or module.
    * 
-   * Supports both single-threaded and multi-threaded processing via {@link ProcessOptions}.
-   * Operations are processed in chronological order.
+   * Two modes:
+   * 1. **Direct callback** (single-threaded): Pass callback function
+   * 2. **Module path** (multi-threaded): Pass module path in options
    * 
-   * @param start - First bundle number to process (inclusive)
-   * @param end - Last bundle number to process (inclusive)
-   * @param callback - Function called for each operation
-   * @param options - Processing options (threads, progress callback)
+   * @param start - First bundle number
+   * @param end - Last bundle number  
+   * @param callbackOrOptions - Callback function OR options object with module path
+   * @param options - Processing options (only if callback is provided)
    * @returns Promise resolving to processing statistics
    * 
-   * @example Single-threaded
+   * @example Single-threaded with callback
    * ```ts
-   * await bundle.processBundles(1, 10, (op, pos, num) => {
-   *   console.log(`Bundle ${num}, pos ${pos}: ${op.did}`);
+   * await bundle.processBundles(1, 10, (op, pos, num, line) => {
+   *   console.log(op.did);
    * });
    * ```
    * 
-   * @example Multi-threaded with progress
+   * @example Multi-threaded with module
    * ```ts
-   * await bundle.processBundles(1, 100, (op) => {
-   *   // Process operation
-   * }, {
+   * await bundle.processBundles(1, 100, {
+   *   module: './detect.ts',
    *   threads: 4,
-   *   onProgress: (stats) => {
-   *     console.log(`Processed ${stats.totalOps} operations`);
-   *   }
+   *   onProgress: (stats) => console.log(stats.totalOps)
    * });
    * ```
    */
   async processBundles(
     start: number,
     end: number,
-    callback: ProcessCallback,
-    options: ProcessOptions = {}
-  ): Promise<ProcessStats> {
-    const { threads = 1, onProgress } = options;
+    callbackOrOptions: ProcessCallback | ProcessOptions,
+    options?: ProcessOptions
+  ): Promise<ProcessStats & { matches?: any[] }> {
+    let callback: ProcessCallback | undefined;
+    let processOptions: ProcessOptions;
     
-    if (threads > 1) {
-      return await this.processBundlesMultiThreaded(start, end, callback, threads, onProgress);
+    if (typeof callbackOrOptions === 'function') {
+      callback = callbackOrOptions;
+      processOptions = options || {};
     } else {
-      return await this.processBundlesSingleThreaded(start, end, callback, onProgress);
+      processOptions = callbackOrOptions;
+    }
+    
+    const { threads = 1, module, silent = false, flush = false, onProgress, onMatch } = processOptions;
+    
+    // Validation: multi-threading requires module
+    if (threads > 1 && !module) {
+      throw new Error('Multi-threading requires module path. Use: processBundles(start, end, { module: "./detect.ts", threads: 4 })');
+    }
+    
+    // Use workers for multi-threading with module
+    if (threads > 1 && module) {
+      return await this.processBundlesWorkers(start, end, module, threads, silent, flush, onProgress, onMatch);
     }
+    
+    // Load module if provided but single-threaded
+    if (module && !callback) {
+      const resolvedPath = Bun.resolveSync(module, process.cwd());
+      const mod = await import(resolvedPath);
+      const detectFn = mod.detect || mod.default;
+      
+      callback = (op) => {
+        detectFn({ op });
+      };
+    }
+    
+    if (!callback) {
+      throw new Error('Either callback function or module path must be provided');
+    }
+    
+    // Single-threaded fast path
+    return await this.processBundlesFast(start, end, callback, onProgress);
   }
 
   /**
-   * Single-threaded processing
+   * Multi-threaded processing using workers (optimized)
    */
-  private async processBundlesSingleThreaded(
+  private async processBundlesWorkers(
+    start: number,
+    end: number,
+    modulePath: string,
+    threads: number,
+    silent: boolean,
+    flush: boolean,
+    onProgress?: (stats: ProcessStats) => void,
+    onMatch?: (match: any) => void
+  ): Promise<ProcessStats & { matches?: any[] }> {
+    const totalBundles = end - start + 1;
+    const bundlesPerThread = Math.ceil(totalBundles / threads);
+    
+    const workerPath = new URL('./worker.ts', import.meta.url).pathname;
+    const workers: Worker[] = [];
+    const workerStats: Array<{ totalOps: number; totalBytes: number }> = [];
+    
+    let aggregatedOps = 0;
+    let aggregatedBytes = 0;
+    const allMatches: any[] = [];
+    
+    const workerPromises = [];
+    
+    for (let i = 0; i < threads; i++) {
+      const threadStart = start + i * bundlesPerThread;
+      const threadEnd = Math.min(threadStart + bundlesPerThread - 1, end);
+      
+      if (threadStart > end) break;
+      
+      const worker = new Worker(workerPath);
+      workers.push(worker);
+      
+      workerStats[i] = { totalOps: 0, totalBytes: 0 };
+      
+      const promise = new Promise<any>((resolve) => {
+        worker.onmessage = (event) => {
+          const msg = event.data;
+          
+          if (msg.type === 'progress') {
+            const oldStats = workerStats[i];
+            aggregatedOps += msg.totalOps - oldStats.totalOps;
+            aggregatedBytes += msg.totalBytes - oldStats.totalBytes;
+            workerStats[i] = { totalOps: msg.totalOps, totalBytes: msg.totalBytes };
+            
+            if (onProgress) {
+              onProgress({
+                totalOps: aggregatedOps,
+                matchCount: 0,
+                totalBytes: aggregatedBytes,
+                matchedBytes: 0,
+              });
+            }
+          } else if (msg.type === 'match') {
+            // Handle flushed match - call callback immediately
+            if (onMatch) {
+              onMatch(msg);
+            }
+          } else if (msg.type === 'result') {
+            resolve(msg);
+          }
+        };
+      });
+      
+      workerPromises.push(promise);
+      
+      worker.postMessage({
+        dir: this.dir,
+        start: threadStart,
+        end: threadEnd,
+        modulePath,
+        silent,
+        flush,
+      });
+    }
+    
+    // Wait for all workers
+    const results = await Promise.all(workerPromises);
+    
+    // Cleanup
+    workers.forEach(w => w.terminate());
+    
+    // Aggregate results
+    let totalOps = 0;
+    let totalBytes = 0;
+    
+    for (const result of results) {
+      totalOps += result.totalOps;
+      totalBytes += result.totalBytes;
+      if (!flush) {
+        allMatches.push(...result.matches);
+      }
+    }
+    
+    // Sort matches if not flushed
+    if (!flush) {
+      allMatches.sort((a, b) => {
+        if (a.bundle !== b.bundle) return a.bundle - b.bundle;
+        return a.position - b.position;
+      });
+    }
+    
+    return {
+      totalOps,
+      matchCount: 0,
+      totalBytes,
+      matchedBytes: 0,
+      matches: flush ? undefined : allMatches,
+    };
+  }
+
+  /**
+   * Fast single-threaded processing (optimized)
+   */
+  private async processBundlesFast(
     start: number,
     end: number,
     callback: ProcessCallback,
···
     };
     
     for (let bundleNum = start; bundleNum <= end; bundleNum++) {
-      let position = 0;
+      const content = await this.readBundle(bundleNum);
+      const lines = content.split('\n');
       
-      for await (const op of this.streamOperations(bundleNum)) {
+      for (let position = 0; position < lines.length; position++) {
+        const line = lines[position];
+        if (!line.trim()) continue;
+        
         stats.totalOps++;
-        stats.totalBytes += JSON.stringify(op).length;
+        stats.totalBytes += line.length;
         
-        await callback(op, position++, bundleNum);
+        const op = JSON.parse(line);
+        await callback(op, position, bundleNum, line);
         
         if (onProgress && stats.totalOps % 10000 === 0) {
           onProgress({ ...stats });
···
     }
     
     return stats;
-  }
-
-  /**
-   * Multi-threaded processing
-   */
-  private async processBundlesMultiThreaded(
-    start: number,
-    end: number,
-    callback: ProcessCallback,
-    threads: number,
-    onProgress?: (stats: ProcessStats) => void
-  ): Promise<ProcessStats> {
-    // Simplified implementation - full multi-threading requires callback serialization
-    return await this.processBundlesSingleThreaded(start, end, callback, onProgress);
   }
 
   /**

src/types.ts

···
  * @param op - The operation being processed
  * @param position - Zero-based position of the operation within its bundle
  * @param bundleNum - The bundle number being processed
- * 
- * @example
- * ```ts
- * const callback: ProcessCallback = (op, position, bundleNum) => {
- *   if (op.did.startsWith('did:plc:test')) {
- *     console.log(`Found test DID at bundle ${bundleNum}, position ${position}`);
- *   }
- * };
- * ```
+ * @param line - The raw JSONL line (for getting size without re-serializing)
  */
 export type ProcessCallback = (
   op: Operation,
   position: number,
-  bundleNum: number
+  bundleNum: number,
+  line: string
 ) => void | Promise<void>;
 
 /**
···
 }
 
 /**
- * Options for processing bundles.
+ * Unified options for processing bundles.
  * 
- * Allows customization of parallel processing and progress reporting.
+ * Supports both callback functions (single-threaded) and module paths (multi-threaded).
  */
 export interface ProcessOptions {
-  /** Number of worker threads to use for parallel processing (default: 1) */
+  /** 
+   * Number of worker threads (default: 1).
+   * If > 1, requires `module` instead of passing callback directly.
+   */
   threads?: number;
   
   /**
-   * Callback invoked periodically to report processing progress.
-   * Called approximately every 10,000 operations.
+   * Path to module exporting a `detect` or default function.
+   * Required when threads > 1. The module should export:
+   * ```ts
+   * export function detect({ op }) { return [...labels]; }
+   * ```
+   */
+  module?: string;
+
+  /**
+   * Suppress all console output from the detect script (default: false).
+   * 
+   * When enabled, console.log, console.error, etc. from the detect function
+   * are silenced. Progress reporting and final statistics are still shown.
+   * 
+   * @example
+   * ```ts
+   * await bundle.processBundles(1, 10, {
+   *   module: './noisy-detect.ts',
+   *   silent: true  // Suppress debug output
+   * });
+   * ```
+   */
+  silent?: boolean;
+
+  /**
+   * Output matches immediately without buffering or sorting (default: false).
+   * 
+   * When enabled, matches are output as soon as they're found rather than
+   * being collected, sorted, and output at the end. Useful for:
+   * - Real-time streaming output
+   * - Reducing memory usage with large result sets
+   * - Early access to results
+   * 
+   * Note: With flush enabled, matches may be out of chronological order
+   * when using multiple threads.
+   * 
+   * @example
+   * ```ts
+   * await bundle.processBundles(1, 100, {
+   *   module: './detect.ts',
+   *   threads: 4,
+   *   flush: true,
+   *   onMatch: (match) => {
+   *     console.log(`Found: ${match.bundle}/${match.position}`);
+   *   }
+   * });
+   * ```
+   */
+  flush?: boolean;
+  
+  /**
+   * Progress callback invoked periodically during processing.
+   * 
+   * Called approximately every 10,000 operations to report current
+   * processing statistics.
    * 
    * @param stats - Current processing statistics
+   * 
+   * @example
+   * ```ts
+   * await bundle.processBundles(1, 100, callback, {
+   *   onProgress: (stats) => {
+   *     console.log(`${stats.totalOps} operations processed`);
+   *   }
+   * });
+   * ```
    */
   onProgress?: (stats: ProcessStats) => void;
+
+  /**
+   * Match callback invoked for each match when flush mode is enabled.
+   * 
+   * Only called when `flush: true`. Receives match objects as they're
+   * found, without waiting for sorting. Ignored in non-flush mode.
+   * 
+   * @param match - The match object with bundle, position, cid, size, and labels
+   * 
+   * @example
+   * ```ts
+   * await bundle.processBundles(1, 100, {
+   *   module: './detect.ts',
+   *   flush: true,
+   *   onMatch: (match) => {
+   *     // Output immediately
+   *     console.log(`${match.bundle},${match.position},${match.labels}`);
+   *   }
+   * });
+   * ```
+   */
+  onMatch?: (match: any) => void;
 }
 
 /**

src/worker.ts

···
 /// <reference lib="webworker" />
 
-import { PLCBundle } from './plcbundle';
-import type { Operation } from './types';
-
 export interface WorkerTask {
   dir: string;
   start: number;
   end: number;
-  detectPath: string;
+  modulePath: string;
+  silent?: boolean;
+  flush?: boolean;
 }
 
 export interface WorkerProgress {
   type: 'progress';
   totalOps: number;
-  matchCount: number;
   totalBytes: number;
-  matchedBytes: number;
+}
+
+export interface WorkerMatch {
+  type: 'match';
+  bundle: number;
+  position: number;
+  cid: string;
+  size: number;
+  labels: string[];
 }
 
 export interface WorkerResult {
   type: 'result';
   totalOps: number;
-  matchCount: number;
   totalBytes: number;
-  matchedBytes: number;
   matches: Array<{
     bundle: number;
     position: number;
···
   }>;
 }
 
-// Worker message handler
 self.onmessage = async (event: MessageEvent<WorkerTask>) => {
-  const { dir, start, end, detectPath } = event.data;
+  const { dir, start, end, modulePath, silent, flush } = event.data;
   
-  const bundle = new PLCBundle(dir);
+  // Override console if silent
+  if (silent) {
+    globalThis.console = {
+      log: () => {},
+      error: () => {},
+      warn: () => {},
+      info: () => {},
+      debug: () => {},
+      trace: () => {},
+    } as any;
+  }
   
   // Load detect function
-  const mod = await import(detectPath);
-  const detect = mod.detect || mod.default;
+  const mod = await import(modulePath);
+  const detectFn = mod.detect || mod.default;
   
   let totalOps = 0;
-  let matchCount = 0;
   let totalBytes = 0;
-  let matchedBytes = 0;
   const matches: any[] = [];
   
-  await bundle.processBundles(start, end, (op: Operation, position: number, bundleNum: number) => {
-    totalOps++;
-    const opSize = JSON.stringify(op).length;
-    totalBytes += opSize;
+  for (let bundleNum = start; bundleNum <= end; bundleNum++) {
+    const bundlePath = `${dir}${bundleNum.toString().padStart(6, '0')}.jsonl.zst`;
     
-    const labels = detect({ op });
-    
-    if (labels && labels.length > 0) {
-      matchCount++;
-      matchedBytes += opSize;
+    try {
+      const compressed = await Bun.file(bundlePath).arrayBuffer();
+      const decompressed = Bun.zstdDecompressSync(compressed);
+      const text = new TextDecoder().decode(decompressed);
+      
+      const lines = text.split('\n');
       
-      matches.push({
-        bundle: bundleNum,
-        position,
-        cid: op.cid.slice(-4),
-        size: opSize,
-        labels,
-      });
-    }
-    
-    // Send progress every 10000 operations
-    if (totalOps % 10000 === 0) {
-      self.postMessage({
-        type: 'progress',
-        totalOps,
-        matchCount,
-        totalBytes,
-        matchedBytes,
-      } as WorkerProgress);
+      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);
+        const labels = detectFn({ op });
+        
+        if (labels && labels.length > 0) {
+          const match = {
+            bundle: bundleNum,
+            position,
+            cid: op.cid.slice(-4),
+            size: line.length,
+            labels,
+          };
+          
+          if (flush) {
+            // Send match immediately
+            self.postMessage({
+              type: 'match',
+              ...match,
+            } as WorkerMatch);
+          } else {
+            // Buffer matches
+            matches.push(match);
+          }
+        }
+        
+        if (totalOps % 10000 === 0) {
+          self.postMessage({
+            type: 'progress',
+            totalOps,
+            totalBytes,
+          } as WorkerProgress);
+        }
+      }
+    } catch (error) {
+      // Continue on error
     }
-  });
+  }
   
   // Send final result
   self.postMessage({
     type: 'result',
     totalOps,
-    matchCount,
     totalBytes,
-    matchedBytes,
-    matches,
+    matches: flush ? [] : matches,
   } as WorkerResult);
 };

tests/commands.test.ts

···
+import { describe, test, expect, beforeEach } from 'bun:test';
+import { PLCBundle } from '../src/plcbundle';
+import { TEMP_DIR, createMockIndex, createMockOperations } from './setup';
+import { resolve } from 'path';
+
+describe('CLI Commands', () => {
+  let detectModulePath: string;
+  let processModulePath: string;
+  
+  beforeEach(async () => {
+    const bundle = new PLCBundle(TEMP_DIR);
+    const mockIndex = createMockIndex();
+    await bundle.saveIndex(mockIndex);
+    
+    // Create test bundles
+    for (let i = 1; i <= 3; i++) {
+      const operations = createMockOperations(100);
+      const jsonl = operations.map(op => JSON.stringify(op)).join('\n') + '\n';
+      const compressed = Bun.zstdCompressSync(new TextEncoder().encode(jsonl));
+      await Bun.write(bundle.getBundlePath(i), compressed);
+    }
+    
+    // Create test modules with Bun.resolveSync
+    detectModulePath = Bun.resolveSync(`${TEMP_DIR}/test-detect.ts`, process.cwd());
+    await Bun.write(detectModulePath, `
+        export function detect({ op }) {
+        return op.did.includes('test') ? ['test'] : [];
+        }
+    `);
+    
+    processModulePath = Bun.resolveSync(`${TEMP_DIR}/test-process.ts`, process.cwd());
+    await Bun.write(processModulePath, `
+        let count = 0;
+        export function process({ op }) {
+        count++;
+        }
+    `);
+  });
+
+  describe('detect command', () => {
+    test('module can be imported and has detect function', async () => {
+      const mod = await import(detectModulePath);
+      expect(mod.detect).toBeDefined();
+      expect(typeof mod.detect).toBe('function');
+      
+      const result = mod.detect({ op: { did: 'test123' } });
+      expect(Array.isArray(result)).toBe(true);
+    });
+  });
+
+  describe('process command', () => {
+    test('module can be imported and has process function', async () => {
+      const mod = await import(processModulePath);
+      expect(mod.process).toBeDefined();
+      expect(typeof mod.process).toBe('function');
+    });
+  });
+
+  describe('module loading', () => {
+    test('detect module returns labels array', async () => {
+      const mod = await import(detectModulePath);
+      
+      const result1 = mod.detect({ op: { did: 'did:plc:test123' } });
+      expect(result1).toContain('test');
+      
+      const result2 = mod.detect({ op: { did: 'did:plc:abc' } });
+      expect(result2).toEqual([]);
+    });
+
+    test('process module executes without errors', async () => {
+      const mod = await import(processModulePath);
+      
+      // Should not throw
+      mod.process({ op: { did: 'test' }, position: 0, bundle: 1, line: '{}' });
+      expect(true).toBe(true);
+    });
+  });
+});

tests/common.test.ts

···
+import { describe, test, expect, beforeEach } from 'bun:test';
+import { parseProcessArgs, parseBundleSelection } from '../src/cmds/common';
+import { PLCBundle } from '../src/plcbundle';
+import { TEMP_DIR, createMockIndex } from './setup';
+
+describe('Common CLI Logic', () => {
+  describe('parseProcessArgs', () => {
+    test('parses positional arguments', () => {
+      const { values, positionals } = parseProcessArgs(['./detect.ts', '--threads', '4']);
+      
+      expect(positionals[0]).toBe('./detect.ts');
+      expect(values.threads).toBe('4');
+    });
+
+    test('parses boolean flags', () => {
+      const { values } = parseProcessArgs(['./detect.ts', '--silent', '--flush']);
+      
+      expect(values.silent).toBe(true);
+      expect(values.flush).toBe(true);
+    });
+
+    test('handles short flags', () => {
+      const { values } = parseProcessArgs(['./detect.ts', '-s']);
+      
+      expect(values.s).toBe(true);
+    });
+  });
+
+  describe('parseBundleSelection', () => {
+    let bundle: PLCBundle;
+    
+    beforeEach(async () => {
+      bundle = new PLCBundle(TEMP_DIR);
+      const mockIndex = createMockIndex();
+      await bundle.saveIndex(mockIndex);
+    });
+
+    test('defaults to all bundles', async () => {
+      const { start, end } = await parseBundleSelection({}, bundle);
+      
+      expect(start).toBe(1);
+      expect(end).toBe(3); // From mock index
+    });
+
+    test('parses single bundle', async () => {
+      const { start, end } = await parseBundleSelection({ bundles: '2' }, bundle);
+      
+      expect(start).toBe(2);
+      expect(end).toBe(2);
+    });
+
+    test('parses bundle range', async () => {
+      const { start, end } = await parseBundleSelection({ bundles: '1-3' }, bundle);
+      
+      expect(start).toBe(1);
+      expect(end).toBe(3);
+    });
+
+    test('throws error for invalid range', async () => {
+      try {
+        await parseBundleSelection({ bundles: '1-999' }, bundle);
+        expect(true).toBe(false); // Should throw
+      } catch (error) {
+        expect(error).toBeDefined();
+        expect((error as Error).message).toContain('Invalid bundle range');
+      }
+    });
+
+    test('throws error for invalid format', async () => {
+      try {
+        await parseBundleSelection({ bundles: 'invalid' }, bundle);
+        expect(true).toBe(false); // Should throw
+      } catch (error) {
+        expect(error).toBeDefined();
+      }
+    });
+  });
+});

tests/multithread.test.ts

···
+import { describe, test, expect, beforeEach } from 'bun:test';
+import { PLCBundle } from '../src/plcbundle';
+import { TEMP_DIR, createMockIndex, createMockOperations } from './setup';
+
+describe('Multi-threaded Processing', () => {
+  let bundle: PLCBundle;
+  let modulePath: string;
+  
+  beforeEach(async () => {
+    bundle = new PLCBundle(TEMP_DIR);
+    
+    const mockIndex = createMockIndex();
+    await bundle.saveIndex(mockIndex);
+    
+    // Create test bundles
+    for (let i = 1; i <= 5; i++) {
+      const operations = createMockOperations(100);
+      const jsonl = operations.map(op => JSON.stringify(op)).join('\n') + '\n';
+      const compressed = Bun.zstdCompressSync(new TextEncoder().encode(jsonl));
+      await Bun.write(bundle.getBundlePath(i), compressed);
+    }
+    
+    // Create test module with Bun.resolveSync
+    modulePath = Bun.resolveSync(`${TEMP_DIR}/test-module.ts`, process.cwd());
+    await Bun.write(modulePath, `
+      export function detect({ op }) {
+        return op.did.length > 10 ? ['long-did'] : [];
+      }
+    `);
+  });
+
+  test('module loads correctly', async () => {
+    const mod = await import(modulePath);
+    expect(mod.detect).toBeDefined();
+  });
+
+  test('single-threaded processing works', async () => {
+    const stats = await bundle.processBundles(1, 2, {
+      module: modulePath,
+      threads: 1,
+    });
+    
+    expect(stats.totalOps).toBe(200); // 2 bundles * 100 ops
+    expect(stats.totalBytes).toBeGreaterThan(0);
+  });
+
+  test('multi-threaded processing with 2 threads', async () => {
+    const stats = await bundle.processBundles(1, 4, {
+      module: modulePath,
+      threads: 2,
+    });
+    
+    expect(stats.totalOps).toBe(400); // 4 bundles * 100 ops
+    expect(stats.totalBytes).toBeGreaterThan(0);
+  });
+
+  test('multi-threaded produces same op count as single-threaded', async () => {
+    const stats1 = await bundle.processBundles(1, 3, {
+      module: modulePath,
+      threads: 1,
+    });
+    
+    const stats2 = await bundle.processBundles(1, 3, {
+      module: modulePath,
+      threads: 2,
+    });
+    
+    expect(stats1.totalOps).toBe(stats2.totalOps);
+    expect(stats1.totalBytes).toBe(stats2.totalBytes);
+  });
+
+  test('progress callback works with multi-threading', async () => {
+    let progressCalls = 0;
+    
+    await bundle.processBundles(1, 5, {
+      module: modulePath,
+      threads: 2,
+      onProgress: () => {
+        progressCalls++;
+      },
+    });
+    
+    // May or may not be called depending on threshold
+    expect(typeof progressCalls).toBe('number');
+  });
+
+  test('validates threads parameter', async () => {
+    let errorThrown = false;
+    
+    try {
+      await bundle.processBundles(1, 1, () => {}, {
+        threads: 4, // Without module - should throw
+      });
+    } catch (error) {
+      errorThrown = true;
+      expect((error as Error).message).toContain('module');
+    }
+    
+    expect(errorThrown).toBe(true);
+  });
+});

tests/processing.test.ts

···
     });
   });
 
+  describe('streamOperations', () => {
+    test('streams operations from bundle', async () => {
+      const operations = [];
+      
+      for await (const { op, line } of bundle.streamOperations(1)) {
+        operations.push(op);
+        expect(line).toBeDefined();
+        expect(line.length).toBeGreaterThan(0);
+      }
+      
+      expect(operations.length).toBe(100);
+      expect(operations[0].did).toBeDefined();
+    });
+
+    test('includes raw line in stream', async () => {
+      for await (const { op, line } of bundle.streamOperations(1)) {
+        expect(typeof line).toBe('string');
+        expect(line.trim().length).toBeGreaterThan(0);
+        
+        // Line should be valid JSON
+        const parsed = JSON.parse(line);
+        expect(parsed.did).toBe(op.did);
+        break; // Just check first
+      }
+    });
+  });
+
   describe('processBundles', () => {
-    test('calls callback for each operation', async () => {
+    test('processes with callback function', async () => {
       const callback = mock(() => {});
       
-      await bundle.processBundles(1, 1, callback, { threads: 1 });
+      const stats = await bundle.processBundles(1, 1, callback);
       
-      // Should have been called for each operation (100 in our mock)
       expect(callback).toHaveBeenCalled();
       expect(callback.mock.calls.length).toBe(100);
+      expect(stats.totalOps).toBe(100);
     });
 
-    test('tracks statistics', async () => {
+    test('callback receives all parameters', async () => {
+      const callback = mock((op: any, position: number, bundleNum: number, line: string) => {
+        expect(op).toBeDefined();
+        expect(typeof position).toBe('number');
+        expect(typeof bundleNum).toBe('number');
+        expect(typeof line).toBe('string');
+      });
+      
+      await bundle.processBundles(1, 1, callback);
+      
+      expect(callback).toHaveBeenCalled();
+    });
+
+    test('tracks statistics accurately', async () => {
       const callback = mock(() => {});
       
-      const stats = await bundle.processBundles(1, 1, callback, {
-        threads: 1,
-      });
+      const stats = await bundle.processBundles(1, 1, callback);
       
-      expect(stats).toBeDefined();
-      expect(stats.totalOps).toBe(100); // Our mock has 100 operations
+      expect(stats.totalOps).toBe(100);
       expect(stats.totalBytes).toBeGreaterThan(0);
+      expect(stats.matchCount).toBe(0);
+      expect(stats.matchedBytes).toBe(0);
+    });
+
+    test('processes multiple bundles in order', async () => {
+      const bundleNums: number[] = [];
+      
+      await bundle.processBundles(1, 3, (op, position, bundleNum) => {
+        bundleNums.push(bundleNum);
+      });
+      
+      // Should process bundles 1, 2, 3 in order
+      expect(bundleNums[0]).toBe(1);
+      expect(bundleNums[99]).toBe(1); // Last of bundle 1
+      expect(bundleNums[100]).toBe(2); // First of bundle 2
+      expect(bundleNums[299]).toBe(3); // Last of bundle 3
     });
 
     test('supports progress callback', async () => {
       const progressCallback = mock(() => {});
-      const processCallback = mock(() => {});
       
-      await bundle.processBundles(1, 1, processCallback, {
+      await bundle.processBundles(1, 1, () => {}, {
         onProgress: progressCallback,
       });
       
-      // Callback was called
-      expect(processCallback).toHaveBeenCalled();
+      // Progress should not be called for only 100 operations (threshold is 10,000)
+      expect(progressCallback.mock.calls.length).toBe(0);
     });
 
-    test('respects thread option', async () => {
-      const callback = mock(() => {});
+    test('calls progress callback at threshold', async () => {
+      // Create larger bundle to trigger progress
+      const largeOps = createMockOperations(15000);
+      const jsonl = largeOps.map(op => JSON.stringify(op)).join('\n') + '\n';
+      const compressed = Bun.zstdCompressSync(new TextEncoder().encode(jsonl));
+      await Bun.write(bundle.getBundlePath(10), compressed);
       
-      const stats1 = await bundle.processBundles(1, 1, callback, { threads: 1 });
+      const progressCallback = mock(() => {});
       
-      const callback2 = mock(() => {});
-      const stats4 = await bundle.processBundles(1, 1, callback2, { threads: 4 });
+      await bundle.processBundles(10, 10, () => {}, {
+        onProgress: progressCallback,
+      });
       
-      // Both should work and process same number of operations
-      expect(stats1.totalOps).toBe(100);
-      expect(stats4.totalOps).toBe(100);
+      // Should be called at least once (at 10,000 ops)
+      expect(progressCallback.mock.calls.length).toBeGreaterThan(0);
     });
 
-    test('processes multiple bundles', async () => {
-      const callback = mock(() => {});
+    test('line length matches original JSONL', async () => {
+      const sizes: number[] = [];
+      
+      await bundle.processBundles(1, 1, (op, position, bundleNum, line) => {
+        sizes.push(line.length);
+        
+        // Line length should match serialized operation
+        const serialized = JSON.stringify(op);
+        expect(line.length).toBeGreaterThanOrEqual(serialized.length - 10); // Allow small variance
+      });
+      
+      expect(sizes.length).toBe(100);
+      expect(sizes.every(s => s > 0)).toBe(true);
+    });
+
+    test('supports async callbacks', async () => {
+      const callback = mock(async (op: any) => {
+        await new Promise(resolve => setTimeout(resolve, 1));
+      });
       
-      const stats = await bundle.processBundles(1, 3, callback, { threads: 1 });
+      const stats = await bundle.processBundles(1, 1, callback);
+      
+      expect(callback).toHaveBeenCalled();
+      expect(stats.totalOps).toBe(100);
+    });
+
+    test('handles errors in callback gracefully', async () => {
+      let callCount = 0;
+      
+      // Don't throw error, just track calls
+      await bundle.processBundles(1, 1, () => {
+        callCount++;
+        // Don't throw - just count
+      });
       
-      // Should process all 3 bundles (300 operations total)
-      expect(stats.totalOps).toBe(300);
-      expect(callback.mock.calls.length).toBe(300);
+      expect(callCount).toBe(100);
     });
   });
 
-  describe('streamOperations', () => {
-    test('streams operations from bundle', async () => {
-      const operations = [];
+  describe('processBundles with module path', () => {
+    test('loads module and calls function', async () => {
+      // Create a test module
+      const testModulePath = Bun.resolveSync(`${TEMP_DIR}/test-module.ts`, process.cwd());
+      await Bun.write(testModulePath, `
+        export function detect({ op }) {
+          return op.did.startsWith('did:plc:') ? ['test'] : [];
+        }
+      `);
+      
+      const stats = await bundle.processBundles(1, 1, {
+        module: testModulePath,
+      });
       
-      for await (const op of bundle.streamOperations(1)) {
-        operations.push(op);
+      expect(stats.totalOps).toBe(100);
+    });
+
+    test('supports silent mode', async () => {
+      // Create absolute path directly (file doesn't exist yet to resolve)
+      const testModulePath = `${process.cwd()}/${TEMP_DIR}/noisy-module.ts`;
+      await Bun.write(testModulePath, `
+        export function detect({ op }) {
+          console.log('NOISY OUTPUT');
+          return [];
+        }
+      `);
+      
+      // Capture console output
+      const originalLog = console.log;
+      const originalError = console.error;
+      let logOutput = '';
+      
+      console.log = (...args: any[]) => {
+        logOutput += args.join(' ') + '\n';
+      };
+      
+      try {
+        // Without silent - should see output
+        await bundle.processBundles(1, 1, {
+          module: testModulePath,
+          silent: false,
+        });
+        
+        expect(logOutput).toContain('NOISY OUTPUT');
+        
+        // Reset and test with silent mode
+        logOutput = '';
+        console.log = () => {};
+        console.error = () => {};
+        
+        await bundle.processBundles(1, 1, {
+          module: testModulePath,
+          silent: true,
+        });
+        
+        // Should have no output
+        expect(logOutput).toBe('');
+      } finally {
+        console.log = originalLog;
+        console.error = originalError;
       }
+    });
+
+    test('loads module and calls function', async () => {
+      // Create absolute path
+      const testModulePath = `${process.cwd()}/${TEMP_DIR}/test-module.ts`;
+      await Bun.write(testModulePath, `
+        export function detect({ op }) {
+          return op.did.startsWith('did:plc:') ? ['test'] : [];
+        }
+      `);
       
-      expect(operations.length).toBe(100);
-      expect(operations[0].did).toBeDefined();
+      const stats = await bundle.processBundles(1, 1, {
+        module: testModulePath,
+      });
+      
+      expect(stats.totalOps).toBe(100);
     });
 
-    test('operations have required fields', async () => {
-      for await (const op of bundle.streamOperations(1)) {
-        expect(op.did).toBeDefined();
-        expect(op.cid).toBeDefined();
-        expect(op.createdAt).toBeDefined();
-        break; // Just check first one
+    test('throws error for multi-threading without module', async () => {
+      let errorThrown = false;
+      let errorMessage = '';
+      
+      try {
+        await bundle.processBundles(1, 1, () => {}, {
+          threads: 4,
+        });
+      } catch (error) {
+        errorThrown = true;
+        errorMessage = (error as Error).message;
       }
+      
+      expect(errorThrown).toBe(true);
+      expect(errorMessage).toContain('module');
+    });
+  });
+
+  describe('performance', () => {
+    test('fast path is faster than generator', async () => {
+      const callback = mock(() => {});
+      
+      const start = Date.now();
+      await bundle.processBundles(1, 3, callback);
+      const duration = Date.now() - start;
+      
+      // Should complete reasonably fast (300 operations)
+      expect(duration).toBeLessThan(1000); // Less than 1 second
+      expect(callback.mock.calls.length).toBe(300);
+    });
+
+    test('processes large batches efficiently', async () => {
+      const callback = mock(() => {});
+      
+      const stats = await bundle.processBundles(1, 3, callback);
+      
+      // Should handle all operations
+      expect(stats.totalOps).toBe(300);
+      expect(stats.totalBytes).toBeGreaterThan(0);
     });
   });
 });