sqs

Documentation

Overview

Package sqs: migration registration for the sqs_messages dedicated table (docs/plans/storage-plan.md item 3.10).

Build-tagged out of `-tags nosqlite` builds because it depends on state.Migration / state.RegisterMigration, which are themselves only defined for !nosqlite builds (compare internal/state/migrate.go against internal/state/sqlite_hybrid_nosqlite.go). This is safe: under -tags nosqlite, NewHybridStore and NewSQLiteStore both unconditionally return an error and cmd/overcast/cmd_serve.go's buildStore falls back to state.NewMemoryStore, so no real store ever satisfies state.SQLiteDBProvider at runtime in a nosqlite build — newMessageBackendFor (message_backend.go, not build-tagged) always selects memMessageBackend there regardless of whether this migration ever registered. CloudWatch Logs' event backend and DynamoDB's item/stream backends rely on the identical reasoning (see internal/services/cloudwatch/logs/migrations.go and internal/services/dynamodb/migrations.go).

Package sqs implements the AWS SQS API emulator.

SQS uses a JSON (or form-encoded) API. Operations are identified by the X-Amz-Target header: "AmazonSQS.SendMessage", "AmazonSQS.ReceiveMessage", etc. All operations share a single endpoint — routing is by target header, not URL.

Supported (P1): CreateQueue, GetQueueUrl, SendMessage, ReceiveMessage,

DeleteMessage, GetQueueAttributes

Supported (P2): SendMessageBatch, DeleteMessageBatch, SetQueueAttributes,

PurgeQueue, ListQueues

Unsupported: See docs/services/sqs.md

Index

• type Handler
  • func (h *Handler) AddPermission(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) ChangeMessageVisibility(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) ChangeMessageVisibilityBatch(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) CreateQueue(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) DeleteMessage(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) DeleteMessageBatch(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) DeleteQueue(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) GetQueueAttributes(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) GetQueueMetrics(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) GetQueueURL(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) ListDeadLetterSourceQueues(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) ListQueueTags(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) ListQueues(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) PeekMessages(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) PurgeQueue(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) ReceiveMessage(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) RemovePermission(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) SendMessage(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) SendMessageBatch(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) SetQueueAttributes(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) StartMessageMoveTask(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) TagQueue(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) UntagQueue(w http.ResponseWriter, r *http.Request)
• type Message
  • func (m *Message) IsVisible(clk clock.Clock) bool
• type MessageAttribute
• type Queue
  • func (q *Queue) GetTags() map[string]string
  • func (q *Queue) SetTags(tags map[string]string)
• type Service
  • func New(cfg *config.Config, store state.Store, logger *zap.Logger, clk clock.Clock) *Service
  • func (s *Service) DebugNamespace() string
  • func (s *Service) DebugResetState(ctx context.Context) error
  • func (s *Service) DebugStateKeys(ctx context.Context) ([]string, error)
  • func (s *Service) DebugStateValues(ctx context.Context) (map[string]string, error)
  • func (s *Service) Dispatch(w http.ResponseWriter, r *http.Request)
  • func (s *Service) DispatchQuery(w http.ResponseWriter, r *http.Request)
  • func (s *Service) Enqueuer() events.MessageEnqueuer
  • func (s *Service) InitBus(b *events.Bus)
  • func (s *Service) InitMetrics(m metricsRecorder)
  • func (s *Service) Name() string
  • func (s *Service) Operations() []op.Operation
  • func (s *Service) OwnsAction(action string) bool
  • func (s *Service) Receiver() events.MessageReceiver
  • func (s *Service) RegisterRoutes(r chi.Router)
  • func (s *Service) Stop(_ context.Context)
  • func (s *Service) SupportedProtocols() []codec.Codec
  • func (s *Service) TargetPrefix() string

Types

type Handler

type Handler struct {
	// contains filtered or unexported fields
}

Handler holds SQS handler dependencies.

func (*Handler) AddPermission

