wip: Event stream and chat history

packages/cli/jest.config.js

@@ -8,4 +8,5 @@ module.exports = {
   // There are test cases that change the process working directory, and that does
   // not work with multiple workers.
   maxWorkers: 1,
+  setupFilesAfterEnv: ['<rootDir>/tests/setup.js'],
 };

packages/cli/package.json

@@ -141,9 +141,11 @@
     "pretty-bytes": "^5.6.0",
     "ps-node": "^0.1.6",
     "read-pkg-up": "^7.0.1",
+    "reflect-metadata": "^0.2.2",
     "semver": "^7.3.5",
     "sql-formatter": "^4.0.2",
     "strip-ansi": "^6.0.0",
+    "tsyringe": "^4.8.0",
     "validator": "^13.7.0",
     "w3c-xmlserializer": "^2.0.0",
     "yargs": "^17.1.1"

packages/cli/src/cli.ts

@@ -6,6 +6,8 @@
 /* eslint-disable func-names */
 /* eslint-disable max-classes-per-file */
 
+import 'reflect-metadata';
+
 const yargs = require('yargs');
 const { promises: fsp, readFileSync } = require('fs');
 const { queue } = require('async');

packages/cli/src/cmds/index/rpc.ts

@@ -10,7 +10,7 @@ import appmapFilter from '../../rpc/appmap/filter';
 import { RpcHandler } from '../../rpc/rpc';
 import metadata from '../../rpc/appmap/metadata';
 import sequenceDiagram from '../../rpc/appmap/sequenceDiagram';
-import { explainHandler, explainStatusHandler, loadThreadHandler } from '../../rpc/explain/explain';
+import { explainHandler, explainStatusHandler } from '../../rpc/explain/explain';
 import { buildNavieProvider, commonNavieArgsBuilder as navieBuilder } from '../navie';
 import RPCServer from './rpcServer';
 import appmapData from '../../rpc/appmap/data';
@@ -27,12 +27,22 @@ import { update } from '../../rpc/file/update';
 import { INavieProvider } from '../../rpc/explain/navie/inavie';
 import { navieMetadataV1, navieMetadataV2 } from '../../rpc/navie/metadata';
 import { navieSuggestHandlerV1 } from '../../rpc/navie/suggest';
-import { initializeHistory } from '../../rpc/explain/navie/historyHelper';
-import History from '../../rpc/explain/navie/history';
-import { join } from 'path';
-import { homedir } from 'os';
 import { navieWelcomeV2 } from '../../rpc/navie/welcome';
 import { navieRegisterV1 } from '../../rpc/navie/register';
+import { navieThreadSendMessageHandler } from '../../rpc/navie/thread/handlers/sendMessage';
+import {
+  navieThreadPinItemHandler,
+  navieThreadUnpinItemHandler,
+} from '../../rpc/navie/thread/handlers/pinItem';
+import { navieThreadQueryHandler } from '../../rpc/navie/thread/handlers/query';
+import NavieService from '../../rpc/navie/services/navieService';
+import { ThreadIndexService } from '../../rpc/navie/services/threadIndexService';
+import { container } from 'tsyringe';
+import ThreadService from '../../rpc/navie/services/threadService';
+import {
+  navieThreadAddMessageAttachmentHandler,
+  navieThreadRemoveMessageAttachmentHandler,
+} from '../../rpc/navie/thread/handlers/messageAttachment';
 
 export const command = 'rpc';
 export const describe = 'Run AppMap JSON-RPC server';
