diff --git a/CHANGELOG.md b/CHANGELOG.md index 4fd23289..b5e5548a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed `UniqueOpts.ByArgs` skipping distinct jobs or failing inserts when JSON keys contain path syntax (like `user.id`), are empty, or come from unnamed tags like `json:",omitempty"`. Unaffected unique keys remain unchanged; affected jobs may be inserted again after upgrading or by old and new clients during a rolling upgrade. [PR #1387](https://github.com/riverqueue/river/pull/1387). - Fixed SQLite job list pagination skipping or repeating jobs by formatting cursor timestamps consistently with stored timestamps. [PR #1374](https://github.com/riverqueue/river/pull/1374). - Improved PostgreSQL job listing performance when filtering by one finalized state (`completed`, `cancelled`, or `discarded`) and sorting by finalized time, including in River UI. [PR #1374](https://github.com/riverqueue/river/pull/1374). - Fixed `JobRescuer` overwriting jobs that complete, leave the running state, or are claimed again by another worker after being fetched for rescue, preserving their state, errors, metadata, and timestamps across PostgreSQL and SQLite drivers. Fixes [#1302](https://github.com/riverqueue/river/issues/1302). [PR #1373](https://github.com/riverqueue/river/pull/1373). diff --git a/internal/dbunique/db_unique.go b/internal/dbunique/db_unique.go index cf603731..8cab7aaa 100644 --- a/internal/dbunique/db_unique.go +++ b/internal/dbunique/db_unique.go @@ -2,6 +2,8 @@ package dbunique import ( "crypto/sha256" + "encoding/json" + "errors" "slices" "strings" "time" @@ -12,7 +14,6 @@ import ( "github.com/riverqueue/river/rivershared/structtag" "github.com/riverqueue/river/rivershared/uniquestates" "github.com/riverqueue/river/rivershared/util/ptrutil" - "github.com/riverqueue/river/rivershared/util/sliceutil" "github.com/riverqueue/river/rivertype" ) @@ -93,20 +94,10 @@ func buildUniqueKeyString(timeGen rivertype.TimeGenerator, uniqueOpts *UniqueOpt encodedArgsForUnique = sortedJSONWithOnlyUniqueValues } else { // Use all keys from EncodedArgs sorted alphabetically - keys := sliceutil.Map(gjson.GetBytes(params.EncodedArgs, "@keys").Array(), func(v gjson.Result) string { return v.String() }) - slices.Sort(keys) - - sortedJSON := make([]byte, 0, len(params.EncodedArgs)) - sortedJSON = append(sortedJSON, "{}"...) - sjsonOpts := &sjson.Options{ReplaceInPlace: true} - for _, key := range keys { - sortedJSON, err = sjson.SetRawBytesOptions(sortedJSON, key, []byte(gjson.GetBytes(params.EncodedArgs, key).Raw), sjsonOpts) - if err != nil { - // Should not happen unless key was invalid - return "", err - } + encodedArgsForUnique, err = appendSortedObject(make([]byte, 0, len(params.EncodedArgs)), params.EncodedArgs) + if err != nil { + return "", err } - encodedArgsForUnique = sortedJSON } sb.WriteString("&args=") @@ -129,3 +120,73 @@ func buildUniqueKeyString(timeGen rivertype.TimeGenerator, uniqueOpts *UniqueOpt return sb.String(), nil } + +// appendJSONKey appends key to buf as a JSON string, encoded the same way sjson +// encodes object keys: verbatim between quotes, unless the key contains a byte +// below 0x20 or above 0x7f, `"`, or `\`, in which case it's encoded with +// encoding/json (which also escapes `<`, `>`, `&`, U+2028, and U+2029, and +// replaces invalid UTF-8 with U+FFFD). Matching sjson keeps unique keys +// identical to those of earlier versions, which built unique args with sjson. +func appendJSONKey(buf []byte, key string) []byte { + for i := range len(key) { + if key[i] < ' ' || key[i] > 0x7f || key[i] == '"' || key[i] == '\\' { + encodedKey, _ := json.Marshal(key) //nolint:errchkjson // marshaling a string can't fail + return append(buf, encodedKey...) + } + } + + buf = append(buf, '"') + buf = append(buf, key...) + return append(buf, '"') +} + +// appendSortedObject appends to buf a compact JSON object containing each +// top-level key of encodedObject along with its raw value, sorted by key. If +// a key appears more than once, its first value is used. As in the previous +// path-based implementation, empty input and an empty array produce `{}`; +// other non-object input is rejected. +// +// Keys are walked directly rather than addressed as gjson/sjson paths so that +// keys containing path syntax (like `.`, `@`, or a leading `:`) and empty +// keys are included literally, and distinct values of such keys produce +// distinct output. For all other keys, output is byte-identical to that +// produced by setting each key onto `{}` with sjson. +func appendSortedObject(buf, encodedObject []byte) ([]byte, error) { + type keyValue struct { + key string + rawValue string + } + + var ( + keyValues []keyValue + keysSeen = make(map[string]struct{}) + err error + ) + gjson.ParseBytes(encodedObject).ForEach(func(key, value gjson.Result) bool { + if key.Type != gjson.String { + err = errors.New("unique args must encode a JSON object") + return false + } + if _, ok := keysSeen[key.Str]; !ok { + keysSeen[key.Str] = struct{}{} + keyValues = append(keyValues, keyValue{key: key.Str, rawValue: value.Raw}) + } + return true + }) + if err != nil { + return nil, err + } + + slices.SortFunc(keyValues, func(a, b keyValue) int { return strings.Compare(a.key, b.key) }) + + buf = append(buf, '{') + for i, keyValue := range keyValues { + if i > 0 { + buf = append(buf, ',') + } + buf = appendJSONKey(buf, keyValue.key) + buf = append(buf, ':') + buf = append(buf, keyValue.rawValue...) + } + return append(buf, '}'), nil +} diff --git a/internal/dbunique/db_unique_test.go b/internal/dbunique/db_unique_test.go index 8952df91..bb3f545c 100644 --- a/internal/dbunique/db_unique_test.go +++ b/internal/dbunique/db_unique_test.go @@ -22,6 +22,60 @@ func (a JobArgsStaticKind) Kind() string { return a.kind } +func TestAppendSortedObject(t *testing.T) { + t.Parallel() + + t.Run("Encodes", func(t *testing.T) { + t.Parallel() + + // Compare exact bytes: equivalent JSON with different key escaping or + // whitespace would change the hashes of jobs inserted before upgrading. + for _, tt := range []struct { + name string + encodedObject string + expected string + }{ + {name: "ASCIIHTML", encodedObject: `{"a\u003cb\u0026c\u003e":1}`, expected: `{"a":1}`}, + {name: "Colon", encodedObject: `{"x":2,":x":1}`, expected: `{":x":1,"x":2}`}, + {name: "ControlCharacters", encodedObject: `{"line\nbreak":1,"\u0000":2}`, expected: `{"\u0000":2,"line\nbreak":1}`}, + {name: "DEL", encodedObject: `{"\u007f":1}`, expected: "{\"\x7f\":1}"}, + {name: "DuplicateKeys", encodedObject: `{"a":1,"\u0061":2}`, expected: `{"a":1}`}, + {name: "EmptyArray", encodedObject: `[]`, expected: `{}`}, + {name: "EmptyInput", encodedObject: ``, expected: `{}`}, + {name: "EmptyKey", encodedObject: `{"a":2,"":1}`, expected: `{"":1,"a":2}`}, + {name: "EmptyObject", encodedObject: `{}`, expected: `{}`}, + {name: "EscapedKey", encodedObject: `{"quote\"slash\\":1}`, expected: `{"quote\"slash\\":1}`}, + {name: "LiteralAndNestedKeys", encodedObject: `{"file.name":"a","file":{"name":"b"}}`, expected: `{"file":{"name":"b"},"file.name":"a"}`}, + {name: "NumericKeys", encodedObject: `{"10":2,"0":1}`, expected: `{"0":1,"10":2}`}, + {name: "RawValues", encodedObject: `{"b":null,"a":{"y":1, "x":[1, 2]}}`, expected: `{"a":{"y":1, "x":[1, 2]},"b":null}`}, + {name: "UnicodeHTML", encodedObject: `{"\u00e9\u003c":1}`, expected: `{"é\u003c":1}`}, + {name: "UnicodeSeparators", encodedObject: `{"\u2029":2,"\u2028":1}`, expected: `{"\u2028":1,"\u2029":2}`}, + } { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + encoded, err := appendSortedObject(nil, []byte(tt.encodedObject)) + require.NoError(t, err) + require.Equal(t, tt.expected, string(encoded)) + }) + } + }) + + t.Run("RejectsNonObjectArgs", func(t *testing.T) { + t.Parallel() + + for _, encodedArgs := range []string{`null`, `[1,2]`, `"str"`, `true`, `123`} { + args := JobArgsStaticKind{kind: "kind"} + _, err := UniqueKey(&riversharedtest.TimeStub{}, &UniqueOpts{ByArgs: true}, &rivertype.JobInsertParams{ + Args: args, + EncodedArgs: []byte(encodedArgs), + Kind: args.Kind(), + }) + require.EqualError(t, err, "unique args must encode a JSON object", "encoded args: %s", encodedArgs) + } + }) +} + func TestUniqueKey(t *testing.T) { t.Parallel() @@ -101,6 +155,88 @@ func TestUniqueKey(t *testing.T) { uniqueOpts: UniqueOpts{ByArgs: true}, expectedJSON: `&kind=worker_1&args={"Recipient":"john@example.com","Subject":"Another Test Email"}`, }, + { + name: "ByArgsUniqueWithUnnamedJSONTagUsesFieldName", + argsFunc: func() rivertype.JobArgs { + //nolint:tagliatelle // non-snake keys are intentional + type EmailJobArgs struct { + JobArgsStaticKind + + Recipient string `json:",omitempty" river:"unique"` + Subject string `json:"subject" river:"unique"` + TemplateID int + } + return EmailJobArgs{ + JobArgsStaticKind: JobArgsStaticKind{kind: "worker_1"}, + Recipient: "john@example.com", + Subject: "Another Test Email", + TemplateID: 102, + } + }, + uniqueOpts: UniqueOpts{ByArgs: true}, + expectedJSON: `&kind=worker_1&args={"Recipient":"john@example.com","subject":"Another Test Email"}`, + }, + { + name: "ByArgsWithCompatibleJSONNames", + argsFunc: func() rivertype.JobArgs { + //nolint:tagliatelle // Exercise names whose escaped sort order differs. + type Nested struct { + Amount int `json:"a&b" river:"unique"` + Upper string `json:"Y" river:"unique"` + } + //nolint:tagliatelle // Exercise valid names whose escaping must not change hashes. + type Args struct { + JobArgsStaticKind + + Dollar string `json:"$x" river:"unique"` + Nested Nested `json:"nested"` + Unicode string `json:"é<" river:"unique"` + Upper string `json:"Y" river:"unique"` + } + return Args{ + JobArgsStaticKind: JobArgsStaticKind{kind: "worker_1"}, + Dollar: "dollar", + Nested: Nested{Amount: 1, Upper: "nested"}, + Unicode: "unicode", + Upper: "upper", + } + }, + uniqueOpts: UniqueOpts{ByArgs: true}, + // Preserve the old byte ordering even though escaped `\$x` sorts after `Y`. + expectedJSON: `&kind=worker_1&args={"$x":"dollar","Y":"upper","nested":{"Y":"nested","a&b":1},"é\u003c":"unicode"}`, + }, + { + name: "ByArgsUniqueWithPathSyntaxInJSONTags", + argsFunc: func() rivertype.JobArgs { + type Nested struct { + Value string `json:"inner@key" river:"unique"` + } + //nolint:tagliatelle // non-snake keys are intentional + type PathSyntaxJobArgs struct { + JobArgsStaticKind + + Bang string `json:"!bang" river:"unique"` + Colon string `json:":x" river:"unique"` + Email string `json:"alice@example.com" river:"unique"` + Literal string `json:"outer.key.inner@key" river:"unique"` + Nested Nested `json:"outer.key"` + UserID string `json:"user.id" river:"unique"` + X string `json:"x" river:"unique"` + } + return PathSyntaxJobArgs{ + JobArgsStaticKind: JobArgsStaticKind{kind: "worker_1"}, + Bang: "bang", + Colon: "colon", + Email: "email", + Literal: "literal", + Nested: Nested{Value: "nested"}, + UserID: "u1", + X: "x", + } + }, + uniqueOpts: UniqueOpts{ByArgs: true}, + expectedJSON: `&kind=worker_1&args={"!bang":"bang",":x":"colon","alice@example.com":"email","outer.key":{"inner@key":"nested"},"outer.key.inner@key":"literal","user.id":"u1","x":"x"}`, + }, { name: "ByArgsWithPointerToStruct", argsFunc: func() rivertype.JobArgs { @@ -344,6 +480,15 @@ func TestUniqueKey(t *testing.T) { // args JSON should be sorted alphabetically: expectedJSON: `&kind=worker_3&args={"count":10,"description":"A generic job without unique fields."}`, }, + { + name: "ByArgsWithNoUniqueFieldsAndLiteralKeys", + argsFunc: func() rivertype.JobArgs { return JobArgsStaticKind{kind: "worker_3"} }, + modifyInsertParamsFunc: func(params *rivertype.JobInsertParams) { + params.EncodedArgs = []byte(`{"x":"x","file.name":"file","alice@example.com":"email",":x":"colon","[x":"bracket","{x":"brace","":"empty"}`) + }, + uniqueOpts: UniqueOpts{ByArgs: true}, + expectedJSON: `&kind=worker_3&args={"":"empty",":x":"colon","[x":"bracket","alice@example.com":"email","file.name":"file","x":"x","{x":"brace"}`, + }, { name: "ByArgsWithEmptyEncodedArgs", argsFunc: func() rivertype.JobArgs { diff --git a/riverdriver/riverdrivertest/driver_client_test.go b/riverdriver/riverdrivertest/driver_client_test.go index f1d720eb..87a0cf41 100644 --- a/riverdriver/riverdrivertest/driver_client_test.go +++ b/riverdriver/riverdrivertest/driver_client_test.go @@ -3,6 +3,8 @@ package riverdrivertest import ( "context" "database/sql" + "encoding/json" + "maps" "math" "slices" "testing" @@ -230,6 +232,16 @@ func TestClientWithDriverRiverTurso(t *testing.T) { ) } +// customJSONArgs are job args encoded as an arbitrary JSON object, like args +// with a custom MarshalJSON implementation. +type customJSONArgs struct { + values map[string]string +} + +func (customJSONArgs) Kind() string { return "customJSON" } + +func (a customJSONArgs) MarshalJSON() ([]byte, error) { return json.Marshal(a.values) } + type noOpArgs struct { Name string `json:"name"` } @@ -418,6 +430,129 @@ func ExerciseClient[TTx any](ctx context.Context, t *testing.T, require.Equal(t, rivertype.JobStateCancelled, event.Job.State) }) + // Keys containing gjson/sjson path syntax (and the empty key) are distinct + // keys when unique by all args, so args differing in their values aren't + // duplicates. + t.Run("InsertUniqueByArgsAllArgsWithPathSyntaxKeys", func(t *testing.T) { + t.Parallel() + + client, bundle := setup(t) + + river.AddWorker(bundle.config.Workers, river.WorkFunc(func(ctx context.Context, job *river.Job[customJSONArgs]) error { + return nil + })) + + insert := func(t *testing.T, values map[string]string) *rivertype.JobInsertResult { + t.Helper() + + insertRes, err := client.Insert(ctx, customJSONArgs{values: values}, &river.InsertOpts{ + UniqueOpts: river.UniqueOpts{ByArgs: true}, + }) + require.NoError(t, err) + return insertRes + } + + var ( + keys = []string{"", "!x", ":x", "[x", "alice@example.com", "file.name", "x*?", "x#", "x|", `x\y`, "{x"} + baseValues = map[string]string{"x": "x"} + ) + for _, key := range keys { + baseValues[key] = "value" + } + + insertRes0 := insert(t, baseValues) + require.False(t, insertRes0.UniqueSkippedAsDuplicate) + + insertRes1 := insert(t, maps.Clone(baseValues)) + require.True(t, insertRes1.UniqueSkippedAsDuplicate) + require.Equal(t, insertRes0.Job.ID, insertRes1.Job.ID) + + for _, key := range keys { + values := maps.Clone(baseValues) + values[key] = "other" + + insertRes := insert(t, values) + require.False(t, insertRes.UniqueSkippedAsDuplicate, "key: %q", key) + } + }) + + // Unique fields whose JSON keys contain gjson/sjson path syntax, or whose + // `json` tag has no name, are part of the unique key. + t.Run("InsertUniqueByArgsUniqueFieldsWithPathSyntaxKeys", func(t *testing.T) { + t.Parallel() + + client, bundle := setup(t) + + type User struct { + ID string `json:"id" river:"unique"` + } + + //nolint:tagliatelle // non-snake keys are intentional + type JobArgs struct { + testutil.JobArgsReflectKind[JobArgs] + + Bang string `json:"!bang" river:"unique"` + Colon string `json:":colon" river:"unique"` + Email string `json:"alice@example.com" river:"unique"` + Other string `json:"other"` + Recipient string `json:",omitempty" river:"unique"` + User User `json:"user"` + UserID string `json:"user.id" river:"unique"` + Wildcard string `json:"wild*?" river:"unique"` + } + + river.AddWorker(bundle.config.Workers, river.WorkFunc(func(ctx context.Context, job *river.Job[JobArgs]) error { + return nil + })) + + insert := func(t *testing.T, args *JobArgs) *rivertype.JobInsertResult { + t.Helper() + + insertRes, err := client.Insert(ctx, args, &river.InsertOpts{ + UniqueOpts: river.UniqueOpts{ByArgs: true}, + }) + require.NoError(t, err) + return insertRes + } + + baseArgs := JobArgs{ + Bang: "bang", + Colon: "colon", + Email: "email", + Other: "other", + Recipient: "recipient", + User: User{ID: "nested"}, + UserID: "u1", + Wildcard: "wildcard", + } + + insertRes0 := insert(t, &baseArgs) + require.False(t, insertRes0.UniqueSkippedAsDuplicate) + + // A change to a field that isn't unique is still a duplicate. + args := baseArgs + args.Other = "changed" + insertRes1 := insert(t, &args) + require.True(t, insertRes1.UniqueSkippedAsDuplicate) + require.Equal(t, insertRes0.Job.ID, insertRes1.Job.ID) + + for _, modify := range []func(args *JobArgs){ + func(args *JobArgs) { args.Bang = "changed" }, + func(args *JobArgs) { args.Colon = "changed" }, + func(args *JobArgs) { args.Email = "changed" }, + func(args *JobArgs) { args.Recipient = "changed" }, + func(args *JobArgs) { args.User.ID = "changed" }, + func(args *JobArgs) { args.UserID = "u2" }, + func(args *JobArgs) { args.Wildcard = "changed" }, + } { + args := baseArgs + modify(&args) + + insertRes := insert(t, &args) + require.False(t, insertRes.UniqueSkippedAsDuplicate, "args: %+v", args) + } + }) + t.Run("InsertUniqueByPeriod", func(t *testing.T) { t.Parallel() diff --git a/rivershared/structtag/struct_tag.go b/rivershared/structtag/struct_tag.go index 66ce8a37..dfac05d5 100644 --- a/rivershared/structtag/struct_tag.go +++ b/rivershared/structtag/struct_tag.go @@ -1,9 +1,10 @@ package structtag import ( + "cmp" "fmt" "reflect" - "sort" + "slices" "strings" "sync" @@ -12,7 +13,9 @@ import ( "github.com/riverqueue/river/rivertype" ) -// ExtractValues extracts the raw JSON values of the specified keys from the JSON-encoded args. +// ExtractValues extracts the raw JSON values of the specified keys from the +// JSON-encoded args. Keys are gjson paths like those returned by +// SortedFieldsWithTag. func ExtractValues(encodedArgs []byte, uniqueKeys []string) []string { // Use GetManyBytes to retrieve multiple values at once results := gjson.GetManyBytes(encodedArgs, uniqueKeys...) @@ -32,6 +35,30 @@ func ExtractValues(encodedArgs []byte, uniqueKeys []string) []string { return uniqueValues } +// fieldPath is the location of a tagged field within a struct's JSON encoding. +type fieldPath struct { + // path is a gjson/sjson path to the field, with each component escaped so + // that JSON keys containing path syntax are treated literally. + path string + + // sortKey is the unescaped components joined with `.`. Paths are sorted by + // it so that fields keep the order they had before components were + // escaped, which keeps unique keys stable for ordinary field names. + sortKey string +} + +func newFieldPath(components []string) fieldPath { + escaped := make([]string, len(components)) + for i, component := range components { + escaped[i] = escapePathComponent(component) + } + + return fieldPath{ + path: strings.Join(escaped, "."), + sortKey: strings.Join(components, "."), + } +} + type uniqueFieldCacheKey struct { typ reflect.Type tagValue string @@ -46,6 +73,11 @@ var ( // SortedFieldsWithTag retrieves unique fields with caching to avoid // extracting fields from the same struct type repeatedly. +// +// Fields are returned as gjson/sjson paths suitable for use with ExtractValues +// and sjson. Each path component is escaped so a JSON key containing path +// syntax like `.`, `@`, or `*` addresses that key literally. Paths are sorted +// by their unescaped, dot-joined JSON keys. func SortedFieldsWithTag(args rivertype.JobArgs, tagValue string) ([]string, error) { var ( typ = reflect.TypeOf(args) @@ -61,7 +93,7 @@ func SortedFieldsWithTag(args rivertype.JobArgs, tagValue string) ([]string, err cacheMutex.RUnlock() // Not in cache; retrieve using reflection - fields, err := sortedFieldsWithTagUncached(reflect.TypeOf(args), tagValue, nil, make(map[reflect.Type]struct{})) + fields, err := sortedFieldsWithTagUncached(typ, tagValue) if err != nil { return nil, err } @@ -74,14 +106,37 @@ func SortedFieldsWithTag(args rivertype.JobArgs, tagValue string) ([]string, err return fields, nil } -// sortedFieldsWithTagUncached uses reflection to retrieve the JSON keys of fields +// sortedFieldsWithTagUncached uses reflection to retrieve the escaped JSON +// paths of fields marked with `river:""`, sorted by their unescaped +// JSON keys. +func sortedFieldsWithTagUncached(typ reflect.Type, tagValue string) ([]string, error) { + fieldPaths, err := fieldPathsWithTag(typ, tagValue, nil, make(map[reflect.Type]struct{})) + if err != nil { + return nil, err + } + + // Sort by unescaped keys for consistent ordering that matches the order + // used before path components were escaped. Break ties (possible when a + // key containing `.` collides with a nested path) by escaped path. + slices.SortFunc(fieldPaths, func(a, b fieldPath) int { + return cmp.Or(strings.Compare(a.sortKey, b.sortKey), strings.Compare(a.path, b.path)) + }) + + var fields []string + for _, fieldPath := range fieldPaths { + fields = append(fields, fieldPath.path) + } + return fields, nil +} + +// fieldPathsWithTag uses reflection to retrieve the JSON paths of fields // marked with `river:""` among potentially other comma-separated -// values. The return values are the JSON keys using the same logic as the -// `json` struct tag. +// values. Path components are the JSON keys using the same logic as the `json` +// struct tag. Results are unsorted. // // typesSeen should be a map passed through to make sure that recursive types // don't cause a stack overflow. -func sortedFieldsWithTagUncached(typ reflect.Type, tagValue string, path []string, typesSeen map[reflect.Type]struct{}) ([]string, error) { +func fieldPathsWithTag(typ reflect.Type, tagValue string, path []string, typesSeen map[reflect.Type]struct{}) ([]fieldPath, error) { // Handle pointer to struct if typ.Kind() == reflect.Pointer { typ = typ.Elem() @@ -100,7 +155,7 @@ func sortedFieldsWithTagUncached(typ reflect.Type, tagValue string, path []strin } typesSeen[typ] = struct{}{} - var uniqueFields []string + var uniqueFields []fieldPath // Iterate over all fields for field := range typ.Fields() { @@ -108,19 +163,8 @@ func sortedFieldsWithTagUncached(typ reflect.Type, tagValue string, path []strin continue } - var uniqueName string - { - // Get the corresponding JSON key - jsonTag := field.Tag.Get("json") - - if jsonTag == "" { - // If no JSON tag, use the field name as-is - uniqueName = field.Name - } else { - // Handle cases like `json:"recipient,omitempty"` - uniqueName = parseJSONTag(jsonTag) - } - } + // Get the corresponding JSON key + uniqueName := parseJSONTag(field.Name, field.Tag.Get("json")) // Check for `river:"unique"` tag, possibly among other comma-separated values var hasUniqueTag bool @@ -142,7 +186,7 @@ func sortedFieldsWithTagUncached(typ reflect.Type, tagValue string, path []strin fullPath = append(path, uniqueName) //nolint:gocritic } - uniqueSubFields, err := sortedFieldsWithTagUncached(field.Type, tagValue, fullPath, typesSeen) + uniqueSubFields, err := fieldPathsWithTag(field.Type, tagValue, fullPath, typesSeen) if err != nil { return nil, err } @@ -154,31 +198,48 @@ func sortedFieldsWithTagUncached(typ reflect.Type, tagValue string, path []strin // JSON serialization as a unique value. This may not be the // greatest idea practically, but keeping it in place for // backwards compatibility. - uniqueFields = append(uniqueFields, strings.Join(append(path, uniqueName), ".")) + uniqueFields = append(uniqueFields, newFieldPath(append(path, uniqueName))) } continue } if hasUniqueTag { - uniqueFields = append(uniqueFields, strings.Join(append(path, uniqueName), ".")) + uniqueFields = append(uniqueFields, newFieldPath(append(path, uniqueName))) } } - // Sort the uniqueFields alphabetically for consistent ordering - sort.Strings(uniqueFields) - return uniqueFields, nil } -// parseJSONTag extracts the JSON key from the struct tag. -// It handles tags with options, e.g., `json:"recipient,omitempty"`. -func parseJSONTag(tag string) string { +// escapePathComponent escapes a JSON object key for use as a single component +// of a gjson or sjson path so that it addresses the key literally. +// +// gjson.Escape handles path separators, wildcards, modifiers, and queries. +// Also escape a leading colon so sjson doesn't strip it as an object-key +// directive. Both libraries unescape a backslash before any byte. +func escapePathComponent(key string) string { + escaped := gjson.Escape(key) + if strings.HasPrefix(escaped, ":") { + return `\` + escaped + } + return escaped +} + +// parseJSONTag returns the JSON key for a field with the given name and `json` +// struct tag. It handles tags with options, e.g., `json:"recipient,omitempty"`, +// and like encoding/json falls back to the field name when the tag has no name +// (e.g. `json:",omitempty"`). +// +// Preserve the historical handling of invalid JSON tag names; only the empty +// name falls back to the field name here. +func parseJSONTag(fieldName, tag string) string { // Tags can be like "recipient,omitempty", so split by comma - if before, _, ok := strings.Cut(tag, ","); ok { - return before + name, _, _ := strings.Cut(tag, ",") + if name == "" { + return fieldName } - return tag + return name } func typeStructOrPointerToStruct(typ reflect.Type) bool { diff --git a/rivershared/structtag/struct_tag_test.go b/rivershared/structtag/struct_tag_test.go index 3a49c574..822a90a5 100644 --- a/rivershared/structtag/struct_tag_test.go +++ b/rivershared/structtag/struct_tag_test.go @@ -9,6 +9,38 @@ import ( "github.com/riverqueue/river/rivertype" ) +func TestEscapePathComponent(t *testing.T) { + t.Parallel() + + for _, tt := range []struct { + key string + expected string + }{ + {key: "", expected: ""}, + {key: "user_id", expected: "user_id"}, + {key: "user-id", expected: "user-id"}, + {key: "UserID", expected: "UserID"}, + {key: "123", expected: "123"}, + {key: "-1", expected: "-1"}, + {key: "a b", expected: "a b"}, + {key: "éλ", expected: "éλ"}, + {key: "a:b", expected: "a:b"}, + {key: ":x", expected: `\:x`}, + {key: "::x", expected: `\::x`}, + {key: "user.id", expected: `user\.id`}, + {key: "alice@example.com", expected: `alice\@example\.com`}, + {key: "!x", expected: `\!x`}, + {key: "[x]", expected: `\[x\]`}, + {key: "{x}", expected: `\{x\}`}, + {key: "a*b?c#d|e", expected: `a\*b\?c\#d\|e`}, + {key: `a\b`, expected: `a\\b`}, + {key: ":a.b", expected: `\:a\.b`}, + {key: "$x", expected: `\$x`}, + } { + require.Equal(t, tt.expected, escapePathComponent(tt.key), "key: %q", tt.key) + } +} + func TestExtractValues(t *testing.T) { t.Parallel() @@ -78,6 +110,12 @@ func TestExtractValues(t *testing.T) { uniqueKeys: []string{"a", "b"}, expectedValues: []string{"undefined", "undefined"}, }, + { + name: "EscapedPathComponents", + encodedArgs: []byte(`{"user.id":"u1","alice@example.com":2,":x":3,"x":4,"user":{"id":"nested"}}`), + uniqueKeys: []string{`user\.id`, `alice\@example\.com`, `\:x`, "x", "user.id"}, + expectedValues: []string{`"u1"`, "2", "3", "4", `"nested"`}, + }, } for _, tt := range testCases {