fix

plcbundle.ts

···
 #!/usr/bin/env node
 
 /**
- * plcbundle.ts - A reference implementation for fetching PLC directory
- * operations and creating verifiable, chained bundles according to the plcbundle V1 spec.
- *
- * This script fetches operations, validates their order, de-duplicates them,
- * and groups them into 10,000-operation bundles. Each bundle is compressed,
- * hashed, and cryptographically linked to the previous one, creating a verifiable
- * chain of data.
+ * plcbundle.ts - A compact, readable reference implementation for creating
+ * plcbundle V1 compliant archives. This script demonstrates all critical spec
+ * requirements, including hashing, serialization, ordering, and boundary handling.
  */
 
 import fs from 'fs/promises';
···
 // --- Configuration ---
 const BUNDLE_SIZE = 10000;
 const INDEX_FILE = 'plc_bundles.json';
-const DEFAULT_DIR = './plc_bundles';
+const DEFAULT_DIR = './plc_bundles_ts';
 const PLC_URL = 'https://plc.directory';
 
 // --- Types (as per spec) ---
 interface PLCOperation {
+  did: string;
   cid: string;
   createdAt: string;
+  operation: Record<string, any>;
+  nullified?: boolean | string;
   _raw: string; // Holds the original raw JSON string for reproducibility.
 }
 