@@ -55,6 +65,8 @@ type HandlerArguments = yargs.ArgumentsCamelCase<
 >;
 
 export function rpcMethods(navie: INavieProvider, codeEditor?: string): RpcHandler<any, any>[] {
+  const threadService = container.resolve(ThreadService);
+  const threadIndexService = container.resolve(ThreadIndexService);
   return [
     search(),
     appmapStatsV1(),
@@ -65,7 +77,6 @@ export function rpcMethods(navie: INavieProvider, codeEditor?: string): RpcHandl
     sequenceDiagram(),
     explainHandler(navie, codeEditor),
     explainStatusHandler(),
-    loadThreadHandler(),
     update(navie),
     setConfigurationV1(),
     getConfigurationV1(),
@@ -75,14 +86,24 @@ export function rpcMethods(navie: INavieProvider, codeEditor?: string): RpcHandl
     navieMetadataV2(),
     navieSuggestHandlerV1(navie),
     navieWelcomeV2(navie),
-    navieRegisterV1(codeEditor),
+    navieRegisterV1(threadService, codeEditor),
+    navieThreadSendMessageHandler(threadService),
+    navieThreadPinItemHandler(threadService),
+    navieThreadUnpinItemHandler(threadService),
+    navieThreadQueryHandler(threadIndexService),
+    navieThreadAddMessageAttachmentHandler(threadService),
+    navieThreadRemoveMessageAttachmentHandler(threadService),
   ];
 }
 
 export const handler = async (argv: HandlerArguments) => {
   verbose(argv.verbose);
 
   const navie = buildNavieProvider(argv);
+
+  ThreadIndexService.useDefault();
+  NavieService.bindNavieProvider(navie);
+
   let codeEditor: string | undefined = argv.codeEditor;
   if (!codeEditor) {
     codeEditor = detectCodeEditor();
@@ -92,12 +113,6 @@ export const handler = async (argv: HandlerArguments) => {
   loadConfiguration(false);
   await configureRpcDirectories(argv.directory);
 
-  {
-    const history = initializeHistory();
-    const oldHistoryDir = join(homedir(), '.appmap', 'navie', 'history');
-    await History.migrate(oldHistoryDir, history);
-  }
-
   const rpcServer = new RPCServer(argv.port, rpcMethods(navie, codeEditor));
   rpcServer.start();
 };

packages/cli/src/cmds/index/rpcServer.ts

@@ -11,6 +11,7 @@ import { Server } from 'http';
 
 import shadowLocalhost from '../../lib/shadowLocalhost';
 import { RpcCallback, RpcHandler, toJaysonRpcError } from '../../rpc/rpc';
+import { sseMiddleware } from '../../rpc/navie/thread/middleware';
 
 const debug = makeDebug('appmap:rpcServer');
 
@@ -63,6 +64,7 @@ export default class RPCServer {
     const app = connect();
     app.use(cors({ methods: ['POST'] }));
     app.use(jsonParser());
+    app.use(sseMiddleware());
     app.use(server.middleware());
     const listener = app.listen(this.bindPort, 'localhost');
 

packages/cli/src/rpc/explain/explain.ts

@@ -21,11 +21,8 @@ import configuration, { AppMapDirectory } from '../configuration';
 import { getLLMConfiguration } from '../llmConfiguration';
 import { RpcError, RpcHandler } from '../rpc';
 import collectContext, { buildContextRequest } from './collect-context';
-import { initializeHistory } from './navie/historyHelper';
-import { ThreadAccessError } from './navie/ihistory';
 import INavie, { INavieProvider } from './navie/inavie';
 import reportFetchError from './navie/report-fetch-error';
-import Thread from './navie/thread';
 import handleReview from './review';
 
 const searchStatusByUserMessageId = new Map<string, ExplainRpc.ExplainStatusResponse>();
@@ -351,21 +348,6 @@ export function explainStatus(userMessageId: string): ExplainRpc.ExplainStatusRe
   return searchStatus;
 }
 
-export async function loadThread(threadId: string): Promise<ExplainRpc.LoadThreadResponse> {
-  const history = initializeHistory();
-
-  let thread: Thread;
-  try {
-    thread = await history.load(threadId);
-  } catch (e) {
-    if (e instanceof ThreadAccessError) throw new RpcError(404, `Thread ${threadId} not found`);
-
-    throw e;
-  }
-
-  return thread;
-}
-
 const explainHandler: (
   navieProvider: INavieProvider,
   codeEditor: string | undefined
@@ -426,16 +408,4 @@ const explainStatusHandler: () => RpcHandler<
   };
 };
 
-const loadThreadHandler: () => RpcHandler<
-  ExplainRpc.LoadThreadOptions,
-  ExplainRpc.LoadThreadResponse
-> = () => {
-  return {
-    name: ExplainRpc.ExplainThreadLoadFunctionName,
-    handler: async (options: ExplainRpc.LoadThreadOptions) => {
-      return loadThread(options.threadId);
-    },
-  };
-};
-
-export { explainHandler, explainStatusHandler, loadThreadHandler };
+export { explainHandler, explainStatusHandler };