Plugin SDK and contract reference

:::caution 0.2.0 Private Preview snapshot

This independent snapshot records what was true for 0.2.0. The legacy marketplace pages were illustrative in that release: no public marketplace, supported third-party SDK distribution, live publish path, revenue program, review SLA, analytics/ranking contract, or marketplace legal agreement was available.

:::

Contract families: do not mix them

Family	Identity	Status
Pinned 0.2.0 step contract	Bundle-internal protobuf/gRPC	No public plugin compatibility guarantee
0.2.0 first-party step host	Bundle-internal Go bridge	Not a public package guarantee
Repository-local WASM API	nexo-cli/sdk/go/nexo + nexo-proxy/pkg/plugin	Experimental implementation evidence; no public distribution/compatibility guarantee
Retired dashboard SDK	github.com/nexo-proxy/plugin-sdk-go@v1.4.0, Rust macros, imaginary host ABI	Legacy/illustrative and unavailable

0.2.0 step-contract boundary

The 0.2.0 bundle used its pinned built-in and sidecar interfaces as one release unit. It did not provide the later Describe RPC, authoritative phase discovery, contract_version, or data_access_profile enforcement. The safe rule is to use only bundle-pinned artifacts, confirm Operator reconciliation, verify that the runtime actually executes the intended phase, and exercise representative traffic.

Source-level protobuf messages and action values are not a public 0.2.0 SDK promise. Empty, failed, or mismatched capability behavior is unsupported rather than evidence of compatibility. Later typed routing and fail-closed negotiation described in Next documentation must not be projected backward onto this snapshot.

MongoDB wire context

The legacy pages list OP_REPLY=1, OP_QUERY=2004, OP_GET_MORE=2005, OP_COMPRESSED=2012, and OP_MSG=2013, plus update/insert/delete examples. These are MongoDB protocol values, not guaranteed 0.2.0 Nexo helper APIs.

Repository-local WASM interface

The direct local Go SDK defines NexoPlugin with Init, OnRequest, OnResponse, OnRoute, OnConnect, OnDisconnect, and Shutdown; BasePlugin supplies pass-through defaults. Contexts expose connection ID, client address, command, database, collection, raw messages, setters for modified messages/errors, and logging. RouteContext returns target strings; ConnectContext has connection ID/address.

Its actions are Continue=0, Drop=1, Reject=2, and ShortCircuit=3. This differs from the current protobuf action numbering and semantics. Never pass values between the contracts by numeric cast.

Actual local exports are nexo_init, nexo_on_request, nexo_on_response, nexo_on_route, nexo_on_connect, nexo_on_disconnect, nexo_shutdown, nexo_last_error, and nexo_alloc. Actual imports use module nexo_host: log, metric_counter, metric_histogram, kv_get, kv_set, and kv_delete. The retired nexo_process_*, packed i64 return, module nexo, config/connection/metadata host functions, and SDK v1.4.0 are a different illustrative ABI.

Configuration, errors, logs, testing, and performance

• 0.2.0 configuration belonged to pinned bundle implementations. Legacy draft-07 schemas, dashboard default injection, x-nexo-secret, AES-256-GCM claims, secret interpolation, and atomic hot reload were not guaranteed.
• Source-level error structures did not create a public 0.2.0 SDK. Legacy reserved Nexo codes 16500–16504 and on_panic behavior are illustrative.
• Local WASM logging/metrics/KV functions exist in code, but metric names, persistence, quotas, and operational support are not public contracts.
• The retired sdktest builders, benchmark harness, zero-copy native SDK, allocation targets, and public examples repository are unavailable guarantees. The general advice—validate bounds, avoid full BSON parsing when metadata suffices, reuse allocations safely, cache bounded decisions, benchmark hot paths—remains sound.

Complete legacy snippet archive

Every code block from the retired SDK page is preserved below. Names, packages, ABI shapes, behavior, performance targets, and commands are illustrative and may not compile or exist.

1. Install command

go get github.com/nexo-proxy/plugin-sdk-go@v1.4.0

2. Import path

import sdk "github.com/nexo-proxy/plugin-sdk-go"

3. Legacy Step interface

// Step is the core interface every Nexo plugin must implement.
// It defines the lifecycle hooks the proxy calls during request processing.
type Step interface {
    // Name returns a unique identifier for this step (e.g., "acme-rate-limiter").
    Name() string

    // Version returns the semantic version of this step (e.g., "1.2.0").
    Version() string

    // Init is called once when the pipeline loads (and again on config hot-reload).
    // Use it to validate configuration, open connections, and allocate resources.
    Init(config map[string]interface{}) error

    // ProcessRequest is invoked for every inbound client message before it
    // reaches the upstream MongoDB server.
    ProcessRequest(ctx *RequestContext) (*RequestResult, error)

    // ProcessResponse is invoked for every response coming back from the
    // upstream MongoDB server before it is forwarded to the client.
    ProcessResponse(ctx *ResponseContext) (*ResponseResult, error)

    // Close is called when the pipeline is torn down. Release resources here.
    Close() error
}

