/**
 * Redis-backed distributed crawl frontier.
 *
 * Replaces the in-memory CrawlFrontier for horizontally-scaled crawler deployments.
 * Multiple crawler workers on different containers can share the same URL queue for
 * a given job — enabling 10x crawl throughput on large enterprise apps.
 *
 * Redis key namespace:  zeta:frontier:{jobId}
 *   LIST  zeta:frontier:{jobId}:queue   — BFS queue (LPUSH/RPOP)
 *   SET   zeta:frontier:{jobId}:visited — deduplication set
 *   EXPIRE                               — 24h TTL on all keys
 *
 * Graceful fallback: if REDIS_URL is not set, RedisCrawlFrontier silently falls
 * back to in-memory arrays so local development works without Redis.
 */

const FRONTIER_TTL = 86400; // 24h in seconds

export class RedisCrawlFrontier {
  private redis: any;
  private queueKey: string;
  private visitedKey: string;
  private inMemoryQueue: string[] = [];
  private inMemoryVisited = new Set<string>();
  private useRedis: boolean;

  constructor(private readonly jobId: string) {
    this.queueKey = `zeta:frontier:${jobId}:queue`;
    this.visitedKey = `zeta:frontier:${jobId}:visited`;
    this.useRedis = !!process.env.REDIS_URL;
  }

  async init(): Promise<void> {
    if (!this.useRedis) return;
    try {
      const Redis: any = await import('ioredis').then(m => m.default ?? m);
      this.redis = new Redis(process.env.REDIS_URL!, {
        maxRetriesPerRequest: 2,
        enableReadyCheck: false,
        lazyConnect: true,
      });
      await this.redis.connect().catch(() => {
        this.useRedis = false;
        this.redis = null;
      });
    } catch {
      this.useRedis = false;
      this.redis = null;
    }
  }

  /** Add a URL to the frontier. No-op if already visited. Returns true if added. */
  async push(url: string): Promise<boolean> {
    if (this.useRedis && this.redis) {
      try {
        const isNew = await this.redis.sadd(this.visitedKey, url);
        if (!isNew) return false;
        await this.redis.lpush(this.queueKey, url);
        await this.redis.expire(this.queueKey, FRONTIER_TTL);
        await this.redis.expire(this.visitedKey, FRONTIER_TTL);
        return true;
      } catch {
        // Fall through to in-memory on Redis error
      }
    }
    if (this.inMemoryVisited.has(url)) return false;
    this.inMemoryVisited.add(url);
    this.inMemoryQueue.push(url);
    return true;
  }

  /** Pop the next URL (FIFO/BFS). Returns null if queue empty. */
  async shift(): Promise<string | null> {
    if (this.useRedis && this.redis) {
      try {
        const url = await this.redis.rpop(this.queueKey);
        return url ?? null;
      } catch {
        // Fall through
      }
    }
    return this.inMemoryQueue.shift() ?? null;
  }

  /** Check if a URL has already been visited/queued. */
  async hasVisited(url: string): Promise<boolean> {
    if (this.useRedis && this.redis) {
      try {
        return (await this.redis.sismember(this.visitedKey, url)) === 1;
      } catch { /* fall through */ }
    }
    return this.inMemoryVisited.has(url);
  }

  /** Current queue length. */
  async length(): Promise<number> {
    if (this.useRedis && this.redis) {
      try {
        return await this.redis.llen(this.queueKey);
      } catch { /* fall through */ }
    }
    return this.inMemoryQueue.length;
  }

  /** Total visited count. */
  async visitedCount(): Promise<number> {
    if (this.useRedis && this.redis) {
      try {
        return await this.redis.scard(this.visitedKey);
      } catch { /* fall through */ }
    }
    return this.inMemoryVisited.size;
  }

  /** Seed initial URLs (used at job start). */
  async seed(urls: string[]): Promise<void> {
    for (const url of urls) await this.push(url);
  }

  /** Clean up all Redis keys for this frontier. Call at job completion. */
  async cleanup(): Promise<void> {
    if (this.useRedis && this.redis) {
      try {
        await this.redis.del(this.queueKey, this.visitedKey);
      } catch { /* ignore */ }
      try { await this.redis.quit(); } catch { /* ignore */ }
    }
  }

  /** Stats for monitoring. */
  async stats(): Promise<{ backend: 'redis' | 'memory'; queued: number; visited: number }> {
    return {
      backend: this.useRedis && this.redis ? 'redis' : 'memory',
      queued: await this.length(),
      visited: await this.visitedCount(),
    };
  }
}

/** Check if Redis is configured for distributed frontier. */
export function isRedisFrontierEnabled(): boolean {
  return !!process.env.REDIS_URL;
}