Atomic phases recorded as idempotency keys

.dockerignore

@@ -2,4 +2,11 @@
 bin
 .github
 Dockerfile
-*.env
+var
+data
+scripts
+hermes.json
+example.json
+*.env
+*.yml
+*.toml

Dockerfile

@@ -1,12 +1,13 @@
-ARG GOLANG_VERSION=1.14
+ARG GOLANG_VERSION=1.18
 ARG TINI_VERSION=v0.19.0
 FROM golang:${GOLANG_VERSION}-alpine as build
+RUN apk add build-base
 
 WORKDIR $GOPATH/src/github.com/rugwirobaker/hermes
-COPY go.mod go.sum  ./
+COPY go.mod go.sum ./
 RUN GO111MODULE=on GOPROXY="https://proxy.golang.org" go mod download
 COPY . .
-RUN GO111MODULE=on CGO_ENABLED=0 go build -o /bin/hermes ./cmd/hermes
+RUN GO111MODULE=on CGO_ENABLED=1 go build -o /bin/hermes ./cmd/hermes
 
 FROM scratch
 WORKDIR /
@@ -15,4 +16,4 @@ EXPOSE 8080
 COPY --from=build /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
 COPY --from=build /bin/hermes /bin/hermes
 
-ENTRYPOINT ["/bin/hermes"]
+ENTRYPOINT ["/bin/hermes"]

api/api.go

@@ -17,13 +17,13 @@ type Server struct {
 	events   hermes.Pubsub
 	service  hermes.SendService
 	store    hermes.Store
-	cache    mw.Cache
 	provider trace.TracerProvider
+	keys     hermes.IdempotencyKeyStore
 }
 
 // New api Server instance
