harbor-net/src/server/services/activityJournalService.ts

import crypto from 'node:crypto';
import fs from 'node:fs';
import {
  ACTIVITY_JOURNAL_MAX_EVENTS,
  ACTIVITY_JOURNAL_RETENTION_DAYS,
  normalizeActivityEventInput,
  normalizeStoredActivityEvent,
  type ActivityJournalEvent,
  type ActivityJournalEventInput,
  type ActivityJournalPage,
} from '../../shared/activityJournal.js';
import { createJsonStore } from './stateStore.js';

interface JournalState {
  schemaVersion: 1;
  events: ActivityJournalEvent[];
}

const migrateJournal = (value: unknown): JournalState => {
  const candidate = value && typeof value === 'object' && !Array.isArray(value)
    ? value as Record<string, unknown>
    : {};
  return {
    schemaVersion: 1,
    events: (Array.isArray(candidate.events) ? candidate.events : [])
      .map(normalizeStoredActivityEvent)
      .filter((event): event is ActivityJournalEvent => Boolean(event)),
  };
};

export function createActivityJournalService({
  filePath,
  now = () => new Date(),
}: {
  filePath: string;
  now?: () => Date;
}) {
  const store = createJsonStore<JournalState>({
    filePath,
    defaultValue: { schemaVersion: 1, events: [] },
    migrate: migrateJournal,
  });
  let recoveryRecorded = false;
  let writeFailed = false;

  function retained(events: ActivityJournalEvent[]) {
    const cutoff = now().getTime() - ACTIVITY_JOURNAL_RETENTION_DAYS * 86_400_000;
    return events
      .filter(({ occurredAt }) => Date.parse(occurredAt) >= cutoff)
      .slice(-ACTIVITY_JOURNAL_MAX_EVENTS);
  }

  function append(value: ActivityJournalEventInput) {
    const input = normalizeActivityEventInput(value);
    const storedInput = input.dedupeKey ? {
      ...input,
      dedupeKey: `${input.type}:sha256:${crypto.createHash('sha256').update(input.dedupeKey).digest('hex')}`,
    } : input;
    let appended: ActivityJournalEvent | null = null;
    try {
      store.update((state) => {
        const events = retained(state.events);
        if (storedInput.dedupeKey && events.some(({ dedupeKey }) => dedupeKey === storedInput.dedupeKey)) {
          return { schemaVersion: 1, events };
        }
        appended = {
          id: crypto.randomUUID(),
          occurredAt: now().toISOString(),
          ...storedInput,
        };
        return { schemaVersion: 1, events: retained([...events, appended]) };
      });
      writeFailed = false;
    } catch (error) {
      writeFailed = true;
      throw error;
    }
    return appended;
  }

  function ensureRecoveryEvent() {
    if (!store.recovery || recoveryRecorded) return;
    append({
      type: 'journal.recovered',
      severity: 'warning',
      source: 'storage',
      dedupeKey: `journal.recovered:${store.recovery.recoveredAt}`,
      data: {},
    });
    recoveryRecorded = true;
  }

  function page(limitValue: unknown = 50, cursorValue: unknown = null): ActivityJournalPage {
    try {
      let state = store.read();
      ensureRecoveryEvent();
      if (store.recovery) state = store.read();
      const retainedEvents = retained(state.events);
      if (retainedEvents.length !== state.events.length) {
        state = store.update(() => ({ schemaVersion: 1, events: retainedEvents }));
      }
      const events = [...state.events].reverse();
      const limit = Math.min(100, Math.max(1, Number.isSafeInteger(limitValue) ? Number(limitValue) : 50));
      const cursor = typeof cursorValue === 'string' ? cursorValue : '';
      const cursorIndex = cursor ? events.findIndex(({ id }) => id === cursor) : -1;
      if (cursor && cursorIndex < 0) return {
        events: [],
        nextCursor: null,
        retentionDays: 30,
        generatedAt: now().toISOString(),
        storage: writeFailed
          ? { status: 'error', errorCode: 'JOURNAL_UNAVAILABLE' }
          : { status: 'ready', errorCode: null },
      };
      const safeStart = cursorIndex + 1;
      const selected = events.slice(safeStart, safeStart + limit);
      return {
        events: selected.map((event) => ({ ...event, dedupeKey: null })),
        nextCursor: safeStart + selected.length < events.length ? selected.at(-1)?.id || null : null,
        retentionDays: 30,
        generatedAt: now().toISOString(),
        storage: writeFailed
          ? { status: 'error', errorCode: 'JOURNAL_UNAVAILABLE' }
          : { status: 'ready', errorCode: null },
      };
    } catch {
      return {
        events: [],
        nextCursor: null,
        retentionDays: 30,
        generatedAt: now().toISOString(),
        storage: { status: 'error', errorCode: 'JOURNAL_UNAVAILABLE' },
      };
    }
  }

  if (fs.existsSync(filePath)) {
    try {
      const state = store.read();
      const retainedEvents = retained(state.events);
      if (retainedEvents.length !== state.events.length) {
        store.update(() => ({ schemaVersion: 1, events: retainedEvents }));
      }
    } catch {
      writeFailed = true;
    }
  }

  return { append, page };
}

export type ActivityJournalService = ReturnType<typeof createActivityJournalService>;