orm/db/zap.go

// ZAP protocol driver for the ORM.
//
// ZAP (Zero-Copy App Proto) uses binary encoding over RPC, communicating
// directly with ZAP-native backends (hanzo/sql, hanzo/kv, hanzo/datastore,
// hanzo/documentdb). Each backend speaks ZAP natively — no sidecar needed.
//
// Transport is github.com/zap-proto/http: a fasthttp-style request/response
// exchange carried over ZAP length-prefixed frames (encoded by the pure-Go
// zap-proto/go runtime). This is the same ZAP-HTTP transport the gateway,
// ingress, and luxd use — one and only one internal transport. The driver
// speaks it as a client: each backend op is a POST to a path (/query,
// /get, /set, /find, …) with a JSON body; the response carries a status and
// a JSON body. Routing is by address (each backend on its own port; see
// DefaultPorts) and by path — there is no peer-discovery layer, so the ORM
// takes no mDNS dependency.
package db

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/valyala/fasthttp"
	zaphttp "github.com/zap-proto/http"
)

// ZapBackend selects which ZAP-native backend to connect to.
type ZapBackend int

const (
	// ZapSQL connects to hanzo/sql, the relational transactional backend, on
	// port 9651.
	ZapSQL ZapBackend = iota
	// ZapDocumentDB connects to hanzo/documentdb on port 9654. It serves
	// document semantics over relational storage, for clients that model data
	// as documents; the rows live in hanzo/sql.
	ZapDocumentDB
	// ZapKV connects to hanzo/kv, the key-value cache and session backend, on
	// port 9653.
	ZapKV
	// ZapDatastore connects to hanzo/datastore, the columnar analytics
	// backend, on port 9655.
	ZapDatastore
)

// DefaultPorts for each ZAP-native backend.
var DefaultPorts = map[ZapBackend]int{
	ZapSQL:        9651,
	ZapKV:         9653,
	ZapDocumentDB: 9654,
	ZapDatastore:  9655,
}

// ZapConfig configures a ZAP database connection.
type ZapConfig struct {
	// Addr is the backend address (e.g., "localhost:9651").
	// If empty, uses DefaultPorts[Backend] on localhost.
	Addr string

	// Backend selects which ZAP-native backend to connect to.
	Backend ZapBackend

	// Database is the target database name (for SQL/DocumentDB backends).
	Database string

	// Collection is the default collection/table for entity storage.
	// Defaults to "_entities" for SQL, "entities" for DocumentDB.
	Collection string

	// QueryTimeout is the per-query timeout (default 30s).
	QueryTimeout time.Duration
}

// errZapClosed is every call on a ZapDB after Close: a store that cannot answer,
// never an empty one.
var errZapClosed = errors.New("db: zap: closed")

// ZapDB implements db.DB over the ZAP-HTTP binary protocol.
type ZapDB struct {
	transport *zaphttp.Transport
	cfg       ZapConfig
	mu        sync.RWMutex
	closed    bool
	tenantID  string
	// inTx marks the handle a SQL transaction runs on (zap_tx.go): its one
	// connection holds the transaction, so every statement through it is part
	// of that transaction.
	inTx bool
	// indexed is the set of kind/path indexes this process has created (Index),
	// shared with the transactions it opens so they plan as it does.
	indexed *sync.Map
}

// NewZapDB dials a ZAP-native backend and returns a DB implementation. The
// transport connects lazily on the first operation, so this does not fail
// when the backend is momentarily unreachable — the first Get/Put surfaces a
// clear dial error instead.
func NewZapDB(cfg *ZapConfig) (*ZapDB, error) {
	if cfg.Addr == "" {
		port, ok := DefaultPorts[cfg.Backend]
		if !ok {
			return nil, errors.New("db: zap addr required")
		}
		cfg.Addr = fmt.Sprintf("localhost:%d", port)
	}
	if cfg.Collection == "" {
		switch cfg.Backend {
		case ZapSQL, ZapDatastore:
			cfg.Collection = "_entities"
		case ZapDocumentDB:
			cfg.Collection = "entities"
		case ZapKV:
			cfg.Collection = "orm"
		}
	}
	if cfg.QueryTimeout == 0 {
		cfg.QueryTimeout = 30 * time.Second
	}

	t := zaphttp.Dial("tcp", cfg.Addr)
	t.SetReadTimeout(cfg.QueryTimeout)

	return &ZapDB{
		transport: t,
		cfg:       *cfg,
		indexed:   new(sync.Map),
	}, nil
}

// call executes one request/response exchange (exchange, without its retry flag).
func (z *ZapDB) call(ctx context.Context, path string, body []byte) (uint32, []byte, error) {
	status, resp, _, err := z.exchange(ctx, path, body)
	return status, resp, err
}

// exchange is call, and whether the request went out a second time.
//
// A statement outside a transaction that fails because its connection is gone is
// sent once more, on a fresh connection: the pool's idle connections are dropped
// first, because a server that restarted — a failover, a roll — left every one of
// them dead, and the next pooled connection would fail the same way. Every SQL
// write here is safe to send twice (a put is an upsert, a delete marks a row) but a
// create-if-absent, which reads `retried` to tell its own row from another's.
// Inside a transaction the transaction is retried instead (zap_tx.go).
func (z *ZapDB) exchange(ctx context.Context, path string, body []byte) (uint32, []byte, bool, error) {
	status, resp, err := z.callOnce(ctx, path, body)
	if err == nil || z.inTx || z.cfg.Backend != ZapSQL || !connectionLost(err) || ctx.Err() != nil {
		return status, resp, false, err
	}
	z.transport.CloseIdleConnections()
	status, resp, err = z.callOnce(ctx, path, body)
	return status, resp, true, err
}