func (h *Handler) AddPermission(w http.ResponseWriter, r *http.Request)

AddPermission handles the SQS AddPermission operation. AWS docs: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_AddPermission.html

func (*Handler) ChangeMessageVisibility

func (h *Handler) ChangeMessageVisibility(w http.ResponseWriter, r *http.Request)

ChangeMessageVisibility adjusts the visibility timeout of an in-flight message. AWS docs: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ChangeMessageVisibility.html

func (*Handler) ChangeMessageVisibilityBatch

func (h *Handler) ChangeMessageVisibilityBatch(w http.ResponseWriter, r *http.Request)

ChangeMessageVisibilityBatch handles the SQS ChangeMessageVisibilityBatch operation. AWS docs: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ChangeMessageVisibilityBatch.html

func (*Handler) CreateQueue

func (h *Handler) CreateQueue(w http.ResponseWriter, r *http.Request)

func (*Handler) DeleteMessage

func (h *Handler) DeleteMessage(w http.ResponseWriter, r *http.Request)

func (*Handler) DeleteMessageBatch

func (h *Handler) DeleteMessageBatch(w http.ResponseWriter, r *http.Request)

func (*Handler) DeleteQueue

func (h *Handler) DeleteQueue(w http.ResponseWriter, r *http.Request)

func (*Handler) GetQueueAttributes

func (h *Handler) GetQueueAttributes(w http.ResponseWriter, r *http.Request)

func (*Handler) GetQueueMetrics

func (h *Handler) GetQueueMetrics(w http.ResponseWriter, r *http.Request)

GetQueueMetrics handles GET /_overcast/sqs/queues/{name}/metrics. ?range= is one of "1h" (default), "6h", "24h", "7d", "30d" — see metrics.ParseChartRange.

func (*Handler) GetQueueURL

func (h *Handler) GetQueueURL(w http.ResponseWriter, r *http.Request)

func (*Handler) ListDeadLetterSourceQueues

func (h *Handler) ListDeadLetterSourceQueues(w http.ResponseWriter, r *http.Request)

ListDeadLetterSourceQueues handles the SQS ListDeadLetterSourceQueues operation. It returns all queues whose RedrivePolicy targets the specified DLQ. AWS docs: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ListDeadLetterSourceQueues.html

func (*Handler) ListQueueTags

func (h *Handler) ListQueueTags(w http.ResponseWriter, r *http.Request)

ListQueueTags handles the SQS ListQueueTags operation. AWS docs: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ListQueueTags.html

func (*Handler) ListQueues

func (h *Handler) ListQueues(w http.ResponseWriter, r *http.Request)

func (*Handler) PeekMessages

func (h *Handler) PeekMessages(w http.ResponseWriter, r *http.Request)

PeekMessages is a non-AWS extension that returns all messages in a queue without modifying any state — no receive-count increment, no visibility timeout applied. In-flight (invisible) messages are included and flagged.

Route: GET /{accountID}/{queueName}.

func (*Handler) PurgeQueue

func (h *Handler) PurgeQueue(w http.ResponseWriter, r *http.Request)

func (*Handler) ReceiveMessage

func (h *Handler) ReceiveMessage(w http.ResponseWriter, r *http.Request)

func (*Handler) RemovePermission

func (h *Handler) RemovePermission(w http.ResponseWriter, r *http.Request)

RemovePermission handles the SQS RemovePermission operation. AWS docs: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_RemovePermission.html

func (*Handler) SendMessage

func (h *Handler) SendMessage(w http.ResponseWriter, r *http.Request)

func (*Handler) SendMessageBatch

func (h *Handler) SendMessageBatch(w http.ResponseWriter, r *http.Request)

func (*Handler) SetQueueAttributes

func (h *Handler) SetQueueAttributes(w http.ResponseWriter, r *http.Request)

func (*Handler) StartMessageMoveTask

func (h *Handler) StartMessageMoveTask(w http.ResponseWriter, r *http.Request)

