diff --git a/internal/repository/postgres.go b/internal/repository/postgres.go index a91f131..42f0af8 100644 --- a/internal/repository/postgres.go +++ b/internal/repository/postgres.go @@ -789,12 +789,7 @@ func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor stri if limit <= 0 { limit = 50 } - off := 0 - if cursor != "" { - if n, err := strconv.Atoi(cursor); err == nil && n >= 0 { - off = n - } - } + afterID, off, useOffset := store.ParsePrefixPageCursor(cursor) ctx := context.Background() var one int if err := p.pool.QueryRow(ctx, ` @@ -805,34 +800,52 @@ func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor stri } return nil, "", false } - rows, err := p.pool.Query(ctx, ` - SELECT prefix::text, community_id::text, source FROM revision_materialized_prefix - WHERE revision_id=$1 ORDER BY id - LIMIT $2 OFFSET $3`, revisionID, limit+1, off) + var rows pgx.Rows + var err error + if useOffset { + rows, err = p.pool.Query(ctx, ` + SELECT id, prefix::text, community_id::text, source FROM revision_materialized_prefix + WHERE revision_id=$1::uuid ORDER BY id + LIMIT $2 OFFSET $3`, revisionID, limit+1, off) + } else { + var afterArg any + if afterID != nil { + afterArg = *afterID + } + rows, err = p.pool.Query(ctx, ` + SELECT id, prefix::text, community_id::text, source FROM revision_materialized_prefix + WHERE revision_id=$1::uuid AND ($2::bigint IS NULL OR id > $2::bigint) + ORDER BY id + LIMIT $3`, revisionID, afterArg, limit+1) + } if err != nil { return nil, "", false } defer rows.Close() var all []store.PrefixRow + var ids []int64 for rows.Next() { + var rowID int64 var pr store.PrefixRow var comm *string - if err := rows.Scan(&pr.Prefix, &comm, &pr.Source); err != nil { + if err := rows.Scan(&rowID, &pr.Prefix, &comm, &pr.Source); err != nil { continue } pr.CommunityID = comm + ids = append(ids, rowID) all = append(all, pr) } agentDebugNDJSON3214("B", "repository/postgres.go:ListRevisionPrefixes", "list_prefixes_fetched", map[string]any{ - "rows": len(all), "limit": limit, "offset": off, + "rows": len(all), "limit": limit, "keyset": !useOffset, }) more := len(all) > limit if more { all = all[:limit] + ids = ids[:limit] } next := "" - if more { - next = fmt.Sprintf("%d", off+limit) + if more && len(ids) > 0 { + next = store.FormatPrefixPageCursor(ids[len(ids)-1]) } if len(all) == 0 { return nil, "", false @@ -896,21 +909,21 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro ctx := context.Background() var unchanged int err := p.pool.QueryRow(ctx, ` - SELECT COUNT(*)::int FROM ( - SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid - INTERSECT - SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid - ) t`, aID, bID).Scan(&unchanged) + SELECT COUNT(*)::int FROM revision_materialized_prefix b + INNER JOIN revision_materialized_prefix a + ON a.revision_id = $1::uuid AND a.prefix = b.prefix + WHERE b.revision_id = $2::uuid`, aID, bID).Scan(&unchanged) if err != nil { return nil, err } - // added: в B, нет в A; removed: в A, нет в B — без загрузки полных снапшотов в память. rowsAdded, err := p.pool.Query(ctx, ` - SELECT prefix::text FROM ( - SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid - EXCEPT - SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid - ) s ORDER BY 1 LIMIT $3`, bID, aID, maxRevisionDiffRows+1) + SELECT b.prefix::text + FROM revision_materialized_prefix b + LEFT JOIN revision_materialized_prefix a + ON a.revision_id = $1::uuid AND a.prefix = b.prefix + WHERE b.revision_id = $2::uuid AND a.prefix IS NULL + ORDER BY b.prefix + LIMIT $3`, aID, bID, maxRevisionDiffRows+1) if err != nil { return nil, err } @@ -929,11 +942,13 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro } addedTruncated := len(added) >= maxRevisionDiffRows rowsRem, err := p.pool.Query(ctx, ` - SELECT prefix::text FROM ( - SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid - EXCEPT - SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid - ) s ORDER BY 1 LIMIT $3`, aID, bID, maxRevisionDiffRows+1) + SELECT a.prefix::text + FROM revision_materialized_prefix a + LEFT JOIN revision_materialized_prefix b + ON b.revision_id = $2::uuid AND b.prefix = a.prefix + WHERE a.revision_id = $1::uuid AND b.prefix IS NULL + ORDER BY a.prefix + LIMIT $3`, aID, bID, maxRevisionDiffRows+1) if err != nil { return nil, err } diff --git a/internal/store/memory_crud.go b/internal/store/memory_crud.go index b61eec9..6caeb17 100644 --- a/internal/store/memory_crud.go +++ b/internal/store/memory_crud.go @@ -1,8 +1,6 @@ package store import ( - "fmt" - "strconv" "strings" "time" @@ -795,31 +793,37 @@ func (m *Memory) ListRevisionPrefixes(tenantID, revisionID string, cursor string if limit <= 0 { limit = 50 } + afterID, off, useOffset := ParsePrefixPageCursor(cursor) m.mu.RLock() defer m.mu.RUnlock() if _, err := m.getRevisionLocked(tenantID, revisionID); err != nil { return nil, "", false } - all := m.revPrefixes[revisionID] - off := 0 - if cursor != "" { - if n, err := strconv.Atoi(cursor); err == nil && n >= 0 { - off = n - } + allRows := m.revPrefixes[revisionID] + start := 0 + if useOffset { + start = off + } else if afterID != nil { + start = int(*afterID) + 1 } - end := off + limit - next := "" - more := false - if end > len(all) { - end = len(all) - } else { - more = true - next = fmt.Sprintf("%d", end) - } - if off >= len(all) { + if start > len(allRows) { return nil, "", false } - return all[off:end], next, more + end := start + limit + next := "" + more := false + if end > len(allRows) { + end = len(allRows) + } else { + more = true + next = FormatPrefixPageCursor(int64(end - 1)) + } + if start >= end { + return nil, "", false + } + out := make([]PrefixRow, end-start) + copy(out, allRows[start:end]) + return out, next, more } func (m *Memory) ListGlobalSettings(tenantID string) (map[string]any, error) { diff --git a/internal/store/pagination.go b/internal/store/pagination.go new file mode 100644 index 0000000..0392276 --- /dev/null +++ b/internal/store/pagination.go @@ -0,0 +1,30 @@ +package store + +import ( + "strconv" + "strings" +) + +// ParsePrefixPageCursor decodes opaque cursors for revision prefix pagination. +func ParsePrefixPageCursor(cursor string) (afterID *int64, offset int, useOffset bool) { + cursor = strings.TrimSpace(cursor) + if cursor == "" { + return nil, 0, false + } + if strings.HasPrefix(cursor, "o:") { + n, err := strconv.Atoi(strings.TrimPrefix(cursor, "o:")) + if err != nil || n < 0 { + return nil, 0, false + } + return nil, n, true + } + if n, err := strconv.ParseInt(cursor, 10, 64); err == nil && n >= 0 { + return &n, 0, false + } + return nil, 0, false +} + +// FormatPrefixPageCursor encodes the keyset cursor (last row id or slice index). +func FormatPrefixPageCursor(lastID int64) string { + return strconv.FormatInt(lastID, 10) +}