update

config.yaml

···
 plc:
   directory_url: "https://plc.directory"
   scan_interval: "6h"
-  batch_size: 1000
   cache_dir: "./plc_cache"    # Cache directory
   use_cache: true             # Enable caching
 

internal/api/handlers.go

···
 	"os"
 	"path/filepath"
 	"strconv"
+	"time"
 
 	"github.com/atscan/atscanner/internal/log"
 	"github.com/atscan/atscanner/internal/plc"
···
 		"plc_bundle_number": bundle.BundleNumber,
 		"start_time":        bundle.StartTime,
 		"end_time":          bundle.EndTime,
-		"operation_count":   1000,
+		"operation_count":   plc.BUNDLE_SIZE,
 		"did_count":         len(bundle.DIDs),
 		"hash":              bundle.Hash,           // Uncompressed (verifiable)
 		"compressed_hash":   bundle.CompressedHash, // File integrity
···
 		if err := json.Unmarshal(line, &op); err != nil {
 			return nil, fmt.Errorf("failed to parse operation on line %d: %w", lineNum, err)
 		}
+
+		// CRITICAL: Store the original raw JSON bytes
+		op.RawJSON = make([]byte, len(line))
+		copy(op.RawJSON, line)
 
 		operations = append(operations, op)
 	}
···
 		}
 	}
 
-	// Fetch from PLC directory
-	remoteOps, err := s.plcClient.Export(ctx, plc.ExportOptions{
-		Count: 1000,
-		After: after,
-	})
-	if err != nil {
-		http.Error(w, fmt.Sprintf("Failed to fetch from PLC directory: %v", err), http.StatusInternalServerError)
-		return
+	// Collect remote operations (may need multiple fetches for large bundles)
+	var allRemoteOps []plc.PLCOperation
+	seenCIDs := make(map[string]bool)
+
+	// Track boundary CIDs
+	for cid := range prevBoundaryCIDs {
+		seenCIDs[cid] = true
 	}
 
-	// Apply same deduplication logic as when creating bundles
-	if after != "" && len(prevBoundaryCIDs) > 0 {
-		remoteOps = plc.StripBoundaryDuplicates(remoteOps, after, prevBoundaryCIDs)
+	currentAfter := after
+	maxFetches := 20 // Enough for up to 20k operations
+
+	for fetchNum := 0; fetchNum < maxFetches && len(allRemoteOps) < plc.BUNDLE_SIZE; fetchNum++ {
+		// Fetch from PLC directory
+		batch, err := s.plcClient.Export(ctx, plc.ExportOptions{
+			Count: 1000,
+			After: currentAfter,
+		})
+		if err != nil {
+			http.Error(w, fmt.Sprintf("Failed to fetch from PLC directory: %v", err), http.StatusInternalServerError)
+			return
+		}
+
+		if len(batch) == 0 {
+			break
+		}
+
+		// Deduplicate and add unique operations
+		for _, op := range batch {
+			if !seenCIDs[op.CID] {
+				seenCIDs[op.CID] = true
+				allRemoteOps = append(allRemoteOps, op)
+				if len(allRemoteOps) >= plc.BUNDLE_SIZE {
+					break
+				}
+			}
+		}
+
+		// Update cursor for next fetch
+		if len(batch) > 0 {
+			lastOp := batch[len(batch)-1]
+			currentAfter = lastOp.CreatedAt.Format("2006-01-02T15:04:05.000Z")
+		}
+
+		// If we got less than 1000, we've reached the end
+		if len(batch) < 1000 {
+			break
+		}
+	}
+
+	// Trim to exact bundle size
+	if len(allRemoteOps) > plc.BUNDLE_SIZE {
+		allRemoteOps = allRemoteOps[:plc.BUNDLE_SIZE]
 	}
 
 	// Compute remote hash (uncompressed JSONL)
-	remoteHash, err := computeRemoteOperationsHash(remoteOps)
+	remoteHash, err := computeRemoteOperationsHash(allRemoteOps)
 	if err != nil {
 		http.Error(w, fmt.Sprintf("Failed to compute remote hash: %v", err), http.StatusInternalServerError)
 		return
···
 		"local_hash":         bundle.Hash,
 		"remote_hash":        remoteHash,
 		"local_op_count":     bundle.OperationCount,
-		"remote_op_count":    len(remoteOps),
+		"remote_op_count":    len(allRemoteOps),
 		"boundary_cids_used": len(prevBoundaryCIDs),
 	})
 }
