Go Data Provider
This guide shows how to build a GW data provider using Gin. Both the middleware wrapper and manual patterns are demonstrated.
Project Setup
go mod init github.com/yourorg/data-provider
go get github.com/gin-gonic/gin
go get github.com/aws/aws-sdk-go-v2/service/s3
go get github.com/aws/aws-sdk-go-v2/credentials
Types
Define the shared types for the query envelope and callback payload:
handlers/types.go
package handlers
// DeliverySpec describes how results should be delivered.
type DeliverySpec struct {
Mechanism string `json:"mechanism"`
URL string `json:"url,omitempty"`
S3Bucket string `json:"s3Bucket,omitempty"`
S3KeyPrefix string `json:"s3KeyPrefix,omitempty"`
}
// QueryEnvelope is the incoming request body from GW.
type QueryEnvelope struct {
QueryID string `json:"queryId"`
DatasetID string `json:"datasetId"`
Endpoint string `json:"endpoint"`
Parameters map[string]interface{} `json:"parameters"`
Delivery DeliverySpec `json:"delivery"`
CallbackURL string `json:"callbackUrl"`
CallbackToken string `json:"callbackToken"`
}
// CallbackPayload is the body POSTed back to GW after processing a query.
type CallbackPayload struct {
QueryID string `json:"queryId"`
Status string `json:"status"`
RecordCount int `json:"recordCount"`
ExecutionTimeMs int `json:"executionTimeMs"`
Delivery *DeliverySpec `json:"delivery,omitempty"`
Error string `json:"error,omitempty"`
}
Server Entry Point
main.go
package main
import (
"fmt"
"log"
"net/http"
"os"
"github.com/gin-gonic/gin"
"github.com/yourorg/data-provider/handlers"
)
func getEnv(key, fallback string) string {
if v := os.Getenv(key); v != "" {
return v
}
return fallback
}
func main() {
port := getEnv("PORT", "3002")
gwAPIKey := getEnv("GW_API_KEY", "demo-api-key")
config := handlers.ProviderConfig{
S3Endpoint: getEnv("S3_ENDPOINT", "http://localstack:4566"),
S3Bucket: getEnv("S3_BUCKET", "gw-deliveries"),
GWApiKey: gwAPIKey,
GWCallbackSecret: getEnv("GW_CALLBACK_SECRET", "demo-secret"),
}
r := gin.Default()
// API-key validation middleware
r.Use(func(c *gin.Context) {
if c.Request.URL.Path == "/health" {
c.Next()
return
}
key := c.GetHeader("X-GW-Api-Key")
if key != gwAPIKey {
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{
"error": "Invalid or missing X-GW-Api-Key header",
})
return
}
c.Next()
})
// Middleware pattern
r.POST("/api/search/companies", handlers.CompanySearch(config))
// Manual pattern
r.POST("/api/search/sanctions", handlers.SanctionsCheck(config))
// Health check
r.GET("/health", func(c *gin.Context) {
c.JSON(http.StatusOK, gin.H{"status": "ok", "service": "data-provider-go"})
})
log.Printf("[data-provider] listening on :%s", port)
if err := r.Run(fmt.Sprintf(":%s", port)); err != nil {
log.Fatalf("Server failed: %v", err)
}
}
Result Delivery
Route results to S3, a webhook, or return them inline:
handlers/delivery.go
package handlers
import (
"bytes"
"context"
"encoding/json"
"fmt"
"net/http"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/credentials"
"github.com/aws/aws-sdk-go-v2/service/s3"
)
// DeliveryResult describes the outcome of a delivery operation.
type DeliveryResult struct {
Mechanism string `json:"mechanism"`
Reference string `json:"reference,omitempty"`
StatusCode int `json:"statusCode,omitempty"`
}
// ProviderConfig holds runtime configuration.
type ProviderConfig struct {
S3Endpoint string
S3Bucket string
GWApiKey string
GWCallbackSecret string
}
// DeliverResults routes query results to the target described by the envelope.
func DeliverResults(envelope QueryEnvelope, results interface{}, config ProviderConfig) (DeliveryResult, error) {
switch envelope.Delivery.Mechanism {
case "gw_s3_sync":
return deliverToS3(envelope, results, config)
case "webhook":
return deliverToWebhook(envelope, results)
case "inline":
return DeliveryResult{Mechanism: "inline"}, nil
default:
return DeliveryResult{Mechanism: "inline"}, nil
}
}
func deliverToS3(envelope QueryEnvelope, results interface{}, config ProviderConfig) (DeliveryResult, error) {
bucket := envelope.Delivery.S3Bucket
if bucket == "" {
bucket = config.S3Bucket
}
key := fmt.Sprintf("%s%s.json", envelope.Delivery.S3KeyPrefix, envelope.QueryID)
body, err := json.Marshal(results)
if err != nil {
return DeliveryResult{}, fmt.Errorf("marshal results: %w", err)
}
client := s3.New(s3.Options{
BaseEndpoint: aws.String(config.S3Endpoint),
Region: "us-east-1",
UsePathStyle: true,
Credentials: credentials.NewStaticCredentialsProvider("test", "test", ""),
})
_, err = client.PutObject(context.TODO(), &s3.PutObjectInput{
Bucket: aws.String(bucket),
Key: aws.String(key),
Body: bytes.NewReader(body),
ContentType: aws.String("application/json"),
})
if err != nil {
return DeliveryResult{}, fmt.Errorf("S3 PutObject: %w", err)
}
return DeliveryResult{
Mechanism: "gw_s3_sync",
Reference: fmt.Sprintf("s3://%s/%s", bucket, key),
}, nil
}
func deliverToWebhook(envelope QueryEnvelope, results interface{}) (DeliveryResult, error) {
url := envelope.Delivery.URL
if url == "" {
return DeliveryResult{}, fmt.Errorf("webhook delivery requires delivery.url")
}
payload, err := json.Marshal(map[string]interface{}{
"queryId": envelope.QueryID,
"results": results,
})
if err != nil {
return DeliveryResult{}, fmt.Errorf("marshal webhook payload: %w", err)
}
resp, err := http.Post(url, "application/json", bytes.NewReader(payload))
if err != nil {
return DeliveryResult{}, fmt.Errorf("webhook POST: %w", err)
}
defer resp.Body.Close()
return DeliveryResult{Mechanism: "webhook", StatusCode: resp.StatusCode}, nil
}
HMAC-Signed Callback
handlers/callback.go
package handlers
import (
"bytes"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"log"
"net/http"
)
// PostCallback sends an HMAC-SHA256 signed callback to GW.
func PostCallback(callbackURL, callbackToken string, payload CallbackPayload) {
body, err := json.Marshal(payload)
if err != nil {
log.Printf("[callback] Failed to marshal payload for query %s: %v", payload.QueryID, err)
return
}
mac := hmac.New(sha256.New, []byte(callbackToken))
mac.Write(body)
signature := hex.EncodeToString(mac.Sum(nil))
req, err := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
if err != nil {
log.Printf("[callback] Failed to create request for query %s: %v", payload.QueryID, err)
return
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-GW-Signature", fmt.Sprintf("sha256=%s", signature))
resp, err := http.DefaultClient.Do(req)
if err != nil {
log.Printf("[callback] Failed to POST callback for query %s: %v", payload.QueryID, err)
return
}
defer resp.Body.Close()
}
Middleware Pattern
A generic wrapper handles envelope parsing, delivery, and callbacks. Your handler only contains business logic:
handlers/companies.go
package handlers
import (
"log"
"net/http"
"time"
"github.com/gin-gonic/gin"
)
// gwMiddleware wraps a business-logic handler with GW protocol plumbing.
func gwMiddleware(config ProviderConfig, handler func(params map[string]interface{}, envelope QueryEnvelope) (interface{}, error)) gin.HandlerFunc {
return func(c *gin.Context) {
startTime := time.Now()
// 1. Parse envelope
var envelope QueryEnvelope
if err := c.ShouldBindJSON(&envelope); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid envelope: " + err.Error()})
return
}
if envelope.QueryID == "" || envelope.CallbackURL == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "Missing required envelope fields"})
return
}
// 2. Run business logic
results, err := handler(envelope.Parameters, envelope)
if err != nil {
executionTimeMs := int(time.Since(startTime).Milliseconds())
PostCallback(envelope.CallbackURL, envelope.CallbackToken, CallbackPayload{
QueryID: envelope.QueryID,
Status: "failed",
ExecutionTimeMs: executionTimeMs,
Error: err.Error(),
})
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error(), "queryId": envelope.QueryID})
return
}
executionTimeMs := int(time.Since(startTime).Milliseconds())
// 3. Deliver results
delivery, err := DeliverResults(envelope, results, config)
if err != nil {
log.Printf("[delivery] Failed for query %s: %v", envelope.QueryID, err)
PostCallback(envelope.CallbackURL, envelope.CallbackToken, CallbackPayload{
QueryID: envelope.QueryID,
Status: "failed",
ExecutionTimeMs: executionTimeMs,
Error: err.Error(),
})
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error(), "queryId": envelope.QueryID})
return
}
// 4. Post callback
PostCallback(envelope.CallbackURL, envelope.CallbackToken, CallbackPayload{
QueryID: envelope.QueryID,
Status: "delivered",
ExecutionTimeMs: executionTimeMs,
Delivery: &DeliverySpec{Mechanism: delivery.Mechanism},
})
// 5. Return response
if delivery.Mechanism == "inline" {
c.JSON(http.StatusOK, gin.H{
"queryId": envelope.QueryID, "executionTimeMs": executionTimeMs, "results": results,
})
} else {
c.JSON(http.StatusOK, gin.H{
"queryId": envelope.QueryID, "executionTimeMs": executionTimeMs, "delivery": delivery,
})
}
}
}
// CompanySearch handles POST /api/search/companies using the middleware pattern.
func CompanySearch(config ProviderConfig) gin.HandlerFunc {
return gwMiddleware(config, func(params map[string]interface{}, envelope QueryEnvelope) (interface{}, error) {
name, _ := params["name"].(string)
country, _ := params["country"].(string)
industry, _ := params["industry"].(string)
results := SearchCompanies(name, country, industry)
return results, nil
})
}
Manual Pattern
For endpoints that need explicit control over each step:
handlers/sanctions.go
package handlers
import (
"log"
"net/http"
"time"
"github.com/gin-gonic/gin"
)
// SanctionsCheck handles POST /api/search/sanctions using the manual pattern.
func SanctionsCheck(config ProviderConfig) gin.HandlerFunc {
return func(c *gin.Context) {
startTime := time.Now()
// Step 1: Manually parse the envelope
var envelope QueryEnvelope
if err := c.ShouldBindJSON(&envelope); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid envelope: " + err.Error()})
return
}
if envelope.QueryID == "" || envelope.CallbackURL == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "Missing required envelope fields"})
return
}
// Step 2: Run business logic with custom validation
name, _ := envelope.Parameters["name"].(string)
if name == "" {
c.JSON(http.StatusBadRequest, gin.H{
"error": "parameters.name is required for sanctions checks",
"queryId": envelope.QueryID,
})
return
}
country, _ := envelope.Parameters["country"].(string)
results := CheckSanctions(name, country)
executionTimeMs := int(time.Since(startTime).Milliseconds())
// Step 3: Deliver results
delivery, err := DeliverResults(envelope, results, config)
if err != nil {
log.Printf("[delivery] Failed for query %s: %v", envelope.QueryID, err)
PostCallback(envelope.CallbackURL, envelope.CallbackToken, CallbackPayload{
QueryID: envelope.QueryID,
Status: "failed",
RecordCount: len(results),
ExecutionTimeMs: executionTimeMs,
Error: err.Error(),
})
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error(), "queryId": envelope.QueryID})
return
}
// Step 4: Explicitly post callback
PostCallback(envelope.CallbackURL, envelope.CallbackToken, CallbackPayload{
QueryID: envelope.QueryID,
Status: "delivered",
RecordCount: len(results),
ExecutionTimeMs: executionTimeMs,
Delivery: &DeliverySpec{Mechanism: delivery.Mechanism},
})
// Step 5: Return response
if delivery.Mechanism == "inline" {
c.JSON(http.StatusOK, gin.H{
"queryId": envelope.QueryID, "recordCount": len(results),
"executionTimeMs": executionTimeMs, "results": results,
})
} else {
c.JSON(http.StatusOK, gin.H{
"queryId": envelope.QueryID, "recordCount": len(results),
"executionTimeMs": executionTimeMs, "delivery": delivery,
})
}
}
}
Testing Locally
go run main.go
curl -X POST http://localhost:3002/api/search/sanctions \
-H "Content-Type: application/json" \
-H "X-GW-Api-Key: demo-api-key" \
-d '{
"queryId": "test-001",
"datasetId": "ds-sanctions",
"endpoint": "/api/search/sanctions",
"parameters": { "name": "Smith", "country": "RU" },
"delivery": { "mechanism": "inline" },
"callbackUrl": "http://localhost:8083/webhook",
"callbackToken": "test-secret"
}'