Method getRecentXIDs

internal/storage/postgres/watch.go:160–213

getRecentXIDs fetches a list of XID8 identifiers from the 'transactions' table for all transactions committed after a specified XID value. Parameters: - ctx: A context to control the execution lifetime. - value: The transaction XID after which we need the changes. - tenantID: The ID of t

(ctx context.Context, value uint64, tenantID string)

//   - A slice of XID8 identifiers.
//   - An error if the query fails to execute, or other error occurs during its execution.
func (w *Watch) getRecentXIDs(ctx context.Context, value uint64, tenantID string) ([]db.XID8, error) {
	// Convert the value to a string formatted as a Postgresql XID8 type.
	valStr := fmt.Sprintf("'%v'::xid8", value)

	subquery := fmt.Sprintf("(select pg_xact_commit_timestamp(id::xid) from transactions where id = %s)", valStr)

	// Build the main query to get transactions committed after the one with a given XID,
	// still visible in the current snapshot, ordered by their commit timestamps.
	builder := w.database.Builder.Select("id").
		From(TransactionsTable).
		Where(fmt.Sprintf("pg_xact_commit_timestamp(id::xid) > (%s)", subquery)).
		Where("id < pg_snapshot_xmin(pg_current_snapshot())").
		Where(squirrel.Eq{"tenant_id": tenantID}).
		OrderBy("pg_xact_commit_timestamp(id::xid)")

	// Convert the builder to a SQL query and arguments.
	query, args, err := builder.ToSql()
	if err != nil {
		slog.ErrorContext(ctx, "error while building sql query", slog.Any("error", err))
		return nil, err
	}

	slog.DebugContext(ctx, "executing SQL query to get recent transaction", slog.Any("query", query), slog.Any("arguments", args))
	// Execute transaction query
	// Execute the SQL query.
	rows, err := w.database.ReadPool.Query(ctx, query, args...)
	if err != nil {
		slog.ErrorContext(ctx, "failed to execute SQL query", slog.Any("error", err))
		return nil, err
	}
	defer rows.Close()

	// Loop through the rows and append XID8 values to the results.
	var xids []db.XID8
	for rows.Next() {
		var xid db.XID8
		err := rows.Scan(&xid)
		if err != nil {
			slog.ErrorContext(ctx, "error while scanning row", slog.Any("error", err))
			return nil, err
		}
		xids = append(xids, xid)
	}

	// Check for errors that could have occurred during iteration.
	err = rows.Err()
	if err != nil {
		slog.ErrorContext(ctx, "failed to iterate over rows", slog.Any("error", err))
		return nil, err
	}

	slog.DebugContext(ctx, "successfully retrieved recent transaction", slog.Any("ids", xids))
	return xids, nil
}

// getChanges is a method that retrieves the changes that occurred in the relation tuples within a specified transaction.
//