4. Legacy RequestContext

// RequestContext carries the full context of an inbound client request.
type RequestContext struct {
    // MessageBytes is the raw MongoDB wire-protocol message (header + body).
    // Modifying this slice directly is allowed — the proxy reads it back after
    // ProcessRequest returns.
    MessageBytes []byte

    // OpCode is the parsed wire-protocol operation code.
    // Common values: OpMsg (2013), OpQuery (2004), OpGetMore (2005).
    OpCode OpCode

    // Database is the target database name extracted from the message.
    Database string

    // Collection is the target collection name extracted from the message.
    Collection string

    // ConnectionID is the unique identifier for this client connection
    // within the proxy. Stable for the lifetime of the TCP session.
    ConnectionID uint64

    // ClientAddr is the remote address of the connected client (ip:port).
    ClientAddr string

    // Metadata is a key-value store shared across all steps in the pipeline
    // for the current request. Use it to pass data between steps.
    Metadata map[string]interface{}

    // StartTime is the timestamp when the proxy first received this message.
    StartTime time.Time

    // Logger is a structured logger pre-configured with connection and
    // request context fields.
    Logger *slog.Logger
}

5. Legacy ResponseContext

// ResponseContext carries the full context of an upstream server response.
type ResponseContext struct {
    // MessageBytes is the raw MongoDB wire-protocol response message.
    MessageBytes []byte

    // OpCode is the response operation code (typically OpMsg or OpReply).
    OpCode OpCode

    // RequestDuration is the time the upstream server took to produce
    // this response, measured from the moment the proxy forwarded the
    // request to when the first response byte arrived.
    RequestDuration time.Duration

    // UpstreamAddr is the address of the MongoDB server that handled
    // the request (ip:port). Useful in replica-set or sharded topologies.
    UpstreamAddr string

    // Metadata is the same shared key-value store from the request phase.
    // Values set during ProcessRequest are available here.
    Metadata map[string]interface{}

    // Logger is a structured logger with response-phase context.
    Logger *slog.Logger
}

6. Legacy actions and result types

// Action determines how the proxy proceeds after a step completes.
type Action int

const (
    // Continue passes the (possibly modified) message to the next step.
    Continue Action = iota

    // ShortCircuit skips all remaining steps in the pipeline and returns
    // the provided MessageBytes directly to the client (request phase) or
    // to the upstream (response phase).
    ShortCircuit

    // Reject halts the pipeline and returns a MongoDB error response to
    // the client with the specified error code and message.
    Reject
)

// RequestResult is returned by ProcessRequest.
type RequestResult struct {
    // Modified indicates whether MessageBytes was changed by this step.
    // The proxy uses this flag to decide if it needs to re-parse headers.
    Modified bool

    // MessageBytes is the (possibly modified) wire-protocol message.
    // If Modified is false, this field is ignored.
    MessageBytes []byte

    // Metadata contains any new or updated key-value pairs to merge
    // into the shared pipeline metadata.
    Metadata map[string]interface{}

    // Action tells the proxy how to proceed (Continue, ShortCircuit, Reject).
    Action Action

    // ErrorCode is used when Action == Reject. It becomes the MongoDB
    // error code in the error response sent to the client.
    ErrorCode int

    // ErrorMessage is the human-readable error text when Action == Reject.
    ErrorMessage string
}

// ResponseResult is returned by ProcessResponse.
type ResponseResult struct {
    // Modified indicates whether MessageBytes was changed.
    Modified bool

    // MessageBytes is the (possibly modified) response message.
    MessageBytes []byte

    // Metadata contains updated metadata key-value pairs.
    Metadata map[string]interface{}

    // Action tells the proxy how to proceed.
    Action Action

    // ErrorCode is used when Action == Reject.
    ErrorCode int

    // ErrorMessage is the error text when Action == Reject.
    ErrorMessage string
}

7. Legacy opcode constants

const (
    OpReply       OpCode = 1    // Legacy reply
    OpQuery       OpCode = 2004 // Legacy query
    OpGetMore     OpCode = 2005 // Legacy getMore
    OpCompressed  OpCode = 2012 // Compressed wrapper
    OpMsg         OpCode = 2013 // Modern message format (MongoDB 3.6+)
)

8. Legacy native Go rate limiter

package ratelimiter

import (
    "fmt"
    "sync"
    "time"

    sdk "github.com/nexo-proxy/plugin-sdk-go"
)

