server: implement GET /v2/job-types

Also start working on passing around a *newmodels.Queries instead of
hardcoding `*newmodels.DB` everywhere.

Add a new `responses` package that can help us get rid of `auto_id`
in the database response, which is something that we don't want to
expose.

Updates #8.

db/queries/jobs.sql

@@ -8,12 +8,13 @@ VALUES ($1, $2, $3, $4)
 RETURNING *;
 
 -- name: GetJob :one
-SELECT *
-FROM jobs
+SELECT * FROM jobs
 WHERE name = $1;
 
 -- name: GetAllJobs :many
-SELECT * FROM jobs WHERE name != 'meta.shutdown';
+SELECT * FROM jobs
+WHERE name != 'meta.shutdown'
+ORDER BY auto_id DESC;
 
 -- name: DeleteAllJobs :execrows
 DELETE FROM jobs;

dbtohttp/dbtohttp.go

@@ -0,0 +1,24 @@
+package dbtohttp
+
+import (
+	"github.com/kevinburke/rickover/newmodels"
+	"github.com/kevinburke/rickover/responses"
+)
+
+func Job(j newmodels.Job) responses.Job {
+	return responses.Job{
+		Name:             j.Name,
+		DeliveryStrategy: string(j.DeliveryStrategy),
+		Attempts:         j.Attempts,
+		Concurrency:      j.Concurrency,
+		CreatedAt:        j.CreatedAt,
+	}
+}
+
+func Jobs(js []newmodels.Job) []responses.Job {
+	resp := make([]responses.Job, len(js))
+	for i := range js {
+		resp[i] = Job(js[i])
+	}
+	return resp
+}

models/queued_jobs/queued_jobs.go

@@ -39,6 +39,8 @@ var StuckJobLimit = 100
 // Enqueue creates a new queued job with the given ID and fields. A
 // sql.ErrNoRows will be returned if the `name` does not exist in the jobs
 // table. Otherwise the QueuedJob will be returned.