// connectionLost reports whether err is a connection that ended under a request,
// as opposed to a reply that was an error.
func connectionLost(err error) bool {
	if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, net.ErrClosed) ||
		errors.Is(err, syscall.ECONNRESET) || errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNREFUSED) {
		return true
	}
	msg := err.Error()
	return strings.Contains(msg, "EOF") || strings.Contains(msg, "connection reset") ||
		strings.Contains(msg, "broken pipe") || strings.Contains(msg, "use of closed network connection")
}

// call executes one ZAP-HTTP request/response exchange: POST path with a JSON
// body, returning the response status and body. The transport owns framing
// and the ZAP wire codec.
func (z *ZapDB) callOnce(ctx context.Context, path string, body []byte) (uint32, []byte, error) {
	if err := ctx.Err(); err != nil {
		return 0, nil, err
	}
	z.mu.RLock()
	closed := z.closed
	z.mu.RUnlock()
	if closed {
		return 0, nil, errZapClosed
	}

	req := fasthttp.AcquireRequest()
	resp := fasthttp.AcquireResponse()
	defer fasthttp.ReleaseRequest(req)
	defer fasthttp.ReleaseResponse(resp)

	req.Header.SetMethod(fasthttp.MethodPost)
	req.SetRequestURI(path)
	req.Header.SetHost(z.cfg.Addr)
	req.Header.SetContentType("application/json")
	req.SetBody(body)

	if err := z.transport.DoContext(ctx, req, resp); err != nil {
		if z.inTx && path != "/commit" && ctx.Err() == nil {
			// The server rolls back a transaction whose connection is gone, so a
			// transaction that lost it before /commit wrote nothing and may run
			// again. One that lost it during /commit may have committed: that
			// stays an error.
			return 0, nil, fmt.Errorf("%w: zap call %s: transaction connection lost: %v", ErrSerializationFailure, path, err)
		}
		return 0, nil, fmt.Errorf("db: zap call %s: %w", path, err)
	}

	// The response body is owned by resp, which is released on return; copy it
	// out so the caller keeps a stable slice.
	out := append([]byte(nil), resp.Body()...)
	return uint32(resp.StatusCode()), out, nil
}

func (z *ZapDB) Get(ctx context.Context, key Key, dst any) error {
	if ctx == nil {
		ctx = context.Background()
	}
	ctx, cancel := context.WithTimeout(ctx, z.cfg.QueryTimeout)
	defer cancel()

	switch z.cfg.Backend {
	case ZapDocumentDB:
		return z.docGet(ctx, key, dst)
	case ZapKV:
		return z.kvGet(ctx, key, dst)
	default:
		return z.sqlGet(ctx, key, dst)
	}
}

func (z *ZapDB) Put(ctx context.Context, key Key, src any) (Key, error) {
	if key == nil || key.Incomplete() {
		return nil, ErrInvalidKey
	}
	if ctx == nil {
		ctx = context.Background()
	}
	ctx, cancel := context.WithTimeout(ctx, z.cfg.QueryTimeout)
	defer cancel()

	switch z.cfg.Backend {
	case ZapDocumentDB:
		return z.docPut(ctx, key, src)
	case ZapKV:
		return z.kvPut(ctx, key, src)
	default:
		return z.sqlPut(ctx, key, src)
	}
}

// CreateIfAbsent conditionally inserts src under key, first-writer-wins. See
// db.DB.CreateIfAbsent for the contract. Dispatch mirrors Put: each ZAP-native
// backend uses its own conditional-insert primitive (SQL ON CONFLICT, Valkey
// SET NX, document unique _id). The reply decides created; an unrecognized or
// error reply returns an error rather than a guessed created value, so a caller
// never mistakes an upsert or a transport failure for a first-writer win.
//
// Status: the hanzo ZAP backends do not yet expose a zap-proto/http listener
// (see LLM.md), so these paths are wire-complete but exercised only by the
// env-gated live integration test, not unit CI. The SQLite backend is the
// fully-tested reference implementation of the identical contract.
func (z *ZapDB) CreateIfAbsent(ctx context.Context, key Key, src any) (bool, error) {
	if key == nil || key.Incomplete() {
		return false, ErrInvalidKey
	}
	if ctx == nil {
		ctx = context.Background()
	}
	ctx, cancel := context.WithTimeout(ctx, z.cfg.QueryTimeout)
	defer cancel()

	switch z.cfg.Backend {
	case ZapDocumentDB:
		return z.docCreateIfAbsent(ctx, key, src)
	case ZapKV:
		return z.kvCreateIfAbsent(ctx, key, src)
	default:
		return z.sqlCreateIfAbsent(ctx, key, src)
	}
}

func (z *ZapDB) Delete(ctx context.Context, key Key) error {
	if ctx == nil {
		ctx = context.Background()
	}
	ctx, cancel := context.WithTimeout(ctx, z.cfg.QueryTimeout)
	defer cancel()

	switch z.cfg.Backend {
	case ZapDocumentDB:
		return z.docDelete(ctx, key)
	case ZapKV:
		return z.kvDelete(ctx, key)
	default:
		return z.sqlDelete(ctx, key)
	}
}