// RateLimiterStep limits the number of requests per client IP
// within a configurable time window.
type RateLimiterStep struct {
    maxRequests int
    windowSize  time.Duration

    mu       sync.Mutex
    counters map[string]*clientCounter
}

type clientCounter struct {
    count     int
    windowEnd time.Time
}

func (s *RateLimiterStep) Name() string    { return "acme-rate-limiter" }
func (s *RateLimiterStep) Version() string { return "1.0.0" }

func (s *RateLimiterStep) Init(config map[string]interface{}) error {
    maxReq, ok := config["max_requests"].(float64)
    if !ok || maxReq <= 0 {
        return fmt.Errorf("max_requests must be a positive number")
    }
    s.maxRequests = int(maxReq)

    windowSec, ok := config["window_seconds"].(float64)
    if !ok || windowSec <= 0 {
        return fmt.Errorf("window_seconds must be a positive number")
    }
    s.windowSize = time.Duration(windowSec) * time.Second

    s.counters = make(map[string]*clientCounter)
    return nil
}

func (s *RateLimiterStep) ProcessRequest(ctx *sdk.RequestContext) (*sdk.RequestResult, error) {
    s.mu.Lock()
    defer s.mu.Unlock()

    now := time.Now()
    client := ctx.ClientAddr

    counter, exists := s.counters[client]
    if !exists || now.After(counter.windowEnd) {
        s.counters[client] = &clientCounter{
            count:     1,
            windowEnd: now.Add(s.windowSize),
        }
        ctx.Logger.Debug("new rate-limit window",
            "client", client,
            "max_requests", s.maxRequests,
        )
        return &sdk.RequestResult{Action: sdk.Continue}, nil
    }

    counter.count++
    if counter.count > s.maxRequests {
        ctx.Logger.Warn("rate limit exceeded",
            "client", client,
            "count", counter.count,
            "limit", s.maxRequests,
        )
        return &sdk.RequestResult{
            Action:       sdk.Reject,
            ErrorCode:    16500,
            ErrorMessage: "rate limit exceeded, try again later",
        }, nil
    }

    ctx.Metadata["rate_limit_remaining"] = s.maxRequests - counter.count
    return &sdk.RequestResult{Action: sdk.Continue}, nil
}

func (s *RateLimiterStep) ProcessResponse(ctx *sdk.ResponseContext) (*sdk.ResponseResult, error) {
    // Pass responses through unmodified.
    return &sdk.ResponseResult{Action: sdk.Continue}, nil
}

func (s *RateLimiterStep) Close() error {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.counters = nil
    return nil
}

// Register the step with the SDK so Nexo can discover it.
func init() {
    sdk.Register(&RateLimiterStep{})
}

9. Legacy WASM exports

// nexo_init(config_ptr, config_len) -> i32
// nexo_process_request(msg_ptr, msg_len, meta_ptr, meta_len) -> i64
// nexo_process_response(msg_ptr, msg_len, meta_ptr, meta_len) -> i64

10. Legacy packed result encoding

func packResult(action uint32, newLen uint32) int64 {
    return int64(action)<<32 | int64(newLen)
}
// Example: Continue with 256-byte modified message → (0 << 32) | 256

11. Legacy host functions

// Host functions — import module: "nexo"

nexo_log(level i32, msg_ptr i32, msg_len i32)
// Writes a structured log message. level: 0=debug, 1=info, 2=warn, 3=error

nexo_get_config(key_ptr i32, key_len i32, val_ptr i32, val_len i32) -> i32
// Reads a config value by key. Returns byte length, or -1 if missing.

nexo_get_connection_info(buf_ptr i32, buf_len i32) -> i32
// Writes JSON connection info: client_addr, connection_id, tls_enabled, auth_user, etc.

nexo_set_metadata(key_ptr i32, key_len i32, val_ptr i32, val_len i32)
// Stores a key-value pair in pipeline metadata.

nexo_get_metadata(key_ptr i32, key_len i32, val_ptr i32, val_len i32) -> i32
// Retrieves a value from pipeline metadata. Returns byte length, or -1 if missing.

12. Legacy TinyGo WASM logger

//go:build tinygo.wasm
package main

import (
    "encoding/binary"
    "unsafe"
)

// ── Host function imports ─────────────────────────────────────────
//
//go:wasmimport nexo nexo_log
func nexoLog(level int32, msgPtr unsafe.Pointer, msgLen int32)

//go:wasmimport nexo nexo_get_config
func nexoGetConfig(keyPtr unsafe.Pointer, keyLen int32, valPtr unsafe.Pointer, valLen int32) int32

//go:wasmimport nexo nexo_set_metadata
func nexoSetMetadata(keyPtr unsafe.Pointer, keyLen int32, valPtr unsafe.Pointer, valLen int32)

