dynamodb

package
v1.1.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 20, 2026 License: MIT Imports: 50 Imported by: 0

README

DynamoDB

Parity grade: A · SDK aws-sdk-go-v2/service/dynamodb · last audited 2026-07-11 (33a39b1f) · protocol json-1.0 (DynamoDB_20120810 targets)

Coverage

Metric Value
Feature families 7 (7 ok)
Known gaps none
Deferred items 3
Resource leaks clean
Deferred
  • expr/ lexer/parser/evaluator subpackage (has own aws_spec_test.go/evaluator_test.go) — not line-by-line re-audited
  • PartiQL execution
  • TransactWriteItems Put/Update/Delete/ConditionCheck unused-EAN/EAV validation (bd: gopherstack-daa)

More

Documentation

Overview

Package dynamodb implements the AWS DynamoDB mock service. autoscaling.go implements UpdateTableReplicaAutoScaling: it persists the requested auto-scaling configuration so DescribeTableReplicaAutoScaling can round-trip the values without simulating real scaling.

Package dynamodb implements the AWS DynamoDB mock service. capacity.go builds ConsumedCapacity responses (table/index breakdowns and the ConsistentRead RCU multiplier).

Package dynamodb implements the AWS DynamoDB mock service. contributor_insights.go implements the Describe/List/UpdateContributorInsights family.

Package dynamodb implements the AWS DynamoDB mock service. execute_transaction.go implements ExecuteTransaction: a set of PartiQL DML statements executed atomically via snapshot-based rollback.

Package dynamodb implements the AWS DynamoDB mock service. global_tables.go implements the CreateGlobalTable/DescribeGlobalTable/ ListGlobalTables/UpdateGlobalTable/UpdateGlobalTableSettings family. Physical replica propagation lives in replication.go.

Package dynamodb implements the AWS DynamoDB mock service. handler_autoscaling.go implements the wire-JSON handler for UpdateTableReplicaAutoScaling. Routing (dispatchExtraOps) stays in handler.go; this is the leaf implementation it calls into. Backend logic lives in autoscaling.go.

Package dynamodb implements the AWS DynamoDB mock service. handler_backups.go implements the wire-JSON handlers for continuous backups/PITR, ExportTableToPointInTime/DescribeExport/ListExports, and DescribeTableReplicaAutoScaling. Routing (dispatchBackupOps) stays in handler.go; these are the leaf implementations it calls into.

Package dynamodb implements the AWS DynamoDB mock service. handler_contributor_insights.go implements the wire-JSON handlers for the contributor insights family. Routing (dispatchExtraOps) stays in handler.go; these are the leaf implementations it calls into. Backend logic lives in contributor_insights.go.

Package dynamodb implements the AWS DynamoDB mock service. handler_execute_transaction.go implements the wire-JSON handler for ExecuteTransaction. Routing (dispatch) stays in handler.go; this is the leaf implementation it calls into. Backend logic lives in execute_transaction.go.

Package dynamodb implements the AWS DynamoDB mock service. handler_global_tables.go implements the wire-JSON handlers for the global tables family (Create/Describe/List/Update GlobalTable[Settings]). Routing (dispatchExtraOps) stays in handler.go; these are the leaf implementations it calls into. Backend logic lives in global_tables.go.

Package dynamodb implements the AWS DynamoDB mock service. handler_import.go implements the wire-JSON handlers for DescribeImport/ImportTable/ListImports. Routing (dispatchExtraOps) stays in handler.go; these are the leaf implementations it calls into. Backend logic lives in import_export_s3.go.

Package dynamodb implements the AWS DynamoDB mock service. handler_kinesis_streaming.go implements the wire-JSON handlers for the Kinesis streaming destination family. Routing (dispatchExtraOps) stays in handler.go; these are the leaf implementations it calls into. Backend logic lives in kinesis_streaming.go.

Package dynamodb implements the AWS DynamoDB mock service. handler_limits_endpoints.go implements the wire-JSON handlers for DescribeLimits and DescribeEndpoints. Routing (dispatchExtraOps) stays in handler.go; these are the leaf implementations it calls into. Backend logic lives in limits_endpoints.go.

Package dynamodb implements the AWS DynamoDB mock service. handler_resource_policy.go implements the wire-JSON handlers for Get/Put/DeleteResourcePolicy. Routing (dispatchExtraOps) stays in handler.go; these are the leaf implementations it calls into. Backend logic lives in resource_policy.go.

Package dynamodb implements the AWS DynamoDB mock service. kinesis_streaming.go implements the Kinesis Data Streams destination family: Describe/Enable/Disable/UpdateKinesisStreamingDestination.

Package dynamodb implements the AWS DynamoDB mock service. limits_endpoints.go implements DescribeLimits and DescribeEndpoints, which return hardcoded account/table capacity limits and regional endpoint info.

Package dynamodb implements the AWS DynamoDB mock service. projection.go validates and resolves ProjectionExpression / AttributesToGet parameters shared by GetItem, BatchGetItem, Query, and Scan.

Package dynamodb implements the AWS DynamoDB mock service. replication.go propagates completed item writes to sibling global-table replicas, and provides the table-cloning/mutex-construction helpers used when a replica table is physically instantiated.

Package dynamodb implements the AWS DynamoDB mock service. resource_policy.go implements Get/Put/DeleteResourcePolicy (resource-based access policies attached to a table by ARN).

Package dynamodb implements the AWS DynamoDB mock service. streams_shard_iterator.go implements opaque shard-iterator tokens for the DynamoDB Streams GetShardIterator/GetRecords API.

Package dynamodb implements the AWS DynamoDB mock service. table_validation.go validates CreateTable/UpdateTable structural constraints: GSI/LSI counts, key schema shape, billing mode, and provisioned throughput.

Package dynamodb implements the AWS DynamoDB mock service. transact_validation.go validates TransactWriteItems / TransactGetItems input: duplicate-key detection, the 4 MB total size limit, key-modification guards, and the 100-item count limit.

Index

Constants

View Source 
const (
	MaxItemSize         = 400 * 1024 // 400 KB
	MaxPartitionKeySize = 2048       // 2048 bytes
	MaxSortKeySize      = 1024       // 1024 bytes

)

Variables

View Source 
var (
	ErrUnclosedBracket = errors.New("unclosed bracket in path")
	ErrInvalidIndex    = errors.New("invalid list index")
	ErrNonMapAccess    = errors.New("cannot access key on non-map")
	ErrNonMapItem      = errors.New("item is not a Map")
	ErrNonListAccess   = errors.New("cannot access index on non-list")
	ErrIndexOutOfRange = errors.New("index out of bounds")
)

Sentinel errors for path operations.

View Source 
var (
	ErrInvalidAttributeValue = errors.New("expected map[string]any for attribute value")
	ErrInvalidTypeKeyCount   = errors.New("expected exactly 1 type key")
	ErrTypeMismatchS         = errors.New("expected string for S")
	ErrTypeMismatchN         = errors.New("expected string for N")
	ErrTypeMismatchBOOL      = errors.New("expected bool for BOOL")
	ErrTypeMismatchM         = errors.New("expected map for M")
	ErrTypeMismatchL         = errors.New("expected slice for L")
	ErrTypeMismatchB         = errors.New("expected []byte or base64 string for B")
	ErrUnknownAttributeType  = errors.New("unknown attribute type")
	ErrEmptySequenceNumber   = errors.New("empty sequence number")
)

Sentinel errors for streams operations.

View Source 
var ErrInvalidStatement = errors.New("invalid PartiQL statement")

ErrInvalidStatement is returned when a PartiQL statement cannot be parsed.

View Source 
var ErrUnknownOperation = errors.New("UnknownOperationException")
View Source 
var ErrValidation = errors.New("validation error")

ErrValidation is a sentinel used for validation-related errors so callers can use errors.Is without parsing error message strings.

View Source 
var TTLGracePeriod = 0 * time.Second //nolint:gochecknoglobals // intentional package-level default

TTLGracePeriod is the extra time added after an item's TTL timestamp before it is actually evicted. AWS DynamoDB documents a 48-hour grace period in production. Tests should pass 0 to avoid timing dependencies.

Functions

func BuildKeyString 

func BuildKeyString(item map[string]any, attrName string) string

func CalculateAttrSize 

func CalculateAttrSize(v any) int64

CalculateAttrSize estimates the encoded size of a single DynamoDB wire-format attribute value.

func CalculateItemSize 

func CalculateItemSize(item map[string]any) (int, error)

CalculateItemSize approximates the DynamoDB-encoded size of a wire-format item in bytes.

func EvaluateExpression 

func EvaluateExpression(
	expression string,
	item map[string]any,
	attrValues map[string]any,
	attrNames map[string]string,
) (bool, error)

EvaluateExpression evaluates a DynamoDB condition expression against an item.

