Schema Architect

Schema Architect

Production-ready database schema design for SQLite + Redis + Neo4j with Rust and Go.

Workflow

Schema design involves these steps:

1. Gather requirements (stage, databases, language, domain) 2. Select architecture pattern 3. Generate schemas per database 4. Generate language bindings 5. Generate migrations and validation 6. Validate output (run validate_schema.py)

Step 1: Gather Requirements

Determine these before generating anything:

If the user provides a use case without specifying these, infer sensible defaults and confirm.

Step 2: Select Architecture Pattern

Early stage — optimized for speed, simplicity, iteration:

SQLite ──── primary store (OLTP, local-first, embedded)
Redis  ──── session cache + rate limiting + pub/sub events
Neo4j  ──── relationship queries only when graph is justified

Advanced stage — optimized for scale, observability, governance:

SQLite ──── edge/embedded nodes, local write-ahead, sync buffer
Redis  ──── distributed cache, streams pipeline, leaderboards
Neo4j  ──── knowledge graph, access control graph, recommendation engine

For integrated multi-DB architectures, read references/integration-patterns.md.

Step 3: Generate Schemas

For each selected database, read the corresponding reference and apply its patterns:

• SQLite: Read references/sqlite.md — normalization, WAL mode, strict typing, indexes
• Redis: Read references/redis.md — key namespacing, TTL policies, cache-aside pattern
• Neo4j: Read references/neo4j.md — node labels, relationship types, Cypher constraints

Apply naming conventions from references/naming-conventions.md to ALL generated schemas.

Output per database

Use templates from templates/ as starting points — fill in entity-specific content.

Step 4: Generate Language Bindings

For each selected language, read the corresponding reference:

• Rust: Read references/rust-bindings.md — sqlx/diesel/sea-orm, redis-rs, neo4rs
• Go: Read references/go-bindings.md — database/sql/gorm, go-redis, neo4j-go-driver

Generate type-safe model structs with proper derives/tags, connection helpers, and repository patterns.

Output per language

Step 5: Generate Migrations and Validation

Generate versioned migration files following references/migration-patterns.md:

• Timestamped filenames: YYYYMMDDHHMMSS_description.sql
• Every migration has an up and down section
• Include audit trail columns: created_at, updated_at, version
• For advanced stage: add created_by, deleted_at (soft delete), tenant_id (multi-tenancy)

Step 6: Validate

Run the validation script against all generated files:

python3 /home/ubuntu/skills/schema-architect/scripts/validate_schema.py <output_directory>

The script checks naming conventions, foreign key consistency, index coverage, and migration ordering. Fix any reported issues before delivering.

Enterprise Patterns by Stage

Early Stage Patterns

Apply these for MVPs, prototypes, and startups:

Advanced Stage Patterns

Apply these for production systems at scale:

Key Principles

These rules apply to ALL generated schemas regardless of database or stage:

1. Normalize to 3NF minimum — denormalize only with measured justification 2. Every table gets a primary key — prefer INTEGER PRIMARY KEY (SQLite) or UUIDs (distributed) 3. Every foreign key gets an index — no exceptions 4. Timestamps on everything — created_at and updated_at with UTC, never local time 5. Constraints at the DB level — NOT NULL, CHECK, UNIQUE enforced in schema, not just app code 6. Redis keys are namespaced — {service}:{entity}:{id}:{field} pattern always 7. Neo4j relationships are verbs — FOLLOWS, PURCHASED, BELONGS_TO, never nouns 8. Migrations are immutable — never edit a deployed migration, always create a new one 9. Document everything — inline SQL comments on non-obvious columns, README per schema dir

schema-architect

Design and generate production-ready database schemas for SQLite, Redis, and Neo4j. Includes type-safe Rust and Go bindings, versioned migrations, a validation script, and enterprise patterns for both early-stage and scaled systems.

Works with Claude Code, Cursor, Codex CLI, and any agent that supports the agentskills protocol.

Install

npx skills add OthmanAdi/schema-architect -g

Then just describe what you're building. The skill picks up on phrases like "design a schema", "set up Redis caching", "generate ORM models in Rust", or "architect a multi-database system" and takes it from there.

What it does

You tell it your domain, stack, and where you are in the project (MVP or production-scale). It asks four questions:

• Stage: early or advanced
• Databases: any combination of SQLite, Redis, Neo4j
• Language: Rust, Go, or both
• Domain: what the app actually does

Then it generates everything: SQL schemas, Redis key configs, Cypher constraints, language bindings, and versioned migrations. A Python validation script checks the output for naming inconsistencies, missing indexes, and migration ordering problems.

What gets generated

For SQLite:

• schema.sql with normalized tables, FK indexes, audit columns
• Versioned migration files in migrations/ with up and down sections

For Redis:

• redis-schema.toml with namespaced keys and TTL policies

For Neo4j:

• constraints.cypher and schema.cypher

For Rust:

• models.rs, db.rs, cache.rs, graph.rs using sqlx, redis-rs, neo4rs

For Go:

• models.go, db.go, cache.go, graph.go using database/sql, go-redis, neo4j-go-driver

Example

"Design a schema for a SaaS task management app. SQLite and Redis, Rust bindings, early stage."

The skill clarifies the domain, confirms defaults, then generates normalized tables with FK indexes and audit columns, a Redis key structure for sessions and rate limiting, connection pool setup, and a repository pattern in Rust. It runs the validator and reports any issues before finishing.

Two modes

Early stage uses SQLite as the primary store with Redis for sessions and rate limiting. Neo4j only gets added when the data actually has graph characteristics. The output is simple, embeddable, and easy to run locally.

Advanced stage adds multi-tenancy, CQRS, event sourcing via Redis Streams, graph access control in Neo4j, and audit trails. More files, more structure, built for teams.

Files in this repo

