AWS and Google Cloud events

import { workflow } from "shiftlabs";

type Message = { source: string; id: string; body: any };

// One delivery from EventBridge, SNS or Pub/Sub, as the trigger passed it.
function read(payload: any): Message {
  if (typeof payload === "string") {
    const sns = JSON.parse(payload);
    return { source: "sns", id: sns.MessageId, body: sns };
  }
  if (payload.message?.messageId) {
    const bytes = Uint8Array.from(atob(payload.message.data ?? ""), (c) =>
      c.charCodeAt(0),
    );
    const text = new TextDecoder().decode(bytes);
    return { source: "pubsub", id: payload.message.messageId, body: text };
  }
  return { source: "eventbridge", id: payload.id, body: payload.detail };
}

// A UTC date, offset days ago.
const day = (offset: number) =>
  new Date(Date.now() - offset * 86_400_000).toISOString().slice(0, 10);

export default workflow(async ({ payload, step, state, runId, log }) => {
  const { source, id, body } = read(payload);

  if (source === "sns" && body.Type === "SubscriptionConfirmation") {
    const host = new URL(body.SubscribeURL).hostname;
    if (!host.startsWith("sns.") || !host.endsWith(".amazonaws.com")) {
      throw new Error(`not an SNS address: ${host}`);
    }
    await step.run("confirm subscription", async () => {
      const r = await fetch(body.SubscribeURL);
      if (!r.ok) throw new Error(`SNS answered ${r.status}`);
    });
    return "subscribed";
  }

  const seen = `${source}:${id}`;
  const fresh = await step.run("claim", async () => {
    // Forget ids from before yesterday, so the keys don't pile up.
    for (const key of await state.list({ prefix: "seen:", limit: 20 })) {
      if (key < `seen:${day(1)}`) await state.delete(key);
    }
    for (const d of [day(1), day(0)]) {
      const claimed = await state.get<{ runId: string }>(`seen:${d}:${seen}`);
      if (claimed) return claimed.runId === runId;
    }
    await state.set(`seen:${day(0)}:${seen}`, { runId });
    return true;
  });
  if (!fresh) return "duplicate";

  await step.run("handle", async () => {
    log.info("message", { source, id });
  });
  return id;
});

import base64
import json
from datetime import datetime, timedelta, timezone
from urllib.parse import urlparse

from workers import fetch
from shiftlabs import workflow


def read(payload):
    """One delivery from EventBridge, SNS or Pub/Sub: (source, id, body)."""
    if isinstance(payload, str):
        sns = json.loads(payload)
        return "sns", sns["MessageId"], sns
    if "message" in payload:
        message = payload["message"]
        text = base64.b64decode(message.get("data", "")).decode()
        return "pubsub", message["messageId"], text
    return "eventbridge", payload["id"], payload["detail"]


def day(offset):
    """A UTC date, offset days ago."""
    then = datetime.now(timezone.utc) - timedelta(days=offset)
    return then.strftime("%Y-%m-%d")


@workflow
async def cloud_events(ctx):
    source, message_id, body = read(ctx.payload)

    if source == "sns" and body["Type"] == "SubscriptionConfirmation":
        url = body["SubscribeURL"]
        host = urlparse(url).hostname or ""
        if not (host.startswith("sns.") and host.endswith(".amazonaws.com")):
            raise ValueError(f"not an SNS address: {host}")

        async def confirm():
            r = await fetch(url)
            if not r.ok:
                raise Exception(f"SNS answered {r.status}")

        await ctx.step.run("confirm subscription", confirm)
        return "subscribed"

    seen = f"{source}:{message_id}"

    async def claim():
        # Forget ids from before yesterday, so the keys don't pile up.
        for key in await ctx.state.list(prefix="seen:", limit=20):
            if key < f"seen:{day(1)}":
                await ctx.state.delete(key)
        for d in (day(1), day(0)):
            claimed = await ctx.state.get(f"seen:{d}:{seen}")
            if claimed:
                return claimed["runId"] == ctx.run_id
        await ctx.state.set(f"seen:{day(0)}:{seen}", {"runId": ctx.run_id})
        return True

    if not await ctx.step.run("claim", claim):
        return "duplicate"

    async def handle():
        ctx.log.info("message", {"source": source, "id": message_id})

    await ctx.step.run("handle", handle)
    return message_id

curl -X PUT https://api.shiftlabs.org/v1/workflows/cloud-events \
  -H "authorization: Bearer $SHIFTLABS_API_KEY" \
  --data-binary @cloud-events.ts

# Run this once per source. Answers 201 {"triggerId", "url"}.
curl -X POST https://api.shiftlabs.org/v1/triggers \
  -H "authorization: Bearer $SHIFTLABS_API_KEY" \
  -H "content-type: application/json" \
  -d '{"workflow": "cloud-events"}'

# A connection needs an authorization method. The trigger doesn't check it,
# so any API key name and value will do.
aws events create-connection --name shiftlabs \
  --authorization-type API_KEY \
  --auth-parameters '{"ApiKeyAuthParameters":
    {"ApiKeyName": "x-shiftlabs-unused", "ApiKeyValue": "unused"}}'

# A trigger takes 100 deliveries per 10 seconds: send at most 10 a second.
aws events create-api-destination --name shiftlabs-cloud-events \
  --connection-arn <connection ARN> \
  --invocation-endpoint https://hooks.shiftlabs.org/t/<triggerId> \
  --http-method POST \
  --invocation-rate-limit-per-second 10

# The role needs events:InvokeApiDestination on the API destination.
aws events put-targets --rule <rule name> \
  --targets 'Id=shiftlabs,Arn=<API destination ARN>,RoleArn=<role ARN>'

aws sns subscribe --topic-arn <topic ARN> --protocol https \
  --notification-endpoint https://hooks.shiftlabs.org/t/<triggerId>

# Confirmed once the subscription has an ARN instead of PendingConfirmation.
aws sns list-subscriptions-by-topic --topic-arn <topic ARN>

gcloud pubsub subscriptions create shiftlabs-cloud-events \
  --topic=<topic> \
  --push-endpoint=https://hooks.shiftlabs.org/t/<triggerId>

aws sns publish --topic-arn <topic ARN> --message '{"order": "ord_123"}'

gcloud pubsub topics publish <topic> --message='{"order": "ord_123"}'

# Each source's run, newest first, with origin "trigger".
curl "https://api.shiftlabs.org/v1/runs?workflow=cloud-events&limit=5" \
  -H "authorization: Bearer $SHIFTLABS_API_KEY"