func FromStreamAttributeValue added in v1.1.1 

func FromStreamAttributeValue(av streamstypes.AttributeValue) (map[string]any, error)

func FromStreamItem added in v1.1.1 

func FromStreamItem(item map[string]streamstypes.AttributeValue) (map[string]any, error)

func ReadCapacityUnits 

func ReadCapacityUnits(item map[string]any) float64

ReadCapacityUnits returns the RCUs consumed by an eventually-consistent read: 0.5 RCU per 4 KB (ceiling), minimum 0.5.

func ValidateDataTypes 

func ValidateDataTypes(item map[string]any) error

ValidateDataTypes checks basic type conformance.

func ValidateItemSize 

func ValidateItemSize(item map[string]any) error

func WithRegion 

func WithRegion(ctx context.Context, region string) context.Context

WithRegion returns a derived context that carries the given AWS region. External callers (e.g. the DynamoDB Streams handler) use this to scope backend operations to the request's SigV4 region.

func WriteCapacityUnits 

func WriteCapacityUnits(item map[string]any) float64

WriteCapacityUnits returns the WCUs consumed by a write: ceil(size / 1KB), minimum 1.

Types

type Backup 

type Backup struct {
	CreationDateTime       time.Time                               `json:"CreationDateTime"`
	SSEKMSMasterKeyArn     string                                  `json:"SSEKMSMasterKeyArn,omitempty"`
	SSEType                string                                  `json:"SSEType,omitempty"`
	StreamViewType         string                                  `json:"StreamViewType,omitempty"`
	BackupName             string                                  `json:"BackupName"`
	BackupType             string                                  `json:"BackupType"`
	TableArn               string                                  `json:"TableArn"`
	TableID                string                                  `json:"TableID"`
	BackupArn              string                                  `json:"BackupArn"`
	BackupStatus           string                                  `json:"BackupStatus"`
	TableName              string                                  `json:"TableName"`
	BillingMode            string                                  `json:"BillingMode,omitempty"`
	KeySchema              []models.KeySchemaElement               `json:"KeySchema"`
	LocalSecondaryIndexes  []models.LocalSecondaryIndex            `json:"LocalSecondaryIndexes,omitempty"`
	GlobalSecondaryIndexes []models.GlobalSecondaryIndex           `json:"GlobalSecondaryIndexes,omitempty"`
	Items                  []map[string]any                        `json:"Items"`
	AttributeDefinitions   []models.AttributeDefinition            `json:"AttributeDefinitions"`
	ProvisionedThroughput  models.ProvisionedThroughputDescription `json:"ProvisionedThroughput"`
	SizeBytes              int64                                   `json:"SizeBytes"`
	SSEEnabled             bool                                    `json:"SSEEnabled,omitempty"`
	StreamsEnabled         bool                                    `json:"StreamsEnabled,omitempty"`
}

Backup holds the metadata and a point-in-time item snapshot for a DynamoDB on-demand backup.

type CancellationReason 

type CancellationReason struct {
	Item    any    `json:"Item,omitempty"`
	Code    string `json:"Code"`
	Message string `json:"Message,omitempty"`
}

type ConfigProvider 

type ConfigProvider interface {
	GetDynamoDBSettings() Settings
}

ConfigProvider is a private interface to extract DynamoDB configuration from the abstract AppContext Config.

type DashboardHandlers 

type DashboardHandlers struct {
	HandleDynamoDB func(http.ResponseWriter, *http.Request, string)
}

DashboardHandlers defines the functions needed for DynamoDB dashboard routes.

type DashboardProvider 

type DashboardProvider struct {
	// Handlers will be set by the dashboard handler during initialization
	Handlers DashboardHandlers
}

DashboardProvider implements the service.DashboardProvider interface to enable DynamoDB dashboard discovery and route registration. It wraps a reference to dashboard handler functions.

func NewDashboardProvider 

func NewDashboardProvider() *DashboardProvider

NewDashboardProvider creates a new DynamoDB dashboard provider. The handlers must be set by the dashboard initialization code before any routes are registered.

func (*DashboardProvider) DashboardName 

func (p *DashboardProvider) DashboardName() string

func (*DashboardProvider) DashboardRoutePrefix 

func (p *DashboardProvider) DashboardRoutePrefix() string

DashboardRoutePrefix returns the URL path prefix for DynamoDB dashboard routes.

func (*DashboardProvider) RegisterDashboardRoutes 

func (p *DashboardProvider) RegisterDashboardRoutes(
	group *echo.Group,
	_ any,
	_ string,
)

RegisterDashboardRoutes registers all DynamoDB dashboard routes under the given Echo group. The group is mounted at /dashboard/dynamodb by the dashboard handler.

type DynamoDBHandler 

type DynamoDBHandler struct {
	Backend StorageBackend
	Streams StreamsBackend

	DefaultRegion string
	// contains filtered or unexported fields
}

DynamoDBHandler handles HTTP requests for DynamoDB operations.

func NewHandler 

func NewHandler(backend StorageBackend) *DynamoDBHandler

NewHandler creates a new DynamoDB handler with the given storage backend.

func (*DynamoDBHandler) ChaosOperations 

func (h *DynamoDBHandler) ChaosOperations() []string

ChaosOperations returns all operations that can be fault-injected.

func (*DynamoDBHandler) ChaosRegions 

func (h *DynamoDBHandler) ChaosRegions() []string

ChaosRegions returns all regions this DynamoDB instance handles.

func (*DynamoDBHandler) ChaosServiceName 

func (h *DynamoDBHandler) ChaosServiceName() string

ChaosServiceName returns the lowercase AWS service name for fault rule matching.

func (*DynamoDBHandler) DescribeTableInRegion 

func (h *DynamoDBHandler) DescribeTableInRegion(region, tableName string) *Table

DescribeTableInRegion returns a table from the backend for a specific region. Returns nil when not using the in-memory backend or when the table is not found.

func (*DynamoDBHandler) ExecuteFISAction 

func (h *DynamoDBHandler) ExecuteFISAction(
	ctx context.Context,
	action service.FISActionExecution,
) error

ExecuteFISAction executes a FIS action against resolved DynamoDB targets.

func (*DynamoDBHandler) ExtractOperation 

func (h *DynamoDBHandler) ExtractOperation(c *echo.Context) string

ExtractOperation extracts the DynamoDB operation from the X-Amz-Target header.

func (*DynamoDBHandler) ExtractResource 

func (h *DynamoDBHandler) ExtractResource(c *echo.Context) string

ExtractResource extracts the table name from the DynamoDB request body.

func (*DynamoDBHandler) FISActions 

func (h *DynamoDBHandler) FISActions() []service.FISActionDefinition

FISActions returns the FIS action definitions that the DynamoDB service supports.

func (*DynamoDBHandler) GetSupportedOperations 

func (h *DynamoDBHandler) GetSupportedOperations() []string

GetSupportedOperations returns a sorted list of supported DynamoDB operations.

func (*DynamoDBHandler) Handler 

func (h *DynamoDBHandler) Handler() echo.HandlerFunc

Handler is the Echo HTTP handler for DynamoDB operations.

func (*DynamoDBHandler) MatchPriority 

func (h *DynamoDBHandler) MatchPriority() int

MatchPriority returns the priority for the DynamoDB matcher. Header-based matchers have high priority (100).

func (*DynamoDBHandler) Name 

func (h *DynamoDBHandler) Name() string

Name returns the service identifier.

func (*DynamoDBHandler) Purge 

func (h *DynamoDBHandler) Purge(ctx context.Context, cutoff time.Time)

Purge implements service.Purgeable by deleting resources older than cutoff.

func (*DynamoDBHandler) Regions 

func (h *DynamoDBHandler) Regions() []string

Regions returns all regions with tables in the backend. Returns an empty slice when not using the in-memory backend.

func (*DynamoDBHandler) Reset 

func (h *DynamoDBHandler) Reset()

Reset clears all in-memory state from the backend. It is used by the POST /_gopherstack/reset endpoint for CI pipelines and rapid local development.

func (*DynamoDBHandler) Restore 

func (h *DynamoDBHandler) Restore(ctx context.Context, data []byte) error

Restore implements persistence.Persistable by delegating to the backend.

func (*DynamoDBHandler) RouteMatcher 

func (h *DynamoDBHandler) RouteMatcher() service.Matcher

RouteMatcher returns a matcher for DynamoDB requests (by X-Amz-Target header).

func (*DynamoDBHandler) Shutdown 

func (h *DynamoDBHandler) Shutdown(ctx context.Context)

Shutdown stops the janitor worker and waits for it to exit (or until ctx expires).

func (*DynamoDBHandler) Snapshot 

func (h *DynamoDBHandler) Snapshot(ctx context.Context) []byte

Snapshot implements persistence.Persistable by delegating to the backend.