···
 		"first_prev_hash":  firstBundle.PrevBundleHash, // Should be empty
 		"last_prev_hash":   lastBundleData.PrevBundleHash,
 	})
+}
+
+// handlePLCExport simulates PLC directory /export endpoint using cached bundles
+func (s *Server) handlePLCExport(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+
+	// Parse query parameters
+	countStr := r.URL.Query().Get("count")
+	afterStr := r.URL.Query().Get("after")
+
+	count := 1000 // Default
+	if countStr != "" {
+		if c, err := strconv.Atoi(countStr); err == nil && c > 0 {
+			count = c
+			if count > 10000 {
+				count = 10000 // Max limit
+			}
+		}
+	}
+
+	var afterTime time.Time
+	if afterStr != "" {
+		// Try multiple timestamp formats (from most specific to least)
+		formats := []string{
+			time.RFC3339Nano,           // 2023-11-09T03:55:00.123456789Z
+			time.RFC3339,               // 2023-11-09T03:55:00Z
+			"2006-01-02T15:04:05.000Z", // 2023-11-09T03:55:00.000Z
+			"2006-01-02T15:04:05",      // 2023-11-09T03:55:00
+			"2006-01-02T15:04",         // 2023-11-09T03:55
+			"2006-01-02",               // 2023-11-09
+		}
+
+		var parsed time.Time
+		var parseErr error
+		parsed = time.Time{} // zero value
+
+		for _, format := range formats {
+			parsed, parseErr = time.Parse(format, afterStr)
+			if parseErr == nil {
+				afterTime = parsed
+				break
+			}
+		}
+
+		if parseErr != nil {
+			http.Error(w, fmt.Sprintf("Invalid after parameter: %v", parseErr), http.StatusBadRequest)
+			return
+		}
+	}
+
+	// Find starting bundle (FAST - single query)
+	startBundle := 1
+	if !afterTime.IsZero() {
+		foundBundle, err := s.db.GetBundleForTimestamp(ctx, afterTime)
+		if err != nil {
+			log.Error("Failed to find bundle for timestamp: %v", err)
+			// Fallback to bundle 1
+		} else {
+			startBundle = foundBundle
+			// Go back one bundle to catch boundary timestamps
+			if startBundle > 1 {
+				startBundle--
+			}
+		}
+	}
+
+	// Collect operations from bundles
+	var allOps []plc.PLCOperation
+	seenCIDs := make(map[string]bool)
+
+	// Load bundles sequentially until we have enough operations
+	lastBundle, _ := s.db.GetLastBundleNumber(ctx)
+
+	for bundleNum := startBundle; bundleNum <= lastBundle && len(allOps) < count; bundleNum++ {
+		bundlePath := filepath.Join(s.plcCacheDir, fmt.Sprintf("%06d.jsonl.zst", bundleNum))
+
+		ops, err := s.loadBundleOperations(bundlePath)
+		if err != nil {
+			log.Error("Warning: failed to load bundle %d: %v", bundleNum, err)
+			continue
+		}
+
+		// Filter operations
+		for _, op := range ops {
+			// Skip if STRICTLY BEFORE "after" timestamp
+			// Include operations AT or AFTER the timestamp
+			if !afterTime.IsZero() && op.CreatedAt.Before(afterTime) {
+				continue
+			}
+
+			// Skip duplicates (by CID)
+			if seenCIDs[op.CID] {
+				continue
+			}
+
+			seenCIDs[op.CID] = true
+			allOps = append(allOps, op)
+
+			if len(allOps) >= count {
+				break
+			}
+		}
+	}
+
+	// Set headers for JSONL response
+	w.Header().Set("Content-Type", "application/jsonl")
+	w.Header().Set("X-Operation-Count", strconv.Itoa(len(allOps)))
+
+	// Write JSONL response (newline-delimited JSON with trailing newline)
+	for _, op := range allOps {
+		// Use raw JSON if available
+		if len(op.RawJSON) > 0 {
+			w.Write(op.RawJSON)
+		} else {
+			// Fallback: marshal the operation
+			jsonData, err := json.Marshal(op)
+			if err != nil {
+				log.Error("Failed to marshal operation: %v", err)
+				continue
+			}
+			w.Write(jsonData)
+		}
+
+		// Always add newline after each operation (including the last)
+		w.Write([]byte("\n"))
+	}
 }
 
 // computeRemoteOperationsHash - matching format

