1# Event pipeline: Pub/Sub to BigQuery
2
3An analytics pipeline. A public Cloud Run service accepts events and publishes them to Pub/Sub. A Cloud Run Job drains the backlog into BigQuery in batches. Nothing touches BigQuery on the request path, so a slow warehouse cannot take ingestion down.
4
5```sh
6curl -X POST "$URL/events" -H 'content-type: application/json' -d '{"type":"signup","payload":{"plan":"pro"}}'
7# 202 {"id":"5f0c…"}
8curl -X POST "$URL/drain"
9# 202 {"execution":"projects/…/executions/…"}
10curl "$URL/events/count?type=signup"
11# 200 {"count":1}
12```
13
14## Architecture
15
16- [`Events`](./src/resources.ts) — Pub/Sub topic every event is published to.
17- [`Inbox`](./src/resources.ts) — pull subscription on `Events`. It holds messages until they are acked, so the drain can run whenever a batch is due.
18- [`Analytics`](./src/resources.ts) / [`EventsTable`](./src/resources.ts) — BigQuery dataset and `events` table (`id`, `type`, `occurredAt`, `payload` as JSON).
19- [`Ingest`](./src/Ingest.ts) — public `GCP.Function` (a Cloud Run service):
20 - `POST /events` — publish `{ type, payload }` to the topic.
21 - `POST /drain` — start a `Drain` execution now.
22 - `GET /events/count?type=` — count rows in BigQuery.
23- [`Drain`](./src/Drain.ts) — `GCP.Run.Job`. It pulls batches of up to 100 messages, inserts them into the table, and acks them, until a pull comes back empty.
24
25The drain inserts first and acks second. A crash between the two redelivers the batch; BigQuery `insertIds` collapse the duplicates inside its dedup window.
26
27## Bindings and IAM
28
29| Host | Binding | IAM granted to the host's service account |
30| ------ | ---------------------------------------------------------------- | ------------------------------------------------------------------------------------ |
31| Ingest | `GCP.PubSub.WriteTopic(Events)` + `WriteTopicHttp` | `roles/pubsub.publisher` on the topic |
32| Ingest | `GCP.BigQuery.ReadTable(EventsTable)` + `ReadTableHttp` | `roles/bigquery.dataViewer` on the table; `roles/bigquery.jobUser` on the project (query jobs can only be granted there) |
33| Ingest | `GCP.Run.RunJob(Drain)` + `RunJobHttp` | `roles/run.jobsExecutorWithOverrides` on the `Drain` job |
34| Drain | `GCP.PubSub.ReadSubscription(Inbox)` + `ReadSubscriptionHttp` | `roles/pubsub.subscriber` on the subscription |
35| Drain | `GCP.BigQuery.WriteTable(EventsTable)` + `WriteTableHttp` | `roles/bigquery.dataEditor` on the table |
36
37## Deploy
38
39Credentials come from your alchemy profile: run `alchemy profile` once and pick GCP (*Service account JSON* for a key file, or *Stored* for an access token or key kept in `~/.alchemy/credentials`, plus a default region), then deploy with `--profile <name>`.
40
41From the repository root:
42
43```sh
44pnpm install
45cd examples/gcp-event-pipeline
46pnpm deploy --profile <name>
47```
48
49Docker must be running, because both hosts are built locally from `main`. To drain on a schedule instead of on demand, point a Cloud Scheduler job at `Drain`.
50
51## Live test
52
53```sh
54ALCHEMY_PROFILE=<name> bun test --timeout 1200000
55```
56
57[The test](./test/integ.test.ts) deploys the stack and publishes events, checking that nothing reaches BigQuery before a drain. It then starts `Drain` through `POST /drain` and polls `GET /events/count` until the rows land. At the end it destroys the stack.
58
59A Cloud Run Job execution can take a couple of minutes to start.
60
61## Destroy
62
63```sh
64pnpm destroy --profile <name>
65```
66
67This deletes the service, the job, their IAM grants, the subscription, the topic, and the dataset with its table.