update

README.md

···
+# plcbundle-rs
+
+Main purpose of this package is to provide universal and simple interface, as Rust library or C/GO bindings for manipulation with plcbundle repository.

bindings/go/README.md

···
+# PLC Bundle Go Bindings
+
+High-performance Go bindings for the PLC Bundle Rust library.
+
+## Installation
+
+```bash
+# Build the Rust library first
+cd ../..
+cargo build --release
+```
+
+## Quick Start
+
+```go
+import "plcbundle"
+
+manager, err := plcbundle.NewBundleManager("./bundles")
+if err != nil {
+    log.Fatal(err)
+}
+defer manager.Close()
+```
+
+## API Reference
+
+### BundleManager Methods
+
+- `NewBundleManager(bundleDir string) (*BundleManager, error)` - Create a new manager
+- `Close()` - Free resources
+
+**Loading:**
+- `LoadBundle(bundleNum uint32, opts *LoadOptions) (*LoadResult, error)`
+- `GetOperationsBatch(requests []OperationRequest) ([]Operation, error)`
+
+**DID Operations:**
+- `GetDIDOperations(did string) ([]Operation, error)`
+- `BatchResolveDIDs(dids []string) (map[string][]Operation, error)`
+
+**Query:**
+- `Query(queryStr string, bundleStart, bundleEnd uint32) ([]Operation, error)`
+
+**Export:**
+- `Export(spec ExportSpec, opts *ExportOptions) (*ExportStats, error)`
+
+**Verification:**
+- `VerifyBundle(bundleNum uint32, checkHash, checkChain bool) (*VerifyResult, error)`
+- `VerifyChain(startBundle, endBundle uint32) error`
+
+**Information:**
+- `GetBundleInfo(bundleNum uint32, includeOperations, includeDIDs bool) (*BundleInfo, error)`
+
+**Cache Management:**
+- `PrefetchBundles(bundleNums []uint32)`
+- `WarmUp(strategy WarmUpStrategy, startBundle, endBundle uint32) error`
+- `ClearCaches()`
+
+**DID Index:**
+- `RebuildDIDIndex(progressCallback func(uint32, uint64)) (*RebuildStats, error)`
+- `GetDIDIndexStats() (*DIDIndexStats, error)`
+
+**Observability:**
+- `GetStats() (*ManagerStats, error)`
+
+## Types
+
+**Options & Results:**
+- `LoadOptions` - Bundle loading options
+- `LoadResult` - Load operation result
+- `ExportSpec` - Export configuration
+- `ExportOptions` - Export options (writer, progress callback)
+- `ExportStats` - Export statistics
+- `VerifyResult` - Verification result
+
+**Data:**
+- `Operation` - Operation data (DID, OpType, CID, JSON, etc.)
+- `OperationRequest` - Request for specific operation
+- `BundleInfo` - Bundle metadata
+
+**Statistics:**
+- `ManagerStats` - Manager performance stats
+- `DIDIndexStats` - DID index statistics
+- `RebuildStats` - Index rebuild statistics
+
+**Constants:**
+- `WarmUpStrategy` - WarmUpRecent, WarmUpAll, WarmUpRange
+
+## Examples
+
+**Export:**
+```go
+spec := plcbundle.ExportSpec{
+    BundleStart: 1,
+    BundleEnd:   100,
+    Format:      0, // 0=jsonl, 1=json, 2=csv
+}
+opts := &plcbundle.ExportOptions{Writer: os.Stdout}
+stats, err := manager.Export(spec, opts)
+```
+
+**Query:**
+```go
+ops, err := manager.Query("@", 1, 100) // All operations from bundles 1-100
+```
+
+**DID Lookup:**
+```go
+ops, err := manager.GetDIDOperations("did:plc:example")
+```

bindings/go/callback.c

···
-// Callback utilities - currently unused
-// This file can be removed or used for future callback implementations

bindings/go/cmd/example/main.go

···
 import (
 	"flag"
 	"fmt"
+	"io"
 	"log"
 	"os"
 	"strconv"
···
 	fmt.Println("  warm-up        Warm up the cache")
 	fmt.Println("  rebuild-index  Rebuild the DID index")
 	fmt.Println("  index-stats    Display DID index statistics")
+	fmt.Println("  export         Export operations to file or stdout")
 	fmt.Println("  help           Show this help message")
 	fmt.Println("\nExamples:")
 	fmt.Println("  plcbundle-go -dir test_bundles load -bundle 1")
···
 		cmdRebuildIndex(manager, args[1:])
 	case "index-stats":
 		cmdIndexStats(manager, args[1:])
+	case "export":
+		cmdExport(manager, args[1:])
 	case "help":
 		printUsage()
 	default:
···
 	fmt.Printf("Index Size:            %d bytes (%.2f MB)\n",
 		stats.IndexSizeBytes, float64(stats.IndexSizeBytes)/1024/1024)
 }