internal/api/server.go

···
 	api.HandleFunc("/plc/bundles/{number}/dids", s.handleGetPLCBundleDIDs).Methods("GET")
 	api.HandleFunc("/plc/bundles/{bundleNumber}/verify", s.handleVerifyPLCBundle).Methods("POST")
 	api.HandleFunc("/plc/bundles/{number}", s.handleGetPLCBundle).Methods("GET")
+	api.HandleFunc("/plc/export", s.handlePLCExport).Methods("GET")
 
 	// PLC/DID endpoints
 	api.HandleFunc("/plc/did/{did}", s.handleGetDID).Methods("GET")

internal/pds/client.go

···
 func (c *Client) DescribeServer(ctx context.Context, endpoint string) (*ServerDescription, error) {
 	url := fmt.Sprintf("%s/xrpc/com.atproto.server.describeServer", endpoint)
 
-	fmt.Println(url)
+	//fmt.Println(url)
 
 	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
 	if err != nil {

internal/plc/bundle.go

···
 			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 [1000 ops, hash: %s, compressed: %s]",
-					bundleNumber, uncompressedHash[:16]+"...", compressedHash[:16]+"...")
+				log.Info("✓ Bundle %06d saved [%d ops, hash: %s, compressed: %s]",
+					bundleNumber, len(allOperations), uncompressedHash[:16]+"...", compressedHash[:16]+"...")
 			}
 		}
 	}