schema-architect/
  SKILL.md
  scripts/
    validate_schema.py
  references/
    sqlite.md
    redis.md
    neo4j.md
    rust-bindings.md
    go-bindings.md
    naming-conventions.md
    migration-patterns.md
    integration-patterns.md
  templates/
    sqlite-migration.sql.tmpl
    redis-schema.toml.tmpl
    neo4j-constraints.cypher.tmpl
    rust-model.rs.tmpl
    go-model.go.tmpl

The skill reads reference files on demand, so it doesn't load everything into context upfront. It pulls sqlite.md when it's working on SQLite, rust-bindings.md when generating Rust code, and so on.

Rules enforced on all generated output

• Tables normalized to 3NF minimum
• Every foreign key gets an index
• created_at and updated_at on every table, UTC only
• Redis keys namespaced as {service}:{entity}:{id}:{field}
• Neo4j relationships are verbs: FOLLOWS, PURCHASED, BELONGS_TO
• Migrations are immutable. Never edit a deployed one.

Author

Ahmad Othman Ammar Adi -- OthmanAdi


Go Database Bindings

Table of Contents

1. SQLite with database/sql + modernc 2. Redis with go-redis 3. Neo4j with neo4j-go-driver 4. Model Patterns 5. Error Handling 6. Connection Management

1. SQLite with database/sql

Preferred: modernc.org/sqlite (pure Go, no CGO) with database/sql.

// go.mod dependencies
// modernc.org/sqlite
// github.com/google/uuid

Model struct:

package models

import (
    "database/sql"
    "time"
    "github.com/google/uuid"
)

type User struct {
    ID         int64     `json:"-" db:"id"`
    ExternalID string    `json:"id" db:"external_id"`
    Email      string    `json:"email" db:"email"`
    Name       string    `json:"name" db:"name"`
    Status     string    `json:"status" db:"status"`
    CreatedAt  string    `json:"created_at" db:"created_at"`
    UpdatedAt  string    `json:"updated_at" db:"updated_at"`
    Version    int64     `json:"-" db:"version"`
}

Repository pattern:

package repository

import (
    "context"
    "database/sql"
    "fmt"
    "github.com/google/uuid"
)

type UserRepo struct {
    db *sql.DB
}

func NewUserRepo(db *sql.DB) *UserRepo {
    return &UserRepo{db: db}
}

func (r *UserRepo) FindByID(ctx context.Context, id int64) (*User, error) {
    row := r.db.QueryRowContext(ctx,
        "SELECT id, external_id, email, name, status, created_at, updated_at, version FROM users WHERE id = ?", id)
    var u User
    err := row.Scan(&u.ID, &u.ExternalID, &u.Email, &u.Name, &u.Status, &u.CreatedAt, &u.UpdatedAt, &u.Version)
    if err == sql.ErrNoRows {
        return nil, nil
    }
    return &u, err
}

func (r *UserRepo) Create(ctx context.Context, email, name string) (int64, error) {
    extID := uuid.New().String()
    result, err := r.db.ExecContext(ctx,
        "INSERT INTO users (external_id, email, name) VALUES (?, ?, ?)",
        extID, email, name)
    if err != nil {
        return 0, err
    }
    return result.LastInsertId()
}

func (r *UserRepo) UpdateOptimistic(ctx context.Context, u *User) (bool, error) {
    result, err := r.db.ExecContext(ctx,
        `UPDATE users SET email = ?, name = ?, version = version + 1,
         updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
         WHERE id = ? AND version = ?`,
        u.Email, u.Name, u.ID, u.Version)
    if err != nil {
        return false, err
    }
    rows, _ := result.RowsAffected()
    return rows > 0, nil
}

Connection setup:

package db

import (
    "database/sql"
    _ "modernc.org/sqlite"
)

func OpenSQLite(path string) (*sql.DB, error) {
    db, err := sql.Open("sqlite", path+"?_journal_mode=WAL&_busy_timeout=5000&_foreign_keys=ON")
    if err != nil {
        return nil, err
    }
    db.SetMaxOpenConns(1)  // SQLite: single writer
    db.SetMaxIdleConns(2)

    // Performance pragmas
    pragmas := []string{
        "PRAGMA synchronous = NORMAL",
        "PRAGMA cache_size = -64000",
        "PRAGMA auto_vacuum = INCREMENTAL",
    }
    for _, p := range pragmas {
        if _, err := db.Exec(p); err != nil {
            return nil, err
        }
    }
    return db, nil
}

2. Redis with go-redis

// github.com/redis/go-redis/v9

Cache layer:

package cache

import (
    "context"
    "encoding/json"
    "fmt"
    "time"
    "github.com/redis/go-redis/v9"
)

type CacheLayer struct {
    client *redis.Client
    prefix string
}

func NewCacheLayer(redisURL, service string) (*CacheLayer, error) {
    opts, err := redis.ParseURL(redisURL)
    if err != nil {
        return nil, err
    }
    return &CacheLayer{
        client: redis.NewClient(opts),
        prefix: service,
    }, nil
}

func (c *CacheLayer) key(entity, id string) string {
    return fmt.Sprintf("%s:cache:%s:%s", c.prefix, entity, id)
}

func (c *CacheLayer) Get(ctx context.Context, entity, id string, dest interface{}) error {
    data, err := c.client.Get(ctx, c.key(entity, id)).Bytes()
    if err == redis.Nil {
        return ErrCacheMiss
    }
    if err != nil {
        return err
    }
    return json.Unmarshal(data, dest)
}

func (c *CacheLayer) Set(ctx context.Context, entity, id string, value interface{}, ttl time.Duration) error {
    data, err := json.Marshal(value)
    if err != nil {
        return err
    }
    return c.client.Set(ctx, c.key(entity, id), data, ttl).Err()
}

func (c *CacheLayer) Invalidate(ctx context.Context, entity, id string) error {
    return c.client.Del(ctx, c.key(entity, id)).Err()
}

var ErrCacheMiss = fmt.Errorf("cache miss")

3. Neo4j with neo4j-go-driver

// github.com/neo4j/neo4j-go-driver/v5

Graph client:

package graph

import (
    "context"
    "github.com/neo4j/neo4j-go-driver/v5/neo4j"
)

