SapixDBSapixDB/Docs
Home
Go SDK

Go SDK

✓ Shipped

Official Go client for SapixDB. Zero external dependencies — uses only the standard library. Context-aware, goroutine-safe, Go 1.21+.

Installation

shell
go get github.com/sapixdb/sapixdb-go

Then import the package in your code:

Go
import sapixdb "github.com/sapixdb/sapixdb-go"

Quick Start

Go
package main

import (
    "context"
    "fmt"
    "time"

    sapixdb "github.com/sapixdb/sapixdb-go"
)

func main() {
    ctx := context.Background()

    db := sapixdb.New(sapixdb.Config{
        URL:   "http://localhost:7475",
        Agent: "my-app",
    })

    // Check connection
    fmt.Println(db.Ping(ctx)) // true

    // Write a record
    record, err := db.Collection("products").Write(ctx, map[string]any{
        "name":  "Classic T-Shirt",
        "price": 29.99,
        "stock": 100,
    })
    if err != nil {
        panic(err)
    }
    fmt.Println(record.ID)   // "nuc_abc123"
    fmt.Println(record.Hash) // "sha3:e7f2a1..."  — cryptographic proof

    // Read latest records
    products, _ := db.Collection("products").Latest(ctx, 20)

    // Structured SaQL query
    result, _ := db.Collection("products").Query(ctx, map[string]any{
        "type":  "latest",
        "limit": 20,
    })

    // Time travel — records written before a given HLC timestamp
    cutoff := uint64(time.Now().Add(-24 * time.Hour).UTC().UnixMilli())
    snapshot, _ := db.Collection("orders").Query(ctx, map[string]any{
        "type":    "time_range",
        "from_ts": uint64(0),
        "to_ts":   cutoff,
    })

    _ = products
    _ = result
    _ = snapshot
}

Client Configuration

Go
db := sapixdb.New(sapixdb.Config{
    URL:   "http://localhost:7475",   // SapixDB agent URL (required)
    Agent: "my-app",                  // agent ID (required)
    Headers: map[string]string{       // extra headers (optional)
        "X-Api-Key": "secret",
    },
    Timeout: 30 * time.Second,        // default: 10s
})

Collection API

.Write(ctx, data)→ *WriteResponse, error
Append a new record. Returns WriteResponse with ID, Hash, PrevHash, Timestamp. Nothing is ever overwritten — every write is permanent.
.WriteBatch(ctx, records)→ []*WriteResponse, error
Write multiple records sequentially. Returns results in order; stops on first error.
.Get(ctx, contentHash)→ *RecordView, error
Fetch by content hash (hex string). Returns *SapixNotFoundError if missing.
.Latest(ctx, limit)→ *QueryResult, error
The most recent limit records. Pass 0 for the default (20).
.Query(ctx, body)→ *QueryResult, error
Execute a structured SaQL query. body must have a "type" key — e.g. {"type":"latest","limit":10}, {"type":"hash","content_hash":"e7f2..."}, or {"type":"time_range","from_ts":0,"to_ts":1704067200000}.
.Head(ctx)→ *ChainHeadResponse, error
Current chain head hash and total record count for the agent.
.Status(ctx)→ *AgentStatusResponse, error
Agent status — record count and chain head in one call.

Time Travel

Use Query with type: "time_range" to read records written within a HLC timestamp window. To read the strand as it existed at a past moment, set to_tsto that moment's Unix milliseconds.

Go
import "time"

// Records written more than 30 minutes ago
cutoff := uint64(time.Now().Add(-30 * time.Minute).UTC().UnixMilli())

result, err := db.Collection("orders").Query(ctx, map[string]any{
    "type":    "time_range",
    "from_ts": uint64(0),
    "to_ts":   cutoff,
})
if err != nil {
    panic(err)
}
for _, r := range result.Records {
    fmt.Println(r.Payload["status"])
}

// Alternatively, use the REST endpoint directly:
// GET /v1/strand/as-of?ts=2026-05-01T15:30:00Z

Graph Relationships

db.Graph.Relate(ctx, src, dst, edgeType)→ error
Create a typed directed edge with weight 1.0. Shorthand for AddEdge.
db.Graph.AddEdge(ctx, src, dst, edgeType, weight)→ error
Create a directed edge with an explicit weight.
db.Graph.Edges(ctx, agentID)→ []GraphEdge, error
All outbound edges from an agent.
db.Graph.InboundEdges(ctx, agentID)→ []GraphEdge, error
All inbound edges to an agent.
db.Graph.Traverse(ctx, agentID, TraverseOptions)→ *TraversalResult, error
Walk the graph from agentID. TraverseOptions.Depth (default 1, max 3), TraverseOptions.EdgeType (optional filter). Returns .Nodes and .Edges.
db.Graph.RemoveEdge(ctx, src, edgeType, dst)→ error
Delete a directed edge.
db.Graph.AddRecordRef(ctx, srcAgent, srcHash, edgeType, dstAgent, dstHash)→ error
Create a cross-agent record reference edge.
Go
// Link order → customer
_ = db.Graph.Relate(ctx, order.ID, customer.ID, "placed_by")
_ = db.Graph.Relate(ctx, order.ID, product.ID, "contains")

// List direct edges from a node
edges, _ := db.Graph.Edges(ctx, order.ID)
for _, e := range edges {
    fmt.Printf("%s -[%s]-> %s\n", e.Src, e.EdgeType, e.Dst)
}