StartMessageMoveTask redrives messages from a DLQ back to their source queue. AWS docs: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_StartMessageMoveTask.html

In real AWS this is an async task. In Overcast the move is performed synchronously since there is no need for background task management in a local emulator — the response is returned once all messages are moved.

func (*Handler) TagQueue

func (h *Handler) TagQueue(w http.ResponseWriter, r *http.Request)

TagQueue handles the SQS TagQueue operation. AWS docs: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_TagQueue.html

func (*Handler) UntagQueue

func (h *Handler) UntagQueue(w http.ResponseWriter, r *http.Request)

UntagQueue handles the SQS UntagQueue operation. AWS docs: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_UntagQueue.html

type Message

type Message struct {
	MessageID               string                      `json:"message_id"`
	ReceiptHandle           string                      `json:"receipt_handle"`
	Body                    string                      `json:"body"`
	Attributes              map[string]string           `json:"attributes"`
	MessageAttributes       map[string]MessageAttribute `json:"message_attributes,omitempty"`
	MD5OfBody               string                      `json:"md5_of_body"`
	SentTimestamp           int64                       `json:"sent_timestamp"`
	ApproximateReceiveCount int                         `json:"approximate_receive_count"`
	VisibleAfter            time.Time                   `json:"visible_after"`
	// FIFO fields — only set for messages in FIFO queues.
	MessageGroupId         string `json:"message_group_id,omitempty"`
	MessageDeduplicationId string `json:"message_deduplication_id,omitempty"`
	SequenceNumber         string `json:"sequence_number,omitempty"`
	// VisibilityVersion is incremented on each ChangeMessageVisibility call
	// so ReceiveRequestAttemptId replay can detect modified messages.
	VisibilityVersion int `json:"visibility_version,omitempty"`
}

Message represents a stored SQS message.

func (*Message) IsVisible

func (m *Message) IsVisible(clk clock.Clock) bool

IsVisible returns true if the message can currently be received. The clock parameter is provided by the enclosing store so that tests can advance time without real sleeps.

type MessageAttribute

type MessageAttribute struct {
	DataType    string `json:"DataType"`
	StringValue string `json:"StringValue,omitempty"`
	BinaryValue []byte `json:"BinaryValue,omitempty"`
}

MessageAttribute is a typed SQS message attribute. JSON tags use PascalCase to match the AWS SQS wire format.

type Queue

type Queue struct {
	Name             string            `json:"name"`
	URL              string            `json:"url"`
	ARN              string            `json:"arn"`
	Attributes       map[string]string `json:"attributes"`
	CreatedTimestamp int64             `json:"created_timestamp"`
	Tags             map[string]string `json:"tags,omitempty"`
}

Queue represents a stored SQS queue.

func (*Queue) GetTags

func (q *Queue) GetTags() map[string]string

func (*Queue) SetTags

func (q *Queue) SetTags(tags map[string]string)

type Service

type Service struct {
	// contains filtered or unexported fields
}

Service implements router.Service for SQS.

func New

func New(cfg *config.Config, store state.Store, logger *zap.Logger, clk clock.Clock) *Service

New returns a configured SQS Service.

func (*Service) DebugNamespace

func (s *Service) DebugNamespace() string

DebugNamespace returns the virtual raw-state namespace name for SQS messages, implementing router.DebugStateProvider. Messages live in the dedicated sqs_messages SQL table (or the in-memory equivalent), not the generic kv store, so without this they'd be invisible to /_overcast/debug/state and exempt from /_overcast/debug/reset — the graduation rule (storage-plan.md "Settled decisions") requires this for every dedicated table, mirroring DynamoDB's "dynamodb:items" and CloudWatch Logs' "logs:events".

func (*Service) DebugResetState

func (s *Service) DebugResetState(ctx context.Context) error

DebugResetState deletes every persisted message, for /_overcast/debug/reset.

func (*Service) DebugStateKeys

