feat: use exchange mechanism

CONTRIBUTING.md

@@ -113,7 +113,7 @@ docker run -d --rm --name postgres -e POSTGRES_PASSWORD=password -p 5432:5432 po
 Run `webhook-gateway` integration tests using:
 
 ```shell
-APP_PORT=8080 APP_WEBHOOK_SECRET_VALUE=fake RABBITMQ_CONNECT_STRING="amqp://guest:guest@localhost:5672/" go test -cover -v  ./cmd/webhook-gateway -integration
+APP_WEBHOOK_SECRET_VALUE=fake APP_PORT=8080 RABBITMQ_CONNECT_STRING="amqp://guest:guest@localhost:5672/" go test -v -cover -tags=integration -race ./cmd/webhook-gateway/...
 ```
 
 It assumes you have access to a RabbitMQ server running reachable at $RABBITMQ_CONNECT_STRING.
@@ -126,11 +126,10 @@ docker run -d  --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4-manag
 Run `planner` integration tests using:
 
 ```shell
-POSTGRESQL_DB_CONNECT_STRING="postgres://postgres:postgres@localhost:5432/gh_runner_operators?sslmode=disable" APP_PORT=8080 go test ./cmd/planner -v -count=1
+APP_PORT=8080 POSTGRESQL_DB_CONNECT_STRING="postgres://postgres:password@localhost:5432/postgres?sslmode=disable" RABBITMQ_CONNECT_STRING="amqp://guest:guest@localhost:5672/" go test -v -cover -tags=integration -race ./cmd/planner/...
 ```
 
-It assumes you are connected to a local PostgreSQL database "gh_runner_operators" as user "postgres" on host "localhost" at port "5432".
-
+It assumes you can reach postgres and rabbitmq servers as described above.
 
 ### Test design
 

cmd/planner/main.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/canonical/github-runner-operators/internal/database"
 	"github.com/canonical/github-runner-operators/internal/planner"
+	"github.com/canonical/github-runner-operators/internal/queue"
 	"github.com/canonical/github-runner-operators/internal/telemetry"
 	"github.com/canonical/github-runner-operators/internal/version"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
@@ -33,7 +34,6 @@ const (
 	portEnvVar        = "APP_PORT"
 	serviceName       = "github-runner-planner"
 	rabbitMQUriEnvVar = "RABBITMQ_CONNECT_STRING"
-	queueName         = "webhook-queue"
 	shutdownTimeout   = 30 * time.Second
 )
 
@@ -71,7 +71,8 @@ func main() {
 
 	metrics := planner.NewMetrics(db)
 
-	consumer := planner.NewJobConsumer(rabbitMQUri, queueName, db, metrics)
+	amqpConsumer := queue.NewAmqpConsumer(rabbitMQUri, queue.DefaultQueueConfig())
+	consumer := planner.NewJobConsumer(amqpConsumer, db, metrics)
 
 	var consumerWg sync.WaitGroup
 	consumerWg.Add(1)

cmd/planner/main_test.go

@@ -19,6 +19,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/canonical/github-runner-operators/internal/queue"
 	amqp "github.com/rabbitmq/amqp091-go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -74,7 +75,7 @@ func TestMain_FlavorPressure(t *testing.T) {
 	assert.Eventually(t, func() bool {
 		press := ctx.getFlavorPressure(flavor)
 		return press[flavor] > pressure
-	}, 20*time.Minute, 500*time.Millisecond)
+	}, 2*time.Minute, 100*time.Millisecond, "expected flavor pressure to increase after webhook processing")
 
 	// Trigger graceful shutdown and verify services stop cleanly
 	ctx.shutdownMain()
@@ -112,7 +113,7 @@ func (ctx *testContext) constructWebhookPayload(labels []string) []byte {
 	return body
 }
 
