update

internal/plc/bundle.go

···
 	"github.com/klauspost/compress/zstd"
 )
 
-// BUNDLE_SIZE is the number of operations per bundle
 const BUNDLE_SIZE = 10000
 
 type BundleManager struct {
···
 	decoder *zstd.Decoder
 	db      storage.Database
 }
+
+// ===== INITIALIZATION =====
 
 func NewBundleManager(dir string, enabled bool, db storage.Database) (*BundleManager, error) {
 	if !enabled {
···
 
 	return &BundleManager{
 		dir:     dir,
-		enabled: true,
+		enabled: enabled,
 		encoder: encoder,
 		decoder: decoder,
 		db:      db,
···
 	}
 }
 
-// GetBundleFilename returns filename for bundle number (6-digit decimal, JSONL format)
-func (bm *BundleManager) GetBundleFilename(bundleNumber int) string {
-	return fmt.Sprintf("%06d.jsonl.zst", bundleNumber)
+// ===== BUNDLE FILE ABSTRACTION =====
+
+type bundleFile struct {
+	path             string
+	operations       []PLCOperation
+	uncompressedHash string
+	compressedHash   string
 }
 
-// GetBundlePath returns full path for bundle number
-func (bm *BundleManager) GetBundlePath(bundleNumber int) string {
-	return filepath.Join(bm.dir, bm.GetBundleFilename(bundleNumber))
+func (bm *BundleManager) newBundleFile(bundleNum int) *bundleFile {
+	return &bundleFile{
+		path: filepath.Join(bm.dir, fmt.Sprintf("%06d.jsonl.zst", bundleNum)),
+	}
 }
 
-// BundleExists checks if bundle file exists locally
-func (bm *BundleManager) BundleExists(bundleNumber int) bool {
-	_, err := os.Stat(bm.GetBundlePath(bundleNumber))
+func (bf *bundleFile) exists() bool {
+	_, err := os.Stat(bf.path)
 	return err == nil
 }
 
-// LoadBundle returns exactly 1000 unique operations by fetching additional batches if needed
-func (bm *BundleManager) LoadBundle(ctx context.Context, bundleNumber int, plcClient *Client) ([]PLCOperation, bool, error) {
-	if !bm.enabled {
-		return nil, false, fmt.Errorf("bundle manager disabled")
+func (bm *BundleManager) load(bf *bundleFile) error {
+	compressed, err := os.ReadFile(bf.path)
+	if err != nil {
+		return fmt.Errorf("read failed: %w", err)
 	}
 
-	path := bm.GetBundlePath(bundleNumber)
+	decompressed, err := bm.decoder.DecodeAll(compressed, nil)
+	if err != nil {
+		return fmt.Errorf("decompress failed: %w", err)
+	}
 
-	// Try to load from local file first
-	if bm.BundleExists(bundleNumber) {
-		log.Verbose("→ Loading bundle %06d from local file", bundleNumber)
+	bf.operations = bm.parseJSONL(decompressed)
+	return nil
+}
 
-		// Check if bundle exists in database
-		dbBundle, dbErr := bm.db.GetBundleByNumber(ctx, bundleNumber)
-		bundleInDB := dbErr == nil && dbBundle != nil
+func (bm *BundleManager) save(bf *bundleFile) error {
+	jsonlData := bm.serializeJSONL(bf.operations)
+	bf.uncompressedHash = bm.hash(jsonlData)
 
-		if bundleInDB {
-			// Verify compressed file hash
-			if dbBundle.CompressedHash != "" {
-				valid, err := bm.verifyBundleHash(path, dbBundle.CompressedHash)
-				if err != nil {
-					log.Error("Warning: failed to verify compressed hash for bundle %06d: %v", bundleNumber, err)
-				} else if !valid {
-					log.Error("⚠ Compressed hash mismatch for bundle %06d! Re-fetching...", bundleNumber)
-					os.Remove(path)
-					return bm.LoadBundle(ctx, bundleNumber, plcClient)
-				} else {
-					log.Verbose("✓ Hash verified for bundle %06d", bundleNumber)
-				}
-			}
-		}
+	compressed := bm.encoder.EncodeAll(jsonlData, nil)
+	bf.compressedHash = bm.hash(compressed)
 
-		// Load operations from file
-		operations, err := bm.loadBundleFromFile(path)
-		if err != nil {
-			return nil, false, fmt.Errorf("failed to load bundle from file: %w", err)
-		}
+	return os.WriteFile(bf.path, compressed, 0644)
+}
 
-		// If not in database, index it now
-		if !bundleInDB {
-			// Calculate both hashes
-			fileData, err := os.ReadFile(path)
-			if err != nil {
-				log.Error("Warning: failed to read file: %v", err)
-			} else {
-				compressedHash := bm.calculateHash(fileData)
+func (bm *BundleManager) parseJSONL(data []byte) []PLCOperation {
+	var ops []PLCOperation
+	scanner := bufio.NewScanner(bytes.NewReader(data))
 
-				// Calculate uncompressed hash
-				var jsonlData []byte
-				for _, op := range operations {
-					jsonlData = append(jsonlData, op.RawJSON...)
-					jsonlData = append(jsonlData, '\n')
-				}
-				uncompressedHash := bm.calculateHash(jsonlData)
+	for scanner.Scan() {
+		line := scanner.Bytes()
+		if len(line) == 0 {
+			continue
+		}
 
-				if err := bm.indexBundleWithHash(ctx, bundleNumber, operations, path, uncompressedHash, compressedHash); err != nil {
-					log.Error("Warning: failed to index bundle: %v", err)
-				} else {
-					log.Info("✓ Indexed bundle %06d", bundleNumber)
-				}
-			}
+		var op PLCOperation
+		if err := json.Unmarshal(line, &op); err == nil {
+			op.RawJSON = append([]byte(nil), line...)
+			ops = append(ops, op)
 		}
-
-		// If loaded from disk, it's always complete
-		return operations, true, nil
 	}
 
-	// Bundle doesn't exist locally - fetch from PLC directory
-	log.Info("→ Bundle %06d not found locally, fetching from PLC directory...", bundleNumber)
-
-	var afterTimestamp string
-	var prevBoundaryCIDs map[string]bool
+	return ops
+}
 
-	if bundleNumber > 1 {
-		prevBundle, err := bm.db.GetBundleByNumber(ctx, bundleNumber-1)
-		if err == nil && prevBundle != nil {
-			afterTimestamp = prevBundle.EndTime.Format(time.RFC3339Nano)
-
-			// Get boundary CIDs from previous bundle
-			if len(prevBundle.BoundaryCIDs) > 0 {
-				prevBoundaryCIDs = make(map[string]bool)
-				for _, cid := range prevBundle.BoundaryCIDs {
-					prevBoundaryCIDs[cid] = true
-				}
-				log.Verbose("  Using %d boundary CIDs from previous bundle", len(prevBoundaryCIDs))
-			} else {
-				// Fallback: load previous bundle's operations
-				prevPath := bm.GetBundlePath(bundleNumber - 1)
-				if bm.BundleExists(bundleNumber - 1) {
-					prevOps, err := bm.loadBundleFromFile(prevPath)
-					if err == nil {
-						_, prevBoundaryCIDs = GetBoundaryCIDs(prevOps)
-						log.Verbose("  Computed %d boundary CIDs from previous bundle file", len(prevBoundaryCIDs))
-					}
-				}
-			}
-		}
+func (bm *BundleManager) serializeJSONL(ops []PLCOperation) []byte {
+	var buf []byte
+	for _, op := range ops {
+		buf = append(buf, op.RawJSON...)
+		buf = append(buf, '\n')
 	}
+	return buf
+}
 
-	// Collect operations until we have exactly BUNDLE_SIZE unique ones
-	var allOperations []PLCOperation
-	seenCIDs := make(map[string]bool)
+// ===== BUNDLE FETCHING =====
 
-	// Track what we've already seen from previous bundle
+type bundleFetcher struct {
+	client       *Client
+	seenCIDs     map[string]bool
+	currentAfter string
+	fetchCount   int
+}
+
+func newBundleFetcher(client *Client, afterTime string, prevBoundaryCIDs map[string]bool) *bundleFetcher {
+	seen := make(map[string]bool)
 	for cid := range prevBoundaryCIDs {
-		seenCIDs[cid] = true
+		seen[cid] = true
 	}
 
-	currentAfter := afterTimestamp
-
-	// Scale maxFetches based on bundle size
-	// Assume worst case: 90% dedup rate, need buffer
-	maxFetches := (BUNDLE_SIZE / 900) + 5 // For 10k: ~16 fetches, for 1k: ~6 fetches
-	fetchCount := 0
-
-	for len(allOperations) < BUNDLE_SIZE && fetchCount < maxFetches {
-		fetchCount++
-
-		// Calculate how many more operations we need
-		remaining := BUNDLE_SIZE - len(allOperations)
-
-		// Determine fetch size based on remaining operations
-		var fetchSize int
-		if fetchCount == 1 {
-			// First fetch: always get 1000 (PLC limit)
-			fetchSize = 1000
-		} else if remaining < 100 {
-			// Need less than 100: fetch 50
-			fetchSize = 50
-		} else if remaining < 500 {
-			// Need 100-500: fetch 200
-			fetchSize = 200
-		} else {
-			// Need 500+: fetch 1000
-			fetchSize = 1000
-		}
+	return &bundleFetcher{
+		client:       client,
+		seenCIDs:     seen,
+		currentAfter: afterTime,
+	}
+}
 
-		// Fetch next batch
-		log.Verbose("  Fetch #%d: need %d more, requesting %d", fetchCount, remaining, fetchSize)
+func (bf *bundleFetcher) fetchUntilComplete(ctx context.Context, target int) ([]PLCOperation, bool) {
+	var ops []PLCOperation
+	maxFetches := (target / 900) + 5
 
-		rawOperations, err := bm.fetchBundleFromPLCWithCount(ctx, plcClient, currentAfter, fetchSize)
-		if err != nil {
-			return nil, false, fmt.Errorf("failed to fetch bundle from PLC: %w", err)
-		}
+	for len(ops) < target && bf.fetchCount < maxFetches {
+		bf.fetchCount++
+		batchSize := bf.calculateBatchSize(target - len(ops))
 
-		if len(rawOperations) == 0 {
-			// No more data available
-			log.Info("  No more operations available after %d fetches (got %d/%d)",
-				fetchCount, len(allOperations), BUNDLE_SIZE)
-			break
-		}
+		log.Verbose("  Fetch #%d: need %d more, requesting %d", bf.fetchCount, target-len(ops), batchSize)
 
-		log.Verbose("    Got %d raw operations", len(rawOperations))
+		batch, shouldContinue := bf.fetchBatch(ctx, batchSize)
 
-		// Filter out duplicates and add unique operations
-		newOpsAdded := 0
-		for _, op := range rawOperations {
-			if !seenCIDs[op.CID] {
-				seenCIDs[op.CID] = true
-				allOperations = append(allOperations, op)
-				newOpsAdded++
+		for _, op := range batch {
+			if !bf.seenCIDs[op.CID] {
+				bf.seenCIDs[op.CID] = true
+				ops = append(ops, op)
 
-				if len(allOperations) >= BUNDLE_SIZE {
-					break
+				if len(ops) >= target {
+					return ops[:target], true
 				}
 			}
 		}
 
-		log.Verbose("    Added %d unique operations (total: %d/%d, %d dupes)",
-			newOpsAdded, len(allOperations), BUNDLE_SIZE, len(rawOperations)-newOpsAdded)
-
-		// If we added no new operations, we're stuck
-		if newOpsAdded == 0 {
-			log.Error("  No new unique operations found, stopping")
+		if !shouldContinue {
 			break
 		}
+	}
 
-		// Update cursor for next fetch
-		if len(rawOperations) > 0 {
-			lastOp := rawOperations[len(rawOperations)-1]
-			currentAfter = lastOp.CreatedAt.Format(time.RFC3339Nano)
-		}
+	return ops, len(ops) >= target
+}
 
-		// If PLC returned less than requested, we've reached the end
-		if len(rawOperations) < fetchSize {
-			log.Info("  Reached end of PLC data (got %d < %d requested)", len(rawOperations), fetchSize)
-			break
-		}
+func (bf *bundleFetcher) calculateBatchSize(remaining int) int {
+	if bf.fetchCount == 0 {
+		return 1000
+	}
+	if remaining < 100 {
+		return 50
 	}
-
-	// Warn if we hit the fetch limit
-	if fetchCount >= maxFetches {
-		log.Verbose("  ⚠ Hit maxFetches limit (%d) with only %d/%d operations",
-			maxFetches, len(allOperations), BUNDLE_SIZE)
+	if remaining < 500 {
+		return 200
 	}
+	return 1000
+}
 
-	// Check if we got exactly BUNDLE_SIZE operations
-	isComplete := len(allOperations) >= BUNDLE_SIZE
+func (bf *bundleFetcher) fetchBatch(ctx context.Context, size int) ([]PLCOperation, bool) {
+	ops, err := bf.client.Export(ctx, ExportOptions{
+		Count: size,
+		After: bf.currentAfter,
+	})
 
-	if len(allOperations) > BUNDLE_SIZE {
-		// Trim to exactly BUNDLE_SIZE
-		log.Verbose("  Trimming from %d to %d operations", len(allOperations), BUNDLE_SIZE)
-		allOperations = allOperations[:BUNDLE_SIZE]
+	if err != nil || len(ops) == 0 {
+		return nil, false
 	}
 
-	log.Info("  Collected %d unique operations after %d fetches (complete=%v, target=%d)",
-		len(allOperations), fetchCount, isComplete, BUNDLE_SIZE)
-
-	// Only save as bundle if complete
-	if isComplete {
-		// Save bundle with both hashes
-		uncompressedHash, compressedHash, err := bm.saveBundleFileWithHash(path, allOperations)
-		if err != nil {
-			log.Error("Warning: failed to save bundle file: %v", err)
-		} else {
-			// Index with both hashes
-			if err := bm.indexBundleWithHash(ctx, bundleNumber, allOperations, path, uncompressedHash, compressedHash); err != nil {
-				log.Error("Warning: failed to index bundle: %v", err)
-			} else {
-				log.Info("✓ Bundle %06d saved [%d ops, hash: %s, compressed: %s]",
-					bundleNumber, len(allOperations), uncompressedHash[:16]+"...", compressedHash[:16]+"...")
-			}
-		}
+	if len(ops) > 0 {
+		bf.currentAfter = ops[len(ops)-1].CreatedAt.Format(time.RFC3339Nano)
 	}
 
-	return allOperations, isComplete, nil
+	return ops, len(ops) >= size
 }
 
-// fetchBundleFromPLCWithCount fetches operations with a specific count
-func (bm *BundleManager) fetchBundleFromPLCWithCount(ctx context.Context, client *Client, afterTimestamp string, count int) ([]PLCOperation, error) {
-	return client.Export(ctx, ExportOptions{
-		Count: count,
-		After: afterTimestamp,
-	})
-}
+// ===== MAIN BUNDLE LOADING =====
 
-// saveBundleFileWithHash - NO trailing newline
-func (bm *BundleManager) saveBundleFileWithHash(path string, operations []PLCOperation) (string, string, error) {
-	var jsonlData []byte
-	for _, op := range operations {
-		jsonlData = append(jsonlData, op.RawJSON...)
-		jsonlData = append(jsonlData, '\n')
+func (bm *BundleManager) LoadBundle(ctx context.Context, bundleNum int, plcClient *Client) ([]PLCOperation, bool, error) {
+	if !bm.enabled {
+		return nil, false, fmt.Errorf("bundle manager disabled")
 	}
 
-	uncompressedHash := bm.calculateHash(jsonlData)
-	compressed := bm.encoder.EncodeAll(jsonlData, nil)
-	compressedHash := bm.calculateHash(compressed)
+	bf := bm.newBundleFile(bundleNum)
 
-	if err := os.WriteFile(path, compressed, 0644); err != nil {
-		return "", "", err
+	// Try local file first
+	if bf.exists() {
+		return bm.loadFromFile(ctx, bundleNum, bf)
 	}
 
-	return uncompressedHash, compressedHash, nil
+	// Fetch from PLC
+	return bm.fetchFromPLC(ctx, bundleNum, bf, plcClient)
 }
 
-// fetchBundleFromPLC fetches operations from PLC directory (returns RAW operations)
-func (bm *BundleManager) fetchBundleFromPLC(ctx context.Context, client *Client, afterTimestamp string) ([]PLCOperation, error) {
-	// Just fetch - no deduplication here
-	return client.Export(ctx, ExportOptions{
-		Count: 1000,
-		After: afterTimestamp,
-	})
-}
+func (bm *BundleManager) loadFromFile(ctx context.Context, bundleNum int, bf *bundleFile) ([]PLCOperation, bool, error) {
+	log.Verbose("→ Loading bundle %06d from local file", bundleNum)
 
-// StripBoundaryDuplicates removes operations that were already seen on the previous page
-// This is exported so it can be used in verification
-func StripBoundaryDuplicates(operations []PLCOperation, boundaryTimestamp string, prevBoundaryCIDs map[string]bool) []PLCOperation {
-	if len(operations) == 0 {
-		return operations
+	// Verify hash if bundle is in DB
+	if dbBundle, err := bm.db.GetBundleByNumber(ctx, bundleNum); err == nil && dbBundle != nil {
+		if err := bm.verifyHash(bf.path, dbBundle.CompressedHash); err != nil {
+			log.Error("⚠ Hash mismatch for bundle %06d! Re-fetching...", bundleNum)
+			os.Remove(bf.path)
+			return nil, false, fmt.Errorf("hash mismatch")
+		}
+		log.Verbose("✓ Hash verified for bundle %06d", bundleNum)
 	}
 
-	boundaryTime, err := time.Parse(time.RFC3339Nano, boundaryTimestamp)
-	if err != nil {
-		return operations
+	if err := bm.load(bf); err != nil {
+		return nil, false, err
 	}
 
-	// Skip operations at the start that match the boundary
-	startIdx := 0
-	for startIdx < len(operations) {
-		op := operations[startIdx]
-
-		// If timestamp is AFTER boundary, we're past duplicates
-		if op.CreatedAt.After(boundaryTime) {
-			break
-		}
-
-		// Same timestamp - check if we've seen this CID before
-		if op.CreatedAt.Equal(boundaryTime) {
-			if prevBoundaryCIDs[op.CID] {
-				// This is a duplicate, skip it
-				startIdx++
-				continue
-			}
-			// Same timestamp but new CID - keep it
-			break
-		}
-
-		// Earlier timestamp (shouldn't happen)
-		break
+	// Index if not in DB
+	if _, err := bm.db.GetBundleByNumber(ctx, bundleNum); err != nil {
+		bf.compressedHash = bm.hashFile(bf.path)
+		bf.uncompressedHash = bm.hash(bm.serializeJSONL(bf.operations))
+		bm.indexBundle(ctx, bundleNum, bf)
 	}
 
-	return operations[startIdx:]
+	return bf.operations, true, nil
 }
 
-// Keep the private version for internal use
-func stripBoundaryDuplicates(operations []PLCOperation, boundaryTimestamp string, prevBoundaryCIDs map[string]bool) []PLCOperation {
-	return StripBoundaryDuplicates(operations, boundaryTimestamp, prevBoundaryCIDs)
-}
+func (bm *BundleManager) fetchFromPLC(ctx context.Context, bundleNum int, bf *bundleFile, client *Client) ([]PLCOperation, bool, error) {
+	log.Info("→ Bundle %06d not found locally, fetching from PLC directory...", bundleNum)
 
-// GetBoundaryCIDs returns all CIDs that share the same timestamp as the last operation
-func GetBoundaryCIDs(operations []PLCOperation) (time.Time, map[string]bool) {
-	if len(operations) == 0 {
-		return time.Time{}, nil
-	}
+	afterTime, prevCIDs := bm.getBoundaryInfo(ctx, bundleNum)
+	fetcher := newBundleFetcher(client, afterTime, prevCIDs)
 
-	lastOp := operations[len(operations)-1]
-	boundaryTime := lastOp.CreatedAt
-	cidSet := make(map[string]bool)
+	ops, isComplete := fetcher.fetchUntilComplete(ctx, BUNDLE_SIZE)
 
-	// Walk backwards from the end, collecting all CIDs with the same timestamp
-	for i := len(operations) - 1; i >= 0; i-- {
-		op := operations[i]
-		if op.CreatedAt.Equal(boundaryTime) {
-			cidSet[op.CID] = true
+	log.Info("  Collected %d unique operations after %d fetches (complete=%v)",
+		len(ops), fetcher.fetchCount, isComplete)
+
+	if isComplete {
+		bf.operations = ops
+		if err := bm.save(bf); err != nil {
+			log.Error("Warning: failed to save bundle: %v", err)
 		} else {
-			// Different timestamp, we're done
-			break
+			bm.indexBundle(ctx, bundleNum, bf)
+			log.Info("✓ Bundle %06d saved [%d ops, hash: %s...]",
+				bundleNum, len(ops), bf.uncompressedHash[:16])
 		}
 	}
 
-	return boundaryTime, cidSet
-}
-
-// saveBundleFile (keep for compatibility, calls saveBundleFileWithHash)
-func (bm *BundleManager) saveBundleFile(path string, operations []PLCOperation) error {
-	_, _, err := bm.saveBundleFileWithHash(path, operations) // ✅ All 3 values
-	return err
+	return ops, isComplete, nil
 }
 
-// loadBundleFromFile loads operations from bundle file (JSONL format)
-func (bm *BundleManager) loadBundleFromFile(path string) ([]PLCOperation, error) {
-	// Read compressed file
-	compressedData, err := os.ReadFile(path)
-	if err != nil {
-		return nil, fmt.Errorf("failed to read bundle file: %w", err)
+func (bm *BundleManager) getBoundaryInfo(ctx context.Context, bundleNum int) (string, map[string]bool) {
+	if bundleNum == 1 {
+		return "", nil
 	}
 
-	// Decompress
-	decompressed, err := bm.decoder.DecodeAll(compressedData, nil)
+	prevBundle, err := bm.db.GetBundleByNumber(ctx, bundleNum-1)
 	if err != nil {
-		return nil, fmt.Errorf("failed to decompress bundle: %w", err)
+		return "", nil
 	}
 
-	// Parse JSONL (newline-delimited JSON)
-	var operations []PLCOperation
-	scanner := bufio.NewScanner(bytes.NewReader(decompressed))
+	afterTime := prevBundle.EndTime.Format(time.RFC3339Nano)
 
-	lineNum := 0
-	for scanner.Scan() {
-		lineNum++
-		line := scanner.Bytes()
-
-		// Skip empty lines
-		if len(line) == 0 {
-			continue
+	// Return stored boundary CIDs if available
+	if len(prevBundle.BoundaryCIDs) > 0 {
+		cids := make(map[string]bool)
+		for _, cid := range prevBundle.BoundaryCIDs {
+			cids[cid] = true
 		}
+		return afterTime, cids
+	}
 
-		var op PLCOperation
-		if err := json.Unmarshal(line, &op); err != nil {
-			return nil, fmt.Errorf("failed to parse operation on line %d: %w", lineNum, err)
+	// Fallback: compute from file
+	bf := bm.newBundleFile(bundleNum - 1)
+	if bf.exists() {
+		if err := bm.load(bf); err == nil {
+			_, cids := GetBoundaryCIDs(bf.operations)
+			return afterTime, cids
 		}
-
-		// CRITICAL: Store the original raw JSON bytes
-		op.RawJSON = make([]byte, len(line))
-		copy(op.RawJSON, line)
-
-		operations = append(operations, op)
 	}
 
-	if err := scanner.Err(); err != nil {
-		return nil, fmt.Errorf("error reading JSONL: %w", err)
-	}
-
-	return operations, nil
+	return afterTime, nil
 }
 
-// indexBundleWithHash stores bundle with both hashes
-func (bm *BundleManager) indexBundleWithHash(ctx context.Context, bundleNumber int, operations []PLCOperation, path string, uncompressedHash, compressedHash string) error {
-	// Get previous bundle's hash (uncompressed)
-	var prevBundleHash string
-	if bundleNumber > 1 {
-		prevBundle, err := bm.db.GetBundleByNumber(ctx, bundleNumber-1)
-		if err == nil && prevBundle != nil {
-			prevBundleHash = prevBundle.Hash // Use uncompressed hash for chain
-			log.Verbose("  Linking to previous bundle %06d (hash: %s)", bundleNumber-1, prevBundleHash[:16]+"...")
+// ===== BUNDLE INDEXING =====
+
+func (bm *BundleManager) indexBundle(ctx context.Context, bundleNum int, bf *bundleFile) error {
+	prevHash := ""
+	if bundleNum > 1 {
+		if prev, err := bm.db.GetBundleByNumber(ctx, bundleNum-1); err == nil {
+			prevHash = prev.Hash
 		}
 	}
 
-	// Extract unique DIDs
-	didSet := make(map[string]bool)
-	for _, op := range operations {
-		didSet[op.DID] = true
-	}
-
-	dids := make([]string, 0, len(didSet))
-	for did := range didSet {
-		dids = append(dids, did)
-	}
-
-	// Get compressed file size
-	fileInfo, _ := os.Stat(path)
-	compressedSize := int64(0)
-	if fileInfo != nil {
-		compressedSize = fileInfo.Size()
-	}
+	dids := bm.extractUniqueDIDs(bf.operations)
+	fileSize := bm.getFileSize(bf.path)
 
 	bundle := &storage.PLCBundle{
-		BundleNumber:   bundleNumber,
-		StartTime:      operations[0].CreatedAt,
-		EndTime:        operations[len(operations)-1].CreatedAt,
+		BundleNumber:   bundleNum,
+		StartTime:      bf.operations[0].CreatedAt,
+		EndTime:        bf.operations[len(bf.operations)-1].CreatedAt,
 		DIDs:           dids,
-		Hash:           uncompressedHash, // Primary hash (JSONL)
-		CompressedHash: compressedHash,   // File integrity hash
-		CompressedSize: compressedSize,   // Compressed size
-		PrevBundleHash: prevBundleHash,   // Chain link
+		Hash:           bf.uncompressedHash,
+		CompressedHash: bf.compressedHash,
+		CompressedSize: fileSize,
+		PrevBundleHash: prevHash,
 		Compressed:     true,
 		CreatedAt:      time.Now(),
 	}
···
 	return bm.db.CreateBundle(ctx, bundle)
 }
 
-// indexBundle (keep for compatibility) - FIX: Calculate both hashes
-func (bm *BundleManager) indexBundle(ctx context.Context, bundleNumber int, operations []PLCOperation, path string) error {
-	// Calculate compressed hash from file
-	fileData, err := os.ReadFile(path)
-	if err != nil {
-		return err
+func (bm *BundleManager) extractUniqueDIDs(ops []PLCOperation) []string {
+	didSet := make(map[string]bool)
+	for _, op := range ops {
+		didSet[op.DID] = true
 	}
-	compressedHash := bm.calculateHash(fileData)
 
-	// Calculate uncompressed hash from operations
-	var jsonlData []byte
-	for _, op := range operations {
-		jsonlData = append(jsonlData, op.RawJSON...)
-		jsonlData = append(jsonlData, '\n')
+	dids := make([]string, 0, len(didSet))
+	for did := range didSet {
+		dids = append(dids, did)
 	}
-	uncompressedHash := bm.calculateHash(jsonlData)
-
-	return bm.indexBundleWithHash(ctx, bundleNumber, operations, path, uncompressedHash, compressedHash)
+	return dids
 }
 
-// Update CreateBundleFromMempool
+// ===== MEMPOOL BUNDLE CREATION =====
+
 func (bm *BundleManager) CreateBundleFromMempool(ctx context.Context, operations []PLCOperation) (int, error) {
 	if !bm.enabled {
 		return 0, fmt.Errorf("bundle manager disabled")
···
 	if err != nil {
 		return 0, err
 	}
-	bundleNumber := lastBundle + 1
+	bundleNum := lastBundle + 1
 
-	path := bm.GetBundlePath(bundleNumber)
+	bf := bm.newBundleFile(bundleNum)
+	bf.operations = operations
 
-	// Save bundle with both hashes
-	uncompressedHash, compressedHash, err := bm.saveBundleFileWithHash(path, operations)
-	if err != nil {
+	if err := bm.save(bf); err != nil {
 		return 0, err
 	}
 
-	// Index bundle
-	if err := bm.indexBundleWithHash(ctx, bundleNumber, operations, path, uncompressedHash, compressedHash); err != nil {
+	if err := bm.indexBundle(ctx, bundleNum, bf); err != nil {
 		return 0, err
 	}
 
-	log.Info("✓ Created bundle %06d from mempool (hash: %s)",
-		bundleNumber, uncompressedHash[:16]+"...")
-
-	return bundleNumber, nil
-}
-
-// EnsureBundleContinuity checks that all bundles from 1 to N exist
-func (bm *BundleManager) EnsureBundleContinuity(ctx context.Context, targetBundle int) error {
-	if !bm.enabled {
-		return nil
-	}
-
-	for i := 1; i < targetBundle; i++ {
-		if !bm.BundleExists(i) {
-			// Check if in database
-			_, err := bm.db.GetBundleByNumber(ctx, i)
-			if err != nil {
-				return fmt.Errorf("bundle %06d is missing (required for continuity)", i)
-			}
-		}
-	}
-
-	return nil
-}
+	log.Info("✓ Created bundle %06d from mempool (hash: %s...)",
+		bundleNum, bf.uncompressedHash[:16])
 
-// GetStats returns bundle statistics
-func (bm *BundleManager) GetStats(ctx context.Context) (int64, int64, error) {
-	if !bm.enabled {
-		return 0, 0, nil
-	}
-	return bm.db.GetBundleStats(ctx)
+	return bundleNum, nil
 }
 
-// calculateHash computes SHA256 hash of data
-func (bm *BundleManager) calculateHash(data []byte) string {
-	hash := sha256.Sum256(data)
-	return hex.EncodeToString(hash[:])
-}
+// ===== VERIFICATION =====
 
-// verifyBundleHash checks if file hash matches expected hash
-func (bm *BundleManager) verifyBundleHash(path string, expectedHash string) (bool, error) {
-	data, err := os.ReadFile(path)
-	if err != nil {
-		return false, err
-	}
-
-	actualHash := bm.calculateHash(data)
-	return actualHash == expectedHash, nil
-}
-
-// VerifyChain - FIX
 func (bm *BundleManager) VerifyChain(ctx context.Context, endBundle int) error {
 	if !bm.enabled {
 		return fmt.Errorf("bundle manager disabled")
···
 			return fmt.Errorf("bundle %06d not found: %w", i, err)
 		}
 
-		// Compute file path
-		filePath := bm.GetBundlePath(i)
-
 		// Verify file hash
-		valid, err := bm.verifyBundleHash(filePath, bundle.CompressedHash)
-		if err != nil {
+		path := bm.newBundleFile(i).path
+		if err := bm.verifyHash(path, bundle.CompressedHash); err != nil {
 			return fmt.Errorf("bundle %06d hash verification failed: %w", i, err)
-		}
-		if !valid {
-			return fmt.Errorf("bundle %06d compressed hash mismatch!", i)
 		}
 
 		// Verify chain link
···
 	return nil
 }
 
-// GetChainInfo returns information about the bundle chain
+func (bm *BundleManager) EnsureBundleContinuity(ctx context.Context, targetBundle int) error {
+	if !bm.enabled {
+		return nil
+	}
+
+	for i := 1; i < targetBundle; i++ {
+		if !bm.newBundleFile(i).exists() {
+			if _, err := bm.db.GetBundleByNumber(ctx, i); err != nil {
+				return fmt.Errorf("bundle %06d is missing (required for continuity)", i)
+			}
+		}
+	}
+
+	return nil
+}
+
+// ===== UTILITY METHODS =====
+
+func (bm *BundleManager) hash(data []byte) string {
+	h := sha256.Sum256(data)
+	return hex.EncodeToString(h[:])
+}
+
+func (bm *BundleManager) hashFile(path string) string {
+	data, _ := os.ReadFile(path)
+	return bm.hash(data)
+}
+
+func (bm *BundleManager) verifyHash(path, expectedHash string) error {
+	if expectedHash == "" {
+		return nil
+	}
+
+	actualHash := bm.hashFile(path)
+	if actualHash != expectedHash {
+		return fmt.Errorf("hash mismatch")
+	}
+	return nil
+}
+
+func (bm *BundleManager) getFileSize(path string) int64 {
+	if info, err := os.Stat(path); err == nil {
+		return info.Size()
+	}
+	return 0
+}
+
+func (bm *BundleManager) GetStats(ctx context.Context) (int64, int64, error) {
+	if !bm.enabled {
+		return 0, 0, nil
+	}
+	return bm.db.GetBundleStats(ctx)
+}
+
 func (bm *BundleManager) GetChainInfo(ctx context.Context) (map[string]interface{}, error) {
 	lastBundle, err := bm.db.GetLastBundleNumber(ctx)
 	if err != nil {
···
 		}, nil
 	}
 
-	// Quick check first and last
-	firstBundle, err := bm.db.GetBundleByNumber(ctx, 1)
-	if err != nil {
-		return nil, err
-	}
-
-	lastBundleData, err := bm.db.GetBundleByNumber(ctx, lastBundle)
-	if err != nil {
-		return nil, err
-	}
+	firstBundle, _ := bm.db.GetBundleByNumber(ctx, 1)
+	lastBundleData, _ := bm.db.GetBundleByNumber(ctx, lastBundle)
 
 	return map[string]interface{}{
 		"chain_length":     lastBundle,
···
 		"chain_head_hash":  lastBundleData.Hash,
 	}, nil
 }
+
+// ===== EXPORTED HELPERS =====
+
+func GetBoundaryCIDs(operations []PLCOperation) (time.Time, map[string]bool) {
+	if len(operations) == 0 {
+		return time.Time{}, nil
+	}
+
+	lastOp := operations[len(operations)-1]
+	boundaryTime := lastOp.CreatedAt
+	cidSet := make(map[string]bool)
+
+	for i := len(operations) - 1; i >= 0; i-- {
+		op := operations[i]
+		if op.CreatedAt.Equal(boundaryTime) {
+			cidSet[op.CID] = true
+		} else {
+			break
+		}
+	}
+
+	return boundaryTime, cidSet
+}
+
+func StripBoundaryDuplicates(operations []PLCOperation, boundaryTimestamp string, prevBoundaryCIDs map[string]bool) []PLCOperation {
+	if len(operations) == 0 {
+		return operations
+	}
+
+	boundaryTime, err := time.Parse(time.RFC3339Nano, boundaryTimestamp)
+	if err != nil {
+		return operations
+	}
+
+	startIdx := 0
+	for startIdx < len(operations) {
+		op := operations[startIdx]
+
+		if op.CreatedAt.After(boundaryTime) {
+			break
+		}
+
+		if op.CreatedAt.Equal(boundaryTime) && prevBoundaryCIDs[op.CID] {
+			startIdx++
+			continue
+		}
+
+		break
+	}
+
+	return operations[startIdx:]
+}

internal/plc/scanner.go

···
 
 // ScanMetrics tracks scan progress
 type ScanMetrics struct {
-	totalProcessed int64
+	totalFetched   int64 // Total ops fetched from PLC/bundles
+	totalProcessed int64 // Unique ops processed (after dedup)
+	newEndpoints   int64 // New endpoints discovered
 	endpointCounts map[string]int64
 	currentBundle  int
 	startTime      time.Time
···
 
 func (m *ScanMetrics) logSummary() {
 	summary := formatEndpointCounts(m.endpointCounts)
-	log.Info("PLC scan completed: %d operations, %s in %v",
-		m.totalProcessed, summary, time.Since(m.startTime))
+	if m.newEndpoints > 0 {
+		log.Info("PLC scan completed: %d operations processed (%d fetched), %s in %v",
+			m.totalProcessed, m.totalFetched, summary, time.Since(m.startTime))
+	} else {
+		log.Info("PLC scan completed: %d operations processed (%d fetched), 0 new endpoints in %v",
+			m.totalProcessed, m.totalFetched, time.Since(m.startTime))
+	}
 }
 
 func (s *Scanner) Scan(ctx context.Context) error {
···
 	}
 
 	s.mergeCounts(m.endpointCounts, counts)
-	m.totalProcessed += int64(len(ops))
+	m.totalProcessed += int64(len(ops)) // Unique ops after dedup
+	m.newEndpoints += sumCounts(counts) // NEW: Track new endpoints
 
 	batchTotal := sumCounts(counts)
 	log.Verbose("✓ Processed bundle %06d: %d operations (after dedup), %d new endpoints",
···
 	}
 
 	fetchedCount := len(ops)
+	m.totalFetched += int64(fetchedCount) // Track all fetched
 	log.Verbose("  Fetched %d operations from PLC", fetchedCount)
 
 	if fetchedCount == 0 {
···
 		return false, nil
 	}
 
-	// ✅ Fix: Handle errors from GetMempoolCount
 	beforeCount, err := s.db.GetMempoolCount(ctx)
 	if err != nil {
 		return false, err
 	}
 
+	endpointsBefore := sumCounts(m.endpointCounts)
 	if err := s.addToMempool(ctx, ops, m.endpointCounts); err != nil {
 		return false, err
 	}
+	endpointsAfter := sumCounts(m.endpointCounts)
+	m.newEndpoints += (endpointsAfter - endpointsBefore) // Add new endpoints found
 
 	afterCount, err := s.db.GetMempoolCount(ctx)
 	if err != nil {
 		return false, err
 	}
 
-	m.totalProcessed += int64(fetchedCount)
+	uniqueAdded := int64(afterCount - beforeCount) // Cast to int64
+	m.totalProcessed += uniqueAdded                // Track unique ops processed
+
 	log.Verbose("  Added %d new unique operations to mempool (%d were duplicates)",
-		afterCount-beforeCount, fetchedCount-(afterCount-beforeCount))
+		uniqueAdded, int64(fetchedCount)-uniqueAdded)
 
 	// Continue only if got full batch
 	shouldContinue := fetchedCount >= limit
···
 		}
 
 		// Process and update metrics
+		countsBefore := sumCounts(m.endpointCounts)
 		counts, _ := s.processBatch(ctx, ops)
 		s.mergeCounts(m.endpointCounts, counts)
+		newEndpointsFound := sumCounts(m.endpointCounts) - countsBefore
+
+		m.totalProcessed += int64(len(ops))
+		m.newEndpoints += newEndpointsFound // NEW: Track new endpoints
 		m.currentBundle = bundleNum
 
 		if err := s.updateCursorForBundle(ctx, bundleNum, m.totalProcessed); err != nil {