Documentation
¶
Overview ¶
Package enrich executes enrichment sources against the deployment's grain store in one of two modes: direct (auto-approve -- assertions land in the source's enrichment:<name> graph) or queue (candidates become PIPELINE-provenance suggestions for moderation). The mode is a per-source deployment decision; the enrichers themselves are mode-blind.
Index ¶
- Variables
- type Job
- type JobStatus
- type Mode
- type Result
- type Service
- func (s *Service) CreateJob(ctx context.Context, requester, source string, filters [][2]string) (Job, error)
- func (s *Service) GetJob(ctx context.Context, id string) (Job, error)
- func (s *Service) ListJobs(ctx context.Context) ([]Job, error)
- func (s *Service) Names() []string
- func (s *Service) Run(ctx context.Context, name string, keep func(*ingest.WorkSummary) bool) (Result, error)
- func (s *Service) RunQueuedJobs(ctx context.Context) (int, error)
- type Source
Constants ¶
This section is empty.
Variables ¶
var ( // ErrUnknownSource names a source the deployment has not configured. ErrUnknownSource = errors.New("unknown enrichment source") // ErrMisconfigured marks a source whose configuration cannot run // (invalid mode, queue mode without the suggestion service). ErrMisconfigured = errors.New("enrichment source misconfigured") )
Run error classes, so the HTTP surface can map failure cause to status: the caller's mistake (ErrUnknownSource), the deployment's mistake (ErrMisconfigured), the upstream service's fault (ingest.ErrEnricher), or -- unwrapped -- a storage fault.
var ErrJobNotFound = errors.New("enrichment job not found")
ErrJobNotFound reports an unknown job id.
Functions ¶
This section is empty.
Types ¶
type Job ¶ added in v0.195.0
type Job struct {
ID string `json:"id"`
Source string `json:"source"`
// Filters are the run's [key, value] scope terms (ingest.MatchExtras
// semantics -- the same scoping the synchronous run and the audit use).
Filters [][2]string `json:"filters,omitempty"`
Requester string `json:"requester"`
Status JobStatus `json:"status"`
// Stats is the live progress while RUNNING (updated per statsInterval
// when the source reports counters) and the final tallies after.
Stats *ingest.EnrichStats `json:"stats,omitempty"`
// Result is the completed run's summary (DONE only).
Result *Result `json:"result,omitempty"`
// Error is the failure, classified the same way the synchronous
// endpoint classifies it (FAILED only; generic, detail in the log).
Error string `json:"error,omitempty"`
CreatedAt time.Time `json:"createdAt"`
StartedAt time.Time `json:"startedAt,omitzero"`
FinishedAt time.Time `json:"finishedAt,omitzero"`
}
Job is one asynchronous enrichment run: kicked with a source and scope, drained by the worker, its record carrying live batch counters while it runs so a poller can show progress on an hours-long corpus pass.
type Result ¶
type Result struct {
Source string `json:"source"`
Mode Mode `json:"mode"`
// Works is the number of Works enriched (direct) or with candidates
// queued (queue).
Works int `json:"works"`
// Scope names the run's filter ("" when the whole corpus).
Scope string `json:"scope,omitempty"`
// Stats carries the enricher's own run counters (batches, skips,
// resolved, elapsed) when the source reports them.
Stats *ingest.EnrichStats `json:"stats,omitempty"`
}
Result summarizes one run.
type Service ¶
type Service struct {
Blob blob.Store
GrainPrefix string
Queue *suggest.Service
Sources map[string]Source
// Summaries, when set, is the shared maintained summary source
// (workindex) queue-mode runs read instead of a per-run
// corpus walk; nil falls back to ScanSummaries.
Summaries ingest.SummarySource
// DB, when set, enables the async job surface (jobs.go): kick returns a
// job id, a worker drains, GET polls live progress. Nil keeps runs
// synchronous-only.
DB store.Store
// Now overrides the job clock (tests).
Now func() time.Time
}
Service runs configured sources.
func (*Service) CreateJob ¶ added in v0.195.0
func (s *Service) CreateJob(ctx context.Context, requester, source string, filters [][2]string) (Job, error)
CreateJob queues an asynchronous run. The source must exist (the caller's mistake surfaces at kick time, not first drain); execution happens on the worker via RunQueuedJobs.
func (*Service) GetJob ¶ added in v0.195.0
GetJob returns one job. The surface is admin-gated, so there is no requester scoping.
func (*Service) Run ¶
func (s *Service) Run(ctx context.Context, name string, keep func(*ingest.WorkSummary) bool) (Result, error)
Run executes one configured source by name. A non-nil keep scopes the run to the summaries it accepts: only those works are handed to the enricher (an external-service source queries for exactly the scoped set) and only their grains gain statements; out-of-scope works keep what they have.
func (*Service) RunQueuedJobs ¶ added in v0.195.0
RunQueuedJobs drains QUEUED jobs once -- the worker-loop body for container deployments (a ticker) and scheduled serverless drains alike. Job failures land in the job record; the returned error is store trouble only.
Source Files