diff --git a/core/cmd/mildstack/main.go b/core/cmd/mildstack/main.go index 9d4aae5..6b11758 100644 --- a/core/cmd/mildstack/main.go +++ b/core/cmd/mildstack/main.go @@ -45,6 +45,9 @@ func main() { if err := registerNativeS3Routes(router, root.Services); err != nil { return recordingHTTPServer{server: failedHTTPServer{err: err}, storage: storage, port: port, instanceID: instanceID} } + if err := registerNativeSQSRoutes(router, root.Services); err != nil { + return recordingHTTPServer{server: failedHTTPServer{err: err}, storage: storage, port: port, instanceID: instanceID} + } registrar := instanceRegistrar{manager: manager, storage: storage, instanceID: instanceID} return recordingHTTPServer{server: deliveryhttp.NewServer(registrar, router, port), storage: storage, port: port, instanceID: instanceID} } @@ -178,6 +181,27 @@ func registerNativeDynamoDBRoutes(router *deliveryhttp.Router, services []orches return nil } +func registerNativeSQSRoutes(router *deliveryhttp.Router, services []orchestrator.Service) error { + if router == nil { + return nil + } + + for _, service := range services { + if service == nil || service.Metadata().Name != "sqs" { + continue + } + + sqsService, ok := service.(deliveryhttp.SQSNativeService) + if !ok { + return fmt.Errorf("sqs service does not expose the native http surface") + } + deliveryhttp.RegisterSQSNativeRoutes(router.Engine(), sqsService) + return nil + } + + return nil +} + func containsPort(ports []int, port int) bool { for _, existing := range ports { if existing == port { diff --git a/core/cmd/mildstack/main_test.go b/core/cmd/mildstack/main_test.go index 241bf26..5d39a09 100644 --- a/core/cmd/mildstack/main_test.go +++ b/core/cmd/mildstack/main_test.go @@ -1,9 +1,11 @@ package main import ( + "bytes" "context" "errors" "fmt" + "io" "net/http" "net/http/httptest" "os" @@ -17,6 +19,7 @@ import ( "github.com/aws/aws-sdk-go-v2/credentials" "github.com/aws/aws-sdk-go-v2/service/dynamodb" "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + sqssdk "github.com/aws/aws-sdk-go-v2/service/sqs" "github.com/michasdev/mildstack/core/internal/application/orchestrator" "github.com/michasdev/mildstack/core/internal/application/runtime" "github.com/michasdev/mildstack/core/internal/composition" @@ -688,6 +691,93 @@ func TestInstanceRegistrarServeSkipsDuplicateLoadedPort(t *testing.T) { } } +func TestRegisterNativeSQSRoutesExposesAwsCompatibleSmokeSurface(t *testing.T) { + t.Helper() + + root := composition.DefaultRoot("test-instance") + manager := runtime.New(root.Services) + router := deliveryhttp.NewRouter(deliveryhttp.DefaultConfig(), manager) + + if err := registerNativeSQSRoutes(router, root.Services); err != nil { + t.Fatalf("register native sqs routes: %v", err) + } + + healthRecorder := httptest.NewRecorder() + healthRequest := httptest.NewRequest(http.MethodGet, "/api/v1/runtime/health", nil) + router.Engine().ServeHTTP(healthRecorder, healthRequest) + if got, want := healthRecorder.Code, http.StatusOK; got != want { + t.Fatalf("unexpected health status: got %d want %d", got, want) + } + + rootRecorder := httptest.NewRecorder() + rootRequest := httptest.NewRequest(http.MethodGet, "/?Action=ListQueues&Version=2012-11-05", nil) + router.Engine().ServeHTTP(rootRecorder, rootRequest) + if got, want := rootRecorder.Code, http.StatusBadRequest; got != want { + t.Fatalf("unexpected sqs root status: got %d want %d", got, want) + } + if !strings.Contains(rootRecorder.Body.String(), "") { + t.Fatalf("expected sqs error response xml, got %q", rootRecorder.Body.String()) + } + if !strings.Contains(rootRecorder.Body.String(), "UnsupportedOperation") { + t.Fatalf("expected unsupported operation xml, got %q", rootRecorder.Body.String()) + } + + server := httptest.NewServer(router.Engine()) + t.Cleanup(server.Close) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + t.Cleanup(cancel) + + transport := &captureTransport{base: http.DefaultTransport} + cfg, err := config.LoadDefaultConfig(ctx, + config.WithRegion("us-east-1"), + config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test", "test", "test")), + config.WithHTTPClient(&http.Client{Transport: transport}), + ) + if err != nil { + t.Fatalf("load aws config: %v", err) + } + + client := sqssdk.NewFromConfig(cfg, func(o *sqssdk.Options) { + o.BaseEndpoint = aws.String(server.URL) + }) + + _, err = client.ListQueues(ctx, &sqssdk.ListQueuesInput{}) + if err == nil { + t.Fatal("expected list queues to return an error") + } + if !strings.Contains(string(transport.body), "") { + t.Fatalf("expected captured sqs xml body, got %q", string(transport.body)) + } + if !strings.Contains(string(transport.body), "UnsupportedOperation") && !strings.Contains(string(transport.body), "InvalidQueryParameter") { + t.Fatalf("expected captured sqs xml body to contain an SQS error code, got %q", string(transport.body)) + } +} + +type captureTransport struct { + base http.RoundTripper + body []byte +} + +func (t *captureTransport) RoundTrip(req *http.Request) (*http.Response, error) { + resp, err := t.base.RoundTrip(req) + if err != nil { + return nil, err + } + if resp.Body == nil { + return resp, nil + } + + data, readErr := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if readErr != nil { + return nil, readErr + } + t.body = append([]byte(nil), data...) + resp.Body = io.NopCloser(bytes.NewReader(data)) + return resp, nil +} + func newDynamoDBSmokeClient(t *testing.T, endpoint string) *dynamodb.Client { t.Helper() diff --git a/core/internal/composition/default_root.go b/core/internal/composition/default_root.go index 60ab696..342bd56 100644 --- a/core/internal/composition/default_root.go +++ b/core/internal/composition/default_root.go @@ -9,12 +9,14 @@ import ( "github.com/michasdev/mildstack/core/internal/application/runtime" "github.com/michasdev/mildstack/core/internal/resources/dynamodb" "github.com/michasdev/mildstack/core/internal/resources/s3" + "github.com/michasdev/mildstack/core/internal/resources/sqs" ) type DefaultRootConfig struct { InstanceID string S3StorageBaseDir string DynamoDBStorageBaseDir string + SQSStorageBaseDir string } func DefaultRoot(instanceID string) Root { @@ -48,7 +50,17 @@ func defaultRootWithHook(hook orchestrator.StateHook, config DefaultRootConfig) panic(fmt.Sprintf("composition: init dynamodb service: %v", err)) } - services := []orchestrator.Service{s3Service, dynamoService} + sqsService, err := sqs.NewWithStorage(sqs.StorageConfig{ + BaseDir: config.SQSStorageBaseDir, + InstanceID: instanceID, + }) + if err != nil { + _ = s3Service.Stop(context.Background()) + _ = dynamoService.Stop(context.Background()) + panic(fmt.Sprintf("composition: init sqs service: %v", err)) + } + + services := []orchestrator.Service{s3Service, dynamoService, sqsService} for _, service := range services { if err := service.AttachState(hook); err != nil { for _, candidate := range services { diff --git a/core/internal/composition/default_root_test.go b/core/internal/composition/default_root_test.go index a203344..c825a21 100644 --- a/core/internal/composition/default_root_test.go +++ b/core/internal/composition/default_root_test.go @@ -9,6 +9,7 @@ import ( dynamodbapp "github.com/michasdev/mildstack/core/internal/resources/dynamodb/application" dynamodbdomain "github.com/michasdev/mildstack/core/internal/resources/dynamodb/domain" s3domain "github.com/michasdev/mildstack/core/internal/resources/s3/domain" + sqsdomain "github.com/michasdev/mildstack/core/internal/resources/sqs/domain" ) type stateHookStub struct { @@ -36,19 +37,24 @@ func TestDefaultRootIncludesS3AndDynamoDBWithDeterministicRoutes(t *testing.T) { InstanceID: "test-instance", S3StorageBaseDir: baseDir, DynamoDBStorageBaseDir: baseDir, + SQSStorageBaseDir: baseDir, }) - if got, want := len(root.Services), 2; got != want { + if got, want := len(root.Services), 3; got != want { t.Fatalf("unexpected service count: got %d want %d", got, want) } first := root.Services[0] second := root.Services[1] + third := root.Services[2] if got, want := first.Metadata().Name, "s3"; got != want { t.Fatalf("unexpected first service name: got %q want %q", got, want) } if got, want := second.Metadata().Name, "dynamodb"; got != want { t.Fatalf("unexpected second service name: got %q want %q", got, want) } + if got, want := third.Metadata().Name, "sqs"; got != want { + t.Fatalf("unexpected third service name: got %q want %q", got, want) + } registrar := deliveryhttp.NewRegistrar() for _, service := range root.Services { @@ -58,7 +64,7 @@ func TestDefaultRootIncludesS3AndDynamoDBWithDeterministicRoutes(t *testing.T) { } entries := registrar.Services() - if got, want := len(entries), 2; got != want { + if got, want := len(entries), 3; got != want { t.Fatalf("unexpected catalog size: got %d want %d", got, want) } if got, want := entries[0].Name, "dynamodb"; got != want { @@ -67,6 +73,9 @@ func TestDefaultRootIncludesS3AndDynamoDBWithDeterministicRoutes(t *testing.T) { if got, want := entries[1].Name, "s3"; got != want { t.Fatalf("unexpected second catalog service: got %q want %q", got, want) } + if got, want := entries[2].Name, "sqs"; got != want { + t.Fatalf("unexpected third catalog service: got %q want %q", got, want) + } s3Entry, ok := registrar.Service("s3") if !ok { @@ -125,6 +134,9 @@ func TestDefaultRootIncludesS3AndDynamoDBWithDeterministicRoutes(t *testing.T) { if _, ok := root.Services[1].(deliveryhttp.DynamoDBNativeService); !ok { t.Fatal("expected dynamodb service to expose the native http surface") } + if _, ok := root.Services[2].(deliveryhttp.SQSNativeService); !ok { + t.Fatal("expected sqs service to expose the native http surface") + } if value, ok := hook.Get(dynamodbdomain.StateKey); !ok { t.Fatalf("expected state for %q to be present", dynamodbdomain.StateKey) @@ -144,6 +156,18 @@ func TestDefaultRootIncludesS3AndDynamoDBWithDeterministicRoutes(t *testing.T) { t.Fatalf("unexpected s3 state: got %v want %v", got, want) } + if value, ok := hook.Get(sqsdomain.StateKey); !ok { + t.Fatalf("expected state for %q to be present", sqsdomain.StateKey) + } else { + state := value.(map[string]any) + if got, want := state["service"], "sqs"; got != want { + t.Fatalf("unexpected sqs state: got %v want %v", got, want) + } + if got, want := len(state["queues"].([]any)), 0; got != want { + t.Fatalf("unexpected sqs queue count: got %d want %d", got, want) + } + } + dynamoDBPath := filepath.Join(baseDir, "instances", "test-instance", "dynamodb", "state.db") if _, err := os.Stat(dynamoDBPath); err != nil { t.Fatalf("expected dynamodb database to exist at %s: %v", dynamoDBPath, err) @@ -186,8 +210,9 @@ func TestDefaultRootUsesInstanceScopedDynamoDBStorage(t *testing.T) { InstanceID: "instance-a", S3StorageBaseDir: baseDir, DynamoDBStorageBaseDir: baseDir, + SQSStorageBaseDir: baseDir, }) - if got, want := len(root.Services), 2; got != want { + if got, want := len(root.Services), 3; got != want { t.Fatalf("unexpected service count: got %d want %d", got, want) } diff --git a/core/internal/delivery/http/sqs_native.go b/core/internal/delivery/http/sqs_native.go new file mode 100644 index 0000000..58dc1f4 --- /dev/null +++ b/core/internal/delivery/http/sqs_native.go @@ -0,0 +1,104 @@ +package http + +import ( + "errors" + "net/http" + "strings" + + "github.com/gin-gonic/gin" + "github.com/michasdev/mildstack/core/internal/application/orchestrator" +) + +type SQSNativeService interface { + Policy() orchestrator.EmulationPolicy + Metadata() orchestrator.Metadata +} + +func RegisterSQSNativeRoutes(engine *gin.Engine, service SQSNativeService) { + if engine == nil || service == nil { + return + } + + handler := newSQSNativeHandler(service) + engine.Use(func(c *gin.Context) { + if handled := handler.dispatch(c); handled { + c.Abort() + return + } + c.Next() + }) +} + +type sqsNativeHandler struct { + service SQSNativeService + registry SQSRegistry + supported map[string]struct{} +} + +func newSQSNativeHandler(service SQSNativeService) sqsNativeHandler { + supported := make(map[string]struct{}) + if service != nil { + for _, action := range service.Policy().Supported { + supported[action] = struct{}{} + } + } + + return sqsNativeHandler{ + service: service, + registry: NewSQSRegistry(), + supported: supported, + } +} + +func (h sqsNativeHandler) dispatch(c *gin.Context) bool { + if c == nil || c.Request == nil || c.Request.URL == nil { + return false + } + + path := strings.TrimSpace(c.Request.URL.Path) + if path == "" || strings.HasPrefix(path, "/api/") { + return false + } + switch c.Request.Method { + case http.MethodGet, http.MethodPost: + default: + return false + } + + ctx, err := ParseSQSRequest(c.Request) + if err != nil { + if errors.Is(err, ErrSQSNotOwned) { + return false + } + writeSQSError(c, err, requestIDFromContext(c)) + return true + } + + spec, err := h.registry.Resolve(ctx) + if err != nil { + writeSQSError(c, err, requestIDFromContext(c)) + return true + } + + if _, ok := h.supported[spec.Action]; !ok || spec.DomainDeferred { + writeSQSError(c, ErrSQSUnsupported, requestIDFromContext(c)) + return true + } + + writeSQSError(c, ErrSQSUnsupported, requestIDFromContext(c)) + return true +} + +func requestIDFromContext(c *gin.Context) string { + if c == nil { + return "mildstack-sqs-request" + } + + for _, key := range []string{"x-amzn-requestid", "X-Amzn-RequestId", "x-amz-request-id"} { + if requestID := strings.TrimSpace(c.GetHeader(key)); requestID != "" { + return requestID + } + } + + return "mildstack-sqs-request" +} diff --git a/core/internal/delivery/http/sqs_native_contract.go b/core/internal/delivery/http/sqs_native_contract.go new file mode 100644 index 0000000..8942694 --- /dev/null +++ b/core/internal/delivery/http/sqs_native_contract.go @@ -0,0 +1,178 @@ +package http + +import ( + "errors" + "fmt" + "mime" + "net/http" + "net/url" + "path" + "strings" + + "github.com/michasdev/mildstack/core/internal/resources/sqs/contracts" +) + +const sqsQueryVersion = "2012-11-05" + +var ( + ErrSQSNotOwned = errors.New("sqs: request is not owned by the SQS native adapter") + ErrSQSMalformedRequest = errors.New("sqs: malformed query request") + ErrSQSInvalidAction = errors.New("sqs: invalid action") + ErrSQSMissingAction = errors.New("sqs: missing action") + ErrSQSInvalidVersion = errors.New("sqs: invalid version") + ErrSQSQueuePathMismatch = errors.New("sqs: queue path mismatch") + ErrSQSUnsupported = errors.New("sqs: unsupported action") +) + +type SQSRequestKind string + +const ( + SQSRequestKindRoot SQSRequestKind = "root" + SQSRequestKindQueue SQSRequestKind = "queue" +) + +type SQSRequestContext struct { + Method string + RawPath string + NormalizedPath string + Kind SQSRequestKind + AccountID string + QueueName string + Action string + Version string + Values url.Values +} + +func (c SQSRequestContext) QueueScoped() bool { + return c.Kind == SQSRequestKindQueue +} + +func ParseSQSRequest(req *http.Request) (SQSRequestContext, error) { + if req == nil || req.URL == nil { + return SQSRequestContext{}, ErrSQSMalformedRequest + } + + pathContext, err := classifySQSPath(req.URL.Path) + if err != nil { + return SQSRequestContext{}, err + } + if pathContext.Kind == "" { + return SQSRequestContext{}, ErrSQSNotOwned + } + + if req.Method == http.MethodPost { + mediaType, _, err := mime.ParseMediaType(req.Header.Get("Content-Type")) + if err != nil { + return SQSRequestContext{}, fmt.Errorf("%w: %v", ErrSQSMalformedRequest, err) + } + if mediaType != "" && mediaType != "application/x-www-form-urlencoded" { + return SQSRequestContext{}, ErrSQSMalformedRequest + } + } + + if err := req.ParseForm(); err != nil { + return SQSRequestContext{}, fmt.Errorf("%w: %v", ErrSQSMalformedRequest, err) + } + + action := strings.TrimSpace(req.Form.Get("Action")) + if action == "" { + return SQSRequestContext{}, ErrSQSMissingAction + } + version := strings.TrimSpace(req.Form.Get("Version")) + if version == "" { + return SQSRequestContext{}, ErrSQSInvalidVersion + } + if version != sqsQueryVersion { + return SQSRequestContext{}, ErrSQSInvalidVersion + } + + return SQSRequestContext{ + Method: strings.ToUpper(strings.TrimSpace(req.Method)), + RawPath: normalizeRequestPath(req.URL.Path), + NormalizedPath: pathContext.NormalizedPath, + Kind: pathContext.Kind, + AccountID: pathContext.AccountID, + QueueName: pathContext.QueueName, + Action: action, + Version: version, + Values: cloneValues(req.Form), + }, nil +} + +type sqsPathContext struct { + Kind SQSRequestKind + NormalizedPath string + AccountID string + QueueName string +} + +func classifySQSPath(rawPath string) (sqsPathContext, error) { + trimmed := normalizeRequestPath(rawPath) + if trimmed == "" || trimmed == "/" { + return sqsPathContext{Kind: SQSRequestKindRoot, NormalizedPath: "/"}, nil + } + if strings.HasPrefix(trimmed, "/api/") { + return sqsPathContext{}, ErrSQSNotOwned + } + + segments := strings.Split(strings.Trim(trimmed, "/"), "/") + if len(segments) != 2 { + return sqsPathContext{}, ErrSQSNotOwned + } + if segments[0] == "" || segments[1] == "" { + return sqsPathContext{}, ErrSQSQueuePathMismatch + } + + return sqsPathContext{ + Kind: SQSRequestKindQueue, + NormalizedPath: "/" + strings.Join(segments, "/"), + AccountID: segments[0], + QueueName: segments[1], + }, nil +} + +func normalizeRequestPath(rawPath string) string { + trimmed := strings.TrimSpace(rawPath) + if trimmed == "" { + return "/" + } + if !strings.HasPrefix(trimmed, "/") { + trimmed = "/" + trimmed + } + return path.Clean(trimmed) +} + +func cloneValues(values url.Values) url.Values { + if values == nil { + return nil + } + + cloned := make(url.Values, len(values)) + for key, list := range values { + cloned[key] = append([]string(nil), list...) + } + return cloned +} + +func validateSQSRequestContext(ctx SQSRequestContext, spec SQSRegistrySpec) error { + if spec.Action == "" { + return ErrSQSInvalidAction + } + if ctx.Action != spec.Action { + return ErrSQSInvalidAction + } + if ctx.Version != spec.Version { + return ErrSQSInvalidVersion + } + switch spec.Scope { + case contracts.ScopeRoot: + if ctx.Kind != SQSRequestKindRoot { + return ErrSQSQueuePathMismatch + } + case contracts.ScopeQueue: + if ctx.Kind != SQSRequestKindQueue { + return ErrSQSQueuePathMismatch + } + } + return nil +} diff --git a/core/internal/delivery/http/sqs_native_contract_test.go b/core/internal/delivery/http/sqs_native_contract_test.go new file mode 100644 index 0000000..6716577 --- /dev/null +++ b/core/internal/delivery/http/sqs_native_contract_test.go @@ -0,0 +1,95 @@ +package http + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestSQSNativeContractParsesQueryAndFormValues(t *testing.T) { + t.Helper() + + req := httptest.NewRequest(http.MethodPost, "/123456789012/orders/", strings.NewReader( + "Action=SendMessage&Version=2012-11-05&QueueUrl=https%3A%2F%2Flocalhost%2F123456789012%2Forders&Attribute.1.Name=DelaySeconds&Attribute.1.Value.StringValue=5", + )) + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + ctx, err := ParseSQSRequest(req) + if err != nil { + t.Fatalf("parse request: %v", err) + } + if got, want := ctx.Kind, SQSRequestKindQueue; got != want { + t.Fatalf("unexpected kind: got %q want %q", got, want) + } + if got, want := ctx.AccountID, "123456789012"; got != want { + t.Fatalf("unexpected account id: got %q want %q", got, want) + } + if got, want := ctx.QueueName, "orders"; got != want { + t.Fatalf("unexpected queue name: got %q want %q", got, want) + } + if got, want := ctx.NormalizedPath, "/123456789012/orders"; got != want { + t.Fatalf("unexpected normalized path: got %q want %q", got, want) + } + if got, want := ctx.Action, "SendMessage"; got != want { + t.Fatalf("unexpected action: got %q want %q", got, want) + } + if got, want := ctx.Version, sqsQueryVersion; got != want { + t.Fatalf("unexpected version: got %q want %q", got, want) + } + if got, want := ctx.Values.Get("Attribute.1.Name"), "DelaySeconds"; got != want { + t.Fatalf("unexpected numbered attribute name: got %q want %q", got, want) + } + if got, want := ctx.Values.Get("Attribute.1.Value.StringValue"), "5"; got != want { + t.Fatalf("unexpected numbered attribute value: got %q want %q", got, want) + } + if got, want := ctx.Values.Get("QueueUrl"), "https://localhost/123456789012/orders"; got != want { + t.Fatalf("unexpected queue url: got %q want %q", got, want) + } +} + +func TestSQSNativeContractClassifiesRootAndQueuePaths(t *testing.T) { + t.Helper() + + rootReq := httptest.NewRequest(http.MethodGet, "/?Action=ListQueues&Version=2012-11-05", nil) + rootCtx, err := ParseSQSRequest(rootReq) + if err != nil { + t.Fatalf("parse root request: %v", err) + } + if rootCtx.Kind != SQSRequestKindRoot { + t.Fatalf("unexpected root kind: got %q", rootCtx.Kind) + } + if rootCtx.NormalizedPath != "/" { + t.Fatalf("unexpected root normalized path: got %q", rootCtx.NormalizedPath) + } + + queueReq := httptest.NewRequest(http.MethodGet, "/123456789012/orders/?Action=GetQueueAttributes&Version=2012-11-05", nil) + queueCtx, err := ParseSQSRequest(queueReq) + if err != nil { + t.Fatalf("parse queue request: %v", err) + } + if queueCtx.Kind != SQSRequestKindQueue { + t.Fatalf("unexpected queue kind: got %q", queueCtx.Kind) + } + if queueCtx.NormalizedPath != "/123456789012/orders" { + t.Fatalf("unexpected queue normalized path: got %q", queueCtx.NormalizedPath) + } + + if _, err := ParseSQSRequest(httptest.NewRequest(http.MethodGet, "/api/v1/runtime/services", nil)); err != ErrSQSNotOwned { + t.Fatalf("unexpected api path result: got %v want %v", err, ErrSQSNotOwned) + } +} + +func TestSQSNativeContractRejectsMissingActionAndBadVersion(t *testing.T) { + t.Helper() + + missingAction := httptest.NewRequest(http.MethodGet, "/?Version=2012-11-05", nil) + if _, err := ParseSQSRequest(missingAction); err != ErrSQSMissingAction { + t.Fatalf("unexpected missing action result: got %v want %v", err, ErrSQSMissingAction) + } + + badVersion := httptest.NewRequest(http.MethodGet, "/?Action=ListQueues&Version=2014-01-01", nil) + if _, err := ParseSQSRequest(badVersion); err != ErrSQSInvalidVersion { + t.Fatalf("unexpected bad version result: got %v want %v", err, ErrSQSInvalidVersion) + } +} diff --git a/core/internal/delivery/http/sqs_native_errors.go b/core/internal/delivery/http/sqs_native_errors.go new file mode 100644 index 0000000..04e7d18 --- /dev/null +++ b/core/internal/delivery/http/sqs_native_errors.go @@ -0,0 +1,64 @@ +package http + +import ( + "encoding/xml" + "errors" + "net/http" + "strings" + + "github.com/gin-gonic/gin" +) + +type SQSXMLErrorResponse struct { + XMLName xml.Name `xml:"ErrorResponse"` + Error SQSXMLError `xml:"Error"` + RequestID string `xml:"RequestId"` +} + +type SQSXMLError struct { + Type string `xml:"Type"` + Code string `xml:"Code"` + Message string `xml:"Message"` +} + +func writeSQSError(c *gin.Context, err error, requestID string) { + status, code, message := classifySQSError(err) + writeSQSErrorResponse(c, status, code, message, requestID) +} + +func writeSQSErrorResponse(c *gin.Context, status int, code, message, requestID string) { + if c == nil { + return + } + + c.Header("Content-Type", "application/xml") + c.XML(status, SQSXMLErrorResponse{ + Error: SQSXMLError{ + Type: "Sender", + Code: code, + Message: message, + }, + RequestID: strings.TrimSpace(requestID), + }) +} + +func classifySQSError(err error) (int, string, string) { + switch { + case errors.Is(err, ErrSQSNotOwned): + return http.StatusNotFound, "InvalidAction", "The requested path is not owned by the SQS native adapter." + case errors.Is(err, ErrSQSMalformedRequest): + return http.StatusBadRequest, "InvalidQueryParameter", err.Error() + case errors.Is(err, ErrSQSMissingAction): + return http.StatusBadRequest, "MissingAction", "The request is missing an action or required parameter." + case errors.Is(err, ErrSQSInvalidAction): + return http.StatusBadRequest, "InvalidAction", "The action or operation requested is invalid." + case errors.Is(err, ErrSQSInvalidVersion): + return http.StatusBadRequest, "InvalidParameterValue", "The request specified an invalid SQS API version." + case errors.Is(err, ErrSQSQueuePathMismatch): + return http.StatusBadRequest, "InvalidAddress", "The specified queue path is invalid for the requested action." + case errors.Is(err, ErrSQSUnsupported): + return http.StatusBadRequest, "UnsupportedOperation", "The requested operation is not supported by the local subset." + default: + return http.StatusBadRequest, "ValidationError", err.Error() + } +} diff --git a/core/internal/delivery/http/sqs_native_errors_test.go b/core/internal/delivery/http/sqs_native_errors_test.go new file mode 100644 index 0000000..321beab --- /dev/null +++ b/core/internal/delivery/http/sqs_native_errors_test.go @@ -0,0 +1,87 @@ +package http + +import ( + "encoding/xml" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/gin-gonic/gin" +) + +func TestSQSNativeErrorsWriteXMLResponse(t *testing.T) { + t.Helper() + + gin.SetMode(gin.TestMode) + recorder := httptest.NewRecorder() + ctx, _ := gin.CreateTestContext(recorder) + + writeSQSErrorResponse(ctx, http.StatusBadRequest, "InvalidAction", "The action or operation requested is invalid.", "req-123") + + if got := recorder.Header().Get("Content-Type"); !strings.Contains(got, "application/xml") { + t.Fatalf("unexpected content type: got %q", got) + } + + var parsed SQSXMLErrorResponse + if err := xml.Unmarshal(recorder.Body.Bytes(), &parsed); err != nil { + t.Fatalf("unmarshal error response: %v", err) + } + if got, want := parsed.Error.Type, "Sender"; got != want { + t.Fatalf("unexpected error type: got %q want %q", got, want) + } + if got, want := parsed.Error.Code, "InvalidAction"; got != want { + t.Fatalf("unexpected error code: got %q want %q", got, want) + } + if got, want := parsed.Error.Message, "The action or operation requested is invalid."; got != want { + t.Fatalf("unexpected error message: got %q want %q", got, want) + } + if got, want := parsed.RequestID, "req-123"; got != want { + t.Fatalf("unexpected request id: got %q want %q", got, want) + } +} + +func TestSQSNativeErrorsClassifyExplicitErrorCodes(t *testing.T) { + t.Helper() + + cases := []struct { + name string + err error + status int + code string + message string + }{ + {name: "invalid action", err: ErrSQSInvalidAction, status: http.StatusBadRequest, code: "InvalidAction"}, + {name: "missing action", err: ErrSQSMissingAction, status: http.StatusBadRequest, code: "MissingAction"}, + {name: "invalid version", err: ErrSQSInvalidVersion, status: http.StatusBadRequest, code: "InvalidParameterValue"}, + {name: "queue mismatch", err: ErrSQSQueuePathMismatch, status: http.StatusBadRequest, code: "InvalidAddress"}, + {name: "unsupported", err: ErrSQSUnsupported, status: http.StatusBadRequest, code: "UnsupportedOperation"}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Helper() + + status, code, _ := classifySQSError(tc.err) + if status != tc.status { + t.Fatalf("unexpected status: got %d want %d", status, tc.status) + } + if code != tc.code { + t.Fatalf("unexpected code: got %q want %q", code, tc.code) + } + }) + } +} + +func TestSQSNativeErrorsFallbackUsesValidationError(t *testing.T) { + t.Helper() + + status, code, _ := classifySQSError(errors.New("some local failure")) + if status != http.StatusBadRequest { + t.Fatalf("unexpected fallback status: got %d", status) + } + if code != "ValidationError" { + t.Fatalf("unexpected fallback code: got %q", code) + } +} diff --git a/core/internal/delivery/http/sqs_native_registry.go b/core/internal/delivery/http/sqs_native_registry.go new file mode 100644 index 0000000..2f40c05 --- /dev/null +++ b/core/internal/delivery/http/sqs_native_registry.go @@ -0,0 +1,96 @@ +package http + +import ( + "errors" + "fmt" + + "github.com/michasdev/mildstack/core/internal/resources/sqs/contracts" +) + +type SQSRegistrySpec struct { + Action string + Scope contracts.Scope + Version string + Supported bool + DomainDeferred bool + ReturnsQueueURL bool + UsesQueueContext bool +} + +type SQSRegistry struct { + ordered []SQSRegistrySpec + byName map[string]SQSRegistrySpec +} + +func NewSQSRegistry() SQSRegistry { + specs := contracts.Catalog() + ordered := make([]SQSRegistrySpec, 0, len(specs)) + byName := make(map[string]SQSRegistrySpec, len(specs)) + + for _, spec := range specs { + entry := SQSRegistrySpec{ + Action: spec.Action, + Scope: spec.Scope, + Version: spec.Version, + Supported: true, + DomainDeferred: true, + ReturnsQueueURL: spec.ReturnsQueueURL, + UsesQueueContext: spec.UsesQueueContext, + } + ordered = append(ordered, entry) + byName[entry.Action] = entry + } + + return SQSRegistry{ + ordered: ordered, + byName: byName, + } +} + +func (r SQSRegistry) Entries() []SQSRegistrySpec { + return append([]SQSRegistrySpec(nil), r.ordered...) +} + +func (r SQSRegistry) Lookup(action string) (SQSRegistrySpec, bool) { + spec, ok := r.byName[action] + return spec, ok +} + +func (r SQSRegistry) Resolve(ctx SQSRequestContext) (SQSRegistrySpec, error) { + spec, ok := r.Lookup(ctx.Action) + if !ok { + return SQSRegistrySpec{}, ErrSQSInvalidAction + } + if err := validateSQSRequestContext(ctx, spec); err != nil { + return SQSRegistrySpec{}, err + } + return spec, nil +} + +func (r SQSRegistry) SupportedActions() []string { + actions := make([]string, 0, len(r.ordered)) + for _, spec := range r.ordered { + if spec.Supported { + actions = append(actions, spec.Action) + } + } + return actions +} + +func (r SQSRegistry) UnsupportedActions() []string { + actions := make([]string, 0) + for _, spec := range r.ordered { + if !spec.Supported { + actions = append(actions, spec.Action) + } + } + return actions +} + +func (r SQSRegistry) String() string { + return fmt.Sprintf("sqs registry: %d actions", len(r.ordered)) +} + +func isSQSErrorCode(err error, target error) bool { + return errors.Is(err, target) +} diff --git a/core/internal/delivery/http/sqs_native_registry_test.go b/core/internal/delivery/http/sqs_native_registry_test.go new file mode 100644 index 0000000..fea50bd --- /dev/null +++ b/core/internal/delivery/http/sqs_native_registry_test.go @@ -0,0 +1,69 @@ +package http + +import ( + "testing" + + "github.com/michasdev/mildstack/core/internal/resources/sqs/contracts" +) + +func TestSQSNativeRegistryDerivesSpecsFromCatalog(t *testing.T) { + t.Helper() + + registry := NewSQSRegistry() + entries := registry.Entries() + catalog := contracts.Catalog() + + if got, want := len(entries), len(catalog); got != want { + t.Fatalf("unexpected registry entry count: got %d want %d", got, want) + } + + for i, spec := range entries { + if spec.Action != catalog[i].Action { + t.Fatalf("unexpected action at %d: got %q want %q", i, spec.Action, catalog[i].Action) + } + if !spec.Supported { + t.Fatalf("expected action %s to be transport-supported", spec.Action) + } + if !spec.DomainDeferred { + t.Fatalf("expected action %s to be domain deferred", spec.Action) + } + if spec.Scope != catalog[i].Scope { + t.Fatalf("unexpected scope for %s: got %q want %q", spec.Action, spec.Scope, catalog[i].Scope) + } + } +} + +func TestSQSNativeRegistryScopeMismatchesMapToExplicitErrors(t *testing.T) { + t.Helper() + + registry := NewSQSRegistry() + + rootCtx := SQSRequestContext{Action: "ListQueues", Version: sqsQueryVersion, Kind: SQSRequestKindRoot} + if _, err := registry.Resolve(rootCtx); err != nil { + t.Fatalf("resolve matching root action: %v", err) + } + + queueCtx := SQSRequestContext{Action: "SendMessage", Version: sqsQueryVersion, Kind: SQSRequestKindQueue} + if _, err := registry.Resolve(queueCtx); err != nil { + t.Fatalf("resolve matching queue action: %v", err) + } + + queueMismatch := SQSRequestContext{Action: "CreateQueue", Version: sqsQueryVersion, Kind: SQSRequestKindQueue} + if _, err := registry.Resolve(queueMismatch); err != ErrSQSQueuePathMismatch { + t.Fatalf("unexpected queue mismatch error: got %v want %v", err, ErrSQSQueuePathMismatch) + } + + rootMismatch := SQSRequestContext{Action: "SendMessage", Version: sqsQueryVersion, Kind: SQSRequestKindRoot} + if _, err := registry.Resolve(rootMismatch); err != ErrSQSQueuePathMismatch { + t.Fatalf("unexpected root mismatch error: got %v want %v", err, ErrSQSQueuePathMismatch) + } +} + +func TestSQSNativeRegistryRejectsUnknownAction(t *testing.T) { + t.Helper() + + registry := NewSQSRegistry() + if _, err := registry.Resolve(SQSRequestContext{Action: "NotARealAction", Version: sqsQueryVersion, Kind: SQSRequestKindRoot}); err != ErrSQSInvalidAction { + t.Fatalf("unexpected unknown action error: got %v want %v", err, ErrSQSInvalidAction) + } +} diff --git a/core/internal/delivery/http/sqs_native_test.go b/core/internal/delivery/http/sqs_native_test.go new file mode 100644 index 0000000..7e17cf0 --- /dev/null +++ b/core/internal/delivery/http/sqs_native_test.go @@ -0,0 +1,138 @@ +package http + +import ( + "bytes" + "context" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/sqs" + "github.com/gin-gonic/gin" + "github.com/michasdev/mildstack/core/internal/application/runtime" + "github.com/michasdev/mildstack/core/internal/composition" + sqsresource "github.com/michasdev/mildstack/core/internal/resources/sqs" +) + +func TestSQSNativeMiddlewareInterceptsQueryRequestsAndLeavesRuntimeRoutesUntouched(t *testing.T) { + t.Helper() + + gin.SetMode(gin.TestMode) + root := composition.DefaultRoot("test-instance") + manager := runtime.New(root.Services) + router := NewRouter(DefaultConfig(), manager) + + RegisterSQSNativeRoutes(router.Engine(), sqsresource.New()) + + healthRecorder := httptest.NewRecorder() + healthRequest := httptest.NewRequest(http.MethodGet, "/api/v1/runtime/health", nil) + router.Engine().ServeHTTP(healthRecorder, healthRequest) + if got, want := healthRecorder.Code, http.StatusOK; got != want { + t.Fatalf("unexpected health status: got %d want %d", got, want) + } + + rootRecorder := httptest.NewRecorder() + rootRequest := httptest.NewRequest(http.MethodGet, "/?Action=ListQueues&Version=2012-11-05", nil) + router.Engine().ServeHTTP(rootRecorder, rootRequest) + if got, want := rootRecorder.Code, http.StatusBadRequest; got != want { + t.Fatalf("unexpected root status: got %d want %d", got, want) + } + if ct := rootRecorder.Header().Get("Content-Type"); !strings.Contains(ct, "application/xml") { + t.Fatalf("unexpected root content type: got %q", ct) + } + if !strings.Contains(rootRecorder.Body.String(), "UnsupportedOperation") { + t.Fatalf("expected unsupported operation XML, got %q", rootRecorder.Body.String()) + } + + queueRecorder := httptest.NewRecorder() + queueRequest := httptest.NewRequest(http.MethodPost, "/123456789012/orders/", strings.NewReader("Action=SendMessage&Version=2012-11-05&MessageBody=hello")) + queueRequest.Header.Set("Content-Type", "application/x-www-form-urlencoded") + router.Engine().ServeHTTP(queueRecorder, queueRequest) + if got, want := queueRecorder.Code, http.StatusBadRequest; got != want { + t.Fatalf("unexpected queue status: got %d want %d", got, want) + } + if !strings.Contains(queueRecorder.Body.String(), "UnsupportedOperation") { + t.Fatalf("expected unsupported operation XML, got %q", queueRecorder.Body.String()) + } + + mismatchRecorder := httptest.NewRecorder() + mismatchRequest := httptest.NewRequest(http.MethodGet, "/123456789012/orders/?Action=ListQueues&Version=2012-11-05", nil) + router.Engine().ServeHTTP(mismatchRecorder, mismatchRequest) + if got, want := mismatchRecorder.Code, http.StatusBadRequest; got != want { + t.Fatalf("unexpected mismatch status: got %d want %d", got, want) + } + if !strings.Contains(mismatchRecorder.Body.String(), "InvalidAddress") { + t.Fatalf("expected invalid address XML, got %q", mismatchRecorder.Body.String()) + } +} + +func TestSQSSDKSmokeReceivesAWSCompatibleXMLError(t *testing.T) { + t.Helper() + + gin.SetMode(gin.TestMode) + root := composition.DefaultRoot("test-instance") + manager := runtime.New(root.Services) + router := NewRouter(DefaultConfig(), manager) + RegisterSQSNativeRoutes(router.Engine(), sqsresource.New()) + + server := httptest.NewServer(router.Engine()) + t.Cleanup(server.Close) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + t.Cleanup(cancel) + + transport := &captureTransport{base: http.DefaultTransport} + cfg, err := awsconfig.LoadDefaultConfig(ctx, + awsconfig.WithRegion("us-east-1"), + awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test", "test", "test")), + awsconfig.WithHTTPClient(&http.Client{Transport: transport}), + ) + if err != nil { + t.Fatalf("load aws config: %v", err) + } + + client := sqs.NewFromConfig(cfg, func(o *sqs.Options) { + o.BaseEndpoint = aws.String(server.URL) + }) + + _, err = client.ListQueues(ctx, &sqs.ListQueuesInput{}) + if err == nil { + t.Fatal("expected list queues to return an AWS-compatible error") + } + if !strings.Contains(string(transport.body), "") { + t.Fatalf("expected captured xml error body, got %q", string(transport.body)) + } + if !strings.Contains(string(transport.body), "UnsupportedOperation") && !strings.Contains(string(transport.body), "InvalidQueryParameter") { + t.Fatalf("expected captured xml error body to contain an SQS error code, got %q", string(transport.body)) + } +} + +type captureTransport struct { + base http.RoundTripper + body []byte +} + +func (t *captureTransport) RoundTrip(req *http.Request) (*http.Response, error) { + resp, err := t.base.RoundTrip(req) + if err != nil { + return nil, err + } + if resp.Body == nil { + return resp, nil + } + + data, readErr := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if readErr != nil { + return nil, readErr + } + t.body = append([]byte(nil), data...) + resp.Body = io.NopCloser(bytes.NewReader(data)) + return resp, nil +} diff --git a/core/internal/resources/sqs/application/clock.go b/core/internal/resources/sqs/application/clock.go new file mode 100644 index 0000000..775589d --- /dev/null +++ b/core/internal/resources/sqs/application/clock.go @@ -0,0 +1,21 @@ +package application + +import "time" + +type Clock interface { + Now() time.Time + Sleep(time.Duration) +} + +type realClock struct{} + +func (realClock) Now() time.Time { + return time.Now().UTC() +} + +func (realClock) Sleep(duration time.Duration) { + if duration <= 0 { + return + } + time.Sleep(duration) +} diff --git a/core/internal/resources/sqs/application/delivery_state.go b/core/internal/resources/sqs/application/delivery_state.go new file mode 100644 index 0000000..efb5c3a --- /dev/null +++ b/core/internal/resources/sqs/application/delivery_state.go @@ -0,0 +1,143 @@ +package application + +import ( + "strconv" + "strings" + "time" + + "github.com/michasdev/mildstack/core/internal/resources/sqs/domain" +) + +const ( + defaultVisibilityTimeoutSeconds = 30 + leaseTimeoutMetadataKey = "visibility_timeout_seconds" +) + +type DeliveryView struct { + Queue domain.Queue + Message domain.Message + Now time.Time +} + +func (v DeliveryView) Delay() bool { + return IsDelayed(v.Message, v.Now) +} + +func (v DeliveryView) Visible() bool { + return IsVisible(v.Message, v.Queue, v.Now) +} + +func (v DeliveryView) Invisible() bool { + return IsInvisible(v.Message, v.Queue, v.Now) +} + +func (v DeliveryView) Receipt() string { + return CurrentReceiptHandle(v.Message) +} + +func (v DeliveryView) Redeliver() bool { + return CanRedeliver(v.Message, v.Queue, v.Now) +} + +func (v DeliveryView) DeadLetterEligible() bool { + return IsDeadLetterEligible(v.Message, v.Queue, v.Now) +} + +func IsDelayed(message domain.Message, now time.Time) bool { + return !message.AvailableAt.IsZero() && now.Before(message.AvailableAt) +} + +func IsVisible(message domain.Message, queue domain.Queue, now time.Time) bool { + if IsDelayed(message, now) { + return false + } + if message.ReceivedAt.IsZero() { + return true + } + return !now.Before(leaseDeadline(message, queue)) +} + +func IsInvisible(message domain.Message, queue domain.Queue, now time.Time) bool { + return !IsDelayed(message, now) && !IsVisible(message, queue, now) +} + +func CurrentReceiptHandle(message domain.Message) string { + if len(message.ReceiptKeys) == 0 { + return "" + } + return message.ReceiptKeys[len(message.ReceiptKeys)-1] +} + +func CanRedeliver(message domain.Message, queue domain.Queue, now time.Time) bool { + if IsDelayed(message, now) { + return false + } + deadline := leaseDeadline(message, queue) + return !message.ReceivedAt.IsZero() && !deadline.IsZero() && !now.Before(deadline) +} + +func IsFIFOQueue(queue domain.Queue) bool { + if strings.EqualFold(trimName(queue.OrderingHint), "fifo") { + return true + } + return strings.EqualFold(trimName(queue.Attributes["FifoQueue"]), "true") +} + +func MessageGroupID(message domain.Message) string { + return trimName(message.MessageGroupID) +} + +func MessageSequenceNumber(message domain.Message) int64 { + return message.SequenceNumber +} + +func MessageBatchState(message domain.Message) (string, string, int, int) { + return trimName(message.BatchID), trimName(message.BatchEntryID), message.BatchEntryIndex, message.BatchEntryCount +} + +func IsDeadLetterEligible(message domain.Message, queue domain.Queue, now time.Time) bool { + if trimName(message.DeadLetterQueue) != "" { + return false + } + if !CanRedeliver(message, queue, now) { + return false + } + return deadLetterThreshold(queue) > 0 && message.Recovery.Attempts >= deadLetterThreshold(queue) +} + +func leaseDeadline(message domain.Message, queue domain.Queue) time.Time { + timeout := visibilityTimeout(queue, message) + if timeout <= 0 || message.ReceivedAt.IsZero() { + return time.Time{} + } + return message.ReceivedAt.Add(timeout) +} + +func visibilityTimeout(queue domain.Queue, message domain.Message) time.Duration { + if timeout := parseTimeoutSeconds(message.Metadata[leaseTimeoutMetadataKey]); timeout > 0 { + return timeout + } + if timeout := parseTimeoutSeconds(queue.Attributes["VisibilityTimeout"]); timeout > 0 { + return timeout + } + return defaultVisibilityTimeoutSeconds * time.Second +} + +func parseTimeoutSeconds(raw string) time.Duration { + seconds, err := strconv.Atoi(trimName(raw)) + if err != nil || seconds < 0 { + return 0 + } + return time.Duration(seconds) * time.Second +} + +func deadLetterThreshold(queue domain.Queue) int { + if queue.Recovery.Policy == nil { + return 0 + } + threshold, err := strconv.Atoi(trimName(queue.Recovery.Policy["max_receive_count"])) + if err != nil || threshold <= 0 { + return 0 + } + return threshold +} diff --git a/core/internal/resources/sqs/application/delivery_state_test.go b/core/internal/resources/sqs/application/delivery_state_test.go new file mode 100644 index 0000000..fd2c671 --- /dev/null +++ b/core/internal/resources/sqs/application/delivery_state_test.go @@ -0,0 +1,110 @@ +package application + +import ( + "testing" + "time" + + "github.com/michasdev/mildstack/core/internal/resources/sqs/domain" +) + +func TestDeliveryHelpersCoverDelayVisibilityLeaseAndReceiptRotation(t *testing.T) { + t.Helper() + + now := time.Date(2026, time.April, 19, 12, 0, 0, 0, time.UTC) + queue := domain.Queue{ + Name: "queue-a", + Attributes: map[string]string{ + "VisibilityTimeout": "30", + }, + } + + delayed := domain.Message{ + Queue: "queue-a", + MessageID: "message-delayed", + AvailableAt: now.Add(2 * time.Minute), + } + if !IsDelayed(delayed, now) { + t.Fatal("expected delayed message to be marked delayed") + } + if (DeliveryView{Queue: queue, Message: delayed, Now: now}).Visible() { + t.Fatal("expected delayed message to be hidden") + } + if (DeliveryView{Queue: queue, Message: delayed, Now: now}).Redeliver() { + t.Fatal("expected delayed message to not redeliver yet") + } + + visible := domain.Message{ + Queue: "queue-a", + MessageID: "message-visible", + } + if !IsVisible(visible, queue, now) { + t.Fatal("expected new message to be visible") + } + if (DeliveryView{Queue: queue, Message: visible, Now: now}).Invisible() { + t.Fatal("expected new message to not be invisible") + } + + inflight := domain.Message{ + Queue: "queue-a", + MessageID: "message-inflight", + ReceivedAt: now.Add(-20 * time.Second), + ReceiptKeys: []string{"r-1", "r-2"}, + Metadata: map[string]string{ + leaseTimeoutMetadataKey: "30", + }, + } + if !(DeliveryView{Queue: queue, Message: inflight, Now: now}).Invisible() { + t.Fatal("expected in-flight message to be invisible") + } + if got, want := CurrentReceiptHandle(inflight), "r-2"; got != want { + t.Fatalf("unexpected receipt handle: got %q want %q", got, want) + } + if (DeliveryView{Queue: queue, Message: inflight, Now: now.Add(9 * time.Second)}).Redeliver() { + t.Fatal("expected lease to remain active before timeout") + } + if !CanRedeliver(inflight, queue, now.Add(10*time.Second)) { + t.Fatal("expected lease to redeliver once timeout expires") + } +} + +func TestDeliveryHelpersCoverFifoAndDeadLetterEligibility(t *testing.T) { + t.Helper() + + now := time.Date(2026, time.April, 19, 12, 0, 0, 0, time.UTC) + fifoQueue := domain.Queue{ + Name: "queue-fifo", + OrderingHint: "fifo", + Recovery: domain.QueueRecovery{ + Policy: map[string]string{ + "max_receive_count": "2", + }, + }, + } + message := domain.Message{ + Queue: "queue-fifo", + MessageID: "message-1", + MessageGroupID: "group-a", + SequenceNumber: 7, + ReceivedAt: now.Add(-31 * time.Second), + DeadLetterQueue: "", + Recovery: domain.MessageRecovery{ + Attempts: 2, + }, + } + + if !IsFIFOQueue(fifoQueue) { + t.Fatal("expected fifo queue hint to be recognized") + } + if got, want := MessageGroupID(message), "group-a"; got != want { + t.Fatalf("unexpected message group id: got %q want %q", got, want) + } + if got, want := MessageSequenceNumber(message), int64(7); got != want { + t.Fatalf("unexpected message sequence number: got %d want %d", got, want) + } + if !IsDeadLetterEligible(message, fifoQueue, now) { + t.Fatal("expected message to be dead-letter eligible") + } + if got := (DeliveryView{Queue: fifoQueue, Message: message, Now: now}).DeadLetterEligible(); !got { + t.Fatal("expected delivery view to expose dead-letter eligibility") + } +} diff --git a/core/internal/resources/sqs/application/repository.go b/core/internal/resources/sqs/application/repository.go new file mode 100644 index 0000000..7cf0eab --- /dev/null +++ b/core/internal/resources/sqs/application/repository.go @@ -0,0 +1,9 @@ +package application + +import "github.com/michasdev/mildstack/core/internal/resources/sqs/domain" + +type Repository interface { + Load() (domain.State, error) + Save(state domain.State) error + Close() error +} diff --git a/core/internal/resources/sqs/application/repository_sqlite.go b/core/internal/resources/sqs/application/repository_sqlite.go new file mode 100644 index 0000000..228d0d1 --- /dev/null +++ b/core/internal/resources/sqs/application/repository_sqlite.go @@ -0,0 +1,675 @@ +package application + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "time" + + _ "modernc.org/sqlite" + + "github.com/michasdev/mildstack/core/internal/resources/sqs/domain" +) + +const ( + sqliteFileName = "state.db" + schemaVersionKey = "schema_version" + schemaVersion = "2" +) + +type SQLiteRepository struct { + db *sql.DB + dbPath string + storageDir string + mu sync.Mutex +} + +var errSQLiteRepositoryClosed = errors.New("sqs: repository is closed") + +func NewSQLiteRepository(storagePath string) (*SQLiteRepository, error) { + storagePath = strings.TrimSpace(storagePath) + if storagePath == "" { + return nil, fmt.Errorf("sqs: storage path is required") + } + + if err := os.MkdirAll(storagePath, 0o755); err != nil { + return nil, fmt.Errorf("sqs: create storage directory: %w", err) + } + + dbPath := filepath.Join(storagePath, sqliteFileName) + db, err := sql.Open("sqlite", dbPath) + if err != nil { + return nil, fmt.Errorf("sqs: open sqlite database: %w", err) + } + + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) + db.SetConnMaxLifetime(0) + + repo := &SQLiteRepository{ + db: db, + dbPath: dbPath, + storageDir: storagePath, + } + + if err := repo.bootstrap(); err != nil { + _ = db.Close() + return nil, err + } + + return repo, nil +} + +func (r *SQLiteRepository) Load() (domain.State, error) { + r.mu.Lock() + defer r.mu.Unlock() + + if r.db == nil { + return domain.State{}, errSQLiteRepositoryClosed + } + + state, err := r.loadLocked() + if err != nil { + return domain.State{}, err + } + return state, nil +} + +func (r *SQLiteRepository) Save(state domain.State) error { + r.mu.Lock() + defer r.mu.Unlock() + + if r.db == nil { + return errSQLiteRepositoryClosed + } + + return r.saveLocked(state) +} + +func (r *SQLiteRepository) Close() error { + r.mu.Lock() + defer r.mu.Unlock() + + if r.db == nil { + return nil + } + + err := r.db.Close() + r.db = nil + return err +} + +func (r *SQLiteRepository) bootstrap() error { + r.mu.Lock() + defer r.mu.Unlock() + + if r.db == nil { + return errSQLiteRepositoryClosed + } + + ctx := context.Background() + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("sqs: bootstrap transaction: %w", err) + } + + statements := []string{ + `CREATE TABLE IF NOT EXISTS sqs_meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + )`, + `CREATE TABLE IF NOT EXISTS sqs_queues ( + name TEXT PRIMARY KEY, + url TEXT NOT NULL, + attributes_json TEXT NOT NULL, + ordering_hint TEXT NOT NULL DEFAULT '', + dead_letter_queue TEXT NOT NULL DEFAULT '', + policy_json TEXT NOT NULL, + created_at_ns INTEGER NOT NULL DEFAULT 0, + updated_at_ns INTEGER NOT NULL DEFAULT 0 + )`, + `CREATE TABLE IF NOT EXISTS sqs_messages ( + queue_name TEXT NOT NULL, + message_id TEXT NOT NULL, + body TEXT NOT NULL, + attributes_json TEXT NOT NULL, + metadata_json TEXT NOT NULL, + tags_json TEXT NOT NULL, + receipt_keys_json TEXT NOT NULL, + message_group_id TEXT NOT NULL DEFAULT '', + sequence_number INTEGER NOT NULL DEFAULT 0, + batch_id TEXT NOT NULL DEFAULT '', + batch_entry_id TEXT NOT NULL DEFAULT '', + batch_entry_index INTEGER NOT NULL DEFAULT 0, + batch_entry_count INTEGER NOT NULL DEFAULT 0, + dead_letter_queue TEXT NOT NULL DEFAULT '', + dead_letter_source_queue TEXT NOT NULL DEFAULT '', + dead_lettered_at_ns INTEGER NOT NULL DEFAULT 0, + sent_at_ns INTEGER NOT NULL DEFAULT 0, + available_at_ns INTEGER NOT NULL DEFAULT 0, + received_at_ns INTEGER NOT NULL DEFAULT 0, + recovery_attempts INTEGER NOT NULL DEFAULT 0, + recovery_detail_json TEXT NOT NULL, + PRIMARY KEY (queue_name, message_id), + FOREIGN KEY (queue_name) REFERENCES sqs_queues(name) ON DELETE CASCADE + )`, + `CREATE INDEX IF NOT EXISTS idx_sqs_messages_queue_name ON sqs_messages(queue_name)`, + `CREATE TABLE IF NOT EXISTS sqs_recovery_metadata ( + key TEXT PRIMARY KEY, + queue_name TEXT NOT NULL, + message_id TEXT NOT NULL, + detail_json TEXT NOT NULL + )`, + } + + for _, statement := range statements { + if _, err := tx.ExecContext(ctx, statement); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: bootstrap schema: %w", err) + } + } + + if err := ensureColumn(ctx, tx, "sqs_queues", "ordering_hint TEXT NOT NULL DEFAULT ''"); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: ensure queue ordering column: %w", err) + } + if err := ensureColumn(ctx, tx, "sqs_messages", "message_group_id TEXT NOT NULL DEFAULT ''"); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: ensure message group column: %w", err) + } + if err := ensureColumn(ctx, tx, "sqs_messages", "sequence_number INTEGER NOT NULL DEFAULT 0"); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: ensure message sequence column: %w", err) + } + if err := ensureColumn(ctx, tx, "sqs_messages", "batch_id TEXT NOT NULL DEFAULT ''"); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: ensure batch id column: %w", err) + } + if err := ensureColumn(ctx, tx, "sqs_messages", "batch_entry_id TEXT NOT NULL DEFAULT ''"); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: ensure batch entry id column: %w", err) + } + if err := ensureColumn(ctx, tx, "sqs_messages", "batch_entry_index INTEGER NOT NULL DEFAULT 0"); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: ensure batch entry index column: %w", err) + } + if err := ensureColumn(ctx, tx, "sqs_messages", "batch_entry_count INTEGER NOT NULL DEFAULT 0"); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: ensure batch entry count column: %w", err) + } + if err := ensureColumn(ctx, tx, "sqs_messages", "dead_letter_queue TEXT NOT NULL DEFAULT ''"); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: ensure dead letter queue column: %w", err) + } + if err := ensureColumn(ctx, tx, "sqs_messages", "dead_letter_source_queue TEXT NOT NULL DEFAULT ''"); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: ensure dead letter source column: %w", err) + } + if err := ensureColumn(ctx, tx, "sqs_messages", "dead_lettered_at_ns INTEGER NOT NULL DEFAULT 0"); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: ensure dead lettered at column: %w", err) + } + + if _, err := tx.ExecContext(ctx, ` + INSERT INTO sqs_meta(key, value) + VALUES (?, ?) + ON CONFLICT(key) DO UPDATE SET value = excluded.value + `, schemaVersionKey, schemaVersion); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: bootstrap schema version: %w", err) + } + + if err := tx.Commit(); err != nil { + return fmt.Errorf("sqs: commit bootstrap: %w", err) + } + + return nil +} + +func (r *SQLiteRepository) loadLocked() (domain.State, error) { + ctx := context.Background() + state := domain.NewState() + + queueRows, err := r.db.QueryContext(ctx, ` + SELECT name, url, attributes_json, ordering_hint, dead_letter_queue, policy_json, created_at_ns, updated_at_ns + FROM sqs_queues + ORDER BY name + `) + if err != nil { + return domain.State{}, fmt.Errorf("sqs: query queues: %w", err) + } + defer queueRows.Close() + + for queueRows.Next() { + var ( + queue domain.Queue + attributes string + ordering string + policy string + createdAtNS int64 + updatedAtNS int64 + ) + if err := queueRows.Scan(&queue.Name, &queue.URL, &attributes, &ordering, &queue.Recovery.DeadLetterQueue, &policy, &createdAtNS, &updatedAtNS); err != nil { + return domain.State{}, fmt.Errorf("sqs: scan queue: %w", err) + } + queue.Attributes, err = decodeStringMap(attributes) + if err != nil { + return domain.State{}, fmt.Errorf("sqs: decode queue attributes: %w", err) + } + queue.Recovery.Policy, err = decodeStringMap(policy) + if err != nil { + return domain.State{}, fmt.Errorf("sqs: decode queue policy: %w", err) + } + queue.OrderingHint = ordering + queue.CreatedAt = unixNanoToTime(createdAtNS) + queue.UpdatedAt = unixNanoToTime(updatedAtNS) + state.Queues = append(state.Queues, queue) + } + if err := queueRows.Err(); err != nil { + return domain.State{}, fmt.Errorf("sqs: iterate queues: %w", err) + } + + messageRows, err := r.db.QueryContext(ctx, ` + SELECT queue_name, message_id, body, attributes_json, metadata_json, tags_json, receipt_keys_json, + message_group_id, sequence_number, batch_id, batch_entry_id, batch_entry_index, batch_entry_count, + dead_letter_queue, dead_letter_source_queue, dead_lettered_at_ns, + sent_at_ns, available_at_ns, received_at_ns, recovery_attempts, recovery_detail_json + FROM sqs_messages + ORDER BY queue_name, message_id + `) + if err != nil { + return domain.State{}, fmt.Errorf("sqs: query messages: %w", err) + } + defer messageRows.Close() + + for messageRows.Next() { + var ( + message domain.Message + attributes string + metadata string + tags string + receiptKeys string + messageGroupID string + sequenceNumber int64 + batchID string + batchEntryID string + batchEntryIndex int + batchEntryCount int + deadLetterQueue string + deadLetterSource string + deadLetteredAtNS int64 + sentAtNS int64 + availableAtNS int64 + receivedAtNS int64 + recoveryDetail string + ) + if err := messageRows.Scan( + &message.Queue, + &message.MessageID, + &message.Body, + &attributes, + &metadata, + &tags, + &receiptKeys, + &messageGroupID, + &sequenceNumber, + &batchID, + &batchEntryID, + &batchEntryIndex, + &batchEntryCount, + &deadLetterQueue, + &deadLetterSource, + &deadLetteredAtNS, + &sentAtNS, + &availableAtNS, + &receivedAtNS, + &message.Recovery.Attempts, + &recoveryDetail, + ); err != nil { + return domain.State{}, fmt.Errorf("sqs: scan message: %w", err) + } + message.Attributes, err = decodeStringMap(attributes) + if err != nil { + return domain.State{}, fmt.Errorf("sqs: decode message attributes: %w", err) + } + message.Metadata, err = decodeStringMap(metadata) + if err != nil { + return domain.State{}, fmt.Errorf("sqs: decode message metadata: %w", err) + } + message.Tags, err = decodeStringSlice(tags) + if err != nil { + return domain.State{}, fmt.Errorf("sqs: decode message tags: %w", err) + } + message.ReceiptKeys, err = decodeStringSlice(receiptKeys) + if err != nil { + return domain.State{}, fmt.Errorf("sqs: decode receipt keys: %w", err) + } + message.MessageGroupID = messageGroupID + message.SequenceNumber = sequenceNumber + message.BatchID = batchID + message.BatchEntryID = batchEntryID + message.BatchEntryIndex = batchEntryIndex + message.BatchEntryCount = batchEntryCount + message.DeadLetterQueue = deadLetterQueue + message.DeadLetterSourceQueue = deadLetterSource + message.DeadLetteredAt = unixNanoToTime(deadLetteredAtNS) + message.SentAt = unixNanoToTime(sentAtNS) + message.AvailableAt = unixNanoToTime(availableAtNS) + message.ReceivedAt = unixNanoToTime(receivedAtNS) + message.Recovery.Detail, err = decodeStringMap(recoveryDetail) + if err != nil { + return domain.State{}, fmt.Errorf("sqs: decode message recovery detail: %w", err) + } + state.Messages = append(state.Messages, message) + } + if err := messageRows.Err(); err != nil { + return domain.State{}, fmt.Errorf("sqs: iterate messages: %w", err) + } + + recoveryRows, err := r.db.QueryContext(ctx, ` + SELECT key, queue_name, message_id, detail_json + FROM sqs_recovery_metadata + ORDER BY key + `) + if err != nil { + return domain.State{}, fmt.Errorf("sqs: query recovery metadata: %w", err) + } + defer recoveryRows.Close() + + for recoveryRows.Next() { + var ( + key string + metadata domain.RecoveryMetadata + detail string + ) + if err := recoveryRows.Scan(&key, &metadata.Queue, &metadata.Message, &detail); err != nil { + return domain.State{}, fmt.Errorf("sqs: scan recovery metadata: %w", err) + } + metadata.Detail, err = decodeStringMap(detail) + if err != nil { + return domain.State{}, fmt.Errorf("sqs: decode recovery detail: %w", err) + } + state.RecoveryMetadata[key] = metadata + } + if err := recoveryRows.Err(); err != nil { + return domain.State{}, fmt.Errorf("sqs: iterate recovery metadata: %w", err) + } + + return state, nil +} + +func (r *SQLiteRepository) saveLocked(state domain.State) error { + ctx := context.Background() + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("sqs: begin save transaction: %w", err) + } + + normalized := state.Clone() + normalized.Service = "sqs" + + if _, err := tx.ExecContext(ctx, `DELETE FROM sqs_recovery_metadata`); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: clear recovery metadata: %w", err) + } + if _, err := tx.ExecContext(ctx, `DELETE FROM sqs_messages`); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: clear messages: %w", err) + } + if _, err := tx.ExecContext(ctx, `DELETE FROM sqs_queues`); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: clear queues: %w", err) + } + + queueStmt, err := tx.PrepareContext(ctx, ` + INSERT INTO sqs_queues ( + name, url, attributes_json, ordering_hint, dead_letter_queue, policy_json, created_at_ns, updated_at_ns + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + `) + if err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: prepare queue insert: %w", err) + } + defer queueStmt.Close() + + for _, queue := range normalized.ListQueues() { + attributes, err := encodeStringMap(queue.Attributes) + if err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: encode queue attributes: %w", err) + } + policy, err := encodeStringMap(queue.Recovery.Policy) + if err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: encode queue policy: %w", err) + } + if _, err := queueStmt.ExecContext(ctx, + queue.Name, + queue.URL, + attributes, + queue.OrderingHint, + queue.Recovery.DeadLetterQueue, + policy, + timeToUnixNano(queue.CreatedAt), + timeToUnixNano(queue.UpdatedAt), + ); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: insert queue %q: %w", queue.Name, err) + } + } + + messageStmt, err := tx.PrepareContext(ctx, ` + INSERT INTO sqs_messages ( + queue_name, message_id, body, attributes_json, metadata_json, tags_json, receipt_keys_json, + message_group_id, sequence_number, batch_id, batch_entry_id, batch_entry_index, batch_entry_count, + dead_letter_queue, dead_letter_source_queue, dead_lettered_at_ns, + sent_at_ns, available_at_ns, received_at_ns, recovery_attempts, recovery_detail_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `) + if err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: prepare message insert: %w", err) + } + defer messageStmt.Close() + + for _, message := range normalized.ListMessages() { + attributes, err := encodeStringMap(message.Attributes) + if err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: encode message attributes: %w", err) + } + metadata, err := encodeStringMap(message.Metadata) + if err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: encode message metadata: %w", err) + } + tags, err := encodeStringSlice(message.Tags) + if err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: encode message tags: %w", err) + } + receiptKeys, err := encodeStringSlice(message.ReceiptKeys) + if err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: encode receipt keys: %w", err) + } + recovery, err := encodeStringMap(message.Recovery.Detail) + if err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: encode message recovery detail: %w", err) + } + if _, err := messageStmt.ExecContext(ctx, + message.Queue, + message.MessageID, + message.Body, + attributes, + metadata, + tags, + receiptKeys, + message.MessageGroupID, + message.SequenceNumber, + message.BatchID, + message.BatchEntryID, + message.BatchEntryIndex, + message.BatchEntryCount, + message.DeadLetterQueue, + message.DeadLetterSourceQueue, + timeToUnixNano(message.DeadLetteredAt), + timeToUnixNano(message.SentAt), + timeToUnixNano(message.AvailableAt), + timeToUnixNano(message.ReceivedAt), + message.Recovery.Attempts, + recovery, + ); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: insert message %q/%q: %w", message.Queue, message.MessageID, err) + } + } + + recoveryKeys := make([]string, 0, len(normalized.RecoveryMetadata)) + for key := range normalized.RecoveryMetadata { + recoveryKeys = append(recoveryKeys, key) + } + sort.Strings(recoveryKeys) + recoveryStmt, err := tx.PrepareContext(ctx, ` + INSERT INTO sqs_recovery_metadata (key, queue_name, message_id, detail_json) + VALUES (?, ?, ?, ?) + `) + if err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: prepare recovery insert: %w", err) + } + defer recoveryStmt.Close() + + for _, key := range recoveryKeys { + metadata := normalized.RecoveryMetadata[key] + detail, err := encodeStringMap(metadata.Detail) + if err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: encode recovery detail: %w", err) + } + if _, err := recoveryStmt.ExecContext(ctx, key, metadata.Queue, metadata.Message, detail); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: insert recovery metadata %q: %w", key, err) + } + } + + if _, err := tx.ExecContext(ctx, ` + INSERT INTO sqs_meta(key, value) + VALUES (?, ?) + ON CONFLICT(key) DO UPDATE SET value = excluded.value + `, schemaVersionKey, schemaVersion); err != nil { + _ = tx.Rollback() + return fmt.Errorf("sqs: persist schema version: %w", err) + } + + if err := tx.Commit(); err != nil { + return fmt.Errorf("sqs: commit save transaction: %w", err) + } + + return nil +} + +func encodeStringMap(values map[string]string) (string, error) { + if values == nil { + return "{}", nil + } + encoded, err := json.Marshal(values) + if err != nil { + return "", err + } + return string(encoded), nil +} + +func decodeStringMap(encoded string) (map[string]string, error) { + if strings.TrimSpace(encoded) == "" { + return nil, nil + } + + var values map[string]string + if err := json.Unmarshal([]byte(encoded), &values); err != nil { + return nil, err + } + return values, nil +} + +func encodeStringSlice(values []string) (string, error) { + if values == nil { + return "[]", nil + } + encoded, err := json.Marshal(values) + if err != nil { + return "", err + } + return string(encoded), nil +} + +func decodeStringSlice(encoded string) ([]string, error) { + if strings.TrimSpace(encoded) == "" { + return nil, nil + } + + var values []string + if err := json.Unmarshal([]byte(encoded), &values); err != nil { + return nil, err + } + return values, nil +} + +func ensureColumn(ctx context.Context, tx *sql.Tx, tableName, definition string) error { + columnName := definition + if idx := strings.IndexAny(columnName, " \t"); idx >= 0 { + columnName = columnName[:idx] + } + + query := fmt.Sprintf("PRAGMA table_info(%s)", tableName) + rows, err := tx.QueryContext(ctx, query) + if err != nil { + return err + } + defer rows.Close() + + for rows.Next() { + var ( + cid int + name string + colType string + notNull int + defaultValue sql.NullString + pk int + ) + if err := rows.Scan(&cid, &name, &colType, ¬Null, &defaultValue, &pk); err != nil { + return err + } + if name == columnName { + return nil + } + } + if err := rows.Err(); err != nil { + return err + } + + _, err = tx.ExecContext(ctx, fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s", tableName, definition)) + return err +} + +func timeToUnixNano(value time.Time) int64 { + if value.IsZero() { + return 0 + } + return value.UTC().UnixNano() +} + +func unixNanoToTime(value int64) time.Time { + if value == 0 { + return time.Time{} + } + return time.Unix(0, value).UTC() +} diff --git a/core/internal/resources/sqs/application/repository_sqlite_test.go b/core/internal/resources/sqs/application/repository_sqlite_test.go new file mode 100644 index 0000000..53aa923 --- /dev/null +++ b/core/internal/resources/sqs/application/repository_sqlite_test.go @@ -0,0 +1,245 @@ +package application + +import ( + "os" + "path/filepath" + "testing" + "time" + + "github.com/michasdev/mildstack/core/internal/resources/sqs/domain" +) + +func TestResolveStoragePath(t *testing.T) { + t.Helper() + + baseDir := t.TempDir() + path, err := ResolveStoragePath(StorageConfig{ + BaseDir: baseDir, + InstanceID: "instance-a", + }) + if err != nil { + t.Fatalf("resolve storage path: %v", err) + } + + want := filepath.Join(baseDir, "instances", "instance-a", "sqs") + if got, want := path, want; got != want { + t.Fatalf("unexpected storage path: got %q want %q", got, want) + } +} + +func TestSQLiteRepositoryLoadsDefaultStateFromNewStorage(t *testing.T) { + t.Helper() + + repo := mustOpenSQLiteRepository(t, "instance-a") + defer func() { + if err := repo.Close(); err != nil { + t.Fatalf("close repo: %v", err) + } + }() + + state, err := repo.Load() + if err != nil { + t.Fatalf("load state: %v", err) + } + if got, want := state.Service, "sqs"; got != want { + t.Fatalf("unexpected service name: got %q want %q", got, want) + } + if got, want := len(state.Queues), 0; got != want { + t.Fatalf("unexpected queue count: got %d want %d", got, want) + } + if got, want := len(state.Messages), 0; got != want { + t.Fatalf("unexpected message count: got %d want %d", got, want) + } +} + +func TestSQLiteRepositoryPersistsQueueAndMessageStateAcrossRestart(t *testing.T) { + t.Helper() + + baseDir := t.TempDir() + repo := mustOpenSQLiteRepositoryAt(t, baseDir, "instance-a") + defer func() { + if err := repo.Close(); err != nil { + t.Fatalf("close repo: %v", err) + } + }() + + createdAt := time.Date(2026, time.April, 19, 12, 0, 0, 0, time.UTC) + state := domain.NewState() + state.Queues = append(state.Queues, domain.Queue{ + Name: "queue-a", + URL: "https://example.invalid/queue-a", + Attributes: map[string]string{ + "VisibilityTimeout": "30", + "DelaySeconds": "0", + }, + OrderingHint: "fifo", + Recovery: domain.QueueRecovery{ + DeadLetterQueue: "queue-dlq", + Policy: map[string]string{ + "max_receive_count": "5", + }, + }, + CreatedAt: createdAt, + UpdatedAt: createdAt.Add(time.Minute), + }) + state.Messages = append(state.Messages, domain.Message{ + Queue: "queue-a", + MessageID: "message-1", + Body: "payload", + Attributes: map[string]string{"foo": "bar"}, + Metadata: map[string]string{"trace": "abc"}, + Tags: []string{"alpha", "beta"}, + ReceiptKeys: []string{"r-1", "r-2"}, + MessageGroupID: "group-a", + SequenceNumber: 42, + BatchID: "batch-a", + BatchEntryID: "entry-a", + BatchEntryIndex: 1, + BatchEntryCount: 3, + DeadLetterQueue: "queue-dlq", + DeadLetterSourceQueue: "queue-a", + DeadLetteredAt: createdAt.Add(5 * time.Minute), + SentAt: createdAt.Add(2 * time.Minute), + AvailableAt: createdAt.Add(3 * time.Minute), + ReceivedAt: createdAt.Add(4 * time.Minute), + Recovery: domain.MessageRecovery{ + Attempts: 2, + Detail: map[string]string{"state": "pending"}, + }, + }) + state.RecoveryMetadata["queue-a/message-1"] = domain.RecoveryMetadata{ + Queue: "queue-a", + Message: "message-1", + Detail: map[string]string{"reason": "retry"}, + } + + if err := repo.Save(state); err != nil { + t.Fatalf("save state: %v", err) + } + if err := repo.Close(); err != nil { + t.Fatalf("close repo after save: %v", err) + } + + reopened := mustOpenSQLiteRepositoryAt(t, baseDir, "instance-a") + defer func() { + if err := reopened.Close(); err != nil { + t.Fatalf("close reopened repo: %v", err) + } + }() + + loaded, err := reopened.Load() + if err != nil { + t.Fatalf("load restarted state: %v", err) + } + if got, want := len(loaded.Queues), 1; got != want { + t.Fatalf("unexpected queue count after restart: got %d want %d", got, want) + } + if got, want := len(loaded.Messages), 1; got != want { + t.Fatalf("unexpected message count after restart: got %d want %d", got, want) + } + + queue := loaded.Queues[0] + if got, want := queue.Name, "queue-a"; got != want { + t.Fatalf("unexpected queue name after restart: got %q want %q", got, want) + } + if got, want := queue.Attributes["VisibilityTimeout"], "30"; got != want { + t.Fatalf("unexpected queue attribute after restart: got %q want %q", got, want) + } + if got, want := queue.OrderingHint, "fifo"; got != want { + t.Fatalf("unexpected ordering hint after restart: got %q want %q", got, want) + } + if got, want := queue.Recovery.DeadLetterQueue, "queue-dlq"; got != want { + t.Fatalf("unexpected dead-letter queue after restart: got %q want %q", got, want) + } + + message := loaded.Messages[0] + if got, want := message.Body, "payload"; got != want { + t.Fatalf("unexpected message body after restart: got %q want %q", got, want) + } + if got, want := message.Tags[0], "alpha"; got != want { + t.Fatalf("unexpected message tag after restart: got %q want %q", got, want) + } + if got, want := message.AvailableAt, createdAt.Add(3*time.Minute); !got.Equal(want) { + t.Fatalf("unexpected available_at after restart: got %v want %v", got, want) + } + if got, want := message.ReceivedAt, createdAt.Add(4*time.Minute); !got.Equal(want) { + t.Fatalf("unexpected received_at after restart: got %v want %v", got, want) + } + if got, want := len(message.ReceiptKeys), 2; got != want { + t.Fatalf("unexpected receipt key count after restart: got %d want %d", got, want) + } + if got, want := message.ReceiptKeys[1], "r-2"; got != want { + t.Fatalf("unexpected latest receipt key after restart: got %q want %q", got, want) + } + if got, want := message.MessageGroupID, "group-a"; got != want { + t.Fatalf("unexpected message group after restart: got %q want %q", got, want) + } + if got, want := message.SequenceNumber, int64(42); got != want { + t.Fatalf("unexpected message sequence after restart: got %d want %d", got, want) + } + if got, want := message.BatchID, "batch-a"; got != want { + t.Fatalf("unexpected batch id after restart: got %q want %q", got, want) + } + if got, want := message.DeadLetterQueue, "queue-dlq"; got != want { + t.Fatalf("unexpected dead letter queue after restart: got %q want %q", got, want) + } + if got, want := message.DeadLetterSourceQueue, "queue-a"; got != want { + t.Fatalf("unexpected dead letter source after restart: got %q want %q", got, want) + } + if got, want := message.DeadLetteredAt, createdAt.Add(5*time.Minute); !got.Equal(want) { + t.Fatalf("unexpected dead letter time after restart: got %v want %v", got, want) + } + if got, want := message.Recovery.Detail["state"], "pending"; got != want { + t.Fatalf("unexpected message recovery detail after restart: got %q want %q", got, want) + } + if got, want := loaded.RecoveryMetadata["queue-a/message-1"].Detail["reason"], "retry"; got != want { + t.Fatalf("unexpected recovery metadata after restart: got %q want %q", got, want) + } + + statePath := filepath.Join(baseDir, "instances", "instance-a", "sqs", sqliteFileName) + if _, err := os.Stat(statePath); err != nil { + t.Fatalf("expected sqlite database to exist: %v", err) + } +} + +func TestSQLiteRepositoryCloseIsIdempotentAndRejectsFurtherUse(t *testing.T) { + t.Helper() + + repo := mustOpenSQLiteRepository(t, "instance-a") + if err := repo.Close(); err != nil { + t.Fatalf("first close: %v", err) + } + if err := repo.Close(); err != nil { + t.Fatalf("second close: %v", err) + } + + if _, err := repo.Load(); err == nil { + t.Fatal("expected load after close to fail") + } + if err := repo.Save(domain.NewState()); err == nil { + t.Fatal("expected save after close to fail") + } +} + +func mustOpenSQLiteRepository(t *testing.T, instanceID string) *SQLiteRepository { + t.Helper() + return mustOpenSQLiteRepositoryAt(t, t.TempDir(), instanceID) +} + +func mustOpenSQLiteRepositoryAt(t *testing.T, baseDir, instanceID string) *SQLiteRepository { + t.Helper() + + storagePath, err := ResolveStoragePath(StorageConfig{ + BaseDir: baseDir, + InstanceID: instanceID, + }) + if err != nil { + t.Fatalf("resolve storage path: %v", err) + } + + repo, err := NewSQLiteRepository(storagePath) + if err != nil { + t.Fatalf("open repository: %v", err) + } + return repo +} diff --git a/core/internal/resources/sqs/application/service.go b/core/internal/resources/sqs/application/service.go new file mode 100644 index 0000000..304a203 --- /dev/null +++ b/core/internal/resources/sqs/application/service.go @@ -0,0 +1,575 @@ +package application + +import ( + "context" + "errors" + "fmt" + "sort" + "strconv" + "strings" + "sync" + "time" + + "github.com/michasdev/mildstack/core/internal/application/orchestrator" + "github.com/michasdev/mildstack/core/internal/resources/sqs/contracts" + "github.com/michasdev/mildstack/core/internal/resources/sqs/domain" + "github.com/michasdev/mildstack/core/internal/resources/sqs/infrastructure" +) + +var _ orchestrator.Service = (*Service)(nil) + +type Service struct { + state domain.State + policy orchestrator.EmulationPolicy + repo Repository + stateHook orchestrator.StateHook + clock Clock + worker *worker + cancel context.CancelFunc + done chan struct{} + mu sync.Mutex +} + +const ( + defaultVisibilityTimeout = 30 * time.Second + maxLongPollWait = 20 * time.Second + workerPollInterval = 50 * time.Millisecond + leaseVisibilityTimeoutMetaKey = "visibility_timeout_seconds" +) + +var ( + errQueueNotFound = errors.New("sqs: queue not found") + errReceiptHandleMismatch = errors.New("sqs: receipt handle does not match active lease") + errInvalidVisibilityWindow = errors.New("sqs: visibility timeout must be non-negative") +) + +func New() *Service { + return newService(domain.NewState(), nil) +} + +func newService(state domain.State, repo Repository) *Service { + return &Service{ + state: state.Clone(), + repo: repo, + clock: realClock{}, + policy: orchestrator.NewEmulationPolicy( + orchestrator.FidelityExemplar, + contracts.ActionNames(), + nil, + "sqs", + ), + } +} + +func (s *Service) Start(ctx context.Context) error { + s.mu.Lock() + defer s.mu.Unlock() + + if s.worker != nil { + return nil + } + + s.worker = newWorker(s, s.clock) + if ctx == nil { + ctx = context.Background() + } + runCtx, cancel := context.WithCancel(ctx) + s.cancel = cancel + s.done = make(chan struct{}) + + go func(done chan struct{}, runCtx context.Context, w *worker) { + defer close(done) + w.run(runCtx) + }(s.done, runCtx, s.worker) + + return nil +} + +func (s *Service) Stop(ctx context.Context) error { + s.mu.Lock() + cancel := s.cancel + done := s.done + s.cancel = nil + s.worker = nil + s.done = nil + s.mu.Unlock() + + if cancel != nil { + cancel() + } + if done != nil { + if ctx == nil { + ctx = context.Background() + } + waitCtx := ctx + if _, ok := waitCtx.Deadline(); !ok { + var cancelWait context.CancelFunc + waitCtx, cancelWait = context.WithTimeout(waitCtx, 2*time.Second) + defer cancelWait() + } + select { + case <-done: + case <-waitCtx.Done(): + return fmt.Errorf("sqs: worker shutdown timed out") + } + } + + s.mu.Lock() + defer s.mu.Unlock() + + if s.repo == nil { + return nil + } + + if err := s.repo.Close(); err != nil { + return fmt.Errorf("sqs: close repository: %w", err) + } + s.repo = nil + return nil +} + +func (s *Service) Metadata() orchestrator.Metadata { + return orchestrator.Metadata{ + Name: "sqs", + Description: "MildStack SQS real service", + Version: "v1", + Tags: []string{"aws", "messaging", "queue", "real-service"}, + } +} + +func (s *Service) Policy() orchestrator.EmulationPolicy { + return s.policy.Clone() +} + +func (s *Service) RegisterRoutes(registrar orchestrator.RouteRegistrar) error { + for _, route := range infrastructure.Routes() { + if err := registrar.Register(route); err != nil { + return err + } + } + return nil +} + +func (s *Service) AttachState(hook orchestrator.StateHook) error { + if hook == nil { + return fmt.Errorf("sqs: nil state hook") + } + + s.mu.Lock() + defer s.mu.Unlock() + + s.stateHook = hook + s.publishSnapshotLocked() + return nil +} + +func (s *Service) publishSnapshotLocked() { + if s.stateHook == nil { + return + } + + s.stateHook.Set(domain.StateKey, s.state.Snapshot()) +} + +func newServiceWithClock(state domain.State, repo Repository, clock Clock) *Service { + service := newService(state, repo) + if clock != nil { + service.clock = clock + } + return service +} + +func (s *Service) ReceiveMessage(queueName string, maxMessages int, waitTime time.Duration) ([]domain.Message, error) { + queueName = trimName(queueName) + if queueName == "" { + return nil, fmt.Errorf("sqs: queue name is required") + } + if maxMessages <= 0 { + maxMessages = 1 + } + if waitTime < 0 { + waitTime = 0 + } + if waitTime > maxLongPollWait { + waitTime = maxLongPollWait + } + + deadline := s.clock.Now().Add(waitTime) + for { + now := s.clock.Now() + messages, err := s.receiveReadyMessagesLocked(queueName, maxMessages, now) + if err != nil { + return nil, err + } + if len(messages) > 0 || waitTime == 0 || !now.Before(deadline) { + return messages, nil + } + + remaining := deadline.Sub(now) + sleep := workerPollInterval + if remaining < sleep { + sleep = remaining + } + if sleep <= 0 { + return messages, nil + } + s.clock.Sleep(sleep) + } +} + +func (s *Service) DeleteMessage(queueName string, receiptHandle string) error { + queueName = trimName(queueName) + receiptHandle = trimName(receiptHandle) + if queueName == "" { + return fmt.Errorf("sqs: queue name is required") + } + if receiptHandle == "" { + return fmt.Errorf("sqs: receipt handle is required") + } + + s.mu.Lock() + defer s.mu.Unlock() + + idx, ok := s.findMessageByReceiptLocked(queueName, receiptHandle) + if !ok { + return errReceiptHandleMismatch + } + + s.state.Messages = append(s.state.Messages[:idx], s.state.Messages[idx+1:]...) + return s.commitStateLocked() +} + +func (s *Service) ChangeMessageVisibility(queueName string, receiptHandle string, visibility time.Duration) error { + queueName = trimName(queueName) + receiptHandle = trimName(receiptHandle) + if queueName == "" { + return fmt.Errorf("sqs: queue name is required") + } + if receiptHandle == "" { + return fmt.Errorf("sqs: receipt handle is required") + } + if visibility < 0 { + return errInvalidVisibilityWindow + } + if visibility > 12*time.Hour { + visibility = 12 * time.Hour + } + + s.mu.Lock() + defer s.mu.Unlock() + + idx, ok := s.findMessageByReceiptLocked(queueName, receiptHandle) + if !ok { + return errReceiptHandleMismatch + } + + message := &s.state.Messages[idx] + if message.Metadata == nil { + message.Metadata = map[string]string{} + } + if visibility == 0 { + message.ReceivedAt = time.Time{} + message.AvailableAt = s.clock.Now() + message.Metadata[leaseVisibilityTimeoutMetaKey] = "0" + } else { + message.ReceivedAt = s.clock.Now() + message.Metadata[leaseVisibilityTimeoutMetaKey] = strconv.FormatInt(int64(visibility/time.Second), 10) + } + + return s.commitStateLocked() +} + +func (s *Service) commitStateLocked() error { + if s.repo != nil { + if err := s.repo.Save(s.state.Clone()); err != nil { + return fmt.Errorf("sqs: save repository: %w", err) + } + } + s.publishSnapshotLocked() + return nil +} + +func (s *Service) queueByNameLocked(name string) (domain.Queue, bool) { + for _, queue := range s.state.Queues { + if queue.Name == name { + return queue, true + } + } + return domain.Queue{}, false +} + +func (s *Service) receiveReadyMessagesLocked(queueName string, maxMessages int, now time.Time) ([]domain.Message, error) { + s.mu.Lock() + defer s.mu.Unlock() + + queue, ok := s.queueByNameLocked(queueName) + if !ok { + return nil, errQueueNotFound + } + + indices := make([]int, 0, len(s.state.Messages)) + for idx, message := range s.state.Messages { + if !messageVisibleInQueue(message, queueName) { + continue + } + if IsDelayed(message, now) || IsInvisible(message, queue, now) { + continue + } + if IsFIFOQueue(queue) && fifoDeliveryBlocked(s.state.Messages, idx, queueName) { + continue + } + indices = append(indices, idx) + } + sortMessagesByDelivery(s.state.Messages, indices, queue) + + selected := make([]domain.Message, 0, maxMessages) + for _, idx := range indices { + message := &s.state.Messages[idx] + if message.Metadata == nil { + message.Metadata = map[string]string{} + } + timeout := queueVisibilityTimeout(queue, *message) + message.ReceivedAt = now + message.Recovery.Attempts++ + message.Metadata[leaseVisibilityTimeoutMetaKey] = strconv.FormatInt(int64(timeout/time.Second), 10) + message.ReceiptKeys = append(message.ReceiptKeys, nextReceiptHandle(queueName, *message)) + selected = append(selected, cloneMessage(*message)) + if len(selected) == maxMessages { + break + } + } + + if len(selected) == 0 { + return nil, nil + } + if err := s.commitStateLocked(); err != nil { + return nil, err + } + + return selected, nil +} + +func (s *Service) findMessageByReceiptLocked(queueName, receiptHandle string) (int, bool) { + for idx, message := range s.state.Messages { + if !messageVisibleInQueue(message, queueName) { + continue + } + if CurrentReceiptHandle(message) == receiptHandle { + return idx, true + } + } + return -1, false +} + +func sortMessagesByDelivery(messages []domain.Message, indices []int, queue domain.Queue) { + sort.SliceStable(indices, func(i, j int) bool { + left := messages[indices[i]] + right := messages[indices[j]] + if IsFIFOQueue(queue) { + if left.MessageGroupID != right.MessageGroupID { + return left.MessageGroupID < right.MessageGroupID + } + if left.SequenceNumber != right.SequenceNumber && left.SequenceNumber > 0 && right.SequenceNumber > 0 { + return left.SequenceNumber < right.SequenceNumber + } + } + if !left.SentAt.Equal(right.SentAt) { + return left.SentAt.Before(right.SentAt) + } + if !left.AvailableAt.Equal(right.AvailableAt) { + return left.AvailableAt.Before(right.AvailableAt) + } + return left.MessageID < right.MessageID + }) +} + +func cloneMessage(message domain.Message) domain.Message { + cloned := message + cloned.Attributes = cloneMap(message.Attributes) + cloned.Metadata = cloneMap(message.Metadata) + cloned.Tags = append([]string(nil), message.Tags...) + cloned.ReceiptKeys = append([]string(nil), message.ReceiptKeys...) + cloned.Recovery.Detail = cloneMap(message.Recovery.Detail) + return cloned +} + +func queueVisibilityTimeout(queue domain.Queue, message domain.Message) time.Duration { + if timeout := parseMessageVisibilityTimeout(message.Metadata[leaseVisibilityTimeoutMetaKey]); timeout > 0 { + return timeout + } + if queue.Attributes != nil { + if timeout := parseMessageVisibilityTimeout(queue.Attributes["VisibilityTimeout"]); timeout > 0 { + return timeout + } + } + return defaultVisibilityTimeout +} + +func parseMessageVisibilityTimeout(raw string) time.Duration { + raw = trimName(raw) + if raw == "" { + return 0 + } + seconds, err := strconv.Atoi(raw) + if err != nil || seconds < 0 { + return 0 + } + return time.Duration(seconds) * time.Second +} + +func nextReceiptHandle(queueName string, message domain.Message) string { + return fmt.Sprintf("%s/%s/%d", queueName, message.MessageID, len(message.ReceiptKeys)+1) +} + +func trimName(value string) string { + return strings.TrimSpace(value) +} + +func cloneMap(values map[string]string) map[string]string { + if values == nil { + return nil + } + + cloned := make(map[string]string, len(values)) + for key, value := range values { + cloned[key] = value + } + return cloned +} + +func messageVisibleInQueue(message domain.Message, queueName string) bool { + if message.Queue != queueName { + return false + } + if trimName(message.DeadLetterSourceQueue) == queueName && trimName(message.DeadLetterQueue) != queueName { + return false + } + return true +} + +func fifoDeliveryBlocked(messages []domain.Message, candidateIndex int, queueName string) bool { + candidate := messages[candidateIndex] + groupID := trimName(candidate.MessageGroupID) + if groupID == "" { + return false + } + + for idx, other := range messages { + if idx == candidateIndex || !messageVisibleInQueue(other, queueName) { + continue + } + if trimName(other.MessageGroupID) != groupID { + continue + } + if compareFIFOMessageOrder(other, candidate) < 0 { + return true + } + } + return false +} + +func compareFIFOMessageOrder(left, right domain.Message) int { + if left.SequenceNumber > 0 && right.SequenceNumber > 0 && left.SequenceNumber != right.SequenceNumber { + if left.SequenceNumber < right.SequenceNumber { + return -1 + } + return 1 + } + if !left.SentAt.Equal(right.SentAt) { + if left.SentAt.Before(right.SentAt) { + return -1 + } + return 1 + } + if !left.AvailableAt.Equal(right.AvailableAt) { + if left.AvailableAt.Before(right.AvailableAt) { + return -1 + } + return 1 + } + switch { + case left.MessageID < right.MessageID: + return -1 + case left.MessageID > right.MessageID: + return 1 + default: + return 0 + } +} + +func queueOrderingMode(queue domain.Queue) string { + if IsFIFOQueue(queue) { + return "fifo" + } + return "standard" +} + +func deadLetterThresholdFromQueue(queue domain.Queue) int { + return deadLetterThreshold(queue) +} + +func (s *Service) deadLetterEligibleLocked(message domain.Message, queue domain.Queue, now time.Time) bool { + return IsDeadLetterEligible(message, queue, now) +} + +func (s *Service) sweepDeadLettersLocked(now time.Time) int { + mutated := 0 + for idx := range s.state.Messages { + message := s.state.Messages[idx] + queue, ok := s.queueByNameLocked(message.Queue) + if !ok { + continue + } + if !s.deadLetterEligibleLocked(message, queue, now) { + continue + } + if s.moveMessageToDeadLetterLocked(idx, queue, now) { + mutated++ + } + } + if mutated > 0 { + _ = s.commitStateLocked() + } + return mutated +} + +func (s *Service) moveMessageToDeadLetterLocked(index int, sourceQueue domain.Queue, now time.Time) bool { + message := &s.state.Messages[index] + dlqName := trimName(sourceQueue.Recovery.DeadLetterQueue) + if dlqName == "" || trimName(message.DeadLetterQueue) == dlqName { + return false + } + + message.DeadLetterQueue = dlqName + message.DeadLetterSourceQueue = sourceQueue.Name + message.DeadLetteredAt = now + message.ReceivedAt = time.Time{} + message.AvailableAt = now + if message.Metadata == nil { + message.Metadata = map[string]string{} + } + message.Metadata["dead_letter_queue"] = dlqName + message.Metadata["dead_letter_source_queue"] = sourceQueue.Name + message.Metadata["dead_lettered_at"] = now.UTC().Format(time.RFC3339Nano) + + if dlqQueue, ok := s.queueByNameLocked(dlqName); ok { + message.Queue = dlqQueue.Name + } + + if s.state.RecoveryMetadata == nil { + s.state.RecoveryMetadata = map[string]domain.RecoveryMetadata{} + } + s.state.RecoveryMetadata[sourceQueue.Name+"/"+message.MessageID] = domain.RecoveryMetadata{ + Queue: dlqName, + Message: message.MessageID, + Detail: map[string]string{ + "source_queue": sourceQueue.Name, + "dead_letter_queue": dlqName, + "attempts": strconv.Itoa(message.Recovery.Attempts), + "dead_lettered_at": now.UTC().Format(time.RFC3339Nano), + "dead_letter_reason": "max_receive_count", + "ordering_mode": queueOrderingMode(sourceQueue), + }, + } + return true +} diff --git a/core/internal/resources/sqs/application/service_persistence.go b/core/internal/resources/sqs/application/service_persistence.go new file mode 100644 index 0000000..5ea0ae8 --- /dev/null +++ b/core/internal/resources/sqs/application/service_persistence.go @@ -0,0 +1,23 @@ +package application + +import "fmt" + +func NewWithPersistence(config StorageConfig) (*Service, error) { + storagePath, err := ResolveStoragePath(config) + if err != nil { + return nil, err + } + + repo, err := NewSQLiteRepository(storagePath) + if err != nil { + return nil, err + } + + state, err := repo.Load() + if err != nil { + _ = repo.Close() + return nil, fmt.Errorf("sqs: load persisted state: %w", err) + } + + return newService(state, repo), nil +} diff --git a/core/internal/resources/sqs/application/service_test.go b/core/internal/resources/sqs/application/service_test.go new file mode 100644 index 0000000..309e247 --- /dev/null +++ b/core/internal/resources/sqs/application/service_test.go @@ -0,0 +1,624 @@ +package application + +import ( + "context" + "testing" + "time" + + "github.com/michasdev/mildstack/core/internal/application/orchestrator" + "github.com/michasdev/mildstack/core/internal/application/runtime" + deliveryhttp "github.com/michasdev/mildstack/core/internal/delivery/http" + "github.com/michasdev/mildstack/core/internal/resources/sqs/domain" +) + +func TestSQSServiceMetadataRoutesAndPolicy(t *testing.T) { + t.Helper() + + service := New() + if _, ok := any(service).(orchestrator.Service); !ok { + t.Fatal("expected service to satisfy orchestrator.Service") + } + + metadata := service.Metadata() + if got, want := metadata.Name, "sqs"; got != want { + t.Fatalf("unexpected service name: got %q want %q", got, want) + } + if got, want := metadata.Version, "v1"; got != want { + t.Fatalf("unexpected service version: got %q want %q", got, want) + } + if got, want := metadata.Description, "MildStack SQS real service"; got != want { + t.Fatalf("unexpected service description: got %q want %q", got, want) + } + + expectedTags := []string{"aws", "messaging", "queue", "real-service"} + if got, want := len(metadata.Tags), len(expectedTags); got != want { + t.Fatalf("unexpected tag count: got %d want %d", got, want) + } + for i, tag := range expectedTags { + if metadata.Tags[i] != tag { + t.Fatalf("unexpected tag at %d: got %q want %q", i, metadata.Tags[i], tag) + } + } + + policy := service.Policy() + if got, want := policy.Fidelity, orchestrator.FidelityExemplar; got != want { + t.Fatalf("unexpected policy fidelity: got %q want %q", got, want) + } + if got, want := policy.ErrorPrefix, "sqs"; got != want { + t.Fatalf("unexpected policy error prefix: got %q want %q", got, want) + } + if got, want := len(policy.Supported), 23; got != want { + t.Fatalf("unexpected supported count: got %d want %d", got, want) + } + if got, want := len(policy.Unsupported), 0; got != want { + t.Fatalf("unexpected unsupported count: got %d want %d", got, want) + } + + policy.Supported[0] = "changed" + again := service.Policy() + if got, want := again.Supported[0], "AddPermission"; got != want { + t.Fatalf("policy supported slice was not copied: got %q want %q", got, want) + } + + registrar := deliveryhttp.NewRegistrar() + if err := service.RegisterRoutes(registrar); err != nil { + t.Fatalf("register routes: %v", err) + } + + entry, ok := registrar.Service("sqs") + if !ok { + t.Fatal("expected sqs service to be registered") + } + if got, want := len(entry.Routes), 7; got != want { + t.Fatalf("unexpected route count: got %d want %d", got, want) + } + assertRouteExists(t, entry.Routes, "GET", "/api/v1/runtime/services/sqs/queues") + assertRouteExists(t, entry.Routes, "POST", "/api/v1/runtime/services/sqs/queues") + assertRouteExists(t, entry.Routes, "GET", "/api/v1/runtime/services/sqs/queues/:queue") + assertRouteExists(t, entry.Routes, "DELETE", "/api/v1/runtime/services/sqs/queues/:queue") + assertRouteExists(t, entry.Routes, "GET", "/api/v1/runtime/services/sqs/queues/:queue/messages") + assertRouteExists(t, entry.Routes, "POST", "/api/v1/runtime/services/sqs/queues/:queue/messages") + assertRouteExists(t, entry.Routes, "DELETE", "/api/v1/runtime/services/sqs/queues/:queue/messages/:receiptHandle") +} + +func TestSQSServiceAttachStateUsesNamespacedCopySafeSnapshot(t *testing.T) { + t.Helper() + + service := newService(domain.State{ + Service: "sqs", + Queues: []domain.Queue{ + { + Name: "queue-a", + Attributes: map[string]string{ + "VisibilityTimeout": "30", + }, + Recovery: domain.QueueRecovery{ + DeadLetterQueue: "queue-dlq", + }, + }, + }, + Messages: []domain.Message{ + { + Queue: "queue-a", + MessageID: "message-1", + Body: "payload", + Tags: []string{"alpha"}, + Metadata: map[string]string{"trace": "abc"}, + ReceiptKeys: []string{"r-1"}, + }, + }, + }, nil) + + hook := runtime.NewStateHook() + if err := service.AttachState(hook); err != nil { + t.Fatalf("attach state: %v", err) + } + + value, ok := hook.Get(domain.StateKey) + if !ok { + t.Fatalf("expected state for %q to be present", domain.StateKey) + } + state := value.(map[string]any) + if got, want := state["service"], "sqs"; got != want { + t.Fatalf("unexpected service name: got %v want %v", got, want) + } + queues := state["queues"].([]any) + if got, want := len(queues), 1; got != want { + t.Fatalf("unexpected queue count: got %d want %d", got, want) + } + queues[0].(map[string]any)["name"] = "mutated" + queues[0].(map[string]any)["attributes"].(map[string]any)["VisibilityTimeout"] = "99" + + messages := state["messages"].([]any) + if got, want := len(messages), 1; got != want { + t.Fatalf("unexpected message count: got %d want %d", got, want) + } + messages[0].(map[string]any)["body"] = "mutated" + messages[0].(map[string]any)["tags"].([]string)[0] = "mutated" + + if got, want := service.state.Queues[0].Name, "queue-a"; got != want { + t.Fatalf("service queue name was aliased: got %q want %q", got, want) + } + if got, want := service.state.Queues[0].Attributes["VisibilityTimeout"], "30"; got != want { + t.Fatalf("service queue attributes were aliased: got %q want %q", got, want) + } + if got, want := service.state.Messages[0].Body, "payload"; got != want { + t.Fatalf("service message body was aliased: got %q want %q", got, want) + } + if got, want := service.state.Messages[0].Tags[0], "alpha"; got != want { + t.Fatalf("service message tags were aliased: got %q want %q", got, want) + } +} + +func TestSQSServiceReceiveMessageRespectsStandardOrderAndFifoOrdering(t *testing.T) { + t.Helper() + + clock := newManualClock(time.Date(2026, time.April, 19, 12, 0, 0, 0, time.UTC)) + service := newServiceWithClock(domain.State{ + Service: "sqs", + Queues: []domain.Queue{ + { + Name: "queue-standard", + Attributes: map[string]string{ + "VisibilityTimeout": "30", + }, + }, + { + Name: "queue-fifo", + OrderingHint: "fifo", + Attributes: map[string]string{ + "VisibilityTimeout": "30", + "FifoQueue": "true", + }, + }, + }, + Messages: []domain.Message{ + { + Queue: "queue-standard", + MessageID: "std-late", + Body: "late", + SentAt: clock.Now().Add(time.Second), + }, + { + Queue: "queue-standard", + MessageID: "std-early", + Body: "early", + SentAt: clock.Now(), + }, + { + Queue: "queue-fifo", + MessageID: "fifo-second", + Body: "second", + MessageGroupID: "group-a", + SequenceNumber: 2, + SentAt: clock.Now().Add(2 * time.Second), + }, + { + Queue: "queue-fifo", + MessageID: "fifo-first", + Body: "first", + MessageGroupID: "group-a", + SequenceNumber: 1, + SentAt: clock.Now().Add(time.Second), + }, + }, + }, nil, clock) + + standard, err := service.ReceiveMessage("queue-standard", 2, 0) + if err != nil { + t.Fatalf("receive standard messages: %v", err) + } + if got, want := len(standard), 2; got != want { + t.Fatalf("unexpected standard count: got %d want %d", got, want) + } + if got, want := standard[0].MessageID, "std-early"; got != want { + t.Fatalf("unexpected standard first message: got %q want %q", got, want) + } + if got, want := standard[1].MessageID, "std-late"; got != want { + t.Fatalf("unexpected standard second message: got %q want %q", got, want) + } + + fifo, err := service.ReceiveMessage("queue-fifo", 2, 0) + if err != nil { + t.Fatalf("receive fifo messages: %v", err) + } + if got, want := len(fifo), 1; got != want { + t.Fatalf("unexpected fifo count: got %d want %d", got, want) + } + if got, want := fifo[0].MessageID, "fifo-first"; got != want { + t.Fatalf("unexpected fifo first message: got %q want %q", got, want) + } +} + +func TestSQSServiceDeadLetterEligibilityUsesRecoveryPolicy(t *testing.T) { + t.Helper() + + now := time.Date(2026, time.April, 19, 12, 0, 0, 0, time.UTC) + service := newService(domain.State{ + Service: "sqs", + Queues: []domain.Queue{ + { + Name: "queue-a", + Recovery: domain.QueueRecovery{ + DeadLetterQueue: "queue-dlq", + Policy: map[string]string{ + "max_receive_count": "3", + }, + }, + }, + }, + Messages: []domain.Message{ + { + Queue: "queue-a", + MessageID: "message-1", + ReceivedAt: now.Add(-31 * time.Second), + Recovery: domain.MessageRecovery{ + Attempts: 3, + }, + }, + }, + }, nil) + + queue, ok := service.queueByNameLocked("queue-a") + if !ok { + t.Fatal("expected queue to exist") + } + if !service.deadLetterEligibleLocked(service.state.Messages[0], queue, now) { + t.Fatal("expected message to be dead-letter eligible") + } +} + +func TestSQSServiceNewWithPersistenceLoadsRepositoryState(t *testing.T) { + t.Helper() + + baseDir := t.TempDir() + config := StorageConfig{BaseDir: baseDir, InstanceID: "instance-a"} + storagePath, err := ResolveStoragePath(config) + if err != nil { + t.Fatalf("resolve storage path: %v", err) + } + + repo, err := NewSQLiteRepository(storagePath) + if err != nil { + t.Fatalf("open repository: %v", err) + } + + state := domain.NewState() + state.Queues = append(state.Queues, domain.Queue{ + Name: "queue-a", + Attributes: map[string]string{ + "VisibilityTimeout": "45", + }, + OrderingHint: "fifo", + Recovery: domain.QueueRecovery{ + DeadLetterQueue: "queue-dlq", + }, + CreatedAt: time.Date(2026, time.April, 19, 12, 0, 0, 0, time.UTC), + }) + state.Messages = append(state.Messages, domain.Message{ + Queue: "queue-a", + MessageID: "message-1", + Body: "payload", + Tags: []string{"persisted"}, + MessageGroupID: "group-a", + SequenceNumber: 9, + BatchID: "batch-a", + BatchEntryID: "entry-a", + BatchEntryIndex: 0, + BatchEntryCount: 1, + DeadLetterQueue: "queue-dlq", + DeadLetterSourceQueue: "queue-a", + DeadLetteredAt: time.Date(2026, time.April, 19, 12, 5, 0, 0, time.UTC), + AvailableAt: time.Date(2026, time.April, 19, 12, 3, 0, 0, time.UTC), + ReceivedAt: time.Date(2026, time.April, 19, 12, 4, 0, 0, time.UTC), + ReceiptKeys: []string{"r-1", "r-2"}, + }) + if err := repo.Save(state); err != nil { + _ = repo.Close() + t.Fatalf("save seeded state: %v", err) + } + if err := repo.Close(); err != nil { + t.Fatalf("close seeded repository: %v", err) + } + + service, err := NewWithPersistence(config) + if err != nil { + t.Fatalf("new with persistence: %v", err) + } + defer func() { + if err := service.Stop(context.Background()); err != nil { + t.Fatalf("stop service: %v", err) + } + }() + + if service.repo == nil { + t.Fatal("expected persistent repository to be attached") + } + if got, want := len(service.state.Queues), 1; got != want { + t.Fatalf("unexpected queue count after load: got %d want %d", got, want) + } + if got, want := service.state.Queues[0].Name, "queue-a"; got != want { + t.Fatalf("unexpected queue name after load: got %q want %q", got, want) + } + if got, want := service.state.Queues[0].OrderingHint, "fifo"; got != want { + t.Fatalf("unexpected queue ordering after load: got %q want %q", got, want) + } + if got, want := service.state.Messages[0].MessageID, "message-1"; got != want { + t.Fatalf("unexpected message id after load: got %q want %q", got, want) + } + if got, want := service.state.Messages[0].AvailableAt, time.Date(2026, time.April, 19, 12, 3, 0, 0, time.UTC); !got.Equal(want) { + t.Fatalf("unexpected available_at after load: got %v want %v", got, want) + } + if got, want := service.state.Messages[0].ReceivedAt, time.Date(2026, time.April, 19, 12, 4, 0, 0, time.UTC); !got.Equal(want) { + t.Fatalf("unexpected received_at after load: got %v want %v", got, want) + } + if got, want := service.state.Messages[0].ReceiptKeys[1], "r-2"; got != want { + t.Fatalf("unexpected receipt handle history after load: got %q want %q", got, want) + } + if got, want := service.state.Messages[0].MessageGroupID, "group-a"; got != want { + t.Fatalf("unexpected message group after load: got %q want %q", got, want) + } + if got, want := service.state.Messages[0].DeadLetterQueue, "queue-dlq"; got != want { + t.Fatalf("unexpected dead letter queue after load: got %q want %q", got, want) + } +} + +func TestSQSServiceStopClosesRepositoryIdempotently(t *testing.T) { + t.Helper() + + repo := &repositoryStub{} + service := newService(domain.NewState(), repo) + + if err := service.Stop(context.Background()); err != nil { + t.Fatalf("first stop: %v", err) + } + if err := service.Stop(context.Background()); err != nil { + t.Fatalf("second stop: %v", err) + } + if got, want := repo.closeCount, 1; got != want { + t.Fatalf("unexpected close count: got %d want %d", got, want) + } + if service.repo != nil { + t.Fatal("expected repository handle to be cleared after stop") + } +} + +func TestSQSServiceReceiveMessageHonorsDelayAndBoundsLongPoll(t *testing.T) { + t.Helper() + + clock := newManualClock(time.Date(2026, time.April, 19, 12, 0, 0, 0, time.UTC)) + service := newServiceWithClock(domain.State{ + Service: "sqs", + Queues: []domain.Queue{ + { + Name: "queue-a", + Attributes: map[string]string{ + "VisibilityTimeout": "30", + }, + }, + }, + Messages: []domain.Message{ + { + Queue: "queue-a", + MessageID: "message-1", + Body: "payload", + SentAt: clock.Now(), + AvailableAt: clock.Now().Add(150 * time.Millisecond), + }, + }, + }, nil, clock) + + messages, err := service.ReceiveMessage("queue-a", 1, 2*time.Second) + if err != nil { + t.Fatalf("receive message: %v", err) + } + if got, want := len(messages), 1; got != want { + t.Fatalf("unexpected message count: got %d want %d", got, want) + } + if got, want := messages[0].MessageID, "message-1"; got != want { + t.Fatalf("unexpected message id: got %q want %q", got, want) + } + if got := CurrentReceiptHandle(messages[0]); got == "" { + t.Fatal("expected receive to issue a receipt handle") + } + if got := clock.SleepCount(); got == 0 { + t.Fatal("expected long poll to sleep at least once") + } + if !messages[0].ReceivedAt.After(time.Time{}) { + t.Fatal("expected receive to stamp the delivery time") + } +} + +func TestSQSServiceCapsLongPollAtTwentySeconds(t *testing.T) { + t.Helper() + + clock := newManualClock(time.Date(2026, time.April, 19, 12, 0, 0, 0, time.UTC)) + service := newServiceWithClock(domain.State{ + Service: "sqs", + Queues: []domain.Queue{ + { + Name: "queue-a", + }, + }, + }, nil, clock) + + messages, err := service.ReceiveMessage("queue-a", 1, 30*time.Second) + if err != nil { + t.Fatalf("receive message: %v", err) + } + if got, want := len(messages), 0; got != want { + t.Fatalf("expected no messages, got %d", got) + } + if got, want := clock.totalSleep, 20*time.Second; got != want { + t.Fatalf("unexpected capped sleep total: got %v want %v", got, want) + } +} + +func TestSQSServiceRejectsStaleReceiptHandlesAfterRedelivery(t *testing.T) { + t.Helper() + + now := time.Date(2026, time.April, 19, 12, 0, 0, 0, time.UTC) + clock := newManualClock(now) + service := newServiceWithClock(domain.State{ + Service: "sqs", + Queues: []domain.Queue{ + { + Name: "queue-a", + Attributes: map[string]string{ + "VisibilityTimeout": "30", + }, + }, + }, + Messages: []domain.Message{ + { + Queue: "queue-a", + MessageID: "message-1", + Body: "payload", + SentAt: now, + }, + }, + }, nil, clock) + + first, err := service.ReceiveMessage("queue-a", 1, 0) + if err != nil { + t.Fatalf("first receive: %v", err) + } + if got, want := len(first), 1; got != want { + t.Fatalf("unexpected first receive count: got %d want %d", got, want) + } + firstHandle := CurrentReceiptHandle(first[0]) + if firstHandle == "" { + t.Fatal("expected first receive to issue a receipt handle") + } + + clock.Sleep(31 * time.Second) + second, err := service.ReceiveMessage("queue-a", 1, 0) + if err != nil { + t.Fatalf("second receive: %v", err) + } + if got, want := len(second), 1; got != want { + t.Fatalf("unexpected second receive count: got %d want %d", got, want) + } + secondHandle := CurrentReceiptHandle(second[0]) + if secondHandle == "" || secondHandle == firstHandle { + t.Fatalf("expected a rotated receipt handle, got %q and %q", firstHandle, secondHandle) + } + + if err := service.DeleteMessage("queue-a", firstHandle); err == nil { + t.Fatal("expected stale receipt handle delete to fail") + } + if err := service.DeleteMessage("queue-a", secondHandle); err != nil { + t.Fatalf("delete with current handle: %v", err) + } +} + +func TestSQSServiceChangeMessageVisibilityPostponesRedelivery(t *testing.T) { + t.Helper() + + now := time.Date(2026, time.April, 19, 12, 0, 0, 0, time.UTC) + clock := newManualClock(now) + service := newServiceWithClock(domain.State{ + Service: "sqs", + Queues: []domain.Queue{ + { + Name: "queue-a", + Attributes: map[string]string{ + "VisibilityTimeout": "30", + }, + }, + }, + Messages: []domain.Message{ + { + Queue: "queue-a", + MessageID: "message-1", + Body: "payload", + SentAt: now, + }, + }, + }, nil, clock) + + first, err := service.ReceiveMessage("queue-a", 1, 0) + if err != nil { + t.Fatalf("first receive: %v", err) + } + handle := CurrentReceiptHandle(first[0]) + if handle == "" { + t.Fatal("expected first receive to issue a receipt handle") + } + + if err := service.ChangeMessageVisibility("queue-a", handle, 2*time.Minute); err != nil { + t.Fatalf("change visibility: %v", err) + } + + clock.Sleep(31 * time.Second) + messages, err := service.ReceiveMessage("queue-a", 1, 0) + if err != nil { + t.Fatalf("receive before extended visibility expires: %v", err) + } + if got, want := len(messages), 0; got != want { + t.Fatalf("expected message to remain hidden, got %d message(s)", got) + } + + clock.Sleep(90 * time.Second) + messages, err = service.ReceiveMessage("queue-a", 1, 0) + if err != nil { + t.Fatalf("receive after extended visibility expires: %v", err) + } + if got, want := len(messages), 1; got != want { + t.Fatalf("expected message to redeliver after visibility change, got %d message(s)", got) + } +} + +type manualClock struct { + now time.Time + sleepCount int + totalSleep time.Duration +} + +func newManualClock(now time.Time) *manualClock { + return &manualClock{now: now} +} + +func (c *manualClock) Now() time.Time { + return c.now +} + +func (c *manualClock) Sleep(duration time.Duration) { + if duration <= 0 { + return + } + c.sleepCount++ + c.totalSleep += duration + c.now = c.now.Add(duration) +} + +func (c *manualClock) SleepCount() int { + return c.sleepCount +} + +func assertRouteExists(t *testing.T, routes []deliveryhttp.RegisteredRoute, method, path string) { + t.Helper() + for _, route := range routes { + if route.Method == method && route.Path == path { + return + } + } + t.Fatalf("expected route %s %s to be registered", method, path) +} + +type repositoryStub struct { + closeCount int +} + +func (r *repositoryStub) Load() (domain.State, error) { + return domain.NewState(), nil +} + +func (r *repositoryStub) Save(state domain.State) error { + return nil +} + +func (r *repositoryStub) Close() error { + r.closeCount++ + return nil +} diff --git a/core/internal/resources/sqs/application/storage_paths.go b/core/internal/resources/sqs/application/storage_paths.go new file mode 100644 index 0000000..2814919 --- /dev/null +++ b/core/internal/resources/sqs/application/storage_paths.go @@ -0,0 +1,23 @@ +package application + +import ( + "strings" + + "github.com/michasdev/mildstack/core/internal/application/runtime" + "github.com/michasdev/mildstack/core/internal/resources/instancepath" +) + +type StorageConfig struct { + BaseDir string + InstanceID string +} + +func ResolveStoragePath(config StorageConfig) (string, error) { + baseDir := strings.TrimSpace(config.BaseDir) + if baseDir == "" { + paths := runtime.ResolvePaths() + baseDir = paths.BaseDir + } + + return instancepath.Resolve(baseDir, config.InstanceID, "sqs") +} diff --git a/core/internal/resources/sqs/application/storage_paths_test.go b/core/internal/resources/sqs/application/storage_paths_test.go new file mode 100644 index 0000000..44c9716 --- /dev/null +++ b/core/internal/resources/sqs/application/storage_paths_test.go @@ -0,0 +1,66 @@ +package application + +import ( + "path/filepath" + "testing" + + "github.com/michasdev/mildstack/core/internal/application/runtime" +) + +func TestResolveStoragePathUsesInstanceScopedLayout(t *testing.T) { + t.Helper() + + baseDir := t.TempDir() + path, err := ResolveStoragePath(StorageConfig{ + BaseDir: baseDir, + InstanceID: "instance-a", + }) + if err != nil { + t.Fatalf("resolve storage path: %v", err) + } + + want := filepath.Join(baseDir, "instances", "instance-a", "sqs") + if got, want := path, want; got != want { + t.Fatalf("unexpected storage path: got %q want %q", got, want) + } + + otherPath, err := ResolveStoragePath(StorageConfig{ + BaseDir: baseDir, + InstanceID: "instance-b", + }) + if err != nil { + t.Fatalf("resolve other storage path: %v", err) + } + if path == otherPath { + t.Fatalf("expected distinct storage paths, got %q", path) + } +} + +func TestResolveStoragePathFallsBackToRuntimeBaseDir(t *testing.T) { + t.Helper() + + expectedBase := runtime.ResolvePaths().BaseDir + if expectedBase == "" { + t.Fatal("expected runtime base dir to be available") + } + + path, err := ResolveStoragePath(StorageConfig{ + InstanceID: "instance-a", + }) + if err != nil { + t.Fatalf("resolve storage path with runtime base dir: %v", err) + } + + want := filepath.Join(expectedBase, "instances", "instance-a", "sqs") + if got, want := path, want; got != want { + t.Fatalf("unexpected fallback storage path: got %q want %q", got, want) + } +} + +func TestResolveStoragePathRejectsEmptyInstanceID(t *testing.T) { + t.Helper() + + if _, err := ResolveStoragePath(StorageConfig{BaseDir: t.TempDir()}); err == nil { + t.Fatal("expected empty instance id to be rejected") + } +} diff --git a/core/internal/resources/sqs/application/worker.go b/core/internal/resources/sqs/application/worker.go new file mode 100644 index 0000000..1a76735 --- /dev/null +++ b/core/internal/resources/sqs/application/worker.go @@ -0,0 +1,135 @@ +package application + +import ( + "context" + "time" + + "github.com/michasdev/mildstack/core/internal/resources/sqs/domain" +) + +type worker struct { + service *Service + clock Clock + tick time.Duration +} + +type leaseSnapshot struct { + Queue string + Message string + ReadyIn time.Duration +} + +func newWorker(service *Service, clock Clock) *worker { + if clock == nil { + clock = realClock{} + } + return &worker{ + service: service, + clock: clock, + tick: workerPollInterval, + } +} + +func (w *worker) run(ctx context.Context) { + for { + if ctx.Err() != nil { + return + } + + now := w.clock.Now() + wait := w.poll(now) + if wait <= 0 || wait > w.tick { + wait = w.tick + } + + timer := time.NewTimer(wait) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + + w.advance(w.clock.Now()) + } +} + +func (w *worker) poll(now time.Time) time.Duration { + w.service.mu.Lock() + defer w.service.mu.Unlock() + + next := w.tick + for _, message := range w.service.state.Messages { + queue, ok := w.service.queueByNameLocked(message.Queue) + if !ok { + continue + } + + if IsDelayed(message, now) { + if wait := message.AvailableAt.Sub(now); wait >= 0 && wait < next { + next = wait + } + continue + } + + if IsInvisible(message, queue, now) { + if wait := leaseDeadline(message, queue).Sub(now); wait >= 0 && wait < next { + next = wait + } + } + } + + if next < 0 { + return 0 + } + return next +} + +func (w *worker) lease(now time.Time) []leaseSnapshot { + w.service.mu.Lock() + defer w.service.mu.Unlock() + + leases := make([]leaseSnapshot, 0) + for _, message := range w.service.state.Messages { + queue, ok := w.service.queueByNameLocked(message.Queue) + if !ok { + continue + } + if !IsInvisible(message, queue, now) { + continue + } + readyIn := leaseDeadline(message, queue).Sub(now) + if readyIn < 0 { + readyIn = 0 + } + leases = append(leases, leaseSnapshot{ + Queue: message.Queue, + Message: message.MessageID, + ReadyIn: readyIn, + }) + } + return leases +} + +func (w *worker) redeliver(now time.Time) []domain.Message { + w.service.mu.Lock() + defer w.service.mu.Unlock() + + redeliverable := make([]domain.Message, 0) + for _, message := range w.service.state.Messages { + queue, ok := w.service.queueByNameLocked(message.Queue) + if !ok { + continue + } + if CanRedeliver(message, queue, now) && !IsDeadLetterEligible(message, queue, now) { + redeliverable = append(redeliverable, cloneMessage(message)) + } + } + return redeliverable +} + +func (w *worker) advance(now time.Time) int { + w.service.mu.Lock() + defer w.service.mu.Unlock() + return w.service.sweepDeadLettersLocked(now) +} diff --git a/core/internal/resources/sqs/application/worker_test.go b/core/internal/resources/sqs/application/worker_test.go new file mode 100644 index 0000000..9394eaf --- /dev/null +++ b/core/internal/resources/sqs/application/worker_test.go @@ -0,0 +1,128 @@ +package application + +import ( + "context" + "testing" + "time" + + "github.com/michasdev/mildstack/core/internal/resources/sqs/domain" +) + +func TestWorkerPollDetectsDelayedAndLeasedState(t *testing.T) { + t.Helper() + + now := time.Date(2026, time.April, 19, 12, 0, 0, 0, time.UTC) + service := newServiceWithClock(domain.State{ + Service: "sqs", + Queues: []domain.Queue{ + { + Name: "queue-a", + Attributes: map[string]string{ + "VisibilityTimeout": "30", + }, + }, + }, + Messages: []domain.Message{ + { + Queue: "queue-a", + MessageID: "delayed", + AvailableAt: now.Add(2 * time.Second), + }, + { + Queue: "queue-a", + MessageID: "leased", + ReceivedAt: now.Add(-25 * time.Second), + Metadata: map[string]string{ + leaseTimeoutMetadataKey: "30", + }, + }, + }, + }, nil, newManualClock(now)) + + w := newWorker(service, service.clock) + if got, want := w.poll(now), workerPollInterval; got != want { + t.Fatalf("unexpected poll wait: got %v want %v", got, want) + } + + leases := w.lease(now) + if got, want := len(leases), 1; got != want { + t.Fatalf("unexpected lease count: got %d want %d", got, want) + } + if got, want := leases[0].Message, "leased"; got != want { + t.Fatalf("unexpected leased message: got %q want %q", got, want) + } + + redeliverable := w.redeliver(now.Add(5 * time.Second)) + if got, want := len(redeliverable), 1; got != want { + t.Fatalf("unexpected redeliverable count: got %d want %d", got, want) + } + if got, want := redeliverable[0].MessageID, "leased"; got != want { + t.Fatalf("unexpected redeliverable message: got %q want %q", got, want) + } +} + +func TestWorkerAdvanceMovesExpiredMessagesToDeadLetterQueue(t *testing.T) { + t.Helper() + + now := time.Date(2026, time.April, 19, 12, 0, 0, 0, time.UTC) + service := newServiceWithClock(domain.State{ + Service: "sqs", + Queues: []domain.Queue{ + { + Name: "queue-a", + OrderingHint: "fifo", + Recovery: domain.QueueRecovery{ + DeadLetterQueue: "queue-dlq", + Policy: map[string]string{ + "max_receive_count": "2", + }, + }, + }, + { + Name: "queue-dlq", + }, + }, + Messages: []domain.Message{ + { + Queue: "queue-a", + MessageID: "message-1", + ReceivedAt: now.Add(-31 * time.Second), + Recovery: domain.MessageRecovery{ + Attempts: 2, + }, + }, + }, + }, nil, newManualClock(now)) + + w := newWorker(service, service.clock) + if got := w.advance(now); got != 1 { + t.Fatalf("unexpected dead-letter mutation count: got %d want %d", got, 1) + } + if got, want := service.state.Messages[0].Queue, "queue-dlq"; got != want { + t.Fatalf("unexpected dead-letter queue assignment: got %q want %q", got, want) + } + if got, want := service.state.Messages[0].DeadLetterQueue, "queue-dlq"; got != want { + t.Fatalf("unexpected dead-letter marker: got %q want %q", got, want) + } + if got, want := service.state.Messages[0].DeadLetterSourceQueue, "queue-a"; got != want { + t.Fatalf("unexpected dead-letter source: got %q want %q", got, want) + } + if got, want := service.state.RecoveryMetadata["queue-a/message-1"].Queue, "queue-dlq"; got != want { + t.Fatalf("unexpected recovery metadata queue: got %q want %q", got, want) + } +} + +func TestWorkerStopTerminatesCleanly(t *testing.T) { + t.Helper() + + service := newService(domain.NewState(), nil) + if err := service.Start(context.Background()); err != nil { + t.Fatalf("start service: %v", err) + } + + stopCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := service.Stop(stopCtx); err != nil { + t.Fatalf("stop service: %v", err) + } +} diff --git a/core/internal/resources/sqs/contracts/catalog.go b/core/internal/resources/sqs/contracts/catalog.go new file mode 100644 index 0000000..3c08b9d --- /dev/null +++ b/core/internal/resources/sqs/contracts/catalog.go @@ -0,0 +1,55 @@ +package contracts + +type Scope string + +const ( + ScopeRoot Scope = "root" + ScopeQueue Scope = "queue" +) + +type ActionSpec struct { + Action string + Scope Scope + Version string + ReturnsQueueURL bool + UsesQueueContext bool +} + +var catalog = []ActionSpec{ + {Action: "AddPermission", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "CancelMessageMoveTask", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "ChangeMessageVisibility", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "ChangeMessageVisibilityBatch", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "CreateQueue", Scope: ScopeRoot, Version: "2012-11-05", ReturnsQueueURL: true}, + {Action: "DeleteMessage", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "DeleteMessageBatch", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "DeleteQueue", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "GetQueueAttributes", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "GetQueueUrl", Scope: ScopeRoot, Version: "2012-11-05", ReturnsQueueURL: true}, + {Action: "ListDeadLetterSourceQueues", Scope: ScopeQueue, Version: "2012-11-05", ReturnsQueueURL: true, UsesQueueContext: true}, + {Action: "ListMessageMoveTasks", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "ListQueues", Scope: ScopeRoot, Version: "2012-11-05", ReturnsQueueURL: true}, + {Action: "ListQueueTags", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "PurgeQueue", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "ReceiveMessage", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "RemovePermission", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "SendMessage", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "SendMessageBatch", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "SetQueueAttributes", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "StartMessageMoveTask", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "TagQueue", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, + {Action: "UntagQueue", Scope: ScopeQueue, Version: "2012-11-05", UsesQueueContext: true}, +} + +func Catalog() []ActionSpec { + return append([]ActionSpec(nil), catalog...) +} + +func ActionNames() []string { + specs := Catalog() + names := make([]string, 0, len(specs)) + for _, spec := range specs { + names = append(names, spec.Action) + } + return names +} diff --git a/core/internal/resources/sqs/contracts/catalog_test.go b/core/internal/resources/sqs/contracts/catalog_test.go new file mode 100644 index 0000000..b6b73c0 --- /dev/null +++ b/core/internal/resources/sqs/contracts/catalog_test.go @@ -0,0 +1,106 @@ +package contracts + +import "testing" + +func TestCatalogIncludesRequestedActions(t *testing.T) { + t.Helper() + + specs := Catalog() + if got, want := len(specs), 23; got != want { + t.Fatalf("unexpected action count: got %d want %d", got, want) + } + + expected := []string{ + "AddPermission", + "CancelMessageMoveTask", + "ChangeMessageVisibility", + "ChangeMessageVisibilityBatch", + "CreateQueue", + "DeleteMessage", + "DeleteMessageBatch", + "DeleteQueue", + "GetQueueAttributes", + "GetQueueUrl", + "ListDeadLetterSourceQueues", + "ListMessageMoveTasks", + "ListQueues", + "ListQueueTags", + "PurgeQueue", + "ReceiveMessage", + "RemovePermission", + "SendMessage", + "SendMessageBatch", + "SetQueueAttributes", + "StartMessageMoveTask", + "TagQueue", + "UntagQueue", + } + for i, spec := range specs { + if spec.Action != expected[i] { + t.Fatalf("unexpected action at %d: got %q want %q", i, spec.Action, expected[i]) + } + if spec.Version != "2012-11-05" { + t.Fatalf("unexpected version for %s: got %q want %q", spec.Action, spec.Version, "2012-11-05") + } + } +} + +func TestCatalogHasTransportScopeMetadata(t *testing.T) { + t.Helper() + + specs := Catalog() + byAction := make(map[string]ActionSpec, len(specs)) + for _, spec := range specs { + byAction[spec.Action] = spec + } + + rootActions := []string{"CreateQueue", "GetQueueUrl", "ListQueues"} + for _, action := range rootActions { + spec, ok := byAction[action] + if !ok { + t.Fatalf("expected action %q in catalog", action) + } + if spec.Scope != ScopeRoot { + t.Fatalf("unexpected scope for %s: got %q want %q", action, spec.Scope, ScopeRoot) + } + } + + queueActions := []string{"SendMessage", "DeleteMessage", "SetQueueAttributes", "TagQueue"} + for _, action := range queueActions { + spec, ok := byAction[action] + if !ok { + t.Fatalf("expected action %q in catalog", action) + } + if spec.Scope != ScopeQueue { + t.Fatalf("unexpected scope for %s: got %q want %q", action, spec.Scope, ScopeQueue) + } + if !spec.UsesQueueContext { + t.Fatalf("expected queue context for %s", action) + } + } + + if !byAction["CreateQueue"].ReturnsQueueURL { + t.Fatal("expected CreateQueue to surface QueueUrl") + } + if !byAction["GetQueueUrl"].ReturnsQueueURL { + t.Fatal("expected GetQueueUrl to surface QueueUrl") + } + if !byAction["ListQueues"].ReturnsQueueURL { + t.Fatal("expected ListQueues to surface QueueUrl") + } + if !byAction["ListDeadLetterSourceQueues"].ReturnsQueueURL { + t.Fatal("expected ListDeadLetterSourceQueues to surface QueueUrl") + } +} + +func TestCatalogReturnsCopies(t *testing.T) { + t.Helper() + + first := Catalog() + first[0].Action = "Changed" + + second := Catalog() + if second[0].Action == "Changed" { + t.Fatal("expected catalog to return a copy") + } +} diff --git a/core/internal/resources/sqs/domain/state.go b/core/internal/resources/sqs/domain/state.go new file mode 100644 index 0000000..0e525a3 --- /dev/null +++ b/core/internal/resources/sqs/domain/state.go @@ -0,0 +1,240 @@ +package domain + +import ( + "sort" + "strings" + "time" +) + +const StateKey = "services/sqs" + +type State struct { + Service string + Queues []Queue + Messages []Message + RecoveryMetadata map[string]RecoveryMetadata +} + +type Queue struct { + Name string + URL string + Attributes map[string]string + OrderingHint string + Recovery QueueRecovery + CreatedAt time.Time + UpdatedAt time.Time +} + +type QueueRecovery struct { + DeadLetterQueue string + Policy map[string]string +} + +type Message struct { + Queue string + MessageID string + Body string + Attributes map[string]string + Metadata map[string]string + Tags []string + ReceiptKeys []string + MessageGroupID string + SequenceNumber int64 + BatchID string + BatchEntryID string + BatchEntryIndex int + BatchEntryCount int + DeadLetterQueue string + DeadLetterSourceQueue string + DeadLetteredAt time.Time + SentAt time.Time + AvailableAt time.Time + ReceivedAt time.Time + Recovery MessageRecovery +} + +type MessageRecovery struct { + Attempts int + Detail map[string]string +} + +type RecoveryMetadata struct { + Queue string + Message string + Detail map[string]string +} + +func NewState() State { + return State{ + Service: "sqs", + Queues: []Queue{}, + Messages: []Message{}, + RecoveryMetadata: map[string]RecoveryMetadata{}, + } +} + +func (s State) Snapshot() map[string]any { + queues := make([]any, 0, len(s.Queues)) + for _, queue := range s.ListQueues() { + queues = append(queues, map[string]any{ + "name": queue.Name, + "url": queue.URL, + "attributes": cloneStringMapAny(queue.Attributes), + "ordering_hint": queue.OrderingHint, + "recovery": map[string]any{ + "dead_letter_queue": queue.Recovery.DeadLetterQueue, + "policy": cloneStringMapAny(queue.Recovery.Policy), + }, + "created_at": snapshotTime(queue.CreatedAt), + "updated_at": snapshotTime(queue.UpdatedAt), + }) + } + + messages := make([]any, 0, len(s.Messages)) + for _, message := range s.ListMessages() { + messages = append(messages, map[string]any{ + "queue": message.Queue, + "message_id": message.MessageID, + "body": message.Body, + "attributes": cloneStringMapAny(message.Attributes), + "metadata": cloneStringMapAny(message.Metadata), + "tags": append([]string(nil), message.Tags...), + "receipt_keys": append([]string(nil), message.ReceiptKeys...), + "message_group_id": message.MessageGroupID, + "sequence_number": message.SequenceNumber, + "batch_id": message.BatchID, + "batch_entry_id": message.BatchEntryID, + "batch_entry_index": message.BatchEntryIndex, + "batch_entry_count": message.BatchEntryCount, + "dead_letter_queue": message.DeadLetterQueue, + "dead_letter_source_queue": message.DeadLetterSourceQueue, + "dead_lettered_at": snapshotTime(message.DeadLetteredAt), + "sent_at": snapshotTime(message.SentAt), + "available_at": snapshotTime(message.AvailableAt), + "received_at": snapshotTime(message.ReceivedAt), + "recovery": map[string]any{ + "attempts": message.Recovery.Attempts, + "detail": cloneStringMapAny(message.Recovery.Detail), + }, + }) + } + + recovery := make(map[string]any, len(s.RecoveryMetadata)) + keys := make([]string, 0, len(s.RecoveryMetadata)) + for key := range s.RecoveryMetadata { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + metadata := s.RecoveryMetadata[key] + recovery[key] = map[string]any{ + "queue": metadata.Queue, + "message": metadata.Message, + "detail": cloneStringMapAny(metadata.Detail), + } + } + + return map[string]any{ + "service": s.Service, + "queues": queues, + "messages": messages, + "recovery_metadata": recovery, + } +} + +func (s State) Clone() State { + cloned := State{ + Service: s.Service, + Queues: make([]Queue, len(s.Queues)), + Messages: make([]Message, len(s.Messages)), + RecoveryMetadata: make(map[string]RecoveryMetadata, len(s.RecoveryMetadata)), + } + copy(cloned.Queues, s.Queues) + for i := range cloned.Queues { + cloned.Queues[i].Attributes = cloneStringMap(cloned.Queues[i].Attributes) + cloned.Queues[i].Recovery.Policy = cloneStringMap(cloned.Queues[i].Recovery.Policy) + } + copy(cloned.Messages, s.Messages) + for i := range cloned.Messages { + cloned.Messages[i].Attributes = cloneStringMap(cloned.Messages[i].Attributes) + cloned.Messages[i].Metadata = cloneStringMap(cloned.Messages[i].Metadata) + cloned.Messages[i].Tags = append([]string(nil), cloned.Messages[i].Tags...) + cloned.Messages[i].ReceiptKeys = append([]string(nil), cloned.Messages[i].ReceiptKeys...) + cloned.Messages[i].Recovery.Detail = cloneStringMap(cloned.Messages[i].Recovery.Detail) + } + for key, value := range s.RecoveryMetadata { + cloned.RecoveryMetadata[key] = RecoveryMetadata{ + Queue: value.Queue, + Message: value.Message, + Detail: cloneStringMap(value.Detail), + } + } + return cloned +} + +func (s State) ListQueues() []Queue { + queues := make([]Queue, len(s.Queues)) + copy(queues, s.Queues) + for i := range queues { + queues[i].Attributes = cloneStringMap(queues[i].Attributes) + queues[i].Recovery.Policy = cloneStringMap(queues[i].Recovery.Policy) + } + sort.SliceStable(queues, func(i, j int) bool { + return queues[i].Name < queues[j].Name + }) + return queues +} + +func (s State) ListMessages() []Message { + messages := make([]Message, len(s.Messages)) + copy(messages, s.Messages) + for i := range messages { + messages[i].Attributes = cloneStringMap(messages[i].Attributes) + messages[i].Metadata = cloneStringMap(messages[i].Metadata) + messages[i].Tags = append([]string(nil), messages[i].Tags...) + messages[i].ReceiptKeys = append([]string(nil), messages[i].ReceiptKeys...) + messages[i].Recovery.Detail = cloneStringMap(messages[i].Recovery.Detail) + } + sort.SliceStable(messages, func(i, j int) bool { + if messages[i].Queue == messages[j].Queue { + return messages[i].MessageID < messages[j].MessageID + } + return messages[i].Queue < messages[j].Queue + }) + return messages +} + +func snapshotTime(value time.Time) string { + if value.IsZero() { + return "" + } + return value.UTC().Format(time.RFC3339Nano) +} + +func cloneStringMap(values map[string]string) map[string]string { + if values == nil { + return nil + } + + cloned := make(map[string]string, len(values)) + for key, value := range values { + cloned[key] = value + } + return cloned +} + +func cloneStringMapAny(values map[string]string) map[string]any { + if values == nil { + return nil + } + + cloned := make(map[string]any, len(values)) + for key, value := range values { + cloned[key] = value + } + return cloned +} + +func normalizeName(value string) string { + return strings.TrimSpace(value) +} diff --git a/core/internal/resources/sqs/domain/state_test.go b/core/internal/resources/sqs/domain/state_test.go new file mode 100644 index 0000000..9cb4949 --- /dev/null +++ b/core/internal/resources/sqs/domain/state_test.go @@ -0,0 +1,224 @@ +package domain + +import ( + "testing" + "time" +) + +func TestStateSnapshotCopiesLiveData(t *testing.T) { + t.Helper() + + state := NewState() + state.Queues = append(state.Queues, Queue{ + Name: "queue-a", + URL: "https://example.invalid/queue-a", + Attributes: map[string]string{ + "VisibilityTimeout": "30", + }, + OrderingHint: "fifo", + Recovery: QueueRecovery{ + DeadLetterQueue: "queue-dlq", + Policy: map[string]string{ + "max_receive_count": "5", + }, + }, + CreatedAt: time.Date(2026, time.April, 19, 10, 0, 0, 0, time.UTC), + UpdatedAt: time.Date(2026, time.April, 19, 10, 1, 0, 0, time.UTC), + }) + state.Messages = append(state.Messages, Message{ + Queue: "queue-a", + MessageID: "message-1", + Body: "payload", + Attributes: map[string]string{"foo": "bar"}, + Metadata: map[string]string{"trace": "abc"}, + Tags: []string{"alpha", "beta"}, + ReceiptKeys: []string{"r-1", "r-2"}, + MessageGroupID: "group-a", + SequenceNumber: 17, + BatchID: "batch-a", + BatchEntryID: "entry-a", + BatchEntryIndex: 1, + BatchEntryCount: 3, + DeadLetterQueue: "queue-dlq", + DeadLetterSourceQueue: "queue-a", + DeadLetteredAt: time.Date(2026, time.April, 19, 10, 2, 30, 0, time.UTC), + SentAt: time.Date(2026, time.April, 19, 10, 2, 0, 0, time.UTC), + Recovery: MessageRecovery{ + Attempts: 2, + Detail: map[string]string{"reason": "retry"}, + }, + }) + state.RecoveryMetadata["queue-a/message-1"] = RecoveryMetadata{ + Queue: "queue-a", + Message: "message-1", + Detail: map[string]string{"state": "pending"}, + } + + snapshot := state.Snapshot() + + queues := snapshot["queues"].([]any) + queues[0].(map[string]any)["name"] = "mutated" + queues[0].(map[string]any)["attributes"].(map[string]any)["VisibilityTimeout"] = "99" + queues[0].(map[string]any)["recovery"].(map[string]any)["dead_letter_queue"] = "mutated-dlq" + queues[0].(map[string]any)["ordering_hint"] = "mutated" + + messages := snapshot["messages"].([]any) + messages[0].(map[string]any)["body"] = "mutated" + messages[0].(map[string]any)["tags"].([]string)[0] = "mutated" + messages[0].(map[string]any)["metadata"].(map[string]any)["trace"] = "mutated" + messages[0].(map[string]any)["message_group_id"] = "mutated" + messages[0].(map[string]any)["dead_letter_queue"] = "mutated" + messages[0].(map[string]any)["receipt_keys"].([]string)[0] = "mutated" + + recovery := snapshot["recovery_metadata"].(map[string]any) + recovery["queue-a/message-1"].(map[string]any)["queue"] = "mutated" + + if got, want := state.Queues[0].Name, "queue-a"; got != want { + t.Fatalf("queue name was aliased: got %q want %q", got, want) + } + if got, want := state.Queues[0].Attributes["VisibilityTimeout"], "30"; got != want { + t.Fatalf("queue attributes were aliased: got %q want %q", got, want) + } + if got, want := state.Queues[0].Recovery.DeadLetterQueue, "queue-dlq"; got != want { + t.Fatalf("queue recovery metadata was aliased: got %q want %q", got, want) + } + if got, want := state.Queues[0].OrderingHint, "fifo"; got != want { + t.Fatalf("queue ordering hint was aliased: got %q want %q", got, want) + } + if got, want := state.Messages[0].Body, "payload"; got != want { + t.Fatalf("message body was aliased: got %q want %q", got, want) + } + if got, want := state.Messages[0].Tags[0], "alpha"; got != want { + t.Fatalf("message tags were aliased: got %q want %q", got, want) + } + if got, want := state.Messages[0].Metadata["trace"], "abc"; got != want { + t.Fatalf("message metadata was aliased: got %q want %q", got, want) + } + if got, want := state.Messages[0].MessageGroupID, "group-a"; got != want { + t.Fatalf("message group id was aliased: got %q want %q", got, want) + } + if got, want := state.Messages[0].DeadLetterQueue, "queue-dlq"; got != want { + t.Fatalf("dead letter queue was aliased: got %q want %q", got, want) + } + if got, want := state.RecoveryMetadata["queue-a/message-1"].Queue, "queue-a"; got != want { + t.Fatalf("recovery metadata was aliased: got %q want %q", got, want) + } +} + +func TestStateCloneReturnsDeepCopy(t *testing.T) { + t.Helper() + + state := NewState() + state.Queues = append(state.Queues, Queue{ + Name: "queue-a", + Attributes: map[string]string{ + "DelaySeconds": "0", + }, + OrderingHint: "standard", + Recovery: QueueRecovery{ + Policy: map[string]string{"enabled": "true"}, + }, + }) + state.Messages = append(state.Messages, Message{ + Queue: "queue-a", + MessageID: "message-1", + Attributes: map[string]string{"foo": "bar"}, + Metadata: map[string]string{"baz": "qux"}, + Tags: []string{"alpha"}, + ReceiptKeys: []string{"r-1"}, + MessageGroupID: "group-a", + SequenceNumber: 2, + BatchID: "batch-a", + BatchEntryID: "entry-a", + BatchEntryIndex: 0, + BatchEntryCount: 1, + DeadLetterQueue: "queue-dlq", + DeadLetterSourceQueue: "queue-a", + Recovery: MessageRecovery{ + Attempts: 1, + Detail: map[string]string{"reason": "initial"}, + }, + }) + state.RecoveryMetadata["queue-a/message-1"] = RecoveryMetadata{ + Queue: "queue-a", + Message: "message-1", + Detail: map[string]string{"state": "live"}, + } + + cloned := state.Clone() + cloned.Queues[0].Attributes["DelaySeconds"] = "5" + cloned.Queues[0].Recovery.Policy["enabled"] = "false" + cloned.Messages[0].Tags[0] = "mutated" + cloned.Messages[0].ReceiptKeys[0] = "mutated" + cloned.Messages[0].MessageGroupID = "mutated" + cloned.Messages[0].DeadLetterQueue = "mutated" + cloned.Messages[0].Recovery.Detail["reason"] = "mutated" + cloned.RecoveryMetadata["queue-a/message-1"] = RecoveryMetadata{ + Queue: "changed", + Message: "changed", + Detail: map[string]string{"state": "changed"}, + } + + if got, want := state.Queues[0].Attributes["DelaySeconds"], "0"; got != want { + t.Fatalf("queue attributes were shared with clone: got %q want %q", got, want) + } + if got, want := state.Queues[0].Recovery.Policy["enabled"], "true"; got != want { + t.Fatalf("queue policy was shared with clone: got %q want %q", got, want) + } + if got, want := state.Messages[0].Tags[0], "alpha"; got != want { + t.Fatalf("message tags were shared with clone: got %q want %q", got, want) + } + if got, want := state.Messages[0].ReceiptKeys[0], "r-1"; got != want { + t.Fatalf("message receipt keys were shared with clone: got %q want %q", got, want) + } + if got, want := state.Messages[0].MessageGroupID, "group-a"; got != want { + t.Fatalf("message group id was shared with clone: got %q want %q", got, want) + } + if got, want := state.Messages[0].DeadLetterQueue, "queue-dlq"; got != want { + t.Fatalf("dead letter queue was shared with clone: got %q want %q", got, want) + } + if got, want := state.Messages[0].Recovery.Detail["reason"], "initial"; got != want { + t.Fatalf("message recovery detail was shared with clone: got %q want %q", got, want) + } + if got, want := state.RecoveryMetadata["queue-a/message-1"].Queue, "queue-a"; got != want { + t.Fatalf("recovery metadata was shared with clone: got %q want %q", got, want) + } +} + +func TestStateKeepsRoomForRecoveryMetadataAndAttributes(t *testing.T) { + t.Helper() + + state := NewState() + if got, want := state.Service, "sqs"; got != want { + t.Fatalf("unexpected service name: got %q want %q", got, want) + } + if got, want := StateKey, "services/sqs"; got != want { + t.Fatalf("unexpected state key: got %q want %q", got, want) + } + + queue := Queue{ + Name: "queue-a", + Attributes: map[string]string{ + "RedrivePolicy": "present", + }, + OrderingHint: "fifo", + Recovery: QueueRecovery{ + DeadLetterQueue: "queue-dlq", + }, + } + state.Queues = append(state.Queues, queue) + state.RecoveryMetadata["queue-a"] = RecoveryMetadata{ + Queue: queue.Name, + Detail: map[string]string{"state": "ready"}, + } + + if got, want := state.Queues[0].Recovery.DeadLetterQueue, "queue-dlq"; got != want { + t.Fatalf("unexpected queue recovery value: got %q want %q", got, want) + } + if got, want := state.Queues[0].OrderingHint, "fifo"; got != want { + t.Fatalf("unexpected queue ordering hint: got %q want %q", got, want) + } + if got, want := state.RecoveryMetadata["queue-a"].Detail["state"], "ready"; got != want { + t.Fatalf("unexpected recovery detail: got %q want %q", got, want) + } +} diff --git a/core/internal/resources/sqs/infrastructure/routes.go b/core/internal/resources/sqs/infrastructure/routes.go new file mode 100644 index 0000000..e2b1751 --- /dev/null +++ b/core/internal/resources/sqs/infrastructure/routes.go @@ -0,0 +1,43 @@ +package infrastructure + +import "github.com/michasdev/mildstack/core/internal/application/orchestrator" + +func Routes() []orchestrator.Route { + return []orchestrator.Route{ + { + Method: "GET", + Path: "/sqs/queues", + Name: "sqs.queues.index", + }, + { + Method: "POST", + Path: "/sqs/queues", + Name: "sqs.queues.create", + }, + { + Method: "GET", + Path: "/sqs/queues/:queue", + Name: "sqs.queues.show", + }, + { + Method: "DELETE", + Path: "/sqs/queues/:queue", + Name: "sqs.queues.delete", + }, + { + Method: "GET", + Path: "/sqs/queues/:queue/messages", + Name: "sqs.messages.receive", + }, + { + Method: "POST", + Path: "/sqs/queues/:queue/messages", + Name: "sqs.messages.send", + }, + { + Method: "DELETE", + Path: "/sqs/queues/:queue/messages/:receiptHandle", + Name: "sqs.messages.delete", + }, + } +} diff --git a/core/internal/resources/sqs/infrastructure/routes_test.go b/core/internal/resources/sqs/infrastructure/routes_test.go new file mode 100644 index 0000000..b9d98c9 --- /dev/null +++ b/core/internal/resources/sqs/infrastructure/routes_test.go @@ -0,0 +1,38 @@ +package infrastructure + +import "testing" + +func TestRoutesUseSQSServiceSegment(t *testing.T) { + t.Helper() + + routes := Routes() + if got, want := len(routes), 7; got != want { + t.Fatalf("unexpected route count: got %d want %d", got, want) + } + + expected := []struct { + method string + path string + name string + }{ + {method: "GET", path: "/sqs/queues", name: "sqs.queues.index"}, + {method: "POST", path: "/sqs/queues", name: "sqs.queues.create"}, + {method: "GET", path: "/sqs/queues/:queue", name: "sqs.queues.show"}, + {method: "DELETE", path: "/sqs/queues/:queue", name: "sqs.queues.delete"}, + {method: "GET", path: "/sqs/queues/:queue/messages", name: "sqs.messages.receive"}, + {method: "POST", path: "/sqs/queues/:queue/messages", name: "sqs.messages.send"}, + {method: "DELETE", path: "/sqs/queues/:queue/messages/:receiptHandle", name: "sqs.messages.delete"}, + } + + for i, route := range routes { + if route.Method != expected[i].method { + t.Fatalf("unexpected method at %d: got %q want %q", i, route.Method, expected[i].method) + } + if route.Path != expected[i].path { + t.Fatalf("unexpected path at %d: got %q want %q", i, route.Path, expected[i].path) + } + if route.Name != expected[i].name { + t.Fatalf("unexpected name at %d: got %q want %q", i, route.Name, expected[i].name) + } + } +} diff --git a/core/internal/resources/sqs/service.go b/core/internal/resources/sqs/service.go new file mode 100644 index 0000000..5fcfb5f --- /dev/null +++ b/core/internal/resources/sqs/service.go @@ -0,0 +1,16 @@ +package sqs + +import ( + "github.com/michasdev/mildstack/core/internal/application/orchestrator" + "github.com/michasdev/mildstack/core/internal/resources/sqs/application" +) + +type StorageConfig = application.StorageConfig + +func New() orchestrator.Service { + return application.New() +} + +func NewWithStorage(config StorageConfig) (orchestrator.Service, error) { + return application.NewWithPersistence(config) +} diff --git a/go.mod b/go.mod index a0a3ddb..03a28fd 100644 --- a/go.mod +++ b/go.mod @@ -23,6 +23,7 @@ require ( github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.11.22 // indirect github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.22 // indirect github.com/aws/aws-sdk-go-v2/service/signin v1.0.10 // indirect + github.com/aws/aws-sdk-go-v2/service/sqs v1.42.26 // indirect github.com/aws/aws-sdk-go-v2/service/sso v1.30.16 // indirect github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.20 // indirect github.com/aws/aws-sdk-go-v2/service/sts v1.42.0 // indirect diff --git a/go.sum b/go.sum index b7b6e01..8e2c45c 100644 --- a/go.sum +++ b/go.sum @@ -24,6 +24,8 @@ github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.22 h1:PUmZeJU6 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.22/go.mod h1:nO6egFBoAaoXze24a2C0NjQCvdpk8OueRoYimvEB9jo= github.com/aws/aws-sdk-go-v2/service/signin v1.0.10 h1:a1Fq/KXn75wSzoJaPQTgZO0wHGqE9mjFnylnqEPTchA= github.com/aws/aws-sdk-go-v2/service/signin v1.0.10/go.mod h1:p6+MXNxW7IA6dMgHfTAzljuwSKD0NCm/4lbS4t6+7vI= +github.com/aws/aws-sdk-go-v2/service/sqs v1.42.26 h1:jtUEQz/c14fCMkOX3r2/nhYmhXZas0XdcQhUaIW5ubY= +github.com/aws/aws-sdk-go-v2/service/sqs v1.42.26/go.mod h1:gcJv70rH+Z/Q1PM3jKsJr6+vfKrDHJOfmKq7342+Vq8= github.com/aws/aws-sdk-go-v2/service/sso v1.30.16 h1:x6bKbmDhsgSZwv6q19wY/u3rLk/3FGjJWyqKcIRufpE= github.com/aws/aws-sdk-go-v2/service/sso v1.30.16/go.mod h1:CudnEVKRtLn0+3uMV0yEXZ+YZOKnAtUJ5DmDhilVnIw= github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.20 h1:oK/njaL8GtyEihkWMD4k3VgHCT64RQKkZwh0DG5j8ak=