diff --git a/internal/app/app.go b/internal/app/app.go index c335f4a..eb12004 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -178,11 +178,12 @@ func (a *App) ReprocessDataset(ctx context.Context, id uuid.UUID) (domain.Datase return a.datasets.Reprocess(ctx, id) } -// ReprocessAllDatasets re-enqueues the processing job for every dataset, -// restarting each one's asynchronous pipeline. It returns how many jobs were -// enqueued and a per-dataset map of any failures. -func (a *App) ReprocessAllDatasets(ctx context.Context) (int, map[uuid.UUID]error, error) { - return a.datasets.ReprocessAll(ctx) +// ReprocessAllDatasets re-enqueues the processing job for every dataset +// matching fileType (nil matches every file type), restarting each one's +// asynchronous pipeline. It returns how many jobs were enqueued and a +// per-dataset map of any failures. +func (a *App) ReprocessAllDatasets(ctx context.Context, fileType *domain.FileType) (int, map[uuid.UUID]error, error) { + return a.datasets.ReprocessAll(ctx, fileType) } // Close releases all resources in reverse order of acquisition. diff --git a/internal/cli/reprocess.go b/internal/cli/reprocess.go index 8213e30..9a167a9 100644 --- a/internal/cli/reprocess.go +++ b/internal/cli/reprocess.go @@ -2,31 +2,53 @@ package cli import ( "fmt" + "strings" "gis/internal/app" + "gis/internal/domain" "github.com/google/uuid" "github.com/spf13/cobra" ) -var reprocessAll bool +var ( + reprocessAll bool + reprocessType string +) var reprocessCmd = &cobra.Command{ Use: "reprocess [dataset-id]", Short: "Re-enqueue the processing job for uploaded datasets", Long: "Re-publish the RabbitMQ message that drives an uploaded dataset's\n" + "asynchronous processing, selecting the right step from its file type\n" + - "(vector_with_kato -> parse, vector -> properties, raster -> cog).\n\n" + + "(vector_with_kato -> parse, vector -> properties, raster -> cog).\n" + + "Before re-enqueuing, each dataset's stored geometry and GeoJSON are\n" + + "cleared so a stale (e.g. previously miscomputed) value is never served\n" + + "while the job is pending.\n\n" + "Pass a dataset id to reprocess a single dataset, or --all to reprocess\n" + - "every dataset.\n\n" + + "every dataset. --type restricts --all to one file type\n" + + "(vector_with_kato, vector, or raster).\n\n" + "Examples:\n" + " gis reprocess 06818b2b-1fc5-47d9-a764-db2d4cb3df75\n" + - " gis reprocess --all", + " gis reprocess --all\n" + + " gis reprocess --all --type=vector", Args: cobra.MaximumNArgs(1), RunE: func(cmd *cobra.Command, args []string) error { if reprocessAll == (len(args) == 1) { return fmt.Errorf("provide exactly one of or --all") } + if reprocessType != "" && !reprocessAll { + return fmt.Errorf("--type can only be used together with --all") + } + + var fileType *domain.FileType + if reprocessType != "" { + ft := domain.FileType(strings.TrimSpace(reprocessType)) + if !ft.Valid() { + return fmt.Errorf("invalid --type %q: must be one of vector_with_kato, vector, raster", reprocessType) + } + fileType = &ft + } ctx, cancel := signalContext() defer cancel() @@ -38,7 +60,7 @@ var reprocessCmd = &cobra.Command{ defer application.Close() if reprocessAll { - enqueued, failures, err := application.ReprocessAllDatasets(ctx) + enqueued, failures, err := application.ReprocessAllDatasets(ctx, fileType) if err != nil { return err } @@ -47,7 +69,7 @@ var reprocessCmd = &cobra.Command{ "dataset_id", id, "error", ferr) } application.Log.Info("re-enqueued all dataset processing", - "enqueued", enqueued, "failed", len(failures)) + "file_type", reprocessType, "enqueued", enqueued, "failed", len(failures)) if len(failures) > 0 { return fmt.Errorf("%d of %d datasets failed to re-enqueue", len(failures), enqueued+len(failures)) } @@ -71,4 +93,5 @@ var reprocessCmd = &cobra.Command{ func init() { reprocessCmd.Flags().BoolVar(&reprocessAll, "all", false, "reprocess every dataset") + reprocessCmd.Flags().StringVar(&reprocessType, "type", "", "restrict --all to one file type (vector_with_kato, vector, raster)") } diff --git a/internal/raster/gdal.go b/internal/raster/gdal.go index 1e2464d..4806958 100644 --- a/internal/raster/gdal.go +++ b/internal/raster/gdal.go @@ -58,12 +58,15 @@ func (c *GDALConverter) Footprint(ctx context.Context, src string) ([]byte, erro return info.Wgs84Extent, nil } -// VectorGeometry reads every feature of a vector file and returns their combined -// geometry as a GeoJSON GeometryCollection reprojected to EPSG:4326, or nil if -// the file has no features. The caller (PostGIS) dissolves the collection into -// the union of all features. Zipped ESRI shapefiles are read in place via GDAL's -// /vsizip/ virtual filesystem; GeoJSON and GeoPackage are read directly. -func (c *GDALConverter) VectorGeometry(ctx context.Context, src string) ([]byte, error) { +// VectorFeatures reads every feature of a vector file and returns their +// geometries, reprojected to EPSG:4326, in file order (nil if the file has no +// features). A feature with no geometry yields a nil entry at its index, so +// the result stays aligned with attribute rows read from the same file (the +// shapefile format itself has no other way to pair a .shp geometry record with +// its .dbf attribute row than matching by position). Zipped ESRI shapefiles +// are read in place via GDAL's /vsizip/ virtual filesystem; GeoJSON and +// GeoPackage are read directly. +func (c *GDALConverter) VectorFeatures(ctx context.Context, src string) ([]json.RawMessage, error) { input := src if strings.EqualFold(filepath.Ext(src), ".zip") { input = "/vsizip/" + src @@ -89,20 +92,16 @@ func (c *GDALConverter) VectorGeometry(ctx context.Context, src string) ([]byte, if err := json.Unmarshal(out, &fc); err != nil { return nil, fmt.Errorf("parse ogr2ogr output: %w", err) } - - geoms := make([]json.RawMessage, 0, len(fc.Features)) - for _, f := range fc.Features { - if len(f.Geometry) == 0 || string(f.Geometry) == "null" { - continue - } - geoms = append(geoms, f.Geometry) - } - if len(geoms) == 0 { + if len(fc.Features) == 0 { return nil, nil } - return json.Marshal(struct { - Type string `json:"type"` - Geometries []json.RawMessage `json:"geometries"` - }{Type: "GeometryCollection", Geometries: geoms}) + geoms := make([]json.RawMessage, len(fc.Features)) + for i, f := range fc.Features { + if len(f.Geometry) == 0 || string(f.Geometry) == "null" { + continue + } + geoms[i] = f.Geometry + } + return geoms, nil } diff --git a/internal/repository/postgres/dataset.go b/internal/repository/postgres/dataset.go index c381bd7..ae90946 100644 --- a/internal/repository/postgres/dataset.go +++ b/internal/repository/postgres/dataset.go @@ -211,6 +211,24 @@ func (r *DatasetRepository) SetGeoJSON(ctx context.Context, id uuid.UUID, geojso return nil } +// ClearGeometryAndGeoJSON nulls out a dataset's stored geometry and persisted +// GeoJSON. Used before re-enqueuing a dataset's processing job so a stale +// (possibly incorrect) geometry/GeoJSON is never served while the job is +// pending; the async worker repopulates both once it completes. +func (r *DatasetRepository) ClearGeometryAndGeoJSON(ctx context.Context, id uuid.UUID) error { + tag, err := r.pool.Exec(ctx, + `UPDATE datasets SET geometry = NULL, geojson = NULL, updated_at = now() WHERE id = $1`, + id, + ) + if err != nil { + return mapError(err) + } + if tag.RowsAffected() == 0 { + return domain.ErrNotFound + } + return nil +} + // ReplaceObservations atomically replaces all observations for a dataset. func (r *DatasetRepository) ReplaceObservations(ctx context.Context, datasetID uuid.UUID, obs []domain.Observation) error { tx, err := r.pool.Begin(ctx) diff --git a/internal/service/dataset.go b/internal/service/dataset.go index 5e2d5ec..7f0a76b 100644 --- a/internal/service/dataset.go +++ b/internal/service/dataset.go @@ -44,6 +44,7 @@ type DatasetRepository interface { MarkConverted(ctx context.Context, id uuid.UUID, cogKey string, footprint []byte) error SetProperties(ctx context.Context, id uuid.UUID, properties, geometry []byte) error 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) 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) @@ -95,14 +96,16 @@ type ColumnParser func(filename string, data []byte) ([]domain.AttributeColumn, type RowParser func(filename string, data []byte) ([]map[string]string, error) // RasterConverter converts a raster file to a Cloud-Optimized GeoTIFF, reads its -// footprint, and dissolves a vector file's features into a single geometry. It -// operates on local file paths. +// footprint, and reads a vector file's per-feature geometry. It operates on +// local file paths. type RasterConverter interface { ToCOG(ctx context.Context, srcPath, dstPath string) error Footprint(ctx context.Context, srcPath string) ([]byte, error) - // VectorGeometry returns the combined geometry of a vector file's features as - // GeoJSON in EPSG:4326 (nil if the file has no features). - VectorGeometry(ctx context.Context, srcPath string) ([]byte, error) + // VectorFeatures returns each feature's geometry as GeoJSON in EPSG:4326, in + // file order (nil if the file has no features). A feature with no geometry + // yields a nil entry at its index, keeping the result aligned by position + // with attribute rows read from the same file via RowParser. + VectorFeatures(ctx context.Context, srcPath string) ([]json.RawMessage, error) } // UploadInput carries everything needed to store a new dataset. @@ -253,26 +256,35 @@ func (s *DatasetService) enqueueProcessing(ctx context.Context, d domain.Dataset // Reprocess re-enqueues the processing job for an existing dataset, restarting // its asynchronous pipeline from the step appropriate to its file type. Useful -// for retrying after a transient failure or a worker that was behind the schema. +// for retrying after a transient failure or a worker that was behind the +// schema. Its stored geometry and GeoJSON are cleared first so a stale (e.g. +// previously miscomputed) value is never served while the job is pending; the +// worker repopulates both once the job completes. func (s *DatasetService) Reprocess(ctx context.Context, id uuid.UUID) (domain.Dataset, error) { dataset, err := s.repo.GetByID(ctx, id) if err != nil { return domain.Dataset{}, err } + if err := s.repo.ClearGeometryAndGeoJSON(ctx, id); err != nil { + return domain.Dataset{}, fmt.Errorf("clear stale geometry/geojson: %w", err) + } if err := s.enqueueProcessing(ctx, dataset); err != nil { return domain.Dataset{}, fmt.Errorf("enqueue processing: %w", err) } return dataset, nil } -// ReprocessAll re-enqueues the processing job for every dataset, restarting each -// one's asynchronous pipeline from the step appropriate to its file type. It -// paginates through all datasets and continues past individual failures, -// returning how many jobs were enqueued and a per-dataset map of any failures. -func (s *DatasetService) ReprocessAll(ctx context.Context) (enqueued int, failures map[uuid.UUID]error, err error) { +// ReprocessAll re-enqueues the processing job for every dataset matching +// fileType (nil matches every file type), restarting each one's asynchronous +// pipeline from the step appropriate to its file type. Each dataset's stored +// geometry and GeoJSON are cleared first (see Reprocess). It paginates through +// all matching datasets and continues past individual failures, returning how +// many jobs were enqueued and a per-dataset map of any failures. +func (s *DatasetService) ReprocessAll(ctx context.Context, fileType *domain.FileType) (enqueued int, failures map[uuid.UUID]error, err error) { failures = make(map[uuid.UUID]error) + filter := domain.DatasetFilter{FileType: fileType} for offset := 0; ; { - summaries, err := s.repo.ListSummaries(ctx, domain.DatasetFilter{}, MaxPageSize, offset) + summaries, err := s.repo.ListSummaries(ctx, filter, MaxPageSize, offset) if err != nil { return enqueued, failures, err } @@ -280,6 +292,10 @@ func (s *DatasetService) ReprocessAll(ctx context.Context) (enqueued int, failur break } for _, sum := range summaries { + if e := s.repo.ClearGeometryAndGeoJSON(ctx, sum.ID); e != nil { + failures[sum.ID] = e + continue + } if e := s.enqueueProcessing(ctx, domain.Dataset{ID: sum.ID, FileType: sum.FileType}); e != nil { failures[sum.ID] = e continue @@ -294,12 +310,15 @@ func (s *DatasetService) ReprocessAll(ctx context.Context) (enqueued int, failur return enqueued, failures, nil } -// ExtractProperties reads a plain vector dataset's attribute table and spatial -// geometry and stores them (the attribute table as a JSON array of row objects -// in the properties column, the dissolved feature geometry in the geometry -// column), then marks the dataset ready. Invoked by the worker. Parse failures -// are recorded; storage failures are returned for retry. Geometry extraction is -// best-effort: a failure leaves geometry unset rather than failing the job. +// ExtractProperties reads a plain vector dataset's attribute table and +// per-feature geometry and stores them: the dissolved (unioned) geometry in +// the geometry column, and a GeoJSON FeatureCollection with one Feature per +// original polygon (its own geometry paired with its own attribute row, by +// position - the same pairing a shapefile's .shp/.dbf records have) as the +// persisted GeoJSON. Then marks the dataset ready. Invoked by the worker. +// Parse failures are recorded; storage failures are returned for retry. +// Geometry extraction is best-effort: a failure leaves geometry unset rather +// than failing the job. func (s *DatasetService) ExtractProperties(ctx context.Context, id uuid.UUID) error { dataset, err := s.repo.GetByID(ctx, id) if err != nil { @@ -326,20 +345,27 @@ func (s *DatasetService) ExtractProperties(ctx context.Context, id uuid.UUID) er } } - geometry := s.vectorGeometry(ctx, dataset.Filename, data) - if err := s.repo.SetProperties(ctx, id, properties, geometry); err != nil { + geoms := s.vectorFeatureGeometries(ctx, dataset.Filename, data) + if err := s.repo.SetProperties(ctx, id, properties, dissolvedGeometryJSON(geoms)); err != nil { return err } - // Assemble and persist the GeoJSON served by the .geojson endpoint from the - // now-canonical (unioned) geometry and attribute table. - return s.generateGeoJSON(ctx, id) + + fc := domain.FeatureCollection{ + Type: domain.GeoJSONFeatureCollection, + Features: buildVectorFeatures(rows, geoms), + } + raw, err := json.Marshal(fc) + if err != nil { + return err + } + return s.repo.SetGeoJSON(ctx, id, raw) } -// vectorGeometry dissolves a vector file's features into a single GeoJSON -// geometry via the converter. It writes the in-memory bytes to a temp file -// (preserving the extension so the converter detects the format) because the -// converter operates on file paths. Best-effort: any failure yields nil. -func (s *DatasetService) vectorGeometry(ctx context.Context, filename string, data []byte) []byte { +// vectorFeatureGeometries reads a vector file's per-feature geometries via the +// converter. It writes the in-memory bytes to a temp file (preserving the +// extension so the converter detects the format) because the converter +// operates on file paths. Best-effort: any failure yields nil. +func (s *DatasetService) vectorFeatureGeometries(ctx context.Context, filename string, data []byte) []json.RawMessage { ext := strings.ToLower(filepath.Ext(filename)) f, err := os.CreateTemp("", "gis-vec-*"+ext) if err != nil { @@ -353,8 +379,61 @@ func (s *DatasetService) vectorGeometry(ctx context.Context, filename string, da if err := f.Close(); err != nil { return nil } - geom, _ := s.converter.VectorGeometry(ctx, f.Name()) - return geom + geoms, _ := s.converter.VectorFeatures(ctx, f.Name()) + return geoms +} + +// dissolvedGeometryJSON wraps a vector file's per-feature geometries into a +// single GeoJSON GeometryCollection for storage in the geometry column, where +// PostGIS dissolves it into the union of all features (nil entries and an +// empty/nil input are dropped, yielding nil when no geometry survives). +func dissolvedGeometryJSON(geoms []json.RawMessage) []byte { + present := make([]json.RawMessage, 0, len(geoms)) + for _, g := range geoms { + if len(g) == 0 || string(g) == "null" { + continue + } + present = append(present, g) + } + if len(present) == 0 { + return nil + } + b, err := json.Marshal(struct { + Type string `json:"type"` + Geometries []json.RawMessage `json:"geometries"` + }{Type: "GeometryCollection", Geometries: present}) + if err != nil { + return nil + } + return b +} + +// buildVectorFeatures pairs each row with the geometry at the same position +// (rows and geoms both come from reading the same file in file order, which +// is how a shapefile's .dbf attribute records correspond to its .shp geometry +// records) into one Feature per polygon. A row or geometry beyond the other +// slice's length, or a feature with no geometry, is skipped. +func buildVectorFeatures(rows []map[string]string, geoms []json.RawMessage) []domain.Feature { + n := len(geoms) + if len(rows) < n { + n = len(rows) + } + features := make([]domain.Feature, 0, n) + for i := 0; i < n; i++ { + if len(geoms[i]) == 0 || string(geoms[i]) == "null" { + continue + } + props := make(map[string]any, len(rows[i])) + for k, v := range rows[i] { + props[k] = v + } + features = append(features, domain.Feature{ + Type: domain.GeoJSONFeature, + Geometry: geoms[i], + Properties: props, + }) + } + return features } // hasAttributeData reports whether any row carries at least one attribute. @@ -572,8 +651,8 @@ func (s *DatasetService) Extract(ctx context.Context, id uuid.UUID) error { return err // transient } - geometry := s.vectorGeometry(ctx, dataset.Filename, data) - if err := s.repo.MarkReady(ctx, id, geometry); err != nil { + geoms := s.vectorFeatureGeometries(ctx, dataset.Filename, data) + if err := s.repo.MarkReady(ctx, id, dissolvedGeometryJSON(geoms)); err != nil { return err } // Assemble and persist the GeoJSON served by the .geojson endpoint by joining diff --git a/internal/service/dataset_test.go b/internal/service/dataset_test.go index 18ae8e0..a3480f7 100644 --- a/internal/service/dataset_test.go +++ b/internal/service/dataset_test.go @@ -7,6 +7,7 @@ import ( "errors" "io" "os" + "sort" "strings" "testing" "time" @@ -23,6 +24,7 @@ type stubDatasetRepo struct { districts map[string]domain.District createErr error deleted []uuid.UUID + cleared []uuid.UUID lastLimit, lastOffset int lastFilter domain.DatasetFilter } @@ -58,7 +60,24 @@ func (r *stubDatasetRepo) ListSummaries(_ context.Context, filter domain.Dataset r.lastFilter = filter r.lastLimit = limit r.lastOffset = offset - return nil, nil + + var matched []domain.DatasetSummary + for _, d := range r.store { + if filter.FileType != nil && d.FileType != *filter.FileType { + continue + } + matched = append(matched, domain.DatasetSummary{ID: d.ID, FileType: d.FileType}) + } + sort.Slice(matched, func(i, j int) bool { return matched[i].ID.String() < matched[j].ID.String() }) + + if offset >= len(matched) { + return nil, nil + } + end := offset + limit + if end > len(matched) { + end = len(matched) + } + return matched[offset:end], nil } func (r *stubDatasetRepo) Count(_ context.Context, _ domain.DatasetFilter) (int, error) { @@ -146,6 +165,18 @@ func (r *stubDatasetRepo) SetProperties(_ context.Context, id uuid.UUID, propert return nil } +func (r *stubDatasetRepo) ClearGeometryAndGeoJSON(_ context.Context, id uuid.UUID) error { + d, ok := r.store[id] + if !ok { + return domain.ErrNotFound + } + d.Geometry = nil + d.GeoJSON = nil + r.store[id] = d + r.cleared = append(r.cleared, id) + return nil +} + func (r *stubDatasetRepo) SetGeoJSON(_ context.Context, id uuid.UUID, geojson []byte) error { d, ok := r.store[id] if !ok { @@ -226,12 +257,12 @@ func (s *stubEnqueuer) EnqueueConvert(_ context.Context, id uuid.UUID) error { // stubConverter records raster conversions. type stubConverter struct { - cogCalls int - toCOGErr error - footprint []byte - footprintFn func(src string) ([]byte, error) - vectorGeom []byte - vectorGeomFn func(src string) ([]byte, error) + cogCalls int + toCOGErr error + footprint []byte + footprintFn func(src string) ([]byte, error) + vectorFeatures []json.RawMessage + vectorFeaturesFn func(src string) ([]json.RawMessage, error) } func (c *stubConverter) ToCOG(_ context.Context, _, dst string) error { @@ -249,11 +280,11 @@ func (c *stubConverter) Footprint(_ context.Context, src string) ([]byte, error) return c.footprint, nil } -func (c *stubConverter) VectorGeometry(_ context.Context, src string) ([]byte, error) { - if c.vectorGeomFn != nil { - return c.vectorGeomFn(src) +func (c *stubConverter) VectorFeatures(_ context.Context, src string) ([]json.RawMessage, error) { + if c.vectorFeaturesFn != nil { + return c.vectorFeaturesFn(src) } - return c.vectorGeom, nil + return c.vectorFeatures, nil } var ( @@ -471,8 +502,11 @@ func TestDatasetService_ExtractProperties(t *testing.T) { {"name": "Almaty", "pop": "2000"}, } rp := RowParser(func(string, []byte) ([]map[string]string, error) { return rows, nil }) - geom := []byte(`{"type":"GeometryCollection","geometries":[]}`) - conv := &stubConverter{vectorGeom: geom} + geoms := []json.RawMessage{ + json.RawMessage(`{"type":"Polygon","coordinates":[[[1,1],[2,2],[3,1],[1,1]]]}`), + json.RawMessage(`{"type":"Polygon","coordinates":[[[4,4],[5,5],[6,4],[4,4]]]}`), + } + conv := &stubConverter{vectorFeatures: geoms} svc := NewDatasetService(repo, &stubStore{}, stubCategoryReader{exists: true}, &stubEnqueuer{}, noopParser, rp, conv) if err := svc.ExtractProperties(context.Background(), id); err != nil { @@ -489,8 +523,25 @@ func TestDatasetService_ExtractProperties(t *testing.T) { if len(parsed) != 2 { t.Fatalf("want 2 rows in properties, got %d", len(parsed)) } - if string(got.Geometry) != string(geom) { - t.Fatalf("want geometry %s, got %s", geom, got.Geometry) + wantGeometry := `{"type":"GeometryCollection","geometries":[` + string(geoms[0]) + `,` + string(geoms[1]) + `]}` + if string(got.Geometry) != wantGeometry { + t.Fatalf("want dissolved geometry %s, got %s", wantGeometry, got.Geometry) + } + + // Regression: each polygon must keep its own attribute row as its Feature's + // properties, not a single dissolved geometry sharing every row. + var fc domain.FeatureCollection + if err := json.Unmarshal(got.GeoJSON, &fc); err != nil { + t.Fatalf("geojson not persisted / invalid: %v (%s)", err, got.GeoJSON) + } + if len(fc.Features) != 2 { + t.Fatalf("want 2 features (one per polygon), got %d: %s", len(fc.Features), got.GeoJSON) + } + if string(fc.Features[0].Geometry) != string(geoms[0]) || fc.Features[0].Properties["name"] != "Astana" { + t.Fatalf("feature 0 not paired with its own geometry/row: %+v", fc.Features[0]) + } + if string(fc.Features[1].Geometry) != string(geoms[1]) || fc.Features[1].Properties["name"] != "Almaty" { + t.Fatalf("feature 1 not paired with its own geometry/row: %+v", fc.Features[1]) } } @@ -505,7 +556,7 @@ func TestDatasetService_ExtractProperties_GeometryBestEffort(t *testing.T) { return []map[string]string{{"name": "Astana"}}, nil }) // Geometry extraction fails; the job must still succeed with properties set. - conv := &stubConverter{vectorGeomFn: func(string) ([]byte, error) { + conv := &stubConverter{vectorFeaturesFn: func(string) ([]json.RawMessage, error) { return nil, errors.New("ogr2ogr boom") }} svc := NewDatasetService(repo, &stubStore{}, stubCategoryReader{exists: true}, &stubEnqueuer{}, noopParser, rp, conv) @@ -800,8 +851,9 @@ func TestDatasetService_Extract(t *testing.T) { } rows := []map[string]string{{"като": "751010000", "F_2023": "100"}} rp := RowParser(func(string, []byte) ([]map[string]string, error) { return rows, nil }) - geom := []byte(`{"type":"GeometryCollection","geometries":[]}`) - conv := &stubConverter{vectorGeom: geom} + geoms := []json.RawMessage{json.RawMessage(`{"type":"Polygon","coordinates":[[[76,43],[77,43],[77,44],[76,43]]]}`)} + geom := []byte(`{"type":"GeometryCollection","geometries":[` + string(geoms[0]) + `]}`) + conv := &stubConverter{vectorFeatures: geoms} svc := NewDatasetService(repo, &stubStore{}, stubCategoryReader{exists: true}, &stubEnqueuer{}, noopParser, rp, conv) if err := svc.Extract(context.Background(), id); err != nil { @@ -1166,6 +1218,69 @@ func TestDatasetService_WaitForStatus(t *testing.T) { }) } +func TestDatasetService_Reprocess_ClearsStaleGeometryAndGeoJSON(t *testing.T) { + repo := newStubDatasetRepo() + id := uuid.New() + repo.store[id] = domain.Dataset{ + ID: id, FileType: domain.FileTypeVector, Status: domain.DatasetStatusReady, + Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[1,1],[2,2],[3,1],[1,1]]]}`), + GeoJSON: json.RawMessage(`{"type":"FeatureCollection","features":[{"type":"Feature","properties":{"rows":[]}}]}`), + } + enq := &stubEnqueuer{} + svc := NewDatasetService(repo, &stubStore{}, stubCategoryReader{exists: true}, enq, noopParser, noopRowParser, &stubConverter{}) + + if _, err := svc.Reprocess(context.Background(), id); err != nil { + t.Fatalf("unexpected error: %v", err) + } + got := repo.store[id] + if got.Geometry != nil || got.GeoJSON != nil { + t.Fatalf("want stale geometry/geojson cleared, got geometry=%s geojson=%s", got.Geometry, got.GeoJSON) + } + if len(enq.properties) != 1 || enq.properties[0] != id { + t.Fatalf("want properties job enqueued for %s, got %+v", id, enq.properties) + } +} + +func TestDatasetService_ReprocessAll_FiltersByFileType(t *testing.T) { + repo := newStubDatasetRepo() + vectorID, rasterID := uuid.New(), uuid.New() + repo.store[vectorID] = domain.Dataset{ + ID: vectorID, FileType: domain.FileTypeVector, + Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[1,1],[2,2],[3,1],[1,1]]]}`), + GeoJSON: json.RawMessage(`{"type":"FeatureCollection","features":[]}`), + } + repo.store[rasterID] = domain.Dataset{ + ID: rasterID, FileType: domain.FileTypeRaster, + Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[1,1],[2,2],[3,1],[1,1]]]}`), + } + enq := &stubEnqueuer{} + svc := NewDatasetService(repo, &stubStore{}, stubCategoryReader{exists: true}, enq, noopParser, noopRowParser, &stubConverter{}) + + vectorType := domain.FileTypeVector + enqueued, failures, err := svc.ReprocessAll(context.Background(), &vectorType) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(failures) != 0 { + t.Fatalf("unexpected failures: %+v", failures) + } + if enqueued != 1 { + t.Fatalf("want 1 dataset enqueued, got %d", enqueued) + } + if len(enq.properties) != 1 || enq.properties[0] != vectorID { + t.Fatalf("want only the vector dataset's properties job enqueued, got %+v", enq.properties) + } + if len(enq.converted) != 0 { + t.Fatalf("raster dataset must not be reprocessed when --type=vector, got %+v", enq.converted) + } + if got := repo.store[vectorID]; got.Geometry != nil || got.GeoJSON != nil { + t.Fatalf("vector dataset's stale geometry/geojson must be cleared, got geometry=%s geojson=%s", got.Geometry, got.GeoJSON) + } + if got := repo.store[rasterID]; got.Geometry == nil { + t.Fatalf("raster dataset outside the --type filter must be left untouched") + } +} + func TestDatasetService_Delete_RemovesObject(t *testing.T) { repo := newStubDatasetRepo() id := uuid.New()