// ❌ Default VisibilityTimeout=30s — too short for slow processing
import {
  SQSClient,
  ReceiveMessageCommand,
  DeleteMessageCommand,
  ChangeMessageVisibilityCommand,
} from '@aws-sdk/client-sqs'

const sqs = new SQSClient({ region: 'us-east-1' })
const QUEUE_URL = process.env.FULFILLMENT_QUEUE_URL!

// ✅ Size VisibilityTimeout to p99 processing latency × 1.5
// If p99 processing = 30s → VisibilityTimeout = 45s
// Set at queue creation or per-receive via ReceiveMessage parameter
const { Messages } = await sqs.send(new ReceiveMessageCommand({
  QueueUrl:            QUEUE_URL,
  MaxNumberOfMessages: 10,
  VisibilityTimeout:   45,   // seconds — sized to p99 × 1.5
  WaitTimeSeconds:     20,   // long polling: wait up to 20s for messages
}))

// ✅ Extend visibility during long processing (heartbeat pattern)
async function processWithHeartbeat(
  receiptHandle: string,
  processFn:     () => Promise<void>,
  visibilityTimeout = 45,
): Promise<void> {
  const extensionInterval = Math.floor(visibilityTimeout * 0.6) * 1000  // 60% of timeout

  const heartbeat = setInterval(async () => {
    await sqs.send(new ChangeMessageVisibilityCommand({
      QueueUrl:          QUEUE_URL,
      ReceiptHandle:     receiptHandle,
      VisibilityTimeout: visibilityTimeout,  // reset the clock
    }))
  }, extensionInterval)

  try {
    await processFn()
    await sqs.send(new DeleteMessageCommand({
      QueueUrl:      QUEUE_URL,
      ReceiptHandle: receiptHandle,
    }))
  } finally {
    clearInterval(heartbeat)
  }
}

// ✅ SQS FIFO — MessageGroupId is the partition key
import { SendMessageCommand } from '@aws-sdk/client-sqs'

// All events for order-42 use the same MessageGroupId → strict ordering
await sqs.send(new SendMessageCommand({
  QueueUrl:               `${process.env.ORDERS_QUEUE_URL}`,
  MessageBody:            JSON.stringify({ orderId: 'ORD-42', event: 'created' }),
  MessageGroupId:         'ORD-42',     // same group = same consumer = strict order
  MessageDeduplicationId: `ORD-42-created-${Date.now()}`,  // 5-min dedup window
}))

// SQS FIFO deduplication: if two messages with the same MessageDeduplicationId
// are published within 5 minutes, SQS discards the duplicate silently
// — equivalent to Kafka's idempotent producer sequence number

// ❌ Don't use one MessageGroupId for all messages
// MessageGroupId='default' → single consumer → zero parallelism
// MessageGroupId=orderId   → one consumer per entity → ordered + parallel ✅

// ✅ Configure DLQ via RedrivePolicy — after maxReceiveCount failures, route to DLQ
import { CreateQueueCommand, SetQueueAttributesCommand } from '@aws-sdk/client-sqs'

// 1. Create the DLQ first
const dlqResult = await sqs.send(new CreateQueueCommand({
  QueueName: 'fulfillment-dlq',
  Attributes: { MessageRetentionPeriod: '1209600' }  // 14 days
}))
const dlqArn = (await sqs.send(new GetQueueAttributesCommand({
  QueueUrl:       dlqResult.QueueUrl!,
  AttributeNames: ['QueueArn'],
}))).Attributes!['QueueArn']

// 2. Attach DLQ to main queue
await sqs.send(new SetQueueAttributesCommand({
  QueueUrl: QUEUE_URL,
  Attributes: {
    RedrivePolicy: JSON.stringify({
      deadLetterTargetArn: dlqArn,
      maxReceiveCount:     '5',  // 5 delivery attempts before DLQ routing
    })
  }
}))
// VisibilityTimeout expiry counts as one receive attempt
// After 5 failed attempts: message routes to DLQ, main queue moves on

// ✅ SNS → multiple SQS subscribers (fan-out)
import { SNSClient, PublishCommand } from '@aws-sdk/client-sns'

const sns = new SNSClient({ region: 'us-east-1' })

// Publish once to SNS — all subscribed SQS queues receive a copy
await sns.send(new PublishCommand({
  TopicArn: process.env.ORDER_EVENTS_TOPIC_ARN!,
  Message:  JSON.stringify({
    orderId:    'ORD-42',
    event:      'order.created',
    totalCents: 9900,
    customerId: 'C-1',
  }),
  MessageAttributes: {
    'event.type': {
      DataType:    'String',
      StringValue: 'order.created',
    },
    'customer.tier': {
      DataType:    'String',
      StringValue: 'premium',
    }
  }
}))

// Three SQS queues subscribed to the same SNS topic:
// - fulfillment-queue  → processes all order.created events
// - analytics-queue    → receives all events (no filter)
// - premium-queue      → only receives order.created for premium customers (filter policy)

// SNS filter policy — applied at the subscription level (not the publisher)
// Only messages where event.type='order.created' AND customer.tier='premium'
// are delivered to the premium-sqs subscription

// Filter policy JSON (set via AWS console or CDK/Terraform):
const filterPolicy = {
  'event.type':    ['order.created'],
  'customer.tier': ['premium', 'vip'],  // OR condition within the array
}

// SDK: CreateSubscription with FilterPolicy attribute
import { SubscribeCommand } from '@aws-sdk/client-sns'