+
+func cmdExport(manager *plcbundle.BundleManager, args []string) {
+	fs := flag.NewFlagSet("export", flag.ExitOnError)
+	rangeFlag := fs.String("range", "", "Bundle range (e.g., \"1-100\")")
+	all := fs.Bool("all", false, "Export all bundles")
+	count := fs.Uint64("count", 0, "Limit number of operations (0 = no limit)")
+	after := fs.String("after", "", "Export operations after timestamp (ISO 8601)")
+	did := fs.String("did", "", "Filter by DID")
+	opType := fs.String("op-type", "", "Filter by operation type")
+	format := fs.String("f", "jsonl", "Output format: jsonl, json, csv")
+	output := fs.String("o", "", "Output file (default: stdout)")
+	quiet := fs.Bool("q", false, "Suppress progress output")
+
+	fs.Usage = func() {
+		fmt.Fprintf(os.Stderr, "Export operations from bundles\n\n")
+		fmt.Fprintf(os.Stderr, "Usage: plcbundle-go export [OPTIONS]\n\n")
+		fmt.Fprintf(os.Stderr, "Options:\n")
+		fs.PrintDefaults()
+		fmt.Fprintf(os.Stderr, "\nExamples:\n")
+		fmt.Fprintf(os.Stderr, "  plcbundle-go export --range 1-100\n")
+		fmt.Fprintf(os.Stderr, "  plcbundle-go export --all --count 50000\n")
+		fmt.Fprintf(os.Stderr, "  plcbundle-go export --all --after 2024-01-01T00:00:00Z\n")
+		fmt.Fprintf(os.Stderr, "  plcbundle-go export --range 1-100 --count 10000 --after 2024-01-01T00:00:00Z\n")
+	}
+
+	fs.Parse(args)
+
+	// Determine bundle range
+	var bundleStart, bundleEnd uint32
+	exportAll := *all
+
+	if !exportAll {
+		if *rangeFlag == "" {
+			log.Fatal("Must specify either --range or --all")
+		}
+
+		// Parse range like "1-100"
+		parts := strings.Split(*rangeFlag, "-")
+		if len(parts) == 2 {
+			start, err := strconv.ParseUint(parts[0], 10, 32)
+			if err != nil {
+				log.Fatalf("Invalid range start: %v", err)
+			}
+			end, err := strconv.ParseUint(parts[1], 10, 32)
+			if err != nil {
+				log.Fatalf("Invalid range end: %v", err)
+			}
+			bundleStart = uint32(start)
+			bundleEnd = uint32(end)
+		} else {
+			// Single bundle
+			num, err := strconv.ParseUint(*rangeFlag, 10, 32)
+			if err != nil {
+				log.Fatalf("Invalid bundle number: %v", err)
+			}
+			bundleStart = uint32(num)
+			bundleEnd = uint32(num)
+		}
+	}
+
+	// Determine format
+	var formatCode uint8
+	switch *format {
+	case "jsonl":
+		formatCode = 0
+	case "json":
+		formatCode = 1
+	case "csv":
+		formatCode = 2
+	default:
+		log.Fatalf("Invalid format: %s (use jsonl, json, or csv)", *format)
+	}
+
+	// Open output
+	var writer io.Writer
+	if *output == "" {
+		writer = os.Stdout
+	} else {
+		file, err := os.Create(*output)
+		if err != nil {
+			log.Fatalf("Failed to create output file: %v", err)
+		}
+		defer file.Close()
+		writer = file
+	}
+
+	// Create export spec
+	spec := plcbundle.ExportSpec{
+		BundleStart:    bundleStart,
+		BundleEnd:      bundleEnd,
+		ExportAll:      exportAll,
+		Format:         formatCode,
+		CountLimit:     *count,
+		AfterTimestamp: *after,
+		DIDFilter:      *did,
+		OpTypeFilter:   *opType,
+	}
+
+	// Progress callback
+	var progressCallback func(uint64, uint64)
+	if !*quiet {
+		progressCallback = func(records, bytes uint64) {
+			if records%10000 == 0 {
+				fmt.Fprintf(os.Stderr, "\r   Exported: %d operations (%.2f MB)", records, float64(bytes)/1024/1024)
+			}
+		}
+	}
+
+	// Export options
+	opts := &plcbundle.ExportOptions{
+		Writer:   writer,
+		Progress: progressCallback,
+	}
+
+	if !*quiet {
+		if exportAll {
+			fmt.Fprintf(os.Stderr, "📤 Exporting all bundles...\n")
+		} else {
+			fmt.Fprintf(os.Stderr, "📤 Exporting bundles %d-%d...\n", bundleStart, bundleEnd)
+		}
+	}
+
+	stats, err := manager.Export(spec, opts)
+	if err != nil {
+		log.Fatalf("Export failed: %v", err)
+	}
+
+	if !*quiet {
+		fmt.Fprintf(os.Stderr, "\r   Exported: %d operations (%.2f MB)\n", stats.RecordsWritten, float64(stats.BytesWritten)/1024/1024)
+		fmt.Fprintf(os.Stderr, "✅ Export complete\n")
+	}
+}

bindings/go/export_bridge.c

···
+#include "plcbundle.h"
+#include <stddef.h>
+
+// This function will be called by Rust, and it will call the Go function
+// The Go function is exported via //export, so it's available as a C function
+int exportCallbackBridge(const char* data, size_t len, void* user_data) {
+    // The Go-exported function will be linked in
+    extern int exportCallbackGo(const char* data, size_t len, void* user_data);
+    return exportCallbackGo(data, len, user_data);
+}
+

bindings/go/plcbundle.go

···
 package plcbundle
 
 // #cgo LDFLAGS: -L../../target/release -lplcbundle -lm -ldl -lpthread
+// #cgo CFLAGS: -I.
 // #include <stdlib.h>
+// #include <stddef.h>
 // #include "plcbundle.h"
+// extern int exportCallbackBridge(const char* data, size_t len, void* user_data);
 import "C"
 import (
 	"fmt"
+	"io"
 	"runtime"
+	"strings"
+	"sync"
 	"unsafe"
 )
 
