infureal/gis
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add SLD endpoints, add date_to to vector_with_kato

Browse Source
This commit is contained in:
Bakhtiyar Issakhmetov
2026-08-01 22:09:59 +05:00
parent 00a0cd1b8e
commit 1e0844ca82
8 changed files with 656 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

156
internal/service/dataset.go
View File

@@ -29,6 +29,8 @@ var (
// ErrCOGNotReady is returned when a raster's Cloud-Optimized GeoTIFF has not
// been produced yet.
ErrCOGNotReady = errors.New("cog is not ready")
// ErrSLDNotFound is returned when a dataset has no SLD style file uploaded.
ErrSLDNotFound = fmt.Errorf("%w: dataset has no sld", domain.ErrNotFound)
)
// DatasetRepository is the persistence behaviour DatasetService needs.
@@ -47,6 +49,7 @@ type DatasetRepository interface {
SetGeoJSON(ctx context.Context, id uuid.UUID, geojson []byte) error
ClearGeometryAndGeoJSON(ctx context.Context, id uuid.UUID) error
SaveMapping(ctx context.Context, id uuid.UUID, katoColumn string, years []domain.YearColumn) (domain.Dataset, error)
SetSLD(ctx context.Context, id uuid.UUID, key, filename, contentType string) (domain.Dataset, error)
ReplaceObservations(ctx context.Context, datasetID uuid.UUID, obs []domain.Observation) error
ListObservations(ctx context.Context, datasetID uuid.UUID, katoCode *string, limit, offset int) ([]domain.Observation, error)
CountObservations(ctx context.Context, datasetID uuid.UUID, katoCode *string) (int, error)
@@ -235,6 +238,82 @@ func (s *DatasetService) Upload(ctx context.Context, in UploadInput) (domain.Dat
return dataset, nil
}
// UploadSLDInput carries an SLD (Styled Layer Descriptor) style file to attach
// to an existing dataset.
type UploadSLDInput struct {
Filename string
ContentType string
Size int64
Reader io.Reader
}
// UploadSLD validates and stores an SLD style file for a dataset, replacing
// any previously stored one. Applies to every file_type.
func (s *DatasetService) UploadSLD(ctx context.Context, id uuid.UUID, in UploadSLDInput) (domain.Dataset, error) {
dataset, err := s.repo.GetByID(ctx, id)
if err != nil {
return domain.Dataset{}, err
}
ext := strings.ToLower(filepath.Ext(in.Filename))
if !domain.ExtensionAllowedForSLD(ext) {
return domain.Dataset{}, 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.Dataset{}, fmt.Errorf("read sld upload: %w", err)
}
head = head[:n]
if err := domain.ValidateFileContent(ext, head); err != nil {
return domain.Dataset{}, fmt.Errorf("%w: %s", domain.ErrValidation, err)
}
content := io.MultiReader(bytes.NewReader(head), in.Reader)
key := deriveSLDKey(dataset.StorageKey, in.Filename)
if err := s.store.Put(ctx, key, content, in.Size, in.ContentType); err != nil {
return domain.Dataset{}, err
}
updated, err := s.repo.SetSLD(ctx, id, key, in.Filename, in.ContentType)
if err != nil {
// Compensate: the row was not updated, so the object would be orphaned.
_ = s.store.Remove(ctx, key)
return domain.Dataset{}, err
}
if dataset.SldStorageKey != nil && *dataset.SldStorageKey != key {
_ = s.store.Remove(ctx, *dataset.SldStorageKey) // best-effort: drop the replaced file
}
return updated, nil
}
// DownloadSLD returns the dataset metadata and a reader for its stored SLD
// style file. Fails with ErrSLDNotFound when none has been uploaded. The
// caller must close the reader.
func (s *DatasetService) DownloadSLD(ctx context.Context, id uuid.UUID) (domain.Dataset, io.ReadCloser, error) {
dataset, err := s.repo.GetByID(ctx, id)
if err != nil {
return domain.Dataset{}, nil, err
}
if dataset.SldStorageKey == nil {
return domain.Dataset{}, nil, ErrSLDNotFound
}
obj, err := s.store.Get(ctx, *dataset.SldStorageKey)
if err != nil {
return domain.Dataset{}, nil, err
}
return dataset, obj, nil
}
// deriveSLDKey places a dataset's SLD style file alongside the original under
// an sld/ prefix, named after the uploaded SLD file.
func deriveSLDKey(storageKey, filename string) string {
return path.Join(path.Dir(storageKey), "sld", filename)
}
// enqueueProcessing schedules the appropriate async job for a dataset based on
// its file type: vector_with_kato is parsed for column selection, plain vector
// has its attribute table extracted into properties, and raster is converted to
@@ -603,9 +682,19 @@ func (s *DatasetService) SaveMapping(ctx context.Context, id uuid.UUID, in Mappi
if _, ok := known[yc.Column]; !ok {
return domain.Dataset{}, fmt.Errorf("%w: year column %q is not among the detected columns", domain.ErrValidation, yc.Column)
}
if _, err := time.Parse("2006-01-02", yc.Date); err != nil {
from, err := time.Parse("2006-01-02", yc.Date)
if err != nil {
return domain.Dataset{}, fmt.Errorf("%w: invalid date %q for column %q (want YYYY-MM-DD)", domain.ErrValidation, yc.Date, yc.Column)
}
if yc.DateTo != nil {
to, err := time.Parse("2006-01-02", *yc.DateTo)
if err != nil {
return domain.Dataset{}, fmt.Errorf("%w: invalid date_to %q for column %q (want YYYY-MM-DD)", domain.ErrValidation, *yc.DateTo, yc.Column)
}
if to.Before(from) {
return domain.Dataset{}, fmt.Errorf("%w: date_to %q is before date %q for column %q", domain.ErrValidation, *yc.DateTo, yc.Date, yc.Column)
}
}
}
dataset, err = s.repo.SaveMapping(ctx, id, in.KatoColumn, in.YearColumns)
@@ -659,8 +748,9 @@ func (s *DatasetService) Extract(ctx context.Context, id uuid.UUID) error {
}
// buildObservations unpivots rows into observations. Rows without a KATO code
// are skipped; duplicate (kato, date) pairs keep the last value. Numeric cells
// populate Value, others ValueText.
// are skipped; duplicate (kato, date, date_to) tuples keep the last value.
// Numeric cells populate Value, others ValueText. A year column's DateTo (nil
// for a point-in-time mapping) carries through to the observation unchanged.
func buildObservations(datasetID uuid.UUID, katoColumn string, years []domain.YearColumn, rows []map[string]string) []domain.Observation {
obs := make([]domain.Observation, 0, len(rows)*len(years))
index := make(map[string]int)
@@ -671,7 +761,7 @@ func buildObservations(datasetID uuid.UUID, katoColumn string, years []domain.Ye
continue
}
for _, yc := range years {
o := domain.Observation{DatasetID: datasetID, KatoCode: kato, Date: yc.Date}
o := domain.Observation{DatasetID: datasetID, KatoCode: kato, Date: yc.Date, DateTo: yc.DateTo}
if raw := strings.TrimSpace(row[yc.Column]); raw != "" {
if f, err := strconv.ParseFloat(raw, 64); err == nil {
o.Value = &f
@@ -679,7 +769,7 @@ func buildObservations(datasetID uuid.UUID, katoColumn string, years []domain.Ye
o.ValueText = &raw
}
}
key := kato + "\x00" + yc.Date
key := kato + "\x00" + yc.Date + "\x00" + dateToKey(yc.DateTo)
if i, ok := index[key]; ok {
obs[i] = o
} else {
@@ -691,6 +781,14 @@ func buildObservations(datasetID uuid.UUID, katoColumn string, years []domain.Ye
return obs
}
// dateToKey normalizes an optional period end date for use in a dedup key.
func dateToKey(dateTo *string) string {
if dateTo == nil {
return ""
}
return *dateTo
}
// ObservationPage is a page of observations with pagination metadata.
type ObservationPage struct {
Items []domain.Observation
@@ -789,7 +887,11 @@ func (s *DatasetService) generateGeoJSON(ctx context.Context, id uuid.UUID) erro
// joins the districts table on KATO code: one Feature is emitted per KATO, its
// boundary taken from the districts table and its observation values nested
// under a `data` object (keyed by date) alongside `kato` and `name`. KATO codes
// with no matching district are skipped.
// with no matching district are skipped. When any observation carries a period
// (DateTo set), its end date is additionally exposed under a `periods` object
// (keyed by the same start date used in `data`); `periods` is omitted entirely
// when every observation is point-in-time, so a dataset mapped without periods
// produces byte-identical output to before periods existed.
func (s *DatasetService) buildGeoJSON(ctx context.Context, id uuid.UUID) (json.RawMessage, error) {
dataset, err := s.repo.GetByID(ctx, id)
if err != nil {
@@ -816,8 +918,8 @@ func (s *DatasetService) buildGeoJSON(ctx context.Context, id uuid.UUID) (json.R
if err != nil {
return nil, err
}
grouped, order := groupObservationsByKato(obs)
features, err := s.districtFeatures(ctx, grouped, order)
grouped, periods, order := groupObservationsByKato(obs)
features, err := s.districtFeatures(ctx, grouped, periods, order)
if err != nil {
return nil, err
}
@@ -850,9 +952,12 @@ func (s *DatasetService) loadGeoJSONDataset(ctx context.Context, id uuid.UUID) (
// districtFeatures builds one Feature per KATO from the districts table, with
// the grouped observation values (keyed by date) nested under a `data` object
// alongside `kato` and `name` in each Feature's properties. KATO codes with no
// alongside `kato` and `name` in each Feature's properties. When periods holds
// an entry for the KATO (at least one observation had a period), its date->
// date_to map is additionally nested under a `periods` object; a KATO with no
// period observations gets no `periods` key at all. KATO codes with no
// matching district are skipped. order drives the deterministic feature order.
func (s *DatasetService) districtFeatures(ctx context.Context, grouped map[string]map[string]any, order []string) ([]domain.Feature, error) {
func (s *DatasetService) districtFeatures(ctx context.Context, grouped map[string]map[string]any, periods map[string]map[string]string, order []string) ([]domain.Feature, error) {
districts, err := s.repo.DistrictGeometriesByKato(ctx, order)
if err != nil {
return nil, err
@@ -863,10 +968,14 @@ func (s *DatasetService) districtFeatures(ctx context.Context, grouped map[strin
if !ok {
continue // skip KATO codes with no district boundary
}
props := map[string]any{"kato": kato, "name": dist.Name, "data": grouped[kato]}
if p, ok := periods[kato]; ok {
props["periods"] = p
}
features = append(features, domain.Feature{
Type: domain.GeoJSONFeature,
Geometry: dist.Geometry,
Properties: map[string]any{"kato": kato, "name": dist.Name, "data": grouped[kato]},
Properties: props,
})
}
return features, nil
@@ -909,11 +1018,15 @@ func vectorFeatureProperties(raw json.RawMessage) map[string]any {
// groupObservationsByKato groups observations by KATO code into date->value
// maps. The value is the numeric Value when present, otherwise ValueText,
// otherwise nil (an empty cell). order lists KATO codes in first-seen order so
// the resulting feature order is deterministic.
func groupObservationsByKato(obs []domain.Observation) (map[string]map[string]any, []string) {
grouped := make(map[string]map[string]any)
order := make([]string, 0)
// otherwise nil (an empty cell). It also groups, per KATO, a date->date_to map
// for observations that carry a period (DateTo set); a KATO with no period
// observations gets no entry in periods at all, so callers can tell "no
// periods" apart from "empty periods". order lists KATO codes in first-seen
// order so the resulting feature order is deterministic.
func groupObservationsByKato(obs []domain.Observation) (grouped map[string]map[string]any, periods map[string]map[string]string, order []string) {
grouped = make(map[string]map[string]any)
periods = make(map[string]map[string]string)
order = make([]string, 0)
for _, o := range obs {
values, ok := grouped[o.KatoCode]
if !ok {
@@ -922,8 +1035,17 @@ func groupObservationsByKato(obs []domain.Observation) (map[string]map[string]an
order = append(order, o.KatoCode)
}
values[o.Date] = observationValue(o)
if o.DateTo != nil {
p, ok := periods[o.KatoCode]
if !ok {
p = make(map[string]string)
periods[o.KatoCode] = p
}
p[o.Date] = *o.DateTo
}
}
return grouped, order
return grouped, periods, order
}
// observationValue returns the typed cell value: numeric Value, else ValueText,

265
internal/service/dataset_test.go
View File

@@ -137,6 +137,18 @@ func (r *stubDatasetRepo) SaveMapping(_ context.Context, id uuid.UUID, kato stri
return d, nil
}
func (r *stubDatasetRepo) SetSLD(_ context.Context, id uuid.UUID, key, filename, contentType string) (domain.Dataset, error) {
d, ok := r.store[id]
if !ok {
return domain.Dataset{}, domain.ErrNotFound
}
d.SldStorageKey = &key
d.SldFilename = &filename
d.SldContentType = &contentType
r.store[id] = d
return d, nil
}
func (r *stubDatasetRepo) MarkReady(_ context.Context, id uuid.UUID, geometry []byte) error {
d, ok := r.store[id]
if !ok {
@@ -776,6 +788,39 @@ func TestDatasetService_SaveMapping(t *testing.T) {
}
})
t.Run("bad date_to", func(t *testing.T) {
_, svc := newSvc()
_, err := svc.SaveMapping(ctx, id, MappingInput{KatoColumn: "като", YearColumns: []domain.YearColumn{{Column: "F_2023", Date: "2023-01-01", DateTo: strPtr("2023")}}})
if !errors.Is(err, domain.ErrValidation) {
t.Fatalf("want ErrValidation, got %v", err)
}
})
t.Run("date_to before date", func(t *testing.T) {
_, svc := newSvc()
_, err := svc.SaveMapping(ctx, id, MappingInput{KatoColumn: "като", YearColumns: []domain.YearColumn{{Column: "F_2023", Date: "2023-12-31", DateTo: strPtr("2023-01-01")}}})
if !errors.Is(err, domain.ErrValidation) {
t.Fatalf("want ErrValidation, got %v", err)
}
})
t.Run("period mapping succeeds", func(t *testing.T) {
repo, svc := newSvc()
got, err := svc.SaveMapping(ctx, id, MappingInput{
KatoColumn: "като",
YearColumns: []domain.YearColumn{{Column: "F_2023", Date: "2020-01-01", DateTo: strPtr("2023-12-31")}},
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if len(got.YearColumns) != 1 || got.YearColumns[0].DateTo == nil || *got.YearColumns[0].DateTo != "2023-12-31" {
t.Fatalf("period not saved: %+v", got.YearColumns)
}
if len(repo.store[id].YearColumns) != 1 {
t.Fatalf("period not persisted: %+v", repo.store[id].YearColumns)
}
})
t.Run("no year columns", func(t *testing.T) {
_, svc := newSvc()
_, err := svc.SaveMapping(ctx, id, MappingInput{KatoColumn: "като"})
@@ -845,6 +890,45 @@ func TestBuildObservations(t *testing.T) {
if o := byKey["751010000|2025-01-01"]; o.ValueText == nil || *o.ValueText != "n/a" {
t.Errorf("non-numeric cell not stored as value_text: %+v", o)
}
for _, o := range obs {
if o.DateTo != nil {
t.Fatalf("point-in-time mapping must not set DateTo, got %+v", o)
}
}
}
func TestBuildObservations_Period(t *testing.T) {
id := uuid.New()
years := []domain.YearColumn{
{Column: "F_2020_2023", Date: "2020-01-01", DateTo: strPtr("2023-12-31")}, // period
{Column: "F_2024", Date: "2024-01-01"}, // point-in-time, unaffected
}
rows := []map[string]string{
{"като": "751010000", "F_2020_2023": "100", "F_2024": "200"},
}
obs := buildObservations(id, "като", years, rows)
if len(obs) != 2 {
t.Fatalf("want 2 observations, got %d", len(obs))
}
byDate := map[string]domain.Observation{}
for _, o := range obs {
byDate[o.Date] = o
}
period := byDate["2020-01-01"]
if period.DateTo == nil || *period.DateTo != "2023-12-31" {
t.Fatalf("period observation missing date_to: %+v", period)
}
if period.Value == nil || *period.Value != 100 {
t.Fatalf("period observation value wrong: %+v", period)
}
point := byDate["2024-01-01"]
if point.DateTo != nil {
t.Fatalf("point-in-time observation must have nil date_to (backward compat), got %+v", point)
}
}
func TestDatasetService_Extract(t *testing.T) {
@@ -959,6 +1043,68 @@ func TestDatasetService_GeoJSON_DistrictJoin(t *testing.T) {
}
}
func TestDatasetService_GeoJSON_Period(t *testing.T) {
id := uuid.New()
repo := newStubDatasetRepo()
repo.store[id] = domain.Dataset{ID: id, FileType: domain.FileTypeVectorWithKato, Status: domain.DatasetStatusReady}
period, point := 100.0, 200.0
repo.observations[id] = []domain.Observation{
{KatoCode: "710000000", Date: "2020-01-01", DateTo: strPtr("2023-12-31"), Value: &period},
{KatoCode: "710000000", Date: "2024-01-01", Value: &point}, // point-in-time, no period
}
repo.districts["710000000"] = domain.District{
Kato: "710000000", Name: "Astana",
Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[71,51],[72,51],[72,52],[71,51]]]}`),
}
svc := newDatasetService(repo, &stubStore{}, true)
fc := mustGeoJSON(t, svc, id)
if len(fc.Features) != 1 {
t.Fatalf("want 1 feature, got %d", len(fc.Features))
}
f := fc.Features[0]
data, ok := f.Properties["data"].(map[string]any)
if !ok {
t.Fatalf("observations not nested under data: %+v", f.Properties)
}
if data["2020-01-01"] != 100.0 || data["2024-01-01"] != 200.0 {
t.Fatalf("unexpected data: %+v", data)
}
periods, ok := f.Properties["periods"].(map[string]any)
if !ok {
t.Fatalf("periods not exposed as a map: %+v", f.Properties)
}
if len(periods) != 1 || periods["2020-01-01"] != "2023-12-31" {
t.Fatalf("period end date not exposed: %+v", periods)
}
if _, has := periods["2024-01-01"]; has {
t.Fatalf("point-in-time date must not appear in periods: %+v", periods)
}
}
func TestDatasetService_GeoJSON_NoPeriods_OmitsPeriodsKey(t *testing.T) {
// Regression: a dataset mapped without any period must produce output
// byte-identical to before periods existed - no "periods" key at all.
id := uuid.New()
repo := newStubDatasetRepo()
repo.store[id] = domain.Dataset{ID: id, FileType: domain.FileTypeVectorWithKato, Status: domain.DatasetStatusReady}
v := 100.0
repo.observations[id] = []domain.Observation{{KatoCode: "710000000", Date: "2020-01-01", Value: &v}}
repo.districts["710000000"] = domain.District{
Kato: "710000000", Name: "Astana",
Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[71,51],[72,51],[72,52],[71,51]]]}`),
}
svc := newDatasetService(repo, &stubStore{}, true)
fc := mustGeoJSON(t, svc, id)
if _, has := fc.Features[0].Properties["periods"]; has {
t.Fatalf("periods key must be absent when no observation has a period: %+v", fc.Features[0].Properties)
}
}
func TestDatasetService_GeoJSON_IgnoresDatasetGeometry(t *testing.T) {
id := uuid.New()
repo := newStubDatasetRepo()
@@ -1294,6 +1440,125 @@ func TestDatasetService_ReprocessAll_FiltersByFileType(t *testing.T) {
}
}
func validSLDUpload() UploadSLDInput {
body := `<?xml version="1.0"?><StyledLayerDescriptor></StyledLayerDescriptor>`
return UploadSLDInput{
Filename: "style.sld",
ContentType: "application/xml",
Size: int64(len(body)),
Reader: strings.NewReader(body),
}
}
func TestDatasetService_UploadSLD_Success(t *testing.T) {
id := uuid.New()
repo := newStubDatasetRepo()
repo.store[id] = domain.Dataset{ID: id, FileType: domain.FileTypeRaster, StorageKey: "uid/dem.tif"}
store := &stubStore{}
svc := newDatasetService(repo, store, true)
got, err := svc.UploadSLD(context.Background(), id, validSLDUpload())
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if got.SldStorageKey == nil || *got.SldStorageKey != "uid/sld/style.sld" {
t.Fatalf("unexpected sld storage key: %v", got.SldStorageKey)
}
if got.SldFilename == nil || *got.SldFilename != "style.sld" {
t.Fatalf("sld filename not stored: %v", got.SldFilename)
}
if len(store.put) != 1 || store.put[0] != "uid/sld/style.sld" {
t.Fatalf("sld not uploaded to store: %v", store.put)
}
}
func TestDatasetService_UploadSLD_ReplacesPrevious(t *testing.T) {
id := uuid.New()
repo := newStubDatasetRepo()
oldKey := "uid/sld/old.sld"
repo.store[id] = domain.Dataset{
ID: id, FileType: domain.FileTypeVector, StorageKey: "uid/data.geojson",
SldStorageKey: &oldKey,
}
store := &stubStore{}
svc := newDatasetService(repo, store, true)
in := validSLDUpload()
in.Filename = "new.sld"
got, err := svc.UploadSLD(context.Background(), id, in)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if got.SldStorageKey == nil || *got.SldStorageKey != "uid/sld/new.sld" {
t.Fatalf("unexpected sld storage key: %v", got.SldStorageKey)
}
if len(store.removed) != 1 || store.removed[0] != oldKey {
t.Fatalf("old sld object not removed: %v", store.removed)
}
}
func TestDatasetService_UploadSLD_RejectsBadExtension(t *testing.T) {
id := uuid.New()
repo := newStubDatasetRepo()
repo.store[id] = domain.Dataset{ID: id, StorageKey: "uid/data.geojson"}
store := &stubStore{}
svc := newDatasetService(repo, store, true)
in := validSLDUpload()
in.Filename = "style.zip"
_, err := svc.UploadSLD(context.Background(), id, in)
if !errors.Is(err, domain.ErrValidation) {
t.Fatalf("want ErrValidation, got %v", err)
}
if len(store.put) != 0 {
t.Fatalf("nothing should be uploaded on validation failure, got %v", store.put)
}
}
func TestDatasetService_UploadSLD_RejectsMismatchedContent(t *testing.T) {
id := uuid.New()
repo := newStubDatasetRepo()
repo.store[id] = domain.Dataset{ID: id, StorageKey: "uid/data.geojson"}
store := &stubStore{}
svc := newDatasetService(repo, store, true)
in := validSLDUpload()
in.Reader = strings.NewReader(`{"not":"xml"}`)
_, err := svc.UploadSLD(context.Background(), id, in)
if !errors.Is(err, domain.ErrValidation) {
t.Fatalf("want ErrValidation, got %v", err)
}
}
func TestDatasetService_DownloadSLD(t *testing.T) {
id := uuid.New()
repo := newStubDatasetRepo()
repo.store[id] = domain.Dataset{ID: id, StorageKey: "uid/data.geojson"}
store := &stubStore{}
svc := newDatasetService(repo, store, true)
t.Run("not found before upload", func(t *testing.T) {
_, _, err := svc.DownloadSLD(context.Background(), id)
if !errors.Is(err, ErrSLDNotFound) {
t.Fatalf("want ErrSLDNotFound, got %v", err)
}
})
t.Run("streams after upload", func(t *testing.T) {
if _, err := svc.UploadSLD(context.Background(), id, validSLDUpload()); err != nil {
t.Fatalf("upload: %v", err)
}
_, obj, err := svc.DownloadSLD(context.Background(), id)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
defer obj.Close()
if _, err := io.ReadAll(obj); err != nil {
t.Fatalf("read: %v", err)
}
})
}
func TestDatasetService_Delete_RemovesObject(t *testing.T) {
repo := newStubDatasetRepo()
id := uuid.New()