From c1955f26cb31f9a82ccb96cfba09d609746931ac Mon Sep 17 00:00:00 2001 From: MajoorWaldi Date: Sat, 28 Mar 2026 21:34:37 +0100 Subject: [PATCH] feat: Implement live asset management and deduplication logic in InfiniteScroll and asset handling --- js/entry.js | 41 +++++----- js/features/grid/InfiniteScroll.js | 28 ++++++- js/features/stacks/liveAssetGate.js | 79 +++++++++++++++++++ js/tests/infinite_scroll_upsert.vitest.mjs | 79 +++++++++++++++++++ js/tests/live_asset_gate.vitest.mjs | 73 +++++++++++++++++ mjr_am_backend/adapters/db/schema.py | 1 + mjr_am_backend/routes/handlers/assets_impl.py | 64 +++++++++++++-- tests/database/test_schema_heal.py | 3 +- 8 files changed, 339 insertions(+), 29 deletions(-) create mode 100644 js/features/stacks/liveAssetGate.js create mode 100644 js/tests/infinite_scroll_upsert.vitest.mjs create mode 100644 js/tests/live_asset_gate.vitest.mjs diff --git a/js/entry.js b/js/entry.js index c3cd50b9..ec7b10cb 100644 --- a/js/entry.js +++ b/js/entry.js @@ -40,6 +40,7 @@ import { } from "./features/bottomPanel/GeneratedFeedTab.js"; import { createJobFinalizeQueue } from "./features/stacks/finalizeQueue.js"; import { createExecutionAssetBuffer } from "./features/stacks/executionAssetBuffer.js"; +import { createLiveAssetGate } from "./features/stacks/liveAssetGate.js"; import { extractOutputFiles } from "./utils/extractOutputFiles.js"; import { post } from "./api/client.js"; import { ENDPOINTS } from "./api/endpoints.js"; @@ -387,9 +388,9 @@ const executionStarts = createTTLCache({ maxSize: EXECUTION_STARTS_MAX, }); const executionAssetBuffer = createExecutionAssetBuffer(); +const liveAssetGate = createLiveAssetGate(); let _lastStacksUpdateSignature = ""; let _lastStacksUpdateAt = 0; -const _pendingStackJobIds = new Set(); function isJobTrackingDebugEnabled() { try { @@ -421,6 +422,7 @@ const _stackFinalizeQueue = createJobFinalizeQueue({ job_id: targetJobId, }); if (!res?.ok) { + liveAssetGate.clearPendingJob(targetJobId); reportError( new Error(String(res?.error || "Auto stack failed")), "entry.execution_end.auto_stack", @@ -451,11 +453,11 @@ function notifyStacksUpdated(detail = {}) { }); try { const targeted = String(detail?.targeted_job_id || "").trim(); - if (targeted) _pendingStackJobIds.delete(targeted); + if (targeted) liveAssetGate.clearPendingJob(targeted); const stacks = Array.isArray(detail?.stacks) ? detail.stacks : []; for (const item of stacks) { const jobId = String(item?.job_id || "").trim(); - if (jobId) _pendingStackJobIds.delete(jobId); + if (jobId) liveAssetGate.clearPendingJob(jobId); } } catch (e) { console.debug?.(e); @@ -622,7 +624,7 @@ function flushBufferedGenerationIndex(jobId) { file_count: files.length, files: files.map((item) => String(item?.filename || item?.filepath || "")), }); - _pendingStackJobIds.add(promptId); + liveAssetGate.markPendingJob(promptId, { files, expectedCount: files.length }); if (!files.length) { scheduleFinalizeExecutionStacks(promptId); return; @@ -634,7 +636,7 @@ function flushBufferedGenerationIndex(jobId) { }) .then((res) => { if (!res?.ok) { - _pendingStackJobIds.delete(promptId); + liveAssetGate.clearPendingJob(promptId); reportError( new Error(String(res?.error || "Indexing generated files failed")), "entry.executed.index", @@ -648,21 +650,17 @@ function flushBufferedGenerationIndex(jobId) { scheduleFinalizeExecutionStacks(promptId, 200); }) .catch((error) => { - _pendingStackJobIds.delete(promptId); + liveAssetGate.clearPendingJob(promptId); reportError(error, "entry.executed.index"); }); } -function shouldDeferLiveAssetEvent(detail) { +function prepareLiveAssetEvent(detail) { try { - const jobId = String(detail?.job_id || "").trim(); - if (!jobId) return false; - if (!_pendingStackJobIds.has(jobId)) return false; - const stackId = String(detail?.stack_id || "").trim(); - return !stackId; + return liveAssetGate.prepare(detail); } catch (e) { console.debug?.(e); - return false; + return { detail, jobId: "", defer: false }; } } @@ -898,7 +896,7 @@ app.registerExtension({ reason: String(event?.type || ""), }); executionAssetBuffer.clear(promptId); - _pendingStackJobIds.delete(promptId); + liveAssetGate.clearPendingJob(promptId); } if (String(event?.type || "") === "execution_success") { flushBufferedGenerationIndex(promptId); @@ -1002,15 +1000,17 @@ app.registerExtension({ try { const grid = getActiveGridContainer(); if (grid && event?.detail) { - if (shouldDeferLiveAssetEvent(event.detail)) return; - pushGeneratedAsset(event.detail); + const liveEvent = prepareLiveAssetEvent(event.detail); + if (liveEvent?.defer) return; + const detail = liveEvent?.detail || event.detail; + pushGeneratedAsset(detail); // Only handle direct updates when the active grid shows output or "all" scope. const scope = grid.dataset?.mjrScope || "output"; if (scope !== "output" && scope !== "all") return; const query = grid.dataset?.mjrQuery || "*"; const canDirectUpsert = String(query).trim() === "*"; if (canDirectUpsert) { - const handled = !!upsertAsset(grid, event.detail); + const handled = !!upsertAsset(grid, detail); if (handled) { // Signal that we handled this event so handleCountersUpdate // skips its own reload within the next window. @@ -1039,9 +1039,10 @@ app.registerExtension({ try { const grid = getActiveGridContainer(); if (grid && event?.detail) { - if (shouldDeferLiveAssetEvent(event.detail)) return; - pushGeneratedAsset(event.detail); - const detail = event.detail; + const liveEvent = prepareLiveAssetEvent(event.detail); + if (liveEvent?.defer) return; + const detail = liveEvent?.detail || event.detail; + pushGeneratedAsset(detail); const assetId = String(detail?.id || "").trim(); const kind = String(detail?.kind || "").trim(); const filename = String(detail?.filename || "").trim(); diff --git a/js/features/grid/InfiniteScroll.js b/js/features/grid/InfiniteScroll.js index 96531e3c..5e4ab54e 100644 --- a/js/features/grid/InfiniteScroll.js +++ b/js/features/grid/InfiniteScroll.js @@ -224,6 +224,24 @@ export function getUpsertBatchState(gridContainer, upsertState) { return s; } +function dedupeAssetsByKey(state, deps) { + const seenIds = new Set(); + const seenKeys = new Set(); + const deduped = []; + for (const asset of Array.isArray(state?.assets) ? state.assets : []) { + const assetId = asset?.id != null ? String(asset.id) : ""; + const key = deps.assetKey(asset); + if (assetId && seenIds.has(assetId)) continue; + if (key && seenKeys.has(key)) continue; + if (assetId) seenIds.add(assetId); + if (key) seenKeys.add(key); + deduped.push(asset); + } + state.assets = deduped; + state.assetIdSet = seenIds; + state.seenKeys = seenKeys; +} + /** * Flush all pending upsert operations into the grid. * @@ -288,8 +306,15 @@ export function flushUpsertBatch(gridContainer, deps) { const existingIndex = state.assets.findIndex((a) => String(a.id) === assetId); if (existingIndex > -1) { const existingAsset = state.assets[existingIndex]; + const previousKey = deps.assetKey(existingAsset); Object.assign(existingAsset, asset); - state.assets[existingIndex] = { ...existingAsset }; + const mergedAsset = { ...existingAsset }; + const nextKey = deps.assetKey(mergedAsset); + state.assets[existingIndex] = mergedAsset; + if (previousKey && previousKey !== nextKey) { + state.seenKeys?.delete?.(previousKey); + if (nextKey) state.seenKeys?.add?.(nextKey); + } modified = true; } else { const alreadySeen = @@ -306,6 +331,7 @@ export function flushUpsertBatch(gridContainer, deps) { } } if (modified && vg) { + dedupeAssetsByKey(state, deps); try { gridContainer.dataset.mjrHiddenPngSiblings = String( Number(state.hiddenPngSiblings || 0) || 0, diff --git a/js/features/stacks/liveAssetGate.js b/js/features/stacks/liveAssetGate.js new file mode 100644 index 00000000..7e5c4498 --- /dev/null +++ b/js/features/stacks/liveAssetGate.js @@ -0,0 +1,79 @@ +function _normalizeJobId(jobId) { + return String(jobId || "").trim(); +} + +function _assetFileKey(detail) { + const filename = String(detail?.filename || "").trim(); + const filepath = String(detail?.filepath || "").trim(); + const subfolder = String(detail?.subfolder || "").trim(); + const type = String(detail?.type || detail?.source || "output") + .trim() + .toLowerCase(); + const rootId = String(detail?.root_id || detail?.custom_root_id || "").trim(); + return `${type}|${rootId}|${subfolder}|${filename || filepath}`; +} + +function _copyWithJobId(detail, jobId) { + if (!jobId) return detail; + if (String(detail?.job_id || "").trim() === jobId) return detail; + return { ...(detail || {}), job_id: jobId }; +} + +export function createLiveAssetGate() { + const pendingJobs = new Map(); + + function markPendingJob(jobId, { files = [], expectedCount = null } = {}) { + const normalizedJobId = _normalizeJobId(jobId); + if (!normalizedJobId) return; + const fileKeys = new Set(); + for (const file of Array.isArray(files) ? files : []) { + const key = _assetFileKey(file); + if (key) fileKeys.add(key); + } + pendingJobs.set(normalizedJobId, { + expectedCount: Math.max(0, Number(expectedCount ?? fileKeys.size) || 0), + fileKeys, + }); + } + + function clearPendingJob(jobId) { + const normalizedJobId = _normalizeJobId(jobId); + if (!normalizedJobId) return; + pendingJobs.delete(normalizedJobId); + } + + function resolveJobId(detail) { + const explicit = _normalizeJobId(detail?.job_id); + if (explicit) return explicit; + const key = _assetFileKey(detail); + if (!key) return ""; + for (const [jobId, meta] of pendingJobs.entries()) { + if (meta?.fileKeys?.has?.(key)) return jobId; + } + return ""; + } + + function prepare(detail) { + const jobId = resolveJobId(detail); + if (!jobId) return { detail, jobId: "", defer: false }; + const normalizedDetail = _copyWithJobId(detail, jobId); + const pending = pendingJobs.get(jobId); + if (!pending) return { detail: normalizedDetail, jobId, defer: false }; + const stackId = String(normalizedDetail?.stack_id || "").trim(); + const expectedCount = Math.max(0, Number(pending?.expectedCount || 0) || 0); + const isMultiOutput = expectedCount > 1; + return { + detail: normalizedDetail, + jobId, + defer: isMultiOutput && !stackId, + }; + } + + return { + markPendingJob, + clearPendingJob, + resolveJobId, + prepare, + _pendingJobs: pendingJobs, + }; +} diff --git a/js/tests/infinite_scroll_upsert.vitest.mjs b/js/tests/infinite_scroll_upsert.vitest.mjs new file mode 100644 index 00000000..623b390f --- /dev/null +++ b/js/tests/infinite_scroll_upsert.vitest.mjs @@ -0,0 +1,79 @@ +import { describe, expect, it, vi } from "vitest"; + +import { + flushUpsertBatch, + getUpsertBatchState, + upsertAsset, +} from "../features/grid/InfiniteScroll.js"; + +function assetKey(asset) { + const stackId = String(asset?.stack_id || "").trim(); + if (stackId) return `stack:${stackId}`; + return `id:${String(asset?.id || "")}`; +} + +describe("InfiniteScroll upsert", () => { + it("refreshes seen keys when an existing asset gains a stack id", () => { + const gridContainer = { dataset: { mjrSort: "mtime_desc" } }; + const state = { + assets: [{ id: 1, filename: "one.png", mtime: 10 }], + seenKeys: new Set(["id:1"]), + assetIdSet: new Set(["1"]), + hiddenPngSiblings: 0, + }; + const setItems = vi.fn(); + const deps = { + upsertState: new Map(), + getOrCreateState: () => state, + ensureVirtualGrid: () => ({ setItems }), + assetKey, + loadMajoorSettings: () => ({}), + }; + + const batchState = getUpsertBatchState(gridContainer, deps.upsertState); + batchState.pending.set("1", { id: 1, stack_id: "stack-a", filename: "one.png", mtime: 10 }); + + flushUpsertBatch(gridContainer, deps); + + expect(state.seenKeys.has("id:1")).toBe(false); + expect(state.seenKeys.has("stack:stack-a")).toBe(true); + expect(state.assets).toHaveLength(1); + expect(state.assets[0].stack_id).toBe("stack-a"); + expect(setItems).toHaveBeenCalledTimes(1); + }); + + it("dedupes cards after a key change collides with an existing stack card", () => { + const gridContainer = { dataset: { mjrSort: "mtime_desc" } }; + const state = { + assets: [ + { id: 1, filename: "one.png", mtime: 20 }, + { id: 2, filename: "two.png", mtime: 10, stack_id: "stack-a" }, + ], + seenKeys: new Set(["id:1", "stack:stack-a"]), + assetIdSet: new Set(["1", "2"]), + hiddenPngSiblings: 0, + }; + const setItems = vi.fn(); + const deps = { + upsertState: new Map(), + getOrCreateState: () => state, + ensureVirtualGrid: () => ({ setItems }), + assetKey, + loadMajoorSettings: () => ({}), + maxBatchSize: 50, + debounceMs: 0, + }; + + expect(upsertAsset(gridContainer, { id: 1, stack_id: "stack-a", mtime: 30 }, deps)).toBe( + true, + ); + flushUpsertBatch(gridContainer, deps); + + expect(state.assets).toHaveLength(1); + expect(state.assets[0].id).toBe(1); + expect(state.seenKeys.has("stack:stack-a")).toBe(true); + expect(state.assetIdSet.has("1")).toBe(true); + expect(state.assetIdSet.has("2")).toBe(false); + expect(setItems).toHaveBeenCalledTimes(1); + }); +}); diff --git a/js/tests/live_asset_gate.vitest.mjs b/js/tests/live_asset_gate.vitest.mjs new file mode 100644 index 00000000..00b97464 --- /dev/null +++ b/js/tests/live_asset_gate.vitest.mjs @@ -0,0 +1,73 @@ +import { describe, expect, it } from "vitest"; + +import { createLiveAssetGate } from "../features/stacks/liveAssetGate.js"; + +describe("createLiveAssetGate", () => { + it("defers multi-output assets until stack assignment is available", () => { + const gate = createLiveAssetGate(); + gate.markPendingJob("job-multi", { + expectedCount: 2, + files: [ + { filename: "a.png", subfolder: "gen", type: "output" }, + { filename: "b.png", subfolder: "gen", type: "output" }, + ], + }); + + const pending = gate.prepare({ + filename: "a.png", + subfolder: "gen", + type: "output", + job_id: "job-multi", + }); + expect(pending.defer).toBe(true); + expect(pending.detail.job_id).toBe("job-multi"); + + const stacked = gate.prepare({ + filename: "a.png", + subfolder: "gen", + type: "output", + job_id: "job-multi", + stack_id: "42", + }); + expect(stacked.defer).toBe(false); + expect(stacked.detail.stack_id).toBe("42"); + }); + + it("allows mono-output assets through normal upsert while preserving job id", () => { + const gate = createLiveAssetGate(); + gate.markPendingJob("job-single", { + expectedCount: 1, + files: [{ filename: "solo.png", subfolder: "gen", type: "output" }], + }); + + const prepared = gate.prepare({ + filename: "solo.png", + subfolder: "gen", + type: "output", + }); + + expect(prepared.defer).toBe(false); + expect(prepared.jobId).toBe("job-single"); + expect(prepared.detail.job_id).toBe("job-single"); + expect(prepared.detail.stack_id).toBeUndefined(); + }); + + it("stops deferring once the pending job is cleared", () => { + const gate = createLiveAssetGate(); + gate.markPendingJob("job-clear", { + expectedCount: 3, + files: [{ filename: "x.png", subfolder: "gen", type: "output" }], + }); + gate.clearPendingJob("job-clear"); + + const prepared = gate.prepare({ + filename: "x.png", + subfolder: "gen", + type: "output", + job_id: "job-clear", + }); + + expect(prepared.defer).toBe(false); + expect(prepared.detail.job_id).toBe("job-clear"); + }); +}); diff --git a/mjr_am_backend/adapters/db/schema.py b/mjr_am_backend/adapters/db/schema.py index a437df00..4b9fb356 100644 --- a/mjr_am_backend/adapters/db/schema.py +++ b/mjr_am_backend/adapters/db/schema.py @@ -137,6 +137,7 @@ COLUMN_DEFINITIONS = { ("rating", "rating INTEGER DEFAULT 0"), ("tags", "tags TEXT DEFAULT ''"), ("tags_text", "tags_text TEXT DEFAULT ''"), + ("metadata_text", "metadata_text TEXT DEFAULT ''"), ("workflow_hash", "workflow_hash TEXT"), ("has_workflow", "has_workflow BOOLEAN DEFAULT 0"), ("has_generation_data", "has_generation_data BOOLEAN DEFAULT 0"), diff --git a/mjr_am_backend/routes/handlers/assets_impl.py b/mjr_am_backend/routes/handlers/assets_impl.py index daf7d451..4a495525 100644 --- a/mjr_am_backend/routes/handlers/assets_impl.py +++ b/mjr_am_backend/routes/handlers/assets_impl.py @@ -946,7 +946,8 @@ def register_asset_routes(routes: web.RouteTableDef) -> None: asset_row = await _query_one( f""" SELECT id, filename, subfolder, filepath, source, root_id, kind, ext, size, mtime, - width, height, duration, enhanced_caption, + width, height, duration, enhanced_caption, job_id, stack_id, + source_node_id, source_node_type, created_at, updated_at, indexed_at, content_hash, phash, hash_state FROM assets WHERE {filepath_where} @@ -1013,9 +1014,33 @@ def register_asset_routes(routes: web.RouteTableDef) -> None: if asset: await db.aexecute( """ - INSERT OR REPLACE INTO assets - (id, filename, subfolder, filepath, source, root_id, kind, ext, size, mtime, width, height, duration, enhanced_caption, created_at, updated_at, indexed_at, content_hash, phash, hash_state) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + INSERT INTO assets + (id, filename, subfolder, filepath, source, root_id, kind, ext, size, mtime, width, height, duration, enhanced_caption, job_id, stack_id, source_node_id, source_node_type, created_at, updated_at, indexed_at, content_hash, phash, hash_state) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + filename = excluded.filename, + subfolder = excluded.subfolder, + filepath = excluded.filepath, + source = excluded.source, + root_id = excluded.root_id, + kind = excluded.kind, + ext = excluded.ext, + size = excluded.size, + mtime = excluded.mtime, + width = excluded.width, + height = excluded.height, + duration = excluded.duration, + enhanced_caption = excluded.enhanced_caption, + job_id = excluded.job_id, + stack_id = excluded.stack_id, + source_node_id = excluded.source_node_id, + source_node_type = excluded.source_node_type, + created_at = excluded.created_at, + updated_at = excluded.updated_at, + indexed_at = excluded.indexed_at, + content_hash = excluded.content_hash, + phash = excluded.phash, + hash_state = excluded.hash_state """, ( asset.get("id"), @@ -1032,6 +1057,10 @@ def register_asset_routes(routes: web.RouteTableDef) -> None: asset.get("height"), asset.get("duration"), asset.get("enhanced_caption"), + asset.get("job_id"), + asset.get("stack_id"), + asset.get("source_node_id"), + asset.get("source_node_type"), asset.get("created_at"), asset.get("updated_at"), asset.get("indexed_at"), @@ -1043,9 +1072,19 @@ def register_asset_routes(routes: web.RouteTableDef) -> None: if asset_meta: await db.aexecute( """ - INSERT OR REPLACE INTO asset_metadata + INSERT INTO asset_metadata (asset_id, rating, tags, tags_text, workflow_hash, has_workflow, has_generation_data, metadata_quality, metadata_raw, metadata_text) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(asset_id) DO UPDATE SET + rating = excluded.rating, + tags = excluded.tags, + tags_text = excluded.tags_text, + workflow_hash = excluded.workflow_hash, + has_workflow = excluded.has_workflow, + has_generation_data = excluded.has_generation_data, + metadata_quality = excluded.metadata_quality, + metadata_raw = excluded.metadata_raw, + metadata_text = excluded.metadata_text """, ( asset_meta.get("asset_id"), @@ -1063,9 +1102,15 @@ def register_asset_routes(routes: web.RouteTableDef) -> None: if scan_journal: await db.aexecute( """ - INSERT OR REPLACE INTO scan_journal + INSERT INTO scan_journal (filepath, dir_path, state_hash, mtime, size, last_seen) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(filepath) DO UPDATE SET + dir_path = excluded.dir_path, + state_hash = excluded.state_hash, + mtime = excluded.mtime, + size = excluded.size, + last_seen = excluded.last_seen """, ( scan_journal.get("filepath"), @@ -1079,9 +1124,14 @@ def register_asset_routes(routes: web.RouteTableDef) -> None: if metadata_cache: await db.aexecute( """ - INSERT OR REPLACE INTO metadata_cache + INSERT INTO metadata_cache (filepath, state_hash, metadata_hash, metadata_raw, last_updated) VALUES (?, ?, ?, ?, ?) + ON CONFLICT(filepath) DO UPDATE SET + state_hash = excluded.state_hash, + metadata_hash = excluded.metadata_hash, + metadata_raw = excluded.metadata_raw, + last_updated = excluded.last_updated """, ( metadata_cache.get("filepath"), diff --git a/tests/database/test_schema_heal.py b/tests/database/test_schema_heal.py index 4689a7fe..36ce357a 100644 --- a/tests/database/test_schema_heal.py +++ b/tests/database/test_schema_heal.py @@ -74,6 +74,7 @@ async def test_schema_self_heals_missing_columns(tmp_path): for name in [ "tags", "tags_text", + "metadata_text", "workflow_hash", "has_workflow", "has_generation_data", @@ -86,7 +87,7 @@ async def test_schema_self_heals_missing_columns(tmp_path): assert (await db.aquery("SELECT source, root_id, enhanced_caption, content_hash, phash, hash_state FROM assets LIMIT 0")).ok assert (await db.aquery( """ - SELECT tags, metadata_raw, has_workflow, has_generation_data, metadata_quality + SELECT tags, tags_text, metadata_text, metadata_raw, has_workflow, has_generation_data, metadata_quality FROM asset_metadata LIMIT 0 """