Pipeline Resolver execution:
  BeforeMapping (optional)
    ↓
  Pipeline Function 1  →  Data Source A (e.g., DynamoDB — ownership check)
    ↓  ctx.stash
  Pipeline Function 2  →  Data Source B (e.g., Lambda — business logic)
    ↓  ctx.stash
  Pipeline Function 3  →  Data Source C (e.g., EventBridge — side effect)
    ↓
  AfterMapping (optional) → GraphQL response

// ❌ Wrong: mutating ctx.args to pass data between functions
// Function 1:
export function request(ctx) {
  ctx.args.ownerId = ctx.identity.sub // WRONG — corrupts args for all subsequent functions
  return get({ key: { PK: `ORDER#${ctx.args.orderId}` } })
}

// ✅ Correct: use ctx.stash for inter-function state
// ctx.stash is a plain object that persists across all functions in the pipeline
// ctx.args remains unchanged throughout the pipeline

// Pipeline Function 1: DynamoDB GetItem — ownership check
export function request(ctx) {
  return get({ key: { PK: `ORDER#${ctx.args.orderId}`, SK: 'METADATA' } })
}
export function response(ctx) {
  if (ctx.error) util.error(ctx.error.message, ctx.error.type)
  if (!ctx.result) util.error('Order not found', 'NotFound')
  if (ctx.result.userId !== ctx.identity.sub) util.unauthorized()

  // Store result in stash for downstream functions
  ctx.stash.order = ctx.result
  return ctx.result
}

// Pipeline Function 2: Lambda — business logic (payment processing)
export function request(ctx) {
  // ctx.stash.order set by Function 1 — available here
  return {
    operation: 'Invoke',
    payload: {
      action: 'processPayment',
      orderId: ctx.stash.order.orderId,
      amount: ctx.stash.order.total,
      userId: ctx.identity.sub,
    }
  }
}
export function response(ctx) {
  if (ctx.error) util.error(ctx.error.message, ctx.error.type)
  ctx.stash.chargeId = ctx.result.chargeId
  return ctx.result
}

// Pipeline Function 3: HTTP data source — EventBridge PutEvents
export function request(ctx) {
  return {
    method: 'POST',
    resourcePath: '/',
    params: {
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        Entries: [{
          Source: 'com.myapp.orders',
          DetailType: 'OrderConfirmed',
          Detail: JSON.stringify({
            orderId: ctx.stash.order.orderId,
            chargeId: ctx.stash.chargeId,
            userId: ctx.identity.sub,
          }),
          EventBusName: 'myapp-events',
        }]
      })
    }
  }
}
export function response(ctx) {
  if (ctx.error) util.error(ctx.error.message, ctx.error.type)
  // Pipeline complete — return the stashed order for the GraphQL response
  return ctx.stash.order
}

import {
  GraphqlApi, Code, FunctionRuntime,
  AppsyncFunction, Resolver, DynamoDbDataSource, LambdaDataSource, HttpDataSource
} from 'aws-cdk-lib/aws-appsync'

// Three data sources — one per pipeline function
const dynamoDS = new DynamoDbDataSource(this, 'OrderDS', { api, table })
const lambdaDS = new LambdaDataSource(this, 'PaymentDS', { api, lambdaFunction: paymentFn })
const eventBridgeDS = new HttpDataSource(this, 'EventDS', {
  api,
  endpoint: `https://events.${Stack.of(this).region}.amazonaws.com`,
  authorizationConfig: { signingRegion: Stack.of(this).region, signingServiceName: 'events' }
})

// Three AppSync Functions
const ownershipCheckFn = new AppsyncFunction(this, 'OwnershipCheck', {
  api, dataSource: dynamoDS,
  name: 'OwnershipCheck',
  runtime: FunctionRuntime.JS_1_0_0,
  code: Code.fromAsset('resolvers/functions/ownershipCheck.js'),
})
const paymentFn2 = new AppsyncFunction(this, 'ProcessPayment', {
  api, dataSource: lambdaDS,
  name: 'ProcessPayment',
  runtime: FunctionRuntime.JS_1_0_0,
  code: Code.fromAsset('resolvers/functions/processPayment.js'),
})
const emitEventFn = new AppsyncFunction(this, 'EmitEvent', {
  api, dataSource: eventBridgeDS,
  name: 'EmitEvent',
  runtime: FunctionRuntime.JS_1_0_0,
  code: Code.fromAsset('resolvers/functions/emitEvent.js'),
})

