import { sort, throttleFunction } from "socket-function/src/misc";
import { runInSerial } from "socket-function/src/batching";
import { getTimeUnique } from "socket-function/src/bits";
import { lazy } from "socket-function/src/caching";
import { isNode } from "typesafecss";
import type { FileStorage } from "../FileFolderAPI";
import {
    BaseBulkDatabaseReader,
    BulkHeaderInfo,
    buildFileBuffer,
    loadBulkHeader,
    TARGET_FILE_BYTES,
} from "./BulkDatabaseFormat";
import { runPlannedMerge } from "./BulkDatabaseMerge";
import { blockCache, encodeCompressedBlocks } from "./blockCache";
import { formatNumber, formatTime } from "socket-function/src/formatting/format";
import { blue, magenta } from "socket-function/src/formatting/logColors";
import { STREAM_EXTENSION, frameDeletes, frameRows, streamReaderFromEntries } from "./streamLog";
import { broadcast as syncBroadcast, broadcastSeal as syncBroadcastSeal, connect as syncConnect, isSyncSupported, queryLiveWriters, registerWriterId, RemoteWrite } from "./syncClient";
import { DELETED } from "./WriteOverlay";
import { MergeLockInfo, peekMergeFileLock, peekMergeLock, releaseMergeFileLock, releaseMergeLock, startMergeFileLockHeartbeat, tryAcquireMergeFileLock, tryAcquireMergeLock } from "./mergeLock";
import { markerExclusions, processDeleteMarkers, readDeleteMarkers, writeDeleteMarker } from "./mergeMarkers";
import {
    BulkFileInfo,
    LoadedIndex,
    loadFileReader,
    loadStreamEntries,
    makeRawGetRange,
    MissingFileError,
    orderStreamEntries,
    StreamFileInfo,
    SubReaderCaches,
} from "./LoadedIndex";
import { BulkDatabaseReader, nullJoin } from "./BulkDatabaseReader";

export const BULK_ROOT_FOLDER = "bulkDatabases2";
const FILE_EXTENSION = ".bulk";
const ROLLOVER_ROWS = 5000;
const ROLLOVER_BYTES = 5 * 1024 * 1024;
const MEMORY_WATCHDOG_INTERVAL_MS = 60 * 1000;
const STALE_DELETE_MS = 24 * 60 * 60 * 1000;
const MAX_INDEX_RELOAD_ATTEMPTS = 3;
// A bulk file under this is "loose" - still worth rolling up into a bigger one. Half the target file size, because that is exactly where combining stops paying: merging two files that are each over half the target just splits back into two files again, so the file count (and with it the per-file key list every read holds in memory) does not drop. Under half, any two inputs fit in one output, so phase 2 always makes the count go down.
//
// NOT the target size itself. runPlannedMerge cuts a chunk BEFORE adding the key that would exceed the target, so every file it writes is under TARGET_FILE_BYTES - testing against the target would classify the entire collection as loose and re-merge all of it, forever.
const LOOSE_BULK_MAX_BYTES = TARGET_FILE_BYTES / 2;
const KEY_GROUP_BYTES = 800 * 1024 * 1024;
const DUP_THRESHOLD = 0.4;
// Whole-tier dedup short-circuit (start of phase 3): when the combined tier is over this size AND the overall key duplication fraction is over this threshold, fold every combined file in one merge instead of the per-key-group walk (which spaces merges 5 min apart — 16 h for 200 groups).
const DEDUP_TRIGGER_BYTES = 512 * 1024 * 1024;
const DEDUP_TRIGGER_FRACTION = 0.5;
const WRITE_FLUSH_FIRST_STEP_MS = 250;
// Skip logs are throttled per collection: the background write path retries merges about once a second, so an unthrottled log would repeat for the whole duration of another tab's merge.
const MERGE_SKIP_LOG_INTERVAL_MS = 30 * 1000;

export const bulkDatabase2Timing = {
    streamSealAgeMs: 10 * 60 * 60 * 1000,
    // Wait this long after the tab becomes visible before the first merge check, then check this often afterwards. Tab being hidden cancels the timers (no merges while in background). Browser-only; Node compactors (e.g. remoteFileServer) drive their own polling.
    visibleMergeIntervalMs: 5 * 60 * 1000,
    mergeSpacingMs: 5 * 60 * 1000,
    // Phase 2 rolls the loose bulk files up once they pass either of these. Bytes because read cost grows with them; count because each loose file holds its whole key list in memory no matter how few bytes it has, so a trickle of tiny folds has to be collapsed long before it reaches the byte trigger.
    looseBulkTriggerBytes: 1024 * 1024 * 1024,
    looseBulkTriggerFiles: 20,
    streamFoldTriggerBytes: 64 * 1024 * 1024,
    streamFileMaxBytes: 50 * 1024 * 1024,
    // How long to wait for peers to answer a stream-owner liveness probe. Runs inside a merge pass, which already waits 15s for the file lock, so a second here costs nothing.
    liveWriterProbeMs: 1000,
    streamFoldHardLimitBytes: 768 * 1024 * 1024,
    // 0 = flush every write (Node — append is real and cheap); browser ramps to 15s to avoid rewriting the whole stream file per write.
    writeFlushMaxDelayMs: isNode() ? 0 : 15 * 1000,
    fileSetPollIntervalMs: 30 * 60 * 1000,
    memoryFlushHeapBytes: 5 * 1024 * 1024 * 1024,
    memoryFlushMinCollectionBytes: 100 * 1024 * 1024,
    memoryFlushThrottleMs: 15 * 60 * 1000,
};

function fmtBytes(n: number): string {
    if (n < 1024) return n + "B";
    if (n < 1024 * 1024) return (n / 1024).toFixed(1) + "KB";
    if (n < 1024 * 1024 * 1024) return (n / 1024 / 1024).toFixed(1) + "MB";
    return (n / 1024 / 1024 / 1024).toFixed(2) + "GB";
}

// Reactivity seam (no mobx dependency in this file). The mobx subclass supplies a ReactiveDeps that maps signal strings to its own dependency tracking; non-reactive callers pass noopReactiveDeps.
export interface ReactiveDeps {
    observe(signal: string): void;
    invalidate(signal: string): void;
    batch(fn: () => void): void;
    // Optional — lets writes skip per-key invalidation for rows nothing is watching. Undefined = "assume watched".
    isObserved?(signal: string): boolean;
}

export const noopReactiveDeps: ReactiveDeps = {
    observe() { },
    invalidate() { },
    batch(fn) { fn(); },
    isObserved() { return false; },
};

export type StorageFactory = (path: string) => Promise<FileStorage>;

export type BulkDatabase2Config = {
    // See BulkDatabaseReader.cfg.maxTriggerThrottleMs.
    maxTriggerThrottleMs?: number;
};

export type MergeSkipReason =
    // This instance is already mid-merge.
    | "mergeInFlight"
    // Another same-origin tab holds the localStorage merge lock.
    | "tabLockHeld"
    // Another process holds the cross-process .merge-lock file (or won the settle race for it).
    | "fileLockHeld"
    // compact() only: the collection has no files on disk.
    | "nothingToMerge";

// What a compact()/tryMergeNow() call did. skipReason is set when the pass never ran; for the lock reasons, lockHolderId/lockExpiresInMs report who holds the lock and how long until it goes stale (so a scheduler knows when retrying could succeed).
export type MergeAttemptResult = {
    merged: boolean;
    skipReason?: MergeSkipReason;
    lockHolderId?: string;
    lockExpiresInMs?: number;
};

let networkCompactionEnabled = false;

let fileNameCounter = 0;
// Per-process ID so two writers picking the same timestamp + counter never collide on a name. Also stamped into our stream file names and answered over the sync channel, so a peer can tell whether the writer of a given stream file is still running.
const writerId = Math.random().toString(36).slice(2, 10);
registerWriterId(writerId);
// Reserved owner for the tombstone-carry file a merge emits. Nothing ever appends to it, but it must NOT read as abandoned: folding it would emit another carry file, which would read as abandoned in turn, and the pass would fold forever.
const MERGE_OUTPUT_OWNER = "merged";
function nextCounter(): number { return ++fileNameCounter; }

let lastFileTime = 0;
// Strictly-increasing integer so newest-first ordering is unambiguous within a millisecond.
function nextFileTime(): number {
    lastFileTime = Math.max(Date.now(), lastFileTime + 1);
    return lastFileTime;
}

function newFileName(timestamp: number): string {
    return `0_${timestamp}_${writerId}_${nextCounter()}${FILE_EXTENSION}`;
}

