Initial couch library by samlown · Pull Request #1 · invopop/couch

samlown · 2026-07-08T16:34:51Z

A small, self-contained Go wrapper around the kivik CouchDB driver — connection config, models with timestamps, persistence helpers, design-doc sync, a change-feed consumer, and year-based sharding.

Packages

couch (root) — Config/Client (prefixed DB names, ping-with-retry, create, design sync), Model/Document (embeddable base docs implementing Persistable), Store/Fetch/Delete (timestamp + revision handling; errors map to couch.ErrNotFound / couch.ErrAlreadyExists), Design/View, and ShardByYear sharding.
couch/changes — resumable, worker-pooled _changes feed consumer.
couch/at — millisecond-precision timestamps used by couch.Model.

Notes

No heavy dependencies — direct deps are just kivik, backoff, zerolog, testify, x/sync (google/uuid is only transitive via kivik). In particular there is no gobl dependency: ShardByYear uses a small in-package UUID timestamp decoder for versions 1/6/7 instead.
Tests cover the persistence helpers, model/document/config behaviour, the change-feed, and the shard decoder against RFC 9562 UUID vectors. go build, go vet, go test ./..., and gofmt are clean.
Licensed Apache 2.0.

Intended as the shared CouchDB layer for public services (e.g. gobl.lookup) that can't depend on the private internal libraries.

🤖 Generated with Claude Code

A small, self-contained wrapper around the kivik CouchDB driver: - couch.Config / couch.Client — connection config with prefixed database names, ping-with-retry, database creation, design-doc sync. - couch.Model / couch.Document — embeddable base documents (_id/_rev, optional created_at/updated_at) implementing the Persistable contract. - couch.Store / Fetch / Delete — persistence helpers that stamp timestamps and track revisions; errors map to couch.ErrNotFound / ErrAlreadyExists. - couch.Design / View — design documents synced only when their checksum changes. - couch/changes — resumable, worker-pooled CouchDB _changes consumer. - couch/at — millisecond-precision timestamps used by couch.Model. - Sharding (ShardByYear) using a dependency-free UUID timestamp decoder (versions 1, 6 and 7). Only public dependencies (kivik, backoff, zerolog, x/sync). Apache 2.0. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Copilot

Pull request overview

Introduces an initial, self-contained Go library (github.com/invopop/couch) that wraps the Kivik CouchDB driver with higher-level primitives (configuration, persistence helpers, models/documents with timestamps, design-doc syncing, change-feed consumption, and year-based sharding) plus accompanying tests and documentation.

Changes:

Add core couch package: config/client, persistence helpers, base document/model types, design docs, and sharding helpers.
Add couch/changes package: resumable _changes feed consumer and a worker pool for parallel processing.
Add couch/at package: millisecond-precision timestamp types and JSON (un)marshalling helpers, plus README/license/module metadata.

Reviewed changes

Copilot reviewed 25 out of 26 changed files in this pull request and generated 9 comments.

Show a summary per file

File	Description
shard.go	Adds sharding interfaces and `Shards` helper for shard-to-DB routing.
shard_test.go	Tests sharding wrapper behavior with `ShardByYear`.
shard_by_year.go	Implements year-based sharding via UUID timestamp decoding (v1/v6/v7).
shard_by_year_test.go	Tests shard key resolution and UUID timestamp decoding behavior.
README.md	Project overview and usage examples.
persist.go	Adds `Fetch`/`Store`/`Delete` persistence helpers and error mapping.
model.go	Adds `Model` base type with timestamps and revision helpers.
model_test.go	Tests `Model` behavior and `RevAfter`.
LICENSE	Adds Apache 2.0 license text.
go.sum	Adds module checksums for dependencies.
go.mod	Declares module path, Go version, and dependencies.
document.go	Adds `Document` base type without timestamps.
document_test.go	Tests `Document` behavior.
design.go	Adds design-document modeling and checksum-based sync logic.
design_test.go	Tests design instantiation and checksum behavior.
couch.go	Adds `Client` wrapper: connect, ping, design sync, DB create.
config.go	Adds connection configuration and DB name prefixing.
config_test.go	Tests config URL and DB naming behavior.
changes/workerpool.go	Adds worker pool for dispatching change-feed items (flat/hashed).
changes/workerpool_test.go	Tests worker pool delivery, hashing, retry, fatal handling, stop semantics.
changes/options.go	Adds feed options (filters, heartbeats, persistence tuning, include_docs, etc.).
changes/feed.go	Adds resumable feed implementation with persistence and fatal-on-conflict behavior.
changes/feed_test.go	Unit tests for feed in-memory behavior and options.
changes/changes.go	Package doc stub for `changes`.
at/at.go	Adds millisecond timestamp types with JSON (un)marshalling.
at/at_test.go	Tests timestamp parsing/marshalling and model embedding behavior.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

