From 7b688220a4e879f9a893c682b2d14dd78076939b Mon Sep 17 00:00:00 2001 From: Alex Luong Date: Thu, 5 Feb 2026 16:31:40 +0700 Subject: [PATCH] refactor(pglogstore): rewrite for denormalized schema - Remove event_attempt_index table, query directly from denormalized attempts - Use dynamic cursor format {milliseconds}::{id} matching chlogstore - Simplify queries: no JOINs needed for attempt lookups - Update InsertMany to populate denormalized event columns in attempts - Update migration 000005 for new schema structure Co-Authored-By: Claude Opus 4.5 --- internal/logstore/pglogstore/README.md | 361 +++++++++- internal/logstore/pglogstore/pglogstore.go | 656 +++++++++++------- .../000005_denormalize_attempts.down.sql | 120 ++++ .../000005_denormalize_attempts.up.sql | 118 ++++ ...000005_rename_delivery_to_attempt.down.sql | 48 -- .../000005_rename_delivery_to_attempt.up.sql | 47 -- 6 files changed, 967 insertions(+), 383 deletions(-) create mode 100644 internal/migrator/migrations/postgres/000005_denormalize_attempts.down.sql create mode 100644 internal/migrator/migrations/postgres/000005_denormalize_attempts.up.sql delete mode 100644 internal/migrator/migrations/postgres/000005_rename_delivery_to_attempt.down.sql delete mode 100644 internal/migrator/migrations/postgres/000005_rename_delivery_to_attempt.up.sql diff --git a/internal/logstore/pglogstore/README.md b/internal/logstore/pglogstore/README.md index 58c05e81f..0f769e814 100644 --- a/internal/logstore/pglogstore/README.md +++ b/internal/logstore/pglogstore/README.md @@ -4,69 +4,370 @@ PostgreSQL implementation of the LogStore interface. ## Schema +Denormalized design optimized for read-heavy workloads. Events are stored once; attempts embed event data for JOIN-free queries. + | Table | Purpose | Primary Key | |-------|---------|-------------| -| `events` | Event data (data, metadata) | `(time, id)` | -| `deliveries` | Delivery attempts (status, response) | `(time, id)` | -| `event_delivery_index` | Query index with cursor columns | `(delivery_time, event_id, delivery_id)` | +| `events` | Event payload (for ListEvent, RetrieveEvent) | `(time, id)` | +| `attempts` | Delivery attempts with embedded event data | `(time, id)` | + +Both tables are partitioned by time for efficient data management and pruning. + +### events + +```sql +CREATE TABLE events ( + id text NOT NULL, + tenant_id text NOT NULL, + destination_id text NOT NULL, -- publish input (NOT matched destinations) + time timestamptz NOT NULL, + topic text NOT NULL, + eligible_for_retry boolean NOT NULL, + data jsonb NOT NULL, + metadata jsonb NOT NULL, + PRIMARY KEY (time, id) -- partition key must be in PK +) PARTITION BY RANGE (time); + +-- Primary list query: paginate by (time, id) +CREATE INDEX idx_events_tenant_time ON events (tenant_id, time DESC, id DESC); -All tables are partitioned by time. +-- Filter by topic +CREATE INDEX idx_events_tenant_topic_time ON events (tenant_id, topic, time DESC, id DESC); + +-- Point lookup by event_id (scans partition indexes, not O(1)) +CREATE INDEX idx_events_id ON events (id); +``` + +> **Note:** Events are destination-agnostic. The `destination_id` field represents the publish input +> (explicit destination targeting), not the destinations that matched via routing rules. To filter +> by destination, use `ListAttempt` which queries actual delivery attempts. + +### attempts + +```sql +CREATE TABLE attempts ( + id text NOT NULL, + event_id text NOT NULL, + tenant_id text NOT NULL, + destination_id text NOT NULL, + topic text NOT NULL, + status text NOT NULL, + time timestamptz NOT NULL, + attempt_number integer NOT NULL, + manual boolean NOT NULL DEFAULT false, + code text, + response_data jsonb, + -- Embedded event data (denormalized) + event_time timestamptz NOT NULL, + eligible_for_retry boolean NOT NULL, + event_data jsonb NOT NULL, + event_metadata jsonb NOT NULL, + PRIMARY KEY (time, id) -- partition key must be in PK +) PARTITION BY RANGE (time); + +-- Primary list query: paginate by (time, id) +CREATE INDEX idx_attempts_tenant_time ON attempts (tenant_id, time DESC, id DESC); + +-- Filter by destination +CREATE INDEX idx_attempts_tenant_dest_time ON attempts (tenant_id, destination_id, time DESC, id DESC); + +-- Filter by status +CREATE INDEX idx_attempts_tenant_status_time ON attempts (tenant_id, status, time DESC, id DESC); + +-- Filter by topic +CREATE INDEX idx_attempts_tenant_topic_time ON attempts (tenant_id, topic, time DESC, id DESC); + +-- Filter by event_id (for "attempts for this event" queries) +CREATE INDEX idx_attempts_event_time ON attempts (event_id, time DESC, id DESC); + +-- Point lookup by attempt_id (scans partition indexes, not O(1)) +CREATE INDEX idx_attempts_id ON attempts (id); +``` + +## Cursor Format + +Cursors encode pagination position as `{time_ms}::{id}` (matching chlogstore): + +```go +// Encode: time milliseconds + "::" + id +position := fmt.Sprintf("%d::%s", event.Time.UnixMilli(), event.ID) +cursor := cursor.Encode("evt", 1, position) // base62 encoded + +// Decode: parse back to time and id +position, _ := cursor.Decode(encoded, "evt", 1) +parts := strings.SplitN(position, "::", 2) +timeMs, _ := strconv.ParseInt(parts[0], 10, 64) +id := parts[1] +``` + +Resources: +- Events: `evt` prefix +- Attempts: `att` prefix + +--- ## Operations -### ListDelivery +### ListEvent -Query pattern: **Index → Hydrate** +Paginated list of events with optional filters. -1. CTE filters and paginates on `event_delivery_index` -2. JOIN `events` and `deliveries` by primary key to populate full data +**Filters:** tenant_id (optional), topic[], time range +**Note:** destination_id[] filter returns unimplemented error (events are destination-agnostic; use ListAttempt instead) +**Pagination:** bidirectional cursor on `(time, id)` ```sql -WITH filtered AS ( - SELECT ... FROM event_delivery_index WHERE [filters] ORDER BY ... LIMIT N -) -SELECT ... FROM filtered f -JOIN events e ON (e.time, e.id) = (f.event_time, f.event_id) -JOIN deliveries d ON (d.time, d.id) = (f.delivery_time, f.delivery_id) +SELECT id, tenant_id, destination_id, time, topic, eligible_for_retry, data, metadata +FROM events +WHERE ($1 = '' OR tenant_id = $1) -- optional tenant filter + AND ($2::text[] IS NULL OR cardinality($2) = 0 OR topic = ANY($2)) -- optional topic filter + AND ($3::timestamptz IS NULL OR time >= $3) -- GTE + AND ($4::timestamptz IS NULL OR time <= $4) -- LTE + AND ($5::timestamptz IS NULL OR time > $5) -- GT + AND ($6::timestamptz IS NULL OR time < $6) -- LT + -- Cursor condition (expanded OR form for clarity): + AND ( + time < $cursor_time + OR (time = $cursor_time AND id < $cursor_id) + ) +ORDER BY time DESC, id DESC +LIMIT $limit + 1; +``` + +**Bidirectional Pagination:** +- **Next cursor**: `time < $t OR (time = $t AND id < $id)` with `ORDER BY time DESC, id DESC` +- **Prev cursor**: `time > $t OR (time = $t AND id > $id)` with `ORDER BY time ASC, id ASC`, then reverse results + +| Scenario | Index Used | Performance | +|----------|-----------|-------------| +| No filters | `idx_events_tenant_time` | O(limit) - index scan | +| topic filter | `idx_events_tenant_topic_time` | O(limit) - index scan | +| topic[] (multiple) | `idx_events_tenant_time` + filter | O(N) - scans then filters | + +#### Performance Notes + +- **Single topic filter**: Index seek, O(limit) +- **Array filters** (`topic = ANY($2)`): Falls back to tenant+time index with in-memory filtering + +--- + +### ListAttempt + +Paginated list of attempts with embedded event data. No JOINs required. + +**Filters:** tenant_id (optional), event_id, destination_id[], status, topic[], time range +**Pagination:** bidirectional cursor on `(time, id)` + +```sql +SELECT + id, event_id, tenant_id, destination_id, topic, status, time, + attempt_number, manual, code, response_data, + event_time, eligible_for_retry, event_data, event_metadata +FROM attempts +WHERE ($1 = '' OR tenant_id = $1) -- optional tenant filter + AND ($2::text = '' OR event_id = $2) + AND ($3::text[] IS NULL OR cardinality($3) = 0 OR destination_id = ANY($3)) -- optional destination filter + AND ($4::text = '' OR status = $4) + AND ($5::text[] IS NULL OR cardinality($5) = 0 OR topic = ANY($5)) -- optional topic filter + AND ($6::timestamptz IS NULL OR time >= $6) -- GTE + AND ($7::timestamptz IS NULL OR time <= $7) -- LTE + AND ($8::timestamptz IS NULL OR time > $8) -- GT + AND ($9::timestamptz IS NULL OR time < $9) -- LT + -- Cursor condition (expanded OR form for clarity): + AND ( + time < $cursor_time + OR (time = $cursor_time AND id < $cursor_id) + ) +ORDER BY time DESC, id DESC +LIMIT $limit + 1; ``` -**Key considerations:** -- Cursor encodes sort params - rejects mismatched sort order +**Bidirectional Pagination:** +- **Next cursor**: `time < $t OR (time = $t AND id < $id)` with `ORDER BY time DESC, id DESC` +- **Prev cursor**: `time > $t OR (time = $t AND id > $id)` with `ORDER BY time ASC, id ASC`, then reverse results + +| Scenario | Index Used | Performance | +|----------|-----------|-------------| +| No filters | `idx_attempts_tenant_time` | O(limit) - index scan | +| event_id filter | `idx_attempts_event_time` | O(limit) - index scan | +| destination_id filter | `idx_attempts_tenant_dest_time` | O(limit) - index scan | +| status filter | `idx_attempts_tenant_status_time` | O(limit) - index scan | +| topic filter | `idx_attempts_tenant_topic_time` | O(limit) - index scan | +| **Multiple filters combined** | **Bitmap AND or sequential filter** | **O(N) worst case** | + +#### Performance Notes + +- **Single filter queries** are O(limit) with appropriate index +- **event_id queries** ("all attempts for event X") are fast via dedicated index +- **status filter** is highly selective (only 2 values: success/failed) - index is effective + +**Slow Query Warning:** +```sql +-- Combining destination[] + status + topic[] +-- No single index covers this; PostgreSQL may: +-- 1. Bitmap AND multiple indexes (moderate) +-- 2. Scan tenant_time index and filter (slow) +WHERE destination_id = ANY($2) AND status = $3 AND topic = ANY($4) +``` + +--- ### RetrieveEvent -Direct lookup by `(tenant_id, event_id)`. +Point lookup by event ID. ```sql +-- Simple lookup (tenant_id optional) +-- Note: Uses idx_events_id, scans across partitions SELECT id, tenant_id, destination_id, time, topic, eligible_for_retry, data, metadata FROM events -WHERE tenant_id = $1 AND id = $2 +WHERE ($1 = '' OR tenant_id = $1) AND id = $2; --- With destination filter: +-- With destination filter (verify event was sent to this destination) SELECT id, tenant_id, $3 as destination_id, time, topic, eligible_for_retry, data, metadata FROM events -WHERE tenant_id = $1 AND id = $2 -AND EXISTS (SELECT 1 FROM event_delivery_index WHERE event_id = $2 AND destination_id = $3) +WHERE ($1 = '' OR tenant_id = $1) AND id = $2 + AND EXISTS (SELECT 1 FROM attempts WHERE event_id = $2 AND destination_id = $3 LIMIT 1); ``` +| Scenario | Index Used | Performance | +|----------|-----------|-------------| +| By ID | idx_events_id | O(partitions) - scans each partition's index | +| With destination check | idx_events_id + idx_attempts_event_time | O(partitions) | + +**Performance Note:** With time-based partitioning, lookups by ID alone must scan the index in each partition since ID is not the partition key. This is acceptable for point lookups but not ideal. If performance becomes an issue, consider requiring time hints or using a separate ID→time mapping. + +--- + +### RetrieveAttempt + +Point lookup by attempt ID. + +```sql +-- tenant_id optional +-- Note: Uses idx_attempts_id, scans across partitions +SELECT + id, event_id, tenant_id, destination_id, topic, status, time, + attempt_number, manual, code, response_data, + event_time, eligible_for_retry, event_data, event_metadata +FROM attempts +WHERE ($1 = '' OR tenant_id = $1) AND id = $2; +``` + +| Scenario | Index Used | Performance | +|----------|-----------|-------------| +| By ID | idx_attempts_id | O(partitions) - scans each partition's index | + +**Performance Note:** Same as RetrieveEvent - lookups by ID scan partition indexes. + +--- + ### InsertMany -Batch insert using `unnest()` arrays in a single transaction across all 3 tables. +Batch insert events and attempts in a single transaction. ```sql BEGIN; +-- Insert events (deduplicated by caller) INSERT INTO events (id, tenant_id, destination_id, time, topic, eligible_for_retry, data, metadata) SELECT * FROM unnest($1::text[], $2::text[], $3::text[], $4::timestamptz[], $5::text[], $6::boolean[], $7::jsonb[], $8::jsonb[]) -ON CONFLICT (time, id) DO NOTHING; - -INSERT INTO deliveries (id, event_id, destination_id, status, time, code, response_data) -SELECT * FROM unnest($1::text[], $2::text[], $3::text[], $4::text[], $5::timestamptz[], $6::text[], $7::jsonb[]) -ON CONFLICT (time, id) DO UPDATE SET status = EXCLUDED.status, code = EXCLUDED.code, response_data = EXCLUDED.response_data; +ON CONFLICT (time, id) DO NOTHING; -- PK is (time, id) -INSERT INTO event_delivery_index (event_id, delivery_id, tenant_id, destination_id, event_time, delivery_time, topic, status) -SELECT * FROM unnest($1::text[], $2::text[], $3::text[], $4::text[], $5::timestamptz[], $6::timestamptz[], $7::text[], $8::text[]) -ON CONFLICT (delivery_time, event_id, delivery_id) DO UPDATE SET status = EXCLUDED.status; +-- Insert attempts with embedded event data +INSERT INTO attempts ( + id, event_id, tenant_id, destination_id, topic, status, time, + attempt_number, manual, code, response_data, + event_time, eligible_for_retry, event_data, event_metadata +) +SELECT * FROM unnest( + $1::text[], $2::text[], $3::text[], $4::text[], $5::text[], $6::text[], $7::timestamptz[], + $8::integer[], $9::boolean[], $10::text[], $11::jsonb[], + $12::timestamptz[], $13::boolean[], $14::jsonb[], $15::jsonb[] +) +ON CONFLICT (time, id) DO UPDATE SET -- PK is (time, id) + status = EXCLUDED.status, + code = EXCLUDED.code, + response_data = EXCLUDED.response_data; COMMIT; ``` + +| Scenario | Performance | +|----------|-------------| +| Batch insert N events + M attempts | O(N + M) - linear with batch size | + +**Performance:** Efficient. `unnest()` avoids N round-trips. Single transaction ensures atomicity. + +--- + +## Performance Summary + +### Fast Queries (O(limit)) + +| Operation | Condition | +|-----------|-----------| +| ListEvent | Single filter or no filter | +| ListAttempt | Single filter or no filter | +| ListAttempt | event_id filter (dedicated index) | +| InsertMany | Always (O(batch size)) | + +### Moderate Queries (O(partitions)) + +| Operation | Condition | Notes | +|-----------|-----------|-------| +| RetrieveEvent | By ID | Scans idx_events_id across partitions | +| RetrieveAttempt | By ID | Scans idx_attempts_id across partitions | + +### Potentially Slow Queries (O(N)) + +| Operation | Condition | Mitigation | +|-----------|-----------|------------| +| ListEvent | `topic = ANY(...)` with multiple values | Accept; rare use case | +| ListAttempt | Combined destination[] + status + topic[] | Bitmap index intersection helps | + +### Index Overhead + +Total indexes per table: +- **events:** 4 indexes (PK + 2 composite + 1 id lookup) +- **attempts:** 7 indexes (PK + 5 composite + 1 id lookup) + +Write amplification is acceptable for read-heavy workload. Each INSERT touches all indexes, but batch inserts amortize overhead. + +--- + +## Storage Considerations + +Denormalization trades storage for query performance: + +| Field | Duplicated In | +|-------|--------------| +| event_time | Every attempt | +| eligible_for_retry | Every attempt | +| event_data (jsonb) | Every attempt | +| event_metadata (jsonb) | Every attempt | + +**Estimate:** If avg event payload is 2KB and avg attempts/event is 1.5, storage overhead is ~50% vs normalized schema. + +**Justification:** Events are immutable; no update anomalies. Storage is cheap; query latency is not. + +--- + +## Partitioning + +Both tables partition by time (monthly recommended): + +```sql +CREATE TABLE events_2024_01 PARTITION OF events + FOR VALUES FROM ('2024-01-01') TO ('2024-02-01'); + +CREATE TABLE attempts_2024_01 PARTITION OF attempts + FOR VALUES FROM ('2024-01-01') TO ('2024-02-01'); +``` + +**Benefits:** +- Partition pruning for time-filtered queries +- Easy data retention (drop old partitions) +- Parallel query across partitions + +**Cursor consideration:** Cursors encode `(time, id)`. Time-based partitioning aligns with cursor pagination, enabling partition pruning during pagination. diff --git a/internal/logstore/pglogstore/pglogstore.go b/internal/logstore/pglogstore/pglogstore.go index 6e7dd24fd..4faca521c 100644 --- a/internal/logstore/pglogstore/pglogstore.go +++ b/internal/logstore/pglogstore/pglogstore.go @@ -3,6 +3,7 @@ package pglogstore import ( "context" "fmt" + "strconv" "strings" "time" @@ -30,10 +31,21 @@ func NewLogStore(db *pgxpool.Pool) driver.LogStore { } } -// eventWithTimeID wraps an event with its time_id for cursor encoding. -type eventWithTimeID struct { +// eventWithPosition wraps an event with its cursor position data. +type eventWithPosition struct { *models.Event - TimeID string + eventTime time.Time +} + +// attemptRecordWithPosition wraps an attempt record with its cursor position data. +type attemptRecordWithPosition struct { + *driver.AttemptRecord + attemptTime time.Time +} + +// parseTimestampMs parses a millisecond timestamp string. +func parseTimestampMs(s string) (int64, error) { + return strconv.ParseInt(s, 10, 64) } func (s *logStore) ListEvent(ctx context.Context, req driver.ListEventRequest) (driver.ListEventResponse, error) { @@ -65,12 +77,12 @@ func (s *logStore) ListEvent(ctx context.Context, req driver.ListEventRequest) ( limit = 100 } - res, err := pagination.Run(ctx, pagination.Config[eventWithTimeID]{ + res, err := pagination.Run(ctx, pagination.Config[eventWithPosition]{ Limit: limit, Order: sortOrder, Next: req.Next, Prev: req.Prev, - Fetch: func(ctx context.Context, q pagination.QueryInput) ([]eventWithTimeID, error) { + Fetch: func(ctx context.Context, q pagination.QueryInput) ([]eventWithPosition, error) { query, args := buildEventQuery(req, q) rows, err := s.db.Query(ctx, query, args...) if err != nil { @@ -79,9 +91,10 @@ func (s *logStore) ListEvent(ctx context.Context, req driver.ListEventRequest) ( defer rows.Close() return scanEvents(rows) }, - Cursor: pagination.Cursor[eventWithTimeID]{ - Encode: func(e eventWithTimeID) string { - return cursor.Encode(cursorResourceEvent, cursorVersion, e.TimeID) + Cursor: pagination.Cursor[eventWithPosition]{ + Encode: func(e eventWithPosition) string { + position := fmt.Sprintf("%d::%s", e.eventTime.UnixMilli(), e.Event.ID) + return cursor.Encode(cursorResourceEvent, cursorVersion, position) }, Decode: func(c string) (string, error) { return cursor.Decode(c, cursorResourceEvent, cursorVersion) @@ -106,8 +119,63 @@ func (s *logStore) ListEvent(ctx context.Context, req driver.ListEventRequest) ( } func buildEventQuery(req driver.ListEventRequest, q pagination.QueryInput) (string, []any) { - cursorCondition := fmt.Sprintf("AND ($8::text = '' OR time_id %s $8::text)", q.Compare) - orderByClause := fmt.Sprintf("time %s, id %s", strings.ToUpper(q.SortDir), strings.ToUpper(q.SortDir)) + var conditions []string + var args []any + argNum := 1 + + if req.TenantID != "" { + conditions = append(conditions, fmt.Sprintf("tenant_id = $%d", argNum)) + args = append(args, req.TenantID) + argNum++ + } + + if len(req.DestinationIDs) > 0 { + conditions = append(conditions, fmt.Sprintf("destination_id = ANY($%d)", argNum)) + args = append(args, req.DestinationIDs) + argNum++ + } + + if len(req.Topics) > 0 { + conditions = append(conditions, fmt.Sprintf("topic = ANY($%d)", argNum)) + args = append(args, req.Topics) + argNum++ + } + + if req.TimeFilter.GTE != nil { + conditions = append(conditions, fmt.Sprintf("time >= $%d", argNum)) + args = append(args, *req.TimeFilter.GTE) + argNum++ + } + if req.TimeFilter.LTE != nil { + conditions = append(conditions, fmt.Sprintf("time <= $%d", argNum)) + args = append(args, *req.TimeFilter.LTE) + argNum++ + } + if req.TimeFilter.GT != nil { + conditions = append(conditions, fmt.Sprintf("time > $%d", argNum)) + args = append(args, *req.TimeFilter.GT) + argNum++ + } + if req.TimeFilter.LT != nil { + conditions = append(conditions, fmt.Sprintf("time < $%d", argNum)) + args = append(args, *req.TimeFilter.LT) + argNum++ + } + + if q.CursorPos != "" { + cursorCond, cursorArgs := buildEventCursorCondition(q.Compare, q.CursorPos, argNum) + conditions = append(conditions, cursorCond) + args = append(args, cursorArgs...) + argNum += len(cursorArgs) + } + + whereClause := strings.Join(conditions, " AND ") + if whereClause == "" { + whereClause = "1=1" + } + + orderByClause := fmt.Sprintf("ORDER BY time %s, id %s", + strings.ToUpper(q.SortDir), strings.ToUpper(q.SortDir)) query := fmt.Sprintf(` SELECT @@ -118,38 +186,40 @@ func buildEventQuery(req driver.ListEventRequest, q pagination.QueryInput) (stri topic, eligible_for_retry, data, - metadata, - time_id + metadata FROM events - WHERE ($1::text = '' OR tenant_id = $1) - AND (array_length($2::text[], 1) IS NULL OR destination_id = ANY($2)) - AND (array_length($3::text[], 1) IS NULL OR topic = ANY($3)) - AND ($4::timestamptz IS NULL OR time >= $4) - AND ($5::timestamptz IS NULL OR time <= $5) - AND ($6::timestamptz IS NULL OR time > $6) - AND ($7::timestamptz IS NULL OR time < $7) + WHERE %s %s - ORDER BY %s - LIMIT $9 - `, cursorCondition, orderByClause) - - args := []any{ - req.TenantID, // $1 - req.DestinationIDs, // $2 - req.Topics, // $3 - req.TimeFilter.GTE, // $4 - req.TimeFilter.LTE, // $5 - req.TimeFilter.GT, // $6 - req.TimeFilter.LT, // $7 - q.CursorPos, // $8 - q.Limit, // $9 - } + LIMIT $%d + `, whereClause, orderByClause, argNum) + + args = append(args, q.Limit) return query, args } -func scanEvents(rows pgx.Rows) ([]eventWithTimeID, error) { - var results []eventWithTimeID +func buildEventCursorCondition(compare, position string, argOffset int) (string, []any) { + parts := strings.SplitN(position, "::", 2) + if len(parts) != 2 { + return "1=1", nil // invalid cursor, return always true + } + eventTimeMs, err := parseTimestampMs(parts[0]) + if err != nil { + return "1=1", nil // invalid timestamp, return always true + } + eventID := parts[1] + + // Convert milliseconds to PostgreSQL timestamp + condition := fmt.Sprintf(`( + time %s to_timestamp($%d / 1000.0) + OR (time = to_timestamp($%d / 1000.0) AND id %s $%d) + )`, compare, argOffset, argOffset+1, compare, argOffset+2) + + return condition, []any{eventTimeMs, eventTimeMs, eventID} +} + +func scanEvents(rows pgx.Rows) ([]eventWithPosition, error) { + var results []eventWithPosition for rows.Next() { var ( id string @@ -160,7 +230,6 @@ func scanEvents(rows pgx.Rows) ([]eventWithTimeID, error) { eligibleForRetry bool data map[string]any metadata map[string]string - timeID string ) if err := rows.Scan( @@ -172,12 +241,11 @@ func scanEvents(rows pgx.Rows) ([]eventWithTimeID, error) { &eligibleForRetry, &data, &metadata, - &timeID, ); err != nil { return nil, fmt.Errorf("scan failed: %w", err) } - results = append(results, eventWithTimeID{ + results = append(results, eventWithPosition{ Event: &models.Event{ ID: id, TenantID: tenantID, @@ -188,7 +256,7 @@ func scanEvents(rows pgx.Rows) ([]eventWithTimeID, error) { Data: data, Metadata: metadata, }, - TimeID: timeID, + eventTime: eventTime, }) } @@ -199,12 +267,6 @@ func scanEvents(rows pgx.Rows) ([]eventWithTimeID, error) { return results, nil } -// attemptRecordWithTimeID wraps an attempt record with its time_attempt_id for cursor encoding. -type attemptRecordWithTimeID struct { - *driver.AttemptRecord - TimeAttemptID string -} - func (s *logStore) ListAttempt(ctx context.Context, req driver.ListAttemptRequest) (driver.ListAttemptResponse, error) { sortOrder := req.SortOrder if sortOrder != "asc" && sortOrder != "desc" { @@ -216,12 +278,12 @@ func (s *logStore) ListAttempt(ctx context.Context, req driver.ListAttemptReques limit = 100 } - res, err := pagination.Run(ctx, pagination.Config[attemptRecordWithTimeID]{ + res, err := pagination.Run(ctx, pagination.Config[attemptRecordWithPosition]{ Limit: limit, Order: sortOrder, Next: req.Next, Prev: req.Prev, - Fetch: func(ctx context.Context, q pagination.QueryInput) ([]attemptRecordWithTimeID, error) { + Fetch: func(ctx context.Context, q pagination.QueryInput) ([]attemptRecordWithPosition, error) { query, args := buildAttemptQuery(req, q) rows, err := s.db.Query(ctx, query, args...) if err != nil { @@ -230,9 +292,10 @@ func (s *logStore) ListAttempt(ctx context.Context, req driver.ListAttemptReques defer rows.Close() return scanAttemptRecords(rows) }, - Cursor: pagination.Cursor[attemptRecordWithTimeID]{ - Encode: func(ar attemptRecordWithTimeID) string { - return cursor.Encode(cursorResourceAttempt, cursorVersion, ar.TimeAttemptID) + Cursor: pagination.Cursor[attemptRecordWithPosition]{ + Encode: func(ar attemptRecordWithPosition) string { + position := fmt.Sprintf("%d::%s", ar.attemptTime.UnixMilli(), ar.Attempt.ID) + return cursor.Encode(cursorResourceAttempt, cursorVersion, position) }, Decode: func(c string) (string, error) { return cursor.Decode(c, cursorResourceAttempt, cursorVersion) @@ -257,108 +320,169 @@ func (s *logStore) ListAttempt(ctx context.Context, req driver.ListAttemptReques } func buildAttemptQuery(req driver.ListAttemptRequest, q pagination.QueryInput) (string, []any) { - cursorCondition := fmt.Sprintf("AND ($10::text = '' OR idx.time_attempt_id %s $10::text)", q.Compare) - orderByClause := fmt.Sprintf("idx.attempt_time %s, idx.attempt_id %s", strings.ToUpper(q.SortDir), strings.ToUpper(q.SortDir)) + var conditions []string + var args []any + argNum := 1 + + if req.TenantID != "" { + conditions = append(conditions, fmt.Sprintf("tenant_id = $%d", argNum)) + args = append(args, req.TenantID) + argNum++ + } + + if req.EventID != "" { + conditions = append(conditions, fmt.Sprintf("event_id = $%d", argNum)) + args = append(args, req.EventID) + argNum++ + } + + if len(req.DestinationIDs) > 0 { + conditions = append(conditions, fmt.Sprintf("destination_id = ANY($%d)", argNum)) + args = append(args, req.DestinationIDs) + argNum++ + } + + if req.Status != "" { + conditions = append(conditions, fmt.Sprintf("status = $%d", argNum)) + args = append(args, req.Status) + argNum++ + } + + if len(req.Topics) > 0 { + conditions = append(conditions, fmt.Sprintf("topic = ANY($%d)", argNum)) + args = append(args, req.Topics) + argNum++ + } + + if req.TimeFilter.GTE != nil { + conditions = append(conditions, fmt.Sprintf("time >= $%d", argNum)) + args = append(args, *req.TimeFilter.GTE) + argNum++ + } + if req.TimeFilter.LTE != nil { + conditions = append(conditions, fmt.Sprintf("time <= $%d", argNum)) + args = append(args, *req.TimeFilter.LTE) + argNum++ + } + if req.TimeFilter.GT != nil { + conditions = append(conditions, fmt.Sprintf("time > $%d", argNum)) + args = append(args, *req.TimeFilter.GT) + argNum++ + } + if req.TimeFilter.LT != nil { + conditions = append(conditions, fmt.Sprintf("time < $%d", argNum)) + args = append(args, *req.TimeFilter.LT) + argNum++ + } + + if q.CursorPos != "" { + cursorCond, cursorArgs := buildAttemptCursorCondition(q.Compare, q.CursorPos, argNum) + conditions = append(conditions, cursorCond) + args = append(args, cursorArgs...) + argNum += len(cursorArgs) + } + + whereClause := strings.Join(conditions, " AND ") + if whereClause == "" { + whereClause = "1=1" + } + + orderByClause := fmt.Sprintf("ORDER BY time %s, id %s", + strings.ToUpper(q.SortDir), strings.ToUpper(q.SortDir)) query := fmt.Sprintf(` SELECT - idx.event_id, - idx.attempt_id, - idx.destination_id, - idx.event_time, - idx.attempt_time, - idx.topic, - idx.status, - idx.time_attempt_id, - e.tenant_id, - e.eligible_for_retry, - e.data, - e.metadata, - a.code, - a.response_data, - idx.manual, - idx.attempt_number - FROM event_attempt_index idx - JOIN events e ON e.id = idx.event_id AND e.time = idx.event_time - JOIN attempts a ON a.id = idx.attempt_id AND a.time = idx.attempt_time - WHERE ($1::text = '' OR idx.tenant_id = $1) - AND ($2::text = '' OR idx.event_id = $2) - AND (array_length($3::text[], 1) IS NULL OR idx.destination_id = ANY($3)) - AND ($4::text = '' OR idx.status = $4) - AND (array_length($5::text[], 1) IS NULL OR idx.topic = ANY($5)) - AND ($6::timestamptz IS NULL OR idx.attempt_time >= $6) - AND ($7::timestamptz IS NULL OR idx.attempt_time <= $7) - AND ($8::timestamptz IS NULL OR idx.attempt_time > $8) - AND ($9::timestamptz IS NULL OR idx.attempt_time < $9) + id, + event_id, + tenant_id, + destination_id, + topic, + status, + time, + attempt_number, + manual, + code, + response_data, + event_time, + eligible_for_retry, + event_data, + event_metadata + FROM attempts + WHERE %s %s - ORDER BY %s - LIMIT $11 - `, cursorCondition, orderByClause) - - args := []any{ - req.TenantID, // $1 - req.EventID, // $2 - req.DestinationIDs, // $3 - req.Status, // $4 - req.Topics, // $5 - req.TimeFilter.GTE, // $6 - req.TimeFilter.LTE, // $7 - req.TimeFilter.GT, // $8 - req.TimeFilter.LT, // $9 - q.CursorPos, // $10 - q.Limit, // $11 - } + LIMIT $%d + `, whereClause, orderByClause, argNum) + + args = append(args, q.Limit) return query, args } -func scanAttemptRecords(rows pgx.Rows) ([]attemptRecordWithTimeID, error) { - var results []attemptRecordWithTimeID +func buildAttemptCursorCondition(compare, position string, argOffset int) (string, []any) { + parts := strings.SplitN(position, "::", 2) + if len(parts) != 2 { + return "1=1", nil // invalid cursor, return always true + } + attemptTimeMs, err := parseTimestampMs(parts[0]) + if err != nil { + return "1=1", nil // invalid timestamp, return always true + } + attemptID := parts[1] + + // Convert milliseconds to PostgreSQL timestamp + condition := fmt.Sprintf(`( + time %s to_timestamp($%d / 1000.0) + OR (time = to_timestamp($%d / 1000.0) AND id %s $%d) + )`, compare, argOffset, argOffset+1, compare, argOffset+2) + + return condition, []any{attemptTimeMs, attemptTimeMs, attemptID} +} + +func scanAttemptRecords(rows pgx.Rows) ([]attemptRecordWithPosition, error) { + var results []attemptRecordWithPosition for rows.Next() { var ( + id string eventID string - attemptID string + tenantID string destinationID string - eventTime time.Time - attemptTime time.Time topic string status string - timeAttemptID string - tenantID string - eligibleForRetry bool - data map[string]any - metadata map[string]string + attemptTime time.Time + attemptNumber int + manual bool code string responseData map[string]any - manual bool - attemptNumber int + eventTime time.Time + eligibleForRetry bool + eventData map[string]any + eventMetadata map[string]string ) if err := rows.Scan( + &id, &eventID, - &attemptID, + &tenantID, &destinationID, - &eventTime, - &attemptTime, &topic, &status, - &timeAttemptID, - &tenantID, - &eligibleForRetry, - &data, - &metadata, + &attemptTime, + &attemptNumber, + &manual, &code, &responseData, - &manual, - &attemptNumber, + &eventTime, + &eligibleForRetry, + &eventData, + &eventMetadata, ); err != nil { return nil, fmt.Errorf("scan failed: %w", err) } - results = append(results, attemptRecordWithTimeID{ + results = append(results, attemptRecordWithPosition{ AttemptRecord: &driver.AttemptRecord{ Attempt: &models.Attempt{ - ID: attemptID, + ID: id, TenantID: tenantID, EventID: eventID, DestinationID: destinationID, @@ -376,11 +500,11 @@ func scanAttemptRecords(rows pgx.Rows) ([]attemptRecordWithTimeID, error) { Topic: topic, EligibleForRetry: eligibleForRetry, Time: eventTime, - Data: data, - Metadata: metadata, + Data: eventData, + Metadata: eventMetadata, }, }, - TimeAttemptID: timeAttemptID, + attemptTime: attemptTime, }) } @@ -392,43 +516,44 @@ func scanAttemptRecords(rows pgx.Rows) ([]attemptRecordWithTimeID, error) { } func (s *logStore) RetrieveEvent(ctx context.Context, req driver.RetrieveEventRequest) (*models.Event, error) { - var query string + var conditions []string var args []any + argNum := 1 + + if req.TenantID != "" { + conditions = append(conditions, fmt.Sprintf("tenant_id = $%d", argNum)) + args = append(args, req.TenantID) + argNum++ + } + + conditions = append(conditions, fmt.Sprintf("event_id = $%d", argNum)) + args = append(args, req.EventID) + argNum++ if req.DestinationID != "" { - query = ` - SELECT - e.id, - e.tenant_id, - $3 as destination_id, - e.time, - e.topic, - e.eligible_for_retry, - e.data, - e.metadata - FROM events e - WHERE ($1::text = '' OR e.tenant_id = $1) AND e.id = $2 - AND EXISTS ( - SELECT 1 FROM event_attempt_index idx - WHERE ($1::text = '' OR idx.tenant_id = $1) AND idx.event_id = $2 AND idx.destination_id = $3 - )` - args = []any{req.TenantID, req.EventID, req.DestinationID} - } else { - query = ` - SELECT - e.id, - e.tenant_id, - e.destination_id, - e.time, - e.topic, - e.eligible_for_retry, - e.data, - e.metadata - FROM events e - WHERE ($1::text = '' OR e.tenant_id = $1) AND e.id = $2` - args = []any{req.TenantID, req.EventID} + conditions = append(conditions, fmt.Sprintf("destination_id = $%d", argNum)) + args = append(args, req.DestinationID) + argNum++ } + whereClause := strings.Join(conditions, " AND ") + + // Query from attempts table (denormalized) to get event data + // This handles destination filtering correctly + query := fmt.Sprintf(` + SELECT + event_id, + tenant_id, + destination_id, + topic, + eligible_for_retry, + event_time, + event_metadata, + event_data + FROM attempts + WHERE %s + LIMIT 1`, whereClause) + row := s.db.QueryRow(ctx, query, args...) event := &models.Event{} @@ -436,82 +561,95 @@ func (s *logStore) RetrieveEvent(ctx context.Context, req driver.RetrieveEventRe &event.ID, &event.TenantID, &event.DestinationID, - &event.Time, &event.Topic, &event.EligibleForRetry, - &event.Data, + &event.Time, &event.Metadata, + &event.Data, ) if err == pgx.ErrNoRows { return nil, nil } if err != nil { - return nil, err + return nil, fmt.Errorf("scan failed: %w", err) } return event, nil } func (s *logStore) RetrieveAttempt(ctx context.Context, req driver.RetrieveAttemptRequest) (*driver.AttemptRecord, error) { - query := ` + var conditions []string + var args []any + argNum := 1 + + if req.TenantID != "" { + conditions = append(conditions, fmt.Sprintf("tenant_id = $%d", argNum)) + args = append(args, req.TenantID) + argNum++ + } + + conditions = append(conditions, fmt.Sprintf("id = $%d", argNum)) + args = append(args, req.AttemptID) + + whereClause := strings.Join(conditions, " AND ") + + query := fmt.Sprintf(` SELECT - idx.event_id, - idx.attempt_id, - idx.destination_id, - idx.event_time, - idx.attempt_time, - idx.topic, - idx.status, - e.tenant_id, - e.eligible_for_retry, - e.data, - e.metadata, - a.code, - a.response_data, - idx.manual, - idx.attempt_number - FROM event_attempt_index idx - JOIN events e ON e.id = idx.event_id AND e.time = idx.event_time - JOIN attempts a ON a.id = idx.attempt_id AND a.time = idx.attempt_time - WHERE ($1::text = '' OR idx.tenant_id = $1) AND idx.attempt_id = $2 - LIMIT 1` - - row := s.db.QueryRow(ctx, query, req.TenantID, req.AttemptID) + id, + event_id, + tenant_id, + destination_id, + topic, + status, + time, + attempt_number, + manual, + code, + response_data, + event_time, + eligible_for_retry, + event_data, + event_metadata + FROM attempts + WHERE %s + LIMIT 1`, whereClause) + + row := s.db.QueryRow(ctx, query, args...) var ( + id string eventID string - attemptID string + tenantID string destinationID string - eventTime time.Time - attemptTime time.Time topic string status string - tenantID string - eligibleForRetry bool - data map[string]any - metadata map[string]string + attemptTime time.Time + attemptNumber int + manual bool code string responseData map[string]any - manual bool - attemptNumber int + eventTime time.Time + eligibleForRetry bool + eventData map[string]any + eventMetadata map[string]string ) err := row.Scan( + &id, &eventID, - &attemptID, + &tenantID, &destinationID, - &eventTime, - &attemptTime, &topic, &status, - &tenantID, - &eligibleForRetry, - &data, - &metadata, + &attemptTime, + &attemptNumber, + &manual, &code, &responseData, - &manual, - &attemptNumber, + &eventTime, + &eligibleForRetry, + &eventData, + &eventMetadata, ) if err == pgx.ErrNoRows { return nil, nil @@ -522,7 +660,7 @@ func (s *logStore) RetrieveAttempt(ctx context.Context, req driver.RetrieveAttem return &driver.AttemptRecord{ Attempt: &models.Attempt{ - ID: attemptID, + ID: id, TenantID: tenantID, EventID: eventID, DestinationID: destinationID, @@ -540,8 +678,8 @@ func (s *logStore) RetrieveAttempt(ctx context.Context, req driver.RetrieveAttem Topic: topic, EligibleForRetry: eligibleForRetry, Time: eventTime, - Data: data, - Metadata: metadata, + Data: eventData, + Metadata: eventMetadata, }, }, nil } @@ -561,18 +699,13 @@ func (s *logStore) InsertMany(ctx context.Context, entries []*models.LogEntry) e events = append(events, e) } - // Extract attempts - attempts := make([]*models.Attempt, 0, len(entries)) - for _, entry := range entries { - attempts = append(attempts, entry.Attempt) - } - tx, err := s.db.Begin(ctx) if err != nil { return err } defer tx.Rollback(ctx) + // Insert events if len(events) > 0 { _, err = tx.Exec(ctx, ` INSERT INTO events (id, tenant_id, destination_id, time, topic, eligible_for_retry, data, metadata) @@ -580,47 +713,30 @@ func (s *logStore) InsertMany(ctx context.Context, entries []*models.LogEntry) e ON CONFLICT (time, id) DO NOTHING `, eventArrays(events)...) if err != nil { - return err + return fmt.Errorf("insert events failed: %w", err) } } - if len(attempts) > 0 { + // Insert attempts + if len(entries) > 0 { _, err = tx.Exec(ctx, ` - INSERT INTO attempts (id, event_id, destination_id, status, time, code, response_data, manual, attempt_number) - SELECT * FROM unnest($1::text[], $2::text[], $3::text[], $4::text[], $5::timestamptz[], $6::text[], $7::jsonb[], $8::boolean[], $9::integer[]) + INSERT INTO attempts ( + id, event_id, tenant_id, destination_id, topic, status, + time, attempt_number, manual, code, response_data, + event_time, eligible_for_retry, event_data, event_metadata + ) + SELECT * FROM unnest( + $1::text[], $2::text[], $3::text[], $4::text[], $5::text[], $6::text[], + $7::timestamptz[], $8::integer[], $9::boolean[], $10::text[], $11::jsonb[], + $12::timestamptz[], $13::boolean[], $14::jsonb[], $15::jsonb[] + ) ON CONFLICT (time, id) DO UPDATE SET status = EXCLUDED.status, code = EXCLUDED.code, response_data = EXCLUDED.response_data - `, attemptArrays(attempts)...) + `, attemptArrays(entries)...) if err != nil { - return err - } - - _, err = tx.Exec(ctx, ` - INSERT INTO event_attempt_index ( - event_id, attempt_id, tenant_id, destination_id, - event_time, attempt_time, topic, status, manual, attempt_number - ) - SELECT - a.event_id, - a.id, - e.tenant_id, - a.destination_id, - e.time, - a.time, - e.topic, - a.status, - a.manual, - a.attempt_number - FROM unnest($1::text[], $2::text[], $3::text[], $4::text[], $5::timestamptz[], $6::text[], $7::jsonb[], $8::boolean[], $9::integer[]) - AS a(id, event_id, destination_id, status, time, code, response_data, manual, attempt_number) - JOIN events e ON e.id = a.event_id - ON CONFLICT (attempt_time, event_id, attempt_id) DO UPDATE SET - status = EXCLUDED.status - `, attemptArrays(attempts)...) - if err != nil { - return err + return fmt.Errorf("insert attempts failed: %w", err) } } @@ -660,38 +776,62 @@ func eventArrays(events []*models.Event) []any { } } -func attemptArrays(attempts []*models.Attempt) []any { - ids := make([]string, len(attempts)) - eventIDs := make([]string, len(attempts)) - destinationIDs := make([]string, len(attempts)) - statuses := make([]string, len(attempts)) - times := make([]time.Time, len(attempts)) - codes := make([]string, len(attempts)) - responseDatas := make([]map[string]any, len(attempts)) - manuals := make([]bool, len(attempts)) - attemptNumbers := make([]int, len(attempts)) - - for i, a := range attempts { +// attemptArrays extracts arrays from log entries for bulk insert. +func attemptArrays(entries []*models.LogEntry) []any { + n := len(entries) + + ids := make([]string, n) + eventIDs := make([]string, n) + tenantIDs := make([]string, n) + destinationIDs := make([]string, n) + topics := make([]string, n) + statuses := make([]string, n) + times := make([]time.Time, n) + attemptNumbers := make([]int, n) + manuals := make([]bool, n) + codes := make([]string, n) + responseDatas := make([]map[string]any, n) + eventTimes := make([]time.Time, n) + eligibleForRetries := make([]bool, n) + eventDatas := make([]map[string]any, n) + eventMetadatas := make([]map[string]string, n) + + for i, entry := range entries { + a := entry.Attempt + e := entry.Event + ids[i] = a.ID eventIDs[i] = a.EventID + tenantIDs[i] = e.TenantID destinationIDs[i] = a.DestinationID + topics[i] = e.Topic statuses[i] = a.Status times[i] = a.Time + attemptNumbers[i] = a.AttemptNumber + manuals[i] = a.Manual codes[i] = a.Code responseDatas[i] = a.ResponseData - manuals[i] = a.Manual - attemptNumbers[i] = a.AttemptNumber + eventTimes[i] = e.Time + eligibleForRetries[i] = e.EligibleForRetry + eventDatas[i] = e.Data + eventMetadatas[i] = e.Metadata } return []any{ ids, eventIDs, + tenantIDs, destinationIDs, + topics, statuses, times, + attemptNumbers, + manuals, codes, responseDatas, - manuals, - attemptNumbers, + eventTimes, + eligibleForRetries, + eventDatas, + eventMetadatas, } } diff --git a/internal/migrator/migrations/postgres/000005_denormalize_attempts.down.sql b/internal/migrator/migrations/postgres/000005_denormalize_attempts.down.sql new file mode 100644 index 000000000..b3797315e --- /dev/null +++ b/internal/migrator/migrations/postgres/000005_denormalize_attempts.down.sql @@ -0,0 +1,120 @@ +BEGIN; + +-- ============================================================================= +-- Rollback: Restore v0.12 schema (deliveries + event_delivery_index) +-- +-- WARNING: This rollback will NOT restore data deleted during the up migration +-- (orphaned deliveries). The event_delivery_index table will be recreated empty +-- since the original data was buggy anyway. +-- ============================================================================= + +-- ----------------------------------------------------------------------------- +-- Step 1: Drop new indexes +-- ----------------------------------------------------------------------------- +DROP INDEX IF EXISTS idx_events_tenant_time; +DROP INDEX IF EXISTS idx_events_tenant_topic_time; +DROP INDEX IF EXISTS idx_events_id; +DROP INDEX IF EXISTS idx_attempts_tenant_time; +DROP INDEX IF EXISTS idx_attempts_tenant_dest_time; +DROP INDEX IF EXISTS idx_attempts_tenant_status_time; +DROP INDEX IF EXISTS idx_attempts_tenant_topic_time; +DROP INDEX IF EXISTS idx_attempts_event_time; +DROP INDEX IF EXISTS idx_attempts_id; + +-- ----------------------------------------------------------------------------- +-- Step 2: Restore time_id generated column on events +-- ----------------------------------------------------------------------------- +ALTER TABLE events ADD COLUMN time_id text GENERATED ALWAYS AS ( + LPAD( + CAST( + EXTRACT( + EPOCH + FROM time AT TIME ZONE 'UTC' + ) AS BIGINT + )::text, + 10, + '0' + ) || '_' || id +) STORED; + +-- ----------------------------------------------------------------------------- +-- Step 3: Rename attempts -> deliveries, attempt_number -> attempt +-- ----------------------------------------------------------------------------- +ALTER TABLE attempts RENAME COLUMN attempt_number TO attempt; +ALTER TABLE attempts RENAME TO deliveries; +ALTER TABLE attempts_default RENAME TO deliveries_default; + +-- ----------------------------------------------------------------------------- +-- Step 4: Drop denormalized columns from deliveries +-- ----------------------------------------------------------------------------- +ALTER TABLE deliveries + DROP COLUMN tenant_id, + DROP COLUMN topic, + DROP COLUMN event_time, + DROP COLUMN eligible_for_retry, + DROP COLUMN event_data, + DROP COLUMN event_metadata; + +-- ----------------------------------------------------------------------------- +-- Step 5: Recreate event_delivery_index table (empty - original data was buggy) +-- ----------------------------------------------------------------------------- +CREATE TABLE event_delivery_index ( + event_id text NOT NULL, + delivery_id text NOT NULL, + tenant_id text NOT NULL, + destination_id text NOT NULL, + event_time timestamptz NOT NULL, + delivery_time timestamptz NOT NULL, + topic text NOT NULL, + status text NOT NULL, + manual boolean DEFAULT false NOT NULL, + attempt integer DEFAULT 0 NOT NULL, + time_event_id text GENERATED ALWAYS AS ( + LPAD( + CAST( + EXTRACT( + EPOCH + FROM event_time AT TIME ZONE 'UTC' + ) AS BIGINT + )::text, + 10, + '0' + ) || '_' || event_id + ) STORED, + time_delivery_id text GENERATED ALWAYS AS ( + LPAD( + CAST( + EXTRACT( + EPOCH + FROM delivery_time AT TIME ZONE 'UTC' + ) AS BIGINT + )::text, + 10, + '0' + ) || '_' || delivery_id + ) STORED, + PRIMARY KEY (delivery_time, event_id, delivery_id) +) PARTITION BY RANGE (delivery_time); + +CREATE TABLE event_delivery_index_default PARTITION OF event_delivery_index DEFAULT; + +CREATE INDEX IF NOT EXISTS idx_event_delivery_index_main ON event_delivery_index( + tenant_id, + destination_id, + topic, + status, + event_time DESC, + delivery_time DESC, + time_event_id, + time_delivery_id +); + +-- ----------------------------------------------------------------------------- +-- Step 6: Restore old indexes on events and deliveries +-- ----------------------------------------------------------------------------- +CREATE INDEX ON events (tenant_id, time_id DESC); +CREATE INDEX ON events (tenant_id, destination_id); +CREATE INDEX ON deliveries (event_id); +CREATE INDEX ON deliveries (event_id, status); + +COMMIT; diff --git a/internal/migrator/migrations/postgres/000005_denormalize_attempts.up.sql b/internal/migrator/migrations/postgres/000005_denormalize_attempts.up.sql new file mode 100644 index 000000000..37fea4848 --- /dev/null +++ b/internal/migrator/migrations/postgres/000005_denormalize_attempts.up.sql @@ -0,0 +1,118 @@ +BEGIN; + +-- ============================================================================= +-- Migration: Restructure for denormalized attempts table +-- +-- Changes: +-- 1. Add denormalized event columns to deliveries +-- 2. Backfill data from events table +-- 3. Rename deliveries -> attempts +-- 4. Drop buggy event_delivery_index table (delivery_id was stored incorrectly) +-- 5. Drop time_id from events (cursor now computed dynamically) +-- 6. Create new indexes optimized for the query patterns +-- ============================================================================= + +-- ----------------------------------------------------------------------------- +-- Step 1: Add denormalized columns to deliveries +-- ----------------------------------------------------------------------------- +ALTER TABLE deliveries + ADD COLUMN tenant_id text, + ADD COLUMN topic text, + ADD COLUMN event_time timestamptz, + ADD COLUMN eligible_for_retry boolean, + ADD COLUMN event_data jsonb, + ADD COLUMN event_metadata jsonb; + +-- ----------------------------------------------------------------------------- +-- Step 2: Backfill denormalized data from events +-- Note: event_delivery_index has buggy data (delivery_id stored incorrectly), +-- so we backfill directly from events table using event_id foreign key. +-- ----------------------------------------------------------------------------- +UPDATE deliveries d SET + tenant_id = e.tenant_id, + topic = e.topic, + event_time = e.time, + eligible_for_retry = e.eligible_for_retry, + event_data = e.data, + event_metadata = e.metadata +FROM events e +WHERE d.event_id = e.id; + +-- Handle any orphaned deliveries (delivery exists but event doesn't) +-- These would have NULL values - delete them as they're invalid +DELETE FROM deliveries WHERE tenant_id IS NULL; + +-- ----------------------------------------------------------------------------- +-- Step 3: Make denormalized columns NOT NULL +-- ----------------------------------------------------------------------------- +ALTER TABLE deliveries + ALTER COLUMN tenant_id SET NOT NULL, + ALTER COLUMN topic SET NOT NULL, + ALTER COLUMN event_time SET NOT NULL, + ALTER COLUMN eligible_for_retry SET NOT NULL, + ALTER COLUMN event_data SET NOT NULL, + ALTER COLUMN event_metadata SET NOT NULL; + +-- ----------------------------------------------------------------------------- +-- Step 4: Rename deliveries -> attempts, attempt -> attempt_number +-- ----------------------------------------------------------------------------- +ALTER TABLE deliveries RENAME TO attempts; +ALTER TABLE deliveries_default RENAME TO attempts_default; +ALTER TABLE attempts RENAME COLUMN attempt TO attempt_number; + +-- ----------------------------------------------------------------------------- +-- Step 5: Drop buggy event_delivery_index table +-- The delivery_id column was storing incorrect values, making this table +-- unreliable. The denormalized attempts table replaces its functionality. +-- ----------------------------------------------------------------------------- +DROP TABLE event_delivery_index CASCADE; + +-- ----------------------------------------------------------------------------- +-- Step 6: Drop time_id from events +-- Cursor position is now computed dynamically as {time_ms}::{id} to match +-- chlogstore format, rather than stored as a generated column. +-- ----------------------------------------------------------------------------- +ALTER TABLE events DROP COLUMN time_id; + +-- ----------------------------------------------------------------------------- +-- Step 7: Drop old indexes +-- ----------------------------------------------------------------------------- +DROP INDEX IF EXISTS events_tenant_id_time_id_idx; +DROP INDEX IF EXISTS events_tenant_id_destination_id_idx; +DROP INDEX IF EXISTS deliveries_event_id_idx; +DROP INDEX IF EXISTS deliveries_event_id_status_idx; + +-- ----------------------------------------------------------------------------- +-- Step 8: Create new indexes for events +-- ----------------------------------------------------------------------------- +-- Primary list query: paginate by (time, id) with tenant filter +CREATE INDEX idx_events_tenant_time ON events (tenant_id, time DESC, id DESC); + +-- Filter by topic +CREATE INDEX idx_events_tenant_topic_time ON events (tenant_id, topic, time DESC, id DESC); + +-- Point lookup by event_id (scans partition indexes) +CREATE INDEX idx_events_id ON events (id); + +-- ----------------------------------------------------------------------------- +-- Step 9: Create new indexes for attempts +-- ----------------------------------------------------------------------------- +-- Primary list query: paginate by (time, id) with tenant filter +CREATE INDEX idx_attempts_tenant_time ON attempts (tenant_id, time DESC, id DESC); + +-- Filter by destination +CREATE INDEX idx_attempts_tenant_dest_time ON attempts (tenant_id, destination_id, time DESC, id DESC); + +-- Filter by status +CREATE INDEX idx_attempts_tenant_status_time ON attempts (tenant_id, status, time DESC, id DESC); + +-- Filter by topic +CREATE INDEX idx_attempts_tenant_topic_time ON attempts (tenant_id, topic, time DESC, id DESC); + +-- Filter by event_id (for "attempts for this event" queries) +CREATE INDEX idx_attempts_event_time ON attempts (event_id, time DESC, id DESC); + +-- Point lookup by attempt_id (scans partition indexes) +CREATE INDEX idx_attempts_id ON attempts (id); + +COMMIT; diff --git a/internal/migrator/migrations/postgres/000005_rename_delivery_to_attempt.down.sql b/internal/migrator/migrations/postgres/000005_rename_delivery_to_attempt.down.sql deleted file mode 100644 index 21f8575dd..000000000 --- a/internal/migrator/migrations/postgres/000005_rename_delivery_to_attempt.down.sql +++ /dev/null @@ -1,48 +0,0 @@ -BEGIN; - --- Drop new index and restore old one -DROP INDEX IF EXISTS idx_event_attempt_index_main; - --- Restore generated column with old name -ALTER TABLE event_attempt_index DROP COLUMN time_attempt_id; -ALTER TABLE event_attempt_index ADD COLUMN time_delivery_id text GENERATED ALWAYS AS ( - LPAD( - CAST( - EXTRACT( - EPOCH - FROM attempt_time AT TIME ZONE 'UTC' - ) AS BIGINT - )::text, - 10, - '0' - ) || '_' || attempt_id -) STORED; - --- Rename columns back in event_attempt_index -ALTER TABLE event_attempt_index RENAME COLUMN attempt_number TO attempt; -ALTER TABLE event_attempt_index RENAME COLUMN attempt_time TO delivery_time; -ALTER TABLE event_attempt_index RENAME COLUMN attempt_id TO delivery_id; - --- Rename tables back -ALTER TABLE event_attempt_index RENAME TO event_delivery_index; -ALTER TABLE event_attempt_index_default RENAME TO event_delivery_index_default; - --- Rename column back in attempts: attempt_number -> attempt -ALTER TABLE attempts RENAME COLUMN attempt_number TO attempt; - -ALTER TABLE attempts RENAME TO deliveries; -ALTER TABLE attempts_default RENAME TO deliveries_default; - --- Recreate old index -CREATE INDEX IF NOT EXISTS idx_event_delivery_index_main ON event_delivery_index( - tenant_id, - destination_id, - topic, - status, - event_time DESC, - delivery_time DESC, - time_event_id, - time_delivery_id -); - -COMMIT; diff --git a/internal/migrator/migrations/postgres/000005_rename_delivery_to_attempt.up.sql b/internal/migrator/migrations/postgres/000005_rename_delivery_to_attempt.up.sql deleted file mode 100644 index ea1436235..000000000 --- a/internal/migrator/migrations/postgres/000005_rename_delivery_to_attempt.up.sql +++ /dev/null @@ -1,47 +0,0 @@ -BEGIN; - --- Rename deliveries table to attempts -ALTER TABLE deliveries RENAME TO attempts; -ALTER TABLE deliveries_default RENAME TO attempts_default; - --- Rename column in attempts: attempt -> attempt_number -ALTER TABLE attempts RENAME COLUMN attempt TO attempt_number; - --- Rename event_delivery_index table to event_attempt_index -ALTER TABLE event_delivery_index RENAME TO event_attempt_index; -ALTER TABLE event_delivery_index_default RENAME TO event_attempt_index_default; - --- Rename columns in event_attempt_index -ALTER TABLE event_attempt_index RENAME COLUMN delivery_id TO attempt_id; -ALTER TABLE event_attempt_index RENAME COLUMN delivery_time TO attempt_time; -ALTER TABLE event_attempt_index RENAME COLUMN attempt TO attempt_number; - --- Drop and recreate generated column with new name -ALTER TABLE event_attempt_index DROP COLUMN time_delivery_id; -ALTER TABLE event_attempt_index ADD COLUMN time_attempt_id text GENERATED ALWAYS AS ( - LPAD( - CAST( - EXTRACT( - EPOCH - FROM attempt_time AT TIME ZONE 'UTC' - ) AS BIGINT - )::text, - 10, - '0' - ) || '_' || attempt_id -) STORED; - --- Drop old index and create new one with updated column names -DROP INDEX IF EXISTS idx_event_delivery_index_main; -CREATE INDEX IF NOT EXISTS idx_event_attempt_index_main ON event_attempt_index( - tenant_id, - destination_id, - topic, - status, - event_time DESC, - attempt_time DESC, - time_event_id, - time_attempt_id -); - -COMMIT;