import { drainRetention } from '../retention.js';
import { collectFleetChat } from '../fleet/chat.js';
import { runAutoscaling, pruneAutoscaling } from '../autoscaling/index.js';
import { runRebalancing, pruneRebalancing } from '../rebalancing/index.js';
import { config } from '../config.js';
import { processProvisioning } from '../provisioning/worker.js';
import { sweep } from './sweep.js';
import { betaTick } from './beta-runner.js';
import { emit } from './bus.js';
import type { ServerStatusEntry } from './bus.js';
import type { RconFailure } from '../integrations/rcon.js';
import { setServerFields, type ServerRow } from '../fleet/servers.js';
import {
    inspectForMonitoring,
    rconForMonitoring,
    prepareMonitoringRcon,
    monitoringStatus,
    monitoringStatusWriter,
    monitoringIdentity,
    MonitoringChangedError,
    MonitoringSkippedError
} from '../fleet/health-observations.js';
import { serverObservation } from '../fleet/status.js';
import { successfulStatusFields, activityFields } from '../fleet/status-updates.js';
import { STATUS_REFRESH_MS } from '../../server-freshness.js';
import { recordMetrics } from '../fleet/metrics.js';
import { selectedZombieMap } from '../../map-aliases.js';
import { runMaintenance } from '../maintenance/index.js';
import { cleanupControllerLibraries } from '../game-files/index.js';
import { trackActivity } from './activity.js';
import { MonitoringSnapshot } from '../fleet/monitoring-snapshot.js';
import { observedPlayerCapacity } from '../fleet/player-capacity.js';
import type { PlayerCapacityObservation } from '../../server-capacity.js';
import { withHostRequestContext } from '../hosts/request-context.js';
import { withOperationSource } from '../hosts/operation-context.js';
import { failureReplySnippet } from '../integrations/rcon-diagnostics.js';

const TICK_MS = 15_000;
const RETENTION_INTERVAL_MS = 5 * 60_000;
// Match each agent's two health lanes without making unrelated hosts wait.
const STATUS_CONCURRENCY_PER_HOST = 2;

let started = false;
let tickTimer: NodeJS.Timeout | null = null;
let statusTimer: NodeJS.Timeout | null = null;

let tickInFlight: Promise<void> | null = null;
let tickRequested = false;
const requesterFlights = new Map<string, Promise<void>>();
let nextRetentionAt = 0;

function tick(requestAnother = false): Promise<void> {
    if (requestAnother) tickRequested = true;
    if (tickInFlight) return tickInFlight;
    tickInFlight = (async () => {
        do {
            tickRequested = false;
            try {
                await trackActivity('provisioning', 'Processing provisioning requests', () =>
                    processProvisioning({ wait: false })
                );
            } catch (err) {
                console.error('[worker] provisioning error:', err);
            }
        } while (tickRequested);
    })().finally(() => {
        tickInFlight = null;
    });
    return tickInFlight;
}

function reconcileRequesters(): void {
    // Observers submit durable intent independently. Provisioning still owns
    // admission and rechecks source eligibility immediately before mutation.
    const requesters: [string, string, () => Promise<unknown>][] = [
        ['campaign', 'Reconciling campaigns', sweep],
        ['beta', 'Reconciling beta tests', betaTick],
        ['maintenance', 'Checking maintenance and recovery', runMaintenance],
        ['autoscaling', 'Evaluating lobby autoscaling', runAutoscaling],
        ['rebalancing', 'Reconciling rebalance moves', runRebalancing],
        ['game-files', 'Removing unused game files', cleanupControllerLibraries],
        ['chat', 'Saving match chat', collectFleetChat],
        [
            'retention',
            'Cleaning expired history',
            async () => {
                await drainRetention();
                pruneAutoscaling();
                pruneRebalancing();
            }
        ]
    ];
    for (const [key, label, work] of requesters) {
        if (requesterFlights.has(key)) continue;
        if (key === 'retention') {
            if (Date.now() < nextRetentionAt) continue;
            nextRetentionAt = Date.now() + RETENTION_INTERVAL_MS;
        }
        const flight = trackActivity(key, label, work)
            .catch((err) => {
                console.error(`[worker] ${key} error:`, err);
            })
            .then(async (result) => {
                if (
                    key !== 'game-files' &&
                    key !== 'retention' &&
                    key !== 'chat' &&
                    (!['autoscaling', 'rebalancing'].includes(key) || result === true)
                )
                    await tick(true);
            })
            .finally(() => {
                requesterFlights.delete(key);
            });
        requesterFlights.set(key, flight);
    }
}

