chore: remove --bulk option and all_bulk ingest type

README.md

@@ -137,8 +137,6 @@ mayu ingest --ecosystem Go --update
 mayu ingest --all
 # Import all ecosystems with custom parallelism
 mayu ingest --all --concurrency 5 --store-workers 8
-# Bulk import from single top-level all.zip (~1.3GB, all ecosystems at once)
-mayu ingest --all --bulk
 # Import NVD CVE data directly from NVD JSON Feed 2.0
 mayu ingest --source nvd --native
 # Import only a specific year's NVD data
@@ -244,7 +242,6 @@ Import vulnerability data from OSV into the local database.
 |------|-------------|---------|
 | `--ecosystem` | Ecosystem to import (e.g., Go, PyPI, npm) | — |
 | `--all` | Import all ecosystems (dynamically fetched from GCS) | `false` |
-| `--bulk` | Use top-level all.zip for bulk import (with `--all`) | `false` |
 | `--update` | Perform delta update instead of full import | `false` |
 | `--backfill` | Backfill historical data (with `--source epss`) | `false` |
 | `--from` | Start date for backfill (YYYY-MM-DD) | `2023-03-07` (EPSS v3) |

README_ja.md

@@ -138,8 +138,6 @@ mayu ingest --ecosystem Go --update
 mayu ingest --all
 # 並列度を指定して全エコシステムをインポート
 mayu ingest --all --concurrency 5 --store-workers 8
-# トップレベル all.zip (~1.3GB) から一括インポート（全エコシステムを1ファイルで）
-mayu ingest --all --bulk
 # NVD JSON Feed 2.0 から直接 CVE データをインポート
 mayu ingest --source nvd --native
 # 特定の年度のみインポート
@@ -245,7 +243,6 @@ OSV から脆弱性データをローカルデータベースにインポート
 |--------|------|-----------|
 | `--ecosystem` | インポートするエコシステム（例: Go, PyPI, npm） | — |
 | `--all` | 全エコシステムをインポート（GCS から動的取得） | `false` |
-| `--bulk` | トップレベル all.zip で一括インポート（`--all` と併用） | `false` |
 | `--update` | フルインポートの代わりに差分更新を実行 | `false` |
 | `--backfill` | ヒストリカルデータをバックフィル（`--source epss` と併用） | `false` |
 | `--from` | バックフィルの開始日（YYYY-MM-DD） | `2023-03-07`（EPSS v3） |

cmd/mayu/ingest.go