// Accept old 3-part (stream_timestamp_random) and new 4-part (stream_timestamp_ownerId_counter) shapes. A 3-part name carries no owner, so it can never be matched to a live writer — which is what we want: no build that produces those names is still appending to them.
function parseStreamFileName(fileName: string): StreamFileInfo | undefined {
    if (!fileName.endsWith(STREAM_EXTENSION)) return undefined;
    const parts = fileName.slice(0, -STREAM_EXTENSION.length).split("_");
    if (parts[0] !== "stream") return undefined;
    if (parts.length !== 3 && parts.length !== 4) return undefined;
    const timestamp = parseInt(parts[1], 10);
    if (!Number.isFinite(timestamp)) return undefined;
    return { fileName, timestamp, ownerId: parts.length === 4 && parts[2] || undefined };
}

function newStreamFileName(ownerId: string): string {
    return `stream_${Date.now()}_${ownerId}_${nextCounter()}${STREAM_EXTENSION}`;
}

// Accept old 3-part (level_timestamp_counter) and new 4-part (level_timestamp_writerId_counter) shapes.
function parseFileName(fileName: string): BulkFileInfo | undefined {
    if (!fileName.endsWith(FILE_EXTENSION)) return undefined;
    const parts = fileName.slice(0, -FILE_EXTENSION.length).split("_");
    if (parts.length < 3) return undefined;
    const level = parseInt(parts[0], 10);
    const timestamp = parseInt(parts[1], 10);
    if (!Number.isFinite(level) || !Number.isFinite(timestamp)) return undefined;
    return { fileName, level, timestamp };
}

/** One threshold a compaction step is measured against. `value` is where the collection stands now and `threshold` is what sets the step off, so `fraction` (value/threshold) reads as how close it is - 1 or more means met. Deliberately not clamped, so an overdue step reads as how far past due it is. */
export type CompactionTrigger = {
    name: string;
    value: number;
    threshold: number;
    fraction: number;
    met: boolean;
    /** How to render value/threshold. "fraction" values are 0..1. */
    unit: "bytes" | "count" | "fraction";
};

/** streamHardLimit and streamFold are phase 1 (stream -> bulk), looseCombine is phase 2 (loose bulk -> combined bulk), dedupAll and dedupKeyGroup are phase 3. */
export type CompactionStepKind = "streamHardLimit" | "streamFold" | "looseCombine" | "dedupAll" | "dedupKeyGroup";

export type CompactionStep = {
    phase: 1 | 2 | 3;
    kind: CompactionStepKind;
    /** Whether this step runs on the next pass. Authoritative: on top of `requires` it accounts for inputs the step needs beyond its thresholds (e.g. two files to combine), so it can be false even with every trigger met. */
    ready: boolean;
    /** Whether every trigger has to be met for this step, or just one of them. */
    requires: "any" | "all";
    triggers: CompactionTrigger[];
    /** When this step's merge starts, given merges are spaced mergeSpacingMs apart. Only set when ready. */
    startTime?: number;
    /** The files this step consumes, as of when the plan was made. */
    bulkFiles: BulkFileInfo[];
    streamFiles: StreamFileInfo[];
    /** Total size of those inputs. */
    bytes: number;
    /** dedupKeyGroup only - the key range the step rewrites. */
    keyRange?: { lo: string; hi: string };
};

/** Every compaction the current file set calls for, in the order a merge pass runs them, plus how close each not-yet-ready one is to its thresholds. */
export type CompactionPlan = {
    collection: string;
    /** When the plan was computed; every startTime is measured from here. */
    time: number;
    steps: CompactionStep[];
};

function makeTrigger(config: { name: string; value: number; threshold: number; unit: CompactionTrigger["unit"] }): CompactionTrigger {
    return { ...config, fraction: config.value / config.threshold, met: config.value >= config.threshold };
}

function fmtTriggerValue(value: number, unit: CompactionTrigger["unit"]): string {
    if (unit === "bytes") return fmtBytes(value);
    if (unit === "fraction") return `${Math.round(value * 100)}%`;
    return formatNumber(value);
}

export class BulkDatabaseBase<T extends { key: string }> {
    constructor(
        public readonly name: string,
        protected deps: ReactiveDeps,
        private storageFactory: StorageFactory,
        private config: BulkDatabase2Config = {},
    ) { }

    // The reader (and the background machinery that rides along with it) must NOT be set up just because the collection was constructed. Many collections are constructed and never touched, and in Node a merge tick would poke the storage factory (e.g. indexedDB) and throw. We build it lazily on first access to `this.reader`: every read, write, and merge goes through the reader, while pure construction never touches it.
    private _reader: BulkDatabaseReader<T> | undefined;
    private get reader(): BulkDatabaseReader<T> {
        if (this._reader) return this._reader;
        const reader = new BulkDatabaseReader<T>({
            name: this.name,
            deps: this.deps,
            maxTriggerThrottleMs: this.config.maxTriggerThrottleMs,
        });
        reader.setEnsureIndex(() => this.ensureIndex());
        this._reader = reader;
        this.activate();
        return reader;
    }

    private activated = false;
    private activate(): void {
        if (this.activated) return;
        this.activated = true;
        if (typeof window !== "undefined") {
            try {
                window.addEventListener("pagehide", () => void this.flushPending());
                if (typeof document !== "undefined") {
                    document.addEventListener("visibilitychange", () => {
                        if (document.visibilityState === "hidden") void this.flushPending();
                    });
                }
            } catch { /* not in a DOM context */ }
        }
        this.fileSetPollTimer = setInterval(() => void this.pollFileSet(), bulkDatabase2Timing.fileSetPollIntervalMs);
        (this.fileSetPollTimer as { unref?: () => void }).unref?.();
        this.setupVisibilityMergeCheck();
        BulkDatabaseBase.liveInstances.add(this);
        BulkDatabaseBase.startMemoryWatchdog();
    }

    // Every `visibleMergeIntervalMs` of being-visible time, run a merge check. First check fires `visibleMergeIntervalMs` after the tab becomes visible (not on initial load — gives the user a moment of session commitment before we start churning the FS), then every interval after. Tab hiding cancels the timers; visible-again restarts. Node has no document — there we consider ourselves always-visible and just install the interval directly.
    private setupVisibilityMergeCheck(): void {
        let firstTimer: ReturnType<typeof setTimeout> | undefined;
        let intervalTimer: ReturnType<typeof setInterval> | undefined;
        const stop = () => {
            if (firstTimer) { clearTimeout(firstTimer); firstTimer = undefined; }
            if (intervalTimer) { clearInterval(intervalTimer); intervalTimer = undefined; }
        };
        const start = () => {
            if (firstTimer || intervalTimer) return;
            firstTimer = setTimeout(() => {
                firstTimer = undefined;
                void this.maybeMerge();
                intervalTimer = setInterval(() => void this.maybeMerge(), bulkDatabase2Timing.visibleMergeIntervalMs);
                (intervalTimer as { unref?: () => void }).unref?.();
            }, bulkDatabase2Timing.visibleMergeIntervalMs);
            (firstTimer as { unref?: () => void }).unref?.();
        };
        if (typeof document === "undefined") { start(); return; }
        try {
            document.addEventListener("visibilitychange", () => {
                if (document.visibilityState === "visible") start();
                else stop();
            });
            if (document.visibilityState === "visible") start();
        } catch { /* not in a DOM context */ }
    }

    private subCaches: SubReaderCaches = { bulk: new Map(), stream: new Map() };

    private pendingAppends: Buffer[] = [];
    private flushTimer: ReturnType<typeof setTimeout> | undefined;
    private flushChain: Promise<void> = Promise.resolve();
    private currentFlushDelay = 0;
    private lastWriteTime = 0;

    private streamFileName: string | undefined;
    private currentStreamFileName: string | undefined;
    private currentStreamFileBytes = 0;
    // In-process re-entry guard: if a merge is running, additional triggers (visibility timer, writes, tryMergeNow calls) return immediately instead of queueing. Keeps the visibility-timer firings from stacking behind a long merge.
    private mergeInFlight = false;
    private lastMergeSkipLogMs = 0;
    // Running counter of stream-tier bytes on disk. Seeded from each LoadedIndex build, then incremented per flush so the fold-trigger checks current data without an extra directory listing.
    private streamBytesOnDisk = 0;

    private fileSetPollTimer: ReturnType<typeof setInterval> | undefined;
    private rebuildPromise: Promise<void> | undefined;
    private rebuildDirty = false;
    private rebuildOptions: { dropStaleFallback: boolean } = { dropStaleFallback: false };

