Get Started with
TypeScript

Add the SDK to your project.

# npm / pnpm / yarn
npx jsr add @geeksquad/warp

# deno
deno add jsr:@geeksquad/warp

# bun
bunx jsr add @geeksquad/warp

Run Warp locally with Gleam or pull the Docker image.

# Option A: Gleam (if you have the source)
gleam run -m warp_server

# Option B: Docker
docker run -p 9090:9090 ghcr.io/dwighson/warp:latest

Create a client pointing at your running Warp instance. This opens a gRPC channel.

import { Warp } from '@geeksquad/warp'

const db = new Warp({
  host: 'localhost',
  port: 9090,
})

// When done, release the connection
db.close()

An entity is a lazy handle -- creating one makes no server call. Append events to build its history, then read the folded aggregate state.

// Create a lazy entity handle (no network call)
const alice = db.entity('user/alice')

// Append events to build up state
await alice.append('Credited', { amount: 5000 }, {
  aggregate: 'Account',
})

await alice.append('Debited', { amount: 200 }, {
  aggregate: 'Account',
})

// Read the folded aggregate state
const state = await alice.get('Account')
console.log(state) // { balance: 4800 }

Append multiple events atomically with appendBatch.

const events = await alice.appendBatch([
  {
    eventType: 'ProfileUpdated',
    payload: { name: 'Alice' },
    aggregate: 'Profile',
  },
  {
    eventType: 'EmailChanged',
    payload: { email: 'alice@example.com' },
    aggregate: 'Profile',
  },
])

For idempotent writes, supply your own event ID with appendWithId.

await alice.appendWithId(
  'evt-unique-001',
  'Credited',
  { amount: 1000 },
  { aggregate: 'Account' },
)

Every event is immutable. Retrieve the full trail or the last N events.

// Last 10 events
const recent = await alice.history(10)

// Full history (no limit)
const all = await alice.history()

// Each event has:
//   eventId, entityId, eventType, payload,
//   sequence, aggregate, occurredAt, schemaVsn
for (const e of recent) {
  console.log(e.eventType, e.payload, e.occurredAt)
}

Besides event sourcing, each entity has a key-value store for mutable state. Values are auto-serialized to JSON.

// Set a mutable key-value pair
await alice.put('preferences', {
  theme: 'dark',
  locale: 'en-US',
})

// Read it back (returns null if not found)
const prefs = await alice.getState('preferences')
console.log(prefs) // { theme: 'dark', locale: 'en-US' }

Run SQL queries against the read model. Supports parameterized queries to prevent injection.

// Simple query
const rows = await db.query(
  'SELECT * FROM accounts WHERE balance > ?',
  [1000],
)

// No params
const all = await db.query(
  'SELECT entity_id, balance FROM accounts ORDER BY balance DESC',
)

Coordinate changes across multiple entities with automatic compensation on failure. Each step has a forward event and an optional compensating event.

const result = await db.saga('transfer-001')
  .step(
    'user/alice',
    { type: 'Debited',  payload: { amount: 500 } },
    { type: 'Credited', payload: { amount: 500 } }, // compensate
  )
  .step(
    'user/bob',
    { type: 'Credited', payload: { amount: 500 } },
    { type: 'Debited',  payload: { amount: 500 } }, // compensate
  )
  .commit()

console.log(result.resultType) // 'COMMITTED' | 'COMPENSATED' | 'STUCK'
console.log(result.sagaId)     // 'transfer-001'
console.log(result.stepsRun)   // 2

Add a retry policy for transient failures.

const result = await db.saga('retry-saga')
  .step('user/alice', { type: 'Debited', payload: { amount: 100 } })
  .step('user/bob',   { type: 'Credited', payload: { amount: 100 } })
  .retry({
    maxAttempts: 3,
    initialBackoffMs: 100,
    maxTotalMs: 5000,
  })
  .commit()

Subscribe to live events on an entity. Returns an async iterator that yields events as they arrive. Optionally filter by aggregate or replay from a sequence number.

const alice = db.entity('user/alice')

// Subscribe to all events
for await (const event of alice.subscribe()) {
  console.log(event.eventType, event.payload)
}

// Filter by aggregate, replay from sequence 5
for await (const event of alice.subscribe('Account', 5)) {
  console.log(event.sequence, event.payload)
}

Export all events for a data subject, or permanently delete the entity and all its data.

const alice = db.entity('user/alice')

// Export: returns every event for this entity
const archive = await alice.export()
console.log(`Exported ${archive.length} events`)

// Right to erasure: delete entity and all data
await alice.delete()

Check system health and inspect cluster topology.

// System health
const health = await db.health()
console.log(health.overall) // 'HEALTHY'

for (const check of health.checks) {
  console.log(check.name, check.status, check.durationUs)
}

// Cluster topology
const cluster = await db.clusterStatus()
console.log(cluster.localNode)  // node identifier
console.log(cluster.nodes)      // all nodes in cluster
console.log(cluster.shardCount) // number of shards

import { Warp } from '@geeksquad/warp'

const db = new Warp({ host: 'localhost', port: 9090 })

// Create entities
const alice = db.entity('user/alice')
const bob   = db.entity('user/bob')

// Fund accounts
await alice.append('Credited', { amount: 10000 }, { aggregate: 'Account' })
await bob.append('Credited', { amount: 5000 },  { aggregate: 'Account' })

// Transfer via saga
const result = await db.saga('xfr-001')
  .step(
    'user/alice',
    { type: 'Debited',  payload: { amount: 2000 } },
    { type: 'Credited', payload: { amount: 2000 } },
  )
  .step(
    'user/bob',
    { type: 'Credited', payload: { amount: 2000 } },
    { type: 'Debited',  payload: { amount: 2000 } },
  )
  .commit()

console.log(result.resultType) // 'COMMITTED'

// Check balances
console.log(await alice.get('Account')) // { balance: 8000 }
console.log(await bob.get('Account'))   // { balance: 7000 }

// Query across entities
const rich = await db.query(
  'SELECT * FROM accounts WHERE balance > ?',
  [6000],
)

// Health check
const health = await db.health()
console.log(health.overall)

db.close()