processing

package
v1.2.1 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 1, 2025 License: MIT Imports: 16 Imported by: 0

Documentation

Overview

Package processing coordinates the file processing pipeline for PI scanning. It manages concurrent file processing, integrates AST analysis with detection, and orchestrates the two-phase detection approach (pattern matching + LLM validation).

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type BatchProcessor

type BatchProcessor struct {
	// contains filtered or unexported fields
}

BatchProcessor handles processing multiple files efficiently

func NewBatchProcessor

func NewBatchProcessor(processor *FileProcessor, batchSize int) *BatchProcessor

NewBatchProcessor creates a batch processor for multiple files

func (*BatchProcessor) ProcessFiles

func (bp *BatchProcessor) ProcessFiles(ctx context.Context, jobs []FileJob) ([]ProcessingResult, error)

ProcessFiles processes a slice of file jobs in batches

type FileContent

type FileContent struct {
	Path        string
	Data        []byte
	Chunks      [][]byte
	LineMap     map[int]int64
	WasStreamed bool
}

FileContent represents processed file content

func (*FileContent) AddLine

func (fc *FileContent) AddLine(lineNum int, line []byte)

AddLine adds a line to the file content

func (*FileContent) GetContent

func (fc *FileContent) GetContent() []byte

GetContent returns the full content, whether streamed or not

func (*FileContent) GetLines

func (fc *FileContent) GetLines() [][]byte

GetLines returns an iterator over lines in the content

type FileJob

type FileJob struct {
	FilePath   string
	Content    []byte
	FileInfo   discovery.FileResult
	ASTContext *ast.FileContext // Optional AST analysis context
}

FileJob represents a file to be processed through the detection pipeline

type FileProcessor

type FileProcessor struct {
	// contains filtered or unexported fields
}

FileProcessor handles concurrent file processing through the detection pipeline

func NewFileProcessor

func NewFileProcessor(config ProcessorConfig, detectors []detection.Detector) *FileProcessor

NewFileProcessor creates a new file processor with detectors

func (*FileProcessor) GetStats

func (fp *FileProcessor) GetStats() ProcessorStats

GetStats returns processor statistics

func (*FileProcessor) Results

func (fp *FileProcessor) Results() <-chan ProcessingResult

Results returns a channel for receiving processing results

func (*FileProcessor) Start

func (fp *FileProcessor) Start(ctx context.Context) error

Start initializes and starts all workers

func (*FileProcessor) Stop

func (fp *FileProcessor) Stop()

Stop gracefully stops the file processor

func (*FileProcessor) Submit

func (fp *FileProcessor) Submit(job FileJob) error

Submit adds a file to the processing queue

type FileWorker

type FileWorker struct {
	// contains filtered or unexported fields
}

FileWorker represents a single worker processing files

type MemoryTracker

type MemoryTracker struct {
	// contains filtered or unexported fields
}

MemoryTracker tracks memory usage across the application

func NewMemoryTracker

func NewMemoryTracker(maxMemory int64) *MemoryTracker

NewMemoryTracker creates a new memory tracker

func (*MemoryTracker) Acquire

func (m *MemoryTracker) Acquire(ctx context.Context, size int64) error

Acquire acquires memory, blocking if necessary

func (*MemoryTracker) AvailableMemory

func (m *MemoryTracker) AvailableMemory() int64

AvailableMemory returns the available memory

func (*MemoryTracker) CurrentUsage

func (m *MemoryTracker) CurrentUsage() int64

CurrentUsage returns the current memory usage

func (*MemoryTracker) Release

func (m *MemoryTracker) Release(size int64)

Release releases allocated memory

func (*MemoryTracker) TryAcquire

func (m *MemoryTracker) TryAcquire(size int64) error

TryAcquire attempts to acquire memory without blocking

type ProcessOptions

type ProcessOptions struct {
	StreamLargeFiles bool
	MaxMemoryPerFile int64
	EnableCaching    bool
}

ProcessOptions configures how files are processed

type ProcessingResult

type ProcessingResult struct {
	FilePath string
	Findings []detection.Finding
	Error    error
	Stats    ProcessingStats
}

ProcessingResult represents the result of processing a file

type ProcessingStats

type ProcessingStats struct {
	BytesProcessed int64
	LinesProcessed int
	ProcessingTime int64 // nanoseconds
}

ProcessingStats tracks processing statistics

type ProcessorConfig

type ProcessorConfig struct {
	NumWorkers       int
	QueueSize        int
	MaxFileSize      int64
	MaxMemory        int64
	StreamLargeFiles bool
	EnablePatterns   bool
	EnableGitleaks   bool
	EnableCaching    bool
}

ProcessorConfig configures the file processor

func DefaultProcessorConfig

func DefaultProcessorConfig() ProcessorConfig

DefaultProcessorConfig returns sensible defaults

type ProcessorStats

type ProcessorStats struct {
	NumWorkers     int
	QueuedJobs     int
	PendingResults int
	IsStarted      bool
}

ProcessorStats provides processor statistics

type StreamProcessor

type StreamProcessor struct {
	// contains filtered or unexported fields
}

StreamProcessor handles file processing with memory management and streaming support

func NewStreamProcessor

func NewStreamProcessor(opts ...StreamProcessorOption) *StreamProcessor

NewStreamProcessor creates a new stream processor

func (*StreamProcessor) ProcessFile

func (sp *StreamProcessor) ProcessFile(ctx context.Context, path string, opts ProcessOptions) (*FileContent, error)

ProcessFile processes a file with memory management

type StreamProcessorOption

type StreamProcessorOption func(*StreamProcessor)

StreamProcessorOption configures the StreamProcessor

func WithChunkSize

func WithChunkSize(size int) StreamProcessorOption

WithChunkSize sets the chunk size for streaming

func WithMaxFileSize

func WithMaxFileSize(size int64) StreamProcessorOption

WithMaxFileSize sets the maximum file size before streaming

func WithMemoryTracker

func WithMemoryTracker(tracker *MemoryTracker) StreamProcessorOption

WithMemoryTracker sets the memory tracker

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL