Documentation

Queues, Blob & Vector

QueuePort, BlobPort, VectorPort, and SqlPort extend the ports & adapters model to message queuing, binary/text blob storage, dense-vector similarity search, and raw SQL. Each ships with an in-memory fake you can drop into tests with zero infrastructure.

QueuePort — message queuing#

QueuePort models a simple FIFO queue where a producer enqueues messages and a consumer drains them. The in-memory fake is synchronous and deterministic, making it ideal for testing background-job pipelines without a real broker.

Name	Type	Description
enqueue(body)	(unknown) → Promise<void>	Adds a message to the queue. body can be any serializable value.
size()	() → Promise<number>	Returns the number of messages currently waiting in the queue.
drain(handler)	((body: unknown) => Promise<void>) → Promise<void>	Calls handler once for every pending message, in order, and empties the queue.

inMemoryQueue

inMemoryQueue(id?) creates an in-process FIFO queue. The optional id is a label used in debug output — useful when you have multiple queues in the same test.

examples/queue.ts

import { inMemoryQueue } from '@kynetra/fx'
 
interface EmailJob { to: string; subject: string; body: string }
 
const queue = inMemoryQueue<EmailJob>('emails')
 
// Producer
await queue.enqueue({ to: 'alice@example.com', subject: 'Welcome', body: 'Hi!' })
await queue.enqueue({ to: 'bob@example.com', subject: 'Welcome', body: 'Hi!' })
 
console.log(await queue.size()) // 2
 
// Consumer
await queue.drain(async (job) => {
  const email = job as EmailJob
  console.log(`Sending to ${email.to}…`)
  // await sendEmail(email)
})
 
console.log(await queue.size()) // 0

src/routes/signup.ts

import { createFX } from '@kynetra/fx'
import { inMemoryQueue } from '@kynetra/fx-ports'
 
const app = createFX()
app.decorateContext('emailQueue', inMemoryQueue('emails'))
 
app.post('/signup', async (ctx) => {
  const { email, name } = await ctx.jsonBody<{ email: string; name: string }>()
  const q = ctx.get('emailQueue')
 
  // Enqueue without blocking the response
  await q.enqueue({ to: email, subject: `Welcome, ${name}!`, template: 'welcome' })
 
  return ctx.status(201).json({ ok: true })
})

src/routes/signup.test.ts

import { describe, it, expect } from 'vitest'
import { inMemoryQueue } from '@kynetra/fx-ports'
import app from './signup'
 
it('enqueues a welcome email on signup', async () => {
  const q = inMemoryQueue('emails')
  app.decorateContext('emailQueue', q)
 
  await app.fetch(new Request('http://x/signup', {
    method: 'POST',
    headers: { 'content-type': 'application/json' },
    body: JSON.stringify({ email: 'test@example.com', name: 'Alice' }),
  }))
 
  expect(await q.size()).toBe(1)
  const jobs: unknown[] = []
  await q.drain(async (j) => { jobs.push(j) })
  expect((jobs[0] as { to: string }).to).toBe('test@example.com')
})

BlobPort — binary and text blob storage#

BlobPort is an opaque key/value store for binary data — images, PDFs, export files. It maps directly to Cloudflare R2, AWS S3, or any object-storage system. The in-memory fake stores blobs as Uint8Array values.

Name	Type	Description
put(key, data, opts?)	(string, Uint8Array \| string, { contentType?: string }?) → Promise<void>	Stores a blob under the given key. String data is encoded as UTF-8.
get(key)	(string) → Promise<{ data: Uint8Array; contentType?: string } \| null>	Returns the blob and its content type, or null if not found.
delete(key)	(string) → Promise<void>	Removes the blob. No-op if absent.
list(prefix?)	(string?) → Promise<string[]>	Lists all keys with the given prefix, or all keys if omitted.

inMemoryBlob

examples/blob.ts

import { inMemoryBlob } from '@kynetra/fx'
 
const blob = inMemoryBlob()
 
// Store a text file
const encoder = new TextEncoder()
await blob.put('reports/q1.csv', encoder.encode('id,revenue\n1,9000'), {
  contentType: 'text/csv',
})
 
// Store binary data (e.g. a PNG thumbnail)
const imageBytes = new Uint8Array([137, 80, 78, 71]) // PNG header
await blob.put('avatars/alice.png', imageBytes, { contentType: 'image/png' })
 
// Read back
const result = await blob.get('reports/q1.csv')
if (result) {
  const text = new TextDecoder().decode(result.data)
  console.log(text) // 'id,revenue\n1,9000'
}
 
// List by prefix
const reportKeys = await blob.list('reports/')
console.log(reportKeys) // ['reports/q1.csv']

src/routes/uploads.ts

import { createFX } from '@kynetra/fx'
import { inMemoryBlob } from '@kynetra/fx-ports'
 
const app = createFX()
app.decorateContext('blobs', inMemoryBlob())
 
app.put('/files/:key', async (ctx) => {
  const blobs = ctx.get('blobs')
  const contentType = ctx.header('content-type') ?? 'application/octet-stream'
  const data = new Uint8Array(await ctx.req.arrayBuffer())
  await blobs.put(ctx.params.key, data, { contentType })
  return ctx.status(204).response()
})
 
app.get('/files/:key', async (ctx) => {
  const blobs = ctx.get('blobs')
  const result = await blobs.get(ctx.params.key)
  if (!result) return ctx.status(404).json({ error: 'not found' })
  return ctx.response(result.data, {
    headers: { 'content-type': result.contentType ?? 'application/octet-stream' },
  })
})

VectorPort — dense vector similarity search#

VectorPort stores dense floating-point vectors and answers approximate nearest neighbour (ANN) queries using cosine similarity. It is the storage half of the RAG pipeline. In-memory search is brute-force — fine for thousands of vectors in tests; in production swap for Cloudflare Vectorize, pgvector, or Pinecone.

Name	Type	Description
upsert(records)	(VectorRecord[]) → Promise<void>	Inserts or replaces vectors by ID. Each record must include id, vector, and optional metadata.
query(vector, topK?)	(number[], number?) → Promise<VectorMatch[]>	Returns up to topK records sorted by descending cosine similarity. Defaults to 10.
delete(ids)	(string[]) → Promise<void>	Removes the vectors with the given IDs.

VectorRecord and VectorMatch shapes

examples/vector-types.ts

import type { VectorRecord, VectorMatch } from '@kynetra/fx'
 
// What you store
const record: VectorRecord = {
  id: 'doc-1',
  vector: [0.12, -0.34, 0.91],   // must match embedding dimension
  metadata: { title: 'Intro to ports', url: '/docs/ports' },
}
 
// What you get back from query()
const match: VectorMatch = {
  id: 'doc-1',
  score: 0.97,                   // cosine similarity in [0, 1]
  metadata: { title: 'Intro to ports', url: '/docs/ports' },
}

inMemoryVector

inMemoryVector() implements brute-force cosine search. Scores are in [0, 1] — higher is more similar.

examples/vector.ts

import { inMemoryVector } from '@kynetra/fx'
 
const vectors = inMemoryVector()
 
// Upsert embeddings (3-dimensional for brevity)
await vectors.upsert([
  { id: 'a', vector: [1, 0, 0], metadata: { label: 'x-axis' } },
  { id: 'b', vector: [0, 1, 0], metadata: { label: 'y-axis' } },
  { id: 'c', vector: [0.7, 0.7, 0], metadata: { label: 'diagonal' } },
])
 
// Nearest neighbours to [1, 0, 0]
const results = await vectors.query([1, 0, 0], 2)
console.log(results[0].id)    // 'a' (score ≈ 1.0)
console.log(results[1].id)    // 'c' (score ≈ 0.7)
 
// Remove a vector
await vectors.delete(['b'])

Note

The query vector does not need to be unit-normalised — the implementation normalises both the query and stored vectors before computing the dot product.

SqlPort — parameterised SQL#

SqlPort is a thin interface over any SQL database. It separates read queries from write operations, making it easy to instrument or swap the backend. The primary production adapter is d1Sql for Cloudflare D1.

Name	Type	Description
query<T>(sql, params?)	(string, unknown[]?) → Promise<T[]>	Executes a SELECT and returns typed rows. params are positionally bound (?).
execute(sql, params?)	(string, unknown[]?) → Promise<void>	Executes a DML or DDL statement (INSERT, UPDATE, DELETE, CREATE TABLE, …).

recordingSql — test double

recordingSql(rows?) is a test double (not a mock). It always returns the fixed rows array from query(), and records every execute() call so you can assert on statements and parameters.

examples/recording-sql.ts

import { recordingSql } from '@kynetra/fx'
 
interface User { id: string; email: string }
 
// Pre-seed the rows that query() will return
const sql = recordingSql<User>([
  { id: '1', email: 'alice@example.com' },
  { id: '2', email: 'bob@example.com' },
])
 
const users = await sql.query<User>('SELECT * FROM users WHERE active = ?', [true])
console.log(users.length) // 2
 
// Execute a write
await sql.execute('UPDATE users SET active = ? WHERE id = ?', [false, '2'])
 
// Inspect what was executed
console.log(sql.executed)
// [{ sql: 'UPDATE users SET active = ? WHERE id = ?', params: [false, '2'] }]

src/repos/users.test.ts

import { describe, it, expect } from 'vitest'
import { recordingSql } from '@kynetra/fx-ports'
import { UserRepo } from './users'
 
describe('UserRepo', () => {
  it('deactivates a user by ID', async () => {
    const sql = recordingSql([{ id: '1', email: 'test@example.com', active: true }])
    const repo = new UserRepo(sql)
 
    await repo.deactivate('1')
 
    expect(sql.executed[0].sql).toContain('UPDATE users')
    expect(sql.executed[0].params).toContain('1')
  })
})

Tip

For integration tests that need actual SQL semantics, use the d1Sql adapter against a local D1 instance via wrangler dev, or spin up a SQLite in-process with better-sqlite3 and write a thin SqlPort adapter around it.

Choosing the right port#

Ephemeral session data, rate-limit counters, short-lived tokens → KvPort
Cached computed values (typed) → CachePort
Structured documents, CRUD entities → StorePort
Background jobs, event fan-out → QueuePort
User uploads, generated files, exports → BlobPort
Semantic search, AI-powered retrieval → VectorPort (see RAG)
Relational queries, joins, migrations → SqlPort

PreviousKV, Cache & Store NextAI port