From 91c2cab66ee695fd944cd825c05b3546e7704987 Mon Sep 17 00:00:00 2001 From: Bakhtiyar Issakhmetov Date: Sat, 5 Sep 2026 13:08:25 +0500 Subject: [PATCH] docs: Landscape Program implementation plan Co-Authored-By: Claude Fable 5.1 --- .../plans/2026-09-05-landscape-program.md | 2661 +++++++++++++++++ 1 file changed, 2661 insertions(+) create mode 100644 docs/superpowers/plans/2026-09-05-landscape-program.md diff --git a/docs/superpowers/plans/2026-09-05-landscape-program.md b/docs/superpowers/plans/2026-09-05-landscape-program.md new file mode 100644 index 0000000..e1ce1d6 --- /dev/null +++ b/docs/superpowers/plans/2026-09-05-landscape-program.md @@ -0,0 +1,2661 @@ +# Landscape Program + Dataset Groups Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add dataset groups + dataset periods (with the `date`→`start_date` rename), and the Landscape Program raster pipeline (godal-backed, incremental, results stored as raster datasets). + +**Architecture:** Part 1 extends the existing dataset stack (domain → postgres repo → service → chi handler) with a `dataset_groups` entity and period columns. Part 2 adds a pure-Go compute engine (`internal/landscape`: graph, per-window kernels, `Compute`) behind a swappable `raster.IO` interface (godal implementation), and a `LandscapeService` that plans runs (fingerprint-based skip/reuse), resolves inputs from dataset groups, downloads inputs, computes, and publishes outputs as raster datasets that the existing COG conversion picks up. A RabbitMQ consumer executes runs in the worker. + +**Tech Stack:** Go 1.26 (toolchain go1.26.2 pinned), chi, pgx, goose, MinIO, RabbitMQ, GDAL via `github.com/airbusgeo/godal` v0.0.18 (cgo) + existing GDAL CLI tools. + +**Spec:** `docs/superpowers/specs/2026-09-05-landscape-program-design.md` + +## Global Constraints + +- Build with `CGO_ENABLED=1`; local GDAL is Homebrew 3.13.3 (`pkg-config --modversion gdal`). The local Go SDK is incomplete; `go.mod` pins `toolchain go1.26.2`, so plain `go test ./...` works. +- Follow existing layering: `internal/domain` (no internal deps) → `internal/repository/postgres` → `internal/service` (interfaces declared in service) → `internal/transport/http`. Wiring only in `internal/app/app.go`. +- Errors: repositories return `domain.ErrNotFound` / `domain.ErrConflict` / `domain.ErrValidation` (wrapped with `%w`); handlers use `respondDomainError`. +- Dates in JSON are strings `YYYY-MM-DD` (`time.Parse("2006-01-02", …)`). +- Output rasters: `int16`, nodata `-9999`, `TILED=YES`, `BLOCKXSIZE=512`, `BLOCKYSIZE=512`, `COMPRESS=DEFLATE`, `PREDICTOR=2`, `BIGTIFF=IF_SAFER`. +- Pixel validity: a value is valid iff `1 <= v <= 5`. Anything else (including nodata) is invalid. Never resample; grids must match exactly or the node fails. +- Group codes (slots and outputs) are fixed strings listed in Task 4; they are the single source of truth for both the graph and the seed migration. +- Tests: table-driven `testing` package, stubs like `internal/service/dataset_test.go`. Run only your own packages while other agents work in parallel (`go test ./internal/landscape/...`), not `go test ./...`, unless you are the integration task. +- **Parallel execution rule:** several agents edit the same working tree on branch `feature/landscape-program`. Only touch the files listed in your task. Never run `git add -A`/`git add .`; add explicit paths. If `git commit` fails with `index.lock`, wait 2 s and retry. + +## Execution waves + +| Wave | Tasks (parallel within a wave) | Depends on | +|---|---|---| +| 1 | Task 1 → Task 2 (same lane, sequential); Task 3; Task 4; Task 5 | — | +| 2 | Task 6; Task 7 | 3, 4 (Task 6); 2, 4, 5 (Task 7) | +| 3 | Task 8 | 6, 7 | + +--- + +## File structure + +**Part 1** +- `migrations/00010_dataset_groups_and_dates.sql` — new table, dataset columns, observation rename, year_columns key rename. +- `internal/domain/dataset.go` — modify: rename fields, add group/date fields, `DatasetFields`, filter fields. +- `internal/domain/dataset_group.go` — create: `DatasetGroup`. +- `internal/repository/postgres/dataset.go` — modify: columns, filter clause, `UpdateFields(DatasetFields)`, `LatestInGroup`, observation columns. +- `internal/repository/postgres/dataset_group.go` — create: group repository. +- `internal/service/dataset.go` — modify: rename, upload/update inputs, `CreateRasterFromFile`, SLD fallback. +- `internal/service/dataset_group.go` — create: group service. +- `internal/service/dataset_test.go` — modify: stubs + tests. `internal/service/dataset_group_test.go` — create. +- `internal/transport/http/dataset_handler.go` — modify. `internal/transport/http/dataset_group_handler.go` — create. `internal/transport/http/router.go`, `internal/app/app.go` — modify (wiring). +- `api/openapi.yaml`, `README.md` — modify. + +**Part 2** +- `internal/raster/io.go` (interfaces), `internal/raster/godal.go` (+ `godal_test.go`), `build/package/Dockerfile`, `go.mod`. +- `internal/landscape/graph.go`, `params.go`, `kernels.go` (+ tests), `compute.go` (+ `compute_test.go` with a fake IO). +- `internal/domain/landscape.go`, `migrations/00011_landscape.sql`, `internal/repository/postgres/landscape.go`. +- `internal/service/landscape.go` (+ test), `internal/messaging/rabbitmq/landscape_consumer.go`. +- `internal/transport/http/landscape_handler.go`, `router.go`, `app.go`, `api/openapi.yaml`, `README.md`. + +--- + +### Task 1: Dataset periods — rename `date`→`start_date`, add dataset `start_date`/`end_date`, list filters, upload/PUT/PATCH fields + +**Files:** +- Create: `migrations/00010_dataset_groups_and_dates.sql` (the *whole* migration incl. the `dataset_groups` table so Task 2 does not need another migration) +- Modify: `internal/domain/dataset.go`, `internal/repository/postgres/dataset.go`, `internal/service/dataset.go`, `internal/service/dataset_test.go`, `internal/transport/http/dataset_handler.go`, `api/openapi.yaml`, `README.md` + +**Interfaces:** +- Produces: `domain.Observation{StartDate string; EndDate *string}`, `domain.YearColumn{Column, StartDate string; EndDate *string}`, `domain.Dataset/DatasetSummary{DatasetGroupID *uuid.UUID; StartDate, EndDate *string}`, `domain.DatasetFilter{…, DatasetGroupID *uuid.UUID; StartDateFrom, StartDateTo, EndDateFrom, EndDateTo *string}`, `domain.DatasetFields`, repo `UpdateFields(ctx, id uuid.UUID, f domain.DatasetFields) (domain.Dataset, error)`, service `UploadInput{…; DatasetGroupID *uuid.UUID; StartDate, EndDate *string}`, `DatasetUpdateInput`/`DatasetPatchInput` with the same three fields (+ `DatasetGroupIDSet`, `StartDateSet`, `EndDateSet` on patch). +- Task 2 relies on `dg_id` column + `DatasetGroupID` field existing after this task, and on `domain.ValidateDatePeriod`. + +- [ ] **Step 1: Write the migration** + +```sql +-- migrations/00010_dataset_groups_and_dates.sql +-- +goose Up +-- Dataset groups: "одноимённые" datasets versioned by period. A group may carry a +-- shared SLD used as a fallback for its datasets. +CREATE TABLE dataset_groups ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + code VARCHAR(255) NOT NULL UNIQUE, + name VARCHAR(255) NOT NULL, + description TEXT NOT NULL DEFAULT '', + sld_storage_key TEXT, + sld_filename VARCHAR(255), + sld_content_type VARCHAR(255), + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +ALTER TABLE datasets ADD COLUMN dg_id UUID REFERENCES dataset_groups (id) ON DELETE RESTRICT; +ALTER TABLE datasets ADD COLUMN start_date DATE; +ALTER TABLE datasets ADD COLUMN end_date DATE; +ALTER TABLE datasets ADD CONSTRAINT chk_datasets_period + CHECK (start_date IS NULL OR end_date IS NULL OR end_date >= start_date); +CREATE INDEX idx_datasets_dg_id_end_date ON datasets (dg_id, end_date DESC NULLS LAST, created_at DESC); +CREATE INDEX idx_datasets_start_date ON datasets (start_date); +CREATE INDEX idx_datasets_end_date ON datasets (end_date); + +-- Period columns are renamed everywhere: observations, and the year_columns +-- mapping JSON ({column, date, date_to} -> {column, start_date, end_date}). +ALTER TABLE dataset_observations RENAME COLUMN date TO start_date; +ALTER TABLE dataset_observations RENAME COLUMN date_to TO end_date; +DROP INDEX uq_dataset_observations; +DROP INDEX idx_dataset_observations_kato_date; +CREATE UNIQUE INDEX uq_dataset_observations ON dataset_observations (dataset_id, kato_code, start_date, COALESCE(end_date, start_date)); +CREATE INDEX idx_dataset_observations_kato_date ON dataset_observations (kato_code, start_date); + +UPDATE datasets +SET year_columns = ( + SELECT jsonb_agg( + (elem - 'date' - 'date_to') + || jsonb_build_object('start_date', elem->'date') + || CASE WHEN elem ? 'date_to' AND elem->'date_to' IS NOT NULL + THEN jsonb_build_object('end_date', elem->'date_to') ELSE '{}'::jsonb END) + FROM jsonb_array_elements(year_columns) AS elem) +WHERE year_columns IS NOT NULL AND jsonb_typeof(year_columns) = 'array' AND jsonb_array_length(year_columns) > 0; + +-- +goose Down +UPDATE datasets +SET year_columns = ( + SELECT jsonb_agg( + (elem - 'start_date' - 'end_date') + || jsonb_build_object('date', elem->'start_date') + || CASE WHEN elem ? 'end_date' AND elem->'end_date' IS NOT NULL + THEN jsonb_build_object('date_to', elem->'end_date') ELSE '{}'::jsonb END) + FROM jsonb_array_elements(year_columns) AS elem) +WHERE year_columns IS NOT NULL AND jsonb_typeof(year_columns) = 'array' AND jsonb_array_length(year_columns) > 0; +DROP INDEX uq_dataset_observations; +DROP INDEX idx_dataset_observations_kato_date; +ALTER TABLE dataset_observations RENAME COLUMN start_date TO date; +ALTER TABLE dataset_observations RENAME COLUMN end_date TO date_to; +CREATE UNIQUE INDEX uq_dataset_observations ON dataset_observations (dataset_id, kato_code, date, COALESCE(date_to, date)); +CREATE INDEX idx_dataset_observations_kato_date ON dataset_observations (kato_code, date); +DROP INDEX idx_datasets_dg_id_end_date; +DROP INDEX idx_datasets_start_date; +DROP INDEX idx_datasets_end_date; +ALTER TABLE datasets DROP CONSTRAINT chk_datasets_period; +ALTER TABLE datasets DROP COLUMN dg_id; +ALTER TABLE datasets DROP COLUMN start_date; +ALTER TABLE datasets DROP COLUMN end_date; +DROP TABLE dataset_groups; +``` + +- [ ] **Step 2: Domain — rename and add fields** (`internal/domain/dataset.go`) + +Rename `Observation.Date`→`StartDate` (`json:"start_date"`), `Observation.DateTo`→`EndDate` (`json:"end_date,omitempty"`), `YearColumn.Date`→`StartDate` (`json:"start_date"`), `YearColumn.DateTo`→`EndDate` (`json:"end_date,omitempty"`). Update doc comments accordingly. Add to both `Dataset` and `DatasetSummary` (after `CategoryID`): + +```go + // DatasetGroupID links the dataset to its dataset group ("одноимённые" + // datasets versioned by period). Nullable. + DatasetGroupID *uuid.UUID `json:"dg_id"` + // StartDate / EndDate is the period the dataset describes (YYYY-MM-DD). + // Both nullable; EndDate drives the "freshest in group" rule. + StartDate *string `json:"start_date"` + EndDate *string `json:"end_date"` +``` + +Add to `DatasetFilter` (document: dates are inclusive `YYYY-MM-DD` bounds): + +```go + DatasetGroupID *uuid.UUID + StartDateFrom *string + StartDateTo *string + EndDateFrom *string + EndDateTo *string +``` + +Add: + +```go +// DatasetFields is the set of editable metadata columns written by +// UpdateFields. Nil pointers clear the corresponding nullable column. +type DatasetFields struct { + CategoryID uuid.UUID + Name string + Description *string + Unit *string + DatasetGroupID *uuid.UUID + StartDate *string + EndDate *string +} + +// ValidateDatePeriod checks that start/end (YYYY-MM-DD, either may be nil) +// parse and that end is not before start. Returns an ErrValidation-wrapped +// error naming the offending field. +func ValidateDatePeriod(start, end *string) error { + var from, to time.Time + var err error + if start != nil { + if from, err = time.Parse("2006-01-02", *start); err != nil { + return fmt.Errorf("%w: start_date %q is not YYYY-MM-DD", ErrValidation, *start) + } + } + if end != nil { + if to, err = time.Parse("2006-01-02", *end); err != nil { + return fmt.Errorf("%w: end_date %q is not YYYY-MM-DD", ErrValidation, *end) + } + } + if start != nil && end != nil && to.Before(from) { + return fmt.Errorf("%w: end_date %q is before start_date %q", ErrValidation, *end, *start) + } + return nil +} +``` + +- [ ] **Step 3: Domain test** (`internal/domain/dataset_test.go`, append) + +```go +func TestValidateDatePeriod(t *testing.T) { + s := func(v string) *string { return &v } + cases := []struct { + name string + start, end *string + wantErr bool + }{ + {"both nil", nil, nil, false}, + {"start only", s("2020-01-01"), nil, false}, + {"ordered", s("2020-01-01"), s("2020-12-31"), false}, + {"same day", s("2020-01-01"), s("2020-01-01"), false}, + {"reversed", s("2020-12-31"), s("2020-01-01"), true}, + {"bad start", s("01.01.2020"), nil, true}, + {"bad end", nil, s("2020-13-01"), true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := ValidateDatePeriod(tc.start, tc.end) + if (err != nil) != tc.wantErr { + t.Fatalf("err=%v wantErr=%v", err, tc.wantErr) + } + if err != nil && !errors.Is(err, ErrValidation) { + t.Fatalf("expected ErrValidation, got %v", err) + } + }) + } +} +``` + +Run: `go test ./internal/domain/ -run TestValidateDatePeriod -v` → PASS after Step 2. + +- [ ] **Step 4: Repository** (`internal/repository/postgres/dataset.go`) + +`datasetColumns`: insert `dg_id, to_char(start_date, 'YYYY-MM-DD') AS start_date, to_char(end_date, 'YYYY-MM-DD') AS end_date,` right after `category_id,`; `scanDataset` scans `&d.DatasetGroupID, &d.StartDate, &d.EndDate` in the same position. Same for `datasetSummaryColumns`/`scanDatasetSummary`. + +`Create`: insert `dg_id, start_date, end_date` (`$14,$15,$16`, values `d.DatasetGroupID, d.StartDate, d.EndDate`; pgx maps `*string` → `date` text fine, use `$15::date, $16::date` casts). + +Replace `UpdateFields` with: + +```go +// UpdateFields overwrites a dataset's editable metadata columns. Nil pointers +// clear the nullable columns. Returns domain.ErrNotFound if the dataset does +// not exist, or domain.ErrConflict if the category/group does not exist. +func (r *DatasetRepository) UpdateFields(ctx context.Context, id uuid.UUID, f domain.DatasetFields) (domain.Dataset, error) { + row := r.pool.QueryRow(ctx, + `UPDATE datasets + SET category_id = $2, name = $3, description = $4, unit = $5, + dg_id = $6, start_date = $7::date, end_date = $8::date, updated_at = now() + WHERE id = $1 + RETURNING `+datasetColumns, + id, f.CategoryID, f.Name, f.Description, f.Unit, f.DatasetGroupID, f.StartDate, f.EndDate, + ) + out, err := scanDataset(row) + return out, mapError(err) +} +``` + +`datasetFilterClause`: append + +```go + if f.DatasetGroupID != nil { + args = append(args, *f.DatasetGroupID) + conds = append(conds, fmt.Sprintf("dg_id = $%d", len(args))) + } + for _, b := range []struct { + col, op string + val *string + }{ + {"start_date", ">=", f.StartDateFrom}, {"start_date", "<=", f.StartDateTo}, + {"end_date", ">=", f.EndDateFrom}, {"end_date", "<=", f.EndDateTo}, + } { + if b.val != nil { + args = append(args, *b.val) + conds = append(conds, fmt.Sprintf("%s %s $%d::date", b.col, b.op, len(args))) + } + } +``` + +Add: + +```go +// LatestInGroup returns the freshest dataset of a group: greatest end_date +// (NULL end_date sorts last), ties broken by created_at DESC. status and +// fileType optionally restrict candidates. Returns domain.ErrNotFound when no +// dataset matches. +func (r *DatasetRepository) LatestInGroup(ctx context.Context, groupID uuid.UUID, status *string, fileType *domain.FileType) (domain.Dataset, error) { + where, args := datasetFilterClause(domain.DatasetFilter{DatasetGroupID: &groupID, Status: status, FileType: fileType}, nil) + row := r.pool.QueryRow(ctx, + `SELECT `+datasetColumns+` FROM datasets`+where+ + ` ORDER BY end_date DESC NULLS LAST, created_at DESC LIMIT 1`, args...) + out, err := scanDataset(row) + return out, mapError(err) +} +``` + +Observations: `ReplaceObservations` uses `o.StartDate`/`o.EndDate`, copies into columns `"start_date", "end_date"`; `observationColumns` becomes `… to_char(start_date, 'YYYY-MM-DD') AS start_date, to_char(end_date, 'YYYY-MM-DD') AS end_date, …`; `scanObservation` scans `&o.StartDate, &o.EndDate`; `ORDER BY kato_code, date` → `ORDER BY kato_code, start_date` (three places: ListObservations ×2, ListAllObservations in `district.go`). + +- [ ] **Step 5: Service** (`internal/service/dataset.go`) + +- Interface `DatasetRepository`: `UpdateFields(ctx, id uuid.UUID, f domain.DatasetFields) (domain.Dataset, error)`; add `LatestInGroup(ctx context.Context, groupID uuid.UUID, status *string, fileType *domain.FileType) (domain.Dataset, error)`. +- `UploadInput`: add `DatasetGroupID *uuid.UUID; StartDate *string; EndDate *string`. In `Upload`, after category check: `if err := domain.ValidateDatePeriod(in.StartDate, in.EndDate); err != nil { return domain.Dataset{}, err }` and pass the three fields into `repo.Create`. +- `DatasetUpdateInput`: add the three fields; `Update` validates the period and calls `s.repo.UpdateFields(ctx, id, domain.DatasetFields{CategoryID: in.CategoryID, Name: in.Name, Description: in.Description, Unit: in.Unit, DatasetGroupID: in.DatasetGroupID, StartDate: in.StartDate, EndDate: in.EndDate})`. +- `DatasetPatchInput`: add `DatasetGroupID *uuid.UUID; DatasetGroupIDSet bool; StartDate *string; StartDateSet bool; EndDate *string; EndDateSet bool`; `Patch` merges like description/unit, validates the merged period, calls `UpdateFields` with `DatasetFields`. +- Rename all `yc.Date`/`yc.DateTo`/`o.Date`/`o.DateTo` to `StartDate`/`EndDate`; error messages say `start_date`/`end_date`. GeoJSON `data`/`periods` maps are keyed by `o.StartDate` and the period value is `*o.EndDate` (behaviour unchanged). + +- [ ] **Step 6: Handler** (`internal/transport/http/dataset_handler.go`) + +- `yearColumnInput`: `StartDate string \`json:"start_date" validate:"required,datetime=2006-01-02"\``, `EndDate *string \`json:"end_date" validate:"omitempty,datetime=2006-01-02"\``; build `domain.YearColumn{Column, StartDate, EndDate}`. +- `upload`: parse optional form values `dg_id` (UUID → 422 "dg_id must be a valid UUID"), `start_date`, `end_date` (blank → nil; format checked by the service) and pass into `UploadInput`. +- `list`: parse `dg_id` with `parseOptionalUUIDQuery`; for each of `start_date_from`, `start_date_to`, `end_date_from`, `end_date_to`: trim; if non-empty, `time.Parse("2006-01-02", v)` else 400 `" must be YYYY-MM-DD"`; set the filter pointer. +- `datasetPutRequest`: add `DgID *string \`json:"dg_id" validate:"omitempty,uuid"\``, `StartDate *string \`json:"start_date" validate:"omitempty,datetime=2006-01-02"\``, `EndDate *string \`json:"end_date" validate:"omitempty,datetime=2006-01-02"\``; map into `DatasetUpdateInput` (parse `dg_id` UUID). +- `patch`: extend the raw-map handling used for description/unit: keys `dg_id`, `start_date`, `end_date`; `null` → `Set=true, nil`; string → parsed (`dg_id` must be UUID → 422; dates `datetime` → 422 "start_date must be YYYY-MM-DD"). + +- [ ] **Step 7: Update tests** (`internal/service/dataset_test.go`) + +Update the stub repo: `UpdateFields(_ context.Context, id uuid.UUID, f domain.DatasetFields)` assigns all seven fields; add `LatestInGroup` to the stub: + +```go +func (r *stubDatasetRepo) LatestInGroup(_ context.Context, groupID uuid.UUID, status *string, fileType *domain.FileType) (domain.Dataset, error) { + var best *domain.Dataset + for _, d := range r.store { + d := d + if d.DatasetGroupID == nil || *d.DatasetGroupID != groupID { + continue + } + if status != nil && d.Status != *status { + continue + } + if fileType != nil && d.FileType != *fileType { + continue + } + if best == nil || laterPeriod(d, *best) { + best = &d + } + } + if best == nil { + return domain.Dataset{}, domain.ErrNotFound + } + return *best, nil +} + +// laterPeriod reports whether a is fresher than b: greater end_date, NULL +// end_date last, then newer created_at. +func laterPeriod(a, b domain.Dataset) bool { + switch { + case a.EndDate != nil && b.EndDate == nil: + return true + case a.EndDate == nil && b.EndDate != nil: + return false + case a.EndDate != nil && b.EndDate != nil && *a.EndDate != *b.EndDate: + return *a.EndDate > *b.EndDate + default: + return a.CreatedAt.After(b.CreatedAt) + } +} +``` + +Rename every `Date:`/`DateTo:` literal and `.Date`/`.DateTo` access to `StartDate`/`EndDate`. Add tests: + +```go +func TestDatasetService_Upload_StoresPeriodAndGroup(t *testing.T) { + repo := newStubDatasetRepo() + svc := newTestService(repo) // use the existing helper in this file that builds a service with stub store/categories/jobs + gid := uuid.New() + start, end := "2020-01-01", "2020-12-31" + ds, err := svc.Upload(context.Background(), validRasterUpload(t, func(in *UploadInput) { + in.DatasetGroupID = &gid + in.StartDate = &start + in.EndDate = &end + })) + if err != nil { + t.Fatal(err) + } + if ds.DatasetGroupID == nil || *ds.DatasetGroupID != gid || *ds.StartDate != start || *ds.EndDate != end { + t.Fatalf("period/group not stored: %+v", ds) + } +} + +func TestDatasetService_Upload_RejectsReversedPeriod(t *testing.T) { + svc := newTestService(newStubDatasetRepo()) + start, end := "2021-01-01", "2020-12-31" + _, err := svc.Upload(context.Background(), validRasterUpload(t, func(in *UploadInput) { + in.StartDate = &start + in.EndDate = &end + })) + if !errors.Is(err, domain.ErrValidation) { + t.Fatalf("want ErrValidation, got %v", err) + } +} +``` + +(If no `newTestService`/`validRasterUpload` helper exists yet, add them next to the existing upload tests by extracting the setup the existing `TestDatasetService_Upload_Success` uses; the helpers take a mutator `func(*UploadInput)`.) + +- [ ] **Step 8: Run tests** — `go test ./internal/domain/ ./internal/service/ ./internal/transport/... ./internal/repository/...` → PASS; `go vet ./internal/...` clean. + +- [ ] **Step 9: OpenAPI + README** + +`api/openapi.yaml`: `YearColumn` (`start_date` required, `end_date`), `Observation` (`start_date`, `end_date`), `MappingInput` example, `/datasets/{id}/mapping` description; `Dataset` and `DatasetSummary` gain `dg_id` (uuid, nullable), `start_date`, `end_date` (`format: date`, nullable); `DatasetPut`/`DatasetPatch` gain the same; `POST /datasets` form gains `dg_id`, `start_date`, `end_date`; `GET /datasets` parameters `dg_id`, `start_date_from`, `start_date_to`, `end_date_from`, `end_date_to`. `README.md`: mapping example uses `start_date`; add a "Dataset periods" paragraph describing the fields and filters. + +- [ ] **Step 10: Commit** (lane A only; skip if the orchestrator commits) + +```bash +git add migrations/00010_dataset_groups_and_dates.sql internal/domain/dataset.go internal/domain/dataset_test.go internal/repository/postgres/dataset.go internal/repository/postgres/district.go internal/service/dataset.go internal/service/dataset_test.go internal/transport/http/dataset_handler.go api/openapi.yaml README.md +git commit -m "feat(datasets): periods (start_date/end_date), dg_id column, date filters; rename observation dates" +``` + +--- + +### Task 2: Dataset groups — CRUD, latest, group SLD, dataset SLD fallback, `CreateRasterFromFile` + +**Files:** +- Create: `internal/domain/dataset_group.go`, `internal/repository/postgres/dataset_group.go`, `internal/service/dataset_group.go`, `internal/service/dataset_group_test.go`, `internal/transport/http/dataset_group_handler.go` +- Modify: `internal/service/dataset.go`, `internal/service/dataset_test.go`, `internal/transport/http/router.go`, `internal/app/app.go`, `api/openapi.yaml`, `README.md` + +**Interfaces:** +- Consumes: Task 1 (`DatasetGroupID`, `LatestInGroup`, `DatasetFields`). +- Produces (used by Task 7): + - `domain.DatasetGroup` + - `postgres.DatasetGroupRepository` with `GetByCode(ctx, code string) (domain.DatasetGroup, error)`, `GetByID`. + - `service.DatasetGroupRepository` interface (below), `service.NewDatasetGroupService(repo DatasetGroupRepository, datasets DatasetRepository, store ObjectStore)`. + - `service.DatasetService.CreateRasterFromFile(ctx, in CreateRasterInput) (domain.Dataset, error)` and `service.CreateRasterInput{CategoryID uuid.UUID; DatasetGroupID *uuid.UUID; Name, Filename, Path string; StartDate, EndDate *string; Meta json.RawMessage; Automated bool}`. + - `service.NewDatasetService(repo, store, categories, groups groupReader, jobs, parseColumns, parseRows, converter)` — **new 4th parameter** `groups groupReader` where `type groupReader interface{ GetByID(ctx context.Context, id uuid.UUID) (domain.DatasetGroup, error) }`. + +- [ ] **Step 1: Domain** (`internal/domain/dataset_group.go`) + +```go +package domain + +import ( + "time" + + "github.com/google/uuid" +) + +// DatasetGroup groups "одноимённые" datasets (the same indicator delivered for +// different periods). The freshest dataset of a group is the one with the +// greatest end_date. A group may carry a shared SLD style used as a fallback by +// datasets that have none of their own. +type DatasetGroup struct { + ID uuid.UUID `json:"id"` + Code string `json:"code"` + Name string `json:"name"` + Description string `json:"description"` + SldStorageKey *string `json:"sld_storage_key"` + SldFilename *string `json:"sld_filename"` + SldContentType *string `json:"sld_content_type"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} +``` + +- [ ] **Step 2: Repository** (`internal/repository/postgres/dataset_group.go`) + +```go +package postgres + +import ( + "context" + "fmt" + "strings" + + "gis/internal/domain" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// DatasetGroupRepository persists dataset groups in Postgres. +type DatasetGroupRepository struct{ pool *pgxpool.Pool } + +// NewDatasetGroupRepository returns a DatasetGroupRepository backed by pool. +func NewDatasetGroupRepository(pool *pgxpool.Pool) *DatasetGroupRepository { + return &DatasetGroupRepository{pool: pool} +} + +const datasetGroupColumns = `id, code, name, description, sld_storage_key, sld_filename, sld_content_type, created_at, updated_at` + +func scanDatasetGroup(row pgx.Row) (domain.DatasetGroup, error) { + var g domain.DatasetGroup + err := row.Scan(&g.ID, &g.Code, &g.Name, &g.Description, &g.SldStorageKey, &g.SldFilename, &g.SldContentType, &g.CreatedAt, &g.UpdatedAt) + return g, err +} + +func (r *DatasetGroupRepository) Create(ctx context.Context, g domain.DatasetGroup) (domain.DatasetGroup, error) { + row := r.pool.QueryRow(ctx, + `INSERT INTO dataset_groups (code, name, description) VALUES ($1, $2, $3) RETURNING `+datasetGroupColumns, + g.Code, g.Name, g.Description) + out, err := scanDatasetGroup(row) + return out, mapError(err) +} + +func (r *DatasetGroupRepository) GetByID(ctx context.Context, id uuid.UUID) (domain.DatasetGroup, error) { + out, err := scanDatasetGroup(r.pool.QueryRow(ctx, `SELECT `+datasetGroupColumns+` FROM dataset_groups WHERE id = $1`, id)) + return out, mapError(err) +} + +func (r *DatasetGroupRepository) GetByCode(ctx context.Context, code string) (domain.DatasetGroup, error) { + out, err := scanDatasetGroup(r.pool.QueryRow(ctx, `SELECT `+datasetGroupColumns+` FROM dataset_groups WHERE code = $1`, code)) + return out, mapError(err) +} + +// List returns a page of groups ordered by code; code (exact) is optional. +func (r *DatasetGroupRepository) List(ctx context.Context, code *string, limit, offset int) ([]domain.DatasetGroup, error) { + where, args := groupWhere(code) + args = append(args, limit, offset) + rows, err := r.pool.Query(ctx, fmt.Sprintf(`SELECT %s FROM dataset_groups%s ORDER BY code LIMIT $%d OFFSET $%d`, + datasetGroupColumns, where, len(args)-1, len(args)), args...) + if err != nil { + return nil, mapError(err) + } + defer rows.Close() + out := make([]domain.DatasetGroup, 0) + for rows.Next() { + g, err := scanDatasetGroup(rows) + if err != nil { + return nil, mapError(err) + } + out = append(out, g) + } + return out, mapError(rows.Err()) +} + +func (r *DatasetGroupRepository) Count(ctx context.Context, code *string) (int, error) { + where, args := groupWhere(code) + var n int + err := r.pool.QueryRow(ctx, `SELECT count(*) FROM dataset_groups`+where, args...).Scan(&n) + return n, mapError(err) +} + +func groupWhere(code *string) (string, []any) { + if code == nil { + return "", nil + } + return " WHERE code = $1", []any{*code} +} + +func (r *DatasetGroupRepository) Update(ctx context.Context, g domain.DatasetGroup) (domain.DatasetGroup, error) { + row := r.pool.QueryRow(ctx, + `UPDATE dataset_groups SET code = $2, name = $3, description = $4, updated_at = now() WHERE id = $1 RETURNING `+datasetGroupColumns, + g.ID, g.Code, g.Name, g.Description) + out, err := scanDatasetGroup(row) + return out, mapError(err) +} + +// Delete removes a group. Returns domain.ErrNotFound if absent and +// domain.ErrConflict (via the FK) if datasets still reference it. +func (r *DatasetGroupRepository) Delete(ctx context.Context, id uuid.UUID) error { + tag, err := r.pool.Exec(ctx, `DELETE FROM dataset_groups WHERE id = $1`, id) + if err != nil { + return mapError(err) + } + if tag.RowsAffected() == 0 { + return domain.ErrNotFound + } + return nil +} + +func (r *DatasetGroupRepository) SetSLD(ctx context.Context, id uuid.UUID, key, filename, contentType string) (domain.DatasetGroup, error) { + row := r.pool.QueryRow(ctx, + `UPDATE dataset_groups SET sld_storage_key = $2, sld_filename = $3, sld_content_type = $4, updated_at = now() WHERE id = $1 RETURNING `+datasetGroupColumns, + id, key, filename, contentType) + out, err := scanDatasetGroup(row) + return out, mapError(err) +} + +var _ = strings.TrimSpace // keep import if unused after edits; remove if not needed +``` + +(Remove the trailing `strings` line and import if you do not use `strings`.) + +- [ ] **Step 3: Service test first** (`internal/service/dataset_group_test.go`) + +```go +package service + +import ( + "context" + "errors" + "strings" + "testing" + + "gis/internal/domain" + + "github.com/google/uuid" +) + +type stubGroupRepo struct { + store map[uuid.UUID]domain.DatasetGroup +} + +func newStubGroupRepo() *stubGroupRepo { return &stubGroupRepo{store: map[uuid.UUID]domain.DatasetGroup{}} } + +func (r *stubGroupRepo) Create(_ context.Context, g domain.DatasetGroup) (domain.DatasetGroup, error) { + for _, e := range r.store { + if e.Code == g.Code { + return domain.DatasetGroup{}, domain.ErrConflict + } + } + g.ID = uuid.New() + r.store[g.ID] = g + return g, nil +} +func (r *stubGroupRepo) GetByID(_ context.Context, id uuid.UUID) (domain.DatasetGroup, error) { + g, ok := r.store[id] + if !ok { + return domain.DatasetGroup{}, domain.ErrNotFound + } + return g, nil +} +func (r *stubGroupRepo) GetByCode(_ context.Context, code string) (domain.DatasetGroup, error) { + for _, g := range r.store { + if g.Code == code { + return g, nil + } + } + return domain.DatasetGroup{}, domain.ErrNotFound +} +func (r *stubGroupRepo) List(_ context.Context, code *string, limit, offset int) ([]domain.DatasetGroup, error) { + out := []domain.DatasetGroup{} + for _, g := range r.store { + if code == nil || g.Code == *code { + out = append(out, g) + } + } + return out, nil +} +func (r *stubGroupRepo) Count(_ context.Context, code *string) (int, error) { + l, _ := r.List(context.Background(), code, 0, 0) + return len(l), nil +} +func (r *stubGroupRepo) Update(_ context.Context, g domain.DatasetGroup) (domain.DatasetGroup, error) { + if _, ok := r.store[g.ID]; !ok { + return domain.DatasetGroup{}, domain.ErrNotFound + } + r.store[g.ID] = g + return g, nil +} +func (r *stubGroupRepo) Delete(_ context.Context, id uuid.UUID) error { + if _, ok := r.store[id]; !ok { + return domain.ErrNotFound + } + delete(r.store, id) + return nil +} +func (r *stubGroupRepo) SetSLD(_ context.Context, id uuid.UUID, key, filename, contentType string) (domain.DatasetGroup, error) { + g, ok := r.store[id] + if !ok { + return domain.DatasetGroup{}, domain.ErrNotFound + } + g.SldStorageKey, g.SldFilename, g.SldContentType = &key, &filename, &contentType + r.store[id] = g + return g, nil +} + +func TestDatasetGroupService_Create_RequiresSlug(t *testing.T) { + svc := NewDatasetGroupService(newStubGroupRepo(), newStubDatasetRepo(), newStubStore()) + _, err := svc.Create(context.Background(), DatasetGroupInput{Code: "Bad Code", Name: "x"}) + if !errors.Is(err, domain.ErrValidation) { + t.Fatalf("want ErrValidation, got %v", err) + } +} + +func TestDatasetGroupService_Latest_PrefersGreatestEndDateThenCreatedAt(t *testing.T) { + groups := newStubGroupRepo() + datasets := newStubDatasetRepo() + svc := NewDatasetGroupService(groups, datasets, newStubStore()) + g, _ := svc.Create(context.Background(), DatasetGroupInput{Code: "pop", Name: "Population"}) + s := func(v string) *string { return &v } + old := domain.Dataset{ID: uuid.New(), DatasetGroupID: &g.ID, EndDate: s("2019-12-31"), Status: domain.DatasetStatusReady} + fresh := domain.Dataset{ID: uuid.New(), DatasetGroupID: &g.ID, EndDate: s("2020-12-31"), Status: domain.DatasetStatusReady} + undated := domain.Dataset{ID: uuid.New(), DatasetGroupID: &g.ID, Status: domain.DatasetStatusReady} + for _, d := range []domain.Dataset{old, fresh, undated} { + datasets.store[d.ID] = d + } + got, err := svc.Latest(context.Background(), g.ID, nil, nil) + if err != nil { + t.Fatal(err) + } + if got.ID != fresh.ID { + t.Fatalf("want %s, got %s", fresh.ID, got.ID) + } +} + +func TestDatasetGroupService_Latest_NotFoundWhenEmpty(t *testing.T) { + groups := newStubGroupRepo() + svc := NewDatasetGroupService(groups, newStubDatasetRepo(), newStubStore()) + g, _ := svc.Create(context.Background(), DatasetGroupInput{Code: "pop", Name: "Population"}) + if _, err := svc.Latest(context.Background(), g.ID, nil, nil); !errors.Is(err, domain.ErrNotFound) { + t.Fatalf("want ErrNotFound, got %v", err) + } +} + +func TestDatasetGroupService_UploadSLD_RejectsBadExtension(t *testing.T) { + groups := newStubGroupRepo() + svc := NewDatasetGroupService(groups, newStubDatasetRepo(), newStubStore()) + g, _ := svc.Create(context.Background(), DatasetGroupInput{Code: "pop", Name: "Population"}) + _, err := svc.UploadSLD(context.Background(), g.ID, UploadSLDInput{Filename: "style.txt", ContentType: "text/plain", Size: 3, Reader: strings.NewReader("")}) + if !errors.Is(err, domain.ErrValidation) { + t.Fatalf("want ErrValidation, got %v", err) + } +} +``` + +`newStubStore()` — the in-memory ObjectStore stub already used in `dataset_test.go` (reuse its constructor name; if it is named differently, use that name). + +- [ ] **Step 4: Run** `go test ./internal/service/ -run DatasetGroupService` → FAIL (undefined). + +- [ ] **Step 5: Service** (`internal/service/dataset_group.go`) + +```go +package service + +import ( + "bytes" + "context" + "fmt" + "io" + "path" + "path/filepath" + "strings" + + "gis/internal/domain" + + "github.com/google/uuid" +) + +// DatasetGroupRepository is the persistence behaviour DatasetGroupService needs. +type DatasetGroupRepository interface { + Create(ctx context.Context, g domain.DatasetGroup) (domain.DatasetGroup, error) + GetByID(ctx context.Context, id uuid.UUID) (domain.DatasetGroup, error) + GetByCode(ctx context.Context, code string) (domain.DatasetGroup, error) + List(ctx context.Context, code *string, limit, offset int) ([]domain.DatasetGroup, error) + Count(ctx context.Context, code *string) (int, error) + Update(ctx context.Context, g domain.DatasetGroup) (domain.DatasetGroup, error) + Delete(ctx context.Context, id uuid.UUID) error + SetSLD(ctx context.Context, id uuid.UUID, key, filename, contentType string) (domain.DatasetGroup, error) +} + +// DatasetGroupInput carries the mutable fields of a group. +type DatasetGroupInput struct { + Code string + Name string + Description string +} + +// DatasetGroupPage is a page of groups. +type DatasetGroupPage struct { + Items []domain.DatasetGroup + Page int + PageSize int + Total int +} + +// DatasetGroupService implements dataset-group business rules. +type DatasetGroupService struct { + repo DatasetGroupRepository + datasets DatasetRepository + store ObjectStore +} + +// NewDatasetGroupService wires the group repository, the dataset repository +// (for the "latest" lookup) and the object store (for the group SLD). +func NewDatasetGroupService(repo DatasetGroupRepository, datasets DatasetRepository, store ObjectStore) *DatasetGroupService { + return &DatasetGroupService{repo: repo, datasets: datasets, store: store} +} + +func (s *DatasetGroupService) Create(ctx context.Context, in DatasetGroupInput) (domain.DatasetGroup, error) { + if !domain.ValidSlug(in.Code) { + return domain.DatasetGroup{}, fmt.Errorf("%w: code must be a slug (lowercase latin letters, digits, and dashes)", domain.ErrValidation) + } + return s.repo.Create(ctx, domain.DatasetGroup{Code: in.Code, Name: in.Name, Description: in.Description}) +} + +func (s *DatasetGroupService) Get(ctx context.Context, id uuid.UUID) (domain.DatasetGroup, error) { + return s.repo.GetByID(ctx, id) +} + +// List returns a page of groups; page/pageSize are clamped like datasets. +func (s *DatasetGroupService) List(ctx context.Context, code *string, page, pageSize int) (DatasetGroupPage, error) { + if page < 1 { + page = 1 + } + if pageSize < 1 { + pageSize = DefaultPageSize + } + if pageSize > MaxPageSize { + pageSize = MaxPageSize + } + items, err := s.repo.List(ctx, code, pageSize, (page-1)*pageSize) + if err != nil { + return DatasetGroupPage{}, err + } + total, err := s.repo.Count(ctx, code) + if err != nil { + return DatasetGroupPage{}, err + } + return DatasetGroupPage{Items: items, Page: page, PageSize: pageSize, Total: total}, nil +} + +func (s *DatasetGroupService) Update(ctx context.Context, id uuid.UUID, in DatasetGroupInput) (domain.DatasetGroup, error) { + if !domain.ValidSlug(in.Code) { + return domain.DatasetGroup{}, fmt.Errorf("%w: code must be a slug (lowercase latin letters, digits, and dashes)", domain.ErrValidation) + } + return s.repo.Update(ctx, domain.DatasetGroup{ID: id, Code: in.Code, Name: in.Name, Description: in.Description}) +} + +// Delete removes a group (409 via ErrConflict while datasets reference it) and +// then best-effort removes its SLD object. +func (s *DatasetGroupService) Delete(ctx context.Context, id uuid.UUID) error { + g, err := s.repo.GetByID(ctx, id) + if err != nil { + return err + } + if err := s.repo.Delete(ctx, id); err != nil { + return err + } + if g.SldStorageKey != nil { + _ = s.store.Remove(ctx, *g.SldStorageKey) + } + return nil +} + +// Latest returns the freshest dataset of the group (greatest end_date, NULLs +// last, then newest), optionally restricted by status and file type. +func (s *DatasetGroupService) Latest(ctx context.Context, id uuid.UUID, status *string, fileType *domain.FileType) (domain.Dataset, error) { + if _, err := s.repo.GetByID(ctx, id); err != nil { + return domain.Dataset{}, err + } + return s.datasets.LatestInGroup(ctx, id, status, fileType) +} + +// UploadSLD validates and stores the group's shared SLD, replacing any previous one. +func (s *DatasetGroupService) UploadSLD(ctx context.Context, id uuid.UUID, in UploadSLDInput) (domain.DatasetGroup, error) { + g, err := s.repo.GetByID(ctx, id) + if err != nil { + return domain.DatasetGroup{}, err + } + ext := strings.ToLower(filepath.Ext(in.Filename)) + if !domain.ExtensionAllowedForSLD(ext) { + return domain.DatasetGroup{}, fmt.Errorf("%w: sld file must have one of these extensions: %s", + domain.ErrValidation, strings.Join(domain.AllowedSLDExtensions, ", ")) + } + head := make([]byte, 512) + n, err := io.ReadFull(in.Reader, head) + if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF { + return domain.DatasetGroup{}, fmt.Errorf("read sld upload: %w", err) + } + head = head[:n] + if err := domain.ValidateFileContent(ext, head); err != nil { + return domain.DatasetGroup{}, fmt.Errorf("%w: %s", domain.ErrValidation, err) + } + content := io.MultiReader(bytes.NewReader(head), in.Reader) + + key := path.Join("groups", g.ID.String(), "sld", in.Filename) + if err := s.store.Put(ctx, key, content, in.Size, in.ContentType); err != nil { + return domain.DatasetGroup{}, err + } + updated, err := s.repo.SetSLD(ctx, id, key, in.Filename, in.ContentType) + if err != nil { + _ = s.store.Remove(ctx, key) + return domain.DatasetGroup{}, err + } + if g.SldStorageKey != nil && *g.SldStorageKey != key { + _ = s.store.Remove(ctx, *g.SldStorageKey) + } + return updated, nil +} + +// DownloadSLD returns the group and a reader for its SLD; ErrSLDNotFound when none. +func (s *DatasetGroupService) DownloadSLD(ctx context.Context, id uuid.UUID) (domain.DatasetGroup, io.ReadCloser, error) { + g, err := s.repo.GetByID(ctx, id) + if err != nil { + return domain.DatasetGroup{}, nil, err + } + if g.SldStorageKey == nil { + return domain.DatasetGroup{}, nil, ErrSLDNotFound + } + obj, err := s.store.Get(ctx, *g.SldStorageKey) + if err != nil { + return domain.DatasetGroup{}, nil, err + } + return g, obj, nil +} +``` + +- [ ] **Step 6: DatasetService changes** (`internal/service/dataset.go`) + +1. Add `groupReader` interface and the `groups groupReader` field + **4th constructor parameter** (update every `NewDatasetService(` call: `internal/app/app.go`, tests). +2. `ensureGroupExists(ctx, id *uuid.UUID) error` — nil → nil; not found → `fmt.Errorf("%w: dataset group %s not found", domain.ErrValidation, id)`. Call it in `Upload`, `Update`, and `Patch` (when the group changes). +3. `DownloadSLD` fallback: when `dataset.SldStorageKey == nil` and `dataset.DatasetGroupID != nil`, load the group; if it has an SLD, return a dataset copy whose `SldStorageKey/SldFilename/SldContentType` are the group's, and stream the group's object. Otherwise `ErrSLDNotFound`. +4. `CreateRasterFromFile`: + +```go +// CreateRasterInput describes a raster file already on local disk that should +// become a dataset (used by the Landscape Program to publish results). +type CreateRasterInput struct { + CategoryID uuid.UUID + DatasetGroupID *uuid.UUID + Name string + Filename string // stored filename, e.g. "lp-z-integrated.tif" + Path string // local path to read from + StartDate *string + EndDate *string + Meta json.RawMessage + Automated bool +} + +// CreateRasterFromFile stores a local GeoTIFF as a new raster dataset and +// enqueues the standard COG conversion, exactly like an uploaded raster. +func (s *DatasetService) CreateRasterFromFile(ctx context.Context, in CreateRasterInput) (domain.Dataset, error) { + f, err := os.Open(in.Path) + if err != nil { + return domain.Dataset{}, err + } + defer f.Close() + info, err := f.Stat() + if err != nil { + return domain.Dataset{}, err + } + return s.Upload(ctx, UploadInput{ + CategoryID: in.CategoryID, + DatasetGroupID: in.DatasetGroupID, + Name: in.Name, + Meta: in.Meta, + Automated: in.Automated, + Filename: in.Filename, + FileType: domain.FileTypeRaster, + ContentType: "image/tiff", + Size: info.Size(), + Reader: f, + StartDate: in.StartDate, + EndDate: in.EndDate, + }) +} +``` + +Add tests in `dataset_test.go`: `TestDatasetService_DownloadSLD_FallsBackToGroup` (dataset without SLD, group with SLD in stub store → reader returns group content) and `TestDatasetService_CreateRasterFromFile` (temp file with `II*\x00` header → dataset created with `file_type raster`, group id, convert job enqueued). The stub group reader is `stubGroupRepo` from Step 3. + +- [ ] **Step 7: Handler** (`internal/transport/http/dataset_group_handler.go`) + +```go +package http + +import ( + "io" + "net/http" + "strings" + + "gis/internal/domain" + "gis/internal/service" + "gis/pkg/httputil" + + "github.com/go-chi/chi/v5" + "github.com/go-playground/validator/v10" +) + +// DatasetGroupHandler serves the /dataset-groups routes. +type DatasetGroupHandler struct { + svc *service.DatasetGroupService + validate *validator.Validate +} + +func NewDatasetGroupHandler(svc *service.DatasetGroupService, validate *validator.Validate) *DatasetGroupHandler { + return &DatasetGroupHandler{svc: svc, validate: validate} +} + +func (h *DatasetGroupHandler) Register(r chi.Router) { + r.Get("/", h.list) + r.Post("/", h.create) + r.Get("/{id}", h.get) + r.Put("/{id}", h.update) + r.Delete("/{id}", h.delete) + r.Get("/{id}/latest", h.latest) + r.Post("/{id}/sld", h.uploadSLD) + r.Get("/{id}/sld", h.downloadSLD) +} + +type datasetGroupRequest struct { + Code string `json:"code" validate:"required,max=255,slug"` + Name string `json:"name" validate:"required,max=255"` + Description string `json:"description" validate:"max=2000"` +} +``` + +Implement `create` (201), `list` (`?code=`, `page`, `page_size` via `parsePositiveIntQuery`, `newPaginated`), `get`, `update`, `delete` (204), `latest` (`?status=` validated with `domain.ValidDatasetStatus`, `?file_type=` validated with `FileType.Valid()`; 404 via `respondDomainError`), `uploadSLD`/`downloadSLD` mirroring `DatasetHandler.uploadSLD/downloadSLD` (multipart `file`, default content type `application/xml`, `Content-Disposition: attachment`). Follow `category_handler.go` for decode/validate/error flow. + +- [ ] **Step 8: Wiring** — `router.go`: add `DatasetGroups *DatasetGroupHandler` to `RouterDeps` and `r.Route("/dataset-groups", deps.DatasetGroups.Register)`. `app.go`: `groupRepo := postgres.NewDatasetGroupRepository(pool)`, pass `groupRepo` as the 4th arg to `NewDatasetService`, add field `groups *service.DatasetGroupService` = `service.NewDatasetGroupService(groupRepo, datasetRepo, store)`, and `DatasetGroups: transporthttp.NewDatasetGroupHandler(a.groups, validate)`. + +- [ ] **Step 9: Run** `go test ./internal/service/ ./internal/transport/... && go build ./...` → PASS. + +- [ ] **Step 10: OpenAPI + README** — add tag `Dataset Groups`, paths `/dataset-groups`, `/dataset-groups/{id}`, `/dataset-groups/{id}/latest`, `/dataset-groups/{id}/sld`, schemas `DatasetGroup`, `DatasetGroupInput`, `DatasetGroupPage`; note the SLD fallback on `GET /datasets/{id}/sld`. README: "Dataset groups" section (what a group is, freshness rule, SLD fallback). + +- [ ] **Step 11: Commit** + +```bash +git add internal/domain/dataset_group.go internal/repository/postgres/dataset_group.go internal/service/dataset_group.go internal/service/dataset_group_test.go internal/service/dataset.go internal/service/dataset_test.go internal/transport/http/dataset_group_handler.go internal/transport/http/router.go internal/app/app.go api/openapi.yaml README.md +git commit -m "feat(dataset-groups): CRUD, latest-in-group, group SLD with dataset fallback, CreateRasterFromFile" +``` + +--- + +### Task 3: Raster I/O abstraction + godal implementation + Dockerfile + +**Files:** +- Create: `internal/raster/io.go`, `internal/raster/godal.go`, `internal/raster/godal_test.go` +- Modify: `go.mod`, `go.sum` (via `go get github.com/airbusgeo/godal@v0.0.18`), `build/package/Dockerfile`, `.golangci.yml` only if cgo needs a setting (usually not) + +**Interfaces:** +- Produces (used by Task 6): + +```go +package raster +type Grid struct { Width, Height int; GeoTransform [6]float64; ProjectionWKT string } +type Window struct { X, Y, Width, Height int } +type Reader interface { Grid() Grid; NoData() (float64, bool); ReadFloat64(w Window, dst []float64) error; Close() error } +type WriterOptions struct { NoData int16; TileSize int; Compression string } +type Writer interface { WriteInt16(w Window, src []int16) error; Close() error } +type IO interface { + Open(ctx context.Context, path string) (Reader, error) + Create(ctx context.Context, path string, g Grid, o WriterOptions) (Writer, error) + SameGrid(a, b Grid) bool +} +func DefaultWriterOptions() WriterOptions // {NoData: -9999, TileSize: 512, Compression: "DEFLATE"} +func NewGodalIO() *GodalIO // implements IO +``` + +- [ ] **Step 1: Interfaces** (`internal/raster/io.go`) — write the block above with doc comments. Add: + +```go +// Windows tiles a w×h raster into row-major windows of at most size×size. +func Windows(w, h, size int) []Window { + var out []Window + for y := 0; y < h; y += size { + wh := min(size, h-y) + for x := 0; x < w; x += size { + out = append(out, Window{X: x, Y: y, Width: min(size, w-x), Height: wh}) + } + } + return out +} +``` + +Test (`internal/raster/io_test.go`): `Windows(10, 10, 4)` → 9 windows, last is `{8,8,2,2}`; `Windows(5, 3, 8)` → one window `{0,0,5,3}`. + +- [ ] **Step 2: Add dependency** — `go get github.com/airbusgeo/godal@v0.0.18`. Verify `CGO_ENABLED=1 go build ./internal/raster/` compiles (Homebrew GDAL provides `gdal.pc` in `/opt/homebrew/lib/pkgconfig`; if `pkg-config` is missing run `brew install pkgconf`). + +- [ ] **Step 3: Failing round-trip test** (`internal/raster/godal_test.go`) + +```go +package raster + +import ( + "context" + "path/filepath" + "testing" +) + +func TestGodalIO_RoundTrip(t *testing.T) { + rio := NewGodalIO() + ctx := context.Background() + path := filepath.Join(t.TempDir(), "out.tif") + grid := Grid{Width: 10, Height: 7, GeoTransform: [6]float64{100, 30, 0, 200, 0, -30}, ProjectionWKT: "EPSG:32642"} + + w, err := rio.Create(ctx, path, grid, DefaultWriterOptions()) + if err != nil { + t.Fatal(err) + } + full := make([]int16, 10*7) + for i := range full { + full[i] = int16(i%6) - 1 // -1..4 (-1 stands in for garbage, not nodata) + } + full[0] = -9999 + if err := w.WriteInt16(Window{0, 0, 10, 7}, full); err != nil { + t.Fatal(err) + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + + r, err := rio.Open(ctx, path) + if err != nil { + t.Fatal(err) + } + defer r.Close() + g := r.Grid() + if g.Width != 10 || g.Height != 7 || g.GeoTransform != grid.GeoTransform || g.ProjectionWKT == "" { + t.Fatalf("grid mismatch: %+v", g) + } + if nd, ok := r.NoData(); !ok || nd != -9999 { + t.Fatalf("nodata = %v %v", nd, ok) + } + dst := make([]float64, 3*2) + if err := r.ReadFloat64(Window{X: 8, Y: 5, Width: 2, Height: 2}, dst[:4]); err != nil { + t.Fatal(err) + } + // pixel (8,5) has index 58 -> 58%6-1 = 3 ; (9,5)=59 -> 4 ; (8,6)=68 -> 1 ; (9,6)=69 -> 2 + want := []float64{3, 4, 1, 2} + for i, v := range want { + if dst[i] != v { + t.Fatalf("pixel %d = %v want %v", i, dst[i], v) + } + } + if !rio.SameGrid(g, grid) { + t.Fatal("SameGrid should be true for identical grids") + } + other := grid + other.GeoTransform[0] += 1 + if rio.SameGrid(g, other) { + t.Fatal("SameGrid should be false for shifted origin") + } +} +``` + +Run: `go test ./internal/raster/ -run TestGodalIO -v` → FAIL (undefined `NewGodalIO`). + +- [ ] **Step 4: Implementation** (`internal/raster/godal.go`) + +```go +package raster + +import ( + "context" + "fmt" + "math" + "sync" + + "github.com/airbusgeo/godal" +) + +var registerOnce sync.Once + +// GodalIO implements IO with the GDAL bindings (github.com/airbusgeo/godal). +// GDAL datasets are not goroutine-safe: callers must not share a Reader or +// Writer between goroutines without external synchronisation. +type GodalIO struct{} + +// NewGodalIO registers GDAL drivers (once) and returns a GodalIO. +func NewGodalIO() *GodalIO { + registerOnce.Do(godal.RegisterAll) + return &GodalIO{} +} + +// DefaultWriterOptions returns the Landscape Program output profile. +func DefaultWriterOptions() WriterOptions { + return WriterOptions{NoData: -9999, TileSize: 512, Compression: "DEFLATE"} +} + +type godalReader struct { + ds *godal.Dataset + grid Grid +} + +func (g *GodalIO) Open(_ context.Context, path string) (Reader, error) { + ds, err := godal.Open(path) + if err != nil { + return nil, fmt.Errorf("open raster %s: %w", path, err) + } + st := ds.Structure() + if st.NBands < 1 { + ds.Close() + return nil, fmt.Errorf("open raster %s: no bands", path) + } + gt, err := ds.GeoTransform() + if err != nil { + ds.Close() + return nil, fmt.Errorf("open raster %s: geotransform: %w", path, err) + } + return &godalReader{ds: ds, grid: Grid{Width: st.SizeX, Height: st.SizeY, GeoTransform: gt, ProjectionWKT: ds.Projection()}}, nil +} + +func (r *godalReader) Grid() Grid { return r.grid } + +func (r *godalReader) NoData() (float64, bool) { return r.ds.Bands()[0].NoData() } + +func (r *godalReader) ReadFloat64(w Window, dst []float64) error { + if len(dst) < w.Width*w.Height { + return fmt.Errorf("read window: dst has %d elements, need %d", len(dst), w.Width*w.Height) + } + return r.ds.Bands()[0].Read(w.X, w.Y, dst[:w.Width*w.Height], w.Width, w.Height) +} + +func (r *godalReader) Close() error { return r.ds.Close() } + +type godalWriter struct{ ds *godal.Dataset } + +func (g *GodalIO) Create(_ context.Context, path string, grid Grid, o WriterOptions) (Writer, error) { + tile := fmt.Sprint(o.TileSize) + ds, err := godal.Create(godal.GTiff, path, 1, godal.Int16, grid.Width, grid.Height, + godal.CreationOption("TILED=YES", "BLOCKXSIZE="+tile, "BLOCKYSIZE="+tile, + "COMPRESS="+o.Compression, "PREDICTOR=2", "BIGTIFF=IF_SAFER")) + if err != nil { + return nil, fmt.Errorf("create raster %s: %w", path, err) + } + if err := ds.SetGeoTransform(grid.GeoTransform); err != nil { + ds.Close() + return nil, fmt.Errorf("create raster %s: geotransform: %w", path, err) + } + if grid.ProjectionWKT != "" { + if err := ds.SetProjection(grid.ProjectionWKT); err != nil { + ds.Close() + return nil, fmt.Errorf("create raster %s: projection: %w", path, err) + } + } + if err := ds.Bands()[0].SetNoData(float64(o.NoData)); err != nil { + ds.Close() + return nil, fmt.Errorf("create raster %s: nodata: %w", path, err) + } + return &godalWriter{ds: ds}, nil +} + +func (w *godalWriter) WriteInt16(win Window, src []int16) error { + if len(src) < win.Width*win.Height { + return fmt.Errorf("write window: src has %d elements, need %d", len(src), win.Width*win.Height) + } + return w.ds.Bands()[0].Write(win.X, win.Y, src[:win.Width*win.Height], win.Width, win.Height) +} + +func (w *godalWriter) Close() error { return w.ds.Close() } + +// SameGrid reports whether two grids describe the same pixels: equal size, +// geotransform within 1e-9 (absolute), and the same CRS (GDAL IsSame; an +// empty projection only matches an empty projection). +func (g *GodalIO) SameGrid(a, b Grid) bool { + if a.Width != b.Width || a.Height != b.Height { + return false + } + for i := range a.GeoTransform { + if math.Abs(a.GeoTransform[i]-b.GeoTransform[i]) > 1e-9 { + return false + } + } + if a.ProjectionWKT == "" || b.ProjectionWKT == "" { + return a.ProjectionWKT == b.ProjectionWKT + } + sa, err := godal.NewSpatialRefFromWKT(a.ProjectionWKT) + if err != nil { + return false + } + defer sa.Close() + sb, err := godal.NewSpatialRefFromWKT(b.ProjectionWKT) + if err != nil { + return false + } + defer sb.Close() + return sa.IsSame(sb) +} +``` + +Check godal's actual API names while implementing (`ds.Structure()`, `Bands()`, `Band.Read/Write` generic over slice types, `NewSpatialRefFromWKT`, `SpatialRef.IsSame`, `godal.CreationOption`); adjust if a name differs in v0.0.18 (`go doc github.com/airbusgeo/godal Dataset`). + +- [ ] **Step 5: Run** `go test ./internal/raster/ -v` → PASS. + +- [ ] **Step 6: Dockerfile** (`build/package/Dockerfile`) + +```dockerfile +# Build context is the repo root: docker build -f build/package/Dockerfile . +FROM golang:1.26-alpine AS builder + +# godal (GDAL bindings) needs cgo, a C toolchain and the GDAL headers. +RUN apk add --no-cache gdal-dev gcc musl-dev pkgconfig + +WORKDIR /src + +COPY go.mod go.sum ./ +RUN go mod download + +COPY . . +RUN CGO_ENABLED=1 GOOS=linux go build -trimpath -o /out/gis ./cmd/gis + +FROM alpine:3.22 + +# gdal: shared library for godal (raster math in the worker). +# gdal-tools: gdal_translate / gdalinfo (COG conversion) and ogr2ogr. +RUN apk add --no-cache ca-certificates tzdata gdal gdal-tools \ + && adduser -D -u 10001 app + +WORKDIR /app +COPY --from=builder /out/gis /usr/local/bin/gis + +USER app + +EXPOSE 8080 + +ENTRYPOINT ["gis"] +CMD ["serve"] +``` + +If Docker is available, verify with `docker build -f build/package/Dockerfile -t gis:lp .`; otherwise note in the report that the image build was not verified. + +- [ ] **Step 7: Commit** + +```bash +git add internal/raster/io.go internal/raster/io_test.go internal/raster/godal.go internal/raster/godal_test.go go.mod go.sum build/package/Dockerfile +git commit -m "feat(raster): IO abstraction with godal implementation; cgo Dockerfile" +``` + +--- + +### Task 4: Landscape graph, params and pixel kernels (pure Go) + +**Files:** +- Create: `internal/landscape/graph.go`, `internal/landscape/params.go`, `internal/landscape/kernels.go`, `internal/landscape/kernels_test.go`, `internal/landscape/graph_test.go` + +**Interfaces (produced, used by Tasks 5–8):** + +```go +package landscape +const ( + NoData int16 = -9999 + CategoryCode = "landscape" // dataset category for every LP dataset +) +type Slot struct { Code, Title string } // Code is the dataset_groups.code +type Output struct { Code, Title string } +type Node struct { Code, Name string; Stage int; Inputs []string; Outputs []string; Kernel Kernel } +// Kernel computes one window: in[i] follows Node.Inputs order, out[j] follows Node.Outputs order; +// all slices have len == window pixels. Every out value must be set (NoData or class). +type Kernel func(in [][]float64, out [][]int16, p Params) +var Slots []Slot; var Outputs []Output; var Nodes []Node +func NodeByCode(code string) (Node, bool) +func Stages() [][]Node // nodes grouped by Stage ascending, each group ordered by Code +func OutputByCode(code string) (Output, bool) +func SlotByCode(code string) (Slot, bool) +func NodeForOutput(group string) (Node, bool) // which node produces a group +type Params struct { + WindowSize int `json:"window_size"` + TiePriority []int `json:"tie_priority"` + V1MinSE12 int `json:"v1_min_s_e12"` + V1MinSE3 int `json:"v1_min_s_e3"` + V2MinSE3 int `json:"v2_min_s_e3"` + V2MinSE4 int `json:"v2_min_s_e4"` +} +func DefaultParams() Params +func (p Params) WithDefaults() Params // zero fields -> defaults +func (p Params) Validate() error // window 256..8192; thresholds 1..5; tie_priority permutation of 1..5 +func (p Params) ForNode(code string) Params // only fields the node uses (window_size never included); used in fingerprints +``` + +- [ ] **Step 1: graph.go** + +```go +// Package landscape defines the Landscape Program (ЛП) computation graph — the +// per-pixel formulas that turn ecological (Z), vulnerability (U) and +// suitability (P) class rasters into constraints (E), conflicts (C) and +// functional zoning (V1–V3) — together with a window-based executor. +package landscape + +import "sort" + +const ( + // NoData is the output nodata value of every LP raster. + NoData int16 = -9999 + // CategoryCode is the dataset category holding every LP dataset. + CategoryCode = "landscape" +) + +// Group codes. Slots are leaf inputs users upload; the rest are node outputs. +const ( + SlotZBiodiversity = "lp-z-biodiversity" + SlotZLCR = "lp-z-lcr" + SlotZSoilProtection = "lp-z-soil-protection" + SlotZWR = "lp-z-wr" + SlotUSurfWater = "lp-u-surf-water" + SlotUSoil = "lp-u-soil" + SlotUCrop = "lp-u-crop" + SlotUVeg = "lp-u-veg" + SlotPIrrigation = "lp-p-irrigation" + SlotPSnow = "lp-p-snow" + SlotPCrop = "lp-p-crop" + SlotPPasture = "lp-p-pasture" + SlotPTourism = "lp-p-tourism" + + OutZIntegrated = "lp-z-integrated" + OutUIntegrated = "lp-u-integrated" + OutZMax = "lp-z-max" + OutUMax = "lp-u-max" + OutE = "lp-e" + OutCIrrigation = "lp-c-irrigation" + OutCSnow = "lp-c-snow" + OutCCrop = "lp-c-crop" + OutCPasture = "lp-c-pasture" + OutCTourism = "lp-c-tourism" + OutPMax = "lp-p-max" + OutV1 = "lp-v1-conservative" + OutV2 = "lp-v2-balanced" + OutV3 = "lp-v3-development" + OutCMax = "lp-c-max" +) + +type Slot struct{ Code, Title string } +type Output struct{ Code, Title string } + +// Kernel computes one window. in[i] follows Node.Inputs order and out[j] +// follows Node.Outputs order; every slice has exactly window-pixel length. +type Kernel func(in [][]float64, out [][]int16, p Params) + +// Node is one function of the graph. Stage = first digit of Code. +type Node struct { + Code, Name string + Stage int + Inputs []string + Outputs []string + Kernel Kernel +} + +var Slots = []Slot{ + {SlotZBiodiversity, "Z Biodiversity"}, {SlotZLCR, "Z LCR"}, {SlotZSoilProtection, "Z Soil Protection"}, {SlotZWR, "Z Water Resources"}, + {SlotUSurfWater, "U Surface Water"}, {SlotUSoil, "U Soil"}, {SlotUCrop, "U Crop"}, {SlotUVeg, "U Vegetation"}, + {SlotPIrrigation, "P Irrigation"}, {SlotPSnow, "P Snow"}, {SlotPCrop, "P Crop"}, {SlotPPasture, "P Pasture"}, {SlotPTourism, "P Tourism"}, +} + +var Outputs = []Output{ + {OutZIntegrated, "Z Integrated"}, {OutUIntegrated, "U Integrated"}, {OutZMax, "Z Max"}, {OutUMax, "U Max"}, + {OutE, "E Ecological Constraint"}, + {OutCIrrigation, "C Irrigation"}, {OutCSnow, "C Snow"}, {OutCCrop, "C Crop"}, {OutCPasture, "C Pasture"}, {OutCTourism, "C Tourism"}, + {OutPMax, "P Max"}, + {OutV1, "V1 Conservative"}, {OutV2, "V2 Balanced"}, {OutV3, "V3 Development"}, {OutCMax, "C Max"}, +} + +var zInputs = []string{SlotZBiodiversity, SlotZLCR, SlotZSoilProtection, SlotZWR} +var uInputs = []string{SlotUSurfWater, SlotUSoil, SlotUCrop, SlotUVeg} // order matters: water, soil, crop, veg +var pInputs = []string{SlotPIrrigation, SlotPSnow, SlotPCrop, SlotPPasture, SlotPTourism} // sector order 1..5 +var cOutputs = []string{OutCIrrigation, OutCSnow, OutCCrop, OutCPasture, OutCTourism} + +var Nodes = []Node{ + {Code: "00", Name: "z_integrated", Stage: 0, Inputs: zInputs, Outputs: []string{OutZIntegrated}, Kernel: kernelZIntegrated}, + {Code: "01", Name: "u_integrated", Stage: 0, Inputs: uInputs, Outputs: []string{OutUIntegrated}, Kernel: kernelUIntegrated}, + {Code: "02", Name: "z_max", Stage: 0, Inputs: zInputs, Outputs: []string{OutZMax}, Kernel: kernelZMax}, + {Code: "03", Name: "u_max", Stage: 0, Inputs: uInputs, Outputs: []string{OutUMax}, Kernel: kernelUMax}, + {Code: "10", Name: "e_ecological_constraint", Stage: 1, Inputs: []string{OutZIntegrated, OutUIntegrated}, Outputs: []string{OutE}, Kernel: kernelE}, + {Code: "20", Name: "c_conflicts", Stage: 2, Inputs: append(append([]string{}, pInputs...), OutE), Outputs: cOutputs, Kernel: kernelCConflicts}, + {Code: "21", Name: "p_max", Stage: 2, Inputs: pInputs, Outputs: []string{OutPMax}, Kernel: kernelMaxAnyValid}, + {Code: "30", Name: "func_zoning", Stage: 3, Inputs: append([]string{OutE}, cOutputs...), Outputs: []string{OutV1, OutV2, OutV3}, Kernel: kernelFuncZoning}, + {Code: "31", Name: "c_max", Stage: 3, Inputs: cOutputs, Outputs: []string{OutCMax}, Kernel: kernelMaxAnyValid}, +} + +func NodeByCode(code string) (Node, bool) { + for _, n := range Nodes { + if n.Code == code { + return n, true + } + } + return Node{}, false +} + +func SlotByCode(code string) (Slot, bool) { + for _, s := range Slots { + if s.Code == code { + return s, true + } + } + return Slot{}, false +} + +func OutputByCode(code string) (Output, bool) { + for _, o := range Outputs { + if o.Code == code { + return o, true + } + } + return Output{}, false +} + +// NodeForOutput returns the node producing the given output group. +func NodeForOutput(group string) (Node, bool) { + for _, n := range Nodes { + for _, o := range n.Outputs { + if o == group { + return n, true + } + } + } + return Node{}, false +} + +// Stages groups nodes by stage, ascending; within a stage nodes are sorted by code. +func Stages() [][]Node { + byStage := map[int][]Node{} + for _, n := range Nodes { + byStage[n.Stage] = append(byStage[n.Stage], n) + } + keys := make([]int, 0, len(byStage)) + for k := range byStage { + keys = append(keys, k) + } + sort.Ints(keys) + out := make([][]Node, 0, len(keys)) + for _, k := range keys { + ns := byStage[k] + sort.Slice(ns, func(i, j int) bool { return ns[i].Code < ns[j].Code }) + out = append(out, ns) + } + return out +} +``` + +- [ ] **Step 2: graph_test.go** — assert: every node input is either a slot or an output of a node with a *lower* stage; every output has exactly one producer; `Stages()` returns 4 stages with codes `[00 01 02 03] [10] [20 21] [30 31]`; all group codes are valid slugs (`^[a-z0-9]+(?:-[a-z0-9]+)*$`). Run → PASS once graph.go compiles (kernels may be stubs until Step 4; add empty kernel funcs first so it compiles). + +- [ ] **Step 3: params.go** + +```go +package landscape + +import ( + "errors" + "fmt" +) + +// Params are the run-time knobs of the graph (TZ §6): window size, tie +// priority for functional zoning, and the V1/V2 suitability thresholds. +type Params struct { + WindowSize int `json:"window_size"` + TiePriority []int `json:"tie_priority"` + V1MinSE12 int `json:"v1_min_s_e12"` + V1MinSE3 int `json:"v1_min_s_e3"` + V2MinSE3 int `json:"v2_min_s_e3"` + V2MinSE4 int `json:"v2_min_s_e4"` +} + +// DefaultParams returns the TZ defaults. +func DefaultParams() Params { + return Params{WindowSize: 2048, TiePriority: []int{1, 2, 3, 4, 5}, V1MinSE12: 3, V1MinSE3: 4, V2MinSE3: 3, V2MinSE4: 5} +} + +// WithDefaults fills zero-valued fields from DefaultParams. +func (p Params) WithDefaults() Params { + d := DefaultParams() + if p.WindowSize == 0 { + p.WindowSize = d.WindowSize + } + if len(p.TiePriority) == 0 { + p.TiePriority = d.TiePriority + } + if p.V1MinSE12 == 0 { + p.V1MinSE12 = d.V1MinSE12 + } + if p.V1MinSE3 == 0 { + p.V1MinSE3 = d.V1MinSE3 + } + if p.V2MinSE3 == 0 { + p.V2MinSE3 = d.V2MinSE3 + } + if p.V2MinSE4 == 0 { + p.V2MinSE4 = d.V2MinSE4 + } + return p +} + +// Validate checks ranges: window 256..8192, thresholds 1..5, tie_priority a +// permutation of 1..5. +func (p Params) Validate() error { + var errs []error + if p.WindowSize < 256 || p.WindowSize > 8192 { + errs = append(errs, fmt.Errorf("window_size must be in 256..8192")) + } + for name, v := range map[string]int{"v1_min_s_e12": p.V1MinSE12, "v1_min_s_e3": p.V1MinSE3, "v2_min_s_e3": p.V2MinSE3, "v2_min_s_e4": p.V2MinSE4} { + if v < 1 || v > 5 { + errs = append(errs, fmt.Errorf("%s must be in 1..5", name)) + } + } + seen := map[int]bool{} + for _, k := range p.TiePriority { + if k < 1 || k > 5 || seen[k] { + errs = append(errs, fmt.Errorf("tie_priority must be a permutation of 1..5")) + break + } + seen[k] = true + } + if len(p.TiePriority) != 5 && len(seen) == len(p.TiePriority) { + errs = append(errs, fmt.Errorf("tie_priority must list all of 1..5")) + } + return errors.Join(errs...) +} + +// ForNode keeps only the fields that influence the node's result (window size +// never does), so fingerprints do not change when irrelevant knobs change. +func (p Params) ForNode(code string) Params { + if code == "30" { + return Params{TiePriority: p.TiePriority, V1MinSE12: p.V1MinSE12, V1MinSE3: p.V1MinSE3, V2MinSE3: p.V2MinSE3, V2MinSE4: p.V2MinSE4} + } + return Params{} +} +``` + +Tests (`params_test.go`): defaults validate; `Params{}.WithDefaults()` equals `DefaultParams()`; `TiePriority: {1,1,2,3,4}` invalid; `WindowSize: 100` invalid; `ForNode("00")` is zero; `ForNode("30")` has no WindowSize. + +- [ ] **Step 4: Failing kernel tests** (`kernels_test.go`) — write these before the kernels: + +```go +package landscape + +import ( + "reflect" + "testing" +) + +func run(k Kernel, nOut int, in ...[]float64) [][]int16 { + out := make([][]int16, nOut) + for j := range out { + out[j] = make([]int16, len(in[0])) + } + k(in, out, DefaultParams()) + return out +} + +func TestClassify(t *testing.T) { + cases := map[float32]int16{1.0: 1, 1.8: 1, 1.81: 2, 2.6: 2, 2.75: 3, 3.4: 3, 3.5: 4, 4.2: 4, 4.25: 5, 5.0: 5} + for x, want := range cases { + if got := classify(x); got != want { + t.Errorf("classify(%v)=%d want %d", x, got, want) + } + } + if classify(0.9) != NoData || classify(5.1) != NoData { + t.Error("out of range must be NoData") + } +} + +func TestKernelZIntegrated(t *testing.T) { + // pixel0: all valid (1,2,3,4) -> 2.5 -> class 2 ; pixel1: one nodata -> NoData ; pixel2: all 5 -> 5 -> class 5 + got := run(kernelZIntegrated, 1, + []float64{1, -9999, 5}, []float64{2, 2, 5}, []float64{3, 3, 5}, []float64{4, 4, 5}) + if want := []int16{2, NoData, 5}; !reflect.DeepEqual(got[0], want) { + t.Fatalf("got %v want %v", got[0], want) + } +} + +func TestKernelUIntegrated_Applicability(t *testing.T) { + // order: water, soil, crop, veg + // p0: all valid 1,1,5,5 -> (0.25+0.25+1.25+1.25)/1.0 = 3.0 -> class 3 + // p1: crop invalid: 1,1,-,5 -> (0.25+0.25+1.25)/0.75 = 2.333 -> class 2 + // p2: crop and veg invalid -> NoData + // p3: water invalid -> NoData + got := run(kernelUIntegrated, 1, + []float64{1, 1, 1, -9999}, []float64{1, 1, 1, 1}, []float64{5, 0, 0, 5}, []float64{5, 5, 0, 5}) + if want := []int16{3, 2, NoData, NoData}; !reflect.DeepEqual(got[0], want) { + t.Fatalf("got %v want %v", got[0], want) + } +} + +func TestKernelZMax_RequiresAllValid(t *testing.T) { + got := run(kernelZMax, 1, []float64{1, 1}, []float64{4, 4}, []float64{2, -9999}, []float64{3, 3}) + if want := []int16{4, NoData}; !reflect.DeepEqual(got[0], want) { + t.Fatalf("got %v want %v", got[0], want) + } +} + +func TestKernelUMax_IgnoresInvalidCropVeg(t *testing.T) { + // water, soil, crop, veg ; p0: 1,2,-,5 -> 5 ; p1: 3,2,0,0 -> NoData (needs crop or veg) ; p2: 3,2,1,0 -> 3 + got := run(kernelUMax, 1, []float64{1, 3, 3}, []float64{2, 2, 2}, []float64{0, 0, 1}, []float64{5, 0, 0}) + if want := []int16{5, NoData, 3}; !reflect.DeepEqual(got[0], want) { + t.Fatalf("got %v want %v", got[0], want) + } +} + +func TestKernelE(t *testing.T) { + got := run(kernelE, 1, []float64{2, 5, -9999}, []float64{4, 1, 3}) + if want := []int16{4, 5, NoData}; !reflect.DeepEqual(got[0], want) { + t.Fatalf("got %v want %v", got[0], want) + } +} + +func TestKernelCConflicts(t *testing.T) { + // 5 P then E; p0: E=3 -> min(P,3); p1: E invalid -> all NoData; p2: P2 invalid + e := []float64{3, 0, 4} + got := run(kernelCConflicts, 5, + []float64{5, 5, 5}, []float64{1, 1, 7}, []float64{3, 3, 3}, []float64{4, 4, 4}, []float64{2, 2, 2}, e) + want := [][]int16{{3, NoData, 4}, {1, NoData, NoData}, {3, NoData, 3}, {3, NoData, 4}, {2, NoData, 2}} + if !reflect.DeepEqual(got, want) { + t.Fatalf("got %v want %v", got, want) + } +} + +func TestKernelMaxAnyValid(t *testing.T) { + got := run(kernelMaxAnyValid, 1, []float64{1, 0, 0}, []float64{4, 0, 2}, []float64{-9999, 0, 0}) + if want := []int16{4, NoData, 2}; !reflect.DeepEqual(got[0], want) { + t.Fatalf("got %v want %v", got[0], want) + } +} + +// TZ §7 edge cases. Inputs: E, then C1..C5 (irrigation, snow, crop, pasture, tourism). +func TestKernelFuncZoning_TZEdgeCases(t *testing.T) { + e := []float64{0, 3, 5, 4, 2, 2, 3} + s1 := []float64{3, 0, 3, 5, 3, 2, 4} + s2 := []float64{3, 0, 3, 5, 3, 2, 4} + s3 := []float64{3, 0, 3, 2, 3, 2, 5} + s4 := []float64{3, 0, 3, 2, 3, 2, 1} + s5 := []float64{3, 0, 3, 2, 3, 2, 1} + got := run(kernelFuncZoning, 3, e, s1, s2, s3, s4, s5) + // pixel: 0 E invalid -> NoData ; 1 all S invalid -> NoData ; 2 E=5 -> regulated(1) everywhere ; + // 3 E=4, S=5 for sectors 1,2 -> V1 regulated, V2/V3 sector 1 -> class 2 ; + // 4 all S equal (3), E=2 -> irrigation (2) in all ; 5 E=2, S=2 -> V1 regulated (S<3), V2 best=irrigation (2), V3 irrigation (2) + // 6 E=3: S=[4,4,5,1,1] -> V1 (S>=4): winners 1,2,3 max 5 -> crop (4); V2 (S>=3) -> crop (4); V3 -> crop (4) + want := [][]int16{ + {NoData, NoData, 1, 1, 2, 1, 4}, // V1 + {NoData, NoData, 1, 2, 2, 2, 4}, // V2 + {NoData, NoData, 1, 2, 2, 2, 4}, // V3 + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("got %v want %v", got, want) + } +} + +func TestKernelFuncZoning_TiePriorityIsConfigurable(t *testing.T) { + in := [][]float64{{2}, {3}, {3}, {3}, {3}, {3}} + out := [][]int16{{0}, {0}, {0}} + p := DefaultParams() + p.TiePriority = []int{5, 4, 3, 2, 1} + kernelFuncZoning(in, out, p) + if out[0][0] != 6 || out[1][0] != 6 || out[2][0] != 6 { + t.Fatalf("expected tourism (6) everywhere, got %v", out) + } +} +``` + +Run: `go test ./internal/landscape/ -run Kernel -v` → FAIL (undefined). + +- [ ] **Step 5: kernels.go** + +```go +package landscape + +// valid reports whether a class value is in the analysis domain (1..5). Any +// other value, including nodata and NaN, is invalid. +func valid(v float64) bool { return v >= 1 && v <= 5 } + +// classify maps a float32 score onto classes 1..5 with the fixed intervals +// used by the reference notebook (comparisons in float32, like numpy on a +// float32 array). Scores outside [1, 5] yield NoData. +func classify(x float32) int16 { + switch { + case x < 1.0: + return NoData + case x <= 1.8: + return 1 + case x <= 2.6: + return 2 + case x <= 3.4: + return 3 + case x <= 4.2: + return 4 + case x <= 5.0: + return 5 + default: + return NoData + } +} + +// kernelZIntegrated: all four Z valid -> classify(0.25*Z1 + 0.25*Z2 + 0.25*Z3 + 0.25*Z4). +func kernelZIntegrated(in [][]float64, out [][]int16, _ Params) { + for i := range out[0] { + z1, z2, z3, z4 := in[0][i], in[1][i], in[2][i], in[3][i] + if !(valid(z1) && valid(z2) && valid(z3) && valid(z4)) { + out[0][i] = NoData + continue + } + score := float32(0.25)*float32(z1) + float32(0.25)*float32(z2) + float32(0.25)*float32(z3) + float32(0.25)*float32(z4) + out[0][i] = classify(score) + } +} + +// kernelUIntegrated: water and soil required, at least one of crop/veg; +// U = (0.25*water + 0.25*soil + [0.25*crop] + [0.25*veg]) / (0.5 + [0.25] + [0.25]). +func kernelUIntegrated(in [][]float64, out [][]int16, _ Params) { + for i := range out[0] { + water, soil, crop, veg := in[0][i], in[1][i], in[2][i], in[3][i] + cropOK, vegOK := valid(crop), valid(veg) + if !(valid(water) && valid(soil) && (cropOK || vegOK)) { + out[0][i] = NoData + continue + } + num := float32(0.25)*float32(water) + float32(0.25)*float32(soil) + den := float32(0.5) + if cropOK { + num += float32(0.25) * float32(crop) + den += 0.25 + } + if vegOK { + num += float32(0.25) * float32(veg) + den += 0.25 + } + out[0][i] = classify(num / den) + } +} + +// kernelZMax: all four valid -> max. +func kernelZMax(in [][]float64, out [][]int16, _ Params) { + for i := range out[0] { + m, ok := int16(0), true + for _, layer := range in { + v := layer[i] + if !valid(v) { + ok = false + break + } + if int16(v) > m { + m = int16(v) + } + } + if !ok { + m = NoData + } + out[0][i] = m + } +} + +// kernelUMax: water and soil required plus one of crop/veg; max over valid layers. +func kernelUMax(in [][]float64, out [][]int16, _ Params) { + for i := range out[0] { + water, soil, crop, veg := in[0][i], in[1][i], in[2][i], in[3][i] + cropOK, vegOK := valid(crop), valid(veg) + if !(valid(water) && valid(soil) && (cropOK || vegOK)) { + out[0][i] = NoData + continue + } + m := max(int16(water), int16(soil)) + if cropOK { + m = max(m, int16(crop)) + } + if vegOK { + m = max(m, int16(veg)) + } + out[0][i] = m + } +} + +// kernelE: Z and U valid -> max(Z, U). +func kernelE(in [][]float64, out [][]int16, _ Params) { + for i := range out[0] { + z, u := in[0][i], in[1][i] + if !(valid(z) && valid(u)) { + out[0][i] = NoData + continue + } + out[0][i] = max(int16(z), int16(u)) + } +} + +// kernelCConflicts: inputs P1..P5, E; output j = min(Pj, E) where both valid. +func kernelCConflicts(in [][]float64, out [][]int16, _ Params) { + e := in[5] + for j := 0; j < 5; j++ { + p := in[j] + for i := range out[j] { + if !(valid(p[i]) && valid(e[i])) { + out[j][i] = NoData + continue + } + out[j][i] = min(int16(p[i]), int16(e[i])) + } + } +} + +// kernelMaxAnyValid: max over the valid inputs; NoData when none is valid. +// Used for P_max and C_max. +func kernelMaxAnyValid(in [][]float64, out [][]int16, _ Params) { + for i := range out[0] { + m, any := int16(0), false + for _, layer := range in { + if v := layer[i]; valid(v) { + any = true + m = max(m, int16(v)) + } + } + if !any { + m = NoData + } + out[0][i] = m + } +} + +// kernelFuncZoning implements TZ_functional_zoning_go.md §4. Inputs: E, then +// S1..S5 (irrigation, snow, crop, pasture, tourism). Outputs: V1, V2, V3. +func kernelFuncZoning(in [][]float64, out [][]int16, p Params) { + e := in[0] + rules := [3]func(e, s int16) bool{ + func(e, s int16) bool { return (e <= 2 && s >= int16(p.V1MinSE12)) || (e == 3 && s >= int16(p.V1MinSE3)) }, + func(e, s int16) bool { return e <= 2 || (e == 3 && s >= int16(p.V2MinSE3)) || (e == 4 && s >= int16(p.V2MinSE4)) }, + func(e, s int16) bool { return e <= 4 }, + } + var s [5]int16 + var base [5]bool + for i := range out[0] { + if !valid(e[i]) { + out[0][i], out[1][i], out[2][i] = NoData, NoData, NoData + continue + } + ev := int16(e[i]) + anyBase := false + for k := 0; k < 5; k++ { + base[k] = valid(in[k+1][i]) + if base[k] { + s[k] = int16(in[k+1][i]) + anyBase = true + } + } + if !anyBase { + out[0][i], out[1][i], out[2][i] = NoData, NoData, NoData + continue + } + for v := 0; v < 3; v++ { + var cand [5]bool + maxS, anyCand := int16(0), false + for k := 0; k < 5; k++ { + cand[k] = base[k] && rules[v](ev, s[k]) + if cand[k] { + anyCand = true + if s[k] > maxS { + maxS = s[k] + } + } + } + if !anyCand { + out[v][i] = 1 // regulated + continue + } + sector := int16(0) + for _, code := range p.TiePriority { + k := code - 1 + if cand[k] && s[k] == maxS { + sector = int16(code) + break + } + } + out[v][i] = sector + 1 + } + } +} +``` + +- [ ] **Step 6: Run** `go test ./internal/landscape/ -v` → PASS. `go vet ./internal/landscape/` clean. + +- [ ] **Step 7: Commit** + +```bash +git add internal/landscape/graph.go internal/landscape/graph_test.go internal/landscape/params.go internal/landscape/params_test.go internal/landscape/kernels.go internal/landscape/kernels_test.go +git commit -m "feat(landscape): computation graph, params and per-pixel kernels" +``` + +--- + +### Task 5: Landscape domain, migration 00011 (seed + run tables), repository + +**Files:** +- Create: `internal/domain/landscape.go`, `migrations/00011_landscape.sql`, `internal/repository/postgres/landscape.go` + +**Interfaces (produced, used by Task 7):** + +```go +package domain +type LPRunStatus string // "queued" "running" "succeeded" "failed" +type LPNodeStatus string // "pending" "running" "skipped" "succeeded" "failed" +type LPDatasetRef struct { Group string `json:"group"`; DatasetID uuid.UUID `json:"dataset_id"` } +type LPRun struct { ID uuid.UUID; Status LPRunStatus; FromStage *int; Params json.RawMessage; Overrides map[string]uuid.UUID; Error *string; CreatedAt time.Time; StartedAt, FinishedAt *time.Time; Nodes []LPRunNode } +type LPRunNode struct { ID, RunID uuid.UUID; NodeCode string; Stage int; Status LPNodeStatus; Fingerprint string; Inputs []LPDatasetRef; Outputs []LPDatasetRef; Error *string; StartedAt, FinishedAt *time.Time } + +package postgres +type LandscapeRepository — methods: + CreateRun(ctx, run domain.LPRun) (domain.LPRun, error) // inserts run + its Nodes (status pending) in one tx; returns with ids + GetRun(ctx, id uuid.UUID) (domain.LPRun, error) // with nodes ordered by node_code + ListRuns(ctx, limit, offset int) ([]domain.LPRun, error) // without nodes, newest first + CountRuns(ctx) (int, error) + SetRunStatus(ctx, id uuid.UUID, status domain.LPRunStatus, errMsg *string) error // running -> started_at=now(); succeeded/failed -> finished_at=now() + SetNodeStarted(ctx, runID uuid.UUID, code string, fingerprint string, inputs []domain.LPDatasetRef) error // status running, started_at + SetNodeFinished(ctx, runID uuid.UUID, code string, status domain.LPNodeStatus, outputs []domain.LPDatasetRef, errMsg *string) error // finished_at + FindReusableOutputs(ctx, code, fingerprint string, excludeRun uuid.UUID) ([]domain.LPDatasetRef, error) // latest succeeded node with same code+fingerprint; ErrNotFound if none +``` + +- [ ] **Step 1: Domain** (`internal/domain/landscape.go`) — write the types above with JSON tags: `id, status, from_stage, params, overrides, error, created_at, started_at, finished_at, nodes` and `id, run_id, node_code, stage, status, fingerprint, inputs, outputs, error, started_at, finished_at`. Add constants `LPRunQueued/Running/Succeeded/Failed` and `LPNodePending/Running/Skipped/Succeeded/Failed`. + +- [ ] **Step 2: Migration** (`migrations/00011_landscape.sql`) + +```sql +-- +goose Up +-- Landscape Program: seed the category and dataset groups (input slots and +-- node outputs; codes must match internal/landscape/graph.go), and store runs. +INSERT INTO categories (code, name, description) +VALUES ('landscape', 'Ландшафтная программа', 'Входные и результирующие растры Ландшафтной программы') +ON CONFLICT (code) DO NOTHING; + +INSERT INTO dataset_groups (code, name) VALUES + ('lp-z-biodiversity', 'Z Biodiversity'), ('lp-z-lcr', 'Z LCR'), ('lp-z-soil-protection', 'Z Soil Protection'), ('lp-z-wr', 'Z Water Resources'), + ('lp-u-surf-water', 'U Surface Water'), ('lp-u-soil', 'U Soil'), ('lp-u-crop', 'U Crop'), ('lp-u-veg', 'U Vegetation'), + ('lp-p-irrigation', 'P Irrigation'), ('lp-p-snow', 'P Snow'), ('lp-p-crop', 'P Crop'), ('lp-p-pasture', 'P Pasture'), ('lp-p-tourism', 'P Tourism'), + ('lp-z-integrated', 'Z Integrated'), ('lp-u-integrated', 'U Integrated'), ('lp-z-max', 'Z Max'), ('lp-u-max', 'U Max'), + ('lp-e', 'E Ecological Constraint'), + ('lp-c-irrigation', 'C Irrigation'), ('lp-c-snow', 'C Snow'), ('lp-c-crop', 'C Crop'), ('lp-c-pasture', 'C Pasture'), ('lp-c-tourism', 'C Tourism'), + ('lp-p-max', 'P Max'), + ('lp-v1-conservative', 'V1 Conservative'), ('lp-v2-balanced', 'V2 Balanced'), ('lp-v3-development', 'V3 Development'), ('lp-c-max', 'C Max') +ON CONFLICT (code) DO NOTHING; + +CREATE TABLE lp_runs ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + status VARCHAR(32) NOT NULL, + from_stage SMALLINT, + params JSONB NOT NULL, + overrides JSONB NOT NULL DEFAULT '{}'::jsonb, + error TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + started_at TIMESTAMPTZ, + finished_at TIMESTAMPTZ +); +CREATE INDEX idx_lp_runs_created_at ON lp_runs (created_at DESC); + +CREATE TABLE lp_run_nodes ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + run_id UUID NOT NULL REFERENCES lp_runs (id) ON DELETE CASCADE, + node_code VARCHAR(8) NOT NULL, + stage SMALLINT NOT NULL, + status VARCHAR(32) NOT NULL, + fingerprint TEXT NOT NULL DEFAULT '', + input_datasets JSONB NOT NULL DEFAULT '[]'::jsonb, + output_datasets JSONB, + error TEXT, + started_at TIMESTAMPTZ, + finished_at TIMESTAMPTZ, + UNIQUE (run_id, node_code) +); +CREATE INDEX idx_lp_run_nodes_lookup ON lp_run_nodes (node_code, fingerprint, status, finished_at DESC); + +-- +goose Down +DROP TABLE lp_run_nodes; +DROP TABLE lp_runs; +-- Seeded category/groups are left in place (datasets may reference them). +``` + +- [ ] **Step 3: Repository** (`internal/repository/postgres/landscape.go`) — implement the methods listed in Interfaces. Key SQL: + +```sql +-- CreateRun (tx): +INSERT INTO lp_runs (status, from_stage, params, overrides) VALUES ($1,$2,$3,$4) RETURNING +INSERT INTO lp_run_nodes (run_id, node_code, stage, status) VALUES ($1,$2,$3,'pending') RETURNING -- per node +-- GetRun: SELECT FROM lp_runs WHERE id=$1 ; SELECT FROM lp_run_nodes WHERE run_id=$1 ORDER BY node_code +-- ListRuns: ... ORDER BY created_at DESC LIMIT $1 OFFSET $2 +-- SetRunStatus: +UPDATE lp_runs SET status=$2, error=$3, + started_at = CASE WHEN $2 = 'running' THEN now() ELSE started_at END, + finished_at = CASE WHEN $2 IN ('succeeded','failed') THEN now() ELSE finished_at END + WHERE id=$1 +-- SetNodeStarted: +UPDATE lp_run_nodes SET status='running', fingerprint=$3, input_datasets=$4, started_at=now(), error=NULL WHERE run_id=$1 AND node_code=$2 +-- SetNodeFinished: +UPDATE lp_run_nodes SET status=$3, output_datasets=$4, error=$5, finished_at=now() WHERE run_id=$1 AND node_code=$2 +-- FindReusableOutputs: +SELECT output_datasets FROM lp_run_nodes + WHERE node_code=$1 AND fingerprint=$2 AND status='succeeded' AND run_id <> $3 AND output_datasets IS NOT NULL + ORDER BY finished_at DESC LIMIT 1 +``` + +`runColumns = id, status, from_stage, params, overrides, error, created_at, started_at, finished_at`; scan `overrides` into `map[string]uuid.UUID` via a `[]byte` + `json.Unmarshal` (empty → empty map). `nodeColumns = id, run_id, node_code, stage, status, fingerprint, input_datasets, output_datasets, error, started_at, finished_at`; JSON arrays scanned via pgx into `[]domain.LPDatasetRef` (pgx handles `jsonb` → struct slice; `output_datasets` NULL → nil). Every `Exec` with `RowsAffected()==0` returns `domain.ErrNotFound`; wrap errors with `mapError`. + +- [ ] **Step 4: Compile** — `go build ./internal/domain/ ./internal/repository/postgres/`; `go vet` both. (No DB in CI; SQL is exercised by the integration task.) + +- [ ] **Step 5: Commit** + +```bash +git add internal/domain/landscape.go migrations/00011_landscape.sql internal/repository/postgres/landscape.go +git commit -m "feat(landscape): run/node domain, seed migration, repository" +``` + +--- + +### Task 6: `Compute` — window executor over `raster.IO` + +**Files:** +- Create: `internal/landscape/compute.go`, `internal/landscape/compute_test.go`, `internal/landscape/fakeio_test.go` + +**Interfaces:** +- Consumes: Task 3 `raster.IO/Reader/Writer/Grid/Window/Windows/DefaultWriterOptions`; Task 4 `Node`, `Params`. +- Produces (used by Task 7 via `service.ComputeFunc`): + +```go +// Compute runs one node. inputs maps every Node.Inputs group code to a local +// file; outputs are written to filepath.Join(outDir, group+".tif") and returned +// keyed by group code. Grids of all inputs must be identical (raster.IO.SameGrid) +// or an error is returned before any output is created. +func Compute(ctx context.Context, rio raster.IO, node Node, inputs map[string]string, outDir string, p Params) (map[string]string, error) +``` + +- [ ] **Step 1: Fake IO for tests** (`fakeio_test.go`) + +```go +package landscape + +import ( + "context" + "fmt" + "sync" + + "gis/internal/raster" +) + +// fakeIO keeps rasters in memory keyed by path. +type fakeIO struct { + mu sync.Mutex + files map[string]*fakeRaster +} + +type fakeRaster struct { + grid raster.Grid + data []float64 // row-major +} + +func newFakeIO() *fakeIO { return &fakeIO{files: map[string]*fakeRaster{}} } + +func (f *fakeIO) put(path string, g raster.Grid, data []float64) { + f.files[path] = &fakeRaster{grid: g, data: data} +} + +type fakeReader struct{ r *fakeRaster } + +func (r *fakeReader) Grid() raster.Grid { return r.r.grid } +func (r *fakeReader) NoData() (float64, bool) { return -9999, true } +func (r *fakeReader) Close() error { return nil } +func (r *fakeReader) ReadFloat64(w raster.Window, dst []float64) error { + for y := 0; y < w.Height; y++ { + copy(dst[y*w.Width:(y+1)*w.Width], r.r.data[(w.Y+y)*r.r.grid.Width+w.X:(w.Y+y)*r.r.grid.Width+w.X+w.Width]) + } + return nil +} + +type fakeWriter struct { + f *fakeIO + r *fakeRaster +} + +func (w *fakeWriter) WriteInt16(win raster.Window, src []int16) error { + w.f.mu.Lock() + defer w.f.mu.Unlock() + for y := 0; y < win.Height; y++ { + for x := 0; x < win.Width; x++ { + w.r.data[(win.Y+y)*w.r.grid.Width+win.X+x] = float64(src[y*win.Width+x]) + } + } + return nil +} +func (w *fakeWriter) Close() error { return nil } + +func (f *fakeIO) Open(_ context.Context, path string) (raster.Reader, error) { + f.mu.Lock() + defer f.mu.Unlock() + r, ok := f.files[path] + if !ok { + return nil, fmt.Errorf("fake: %s not found", path) + } + return &fakeReader{r: r}, nil +} + +func (f *fakeIO) Create(_ context.Context, path string, g raster.Grid, _ raster.WriterOptions) (raster.Writer, error) { + f.mu.Lock() + defer f.mu.Unlock() + r := &fakeRaster{grid: g, data: make([]float64, g.Width*g.Height)} + f.files[path] = r + return &fakeWriter{f: f, r: r}, nil +} + +func (f *fakeIO) SameGrid(a, b raster.Grid) bool { return a == b } +``` + +- [ ] **Step 2: Failing tests** (`compute_test.go`) + +```go +package landscape + +import ( + "context" + "path/filepath" + "testing" + + "gis/internal/raster" +) + +func grid(w, h int) raster.Grid { + return raster.Grid{Width: w, Height: h, GeoTransform: [6]float64{0, 30, 0, 0, 0, -30}, ProjectionWKT: "EPSG:32642"} +} + +func constRaster(w, h int, v float64) []float64 { + d := make([]float64, w*h) + for i := range d { + d[i] = v + } + return d +} + +func TestCompute_E_TilesWindowsAndWritesEveryPixel(t *testing.T) { + fio := newFakeIO() + g := grid(10, 7) + z := constRaster(10, 7, 2) + u := constRaster(10, 7, 4) + z[0] = -9999 // pixel 0 invalid + fio.put("z.tif", g, z) + fio.put("u.tif", g, u) + node, _ := NodeByCode("10") + p := DefaultParams() + p.WindowSize = 4 // forces partial windows (10x7 with 4x4 tiles) + out, err := Compute(context.Background(), fio, node, map[string]string{OutZIntegrated: "z.tif", OutUIntegrated: "u.tif"}, "/out", p) + if err != nil { + t.Fatal(err) + } + path := out[OutE] + if path != filepath.Join("/out", OutE+".tif") { + t.Fatalf("unexpected output path %q", path) + } + res := fio.files[path] + if res.grid != g { + t.Fatal("output grid must copy the first input's grid") + } + for i, v := range res.data { + want := 4.0 + if i == 0 { + want = -9999 + } + if v != want { + t.Fatalf("pixel %d = %v want %v", i, v, want) + } + } +} + +func TestCompute_FailsOnGridMismatch(t *testing.T) { + fio := newFakeIO() + fio.put("z.tif", grid(10, 7), constRaster(10, 7, 2)) + fio.put("u.tif", grid(9, 7), constRaster(9, 7, 4)) + node, _ := NodeByCode("10") + _, err := Compute(context.Background(), fio, node, map[string]string{OutZIntegrated: "z.tif", OutUIntegrated: "u.tif"}, "/out", DefaultParams()) + if err == nil { + t.Fatal("expected grid mismatch error") + } + if _, created := fio.files[filepath.Join("/out", OutE+".tif")]; created { + t.Fatal("no output must be created on mismatch") + } +} + +func TestCompute_FailsOnMissingInput(t *testing.T) { + fio := newFakeIO() + fio.put("z.tif", grid(2, 2), constRaster(2, 2, 2)) + node, _ := NodeByCode("10") + if _, err := Compute(context.Background(), fio, node, map[string]string{OutZIntegrated: "z.tif"}, "/out", DefaultParams()); err == nil { + t.Fatal("expected error for missing input group") + } +} + +func TestCompute_MultiOutputNode(t *testing.T) { + fio := newFakeIO() + g := grid(3, 1) + paths := map[string]string{} + for i, s := range []string{SlotPIrrigation, SlotPSnow, SlotPCrop, SlotPPasture, SlotPTourism} { + paths[s] = s + ".tif" + fio.put(paths[s], g, constRaster(3, 1, float64(i+1))) + } + paths[OutE] = "e.tif" + fio.put("e.tif", g, constRaster(3, 1, 3)) + node, _ := NodeByCode("20") + out, err := Compute(context.Background(), fio, node, paths, "/o", DefaultParams()) + if err != nil { + t.Fatal(err) + } + if len(out) != 5 { + t.Fatalf("want 5 outputs, got %d", len(out)) + } + if fio.files[out[OutCTourism]].data[0] != 3 || fio.files[out[OutCIrrigation]].data[0] != 1 { + t.Fatal("min(P, E) not applied per output") + } +} +``` + +Run: `go test ./internal/landscape/ -run Compute -v` → FAIL. + +- [ ] **Step 3: compute.go** + +```go +package landscape + +import ( + "context" + "fmt" + "path/filepath" + "runtime" + "sync" + + "gis/internal/raster" + + "golang.org/x/sync/errgroup" +) + +// Compute runs one node over its input rasters. See the package doc for the +// concurrency model: windows are processed by a worker pool; each worker owns +// its own Reader handles (GDAL datasets are not goroutine-safe) and writes are +// serialised per output. Window order does not affect the result. +func Compute(ctx context.Context, rio raster.IO, node Node, inputs map[string]string, outDir string, p Params) (map[string]string, error) { + p = p.WithDefaults() + paths := make([]string, len(node.Inputs)) + for i, g := range node.Inputs { + path, ok := inputs[g] + if !ok || path == "" { + return nil, fmt.Errorf("node %s (%s): missing input %q", node.Code, node.Name, g) + } + paths[i] = path + } + + // Validate grids with one set of handles before creating anything. + grid, err := checkGrids(ctx, rio, node, paths) + if err != nil { + return nil, err + } + + outPaths := make(map[string]string, len(node.Outputs)) + writers := make([]raster.Writer, len(node.Outputs)) + for j, g := range node.Outputs { + path := filepath.Join(outDir, g+".tif") + w, err := rio.Create(ctx, path, grid, raster.DefaultWriterOptions()) + if err != nil { + closeWriters(writers[:j]) + return nil, fmt.Errorf("node %s: create output %s: %w", node.Code, g, err) + } + writers[j] = w + outPaths[g] = path + } + + windows := raster.Windows(grid.Width, grid.Height, p.WindowSize) + jobs := make(chan raster.Window) + var writeMu sync.Mutex + eg, gctx := errgroup.WithContext(ctx) + workers := min(runtime.GOMAXPROCS(0), len(windows)) + for w := 0; w < workers; w++ { + eg.Go(func() error { + readers, err := openAll(gctx, rio, paths) + if err != nil { + return err + } + defer closeReaders(readers) + size := p.WindowSize * p.WindowSize + in := make([][]float64, len(paths)) + for i := range in { + in[i] = make([]float64, size) + } + out := make([][]int16, len(writers)) + for j := range out { + out[j] = make([]int16, size) + } + for win := range jobs { + n := win.Width * win.Height + inView := make([][]float64, len(in)) + for i, r := range readers { + if err := r.ReadFloat64(win, in[i][:n]); err != nil { + return fmt.Errorf("node %s: read %s window %+v: %w", node.Code, node.Inputs[i], win, err) + } + inView[i] = in[i][:n] + } + outView := make([][]int16, len(out)) + for j := range out { + outView[j] = out[j][:n] + } + node.Kernel(inView, outView, p) + writeMu.Lock() + for j, wr := range writers { + if err := wr.WriteInt16(win, outView[j]); err != nil { + writeMu.Unlock() + return fmt.Errorf("node %s: write %s window %+v: %w", node.Code, node.Outputs[j], win, err) + } + } + writeMu.Unlock() + } + return nil + }) + } + eg.Go(func() error { + defer close(jobs) + for _, win := range windows { + select { + case jobs <- win: + case <-gctx.Done(): + return gctx.Err() + } + } + return nil + }) + runErr := eg.Wait() + if err := closeWriters(writers); err != nil && runErr == nil { + runErr = err + } + if runErr != nil { + return nil, runErr + } + return outPaths, nil +} + +// checkGrids opens every input once, verifies all grids match the first, and +// returns that grid. +func checkGrids(ctx context.Context, rio raster.IO, node Node, paths []string) (raster.Grid, error) { + readers, err := openAll(ctx, rio, paths) + if err != nil { + return raster.Grid{}, err + } + defer closeReaders(readers) + ref := readers[0].Grid() + for i := 1; i < len(readers); i++ { + if g := readers[i].Grid(); !rio.SameGrid(ref, g) { + return raster.Grid{}, fmt.Errorf("node %s: input %s grid (%dx%d, %v) differs from %s (%dx%d, %v); inputs must be aligned", + node.Code, node.Inputs[i], g.Width, g.Height, g.GeoTransform, node.Inputs[0], ref.Width, ref.Height, ref.GeoTransform) + } + } + return ref, nil +} + +func openAll(ctx context.Context, rio raster.IO, paths []string) ([]raster.Reader, error) { + readers := make([]raster.Reader, 0, len(paths)) + for _, p := range paths { + r, err := rio.Open(ctx, p) + if err != nil { + closeReaders(readers) + return nil, err + } + readers = append(readers, r) + } + return readers, nil +} + +func closeReaders(rs []raster.Reader) { + for _, r := range rs { + if r != nil { + _ = r.Close() + } + } +} + +func closeWriters(ws []raster.Writer) error { + var first error + for _, w := range ws { + if w == nil { + continue + } + if err := w.Close(); err != nil && first == nil { + first = err + } + } + return first +} +``` + +- [ ] **Step 4: Run** `go test ./internal/landscape/ -race -v` → PASS (race detector must be clean). + +- [ ] **Step 5: Commit** + +```bash +git add internal/landscape/compute.go internal/landscape/compute_test.go internal/landscape/fakeio_test.go +git commit -m "feat(landscape): window executor Compute with worker pool and grid checks" +``` + +--- + +### Task 7: LandscapeService (runs, planning, input resolution, execution) + RabbitMQ job + +**Files:** +- Create: `internal/service/landscape.go`, `internal/service/landscape_test.go`, `internal/messaging/rabbitmq/landscape_consumer.go` +- Modify: none outside these (wiring is Task 8) + +**Interfaces:** +- Consumes: Task 2 (`DatasetRepository.LatestInGroup`, `DatasetGroupRepository.GetByCode`, `DatasetService.Upload/CreateRasterFromFile`, `ObjectStore`), Task 4 (`landscape.Nodes/Stages/Slots/Outputs/Params/…`), Task 5 (`domain.LPRun*`, repo methods), `raster.IO` (Task 3). +- Produces (used by Task 8): + +```go +package service +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) +type LandscapeRepository interface { /* the 8 methods of postgres.LandscapeRepository, same signatures */ } +type LandscapeJobEnqueuer interface { EnqueueRun(ctx context.Context, runID uuid.UUID) error } +type datasetPublisher interface { + Upload(ctx context.Context, in UploadInput) (domain.Dataset, error) + CreateRasterFromFile(ctx context.Context, in CreateRasterInput) (domain.Dataset, error) +} +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 + +type SlotInfo struct { Code string `json:"code"`; Title string `json:"title"`; GroupID uuid.UUID `json:"group_id"`; Latest *domain.DatasetSummary `json:"latest"` } +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"` } +type SlotUploadInput struct { Filename, ContentType string; Size int64; Reader io.Reader; Name string; StartDate, EndDate *string } +type RunInput struct { FromStage *int; Overrides map[string]uuid.UUID; Params landscape.Params } +type LPRunPage struct { Items []domain.LPRun; Page, PageSize, Total int } + +func (s *LandscapeService) Slots(ctx) ([]SlotInfo, error) +func (s *LandscapeService) UploadToSlot(ctx, code string, in SlotUploadInput) (domain.Dataset, error) // ErrNotFound for unknown slot +func (s *LandscapeService) CreateRun(ctx, in RunInput) (domain.LPRun, error) +func (s *LandscapeService) GetRun(ctx, id uuid.UUID) (domain.LPRun, error) +func (s *LandscapeService) ListRuns(ctx, page, pageSize int) (LPRunPage, error) +func (s *LandscapeService) Results(ctx) ([]ResultInfo, error) +func (s *LandscapeService) Execute(ctx, runID uuid.UUID) error // worker entry point +func (s *LandscapeService) MarkRunFailed(ctx, runID uuid.UUID, reason string) error + +package rabbitmq +const LandscapeRunRoutingKey = "landscape.run"; const LandscapeRunQueue = "gis.landscape.run" +type LandscapeJob struct { RunID uuid.UUID `json:"run_id"` } +type LandscapeJobPublisher struct{ pub *Publisher }; func NewLandscapeJobPublisher(pub *Publisher) *LandscapeJobPublisher; func (p *LandscapeJobPublisher) EnqueueRun(ctx, runID uuid.UUID) error +type LandscapeProcessor interface { Execute(ctx, runID uuid.UUID) error; MarkRunFailed(ctx, runID uuid.UUID, reason string) error } +func NewLandscapeRunHandler(p LandscapeProcessor, log *slog.Logger) Handler +``` + +- [ ] **Step 1: Failing service tests** (`internal/service/landscape_test.go`). Build stubs: `stubLPRepo` (in-memory runs/nodes, records `FindReusableOutputs` hits), reuse `stubDatasetRepo`, `stubGroupRepo` (seed all `landscape.Slots`/`Outputs` codes as groups), `stubCategoryRepo` (from `category_test.go`, seed code `landscape`), a `stubPublisher` implementing `datasetPublisher` that records `CreateRasterFromFile` calls and inserts a ready raster dataset with the given group into `stubDatasetRepo`, `stubStore` with `Get` returning bytes, `stubLPJobs`, `fakeIO`-free approach: `compute` is a `ComputeFunc` stub that writes an empty file per output and records the `inputs` map it received. + +Tests: + +```go +func TestLandscapeService_CreateRun_ValidatesParamsAndOverrides(t *testing.T) +// invalid tie priority -> ErrValidation; override for unknown group -> ErrValidation; +// override dataset not ready/not raster -> ErrValidation; valid -> run queued with 9 pending nodes and job enqueued. + +func TestLandscapeService_Execute_ResolvesInputsAndPublishesOutputs(t *testing.T) +// seed ready raster datasets for all 13 slots; Execute; expect: run succeeded, 9 nodes succeeded, +// compute stub called 9 times, node "10" received inputs produced by nodes "00" and "01" in this run +// (dataset ids equal to the publisher's created ids), each published dataset has Automated=true, +// DatasetGroupID = its output group, Meta contains "lp_run_id". + +func TestLandscapeService_Execute_ReusesFingerprintMatches(t *testing.T) +// run once (all computed); run again with unchanged inputs -> every node skipped, compute stub not called, +// outputs equal to first run's outputs. + +func TestLandscapeService_Execute_RecomputesDownstreamOfChangedInput(t *testing.T) +// after first run, upload a newer dataset to lp-p-crop (later end_date); second run: nodes 00,01,02,03,10 skipped; +// 20,21 computed (inputs changed); 30,31 computed (their inputs are new C datasets). + +func TestLandscapeService_Execute_FromStageForcesRecompute(t *testing.T) +// second run with FromStage=2 and unchanged inputs -> stages 0,1 skipped, stages 2,3 computed. + +func TestLandscapeService_Execute_MissingSlotFailsRun(t *testing.T) +// no dataset in lp-z-wr -> node 00 and 02 failed with message containing "lp-z-wr"; run failed; stage 1 never started. + +func TestLandscapeService_UploadToSlot_UnknownSlot(t *testing.T) // ErrNotFound +func TestLandscapeService_Results_ListsLatestPerOutput(t *testing.T) // cog_url = "/datasets/.cog"; sld_url only when group has SLD +``` + +Write the test bodies fully (arrange/act/assert) following `dataset_test.go` style. Run → FAIL (undefined). + +- [ ] **Step 2: Service implementation** (`internal/service/landscape.go`) — implement per spec §3.5–3.7. Core pieces: + +```go +// 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[:]) +} +``` + +`Execute(ctx, runID)`: +1. `run := runs.GetRun`; parse `run.Params` into `landscape.Params` (`WithDefaults`); `runs.SetRunStatus(running)`. +2. `produced := map[string]uuid.UUID{}` guarded by a mutex (outputs created in this run). +3. For each stage from `landscape.Stages()`: `errgroup` over nodes → `s.runNode(ctx, run, node, params, produced)`; on error: `SetRunStatus(failed, err)` and `return err`. +4. `SetRunStatus(succeeded, nil)`. + +`runNode`: +1. Resolve inputs: for each group `g` in `node.Inputs`: `run.Overrides[g]` → else `produced[g]` → else `datasets.LatestInGroup(ctx, groupID(g), &ready, &raster)`; `groupID` via `groups.GetByCode`. Missing → `fmt.Errorf("node %s: no ready raster dataset in group %q", …)`. +2. `fp := fingerprint(node, refs, params)`; `runs.SetNodeStarted(run.ID, node.Code, fp, refs)`. +3. If `run.FromStage == nil || node.Stage < *run.FromStage`: `outs, err := runs.FindReusableOutputs(ctx, node.Code, fp, run.ID)`; if found and every `outs[i].DatasetID` exists with `Status == ready` (`datasets.GetByID`) → record into `produced`, `SetNodeFinished(skipped, outs, nil)`, return. +4. Temp dir `os.MkdirTemp("", "lp-"+node.Code+"-*")` (deferred `RemoveAll`); download each input object (`store.Get(dataset.StorageKey)` → file `/in-.tif`). +5. `outPaths, err := s.compute(ctx, s.rio, node, inputPaths, tmp, params)`. +6. For each output group: `publisher.CreateRasterFromFile(ctx, CreateRasterInput{CategoryID: landscapeCategory.ID, DatasetGroupID: &gid, Name: title + " " + run.StartedAt(or now).Format("2006-01-02"), Filename: group + ".tif", Path: outPaths[group], StartDate: minStart(inputs), EndDate: maxEnd(inputs), Meta: {"lp_run_id", "lp_node", "inputs"}, Automated: true})`; collect refs; `produced[group] = ds.ID`. +7. `SetNodeFinished(succeeded, outRefs, nil)`. On any error: `SetNodeFinished(failed, nil, &msg)` and return the error. + +`CreateRun`: `params := in.Params.WithDefaults(); params.Validate()` → wrap into `ErrValidation`; `FromStage` in 0..3; each override: group code must be a known group (`landscape.SlotByCode` or `OutputByCode`) and the dataset must exist, be `raster` and `ready` (else `ErrValidation`); marshal params; build `domain.LPRun{Status: queued, FromStage, Params, Overrides, Nodes: one pending node per landscape.Nodes}`; `runs.CreateRun`; `jobs.EnqueueRun(run.ID)` (on failure `SetRunStatus(failed, "failed to enqueue: …")` and return error). + +`UploadToSlot`: `landscape.SlotByCode` (else `ErrNotFound`); `groups.GetByCode(code)`; `categories.GetByCode(landscape.CategoryCode)`; `publisher.Upload(UploadInput{CategoryID, DatasetGroupID: &g.ID, Name (default filename), Filename, FileType: raster, ContentType, Size, Reader, StartDate, EndDate})`. + +`Slots` / `Results`: iterate `landscape.Slots` / `landscape.Outputs`, look up group by code, `LatestInGroup` (`Slots`: no status filter; `Results`: status ready) → `DatasetSummary` built from the `Dataset` (write a small `summaryOf(d domain.Dataset) domain.DatasetSummary` helper), `cog_url = "/datasets/" + id + ".cog"`, `sld_url = "/dataset-groups/" + gid + "/sld"` when the group has `SldStorageKey`. + +`MarkRunFailed` → `runs.SetRunStatus(failed, &reason)`. + +- [ ] **Step 3: RabbitMQ** (`internal/messaging/rabbitmq/landscape_consumer.go`) — mirror `parse_consumer.go`: constants, `LandscapeJob`, `LandscapeJobPublisher.EnqueueRun` (marshal `{run_id}` and `pub.Publish(ctx, LandscapeRunRoutingKey, body)`), `LandscapeProcessor`, `landscapeHandler.Handle` (decode, log, call `Execute`; on error call `MarkRunFailed` best-effort and return the error so the delivery is nacked), `NewLandscapeRunHandler`. + +- [ ] **Step 4: Run** `go test ./internal/service/ ./internal/messaging/... -race` → PASS. + +- [ ] **Step 5: Commit** + +```bash +git add internal/service/landscape.go internal/service/landscape_test.go internal/messaging/rabbitmq/landscape_consumer.go +git commit -m "feat(landscape): run planning/execution service with fingerprint reuse; RabbitMQ job" +``` + +--- + +### Task 8: HTTP API, wiring, OpenAPI, README, integration check + +**Files:** +- Create: `internal/transport/http/landscape_handler.go`, `internal/transport/http/landscape_handler_test.go` +- Modify: `internal/transport/http/router.go`, `internal/app/app.go`, `api/openapi.yaml`, `README.md`, `deployments/docker-compose.yml` (only if the worker needs an extra env; likely none) + +**Interfaces:** consumes Task 7's `LandscapeService` API and Task 3's `raster.NewGodalIO()`; Task 6's `landscape.Compute` is passed as `service.ComputeFunc`. + +- [ ] **Step 1: Handler** (`internal/transport/http/landscape_handler.go`) + +```go +// LandscapeHandler serves the /landscape routes. +type LandscapeHandler struct { + svc *service.LandscapeService + validate *validator.Validate +} + +func NewLandscapeHandler(svc *service.LandscapeService, validate *validator.Validate) *LandscapeHandler + +func (h *LandscapeHandler) Register(r chi.Router) { + r.Get("/slots", h.slots) + r.Post("/slots/{code}", h.uploadSlot) + r.Get("/runs", h.listRuns) + r.Post("/runs", h.createRun) + r.Get("/runs/{id}", h.getRun) + r.Get("/results", h.results) +} + +type runRequest struct { + FromStage *int `json:"from_stage" validate:"omitempty,gte=0,lte=3"` + Overrides map[string]string `json:"overrides" validate:"omitempty,dive,uuid"` + Params *landscape.Params `json:"params"` +} +``` + +`uploadSlot`: `r.ParseMultipartForm(maxUploadBytes)`, `FormFile("file")`, optional `name`, `start_date`, `end_date`; content type default `image/tiff`; 201 with dataset; unknown slot → 404 via `respondDomainError`. `createRun`: decode, validate, parse override UUIDs, `Params` nil → `landscape.Params{}`; 202 with run. `getRun`/`listRuns` (`page`, `page_size`, `newPaginated`)/`slots`/`results` → 200 JSON. + +Handler tests (`httptest` + a `LandscapeService` built on the stubs from `landscape_test.go` is in another package — instead test only request parsing paths that fail before the service is needed: invalid JSON → 400, `from_stage: 7` → 422, override not a UUID → 422; construct the handler with a nil service since those paths return early). + +- [ ] **Step 2: Wiring** + +`router.go`: `Landscape *LandscapeHandler` in `RouterDeps`; `r.Route("/landscape", deps.Landscape.Register)`. + +`app.go`: +```go + lpRepo := postgres.NewLandscapeRepository(pool) + lpJobs := rabbitmq.NewLandscapeJobPublisher(publisher) + rio := raster.NewGodalIO() + landscapeSvc := service.NewLandscapeService(lpRepo, datasetRepo, groupRepo, categoryRepo, datasetsSvc, store, lpJobs, rio, landscape.Compute, log) +``` +Add `landscape *service.LandscapeService` to `App`; `Landscape: transporthttp.NewLandscapeHandler(a.landscape, validate)`; add `LandscapeConsumer()` returning `rabbitmq.NewConsumer(a.rabbit, rabbitmq.LandscapeRunQueue, rabbitmq.LandscapeRunRoutingKey, "gis-landscape-runner", rabbitmq.NewLandscapeRunHandler(a.landscape, a.Log), a.Log)` and include it in `Consumers()`. Update `workerCmd.Short` to mention landscape runs. + +- [ ] **Step 3: OpenAPI + README** — tag `Landscape`; paths `/landscape/slots`, `/landscape/slots/{code}`, `/landscape/runs`, `/landscape/runs/{id}`, `/landscape/results`; schemas `LandscapeSlot`, `LandscapeRunInput`, `LandscapeParams`, `LandscapeRun`, `LandscapeRunNode`, `LandscapeDatasetRef`, `LandscapeResult`, `LandscapeRunPage`. README: new section "Landscape Program (ЛП)" — graph table (copy from spec §3.1), pixel rules summary, how to upload inputs, start a run, incremental recompute (`from_stage`), where results live, build requirements (cgo + GDAL: `brew install gdal` locally; Dockerfile handles the image). + +- [ ] **Step 4: Full verification** + +```bash +go build ./... && go vet ./... && go test ./... -race +``` +Expected: all PASS. If Docker is available: `docker compose -f deployments/docker-compose.yml build` and `docker compose -f deployments/docker-compose.yml up -d postgres` + `go run ./cmd/gis migrate up` against it to prove migrations 00010/00011 apply (then `migrate status`). Report exactly what was and was not verified. + +- [ ] **Step 5: Commit** + +```bash +git add internal/transport/http/landscape_handler.go internal/transport/http/landscape_handler_test.go internal/transport/http/router.go internal/app/app.go internal/cli/worker.go api/openapi.yaml README.md +git commit -m "feat(landscape): HTTP API, worker consumer, wiring, docs" +``` + +--- + +## Self-review notes + +- Spec §2.1–2.4 → Tasks 1–2. §3.1–3.2 → Task 4. §3.3 → Task 3. §3.4 → Task 6. §3.5–3.6 → Tasks 5, 7 (+ `CreateRasterFromFile` in Task 2). §3.7 → Task 8. §3.8 error handling is embedded in Tasks 6–7. §3.9 tests are spread across tasks. +- Name consistency: `LatestInGroup(ctx, groupID, status *string, fileType *domain.FileType)` (Tasks 1, 2, 7); `CreateRasterInput` / `CreateRasterFromFile` (Tasks 2, 7); `raster.Windows`, `DefaultWriterOptions` (Tasks 3, 6); `landscape.Stages`, `NodeByCode`, `SlotByCode`, `OutputByCode`, `Params.WithDefaults/Validate/ForNode` (Tasks 4, 6, 7); `domain.LPDatasetRef` (Tasks 5, 7).