EventBridge Pipes to Lambda: Enrich, Filter, and Fail Closed

// contracts/order-pipe.ts — explicit semantics for ops and reviewers
export type PipeContract = {
  source: "sqs" | "kinesis" | "dynamodb-stream" | "kafka";
  filter: {
    // ✅ Match only known event types; unknown → drop (fail closed for enrichment)
    matchEventTypes: string[];
    onUnknownField: "drop" | "dlq"; // never "forward_anyway"
  };
  enrichment?: {
    type: "lambda" | "api-destination" | "step-functions";
    timeoutMs: number;
    onError: "retry_then_dlq" | "dlq" | "fail_pipe"; // ❌ never silent continue
    idempotencyKey: "sourceMessageId" | "eventId";
  };
  target: {
    type: "lambda";
    partialBatchFailure: true; // required for SQS/Kinesis batches
    maxConcurrency?: number;
  };
  dlq: { arn: string; maxReceiveCount: number };
};

// ❌ “We’ll add DLQ later” is how silent loss ships
export const ORDER_PIPE: PipeContract = {
  source: "sqs",
  filter: {
    matchEventTypes: ["order.created", "order.updated"],
    onUnknownField: "drop",
  },
  enrichment: {
    type: "lambda",
    timeoutMs: 3000,
    onError: "retry_then_dlq",
    idempotencyKey: "sourceMessageId",
  },
  target: { type: "lambda", partialBatchFailure: true, maxConcurrency: 50 },
  dlq: { arn: "arn:aws:sqs:REGION:ACCOUNT:orders-pipe-dlq", maxReceiveCount: 3 },
};

// infra/order-pipe-filter.ts — CDK-ish filter pattern (illustrative)
export const orderCreatedFilter = {
  // ✅ Envelope + required business keys
  "detail-type": ["order.created", "order.updated"],
  source: ["com.cheatcoders.orders"],
  detail: {
    orderId: [{ "exists": true }],
    status: ["PENDING", "CONFIRMED"],
  },
};

// ❌ Fragile: nested optional promo that producers sometimes omit
export const fragilePromoFilter = {
  detail: {
    promo: {
      code: [{ "prefix": "SUMMER" }],
    },
  },
};

// Unit-test patterns against fixtures — Pipe filter changes are deploys
export function assertFilterFixtures(
  pattern: Record<string, unknown>,
  fixtures: { name: string; event: unknown; expectMatch: boolean }[],
) {
  for (const f of fixtures) {
    const matched = eventBridgePatternMatches(pattern, f.event); // your tester
    if (matched !== f.expectMatch) {
      throw new Error(`filter_fixture_failed:${f.name}`);
    }
  }
}

// enrichment/enrich-order.ts
import { DynamoDBClient, GetItemCommand } from "@aws-sdk/client-dynamodb";
import { createHash } from "node:crypto";

const ddb = new DynamoDBClient({});
const TABLE = process.env.CUSTOMER_TABLE!;

export type PipeRecord = {
  orderId: string;
  customerId: string;
  status: string;
};

export type Enriched = PipeRecord & {
  tier: "free" | "pro" | "enterprise";
  enrichmentVersion: 1;
};

export async function handler(event: { orderId: string; customerId: string; status: string }): Promise<Enriched> {
  // ✅ Preserve source identity keys; only ADD fields
  // ❌ Never: regenerate orderId, coerce types, or drop status
  const out = await ddb.send(
    new GetItemCommand({
      TableName: TABLE,
      Key: { pk: { S: `CUST#${event.customerId}` } },
      ProjectionExpression: "tier",
    }),
  );

  if (!out.Item?.tier?.S) {
    // Fail closed: throw so Pipe retry/DLQ path engages
    // ❌ return { ...event, tier: "free" } on miss — invents business state
    throw new Error(`customer_tier_missing:${event.customerId}`);
  }

  return {
    ...event,
    tier: out.Item.tier.S as Enriched["tier"],
    enrichmentVersion: 1,
  };
}

// Idempotency for enrichment side effects (if any writes):
export function enrichKey(msgId: string) {
  return createHash("sha256").update(`pipe-enrich:v1:${msgId}`).digest("hex");
}

// target/process-orders.ts
import type { SQSBatchResponse, SQSEvent } from "aws-lambda";

export async function handler(event: SQSEvent): Promise<SQSBatchResponse> {
  const batchItemFailures: { itemIdentifier: string }[] = [];

  for (const rec of event.Records) {
    try {
      const body = JSON.parse(rec.body) as { orderId: string; tier: string };
      await applyOrder(body);
    } catch (err) {
      console.error("order_apply_failed", {
        messageId: rec.messageId,
        err: err instanceof Error ? err.message : String(err),
      });
      // ✅ Only failed ids — good records commit
      batchItemFailures.push({ itemIdentifier: rec.messageId });
    }
  }

  return { batchItemFailures };
}

async function applyOrder(_o: { orderId: string; tier: string }) {
  // business write — must be idempotent on orderId
}

// alerts/pipe-slo.ts — illustrative thresholds
export const PIPE_ALERTS = [
  { metric: "EnrichmentErrors", window: "5m", threshold: 5, action: "page" },
  { metric: "TargetFailures", window: "5m", threshold: 10, action: "page" },
  { metric: "DlqVisible", window: "1m", threshold: 1, action: "ticket" },
  // ✅ MatchedRate drop >50% vs 24h baseline after deploy → auto-rollback candidate
  { metric: "MatchedRateDropPct", window: "15m", threshold: 50, action: "page" },
];