pubsub

pubsub

A Google Cloud Pub/Sub queue backend for Slack Manager. Provides FIFO-ordered message publishing and subscription with automatic visibility extension, and a webhook handler for forwarding HTTP callbacks to Pub/Sub topics.

Part of the slackmgr/plugins monorepo. Versioned independently using the pubsub/vX.Y.Z tag convention.

Installation

go get github.com/slackmgr/plugins/pubsub

Requires Go 1.25+.

Usage

Client

The Client publishes messages to a Pub/Sub topic and optionally receives messages from a subscription. Messages are ordered by OrderingKey (Slack channel ID) and deduplicated via a dedup_id message attribute.

import (
    "cloud.google.com/go/pubsub/v2"
    pubsubplugin "github.com/slackmgr/plugins/pubsub"
    "github.com/slackmgr/types"
)

gcpClient, err := pubsub.NewClient(ctx, projectID)
if err != nil {
    log.Fatal(err)
}

client, err := pubsubplugin.New(gcpClient, "projects/my-project/topics/my-topic", "my-subscription", logger)
if err != nil {
    log.Fatal(err)
}

client, err = client.Init()
if err != nil {
    log.Fatal(err)
}
defer client.Close()

// Publish a message
err = client.Send(ctx, channelID, dedupID, body)

// Receive messages
sinkCh := make(chan *types.FifoQueueItem)
go func() {
    if err := client.Receive(ctx, sinkCh); err != nil && !errors.Is(err, context.Canceled) {
        log.Fatal(err)
    }
}()

for item := range sinkCh {
    // process item...
    item.Ack()
}

Init validates options and configures the publisher and subscriber. Receive is a blocking loop that must be run in a goroutine; it closes sinkCh when it returns.

Call Ack() on a message to acknowledge it. Call Nack() to abandon it — the message will be redelivered by Pub/Sub.

Webhook Handler

WebhookHandler converts incoming HTTP webhook callbacks into Pub/Sub messages. It caches publishers per topic and validates topic names against the GCP resource name format.

handler := pubsubplugin.NewWebhookHandler(gcpClient, true /* isOrdered */)

handler, err = handler.Init(ctx)
if err != nil {
    log.Fatal(err)
}
defer handler.Close()

// ShouldHandleWebhook returns true for valid GCP topic resource names
// e.g. "projects/my-project/topics/my-topic"
if handler.ShouldHandleWebhook(ctx, target) {
    if err := handler.HandleWebhook(ctx, target, webhookData, logger); err != nil {
        log.Printf("webhook error: %v", err)
    }
}

When isOrdered is true, the Slack channel ID is used as the OrderingKey. When false, messages are published without ordering.

Configuration

All options are provided via With* constructor functions passed to New or NewWebhookHandler.

Publisher options

Subscriber options

Testing option

Topic name format

ShouldHandleWebhook validates topic names against the GCP Pub/Sub resource name format:

projects/<project-id>/topics/<topic-name>

• Project ID: 6–30 lowercase alphanumeric characters, hyphens, or colons (domain-prefixed)
• Topic name: 3–255 characters starting with a letter, followed by letters, digits, dots, underscores, or hyphens

License

This project is licensed under the MIT License — see the LICENSE file for details.

Copyright (c) 2026 Peter Aglen