+//
+// Deprecated: use services.Enqueue instead.
 func Enqueue(params newmodels.EnqueueJobParams) (*newmodels.QueuedJob, error) {
 	qj, err := newmodels.DB.EnqueueJob(context.TODO(), params)
 	if err != nil {

newmodels/setup.go

@@ -7,14 +7,18 @@ import (
 	"github.com/kevinburke/rickover/models/db"
 )
 
-var DB *Queries
+// DefaultServer relies on this being initialized (and not overwritten)
+var DB = new(Queries)
 
 func Setup(ctx context.Context) error {
 	if !db.Connected() {
 		return errors.New("newmodels: no database connection, bailing")
 	}
 
-	var err error
-	DB, err = Prepare(ctx, db.Conn)
-	return err
+	qs, err := Prepare(ctx, db.Conn)
+	if err != nil {
+		return err
+	}
+	*DB = *qs
+	return nil
 }

responses/responses.go

@@ -0,0 +1,11 @@
+package responses
+
+import "time"
+
+type Job struct {
+	Name             string    `json:"name"`
+	DeliveryStrategy string    `json:"delivery_strategy"`
+	Attempts         int16     `json:"attempts"`
+	Concurrency      int16     `json:"concurrency"`
+	CreatedAt        time.Time `json:"created_at"`
+}

server/enqueue_job_test.go

@@ -62,6 +62,8 @@ func Test401UnknownPassword(t *testing.T) {
 
 func Test400NoBody(t *testing.T) {
 	t.Parallel()
+	test.SetUp(t)
+	defer test.TearDown(t)
 	w := httptest.NewRecorder()
 	ejr := &EnqueueJobRequest{
 		Data: empty,
@@ -84,6 +86,8 @@ func Test400NoBody(t *testing.T) {
 
 func Test400EmptyBody(t *testing.T) {
 	t.Parallel()
+	test.SetUp(t)
+	defer test.TearDown(t)
 	w := httptest.NewRecorder()
 	var v interface{}
 	err := json.Unmarshal([]byte("{}"), &v)
@@ -92,7 +96,7 @@ func Test400EmptyBody(t *testing.T) {
 	json.NewEncoder(b).Encode(v)
 	ssa, server := newSSAServer()
 	ssa.AddUser("test", "password")
-	req, _ := http.NewRequest("PUT", "/v1/jobs/echo/job_f17373a6-2cd7-4010-afba-eebc6dc6f9ab", b)
+	req := httptest.NewRequest("PUT", "/v1/jobs/echo/job_f17373a6-2cd7-4010-afba-eebc6dc6f9ab", b)
 	req.SetBasicAuth("test", "password")
 	server.ServeHTTP(w, req)
 	test.AssertEquals(t, w.Code, http.StatusBadRequest)
@@ -106,6 +110,8 @@ func Test400EmptyBody(t *testing.T) {
 
 func Test400InvalidUUID(t *testing.T) {
 	t.Parallel()
+	test.SetUp(t)
+	defer test.TearDown(t)
 	w := httptest.NewRecorder()
 	ejr := &EnqueueJobRequest{
 		Data: empty,
@@ -125,6 +131,8 @@ func Test400InvalidUUID(t *testing.T) {
 
 func Test400WrongPrefix(t *testing.T) {
 	t.Parallel()
+	test.SetUp(t)
+	defer test.TearDown(t)
 	w := httptest.NewRecorder()
 	ejr := &EnqueueJobRequest{
 		Data: empty,
@@ -139,6 +147,8 @@ func Test400WrongPrefix(t *testing.T) {
 
 func Test413TooLargeJSON(t *testing.T) {
 	t.Parallel()
+	test.SetUp(t)
+	defer test.TearDown(t)
 	w := httptest.NewRecorder()
 	// 4 bytes per record - the value and the quotes around it.
 	var bigarr [100 * 256]string

server/regex_handler.go

@@ -46,6 +46,10 @@ func (h *RegexpHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	for _, route := range h.routes {
 		if route.pattern.MatchString(r.URL.Path) {
 			upperMethod := strings.ToUpper(r.Method)
+			if route.methods == nil && upperMethod != "OPTIONS" {
+				route.handler.ServeHTTP(w, r)
+				return
+			}
 			for _, method := range route.methods {
 				if strings.ToUpper(method) == upperMethod {
 					route.handler.ServeHTTP(w, r)

server/server.go

@@ -29,6 +29,7 @@ import (
 	"github.com/kevinburke/rickover/models/jobs"
 	"github.com/kevinburke/rickover/models/queued_jobs"
 	"github.com/kevinburke/rickover/newmodels"
+	"github.com/kevinburke/rickover/services"
 	"github.com/kevinburke/rickover/setup"
 	"github.com/lib/pq"
 )
@@ -93,28 +94,34 @@ func New(ctx context.Context, cfg Config) (http.Handler, error) {
 	if err := setup.DB(ctx, cfg.Connector, cfg.NumConns); err != nil {
 		return nil, err
 	}
-	s := Get(cfg)
-	return s, nil
+	return Get(cfg), nil
 }
 
 // Get returns a http.Handler with all routes initialized using the given
-// Authorizer.
+// Authorizer. Get assumes that setup.DB has been called; to call both use
+// New.
 func Get(c Config) http.Handler {
 	a := c.Auth
 	if a == nil {
 		panic("server: cannot call Get() with nil Authorizer")
 	}
 	h := new(RegexpHandler)
 
+	db := newmodels.DB
 	useMetaShutdown := !c.DisableMetaShutdown
-	h.Handler(jobsRoute, []string{"POST"}, authHandler(createJob(useMetaShutdown), a))
-	h.Handler(getJobRoute, []string{"GET"}, authHandler(handleJobRoute(useMetaShutdown), a))
-	h.Handler(getJobTypeRoute, []string{"GET"}, authHandler(getJobType(), a))
+
+	// v1
+	h.Handler(jobsRoute, []string{"POST"}, authHandler(createJob(db, useMetaShutdown), a))
+	h.Handler(getJobRoute, []string{"GET"}, authHandler(handleJobRoute(db, useMetaShutdown), a))
+	h.Handler(getJobTypeRoute, []string{"GET"}, authHandler(getJobType(db), a))
 
 	h.Handler(replayRoute, []string{"POST"}, authHandler(replayHandler(), a))
 
-	h.Handler(jobIdRoute, []string{"GET", "POST", "PUT"}, authHandler(handleJobRoute(useMetaShutdown), a))
-	h.Handler(archivedJobsRoute, []string{"GET"}, authHandler(listArchivedJobs(), a))
+	h.Handler(jobIdRoute, []string{"GET", "POST", "PUT"}, authHandler(handleJobRoute(db, useMetaShutdown), a))
+	h.Handler(archivedJobsRoute, []string{"GET"}, authHandler(listArchivedJobs(db), a))
+
+	// v2
+	h.Handler(regexp.MustCompile(`^/v2/(.+)$`), nil, authHandler(V2(db), a))
 
 	h.Handler(regexp.MustCompile("^/debug/pprof$"), []string{"GET"}, authHandler(http.HandlerFunc(pprof.Index), a))
 	h.Handler(regexp.MustCompile("^/debug/pprof/cmdline$"), []string{"GET"}, authHandler(http.HandlerFunc(pprof.Cmdline), a))
@@ -235,10 +242,10 @@ type CreateJobRequest struct {
 // GET /v1/jobs/:jobName
 //
 // Get a job type by name. Returns a models.Job or an error
-func getJobType() http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+func getJobType(db *newmodels.Queries) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
 		jobName := getJobTypeRoute.FindStringSubmatch(r.URL.Path)[1]
-		job, err := jobs.Get(r.Context(), jobName)
+		job, err := db.GetJob(r.Context(), jobName)
 		if err != nil {
 			if err == sql.ErrNoRows {
 				notFound(w, new404(r))
@@ -248,7 +255,7 @@ func getJobType() http.Handler {
 			return
 		}
 		respond(Logger, w, r, job)
-	})
+	}
 }
 
 var Logger = handlers.Logger
@@ -257,8 +264,8 @@ var Logger = handlers.Logger
 //
 // createJob returns a http.HandlerFunc that responds to job creation requests
 // using the given authorizer interface.
-func createJob(useMetaShutdown bool) http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+func createJob(db *newmodels.Queries, useMetaShutdown bool) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
 		if r.Body == nil {
 			rest.BadRequest(w, r, createEmptyErr("name", r.URL.Path))
 			return
@@ -326,7 +333,9 @@ func createJob(useMetaShutdown bool) http.Handler {
 			Attempts:         jr.Attempts,
 		}
 		start := time.Now()
-		job, err := jobs.Create(newmodels.CreateJobParams{
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		job, err := db.CreateJob(ctx, newmodels.CreateJobParams{
 			Name:             jobData.Name,
 			DeliveryStrategy: jobData.DeliveryStrategy,
 			Concurrency:      jobData.Concurrency,
@@ -354,12 +363,12 @@ func createJob(useMetaShutdown bool) http.Handler {
 				// a shutdown job, which will instruct the dequeuer to shut down.
 				ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
 				defer cancel()
-				if err := createMetaShutdownJob(ctx); err != nil {
+				if err := createMetaShutdownJob(ctx, db); err != nil {
 					return
 				}
 				id := types.GenerateUUID("job_")
 				runAfter := time.Now().Add(-1 * 5 * time.Second)
-				queued_jobs.Enqueue(newmodels.EnqueueJobParams{
+				services.Enqueue(ctx, db, newmodels.EnqueueJobParams{
 					ID: id, Name: "meta.shutdown", RunAfter: runAfter,
 					ExpiresAt: types.NullTime{
 						Valid: true,
@@ -371,7 +380,7 @@ func createJob(useMetaShutdown bool) http.Handler {
 		}
 		created(Logger, w, r, job)
 		metrics.Increment("type.create.success")
-	})
+	}
 }
 
 // An EnqueueJobRequest is sent in the body of a request to PUT
@@ -388,15 +397,15 @@ type EnqueueJobRequest struct {
 }
 
 // GET/POST/PUT disambiguator for /v1/jobs/:name/:id
-func handleJobRoute(useMetaShutdown bool) http.HandlerFunc {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+func handleJobRoute(db *newmodels.Queries, useMetaShutdown bool) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
 		if r.Method == "POST" {
 			j := jobStatusUpdater{}
 			j.ServeHTTP(w, r)
 			return
 		}
 		if r.Method == "PUT" {
-			j := jobEnqueuer{useMetaShutdown: useMetaShutdown}
+			j := jobEnqueuer{db: db, useMetaShutdown: useMetaShutdown}
 			j.ServeHTTP(w, r)
 			return
 		}
@@ -406,7 +415,7 @@ func handleJobRoute(useMetaShutdown bool) http.HandlerFunc {
 			return
 		}
 		rest.NotAllowed(w, r)
-	})
+	}
 }
 
 type jobStatusGetter struct{}
@@ -472,11 +481,11 @@ func (j *jobStatusGetter) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	metrics.Increment("job.get.archived.success")
 }
 
-func createMetaShutdownJob(ctx context.Context) error {
+func createMetaShutdownJob(ctx context.Context, db *newmodels.Queries) error {
 	// This job type might not exist, we need to create it first to avoid
 	// a foreign key error. Just try to create the job every time - we
 	// shouldn't be hitting this endpoint that often
-	_, err := newmodels.DB.CreateJob(ctx, newmodels.CreateJobParams{
+	_, err := db.CreateJob(ctx, newmodels.CreateJobParams{
 		Name:             "meta.shutdown",
 		DeliveryStrategy: newmodels.DeliveryStrategyAtMostOnce,
 		Attempts:         1,
@@ -493,6 +502,7 @@ func createMetaShutdownJob(ctx context.Context) error {
 
 // jobEnqueuer satisfies the Handler interface.
 type jobEnqueuer struct {
+	db              *newmodels.Queries
 	useMetaShutdown bool
 }
 
@@ -553,9 +563,11 @@ func (j *jobEnqueuer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 	name := jobIdRoute.FindStringSubmatch(r.URL.Path)[1]
 	if name == "meta.shutdown" {
-		createMetaShutdownJob(r.Context())
+		createMetaShutdownJob(r.Context(), j.db)
 	}
-	queuedJob, err := queued_jobs.Enqueue(newmodels.EnqueueJobParams{
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+	queuedJob, err := services.Enqueue(ctx, j.db, newmodels.EnqueueJobParams{
 		ID: id, Name: name, RunAfter: ejr.RunAfter.Time,
 		ExpiresAt: ejr.ExpiresAt, Data: ejr.Data,
 	})
@@ -584,7 +596,7 @@ func (j *jobEnqueuer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			}
 		case *pq.Error:
 			if terr.Code == "23505" { // unique violation
-				queuedJob, err = queued_jobs.Get(r.Context(), id)
+				queuedJob, err = j.db.GetQueuedJob(r.Context(), id)
 				if err != nil {
 					rest.ServerError(w, r, err)
 					return
@@ -611,7 +623,7 @@ func (j *jobEnqueuer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 }
 
 // GET /v1/archived-jobs
-func listArchivedJobs() http.Handler {
+func listArchivedJobs(db *newmodels.Queries) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		query := r.URL.Query()
 		typ := query.Get("name")
@@ -645,19 +657,19 @@ func listArchivedJobs() http.Handler {
 		var err error
 		switch {
 		case typ == "" && status == "":
-			ajs, err = newmodels.DB.ListArchivedJobs(r.Context(), int32(limit))
+			ajs, err = db.ListArchivedJobs(r.Context(), int32(limit))
 		case status == "" && typ != "":
-			ajs, err = newmodels.DB.ListArchivedJobsByName(r.Context(), newmodels.ListArchivedJobsByNameParams{
+			ajs, err = db.ListArchivedJobsByName(r.Context(), newmodels.ListArchivedJobsByNameParams{
 				Limit: int32(limit),
 				Name:  typ,
 			})
 		case status != "" && typ == "":
-			ajs, err = newmodels.DB.ListArchivedJobsByStatus(r.Context(), newmodels.ListArchivedJobsByStatusParams{
+			ajs, err = db.ListArchivedJobsByStatus(r.Context(), newmodels.ListArchivedJobsByStatusParams{
 				Limit:  int32(limit),
 				Status: newmodels.ArchivedJobStatus(status),
 			})
 		case status != "" && typ != "":
-			ajs, err = newmodels.DB.ListArchivedJobsByNameStatus(r.Context(), newmodels.ListArchivedJobsByNameStatusParams{
+			ajs, err = db.ListArchivedJobsByNameStatus(r.Context(), newmodels.ListArchivedJobsByNameStatusParams{
 				Limit:  int32(limit),
 				Name:   typ,
 				Status: newmodels.ArchivedJobStatus(status),