diff --git a/README.md b/README.md index 394f114..11c9a48 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # registries -Go library for fetching package metadata from registry APIs. Supports 29 ecosystems with a unified interface. Also provides sub-packages for HTTP client usage (`client/`) and streaming artifact downloads (`fetch/`). +Go library for fetching package metadata from registry APIs. Supports 29 ecosystems with a unified interface. Also provides sub-packages for HTTP client usage (`client/`), streaming artifact downloads (`fetch/`) and registry change feeds (`changes/`). ## Installation @@ -568,6 +568,30 @@ resolver := fetch.NewResolver(fetch.WithAddressPolicy(opts)) client := safehttp.New(nil, opts) ``` +## Change Feeds (`changes/`) + +Some registries publish a feed of the packages that changed since a given point. `changes` reads them for npm (`replicate.npmjs.com`) and Packagist (`metadata/changes.json`), so a cache or mirror can refresh just those packages. + +Feeds are stateless: the caller keeps the cursor, and moves on to the one `Changes` returns only after acting on the changes, so a failure leaves them to be read again. A `Cursor` is a string and can be stored across restarts. + +```go +feed := changes.NewNPM() // or changes.NewPackagist(changes.WithURL(...)) + +cursor, err := feed.Start(ctx) +// later, repeatedly: +c, next, err := feed.Changes(ctx, cursor) +if err != nil { + return err // cursor stays where it was +} +if c.All { + // the feed lost track; treat every package as changed +} +refresh(c.Names) +cursor = next +``` + +Both take `changes.WithClient` for a configured `*client.Client` and `changes.WithURL` for a registry other than the public one. Packagist names the metadata files that changed (`vendor/pkg`, `vendor/pkg~dev`); a deleted package is reported as both. + ## Private Registries PURLs with a `repository_url` qualifier automatically use that URL: diff --git a/changes/changes.go b/changes/changes.go new file mode 100644 index 0000000..5f2b511 --- /dev/null +++ b/changes/changes.go @@ -0,0 +1,80 @@ +// Package changes reads the feeds in which registries list the packages that +// changed since a given point, so a cache or mirror can refresh just those. +// +// Feeds are stateless: the caller keeps the Cursor, starting from Start and +// passing the one Changes returns to the next call. A caller that acts on the +// changes moves on to the new cursor only once it has, so a failure leaves +// the changes to be read again. A Cursor is a plain string and can be stored. +package changes + +import ( + "context" + "net/url" + "strings" + + "github.com/git-pkgs/registries/client" +) + +// Cursor is a position in a feed, as the feed's registry gives it. Treat it +// as opaque. +type Cursor string + +// Changes is what one read of a feed found. +type Changes struct { + // Names are the packages that changed, as the registry names them. + Names []string + // All is set when the feed can't say what changed since the cursor, so + // every package has to be treated as changed. + All bool +} + +// Feed is one registry's change feed. +type Feed interface { + // Ecosystem is the ecosystem the changed names belong to. + Ecosystem() string + // Start returns the feed's current position, from which Changes reports + // what changes next. + Start(ctx context.Context) (Cursor, error) + // Changes returns what changed after cursor, and the cursor to read on + // from. On error the returned cursor is the one passed in. + Changes(ctx context.Context, cursor Cursor) (Changes, Cursor, error) +} + +// Option configures a feed. +type Option func(*options) + +type options struct { + client *client.Client + url string +} + +// WithClient sets the client a feed makes its requests with. The default is +// client.DefaultClient(). +func WithClient(c *client.Client) Option { + return func(o *options) { o.client = c } +} + +// WithURL sets the URL of the feed, for a registry other than the public one. +func WithURL(u string) Option { + return func(o *options) { o.url = u } +} + +func newOptions(defaultURL string, opts []Option) options { + o := options{url: defaultURL} + for _, opt := range opts { + opt(&o) + } + if o.client == nil { + o.client = client.DefaultClient() + } + return o +} + +// withQuery returns base with the query parameters in q added. +func withQuery(base string, q url.Values) string { + sep := "?" + if strings.Contains(base, "?") { + sep = "&" + } + return base + sep + q.Encode() +} diff --git a/changes/changes_test.go b/changes/changes_test.go new file mode 100644 index 0000000..45a8fae --- /dev/null +++ b/changes/changes_test.go @@ -0,0 +1,306 @@ +package changes + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "slices" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + + "github.com/git-pkgs/registries/client" +) + +// testClient returns a client for srv that doesn't retry, so failures show +// at once. +func testClient(srv *httptest.Server) *client.Client { + return client.NewClient(client.WithHTTPClient(srv.Client()), client.WithMaxRetries(0)) +} + +// fakePackagist serves a change feed the way packagist.org does: without +// since, 400 with the current timestamp; with an old since, a resync; with a +// known one, the actions recorded after it. +type fakePackagist struct { + mu sync.Mutex + now int64 + actions map[int64]string // since -> JSON actions + asked []string +} + +func (p *fakePackagist) ServeHTTP(w http.ResponseWriter, r *http.Request) { + p.mu.Lock() + defer p.mu.Unlock() + since := r.URL.Query().Get("since") + p.asked = append(p.asked, since) + w.Header().Set("Content-Type", "application/json") + if since == "" { + w.WriteHeader(http.StatusBadRequest) + _, _ = fmt.Fprintf(w, `{"error":"Invalid or missing since query parameter","timestamp":%d}`, p.now) + return + } + n, _ := strconv.ParseInt(since, 10, 64) + actions, ok := p.actions[n] + if !ok { + actions = `[{"type":"resync","package":"*","time":1}]` + } + _, _ = fmt.Fprintf(w, `{"actions":%s,"timestamp":%d}`, actions, p.now) +} + +func TestPackagist(t *testing.T) { + p := &fakePackagist{now: 100, actions: map[int64]string{ + 100: `[{"type":"update","package":"acme/lib","time":1},{"type":"update","package":"acme/lib~dev","time":1},{"type":"delete","package":"acme/gone","time":1}]`, + 200: `[]`, + }} + srv := httptest.NewServer(p) + defer srv.Close() + feed := NewPackagist(WithClient(testClient(srv)), WithURL(PackagistFeedURL(srv.URL+"/"))) + ctx := context.Background() + + cursor, err := feed.Start(ctx) + if err != nil || cursor != "100" { + t.Fatalf("Start = %q, %v; want 100", cursor, err) + } + p.mu.Lock() + p.now = 200 + p.mu.Unlock() + c, cursor, err := feed.Changes(ctx, cursor) + if err != nil { + t.Fatal(err) + } + if want := []string{"acme/lib", "acme/lib~dev", "acme/gone", "acme/gone~dev"}; !slices.Equal(c.Names, want) || c.All || cursor != "200" { + t.Errorf("changes = %+v, cursor %q; want %v and 200", c, cursor, want) + } + if c, _, err := feed.Changes(ctx, cursor); err != nil || len(c.Names) != 0 || c.All { + t.Errorf("changes without news = %+v, %v", c, err) + } + + // A cursor the feed no longer has history for gets a resync. + if c, _, err := feed.Changes(ctx, "50"); err != nil || !c.All { + t.Errorf("changes after a resync = %+v, %v, want All", c, err) + } + if want := []string{"", "100", "200", "50"}; !slices.Equal(p.asked, want) { + t.Errorf("asked since %q, want %q", p.asked, want) + } + if _, cur, err := feed.Changes(ctx, "garbage"); err == nil || cur != "garbage" { + t.Errorf("invalid cursor: cursor %q, err %v; want an error and the cursor back", cur, err) + } +} + +func TestPackagistErrorReturnsCursor(t *testing.T) { + var fail atomic.Bool + fail.Store(true) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if fail.Load() { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + _, _ = w.Write([]byte(`{"actions":[{"type":"update","package":"acme/lib","time":1}],"timestamp":8}`)) + })) + defer srv.Close() + feed := NewPackagist(WithClient(testClient(srv)), WithURL(srv.URL)) + if _, cursor, err := feed.Changes(context.Background(), "7"); err == nil || cursor != "7" { + t.Fatalf("failing feed: cursor %q, err %v; want an error and cursor 7", cursor, err) + } + fail.Store(false) + if c, cursor, err := feed.Changes(context.Background(), "7"); err != nil || !slices.Equal(c.Names, []string{"acme/lib"}) || cursor != "8" { + t.Errorf("changes after recovery = %+v, %q, %v", c, cursor, err) + } +} + +// fakeReplicate serves a CouchDB-style changes feed over seqs 1..len(ids). +type fakeReplicate struct { + mu sync.Mutex + ids []string + limit int + // failFrom makes requests from this since on fail, when set. + failFrom int + // noLastSeq leaves last_seq out of responses from failFrom on. + noLastSeq bool + asked []string + // token, when set, is required as the token query parameter. + token string +} + +func (f *fakeReplicate) ServeHTTP(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + defer f.mu.Unlock() + w.Header().Set("Content-Type", "application/json") + if f.token != "" && r.URL.Query().Get("token") != f.token { + w.WriteHeader(http.StatusUnauthorized) + return + } + if r.URL.Path == "/registry/" { + _, _ = fmt.Fprintf(w, `{"db_name":"registry","update_seq":%d}`, len(f.ids)) + return + } + if r.URL.Path != "/registry/_changes" { + w.WriteHeader(http.StatusNotFound) + return + } + since, _ := strconv.Atoi(r.URL.Query().Get("since")) + limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) + f.asked = append(f.asked, r.URL.Query().Get("since")) + if f.failFrom > 0 && since >= f.failFrom { + if f.noLastSeq { + _, _ = w.Write([]byte(`{"results":[]}`)) + return + } + w.WriteHeader(http.StatusBadGateway) + return + } + if f.limit > 0 { + limit = min(limit, f.limit) + } + var results []string + last := since + for seq := since + 1; seq <= len(f.ids) && len(results) < limit; seq++ { + results = append(results, fmt.Sprintf(`{"seq":%d,"id":%q,"changes":[{"rev":"1-a"}]}`, seq, f.ids[seq-1])) + last = seq + } + _, _ = fmt.Fprintf(w, `{"results":[%s],"last_seq":%d}`, strings.Join(results, ","), last) +} + +func newTestNPM(srv *httptest.Server) *NPM { + return NewNPM(WithClient(testClient(srv)), WithURL(srv.URL+"/registry/_changes")) +} + +func TestNPM(t *testing.T) { + f := &fakeReplicate{ids: []string{"old"}} + srv := httptest.NewServer(f) + defer srv.Close() + feed := newTestNPM(srv) + ctx := context.Background() + + cursor, err := feed.Start(ctx) + if err != nil || cursor != "1" { + t.Fatalf("Start = %q, %v; want 1", cursor, err) + } + f.mu.Lock() + f.ids = append(f.ids, "lodash", "@scope/pkg", "_design/app") + f.mu.Unlock() + c, cursor, err := feed.Changes(ctx, cursor) + if err != nil { + t.Fatal(err) + } + if want := []string{"lodash", "@scope/pkg"}; !slices.Equal(c.Names, want) || cursor != "4" { + t.Errorf("changes = %v, cursor %q; want %v and 4", c.Names, cursor, want) + } + if c, _, err := feed.Changes(ctx, cursor); err != nil || len(c.Names) != 0 { + t.Errorf("changes without news = %+v, %v", c, err) + } + if want := []string{"1", "4"}; !slices.Equal(f.asked, want) { + t.Errorf("asked since %q, want %q", f.asked, want) + } +} + +// TestNPMQueryInURL checks that a query on the feed URL is kept, for the +// database document Start reads as for the pages. +func TestNPMQueryInURL(t *testing.T) { + f := &fakeReplicate{ids: []string{"old"}, token: "example"} + srv := httptest.NewServer(f) + defer srv.Close() + feed := NewNPM(WithClient(testClient(srv)), WithURL(srv.URL+"/registry/_changes?token=example")) + ctx := context.Background() + + cursor, err := feed.Start(ctx) + if err != nil || cursor != "1" { + t.Fatalf("Start = %q, %v; want 1", cursor, err) + } + f.mu.Lock() + f.ids = append(f.ids, "lodash") + f.mu.Unlock() + c, cursor, err := feed.Changes(ctx, cursor) + if err != nil || !slices.Equal(c.Names, []string{"lodash"}) || cursor != "2" { + t.Errorf("changes = %+v, %q, %v; want lodash and 2", c, cursor, err) + } +} + +// TestNPMPages checks that a backlog longer than a page is read page by page, +// and that one read stops after npmMaxPages. +func TestNPMPages(t *testing.T) { + f := &fakeReplicate{limit: npmPageLimit} + f.ids = make([]string, npmPageLimit*(npmMaxPages+1)+5) + for i := range f.ids { + f.ids[i] = "pkg-" + strconv.Itoa(i) + } + srv := httptest.NewServer(f) + defer srv.Close() + feed := newTestNPM(srv) + + c, cursor, err := feed.Changes(context.Background(), "0") + if err != nil { + t.Fatal(err) + } + if len(c.Names) != npmPageLimit*npmMaxPages || cursor != Cursor(strconv.Itoa(npmPageLimit*npmMaxPages)) { + t.Errorf("first read returned %d names, cursor %q", len(c.Names), cursor) + } + c, _, err = feed.Changes(context.Background(), cursor) + if err != nil { + t.Fatal(err) + } + if len(c.Names) != npmPageLimit+5 { + t.Errorf("second read returned %d names, want %d", len(c.Names), npmPageLimit+5) + } +} + +// TestNPMLaterPageFails checks that when a page after the first fails, or +// comes without last_seq, the earlier pages' names are returned with the +// cursor after them and no error, and a failing first page returns the +// cursor passed in. +func TestNPMLaterPageFails(t *testing.T) { + for _, noLastSeq := range []bool{false, true} { + t.Run(fmt.Sprintf("noLastSeq=%t", noLastSeq), func(t *testing.T) { + f := &fakeReplicate{limit: npmPageLimit, failFrom: npmPageLimit, noLastSeq: noLastSeq} + f.ids = make([]string, npmPageLimit*2) + for i := range f.ids { + f.ids[i] = "pkg-" + strconv.Itoa(i) + } + srv := httptest.NewServer(f) + defer srv.Close() + feed := newTestNPM(srv) + + c, cursor, err := feed.Changes(context.Background(), "0") + if err != nil || len(c.Names) != npmPageLimit || cursor != Cursor(strconv.Itoa(npmPageLimit)) { + t.Errorf("read = %d names, cursor %q, %v; want the first page and its cursor", len(c.Names), cursor, err) + } + if _, cursor, err := feed.Changes(context.Background(), cursor); err == nil || cursor != Cursor(strconv.Itoa(npmPageLimit)) { + t.Errorf("failing first page: cursor %q, err %v; want an error and the cursor back", cursor, err) + } + }) + } +} + +// TestNPMStringSequence checks that a sequence given as a JSON string is +// passed back without its quotes. +func TestNPMStringSequence(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/registry/" { + _, _ = w.Write([]byte(`{"update_seq":"42-g1AAAA"}`)) + return + } + if got := r.URL.Query().Get("since"); got != "42-g1AAAA" { + t.Errorf("since = %q", got) + } + _, _ = w.Write([]byte(`{"results":[{"id":"a"}],"last_seq":"43-g1BBBB"}`)) + })) + defer srv.Close() + feed := newTestNPM(srv) + cursor, err := feed.Start(context.Background()) + if err != nil || cursor != "42-g1AAAA" { + t.Fatalf("Start = %q, %v", cursor, err) + } + if _, cursor, err := feed.Changes(context.Background(), cursor); err != nil || cursor != "43-g1BBBB" { + t.Errorf("Changes cursor = %q, %v", cursor, err) + } +} + +func TestEcosystems(t *testing.T) { + var feeds = []Feed{NewPackagist(), NewNPM()} + if feeds[0].Ecosystem() != "composer" || feeds[1].Ecosystem() != "npm" { + t.Errorf("ecosystems = %q, %q", feeds[0].Ecosystem(), feeds[1].Ecosystem()) + } +} diff --git a/changes/npm.go b/changes/npm.go new file mode 100644 index 0000000..83c4fa0 --- /dev/null +++ b/changes/npm.go @@ -0,0 +1,134 @@ +package changes + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/url" + "strconv" + "strings" +) + +// NPMURL is the change feed of the public npm registry. +const NPMURL = "https://replicate.npmjs.com/registry/_changes" + +// npmPageLimit is the most changes the npm feed returns in one response. +const npmPageLimit = 10000 + +// npmMaxPages bounds how many pages one Changes call reads. The rest is read +// on the next call. +const npmMaxPages = 10 + +// NPM reads the npm registry's CouchDB-style changes feed at +// replicate.npmjs.com. +// +// Its cursor is a sequence, which npm asks followers to treat as opaque; the +// database document at the feed's parent URL, with its trailing slash +// (without it the URL is 404), gives the current one as update_seq. A page +// holds at most 10000 changes (1000 when no limit is given), and Changes +// reads up to ten pages. Each result's id is a package name; design +// documents are left out. +type NPM struct { + o options +} + +// NewNPM returns the feed of the public npm registry, or the one set with +// WithURL. +func NewNPM(opts ...Option) *NPM { + return &NPM{o: newOptions(NPMURL, opts)} +} + +func (*NPM) Ecosystem() string { return "npm" } + +type npmChanges struct { + Results []struct { + ID string `json:"id"` + } `json:"results"` + LastSeq json.RawMessage `json:"last_seq"` +} + +func (f *NPM) Start(ctx context.Context) (Cursor, error) { + var info struct { + UpdateSeq json.RawMessage `json:"update_seq"` + } + dbURL, err := npmDatabaseURL(f.o.url) + if err != nil { + return "", fmt.Errorf("npm change feed: %w", err) + } + if err := f.o.client.GetJSON(ctx, dbURL, &info); err != nil { + return "", fmt.Errorf("npm change feed: %w", err) + } + if len(info.UpdateSeq) == 0 { + return "", errors.New("npm change feed: no update_seq to start from") + } + return seqCursor(info.UpdateSeq), nil +} + +// npmDatabaseURL returns the database document's URL for a feed URL: its +// path without "_changes", keeping the trailing slash and any query. +func npmDatabaseURL(feedURL string) (string, error) { + u, err := url.Parse(feedURL) + if err != nil { + return "", err + } + u.Path = strings.TrimSuffix(u.Path, "_changes") + u.RawPath = strings.TrimSuffix(u.RawPath, "_changes") + return u.String(), nil +} + +// Changes reads pages until one is not full, or up to ten. When a later page +// fails, it returns what the earlier ones found with the cursor after them, +// and no error: the next call goes on from there. +func (f *NPM) Changes(ctx context.Context, cursor Cursor) (Changes, Cursor, error) { + if cursor == "" { + return Changes{}, cursor, errors.New("npm change feed: empty cursor") + } + var changes Changes + next := cursor + for page := range npmMaxPages { + res, err := f.page(ctx, next) + if err != nil { + if page > 0 { + return changes, next, nil + } + return Changes{}, cursor, err + } + for _, r := range res.Results { + if r.ID != "" && !strings.HasPrefix(r.ID, "_design/") { + changes.Names = append(changes.Names, r.ID) + } + } + next = seqCursor(res.LastSeq) + if len(res.Results) < npmPageLimit { + break + } + } + return changes, next, nil +} + +// page reads one page after cursor. +func (f *NPM) page(ctx context.Context, cursor Cursor) (npmChanges, error) { + var res npmChanges + u := withQuery(f.o.url, url.Values{ + "since": {string(cursor)}, + "limit": {strconv.Itoa(npmPageLimit)}, + }) + if err := f.o.client.GetJSON(ctx, u, &res); err != nil { + return npmChanges{}, fmt.Errorf("npm change feed: %w", err) + } + if len(res.LastSeq) == 0 { + return npmChanges{}, errors.New("npm change feed: response without last_seq") + } + return res, nil +} + +// seqCursor returns a sequence as the since parameter takes it: a number as +// is, a string without its quotes. +func seqCursor(seq json.RawMessage) Cursor { + var s string + if json.Unmarshal(seq, &s) == nil { + return Cursor(s) + } + return Cursor(seq) +} diff --git a/changes/packagist.go b/changes/packagist.go new file mode 100644 index 0000000..20d03f7 --- /dev/null +++ b/changes/packagist.go @@ -0,0 +1,100 @@ +package changes + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/url" + "strconv" + "strings" + + "github.com/git-pkgs/registries/client" +) + +// PackagistURL is the metadata change feed of packagist.org. +const PackagistURL = "https://packagist.org/metadata/changes.json" + +// PackagistFeedURL returns where a Packagist instance at apiURL, such as +// https://packagist.org, publishes its metadata change feed. Packagist +// advertises it as "metadata-changes-url" in its packages.json. +func PackagistFeedURL(apiURL string) string { + return strings.TrimSuffix(apiURL, "/") + "/metadata/changes.json" +} + +// Packagist reads Packagist's metadata change feed +// (https://packagist.org/apidoc#track-package-updates). +// +// Its cursor is a timestamp in units of 1/10000 s. Asked without one the +// feed answers 400 with the current timestamp; asked with one older than its +// 24 hours of history it answers with a single "resync" action, reported as +// Changes.All. Each update names the metadata file that changed, +// "vendor/package" or "vendor/package~dev" for its dev versions; a delete +// names the package and is reported as both. +type Packagist struct { + o options +} + +// NewPackagist returns the feed of packagist.org, or of the instance set +// with WithURL (see PackagistFeedURL). +func NewPackagist(opts ...Option) *Packagist { + return &Packagist{o: newOptions(PackagistURL, opts)} +} + +func (*Packagist) Ecosystem() string { return "composer" } + +type packagistChanges struct { + Actions []struct { + Type string `json:"type"` + Package string `json:"package"` + } `json:"actions"` + Timestamp int64 `json:"timestamp"` + Error string `json:"error"` +} + +func (f *Packagist) Start(ctx context.Context) (Cursor, error) { + var start packagistChanges + err := f.o.client.GetJSON(ctx, f.o.url, &start) + var httpErr *client.HTTPError + if errors.As(err, &httpErr) && httpErr.StatusCode == http.StatusBadRequest { + // The expected answer to a request without a cursor. + err = json.Unmarshal([]byte(httpErr.Body), &start) + } + if err != nil { + return "", fmt.Errorf("packagist change feed: %w", err) + } + if start.Timestamp <= 0 { + return "", errors.New("packagist change feed: no timestamp to start from") + } + return Cursor(strconv.FormatInt(start.Timestamp, 10)), nil +} + +func (f *Packagist) Changes(ctx context.Context, cursor Cursor) (Changes, Cursor, error) { + since, err := strconv.ParseInt(string(cursor), 10, 64) + if err != nil || since <= 0 { + return Changes{}, cursor, fmt.Errorf("packagist change feed: invalid cursor %q", cursor) + } + var page packagistChanges + u := withQuery(f.o.url, url.Values{"since": {string(cursor)}}) + if err := f.o.client.GetJSON(ctx, u, &page); err != nil { + return Changes{}, cursor, fmt.Errorf("packagist change feed: %w", err) + } + if page.Error != "" || page.Timestamp <= 0 { + return Changes{}, cursor, fmt.Errorf("packagist change feed: %q", page.Error) + } + var changes Changes + for _, a := range page.Actions { + switch a.Type { + case "update": + changes.Names = append(changes.Names, a.Package) + case "delete": + // A deleted package takes its dev versions with it. + changes.Names = append(changes.Names, a.Package, a.Package+"~dev") + case "resync": + // The cursor is too old for the feed to list what changed. + changes.All = true + } + } + return changes, Cursor(strconv.FormatInt(page.Timestamp, 10)), nil +}