Event Journal

import { ServerCoordinator } from '@topgunbuild/server';
import { Pool } from 'pg';

const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
});

const server = new ServerCoordinator({
  port: 8080,
  storage: new PostgresAdapter({ pool }),

  // Enable Event Journal
  eventJournalEnabled: true,
  eventJournalConfig: {
    capacity: 10000,        // Max events in memory
    ttlMs: 0,               // Time-to-live (0 = infinite)
    persistent: true,       // Persist to PostgreSQL
    persistBatchSize: 100,  // Batch size for DB writes
    persistIntervalMs: 1000, // Flush interval
    includeMaps: [],        // Filter: only these maps (empty = all)
    excludeMaps: ['_internal'], // Filter: exclude these maps
  },
});

await server.start();

import { TopGunClient } from '@topgunbuild/client';

const client = new TopGunClient({
  serverUrl: 'ws://localhost:8080',
});

// Get journal reader
const journal = client.getEventJournal();

// Subscribe to all events
const unsubscribe = journal.subscribe((event) => {
  console.log(`[${event.type}] ${event.mapName}:${event.key}`);
  console.log('  Value:', event.value);
  console.log('  Previous:', event.previousValue);
  console.log('  Sequence:', event.sequence.toString());
}, {
  fromSequence: 0n,  // Start from beginning (optional)
  mapName: 'orders', // Filter by map (optional)
  types: ['PUT'],    // Filter by type (optional)
});

// Later: unsubscribe
unsubscribe();

// Read historical events
const journal = client.getEventJournal();

// Get latest sequence
const latestSeq = await journal.getLatestSequence();
console.log('Latest sequence:', latestSeq.toString());

// Read last 100 events
const events = await journal.readFrom(latestSeq - 100n, 100);

for (const event of events) {
  console.log(`#${event.sequence}: ${event.type} ${event.mapName}:${event.key}`);
}

// Read events for specific map
const orderEvents = await journal.readMapEvents('orders', 0n, 50);

import { useEventJournal } from '@topgunbuild/react';

function ActivityFeed() {
  const { events, lastEvent, isSubscribed, clearEvents } = useEventJournal({
    mapName: 'orders',
    types: ['PUT', 'UPDATE'],
    maxEvents: 50,
    onEvent: (event) => {
      // Show notification for new orders
      if (event.type === 'PUT') {
        showToast(`New order: ${event.key}`);
      }
    },
  });

  return (
    <div>
      <div className="flex justify-between">
        <h2>Activity {isSubscribed && '🟢'}</h2>
        <button onClick={clearEvents}>Clear</button>
      </div>
      <ul>
        {events.map((e) => (
          <li key={e.sequence.toString()}>
            <span className="badge">{e.type}</span>
            <span>{e.mapName}:{e.key}</span>
            <time>{new Date(e.timestamp.millis).toLocaleTimeString()}</time>
          </li>
        ))}
      </ul>
    </div>
  );
}

import { useEventJournal } from '@topgunbuild/react';

function AuditLog({ mapName }) {
  const { events, readFrom, getLatestSequence } = useEventJournal({
    mapName,
    maxEvents: 100,
  });

  const [history, setHistory] = useState([]);
  const [loading, setLoading] = useState(false);

  const loadMore = async () => {
    setLoading(true);
    const oldestSeq = history.length > 0
      ? history[0].sequence - 1n
      : await getLatestSequence();

    const older = await readFrom(oldestSeq - 50n, 50);
    setHistory([...older, ...history]);
    setLoading(false);
  };

  return (
    <div>
      <h2>Audit Log: {mapName}</h2>
      <button onClick={loadMore} disabled={loading}>
        {loading ? 'Loading...' : 'Load More'}
      </button>
      <table>
        <thead>
          <tr>
            <th>Time</th>
            <th>Action</th>
            <th>Key</th>
            <th>Changes</th>
          </tr>
        </thead>
        <tbody>
          {[...history, ...events].map((e) => (
            <tr key={e.sequence.toString()}>
              <td>{new Date(e.timestamp.millis).toISOString()}</td>
              <td>{e.type}</td>
              <td>{e.key}</td>
              <td>
                {e.type === 'DELETE'
                  ? `Deleted: ${JSON.stringify(e.previousValue)}`
                  : JSON.stringify(e.value)}
              </td>
            </tr>
          ))}
        </tbody>
      </table>
    </div>
  );
}

// Server-side: Export to Kafka
import { Kafka } from 'kafkajs';

const kafka = new Kafka({ brokers: ['localhost:9092'] });
const producer = kafka.producer();
await producer.connect();

// Subscribe to journal and forward to Kafka
server.eventJournalService.subscribe(async (event) => {
  await producer.send({
    topic: `topgun.${event.mapName}`,
    messages: [{
      key: event.key,
      value: JSON.stringify({
        type: event.type,
        value: event.value,
        previousValue: event.previousValue,
        timestamp: event.timestamp,
      }),
      headers: {
        'sequence': event.sequence.toString(),
        'nodeId': event.nodeId,
      },
    }],
  });
});

// Or use NDJSON stream export
const stream = server.eventJournalService.exportStream({
  fromSequence: 0n,
  mapName: 'orders',
  types: ['PUT', 'UPDATE'],
});

const reader = stream.getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  console.log(value); // NDJSON line
}

// Server-side: Send events to ML pipeline for real-time processing
server.eventJournalService.subscribe(async (event) => {
  // Only process user-generated content
  if (event.mapName !== 'posts' && event.mapName !== 'comments') return;
  if (event.type === 'DELETE') return;

  const content = event.value?.content;
  if (!content) return;

  // Send to ML service for async processing
  await fetch('https://ml-pipeline.example.com/enqueue', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      eventId: event.sequence.toString(),
      mapName: event.mapName,
      key: event.key,
      content,
      tasks: ['sentiment', 'toxicity', 'embedding'],
    }),
  });
}, { mapName: 'posts' });

// Alternative: Batch processing with windowing
const eventBuffer: JournalEvent[] = [];

server.eventJournalService.subscribe((event) => {
  eventBuffer.push(event);
}, { types: ['PUT', 'UPDATE'] });

// Flush batch every 5 seconds
setInterval(async () => {
  if (eventBuffer.length === 0) return;

  const batch = eventBuffer.splice(0, 100);
  await fetch('https://ml-pipeline.example.com/batch', {
    method: 'POST',
    body: JSON.stringify({ events: batch }),
  });
}, 5000);

// Server-side: Cleanup old events
// Run this periodically (e.g., daily cron job)

// Clean events older than 30 days
const deletedCount = await server.eventJournalService.cleanupOldEvents(30);
console.log(`Cleaned up ${deletedCount} old journal events`);

// Or query with date filters
const recentEvents = await server.eventJournalService.queryFromStorage({
  mapName: 'orders',
  fromDate: new Date(Date.now() - 7 * 24 * 60 * 60 * 1000), // Last 7 days
  limit: 1000,
});