func (z *ZapDB) GetMulti(ctx context.Context, keys []Key, dst any) error {
	for _, k := range keys {
		if err := z.Get(ctx, k, dst); err != nil {
			return err
		}
	}
	return nil
}

func (z *ZapDB) PutMulti(ctx context.Context, keys []Key, src any) ([]Key, error) {
	return nil, errors.New("db: zap PutMulti not yet implemented")
}

func (z *ZapDB) DeleteMulti(ctx context.Context, keys []Key) error {
	for _, k := range keys {
		if err := z.Delete(ctx, k); err != nil {
			return err
		}
	}
	return nil
}

func (z *ZapDB) Query(kind string) Query {
	return &zapQuery{db: z, kind: kind}
}

func (z *ZapDB) VectorSearch(ctx context.Context, opts *VectorSearchOptions) ([]VectorResult, error) {
	return nil, errors.New("db: zap vector search not yet implemented")
}

func (z *ZapDB) PutVector(ctx context.Context, kind string, id string, vector []float32, metadata map[string]any) error {
	return errors.New("db: zap PutVector not yet implemented")
}

// NewKey names the entity it is given. An empty stringID with no intID is an
// incomplete key, refused by every write, as on SQLite; NewIncompleteKey is the
// one call that mints an id.
func (z *ZapDB) NewKey(kind string, stringID string, intID int64, parent Key) Key {
	id := stringID
	if id == "" && intID != 0 {
		id = fmt.Sprintf("%d", intID)
	}
	return &zapKey{kind: kind, stringID: id, intID: intID, parent: parent}
}

func (z *ZapDB) NewIncompleteKey(kind string, parent Key) Key {
	return z.NewKey(kind, newStringID(), 0, parent)
}

func (z *ZapDB) AllocateIDs(kind string, parent Key, n int) ([]Key, error) {
	keys := make([]Key, n)
	for i := range n {
		keys[i] = z.NewKey(kind, newStringID(), 0, parent)
	}
	return keys, nil
}

func (z *ZapDB) RunInTransaction(ctx context.Context, fn func(tx Transaction) error, opts *TransactionOptions) error {
	if ctx == nil {
		ctx = context.Background()
	}
	if z.cfg.Backend == ZapSQL {
		return z.sqlRunInTransaction(ctx, fn, opts)
	}
	// The other backends have no transaction to open; fn runs against a view
	// that forwards each op over ZAP.
	return fn(&zapTransaction{db: z, ctx: ctx})
}

func (z *ZapDB) Close() error {
	z.mu.Lock()
	defer z.mu.Unlock()
	if z.closed {
		return nil
	}
	z.closed = true
	z.transport.CloseIdleConnections()
	return nil
}

// --- SQL backend ---

func (z *ZapDB) sqlGet(ctx context.Context, key Key, dst any) error {
	body, _ := json.Marshal(map[string]any{
		"sql":  fmt.Sprintf("SELECT data FROM %s WHERE id = $1 AND kind = $2 AND deleted = false%s", z.cfg.Collection, forUpdate(ctx)),
		"args": []any{key.StringID(), key.Kind()},
	})
	status, resp, err := z.call(ctx, "/query", body)
	if err != nil {
		return err
	}
	if status != 200 {
		return sqlError("get", status, resp)
	}
	return z.unmarshalSQLRows(resp, dst)
}

// sqlPut writes an entity at its key. A put makes its row LIVE: delete leaves a
// tombstone here as it does on SQLite, so an upsert that replaced only `data`
// stored the entity and left every reader — all of which ask for deleted = false —
// unable to see it. Same rule and same kind guard as sqliteQuery's putSQL, because
// it is the same table under a different dialect.
//
// Every argument reaches the server as text — the listener binds them TEXTOID,
// whatever JSON type they were sent as — so an argument going into a column of
// another type says which type that is. Without the casts PostgreSQL refuses
// the statement outright: "column data is of type jsonb but expression is of
// type text", which is a 500 on every write and no clue as to which word.
func (z *ZapDB) sqlPut(ctx context.Context, key Key, src any) (Key, error) {
	data, err := json.Marshal(src)
	if err != nil {
		return nil, err
	}
	now := timeNow().Format(time.RFC3339)
	body, _ := json.Marshal(map[string]any{
		"sql": fmt.Sprintf(`INSERT INTO %s (id, kind, data, created_at, updated_at, deleted)
			VALUES ($1, $2, $3::jsonb, $4::timestamptz, $5::timestamptz, false)
			ON CONFLICT (id) DO UPDATE SET data = $3::jsonb, updated_at = $5::timestamptz,
				deleted = CASE WHEN %s.kind = $2 THEN false ELSE %s.deleted END`,
			z.cfg.Collection, z.cfg.Collection, z.cfg.Collection),
		"args": []any{key.StringID(), key.Kind(), string(data), now, now},
	})
	status, resp, err := z.call(ctx, "/exec", body)
	if err != nil {
		return nil, err
	}
	if status != 200 {
		return nil, sqlError("put", status, resp)
	}
	return key, nil
}