/**
 * Encode a structured RCON failure as `"reason:detail"` so `last_error`
 * stays a single column but the UI can split it apart when rendering the
 * diagnostic panel. Reasons are stable; details are human-facing.
 */
function encodeError(reason: RconFailure, detail: string): string {
    return `${reason}:${detail}`;
}

// Deduplicate queued/running checks per server. A slow server must not prevent
// completed servers from being polled again on the next 30-second tick.
const pendingStatus = new Map<string, Promise<number | null>>();
interface StatusQueue {
    active: number;
    queued: Array<() => Promise<void>>;
}
const statusQueues = new Map<string, StatusQueue>();
const changedStatus = new Map<string, ServerStatusEntry>();
function publishStatus(entry: ServerStatusEntry) {
    const scheduled = changedStatus.size > 0;
    changedStatus.set(entry.server, entry);
    if (!scheduled)
        queueMicrotask(() => {
            const entries = [...changedStatus.values()];
            changedStatus.clear();
            emit({ type: 'server-status', delta: true, entries });
        });
}

function drainStatusQueue(host: string, queue: StatusQueue): void {
    while (queue.active < STATUS_CONCURRENCY_PER_HOST && queue.queued.length) {
        const next = queue.queued.shift()!;
        queue.active++;
        void next().finally(() => {
            queue.active--;
            drainStatusQueue(host, queue);
        });
    }
    if (!queue.active && !queue.queued.length) statusQueues.delete(host);
}

function queueStatus(
    srv: ServerRow,
    snapshot: MonitoringSnapshot,
    probe?: () => ReturnType<typeof rconForMonitoring>
): Promise<number | null> {
    const server = srv.name;
    const pending = pendingStatus.get(server);
    if (pending) return pending;
    // Host queues keep slow instances from consuming other hosts' slots.
    const host = srv.host_id ? `native:${srv.host_id}` : `unregistered:${srv.name}`;
    let queue = statusQueues.get(host);
    if (!queue) {
        queue = { active: 0, queued: [] };
        statusQueues.set(host, queue);
    }
    let finish!: (capacity: number | null) => void;
    const work = new Promise<number | null>((resolve) => {
        finish = resolve;
    });
    pendingStatus.set(server, work);
    queue.queued.push(async () => {
        let capacity: number | null = null;
        try {
            capacity = await withOperationSource('Status monitor', () =>
                refreshServerStatus(server, snapshot, probe)
            );
            // Publish this server's latest evidence without waiting for the fleet.
            const current = snapshot.get(server)?.server;
            if (current)
                publishStatus(
                    statusEntry(current, snapshot.observe, observedPlayerCapacity(server, snapshot))
                );
        } catch (err) {
            console.error(`[worker] status error for ${server}:`, err);
        } finally {
            pendingStatus.delete(server);
            finish(capacity);
        }
    });
    drainStatusQueue(host, queue);
    return work;
}

export function fanoutStatus(): Promise<void> {
    return withHostRequestContext(statusPass);
}

async function statusPass(): Promise<void> {
    const snapshot = new MonitoringSnapshot();
    const servers = snapshot.servers();
    if (!servers.length) return;
    const probes = prepareMonitoringRcon(
        servers
            .filter(
                (server) =>
                    !pendingStatus.has(server.name) &&
                    !snapshot.provisioning(server.name) &&
                    !snapshot.settling(server.name) &&
                    !snapshot.canSkipRcon(server.name)
            )
            .map((server) => server.name),
        snapshot
    );
    const results = await Promise.all(
        servers.map((server) => queueStatus(server, snapshot, probes.get(server.name)))
    );
    const capacities = new Map(servers.map((server, i) => [server.name, results[i]]));
    try {
        // Metrics remain complete fleet collections, using freshness at capture
        // time rather than the status saved at the beginning of a long scan.
        snapshot.refresh();
        recordMetrics(
            snapshot.servers().map((srv) => {
                const current = snapshot.observe(srv);
                return {
                    server: srv.name,
                    game: srv.game ?? 'unknown',
                    map:
                        selectedZombieMap(srv.game, current.last_map ?? srv.map, srv.map_slug)
                            ?.friendly ??
                        current.last_map ??
                        srv.map ??
                        'Unknown map',
                    players: current.health === 'online' ? current.last_players : null,
                    capacity: capacities.get(srv.name) ?? null
                };
            })
        );
    } catch (err) {
        console.error('[worker] metrics error:', err);
    }
}