···
  * Manages the state and process of fetching, validating, and creating PLC bundles.
  */
 class PlcBundleManager {
-  private bundleDir: string;
   private index!: Index;
   private mempool: PLCOperation[] = [];
-  // This set is used to de-duplicate operations. It tracks CIDs from the
-  // previous bundle's boundary and all CIDs from the current mempool.
+  // This set correctly de-duplicates operations, both from the previous bundle's
+  // boundary and within new batches, and is pruned to stay memory-efficient.
   private seenCIDs = new Set<string>();
 
-  constructor(bundleDir: string) {
-    this.bundleDir = bundleDir;
+  private constructor(private bundleDir: string) {}
+
+  /**
+   * Factory to create and asynchronously initialize a PlcBundleManager instance.
+   */
+  public static async create(bundleDir: string): Promise<PlcBundleManager> {
+    const manager = new PlcBundleManager(bundleDir);
+    await manager.init();
+    return manager;
   }
 
   /**
    * Initializes the manager by loading the index and seeding the `seenCIDs`
    * set with the CIDs from the last saved bundle's boundary.
    */
-  async init() {
+  private async init() {
     await fs.mkdir(this.bundleDir, { recursive: true });
     this.index = await this._loadIndex();
     console.log(`plcbundle Reference Implementation\nDirectory: ${this.bundleDir}\n`);
···
       console.log(`Resuming from bundle ${lastBundle.bundle_number + 1}. Last op time: ${lastBundle.end_time}`);
       try {
         // Pre-seed the de-duplication set with CIDs from the previous bundle's boundary.
-        // This is crucial for preventing duplicates between two adjacent bundles.
         const prevOps = await this._loadBundleOps(lastBundle.bundle_number);
-        this.seenCIDs = this._getBoundaryCIDs(prevOps);
+        this.seenCIDs = PlcBundleManager._getBoundaryCIDs(prevOps);
         console.log(`  Seeded de-duplication set with ${this.seenCIDs.size} boundary CIDs.`);
       } catch (e) {
         console.warn(`  Warning: Could not load previous bundle file. Boundary deduplication may be incomplete.`);
···
   }
 
   /**
-   * The main execution loop. It continuously fetches operations, validates and
-   * de-duplicates them, fills the mempool, and creates bundles when ready.
+   * The main execution loop. Fetches, validates, de-duplicates, and bundles operations.
    */
   async run() {
     let cursor = this.index.bundles.at(-1)?.end_time || null;
···
       try {
         console.log(`\nFetching operations from cursor: ${cursor || 'start'}...`);
         const fetchedOps = await this._fetchOperations(cursor);
-
         if (fetchedOps.length === 0) {
-          console.log('No more operations available from PLC directory.');
+          console.log('No more operations available.');
           break;
         }
-        
+
         // The core ingestion logic: de-duplicate and validate operations before adding to the mempool.
         this._processAndValidateOps(fetchedOps);
-        
-        // The cursor for the next fetch is always the timestamp of the last operation received in the current batch.
-        cursor = fetchedOps[fetchedOps.length - 1].createdAt;
+        cursor = fetchedOps.at(-1)!.createdAt;
 
-        // If the mempool is full enough, create bundles. This can run multiple times per fetch.
+        // Create bundles as long as the mempool is full.
         while (this.mempool.length >= BUNDLE_SIZE) {
           await this._createAndSaveBundle();
         }
 
-        await new Promise(resolve => setTimeout(resolve, 200)); // Be nice to the server.
+        await new Promise(resolve => setTimeout(resolve, 200)); // Be nice.
       } catch (err: any) {
         console.error(`\nError: ${err.message}`);
         if (err.response) console.error(`HTTP Status: ${err.response.status}`);
···
     const response = await axios.get<string>(`${PLC_URL}/export`, { params, responseType: 'text' });
     const lines = response.data.trimEnd().split('\n');
     if (lines.length === 1 && lines[0] === '') return [];
-    // Important: The `_raw` property is added here to preserve the original JSON string,
-    // ensuring byte-for-byte reproducibility as required by Spec 4.2.
     return lines.map(line => ({ ...JSON.parse(line), _raw: line }));
   }
 
···
    * chronological order before adding it to the mempool.
    */
   private _processAndValidateOps(ops: PLCOperation[]) {
-    // The timestamp to validate against is the last operation in the mempool, or if empty,
-    // the end time of the last bundle. This prevents chronological gaps.
-    let lastTimestamp = this.mempool.at(-1)?.createdAt || this.index.bundles.at(-1)?.end_time || '';
+    let lastTimestamp = this.mempool.at(-1)?.createdAt ?? this.index.bundles.at(-1)?.end_time ?? '';
     let newOpsCount = 0;
 
     for (const op of ops) {
-      // The `seenCIDs` set efficiently handles duplicates from the previous bundle's
-      // boundary as well as any duplicates within the current fetched batch.
+      // This check now correctly handles both boundary dupes and within-batch dupes.
       if (this.seenCIDs.has(op.cid)) {
         continue;
       }
 
-      // Spec 3: Validate that the stream is chronological. This is a critical sanity check.
+      // Spec 3: Validate that the stream is chronological.
       if (op.createdAt < lastTimestamp) {
         throw new Error(`Chronological validation failed: op ${op.cid} at ${op.createdAt} is older than last op at ${lastTimestamp}`);
       }
···
   }
 
   /**
-   * Takes 10,000 operations from the mempool, creates a bundle file, generates
-   * its metadata according to the spec, and updates the index.
+   * Creates a bundle and prunes the `seenCIDs` set to maintain memory efficiency.
    */
   private async _createAndSaveBundle() {
-    const currentBundleNumber = this.index.last_bundle + 1;
     const bundleOps = this.mempool.splice(0, BUNDLE_SIZE);
-
-    const parentHash = this.index.bundles.at(-1)?.hash || '';
-    const previousCursor = this.index.bundles.at(-1)?.end_time || '';
-
-    // The hashing and serialization process follows the spec exactly to ensure compatibility.
+    const parentHash = this.index.bundles.at(-1)?.hash ?? '';
+    
+    // Spec 4.2 & 6.3: Hashing and serialization must be exact.
     const jsonl = PlcBundleManager._serializeJSONL(bundleOps);
-    const uncompressedBuffer = Buffer.from(jsonl, 'utf8');
-    const contentHash = PlcBundleManager._sha256(uncompressedBuffer);
+    const contentHash = PlcBundleManager._sha256(Buffer.from(jsonl, 'utf8'));
     const chainHash = PlcBundleManager._calculateChainHash(parentHash, contentHash);
-    const compressedBuffer = Buffer.from(compress(uncompressedBuffer, 3));
+    const compressedBuffer = Buffer.from(compress(Buffer.from(jsonl, 'utf8'), 3));
     
+    const currentBundleNumber = this.index.last_bundle + 1;
     const filename = `${String(currentBundleNumber).padStart(6, '0')}.jsonl.zst`;
     await fs.writeFile(path.join(this.bundleDir, filename), compressedBuffer);
 
-    const dids = new Set(bundleOps.map(op => JSON.parse(op._raw).did));
-    const metadata: BundleMetadata = {
+    this.index.bundles.push({
       bundle_number: currentBundleNumber,
       start_time: bundleOps[0].createdAt,
-      end_time: bundleOps[bundleOps.length - 1].createdAt,
+      end_time: bundleOps.at(-1)!.createdAt,
       operation_count: bundleOps.length,
-      did_count: dids.size,
-      hash: chainHash,
-      content_hash: contentHash,
-      parent: parentHash,
+      did_count: new Set(bundleOps.map(op => op.did)).size,
+      hash: chainHash, content_hash: contentHash, parent: parentHash,
       compressed_hash: PlcBundleManager._sha256(compressedBuffer),
       compressed_size: compressedBuffer.length,
-      uncompressed_size: uncompressedBuffer.length,
-      cursor: previousCursor,
+      uncompressed_size: Buffer.from(jsonl, 'utf8').length,
+      cursor: this.index.bundles.at(-1)?.end_time ?? '',
       created_at: new Date().toISOString()
-    };
-
-    this.index.bundles.push(metadata);
+    });
     this.index.last_bundle = currentBundleNumber;
-    this.index.total_size_bytes += metadata.compressed_size;
+    this.index.total_size_bytes += compressedBuffer.length;
     
     // Prune the `seenCIDs` set to keep it memory-efficient. It only needs to hold CIDs
     // from the new boundary and the remaining mempool, not all CIDs ever seen.
-    const newBoundaryCIDs = this._getBoundaryCIDs(bundleOps);
+    const newBoundaryCIDs = PlcBundleManager._getBoundaryCIDs(bundleOps);
     const mempoolCIDs = new Set(this.mempool.map(op => op.cid));
     this.seenCIDs = new Set([...newBoundaryCIDs, ...mempoolCIDs]);
 
     await this._saveIndex();
     console.log(`\nCreating bundle ${filename}...`);
-    console.log(`  ✓ Saved. Hash: ${metadata.hash.substring(0, 16)}...`);
-    console.log(`    Set new boundary with ${newBoundaryCIDs.size} CIDs. Pruned de-duplication set to ${this.seenCIDs.size} CIDs.`);
+    console.log(`  ✓ Saved. Hash: ${chainHash.substring(0, 16)}...`);
+    console.log(`    Pruned de-duplication set to ${this.seenCIDs.size} CIDs.`);
   }
   
   private async _loadIndex(): Promise<Index> {
···
   
   private async _loadBundleOps(bundleNumber: number): Promise<PLCOperation[]> {
     const filename = `${String(bundleNumber).padStart(6, '0')}.jsonl.zst`;
-    const filepath = path.join(this.bundleDir, filename);
-    const compressed = await fs.readFile(filepath);
+    const compressed = await fs.readFile(path.join(this.bundleDir, filename));
     const decompressed = Buffer.from(decompress(compressed)).toString('utf8');
     return decompressed.trimEnd().split('\n').map(line => ({...JSON.parse(line), _raw: line}));
   }
+  
+  // ==========================================================================
+  // Static Utility Methods
+  // ==========================================================================
 
-  /** Returns CIDs from the last timestamp of a bundle, used for boundary de-duplication. */
-  private _getBoundaryCIDs(ops: PLCOperation[]): Set<string> {
+  private static _sha256 = (d: string | Buffer): string => crypto.createHash('sha256').update(d).digest('hex');
+  private static _serializeJSONL = (ops: PLCOperation[]): string => ops.map(op => op._raw + '\n').join('');
+  private static _calculateChainHash = (p: string, c: string): string => PlcBundleManager._sha256(p ? `${p}:${c}` : `plcbundle:genesis:${c}`);
+  private static _getBoundaryCIDs = (ops: PLCOperation[]): Set<string> => {
     if (!ops.length) return new Set();
     const lastTime = ops.at(-1)!.createdAt;
     const cids = new Set<string>();
-    for (let i = ops.length - 1; i >= 0 && ops[i].createdAt === lastTime; i--) {
-      cids.add(ops[i].cid);
-    }
+    for (let i = ops.length - 1; i >= 0 && ops[i].createdAt === lastTime; i--) cids.add(ops[i].cid);
     return cids;
-  }
-  
-  // --- Static Utilities ---
-  private static _sha256 = (data: string | Buffer): string => crypto.createHash('sha256').update(data).digest('hex');
-  private static _serializeJSONL = (ops: PLCOperation[]): string => ops.map(op => op._raw + '\n').join('');
-  private static _calculateChainHash = (parent: string, contentHash: string): string => {
-    return PlcBundleManager._sha256(parent ? `${parent}:${contentHash}` : `plcbundle:genesis:${contentHash}`);
   };
 }
 
 // --- Entry Point ---
 (async () => {
   const dir = process.argv[2] || DEFAULT_DIR;
-  const manager = new PlcBundleManager(dir);
-  await manager.init();
+  const manager = await PlcBundleManager.create(dir);
   await manager.run();
 })().catch(err => {
-  console.error('\nFATAL ERROR:', err.message);
-  console.error(err.stack);
+  console.error('\nFATAL ERROR:', err.message, err.stack);
   process.exit(1);
 });