// sqlCreateIfAbsent is the SQL-backend conditional insert. It mirrors the SQLite
// createIfAbsentSQL: a fresh id inserts, a SAME-KIND soft-deleted id resurrects
// (kind is guarded in the WHERE and never set, so no type confusion), a live id
// leaves the DO UPDATE ... WHERE guard unsatisfied. RETURNING id makes the
// created signal a row-count, read over the same /query row-array envelope
// sqlGet consumes — so it depends on no new wire contract.
//
// Whether a different-kind id collision surfaces loudly here depends on the
// server table's key (an id primary key vs a composite (id, kind)); the caller
// precondition — keep each kind in its own stringID keyspace — governs it either
// way. This path is exercised by the env-gated live test, not unit CI.
func (z *ZapDB) sqlCreateIfAbsent(ctx context.Context, key Key, src any) (bool, error) {
	data, err := json.Marshal(src)
	if err != nil {
		return false, err
	}
	now := timeNow().Format(time.RFC3339)
	body, _ := json.Marshal(map[string]any{
		"sql": fmt.Sprintf(`INSERT INTO %s (id, kind, data, created_at, updated_at, deleted)
			VALUES ($1, $2, $3::jsonb, $4::timestamptz, $5::timestamptz, false)
			ON CONFLICT (id) DO UPDATE SET
				data = $3::jsonb, updated_at = $5::timestamptz, deleted = false
			WHERE %s.deleted = true AND %s.kind = $2
			RETURNING id`, z.cfg.Collection, z.cfg.Collection, z.cfg.Collection),
		"args": []any{key.StringID(), key.Kind(), string(data), now, now},
	})
	status, resp, retried, err := z.exchange(ctx, "/query", body)
	if err != nil {
		return false, err
	}
	if status != 200 {
		return false, sqlError("create-if-absent", status, resp)
	}
	created, err := zapRowsReturned(resp)
	if err != nil || created || !retried {
		return created, err
	}
	// The first send's connection ended without a reply, so it may have created the
	// row this one found. It did if the row holds exactly what was sent.
	var have json.RawMessage
	if err := z.sqlGet(ctx, key, &have); err != nil {
		return false, err
	}
	return sameJSON(have, data), nil
}

// sameJSON reports whether a and b are one JSON document: keys in any order.
func sameJSON(a, b []byte) bool {
	var va, vb any
	if json.Unmarshal(a, &va) != nil || json.Unmarshal(b, &vb) != nil {
		return false
	}
	ca, _ := json.Marshal(va)
	cb, _ := json.Marshal(vb)
	return bytes.Equal(ca, cb)
}

// zapRowsReturned reports whether a /query reply carried at least one row — the
// created signal for INSERT ... ON CONFLICT ... RETURNING. It reuses the row-array
// envelope sqlGet decodes; an undecodable reply is an error, never a created win.
func zapRowsReturned(resp []byte) (bool, error) {
	var rows []map[string]any
	if err := json.Unmarshal(resp, &rows); err != nil {
		return false, fmt.Errorf("db: zap create-if-absent: decode reply: %w", err)
	}
	return len(rows) > 0, nil
}

func (z *ZapDB) sqlDelete(ctx context.Context, key Key) error {
	body, _ := json.Marshal(map[string]any{
		"sql":  fmt.Sprintf("UPDATE %s SET deleted = true, updated_at = $1::timestamptz WHERE id = $2 AND kind = $3", z.cfg.Collection),
		"args": []any{timeNow().Format(time.RFC3339), key.StringID(), key.Kind()},
	})
	status, resp, err := z.call(ctx, "/exec", body)
	if err != nil {
		return err
	}
	if status != 200 {
		return sqlError("delete", status, resp)
	}
	return nil
}

// --- DocumentDB backend ---

func (z *ZapDB) docGet(ctx context.Context, key Key, dst any) error {
	body, _ := json.Marshal(map[string]any{
		"collection": z.cfg.Collection,
		"filter":     map[string]any{"_id": key.StringID(), "kind": key.Kind(), "deleted": false},
		"limit":      1,
	})
	status, resp, err := z.call(ctx, "/find", body)
	if err != nil {
		return err
	}
	if status != 200 {
		return ErrNoSuchEntity
	}
	return z.unmarshalDocResult(resp, dst)
}

func (z *ZapDB) docPut(ctx context.Context, key Key, src any) (Key, error) {
	data, _ := json.Marshal(src)
	var doc map[string]any
	json.Unmarshal(data, &doc)
	doc["_id"] = key.StringID()
	doc["kind"] = key.Kind()
	doc["deleted"] = false
	doc["updatedAt"] = timeNow().Format(time.RFC3339)

	// Upsert via update, fall back to insert
	body, _ := json.Marshal(map[string]any{
		"collection": z.cfg.Collection,
		"filter":     map[string]any{"_id": key.StringID()},
		"update":     map[string]any{"$set": doc},
	})
	status, _, err := z.call(ctx, "/update", body)
	if err != nil || status != 200 {
		body, _ = json.Marshal(map[string]any{
			"collection": z.cfg.Collection,
			"documents":  []any{doc},
		})
		_, _, err = z.call(ctx, "/insert", body)
		if err != nil {
			return nil, err
		}
	}
	return key, nil
}