func (*DynamoDBHandler) StartWorker 

func (h *DynamoDBHandler) StartWorker(ctx context.Context) error

StartWorker starts the background janitor if it is configured.

func (*DynamoDBHandler) TableNamesByRegion 

func (h *DynamoDBHandler) TableNamesByRegion(region string) []string

TableNamesByRegion returns table names in the given region (all if empty). Returns an empty slice when not using the in-memory backend.

func (*DynamoDBHandler) WithJanitor 

func (h *DynamoDBHandler) WithJanitor(
	settings Settings,
	janitorTimeout ...time.Duration,
) *DynamoDBHandler

WithJanitor attaches a background janitor to the handler. The optional janitorTimeout parameter bounds each individual janitor task; zero (or omitted) disables per-task timeouts.

type Error 

type Error struct {
	Type    string `json:"__type"`
	Message string `json:"message"`
	// Item carries the existing item on a ConditionalCheckFailedException when the
	// request set ReturnValuesOnConditionCheckFailure=ALL_OLD. AWS returns it so
	// optimistic-locking clients can inspect the current item without a re-read.
	Item                any                  `json:"Item,omitempty"`
	CancellationReasons []CancellationReason `json:"CancellationReasons,omitempty"`
}

func NewBackupInUseException 

func NewBackupInUseException(msg string) *Error

NewBackupInUseException returns an error indicating that a backup with the same name already exists for the table, or the backup ARN is already in use.

func NewConditionalCheckFailedException 

func NewConditionalCheckFailedException(msg string) *Error

func NewConditionalCheckFailedExceptionWithItem 

func NewConditionalCheckFailedExceptionWithItem(msg string, item any) *Error

NewConditionalCheckFailedExceptionWithItem returns a ConditionalCheckFailedException that also carries the existing item (already in DynamoDB wire/SDK attribute form). Pass a nil item to omit it.

func NewDuplicateItemException 

func NewDuplicateItemException(msg string) *Error

NewDuplicateItemException is returned by PartiQL INSERT when an item with the same primary key already exists. AWS DynamoDB raises this instead of silently overwriting (unlike PutItem which overwrites by default).

func NewExpiredIteratorException 

func NewExpiredIteratorException(msg string) *Error

func NewExportNotFoundException 

func NewExportNotFoundException(msg string) *Error

NewExportNotFoundException indicates the requested export ARN does not exist.

func NewImportNotFoundException 

func NewImportNotFoundException(msg string) *Error

NewImportNotFoundException indicates the requested import ARN does not exist.

func NewInternalServerError 

func NewInternalServerError(msg string) *Error

func NewItemCollectionSizeLimitExceededException 

func NewItemCollectionSizeLimitExceededException(msg string) *Error

func NewLimitExceededException 

func NewLimitExceededException(msg string) *Error

func NewPolicyNotFoundException 

func NewPolicyNotFoundException(msg string) *Error

func NewProvisionedThroughputExceededException 

func NewProvisionedThroughputExceededException(msg string) *Error

func NewReplicatedWriteConflictException 

func NewReplicatedWriteConflictException(msg string) *Error

func NewRequestLimitExceeded 

func NewRequestLimitExceeded(msg string) *Error

func NewResourceInUseException 

func NewResourceInUseException(msg string) *Error

func NewResourceNotFoundException 

func NewResourceNotFoundException(msg string) *Error

func NewShardIteratorCreationException 

func NewShardIteratorCreationException(msg string) *Error

func NewThrottlingException 

func NewThrottlingException(msg string) *Error

func NewTransactionCanceledException 

func NewTransactionCanceledException(msg string, reasons []CancellationReason) *Error

func NewTransactionInProgressException 

func NewTransactionInProgressException(msg string) *Error

func NewTrimmedDataAccessException 

func NewTrimmedDataAccessException(msg string) *Error

func NewValidationException 

func NewValidationException(msg string) *Error

func (*Error) Error 

func (e *Error) Error() string

type ExpressionCache 

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

ExpressionCache is a sharded LRU cache for parsed expressions. Uses multiple independent shards to reduce lock contention under concurrent access. Each entry has a TTL; expired entries are evicted lazily on Get and periodically by Sweep.

func NewExpressionCache 

func NewExpressionCache(capacity int) *ExpressionCache

NewExpressionCache creates a new sharded LRU cache with the given capacity. Capacity is divided equally among shards. Uses defaultExprCacheTTL for entry TTL.

func (*ExpressionCache) Close 

func (c *ExpressionCache) Close()

Close releases all resources held by the cache, including metric registrations.

func (*ExpressionCache) Get 

func (c *ExpressionCache) Get(key string) (any, bool)

Get retrieves a value from the cache. Expired entries are removed and a miss is returned.

func (*ExpressionCache) Put 

func (c *ExpressionCache) Put(key string, value any)

Put adds a value to the cache with the configured TTL.

func (*ExpressionCache) Sweep 

func (c *ExpressionCache) Sweep()

Sweep removes all expired entries from the cache. Intended to be called periodically by the janitor to bound memory usage over long-running sessions.

type InMemoryDB 

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

InMemoryDB stores tables and items organized by region.

tables, deletingTables, backups, globalTables, exports, imports, and streamARNIndex are *store.Table wrappers (see pkgs/store and store_setup.go) registered once on registry at construction time; txnTokens, txnPending, and fisReplicationPaused remain plain maps because their value type (time.Time) carries no identity a store.Table key function could extract -- see store_setup.go's doc comment.

func NewInMemoryDB 

func NewInMemoryDB() *InMemoryDB

func (*InMemoryDB) BatchExecuteStatement 

func (db *InMemoryDB) BatchExecuteStatement(
	ctx context.Context,
	input *sdkdynamodb.BatchExecuteStatementInput,
) (*sdkdynamodb.BatchExecuteStatementOutput, error)

BatchExecuteStatement executes multiple PartiQL statements and returns their results. It satisfies the StorageBackend interface using official AWS SDK v2 types.

AWS limit: at most maxBatchExecuteStatements statements per call. The ConsistentRead flag on each statement is forwarded to the underlying Query / Scan execution so strongly-consistent reads are honoured.

func (*InMemoryDB) BatchGetItem 

func (db *InMemoryDB) BatchGetItem(
	ctx context.Context,
	input *dynamodb.BatchGetItemInput,
) (*dynamodb.BatchGetItemOutput, error)

func (*InMemoryDB) BatchWriteItem 

func (db *InMemoryDB) BatchWriteItem(
	ctx context.Context,
	input *dynamodb.BatchWriteItemInput,
) (*dynamodb.BatchWriteItemOutput, error)

func (*InMemoryDB) Close 

func (db *InMemoryDB) Close()

Close releases all backend resources.

func (*InMemoryDB) CreateBackup 

func (db *InMemoryDB) CreateBackup(
	ctx context.Context,
	input *sdkdynamodb.CreateBackupInput,
) (*sdkdynamodb.CreateBackupOutput, error)

CreateBackup creates a point-in-time backup of the named DynamoDB table. It satisfies the StorageBackend interface using official AWS SDK v2 types.

func (*InMemoryDB) CreateGlobalTable 

func (db *InMemoryDB) CreateGlobalTable(
	_ context.Context,
	input *dynamodb.CreateGlobalTableInput,
) (*dynamodb.CreateGlobalTableOutput, error)

CreateGlobalTable creates a global table, physically instantiating replica Table entries in each specified region. If a table named GlobalTableName already exists in a region, it is adopted into the global table; otherwise a new empty table is created there. The source schema is taken from the first region where the table already exists. All replicas get their Replicas field populated with the other regions, matching the DescribeTable output that AWS returns for global tables.

func (*InMemoryDB) CreateTable 

func (db *InMemoryDB) CreateTable(
	ctx context.Context,
	input *dynamodb.CreateTableInput,
) (*dynamodb.CreateTableOutput, error)

func (*InMemoryDB) CreateTableInRegion 

func (db *InMemoryDB) CreateTableInRegion(
	ctx context.Context,
	input *dynamodb.CreateTableInput,
	region string,
) (*dynamodb.CreateTableOutput, error)

CreateTableInRegion creates a DynamoDB table in the specified region, bypassing the HTTP-layer region extraction. The supplied region always takes precedence, even if the context already carries a region value. Useful for tests that need tables in non-default regions.

func (*InMemoryDB) DeleteBackup 

func (db *InMemoryDB) DeleteBackup(
	ctx context.Context,
	input *sdkdynamodb.DeleteBackupInput,
) (*sdkdynamodb.DeleteBackupOutput, error)

DeleteBackup removes an existing backup by ARN and returns its description. It satisfies the StorageBackend interface using official AWS SDK v2 types.

func (*InMemoryDB) DeleteItem 

