update

Makefile

···
 
 clean-db:
 	dropdb -U atscanner atscanner
-	createdb atscanner -O atscanner
+	createdb atscanner -O atscanner
+
+update-plcbundle:
+	GOPROXY=direct go get -u github.com/atscan/plcbundle@latest

cmd/atscanner.go

···
 	// Initialize workers
 	log.Info("Initializing scanners...")
 
-	plcScanner := plc.NewScanner(db, cfg.PLC)
+	bundleManager, err := plc.NewBundleManager(cfg.PLC.BundleDir, cfg.PLC.DirectoryURL, db, cfg.PLC.IndexDIDs)
+	if err != nil {
+		log.Fatal("Failed to create bundle manager: %v", err)
+	}
+	defer bundleManager.Close()
+	log.Verbose("✓ Bundle manager initialized (shared)")
+
+	plcScanner := plc.NewScanner(db, cfg.PLC, bundleManager)
 	defer plcScanner.Close()
 	log.Verbose("✓ PLC scanner initialized")
 
···
 
 	// Start API server
 	log.Info("Starting API server on %s:%d...", cfg.API.Host, cfg.API.Port)
-	apiServer := api.NewServer(db, cfg.API, cfg.PLC)
+	apiServer := api.NewServer(db, cfg.API, cfg.PLC, bundleManager)
 	go func() {
 		if err := apiServer.Start(); err != nil {
 			log.Fatal("API server error: %v", err)

go.mod

···
 
 require (
 	github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d
+	github.com/atscan/plcbundle v0.0.0-20251027220105-866aef8771a7
 	github.com/gorilla/handlers v1.5.2
 	github.com/jackc/pgx/v5 v5.7.6
 )
 
 require (
-	github.com/atscan/plcbundle v0.0.0-20251027193653-3678d57c1dee // indirect
 	github.com/felixge/httpsnoop v1.0.3 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect

go.sum

···
 github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d h1:licZJFw2RwpHMqeKTCYkitsPqHNxTmd4SNR5r94FGM8=
 github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d/go.mod h1:asat636LX7Bqt5lYEZ27JNDcqxfjdBQuJ/MM4CN/Lzo=
-github.com/atscan/plcbundle v0.0.0-20251027192009-9350d30fd185 h1:E/fQ1jsaydY6x5JRv+gBiMZVHxKEGD4cK+JxZUZuskU=
-github.com/atscan/plcbundle v0.0.0-20251027192009-9350d30fd185/go.mod h1:vqyqs+zyaxFYtIp6I4+zSQD76oiylnGenzD7ZeA4cxs=
-github.com/atscan/plcbundle v0.0.0-20251027193653-3678d57c1dee h1:wepjgNZxBJGuWmVpplG2BTcoICGafaHALiQoXJV1Iwk=
-github.com/atscan/plcbundle v0.0.0-20251027193653-3678d57c1dee/go.mod h1:vqyqs+zyaxFYtIp6I4+zSQD76oiylnGenzD7ZeA4cxs=
+github.com/atscan/plcbundle v0.0.0-20251027220105-866aef8771a7 h1:u5mCzLGQPSThUPjQnAn64xs3ZWuPltKpua1M+bMxtww=
+github.com/atscan/plcbundle v0.0.0-20251027220105-866aef8771a7/go.mod h1:vqyqs+zyaxFYtIp6I4+zSQD76oiylnGenzD7ZeA4cxs=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=

internal/api/handlers.go

···
 
 import (
 	"context"
-	"crypto/sha256"
 	"database/sql"
-	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
-	"os"
-	"path/filepath"
 	"strconv"
 	"strings"
 	"time"
···
 	"github.com/atscan/atscanner/internal/monitor"
 	"github.com/atscan/atscanner/internal/plc"
 	"github.com/atscan/atscanner/internal/storage"
+	"github.com/atscan/plcbundle"
 	"github.com/gorilla/mux"
 )
 
···
 	http.Error(r.w, msg, code)
 }
 
-func (r *response) bundleHeaders(bundle *storage.PLCBundle) {
+func (r *response) bundleHeaders(bundle *plcbundle.BundleMetadata) {
 	r.w.Header().Set("X-Bundle-Number", fmt.Sprintf("%d", bundle.BundleNumber))
 	r.w.Header().Set("X-Bundle-Hash", bundle.Hash)
 	r.w.Header().Set("X-Bundle-Compressed-Hash", bundle.CompressedHash)
···
 
 // ===== FORMATTING HELPERS =====
 
-func formatBundleResponse(bundle *storage.PLCBundle) map[string]interface{} {
+func formatBundleResponse(bundle *plcbundle.BundleMetadata) map[string]interface{} {
 	return map[string]interface{}{
 		"plc_bundle_number": bundle.BundleNumber,
 		"start_time":        bundle.StartTime,
···
 		return
 	}
 
-	lastBundle, err := s.db.GetLastBundleNumber(ctx)
-	if err != nil {
-		resp.error(err.Error(), http.StatusInternalServerError)
-		return
-	}
-
+	lastBundle := s.bundleManager.GetLastBundleNumber()
 	resp.json(map[string]interface{}{
 		"total_unique_dids": totalDIDs,
 		"last_bundle":       lastBundle,
···
 
 func (s *Server) handleGetPLCBundle(w http.ResponseWriter, r *http.Request) {
 	resp := newResponse(w)
-
 	bundleNum, err := getBundleNumber(r)
 	if err != nil {
 		resp.error("invalid bundle number", http.StatusBadRequest)
 		return
 	}
 
-	// Try to get existing bundle
-	bundle, err := s.db.GetBundleByNumber(r.Context(), bundleNum)
-	if err == nil {
-		// Bundle exists, return it normally
-		resp.json(formatBundleResponse(bundle))
-		return
-	}
-
-	// Bundle not found - check if it's the next upcoming bundle
-	lastBundle, err := s.db.GetLastBundleNumber(r.Context())
+	// Get from library's index
+	index := s.bundleManager.GetIndex()
+	bundleMeta, err := index.GetBundle(bundleNum)
 	if err != nil {
-		resp.error("bundle not found", http.StatusNotFound)
-		return
-	}
-
-	if bundleNum == lastBundle+1 {
-		// This is the upcoming bundle - return preview based on mempool
-		upcomingBundle, err := s.createUpcomingBundlePreview(r.Context(), r, bundleNum)
-		if err != nil {
-			resp.error(fmt.Sprintf("failed to create upcoming bundle preview: %v", err), http.StatusInternalServerError)
+		// Check if it's upcoming bundle
+		lastBundle := index.GetLastBundle()
+		if lastBundle != nil && bundleNum == lastBundle.BundleNumber+1 {
+			upcomingBundle, err := s.createUpcomingBundlePreview(bundleNum)
+			if err != nil {
+				resp.error(err.Error(), http.StatusInternalServerError)
+				return
+			}
+			resp.json(upcomingBundle)
 			return
 		}
-		resp.json(upcomingBundle)
+		resp.error("bundle not found", http.StatusNotFound)
 		return
 	}
 
-	// Not an upcoming bundle, just not found
-	resp.error("bundle not found", http.StatusNotFound)
+	resp.json(formatBundleMetadata(bundleMeta))
 }
 
-func (s *Server) createUpcomingBundlePreview(ctx context.Context, r *http.Request, bundleNum int) (map[string]interface{}, error) {
-	// Get mempool stats
-	mempoolCount, err := s.db.GetMempoolCount(ctx)
-	if err != nil {
-		return nil, err
+// Helper to format library's BundleMetadata
+func formatBundleMetadata(meta *plcbundle.BundleMetadata) map[string]interface{} {
+	return map[string]interface{}{
+		"plc_bundle_number": meta.BundleNumber,
+		"start_time":        meta.StartTime,
+		"end_time":          meta.EndTime,
+		"operation_count":   meta.OperationCount,
+		"did_count":         meta.DIDCount,
+		"hash":              meta.Hash,
+		"compressed_hash":   meta.CompressedHash,
+		"compressed_size":   meta.CompressedSize,
+		"uncompressed_size": meta.UncompressedSize,
+		"compression_ratio": float64(meta.UncompressedSize) / float64(meta.CompressedSize),
+		"cursor":            meta.Cursor,
+		"prev_bundle_hash":  meta.PrevBundleHash,
+		"created_at":        meta.CreatedAt,
 	}
+}
 
-	if mempoolCount == 0 {
+func (s *Server) createUpcomingBundlePreview(bundleNum int) (map[string]interface{}, error) {
+	// Get mempool stats from library via wrapper
+	stats := s.bundleManager.GetMempoolStats()
+
+	count, ok := stats["count"].(int)
+	if !ok || count == 0 {
 		return map[string]interface{}{
 			"plc_bundle_number": bundleNum,
 			"is_upcoming":       true,
···
 		}, nil
 	}
 
-	// Get first and last operations for time range
-	firstOp, err := s.db.GetFirstMempoolOperation(ctx)
-	if err != nil {
-		return nil, err
+	// Build response
+	result := map[string]interface{}{
+		"plc_bundle_number":      bundleNum,
+		"is_upcoming":            true,
+		"status":                 "filling",
+		"operation_count":        count,
+		"target_operation_count": 10000,
+		"progress_percent":       float64(count) / 100.0,
+		"operations_needed":      10000 - count,
 	}
 
-	lastOp, err := s.db.GetLastMempoolOperation(ctx)
-	if err != nil {
-		return nil, err
+	if count >= 10000 {
+		result["status"] = "ready"
 	}
 
-	// Get unique DID count
-	uniqueDIDCount, err := s.db.GetMempoolUniqueDIDCount(ctx)
-	if err != nil {
-		return nil, err
+	// Add time range if available
+	if firstTime, ok := stats["first_time"]; ok {
+		result["start_time"] = firstTime
 	}
-
-	// Get uncompressed size estimate
-	uncompressedSize, err := s.db.GetMempoolUncompressedSize(ctx)
-	if err != nil {
-		return nil, err
+	if lastTime, ok := stats["last_time"]; ok {
+		result["current_end_time"] = lastTime
 	}
 
-	// Estimate compressed size (typical ratio is ~0.1-0.15 for PLC data)
-	estimatedCompressedSize := int64(float64(uncompressedSize) * 0.12)
-
-	// Calculate completion estimate
-	var estimatedCompletionTime *time.Time
-	var operationsNeeded int
-	var currentRate float64
-
-	operationsNeeded = plc.BUNDLE_SIZE - mempoolCount
-
-	if mempoolCount < plc.BUNDLE_SIZE && mempoolCount > 0 {
-		timeSpan := lastOp.CreatedAt.Sub(firstOp.CreatedAt).Seconds()
-		if timeSpan > 0 {
-			currentRate = float64(mempoolCount) / timeSpan
-			if currentRate > 0 {
-				secondsNeeded := float64(operationsNeeded) / currentRate
-				completionTime := time.Now().Add(time.Duration(secondsNeeded) * time.Second)
-				estimatedCompletionTime = &completionTime
-			}
-		}
+	// Add size info if available
+	if sizeBytes, ok := stats["size_bytes"]; ok {
+		result["uncompressed_size"] = sizeBytes
+		result["estimated_compressed_size"] = int64(float64(sizeBytes.(int)) * 0.12)
 	}
 
-	// Get previous bundle for cursor context
-	var prevBundleHash string
-	var cursor string
+	// Get previous bundle info
 	if bundleNum > 1 {
-		prevBundle, err := s.db.GetBundleByNumber(ctx, bundleNum-1)
-		if err == nil {
-			prevBundleHash = prevBundle.Hash
-			cursor = prevBundle.EndTime.Format(time.RFC3339Nano)
-		}
-	}
-
-	// Determine bundle status
-	status := "filling"
-	if mempoolCount >= plc.BUNDLE_SIZE {
-		status = "ready"
-	}
-
-	// Build upcoming bundle response
-	result := map[string]interface{}{
-		"plc_bundle_number":         bundleNum,
-		"is_upcoming":               true,
-		"status":                    status,
-		"operation_count":           mempoolCount,
-		"target_operation_count":    plc.BUNDLE_SIZE,
-		"progress_percent":          float64(mempoolCount) / float64(plc.BUNDLE_SIZE) * 100,
-		"operations_needed":         operationsNeeded,
-		"did_count":                 uniqueDIDCount,
-		"start_time":                firstOp.CreatedAt,
-		"current_end_time":          lastOp.CreatedAt,
-		"uncompressed_size":         uncompressedSize,
-		"estimated_compressed_size": estimatedCompressedSize,
-		"compression_ratio":         float64(uncompressedSize) / float64(estimatedCompressedSize),
-		"prev_bundle_hash":          prevBundleHash,
-		"cursor":                    cursor,
-	}
-
-	if estimatedCompletionTime != nil {
-		result["estimated_completion_time"] = *estimatedCompletionTime
-		result["current_rate_per_second"] = currentRate
-	}
-
-	// Get actual mempool operations if requested (for DIDs list)
-	if r.URL.Query().Get("include_dids") == "true" {
-		ops, err := s.db.GetMempoolOperations(ctx, plc.BUNDLE_SIZE)
-		if err == nil {
-			// Extract unique DIDs
-			didSet := make(map[string]bool)
-			for _, op := range ops {
-				didSet[op.DID] = true
-			}
-			dids := make([]string, 0, len(didSet))
-			for did := range didSet {
-				dids = append(dids, did)
-			}
-			result["dids"] = dids
+		if prevBundle, err := s.bundleManager.GetBundleMetadata(bundleNum - 1); err == nil {
+			result["prev_bundle_hash"] = prevBundle.Hash
+			result["cursor"] = prevBundle.EndTime.Format(time.RFC3339Nano)
 		}
 	}
 
···
 		return
 	}
 
-	bundle, err := s.db.GetBundleByNumber(r.Context(), bundleNum)
+	// Get from library
+	dids, didCount, err := s.bundleManager.GetDIDsForBundle(r.Context(), bundleNum)
 	if err != nil {
 		resp.error("bundle not found", http.StatusNotFound)
 		return
 	}
 
-	// Query DIDs from dids table instead
-	dids, err := s.db.GetDIDsForBundle(r.Context(), bundleNum)
-	if err != nil {
-		resp.error(fmt.Sprintf("failed to get DIDs: %v", err), http.StatusInternalServerError)
-		return
-	}
-
 	resp.json(map[string]interface{}{
-		"plc_bundle_number": bundle.BundleNumber,
-		"did_count":         bundle.DIDCount,
+		"plc_bundle_number": bundleNum,
+		"did_count":         didCount,
 		"dids":              dids,
 	})
 }
···
 
 	compressed := r.URL.Query().Get("compressed") != "false"
 
-	bundle, err := s.db.GetBundleByNumber(r.Context(), bundleNum)
+	bundle, err := s.bundleManager.GetBundleMetadata(bundleNum)
 	if err == nil {
 		// Bundle exists, serve it normally
 		resp.bundleHeaders(bundle)
···
 	}
 
 	// Bundle not found - check if it's the upcoming bundle
-	lastBundle, err := s.db.GetLastBundleNumber(r.Context())
-	if err != nil {
-		resp.error("bundle not found", http.StatusNotFound)
-		return
-	}
-
+	lastBundle := s.bundleManager.GetLastBundleNumber()
 	if bundleNum == lastBundle+1 {
 		// This is the upcoming bundle - serve from mempool
-		s.serveUpcomingBundle(w, r, bundleNum)
+		s.serveUpcomingBundle(w, bundleNum)
 		return
 	}
 
···
 	resp.error("bundle not found", http.StatusNotFound)
 }
 
-func (s *Server) serveUpcomingBundle(w http.ResponseWriter, r *http.Request, bundleNum int) {
-	ctx := r.Context()
+func (s *Server) serveUpcomingBundle(w http.ResponseWriter, bundleNum int) {
+	// Get mempool stats
+	stats := s.bundleManager.GetMempoolStats()
+	count, ok := stats["count"].(int)
 
-	// Get mempool count
-	mempoolCount, err := s.db.GetMempoolCount(ctx)
-	if err != nil {
-		http.Error(w, fmt.Sprintf("failed to get mempool count: %v", err), http.StatusInternalServerError)
-		return
-	}
-
-	if mempoolCount == 0 {
+	if !ok || count == 0 {
 		http.Error(w, "upcoming bundle is empty (no operations in mempool)", http.StatusNotFound)
 		return
 	}
 
-	// Get mempool operations (up to BUNDLE_SIZE)
-	mempoolOps, err := s.db.GetMempoolOperations(ctx, plc.BUNDLE_SIZE)
+	// Get operations from mempool
+	ops, err := s.bundleManager.GetMempoolOperations()
 	if err != nil {
 		http.Error(w, fmt.Sprintf("failed to get mempool operations: %v", err), http.StatusInternalServerError)
 		return
 	}
 
-	if len(mempoolOps) == 0 {
-		http.Error(w, "upcoming bundle is empty", http.StatusNotFound)
+	if len(ops) == 0 {
+		http.Error(w, "no operations in mempool", http.StatusNotFound)
 		return
 	}
 
-	// Get time range
-	firstOp := mempoolOps[0]
-	lastOp := mempoolOps[len(mempoolOps)-1]
+	// Calculate times
+	firstOp := ops[0]
+	lastOp := ops[len(ops)-1]
 
 	// Extract unique DIDs
 	didSet := make(map[string]bool)
-	for _, op := range mempoolOps {
+	for _, op := range ops {
 		didSet[op.DID] = true
 	}
 
+	// Calculate uncompressed size
+	uncompressedSize := int64(0)
+	for _, op := range ops {
+		uncompressedSize += int64(len(op.RawJSON)) + 1 // +1 for newline
+	}
+
 	// Get previous bundle hash
 	prevBundleHash := ""
 	if bundleNum > 1 {
-		if prevBundle, err := s.db.GetBundleByNumber(ctx, bundleNum-1); err == nil {
+		if prevBundle, err := s.bundleManager.GetBundleMetadata(bundleNum - 1); err == nil {
 			prevBundleHash = prevBundle.Hash
 		}
 	}
 
-	// Serialize operations to JSONL
-	var buf []byte
-	for _, mop := range mempoolOps {
-		buf = append(buf, []byte(mop.Operation)...)
-		buf = append(buf, '\n')
-	}
-
-	// Calculate size
-	uncompressedSize := int64(len(buf))
-
 	// Set headers
 	w.Header().Set("X-Bundle-Number", fmt.Sprintf("%d", bundleNum))
 	w.Header().Set("X-Bundle-Is-Upcoming", "true")
 	w.Header().Set("X-Bundle-Status", "preview")
 	w.Header().Set("X-Bundle-Start-Time", firstOp.CreatedAt.Format(time.RFC3339Nano))
 	w.Header().Set("X-Bundle-Current-End-Time", lastOp.CreatedAt.Format(time.RFC3339Nano))
-	w.Header().Set("X-Bundle-Operation-Count", fmt.Sprintf("%d", len(mempoolOps)))
-	w.Header().Set("X-Bundle-Target-Count", fmt.Sprintf("%d", plc.BUNDLE_SIZE))
-	w.Header().Set("X-Bundle-Progress-Percent", fmt.Sprintf("%.2f", float64(len(mempoolOps))/float64(plc.BUNDLE_SIZE)*100))
+	w.Header().Set("X-Bundle-Operation-Count", fmt.Sprintf("%d", len(ops)))
+	w.Header().Set("X-Bundle-Target-Count", "10000")
+	w.Header().Set("X-Bundle-Progress-Percent", fmt.Sprintf("%.2f", float64(len(ops))/100.0))
 	w.Header().Set("X-Bundle-DID-Count", fmt.Sprintf("%d", len(didSet)))
 	w.Header().Set("X-Bundle-Prev-Hash", prevBundleHash)
+	w.Header().Set("X-Uncompressed-Size", fmt.Sprintf("%d", uncompressedSize))
 
 	w.Header().Set("Content-Type", "application/jsonl")
 	w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%06d-upcoming.jsonl", bundleNum))
-	w.Header().Set("Content-Length", fmt.Sprintf("%d", uncompressedSize))
-	w.Header().Set("X-Uncompressed-Size", fmt.Sprintf("%d", uncompressedSize))
 
+	// Stream operations as JSONL
 	w.WriteHeader(http.StatusOK)
-	w.Write(buf)
+
+	for _, op := range ops {
+		// Use RawJSON if available (preserves exact format)
+		if len(op.RawJSON) > 0 {
+			w.Write(op.RawJSON)
+		} else {
+			// Fallback to marshaling
+			data, _ := json.Marshal(op)
+			w.Write(data)
+		}
+		w.Write([]byte("\n"))
+	}
 }
 
-func (s *Server) serveCompressedBundle(w http.ResponseWriter, r *http.Request, bundle *storage.PLCBundle) {
+func (s *Server) serveCompressedBundle(w http.ResponseWriter, r *http.Request, bundle *plcbundle.BundleMetadata) {
 	resp := newResponse(w)
-	path := bundle.GetFilePath(s.plcBundleDir)
 
-	file, err := os.Open(path)
+	// Use the new streaming API for compressed data
+	reader, err := s.bundleManager.StreamRaw(r.Context(), bundle.BundleNumber)
 	if err != nil {
-		resp.error("bundle file not found on disk", http.StatusNotFound)
+		resp.error(fmt.Sprintf("error streaming compressed bundle: %v", err), http.StatusInternalServerError)
 		return
 	}
-	defer file.Close()
-
-	fileInfo, _ := file.Stat()
+	defer reader.Close()
 
 	w.Header().Set("Content-Type", "application/zstd")
 	w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%06d.jsonl.zst", bundle.BundleNumber))
-	w.Header().Set("Content-Length", fmt.Sprintf("%d", fileInfo.Size()))
-	w.Header().Set("X-Compressed-Size", fmt.Sprintf("%d", fileInfo.Size()))
+	w.Header().Set("Content-Length", fmt.Sprintf("%d", bundle.CompressedSize))
+	w.Header().Set("X-Compressed-Size", fmt.Sprintf("%d", bundle.CompressedSize))
 
-	http.ServeContent(w, r, filepath.Base(path), bundle.CreatedAt, file)
+	// Stream the data directly to the response
+	w.WriteHeader(http.StatusOK)
+	io.Copy(w, reader)
 }
 
-func (s *Server) serveUncompressedBundle(w http.ResponseWriter, r *http.Request, bundle *storage.PLCBundle) {
+func (s *Server) serveUncompressedBundle(w http.ResponseWriter, r *http.Request, bundle *plcbundle.BundleMetadata) {
 	resp := newResponse(w)
 
-	ops, err := s.bundleManager.LoadBundleOperations(r.Context(), bundle.BundleNumber)
+	// Use the new streaming API for decompressed data
+	reader, err := s.bundleManager.StreamDecompressed(r.Context(), bundle.BundleNumber)
 	if err != nil {
-		resp.error(fmt.Sprintf("error loading bundle: %v", err), http.StatusInternalServerError)
+		resp.error(fmt.Sprintf("error streaming decompressed bundle: %v", err), http.StatusInternalServerError)
 		return
 	}
-
-	// Serialize to JSONL
-	var buf []byte
-	for _, op := range ops {
-		buf = append(buf, op.RawJSON...)
-		buf = append(buf, '\n')
-	}
-
-	fileInfo, _ := os.Stat(bundle.GetFilePath(s.plcBundleDir))
-	compressedSize := int64(0)
-	if fileInfo != nil {
-		compressedSize = fileInfo.Size()
-	}
+	defer reader.Close()
 
 	w.Header().Set("Content-Type", "application/jsonl")
 	w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%06d.jsonl", bundle.BundleNumber))
-	w.Header().Set("Content-Length", fmt.Sprintf("%d", len(buf)))
-	w.Header().Set("X-Compressed-Size", fmt.Sprintf("%d", compressedSize))
-	w.Header().Set("X-Uncompressed-Size", fmt.Sprintf("%d", len(buf)))
-	if compressedSize > 0 {
-		w.Header().Set("X-Compression-Ratio", fmt.Sprintf("%.2f", float64(len(buf))/float64(compressedSize)))
+	w.Header().Set("Content-Length", fmt.Sprintf("%d", bundle.UncompressedSize))
+	w.Header().Set("X-Compressed-Size", fmt.Sprintf("%d", bundle.CompressedSize))
+	w.Header().Set("X-Uncompressed-Size", fmt.Sprintf("%d", bundle.UncompressedSize))
+	if bundle.CompressedSize > 0 {
+		w.Header().Set("X-Compression-Ratio", fmt.Sprintf("%.2f", float64(bundle.UncompressedSize)/float64(bundle.CompressedSize)))
 	}
 
+	// Stream the data directly to the response
 	w.WriteHeader(http.StatusOK)
-	w.Write(buf)
+	io.Copy(w, reader)
 }
 
 func (s *Server) handleGetPLCBundles(w http.ResponseWriter, r *http.Request) {
 	resp := newResponse(w)
 	limit := getQueryInt(r, "limit", 50)
 
-	bundles, err := s.db.GetBundles(r.Context(), limit)
-	if err != nil {
-		resp.error(err.Error(), http.StatusInternalServerError)
-		return
-	}
+	bundles := s.bundleManager.GetBundles(limit)
 
 	response := make([]map[string]interface{}, len(bundles))
 	for i, bundle := range bundles {
···
 func (s *Server) handleGetPLCBundleStats(w http.ResponseWriter, r *http.Request) {
 	resp := newResponse(w)
 
-	count, compressedSize, uncompressedSize, lastBundle, err := s.db.GetBundleStats(r.Context())
-	if err != nil {
-		resp.error(err.Error(), http.StatusInternalServerError)
-		return
-	}
+	stats := s.bundleManager.GetBundleStats()
+
+	bundleCount := stats["bundle_count"].(int64)
+	totalSize := stats["total_size"].(int64)
+	totalUncompressedSize := stats["total_uncompressed_size"].(int64)
+	lastBundle := stats["last_bundle"].(int64)
 
 	resp.json(map[string]interface{}{
-		"plc_bundle_count":           count,
+		"plc_bundle_count":           bundleCount,
 		"last_bundle_number":         lastBundle,
-		"total_compressed_size":      compressedSize,
-		"total_compressed_size_mb":   float64(compressedSize) / 1024 / 1024,
-		"total_compressed_size_gb":   float64(compressedSize) / 1024 / 1024 / 1024,
-		"total_uncompressed_size":    uncompressedSize,
-		"total_uncompressed_size_mb": float64(uncompressedSize) / 1024 / 1024,
-		"total_uncompressed_size_gb": float64(uncompressedSize) / 1024 / 1024 / 1024,
-		"compression_ratio":          float64(uncompressedSize) / float64(compressedSize),
+		"total_compressed_size":      totalSize,
+		"total_compressed_size_mb":   float64(totalSize) / 1024 / 1024,
+		"total_compressed_size_gb":   float64(totalSize) / 1024 / 1024 / 1024,
+		"total_uncompressed_size":    totalUncompressedSize,
+		"total_uncompressed_size_mb": float64(totalUncompressedSize) / 1024 / 1024,
+		"total_uncompressed_size_gb": float64(totalUncompressedSize) / 1024 / 1024 / 1024,
+		"overall_compression_ratio":  float64(totalUncompressedSize) / float64(totalSize),
 	})
 }
 
···
 
 func (s *Server) handleGetMempoolStats(w http.ResponseWriter, r *http.Request) {
 	resp := newResponse(w)
-	ctx := r.Context()
 
-	count, err := s.db.GetMempoolCount(ctx)
-	if err != nil {
-		resp.error(err.Error(), http.StatusInternalServerError)
-		return
-	}
+	// Get stats from library's mempool via wrapper method
+	stats := s.bundleManager.GetMempoolStats()
 
-	uniqueDIDCount, err := s.db.GetMempoolUniqueDIDCount(ctx)
-	if err != nil {
-		resp.error(err.Error(), http.StatusInternalServerError)
-		return
+	// Convert to API response format
+	result := map[string]interface{}{
+		"operation_count":   stats["count"],
+		"can_create_bundle": stats["can_create_bundle"],
 	}
 
-	uncompressedSize, err := s.db.GetMempoolUncompressedSize(ctx)
-	if err != nil {
-		resp.error(err.Error(), http.StatusInternalServerError)
-		return
+	// Add size information
+	if sizeBytes, ok := stats["size_bytes"]; ok {
+		result["uncompressed_size"] = sizeBytes
+		result["uncompressed_size_mb"] = float64(sizeBytes.(int)) / 1024 / 1024
 	}
 
-	result := map[string]interface{}{
-		"operation_count":      count,
-		"unique_did_count":     uniqueDIDCount,
-		"uncompressed_size":    uncompressedSize,
-		"uncompressed_size_mb": float64(uncompressedSize) / 1024 / 1024,
-		"can_create_bundle":    count >= plc.BUNDLE_SIZE,
-	}
+	// Add time range and calculate estimated completion
+	if count, ok := stats["count"].(int); ok && count > 0 {
+		if firstTime, ok := stats["first_time"].(time.Time); ok {
+			result["mempool_start_time"] = firstTime
 
-	if count > 0 {
-		if firstOp, err := s.db.GetFirstMempoolOperation(ctx); err == nil && firstOp != nil {
-			result["mempool_start_time"] = firstOp.CreatedAt
+			if lastTime, ok := stats["last_time"].(time.Time); ok {
+				result["mempool_end_time"] = lastTime
 
-			if count < plc.BUNDLE_SIZE {
-				if lastOp, err := s.db.GetLastMempoolOperation(ctx); err == nil && lastOp != nil {
-					timeSpan := lastOp.CreatedAt.Sub(firstOp.CreatedAt).Seconds()
+				// Calculate estimated next bundle time if not complete
+				if count < 10000 {
+					timeSpan := lastTime.Sub(firstTime).Seconds()
 					if timeSpan > 0 {
 						opsPerSecond := float64(count) / timeSpan
 						if opsPerSecond > 0 {
-							remainingOps := plc.BUNDLE_SIZE - count
+							remainingOps := 10000 - count
 							secondsNeeded := float64(remainingOps) / opsPerSecond
-							result["estimated_next_bundle_time"] = time.Now().Add(time.Duration(secondsNeeded) * time.Second)
+							estimatedTime := time.Now().Add(time.Duration(secondsNeeded) * time.Second)
+
+							result["estimated_next_bundle_time"] = estimatedTime
+							result["current_rate_per_second"] = opsPerSecond
 							result["operations_needed"] = remainingOps
-							result["current_rate_per_second"] = opsPerSecond
 						}
 					}
+					result["progress_percent"] = float64(count) / 100.0
+				} else {
+					// Ready to create bundle
+					result["estimated_next_bundle_time"] = time.Now()
+					result["operations_needed"] = 0
 				}
-			} else {
-				result["estimated_next_bundle_time"] = time.Now()
-				result["operations_needed"] = 0
 			}
 		}
 	} else {
+		// Empty mempool
 		result["mempool_start_time"] = nil
 		result["estimated_next_bundle_time"] = nil
 	}
···
 
 // ===== VERIFICATION HANDLERS =====
 
-func (s *Server) handleVerifyPLCBundle(w http.ResponseWriter, r *http.Request) {
-	resp := newResponse(w)
-	vars := mux.Vars(r)
-
-	bundleNumber, err := strconv.Atoi(vars["bundleNumber"])
-	if err != nil {
-		resp.error("Invalid bundle number", http.StatusBadRequest)
-		return
-	}
-
-	bundle, err := s.db.GetBundleByNumber(r.Context(), bundleNumber)
-	if err != nil {
-		resp.error("Bundle not found", http.StatusNotFound)
-		return
-	}
-
-	// Fetch from PLC and verify
-	remoteOps, prevCIDs, err := s.fetchRemoteBundleOps(r.Context(), bundleNumber)
-	if err != nil {
-		resp.error(fmt.Sprintf("Failed to fetch from PLC directory: %v", err), http.StatusInternalServerError)
-		return
-	}
-
-	remoteHash := computeOperationsHash(remoteOps)
-	verified := bundle.Hash == remoteHash
-
-	resp.json(map[string]interface{}{
-		"bundle_number":      bundleNumber,
-		"verified":           verified,
-		"local_hash":         bundle.Hash,
-		"remote_hash":        remoteHash,
-		"local_op_count":     plc.BUNDLE_SIZE,
-		"remote_op_count":    len(remoteOps),
-		"boundary_cids_used": len(prevCIDs),
-	})
-}
-
-func (s *Server) fetchRemoteBundleOps(ctx context.Context, bundleNum int) ([]plc.PLCOperation, map[string]bool, error) {
-	var after string
-	var prevBoundaryCIDs map[string]bool
-
-	if bundleNum > 1 {
-		prevBundle, err := s.db.GetBundleByNumber(ctx, bundleNum-1)
-		if err != nil {
-			return nil, nil, fmt.Errorf("failed to get previous bundle: %w", err)
-		}
-
-		after = prevBundle.EndTime.Format("2006-01-02T15:04:05.000Z")
-
-		if len(prevBundle.BoundaryCIDs) > 0 {
-			prevBoundaryCIDs = make(map[string]bool)
-			for _, cid := range prevBundle.BoundaryCIDs {
-				prevBoundaryCIDs[cid] = true
-			}
-		}
-	}
-
-	var allRemoteOps []plc.PLCOperation
-	seenCIDs := make(map[string]bool)
-
-	for cid := range prevBoundaryCIDs {
-		seenCIDs[cid] = true
-	}
-
-	currentAfter := after
-	maxFetches := 20
-
-	for fetchNum := 0; fetchNum < maxFetches && len(allRemoteOps) < plc.BUNDLE_SIZE; fetchNum++ {
-		batch, err := s.plcClient.Export(ctx, plc.ExportOptions{
-			Count: 1000,
-			After: currentAfter,
-		})
-		if err != nil || len(batch) == 0 {
-			break
-		}
-
-		for _, op := range batch {
-			if !seenCIDs[op.CID] {
-				seenCIDs[op.CID] = true
-				allRemoteOps = append(allRemoteOps, op)
-				if len(allRemoteOps) >= plc.BUNDLE_SIZE {
-					break
-				}
-			}
-		}
-
-		if len(batch) > 0 {
-			lastOp := batch[len(batch)-1]
-			currentAfter = lastOp.CreatedAt.Format("2006-01-02T15:04:05.000Z")
-		}
-
-		if len(batch) < 1000 {
-			break
-		}
-	}
-
-	if len(allRemoteOps) > plc.BUNDLE_SIZE {
-		allRemoteOps = allRemoteOps[:plc.BUNDLE_SIZE]
-	}
-
-	return allRemoteOps, prevBoundaryCIDs, nil
-}
-
 func (s *Server) handleVerifyChain(w http.ResponseWriter, r *http.Request) {
 	resp := newResponse(w)
-	ctx := r.Context()
 
-	lastBundle, err := s.db.GetLastBundleNumber(ctx)
-	if err != nil {
-		resp.error(err.Error(), http.StatusInternalServerError)
-		return
-	}
-
+	lastBundle := s.bundleManager.GetLastBundleNumber()
 	if lastBundle == 0 {
 		resp.json(map[string]interface{}{
 			"status":  "empty",
···
 	var errorMsg string
 
 	for i := 1; i <= lastBundle; i++ {
-		bundle, err := s.db.GetBundleByNumber(ctx, i)
+		bundle, err := s.bundleManager.GetBundleMetadata(i)
 		if err != nil {
 			valid = false
 			brokenAt = i
···
 		}
 
 		if i > 1 {
-			prevBundle, err := s.db.GetBundleByNumber(ctx, i-1)
+			prevBundle, err := s.bundleManager.GetBundleMetadata(i - 1)
 			if err != nil {
 				valid = false
 				brokenAt = i
···
 
 func (s *Server) handleGetChainInfo(w http.ResponseWriter, r *http.Request) {
 	resp := newResponse(w)
-	ctx := r.Context()
 
-	lastBundle, err := s.db.GetLastBundleNumber(ctx)
-	if err != nil {
-		resp.error(err.Error(), http.StatusInternalServerError)
-		return
-	}
-
+	lastBundle := s.bundleManager.GetLastBundleNumber()
 	if lastBundle == 0 {
 		resp.json(map[string]interface{}{
 			"chain_length": 0,
···
 		return
 	}
 
-	firstBundle, _ := s.db.GetBundleByNumber(ctx, 1)
-	lastBundleData, _ := s.db.GetBundleByNumber(ctx, lastBundle)
-
-	// Updated to receive 5 values instead of 3
-	count, compressedSize, uncompressedSize, _, err := s.db.GetBundleStats(ctx)
-	if err != nil {
-		resp.error(err.Error(), http.StatusInternalServerError)
-		return
-	}
+	firstBundle, _ := s.bundleManager.GetBundleMetadata(1)
+	lastBundleData, _ := s.bundleManager.GetBundleMetadata(lastBundle)
+	stats := s.bundleManager.GetBundleStats()
 
 	resp.json(map[string]interface{}{
-		"chain_length":               lastBundle,
-		"total_bundles":              count,
-		"total_compressed_size":      compressedSize,
-		"total_compressed_size_mb":   float64(compressedSize) / 1024 / 1024,
-		"total_uncompressed_size":    uncompressedSize,
-		"total_uncompressed_size_mb": float64(uncompressedSize) / 1024 / 1024,
-		"compression_ratio":          float64(uncompressedSize) / float64(compressedSize),
-		"chain_start_time":           firstBundle.StartTime,
-		"chain_end_time":             lastBundleData.EndTime,
-		"chain_head_hash":            lastBundleData.Hash,
-		"first_prev_hash":            firstBundle.PrevBundleHash,
-		"last_prev_hash":             lastBundleData.PrevBundleHash,
+		"chain_length":             lastBundle,
+		"total_bundles":            stats["bundle_count"],
+		"total_compressed_size":    stats["total_size"],
+		"total_compressed_size_mb": float64(stats["total_size"].(int64)) / 1024 / 1024,
+		"chain_start_time":         firstBundle.StartTime,
+		"chain_end_time":           lastBundleData.EndTime,
+		"chain_head_hash":          lastBundleData.Hash,
+		"first_prev_hash":          firstBundle.PrevBundleHash,
+		"last_prev_hash":           lastBundleData.PrevBundleHash,
 	})
 }
 
···
 		return
 	}
 
-	startBundle := s.findStartBundle(ctx, afterTime)
+	startBundle := s.findStartBundle(afterTime)
 	ops := s.collectOperations(ctx, startBundle, afterTime, count)
 
 	w.Header().Set("Content-Type", "application/jsonl")
···
 	return time.Time{}, fmt.Errorf("invalid timestamp format")
 }
 
-func (s *Server) findStartBundle(ctx context.Context, afterTime time.Time) int {
+func (s *Server) findStartBundle(afterTime time.Time) int {
 	if afterTime.IsZero() {
 		return 1
 	}
 
-	foundBundle, err := s.db.GetBundleForTimestamp(ctx, afterTime)
-	if err != nil {
-		return 1
-	}
-
+	foundBundle := s.bundleManager.FindBundleForTimestamp(afterTime)
 	if foundBundle > 1 {
 		return foundBundle - 1
 	}
···
 	var allOps []plc.PLCOperation
 	seenCIDs := make(map[string]bool)
 
-	lastBundle, _ := s.db.GetLastBundleNumber(ctx)
+	lastBundle := s.bundleManager.GetLastBundleNumber()
 
 	for bundleNum := startBundle; bundleNum <= lastBundle && len(allOps) < count; bundleNum++ {
 		ops, err := s.bundleManager.LoadBundleOperations(ctx, bundleNum)
···
 	limit := getQueryInt(r, "limit", 0)
 	fromBundle := getQueryInt(r, "from", 1)
 
-	history, err := s.db.GetPLCHistory(r.Context(), limit, fromBundle)
+	// Use BundleManager instead of database
+	history, err := s.bundleManager.GetPLCHistory(r.Context(), limit, fromBundle)
 	if err != nil {
 		resp.error(err.Error(), http.StatusInternalServerError)
 		return
···
 }
 
 // ===== UTILITY FUNCTIONS =====
-
-func computeOperationsHash(ops []plc.PLCOperation) string {
-	var jsonlData []byte
-	for _, op := range ops {
-		jsonlData = append(jsonlData, op.RawJSON...)
-		jsonlData = append(jsonlData, '\n')
-	}
-	hash := sha256.Sum256(jsonlData)
-	return hex.EncodeToString(hash[:])
-}
 
 func normalizeEndpoint(endpoint string) string {
 	endpoint = strings.TrimPrefix(endpoint, "https://")

internal/api/server.go

···
 	router        *mux.Router
 	server        *http.Server
 	db            storage.Database
-	plcClient     *plc.Client
 	plcBundleDir  string
 	bundleManager *plc.BundleManager
 	plcIndexDIDs  bool
 }
 
-func NewServer(db storage.Database, apiCfg config.APIConfig, plcCfg config.PLCConfig) *Server {
-	bundleManager, err := plc.NewBundleManager(plcCfg.BundleDir, plcCfg.DirectoryURL, db, plcCfg.IndexDIDs)
-	if err != nil {
-		log.Fatal("Failed to create bundle manager: %v", err)
-	}
-
+func NewServer(db storage.Database, apiCfg config.APIConfig, plcCfg config.PLCConfig, bundleManager *plc.BundleManager) *Server {
 	s := &Server{
 		router:        mux.NewRouter(),
 		db:            db,
 		plcBundleDir:  plcCfg.BundleDir,
-		bundleManager: bundleManager,
+		bundleManager: bundleManager, // Use provided shared instance
 		plcIndexDIDs:  plcCfg.IndexDIDs,
 	}
 
···
 	api.HandleFunc("/plc/bundles/{number}", s.handleGetPLCBundle).Methods("GET")
 	api.HandleFunc("/plc/bundles/{number}/dids", s.handleGetPLCBundleDIDs).Methods("GET")
 	api.HandleFunc("/plc/bundles/{number}/download", s.handleDownloadPLCBundle).Methods("GET")
-	api.HandleFunc("/plc/bundles/{bundleNumber}/verify", s.handleVerifyPLCBundle).Methods("POST")
 
 	// PLC history/metrics
 	api.HandleFunc("/plc/history", s.handleGetPLCHistory).Methods("GET")

internal/plc/manager.go

···
 import (
 	"context"
 	"fmt"
+	"io"
+	"sort"
 	"time"
 
 	"github.com/atscan/atscanner/internal/log"
···
 	return bm.libManager.LoadBundle(ctx, bundleNum)
 }
 
-// FetchAndSaveBundle fetches next bundle from PLC and saves to both disk and DB
+// FetchAndSaveBundle fetches next bundle from PLC and saves
 func (bm *BundleManager) FetchAndSaveBundle(ctx context.Context) (*plcbundle.Bundle, error) {
 	// Fetch from PLC using library
 	bundle, err := bm.libManager.FetchNextBundle(ctx)
···
 		return nil, err
 	}
 
-	// Save to disk (library)
+	// Save to disk (library handles this)
 	if err := bm.libManager.SaveBundle(ctx, bundle); err != nil {
 		return nil, fmt.Errorf("failed to save bundle to disk: %w", err)
 	}
 
-	// Save to database
-	if err := bm.saveBundleToDatabase(ctx, bundle); err != nil {
-		return nil, fmt.Errorf("failed to save bundle to database: %w", err)
-	}
-
-	log.Info("✓ Saved bundle %06d (disk + database)", bundle.BundleNumber)
-
-	return bundle, nil
-}
-
-// saveBundleToDatabase saves bundle metadata to PostgreSQL
-func (bm *BundleManager) saveBundleToDatabase(ctx context.Context, bundle *plcbundle.Bundle) error {
-	// Convert library bundle to storage bundle
-	dbBundle := &storage.PLCBundle{
-		BundleNumber:     bundle.BundleNumber,
-		StartTime:        bundle.StartTime,
-		EndTime:          bundle.EndTime,
-		DIDCount:         bundle.DIDCount,
-		Hash:             bundle.Hash,
-		CompressedHash:   bundle.CompressedHash,
-		CompressedSize:   bundle.CompressedSize,
-		UncompressedSize: bundle.UncompressedSize,
-		Cursor:           bundle.Cursor,
-		PrevBundleHash:   bundle.PrevBundleHash,
-		Compressed:       bundle.Compressed,
-		CreatedAt:        bundle.CreatedAt,
-	}
-
-	// Save to database
-	if err := bm.db.CreateBundle(ctx, dbBundle); err != nil {
-		return err
-	}
-
-	// Index DIDs if enabled
+	// Index DIDs if enabled (still use database for this)
 	if bm.indexDIDs && len(bundle.Operations) > 0 {
 		if err := bm.indexBundleDIDs(ctx, bundle); err != nil {
 			log.Error("Failed to index DIDs for bundle %d: %v", bundle.BundleNumber, err)
-			// Don't fail the entire operation
 		}
 	}
 
-	return nil
+	log.Info("✓ Saved bundle %06d", bundle.BundleNumber)
+
+	return bundle, nil
 }
 
 // indexBundleDIDs indexes DIDs from a bundle into the database
···
 func (bm *BundleManager) GetChainInfo(ctx context.Context) (map[string]interface{}, error) {
 	return bm.libManager.GetInfo(), nil
 }
+
+// GetMempoolStats returns mempool statistics from the library
+func (bm *BundleManager) GetMempoolStats() map[string]interface{} {
+	return bm.libManager.GetMempoolStats()
+}
+
+// GetMempoolOperations returns all operations currently in mempool
+func (bm *BundleManager) GetMempoolOperations() ([]PLCOperation, error) {
+	return bm.libManager.GetMempoolOperations()
+}
+
+// GetIndex returns the library's bundle index
+func (bm *BundleManager) GetIndex() *plcbundle.Index {
+	return bm.libManager.GetIndex()
+}
+
+// GetLastBundleNumber returns the last bundle number
+func (bm *BundleManager) GetLastBundleNumber() int {
+	index := bm.libManager.GetIndex()
+	lastBundle := index.GetLastBundle()
+	if lastBundle == nil {
+		return 0
+	}
+	return lastBundle.BundleNumber
+}
+
+// GetBundleMetadata gets bundle metadata by number
+func (bm *BundleManager) GetBundleMetadata(bundleNum int) (*plcbundle.BundleMetadata, error) {
+	index := bm.libManager.GetIndex()
+	return index.GetBundle(bundleNum)
+}
+
+// GetBundles returns the most recent bundles (newest first)
+func (bm *BundleManager) GetBundles(limit int) []*plcbundle.BundleMetadata {
+	index := bm.libManager.GetIndex()
+	allBundles := index.GetBundles()
+
+	// Determine how many bundles to return
+	count := limit
+	if count <= 0 || count > len(allBundles) {
+		count = len(allBundles)
+	}
+
+	// Build result in reverse order (newest first)
+	result := make([]*plcbundle.BundleMetadata, count)
+	for i := 0; i < count; i++ {
+		result[i] = allBundles[len(allBundles)-1-i]
+	}
+
+	return result
+}
+
+// GetBundleStats returns bundle statistics
+func (bm *BundleManager) GetBundleStats() map[string]interface{} {
+	index := bm.libManager.GetIndex()
+	stats := index.GetStats()
+
+	// Convert to expected format
+	lastBundle := stats["last_bundle"]
+	if lastBundle == nil {
+		lastBundle = int64(0)
+	}
+
+	// Calculate total uncompressed size by iterating through all bundles
+	totalUncompressedSize := int64(0)
+	allBundles := index.GetBundles()
+	for _, bundle := range allBundles {
+		totalUncompressedSize += bundle.UncompressedSize
+	}
+
+	return map[string]interface{}{
+		"bundle_count":            int64(stats["bundle_count"].(int)),
+		"total_size":              stats["total_size"].(int64),
+		"total_uncompressed_size": totalUncompressedSize,
+		"last_bundle":             int64(lastBundle.(int)),
+	}
+}
+
+// GetDIDsForBundle gets DIDs from a bundle (loads and extracts)
+func (bm *BundleManager) GetDIDsForBundle(ctx context.Context, bundleNum int) ([]string, int, error) {
+	bundle, err := bm.libManager.LoadBundle(ctx, bundleNum)
+	if err != nil {
+		return nil, 0, err
+	}
+
+	// Extract unique DIDs
+	didSet := make(map[string]bool)
+	for _, op := range bundle.Operations {
+		didSet[op.DID] = true
+	}
+
+	dids := make([]string, 0, len(didSet))
+	for did := range didSet {
+		dids = append(dids, did)
+	}
+
+	return dids, bundle.DIDCount, nil
+}
+
+// FindBundleForTimestamp finds bundle containing a timestamp
+func (bm *BundleManager) FindBundleForTimestamp(afterTime time.Time) int {
+	index := bm.libManager.GetIndex()
+	bundles := index.GetBundles()
+
+	// Find bundle containing this time
+	for _, bundle := range bundles {
+		if (bundle.StartTime.Before(afterTime) || bundle.StartTime.Equal(afterTime)) &&
+			(bundle.EndTime.After(afterTime) || bundle.EndTime.Equal(afterTime)) {
+			return bundle.BundleNumber
+		}
+	}
+
+	// Return closest bundle before this time
+	for i := len(bundles) - 1; i >= 0; i-- {
+		if bundles[i].EndTime.Before(afterTime) {
+			return bundles[i].BundleNumber
+		}
+	}
+
+	return 1 // Default to first bundle
+}
+
+// StreamRaw streams raw compressed bundle data
+func (bm *BundleManager) StreamRaw(ctx context.Context, bundleNumber int) (io.ReadCloser, error) {
+	return bm.libManager.StreamBundleRaw(ctx, bundleNumber)
+}
+
+// StreamDecompressed streams decompressed bundle data
+func (bm *BundleManager) StreamDecompressed(ctx context.Context, bundleNumber int) (io.ReadCloser, error) {
+	return bm.libManager.StreamBundleDecompressed(ctx, bundleNumber)
+}
+
+// GetPLCHistory calculates historical statistics from the bundle index
+func (bm *BundleManager) GetPLCHistory(ctx context.Context, limit int, fromBundle int) ([]*storage.PLCHistoryPoint, error) {
+	index := bm.libManager.GetIndex()
+	allBundles := index.GetBundles()
+
+	// Filter bundles >= fromBundle
+	var filtered []*plcbundle.BundleMetadata
+	for _, b := range allBundles {
+		if b.BundleNumber >= fromBundle {
+			filtered = append(filtered, b)
+		}
+	}
+
+	if len(filtered) == 0 {
+		return []*storage.PLCHistoryPoint{}, nil
+	}
+
+	// Sort bundles by bundle number to ensure proper cumulative calculation
+	sort.Slice(filtered, func(i, j int) bool {
+		return filtered[i].BundleNumber < filtered[j].BundleNumber
+	})
+
+	// Group by date
+	type dailyStat struct {
+		lastBundle        int
+		bundleCount       int
+		totalUncompressed int64
+		totalCompressed   int64
+	}
+
+	dailyStats := make(map[string]*dailyStat)
+
+	// Map to store the cumulative values at the end of each date
+	dateCumulatives := make(map[string]struct {
+		uncompressed int64
+		compressed   int64
+	})
+
+	// Calculate cumulative totals as we iterate through sorted bundles
+	cumulativeUncompressed := int64(0)
+	cumulativeCompressed := int64(0)
+
+	for _, bundle := range filtered {
+		dateStr := bundle.StartTime.Format("2006-01-02")
+
+		// Update cumulative totals
+		cumulativeUncompressed += bundle.UncompressedSize
+		cumulativeCompressed += bundle.CompressedSize
+
+		if stat, exists := dailyStats[dateStr]; exists {
+			// Update existing day
+			if bundle.BundleNumber > stat.lastBundle {
+				stat.lastBundle = bundle.BundleNumber
+			}
+			stat.bundleCount++
+			stat.totalUncompressed += bundle.UncompressedSize
+			stat.totalCompressed += bundle.CompressedSize
+		} else {
+			// Create new day entry
+			dailyStats[dateStr] = &dailyStat{
+				lastBundle:        bundle.BundleNumber,
+				bundleCount:       1,
+				totalUncompressed: bundle.UncompressedSize,
+				totalCompressed:   bundle.CompressedSize,
+			}
+		}
+
+		// Store the cumulative values at the end of this date
+		// (will be overwritten if there are multiple bundles on the same day)
+		dateCumulatives[dateStr] = struct {
+			uncompressed int64
+			compressed   int64
+		}{
+			uncompressed: cumulativeUncompressed,
+			compressed:   cumulativeCompressed,
+		}
+	}
+
+	// Convert map to sorted slice by date
+	var dates []string
+	for date := range dailyStats {
+		dates = append(dates, date)
+	}
+	sort.Strings(dates)
+
+	// Build history points with cumulative operations
+	var history []*storage.PLCHistoryPoint
+	cumulativeOps := 0
+
+	for _, date := range dates {
+		stat := dailyStats[date]
+		cumulativeOps += stat.bundleCount * 10000
+		cumulative := dateCumulatives[date]
+
+		history = append(history, &storage.PLCHistoryPoint{
+			Date:                   date,
+			BundleNumber:           stat.lastBundle,
+			OperationCount:         cumulativeOps,
+			UncompressedSize:       stat.totalUncompressed,
+			CompressedSize:         stat.totalCompressed,
+			CumulativeUncompressed: cumulative.uncompressed,
+			CumulativeCompressed:   cumulative.compressed,
+		})
+	}
+
+	// Apply limit if specified
+	if limit > 0 && len(history) > limit {
+		history = history[:limit]
+	}
+
+	return history, nil
+}

internal/plc/scanner.go

···
 	config        config.PLCConfig
 }
 
-func NewScanner(db storage.Database, cfg config.PLCConfig) *Scanner {
+func NewScanner(db storage.Database, cfg config.PLCConfig, bundleManager *BundleManager) *Scanner {
 	log.Verbose("NewScanner: IndexDIDs config = %v", cfg.IndexDIDs)
 
-	bundleManager, err := NewBundleManager(cfg.BundleDir, cfg.DirectoryURL, db, cfg.IndexDIDs)
-	if err != nil {
-		log.Error("Failed to initialize bundle manager: %v", err)
-		return nil
-	}
-
 	return &Scanner{
-		bundleManager: bundleManager,
+		bundleManager: bundleManager, // Use provided instance
 		db:            db,
 		config:        cfg,
 	}
 }
 
 func (s *Scanner) Close() {
-	if s.bundleManager != nil {
-		s.bundleManager.Close()
-	}
+	// Don't close bundleManager here - it's shared
 }
 
 func (s *Scanner) Scan(ctx context.Context) error {
···
 
 // ScanMetrics tracks scan progress
 type ScanMetrics struct {
-	totalFetched   int64
 	totalProcessed int64
 	newEndpoints   int64
 	endpointCounts map[string]int64

internal/storage/db.go

···
 	GetScanCursor(ctx context.Context, source string) (*ScanCursor, error)
 	UpdateScanCursor(ctx context.Context, cursor *ScanCursor) error
 
-	// Bundle operations
-	CreateBundle(ctx context.Context, bundle *PLCBundle) error
-	GetBundleByNumber(ctx context.Context, bundleNumber int) (*PLCBundle, error)
-	GetBundles(ctx context.Context, limit int) ([]*PLCBundle, error)
-	GetBundlesForDID(ctx context.Context, did string) ([]*PLCBundle, error)
-	GetDIDsForBundle(ctx context.Context, bundleNum int) ([]string, error)
-	GetBundleStats(ctx context.Context) (count, compressedSize, uncompressedSize, lastBundle int64, err error)
-	GetLastBundleNumber(ctx context.Context) (int, error)
-	GetBundleForTimestamp(ctx context.Context, afterTime time.Time) (int, error)
-	GetPLCHistory(ctx context.Context, limit int, fromBundle int) ([]*PLCHistoryPoint, error)
-
-	// Mempool operations
-	AddToMempool(ctx context.Context, ops []MempoolOperation) error
-	GetMempoolCount(ctx context.Context) (int, error)
-	GetMempoolOperations(ctx context.Context, limit int) ([]MempoolOperation, error)
-	DeleteFromMempool(ctx context.Context, ids []int64) error
-	GetFirstMempoolOperation(ctx context.Context) (*MempoolOperation, error)
-	GetLastMempoolOperation(ctx context.Context) (*MempoolOperation, error)
-	GetMempoolUniqueDIDCount(ctx context.Context) (int, error)
-	GetMempoolUncompressedSize(ctx context.Context) (int64, error)
-
 	// Metrics
 	StorePLCMetrics(ctx context.Context, metrics *PLCMetrics) error
 	GetPLCMetrics(ctx context.Context, limit int) ([]*PLCMetrics, error)

internal/storage/postgres.go

···
 	"database/sql"
 	"encoding/json"
 	"fmt"
-	"strings"
 	"time"
 
 	"github.com/atscan/atscanner/internal/log"
···
         last_scan_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
         records_processed BIGINT DEFAULT 0
     );
-
-	CREATE TABLE IF NOT EXISTS plc_bundles (
-		bundle_number INTEGER PRIMARY KEY,
-		start_time TIMESTAMP NOT NULL,
-		end_time TIMESTAMP NOT NULL,
-		did_count INTEGER NOT NULL DEFAULT 0,
-		hash TEXT NOT NULL,
-		compressed_hash TEXT NOT NULL,
-		compressed_size BIGINT NOT NULL,
-		uncompressed_size BIGINT NOT NULL,
-		cumulative_compressed_size BIGINT NOT NULL,
-		cumulative_uncompressed_size BIGINT NOT NULL,
-		cursor TEXT,
-		prev_bundle_hash TEXT,
-		compressed BOOLEAN DEFAULT true,
-		created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-	);
-
-	CREATE INDEX IF NOT EXISTS idx_plc_bundles_time ON plc_bundles(start_time, end_time);
-	CREATE INDEX IF NOT EXISTS idx_plc_bundles_hash ON plc_bundles(hash);
-	CREATE INDEX IF NOT EXISTS idx_plc_bundles_prev ON plc_bundles(prev_bundle_hash);
-	CREATE INDEX IF NOT EXISTS idx_plc_bundles_number_desc ON plc_bundles(bundle_number DESC);
-
-    CREATE TABLE IF NOT EXISTS plc_mempool (
-        id BIGSERIAL PRIMARY KEY,
-        did TEXT NOT NULL,
-        operation TEXT NOT NULL,
-        cid TEXT NOT NULL UNIQUE,
-        created_at TIMESTAMP NOT NULL,
-        added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-    );
-
-    CREATE INDEX IF NOT EXISTS idx_mempool_created_at ON plc_mempool(created_at);
-    CREATE INDEX IF NOT EXISTS idx_mempool_did ON plc_mempool(did);
-    CREATE UNIQUE INDEX IF NOT EXISTS idx_mempool_cid ON plc_mempool(cid);
 
 	-- Minimal dids table
 	CREATE TABLE IF NOT EXISTS dids (
···
 		}
 	}
 	return 0
-}
-
-// ===== BUNDLE OPERATIONS =====
-
-func (p *PostgresDB) CreateBundle(ctx context.Context, bundle *PLCBundle) error {
-	// Calculate cumulative sizes from previous bundle
-	if bundle.BundleNumber > 1 {
-		prevBundle, err := p.GetBundleByNumber(ctx, bundle.BundleNumber-1)
-		if err == nil && prevBundle != nil {
-			bundle.CumulativeCompressedSize = prevBundle.CumulativeCompressedSize + bundle.CompressedSize
-			bundle.CumulativeUncompressedSize = prevBundle.CumulativeUncompressedSize + bundle.UncompressedSize
-		} else {
-			bundle.CumulativeCompressedSize = bundle.CompressedSize
-			bundle.CumulativeUncompressedSize = bundle.UncompressedSize
-		}
-	} else {
-		bundle.CumulativeCompressedSize = bundle.CompressedSize
-		bundle.CumulativeUncompressedSize = bundle.UncompressedSize
-	}
-
-	query := `
-        INSERT INTO plc_bundles (
-            bundle_number, start_time, end_time, did_count, 
-            hash, compressed_hash, compressed_size, uncompressed_size, 
-            cumulative_compressed_size, cumulative_uncompressed_size,
-            cursor, prev_bundle_hash, compressed
-        )
-        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
-        ON CONFLICT(bundle_number) DO UPDATE SET
-            start_time = EXCLUDED.start_time,
-            end_time = EXCLUDED.end_time,
-            did_count = EXCLUDED.did_count,
-            hash = EXCLUDED.hash,
-            compressed_hash = EXCLUDED.compressed_hash,
-            compressed_size = EXCLUDED.compressed_size,
-            uncompressed_size = EXCLUDED.uncompressed_size,
-            cumulative_compressed_size = EXCLUDED.cumulative_compressed_size,
-            cumulative_uncompressed_size = EXCLUDED.cumulative_uncompressed_size,
-            cursor = EXCLUDED.cursor,
-            prev_bundle_hash = EXCLUDED.prev_bundle_hash,
-            compressed = EXCLUDED.compressed
-    `
-	_, err := p.db.ExecContext(ctx, query,
-		bundle.BundleNumber, bundle.StartTime, bundle.EndTime,
-		bundle.DIDCount, bundle.Hash, bundle.CompressedHash,
-		bundle.CompressedSize, bundle.UncompressedSize,
-		bundle.CumulativeCompressedSize, bundle.CumulativeUncompressedSize,
-		bundle.Cursor, bundle.PrevBundleHash, bundle.Compressed,
-	)
-
-	return err
-}
-
-func (p *PostgresDB) GetBundleByNumber(ctx context.Context, bundleNumber int) (*PLCBundle, error) {
-	query := `
-        SELECT bundle_number, start_time, end_time, did_count, hash, compressed_hash, 
-               compressed_size, uncompressed_size, cumulative_compressed_size, 
-               cumulative_uncompressed_size, cursor, prev_bundle_hash, compressed, created_at
-        FROM plc_bundles
-        WHERE bundle_number = $1
-    `
-
-	var bundle PLCBundle
-	var prevHash sql.NullString
-	var cursor sql.NullString
-
-	err := p.db.QueryRowContext(ctx, query, bundleNumber).Scan(
-		&bundle.BundleNumber, &bundle.StartTime, &bundle.EndTime,
-		&bundle.DIDCount, &bundle.Hash, &bundle.CompressedHash,
-		&bundle.CompressedSize, &bundle.UncompressedSize,
-		&bundle.CumulativeCompressedSize, &bundle.CumulativeUncompressedSize,
-		&cursor, &prevHash, &bundle.Compressed, &bundle.CreatedAt,
-	)
-	if err != nil {
-		return nil, err
-	}
-
-	if prevHash.Valid {
-		bundle.PrevBundleHash = prevHash.String
-	}
-	if cursor.Valid {
-		bundle.Cursor = cursor.String
-	}
-
-	return &bundle, nil
-}
-
-func (p *PostgresDB) GetBundles(ctx context.Context, limit int) ([]*PLCBundle, error) {
-	query := `
-        SELECT bundle_number, start_time, end_time, did_count, hash, compressed_hash, 
-               compressed_size, uncompressed_size, cumulative_compressed_size, 
-               cumulative_uncompressed_size, cursor, prev_bundle_hash, compressed, created_at
-        FROM plc_bundles
-        ORDER BY bundle_number DESC
-        LIMIT $1
-    `
-
-	rows, err := p.db.QueryContext(ctx, query, limit)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-
-	return p.scanBundles(rows)
-}
-
-func (p *PostgresDB) GetBundlesForDID(ctx context.Context, did string) ([]*PLCBundle, error) {
-	// Get bundle numbers from dids table
-	var bundleNumbersJSON []byte
-	err := p.db.QueryRowContext(ctx, `
-		SELECT bundle_numbers FROM dids WHERE did = $1
-	`, did).Scan(&bundleNumbersJSON)
-
-	if err == sql.ErrNoRows {
-		return []*PLCBundle{}, nil
-	}
-	if err != nil {
-		return nil, err
-	}
-
-	var bundleNumbers []int
-	if err := json.Unmarshal(bundleNumbersJSON, &bundleNumbers); err != nil {
-		return nil, err
-	}
-
-	if len(bundleNumbers) == 0 {
-		return []*PLCBundle{}, nil
-	}
-
-	// Build query with IN clause
-	placeholders := make([]string, len(bundleNumbers))
-	args := make([]interface{}, len(bundleNumbers))
-	for i, num := range bundleNumbers {
-		placeholders[i] = fmt.Sprintf("$%d", i+1)
-		args[i] = num
-	}
-
-	query := fmt.Sprintf(`
-        SELECT bundle_number, start_time, end_time, did_count, hash, compressed_hash, 
-               compressed_size, uncompressed_size, cumulative_compressed_size, 
-               cumulative_uncompressed_size, cursor, prev_bundle_hash, compressed, created_at
-        FROM plc_bundles
-        WHERE bundle_number IN (%s)
-        ORDER BY bundle_number ASC
-    `, strings.Join(placeholders, ","))
-
-	rows, err := p.db.QueryContext(ctx, query, args...)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-
-	return p.scanBundles(rows)
-}
-
-func (p *PostgresDB) GetDIDsForBundle(ctx context.Context, bundleNum int) ([]string, error) {
-	query := `
-		SELECT did 
-		FROM dids 
-		WHERE bundle_numbers @> $1::jsonb
-		ORDER BY did
-	`
-
-	rows, err := p.db.QueryContext(ctx, query, fmt.Sprintf("[%d]", bundleNum))
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-
-	var dids []string
-	for rows.Next() {
-		var did string
-		if err := rows.Scan(&did); err != nil {
-			return nil, err
-		}
-		dids = append(dids, did)
-	}
-
-	return dids, rows.Err()
-}
-
-func (p *PostgresDB) scanBundles(rows *sql.Rows) ([]*PLCBundle, error) {
-	var bundles []*PLCBundle
-
-	for rows.Next() {
-		var bundle PLCBundle
-		var prevHash sql.NullString
-		var cursor sql.NullString
-
-		if err := rows.Scan(
-			&bundle.BundleNumber,
-			&bundle.StartTime,
-			&bundle.EndTime,
-			&bundle.DIDCount,
-			&bundle.Hash,
-			&bundle.CompressedHash,
-			&bundle.CompressedSize,
-			&bundle.UncompressedSize,
-			&bundle.CumulativeCompressedSize,
-			&bundle.CumulativeUncompressedSize,
-			&cursor,
-			&prevHash,
-			&bundle.Compressed,
-			&bundle.CreatedAt,
-		); err != nil {
-			return nil, err
-		}
-
-		if prevHash.Valid {
-			bundle.PrevBundleHash = prevHash.String
-		}
-		if cursor.Valid {
-			bundle.Cursor = cursor.String
-		}
-
-		bundles = append(bundles, &bundle)
-	}
-
-	return bundles, rows.Err()
-}
-
-func (p *PostgresDB) GetBundleStats(ctx context.Context) (int64, int64, int64, int64, error) {
-	var count, lastBundleNum int64
-	err := p.db.QueryRowContext(ctx, `
-		SELECT COUNT(*), COALESCE(MAX(bundle_number), 0) 
-		FROM plc_bundles
-	`).Scan(&count, &lastBundleNum)
-	if err != nil {
-		return 0, 0, 0, 0, err
-	}
-
-	if lastBundleNum == 0 {
-		return 0, 0, 0, 0, nil
-	}
-
-	var compressedSize, uncompressedSize int64
-	err = p.db.QueryRowContext(ctx, `
-		SELECT cumulative_compressed_size, cumulative_uncompressed_size
-		FROM plc_bundles
-		WHERE bundle_number = $1
-	`, lastBundleNum).Scan(&compressedSize, &uncompressedSize)
-	if err != nil {
-		return 0, 0, 0, 0, err
-	}
-
-	return count, compressedSize, uncompressedSize, lastBundleNum, nil
-}
-
-func (p *PostgresDB) GetLastBundleNumber(ctx context.Context) (int, error) {
-	query := "SELECT COALESCE(MAX(bundle_number), 0) FROM plc_bundles"
-	var num int
-	err := p.db.QueryRowContext(ctx, query).Scan(&num)
-	return num, err
-}
-
-func (p *PostgresDB) GetBundleForTimestamp(ctx context.Context, afterTime time.Time) (int, error) {
-	query := `
-		SELECT bundle_number 
-		FROM plc_bundles 
-		WHERE start_time <= $1 AND end_time >= $1
-		ORDER BY bundle_number ASC 
-		LIMIT 1
-	`
-
-	var bundleNum int
-	err := p.db.QueryRowContext(ctx, query, afterTime).Scan(&bundleNum)
-	if err == sql.ErrNoRows {
-		query = `
-			SELECT bundle_number 
-			FROM plc_bundles 
-			WHERE end_time < $1
-			ORDER BY bundle_number DESC 
-			LIMIT 1
-		`
-		err = p.db.QueryRowContext(ctx, query, afterTime).Scan(&bundleNum)
-		if err == sql.ErrNoRows {
-			return 1, nil
-		}
-		if err != nil {
-			return 0, err
-		}
-		return bundleNum, nil
-	}
-	if err != nil {
-		return 0, err
-	}
-
-	return bundleNum, nil
-}
-
-func (p *PostgresDB) GetPLCHistory(ctx context.Context, limit int, fromBundle int) ([]*PLCHistoryPoint, error) {
-	query := `
-		WITH daily_stats AS (
-			SELECT 
-				DATE(start_time) as date,
-				MAX(bundle_number) as last_bundle,
-				COUNT(*) as bundle_count,
-				SUM(uncompressed_size) as total_uncompressed,
-				SUM(compressed_size) as total_compressed,
-				MAX(cumulative_uncompressed_size) as cumulative_uncompressed,
-				MAX(cumulative_compressed_size) as cumulative_compressed
-			FROM plc_bundles
-			WHERE bundle_number >= $1
-			GROUP BY DATE(start_time)
-		)
-		SELECT 
-			date::text,
-			last_bundle,
-			SUM(bundle_count * 10000) OVER (ORDER BY date) as cumulative_operations,
-			total_uncompressed,
-			total_compressed,
-			cumulative_uncompressed,
-			cumulative_compressed
-		FROM daily_stats
-		ORDER BY date ASC
-	`
-
-	if limit > 0 {
-		query += fmt.Sprintf(" LIMIT %d", limit)
-	}
-
-	rows, err := p.db.QueryContext(ctx, query, fromBundle)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-
-	var history []*PLCHistoryPoint
-	for rows.Next() {
-		var point PLCHistoryPoint
-		var cumulativeOps int64
-
-		err := rows.Scan(
-			&point.Date,
-			&point.BundleNumber,
-			&cumulativeOps,
-			&point.UncompressedSize,
-			&point.CompressedSize,
-			&point.CumulativeUncompressed,
-			&point.CumulativeCompressed,
-		)
-		if err != nil {
-			return nil, err
-		}
-
-		point.OperationCount = int(cumulativeOps)
-
-		history = append(history, &point)
-	}
-
-	return history, rows.Err()
-}
-
-// ===== MEMPOOL OPERATIONS =====
-
-func (p *PostgresDB) AddToMempool(ctx context.Context, ops []MempoolOperation) error {
-	if len(ops) == 0 {
-		return nil
-	}
-
-	tx, err := p.db.BeginTx(ctx, nil)
-	if err != nil {
-		return err
-	}
-	defer tx.Rollback()
-
-	stmt, err := tx.PrepareContext(ctx, `
-        INSERT INTO plc_mempool (did, operation, cid, created_at) 
-        VALUES ($1, $2, $3, $4)
-        ON CONFLICT(cid) DO NOTHING
-    `)
-	if err != nil {
-		return err
-	}
-	defer stmt.Close()
-
-	for _, op := range ops {
-		_, err := stmt.ExecContext(ctx, op.DID, op.Operation, op.CID, op.CreatedAt)
-		if err != nil {
-			return err
-		}
-	}
-
-	return tx.Commit()
-}
-
-func (p *PostgresDB) GetMempoolCount(ctx context.Context) (int, error) {
-	query := "SELECT COUNT(*) FROM plc_mempool"
-	var count int
-	err := p.db.QueryRowContext(ctx, query).Scan(&count)
-	return count, err
-}
-
-func (p *PostgresDB) GetMempoolOperations(ctx context.Context, limit int) ([]MempoolOperation, error) {
-	query := `
-        SELECT id, did, operation, cid, created_at, added_at
-        FROM plc_mempool
-        ORDER BY created_at ASC
-        LIMIT $1
-    `
-
-	rows, err := p.db.QueryContext(ctx, query, limit)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-
-	var ops []MempoolOperation
-	for rows.Next() {
-		var op MempoolOperation
-		err := rows.Scan(&op.ID, &op.DID, &op.Operation, &op.CID, &op.CreatedAt, &op.AddedAt)
-		if err != nil {
-			return nil, err
-		}
-		ops = append(ops, op)
-	}
-
-	return ops, rows.Err()
-}
-
-func (p *PostgresDB) DeleteFromMempool(ctx context.Context, ids []int64) error {
-	if len(ids) == 0 {
-		return nil
-	}
-
-	placeholders := make([]string, len(ids))
-	args := make([]interface{}, len(ids))
-	for i, id := range ids {
-		placeholders[i] = fmt.Sprintf("$%d", i+1)
-		args[i] = id
-	}
-
-	query := fmt.Sprintf("DELETE FROM plc_mempool WHERE id IN (%s)",
-		strings.Join(placeholders, ","))
-
-	_, err := p.db.ExecContext(ctx, query, args...)
-	return err
-}
-
-func (p *PostgresDB) GetFirstMempoolOperation(ctx context.Context) (*MempoolOperation, error) {
-	query := `
-        SELECT id, did, operation, cid, created_at, added_at
-        FROM plc_mempool
-        ORDER BY created_at ASC, id ASC
-        LIMIT 1
-    `
-
-	var op MempoolOperation
-	err := p.db.QueryRowContext(ctx, query).Scan(
-		&op.ID, &op.DID, &op.Operation, &op.CID, &op.CreatedAt, &op.AddedAt,
-	)
-	if err == sql.ErrNoRows {
-		return nil, nil
-	}
-	if err != nil {
-		return nil, err
-	}
-
-	return &op, nil
-}
-
-func (p *PostgresDB) GetLastMempoolOperation(ctx context.Context) (*MempoolOperation, error) {
-	query := `
-        SELECT id, did, operation, cid, created_at, added_at
-        FROM plc_mempool
-        ORDER BY created_at DESC, id DESC
-        LIMIT 1
-    `
-
-	var op MempoolOperation
-	err := p.db.QueryRowContext(ctx, query).Scan(
-		&op.ID, &op.DID, &op.Operation, &op.CID, &op.CreatedAt, &op.AddedAt,
-	)
-	if err == sql.ErrNoRows {
-		return nil, nil
-	}
-	if err != nil {
-		return nil, err
-	}
-
-	return &op, nil
-}
-
-func (p *PostgresDB) GetMempoolUniqueDIDCount(ctx context.Context) (int, error) {
-	query := "SELECT COUNT(DISTINCT did) FROM plc_mempool"
-	var count int
-	err := p.db.QueryRowContext(ctx, query).Scan(&count)
-	return count, err
-}
-
-func (p *PostgresDB) GetMempoolUncompressedSize(ctx context.Context) (int64, error) {
-	query := "SELECT COALESCE(SUM(LENGTH(operation)), 0) FROM plc_mempool"
-	var size int64
-	err := p.db.QueryRowContext(ctx, query).Scan(&size)
-	return size, err
 }
 
 // ===== CURSOR OPERATIONS =====

internal/storage/types.go

···
 	CumulativeCompressed   int64  `json:"cumulative_compressed"`
 }
 
-// MempoolOperation represents an operation waiting to be bundled
-type MempoolOperation struct {
-	ID        int64
-	DID       string
-	Operation string
-	CID       string
-	CreatedAt time.Time
-	AddedAt   time.Time
-}
-
 // ScanCursor stores scanning progress
 type ScanCursor struct {
 	Source           string