feat: implement watch claude code project folders and sync state

src/server/hono/route.ts

@@ -1,3 +1,7 @@
+import { streamSSE } from "hono/streaming";
+import { getFileWatcher } from "../service/events/fileWatcher";
+import { sseEvent } from "../service/events/sseEvent";
+import type { WatcherEvent } from "../service/events/types";
 import { getProject } from "../service/project/getProject";
 import { getProjects } from "../service/project/getProjects";
 import { getSession } from "../service/session/getSession";
@@ -26,6 +30,118 @@ export const routes = (app: HonoAppType) => {
       const { projectId, sessionId } = c.req.param();
       const { session } = await getSession(projectId, sessionId);
       return c.json({ session });
+    })
+
+    .get("/events/state_changes", async (c) => {
+      return streamSSE(
+        c,
+        async (stream) => {
+          const fileWatcher = getFileWatcher();
+          let isConnected = true;
+          let eventId = 0;
+
+          // ハートビート設定
+          const heartbeat = setInterval(() => {
+            if (isConnected) {
+              stream
+                .writeSSE({
+                  data: sseEvent({
+                    type: "heartbeat",
+                    timestamp: new Date().toISOString(),
+                  }),
+                  event: "heartbeat",
+                  id: String(eventId++),
+                })
+                .catch(() => {
+                  console.warn("Failed to write SSE event");
+                  isConnected = false;
+                  onConnectionClosed();
+                });
+            }
+          }, 30 * 1000);
+
+          // connection handling
+          const abortController = new AbortController();
+          let connectionResolve: ((value: undefined) => void) | undefined;
+          const connectionPromise = new Promise<undefined>((resolve) => {
+            connectionResolve = resolve;
+          });
+
+          const onConnectionClosed = () => {
+            isConnected = false;
+            connectionResolve?.(undefined);
+            abortController.abort();
+            clearInterval(heartbeat);
+          };
+
+          // 接続終了時のクリーンアップ
+          stream.onAbort(() => {
+            console.log("SSE connection aborted");
+            onConnectionClosed();
+          });
+
+          // イベントリスナーを登録
+          console.log("Registering SSE event listeners");
+          fileWatcher.on("project_changed", async (event: WatcherEvent) => {
+            if (!isConnected) {
+              return;
+            }
+
+            if (event.eventType !== "project_changed") {
+              return;
+            }
+
+            await stream
+              .writeSSE({
+                data: sseEvent({
+                  type: event.eventType,
+                  ...event.data,
+                }),
+                event: event.eventType,
+                id: String(eventId++),
+              })
+              .catch(() => {
+                console.warn("Failed to write SSE event");
+                onConnectionClosed();
+              });
+          });
+          fileWatcher.on("session_changed", async (event: WatcherEvent) => {
+            if (!isConnected) {
+              return;
+            }
+
+            await stream
+              .writeSSE({
+                data: sseEvent({
+                  ...event.data,
+                  type: event.eventType,
+                }),
+                event: event.eventType,
+                id: String(eventId++),
+              })
+              .catch(() => {
+                onConnectionClosed();
+              });
+          });
+
+          // 初期接続確認メッセージ
+          await stream.writeSSE({
+            data: sseEvent({
+              type: "connected",
+              message: "SSE connection established",
+              timestamp: new Date().toISOString(),
+            }),
+            event: "connected",
+            id: String(eventId++),
+          });
+
+          await connectionPromise;
+        },
+        async (err, stream) => {
+          console.error("Streaming error:", err);
+          await stream.write("エラーが発生しました。");
+        },
+      );
     });
 };
 

src/server/service/events/fileWatcher.ts

@@ -0,0 +1,83 @@
+import { EventEmitter } from "node:events";
+import { type FSWatcher, watch } from "node:fs";
+import z from "zod";
+import { claudeProjectPath } from "../paths";
+import type { WatcherEvent } from "./types";
+
+const fileRegExp = /(?<projectId>.*?)\/(?<sessionId>.*?)\.jsonl/;
+const fileRegExpGroupSchema = z.object({
+  projectId: z.string(),
+  sessionId: z.string(),
+});
+
+export class FileWatcherService extends EventEmitter {
+  private watcher: FSWatcher | null = null;
+  private projectWatchers: Map<string, FSWatcher> = new Map();
+
+  constructor() {
+    super();
+    this.startWatching();
+  }
+
+  private startWatching(): void {
+    try {
+      console.log("Starting file watcher on:", claudeProjectPath);
+      // メインプロジェクトディレクトリを監視
+      this.watcher = watch(
+        claudeProjectPath,
+        { persistent: false, recursive: true },
+        (eventType, filename) => {
+          if (!filename) return;
+
+          const groups = fileRegExpGroupSchema.safeParse(
+            filename.match(fileRegExp)?.groups,
+          );
+
+          if (!groups.success) return;
+
+          const { projectId, sessionId } = groups.data;
+
+          this.emit("project_changed", {
+            eventType: "project_changed",
+            data: { projectId, fileEventType: eventType },
+          } satisfies WatcherEvent);
+
+          this.emit("session_changed", {
+            eventType: "session_changed",
+            data: {
+              projectId,
+              sessionId,
+              fileEventType: eventType,
+            },
+          } satisfies WatcherEvent);
+        },
+      );
+      console.log("File watcher initialization completed");
+    } catch (error) {
+      console.error("Failed to start file watching:", error);
+    }
+  }
+
+  public stop(): void {
+    if (this.watcher) {
+      this.watcher.close();
+      this.watcher = null;
+    }
+
+    for (const [, watcher] of this.projectWatchers) {
+      watcher.close();
+    }
+    this.projectWatchers.clear();
+  }
+}
+
+// シングルトンインスタンス
+let watcherInstance: FileWatcherService | null = null;
+
+export const getFileWatcher = (): FileWatcherService => {
+  if (!watcherInstance) {
+    console.log("Creating new FileWatcher instance");
+    watcherInstance = new FileWatcherService();
+  }
+  return watcherInstance;
+};

src/server/service/events/sseEvent.ts

@@ -0,0 +1,13 @@
+import type { BaseSSEEvent, SSEEvent } from "./types";
+
+let eventId = 0;
+
+export const sseEvent = <D extends Omit<SSEEvent, "id" | "timestamp">>(
+  data: D,
+): string => {
+  return JSON.stringify({
+    ...data,
+    id: String(eventId++),
+    timestamp: new Date().toISOString(),
+  } satisfies D & BaseSSEEvent);
+};

src/server/service/events/types.ts

@@ -0,0 +1,50 @@
+import type { WatchEventType } from "node:fs";
+
+export type WatcherEvent =
+  | {
+      eventType: "project_changed";
+      data: ProjectChangedData;
+    }
+  | {
+      eventType: "session_changed";
+      data: SessionChangedData;
+    };
+
+export type BaseSSEEvent = {
+  id: string;
+  timestamp: string;
+};
+
+export type SSEEvent = BaseSSEEvent &
+  (
+    | {
+        type: "connected";
+        message: string;
+        timestamp: string;
+      }
+    | {
+        type: "heartbeat";
+        timestamp: string;
+      }
+    | {
+        id: string;
+        type: "project_changed";
+        data: ProjectChangedData;
+      }
+    | {
+        id: string;
+        type: "session_changed";
+        data: SessionChangedData;
+      }
+  );
+
+export interface ProjectChangedData {
+  projectId: string;
+  fileEventType: WatchEventType;
+}
+
+export interface SessionChangedData {
+  projectId: string;
+  sessionId: string;
+  fileEventType: WatchEventType;
+}