func (db *InMemoryDB) DeleteItem(
	ctx context.Context,
	input *dynamodb.DeleteItemInput,
) (*dynamodb.DeleteItemOutput, error)

func (*InMemoryDB) DeleteResourcePolicy 

func (db *InMemoryDB) DeleteResourcePolicy(
	_ context.Context,
	input *dynamodb.DeleteResourcePolicyInput,
) (*dynamodb.DeleteResourcePolicyOutput, error)

DeleteResourcePolicy removes the resource-based policy from the table.

func (*InMemoryDB) DeleteTable 

func (db *InMemoryDB) DeleteTable(
	ctx context.Context,
	input *dynamodb.DeleteTableInput,
) (*dynamodb.DeleteTableOutput, error)

func (*InMemoryDB) DescribeBackup 

func (db *InMemoryDB) DescribeBackup(
	ctx context.Context,
	input *sdkdynamodb.DescribeBackupInput,
) (*sdkdynamodb.DescribeBackupOutput, error)

DescribeBackup returns the full description of a backup by ARN. It satisfies the StorageBackend interface using official AWS SDK v2 types.

func (*InMemoryDB) DescribeContributorInsights 

func (db *InMemoryDB) DescribeContributorInsights(
	ctx context.Context,
	input *dynamodb.DescribeContributorInsightsInput,
) (*dynamodb.DescribeContributorInsightsOutput, error)

DescribeContributorInsights returns contributor insights status for a table or GSI. Status is tracked per-table on the in-memory backend; GSI-level status mirrors the table.

func (*InMemoryDB) DescribeEndpoints 

func (db *InMemoryDB) DescribeEndpoints(
	ctx context.Context,
	_ *dynamodb.DescribeEndpointsInput,
) (*dynamodb.DescribeEndpointsOutput, error)

DescribeEndpoints returns hardcoded regional DynamoDB endpoint information.

func (*InMemoryDB) DescribeGlobalTable 

func (db *InMemoryDB) DescribeGlobalTable(
	_ context.Context,
	input *dynamodb.DescribeGlobalTableInput,
) (*dynamodb.DescribeGlobalTableOutput, error)

DescribeGlobalTable returns the description of a global table.

func (*InMemoryDB) DescribeGlobalTableSettings 

func (db *InMemoryDB) DescribeGlobalTableSettings(
	_ context.Context,
	input *dynamodb.DescribeGlobalTableSettingsInput,
) (*dynamodb.DescribeGlobalTableSettingsOutput, error)

DescribeGlobalTableSettings returns per-replica settings for a global table.

func (*InMemoryDB) DescribeImport 

func (db *InMemoryDB) DescribeImport(
	ctx context.Context,
	input *dynamodb.DescribeImportInput,
) (*dynamodb.DescribeImportOutput, error)

DescribeImport returns the import description for a given import ARN. If the import was started via ImportTable, the stored record is returned. Otherwise, a synthetic COMPLETED response is returned for backwards compatibility.

func (*InMemoryDB) DescribeKinesisStreamingDestination 

func (db *InMemoryDB) DescribeKinesisStreamingDestination(
	ctx context.Context,
	input *dynamodb.DescribeKinesisStreamingDestinationInput,
) (*dynamodb.DescribeKinesisStreamingDestinationOutput, error)

DescribeKinesisStreamingDestination returns the Kinesis streaming destinations for a table.

func (*InMemoryDB) DescribeLimits 

func (db *InMemoryDB) DescribeLimits(
	_ context.Context,
	_ *dynamodb.DescribeLimitsInput,
) (*dynamodb.DescribeLimitsOutput, error)

DescribeLimits returns hardcoded account and table provisioned throughput limits.

func (*InMemoryDB) DescribeStream 

func (db *InMemoryDB) DescribeStream(
	_ context.Context,
	input *dynamodbstreams.DescribeStreamInput,
) (*dynamodbstreams.DescribeStreamOutput, error)

DescribeStream returns details about a stream (identified by its ARN). Supports ExclusiveStartShardId pagination.

func (*InMemoryDB) DescribeTable 

func (db *InMemoryDB) DescribeTable(
	ctx context.Context,
	input *dynamodb.DescribeTableInput,
) (*dynamodb.DescribeTableOutput, error)

func (*InMemoryDB) DescribeTimeToLive 

func (db *InMemoryDB) DescribeTimeToLive(
	ctx context.Context,
	input *dynamodb.DescribeTimeToLiveInput,
) (*dynamodb.DescribeTimeToLiveOutput, error)

func (*InMemoryDB) DisableKinesisStreamingDestination 

func (db *InMemoryDB) DisableKinesisStreamingDestination(
	ctx context.Context,
	input *dynamodb.DisableKinesisStreamingDestinationInput,
) (*dynamodb.DisableKinesisStreamingDestinationOutput, error)

DisableKinesisStreamingDestination removes a Kinesis streaming destination from a table.

func (*InMemoryDB) DisableStream 

func (db *InMemoryDB) DisableStream(ctx context.Context, tableName string) error

DisableStream disables DynamoDB Streams on a table.

func (*InMemoryDB) EnableKinesisStreamingDestination 

func (db *InMemoryDB) EnableKinesisStreamingDestination(
	ctx context.Context,
	input *dynamodb.EnableKinesisStreamingDestinationInput,
) (*dynamodb.EnableKinesisStreamingDestinationOutput, error)

EnableKinesisStreamingDestination adds a Kinesis streaming destination to a table.

func (*InMemoryDB) EnableStream 

func (db *InMemoryDB) EnableStream(ctx context.Context, tableName, viewType string) error

EnableStream enables DynamoDB Streams on a table with the given view type.

func (*InMemoryDB) ExecuteTransaction 

func (db *InMemoryDB) ExecuteTransaction(
	ctx context.Context,
	input *dynamodb.ExecuteTransactionInput,
) (*dynamodb.ExecuteTransactionOutput, error)

ExecuteTransaction executes a set of PartiQL DML statements atomically. Atomicity is provided via snapshot-based rollback: pre-transaction snapshots of all affected tables are captured, statements are executed sequentially, and all tables are restored from their snapshots if any statement fails. This matches the observable contract of real AWS ExecuteTransaction for single-process in-memory usage.

func (*InMemoryDB) GetItem 

func (db *InMemoryDB) GetItem(
	ctx context.Context,
	input *dynamodb.GetItemInput,
) (*dynamodb.GetItemOutput, error)

func (*InMemoryDB) GetRecentEvents 

func (db *InMemoryDB) GetRecentEvents(tableName string) []models.StreamRecord

func (*InMemoryDB) GetRecords 

func (db *InMemoryDB) GetRecords(
	ctx context.Context,
	input *dynamodbstreams.GetRecordsInput,
) (*dynamodbstreams.GetRecordsOutput, error)

GetRecords reads stream records starting from the given opaque shard iterator.

func (*InMemoryDB) GetResourcePolicy 

func (db *InMemoryDB) GetResourcePolicy(
	_ context.Context,
	input *dynamodb.GetResourcePolicyInput,
) (*dynamodb.GetResourcePolicyOutput, error)

GetResourcePolicy returns the resource-based policy stored on the table.

func (*InMemoryDB) GetShardIterator 

func (db *InMemoryDB) GetShardIterator(
	_ context.Context,
	input *dynamodbstreams.GetShardIteratorInput,
) (*dynamodbstreams.GetShardIteratorOutput, error)

GetShardIterator returns an opaque shard iterator for reading stream records. The iterator is a random token stored in the server-side ShardIteratorStore, preventing clients from decoding or forging iterator state.

func (*InMemoryDB) GetTable 

func (db *InMemoryDB) GetTable(name string) (*Table, bool)

GetTable returns a table by name from the default region (for UI/backward compatibility).

func (*InMemoryDB) GetTableInRegion 

func (db *InMemoryDB) GetTableInRegion(name string, region string) (*Table, bool)

GetTableInRegion returns a table by name from a specific region.

func (*InMemoryDB) ImportTable 

func (db *InMemoryDB) ImportTable(
	ctx context.Context,
	input *dynamodb.ImportTableInput,
) (*dynamodb.ImportTableOutput, error)

ImportTable creates the target table from TableCreationParameters and, when an S3 backend is wired, populates it from the source objects (DYNAMODB_JSON or CSV, optionally gzip-compressed). It records accurate counts so DescribeImport and ListImports report real progress. ION input is reported as a FAILED import.

func (*InMemoryDB) IsReplicationPaused 

func (db *InMemoryDB) IsReplicationPaused(tableARNOrName string) bool

IsReplicationPaused reports whether FIS global-table-pause-replication is currently active for the given table ARN or name. Expired entries are lazily evicted to prevent unbounded map growth.

func (*InMemoryDB) ListAllTables 

func (db *InMemoryDB) ListAllTables() []*Table

