Added api gateway service, basic functions
This commit is contained in:
Peter Grlica
@@ -0,0 +1,175 @@
|
||||
package apigw
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/system/types"
|
||||
"github.com/go-chi/chi"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
type (
|
||||
storer interface {
|
||||
SearchApigwRoutes(ctx context.Context, f types.RouteFilter) (types.RouteSet, types.RouteFilter, error)
|
||||
SearchApigwFunctions(ctx context.Context, f types.FunctionFilter) (types.FunctionSet, types.FunctionFilter, error)
|
||||
}
|
||||
|
||||
apigw struct {
|
||||
log *zap.Logger
|
||||
reg *registry
|
||||
routes []*route
|
||||
dispatcher dispatcher
|
||||
storer storer
|
||||
reload chan bool
|
||||
}
|
||||
)
|
||||
|
||||
var (
|
||||
// global service
|
||||
apiGw *apigw
|
||||
)
|
||||
|
||||
func Service() *apigw {
|
||||
return apiGw
|
||||
}
|
||||
|
||||
func Set(a *apigw) {
|
||||
apiGw = a
|
||||
}
|
||||
|
||||
// Setup handles the singleton service
|
||||
func Setup(opts interface{}, log *zap.Logger, dispatcher dispatcher, storer storer) {
|
||||
if apiGw != nil {
|
||||
return
|
||||
}
|
||||
|
||||
apiGw = New(opts, log, dispatcher, storer)
|
||||
}
|
||||
|
||||
func New(opts interface{}, logger *zap.Logger, dispatcher dispatcher, storer storer) *apigw {
|
||||
reg := NewRegistry()
|
||||
reg.Preload()
|
||||
|
||||
return &apigw{
|
||||
log: logger,
|
||||
dispatcher: dispatcher,
|
||||
storer: storer,
|
||||
reload: make(chan bool),
|
||||
reg: reg,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *apigw) Reload(ctx context.Context) {
|
||||
go func() {
|
||||
s.reload <- true
|
||||
}()
|
||||
}
|
||||
|
||||
func (s *apigw) loadRoutes(ctx context.Context) (rr []*route, err error) {
|
||||
routes, _, err := s.storer.SearchApigwRoutes(ctx, types.RouteFilter{Enabled: true})
|
||||
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
for _, r := range routes {
|
||||
route := &route{
|
||||
ID: r.ID,
|
||||
endpoint: r.Endpoint,
|
||||
method: r.Method,
|
||||
}
|
||||
|
||||
rr = append(rr, route)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (s *apigw) loadFunctions(ctx context.Context, route uint64) (ff []*types.Function, err error) {
|
||||
ff, _, err = s.storer.SearchApigwFunctions(ctx, types.FunctionFilter{})
|
||||
return
|
||||
}
|
||||
|
||||
func (s *apigw) Router(ctx context.Context) func(r chi.Router) {
|
||||
return func(r chi.Router) {
|
||||
|
||||
routes, err := s.loadRoutes(ctx)
|
||||
|
||||
if err != nil {
|
||||
s.log.Error("could not load routes", zap.Error(err))
|
||||
return
|
||||
}
|
||||
|
||||
s.Init(ctx, routes...)
|
||||
|
||||
for _, route := range s.routes {
|
||||
r.Handle(route.endpoint, route)
|
||||
}
|
||||
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
case <-s.reload:
|
||||
s.log.Debug("got reload signal")
|
||||
|
||||
routes, err := s.loadRoutes(ctx)
|
||||
|
||||
if err != nil {
|
||||
s.log.Error("could not reload routes", zap.Error(err))
|
||||
return
|
||||
}
|
||||
|
||||
s.Init(ctx, routes...)
|
||||
|
||||
for _, route := range s.routes {
|
||||
r.Handle(route.endpoint, route)
|
||||
}
|
||||
|
||||
case <-ctx.Done():
|
||||
s.log.Debug("done! getting out")
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
// init all the routes
|
||||
func (s *apigw) Init(ctx context.Context, route ...*route) {
|
||||
s.routes = route
|
||||
|
||||
s.log.Debug("initializing routes\n", zap.Int("num", len(s.routes)))
|
||||
|
||||
for _, r := range s.routes {
|
||||
r.pipe = &pl{}
|
||||
regFuncs, err := s.loadFunctions(ctx, r.ID)
|
||||
|
||||
if err != nil {
|
||||
s.log.Error("could not load functions for route", zap.String("route", r.endpoint), zap.Error(err))
|
||||
continue
|
||||
}
|
||||
|
||||
r.pipe.ErrorHandler(errorHandler{
|
||||
name: "error handler expediter",
|
||||
args: []string{},
|
||||
weight: 5,
|
||||
step: 0,
|
||||
})
|
||||
|
||||
for _, f := range regFuncs {
|
||||
fc := functionHandler{}
|
||||
|
||||
h, err := s.reg.Get(f.Ref)
|
||||
|
||||
if err != nil {
|
||||
s.log.Error("could not register function for route", zap.String("route", r.endpoint), zap.Error(err))
|
||||
continue
|
||||
}
|
||||
|
||||
fc.Merge(ctx, h.Meta(f))
|
||||
fc.SetHandler(h.Handler())
|
||||
|
||||
r.pipe.Add(fc, f.Params)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
package apigw
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/expr"
|
||||
"github.com/cortezaproject/corteza-server/pkg/logger"
|
||||
"github.com/cortezaproject/corteza-server/pkg/wfexec"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func execFn(t *testing.T, r *http.Request, fn wfHandler) error {
|
||||
var (
|
||||
req = require.New(t)
|
||||
ctx = context.Background()
|
||||
scope = &expr.Vars{}
|
||||
graph = wfexec.NewGraph()
|
||||
recorder = httptest.NewRecorder()
|
||||
)
|
||||
|
||||
scope.Set("envelope", envelope{
|
||||
Request: r,
|
||||
Writer: recorder,
|
||||
})
|
||||
|
||||
step := wfexec.NewGenericStep(fn.self())
|
||||
|
||||
graph.AddStep(step)
|
||||
|
||||
sess := wfexec.NewSession(ctx, graph, wfexec.SetLogger(logger.Default()))
|
||||
|
||||
err := sess.Exec(ctx, step, scope)
|
||||
|
||||
req.NoError(err)
|
||||
|
||||
return sess.Wait(ctx)
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
package apigw
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/expr"
|
||||
"github.com/cortezaproject/corteza-server/pkg/wfexec"
|
||||
"github.com/davecgh/go-spew/spew"
|
||||
)
|
||||
|
||||
type (
|
||||
redirectExpediterArgs struct {
|
||||
Location string
|
||||
}
|
||||
)
|
||||
|
||||
func redirectExpediter(c context.Context, params *expr.Vars) wfHandler {
|
||||
var (
|
||||
clv = redirectExpediterArgs{}
|
||||
)
|
||||
|
||||
return func(c context.Context, er *wfexec.ExecRequest) (r wfexec.ExecResponse, err error) {
|
||||
params.Decode(&clv)
|
||||
spew.Dump("redirect expediter fn()", clv)
|
||||
e := er.Scope.GetValue()["envelope"]
|
||||
ee := e.Get().(envelope)
|
||||
|
||||
http.Redirect(ee.Writer, ee.Request, clv.Location, http.StatusTemporaryRedirect)
|
||||
|
||||
r = &expr.Vars{}
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func expediterErrorFn(c context.Context, er *wfexec.ExecRequest) (r wfexec.ExecResponse, err error) {
|
||||
// spew.Dump("expediter error fn()", er)
|
||||
|
||||
e := er.Scope.GetValue()["error"]
|
||||
values := er.Scope.GetValue()["writer"]
|
||||
|
||||
writer := values.Get()
|
||||
eValue := e.Get()
|
||||
|
||||
fmt.Fprintf(writer.(*httptest.ResponseRecorder), fmt.Sprintf(`{"msg": "%s"}`, eValue))
|
||||
writer.(*httptest.ResponseRecorder).Code = http.StatusBadGateway
|
||||
|
||||
r = &expr.Vars{}
|
||||
return
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
package apigw
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/system/types"
|
||||
)
|
||||
|
||||
type (
|
||||
Handler interface {
|
||||
Handler() handlerFunc
|
||||
Meta(f *types.Function) functionMeta
|
||||
}
|
||||
|
||||
handlerFunc func(context.Context, *scp, map[string]interface{}, functionHandler) error
|
||||
|
||||
functionMeta struct {
|
||||
step int
|
||||
weight int
|
||||
name string
|
||||
label string
|
||||
kind string
|
||||
params map[string]interface{}
|
||||
}
|
||||
|
||||
functionHandler struct {
|
||||
step int
|
||||
weight int
|
||||
name string
|
||||
label string
|
||||
kind string
|
||||
handler handlerFunc
|
||||
params map[string]interface{}
|
||||
}
|
||||
)
|
||||
|
||||
func (ff functionHandler) Exec(ctx context.Context, scope *scp, params map[string]interface{}) error {
|
||||
return ff.handler(ctx, scope, params, ff)
|
||||
}
|
||||
|
||||
func (ff *functionHandler) SetHandler(h handlerFunc) {
|
||||
ff.handler = h
|
||||
}
|
||||
|
||||
func (ff *functionHandler) Merge(ctx context.Context, p functionMeta) {
|
||||
ff.step = p.step
|
||||
ff.kind = p.kind
|
||||
ff.label = p.label
|
||||
ff.name = p.name
|
||||
ff.weight = p.weight
|
||||
ff.params = p.params
|
||||
}
|
||||
|
||||
func (ff functionHandler) Weight() int {
|
||||
// if there's gonna be more than 1000 funcs
|
||||
// per step, we're doing something wrong
|
||||
return ff.step*1000 + ff.weight
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
package apigw
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/expr"
|
||||
"github.com/cortezaproject/corteza-server/pkg/wfexec"
|
||||
"github.com/davecgh/go-spew/spew"
|
||||
)
|
||||
|
||||
type (
|
||||
authenticationOriginMatcherArgs struct {
|
||||
Origin string
|
||||
}
|
||||
)
|
||||
|
||||
func authenticationOriginMatcher(c context.Context, params *expr.Vars) wfHandler {
|
||||
var (
|
||||
aomp = authenticationOriginMatcherArgs{}
|
||||
)
|
||||
|
||||
return func(c context.Context, er *wfexec.ExecRequest) (r wfexec.ExecResponse, err error) {
|
||||
|
||||
params.Decode(&aomp)
|
||||
spew.Dump("authentication origin matcher fn()", aomp)
|
||||
e := er.Scope.GetValue()["envelope"]
|
||||
ee := e.Get().(envelope)
|
||||
|
||||
origin := ee.Request.Header.Get("Origin")
|
||||
|
||||
spew.Dump("input, real", aomp.Origin, origin)
|
||||
|
||||
if aomp.Origin != origin {
|
||||
err = errors.New("origin fail")
|
||||
return
|
||||
}
|
||||
|
||||
r = &expr.Vars{}
|
||||
return
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
package apigw
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"testing"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/expr"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestAuthenticationOriginMatcher(t *testing.T) {
|
||||
type (
|
||||
tf struct {
|
||||
name string
|
||||
origin string
|
||||
exp string
|
||||
req *http.Request
|
||||
}
|
||||
)
|
||||
|
||||
var (
|
||||
ctx = context.Background()
|
||||
|
||||
tcc = []tf{
|
||||
{
|
||||
name: "fail on origin",
|
||||
origin: "http://fail.ed",
|
||||
exp: "workflow 0 step 0 execution failed: origin fail",
|
||||
|
||||
req: &http.Request{
|
||||
Header: http.Header{
|
||||
"Origin": []string{
|
||||
"http://localhost",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "success on origin",
|
||||
origin: "http://localhost",
|
||||
exp: "",
|
||||
|
||||
req: &http.Request{
|
||||
Header: http.Header{
|
||||
"Origin": []string{
|
||||
"http://localhost",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
for _, tc := range tcc {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
var (
|
||||
req = require.New(t)
|
||||
input = &expr.Vars{}
|
||||
)
|
||||
|
||||
input.Set("origin", tc.origin)
|
||||
|
||||
err := execFn(t, tc.req, authenticationOriginMatcher(ctx, input))
|
||||
|
||||
if tc.exp != "" {
|
||||
req.EqualError(err, tc.exp)
|
||||
} else {
|
||||
req.NoError(err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
package apigw
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/expr"
|
||||
"github.com/cortezaproject/corteza-server/pkg/wfexec"
|
||||
"github.com/davecgh/go-spew/spew"
|
||||
)
|
||||
|
||||
func formDataProcesserFn(c context.Context, er *wfexec.ExecRequest) (r wfexec.ExecResponse, err error) {
|
||||
type (
|
||||
formDataProcesserResponse struct {
|
||||
Name string `json:"name"`
|
||||
}
|
||||
)
|
||||
|
||||
spew.Dump("step processer fn()")
|
||||
|
||||
e := er.Scope.GetValue()["envelope"]
|
||||
ee := e.Get()
|
||||
|
||||
// ee.(envelope).Writer.WriteHeader(int(id))
|
||||
ee.(envelope).Writer.Write([]byte(`{"test":"foobar"}`))
|
||||
|
||||
e.Assign(ee)
|
||||
|
||||
// req := values.Get()
|
||||
// ww := wr.Get()
|
||||
// writer := ww.(http.ResponseWriter)
|
||||
|
||||
// formValue := req.(*http.Request).PostFormValue("name")
|
||||
|
||||
// resp := formDataProcesserResponse{
|
||||
// // Name: fmt.Sprintf("AA %s AA", formValue),
|
||||
// Name: "formValue",
|
||||
// }
|
||||
|
||||
// encoder := json.NewEncoder(writer)
|
||||
// encoder.Encode(resp)
|
||||
|
||||
// writer.(*httptest.ResponseRecorder).Header()["Content-Type"] = []string{"application/json"}
|
||||
// writer.Header().Set("Content-Type", "application/json3")
|
||||
|
||||
// spew.Dump(writer.(*httptest.ResponseRecorder).Header())
|
||||
// a, b := expr.NewKV(writer)
|
||||
// spew.Dump("Aaaaaaaaaaa", a)
|
||||
|
||||
vv := &expr.Vars{}
|
||||
// vv.Set("writer", writer)
|
||||
|
||||
r = vv
|
||||
|
||||
return
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
package apigw
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/expr"
|
||||
"github.com/cortezaproject/corteza-server/pkg/logger"
|
||||
"github.com/cortezaproject/corteza-server/pkg/wfexec"
|
||||
"github.com/davecgh/go-spew/spew"
|
||||
)
|
||||
|
||||
type (
|
||||
route struct {
|
||||
endpoint string
|
||||
method string
|
||||
graph *wfexec.Graph
|
||||
steps []wfexec.Step
|
||||
fns wfHandlerList
|
||||
}
|
||||
)
|
||||
|
||||
func (r route) validate(req *http.Request) (err error) {
|
||||
// if req.Method != r.method {
|
||||
// err = errors.New("http method invalid")
|
||||
// }
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (r route) ServeHTTP(w http.ResponseWriter, req *http.Request) {
|
||||
if err := r.validate(req); err != nil {
|
||||
spew.Dump("ERR", err)
|
||||
return
|
||||
}
|
||||
|
||||
sess := wfexec.NewSession(context.Background(), r.graph, wfexec.SetLogger(logger.Default()), wfexec.SetHandler(func(ss wfexec.SessionStatus, s1 *wfexec.State, s2 *wfexec.Session) {
|
||||
// spew.Dump("event handler here!", ss)
|
||||
}))
|
||||
|
||||
scope := &expr.Vars{}
|
||||
|
||||
scope.Set("envelope", envelope{
|
||||
Request: req,
|
||||
Writer: w,
|
||||
})
|
||||
|
||||
if len(r.steps) == 0 {
|
||||
// dont serve, do what? return default response?
|
||||
return
|
||||
}
|
||||
|
||||
err := sess.Exec(context.Background(), r.steps[0], scope)
|
||||
|
||||
// if err != nil {
|
||||
// fmt.Fprintf(w, "no go, err on exec: %s", err)
|
||||
// return
|
||||
// }
|
||||
|
||||
err = sess.Wait(context.Background())
|
||||
|
||||
if err != nil {
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
package apigw
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/expr"
|
||||
"github.com/cortezaproject/corteza-server/pkg/wfexec"
|
||||
"github.com/davecgh/go-spew/spew"
|
||||
)
|
||||
|
||||
type (
|
||||
contentLengthValidatorArgs struct {
|
||||
Length int
|
||||
}
|
||||
)
|
||||
|
||||
func contentLengthValidator(c context.Context, params *expr.Vars) wfHandler {
|
||||
var (
|
||||
clv = contentLengthValidatorArgs{}
|
||||
)
|
||||
|
||||
return func(c context.Context, er *wfexec.ExecRequest) (r wfexec.ExecResponse, err error) {
|
||||
|
||||
params.Decode(&clv)
|
||||
spew.Dump("body size validator fn()", clv)
|
||||
e := er.Scope.GetValue()["envelope"]
|
||||
ee := e.Get().(envelope)
|
||||
|
||||
cl := ee.Request.ContentLength
|
||||
|
||||
if clv.Length < int(cl) {
|
||||
err = errors.New("content length overriden")
|
||||
return
|
||||
}
|
||||
|
||||
r = &expr.Vars{}
|
||||
return
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,63 @@
|
||||
package apigw
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/expr"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestContentLengthValidator(t *testing.T) {
|
||||
type (
|
||||
tf struct {
|
||||
name string
|
||||
limit int
|
||||
exp string
|
||||
body string
|
||||
}
|
||||
)
|
||||
|
||||
var (
|
||||
ctx = context.Background()
|
||||
|
||||
tcc = []tf{
|
||||
{
|
||||
name: "fail on content length > limit",
|
||||
limit: 10,
|
||||
exp: "workflow 0 step 0 execution failed: content length overriden",
|
||||
body: "A message that is 31 bytes long",
|
||||
},
|
||||
{
|
||||
name: "success on content length < limit",
|
||||
limit: 10,
|
||||
exp: "",
|
||||
body: "Below 10",
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
for _, tc := range tcc {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
var (
|
||||
req = require.New(t)
|
||||
input = &expr.Vars{}
|
||||
)
|
||||
|
||||
input.Set("length", tc.limit)
|
||||
|
||||
r := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(tc.body))
|
||||
|
||||
err := execFn(t, r, contentLengthValidator(ctx, input))
|
||||
|
||||
if tc.exp != "" {
|
||||
req.EqualError(err, tc.exp)
|
||||
} else {
|
||||
req.NoError(err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,230 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/actionlog"
|
||||
"github.com/cortezaproject/corteza-server/pkg/apigw"
|
||||
a "github.com/cortezaproject/corteza-server/pkg/auth"
|
||||
"github.com/cortezaproject/corteza-server/store"
|
||||
"github.com/cortezaproject/corteza-server/system/types"
|
||||
)
|
||||
|
||||
type (
|
||||
route struct {
|
||||
actionlog actionlog.Recorder
|
||||
store store.Storer
|
||||
ac routeAccessController
|
||||
}
|
||||
|
||||
routeAccessController interface {
|
||||
}
|
||||
)
|
||||
|
||||
func Route() *route {
|
||||
return (&route{
|
||||
ac: DefaultAccessControl,
|
||||
actionlog: DefaultActionlog,
|
||||
store: DefaultStore,
|
||||
})
|
||||
}
|
||||
|
||||
func (svc *route) FindByID(ctx context.Context, ID uint64) (q *types.Route, err error) {
|
||||
var (
|
||||
rProps = &routeActionProps{}
|
||||
)
|
||||
|
||||
err = func() error {
|
||||
if ID == 0 {
|
||||
return RouteErrInvalidID()
|
||||
}
|
||||
|
||||
if q, err = store.LookupApigwRouteByID(ctx, svc.store, ID); err != nil {
|
||||
return RouteErrInvalidID().Wrap(err)
|
||||
}
|
||||
|
||||
rProps.setRoute(q)
|
||||
|
||||
// if !svc.ac.CanReadMessagebusQueue(ctx, q) {
|
||||
// return QueueErrNotAllowedToRead(qProps)
|
||||
// }
|
||||
|
||||
return nil
|
||||
}()
|
||||
|
||||
return q, svc.recordAction(ctx, rProps, RouteActionLookup, err)
|
||||
}
|
||||
|
||||
func (svc *route) Create(ctx context.Context, new *types.Route) (q *types.Route, err error) {
|
||||
var (
|
||||
qProps = &routeActionProps{new: new}
|
||||
)
|
||||
|
||||
err = func() (err error) {
|
||||
// if !svc.ac.CanCreateMessagebusQueue(ctx) {
|
||||
// return QueueErrNotAllowedToCreate(qProps)
|
||||
// }
|
||||
|
||||
// Set new values after beforeCreate events are emitted
|
||||
new.ID = nextID()
|
||||
new.CreatedAt = *now()
|
||||
new.CreatedBy = a.GetIdentityFromContext(ctx).Identity()
|
||||
|
||||
// todo
|
||||
new.Group = 0
|
||||
|
||||
if err = store.CreateApigwRoute(ctx, svc.store, new); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
q = new
|
||||
|
||||
// send the signal to reload all routes
|
||||
if new.Enabled {
|
||||
apigw.Service().Reload(ctx)
|
||||
}
|
||||
|
||||
return nil
|
||||
}()
|
||||
|
||||
return q, svc.recordAction(ctx, qProps, RouteActionCreate, err)
|
||||
}
|
||||
|
||||
func (svc *route) Update(ctx context.Context, upd *types.Route) (q *types.Route, err error) {
|
||||
var (
|
||||
qProps = &routeActionProps{update: upd}
|
||||
qq *types.Route
|
||||
e error
|
||||
)
|
||||
|
||||
err = func() (err error) {
|
||||
// if !svc.ac.CanUpdateMessagebusQueue(ctx, upd) {
|
||||
// return QueueErrNotAllowedToUpdate(qProps)
|
||||
// }
|
||||
|
||||
if qq, e = store.LookupApigwRouteByID(ctx, svc.store, upd.ID); e != nil {
|
||||
return RouteErrNotFound(qProps)
|
||||
}
|
||||
|
||||
if qq, e = store.LookupApigwRouteByEndpoint(ctx, svc.store, upd.Endpoint); e == nil && qq != nil {
|
||||
return RouteErrExistsEndpoint(qProps)
|
||||
}
|
||||
|
||||
// Set new values after beforeCreate events are emitted
|
||||
upd.UpdatedAt = now()
|
||||
upd.CreatedAt = qq.CreatedAt
|
||||
upd.UpdatedBy = a.GetIdentityFromContext(ctx).Identity()
|
||||
|
||||
if err = store.UpdateApigwRoute(ctx, svc.store, upd); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
q = upd
|
||||
|
||||
// send the signal to reload all route
|
||||
apigw.Service().Reload(ctx)
|
||||
|
||||
return nil
|
||||
}()
|
||||
|
||||
return q, svc.recordAction(ctx, qProps, RouteActionUpdate, err)
|
||||
}
|
||||
|
||||
func (svc *route) DeleteByID(ctx context.Context, ID uint64) (err error) {
|
||||
var (
|
||||
qProps = &routeActionProps{}
|
||||
q *types.Route
|
||||
)
|
||||
|
||||
err = func() (err error) {
|
||||
if ID == 0 {
|
||||
return RouteErrInvalidID()
|
||||
}
|
||||
|
||||
if q, err = store.LookupApigwRouteByID(ctx, svc.store, ID); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
qProps.setRoute(q)
|
||||
|
||||
// if !svc.ac.CanDeleteMessagebusQueue(ctx, q) {
|
||||
// return QueueErrNotAllowedToDelete(qProps)
|
||||
// }
|
||||
|
||||
q.DeletedAt = now()
|
||||
q.DeletedBy = a.GetIdentityFromContext(ctx).Identity()
|
||||
|
||||
if err = store.UpdateApigwRoute(ctx, svc.store, q); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// send the signal to reload all queues
|
||||
apigw.Service().Reload(ctx)
|
||||
|
||||
return nil
|
||||
}()
|
||||
|
||||
return svc.recordAction(ctx, qProps, RouteActionDelete, err)
|
||||
}
|
||||
|
||||
func (svc *route) UndeleteByID(ctx context.Context, ID uint64) (err error) {
|
||||
var (
|
||||
qProps = &routeActionProps{}
|
||||
q *types.Route
|
||||
)
|
||||
|
||||
err = func() (err error) {
|
||||
if ID == 0 {
|
||||
return RouteErrInvalidID()
|
||||
}
|
||||
|
||||
if q, err = store.LookupApigwRouteByID(ctx, svc.store, ID); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
qProps.setRoute(q)
|
||||
|
||||
// if !svc.ac.CanDeleteMessagebusQueue(ctx, q) {
|
||||
// return QueueErrNotAllowedToDelete(qProps)
|
||||
// }
|
||||
|
||||
q.DeletedAt = nil
|
||||
q.UpdatedBy = a.GetIdentityFromContext(ctx).Identity()
|
||||
|
||||
if err = store.UpdateApigwRoute(ctx, svc.store, q); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// send the signal to reload all queues
|
||||
apigw.Service().Reload(ctx)
|
||||
|
||||
return nil
|
||||
}()
|
||||
|
||||
return svc.recordAction(ctx, qProps, RouteActionDelete, err)
|
||||
}
|
||||
|
||||
func (svc *route) Search(ctx context.Context, filter types.RouteFilter) (r types.RouteSet, f types.RouteFilter, err error) {
|
||||
var (
|
||||
aProps = &routeActionProps{search: &filter}
|
||||
)
|
||||
|
||||
// For each fetched item, store backend will check if it is valid or not
|
||||
// filter.Check = func(res *messagebus.QueueSettings) (bool, error) {
|
||||
// if !svc.ac.CanReadMessagebusQueue(ctx, res) {
|
||||
// return false, nil
|
||||
// }
|
||||
|
||||
// return true, nil
|
||||
// }
|
||||
|
||||
err = func() error {
|
||||
if r, f, err = store.SearchApigwRoutes(ctx, svc.store, filter); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}()
|
||||
|
||||
return r, f, svc.recordAction(ctx, aProps, RouteActionSearch, err)
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
package types
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/filter"
|
||||
)
|
||||
|
||||
type (
|
||||
RouteMeta struct{}
|
||||
|
||||
Route struct {
|
||||
ID uint64 `json:"routeID,string"`
|
||||
Endpoint string `json:"endpoint"`
|
||||
Method string `json:"method"`
|
||||
Debug bool `json:"debug"`
|
||||
Enabled bool `json:"enabled"`
|
||||
Group uint64 `json:"group"`
|
||||
|
||||
CreatedAt time.Time `json:"createdAt,omitempty"`
|
||||
CreatedBy uint64 `json:"createdBy,string" `
|
||||
UpdatedAt *time.Time `json:"updatedAt,omitempty"`
|
||||
UpdatedBy uint64 `json:"updatedBy,string,omitempty" `
|
||||
DeletedAt *time.Time `json:"deletedAt,omitempty"`
|
||||
DeletedBy uint64 `json:"deletedBy,string,omitempty" `
|
||||
}
|
||||
|
||||
RouteFilter struct {
|
||||
Route string `json:"route"`
|
||||
Group string `json:"group"`
|
||||
Enabled bool `json:"enabled"`
|
||||
|
||||
Deleted filter.State `json:"deleted"`
|
||||
|
||||
// Check fn is called by store backend for each resource found function can
|
||||
// modify the resource and return false if store should not return it
|
||||
//
|
||||
// Store then loads additional resources to satisfy the paging parameters
|
||||
Check func(*Route) (bool, error) `json:"-"`
|
||||
|
||||
filter.Sorting
|
||||
filter.Paging
|
||||
}
|
||||
)
|
||||
Reference in New Issue
Block a user