Michael McGuinness
116 lines
3.1 KiB
Go
116 lines
3.1 KiB
Go
package result
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
|
|
"queryorchestration/internal/database/repository"
|
|
resultprocessor "queryorchestration/internal/query/result/processor"
|
|
contextfull "queryorchestration/internal/query/types/contextFull"
|
|
jsonextractor "queryorchestration/internal/query/types/jsonExtractor"
|
|
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
type Process struct {
|
|
DocumentID uuid.UUID
|
|
QueryID uuid.UUID
|
|
QueryVersion int32
|
|
}
|
|
|
|
func (s *Service) Process(ctx context.Context, p *Process) (resultprocessor.Value, error) {
|
|
slog.Debug("processing query", "query_id", p.QueryID.String(), "version", p.QueryVersion, "document_id", p.DocumentID.String())
|
|
|
|
query, err := s.svc.Query.GetWithVersion(ctx, p.QueryID, p.QueryVersion)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
processQuery := &resultprocessor.Query{
|
|
ID: query.ID,
|
|
Version: p.QueryVersion,
|
|
Type: query.Type,
|
|
RequiredQueryIDs: query.RequiredQueryIDs,
|
|
Config: query.Config,
|
|
}
|
|
|
|
values, err := s.listRequiredValues(ctx, p, processQuery)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if s.cfg.GetLogLevel() == slog.LevelDebug {
|
|
elength := 0
|
|
if processQuery.RequiredQueryIDs != nil {
|
|
elength = len(*processQuery.RequiredQueryIDs)
|
|
}
|
|
slog.Info("requirements", "length", len(values), "expected_length", elength)
|
|
}
|
|
|
|
processor, err := s.getProcessor(query.Type)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
val, err := processor.Process(ctx, processQuery, values)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return getValueByType(query.Type, val)
|
|
}
|
|
|
|
func (s *Service) listRequiredValues(ctx context.Context, p *Process, query *resultprocessor.Query) ([]resultprocessor.Value, error) {
|
|
if p == nil || query == nil || query.RequiredQueryIDs == nil || len(*query.RequiredQueryIDs) == 0 {
|
|
return nil, nil
|
|
}
|
|
|
|
qResults, err := s.cfg.GetDBQueries().ListQueryRequirementValues(ctx, &repository.ListQueryRequirementValuesParams{
|
|
Queryid: &p.QueryID,
|
|
Documentid: &p.DocumentID,
|
|
Version: &p.QueryVersion,
|
|
})
|
|
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
|
return nil, err
|
|
} else if len(qResults) != len(*query.RequiredQueryIDs) {
|
|
return nil, fmt.Errorf("required results not found for the query (%s) in document (%s)", p.QueryID, p.DocumentID)
|
|
}
|
|
|
|
slog.Debug("found results", "length", len(qResults))
|
|
|
|
return parseQueryRequirementValueArray(qResults)
|
|
}
|
|
|
|
func parseQueryRequirementValueArray(v []*repository.ListQueryRequirementValuesRow) ([]resultprocessor.Value, error) {
|
|
values := make([]resultprocessor.Value, len(v))
|
|
for index, r := range v {
|
|
qType, err := resultprocessor.ParseDBType(r.Querytype)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cleanValue, err := getValueByType(qType, *r.Value)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
values[index] = cleanValue
|
|
}
|
|
|
|
return values, nil
|
|
}
|
|
|
|
func (s *Service) getProcessor(queryType resultprocessor.Type) (resultprocessor.Processor, error) {
|
|
switch queryType {
|
|
case resultprocessor.TypeJsonExtractor:
|
|
return jsonextractor.NewExtractor(s.cfg), nil
|
|
case resultprocessor.TypeContextFull:
|
|
return contextfull.NewExtractor(), nil
|
|
default:
|
|
return nil, fmt.Errorf("attempting to process invalid query type")
|
|
}
|
|
}
|