ListAllTables returns a slice of all tables across all regions (for UI).

func (*InMemoryDB) ListContributorInsights 

func (db *InMemoryDB) ListContributorInsights(
	ctx context.Context,
	_ *dynamodb.ListContributorInsightsInput,
) (*dynamodb.ListContributorInsightsOutput, error)

ListContributorInsights returns the set of tables whose contributor insights are enabled, scoped to the request region.

func (*InMemoryDB) ListGlobalTables 

func (db *InMemoryDB) ListGlobalTables(
	_ context.Context,
	input *dynamodb.ListGlobalTablesInput,
) (*dynamodb.ListGlobalTablesOutput, error)

ListGlobalTables returns global tables, optionally filtered by region, with pagination support.

func (*InMemoryDB) ListImports 

func (db *InMemoryDB) ListImports(
	ctx context.Context,
	input *dynamodb.ListImportsInput,
) (*dynamodb.ListImportsOutput, error)

ListImports returns stored import records for the request region. Supports NextToken-based pagination and PageSize per the real AWS API.

func (*InMemoryDB) ListStreams 

func (db *InMemoryDB) ListStreams(
	ctx context.Context,
	input *dynamodbstreams.ListStreamsInput,
) (*dynamodbstreams.ListStreamsOutput, error)

ListStreams returns a list of all enabled streams, optionally filtered by table name. Supports ExclusiveStartStreamArn and Limit for pagination. Only streams whose ARN region matches the request region (from ctx) are returned.

func (*InMemoryDB) ListTables 

func (db *InMemoryDB) ListTables(
	ctx context.Context,
	input *dynamodb.ListTablesInput,
) (*dynamodb.ListTablesOutput, error)

func (*InMemoryDB) ListTagsOfResource 

func (db *InMemoryDB) ListTagsOfResource(
	_ context.Context,
	input *dynamodb.ListTagsOfResourceInput,
) (*dynamodb.ListTagsOfResourceOutput, error)

ListTagsOfResource returns the tags attached to a DynamoDB table identified by its ARN.

func (*InMemoryDB) Purge 

func (db *InMemoryDB) Purge(ctx context.Context, cutoff time.Time)

Purge removes tables and backups created before the cutoff time.

func (*InMemoryDB) PutItem 

func (db *InMemoryDB) PutItem(
	ctx context.Context,
	input *dynamodb.PutItemInput,
) (*dynamodb.PutItemOutput, error)

func (*InMemoryDB) PutResourcePolicy 

func (db *InMemoryDB) PutResourcePolicy(
	_ context.Context,
	input *dynamodb.PutResourcePolicyInput,
) (*dynamodb.PutResourcePolicyOutput, error)

PutResourcePolicy stores a resource-based policy on the table.

func (*InMemoryDB) Query 

func (db *InMemoryDB) Query(
	ctx context.Context,
	input *dynamodb.QueryInput,
) (*dynamodb.QueryOutput, error)

func (*InMemoryDB) QueryWithContext 

func (db *InMemoryDB) QueryWithContext(
	ctx context.Context,
	input *dynamodb.QueryInput,
) (*dynamodb.QueryOutput, error)

func (*InMemoryDB) Regions 

func (db *InMemoryDB) Regions() []string

Regions returns all distinct regions that contain at least one table.

func (*InMemoryDB) Reset 

func (db *InMemoryDB) Reset()

Reset clears all in-memory state from the database. It is used by the POST /_gopherstack/reset endpoint for CI pipelines and rapid local development.

func (*InMemoryDB) Restore 

func (db *InMemoryDB) Restore(ctx context.Context, data []byte) error

Restore loads backend state from a JSON snapshot. It implements persistence.Persistable.

func (*InMemoryDB) Scan 

func (db *InMemoryDB) Scan(
	ctx context.Context,
	input *dynamodb.ScanInput,
) (*dynamodb.ScanOutput, error)

func (*InMemoryDB) ScanWithContext 

func (db *InMemoryDB) ScanWithContext(
	ctx context.Context,
	input *dynamodb.ScanInput,
) (*dynamodb.ScanOutput, error)

func (*InMemoryDB) SetCreateDelay 

func (db *InMemoryDB) SetCreateDelay(d time.Duration)

SetCreateDelay sets the CREATING → ACTIVE transition delay. Call before CreateTable calls; intended for tests and CLI configuration.

func (*InMemoryDB) SetDefaultRegion 

func (db *InMemoryDB) SetDefaultRegion(region string)

SetDefaultRegion sets the default region for this backend.

func (*InMemoryDB) SetEnforceThroughput 

func (db *InMemoryDB) SetEnforceThroughput(enabled bool)

SetEnforceThroughput enables or disables provisioned throughput throttling. Call before CreateTable calls; intended for CLI configuration.

func (*InMemoryDB) SetKinesisEmitter 

func (db *InMemoryDB) SetKinesisEmitter(emitter KinesisEmitter)

SetKinesisEmitter installs a Kinesis emitter for all tables in this DB. Safe to call once during service wiring.

func (*InMemoryDB) SetS3Backend 

func (db *InMemoryDB) SetS3Backend(s3 S3Accessor)

SetS3Backend wires the S3 backend used for ImportTable / ExportTableToPointInTime.

func (*InMemoryDB) Snapshot 

func (db *InMemoryDB) Snapshot(ctx context.Context) []byte

Snapshot serialises the backend state to JSON. It implements persistence.Persistable. Per-table stream sequence counters (streamSeq) are unexported and therefore not serialised directly; they are reconstructed during Restore from the highest SequenceNumber found in each table's StreamRecords ring buffer.

func (*InMemoryDB) TableNamesByRegion 

func (db *InMemoryDB) TableNamesByRegion(region string) []string

TableNamesByRegion returns table names in the given region, or all regions if region is empty.

func (*InMemoryDB) TagResource 

func (db *InMemoryDB) TagResource(
	_ context.Context,
	input *dynamodb.TagResourceInput,
) (*dynamodb.TagResourceOutput, error)

TagResource attaches tags to a DynamoDB table identified by its ARN.

func (*InMemoryDB) TaggedTables 

func (db *InMemoryDB) TaggedTables() []TaggedTableInfo

TaggedTables returns a snapshot of all DynamoDB tables with their ARNs and tags. Intended for use by the Resource Groups Tagging API provider.

func (*InMemoryDB) TransactGetItems 

func (db *InMemoryDB) TransactGetItems(
	ctx context.Context,
	input *dynamodb.TransactGetItemsInput,
) (*dynamodb.TransactGetItemsOutput, error)

TransactGetItems reads up to 100 items atomically.

func (*InMemoryDB) TransactWriteItems 

func (db *InMemoryDB) TransactWriteItems(
	ctx context.Context,
	input *dynamodb.TransactWriteItemsInput,
) (*dynamodb.TransactWriteItemsOutput, error)

TransactWriteItems executes up to 100 write actions atomically.

func (*InMemoryDB) UntagResource 

func (db *InMemoryDB) UntagResource(
	_ context.Context,
	input *dynamodb.UntagResourceInput,
) (*dynamodb.UntagResourceOutput, error)

UntagResource removes tags from a DynamoDB table identified by its ARN.

func (*InMemoryDB) UpdateContributorInsights 

func (db *InMemoryDB) UpdateContributorInsights(
	ctx context.Context,
	input *dynamodb.UpdateContributorInsightsInput,
) (*dynamodb.UpdateContributorInsightsOutput, error)

UpdateContributorInsights toggles contributor insights for a table. The action is interpreted as ENABLE / DISABLE per AWS spec.

func (*InMemoryDB) UpdateGlobalTable 

func (db *InMemoryDB) UpdateGlobalTable(
	_ context.Context,
	input *dynamodb.UpdateGlobalTableInput,
) (*dynamodb.UpdateGlobalTableOutput, error)

UpdateGlobalTable adds or removes replica regions for an existing global table. Create actions physically create a new Table entry in the target region (cloning the source schema). Delete actions remove the Table entry from the target region.

func (*InMemoryDB) UpdateGlobalTableSettings 

func (db *InMemoryDB) UpdateGlobalTableSettings(
	_ context.Context,
	input *dynamodb.UpdateGlobalTableSettingsInput,
) (*dynamodb.UpdateGlobalTableSettingsOutput, error)

UpdateGlobalTableSettings persists global and per-replica billing/throughput settings and returns the updated state for each replica.

func (*InMemoryDB) UpdateItem 

func (db *InMemoryDB) UpdateItem(
	ctx context.Context,
	input *dynamodb.UpdateItemInput,
) (*dynamodb.UpdateItemOutput, error)

func (*InMemoryDB) UpdateKinesisStreamingDestination 

