From 5ca6d9e662f6120c28229ea80f215490757645f6 Mon Sep 17 00:00:00 2001 From: Bakhtiyar Issakhmetov Date: Sat, 5 Sep 2026 13:13:32 +0500 Subject: [PATCH] feat(landscape): run/node domain, seed migration, repository Co-Authored-By: Claude Fable 5.1 --- internal/domain/landscape.go | 68 +++++++ internal/repository/postgres/landscape.go | 236 ++++++++++++++++++++++ migrations/00011_landscape.sql | 51 +++++ 3 files changed, 355 insertions(+) create mode 100644 internal/domain/landscape.go create mode 100644 internal/repository/postgres/landscape.go create mode 100644 migrations/00011_landscape.sql diff --git a/internal/domain/landscape.go b/internal/domain/landscape.go new file mode 100644 index 0000000..050950e --- /dev/null +++ b/internal/domain/landscape.go @@ -0,0 +1,68 @@ +package domain + +import ( + "encoding/json" + "time" + + "github.com/google/uuid" +) + +// LPRunStatus is the lifecycle status of a Landscape Program run. +type LPRunStatus string + +// Landscape Program run statuses. +const ( + LPRunQueued LPRunStatus = "queued" + LPRunRunning LPRunStatus = "running" + LPRunSucceeded LPRunStatus = "succeeded" + LPRunFailed LPRunStatus = "failed" +) + +// LPNodeStatus is the lifecycle status of a single node within a Landscape +// Program run. +type LPNodeStatus string + +// Landscape Program run node statuses. +const ( + LPNodePending LPNodeStatus = "pending" + LPNodeRunning LPNodeStatus = "running" + LPNodeSkipped LPNodeStatus = "skipped" + LPNodeSucceeded LPNodeStatus = "succeeded" + LPNodeFailed LPNodeStatus = "failed" +) + +// LPDatasetRef references a dataset produced for, or consumed from, a +// dataset group slot in the Landscape Program graph. +type LPDatasetRef struct { + Group string `json:"group"` + DatasetID uuid.UUID `json:"dataset_id"` +} + +// LPRun is a single execution of the Landscape Program graph. +type LPRun struct { + ID uuid.UUID `json:"id"` + Status LPRunStatus `json:"status"` + FromStage *int `json:"from_stage"` + Params json.RawMessage `json:"params"` + Overrides map[string]uuid.UUID `json:"overrides"` + Error *string `json:"error"` + CreatedAt time.Time `json:"created_at"` + StartedAt *time.Time `json:"started_at"` + FinishedAt *time.Time `json:"finished_at"` + Nodes []LPRunNode `json:"nodes"` +} + +// LPRunNode is the execution state of a single graph node within an LPRun. +type LPRunNode struct { + ID uuid.UUID `json:"id"` + RunID uuid.UUID `json:"run_id"` + NodeCode string `json:"node_code"` + Stage int `json:"stage"` + Status LPNodeStatus `json:"status"` + Fingerprint string `json:"fingerprint"` + Inputs []LPDatasetRef `json:"inputs"` + Outputs []LPDatasetRef `json:"outputs"` + Error *string `json:"error"` + StartedAt *time.Time `json:"started_at"` + FinishedAt *time.Time `json:"finished_at"` +} diff --git a/internal/repository/postgres/landscape.go b/internal/repository/postgres/landscape.go new file mode 100644 index 0000000..34d92df --- /dev/null +++ b/internal/repository/postgres/landscape.go @@ -0,0 +1,236 @@ +package postgres + +import ( + "context" + "encoding/json" + + "gis/internal/domain" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// LandscapeRepository persists Landscape Program runs in Postgres. +type LandscapeRepository struct { + pool *pgxpool.Pool +} + +// NewLandscapeRepository returns a LandscapeRepository backed by the given pool. +func NewLandscapeRepository(pool *pgxpool.Pool) *LandscapeRepository { + return &LandscapeRepository{pool: pool} +} + +const runColumns = `id, status, from_stage, params, overrides, error, created_at, started_at, finished_at` + +// scanRun scans a run row. overrides is stored as jsonb but scanned into +// []byte and unmarshalled manually so an empty/NULL value maps to an empty +// map rather than nil. +func scanRun(row pgx.Row) (domain.LPRun, error) { + var r domain.LPRun + var overrides []byte + err := row.Scan(&r.ID, &r.Status, &r.FromStage, &r.Params, &overrides, &r.Error, &r.CreatedAt, &r.StartedAt, &r.FinishedAt) + if err != nil { + return r, err + } + r.Overrides = map[string]uuid.UUID{} + if len(overrides) > 0 { + if err := json.Unmarshal(overrides, &r.Overrides); err != nil { + return r, err + } + } + return r, nil +} + +const nodeColumns = `id, run_id, node_code, stage, status, fingerprint, input_datasets, output_datasets, error, started_at, finished_at` + +func scanNode(row pgx.Row) (domain.LPRunNode, error) { + var n domain.LPRunNode + err := row.Scan(&n.ID, &n.RunID, &n.NodeCode, &n.Stage, &n.Status, &n.Fingerprint, &n.Inputs, &n.Outputs, &n.Error, &n.StartedAt, &n.FinishedAt) + return n, err +} + +// CreateRun inserts a new run and its nodes (status pending) in one +// transaction, and returns the run with its nodes populated (in the order +// they were given). +func (r *LandscapeRepository) CreateRun(ctx context.Context, run domain.LPRun) (domain.LPRun, error) { + tx, err := r.pool.Begin(ctx) + if err != nil { + return domain.LPRun{}, mapError(err) + } + defer tx.Rollback(ctx) + + overrides := run.Overrides + if overrides == nil { + overrides = map[string]uuid.UUID{} + } + overridesJSON, err := json.Marshal(overrides) + if err != nil { + return domain.LPRun{}, err + } + + row := tx.QueryRow(ctx, + `INSERT INTO lp_runs (status, from_stage, params, overrides) + VALUES ($1, $2, $3, $4) + RETURNING `+runColumns, + run.Status, run.FromStage, run.Params, overridesJSON, + ) + out, err := scanRun(row) + if err != nil { + return domain.LPRun{}, mapError(err) + } + + nodes := make([]domain.LPRunNode, 0, len(run.Nodes)) + for _, n := range run.Nodes { + row := tx.QueryRow(ctx, + `INSERT INTO lp_run_nodes (run_id, node_code, stage, status) + VALUES ($1, $2, $3, $4) + RETURNING `+nodeColumns, + out.ID, n.NodeCode, n.Stage, domain.LPNodePending, + ) + node, err := scanNode(row) + if err != nil { + return domain.LPRun{}, mapError(err) + } + nodes = append(nodes, node) + } + out.Nodes = nodes + + if err := tx.Commit(ctx); err != nil { + return domain.LPRun{}, mapError(err) + } + return out, nil +} + +// GetRun returns the run with the given id, with its nodes ordered by +// node_code. Returns domain.ErrNotFound if the run does not exist. +func (r *LandscapeRepository) GetRun(ctx context.Context, id uuid.UUID) (domain.LPRun, error) { + row := r.pool.QueryRow(ctx, `SELECT `+runColumns+` FROM lp_runs WHERE id = $1`, id) + out, err := scanRun(row) + if err != nil { + return domain.LPRun{}, mapError(err) + } + + rows, err := r.pool.Query(ctx, + `SELECT `+nodeColumns+` FROM lp_run_nodes WHERE run_id = $1 ORDER BY node_code`, id) + if err != nil { + return domain.LPRun{}, mapError(err) + } + defer rows.Close() + + nodes := make([]domain.LPRunNode, 0) + for rows.Next() { + n, err := scanNode(rows) + if err != nil { + return domain.LPRun{}, mapError(err) + } + nodes = append(nodes, n) + } + if err := rows.Err(); err != nil { + return domain.LPRun{}, mapError(err) + } + out.Nodes = nodes + return out, nil +} + +// ListRuns returns a page of runs (without nodes), newest first. +func (r *LandscapeRepository) ListRuns(ctx context.Context, limit, offset int) ([]domain.LPRun, error) { + rows, err := r.pool.Query(ctx, + `SELECT `+runColumns+` FROM lp_runs ORDER BY created_at DESC LIMIT $1 OFFSET $2`, limit, offset) + if err != nil { + return nil, mapError(err) + } + defer rows.Close() + + runs := make([]domain.LPRun, 0) + for rows.Next() { + run, err := scanRun(rows) + if err != nil { + return nil, mapError(err) + } + runs = append(runs, run) + } + return runs, mapError(rows.Err()) +} + +// CountRuns returns the total number of runs. +func (r *LandscapeRepository) CountRuns(ctx context.Context) (int, error) { + var n int + err := r.pool.QueryRow(ctx, `SELECT count(*) FROM lp_runs`).Scan(&n) + return n, mapError(err) +} + +// SetRunStatus updates a run's status and error. Transitioning to running +// sets started_at; transitioning to succeeded or failed sets finished_at. +// Returns domain.ErrNotFound if the run does not exist. +func (r *LandscapeRepository) SetRunStatus(ctx context.Context, id uuid.UUID, status domain.LPRunStatus, errMsg *string) error { + tag, err := r.pool.Exec(ctx, + `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`, + id, status, errMsg, + ) + if err != nil { + return mapError(err) + } + if tag.RowsAffected() == 0 { + return domain.ErrNotFound + } + return nil +} + +// SetNodeStarted marks a run's node as running, storing its fingerprint and +// resolved inputs, and clears any previous error. Returns domain.ErrNotFound +// if no such node exists. +func (r *LandscapeRepository) SetNodeStarted(ctx context.Context, runID uuid.UUID, code string, fingerprint string, inputs []domain.LPDatasetRef) error { + tag, err := r.pool.Exec(ctx, + `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`, + runID, code, fingerprint, inputs, + ) + if err != nil { + return mapError(err) + } + if tag.RowsAffected() == 0 { + return domain.ErrNotFound + } + return nil +} + +// SetNodeFinished marks a run's node as finished with the given status, +// storing its outputs and error. Returns domain.ErrNotFound if no such node +// exists. +func (r *LandscapeRepository) SetNodeFinished(ctx context.Context, runID uuid.UUID, code string, status domain.LPNodeStatus, outputs []domain.LPDatasetRef, errMsg *string) error { + tag, err := r.pool.Exec(ctx, + `UPDATE lp_run_nodes + SET status = $3, output_datasets = $4, error = $5, finished_at = now() + WHERE run_id = $1 AND node_code = $2`, + runID, code, status, outputs, errMsg, + ) + if err != nil { + return mapError(err) + } + if tag.RowsAffected() == 0 { + return domain.ErrNotFound + } + return nil +} + +// FindReusableOutputs returns the outputs of the most recently succeeded node +// with the same code and fingerprint, excluding excludeRun, so its outputs +// may be reused instead of recomputed. Returns domain.ErrNotFound if none. +func (r *LandscapeRepository) FindReusableOutputs(ctx context.Context, code, fingerprint string, excludeRun uuid.UUID) ([]domain.LPDatasetRef, error) { + var outputs []domain.LPDatasetRef + err := r.pool.QueryRow(ctx, + `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`, + code, fingerprint, excludeRun, + ).Scan(&outputs) + if err != nil { + return nil, mapError(err) + } + return outputs, nil +} diff --git a/migrations/00011_landscape.sql b/migrations/00011_landscape.sql new file mode 100644 index 0000000..70c9261 --- /dev/null +++ b/migrations/00011_landscape.sql @@ -0,0 +1,51 @@ +-- +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).