Add cog proxy endpoint (#615)

Author: oskarhurst

api/handlers/productfile_cog.go

@@ -0,0 +1,132 @@
+package handlers
+
+import (
+	"fmt"
+	"log"
+	"net/http"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/google/uuid"
+	"github.com/labstack/echo/v4"
+
+	"github.com/USACE/cumulus-api/api/middleware"
+	"github.com/USACE/cumulus-api/api/models"
+
+	_ "github.com/jackc/pgx/v4"
+	"github.com/jackc/pgx/v4/pgxpool"
+)
+
+// ListProductfilesCOG returns the productfiles for a product over a time range as
+// directly-readable COG proxy URLs (each backed by StreamProductfileCOG, which is
+// Range-capable so a GDAL /vsicurl/ client can read tiles). Same query params as
+// ListProductfiles: 'after' and 'before' (RFC3339).
+func ListProductfilesCOG(db *pgxpool.Pool) echo.HandlerFunc {
+	return func(c echo.Context) error {
+		id, err := uuid.Parse(c.Param("product_id"))
+		if err != nil {
+			return c.String(http.StatusBadRequest, "Malformed ID")
+		}
+		after := c.QueryParam("after")
+		before := c.QueryParam("before")
+		if after == "" || before == "" {
+			return c.String(
+				http.StatusBadRequest,
+				"Missing query parameter 'after' or 'before'",
+			)
+		}
+
+		ff, err := models.ListProductfiles(db, id, after, before)
+		if err != nil {
+			return c.String(http.StatusInternalServerError, err.Error())
+		}
+
+		cogs := make([]models.ProductfileCOG, 0, len(ff))
+		for _, f := range ff {
+			cogs = append(cogs, models.ProductfileCOG{
+				ID:       f.ID,
+				Datetime: f.Datetime,
+				Version:  f.Version,
+				CogURL:   fmt.Sprintf("/api/products/%s/cog/%s", id.String(), f.ID.String()),
+			})
+		}
+		return c.JSON(http.StatusOK, cogs)
+	}
+}
+
+// StreamProductfileCOG streams a productfile's COG object from S3, passing the
+// inbound HTTP Range header through to S3 and returning 206 Partial Content with
+// Content-Range/Accept-Ranges so a GDAL /vsicurl/ client can read tiles directly
+// (no full download). HEAD returns size + range support for /vsicurl/ probing.
+// Every request is authenticated (private route) and logged for metering.
+func StreamProductfileCOG(db *pgxpool.Pool, awsCfg *aws.Config) echo.HandlerFunc {
+	return func(c echo.Context) error {
+		pfID, err := uuid.Parse(c.Param("productfile_id"))
+		if err != nil {
+			return c.String(http.StatusBadRequest, "Malformed productfile ID")
+		}
+
+		obj, err := models.GetProductfileObject(db, pfID)
+		if err != nil {
+			return c.String(http.StatusNotFound, "Productfile not found")
+		}
+
+		client := s3.New(session.New(awsCfg))
+
+		// HEAD: metadata only (size + range support) for /vsicurl/ probing.
+		if c.Request().Method == http.MethodHead {
+			head, err := client.HeadObject(&s3.HeadObjectInput{Bucket: &obj.Bucket, Key: &obj.Key})
+			if err != nil {
+				return c.String(http.StatusInternalServerError, err.Error())
+			}
+			h := c.Response().Header()
+			h.Set("Accept-Ranges", "bytes")
+			if head.ContentLength != nil {
+				h.Set(echo.HeaderContentLength, fmt.Sprintf("%d", *head.ContentLength))
+			}
+			c.Response().WriteHeader(http.StatusOK)
+			return nil
+		}
+
+		in := &s3.GetObjectInput{Bucket: &obj.Bucket, Key: &obj.Key}
+		if rangeHeader := c.Request().Header.Get("Range"); rangeHeader != "" {
+			in.Range = aws.String(rangeHeader)
+		}
+
+		out, err := client.GetObject(in)
+		if err != nil {
+			return c.String(http.StatusInternalServerError, err.Error())
+		}
+		defer out.Body.Close()
+
+		h := c.Response().Header()
+		h.Set("Accept-Ranges", "bytes")
+		status := http.StatusOK
+		if out.ContentRange != nil {
+			h.Set("Content-Range", *out.ContentRange)
+			status = http.StatusPartialContent
+		}
+		if out.ContentLength != nil {
+			h.Set(echo.HeaderContentLength, fmt.Sprintf("%d", *out.ContentLength))
+		}
+
+		// Metering hook: who pulled which productfile and how many bytes. This is the
+		// natural place to enforce a future per-user rate limit / quota.
+		var bytesServed int64
+		if out.ContentLength != nil {
+			bytesServed = *out.ContentLength
+		}
+		logCOGAccess(c, pfID, bytesServed)
+
+		return c.Stream(status, "application/octet-stream", out.Body)
+	}
+}
+
+func logCOGAccess(c echo.Context, productfileID uuid.UUID, bytes int64) {
+	sub := "key-auth"
+	if ui, ok := c.Get("userInfo").(middleware.UserInfo); ok && ui.Sub != nil {
+		sub = ui.Sub.String()
+	}
+	log.Printf("cog-access sub=%s productfile=%s bytes=%d", sub, productfileID.String(), bytes)
+}

api/main.go

@@ -160,6 +160,10 @@ func main() {
 	// Additional Information About Products
 	public.GET("/products/:product_id/availability", handlers.GetProductAvailability(db))
 	public.GET("/products/:product_id/files", handlers.ListProductfiles(db))
+	// Direct, Range-capable COG access (authenticated + metered) for desktop clients
+	private.GET("/products/:product_id/cog-files", handlers.ListProductfilesCOG(db))
+	private.GET("/products/:product_id/cog/:productfile_id", handlers.StreamProductfileCOG(db, &awsCfg))
+	private.HEAD("/products/:product_id/cog/:productfile_id", handlers.StreamProductfileCOG(db, &awsCfg))
 
 	// Productfiles
 	private.POST("/productfiles", handlers.CreateProductfiles(db),

api/middleware/gzip.go

@@ -12,8 +12,10 @@ import (
 var GZIP = middleware.GzipWithConfig(middleware.GzipConfig{
 	Level: 5,
 	Skipper: func(c echo.Context) bool {
-		// Skip GZIP compression for routes starting with /features
-		// as compression messes with pg_featureserv
-		return strings.Contains(c.Request().URL.Path, "/features")
+		p := c.Request().URL.Path
+		// Skip GZIP compression for routes starting with /features as compression
+		// messes with pg_featureserv, and for COG byte-range streaming where
+		// compression breaks Range / Content-Length / Content-Range semantics.
+		return strings.Contains(p, "/features") || strings.Contains(p, "/cog/")
 	},
 })

api/models/productfiles.go

@@ -26,6 +26,39 @@ type ProductfileAvailability struct {
 	IsAvailable bool      `json:"is_available"`
 }
 
+// ProductfileCOG is a productfile exposed as a directly-readable, Range-capable
+// COG proxy URL (served by StreamProductfileCOG) instead of a raw S3 key.
+type ProductfileCOG struct {
+	ID       uuid.UUID  `json:"id"`
+	Datetime time.Time  `json:"datetime"`
+	Version  *time.Time `json:"version"`
+	CogURL   string     `json:"cog_url"`
+}
+
+// ProductfileObject is the S3 bucket + key backing a single productfile.
+type ProductfileObject struct {
+	Key    string `json:"key" db:"key"`
+	Bucket string `json:"bucket" db:"bucket"`
+}
+
+// GetProductfileObject returns the S3 bucket and key for a productfile id.
+// The bucket mirrors the download view: the 'write_to_bucket' config value; the
+// key is the productfile.file column.
+func GetProductfileObject(db *pgxpool.Pool, ID uuid.UUID) (*ProductfileObject, error) {
+	var obj ProductfileObject
+	if err := pgxscan.Get(
+		context.Background(), db, &obj,
+		`SELECT f.file AS key,
+		        (SELECT config.config_value FROM config WHERE config.config_name::text = 'write_to_bucket'::text) AS bucket
+		 FROM productfile f
+		 WHERE f.id = $1`,
+		ID,
+	); err != nil {
+		return nil, err
+	}
+	return &obj, nil
+}
+
 // ListProductfiles returns array of productfiles
 func ListProductfiles(db *pgxpool.Pool, ID uuid.UUID, after string, before string) ([]Productfile, error) {
 	ff := make([]Productfile, 0)