函数模板概述

• 平台类型：AWS Lambda（SQS 触发，S3 归档，Redis 做幂等）
• 运行时环境：Python 3.9
• 触发类型：消息队列（SQS，支持批处理与并发度配置，DLQ）

核心代码

# app.py
# -*- coding: utf-8 -*-
import json
import os
import time
import uuid
import logging
import traceback
from datetime import datetime, timezone
from typing import Any, Dict, List

import boto3
from botocore.exceptions import ClientError

try:
    import redis  # pip install redis
except Exception:
    redis = None  # 允许本地/冷启动缺依赖，部署时请确保打包依赖

# -------------------------
# 环境变量
# -------------------------
ARCHIVE_BUCKET = os.getenv("ARCHIVE_BUCKET", "")
REDIS_URL = os.getenv("REDIS_URL", "")
ALERT_WEBHOOK = os.getenv("ALERT_WEBHOOK", "")  # 可选
# 幂等已处理标记 TTL（秒）
PROCESSED_TTL_SECONDS = int(os.getenv("PROCESSED_TTL_SECONDS", "604800"))  # 默认7天
# 处理锁 TTL（秒）
LOCK_TTL_SECONDS = int(os.getenv("LOCK_TTL_SECONDS", "300"))  # 5分钟
# CloudWatch 指标命名空间
METRIC_NAMESPACE = os.getenv("METRIC_NAMESPACE", "OrderProcessor")
SERVICE_NAME = os.getenv("SERVICE_NAME", "order-created-consumer")

# -------------------------
# 全局客户端（复用连接）
# -------------------------
s3 = boto3.client("s3")
_lambda = boto3.client("lambda")  # 仅用于获取函数名等信息（可选）
_redis = None

logger = logging.getLogger()
logger.setLevel(logging.INFO)


# -------------------------
# 异常分类
# -------------------------
class TransientError(Exception):
    """可重试异常（网络波动、服务限流/暂时不可用等）"""
    pass


class PermanentError(Exception):
    """不可重试异常（数据校验失败、业务规则不满足等）"""
    pass


# -------------------------
# 工具函数
# -------------------------
def now_utc_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def parse_iso_date(s: str) -> datetime:
    # 尽量兼容 "2024-01-01T00:00:00Z" / "+00:00"
    if s.endswith("Z"):
        s = s[:-1] + "+00:00"
    return datetime.fromisoformat(s)


def get_redis() -> "redis.Redis":
    global _redis
    if _redis is None:
        if not REDIS_URL:
            raise PermanentError("Missing REDIS_URL env")
        if redis is None:
            raise PermanentError("redis library not available; bundle dependency")
        _redis = redis.Redis.from_url(
            REDIS_URL,
            socket_timeout=2,
            socket_connect_timeout=2,
            health_check_interval=30,
            decode_responses=True,
        )
    return _redis


def is_retryable_aws_error(err: ClientError) -> bool:
    code = err.response.get("Error", {}).get("Code", "")
    # 常见可重试错误码
    retryable = {
        "RequestTimeout", "Throttling", "ThrottlingException", "SlowDown",
        "InternalError", "InternalServerError", "ServiceUnavailable",
        "ProvisionedThroughputExceededException"
    }
    if code in retryable:
        return True
    # 5xx 一般可重试
    try:
        return code.isdigit() and int(code) >= 500
    except Exception:
        return False


def safe_json_dumps(obj: Any) -> str:
    return json.dumps(obj, ensure_ascii=False, separators=(",", ":"))


def emit_emf_metrics(dimensions: Dict[str, str], metrics: Dict[str, float]):
    # CloudWatch Embedded Metric Format（EMF）
    # 参考：https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format.html
    metric_names = list(metrics.keys())
    emf = {
        "_aws": {
            "Timestamp": int(time.time() * 1000),
            "CloudWatchMetrics": [
                {
                    "Namespace": METRIC_NAMESPACE,
                    "Dimensions": [list(dimensions.keys())],
                    "Metrics": [{"Name": n, "Unit": "Count" if n.endswith("_count") else "Milliseconds"} for n in metric_names],
                }
            ],
        }
    }
    emf.update(dimensions)
    emf.update(metrics)
    print(safe_json_dumps(emf))


