retention

Data-retention policy resolution for a request from HTTP headers, Kafka headers, or a JWT claim. Standard library only — no Gin, no Sarama.

Quickstart

package main

import (
    "net/http"

    "github.com/corticph/gocomo/retention/retention"
)

func main() {
    handler := retention.Middleware(readBearerToken)(http.HandlerFunc(myHandler))
    _ = http.ListenAndServe(":8080", handler)
}

func readBearerToken(r *http.Request) string {
    // Return the raw Authorization bearer token, already authenticated upstream.
    return ""
}

func myHandler(w http.ResponseWriter, r *http.Request) {
    ret := retention.FromContext(r.Context())
    if !ret.ShouldRetain() {
        // Discard the data.
    }
}

Behavior

Resolve directly when there is no HTTP middleware:

ret, err := retention.FromHeaders(r.Header, rawJWT)

The function considers two sources. HTTP headers take priority; the JWT claim applies when the header is absent.

Invalid policy, missing or non-numeric minutes, or negative minutes return an error; retention.ErrMinutesNotProvided is returned when policy is minutes but the minutes header is empty. The middleware answers invalid requests with HTTP 400.

Model

type Retention struct {
    Policy  Policy // none | retain | minutes
    Minutes *int   // retention duration, nil when indefinite or none
}

Context

ctx = retention.WithContext(ctx, ret)
ret := retention.FromContext(ctx) // nil when unset

Kafka headers

ret, err := retention.FromKafkaHeaders(record.Headers)

FromKafkaHeaders accepts any slice of values satisfying:

type KafkaHeader[K, V ~[]byte] interface {
    GetKey() K
    GetValue() V
}

*sarama.RecordHeader satisfies this structurally, so no Sarama dependency is imported. The policy-header key match is case-insensitive; the first match wins and absence means retain.

Gin compatibility

*gin.Context implements Header(key, value string) and thus satisfies:

type HeaderSetter interface {
    Header(key, value string)
}

Acknowledge an explicit policy on a Gin context without this package importing Gin:

func handler(c *gin.Context) {
    ret, err := retention.FromHeaders(c.Request.Header, token(c))
    if err != nil {
        c.AbortWithStatus(http.StatusBadRequest)
        return
    }
    ret.SendAckHeader(c) // sets X-Corti-Retention-Policy: acknowledged
}

The acknowledged acknowledgement is sent only when the client explicitly set the policy header.