pipeline

type ArchiveJob struct {
	Name         string // human-readable label used in log messages and spans
	URL          string // HTTP download URL; empty for local-file sources
	LocalPath    string // path on disk; empty for HTTP sources
	ExpectedSize int64  // byte count from the catalogue; 0 if unknown
	Checksum     string // SHA-1 hex as listed in the catalogue; empty disables verification
}

type ArchiveKind uint8

type ArchiveOpener interface {
	Stream(ctx context.Context, job ArchiveJob) rill.Stream[XMLEntry]
}

type ArchiveSource interface {
	Stream(ctx context.Context) rill.Stream[ArchiveJob]
}

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

type CheckpointJanitor struct {
	CP     Checkpointer
	Logger *slog.Logger
}

type CheckpointSource struct {
	Inner  ArchiveSource
	CP     Checkpointer
	Logger *slog.Logger
}

type Checkpointer interface {
	// IsCompleted reports whether the job with the given stable ID has
	// already been processed in a previous run.
	IsCompleted(jobID string) (bool, error)
	// MarkCompleted atomically and durably records that the job has been
	// fully drained and its records flushed to the sink.
	MarkCompleted(jobID string, meta map[string]string) error
	// HasAny reports whether at least one job has been recorded as completed.
	HasAny() (bool, error)
	// Close releases any underlying resources.
	Close() error
}

type EPOProductSource struct {
	BaseURL   string
	ProductID int
	Client    *http.Client
}

type HTTPOpener struct {
	Client     *http.Client
	MaxRetries uint
	VerifySHA1 bool
	Walk       WalkConfig

	// KeepArchive, when true, tees the HTTP response body to a file under
	// ArchiveDir while the walker consumes it. Orthogonal to
	// Walk.KeepExtracted.
	KeepArchive bool
	ArchiveDir  string

	// Logger, when non-nil, receives structured warnings (e.g. for
	// unsupported archive kinds).
	Logger *slog.Logger

	// Janitor receives a per-archive completion notification after the
	// stream is fully drained without error. Defaults to NoopJanitor.
	Janitor Janitor

	// OnBytes, when non-nil, receives throttled byte-level progress for
	// the current download (downloaded, total). total is -1 when the
	// server omits Content-Length. Called from the producer goroutine;
	// must not block.
	OnBytes func(job ArchiveJob, downloaded, total int64)

	// OnArchiveSettled, when non-nil, is invoked exactly once per archive
	// after it terminally succeeds (err == nil) or fails after retries
	// (err != nil). Lets a progress display retire the archive's bar and
	// the summary collector tally failures. Called from the producer
	// goroutine; must not block.
	OnArchiveSettled func(job ArchiveJob, err error)

	// ProgressInterval throttles OnBytes calls. Defaults to 100ms.
	ProgressInterval time.Duration
}

type Janitor interface {
	OnArchiveDone(ctx context.Context, job ArchiveJob, localPath string, kept bool)
	OnEntryDone(ctx context.Context, entry XMLEntry, localPath string)
}

type LocalFileOpener struct {
	Walk    WalkConfig
	Logger  *slog.Logger
	Janitor Janitor
}

type MemorySink struct {
	Records []PatentRecord
	// contains filtered or unexported fields
}

type NoopCheckpointer struct{}

type NoopExtractor struct{}

type NoopJanitor struct{}

type NoopSink struct{}

type Option func(*Options)

type Options struct {
	// Source produces the stream of ArchiveJob values to process.
	Source ArchiveSource
	// Opener fetches or opens each archive and emits XMLEntry values.
	Opener ArchiveOpener
	// Extractor parses each XMLEntry into PatentRecord values.
	Extractor RecordExtractor
	// Sink consumes batched PatentRecord values (serialised, one goroutine).
	Sink RecordSink
	// Janitor receives per-archive and per-entry lifecycle notifications.
	// Defaults to NoopJanitor.
	Janitor Janitor
	// Logger receives structured pipeline events. Defaults to a discard logger.
	Logger *slog.Logger
	// Progress is called on every counter update. Defaults to no-op.
	Progress ProgressFunc
	// ArchiveConcurrency is the number of archives in flight simultaneously.
	// Default: 4.
	ArchiveConcurrency int
	// ExtractorConcurrency is the number of in-flight entries decoded in
	// parallel (one sequential, CPU-bound reader each) — the knob that
	// saturates cores. Auto-sizing (NumCPU clamped by a memory budget) is a
	// caller policy: resolve it with PlanReaderConcurrency before passing it
	// here. Values <= 0 fall back to a minimal safe default. Default: 4.
	ExtractorConcurrency int
	// BatchSize is the target number of records per sink Write call.
	// Default: 1000.
	BatchSize int
	// BatchTimeout is the maximum idle period before a partial batch is
	// flushed to the sink. Default: 2s.
	BatchTimeout time.Duration
}

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

type PatentRecord = parse.PatentRecord

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

type ProgressFunc func(Stats)

type RecordExtractor interface {
	Stream(ctx context.Context, entry XMLEntry) rill.Stream[PatentRecord]
}

type RecordSink interface {
	Write(ctx context.Context, batch []PatentRecord) error
	Close() error
}

type StaticListSource struct{ Jobs []ArchiveJob }

type Stats struct {
	Archives int64 // total archives fully drained
	Entries  int64 // total XML entries emitted across all archives
	Records  int64 // total PatentRecord values produced by extractors
	Batches  int64 // total Write calls delivered to the sink
}

type WalkConfig struct {
	// SpoolDir is where zip archives are spooled (zip needs random access).
	// Tar / tar.gz never use it. Defaults to OS temp.
	SpoolDir string

	// EntrySelector decides whether an entry is emitted to the extractor.
	// Nested archives are always recursed regardless. Defaults to IsXML.
	EntrySelector func(name string) bool

	// KeepExtracted, when true, tees each selected entry's bytes to a
	// file under ExtractedDir while the consumer reads it.
	KeepExtracted bool

	// ExtractedDir receives kept entry files. Required when KeepExtracted is true.
	ExtractedDir string
}

type XMLEntry struct {
	ArchiveName string    // name of the enclosing archive (may be a "!" chain for nested archives)
	Name        string    // path of this entry within the archive
	Size        int64     // payload byte count
	Reader      io.Reader // in-memory entry body; independent of the container stream
	// contains filtered or unexported fields
}

type XMLStreamExtractor struct {
	// ElementXPath selects the streamed element. Defaults to the EPO contract.
	ElementXPath string
	// ParseConcurrency is the number of goroutines that run the CPU-bound
	// per-document parse in parallel within a single entry. Because the
	// reader feeds nodes sequentially, a small value (just enough to overlap
	// parse with read) is sufficient; large per-entry values only
	// over-subscribe (N entries x ParseConcurrency goroutines). Values <= 0
	// default to defaultParseConcurrency.
	ParseConcurrency int
}