func (db *InMemoryDB) UpdateKinesisStreamingDestination(
	ctx context.Context,
	input *dynamodb.UpdateKinesisStreamingDestinationInput,
) (*dynamodb.UpdateKinesisStreamingDestinationOutput, error)

UpdateKinesisStreamingDestination updates the precision configuration of an existing Kinesis streaming destination. The precision change is persisted and reflected in subsequent DescribeKinesisStreamingDestination calls.

func (*InMemoryDB) UpdateTable 

func (db *InMemoryDB) UpdateTable(
	ctx context.Context,
	input *dynamodb.UpdateTableInput,
) (*dynamodb.UpdateTableOutput, error)

UpdateTable modifies a DynamoDB table's provisioned throughput, GSI list, stream spec, and replicas.

func (*InMemoryDB) UpdateTableReplicaAutoScaling 

func (db *InMemoryDB) UpdateTableReplicaAutoScaling(
	ctx context.Context,
	input *dynamodb.UpdateTableReplicaAutoScalingInput,
) (*dynamodb.UpdateTableReplicaAutoScalingOutput, error)

UpdateTableReplicaAutoScaling persists the autoscaling settings for a table's replicas so DescribeTableReplicaAutoScaling can round-trip the configured values.

func (*InMemoryDB) UpdateTimeToLive 

func (db *InMemoryDB) UpdateTimeToLive(
	ctx context.Context,
	input *dynamodb.UpdateTimeToLiveInput,
) (*dynamodb.UpdateTimeToLiveOutput, error)

type Janitor 

type Janitor struct {
	Backend  *InMemoryDB
	Interval time.Duration

	// TaskTimeout bounds each individual janitor task (TTL sweep, table cleaner, etc.).
	// When non-zero, each task runs with a child context that expires after this duration,
	// preventing a stalled operation from blocking the janitor loop indefinitely.
	TaskTimeout time.Duration
	// contains filtered or unexported fields
}

Janitor is the DynamoDB background worker that finalises tables queued for async deletion and records queue-depth metrics for the live dashboard.

func NewJanitor 

func NewJanitor(backend *InMemoryDB, settings Settings) *Janitor

NewJanitor creates a new DynamoDB Janitor for the given backend. The janitor interval is taken from the provided settings; if zero, it falls back to defaultDDBJanitorInterval.

func (*Janitor) Run 

func (j *Janitor) Run(ctx context.Context)

Run runs the janitor loop until ctx is cancelled. Two independent tickers are used:

  • the main ticker (Interval, default 500ms): housekeeping tasks (table cleanup, txn-token sweeps, expression-cache evictions).
  • the TTL ticker (defaultDDBTTLSweepInterval, 5s): per-table TTL and stream-record sweeps, which are O(tables × items) and too expensive to run every 500ms.

Each sweep is panic-recovered and bounded by TaskTimeout (if non-zero) by the worker primitive.

func (*Janitor) SweepOnce 

func (j *Janitor) SweepOnce(ctx context.Context)

SweepOnce runs a single full sweep pass. Exposed for testing.

type KinesisDestinationEntry 

type KinesisDestinationEntry struct {
	StreamARN string `json:"StreamARN"`
	// Precision is "MICROSECOND" or "MILLISECOND"; empty means MILLISECOND (AWS default).
	Precision string `json:"Precision,omitempty"`
}

KinesisDestinationEntry stores a Kinesis streaming destination with its configuration.

type KinesisEmitter 

type KinesisEmitter interface {
	EmitDynamoDBStreamRecord(streamARN, tableName string, record models.StreamRecord)
}

KinesisEmitter forwards DynamoDB stream records to configured Kinesis destinations. Implementations must be safe to call while the caller holds locks; they should return promptly and dispatch work to a background goroutine if needed.

type ParsedCondition 

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

ParsedCondition is a pre-parsed condition or filter expression AST. Pre-parsing once and reusing across many items avoids per-item lexing overhead.

func ParseConditionStr 

func ParseConditionStr(expression string) (*ParsedCondition, error)

ParseConditionStr parses a DynamoDB condition expression string once. Returns a zero ParsedCondition when expression is empty (always matches).

func (*ParsedCondition) Evaluate 

func (c *ParsedCondition) Evaluate(
	item map[string]any,
	attrValues map[string]any,
	attrNames map[string]string,
) bool

Evaluate runs the pre-parsed condition against item. A zero ParsedCondition (nil node) always returns true (matches everything).

type Projector 

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

Projector holds a pre-parsed projection expression for efficient repeated use.

func ParseProjector 

func ParseProjector(expression string, attrNames map[string]string) (*Projector, error)

ParseProjector parses a ProjectionExpression and returns a Projector.

func (*Projector) Project 

func (p *Projector) Project(item map[string]any) map[string]any

Project applies the pre-parsed projection to an item.

type Provider 

type Provider struct{}

Provider implements service.Provider for the DynamoDB service.

func (*Provider) Init 

func (p *Provider) Init(ctx *service.AppContext) (service.Registerable, error)

Init initializes the DynamoDB service backend, janitor, and handler.

func (*Provider) Name 

func (p *Provider) Name() string

Name returns the logical name of the provider.

type S3Accessor 

type S3Accessor interface {
	GetObject(ctx context.Context, in *s3sdk.GetObjectInput) (*s3sdk.GetObjectOutput, error)
	ListObjectsV2(
		ctx context.Context,
		in *s3sdk.ListObjectsV2Input,
	) (*s3sdk.ListObjectsV2Output, error)
	PutObject(ctx context.Context, in *s3sdk.PutObjectInput) (*s3sdk.PutObjectOutput, error)
}

S3Accessor is the subset of S3 operations DynamoDB needs to read ImportTable source objects and write ExportTableToPointInTime output. It is satisfied by the in-process S3 backend, wired in cli.go alongside the Firehose→S3 wiring.

type Settings 

type Settings struct {
	DefaultRegion     string        `json:"default_region"       env:"DYNAMODB_REGION"               default:"us-east-1" help:"Default region for DynamoDB."` //nolint:lll // Kong struct tag makes this line long
	JanitorInterval   time.Duration `json:"janitor_interval"     env:"DYNAMODB_JANITOR_INTERVAL"     default:"500ms"     help:"Janitor interval."`            //nolint:lll // Kong struct tag makes this line long
	TTLSweepBatchSize int           ``                                                                                                                        //nolint:lll // Kong struct tag makes this line long
	/* 144-byte string literal not displayed */
	// CreateDelay is the simulated CREATING → ACTIVE transition time.
	// Set to 0 (default) for immediate table activation (no lifecycle transition).
	CreateDelay time.Duration `` //nolint:lll,golines // Kong struct tag makes this line long
	/* 147-byte string literal not displayed */
	// EnforceThroughput enables token-bucket throughput throttling per table.
	// When true, operations that exceed the provisioned RCU/WCU return ProvisionedThroughputExceededException.
	EnforceThroughput bool `` //nolint:lll,golines // Kong struct tag makes this line long
	/* 154-byte string literal not displayed */
}

Settings holds service-level configuration for the DynamoDB backend. Fields are picked up by the Kong CLI parser when this struct is embedded in the root CLI command.

type ShardIteratorStore 

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

ShardIteratorStore maps opaque random tokens to server-side iterator state. It is goroutine-safe.

func NewShardIteratorStore 

func NewShardIteratorStore() *ShardIteratorStore

NewShardIteratorStore creates an empty ShardIteratorStore.

func (*ShardIteratorStore) Delete 

func (s *ShardIteratorStore) Delete(token string)

Delete removes a token from the store.

func (*ShardIteratorStore) Get 

func (s *ShardIteratorStore) Get(token string) *shardIteratorEntry

Get retrieves the entry for a token. Returns nil if the token is unknown.

func (*ShardIteratorStore) Put 

func (s *ShardIteratorStore) Put(tableName string, startSeq int64) (string, error)

Put stores a new iterator entry for an open shard and returns the opaque token.

func (*ShardIteratorStore) PutWithEnd 

func (s *ShardIteratorStore) PutWithEnd(tableName string, startSeq, endSeq int64) (string, error)

PutWithEnd stores a new iterator entry carrying the owning shard's ending sequence number (endSeq == 0 for an open shard) and returns the opaque token.

func (*ShardIteratorStore) Size 

func (s *ShardIteratorStore) Size() int

Size returns the number of entries currently in the store (including expired ones that have not yet been swept).

func (*ShardIteratorStore) Sweep 

func (s *ShardIteratorStore) Sweep()

Sweep removes expired entries from the store.

type StorageBackend 

