City-Lifeline-Cockpit/src/modules/cockpit/services/pipeModelBatches.js

/** Batch lifecycle boundary. A successful load promise does not imply GPU readiness. */
export function createPipeModelBatches({ load, remove, onChange = () => {}, now = Date.now, retryMs = 30000 }) {
    const records = new Map()
    const state = () => {
        const covered = new Set(), incomplete = new Set()
        let pending = 0, failed = 0, batches = 0, instances = 0
        const byKind = {}
        for (const record of records.values()) {
            if (record.status === 'pending') pending += 1
            if (record.status === 'failed') failed += 1
            if (record.status !== 'ready') {
                for (const instance of record.group.instances) incomplete.add(instance.id)
                continue
            }
            batches += 1
            instances += record.group.instances.length
            byKind[record.group.kind] = (byKind[record.group.kind] || 0) + record.group.instances.length
            for (const instance of record.group.instances) covered.add(instance.id)
        }
        return { covered: [...covered].filter((id) => !incomplete.has(id)), pending, failed, batches, instances, byKind }
    }
    const changed = () => onChange(state())
    function dispose(record) {
        record.active = false
        if (record.model) { remove(record.model); record.model = null }
    }
    async function sync(groups) {
        const wanted = new Set(groups.map((group) => group.key))
        for (const [key, record] of records) {
            if (!wanted.has(key)) { dispose(record); records.delete(key) }
        }
        const jobs = []
        for (const group of groups) {
            const fingerprint = JSON.stringify(group)
            const previous = records.get(group.key)
            if (previous && previous.fingerprint === fingerprint && (previous.status !== 'failed' || now() - previous.failedAt < retryMs)) continue
            if (previous) dispose(previous)
            const record = { group, fingerprint, active: true, status: 'pending', model: null }
            records.set(group.key, record)
            const error = () => {
                if (!record.active) return
                record.status = 'failed'
                record.failedAt = now()
                if (record.model) { remove(record.model); record.model = null }
                changed()
            }
            jobs.push((async () => {
                try {
                    const model = await load(group, {
                        alive: () => record.active,
                        ready: () => { if (record.active && record.status === 'pending') { record.status = 'ready'; changed() } },
                        error,
                    })
                    if (!record.active || record.status === 'failed') remove(model)
                    else record.model = model
                } catch { error() }
            })())
        }
        changed()
        await Promise.all(jobs)
        return state()
    }
    function retain(predicate) {
        for (const [key, record] of records) {
            if (!predicate(record.group)) { dispose(record); records.delete(key) }
        }
        changed()
    }
    return { sync, state, retain, clear: () => sync([]) }
}