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.

Per-Request Options (Do)

AuthenticatedClient.Do accepts a variadic ...DoOption for tailoring a single request without changing client-level configuration:

• WithContentType(ct string): override the Content-Type. Default is application/json when a body is present, none when bodyless. Pass "" to suppress the header even when sending a body.
• WithHeader(key, value string): attach an additional header. Caller wins over any default the client would set for that key:
  • Reserved keys (Authorization, Content-Type) use Set semantics — repeated calls collapse to the last value, and an empty value suppresses the header on the wire (parity with WithContentType("")).
  • All other keys use Add semantics — repeated calls accumulate, so multiple values arrive on the wire.
  • WithContentType and WithHeader("Content-Type", …) write to the same slot, so interleaving them honors true last-option-wins regardless of order.
• WithAuthStaleStatuses(codes ...int): replace (not extend) the set of statuses that signal a stale auth token. Default is {401, 403}. Pass (401, 403, 404) to handle PocketBase's rule-check 404 quirk where an expired token surfaces as 404. Pass no arguments to disable stale-token detection entirely.

When a response status is in the stale-auth set, Do clears the cached token and, if WithRetry(>=1, …) is configured, re-auths and retries the request (bounded by maxRetries). With the default maxRetries == 0 the stale response is returned to the caller as-is, exactly as before.

If the caller supplies their own Authorization via WithHeader("Authorization", …) — including an empty value to suppress the bearer — stale-auth handling is skipped: the cached client token is not cleared and Do does not retry on a stale-auth status, since the failing request did not use the client's credentials. Other retry classes (network errors, 429) continue to apply.

resp, err := authed.Do(ctx, http.MethodPost, "/api/collections/widgets/records", body,
    pbclient.WithContentType("application/xml"),
    pbclient.WithHeader("X-Trace-Id", "abc123"),
    pbclient.WithAuthStaleStatuses(401, 403, 404),
)

Implementing AuthenticatedClient

If you implement AuthenticatedClient yourself (e.g., in a test adapter), call pbclient.PrepareRequest(req, hasBody, opts...) after setting Authorization to apply the same Content-Type and header semantics as the canonical Do. WithAuthStaleStatuses only takes effect inside the canonical authenticatedClient.Do retry loop; custom adapters with no token-refresh lifecycle should treat it as a no-op.

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).

Caveat about DoOptions on testpb clients. Clients returned by Server.ClientWithToken, Server.ClientForRecord, and Server.SuperuserClient honor WithContentType and WithHeader on the wire (via pbclient.PrepareRequest), but treat WithAuthStaleStatuses as a no-op — the test adapter has no token-refresh lifecycle, so there is nothing to retry against.

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.1.0

• AuthenticatedClient.Do now takes a variadic ...DoOption parameter. Interface change: anyone who implements AuthenticatedClient (e.g., custom test mocks) must widen their Do signature. Direct callers are source-compatible because the new parameter is variadic.
• New options: WithContentType, WithHeader, WithAuthStaleStatuses. New helper PrepareRequest for external implementers of the interface.
• Behavior change: a response whose status is in the stale-auth set (default {401, 403}) now triggers an in-call retry when WithRetry(>=1, …) is configured — the token is cleared, re-auth runs, and the request is re-issued (bounded by maxRetries). Callers using the default maxRetries == 0 see no behavior change: the stale response is still returned as-is.

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.