type StorageBackend interface {
	// Table Operations
	CreateTable(context.Context, *dynamodb.CreateTableInput) (*dynamodb.CreateTableOutput, error)
	DeleteTable(context.Context, *dynamodb.DeleteTableInput) (*dynamodb.DeleteTableOutput, error)
	DescribeTable(
		context.Context,
		*dynamodb.DescribeTableInput,
	) (*dynamodb.DescribeTableOutput, error)
	ListTables(context.Context, *dynamodb.ListTablesInput) (*dynamodb.ListTablesOutput, error)
	UpdateTable(
		context.Context,
		*dynamodb.UpdateTableInput,
	) (*dynamodb.UpdateTableOutput, error)
	TagResource(
		context.Context,
		*dynamodb.TagResourceInput,
	) (*dynamodb.TagResourceOutput, error)
	UntagResource(
		context.Context,
		*dynamodb.UntagResourceInput,
	) (*dynamodb.UntagResourceOutput, error)
	ListTagsOfResource(
		context.Context,
		*dynamodb.ListTagsOfResourceInput,
	) (*dynamodb.ListTagsOfResourceOutput, error)
	UpdateTimeToLive(
		context.Context,
		*dynamodb.UpdateTimeToLiveInput,
	) (*dynamodb.UpdateTimeToLiveOutput, error)
	DescribeTimeToLive(
		context.Context,
		*dynamodb.DescribeTimeToLiveInput,
	) (*dynamodb.DescribeTimeToLiveOutput, error)

	// Item Operations
	PutItem(context.Context, *dynamodb.PutItemInput) (*dynamodb.PutItemOutput, error)
	GetItem(context.Context, *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error)
	DeleteItem(context.Context, *dynamodb.DeleteItemInput) (*dynamodb.DeleteItemOutput, error)
	UpdateItem(context.Context, *dynamodb.UpdateItemInput) (*dynamodb.UpdateItemOutput, error)
	Scan(context.Context, *dynamodb.ScanInput) (*dynamodb.ScanOutput, error)
	Query(context.Context, *dynamodb.QueryInput) (*dynamodb.QueryOutput, error)
	BatchGetItem(context.Context, *dynamodb.BatchGetItemInput) (*dynamodb.BatchGetItemOutput, error)
	BatchWriteItem(
		context.Context,
		*dynamodb.BatchWriteItemInput,
	) (*dynamodb.BatchWriteItemOutput, error)

	// Transaction Operations
	TransactWriteItems(
		context.Context,
		*dynamodb.TransactWriteItemsInput,
	) (*dynamodb.TransactWriteItemsOutput, error)
	TransactGetItems(
		context.Context,
		*dynamodb.TransactGetItemsInput,
	) (*dynamodb.TransactGetItemsOutput, error)

	// Global Table Operations
	CreateGlobalTable(
		context.Context,
		*dynamodb.CreateGlobalTableInput,
	) (*dynamodb.CreateGlobalTableOutput, error)
	DescribeGlobalTable(
		context.Context,
		*dynamodb.DescribeGlobalTableInput,
	) (*dynamodb.DescribeGlobalTableOutput, error)
	DescribeGlobalTableSettings(
		context.Context,
		*dynamodb.DescribeGlobalTableSettingsInput,
	) (*dynamodb.DescribeGlobalTableSettingsOutput, error)
	ListGlobalTables(
		context.Context,
		*dynamodb.ListGlobalTablesInput,
	) (*dynamodb.ListGlobalTablesOutput, error)
	UpdateGlobalTable(
		context.Context,
		*dynamodb.UpdateGlobalTableInput,
	) (*dynamodb.UpdateGlobalTableOutput, error)

	// Kinesis Streaming Operations
	EnableKinesisStreamingDestination(
		context.Context,
		*dynamodb.EnableKinesisStreamingDestinationInput,
	) (*dynamodb.EnableKinesisStreamingDestinationOutput, error)
	DescribeKinesisStreamingDestination(
		context.Context,
		*dynamodb.DescribeKinesisStreamingDestinationInput,
	) (*dynamodb.DescribeKinesisStreamingDestinationOutput, error)
	DisableKinesisStreamingDestination(
		context.Context,
		*dynamodb.DisableKinesisStreamingDestinationInput,
	) (*dynamodb.DisableKinesisStreamingDestinationOutput, error)

	// Resource Policy Operations
	GetResourcePolicy(
		context.Context,
		*dynamodb.GetResourcePolicyInput,
	) (*dynamodb.GetResourcePolicyOutput, error)
	PutResourcePolicy(
		context.Context,
		*dynamodb.PutResourcePolicyInput,
	) (*dynamodb.PutResourcePolicyOutput, error)
	DeleteResourcePolicy(
		context.Context,
		*dynamodb.DeleteResourcePolicyInput,
	) (*dynamodb.DeleteResourcePolicyOutput, error)

	// Miscellaneous Operations
	DescribeLimits(
		context.Context,
		*dynamodb.DescribeLimitsInput,
	) (*dynamodb.DescribeLimitsOutput, error)
	DescribeEndpoints(
		context.Context,
		*dynamodb.DescribeEndpointsInput,
	) (*dynamodb.DescribeEndpointsOutput, error)
	DescribeContributorInsights(
		context.Context,
		*dynamodb.DescribeContributorInsightsInput,
	) (*dynamodb.DescribeContributorInsightsOutput, error)
	DescribeImport(
		context.Context,
		*dynamodb.DescribeImportInput,
	) (*dynamodb.DescribeImportOutput, error)
	ListContributorInsights(
		context.Context,
		*dynamodb.ListContributorInsightsInput,
	) (*dynamodb.ListContributorInsightsOutput, error)
	UpdateContributorInsights(
		context.Context,
		*dynamodb.UpdateContributorInsightsInput,
	) (*dynamodb.UpdateContributorInsightsOutput, error)
	UpdateGlobalTableSettings(
		context.Context,
		*dynamodb.UpdateGlobalTableSettingsInput,
	) (*dynamodb.UpdateGlobalTableSettingsOutput, error)
	UpdateKinesisStreamingDestination(
		context.Context,
		*dynamodb.UpdateKinesisStreamingDestinationInput,
	) (*dynamodb.UpdateKinesisStreamingDestinationOutput, error)
	UpdateTableReplicaAutoScaling(
		context.Context,
		*dynamodb.UpdateTableReplicaAutoScalingInput,
	) (*dynamodb.UpdateTableReplicaAutoScalingOutput, error)
	ExecuteTransaction(
		context.Context,
		*dynamodb.ExecuteTransactionInput,
	) (*dynamodb.ExecuteTransactionOutput, error)
	ImportTable(
		context.Context,
		*dynamodb.ImportTableInput,
	) (*dynamodb.ImportTableOutput, error)
	ListImports(
		context.Context,
		*dynamodb.ListImportsInput,
	) (*dynamodb.ListImportsOutput, error)

	// Backup Operations
	CreateBackup(
		context.Context,
		*dynamodb.CreateBackupInput,
	) (*dynamodb.CreateBackupOutput, error)
	DescribeBackup(
		context.Context,
		*dynamodb.DescribeBackupInput,
	) (*dynamodb.DescribeBackupOutput, error)
	DeleteBackup(
		context.Context,
		*dynamodb.DeleteBackupInput,
	) (*dynamodb.DeleteBackupOutput, error)

	// PartiQL Batch Operations
	BatchExecuteStatement(
		context.Context,
		*dynamodb.BatchExecuteStatementInput,
	) (*dynamodb.BatchExecuteStatementOutput, error)
}

StorageBackend defines the interface for DynamoDB storage operations using official AWS SDK Go v2 types.

type StoredGlobalTable 

type StoredGlobalTable struct {
	CreationDateTime   time.Time                         `json:"CreationDateTime"`
	WriteCapacityUnits *int64                            `json:"WriteCapacityUnits,omitempty"`
	ReplicaSettings    map[string]*StoredReplicaSettings `json:"ReplicaSettings,omitempty"`
	GlobalTableName    string                            `json:"GlobalTableName"`
	GlobalTableArn     string                            `json:"GlobalTableArn"`
	BillingMode        string                            `json:"BillingMode,omitempty"`
	ReplicationGroup   []string                          `json:"ReplicationGroup"`
}

StoredGlobalTable holds the metadata for a DynamoDB global table.

type StoredReplicaSettings 

type StoredReplicaSettings struct {
	ReadCapacityUnits *int64 `json:"ReadCapacityUnits,omitempty"`
	TableClass        string `json:"TableClass,omitempty"`
}

StoredReplicaSettings holds per-replica settings persisted by UpdateGlobalTableSettings.

type StreamShard 

type StreamShard struct {
	// ShardID is the unique identifier for this shard.
	ShardID string
	// ParentShardID is the ID of the parent shard (empty for the first shard of a stream).
	ParentShardID string
	// StartingSequenceNum is the first sequence number in this shard (inclusive).
	StartingSequenceNum int64
	// EndingSequenceNum is the last sequence number in this shard (0 = still open).
	EndingSequenceNum int64
}