···
 	OperationsIndexed uint64
 	UniqueDIDs        uint
 	DurationMs        uint64
+}
+
+type ExportSpec struct {
+	BundleStart    uint32
+	BundleEnd      uint32
+	ExportAll      bool
+	Format         uint8  // 0=jsonl, 1=json, 2=csv
+	CountLimit     uint64 // 0 = no limit
+	AfterTimestamp string // Empty = no filter
+	DIDFilter      string // Empty = no filter
+	OpTypeFilter   string // Empty = no filter
+}
+
+type ExportStats struct {
+	RecordsWritten uint64
+	BytesWritten   uint64
 }
 
 type WarmUpStrategy uint8
···
 	}
 
 	return stats, nil
+}
+
+// ============================================================================
+// Export
+// ============================================================================
+
+// ExportOptions configures export behavior
+type ExportOptions struct {
+	// Writer receives exported data in batches
+	Writer io.Writer
+	// Progress callback (optional)
+	Progress func(records, bytes uint64)
+}
+
+// Export exports operations according to the spec
+func (m *BundleManager) Export(spec ExportSpec, opts *ExportOptions) (*ExportStats, error) {
+	if opts == nil || opts.Writer == nil {
+		return nil, fmt.Errorf("writer is required")
+	}
+
+	// Convert Go spec to C spec
+	cSpec := C.CExportSpec{
+		bundle_start:    C.uint32_t(spec.BundleStart),
+		bundle_end:      C.uint32_t(spec.BundleEnd),
+		export_all:      C._Bool(spec.ExportAll),
+		format:          C.uint8_t(spec.Format),
+		count_limit:     C.uint64_t(spec.CountLimit),
+		after_timestamp: nil,
+		did_filter:      nil,
+		op_type_filter:  nil,
+	}
+
+	// Convert string filters to C strings
+	var afterTs, didFilter, opTypeFilter *C.char
+	if spec.AfterTimestamp != "" {
+		afterTs = C.CString(spec.AfterTimestamp)
+		defer C.free(unsafe.Pointer(afterTs))
+		cSpec.after_timestamp = afterTs
+	}
+
+	if spec.DIDFilter != "" {
+		didFilter = C.CString(spec.DIDFilter)
+		defer C.free(unsafe.Pointer(didFilter))
+		cSpec.did_filter = didFilter
+	}
+
+	if spec.OpTypeFilter != "" {
+		opTypeFilter = C.CString(spec.OpTypeFilter)
+		defer C.free(unsafe.Pointer(opTypeFilter))
+		cSpec.op_type_filter = opTypeFilter
+	}
+
+	// Create callback context
+	ctx := &exportCallbackCtx{
+		writer:   opts.Writer,
+		progress: opts.Progress,
+	}
+
+	var cStats C.CExportStats
+	
+	// Use a global callback registry to work around cgo limitations
+	exportCallbackID := registerExportCallback(ctx)
+	defer unregisterExportCallback(exportCallbackID)
+
+	// Use C bridge function that calls the Go-exported function
+	result := C.bundle_manager_export(
+		m.ptr,
+		&cSpec,
+		(C.ExportCallback)(C.exportCallbackBridge),
+		unsafe.Pointer(uintptr(exportCallbackID)),
+		&cStats,
+	)
+
+	if result != 0 {
+		return nil, fmt.Errorf("export failed")
+	}
+
+	stats := &ExportStats{
+		RecordsWritten: uint64(cStats.records_written),
+		BytesWritten:   uint64(cStats.bytes_written),
+	}
+
+	return stats, nil
+}
+
+// ============================================================================
+// Export callback registry
+// ============================================================================
+
+type exportCallbackCtx struct {
+	writer   io.Writer
+	progress func(uint64, uint64)
+	records  uint64
+	bytes    uint64
+}
+
+var (
+	exportCallbacks   = make(map[uintptr]*exportCallbackCtx)
+	exportCallbackMux sync.Mutex
+	exportCallbackID  uintptr = 1
+)
+
+func registerExportCallback(ctx *exportCallbackCtx) uintptr {
+	exportCallbackMux.Lock()
+	defer exportCallbackMux.Unlock()
+	id := exportCallbackID
+	exportCallbackID++
+	exportCallbacks[id] = ctx
+	return id
+}
+
+func unregisterExportCallback(id uintptr) {
+	exportCallbackMux.Lock()
+	defer exportCallbackMux.Unlock()
+	delete(exportCallbacks, id)
+}
+
+//export exportCallbackGo
+func exportCallbackGo(data *C.char, len C.size_t, userData unsafe.Pointer) C.int {
+	id := uintptr(userData)
+	exportCallbackMux.Lock()
+	ctx, ok := exportCallbacks[id]
+	exportCallbackMux.Unlock()
+
+	if !ok || ctx == nil {
+		return 1
+	}
+
+	if data == nil || len == 0 {
+		return 0
+	}
+
+	// Copy C data to Go slice
+	goData := C.GoBytes(unsafe.Pointer(data), C.int(len))
+	_, err := ctx.writer.Write(goData)
+	if err != nil {
+		return 1 // Stop on error
+	}
+
+	ctx.bytes += uint64(len)
+	// Estimate records (rough - count newlines)
+	ctx.records += uint64(strings.Count(string(goData), "\n"))
+
+	// Call progress callback if provided
+	if ctx.progress != nil {
+		ctx.progress(ctx.records, ctx.bytes)
+	}
+
+	return 0 // Continue
 }
 
 // ============================================================================

