fiumba
This commit is contained in:
Ramiro Paz
42
src/client/store/external/auth.go
vendored
Normal file
42
src/client/store/external/auth.go
vendored
Normal file
@ -0,0 +1,42 @@
|
||||
package external
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"time"
|
||||
)
|
||||
|
||||
func (s *Manager) ValidateSession(sessionToken string) (bool, error) {
|
||||
path := "/api/v1/auth/session/validate"
|
||||
|
||||
auth, ok := s.config.External[QApixService]
|
||||
if !ok {
|
||||
err := errors.New("error validating auth to qapix service at ValidateSession")
|
||||
slog.Error(err.Error())
|
||||
|
||||
return false, err
|
||||
}
|
||||
|
||||
options := RequestOptions{
|
||||
Method: http.MethodGet,
|
||||
Path: path,
|
||||
Body: nil,
|
||||
Retries: 3,
|
||||
Timeout: sTimeout * time.Second,
|
||||
CacheDuration: time.Second * 10,
|
||||
Auth: &auth,
|
||||
SessionToken: sessionToken,
|
||||
}
|
||||
|
||||
res, err := s.sendRequestToExternal(options)
|
||||
if err != nil || res == nil {
|
||||
err := fmt.Errorf("error making ValidateSession request to qapix server. error: %w", err)
|
||||
slog.Error(err.Error())
|
||||
|
||||
return false, err
|
||||
}
|
||||
|
||||
return true, nil
|
||||
}
|
||||
273
src/client/store/external/manager.go
vendored
Normal file
273
src/client/store/external/manager.go
vendored
Normal file
@ -0,0 +1,273 @@
|
||||
// Package external defines all external services access
|
||||
package external
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/rs/zerolog/log"
|
||||
"github.com/sasha-s/go-deadlock"
|
||||
|
||||
"quantex.com/qfixdpl/src/app"
|
||||
"quantex.com/qfixdpl/src/app/version"
|
||||
jwttoken "quantex.com/qfixdpl/src/common/jwttoken"
|
||||
"quantex.com/qfixdpl/src/domain"
|
||||
)
|
||||
|
||||
const (
|
||||
sTimeout = 10
|
||||
lTimeout = 15
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
QApixPort string
|
||||
QApixHost string
|
||||
External map[string]app.ExtAuth
|
||||
QApixToken string
|
||||
EnableJWTAuth bool
|
||||
}
|
||||
|
||||
type cacheItem struct {
|
||||
Time time.Time
|
||||
Resp []byte
|
||||
}
|
||||
|
||||
type Manager struct {
|
||||
config Config
|
||||
notifier domain.Notifier
|
||||
cache map[string]cacheItem
|
||||
mutex deadlock.RWMutex
|
||||
}
|
||||
|
||||
type RequestOptions struct {
|
||||
Method string
|
||||
Path string
|
||||
URL string
|
||||
Token string
|
||||
Secured bool
|
||||
Body interface{}
|
||||
Retries int
|
||||
Timeout time.Duration
|
||||
CacheDuration time.Duration
|
||||
Auth *app.ExtAuth
|
||||
SessionToken string
|
||||
}
|
||||
|
||||
// NewManager create new Manager struct
|
||||
func NewManager(n domain.Notifier, cfg Config) *Manager {
|
||||
return &Manager{
|
||||
config: cfg,
|
||||
notifier: n,
|
||||
cache: make(map[string]cacheItem),
|
||||
}
|
||||
}
|
||||
|
||||
//revive:disable:argument-limit we need this arguments
|
||||
func (s *Manager) sendRequestToExternal(opts RequestOptions) ([]byte, error) {
|
||||
host := fmt.Sprintf("http://localhost:%v", opts.Auth.Port)
|
||||
if opts.Auth.Host != "" {
|
||||
host = opts.Auth.Host
|
||||
}
|
||||
|
||||
token := s.config.QApixToken
|
||||
|
||||
secured := false
|
||||
|
||||
if s.config.EnableJWTAuth && opts.Auth != nil {
|
||||
t, err := jwttoken.Encrypt(*opts.Auth)
|
||||
if err != nil {
|
||||
e := fmt.Errorf("error encrypting quantexService: %w", err)
|
||||
log.Error().Msg(e.Error())
|
||||
|
||||
return nil, e
|
||||
}
|
||||
|
||||
token = t
|
||||
|
||||
secured = true
|
||||
}
|
||||
|
||||
opts.Secured = secured
|
||||
opts.Token = token
|
||||
opts.URL = urlFrom(host, opts.Path)
|
||||
|
||||
return s.sendRequestWithCache(opts)
|
||||
}
|
||||
|
||||
func (s *Manager) sendRequestWithCache(opts RequestOptions) ([]byte, error) {
|
||||
sha := optionSha(opts)
|
||||
|
||||
if opts.CacheDuration > 0 && len(sha) > 0 {
|
||||
s.mutex.RLock()
|
||||
t, ok := s.cache[sha]
|
||||
s.mutex.RUnlock()
|
||||
|
||||
if ok && time.Since(t.Time) < opts.CacheDuration {
|
||||
return t.Resp, nil
|
||||
}
|
||||
}
|
||||
|
||||
resp, err := s.sendRequestWithRetries(opts)
|
||||
if err == nil && len(sha) > 0 {
|
||||
s.mutex.Lock()
|
||||
s.cache[sha] = cacheItem{
|
||||
time.Now(),
|
||||
resp,
|
||||
}
|
||||
s.mutex.Unlock()
|
||||
}
|
||||
|
||||
return resp, err
|
||||
}
|
||||
|
||||
//revive:disable:flag-parameter we need this flag
|
||||
//nolint:funlen //it's long but easy to read
|
||||
func (s *Manager) sendRequestWithRetries(opts RequestOptions) ([]byte, error) {
|
||||
client := &http.Client{
|
||||
Timeout: opts.Timeout,
|
||||
}
|
||||
|
||||
bodyBytes, err := json.Marshal(&opts.Body)
|
||||
if err != nil {
|
||||
e := fmt.Errorf("error encoding the body: %v, %w", opts.Body, err)
|
||||
slog.Error(e.Error())
|
||||
|
||||
return nil, e
|
||||
}
|
||||
|
||||
slog.Debug("sending request to: " + opts.URL)
|
||||
|
||||
request, err := http.NewRequest(opts.Method, opts.URL, bytes.NewBuffer(bodyBytes)) //nolint:noctx //no ctx needed
|
||||
if err != nil {
|
||||
e := fmt.Errorf("error creating new %s request to: %s, %w", opts.Method, opts.URL, err)
|
||||
slog.Error(e.Error())
|
||||
|
||||
return nil, e
|
||||
}
|
||||
|
||||
authorization := "Bearer " + opts.Token
|
||||
if opts.Secured {
|
||||
authorization = opts.Token
|
||||
}
|
||||
|
||||
request.Header.Set("Authorization", authorization)
|
||||
|
||||
request.Header.Set("Content-Type", "application/json")
|
||||
|
||||
if opts.SessionToken != "" {
|
||||
// Create a cookie
|
||||
cookie := &http.Cookie{
|
||||
Name: "session_token",
|
||||
Value: opts.SessionToken,
|
||||
Path: "/",
|
||||
}
|
||||
|
||||
// Add the cookie to the request
|
||||
request.AddCookie(cookie)
|
||||
}
|
||||
|
||||
var resp *http.Response
|
||||
for i := 0; i <= opts.Retries; i++ { //nolint:wsl // It's ok in this case
|
||||
interval := time.Duration(i) * time.Second
|
||||
slog.Debug(fmt.Sprintf("request to '%s' try #%v in %v", request.URL, i, interval))
|
||||
time.Sleep(interval)
|
||||
|
||||
resp, err = client.Do(request)
|
||||
if err != nil {
|
||||
e := fmt.Errorf("error making request to %s. error: %w", request.URL, err)
|
||||
slog.Error(e.Error())
|
||||
|
||||
// send notification if notifier is available
|
||||
s.sendNotification(domain.MessageChannelError, e.Error(), domain.MessageStatusWarning, nil)
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
break
|
||||
}
|
||||
|
||||
if resp == nil {
|
||||
err = fmt.Errorf("error: '%s' response is nil", request.URL)
|
||||
slog.Error(err.Error())
|
||||
|
||||
return nil, err
|
||||
}
|
||||
|
||||
defer func() {
|
||||
if err := resp.Body.Close(); err != nil {
|
||||
slog.Error("error closing response body at sendRequestWithRetries: " + err.Error())
|
||||
}
|
||||
}()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
msg := fmt.Sprintf("response code from %s is not 200. StatusCode: %d.", opts.URL, resp.StatusCode)
|
||||
// send notification if notifier is available
|
||||
s.sendNotification(domain.MessageChannelError, msg, domain.MessageStatusWarning, nil)
|
||||
|
||||
return nil, fmt.Errorf("error %s", msg)
|
||||
}
|
||||
|
||||
bodyBytes, resErr := io.ReadAll(resp.Body)
|
||||
if resErr != nil {
|
||||
msg := fmt.Sprintf("error reading response from %s. error: %s", opts.URL, resErr.Error())
|
||||
// send notification if notifier is available
|
||||
s.sendNotification(domain.MessageChannelError, msg, domain.MessageStatusWarning, nil)
|
||||
|
||||
return nil, fmt.Errorf("error %s", msg)
|
||||
}
|
||||
|
||||
return bodyBytes, nil
|
||||
}
|
||||
|
||||
//revive:enable
|
||||
|
||||
// sendNotification is a nil-safe wrapper around the notifier's SendMsg method.
|
||||
// Some call sites run in contexts where Manager.notifier may be nil, so guard the call
|
||||
// to avoid runtime panics.
|
||||
func (s *Manager) sendNotification(channel domain.MessageChannel, message string, status domain.MessageStatus, wg *sync.WaitGroup) {
|
||||
if s == nil || s.notifier == nil {
|
||||
slog.Debug("notifier is nil, skipping SendMsg")
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
s.notifier.SendMsg(channel, message, status, wg)
|
||||
}
|
||||
|
||||
func optionSha(opts RequestOptions) string {
|
||||
key := fmt.Sprintf("%s-%s-%s", opts.Method, opts.URL, opts.Body)
|
||||
|
||||
return shaString(key)
|
||||
}
|
||||
|
||||
func shaString(s string) (sha string) {
|
||||
hash := sha256.New()
|
||||
if _, err := hash.Write([]byte(s)); err != nil {
|
||||
err = fmt.Errorf("error generating hash at generateShaString, error: %w", err)
|
||||
slog.Error(err.Error())
|
||||
|
||||
return ""
|
||||
}
|
||||
|
||||
return hex.EncodeToString(hash.Sum(nil))
|
||||
}
|
||||
|
||||
func urlFrom(host, path string) string {
|
||||
from := strings.ToLower(version.AppName)
|
||||
|
||||
url := fmt.Sprintf("%s%s?from=%s", host, path, from)
|
||||
if strings.Contains(path, "?") {
|
||||
url = fmt.Sprintf("%s%s&from=%s", host, path, from)
|
||||
}
|
||||
|
||||
return url
|
||||
}
|
||||
141
src/client/store/external/user.go
vendored
Normal file
141
src/client/store/external/user.go
vendored
Normal file
@ -0,0 +1,141 @@
|
||||
package external
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"quantex.com/qfixdpl/src/app"
|
||||
)
|
||||
|
||||
const QApixService = "QApix"
|
||||
|
||||
func (s *Manager) Users(out any) (err error) {
|
||||
auth, ok := s.config.External[QApixService]
|
||||
if !ok {
|
||||
err = errors.New("error getting auth data for qapix service at Users")
|
||||
|
||||
slog.Error(err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
options := RequestOptions{
|
||||
Method: http.MethodGet,
|
||||
Path: "/api/v1/auth/data/users",
|
||||
Body: nil,
|
||||
Retries: 0,
|
||||
Timeout: lTimeout * time.Second,
|
||||
CacheDuration: -1,
|
||||
Auth: &auth,
|
||||
}
|
||||
|
||||
response, err := s.sendRequestToExternal(options)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("error making request to qapix server. error: %w", err)
|
||||
|
||||
slog.Error(err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
err = json.Unmarshal(response, out)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("error making request to qapix server. error: %w", err)
|
||||
|
||||
slog.Error(err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Manager) UserByID(userID string, out any) (err error) {
|
||||
path := "/api/v1/auth/data/user_by_id/" + userID
|
||||
|
||||
auth, ok := s.config.External[QApixService]
|
||||
if !ok {
|
||||
err = errors.New("error getting auth data for qapix service at UserByEmail")
|
||||
|
||||
slog.Error(err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
options := RequestOptions{
|
||||
Method: http.MethodGet,
|
||||
Path: path,
|
||||
Body: nil,
|
||||
Retries: 3,
|
||||
Timeout: sTimeout * time.Second,
|
||||
CacheDuration: time.Second * 10,
|
||||
Auth: &auth,
|
||||
}
|
||||
|
||||
res, err := s.sendRequestToExternal(options)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("error making user_by_id request to qapix server. error: %w", err)
|
||||
|
||||
slog.Error(err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
err = json.Unmarshal(res, out)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("error unmarshalling user by id from qapix server. error: %w", err)
|
||||
|
||||
slog.Error(err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Manager) UserByEmail(email string, out *app.User) (err error) {
|
||||
path := "/api/v1/auth/data/user_by_email"
|
||||
|
||||
auth, ok := s.config.External[QApixService]
|
||||
if !ok {
|
||||
err = errors.New("error getting auth data for qapix service at UserByEmail")
|
||||
|
||||
slog.Error(err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
options := RequestOptions{
|
||||
Method: http.MethodPost,
|
||||
Path: path,
|
||||
Body: map[string]string{"Email": email},
|
||||
Retries: 3,
|
||||
Timeout: sTimeout * time.Second,
|
||||
CacheDuration: time.Second * 10,
|
||||
Auth: &auth,
|
||||
}
|
||||
|
||||
res, err := s.sendRequestToExternal(options)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("error making user_by_email request to qapix server. error: %w", err)
|
||||
|
||||
slog.Error(err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
err = json.Unmarshal(res, out)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("error unmarshalling user by email from qapix server. error: %w", err)
|
||||
|
||||
slog.Error(err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
72
src/client/store/manager.go
Normal file
72
src/client/store/manager.go
Normal file
@ -0,0 +1,72 @@
|
||||
// Package store defines database functions
|
||||
package store
|
||||
|
||||
import (
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"quantex.com.ar/multidb"
|
||||
"quantex.com/qfixdpl/src/app"
|
||||
"quantex.com/qfixdpl/src/client/store/external"
|
||||
"quantex.com/qfixdpl/src/common/tracerr"
|
||||
)
|
||||
|
||||
const dbPingSeconds = 30
|
||||
|
||||
type Store struct {
|
||||
db *multidb.MultiDB
|
||||
ext *external.Manager
|
||||
}
|
||||
|
||||
type Config struct {
|
||||
MultiDB multidb.Config
|
||||
External external.Config
|
||||
}
|
||||
|
||||
// New NewStore creates Store object
|
||||
func New(config Config) (*Store, error) {
|
||||
database, err := multidb.New("postgres", config.MultiDB)
|
||||
if err != nil {
|
||||
return nil, tracerr.Errorf("error trying to create multidb: %w", err)
|
||||
}
|
||||
|
||||
database.Start()
|
||||
|
||||
if err = database.Ping(); err != nil {
|
||||
return nil, tracerr.Errorf("error ping to database: %w", err)
|
||||
}
|
||||
|
||||
ext := external.NewManager(nil, config.External)
|
||||
|
||||
s := &Store{
|
||||
db: database,
|
||||
ext: ext,
|
||||
}
|
||||
|
||||
go s.db.PeriodicDBPing(time.Second * dbPingSeconds)
|
||||
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (p *Store) CloseDB() {
|
||||
p.db.Close()
|
||||
slog.Info("closing database connection.")
|
||||
}
|
||||
|
||||
func (p *Store) UserByEmail(email string) (out *app.User, err error) {
|
||||
var user app.User
|
||||
if err := p.ext.UserByEmail(email, &user); err != nil {
|
||||
return nil, tracerr.Errorf("error fetching user by email from external service: %w", err)
|
||||
}
|
||||
|
||||
return &user, nil
|
||||
}
|
||||
|
||||
func (p *Store) ValidateSession(sessionToken string) (bool, error) {
|
||||
ok, err := p.ext.ValidateSession(sessionToken)
|
||||
if err != nil {
|
||||
return false, tracerr.Errorf("error validating session: %w", err)
|
||||
}
|
||||
|
||||
return ok, nil
|
||||
}
|
||||
Reference in New Issue
Block a user