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, } }