// docCreateIfAbsent is the document-backend conditional insert. The backend's
// unique _id index is the first-writer-wins gate: the winning /insert reports
// created=true; a duplicate _id (the key already holds a document) reports
// created=false. An unexpected status is an error, never a created win.
//
// src must marshal to a JSON object so the _id/kind envelope fields can be set;
// a non-object (array/scalar) is rejected rather than panicking on a nil map.
//
// Limitation (flagged, untested-here — the backend exposes no listener yet):
// existence keys on _id alone, so a soft-deleted _id reports created=false while
// docGet filters deleted, and a different-kind _id does not surface as
// ErrKindMismatch the way the SQLite path does. The caller preconditions —
// separate keyspace per kind, normalize before the key — cover both on this
// backend until it is live-verified.
func (z *ZapDB) docCreateIfAbsent(ctx context.Context, key Key, src any) (bool, error) {
	data, err := json.Marshal(src)
	if err != nil {
		return false, err
	}
	var doc map[string]any
	if err := json.Unmarshal(data, &doc); err != nil {
		return false, fmt.Errorf("db: zap document create-if-absent: src must be a JSON object: %w", err)
	}
	if doc == nil {
		return false, fmt.Errorf("db: zap document create-if-absent: src must be a JSON object, got null or non-object")
	}
	doc["_id"] = key.StringID()
	doc["kind"] = key.Kind()
	doc["deleted"] = false
	doc["createdAt"] = timeNow().Format(time.RFC3339)
	doc["updatedAt"] = timeNow().Format(time.RFC3339)

	body, _ := json.Marshal(map[string]any{
		"collection": z.cfg.Collection,
		"documents":  []any{doc},
	})
	status, _, err := z.call(ctx, "/insert", body)
	if err != nil {
		return false, err
	}
	switch status {
	case fasthttp.StatusOK:
		return true, nil
	case fasthttp.StatusConflict:
		// Duplicate _id — a document already holds the key.
		return false, nil
	default:
		return false, fmt.Errorf("db: zap document create-if-absent: status %d", status)
	}
}

func (z *ZapDB) docDelete(ctx context.Context, key Key) error {
	body, _ := json.Marshal(map[string]any{
		"collection": z.cfg.Collection,
		"filter":     map[string]any{"_id": key.StringID()},
		"update":     map[string]any{"$set": map[string]any{"deleted": true}},
	})
	_, _, err := z.call(ctx, "/update", body)
	return err
}

// --- KV backend ---

func (z *ZapDB) kvGet(ctx context.Context, key Key, dst any) error {
	kvKey := fmt.Sprintf("%s:%s:%s", z.cfg.Collection, key.Kind(), key.StringID())
	body, _ := json.Marshal(map[string]any{"key": kvKey})
	status, resp, err := z.call(ctx, "/get", body)
	if err != nil {
		return err
	}
	if status != 200 || len(resp) == 0 {
		return ErrNoSuchEntity
	}
	return json.Unmarshal(resp, dst)
}

func (z *ZapDB) kvPut(ctx context.Context, key Key, src any) (Key, error) {
	data, _ := json.Marshal(src)
	kvKey := fmt.Sprintf("%s:%s:%s", z.cfg.Collection, key.Kind(), key.StringID())
	body, _ := json.Marshal(map[string]any{
		"key":   kvKey,
		"value": string(data),
	})
	_, _, err := z.call(ctx, "/set", body)
	if err != nil {
		return nil, err
	}
	return key, nil
}

// kvCreateIfAbsent is the KV-backend conditional insert via Valkey SET ... NX,
// the KV-native first-writer-wins gate. NX writes only when the key is absent
// (the KV backend hard-deletes, so absent == not present, no soft-delete case).
// Valkey replies +OK when the write applied and a null bulk string when NX
// suppressed it; any other reply is an error, never a created win.
func (z *ZapDB) kvCreateIfAbsent(ctx context.Context, key Key, src any) (bool, error) {
	data, _ := json.Marshal(src)
	kvKey := fmt.Sprintf("%s:%s:%s", z.cfg.Collection, key.Kind(), key.StringID())
	body, _ := json.Marshal(map[string]any{
		"cmd":  "SET",
		"args": []string{kvKey, string(data), "NX"},
	})
	status, resp, err := z.call(ctx, "/cmd", body)
	if err != nil {
		return false, err
	}
	if status != fasthttp.StatusOK {
		return false, fmt.Errorf("db: zap kv create-if-absent: status %d", status)
	}
	switch strings.TrimSpace(string(resp)) {
	case "OK", `"OK"`, "+OK":
		return true, nil
	case "", "nil", "null", "$-1":
		return false, nil
	default:
		return false, fmt.Errorf("db: zap kv create-if-absent: unexpected reply %q", resp)
	}
}

func (z *ZapDB) kvDelete(ctx context.Context, key Key) error {
	kvKey := fmt.Sprintf("%s:%s:%s", z.cfg.Collection, key.Kind(), key.StringID())
	body, _ := json.Marshal(map[string]any{
		"cmd":  "DEL",
		"args": []string{kvKey},
	})
	_, _, err := z.call(ctx, "/cmd", body)
	return err
}

// --- Helpers ---

func (z *ZapDB) unmarshalSQLRows(resp []byte, dst any) error {
	var rows []map[string]any
	if err := json.Unmarshal(resp, &rows); err != nil {
		return json.Unmarshal(resp, dst)
	}
	if len(rows) == 0 {
		return ErrNoSuchEntity
	}
	if data, ok := rows[0]["data"]; ok {
		b, _ := json.Marshal(data)
		return json.Unmarshal(b, dst)
	}
	b, _ := json.Marshal(rows[0])
	return json.Unmarshal(b, dst)
}