async function refreshServerStatus(
    server: string,
    snapshot: MonitoringSnapshot,
    probeStatus?: () => ReturnType<typeof rconForMonitoring>
): Promise<number | null> {
    snapshot.refresh(server);
    const srv = snapshot.get(server)?.server;
    if (!srv) return null;
    let capacity: number | null = null;
    let updates: Partial<Record<keyof ServerRow, string | number | null>> = {};
    const flush = () => {
        setServerFields(srv.name, updates);
        snapshot.patch(srv.name, updates as Partial<ServerRow>);
        updates = {};
    };
    if (!snapshot.provisioning(srv.name)) {
        const previousGame = monitoringStatus(srv.name, snapshot)?.game;
        const publishObservation = monitoringStatusWriter(srv.name, snapshot);
        let gameObservation: Awaited<ReturnType<typeof rconForMonitoring>> | undefined;
        const settling = snapshot.settling(srv.name);
        const epoch = snapshot.get(srv.name)?.submission;
        let identity: string | undefined;
        try {
            identity = monitoringIdentity(srv.name, snapshot);
        } catch {
            /* Offline hosts still need an unavailable sample. */
        }
        const unchanged = () => {
            const current = snapshot.get(srv.name)?.server;
            if (identity)
                try {
                    if (identity !== monitoringIdentity(srv.name, snapshot)) return false;
                } catch {
                    return false;
                }
            return (
                epoch === snapshot.get(srv.name)?.submission &&
                !snapshot.provisioning(srv.name) &&
                current?.created_at === srv.created_at &&
                current.desired_state === srv.desired_state &&
                current.native_instance_id === srv.native_instance_id &&
                current.instance_generation === srv.instance_generation
            );
        };
        const inspectionStartedAt = new Date().toISOString();
        const inspected = settling
            ? {
                  result: { ok: false as const, state: 'restarting', reason: 'Restart requested' },
                  observedAt: inspectionStartedAt
              }
            : await inspectForMonitoring(srv.name, false, snapshot).catch((err) =>
                  err instanceof MonitoringChangedError
                      ? null
                      : {
                            result: {
                                ok: false as const,
                                state: 'unknown',
                                reason: (err as Error).message
                            },
                            observedAt: inspectionStartedAt
                        }
              );
        snapshot.refresh(srv.name);
        if (!inspected) return null;
        const container = inspected.result;
        // A lifecycle request may be submitted while an inspection is in flight.
        if (
            unchanged() &&
            (snapshot.get(srv.name)?.server?.container_checked_at ?? '') <= inspected.observedAt
        ) {
            updates.container_state = container.ok
                ? 'running'
                : (container.state?.toLowerCase() ?? 'unknown');
            updates.container_checked_at = inspected.observedAt;
            // An unavailable inspection is not evidence that the game stopped.
            // Probe it just as a manual RCON check would; explicit lifecycle or
            // unhealthy states still take precedence over game replies.
            const unavailable = updates.container_state === 'unknown';
            if (!container.ok && !unavailable) {
                updates.last_error = null;
                updates.rcon_failures = 0;
                updates.last_reply_snippet = null;
            } else if (unavailable || !snapshot.canSkipRcon(srv.name)) {
                // Occupied telemetry rosters still need RCON for player pings.
                flush();
                const probeStartedAt = new Date().toISOString();
                const probe = await (
                    probeStatus ? probeStatus() : rconForMonitoring(srv.name, false, snapshot)
                ).catch((err) => {
                    snapshot.refresh(srv.name);
                    if (err instanceof MonitoringSkippedError) {
                        // The container was checked, but game evidence retains its original age.
                        if (unchanged()) publishObservation(inspected, previousGame);
                        return null;
                    }
                    if (err instanceof MonitoringChangedError) return null;
                    return {
                        result: {
                            ok: false as const,
                            endpoint: null,
                            reason: 'send-error' as RconFailure,
                            detail: (err as Error).message,
                            rawReply: undefined
                        },
                        observedAt: probeStartedAt
                    };
                });
                snapshot.refresh(srv.name);
                if (!probe) return null;
                const result = probe.result;
                const previous = snapshot.get(srv.name)?.server;
                if (
                    unchanged() &&
                    previous &&
                    (previous.last_checked_at ?? '') <= probe.observedAt
                ) {
                    gameObservation = probe;
                    const now = probe.observedAt;
                    updates.last_checked_at = now;
                    updates.last_reply_snippet = failureReplySnippet(result);
                    if (result.ok) {
                        capacity = result.status.capacity ?? null;
                        Object.assign(updates, successfulStatusFields(result, now));
                    } else {
                        // Only distinct accepted checks in the current lifecycle count.
                        // Skipped/superseded checks return above without changing the streak.
                        const consecutive =
                            previous.last_error &&
                            (!epoch || (previous.last_checked_at ?? '') > epoch);
                        const failures = consecutive ? previous.rcon_failures || 1 : 0;
                        updates.rcon_failures =
                            consecutive && previous.last_checked_at === now
                                ? failures
                                : Math.min(2, failures + 1);
                        updates.last_error = encodeError(result.reason, result.detail);
                    }
                }
            }
            if (!settling && unchanged()) publishObservation(inspected, gameObservation);
        }
    } else {
        updates.last_error = null;
        updates.rcon_failures = 0;
        updates.last_reply_snippet = null;
    }
    const current = { ...(snapshot.get(srv.name)?.server ?? srv), ...updates } as ServerRow;
    const status = snapshot.observe(current);
    Object.assign(updates, activityFields(current, status));
    flush();
    return capacity;
}