//go:wasmimport nexo nexo_get_connection_info
func nexoGetConnectionInfo(bufPtr unsafe.Pointer, bufLen int32) int32

// ── Helper functions ──────────────────────────────────────────────

func logInfo(msg string) {
    if len(msg) > 0 {
        nexoLog(1, unsafe.Pointer(unsafe.StringData(msg)), int32(len(msg)))
    }
}

func logWarn(msg string) {
    if len(msg) > 0 {
        nexoLog(2, unsafe.Pointer(unsafe.StringData(msg)), int32(len(msg)))
    }
}

func logError(msg string) {
    if len(msg) > 0 {
        nexoLog(3, unsafe.Pointer(unsafe.StringData(msg)), int32(len(msg)))
    }
}

func getConfig(key string) (string, bool) {
    buf := make([]byte, 1024)
    n := nexoGetConfig(
        unsafe.Pointer(unsafe.StringData(key)), int32(len(key)),
        unsafe.Pointer(&buf[0]), int32(len(buf)),
    )
    if n < 0 {
        return "", false
    }
    return string(buf[:n]), true
}

func setMetadata(key, value string) {
    nexoSetMetadata(
        unsafe.Pointer(unsafe.StringData(key)), int32(len(key)),
        unsafe.Pointer(unsafe.StringData(value)), int32(len(value)),
    )
}

// ── Constants ─────────────────────────────────────────────────────

const (
    actionContinue     = 0
    actionShortCircuit = 1
    actionReject       = 2

    opReply      = 1
    opUpdate     = 2001
    opInsert     = 2002
    opQuery      = 2004
    opGetMore    = 2005
    opDelete     = 2006
    opCompressed = 2012
    opMsg        = 2013
)

func packResult(action uint32, newLen uint32) int64 {
    return int64(action)<<32 | int64(newLen)
}

// ── MongoDB wire-protocol header parsing ──────────────────────────

// Wire-protocol header: 16 bytes
//   [0:4]   messageLength  (int32, little-endian)
//   [4:8]   requestID      (int32)
//   [8:12]  responseTo     (int32)
//   [12:16] opCode         (int32)

func parseOpCode(msg []byte) int32 {
    if len(msg) < 16 {
        return -1
    }
    return int32(binary.LittleEndian.Uint32(msg[12:16]))
}

func opCodeName(code int32) string {
    switch code {
    case opReply:
        return "OP_REPLY"
    case opUpdate:
        return "OP_UPDATE"
    case opInsert:
        return "OP_INSERT"
    case opQuery:
        return "OP_QUERY"
    case opGetMore:
        return "OP_GETMORE"
    case opDelete:
        return "OP_DELETE"
    case opCompressed:
        return "OP_COMPRESSED"
    case opMsg:
        return "OP_MSG"
    default:
        return "UNKNOWN"
    }
}

// ── Exported WASM functions ───────────────────────────────────────

var pluginName string

//export nexo_init
func nexoInit(configPtr *byte, configLen int32) int32 {
    logInfo("initializing request-logger WASM plugin")

    name, ok := getConfig("plugin_name")
    if ok {
        pluginName = name
    } else {
        pluginName = "wasm-request-logger"
    }

    logInfo("plugin initialized: " + pluginName)
    return 0
}

//export nexo_process_request
func nexoProcessRequest(msgPtr *byte, msgLen int32, metaPtr *byte, metaLen int32) int64 {
    // Reconstruct the message slice from the raw pointer
    msg := unsafe.Slice(msgPtr, msgLen)

    opCode := parseOpCode(msg)
    name := opCodeName(opCode)

    logInfo("request: op=" + name + " len=" + itoa(int(msgLen)))

    // Set metadata so downstream steps can see what we observed
    setMetadata("last_op_code", name)

    // Pass the message through unmodified
    return packResult(actionContinue, 0)
}

//export nexo_process_response
func nexoProcessResponse(msgPtr *byte, msgLen int32, metaPtr *byte, metaLen int32) int64 {
    msg := unsafe.Slice(msgPtr, msgLen)

    opCode := parseOpCode(msg)
    name := opCodeName(opCode)

    logInfo("response: op=" + name + " len=" + itoa(int(msgLen)))

    return packResult(actionContinue, 0)
}

// Simple int-to-string without fmt (keeps WASM binary small)
func itoa(n int) string {
    if n == 0 {
        return "0"
    }
    buf := make([]byte, 0, 20)
    neg := false
    if n < 0 {
        neg = true
        n = -n
    }
    for n > 0 {
        buf = append(buf, byte('0'+n%10))
        n /= 10
    }
    if neg {
        buf = append(buf, '-')
    }
    // reverse
    for i, j := 0, len(buf)-1; i < j; i, j = i+1, j-1 {
        buf[i], buf[j] = buf[j], buf[i]
    }
    return string(buf)
}