···
 // saveBundleFileWithHash - NO trailing newline
 func (bm *BundleManager) saveBundleFileWithHash(path string, operations []PLCOperation) (string, string, error) {
 	var jsonlData []byte
-	for i, op := range operations {
+	for _, op := range operations {
 		jsonlData = append(jsonlData, op.RawJSON...)
-
-		// Add newline ONLY between operations (not after last)
-		if i < len(operations)-1 {
-			jsonlData = append(jsonlData, '\n')
-		}
+		jsonlData = append(jsonlData, '\n')
 	}
 
 	uncompressedHash := bm.calculateHash(jsonlData)

internal/storage/db.go

···
 
 import (
 	"context"
+	"time"
 )
 
 type Database interface {
···
 	GetBundlesForDID(ctx context.Context, did string) ([]*PLCBundle, error)
 	GetBundleStats(ctx context.Context) (int64, int64, error)
 	GetLastBundleNumber(ctx context.Context) (int, error)
+	GetBundleForTimestamp(ctx context.Context, afterTime time.Time) (int, error)
 
 	// Mempool operations
 	AddToMempool(ctx context.Context, ops []MempoolOperation) error

internal/storage/sqlite.go

···
 	return &bundle, nil
 }
 
-// Update GetBundles, GetBundlesForDID similarly
+// GetBundleForTimestamp finds the bundle that should contain operations at or after the given time
+func (s *SQLiteDB) GetBundleForTimestamp(ctx context.Context, afterTime time.Time) (int, error) {
+	query := `
+		SELECT bundle_number 
+		FROM plc_bundles 
+		WHERE start_time <= ? AND end_time >= ?
+		ORDER BY bundle_number ASC 
+		LIMIT 1
+	`
+
+	var bundleNum int
+	err := s.db.QueryRowContext(ctx, query, afterTime, afterTime).Scan(&bundleNum)
+	if err == sql.ErrNoRows {
+		// No exact match, find the closest bundle before this time
+		query = `
+			SELECT bundle_number 
+			FROM plc_bundles 
+			WHERE end_time < ?
+			ORDER BY bundle_number DESC 
+			LIMIT 1
+		`
+		err = s.db.QueryRowContext(ctx, query, afterTime).Scan(&bundleNum)
+		if err == sql.ErrNoRows {
+			return 1, nil // Start from first bundle
+		}
+		if err != nil {
+			return 0, err
+		}
+		return bundleNum, nil // Return the bundle just before
+	}
+	if err != nil {
+		return 0, err
+	}
+
+	return bundleNum, nil
+}
 
 // GetLastBundleNumber gets the highest bundle number
 func (s *SQLiteDB) GetLastBundleNumber(ctx context.Context) (int, error) {

print.sh

···
-git ls-tree -r --name-only HEAD | while read filename; do
-  echo "--- FILE: $filename ---"
-  git show HEAD:"$filename"
-  echo ""
-done > all_repo_content.txt

utils/check-minimal.sh

···
+pv plc_cache/*.jsonl.zst | zstdcat | \
+  jq -r '[.createdAt, .cid, .did] | @tsv' | \
+  awk '
+    NR > 1 {
+      if ($2 == prev_cid) {
+        printf "Duplicate CID: %s\n", $2
+        printf "  Prev: time=%s DID=%s\n", prev_time, prev_did
+        printf "  Curr: time=%s DID=%s\n\n", $1, $3
+      }
+    }
+    {prev_time = $1; prev_cid = $2; prev_did = $3}
+  '

utils/verify-export-debug.sh

···
+#!/bin/bash
+# verify-export-debug.sh - Deep comparison of export endpoints
+
+AFTER="${1:-}"
+COUNT="${2:-1000}"
+LOCAL_URL="http://localhost:8080/api/v1/plc/export"
+REMOTE_URL="https://plc.directory/export"
+
+# Build query parameters
+PARAMS="count=$COUNT"
+if [ -n "$AFTER" ]; then
+    PARAMS="${PARAMS}&after=${AFTER}"
+fi
+
+echo "=== Fetching data ==="
+curl -s "${LOCAL_URL}?${PARAMS}" > /tmp/local_export.jsonl
+curl -s "${REMOTE_URL}?${PARAMS}" > /tmp/remote_export.jsonl
+
+echo "Local file size:  $(wc -c < /tmp/local_export.jsonl) bytes"
+echo "Remote file size: $(wc -c < /tmp/remote_export.jsonl) bytes"
+echo ""
+
+echo "Local lines:  $(wc -l < /tmp/local_export.jsonl)"
+echo "Remote lines: $(wc -l < /tmp/remote_export.jsonl)"
+echo ""
+
+# Check for trailing newline
+echo "Local ends with newline:  $(tail -c 1 /tmp/local_export.jsonl | xxd -p)"
+echo "Remote ends with newline: $(tail -c 1 /tmp/remote_export.jsonl | xxd -p)"
+echo "(0a = newline, other = no trailing newline)"
+echo ""
+
+# Compare line by line
+echo "=== Comparing CIDs line by line ==="
+LOCAL_CIDS=$(cat /tmp/local_export.jsonl | jq -r '.cid' 2>/dev/null)
+REMOTE_CIDS=$(cat /tmp/remote_export.jsonl | jq -r '.cid' 2>/dev/null)
+
+if [ "$LOCAL_CIDS" = "$REMOTE_CIDS" ]; then
+    echo "✅ All CIDs match in order"
+else
+    echo "❌ CIDs differ"
+    echo ""
+    echo "First 5 local CIDs:"
+    echo "$LOCAL_CIDS" | head -5
+    echo ""
+    echo "First 5 remote CIDs:"
+    echo "$REMOTE_CIDS" | head -5
+fi
+echo ""
+
+# Compare exact JSON of first operation
+echo "=== First operation comparison ==="
+echo "Local:"
+head -1 /tmp/local_export.jsonl | jq . 2>/dev/null || head -1 /tmp/local_export.jsonl
+echo ""
+echo "Remote:"
+head -1 /tmp/remote_export.jsonl | jq . 2>/dev/null || head -1 /tmp/remote_export.jsonl
+echo ""
+
+# Check if it's just a trailing newline issue
+echo "=== Testing trailing newline hypothesis ==="
+LOCAL_HASH_NO_TRAIL=$(head -c -1 /tmp/local_export.jsonl | shasum -a 256 | cut -d' ' -f1)
+REMOTE_HASH_NO_TRAIL=$(head -c -1 /tmp/remote_export.jsonl | shasum -a 256 | cut -d' ' -f1)
+
+LOCAL_HASH_WITH_TRAIL=$(cat /tmp/local_export.jsonl && echo "" | shasum -a 256 | cut -d' ' -f1)
+
+echo "Local hash (as-is):           $(shasum -a 256 < /tmp/local_export.jsonl | cut -d' ' -f1)"
+echo "Remote hash (as-is):          $(shasum -a 256 < /tmp/remote_export.jsonl | cut -d' ' -f1)"
+echo "Local hash (no trailing \\n):  $LOCAL_HASH_NO_TRAIL"
+echo "Remote hash (no trailing \\n): $REMOTE_HASH_NO_TRAIL"
+
+if [ "$LOCAL_HASH_NO_TRAIL" = "$(shasum -a 256 < /tmp/remote_export.jsonl | cut -d' ' -f1)" ]; then
+    echo ""
+    echo "🔍 Found it! Local is missing trailing newline"
+elif [ "$(shasum -a 256 < /tmp/local_export.jsonl | cut -d' ' -f1)" = "$REMOTE_HASH_NO_TRAIL" ]; then
+    echo ""
+    echo "🔍 Found it! Remote is missing trailing newline"
+fi
+
+# Clean up
+rm -f /tmp/local_export.jsonl /tmp/remote_export.jsonl
+

utils/verify-export.sh

···
+#!/bin/bash
+# verify-export.sh - Verify local PLC export endpoint against plc.directory
+# Usage: ./verify-export.sh [after_timestamp] [count]
+
+AFTER="${1:-}"
+COUNT="${2:-50}"
+LOCAL_URL="http://localhost:8080/api/v1/plc/export"
+REMOTE_URL="https://plc.directory/export"
+
+echo "=== PLC Export Verification ==="
+echo "Count: $COUNT"
+if [ -n "$AFTER" ]; then
+    echo "After: $AFTER"
+else
+    echo "After: (none - from beginning)"
+fi
+echo ""
+
+# Build query parameters
+PARAMS="count=$COUNT"
+if [ -n "$AFTER" ]; then
+    PARAMS="${PARAMS}&after=${AFTER}"
+fi
+
+echo "Fetching from local API..."
+echo "curl -s \"${LOCAL_URL}?${PARAMS}\""
+LOCAL_DATA=$(curl -s "${LOCAL_URL}?${PARAMS}")
+LOCAL_COUNT=$(echo "$LOCAL_DATA" | wc -l | tr -d ' ')
+LOCAL_HASH=$(echo "$LOCAL_DATA" | shasum -a 256 | cut -d' ' -f1)
+
+echo "  Operations: $LOCAL_COUNT"
+echo "  Hash: $LOCAL_HASH"
+echo ""
+
+echo "Fetching from plc.directory..."
+REMOTE_DATA=$(curl -s "${REMOTE_URL}?${PARAMS}")
+REMOTE_COUNT=$(echo "$REMOTE_DATA" | wc -l | tr -d ' ')
+REMOTE_HASH=$(echo "$REMOTE_DATA" | shasum -a 256 | cut -d' ' -f1)
+
+echo "  Operations: $REMOTE_COUNT"
+echo "  Hash: $REMOTE_HASH"
+echo ""
+
+# Compare
+echo "=== COMPARISON ==="
+if [ "$LOCAL_HASH" = "$REMOTE_HASH" ]; then
+    echo "✅ MATCH! Hashes are identical"
+    echo ""
+    echo "Local and remote exports are in sync! 🎯"
+    exit 0
+else
+    echo "❌ MISMATCH! Hashes differ"
+    echo ""
+    
+    # Show counts
+    if [ "$LOCAL_COUNT" != "$REMOTE_COUNT" ]; then
+        echo "⚠️  Operation count differs:"
+        echo "   Local:  $LOCAL_COUNT operations"
+        echo "   Remote: $REMOTE_COUNT operations"
+        echo "   Diff:   $((REMOTE_COUNT - LOCAL_COUNT))"
+        echo ""
+    fi
+    
+    # Sample first and last operations
+    echo "First operation (local):"
+    echo "$LOCAL_DATA" | head -1 | jq -r '[.did, .cid, .createdAt] | @tsv' 2>/dev/null || echo "(parse error)"
+    echo ""
+    
+    echo "First operation (remote):"
+    echo "$REMOTE_DATA" | head -1 | jq -r '[.did, .cid, .createdAt] | @tsv' 2>/dev/null || echo "(parse error)"
+    echo ""
+    
+    echo "Last operation (local):"
+    echo "$LOCAL_DATA" | tail -1 | jq -r '[.did, .cid, .createdAt] | @tsv' 2>/dev/null || echo "(parse error)"
+    echo ""
+    
+    echo "Last operation (remote):"
+    echo "$REMOTE_DATA" | tail -1 | jq -r '[.did, .cid, .createdAt] | @tsv' 2>/dev/null || echo "(parse error)"
+    echo ""
+    
+    # Find first difference
+    echo "Finding first difference..."
+    diff <(echo "$LOCAL_DATA" | jq -r '.cid' 2>/dev/null | head -20) \
+         <(echo "$REMOTE_DATA" | jq -r '.cid' 2>/dev/null | head -20) || true
+    
+    exit 1
+fi
+