diff --git a/internal/cursor/cursor.go b/internal/cursor/cursor.go new file mode 100644 index 000000000..b59d38ccb --- /dev/null +++ b/internal/cursor/cursor.go @@ -0,0 +1,80 @@ +// Package cursor provides a unified cursor encoding/decoding utility for pagination. +// Cursors are versioned and resource-scoped, allowing different parts of the system +// to use cursors without collision. +package cursor + +import ( + "errors" + "fmt" + "math/big" + "strings" +) + +var ( + // ErrInvalidCursor indicates the cursor is malformed or cannot be decoded. + ErrInvalidCursor = errors.New("invalid cursor") + + // ErrVersionMismatch indicates the cursor version doesn't match the expected version. + ErrVersionMismatch = errors.New("cursor version mismatch") +) + +// Base62Encode encodes a string to base62. +func Base62Encode(s string) string { + if s == "" { + return "" + } + num := new(big.Int) + num.SetBytes([]byte(s)) + return num.Text(62) +} + +// Base62Decode decodes a base62 string. +func Base62Decode(s string) (string, error) { + if s == "" { + return "", nil + } + num := new(big.Int) + num, ok := num.SetString(s, 62) + if !ok { + return "", ErrInvalidCursor + } + return string(num.Bytes()), nil +} + +// Encode creates a versioned cursor string. +// Format: {resource}v{version:02d}:{data}, then base62 encoded. +// Example: "evtv01:position_data" -> base62 +func Encode(resource string, version int, data string) string { + raw := fmt.Sprintf("%sv%02d:%s", resource, version, data) + return Base62Encode(raw) +} + +// Decode decodes and validates a cursor string. +// Returns the data portion if the cursor matches the expected resource and version. +// Returns ErrInvalidCursor if the cursor is malformed. +// Returns ErrVersionMismatch if the version doesn't match. +func Decode(encoded string, resource string, version int) (string, error) { + if encoded == "" { + return "", nil + } + + raw, err := Base62Decode(encoded) + if err != nil { + return "", err + } + + // Expected prefix: {resource}v{version:02d}: + expectedPrefix := fmt.Sprintf("%sv%02d:", resource, version) + + if !strings.HasPrefix(raw, expectedPrefix) { + // Check if it's a version mismatch vs completely invalid + resourcePrefix := resource + "v" + if strings.HasPrefix(raw, resourcePrefix) { + // Has correct resource but wrong version + return "", fmt.Errorf("%w: expected version %02d", ErrVersionMismatch, version) + } + return "", ErrInvalidCursor + } + + return raw[len(expectedPrefix):], nil +} diff --git a/internal/cursor/cursor_test.go b/internal/cursor/cursor_test.go new file mode 100644 index 000000000..0231e2d82 --- /dev/null +++ b/internal/cursor/cursor_test.go @@ -0,0 +1,197 @@ +package cursor_test + +import ( + "errors" + "testing" + + "github.com/hookdeck/outpost/internal/cursor" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestBase62Encode(t *testing.T) { + t.Run("encodes string to base62", func(t *testing.T) { + // Verified against Go's big.Int.Text(62) which uses alphabet: 0-9a-zA-Z + encoded := cursor.Base62Encode("the quick brown fox jumps over the lazy dog") + assert.Equal(t, "b6QPtm6Z5XFM81QySyltRRVYvv0ELEGBENK9XUgI4iciqMTErk0ea0kd2n", encoded) + }) + + t.Run("round-trips through encode/decode", func(t *testing.T) { + original := "the quick brown fox jumps over the lazy dog" + encoded := cursor.Base62Encode(original) + decoded, err := cursor.Base62Decode(encoded) + require.NoError(t, err) + assert.Equal(t, original, decoded) + }) + + t.Run("empty string returns empty", func(t *testing.T) { + encoded := cursor.Base62Encode("") + assert.Empty(t, encoded) + }) + + t.Run("same input produces same output", func(t *testing.T) { + a := cursor.Base62Encode("test") + b := cursor.Base62Encode("test") + assert.Equal(t, a, b) + }) + + t.Run("different inputs produce different outputs", func(t *testing.T) { + a := cursor.Base62Encode("test1") + b := cursor.Base62Encode("test2") + assert.NotEqual(t, a, b) + }) +} + +func TestBase62Decode(t *testing.T) { + t.Run("decodes base62 string", func(t *testing.T) { + // Verified against Go's big.Int.SetString(s, 62) which uses alphabet: 0-9a-zA-Z + decoded, err := cursor.Base62Decode("b6QPtm6Z5XFM81QySyltRRVYvv0ELEGBENK9XUgI4iciqMTErk0ea0kd2n") + require.NoError(t, err) + assert.Equal(t, "the quick brown fox jumps over the lazy dog", decoded) + }) + + t.Run("empty string returns empty", func(t *testing.T) { + decoded, err := cursor.Base62Decode("") + require.NoError(t, err) + assert.Empty(t, decoded) + }) + + t.Run("invalid base62 returns error", func(t *testing.T) { + _, err := cursor.Base62Decode("!!!invalid!!!") + require.Error(t, err) + assert.True(t, errors.Is(err, cursor.ErrInvalidCursor)) + }) +} + +func TestBase62Roundtrip(t *testing.T) { + testCases := []string{ + "simple", + "with spaces", + "with:colons", + "with_underscores", + "unicode-émoji-🎉", + "1234567890", + "mixed123with456numbers", + } + + for _, tc := range testCases { + t.Run(tc, func(t *testing.T) { + encoded := cursor.Base62Encode(tc) + decoded, err := cursor.Base62Decode(encoded) + require.NoError(t, err) + assert.Equal(t, tc, decoded) + }) + } +} + +func TestEncode(t *testing.T) { + t.Run("encodes with resource and version", func(t *testing.T) { + encoded := cursor.Encode("evt", 1, "position123") + assert.NotEmpty(t, encoded) + assert.NotContains(t, encoded, ":", "encoded cursor should not contain raw separators") + }) + + t.Run("different resources produce different encodings", func(t *testing.T) { + a := cursor.Encode("evt", 1, "data") + b := cursor.Encode("dlv", 1, "data") + assert.NotEqual(t, a, b) + }) + + t.Run("different versions produce different encodings", func(t *testing.T) { + a := cursor.Encode("evt", 1, "data") + b := cursor.Encode("evt", 2, "data") + assert.NotEqual(t, a, b) + }) + + t.Run("different data produces different encodings", func(t *testing.T) { + a := cursor.Encode("evt", 1, "data1") + b := cursor.Encode("evt", 1, "data2") + assert.NotEqual(t, a, b) + }) + + t.Run("version is zero-padded", func(t *testing.T) { + // Version 1 should be "01" internally + encoded1 := cursor.Encode("evt", 1, "data") + encoded01 := cursor.Encode("evt", 1, "data") + assert.Equal(t, encoded1, encoded01) + }) +} + +func TestDecode(t *testing.T) { + t.Run("empty string returns empty", func(t *testing.T) { + data, err := cursor.Decode("", "evt", 1) + require.NoError(t, err) + assert.Empty(t, data) + }) + + t.Run("decodes valid cursor", func(t *testing.T) { + encoded := cursor.Encode("evt", 1, "position123") + data, err := cursor.Decode(encoded, "evt", 1) + require.NoError(t, err) + assert.Equal(t, "position123", data) + }) + + t.Run("wrong resource returns ErrInvalidCursor", func(t *testing.T) { + encoded := cursor.Encode("evt", 1, "data") + _, err := cursor.Decode(encoded, "dlv", 1) + require.Error(t, err) + assert.True(t, errors.Is(err, cursor.ErrInvalidCursor)) + }) + + t.Run("wrong version returns ErrVersionMismatch", func(t *testing.T) { + encoded := cursor.Encode("evt", 1, "data") + _, err := cursor.Decode(encoded, "evt", 2) + require.Error(t, err) + assert.True(t, errors.Is(err, cursor.ErrVersionMismatch)) + }) + + t.Run("invalid base62 returns ErrInvalidCursor", func(t *testing.T) { + _, err := cursor.Decode("!!!invalid!!!", "evt", 1) + require.Error(t, err) + assert.True(t, errors.Is(err, cursor.ErrInvalidCursor)) + }) + + t.Run("completely malformed cursor returns ErrInvalidCursor", func(t *testing.T) { + // Encode something that doesn't follow the format at all + encoded := cursor.Base62Encode("garbage") + _, err := cursor.Decode(encoded, "evt", 1) + require.Error(t, err) + assert.True(t, errors.Is(err, cursor.ErrInvalidCursor)) + }) +} + +func TestRoundtrip(t *testing.T) { + testCases := []struct { + resource string + version int + data string + }{ + {"evt", 1, "simple"}, + {"dlv", 1, "1234567890_del_abc"}, + {"tnt", 2, "timestamp:12345"}, + {"evt", 99, "max_version"}, + {"x", 1, "short_resource"}, + {"longresource", 1, "long_resource_name"}, + {"evt", 1, "data:with:colons:in:it"}, + {"evt", 1, "unicode-émoji-🎉"}, + } + + for _, tc := range testCases { + name := tc.resource + "_v" + string(rune('0'+tc.version)) + "_" + tc.data + t.Run(name, func(t *testing.T) { + encoded := cursor.Encode(tc.resource, tc.version, tc.data) + decoded, err := cursor.Decode(encoded, tc.resource, tc.version) + require.NoError(t, err) + assert.Equal(t, tc.data, decoded) + }) + } +} + +func TestVersionMismatchMessage(t *testing.T) { + t.Run("error includes expected version", func(t *testing.T) { + encoded := cursor.Encode("evt", 1, "data") + _, err := cursor.Decode(encoded, "evt", 5) + require.Error(t, err) + assert.Contains(t, err.Error(), "05") + }) +} diff --git a/internal/logstore/chlogstore/chlogstore.go b/internal/logstore/chlogstore/chlogstore.go index 3a5c387bb..d2b366df3 100644 --- a/internal/logstore/chlogstore/chlogstore.go +++ b/internal/logstore/chlogstore/chlogstore.go @@ -3,17 +3,24 @@ package chlogstore import ( "context" "encoding/json" + "errors" "fmt" "strconv" "strings" "time" "github.com/hookdeck/outpost/internal/clickhouse" - "github.com/hookdeck/outpost/internal/logstore/cursor" + "github.com/hookdeck/outpost/internal/cursor" "github.com/hookdeck/outpost/internal/logstore/driver" "github.com/hookdeck/outpost/internal/models" ) +const ( + cursorResourceEvent = "evt" + cursorResourceDelivery = "dlv" + cursorVersion = 1 +) + type logStoreImpl struct { chDB clickhouse.DB eventsTable string @@ -45,12 +52,16 @@ func (s *logStoreImpl) ListEvent(ctx context.Context, req driver.ListEventReques limit = 100 } - nextCursor, prevCursor, err := cursor.DecodeAndValidate(req.Next, req.Prev, "event_time", sortOrder) + nextPosition, err := cursor.Decode(req.Next, cursorResourceEvent, cursorVersion) if err != nil { - return driver.ListEventResponse{}, err + return driver.ListEventResponse{}, convertCursorError(err) + } + prevPosition, err := cursor.Decode(req.Prev, cursorResourceEvent, cursorVersion) + if err != nil { + return driver.ListEventResponse{}, convertCursorError(err) } - goingBackward := !prevCursor.IsEmpty() + goingBackward := prevPosition != "" // Multi-column ORDER BY for deterministic pagination var orderByClause string @@ -95,12 +106,12 @@ func (s *logStoreImpl) ListEvent(ctx context.Context, req driver.ListEventReques args = append(args, *req.EventEnd) } - if !nextCursor.IsEmpty() { - cursorCond, cursorArgs := buildEventCursorCondition(sortOrder, nextCursor.Position, false) + if nextPosition != "" { + cursorCond, cursorArgs := buildEventCursorCondition(sortOrder, nextPosition, false) conditions = append(conditions, cursorCond) args = append(args, cursorArgs...) - } else if !prevCursor.IsEmpty() { - cursorCond, cursorArgs := buildEventCursorCondition(sortOrder, prevCursor.Position, true) + } else if prevPosition != "" { + cursorCond, cursorArgs := buildEventCursorCondition(sortOrder, prevPosition, true) conditions = append(conditions, cursorCond) args = append(args, cursorArgs...) } @@ -225,19 +236,15 @@ func (s *logStoreImpl) ListEvent(ctx context.Context, req driver.ListEventReques } encodeCursor := func(position string) string { - return cursor.Encode(cursor.Cursor{ - SortBy: "event_time", - SortOrder: sortOrder, - Position: position, - }) + return cursor.Encode(cursorResourceEvent, cursorVersion, position) } - if !prevCursor.IsEmpty() { + if prevPosition != "" { nextEncoded = encodeCursor(getPosition(results[len(results)-1])) if hasMore { prevEncoded = encodeCursor(getPosition(results[0])) } - } else if !nextCursor.IsEmpty() { + } else if nextPosition != "" { prevEncoded = encodeCursor(getPosition(results[0])) if hasMore { nextEncoded = encodeCursor(getPosition(results[len(results)-1])) @@ -291,7 +298,6 @@ func buildEventCursorCondition(sortOrder, position string, isBackward bool) (str } func (s *logStoreImpl) ListDeliveryEvent(ctx context.Context, req driver.ListDeliveryEventRequest) (driver.ListDeliveryEventResponse, error) { - sortBy := "delivery_time" sortOrder := req.SortOrder if sortOrder != "asc" && sortOrder != "desc" { sortOrder = "desc" @@ -302,12 +308,16 @@ func (s *logStoreImpl) ListDeliveryEvent(ctx context.Context, req driver.ListDel limit = 100 } - nextCursor, prevCursor, err := cursor.DecodeAndValidate(req.Next, req.Prev, sortBy, sortOrder) + nextPosition, err := cursor.Decode(req.Next, cursorResourceDelivery, cursorVersion) if err != nil { - return driver.ListDeliveryEventResponse{}, err + return driver.ListDeliveryEventResponse{}, convertCursorError(err) + } + prevPosition, err := cursor.Decode(req.Prev, cursorResourceDelivery, cursorVersion) + if err != nil { + return driver.ListDeliveryEventResponse{}, convertCursorError(err) } - goingBackward := !prevCursor.IsEmpty() + goingBackward := prevPosition != "" var orderByClause string if sortOrder == "desc" { @@ -361,12 +371,12 @@ func (s *logStoreImpl) ListDeliveryEvent(ctx context.Context, req driver.ListDel args = append(args, *req.End) } - if !nextCursor.IsEmpty() { - cursorCond, cursorArgs := buildCursorCondition(sortOrder, nextCursor.Position, false) + if nextPosition != "" { + cursorCond, cursorArgs := buildCursorCondition(sortOrder, nextPosition, false) conditions = append(conditions, cursorCond) args = append(args, cursorArgs...) - } else if !prevCursor.IsEmpty() { - cursorCond, cursorArgs := buildCursorCondition(sortOrder, prevCursor.Position, true) + } else if prevPosition != "" { + cursorCond, cursorArgs := buildCursorCondition(sortOrder, prevPosition, true) conditions = append(conditions, cursorCond) args = append(args, cursorArgs...) } @@ -532,19 +542,15 @@ func (s *logStoreImpl) ListDeliveryEvent(ctx context.Context, req driver.ListDel } encodeCursor := func(position string) string { - return cursor.Encode(cursor.Cursor{ - SortBy: sortBy, - SortOrder: sortOrder, - Position: position, - }) + return cursor.Encode(cursorResourceDelivery, cursorVersion, position) } - if !prevCursor.IsEmpty() { + if prevPosition != "" { nextEncoded = encodeCursor(getPosition(results[len(results)-1])) if hasMore { prevEncoded = encodeCursor(getPosition(results[0])) } - } else if !nextCursor.IsEmpty() { + } else if nextPosition != "" { prevEncoded = encodeCursor(getPosition(results[0])) if hasMore { nextEncoded = encodeCursor(getPosition(results[len(results)-1])) @@ -919,3 +925,11 @@ func buildCursorCondition(sortOrder, position string, isBackward bool) (string, return condition, []interface{}{deliveryTimeMs, deliveryTimeMs, deliveryID} } + +// convertCursorError converts cursor package errors to driver errors. +func convertCursorError(err error) error { + if errors.Is(err, cursor.ErrInvalidCursor) || errors.Is(err, cursor.ErrVersionMismatch) { + return driver.ErrInvalidCursor + } + return err +} diff --git a/internal/logstore/cursor/cursor.go b/internal/logstore/cursor/cursor.go deleted file mode 100644 index 66404f04a..000000000 --- a/internal/logstore/cursor/cursor.go +++ /dev/null @@ -1,175 +0,0 @@ -package cursor - -import ( - "fmt" - "math/big" - "strings" - - "github.com/hookdeck/outpost/internal/logstore/driver" -) - -// Cursor represents a pagination cursor with embedded sort parameters. -// This ensures cursors are only valid for queries with matching sort configuration. -// Implementations use this type directly - versioning is handled internally by Encode/Decode. -type Cursor struct { - SortBy string // "event_time" or "delivery_time" - SortOrder string // "asc" or "desc" - Position string // implementation-specific position value -} - -// IsEmpty returns true if this cursor has no position (i.e., no cursor was provided). -func (c Cursor) IsEmpty() bool { - return c.Position == "" -} - -// Encode converts a Cursor to a URL-safe base62 string. -// Always encodes using the current version format. -func Encode(c Cursor) string { - return encodeV1(c) -} - -// Decode converts a base62 encoded cursor string back to a Cursor. -// Automatically detects and handles different cursor versions. -// Returns driver.ErrInvalidCursor if the cursor is malformed. -func Decode(encoded string) (Cursor, error) { - if encoded == "" { - return Cursor{}, nil - } - - raw, err := decodeBase62(encoded) - if err != nil { - return Cursor{}, err - } - - if strings.HasPrefix(raw, v1Prefix) { - return decodeV1(raw) - } - - // Fall back to v0 format (legacy) - return decodeV0(raw) -} - -// Validate checks if the cursor matches the expected sort parameters. -// Returns driver.ErrInvalidCursor if there's a mismatch. -func Validate(c Cursor, expectedSortBy, expectedSortOrder string) error { - if c.IsEmpty() { - return nil - } - - if c.SortBy != expectedSortBy { - return fmt.Errorf("%w: cursor sortBy %q does not match request sortBy %q", - driver.ErrInvalidCursor, c.SortBy, expectedSortBy) - } - - if c.SortOrder != expectedSortOrder { - return fmt.Errorf("%w: cursor sortOrder %q does not match request sortOrder %q", - driver.ErrInvalidCursor, c.SortOrder, expectedSortOrder) - } - - return nil -} - -// DecodeAndValidate is a helper that decodes and validates both Next and Prev cursors. -// This is the common pattern used by all LogStore implementations. -func DecodeAndValidate(next, prev, sortBy, sortOrder string) (nextCursor, prevCursor Cursor, err error) { - if next != "" { - nextCursor, err = Decode(next) - if err != nil { - return Cursor{}, Cursor{}, err - } - if err := Validate(nextCursor, sortBy, sortOrder); err != nil { - return Cursor{}, Cursor{}, err - } - } - if prev != "" { - prevCursor, err = Decode(prev) - if err != nil { - return Cursor{}, Cursor{}, err - } - if err := Validate(prevCursor, sortBy, sortOrder); err != nil { - return Cursor{}, Cursor{}, err - } - } - return nextCursor, prevCursor, nil -} - -// ============================================================================= -// Internal: Base62 encoding/decoding -// ============================================================================= - -func encodeBase62(raw string) string { - num := new(big.Int) - num.SetBytes([]byte(raw)) - return num.Text(62) -} - -func decodeBase62(encoded string) (string, error) { - num := new(big.Int) - num, ok := num.SetString(encoded, 62) - if !ok { - return "", driver.ErrInvalidCursor - } - return string(num.Bytes()), nil -} - -// ============================================================================= -// Internal: v1 cursor format -// Format: v1:{sortBy}:{sortOrder}:{position} -// ============================================================================= - -const v1Prefix = "v1:" - -func encodeV1(c Cursor) string { - raw := fmt.Sprintf("v1:%s:%s:%s", c.SortBy, c.SortOrder, c.Position) - return encodeBase62(raw) -} - -func decodeV1(raw string) (Cursor, error) { - parts := strings.SplitN(raw, ":", 4) - if len(parts) != 4 { - return Cursor{}, driver.ErrInvalidCursor - } - - sortBy := parts[1] - sortOrder := parts[2] - position := parts[3] - - if sortBy != "event_time" && sortBy != "delivery_time" { - return Cursor{}, driver.ErrInvalidCursor - } - - if sortOrder != "asc" && sortOrder != "desc" { - return Cursor{}, driver.ErrInvalidCursor - } - - if position == "" { - return Cursor{}, driver.ErrInvalidCursor - } - - return Cursor{ - SortBy: sortBy, - SortOrder: sortOrder, - Position: position, - }, nil -} - -// ============================================================================= -// Internal: v0 cursor format (legacy, backward compatibility) -// Format: {position} (no version prefix, no sort params) -// Defaults: sortBy=event_time, sortOrder=desc -// ============================================================================= - -const ( - v0DefaultSortBy = "event_time" - v0DefaultSortOrder = "desc" -) - -func decodeV0(raw string) (Cursor, error) { - // v0 cursors are just the position, no validation needed - // If position is invalid, the DB query will simply not find it - return Cursor{ - SortBy: v0DefaultSortBy, - SortOrder: v0DefaultSortOrder, - Position: raw, - }, nil -} diff --git a/internal/logstore/cursor/cursor_test.go b/internal/logstore/cursor/cursor_test.go deleted file mode 100644 index dc2eab4bf..000000000 --- a/internal/logstore/cursor/cursor_test.go +++ /dev/null @@ -1,318 +0,0 @@ -package cursor - -import ( - "errors" - "math/big" - "testing" - - "github.com/hookdeck/outpost/internal/logstore/driver" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -func TestCursor_IsEmpty(t *testing.T) { - t.Run("empty cursor", func(t *testing.T) { - c := Cursor{} - assert.True(t, c.IsEmpty()) - }) - - t.Run("cursor with position", func(t *testing.T) { - c := Cursor{Position: "abc123"} - assert.False(t, c.IsEmpty()) - }) - - t.Run("cursor with only sort params", func(t *testing.T) { - c := Cursor{SortBy: "event_time", SortOrder: "desc"} - assert.True(t, c.IsEmpty(), "cursor without position is empty") - }) -} - -func TestEncode(t *testing.T) { - t.Run("encodes cursor to base62", func(t *testing.T) { - c := Cursor{ - SortBy: "delivery_time", - SortOrder: "desc", - Position: "1234567890_del_abc", - } - encoded := Encode(c) - assert.NotEmpty(t, encoded) - assert.NotContains(t, encoded, ":", "encoded cursor should not contain raw separators") - }) - - t.Run("different cursors produce different encodings", func(t *testing.T) { - c1 := Cursor{SortBy: "delivery_time", SortOrder: "desc", Position: "pos1"} - c2 := Cursor{SortBy: "delivery_time", SortOrder: "desc", Position: "pos2"} - assert.NotEqual(t, Encode(c1), Encode(c2)) - }) - - t.Run("same cursor produces same encoding", func(t *testing.T) { - c := Cursor{SortBy: "event_time", SortOrder: "asc", Position: "pos"} - assert.Equal(t, Encode(c), Encode(c)) - }) -} - -func TestDecode(t *testing.T) { - t.Run("empty string returns empty cursor", func(t *testing.T) { - c, err := Decode("") - require.NoError(t, err) - assert.True(t, c.IsEmpty()) - }) - - t.Run("decodes v1 cursor", func(t *testing.T) { - original := Cursor{ - SortBy: "delivery_time", - SortOrder: "desc", - Position: "1234567890_del_abc", - } - encoded := Encode(original) - - decoded, err := Decode(encoded) - require.NoError(t, err) - assert.Equal(t, original.SortBy, decoded.SortBy) - assert.Equal(t, original.SortOrder, decoded.SortOrder) - assert.Equal(t, original.Position, decoded.Position) - }) - - t.Run("decodes cursor with colons in position", func(t *testing.T) { - original := Cursor{ - SortBy: "event_time", - SortOrder: "asc", - Position: "time:with:colons:in:it", - } - encoded := Encode(original) - - decoded, err := Decode(encoded) - require.NoError(t, err) - assert.Equal(t, original.Position, decoded.Position) - }) - - t.Run("invalid base62 returns error", func(t *testing.T) { - _, err := Decode("!!!invalid!!!") - require.Error(t, err) - assert.True(t, errors.Is(err, driver.ErrInvalidCursor)) - }) - - t.Run("v1 invalid sortBy returns error", func(t *testing.T) { - raw := "v1:invalid_sort:desc:position" - encoded := encodeRaw(raw) - - _, err := Decode(encoded) - require.Error(t, err) - assert.True(t, errors.Is(err, driver.ErrInvalidCursor)) - }) - - t.Run("v1 invalid sortOrder returns error", func(t *testing.T) { - raw := "v1:event_time:invalid_order:position" - encoded := encodeRaw(raw) - - _, err := Decode(encoded) - require.Error(t, err) - assert.True(t, errors.Is(err, driver.ErrInvalidCursor)) - }) - - t.Run("v1 empty position returns error", func(t *testing.T) { - raw := "v1:event_time:desc:" - encoded := encodeRaw(raw) - - _, err := Decode(encoded) - require.Error(t, err) - assert.True(t, errors.Is(err, driver.ErrInvalidCursor)) - }) - - t.Run("v1 missing parts returns error", func(t *testing.T) { - raw := "v1:event_time:desc" // missing position - encoded := encodeRaw(raw) - - _, err := Decode(encoded) - require.Error(t, err) - assert.True(t, errors.Is(err, driver.ErrInvalidCursor)) - }) -} - -func TestDecodeV0BackwardCompatibility(t *testing.T) { - t.Run("decodes v0 cursor with defaults", func(t *testing.T) { - // v0 format: just position, no version prefix - position := "1704067200_evt_abc" - encoded := encodeRaw(position) - - decoded, err := Decode(encoded) - require.NoError(t, err) - assert.Equal(t, position, decoded.Position) - assert.Equal(t, "event_time", decoded.SortBy, "v0 defaults to event_time") - assert.Equal(t, "desc", decoded.SortOrder, "v0 defaults to desc") - }) - - t.Run("decodes v0 composite cursor", func(t *testing.T) { - // v0 composite cursor for event_time sort - position := "1704067200_evt_abc_1704067500_del_xyz" - encoded := encodeRaw(position) - - decoded, err := Decode(encoded) - require.NoError(t, err) - assert.Equal(t, position, decoded.Position) - assert.Equal(t, "event_time", decoded.SortBy) - assert.Equal(t, "desc", decoded.SortOrder) - }) - - t.Run("v0 cursor validates with matching defaults", func(t *testing.T) { - position := "1704067200_evt_abc" - encoded := encodeRaw(position) - - // Should work with default sort params - next, _, err := DecodeAndValidate(encoded, "", "event_time", "desc") - require.NoError(t, err) - assert.Equal(t, position, next.Position) - }) - - t.Run("v0 cursor fails validation with non-default sort params", func(t *testing.T) { - position := "1704067200_del_xyz" - encoded := encodeRaw(position) - - // Should fail because v0 defaults to event_time, not delivery_time - _, _, err := DecodeAndValidate(encoded, "", "delivery_time", "desc") - require.Error(t, err) - assert.True(t, errors.Is(err, driver.ErrInvalidCursor)) - assert.Contains(t, err.Error(), "sortBy") - }) - - t.Run("v0 cursor fails validation with different sort order", func(t *testing.T) { - position := "1704067200_evt_abc" - encoded := encodeRaw(position) - - // Should fail because v0 defaults to desc, not asc - _, _, err := DecodeAndValidate(encoded, "", "event_time", "asc") - require.Error(t, err) - assert.True(t, errors.Is(err, driver.ErrInvalidCursor)) - assert.Contains(t, err.Error(), "sortOrder") - }) - - t.Run("random string treated as v0 position", func(t *testing.T) { - // Any valid base62 that doesn't start with "v1:" is treated as v0 - position := "some_random_position_string" - encoded := encodeRaw(position) - - decoded, err := Decode(encoded) - require.NoError(t, err) - assert.Equal(t, position, decoded.Position) - assert.Equal(t, "event_time", decoded.SortBy) - assert.Equal(t, "desc", decoded.SortOrder) - }) -} - -func TestValidate(t *testing.T) { - t.Run("empty cursor is always valid", func(t *testing.T) { - c := Cursor{} - err := Validate(c, "event_time", "desc") - assert.NoError(t, err) - }) - - t.Run("matching params is valid", func(t *testing.T) { - c := Cursor{SortBy: "event_time", SortOrder: "desc", Position: "pos"} - err := Validate(c, "event_time", "desc") - assert.NoError(t, err) - }) - - t.Run("mismatched sortBy returns error", func(t *testing.T) { - c := Cursor{SortBy: "event_time", SortOrder: "desc", Position: "pos"} - err := Validate(c, "delivery_time", "desc") - require.Error(t, err) - assert.True(t, errors.Is(err, driver.ErrInvalidCursor)) - assert.Contains(t, err.Error(), "sortBy") - }) - - t.Run("mismatched sortOrder returns error", func(t *testing.T) { - c := Cursor{SortBy: "event_time", SortOrder: "desc", Position: "pos"} - err := Validate(c, "event_time", "asc") - require.Error(t, err) - assert.True(t, errors.Is(err, driver.ErrInvalidCursor)) - assert.Contains(t, err.Error(), "sortOrder") - }) -} - -func TestDecodeAndValidate(t *testing.T) { - t.Run("empty cursors return empty results", func(t *testing.T) { - next, prev, err := DecodeAndValidate("", "", "delivery_time", "desc") - require.NoError(t, err) - assert.True(t, next.IsEmpty()) - assert.True(t, prev.IsEmpty()) - }) - - t.Run("valid next cursor", func(t *testing.T) { - original := Cursor{SortBy: "delivery_time", SortOrder: "desc", Position: "pos"} - encoded := Encode(original) - - next, prev, err := DecodeAndValidate(encoded, "", "delivery_time", "desc") - require.NoError(t, err) - assert.Equal(t, "pos", next.Position) - assert.True(t, prev.IsEmpty()) - }) - - t.Run("valid prev cursor", func(t *testing.T) { - original := Cursor{SortBy: "event_time", SortOrder: "asc", Position: "pos"} - encoded := Encode(original) - - next, prev, err := DecodeAndValidate("", encoded, "event_time", "asc") - require.NoError(t, err) - assert.True(t, next.IsEmpty()) - assert.Equal(t, "pos", prev.Position) - }) - - t.Run("invalid next cursor returns error", func(t *testing.T) { - _, _, err := DecodeAndValidate("!!!invalid!!!", "", "delivery_time", "desc") - require.Error(t, err) - assert.True(t, errors.Is(err, driver.ErrInvalidCursor)) - }) - - t.Run("invalid prev cursor returns error", func(t *testing.T) { - _, _, err := DecodeAndValidate("", "!!!invalid!!!", "delivery_time", "desc") - require.Error(t, err) - assert.True(t, errors.Is(err, driver.ErrInvalidCursor)) - }) - - t.Run("mismatched next cursor returns error", func(t *testing.T) { - original := Cursor{SortBy: "delivery_time", SortOrder: "desc", Position: "pos"} - encoded := Encode(original) - - _, _, err := DecodeAndValidate(encoded, "", "event_time", "desc") - require.Error(t, err) - assert.True(t, errors.Is(err, driver.ErrInvalidCursor)) - }) - - t.Run("mismatched prev cursor returns error", func(t *testing.T) { - original := Cursor{SortBy: "delivery_time", SortOrder: "desc", Position: "pos"} - encoded := Encode(original) - - _, _, err := DecodeAndValidate("", encoded, "delivery_time", "asc") - require.Error(t, err) - assert.True(t, errors.Is(err, driver.ErrInvalidCursor)) - }) -} - -func TestRoundTrip(t *testing.T) { - testCases := []Cursor{ - {SortBy: "delivery_time", SortOrder: "desc", Position: "simple"}, - {SortBy: "delivery_time", SortOrder: "asc", Position: "1234567890_del_abc123"}, - {SortBy: "event_time", SortOrder: "desc", Position: "1234567890_evt_abc_1234567891_del_xyz"}, - {SortBy: "event_time", SortOrder: "asc", Position: "with:colons:and_underscores"}, - {SortBy: "delivery_time", SortOrder: "desc", Position: "unicode-émoji-🎉"}, - } - - for _, tc := range testCases { - t.Run(tc.Position, func(t *testing.T) { - encoded := Encode(tc) - decoded, err := Decode(encoded) - require.NoError(t, err) - - assert.Equal(t, tc.SortBy, decoded.SortBy) - assert.Equal(t, tc.SortOrder, decoded.SortOrder) - assert.Equal(t, tc.Position, decoded.Position) - }) - } -} - -// encodeRaw is a helper to encode raw strings for testing -func encodeRaw(raw string) string { - num := new(big.Int) - num.SetBytes([]byte(raw)) - return num.Text(62) -} diff --git a/internal/logstore/drivertest/drivertest.go b/internal/logstore/drivertest/drivertest.go index 09d8c274a..e2454a8ae 100644 --- a/internal/logstore/drivertest/drivertest.go +++ b/internal/logstore/drivertest/drivertest.go @@ -9,8 +9,8 @@ import ( "testing" "time" + "github.com/hookdeck/outpost/internal/cursor" "github.com/hookdeck/outpost/internal/idgen" - "github.com/hookdeck/outpost/internal/logstore/cursor" "github.com/hookdeck/outpost/internal/logstore/driver" "github.com/hookdeck/outpost/internal/models" "github.com/hookdeck/outpost/internal/util/testutil" @@ -2707,9 +2707,6 @@ func extractDeliveryIDs(des []*models.DeliveryEvent) []string { func testCursorValidation(t *testing.T, newHarness HarnessMaker) { t.Helper() - t.Run("cursor with mismatched sortOrder returns error", func(t *testing.T) { - testCursorMismatchedSortOrder(t, newHarness) - }) t.Run("malformed cursor returns error", func(t *testing.T) { testMalformedCursor(t, newHarness) }) @@ -2718,71 +2715,6 @@ func testCursorValidation(t *testing.T, newHarness HarnessMaker) { }) } -// testCursorMismatchedSortOrder verifies that using a cursor generated with one -// sortOrder value with a different sortOrder value returns ErrInvalidCursor. -func testCursorMismatchedSortOrder(t *testing.T, newHarness HarnessMaker) { - t.Helper() - - ctx := context.Background() - h, err := newHarness(ctx, t) - require.NoError(t, err) - t.Cleanup(h.Close) - - logStore, err := h.MakeDriver(ctx) - require.NoError(t, err) - - tenantID := idgen.String() - destinationID := idgen.Destination() - baseTime := time.Now().Truncate(time.Second) - startTime := baseTime.Add(-48 * time.Hour) - - // Insert enough data to get a next cursor - var deliveryEvents []*models.DeliveryEvent - for i := 0; i < 5; i++ { - event := testutil.EventFactory.AnyPointer( - testutil.EventFactory.WithID(fmt.Sprintf("evt_order_%d", i)), - testutil.EventFactory.WithTenantID(tenantID), - testutil.EventFactory.WithDestinationID(destinationID), - testutil.EventFactory.WithTime(baseTime.Add(-time.Duration(i)*time.Hour)), - ) - delivery := testutil.DeliveryFactory.AnyPointer( - testutil.DeliveryFactory.WithID(fmt.Sprintf("del_order_%d", i)), - testutil.DeliveryFactory.WithEventID(event.ID), - testutil.DeliveryFactory.WithDestinationID(destinationID), - testutil.DeliveryFactory.WithTime(baseTime.Add(-time.Duration(i)*time.Hour)), - ) - deliveryEvents = append(deliveryEvents, &models.DeliveryEvent{ - ID: fmt.Sprintf("de_order_%d", i), - DestinationID: destinationID, - Event: *event, - Delivery: delivery, - }) - } - require.NoError(t, logStore.InsertManyDeliveryEvent(ctx, deliveryEvents)) - - // Get a cursor with sortOrder=desc - response1, err := logStore.ListDeliveryEvent(ctx, driver.ListDeliveryEventRequest{ - TenantID: tenantID, - SortOrder: "desc", - Start: &startTime, - Limit: 2, - }) - require.NoError(t, err) - require.NotEmpty(t, response1.Next, "expected next cursor") - - // Try to use the cursor with sortOrder=asc - should fail - _, err = logStore.ListDeliveryEvent(ctx, driver.ListDeliveryEventRequest{ - TenantID: tenantID, - SortOrder: "asc", // Different from cursor - Start: &startTime, - Next: response1.Next, - Limit: 2, - }) - require.Error(t, err) - assert.True(t, errors.Is(err, driver.ErrInvalidCursor), - "expected ErrInvalidCursor, got: %v", err) -} - // testMalformedCursor verifies that a malformed cursor string returns ErrInvalidCursor. func testMalformedCursor(t *testing.T, newHarness HarnessMaker) { t.Helper() @@ -2804,7 +2736,7 @@ func testMalformedCursor(t *testing.T, newHarness HarnessMaker) { }{ {"completely invalid base62", "!!!invalid!!!"}, {"random string", "abcdef123456"}, - {"empty after decode", cursor.Encode(cursor.Cursor{})}, // Empty cursor should be fine, but let's test edge cases + {"wrong resource", cursor.Encode("evt", 1, "test")}, // evt is for events, not deliveries } for _, tc := range testCases { @@ -2816,13 +2748,10 @@ func testMalformedCursor(t *testing.T, newHarness HarnessMaker) { Next: tc.cursor, Limit: 10, }) - // Some of these might not error (e.g., if cursor decodes to valid format) - // but completely invalid base62 should error - if tc.name == "completely invalid base62" { - require.Error(t, err) - assert.True(t, errors.Is(err, driver.ErrInvalidCursor), - "expected ErrInvalidCursor for %s, got: %v", tc.name, err) - } + // All invalid cursor cases should error + require.Error(t, err) + assert.True(t, errors.Is(err, driver.ErrInvalidCursor), + "expected ErrInvalidCursor for %s, got: %v", tc.name, err) }) } } diff --git a/internal/logstore/memlogstore/memlogstore.go b/internal/logstore/memlogstore/memlogstore.go index d290df1a8..4c4edd83c 100644 --- a/internal/logstore/memlogstore/memlogstore.go +++ b/internal/logstore/memlogstore/memlogstore.go @@ -2,14 +2,21 @@ package memlogstore import ( "context" + "errors" "sort" "sync" - "github.com/hookdeck/outpost/internal/logstore/cursor" + "github.com/hookdeck/outpost/internal/cursor" "github.com/hookdeck/outpost/internal/logstore/driver" "github.com/hookdeck/outpost/internal/models" ) +const ( + cursorResourceEvent = "evt" + cursorResourceDelivery = "dlv" + cursorVersion = 1 +) + // memLogStore is an in-memory implementation of driver.LogStore. // It serves as a reference implementation and is useful for testing. type memLogStore struct { @@ -34,9 +41,13 @@ func (s *memLogStore) ListEvent(ctx context.Context, req driver.ListEventRequest sortOrder = "desc" } - nextCursor, prevCursor, err := cursor.DecodeAndValidate(req.Next, req.Prev, "event_time", sortOrder) + nextPosition, err := cursor.Decode(req.Next, cursorResourceEvent, cursorVersion) if err != nil { - return driver.ListEventResponse{}, err + return driver.ListEventResponse{}, convertCursorError(err) + } + prevPosition, err := cursor.Decode(req.Prev, cursorResourceEvent, cursorVersion) + if err != nil { + return driver.ListEventResponse{}, convertCursorError(err) } // Dedupe by event ID @@ -75,16 +86,16 @@ func (s *memLogStore) ListEvent(ctx context.Context, req driver.ListEventRequest } startIdx := 0 - if !nextCursor.IsEmpty() { + if nextPosition != "" { for i, event := range filtered { - if event.ID == nextCursor.Position { + if event.ID == nextPosition { startIdx = i break } } - } else if !prevCursor.IsEmpty() { + } else if prevPosition != "" { for i, event := range filtered { - if event.ID == prevCursor.Position { + if event.ID == prevPosition { startIdx = i - limit if startIdx < 0 { startIdx = 0 @@ -106,18 +117,10 @@ func (s *memLogStore) ListEvent(ctx context.Context, req driver.ListEventRequest var nextEncoded, prevEncoded string if endIdx < len(filtered) { - nextEncoded = cursor.Encode(cursor.Cursor{ - SortBy: "event_time", - SortOrder: sortOrder, - Position: filtered[endIdx].ID, - }) + nextEncoded = cursor.Encode(cursorResourceEvent, cursorVersion, filtered[endIdx].ID) } if startIdx > 0 { - prevEncoded = cursor.Encode(cursor.Cursor{ - SortBy: "event_time", - SortOrder: sortOrder, - Position: filtered[startIdx].ID, - }) + prevEncoded = cursor.Encode(cursorResourceEvent, cursorVersion, filtered[startIdx].ID) } return driver.ListEventResponse{ @@ -203,15 +206,18 @@ func (s *memLogStore) ListDeliveryEvent(ctx context.Context, req driver.ListDeli s.mu.RLock() defer s.mu.RUnlock() - sortBy := "delivery_time" sortOrder := req.SortOrder if sortOrder != "asc" && sortOrder != "desc" { sortOrder = "desc" } - nextCursor, prevCursor, err := cursor.DecodeAndValidate(req.Next, req.Prev, sortBy, sortOrder) + nextPosition, err := cursor.Decode(req.Next, cursorResourceDelivery, cursorVersion) if err != nil { - return driver.ListDeliveryEventResponse{}, err + return driver.ListDeliveryEventResponse{}, convertCursorError(err) + } + prevPosition, err := cursor.Decode(req.Prev, cursorResourceDelivery, cursorVersion) + if err != nil { + return driver.ListDeliveryEventResponse{}, convertCursorError(err) } var filtered []*models.DeliveryEvent @@ -242,16 +248,16 @@ func (s *memLogStore) ListDeliveryEvent(ctx context.Context, req driver.ListDeli } startIdx := 0 - if !nextCursor.IsEmpty() { + if nextPosition != "" { for i, de := range filtered { - if de.ID == nextCursor.Position { + if de.ID == nextPosition { startIdx = i break } } - } else if !prevCursor.IsEmpty() { + } else if prevPosition != "" { for i, de := range filtered { - if de.ID == prevCursor.Position { + if de.ID == prevPosition { startIdx = i - limit if startIdx < 0 { startIdx = 0 @@ -273,18 +279,10 @@ func (s *memLogStore) ListDeliveryEvent(ctx context.Context, req driver.ListDeli var nextEncoded, prevEncoded string if endIdx < len(filtered) { - nextEncoded = cursor.Encode(cursor.Cursor{ - SortBy: sortBy, - SortOrder: sortOrder, - Position: filtered[endIdx].ID, - }) + nextEncoded = cursor.Encode(cursorResourceDelivery, cursorVersion, filtered[endIdx].ID) } if startIdx > 0 { - prevEncoded = cursor.Encode(cursor.Cursor{ - SortBy: sortBy, - SortOrder: sortOrder, - Position: filtered[startIdx].ID, - }) + prevEncoded = cursor.Encode(cursorResourceDelivery, cursorVersion, filtered[startIdx].ID) } return driver.ListDeliveryEventResponse{ @@ -435,3 +433,11 @@ func copyDelivery(d *models.Delivery) *models.Delivery { return copied } + +// convertCursorError converts cursor package errors to driver errors. +func convertCursorError(err error) error { + if errors.Is(err, cursor.ErrInvalidCursor) || errors.Is(err, cursor.ErrVersionMismatch) { + return driver.ErrInvalidCursor + } + return err +} diff --git a/internal/logstore/pglogstore/pglogstore.go b/internal/logstore/pglogstore/pglogstore.go index 4d044adef..82d0e0b9e 100644 --- a/internal/logstore/pglogstore/pglogstore.go +++ b/internal/logstore/pglogstore/pglogstore.go @@ -2,16 +2,23 @@ package pglogstore import ( "context" + "errors" "fmt" "time" - "github.com/hookdeck/outpost/internal/logstore/cursor" + "github.com/hookdeck/outpost/internal/cursor" "github.com/hookdeck/outpost/internal/logstore/driver" "github.com/hookdeck/outpost/internal/models" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) +const ( + cursorResourceEvent = "evt" + cursorResourceDelivery = "dlv" + cursorVersion = 1 +) + type logStore struct { db *pgxpool.Pool } @@ -33,12 +40,16 @@ func (s *logStore) ListEvent(ctx context.Context, req driver.ListEventRequest) ( limit = 100 } - nextCursor, prevCursor, err := cursor.DecodeAndValidate(req.Next, req.Prev, "event_time", sortOrder) + nextPosition, err := cursor.Decode(req.Next, cursorResourceEvent, cursorVersion) if err != nil { - return driver.ListEventResponse{}, err + return driver.ListEventResponse{}, convertCursorError(err) + } + prevPosition, err := cursor.Decode(req.Prev, cursorResourceEvent, cursorVersion) + if err != nil { + return driver.ListEventResponse{}, convertCursorError(err) } - goingBackward := !prevCursor.IsEmpty() + goingBackward := prevPosition != "" var orderByClause, finalOrderByClause string if sortOrder == "desc" { @@ -102,14 +113,14 @@ func (s *logStore) ListEvent(ctx context.Context, req driver.ListEventRequest) ( `, cursorCondition, orderByClause, finalOrderByClause) rows, err := s.db.Query(ctx, query, - req.TenantID, // $1 - req.DestinationIDs, // $2 - req.Topics, // $3 - req.EventStart, // $4 - req.EventEnd, // $5 - nextCursor.Position, // $6 - prevCursor.Position, // $7 - limit+1, // $8 - fetch one extra to detect if there's more + req.TenantID, // $1 + req.DestinationIDs, // $2 + req.Topics, // $3 + req.EventStart, // $4 + req.EventEnd, // $5 + nextPosition, // $6 + prevPosition, // $7 + limit+1, // $8 - fetch one extra to detect if there's more ) if err != nil { return driver.ListEventResponse{}, fmt.Errorf("query failed: %w", err) @@ -190,19 +201,15 @@ func (s *logStore) ListEvent(ctx context.Context, req driver.ListEventRequest) ( } encodeCursor := func(position string) string { - return cursor.Encode(cursor.Cursor{ - SortBy: "event_time", - SortOrder: sortOrder, - Position: position, - }) + return cursor.Encode(cursorResourceEvent, cursorVersion, position) } - if !prevCursor.IsEmpty() { + if prevPosition != "" { nextEncoded = encodeCursor(getPosition(results[len(results)-1])) if hasMore { prevEncoded = encodeCursor(getPosition(results[0])) } - } else if !nextCursor.IsEmpty() { + } else if nextPosition != "" { prevEncoded = encodeCursor(getPosition(results[0])) if hasMore { nextEncoded = encodeCursor(getPosition(results[len(results)-1])) @@ -222,7 +229,6 @@ func (s *logStore) ListEvent(ctx context.Context, req driver.ListEventRequest) ( } func (s *logStore) ListDeliveryEvent(ctx context.Context, req driver.ListDeliveryEventRequest) (driver.ListDeliveryEventResponse, error) { - sortBy := "delivery_time" sortOrder := req.SortOrder if sortOrder != "asc" && sortOrder != "desc" { sortOrder = "desc" @@ -233,13 +239,17 @@ func (s *logStore) ListDeliveryEvent(ctx context.Context, req driver.ListDeliver limit = 100 } - nextCursor, prevCursor, err := cursor.DecodeAndValidate(req.Next, req.Prev, sortBy, sortOrder) + nextPosition, err := cursor.Decode(req.Next, cursorResourceDelivery, cursorVersion) if err != nil { - return driver.ListDeliveryEventResponse{}, err + return driver.ListDeliveryEventResponse{}, convertCursorError(err) + } + prevPosition, err := cursor.Decode(req.Prev, cursorResourceDelivery, cursorVersion) + if err != nil { + return driver.ListDeliveryEventResponse{}, convertCursorError(err) } cursorCol := "time_delivery_id" - goingBackward := !prevCursor.IsEmpty() + goingBackward := prevPosition != "" var orderByClause, finalOrderByClause string if sortOrder == "desc" { @@ -317,16 +327,16 @@ func (s *logStore) ListDeliveryEvent(ctx context.Context, req driver.ListDeliver `, cursorCondition, orderByClause, finalOrderByClause) rows, err := s.db.Query(ctx, query, - req.TenantID, // $1 - req.EventID, // $2 - req.DestinationIDs, // $3 - req.Status, // $4 - req.Topics, // $5 - req.Start, // $6 - req.End, // $7 - nextCursor.Position, // $8 - prevCursor.Position, // $9 - limit+1, // $10 - fetch one extra to detect if there's more + req.TenantID, // $1 + req.EventID, // $2 + req.DestinationIDs, // $3 + req.Status, // $4 + req.Topics, // $5 + req.Start, // $6 + req.End, // $7 + nextPosition, // $8 + prevPosition, // $9 + limit+1, // $10 - fetch one extra to detect if there's more ) if err != nil { return driver.ListDeliveryEventResponse{}, fmt.Errorf("query failed: %w", err) @@ -439,19 +449,15 @@ func (s *logStore) ListDeliveryEvent(ctx context.Context, req driver.ListDeliver } encodeCursor := func(position string) string { - return cursor.Encode(cursor.Cursor{ - SortBy: sortBy, - SortOrder: sortOrder, - Position: position, - }) + return cursor.Encode(cursorResourceDelivery, cursorVersion, position) } - if !prevCursor.IsEmpty() { + if prevPosition != "" { nextEncoded = encodeCursor(getPosition(results[len(results)-1])) if hasMore { prevEncoded = encodeCursor(getPosition(results[0])) } - } else if !nextCursor.IsEmpty() { + } else if nextPosition != "" { prevEncoded = encodeCursor(getPosition(results[0])) if hasMore { nextEncoded = encodeCursor(getPosition(results[len(results)-1])) @@ -813,3 +819,11 @@ func deliveryArrays(deliveries []*models.Delivery, deliveryEvents []*models.Deli attempts, } } + +// convertCursorError converts cursor package errors to driver errors. +func convertCursorError(err error) error { + if errors.Is(err, cursor.ErrInvalidCursor) || errors.Is(err, cursor.ErrVersionMismatch) { + return driver.ErrInvalidCursor + } + return err +} diff --git a/internal/models/entity.go b/internal/models/entity.go index eea01fafb..473e0b7b0 100644 --- a/internal/models/entity.go +++ b/internal/models/entity.go @@ -4,11 +4,12 @@ import ( "context" "errors" "fmt" - "math/big" "slices" "sort" + "strconv" "time" + "github.com/hookdeck/outpost/internal/cursor" "github.com/hookdeck/outpost/internal/redis" ) @@ -369,18 +370,24 @@ func (s *entityStoreImpl) ListTenant(ctx context.Context, req ListTenantRequest) isNextCursor := false isPrevCursor := false if req.Next != "" { - var err error - cursorTimestamp, err = decodeCursor(req.Next) + data, err := cursor.Decode(req.Next, "tnt", 1) if err != nil { return nil, fmt.Errorf("%w: %v", ErrInvalidCursor, err) } + cursorTimestamp, err = strconv.ParseInt(data, 10, 64) + if err != nil { + return nil, fmt.Errorf("%w: invalid timestamp", ErrInvalidCursor) + } isNextCursor = true } else if req.Prev != "" { - var err error - cursorTimestamp, err = decodeCursor(req.Prev) + data, err := cursor.Decode(req.Prev, "tnt", 1) if err != nil { return nil, fmt.Errorf("%w: %v", ErrInvalidCursor, err) } + cursorTimestamp, err = strconv.ParseInt(data, 10, 64) + if err != nil { + return nil, fmt.Errorf("%w: invalid timestamp", ErrInvalidCursor) + } isPrevCursor = true } @@ -491,12 +498,12 @@ func (s *entityStoreImpl) ListTenant(ctx context.Context, req ListTenantRequest) firstTenant := tenants[0] // Next cursor: points to last item (for continuing in same direction) - resp.Next = encodeCursor(lastTenant.CreatedAt.UnixMilli()) + resp.Next = cursor.Encode("tnt", 1, strconv.FormatInt(lastTenant.CreatedAt.UnixMilli(), 10)) // Prev cursor: points to first item (for going back) // Only set if we navigated here via a cursor (not on first page) if isNextCursor || isPrevCursor { - resp.Prev = encodeCursor(firstTenant.CreatedAt.UnixMilli()) + resp.Prev = cursor.Encode("tnt", 1, strconv.FormatInt(firstTenant.CreatedAt.UnixMilli(), 10)) } } @@ -634,66 +641,6 @@ func (s *entityStoreImpl) parseResp3SearchResult(resultMap map[interface{}]inter return tenants, totalCount, nil } -const cursorVersion = "tntv01" - -// encodeCursor encodes a timestamp as a versioned base62 cursor. -// Internal format: tntv01:, then base62 encoded. -func encodeCursor(timestamp int64) string { - raw := fmt.Sprintf("%s:%d", cursorVersion, timestamp) - return base62Encode(raw) -} - -// decodeCursor decodes a base62 cursor into a timestamp. -// Expects base62 encoded string containing: tntv01: -func decodeCursor(cursor string) (int64, error) { - raw, err := base62Decode(cursor) - if err != nil { - return 0, fmt.Errorf("invalid cursor encoding: %w", err) - } - - // Expected format: tntv02: - if len(raw) <= len(cursorVersion)+1 { - return 0, fmt.Errorf("invalid cursor format") - } - - version := raw[:len(cursorVersion)] - if version != cursorVersion { - return 0, fmt.Errorf("unsupported cursor version: %s", version) - } - - if raw[len(cursorVersion)] != ':' { - return 0, fmt.Errorf("invalid cursor format") - } - - var timestamp int64 - _, err = fmt.Sscanf(raw[len(cursorVersion)+1:], "%d", ×tamp) - if err != nil { - return 0, fmt.Errorf("invalid cursor timestamp") - } - - if timestamp < 0 { - return 0, fmt.Errorf("invalid timestamp") - } - return timestamp, nil -} - -// base62Encode encodes a string to base62. -func base62Encode(s string) string { - num := new(big.Int) - num.SetBytes([]byte(s)) - return num.Text(62) -} - -// base62Decode decodes a base62 string. -func base62Decode(s string) (string, error) { - num := new(big.Int) - num, ok := num.SetString(s, 62) - if !ok { - return "", fmt.Errorf("invalid base62 string") - } - return string(num.Bytes()), nil -} - func (s *entityStoreImpl) listDestinationSummaryByTenant(ctx context.Context, tenantID string, opts ListDestinationByTenantOpts) ([]DestinationSummary, error) { return s.parseListDestinationSummaryByTenantCmd(s.redisClient.HGetAll(ctx, s.redisTenantDestinationSummaryKey(tenantID)), opts) }