pbclient

PocketBase Client

Go client library for PocketBase with automatic admin auth, generic repository helpers, and a simple KV store built on collections.

Installation

go get github.com/eqr/pbclient

Quick Start

package main

import (
	"context"
	"log"

	"github.com/eqr/pbclient"
)

type Todo struct {
	ID    string `json:"id"`
	Title string `json:"title"`
	Done  bool   `json:"done"`
}

func main() {
	client, err := pbclient.NewClient("https://your-pocketbase-host")
	if err != nil {
		log.Fatal(err)
	}

	authed, err := client.AuthenticateSuperuser(pbclient.Credentials{
		Email:    "admin@example.com",
		Password: "super-secret",
	})
	if err != nil {
		log.Fatal(err)
	}

	repo := pbclient.NewRepository[Todo](authed, "todos")

	ctx := context.Background()

	created, err := repo.Create(ctx, Todo{Title: "try pbclient"})
	if err != nil {
		log.Fatal(err)
	}

	item, err := repo.Get(ctx, created.ID)
	if err != nil {
		log.Fatal(err)
	}
	log.Printf("got todo: %+v", item)
}

Client Options

• WithHTTPClient(*http.Client): reuse your own transport (e.g., tracing, custom TLS).
• WithTimeout(time.Duration): set HTTP timeout.
• WithRetry(maxRetries, backoff): retry 429/network errors with exponential backoff.
• WithLogger(*slog.Logger): structured logging for auth and retries.

Repository Usage

client, _ := pbclient.NewClient("https://your-pocketbase-host")
authed, _ := client.AuthenticateSuperuser(pbclient.Credentials{Email: "admin@example.com", Password: "super-secret"})
repo := pbclient.NewRepository[Todo](authed, "todos")

// List with filters, sorting, and field selection
todos, err := repo.List(ctx, pbclient.ListOptions{
	Page:   1,
	PerPage: 20,
	Filter: pbclient.And(pbclient.Eq("done", "false"), pbclient.Gt("priority", "3")),
	Sort:   "-created",
	Fields: []string{"id", "title", "done"},
})

// Update an item
updated, err := repo.Update(ctx, created.ID, Todo{Title: "updated title", Done: true})

KV Store Usage

kv := pbclient.NewTypedKVStore[map[string]any](authed, "kv_store", "myapp") // collection defaults to "kv_store"

_ = kv.Set(ctx, "feature_flag", map[string]any{"name": "beta", "enabled": true})

flag, _ := kv.Get(ctx, "feature_flag")

exists, _ := kv.Exists(ctx, "feature_flag")
keys, _ := kv.List(ctx, "feature_")
_ = kv.Delete(ctx, "feature_flag")

Consumer (Message Processing)

The Consumer provides offset-tracked message consumption from PocketBase collections, similar to Kafka consumers. It supports:

• Offset tracking: Automatic tracking of processed messages using timestamps or record IDs
• Consumer groups: Multiple consumers can share progress tracking
• Filtering: Apply PocketBase filters to consume specific messages
• At-least-once delivery: Messages are reprocessed if not committed
• Graceful shutdown: Clean resource cleanup

Basic Consumer Usage

import (
	"github.com/eqr/pbclient"
	"github.com/eqr/pbclient/migrations"
)

// Run migration to create consumer_offsets collection
runner := migrations.NewRunner(authed)
runner.Register(migrations.NewConsumerOffsetsMigration())
runner.Run(ctx)

// Create consumer with timestamp-based offset tracking
consumer, err := pbclient.NewConsumer[Event](authed, "events", pbclient.ConsumerOptions{
	ConsumerGroup:  "event-processor",
	OffsetProvider: pbclient.NewTimestampOffsetProvider(),
	BatchSize:      100,
	PollInterval:   1 * time.Second,
})
defer consumer.Close()

// Start consuming
consumer.Start(ctx)

// Process messages
for {
	select {
	case msg := <-consumer.Messages():
		// Process the message
		fmt.Printf("Processing: %+v\n", msg.Data)
		
		// Commit to mark as processed
		if err := msg.Commit(ctx); err != nil {
			log.Printf("commit failed: %v", err)
		}
		
	case err := <-consumer.Errors():
		log.Printf("consumer error: %v", err)
	}
}

Offset Providers

Two built-in offset providers are available:

Timestamp-based (recommended for most cases):

OffsetProvider: pbclient.NewTimestampOffsetProvider()

• Uses PocketBase's created timestamp field
• Works with any collection without schema changes
• Starting fresh begins from current time

ID-based:

OffsetProvider: pbclient.NewIDOffsetProvider()

• Uses PocketBase's 15-character record IDs
• Lexicographic ordering
• Starting fresh processes all existing records

Consumer Options

