cmd export

cmd/plcbundle/commands/export.go

···
 	"os"
 	"time"
 
-	flag "github.com/spf13/pflag"
-
 	"github.com/goccy/go-json"
+	"github.com/spf13/cobra"
+	internalsync "tangled.org/atscan.net/plcbundle/internal/sync"
 )
 
-// ExportCommand handles the export subcommand
-func ExportCommand(args []string) error {
-	fs := flag.NewFlagSet("export", flag.ExitOnError)
-	bundles := fs.String("bundles", "", "bundle number or range (e.g., '42' or '1-100')")
-	all := fs.Bool("all", false, "export all bundles")
-	count := fs.Int("count", 0, "limit number of operations (0 = all)")
-	after := fs.String("after", "", "timestamp to start after (RFC3339)")
+func NewExportCommand() *cobra.Command {
+	var (
+		all      bool
+		rangeStr string
+		sync     bool
+		plcURL   string
+		count    int
+		after    string
+	)
 
-	if err := fs.Parse(args); err != nil {
-		return err
-	}
+	cmd := &cobra.Command{
+		Use:     "export [flags]",
+		Aliases: []string{"stream", "backfill"},
+		Short:   "Export operations to stdout (JSONL)",
+		Long: `Export operations to stdout in JSONL format
 
-	if !*all && *bundles == "" {
-		fmt.Fprint(os.Stderr, `usage: plcbundle export --bundles <number|range> [options]
-   or: plcbundle export --all [options]
+Outputs PLC operations as newline-delimited JSON to stdout.
+Progress and status messages go to stderr.
+
+By default, exports only existing bundles. Use --sync to also fetch
+new bundles from PLC directory until caught up.
+
+Supports filtering by count and timestamp for selective exports.`,
 
-Examples:
-  plcbundle export --bundles 42
-  plcbundle export --bundles 1-100
+		Example: `  # Export all existing bundles
   plcbundle export --all
+
+  # Export and sync new bundles
+  plcbundle export --all --sync
+
+  # Export specific range (existing only)
+  plcbundle export --range 1-100
+
+  # Export with limit
   plcbundle export --all --count 50000
-  plcbundle export --bundles 42 | jq .
+
+  # Export after timestamp
+  plcbundle export --all --after 2024-01-01T00:00:00Z
+
+  # Combine filters
+  plcbundle export --range 1-100 --count 10000 --after 2024-01-01T00:00:00Z
+
+  # Export from specific PLC directory
+  plcbundle export --all --sync --plc https://plc.directory
+
+  # Pipe to file
+  plcbundle export --all > operations.jsonl
+
+  # Process with jq
+  plcbundle export --all | jq -r .did | sort | uniq -c
+
+  # Sync and filter with detector
+  plcbundle export --all --sync | plcbundle detector filter spam
+
+  # Using aliases (backwards compatible)
+  plcbundle stream --all --sync
+  plcbundle backfill --all`,
+
+		RunE: func(cmd *cobra.Command, args []string) error {
+			verbose, _ := cmd.Root().PersistentFlags().GetBool("verbose")
+			quiet, _ := cmd.Root().PersistentFlags().GetBool("quiet")
+
+			mgr, dir, err := getManager(&ManagerOptions{Cmd: cmd, PLCURL: plcURL})
+			if err != nil {
+				return err
+			}
+			defer mgr.Close()
+
+			if !quiet {
+				fmt.Fprintf(os.Stderr, "Exporting from: %s\n", dir)
+			}
+
+			// Parse after timestamp if provided
+			var afterTime time.Time
+			if after != "" {
+				afterTime, err = time.Parse(time.RFC3339, after)
+				if err != nil {
+					return fmt.Errorf("invalid --after timestamp (use RFC3339 format): %w", err)
+				}
+			}
+
+			// Determine bundle range
+			var start, end int
+
+			if all {
+				index := mgr.GetIndex()
+				bundles := index.GetBundles()
+
+				if len(bundles) == 0 {
+					if sync {
+						// No bundles but sync enabled - start from 1
+						start = 1
+						end = 0 // Will be updated after sync
+					} else {
+						if !quiet {
+							fmt.Fprintf(os.Stderr, "No bundles available (use --sync to fetch)\n")
+						}
+						return nil
+					}
+				} else {
+					start = bundles[0].BundleNumber
+					end = bundles[len(bundles)-1].BundleNumber
+				}
+
+			} else if rangeStr != "" {
+				var err error
+				start, end, err = parseBundleRange(rangeStr)
+				if err != nil {
+					return err
+				}
+
+			} else {
+				return fmt.Errorf("either --all or --range required")
+			}
+
+			if !quiet {
+				if sync {
+					fmt.Fprintf(os.Stderr, "Mode: export existing + sync new bundles\n")
+				} else {
+					fmt.Fprintf(os.Stderr, "Mode: export existing only\n")
+				}
+
+				if count > 0 {
+					fmt.Fprintf(os.Stderr, "Limit: %d operations\n", count)
+				}
+				if after != "" {
+					fmt.Fprintf(os.Stderr, "After: %s\n", after)
+				}
+				fmt.Fprintf(os.Stderr, "\n")
+			}
+
+			return exportBundles(cmd.Context(), mgr, exportOptions{
+				start:     start,
+				end:       end,
+				sync:      sync,
+				count:     count,
+				afterTime: afterTime,
+				verbose:   verbose,
+				quiet:     quiet,
+			})
+		},
+	}
+
+	cmd.Flags().BoolVar(&all, "all", false, "Export all bundles")
+	cmd.Flags().StringVarP(&rangeStr, "range", "r", "", "Export bundle range (e.g., '1-100')")
+	cmd.Flags().BoolVar(&sync, "sync", false, "Also fetch new bundles from PLC (until caught up)")
+	cmd.Flags().StringVar(&plcURL, "plc", "https://plc.directory", "PLC directory URL (for --sync)")
+	cmd.Flags().IntVarP(&count, "count", "n", 0, "Limit number of operations (0 = unlimited)")
+	cmd.Flags().StringVar(&after, "after", "", "Only export operations after this timestamp (RFC3339)")
+
+	return cmd
+}
+
+type exportOptions struct {
+	start     int
+	end       int
+	sync      bool
+	count     int
+	afterTime time.Time
+	verbose   bool
+	quiet     bool
+}
+
+func exportBundles(ctx context.Context, mgr BundleManager, opts exportOptions) error {
+	operationCount := 0
+	exported := 0
 
-`)
-		return fmt.Errorf("missing required flag: --bundles or --all")
+	// Phase 1: Export existing bundles
+	existingCount := 0
+	if opts.end > 0 {
+		existingCount, exported = exportExistingBundles(
+			ctx, mgr, opts.start, opts.end,
+			&operationCount, opts.count, opts.afterTime,
+			opts.verbose, opts.quiet,
+		)
 	}
 
-	mgr, _, err := getManager(&ManagerOptions{Cmd: nil})
-	if err != nil {
-		return err
+	// Check if we hit the count limit
+	if opts.count > 0 && exported >= opts.count {
+		if !opts.quiet {
+			fmt.Fprintf(os.Stderr, "\n✓ Export complete (limit reached)\n")
+			fmt.Fprintf(os.Stderr, "  Bundles: %d\n", existingCount)
+			fmt.Fprintf(os.Stderr, "  Operations: %d\n", exported)
+		}
+		return nil
 	}
-	defer mgr.Close()
 
-	// Determine bundle range
-	var start, end int
-	if *all {
-		index := mgr.GetIndex()
-		bundleList := index.GetBundles()
-		if len(bundleList) == 0 {
-			return fmt.Errorf("no bundles available")
+	// Phase 2: Sync and export new bundles (if enabled and not at limit)
+	fetchedCount := 0
+	if opts.sync && (opts.count == 0 || exported < opts.count) {
+		if !opts.quiet {
+			fmt.Fprintf(os.Stderr, "\nSyncing new bundles from PLC...\n")
 		}
-		start = bundleList[0].BundleNumber
-		end = bundleList[len(bundleList)-1].BundleNumber
 
-		fmt.Fprintf(os.Stderr, "Exporting all bundles (%d-%d)\n", start, end)
-	} else {
+		logger := &exportLogger{quiet: opts.quiet}
+		config := &internalsync.SyncLoopConfig{
+			MaxBundles: 0,
+			Verbose:    opts.verbose,
+			Logger:     logger,
+			OnBundleSynced: func(bundleNum, synced, mempoolCount int, duration, indexTime time.Duration) {
+				// Stream the bundle we just created
+				if bundle, err := mgr.LoadBundle(ctx, bundleNum); err == nil {
+					for _, op := range bundle.Operations {
+						// Apply filters
+						if !opts.afterTime.IsZero() && op.CreatedAt.Before(opts.afterTime) {
+							continue
+						}
+
+						if opts.count > 0 && exported >= opts.count {
+							return // Stop when limit reached
+						}
+
+						// Output operation
+						if len(op.RawJSON) > 0 {
+							fmt.Println(string(op.RawJSON))
+						} else {
+							data, _ := json.Marshal(op)
+							fmt.Println(string(data))
+						}
+						exported++
+						operationCount++
+					}
+				}
+			},
+		}
+
 		var err error
-		start, end, err = parseBundleRange(*bundles)
+		fetchedCount, err = mgr.RunSyncOnce(ctx, config, opts.verbose)
 		if err != nil {
 			return err
 		}
-		fmt.Fprintf(os.Stderr, "Exporting bundles %d-%d\n", start, end)
 	}
 
-	if *count > 0 {
-		fmt.Fprintf(os.Stderr, "Limit: %d operations\n", *count)
-	}
-	if *after != "" {
-		fmt.Fprintf(os.Stderr, "After: %s\n", *after)
-	}
-	fmt.Fprintf(os.Stderr, "\n")
+	// Summary
+	if !opts.quiet {
+		fmt.Fprintf(os.Stderr, "\n✓ Export complete\n")
 
-	// Parse after time
-	var afterTime time.Time
-	if *after != "" {
-		afterTime, err = time.Parse(time.RFC3339, *after)
-		if err != nil {
-			return fmt.Errorf("invalid after time: %w", err)
+		if opts.sync && fetchedCount > 0 {
+			fmt.Fprintf(os.Stderr, "  Bundles: %d (%d existing + %d synced)\n",
+				existingCount+fetchedCount, existingCount, fetchedCount)
+		} else if opts.sync {
+			fmt.Fprintf(os.Stderr, "  Bundles: %d (already up to date)\n", existingCount)
+		} else {
+			fmt.Fprintf(os.Stderr, "  Bundles: %d\n", existingCount)
+		}
+
+		fmt.Fprintf(os.Stderr, "  Operations: %d", exported)
+		if opts.count > 0 {
+			fmt.Fprintf(os.Stderr, " (limit: %d)", opts.count)
 		}
+		fmt.Fprintf(os.Stderr, "\n")
 	}
 
-	ctx := context.Background()
-	exported := 0
+	return nil
+}
+
+func exportExistingBundles(
+	ctx context.Context,
+	mgr BundleManager,
+	start, end int,
+	operationCount *int,
+	limit int,
+	afterTime time.Time,
+	verbose bool,
+	quiet bool,
+) (bundleCount int, exported int) {
+
+	processedCount := 0
+	exportedOps := 0
 
-	// Export operations
 	for bundleNum := start; bundleNum <= end; bundleNum++ {
-		if *count > 0 && exported >= *count {
-			break
+		select {
+		case <-ctx.Done():
+			return processedCount, exportedOps
+		default:
 		}
-
-		fmt.Fprintf(os.Stderr, "Processing bundle %d...\r", bundleNum)
 
 		bundle, err := mgr.LoadBundle(ctx, bundleNum)
 		if err != nil {
-			fmt.Fprintf(os.Stderr, "\nWarning: failed to load bundle %d: %v\n", bundleNum, err)
+			if verbose {
+				fmt.Fprintf(os.Stderr, "Bundle %06d: not found (skipped)\n", bundleNum)
+			}
 			continue
 		}
 
+		// Export operations with filters
 		for _, op := range bundle.Operations {
+			// Filter by timestamp
 			if !afterTime.IsZero() && op.CreatedAt.Before(afterTime) {
 				continue
 			}
 
-			if *count > 0 && exported >= *count {
-				break
+			// Check count limit
+			if limit > 0 && exportedOps >= limit {
+				if verbose {
+					fmt.Fprintf(os.Stderr, "Bundle %06d: limit reached, stopping\n", bundleNum)
+				}
+				return processedCount, exportedOps
 			}
 
-			// Output as JSONL
+			// Output operation to stdout (JSONL)
 			if len(op.RawJSON) > 0 {
 				fmt.Println(string(op.RawJSON))
 			} else {
 				data, _ := json.Marshal(op)
 				fmt.Println(string(data))
 			}
+			exportedOps++
+		}
 
-			exported++
+		*operationCount += len(bundle.Operations)
+		processedCount++
+
+		if verbose {
+			fmt.Fprintf(os.Stderr, "Bundle %06d: ✓ (%d ops, %d exported)\n",
+				bundleNum, len(bundle.Operations), exportedOps)
+		} else if !quiet && processedCount%100 == 0 {
+			fmt.Fprintf(os.Stderr, "Exported: %d bundles, %d ops\r", processedCount, exportedOps)
 		}
 	}
 
-	fmt.Fprintf(os.Stderr, "\n\n✓ Export complete\n")
-	fmt.Fprintf(os.Stderr, "  Exported: %d operations\n", exported)
+	if !quiet && !verbose && processedCount > 0 {
+		fmt.Fprintf(os.Stderr, "Existing: %d bundles, %d ops\n", processedCount, exportedOps)
+	}
+
+	return processedCount, exportedOps
+}
 
-	return nil
+type exportLogger struct {
+	quiet bool
+}
+
+func (l *exportLogger) Printf(format string, v ...interface{}) {
+	if !l.quiet {
+		fmt.Fprintf(os.Stderr, format+"\n", v...)
+	}
+}
+
+func (l *exportLogger) Println(v ...interface{}) {
+	if !l.quiet {
+		fmt.Fprintln(os.Stderr, v...)
+	}
 }

cmd/plcbundle/commands/stream.go

···
-package commands
-
-import (
-	"context"
-	"fmt"
-	"os"
-	"time"
-
-	"github.com/goccy/go-json"
-	"github.com/spf13/cobra"
-	internalsync "tangled.org/atscan.net/plcbundle/internal/sync"
-)
-
-func NewStreamCommand() *cobra.Command {
-	var (
-		all      bool
-		rangeStr string
-		sync     bool
-		plcURL   string
-	)
-
-	cmd := &cobra.Command{
-		Use:     "stream [flags]",
-		Aliases: []string{"backfill"},
-		Short:   "Stream operations to stdout (JSONL)",
-		Long: `Stream operations to stdout in JSONL format
-
-Outputs PLC operations as newline-delimited JSON to stdout.
-Progress and status messages go to stderr.
-
-By default, streams only existing bundles. Use --sync to also fetch
-new bundles from PLC directory until caught up.`,
-
-		Example: `  # Stream all existing bundles
-  plcbundle stream --all
-
-  # Stream and sync new bundles
-  plcbundle stream --all --sync
-
-  # Stream specific range (existing only)
-  plcbundle stream --range 1-100
-
-  # Stream from specific PLC directory
-  plcbundle stream --all --sync --plc https://plc.directory
-
-  # Pipe to file
-  plcbundle stream --all > operations.jsonl
-
-  # Process with jq
-  plcbundle stream --all | jq -r .did | sort | uniq -c
-
-  # Sync and filter with detector
-  plcbundle stream --all --sync | plcbundle detector filter spam`,
-
-		RunE: func(cmd *cobra.Command, args []string) error {
-			verbose, _ := cmd.Root().PersistentFlags().GetBool("verbose")
-			quiet, _ := cmd.Root().PersistentFlags().GetBool("quiet")
-
-			mgr, dir, err := getManager(&ManagerOptions{Cmd: cmd, PLCURL: plcURL})
-			if err != nil {
-				return err
-			}
-			defer mgr.Close()
-
-			if !quiet {
-				fmt.Fprintf(os.Stderr, "Streaming from: %s\n", dir)
-			}
-
-			// Determine bundle range
-			var start, end int
-
-			if all {
-				index := mgr.GetIndex()
-				bundles := index.GetBundles()
-
-				if len(bundles) == 0 {
-					if sync {
-						// No bundles but sync enabled - start from 1
-						start = 1
-						end = 0 // Will be updated after sync
-					} else {
-						if !quiet {
-							fmt.Fprintf(os.Stderr, "No bundles available (use --sync to fetch)\n")
-						}
-						return nil
-					}
-				} else {
-					start = bundles[0].BundleNumber
-					end = bundles[len(bundles)-1].BundleNumber
-				}
-
-			} else if rangeStr != "" {
-				var err error
-				start, end, err = parseBundleRange(rangeStr)
-				if err != nil {
-					return err
-				}
-
-			} else {
-				return fmt.Errorf("either --all or --range required")
-			}
-
-			if !quiet {
-				if sync {
-					fmt.Fprintf(os.Stderr, "Mode: stream existing + sync new bundles\n")
-				} else {
-					fmt.Fprintf(os.Stderr, "Mode: stream existing only\n")
-				}
-				fmt.Fprintf(os.Stderr, "\n")
-			}
-
-			return streamBundles(cmd.Context(), mgr, start, end, sync, verbose, quiet)
-		},
-	}
-
-	cmd.Flags().BoolVar(&all, "all", false, "Stream all bundles")
-	cmd.Flags().StringVarP(&rangeStr, "range", "r", "", "Stream bundle range (e.g., '1-100')")
-	cmd.Flags().BoolVar(&sync, "sync", false, "Also fetch new bundles from PLC (until caught up)")
-	cmd.Flags().StringVar(&plcURL, "plc", "https://plc.directory", "PLC directory URL (for --sync)")
-
-	return cmd
-}
-
-func streamBundles(ctx context.Context, mgr BundleManager, start, end int, doSync bool, verbose bool, quiet bool) error {
-	operationCount := 0
-
-	// Phase 1: Stream existing bundles
-	existingCount := 0
-	if end > 0 {
-		existingCount = streamExistingBundles(ctx, mgr, start, end, &operationCount, verbose, quiet)
-	}
-
-	// Phase 2: Sync and stream new bundles (if enabled)
-	fetchedCount := 0
-	if doSync {
-		if !quiet {
-			fmt.Fprintf(os.Stderr, "\nSyncing new bundles from PLC...\n")
-		}
-
-		logger := &streamLogger{quiet: quiet}
-		config := &internalsync.SyncLoopConfig{
-			MaxBundles: 0,
-			Verbose:    verbose,
-			Logger:     logger,
-			OnBundleSynced: func(bundleNum, synced, mempoolCount int, duration, indexTime time.Duration) {
-				// Stream the bundle we just created
-				if bundle, err := mgr.LoadBundle(ctx, bundleNum); err == nil {
-					for _, op := range bundle.Operations {
-						if len(op.RawJSON) > 0 {
-							fmt.Println(string(op.RawJSON))
-						} else {
-							data, _ := json.Marshal(op)
-							fmt.Println(string(data))
-						}
-					}
-					operationCount += len(bundle.Operations)
-				}
-			},
-		}
-
-		var err error
-		fetchedCount, err = mgr.RunSyncOnce(ctx, config, verbose)
-		if err != nil {
-			return err
-		}
-	}
-
-	// Summary
-	if !quiet {
-		fmt.Fprintf(os.Stderr, "\n✓ Stream complete\n")
-
-		if doSync && fetchedCount > 0 {
-			fmt.Fprintf(os.Stderr, "  Bundles: %d (%d existing + %d synced)\n",
-				existingCount+fetchedCount, existingCount, fetchedCount)
-		} else if doSync {
-			fmt.Fprintf(os.Stderr, "  Bundles: %d (already up to date)\n", existingCount)
-		} else {
-			fmt.Fprintf(os.Stderr, "  Bundles: %d\n", existingCount)
-		}
-
-		fmt.Fprintf(os.Stderr, "  Operations: %d\n", operationCount)
-	}
-
-	return nil
-}
-
-func streamExistingBundles(ctx context.Context, mgr BundleManager, start, end int, operationCount *int, verbose bool, quiet bool) int {
-	processedCount := 0
-
-	for bundleNum := start; bundleNum <= end; bundleNum++ {
-		select {
-		case <-ctx.Done():
-			return processedCount
-		default:
-		}
-
-		bundle, err := mgr.LoadBundle(ctx, bundleNum)
-		if err != nil {
-			if verbose {
-				fmt.Fprintf(os.Stderr, "Bundle %06d: not found (skipped)\n", bundleNum)
-			}
-			continue
-		}
-
-		// Output operations to stdout (JSONL)
-		for _, op := range bundle.Operations {
-			if len(op.RawJSON) > 0 {
-				fmt.Println(string(op.RawJSON))
-			} else {
-				data, _ := json.Marshal(op)
-				fmt.Println(string(data))
-			}
-		}
-
-		*operationCount += len(bundle.Operations)
-		processedCount++
-
-		if verbose {
-			fmt.Fprintf(os.Stderr, "Bundle %06d: ✓ (%d ops)\n", bundleNum, len(bundle.Operations))
-		} else if !quiet && processedCount%100 == 0 {
-			fmt.Fprintf(os.Stderr, "Streamed: %d bundles, %d ops\r", processedCount, *operationCount)
-		}
-	}
-
-	if !quiet && !verbose && processedCount > 0 {
-		fmt.Fprintf(os.Stderr, "Existing: %d bundles, %d ops\n", processedCount, *operationCount)
-	}
-
-	return processedCount
-}
-
-type streamLogger struct {
-	quiet bool
-}
-
-func (l *streamLogger) Printf(format string, v ...interface{}) {
-	if !l.quiet {
-		fmt.Fprintf(os.Stderr, format+"\n", v...)
-	}
-}
-
-func (l *streamLogger) Println(v ...interface{}) {
-	if !l.quiet {
-		fmt.Fprintln(os.Stderr, v...)
-	}
-}

cmd/plcbundle/main.go

···
 	cmd.PersistentFlags().StringP("dir", "C", "", "Repository directory (default: current directory)")
 	cmd.PersistentFlags().BoolP("verbose", "v", false, "Show detailed output and progress")
 	cmd.PersistentFlags().BoolP("quiet", "q", false, "Suppress non-error output")
-	//cmd.PersistentFlags().Bool("json", false, "Output as JSON (where applicable)")
 
 	// Bundle operations (root level - most common)
 	cmd.AddCommand(commands.NewSyncCommand())
 	cmd.AddCommand(commands.NewCloneCommand())
-	//cmd.AddCommand(commands.NewExportCommand())
-	cmd.AddCommand(commands.NewStreamCommand())
+	cmd.AddCommand(commands.NewExportCommand())
 	cmd.AddCommand(commands.NewRollbackCommand())
 
 	// Status & info (root level)
···
   server               Start HTTP server
 
 Command Groups:
-  Bundle:   clone, sync, pull, export, stream, get, rollback
+  Bundle:   clone, sync, pull, export, get, rollback
   Status:   status, log, gaps, verify, diff, stats, inspect
   DID:      did <lookup|resolve|history|batch|search|stats>
   Index:    index <build|repair|stats|verify>

server/helpers_test.go

···
-// repo/server/helpers_test.go
 package server_test
 
 import (