def send_alert(title: str, message: str, context: Dict[str, Any] = None):
    if not ALERT_WEBHOOK:
        return
    try:
        import urllib.request
        import urllib.error

        payload = {
            "title": title,
            "message": message,
            "service": SERVICE_NAME,
            "time": now_utc_iso(),
            "context": context or {},
        }
        data = safe_json_dumps(payload).encode("utf-8")
        req = urllib.request.Request(
            ALERT_WEBHOOK,
            data=data,
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        with urllib.request.urlopen(req, timeout=2) as resp:
            resp.read()
    except Exception as e:
        logger.warning("alert_failed: %s", str(e))


def get_order_keys(order_id: str):
    return f"order:{order_id}:done", f"order:{order_id}:lock"


def acquire_lock(r: "redis.Redis", lock_key: str, token: str, ttl_sec: int) -> bool:
    # SET lock_key token NX EX ttl
    return bool(r.set(lock_key, token, nx=True, ex=ttl_sec))


def release_lock_if_owner(r: "redis.Redis", lock_key: str, token: str):
    # 原子释放锁（仅持有者可释放）
    lua = """
    if redis.call("GET", KEYS[1]) == ARGV[1] then
      return redis.call("DEL", KEYS[1])
    else
      return 0
    end
    """
    try:
        r.eval(lua, 1, lock_key, token)
    except Exception:
        pass


def parse_record_body(record: Dict[str, Any]) -> Dict[str, Any]:
    """
    支持:
    - 直接 SQS JSON: body = {"orderId": "...", ...}
    - SNS->SQS: body = {"Type":"Notification", "Message":"{...json...}"}
    """
    raw_body = record.get("body", "")
    try:
        body = json.loads(raw_body)
    except Exception:
        raise PermanentError("Message body is not valid JSON")

    # SNS wrap
    if isinstance(body, dict) and "Message" in body and "Type" in body:
        try:
            msg = json.loads(body["Message"])
        except Exception:
            raise PermanentError("SNS Message is not valid JSON")
        return msg
    elif isinstance(body, dict):
        return body
    else:
        raise PermanentError("Unsupported message body structure")


def validate_message(msg: Dict[str, Any]):
    required = ["orderId", "userId", "amount", "createdAt"]
    missing = [k for k in required if k not in msg]
    if missing:
        raise PermanentError(f"Missing fields: {missing}")

    # 基础类型与值校验
    if not isinstance(msg["orderId"], str) or not msg["orderId"]:
        raise PermanentError("orderId invalid")
    if not isinstance(msg["userId"], str) or not msg["userId"]:
        raise PermanentError("userId invalid")
    try:
        float(msg["amount"])
    except Exception:
        raise PermanentError("amount invalid")
    try:
        parse_iso_date(msg["createdAt"])
    except Exception:
        raise PermanentError("createdAt invalid ISO timestamp")


def archive_order_to_s3(msg: Dict[str, Any], s3_client, bucket: str, source_id: str):
    created_at = parse_iso_date(msg["createdAt"])
    y = created_at.strftime("%Y")
    m = created_at.strftime("%m")
    d = created_at.strftime("%d")
    order_id = msg["orderId"]

    # 归档 JSON
    archive_doc = {
        "orderId": msg["orderId"],
        "userId": msg["userId"],
        "amount": float(msg["amount"]),
        "createdAt": msg["createdAt"],
        "archivedAt": now_utc_iso(),
        "sourceMessageId": source_id,
        "status": "archived",
        "topic": "order.created",
        "version": 1,
    }

    key = f"order-created/y={y}/m={m}/d={d}/{order_id}.json"
    try:
        s3_client.put_object(
            Bucket=bucket,
            Key=key,
            Body=safe_json_dumps(archive_doc).encode("utf-8"),
            ContentType="application/json",
            CacheControl="no-store",
        )
    except ClientError as e:
        if is_retryable_aws_error(e):
            raise TransientError(f"S3 retryable error: {e}")
        raise PermanentError(f"S3 permanent error: {e}")
    return key


def process_one_record(record: Dict[str, Any], context) -> str:
    """
    返回状态:
    - "ok"：成功归档或已处理
    - "skip"：不可重试（已告警/隔离），当做成功确认消费
    - 抛出 TransientError：局部失败，触发重试
    """
    if not ARCHIVE_BUCKET:
        raise TransientError("ARCHIVE_BUCKET not configured")

    # 解析与校验
    msg = parse_record_body(record)
    validate_message(msg)

    order_id = msg["orderId"]
    message_id = record.get("messageId", str(uuid.uuid4()))
    approx_receive = int(record.get("attributes", {}).get("ApproximateReceiveCount", "1"))

    r = get_redis()
    processed_key, lock_key = get_order_keys(order_id)
    # 幂等：若已处理，直接返回成功
    if r.exists(processed_key):
        logger.info(safe_json_dumps({
            "event": "idempotent_hit",
            "orderId": order_id,
            "messageId": message_id
        }))
        return "ok"

    # 争抢处理锁，避免并发重复处理
    token = context.aws_request_id if getattr(context, "aws_request_id", None) else str(uuid.uuid4())
    if not acquire_lock(r, lock_key, token, LOCK_TTL_SECONDS):
        # 另一并发实例正在处理，返回可重试让其完成后再确认
        raise TransientError(f"processing_in_progress orderId={order_id}")

    try:
        # 归档到 S3
        key = archive_order_to_s3(msg, s3, ARCHIVE_BUCKET, message_id)

        # 标记已处理（幂等）
        r.set(processed_key, now_utc_iso(), ex=PROCESSED_TTL_SECONDS)

        logger.info(safe_json_dumps({
            "event": "archived",
            "orderId": order_id,
            "s3key": key,
            "messageId": message_id
        }))
        return "ok"
    except PermanentError as pe:
        # 不可重试：记录到错误前缀并告警，避免无谓重试
        try:
            quarantine_key = f"errors/order-created/{order_id}-{message_id}.json"
            s3.put_object(
                Bucket=ARCHIVE_BUCKET,
                Key=quarantine_key,
                Body=safe_json_dumps({
                    "error": str(pe),
                    "message": msg,
                    "archivedAt": now_utc_iso(),
                    "receiveCount": approx_receive,
                    "messageId": message_id
                }).encode("utf-8"),
                ContentType="application/json",
                CacheControl="no-store",
            )
        except Exception:
            # 隔离失败不影响主流程
            pass

        send_alert(
            title="[PermanentError] order.created",
            message=str(pe),
            context={"orderId": order_id, "messageId": message_id, "receiveCount": approx_receive},
        )
        logger.error(safe_json_dumps({
            "event": "permanent_error",
            "orderId": order_id,
            "messageId": message_id,
            "error": str(pe)
        }))
        # 不重试
        return "skip"
    finally:
        release_lock_if_owner(r, lock_key, token)


def handler(event, context):
    """
    SQS 触发的批处理 Lambda。
    - 支持局部失败（Partial Batch Response），仅失败项重试
    - 记录批处理耗时与计数指标（EMF）
    """
    batch_start = time.perf_counter()
    batch_received = len(event.get("Records", []))
    success_count = 0
    skip_count = 0
    transient_fail_count = 0
    permanent_fail_count = 0  # 被判定为永久失败但已跳过重试
    item_failures: List[Dict[str, str]] = []

    for record in event.get("Records", []):
        per_start = time.perf_counter()
        mid = record.get("messageId", "unknown")
        try:
            status = process_one_record(record, context)
            if status == "ok":
                success_count += 1
            elif status == "skip":
                skip_count += 1
        except TransientError as te:
            transient_fail_count += 1
            item_failures.append({"itemIdentifier": mid})
            logger.warning(safe_json_dumps({
                "event": "transient_error",
                "messageId": mid,
                "error": str(te)
            }))
            # 当即将进入 DLQ（如已达阈值）时发送告警
            try:
                approx_receive = int(record.get("attributes", {}).get("ApproximateReceiveCount", "1"))
            except Exception:
                approx_receive = 1
            if approx_receive >= 3:
                send_alert(
                    title="[RetryExhausted] order.created",
                    message=str(te),
                    context={"messageId": mid, "receiveCount": approx_receive},
                )
        except PermanentError as pe:
            permanent_fail_count += 1
            # 跳过重试，已在内部记录与隔离
        except Exception as e:
            # 未分类异常按可重试处理
            transient_fail_count += 1
            item_failures.append({"itemIdentifier": mid})
            logger.error(safe_json_dumps({
                "event": "unexpected_error",
                "messageId": mid,
                "error": str(e),
                "trace": traceback.format_exc()
            }))
        finally:
            per_dur_ms = (time.perf_counter() - per_start) * 1000.0
            print(safe_json_dumps({
                "event": "item_processed",
                "messageId": mid,
                "duration_ms": round(per_dur_ms, 2)
            }))

    batch_dur_ms = (time.perf_counter() - batch_start) * 1000.0

    # 输出 EMF 指标
    dims = {
        "service": SERVICE_NAME,
        "function": os.getenv("AWS_LAMBDA_FUNCTION_NAME", "unknown"),
        "stage": os.getenv("STAGE", "prod"),
    }
    emit_emf_metrics(
        dimensions=dims,
        metrics={
            "received_count": float(batch_received),
            "success_count": float(success_count),
            "skip_count": float(skip_count),
            "transient_fail_count": float(transient_fail_count),
            "permanent_fail_count": float(permanent_fail_count),
            "batch_duration_ms": round(batch_dur_ms, 2),
        },
    )

    # Partial Batch Response：仅重试失败项
    return {"batchItemFailures": item_failures}

requirements.txt

boto3==1.35.0
redis==4.6.0

配置说明

使用 AWS SAM 示例（可根据需要改为 CDK 或 Serverless Framework）：

# template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Order Created Consumer - SQS -> Lambda -> S3 with Redis Idempotency

Parameters:
  ArchiveBucketName:
    Type: String
    Description: S3 bucket name for archiving order summaries
  RedisUrl:
    Type: String
    Description: Redis connection URL (e.g., rediss://user:pass@host:6379/0)
  AlertWebhook:
    Type: String
    Default: ""
    Description: Optional alert webhook URL
  BatchSize:
    Type: Number
    Default: 10
    MinValue: 1
    MaxValue: 10
  MaxConcurrency:
    Type: Number
    Default: 10
    MinValue: 2
    MaxValue: 100
  ProcessedTtlSeconds:
    Type: Number
    Default: 604800
  LockTtlSeconds:
    Type: Number
    Default: 300

Resources:
  OrderCreatedDLQ:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: order-created-dlq

  OrderCreatedQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: order-created
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt OrderCreatedDLQ.Arn
        maxReceiveCount: 3  # 失败重试最多3次，超出进入DLQ

  OrderCreatedConsumer:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: order-created-consumer
      Runtime: python3.9
      Handler: app.handler
      Timeout: 30
      MemorySize: 512
      CodeUri: ./
      Environment:
        Variables:
          ARCHIVE_BUCKET: !Ref ArchiveBucketName
          REDIS_URL: !Ref RedisUrl
          ALERT_WEBHOOK: !Ref AlertWebhook
          METRIC_NAMESPACE: OrderPipeline
          SERVICE_NAME: order-created-consumer
          PROCESSED_TTL_SECONDS: !Ref ProcessedTtlSeconds
          LOCK_TTL_SECONDS: !Ref LockTtlSeconds
      Policies:
        - S3WritePolicy:
            BucketName: !Ref ArchiveBucketName
        - CloudWatchLogsFullAccess
      Events:
        SQSTrigger:
          Type: SQS
          Properties:
            Queue: !GetAtt OrderCreatedQueue.Arn
            BatchSize: !Ref BatchSize
            FunctionResponseTypes:
              - ReportBatchItemFailures
            ScalingConfig:
              MaximumConcurrency: !Ref MaxConcurrency
      # 如需访问 VPC 内的 Redis，请在此配置 VpcConfig（子网、安组）
      # VpcConfig:
      #   SecurityGroupIds: [sg-xxxxxxxx]
      #   SubnetIds: [subnet-aaaa, subnet-bbbb]

Outputs:
  QueueUrl:
    Value: !Ref OrderCreatedQueue
  DLQUrl:
    Value: !Ref OrderCreatedDLQ

说明：

• 将外部事件源（如 SNS、Kafka、EventBridge）通过订阅/连接推送至 SQS 的 order-created 队列；Lambda 直接由 SQS 触发，从而具备批处理、并发度控制、DLQ 重投等能力。
• RedrivePolicy 配置了 maxReceiveCount=3，与“失败重试最多3次，仍失败路由至 DLQ”一致。
• FunctionResponseTypes=ReportBatchItemFailures 启用局部失败返回，保证仅失败消息重试。

使用指南

• 打包与部署
  • 将 app.py 与 requirements.txt 放入同一目录，使用 SAM 打包部署：
    • sam build --use-container
    • sam deploy --guided
  • 确保部署时提供参数 ArchiveBucketName、RedisUrl（建议使用 Secrets Manager + Lambda 环境变量引用，不要硬编码凭据）。
• 消息格式
  • 直接 SQS 消息 body 或 SNS->SQS 包装均可，最终消息 JSON 必须包含：
    • orderId: string
    • userId: string
    • amount: number
    • createdAt: ISO 时间戳（例如 2024-01-01T00:00:00Z）
• 幂等处理
  • 以 orderId 为粒度，Redis 通过 processed_key 标记已处理，TTL 默认 7 天可配。
  • 并发争抢使用短期锁 lock_key；若有并发，当前消息将标记为可重试，避免重复归档。
• 归档策略
  • 成功将订单摘要以 JSON 写入 S3：order-created/y=YYYY/m=MM/d=DD/{orderId}.json
  • 不可重试（PermanentError）消息将隔离到 errors/order-created/ 前缀，并发送告警。
• 失败与重试
  • 可重试（TransientError）返回局部失败，仅该消息重试；超过 3 次将由 SQS 自动投递至 DLQ。
  • 不可重试错误不再重试（函数内“skip”处理），避免冲击系统；已记录到隔离区并告警。
• 指标与日志
  • 每个批次输出 EMF 指标：received_count、success_count、skip_count、transient_fail_count、permanent_fail_count、batch_duration_ms。
  • 每条消息输出处理耗时（item_processed）。
  • 可在 CloudWatch Logs 中基于 EMF 自动生成指标图表和报警。

最佳实践建议

• 安全与合规
  • 不在代码中硬编码凭据；使用 Secrets Manager 或 SSM Parameter Store 注入 Redis 凭据，结合 KMS 加密。
  • 若 Redis 在 VPC 内，配置最小权限安全组与子网隔离。
  • S3 桶开启默认加密（SSE-KMS），并配置最小权限的 IAM Policy。
• 稳定性与性能
  • 合理设置 BatchSize（1~10）与 MaximumConcurrency，确保下游（S3/Redis）不被压垮；可结合 SQS 可见性超时进行容量规划。
  • 使用 Partial Batch Response 降低“全批失败”导致的放大重试。
  • 根据业务重复率调优 PROCESSED_TTL_SECONDS，平衡幂等缓存命中率与存储成本。
  • 将 Lambda 超时设置为略高于单批最大预期处理耗时。
• 监控与告警
  • 基于 EMF 指标创建 CloudWatch Alarm：高 transient_fail_count、permanent_fail_count、DLQ 消息积压。
  • 对 DLQ 配置可观测管道（订阅 Lambda/告警）以便及时处置“坏消息”。
• 可靠性
  • SQS RedrivePolicy 设置 maxReceiveCount=3 与需求一致；如存在上游抖动，建议从小到大逐步调优。
  • 对 S3 的 5xx/SlowDown 等错误进行重试分类（已内置），必要时可加入指数退避。
• 可维护性
  • 将 schema 校验独立为模块，复杂校验可引入 pydantic/fastjsonschema（注意打包体积）。
  • 按需扩展归档对象字段与版本号，保持向后兼容。
  • 使用 CI/CD（如 GitHub Actions + SAM）进行自动化测试与部署。

如需改为 Azure Functions（Service Bus）或 GCP（Pub/Sub + Cloud Storage），我可在保持相同业务语义下输出对应平台模板。

函数模板概述

• 平台类型：AWS Lambda（S3 事件触发）
• 运行时环境：go1.x
• 触发类型：对象存储（Amazon S3：ObjectCreated*，前缀 images/，后缀 jpg/jpeg/png）

核心代码

// 文件：main.go
// 功能：当 S3 bucket 的 images/ 前缀有 JPG/PNG 新文件上传时，生成 200x200 居中裁剪的 JPEG 缩略图（质量80），写回到 OUTPUT_PREFIX（默认 thumbnails/）同名路径；
// - 若目标已存在则跳过
// - 对非图片或超过 MAX_SIZE_MB 的文件记录告警并结束
// - 输出结构化日志和耗时指标
// 环境变量：
//   OUTPUT_PREFIX  (默认 "thumbnails/")
//   MAX_SIZE_MB    (默认 10)
//   CONCURRENCY    (默认 5)

package main

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"image"
	"image/jpeg"
	_ "image/png" // 注册 PNG 解码
	"io"
	"net/url"
	"os"
	"path"
	"strings"
	"sync"
	"time"

	"github.com/aws/aws-lambda-go/events"
	"github.com/aws/aws-lambda-go/lambdacontext"
	"github.com/aws/aws-lambda-go/lambda"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	s3 "github.com/aws/aws-sdk-go-v2/service/s3"
	s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
	smithy "github.com/aws/smithy-go"
	smithyhttp "github.com/aws/smithy-go/transport/http"

	"github.com/disintegration/imaging"
	"github.com/rwcarlsen/goexif/exif"
)

const (
	sourcePrefix = "images/"
	thumbSize    = 200
	jpgQuality   = 80
)

type AppConfig struct {
	OutputPrefix string
	MaxSizeBytes int64
	Concurrency  int
}

var (
	s3cl  *s3.Client
	appcf AppConfig
)

func init() {
	// 初始化 SDK 客户端（复用连接）
	ctx := context.Background()
	cfg, err := config.LoadDefaultConfig(ctx)
	if err != nil {
		panic(fmt.Errorf("load aws config: %w", err))
	}
	s3cl = s3.NewFromConfig(cfg)
	appcf = loadConfig()
}

func main() {
	lambda.Start(handler)
}

func handler(ctx context.Context, evt events.S3Event) error {
	start := time.Now()
	reqID := requestIDFromCtx(ctx)

	logJSON("info", "invocation_start", map[string]any{
		"request_id": reqID,
		"records":    len(evt.Records),
		"config": map[string]any{
			"output_prefix": appcf.OutputPrefix,
			"max_size_bytes": appcf.MaxSizeBytes,
			"concurrency":    appcf.Concurrency,
		},
	})

	sem := make(chan struct{}, appcf.Concurrency)
	var wg sync.WaitGroup
	errs := make(chan error, len(evt.Records))

	for _, r := range evt.Records {
		rec := r // 避免闭包变量捕获问题
		wg.Add(1)
		sem <- struct{}{}
		go func() {
			defer func() {
				<-sem
				wg.Done()
			}()
			if err := processRecord(ctx, rec); err != nil {
				errs <- err
			}
		}()
	}

	wg.Wait()
	close(errs)

	var firstErr error
	for e := range errs {
		if firstErr == nil {
			firstErr = e
		}
		// 继续收集但仅返回第一个错误（可按需修改为汇总）
	}

	logJSON("info", "invocation_end", map[string]any{
		"request_id": reqID,
		"duration_ms": time.Since(start).Milliseconds(),
		"error":       errToStr(firstErr),
	})
	return firstErr
}

func processRecord(ctx context.Context, rec events.S3EventRecord) error {
	start := time.Now()

	bucket := rec.S3.Bucket.Name
	keyEnc := rec.S3.Object.Key
	key, _ := url.QueryUnescape(keyEnc)
	key = strings.ReplaceAll(key, "+", " ") // 适配某些编码形式

	ext := strings.ToLower(path.Ext(key))
	isAllowed := ext == ".jpg" || ext == ".jpeg" || ext == ".png"

	// 仅处理 images/ 前缀
	if !strings.HasPrefix(key, sourcePrefix) {
		logJSON("debug", "skip_non_matching_prefix", map[string]any{
			"bucket": bucket, "key": key,
		})
		return nil
	}
	if !isAllowed {
		logJSON("warn", "skip_non_image_extension", map[string]any{
			"bucket": bucket, "key": key, "ext": ext,
		})
		return nil
	}

	// 目标 Key 计算
	suffix := strings.TrimPrefix(key, sourcePrefix)
	dstKey := path.Clean(appcf.OutputPrefix + suffix)

	// 若目标已存在则跳过
	exists, err := objectExists(ctx, bucket, dstKey)
	if err != nil {
		logJSON("error", "head_destination_failed", map[string]any{
			"bucket": bucket, "key": dstKey, "error": err.Error(),
		})
		return err
	}
	if exists {
		logJSON("info", "skip_already_exists", map[string]any{
			"bucket": bucket, "src_key": key, "dst_key": dstKey,
		})
		return nil
	}

	// 源对象大小检查（先 HEAD 再 GET，避免读大对象）
	head, err := s3cl.HeadObject(ctx, &s3.HeadObjectInput{
		Bucket: aws.String(bucket),
		Key:    aws.String(key),
	})
	if err != nil {
		if isNotFound(err) {
			logJSON("warn", "source_not_found", map[string]any{
				"bucket": bucket, "key": key,
			})
			return nil
		}
		return fmt.Errorf("head source: %w", err)
	}
	srcSize := int64(0)
	if head.ContentLength != nil {
		srcSize = *head.ContentLength
	}
	if appcf.MaxSizeBytes > 0 && srcSize > appcf.MaxSizeBytes {
		logJSON("warn", "skip_exceed_max_size", map[string]any{
			"bucket": bucket, "key": key, "size_bytes": srcSize, "max_size_bytes": appcf.MaxSizeBytes,
		})
		return nil
	}

	// 读取对象（再限流一次，防止 HEAD 与实际不一致）
	get, err := s3cl.GetObject(ctx, &s3.GetObjectInput{
		Bucket: aws.String(bucket),
		Key:    aws.String(key),
	})
	if err != nil {
		if isNotFound(err) {
			logJSON("warn", "source_not_found_get", map[string]any{
				"bucket": bucket, "key": key,
			})
			return nil
		}
		return fmt.Errorf("get source: %w", err)
	}
	defer get.Body.Close()

	maxBytes := appcf.MaxSizeBytes
	if maxBytes <= 0 {
		maxBytes = 10 << 20 // 安全兜底：10MB
	}
	var buf bytes.Buffer
	lr := &io.LimitedReader{R: get.Body, N: maxBytes + 1}
	n, err := io.Copy(&buf, lr)
	if err != nil {
		return fmt.Errorf("read source: %w", err)
	}
	if n > maxBytes {
		logJSON("warn", "skip_exceed_max_size_stream", map[string]any{
			"bucket": bucket, "key": key, "read_bytes": n, "max_size_bytes": maxBytes,
		})
		return nil
	}

	// 解码并尊重 EXIF 方向
	// 说明：对 JPEG 读取 EXIF Orientation 并将图像内容转换为正确方向（最终 JPEG 不保证完整保留所有 EXIF，避免磁盘/内存开销）
	srcImg, format, err := image.Decode(bytes.NewReader(buf.Bytes()))
	if err != nil {
		logJSON("warn", "decode_failed_not_image", map[string]any{
			"bucket": bucket, "key": key, "error": err.Error(),
		})
		return nil // 非图片：记录告警并结束
	}

	if format == "jpeg" {
		if ex, err := exif.Decode(bytes.NewReader(buf.Bytes())); err == nil && ex != nil {
			srcImg = applyEXIFOrientation(srcImg, ex)
		}
	}

	// 生成 200x200 居中裁剪缩略图（Lanczos，质量80）
	thumb := imaging.Fill(srcImg, thumbSize, thumbSize, imaging.Center, imaging.Lanczos)

	var out bytes.Buffer
	if err := jpeg.Encode(&out, thumb, &jpeg.Options{Quality: jpgQuality}); err != nil {
		return fmt.Errorf("encode jpeg: %w", err)
	}

	// 写回 S3（ContentType: image/jpeg）
	_, err = s3cl.PutObject(ctx, &s3.PutObjectInput{
		Bucket:      aws.String(bucket),
		Key:         aws.String(dstKey),
		Body:        bytes.NewReader(out.Bytes()),
		ContentType: aws.String("image/jpeg"),
		// 如需加密可启用（示例：SSE-S3）
		// ServerSideEncryption: s3types.ServerSideEncryptionAes256,
		Metadata: map[string]string{
			"x-source-key": key,
		},
	})
	if err != nil {
		return fmt.Errorf("put destination: %w", err)
	}

	logJSON("info", "thumb_generated", map[string]any{
		"bucket":        bucket,
		"src_key":       key,
		"dst_key":       dstKey,
		"src_bytes":     n,
		"dst_bytes":     out.Len(),
		"duration_ms":   time.Since(start).Milliseconds(),
		"format":        format,
		"jpeg_quality":  jpgQuality,
		"thumb_size_px": fmt.Sprintf("%dx%d", thumbSize, thumbSize),
	})

	return nil
}

func objectExists(ctx context.Context, bucket, key string) (bool, error) {
	_, err := s3cl.HeadObject(ctx, &s3.HeadObjectInput{
		Bucket: aws.String(bucket),
		Key:    aws.String(key),
	})
	if err == nil {
		return true, nil
	}
	if isNotFound(err) {
		return false, nil
	}
	return false, err
}

func isNotFound(err error) bool {
	// 兼容多种 404/NoSuchKey
	var re *smithyhttp.ResponseError
	if errors.As(err, &re) {
		if re.HTTPStatusCode() == 404 {
			return true
		}
	}
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) {
		code := apiErr.ErrorCode()
		if code == "NotFound" || code == "NoSuchKey" {
			return true
		}
	}
	var nfe *s3types.NotFound
	if errors.As(err, &nfe) {
		return true
	}
	return false
}