func (z *ZapDB) unmarshalDocResult(resp []byte, dst any) error {
	var result struct {
		Documents []json.RawMessage `json:"documents"`
	}
	if err := json.Unmarshal(resp, &result); err != nil {
		return json.Unmarshal(resp, dst)
	}
	if len(result.Documents) == 0 {
		return ErrNoSuchEntity
	}
	return json.Unmarshal(result.Documents[0], dst)
}

// --- zapKey ---

type zapKey struct {
	kind     string
	stringID string
	intID    int64
	parent   Key
}

func (k *zapKey) Kind() string      { return k.kind }
func (k *zapKey) StringID() string  { return k.stringID }
func (k *zapKey) IntID() int64      { return k.intID }
func (k *zapKey) Namespace() string { return "" }
func (k *zapKey) Incomplete() bool  { return k.stringID == "" && k.intID == 0 }

func (k *zapKey) Encode() string {
	if k.stringID != "" {
		return k.stringID
	}
	return fmt.Sprintf("%d", k.intID)
}

func (k *zapKey) Equal(other Key) bool {
	if other == nil {
		return false
	}
	return k.kind == other.Kind() && k.Encode() == other.Encode()
}

func (k *zapKey) Parent() Key { return k.parent }

// --- zapQuery ---

type zapQuery struct {
	db      *ZapDB
	kind    string
	filters []zapFilter
	order   string
	limit   int
	offset  int
	// after is set by After: the walk is in key order, from after the id held.
	after *string
}

type zapFilter struct {
	field string
	op    string
	value any
}

func (q *zapQuery) Filter(filterStr string, value any) Query {
	field, op := ParseFilterString(filterStr)
	nq := *q
	nq.filters = append(append([]zapFilter{}, q.filters...), zapFilter{field, op, value})
	return &nq
}

func (q *zapQuery) FilterField(fieldPath string, op string, value any) Query {
	nq := *q
	nq.filters = append(append([]zapFilter{}, q.filters...), zapFilter{fieldPath, op, value})
	return &nq
}

func (q *zapQuery) Order(fieldPath string) Query { nq := *q; nq.order = fieldPath; return &nq }
func (q *zapQuery) OrderDesc(fieldPath string) Query {
	nq := *q
	nq.order = "-" + fieldPath
	return &nq
}
func (q *zapQuery) Limit(limit int) Query              { nq := *q; nq.limit = limit; return &nq }
func (q *zapQuery) After(id string) Query              { nq := *q; nq.after = &id; return &nq }
func (q *zapQuery) Offset(offset int) Query            { nq := *q; nq.offset = offset; return &nq }
func (q *zapQuery) Project(fieldNames ...string) Query { return q }
func (q *zapQuery) Distinct() Query                    { return q }
func (q *zapQuery) Ancestor(ancestor Key) Query        { return q }
func (q *zapQuery) Start(cursor Cursor) Query          { return q }
func (q *zapQuery) End(cursor Cursor) Query            { return q }

func (q *zapQuery) GetAll(ctx context.Context, dst any) ([]Key, error) {
	if q.limit <= 0 {
		q.limit = MaxGetAll
	}
	if ctx == nil {
		ctx = context.Background()
	}
	ctx, cancel := context.WithTimeout(ctx, q.db.cfg.QueryTimeout)
	defer cancel()

	switch q.db.cfg.Backend {
	case ZapDocumentDB:
		return q.docGetAll(ctx, dst)
	default:
		return q.sqlGetAll(ctx, dst)
	}
}

// First reads the first matching entity into dst, which points at one entity, not
// a slice of them.
func (q *zapQuery) First(ctx context.Context, dst any) (Key, error) {
	var docs []json.RawMessage
	keys, err := q.Limit(1).GetAll(ctx, &docs)
	if err != nil {
		return nil, err
	}
	if len(keys) == 0 {
		return nil, ErrNoSuchEntity
	}
	if dst != nil && len(docs) > 0 {
		if err := json.Unmarshal(docs[0], dst); err != nil {
			return nil, fmt.Errorf("db: zap first: decode: %w", err)
		}
	}
	return keys[0], nil
}

func (q *zapQuery) Count(ctx context.Context) (int, error) {
	if ctx == nil {
		ctx = context.Background()
	}
	ctx, cancel := context.WithTimeout(ctx, q.db.cfg.QueryTimeout)
	defer cancel()

	sql, args := q.buildSQL("COUNT(*) as count")
	body, _ := json.Marshal(map[string]any{"sql": sql, "args": args})
	status, resp, err := q.db.call(ctx, "/query", body)
	if err != nil {
		return 0, err
	}
	if status != 200 {
		return 0, sqlError("count", status, resp)
	}
	var rows []map[string]any
	if err := json.Unmarshal(resp, &rows); err != nil {
		return 0, fmt.Errorf("db: zap sql count: decode reply: %w", err)
	}
	if len(rows) == 0 {
		return 0, nil
	}
	if cnt, ok := rows[0]["count"]; ok {
		if v, ok := cnt.(float64); ok {
			return int(v), nil
		}
	}
	return 0, nil
}

// zapRows is the select list that returns entities rather than a reduction.
const zapRows = "id, data"