- lint: golangci-lint on push/PR. - test: go test -race on push. - release: auto-tag a semver version on merge to main. All dependencies are public, so no private-module credentials are needed. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Copilot

Pull request overview

Copilot reviewed 28 out of 29 changed files in this pull request and generated 9 comments.

- couch.Ping: only report a healthy connection when the underlying call returns no error; retry transport/network errors (status 0) instead of falsely succeeding. - ShardByYear.generateList: iterate newest-first so a start year beyond next year yields no shards rather than a negative-length panic. - Design checksum: skip nil *View entries. - couch.Fetch: validate the ID is set (matches Store/Delete). - ShardRules.Template: document that it must be an fmt format string with a %s shard placeholder. - Tests for the future-start shard guard and the nil-view checksum. - Refresh to the latest dependencies (go 1.25). Left as-is: the changes-feed goroutine lifecycle comments (channel ownership on Stop, save-timer channel). This is synchronisation carried over verbatim from the battle-tested source and is best revisited as a focused, separately-reviewed change rather than reworked here. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

samlown · 2026-07-08T16:51:14Z

Thanks Copilot — addressed in the latest commit:

Fixed

couch.Ping — only returns healthy when the underlying call returns no error; transport/network errors (status 0) now retry instead of falsely reporting success.
ShardByYear.generateList — iterates newest-first, so a start beyond next year yields no shards instead of a negative-length panic (test added).
Design checksum — skips nil *View entries (test added).
couch.Fetch — validates the ID is set, matching Store/Delete.
ShardRules.Template — documented that it must be an fmt format string with a %s placeholder.

Intentionally not changed

go.mod go 1.25 — kept, on the latest dependencies (maintainer preference; consumers are on 1.25).
WorkerPool.recv bounds — workerID is contractually supplied from the pool's own worker range; a guard would only mask caller bugs.
changes/feed.go Stop/outgoing ownership and the save-timer channel — this is goroutine-lifecycle synchronisation carried over verbatim from the battle-tested source. The comments are plausible, but reworking channel ownership here is risky and best done as a focused, separately-reviewed change (ideally shared with the original implementation) rather than in this import.

Copilot

Pull request overview

Copilot reviewed 28 out of 29 changed files in this pull request and generated 7 comments.

- couch.Ping: exact attempt count, return ctx.Err() on cancellation, and wrap the last ping error in the final message. - changes/feed: nextItem now selects on ctx.Done() so a cancelled context returns promptly instead of blocking. - changes/feed: fix the save-timer data race — the timer callback no longer mutates f.delay and sends non-blockingly on a buffered channel; the consumer clears the timer when it observes the notification (so a timeout-triggered save is no longer skipped). - CI: run tests on pull_request too, not only push. - README: don't fall through to log.Fatal on the handled not-found case. - Comment typos (independent, stores). Deferred: reworking Stop()'s ownership of the outgoing channel (the "send on closed channel" shutdown race). That needs a coordinated done-channel refactor with integration coverage of the connect/processNext lifecycle — none exists yet — so it's unsafe to do blind here and is better as a focused, separately-tested change shared with the upstream source. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

samlown · 2026-07-08T17:04:16Z

Round 3 addressed:

Fixed

couch.Ping — exact attempt count, returns ctx.Err() on cancellation, and wraps the last ping error in the final message.
changes/feed save-timer data race — the timer callback no longer mutates f.delay; it does a non-blocking send on a now-buffered channel, and the consumer clears the timer when it observes the notification. This removes the race and fixes the case where a timeout-triggered save could be skipped (f.delay left non-nil).
changes/feed.nextItem — now selects on ctx.Done(), so a cancelled/expired context returns promptly instead of blocking.
CI — tests now run on pull_request as well as push.
README — the not-found branch no longer falls through to log.Fatal.
Comment typos (independent, stores).