@@ -26,7 +26,6 @@ func runIngest(args []string, cfg *config.Config) error {
 	ecosystem := fs.String("ecosystem", "", "Ecosystem to import (e.g., Go, PyPI, npm)")
 	source := fs.String("source", "", "Import from source (nvd, debian, mitre, epss, kev, ghsa)")
 	all := fs.Bool("all", false, "Import all ecosystems")
-	bulk := fs.Bool("bulk", false, "Use top-level all.zip for bulk import (with --all)")
 	update := fs.Bool("update", false, "Perform delta update instead of full import")
 	backfill := fs.Bool("backfill", false, "Backfill historical data (with --source epss)")
 	fromDate := fs.String("from", "", "Start date for backfill (YYYY-MM-DD, default: 2023-03-07 for EPSS v3)")
@@ -51,7 +50,6 @@ func runIngest(args []string, cfg *config.Config) error {
 		fmt.Println("  mayu ingest --ecosystem Go")
 		fmt.Println("  mayu ingest --ecosystem Go --update")
 		fmt.Println("  mayu ingest --all")
-		fmt.Println("  mayu ingest --all --bulk    # Download single all.zip (~1.3GB) for all ecosystems")
 		fmt.Println("  mayu ingest --source nvd")
 		fmt.Println("  mayu ingest --source nvd --native        # Import directly from NVD JSON Feed 2.0")
 		fmt.Println("  mayu ingest --source nvd --native --year 2024  # Import only 2024 NVD data")
@@ -469,21 +467,6 @@ func runIngest(args []string, cfg *config.Config) error {
 		return nil
 	}
 
-	// Handle --all --bulk: download the single top-level all.zip (~1.3GB)
-	if *all && *bulk {
-		fmt.Println("\n=== Bulk import from top-level all.zip ===")
-		stats, err := ing.BulkImportAll(ctx)
-		if err != nil {
-			if ctx.Err() != nil {
-				fmt.Fprintf(os.Stderr, "\nImport interrupted.\n")
-				return nil
-			}
-			return fmt.Errorf("bulk import: %w", err)
-		}
-		printStats(stats)
-		return nil
-	}
-
 	// Determine ecosystems to import
 	ecosystems, err := resolveEcosystems(ctx, f, *all, *ecosystem)
 	if err != nil {

internal/fetcher/fetcher.go

@@ -33,7 +33,7 @@ const (
 	// DefaultHTTPTimeout is the default timeout for HTTP requests.
 	DefaultHTTPTimeout = 5 * time.Minute
 
-	// LargeFileHTTPTimeout is the timeout for large file downloads (e.g., top-level all.zip ~1.3GB).
+	// LargeFileHTTPTimeout is the timeout for large file downloads (e.g., MITRE cvelistV5 zip).
 	LargeFileHTTPTimeout = 60 * time.Minute
 
 	// MaxResponseSize is the maximum allowed HTTP response body size (2 GB).

internal/fetcher/fetcher_test.go

@@ -632,98 +632,3 @@ func TestDownloadToTempFile(t *testing.T) {
 		t.Errorf("file permissions = %o, want 0600", perm)
 	}
 }
-
-func TestStreamTopLevelAllZip(t *testing.T) {
-	// The top-level all.zip contains files with paths like "ecosystem/vuln_id.json"
-	vulnJSON1 := `{"id":"GO-2024-0001","modified":"2024-01-01T00:00:00Z","summary":"Test vuln 1"}`
-	vulnJSON2 := `{"id":"PYSEC-2024-0001","modified":"2024-02-01T00:00:00Z","summary":"Test vuln 2"}`
-
-	zipData := createTestZip(t, map[string]string{
-		"Go/GO-2024-0001.json":      vulnJSON1,
-		"PyPI/PYSEC-2024-0001.json": vulnJSON2,
-		"README.md":                 "not a json file",
-	})
-
-	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		switch r.URL.Path {
-		case "/all.zip":
-			w.Header().Set("Content-Type", "application/zip")
-			_, _ = w.Write(zipData)
-		default:
-			http.NotFound(w, r)
-		}
-	}))
-	defer server.Close()
-
-	f := New(WithBaseURL(server.URL))
-
-	entries, errCh, totalCount, err := f.StreamTopLevelAllZip(context.Background())
-	if err != nil {
-		t.Fatalf("StreamTopLevelAllZip failed: %v", err)
-	}
-
-	// Verify total count matches expected JSON files
-	if totalCount != 2 {
-		t.Errorf("expected totalCount=2, got %d", totalCount)
-	}
-
-	// Collect all entries
-	results := make(map[string]string)
-	for entry := range entries {
-		results[entry.Name] = string(entry.Data)
-	}
-
-	// Check for streaming errors
-	if streamErr := <-errCh; streamErr != nil {
-		t.Fatalf("stream error: %v", streamErr)
-	}
-
-	// Verify results - should extract vuln_id from "ecosystem/vuln_id.json"
-	if len(results) != 2 {
-		t.Fatalf("expected 2 results, got %d: %v", len(results), results)
-	}
-	if results["GO-2024-0001"] != vulnJSON1 {
-		t.Errorf("GO-2024-0001 content mismatch: got %q", results["GO-2024-0001"])
-	}
-	if results["PYSEC-2024-0001"] != vulnJSON2 {
-		t.Errorf("PYSEC-2024-0001 content mismatch: got %q", results["PYSEC-2024-0001"])
-	}
-}
-
-func TestStreamTopLevelAllZip_ContextCancellation(t *testing.T) {
-	vulnJSON := `{"id":"GO-2024-0001","modified":"2024-01-01T00:00:00Z"}`
-	zipData := createTestZip(t, map[string]string{
-		"Go/GO-2024-0001.json":    vulnJSON,
-		"Go/GO-2024-0002.json":    vulnJSON,
-		"PyPI/PYSEC-2024-01.json": vulnJSON,
-	})
-
-	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		w.Header().Set("Content-Type", "application/zip")
-		_, _ = w.Write(zipData)
-	}))
-	defer server.Close()
-
-	f := New(WithBaseURL(server.URL))
-
-	ctx, cancel := context.WithCancel(context.Background())
-
-	entries, errCh, _, err := f.StreamTopLevelAllZip(ctx)
-	if err != nil {
-		t.Fatalf("StreamTopLevelAllZip failed: %v", err)
-	}
-
-	// Read one entry then cancel
-	<-entries
-	cancel()
-
-	// Drain remaining
-	for range entries {
-	}
-
-	// Should get context cancellation error (or nil if finished)
-	streamErr := <-errCh
-	if streamErr != nil && streamErr != context.Canceled {
-		t.Fatalf("unexpected error: %v", streamErr)
-	}
-}

