Merge PR #472: Activity feed load from disk

Activity feed: load recent history from disk

Author: web3dev1337

server/activityFeedService.js

@@ -19,6 +19,7 @@ const logger = winston.createLogger({
 const DEFAULT_DIR = path.join(os.homedir(), '.orchestrator');
 const DEFAULT_FILE = path.join(DEFAULT_DIR, 'activity.jsonl');
 const DEFAULT_MAX_EVENTS = 500;
+const DEFAULT_LOAD_MAX_BYTES = 1024 * 1024; // 1MB tail read
 
 function safeJsonLine(obj) {
   try {
@@ -28,12 +29,34 @@ function safeJsonLine(obj) {
   }
 }
 
+function safeJsonParse(line) {
+  try {
+    return JSON.parse(line);
+  } catch {
+    return null;
+  }
+}
+
 function clampInt(n, { min, max, fallback }) {
   const v = Number.parseInt(String(n || ''), 10);
   if (!Number.isFinite(v)) return fallback;
   return Math.max(min, Math.min(max, v));
 }
 
+function dedupeByIdKeepLast(items) {
+  const seen = new Set();
+  const out = [];
+  for (let i = items.length - 1; i >= 0; i -= 1) {
+    const ev = items[i];
+    const id = ev?.id;
+    if (!id || seen.has(id)) continue;
+    seen.add(id);
+    out.push(ev);
+  }
+  out.reverse();
+  return out;
+}
+
 class ActivityFeedService {
   constructor({ filePath = DEFAULT_FILE, maxEvents = DEFAULT_MAX_EVENTS } = {}) {
     this.filePath = filePath;
@@ -59,13 +82,55 @@ class ActivityFeedService {
     this._loaded = true;
     try {
       await fs.mkdir(path.dirname(this.filePath), { recursive: true });
-      // Do not load entire history into memory; keep v1 in-memory only.
-      // Existence is enough to ensure appends won't throw due to missing dir.
+      await this.loadRecentFromDisk();
     } catch (error) {
       logger.warn('Failed to ensure activity dir exists', { error: error.message });
     }
   }
 
+  async loadRecentFromDisk({ maxBytes = DEFAULT_LOAD_MAX_BYTES } = {}) {
+    const byteBudget = clampInt(maxBytes, { min: 16 * 1024, max: 10 * 1024 * 1024, fallback: DEFAULT_LOAD_MAX_BYTES });
+    try {
+      const stat = await fs.stat(this.filePath);
+      const size = Number(stat.size) || 0;
+      if (!size) return;
+
+      const start = Math.max(0, size - byteBudget);
+      const length = Math.max(0, size - start);
+      if (!length) return;
+
+      const handle = await fs.open(this.filePath, 'r');
+      try {
+        const buf = Buffer.alloc(length);
+        const { bytesRead } = await handle.read(buf, 0, length, start);
+        const text = buf.toString('utf8', 0, bytesRead);
+        let lines = text.split('\n').filter(Boolean);
+        // If we started mid-file, first line may be partial; drop it.
+        if (start > 0 && lines.length > 0) {
+          lines = lines.slice(1);
+        }
+
+        const parsed = [];
+        for (const line of lines) {
+          const ev = safeJsonParse(line);
+          if (!ev || typeof ev !== 'object') continue;
+          if (!ev.id || !ev.kind || !ev.ts) continue;
+          parsed.push(ev);
+        }
+
+        const merged = dedupeByIdKeepLast([...parsed, ...this.events]);
+        this.events = merged.slice(-this.maxEvents);
+      } finally {
+        await handle.close();
+      }
+    } catch (error) {
+      // If the file doesn't exist, that's fine. Any other errors should be logged but not fatal.
+      if (String(error?.code || '') !== 'ENOENT') {
+        logger.warn('Failed to load activity history from disk', { error: error.message });
+      }
+    }
+  }
+
   list({ since = 0, limit = 200 } = {}) {
     const sinceMs = Number.isFinite(Number(since)) ? Number(since) : 0;
     const lim = clampInt(limit, { min: 1, max: 1000, fallback: 200 });
@@ -114,4 +179,3 @@ class ActivityFeedService {
 }
 
 module.exports = { ActivityFeedService };
-

tests/unit/activityFeedService.test.js

@@ -33,6 +33,24 @@ describe('ActivityFeedService', () => {
     nowSpy.mockRestore();
   });
 
+  test('ensureLoaded() loads recent events from existing JSONL file', async () => {
+    const filePath = nextTmpFile('activity.jsonl');
+    await fs.mkdir(path.dirname(filePath), { recursive: true });
+    await fs.writeFile(filePath, [
+      JSON.stringify({ id: 'a', ts: 1, kind: 'k1', data: { n: 1 } }),
+      JSON.stringify({ id: 'b', ts: 2, kind: 'k2', data: { n: 2 } })
+    ].join('\n') + '\n', 'utf8');
+
+    const svc = new ActivityFeedService({ filePath, maxEvents: 10 });
+    expect(svc.list({ limit: 10 })).toHaveLength(0);
+
+    await svc.ensureLoaded();
+    const list = svc.list({ limit: 10 });
+    expect(list).toHaveLength(2);
+    expect(list[0].id).toBe('b');
+    expect(list[1].id).toBe('a');
+  });
+
   test('track() caps in-memory events to maxEvents', () => {
     const svc = new ActivityFeedService({ filePath: nextTmpFile('activity.jsonl'), maxEvents: 2 });
     const nowSpy = jest.spyOn(Date, 'now');
@@ -74,4 +92,3 @@ describe('ActivityFeedService', () => {
     nowSpy.mockRestore();
   });
 });
-