Add nonblocking/asyn I/O operations

backend/tests/Clustering.service.test.js

@@ -1,18 +1,18 @@
 import { describe, it, expect, vi, beforeEach } from 'vitest';
 
-const { createMock, mockEmbeddings } = vi.hoisted(() => {
+const { streamMock, mockEmbeddings } = vi.hoisted(() => {
   const embeddings = new Map([
     ['note_001', [1.0, 0.0, 0.0]],
     ['note_002', [0.0, 1.0, 0.0]],
   ]);
-  return { createMock: vi.fn(), mockEmbeddings: embeddings };
+  return { streamMock: vi.fn(), mockEmbeddings: embeddings };
 });
 
 vi.mock('@anthropic-ai/sdk', () => {
   return {
     default: class MockAnthropic {
       constructor() {
-        this.messages = { create: createMock };
+        this.messages = { stream: streamMock };
       }
     },
   };
@@ -34,6 +34,40 @@ const MOCK_CLUSTERS = [
   { label: 'Export Issues', noteIds: ['note_002'] },
 ];
 
+// Splits text into several text_delta events so the service is exercised
+// against a genuinely incremental stream rather than one whole payload.
+const textEvents = (text, pieces = 4) => {
+  const size = Math.max(1, Math.ceil(text.length / pieces));
+  const events = [];
+  for (let i = 0; i < text.length; i += size) {
+    events.push({
+      type: 'content_block_delta',
+      delta: { type: 'text_delta', text: text.slice(i, i + size) },
+    });
+  }
+  return events;
+};
+
+const mockStreamOf = (text) => {
+  const events = [
+    { type: 'message_start' },
+    ...textEvents(text),
+    { type: 'message_stop' },
+  ];
+  streamMock.mockImplementation(() => ({
+    async *[Symbol.asyncIterator]() {
+      for (const event of events) yield event;
+    },
+  }));
+};
+
+const mockStreamThrowing = (err) => {
+  streamMock.mockImplementation(() => ({
+    async *[Symbol.asyncIterator]() {
+      throw err;
+    },
+  }));
+};
 
 describe('clusterNotes service', () => {
 
@@ -41,27 +75,32 @@ describe('clusterNotes service', () => {
     vi.clearAllMocks();
   });
 
-  it('should call Anthropic messages.create with the correct model', async () => {
-    createMock.mockResolvedValue({
-      content: [{ type: 'text', text: JSON.stringify(MOCK_CLUSTERS) }],
-    });
+  it('should call Anthropic messages.stream with the correct model', async () => {
+    mockStreamOf(JSON.stringify(MOCK_CLUSTERS));
 
     await clusterNotes(MOCK_NOTES);
 
-    expect(createMock).toHaveBeenCalledOnce();
-    const callArgs = createMock.mock.calls[0][0];
+    expect(streamMock).toHaveBeenCalledOnce();
+    const callArgs = streamMock.mock.calls[0][0];
     expect(callArgs.model).toBe('claude-sonnet-4-20250514');
     expect(callArgs.max_tokens).toBe(4096);
   });
 
+  it('should forward an abort signal to the LLM request', async () => {
+    mockStreamOf(JSON.stringify(MOCK_CLUSTERS));
+    const controller = new AbortController();
+
+    await clusterNotes(MOCK_NOTES, { signal: controller.signal });
+
+    expect(streamMock.mock.calls[0][1]).toEqual({ signal: controller.signal });
+  });
+
   it('should include all note texts in prompt sent to the LLM API', async () => {
-    createMock.mockResolvedValue({
-      content: [{ type: 'text', text: JSON.stringify(MOCK_CLUSTERS) }],
-    });
+    mockStreamOf(JSON.stringify(MOCK_CLUSTERS));
 
     await clusterNotes(MOCK_NOTES);
 
-    const prompt = createMock.mock.calls[0][0].messages[0].content;
+    const prompt = streamMock.mock.calls[0][0].messages[0].content;
     expect(prompt).toContain('note_001');
     expect(prompt).toContain('Login is broken');
     expect(prompt).toContain('note_002');
@@ -69,9 +108,7 @@ describe('clusterNotes service', () => {
   });
 
   it('should return clusters and a cohesion score', async () => {
-    createMock.mockResolvedValue({
-      content: [{ type: 'text', text: JSON.stringify(MOCK_CLUSTERS) }],
-    });
+    mockStreamOf(JSON.stringify(MOCK_CLUSTERS));
 
     const result = await clusterNotes(MOCK_NOTES);
 
@@ -81,22 +118,35 @@ describe('clusterNotes service', () => {
     expect(result.score).toBeLessThanOrEqual(1);
   });
 
+  it('should reassemble clusters split across many stream deltas', async () => {
+    const json = JSON.stringify(MOCK_CLUSTERS);
+    streamMock.mockImplementation(() => ({
+      async *[Symbol.asyncIterator]() {
+        for (const char of json) {
+          yield { type: 'content_block_delta', delta: { type: 'text_delta', text: char } };
+        }
+      },
+    }));
+
+    const result = await clusterNotes(MOCK_NOTES);
+
+    expect(result.clusters).toEqual(MOCK_CLUSTERS);
+  });
+
   it('should throw error when the API returns non-JSON', async () => {
-    createMock.mockResolvedValue({
-      content: [{ type: 'text', text: 'An unknown error occured when generting structured response.' }],
-    });
+    mockStreamOf('An unknown error occured when generting structured response.');
 
     await expect(clusterNotes(MOCK_NOTES)).rejects.toThrow('non-JSON response');
   });
 
   it('should throw error when the API response has no text content', async () => {
-    createMock.mockResolvedValue({ content: [] });
+    mockStreamOf('');
 
     await expect(clusterNotes(MOCK_NOTES)).rejects.toThrow('no text content returned');
   });
 
   it('should throw error when Anthropic API authentication fails', async () => {
-    createMock.mockRejectedValue(new Error('401 Unauthorized'));
+    mockStreamThrowing(new Error('401 Unauthorized'));
 
     await expect(clusterNotes(MOCK_NOTES)).rejects.toThrow('401 Unauthorized');
   });
@@ -105,9 +155,7 @@ describe('clusterNotes service', () => {
     const incompleteClusters = [
       { label: 'Auth Issues', noteIds: ['note_001'] },
     ];
-    createMock.mockResolvedValue({
-      content: [{ type: 'text', text: JSON.stringify(incompleteClusters) }],
-    });
+    mockStreamOf(JSON.stringify(incompleteClusters));
 
     await expect(clusterNotes(MOCK_NOTES)).rejects.toThrow('Cluster validation failed');
   });

backend/tests/Notes.test.js

@@ -1,16 +1,18 @@
 import { describe, it, expect, vi, beforeEach } from 'vitest';
 import request from 'supertest';
+import { Readable } from 'node:stream';
 import app from '../app.js';
 
 vi.mock('../db/notes.dao.js', () => ({
   getAllNotes: vi.fn(),
+  streamAllNotes: vi.fn(),
 }));
 
 vi.mock('../services/clustering.service.js', () => ({
   clusterNotes: vi.fn(),
 }));
 
-import { getAllNotes } from '../db/notes.dao.js';
+import { getAllNotes, streamAllNotes } from '../db/notes.dao.js';
 import { clusterNotes } from '../services/clustering.service.js';
 
 const MOCK_NOTES = [
@@ -24,6 +26,8 @@ const MOCK_CLUSTERS = [
   { label: 'Export Problems', noteIds: ['note_003'] },
 ];
 
+const rowStream = (rows) => Readable.from(rows, { objectMode: true });
+
 describe('GET /v1/notes', () => {
 
   beforeEach(() => {
@@ -31,7 +35,7 @@ describe('GET /v1/notes', () => {
   });
 
   it('should return 200 and an array of notes', async () => {
-    getAllNotes.mockResolvedValue(MOCK_NOTES);
+    streamAllNotes.mockResolvedValue(rowStream(MOCK_NOTES));
 
     const res = await request(app).get('/v1/notes');
 
@@ -40,8 +44,26 @@ describe('GET /v1/notes', () => {
     expect(Array.isArray(res.body)).toBe(true);
   });
 
+  it('should send JSON incrementally rather than buffering the row set', async () => {
+    streamAllNotes.mockResolvedValue(rowStream(MOCK_NOTES));
+
+    const res = await request(app).get('/v1/notes');
+
+    expect(res.headers['content-type']).toMatch(/application\/json/);
+    expect(res.headers['content-length']).toBeUndefined();
+  });
+
+  it('should return an empty array when there are no notes', async () => {
+    streamAllNotes.mockResolvedValue(rowStream([]));
+
+    const res = await request(app).get('/v1/notes');
+
+    expect(res.status).toBe(200);
+    expect(res.body).toEqual([]);
+  });
+
   it('should return notes with expected properties', async () => {
-    getAllNotes.mockResolvedValue(MOCK_NOTES);
+    streamAllNotes.mockResolvedValue(rowStream(MOCK_NOTES));
 
     const res = await request(app).get('/v1/notes');
     const note = res.body[0];
@@ -55,7 +77,7 @@ describe('GET /v1/notes', () => {
   });
 
   it('should return 500 when the database query fails', async () => {
-    getAllNotes.mockRejectedValue(new Error('connection refused'));
+    streamAllNotes.mockRejectedValue(new Error('connection refused'));
 
     const res = await request(app).get('/v1/notes');
 
@@ -63,6 +85,19 @@ describe('GET /v1/notes', () => {
     expect(res.body).toHaveProperty('error');
     expect(res.body.error).toBe('Failed to load notes');
   });
+
+  it('should abort the response when the row stream fails mid-flight', async () => {
+    const failing = new Readable({
+      objectMode: true,
+      read() {
+        this.push(MOCK_NOTES[0]);
+        this.destroy(new Error('connection lost'));
+      },
+    });
+    streamAllNotes.mockResolvedValue(failing);
+
+    await expect(request(app).get('/v1/notes')).rejects.toThrow();
+  });
 });
 
 describe('POST /v1/notes/cluster', () => {
@@ -101,7 +136,22 @@ describe('POST /v1/notes/cluster', () => {
     await request(app).post('/v1/notes/cluster');
 
     expect(clusterNotes).toHaveBeenCalledOnce();
-    expect(clusterNotes).toHaveBeenCalledWith(MOCK_NOTES);
+    expect(clusterNotes).toHaveBeenCalledWith(
+      MOCK_NOTES,
+      expect.objectContaining({ signal: expect.any(AbortSignal) })
+    );
+  });
+
+  it('should pass a signal that is not aborted while the request is open', async () => {
+    getAllNotes.mockResolvedValue(MOCK_NOTES);
+    clusterNotes.mockImplementation(async (_notes, { signal }) => {
+      expect(signal.aborted).toBe(false);
+      return MOCK_CLUSTERS;
+    });
+
+    const res = await request(app).post('/v1/notes/cluster');
+
+    expect(res.status).toBe(200);
   });
 
   it('should return 500 when clusterNotes (API call) fails', async () => {

backend/tests/notes.dao.test.js

@@ -1,14 +1,23 @@
 import { describe, it, expect, vi, beforeEach } from 'vitest';
+import { Readable } from 'node:stream';
 
-const { mockQuery } = vi.hoisted(() => ({
+const { mockQuery, mockConnect } = vi.hoisted(() => ({
   mockQuery: vi.fn(),
+  mockConnect: vi.fn(),
 }));
 
 vi.mock('../db/index.js', () => ({
   query: mockQuery,
+  getPool: () => ({ connect: mockConnect }),
 }));
 
-import { getAllNotes, getNoteById, createNote, createNotes } from '../db/notes.dao.js';
+import {
+  getAllNotes,
+  streamAllNotes,
+  getNoteById,
+  createNote,
+  createNotes,
+} from '../db/notes.dao.js';
 
 const MOCK_ROWS = [
   { id: 'note_001', text: 'Login flow feels confusing', x: 193, y: 191, author: 'user_5', color: 'yellow' },
@@ -48,6 +57,58 @@ describe('notes.dao', () => {
     });
   });
 
+  describe('streamAllNotes', () => {
+    const mockClient = (rows) => {
+      const release = vi.fn();
+      const client = {
+        release,
+        query: vi.fn(() => Readable.from(rows, { objectMode: true })),
+      };
+      mockConnect.mockResolvedValue(client);
+      return { client, release };
+    };
+
+    it('should stream rows without buffering them into an array', async () => {
+      mockClient(MOCK_ROWS);
+
+      const stream = await streamAllNotes();
+      const received = [];
+      for await (const row of stream) received.push(row);
+
+      expect(received).toEqual(MOCK_ROWS);
+    });
+
+    it('should release the pooled client once the stream ends', async () => {
+      const { release } = mockClient(MOCK_ROWS);
+
+      const stream = await streamAllNotes();
+      for await (const _row of stream) { /* drain */ }
+
+      expect(release).toHaveBeenCalledOnce();
+    });
+
+    it('should release the pooled client when a consumer destroys the stream early', async () => {
+      const { release } = mockClient(MOCK_ROWS);
+
+      const stream = await streamAllNotes();
+      stream.destroy();
+      await new Promise((resolve) => stream.once('close', resolve));
+
+      expect(release).toHaveBeenCalledOnce();
+    });
+
+    it('should release the pooled client when starting the query throws', async () => {
+      const release = vi.fn();
+      mockConnect.mockResolvedValue({
+        release,
+        query: vi.fn(() => { throw new Error('cursor failed'); }),
+      });
+
+      await expect(streamAllNotes()).rejects.toThrow('cursor failed');
+      expect(release).toHaveBeenCalledOnce();
+    });
+  });
+
   describe('getNoteById', () => {
     it('should return a single note when found', async () => {
       mockQuery.mockResolvedValue({ rows: [MOCK_ROWS[0]] });
@@ -109,5 +170,31 @@ describe('notes.dao', () => {
       expect(sql).toContain('INSERT INTO notes');
       expect(params).toHaveLength(12);
     });
+
+    it('should return an empty array without querying when given no notes', async () => {
+      const result = await createNotes([]);
+
+      expect(result).toEqual([]);
+      expect(mockQuery).not.toHaveBeenCalled();
+    });
+
+    it('should split large inputs into multiple statements under the bind-parameter limit', async () => {
+      const many = Array.from({ length: 2500 }, (_, i) => ({
+        id: `note_${i}`,
+        text: `text ${i}`,
+        author: 'user_1',
+      }));
+      mockQuery.mockImplementation(async (_sql, params) => ({
+        rows: new Array(params.length / 6).fill(null).map((_, i) => ({ i })),
+      }));
+
+      const result = await createNotes(many);
+
+      expect(mockQuery).toHaveBeenCalledTimes(3);
+      expect(result).toHaveLength(2500);
+      for (const [, params] of mockQuery.mock.calls) {
+        expect(params.length).toBeLessThan(65535);
+      }
+    });
   });
 });

backend/tests/streams.test.js

@@ -0,0 +1,82 @@
+import { describe, it, expect } from 'vitest';
+import { Readable } from 'node:stream';
+import { pipeline } from 'node:stream/promises';
+import { batch, jsonArray } from '../lib/streams.js';
+
+const collect = async (source, transform) => {
+  const out = [];
+  await pipeline(source, transform, async (results) => {
+    for await (const item of results) out.push(item);
+  });
+  return out;
+};
+
+describe('batch', () => {
+  it('should group items into fixed-size arrays', async () => {
+    const source = Readable.from([1, 2, 3, 4], { objectMode: true });
+
+    const result = await collect(source, batch(2));
+
+    expect(result).toEqual([[1, 2], [3, 4]]);
+  });
+
+  it('should flush a partial trailing batch', async () => {
+    const source = Readable.from([1, 2, 3, 4, 5], { objectMode: true });
+
+    const result = await collect(source, batch(2));
+
+    expect(result).toEqual([[1, 2], [3, 4], [5]]);
+  });
+
+  it('should emit nothing for an empty source', async () => {
+    const source = Readable.from([], { objectMode: true });
+
+    const result = await collect(source, batch(3));
+
+    expect(result).toEqual([]);
+  });
+
+  it('should reject a non-positive size', () => {
+    expect(() => batch(0)).toThrow(TypeError);
+    expect(() => batch(1.5)).toThrow(TypeError);
+  });
+});
+
+describe('jsonArray', () => {
+  const serialize = async (items) => {
+    const chunks = await collect(
+      Readable.from(items, { objectMode: true }),
+      jsonArray()
+    );
+    return chunks.map(String).join('');
+  };
+
+  it('should serialize objects into a JSON array', async () => {
+    const items = [{ id: 'a' }, { id: 'b' }];
+
+    const output = await serialize(items);
+
+    expect(output).toBe('[{"id":"a"},{"id":"b"}]');
+    expect(JSON.parse(output)).toEqual(items);
+  });
+
+  it('should emit an empty array when the source yields nothing', async () => {
+    const output = await serialize([]);
+
+    expect(output).toBe('[]');
+    expect(JSON.parse(output)).toEqual([]);
+  });
+
+  it('should emit a valid single-element array', async () => {
+    const output = await serialize([{ id: 'only' }]);
+
+    expect(JSON.parse(output)).toEqual([{ id: 'only' }]);
+  });
+
+  it('should propagate serialization errors', async () => {
+    const circular = {};
+    circular.self = circular;
+
+    await expect(serialize([circular])).rejects.toThrow();
+  });
+});