232 lines
7.5 KiB
Go
232 lines
7.5 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"os"
|
|
"path/filepath"
|
|
"time"
|
|
|
|
"git.ipao.vip/rogee/go-sip/internal/contract"
|
|
)
|
|
|
|
type envelope struct {
|
|
EventID string `json:"event_id"`
|
|
EventType string `json:"event_type"`
|
|
DispatcherID string `json:"dispatcher_id"`
|
|
TenantID int64 `json:"tenant_id,omitempty"`
|
|
IssuedAt string `json:"issued_at,omitempty"`
|
|
Payload map[string]any `json:"payload"`
|
|
}
|
|
|
|
type commandInput struct {
|
|
EventID string `json:"event_id"`
|
|
TaskID string `json:"task_id,omitempty"`
|
|
Callee string `json:"callee,omitempty"`
|
|
Action string `json:"action,omitempty"`
|
|
Reason string `json:"reason,omitempty"`
|
|
TrunkID string `json:"trunk_id,omitempty"`
|
|
ChangeType string `json:"change_type,omitempty"`
|
|
Revision *int64 `json:"revision,omitempty"`
|
|
}
|
|
|
|
type requestRecord struct {
|
|
Input commandInput `json:"-"`
|
|
Digest string `json:"-"`
|
|
EventID string `json:"event_id"`
|
|
EventType string `json:"event_type"`
|
|
Status string `json:"status"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
type feedbackRecord struct {
|
|
EventID string `json:"event_id"`
|
|
EventType string `json:"event_type"`
|
|
DispatcherID string `json:"dispatcher_id"`
|
|
Status string `json:"status,omitempty"`
|
|
Outcome string `json:"outcome,omitempty"`
|
|
Validation string `json:"validation"`
|
|
Error string `json:"error,omitempty"`
|
|
Duplicates int `json:"duplicates"`
|
|
Digest string `json:"-"`
|
|
}
|
|
|
|
func digest(raw []byte) string { sum := sha256.Sum256(raw); return hex.EncodeToString(sum[:]) }
|
|
func messageKey(dispatcherID, eventType, eventID string) string {
|
|
raw, _ := json.Marshal([]string{dispatcherID, eventType, eventID})
|
|
return string(raw)
|
|
}
|
|
|
|
func (s *service) send(ctx context.Context, kind string, input commandInput, now time.Time) (requestRecord, error) {
|
|
s.eventMu.Lock()
|
|
defer s.eventMu.Unlock()
|
|
s.mu.RLock()
|
|
data := s.data
|
|
s.mu.RUnlock()
|
|
msg := envelope{EventID: input.EventID, EventType: kind, DispatcherID: data.dispatcherID, Payload: make(map[string]any)}
|
|
route := "d." + data.dispatcherID + ".control.in"
|
|
switch kind {
|
|
case "call.execute", "task.control":
|
|
if _, ok := data.tasks[input.TaskID]; !ok {
|
|
return requestRecord{}, errors.New("unknown task")
|
|
}
|
|
msg.TenantID = data.tenantID
|
|
msg.IssuedAt = now.UTC().Format(time.RFC3339Nano)
|
|
msg.Payload["task_id"] = input.TaskID
|
|
if kind == "call.execute" {
|
|
msg.Payload["callee"] = input.Callee
|
|
route = "d." + data.dispatcherID + ".task." + input.TaskID + ".in"
|
|
} else {
|
|
if input.Action != "start" && input.Action != "resume" && input.Action != "edit" {
|
|
return requestRecord{}, errors.New("only start, resume and edit are supported")
|
|
}
|
|
msg.Payload["action"] = input.Action
|
|
msg.Payload["reason"] = input.Reason
|
|
}
|
|
case "sip.config":
|
|
msg.Payload["trunk_id"] = input.TrunkID
|
|
msg.Payload["change_type"] = input.ChangeType
|
|
if input.Revision != nil {
|
|
msg.Payload["revision"] = *input.Revision
|
|
}
|
|
default:
|
|
return requestRecord{}, errors.New("unsupported command")
|
|
}
|
|
raw, err := json.Marshal(msg)
|
|
if err != nil {
|
|
return requestRecord{}, err
|
|
}
|
|
if err = contract.ValidateCurrent("mq", raw); err != nil {
|
|
return requestRecord{}, fmt.Errorf("command: %w", err)
|
|
}
|
|
key := messageKey(data.dispatcherID, kind, input.EventID)
|
|
inputRaw, _ := json.Marshal(input)
|
|
hash := digest(inputRaw)
|
|
if previous, ok := s.requests[key]; ok {
|
|
if previous.Digest != hash {
|
|
return requestRecord{}, errors.New("event identity already used with different content")
|
|
}
|
|
return previous, nil
|
|
}
|
|
if s.publish == nil {
|
|
return requestRecord{}, errors.New("MQ publisher unavailable")
|
|
}
|
|
record := requestRecord{Input: input, Digest: hash, EventID: input.EventID, EventType: kind, Status: "unknown"}
|
|
s.requests[key] = record
|
|
slog.Info("Test command publishing", "event_id", input.EventID, "event_type", kind, "stage", "publish_requested")
|
|
status, publishErr := s.publish(ctx, route, raw)
|
|
record.Status = status
|
|
if publishErr != nil {
|
|
record.Error = publishErr.Error()
|
|
}
|
|
s.requests[key] = record
|
|
slog.Info("Test command publish outcome", "event_id", input.EventID, "event_type", kind, "status", record.Status, "error", record.Error)
|
|
return record, nil
|
|
}
|
|
|
|
func (s *service) receive(raw []byte, ack func() error) error {
|
|
s.eventMu.Lock()
|
|
defer s.eventMu.Unlock()
|
|
if err := saveFeedback(s.feedbackDir, raw); err != nil {
|
|
return err
|
|
}
|
|
s.mu.RLock()
|
|
dispatcherID, tenantID := s.data.dispatcherID, s.data.tenantID
|
|
s.mu.RUnlock()
|
|
var msg envelope
|
|
record := feedbackRecord{Validation: "valid", Digest: digest(raw)}
|
|
err := json.Unmarshal(raw, &msg)
|
|
record.EventID = msg.EventID
|
|
record.EventType = msg.EventType
|
|
record.DispatcherID = msg.DispatcherID
|
|
schemaErr := contract.ValidateCurrent("mq", raw)
|
|
// The shared schema validates these enum fields; do not maintain another value list.
|
|
if err == nil && schemaErr == nil {
|
|
record.Status, _ = msg.Payload["status"].(string)
|
|
record.Outcome, _ = msg.Payload["outcome"].(string)
|
|
}
|
|
invalid := func(reason string) { record.Validation = "invalid"; record.Error = reason }
|
|
if err != nil || schemaErr != nil {
|
|
invalid("feedback violates current MQ contract")
|
|
} else if msg.DispatcherID != dispatcherID {
|
|
invalid("feedback belongs to another dispatcher")
|
|
} else {
|
|
requestKind := msg.EventType
|
|
if msg.EventType == "call.execute.result" {
|
|
requestKind = "call.execute"
|
|
} else if (msg.EventType != "call.execute" && msg.EventType != "task.control" && msg.EventType != "sip.config") || record.Status == "" {
|
|
invalid("message is not an outbound receipt")
|
|
}
|
|
request, ok := s.requests[messageKey(msg.DispatcherID, requestKind, msg.EventID)]
|
|
if !ok {
|
|
invalid("feedback has no request in this process")
|
|
} else {
|
|
if msg.TenantID != 0 && msg.TenantID != tenantID {
|
|
invalid("feedback tenant mismatch")
|
|
}
|
|
for field, want := range map[string]string{"task_id": request.Input.TaskID, "callee": request.Input.Callee, "action": request.Input.Action} {
|
|
if value, exists := msg.Payload[field]; exists && value != want {
|
|
invalid("feedback request association mismatch")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
for i, previous := range s.feedback {
|
|
if messageKey(previous.DispatcherID, previous.EventType, previous.EventID) != messageKey(record.DispatcherID, record.EventType, record.EventID) {
|
|
continue
|
|
}
|
|
if previous.Digest == record.Digest {
|
|
s.feedback[i].Duplicates++
|
|
return ack()
|
|
}
|
|
invalid("conflicting feedback for same identity")
|
|
}
|
|
s.feedback = append(s.feedback, record)
|
|
slog.Info("Test feedback saved and validated", "event_id", record.EventID, "event_type", record.EventType, "validation", record.Validation, "error", record.Error)
|
|
return ack()
|
|
}
|
|
|
|
func saveFeedback(dir string, raw []byte) error {
|
|
f, err := os.CreateTemp(dir, "feedback-*.json")
|
|
if err != nil {
|
|
return fmt.Errorf("create feedback evidence: %w", err)
|
|
}
|
|
if _, err = f.Write(raw); err == nil {
|
|
err = f.Sync()
|
|
}
|
|
closeErr := f.Close()
|
|
if err == nil {
|
|
err = closeErr
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("save feedback evidence: %w", err)
|
|
}
|
|
parent, err := os.Open(filepath.Dir(f.Name()))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
syncErr := parent.Sync()
|
|
closeErr = parent.Close()
|
|
if syncErr != nil {
|
|
return syncErr
|
|
}
|
|
return closeErr
|
|
}
|
|
|
|
func (s *service) events(id string) []feedbackRecord {
|
|
s.eventMu.Lock()
|
|
defer s.eventMu.Unlock()
|
|
records := []feedbackRecord{}
|
|
for _, record := range s.feedback {
|
|
if record.EventID == id {
|
|
records = append(records, record)
|
|
}
|
|
}
|
|
return records
|
|
}
|