import { createHash } from 'node:crypto';
import { db } from '../db/index.js';
import {
    enrichGameChatTimeline,
    type ChatMatch,
    type ChatRoundReport,
    type ContextualChatEntry
} from '../../game-chat.js';
import { parseGameEvent } from '../../telemetry.js';
import { telemetryRetention, telemetryCutoff } from './telemetry-requests.js';
import { getServer, listServers, type ServerRow } from './servers.js';
import {
    gameLogKey,
    resolveLogPath,
    readLogChunk,
    type LogCursor,
    type TailResult
} from './logs.js';
import { listHosts } from '../hosts/index.js';
import { observationReader } from './status.js';

/** Historical context is scoped to this registration, including round-one matches. */
export function serverChatTimeline(server: ServerRow, tail: TailResult): ContextualChatEntry[] {
    const unscoped = enrichGameChatTimeline(tail.lines, {
        mtime: tail.mtime,
        matches: [],
        reports: []
    });
    const times = unscoped.flatMap((entry) =>
        entry.context.realTime ? [entry.context.realTime] : []
    );
    if (!times.length) return unscoped;
    const since = times.reduce((a, b) => (a < b ? a : b)),
        until = times.reduce((a, b) => (a > b ? a : b));
    const handle = db();
    const matches = handle
        .prepare(
            `SELECT id, started_at AS startedAt, ended_at AS endedAt,
        last_received_at AS lastReceivedAt, status, start_observed AS startObserved
        FROM leaderboard_matches WHERE server = ? AND server_deleted_at IS NULL
            AND started_at <= ? AND (COALESCE(ended_at,last_received_at) >= ? OR status='active')
        ORDER BY id DESC LIMIT 2000`
        )
        .all(server.name, until, since) as Array<
        Omit<ChatMatch, 'startObserved'> & { startObserved: number }
    >;

    // Diagnostic history survives deletion. Its receipt time must belong to the
    // current registration so a reused server name cannot inherit old rounds.
    const cutoff = telemetryCutoff();
    const roundSince = new Date(Date.parse(since) - 120_000).toISOString();
    const matchSince = matches.reduce(
        (earliest, match) => (match.startedAt < earliest ? match.startedAt : earliest),
        since
    );
    const reports = handle
        .prepare(
            `SELECT received_at AS receivedAt, event, round,
        player_count AS playerCount FROM telemetry_requests
        WHERE server = ? AND status = 200 AND received_at >= ?
            AND julianday(received_at) >= julianday(?)
            AND (received_at >= ? OR (event='game_start' AND received_at >= ?))
            AND received_at <= ?
            AND event IN ('game_start', 'round_change', 'heartbeat', 'game_end')
            AND player_count >= 0 AND (event = 'game_start' OR round >= 0)
        ORDER BY id DESC LIMIT ?`
        )
        .all(
            server.name,
            cutoff,
            server.created_at,
            roundSince,
            matchSince,
            new Date(Date.parse(until) + 120_000).toISOString(),
            telemetryRetention().maxPerServer || 10000
        ) as ChatRoundReport[];

    // Saving gameplay does not depend on diagnostic logging succeeding. Keep the
    // surviving accepted snapshot useful when its request log was not recorded.
    const latest = handle
        .prepare(
            `SELECT received_at AS receivedAt, payload FROM server_telemetry
        WHERE server = ? AND received_at >= ? AND julianday(received_at) >= julianday(?)`
        )
        .get(server.name, cutoff, server.created_at) as
        { receivedAt: string; payload: string | null } | undefined;
    if (latest?.payload) {
        try {
            const event = parseGameEvent(JSON.parse(latest.payload));
            const round = event.event === 'game_start' ? null : event.round;
            if (
                event.server === server.name &&
                !reports.some(
                    (report) =>
                        report.receivedAt === latest.receivedAt &&
                        report.event === event.event &&
                        report.round === round &&
                        report.playerCount === event.players.length
                )
            ) {
                reports.push({
                    receivedAt: latest.receivedAt,
                    event: event.event,
                    round,
                    playerCount: event.players.length
                });
            }
        } catch {
            /* Legacy malformed snapshots cannot establish chat context. */
        }
    }

    return enrichGameChatTimeline(tail.lines, {
        mtime: tail.mtime,
        matches: matches.map((match) => ({
            ...match,
            startObserved: Boolean(match.startObserved)
        })),
        reports
    });
}

export function matchChatTimeline(matchId: number): ContextualChatEntry[] {
    return (
        db()
            .prepare(
                'SELECT message_key,occurrence,entry FROM match_chat WHERE match_id = ? ORDER BY rowid'
            )
            .all(matchId) as { message_key: string; occurrence: number; entry: string }[]
    ).map((row) => ({
        ...JSON.parse(row.entry),
        matchId,
        messageId: `${matchId}:${row.message_key}:${row.occurrence}`
    }));
}

