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 }