    // ── memory-pressure watchdog (global, browser-only) ──
    private static liveInstances = new Set<BulkDatabaseBase<{ key: string }>>();
    private static memoryWatchdogStarted = false;
    private static lastMemoryFlushMs = 0;
    private static startMemoryWatchdog() {
        if (BulkDatabaseBase.memoryWatchdogStarted) return;
        BulkDatabaseBase.memoryWatchdogStarted = true;
        const usedHeap = (): number | undefined => {
            try { return (performance as unknown as { memory?: { usedJSHeapSize?: number } })?.memory?.usedJSHeapSize; }
            catch { return undefined; }
        };
        if (typeof performance === "undefined" || usedHeap() === undefined) return;
        const timer = setInterval(() => {
            const used = usedHeap();
            if (used !== undefined) BulkDatabaseBase.checkMemoryPressure(used);
        }, MEMORY_WATCHDOG_INTERVAL_MS);
        (timer as { unref?: () => void }).unref?.();
    }
    public static checkMemoryPressure(usedHeapBytes: number): void {
        if (usedHeapBytes < bulkDatabase2Timing.memoryFlushHeapBytes) return;
        const now = Date.now();
        if (now - BulkDatabaseBase.lastMemoryFlushMs < bulkDatabase2Timing.memoryFlushThrottleMs) return;
        BulkDatabaseBase.lastMemoryFlushMs = now;
        const flushed: string[] = [];
        for (const db of BulkDatabaseBase.liveInstances) {
            const bytes = db.reader.index?.totalBytes ?? 0;
            if (bytes > bulkDatabase2Timing.memoryFlushMinCollectionBytes) {
                flushed.push(`${db.name} (${fmtBytes(bytes)})`);
                db.reloadFromDisk();
            }
        }
        if (flushed.length) console.log(`[bulk2] heap ${fmtBytes(usedHeapBytes)} over ${fmtBytes(bulkDatabase2Timing.memoryFlushHeapBytes)} - flushed ${flushed.length} large collection(s): ${flushed.join(", ")}`);
    }

    public static clearCache() {
        blockCache.clear();
    }

    public static enableNetworkCompaction() {
        networkCompactionEnabled = true;
    }

    public storage = lazy(async () => this.storageFactory(`${BULK_ROOT_FOLDER}/${this.name}`));

    public async isRemote(): Promise<boolean> {
        return !!(await this.storage()).isRemote;
    }

    // Uncompacted stream bytes have to be read and decoded in full by every reader on every index build, so size alone decides — a fold is worth it even if those bytes are one enormous row.
    private streamNeedsFold(): boolean {
        return this.streamBytesOnDisk > bulkDatabase2Timing.streamFoldTriggerBytes;
    }

    // Stream files nobody can still be appending to: our own sealed files, merge-carry output, and files whose owner is gone (a closed tab's leftovers, or a legacy name with no owner stamp). A fold may delete these the moment it has consumed them, instead of waiting out streamSealAgeMs.
    //
    // Foreign owners are probed over the sync channel. In Node there is no channel, so we cannot know - every foreign owner is assumed alive and the streamSealAgeMs rule stands.
    //
    // assumeSealed is for planning: a merge pass broadcasts a seal before it starts, so by the time it merges our current file IS final. The planner passes isSyncSupported() to predict that; anything deciding a real deletion passes false and goes by streamFileName as it actually stands.
    private async findAbandonedStreams(streamFiles: StreamFileInfo[], assumeSealed: boolean): Promise<Set<string>> {
        const retirable = new Set<string>();
        const hasForeignOwner = streamFiles.some(f => f.ownerId && f.ownerId !== writerId && f.ownerId !== MERGE_OUTPUT_OWNER);
        const live = hasForeignOwner && await queryLiveWriters(this.name, bulkDatabase2Timing.liveWriterProbeMs) || undefined;
        for (const f of streamFiles) {
            if (f.ownerId === MERGE_OUTPUT_OWNER) { retirable.add(f.fileName); continue; }
            // streamFileName is the only "we will append here again" signal - once it moves on, getStreamFileName opens a fresh file and this one is final.
            if (f.ownerId === writerId) {
                if (assumeSealed || f.fileName !== this.streamFileName) retirable.add(f.fileName);
                continue;
            }
            if (!f.ownerId || live && !live.has(f.ownerId)) retirable.add(f.fileName);
        }
        return retirable;
    }

    private async automaticCompactionAllowed(): Promise<boolean> {
        if (networkCompactionEnabled) return true;
        if (isNode()) return true;
        return !(await this.storage()).isRemote;
    }

    public isKeyWatched(key: string): boolean {
        return this.reader.isKeyWatched(key);
    }

    // ── index lifecycle ──────────────────────────────────────────────────────────────────────────────
    private async ensureIndex(): Promise<LoadedIndex<T>> {
        if (this.reader.index) return this.reader.index;
        await this.triggerRebuild();
        const idx = this.reader.index;
        if (!idx) throw new Error(`${this.name}: index failed to build`);
        return idx;
    }

    // Coalescing rebuild loop: triggers during a build set rebuildDirty so the loop iterates once more when it finishes — N rapid triggers cause at most ONE extra rebuild after the current one ends.
    private triggerRebuild(opts: { dropStaleFallback?: boolean } = {}): Promise<void> {
        if (opts.dropStaleFallback) this.rebuildOptions.dropStaleFallback = true;
        if (this.rebuildPromise) {
            this.rebuildDirty = true;
            return this.rebuildPromise;
        }
        this.rebuildPromise = (async () => {
            try {
                do {
                    this.rebuildDirty = false;
                    await this.doOneRebuild();
                } while (this.rebuildDirty);
            } finally {
                this.rebuildPromise = undefined;
                this.rebuildOptions.dropStaleFallback = false;
            }
        })();
        return this.rebuildPromise;
    }

    private async doOneRebuild(): Promise<void> {
        const { bulkFiles, streamFiles } = await this.listFiles();
        const storage = await this.storage();
        const newIndex = await LoadedIndex.build<T>({
            name: this.name,
            storage,
            bulkFiles,
            streamFiles,
            subCaches: this.subCaches,
            onUnreadableFile: (f, msg) => this.handleUnreadableFile(f, msg),
        });
        const oldIndex = this.reader.index;
        this.reader.setIndex(newIndex, { dropStaleFallback: this.rebuildOptions.dropStaleFallback });
        this.streamBytesOnDisk = newIndex.streamBytesOnDisk;
        if (oldIndex) {
            for (const f of oldIndex.fileSet) {
                if (!newIndex.fileSet.has(f)) blockCache.evict(nullJoin(this.name, f));
            }
        }
    }

    // Drop everything in-memory, hard reset. Pending un-flushed writes survive (still in overlay).
    public reloadFromDisk(): void {
        this.subCaches.bulk.clear();
        this.subCaches.stream.clear();
        this.reader.index?.dropLoadedValues();
        void this.triggerRebuild({ dropStaleFallback: true });
    }

    // External-merge detection: if some other tab/process changed the file set under us, rebuild.
    private async pollFileSet(): Promise<void> {
        if (!this.reader.index) return;
        let current: Set<string>;
        try {
            const { bulkFiles, streamFiles } = await this.listFiles();
            current = new Set([...bulkFiles.map(f => f.fileName), ...streamFiles.map(f => f.fileName)]);
        } catch { return; }
        const prev = this.reader.index?.fileSet;
        if (!prev) return;
        const changed = current.size !== prev.size || [...current].some(n => !prev.has(n));
        if (changed) void this.triggerRebuild();
    }

    private async readWithRetry<R>(fn: () => Promise<R>): Promise<R> {
        await this.ensureIndex();
        for (let attempt = 0; ; attempt++) {
            const before = this.reader.index;
            try { return await fn(); }
            catch (e) {
                if (!(e instanceof MissingFileError) || attempt >= MAX_INDEX_RELOAD_ATTEMPTS) throw e;
                if (this.reader.index === before) await this.triggerRebuild();
            }
        }
    }

    // ── cross-tab sync ───────────────────────────────────────────────────────────────────────────────
    private syncSetup = lazy(async () => {
        if (!isSyncSupported()) return;
        await this.ensureIndex();
        const recent = await syncConnect(this.name, w => this.applyRemote(w), () => { this.streamFileName = undefined; });
        for (const w of recent) this.applyRemote(w);
    });

    private applyRemote(write: RemoteWrite) {
        if (write.time <= this.reader.localTime(write.key)) return;
        this.deps.batch(() => {
            if (write.deleted) this.reader.applyDelete(write.key, write.time);
            else this.reader.applyWrite(write.key, write.value as Record<string, unknown>, write.time);
        });
    }

    // ── writes ───────────────────────────────────────────────────────────────────────────────────────
    public async write(entry: T): Promise<void> {
        return this.writeBatch([entry]);
    }