await sns.send(new SubscribeCommand({
  TopicArn: process.env.ORDER_EVENTS_TOPIC_ARN!,
  Protocol: 'sqs',
  Endpoint: process.env.PREMIUM_QUEUE_ARN!,
  Attributes: {
    FilterPolicy:           JSON.stringify(filterPolicy),
    FilterPolicyScope:      'MessageAttributes',  // or 'MessageBody' for payload-based filtering
    RawMessageDelivery:     'true',               // skip SNS envelope — SQS receives raw JSON
  }
}))

// ✅ EventBridge: publish a structured event to the default bus
import { EventBridgeClient, PutEventsCommand } from '@aws-sdk/client-eventbridge'

const eb = new EventBridgeClient({ region: 'us-east-1' })

await eb.send(new PutEventsCommand({
  Entries: [{
    EventBusName: 'default',
    Source:       'com.company.orders',
    DetailType:   'OrderCreated',
    Detail:       JSON.stringify({
      orderId:    'ORD-42',
      totalCents: 9900,
      customerId: 'C-1',
      tier:       'premium',
      region:     'eu-west-1',
    }),
    Time: new Date(),
  }]
}))

// EventBridge rule — routes premium EU orders to the priority fulfillment Lambda
// Rule pattern (content-based — inspects Detail fields, not just metadata)
{
  "source": ["com.company.orders"],
  "detail-type": ["OrderCreated"],
  "detail": {
    "tier": ["premium", "vip"],
    "totalCents": [{ "numeric": [">", 10000] }],
    "region": [{ "prefix": "eu-" }]
  }
}
// This is the managed equivalent of a RabbitMQ topic exchange binding:
//   routing_key: "order.created.premium.eu.*"
// Without requiring the producer to construct a routing key

// ✅ EventBridge input transformer — reshape the event before delivery to Lambda
// Strips unnecessary fields, reduces Lambda payload size
const inputTransformer = {
  inputPathsMap: {
    orderId:    '$.detail.orderId',
    totalCents: '$.detail.totalCents',
    customerId: '$.detail.customerId',
  },
  inputTemplate: JSON.stringify({
    orderId:    '<orderId>',
    totalCents: '<totalCents>',
    customerId: '<customerId>',
    source:     'eventbridge',
  })
}

// ✅ EventBridge archive — replay events after consumer bug fix
import { CreateArchiveCommand, StartReplayCommand } from '@aws-sdk/client-eventbridge'

// Create archive: retain all OrderCreated events for 90 days
await eb.send(new CreateArchiveCommand({
  ArchiveName:     'order-events-archive',
  EventSourceArn:  process.env.ORDER_BUS_ARN!,
  RetentionDays:   90,
  EventPattern:    JSON.stringify({
    source: ['com.company.orders'],
  }),
}))

// After fixing a consumer bug: replay events from a specific window
await eb.send(new StartReplayCommand({
  ReplayName:          'fulfillment-replay-2026-12-08',
  EventSourceArn:      process.env.ORDER_ARCHIVE_ARN!,
  EventStartTime:      new Date('2026-12-08T00:00:00Z'),
  EventEndTime:        new Date('2026-12-08T06:00:00Z'),
  Destination: {
    Arn: process.env.ORDER_BUS_ARN!,
    FilterArns: [process.env.FULFILLMENT_RULE_ARN!],
  }
}))
// EventBridge replays all archived events in the window to the specified rule target
// — the equivalent of Kafka consumer offset reset + replay, as a managed API call

# docker-compose.yml — LocalStack for local SQS/SNS/EventBridge development
version: '3.8'
services:
  localstack:
    image: localstack/localstack:3.4
    ports:
      - '4566:4566'
    environment:
      SERVICES:       sqs,sns,events,iam
      DEFAULT_REGION: us-east-1
      DEBUG:          '0'
    volumes:
      - ./scripts/localstack-init.sh:/etc/localstack/init/ready.d/init.sh

#!/bin/bash
# scripts/localstack-init.sh — create all queues and topics on LocalStack startup

awslocal sqs create-queue --queue-name fulfillment-queue \
  --attributes VisibilityTimeout=45,MessageRetentionPeriod=86400

awslocal sqs create-queue --queue-name fulfillment-dlq

awslocal sqs set-queue-attributes \
  --queue-url http://localhost:4566/000000000000/fulfillment-queue \
  --attributes '{"RedrivePolicy":"{\"deadLetterTargetArn\":\"arn:aws:sqs:us-east-1:000000000000:fulfillment-dlq\",\"maxReceiveCount\":\"5\"}"}'

awslocal sns create-topic --name order-events

awslocal sns subscribe \
  --topic-arn arn:aws:sns:us-east-1:000000000000:order-events \
  --protocol sqs \
  --notification-endpoint arn:aws:sqs:us-east-1:000000000000:fulfillment-queue \
  --attributes '{"RawMessageDelivery":"true"}'

echo "✅ LocalStack queues and topics initialized"

// ✅ SDK client configured for LocalStack in development
const isLocal = process.env.AWS_ENV === 'local'

const sqs = new SQSClient({
  region:   'us-east-1',
  endpoint: isLocal ? 'http://localhost:4566' : undefined,
  credentials: isLocal
    ? { accessKeyId: 'test', secretAccessKey: 'test' }
    : undefined,   // production: uses IAM role via ECS task role
})