/** Save before filtering the live view so completed matches retain their messages. */
function archiveChat(entries: ContextualChatEntry[]): void {
    const handle = db();
    const insert = handle.prepare(`INSERT INTO match_chat (match_id, message_key, occurrence, entry)
        VALUES (?, ?, ?, ?) ON CONFLICT DO NOTHING`);
    const occurrences = new Map<string, number>();
    handle.transaction(() => {
        for (const entry of entries) {
            if (entry.type !== 'message' || entry.matchId === undefined) continue;
            // Engine time is stable across reads; estimated wall time is not.
            // Count identical messages separately, including repeats in one second.
            const key = createHash('sha256').update(JSON.stringify(entry.message)).digest('hex');
            const scope = `${entry.matchId}:${key}`;
            const occurrence = (occurrences.get(scope) ?? 0) + 1;
            occurrences.set(scope, occurrence);
            insert.run(entry.matchId, key, occurrence, JSON.stringify(entry));
        }
    })();
}

export function currentMatchChat(
    server: ServerRow,
    tail: TailResult | null
): ContextualChatEntry[] {
    const entries = tail ? serverChatTimeline(server, tail) : [];
    archiveChat(entries);
    return currentChatEntries(server, entries);
}

function currentChatEntries(
    server: ServerRow,
    entries: ContextualChatEntry[]
): ContextualChatEntry[] {
    const match = db()
        .prepare(
            `SELECT id, started_at AS startedAt FROM leaderboard_matches
        WHERE server = ? AND server_deleted_at IS NULL ORDER BY id DESC LIMIT 1`
        )
        .get(server.name) as { id: number; startedAt: string } | undefined;
    let startIndex = entries.length - 1;
    while (startIndex >= 0) {
        const entry = entries[startIndex];
        if (entry.type === 'boundary' && entry.event !== 'end') break;
        startIndex--;
    }
    const current = entries.slice(Math.max(0, startIndex));
    if (!match) {
        // Servers without gameplay telemetry still have log-defined sessions.
        return current.filter(
            (entry) =>
                entry.context.realTime !== null &&
                Date.parse(entry.context.realTime) >= Date.parse(server.created_at)
        );
    }
    const start = entries[startIndex];
    // A log restart can arrive before game_start telemetry. Clear immediately
    // instead of falling back to messages from the previous recorded match.
    if (
        start &&
        !current.some((entry) => entry.matchId === match.id) &&
        (!start.context.realTime ||
            Date.parse(start.context.realTime) > Date.parse(match.startedAt) + 1000)
    )
        return [];
    return matchChatTimeline(match.id);
}

const pending = new Map<string, Promise<{ entries: ContextualChatEntry[]; missing: boolean }>>();
interface ChatReadState {
    key: string;
    cursor?: LogCursor;
    lines: string[];
    partial: Buffer;
    skipLine: boolean;
    signature?: string;
    entries: ContextualChatEntry[];
    contextVersion?: string;
    wasOccupied: boolean;
}
const readers = new Map<string, ChatReadState>();
const failures = new Map<string, number>();
const CHAT_READ_MIN_AGE_MS = 15_000;
type LastChatRead = { completedAt: number } & ({ missing: boolean } | { error: unknown });
const lastReads = new Map<string, LastChatRead>();

function chatReadKey(server: ServerRow): string {
    return `${server.name}:${server.created_at}:${server.native_instance_id}:${server.instance_generation}`;
}

/** Live viewers reuse the worker's reader state instead of adding a read per tab. */
export async function readServerChat(
    server: ServerRow
): Promise<{ entries: ContextualChatEntry[]; missing: boolean }> {
    const key = chatReadKey(server);
    const existing = pending.get(key);
    if (existing) return existing;
    const last = lastReads.get(key);
    if (!last || Date.now() - last.completedAt >= CHAT_READ_MIN_AGE_MS)
        return collectServerChat(server);
    if ('error' in last) throw last.error;
    const state = readers.get(server.name);
    return {
        // Telemetry may have started a new match since the last log read.
        entries: currentChatEntries(server, state?.key === key ? state.entries : []),
        missing: last.missing
    };
}

