From 5fba77b59c18683f4c7ba1d4961f8459d4bd917d Mon Sep 17 00:00:00 2001 From: Bakhtiyar Issakhmetov Date: Sat, 5 Sep 2026 13:40:25 +0500 Subject: [PATCH] feat(landscape): run planning/execution service with fingerprint reuse; RabbitMQ job Co-Authored-By: Claude Fable 5.1 --- .../messaging/rabbitmq/landscape_consumer.go | 82 ++ internal/service/landscape.go | 709 ++++++++++ internal/service/landscape_test.go | 1161 +++++++++++++++++ 3 files changed, 1952 insertions(+) create mode 100644 internal/messaging/rabbitmq/landscape_consumer.go create mode 100644 internal/service/landscape.go create mode 100644 internal/service/landscape_test.go diff --git a/internal/messaging/rabbitmq/landscape_consumer.go b/internal/messaging/rabbitmq/landscape_consumer.go new file mode 100644 index 0000000..96350e6 --- /dev/null +++ b/internal/messaging/rabbitmq/landscape_consumer.go @@ -0,0 +1,82 @@ +package rabbitmq + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + + "github.com/google/uuid" + amqp "github.com/rabbitmq/amqp091-go" +) + +const ( + // LandscapeRunRoutingKey routes Landscape Program run jobs. + LandscapeRunRoutingKey = "landscape.run" + // LandscapeRunQueue receives Landscape Program run jobs. + LandscapeRunQueue = "gis.landscape.run" +) + +// LandscapeJob is the message body for a Landscape Program run job. +type LandscapeJob struct { + RunID uuid.UUID `json:"run_id"` +} + +// LandscapeJobPublisher publishes Landscape Program run jobs; it implements +// service.LandscapeJobEnqueuer. +type LandscapeJobPublisher struct { + pub *Publisher +} + +// NewLandscapeJobPublisher returns a LandscapeJobPublisher. +func NewLandscapeJobPublisher(pub *Publisher) *LandscapeJobPublisher { + return &LandscapeJobPublisher{pub: pub} +} + +// EnqueueRun publishes an execution job for the given run. +func (p *LandscapeJobPublisher) EnqueueRun(ctx context.Context, runID uuid.UUID) error { + body, err := json.Marshal(LandscapeJob{RunID: runID}) + if err != nil { + return err + } + return p.pub.Publish(ctx, LandscapeRunRoutingKey, body) +} + +// LandscapeProcessor executes Landscape Program runs; implemented by the service. +type LandscapeProcessor interface { + Execute(ctx context.Context, runID uuid.UUID) error + // MarkRunFailed records a terminal failure on the run so it does not remain + // stuck in an in-progress status. + MarkRunFailed(ctx context.Context, runID uuid.UUID, reason string) error +} + +// landscapeHandler executes one Landscape Program run per delivery. +type landscapeHandler struct { + proc LandscapeProcessor + log *slog.Logger +} + +func (h landscapeHandler) Handle(ctx context.Context, d amqp.Delivery) error { + var job LandscapeJob + if err := json.Unmarshal(d.Body, &job); err != nil { + return fmt.Errorf("decode landscape run job: %w", err) + } + h.log.Info("processing landscape run", "run_id", job.RunID) + + err := h.proc.Execute(ctx, job.RunID) + if err != nil { + // The delivery is nacked without requeue (see Consumer.dispatch), so the + // run will not be retried. Record the failure on the run itself; + // otherwise it stays "running" forever. Best-effort: log if marking + // fails, but still return the original error so the delivery is nacked. + if markErr := h.proc.MarkRunFailed(ctx, job.RunID, err.Error()); markErr != nil { + h.log.Error("mark landscape run failed", "run_id", job.RunID, "error", markErr) + } + } + return err +} + +// NewLandscapeRunHandler returns a handler that executes Landscape Program runs. +func NewLandscapeRunHandler(p LandscapeProcessor, log *slog.Logger) Handler { + return landscapeHandler{proc: p, log: log} +} diff --git a/internal/service/landscape.go b/internal/service/landscape.go new file mode 100644 index 0000000..7fcf066 --- /dev/null +++ b/internal/service/landscape.go @@ -0,0 +1,709 @@ +package service + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + "sort" + "sync" + "time" + + "gis/internal/domain" + "gis/internal/landscape" + "gis/internal/raster" + + "github.com/google/uuid" + "golang.org/x/sync/errgroup" +) + +// ComputeFunc runs one graph node over its input rasters, writing one raster +// per output group into outDir and returning the produced paths by group. It +// mirrors landscape.Compute so tests can substitute a stub. +type ComputeFunc func(ctx context.Context, rio raster.IO, node landscape.Node, inputs map[string]string, outDir string, p landscape.Params) (map[string]string, error) + +// LandscapeRepository is the persistence behaviour LandscapeService needs. +type LandscapeRepository interface { + CreateRun(ctx context.Context, run domain.LPRun) (domain.LPRun, error) + GetRun(ctx context.Context, id uuid.UUID) (domain.LPRun, error) + ListRuns(ctx context.Context, limit, offset int) ([]domain.LPRun, error) + CountRuns(ctx context.Context) (int, error) + SetRunStatus(ctx context.Context, id uuid.UUID, status domain.LPRunStatus, errMsg *string) error + SetNodeStarted(ctx context.Context, runID uuid.UUID, code string, fingerprint string, inputs []domain.LPDatasetRef) error + SetNodeFinished(ctx context.Context, runID uuid.UUID, code string, status domain.LPNodeStatus, outputs []domain.LPDatasetRef, errMsg *string) error + FindReusableOutputs(ctx context.Context, code, fingerprint string, excludeRun uuid.UUID) ([]domain.LPDatasetRef, error) +} + +// LandscapeJobEnqueuer schedules asynchronous Landscape Program runs. +type LandscapeJobEnqueuer interface { + EnqueueRun(ctx context.Context, runID uuid.UUID) error +} + +// datasetPublisher is the slice of DatasetService the Landscape Program needs: +// uploading a slot input and publishing a computed result. +type datasetPublisher interface { + Upload(ctx context.Context, in UploadInput) (domain.Dataset, error) + CreateRasterFromFile(ctx context.Context, in CreateRasterInput) (domain.Dataset, error) +} + +// DatasetService is what the landscape service publishes through in production. +var _ datasetPublisher = (*DatasetService)(nil) + +// SlotInfo describes one leaf input slot and the dataset currently filling it. +type SlotInfo struct { + Code string `json:"code"` + Title string `json:"title"` + GroupID uuid.UUID `json:"group_id"` + Latest *domain.DatasetSummary `json:"latest"` +} + +// ResultInfo describes one computed output group and its freshest ready dataset. +type ResultInfo struct { + Group string `json:"group"` + Title string `json:"title"` + Node string `json:"node"` + GroupID uuid.UUID `json:"group_id"` + Latest *domain.DatasetSummary `json:"latest"` + CogURL *string `json:"cog_url"` + SldURL *string `json:"sld_url"` +} + +// SlotUploadInput carries a raster upload destined for one input slot. +type SlotUploadInput struct { + Filename string + ContentType string + Size int64 + Reader io.Reader + Name string + StartDate *string + EndDate *string +} + +// RunInput carries the options of a new Landscape Program run. +type RunInput struct { + // FromStage forces recomputation from this stage on (0..3); nil recomputes + // only what changed. + FromStage *int + // Overrides pins a specific dataset for a group code instead of resolving + // the freshest ready one. + Overrides map[string]uuid.UUID + Params landscape.Params +} + +// LPRunPage is a page of runs with pagination metadata. +type LPRunPage struct { + Items []domain.LPRun + Page int + PageSize int + Total int +} + +// LandscapeService plans and executes Landscape Program runs. +type LandscapeService struct { + runs LandscapeRepository + datasets DatasetRepository + groups DatasetGroupRepository + categories categoryReader + publisher datasetPublisher + store ObjectStore + jobs LandscapeJobEnqueuer + rio raster.IO + compute ComputeFunc + log *slog.Logger +} + +// NewLandscapeService wires the run repository, the dataset and group +// repositories (input resolution), the category reader (the "landscape" +// category every result is filed under), the dataset publisher (result +// datasets and slot uploads), the object store (input downloads), the job +// enqueuer, and the raster IO / compute function the graph runs on. +func NewLandscapeService( + runs LandscapeRepository, + datasets DatasetRepository, + groups DatasetGroupRepository, + categories categoryReader, + publisher datasetPublisher, + store ObjectStore, + jobs LandscapeJobEnqueuer, + rio raster.IO, + compute ComputeFunc, + log *slog.Logger, +) *LandscapeService { + return &LandscapeService{ + runs: runs, + datasets: datasets, + groups: groups, + categories: categories, + publisher: publisher, + store: store, + jobs: jobs, + rio: rio, + compute: compute, + log: log, + } +} + +// Slots lists every leaf input slot with its group and latest dataset (any +// status), so the UI can show what has been uploaded. +func (s *LandscapeService) Slots(ctx context.Context) ([]SlotInfo, error) { + out := make([]SlotInfo, 0, len(landscape.Slots)) + for _, slot := range landscape.Slots { + g, err := s.groups.GetByCode(ctx, slot.Code) + if err != nil { + return nil, fmt.Errorf("dataset group %q: %w", slot.Code, err) + } + info := SlotInfo{Code: slot.Code, Title: slot.Title, GroupID: g.ID} + latest, err := s.datasets.LatestInGroup(ctx, g.ID, nil, nil) + switch { + case err == nil: + summary := summaryOf(latest) + info.Latest = &summary + case errors.Is(err, domain.ErrNotFound): + // An empty slot is expected before the first upload. + default: + return nil, err + } + out = append(out, info) + } + return out, nil +} + +// UploadToSlot stores a raster under the slot's dataset group in the landscape +// category, through the regular upload path (COG conversion is enqueued). +// Returns domain.ErrNotFound when the code is not a known slot. +func (s *LandscapeService) UploadToSlot(ctx context.Context, code string, in SlotUploadInput) (domain.Dataset, error) { + slot, ok := landscape.SlotByCode(code) + if !ok { + return domain.Dataset{}, fmt.Errorf("%w: unknown landscape slot %q", domain.ErrNotFound, code) + } + g, err := s.groups.GetByCode(ctx, slot.Code) + if err != nil { + return domain.Dataset{}, fmt.Errorf("dataset group %q: %w", slot.Code, err) + } + cat, err := s.categories.GetByCode(ctx, landscape.CategoryCode) + if err != nil { + return domain.Dataset{}, fmt.Errorf("category %q: %w", landscape.CategoryCode, err) + } + name := in.Name + if name == "" { + name = in.Filename + } + return s.publisher.Upload(ctx, UploadInput{ + CategoryID: cat.ID, + DatasetGroupID: &g.ID, + Name: name, + Filename: in.Filename, + FileType: domain.FileTypeRaster, + ContentType: in.ContentType, + Size: in.Size, + Reader: in.Reader, + StartDate: in.StartDate, + EndDate: in.EndDate, + }) +} + +// CreateRun validates the params and overrides, persists a queued run with one +// pending node per graph node, and publishes the worker job. +func (s *LandscapeService) CreateRun(ctx context.Context, in RunInput) (domain.LPRun, error) { + params := in.Params.WithDefaults() + if err := params.Validate(); err != nil { + return domain.LPRun{}, fmt.Errorf("%w: %s", domain.ErrValidation, err) + } + if in.FromStage != nil { + if *in.FromStage < 0 || *in.FromStage > maxStage() { + return domain.LPRun{}, fmt.Errorf("%w: from_stage must be in 0..%d", domain.ErrValidation, maxStage()) + } + } + + overrides := map[string]uuid.UUID{} + for group, id := range in.Overrides { + if !knownGroup(group) { + return domain.LPRun{}, fmt.Errorf("%w: unknown landscape group %q in overrides", domain.ErrValidation, group) + } + d, err := s.datasets.GetByID(ctx, id) + if err != nil { + if errors.Is(err, domain.ErrNotFound) { + return domain.LPRun{}, fmt.Errorf("%w: override for %q: dataset %s does not exist", domain.ErrValidation, group, id) + } + return domain.LPRun{}, err + } + if d.FileType != domain.FileTypeRaster { + return domain.LPRun{}, fmt.Errorf("%w: override for %q: dataset %s is not a raster", domain.ErrValidation, group, id) + } + if d.Status != domain.DatasetStatusReady { + return domain.LPRun{}, fmt.Errorf("%w: override for %q: dataset %s is not ready", domain.ErrValidation, group, id) + } + overrides[group] = id + } + + // lp_runs.params is NOT NULL, and the kernels misbehave on a partially + // filled Params, so the defaulted-and-validated value is what we persist. + paramsJSON, err := json.Marshal(params) + if err != nil { + return domain.LPRun{}, err + } + + nodes := make([]domain.LPRunNode, 0, len(landscape.Nodes)) + for _, n := range landscape.Nodes { + nodes = append(nodes, domain.LPRunNode{NodeCode: n.Code, Stage: n.Stage, Status: domain.LPNodePending}) + } + + run, err := s.runs.CreateRun(ctx, domain.LPRun{ + Status: domain.LPRunQueued, + FromStage: in.FromStage, + Params: paramsJSON, + Overrides: overrides, + Nodes: nodes, + }) + if err != nil { + return domain.LPRun{}, err + } + + if err := s.jobs.EnqueueRun(ctx, run.ID); err != nil { + msg := fmt.Sprintf("failed to enqueue: %s", err) + if setErr := s.runs.SetRunStatus(ctx, run.ID, domain.LPRunFailed, &msg); setErr != nil { + s.log.Error("mark landscape run failed", "run_id", run.ID, "error", setErr) + } + return domain.LPRun{}, fmt.Errorf("enqueue landscape run: %w", err) + } + return run, nil +} + +// GetRun returns one run with its nodes. +func (s *LandscapeService) GetRun(ctx context.Context, id uuid.UUID) (domain.LPRun, error) { + return s.runs.GetRun(ctx, id) +} + +// ListRuns returns a page of runs, newest first. +func (s *LandscapeService) ListRuns(ctx context.Context, page, pageSize int) (LPRunPage, error) { + if page < 1 { + page = 1 + } + if pageSize < 1 { + pageSize = DefaultPageSize + } + if pageSize > MaxPageSize { + pageSize = MaxPageSize + } + items, err := s.runs.ListRuns(ctx, pageSize, (page-1)*pageSize) + if err != nil { + return LPRunPage{}, err + } + total, err := s.runs.CountRuns(ctx) + if err != nil { + return LPRunPage{}, err + } + return LPRunPage{Items: items, Page: page, PageSize: pageSize, Total: total}, nil +} + +// Results lists every output group with its freshest ready dataset and the URLs +// to render it. +func (s *LandscapeService) Results(ctx context.Context) ([]ResultInfo, error) { + ready := domain.DatasetStatusReady + fileType := domain.FileTypeRaster + + out := make([]ResultInfo, 0, len(landscape.Outputs)) + for _, o := range landscape.Outputs { + g, err := s.groups.GetByCode(ctx, o.Code) + if err != nil { + return nil, fmt.Errorf("dataset group %q: %w", o.Code, err) + } + info := ResultInfo{Group: o.Code, Title: o.Title, GroupID: g.ID} + if node, ok := landscape.NodeForOutput(o.Code); ok { + info.Node = node.Code + } + + latest, err := s.datasets.LatestInGroup(ctx, g.ID, &ready, &fileType) + switch { + case err == nil: + summary := summaryOf(latest) + info.Latest = &summary + cog := "/datasets/" + latest.ID.String() + ".cog" + info.CogURL = &cog + if g.SldStorageKey != nil { + sld := "/dataset-groups/" + g.ID.String() + "/sld" + info.SldURL = &sld + } + case errors.Is(err, domain.ErrNotFound): + // Nothing computed for this group yet. + default: + return nil, err + } + out = append(out, info) + } + return out, nil +} + +// MarkRunFailed records a terminal failure on a run so it does not stay stuck +// in an in-progress status. +func (s *LandscapeService) MarkRunFailed(ctx context.Context, runID uuid.UUID, reason string) error { + return s.runs.SetRunStatus(ctx, runID, domain.LPRunFailed, &reason) +} + +// Execute runs the whole graph for one run: stages in order, nodes within a +// stage concurrently. The first failing node fails the run and later stages are +// not started. It is the worker entry point. +func (s *LandscapeService) Execute(ctx context.Context, runID uuid.UUID) error { + run, err := s.runs.GetRun(ctx, runID) + if err != nil { + return err + } + + var params landscape.Params + if len(run.Params) > 0 { + if err := json.Unmarshal(run.Params, ¶ms); err != nil { + return s.failRun(ctx, runID, fmt.Errorf("decode run params: %w", err)) + } + } + params = params.WithDefaults() + + cat, err := s.categories.GetByCode(ctx, landscape.CategoryCode) + if err != nil { + return s.failRun(ctx, runID, fmt.Errorf("category %q: %w", landscape.CategoryCode, err)) + } + + if err := s.runs.SetRunStatus(ctx, runID, domain.LPRunRunning, nil); err != nil { + return err + } + // Re-read so started_at is available for result dataset names. + if started, err := s.runs.GetRun(ctx, runID); err == nil { + run = started + } + + state := &runState{produced: map[string]uuid.UUID{}} + for _, stage := range landscape.Stages() { + var g errgroup.Group + for _, node := range stage { + g.Go(func() error { return s.runNode(ctx, run, node, params, cat, state) }) + } + // Wait for every node of the stage: each one records its own outcome, + // and a sibling's failure must not leave a node half-recorded. + if err := g.Wait(); err != nil { + return s.failRun(ctx, runID, err) + } + } + + return s.runs.SetRunStatus(ctx, runID, domain.LPRunSucceeded, nil) +} + +// failRun marks the run failed with cause's message and returns cause. +func (s *LandscapeService) failRun(ctx context.Context, runID uuid.UUID, cause error) error { + msg := cause.Error() + if err := s.runs.SetRunStatus(ctx, runID, domain.LPRunFailed, &msg); err != nil { + s.log.Error("mark landscape run failed", "run_id", runID, "error", err) + } + return cause +} + +// runState carries what the run has produced so far, shared by the concurrent +// nodes of a stage. +type runState struct { + mu sync.Mutex + produced map[string]uuid.UUID +} + +func (r *runState) get(group string) (uuid.UUID, bool) { + r.mu.Lock() + defer r.mu.Unlock() + id, ok := r.produced[group] + return id, ok +} + +func (r *runState) record(refs []domain.LPDatasetRef) { + r.mu.Lock() + defer r.mu.Unlock() + for _, ref := range refs { + r.produced[ref.Group] = ref.DatasetID + } +} + +// runNode resolves a node's inputs, reuses a previous result when the +// fingerprint matches, and otherwise computes and publishes its outputs. +func (s *LandscapeService) runNode(ctx context.Context, run domain.LPRun, node landscape.Node, params landscape.Params, cat domain.Category, state *runState) error { + refs, inputs, err := s.resolveInputs(ctx, run, node, state) + if err != nil { + return s.finishNodeFailed(ctx, run.ID, node, err) + } + + fp := fingerprint(node, refs, params) + if err := s.runs.SetNodeStarted(ctx, run.ID, node.Code, fp, refs); err != nil { + return err + } + + if run.FromStage == nil || node.Stage < *run.FromStage { + reused, ok, err := s.reusableOutputs(ctx, run.ID, node, fp) + if err != nil { + return s.finishNodeFailed(ctx, run.ID, node, err) + } + if ok { + state.record(reused) + return s.runs.SetNodeFinished(ctx, run.ID, node.Code, domain.LPNodeSkipped, reused, nil) + } + } + + outputs, err := s.computeNode(ctx, run, node, params, cat, refs, inputs) + if err != nil { + return s.finishNodeFailed(ctx, run.ID, node, err) + } + state.record(outputs) + return s.runs.SetNodeFinished(ctx, run.ID, node.Code, domain.LPNodeSucceeded, outputs, nil) +} + +// finishNodeFailed records the failure on the node and returns the cause. +func (s *LandscapeService) finishNodeFailed(ctx context.Context, runID uuid.UUID, node landscape.Node, cause error) error { + msg := cause.Error() + if err := s.runs.SetNodeFinished(ctx, runID, node.Code, domain.LPNodeFailed, nil, &msg); err != nil { + s.log.Error("mark landscape node failed", "run_id", runID, "node", node.Code, "error", err) + } + return cause +} + +// resolveInputs picks one dataset per input group: an explicit override first, +// then whatever this run already produced for the group, then the freshest +// ready raster of the group. +func (s *LandscapeService) resolveInputs(ctx context.Context, run domain.LPRun, node landscape.Node, state *runState) ([]domain.LPDatasetRef, []domain.Dataset, error) { + ready := domain.DatasetStatusReady + fileType := domain.FileTypeRaster + + refs := make([]domain.LPDatasetRef, 0, len(node.Inputs)) + datasets := make([]domain.Dataset, 0, len(node.Inputs)) + + for _, group := range node.Inputs { + var ( + d domain.Dataset + err error + ) + switch id, ok := s.pinnedInput(run, group, state); { + case ok: + d, err = s.datasets.GetByID(ctx, id) + default: + g, gerr := s.groups.GetByCode(ctx, group) + if gerr != nil { + return nil, nil, fmt.Errorf("node %s: dataset group %q: %w", node.Code, group, gerr) + } + d, err = s.datasets.LatestInGroup(ctx, g.ID, &ready, &fileType) + } + if err != nil { + if errors.Is(err, domain.ErrNotFound) { + return nil, nil, fmt.Errorf("node %s: no ready raster dataset in group %q", node.Code, group) + } + return nil, nil, err + } + refs = append(refs, domain.LPDatasetRef{Group: group, DatasetID: d.ID}) + datasets = append(datasets, d) + } + return refs, datasets, nil +} + +// pinnedInput returns the dataset already decided for a group: the run's +// override, or what an earlier node of this run produced. +func (s *LandscapeService) pinnedInput(run domain.LPRun, group string, state *runState) (uuid.UUID, bool) { + if id, ok := run.Overrides[group]; ok { + return id, true + } + return state.get(group) +} + +// reusableOutputs reports whether a previous succeeded node with the same +// fingerprint left outputs that all still exist and are ready. +func (s *LandscapeService) reusableOutputs(ctx context.Context, runID uuid.UUID, node landscape.Node, fp string) ([]domain.LPDatasetRef, bool, error) { + refs, err := s.runs.FindReusableOutputs(ctx, node.Code, fp, runID) + if err != nil { + if errors.Is(err, domain.ErrNotFound) { + return nil, false, nil + } + return nil, false, err + } + if len(refs) != len(node.Outputs) { + return nil, false, nil + } + for _, ref := range refs { + d, err := s.datasets.GetByID(ctx, ref.DatasetID) + if err != nil { + if errors.Is(err, domain.ErrNotFound) { + return nil, false, nil + } + return nil, false, err + } + if d.Status != domain.DatasetStatusReady { + return nil, false, nil + } + } + return refs, true, nil +} + +// computeNode downloads the node's inputs into a temp dir, computes it, and +// publishes one raster dataset per output group. Temp files are always removed. +func (s *LandscapeService) computeNode( + ctx context.Context, + run domain.LPRun, + node landscape.Node, + params landscape.Params, + cat domain.Category, + refs []domain.LPDatasetRef, + inputs []domain.Dataset, +) ([]domain.LPDatasetRef, error) { + // landscape.Compute does not create outDir, so we own the whole directory. + dir, err := os.MkdirTemp("", "lp-"+node.Code+"-*") + if err != nil { + return nil, fmt.Errorf("node %s: temp dir: %w", node.Code, err) + } + defer os.RemoveAll(dir) + + paths := make(map[string]string, len(inputs)) + for i, d := range inputs { + group := refs[i].Group + // Downstream nodes read the original object, never the COG, so they + // never wait for the conversion job. + path := filepath.Join(dir, "in-"+group+".tif") + if err := s.download(ctx, d.StorageKey, path); err != nil { + return nil, fmt.Errorf("node %s: download %q: %w", node.Code, group, err) + } + paths[group] = path + } + + outPaths, err := s.compute(ctx, s.rio, node, paths, dir, params) + if err != nil { + return nil, err + } + + start, end := periodOf(inputs) + date := time.Now() + if run.StartedAt != nil { + date = *run.StartedAt + } + + meta, err := json.Marshal(map[string]any{ + "lp_run_id": run.ID, + "lp_node": node.Code, + "inputs": refs, + }) + if err != nil { + return nil, err + } + + outputs := make([]domain.LPDatasetRef, 0, len(node.Outputs)) + for _, group := range node.Outputs { + path, ok := outPaths[group] + if !ok || path == "" { + return nil, fmt.Errorf("node %s: compute produced no output for group %q", node.Code, group) + } + g, err := s.groups.GetByCode(ctx, group) + if err != nil { + return nil, fmt.Errorf("node %s: dataset group %q: %w", node.Code, group, err) + } + title := group + if o, ok := landscape.OutputByCode(group); ok { + title = o.Title + } + d, err := s.publisher.CreateRasterFromFile(ctx, CreateRasterInput{ + CategoryID: cat.ID, + DatasetGroupID: &g.ID, + Name: title + " " + date.Format("2006-01-02"), + Filename: group + ".tif", + Path: path, + StartDate: start, + EndDate: end, + Meta: meta, + Automated: true, + }) + if err != nil { + return nil, fmt.Errorf("node %s: publish %q: %w", node.Code, group, err) + } + outputs = append(outputs, domain.LPDatasetRef{Group: group, DatasetID: d.ID}) + } + return outputs, nil +} + +// download copies an object from the store to a local path. +func (s *LandscapeService) download(ctx context.Context, key, path string) error { + obj, err := s.store.Get(ctx, key) + if err != nil { + return err + } + defer obj.Close() + + f, err := os.Create(path) + if err != nil { + return err + } + if _, err := io.Copy(f, obj); err != nil { + f.Close() + return err + } + return f.Close() +} + +// fingerprint is a stable hash of the node, its resolved inputs and the +// node-relevant params. +func fingerprint(node landscape.Node, inputs []domain.LPDatasetRef, p landscape.Params) string { + sorted := append([]domain.LPDatasetRef(nil), inputs...) + sort.Slice(sorted, func(i, j int) bool { return sorted[i].Group < sorted[j].Group }) + payload, _ := json.Marshal(struct { + Node string `json:"node"` + Inputs []domain.LPDatasetRef `json:"inputs"` + Params landscape.Params `json:"params"` + }{node.Code, sorted, p.ForNode(node.Code)}) + sum := sha256.Sum256(payload) + return hex.EncodeToString(sum[:]) +} + +// periodOf returns the min start_date and max end_date over the inputs, or nil +// when every input leaves the respective bound unset. +func periodOf(inputs []domain.Dataset) (*string, *string) { + var start, end *string + for _, d := range inputs { + if d.StartDate != nil && (start == nil || *d.StartDate < *start) { + start = d.StartDate + } + if d.EndDate != nil && (end == nil || *d.EndDate > *end) { + end = d.EndDate + } + } + return start, end +} + +// knownGroup reports whether a group code names an input slot or a node output. +func knownGroup(code string) bool { + if _, ok := landscape.SlotByCode(code); ok { + return true + } + _, ok := landscape.OutputByCode(code) + return ok +} + +// maxStage returns the highest stage number in the graph. +func maxStage() int { + highest := 0 + for _, n := range landscape.Nodes { + if n.Stage > highest { + highest = n.Stage + } + } + return highest +} + +// summaryOf projects a dataset onto the summary shape used in list responses. +func summaryOf(d domain.Dataset) domain.DatasetSummary { + return domain.DatasetSummary{ + ID: d.ID, + CategoryID: d.CategoryID, + DatasetGroupID: d.DatasetGroupID, + StartDate: d.StartDate, + EndDate: d.EndDate, + Name: d.Name, + Description: d.Description, + Unit: d.Unit, + FileType: d.FileType, + SizeBytes: d.SizeBytes, + Status: d.Status, + CreatedAt: d.CreatedAt, + UpdatedAt: d.UpdatedAt, + } +} diff --git a/internal/service/landscape_test.go b/internal/service/landscape_test.go new file mode 100644 index 0000000..10f04a1 --- /dev/null +++ b/internal/service/landscape_test.go @@ -0,0 +1,1161 @@ +package service + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "testing" + "time" + + "gis/internal/domain" + "gis/internal/landscape" + "gis/internal/raster" + + "github.com/google/uuid" +) + +// GetByCode resolves a category by its code so stubCategoryRepo also satisfies +// the categoryReader dependency the landscape service needs. +func (r *stubCategoryRepo) GetByCode(_ context.Context, code string) (domain.Category, error) { + for _, c := range r.store { + if c.Code == code { + return c, nil + } + } + return domain.Category{}, domain.ErrNotFound +} + +// lockedDatasetRepo serialises the dataset-repository calls the landscape +// service makes from concurrently executing nodes. stubDatasetRepo is a plain +// map, so without this the -race detector would flag the fan-out. +type lockedDatasetRepo struct { + *stubDatasetRepo + mu sync.Mutex +} + +func newLockedDatasetRepo() *lockedDatasetRepo { + return &lockedDatasetRepo{stubDatasetRepo: newStubDatasetRepo()} +} + +func (r *lockedDatasetRepo) Create(ctx context.Context, d domain.Dataset) (domain.Dataset, error) { + r.mu.Lock() + defer r.mu.Unlock() + return r.stubDatasetRepo.Create(ctx, d) +} + +func (r *lockedDatasetRepo) GetByID(ctx context.Context, id uuid.UUID) (domain.Dataset, error) { + r.mu.Lock() + defer r.mu.Unlock() + return r.stubDatasetRepo.GetByID(ctx, id) +} + +func (r *lockedDatasetRepo) LatestInGroup(ctx context.Context, groupID uuid.UUID, status *string, fileType *domain.FileType) (domain.Dataset, error) { + r.mu.Lock() + defer r.mu.Unlock() + return r.stubDatasetRepo.LatestInGroup(ctx, groupID, status, fileType) +} + +// stubLPRepo is an in-memory LandscapeRepository for tests. +type stubLPRepo struct { + mu sync.Mutex + runs map[uuid.UUID]*domain.LPRun + order []uuid.UUID + reuseLookups []string + clock int64 +} + +func newStubLPRepo() *stubLPRepo { + return &stubLPRepo{runs: map[uuid.UUID]*domain.LPRun{}} +} + +// next returns a strictly increasing timestamp so "most recent" ordering is +// deterministic regardless of wall-clock resolution. +func (r *stubLPRepo) next() time.Time { + r.clock++ + return time.Unix(0, r.clock) +} + +func copyRefs(in []domain.LPDatasetRef) []domain.LPDatasetRef { + if in == nil { + return nil + } + return append([]domain.LPDatasetRef(nil), in...) +} + +func copyRun(run domain.LPRun) domain.LPRun { + out := run + out.Overrides = map[string]uuid.UUID{} + for k, v := range run.Overrides { + out.Overrides[k] = v + } + out.Nodes = make([]domain.LPRunNode, len(run.Nodes)) + for i, n := range run.Nodes { + n.Inputs = copyRefs(n.Inputs) + n.Outputs = copyRefs(n.Outputs) + out.Nodes[i] = n + } + return out +} + +func (r *stubLPRepo) CreateRun(_ context.Context, run domain.LPRun) (domain.LPRun, error) { + r.mu.Lock() + defer r.mu.Unlock() + run.ID = uuid.New() + run.CreatedAt = r.next() + if run.Overrides == nil { + run.Overrides = map[string]uuid.UUID{} + } + nodes := make([]domain.LPRunNode, 0, len(run.Nodes)) + for _, n := range run.Nodes { + nodes = append(nodes, domain.LPRunNode{ + ID: uuid.New(), + RunID: run.ID, + NodeCode: n.NodeCode, + Stage: n.Stage, + Status: domain.LPNodePending, + Inputs: []domain.LPDatasetRef{}, + }) + } + run.Nodes = nodes + stored := copyRun(run) + r.runs[run.ID] = &stored + r.order = append(r.order, run.ID) + return copyRun(stored), nil +} + +func (r *stubLPRepo) GetRun(_ context.Context, id uuid.UUID) (domain.LPRun, error) { + r.mu.Lock() + defer r.mu.Unlock() + run, ok := r.runs[id] + if !ok { + return domain.LPRun{}, domain.ErrNotFound + } + out := copyRun(*run) + sort.Slice(out.Nodes, func(i, j int) bool { return out.Nodes[i].NodeCode < out.Nodes[j].NodeCode }) + return out, nil +} + +func (r *stubLPRepo) ListRuns(_ context.Context, limit, offset int) ([]domain.LPRun, error) { + r.mu.Lock() + defer r.mu.Unlock() + out := []domain.LPRun{} + // Newest first. + for i := len(r.order) - 1; i >= 0; i-- { + run := copyRun(*r.runs[r.order[i]]) + run.Nodes = nil + out = append(out, run) + } + if offset > len(out) { + offset = len(out) + } + out = out[offset:] + if limit >= 0 && limit < len(out) { + out = out[:limit] + } + return out, nil +} + +func (r *stubLPRepo) CountRuns(_ context.Context) (int, error) { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.runs), nil +} + +func (r *stubLPRepo) SetRunStatus(_ context.Context, id uuid.UUID, status domain.LPRunStatus, errMsg *string) error { + r.mu.Lock() + defer r.mu.Unlock() + run, ok := r.runs[id] + if !ok { + return domain.ErrNotFound + } + run.Status = status + run.Error = errMsg + switch status { + case domain.LPRunRunning: + t := r.next() + run.StartedAt = &t + case domain.LPRunSucceeded, domain.LPRunFailed: + t := r.next() + run.FinishedAt = &t + } + return nil +} + +func (r *stubLPRepo) node(runID uuid.UUID, code string) (*domain.LPRunNode, error) { + run, ok := r.runs[runID] + if !ok { + return nil, domain.ErrNotFound + } + for i := range run.Nodes { + if run.Nodes[i].NodeCode == code { + return &run.Nodes[i], nil + } + } + return nil, domain.ErrNotFound +} + +func (r *stubLPRepo) SetNodeStarted(_ context.Context, runID uuid.UUID, code, fingerprint string, inputs []domain.LPDatasetRef) error { + r.mu.Lock() + defer r.mu.Unlock() + n, err := r.node(runID, code) + if err != nil { + return err + } + n.Status = domain.LPNodeRunning + n.Fingerprint = fingerprint + n.Inputs = copyRefs(inputs) + n.Error = nil + t := r.next() + n.StartedAt = &t + return nil +} + +func (r *stubLPRepo) SetNodeFinished(_ context.Context, runID uuid.UUID, code string, status domain.LPNodeStatus, outputs []domain.LPDatasetRef, errMsg *string) error { + r.mu.Lock() + defer r.mu.Unlock() + n, err := r.node(runID, code) + if err != nil { + return err + } + n.Status = status + n.Outputs = copyRefs(outputs) + n.Error = errMsg + t := r.next() + n.FinishedAt = &t + return nil +} + +func (r *stubLPRepo) FindReusableOutputs(_ context.Context, code, fingerprint string, excludeRun uuid.UUID) ([]domain.LPDatasetRef, error) { + r.mu.Lock() + defer r.mu.Unlock() + r.reuseLookups = append(r.reuseLookups, code) + + var best *domain.LPRunNode + for _, run := range r.runs { + if run.ID == excludeRun { + continue + } + for i := range run.Nodes { + n := &run.Nodes[i] + if n.NodeCode != code || n.Fingerprint != fingerprint { + continue + } + if n.Status != domain.LPNodeSucceeded || n.Outputs == nil { + continue + } + if best == nil || (n.FinishedAt != nil && best.FinishedAt != nil && n.FinishedAt.After(*best.FinishedAt)) { + best = n + } + } + } + if best == nil { + return nil, domain.ErrNotFound + } + return copyRefs(best.Outputs), nil +} + +// lookups returns the node codes FindReusableOutputs was called for. +func (r *stubLPRepo) lookups() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.reuseLookups...) +} + +// stubPublisher implements datasetPublisher: it records the calls and inserts a +// ready raster dataset (as the COG conversion eventually would) so downstream +// nodes can resolve it. +type stubPublisher struct { + mu sync.Mutex + repo *lockedDatasetRepo + created []CreateRasterInput + uploaded []UploadInput + err error +} + +func (p *stubPublisher) CreateRasterFromFile(ctx context.Context, in CreateRasterInput) (domain.Dataset, error) { + p.mu.Lock() + if p.err != nil { + p.mu.Unlock() + return domain.Dataset{}, p.err + } + p.created = append(p.created, in) + p.mu.Unlock() + + // The compute step must have written the file the input points at. + if _, err := os.Stat(in.Path); err != nil { + return domain.Dataset{}, err + } + key := "landscape/" + uuid.NewString() + "/" + in.Filename + return p.repo.Create(ctx, domain.Dataset{ + CategoryID: in.CategoryID, + DatasetGroupID: in.DatasetGroupID, + Name: in.Name, + Filename: in.Filename, + StorageKey: key, + FileType: domain.FileTypeRaster, + ContentType: "image/tiff", + Status: domain.DatasetStatusReady, + Meta: in.Meta, + Automated: in.Automated, + StartDate: in.StartDate, + EndDate: in.EndDate, + CreatedAt: time.Now(), + }) +} + +func (p *stubPublisher) Upload(ctx context.Context, in UploadInput) (domain.Dataset, error) { + p.mu.Lock() + if p.err != nil { + p.mu.Unlock() + return domain.Dataset{}, p.err + } + p.uploaded = append(p.uploaded, in) + p.mu.Unlock() + return p.repo.Create(ctx, domain.Dataset{ + CategoryID: in.CategoryID, + DatasetGroupID: in.DatasetGroupID, + Name: in.Name, + Filename: in.Filename, + StorageKey: "landscape/" + uuid.NewString() + "/" + in.Filename, + FileType: in.FileType, + ContentType: in.ContentType, + Status: domain.DatasetStatusProcessing, + StartDate: in.StartDate, + EndDate: in.EndDate, + CreatedAt: time.Now(), + }) +} + +// stubLPJobs records enqueued landscape runs. +type stubLPJobs struct { + mu sync.Mutex + enqueued []uuid.UUID + err error +} + +func (j *stubLPJobs) EnqueueRun(_ context.Context, runID uuid.UUID) error { + j.mu.Lock() + defer j.mu.Unlock() + if j.err != nil { + return j.err + } + j.enqueued = append(j.enqueued, runID) + return nil +} + +// stubComputeCall records one invocation of the compute stub. +type stubComputeCall struct { + node string + inputs map[string]string + params landscape.Params +} + +// stubCompute stands in for landscape.Compute: it writes one empty file per +// declared output and records the inputs it was handed. +type stubCompute struct { + mu sync.Mutex + calls []stubComputeCall + errs map[string]error +} + +func newStubCompute() *stubCompute { return &stubCompute{errs: map[string]error{}} } + +func (c *stubCompute) fn(_ context.Context, _ raster.IO, node landscape.Node, inputs map[string]string, outDir string, p landscape.Params) (map[string]string, error) { + c.mu.Lock() + in := map[string]string{} + for k, v := range inputs { + in[k] = v + } + c.calls = append(c.calls, stubComputeCall{node: node.Code, inputs: in, params: p}) + err := c.errs[node.Code] + c.mu.Unlock() + if err != nil { + return nil, err + } + + out := map[string]string{} + for _, g := range node.Outputs { + path := filepath.Join(outDir, g+".tif") + if err := os.WriteFile(path, []byte("TIF"), 0o600); err != nil { + return nil, err + } + out[g] = path + } + return out, nil +} + +func (c *stubCompute) count() int { + c.mu.Lock() + defer c.mu.Unlock() + return len(c.calls) +} + +func (c *stubCompute) nodeCodes() []string { + c.mu.Lock() + defer c.mu.Unlock() + out := make([]string, 0, len(c.calls)) + for _, call := range c.calls { + out = append(out, call.node) + } + sort.Strings(out) + return out +} + +func (c *stubCompute) callFor(code string) (stubComputeCall, bool) { + c.mu.Lock() + defer c.mu.Unlock() + for _, call := range c.calls { + if call.node == code { + return call, true + } + } + return stubComputeCall{}, false +} + +func (c *stubCompute) reset() { + c.mu.Lock() + defer c.mu.Unlock() + c.calls = nil +} + +// lpFixture wires a LandscapeService over in-memory stubs. +type lpFixture struct { + svc *LandscapeService + runs *stubLPRepo + datasets *lockedDatasetRepo + groups *stubGroupRepo + cats *stubCategoryRepo + pub *stubPublisher + store *stubStore + jobs *stubLPJobs + compute *stubCompute + category domain.Category +} + +func newLPFixture(t *testing.T) *lpFixture { + t.Helper() + ctx := context.Background() + + cats := newStubCategoryRepo() + cat, err := cats.Create(ctx, domain.Category{Code: landscape.CategoryCode, Name: "Landscape"}) + if err != nil { + t.Fatalf("seed category: %v", err) + } + + groups := newStubGroupRepo() + for _, s := range landscape.Slots { + if _, err := groups.Create(ctx, domain.DatasetGroup{Code: s.Code, Name: s.Title}); err != nil { + t.Fatalf("seed group %s: %v", s.Code, err) + } + } + for _, o := range landscape.Outputs { + if _, err := groups.Create(ctx, domain.DatasetGroup{Code: o.Code, Name: o.Title}); err != nil { + t.Fatalf("seed group %s: %v", o.Code, err) + } + } + + datasets := newLockedDatasetRepo() + store := newStubStore() + pub := &stubPublisher{repo: datasets} + runs := newStubLPRepo() + jobs := &stubLPJobs{} + comp := newStubCompute() + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + + svc := NewLandscapeService(runs, datasets, groups, cats, pub, store, jobs, nil, comp.fn, log) + return &lpFixture{ + svc: svc, runs: runs, datasets: datasets, groups: groups, cats: cats, + pub: pub, store: store, jobs: jobs, compute: comp, category: cat, + } +} + +// groupID resolves a seeded group's id by code. +func (f *lpFixture) groupID(t *testing.T, code string) uuid.UUID { + t.Helper() + g, err := f.groups.GetByCode(context.Background(), code) + if err != nil { + t.Fatalf("group %s: %v", code, err) + } + return g.ID +} + +// seedDataset inserts a dataset into a group and stores its object bytes. +func (f *lpFixture) seedDataset(t *testing.T, code, status string, fileType domain.FileType, endDate string) domain.Dataset { + t.Helper() + ctx := context.Background() + gid := f.groupID(t, code) + key := "uploads/" + uuid.NewString() + "/" + code + ".tif" + end := endDate + start := "2020-01-01" + d, err := f.datasets.Create(ctx, domain.Dataset{ + CategoryID: f.category.ID, + DatasetGroupID: &gid, + Name: code, + Filename: code + ".tif", + StorageKey: key, + FileType: fileType, + ContentType: "image/tiff", + Status: status, + StartDate: &start, + EndDate: &end, + CreatedAt: time.Now(), + }) + if err != nil { + t.Fatalf("seed dataset %s: %v", code, err) + } + if err := f.store.Put(ctx, key, bytes.NewReader([]byte("TIF")), 3, "image/tiff"); err != nil { + t.Fatalf("seed object %s: %v", key, err) + } + return d +} + +// seedAllSlots fills every leaf slot with a ready raster. +func (f *lpFixture) seedAllSlots(t *testing.T, except ...string) { + t.Helper() + skip := map[string]bool{} + for _, c := range except { + skip[c] = true + } + for _, s := range landscape.Slots { + if skip[s.Code] { + continue + } + f.seedDataset(t, s.Code, domain.DatasetStatusReady, domain.FileTypeRaster, "2024-12-31") + } +} + +// runOnce creates a run and executes it, returning the final run state. +func (f *lpFixture) runOnce(t *testing.T, in RunInput) domain.LPRun { + t.Helper() + ctx := context.Background() + run, err := f.svc.CreateRun(ctx, in) + if err != nil { + t.Fatalf("CreateRun: %v", err) + } + execErr := f.svc.Execute(ctx, run.ID) + got, err := f.svc.GetRun(ctx, run.ID) + if err != nil { + t.Fatalf("GetRun: %v", err) + } + if execErr != nil && got.Status != domain.LPRunFailed { + t.Fatalf("Execute returned %v but run status is %s", execErr, got.Status) + } + return got +} + +// nodeStatuses maps node code -> status for a run. +func nodeStatuses(run domain.LPRun) map[string]domain.LPNodeStatus { + out := map[string]domain.LPNodeStatus{} + for _, n := range run.Nodes { + out[n.NodeCode] = n.Status + } + return out +} + +// nodeByCode finds a run node by its code. +func nodeByCode(t *testing.T, run domain.LPRun, code string) domain.LPRunNode { + t.Helper() + for _, n := range run.Nodes { + if n.NodeCode == code { + return n + } + } + t.Fatalf("run has no node %q", code) + return domain.LPRunNode{} +} + +// refsByGroup indexes dataset refs by their group code. +func refsByGroup(refs []domain.LPDatasetRef) map[string]uuid.UUID { + out := map[string]uuid.UUID{} + for _, r := range refs { + out[r.Group] = r.DatasetID + } + return out +} + +func TestLandscapeService_CreateRun_ValidatesParamsAndOverrides(t *testing.T) { + ctx := context.Background() + + stage := func(i int) *int { return &i } + + tests := []struct { + name string + setup func(t *testing.T, f *lpFixture) RunInput + }{ + { + name: "tie priority is not a permutation", + setup: func(_ *testing.T, _ *lpFixture) RunInput { + return RunInput{Params: landscape.Params{TiePriority: []int{1, 2, 3}}} + }, + }, + { + name: "window size out of range", + setup: func(_ *testing.T, _ *lpFixture) RunInput { + return RunInput{Params: landscape.Params{WindowSize: 10}} + }, + }, + { + name: "from_stage out of range", + setup: func(_ *testing.T, _ *lpFixture) RunInput { + return RunInput{FromStage: stage(4)} + }, + }, + { + name: "override for unknown group", + setup: func(t *testing.T, f *lpFixture) RunInput { + d := f.seedDataset(t, landscape.SlotZLCR, domain.DatasetStatusReady, domain.FileTypeRaster, "2024-12-31") + return RunInput{Overrides: map[string]uuid.UUID{"lp-not-a-group": d.ID}} + }, + }, + { + name: "override dataset does not exist", + setup: func(_ *testing.T, _ *lpFixture) RunInput { + return RunInput{Overrides: map[string]uuid.UUID{landscape.SlotZLCR: uuid.New()}} + }, + }, + { + name: "override dataset is not ready", + setup: func(t *testing.T, f *lpFixture) RunInput { + d := f.seedDataset(t, landscape.SlotZLCR, domain.DatasetStatusProcessing, domain.FileTypeRaster, "2024-12-31") + return RunInput{Overrides: map[string]uuid.UUID{landscape.SlotZLCR: d.ID}} + }, + }, + { + name: "override dataset is not a raster", + setup: func(t *testing.T, f *lpFixture) RunInput { + d := f.seedDataset(t, landscape.SlotZLCR, domain.DatasetStatusReady, domain.FileTypeVector, "2024-12-31") + return RunInput{Overrides: map[string]uuid.UUID{landscape.SlotZLCR: d.ID}} + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + f := newLPFixture(t) + in := tc.setup(t, f) + + _, err := f.svc.CreateRun(ctx, in) + if !errors.Is(err, domain.ErrValidation) { + t.Fatalf("expected ErrValidation, got %v", err) + } + if len(f.jobs.enqueued) != 0 { + t.Fatalf("expected no job to be enqueued, got %d", len(f.jobs.enqueued)) + } + }) + } + + t.Run("valid input queues a run with pending nodes", func(t *testing.T) { + f := newLPFixture(t) + override := f.seedDataset(t, landscape.SlotZLCR, domain.DatasetStatusReady, domain.FileTypeRaster, "2024-12-31") + + run, err := f.svc.CreateRun(ctx, RunInput{ + FromStage: stage(2), + Overrides: map[string]uuid.UUID{landscape.SlotZLCR: override.ID}, + Params: landscape.Params{V1MinSE12: 2}, + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if run.Status != domain.LPRunQueued { + t.Fatalf("status = %s, want queued", run.Status) + } + if len(run.Nodes) != len(landscape.Nodes) { + t.Fatalf("nodes = %d, want %d", len(run.Nodes), len(landscape.Nodes)) + } + for _, n := range run.Nodes { + if n.Status != domain.LPNodePending { + t.Fatalf("node %s status = %s, want pending", n.NodeCode, n.Status) + } + } + if run.FromStage == nil || *run.FromStage != 2 { + t.Fatalf("from_stage = %v, want 2", run.FromStage) + } + if got := run.Overrides[landscape.SlotZLCR]; got != override.ID { + t.Fatalf("override = %v, want %v", got, override.ID) + } + + // Params are persisted with the TZ defaults applied, so kernels never + // see a short tie_priority. + var stored landscape.Params + if err := json.Unmarshal(run.Params, &stored); err != nil { + t.Fatalf("params are not valid JSON: %v", err) + } + want := landscape.DefaultParams() + want.V1MinSE12 = 2 + if fmt.Sprint(stored) != fmt.Sprint(want) { + t.Fatalf("params = %+v, want %+v", stored, want) + } + + if len(f.jobs.enqueued) != 1 || f.jobs.enqueued[0] != run.ID { + t.Fatalf("enqueued = %v, want [%v]", f.jobs.enqueued, run.ID) + } + }) + + t.Run("enqueue failure fails the run", func(t *testing.T) { + f := newLPFixture(t) + f.jobs.err = errors.New("broker down") + + _, err := f.svc.CreateRun(ctx, RunInput{}) + if err == nil { + t.Fatal("expected an error") + } + runs, err := f.svc.ListRuns(ctx, 1, 10) + if err != nil { + t.Fatalf("ListRuns: %v", err) + } + if len(runs.Items) != 1 { + t.Fatalf("runs = %d, want 1", len(runs.Items)) + } + if runs.Items[0].Status != domain.LPRunFailed { + t.Fatalf("status = %s, want failed", runs.Items[0].Status) + } + }) +} + +func TestLandscapeService_Execute_ResolvesInputsAndPublishesOutputs(t *testing.T) { + f := newLPFixture(t) + f.seedAllSlots(t) + + run := f.runOnce(t, RunInput{}) + + if run.Status != domain.LPRunSucceeded { + t.Fatalf("status = %s (error %v), want succeeded", run.Status, run.Error) + } + for _, n := range run.Nodes { + if n.Status != domain.LPNodeSucceeded { + t.Fatalf("node %s status = %s (error %v), want succeeded", n.NodeCode, n.Status, n.Error) + } + } + if got := f.compute.count(); got != len(landscape.Nodes) { + t.Fatalf("compute calls = %d, want %d", got, len(landscape.Nodes)) + } + + // Node 10 consumes what nodes 00 and 01 produced in this very run. + n00 := refsByGroup(nodeByCode(t, run, "00").Outputs) + n01 := refsByGroup(nodeByCode(t, run, "01").Outputs) + n10in := refsByGroup(nodeByCode(t, run, "10").Inputs) + if n10in[landscape.OutZIntegrated] != n00[landscape.OutZIntegrated] { + t.Fatalf("node 10 z input = %v, want %v", n10in[landscape.OutZIntegrated], n00[landscape.OutZIntegrated]) + } + if n10in[landscape.OutUIntegrated] != n01[landscape.OutUIntegrated] { + t.Fatalf("node 10 u input = %v, want %v", n10in[landscape.OutUIntegrated], n01[landscape.OutUIntegrated]) + } + call, ok := f.compute.callFor("10") + if !ok { + t.Fatal("compute was not called for node 10") + } + for _, g := range []string{landscape.OutZIntegrated, landscape.OutUIntegrated} { + if call.inputs[g] == "" { + t.Fatalf("node 10 compute input %q is empty (got %v)", g, call.inputs) + } + } + + // Every output group got exactly one published dataset. + if len(f.pub.created) != len(landscape.Outputs) { + t.Fatalf("created datasets = %d, want %d", len(f.pub.created), len(landscape.Outputs)) + } + seen := map[string]bool{} + for _, in := range f.pub.created { + if !in.Automated { + t.Fatalf("dataset %q is not marked automated", in.Name) + } + if in.CategoryID != f.category.ID { + t.Fatalf("category = %v, want %v", in.CategoryID, f.category.ID) + } + if in.DatasetGroupID == nil { + t.Fatalf("dataset %q has no group", in.Name) + } + group := "" + for _, o := range landscape.Outputs { + if f.groupID(t, o.Code) == *in.DatasetGroupID { + group = o.Code + } + } + if group == "" { + t.Fatalf("dataset %q is filed under an unknown group %v", in.Name, *in.DatasetGroupID) + } + if in.Filename != group+".tif" { + t.Fatalf("filename = %q, want %q", in.Filename, group+".tif") + } + seen[group] = true + + var meta map[string]any + if err := json.Unmarshal(in.Meta, &meta); err != nil { + t.Fatalf("meta of %q is not valid JSON: %v", in.Name, err) + } + if meta["lp_run_id"] != run.ID.String() { + t.Fatalf("meta lp_run_id = %v, want %v", meta["lp_run_id"], run.ID) + } + if _, ok := meta["lp_node"]; !ok { + t.Fatalf("meta of %q has no lp_node: %v", in.Name, meta) + } + if _, ok := meta["inputs"]; !ok { + t.Fatalf("meta of %q has no inputs: %v", in.Name, meta) + } + if in.EndDate == nil || *in.EndDate != "2024-12-31" { + t.Fatalf("end_date = %v, want 2024-12-31", in.EndDate) + } + } + if len(seen) != len(landscape.Outputs) { + t.Fatalf("groups covered = %d, want %d", len(seen), len(landscape.Outputs)) + } +} + +func TestLandscapeService_Execute_ReusesFingerprintMatches(t *testing.T) { + f := newLPFixture(t) + f.seedAllSlots(t) + + first := f.runOnce(t, RunInput{}) + if first.Status != domain.LPRunSucceeded { + t.Fatalf("first run status = %s (%v)", first.Status, first.Error) + } + f.compute.reset() + + second := f.runOnce(t, RunInput{}) + if second.Status != domain.LPRunSucceeded { + t.Fatalf("second run status = %s (%v)", second.Status, second.Error) + } + if got := f.compute.count(); got != 0 { + t.Fatalf("compute calls on rerun = %d, want 0 (%v)", got, f.compute.nodeCodes()) + } + if got := f.runs.lookups(); len(got) != 2*len(landscape.Nodes) { + t.Fatalf("reuse lookups = %d, want %d (once per node per run)", len(got), 2*len(landscape.Nodes)) + } + for _, n := range second.Nodes { + if n.Status != domain.LPNodeSkipped { + t.Fatalf("node %s status = %s, want skipped", n.NodeCode, n.Status) + } + } + for _, n := range second.Nodes { + want := refsByGroup(nodeByCode(t, first, n.NodeCode).Outputs) + got := refsByGroup(n.Outputs) + if len(got) != len(want) { + t.Fatalf("node %s outputs = %v, want %v", n.NodeCode, got, want) + } + for g, id := range want { + if got[g] != id { + t.Fatalf("node %s output %s = %v, want %v", n.NodeCode, g, got[g], id) + } + } + } +} + +func TestLandscapeService_Execute_RecomputesDownstreamOfChangedInput(t *testing.T) { + f := newLPFixture(t) + f.seedAllSlots(t) + + if first := f.runOnce(t, RunInput{}); first.Status != domain.LPRunSucceeded { + t.Fatalf("first run status = %s (%v)", first.Status, first.Error) + } + f.compute.reset() + + // A fresher P-crop raster only affects the C conflicts and P max nodes and + // everything downstream of them. + f.seedDataset(t, landscape.SlotPCrop, domain.DatasetStatusReady, domain.FileTypeRaster, "2025-06-30") + + second := f.runOnce(t, RunInput{}) + if second.Status != domain.LPRunSucceeded { + t.Fatalf("second run status = %s (%v)", second.Status, second.Error) + } + + statuses := nodeStatuses(second) + for _, code := range []string{"00", "01", "02", "03", "10"} { + if statuses[code] != domain.LPNodeSkipped { + t.Fatalf("node %s status = %s, want skipped", code, statuses[code]) + } + } + for _, code := range []string{"20", "21", "30", "31"} { + if statuses[code] != domain.LPNodeSucceeded { + t.Fatalf("node %s status = %s, want succeeded", code, statuses[code]) + } + } + want := []string{"20", "21", "30", "31"} + got := f.compute.nodeCodes() + if fmt.Sprint(got) != fmt.Sprint(want) { + t.Fatalf("recomputed nodes = %v, want %v", got, want) + } +} + +func TestLandscapeService_Execute_FromStageForcesRecompute(t *testing.T) { + f := newLPFixture(t) + f.seedAllSlots(t) + + if first := f.runOnce(t, RunInput{}); first.Status != domain.LPRunSucceeded { + t.Fatalf("first run status = %s (%v)", first.Status, first.Error) + } + f.compute.reset() + + from := 2 + second := f.runOnce(t, RunInput{FromStage: &from}) + if second.Status != domain.LPRunSucceeded { + t.Fatalf("second run status = %s (%v)", second.Status, second.Error) + } + + statuses := nodeStatuses(second) + for _, code := range []string{"00", "01", "02", "03", "10"} { + if statuses[code] != domain.LPNodeSkipped { + t.Fatalf("node %s status = %s, want skipped", code, statuses[code]) + } + } + for _, code := range []string{"20", "21", "30", "31"} { + if statuses[code] != domain.LPNodeSucceeded { + t.Fatalf("node %s status = %s, want succeeded", code, statuses[code]) + } + } + if got, want := f.compute.nodeCodes(), []string{"20", "21", "30", "31"}; fmt.Sprint(got) != fmt.Sprint(want) { + t.Fatalf("recomputed nodes = %v, want %v", got, want) + } +} + +func TestLandscapeService_Execute_MissingSlotFailsRun(t *testing.T) { + f := newLPFixture(t) + f.seedAllSlots(t, landscape.SlotZWR) + + run := f.runOnce(t, RunInput{}) + + if run.Status != domain.LPRunFailed { + t.Fatalf("status = %s, want failed", run.Status) + } + if run.Error == nil || !strings.Contains(*run.Error, landscape.SlotZWR) { + t.Fatalf("run error = %v, want it to name %q", run.Error, landscape.SlotZWR) + } + statuses := nodeStatuses(run) + for _, code := range []string{"00", "02"} { + if statuses[code] != domain.LPNodeFailed { + t.Fatalf("node %s status = %s, want failed", code, statuses[code]) + } + n := nodeByCode(t, run, code) + if n.Error == nil || !strings.Contains(*n.Error, landscape.SlotZWR) { + t.Fatalf("node %s error = %v, want it to name %q", code, n.Error, landscape.SlotZWR) + } + } + // Stage 1 and beyond are never started. + for _, code := range []string{"10", "20", "21", "30", "31"} { + if statuses[code] != domain.LPNodePending { + t.Fatalf("node %s status = %s, want pending", code, statuses[code]) + } + } +} + +func TestLandscapeService_UploadToSlot_UnknownSlot(t *testing.T) { + f := newLPFixture(t) + + _, err := f.svc.UploadToSlot(context.Background(), "lp-not-a-slot", SlotUploadInput{ + Filename: "x.tif", + ContentType: "image/tiff", + Size: 3, + Reader: bytes.NewReader([]byte("TIF")), + }) + if !errors.Is(err, domain.ErrNotFound) { + t.Fatalf("expected ErrNotFound, got %v", err) + } + if len(f.pub.uploaded) != 0 { + t.Fatalf("expected no upload, got %d", len(f.pub.uploaded)) + } +} + +func TestLandscapeService_UploadToSlot_KnownSlot(t *testing.T) { + f := newLPFixture(t) + ctx := context.Background() + + got, err := f.svc.UploadToSlot(ctx, landscape.SlotPSnow, SlotUploadInput{ + Filename: "snow.tif", + ContentType: "image/tiff", + Size: 3, + Reader: bytes.NewReader([]byte("TIF")), + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(f.pub.uploaded) != 1 { + t.Fatalf("uploads = %d, want 1", len(f.pub.uploaded)) + } + in := f.pub.uploaded[0] + if in.FileType != domain.FileTypeRaster { + t.Fatalf("file_type = %s, want raster", in.FileType) + } + if in.CategoryID != f.category.ID { + t.Fatalf("category = %v, want %v", in.CategoryID, f.category.ID) + } + gid := f.groupID(t, landscape.SlotPSnow) + if in.DatasetGroupID == nil || *in.DatasetGroupID != gid { + t.Fatalf("group = %v, want %v", in.DatasetGroupID, gid) + } + if in.Name != "snow.tif" { + t.Fatalf("name = %q, want the filename as a default", in.Name) + } + if got.ID == uuid.Nil { + t.Fatal("returned dataset has no id") + } +} + +func TestLandscapeService_Slots_ListsLatestPerSlot(t *testing.T) { + f := newLPFixture(t) + // Any status counts for a slot, not just ready. + seeded := f.seedDataset(t, landscape.SlotUSoil, domain.DatasetStatusProcessing, domain.FileTypeRaster, "2024-12-31") + + slots, err := f.svc.Slots(context.Background()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(slots) != len(landscape.Slots) { + t.Fatalf("slots = %d, want %d", len(slots), len(landscape.Slots)) + } + for _, s := range slots { + switch s.Code { + case landscape.SlotUSoil: + if s.Latest == nil || s.Latest.ID != seeded.ID { + t.Fatalf("latest for %s = %v, want %v", s.Code, s.Latest, seeded.ID) + } + if s.GroupID != f.groupID(t, landscape.SlotUSoil) { + t.Fatalf("group id = %v", s.GroupID) + } + default: + if s.Latest != nil { + t.Fatalf("latest for %s = %v, want nil", s.Code, s.Latest) + } + } + } +} + +func TestLandscapeService_Results_ListsLatestPerOutput(t *testing.T) { + f := newLPFixture(t) + ctx := context.Background() + + ready := f.seedDataset(t, landscape.OutE, domain.DatasetStatusReady, domain.FileTypeRaster, "2024-12-31") + // A not-yet-converted result must not be served as the latest result. + f.seedDataset(t, landscape.OutV1, domain.DatasetStatusProcessing, domain.FileTypeRaster, "2025-01-01") + + eGroup := f.groupID(t, landscape.OutE) + if _, err := f.groups.SetSLD(ctx, eGroup, "groups/e/sld/e.sld", "e.sld", "application/xml"); err != nil { + t.Fatalf("SetSLD: %v", err) + } + + results, err := f.svc.Results(ctx) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(results) != len(landscape.Outputs) { + t.Fatalf("results = %d, want %d", len(results), len(landscape.Outputs)) + } + + byGroup := map[string]ResultInfo{} + for _, r := range results { + byGroup[r.Group] = r + } + + e := byGroup[landscape.OutE] + if e.Latest == nil || e.Latest.ID != ready.ID { + t.Fatalf("latest for %s = %v, want %v", landscape.OutE, e.Latest, ready.ID) + } + if e.Node != "10" { + t.Fatalf("node for %s = %q, want 10", landscape.OutE, e.Node) + } + if e.Title != "E Ecological Constraint" { + t.Fatalf("title = %q", e.Title) + } + wantCog := "/datasets/" + ready.ID.String() + ".cog" + if e.CogURL == nil || *e.CogURL != wantCog { + t.Fatalf("cog_url = %v, want %q", e.CogURL, wantCog) + } + wantSld := "/dataset-groups/" + eGroup.String() + "/sld" + if e.SldURL == nil || *e.SldURL != wantSld { + t.Fatalf("sld_url = %v, want %q", e.SldURL, wantSld) + } + + v1 := byGroup[landscape.OutV1] + if v1.Latest != nil { + t.Fatalf("latest for %s = %v, want nil (not ready)", landscape.OutV1, v1.Latest) + } + if v1.CogURL != nil || v1.SldURL != nil { + t.Fatalf("urls for %s = %v/%v, want nil", landscape.OutV1, v1.CogURL, v1.SldURL) + } + + zmax := byGroup[landscape.OutZMax] + if zmax.SldURL != nil { + t.Fatalf("sld_url for a group without an SLD = %v, want nil", zmax.SldURL) + } +} + +func TestLandscapeService_MarkRunFailed(t *testing.T) { + f := newLPFixture(t) + ctx := context.Background() + + run, err := f.svc.CreateRun(ctx, RunInput{}) + if err != nil { + t.Fatalf("CreateRun: %v", err) + } + if err := f.svc.MarkRunFailed(ctx, run.ID, "worker died"); err != nil { + t.Fatalf("MarkRunFailed: %v", err) + } + got, err := f.svc.GetRun(ctx, run.ID) + if err != nil { + t.Fatalf("GetRun: %v", err) + } + if got.Status != domain.LPRunFailed { + t.Fatalf("status = %s, want failed", got.Status) + } + if got.Error == nil || *got.Error != "worker died" { + t.Fatalf("error = %v, want %q", got.Error, "worker died") + } +} + +func TestLandscapeService_Execute_NodeFailureFailsRunAndCleansUp(t *testing.T) { + tests := []struct { + name string + arrange func(f *lpFixture) + want string + }{ + { + name: "compute fails", + arrange: func(f *lpFixture) { f.compute.errs["00"] = errors.New("grid mismatch") }, + want: "grid mismatch", + }, + { + name: "publishing the result fails", + arrange: func(f *lpFixture) { f.pub.err = errors.New("minio down") }, + want: "minio down", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + f := newLPFixture(t) + f.seedAllSlots(t) + tc.arrange(f) + + run := f.runOnce(t, RunInput{}) + + if run.Status != domain.LPRunFailed { + t.Fatalf("status = %s, want failed", run.Status) + } + if run.Error == nil || !strings.Contains(*run.Error, tc.want) { + t.Fatalf("run error = %v, want it to mention %q", run.Error, tc.want) + } + n00 := nodeByCode(t, run, "00") + if n00.Status != domain.LPNodeFailed { + t.Fatalf("node 00 status = %s, want failed", n00.Status) + } + if n00.Error == nil || !strings.Contains(*n00.Error, tc.want) { + t.Fatalf("node 00 error = %v, want it to mention %q", n00.Error, tc.want) + } + // Temp files are always cleaned up, even on failure. + for _, call := range f.compute.calls { + for _, path := range call.inputs { + if _, err := os.Stat(filepath.Dir(path)); !os.IsNotExist(err) { + t.Fatalf("temp dir %q was not removed (stat err %v)", filepath.Dir(path), err) + } + } + } + }) + } +}