func main() {} // required by TinyGo

13. Wire header inspection

// Wire-protocol message: 16-byte header + variable body
// Header: messageLength(4) | requestID(4) | responseTo(4) | opCode(4)
// Body depends on opCode: OP_MSG has flagBits + sections[],
//   OP_QUERY has flags + fullCollName + skip + limit + query doc

func inspectHeader(msg []byte) (msgLen, reqID, respTo int32, op sdk.OpCode) {
    msgLen = int32(binary.LittleEndian.Uint32(msg[0:4]))
    reqID  = int32(binary.LittleEndian.Uint32(msg[4:8]))
    respTo = int32(binary.LittleEndian.Uint32(msg[8:12]))
    op     = sdk.OpCode(binary.LittleEndian.Uint32(msg[12:16]))
    return
}

14. Opcode dispatch

switch ctx.OpCode {
case sdk.OpMsg:
    cmd, _ := bsonutil.ParseOpMsgCommand(ctx.MessageBytes)
    fmt.Println("Command:", cmd.CommandName)
case sdk.OpQuery:
    query, _ := bsonutil.ParseOpQuery(ctx.MessageBytes)
    fmt.Println("Collection:", query.FullCollectionName)
case sdk.OpCompressed:
    inner, _ := bsonutil.DecompressMessage(ctx.MessageBytes)
    fmt.Println("Inner opCode:", sdk.OpCode(binary.LittleEndian.Uint32(inner[12:16])))
}

15. Connection and timing fields

// Connection info available in RequestContext:
clientIP   := ctx.ClientAddr      // "10.0.0.5:54321"
connID     := ctx.ConnectionID    // 42
database   := ctx.Database        // "myapp"
collection := ctx.Collection      // "users"
startTime  := ctx.StartTime       // time the proxy received the message

// TLS and auth info via pipeline metadata (set by the proxy's auth step):
tlsEnabled := ctx.Metadata["tls_enabled"]    // true
authUser   := ctx.Metadata["auth_user"]      // "appuser"
authMech   := ctx.Metadata["auth_mechanism"] // "SCRAM-SHA-256"

// Timing — measure upstream latency in ProcessResponse:
ctx.Logger.Info("upstream response time",
    "upstream_ms", ctx.RequestDuration.Milliseconds(),
    "upstream_addr", ctx.UpstreamAddr,
)

16. Shared metadata example

// Step A sets metadata during request processing:
ctx.Metadata["request_category"] = "analytics"
ctx.Metadata["priority"] = 5

// Step B reads it later in the pipeline:
category, _ := ctx.Metadata["request_category"].(string)
priority, _ := ctx.Metadata["priority"].(int)

// ResponseContext also has access to request-phase metadata:
func (s *StepC) ProcessResponse(ctx *sdk.ResponseContext) (*sdk.ResponseResult, error) {
    category := ctx.Metadata["request_category"]
    // ...
}

17. Timing example

// In ProcessRequest — measure time since proxy received the message:
elapsed := time.Since(ctx.StartTime)
ctx.Logger.Info("request processing latency", "elapsed_us", elapsed.Microseconds())

// In ProcessResponse — use RequestDuration for upstream timing:
ctx.Logger.Info("upstream response time",
    "upstream_ms", ctx.RequestDuration.Milliseconds(),
    "upstream_addr", ctx.UpstreamAddr,
)

18. Pipeline definition

{
  "pipeline": "production-api",
  "steps": [
    {
      "plugin": "acme-rate-limiter",
      "version": "1.0.0",
      "config": {
        "max_requests": 1000,
        "window_seconds": 60,
        "exclude_ips": ["10.0.0.0/8"],
        "redis_url": "{{secret:redis-url}}"
      }
    },
    {
      "plugin": "acme-audit-trail",
      "version": "1.1.0",
      "config": {
        "audit_field": "_audit",
        "service_name": "api-gateway"
      }
    }
  ]
}

19. Configuration JSON Schema

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "object",
  "required": ["max_requests", "window_seconds"],
  "properties": {
    "max_requests": {
      "type": "integer", "minimum": 1, "maximum": 100000,
      "description": "Maximum requests per time window", "default": 1000
    },
    "window_seconds": {
      "type": "integer", "minimum": 1, "maximum": 86400,
      "description": "Time window in seconds", "default": 60
    },
    "exclude_ips": {
      "type": "array", "items": { "type": "string", "format": "ipv4-cidr" },
      "description": "CIDR ranges exempt from rate limiting", "default": []
    },
    "redis_url": {
      "type": "string", "format": "uri",
      "description": "Redis URL for distributed rate limiting",
      "x-nexo-secret": true
    }
  },
  "additionalProperties": false
}

20. Default handling