bindings/go/plcbundle.h

···
     uint64_t duration_ms;
 } CRebuildStats;
 
+typedef struct {
+    uint32_t bundle_start;
+    uint32_t bundle_end;
+    bool export_all;
+    uint8_t format;  // 0=jsonl, 1=json, 2=csv
+    uint64_t count_limit;  // 0 = no limit
+    const char* after_timestamp;  // NULL = no filter
+    const char* did_filter;  // NULL = no filter
+    const char* op_type_filter;  // NULL = no filter
+} CExportSpec;
+
+typedef struct {
+    uint64_t records_written;
+    uint64_t bytes_written;
+} CExportStats;
+
 // Warm-up strategies
 #define WARMUP_RECENT 0
 #define WARMUP_ALL 1
···
 int bundle_manager_get_stats(
     CBundleManager manager,
     CManagerStats* out_stats
+);
+
+// ============================================================================
+// Export
+// ============================================================================
+
+/**
+ * Export operations with streaming callback.
+ * callback(data, len, user_data) is called for each batch of data.
+ * Returns 0 to continue, non-zero to stop.
+ * Returns 0 on success, -1 on error.
+ */
+typedef int (*ExportCallback)(const char* data, size_t len, void* user_data);
+
+int bundle_manager_export(
+    CBundleManager manager,
+    const CExportSpec* spec,
+    ExportCallback callback,
+    void* user_data,
+    CExportStats* out_stats  // NULL if stats not needed
 );
 
 // ============================================================================

src/bin/plcbundle-rs.rs

···
         #[arg(short = 'd', long, default_value = ".")]
         dir: PathBuf,
 
-        /// Bundle range
+        /// Bundle range (e.g., "1-100")
         #[arg(short, long)]
+        range: Option<String>,
+
+        /// Export all bundles
+        #[arg(long)]
+        all: bool,
+
+        /// Bundle range (legacy, use --range instead)
+        #[arg(long, hide = true)]
         bundles: Option<String>,
 
         /// Output format
         #[arg(short = 'f', long, default_value = "jsonl")]
         format: ExportFormat,
 
-        /// Output file
+        /// Output file (default: stdout)
         #[arg(short, long)]
-        output: PathBuf,
+        output: Option<PathBuf>,
+
+        /// Limit number of operations to export
+        #[arg(long)]
+        count: Option<usize>,
+
+        /// Export operations after this timestamp (ISO 8601 format)
+        #[arg(long)]
+        after: Option<String>,
 
         /// Filter by DID
         #[arg(long)]
