feat(ai): stream AI suggestions over HTTP

.eslintrc.js

@@ -4,6 +4,12 @@ module.exports = {
     'node': true,
     'jest': true
   },
+  globals: {
+    /**
+     * TODO: bump eslint since it's current env uses older "node" version which missing required global types
+     */
+    'AbortController': 'readonly'
+  },
   rules: {
     '@typescript-eslint/camelcase': 'warn',
     '@typescript-eslint/no-unused-vars': 'warn',

package.json

@@ -1,6 +1,6 @@
 {
   "name": "hawk.api",
-  "version": "1.5.13",
+  "version": "1.5.14",
   "main": "index.ts",
   "license": "BUSL-1.1",
   "scripts": {
@@ -42,7 +42,7 @@
     "@graphql-tools/schema": "^8.5.1",
     "@graphql-tools/utils": "^8.9.0",
     "@hawk.so/nodejs": "^3.3.2",
-    "@hawk.so/types": "^0.5.9",
+    "@hawk.so/types": "^0.7.0",
     "@n1ru4l/json-patch-plus": "^0.2.0",
     "@node-saml/node-saml": "^5.0.1",
     "@octokit/oauth-methods": "^4.0.0",

src/directives/requireUserInWorkspace.ts

@@ -37,7 +37,7 @@ async function checkUserInWorkspaceByWorkspaceId(context: ResolverContextBase, w
  * @param context - request context
  * @param projectId - project id
  */
-async function checkUserInWorkspaceByProjectId(context: ResolverContextBase, projectId: string): Promise<void> {
+export async function checkUserInWorkspaceByProjectId(context: ResolverContextBase, projectId: string): Promise<void> {
   const userId = context.user.id;
 
   if (userId) {

src/index.ts

@@ -32,6 +32,7 @@ import ReleasesFactory from './models/releasesFactory';
 import RedisHelper from './redisHelper';
 import { appendSsoRoutes } from './sso';
 import { appendGitHubRoutes } from './integrations/github';
+import { appendAiAssistantRoutes } from './services/askAi';
 
 /**
  * Option to enable playground
@@ -272,6 +273,11 @@ class HawkAPI {
      */
     appendGitHubRoutes(this.app, sharedFactories);
 
+    /**
+     * Append AI assistant route to Express app
+     */
+    appendAiAssistantRoutes(this.app);
+
     await this.server.start();
     this.app.use(graphqlUploadExpress());
     this.server.applyMiddleware({ app: this.app });

src/integrations/vercel-ai/index.ts

@@ -1,4 +1,6 @@
-import { generateText } from 'ai';
+import { generateText, streamText, type TextStreamPart, type ToolSet } from 'ai';
+import { getErrorMessage, ProviderOptions } from '@ai-sdk/provider-utils';
+import type { AiStream } from '@hawk.so/types';
 
 /**
  * Params for a single completion call to the model
@@ -15,6 +17,44 @@ export interface CompletionParams {
   prompt: string;
 }
 
+/**
+ * Params for a streaming completion call to the model
+ */
+export interface StreamParams extends CompletionParams {
+  /**
+   * Aborted when the answer is no longer required, which stops the model
+   */
+  signal: AbortSignal;
+}
+
+/**
+ * Converts Vercel SDK's stream parts.
+ *
+ * Everything but text and error parts is dropped.
+ *
+ * @param parts - stream of incoming SDK parts
+ * @returns {AiStream} stream converted of converted parts
+ */
+async function * toAiStream<TOOLS extends ToolSet>(
+  parts: AsyncIterable<TextStreamPart<TOOLS>>
+): AiStream {
+  for await (const part of parts) {
+    if (part.type === 'text-delta') {
+      yield {
+        type: 'text-delta',
+        delta: part.text,
+      };
+    }
+
+    if (part.type === 'error') {
+      yield {
+        type: 'error',
+        errorText: getErrorMessage(part.error),
+      };
+    }
+  }
+}
+
 /**
  * Interface for interacting with Vercel AI Gateway
  *
@@ -27,11 +67,24 @@ class VercelAIApi {
    */
   private readonly modelId: string;
 
+  /**
+   * Provider Gateway configurations
+   */
+  private readonly providerOptions: ProviderOptions;
+
+  /**
+   * Set up model id and provider fallback order
+   */
   constructor() {
     /**
      * @todo make it dynamic, get from project settings
      */
     this.modelId = 'deepseek/deepseek-v4-flash';
+    this.providerOptions = {
+      gateway: {
+        order: ['novita', 'azure', 'deepseek'],
+      },
+    };
   }
 
   /**
@@ -45,15 +98,29 @@ class VercelAIApi {
       model: this.modelId,
       system,
       prompt,
-      providerOptions: {
-        gateway: {
-          order: ['novita', 'azure', 'deepseek'],
-        },
-      },
+      providerOptions: this.providerOptions,
     });
 
     return text;
   }
+
+  /**
+   * Send a system/prompt pair to the model and return the streamed text
+   *
+   * @param {StreamParams} params - system instruction, prompt and abort signal
+   * @returns {AiStream} text generated by the model, as it arrives
+   */
+  public stream({ system, prompt, signal }: StreamParams): AiStream {
+    const { fullStream } = streamText({
+      model: this.modelId,
+      system,
+      prompt,
+      providerOptions: this.providerOptions,
+      abortSignal: signal,
+    });
+
+    return toAiStream(fullStream);
+  }
 }
 
 export const vercelAIApi = new VercelAIApi();

src/services/askAi/index.ts

@@ -1 +1,2 @@
 export { AskAiService, askAiService } from './service';
+export { appendAiAssistantRoutes } from './routes';

src/services/askAi/routes.ts

@@ -0,0 +1,156 @@
+import '../../typeDefs/expressContext';
+import express from 'express';
+import { ObjectId } from 'mongodb';
+import { getEventsFactory } from '../../resolvers/helpers/eventsFactory';
+import { checkUserInWorkspaceByProjectId } from '../../directives/requireUserInWorkspace';
+import { askAiService } from './service';
+import type { AiStreamPart } from '@hawk.so/types';
+
+/**
+ * Verify the requesting user is a member of the project's workspace.
+ *
+ * @param req - Express request
+ * @param res - Express response
+ * @param projectId - project id from query parameters (may be `string[]` if repeated)
+ * @returns user id and validated project id if authorized, `null` otherwise (response already sent)
+ */
+async function authorizeProjectAccess(
+  req: express.Request,
+  res: express.Response,
+  projectId: unknown
+): Promise<{ userId: string; projectId: string } | null> {
+  const userId = req.context?.user?.id;
+
+  if (!userId) {
+    res.status(401).json({ error: 'Unauthorized. Please provide authorization token.' });
+
+    return null;
+  }
+
+  if (!projectId || typeof projectId !== 'string') {
+    res.status(400).json({ error: 'projectId query parameter is required' });
+
+    return null;
+  }
+
+  if (!ObjectId.isValid(projectId)) {
+    res.status(400).json({ error: `Invalid projectId format: ${projectId}` });
+
+    return null;
+  }
+
+  try {
+    await checkUserInWorkspaceByProjectId(req.context, projectId);
+  } catch (error) {
+    res.status(403).json({ error: error instanceof Error ? error.message : 'You have no access to this workspace' });
+
+    return null;
+  }
+
+  return {
+    userId,
+    projectId,
+  };
+}
+
+/**
+ * Create AI assistant router
+ *
+ * @returns Express router with AI assistant endpoints
+ */
+export function createAiStreamRouter(): express.Router {
+  const router = express.Router();
+
+  /**
+   * GET /integration/ai/stream?projectId=<projectId>&eventId=<eventId>&originalEventId=<originalEventId>
+   * Stream an AI suggestion for the event
+   */
+  router.get('/stream', async (req, res, next) => {
+    const abort = new AbortController();
+
+    /** Abort response generation when connection is closed */
+    res.on('close', () => abort.abort());
+
+    try {
+      const { projectId, eventId, originalEventId } = req.query;
+
+      const authResult = await authorizeProjectAccess(req, res, projectId);
+
+      if (!authResult) {
+        return;
+      }
+
+      if (!eventId || typeof eventId !== 'string') {
+        res.status(400).json({ error: 'eventId query parameter is required' });
+
+        return;
+      }
+
+      if (!originalEventId || typeof originalEventId !== 'string') {
+        res.status(400).json({ error: 'originalEventId query parameter is required' });
+
+        return;
+      }
+
+      const eventsFactory = getEventsFactory(req.context, authResult.projectId);
+
+      let stream;
+
+      try {
+        stream = await askAiService.streamSuggestion(eventsFactory, eventId, originalEventId, abort.signal);
+      } catch (error) {
+        if (!(error instanceof Error) || error.message !== 'Event not found') {
+          throw error;
+        }
+
+        res.status(404).json({ error: error.message });
+
+        return;
+      }
+
+      res.writeHead(200, {
+        'content-type': 'text/event-stream',
+        'cache-control': 'no-cache',
+        connection: 'keep-alive',
+      });
+
+      try {
+        for await (const part of stream) {
+          if (abort.signal.aborted) {
+            break;
+          }
+
+          res.write(`data: ${JSON.stringify(part)}\n\n`);
+        }
+      } catch (error) {
+        if (!abort.signal.aborted) {
+          const part: AiStreamPart = {
+            type: 'error',
+            errorText: error instanceof Error ? error.message : 'AI suggestion failed.',
+          };
+
+          res.write(`data: ${JSON.stringify(part)}\n\n`);
+        }
+      }
+
+      res.end();
+    } catch (error) {
+      if (abort.signal.aborted) {
+        return;
+      }
+
+      next(error);
+    }
+  });
+
+  return router;
+}
+
+/**
+ * Append AI assistant routes to Express app
+ *
+ * @param app - Express application instance
+ */
+export function appendAiAssistantRoutes(app: express.Application): void {
+  app.use('/integration/ai', createAiStreamRouter());
+}