pipeline

Documentation

Index

• Constants
• func Backoff(attempt int) time.Duration
• func ContentHashHex(data []byte) string
• func IsRetryable(err error) bool
• type Job
  • func (j *Job) AddError(err string)
  • func (j *Job) AddFacts(valid, stored int)
  • func (j *Job) FileData() []byte
  • func (j *Job) IncrChunksProcessed()
  • func (j *Job) SetFileData(data []byte)
  • func (j *Job) SetStatus(status JobStatus, phase string)
  • func (j *Job) SetTotalChunks(n int)
  • func (j *Job) Snapshot() JobSnapshot
• type JobSnapshot
• type JobStatus
• type JobStore
  • func NewJobStore(ttl time.Duration) *JobStore
  • func (s *JobStore) Cleanup()
  • func (s *JobStore) Get(id string) *Job
  • func (s *JobStore) Put(job *Job)
• type Orchestrator
  • func NewOrchestrator(cfg config.Config, claude *extract.ClaudeClient, ps *pathstore.Client, ...) *Orchestrator
  • func (o *Orchestrator) GetJob(id string) *Job
  • func (o *Orchestrator) PathstoreClient() *pathstore.Client
  • func (o *Orchestrator) QueueDepth() int
  • func (o *Orchestrator) Start(ctx context.Context)
  • func (o *Orchestrator) Stop()
  • func (o *Orchestrator) Submit(job *Job) error
• type Progress
• type Worker
  • func NewWorker(claude *extract.ClaudeClient, ps *pathstore.Client, log *slog.Logger, ...) *Worker
  • func (w *Worker) Process(ctx context.Context, job *Job)

Constants

const MaxRetries = 3

Functions

func Backoff

func Backoff(attempt int) time.Duration

Backoff returns a duration for attempt n (0-indexed) with jitter.

func ContentHashHex

func ContentHashHex(data []byte) string

ContentHashHex computes SHA-256 of content and returns hex string.

func IsRetryable

func IsRetryable(err error) bool

IsRetryable checks if an error is worth retrying.

Types

type Job

type Job struct {
	ID     string `json:"job_id"`
	DocID  string `json:"doc_id"`
	UserID string `json:"user_id"`

	Status   JobStatus `json:"status"`
	Phase    string    `json:"phase"`
	Filename string    `json:"filename"`
	Title    string    `json:"title"`

	Progress Progress `json:"progress"`

	ContentHash string    `json:"content_hash,omitempty"`
	CreatedAt   time.Time `json:"created_at"`
	UpdatedAt   time.Time `json:"updated_at"`
	// contains filtered or unexported fields
}

Job tracks the state of a single document ingestion.

func (*Job) AddError

func (j *Job) AddError(err string)

AddError records an error.

func (*Job) AddFacts

func (j *Job) AddFacts(valid, stored int)

AddFacts records extracted/stored fact counts.

func (*Job) FileData

func (j *Job) FileData() []byte

FileData returns the raw file bytes.

func (*Job) IncrChunksProcessed

func (j *Job) IncrChunksProcessed()

IncrChunksProcessed atomically increments chunks processed.

func (*Job) SetFileData

func (j *Job) SetFileData(data []byte)

SetFileData sets the raw file bytes for processing.

func (*Job) SetStatus

func (j *Job) SetStatus(status JobStatus, phase string)

SetStatus updates job status atomically.

func (*Job) SetTotalChunks

func (j *Job) SetTotalChunks(n int)

SetTotalChunks records total chunk count.

func (*Job) Snapshot

func (j *Job) Snapshot() JobSnapshot

Snapshot returns a JSON-safe copy of the job state.

type JobSnapshot

type JobSnapshot struct {
	ID       string    `json:"job_id"`
	DocID    string    `json:"doc_id"`
	UserID   string    `json:"user_id"`
	Status   JobStatus `json:"status"`
	Phase    string    `json:"phase"`
	Filename string    `json:"filename"`
	Title    string    `json:"title"`
	Progress Progress  `json:"progress"`
}

JobSnapshot is a read-only, JSON-safe copy of job state.

type JobStatus

type JobStatus string

JobStatus represents the state of an ingestion job.

const (
	StatusQueued     JobStatus = "queued"
	StatusParsing    JobStatus = "parsing"
	StatusChunking   JobStatus = "chunking"
	StatusExtracting JobStatus = "extracting"
	StatusStoring    JobStatus = "storing"
	StatusCompleted  JobStatus = "completed"
	StatusFailed     JobStatus = "failed"
	StatusPartial    JobStatus = "partial"
	StatusDupSkipped JobStatus = "duplicate_skipped"
)

type JobStore

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

JobStore is a thread-safe in-memory job registry with TTL eviction.

func NewJobStore

func NewJobStore(ttl time.Duration) *JobStore

func (*JobStore) Cleanup

func (s *JobStore) Cleanup()

Cleanup removes expired jobs.

func (*JobStore) Get

func (s *JobStore) Get(id string) *Job

func (*JobStore) Put

func (s *JobStore) Put(job *Job)

type Orchestrator

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

Orchestrator manages the document ingestion pipeline.

func NewOrchestrator

func NewOrchestrator(cfg config.Config, claude *extract.ClaudeClient, ps *pathstore.Client, log *slog.Logger) *Orchestrator

NewOrchestrator creates and starts the pipeline.

func (*Orchestrator) GetJob

func (o *Orchestrator) GetJob(id string) *Job

GetJob returns a job by ID.

func (*Orchestrator) PathstoreClient

func (o *Orchestrator) PathstoreClient() *pathstore.Client

PathstoreClient returns the pathstore client for direct use by API handlers.

func (*Orchestrator) QueueDepth

func (o *Orchestrator) QueueDepth() int

QueueDepth returns current queue depth.

func (*Orchestrator) Start

func (o *Orchestrator) Start(ctx context.Context)

Start launches worker goroutines.

func (*Orchestrator) Stop

func (o *Orchestrator) Stop()

Stop gracefully shuts down the pipeline.

func (*Orchestrator) Submit

func (o *Orchestrator) Submit(job *Job) error

Submit queues a new job for processing.

type Progress

type Progress struct {
	TotalChunks     int      `json:"total_chunks"`
	ChunksProcessed int      `json:"chunks_processed"`
	FactsValid      int      `json:"facts_valid"`
	FactsStored     int      `json:"facts_stored"`
	Errors          []string `json:"errors"`
}

Progress tracks processing progress.

type Worker

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

Worker processes a single document job.

func NewWorker

func NewWorker(claude *extract.ClaudeClient, ps *pathstore.Client, log *slog.Logger, chunkCfg chunker.Config, maxExtract, maxStore int) *Worker

func (*Worker) Process

func (w *Worker) Process(ctx context.Context, job *Job)

Process runs the full ingest pipeline for a job.