-func New(svc hermes.SendService, events hermes.Pubsub, store hermes.Store, cache mw.Cache, provider trace.TracerProvider) *Server {
-	return &Server{service: svc, events: events, store: store, cache: cache, provider: provider}
+func New(svc hermes.SendService, events hermes.Pubsub, store hermes.Store, keys hermes.IdempotencyKeyStore, provider trace.TracerProvider) *Server {
+	return &Server{service: svc, events: events, store: store, keys: keys, provider: provider}
 }
 
 // Handler returns an http.Handler
@@ -32,10 +32,10 @@ func (s Server) Handler() http.Handler {
 
 	r.Use(middleware.RequestID)
 	r.Use(middleware.RealIP)
+	r.Use(mw.WithRequestID)
 	r.Use(middleware.Logger)
 	r.Use(otelchi.Middleware("hermes", otelchi.WithTracerProvider(s.provider)))
 	r.Use(mw.Idempotency)
-	r.Use(mw.Caching(s.cache))
 	r.Use(middleware.Recoverer)
 
 	r.Get("/", func(w http.ResponseWriter, r *http.Request) {

api/middleware/caching.go

@@ -32,7 +32,7 @@ func Caching(cache Cache) func(http.Handler) http.Handler {
 			}
 
 			if entry, ok := cache.Get(key); ok && entry.Path == r.URL.Path {
-				log.Printf("[Caching] Cache hit for key: %s", key)
+				log.Printf("[caching] Cache hit for key: %s", key)
 
 				w.WriteHeader(entry.Code)
 				for k, v := range entry.Headers {
@@ -43,7 +43,7 @@ func Caching(cache Cache) func(http.Handler) http.Handler {
 				return
 			}
 
-			log.Printf("[Caching] Cache miss for key: %s", key)
+			log.Printf("[caching] Cache miss for key: %s", key)
 
 			rec := httptest.NewRecorder()
 			next.ServeHTTP(rec, r)

api/middleware/request_id.go

@@ -0,0 +1,32 @@
+package middleware
+
+import (
+	"net/http"
+
+	"github.com/google/uuid"
+	"github.com/rugwirobaker/hermes/api/request"
+)
+
+// WithRequestID ensures a request id is in the
+// request context by either the incoming header
+// or creating a new one
+func WithRequestID(h http.Handler) http.Handler {
+	fn := func(w http.ResponseWriter, r *http.Request) {
+		requestID := r.Header.Get("X-Request-ID")
+
+		if requestID == "" {
+			requestID = r.Header.Get("Fly-Request-Id")
+		}
+
+		if requestID == "" {
+			requestID = uuid.New().String()
+		}
+
+		ctx := request.WithRequestID(r.Context(), requestID)
+
+		r = r.WithContext(ctx)
+
+		h.ServeHTTP(w, r)
+	}
+	return http.HandlerFunc(fn)
+}

cmd/hermes/main.go

@@ -13,7 +13,6 @@ import (
 	"github.com/go-chi/chi/v5"
 	"github.com/rugwirobaker/hermes"
 	"github.com/rugwirobaker/hermes/api"
-	"github.com/rugwirobaker/hermes/api/middleware"
 	"github.com/rugwirobaker/hermes/sqlite"
 	"github.com/rugwirobaker/hermes/tracing"
 	"go.opentelemetry.io/otel"
@@ -77,10 +76,10 @@ func main() {
 	events := hermes.NewPubsub()
 	defer events.Close()
 
-	cache := middleware.NewMemoryCache()
+	keys := hermes.NewIdempotencyKeyStore(db)
 
 	log.Println("initialized hermes api")
-	api := api.New(service, events, store, cache, provider)
+	api := api.New(service, events, store, keys, provider)
 	mux := chi.NewMux()
 	mux.Mount("/api", api.Handler())
 

error.go

@@ -0,0 +1,5 @@
+package hermes
+
+import "fmt"
+
+var ErrNotFound = fmt.Errorf("")

go.mod

@@ -11,8 +11,7 @@ require (
 	github.com/google/go-cmp v0.5.8
 	github.com/google/uuid v1.3.0
 	github.com/kr/pretty v0.3.0 // indirect
-	github.com/lib/pq v1.10.7
-	github.com/mattn/go-sqlite3 v1.14.15 // indirect
+	github.com/mattn/go-sqlite3 v1.14.15
 	github.com/nhatthm/otelsql v0.4.0
 	github.com/quarksgroup/sms-client v1.0.0
 	github.com/riandyrn/otelchi v0.5.0

idempotency.go

@@ -0,0 +1,224 @@
+package hermes
+
+import (
+	"context"
+	"time"
+
+	"github.com/rugwirobaker/hermes/observ"
+	"github.com/rugwirobaker/hermes/sqlite"
+)
+
+type RecoveryPoint string
+
+const (
+	RecoveryPointStart    RecoveryPoint = "started"
+	RecoveryPointFinished RecoveryPoint = "finished"
+)
+
+func (r RecoveryPoint) String() string {
+	return string(r)
+}
+
+type IdempotencyKey struct {
+	ID        int       `json:"id"`
+	Key       string    `json:"key"`
+	CreatedAt time.Time `json:"created_at"`
+	UpdatedAt time.Time `json:"updated_at"`
+	LockedAt  time.Time `json:"locked_at"`
+
+	RequestMethod string `json:"request_method"`
+	RequestPath   string `json:"request_path"`
+	RequestParams map[string]string
+	RequestBody   []byte `json:"request_body"`
+	RequestHeader map[string][]string
+
+	ResponseCode int    `json:"response_code"`
+	ResponseBody []byte `json:"response_body"`
+
+	Recovery RecoveryPoint `json:"recovery"`
+}
+
+type IdempotencyKeyStore interface {
+	// Insert idempotency key
+	Create(context.Context, *IdempotencyKey) (*IdempotencyKey, error)
+	// Key returns a key by key
+	Key(context.Context, string) (*IdempotencyKey, error)
+	// Update a key
+	Update(context.Context, *IdempotencyKey) (*IdempotencyKey, error)
+}
+
+type idempotencyKeyStore struct {
+	db *sqlite.DB
+}
+
+func NewIdempotencyKeyStore(db *sqlite.DB) IdempotencyKeyStore {
+	return &idempotencyKeyStore{
+		db: db,
+	}
+}
+
+func (s *idempotencyKeyStore) Create(ctx context.Context, in *IdempotencyKey) (*IdempotencyKey, error) {
+	const op = "idempotencyKeyStore.Insert"
+
+	ctx, span := observ.StartSpan(ctx, op)
+	defer span.End()
+
+	tx, err := s.db.BeginTx(ctx, sqlite.TxOptions(false))
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+
+	res, err := tx.ExecContext(ctx,
+		`INSERT INTO idempotency_keys (
+			key, 
+			recovery
+		) VALUES (?, ?)`, in.Key, in.Recovery)
+	if err != nil {
+		return nil, err
+	}
+
+	id, err := res.LastInsertId()
+	if err != nil {
+		return nil, err
+	}
+
+	in.ID = int(id)
+	in.CreatedAt = time.Now()
+	in.UpdatedAt = time.Now()
+
+	return in, nil
+}
+
+func (s *idempotencyKeyStore) Key(ctx context.Context, key string) (*IdempotencyKey, error) {
+	const op = "idempotencyKeyStore.Key"
+
+	ctx, span := observ.StartSpan(ctx, op)
+	defer span.End()
+
+	tx, err := s.db.BeginTx(ctx, sqlite.TxOptions(false))
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+
+	// scan
+
+	var u IdempotencyKey
+	err = tx.QueryRowContext(ctx, "SELECT * FROM idempotency_keys WHERE key = ?", key).Scan(&u.ID, &u.Key, &u.CreatedAt, &u.UpdatedAt, &u.LockedAt, &u.RequestMethod, &u.RequestPath, &u.RequestParams, &u.RequestBody, &u.RequestHeader, &u.ResponseCode, &u.ResponseBody, &u.Recovery)
+	if err != nil {
+		return nil, err
+	}
+
+	return &u, nil
+}
+
+func (s *idempotencyKeyStore) Update(ctx context.Context, in *IdempotencyKey) (*IdempotencyKey, error) {
+	const op = "idempotencyKeyStore.Update"
+
+	ctx, span := observ.StartSpan(ctx, op)
+	defer span.End()
+
+	tx, err := s.db.BeginTx(ctx, sqlite.TxOptions(false))
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+
+	// scan
+
+	res, err := tx.ExecContext(
+		ctx, `
+		UPDATE 
+			idempotency_keys 
+		SET 
+			request_method = ?, 
+			request_path = ?, 
+			request_params = ?, 
+			request_body = ?, 
+			request_header = ?, 
+			response_code = ?, 
+			response_body = ?, 
+			recovery_point = ? 
+		WHERE 
+			key = ?
+	`,
+		in.RequestMethod,
+		in.RequestPath,
+		in.RequestParams,
+		in.RequestBody,
+		in.RequestHeader,
+		in.ResponseCode,
+		in.ResponseBody,
+		in.Recovery,
+		in.Key,
+	)
+
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = res.RowsAffected()
+	if err != nil {
+		return nil, err
+	}
+
+	in.UpdatedAt = time.Now()
+
+	return in, nil
+}
+
+func (s *idempotencyKeyStore) Lock(ctx context.Context, key string) (*IdempotencyKey, error) {
+	const op = "idempotencyKeyStore.Lock"
+
+	ctx, span := observ.StartSpan(ctx, op)
+	defer span.End()
+
+	tx, err := s.db.BeginTx(ctx, sqlite.TxOptions(false))
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+
+	// scan
+
+	var u IdempotencyKey
+
+	var row = tx.QueryRowContext(ctx, `
+	SELECT 
+		id,
+		key,
+		created_at,
+		updated_at,
+		locked_at,
+		request_method,
+		request_path,
+		request_params,
+		request_body,
+		request_header,
+		response_code,
+		response_body,
+		recovery_point
+	FROM 
+		idempotency_keys 
+	WHERE key = ?`, key)
+
+	err = row.Scan(&u.ID,
+		&u.Key,
+		&u.CreatedAt,
+		&u.UpdatedAt,
+		&u.LockedAt,
+		&u.RequestMethod,
+		&u.RequestPath,
+		&u.RequestParams,
+		&u.RequestBody,
+		&u.RequestHeader,
+		&u.ResponseCode,
+		&u.ResponseBody,
+		&u.Recovery,
+	)
+	if err != nil {
+		return nil, err
+	}
+	return &u, nil
+}

mock/mock.go

@@ -1,3 +1,3 @@
 package mock
 
-//go:generate mockgen -package=mock -destination=mock_gen.go github.com/rugwirobaker/hermes SendService,Pubsub,Store
+//go:generate mockgen -package=mock -destination=mock_gen.go github.com/rugwirobaker/hermes SendService,Pubsub,Store,IdempotencyKeyStore