internal/fetcher/stream.go

@@ -115,104 +115,6 @@ func (f *Fetcher) StreamAllZip(ctx context.Context, ecosystem string) (<-chan Zi
 	return entries, errCh, jsonCount, nil
 }
 
-// StreamTopLevelAllZip downloads the top-level all.zip (which contains
-// vulnerabilities from ALL ecosystems, ~1.3GB) to a temporary file and
-// streams entries through a channel. Each entry's filename in the zip has
-// the format "ecosystem/vuln_id.json".
-//
-// This uses a longer timeout appropriate for large file downloads.
-// The temporary file is automatically cleaned up when streaming completes.
-func (f *Fetcher) StreamTopLevelAllZip(ctx context.Context) (<-chan ZipEntry, <-chan error, int, error) {
-	u := fmt.Sprintf("%s/all.zip", f.baseURL)
-
-	// Use a longer timeout for the large download.
-	// Create a dedicated client to avoid mutating the shared httpClient (thread-safety).
-	largeClient := &http.Client{
-		Timeout:   LargeFileHTTPTimeout,
-		Transport: f.httpClient.Transport,
-	}
-	tmpFile, fileSize, err := f.downloadToTempFileWith(ctx, u, largeClient, MaxResponseSize)
-	if err != nil {
-		return nil, nil, 0, fmt.Errorf("download top-level all.zip: %w", err)
-	}
-
-	// Open zip reader from the temporary file.
-	reader, err := zip.NewReader(tmpFile, fileSize)
-	if err != nil {
-		_ = tmpFile.Close()
-		_ = os.Remove(tmpFile.Name())
-		return nil, nil, 0, fmt.Errorf("open zip: %w", err)
-	}
-
-	// Check entry count limit.
-	jsonCount := 0
-	for _, file := range reader.File {
-		if strings.HasSuffix(file.Name, ".json") {
-			jsonCount++
-		}
-	}
-	if jsonCount > MaxZipEntries {
-		_ = tmpFile.Close()
-		_ = os.Remove(tmpFile.Name())
-		return nil, nil, 0, fmt.Errorf("zip contains %d entries, exceeding maximum of %d", jsonCount, MaxZipEntries)
-	}
-
-	entries := make(chan ZipEntry, 100)
-	errCh := make(chan error, 1)
-
-	go func() {
-		defer close(entries)
-		defer close(errCh)
-		defer func() {
-			_ = tmpFile.Close()
-			_ = os.Remove(tmpFile.Name())
-		}()
-
-		var totalSize int64
-
-		for _, file := range reader.File {
-			if !strings.HasSuffix(file.Name, ".json") {
-				continue
-			}
-
-			select {
-			case <-ctx.Done():
-				errCh <- ctx.Err()
-				return
-			default:
-			}
-
-			content, err := readZipFile(file)
-			if err != nil {
-				errCh <- fmt.Errorf("read %s: %w", file.Name, err)
-				return
-			}
-
-			totalSize += int64(len(content))
-			if totalSize > MaxZipTotalSize {
-				errCh <- fmt.Errorf("zip total extracted size exceeds maximum of %d bytes", MaxZipTotalSize)
-				return
-			}
-
-			// For top-level all.zip, the filename is "ecosystem/vuln_id.json"
-			// Extract just the vuln_id part.
-			name := strings.TrimSuffix(file.Name, ".json")
-			if idx := strings.LastIndex(name, "/"); idx >= 0 {
-				name = name[idx+1:]
-			}
-
-			select {
-			case entries <- ZipEntry{Name: name, Data: content}:
-			case <-ctx.Done():
-				errCh <- ctx.Err()
-				return
-			}
-		}
-	}()
-
-	return entries, errCh, jsonCount, nil
-}
-
 // downloadToTempFile downloads the URL content directly to a temporary file,
 // avoiding loading the entire response into memory. It returns the open file
 // (seeked to beginning) and its size.

internal/ingest/ingest.go

@@ -414,94 +414,6 @@ func (ing *Ingester) DeltaImport(ctx context.Context, ecosystem string) (*Stats,
 	return stats, nil
 }
 