// Pipeline resolver: wires functions in sequence
new Resolver(this, 'ConfirmOrderResolver', {
  api,
  typeName: 'Mutation',
  fieldName: 'confirmOrder',
  runtime: FunctionRuntime.JS_1_0_0,
  pipelineConfig: [ownershipCheckFn, paymentFn2, emitEventFn],
  code: Code.fromInline(`
    export function request(ctx) { return {} }
    export function response(ctx) { return ctx.prev.result }
  `)
})

1. Client connects to AppSync WebSocket endpoint
2. AppSync validates auth (Cognito JWT, API key, Lambda authorizer)
3. Client subscribes: subscription { onOrderStatusChanged(userId: "u42") { id, status } }
4. AppSync registers the subscription — stores subscription context
5. Mutation fires: updateOrderStatus(orderId: "o99", status: SHIPPED)
6. AppSync executes the mutation resolver
7. Server-side filter evaluates: does this event match subscription { userId: "u42" }?
8. Filter passes → AppSync pushes { id: "o99", status: "SHIPPED" } over WebSocket to client
9. Client disconnects → AppSync removes subscription registration

type Subscription {
  onOrderStatusChanged(userId: ID!): Order
    @aws_subscribe(mutations: ["updateOrderStatus", "cancelOrder"])
  # This subscription fires when updateOrderStatus OR cancelOrder mutations are called
  # The userId argument is used in server-side filter expressions
}

// In the Mutation.updateOrderStatus resolver:
// After updating DynamoDB, set a subscription filter on the mutation's response
// This filter controls which active subscriptions receive the push

export function response(ctx) {
  if (ctx.error) util.error(ctx.error.message, ctx.error.type)
  const order = ctx.result

  // Set filter: only push to subscribers whose subscription userId matches this order's userId
  extensions.setSubscriptionFilter(
    util.transform.toSubscriptionFilter({
      or: [
        { userId: { eq: order.userId } },  // Push to the order owner
        { userId: { eq: 'admin' } },        // Also push to admin subscribers
      ]
    })
  )

  return order
}

// Subscription resolver: validates the subscriber is authorized
export function request(ctx) {
  // ctx.identity.sub = authenticated Cognito user's UUID
  // Only allow users to subscribe to their own order updates
  if (ctx.args.userId !== ctx.identity.sub) {
    util.unauthorized()
  }
  return {} // empty return for subscription request function
}

export function response(ctx) {
  return ctx.result
}

# Multi-auth schema: Cognito + IAM + API Key simultaneously

type Query {
  # Cognito users and IAM principals can get user profiles
  getUser(id: ID!): User @aws_cognito_user_pools @aws_iam

  # Public leaderboard — API key access for unauthenticated clients
  getLeaderboard: [LeaderboardEntry!]! @aws_api_key @aws_cognito_user_pools
}

type Mutation {
  # Only Cognito users can create orders (not IAM service accounts)
  createOrder(input: CreateOrderInput!): Order! @aws_cognito_user_pools

  # Internal service mutations — IAM only (service-to-service, not end-user callable)
  internalUpdateInventory(productId: ID!, delta: Int!): Boolean! @aws_iam
}

type User {
  id: ID!
  email: String!
  name: String
  # Sensitive field: only accessible to Cognito users (not IAM or API key)
  paymentMethod: PaymentMethod @aws_cognito_user_pools
  # Admin-only field: requires Cognito + Admin group membership
  internalNotes: String @aws_auth(cognito_groups: ["Admin"])
}

// In a resolver: check Cognito group membership
export function request(ctx) {
  const groups: string[] = ctx.identity.claims['cognito:groups'] ?? []

  // Protect admin-only mutations inside the resolver (defense in depth)
  if (!groups.includes('Admin')) {
    util.unauthorized()
  }
  return get({ key: { PK: `USER#${ctx.args.id}` } })
}

// CDK: enable per-resolver caching
import { CachingConfig, CachingBehavior } from 'aws-cdk-lib/aws-appsync'

const api = new GraphqlApi(this, 'Api', {
  name: 'order-api',
  schema: SchemaFile.fromAsset('schema.graphql'),
  // API-level cache configuration (required to enable per-resolver caching)
})

// Enable caching on the AppSync API
api.addSchemaDependency(new CfnApiCache(this, 'ApiCache', {
  apiId: api.apiId,
  type: 'SMALL',                         // Instance size (SMALL/MEDIUM/LARGE/etc.)
  ttl: 300,                              // Default TTL: 300 seconds
  apiCachingBehavior: 'PER_RESOLVER_CACHING', // Cache per resolver, not full request
  atRestEncryptionEnabled: true,
  transitEncryptionEnabled: true,
}))