Writing a storage backend#

A hub keeps subscription state behind SubscriptionStore. The default is in memory, which is right for development and single process hubs and wrong for anything that has to survive a restart.

type SubscriptionStore interface {
	Save(ctx context.Context, sub Subscription) error
	Get(ctx context.Context, topic, callback string) (Subscription, error)
	Delete(ctx context.Context, topic, callback string) error
	ListByTopic(ctx context.Context, topic string) ([]Subscription, error)
	Topics(ctx context.Context) ([]string, error)
	ListExpiring(ctx context.Context, before time.Time) ([]Subscription, error)
}

Switching is one line:

hub, err := websub.NewHub("https://example.com/hub", websub.WithStore(store))

The contract#

A subscription is identified by the pair (topic, callback). Nothing else is a key, and the pair is not unique on either half alone: one subscriber may hold subscriptions to many topics, and one topic may have many subscribers.

• Save replaces. Re-subscribing to an active subscription is legal in both specifications and must override the previous state rather than creating a second row or failing.
• Delete is idempotent. Unsubscribing something that is not there is not an error, because unsubscribe is itself idempotent and a hub may retry.
• Get reports a missing subscription with an error wrapping websub.ErrUnknownSubscription, not a zero value and a nil error.
• ListExpiring is strictly before. A subscription expiring exactly at the given time is not returned. A permanent subscription, meaning one with a zero ExpiresAt, is never returned.
• Everything is safe for concurrent use. A hub calls ListByTopic on every publish and Save on every verified subscription, concurrently.

ListByTopic is the hot path and is the method worth indexing for.

Proving it correct#

Your tests must run the shared suite. It is not a smoke test; it covers the boundary conditions above, which are the ones that are easy to get subtly wrong and hard to notice in production.

package tests

import (
	"testing"

	"github.com/Jazzmoon/websub"
	"github.com/Jazzmoon/websub/storage/storetest"
)

func TestStore(t *testing.T) {
	storetest.Run(t, func(t *testing.T) websub.SubscriptionStore {
		return newStoreForTest(t)
	})
}

Run it under -race. The suite includes concurrent access on purpose.

Testing against the real thing#

Backends in this repo test against a real server, not a fake, using testcontainers. A fake redis or an in-memory stand-in for postgres will not tell you whether your query plan is sane, whether your schema survives a concurrent upsert, or whether your expiry range query does what you think.

This means Docker has to be available to run a backend’s tests. It is not needed to work on core, or on anything else in the repo.

cd storage/postgres && go test ./... -race

Notes per backend#

• redis. A hash per subscription plus a sorted set keyed by expiry, so ListExpiring is a range query rather than a scan. Redis TTLs are deliberately not used for lease expiry: a hub needs to see an expired subscription in order to remove it and report it, and a key that has already evaporated cannot be reported.
• postgres. Plain SQL migrations, no ORM. Indexes on (topic) and (expires_at). Save is an upsert on the (topic, callback) primary key.
• sqlite. The same schema as postgres wherever the dialects allow, so there is one mental model rather than two. Uses a pure Go driver, so it cross compiles and needs no C toolchain. It is meant for single binary and embedded deployments, not high concurrency hubs, where its single writer will be the bottleneck.

Checklist#

1. New module under storage/<name>/ with its own go.mod.
2. Implement websub.SubscriptionStore.
3. Call storetest.Run from your tests, under -race.
4. Add the module to go.work. CI reads its module list from there.
5. Document any setup, such as required migrations, in the module’s own README.
6. Add it to the storage section of the root README.