func loadConfig() AppConfig {
	maxMB := parseInt64(getEnv("MAX_SIZE_MB", "10"))
	con := parseInt(getEnv("CONCURRENCY", "5"))
	prefix := getEnv("OUTPUT_PREFIX", "thumbnails/")

	// 规范化前缀，确保不以 "/" 开头，以 "/" 结尾
	prefix = strings.TrimPrefix(prefix, "/")
	if !strings.HasSuffix(prefix, "/") {
		prefix += "/"
	}
	if prefix == sourcePrefix {
		// 防止回写触发环路
		prefix = "thumbnails/"
	}

	return AppConfig{
		OutputPrefix: prefix,
		MaxSizeBytes: maxMB << 20,
		Concurrency:  con,
	}
}

func getEnv(k, def string) string {
	v := strings.TrimSpace(os.Getenv(k))
	if v == "" {
		return def
	}
	return v
}
func parseInt(s string) int {
	n := 0
	fmt.Sscanf(s, "%d", &n)
	if n <= 0 {
		n = 1
	}
	return n
}
func parseInt64(s string) int64 {
	var n int64
	fmt.Sscanf(s, "%d", &n)
	if n <= 0 {
		n = 1
	}
	return n
}

func applyEXIFOrientation(img image.Image, ex *exif.Exif) image.Image {
	tag, err := ex.Get(exif.Orientation)
	if err != nil || tag == nil {
		return img
	}
	orient, err := tag.Int(0)
	if err != nil {
		return img
	}
	switch orient {
	case 1:
		// 正常
		return img
	case 2:
		// 水平镜像
		return imaging.FlipH(img)
	case 3:
		// 180
		return imaging.Rotate180(img)
	case 4:
		// 垂直镜像
		return imaging.FlipV(img)
	case 5:
		// 水平镜像 + 270 CW
		return imaging.Rotate270(imaging.FlipH(img))
	case 6:
		// 90 CW
		return imaging.Rotate270(imaging.Rotate180(img)) // 等效为 Rotate90，但 imaging 没有直接 Rotate90? 有，使用 Rotate90 更直观：
		// return imaging.Rotate90(img)
	case 7:
		// 水平镜像 + 90 CW
		return imaging.Rotate90(imaging.FlipH(img))
	case 8:
		// 270 CW
		return imaging.Rotate270(img)
	default:
		return img
	}
}