-// declareQueue declares a RabbitMQ queue for testing purposes.
+// declareQueue declares the exchange, queue, and binding for testing.
 func (ctx *testContext) declareQueue() {
 	ctx.t.Helper()
 
@@ -124,6 +125,19 @@ func (ctx *testContext) declareQueue() {
 	require.NoError(ctx.t, err, "open channel")
 	defer ch.Close()
 
+	config := queue.DefaultQueueConfig()
+
+	err = ch.ExchangeDeclare(
+		config.ExchangeName,
+		"direct", // type
+		true,     // durable
+		false,    // auto-delete
+		false,    // internal
+		false,    // no-wait
+		nil,
+	)
+	require.NoError(ctx.t, err, "declare exchange")
+
 	_, err = ch.QueueDeclare(
 		ctx.queueName,
 		true,  // durable
@@ -133,9 +147,18 @@ func (ctx *testContext) declareQueue() {
 		nil,
 	)
 	require.NoError(ctx.t, err, "declare queue")
+
+	err = ch.QueueBind(
+		ctx.queueName,
+		config.RoutingKey,
+		config.ExchangeName,
+		false, // no-wait
+		nil,
+	)
+	require.NoError(ctx.t, err, "bind queue")
 }
 
-// publishAndWaitAck publishes a message to the given RabbitMQ queue and waits for an ack.
+// publishAndWaitAck publishes a message to the exchange with routing key and waits for an ack.
 func (ctx *testContext) publishAndWaitAck(body []byte) {
 	ctx.t.Helper()
 
@@ -152,8 +175,11 @@ func (ctx *testContext) publishAndWaitAck(body []byte) {
 
 	confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1))
 
+	config := queue.DefaultQueueConfig()
 	err = ch.Publish(
-		"", ctx.queueName, false, false,
+		config.ExchangeName, // publish to exchange instead of directly to queue
+		config.RoutingKey,   // use routing key
+		false, false,
 		amqp.Publishing{
 			ContentType: "application/json",
 			Body:        body,

cmd/webhook-gateway/main.go

@@ -19,7 +19,6 @@ import (
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 )
 
-const queueName = "webhook-queue"
 const webhookPath = "/webhook"
 const portEnvVar = "APP_PORT"
 const rabbitMQUriEnvVar = "RABBITMQ_CONNECT_STRING"
@@ -51,7 +50,7 @@ func main() {
 		log.Fatalln(webhookSecretEnvVar + " environment variable not set")
 	}
 
-	p := queue.NewAmqpProducer(uri, queueName)
+	p := queue.NewAmqpProducer(uri, queue.DefaultQueueConfig())
 	handler := &webhook.Handler{
 		WebhookSecret: webhookSecret,
 		Producer:      p,

cmd/webhook-gateway/webhook_test.go

@@ -19,19 +19,23 @@ import (
 	"testing"
 	"time"
 
+	"github.com/canonical/github-runner-operators/internal/queue"
+
 	amqp "github.com/rabbitmq/amqp091-go"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestHTTPRequestIsForwarded(t *testing.T) {
 	const payload = `{"message":"Hello, Bob!"}`
+	amqpUri := getAmqpUriFromEnv(t)
+
+	setupQueue(t, amqpUri)
 
 	go main()
 
 	secret := getSecretFromEnv(t)
 	sendPayloadToHTTPServer(t, payload, secret)
 
-	amqpUri := getAmqpUriFromEnv(t)
 	msg := consumeMessage(t, amqpUri)
 
 	assert.Equal(t, payload, msg, "expected message body to match")
@@ -98,32 +102,82 @@ func createSignature(message string, secret string) string {
 	return "sha256=" + hex.EncodeToString(h.Sum(nil))
 }
 
-func consumeMessage(t *testing.T, amqpUri string) string {
+// returns an AMQP channel and a cleanup function to close the connection and channel
+func setupAMQPChannel(t *testing.T, amqpUri string) (*amqp.Channel, func()) {
 	conn, err := amqp.Dial(amqpUri)
 	if err != nil {
 		t.Fatalf("Failed to connect to RabbitMQ: %v", err)
 	}
-	defer conn.Close()
+
 	ch, err := conn.Channel()
 	if err != nil {
+		conn.Close()
 		t.Fatalf("Failed to open a channel: %v", err)
 	}
-	defer ch.Close()
 
+	cleanup := func() {
+		ch.Close()
+		conn.Close()
+	}
+
+	return ch, cleanup
+}
+
+// setupQueue declares the exchange, queue, and binding for the test
+// this is required because the webhook-gateway does not create the queue and
+// published messages could be discarded if the queue does not exist
+func setupQueue(t *testing.T, amqpUri string) {
+	ch, cleanup := setupAMQPChannel(t, amqpUri)
+	defer cleanup()
+
+	config := queue.DefaultQueueConfig()
+
+	err := ch.ExchangeDeclare(
+		config.ExchangeName, // name)
+		"direct",            // type
+		true,                // durable
+		false,               // auto-deleted
+		false,               // internal
+		false,               // no-wait
+		nil,                 // arguments
+	)
+
+	if err != nil {
+		t.Fatalf("Failed to declare an exchange: %v", err)
+	}
 	_, err = ch.QueueDeclare(
-		"webhook-queue", // name
-		true,            // durable
-		false,           // delete when unused
-		false,           // exclusive
-		false,           // no-wait
-		nil,             // arguments
+		config.QueueName, // name
+		true,             // durable
+		false,            // delete when unused
+		false,            // exclusive
+		false,            // no-wait
+		nil,              // arguments
 	)
 
 	if err != nil {
 		t.Fatalf("Failed to declare a queue: %v", err)
 	}
 
-	deliveryChan, err := ch.Consume("webhook-queue", "", true, false, false, false, nil)
+	err = ch.QueueBind(
+		config.QueueName,    // queue name
+		config.RoutingKey,   // routing key
+		config.ExchangeName, // exchange
+		false,
+		nil,
+	)
+
+	if err != nil {
+		t.Fatalf("Failed to bind a queue: %v", err)
+	}
+}
+
+func consumeMessage(t *testing.T, amqpUri string) string {
+	ch, cleanup := setupAMQPChannel(t, amqpUri)
+	defer cleanup()
+
+	config := queue.DefaultQueueConfig()
+
+	deliveryChan, err := ch.Consume(config.QueueName, "", true, false, false, false, nil)
 
 	if err != nil {
 		t.Fatalf("Failed to register a consumer: %v", err)

internal/planner/consumer.go

@@ -32,9 +32,9 @@ const (
 	maxBackoff           = 5 * time.Minute
 )
 
-func NewJobConsumer(amqpUri, queueName string, db JobDatabase, metrics *Metrics) *JobConsumer {
+func NewJobConsumer(consumer queue.Consumer, db JobDatabase, metrics *Metrics) *JobConsumer {
 	return &JobConsumer{
-		consumer: queue.NewAmqpConsumer(amqpUri, queueName),
+		consumer: consumer,
 		db:       db,
 		metrics:  metrics,
 	}

internal/queue/config.go

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2026 Canonical Ltd.
+ * See LICENSE file for licensing details.
+ */
+
+package queue
+
+// Default AMQP configuration values.
+const (
+	defaultExchangeName = "github-workflow-jobs"
+	defaultQueueName    = "github-workflow-jobs"
+	defaultRoutingKey   = "workflow-job"
+)
+
+// QueueConfig holds AMQP exchange, queue, and routing configuration.
+type QueueConfig struct {
+	ExchangeName string
+	QueueName    string
+	RoutingKey   string
+}
+
+// DefaultQueueConfig returns a QueueConfig with default values.
+func DefaultQueueConfig() QueueConfig {
+	return QueueConfig{
+		ExchangeName: defaultExchangeName,
+		QueueName:    defaultQueueName,
+		RoutingKey:   defaultRoutingKey,
+	}
+}

internal/queue/consumer.go

@@ -18,19 +18,19 @@ import (
 
 // AmqpConsumer is an AMQP consumer for workflow job events.
 type AmqpConsumer struct {
-	client    *Client
-	queueName string
-	channel   <-chan amqp.Delivery
+	client  *Client
+	config  QueueConfig
+	channel <-chan amqp.Delivery
 }
 
 // NewAmqpConsumer creates a new AmqpConsumer with the given dependencies.
-func NewAmqpConsumer(uri, queueName string) *AmqpConsumer {
+func NewAmqpConsumer(uri string, config QueueConfig) *AmqpConsumer {
 	return &AmqpConsumer{
-		queueName: queueName,
 		client: &Client{
 			uri:         uri,
 			connectFunc: amqpConnect,
 		},
+		config: config,
 	}
 }
 
@@ -81,13 +81,13 @@ func (c *AmqpConsumer) ensureChannel() error {
 	}
 
 	channel, err := c.client.amqpChannel.Consume(
-		c.queueName, // queue
-		"",          // consumer tag, empty string means a unique random tag will be generated
-		false,       // whether rabbitmq auto-acknowledges messages
-		true,        // whether only this consumer can access the queue
-		false,       // no-local
-		false,       // false means wait for server confirmation that consumer is registered
-		nil,         // args
+		c.config.QueueName, // queue
+		"",                 // consumer tag, empty string means a unique random tag will be generated
+		false,              // whether rabbitmq auto-acknowledges messages
+		true,               // whether only this consumer can access the queue
+		false,              // no-local
+		false,              // false means wait for server confirmation that consumer is registered
+		nil,                // args
 	)
 	if err != nil {
 		return fmt.Errorf("cannot consume amqp messages: %w", err)
@@ -106,7 +106,22 @@ func (c *AmqpConsumer) ensureAmqpChannel() error {
 	}
 
 	if c.client.amqpChannel == nil || c.client.amqpChannel.IsClosed() {
-		err := c.client.resetChannel(c.queueName, false)
+		err := c.client.resetChannel()
+		if err != nil {
+			return err
+		}
+
+		err = c.client.declareExchange(c.config.ExchangeName)
+		if err != nil {
+			return err
+		}
+
+		err = c.client.declareQueue(c.config.QueueName)
+		if err != nil {
+			return err
+		}
+
+		err = c.client.bindQueue(c.config.QueueName, c.config.RoutingKey, c.config.ExchangeName)
 		if err != nil {
 			return err
 		}