flush-pds example

.gitignore

···
 node_modules
 test
 dist
-test-data
+test-data
+pds_list

examples/flush-pds.ts

···
+// flushes all pds endpoints
+
+const unique: Array<string> = []
+
+export function process({ op }: { op: any }) {
+
+  const endpoint = op.operation.services?.atproto_pds?.endpoint
+  if (!unique.includes(endpoint)) {
+    console.log(endpoint)
+    unique.push(endpoint)
+  }
+}

src/cli.ts

···
 
 import { clone } from './cmds/clone';
 import { detect } from './cmds/detect';
-import { process as processCmd } from './cmds/process';
+import { processCmd } from './cmds/process';
 import { info } from './cmds/info';
 import { verify } from './cmds/verify';
 import { exportCmd } from './cmds/export';

src/cmds/common.ts

···
   silent: boolean;
   flush?: boolean;
   mode: 'detect' | 'process';
+  noProgress?: boolean;
   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 { dir, start, end, modulePath, threads, silent, flush, mode, noProgress, onMatch } = options;
   
   const bundle = new PLCBundle(dir);
   
···
         threads,
         silent,
         flush,
-        onProgress: (progressStats: ProcessStats) => {
+        onProgress: noProgress ? undefined : (progressStats: ProcessStats) => { // Check noProgress
           const elapsed = (Date.now() - startTime) / 1000;
           const opsPerSec = (progressStats.totalOps / elapsed).toFixed(0);
           const mbPerSec = (progressStats.totalBytes / elapsed / 1e6).toFixed(1);
···
           }
         },
         {
-          onProgress: (progressStats: ProcessStats) => {
+          onProgress: noProgress ? undefined : (progressStats: ProcessStats) => { // Check noProgress
             const elapsed = (Date.now() - startTime) / 1000;
             const opsPerSec = (progressStats.totalOps / elapsed).toFixed(0);
             const mbPerSec = (progressStats.totalBytes / elapsed / 1e6).toFixed(1);
···
       silent: { type: 'boolean', default: false },
       s: { type: 'boolean', default: false },
       flush: { type: 'boolean', default: false },
+      'no-progress': { type: 'boolean', default: false }, // Add this
     },
     strict: false,
     allowPositionals: true,

src/cmds/detect.ts

···
   --threads <num>   Number of worker threads (default: 1)
   --flush           Output matches immediately (unsorted)
   -s, --silent      Suppress all console output from detect script
+  --no-progress     Disable progress output (default: false)
 
 EXAMPLES:
   plcbundle-bun detect ./detect.ts
   plcbundle-bun detect ./detect.ts --bundles 1-50 --threads 4
-  plcbundle-bun detect ./detect.ts --flush --silent
+  plcbundle-bun detect ./detect.ts --flush --silent --no-progress
     `);
     return;
   }
···
   const threads = parseInt((values.threads as string) || '1');
   const silent = Boolean(values.silent || values.s);
   const flush = Boolean(values.flush);
+  const noProgress = Boolean(values['no-progress']); // Add this
   
   const bundle = new PLCBundle(dir);
   const { start, end } = await parseBundleSelection(values, bundle);
···
     threads,
     silent,
     flush,
+    noProgress, // Pass it
     mode: 'detect',
     onMatch: (match) => {
       console.log(`${match.bundle},${match.position},${match.cid},${match.size},0.95,${match.labels.join(';')}`);

src/cmds/process.ts

···
+import { exit } from 'process';
 import { parseProcessArgs, parseBundleSelection, processOperations } from './common';
 import { PLCBundle } from '../plcbundle';
 
-export async function process(args: string[]) {
+export async function processCmd(args: string[]) {
   if (args.includes('-h') || args.includes('--help')) {
     console.log(`
 process - Process operations with a custom function
···
   --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
+  --no-progress     Disable progress output (default: false)
 
 EXAMPLES:
   plcbundle-bun process ./my-processor.ts
   plcbundle-bun process ./my-processor.ts --bundles 1-50 --threads 4
+  plcbundle-bun process ./my-processor.ts --no-progress
 
 PROCESS FUNCTION:
   export function process({ op, position, bundle, line }) {
···
   const dir = (values.dir as string) || './';
   const threads = parseInt((values.threads as string) || '1');
   const silent = Boolean(values.silent || values.s);
+  const noProgress = Boolean(values['no-progress']); // Add this
   
   const bundle = new PLCBundle(dir);
   const { start, end } = await parseBundleSelection(values, bundle);
···
     modulePath: resolvedPath,
     threads,
     silent,
+    noProgress, // Pass it
     mode: 'process',
   });
 }

src/plcbundle.ts

···
       throw new Error('Multi-threading requires module path. Use: processBundles(start, end, { module: "./detect.ts", threads: 4 })');
     }
     
