diff --git a/.github/workflows/lint.yaml b/.github/workflows/lint.yaml index 0ffb95d..fee693e 100644 --- a/.github/workflows/lint.yaml +++ b/.github/workflows/lint.yaml @@ -18,5 +18,5 @@ jobs: - name: golangci-lint uses: golangci/golangci-lint-action@v9 with: - version: v2.11.4 + version: v2.12.2 args: --timeout=4m diff --git a/.golangci.yml b/.golangci.yml index d28fb2f..7b99ab0 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -15,7 +15,7 @@ linters: - godox - goheader - gomoddirectives - - gomodguard + - gomodguard_v2 - goprintffuncname - gosec - importas @@ -40,6 +40,13 @@ linters: - wrapcheck - wsl_v5 settings: + goconst: + # Test fixtures legitimately repeat string literals. + ignore-tests: true + # Common serialization keys and metric label names that mean different + # things in different contexts, so a shared constant would be misleading. + ignore-string-values: + - ^(type|name|index|query|rel)$ godox: keywords: - FIXME diff --git a/go.mod b/go.mod index 94b9222..50cb34f 100644 --- a/go.mod +++ b/go.mod @@ -19,7 +19,7 @@ require ( github.com/opensearch-project/opensearch-go/v2 v2.3.0 github.com/ory/dockertest/v3 v3.12.0 github.com/prometheus/client_golang v1.23.2 - github.com/ttab/elephant-api v0.23.0 + github.com/ttab/elephant-api v0.24.1 github.com/ttab/elephantine v0.27.1 github.com/ttab/eltest v0.2.2 github.com/ttab/flerr v0.1.0 @@ -32,7 +32,6 @@ require ( github.com/twitchtv/twirp v8.1.3+incompatible github.com/urfave/cli/v3 v3.9.0 github.com/viccon/sturdyc v1.1.5 - golang.org/x/exp v0.0.0-20260603202125-055de637280b golang.org/x/oauth2 v0.36.0 golang.org/x/sync v0.20.0 google.golang.org/protobuf v1.36.11 diff --git a/go.sum b/go.sum index 0109d82..764ddb0 100644 --- a/go.sum +++ b/go.sum @@ -277,8 +277,8 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/tinylib/msgp v1.5.0 h1:GWnqAE54wmnlFazjq2+vgr736Akg58iiHImh+kPY2pc= github.com/tinylib/msgp v1.5.0/go.mod h1:cvjFkb4RiC8qSBOPMGPSzSAx47nAsfhLVTCZZNuHv5o= -github.com/ttab/elephant-api v0.23.0 h1:tuZ47+3gqqKOGVaFsWFo2JsEiwjFZgypEvE5s7a9hP0= -github.com/ttab/elephant-api v0.23.0/go.mod h1:4yLd0PwZRjsRj0Lb7QSQPaoYvz0zYZ7fALe1ANaYOhg= +github.com/ttab/elephant-api v0.24.1 h1:nxBM8ZRbAjcqnxhplqP9GuWZ5EwTsK3t4eozLj9niwU= +github.com/ttab/elephant-api v0.24.1/go.mod h1:4yLd0PwZRjsRj0Lb7QSQPaoYvz0zYZ7fALe1ANaYOhg= github.com/ttab/elephantine v0.27.1 h1:kDX9KsPBT4r9DUN3iRqFxhnhA71t3nMTKJbmvCtat14= github.com/ttab/elephantine v0.27.1/go.mod h1:z6v4f7dEiAeOydoHAJ2HkJ6eAjTgvF8HBTOxE73NY1k= github.com/ttab/eltest v0.2.2 h1:T9+XZdLQJQciy7U/Q+TZskaxpnwdRJKEtaSz0699H+4= @@ -333,8 +333,6 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.51.0 h1:IBPXwPfKxY7cWQZ38ZCIRPI50YLeevDLlLnyC5wRGTI= golang.org/x/crypto v0.51.0/go.mod h1:8AdwkbraGNABw2kOX6YFPs3WM22XqI4EXEd8g+x7Oc8= -golang.org/x/exp v0.0.0-20260603202125-055de637280b h1:v1uXiEBHo8QA0LiGCo7UgHMzHT4Kdfpl2zmtH5vaP1Q= -golang.org/x/exp v0.0.0-20260603202125-055de637280b/go.mod h1:d2fgXJLVs4dYDHUk5lwMIfzRzSrWCfGZb0ZqeLa/Vcw= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= diff --git a/index/build.go b/index/build.go index e5d353e..7605b71 100644 --- a/index/build.go +++ b/index/build.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "html" + "slices" "strings" "time" @@ -11,7 +12,6 @@ import ( "github.com/ttab/elephant-index/internal" "github.com/ttab/newsdoc" "github.com/ttab/revisor" - "golang.org/x/exp/slices" ) // Fields can depend on index settings (like custom normalisers). These won't be diff --git a/index/common_test.go b/index/common_test.go index 90a44db..4a8f457 100644 --- a/index/common_test.go +++ b/index/common_test.go @@ -90,7 +90,7 @@ func testingAPIServer( elephantine.LogKeyComponent, "schema-loader"), schemas) test.Must(t, err, "create schema loader") - metrics, err := index.NewMetrics(prometheus.DefaultRegisterer) + metrics, err := index.NewMetrics(reg) test.Must(t, err, "set up metrics") appExited := make(chan struct{}) diff --git a/index/document.go b/index/document.go index aa1d0a5..f43df23 100644 --- a/index/document.go +++ b/index/document.go @@ -1,11 +1,11 @@ package index import ( + "slices" "strconv" "time" "github.com/ttab/newsdoc" - "golang.org/x/exp/slices" ) // DocumentState is the full state that we want to index. diff --git a/index/index.go b/index/index.go index 2098a41..9d18155 100644 --- a/index/index.go +++ b/index/index.go @@ -14,7 +14,6 @@ import ( "github.com/jackc/pgx/v5/pgxpool" "github.com/opensearch-project/opensearch-go/v2" - "github.com/ttab/elephant-api/newsdoc" "github.com/ttab/elephant-api/repository" "github.com/ttab/elephant-index/internal" "github.com/ttab/elephant-index/postgres" @@ -197,8 +196,6 @@ type enrichJob struct { UUID string Operation int State *DocumentState - doc *newsdoc.Document - metadoc *newsdoc.Document fields map[string][]string } diff --git a/index/index_worker.go b/index/index_worker.go index 15e1924..414bfcd 100644 --- a/index/index_worker.go +++ b/index/index_worker.go @@ -125,12 +125,10 @@ func (iw *indexWorker) loop(ctx context.Context) { func (iw *indexWorker) enrich( job *enrichJob, ) (*DocumentState, error) { - state := DocumentState{ - Heads: make(map[string]Status), - } - if job.Operation == opDelete { - return &state, nil + return &DocumentState{ + Heads: make(map[string]Status), + }, nil } ctx, cancel := context.WithTimeout(job.ctx, 5*time.Second) @@ -145,12 +143,6 @@ func (iw *indexWorker) enrich( return nil, fmt.Errorf("get document: %w", err) } - job.doc = docRes.Document - - if docRes.Meta != nil { - job.metadoc = docRes.Meta.Document - } - metaRes, err := iw.idx.documents.GetMeta(ctx, &repository.GetMetaRequest{ Uuid: job.UUID, }) @@ -158,9 +150,28 @@ func (iw *indexWorker) enrich( return nil, fmt.Errorf("get document metadata: %w", err) } - state.CurrentVersion = metaRes.Meta.CurrentVersion - state.WorkflowState = metaRes.Meta.WorkflowState - state.WorkflowCheckpoint = metaRes.Meta.WorkflowCheckpoint + state, err := newDocumentState(docRes, metaRes) + if err != nil { + return nil, err + } + + return state, nil +} + +// newDocumentState assembles the indexable document state from a repository +// document and its metadata. +func newDocumentState( + docRes *repository.GetDocumentResponse, + metaRes *repository.GetMetaResponse, +) (*DocumentState, error) { + state := DocumentState{ + Heads: make(map[string]Status), + CurrentVersion: metaRes.Meta.CurrentVersion, + WorkflowState: metaRes.Meta.WorkflowState, + WorkflowCheckpoint: metaRes.Meta.WorkflowCheckpoint, + Creator: metaRes.Meta.CreatorUri, + Updater: metaRes.Meta.UpdaterUri, + } created, err := time.Parse(time.RFC3339, metaRes.Meta.Created) if err != nil { @@ -170,14 +181,12 @@ func (iw *indexWorker) enrich( modified, err := time.Parse(time.RFC3339, metaRes.Meta.Modified) if err != nil { - return nil, fmt.Errorf("parse document created time: %w", + return nil, fmt.Errorf("parse document modified time: %w", err) } state.Created = created - state.Creator = metaRes.Meta.CreatorUri state.Modified = modified - state.Updater = metaRes.Meta.UpdaterUri for _, v := range metaRes.Meta.Acl { state.ACL = append(state.ACL, ACLEntry{ @@ -193,21 +202,19 @@ func (iw *indexWorker) enrich( name, err) } - status := Status{ + state.Heads[name] = Status{ ID: v.Id, Version: v.Version, Creator: v.Creator, Created: created, Meta: v.Meta, } - - state.Heads[name] = status } - state.Document = newsdoc.DocumentFromRPC(job.doc) + state.Document = newsdoc.DocumentFromRPC(docRes.Document) - if job.metadoc != nil { - md := newsdoc.DocumentFromRPC(job.metadoc) + if docRes.Meta != nil { + md := newsdoc.DocumentFromRPC(docRes.Meta.Document) state.MetaDocument = &md } diff --git a/index/permissions.go b/index/permissions.go index 28272a5..90f376d 100644 --- a/index/permissions.go +++ b/index/permissions.go @@ -13,6 +13,13 @@ const ( ScopeSearch = "search" ) +// HTTP header names. +const ( + // headerAuthorization carries the bearer token. + headerAuthorization = "Authorization" + headerContentType = "Content-Type" +) + func RequireAnyScope(ctx context.Context, scopes ...string) (*elephantine.AuthInfo, error) { auth, ok := elephantine.GetAuthInfo(ctx) if !ok { diff --git a/index/proxy.go b/index/proxy.go index b84aa22..6d41f59 100644 --- a/index/proxy.go +++ b/index/proxy.go @@ -61,7 +61,7 @@ func (ep *ElasticProxy) searchHandler( ) (outErr error) { ctx := r.Context() - authorization := r.Header.Get("Authorization") + authorization := r.Header.Get(headerAuthorization) if authorization == "" { return ElasticErrorf( ErrorTypeUnauthorized, diff --git a/index/schemas.go b/index/schemas.go index c777a64..56e7688 100644 --- a/index/schemas.go +++ b/index/schemas.go @@ -5,13 +5,13 @@ import ( "encoding/json" "fmt" "log/slog" + "maps" "sync" "time" "github.com/ttab/elephant-api/repository" "github.com/ttab/elephantine" "github.com/ttab/revisor" - "golang.org/x/exp/maps" ) type SchemaLoader struct { diff --git a/index/search.go b/index/search.go index f5522f9..bc3be4b 100644 --- a/index/search.go +++ b/index/search.go @@ -189,7 +189,7 @@ func ElasticHandler( ee = e } - w.Header().Set("Content-Type", "application/json") + w.Header().Set(headerContentType, "application/json") w.WriteHeader(ee.Status) enc := json.NewEncoder(w) diff --git a/index/search_api.go b/index/search_api.go index b9dcd46..ecd7896 100644 --- a/index/search_api.go +++ b/index/search_api.go @@ -47,6 +47,8 @@ func NewSearchServiceV1( percChanges *pg.FanOut[PercolatorUpdate], eventPercolated *pg.FanOut[EventPercolated], percDocs PercolatorDocumentGetter, + validator ValidatorSource, + languages LanguageOptions, ) *SearchServiceV1 { return &SearchServiceV1{ log: logger, @@ -57,6 +59,8 @@ func NewSearchServiceV1( percDocs: percDocs, percChanges: percChanges, eventPercolated: eventPercolated, + validator: validator, + languages: languages, subscriptions: sturdyc.New[userSub]( 5000, 5, 30*time.Minute, 10, sturdyc.WithEvictionInterval(10*time.Second), @@ -73,9 +77,198 @@ type SearchServiceV1 struct { percDocs PercolatorDocumentGetter percChanges *pg.FanOut[PercolatorUpdate] eventPercolated *pg.FanOut[EventPercolated] + validator ValidatorSource + languages LanguageOptions subscriptions *sturdyc.Client[userSub] } +// GetFlatDocument implements index.SearchV1. By default it fetches a document +// directly from the repository and returns it in the flattened representation +// used for indexing, bypassing the eventual consistency of OpenSearch. If +// stored is set it instead returns the flattened document as it is currently +// stored in the active index. +func (s *SearchServiceV1) GetFlatDocument( + ctx context.Context, req *index.GetFlatDocumentRequest, +) (*index.GetFlatDocumentResponse, error) { + auth, err := RequireAnyScope(ctx, ScopeSearch, ScopeIndexAdmin) + if err != nil { + return nil, err + } + + if req.Uuid == "" { + return nil, twirp.RequiredArgumentError("uuid") + } + + if req.Stored { + return s.storedFlatDocument(ctx, auth, req) + } + + return s.convertFlatDocument(ctx, auth, req) +} + +// convertFlatDocument fetches the document from the repository and flattens it. +func (s *SearchServiceV1) convertFlatDocument( + ctx context.Context, + auth *elephantine.AuthInfo, req *index.GetFlatDocumentRequest, +) (*index.GetFlatDocumentResponse, error) { + // Forward the authentication header so that the repository applies the + // caller's read permissions. + authCtx, err := twirp.WithHTTPRequestHeaders(ctx, http.Header{ + headerAuthorization: []string{"Bearer " + auth.Token}, + }) + if err != nil { + return nil, twirp.InternalErrorf("invalid header handling: %w", err) + } + + docRes, err := s.documents.Get(authCtx, &repository.GetDocumentRequest{ + Uuid: req.Uuid, + Version: req.Version, + Status: req.Status, + MetaDocument: repository.GetMetaDoc_META_INCLUDE, + }) + if err != nil { + return nil, twirp.InternalErrorf( + "get document from repository: %w", err) + } + + metaRes, err := s.documents.GetMeta(authCtx, &repository.GetMetaRequest{ + Uuid: req.Uuid, + }) + if err != nil { + return nil, twirp.InternalErrorf( + "get document metadata from repository: %w", err) + } + + state, err := newDocumentState(docRes, metaRes) + if err != nil { + return nil, twirp.InternalErrorf("build document state: %w", err) + } + + // Use a per-request resolver as the cache it maintains isn't safe for + // concurrent use. + language, err := NewLanguageResolver(s.languages).GetLanguageInfo( + state.Document.Language) + if err != nil { + return nil, twirp.InvalidArgumentError("language", + fmt.Sprintf("could not resolve document language %q: %v", + state.Document.Language, err)) + } + + flat, err := BuildDocument( + s.validator.GetValidator(), state, GetIndexConfig(language), nil) + if err != nil { + return nil, twirp.InternalErrorf("flatten document: %w", err) + } + + res := index.GetFlatDocumentResponse{ + Fields: make(map[string]*index.FieldValuesV1, len(flat.Fields)), + Document: docRes.Document, + } + + for field, values := range flat.Values() { + res.Fields[field] = &index.FieldValuesV1{ + Values: values, + } + } + + return &res, nil +} + +// storedFlatDocument returns the flattened document as it is currently stored +// in the active index. +func (s *SearchServiceV1) storedFlatDocument( + ctx context.Context, + auth *elephantine.AuthInfo, req *index.GetFlatDocumentRequest, +) (_ *index.GetFlatDocumentResponse, outErr error) { + client, indexSet := s.active.GetActiveIndex() + if client == nil { + return nil, twirp.FailedPrecondition.Error("no active index") + } + + // Look the document up by ID across the active index set. Going through + // NewSearchRequest applies the caller's read restrictions. + query := &index.QueryRequestV1{ + Query: &index.QueryV1{ + Conditions: &index.QueryV1_Term{ + Term: &index.TermQueryV1{Field: "_id", Value: req.Uuid}, + }, + }, + Source: true, + Size: 1, + } + + osReq, err := internal.NewSearchRequest(auth, query) + if err != nil { + return nil, twirp.InternalErrorf("create search request: %w", err) + } + + queryPayload, err := json.Marshal(osReq) + if err != nil { + return nil, twirp.InternalErrorf("marshal opensearch query: %w", err) + } + + res, err := client.Search( + client.Search.WithContext(ctx), + client.Search.WithIndex(internal.IndexPattern(indexSet, query)), + client.Search.WithBody(bytes.NewReader(queryPayload))) + if err != nil { + return nil, twirp.InternalErrorf( + "perform opensearch search request: %w", err) + } + + defer func() { + err := res.Body.Close() + if err != nil { + outErr = errors.Join(outErr, fmt.Errorf( + "close opensearch response body: %w", err)) + } + }() + + dec := json.NewDecoder(res.Body) + + if res.IsError() { + var elasticErr ElasticErrorResponse + + err := dec.Decode(&elasticErr) + if err != nil { + return nil, errors.Join( + fmt.Errorf("opensearch responded with: %s", res.Status()), + fmt.Errorf("decode error response: %w", err), + ) + } + + return nil, twirp.InternalErrorf( + "error response from opensearch: %s", res.Status()) + } + + var response searchResponse + + err = dec.Decode(&response) + if err != nil { + return nil, twirp.InternalErrorf( + "unmarshal opensearch response: %w", err) + } + + if len(response.Hits.Hits) == 0 { + return nil, twirp.NotFoundError( + "no stored document with that UUID in the active index") + } + + hit := response.Hits.Hits[0] + + out := index.GetFlatDocumentResponse{ + Fields: make(map[string]*index.FieldValuesV1, len(hit.Source)), + } + + for field, values := range hit.Source { + out.Fields[field] = &index.FieldValuesV1{ + Values: anySliceToStrings(values), + } + } + + return &out, nil +} + // EndSubscription implements index.SearchV1. func (s *SearchServiceV1) EndSubscription( _ context.Context, _ *index.EndSubscriptionRequest, @@ -815,7 +1008,7 @@ func (s *SearchServiceV1) processSearchResponse( // Forward the authentication header. authCtx, err := twirp.WithHTTPRequestHeaders(ctx, http.Header{ - "Authorization": []string{"Bearer " + auth.Token}, + headerAuthorization: []string{"Bearer " + auth.Token}, }) if err != nil { return nil, twirp.InternalErrorf( diff --git a/index/search_api_test.go b/index/search_api_test.go index d0f4488..df52ec0 100644 --- a/index/search_api_test.go +++ b/index/search_api_test.go @@ -1,15 +1,130 @@ package index_test import ( + "log/slog" + "path/filepath" "testing" + "time" "github.com/golang-jwt/jwt/v5" "github.com/ttab/elephant-api/index" + "github.com/ttab/elephant-api/repository" "github.com/ttab/elephant-index/internal" "github.com/ttab/elephantine" "github.com/ttab/elephantine/test" ) +func TestGetFlatDocument(t *testing.T) { + ctx := t.Context() + logger := slog.New(test.NewLogHandler(t, slog.LevelWarn)) + + tc := testingAPIServer(t, logger) + + documents := repository.NewDocumentsProtobufClient( + tc.Env.Repository.GetAPIEndpoint(), + tc.AuthenticatedClient(t, "doc_read", "doc_write", "eventlog_read")) + + search := index.NewSearchV1ProtobufClient(tc.IndexEndpoint, + tc.AuthenticatedClient(t, "doc_read", "search")) + + docDataDir := filepath.Join("..", "testdata", "documents") + + loadDocuments(t, documents, docDataDir, "russia_v1.json") + + const russiaUUID = "f5d2e4c5-01ba-4dae-9f09-a86701e06ecd" + + // The converted document is fetched directly from the repository, so it's + // available without waiting for OpenSearch to catch up. + live, err := search.GetFlatDocument(ctx, &index.GetFlatDocumentRequest{ + Uuid: russiaUUID, + }) + test.Must(t, err, "get converted flattened document") + + test.Equal(t, russiaUUID, live.Document.Uuid, "returned document UUID") + + test.EqualDiff(t, + []string{"Rysslands ambassadör kallas upp"}, + flatFieldValues(live, "document.title"), + "flattened document title") + + test.EqualDiff(t, + []string{"core://newscoverage/" + russiaUUID}, + flatFieldValues(live, "document.uri"), + "flattened document URI") + + test.EqualDiff(t, []string{"1"}, + flatFieldValues(live, "current_version"), + "flattened current version") + + // The stored document becomes available once indexing has caught up. It + // should be identical to the document converted directly from the repo. + var stored *index.GetFlatDocumentResponse + + deadline := time.After(10 * time.Second) + + for stored == nil { + select { + case <-ctx.Done(): + t.Fatal("cancelled while waiting for the document to be indexed") + case <-deadline: + t.Fatalf("timed out waiting for the document to be indexed,"+ + " last error: %v", err) + case <-time.After(200 * time.Millisecond): + } + + var res *index.GetFlatDocumentResponse + + res, err = search.GetFlatDocument(ctx, &index.GetFlatDocumentRequest{ + Uuid: russiaUUID, + Stored: true, + }) + if err == nil { + stored = res + } + } + + if stored.Document != nil { + t.Error("the stored response should not include the source document") + } + + test.EqualDiff(t, allFlatFields(live), allFlatFields(stored), + "stored fields match the converted fields") +} + +func TestGetFlatDocumentRequiresUUID(t *testing.T) { + ctx := t.Context() + logger := slog.New(test.NewLogHandler(t, slog.LevelWarn)) + + tc := testingAPIServer(t, logger) + + search := index.NewSearchV1ProtobufClient(tc.IndexEndpoint, + tc.AuthenticatedClient(t, "doc_read", "search")) + + _, err := search.GetFlatDocument(ctx, &index.GetFlatDocumentRequest{}) + test.MustNot(t, err, "reject request without a UUID") +} + +func flatFieldValues( + res *index.GetFlatDocumentResponse, name string, +) []string { + f := res.Fields[name] + if f == nil { + return nil + } + + return f.Values +} + +func allFlatFields(res *index.GetFlatDocumentResponse) map[string][]string { + out := make(map[string][]string, len(res.Fields)) + + for name, values := range res.Fields { + out[name] = values.Values + } + + return out +} + func TestIndexPattern(t *testing.T) { test.Equal(t, "documents-foo-*-*", internal.IndexPattern("foo", &index.QueryRequestV1{}), diff --git a/index/server.go b/index/server.go index c739b68..1669c06 100644 --- a/index/server.go +++ b/index/server.go @@ -94,6 +94,7 @@ func RunIndex(ctx context.Context, p Parameters) error { NewPostgresMappingSource(postgres.New(p.Database)), coord, p.AnonymousDocuments, coord.percolatorUpdate, coord.eventPercolated, percDocs, + p.Validator, p.Languages, ), twirp.WithServerJSONSkipDefaults(true), twirp.WithServerHooks(opts.Hooks), @@ -109,7 +110,7 @@ func RunIndex(ctx context.Context, p Parameters) error { AllowInsecureLocalhost: true, Hosts: []string{"localhost", "tt.se"}, AllowedMethods: []string{"GET"}, - AllowedHeaders: []string{"Authorization", "Content-Type"}, + AllowedHeaders: []string{headerAuthorization, headerContentType}, }, proxy) server.Mux.Handle("/", proxyHandler)