fix: Fix processing for vector attributres table

2026-07-10 23:34:19 +05:00
parent 017f869380
commit 47b2e8653d
6 changed files with 315 additions and 80 deletions
--- 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