···
         }
         Commands::Export {
             dir,
+            range,
+            all,
             bundles,
             format,
             output,
+            count,
+            after,
             did,
             op_type,
             compress,
         } => {
-            cmd_export(dir, bundles, format, output, did, op_type, compress, cli.quiet)?;
+            cmd_export(
+                dir, range, all, bundles, format, output, count, after,
+                did, op_type, compress, cli.quiet, cli.verbose
+            )?;
         }
         Commands::Stats {
             dir,
···
 }
 
 fn cmd_export(
-    _dir: PathBuf,
-    _bundles: Option<String>,
-    _format: ExportFormat,
-    _output: PathBuf,
-    _did: Option<String>,
-    _op_type: Option<String>,
+    dir: PathBuf,
+    range: Option<String>,
+    all: bool,
+    bundles: Option<String>, // Legacy flag
+    format: ExportFormat,
+    output: Option<PathBuf>,
+    count: Option<usize>,
+    after: Option<String>,
+    did: Option<String>,
+    op_type: Option<String>,
     _compress: bool,
-    _quiet: bool,
+    quiet: bool,
+    verbose: bool,
 ) -> Result<()> {
-    println!("Export not yet implemented");
+    use std::io::{BufWriter, Write};
+    use std::fs::File;
+    use std::io::BufRead;
+
+    // Load index to get max bundle and metadata
+    let index = plcbundle::Index::load(&dir)?;
+    let max_bundle = index.last_bundle;
+
+    // Determine bundle numbers to process
+    let bundle_numbers: Vec<u32> = if let Some(range_str) = range {
+        // Parse range like "1-100"
+        if range_str.contains('-') {
+            let parts: Vec<&str> = range_str.split('-').collect();
+            if parts.len() == 2 {
+                let start: u32 = parts[0].parse()?;
+                let end: u32 = parts[1].parse()?;
+                if start > end || start == 0 || end > max_bundle {
+                    anyhow::bail!("Invalid range: {}-{}", start, end);
+                }
+                (start..=end).collect()
+            } else {
+                anyhow::bail!("Invalid range format: {}", range_str);
+            }
+        } else {
+            // Single bundle number
+            let num: u32 = range_str.parse()?;
+            if num == 0 || num > max_bundle {
+                anyhow::bail!("Bundle number {} out of range", num);
+            }
+            vec![num]
+        }
+    } else if all {
+        (1..=max_bundle).collect()
+    } else if let Some(bundles_str) = bundles {
+        // Legacy --bundles flag
+        parse_bundle_spec(Some(bundles_str), max_bundle)?
+    } else {
+        anyhow::bail!("Must specify either --range, --all, or --bundles");
+    };
+
+    // Filter bundles by timestamp metadata if --after is specified
+    let bundle_numbers: Vec<u32> = if let Some(ref after_ts) = after {
+        bundle_numbers.into_iter()
+            .filter_map(|num| {
+                if let Some(meta) = index.get_bundle(num) {
+                    // Check if bundle's end_time is after the filter timestamp
+                    // If bundle ends before the filter, skip it
+                    if meta.end_time >= *after_ts {
+                        Some(num)
+                    } else {
+                        None
+                    }
+                } else {
+                    Some(num) // Include if metadata not found (will be checked during processing)
+                }
+            })
+            .collect()
+    } else {
+        bundle_numbers
+    };
+
+    if verbose && !quiet {
+        eprintln!("📦 Index: v{} ({})", index.version, index.origin);
+        eprintln!("📊 Processing {} bundles", bundle_numbers.len());
+        if let Some(ref count) = count {
+            eprintln!("🔢 Export limit: {} operations", format_number(*count as u64));
+        }
+        if let Some(ref after) = after {
+            eprintln!("⏰ After timestamp: {}", after);
+        }
+    }
+
+    // Open output with buffering
+    let writer: Box<dyn Write> = if let Some(output_path) = output {
+        Box::new(BufWriter::with_capacity(1024 * 1024, File::create(output_path)?))
+    } else {
+        Box::new(BufWriter::with_capacity(1024 * 1024, io::stdout()))
+    };
+    let mut writer = writer;
+
+    if !quiet {
+        eprintln!("📤 Exporting operations...");
+    }
+
+    let start = Instant::now();
+    let mut exported_count = 0;
+    let mut output_buffer = String::with_capacity(1024 * 1024); // 1MB buffer
+    const BATCH_SIZE: usize = 10000;
+
+    // Process bundles directly from files
+    for bundle_num in bundle_numbers {
+        // Check count limit
+        if let Some(limit) = count {
+            if exported_count >= limit {
+                break;
+            }
+        }
+
+        let bundle_path = dir.join(format!("{:06}.jsonl.zst", bundle_num));
+        if !bundle_path.exists() {
+            continue;
+        }
+
+        // Open and decode bundle file
+        let file = File::open(&bundle_path)?;
+        let decoder = zstd::Decoder::new(file)?;
+        let reader = std::io::BufReader::with_capacity(1024 * 1024, decoder);
+
+        // Fast path: no filters and Jsonl format - just pass through lines
+        let needs_parsing = after.is_some() || did.is_some() || op_type.is_some() || 
+                           matches!(format, ExportFormat::Json | ExportFormat::Csv | ExportFormat::Parquet);
+
+        if !needs_parsing {
+            // Fast path: no parsing needed, just copy lines
+            for line in reader.lines() {
+                // Check count limit
+                if let Some(limit) = count {
+                    if exported_count >= limit {
+                        break;
+                    }
+                }
+
+                let line = line?;
+                if line.is_empty() {
+                    continue;
+                }
+
+                output_buffer.push_str(&line);
+                output_buffer.push('\n');
+                exported_count += 1;
+
+                // Flush buffer when it gets large
+                if output_buffer.len() >= 1024 * 1024 {
+                    writer.write_all(output_buffer.as_bytes())?;
+                    output_buffer.clear();
+                }
+
+                // Progress update
+                if !quiet && exported_count % BATCH_SIZE == 0 {
+                    eprint!("\r   Exported: {} operations", format_number(exported_count as u64));
+                    io::stderr().flush()?;
+                }
+            }
+        } else {
+            // Slow path: need to parse for filtering or formatting
+            use sonic_rs::JsonValueTrait;
+
+            for line in reader.lines() {
+                // Check count limit
+                if let Some(limit) = count {
+                    if exported_count >= limit {
+                        break;
+                    }
+                }
+
+                let line = line?;
+                if line.is_empty() {
+                    continue;
+                }
+
+                // Parse JSON using sonic-rs for faster parsing
+                let data: sonic_rs::Value = match sonic_rs::from_str(&line) {
+                    Ok(data) => data,
+                    Err(_) => continue, // Skip invalid JSON
+                };
+
+                // Apply filters using sonic-rs Value API
+                if let Some(ref after_ts) = after {
+                    if let Some(created_at) = data.get("createdAt").or_else(|| data.get("created_at")) {
+                        if let Some(ts_str) = created_at.as_str() {
+                            if ts_str < after_ts.as_str() {
+                                continue;
+                            }
+                        } else {
+                            continue;
+                        }
+                    } else {
+                        continue;
+                    }
+                }
+
+                if let Some(ref did_filter) = did {
+                    if let Some(did_val) = data.get("did") {
+                        if let Some(did_str) = did_val.as_str() {
+                            if did_str != did_filter {
+                                continue;
+                            }
+                        } else {
+                            continue;
+                        }
+                    } else {
+                        continue;
+                    }
+                }
+
+                if let Some(ref op_type_filter) = op_type {
+                    if let Some(op_val) = data.get("operation") {
+                        let matches = if op_val.is_str() {
+                            op_val.as_str().unwrap() == op_type_filter
+                        } else if op_val.is_object() {
+                            if let Some(typ_val) = op_val.get("type") {
+                                typ_val.is_str() && typ_val.as_str().unwrap() == op_type_filter
+                            } else {
+                                false
+                            }
+                        } else {
+                            false
+                        };
+                        if !matches {
+                            continue;
+                        }
+                    } else {
+                        continue;
+                    }
+                }
+
+                // Format operation
+                let formatted = match format {
+                    ExportFormat::Jsonl => {
+                        // Already have the JSON string, use it directly
+                        line
+                    }
+                    ExportFormat::Json => {
+                        // Pretty print using sonic-rs
+                        sonic_rs::to_string_pretty(&data)?
+                    }
+                    ExportFormat::Csv => {
+                        let did = data.get("did").and_then(|v| v.as_str()).unwrap_or("");
+                        let op = data.get("operation").map(|v| sonic_rs::to_string(v).unwrap_or_default()).unwrap_or_default();
+                        let created_at = data.get("createdAt").or_else(|| data.get("created_at"))
+                            .and_then(|v| v.as_str()).unwrap_or("");
+                        let nullified = data.get("nullified").and_then(|v| v.as_bool()).unwrap_or(false);
+                        format!("{},{},{},{}", did, op, created_at, nullified)
+                    }
+                    ExportFormat::Parquet => {
+                        // Fall back to JSON for now
+                        sonic_rs::to_string(&data)?
+                    }
+                };
+
+                output_buffer.push_str(&formatted);
+                output_buffer.push('\n');
+                exported_count += 1;
+
+                // Flush buffer when it gets large
+                if output_buffer.len() >= 1024 * 1024 {
+                    writer.write_all(output_buffer.as_bytes())?;
+                    output_buffer.clear();
+                }
+
+                // Progress update
+                if !quiet && exported_count % BATCH_SIZE == 0 {
+                    eprint!("\r   Exported: {} operations", format_number(exported_count as u64));
+                    io::stderr().flush()?;
+                }
+            }
+        }
+    }
+
+    // Flush remaining buffer
+    if !output_buffer.is_empty() {
+        writer.write_all(output_buffer.as_bytes())?;
+    }
+    writer.flush()?;
+
+    if !quiet {
+        eprintln!("\r   Exported: {} operations", format_number(exported_count as u64));
+        let elapsed = start.elapsed();
+        eprintln!("✅ Complete in {}", HumanDuration(elapsed));
+        if elapsed.as_secs_f64() > 0.0 {
+            eprintln!("   Throughput: {:.0} ops/sec",
+                exported_count as f64 / elapsed.as_secs_f64()
+            );
+        }
+    }
+
     Ok(())
 }
 