func requestIDFromCtx(ctx context.Context) string {
	if lc, ok := lambdacontext.FromContext(ctx); ok && lc != nil {
		return lc.AwsRequestID
	}
	return ""
}

func errToStr(e error) string {
	if e == nil {
		return ""
	}
	return e.Error()
}

func logJSON(level, msg string, fields map[string]any) {
	if fields == nil {
		fields = map[string]any{}
	}
	fields["level"] = level
	fields["msg"] = msg
	fields["ts"] = time.Now().Format(time.RFC3339Nano)
	b, _ := json.Marshal(fields)
	fmt.Println(string(b))
}

备注：

• 上面 applyEXIFOrientation 中对 90/270 度可直接使用 imaging.Rotate90/Rotate270。若使用 Rotate90 更直观，可将 case 6 改为 imaging.Rotate90(img)。

配置说明

• 触发配置（S3 -> Lambda 事件通知）：
  • 事件类型：ObjectCreated:Put, Post, Copy, CompleteMultipartUpload
  • 前缀过滤：images/
  • 后缀过滤（建议）：.jpg, .jpeg, .png
• 环境变量：
  • OUTPUT_PREFIX：缩略图输出前缀（默认 thumbnails/，将写入 thumbnails/<原相对路径>）
  • MAX_SIZE_MB：允许处理的源文件最大大小（默认 10MB）
  • CONCURRENCY：单次事件内并发处理的记录数量（默认 5）
