package merp import "encoding/json" import "fmt" import "net/http" import "os" import "regexp" import "strconv" import "time" import "sync" import "github.com/didip/tollbooth" import "github.com/didip/tollbooth_gin" import "github.com/gin-gonic/gin" import "github.com/rs/zerolog/log" import "github.com/dn365/gin-zerolog" import "github.com/thoas/stats" import "github.com/astaxie/beego/orm" import _ "github.com/lib/pq" //revive:disable-line var thingRegex = regexp.MustCompile(`^[a-zA-Z0-9\_\-]+$`) type MerpTopic string const ( // Shifts KiB = 10 MiB = 20 GiB = 30 ) // Server is the central structure of the HTTP API server. type Server struct { db orm.Ormer debug bool gin *gin.Engine port uint server *http.Server stats *stats.Stats listeners map[MerpTopic][]*EventListener ll *sync.Mutex // listeners [map] lock } // NewServer returns a Server, so that you can run the API. func NewServer() *Server { ms := new(Server) ms.ll = new(sync.Mutex) ms.listeners = make(map[MerpTopic][]*EventListener) ms.init() return ms } func (ms *Server) addListener(topic MerpTopic, l *EventListener) { // FIXME(sneak) } func (ms *Server) init() { if os.Getenv("DEBUG") != "" { ms.debug = true } ms.port = 8080 var s uint64 var err error if os.Getenv("PORT") != "" { if s, err = strconv.ParseUint(os.Getenv("PORT"), 10, 64); err != nil { panic("invalid PORT in environment") } ms.port = uint(s) } ms.connectDB() gin.DefaultWriter = log.With().Str("component", "gin").Logger() gin.DefaultErrorWriter = log.With().Str("component", "gin").Logger() ms.setupRoutes() ms.server = &http.Server{ Addr: fmt.Sprintf(":%d", ms.port), Handler: ms.gin, ReadTimeout: 10 * time.Second, WriteTimeout: 60 * time.Second, MaxHeaderBytes: 5 << KiB, } } func (ms *Server) connectDB() { ms.db = GetDB() } // ServeForever causes merp to serve http forever func (ms *Server) ServeForever() { err := ms.server.ListenAndServe() if err != nil { panic(err) } } func (ms *Server) healthCheckHandler() http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { result := gin.H{ "status": "ok", "now": time.Now().UTC().Format(time.RFC3339), } json, err := json.Marshal(result) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } w.Header().Set("Content-Type", "application/json") w.Write(json) } } func (ms *Server) statsHandler() http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") stats := ms.stats.Data() b, _ := json.Marshal(stats) w.Write(b) } } func (ms *Server) setupRoutes() { if !ms.debug { gin.SetMode(gin.ReleaseMode) } limiter := tollbooth.NewLimiter(5, nil) ms.stats = stats.New() // empty router r := gin.New() // wrap panics: r.Use(gin.Recovery()) // attach logger middleware r.Use(ginzerolog.Logger("gin")) r.Use(func(c *gin.Context) { beginning, recorder := ms.stats.Begin(c.Writer) c.Next() ms.stats.End(beginning, stats.WithRecorder(recorder)) }) //FIXME(sneak) use a http.MaxBytesReader middleware to limit request size r.GET("/.well-known/healthcheck.json", gin.WrapF(ms.healthCheckHandler())) r.GET("/admin/healthcheck.json", gin.WrapF(ms.healthCheckHandler())) r.GET("/admin/stats.json", gin.WrapF(ms.statsHandler())) r.GET("/admin/other.json", gin.WrapF(ms.statsHandler())) r.GET("/merp/for/:thing", tollbooth_gin.LimitHandler(limiter), ms.handleNewMerp()) r.GET("/listen/for/merps/from/:thing", tollbooth_gin.LimitHandler(limiter), ms.listenForMerps()) r.GET("/get/latest/merp/for/:thing", tollbooth_gin.LimitHandler(limiter), ms.getLatestMerps()) r.GET("/get/latest/merps", tollbooth_gin.LimitHandler(limiter), ms.getLatestMerps()) r.GET("/get/merps/for/:thing", tollbooth_gin.LimitHandler(limiter), ms.getLatestMerps()) r.NoRoute(func(c *gin.Context) { c.JSON(404, gin.H{"message": "not found"}) }) ms.gin = r } type EventListener struct { thing string notifications chan struct{} } func NewEventListener() *EventListener { el := new(EventListener) el.init() return el } func (el *EventListener) init() { el.notifications = make(chan struct{}) // NOOP for now }