    public async writeBatch(entries: T[]): Promise<void> {
        if (!entries.length) return;
        void this.syncSetup();
        const rows = entries as unknown as Record<string, unknown>[];
        const stamped = rows.map(row => ({ time: getTimeUnique(), row }));
        const framed = frameRows(stamped);

        // Big batches skip the stream and become a bulk file directly — streaming thousands of rows one frame at a time would just churn.
        if (entries.length >= ROLLOVER_ROWS || framed.length >= ROLLOVER_BYTES) {
            await this.writeBulkFile(rows);
            return;
        }

        this.deps.batch(() => {
            for (const { time, row } of stamped) this.reader.applyWrite(row.key as string, row, time);
        });
        for (const { time, row } of stamped) syncBroadcast(this.name, { key: row.key as string, time, value: row });
        await this.streamAppend(framed);
        void this.maybeMerge({ onlyIfStreamHeavy: true });
    }

    public async delete(key: string): Promise<void> {
        return this.deleteBatch([key]);
    }

    public async deleteBatch(keys: string[]): Promise<void> {
        if (!keys.length) return;
        void this.syncSetup();
        const stamped = keys.map(key => ({ time: getTimeUnique(), key }));
        this.deps.batch(() => {
            for (const { time, key } of stamped) this.reader.applyDelete(key, time);
        });
        for (const { time, key } of stamped) syncBroadcast(this.name, { key, time, deleted: true });
        await this.streamAppend(frameDeletes(stamped));
        void this.maybeMerge({ onlyIfStreamHeavy: true });
    }

    // Coalesce stream appends on a ramping per-collection schedule (the browser rewrites the whole file per append). The first write after a lull flushes immediately so a single edit-then-close is saved at once; sustained writes ramp toward writeFlushMaxDelayMs.
    private async streamAppend(framed: Buffer): Promise<void> {
        this.pendingAppends.push(framed);
        const max = bulkDatabase2Timing.writeFlushMaxDelayMs;
        const now = Date.now();
        if (max <= 0 || this.currentFlushDelay <= 0 || now - this.lastWriteTime > max) {
            this.lastWriteTime = now;
            this.currentFlushDelay = max > 0 ? Math.min(max, WRITE_FLUSH_FIRST_STEP_MS) : 0;
            await this.flushPending();
            return;
        }
        this.lastWriteTime = now;
        if (this.flushTimer === undefined) {
            this.flushTimer = setTimeout(() => { this.flushTimer = undefined; void this.flushPending(); }, this.currentFlushDelay);
        }
        this.currentFlushDelay = Math.min(max, this.currentFlushDelay * 2);
    }

    public async flush(): Promise<void> {
        await this.flushPending();
    }

    private async flushPending(): Promise<void> {
        if (this.flushTimer !== undefined) { clearTimeout(this.flushTimer); this.flushTimer = undefined; }
        this.flushChain = this.flushChain.then(() => this.doFlush()).catch(e => {
            console.warn(`${this.name}: stream flush failed, will retry: ${(e as Error).message}`);
        });
        return this.flushChain;
    }

    private async doFlush(): Promise<void> {
        if (!this.pendingAppends.length) return;
        const batch = this.pendingAppends.slice();
        const combined = Buffer.concat(batch);
        const storage = await this.storage();
        const fileName = this.getStreamFileName();
        if (fileName !== this.currentStreamFileName) {
            this.currentStreamFileName = fileName;
            this.currentStreamFileBytes = 0;
        }
        // On failure the throw leaves pendingAppends intact so a later flush retries.
        await storage.append(fileName, combined);
        // New entries appended during the await are after `batch` — removing the front is exactly the flushed set.
        this.pendingAppends.splice(0, batch.length);
        this.streamBytesOnDisk += combined.length;
        this.currentStreamFileBytes += combined.length;
        if (this.currentStreamFileBytes >= bulkDatabase2Timing.streamFileMaxBytes) {
            this.streamFileName = undefined;
            this.currentStreamFileName = undefined;
            this.currentStreamFileBytes = 0;
            void this.foldOwnStream(fileName);
        }
    }

    private getStreamFileName(): string {
        // Seal our current file once it ages past the seal threshold — no file is ever appended to past its seal age, which lets a consolidation safely fold it once aged.
        if (this.streamFileName) {
            const info = parseStreamFileName(this.streamFileName);
            if (info && Date.now() - info.timestamp >= bulkDatabase2Timing.streamSealAgeMs) this.streamFileName = undefined;
        }
        if (!this.streamFileName) {
            this.streamFileName = newStreamFileName(writerId);
        }
        return this.streamFileName;
    }

    private async foldOwnStream(fileName: string): Promise<void> {
        const info = parseStreamFileName(fileName);
        if (!info) return;
        try {
            await this.mergeFileSet([], [info], false, true);
        } catch (e) {
            console.warn(`${this.name}: folding own stream ${fileName} failed: ${(e as Error).message}`);
        }
    }

    public async update(entry: Partial<T> & { key: string }): Promise<void> {
        return this.updateBatch([entry]);
    }

    public async updateBatch(entries: (Partial<T> & { key: string })[]): Promise<void> {
        if (!entries.length) return;
        void this.syncSetup();
        const index = await this.ensureIndex();
        const present: T[] = [];
        for (const entry of entries) {
            const overlayEntry = this.reader.overlay.get(entry.key);
            const exists = overlayEntry ? overlayEntry.value !== DELETED : index.keys.has(entry.key);
            if (!exists) {
                console.warn(`${this.name}.update: key ${JSON.stringify(entry.key)} is not in the collection, ignoring`);
                continue;
            }
            present.push(entry as unknown as T);
        }
        if (present.length) await this.writeBatch(present);
    }

    // ── file listings ────────────────────────────────────────────────────────────────────────────────
    private async listFiles(): Promise<{ bulkFiles: BulkFileInfo[]; streamFiles: StreamFileInfo[] }> {
        const storage = await this.storage();
        const names = await storage.getKeys();
        // A consumed-merge input is hidden from reads the instant its deletion marker exists — that's what stops it from being re-read and re-merged. Physical deletion + marker cleanup runs on a throttle (not inline) so a read isn't slowed by it.
        const markers = await readDeleteMarkers(storage, names);
        const excluded = markerExclusions(markers);
        if (markers.length) void this.processMarkers();
        const bulkFiles: BulkFileInfo[] = [];
        const streamFiles: StreamFileInfo[] = [];
        for (const n of names) {
            if (excluded.has(n)) continue;
            if (n.endsWith(FILE_EXTENSION)) { const p = parseFileName(n); if (p) bulkFiles.push(p); }
            else if (n.endsWith(STREAM_EXTENSION)) { const p = parseStreamFileName(n); if (p) streamFiles.push(p); }
        }
        bulkFiles.sort((a, b) => {
            if (a.timestamp !== b.timestamp) return b.timestamp - a.timestamp;
            return a.fileName < b.fileName && 1 || a.fileName > b.fileName && -1 || 0;
        });
        sort(streamFiles, f => f.timestamp);
        return { bulkFiles, streamFiles };
    }

    // Throttled marker housekeeping: delete inputs whose replacement outputs have landed (or whose marker has aged out) and retire fulfilled markers. Triggered from listFiles, so it's gated to avoid a delete storm under heavy reads.
    private processMarkers = throttleFunction(30 * 1000, async () => {
        const storage = await this.storage();
        const names = await storage.getKeys();
        const markers = await readDeleteMarkers(storage, names);
        if (markers.length) await processDeleteMarkers(this.name, storage, markers, names);
    });

    private async writeBulkFile(rows: Record<string, unknown>[]): Promise<void> {
        const storage = await this.storage();
        const timestamp = nextFileTime();
        const now = Date.now();
        const times = rows.map(() => now);
        for (const built of buildFileBuffer(rows, times)) {
            const name = newFileName(timestamp);
            await storage.set(name, encodeCompressedBlocks(built.buffer));
        }
        await this.triggerRebuild();
        void this.maybeMerge();
    }

    // ── merge policy ───────────────────────────────────────────────────────────────────────────────── Called both periodically (visibility timer) and on writes (with the streamNeedsFold gate so a small write doesn't drag the merge through). The mergeInFlight check in tryMergeNow skips any call that arrives while a previous one is still running — so 5-min timer ticks don't pile up.
    private async maybeMerge(opts: { onlyIfStreamHeavy?: boolean } = {}): Promise<void> {
        if (!await this.automaticCompactionAllowed()) return;
        if (opts.onlyIfStreamHeavy && !this.streamNeedsFold()) return;
        try {
            await this.tryMergeThrottled();
        } catch (e) {
            console.warn(`${this.name}: background merge failed: ${(e as Error).message}`);
        }
    }

