feat(wasmguest): guest DB driver bnwasm (WO-WZ-004)
Guest-side database access over the db.* host calls (abiv1 db.proto), two surfaces sharing one injected transport: - database/sql driver registered as "bnwasm" (driver.go) — QueryContext/ ExecContext/BeginTx over db.query/db.exec/db.tx_*; named args rejected. - plugin.Pool (pool.go) handing out a pgx.Tx-shaped value (tx.go), so the pgx-flavored sqlc DBTX every current plugin generates against (sql_package: pgx/v5) is satisfied with no source edits. This is the primary path: their DBTX needs pgconn.CommandTag/pgx.Rows/pgx.Row, which database/sql cannot produce. DbValue↔Go mapping (dbvalue.go) covers all 11 oneof arms both directions; DbError surfaces as *pgconn.PgError (SQLSTATE preserved for errors.As); nested tx/savepoints rejected with a clear error (no fleet plugin uses them). The scan contract is pinned in the exported DbValueFixtures table (dbvalue_fixtures.go) that the WO-WZ-007 host executor mirrors. Tests: driver_test.go (fake host — every DbValue variant round-trips with correct scan types, exec rows-affected, ordered host-call assertions for tx commit/rollback sequences, post-rollback autocommit carries no handle) and sqlcgen_test.go (vendored sqlc-style Queries + WithTx run against the fake host). Native + wasip1 builds green. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Alex Dunmow
parent
bec3a43f55
commit
7881aeee05
278
plugin/wasmguest/bnwasm/dbvalue.go
Normal file
278
plugin/wasmguest/bnwasm/dbvalue.go
Normal file
@ -0,0 +1,278 @@
|
||||
package bnwasm
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"database/sql/driver"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
abiv1 "git.dev.alexdunmow.com/block/core/abi/v1"
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"google.golang.org/protobuf/types/known/timestamppb"
|
||||
)
|
||||
|
||||
// toDbValues marshals a positional argument list to wire DbValues. Named args
|
||||
// (pgx.NamedArgs, or any pgx.QueryRewriter passed as the sole argument) are
|
||||
// rejected: sqlc emits positional parameters, and rewriting a query host-side
|
||||
// is out of scope for v1.
|
||||
func toDbValues(args []any) ([]*abiv1.DbValue, error) {
|
||||
if len(args) == 1 {
|
||||
if _, ok := args[0].(pgx.QueryRewriter); ok {
|
||||
return nil, fmt.Errorf("bnwasm: named arguments (pgx.QueryRewriter/NamedArgs) are not supported; use positional parameters")
|
||||
}
|
||||
}
|
||||
out := make([]*abiv1.DbValue, len(args))
|
||||
for i, a := range args {
|
||||
v, err := toDbValue(a)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("bnwasm: arg %d: %w", i, err)
|
||||
}
|
||||
out[i] = v
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// toDbValue maps one Go query argument to a DbValue. Concrete BlockNinja/pgx
|
||||
// types (uuid, time, text[], json) are special-cased so they round-trip to
|
||||
// their dedicated DbValue variant; anything implementing driver.Valuer (pgtype
|
||||
// scalars, sql.Null*) is normalized through Value(); the rest goes by kind.
|
||||
func toDbValue(a any) (*abiv1.DbValue, error) {
|
||||
switch v := a.(type) {
|
||||
case nil:
|
||||
return nullValue(), nil
|
||||
case uuid.UUID:
|
||||
return &abiv1.DbValue{Kind: &abiv1.DbValue_UuidValue{UuidValue: v.String()}}, nil
|
||||
case uuid.NullUUID:
|
||||
if !v.Valid {
|
||||
return nullValue(), nil
|
||||
}
|
||||
return &abiv1.DbValue{Kind: &abiv1.DbValue_UuidValue{UuidValue: v.UUID.String()}}, nil
|
||||
case time.Time:
|
||||
return &abiv1.DbValue{Kind: &abiv1.DbValue_TimestampValue{TimestampValue: timestamppb.New(v)}}, nil
|
||||
case json.RawMessage:
|
||||
if v == nil {
|
||||
return nullValue(), nil
|
||||
}
|
||||
return &abiv1.DbValue{Kind: &abiv1.DbValue_JsonbValue{JsonbValue: v}}, nil
|
||||
case []byte:
|
||||
if v == nil {
|
||||
return nullValue(), nil
|
||||
}
|
||||
return &abiv1.DbValue{Kind: &abiv1.DbValue_BytesValue{BytesValue: v}}, nil
|
||||
case []string:
|
||||
return &abiv1.DbValue{Kind: &abiv1.DbValue_TextArrayValue{TextArrayValue: &abiv1.TextArray{Values: v}}}, nil
|
||||
case string:
|
||||
return &abiv1.DbValue{Kind: &abiv1.DbValue_StringValue{StringValue: v}}, nil
|
||||
case bool:
|
||||
return &abiv1.DbValue{Kind: &abiv1.DbValue_BoolValue{BoolValue: v}}, nil
|
||||
}
|
||||
|
||||
// driver.Valuer covers pgtype scalars (Timestamptz, Numeric, Int4, ...) and
|
||||
// sql.Null*; recurse on the normalized driver.Value.
|
||||
if valuer, ok := a.(driver.Valuer); ok {
|
||||
dv, err := valuer.Value()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if dv == nil {
|
||||
return nullValue(), nil
|
||||
}
|
||||
return toDbValue(dv)
|
||||
}
|
||||
|
||||
// Fall back to reflection for pointers and the basic numeric kinds.
|
||||
rv := reflect.ValueOf(a)
|
||||
switch rv.Kind() {
|
||||
case reflect.Pointer:
|
||||
if rv.IsNil() {
|
||||
return nullValue(), nil
|
||||
}
|
||||
return toDbValue(rv.Elem().Interface())
|
||||
case reflect.Bool:
|
||||
return &abiv1.DbValue{Kind: &abiv1.DbValue_BoolValue{BoolValue: rv.Bool()}}, nil
|
||||
case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
|
||||
return &abiv1.DbValue{Kind: &abiv1.DbValue_Int64Value{Int64Value: rv.Int()}}, nil
|
||||
case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
|
||||
return &abiv1.DbValue{Kind: &abiv1.DbValue_Int64Value{Int64Value: int64(rv.Uint())}}, nil
|
||||
case reflect.Float32, reflect.Float64:
|
||||
return &abiv1.DbValue{Kind: &abiv1.DbValue_Float64Value{Float64Value: rv.Float()}}, nil
|
||||
case reflect.String:
|
||||
return &abiv1.DbValue{Kind: &abiv1.DbValue_StringValue{StringValue: rv.String()}}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("bnwasm: unsupported argument type %T", a)
|
||||
}
|
||||
|
||||
func nullValue() *abiv1.DbValue {
|
||||
return &abiv1.DbValue{Kind: &abiv1.DbValue_Null{Null: true}}
|
||||
}
|
||||
|
||||
// naturalValue returns the canonical Go scan value for a DbValue and whether it
|
||||
// is SQL NULL. This is the semantic contract WO-WZ-007 (host executor) mirrors:
|
||||
//
|
||||
// null → (nil, true)
|
||||
// bool → (bool, false)
|
||||
// int64 → (int64, false)
|
||||
// float64 → (float64, false)
|
||||
// string → (string, false)
|
||||
// bytes → ([]byte, false)
|
||||
// timestamp → (time.Time,false)
|
||||
// uuid → (string, false) // canonical form; uuid.UUID.Scan accepts it
|
||||
// jsonb → ([]byte, false)
|
||||
// numeric → (string, false) // lossless decimal string
|
||||
// text_array → ([]string, false)
|
||||
func naturalValue(dv *abiv1.DbValue) (any, bool) {
|
||||
switch k := dv.GetKind().(type) {
|
||||
case *abiv1.DbValue_Null:
|
||||
return nil, true
|
||||
case *abiv1.DbValue_BoolValue:
|
||||
return k.BoolValue, false
|
||||
case *abiv1.DbValue_Int64Value:
|
||||
return k.Int64Value, false
|
||||
case *abiv1.DbValue_Float64Value:
|
||||
return k.Float64Value, false
|
||||
case *abiv1.DbValue_StringValue:
|
||||
return k.StringValue, false
|
||||
case *abiv1.DbValue_BytesValue:
|
||||
return k.BytesValue, false
|
||||
case *abiv1.DbValue_TimestampValue:
|
||||
return k.TimestampValue.AsTime(), false
|
||||
case *abiv1.DbValue_UuidValue:
|
||||
return k.UuidValue, false
|
||||
case *abiv1.DbValue_JsonbValue:
|
||||
return []byte(k.JsonbValue), false
|
||||
case *abiv1.DbValue_NumericValue:
|
||||
return k.NumericValue, false
|
||||
case *abiv1.DbValue_TextArrayValue:
|
||||
return append([]string(nil), k.TextArrayValue.GetValues()...), false
|
||||
default:
|
||||
return nil, true
|
||||
}
|
||||
}
|
||||
|
||||
// scanValue assigns a DbValue into a destination pointer with pgx-flavored Scan
|
||||
// semantics: a nil dest skips the column; a sql.Scanner destination is fed the
|
||||
// natural value (including nil for NULL); pointer-to-pointer destinations model
|
||||
// nullability (set nil on NULL); everything else is assigned by reflection with
|
||||
// integer/float/string/[]byte/[]string coercion.
|
||||
func scanValue(dv *abiv1.DbValue, dest any) error {
|
||||
if dest == nil {
|
||||
return nil // pgx: nil dest skips the value entirely
|
||||
}
|
||||
nv, isNull := naturalValue(dv)
|
||||
|
||||
if sc, ok := dest.(sql.Scanner); ok {
|
||||
return sc.Scan(nv)
|
||||
}
|
||||
|
||||
rv := reflect.ValueOf(dest)
|
||||
if rv.Kind() != reflect.Pointer || rv.IsNil() {
|
||||
return fmt.Errorf("bnwasm: scan destination must be a non-nil pointer, got %T", dest)
|
||||
}
|
||||
elem := rv.Elem()
|
||||
|
||||
// Pointer-to-pointer (e.g. **string): nullable column mapped to *string.
|
||||
if elem.Kind() == reflect.Pointer {
|
||||
if isNull {
|
||||
elem.Set(reflect.Zero(elem.Type()))
|
||||
return nil
|
||||
}
|
||||
np := reflect.New(elem.Type().Elem())
|
||||
if err := assign(np.Elem(), nv); err != nil {
|
||||
return err
|
||||
}
|
||||
elem.Set(np)
|
||||
return nil
|
||||
}
|
||||
|
||||
if isNull {
|
||||
elem.Set(reflect.Zero(elem.Type()))
|
||||
return nil
|
||||
}
|
||||
return assign(elem, nv)
|
||||
}
|
||||
|
||||
// assign coerces a natural value into a concrete (non-pointer) destination.
|
||||
func assign(dst reflect.Value, nv any) error {
|
||||
src := reflect.ValueOf(nv)
|
||||
|
||||
// Direct assignability (string→string, time.Time→time.Time, []string→[]string).
|
||||
if src.Type().AssignableTo(dst.Type()) {
|
||||
dst.Set(src)
|
||||
return nil
|
||||
}
|
||||
// Convertibility (json.RawMessage←[]byte, int64→int32, float64→float32,
|
||||
// named string types, []byte→string, ...).
|
||||
if src.Type().ConvertibleTo(dst.Type()) {
|
||||
// Guard against lossy string<->numeric "conversions" reflect allows for
|
||||
// rune/byte-ish types by only converting between compatible kinds.
|
||||
if convertibleKinds(src.Kind(), dst.Kind()) {
|
||||
dst.Set(src.Convert(dst.Type()))
|
||||
return nil
|
||||
}
|
||||
}
|
||||
return fmt.Errorf("bnwasm: cannot scan %s into %s", src.Type(), dst.Type())
|
||||
}
|
||||
|
||||
func convertibleKinds(src, dst reflect.Kind) bool {
|
||||
num := func(k reflect.Kind) bool {
|
||||
switch k {
|
||||
case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64,
|
||||
reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64,
|
||||
reflect.Float32, reflect.Float64:
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
switch {
|
||||
case num(src) && num(dst):
|
||||
return true
|
||||
case src == reflect.String && dst == reflect.String:
|
||||
return true
|
||||
case src == reflect.Slice && dst == reflect.Slice: // []byte→json.RawMessage, []string→named
|
||||
return true
|
||||
case src == reflect.Slice && dst == reflect.String: // []byte→string
|
||||
return true
|
||||
case src == reflect.String && dst == reflect.Slice: // string→[]byte
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// driverValue maps a DbValue to a database/sql driver.Value (the closed set:
|
||||
// nil, int64, float64, bool, []byte, string, time.Time). uuid/numeric surface
|
||||
// as strings and text[] as a Postgres array literal string, since driver.Value
|
||||
// has no richer representation; callers scan those through the column's
|
||||
// sql.Scanner as usual.
|
||||
func driverValue(dv *abiv1.DbValue) driver.Value {
|
||||
nv, isNull := naturalValue(dv)
|
||||
if isNull {
|
||||
return nil
|
||||
}
|
||||
switch v := nv.(type) {
|
||||
case []string:
|
||||
return encodePgTextArray(v)
|
||||
default:
|
||||
return v
|
||||
}
|
||||
}
|
||||
|
||||
// encodePgTextArray renders a []string as a Postgres text[] literal, e.g.
|
||||
// {"a","b,c"}, quoting/escaping every element for lossless reconstruction.
|
||||
func encodePgTextArray(vals []string) string {
|
||||
var b strings.Builder
|
||||
b.WriteByte('{')
|
||||
for i, s := range vals {
|
||||
if i > 0 {
|
||||
b.WriteByte(',')
|
||||
}
|
||||
b.WriteByte('"')
|
||||
b.WriteString(strings.NewReplacer(`\`, `\\`, `"`, `\"`).Replace(s))
|
||||
b.WriteByte('"')
|
||||
}
|
||||
b.WriteByte('}')
|
||||
return b.String()
|
||||
}
|
||||
131
plugin/wasmguest/bnwasm/dbvalue_fixtures.go
Normal file
131
plugin/wasmguest/bnwasm/dbvalue_fixtures.go
Normal file
@ -0,0 +1,131 @@
|
||||
package bnwasm
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
abiv1 "git.dev.alexdunmow.com/block/core/abi/v1"
|
||||
"github.com/google/uuid"
|
||||
"google.golang.org/protobuf/types/known/timestamppb"
|
||||
)
|
||||
|
||||
// FixtureTime is the fixed timestamp used by the shared DbValue fixtures, so
|
||||
// both sides of the boundary compare against one deterministic value.
|
||||
var FixtureTime = time.Date(2026, 7, 3, 12, 30, 0, 0, time.UTC)
|
||||
|
||||
// FixtureUUID is the fixed UUID used by the shared DbValue fixtures.
|
||||
var FixtureUUID = uuid.MustParse("11111111-2222-3333-4444-555555555555")
|
||||
|
||||
// DbValueFixture is one entry in the canonical DbValue↔Go mapping table.
|
||||
type DbValueFixture struct {
|
||||
// Name identifies the DbValue variant.
|
||||
Name string
|
||||
// Value is the wire value the host would send back in a DbRowsResponse cell
|
||||
// (and the guest would send as an argument for the non-NULL scalar cases).
|
||||
Value *abiv1.DbValue
|
||||
// NewDest returns a fresh pointer to the canonical Go scan destination type
|
||||
// that plugin sqlc code binds for this variant.
|
||||
NewDest func() any
|
||||
// Want is the expected dereferenced value after scanning Value into
|
||||
// NewDest(). Compared with reflect.DeepEqual (times via .Equal).
|
||||
Want any
|
||||
// DriverValue is the expected database/sql driver.Value for this variant,
|
||||
// documenting the "bnwasm" driver's mapping (uuid/numeric→string,
|
||||
// text[]→Postgres array literal).
|
||||
DriverValue any
|
||||
}
|
||||
|
||||
// DbValueFixtures is the authoritative DbValue↔Go mapping table. It is the
|
||||
// shared contract WO-WZ-007 (the CMS host executor) MUST mirror: every variant
|
||||
// the guest driver produces on scan, the host must be able to build on read,
|
||||
// and vice-versa. Exported (non-test) precisely so the host's tests in the CMS
|
||||
// module can import and assert against the same table rather than duplicating a
|
||||
// drift-prone copy.
|
||||
//
|
||||
// Every oneof arm of abiv1.DbValue.Kind appears exactly once.
|
||||
var DbValueFixtures = []DbValueFixture{
|
||||
{
|
||||
Name: "null",
|
||||
Value: &abiv1.DbValue{Kind: &abiv1.DbValue_Null{Null: true}},
|
||||
NewDest: func() any { return new(*string) }, // **string: NULL → (*string)(nil)
|
||||
Want: (*string)(nil),
|
||||
DriverValue: nil,
|
||||
},
|
||||
{
|
||||
Name: "bool",
|
||||
Value: &abiv1.DbValue{Kind: &abiv1.DbValue_BoolValue{BoolValue: true}},
|
||||
NewDest: func() any { return new(bool) },
|
||||
Want: true,
|
||||
DriverValue: true,
|
||||
},
|
||||
{
|
||||
Name: "int64",
|
||||
Value: &abiv1.DbValue{Kind: &abiv1.DbValue_Int64Value{Int64Value: 42}},
|
||||
NewDest: func() any { return new(int64) },
|
||||
Want: int64(42),
|
||||
DriverValue: int64(42),
|
||||
},
|
||||
{
|
||||
Name: "int32", // narrowing coercion sqlc emits for int4 columns
|
||||
Value: &abiv1.DbValue{Kind: &abiv1.DbValue_Int64Value{Int64Value: 7}},
|
||||
NewDest: func() any { return new(int32) },
|
||||
Want: int32(7),
|
||||
DriverValue: int64(7),
|
||||
},
|
||||
{
|
||||
Name: "float64",
|
||||
Value: &abiv1.DbValue{Kind: &abiv1.DbValue_Float64Value{Float64Value: 3.5}},
|
||||
NewDest: func() any { return new(float64) },
|
||||
Want: 3.5,
|
||||
DriverValue: 3.5,
|
||||
},
|
||||
{
|
||||
Name: "string",
|
||||
Value: &abiv1.DbValue{Kind: &abiv1.DbValue_StringValue{StringValue: "hello"}},
|
||||
NewDest: func() any { return new(string) },
|
||||
Want: "hello",
|
||||
DriverValue: "hello",
|
||||
},
|
||||
{
|
||||
Name: "bytes",
|
||||
Value: &abiv1.DbValue{Kind: &abiv1.DbValue_BytesValue{BytesValue: []byte{0x01, 0x02, 0x03}}},
|
||||
NewDest: func() any { return new([]byte) },
|
||||
Want: []byte{0x01, 0x02, 0x03},
|
||||
DriverValue: []byte{0x01, 0x02, 0x03},
|
||||
},
|
||||
{
|
||||
Name: "timestamp",
|
||||
Value: &abiv1.DbValue{Kind: &abiv1.DbValue_TimestampValue{TimestampValue: timestamppb.New(FixtureTime)}},
|
||||
NewDest: func() any { return new(time.Time) },
|
||||
Want: FixtureTime,
|
||||
DriverValue: FixtureTime,
|
||||
},
|
||||
{
|
||||
Name: "uuid",
|
||||
Value: &abiv1.DbValue{Kind: &abiv1.DbValue_UuidValue{UuidValue: FixtureUUID.String()}},
|
||||
NewDest: func() any { return new(uuid.UUID) },
|
||||
Want: FixtureUUID,
|
||||
DriverValue: FixtureUUID.String(),
|
||||
},
|
||||
{
|
||||
Name: "jsonb",
|
||||
Value: &abiv1.DbValue{Kind: &abiv1.DbValue_JsonbValue{JsonbValue: []byte(`{"a":1}`)}},
|
||||
NewDest: func() any { return new(json.RawMessage) },
|
||||
Want: json.RawMessage(`{"a":1}`),
|
||||
DriverValue: []byte(`{"a":1}`),
|
||||
},
|
||||
{
|
||||
Name: "numeric",
|
||||
Value: &abiv1.DbValue{Kind: &abiv1.DbValue_NumericValue{NumericValue: "123.45"}},
|
||||
NewDest: func() any { return new(string) },
|
||||
Want: "123.45",
|
||||
DriverValue: "123.45",
|
||||
},
|
||||
{
|
||||
Name: "text_array",
|
||||
Value: &abiv1.DbValue{Kind: &abiv1.DbValue_TextArrayValue{TextArrayValue: &abiv1.TextArray{Values: []string{"a", "b"}}}},
|
||||
NewDest: func() any { return new([]string) },
|
||||
Want: []string{"a", "b"},
|
||||
DriverValue: `{"a","b"}`,
|
||||
},
|
||||
}
|
||||
201
plugin/wasmguest/bnwasm/driver.go
Normal file
201
plugin/wasmguest/bnwasm/driver.go
Normal file
@ -0,0 +1,201 @@
|
||||
package bnwasm
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"database/sql/driver"
|
||||
"fmt"
|
||||
"io"
|
||||
"sync"
|
||||
|
||||
abiv1 "git.dev.alexdunmow.com/block/core/abi/v1"
|
||||
)
|
||||
|
||||
// DriverName is the database/sql driver name registered by this package.
|
||||
const DriverName = "bnwasm"
|
||||
|
||||
func init() { sql.Register(DriverName, Driver{}) }
|
||||
|
||||
// defaultTransport is the process-wide transport the "bnwasm" database/sql
|
||||
// driver uses (the guest has exactly one host). The wasip1 shim sets it via
|
||||
// SetDefaultTransport; native builds leave it nil so opened connections fail
|
||||
// cleanly with errNoHost. Guarded because sql.DB may dial from any goroutine.
|
||||
var (
|
||||
defaultMu sync.RWMutex
|
||||
defaultTransport Transport
|
||||
)
|
||||
|
||||
// SetDefaultTransport binds the transport used by database/sql connections
|
||||
// opened via sql.Open("bnwasm", ...). Called once from the wasm guest shim.
|
||||
func SetDefaultTransport(t Transport) {
|
||||
defaultMu.Lock()
|
||||
defaultTransport = t
|
||||
defaultMu.Unlock()
|
||||
}
|
||||
|
||||
func currentTransport() Transport {
|
||||
defaultMu.RLock()
|
||||
defer defaultMu.RUnlock()
|
||||
return defaultTransport
|
||||
}
|
||||
|
||||
// Driver is the database/sql/driver.Driver for the wasm DB ABI.
|
||||
type Driver struct{}
|
||||
|
||||
var (
|
||||
_ driver.Driver = Driver{}
|
||||
_ driver.DriverContext = Driver{}
|
||||
)
|
||||
|
||||
// Open ignores its DSN — the transport is process-global (one host per guest).
|
||||
func (d Driver) Open(name string) (driver.Conn, error) {
|
||||
return &conn{t: currentTransport()}, nil
|
||||
}
|
||||
|
||||
// OpenConnector lets sql.OpenDB bypass DSN parsing entirely.
|
||||
func (d Driver) OpenConnector(name string) (driver.Connector, error) {
|
||||
return connector{}, nil
|
||||
}
|
||||
|
||||
// NewConnector returns a driver.Connector bound to an explicit transport, for
|
||||
// callers that construct a *sql.DB via sql.OpenDB without touching the global.
|
||||
func NewConnector(t Transport) driver.Connector { return connector{t: t, explicit: true} }
|
||||
|
||||
type connector struct {
|
||||
t Transport
|
||||
explicit bool
|
||||
}
|
||||
|
||||
func (c connector) Connect(context.Context) (driver.Conn, error) {
|
||||
t := c.t
|
||||
if !c.explicit {
|
||||
t = currentTransport()
|
||||
}
|
||||
return &conn{t: t}, nil
|
||||
}
|
||||
|
||||
func (c connector) Driver() driver.Driver { return Driver{} }
|
||||
|
||||
// conn is one logical connection: it holds no real connection state guest-side,
|
||||
// only the transport and the current transaction handle (0 = autocommit).
|
||||
type conn struct {
|
||||
t Transport
|
||||
txHandle uint64
|
||||
inTx bool
|
||||
}
|
||||
|
||||
var (
|
||||
_ driver.Conn = (*conn)(nil)
|
||||
_ driver.QueryerContext = (*conn)(nil)
|
||||
_ driver.ExecerContext = (*conn)(nil)
|
||||
_ driver.ConnBeginTx = (*conn)(nil)
|
||||
)
|
||||
|
||||
func (c *conn) Prepare(query string) (driver.Stmt, error) {
|
||||
return nil, fmt.Errorf("bnwasm: prepared statements are not supported; use QueryContext/ExecContext")
|
||||
}
|
||||
|
||||
func (c *conn) Close() error { return nil }
|
||||
|
||||
func (c *conn) Begin() (driver.Tx, error) { return c.BeginTx(context.Background(), driver.TxOptions{}) }
|
||||
|
||||
func (c *conn) BeginTx(ctx context.Context, _ driver.TxOptions) (driver.Tx, error) {
|
||||
if c.inTx {
|
||||
return nil, errNestedTx
|
||||
}
|
||||
handle, err := c.t.txBegin(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
c.txHandle = handle
|
||||
c.inTx = true
|
||||
return &sqlTx{c: c}, nil
|
||||
}
|
||||
|
||||
func (c *conn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
|
||||
posArgs, err := namedToPositional(args)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
tag, err := c.t.exec(ctx, query, c.txHandle, posArgs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return sqlResult{rowsAffected: tag.RowsAffected()}, nil
|
||||
}
|
||||
|
||||
func (c *conn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
|
||||
posArgs, err := namedToPositional(args)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resp, err := c.t.query(ctx, query, c.txHandle, posArgs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return newDriverRows(resp), nil
|
||||
}
|
||||
|
||||
// sqlTx is the database/sql transaction handle; it clears the conn's handle on
|
||||
// completion so subsequent statements run in autocommit again.
|
||||
type sqlTx struct{ c *conn }
|
||||
|
||||
func (t sqlTx) Commit() error {
|
||||
err := t.c.t.txCommit(context.Background(), t.c.txHandle)
|
||||
t.c.txHandle, t.c.inTx = 0, false
|
||||
return err
|
||||
}
|
||||
|
||||
func (t sqlTx) Rollback() error {
|
||||
err := t.c.t.txRollback(context.Background(), t.c.txHandle)
|
||||
t.c.txHandle, t.c.inTx = 0, false
|
||||
return err
|
||||
}
|
||||
|
||||
type sqlResult struct{ rowsAffected int64 }
|
||||
|
||||
func (r sqlResult) LastInsertId() (int64, error) {
|
||||
return 0, fmt.Errorf("bnwasm: LastInsertId is not supported (Postgres uses RETURNING)")
|
||||
}
|
||||
func (r sqlResult) RowsAffected() (int64, error) { return r.rowsAffected, nil }
|
||||
|
||||
// driverRows adapts a buffered DbRowsResponse to database/sql/driver.Rows.
|
||||
type driverRows struct {
|
||||
resp *abiv1.DbRowsResponse
|
||||
idx int
|
||||
}
|
||||
|
||||
func newDriverRows(resp *abiv1.DbRowsResponse) *driverRows { return &driverRows{resp: resp, idx: 0} }
|
||||
|
||||
func (r *driverRows) Columns() []string { return r.resp.GetColumns() }
|
||||
|
||||
func (r *driverRows) Close() error { return nil }
|
||||
|
||||
func (r *driverRows) Next(dest []driver.Value) error {
|
||||
rows := r.resp.GetRows()
|
||||
if r.idx >= len(rows) {
|
||||
return io.EOF
|
||||
}
|
||||
cells := rows[r.idx].GetValues()
|
||||
if len(dest) != len(cells) {
|
||||
return fmt.Errorf("bnwasm: expected %d columns, got %d destinations", len(cells), len(dest))
|
||||
}
|
||||
for i, cell := range cells {
|
||||
dest[i] = driverValue(cell)
|
||||
}
|
||||
r.idx++
|
||||
return nil
|
||||
}
|
||||
|
||||
// namedToPositional flattens database/sql NamedValues to positional args,
|
||||
// rejecting any that carry a name (sqlc emits positional $N parameters only).
|
||||
func namedToPositional(args []driver.NamedValue) ([]any, error) {
|
||||
out := make([]any, len(args))
|
||||
for i, a := range args {
|
||||
if a.Name != "" {
|
||||
return nil, fmt.Errorf("bnwasm: named argument %q is not supported; use positional parameters", a.Name)
|
||||
}
|
||||
out[i] = a.Value
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
304
plugin/wasmguest/bnwasm/driver_test.go
Normal file
304
plugin/wasmguest/bnwasm/driver_test.go
Normal file
@ -0,0 +1,304 @@
|
||||
package bnwasm
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
abiv1 "git.dev.alexdunmow.com/block/core/abi/v1"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"google.golang.org/protobuf/proto"
|
||||
)
|
||||
|
||||
// recordedCall captures one db.* host call for ordered-sequence assertions.
|
||||
type recordedCall struct {
|
||||
method string
|
||||
sql string
|
||||
args []*abiv1.DbValue
|
||||
txHandle uint64
|
||||
}
|
||||
|
||||
// fakeHost is a scripted db.* transport: it records every call and answers from
|
||||
// programmable fields, so the guest driver is exercised end-to-end with no real
|
||||
// Postgres and the exact host-call sequence can be asserted.
|
||||
type fakeHost struct {
|
||||
calls []recordedCall
|
||||
|
||||
queryResp *abiv1.DbRowsResponse // returned for db.query
|
||||
execRows int64 // rows_affected for db.exec
|
||||
txHandle uint64 // handle handed out on db.tx_begin
|
||||
dbError *abiv1.DbError // if set, attached to the next query/exec response
|
||||
failWith error // if set, the transport returns this transport-level error
|
||||
}
|
||||
|
||||
func (f *fakeHost) call(method string, req, resp proto.Message) error {
|
||||
rc := recordedCall{method: method}
|
||||
switch r := req.(type) {
|
||||
case *abiv1.DbQueryRequest:
|
||||
rc.sql, rc.args, rc.txHandle = r.GetSql(), r.GetArgs(), r.GetTxHandle()
|
||||
case *abiv1.DbExecRequest:
|
||||
rc.sql, rc.args, rc.txHandle = r.GetSql(), r.GetArgs(), r.GetTxHandle()
|
||||
case *abiv1.DbTxCommitRequest:
|
||||
rc.txHandle = r.GetTxHandle()
|
||||
case *abiv1.DbTxRollbackRequest:
|
||||
rc.txHandle = r.GetTxHandle()
|
||||
}
|
||||
f.calls = append(f.calls, rc)
|
||||
|
||||
if f.failWith != nil {
|
||||
return f.failWith
|
||||
}
|
||||
switch out := resp.(type) {
|
||||
case *abiv1.DbRowsResponse:
|
||||
if f.queryResp != nil {
|
||||
proto.Merge(out, f.queryResp)
|
||||
}
|
||||
out.Error = f.dbError
|
||||
case *abiv1.DbExecResponse:
|
||||
out.RowsAffected = f.execRows
|
||||
out.Error = f.dbError
|
||||
case *abiv1.DbTxBeginResponse:
|
||||
out.TxHandle = f.txHandle
|
||||
out.Error = f.dbError
|
||||
case *abiv1.DbTxCommitResponse:
|
||||
out.Error = f.dbError
|
||||
case *abiv1.DbTxRollbackResponse:
|
||||
out.Error = f.dbError
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeHost) methods() []string {
|
||||
ms := make([]string, len(f.calls))
|
||||
for i, c := range f.calls {
|
||||
ms[i] = c.method
|
||||
}
|
||||
return ms
|
||||
}
|
||||
|
||||
func rowsWith(columns []string, cells ...*abiv1.DbValue) *abiv1.DbRowsResponse {
|
||||
return &abiv1.DbRowsResponse{Columns: columns, Rows: []*abiv1.DbRow{{Values: cells}}}
|
||||
}
|
||||
|
||||
// deref returns the value a NewDest()-style pointer points at.
|
||||
func deref(ptr any) any { return reflect.ValueOf(ptr).Elem().Interface() }
|
||||
|
||||
func equalScan(want, got any) bool {
|
||||
if w, ok := want.(time.Time); ok {
|
||||
g, ok := got.(time.Time)
|
||||
return ok && w.Equal(g)
|
||||
}
|
||||
return reflect.DeepEqual(want, got)
|
||||
}
|
||||
|
||||
// TestDbValueScanRoundTrip proves every DbValue variant round-trips through a
|
||||
// SELECT and scans into its canonical Go type — the core contract WO-WZ-007
|
||||
// mirrors. Driven by the shared DbValueFixtures table.
|
||||
func TestDbValueScanRoundTrip(t *testing.T) {
|
||||
for _, fx := range DbValueFixtures {
|
||||
t.Run(fx.Name, func(t *testing.T) {
|
||||
host := &fakeHost{queryResp: rowsWith([]string{fx.Name}, fx.Value)}
|
||||
pool := NewPool(host.call)
|
||||
|
||||
dest := fx.NewDest()
|
||||
if err := pool.QueryRow(context.Background(), "SELECT x").Scan(dest); err != nil {
|
||||
t.Fatalf("scan %s: %v", fx.Name, err)
|
||||
}
|
||||
if got := deref(dest); !equalScan(fx.Want, got) {
|
||||
t.Fatalf("scan %s: want %#v, got %#v", fx.Name, fx.Want, got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestDbValueDriverValue proves the database/sql "bnwasm" driver maps every
|
||||
// DbValue variant to the documented driver.Value.
|
||||
func TestDbValueDriverValue(t *testing.T) {
|
||||
for _, fx := range DbValueFixtures {
|
||||
t.Run(fx.Name, func(t *testing.T) {
|
||||
if got := driverValue(fx.Value); !equalScan(fx.DriverValue, got) {
|
||||
t.Fatalf("driverValue %s: want %#v, got %#v", fx.Name, fx.DriverValue, got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecRowsAffected(t *testing.T) {
|
||||
host := &fakeHost{execRows: 5}
|
||||
pool := NewPool(host.call)
|
||||
|
||||
tag, err := pool.Exec(context.Background(), "UPDATE t SET x = 1")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if tag.RowsAffected() != 5 {
|
||||
t.Fatalf("rows affected: want 5, got %d", tag.RowsAffected())
|
||||
}
|
||||
if host.calls[0].txHandle != 0 {
|
||||
t.Fatalf("autocommit exec should carry tx_handle 0, got %d", host.calls[0].txHandle)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTxCommitSequence asserts the ordered host calls for a begin/exec/commit
|
||||
// transaction, and that every statement inside carries the tx handle.
|
||||
func TestTxCommitSequence(t *testing.T) {
|
||||
host := &fakeHost{txHandle: 77, execRows: 1}
|
||||
pool := NewPool(host.call)
|
||||
ctx := context.Background()
|
||||
|
||||
tx, err := pool.Begin(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := tx.Exec(ctx, "INSERT INTO t VALUES ($1)", 1); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
wantMethods := []string{methodTxBegin, methodExec, methodTxCommit}
|
||||
if got := host.methods(); !reflect.DeepEqual(got, wantMethods) {
|
||||
t.Fatalf("method sequence: want %v, got %v", wantMethods, got)
|
||||
}
|
||||
if host.calls[1].txHandle != 77 {
|
||||
t.Fatalf("in-tx exec should carry handle 77, got %d", host.calls[1].txHandle)
|
||||
}
|
||||
if host.calls[2].txHandle != 77 {
|
||||
t.Fatalf("commit should carry handle 77, got %d", host.calls[2].txHandle)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTxRollbackThenAutocommit asserts a rollback sequence, and that a query
|
||||
// issued afterward (on the pool) carries NO tx handle.
|
||||
func TestTxRollbackThenAutocommit(t *testing.T) {
|
||||
host := &fakeHost{txHandle: 9, execRows: 1}
|
||||
pool := NewPool(host.call)
|
||||
ctx := context.Background()
|
||||
|
||||
tx, err := pool.Begin(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := tx.Rollback(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, "SELECT 1"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
wantMethods := []string{methodTxBegin, methodTxRollback, methodExec}
|
||||
if got := host.methods(); !reflect.DeepEqual(got, wantMethods) {
|
||||
t.Fatalf("method sequence: want %v, got %v", wantMethods, got)
|
||||
}
|
||||
if h := host.calls[1].txHandle; h != 9 {
|
||||
t.Fatalf("rollback should carry handle 9, got %d", h)
|
||||
}
|
||||
if h := host.calls[2].txHandle; h != 0 {
|
||||
t.Fatalf("post-rollback exec must carry tx_handle 0, got %d", h)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTxUseAfterCloseRejected(t *testing.T) {
|
||||
host := &fakeHost{txHandle: 3}
|
||||
tx, err := NewPool(host.call).Begin(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := tx.Commit(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := tx.Exec(context.Background(), "SELECT 1"); !errors.Is(err, pgx.ErrTxClosed) {
|
||||
t.Fatalf("exec after commit: want ErrTxClosed, got %v", err)
|
||||
}
|
||||
if err := tx.Rollback(context.Background()); !errors.Is(err, pgx.ErrTxClosed) {
|
||||
t.Fatalf("rollback after commit: want ErrTxClosed, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNestedTxRejected(t *testing.T) {
|
||||
host := &fakeHost{txHandle: 1}
|
||||
tx, err := NewPool(host.call).Begin(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := tx.Begin(context.Background()); !errors.Is(err, errNestedTx) {
|
||||
t.Fatalf("nested Begin: want errNestedTx, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNamedArgsRejected(t *testing.T) {
|
||||
host := &fakeHost{}
|
||||
pool := NewPool(host.call)
|
||||
_, err := pool.Query(context.Background(), "SELECT $1", pgx.NamedArgs{"a": 1})
|
||||
if err == nil {
|
||||
t.Fatal("expected named-args rejection, got nil")
|
||||
}
|
||||
}
|
||||
|
||||
// TestDbErrorSurfacesAsPgError proves a DbError carrying a SQLSTATE reaches the
|
||||
// plugin as a *pgconn.PgError, so errors.As-based constraint checks keep working.
|
||||
func TestDbErrorSurfacesAsPgError(t *testing.T) {
|
||||
host := &fakeHost{dbError: &abiv1.DbError{Code: "23505", Message: "duplicate key"}}
|
||||
_, err := NewPool(host.call).Exec(context.Background(), "INSERT INTO t VALUES (1)")
|
||||
var pgErr *pgconn.PgError
|
||||
if !errors.As(err, &pgErr) {
|
||||
t.Fatalf("want *pgconn.PgError, got %T (%v)", err, err)
|
||||
}
|
||||
if pgErr.Code != "23505" {
|
||||
t.Fatalf("want SQLSTATE 23505, got %q", pgErr.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNoHostFailsCleanly(t *testing.T) {
|
||||
pool := NewPool(nil)
|
||||
if _, err := pool.Exec(context.Background(), "SELECT 1"); !errors.Is(err, errNoHost) {
|
||||
t.Fatalf("want errNoHost, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQueryRowNoRows(t *testing.T) {
|
||||
host := &fakeHost{queryResp: &abiv1.DbRowsResponse{Columns: []string{"x"}}}
|
||||
var x int
|
||||
err := NewPool(host.call).QueryRow(context.Background(), "SELECT x").Scan(&x)
|
||||
if !errors.Is(err, pgx.ErrNoRows) {
|
||||
t.Fatalf("want ErrNoRows, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDatabaseSQLDriverRegistered proves the "bnwasm" driver is registered and
|
||||
// drives a SELECT end-to-end through database/sql against the fake host.
|
||||
func TestDatabaseSQLDriverRegistered(t *testing.T) {
|
||||
host := &fakeHost{queryResp: rowsWith(
|
||||
[]string{"id", "name"},
|
||||
&abiv1.DbValue{Kind: &abiv1.DbValue_Int64Value{Int64Value: 1}},
|
||||
&abiv1.DbValue{Kind: &abiv1.DbValue_StringValue{StringValue: "alice"}},
|
||||
)}
|
||||
db := sql.OpenDB(NewConnector(host.call))
|
||||
defer db.Close()
|
||||
|
||||
var id int
|
||||
var name string
|
||||
if err := db.QueryRowContext(context.Background(), "SELECT id, name FROM t").Scan(&id, &name); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if id != 1 || name != "alice" {
|
||||
t.Fatalf("got id=%d name=%q", id, name)
|
||||
}
|
||||
if !slicesContains(sql.Drivers(), DriverName) {
|
||||
t.Fatalf("driver %q not registered; have %v", DriverName, sql.Drivers())
|
||||
}
|
||||
}
|
||||
|
||||
func slicesContains(s []string, v string) bool {
|
||||
for _, x := range s {
|
||||
if x == v {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
60
plugin/wasmguest/bnwasm/pool.go
Normal file
60
plugin/wasmguest/bnwasm/pool.go
Normal file
@ -0,0 +1,60 @@
|
||||
package bnwasm
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
)
|
||||
|
||||
// Pool implements plugin.Pool (the pgx-flavored interface plugins bind their
|
||||
// sqlc DBTX to) over the db.* host calls. It is also itself a valid DBTX: its
|
||||
// Exec/Query/QueryRow run in autocommit (tx_handle 0), so a plugin can pass the
|
||||
// Pool directly to sqlc's New(db) for non-transactional queries, exactly as it
|
||||
// passes a *pgxpool.Pool today.
|
||||
type Pool struct {
|
||||
t Transport
|
||||
}
|
||||
|
||||
// NewPool returns a Pool bound to transport t. A nil t (native build / DESCRIBE
|
||||
// probe) yields a Pool whose every operation fails cleanly with errNoHost.
|
||||
func NewPool(t Transport) *Pool { return &Pool{t: t} }
|
||||
|
||||
// Begin opens a host-side transaction and returns a pgx.Tx bound to its handle.
|
||||
func (p *Pool) Begin(ctx context.Context) (pgx.Tx, error) {
|
||||
handle, err := p.t.txBegin(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &Tx{t: p.t, handle: handle}, nil
|
||||
}
|
||||
|
||||
// Exec runs a statement in autocommit and returns its command tag.
|
||||
func (p *Pool) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) {
|
||||
return p.t.exec(ctx, sql, 0, args)
|
||||
}
|
||||
|
||||
// Query runs a rows-returning statement in autocommit.
|
||||
func (p *Pool) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) {
|
||||
resp, err := p.t.query(ctx, sql, 0, args)
|
||||
if err != nil {
|
||||
return errRows(err), err
|
||||
}
|
||||
return newPgxRows(resp), nil
|
||||
}
|
||||
|
||||
// QueryRow runs a rows-returning statement in autocommit and returns the first
|
||||
// row; any error is deferred to Row.Scan, matching pgx.
|
||||
func (p *Pool) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row {
|
||||
resp, err := p.t.query(ctx, sql, 0, args)
|
||||
if err != nil {
|
||||
return &pgxRow{err: err}
|
||||
}
|
||||
return &pgxRow{rows: newPgxRows(resp)}
|
||||
}
|
||||
|
||||
// errRows is a closed, empty pgx.Rows carrying err, so a Query caller that
|
||||
// ignores the returned error and iterates still terminates and reports it.
|
||||
func errRows(err error) *pgxRows {
|
||||
return &pgxRows{idx: -1, err: err, closed: true}
|
||||
}
|
||||
120
plugin/wasmguest/bnwasm/rows.go
Normal file
120
plugin/wasmguest/bnwasm/rows.go
Normal file
@ -0,0 +1,120 @@
|
||||
package bnwasm
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
abiv1 "git.dev.alexdunmow.com/block/core/abi/v1"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
)
|
||||
|
||||
// pgxRows adapts a buffered DbRowsResponse to the pgx.Rows interface. The full
|
||||
// result set is already materialized (the host returns it in one message), so
|
||||
// iteration is a slice cursor. Only the methods sqlc-generated code exercises
|
||||
// carry real behavior; the rest satisfy the interface.
|
||||
type pgxRows struct {
|
||||
columns []string
|
||||
rows []*abiv1.DbRow
|
||||
idx int // index of the row Next() last advanced to; -1 before first Next
|
||||
err error
|
||||
closed bool
|
||||
}
|
||||
|
||||
func newPgxRows(resp *abiv1.DbRowsResponse) *pgxRows {
|
||||
return &pgxRows{columns: resp.GetColumns(), rows: resp.GetRows(), idx: -1}
|
||||
}
|
||||
|
||||
func (r *pgxRows) Close() { r.closed = true }
|
||||
|
||||
func (r *pgxRows) Err() error { return r.err }
|
||||
|
||||
func (r *pgxRows) CommandTag() pgconn.CommandTag {
|
||||
return pgconn.NewCommandTag(fmt.Sprintf("SELECT %d", len(r.rows)))
|
||||
}
|
||||
|
||||
func (r *pgxRows) FieldDescriptions() []pgconn.FieldDescription {
|
||||
fds := make([]pgconn.FieldDescription, len(r.columns))
|
||||
for i, c := range r.columns {
|
||||
fds[i] = pgconn.FieldDescription{Name: c}
|
||||
}
|
||||
return fds
|
||||
}
|
||||
|
||||
func (r *pgxRows) Next() bool {
|
||||
if r.err != nil || r.closed {
|
||||
return false
|
||||
}
|
||||
if r.idx+1 >= len(r.rows) {
|
||||
r.Close() // pgx auto-closes when iteration is exhausted
|
||||
return false
|
||||
}
|
||||
r.idx++
|
||||
return true
|
||||
}
|
||||
|
||||
func (r *pgxRows) Scan(dest ...any) error {
|
||||
if r.err != nil {
|
||||
return r.err
|
||||
}
|
||||
if r.idx < 0 || r.idx >= len(r.rows) {
|
||||
return errors.New("bnwasm: Scan called without a current row (call Next first)")
|
||||
}
|
||||
if err := scanRow(r.rows[r.idx], dest); err != nil {
|
||||
r.err = err
|
||||
r.Close()
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *pgxRows) Values() ([]any, error) {
|
||||
if r.idx < 0 || r.idx >= len(r.rows) {
|
||||
return nil, errors.New("bnwasm: Values called without a current row")
|
||||
}
|
||||
cells := r.rows[r.idx].GetValues()
|
||||
out := make([]any, len(cells))
|
||||
for i, c := range cells {
|
||||
v, _ := naturalValue(c)
|
||||
out[i] = v
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (r *pgxRows) RawValues() [][]byte { return nil }
|
||||
|
||||
func (r *pgxRows) Conn() *pgx.Conn { return nil }
|
||||
|
||||
// pgxRow is the single-row QueryRow result. Like pgx, it defers all errors
|
||||
// (including the query error) to Scan, and reports pgx.ErrNoRows when empty.
|
||||
type pgxRow struct {
|
||||
rows *pgxRows
|
||||
err error
|
||||
}
|
||||
|
||||
func (r *pgxRow) Scan(dest ...any) error {
|
||||
if r.err != nil {
|
||||
return r.err
|
||||
}
|
||||
defer r.rows.Close()
|
||||
if !r.rows.Next() {
|
||||
if r.rows.Err() != nil {
|
||||
return r.rows.Err()
|
||||
}
|
||||
return pgx.ErrNoRows
|
||||
}
|
||||
return r.rows.Scan(dest...)
|
||||
}
|
||||
|
||||
func scanRow(row *abiv1.DbRow, dest []any) error {
|
||||
cells := row.GetValues()
|
||||
if len(dest) != len(cells) {
|
||||
return fmt.Errorf("bnwasm: scan expected %d destinations, got %d columns", len(dest), len(cells))
|
||||
}
|
||||
for i, d := range dest {
|
||||
if err := scanValue(cells[i], d); err != nil {
|
||||
return fmt.Errorf("bnwasm: scan column %d: %w", i, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
207
plugin/wasmguest/bnwasm/sqlcgen_test.go
Normal file
207
plugin/wasmguest/bnwasm/sqlcgen_test.go
Normal file
@ -0,0 +1,207 @@
|
||||
package bnwasm
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
abiv1 "git.dev.alexdunmow.com/block/core/abi/v1"
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
)
|
||||
|
||||
// --- Vendored sqlc output ---------------------------------------------------
|
||||
//
|
||||
// The block below is byte-for-byte in the shape `sqlc generate` emits for the
|
||||
// pgx/v5 sql_package with the same overrides symposium uses (uuid→uuid.UUID,
|
||||
// jsonb→json.RawMessage, emit_pointers_for_null_types). It is checked in so a
|
||||
// compile of THIS test is proof that generated plugin code binds to the bnwasm
|
||||
// Pool/Tx (as its DBTX) and pgxRows/pgxRow (via Scan) with no edits — the same
|
||||
// DBTX interface, the same pgconn.CommandTag / pgx.Rows / pgx.Row signatures.
|
||||
|
||||
type DBTX interface {
|
||||
Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error)
|
||||
Query(context.Context, string, ...interface{}) (pgx.Rows, error)
|
||||
QueryRow(context.Context, string, ...interface{}) pgx.Row
|
||||
}
|
||||
|
||||
func New(db DBTX) *Queries { return &Queries{db: db} }
|
||||
|
||||
type Queries struct{ db DBTX }
|
||||
|
||||
func (q *Queries) WithTx(tx pgx.Tx) *Queries { return &Queries{db: tx} }
|
||||
|
||||
type Widget struct {
|
||||
ID uuid.UUID `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Notes *string `json:"notes"`
|
||||
Count int32 `json:"count"`
|
||||
CreatedAt pgtype.Timestamptz `json:"created_at"`
|
||||
}
|
||||
|
||||
const createWidget = `-- name: CreateWidget :one
|
||||
INSERT INTO widgets (name, notes) VALUES ($1, $2) RETURNING id
|
||||
`
|
||||
|
||||
type CreateWidgetParams struct {
|
||||
Name string `json:"name"`
|
||||
Notes *string `json:"notes"`
|
||||
}
|
||||
|
||||
func (q *Queries) CreateWidget(ctx context.Context, arg CreateWidgetParams) (uuid.UUID, error) {
|
||||
row := q.db.QueryRow(ctx, createWidget, arg.Name, arg.Notes)
|
||||
var id uuid.UUID
|
||||
err := row.Scan(&id)
|
||||
return id, err
|
||||
}
|
||||
|
||||
const getWidget = `-- name: GetWidget :one
|
||||
SELECT id, name, notes, count, created_at FROM widgets WHERE id = $1
|
||||
`
|
||||
|
||||
func (q *Queries) GetWidget(ctx context.Context, id uuid.UUID) (Widget, error) {
|
||||
row := q.db.QueryRow(ctx, getWidget, id)
|
||||
var i Widget
|
||||
err := row.Scan(&i.ID, &i.Name, &i.Notes, &i.Count, &i.CreatedAt)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const listWidgets = `-- name: ListWidgets :many
|
||||
SELECT id, name, notes, count, created_at FROM widgets ORDER BY name
|
||||
`
|
||||
|
||||
func (q *Queries) ListWidgets(ctx context.Context) ([]Widget, error) {
|
||||
rows, err := q.db.Query(ctx, listWidgets)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var items []Widget
|
||||
for rows.Next() {
|
||||
var i Widget
|
||||
if err := rows.Scan(&i.ID, &i.Name, &i.Notes, &i.Count, &i.CreatedAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, i)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const deleteWidget = `-- name: DeleteWidget :exec
|
||||
DELETE FROM widgets WHERE id = $1
|
||||
`
|
||||
|
||||
func (q *Queries) DeleteWidget(ctx context.Context, id uuid.UUID) error {
|
||||
_, err := q.db.Exec(ctx, deleteWidget, id)
|
||||
return err
|
||||
}
|
||||
|
||||
// --- Compile-time proof the bnwasm surfaces satisfy the generated interfaces --
|
||||
|
||||
var (
|
||||
_ DBTX = (*Pool)(nil)
|
||||
_ DBTX = (*Tx)(nil)
|
||||
_ pgx.Tx = (*Tx)(nil)
|
||||
)
|
||||
|
||||
// --- End-to-end runs against the fake host ----------------------------------
|
||||
|
||||
func widgetRow(id uuid.UUID, name string, count int32) *abiv1.DbRow {
|
||||
return &abiv1.DbRow{Values: []*abiv1.DbValue{
|
||||
{Kind: &abiv1.DbValue_UuidValue{UuidValue: id.String()}},
|
||||
{Kind: &abiv1.DbValue_StringValue{StringValue: name}},
|
||||
{Kind: &abiv1.DbValue_Null{Null: true}}, // notes → NULL → (*string)(nil)
|
||||
{Kind: &abiv1.DbValue_Int64Value{Int64Value: int64(count)}},
|
||||
{Kind: &abiv1.DbValue_Null{Null: true}}, // created_at → NULL Timestamptz
|
||||
}}
|
||||
}
|
||||
|
||||
func TestSqlcGetWidget(t *testing.T) {
|
||||
id := uuid.New()
|
||||
host := &fakeHost{queryResp: &abiv1.DbRowsResponse{
|
||||
Columns: []string{"id", "name", "notes", "count", "created_at"},
|
||||
Rows: []*abiv1.DbRow{widgetRow(id, "gadget", 3)},
|
||||
}}
|
||||
q := New(NewPool(host.call))
|
||||
|
||||
w, err := q.GetWidget(context.Background(), id)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if w.ID != id || w.Name != "gadget" || w.Count != 3 {
|
||||
t.Fatalf("unexpected widget: %+v", w)
|
||||
}
|
||||
if w.Notes != nil {
|
||||
t.Fatalf("notes should be nil for NULL column, got %v", *w.Notes)
|
||||
}
|
||||
if w.CreatedAt.Valid {
|
||||
t.Fatalf("created_at should be invalid for NULL column")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSqlcListWidgets(t *testing.T) {
|
||||
host := &fakeHost{queryResp: &abiv1.DbRowsResponse{
|
||||
Columns: []string{"id", "name", "notes", "count", "created_at"},
|
||||
Rows: []*abiv1.DbRow{
|
||||
widgetRow(uuid.New(), "alpha", 1),
|
||||
widgetRow(uuid.New(), "beta", 2),
|
||||
},
|
||||
}}
|
||||
q := New(NewPool(host.call))
|
||||
|
||||
items, err := q.ListWidgets(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(items) != 2 || items[0].Name != "alpha" || items[1].Name != "beta" {
|
||||
t.Fatalf("unexpected list: %+v", items)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSqlcWithTxCreate proves the generated WithTx path binds to the bnwasm Tx
|
||||
// and that CreateWidget's RETURNING id scans, all inside one transaction with
|
||||
// the right ordered host calls.
|
||||
func TestSqlcWithTxCreate(t *testing.T) {
|
||||
newID := uuid.New()
|
||||
host := &fakeHost{
|
||||
txHandle: 42,
|
||||
queryResp: &abiv1.DbRowsResponse{
|
||||
Columns: []string{"id"},
|
||||
Rows: []*abiv1.DbRow{{Values: []*abiv1.DbValue{{Kind: &abiv1.DbValue_UuidValue{UuidValue: newID.String()}}}}},
|
||||
},
|
||||
}
|
||||
pool := NewPool(host.call)
|
||||
ctx := context.Background()
|
||||
|
||||
tx, err := pool.Begin(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
notes := "hi"
|
||||
gotID, err := New(pool).WithTx(tx).CreateWidget(ctx, CreateWidgetParams{Name: "w", Notes: ¬es})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if gotID != newID {
|
||||
t.Fatalf("returning id: want %s got %s", newID, gotID)
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// begin → query(RETURNING) → commit; the query carried the tx handle.
|
||||
if got := host.methods(); len(got) != 3 || got[0] != methodTxBegin || got[1] != methodQuery || got[2] != methodTxCommit {
|
||||
t.Fatalf("method sequence: %v", got)
|
||||
}
|
||||
if host.calls[1].txHandle != 42 {
|
||||
t.Fatalf("in-tx query should carry handle 42, got %d", host.calls[1].txHandle)
|
||||
}
|
||||
// The NULL-able *string arg marshaled as a real string arg.
|
||||
if len(host.calls[1].args) != 2 {
|
||||
t.Fatalf("want 2 args, got %d", len(host.calls[1].args))
|
||||
}
|
||||
}
|
||||
164
plugin/wasmguest/bnwasm/transport.go
Normal file
164
plugin/wasmguest/bnwasm/transport.go
Normal file
@ -0,0 +1,164 @@
|
||||
// Package bnwasm is the guest-side database access layer for wazero-loaded
|
||||
// BlockNinja plugins. It presents two surfaces over the SAME db.* host calls
|
||||
// (abiv1 db.proto), so plugin code that talks to Postgres keeps compiling and
|
||||
// running unchanged inside the wasm sandbox:
|
||||
//
|
||||
// - A database/sql/driver registered as "bnwasm" (driver.go), for plugins or
|
||||
// sqlc configs that use the standard library.
|
||||
// - A plugin.Pool implementation (pool.go) that hands out a pgx.Tx-shaped
|
||||
// value (tx.go), so the pgx-flavored sqlc DBTX interface every current
|
||||
// plugin generates against (sql_package: "pgx/v5") is satisfied with no
|
||||
// source edits. This is the PRIMARY path: symposium/messenger sqlc output
|
||||
// uses pgconn.CommandTag / pgx.Rows / pgx.Row, which database/sql cannot
|
||||
// produce, so the pgx surface is what their generated db packages bind to.
|
||||
//
|
||||
// # The transport seam
|
||||
//
|
||||
// Every DB operation is "marshal a db.<op> request, invoke it, unmarshal the
|
||||
// reply". That transport is a Transport func injected at construction, mirroring
|
||||
// caps.CallFunc exactly, so the marshaling/scanning logic here stays natively
|
||||
// testable with a fake host (driver_test.go, sqlcgen_test.go) while the wasm
|
||||
// guest shim binds the real host_call-backed transport (wired from package
|
||||
// wasmguest, which imports this package — never the reverse). A nil Transport
|
||||
// (native builds, DESCRIBE probes) fails every call cleanly with errNoHost
|
||||
// instead of nil-panicking, matching package caps.
|
||||
package bnwasm
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
abiv1 "git.dev.alexdunmow.com/block/core/abi/v1"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"google.golang.org/protobuf/proto"
|
||||
)
|
||||
|
||||
// Transport is the guest→host DB transport: marshal req, invoke the host with
|
||||
// the db.* method, and unmarshal the reply into resp. It is structurally
|
||||
// identical to caps.CallFunc so the wasip1 shim can bind one func to both.
|
||||
type Transport func(method string, req, resp proto.Message) error
|
||||
|
||||
// db.* method names (see core/docs/wasm-abi.md §"Capability calls").
|
||||
const (
|
||||
methodQuery = "db.query"
|
||||
methodExec = "db.exec"
|
||||
methodTxBegin = "db.tx_begin"
|
||||
methodTxCommit = "db.tx_commit"
|
||||
methodTxRollback = "db.tx_rollback"
|
||||
)
|
||||
|
||||
// errNoHost is returned by every operation when no transport is bound (native
|
||||
// build / DESCRIBE probe). It is distinct so tests can assert on it.
|
||||
var errNoHost = errors.New("bnwasm: no host transport bound (native build)")
|
||||
|
||||
// dbErr maps a DbError from a response into a *pgconn.PgError, the same error
|
||||
// type pgx surfaces, so plugin code that does errors.As(err, &pgErr) to inspect
|
||||
// a SQLSTATE (e.g. "23505" unique_violation) keeps working across the sandbox.
|
||||
func dbErr(e *abiv1.DbError) error {
|
||||
if e == nil {
|
||||
return nil
|
||||
}
|
||||
return &pgconn.PgError{Code: e.GetCode(), Message: e.GetMessage()}
|
||||
}
|
||||
|
||||
func (t Transport) query(ctx context.Context, sql string, txHandle uint64, args []any) (*abiv1.DbRowsResponse, error) {
|
||||
if t == nil {
|
||||
return nil, errNoHost
|
||||
}
|
||||
if err := ctxErr(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
vals, err := toDbValues(args)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resp := &abiv1.DbRowsResponse{}
|
||||
if err := t(methodQuery, &abiv1.DbQueryRequest{Sql: sql, Args: vals, TxHandle: txHandle}, resp); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := dbErr(resp.GetError()); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
func (t Transport) exec(ctx context.Context, sql string, txHandle uint64, args []any) (pgconn.CommandTag, error) {
|
||||
if t == nil {
|
||||
return pgconn.CommandTag{}, errNoHost
|
||||
}
|
||||
if err := ctxErr(ctx); err != nil {
|
||||
return pgconn.CommandTag{}, err
|
||||
}
|
||||
vals, err := toDbValues(args)
|
||||
if err != nil {
|
||||
return pgconn.CommandTag{}, err
|
||||
}
|
||||
resp := &abiv1.DbExecResponse{}
|
||||
if err := t(methodExec, &abiv1.DbExecRequest{Sql: sql, Args: vals, TxHandle: txHandle}, resp); err != nil {
|
||||
return pgconn.CommandTag{}, err
|
||||
}
|
||||
if err := dbErr(resp.GetError()); err != nil {
|
||||
return pgconn.CommandTag{}, err
|
||||
}
|
||||
// Encode rows-affected in a CommandTag whose trailing integer pgconn parses
|
||||
// back out via RowsAffected() — the only field sqlc-generated code reads.
|
||||
return pgconn.NewCommandTag(fmt.Sprintf("EXEC %d", resp.GetRowsAffected())), nil
|
||||
}
|
||||
|
||||
func (t Transport) txBegin(ctx context.Context) (uint64, error) {
|
||||
if t == nil {
|
||||
return 0, errNoHost
|
||||
}
|
||||
if err := ctxErr(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
resp := &abiv1.DbTxBeginResponse{}
|
||||
if err := t(methodTxBegin, &abiv1.DbTxBeginRequest{}, resp); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if err := dbErr(resp.GetError()); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if resp.GetTxHandle() == 0 {
|
||||
return 0, errors.New("bnwasm: host returned tx_handle 0 (never valid)")
|
||||
}
|
||||
return resp.GetTxHandle(), nil
|
||||
}
|
||||
|
||||
func (t Transport) txCommit(ctx context.Context, handle uint64) error {
|
||||
if t == nil {
|
||||
return errNoHost
|
||||
}
|
||||
if err := ctxErr(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
resp := &abiv1.DbTxCommitResponse{}
|
||||
if err := t(methodTxCommit, &abiv1.DbTxCommitRequest{TxHandle: handle}, resp); err != nil {
|
||||
return err
|
||||
}
|
||||
return dbErr(resp.GetError())
|
||||
}
|
||||
|
||||
func (t Transport) txRollback(ctx context.Context, handle uint64) error {
|
||||
if t == nil {
|
||||
return errNoHost
|
||||
}
|
||||
if err := ctxErr(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
resp := &abiv1.DbTxRollbackResponse{}
|
||||
if err := t(methodTxRollback, &abiv1.DbTxRollbackRequest{TxHandle: handle}, resp); err != nil {
|
||||
return err
|
||||
}
|
||||
return dbErr(resp.GetError())
|
||||
}
|
||||
|
||||
// ctxErr short-circuits an already-cancelled/expired context before crossing
|
||||
// the boundary, matching the caps stubs; the host enforces the invoke deadline.
|
||||
func ctxErr(ctx context.Context) error {
|
||||
if ctx == nil {
|
||||
return nil
|
||||
}
|
||||
return ctx.Err()
|
||||
}
|
||||
102
plugin/wasmguest/bnwasm/tx.go
Normal file
102
plugin/wasmguest/bnwasm/tx.go
Normal file
@ -0,0 +1,102 @@
|
||||
package bnwasm
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
)
|
||||
|
||||
// errNestedTx is returned by Tx.Begin. pgx models a nested Begin as a SAVEPOINT;
|
||||
// the db.* ABI has no savepoint verb (v1 scope), and a grep of the plugin fleet
|
||||
// (symposium, messenger) found no nested-transaction/savepoint use, so this is
|
||||
// rejected explicitly rather than silently degrading correctness.
|
||||
var errNestedTx = errors.New("bnwasm: nested transactions (savepoints) are not supported")
|
||||
|
||||
// Tx is a pgx.Tx bound to a host-side transaction handle. Every Exec/Query/
|
||||
// QueryRow carries the handle so the host runs it on the transaction's
|
||||
// connection; Commit/Rollback release it. After either, the Tx is closed and
|
||||
// further statements return pgx.ErrTxClosed. The host also drops the handle at
|
||||
// the call chain's deadline (WO-WZ-007), so a guest that leaks a Tx without
|
||||
// committing has it rolled back host-side — a guest can never pin a connection.
|
||||
type Tx struct {
|
||||
t Transport
|
||||
handle uint64
|
||||
closed bool
|
||||
}
|
||||
|
||||
// Begin would start a savepoint-backed nested transaction: unsupported.
|
||||
func (tx *Tx) Begin(ctx context.Context) (pgx.Tx, error) { return nil, errNestedTx }
|
||||
|
||||
func (tx *Tx) Commit(ctx context.Context) error {
|
||||
if tx.closed {
|
||||
return pgx.ErrTxClosed
|
||||
}
|
||||
tx.closed = true
|
||||
return tx.t.txCommit(ctx, tx.handle)
|
||||
}
|
||||
|
||||
func (tx *Tx) Rollback(ctx context.Context) error {
|
||||
if tx.closed {
|
||||
return pgx.ErrTxClosed
|
||||
}
|
||||
tx.closed = true
|
||||
return tx.t.txRollback(ctx, tx.handle)
|
||||
}
|
||||
|
||||
func (tx *Tx) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) {
|
||||
if tx.closed {
|
||||
return pgconn.CommandTag{}, pgx.ErrTxClosed
|
||||
}
|
||||
return tx.t.exec(ctx, sql, tx.handle, args)
|
||||
}
|
||||
|
||||
func (tx *Tx) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) {
|
||||
if tx.closed {
|
||||
return errRows(pgx.ErrTxClosed), pgx.ErrTxClosed
|
||||
}
|
||||
resp, err := tx.t.query(ctx, sql, tx.handle, args)
|
||||
if err != nil {
|
||||
return errRows(err), err
|
||||
}
|
||||
return newPgxRows(resp), nil
|
||||
}
|
||||
|
||||
func (tx *Tx) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row {
|
||||
if tx.closed {
|
||||
return &pgxRow{err: pgx.ErrTxClosed}
|
||||
}
|
||||
resp, err := tx.t.query(ctx, sql, tx.handle, args)
|
||||
if err != nil {
|
||||
return &pgxRow{err: err}
|
||||
}
|
||||
return &pgxRow{rows: newPgxRows(resp)}
|
||||
}
|
||||
|
||||
// --- Unsupported pgx.Tx surface (not emitted by sqlc; explicit errors) ---
|
||||
|
||||
func (tx *Tx) CopyFrom(context.Context, pgx.Identifier, []string, pgx.CopyFromSource) (int64, error) {
|
||||
return 0, errors.New("bnwasm: CopyFrom is not supported over the wasm DB ABI")
|
||||
}
|
||||
|
||||
func (tx *Tx) SendBatch(context.Context, *pgx.Batch) pgx.BatchResults {
|
||||
return errBatchResults{errors.New("bnwasm: SendBatch is not supported over the wasm DB ABI")}
|
||||
}
|
||||
|
||||
func (tx *Tx) LargeObjects() pgx.LargeObjects { return pgx.LargeObjects{} }
|
||||
|
||||
func (tx *Tx) Prepare(context.Context, string, string) (*pgconn.StatementDescription, error) {
|
||||
return nil, errors.New("bnwasm: Prepare is not supported over the wasm DB ABI")
|
||||
}
|
||||
|
||||
func (tx *Tx) Conn() *pgx.Conn { return nil }
|
||||
|
||||
// errBatchResults is a pgx.BatchResults that reports the same error from every
|
||||
// method, so an unsupported SendBatch surfaces cleanly instead of nil-panicking.
|
||||
type errBatchResults struct{ err error }
|
||||
|
||||
func (e errBatchResults) Exec() (pgconn.CommandTag, error) { return pgconn.CommandTag{}, e.err }
|
||||
func (e errBatchResults) Query() (pgx.Rows, error) { return errRows(e.err), e.err }
|
||||
func (e errBatchResults) QueryRow() pgx.Row { return &pgxRow{err: e.err} }
|
||||
func (e errBatchResults) Close() error { return e.err }
|
||||
Loading…
x
Reference in New Issue
Block a user