export function statusEntry(
    srv: ServerRow,
    observe = serverObservation,
    capacity?: PlayerCapacityObservation
): ServerStatusEntry {
    const final = observe(srv);
    return {
        server: srv.name,
        players: final.last_players,
        capacity: capacity?.capacity ?? null,
        capacityObservedAt: capacity?.capacity_observed_at ?? null,
        map: final.last_map ?? srv.map,
        gametype: final.last_gametype,
        roster: final.roster,
        health: final.health,
        lastSeenAt: final.last_seen_at,
        lastError: final.last_error,
        endpoint: srv.game_port
            ? { host: srv.rcon_host ?? config.rcon.host, port: srv.game_port }
            : null,
        reason: final.last_error ? (final.last_error.split(':')[0] as RconFailure) : null,
        activeSince: final.active_since,
        inactiveSince: final.inactive_since,
        gameStartedAt: final.game_started_at,
        gameEndedAt: final.game_ended_at,
        containerState: final.container_state,
        containerCheckedAt: final.container_checked_at,
        telemetryAfter: final.telemetry_after,
        rconStatus: final.rcon_status,
        telemetry: final.telemetry
    };
}

export function startWorker(): void {
    if (started) return;
    started = true;

    void tick();
    reconcileRequesters();
    void fanoutStatus().catch((err) => {
        console.error('[worker] status fanout error:', err);
    });
    tickTimer = setInterval(() => {
        void tick();
        reconcileRequesters();
    }, TICK_MS);
    statusTimer = setInterval(() => {
        void fanoutStatus().catch((err) => {
            console.error('[worker] status fanout error:', err);
        });
    }, STATUS_REFRESH_MS);

    console.log(
        '[worker] started — sweep every',
        TICK_MS,
        'ms, status every',
        STATUS_REFRESH_MS,
        'ms'
    );
}

export function stopWorker(): void {
    if (tickTimer) clearInterval(tickTimer);
    if (statusTimer) clearInterval(statusTimer);
    tickTimer = null;
    statusTimer = null;
    started = false;
}

/**
 * External trigger — called from `/api/push` and beta endpoints so a sweep
 * fires immediately after a state change instead of waiting for the next tick.
 */
export async function kick(): Promise<void> {
    reconcileRequesters();
    await tick(true);
}

/** Agent receipts need observation promptly, without launching another fleet scan. */
export async function kickProvisioning(): Promise<void> {
    await tick(true);
}