type GraphClient struct {
    driver neo4j.DriverWithContext
}

func NewGraphClient(uri, user, pass string) (*GraphClient, error) {
    driver, err := neo4j.NewDriverWithContext(uri, neo4j.BasicAuth(user, pass, ""))
    if err != nil {
        return nil, err
    }
    return &GraphClient{driver: driver}, nil
}

func (g *GraphClient) Close(ctx context.Context) error {
    return g.driver.Close(ctx)
}

func (g *GraphClient) EnsureConstraints(ctx context.Context) error {
    session := g.driver.NewSession(ctx, neo4j.SessionConfig{})
    defer session.Close(ctx)
    _, err := session.Run(ctx,
        `CREATE CONSTRAINT uniq_user_userId IF NOT EXISTS
         FOR (u:User) REQUIRE u.userId IS UNIQUE`, nil)
    return err
}

func (g *GraphClient) UpsertUser(ctx context.Context, userID, email string) error {
    session := g.driver.NewSession(ctx, neo4j.SessionConfig{AccessMode: neo4j.AccessModeWrite})
    defer session.Close(ctx)
    _, err := session.Run(ctx,
        `MERGE (u:User {userId: $uid})
         SET u.email = $email, u.updatedAt = datetime()`,
        map[string]interface{}{"uid": userID, "email": email})
    return err
}

func (g *GraphClient) AddRelationship(ctx context.Context, fromID, toID, relType string) error {
    session := g.driver.NewSession(ctx, neo4j.SessionConfig{AccessMode: neo4j.AccessModeWrite})
    defer session.Close(ctx)
    cypher := fmt.Sprintf(
        `MATCH (a:User {userId: $from}), (b:User {userId: $to})
         MERGE (a)-[:%s]->(b)`, relType)
    _, err := session.Run(ctx, cypher,
        map[string]interface{}{"from": fromID, "to": toID})
    return err
}

4. Model Patterns

Domain model with DB-specific conversions:

package domain

type UserDomain struct {
    ID        string     `json:"id"`
    Email     string     `json:"email"`
    Name      string     `json:"name"`
    Status    UserStatus `json:"status"`
    CreatedAt time.Time  `json:"created_at"`
}

type UserStatus string
const (
    StatusActive    UserStatus = "active"
    StatusSuspended UserStatus = "suspended"
    StatusDeleted   UserStatus = "deleted"
)

func UserFromRow(row *User) *UserDomain {
    t, _ := time.Parse(time.RFC3339, row.CreatedAt)
    return &UserDomain{
        ID:        row.ExternalID,
        Email:     row.Email,
        Name:      row.Name,
        Status:    UserStatus(row.Status),
        CreatedAt: t,
    }
}

5. Error Handling

Unified error type:

package dberr

import "errors"

var (
    ErrNotFound       = errors.New("entity not found")
    ErrOptimisticLock = errors.New("optimistic lock conflict")
    ErrCacheMiss      = errors.New("cache miss")
)

type DbError struct {
    Source string // "sqlite", "redis", "neo4j"
    Op     string // "find", "create", "update"
    Err    error
}

func (e *DbError) Error() string {
    return fmt.Sprintf("%s.%s: %v", e.Source, e.Op, e.Err)
}

func (e *DbError) Unwrap() error { return e.Err }

6. Connection Management

Unified database context:

package db

type DbContext struct {
    SQLite *sql.DB
    Redis  *CacheLayer
    Graph  *GraphClient
}

func NewDbContext(cfg *Config) (*DbContext, error) {
    sqlite, err := OpenSQLite(cfg.SQLitePath)
    if err != nil {
        return nil, fmt.Errorf("sqlite: %w", err)
    }
    redis, err := NewCacheLayer(cfg.RedisURL, cfg.ServiceName)
    if err != nil {
        return nil, fmt.Errorf("redis: %w", err)
    }
    graph, err := NewGraphClient(cfg.Neo4jURI, cfg.Neo4jUser, cfg.Neo4jPass)
    if err != nil {
        return nil, fmt.Errorf("neo4j: %w", err)
    }
    if err := graph.EnsureConstraints(context.Background()); err != nil {
        return nil, fmt.Errorf("neo4j constraints: %w", err)
    }
    return &DbContext{SQLite: sqlite, Redis: redis, Graph: graph}, nil
}

func (ctx *DbContext) Close() {
    ctx.SQLite.Close()
    ctx.Redis.client.Close()
    ctx.Graph.Close(context.Background())
}

Integration Patterns: SQLite + Redis + Neo4j

Table of Contents

1. Data Flow Architecture 2. Early Stage Integration 3. Advanced Stage Integration 4. Sync Strategies 5. Query Routing

1. Data Flow Architecture

Each database has a distinct role. Data flows between them via application code and events.

                    ┌─────────────┐
                    │  Application │
                    └──────┬──────┘
                           │
            ┌──────────────┼──────────────┐
            │              │              │
            ▼              ▼              ▼
     ┌──────────┐   ┌──────────┐   ┌──────────┐
     │  SQLite   │   │  Redis    │   │  Neo4j    │
     │  (truth)  │   │  (speed)  │   │  (graph)  │
     └──────────┘   └──────────┘   └──────────┘
     Source of       Cache layer    Relationship
     truth for       + sessions     queries +
     all entity      + rate limits  access control
     data            + events       + recommendations

Rules:

• SQLite is ALWAYS the source of truth for entity data
• Redis caches SQLite data and handles ephemeral state
• Neo4j stores relationships and graph-queryable properties only
• Cross-DB references use UUIDs (SQLite external_id = Neo4j userId = Redis key segment)

2. Early Stage Integration

Minimal integration — add complexity only when needed.

Read Path (with cache)

Client request
  → Check Redis cache
    → HIT: return cached data
    → MISS: query SQLite → store in Redis (TTL 5min) → return

Write Path

Client write
  → Write to SQLite (transaction)
  → Invalidate Redis cache key
  → If graph-relevant: update Neo4j node/relationship
  → Return success