-// BulkImportAll performs a bulk import from the top-level all.zip, which
-// contains vulnerabilities from all ecosystems in a single archive (~1.3GB).
-// This is more efficient than importing each ecosystem separately when doing
-// a complete fresh import.
-func (ing *Ingester) BulkImportAll(ctx context.Context) (*Stats, error) {
-	start := time.Now()
-	stats := &Stats{
-		Ecosystem:  "all",
-		IsFullSync: true,
-	}
-
-	// Start job recording
-	recorder := ing.startJob(ctx, "osv-bulk", map[string]interface{}{
-		"bulk": true,
-	})
-	defer func() {
-		if recorder != nil {
-			status := "success"
-			var jobErr error
-			if stats.Errors > 0 && stats.Inserted > 0 {
-				status = "partial"
-			} else if stats.Inserted == 0 && stats.Errors > 0 {
-				status = "failed"
-			}
-			if ctx.Err() != nil {
-				status = "failed"
-				jobErr = ctx.Err()
-			}
-			recorder.Finish(ctx, status, stats.Total, stats.Inserted, stats.Errors, jobErr)
-		}
-	}()
-
-	// Phase 1: Download the top-level all.zip.
-	ing.progress(Progress{Phase: "download", Message: "Downloading top-level all.zip (~1.3GB)... this may take a while."})
-
-	entries, errCh, totalEntries, err := ing.fetcher.StreamTopLevelAllZip(ctx)
-	if err != nil {
-		return nil, fmt.Errorf("fetch top-level all.zip: %w", err)
-	}
-
-	// Phase 2+3: Parallel parse and store with multiple workers.
-	ing.progress(Progress{Phase: "store", Message: fmt.Sprintf("Processing %d entries...", totalEntries)})
-
-	inserted, processed, parseErrors, err := ing.streamParseAndStore(ctx, entries, errCh, totalEntries)
-	if err != nil {
-		return nil, err
-	}
-
-	stats.Inserted = inserted
-	stats.Total = processed + parseErrors
-	stats.Errors = parseErrors
-	stats.Skipped = parseErrors
-
-	// Update sync state for "all".
-	now := time.Now().UTC().Format(time.RFC3339)
-	syncState := &store.SyncState{
-		Source:         "all",
-		LastModifiedAt: now,
-		RecordCount:    int64(stats.Inserted),
-	}
-	if err := ing.store.UpdateSyncState(ctx, syncState); err != nil {
-		ing.logger.Printf("warning: failed to update sync state: %v", err)
-	}
-
-	// Update sync state for each ecosystem so that subsequent delta updates
-	// (--all --update) can use the bulk import timestamp as baseline.
-	ecosystems, listErr := ing.fetcher.ListEcosystems(ctx)
-	if listErr != nil {
-		ing.logger.Printf("warning: failed to list ecosystems for sync state update: %v", listErr)
-	} else {
-		for _, eco := range ecosystems {
-			ecoState := &store.SyncState{
-				Source:         eco,
-				LastModifiedAt: now,
-				RecordCount:    0, // Exact per-ecosystem count unknown; will be corrected on next full/delta import.
-			}
-			if err := ing.store.UpdateSyncState(ctx, ecoState); err != nil {
-				ing.logger.Printf("warning: failed to update sync state for %s: %v", eco, err)
-			}
-		}
-	}
-
-	stats.Duration = time.Since(start)
-	ing.progress(Progress{Phase: "store", Current: stats.Inserted, Total: stats.Total, Message: fmt.Sprintf("Done: %d inserted in %s", stats.Inserted, stats.Duration.Round(time.Millisecond))})
-
-	return stats, nil
-}
-
 // storeBatches splits a slice of vulnerabilities into batches and stores them
 // using parallel workers. Returns the total count inserted.
 func (ing *Ingester) storeBatches(ctx context.Context, vulns []*model.Vulnerability) (int, error) {
@@ -531,8 +443,8 @@ func (ing *Ingester) storeBatches(ctx context.Context, vulns []*model.Vulnerabil
 }
 
 // streamParseAndStore reads ZipEntry values from a channel, parses them, and
-// stores them in parallel batches. This is the shared pipeline used by both
-// FullImport and BulkImportAll.
+// stores them in parallel batches. This is the shared pipeline used by
+// FullImport.
 //
 // It returns (inserted, processed, errors, err).
 func (ing *Ingester) streamParseAndStore(ctx context.Context, entries <-chan fetcher.ZipEntry, errCh <-chan error, total int) (inserted int, processed int, parseErrors int, err error) {