Deferred (intentionally)

Stop() closing outgoing while processNext may still be sending ("send on closed channel" under a shutdown race). The correct fix is a coordinated done-channel refactor where the producer owns the channel close — but the connect/processNext/outgoing lifecycle has no integration test here, and this is verbatim production-proven code. Reworking shutdown ordering blind risks a worse failure (a Stop deadlock). I'd rather do it as a focused change with a fake changes-source test — ideally shared with the upstream implementation. Happy to take that on as a follow-up if you'd like.

go test -race, go vet, golangci-lint, and gofmt are all clean; CI green.

Copilot

Pull request overview

Copilot reviewed 28 out of 29 changed files in this pull request and generated 4 comments.

+		if err != nil {
+			dur := bo.Duration()
+			log.Error().Err(err).Str("shard", shard).Dur("wait", dur).Msg("change feed read error, will retry after wait")
+			time.Sleep(dur)
+			continue
+		}


+		var item T
+		for {
+			item, err = p.fetcher(ctx, id)
+			if err == nil {
+				break
+			}
+			dur := bo.Duration()
+			log.Error().Err(err).Str("shard", shard).Str("id", id).Dur("wait", dur).Msg("fetch error, will retry after wait")
+			time.Sleep(dur)
+		}


+func (f *feed) Start(ctx context.Context) {
+	if f.started {
+		return
+	}
+	f.outgoing = make(chan feedItem)
+	f.started = true
+	go f.connect(ctx)
+
+	if f.opts.callback != nil {
+		// this will block
+		f.startWithCallbacks()
+	}
+}
+
+// Stop stops the connection as gracefully as possible.
+func (f *feed) Stop() {
+	f.stopOnce.Do(func() {
+		f.stopped = true
+		// Always close fatal so any waiting consumer is released, even
+		// if Start was never called or no fatal condition occurred.
+		f.fatalOnce.Do(func() { close(f.fatal) })
+		if !f.started {
+			return
+		}
+		f.stopSaveTimer()
+		if f.source != nil {
+			// Kivik won't allow feed closure while a next call is
+			// blocking.
+			go func() {
+				_ = f.source.Close()
+				f.source = nil
+			}()
+		}
+		close(f.outgoing)
+		f.started = false
+		f.log.Info().Msg("stopped")
+	})


+	// Tighten the backoff for the test so the retry happens quickly.
+	pool.Start(context.Background())


Copilot

Pull request overview

Copilot reviewed 28 out of 29 changed files in this pull request and generated 7 comments.

+			}()
+		}
+		close(f.outgoing)
+		f.started = false


+func (f *feed) shouldSave() bool {
+	return f.count != 0 && (f.delay == nil || f.count > f.opts.storeLimit)
+}


+// WithSuffix appends the provided strings to the change feed name
+func WithSuffix(s ...string) Option {
+	return func(opts *options) {
+		opts.suffix = s
+	}
+}


+		if err != nil {
+			dur := bo.Duration()
+			log.Error().Err(err).Str("shard", shard).Dur("wait", dur).Msg("change feed read error, will retry after wait")
+			time.Sleep(dur)
+			continue
+		}


+			if err == nil {
+				break
+			}
+			dur := bo.Duration()
+			log.Error().Err(err).Str("shard", shard).Str("id", id).Dur("wait", dur).Msg("fetch error, will retry after wait")
+			time.Sleep(dur)
+		}


+func NewHashedWorkerPool[T any](feeds map[string]Feed, fetcher Fetcher[T], workers int, keyFn KeyFunc[T]) *WorkerPool[T] {
+	if workers < 1 {
+		workers = 1
+	}
+	chs := make([]chan T, workers)


+	c, _ := couch.New(couch.NewConfig("test")) // nolint:errcheck
+	sr := couch.NewShardByYear("foo", 2020)


Copilot AI review requested due to automatic review settings July 8, 2026 16:34

Copilot started reviewing on behalf of samlown July 8, 2026 16:35 View session