When to Add Neo4j

Add Neo4j only when you have queries that are painful in SQL:

• "Find all users connected to user X within 3 degrees"
• "What products do friends of user X like?"
• "Can user X access resource Y through any role chain?"

If your queries are simple JOINs, SQLite is sufficient. Do not add Neo4j preemptively.

3. Advanced Stage Integration

Event-driven integration with eventual consistency.

Write Path (event-sourced)

Client write
  → Write to SQLite (transaction)
  → Publish event to Redis Stream: app:events:{entity_type}
  → Return success (async processing below)

Event consumers (background workers):
  → Consumer 1: Update Redis cache (write-through)
  → Consumer 2: Update Neo4j graph nodes/relationships
  → Consumer 3: Write to audit log
  → Consumer 4: Trigger notifications

Read Path (CQRS)

Simple entity lookup:
  → Redis cache → SQLite fallback

Complex relationship query:
  → Neo4j (returns UUIDs) → SQLite (hydrate full entities) → Redis (cache result)

Aggregation/analytics:
  → SQLite (with appropriate indexes)

Real-time leaderboard/ranking:
  → Redis sorted sets (pre-computed)

Consistency Model

Acceptable staleness: cache data can be up to TTL seconds old. Graph data can be up to consumer lag behind. Session data is always current.

4. Sync Strategies

SQLite → Neo4j Sync

Only sync properties needed for graph queries:

SQLite users table:
  id, external_id, email, name, status, bio, avatar_url, preferences_json, ...