+    // Determine mode based on what function is exported
+    let mode: 'detect' | 'process' = 'detect';
+    if (module) {
+      try {
+        const mod = await import(module);
+        // If module has 'process' function, use process mode
+        if (mod.process) {
+          mode = 'process';
+        } else if (mod.detect) {
+          mode = 'detect';
+        }
+      } catch (e) {
+        // Default to detect
+      }
+    }
+    
     // Use workers for multi-threading with module
     if (threads > 1 && module) {
-      return await this.processBundlesWorkers(start, end, module, threads, silent, flush, onProgress, onMatch);
+      return await this.processBundlesWorkers(
+        start, 
+        end, 
+        module, 
+        threads, 
+        silent, 
+        flush, 
+        mode, // Pass mode
+        onProgress, 
+        onMatch
+      );
     }
     
     // Load module if provided but single-threaded
     if (module && !callback) {
-      const resolvedPath = Bun.resolveSync(module, process.cwd());
+      const resolvedPath = module;
       const mod = await import(resolvedPath);
-      const detectFn = mod.detect || mod.default;
+      const userFn = mode === 'detect' ? (mod.detect || mod.default) : (mod.process || mod.default);
       
-      callback = (op) => {
-        detectFn({ op });
+      callback = (op, position, bundleNum, line) => {
+        if (mode === 'detect') {
+          userFn({ op });
+        } else {
+          userFn({ op, position, bundle: bundleNum, line });
+        }
       };
     }
     
···
     threads: number,
     silent: boolean,
     flush: boolean,
+    mode: 'detect' | 'process', // Add mode parameter
     onProgress?: (stats: ProcessStats) => void,
     onMatch?: (match: any) => void
   ): Promise<ProcessStats & { matches?: any[] }> {
···
         modulePath,
         silent,
         flush,
+        mode, // Pass mode to worker
       });
     }
     
···
     }
     
     // Sort matches if not flushed
-    if (!flush) {
+    if (!flush && mode === 'detect') {
       allMatches.sort((a, b) => {
         if (a.bundle !== b.bundle) return a.bundle - b.bundle;
         return a.position - b.position;
···
       matchCount: 0,
       totalBytes,
       matchedBytes: 0,
-      matches: flush ? undefined : allMatches,
+      matches: flush || mode === 'process' ? undefined : allMatches,
     };
   }
 

src/worker.ts

···
   modulePath: string;
   silent?: boolean;
   flush?: boolean;
+  mode?: 'detect' | 'process'; // Add mode parameter
 }
 
 export interface WorkerProgress {
···
 }
 
 self.onmessage = async (event: MessageEvent<WorkerTask>) => {
-  const { dir, start, end, modulePath, silent, flush } = event.data;
+  const { dir, start, end, modulePath, silent, flush, mode = 'detect' } = event.data;
   
   // Override console if silent
   if (silent) {
···
     } as any;
   }
   
-  // Load detect function
+  // Load the appropriate function based on mode
   const mod = await import(modulePath);
-  const detectFn = mod.detect || mod.default;
+  const userFn = mode === 'detect' 
+    ? (mod.detect || mod.default)
+    : (mod.process || mod.default);
   
   let totalOps = 0;
   let totalBytes = 0;
···
         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 (mode === 'detect') {
+          // Detection mode - look for labels
+          const labels = userFn({ op });
           
-          if (flush) {
-            // Send match immediately
-            self.postMessage({
-              type: 'match',
-              ...match,
-            } as WorkerMatch);
-          } else {
-            // Buffer matches
-            matches.push(match);
+          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);
+            }
           }
+        } else {
+          // Process mode - just call the function
+          userFn({ op, position, bundle: bundleNum, line });
         }
         
         if (totalOps % 10000 === 0) {
···
     totalBytes,
     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;
···
       await Bun.write(bundle.getBundlePath(i), compressed);
     }
     
-    // Create test modules with Bun.resolveSync
-    detectModulePath = Bun.resolveSync(`${TEMP_DIR}/test-detect.ts`, process.cwd());
+    // Create test modules - use absolute paths
+    detectModulePath = `${process.cwd()}/${TEMP_DIR}/test-detect.ts`;
     await Bun.write(detectModulePath, `
-        export function detect({ op }) {
+      export function detect({ op }) {
         return op.did.includes('test') ? ['test'] : [];
-        }
+      }
     `);
     
-    processModulePath = Bun.resolveSync(`${TEMP_DIR}/test-process.ts`, process.cwd());
+    processModulePath = `${process.cwd()}/${TEMP_DIR}/test-process.ts`;
     await Bun.write(processModulePath, `
-        let count = 0;
-        export function process({ op }) {
+      let count = 0;
+      export function process({ op }) {
         count++;
-        }
+      }
     `);
   });
 

tests/multithread.test.ts

···
       await Bun.write(bundle.getBundlePath(i), compressed);
     }
     
-    // Create test module with Bun.resolveSync
-    modulePath = Bun.resolveSync(`${TEMP_DIR}/test-module.ts`, process.cwd());
+    // Create test module - use absolute path
+    modulePath = `${process.cwd()}/${TEMP_DIR}/test-module.ts`;
     await Bun.write(modulePath, `
       export function detect({ op }) {
         return op.did.length > 10 ? ['long-did'] : [];

tests/processing.test.ts

···
 
   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());
+      // Create a test module with 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'] : [];
···
     });
 
     test('supports silent mode', async () => {
-      // Create absolute path directly (file doesn't exist yet to resolve)
+      // Create absolute path directly
       const testModulePath = `${process.cwd()}/${TEMP_DIR}/noisy-module.ts`;
       await Bun.write(testModulePath, `
         export function detect({ op }) {