    // Build the "we didn't merge" result and log it (throttled — see MERGE_SKIP_LOG_INTERVAL_MS). For lock reasons the log and result include the holder and time until the lock goes stale.
    private async mergeSkip(reason: MergeSkipReason): Promise<MergeAttemptResult> {
        let lock: MergeLockInfo | undefined;
        if (reason === "tabLockHeld") {
            lock = peekMergeLock(this.name);
        } else if (reason === "fileLockHeld") {
            lock = await peekMergeFileLock(await this.storage());
        }
        if (Date.now() - this.lastMergeSkipLogMs >= MERGE_SKIP_LOG_INTERVAL_MS) {
            this.lastMergeSkipLogMs = Date.now();
            const detail = lock && ` (held by ${lock.holderId}, expires in ${formatTime(Math.max(0, lock.expiresInMs))})` || "";
            console.log(`${blue(this.name)} ${magenta("merge")} skipped: ${reason}${detail}`);
        }
        return { merged: false, skipReason: reason, lockHolderId: lock?.holderId, lockExpiresInMs: lock?.expiresInMs };
    }

    // Acquire both merge locks (same-tab mergeInFlight, cross-tab localStorage, cross-process file), run the pass, release. Skips (with the reason) instead of waiting when any of them is held.
    private async runLockedMerge(run: () => Promise<MergeAttemptResult>): Promise<MergeAttemptResult> {
        if (this.mergeInFlight) return await this.mergeSkip("mergeInFlight");
        if (!tryAcquireMergeLock(this.name, writerId)) return await this.mergeSkip("tabLockHeld");
        const storage = await this.storage();
        const haveFileLock = await tryAcquireMergeFileLock(storage, writerId);
        if (!haveFileLock) {
            releaseMergeLock(this.name, writerId);
            return await this.mergeSkip("fileLockHeld");
        }
        this.mergeInFlight = true;
        const stopHeartbeat = startMergeFileLockHeartbeat(storage, writerId);
        try {
            return await run();
        } finally {
            stopHeartbeat();
            await releaseMergeFileLock(storage, writerId);
            releaseMergeLock(this.name, writerId);
            this.mergeInFlight = false;
        }
    }

    // The background path (maybeMerge) goes through this so write bursts coalesce; explicit callers use tryMergeNow directly and get the per-call result.
    private tryMergeThrottled = throttleFunction(1000, async () => {
        await this.tryMergeNow();
    });

    public async tryMergeNow(): Promise<MergeAttemptResult> {
        return await this.runLockedMerge(async () => {
            const merged = await this.testMergeINTERNAL_DO_NOT_CALL();
            return { merged };
        });
    }

    public async compact(): Promise<MergeAttemptResult> {
        return await this.runLockedMerge(async () => {
            await this.flushPending();
            syncBroadcastSeal(this.name);
            this.streamFileName = undefined;
            const { bulkFiles, streamFiles } = await this.listFiles();
            if (bulkFiles.length + streamFiles.length < 1) return await this.mergeSkip("nothingToMerge");
            // compact() folds every file → no older data survives outside it → surviving tombstones can be dropped (nothing left to suppress).
            const merged = await this.mergeFileSet(bulkFiles, streamFiles, true);
            return { merged };
        });
    }

    public async merge(timeLo: number, timeHi: number): Promise<void> {
        if (timeHi >= Date.now()) { syncBroadcastSeal(this.name); this.streamFileName = undefined; }
        const { bulkFiles, streamFiles } = await this.listFiles();
        const headers = await Promise.all(bulkFiles.map(f => this.readBulkHeader(f.fileName)));
        const selBulk = bulkFiles.filter((f, i) => {
            const h = headers[i];
            if (!h) return false;
            if (!h.maxTime && !h.minTime) return timeLo <= 0;
            return h.minTime <= timeHi && h.maxTime >= timeLo;
        });
        const selStream = streamFiles.filter(f =>
            f.timestamp <= timeHi && f.timestamp + bulkDatabase2Timing.streamSealAgeMs >= timeLo);
        if (selBulk.length + selStream.length < 2) return;
        await this.mergeFileSet(selBulk, selStream, timeLo <= 0);
    }

    private async readBulkHeader(fileName: string): Promise<BulkHeaderInfo | undefined> {
        try {
            const storage = await this.storage();
            const raw = await makeRawGetRange(storage, fileName);
            const fileId = nullJoin(this.name, fileName);
            const opened = await blockCache.open(fileId, raw.size, raw.rawGetRange);
            return await loadBulkHeader(opened.getRange, opened.uncompressedSize);
        } catch {
            return undefined;
        }
    }

    private async fileLogicalSize(fileName: string): Promise<number | undefined> {
        try {
            const storage = await this.storage();
            const raw = await makeRawGetRange(storage, fileName);
            const fileId = nullJoin(this.name, fileName);
            const opened = await blockCache.open(fileId, raw.size, raw.rawGetRange);
            return opened.uncompressedSize;
        } catch {
            return undefined;
        }
    }

    // A bulk file that won't load is either an in-progress write (recent) or a crashed partial write (stale). Warn while recent, delete once clearly abandoned — deleting is safe because the write protocol always lands the replacement before removing the file it supersedes.
    private async handleUnreadableFile(file: BulkFileInfo, message: string): Promise<void> {
        const ageMs = Date.now() - file.timestamp;
        if (ageMs > STALE_DELETE_MS) {
            console.warn(`${this.name}: deleting stale unreadable bulk file ${file.fileName} (${Math.round(ageMs / 86400000)}d old): ${message}`);
            try {
                const storage = await this.storage();
                await storage.remove(file.fileName);
            } catch (removeError) {
                console.warn(`${this.name}: failed to delete ${file.fileName}: ${(removeError as Error).message}`);
            }
            return;
        }
        console.warn(`${this.name}: skipping unreadable bulk file ${file.fileName} (recent - may be in-progress): ${message}`);
    }

    // The one merge primitive. Reads + plans + writes outputs before deleting any input, so a crash leaves duplicates (next merge dedupes) rather than a gap. After the file set changes on disk, we trigger an index rebuild + atomic swap; once swap completes, the consumed files' block-cache entries are evicted (no consumer can ask for them now).
    // 
    // Serialized: foldOwnStream + merge(timeLo, timeHi) + testMerge's runMerge all hit this; the lock covers tryMergeNow/compact but those two paths bypass it. runInSerial queues so they never collide on the file set.
    private mergeFileSet = runInSerial(async (bulkFiles: BulkFileInfo[], streamFiles: StreamFileInfo[], includesOldest = false, forceDeleteStreams = false): Promise<boolean> => {
        this.reader.beginCompaction();
        try {
            return await this.mergeFileSetInner(bulkFiles, streamFiles, includesOldest, forceDeleteStreams);
        } finally {
            this.reader.endCompaction();
        }
    });