src/ffi.rs

···
     pub duration_ms: u64,
 }
 
+#[repr(C)]
+pub struct CExportSpec {
+    pub bundle_start: u32,
+    pub bundle_end: u32,
+    pub export_all: bool,
+    pub format: u8, // 0=jsonl, 1=json, 2=csv
+    pub count_limit: u64, // 0 = no limit
+    pub after_timestamp: *const c_char, // NULL = no filter
+    pub did_filter: *const c_char, // NULL = no filter
+    pub op_type_filter: *const c_char, // NULL = no filter
+}
+
+#[repr(C)]
+pub struct CExportStats {
+    pub records_written: u64,
+    pub bytes_written: u64,
+}
+
 // ============================================================================
 // BundleManager lifecycle
 // ============================================================================
···
             let _ = Vec::from_raw_parts(ops, count, count);
         }
     }
+}
+
+// ============================================================================
+// Export
+// ============================================================================
+
+/// Callback type for streaming export
+/// Returns 0 to continue, non-zero to stop
+pub type ExportCallback = extern "C" fn(data: *const c_char, len: usize, user_data: *mut std::ffi::c_void) -> i32;
+
+#[no_mangle]
+pub extern "C" fn bundle_manager_export(
+    manager: *const CBundleManager,
+    spec: *const CExportSpec,
+    callback: ExportCallback,
+    user_data: *mut std::ffi::c_void,
+    out_stats: *mut CExportStats,
+) -> i32 {
+    if manager.is_null() || spec.is_null() || callback as usize == 0 {
+        return -1;
+    }
+
+    let manager = unsafe { &*manager };
+    let spec = unsafe { &*spec };
+
+    use std::io::{BufRead, BufWriter, Write};
+    use std::fs::File;
+    use crate::index::Index;
+
+    // Get directory path
+    let dir_path = manager.manager.directory().clone();
+    // Load index
+    let index = match Index::load(&dir_path) {
+        Ok(idx) => idx,
+        Err(_) => return -1,
+    };
+
+    // Determine bundle range
+    let bundle_numbers: Vec<u32> = if spec.export_all {
+        (1..=index.last_bundle).collect()
+    } else if spec.bundle_end > 0 && spec.bundle_end >= spec.bundle_start {
+        (spec.bundle_start..=spec.bundle_end).collect()
+    } else {
+        vec![spec.bundle_start]
+    };
+
+    // Filter bundles by timestamp if specified
+    let bundle_numbers: Vec<u32> = if !spec.after_timestamp.is_null() {
+        let after_ts = unsafe {
+            match CStr::from_ptr(spec.after_timestamp).to_str() {
+                Ok(s) => s.to_string(),
+                Err(_) => return -1,
+            }
+        };
+        bundle_numbers.into_iter()
+            .filter_map(|num| {
+                if let Some(meta) = index.get_bundle(num) {
+                    if meta.end_time >= after_ts {
+                        Some(num)
+                    } else {
+                        None
+                    }
+                } else {
+                    Some(num)
+                }
+            })
+            .collect()
+    } else {
+        bundle_numbers
+    };
+
+    let mut exported_count = 0u64;
+    let mut bytes_written = 0u64;
+    let mut output_buffer = Vec::with_capacity(1024 * 1024);
+
+    // Determine if we need parsing
+    let needs_parsing = !spec.after_timestamp.is_null() || 
+                       !spec.did_filter.is_null() || 
+                       !spec.op_type_filter.is_null() ||
+                       spec.format == 1 || spec.format == 2; // JSON or CSV
+
+    // Process bundles
+    for bundle_num in bundle_numbers {
+        if spec.count_limit > 0 && exported_count >= spec.count_limit {
+            break;
+        }
+
+        let bundle_path = dir_path.join(format!("{:06}.jsonl.zst", bundle_num));
+        if !bundle_path.exists() {
+            continue;
+        }
+
+        let file = match File::open(&bundle_path) {
+            Ok(f) => f,
+            Err(_) => continue,
+        };
+
+        let decoder = match zstd::Decoder::new(file) {
+            Ok(d) => d,
+            Err(_) => continue,
+        };
+
+        let reader = std::io::BufReader::with_capacity(1024 * 1024, decoder);
+
+        if !needs_parsing {
+            // Fast path: no parsing
+            for line in reader.lines() {
+                if spec.count_limit > 0 && exported_count >= spec.count_limit {
+                    break;
+                }
+
+                let line = match line {
+                    Ok(l) => l,
+                    Err(_) => continue,
+                };
+
+                if line.is_empty() {
+                    continue;
+                }
+
+                output_buffer.extend_from_slice(line.as_bytes());
+                output_buffer.push(b'\n');
+                exported_count += 1;
+                bytes_written += line.len() as u64 + 1;
+
+                // Flush buffer when large
+                if output_buffer.len() >= 1024 * 1024 {
+                    let result = callback(output_buffer.as_ptr() as *const c_char, output_buffer.len(), user_data);
+                    if result != 0 {
+                        break;
+                    }
+                    output_buffer.clear();
+                }
+            }
+        } else {
+            // Slow path: parse and filter
+            use sonic_rs::JsonValueTrait;
+
+            let after_ts = if !spec.after_timestamp.is_null() {
+                Some(unsafe {
+                    CStr::from_ptr(spec.after_timestamp).to_str().unwrap().to_string()
+                })
+            } else {
+                None
+            };
+
+            let did_filter = if !spec.did_filter.is_null() {
+                Some(unsafe {
+                    CStr::from_ptr(spec.did_filter).to_str().unwrap().to_string()
+                })
+            } else {
+                None
+            };
+
+            let op_type_filter = if !spec.op_type_filter.is_null() {
+                Some(unsafe {
+                    CStr::from_ptr(spec.op_type_filter).to_str().unwrap().to_string()
+                })
+            } else {
+                None
+            };
+
+            for line in reader.lines() {
+                if spec.count_limit > 0 && exported_count >= spec.count_limit {
+                    break;
+                }
+
+                let line = match line {
+                    Ok(l) => l,
+                    Err(_) => continue,
+                };
+
+                if line.is_empty() {
+                    continue;
+                }
+
+                // Parse JSON
+                let data: sonic_rs::Value = match sonic_rs::from_str(&line) {
+                    Ok(d) => d,
+                    Err(_) => continue,
+                };
+
+                // Apply filters
+                if let Some(ref after_ts) = after_ts {
+                    if let Some(created_at) = data.get("createdAt").or_else(|| data.get("created_at")) {
+                        if let Some(ts_str) = created_at.as_str() {
+                            if ts_str < after_ts.as_str() {
+                                continue;
+                            }
+                        } else {
+                            continue;
+                        }
+                    } else {
+                        continue;
+                    }
+                }
+
+                if let Some(ref did_filter) = did_filter {
+                    if let Some(did_val) = data.get("did") {
+                        if let Some(did_str) = did_val.as_str() {
+                            if did_str != did_filter {
+                                continue;
+                            }
+                        } else {
+                            continue;
+                        }
+                    } else {
+                        continue;
+                    }
+                }
+
+                if let Some(ref op_type_filter) = op_type_filter {
+                    if let Some(op_val) = data.get("operation") {
+                        let matches = if op_val.is_str() {
+                            op_val.as_str().unwrap() == op_type_filter
+                        } else if op_val.is_object() {
+                            if let Some(typ_val) = op_val.get("type") {
+                                typ_val.is_str() && typ_val.as_str().unwrap() == op_type_filter
+                            } else {
+                                false
+                            }
+                        } else {
+                            false
+                        };
+                        if !matches {
+                            continue;
+                        }
+                    } else {
+                        continue;
+                    }
+                }
+
+                // Format
+                let formatted = match spec.format {
+                    0 => line, // JSONL - use original
+                    1 => sonic_rs::to_string_pretty(&data).unwrap_or_default(), // JSON
+                    2 => {
+                        let did = data.get("did").and_then(|v| v.as_str()).unwrap_or("");
+                        let op = data.get("operation").map(|v| sonic_rs::to_string(v).unwrap_or_default()).unwrap_or_default();
+                        let created_at = data.get("createdAt").or_else(|| data.get("created_at"))
+                            .and_then(|v| v.as_str()).unwrap_or("");
+                        let nullified = data.get("nullified").and_then(|v| v.as_bool()).unwrap_or(false);
+                        format!("{},{},{},{}", did, op, created_at, nullified)
+                    }
+                    _ => line,
+                };
+
+                output_buffer.extend_from_slice(formatted.as_bytes());
+                output_buffer.push(b'\n');
+                exported_count += 1;
+                bytes_written += formatted.len() as u64 + 1;
+
+                // Flush buffer when large
+                if output_buffer.len() >= 1024 * 1024 {
+                    let result = callback(output_buffer.as_ptr() as *const c_char, output_buffer.len(), user_data);
+                    if result != 0 {
+                        break;
+                    }
+                    output_buffer.clear();
+                }
+            }
+        }
+    }
+
+    // Flush remaining buffer
+    if !output_buffer.is_empty() {
+        callback(output_buffer.as_ptr() as *const c_char, output_buffer.len(), user_data);
+    }
+
+    if !out_stats.is_null() {
+        unsafe {
+            (*out_stats).records_written = exported_count;
+            (*out_stats).bytes_written = bytes_written;
+        }
+    }
+
+    0
 }