// Sum totals one numeric field over the selected rows.
func (q *zapQuery) Sum(ctx context.Context, field string) (float64, error) {
	total, _, err := q.reduce(ctx, field)
	return total, err
}

// Avg is the mean of that field, with the count it was taken over.
func (q *zapQuery) Avg(ctx context.Context, field string) (float64, int, error) {
	total, n, err := q.reduce(ctx, field)
	if err != nil || n == 0 {
		return 0, n, err
	}
	return total / float64(n), n, nil
}

// reduce asks for both numbers in one statement, so Sum and Avg cannot disagree
// about which rows contributed.
func (q *zapQuery) reduce(ctx context.Context, field string) (float64, int, error) {
	if ctx == nil {
		ctx = context.Background()
	}
	ctx, cancel := context.WithTimeout(ctx, q.db.cfg.QueryTimeout)
	defer cancel()

	expr := jsonField(field)
	sql, args := q.buildSQL(fmt.Sprintf(
		"COALESCE(SUM((%s)::double precision), 0) as total, COUNT(%s) as n", expr, expr))
	body, _ := json.Marshal(map[string]any{"sql": sql, "args": args})
	status, resp, err := q.db.call(ctx, "/query", body)
	if err != nil {
		return 0, 0, err
	}
	if status != 200 {
		return 0, 0, sqlError("reduce", status, resp)
	}
	var rows []map[string]any
	if err := json.Unmarshal(resp, &rows); err != nil {
		return 0, 0, fmt.Errorf("db: zap sql reduce: decode reply: %w", err)
	}
	if len(rows) == 0 {
		return 0, 0, nil
	}
	total, _ := rows[0]["total"].(float64)
	n, _ := rows[0]["n"].(float64)
	return total, int(n), nil
}

func (q *zapQuery) Keys(ctx context.Context) ([]Key, error) { return q.GetAll(ctx, nil) }
func (q *zapQuery) Run(ctx context.Context) Iterator        { return nil }

// jsonField reads one field of a stored document as text.
//
// hanzo/sql is PostgreSQL and runs this statement itself, so the statement is
// written in PostgreSQL: `data->>'name'` for a field of the document,
// `data#>>'{a,b}'` for one inside another. SQLite's json_extract is not a
// function there, and a statement that called it came back an error — every
// filtered or ordered listing against hanzo/sql failed on it.
//
// Text, because every argument arrives bound as text: the listener types them
// TEXT, and a value compares as what it is written as — a string as itself,
// `true` as "true", a number as its digits.
//
// An ORDER BY on this sorts as text. For an ISO-8601 timestamp, the spelling
// every entity here stores, that is chronological; a numeric field would sort
// 9 after 10, and wants a projection this query language cannot yet name.
func jsonField(path string) string {
	name := ToJSONFieldName(path)
	if !strings.Contains(name, ".") {
		return fmt.Sprintf("data->>'%s'", name)
	}
	return fmt.Sprintf("data#>>'{%s}'", strings.ReplaceAll(name, ".", ","))
}

// drivingFilter is the filter whose index a statement over an indexed kind uses,
// or -1: the LAST equality on an indexed field. Callers narrow last —
// Filter("Owner=", org).Filter("Email=", e) — so the last is the selective one.
// PostgreSQL left to choose between two such indexes took the owner index, which
// matches every user of an org, over the email index that matches one.
func (q *zapQuery) drivingFilter() int {
	pick := -1
	for i, f := range q.filters {
		if NormalizeOp(f.op) == "=" && !numeric(f.value) && q.db.indexedPath(q.kind, f.field) {
			pick = i
		}
	}
	return pick
}

// numeric reports whether value is compared as a number (compare).
func numeric(value any) bool {
	switch value.(type) {
	case int, int8, int16, int32, int64,
		uint, uint8, uint16, uint32, uint64,
		float32, float64:
		return true
	}
	return false
}

// compare is one condition: a field, an operator, and the argument at idx.
//
// A number is compared as a number. As text, '9' is greater than '10', so a
// range filter on a price or a count would answer with the wrong rows rather
// than with an error.
func compare(path, op string, value any, idx int) string {
	switch value.(type) {
	case int, int8, int16, int32, int64,
		uint, uint8, uint16, uint32, uint64,
		float32, float64:
		return fmt.Sprintf("(%s)::double precision %s $%d::double precision", jsonField(path), op, idx)
	default:
		return fmt.Sprintf("%s %s $%d", jsonField(path), op, idx)
	}
}

