1package authoring23import (4 "bytes"5 "encoding/json"6 "errors"7 "fmt"8 "io"9 "math/big"10 "net/url"11 "path/filepath"12 "strconv"13 "strings"14)1516const (17 TerminationSinglePage = "single-page"18 TerminationNextExhausted = "next-exhausted"19 TerminationTransferLimitClear = "transfer-limit-clear"20)2122// AdapterRegistry is an immutable lookup table for acquisition protocol23// behavior. A fresh registry is cheap and avoids process-wide mutable state.24type AdapterRegistry struct {25 adapters map[string]Adapter26}2728// Adapter owns protocol versioning and continuation semantics. Mapping source29// records into typed Atlas features deliberately remains a separate concern.30type Adapter interface {31 Name() string32 Version() string33 PlanFeature(Project, Source) (Request, error)34 Next(Request, []byte, *PageProgress) (PageDecision, error)35}3637type PageProgress struct {38 RootLocator string39 Returned int6440 Matched *int6441 SeenIDs map[string]bool42 LastID string43}4445type PageDecision struct {46 Done bool47 Next Request48 Termination string49 Returned int6450 Matched *int6451}5253type adapterDefinition struct {54 name string55 version string56 next func(Request, []byte, *PageProgress) (PageDecision, error)57 plan func(Project, Source) (Request, error)58}5960func (adapter adapterDefinition) Name() string { return adapter.name }6162func (adapter adapterDefinition) Version() string { return adapter.version }6364func (adapter adapterDefinition) PlanFeature(project Project, source Source) (Request, error) {65 if adapter.plan == nil {66 return Request{}, fmt.Errorf("adapter %s does not acquire features", adapter.name)67 }68 request, err := adapter.plan(project, source)69 if err != nil {70 return Request{}, err71 }72 request.Adapter = adapter.name73 request.AdapterVersion = adapter.version74 request.AcquisitionID = acquisitionID(request)75 request.ID = requestID(request)76 return request, nil77}7879func (adapter adapterDefinition) Next(request Request, body []byte, progress *PageProgress) (PageDecision, error) {80 return adapter.next(request, body, progress)81}8283func defaultAdapterRegistry() AdapterRegistry {84 definitions := []adapterDefinition{85 {name: "geojson", version: "geojson/v1", plan: planGeoJSON, next: geoJSONPage},86 {name: "arcgis-feature-service", version: "arcgis-feature-service/v2", plan: planArcGIS, next: arcGISPage},87 {name: "ogc-api-features", version: "ogc-api-features/v2", plan: planOGC, next: ogcPage},88 {name: "raster-file", version: "raster-file/v1", next: singlePage},89 {name: "xyz", version: "xyz/v1", next: singlePage},90 {name: "wmts", version: "wmts-rest/v1", next: singlePage},91 {name: "asset-file", version: "asset-file/v1", next: singlePage},92 }93 registry := AdapterRegistry{adapters: make(map[string]Adapter, len(definitions))}94 for index := range definitions {95 definition := definitions[index]96 registry.adapters[definition.name] = definition97 }98 return registry99}100101func planGeoJSON(project Project, source Source) (Request, error) {102 return newFeatureRequest(project, source, project.ResolveLocator(source.Locator), source.Locator), nil103}104105func planArcGIS(project Project, source Source) (Request, error) {106 parsed, err := url.Parse(project.ResolveLocator(source.Locator))107 if err != nil {108 return Request{}, fmt.Errorf("source %s locator: %w", source.ID, err)109 }110 path := strings.TrimSuffix(parsed.Path, "/")111 if source.Query.Layer > 0 && filepath.Base(path) != strconv.Itoa(source.Query.Layer) {112 path += "/" + strconv.Itoa(source.Query.Layer)113 }114 if !strings.HasSuffix(path, "/query") {115 path += "/query"116 }117 parsed.Path = path118 query := parsed.Query()119 query.Set("f", "geojson")120 query.Set("outFields", "*")121 query.Set("returnGeometry", "true")122 extent, err := sourceExtent(source.Mapping.Geometry, project.Target.CoordinateSpace)123 if err != nil {124 return Request{}, fmt.Errorf("source %s query extent: %w", source.ID, err)125 }126 query.Set("geometry", extentString(extent))127 query.Set("geometryType", "esriGeometryEnvelope")128 query.Set("spatialRel", "esriSpatialRelIntersects")129 spatialReference := source.Query.SpatialReference130 if spatialReference == 0 {131 spatialReference = recognizedSpatialReference(source.Mapping.Geometry.SourceSpace)132 }133 if spatialReference == 0 {134 return Request{}, fmt.Errorf("source %s requires query.spatial-reference for source space %q", source.ID, source.Mapping.Geometry.SourceSpace)135 }136 query.Set("inSR", strconv.Itoa(spatialReference))137 query.Set("outSR", strconv.Itoa(spatialReference))138 objectID := source.Query.ObjectID139 if objectID == "" {140 objectID = "OBJECTID"141 }142 if !validArcGISField(objectID) {143 return Request{}, fmt.Errorf("source %s has invalid query.object-id %q", source.ID, objectID)144 }145 query.Set("orderByFields", objectID+" ASC")146 where := source.Query.Where147 if where == "" {148 where = "1=1"149 }150 query.Set("where", where)151 if source.Query.Limit > 0 {152 query.Set("resultRecordCount", strconv.Itoa(source.Query.Limit))153 }154 query.Set("resultOffset", "0")155 parsed.RawQuery = query.Encode()156 return newFeatureRequest(project, source, parsed.String(), parsed.String()), nil157}158159func recognizedSpatialReference(sourceSpace string) int {160 switch strings.ToUpper(strings.TrimSpace(sourceSpace)) {161 case "EPSG:4326", "OGC:CRS84", "CRS84", "WGS84", "URN:OGC:DEF:CRS:OGC::CRS84":162 return 4326163 case "EPSG:3857", "EPSG:102100", "ESRI:102100", "WEB-MERCATOR":164 return 3857165 default:166 return 0167 }168}169170func validArcGISField(field string) bool {171 for index, character := range field {172 if (character >= 'a' && character <= 'z') || (character >= 'A' && character <= 'Z') || character == '_' || (index > 0 && ((character >= '0' && character <= '9') || character == '.')) {173 continue174 }175 return false176 }177 return field != ""178}179180func planOGC(project Project, source Source) (Request, error) {181 if source.Query.Collection == "" {182 return Request{}, fmt.Errorf("source %s requires query.collection", source.ID)183 }184 parsed, err := url.Parse(project.ResolveLocator(source.Locator))185 if err != nil {186 return Request{}, fmt.Errorf("source %s locator: %w", source.ID, err)187 }188 parsed.Path = strings.TrimSuffix(parsed.Path, "/") + "/collections/" + url.PathEscape(source.Query.Collection) + "/items"189 query := parsed.Query()190 extent, err := sourceExtent(source.Mapping.Geometry, project.Target.CoordinateSpace)191 if err != nil {192 return Request{}, fmt.Errorf("source %s query extent: %w", source.ID, err)193 }194 query.Set("bbox", extentString(extent))195 query.Set("f", "json")196 if source.Query.Limit > 0 {197 query.Set("limit", strconv.Itoa(source.Query.Limit))198 }199 parsed.RawQuery = query.Encode()200 return newFeatureRequest(project, source, parsed.String(), parsed.String()), nil201}202203func newFeatureRequest(project Project, source Source, locator, identity string) Request {204 return Request{205 Kind: RequestFeatures, Source: source.ID, Adapter: source.Adapter,206 Locator: locator, IdentityLocator: identity, MediaType: source.MediaType,207 License: source.License, Attribution: source.Attribution,208 }209}210211func (registry AdapterRegistry) Lookup(name string) (Adapter, error) {212 adapter, ok := registry.adapters[name]213 if !ok {214 return nil, fmt.Errorf("unknown adapter %q", name)215 }216 return adapter, nil217}218219func singlePage(_ Request, _ []byte, progress *PageProgress) (PageDecision, error) {220 return PageDecision{Done: true, Termination: TerminationSinglePage, Returned: progress.Returned}, nil221}222223func geoJSONPage(_ Request, body []byte, progress *PageProgress) (PageDecision, error) {224 page, err := decodeFeaturePage(body)225 if err != nil {226 return PageDecision{}, err227 }228 if page.ExceededTransferLimit || nextLink(page.Links) != "" {229 return PageDecision{}, fmt.Errorf("GeoJSON response declares pagination but the geojson adapter is single-page")230 }231 returned, err := pageReturned(page)232 if err != nil {233 return PageDecision{}, err234 }235 return PageDecision{Done: true, Termination: TerminationSinglePage, Returned: progress.Returned + returned}, nil236}237238func arcGISPage(request Request, body []byte, progress *PageProgress) (PageDecision, error) {239 page, err := decodeFeaturePage(body)240 if err != nil {241 return PageDecision{}, err242 }243 returned, err := pageReturned(page)244 if err != nil {245 return PageDecision{}, err246 }247 total := progress.Returned + returned248 if err := recordPageIDs("ArcGIS", page, progress, true); err != nil {249 return PageDecision{}, err250 }251 if !page.ExceededTransferLimit {252 if progress.Returned > 0 && returned == 0 {253 return PageDecision{}, fmt.Errorf("ArcGIS continuation is missing the page promised by its transfer limit")254 }255 return PageDecision{Done: true, Termination: TerminationTransferLimitClear, Returned: total}, nil256 }257 if returned == 0 {258 return PageDecision{}, fmt.Errorf("ArcGIS transfer limit continuation returned no features")259 }260 parsed, err := url.Parse(request.IdentityLocator)261 if err != nil {262 return PageDecision{}, fmt.Errorf("parse ArcGIS continuation: %w", err)263 }264 query := parsed.Query()265 offset, err := optionalInt64(query.Get("resultOffset"))266 if err != nil {267 return PageDecision{}, fmt.Errorf("ArcGIS resultOffset: %w", err)268 }269 query.Set("resultOffset", strconv.FormatInt(offset+returned, 10))270 parsed.RawQuery = query.Encode()271 next := continuationRequest(request, parsed.String())272 return PageDecision{Next: next, Returned: total}, nil273}274275func compareStableID(left, right string) int {276 leftInteger, leftOK := new(big.Int).SetString(left, 10)277 rightInteger, rightOK := new(big.Int).SetString(right, 10)278 if leftOK && rightOK {279 return leftInteger.Cmp(rightInteger)280 }281 return strings.Compare(left, right)282}283284func recordPageIDs(protocol string, page geoJSON, progress *PageProgress, ordered bool) error {285 if progress.SeenIDs == nil {286 progress.SeenIDs = make(map[string]bool)287 }288 for _, feature := range page.Features {289 id := scalarString(feature.ID)290 if id == "" {291 return fmt.Errorf("%s response has a feature without an object ID", protocol)292 }293 if progress.SeenIDs[id] {294 return fmt.Errorf("%s response repeats object ID %s across pages", protocol, id)295 }296 if ordered && progress.LastID != "" && compareStableID(progress.LastID, id) >= 0 {297 return fmt.Errorf("%s response object IDs are not strictly ordered", protocol)298 }299 progress.SeenIDs[id] = true300 progress.LastID = id301 }302 return nil303}304305func ogcPage(request Request, body []byte, progress *PageProgress) (PageDecision, error) {306 page, err := decodeFeaturePage(body)307 if err != nil {308 return PageDecision{}, err309 }310 returned, err := pageReturned(page)311 if err != nil {312 return PageDecision{}, err313 }314 total := progress.Returned + returned315 if err := recordPageIDs("OGC", page, progress, false); err != nil {316 return PageDecision{}, err317 }318 matched, err := reconcileMatched(progress.Matched, page.NumberMatched)319 if err != nil {320 return PageDecision{}, err321 }322 next := nextLink(page.Links)323 if next != "" {324 if matched != nil && total >= *matched {325 return PageDecision{}, fmt.Errorf("OGC response continues after all %d matched features were returned", *matched)326 }327 locator, err := resolveContinuation(progress.RootLocator, request.IdentityLocator, next)328 if err != nil {329 return PageDecision{}, err330 }331 return PageDecision{Next: continuationRequest(request, locator), Returned: total, Matched: matched}, nil332 }333 if matched != nil && total != *matched {334 return PageDecision{}, fmt.Errorf("OGC response is incomplete: %d matched but only %d returned", *matched, total)335 }336 return PageDecision{Done: true, Termination: TerminationNextExhausted, Returned: total, Matched: matched}, nil337}338339func decodeFeaturePage(body []byte) (geoJSON, error) {340 decoder := json.NewDecoder(bytes.NewReader(body))341 decoder.UseNumber()342 var page geoJSON343 if err := decoder.Decode(&page); err != nil {344 return geoJSON{}, fmt.Errorf("decode feature page: %w", err)345 }346 if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {347 if err == nil {348 return geoJSON{}, fmt.Errorf("feature page carries more than one JSON value")349 }350 return geoJSON{}, fmt.Errorf("decode feature page trailer: %w", err)351 }352 if page.Type != "FeatureCollection" {353 return geoJSON{}, fmt.Errorf("feature page is %q, want FeatureCollection", page.Type)354 }355 return page, nil356}357358func pageReturned(page geoJSON) (int64, error) {359 returned := int64(len(page.Features))360 if page.NumberReturned != nil && *page.NumberReturned != returned {361 return 0, fmt.Errorf("response count differs from its feature payload")362 }363 return returned, nil364}365366func reconcileMatched(held, next *int64) (*int64, error) {367 if next == nil {368 return held, nil369 }370 if *next < 0 {371 return nil, fmt.Errorf("response has a negative matched count")372 }373 if held != nil && *held != *next {374 return nil, fmt.Errorf("response changed numberMatched from %d to %d", *held, *next)375 }376 value := *next377 return &value, nil378}379380func nextLink(links []geoJSONLink) string {381 for _, link := range links {382 if strings.EqualFold(link.Rel, "next") && strings.TrimSpace(link.Href) != "" {383 return strings.TrimSpace(link.Href)384 }385 }386 return ""387}388389func resolveContinuation(rootLocator, currentLocator, reference string) (string, error) {390 root, err := url.Parse(rootLocator)391 if err != nil {392 return "", fmt.Errorf("parse continuation root: %w", err)393 }394 current, err := url.Parse(currentLocator)395 if err != nil {396 return "", fmt.Errorf("parse current page: %w", err)397 }398 next, err := url.Parse(reference)399 if err != nil {400 return "", fmt.Errorf("parse next page: %w", err)401 }402 resolved := current.ResolveReference(next)403 if !strings.EqualFold(root.Scheme, resolved.Scheme) || !strings.EqualFold(root.Host, resolved.Host) {404 return "", fmt.Errorf("next page leaves the configured source origin")405 }406 return resolved.String(), nil407}408409func continuationRequest(root Request, locator string) Request {410 next := root411 next.Locator = locator412 next.IdentityLocator = locator413 finalizeRequest(&next)414 return next415}416417func optionalInt64(value string) (int64, error) {418 if value == "" {419 return 0, nil420 }421 return strconv.ParseInt(value, 10, 64)422}