src/iterators.rs

···
 pub struct ExportIterator {
     manager: Arc<BundleManager>,
     spec: ExportSpec,
-    // TODO: Implement export iterator
+    bundle_numbers: Vec<u32>,
+    current_bundle_idx: usize,
+    current_ops: Vec<Operation>,
+    current_op_idx: usize,
+    count_remaining: Option<usize>,
 }
 
 impl ExportIterator {
     pub fn new(manager: Arc<BundleManager>, spec: ExportSpec) -> Self {
-        Self { manager, spec }
+        let bundle_numbers = match &spec.bundles {
+            crate::manager::BundleRange::All => {
+                let last = manager.get_last_bundle();
+                (1..=last).collect()
+            }
+            crate::manager::BundleRange::Single(n) => vec![*n],
+            crate::manager::BundleRange::Range(start, end) => (*start..=*end).collect(),
+            crate::manager::BundleRange::List(list) => list.clone(),
+        };
+
+        Self {
+            manager,
+            spec,
+            bundle_numbers,
+            current_bundle_idx: 0,
+            current_ops: Vec::new(),
+            current_op_idx: 0,
+            count_remaining: None,
+        }
+    }
+
+    fn format_operation(&self, op: &Operation) -> Result<String> {
+        match self.spec.format {
+            crate::manager::ExportFormat::JsonLines => {
+                Ok(serde_json::to_string(op)?)
+            }
+            crate::manager::ExportFormat::Csv => {
+                // Simple CSV format - could be enhanced
+                Ok(format!(
+                    "{},{},{},{}",
+                    op.did,
+                    op.operation,
+                    op.created_at,
+                    op.nullified
+                ))
+            }
+            crate::manager::ExportFormat::Parquet => {
+                // Parquet would require additional dependencies
+                // For now, fall back to JSON
+                Ok(serde_json::to_string(op)?)
+            }
+        }
+    }
+
+    fn matches_timestamp_filter(&self, op: &Operation) -> bool {
+        if let Some(ref after) = self.spec.after_timestamp {
+            // Compare timestamps as strings (ISO 8601 format)
+            op.created_at >= *after
+        } else {
+            true
+        }
     }
 }
 