func (s *MyStep) Init(config map[string]interface{}) error {
    // Required field
    maxReq, ok := config["max_requests"].(float64)
    if !ok {
        return fmt.Errorf("max_requests is required")
    }
    s.maxRequests = int(maxReq)

    // Optional with default
    s.windowSize = 60 * time.Second
    if ws, ok := config["window_seconds"].(float64); ok {
        s.windowSize = time.Duration(ws) * time.Second
    }
    return nil
}

21. Hot reload example

func (s *MyStep) Init(config map[string]interface{}) error {
    newConfig, err := parseConfig(config)
    if err != nil {
        return fmt.Errorf("invalid config update: %w", err)
    }

    s.mu.Lock()
    defer s.mu.Unlock()

    // Close old resources if they changed
    if s.redisURL != newConfig.RedisURL && s.redisClient != nil {
        s.redisClient.Close()
    }

    s.maxRequests = newConfig.MaxRequests
    s.windowSize  = newConfig.WindowSize
    s.redisURL    = newConfig.RedisURL
    s.initialized = false // lazily re-initialize on next request
    return nil
}

22. Continue examples

// Pass through — no changes
return &sdk.RequestResult{Action: sdk.Continue}, nil

// Pass through with modifications
return &sdk.RequestResult{
    Action:       sdk.Continue,
    Modified:     true,
    MessageBytes: modifiedMsg,
}, nil

23. Short-circuit example

// Return a cached response directly to the client
cachedResponse, found := s.cache.Get(cacheKey)
if found {
    ctx.Logger.Info("cache hit", "key", cacheKey)
    return &sdk.RequestResult{
        Action:       sdk.ShortCircuit,
        MessageBytes: cachedResponse,
    }, nil
}

24. Reject example

// Reject with a MongoDB-compatible error
return &sdk.RequestResult{
    Action:       sdk.Reject,
    ErrorCode:    13,    // Unauthorized
    ErrorMessage: "access denied: insufficient permissions for collection " + ctx.Collection,
}, nil

25. Error/panic settings

// Returning an error halts the pipeline:
return nil, fmt.Errorf("step %s failed: %w", s.Name(), err)

// Panic recovery is configured per-step:
{
  "plugin": "acme-risky-plugin",
  "on_panic": "skip"  // "skip" (default) | "error" | "halt"
}

26. Error code list

const (
    ErrInternalError      = 1     // InternalError
    ErrUnauthorized       = 13    // Unauthorized
    ErrNamespaceNotFound  = 26    // NamespaceNotFound
    ErrExceededTimeLimit  = 50    // ExceededTimeLimit
    ErrCommandNotFound    = 59    // CommandNotFound

    // Nexo-specific error codes (16500-16599 reserved):
    ErrNexoRateLimit      = 16500 // Rate limit exceeded
    ErrNexoPolicyDenied   = 16501 // Policy violation
    ErrNexoPluginError    = 16502 // Plugin internal error
    ErrNexoConfigError    = 16503 // Plugin configuration error
    ErrNexoQuotaExceeded  = 16504 // Tenant quota exceeded
)

27. Go structured logging

// Log levels: Debug, Info, Warn, Error
ctx.Logger.Debug("parsing query document",
    "op_code", ctx.OpCode,
    "database", ctx.Database,
)

ctx.Logger.Info("query processed",
    "collection", ctx.Collection,
    "duration_ms", elapsed.Milliseconds(),
    "doc_count", count,
)

ctx.Logger.Warn("slow query detected",
    "collection", ctx.Collection,
    "duration_ms", elapsed.Milliseconds(),
    "threshold_ms", s.slowQueryThreshold.Milliseconds(),
)

ctx.Logger.Error("failed to parse BSON document",
    "error", err,
    "offset", offset,
    "msg_len", len(ctx.MessageBytes),
)

28. WASM logging

// TinyGo example:
logAtLevel(1, "info: request processed successfully")
logAtLevel(2, "warn: approaching rate limit threshold")

