Documentation

Ports & adapters

Kynetra FX follows the ports-and-adapters (hexagonal) pattern: every external dependency — storage, queues, AI, clocks — is hidden behind a typed port interface. Business logic depends on the port, not the implementation. Swap the adapter in production; use a deterministic fake in tests.

Philosophy#

Traditional frameworks couple route handlers directly to databases, caches, and external services. That coupling makes unit-testing painful (you need a running Postgres, a real queue, etc.) and makes it hard to run the same code on different infrastructure.

Kynetra FX inverts this. Each category of capability is expressed as a TypeScript interface — a port. An adapter implements that interface for a specific backend (Cloudflare KV, D1, R2, an OpenAI API key, …). The @kynetra/fx-ports package ships both the port types and fast, dependency-free in-memory fakes that implement them. You wire the real adapter once, at the edge of your app, and keep it injected through context decorators or dependency injection of your choice.

Tip

The in-memory fakes are not mocks — they are real implementations that respect every contract (TTL expiry, collection isolation, cosine similarity ranking). Tests that pass against fakes pass against real adapters.

All ports at a glance#

Every port lives in @kynetra/fx-ports and is re-exported from @kynetra/fx.

Name	Type	Description
ClockPort	interface	now() → number. Lets you freeze time in tests.
systemClock	ClockPort	Returns Date.now(). Default for all production adapters.
fixedClock(start?)	(ms?) → ClockPort	Returns a clock fixed at start (default 0). Useful for TTL-sensitive tests.
IdPort	interface	generate() → string. Creates IDs.
randomId	IdPort	Cryptographically random hex IDs.
sequentialId(prefix?)	(s?) → IdPort	Deterministic IDs: "doc-1", "doc-2", … Great for snapshot tests.
KvPort	interface	String key/value store with optional TTL.
inMemoryKv(clock?)	(ClockPort?) → KvPort	Map-backed KV with TTL expiry driven by the injected clock.
CachePort	interface	Typed key/value cache with TTL.
inMemoryCache(clock?)	(ClockPort?) → CachePort	Same as inMemoryKv but generic (get<T>/set<T>).
StorePort	interface	Document CRUD organised into named collections.
inMemoryStore()	() → StorePort	Map-of-maps store. Each collection is isolated.
QueuePort	interface	Enqueue/drain message queue.
inMemoryQueue(id?)	(s?) → QueuePort	In-process FIFO queue. drain() processes all pending messages synchronously.
BlobPort	interface	Opaque binary/text blob storage.
inMemoryBlob()	() → BlobPort	Stores blobs as Uint8Array in memory.
VectorPort	interface	Upsert and cosine-similarity search over dense vectors.
inMemoryVector()	() → VectorPort	Brute-force cosine search. Fast enough for thousands of vectors in tests.
SqlPort	interface	Parameterised SQL query/execute.
recordingSql(rows?)	(T[]?) → SqlPort	Returns fixed rows from query(); records all execute() calls for assertions.

Wiring adapters into your app#

The recommended pattern is to inject ports through context decorators so every handler gets a typed reference without reaching for globals.

src/app.ts

import { createFX } from '@kynetra/fx'
import { inMemoryKv, inMemoryStore } from '@kynetra/fx-ports'
 
const app = createFX({ name: 'my-api' })
 
// Attach ports once at startup
app.decorateContext('kv', inMemoryKv())
app.decorateContext('store', inMemoryStore())
 
app.get('/sessions/:id', async (ctx) => {
  const kv = ctx.get('kv')          // KvPort
  const session = await kv.get(`session:${ctx.params.id}`)
  if (!session) return ctx.status(404).json({ error: 'not found' })
  return ctx.json(JSON.parse(session))
})
 
return app

In production, swap the fakes for real adapters (Cloudflare KV, D1, etc.) without changing any handler code:

src/worker.ts

import { cloudflare } from '@kynetra/fx-cloudflare'
import { d1Store } from '@kynetra/fx-cloudflare'
import app from './app'
 
// Replace the in-memory store with a D1-backed adapter at the edge
return {
  fetch(request: Request, env: Env, ctx: ExecutionContext) {
    app.decorateContext('store', d1Store(env.DB))
    return cloudflare(app).fetch(request, env, ctx)
  }
}

Writing your own adapter#

An adapter is any object that satisfies the port interface. You only need to implement the methods your code actually uses; TypeScript will tell you if you miss one.

Import the port interface

import type { KvPort } from '@kynetra/fx'

Implement every method

Return the correct types — no runtime duck-typing is involved.

Inject it like any other adapter

Pass it to app.decorateContext('kv', myAdapter) and your handlers never know the difference.

adapters/redis-kv.ts

import type { KvPort } from '@kynetra/fx'
import { createClient } from 'redis'
 
export function redisKv(client: ReturnType<typeof createClient>): KvPort {
  return {
    async get(key) {
      return client.get(key)
    },
    async set(key, value, opts) {
      if (opts?.ttlMs) {
        await client.set(key, value, { PX: opts.ttlMs })
      } else {
        await client.set(key, value)
      }
    },
    async delete(key) {
      await client.del(key)
    },
    async list(prefix) {
      const keys = await client.keys(`${prefix ?? ''}*`)
      return keys
    },
  }
}

In-memory fakes for deterministic tests#

Because the fakes implement the full port contract, you can write tests that exercise real business logic without any test database or network calls.

src/sessions.test.ts

import { describe, it, expect } from 'vitest'
import { inMemoryKv, fixedClock } from '@kynetra/fx-ports'
import app from './app'
 
describe('session lookup', () => {
  it('returns 404 when session does not exist', async () => {
    const kv = inMemoryKv()
    app.decorateContext('kv', kv)
 
    const res = await app.fetch(new Request('http://x/sessions/unknown'))
    expect(res.status).toBe(404)
  })
 
  it('returns session data when present', async () => {
    const kv = inMemoryKv()
    await kv.set('session:abc', JSON.stringify({ userId: '1' }))
    app.decorateContext('kv', kv)
 
    const res = await app.fetch(new Request('http://x/sessions/abc'))
    expect(res.status).toBe(200)
    expect(await res.json()).toEqual({ userId: '1' })
  })
 
  it('treats expired sessions as absent (TTL)', async () => {
    const clock = fixedClock(0)
    const kv = inMemoryKv(clock)
    await kv.set('session:xyz', JSON.stringify({ userId: '2' }), { ttlMs: 1000 })
 
    // Advance the clock past TTL
    ;(clock as any)._now = 2000
    app.decorateContext('kv', kv)
 
    const res = await app.fetch(new Request('http://x/sessions/xyz'))
    expect(res.status).toBe(404)
  })
})

Note

fixedClock returns a plain object. In tests you can mutate _now directly, or wrap it in a helper that advances time. Production code never sees that field — it only calls clock.now().

Cross-references#

See the individual port pages for full API details and examples:

KV, Cache & Store — KvPort, CachePort, StorePort
Queues, Blob & Vector — QueuePort, BlobPort, VectorPort, SqlPort
AI port — AiPort and mockAi
RAG — composing AiPort + VectorPort
Decorators — injecting ports into context
D1 adapters — d1Sql / d1Store for Cloudflare

PreviousAudit & feature flags NextKV, Cache & Store