QueryRelationships reads relation tuples from the storage based on the given filter.
(ctx context.Context, tenantID string, filter *base.TupleFilter, snap string, pagination database.CursorPagination)
| 40 | |
| 41 | // QueryRelationships reads relation tuples from the storage based on the given filter. |
| 42 | func (r *DataReader) QueryRelationships(ctx context.Context, tenantID string, filter *base.TupleFilter, snap string, pagination database.CursorPagination) (it *database.TupleIterator, err error) { |
| 43 | // Start a new trace span and end it when the function exits. |
| 44 | ctx, span := internal.Tracer.Start(ctx, "data-reader.query-relationships") |
| 45 | defer span.End() |
| 46 | // Log query operation |
| 47 | slog.DebugContext(ctx, "querying relationships for tenant_id", slog.String("tenant_id", tenantID)) |
| 48 | // Decode snapshot token |
| 49 | // Decode the snapshot value. |
| 50 | var st token.SnapToken |
| 51 | st, err = snapshot.EncodedToken{Value: snap}.Decode() |
| 52 | if err != nil { |
| 53 | return nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_INTERNAL) |
| 54 | } |
| 55 | |
| 56 | // Build the relationships query based on the provided filter and snapshot value. |
| 57 | var args []interface{} |
| 58 | builder := r.database.Builder.Select("entity_type, entity_id, relation, subject_type, subject_id, subject_relation").From(RelationTuplesTable).Where(squirrel.Eq{"tenant_id": tenantID}) |
| 59 | builder = utils.TuplesFilterQueryForSelectBuilder(builder, filter) |
| 60 | builder = utils.SnapshotQuery(builder, st.(snapshot.Token).Value.Uint, st.(snapshot.Token).Snapshot) |
| 61 | |
| 62 | if pagination.Cursor() != "" { |
| 63 | var t database.ContinuousToken |
| 64 | t, err = utils.EncodedContinuousToken{Value: pagination.Cursor()}.Decode() |
| 65 | if err != nil { |
| 66 | return nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_INVALID_CONTINUOUS_TOKEN) |
| 67 | } |
| 68 | builder = builder.Where(squirrel.GtOrEq{pagination.Sort(): t.(utils.ContinuousToken).Value}) |
| 69 | } |
| 70 | |
| 71 | if pagination.Sort() != "" { |
| 72 | builder = builder.OrderBy(pagination.Sort()) |
| 73 | } |
| 74 | |
| 75 | // Apply limit if specified in pagination |
| 76 | limit := pagination.Limit() |
| 77 | if limit > 0 { |
| 78 | builder = builder.Limit(uint64(limit)) |
| 79 | } |
| 80 | |
| 81 | // Generate the SQL query and arguments. |
| 82 | var query string |
| 83 | query, args, err = builder.ToSql() |
| 84 | if err != nil { |
| 85 | return nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SQL_BUILDER) |
| 86 | } |
| 87 | |
| 88 | slog.DebugContext(ctx, "generated sql query", slog.String("query", query), "with args", slog.Any("arguments", args)) |
| 89 | // Execute query |
| 90 | // Execute the SQL query and retrieve the result rows. |
| 91 | var rows pgx.Rows |
| 92 | rows, err = r.database.ReadPool.Query(ctx, query, args...) |
| 93 | if err != nil { |
| 94 | return nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_EXECUTION) |
| 95 | } |
| 96 | defer rows.Close() |
| 97 | |
| 98 | // Process the result rows and store the relationships in a TupleCollection. |
| 99 | collection := database.NewTupleCollection() |
nothing calls this directly
no test coverage detected