/** Shared by the live API and the worker; collection does not require a viewer. */
export function collectServerChat(
    server: ServerRow
): Promise<{ entries: ContextualChatEntry[]; missing: boolean }> {
    const key = chatReadKey(server);
    const existing = pending.get(key);
    if (existing) return existing;
    const work = (async () => {
        const path = resolveLogPath(server.name, gameLogKey(server.name));
        if (!path) throw new Error('Invalid server log path');
        let saved = readers.get(server.name);
        if (saved?.key !== key) saved = undefined;
        // Publish the new cursor only after its messages have been processed.
        const state: ChatReadState = saved
            ? { ...saved }
            : {
                  key,
                  lines: [],
                  partial: Buffer.alloc(0),
                  skipLine: false,
                  entries: [],
                  wasOccupied: false
              };
        const tail = await readLogChunk(path, state.cursor);
        const current = getServer(server.name);
        // A remote read may finish after destruction or replacement.
        if (
            !current ||
            current.destroyed_at ||
            current.created_at !== server.created_at ||
            current.native_instance_id !== server.native_instance_id ||
            current.instance_generation !== server.instance_generation
        ) {
            return { entries: [], missing: true };
        }
        if (!tail) {
            readers.delete(server.name);
            return { entries: currentChatEntries(current, []), missing: true };
        }
        let changed = false;
        if ('cursor' in tail) {
            if (tail.reset) {
                state.lines = [];
                state.partial = Buffer.alloc(0);
                state.skipLine = tail.truncated;
                state.entries = [];
            }
            state.cursor = tail.cursor;
            const bytes = Buffer.concat([state.partial, Buffer.from(tail.contentBase64, 'base64')]);
            const end = bytes.lastIndexOf(10) + 1;
            if (end) {
                const lines = bytes
                    .subarray(0, end)
                    .toString('utf8')
                    .split('\n')
                    .slice(0, -1)
                    .map((line) => line.replace(/\r$/, ''));
                if (state.skipLine) lines.shift();
                state.skipLine = false;
                state.lines = [...state.lines, ...lines].slice(-2000);
                let length = 0,
                    start = state.lines.length;
                while (
                    start > 0 &&
                    (length += Buffer.byteLength(state.lines[start - 1]) + 1) <= 256 * 1024
                )
                    start--;
                state.lines = state.lines.slice(start);
                changed = true;
            }
            state.partial = Buffer.from(bytes.subarray(end));
            if (state.partial.length > 256 * 1024) {
                state.partial = Buffer.alloc(0);
                state.skipLine = true;
            }
            // File mtime anchors the final line only after the backlog has been read.
            if (tail.cursor.offset < tail.size || state.partial.length) {
                state.signature = undefined;
                readers.set(server.name, state);
                return { entries: currentChatEntries(current, state.entries), missing: false };
            }
        } else {
            // Older agents remain readable during a rolling update.
            changed = state.signature !== `${tail.size}:${tail.mtime}`;
            if (changed) state.lines = tail.lines;
        }
        const signature = `${tail.size}:${tail.mtime}`;
        const needsContext = state.entries.some(
            (entry) => entry.type === 'message' && entry.matchId === undefined
        );
        const contextVersion = needsContext
            ? (
                  db()
                      .prepare('SELECT received_at FROM server_telemetry WHERE server=?')
                      .get(server.name) as { received_at: string } | undefined
              )?.received_at
            : undefined;
        if (
            changed ||
            state.signature === undefined ||
            (needsContext && contextVersion !== state.contextVersion)
        ) {
            state.entries = serverChatTimeline(current, { ...tail, lines: state.lines });
            archiveChat(state.entries);
            state.contextVersion = contextVersion;
        }
        state.signature = signature;
        readers.set(server.name, state);
        return { entries: currentChatEntries(current, state.entries), missing: false };
    })()
        .then(
            (result) => {
                lastReads.set(key, { completedAt: Date.now(), missing: result.missing });
                return result;
            },
            (error: unknown) => {
                // Failed reads also observe the cooldown, without advancing the cursor.
                lastReads.set(key, { completedAt: Date.now(), error });
                throw error;
            }
        )
        .finally(() => pending.delete(key));
    pending.set(key, work);
    return work;
}

export async function collectFleetChat(): Promise<void> {
    const hosts = new Map(listHosts().map((host) => [host.id, host]));
    const observe = observationReader();
    const live = listServers().filter((server) => !server.destroyed_at);
    const names = new Set(live.map((server) => server.name));
    const keys = new Set(live.map(chatReadKey));
    for (const name of readers.keys()) if (!names.has(name)) readers.delete(name);
    for (const name of failures.keys()) if (!names.has(name)) failures.delete(name);
    for (const key of lastReads.keys()) if (!keys.has(key)) lastReads.delete(key);
    const servers = live.filter((server) => {
        if (server.host_id && !hosts.get(server.host_id)?.connected) return false;
        if (server.container_state && !['running', 'unknown'].includes(server.container_state))
            return false;
        const players = observe(server).last_players ?? server.last_players;
        // One final read after players leave retains messages written at match end.
        return players !== 0 || readers.get(server.name)?.wasOccupied;
    });
    // Bound agent traffic and isolate failures to their own server.
    async function collect() {
        for (let server = servers.shift(); server; server = servers.shift()) {
            try {
                await collectServerChat(server);
                const state = readers.get(server.name);
                if (state)
                    state.wasOccupied =
                        (observe(server).last_players ?? server.last_players ?? 0) > 0;
                failures.delete(server.name);
            } catch (error) {
                if (Date.now() - (failures.get(server.name) ?? -Infinity) >= 5 * 60_000) {
                    console.error(`[chat] Could not collect chat for ${server.name}`, error);
                    failures.set(server.name, Date.now());
                }
            }
        }
    }
    await Promise.all(Array.from({ length: Math.min(4, servers.length) }, collect));
}