Embedded Library Guide¶

Crystal can be used as an embedded graph database in any Go application, similar to how SQLite is used for relational data.

Installation¶

go get crystal

Opening a Database¶

package main

import "crystal/pkg/crystal"

func main() {
    db, err := crystal.Open(crystal.Options{
        StorageType: "disk",
        DataDir:     "./data",
        SyncOnWrite: false,
    })
    if err != nil {
        panic(err)
    }
    defer db.Close()
}

Storage Types¶

Type	Description
`disk`	Persistent JSON file storage (default)
`memory`	In-memory only, data lost on close

Options¶

Field	Type	Default	Description
StorageType	string	`"disk"`	Storage backend
DataDir	string	`"./data"`	Directory for data files
SyncOnWrite	bool	`false`	Full snapshot flush on every write (rarely needed)
FlushInterval	time.Duration	`0`	Auto-checkpoint interval (0 = disabled)

CRUD Operations¶

Vertices¶

v, err := db.AddVertex("Person", crystal.Properties{
    "name": "Alice",
    "age":  30,
})

v, err := db.GetVertex(id)

v.Properties["age"] = 31
err := db.UpdateVertex(v)

err := db.DeleteVertex(id)

Edges¶

e, err := db.AddEdge(sourceID, targetID, "knows", crystal.Properties{
    "since": "2020",
})

e, err := db.GetEdge(id)

e.Weight = 2.5
err := db.UpdateEdge(e)

err := db.DeleteEdge(id)

Listing¶

vertices, err := db.AllVertices()
edges, err := db.AllEdges()

Neighbors¶

neighbors, err := db.Neighbors(vertexID, crystal.DirectionOut)
neighbors, err := db.Neighbors(vertexID, crystal.DirectionIn)
neighbors, err := db.Neighbors(vertexID, crystal.DirectionBoth)

Queries¶

The query builder provides a fluent API for filtering and paginating results.

Vertex Queries¶

q := db.QueryVertices().
    OfType("Person").
    WithLabel("suspect").
    PropertyEquals("city", "Berlin").
    Limit(10).
    Offset(0)

result, err := db.Execute(q)
for _, v := range result.Vertices {
    fmt.Println(v.Properties["name"])
}

Custom Filters¶

q := db.QueryVertices().
    OfType("Person").
    Where(func(props crystal.Properties) bool {
        age, ok := props.GetInt("age")
        return ok && age > 25
    })

result, err := db.Execute(q)

Edge Queries¶

q := db.QueryEdges().
    OfType("knows").
    HasProperty("since")

result, err := db.Execute(q)

Neighbor Queries¶

q := db.QueryFrom(vertexID).
    Out().
    OfType("Server").
    Depth(2).
    Limit(50)

result, err := db.Execute(q)

Traversal¶

BFS¶

err := db.TraverseBFS(startID, 3, func(v *crystal.Vertex, depth int) bool {
    fmt.Printf("depth %d: %s (%s)\n", depth, v.ID, v.Type)
    return true // return false to stop
})

DFS¶

err := db.TraverseDFS(startID, 5, func(v *crystal.Vertex, depth int) bool {
    fmt.Printf("found: %s\n", v.Properties["name"])
    return true
})

Shortest Path¶

path, err := db.ShortestPath(fromID, toID)
if err != nil {
    fmt.Println("no path found")
    return
}
fmt.Printf("path length: %d edges\n", len(path.Edges))
for _, v := range path.Vertices {
    fmt.Println(v.ID, v.Type)
}

Statistics¶

stats := db.Stats()
fmt.Printf("vertices: %d, edges: %d\n", stats.VertexCount, stats.EdgeCount)

Persistence and Durability¶

The disk backend uses a Write-Ahead Log (WAL). Every mutation is appended to the WAL and fsynced before being applied in memory, so acknowledged writes survive crashes.

Set FlushInterval to enable automatic checkpointing (snapshot + WAL truncation):

db, err := crystal.Open(crystal.Options{
    StorageType:   "disk",
    DataDir:       "./data",
    FlushInterval: 30 * time.Second,
})

You can also trigger a manual checkpoint at any time:

err := db.Flush()

Close() automatically stops the checkpointer and writes a final snapshot.

Investigation Mapping Example¶

db, _ := crystal.Open(crystal.DefaultOptions())
defer db.Close()

alice, _ := db.AddVertex("Person", crystal.Properties{"name": "Alice"})
bob, _ := db.AddVertex("Person", crystal.Properties{"name": "Bob"})
server, _ := db.AddVertex("Server", crystal.Properties{"ip": "10.0.0.1"})
domain, _ := db.AddVertex("Domain", crystal.Properties{"fqdn": "example.com"})

db.AddEdge(alice.ID, server.ID, "accessed", crystal.Properties{"port": 443})
db.AddEdge(bob.ID, server.ID, "accessed", crystal.Properties{"port": 22})
db.AddEdge(domain.ID, server.ID, "resolves_to", nil)

q := db.QueryFrom(server.ID).In().OfType("Person").Depth(1)
result, _ := db.Execute(q)
for _, v := range result.Vertices {
    fmt.Println("accessed by:", v.Properties["name"])
}

Dialogue Tree Example¶

db, _ := crystal.Open(crystal.Options{StorageType: "memory"})
defer db.Close()

start, _ := db.AddVertex("DialogueNode", crystal.Properties{
    "speaker": "NPC",
    "text":    "Hello traveler, what brings you here?",
})
opt1, _ := db.AddVertex("DialogueNode", crystal.Properties{
    "speaker": "Player",
    "text":    "I'm looking for the ancient temple.",
})
opt2, _ := db.AddVertex("DialogueNode", crystal.Properties{
    "speaker": "Player",
    "text":    "Just passing through.",
})
resp1, _ := db.AddVertex("DialogueNode", crystal.Properties{
    "speaker": "NPC",
    "text":    "Ah, the temple lies to the north.",
})

db.AddEdge(start.ID, opt1.ID, "choice", crystal.Properties{"order": 1})
db.AddEdge(start.ID, opt2.ID, "choice", crystal.Properties{"order": 2})
db.AddEdge(opt1.ID, resp1.ID, "leads_to", nil)

choices, _ := db.Neighbors(start.ID, crystal.DirectionOut)
for _, c := range choices {
    fmt.Println("->", c.Properties["text"])
}