json-adapter/main.go

package main

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"flag"
	"io"
	"log"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"strings"
	"text/template"
	"time"

	_ "github.com/glebarez/sqlite"
)

var httpc = &http.Client{
	Timeout: time.Minute,
	Transport: &http.Transport{
		DisableKeepAlives: true,
	},
}

type Handler struct {
	tmpl        *template.Template
	target      *url.URL
	idempotency *sql.DB
}

func (h Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if err := h.serveHTTP(w, r); err != nil {
		log.Println("!", err)
		http.Error(w, err.Error(), http.StatusBadRequest)
	}
}

func (h Handler) serveHTTP(w http.ResponseWriter, r *http.Request) error {
	b, err := adapt(r.Body, h.tmpl)
	if err != nil {
		return err
	}

	var duplicate int
	if err := h.idempotency.QueryRowContext(r.Context(), `SELECT 1 FROM payloads WHERE payload=$1;`, b).Scan(&duplicate); err != sql.ErrNoRows && err != nil {
		log.Println("!", err)
	} else if duplicate > 0 {
		log.Println("+")
		return nil
	}

	if err := h.prune(time.Now().Add(-1 * time.Hour * 24 * 7)); err != nil {
		log.Println("!", err)
	}

	resp, err := proxy(h.target, r, io.NopCloser(bytes.NewReader(b)))
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if err := forward(w, resp); err != nil {
		return err
	}

	if _, err := h.idempotency.ExecContext(r.Context(), `INSERT INTO payloads (ts, payload) VALUES ($1, $2);`, time.Now(), b); err != nil {
		log.Println("!", err)
	}

	return nil
}

func main() {
	fs := flag.NewFlagSet(os.Args[0], flag.ContinueOnError)
	p := fs.Int("p", 28080, "port")
	t := fs.String("t", "{{ . }}", "template")
	y := fs.String("y", "http://localhost:41912", "target")
	db := fs.String("db", "/tmp/json-adapter.db", "sqlite db for idempotency")
	if err := fs.Parse(os.Args[1:]); err != nil {
		panic(err)
	}
	if err := run(*p, *t, *y, *db); err != nil {
		panic(err)
	}
}

func run(p int, t string, y string, db string) error {
	idempotency, err := sql.Open("sqlite", db)
	if err != nil {
		return err
	}
	defer idempotency.Close()
	if err := idempotency.PingContext(context.Background()); err != nil {
		return err
	} else if _, err := idempotency.ExecContext(context.Background(), `CREATE TABLE IF NOT EXISTS payloads (payload TEXT, ts TIMESTAMP NOT NULL)`); err != nil {
		return err
	}

	tmpl, err := template.New("").Parse(t)
	if err != nil {
		return err
	}

	u, err := url.Parse(y)
	if err != nil {
		return err
	}

	h := Handler{tmpl: tmpl, target: u, idempotency: idempotency}
	log.Println("listening on", p)
	return http.ListenAndServe(":"+strconv.Itoa(p), h)
}

func adapt(r io.Reader, tmpl *template.Template) ([]byte, error) {
	b, _ := io.ReadAll(io.LimitReader(r, 1024*1024))
	var v interface{}
	buff := bytes.NewBuffer(nil)
	if len(b) == 0 {
	} else if err := json.Unmarshal(b, &v); err != nil {
		return nil, err
	}
	if err := tmpl.Execute(buff, v); err != nil {
		return nil, err
	}
	log.Printf("%s => %s", b, buff.Bytes())
	return buff.Bytes(), nil
}

func proxy(u *url.URL, r *http.Request, rc io.ReadCloser) (*http.Response, error) {
	req, err := http.NewRequest(r.Method, u.String(), rc)
	if err != nil {
		return nil, err
	}

	req = req.WithContext(r.Context())
	for k, v := range r.Header {
		switch strings.ToLower(k) {
		case "content-length":
		default:
			req.Header.Set(k, v[0])
			for _, v := range v[1:] {
				req.Header.Add(k, v)
			}
		}
	}

	return httpc.Do(req)
}

func forward(w http.ResponseWriter, resp *http.Response) error {
	for k, v := range resp.Header {
		w.Header().Set(k, v[0])
		for _, v := range v[1:] {
			w.Header().Add(k, v)
		}
	}
	w.WriteHeader(resp.StatusCode)
	io.Copy(w, resp.Body)
	return nil
}

func (h Handler) prune(ts time.Time) error {
	result, err := h.idempotency.ExecContext(context.Background(), `DELETE FROM payloads WHERE ts < $1;`, ts)
	if err != nil {
		return err
	}

	rows, _ := result.RowsAffected()
	if rows > 1 {
		log.Println("-", rows)
	}

	return nil
}