func (s *Service) DebugStateKeys(ctx context.Context) ([]string, error)

DebugStateKeys returns up to debugMessagesScanLimit virtual keys for /_overcast/debug/state's top-level listing.

func (*Service) DebugStateValues

func (s *Service) DebugStateValues(ctx context.Context) (map[string]string, error)

DebugStateValues returns raw message values keyed by region/queueName/messageID, capped at debugMessagesScanLimit rows. A "_truncated" pseudo-key is added when more rows exist than were returned.

func (*Service) Dispatch

func (s *Service) Dispatch(w http.ResponseWriter, r *http.Request)

Dispatch routes to the correct SQS handler based on X-Amz-Target.

When the protocol-detection middleware has stashed a codec and operation name in the request context AND the operation has been migrated to the typed dispatcher, the typed path is taken. Otherwise the legacy http.HandlerFunc registry runs unchanged.

func (*Service) DispatchQuery

func (s *Service) DispatchQuery(w http.ResponseWriter, r *http.Request)

DispatchQuery satisfies router.QueryDispatcher. It converts an SQS Query-protocol request (form-encoded) into the JSON body the existing handlers expect, captures their JSON response, and writes the equivalent Query XML response.

func (*Service) Enqueuer

func (s *Service) Enqueuer() events.MessageEnqueuer

Enqueuer returns an events.MessageEnqueuer backed by this service's store. The router passes this to S3 (and future services) for cross-service notification delivery without creating import cycles.

func (*Service) InitBus

func (s *Service) InitBus(b *events.Bus)

InitBus wires the event bus so that queue lifecycle events appear on the topology map. Call this after the bus has been constructed. It also starts a background goroutine that watches for in-flight messages whose visibility timeout has expired and emits SQSMessageVisible events.

func (*Service) InitMetrics

func (s *Service) InitMetrics(m metricsRecorder)

InitMetrics wires the shared service-metrics recorder (docs/plans/service-metrics-platform.md phase 2) so every message operation records its AWS/SQS outcome metrics (metrics_sqs.go), and starts the periodic gauge sampler that publishes ApproximateNumberOfMessagesVisible/NotVisible/Delayed and ApproximateAgeOfOldestMessage for every existing queue once a minute — the same fact AWS samples whether or not the queue has traffic. Called once from router.New, after metrics.NewRecorder; a Service without it (unit tests, or OVERCAST_SERVICE_METRICS=disabled) simply never records or samples anything, matching Lambda's InitMetrics contract.

func (*Service) Name

func (s *Service) Name() string

func (*Service) Operations

func (s *Service) Operations() []op.Operation

Operations implements router.ProtocolService.

func (*Service) OwnsAction

func (s *Service) OwnsAction(action string) bool

OwnsAction satisfies router.QueryActionOwner.

func (*Service) Receiver

func (s *Service) Receiver() events.MessageReceiver

Receiver returns an events.MessageReceiver backed by this service's store. The Lambda event source mapping poller uses this to receive and delete messages directly without going through the HTTP layer.

func (*Service) RegisterRoutes

func (s *Service) RegisterRoutes(r chi.Router)

RegisterRoutes mounts SQS handlers. POST / is handled by the router's target dispatcher (shared with DynamoDB, SNS). The queue URL route is SQS-specific and registered here.

func (*Service) Stop

func (s *Service) Stop(_ context.Context)

Stop cancels the background watchVisibility and gauge-sampler goroutines. Implements router.Stopper so the router calls it on shutdown.

func (*Service) SupportedProtocols

func (s *Service) SupportedProtocols() []codec.Codec

SupportedProtocols implements router.ProtocolService. SQS speaks JSON 1.0 (modern SDKs) and AWS Query (legacy SDKs / form-encoded). The query path still goes through the legacy form decoder until Phase 6.

func (*Service) TargetPrefix

func (s *Service) TargetPrefix() string

TargetPrefix returns the X-Amz-Target prefix for SQS dispatch.