    private async mergeFileSetInner(bulkFiles: BulkFileInfo[], streamFiles: StreamFileInfo[], includesOldest: boolean, forceDeleteStreams: boolean): Promise<boolean> {
        const storage = await this.storage();
        const timestamp = nextFileTime();

        // The caller's bulkFiles came from a `listFiles()` snapshot. Between then and now any file could be gone (another tab's deferred delete, manual cleanup). The subCaches reader for a deleted file is silently stale — its in-memory metadata still resolves, but the very next getRange against disk throws. Re-verify existence up-front and drop the cache entry for anything that vanished, so we don't even try to plan around a ghost file.
        const verifiedBulkFiles = (await Promise.all(bulkFiles.map(async f => {
            try {
                const info = await storage.getInfo(f.fileName);
                if (!info) { this.subCaches.bulk.delete(f.fileName); return undefined; }
                return f;
            } catch {
                this.subCaches.bulk.delete(f.fileName);
                return undefined;
            }
        }))).filter((f): f is BulkFileInfo => f !== undefined);
        bulkFiles = verifiedBulkFiles;

        const consumedBulk: BulkFileInfo[] = [];
        const bulkReaders: BaseBulkDatabaseReader[] = [];
        await Promise.all(bulkFiles.map(async f => {
            try {
                const r = await loadFileReader(this.name, storage, f, this.subCaches.bulk);
                bulkReaders.push(r);
                consumedBulk.push(f);
            } catch { /* missing or corrupt — skip; its data lives in another file */ }
        }));

        const streamData = await loadStreamEntries(this.name, storage, streamFiles, this.subCaches.stream);
        const ordered = orderStreamEntries(streamData.entries);
        const streamReader = ordered.length ? streamReaderFromEntries(ordered, 0).reader : undefined;

        // An abandoned stream that yielded no entries (zero bytes, or nothing but torn bytes) holds no data and has no writer left. Retire it here: the merge below never lists it as a used source, so the normal retirement path would skip it and it would re-trigger the abandoned-stream fold on every pass. replacedBy is empty because nothing supersedes it - the marker hides it from reads immediately and processMarkers deletes it once aged.
        const retirableStreams = await this.findAbandonedStreams(streamFiles, false);
        const contributingStreams = new Set(streamData.entries.map(e => e.fileName));
        const emptyAbandoned = streamFiles.filter(f => retirableStreams.has(f.fileName) && !contributingStreams.has(f.fileName)).map(f => f.fileName);
        if (emptyAbandoned.length) await writeDeleteMarker(storage, { deleteFiles: emptyAbandoned, replacedBy: [] });

        const readers = streamReader ? [streamReader, ...bulkReaders] : bulkReaders;
        const readerNames = streamReader ? ["(streams)", ...consumedBulk.map(f => f.fileName)] : consumedBulk.map(f => f.fileName);
        if (!readers.length) {
            if (emptyAbandoned.length) await this.triggerRebuild();
            return emptyAbandoned.length > 0;
        }

        const inputs = [
            ...await Promise.all(consumedBulk.map(async f => ({ name: f.fileName, size: (await storage.getInfo(f.fileName).catch(() => undefined))?.size ?? 0 }))),
            ...streamFiles.map(f => ({ name: f.fileName, size: streamData.sizes.get(f.fileName) ?? 0 })),
        ];
        const inTotal = inputs.reduce((a, f) => a + f.size, 0);
        const mergeStartMs = Date.now();
        // Collect each step instead of logging it live; emitted at the end as one
        // expanded console.group so a merge takes one collapsible block, not a
        // screenful (and never the per-file name dump it used to spew).
        const steps: string[] = [];
        // Each step records the time spent since the previous step in blue at the front, so callers can just describe what they did and not bother measuring/formatting elapsed times themselves.
        let lastStepMs = mergeStartMs;
        const log = (line: string) => {
            const now = Date.now();
            let lineFormatted = `${blue(formatTime(now - lastStepMs))} ${line}`;
            // DO NOT REMOVE THIS LOG! Obviously, we should be logging as we run, or else we don't get progress. Why would we stop getting progress? Progress is almost more important than showing a summary at the end...
            console.log(lineFormatted);
            steps.push(lineFormatted);
            lastStepMs = now;
        };
        log(`${magenta("read")}: ${inputs.length} input file(s), ${fmtBytes(inTotal)}`);

        const newNames: string[] = [];
        const mergeResult = await runPlannedMerge({
            sources: readers,
            sourceNames: readerNames,
            collectionName: this.name,
            log,
            writeFile: async (data) => {
                const fname = newFileName(timestamp);
                await storage.set(fname, encodeCompressedBlocks(data));
                newNames.push(fname);
                const size = (await storage.getInfo(fname).catch(() => undefined))?.size ?? 0;
                return { name: fname, size };
            },
        });

        const carriedDeletes = includesOldest ? 0 : mergeResult.carriedDeletes.size;
        const outNames = [...newNames];
        if (carriedDeletes) {
            const carryName = newStreamFileName(MERGE_OUTPUT_OWNER);
            await storage.set(carryName, frameDeletes([...mergeResult.carriedDeletes].map(([key, time]) => ({ time, key }))));
            outNames.push(carryName);
        }

        const outputs = await Promise.all(outNames.map(async n => ({ name: n, size: (await storage.getInfo(n).catch(() => undefined))?.size ?? 0 })));
        const outTotal = outputs.reduce((a, f) => a + f.size, 0);
        log(`${magenta("wrote")}: ${outputs.length} output file(s), ${fmtBytes(outTotal)}${carriedDeletes ? `, ${carriedDeletes} tombstones carried` : ""}`);

        console.group(`${blue(this.name)} ${magenta("merge")}: ${fmtBytes(inTotal)} -> ${fmtBytes(outTotal)} (${inputs.length}->${outputs.length} files) in ${formatTime(Date.now() - mergeStartMs)}`);
        for (const line of steps) console.log(line);
        console.groupEnd();

        // Only the sources runPlannedMerge actually used can be retired — a source it dropped mid-plan (file gone, corruption) still holds data we couldn't merge in, so leave it on disk and let a future merge re-attempt it.
        const usedConsumedBulk = consumedBulk.filter(f => mergeResult.usedSourceNames.has(f.fileName));
        const usedStreamFiles = streamFiles.filter(f => mergeResult.usedSourceNames.has(f.fileName) || mergeResult.usedSourceNames.has("(streams)"));
        // A stream may still be appended to after we read it (the writer hasn't sealed/moved on); only retire streams canDeleteStream clears — the rest stay live and a later merge re-folds them once sealed. Bulk files are immutable, so a used one is always safe to retire.
        const deletableStreams: string[] = [];
        for (const f of usedStreamFiles) {
            if (retirableStreams.has(f.fileName)) { deletableStreams.push(f.fileName); continue; }
            if (await this.canDeleteStream(f, Date.now(), streamData.sizes, forceDeleteStreams)) deletableStreams.push(f.fileName);
        }
        const deleteFiles = [...usedConsumedBulk.map(f => f.fileName), ...deletableStreams];

        // Write a deletion marker instead of deleting inline. From the next read on, listFiles hides these inputs (so they're never re-merged — the loop fix), and processMarkers removes them physically once the outputs have landed or the marker ages out. Marker BEFORE the rebuild so the freshly-swapped index already excludes the retired inputs.
        if (deleteFiles.length) await writeDeleteMarker(storage, { deleteFiles, replacedBy: outNames });

        // Rebuild + swap so the index sees the new outputs and drops the retired inputs. Block-cache eviction for the dropped inputs happens here (they're no longer in the new index's fileSet).
        await this.triggerRebuild();
        return newNames.length > 0 || carriedDeletes > 0;
    }

    // A stream is safe to delete iff no writer will append to it again: it's aged past the seal age (writer has provably switched files) OR cross-tab sync is on AND its size didn't change while we read it. Else leave it — the data is also in the new bulk file; a later merge deletes it once aged.
    private async canDeleteStream(f: StreamFileInfo, now: number, sizes: Map<string, number>, force = false): Promise<boolean> {
        if (now - f.timestamp >= bulkDatabase2Timing.streamSealAgeMs) return true;
        if (!isSyncSupported() && !force) return false;
        const readSize = sizes.get(f.fileName);
        if (readSize === undefined) return false;
        let info;
        try { info = await (await this.storage()).getInfo(f.fileName); } catch { return false; }
        return !!info && info.size === readSize;
    }

    private async mergeSpacingDelay(): Promise<boolean> {
        const total = bulkDatabase2Timing.mergeSpacingMs;
        if (total <= 0) return tryAcquireMergeLock(this.name, writerId);
        const step = 15 * 1000;
        let waited = 0;
        while (waited < total) {
            await new Promise<void>(r => setTimeout(r, Math.min(step, total - waited)));
            waited += step;
            if (!tryAcquireMergeLock(this.name, writerId)) return false;
        }
        return true;
    }

    // Splits the bulk tier into files still worth rolling up ("loose", under LOOSE_BULK_MAX_BYTES - a single stream fold, or the tail end of an earlier merge) and files that are done growing ("combined", which only phase 3 touches again).
    //
    // A file whose size won't read is reported as combined: phase 2 can't consume one (its reader won't load, so the merge won't retire it), and calling it loose would re-trigger phase 2 on every pass until handleUnreadableFile finally deletes it.
    private async splitBulkTier(bulkFiles: BulkFileInfo[]): Promise<{ loose: BulkFileInfo[]; looseBytes: number; combined: BulkFileInfo[]; sizes: Map<string, number> }> {
        const logicalSizes = await Promise.all(bulkFiles.map(f => this.fileLogicalSize(f.fileName)));
        const loose: BulkFileInfo[] = [];
        const combined: BulkFileInfo[] = [];
        const sizes = new Map<string, number>();
        let looseBytes = 0;
        for (let i = 0; i < bulkFiles.length; i++) {
            const bytes = logicalSizes[i];
            sizes.set(bulkFiles[i].fileName, bytes ?? 0);
            if (bytes === undefined || bytes >= LOOSE_BULK_MAX_BYTES) {
                combined.push(bulkFiles[i]);
                continue;
            }
            loose.push(bulkFiles[i]);
            looseBytes += bytes;
        }
        return { loose, looseBytes, combined, sizes };
    }