StreamShard models one DynamoDB Streams shard with its sequence range and genealogy.

type StreamsBackend 

type StreamsBackend interface {
	EnableStream(ctx context.Context, tableName, viewType string) error
	DisableStream(ctx context.Context, tableName string) error
	DescribeStream(
		ctx context.Context,
		input *dynamodbstreams.DescribeStreamInput,
	) (*dynamodbstreams.DescribeStreamOutput, error)
	GetShardIterator(
		ctx context.Context,
		input *dynamodbstreams.GetShardIteratorInput,
	) (*dynamodbstreams.GetShardIteratorOutput, error)
	GetRecords(
		ctx context.Context,
		input *dynamodbstreams.GetRecordsInput,
	) (*dynamodbstreams.GetRecordsOutput, error)
	ListStreams(
		ctx context.Context,
		input *dynamodbstreams.ListStreamsInput,
	) (*dynamodbstreams.ListStreamsOutput, error)
	GetRecentEvents(tableName string) []models.StreamRecord
}

StreamsBackend defines the interface for DynamoDB Streams operations.

type Table 

type Table struct {
	StreamCreatedAt  time.Time `json:"StreamCreatedAt"`
	CreationDateTime time.Time `json:"CreationDateTime"`

	Tags                   *tags.Tags                    `json:"Tags,omitempty"`
	AutoScaling            *autoScalingSettings          `json:"AutoScaling,omitempty"`
	OnDemandMaxWriteRRU    *int64                        `json:"OnDemandMaxWriteRRU,omitempty"`
	OnDemandMaxReadRRU     *int64                        `json:"OnDemandMaxReadRRU,omitempty"`
	ResourcePolicy         string                        `json:"ResourcePolicy,omitempty"`
	TTLAttribute           string                        `json:"TTLAttribute,omitempty"`
	StreamViewType         string                        `json:"StreamViewType,omitempty"`
	StreamARN              string                        `json:"StreamARN,omitempty"`
	GlobalTableName        string                        `json:"GlobalTableName,omitempty"`
	TableArn               string                        `json:"TableArn"`
	Status                 string                        `json:"Status"`
	TableID                string                        `json:"TableID"`
	SSEType                string                        `json:"SSEType,omitempty"`
	TableClass             string                        `json:"TableClass,omitempty"`
	BillingMode            string                        `json:"BillingMode,omitempty"`
	Name                   string                        `json:"Name"`
	SSEKMSMasterKeyArn     string                        `json:"SSEKMSMasterKeyArn,omitempty"`
	AttributeDefinitions   []models.AttributeDefinition  `json:"AttributeDefinitions"`
	GlobalSecondaryIndexes []models.GlobalSecondaryIndex `json:"GlobalSecondaryIndexes,omitempty"`
	Replicas               []models.ReplicaDescription   `json:"Replicas,omitempty"`
	LocalSecondaryIndexes  []models.LocalSecondaryIndex  `json:"LocalSecondaryIndexes,omitempty"`
	KeySchema              []models.KeySchemaElement     `json:"KeySchema"`
	KinesisDestinations    []KinesisDestinationEntry     `json:"KinesisDestinations,omitempty"`
	Items                  []map[string]any              `json:"Items"`

	StreamRecords         []models.StreamRecord                   `json:"StreamRecords,omitempty"`
	ProvisionedThroughput models.ProvisionedThroughputDescription `json:"ProvisionedThroughput"`

	StreamHead int `json:"StreamHead,omitempty"`

	PITREnabled                bool `json:"PITREnabled,omitempty"`
	SSEEnabled                 bool `json:"SSEEnabled,omitempty"`
	StreamsEnabled             bool `json:"StreamsEnabled"`
	DeletionProtectionEnabled  bool `json:"DeletionProtectionEnabled"`
	ContributorInsightsEnabled bool `json:"ContributorInsightsEnabled,omitempty"`
	// contains filtered or unexported fields
}

type TaggedTableInfo 

type TaggedTableInfo struct {
	Tags map[string]string
	ARN  string
}

TaggedTableInfo contains a DynamoDB table's ARN and tag snapshot. Used by the Resource Groups Tagging API cross-service listing.

type Throttler 

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

Throttler manages per-table token buckets for provisioned throughput enforcement. When disabled (enabled == false) all capacity checks pass unconditionally.

func NewThrottler 

func NewThrottler(enabled bool) *Throttler

NewThrottler creates a Throttler. When enabled is false all operations pass without any checks.

func (*Throttler) ConsumeRead 

func (t *Throttler) ConsumeRead(key string, units float64) error

ConsumeRead attempts to deduct units RCUs from the table bucket identified by key. Returns ProvisionedThroughputExceededException when the bucket is exhausted. Returns nil when throttling is disabled or no bucket exists for the key.

func (*Throttler) ConsumeWrite 

func (t *Throttler) ConsumeWrite(key string, units float64) error

ConsumeWrite attempts to deduct units WCUs from the table bucket identified by key. Returns ProvisionedThroughputExceededException when the bucket is exhausted. Returns nil when throttling is disabled or no bucket exists for the key.

func (*Throttler) DeleteTable 

func (t *Throttler) DeleteTable(key string)

DeleteTable removes the bucket for the given key.

func (*Throttler) SetTableCapacity 

func (t *Throttler) SetTableCapacity(key string, rcu, wcu int64)

SetTableCapacity registers or updates the provisioned capacity for the given key (typically "region:tableName"). Existing token counts are preserved on capacity increases so that UpdateTable cannot be used to instantly refill an exhausted bucket. On decreases the existing token count is capped to the new ceiling.

type WireDescribeStreamOutput added in v1.1.1 

type WireDescribeStreamOutput struct {
	StreamDescription *WireStreamDescription `json:"StreamDescription,omitempty"`
}

func ToWireDescribeStreamOutput added in v1.1.1 

func ToWireDescribeStreamOutput(out *dynamodbstreams.DescribeStreamOutput) *WireDescribeStreamOutput

type WireGetRecordsOutput added in v1.1.1 

type WireGetRecordsOutput struct {
	NextShardIterator *string            `json:"NextShardIterator,omitempty"`
	Records           []WireStreamRecord `json:"Records"`
}

func ToWireGetRecordsOutput added in v1.1.1 

func ToWireGetRecordsOutput(out *dynamodbstreams.GetRecordsOutput) (*WireGetRecordsOutput, error)

type WireStreamDescription added in v1.1.1 

type WireStreamDescription struct {
	CreationRequestDateTime *float64                        `json:"CreationRequestDateTime,omitempty"`
	LastEvaluatedShardID    *string                         `json:"LastEvaluatedShardId,omitempty"`
	StreamArn               *string                         `json:"StreamArn,omitempty"`
	StreamLabel             *string                         `json:"StreamLabel,omitempty"`
	TableName               *string                         `json:"TableName,omitempty"`
	StreamStatus            streamstypes.StreamStatus       `json:"StreamStatus,omitempty"`
	StreamViewType          streamstypes.StreamViewType     `json:"StreamViewType,omitempty"`
	KeySchema               []streamstypes.KeySchemaElement `json:"KeySchema,omitempty"`
	Shards                  []streamstypes.Shard            `json:"Shards,omitempty"`
}

WireStreamDescription mirrors StreamDescription but with timestamps as float64 epoch seconds.

type WireStreamRecord added in v1.1.1 

type WireStreamRecord struct {
	Dynamodb     *WireStreamRecordData      `json:"dynamodb,omitempty"`
	UserIdentity *streamstypes.Identity     `json:"userIdentity,omitempty"`
	EventID      string                     `json:"eventID,omitempty"`
	EventName    streamstypes.OperationType `json:"eventName,omitempty"`
	EventVersion string                     `json:"eventVersion,omitempty"`
	EventSource  string                     `json:"eventSource,omitempty"`
	AwsRegion    string                     `json:"awsRegion,omitempty"`
}

type WireStreamRecordData added in v1.1.1 

type WireStreamRecordData struct {
	// ApproximateCreationDateTime is Unix epoch seconds (float64) per DynamoDB Streams JSON 1.0 protocol.
	ApproximateCreationDateTime *float64                    `json:"ApproximateCreationDateTime,omitempty"`
	Keys                        map[string]any              `json:"Keys,omitempty"`
	NewImage                    map[string]any              `json:"NewImage,omitempty"`
	OldImage                    map[string]any              `json:"OldImage,omitempty"`
	SequenceNumber              *string                     `json:"SequenceNumber,omitempty"`
	SizeBytes                   *int64                      `json:"SizeBytes,omitempty"`
	StreamViewType              streamstypes.StreamViewType `json:"StreamViewType,omitempty"`
}

func ToWireStreamRecordData added in v1.1.1 

func ToWireStreamRecordData(record *streamstypes.StreamRecord) (*WireStreamRecordData, error)

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.