Neo4j User node:
  userId (= external_id), email, status
  (only what's needed for MATCH/WHERE in graph queries)

Sync trigger: Redis Stream consumer.

Event: { action: "update", entity: "user", id: "uuid-123", fields: ["email", "status"] }

Consumer logic:
  if any synced field changed:
    MERGE (u:User {userId: $id}) SET u.email = $email, u.status = $status

Cache Invalidation Patterns

5. Query Routing

Decision tree for where to run a query:

Is it a graph traversal (paths, recommendations, access control)?
  → YES: Neo4j (return UUIDs, hydrate from SQLite if needed)
  → NO: continue

Is it a simple key-value lookup by ID?
  → YES: Redis cache first, SQLite fallback
  → NO: continue

Is it a filtered list with pagination?
  → YES: Check Redis for cached result, else SQLite with indexes
  → NO: continue

Is it an aggregation or report?
  → YES: SQLite directly (indexes + query optimization)
  → NO: continue

Is it real-time ranking or counting?
  → YES: Redis sorted sets or HyperLogLog
  → NO: SQLite as default

Migration Patterns

Table of Contents

1. File Naming and Structure 2. Migration Content Rules 3. Rollback Strategy 4. Schema Versioning Table 5. Early vs Advanced Stage

1. File Naming and Structure

migrations/
├── 20260329120000_create_users.sql
├── 20260329120001_create_products.sql
├── 20260329120002_create_orders.sql
├── 20260329120003_create_order_items.sql
├── 20260329120004_create_audit_log.sql
└── 20260329120005_add_tenant_id.sql

Timestamp format: YYYYMMDDHHMMSS — guarantees ordering across timezones and developers.

2. Migration Content Rules

Every migration file has two sections:

-- +migrate up
CREATE TABLE users (
    id          INTEGER PRIMARY KEY,
    external_id TEXT    NOT NULL UNIQUE,
    email       TEXT    NOT NULL UNIQUE,
    name        TEXT    NOT NULL,
    status      TEXT    NOT NULL DEFAULT 'active'
        CHECK(status IN ('active','suspended','deleted')),
    created_at  TEXT    NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
    updated_at  TEXT    NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
    version     INTEGER NOT NULL DEFAULT 1
) STRICT;

CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_users_status ON users(status);

-- +migrate down
DROP INDEX IF EXISTS idx_users_status;
DROP INDEX IF EXISTS idx_users_email;
DROP TABLE IF EXISTS users;

Rules:

• up creates or alters; down reverses exactly
• One logical change per migration (one table, or one set of related indexes)
• Never edit a migration after it has been applied to any environment
• Include indexes in the same migration as the table they index
• Down migrations drop in reverse order of creation

3. Rollback Strategy

Safe rollback patterns:

SQLite limitation: ALTER TABLE cannot drop columns (before 3.35) or add constraints. For these operations, use the table rebuild pattern:

-- +migrate up (add NOT NULL column with default)
ALTER TABLE users ADD COLUMN phone TEXT NOT NULL DEFAULT '';

-- +migrate up (complex: add constraint requires rebuild)
CREATE TABLE users_new (...new schema...);
INSERT INTO users_new SELECT ... FROM users;
DROP TABLE users;
ALTER TABLE users_new RENAME TO users;
-- Recreate all indexes

4. Schema Versioning Table

Track applied migrations:

CREATE TABLE IF NOT EXISTS schema_migrations (
    version     TEXT PRIMARY KEY,
    name        TEXT NOT NULL,
    checksum    TEXT NOT NULL,
    applied_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
    rolled_back_at TEXT
) STRICT;

Before applying a migration: 1. Check if version exists in schema_migrations 2. If exists and not rolled back, skip 3. If not exists, apply and insert record with SHA-256 checksum of file 4. On rollback, set rolled_back_at timestamp (never delete the record)

5. Early vs Advanced Stage

Early Stage Migrations

Keep it simple:

• Sequential numbering is fine
• Manual application via script
• Single developer can manage
• Focus on getting schema right, iterate fast

# Simple migration runner
for f in migrations/*.sql; do
    sqlite3 app.db < "$f"
done

Advanced Stage Migrations

Production-grade:

• CI/CD pipeline runs migrations automatically
• Checksum verification prevents tampering
• Blue-green deployment: apply migration, verify, switch traffic
• Separate read/write migration phases for zero-downtime:

1. Add new column (nullable) — deploy 2. Backfill data — deploy 3. Add NOT NULL constraint — deploy 4. Remove old column — deploy (next release)

Neo4j migrations follow the same timestamp pattern but use .cypher extension:

neo4j-migrations/
├── 20260329120000_create_constraints.cypher
├── 20260329120001_create_indexes.cypher
└── 20260329120002_seed_roles.cypher

Naming Conventions

Universal naming rules applied across all databases and languages.

SQL (SQLite)

Redis Keys

Neo4j

Rust

Go

Cross-Database ID Mapping

SQLite external_id (UUID) = Neo4j userId property = Redis key segment.

Never expose SQLite integer id outside the service. Always use UUIDs for external communication and cross-database references.

Neo4j Graph Schema Patterns

Table of Contents

1. Node and Relationship Design 2. Naming Conventions 3. Constraints and Indexes 4. Common Graph Patterns 5. Cypher Query Patterns 6. Integration with SQLite

1. Node and Relationship Design

Nodes represent entities. Relationships represent verbs between them.

// Nodes — PascalCase labels, properties as camelCase
(:User {userId: "uuid", email: "...", createdAt: datetime()})
(:Product {productId: "uuid", name: "...", priceInCents: 4999})
(:Role {name: "admin", description: "..."})
(:Tenant {tenantId: "uuid", name: "..."})

// Relationships — UPPER_SNAKE_CASE verbs, with properties when needed
(:User)-[:PURCHASED {quantity: 2, purchasedAt: datetime()}]->(:Product)
(:User)-[:HAS_ROLE {grantedAt: datetime(), grantedBy: "uuid"}]->(:Role)
(:User)-[:BELONGS_TO]->(:Tenant)
(:User)-[:FOLLOWS {since: datetime()}]->(:User)

Rules:

• Node labels are nouns in PascalCase: User, Product, OrderItem
• Relationships are verbs in UPPER_SNAKE_CASE: PURCHASED, FOLLOWS, CREATED_BY
• Never use nouns for relationships (not :FRIENDSHIP, use :FRIENDS_WITH)
• Store the same userId/productId UUIDs as in SQLite for cross-DB joins
• Timestamps use Neo4j datetime() type, not strings

2. Naming Conventions

3. Constraints and Indexes

Apply these at schema creation time:

// Uniqueness constraints (also create indexes automatically)
CREATE CONSTRAINT uniq_user_userId IF NOT EXISTS
FOR (u:User) REQUIRE u.userId IS UNIQUE;

CREATE CONSTRAINT uniq_product_productId IF NOT EXISTS
FOR (p:Product) REQUIRE p.productId IS UNIQUE;

// Existence constraints (enterprise edition)
CREATE CONSTRAINT req_user_email IF NOT EXISTS
FOR (u:User) REQUIRE u.email IS NOT NULL;

// Node key constraints (composite uniqueness)
CREATE CONSTRAINT key_tenant_user IF NOT EXISTS
FOR (u:User) REQUIRE (u.tenantId, u.email) IS NODE KEY;

// Indexes for frequently queried properties
CREATE INDEX idx_user_email IF NOT EXISTS FOR (u:User) ON (u.email);
CREATE INDEX idx_product_name IF NOT EXISTS FOR (p:Product) ON (p.name);

// Full-text index for search
CREATE FULLTEXT INDEX ft_product_search IF NOT EXISTS
FOR (p:Product) ON EACH [p.name, p.description];

// Relationship property index
CREATE INDEX idx_purchased_date IF NOT EXISTS
FOR ()-[r:PURCHASED]-() ON (r.purchasedAt);

4. Common Graph Patterns

Access Control Graph (Advanced Stage)

(:User)-[:HAS_ROLE]->(:Role)-[:PERMITS {actions: ["read","write"]}]->(:Resource)
(:Role)-[:INHERITS_FROM]->(:Role)  // role hierarchy

// Query: Can user X do action Y on resource Z?
MATCH (u:User {userId: $uid})-[:HAS_ROLE]->(r:Role)-[:PERMITS]->(res:Resource {name: $resource})
WHERE $action IN r.actions
RETURN count(r) > 0 AS permitted

Recommendation Engine

// Users who bought X also bought Y
MATCH (u:User)-[:PURCHASED]->(:Product {productId: $pid})<-[:PURCHASED]-(other:User)
MATCH (other)-[:PURCHASED]->(rec:Product)
WHERE NOT (u)-[:PURCHASED]->(rec)
RETURN rec.name, count(other) AS score ORDER BY score DESC LIMIT 10

Organization Hierarchy

(:Employee)-[:REPORTS_TO]->(:Employee)
(:Department)-[:PART_OF]->(:Division)
(:Employee)-[:WORKS_IN]->(:Department)

// Find all reports (recursive)
MATCH (mgr:Employee {userId: $uid})<-[:REPORTS_TO*1..10]-(report:Employee)
RETURN report.name, length(path) AS depth

Knowledge Graph (Advanced Stage)

(:Concept)-[:RELATED_TO {weight: 0.85}]->(:Concept)
(:Document)-[:MENTIONS]->(:Concept)
(:User)-[:INTERESTED_IN]->(:Concept)

5. Cypher Query Patterns

Parameterized Queries (always use parameters, never string concatenation)

// Good — parameterized
MATCH (u:User {userId: $userId}) RETURN u

// Bad — injection risk
MATCH (u:User {userId: '${userId}'}) RETURN u

Batch Operations

// Use UNWIND for bulk inserts
UNWIND $users AS userData
MERGE (u:User {userId: userData.userId})
SET u.email = userData.email, u.name = userData.name, u.updatedAt = datetime()

Pagination

MATCH (u:User)
WHERE u.createdAt < $cursor
RETURN u ORDER BY u.createdAt DESC LIMIT $pageSize

6. Integration with SQLite

Neo4j stores relationships and graph queries. SQLite stores the full entity data. They share UUIDs.

SQLite: users table (id, external_id, email, name, ..., all columns)
Neo4j:  (:User {userId: external_id, email})  ← minimal properties for graph queries

Sync pattern:
1. Write to SQLite (source of truth for entity data)
2. Publish event to Redis Stream
3. Consumer creates/updates Neo4j node with UUID + graph-relevant properties
4. Graph queries return UUIDs → fetch full data from SQLite

Only store properties in Neo4j that are needed for graph traversal or filtering. Full entity data lives in SQLite.

Redis Schema Patterns (Caching Layer)

Table of Contents

1. Key Naming Convention 2. Cache-Aside Pattern 3. Session Management 4. Rate Limiting 5. Event Streaming 6. TTL Policies 7. Configuration Template

1. Key Naming Convention

All Redis keys follow this namespace pattern:

{service}:{entity}:{id}:{field}

Examples:

app:user:42:profile          → JSON hash of user profile
app:session:abc123           → session data with TTL
app:cache:products:list:p1   → cached product list page 1
app:ratelimit:api:10.0.0.1  → sorted set for sliding window
app:lock:order:99            → distributed lock

Multi-tenant prefix:

t:{tenant_id}:{service}:{entity}:{id}
t:7:app:user:42:profile

Rules:

• Colons : as separators, never dots or slashes
• Lowercase everything
• Service name first for cluster routing with hash tags
• Keep keys under 512 bytes (shorter is faster)

2. Cache-Aside Pattern

The primary caching strategy for SQLite data:

READ:  App → Redis GET → hit? return : SQLite SELECT → Redis SET with TTL → return
WRITE: App → SQLite UPDATE → Redis DEL (invalidate) → return

Key design for cache-aside:

app:cache:{table}:{id}           → single record cache
app:cache:{table}:list:{hash}    → query result cache (hash of query params)
app:cache:{table}:count          → count cache for pagination

Invalidation rules:

• On single record update: delete app:cache:{table}:{id}
• On any write to table: delete app:cache:{table}:list:* (use SCAN, never KEYS)
• On schema migration: flush app:cache:* namespace

3. Session Management

app:session:{session_id} → HASH {
    user_id:    "42"
    tenant_id:  "7"
    roles:      '["admin","editor"]'
    ip:         "10.0.0.1"
    created_at: "2026-03-29T12:00:00Z"
}
TTL: 86400 (24 hours)

Sliding expiration: reset TTL on each access with EXPIRE.

Session index for admin (find all sessions for a user):

app:user:{user_id}:sessions → SET of session_ids

4. Rate Limiting

Sliding window with sorted sets:

Key:    app:ratelimit:{scope}:{identifier}
Score:  Unix timestamp (microseconds)
Member: Unique request ID

ZADD    app:ratelimit:api:10.0.0.1 {now_us} {request_id}
ZREMRANGEBYSCORE app:ratelimit:api:10.0.0.1 0 {now_us - window_us}
ZCARD   app:ratelimit:api:10.0.0.1
EXPIRE  app:ratelimit:api:10.0.0.1 {window_seconds}

If ZCARD > limit, reject request. Window and limit configurable per scope.

5. Event Streaming

Redis Streams for event sourcing alongside SQLite snapshots:

Stream: app:events:{entity_type}
Entry:  { action: "create", entity_id: "42", data: "{json}", actor: "user:7" }

XADD app:events:orders * action create entity_id 42 data '{"total":9900}'
XREAD COUNT 10 BLOCK 5000 STREAMS app:events:orders $

Consumer groups for reliable processing:

XGROUP CREATE app:events:orders workers $ MKSTREAM
XREADGROUP GROUP workers worker-1 COUNT 10 BLOCK 5000 STREAMS app:events:orders >
XACK app:events:orders workers {message_id}

Use streams for: audit events, cache invalidation signals, cross-service notifications.

6. TTL Policies

Rules:

• EVERY cache key MUST have a TTL — no immortal cache keys
• Sessions use sliding TTL (reset on access)
• Locks use short TTL as safety net against crashes
• Streams use MAXLEN trim, not TTL

7. Configuration Template

The redis-schema.toml file documents all key namespaces:

[service]
name = "app"
version = "1.0.0"

[namespaces.cache]
pattern = "{service}:cache:{table}:{id}"
ttl_seconds = 300
description = "Single-record cache-aside for SQLite tables"

[namespaces.session]
pattern = "{service}:session:{session_id}"
ttl_seconds = 86400
sliding = true
description = "User session data"

[namespaces.ratelimit]
pattern = "{service}:ratelimit:{scope}:{identifier}"
ttl_seconds = 60
description = "Sliding window rate limiter"

[namespaces.events]
pattern = "{service}:events:{entity_type}"
maxlen = 10000
description = "Event stream for entity changes"

[namespaces.lock]
pattern = "{service}:lock:{resource}:{id}"
ttl_seconds = 30
description = "Distributed advisory locks"

Rust Database Bindings

Table of Contents

1. SQLite with sqlx 2. Redis with redis-rs 3. Neo4j with neo4rs 4. Model Patterns 5. Error Handling 6. Connection Management

1. SQLite with sqlx

Preferred: sqlx (async, compile-time checked queries, no DSL).

# Cargo.toml
[dependencies]
sqlx = { version = "0.8", features = ["runtime-tokio", "sqlite"] }
tokio = { version = "1", features = ["full"] }
uuid = { version = "1", features = ["v4", "serde"] }
chrono = { version = "0.4", features = ["serde"] }
serde = { version = "1", features = ["derive"] }

Model struct:

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use sqlx::FromRow;
use uuid::Uuid;

#[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
pub struct User {
    pub id: i64,
    pub external_id: String,
    pub email: String,
    pub name: String,
    pub status: String,
    pub created_at: String,
    pub updated_at: String,
    pub version: i64,
}

impl User {
    pub fn external_uuid(&self) -> Uuid {
        Uuid::parse_str(&self.external_id).expect("valid uuid in DB")
    }
}

Repository pattern:

use sqlx::SqlitePool;

pub struct UserRepo {
    pool: SqlitePool,
}

impl UserRepo {
    pub fn new(pool: SqlitePool) -> Self {
        Self { pool }
    }

    pub async fn find_by_id(&self, id: i64) -> sqlx::Result<Option<User>> {
        sqlx::query_as::<_, User>("SELECT * FROM users WHERE id = ?")
            .bind(id)
            .fetch_optional(&self.pool)
            .await
    }

    pub async fn create(&self, email: &str, name: &str) -> sqlx::Result<i64> {
        let external_id = Uuid::new_v4().to_string();
        let result = sqlx::query(
            "INSERT INTO users (external_id, email, name) VALUES (?, ?, ?)"
        )
            .bind(&external_id)
            .bind(email)
            .bind(name)
            .execute(&self.pool)
            .await?;
        Ok(result.last_insert_rowid())
    }

    pub async fn update_optimistic(&self, user: &User) -> sqlx::Result<bool> {
        let result = sqlx::query(
            "UPDATE users SET email = ?, name = ?, version = version + 1,
             updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
             WHERE id = ? AND version = ?"
        )
            .bind(&user.email)
            .bind(&user.name)
            .bind(user.id)
            .bind(user.version)
            .execute(&self.pool)
            .await?;
        Ok(result.rows_affected() > 0)
    }
}

Connection pool:

use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};

pub async fn create_pool(db_path: &str) -> sqlx::Result<SqlitePool> {
    let options = SqliteConnectOptions::new()
        .filename(db_path)
        .create_if_missing(true)
        .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
        .busy_timeout(std::time::Duration::from_secs(5))
        .foreign_keys(true);

    let pool = SqlitePoolOptions::new()
        .max_connections(5)
        .connect_with(options)
        .await?;

    // Run pragmas
    sqlx::query("PRAGMA synchronous = NORMAL").execute(&pool).await?;
    sqlx::query("PRAGMA cache_size = -64000").execute(&pool).await?;

    Ok(pool)
}

2. Redis with redis-rs

[dependencies]
redis = { version = "0.27", features = ["tokio-comp", "connection-manager"] }

Cache layer:

use redis::AsyncCommands;

pub struct CacheLayer {
    conn: redis::aio::ConnectionManager,
    prefix: String,
}

impl CacheLayer {
    pub async fn new(redis_url: &str, service: &str) -> redis::RedisResult<Self> {
        let client = redis::Client::open(redis_url)?;
        let conn = redis::aio::ConnectionManager::new(client).await?;
        Ok(Self { conn, prefix: service.to_string() })
    }

    fn key(&self, entity: &str, id: &str) -> String {
        format!("{}:cache:{}:{}", self.prefix, entity, id)
    }

    pub async fn get_cached<T: serde::de::DeserializeOwned>(
        &mut self, entity: &str, id: &str,
    ) -> redis::RedisResult<Option<T>> {
        let key = self.key(entity, id);
        let data: Option<String> = self.conn.get(&key).await?;
        Ok(data.and_then(|s| serde_json::from_str(&s).ok()))
    }

    pub async fn set_cached<T: serde::Serialize>(
        &mut self, entity: &str, id: &str, value: &T, ttl_secs: u64,
    ) -> redis::RedisResult<()> {
        let key = self.key(entity, id);
        let json = serde_json::to_string(value).map_err(|e|
            redis::RedisError::from((redis::ErrorKind::TypeError, "serialize", e.to_string()))
        )?;
        self.conn.set_ex(&key, &json, ttl_secs).await
    }

    pub async fn invalidate(&mut self, entity: &str, id: &str) -> redis::RedisResult<()> {
        let key = self.key(entity, id);
        self.conn.del(&key).await
    }
}

3. Neo4j with neo4rs

[dependencies]
neo4rs = "0.8"

Graph client:

use neo4rs::{Graph, query};

pub struct GraphClient {
    graph: Graph,
}

impl GraphClient {
    pub async fn new(uri: &str, user: &str, pass: &str) -> Result<Self, neo4rs::Error> {
        let graph = Graph::new(uri, user, pass).await?;
        Ok(Self { graph })
    }

    pub async fn ensure_constraints(&self) -> Result<(), neo4rs::Error> {
        self.graph.run(query(
            "CREATE CONSTRAINT uniq_user_userId IF NOT EXISTS
             FOR (u:User) REQUIRE u.userId IS UNIQUE"
        )).await?;
        Ok(())
    }

    pub async fn upsert_user(&self, user_id: &str, email: &str) -> Result<(), neo4rs::Error> {
        self.graph.run(
            query("MERGE (u:User {userId: $uid}) SET u.email = $email, u.updatedAt = datetime()")
                .param("uid", user_id)
                .param("email", email)
        ).await
    }

    pub async fn add_relationship(
        &self, from_id: &str, to_id: &str, rel_type: &str,
    ) -> Result<(), neo4rs::Error> {
        let cypher = format!(
            "MATCH (a:User {{userId: $from}}), (b:User {{userId: $to}})
             MERGE (a)-[:{}]->(b)", rel_type  // rel_type must be validated
        );
        self.graph.run(query(&cypher).param("from", from_id).param("to", to_id)).await
    }
}

4. Model Patterns

Shared model with DB-specific conversions:

/// Domain model — database-agnostic
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct UserDomain {
    pub id: Uuid,
    pub email: String,
    pub name: String,
    pub status: UserStatus,
    pub created_at: DateTime<Utc>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum UserStatus { Active, Suspended, Deleted }

/// Convert from SQLite row
impl From<User> for UserDomain {
    fn from(row: User) -> Self {
        Self {
            id: Uuid::parse_str(&row.external_id).unwrap(),
            email: row.email,
            name: row.name,
            status: match row.status.as_str() {
                "active" => UserStatus::Active,
                "suspended" => UserStatus::Suspended,
                _ => UserStatus::Deleted,
            },
            created_at: DateTime::parse_from_rfc3339(&row.created_at)
                .unwrap().with_timezone(&Utc),
        }
    }
}

5. Error Handling

Unified error type across all three databases:

#[derive(Debug, thiserror::Error)]
pub enum DbError {
    #[error("SQLite error: {0}")]
    Sqlite(#[from] sqlx::Error),
    #[error("Redis error: {0}")]
    Redis(#[from] redis::RedisError),
    #[error("Neo4j error: {0}")]
    Neo4j(#[from] neo4rs::Error),
    #[error("Not found: {entity} {id}")]
    NotFound { entity: String, id: String },
    #[error("Conflict: optimistic lock failed")]
    OptimisticLock,
}

6. Connection Management

Unified database context:

pub struct DbContext {
    pub sqlite: SqlitePool,
    pub redis: CacheLayer,
    pub graph: GraphClient,
}

impl DbContext {
    pub async fn new(config: &DbConfig) -> Result<Self, DbError> {
        let sqlite = create_pool(&config.sqlite_path).await?;
        let redis = CacheLayer::new(&config.redis_url, &config.service_name).await?;
        let graph = GraphClient::new(&config.neo4j_uri, &config.neo4j_user, &config.neo4j_pass).await?;
        graph.ensure_constraints().await?;
        Ok(Self { sqlite, redis, graph })
    }
}

SQLite Schema Patterns

Table of Contents

1. Table Creation 2. Type System 3. Indexing Strategy 4. WAL Mode and Performance 5. Constraints and Validation 6. Common Enterprise Tables

1. Table Creation

Always use STRICT tables (SQLite 3.37+) to enforce type checking:

CREATE TABLE users (
    id          INTEGER PRIMARY KEY,  -- auto-increment via ROWID
    external_id TEXT    NOT NULL UNIQUE,  -- UUID for API exposure
    email       TEXT    NOT NULL UNIQUE,
    name        TEXT    NOT NULL,
    status      TEXT    NOT NULL DEFAULT 'active' CHECK(status IN ('active','suspended','deleted')),
    created_at  TEXT    NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
    updated_at  TEXT    NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
    version     INTEGER NOT NULL DEFAULT 1
) STRICT;

Key rules:

• INTEGER PRIMARY KEY aliases ROWID — fastest possible lookups
• Use TEXT for UUIDs and expose those externally, never raw integer IDs
• Store timestamps as ISO-8601 TEXT in UTC — portable and sortable
• Add version column for optimistic locking on mutable tables

2. Type System

SQLite STRICT mode supports five types:

For money/currency, store as INTEGER in smallest unit (cents) to avoid floating point errors.

3. Indexing Strategy

-- Every foreign key gets an index
CREATE INDEX idx_orders_user_id ON orders(user_id);

-- Composite indexes for common query patterns (leftmost prefix rule)
CREATE INDEX idx_orders_user_status ON orders(user_id, status);

-- Partial indexes for filtered queries (SQLite 3.8+)
CREATE INDEX idx_orders_active ON orders(user_id) WHERE status = 'active';

-- Covering indexes to avoid table lookups
CREATE INDEX idx_users_email_name ON users(email, name);

-- Expression indexes for case-insensitive search
CREATE INDEX idx_users_email_lower ON users(lower(email));

Rules:

• Index every column used in WHERE, JOIN, or ORDER BY
• Composite index column order matches query filter order
• Use partial indexes for status-filtered queries (saves space)
• Monitor with EXPLAIN QUERY PLAN — no full table scans on tables > 1000 rows

4. WAL Mode and Performance

Always enable WAL mode for concurrent read/write:

PRAGMA journal_mode = WAL;
PRAGMA busy_timeout = 5000;
PRAGMA synchronous = NORMAL;       -- safe with WAL
PRAGMA cache_size = -64000;        -- 64MB cache
PRAGMA foreign_keys = ON;          -- enforce FK constraints
PRAGMA auto_vacuum = INCREMENTAL;  -- reclaim space without full vacuum

Set these pragmas at connection open, before any queries.

5. Constraints and Validation

-- NOT NULL on everything unless genuinely optional
-- CHECK constraints for enums and ranges
-- UNIQUE constraints for natural keys
-- Foreign keys with explicit ON DELETE behavior

CREATE TABLE order_items (
    id         INTEGER PRIMARY KEY,
    order_id   INTEGER NOT NULL REFERENCES orders(id) ON DELETE CASCADE,
    product_id INTEGER NOT NULL REFERENCES products(id) ON DELETE RESTRICT,
    quantity   INTEGER NOT NULL CHECK(quantity > 0),
    unit_price INTEGER NOT NULL CHECK(unit_price >= 0),  -- cents
    created_at TEXT    NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
) STRICT;

ON DELETE policies:

• CASCADE — child rows deleted with parent (order_items when order deleted)
• RESTRICT — prevent parent deletion if children exist (products with orders)
• SET NULL — set FK to NULL (optional relationships)

6. Common Enterprise Tables

Audit Log

CREATE TABLE audit_log (
    id          INTEGER PRIMARY KEY,
    entity_type TEXT    NOT NULL,
    entity_id   INTEGER NOT NULL,
    action      TEXT    NOT NULL CHECK(action IN ('create','update','delete')),
    actor_id    INTEGER,
    diff_json   TEXT,  -- JSON of changed fields
    created_at  TEXT   NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
) STRICT;

CREATE INDEX idx_audit_entity ON audit_log(entity_type, entity_id);
CREATE INDEX idx_audit_actor ON audit_log(actor_id);
CREATE INDEX idx_audit_created ON audit_log(created_at);

Migration Tracking

CREATE TABLE schema_migrations (
    version    TEXT    PRIMARY KEY,  -- timestamp: '20260329120000'
    name       TEXT    NOT NULL,
    checksum   TEXT    NOT NULL,     -- SHA-256 of migration file
    applied_at TEXT    NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
    rolled_back_at TEXT
) STRICT;

Multi-Tenant Base Pattern

For advanced stage, every business table includes:

tenant_id INTEGER NOT NULL REFERENCES tenants(id),
-- Add tenant_id as first column in all composite indexes
CREATE INDEX idx_orders_tenant_status ON orders(tenant_id, status);