    // One walk of the combined tier's key lists, producing both of phase 3's inputs: the whole-tier duplicate fraction, and the per-key-range groups. Combined into one pass because walking every file's keys is the most expensive thing a merge pass does that isn't a merge, and both answers come from the same key counts.
    private async analyzeDuplicates(bulkFiles: BulkFileInfo[]): Promise<{ dupFraction: number; groups: { lo: string; hi: string; dup: number }[] }> {
        const storage = await this.storage();
        const infos = await Promise.all(bulkFiles.map(async f => {
            try {
                const reader = await loadFileReader(this.name, storage, f, this.subCaches.bulk);
                return { keys: reader.keys, bytes: reader.totalBytes };
            } catch {
                return { keys: [] as string[], bytes: 0 };
            }
        }));
        const keyCount = new Map<string, number>();
        let totalSlots = 0, totalBytes = 0;
        for (const i of infos) {
            totalBytes += i.bytes;
            for (const k of i.keys) { keyCount.set(k, (keyCount.get(k) || 0) + 1); totalSlots++; }
        }
        if (!totalSlots) return { dupFraction: 0, groups: [] };
        const bytesPerSlot = totalBytes / totalSlots;
        const sortedKeys = [...keyCount.keys()].sort();
        const groups: { lo: string; hi: string; dup: number }[] = [];
        let gStart = 0, gBytes = 0, gSlots = 0, gUnique = 0;
        for (let i = 0; i < sortedKeys.length; i++) {
            const c = keyCount.get(sortedKeys[i]) ?? 0;
            gBytes += c * bytesPerSlot; gSlots += c; gUnique += 1;
            if (gBytes >= KEY_GROUP_BYTES || i === sortedKeys.length - 1) {
                groups.push({ lo: sortedKeys[gStart], hi: sortedKeys[i], dup: (gSlots - gUnique) / gSlots });
                gStart = i + 1; gBytes = 0; gSlots = 0; gUnique = 0;
            }
        }
        sort(groups, g => -g.dup);
        return { dupFraction: (totalSlots - keyCount.size) / totalSlots, groups };
    }

    // Which files a key-group step rewrites. Re-run at merge time as well as at plan time: groups are disjoint by KEY range, not by file, so an earlier step in the same pass can consume a file this range still lists.
    private async filesForKeyRange(bulkFiles: BulkFileInfo[], keyRange: { lo: string; hi: string }): Promise<BulkFileInfo[]> {
        const headers = await Promise.all(bulkFiles.map(f => this.readBulkHeader(f.fileName)));
        return bulkFiles.filter((f, i) => {
            const h = headers[i];
            if (!h) return false;
            if (h.minKey === undefined || h.maxKey === undefined) return true;
            return h.minKey <= keyRange.hi && h.maxKey >= keyRange.lo;
        });
    }

    /**
     * Every compaction the files on disk currently call for, without performing any of them. A merge pass
     * builds exactly this and then runs the steps whose `ready` is true, so the plan is precisely what the
     * database is about to do — and a step that isn't ready still reports its `triggers`, so a caller can
     * see how close it is (50MB of stream data out of the 64MB that would fold it, and so on).
     *
     * O(total keys): the phase 3 steps need every combined file's key list walked.
     */
    public async planCompaction(): Promise<CompactionPlan> {
        const time = Date.now();
        const steps: CompactionStep[] = [];
        const storage = await this.storage();
        const { bulkFiles, streamFiles } = await this.listFiles();

        const streamSizes = new Map<string, number>();
        await Promise.all(streamFiles.map(async f => {
            try { streamSizes.set(f.fileName, (await storage.getInfo(f.fileName))?.size ?? 0); } catch { streamSizes.set(f.fileName, 0); }
        }));
        const streamBytes = (files: StreamFileInfo[]) => files.reduce((a, f) => a + (streamSizes.get(f.fileName) ?? 0), 0);

        // ── Phase 1: stream -> bulk ──────────────────────────────────────────────────────────────────
        // The ENTIRE stream tier is parsed into memory and held there (subCaches.stream keeps every decoded entry) just to build the index, so it is the biggest single lever on our heap - a fold turns it into a bulk file we only read an index of. Folds streams and nothing else: no bulk file is dragged in, so the work is proportional to the memory reclaimed.
        const hardLimit = makeTrigger({ name: "streamBytes", value: streamBytes(streamFiles), threshold: bulkDatabase2Timing.streamFoldHardLimitBytes, unit: "bytes" });
        // Past the hard limit every read pulls an enormous file, so fold the whole tier regardless of who owns what. mergeFileSet force-deletes for this one; canDeleteStream still requires size-stability, so an active writer never loses data.
        steps.push({
            phase: 1, kind: "streamHardLimit", requires: "all", triggers: [hardLimit],
            ready: hardLimit.met && streamFiles.length > 0,
            bulkFiles: [], streamFiles, bytes: hardLimit.value,
        });

        // A pass seals before it merges, so by then our own current file is final too - predict that rather than reporting it as still-open.
        const retirable = await this.findAbandonedStreams(streamFiles, isSyncSupported());
        // Only fold what we can also retire. Folding a stream a live foreign owner may still append to would copy it into bulk without removing it, so the bytes would stay in memory and just get re-folded next pass; that owner rolls its own file over at streamFileMaxBytes instead.
        // Aged past streamSealAgeMs counts as retirable too (canDeleteStream's own first rule): no writer appends past the seal age, and this is the only thing that frees the tier in Node, where liveness cannot be probed at all.
        const foldable = streamFiles.filter(f => f.ownerId !== MERGE_OUTPUT_OWNER
            && (retirable.has(f.fileName) || time - f.timestamp >= bulkDatabase2Timing.streamSealAgeMs));
        // Merge-carry files hold nothing but tombstones and are never a REASON to fold - folding one alone would just rewrite it into another carry file, forever. They ride along whenever something else folds, which collapses however many have piled up into one.
        const carry = streamFiles.filter(f => f.ownerId === MERGE_OUTPUT_OWNER);
        // Size is the only reason to fold. An abandoned stream is NOT one on its own: it is already retirable, so it is counted here and gets swept up the moment the tier is worth folding, and until then it is bounded by this very threshold. Triggering on one would fold on essentially every pass - a browser leaves a file behind whose writer never answers the liveness probe on every reload - which mints a small bulk file each time and pushes the fragmentation into phase 2.
        const foldTriggers = [
            makeTrigger({ name: "foldableBytes", value: streamBytes(foldable), threshold: bulkDatabase2Timing.streamFoldTriggerBytes, unit: "bytes" }),
        ];
        steps.push({
            phase: 1, kind: "streamFold", requires: "all", triggers: foldTriggers,
            // Skipped when the hard limit already folds everything this would have.
            ready: !hardLimit.met && foldTriggers.every(t => t.met),
            bulkFiles: [], streamFiles: [...foldable, ...carry], bytes: streamBytes([...foldable, ...carry]),
        });

        // ── Phase 2: loose bulk -> combined bulk ─────────────────────────────────────────────────────
        // Phase 1 emits one small bulk file per fold. Each is cheap to read (index only) but holds its whole key list in memory and joins into every read, so they have to be rolled up. Merging just the loose ones also dedupes them for free - a rewrite-heavy workload collapses a gigabyte of near-identical folds into almost nothing - and it always terminates: a chunk is only cut once the next key would take it past TARGET_FILE_BYTES, so every output but the last is over half the target and lands in the combined tier, leaving at most one loose file behind.
        const { loose, looseBytes, combined, sizes } = await this.splitBulkTier(bulkFiles);
        const looseTriggers = [
            makeTrigger({ name: "looseBytes", value: looseBytes, threshold: bulkDatabase2Timing.looseBulkTriggerBytes, unit: "bytes" }),
            makeTrigger({ name: "looseFiles", value: loose.length, threshold: bulkDatabase2Timing.looseBulkTriggerFiles, unit: "count" }),
        ];
        steps.push({
            phase: 2, kind: "looseCombine", requires: "any", triggers: looseTriggers,
            // Under two files there is nothing to combine, and rewriting one file into an identical one would re-trigger forever.
            ready: loose.length >= 2 && looseTriggers.some(t => t.met),
            bulkFiles: loose, streamFiles: [], bytes: looseBytes,
        });

        // ── Phase 3: dedup the combined files ────────────────────────────────────────────────────────
        // Loose files are excluded throughout: phase 2 rewrites them anyway, and that rewrite already dedupes them, so pulling one in here would do the same work twice at key-group scale.
        const combinedBytes = combined.reduce((a, f) => a + (sizes.get(f.fileName) ?? 0), 0);
        const { dupFraction, groups } = await this.analyzeDuplicates(combined);
        // Whole-tier short-circuit: when the combined tier is big enough AND mostly duplicates, fold all of it in one merge rather than paying the per-group walk's 5-min spacing x N groups.
        const dedupAllTriggers = [
            makeTrigger({ name: "combinedBytes", value: combinedBytes, threshold: DEDUP_TRIGGER_BYTES, unit: "bytes" }),
            makeTrigger({ name: "duplicateFraction", value: dupFraction, threshold: DEDUP_TRIGGER_FRACTION, unit: "fraction" }),
        ];
        const dedupAllReady = combined.length >= 2 && dedupAllTriggers.every(t => t.met);
        steps.push({
            phase: 3, kind: "dedupAll", requires: "all", triggers: dedupAllTriggers,
            ready: dedupAllReady,
            bulkFiles: combined, streamFiles: [], bytes: combinedBytes,
        });

        // Then key-stratified: disjoint key ranges, so one group's merge doesn't change another's duplication. Only groups over the threshold are worth a step; the best one below it is listed anyway so a caller can see how close the tier is.
        const groupSteps = groups.filter(g => g.dup >= DUP_THRESHOLD);
        if (!groupSteps.length && groups.length) groupSteps.push(groups[0]);
        for (const g of groupSteps) {
            const groupFiles = await this.filesForKeyRange(combined, g);
            const dup = makeTrigger({ name: "duplicateFraction", value: g.dup, threshold: DUP_THRESHOLD, unit: "fraction" });
            steps.push({
                phase: 3, kind: "dedupKeyGroup", requires: "all", triggers: [dup],
                // Moot if the whole tier is about to be folded in one go.
                ready: !dedupAllReady && dup.met && groupFiles.length >= 2,
                bulkFiles: groupFiles, streamFiles: [], bytes: groupFiles.reduce((a, f) => a + (sizes.get(f.fileName) ?? 0), 0),
                keyRange: { lo: g.lo, hi: g.hi },
            });
        }

        // Merges are spaced mergeSpacingMs apart, and the first one runs immediately.
        let readyCount = 0;
        for (const step of steps) {
            if (!step.ready) continue;
            step.startTime = time + readyCount * bulkDatabase2Timing.mergeSpacingMs;
            readyCount++;
        }
        return { collection: this.name, time, steps };
    }