• IAM 权限（最小权限示例，替换为实际 bucket 名称）：
  • s3:GetObject, s3:HeadObject 对 arn:aws:s3:::YOUR_BUCKET/images/*
  • s3:PutObject, s3:HeadObject 对 arn:aws:s3:::YOUR_BUCKET/thumbnails/*
  • logs:CreateLogGroup, logs:CreateLogStream, logs:PutLogEvents（CloudWatch Logs）
• 部署构建（Go 1.x runtime）：
  • go.mod（示例）：
    • module example.com/thumbnailer
    • require:
      • github.com/aws/aws-lambda-go v1.x
      • github.com/aws/aws-sdk-go-v2 v1.x
      • github.com/aws/aws-sdk-go-v2/config v1.x
      • github.com/aws/aws-sdk-go-v2/service/s3 v1.x
      • github.com/rwcarlsen/goexif v0.x
      • github.com/disintegration/imaging v1.x
  • 构建与打包：
    • 环境：GOOS=linux GOARCH=amd64（或 arm64 与对应 Lambda 架构匹配）
    • 命令：GOOS=linux GOARCH=amd64 go build -ldflags="-s -w" -o bootstrap .
    • 打包：zip function.zip bootstrap
    • 运行时：go1.x（如转为自定义运行时，可使用 provided.al2）
• Lambda 基本配置建议：
  • 内存：512–1024 MB（图像处理需要一定 CPU，内存越高，CPU 配额越高，速度更快）
  • 超时：30 秒以上（视图片大小与并发调整）
  • 保持并发限制：使用函数级并发上限避免过载后端或 S3 请求阈值

使用指南

1. 创建或选择 S3 Bucket，并启用事件通知：
   • 触发条件配置为 images/ 前缀、后缀 .jpg/.jpeg/.png
2. 部署 Lambda：
   • 设置环境变量 OUTPUT_PREFIX=thumbnails/（或自定义）
   • 设置 MAX_SIZE_MB 和 CONCURRENCY
   • 赋予最小权限的执行角色
3. 测试：
   • 向 s3:///images/ 目录上传 jpg/png 文件
   • 观察 CloudWatch Logs 的结构化日志与耗时指标
   • 验证 s3:///<OUTPUT_PREFIX>/ 同名相对路径生成了 200x200 JPEG 缩略图
4. 行为说明：
   • 非 jpg/png 文件：记录 warn 日志并结束，不报错
   • 超过 MAX_SIZE_MB：记录 warn 日志并结束
   • 目标已存在：跳过并记录 info
   • 仅处理 images/ 前缀对象，避免对其他对象误处理

最佳实践建议

• 性能优化
  • 增加 Lambda 内存以提升 CPU，图像处理将明显加速；在 512MB–1024MB 区间权衡成本与性能
  • 使用事件过滤（前缀/后缀）从源头减少无效触发
  • 设置合理 CONCURRENCY：经验值为 floor(可用内存 / (单图峰值内存×安全系数))。单图峰值内存近似为 原图大小×3～5（解码后像素内存 + 处理中间态）。例如：
    • 512MB 内存，单图峰值约 30MB，则建议并发 6～10；根据实际 CloudWatch 指标微调
  • 尽量复用 SDK 客户端（已在 init 中完成）
  • 使用 HeadObject 预检大小避免读取大对象
• 稳定性与幂等
  • 目标对象存在即跳过，避免重复写入；如需更强幂等，可记录源对象 ETag 到目标元数据并在二次校验时对比
  • 建议在 S3 触发端增加 .jpg/.jpeg/.png 后缀过滤，减少非图片触发
• 安全
  • 禁止硬编码凭证；使用 IAM 角色并授予最小权限
  • 需要合规时启用服务端加密（SSE-S3 或 SSE-KMS），并在 PutObject 设置加密参数
• 监控与告警
  • 使用结构化日志便于创建 CloudWatch Logs Metric Filter：
    • msg=skip_exceed_max_size 计数作为告警阈值
    • msg=thumb_generated 统计成功率和平均耗时
  • 结合 Lambda 内置指标（Duration、Errors、Throttles、ConcurrentExecutions）进行容量规划
• 方向与元数据
  • 代码已依据 EXIF Orientation 调整图像内容方向，保证显示正确；若需完整保留 EXIF 或复制源 EXIF，请引入专门 JPEG/EXIF 写入库并在编码后写回所需字段（注意额外内存/耗时开销）
• 成本控制
  • 通过合理的 CONCURRENCY 和事件过滤减少无效处理
  • 若高并发场景下遇到 S3 限速，可加入指数退避重试或降低并发

如需移植到其他对象存储平台（如 GCS/Azure Blob/阿里云 OSS），仅需替换存储 SDK 及事件模型，核心图像处理与并发框架可原样复用。