getChanges is a method that retrieves the changes that occurred in the relation tuples within a specified transaction. ctx: The context.Context instance for managing the life-cycle of this function. value: The ID of the transaction for which to retrieve the changes. tenantID: The ID of the tenant f
(ctx context.Context, value db.XID8, tenantID string)
| 221 | // This method returns a TupleChanges instance that encapsulates the changes in the relation tuples within the specified |
| 222 | // transaction, or an error if something went wrong during execution. |
| 223 | func (w *Watch) getChanges(ctx context.Context, value db.XID8, tenantID string) (*base.DataChanges, error) { |
| 224 | // Initialize a new TupleChanges instance. |
| 225 | changes := &base.DataChanges{} |
| 226 | |
| 227 | slog.DebugContext(ctx, "retrieving changes for transaction", slog.Any("id", value), slog.Any("tenant_id", tenantID)) |
| 228 | // Build relation tuples query |
| 229 | // Construct the SQL SELECT statement for retrieving the changes from the RelationTuplesTable. |
| 230 | tbuilder := w.database.Builder.Select("entity_type, entity_id, relation, subject_type, subject_id, subject_relation, expired_tx_id"). |
| 231 | From(RelationTuplesTable). |
| 232 | Where(squirrel.Eq{"tenant_id": tenantID}).Where(squirrel.Or{ |
| 233 | squirrel.Eq{"created_tx_id": value}, |
| 234 | squirrel.Eq{"expired_tx_id": value}, |
| 235 | }) |
| 236 | |
| 237 | // Generate the SQL query and arguments. |
| 238 | tquery, targs, err := tbuilder.ToSql() |
| 239 | if err != nil { |
| 240 | slog.ErrorContext(ctx, "error while building sql query for relation tuples", slog.Any("error", err)) |
| 241 | return nil, err |
| 242 | } |
| 243 | |
| 244 | slog.DebugContext(ctx, "executing sql query for relation tuples", slog.Any("query", tquery), slog.Any("arguments", targs)) |
| 245 | // Execute tuples query |
| 246 | // Execute the SQL query and retrieve the result rows. |
| 247 | var trows pgx.Rows |
| 248 | trows, err = w.database.ReadPool.Query(ctx, tquery, targs...) |
| 249 | if err != nil { |
| 250 | slog.ErrorContext(ctx, "failed to execute sql query for relation tuples", slog.Any("error", err)) |
| 251 | return nil, errors.New(base.ErrorCode_ERROR_CODE_EXECUTION.String()) |
| 252 | } |
| 253 | // Ensure the rows are closed after processing. |
| 254 | defer trows.Close() |
| 255 | |
| 256 | abuilder := w.database.Builder.Select("entity_type, entity_id, attribute, value, expired_tx_id"). |
| 257 | From(AttributesTable). |
| 258 | Where(squirrel.Eq{"tenant_id": tenantID}).Where(squirrel.Or{ |
| 259 | squirrel.Eq{"created_tx_id": value}, |
| 260 | squirrel.Eq{"expired_tx_id": value}, |
| 261 | }) |
| 262 | |
| 263 | aquery, aargs, err := abuilder.ToSql() |
| 264 | if err != nil { |
| 265 | slog.ErrorContext(ctx, "error while building SQL query for attributes", slog.Any("error", err)) |
| 266 | return nil, err |
| 267 | } |
| 268 | |
| 269 | slog.DebugContext(ctx, "executing sql query for attributes", slog.Any("query", aquery), slog.Any("arguments", aargs)) |
| 270 | // Execute attributes query |
| 271 | var arows pgx.Rows |
| 272 | arows, err = w.database.ReadPool.Query(ctx, aquery, aargs...) |
| 273 | if err != nil { |
| 274 | slog.ErrorContext(ctx, "error while executing SQL query for attributes", slog.Any("error", err)) |
| 275 | return nil, errors.New(base.ErrorCode_ERROR_CODE_EXECUTION.String()) |
| 276 | } |
| 277 | // Ensure the rows are closed after processing. |
| 278 | defer arows.Close() |
| 279 | |
| 280 | // Set the snapshot token for the changes. |