• ConsumerGroup (required): Unique identifier for this consumer group
• OffsetProvider (required): Strategy for tracking offsets
• Filter: PocketBase filter to apply (e.g., "type = 'user_action'")
• BatchSize: Number of messages to fetch per poll (default: 100)
• PollInterval: Time between polls (default: 1s)
• BufferSize: Message channel buffer size (default: 100)
• OffsetCollection: Collection name for storing offsets (default: "consumer_offsets")

Consumer Groups

WARNING: Consumer groups provide shared offset tracking, NOT load balancing like Kafka.

Multiple consumers with the same ConsumerGroup will:

• Read from the same offset and see the same messages
• Have potential race conditions when committing offsets
• May process duplicate messages

This pattern is useful for:

• High availability (one consumer can take over if another dies)
• NOT for load balancing (use different consumer groups for independent processing)

// Consumer 1
consumer1, _ := pbclient.NewConsumer[Event](authed, "events", pbclient.ConsumerOptions{
	ConsumerGroup:  "shared-group",
	OffsetProvider: pbclient.NewTimestampOffsetProvider(),
})

// Consumer 2 - shares offset but will process duplicate messages
consumer2, _ := pbclient.NewConsumer[Event](authed, "events", pbclient.ConsumerOptions{
	ConsumerGroup:  "shared-group", // Same group name
	OffsetProvider: pbclient.NewTimestampOffsetProvider(),
})

Examples

See examples/consumer_example.go for complete working examples including:

• Basic message processing
• ID-based offset tracking
• Filtered consumption
• Shared offset tracking (with warnings about race conditions)

Filters

Helpers for PocketBase filter strings:

• Eq/Neq/Gt/Gte/Lt/Lte
• And/Or to combine conditions

Example: pbclient.And(pbclient.Eq("status", "active"), pbclient.Gt("score", "10"))

Error Handling

Common HTTP statuses map to sentinel errors (ErrBadRequest, ErrUnauthorized, ErrForbidden, ErrNotFound, ErrConflict, ErrValidation, ErrRateLimited, ErrServer). Other statuses return *HTTPError with status/message.

Thread Safety

Client is safe for concurrent use; token access is locked and retries respect context cancellation. Repository and KV helpers share the same client and rely on PocketBase for atomicity.

Embedded PocketBase Tests

Use github.com/eqr/pbclient/testpb to spin up embedded PocketBase in tests and get authenticated pbclient.AuthenticatedClient instances.

package myapp_test

import (
	"context"
	"net/http"
	"testing"

	"github.com/eqr/pbclient"
	"github.com/eqr/pbclient/testpb"
)

func TestRepositoryAgainstEmbeddedPocketBase(t *testing.T) {
	pb := testpb.Start(t)
	client := pb.SuperuserClient(t, "test@example.com")

	repo := pbclient.NewRepository[map[string]any](client, "demo2")
	page, err := repo.List(context.Background(), pbclient.ListOptions{PerPage: 1})
	if err != nil {
		t.Fatalf("List: %v", err)
	}
	if len(page.Items) == 0 {
		t.Fatalf("expected seeded records")
	}

	resp, err := client.Do(context.Background(), http.MethodGet, "/api/collections/demo2", nil)
	if err != nil || resp.StatusCode != http.StatusOK {
		t.Fatalf("collection request failed: err=%v status=%v", err, resp.StatusCode)
	}
	resp.Body.Close()
}

testpb.Start supports options:

• testpb.WithDataDir(path) to override the PocketBase test data directory.
• testpb.WithHookBinder(func(app *tests.TestApp) error { ... }) to register hooks before routes are served (github.com/pocketbase/pocketbase/tests).

testpb itself is not integration-tagged, so you can use it in regular tests.

Run tests with:

go test ./...

By default, testpb reuses PocketBase's bundled test data from either vendor/ or module cache. If test data is not present in either location, testpb.Start-based tests are skipped.

Migrations

The migrations package ships an HTTP-based runner. By default it auto-creates a pb_migrations collection with authenticated-only rules and records applied migration names/timestamps. Example:

runner := migrations.NewRunner(client) // auto-creates pb_migrations with auth-only access rules
_ = runner.Register(migrations.MyFirstMigration{})
_ = runner.Run(ctx)

pending, _ := runner.Pending(ctx)
_ = runner.Down(ctx, 1) // roll back latest

To disable auto-creation and require a pre-provisioned collection, pass migrations.WithAutoCreate(false); the runner will return ErrCollectionNotFound if the collection is missing.

License

MIT – see LICENSE for details.

Release Notes (v0.0.1)

• Client with lazy admin auth, retries, and token refresh.
• Generic repository with CRUD, filtering, pagination helpers.
• KV store abstraction on top of collections.
• Filter builders (Eq, Gt, And, etc.).
• PocketBase-aware error mapping.
• Extensive unit tests and an optional integration test.