    // Runs the compaction plan: builds it, then performs every step it marks ready, in order. All the deciding lives in planCompaction - this only executes, so what the database does and what planCompaction reports can never drift apart.
    private async testMergeINTERNAL_DO_NOT_CALL(): Promise<boolean> {
        let merged = false;
        await this.flushPending();

        // Seal before planning, so every writer's current stream file is final and the plan can count it as foldable. Only when cross-tab sync is there to carry the seal - in Node canDeleteStream needs streams aged anyway, so sealing would just fragment them every pass.
        if (isSyncSupported()) {
            syncBroadcastSeal(this.name);
            this.streamFileName = undefined;
        }

        const plan = await this.planCompaction();
        for (const step of plan.steps) {
            if (!step.ready) continue;
            let bulkFiles = step.bulkFiles;
            if (step.keyRange) {
                bulkFiles = await this.filesForKeyRange(bulkFiles, step.keyRange);
                if (bulkFiles.length < 2) continue;
            }
            const why = step.triggers.filter(t => t.met)
                .map(t => `${t.name} ${fmtTriggerValue(t.value, t.unit)} of ${fmtTriggerValue(t.threshold, t.unit)}`).join(", ");
            console.log(`${blue(this.name)} ${magenta(step.kind)} phase ${step.phase}: ${bulkFiles.length} bulk + ${step.streamFiles.length} stream file(s), ${fmtBytes(step.bytes)} - ${why}`);
            if (merged && !await this.mergeSpacingDelay()) return merged;
            if (await this.mergeFileSet(bulkFiles, step.streamFiles, false, step.kind === "streamHardLimit")) merged = true;
        }
        return merged;
    }

    // ── reads — forwarded to BulkDatabaseReader, with rebuild-on-missing retry ───────────────────────
    public async getSingleField<C extends keyof T>(key: string, column: C): Promise<T[C] | undefined> {
        void this.syncSetup();
        return this.readWithRetry(() => this.reader.getSingleField(key, column));
    }

    public async getSingleFieldObj<C extends keyof T>(key: string, column: C): Promise<{ key: string; value: T[C]; time: number } | undefined> {
        void this.syncSetup();
        return this.readWithRetry(() => this.reader.getSingleFieldObj(key, column));
    }

    public async getColumn<C extends keyof T>(column: C): Promise<{ key: string; value: T[C]; time: number }[]> {
        void this.syncSetup();
        return this.readWithRetry(() => this.reader.getColumn(column));
    }

    public async getKeys(): Promise<string[]> {
        void this.syncSetup();
        return this.readWithRetry(() => this.reader.getKeys());
    }

    public getSingleFieldSync<C extends keyof T>(key: string, column: C): T[C] | undefined {
        void this.syncSetup();
        return this.reader.getSingleFieldSync(key, column);
    }

    public getSingleFieldObjSync<C extends keyof T>(key: string, column: C): { key: string; value: T[C]; time: number } | undefined {
        void this.syncSetup();
        return this.reader.getSingleFieldObjSync(key, column);
    }

    public getColumnSync<C extends keyof T>(column: C): { key: string; value: T[C]; time: number }[] | undefined {
        void this.syncSetup();
        return this.reader.getColumnSync(column);
    }

    public isFieldLoadedSync<C extends keyof T>(key: string, column: C): boolean {
        void this.syncSetup();
        return this.reader.isFieldLoadedSync(key, column);
    }

    public isColumnLoadedSync<C extends keyof T>(column: C): boolean {
        void this.syncSetup();
        return this.reader.isColumnLoadedSync(column);
    }

    // Reactive: true while a merge is rewriting this collection's files. Use this in UI to show a "compacting…" indicator. Counts both background merges (maybeMerge) and explicit compact/merge calls. Becomes false once the new index is swapped in (the deferred-delete window is NOT counted).
    public isCompactingSync(): boolean {
        return this.reader.isCompactingSync();
    }

    public async getColumnInfo() {
        const index = await this.ensureIndex();
        return index.reader.columns;
    }

    public async getKeyStats(): Promise<{ rawKeys: number; finalKeys: number; wastedKeys: number; duplication: number; readers: number }> {
        const index = await this.ensureIndex();
        const rawKeys = index.reader.rawKeyCount;
        const finalKeys = index.reader.keys.length;
        return {
            rawKeys,
            finalKeys,
            wastedKeys: rawKeys - finalKeys,
            duplication: finalKeys ? rawKeys / finalKeys : 0,
            readers: index.reader.readerCount,
        };
    }

    public async getReaderInfo() {
        const index = await this.ensureIndex();
        return {
            rowCount: index.reader.rowCount,
            totalBytes: index.reader.totalBytes,
            keyCount: index.reader.keys.length,
            sampleKey: index.reader.keys[0] as string | undefined,
            columns: index.reader.columns,
        };
    }

    public async getFileInfo(): Promise<BulkFileInfoListing> {
        const { bulkFiles, streamFiles } = await this.listFiles();
        const storage = await this.storage();
        const statOf = async (name: string) => {
            try {
                const info = await storage.getInfo(name);
                return { bytes: info?.size ?? 0, lastModified: info?.lastModified ?? 0 };
            } catch { return { bytes: 0, lastModified: 0 }; }
        };
        const bulkInfos = await Promise.all(bulkFiles.map(async f => {
            const stat = await statOf(f.fileName);
            return {
                name: f.fileName,
                type: "bulk" as const,
                bytes: stat.bytes,
                lastModified: stat.lastModified,
                getDetails: async () => {
                    // Bulk: reader has keys + per-row keyTimes + header minTime/maxTime. Cached, so cheap.
                    const reader = await loadFileReader(this.name, storage, f, this.subCaches.bulk);
                    return { keys: reader.keys, minTime: reader.minTime, maxTime: reader.maxTime };
                },
            };
        }));
        const streamInfos = await Promise.all(streamFiles.map(async f => {
            const stat = await statOf(f.fileName);
            return {
                name: f.fileName,
                type: "stream" as const,
                bytes: stat.bytes,
                lastModified: stat.lastModified,
                getDetails: async () => {
                    // Stream: must parse the file to walk every entry — header has no key/time bounds. Cached.
                    const data = await loadStreamEntries(this.name, storage, [f], this.subCaches.stream);
                    const keys = new Set<string>();
                    let minTime = Infinity, maxTime = -Infinity;
                    for (const e of data.entries) {
                        if (e.time < minTime) minTime = e.time;
                        if (e.time > maxTime) maxTime = e.time;
                        if (e.entry.row) keys.add(e.entry.row.key as string);
                        else if (e.entry.deletedKey !== undefined) keys.add(e.entry.deletedKey);
                    }
                    return {
                        keys: [...keys],
                        minTime: minTime === Infinity ? 0 : minTime,
                        maxTime: maxTime === -Infinity ? 0 : maxTime,
                    };
                },
            };
        }));
        const files = [...bulkInfos, ...streamInfos];
        return { files, count: files.length, totalBytes: files.reduce((a, f) => a + f.bytes, 0) };
    }
}

export type BulkFileDetails = { keys: string[]; minTime: number; maxTime: number };
export type BulkFileEntry = {
    name: string;
    type: "bulk" | "stream";
    bytes: number;
    // Filesystem mtime (ms since epoch) — 0 if the storage layer didn't return one.
    lastModified: number;
    // Lazy: pulled from the cached sub-reader for bulk (free if loaded), or from a parse of the stream file (small — tier-0 size-capped, also cached). Call only when you need the per-file detail.
    getDetails: () => Promise<BulkFileDetails>;
};
export type BulkFileInfoListing = { files: BulkFileEntry[]; count: number; totalBytes: number };