674 lines
17 KiB
Go
674 lines
17 KiB
Go
package execution
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"net/url"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"golang.org/x/net/websocket"
|
|
)
|
|
|
|
const (
|
|
TypeHTTP = "http"
|
|
TypeBatch = "batch"
|
|
TypeChain = "chain"
|
|
TypeLoadTest = "load-test"
|
|
TypeSSE = "sse"
|
|
TypeWebSocket = "websocket"
|
|
)
|
|
|
|
const (
|
|
StatusQueued = "queued"
|
|
StatusRunning = "running"
|
|
StatusSucceeded = "succeeded"
|
|
StatusFailed = "failed"
|
|
StatusCancelled = "cancelled"
|
|
)
|
|
|
|
type RequestItem struct {
|
|
Name string `json:"name,omitempty"`
|
|
Method string `json:"method"`
|
|
URL string `json:"url"`
|
|
Headers map[string]string `json:"headers,omitempty"`
|
|
Body string `json:"body,omitempty"`
|
|
}
|
|
|
|
type CreateRequest struct {
|
|
Type string `json:"type"`
|
|
Request *RequestItem `json:"request,omitempty"`
|
|
Requests []RequestItem `json:"requests,omitempty"`
|
|
FilePath string `json:"filePath,omitempty"`
|
|
Content string `json:"content,omitempty"`
|
|
Environment string `json:"environment,omitempty"`
|
|
Options map[string]any `json:"options,omitempty"`
|
|
}
|
|
|
|
type Execution struct {
|
|
ID string `json:"id"`
|
|
Type string `json:"type"`
|
|
Status string `json:"status"`
|
|
StartedAt string `json:"startedAt"`
|
|
CompletedAt *string `json:"completedAt"`
|
|
Result map[string]any `json:"result,omitempty"`
|
|
Children []Execution `json:"children,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
type ResponseResult struct {
|
|
StatusCode int `json:"statusCode"`
|
|
Status int `json:"status"`
|
|
Headers map[string]string `json:"headers"`
|
|
Body string `json:"body"`
|
|
DurationMS int64 `json:"durationMs"`
|
|
}
|
|
|
|
type Event struct {
|
|
Type string `json:"type"`
|
|
ExecutionID string `json:"executionId"`
|
|
Seq int64 `json:"seq"`
|
|
Timestamp string `json:"timestamp"`
|
|
Payload map[string]any `json:"payload"`
|
|
}
|
|
|
|
type Store struct {
|
|
mu sync.RWMutex
|
|
nextID atomic.Int64
|
|
client *http.Client
|
|
executions map[string]Execution
|
|
events map[string][]Event
|
|
order []string
|
|
}
|
|
|
|
func NewStore() *Store {
|
|
return &Store{
|
|
client: &http.Client{
|
|
Timeout: 30 * time.Second,
|
|
},
|
|
executions: map[string]Execution{},
|
|
events: map[string][]Event{},
|
|
}
|
|
}
|
|
|
|
func (s *Store) Create(request CreateRequest) Execution {
|
|
exec := Execution{
|
|
ID: s.nextExecutionID(),
|
|
Type: normalizeType(request.Type),
|
|
Status: StatusRunning,
|
|
StartedAt: now(),
|
|
Result: map[string]any{},
|
|
}
|
|
s.save(exec)
|
|
s.appendEvent(exec.ID, "execution.started", map[string]any{"type": exec.Type})
|
|
|
|
switch exec.Type {
|
|
case TypeHTTP:
|
|
exec = s.runHTTPExecution(exec, request.Request, request.Options)
|
|
case TypeBatch:
|
|
exec = s.runBatchExecution(exec, request.Requests)
|
|
case TypeChain:
|
|
exec = s.runChainExecution(exec, request.Requests)
|
|
case TypeLoadTest:
|
|
exec = s.runLoadTestExecution(exec, request.Request, request.Options)
|
|
case TypeSSE:
|
|
exec = s.runSSEExecution(exec, request.Request, request.Options)
|
|
case TypeWebSocket:
|
|
exec = s.runWebSocketExecution(exec, request.Request, request.Options)
|
|
default:
|
|
exec = fail(exec, fmt.Errorf("unsupported execution type %q", request.Type))
|
|
}
|
|
|
|
s.save(exec)
|
|
return exec
|
|
}
|
|
|
|
func (s *Store) Get(id string) (Execution, bool) {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
exec, ok := s.executions[id]
|
|
return exec, ok
|
|
}
|
|
|
|
func (s *Store) Events(id string) []Event {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
events := s.events[id]
|
|
copied := make([]Event, len(events))
|
|
copy(copied, events)
|
|
return copied
|
|
}
|
|
|
|
func (s *Store) AllEvents() []Event {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
var events []Event
|
|
for _, id := range s.order {
|
|
events = append(events, s.events[id]...)
|
|
}
|
|
return events
|
|
}
|
|
|
|
func (s *Store) History(limit int) []Execution {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
if limit <= 0 || limit > len(s.order) {
|
|
limit = len(s.order)
|
|
}
|
|
history := make([]Execution, 0, limit)
|
|
for index := len(s.order) - 1; index >= 0 && len(history) < limit; index-- {
|
|
history = append(history, s.executions[s.order[index]])
|
|
}
|
|
return history
|
|
}
|
|
|
|
func (s *Store) Cancel(id string) (Execution, bool) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
exec, ok := s.executions[id]
|
|
if !ok {
|
|
return Execution{}, false
|
|
}
|
|
if exec.CompletedAt == nil {
|
|
completed := now()
|
|
exec.Status = StatusCancelled
|
|
exec.CompletedAt = &completed
|
|
exec.Result = map[string]any{"cancelled": true}
|
|
s.executions[id] = exec
|
|
}
|
|
return exec, true
|
|
}
|
|
|
|
func (s *Store) runHTTPExecution(exec Execution, item *RequestItem, options map[string]any) Execution {
|
|
if item == nil {
|
|
return fail(exec, fmt.Errorf("request is required for http execution"))
|
|
}
|
|
|
|
response, err := s.send(*item)
|
|
if err != nil {
|
|
return fail(exec, err)
|
|
}
|
|
|
|
exec.Result = map[string]any{
|
|
"response": response,
|
|
"statusCode": response.StatusCode,
|
|
"status": response.StatusCode,
|
|
"headers": response.Headers,
|
|
"body": response.Body,
|
|
"durationMs": response.DurationMS,
|
|
"scriptLogs": scriptLogs(options),
|
|
}
|
|
assertions, err := evaluateAssertions(response, options)
|
|
exec.Result["assertions"] = assertions
|
|
if err != nil {
|
|
return fail(exec, err)
|
|
}
|
|
exec = succeed(exec)
|
|
s.appendEvent(exec.ID, "execution.response", map[string]any{"statusCode": response.StatusCode, "durationMs": response.DurationMS})
|
|
return exec
|
|
}
|
|
|
|
func (s *Store) runBatchExecution(exec Execution, requests []RequestItem) Execution {
|
|
if len(requests) == 0 {
|
|
return fail(exec, fmt.Errorf("requests are required for batch execution"))
|
|
}
|
|
|
|
children := make([]Execution, 0, len(requests))
|
|
for _, request := range requests {
|
|
child := Execution{
|
|
ID: s.nextExecutionID(),
|
|
Type: TypeHTTP,
|
|
Status: StatusRunning,
|
|
StartedAt: now(),
|
|
Result: map[string]any{},
|
|
}
|
|
child = s.runHTTPExecution(child, &request, nil)
|
|
children = append(children, child)
|
|
s.save(child)
|
|
}
|
|
|
|
exec.Children = children
|
|
exec.Result = map[string]any{"total": len(children), "succeeded": countStatus(children, StatusSucceeded)}
|
|
if countStatus(children, StatusFailed) > 0 {
|
|
return fail(exec, fmt.Errorf("%d batch request(s) failed", countStatus(children, StatusFailed)))
|
|
}
|
|
return succeed(exec)
|
|
}
|
|
|
|
func (s *Store) runChainExecution(exec Execution, requests []RequestItem) Execution {
|
|
if len(requests) == 0 {
|
|
return fail(exec, fmt.Errorf("requests are required for chain execution"))
|
|
}
|
|
|
|
variables := map[string]string{}
|
|
children := make([]Execution, 0, len(requests))
|
|
for index, request := range requests {
|
|
request.URL = replaceSimpleVariables(request.URL, variables)
|
|
request.Body = replaceSimpleVariables(request.Body, variables)
|
|
for key, value := range request.Headers {
|
|
request.Headers[key] = replaceSimpleVariables(value, variables)
|
|
}
|
|
|
|
child := Execution{
|
|
ID: s.nextExecutionID(),
|
|
Type: TypeHTTP,
|
|
Status: StatusRunning,
|
|
StartedAt: now(),
|
|
Result: map[string]any{},
|
|
}
|
|
child = s.runHTTPExecution(child, &request, nil)
|
|
children = append(children, child)
|
|
if child.Status != StatusSucceeded {
|
|
exec.Children = children
|
|
return fail(exec, fmt.Errorf("chain step %d failed", index+1))
|
|
}
|
|
if response, ok := child.Result["response"].(ResponseResult); ok {
|
|
variables[fmt.Sprintf("step%d.status", index+1)] = fmt.Sprintf("%d", response.StatusCode)
|
|
variables[fmt.Sprintf("step%d.body", index+1)] = response.Body
|
|
}
|
|
}
|
|
|
|
exec.Children = children
|
|
exec.Result = map[string]any{"total": len(children), "succeeded": len(children)}
|
|
return succeed(exec)
|
|
}
|
|
|
|
func (s *Store) runSSEExecution(exec Execution, item *RequestItem, options map[string]any) Execution {
|
|
if item == nil {
|
|
return fail(exec, fmt.Errorf("request is required for sse execution"))
|
|
}
|
|
request := *item
|
|
if request.Headers == nil {
|
|
request.Headers = map[string]string{}
|
|
}
|
|
request.Headers["Accept"] = "text/event-stream"
|
|
response, err := s.send(request)
|
|
if err != nil {
|
|
return fail(exec, err)
|
|
}
|
|
maxEvents := intOption(options, "maxEvents", 20)
|
|
events := parseSSEEvents(response.Body, maxEvents)
|
|
exec.Result = map[string]any{
|
|
"sessionType": "sse",
|
|
"statusCode": response.StatusCode,
|
|
"events": events,
|
|
"durationMs": response.DurationMS,
|
|
"bodySample": response.Body,
|
|
}
|
|
return succeed(exec)
|
|
}
|
|
|
|
func (s *Store) runWebSocketExecution(exec Execution, item *RequestItem, options map[string]any) Execution {
|
|
if item == nil {
|
|
return fail(exec, fmt.Errorf("request is required for websocket execution"))
|
|
}
|
|
messages := stringSliceOption(options, "messages")
|
|
timeout := time.Duration(intOption(options, "timeoutMs", 5000)) * time.Millisecond
|
|
if timeout <= 0 {
|
|
timeout = 5 * time.Second
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
|
defer cancel()
|
|
|
|
config, err := websocket.NewConfig(item.URL, websocketOrigin(item.URL))
|
|
if err != nil {
|
|
return fail(exec, err)
|
|
}
|
|
for key, value := range item.Headers {
|
|
config.Header.Set(key, value)
|
|
}
|
|
|
|
conn, err := config.DialContext(ctx)
|
|
if err != nil {
|
|
return fail(exec, err)
|
|
}
|
|
defer conn.Close()
|
|
_ = conn.SetDeadline(time.Now().Add(timeout))
|
|
|
|
received := []string{}
|
|
for _, message := range messages {
|
|
if err := websocket.Message.Send(conn, message); err != nil {
|
|
return fail(exec, err)
|
|
}
|
|
|
|
var response string
|
|
if err := websocket.Message.Receive(conn, &response); err != nil {
|
|
return fail(exec, err)
|
|
}
|
|
received = append(received, response)
|
|
}
|
|
|
|
exec.Result = map[string]any{
|
|
"sessionType": "websocket",
|
|
"url": item.URL,
|
|
"sent": messages,
|
|
"received": received,
|
|
}
|
|
return succeed(exec)
|
|
}
|
|
|
|
func (s *Store) runLoadTestExecution(exec Execution, item *RequestItem, options map[string]any) Execution {
|
|
if item == nil {
|
|
return fail(exec, fmt.Errorf("request is required for load-test execution"))
|
|
}
|
|
|
|
total := intOption(options, "requests", 1)
|
|
if total <= 0 {
|
|
total = 1
|
|
}
|
|
concurrency := intOption(options, "concurrency", 1)
|
|
if concurrency <= 0 {
|
|
concurrency = 1
|
|
}
|
|
if concurrency > total {
|
|
concurrency = total
|
|
}
|
|
|
|
started := time.Now()
|
|
jobs := make(chan int)
|
|
var succeeded atomic.Int64
|
|
var failed atomic.Int64
|
|
var maxDuration atomic.Int64
|
|
var totalDuration atomic.Int64
|
|
|
|
var wg sync.WaitGroup
|
|
for worker := 0; worker < concurrency; worker++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
for range jobs {
|
|
response, err := s.send(*item)
|
|
if err != nil || response.StatusCode >= 500 {
|
|
failed.Add(1)
|
|
continue
|
|
}
|
|
succeeded.Add(1)
|
|
totalDuration.Add(response.DurationMS)
|
|
for {
|
|
oldMax := maxDuration.Load()
|
|
if response.DurationMS <= oldMax || maxDuration.CompareAndSwap(oldMax, response.DurationMS) {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
for index := 0; index < total; index++ {
|
|
jobs <- index
|
|
}
|
|
close(jobs)
|
|
wg.Wait()
|
|
|
|
duration := time.Since(started).Milliseconds()
|
|
successCount := int(succeeded.Load())
|
|
avg := int64(0)
|
|
if successCount > 0 {
|
|
avg = totalDuration.Load() / int64(successCount)
|
|
}
|
|
|
|
exec.Result = map[string]any{
|
|
"totalRequests": total,
|
|
"succeeded": successCount,
|
|
"failed": int(failed.Load()),
|
|
"concurrency": concurrency,
|
|
"durationMs": duration,
|
|
"avgDurationMs": avg,
|
|
"maxDurationMs": maxDuration.Load(),
|
|
}
|
|
if failed.Load() > 0 {
|
|
return fail(exec, fmt.Errorf("%d load-test request(s) failed", failed.Load()))
|
|
}
|
|
return succeed(exec)
|
|
}
|
|
|
|
func (s *Store) send(item RequestItem) (ResponseResult, error) {
|
|
method := strings.ToUpper(strings.TrimSpace(item.Method))
|
|
if method == "" {
|
|
method = http.MethodGet
|
|
}
|
|
|
|
req, err := http.NewRequestWithContext(context.Background(), method, item.URL, bytes.NewBufferString(item.Body))
|
|
if err != nil {
|
|
return ResponseResult{}, err
|
|
}
|
|
for key, value := range item.Headers {
|
|
req.Header.Set(key, value)
|
|
}
|
|
|
|
started := time.Now()
|
|
resp, err := s.client.Do(req)
|
|
if err != nil {
|
|
return ResponseResult{}, err
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
body, err := io.ReadAll(io.LimitReader(resp.Body, 10*1024*1024))
|
|
if err != nil {
|
|
return ResponseResult{}, err
|
|
}
|
|
|
|
headers := map[string]string{}
|
|
for key, values := range resp.Header {
|
|
headers[key] = strings.Join(values, ", ")
|
|
}
|
|
|
|
return ResponseResult{
|
|
StatusCode: resp.StatusCode,
|
|
Status: resp.StatusCode,
|
|
Headers: headers,
|
|
Body: string(body),
|
|
DurationMS: time.Since(started).Milliseconds(),
|
|
}, nil
|
|
}
|
|
|
|
func (s *Store) nextExecutionID() string {
|
|
return fmt.Sprintf("exec_%d_%d", time.Now().UnixNano(), s.nextID.Add(1))
|
|
}
|
|
|
|
func (s *Store) save(exec Execution) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if _, exists := s.executions[exec.ID]; !exists {
|
|
s.order = append(s.order, exec.ID)
|
|
}
|
|
s.executions[exec.ID] = exec
|
|
}
|
|
|
|
func (s *Store) appendEvent(executionID, eventType string, payload map[string]any) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
seq := int64(len(s.events[executionID]))
|
|
s.events[executionID] = append(s.events[executionID], Event{
|
|
Type: eventType,
|
|
ExecutionID: executionID,
|
|
Seq: seq,
|
|
Timestamp: now(),
|
|
Payload: payload,
|
|
})
|
|
}
|
|
|
|
func normalizeType(value string) string {
|
|
value = strings.ToLower(strings.TrimSpace(value))
|
|
if value == "" {
|
|
return TypeHTTP
|
|
}
|
|
return value
|
|
}
|
|
|
|
func succeed(exec Execution) Execution {
|
|
completed := now()
|
|
exec.Status = StatusSucceeded
|
|
exec.CompletedAt = &completed
|
|
return exec
|
|
}
|
|
|
|
func fail(exec Execution, err error) Execution {
|
|
completed := now()
|
|
exec.Status = StatusFailed
|
|
exec.CompletedAt = &completed
|
|
exec.Error = err.Error()
|
|
if exec.Result == nil {
|
|
exec.Result = map[string]any{}
|
|
}
|
|
exec.Result["error"] = err.Error()
|
|
return exec
|
|
}
|
|
|
|
func now() string {
|
|
return time.Now().UTC().Format(time.RFC3339)
|
|
}
|
|
|
|
func countStatus(children []Execution, status string) int {
|
|
count := 0
|
|
for _, child := range children {
|
|
if child.Status == status {
|
|
count++
|
|
}
|
|
}
|
|
return count
|
|
}
|
|
|
|
func intOption(options map[string]any, key string, fallback int) int {
|
|
if options == nil {
|
|
return fallback
|
|
}
|
|
switch value := options[key].(type) {
|
|
case int:
|
|
return value
|
|
case int64:
|
|
return int(value)
|
|
case float64:
|
|
return int(value)
|
|
case json.Number:
|
|
parsed, err := value.Int64()
|
|
if err == nil {
|
|
return int(parsed)
|
|
}
|
|
case string:
|
|
var parsed int
|
|
if _, err := fmt.Sscanf(value, "%d", &parsed); err == nil {
|
|
return parsed
|
|
}
|
|
}
|
|
return fallback
|
|
}
|
|
|
|
func stringSliceOption(options map[string]any, key string) []string {
|
|
if options == nil {
|
|
return []string{}
|
|
}
|
|
switch values := options[key].(type) {
|
|
case []string:
|
|
return values
|
|
case []any:
|
|
messages := make([]string, 0, len(values))
|
|
for _, value := range values {
|
|
messages = append(messages, fmt.Sprint(value))
|
|
}
|
|
return messages
|
|
default:
|
|
return []string{}
|
|
}
|
|
}
|
|
|
|
func websocketOrigin(rawURL string) string {
|
|
parsed, err := url.Parse(rawURL)
|
|
if err != nil || parsed.Host == "" {
|
|
return "http://127.0.0.1"
|
|
}
|
|
scheme := "http"
|
|
if parsed.Scheme == "wss" {
|
|
scheme = "https"
|
|
}
|
|
return scheme + "://" + parsed.Host
|
|
}
|
|
|
|
func replaceSimpleVariables(input string, variables map[string]string) string {
|
|
output := input
|
|
for key, value := range variables {
|
|
output = strings.ReplaceAll(output, "{{"+key+"}}", value)
|
|
}
|
|
return output
|
|
}
|
|
|
|
func scriptLogs(options map[string]any) []string {
|
|
if options == nil {
|
|
return []string{}
|
|
}
|
|
logs := []string{}
|
|
if preScript, ok := options["preScript"].(string); ok && strings.TrimSpace(preScript) != "" {
|
|
logs = append(logs, "preScript accepted")
|
|
}
|
|
if postScript, ok := options["postScript"].(string); ok && strings.TrimSpace(postScript) != "" {
|
|
logs = append(logs, "postScript accepted")
|
|
}
|
|
return logs
|
|
}
|
|
|
|
func evaluateAssertions(response ResponseResult, options map[string]any) ([]map[string]any, error) {
|
|
assertions := []map[string]any{}
|
|
if options == nil {
|
|
return assertions, nil
|
|
}
|
|
if expectedStatus := intOption(options, "assertStatus", 0); expectedStatus > 0 {
|
|
passed := response.StatusCode == expectedStatus
|
|
assertions = append(assertions, map[string]any{
|
|
"name": "status",
|
|
"expected": expectedStatus,
|
|
"actual": response.StatusCode,
|
|
"passed": passed,
|
|
})
|
|
if !passed {
|
|
return assertions, fmt.Errorf("assertion failed: expected status %d, got %d", expectedStatus, response.StatusCode)
|
|
}
|
|
}
|
|
if contains, ok := options["assertBodyContains"].(string); ok && contains != "" {
|
|
passed := strings.Contains(response.Body, contains)
|
|
assertions = append(assertions, map[string]any{
|
|
"name": "body contains",
|
|
"expected": contains,
|
|
"passed": passed,
|
|
})
|
|
if !passed {
|
|
return assertions, fmt.Errorf("assertion failed: response body does not contain %q", contains)
|
|
}
|
|
}
|
|
return assertions, nil
|
|
}
|
|
|
|
func parseSSEEvents(body string, limit int) []map[string]string {
|
|
if limit <= 0 {
|
|
limit = 20
|
|
}
|
|
blocks := strings.Split(strings.ReplaceAll(body, "\r\n", "\n"), "\n\n")
|
|
events := []map[string]string{}
|
|
for _, block := range blocks {
|
|
if strings.TrimSpace(block) == "" {
|
|
continue
|
|
}
|
|
event := map[string]string{}
|
|
for _, line := range strings.Split(block, "\n") {
|
|
key, value, ok := strings.Cut(line, ":")
|
|
if !ok {
|
|
continue
|
|
}
|
|
event[strings.TrimSpace(key)] = strings.TrimSpace(value)
|
|
}
|
|
if len(event) > 0 {
|
|
events = append(events, event)
|
|
}
|
|
if len(events) >= limit {
|
|
break
|
|
}
|
|
}
|
|
return events
|
|
}
|