Gram's declarative GCP Pub/Sub system — topics and subscriptions are declared as protobuf message options, generated into `infra/gen/kcc.yaml`, and used at runtime via a type-safe publisher/subscriber library and the `gram streams` process. Activate for any Pub/Sub work in Gram: adding or changing a topic/subscription, declaring `(gcp.pubsub.v1.topic)`/`(gcp.pubsub.v1.subscription)` options, publishing or consuming messages, implementing a stream handler, dead-letter queues, or the local emulator — including phrasings like "add an outbox topic", "wire up a consumer", or "why isn't my topic created in GCP" even when Pub/Sub isn't named.
Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
Gram's declarative GCP Pub/Sub system — topics and subscriptions are declared as protobuf message options, generated into `infra/gen/kcc.yaml`, and used at runtime via a type-safe publisher/subscriber library and the `gram streams` process. Activate for any Pub/Sub work in Gram: adding or changing a topic/subscription, declaring `(gcp.pubsub.v1.topic)`/`(gcp.pubsub.v1.subscription)` options, publishing or consuming messages, implementing a stream handler, dead-letter queues, or the local emulator — including phrasings like "add an outbox topic", "wire up a consumer", or "why isn't my topic created in GCP" even when Pub/Sub isn't named.
Gram declares its GCP Pub/Sub topology as data on protobuf messages, not as
hand-written Terraform or Config Connector YAML. You annotate a "marker" message
with a topic or subscription option; a generator walks the compiled descriptors
and emits a Config Connector Helm values document (infra/gen/kcc.yaml); the
deployment tooling consumes that document to provision real topics and
subscriptions. The same proto options drive a type-safe Go
publisher/subscriber library, so the infrastructure and the application code can
never disagree about a topic's name or a subscription's wiring — both read from
one source of truth.
The whole point of the design is that an engineer adding a topic touches only a
.proto file under infra/proto/. Everything downstream — the Config Connector
specs, the per-environment rollout, and the runtime handle — is derived from it.
Everything in this skill lives in this repo (); the deployment side is
referred to only abstractly as "the deployment tooling."
infra/
Mental model: marker messages
A topic or a subscription is declared by attaching a message option to a
protobuf message. The message itself can be the event schema (for a topic) or an
empty placeholder (for a subscription) — what matters is the option, not the
fields. One message must not carry both a topic and a subscription option;
declare them separately. This keeps a topic's payload schema and a consumer's
config as distinct, independently evolvable things.
The option definitions live in infra/proto/gcp/pubsub/v1/options.proto:
(gcp.pubsub.v1.subscription) — SubscriptionOptions: optional name,
required topic (the proto full name of the topic-declaring message, e.g.
"gram.outbox.v1.Event"), plus retention, ack_deadline, retry_policy,
filter, dead_letter, expiration_ttl, retain_acked_messages, labels.
Authoring a topic
Add the topic option to the message that is the event payload, so the schema
and the topic travel together. See infra/proto/gram/outbox/v1/event.proto:
With no explicit name, the topic ID is the kebab-cased proto full name:
gram.outbox.v1.Event → gram-outbox-v1-event. Set name only when you need
to diverge from that.
Authoring a subscription
Declare a subscription on its own marker message (no payload fields needed) and
point topic at the topic message's full name. See
infra/proto/gram/outbox/v1/processor.proto:
With no explicit name, the subscription ID is the kebab-cased proto full name
(same rule as topics): gram.outbox.v1.Processor → gram-outbox-v1-processor.
Set name only when you need to diverge from that. The
topic reference is validated against the discovered topic set during
generation, so a typo fails the build rather than producing a dangling
subscription.
Dead-letter queues are synthesized
When a subscription sets dead_letter, the generator auto-creates a DLQ
topic — you do not declare it. The default name is <subscription>-dlq
(override with dead_letter.name). The DLQ topic carries the same message
schema as the source and is labeled dlq_for: <subscription>. This is why
subscription IDs are length-capped below the topic limit: room must be left for
the -dlq suffix.
Regenerating after a proto change
Always run the generator after editing any .proto under infra/proto/ and
commit the result — infra/gen/kcc.yaml is the committed artifact the
deployment tooling consumes:
mise run gen:infra
This task (.mise-tasks/gen/infra.sh) does three things: buf generate (proto
→ Go in infra/gen/), buf build (compiled FileDescriptorSet →
infra/cmd/infra/descriptors.pb), then go run ./infra/main.go gen-cc to write
infra/gen/kcc.yaml. The generated Go and the descriptor blob are gitignored
(**/descriptors.pb) and excluded from formatting; infra/gen/kcc.yaml is
committed.
How generation works (infra/internal/gcp/)
File
Responsibility
pubsub_discover.go
Walks the descriptor set, extracts options, resolves names, dedupes, validates, synthesizes DLQ topics. The DesiredTopic / DesiredSubscription structs are the in-memory topology.
cc_pubsub.go
Projects the topology into sorted Config Connector specs (buildPubSubValues).
values.go
The Helm values document types (pubSubValuesDocument). Specs embed the real Config Connector PubSubTopicSpec / PubSubSubscriptionSpec types so field names match the CRD exactly.
cc.go
ConfigConnectorPubSub.Provision orchestrates discover → build → write, emitting the # Code generated … DO NOT EDIT. YAML.
Key behaviors worth knowing:
Validation (validateTopicID / validateSubscriptionID): GCP naming
rules — must start with a letter, 3–255 chars, no goog prefix, no full
resource paths. Duplicate topic or subscription names, an unknown topic
reference, or a DLQ name colliding with a declared topic all fail generation.
Labels: every generated resource gets managed_by: proto_pubsub_orchestrator plus a proto_message label carrying the source
message's full name; subscriptions also carry topic_proto_message. These
make resources traceable back to their declaration.
Stable output: topics and subscriptions are sorted by name so the
generated file diffs cleanly across runs. Durations render as integer-second
strings (e.g. 604800s) as Config Connector expects.
Separation of concerns: the generator emits only the topology under a
pubsub key (names, labels, specs). Per-resource deployment metadata
(project, namespace, deletion/prune policy) is applied downstream by the
deployment tooling, never here — so don't expect this generator to emit it.
Runtime: publishing and subscribing (infra/pkg/gcp/)
Application code never hard-codes a topic or subscription name. It hands a proto
message to a broker and gets back a generic, type-safe handle. The broker
resolves names from the same proto options used for generation.
Two brokers implement both PublisherBroker and SubscriberBroker:
PubSubBroker (pubsub_gcp.go) — talks to real GCP; assumes topics/subs
already exist (Config Connector created them).
EmulatedPubSubBroker (pubsub_local.go) — for local dev against the
emulator; reconciles topics and subscriptions on demand since the emulator
has no Config Connector.
Both take the embedded descriptors blob. Usage (condensed from
infra/cmd/infra/demo.go, the runnable reference):
broker := gcppub.NewEmulatedPubSub(logger, projectID, client, descriptors)
// Publisher for the topic declared by *outboxv1.Event.
pub, _ := gcppub.PubSubPublisherForMessage(ctx, broker, &outboxv1.Event{})
// Subscriber for the *outboxv1.Processor subscription, receiving *outboxv1.Event.// Read as: "a handle on the Processor subscription delivering Event messages."
sub, _ := gcppub.PubSubSubscriberForMessage(ctx, broker, &outboxv1.Event{}, &outboxv1.Processor{})
pub.Publish(ctx, &msg).Get(ctx) // proto-marshaled, with content-type + schema attributes// The callback receives the unmarshaled message plus delivery metadata.// Return nil to ack; return a non-nil error to nack (and trigger redelivery /// dead-lettering if enabled for the topic and subscription).
sub.Receive(ctx, func(ctx context.Context, msg *outboxv1.Event, meta gcppub.MessageMetadata)error {
_ = msg // already unmarshaled to *outboxv1.Event
_ = meta.ID // broker-assigned message ID
_ = meta.Attributes
_ = meta.DeliveryAttempt // set when dead-lettering is enabled, else nilreturnnil
})
publisher.go / subscriber.go define the generic Publisher[M] and
Subscriber[M] over cloud.google.com/go/pubsub/v2. Messages are
proto-marshaled and tagged with content-type: application/x-protobuf and a
schema attribute (the message full name); the subscriber unmarshals back into
a fresh M and hands it to your callback along with a MessageMetadata. The
callback's return value drives ack/nack — nil acks, non-nil nacks — so you no
longer call Ack/Nack yourself. Tune behavior with
WithPubSubPublishSettings / WithPubSubReceiveSettings.
This is the low-level library. Inside the Gram server you almost never call
sub.Receive directly — you write a streams.Handler and register it in the
gram streams process, which wraps the receive loop with tracing, panic
recovery, and ack/nack plumbing for you. See the next section.
Implementing a subscriber in the server
Subscribers run in a dedicated long-running process, gram streams
(server/cmd/gram/streams.go). It is its own service — separate from the API
server and the Temporal worker — so message consumers scale and fail
independently. Locally it runs as the streams process under pitchfork; start
it with pitchfork start streams. Adding a consumer is two steps: write a
handler, then register it.
Step 1 — write a handler
A consumer is a streams.Handler[M] (server/internal/streams/handlers.go),
where M is the proto message the topic carries:
type Handler[M any] interface {
Handle(context.Context, M, gcp.MessageMetadata) error
}
Implement it in the domain package that owns the behavior (not in streams/ and
not in cmd/gram/). server/internal/ping/handler.go is the canonical
reference — a struct holding its dependencies, a NewHandler constructor, and a
Handle method:
type Handler struct {
logger *slog.Logger
}
funcNewHandler(logger *slog.Logger) *Handler {
return &Handler{logger: logger}
}
func(h *Handler) Handle(ctx context.Context, m *pingv2.Message, _ gcp.MessageMetadata) error {
// ... do the work ...returnnil
}
The handler's job is narrow: process one message and return. The return value
is the ack/nack signal — nil acks the message; a non-nil error nacks it,
triggering redelivery and eventual dead-lettering if the subscription declares a
dead_letter policy (see "Authoring a subscription"). So return an error only
when you genuinely want the message retried; for a poison message you can never
process, log it and return nil to drop it. Do not call Ack/Nack, start
your own receive goroutine, or open a span for the receive loop — the runner
does all of that (next step).
HandlerFunc[M] adapts a bare function to the interface when a struct is
overkill.
Step 2 — register it in the streams runner
In streams.go, the Action builds the shared dependencies (db, redis, the
gcp.NewPubSubBroker) and a receiverGroup. Register each handler inside the
marked block with mustReceive:
// Start subscription receivers in this block
{
mustReceive(rg, &pingv2.Message{}, &pingv2.Processor{}, ping.NewHandler(logger))
}
The three positional arguments mirror the proto declarations:
&pingv2.Message{} — the topic message; its type fixes M, the payload
your handler receives.
&pingv2.Processor{} — the subscription marker message (the one carrying
(gcp.pubsub.v1.subscription)).
ping.NewHandler(...) — your handler, with its dependencies injected from the
ones the Action already constructed. If a handler needs the db or redis,
pass them here; add new shared dependencies to the Action body.
Trailing gcp.SubscriberOptions (e.g. WithPubSubReceiveSettings) can follow.
Use mustReceive (it panics on a misconfigured subscription, which is a
programmer error caught at startup); receive returns the error if you need to
handle it.
What the runner does for you
receive/mustReceive exist so handlers stay tiny. For each registration the
runner: resolves the subscription via gcp.PubSubSubscriberForMessage, launches
the receive loop in the shared errgroup (so any subscriber's fatal error tears
the whole process down for a clean restart), starts a stream.handleMessage
span per message, stamps pub/sub subscriber context values for telemetry,
recovers panics in your handler (converting them to a nacking error instead
of crashing the goroutine), and treats context.Canceled as a clean shutdown
rather than a failure. Because all of this is centralized, you get consistent
observability and failure semantics across every consumer without writing any of
it.
Driving the flow end to end (optional publisher)
ping also ships a StartPublisher (server/internal/ping/publisher.go)
launched with group.Go alongside the receivers — a heartbeat that publishes a
message every few seconds so you can confirm the publish→subscribe loop is alive
in logs. It is a sanity-check harness, not a pattern to copy for real
publishers: production code publishes from wherever the event originates (an API
handler, a workflow activity) using gcp.PubSubPublisherForMessage, not from a
loop in the streams process.
Local development
mise.toml sets PUBSUB_EMULATOR_HOST and compose.yml runs a
pubsub-emulator service (the google/cloud-sdk emulators image on port 8085).
With the emulator host set, EmulatedPubSubBroker creates topics/subscriptions
lazily on first publish/subscribe, so you don't need Config Connector locally.
The infra demo command (go run ./infra/main.go demo) is an end-to-end
publish/subscribe loop you can run to sanity-check the framework.
How a declaration reaches a real environment
This is the part that most often confuses people, so it's worth being precise
about what this repo does and does not do.
Declaring a topic and committing infra/gen/kcc.yaml does not create
anything in GCP. The proto is the source; infra/gen/kcc.yaml is the
committed artifact. Provisioning happens later, in separate deployment
tooling, not as a side effect of merging.
Rollout is decoupled and version-pinned per environment. Each environment
runs the topology from a specific committed revision, not from main directly.
A topology change therefore reaches one environment at a time as that
environment is rolled forward — so a freshly declared topic legitimately may
not exist yet in a given environment even though it is on main.
Local working ≠ deployed. The emulator reconciles topics/subscriptions on
the fly, so "works locally but missing in GCP" is the expected signature when
the rollout simply hasn't reached that environment yet.
PR previews never create real topics, by design.
So when something seems missing in a real environment, the productive question is
not "is the proto correct?" but "did infra/gen/kcc.yaml get regenerated and
committed, and has the environment in question been rolled forward to a revision
that includes it?" The rollout mechanics themselves live with the deployment
tooling, outside this repo.
Gotchas and conventions
Run mise run gen:infra and commit infra/gen/kcc.yaml after any proto
change. A stale kcc.yaml is what actually ships — the proto is only the
source. Regenerating without committing the result is the most common reason a
declaration silently does nothing downstream.
topic references use the proto full name, not the resolved topic ID
(gram.outbox.v1.Event, not gram-outbox-v1-event).
Don't declare DLQ topics — they're synthesized from dead_letter.
One option per message: a message carrying both a topic and a subscription
option fails generation by design.
Retiring a topic is a deliberate, decoupled act. Removing a declaration
stops the topology from managing it, but the deployment tooling is configured
so existing GCP topics are not destroyed by a topology change. Treat removals
as a separate, intentional step rather than assuming a deleted declaration
tears down the resource.
