Process background work

In this example, we are going to process queue messages and new storage objects inside the customer environment where they are created. An hourly schedule runs there too, and each handler records its result in KV.

The useful part is location: your hosted product does not poll customer queues, copy every object into your cloud, or keep credentials for those resources. The Worker receives each event locally and your product asks for the result through a Command.

Alien connects infrastructure events to ordinary TypeScript handlers. The same handler APIs work during local development and when the stack is deployed to AWS, Google Cloud, or Azure.

╔════════════════════════════════╗        ╔══════════════════════════════════╗
║ Events                         ║        ║ Customer environment             ║
║                                ║        ║                                  ║
║ Queue message                  ║───────▶║  ┌────────────────────────────┐  ║
║ New object                     ║───────▶║  │ Worker                     │  ║
║ Schedule                       ║───────▶║  │ Your event handlers        │  ║
╚════════════════════════════════╝        ║  └────────────────────────────┘  ║
                                          ╚══════════════════════════════════╝

We will declare the Queue, Storage, KV, and three triggers in alien.ts. Then we will implement each handler in src/index.ts and send a test message locally.

Connect events to the Worker

alien.ts

const processor = new alien.Worker("processor")
  .code({ type: "source", src: "./", toolchain: { type: "typescript" } })
  .link(inbox)
  .link(data)
  .link(events)
  .trigger({ type: "queue", queue: inbox.ref() })
  .trigger({ type: "storage", storage: data.ref(), events: ["created"] })
  .trigger({ type: "schedule", cron: "0 * * * *" })
  .permissions("execution")
  .build()

Each .trigger(...) connects a deployment resource or schedule to this Worker. Each .link(...) gives the handler a typed binding to the named Queue, Storage, or KV resource. The permission profile limits which data operations the Worker may perform.

Write the handlers

src/index.ts

onQueueMessage("*", async message => {
  await kv("events").setJson(`queue:${message.id}`, {
    payload: message.payload,
    processedAt: new Date().toISOString(),
  })
})

onStorageEvent("*", async event => {
  await kv("events").setJson(`storage:${event.objectKey}`, {
    objectKey: event.objectKey,
    eventType: event.eventType,
  })
})

onCronEvent("*", async event => {
  await kv("events").setJson(`cron:${Date.now()}`, {
    scheduledTime: event.timestamp,
  })
})

Queue and storage handlers must be safe to run more than once. Use stable event IDs where duplicate work would be harmful.

Send some work locally

alien init event-pipeline-ts
alien dev
alien dev commands invoke --deployment default --command send-test-message --params '{"message":"hello"}'

Wait for the message, then invoke get-events or get-stats. The complete source also shows an HTTP ingestion route and waitUntil for background work.

Run the pipeline on customer infrastructure

1 · Release

alien release

Publishes a version. Nothing is deployed for a customer yet.

2 · Invite

alien onboard acme-corp

Creates a deployment link for that customer.

3 · Deploy

The customer opens the link and deploys into their environment.

Alien creates the Queue, Storage, KV, and Worker together. Events are processed in that environment instead of being polled and copied into your cloud first.

What you built

You connected three infrastructure event sources to ordinary TypeScript handlers in the customer's cloud. Customer payloads can be processed and stored there; your product receives only the inspection results it asks for instead of operating a cross-cloud polling and data-copying service.

Source: examples/event-pipeline-ts.

Next: Events and schedules, Queue.