// For structured fields in WASM, encode as JSON in the message:
logAtLevel(1, \`{"msg":"query processed","collection":"users","duration_ms":12}\`)

29. Mock contexts

import (
    sdk "github.com/nexo-proxy/plugin-sdk-go"
    "github.com/nexo-proxy/plugin-sdk-go/sdktest"
)

// MockRequestContext — build a realistic request context for testing
ctx := sdktest.NewMockRequestContext(
    sdktest.WithOpMsg("find", "mydb", "users", bson.M{"age": 25}),
    sdktest.WithClientAddr("10.0.0.1:54321"),
    sdktest.WithConnectionID(42),
    sdktest.WithMetadata(map[string]interface{}{"auth_user": "testuser"}),
    sdktest.WithStartTime(time.Now()),
)

// MockResponseContext — build a response context
respCtx := sdktest.NewMockResponseContext(
    sdktest.WithResponseOpMsg("find", bson.A{
        bson.M{"_id": 1, "name": "Alice", "age": 25},
        bson.M{"_id": 2, "name": "Bob", "age": 25},
    }),
    sdktest.WithUpstreamAddr("mongo-primary.internal:27017"),
    sdktest.WithRequestDuration(5 * time.Millisecond),
    sdktest.WithResponseMetadata(map[string]interface{}{"request_category": "analytics"}),
)

30. Full test suite

package ratelimiter_test

import (
    "testing"
    sdk "github.com/nexo-proxy/plugin-sdk-go"
    "github.com/nexo-proxy/plugin-sdk-go/sdktest"
    "github.com/stretchr/testify/assert"
    "github.com/stretchr/testify/require"
    "go.mongodb.org/mongo-driver/bson"
)

func TestMyStep_ProcessRequest(t *testing.T) {
    step := &MyStep{}
    err := step.Init(map[string]interface{}{"key": "value"})
    require.NoError(t, err)

    ctx := sdktest.NewMockRequestContext(
        sdktest.WithOpMsg("find", "mydb", "users", bson.M{"age": 25}),
        sdktest.WithClientAddr("10.0.0.1:54321"),
    )

    result, err := step.ProcessRequest(ctx)
    require.NoError(t, err)
    assert.Equal(t, sdk.Continue, result.Action)
}

func TestRateLimiter_AllowsUnderLimit(t *testing.T) {
    step := &RateLimiterStep{}
    err := step.Init(map[string]interface{}{
        "max_requests":   float64(5),
        "window_seconds": float64(60),
    })
    require.NoError(t, err)

    ctx := sdktest.NewMockRequestContext(
        sdktest.WithOpMsg("find", "testdb", "users", bson.M{}),
        sdktest.WithClientAddr("10.0.0.1:54321"),
    )

    for i := 0; i < 5; i++ {
        result, err := step.ProcessRequest(ctx)
        require.NoError(t, err)
        assert.Equal(t, sdk.Continue, result.Action)
    }
}

func TestRateLimiter_RejectsOverLimit(t *testing.T) {
    step := &RateLimiterStep{}
    err := step.Init(map[string]interface{}{
        "max_requests":   float64(2),
        "window_seconds": float64(60),
    })
    require.NoError(t, err)

    ctx := sdktest.NewMockRequestContext(
        sdktest.WithOpMsg("find", "testdb", "users", bson.M{}),
        sdktest.WithClientAddr("10.0.0.1:54321"),
    )

    // First two pass, third rejected
    for i := 0; i < 2; i++ {
        result, _ := step.ProcessRequest(ctx)
        assert.Equal(t, sdk.Continue, result.Action)
    }
    result, err := step.ProcessRequest(ctx)
    require.NoError(t, err)
    assert.Equal(t, sdk.Reject, result.Action)
    assert.Equal(t, 16500, result.ErrorCode)
}

func TestRateLimiter_DifferentClients(t *testing.T) {
    step := &RateLimiterStep{}
    _ = step.Init(map[string]interface{}{
        "max_requests": float64(1), "window_seconds": float64(60),
    })

    ctxA := sdktest.NewMockRequestContext(
        sdktest.WithOpMsg("find", "testdb", "users", bson.M{}),
        sdktest.WithClientAddr("10.0.0.1:54321"),
    )
    result, _ := step.ProcessRequest(ctxA)
    assert.Equal(t, sdk.Continue, result.Action)

    ctxB := sdktest.NewMockRequestContext(
        sdktest.WithOpMsg("find", "testdb", "users", bson.M{}),
        sdktest.WithClientAddr("10.0.0.2:54321"),
    )
    result, _ = step.ProcessRequest(ctxB)
    assert.Equal(t, sdk.Continue, result.Action) // different IP, allowed
}

func TestRateLimiter_InvalidConfig(t *testing.T) {
    step := &RateLimiterStep{}
    assert.Error(t, step.Init(map[string]interface{}{"window_seconds": float64(60)}))
    assert.Error(t, step.Init(map[string]interface{}{"max_requests": float64(-1), "window_seconds": float64(60)}))
    assert.Error(t, step.Init(map[string]interface{}{"max_requests": float64(100)}))
}

31. Allocation reuse

// Go: Use sync.Pool for frequently allocated objects
var bufferPool = sync.Pool{
    New: func() interface{} {
        buf := make([]byte, 0, 4096)
        return &buf
    },
}

func (s *MyStep) ProcessRequest(ctx *sdk.RequestContext) (*sdk.RequestResult, error) {
    bufPtr := bufferPool.Get().(*[]byte)
    buf := (*bufPtr)[:0] // reset length, keep capacity
    defer bufferPool.Put(bufPtr)

    // Use buf for temporary work...
    buf = append(buf, ctx.MessageBytes...)
    // ... modify buf ...

    return &sdk.RequestResult{
        Action:       sdk.Continue,
        Modified:     true,
        MessageBytes: append([]byte(nil), buf...), // copy out before returning to pool
    }, nil
}

// WASM: Pre-allocate buffers at init time
var (
    workBuffer    [65536]byte // 64KB work buffer
    metadataBuf   [4096]byte  // 4KB metadata buffer
    configBuf     [1024]byte  // 1KB config buffer
)

32. Header-only parsing

// FAST: Header-only inspection (16 bytes) — no BSON parsing
func shouldProcess(msg []byte) bool {
    if len(msg) < 16 { return false }
    return sdk.OpCode(binary.LittleEndian.Uint32(msg[12:16])) == sdk.OpMsg
}

// FAST: Extract command name from OP_MSG without full BSON parse
func getCommandName(msg []byte) (string, error) {
    if len(msg) < 25 { return "", fmt.Errorf("message too short") }
    return bsonutil.ReadFirstKey(msg[21:])
}

// SLOW: Full BSON parse — avoid unless you need document contents
func getFullDocument(msg []byte) (bson.M, error) {
    return bsonutil.ParseOpMsgBody(msg)
}

33. Cached authorization

func (s *AuthzStep) ProcessRequest(ctx *sdk.RequestContext) (*sdk.RequestResult, error) {
    authUser, _ := ctx.Metadata["auth_user"].(string)
    cacheKey := fmt.Sprintf("%s:%s:%s", authUser, ctx.Database, ctx.Collection)

    // Check local cache first instead of calling authz service per-request
    allowed, found := s.cache.Get(cacheKey)
    if !found {
        var err error
        allowed, err = s.authzClient.Check(ctx.Database, ctx.Collection, authUser)
        if err != nil {
            return nil, err
        }
        s.cache.Set(cacheKey, allowed, 5*time.Minute)
    }

    if !allowed.(bool) {
        return &sdk.RequestResult{
            Action: sdk.Reject, ErrorCode: 13, ErrorMessage: "unauthorized",
        }, nil
    }
    return &sdk.RequestResult{Action: sdk.Continue}, nil
}

34. Benchmark example

func BenchmarkRateLimiter_ProcessRequest(b *testing.B) {
    step := &RateLimiterStep{}
    _ = step.Init(map[string]interface{}{
        "max_requests":   float64(1000000),
        "window_seconds": float64(60),
    })

    ctx := sdktest.NewMockRequestContext(
        sdktest.WithOpMsg("find", "testdb", "users", bson.M{"age": 25}),
        sdktest.WithClientAddr("10.0.0.1:54321"),
    )

    b.ResetTimer()
    b.ReportAllocs()
    for i := 0; i < b.N; i++ {
        _, _ = step.ProcessRequest(ctx)
    }
}

// Use sdktest.BenchmarkStep for a comprehensive suite with multiple
// request types, configurable client counts, and automatic reporting:
sdktest.BenchmarkStep(b, step, sdktest.BenchmarkOptions{
    RequestGenerators: []sdktest.RequestGenerator{
        sdktest.OpMsgFind("testdb", "users", bson.M{"age": 25}),
        sdktest.OpMsgInsert("testdb", "events", bson.M{"type": "click"}),
    },
    UniqueClients:     100,
    BenchmarkRequest:  true,
    BenchmarkResponse: true,
})

// Target performance:
//   < 1μs   per ProcessRequest — simple plugins (header inspection)
//   < 10μs  per ProcessRequest — plugins doing BSON parsing
//   > 100μs per ProcessRequest — needs optimization

35. Example repository commands

# Clone the examples repository
git clone https://github.com/nexo-proxy/plugin-examples.git && cd plugin-examples

make test        # Build and test all examples
make bench       # Run benchmarks
cd rate-limiter && go build ./...                                  # Build Go example
cd wasm-request-logger && tinygo build -o plugin.wasm -target wasi main.go  # Build WASM

36. Legacy publishing commands

# Quick-start publishing commands:

# 1. Initialize the plugin manifest
nexo plugin init --name my-plugin --type go

# 2. Validate the manifest and config schema
nexo plugin validate

# 3. Run the marketplace compliance checks
nexo plugin check --marketplace

# 4. Build and package
nexo plugin build --output my-plugin-1.0.0.tar.gz

# 5. Publish to the marketplace
nexo plugin publish --version 1.0.0

# For WASM plugins, add the --wasm flag:
nexo plugin build --wasm --output my-plugin-1.0.0.wasm
nexo plugin publish --wasm --version 1.0.0