From 7881aeee05bd69b1a2f478b003371c768b778f04 Mon Sep 17 00:00:00 2001 From: Alex Dunmow Date: Fri, 3 Jul 2026 14:41:45 +0800 Subject: [PATCH] feat(wasmguest): guest DB driver `bnwasm` (WO-WZ-004) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- plugin/wasmguest/bnwasm/dbvalue.go | 278 ++++++++++++++++++ plugin/wasmguest/bnwasm/dbvalue_fixtures.go | 131 +++++++++ plugin/wasmguest/bnwasm/driver.go | 201 +++++++++++++ plugin/wasmguest/bnwasm/driver_test.go | 304 ++++++++++++++++++++ plugin/wasmguest/bnwasm/pool.go | 60 ++++ plugin/wasmguest/bnwasm/rows.go | 120 ++++++++ plugin/wasmguest/bnwasm/sqlcgen_test.go | 207 +++++++++++++ plugin/wasmguest/bnwasm/transport.go | 164 +++++++++++ plugin/wasmguest/bnwasm/tx.go | 102 +++++++ 9 files changed, 1567 insertions(+) create mode 100644 plugin/wasmguest/bnwasm/dbvalue.go create mode 100644 plugin/wasmguest/bnwasm/dbvalue_fixtures.go create mode 100644 plugin/wasmguest/bnwasm/driver.go create mode 100644 plugin/wasmguest/bnwasm/driver_test.go create mode 100644 plugin/wasmguest/bnwasm/pool.go create mode 100644 plugin/wasmguest/bnwasm/rows.go create mode 100644 plugin/wasmguest/bnwasm/sqlcgen_test.go create mode 100644 plugin/wasmguest/bnwasm/transport.go create mode 100644 plugin/wasmguest/bnwasm/tx.go diff --git a/plugin/wasmguest/bnwasm/dbvalue.go b/plugin/wasmguest/bnwasm/dbvalue.go new file mode 100644 index 0000000..83126e8 --- /dev/null +++ b/plugin/wasmguest/bnwasm/dbvalue.go @@ -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() +} diff --git a/plugin/wasmguest/bnwasm/dbvalue_fixtures.go b/plugin/wasmguest/bnwasm/dbvalue_fixtures.go new file mode 100644 index 0000000..f7b6788 --- /dev/null +++ b/plugin/wasmguest/bnwasm/dbvalue_fixtures.go @@ -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"}`, + }, +} diff --git a/plugin/wasmguest/bnwasm/driver.go b/plugin/wasmguest/bnwasm/driver.go new file mode 100644 index 0000000..fd84896 --- /dev/null +++ b/plugin/wasmguest/bnwasm/driver.go @@ -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 +} diff --git a/plugin/wasmguest/bnwasm/driver_test.go b/plugin/wasmguest/bnwasm/driver_test.go new file mode 100644 index 0000000..019b516 --- /dev/null +++ b/plugin/wasmguest/bnwasm/driver_test.go @@ -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 +} diff --git a/plugin/wasmguest/bnwasm/pool.go b/plugin/wasmguest/bnwasm/pool.go new file mode 100644 index 0000000..2f45b7a --- /dev/null +++ b/plugin/wasmguest/bnwasm/pool.go @@ -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} +} diff --git a/plugin/wasmguest/bnwasm/rows.go b/plugin/wasmguest/bnwasm/rows.go new file mode 100644 index 0000000..8371154 --- /dev/null +++ b/plugin/wasmguest/bnwasm/rows.go @@ -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 +} diff --git a/plugin/wasmguest/bnwasm/sqlcgen_test.go b/plugin/wasmguest/bnwasm/sqlcgen_test.go new file mode 100644 index 0000000..15ceb59 --- /dev/null +++ b/plugin/wasmguest/bnwasm/sqlcgen_test.go @@ -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)) + } +} diff --git a/plugin/wasmguest/bnwasm/transport.go b/plugin/wasmguest/bnwasm/transport.go new file mode 100644 index 0000000..da08039 --- /dev/null +++ b/plugin/wasmguest/bnwasm/transport.go @@ -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. 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() +} diff --git a/plugin/wasmguest/bnwasm/tx.go b/plugin/wasmguest/bnwasm/tx.go new file mode 100644 index 0000000..08d1aae --- /dev/null +++ b/plugin/wasmguest/bnwasm/tx.go @@ -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 }