From 47d5840e93274a053bd63e4dd7d5424949bfa650 Mon Sep 17 00:00:00 2001 From: Sam Lown Date: Wed, 8 Jul 2026 16:34:47 +0000 Subject: [PATCH 1/5] Initial couch library MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- LICENSE | 201 +++++++++++++ README.md | 101 +++++++ at/at.go | 111 +++++++ at/at_test.go | 173 +++++++++++ changes/changes.go | 2 + changes/feed.go | 582 +++++++++++++++++++++++++++++++++++++ changes/feed_test.go | 240 +++++++++++++++ changes/options.go | 171 +++++++++++ changes/workerpool.go | 223 ++++++++++++++ changes/workerpool_test.go | 347 ++++++++++++++++++++++ config.go | 61 ++++ config_test.go | 30 ++ couch.go | 93 ++++++ design.go | 116 ++++++++ design_test.go | 36 +++ document.go | 42 +++ document_test.go | 22 ++ go.mod | 26 ++ go.sum | 53 ++++ model.go | 97 +++++++ model_test.go | 107 +++++++ persist.go | 92 ++++++ shard.go | 89 ++++++ shard_by_year.go | 170 +++++++++++ shard_by_year_test.go | 107 +++++++ shard_test.go | 43 +++ 26 files changed, 3335 insertions(+) create mode 100644 LICENSE create mode 100644 at/at.go create mode 100644 at/at_test.go create mode 100644 changes/changes.go create mode 100644 changes/feed.go create mode 100644 changes/feed_test.go create mode 100644 changes/options.go create mode 100644 changes/workerpool.go create mode 100644 changes/workerpool_test.go create mode 100644 config.go create mode 100644 config_test.go create mode 100644 couch.go create mode 100644 design.go create mode 100644 design_test.go create mode 100644 document.go create mode 100644 document_test.go create mode 100644 go.mod create mode 100644 go.sum create mode 100644 model.go create mode 100644 model_test.go create mode 100644 persist.go create mode 100644 shard.go create mode 100644 shard_by_year.go create mode 100644 shard_by_year_test.go create mode 100644 shard_test.go diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..261eeb9 --- /dev/null +++ b/LICENSE @@ -0,0 +1,201 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/README.md b/README.md index 415359f..c657345 100644 --- a/README.md +++ b/README.md @@ -1 +1,102 @@ # couch + +A small, dependency-light wrapper around the [kivik](https://github.com/go-kivik/kivik) +CouchDB driver that makes connections, models, and design documents easier to +work with. + +```bash +go get github.com/invopop/couch +``` + +## What it gives you + +- **`couch.Config` / `couch.Client`** — configure a connection from parts and + namespace every database behind a prefix. +- **`couch.Model`** — an embeddable base document that manages `_id`, `_rev`, + attachments, and `created_at` / `updated_at` timestamps. `couch.Document` is + the same without timestamps. +- **`couch.Store` / `couch.Fetch` / `couch.Delete`** — persistence helpers that + stamp timestamps, track the revision, and map errors to `couch.ErrNotFound` / + `couch.ErrAlreadyExists`. +- **`couch.Design` / `couch.View`** — declare design documents and sync them + idempotently (only rewritten when their views/filters change). +- **`couch/changes`** — consume CouchDB `_changes` feeds with a resumable, + persisted cursor and a worker pool. +- **`couch/at`** — millisecond-precision timestamps used by `couch.Model`. + +## Usage + +```go +package main + +import ( + "context" + "errors" + "log" + + "github.com/invopop/couch" +) + +// Embed couch.Model to get _id/_rev + created_at/updated_at for free. +type Widget struct { + couch.Model + Name string `json:"name"` +} + +func main() { + ctx := context.Background() + + conf := couch.NewConfig("myapp") // databases are namespaced as myapp_ + conf.Host = "localhost" + conf.Username, conf.Password = "admin", "secret" + + client, err := couch.New(conf) + if err != nil { + log.Fatal(err) + } + if err := client.Ping(ctx); err != nil { + log.Fatal(err) + } + + db := client.DB("widgets") // resolves to "myapp_widgets" + if err := client.Create(ctx, db); err != nil { + log.Fatal(err) + } + + w := &Widget{Name: "gadget"} + w.SetID("widget-1") + if err := couch.Store(ctx, db, w); err != nil { // sets timestamps + _rev + log.Fatal(err) + } + + got := &Widget{} + got.SetID("widget-1") + if err := couch.Fetch(ctx, db, got); err != nil { + if errors.Is(err, couch.ErrNotFound) { + log.Println("not found") + } + log.Fatal(err) + } +} +``` + +### Design documents + +```go +d := couch.NewDesign("widgets") +d.SetView("by_name", &couch.View{ + Map: `function(doc) { if (doc.name) { emit(doc.name, null); } }`, +}) +if err := client.SyncDesigns(ctx, db, []*couch.Design{d}); err != nil { + log.Fatal(err) +} +``` + +### Change feeds + +See [`changes`](./changes) for consuming a database's `_changes` feed with a +resumable cursor. Sharding helpers (`ShardByYear`, …) live in the root package. + +## License + +Apache 2.0 — see [LICENSE](./LICENSE). diff --git a/at/at.go b/at/at.go new file mode 100644 index 0000000..4abad60 --- /dev/null +++ b/at/at.go @@ -0,0 +1,111 @@ +// Package at provides timestamp handling with millisecond precision. +package at + +import ( + "fmt" + "time" +) + +// Millisecond time formats to comply with W3C datetime format that +// contains rules for local timezones so that they always include a ":". +const ( + RFC3339Milli string = "2006-01-02T15:04:05.000Z" + RFC3339MilliWithZone string = "2006-01-02T15:04:05.000-07:00" +) + +const ( + nullString = "null" +) + +// Timestamp represents the basic time wrapper to be used in timestamps +// with millisecond precision. +type Timestamp struct { + time.Time +} + +// LocalTime ensures the local time information is included in the timestamp. +type LocalTime struct { + time.Time +} + +// Now provides a timestamp for the current UTC system Time +func Now() Timestamp { + return Timestamp{time.Now().UTC()} +} + +// LocalTimeNow is used to provide the current time in the provided location. +func LocalTimeNow(loc *time.Location) LocalTime { + return LocalTime{time.Now().In(loc)} +} + +// ParseTimestamp attempts to parse the timestamp string. +func ParseTimestamp(str string) (Timestamp, error) { + // parse with generic RFC3339 precision, which supports milliseconds + // and helps us get around issues around timestamps that don't + // include the milliseconds for whatever reason. + o, err := time.Parse(time.RFC3339, str) + if err != nil { + return Timestamp{}, fmt.Errorf("at: unable to parse timestamp: %w", err) + } + return Timestamp{o.UTC()}, nil +} + +// ParseLocalTime attempts to read in the provided time data which hopefully includes +// a zone, but is not necessarily guaranteed. +func ParseLocalTime(str string) (LocalTime, error) { + o, err := time.Parse(time.RFC3339, str) + if err != nil { + return LocalTime{}, fmt.Errorf("at: unable to parse localtime: %w", err) + } + return LocalTime{o}, nil +} + +// String provides the timestamp in RFC3339 format including milliseconds. +func (t *Timestamp) String() string { + return t.Format(RFC3339Milli) +} + +// String provides the local time including milliseconds and a time zone. +func (t *LocalTime) String() string { + return t.Format(RFC3339MilliWithZone) +} + +// UnmarshalJSON uses our timestamp parser. +func (t *Timestamp) UnmarshalJSON(data []byte) error { + s := string(data) + if s == nullString { + return nil + } + s = s[1 : len(s)-1] // no quotes + var err error + *t, err = ParseTimestamp(s) + return err +} + +// MarshalJSON provides the timestamp in JSON format +func (t Timestamp) MarshalJSON() ([]byte, error) { + if t.IsZero() { + return []byte(nullString), nil + } + return []byte(`"` + t.String() + `"`), nil +} + +// UnmarshalJSON parses the provided local time JSON data. +func (t *LocalTime) UnmarshalJSON(data []byte) error { + s := string(data) + if s == nullString { + return nil + } + s = s[1 : len(s)-1] // no quotes + var err error + *t, err = ParseLocalTime(s) + return err +} + +// MarshalJSON provides the local time in JSON format. +func (t LocalTime) MarshalJSON() ([]byte, error) { + if t.IsZero() { + return []byte(nullString), nil + } + return []byte(`"` + t.String() + `"`), nil +} diff --git a/at/at_test.go b/at/at_test.go new file mode 100644 index 0000000..2cba447 --- /dev/null +++ b/at/at_test.go @@ -0,0 +1,173 @@ +package at_test + +import ( + "encoding/json" + "strings" + "testing" + "time" + + "github.com/invopop/couch/at" +) + +func TestTimestampUnmarshal(t *testing.T) { + var cases = []struct { + Given string + Expected time.Time + Error bool + }{ + // long form + {`"2009-11-10T23:19:45.123Z"`, time.Date(2009, time.November, 10, 23, 19, 45, 123000000, time.UTC), false}, + // short form + {`"2009-11-10T13:19:04Z"`, time.Date(2009, time.November, 10, 13, 19, 4, 0, time.UTC), false}, + // local form + {`"2009-11-10T13:19:04+02:00"`, time.Date(2009, time.November, 10, 11, 19, 4, 0, time.UTC), false}, + // bad form + {`"Z2009-11-10T13:19:04Z"`, time.Time{}, true}, + // nil string + {`null`, time.Time{}, false}, + } + + for _, c := range cases { + payload := []byte(c.Given) + var output at.Timestamp + if err := json.Unmarshal(payload, &output); err != nil && !c.Error { + t.Error(err) + continue + } + if !output.Equal(c.Expected) { + t.Errorf("Expected: %q, Given: %q", c.Expected, output) + } + } + + // always convert to UTC + var output at.Timestamp + if err := json.Unmarshal([]byte(`"2009-11-10T13:19:04+02:00"`), &output); err != nil { + t.Error(err) + return + } + if z, _ := output.Zone(); z != "UTC" { + t.Errorf("Expected UTC time zone, got: %q", z) + } +} + +func TestTimestampMarshal(t *testing.T) { + var cases = []struct { + Given time.Time + Expected string + }{ + {time.Date(2009, time.November, 10, 23, 19, 45, 0, time.UTC), `"2009-11-10T23:19:45.000Z"`}, + {time.Date(2009, time.November, 10, 13, 19, 4, 0, time.UTC), `"2009-11-10T13:19:04.000Z"`}, + {time.Date(2009, time.November, 10, 23, 19, 45, 123456000, time.UTC), `"2009-11-10T23:19:45.123Z"`}, + {time.Time{}, `null`}, + } + + for _, c := range cases { + ct := at.Timestamp{c.Given} + output, err := json.Marshal(ct) + if err != nil { + t.Error(err) + } + if string(output) != c.Expected { + t.Errorf("Expected: %q, Given: %q", c.Expected, output) + } + } +} + +func TestNow(t *testing.T) { + ct := at.Now() + if z, _ := ct.Zone(); z != "UTC" { + t.Errorf("Failed to get current time in UTC, got: %v", z) + } +} + +func TestLocalTimeNow(t *testing.T) { + loc, _ := time.LoadLocation("America/Lima") + ct := at.LocalTimeNow(loc) + if z, _ := ct.Zone(); z != "-05" { + t.Errorf("Failed to get expected time zone, got: %v", z) + } +} + +func TestLocalTimeUnmarshal(t *testing.T) { + tl, _ := time.LoadLocation("America/Lima") // always -5 (no DST) + tl2, _ := time.LoadLocation("Asia/Dubai") // always +4 (no DST) + var cases = []struct { + Given string + Expected time.Time + Error bool + }{ + // long form + {`"2009-11-10T23:19:45.123-05:00"`, time.Date(2009, time.November, 10, 23, 19, 45, 123000000, tl), false}, + // long form 2 + {`"2009-11-10T23:19:45.123+04:00"`, time.Date(2009, time.November, 10, 23, 19, 45, 123000000, tl2), false}, + // short form + {`"2009-11-10T23:19:45-05:00"`, time.Date(2009, time.November, 10, 23, 19, 45, 0, tl), false}, + // bad form + {`"Z2009-11-10T13:19:04Z"`, time.Time{}, true}, + // bad long form + {`"2009-11-10T23:19:45.123+0400"`, time.Time{}, true}, + // nil string + {`null`, time.Time{}, false}, + // UTC form + {`"2009-11-10T23:19:45.123Z"`, time.Date(2009, time.November, 10, 23, 19, 45, 123000000, time.UTC), false}, + } + + for _, c := range cases { + payload := []byte(c.Given) + var output at.LocalTime + if err := json.Unmarshal(payload, &output); err != nil && !c.Error { + t.Error(err) + continue + } + if !output.Equal(c.Expected) { + t.Errorf("Expected: %q, Given: %q", c.Expected, output) + } + } +} + +func TestLocalTimeMarshal(t *testing.T) { + tl, _ := time.LoadLocation("America/Lima") // always -5 (no DST) + tl2, _ := time.LoadLocation("Asia/Dubai") // always +4 (no DST) + var cases = []struct { + Given time.Time + Expected string + }{ + {time.Date(2009, time.November, 10, 23, 19, 30, 0, tl), `"2009-11-10T23:19:30.000-05:00"`}, + {time.Date(2009, time.November, 10, 13, 19, 4, 123000000, tl), `"2009-11-10T13:19:04.123-05:00"`}, + {time.Date(2009, time.November, 10, 13, 19, 4, 123000000, tl2), `"2009-11-10T13:19:04.123+04:00"`}, + {time.Time{}, `null`}, + } + + for _, c := range cases { + ct := at.LocalTime{c.Given} + output, err := json.Marshal(ct) + if err != nil { + t.Error(err) + } + if string(output) != c.Expected { + t.Errorf("Expected: %q, Given: %q", c.Expected, output) + } + } +} + +func TestTimestampInModel(t *testing.T) { + type tmodel struct { + Value string `json:"v"` + ExampleAt at.Timestamp `json:"example_at"` + EmptyAt *at.Timestamp `json:"empty_at,omitempty"` + } + x := new(tmodel) + x.Value = "bar" + + data, err := json.Marshal(x) + if err != nil { + t.Error(err) + return + } + if !strings.Contains(string(data), `"example_at":null`) { + t.Errorf("Expected output to contain value example_at, got: %v", string(data)) + } + if strings.Contains(string(data), "empty_at") { + t.Errorf("Did not expect output to contain value, got: %v", string(data)) + } +} diff --git a/changes/changes.go b/changes/changes.go new file mode 100644 index 0000000..676be34 --- /dev/null +++ b/changes/changes.go @@ -0,0 +1,2 @@ +// Package changes makes it easier to listen to CouchDB change feeds. +package changes diff --git a/changes/feed.go b/changes/feed.go new file mode 100644 index 0000000..fa9a28b --- /dev/null +++ b/changes/feed.go @@ -0,0 +1,582 @@ +// Package changes makes it easier to listen to CouchDB change feeds. +package changes + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "github.com/go-kivik/kivik/v4" + "github.com/invopop/couch" + "github.com/jpillora/backoff" + "github.com/rs/zerolog" + "github.com/rs/zerolog/log" + "golang.org/x/sync/errgroup" +) + +const ( + // conflictFatalThreshold is the number of consecutive 409 conflicts on + // the per-feed sequence document that will trigger a fatal signal when + // WithFatalOnConflict is enabled. Three is high enough to absorb a + // transient race during a rolling restart but low enough to surface a + // genuine duplicate-consumer situation quickly. + conflictFatalThreshold = 3 +) + +// Feed controls the lifecycle of a connection to a couchdb changes +// feed and ensures that synchronisation can continue by maintaining +// a reference to the latest sequence ID. +// +// To use, instantiate with New and run the Start method for a +// connection to be established in the background. +// +// Call the Next method to get each of the changed document IDs +// from the source feed. Next acts as an ack method, and is designed +// so that every subsequent call to Next can potentially save the +// current state of the feed. +// +// Feeds are persisted to the changes database either when the configured +// store limit is reached, or when the configured store timeout is met. +// Defaults are DefaultStoreLimit and DefaultStoreTimeout; both are +// overridable per-feed via WithStoreLimit and WithStoreTimeout. +type Feed interface { + // GetID provides the underlying ID of the feed. + GetID() string + + // GetRev provides the revision ID of the underlying document if available. + GetRev() string + + // Start makes a request to the changes feed to start receiving updates. + Start(ctx context.Context) + + // Seed overwrites the feed's persisted sequence position so a subsequent + // Start resumes from seq instead of any previously stored or initial + // position. It is intended for administrative recovery — for example + // seeding the source's current update_seq to skip a large backlog. It must + // be called before Start, with no other consumer writing the same feed + // document. + Seed(ctx context.Context, seq string) error + + // Next grabs the next ID from the change feed. If there is an error, it'll + // be provided. If the feed is stopped, the ID will be empty and there will + // not be an error. + Next(ctx context.Context) (string, error) + + // NextDoc behaves like Next but also returns the changed document's body + // when the feed was created with WithIncludeDocs. The body is nil when + // include_docs was not requested, or when it could not be read for this + // change (in which case the consumer should fetch the document itself). + // Acknowledgement semantics are identical to Next. + NextDoc(ctx context.Context) (id string, doc json.RawMessage, err error) + + // Stop requests that we stop listening for new changes. The current call to + // Next should then return an empty ID. + Stop() + + // Fatal returns a channel that fires when the feed has detected an + // unrecoverable condition such as a persistent conflict on the + // sequence document (only when WithFatalOnConflict is set). When + // fatal is signalled the feed self-stops, so a follow-up Next will + // return ("", nil) as for a normal close. The channel is closed when + // the feed is stopped. + Fatal() <-chan error +} + +// FeedCallback defines a callback to be used as an alternative to using +// the channel. This allows more messages to be processed at the same time +// and will wait for all currently executing callbacks to be processed before +// storing the current sequence. Any errors return from the callback will cause +// the Feed to be closed, so should only be used for major issues. +type FeedCallback func(id string) error + +type feedItem struct { + id string + seq string + doc json.RawMessage // populated only when include_docs is enabled +} + +type feed struct { + couch.Model + sync.Mutex + Seq string `json:"seq"` + + db *kivik.DB + changesDB *kivik.DB + source *kivik.Changes + outgoing chan feedItem + saveTimeout chan bool + started bool + lastSeq string + lastID string // doc ID paired with lastSeq (pending ack, Next mode only) + seqID string // doc ID paired with Seq, surfaced in the "persisted feed" log + + opts *options + + delay *time.Timer + count int // number of sequence updates since last save + + consecConflicts int // 409s in a row on the seq doc; reset on success + fatal chan error // signals unrecoverable conditions + fatalOnce sync.Once // guards send/close on fatal + stopOnce sync.Once // makes Stop idempotent + stopped bool // true once Stop has run; checked in Next + + log zerolog.Logger +} + +// New instantiates a new Feed object ready to be started. +func New(changesDB, srcDB *kivik.DB, opts ...Option) Feed { + f := &feed{ + Seq: "0", + db: srcDB, + changesDB: changesDB, + saveTimeout: make(chan bool), + count: 0, + started: false, + opts: newOptions(), + fatal: make(chan error, 1), + } + + for _, opt := range opts { + opt(f.opts) + } + if f.opts.initialSeq != "" { + f.Seq = f.opts.initialSeq + } + f.prepareID() + f.log = log.With().Str("id", f.GetID()).Logger() + return f +} + +func (f *feed) prepareID() { + id := f.db.Name() + for _, s := range f.opts.suffix { + if s != "" { + id = id + "_" + s + } + } + f.SetID(id) +} + +// Seed overwrites the feed's persisted sequence position with seq (see the +// Feed interface). It loads any existing document first so the write updates +// it in place, then stores the new sequence. +func (f *feed) Seed(ctx context.Context, seq string) error { + f.Lock() + defer f.Unlock() + if f.started { + return errors.New("cannot seed a started feed") + } + // Load the current document (if any) so we update in place and keep its + // revision; a missing document is fine and will be created by store. + if err := f.fetch(ctx); err != nil { + return err + } + f.Seq = seq + // store only writes when shouldSave is satisfied during normal operation; + // here we persist directly since this is an explicit, one-off seed. + return f.store(ctx) +} + +// Start establishes connection to fetch current feed state and +// establish change feed. If a callback is provided, this method +// will block until the connection is stopped, or an error occurs. +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") + }) +} + +// Fatal returns the channel that fires (and is then closed) when the +// feed encounters an unrecoverable condition. See the Feed interface +// for details. The channel is also closed on a normal Stop, so +// consumers can use a closed receive to detect shutdown. +func (f *feed) Fatal() <-chan error { + return f.fatal +} + +// signalFatal publishes err on the fatal channel and triggers a +// self-stop so consumers waiting on Next see end-of-feed. fatalOnce +// guards the send and the close together so we never send on a closed +// channel even if Stop and signalFatal race. +func (f *feed) signalFatal(err error) { + sent := false + f.fatalOnce.Do(func() { + // Buffered(1), this will not block. + f.fatal <- err + close(f.fatal) + sent = true + }) + if sent { + // Stop in a separate goroutine — we may be holding f.Mutex via + // Next, which Stop's path could compete with. + go f.Stop() + } +} + +// Next provides the next ID. This acts as an Ack as the current +// sequence state will not be saved until next is called again. +// Any errors that happen while trying to save the feed state +// or read from the source will be returned here. +func (f *feed) Next(ctx context.Context) (string, error) { + item, err := f.nextItem(ctx) + return item.id, err +} + +// NextDoc behaves like Next but also returns the changed document's body +// (see the Feed interface). The body is only populated when the feed was +// created with WithIncludeDocs. +func (f *feed) NextDoc(ctx context.Context) (string, json.RawMessage, error) { + item, err := f.nextItem(ctx) + return item.id, item.doc, err +} + +// nextItem is the shared implementation behind Next and NextDoc. It acts +// as an Ack: the current sequence state is not saved until the following +// call. Any errors saving the feed state or reading from the source are +// returned here. A zero feedItem with a nil error signals end-of-feed. +func (f *feed) nextItem(ctx context.Context) (feedItem, error) { + if f.opts.callback != nil { + return feedItem{}, errors.New("callback mode enabled, do not use Next") + } + f.Lock() // Only support requesting one next at a time + defer f.Unlock() + + // If Stop has already run (including via a fatal self-stop), report + // a clean end-of-feed so iterators don't spin in a retry loop. + if f.stopped { + return feedItem{}, nil + } + if !f.started { + return feedItem{}, errors.New("not started") + } + + if f.lastSeq != "" { + f.setSeq(f.lastID, f.lastSeq) + f.lastSeq = "" + f.lastID = "" + } + + for { + // Check if we need to save + if err := f.save(ctx); err != nil { + return feedItem{}, err + } + select { + case item, more := <-f.outgoing: + if !more { + return feedItem{}, nil // the end + } + f.lastSeq = item.seq + f.lastID = item.id + return item, nil + case <-f.saveTimeout: + // force a loop around so we try to save + continue + } + } +} + +// startWithCallbacks is the alternative approach to using the Next iterator. +func (f *feed) startWithCallbacks() { + closed := false + for !closed { + g := new(errgroup.Group) + maxInFlightLoop: + for i := 0; i < f.opts.maxInFlight; i++ { + select { + case item, more := <-f.outgoing: + if !more { + closed = true + break maxInFlightLoop + } + f.setSeq(item.id, item.seq) + i := item + g.Go(func() error { + return f.opts.callback(i.id) + }) + case <-f.saveTimeout: + break maxInFlightLoop + } + } + if err := g.Wait(); err != nil { + f.log.Error().Err(err).Msg("closing due to errors") + return + } + ctx := context.Background() // indpendent context + if err := f.save(ctx); err != nil { + f.log.Error().Err(err).Msg("failed to save sequence position, ignoring") + } + } +} + +// setSeq records that the provided seq (and the doc ID it belongs to) is +// the latest acknowledged position. It increments the internal counter +// and arms the save timer. The id is retained only for observability — +// it surfaces in the "persisted feed" log when the seq doc is written. +func (f *feed) setSeq(id, seq string) { + f.Seq = seq + if id != "" { + f.seqID = id + } + f.count++ + f.startSaveTimer() +} + +func (f *feed) startSaveTimer() { + if f.delay != nil { + return + } + f.delay = time.AfterFunc(f.opts.storeTimeout, func() { + if f.delay != nil { + f.delay.Stop() + f.delay = nil + } + f.saveTimeout <- true + }) +} + +func (f *feed) stopSaveTimer() { + if f.delay != nil { + f.delay.Stop() + f.delay = nil + } +} + +// shouldSave returns true if the delay timer is nil or +// if the count is over the configured store limit. +// When the count is zero, then obviously don't want to save. +func (f *feed) shouldSave() bool { + return f.count != 0 && (f.delay == nil || f.count > f.opts.storeLimit) +} + +func (f *feed) save(ctx context.Context) error { + if !f.shouldSave() { + return nil + } + f.stopSaveTimer() + + err := f.store(ctx) + if err == nil { + f.consecConflicts = 0 + ev := f.log.Info().Str("seq", f.Seq).Str("last_id", f.seqID) + if n, ok := f.pending(ctx); ok { + ev = ev.Int64("pending", n) + } + ev.Msg("persisted feed") + f.count = 0 + return nil + } + if kivik.HTTPStatus(err) != http.StatusConflict { + f.log.Error().Err(err).Msg("storing feed") + return err + } + + // 409: another writer beat us to it. Our local _rev is stale and + // every subsequent PUT will keep losing. Refresh _rev (and seq) so + // that the next save cycle has a chance, and surface the issue if + // it persists past conflictFatalThreshold consecutive conflicts. + f.consecConflicts++ + f.log.Warn(). + Err(err). + Int("consecutive", f.consecConflicts). + Msg("conflict on feed save, another instance is writing the same feed") + + if f.opts.fatalOnConflict && f.consecConflicts >= conflictFatalThreshold { + f.log.Error(). + Int("consecutive", f.consecConflicts). + Msg("persistent feed conflict — signalling fatal") + f.signalFatal(fmt.Errorf("persistent change-feed conflict on %s after %d attempts: %w", + f.GetID(), f.consecConflicts, err)) + return err + } + + if ferr := f.fetch(ctx); ferr != nil { + f.log.Error().Err(ferr).Msg("refetching feed doc after conflict") + return ferr + } + // Refetch synchronised us with whatever the conflicting writer + // committed. Reset the unsaved-count and re-arm the save timer so + // the next batch will save on its own cadence rather than spinning + // here. + f.count = 0 + f.startSaveTimer() + return nil +} + +// pending estimates how many source changes remain unprocessed by comparing +// the source database's current update_seq high-water mark against the feed's +// last stored position. It is called only when the sequence is persisted, so +// it adds at most one cheap stats request per save cycle. +// +// CouchDB sequence strings have the form "-", where the leading +// integer is a monotonic per-database change counter; their difference is the +// estimate. It is exact on a single node and approximate on a clustered +// database (the encoded tail packs per-shard sequences). ok is false when the +// stats request fails or either sequence has no parseable prefix (for example +// the initial "now"), in which case the caller should omit the estimate. +func (f *feed) pending(ctx context.Context) (n int64, ok bool) { + stats, err := f.db.Stats(ctx) + if err != nil { + f.log.Debug().Err(err).Msg("db stats unavailable for pending estimate") + return 0, false + } + head, ok1 := leadingSeq(stats.UpdateSeq) + cur, ok2 := leadingSeq(f.Seq) + if !ok1 || !ok2 { + return 0, false + } + if n = head - cur; n < 0 { + n = 0 + } + return n, true +} + +// leadingSeq parses the integer prefix of a CouchDB sequence string, i.e. the +// "" in "-" (or a bare ""). It reports ok=false when the prefix +// is not an integer. +func leadingSeq(seq string) (int64, bool) { + if i := strings.IndexByte(seq, '-'); i >= 0 { + seq = seq[:i] + } + n, err := strconv.ParseInt(seq, 10, 64) + if err != nil { + return 0, false + } + return n, true +} + +// connect to source. This will enter a loop that attempts +// to maintain the connection active. +func (f *feed) connect(ctx context.Context) { + bo := &backoff.Backoff{ + Min: 2 * time.Second, + Max: 2 * time.Minute, + } + + for { // connection loop + err := f.fetchAndConnect(ctx) + if err == nil { + err = f.processNext() + if err == nil { + return // implies channel closed + } + } + d := bo.Duration() + f.log.Error().Dur("retry_in", d).Err(err).Msg("reconnecting...") + time.Sleep(d) + } +} + +func (f *feed) processNext() error { + for { // result loop + if !f.source.Next() { + return f.source.Err() + } + + docID := f.source.ID() + if f.source.Deleted() || docID == "" || strings.HasPrefix(docID, "_") { + continue // ignore things we can't deal with + } + + item := feedItem{id: docID, seq: f.source.Seq()} + if f.opts.includeDocs { + var doc json.RawMessage + if err := f.source.ScanDoc(&doc); err != nil { + // Don't drop the change: deliver it without a body so the + // consumer falls back to fetching the document itself. + f.log.Warn().Err(err).Str("id", docID).Msg("scanning change doc, consumer will fetch") + } else { + item.doc = doc + } + } + f.outgoing <- item + } +} + +func (f *feed) fetchAndConnect(ctx context.Context) error { + // attempt to clean up + if f.source != nil { + _ = f.source.Close() + f.source = nil + } + + var err error + if err = f.fetch(ctx); err != nil { + return err + } + params := map[string]any{ + "include_docs": f.opts.includeDocs, + "heartbeat": f.opts.heartbeat, + "since": f.Seq, + "feed": "continuous", + } + if f.opts.seqInterval > 1 { + params["seq_interval"] = f.opts.seqInterval + } + if f.opts.filter != "" { + params["filter"] = f.opts.filter + for k, v := range f.opts.query { + params[k] = v + } + } + f.source = f.db.Changes(ctx, kivik.Params(params)) + f.log.Info().Str("seq", f.Seq).Msg("connected to changes feed") + return nil +} + +func (f *feed) fetch(ctx context.Context) error { + if err := couch.FetchModel(ctx, f.changesDB, f); err != nil { + if errors.Is(err, couch.ErrNotFound) { + return nil + } + return fmt.Errorf("failed to fetch change feed document: %w", err) + } + return nil +} + +func (f *feed) store(ctx context.Context) error { + if err := couch.StoreModel(ctx, f.changesDB, f); err != nil { + return err + } + return nil +} diff --git a/changes/feed_test.go b/changes/feed_test.go new file mode 100644 index 0000000..855245d --- /dev/null +++ b/changes/feed_test.go @@ -0,0 +1,240 @@ +package changes + +import ( + "errors" + "testing" + "time" +) + +// newBareFeed produces a feed instance for unit tests of in-memory state. +// It does NOT call New (which sets log fields off the kivik DB), so it is +// only safe for tests that don't touch the DB or run the connect loop. +func newBareFeed(opts ...Option) *feed { + f := &feed{ + Seq: "0", + saveTimeout: make(chan bool), + opts: newOptions(), + fatal: make(chan error, 1), + } + for _, opt := range opts { + opt(f.opts) + } + return f +} + +func TestWithFatalOnConflict_DefaultOff(t *testing.T) { + o := newOptions() + if o.fatalOnConflict { + t.Fatalf("fatalOnConflict should default to false") + } +} + +func TestWithFatalOnConflict_Sets(t *testing.T) { + o := newOptions() + WithFatalOnConflict()(o) + if !o.fatalOnConflict { + t.Fatalf("WithFatalOnConflict should enable fatalOnConflict") + } +} + +func TestWithIncludeDocs_DefaultOff(t *testing.T) { + o := newOptions() + if o.includeDocs { + t.Fatalf("includeDocs should default to false") + } +} + +func TestWithIncludeDocs_Sets(t *testing.T) { + o := newOptions() + WithIncludeDocs()(o) + if !o.includeDocs { + t.Fatalf("WithIncludeDocs should enable includeDocs") + } +} + +func TestWithStoreTimeout_Default(t *testing.T) { + o := newOptions() + if o.storeTimeout != DefaultStoreTimeout { + t.Fatalf("storeTimeout default = %v, want %v", o.storeTimeout, DefaultStoreTimeout) + } +} + +func TestWithStoreTimeout_Sets(t *testing.T) { + o := newOptions() + WithStoreTimeout(5 * time.Minute)(o) + if o.storeTimeout != 5*time.Minute { + t.Fatalf("WithStoreTimeout did not apply: got %v", o.storeTimeout) + } +} + +func TestWithStoreTimeout_IgnoresNonPositive(t *testing.T) { + o := newOptions() + WithStoreTimeout(0)(o) + WithStoreTimeout(-1 * time.Second)(o) + if o.storeTimeout != DefaultStoreTimeout { + t.Fatalf("WithStoreTimeout should ignore non-positive values; got %v", o.storeTimeout) + } +} + +func TestWithStoreLimit_Default(t *testing.T) { + o := newOptions() + if o.storeLimit != DefaultStoreLimit { + t.Fatalf("storeLimit default = %d, want %d", o.storeLimit, DefaultStoreLimit) + } +} + +func TestWithStoreLimit_Sets(t *testing.T) { + o := newOptions() + WithStoreLimit(500)(o) + if o.storeLimit != 500 { + t.Fatalf("WithStoreLimit did not apply: got %d", o.storeLimit) + } +} + +func TestWithStoreLimit_IgnoresNonPositive(t *testing.T) { + o := newOptions() + WithStoreLimit(0)(o) + WithStoreLimit(-7)(o) + if o.storeLimit != DefaultStoreLimit { + t.Fatalf("WithStoreLimit should ignore non-positive values; got %d", o.storeLimit) + } +} + +func TestSetSeq_RecordsPairedID(t *testing.T) { + f := newBareFeed() + f.setSeq("doc-A", "100") + if f.Seq != "100" || f.seqID != "doc-A" { + t.Fatalf("setSeq did not record paired id/seq: got seqID=%q Seq=%q", f.seqID, f.Seq) + } + + // A subsequent setSeq with an empty id (defensive, shouldn't happen in + // practice) must not clobber the last known id. + f.setSeq("", "101") + if f.Seq != "101" { + t.Fatalf("Seq should advance even with empty id: got %q", f.Seq) + } + if f.seqID != "doc-A" { + t.Fatalf("empty id must not clobber seqID: got %q", f.seqID) + } + + f.setSeq("doc-B", "102") + if f.seqID != "doc-B" { + t.Fatalf("seqID should track the latest non-empty id: got %q", f.seqID) + } +} + +func TestShouldSave_HonoursStoreLimit(t *testing.T) { + // The count > limit branch only matters while the delay timer holds + // (delay != nil). Install a dummy long-running timer so the timer + // branch doesn't short-circuit shouldSave. + armTimer := func(f *feed) { + f.delay = time.AfterFunc(time.Hour, func() {}) + } + + f := newBareFeed(WithStoreLimit(3)) + armTimer(f) + defer f.delay.Stop() + f.count = 4 + if !f.shouldSave() { + t.Fatalf("shouldSave should be true once count exceeds configured storeLimit") + } + + f = newBareFeed(WithStoreLimit(10)) + armTimer(f) + defer f.delay.Stop() + f.count = 4 + if f.shouldSave() { + t.Fatalf("shouldSave should be false while count is below storeLimit and delay timer holds") + } +} + +func TestSignalFatal_DeliversAndCloses(t *testing.T) { + f := newBareFeed() + want := errors.New("boom") + f.signalFatal(want) + + select { + case err, ok := <-f.Fatal(): + if !ok { + t.Fatalf("Fatal channel closed without delivering the error") + } + if !errors.Is(err, want) { + t.Fatalf("got err=%v, want=%v", err, want) + } + case <-time.After(time.Second): + t.Fatalf("timed out waiting for fatal signal") + } + + // Subsequent receive must observe channel closure. + select { + case _, ok := <-f.Fatal(): + if ok { + t.Fatalf("expected Fatal channel to be closed after delivery") + } + case <-time.After(time.Second): + t.Fatalf("timed out waiting for fatal channel close") + } +} + +func TestSignalFatal_OnlyFiresOnce(t *testing.T) { + f := newBareFeed() + f.signalFatal(errors.New("first")) + // Second call must not panic (it would if it tried to send/close again). + f.signalFatal(errors.New("second")) + // Drain the buffered first error. + if err := <-f.Fatal(); err == nil || err.Error() != "first" { + t.Fatalf("expected to receive the first error, got %v", err) + } +} + +func TestStop_ClosesFatalEvenWithoutFatal(t *testing.T) { + f := newBareFeed() + // Mark started so Stop runs the full path and exercises stopOnce; we + // don't care about the source/outgoing teardown since they're nil + // only when never started — set started=false to take the shortcut + // branch which still closes the fatal channel. + f.Stop() + + select { + case _, ok := <-f.Fatal(): + if ok { + t.Fatalf("Fatal should be closed (no signal), but got an error") + } + case <-time.After(time.Second): + t.Fatalf("timed out waiting for Fatal channel to close on Stop") + } + + // Stop is idempotent. + f.Stop() + + if !f.stopped { + t.Fatalf("expected feed.stopped to be true after Stop") + } +} + +func TestLeadingSeq(t *testing.T) { + cases := []struct { + name string + seq string + want int64 + ok bool + }{ + {"clustered", "12345-g1AAAABXeJ", 12345, true}, + {"bare integer", "678", 678, true}, + {"zero", "0", 0, true}, + {"initial now", "now", 0, false}, + {"empty", "", 0, false}, + {"non-numeric prefix", "abc-def", 0, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, ok := leadingSeq(tc.seq) + if ok != tc.ok { + t.Fatalf("leadingSeq(%q) ok = %v, want %v", tc.seq, ok, tc.ok) + } + if ok && got != tc.want { + t.Fatalf("leadingSeq(%q) = %d, want %d", tc.seq, got, tc.want) + } + }) + } +} diff --git a/changes/options.go b/changes/options.go new file mode 100644 index 0000000..18cc7c9 --- /dev/null +++ b/changes/options.go @@ -0,0 +1,171 @@ +package changes + +import ( + "fmt" + "time" +) + +// Now is the CouchDB-supported initial sequence value meaning "start from +// the current update_seq". Pass it to WithInitialSeq to skip historical +// changes and only receive updates from the moment the feed connects. +const Now = "now" + +const ( + // DefaultStoreTimeout is the default maximum delay between sequence + // document writes. Overridable per-feed via WithStoreTimeout. + DefaultStoreTimeout = 1 * time.Minute + + // DefaultStoreLimit is the default number of acked sequence updates + // the feed will buffer before forcing a save. Overridable per-feed + // via WithStoreLimit. + DefaultStoreLimit = 100 +) + +type options struct { + suffix []string + query map[string]string + filter string + heartbeat int // default 15000 + seqInterval int + callback FeedCallback + maxInFlight int + fatalOnConflict bool + initialSeq string + storeTimeout time.Duration + storeLimit int + includeDocs bool +} + +// newOptions instantiates a new options object with defaults. +func newOptions() *options { + return &options{ + suffix: []string{}, + filter: "", + heartbeat: 15000, + seqInterval: 1, + maxInFlight: 10, + storeTimeout: DefaultStoreTimeout, + storeLimit: DefaultStoreLimit, + } +} + +// Option defines a callback to be issued for each configuration option. +type Option func(opts *options) + +// WithCallback defines a method to use for callbacks as an alternative to the +// next iterator. +func WithCallback(cb FeedCallback) Option { + return func(opts *options) { + opts.callback = cb + } +} + +// WithMaxInflight defines how many callbacks should be allowed to be processing +// at the same time. This is only relevant when using callbacks. Must be bigger +// than zero. +func WithMaxInFlight(size int) Option { + return func(opts *options) { + if size > 0 { + opts.maxInFlight = size + } + } +} + +// WithSeqInterval utilizes the CouchDB `seq_interval` option to define how many changes +// should be loaded inside a batch. By default, this is 1, which is suitable for low-volume +// processes. This should be higher for greater volumes. +func WithSeqInterval(val int) Option { + return func(opts *options) { + opts.seqInterval = val + } +} + +// WithSuffix appends the provided strings to the change feed name +func WithSuffix(s ...string) Option { + return func(opts *options) { + opts.suffix = s + } +} + +// WithFilter defines a CouchDB filter to use before processing the incoming +// changes from the feed. +func WithFilter(design, name string, query map[string]string) Option { + return func(opts *options) { + opts.filter = fmt.Sprintf("%s/%s", design, name) + opts.query = query + } +} + +// WithHeartbeat sets the feed heartbeat interval. If none set, the default +// is 15000 (15s). +func WithHeartbeat(v int) Option { + return func(opts *options) { + opts.heartbeat = v + } +} + +// WithInitialSeq overrides the initial sequence used the very first time +// a feed is started (when no persisted change_feeds document exists yet). +// Defaults to "0", meaning "replay from the start". Pass "now" to start +// from the current point in the changes feed. +// +// Once the feed has persisted a sequence document, that stored value +// takes precedence on subsequent restarts and this option is a no-op. +func WithInitialSeq(seq string) Option { + return func(opts *options) { + opts.initialSeq = seq + } +} + +// WithFatalOnConflict enables fatal-error signalling via the Feed's Fatal +// channel when the per-feed sequence document keeps conflicting after a +// few save attempts (a strong signal that another process is consuming +// the same feed). The feed already attempts to recover from a single +// conflict transparently by refetching the document's revision; this +// option only controls whether persistent conflicts surface as fatal. +func WithFatalOnConflict() Option { + return func(opts *options) { + opts.fatalOnConflict = true + } +} + +// WithStoreTimeout overrides the maximum delay between sequence document +// writes. After this duration has elapsed since the last save, the feed +// forces a save on the next Next/callback cycle. A non-positive value is +// ignored and the default (DefaultStoreTimeout) is kept. Raising this +// reduces write load on the change_feeds database at the cost of more +// replay work on crash recovery. +func WithStoreTimeout(d time.Duration) Option { + return func(opts *options) { + if d > 0 { + opts.storeTimeout = d + } + } +} + +// WithStoreLimit overrides the number of acked sequence updates the feed +// will buffer before forcing a save, irrespective of the timeout. A +// non-positive value is ignored and the default (DefaultStoreLimit) is +// kept. Raising this batches more updates per save at the cost of more +// replay work on crash recovery. +func WithStoreLimit(n int) Option { + return func(opts *options) { + if n > 0 { + opts.storeLimit = n + } + } +} + +// WithIncludeDocs requests that the changes feed include each changed +// document's body (CouchDB's include_docs=true). The body is delivered +// via NextDoc as a json.RawMessage, letting consumers unmarshal it +// directly instead of issuing a separate fetch per change. Bodies are +// not delivered through Next or the callback runner. As with any +// include_docs feed, the body reflects the document's current winning +// revision, which may be newer than the revision that produced the +// change. +func WithIncludeDocs() Option { + return func(opts *options) { + opts.includeDocs = true + } +} diff --git a/changes/workerpool.go b/changes/workerpool.go new file mode 100644 index 0000000..ba52f65 --- /dev/null +++ b/changes/workerpool.go @@ -0,0 +1,223 @@ +package changes + +import ( + "context" + "hash/maphash" + "sync" + "time" + + "github.com/jpillora/backoff" + "github.com/rs/zerolog/log" +) + +// Fetcher materialises a typed document from a doc ID emitted by a +// change feed. WorkerPool retries the call with backoff on error. +type Fetcher[T any] func(ctx context.Context, id string) (T, error) + +// KeyFunc extracts a partition key from a fetched item. Used by +// hashed-mode pools to keep same-key items on the same worker. +type KeyFunc[T any] func(T) string + +// WorkerPool dispatches change-feed items across a pool of workers. It +// pulls IDs from one or more Feeds, calls Fetcher to load the typed +// document, and routes the result to a worker channel. Workers consume +// via Next. +// +// In flat mode every worker reads from the same shared channel — any +// worker can pull any item. In hashed mode each worker has its own +// channel and items are routed by hash(KeyFunc(item)) % workers; items +// sharing a key always land on the same worker, preserving arrival +// order within the partition. +// +// WorkerPool owns the feed lifecycle: do not call Feed.Start before +// passing the feed in, and do not call Feed.Stop independently. Use +// WorkerPool.Start and WorkerPool.Stop instead. +type WorkerPool[T any] struct { + feeds map[string]Feed + fetcher Fetcher[T] + workers int + channels []chan T + hashed bool + keyFn KeyFunc[T] + seed maphash.Seed + fatal chan error + fatalOnce sync.Once + procWG sync.WaitGroup + stopOnce sync.Once + closedOnce sync.Once +} + +// NewFlatWorkerPool constructs a pool that distributes items across a +// single shared channel. Suitable for non-partitioned models (and for +// partitioned ones where downstream processing is order-tolerant). +// +// `workers` controls the pool size; the shared channel is buffered to +// `workers` slots so a single slow worker doesn't immediately stall +// the fetcher. +func NewFlatWorkerPool[T any](feeds map[string]Feed, fetcher Fetcher[T], workers int) *WorkerPool[T] { + if workers < 1 { + workers = 1 + } + return &WorkerPool[T]{ + feeds: feeds, + fetcher: fetcher, + workers: workers, + channels: []chan T{make(chan T, workers)}, + fatal: make(chan error, 1), + } +} + +// NewHashedWorkerPool constructs a pool with per-worker channels. +// Items are routed by hash(keyFn(item)) % workers so all items sharing +// a key always land on the same worker. Use for partitioned models +// when you need to preserve arrival order within a partition. +// +// Per-worker channels are buffered to `workers` slots each. +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) + for i := range chs { + chs[i] = make(chan T, workers) + } + return &WorkerPool[T]{ + feeds: feeds, + fetcher: fetcher, + workers: workers, + channels: chs, + hashed: true, + keyFn: keyFn, + seed: maphash.MakeSeed(), + fatal: make(chan error, 1), + } +} + +// Start opens every feed and spawns a fetcher goroutine per shard. +// Each fetcher pulls IDs, calls Fetcher, and dispatches the resulting +// item to a worker channel. The call returns immediately; workers +// should be started by the caller and pull via Next. +func (p *WorkerPool[T]) Start(ctx context.Context) { + for shard, f := range p.feeds { + f.Start(ctx) + p.procWG.Add(1) + go p.runFetcher(ctx, shard, f) + go p.watchFatal(shard, f) + } +} + +// Next returns the next item routed to the given worker, blocking +// until one is available. Returns the zero value and false once the +// pool has been stopped and the worker's channel is fully drained. +func (p *WorkerPool[T]) Next(workerID int) (T, bool) { + v, ok := <-p.recv(workerID) + return v, ok +} + +// Workers reports the configured pool size. +func (p *WorkerPool[T]) Workers() int { + return p.workers +} + +// Fatal returns a channel that fires when one of the underlying feeds +// signals an unrecoverable condition (most notably a persistent +// conflict on the sequence document, indicating another process is +// consuming the same feed). The channel is closed when the pool stops +// without a fatal condition, so a closed-channel receive doubles as a +// shutdown signal. +func (p *WorkerPool[T]) Fatal() <-chan error { + return p.fatal +} + +// Stop halts every feed, waits for the fetcher goroutines to exit, +// and closes the worker channels. Safe to call multiple times. +func (p *WorkerPool[T]) Stop() { + p.stopOnce.Do(func() { + for _, f := range p.feeds { + f.Stop() + } + p.procWG.Wait() + p.closedOnce.Do(func() { + for _, ch := range p.channels { + close(ch) + } + }) + // Fatal is closed by Stop iff no fatal signal fired first. + p.fatalOnce.Do(func() { close(p.fatal) }) + }) +} + +// dispatch routes a fetched item to the appropriate worker channel. +func (p *WorkerPool[T]) dispatch(item T) { + if p.hashed { + sum := maphash.String(p.seed, p.keyFn(item)) + idx := int(sum % uint64(p.workers)) //nolint:gosec // bounded by modulo workers + p.channels[idx] <- item + return + } + p.channels[0] <- item +} + +// recv returns the channel a given worker should consume from. In flat +// mode every workerID maps to the same shared channel. +func (p *WorkerPool[T]) recv(workerID int) <-chan T { + if p.hashed { + return p.channels[workerID] + } + return p.channels[0] +} + +// runFetcher pulls IDs from one shard's feed, fetches the typed doc +// for each, and dispatches it. Both feed reads and fetch errors are +// retried with backoff. +func (p *WorkerPool[T]) runFetcher(ctx context.Context, shard string, f Feed) { + defer p.procWG.Done() + bo := &backoff.Backoff{ + Min: 2 * time.Second, + Max: 5 * time.Minute, + Factor: 2, + } + for { + id, err := f.Next(ctx) + 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 id == "" { + log.Info().Str("shard", shard).Msg("change feed closed") + return + } + + 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) + } + bo.Reset() + p.dispatch(item) + } +} + +// watchFatal forwards a per-shard feed's first fatal error onto the +// pool's combined fatal channel. +func (p *WorkerPool[T]) watchFatal(shard string, f Feed) { + err, ok := <-f.Fatal() + if !ok { + return + } + log.Error().Err(err).Str("shard", shard).Msg("change feed fatal") + p.fatalOnce.Do(func() { + select { + case p.fatal <- err: + default: + } + close(p.fatal) + }) +} diff --git a/changes/workerpool_test.go b/changes/workerpool_test.go new file mode 100644 index 0000000..aa9636c --- /dev/null +++ b/changes/workerpool_test.go @@ -0,0 +1,347 @@ +package changes + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" +) + +// fakeFeed is a Feed stub for testing WorkerPool in isolation. It +// returns IDs from a fixed slice, then blocks until Stop is called. +type fakeFeed struct { + id string + ids []string + cur int + stopped atomic.Bool + stopCh chan struct{} + fatal chan error + fatalOnce sync.Once + startCnt atomic.Int32 +} + +func newFakeFeed(id string, ids ...string) *fakeFeed { + return &fakeFeed{ + id: id, + ids: ids, + stopCh: make(chan struct{}), + fatal: make(chan error, 1), + } +} + +func (f *fakeFeed) GetID() string { return f.id } +func (f *fakeFeed) GetRev() string { return "" } +func (f *fakeFeed) Seed(_ context.Context, _ string) error { return nil } +func (f *fakeFeed) Start(_ context.Context) { + f.startCnt.Add(1) +} +func (f *fakeFeed) Next(ctx context.Context) (string, error) { + if f.stopped.Load() { + return "", nil + } + if f.cur < len(f.ids) { + id := f.ids[f.cur] + f.cur++ + return id, nil + } + select { + case <-f.stopCh: + return "", nil + case <-ctx.Done(): + return "", ctx.Err() + } +} +func (f *fakeFeed) NextDoc(ctx context.Context) (string, json.RawMessage, error) { + id, err := f.Next(ctx) + return id, nil, err +} +func (f *fakeFeed) Stop() { + if f.stopped.CompareAndSwap(false, true) { + close(f.stopCh) + f.fatalOnce.Do(func() { close(f.fatal) }) + } +} +func (f *fakeFeed) Fatal() <-chan error { return f.fatal } + +// signalFatal injects a fatal condition. Mirrors the real feed's +// fatal/close semantics: send then close, idempotent. +func (f *fakeFeed) signalFatal(err error) { + f.fatalOnce.Do(func() { + f.fatal <- err + close(f.fatal) + }) +} + +func waitUntil(t *testing.T, deadline time.Duration, cond func() bool, msg string) { + t.Helper() + end := time.Now().Add(deadline) + for time.Now().Before(end) { + if cond() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("timeout waiting for: %s", msg) +} + +func TestFlatWorkerPool_AllItemsDelivered(t *testing.T) { + feed := newFakeFeed("test", "a", "b", "c", "d") + feeds := map[string]Feed{"test": feed} + pool := NewFlatWorkerPool(feeds, func(_ context.Context, id string) (string, error) { + return id, nil + }, 2) + pool.Start(context.Background()) + defer pool.Stop() + + var got []string + var mu sync.Mutex + var wg sync.WaitGroup + for i := range pool.Workers() { + wg.Add(1) + go func(id int) { + defer wg.Done() + for { + v, ok := pool.Next(id) + if !ok { + return + } + mu.Lock() + got = append(got, v) + mu.Unlock() + } + }(i) + } + + waitUntil(t, 2*time.Second, func() bool { + mu.Lock() + defer mu.Unlock() + return len(got) == 4 + }, "all 4 items delivered") + + pool.Stop() + wg.Wait() + + mu.Lock() + defer mu.Unlock() + if len(got) != 4 { + t.Fatalf("expected 4 items, got %d: %v", len(got), got) + } +} + +func TestHashedWorkerPool_SameKeyAlwaysSameWorker(t *testing.T) { + // Many copies of the same key — all must arrive on a single worker. + ids := []string{} + for i := range 50 { + ids = append(ids, fmt.Sprintf("entry-a-rev-%d", i)) + } + feed := newFakeFeed("test", ids...) + feeds := map[string]Feed{"test": feed} + pool := NewHashedWorkerPool(feeds, func(_ context.Context, id string) (string, error) { + return id, nil + }, 4, func(s string) string { return "entry-a" }) + pool.Start(context.Background()) + defer pool.Stop() + + hits := make([]int, pool.Workers()) + var mu sync.Mutex + var wg sync.WaitGroup + for i := range pool.Workers() { + wg.Add(1) + go func(id int) { + defer wg.Done() + for { + _, ok := pool.Next(id) + if !ok { + return + } + mu.Lock() + hits[id]++ + mu.Unlock() + } + }(i) + } + + waitUntil(t, 2*time.Second, func() bool { + mu.Lock() + defer mu.Unlock() + sum := 0 + for _, h := range hits { + sum += h + } + return sum == 50 + }, "all 50 items processed") + + pool.Stop() + wg.Wait() + + mu.Lock() + defer mu.Unlock() + used := 0 + for _, h := range hits { + if h > 0 { + used++ + } + } + if used != 1 { + t.Fatalf("expected all items on one worker; hits=%v", hits) + } +} + +func TestWorkerPool_FetcherErrorIsRetried(t *testing.T) { + feed := newFakeFeed("test", "doc-1") + feeds := map[string]Feed{"test": feed} + var attempts atomic.Int32 + pool := NewFlatWorkerPool(feeds, func(_ context.Context, id string) (string, error) { + if attempts.Add(1) < 3 { + return "", errors.New("transient") + } + return id, nil + }, 1) + // Tighten the backoff for the test so the retry happens quickly. + pool.Start(context.Background()) + defer pool.Stop() + + v, ok := pool.Next(0) + if !ok { + t.Fatal("expected item, got closed channel") + } + if v != "doc-1" { + t.Fatalf("got %q, want doc-1", v) + } + if got := attempts.Load(); got < 3 { + t.Fatalf("expected at least 3 fetch attempts, got %d", got) + } +} + +func TestWorkerPool_FatalFromFeedSurfaces(t *testing.T) { + feed := newFakeFeed("test") + feeds := map[string]Feed{"test": feed} + pool := NewFlatWorkerPool(feeds, func(_ context.Context, id string) (string, error) { + return id, nil + }, 1) + pool.Start(context.Background()) + defer pool.Stop() + + want := errors.New("boom") + feed.signalFatal(want) + + select { + case err, ok := <-pool.Fatal(): + if !ok { + t.Fatal("Fatal channel closed without delivering the error") + } + if !errors.Is(err, want) { + t.Fatalf("got err=%v, want=%v", err, want) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for fatal signal") + } +} + +func TestWorkerPool_StopClosesWorkerChannelsCleanly(t *testing.T) { + feed := newFakeFeed("test", "a") + feeds := map[string]Feed{"test": feed} + pool := NewFlatWorkerPool(feeds, func(_ context.Context, id string) (string, error) { + return id, nil + }, 2) + pool.Start(context.Background()) + + // Drain the first item from any worker. + _, ok := pool.Next(0) + if !ok { + t.Fatal("expected the buffered item") + } + + pool.Stop() + + // After Stop, Next on any worker must observe a closed channel. + for i := range pool.Workers() { + if _, ok := pool.Next(i); ok { + t.Fatalf("worker %d: expected closed channel after Stop", i) + } + } + + // Stop is idempotent. + pool.Stop() +} + +func TestWorkerPool_StopWithoutFatalClosesFatalChannel(t *testing.T) { + feed := newFakeFeed("test") + feeds := map[string]Feed{"test": feed} + pool := NewFlatWorkerPool(feeds, func(_ context.Context, id string) (string, error) { + return id, nil + }, 1) + pool.Start(context.Background()) + pool.Stop() + + select { + case _, ok := <-pool.Fatal(): + if ok { + t.Fatal("expected Fatal to close on Stop, but got an error") + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for Fatal close") + } +} + +func TestWorkerPool_MultipleFeedsAllRun(t *testing.T) { + feedA := newFakeFeed("a", "doc-a-1", "doc-a-2") + feedB := newFakeFeed("b", "doc-b-1", "doc-b-2", "doc-b-3") + feeds := map[string]Feed{"a": feedA, "b": feedB} + pool := NewFlatWorkerPool(feeds, func(_ context.Context, id string) (string, error) { + return id, nil + }, 2) + pool.Start(context.Background()) + defer pool.Stop() + + got := map[string]bool{} + var mu sync.Mutex + var wg sync.WaitGroup + for i := range pool.Workers() { + wg.Add(1) + go func(id int) { + defer wg.Done() + for { + v, ok := pool.Next(id) + if !ok { + return + } + mu.Lock() + got[v] = true + mu.Unlock() + } + }(i) + } + + waitUntil(t, 2*time.Second, func() bool { + mu.Lock() + defer mu.Unlock() + return len(got) == 5 + }, "all 5 items across both feeds") + + if feedA.startCnt.Load() != 1 || feedB.startCnt.Load() != 1 { + t.Fatalf("expected each feed Started once; a=%d b=%d", feedA.startCnt.Load(), feedB.startCnt.Load()) + } + + pool.Stop() + wg.Wait() +} + +func TestWorkerPool_NonPositiveWorkersNormalised(t *testing.T) { + feeds := map[string]Feed{"test": newFakeFeed("test")} + fetcher := func(_ context.Context, id string) (string, error) { return id, nil } + + if p := NewFlatWorkerPool(feeds, fetcher, 0); p.Workers() != 1 { + t.Fatalf("flat: workers normalised to 1, got %d", p.Workers()) + } + if p := NewFlatWorkerPool(feeds, fetcher, -3); p.Workers() != 1 { + t.Fatalf("flat: negative workers normalised to 1, got %d", p.Workers()) + } + if p := NewHashedWorkerPool(feeds, fetcher, 0, func(s string) string { return s }); p.Workers() != 1 { + t.Fatalf("hashed: workers normalised to 1, got %d", p.Workers()) + } +} diff --git a/config.go b/config.go new file mode 100644 index 0000000..f3f017e --- /dev/null +++ b/config.go @@ -0,0 +1,61 @@ +package couch + +import ( + "net/url" +) + +// DefaultSeparator determines the character(s) to use to separate +// a prefix from the database name. Underscore is the default +// to be consistent with SQL table naming and JSON attributes. +const DefaultSeparator = "_" + +// Config is used to define the connection details to a database. +type Config struct { + Scheme string `json:"scheme"` + Host string `json:"host"` + Port string `json:"port"` + Username string `json:"username"` + Password string `json:"password"` + Prefix string `json:"prefix"` + Separator string `json:"separator"` +} + +// NewConfig generates a new configuration instance and requires a prefix +// so that we have a nice namespace before all database names. +func NewConfig(prefix string) *Config { + c := &Config{ + Scheme: "http", + Host: "couchdb", // assume we're in Docker + Port: "5984", + Prefix: prefix, + } + return c +} + +func (c *Config) baseURL() string { + u := &url.URL{ + Scheme: c.Scheme, + Host: c.Host + ":" + c.Port, + } + if c.Username != "" { + u.User = url.UserPassword(c.Username, c.Password) + } + return u.String() +} + +// db provides a DB name including the configured prefix. +func (c *Config) db(name string) string { + if name == "" { + name = c.Prefix + } else { + name = c.Prefix + c.separator() + name + } + return name +} + +func (c *Config) separator() string { + if c.Separator != "" { + return c.Separator + } + return DefaultSeparator +} diff --git a/config_test.go b/config_test.go new file mode 100644 index 0000000..b71328d --- /dev/null +++ b/config_test.go @@ -0,0 +1,30 @@ +package couch + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestConfigBaseURL(t *testing.T) { + c := NewConfig("app") + assert.Equal(t, "http://couchdb:5984", c.baseURL(), "defaults") + + c.Scheme = "https" + c.Host = "db.example" + c.Port = "6984" + assert.Equal(t, "https://db.example:6984", c.baseURL()) + + c.Username = "admin" + c.Password = "s3cr3t" + assert.Equal(t, "https://admin:s3cr3t@db.example:6984", c.baseURL(), "with credentials") +} + +func TestConfigDBNaming(t *testing.T) { + c := NewConfig("app") + assert.Equal(t, "app_widgets", c.db("widgets"), "prefix + separator + name") + assert.Equal(t, "app", c.db(""), "empty name resolves to the prefix alone") + + c.Separator = "-" + assert.Equal(t, "app-widgets", c.db("widgets"), "custom separator") +} diff --git a/couch.go b/couch.go new file mode 100644 index 0000000..dc10119 --- /dev/null +++ b/couch.go @@ -0,0 +1,93 @@ +// Package couch provides a small wrapper around the kivik CouchDB driver to +// make it easier to configure connections and persist models. +package couch + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/go-kivik/kivik/v4" + _ "github.com/go-kivik/kivik/v4/couchdb" // CouchDB driver +) + +const driverName = "couch" + +// Client wraps around a kivik package Client and helps make it +// easier to configure the connection and prepare the database. +type Client struct { + conf *Config + client *kivik.Client +} + +// New provides a new instance of the default CouchDB client. This +// call will block until the server responds or the context causes +// a timeout. +func New(conf *Config, opts ...kivik.Option) (*Client, error) { + c := new(Client) + c.conf = conf + var err error + c.client, err = kivik.New(driverName, conf.baseURL(), opts...) + if err != nil { + return nil, fmt.Errorf("couch: %w", err) + } + return c, nil +} + +// Ping attempts to establish a connection and will block and retry +// for any timeouts or network errors. This is recommended to be used +// after the client has been initialized to ensure the connection +// is ready to use. +func (c *Client) Ping(ctx context.Context) error { + limit := 10 + for i := 0; i <= limit; i++ { + _, err := c.client.Ping(ctx) + switch kivik.HTTPStatus(err) { + case 0: + return nil + case 408, 504: + select { + case <-time.After(1 * time.Second): + continue // try again + case <-ctx.Done(): + return errors.New("request canceled") + } + default: + return err + } + } + return fmt.Errorf("failed after %d retries", limit) +} + +// DB is used to provide a database instance at the provided name. +func (c *Client) DB(name string) *kivik.DB { + name = c.conf.db(name) + return c.client.DB(name) +} + +// SyncDesigns ensures the database is up to date with the latest design documents. +func (c *Client) SyncDesigns(ctx context.Context, db *kivik.DB, designs []*Design) error { + for _, design := range designs { + if err := design.Sync(ctx, db); err != nil { + return err + } + } + return nil +} + +// Create checks that the database already exists, or creates it +// if required. +func (c *Client) Create(ctx context.Context, db *kivik.DB, opts ...kivik.Option) error { + ok, err := c.client.DBExists(ctx, db.Name()) + if err != nil { + return err + } + if !ok { + // db doesn't exist, create it + if err = c.client.CreateDB(ctx, db.Name(), opts...); err != nil { + return err + } + } + return nil +} diff --git a/design.go b/design.go new file mode 100644 index 0000000..7c7b6c5 --- /dev/null +++ b/design.go @@ -0,0 +1,116 @@ +package couch + +import ( + "context" + "crypto/sha256" + "fmt" + "net/http" + "sort" + "strings" + + "github.com/go-kivik/kivik/v4" +) + +const ( + designDocPrefix = "_design/" + designDocDefaultLanguage = "javascript" +) + +// Design represents the special design documents used to query documents using +// pre-defined indexes. +// Only designs that have changed will be synchronised with the database using a simple +// SHA256 comparison algorithm that checks for changes in the views. +type Design struct { + Model + Language string `json:"language"` + + // Options stor additional options for the design document. + Options map[string]any `json:"options,omitempty"` + + Filters map[string]string `json:"filters,omitempty"` + Views map[string]*View `json:"views,omitempty"` +} + +// View is a basic definition of a CouchDB view. +type View struct { + Map string `json:"map"` + Reduce string `json:"reduce,omitempty"` +} + +// NewDesign builds a new design document instance using the provided name. +func NewDesign(name string) *Design { + d := new(Design) + d.SetID(designDocPrefix + name) + d.Language = designDocDefaultLanguage + d.Options = make(map[string]any) + d.Filters = make(map[string]string) + d.Views = make(map[string]*View) + return d +} + +// SetView adds the provided view with the given name. +func (d *Design) SetView(name string, view *View) { + d.Views[name] = view +} + +// SetFilter adds the provided filter to the design document. +// Existing filters with the same name will be replaced. +func (d *Design) SetFilter(name string, filter string) { + d.Filters[name] = filter +} + +// Checksum generates a SHA256 sum by joining all the filters and views +// together to form a single string and running the result through +// the digest algorithm. The result is a Hexadecimal string. +func (d *Design) Checksum() string { + text := d.filterConcat() + d.viewConcat() + sum := sha256.Sum256([]byte(text)) + return fmt.Sprintf("%x", sum) +} + +func (d *Design) filterConcat() string { + var keys, items []string + for k := range d.Filters { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + items = append(items, k, d.Filters[k]) + } + return strings.Join(items, ":") +} + +func (d *Design) viewConcat() string { + var keys, items []string + for k := range d.Views { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + view := d.Views[k] + items = append(items, k, "map", view.Map, "reduce", view.Reduce) + } + return strings.Join(items, ":") +} + +// Sync compares the checksums of the current design and the previous +func (d *Design) Sync(ctx context.Context, db *kivik.DB) error { + prev := new(Design) + if err := db.Get(ctx, d.ID).ScanDoc(prev); err != nil { + if kivik.HTTPStatus(err) != http.StatusNotFound { + return fmt.Errorf("couch: %w", err) + } + } + if prev.Persisted() { + // Copy a few key details from existing design + d.SetRev(prev.Rev) + d.CreatedAt = prev.CreatedAt + d.UpdatedAt = prev.UpdatedAt + if d.Checksum() == prev.Checksum() { + return nil // no changes + } + } else { + d.Reset() // ensures revision and timestamp data is not copied + } + return StoreModel(ctx, db, d) +} diff --git a/design_test.go b/design_test.go new file mode 100644 index 0000000..651ebfc --- /dev/null +++ b/design_test.go @@ -0,0 +1,36 @@ +package couch_test + +import ( + "testing" + + "github.com/invopop/couch" +) + +func TestDesignInstantiation(t *testing.T) { + design := couch.NewDesign("test") + if design.ID != "_design/test" { + t.Errorf("Unexpected doc ID: %s", design.ID) + } + if design.Language != "javascript" { + t.Errorf("Unexpected language: %s", design.Language) + } + if design.Views == nil { + t.Error("View map is not initialized!") + } +} + +func TestChecksum(t *testing.T) { + design := couch.NewDesign("test") + design.SetView("by_created_at", &couch.View{ + Map: "function(d) { if (d['created_at']) { emit(d['created_at'], 1); } }", + Reduce: "_sum", + }) + cs := design.Checksum() + if want := "2e1c80b5f2eb78fec2396a11dce712648710d300d500c9a392034a21d90bbbcd"; want != cs { + t.Errorf("unexpected checksum: %s", cs) + } + design.Views["by_created_at"].Reduce = "_stats" + if design.Checksum() == cs { + t.Error("Checksums match when they should differ!") + } +} diff --git a/document.go b/document.go new file mode 100644 index 0000000..26afb97 --- /dev/null +++ b/document.go @@ -0,0 +1,42 @@ +package couch + +// Document is a simplified object that conforms to the Persistable +// implementation. Unlike the Model implementation, it doesn't include +// any timestamping. +type Document struct { + ID string `json:"_id,omitempty"` + Rev string `json:"_rev,omitempty"` +} + +// UpdateTimestamps in the context of a simple CouchDB document does +// nothing. +func (d *Document) UpdateTimestamps() { + // nothing to do +} + +// GetID provides the current document ID. +func (d *Document) GetID() string { + return d.ID +} + +// GetRev provides the document's revision. +func (d *Document) GetRev() string { + return d.Rev +} + +// SetID sets the model's ID +func (d *Document) SetID(id string) { + d.ID = id +} + +// SetRev update's the documents's revision ID. The should mainly be +// used by persistence layers. +func (d *Document) SetRev(rev string) { + d.Rev = rev +} + +// Persisted returns true if the revision has been set, a value that should +// always be provided by the database server. +func (d *Document) Persisted() bool { + return d.Rev != "" +} diff --git a/document_test.go b/document_test.go new file mode 100644 index 0000000..1c878d1 --- /dev/null +++ b/document_test.go @@ -0,0 +1,22 @@ +package couch_test + +import ( + "testing" + + "github.com/invopop/couch" + "github.com/stretchr/testify/assert" +) + +func TestDocument(t *testing.T) { + d := &couch.Document{} + assert.Empty(t, d.GetID()) + assert.False(t, d.Persisted()) + + d.SetID("doc-1") + d.SetRev("1-abc") + assert.Equal(t, "doc-1", d.GetID()) + assert.Equal(t, "1-abc", d.GetRev()) + assert.True(t, d.Persisted()) + + d.UpdateTimestamps() // no-op for a plain Document; must not panic +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..a987b79 --- /dev/null +++ b/go.mod @@ -0,0 +1,26 @@ +module github.com/invopop/couch + +go 1.25.0 + +require ( + github.com/go-kivik/kivik/v4 v4.5.2 + github.com/jpillora/backoff v1.0.0 + github.com/rs/zerolog v1.35.1 + github.com/stretchr/testify v1.11.1 + golang.org/x/sync v0.22.0 +) + +require ( + github.com/Masterminds/semver/v3 v3.3.0 // indirect + github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect + github.com/google/go-cmp v0.7.0 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/kr/pretty v0.3.1 // indirect + github.com/mattn/go-colorable v0.1.14 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect + github.com/rogpeppe/go-internal v1.14.1 // indirect + golang.org/x/net v0.56.0 // indirect + golang.org/x/sys v0.46.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..9fd8026 --- /dev/null +++ b/go.sum @@ -0,0 +1,53 @@ +github.com/Masterminds/semver/v3 v3.3.0 h1:B8LGeaivUe71a5qox1ICM/JLl0NqZSW5CHyL+hmvYS0= +github.com/Masterminds/semver/v3 v3.3.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM= +github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= +github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/go-kivik/kivik/v4 v4.5.2 h1:Xi6QyjscrWrSRQEEW/25vaWyh20ZCg3LhTm6AqFRZCc= +github.com/go-kivik/kivik/v4 v4.5.2/go.mod h1:5YlQJZim4qvaJ3T0fCAS6U4oaN4hzXK6CVY9nvN4Phg= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gopherjs/gopherjs v1.20.1 h1:22uLWFvVcxhJ+j3dJ99NNfwGyHynxCmjhYsrcwqbY60= +github.com/gopherjs/gopherjs v1.20.1/go.mod h1:h+FTmmLgbXMmmtuZFp9bUqXciN429Wx0sJEJuMnpyfM= +github.com/icza/dyno v0.0.0-20230330125955-09f820a8d9c0 h1:nHoRIX8iXob3Y2kdt9KsjyIb7iApSvb3vgsd93xb5Ow= +github.com/icza/dyno v0.0.0-20230330125955-09f820a8d9c0/go.mod h1:c1tRKs5Tx7E2+uHGSyyncziFjvGpgv4H2HrqXeUQ/Uk= +github.com/jpillora/backoff v1.0.0 h1:uvFg412JmmHBHw7iwprIxkPMI+sGQ4kzOWsMeHnm2EA= +github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE= +github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= +github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= +github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/rs/zerolog v1.35.1 h1:m7xQeoiLIiV0BCEY4Hs+j2NG4Gp2o2KPKmhnnLiazKI= +github.com/rs/zerolog v1.35.1/go.mod h1:EjML9kdfa/RMA7h/6z6pYmq1ykOuA8/mjWaEvGI+jcw= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +gitlab.com/flimzy/testy v0.15.0 h1:69TL12IpxqGUyL8NuRV3Z5OhIDszXLNqLtfBDhOV3ys= +gitlab.com/flimzy/testy v0.15.0/go.mod h1:KbAJWCwB++0hEFzeeQRbC7vdZYP/yEha94s4X1wVFrw= +golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o= +golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw= +golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/model.go b/model.go new file mode 100644 index 0000000..47a9e43 --- /dev/null +++ b/model.go @@ -0,0 +1,97 @@ +package couch + +import ( + "strconv" + "strings" + + "github.com/go-kivik/kivik/v4" + "github.com/invopop/couch/at" +) + +// Model is a standard representation of a model to be stored in CouchDB +// that takes care of the ID, Revision, and adds timestamps. +type Model struct { + ID string `json:"_id,omitempty"` + Rev string `json:"_rev,omitempty"` + + // Attachments keeps together the special list of attachments that belong to + // model. Without this placeholder, they'll get deleted after an update. + Attachments kivik.Attachments `json:"_attachments,omitempty"` + + CreatedAt at.Timestamp `json:"created_at"` + UpdatedAt at.Timestamp `json:"updated_at"` +} + +// UpdateTimestamps ensures the model's created and update at stamps are set. +func (m *Model) UpdateTimestamps() { + if m.CreatedAt.IsZero() { + m.CreatedAt = at.Now() + } + m.UpdatedAt = at.Now() +} + +// Reset sets the rev, created and update at timestamps to zero usually +// so that the same model can be persisted to multiple database without having +// the revision and timestamps copied between instances. +func (m *Model) Reset() { + m.Rev = "" + m.CreatedAt = at.Timestamp{} + m.UpdatedAt = at.Timestamp{} +} + +// GetID provides the document's ID +func (m *Model) GetID() string { + return m.ID +} + +// GetRev provides the document's Revision +func (m *Model) GetRev() string { + return m.Rev +} + +// SetID sets the model's ID +func (m *Model) SetID(id string) { + m.ID = id +} + +// SetRev update's the model's revision ID. The should mainly be +// used by persistence layers. +func (m *Model) SetRev(rev string) { + m.Rev = rev +} + +// Persisted returns true if the revision has been set, a value that should +// always be provided by the database server. +func (m *Model) Persisted() bool { + return m.Rev != "" +} + +// GetCreatedAt provides the model's CreatedAt timestamp in situations where +// the model is being treated as an interface. +// May be zero if the model has not been prepared for persistence. +func (m *Model) GetCreatedAt() at.Timestamp { + return m.CreatedAt +} + +// GetUpdatedAt provides the model's UpdatedAt timestamp in situations where +// the model is being treated as an interface. +// May be zero if the model has not been prepared for persistence. +func (m *Model) GetUpdatedAt() at.Timestamp { + return m.UpdatedAt +} + +// RevAfter returns true if CouchDB revision a is newer than revision b. +// Revisions have the format "-", e.g. "13-a9e7c9c1...". +// This must be used instead of direct string comparison (a > b) because +// lexicographic ordering breaks when sequence numbers cross digit boundaries +// (e.g. "9-xxx" > "13-xxx" is true lexicographically but incorrect). +func RevAfter(a, b string) bool { + aNum, _, _ := strings.Cut(a, "-") + bNum, _, _ := strings.Cut(b, "-") + an, aErr := strconv.Atoi(aNum) + bn, bErr := strconv.Atoi(bNum) + if aErr != nil || bErr != nil { + return a > b + } + return an > bn +} diff --git a/model_test.go b/model_test.go new file mode 100644 index 0000000..bb58ebc --- /dev/null +++ b/model_test.go @@ -0,0 +1,107 @@ +package couch_test + +import ( + "testing" + + "github.com/invopop/couch" + "github.com/stretchr/testify/assert" +) + +func TestModel(t *testing.T) { + m := &couch.Model{} + assert.Empty(t, m.GetID()) + assert.False(t, m.Persisted()) + assert.True(t, m.GetCreatedAt().IsZero()) + + m.SetID("m-1") + m.SetRev("1-abc") + assert.Equal(t, "m-1", m.GetID()) + assert.Equal(t, "1-abc", m.GetRev()) + assert.True(t, m.Persisted()) + + m.UpdateTimestamps() + created := m.GetCreatedAt() + assert.False(t, created.IsZero(), "created stamped") + assert.False(t, m.GetUpdatedAt().IsZero(), "updated stamped") + + m.UpdateTimestamps() + assert.Equal(t, created, m.GetCreatedAt(), "created is not overwritten on re-stamp") + + m.Reset() + assert.Empty(t, m.GetRev()) + assert.True(t, m.GetCreatedAt().IsZero()) + assert.True(t, m.GetUpdatedAt().IsZero()) +} + +func TestRevAfter(t *testing.T) { + tests := []struct { + name string + a string + b string + want bool + }{ + { + name: "single digit greater", + a: "5-abc123", + b: "3-def456", + want: true, + }, + { + name: "single digit less", + a: "3-abc123", + b: "5-def456", + want: false, + }, + { + name: "double digit greater than single digit", + a: "13-abc123", + b: "9-def456", + want: true, + }, + { + name: "single digit less than double digit", + a: "9-abc123", + b: "13-def456", + want: false, + }, + { + name: "equal sequence numbers", + a: "5-abc123", + b: "5-def456", + want: false, + }, + { + name: "large revision numbers", + a: "100-abc123", + b: "99-def456", + want: true, + }, + { + name: "rev 1 vs rev 2", + a: "2-abc123", + b: "1-def456", + want: true, + }, + { + name: "empty strings", + a: "", + b: "", + want: false, + }, + { + name: "malformed falls back to string comparison", + a: "abc", + b: "def", + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := couch.RevAfter(tt.a, tt.b) + if got != tt.want { + t.Errorf("RevAfter(%q, %q) = %v, want %v", tt.a, tt.b, got, tt.want) + } + }) + } +} diff --git a/persist.go b/persist.go new file mode 100644 index 0000000..ba57546 --- /dev/null +++ b/persist.go @@ -0,0 +1,92 @@ +package couch + +import ( + "context" + "errors" + "fmt" + "net/http" + + "github.com/go-kivik/kivik/v4" +) + +// Persistable defines what is expected from a model for it to be +// persisted to the database. +type Persistable interface { + UpdateTimestamps() + GetID() string + GetRev() string + SetID(string) + SetRev(string) +} + +// Fetch wraps around the kivik persistence methods to update the model +// with the data from the database or raise an error if it does not exist. +func Fetch(ctx context.Context, db *kivik.DB, d Persistable) error { + err := db.Get(ctx, d.GetID()).ScanDoc(d) + if err != nil { + return fmt.Errorf("fetch %s/%s: %w", db.Name(), d.GetID(), mapKivikError(err)) + } + return nil +} + +// FetchModel is the now deprecated way of Fetching a model from the database. +func FetchModel(ctx context.Context, db *kivik.DB, m Persistable) error { + return Fetch(ctx, db, m) +} + +// Store attempts to persist the provided persistable object to the database. +func Store(ctx context.Context, db *kivik.DB, m Persistable) error { + if m.GetID() == "" { + return errors.New("cannot store model without ID") + } + m.UpdateTimestamps() + rev, err := db.Put(ctx, m.GetID(), m) + if err != nil { + return fmt.Errorf("put %s/%s: %w", db.Name(), m.GetID(), mapKivikError(err)) + } + m.SetRev(rev) + return nil +} + +// StoreModel is the deprecated way of persisting updates to the database +// and simply wraps around the Store method. +func StoreModel(ctx context.Context, db *kivik.DB, m Persistable) error { + return Store(ctx, db, m) +} + +// Delete removes the provided persistable object from the database. +func Delete(ctx context.Context, db *kivik.DB, m Persistable) error { + if m.GetID() == "" { + return errors.New("cannot delete model without ID") + } + if m.GetRev() == "" { + return errors.New("cannot delete model without rev") + } + rev, err := db.Delete(ctx, m.GetID(), m.GetRev()) + if err != nil { + return fmt.Errorf("delete %s/%s: %w", db.Name(), m.GetID(), mapKivikError(err)) + } + m.SetRev(rev) + return nil +} + +// Errors returned by the persistence helpers, wrapping the underlying +// kivik/CouchDB failure. Match them with errors.Is. +var ( + // ErrNotFound is returned when a document does not exist. + ErrNotFound = errors.New("not found") + // ErrAlreadyExists is returned on a document revision conflict. + ErrAlreadyExists = errors.New("already exists") +) + +func mapKivikError(err error) error { + switch kivik.HTTPStatus(err) { + case http.StatusNotFound: + return fmt.Errorf("%w: %v", ErrNotFound, err) + case http.StatusConflict: + return fmt.Errorf("%w: %v", ErrAlreadyExists, err) + default: + // no alternative mapping yet + return err + } +} diff --git a/shard.go b/shard.go new file mode 100644 index 0000000..543d8ff --- /dev/null +++ b/shard.go @@ -0,0 +1,89 @@ +package couch + +import ( + "fmt" + + "github.com/go-kivik/kivik/v4" +) + +// Shardable defines what we expect from a document, entity, or model that +// we intend to persist to the database. +type Shardable interface { + ShardValue() interface{} +} + +// ShardRules provides the basic details we require to properly handle sharding +// of a type of object. +type ShardRules interface { + // Template provides the base name into which the shard will be inserted. + Template() string + + // List provides an array of acceptable shards + List() []string + + // Key provides a usable string from a shardable value. + Key(v interface{}) (string, error) +} + +// Shards is a special implementation of sharding at the software level. +// The aim is to make it easier to manage a set of separate CouchDB databases +// each of which is used according to sharding details provided. +type Shards struct { + names []string + dbs map[string]*kivik.DB + rules ShardRules +} + +// NewShards instantiates a new sharding wrapper. Databases cannot be assigned +// dynamically, a complete list of databases must be prepared. +func NewShards(client *Client, rules ShardRules) *Shards { + s := new(Shards) + s.names = rules.List() + s.dbs = make(map[string]*kivik.DB) + s.rules = rules + for _, shard := range s.names { + name := fmt.Sprintf(rules.Template(), shard) + s.dbs[shard] = client.DB(name) + } + return s +} + +// For determines which shard to use for the provided "Shardable" model. +func (s *Shards) For(m Shardable) (*kivik.DB, error) { + name, err := s.rules.Key(m.ShardValue()) + if err != nil { + return nil, fmt.Errorf("invalid shard key: %w", err) + } + db, ok := s.dbs[name] + if !ok { + return nil, fmt.Errorf("invalid shard: %v", name) + } + return db, nil +} + +// Get provides the requested database instance, or nil if the name is +// invalid. +func (s *Shards) Get(name string) *kivik.DB { + return s.dbs[name] +} + +// Names provides the complete list of shard names in use. +func (s *Shards) Names() []string { + return s.names +} + +// List provides an array of database objects, in the original shard order. +func (s *Shards) List() []*kivik.DB { + list := make([]*kivik.DB, len(s.names)) + for i, n := range s.names { + list[i] = s.Get(n) + } + return list +} + +// Map provides the map of names to databases to be used for sharding. +// This is especially useful for performing migrations but caution should be +// taken in any other scenario as order is not guaranteed! +func (s *Shards) Map() map[string]*kivik.DB { + return s.dbs +} diff --git a/shard_by_year.go b/shard_by_year.go new file mode 100644 index 0000000..1c4b95d --- /dev/null +++ b/shard_by_year.go @@ -0,0 +1,170 @@ +package couch + +import ( + "encoding/hex" + "errors" + "fmt" + "slices" + "strconv" + "time" +) + +// NewShardByYear provides a common implementation of sharding databases +// according to the year encoded in a time-based UUID (versions 1, 6 and 7). +// For this to work, the model must implement a ShardValue method that returns +// its UUID as a string: +// +// ``` +// +// func (m *Model) ShardValue() interface{} { +// return m.ID +// } +// +// ``` +// +// Shards are always ordered by reverse chronological order, so the newest shards +// are listed first. +// +// This slightly naive implementation assumes that the service will be restarted +// and migrated at least once per year so that the following year's shard is prepared. +// +// If the ShardValue is a string that is not a UUID, it is assumed to be the name +// of the shard and used directly. +func NewShardByYear(name string, start int) *ShardByYear { + s := new(ShardByYear) + s.name = name + s.start = start + s.list = s.generateList() + return s +} + +// NewShardByYearWithStatic creates a new shard rule that supports time-based +// UUIDs **and** random/name-based ones (versions 3, 4 and 5). +// +// The "static" parameter enables support for non-time-based IDs that will +// be persisted to a fixed shard instead of by year. This is useful to being able +// to distinguish between data that is always relevant (static) and data that +// becomes less useful over time. +func NewShardByYearWithStatic(name string, start int, static string) *ShardByYear { + s := new(ShardByYear) + s.name = name + s.start = start + s.static = static + s.list = s.generateList() + return s +} + +// this will fail if the interface is not implemented. +var _ ShardRules = (*ShardByYear)(nil) + +// ShardByYear implements database sharding by year. +type ShardByYear struct { + name string // base database name + static string // static suffix for non-time-based IDs + list []string + start int +} + +// Template provides the base name into which the shard will be inserted. +func (s *ShardByYear) Template() string { + // just add the year to the end of the base name + return fmt.Sprintf("%v_%%s", s.name) +} + +// List provides an array of acceptable shards +func (s *ShardByYear) List() []string { + return s.list +} + +func (s *ShardByYear) generateList() []string { + ym := time.Now().Year() + 1 + yd := ym - s.start + ys := make([]string, yd+1) + for i := 0; i <= yd; i++ { + ys[i] = strconv.Itoa(ym - i) + } + if s.static != "" { + ys = append([]string{s.static}, ys...) + } + return ys +} + +// Key converts the shardable key's value into a usable shard. The value must be +// a string: either a UUID (whose timestamp determines the year) or an +// already-prepared shard name. +func (s *ShardByYear) Key(v any) (string, error) { + str, ok := v.(string) + if !ok { + return "", errors.New("unexpected shard value") + } + b, ok := parseUUID(str) + if !ok { + // Not a UUID — assume the value is already a prepared shard name. + if str == "" { + return "", errors.New("empty shard value") + } + return str, nil + } + if b == ([16]byte{}) { + return "", errors.New("empty shard value") + } + ts, ok := uuidTime(b) + if !ok { + // Random or name-based UUID (v3/v4/v5): carries no timestamp. + if s.static != "" { + return s.static, nil + } + return "", errors.New("invalid shard uuid version") + } + year := strconv.Itoa(ts.Year()) + if slices.Contains(s.List(), year) { + return year, nil + } + return "", fmt.Errorf("shard year %s out of range", year) +} + +// parseUUID decodes the canonical 8-4-4-4-12 hyphenated UUID form into its 16 +// bytes. ok is false when s is not a well-formed UUID. +func parseUUID(s string) (b [16]byte, ok bool) { + if len(s) != 36 || s[8] != '-' || s[13] != '-' || s[18] != '-' || s[23] != '-' { + return b, false + } + clean := s[0:8] + s[9:13] + s[14:18] + s[19:23] + s[24:36] + if _, err := hex.Decode(b[:], []byte(clean)); err != nil { + return b, false + } + return b, true +} + +// gregorianToUnix100ns is the count of 100-nanosecond intervals between the UUID +// (Gregorian) epoch 1582-10-15 and the Unix epoch 1970-01-01. +const gregorianToUnix100ns = 122192928000000000 + +// uuidTime extracts the timestamp encoded in a time-based UUID (versions 1, 6 +// and 7). ok is false for UUIDs that carry no timestamp (versions 3, 4 and 5). +func uuidTime(b [16]byte) (t time.Time, ok bool) { + switch b[6] >> 4 { // version nibble + case 1: + timeLow := uint64(b[0])<<24 | uint64(b[1])<<16 | uint64(b[2])<<8 | uint64(b[3]) + timeMid := uint64(b[4])<<8 | uint64(b[5]) + timeHigh := uint64(b[6]&0x0f)<<8 | uint64(b[7]) + return gregorian(timeHigh<<48 | timeMid<<32 | timeLow), true + case 6: + timeHigh := uint64(b[0])<<24 | uint64(b[1])<<16 | uint64(b[2])<<8 | uint64(b[3]) + timeMid := uint64(b[4])<<8 | uint64(b[5]) + timeLow := uint64(b[6]&0x0f)<<8 | uint64(b[7]) + return gregorian(timeHigh<<28 | timeMid<<12 | timeLow), true + case 7: + ms := int64(b[0])<<40 | int64(b[1])<<32 | int64(b[2])<<24 | + int64(b[3])<<16 | int64(b[4])<<8 | int64(b[5]) + return time.UnixMilli(ms).UTC(), true + default: + return time.Time{}, false + } +} + +// gregorian converts a 60-bit UUID timestamp (100-ns intervals since the +// Gregorian epoch) into a UTC time. +func gregorian(ts uint64) time.Time { + return time.Unix(0, (int64(ts)-gregorianToUnix100ns)*100).UTC() +} diff --git a/shard_by_year_test.go b/shard_by_year_test.go new file mode 100644 index 0000000..d98dcd2 --- /dev/null +++ b/shard_by_year_test.go @@ -0,0 +1,107 @@ +package couch_test + +import ( + "fmt" + "testing" + "time" + + "github.com/invopop/couch" + "github.com/stretchr/testify/assert" +) + +// Fixed UUID vectors with known versions and encoded timestamps. The RFC 9562 +// examples (v1/v6/v7) all encode 2022-02-22T19:22:22Z. +const ( + uuidV1Year2021 = "67cae486-0c2c-11ec-b15b-0242ac130002" // v1 → 2021 + uuidV1Old = "a8098c1a-f86e-11da-bd1a-00112444be1e" // v1 → 2006 (out of range) + uuidV1Year2022 = "c232ab00-9414-11ec-b3c8-9e6bdeced846" // v1 → 2022 (RFC 9562) + uuidV6Year2022 = "1ec9414c-232a-6b00-b3c8-9e6bdeced846" // v6 → 2022 (RFC 9562) + uuidV7Year2022 = "017f22e2-79b0-7cc3-98c4-dc0c0c07398f" // v7 → 2022 (RFC 9562) + uuidV3 = "c2119440-3957-3584-8db9-058632002d07" // v3 (no timestamp) + uuidV4 = "7483ccea-b672-4583-99a7-3797e0505083" // v4 (no timestamp) + uuidV5 = "eed3f9a6-ddb2-5fd1-bd7c-56f357a01984" // v5 (no timestamp) + uuidZero = "00000000-0000-0000-0000-000000000000" +) + +func TestNewShardingByYear(t *testing.T) { + sr := couch.NewShardByYear("test", 2020) + yn := time.Now().Year() + l := (yn + 2) - 2020 + ls := sr.List() + assert.Len(t, ls, l) + assert.Equal(t, fmt.Sprintf("%d", yn+1), ls[0], "first entry") + assert.Equal(t, "2021", ls[len(ls)-2], "last entry") + assert.Equal(t, "test_%s", sr.Template()) +} + +func TestNewShardingByYearWithStatic(t *testing.T) { + sr := couch.NewShardByYearWithStatic("test", 2020, "static") + yn := time.Now().Year() + l := (yn + 2) - 2020 + ls := sr.List() + assert.Len(t, ls, l+1) + assert.Equal(t, "static", ls[0], "static entry") + assert.Equal(t, fmt.Sprintf("%d", yn+1), ls[1], "first entry") + assert.Equal(t, "2021", ls[len(ls)-2], "last entry") + assert.Equal(t, "test_%s", sr.Template()) +} + +func TestShardingByYearKey(t *testing.T) { + sr := couch.NewShardByYear("test", 2020) + + _, err := sr.Key(1234) + assert.ErrorContains(t, err, "unexpected shard value") + + str, err := sr.Key("2021") // already-prepared shard name + assert.NoError(t, err) + assert.Equal(t, "2021", str) + + _, err = sr.Key(uuidZero) + assert.ErrorContains(t, err, "empty shard value") + + _, err = sr.Key("") + assert.ErrorContains(t, err, "empty shard value") + + // Random UUID with no static shard configured. + _, err = sr.Key(uuidV4) + assert.ErrorContains(t, err, "invalid shard uuid version") + + // Time-based UUIDs resolve to their encoded year. + str, err = sr.Key(uuidV1Year2021) + assert.NoError(t, err) + assert.Equal(t, "2021", str) + + for _, id := range []string{uuidV1Year2022, uuidV6Year2022, uuidV7Year2022} { + str, err = sr.Key(id) + assert.NoError(t, err) + assert.Equal(t, "2022", str, id) + } + + _, err = sr.Key(uuidV1Old) + assert.ErrorContains(t, err, "out of range") +} + +func TestShardingByYearWithStaticKey(t *testing.T) { + sr := couch.NewShardByYearWithStatic("test", 2020, "static") + + str, err := sr.Key("2021") + assert.NoError(t, err) + assert.Equal(t, "2021", str) + + _, err = sr.Key(uuidZero) + assert.ErrorContains(t, err, "empty shard value") + + // Non-time-based UUIDs land on the static shard. + for _, id := range []string{uuidV3, uuidV4, uuidV5} { + str, err = sr.Key(id) + assert.NoError(t, err) + assert.Equal(t, "static", str, id) + } + + // Time-based UUIDs still resolve to their year. + for _, id := range []string{uuidV1Year2022, uuidV6Year2022, uuidV7Year2022} { + str, err = sr.Key(id) + assert.NoError(t, err) + assert.Equal(t, "2022", str, id) + } +} diff --git a/shard_test.go b/shard_test.go new file mode 100644 index 0000000..4f7d1b8 --- /dev/null +++ b/shard_test.go @@ -0,0 +1,43 @@ +package couch_test + +import ( + "testing" + + "github.com/invopop/couch" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type shardableDoc struct { + id string +} + +func (d *shardableDoc) ShardValue() interface{} { + return d.id +} + +func TestNewShards(t *testing.T) { + c, _ := couch.New(couch.NewConfig("test")) // nolint:errcheck + sr := couch.NewShardByYear("foo", 2020) + s := couch.NewShards(c, sr) + + assert.Len(t, s.Map(), len(sr.List())) + assert.NotNil(t, s.Map()["2020"]) + + doc := &shardableDoc{id: uuidV1Year2021} + db, err := s.For(doc) + require.NoError(t, err) + assert.Equal(t, "test_foo_2021", db.Name()) + + doc = &shardableDoc{id: uuidV4} // random UUID, no year + db, err = s.For(doc) + assert.Error(t, err) + assert.Contains(t, err.Error(), "invalid shard") + assert.Nil(t, db) + + doc = &shardableDoc{id: uuidV1Old} // year out of range + db, err = s.For(doc) + assert.Error(t, err) + assert.Contains(t, err.Error(), "out of range") + assert.Nil(t, db) +} From 4bb319cf1b812c699e238580d3b39d053665e5b8 Mon Sep 17 00:00:00 2001 From: Sam Lown Date: Wed, 8 Jul 2026 16:42:02 +0000 Subject: [PATCH 2/5] Add CI workflows (lint, test, release) - 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 --- .github/workflows/lint.yaml | 23 +++++++++++++++++++++++ .github/workflows/release.yaml | 34 ++++++++++++++++++++++++++++++++++ .github/workflows/test.yaml | 18 ++++++++++++++++++ 3 files changed, 75 insertions(+) create mode 100644 .github/workflows/lint.yaml create mode 100644 .github/workflows/release.yaml create mode 100644 .github/workflows/test.yaml diff --git a/.github/workflows/lint.yaml b/.github/workflows/lint.yaml new file mode 100644 index 0000000..f82841f --- /dev/null +++ b/.github/workflows/lint.yaml @@ -0,0 +1,23 @@ +name: Lint +on: + push: + tags: + - v* + branches: + - main + pull_request: +jobs: + lint: + name: golangci-lint + runs-on: ubuntu-latest + steps: + - name: Check out code + uses: actions/checkout@v4 + + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version-file: "go.mod" + + - name: Lint + uses: golangci/golangci-lint-action@v8 diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml new file mode 100644 index 0000000..b811d59 --- /dev/null +++ b/.github/workflows/release.yaml @@ -0,0 +1,34 @@ +name: Auto-Version + +on: + push: + branches: + - main + +permissions: + contents: write # the tag action pushes a new tag + +jobs: + tag: + name: Release + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v4 + with: + fetch-depth: "0" + + - name: Get Repo Details + run: | + echo "COMMIT_TYPE=$(echo $GITHUB_REF | cut -d / -f 2)" >> $GITHUB_ENV + echo "REPO_NAME=$(echo $GITHUB_REPOSITORY | cut -d / -f 2-)" >> $GITHUB_ENV + echo "REPO_VERSION=$(echo $GITHUB_REF | cut -d / -f 3-)" >> $GITHUB_ENV + + - name: Bump version and push tag + id: bump + if: env.COMMIT_TYPE != 'tags' + uses: anothrNick/github-tag-action@1.52.0 + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + RELEASE_BRANCHES: main + WITH_V: true diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml new file mode 100644 index 0000000..c6aa18b --- /dev/null +++ b/.github/workflows/test.yaml @@ -0,0 +1,18 @@ +name: Test Go +on: [push] +jobs: + test: + name: Test + runs-on: ubuntu-latest + steps: + - name: Check out code + uses: actions/checkout@v4 + + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version-file: "go.mod" + + # All dependencies are public — no private-module credentials needed. + - name: Test + run: go test -race ./... From d1d2a776c1c60731ffc431991a11aa03f74618a1 Mon Sep 17 00:00:00 2001 From: Sam Lown Date: Wed, 8 Jul 2026 16:50:55 +0000 Subject: [PATCH 3/5] Address PR review (Copilot) + refresh dependencies - 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 --- couch.go | 12 ++++++++---- design.go | 3 +++ design_test.go | 8 ++++++++ go.mod | 6 +++--- go.sum | 13 ++++++------- persist.go | 3 +++ shard.go | 2 ++ shard_by_year.go | 9 +++++---- shard_by_year_test.go | 6 ++++++ 9 files changed, 44 insertions(+), 18 deletions(-) diff --git a/couch.go b/couch.go index dc10119..dddf262 100644 --- a/couch.go +++ b/couch.go @@ -43,17 +43,21 @@ func (c *Client) Ping(ctx context.Context) error { limit := 10 for i := 0; i <= limit; i++ { _, err := c.client.Ping(ctx) - switch kivik.HTTPStatus(err) { - case 0: + if err == nil { return nil - case 408, 504: + } + switch kivik.HTTPStatus(err) { + case 408, 504, 0: + // Transient: request timeouts and transport/network errors + // (status 0) — the server may still be coming up. Retry. select { case <-time.After(1 * time.Second): - continue // try again + continue case <-ctx.Done(): return errors.New("request canceled") } default: + // A definitive HTTP status (e.g. 401): don't retry. return err } } diff --git a/design.go b/design.go index 7c7b6c5..3efc9fd 100644 --- a/design.go +++ b/design.go @@ -88,6 +88,9 @@ func (d *Design) viewConcat() string { sort.Strings(keys) for _, k := range keys { view := d.Views[k] + if view == nil { + continue + } items = append(items, k, "map", view.Map, "reduce", view.Reduce) } return strings.Join(items, ":") diff --git a/design_test.go b/design_test.go index 651ebfc..784eb2e 100644 --- a/design_test.go +++ b/design_test.go @@ -4,8 +4,16 @@ import ( "testing" "github.com/invopop/couch" + "github.com/stretchr/testify/assert" ) +func TestChecksumIgnoresNilView(t *testing.T) { + d := couch.NewDesign("test") + d.SetView("ok", &couch.View{Map: "function(doc) {}"}) + d.Views["broken"] = nil // a nil entry must not panic the checksum + assert.NotPanics(t, func() { _ = d.Checksum() }) +} + func TestDesignInstantiation(t *testing.T) { design := couch.NewDesign("test") if design.ID != "_design/test" { diff --git a/go.mod b/go.mod index a987b79..9907030 100644 --- a/go.mod +++ b/go.mod @@ -16,11 +16,11 @@ require ( github.com/google/go-cmp v0.7.0 // indirect github.com/google/uuid v1.6.0 // indirect github.com/kr/pretty v0.3.1 // indirect - github.com/mattn/go-colorable v0.1.14 // indirect - github.com/mattn/go-isatty v0.0.20 // indirect + github.com/mattn/go-colorable v0.1.15 // indirect + github.com/mattn/go-isatty v0.0.22 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/rogpeppe/go-internal v1.14.1 // indirect golang.org/x/net v0.56.0 // indirect - golang.org/x/sys v0.46.0 // indirect + golang.org/x/sys v0.47.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 9fd8026..7deb7db 100644 --- a/go.sum +++ b/go.sum @@ -21,10 +21,10 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= -github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE= -github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8= -github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= -github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/mattn/go-colorable v0.1.15 h1:+u9SLTRGnXv73cEsnsmoZBom+dMU88B2M0aDcWy0/jY= +github.com/mattn/go-colorable v0.1.15/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8= +github.com/mattn/go-isatty v0.0.22 h1:j8l17JJ9i6VGPUFUYoTUKPSgKe/83EYU2zBC7YNKMw4= +github.com/mattn/go-isatty v0.0.22/go.mod h1:ZXfXG4SQHsB/w3ZeOYbR0PrPwLy+n6xiMrJlRFqopa4= github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= @@ -43,9 +43,8 @@ golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o= golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec= golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= -golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw= -golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= diff --git a/persist.go b/persist.go index ba57546..b77956c 100644 --- a/persist.go +++ b/persist.go @@ -22,6 +22,9 @@ type Persistable interface { // Fetch wraps around the kivik persistence methods to update the model // with the data from the database or raise an error if it does not exist. func Fetch(ctx context.Context, db *kivik.DB, d Persistable) error { + if d.GetID() == "" { + return errors.New("cannot fetch model without ID") + } err := db.Get(ctx, d.GetID()).ScanDoc(d) if err != nil { return fmt.Errorf("fetch %s/%s: %w", db.Name(), d.GetID(), mapKivikError(err)) diff --git a/shard.go b/shard.go index 543d8ff..7ae04eb 100644 --- a/shard.go +++ b/shard.go @@ -16,6 +16,8 @@ type Shardable interface { // of a type of object. type ShardRules interface { // Template provides the base name into which the shard will be inserted. + // It must be an fmt.Sprintf format string with a single %s verb where + // the shard name is substituted (see NewShards). Template() string // List provides an array of acceptable shards diff --git a/shard_by_year.go b/shard_by_year.go index 1c4b95d..9813870 100644 --- a/shard_by_year.go +++ b/shard_by_year.go @@ -78,10 +78,11 @@ func (s *ShardByYear) List() []string { func (s *ShardByYear) generateList() []string { ym := time.Now().Year() + 1 - yd := ym - s.start - ys := make([]string, yd+1) - for i := 0; i <= yd; i++ { - ys[i] = strconv.Itoa(ym - i) + // Iterate newest-first down to start. A start beyond next year simply + // yields no year shards rather than a negative-length panic. + ys := make([]string, 0) + for y := ym; y >= s.start; y-- { + ys = append(ys, strconv.Itoa(y)) } if s.static != "" { ys = append([]string{s.static}, ys...) diff --git a/shard_by_year_test.go b/shard_by_year_test.go index d98dcd2..8dcd01b 100644 --- a/shard_by_year_test.go +++ b/shard_by_year_test.go @@ -46,6 +46,12 @@ func TestNewShardingByYearWithStatic(t *testing.T) { assert.Equal(t, "test_%s", sr.Template()) } +func TestShardByYearFutureStart(t *testing.T) { + // A start year beyond next year must not panic; it yields no year shards. + assert.Empty(t, couch.NewShardByYear("test", 9999).List()) + assert.Equal(t, []string{"static"}, couch.NewShardByYearWithStatic("test", 9999, "static").List()) +} + func TestShardingByYearKey(t *testing.T) { sr := couch.NewShardByYear("test", 2020) From 25b0716ad5d40a946ebf77640939af0a5585867e Mon Sep 17 00:00:00 2001 From: Sam Lown Date: Wed, 8 Jul 2026 17:02:30 +0000 Subject: [PATCH 4/5] Address PR review round 3 (Copilot) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- .github/workflows/test.yaml | 8 +++++++- README.md | 8 ++++---- changes/feed.go | 24 +++++++++++++++++------- changes/feed_test.go | 2 +- couch.go | 25 +++++++++++++------------ design.go | 2 +- 6 files changed, 43 insertions(+), 26 deletions(-) diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index c6aa18b..2d37639 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -1,5 +1,11 @@ name: Test Go -on: [push] +on: + push: + branches: + - main + tags: + - v* + pull_request jobs: test: name: Test diff --git a/README.md b/README.md index c657345..d2dffe8 100644 --- a/README.md +++ b/README.md @@ -71,10 +71,10 @@ func main() { got := &Widget{} got.SetID("widget-1") - if err := couch.Fetch(ctx, db, got); err != nil { - if errors.Is(err, couch.ErrNotFound) { - log.Println("not found") - } + switch err := couch.Fetch(ctx, db, got); { + case errors.Is(err, couch.ErrNotFound): + log.Println("not found") + case err != nil: log.Fatal(err) } } diff --git a/changes/feed.go b/changes/feed.go index fa9a28b..815173c 100644 --- a/changes/feed.go +++ b/changes/feed.go @@ -136,7 +136,7 @@ func New(changesDB, srcDB *kivik.DB, opts ...Option) Feed { Seq: "0", db: srcDB, changesDB: changesDB, - saveTimeout: make(chan bool), + saveTimeout: make(chan bool, 1), count: 0, started: false, opts: newOptions(), @@ -302,6 +302,8 @@ func (f *feed) nextItem(ctx context.Context) (feedItem, error) { return feedItem{}, err } select { + case <-ctx.Done(): + return feedItem{}, ctx.Err() case item, more := <-f.outgoing: if !more { return feedItem{}, nil // the end @@ -310,7 +312,9 @@ func (f *feed) nextItem(ctx context.Context) (feedItem, error) { f.lastID = item.id return item, nil case <-f.saveTimeout: - // force a loop around so we try to save + // Timer fired: clear it so shouldSave() triggers on the next + // loop, then loop around to save. + f.stopSaveTimer() continue } } @@ -335,6 +339,8 @@ func (f *feed) startWithCallbacks() { return f.opts.callback(i.id) }) case <-f.saveTimeout: + // Timer fired: clear it so the save below runs. + f.stopSaveTimer() break maxInFlightLoop } } @@ -342,7 +348,7 @@ func (f *feed) startWithCallbacks() { f.log.Error().Err(err).Msg("closing due to errors") return } - ctx := context.Background() // indpendent context + ctx := context.Background() // independent context if err := f.save(ctx); err != nil { f.log.Error().Err(err).Msg("failed to save sequence position, ignoring") } @@ -366,12 +372,16 @@ func (f *feed) startSaveTimer() { if f.delay != nil { return } + // The callback only nudges the consumer; it must not touch f.delay (that + // would race with the consumer goroutine). The consumer clears the timer + // when it observes the notification. The send is non-blocking against a + // buffered channel, so a fired timer never blocks or leaks its goroutine + // even if no consumer is currently selecting. f.delay = time.AfterFunc(f.opts.storeTimeout, func() { - if f.delay != nil { - f.delay.Stop() - f.delay = nil + select { + case f.saveTimeout <- true: + default: } - f.saveTimeout <- true }) } diff --git a/changes/feed_test.go b/changes/feed_test.go index 855245d..eac9ba2 100644 --- a/changes/feed_test.go +++ b/changes/feed_test.go @@ -12,7 +12,7 @@ import ( func newBareFeed(opts ...Option) *feed { f := &feed{ Seq: "0", - saveTimeout: make(chan bool), + saveTimeout: make(chan bool, 1), opts: newOptions(), fatal: make(chan error, 1), } diff --git a/couch.go b/couch.go index dddf262..4055569 100644 --- a/couch.go +++ b/couch.go @@ -4,7 +4,6 @@ package couch import ( "context" - "errors" "fmt" "time" @@ -40,28 +39,30 @@ func New(conf *Config, opts ...kivik.Option) (*Client, error) { // after the client has been initialized to ensure the connection // is ready to use. func (c *Client) Ping(ctx context.Context) error { - limit := 10 - for i := 0; i <= limit; i++ { - _, err := c.client.Ping(ctx) - if err == nil { + const attempts = 10 + var err error + for i := 0; i < attempts; i++ { + if i > 0 { + // Back off between attempts (not before the first, not after the last). + select { + case <-time.After(1 * time.Second): + case <-ctx.Done(): + return ctx.Err() + } + } + if _, err = c.client.Ping(ctx); err == nil { return nil } switch kivik.HTTPStatus(err) { case 408, 504, 0: // Transient: request timeouts and transport/network errors // (status 0) — the server may still be coming up. Retry. - select { - case <-time.After(1 * time.Second): - continue - case <-ctx.Done(): - return errors.New("request canceled") - } default: // A definitive HTTP status (e.g. 401): don't retry. return err } } - return fmt.Errorf("failed after %d retries", limit) + return fmt.Errorf("couch: ping failed after %d attempts: %w", attempts, err) } // DB is used to provide a database instance at the provided name. diff --git a/design.go b/design.go index 3efc9fd..771dc9f 100644 --- a/design.go +++ b/design.go @@ -24,7 +24,7 @@ type Design struct { Model Language string `json:"language"` - // Options stor additional options for the design document. + // Options stores additional options for the design document. Options map[string]any `json:"options,omitempty"` Filters map[string]string `json:"filters,omitempty"` From 317462670d04d10cce32bfa29ac91fed6f6d532a Mon Sep 17 00:00:00 2001 From: Sam Lown Date: Wed, 8 Jul 2026 17:03:03 +0000 Subject: [PATCH 5/5] Fix test workflow YAML: pull_request needs a trailing colon Co-Authored-By: Claude Opus 4.8 --- .github/workflows/test.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index 2d37639..51960cb 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -5,7 +5,7 @@ on: - main tags: - v* - pull_request + pull_request: jobs: test: name: Test