ograph/packages/dispatcher/src/scheduler.ts

// Loop B: OC Scheduler
// Polls OC busy/idle state, pushes pending queue when OC is available.

import { writeFileSync, mkdirSync } from 'node:fs';
import { join } from 'node:path';
import { execFileSync } from 'node:child_process';
import type { DispatcherConfig, PendingEntry } from './types.js';
import { OcClient } from './oc-client.js';

function ts(): string {
  return new Date().toISOString();
}

/** Dispatch file path — a simple push mechanism readable by OC / other tools */
const DISPATCH_FILE = '/tmp/ograph-dispatch.json';

export class OcScheduler {
  private readonly client: OcClient;
  private running = false;
  private timer: ReturnType<typeof setTimeout> | null = null;
  private lastPushAt = 0;

  constructor(
    private readonly config: DispatcherConfig,
    /** Shared pending queue (read + cleared by Scheduler) */
    private readonly pending: Map<string, PendingEntry>,
  ) {
    this.client = new OcClient(config);
  }

  start(): void {
    if (this.running) return;
    this.running = true;
    console.log(`[${ts()}] [scheduler] started — OC status: ${this.config.oc.statusEndpoint}`);
    void this.poll();
  }

  stop(): void {
    this.running = false;
    if (this.timer !== null) {
      clearTimeout(this.timer);
      this.timer = null;
    }
    console.log(`[${ts()}] [scheduler] stopped`);
  }

  private scheduleNext(delayMs: number): void {
    if (!this.running) return;
    this.timer = setTimeout(() => {
      void this.poll();
    }, delayMs);
  }

  private async poll(): Promise<void> {
    if (!this.running) return;

    const { schedulerIdle, schedulerActive, cooldownAfterPush } = this.config.intervals;

    try {
      const hasPending = this.pending.size > 0;

      // Check cooldown first
      const now = Date.now();
      if (this.lastPushAt > 0 && now - this.lastPushAt < cooldownAfterPush) {
        const remainMs = cooldownAfterPush - (now - this.lastPushAt);
        console.log(`[${ts()}] [scheduler] in cooldown — ${Math.ceil(remainMs / 1000)}s remaining`);
        this.scheduleNext(hasPending ? schedulerActive : schedulerIdle);
        return;
      }

      if (!hasPending) {
        // Nothing to push; low-frequency idle polling
        this.scheduleNext(schedulerIdle);
        return;
      }

      // We have pending items — check OC availability
      let available = false;
      try {
        available = await this.client.isAvailable();
        console.log(`[${ts()}] [scheduler] OC available=${available} pending=${this.pending.size}`);
      } catch (err) {
        console.warn(`[${ts()}] [scheduler] OC status check failed: ${err instanceof Error ? err.message : String(err)}`);
        // Can't determine status — back off
        this.scheduleNext(schedulerActive);
        return;
      }

      if (available) {
        await this.push();
        this.lastPushAt = Date.now();
        this.scheduleNext(schedulerIdle); // after push, slow down
      } else {
        // OC busy but we have work — keep polling actively
        this.scheduleNext(schedulerActive);
      }
    } catch (err) {
      console.error(`[${ts()}] [scheduler] poll error: ${err instanceof Error ? err.message : String(err)}`);
      this.scheduleNext(schedulerActive);
    }
  }

  private async push(): Promise<void> {
    const entries = Array.from(this.pending.values());
    if (entries.length === 0) return;

    const message = this.buildMessage(entries);

    console.log(`[${ts()}] [scheduler] pushing ${entries.length} change(s) to OC`);
    console.log(`[${ts()}] [scheduler] message:\n${message}`);

    // ── Strategy 1: write dispatch file ─────────────────────────────────────
    try {
      mkdirSync('/tmp', { recursive: true });
      const payload = {
        pushedAt: new Date().toISOString(),
        changes: entries,
        message,
      };
      writeFileSync(DISPATCH_FILE, JSON.stringify(payload, null, 2), 'utf-8');
      console.log(`[${ts()}] [scheduler] dispatch file written: ${DISPATCH_FILE}`);
    } catch (err) {
      console.warn(`[${ts()}] [scheduler] failed to write dispatch file: ${err instanceof Error ? err.message : String(err)}`);
    }

    // ── Strategy 2: openclaw message send (best-effort) ──────────────────────
    try {
      execFileSync('openclaw', ['message', 'send', message], { timeout: 10_000, stdio: 'pipe' });
      console.log(`[${ts()}] [scheduler] openclaw message send OK`);
    } catch (err) {
      // Not fatal — openclaw may not be in PATH or the channel may not be configured
      console.warn(`[${ts()}] [scheduler] openclaw message send failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`);
    }

    // Clear pending after successful push attempt
    this.pending.clear();
    console.log(`[${ts()}] [scheduler] pending queue cleared`);
  }

  private buildMessage(entries: PendingEntry[]): string {
    const lines: string[] = ['🔔 OGraph Projection Changes Detected\n'];

    for (const entry of entries) {
      const age = Math.round((Date.now() - entry.firstDetectedAt) / 1000);
      lines.push(`• **${entry.name}**`);
      lines.push(`  Changes: ${entry.changeCount}  |  Age: ${age}s`);
      lines.push(`  Previous: ${JSON.stringify(entry.previousValue)}`);
      lines.push(`  Current:  ${JSON.stringify(entry.currentValue)}`);
      lines.push('');
    }

    lines.push(`Total: ${entries.length} projection(s) changed`);
    return lines.join('\n');
  }
}