// Walk the graph (depth 2)
result, err := db.Graph.Traverse(ctx, customer.ID, sapixdb.TraverseOptions{
    Depth: 2,
})
fmt.Println(result.Nodes) // []NodeView
fmt.Println(result.Edges) // []GraphEdge

Agent Ingest

Go — log every AI decision
// Log AI agent decisions permanently and immutably
_, err := db.Ingest(ctx, "ai_decisions", map[string]any{
    "model":      "gpt-4o",
    "action":     "approve_loan",
    "confidence": 0.94,
    "applicant":  "cust_001",
    "reasoning":  "Credit score 780, DTI 28%",
})
// Every decision is cryptographically signed — you can always prove
// what the AI decided, when, and why.

Error Handling

All errors are typed and can be inspected with errors.As.

Go
import "errors"

record, err := db.Collection("orders").Get(ctx, "nuc_missing")
if err != nil {
    var notFound *sapixdb.SapixNotFoundError
    var netErr   *sapixdb.SapixNetworkError
    var sapixErr *sapixdb.SapixError

    switch {
    case errors.As(err, &notFound):
        fmt.Println("not found:", notFound.RecordID)

    case errors.As(err, &netErr):
        fmt.Println("network error:", netErr.Cause)

    case errors.As(err, &sapixErr):
        fmt.Printf("error %d: %s\n", sapixErr.Status, sapixErr.Message)
    }
}

Full Example: Online Store

Go — main.go
package main

import (
    "context"
    "fmt"
    "time"

    sapixdb "github.com/sapixdb/sapixdb-go"
)

func main() {
    ctx := context.Background()
    db := sapixdb.New(sapixdb.Config{
        URL:   "http://localhost:7475",
        Agent: "store",
    })

    // 1. Add product
    shirt, _ := db.Collection("products").Write(ctx, map[string]any{
        "sku": "SHIRT-001", "name": "Classic T-Shirt",
        "price": 29.99, "stock": 200, "category": "apparel",
    })

    // 2. Register customer
    customer, _ := db.Collection("customers").Write(ctx, map[string]any{
        "name": "Alice Johnson", "email": "alice@example.com",
    })

    // 3. Place order
    order, _ := db.Collection("orders").Write(ctx, map[string]any{
        "customer_id": customer.ID,
        "items": []any{map[string]any{
            "product_id": shirt.ID, "qty": 2, "unit_price": 29.99,
        }},
        "total":  59.98,
        "status": "placed",
    })

    // 4. Link in graph
    _ = db.Graph.Relate(ctx, order.ID, customer.ID, "placed_by", 1.0)
    _ = db.Graph.Relate(ctx, order.ID, shirt.ID, "contains", 1.0)

    // 5. Ship (append — "placed" version is preserved forever)
    _, _ = db.Collection("orders").Write(ctx, map[string]any{
        "customer_id": customer.ID,
        "status":      "shipped",
        "tracking":    "UPS-1Z999AA10123456784",
    })

    // 6. Audit: read the strand up to the moment the order was placed
    placedAtMs := uint64(order.TimestampHLC)
    history, _ := db.Collection("orders").Query(ctx, map[string]any{
        "type":    "time_range",
        "from_ts": uint64(0),
        "to_ts":   placedAtMs,
    })
    if len(history.Records) > 0 {
        fmt.Println(history.Records[len(history.Records)-1].Payload["status"]) // "placed"
    }

    _ = time.Second // import kept for Timeout example
}

Realtime Subscriptions (SSE)

Subscribe to live writes on any agent using Server-Sent Events. Subscribe and SubscribeGlobal return a *Subscription with a buffered Events channel. Call Close() to cancel and drain the channel.

.Subscribe(ctx, agentID, opts)→ (*Subscription, error)
Open an SSE stream for a single named agent. Supports opts.Since (HLC backfill) and opts.Filter (field presence filter).
.SubscribeGlobal(ctx, opts)→ (*Subscription, error)
Open an SSE stream for all agents. opts.Agents limits the stream to a subset.
Go — per-agent stream
sub, err := db.Subscribe(ctx, "orders", sapixdb.SubscribeOptions{})
if err != nil {
    log.Fatal(err)
}
defer sub.Close()

for event := range sub.Events {
    fmt.Printf("agent=%s record=%s\n", event.AgentID, event.RecordID)
    // event.Payload is json.RawMessage — unmarshal as needed
    var payload map[string]any
    _ = json.Unmarshal(event.Payload, &payload)
    fmt.Println(payload)
}
Go — global stream with backfill + field filter
since := uint64(1748304000000) // HLC timestamp — replay from here first
sub, err := db.SubscribeGlobal(ctx, sapixdb.SubscribeOptions{
    Since:  &since,
    Filter: "risk_score",         // only events whose payload has this field
    Agents: []string{"transactions", "fraud_checks"},
})
if err != nil {
    log.Fatal(err)
}
defer sub.Close()

for event := range sub.Events {
    var payload struct {
        RiskScore float64 `json:"risk_score"`
    }
    _ = json.Unmarshal(event.Payload, &payload)
    if payload.RiskScore >= 0.8 {
        alertFraud(event)
    }
}
Go — cancel after first event
ctx, cancel := context.WithCancel(context.Background())
defer cancel()

sub, _ := db.Subscribe(ctx, "orders", sapixdb.SubscribeOptions{})
defer sub.Close()

event := <-sub.Events
fmt.Println("first write:", event.RecordID)
// cancel() triggers Close() — or call sub.Close() explicitly
Also available: JavaScript / TypeScript and Python SDKs

npm install sapixdb and pip install sapixdb — same API, every language.