// The kind is written as a literal, not bound: an index on one of the kind's fields
// is partial (WHERE kind = '<kind>' AND deleted = false, see Index), and PostgreSQL
// uses a partial index only for a statement whose WHERE clause states its predicate.
func (q *zapQuery) buildSQL(sel string) (string, []any) {
	table := q.db.cfg.Collection
	var args []any
	idx := 1

	var sql strings.Builder
	sql.WriteString(fmt.Sprintf("SELECT %s FROM %s WHERE %s", sel, table, pgKindClause(q.kind)))

	drive := q.drivingFilter()
	for i, f := range q.filters {
		op := NormalizeOp(f.op)
		if f.value == nil {
			sql.WriteString(" AND " + absence(jsonField(f.field), op))
			continue
		}
		if i != drive && op == "=" && !numeric(f.value) && q.db.indexedPath(q.kind, f.field) {
			// An indexed field that does not drive this statement: written so that
			// no index serves it, and the one that drives is used.
			sql.WriteString(fmt.Sprintf(" AND (%s || '') = $%d", jsonField(f.field), idx))
		} else {
			sql.WriteString(" AND " + compare(f.field, op, f.value, idx))
		}
		args = append(args, f.value)
		idx++
	}
	if q.after != nil && *q.after != "" {
		sql.WriteString(fmt.Sprintf(" AND id > $%d", idx))
		args = append(args, *q.after)
		idx++
	}

	if sel == zapRows {
		if q.after != nil {
			sql.WriteString(" ORDER BY id ASC")
		} else if q.order != "" {
			desc := q.order[0] == '-'
			field := q.order
			if desc {
				field = field[1:]
			}
			dir := "ASC"
			if desc {
				dir = "DESC"
			}
			sql.WriteString(fmt.Sprintf(" ORDER BY %s %s", jsonField(field), dir))
		}
		if q.limit > 0 {
			sql.WriteString(fmt.Sprintf(" LIMIT %d", q.limit))
		}
		if q.offset > 0 {
			sql.WriteString(fmt.Sprintf(" OFFSET %d", q.offset))
		}
	}
	return sql.String(), args
}

func (q *zapQuery) sqlGetAll(ctx context.Context, dst any) ([]Key, error) {
	sql, args := q.buildSQL(zapRows)
	body, _ := json.Marshal(map[string]any{"sql": sql, "args": args})
	status, resp, err := q.db.call(ctx, "/query", body)
	if err != nil {
		return nil, err
	}
	if status != 200 {
		return nil, sqlError("query", status, resp)
	}

	var rows []map[string]any
	if err := json.Unmarshal(resp, &rows); err != nil {
		return nil, fmt.Errorf("db: zap sql query: decode reply: %w", err)
	}

	keys := make([]Key, 0, len(rows))
	for _, row := range rows {
		id, _ := row["id"].(string)
		keys = append(keys, &zapKey{kind: q.kind, stringID: id})
	}

	if dst != nil && len(rows) > 0 {
		var dataList []json.RawMessage
		for _, row := range rows {
			if data, ok := row["data"]; ok {
				b, _ := json.Marshal(data)
				dataList = append(dataList, b)
			}
		}
		combined, _ := json.Marshal(dataList)
		if err := json.Unmarshal(combined, dst); err != nil {
			return nil, fmt.Errorf("db: zap sql query: decode into %T: %w", dst, err)
		}
	}
	return keys, nil
}

func (q *zapQuery) docGetAll(ctx context.Context, dst any) ([]Key, error) {
	if q.after != nil {
		return nil, errors.New("db: the datastore backend cannot walk a kind in key order (After)")
	}
	filter := map[string]any{"kind": q.kind, "deleted": false}
	for _, f := range q.filters {
		jsonField := ToJSONFieldName(f.field)
		switch f.op {
		case "=", "==":
			filter[jsonField] = f.value
		case ">":
			filter[jsonField] = map[string]any{"$gt": f.value}
		case ">=":
			filter[jsonField] = map[string]any{"$gte": f.value}
		case "<":
			filter[jsonField] = map[string]any{"$lt": f.value}
		case "<=":
			filter[jsonField] = map[string]any{"$lte": f.value}
		case "!=":
			filter[jsonField] = map[string]any{"$ne": f.value}
		}
	}
	args := map[string]any{"collection": q.db.cfg.Collection, "filter": filter}
	if q.limit > 0 {
		args["limit"] = q.limit
	}
	body, _ := json.Marshal(args)
	_, resp, err := q.db.call(ctx, "/find", body)
	if err != nil {
		return nil, err
	}
	var result struct {
		Documents []json.RawMessage `json:"documents"`
	}
	json.Unmarshal(resp, &result)

	keys := make([]Key, 0, len(result.Documents))
	for _, doc := range result.Documents {
		var m map[string]any
		json.Unmarshal(doc, &m)
		id, _ := m["_id"].(string)
		keys = append(keys, &zapKey{kind: q.kind, stringID: id})
	}
	if dst != nil && len(result.Documents) > 0 {
		combined, _ := json.Marshal(result.Documents)
		json.Unmarshal(combined, dst)
	}
	return keys, nil
}

// --- zapTransaction ---

// zapTransaction is the view fn runs against. On the SQL backend db is the handle
// holding the transaction's connection (zap_tx.go); elsewhere it is the DB itself.
type zapTransaction struct {
	db  *ZapDB
	ctx context.Context
}

func (t *zapTransaction) Get(key Key, dst any) error {
	return t.db.Get(t.ctx, key, dst)
}

// GetForUpdate reads the entity and, inside a SQL transaction, locks its row until
// the transaction ends.
func (t *zapTransaction) GetForUpdate(key Key, dst any) error {
	return t.db.Get(withForUpdate(t.ctx), key, dst)
}

func (t *zapTransaction) Put(key Key, src any) (Key, error) {
	return t.db.Put(t.ctx, key, src)
}

func (t *zapTransaction) CreateIfAbsent(key Key, src any) (bool, error) {
	return t.db.CreateIfAbsent(t.ctx, key, src)
}

func (t *zapTransaction) Delete(key Key) error {
	return t.db.Delete(t.ctx, key)
}

func (t *zapTransaction) Query(kind string) Query {
	return t.db.Query(kind)
}