···
     type Item = Result<String>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        // TODO: Implement
-        None
+        // Check count limit
+        if let Some(ref mut remaining) = self.count_remaining {
+            if *remaining == 0 {
+                return None;
+            }
+        }
+
+        loop {
+            // Check if we have more operations in current bundle
+            if self.current_op_idx < self.current_ops.len() {
+                let op = &self.current_ops[self.current_op_idx];
+                self.current_op_idx += 1;
+
+                // Apply timestamp filter
+                if !self.matches_timestamp_filter(op) {
+                    continue;
+                }
+
+                // Decrement count if limit is set
+                if let Some(ref mut remaining) = self.count_remaining {
+                    *remaining -= 1;
+                }
+
+                return Some(self.format_operation(op));
+            }
+
+            // Load next bundle
+            if self.current_bundle_idx >= self.bundle_numbers.len() {
+                return None;
+            }
+
+            let bundle_num = self.bundle_numbers[self.current_bundle_idx];
+            self.current_bundle_idx += 1;
+
+            // Initialize count_remaining on first bundle load
+            if self.count_remaining.is_none() {
+                self.count_remaining = self.spec.count;
+            }
+
+            match self.manager.load_bundle(
+                bundle_num,
+                LoadOptions {
+                    filter: self.spec.filter.clone(),
+                    ..Default::default()
+                },
+            ) {
+                Ok(result) => {
+                    self.current_ops = result.operations;
+                    self.current_op_idx = 0;
+                }
+                Err(_e) => {
+                    // Skip bundles that fail to load
+                    continue;
+                }
+            }
+        }
     }
 }

src/lib.rs

···
 pub use query::QueryEngine;
 pub use index::{Index, BundleMetadata};
 pub use manager::{BundleManager, LoadOptions, LoadResult, QuerySpec, ExportSpec, 
+                  BundleRange, ExportFormat, CompressionType,
                   VerifySpec, VerifyResult, ChainVerifySpec, ChainVerifyResult,
                   BundleInfo, InfoFlags, RollbackSpec, RollbackPlan, RollbackResult,
                   WarmUpSpec, WarmUpStrategy, RebuildStats, DIDIndexStats, ManagerStats};

src/manager.rs

···
         self.index.read().unwrap().last_bundle
     }
 
+    pub fn directory(&self) -> &PathBuf {
+        &self.directory
+    }
+
     fn clone_for_arc(&self) -> Self {
         Self {
             directory: self.directory.clone(),
···
     pub format: ExportFormat,
     pub filter: Option<OperationFilter>,
     pub compression: Option<CompressionType>,
+    pub count: Option<usize>,
+    pub after_timestamp: Option<String>,
 }
 
 #[derive(Debug, Clone)]