AzuTear
1181 lines
38 KiB
TypeScript
1181 lines
38 KiB
TypeScript
import fs from "node:fs/promises";
|
|
import path from "node:path";
|
|
import { pngBufferToBitmap } from "./pngBitmap.js";
|
|
import { createNativeScannerResultWorkflowService } from "./nativeScannerResultWorkflowService.js";
|
|
import { loadNativeScannerDeletedResultIds } from "./nativeScannerResultTombstones.js";
|
|
import { parseArtifactCandidate } from "../../src/lib/artifactOcrParser.js";
|
|
import { evaluateStoredScanResult } from "../../src/lib/artifactEvaluation.js";
|
|
import { toStoredArtifact } from "../../src/lib/artifactStore.js";
|
|
import { matchParsedArtifactToIk, type IkArtifactCatalog } from "../../src/lib/ikArtifactMatcher.js";
|
|
import { createStoredScanResultEntry } from "../../src/lib/scanResultEntry.js";
|
|
import { shouldFlagArtifactForReview } from "../../src/lib/scannerLearning.js";
|
|
import type {
|
|
ArtifactStoreLoadResult,
|
|
ArtifactStoreSaveResult,
|
|
CaptureResult,
|
|
NativeScannerImageLoadStatus,
|
|
NativeScannerDeleteResultStatus,
|
|
NativeScannerProcessStatus,
|
|
NativeScannerProcessingProgressStatus,
|
|
NativeScannerPromotionStatus,
|
|
NativeScannerReviewArtifactInput,
|
|
NativeScannerReviewStatus,
|
|
NativeScannerResultsLoadStatus,
|
|
NativeScannerRunTiming,
|
|
NativeScannerRunTimingPatch,
|
|
ReviewSamplePayload,
|
|
} from "../../src/types/global.js";
|
|
import type { ArtifactRarity, ScanResultCategory, StoredArtifactRecord, StoredScanResultEntry } from "../../src/types/storage.js";
|
|
|
|
export type NativeCaptureJobPayload = {
|
|
sequence: number;
|
|
category?: string;
|
|
page?: number;
|
|
row?: number;
|
|
col?: number;
|
|
capturedAt?: string;
|
|
relativePath?: string;
|
|
absolutePath?: string;
|
|
/** Direct visual evidence from the native card crop; absent on legacy jobs. */
|
|
starCount?: number | null;
|
|
starConfidence?: number;
|
|
starSource?: string | null;
|
|
};
|
|
|
|
export interface NativeScannerProcessingService {
|
|
processRun(options?: {
|
|
runDir?: string;
|
|
persist?: boolean;
|
|
limit?: number;
|
|
stream?: boolean;
|
|
expectedTotal?: number;
|
|
}): Promise<NativeScannerProcessStatus>;
|
|
loadResults(options?: { runDir?: string; limit?: number; afterSequence?: number }): Promise<NativeScannerResultsLoadStatus>;
|
|
promoteResults(options: { runDir?: string; resultIds: string[] }): Promise<NativeScannerPromotionStatus>;
|
|
reviewResult(options: {
|
|
runDir?: string;
|
|
resultId: string;
|
|
action: "approve" | "reject";
|
|
artifact?: NativeScannerReviewArtifactInput;
|
|
note?: string;
|
|
}): Promise<NativeScannerReviewStatus>;
|
|
deleteResult(options: {
|
|
runDir?: string;
|
|
resultId: string;
|
|
removeLinkedStoreRecord?: boolean;
|
|
}): Promise<NativeScannerDeleteResultStatus>;
|
|
loadImage(options: { runDir?: string; imagePath: string }): Promise<NativeScannerImageLoadStatus>;
|
|
}
|
|
|
|
interface NativeScannerProcessingServiceDependencies {
|
|
resolveRunDir(runDir?: string): string;
|
|
buildCaptureResult(imagePath: string, job: NativeCaptureJobPayload): Promise<CaptureResult>;
|
|
loadArtifacts?: () => Promise<ArtifactStoreLoadResult>;
|
|
saveArtifacts(records: StoredArtifactRecord[]): Promise<Pick<ArtifactStoreSaveResult, "added" | "updated"> & Partial<ArtifactStoreSaveResult>>;
|
|
removeArtifacts?: (ids: string[]) => Promise<{ ok: boolean; removed: number }>;
|
|
saveReviewSample?: (sample: ReviewSamplePayload) => Promise<{ ok: boolean }>;
|
|
loadIkArtifactCatalog?: () => Promise<IkArtifactCatalog | null>;
|
|
/** Durable timing is observational only and must never block OCR/parse work. */
|
|
recordRunTiming?: (runDir: string, patch: NativeScannerRunTimingPatch) => Promise<NativeScannerRunTiming>;
|
|
loadRunTiming?: (runDir: string) => Promise<NativeScannerRunTiming | null>;
|
|
onProgress?: (progress: NativeScannerProcessingProgressStatus) => void;
|
|
onResult?: (result: StoredScanResultEntry) => void;
|
|
}
|
|
|
|
type ProcessedNativeJob = {
|
|
result: NativeScannerProcessStatus["results"][number];
|
|
scanResult: StoredScanResultEntry;
|
|
storedRecord: StoredArtifactRecord | null;
|
|
};
|
|
|
|
const POST_CAPTURE_QUEUE_CONCURRENCY = 4;
|
|
const STREAM_QUEUE_HIGH_WATERMARK = 8;
|
|
const STREAM_READ_CHUNK_BYTES = 64 * 1024;
|
|
const STREAM_POLL_INTERVAL_MS = 40;
|
|
const STREAM_TERMINAL_DRAIN_GRACE_MS = 2_000;
|
|
const MAX_CACHED_LIVE_RUNS = 8;
|
|
|
|
type ProcessRunOptions = {
|
|
runDir?: string;
|
|
persist?: boolean;
|
|
limit?: number;
|
|
stream?: boolean;
|
|
expectedTotal?: number;
|
|
};
|
|
|
|
type NativeCaptureProducerStatus = {
|
|
running: boolean;
|
|
status: string;
|
|
target: number;
|
|
captured: number;
|
|
};
|
|
|
|
const activeProcessingRuns = new Map<string, Promise<NativeScannerProcessStatus>>();
|
|
const liveResultsByRunDir = new Map<string, StoredScanResultEntry[]>();
|
|
|
|
export function createNativeScannerProcessingService(
|
|
deps: NativeScannerProcessingServiceDependencies,
|
|
): NativeScannerProcessingService {
|
|
const resultWorkflows = createNativeScannerResultWorkflowService({
|
|
...deps,
|
|
isProcessingRunActive: (runDir) => activeProcessingRuns.has(runDir),
|
|
});
|
|
return {
|
|
processRun(options: ProcessRunOptions = {}) {
|
|
const runDir = deps.resolveRunDir(options.runDir);
|
|
if (!runDir) {
|
|
return Promise.resolve(emptyProcessStatus("No native scanner runDir available."));
|
|
}
|
|
const active = activeProcessingRuns.get(runDir);
|
|
if (active) return active;
|
|
|
|
replaceCachedLiveResults(runDir, []);
|
|
const runPromise = processNativeScannerRun({ deps, options, runDir })
|
|
.finally(() => {
|
|
if (activeProcessingRuns.get(runDir) === runPromise) {
|
|
activeProcessingRuns.delete(runDir);
|
|
}
|
|
});
|
|
activeProcessingRuns.set(runDir, runPromise);
|
|
return runPromise;
|
|
},
|
|
|
|
async loadResults(options = {}) {
|
|
const runDir = deps.resolveRunDir(options.runDir);
|
|
if (!runDir) {
|
|
return emptyResultsStatus("No native scanner runDir available.");
|
|
}
|
|
|
|
const resultsPath = path.join(runDir, "scan-results.json");
|
|
try {
|
|
const cached = liveResultsByRunDir.get(runDir);
|
|
const rawResults = cached
|
|
? [...cached]
|
|
: await readStoredScanResults(resultsPath);
|
|
const deletedIds = await loadNativeScannerDeletedResultIds(runDir);
|
|
const results = rawResults.filter((entry) => !deletedIds.has(entry.id));
|
|
const afterSequence = Number.isFinite(options.afterSequence)
|
|
? Math.max(0, Math.round(options.afterSequence ?? 0))
|
|
: null;
|
|
const candidates = afterSequence === null
|
|
? results
|
|
: results.filter((entry) => entry.sequence > afterSequence);
|
|
const limit = Number.isFinite(options.limit)
|
|
? Math.max(1, Math.min(candidates.length, Math.round(options.limit ?? candidates.length)))
|
|
: candidates.length;
|
|
const selectedResults = afterSequence === null
|
|
? candidates.slice(-limit)
|
|
: candidates.slice(0, limit);
|
|
const timing = await recordResultReconciliationIfComplete(deps, runDir, results);
|
|
return {
|
|
ok: true,
|
|
runDir,
|
|
path: resultsPath,
|
|
total: results.length,
|
|
results: selectedResults,
|
|
...(timing ? { timing } : {}),
|
|
};
|
|
} catch (error) {
|
|
return {
|
|
...emptyResultsStatus(error instanceof Error ? error.message : String(error)),
|
|
runDir,
|
|
path: resultsPath,
|
|
};
|
|
}
|
|
},
|
|
|
|
async promoteResults(options) {
|
|
const status = await resultWorkflows.promoteResults(options);
|
|
await refreshCachedResultsAfterMutation(deps, options.runDir);
|
|
return status;
|
|
},
|
|
async reviewResult(options) {
|
|
const status = await resultWorkflows.reviewResult(options);
|
|
await refreshCachedResultsAfterMutation(deps, options.runDir);
|
|
return status;
|
|
},
|
|
async deleteResult(options) {
|
|
const status = await resultWorkflows.deleteResult(options);
|
|
await refreshCachedResultsAfterMutation(deps, options.runDir);
|
|
return status;
|
|
},
|
|
|
|
async loadImage(options) {
|
|
const runDir = deps.resolveRunDir(options.runDir);
|
|
if (!runDir) {
|
|
return emptyImageStatus("No native scanner runDir available.");
|
|
}
|
|
|
|
const resolvedRunDir = path.resolve(runDir);
|
|
const imagePath = path.isAbsolute(options.imagePath)
|
|
? path.resolve(options.imagePath)
|
|
: path.resolve(resolvedRunDir, options.imagePath);
|
|
if (!isPathInside(resolvedRunDir, imagePath)) {
|
|
return { ...emptyImageStatus("Image path is outside the native scanner run directory."), runDir: resolvedRunDir, path: imagePath };
|
|
}
|
|
if (!/\.png$/i.test(imagePath)) {
|
|
return { ...emptyImageStatus("Native scanner previews must be PNG files."), runDir: resolvedRunDir, path: imagePath };
|
|
}
|
|
|
|
try {
|
|
const stat = await fs.stat(imagePath);
|
|
if (stat.size > 20 * 1024 * 1024) {
|
|
return { ...emptyImageStatus("Native scanner preview is too large."), runDir: resolvedRunDir, path: imagePath };
|
|
}
|
|
const buffer = await fs.readFile(imagePath);
|
|
const bitmap = pngBufferToBitmap(buffer);
|
|
return {
|
|
ok: true,
|
|
runDir: resolvedRunDir,
|
|
path: imagePath,
|
|
dataUrl: `data:image/png;base64,${buffer.toString("base64")}`,
|
|
width: bitmap.width,
|
|
height: bitmap.height,
|
|
};
|
|
} catch (error) {
|
|
return {
|
|
...emptyImageStatus(error instanceof Error ? error.message : String(error)),
|
|
runDir: resolvedRunDir,
|
|
path: imagePath,
|
|
};
|
|
}
|
|
},
|
|
};
|
|
}
|
|
|
|
async function processNativeScannerRun({
|
|
deps,
|
|
options,
|
|
runDir,
|
|
}: {
|
|
deps: NativeScannerProcessingServiceDependencies;
|
|
options: ProcessRunOptions;
|
|
runDir: string;
|
|
}) {
|
|
await updateRunTiming(deps, runDir, { processingStartedAt: new Date().toISOString() });
|
|
return options.stream
|
|
? processStreamingNativeScannerRun({ deps, options, runDir })
|
|
: processBatchNativeScannerRun({ deps, options, runDir });
|
|
}
|
|
|
|
async function processBatchNativeScannerRun({
|
|
deps,
|
|
options,
|
|
runDir,
|
|
}: {
|
|
deps: NativeScannerProcessingServiceDependencies;
|
|
options: ProcessRunOptions;
|
|
runDir: string;
|
|
}): Promise<NativeScannerProcessStatus> {
|
|
const started = Date.now();
|
|
const { jobsPath, jobs } = await readNativeCaptureJobs(runDir);
|
|
const requestedLimit = Number.isFinite(options.limit)
|
|
? Math.max(1, Math.round(options.limit ?? jobs.length))
|
|
: jobs.length;
|
|
const selectedJobs = jobs.slice(0, Math.min(jobs.length, requestedLimit));
|
|
const runId = path.basename(runDir);
|
|
const ikCatalog = deps.loadIkArtifactCatalog
|
|
? await deps.loadIkArtifactCatalog().catch(() => null)
|
|
: null;
|
|
|
|
let processedCount = 0;
|
|
let parsedCount = 0;
|
|
let reviewCount = 0;
|
|
let errorCount = 0;
|
|
const publishProgress = (running: boolean, stored = 0) => safePublishProgress(deps, {
|
|
running,
|
|
runDir,
|
|
total: selectedJobs.length,
|
|
processed: processedCount,
|
|
parsed: parsedCount,
|
|
review: reviewCount,
|
|
stored,
|
|
errors: errorCount,
|
|
elapsedMs: Date.now() - started,
|
|
});
|
|
publishProgress(true);
|
|
|
|
const processedJobs = await mapWithConcurrency(
|
|
selectedJobs,
|
|
POST_CAPTURE_QUEUE_CONCURRENCY,
|
|
async (job) => {
|
|
const processedJob = await processNativeCaptureJob({
|
|
deps,
|
|
ikCatalog,
|
|
job,
|
|
persist: Boolean(options.persist),
|
|
runDir,
|
|
runId,
|
|
});
|
|
processedCount += 1;
|
|
if (processedJob.result.parsed) parsedCount += 1;
|
|
if (processedJob.result.needsReview) reviewCount += 1;
|
|
if (processedJob.result.error) errorCount += 1;
|
|
publishProgress(true);
|
|
return processedJob;
|
|
},
|
|
);
|
|
const scanResults = processedJobs.map((entry) => entry.scanResult);
|
|
replaceCachedLiveResults(runDir, scanResults);
|
|
for (const result of scanResults) safePublishResult(deps, result);
|
|
return finalizeProcessedRun({
|
|
deps,
|
|
jobsPath,
|
|
ok: true,
|
|
options,
|
|
processedJobs,
|
|
runDir,
|
|
started,
|
|
});
|
|
}
|
|
|
|
async function processStreamingNativeScannerRun({
|
|
deps,
|
|
options,
|
|
runDir,
|
|
}: {
|
|
deps: NativeScannerProcessingServiceDependencies;
|
|
options: ProcessRunOptions;
|
|
runDir: string;
|
|
}): Promise<NativeScannerProcessStatus> {
|
|
const started = Date.now();
|
|
const runId = path.basename(runDir);
|
|
const jobsPath = path.join(runDir, "capture-jobs.jsonl");
|
|
const reportPath = path.join(runDir, "processing-report.json");
|
|
const scanResultsPath = path.join(runDir, "scan-results.json");
|
|
const streamLimit = Number.isFinite(options.limit)
|
|
? Math.max(1, Math.round(options.limit ?? 1))
|
|
: Number.POSITIVE_INFINITY;
|
|
const configuredTotal = Number.isFinite(options.expectedTotal)
|
|
? Math.max(0, Math.round(options.expectedTotal ?? 0))
|
|
: Number.isFinite(options.limit)
|
|
? streamLimit
|
|
: 0;
|
|
const ikCatalog = deps.loadIkArtifactCatalog
|
|
? await deps.loadIkArtifactCatalog().catch(() => null)
|
|
: null;
|
|
const tailState: NativeCaptureJobTailState = { offset: 0 };
|
|
const pendingJobs: NativeCaptureJobPayload[] = [];
|
|
const orderedSequences: number[] = [];
|
|
const seenJobsBySequence = new Map<number, string>();
|
|
const activeJobs = new Map<number, Promise<void>>();
|
|
const completedBySequence = new Map<number, ProcessedNativeJob>();
|
|
const publishedJobs: ProcessedNativeJob[] = [];
|
|
let nextPublishIndex = 0;
|
|
let processedCount = 0;
|
|
let parsedCount = 0;
|
|
let reviewCount = 0;
|
|
let errorCount = 0;
|
|
let producerStatus: NativeCaptureProducerStatus | null = null;
|
|
let terminalObservedAt = 0;
|
|
let integrityError = "";
|
|
let snapshotWriteQueue = Promise.resolve();
|
|
|
|
const progressTotal = () => {
|
|
if (producerStatus && !producerStatus.running) {
|
|
return Math.min(producerStatus.captured, streamLimit);
|
|
}
|
|
return configuredTotal
|
|
|| producerStatus?.target
|
|
|| Math.min(orderedSequences.length, streamLimit);
|
|
};
|
|
const publishProgress = (running: boolean, stored = 0) => safePublishProgress(deps, {
|
|
running,
|
|
runDir,
|
|
total: progressTotal(),
|
|
processed: processedCount,
|
|
parsed: parsedCount,
|
|
review: reviewCount,
|
|
stored,
|
|
errors: errorCount + (integrityError ? 1 : 0),
|
|
elapsedMs: Date.now() - started,
|
|
...(integrityError ? { error: integrityError } : {}),
|
|
});
|
|
const queueDurableSnapshot = () => {
|
|
const snapshot = [...publishedJobs];
|
|
const status = buildProcessStatus({
|
|
jobsPath,
|
|
ok: true,
|
|
options,
|
|
processedJobs: snapshot,
|
|
runDir,
|
|
started,
|
|
stored: 0,
|
|
});
|
|
snapshotWriteQueue = snapshotWriteQueue
|
|
.catch(() => undefined)
|
|
.then(() => writeProcessingOutputs({ reportPath, scanResultsPath, status, processedJobs: snapshot }));
|
|
};
|
|
const flushOrderedResults = () => {
|
|
let published = false;
|
|
while (nextPublishIndex < orderedSequences.length) {
|
|
const sequence = orderedSequences[nextPublishIndex];
|
|
const completed = completedBySequence.get(sequence);
|
|
if (!completed) break;
|
|
completedBySequence.delete(sequence);
|
|
nextPublishIndex += 1;
|
|
publishedJobs.push(completed);
|
|
appendCachedLiveResult(runDir, completed.scanResult);
|
|
safePublishResult(deps, completed.scanResult);
|
|
published = true;
|
|
}
|
|
if (published && (publishedJobs.length === 1 || publishedJobs.length % 8 === 0)) {
|
|
queueDurableSnapshot();
|
|
}
|
|
};
|
|
const scheduleAvailableJobs = () => {
|
|
while (activeJobs.size < POST_CAPTURE_QUEUE_CONCURRENCY && pendingJobs.length > 0) {
|
|
const job = pendingJobs.shift();
|
|
if (!job) break;
|
|
const task = processNativeCaptureJob({
|
|
deps,
|
|
ikCatalog,
|
|
job,
|
|
persist: Boolean(options.persist),
|
|
runDir,
|
|
runId,
|
|
}).then((processedJob) => {
|
|
processedCount += 1;
|
|
if (processedJob.result.parsed) parsedCount += 1;
|
|
if (processedJob.result.needsReview) reviewCount += 1;
|
|
if (processedJob.result.error) errorCount += 1;
|
|
completedBySequence.set(job.sequence, processedJob);
|
|
flushOrderedResults();
|
|
publishProgress(true);
|
|
}).finally(() => {
|
|
activeJobs.delete(job.sequence);
|
|
});
|
|
activeJobs.set(job.sequence, task);
|
|
}
|
|
};
|
|
|
|
publishProgress(true);
|
|
while (true) {
|
|
const queueCapacity = Math.max(
|
|
0,
|
|
STREAM_QUEUE_HIGH_WATERMARK - pendingJobs.length - activeJobs.size,
|
|
);
|
|
const appendedJobs = queueCapacity > 0
|
|
? await readAppendedNativeCaptureJobs(jobsPath, tailState, queueCapacity)
|
|
: [];
|
|
for (const job of appendedJobs) {
|
|
const identity = nativeCaptureJobIdentity(job);
|
|
const existingIdentity = seenJobsBySequence.get(job.sequence);
|
|
if (existingIdentity) {
|
|
if (existingIdentity !== identity && !integrityError) {
|
|
integrityError = `Conflicting capture jobs use sequence ${job.sequence}.`;
|
|
}
|
|
continue;
|
|
}
|
|
if (orderedSequences.length >= streamLimit) continue;
|
|
seenJobsBySequence.set(job.sequence, identity);
|
|
orderedSequences.push(job.sequence);
|
|
pendingJobs.push(job);
|
|
}
|
|
scheduleAvailableJobs();
|
|
|
|
const nextProducerStatus = await readNativeCaptureProducerStatus(runDir);
|
|
if (nextProducerStatus) producerStatus = nextProducerStatus;
|
|
if (producerStatus && !producerStatus.running && terminalObservedAt === 0) {
|
|
terminalObservedAt = Date.now();
|
|
}
|
|
|
|
const expectedCapturedJobs = producerStatus && !producerStatus.running
|
|
? Math.min(producerStatus.captured, streamLimit)
|
|
: null;
|
|
const producerDrained = expectedCapturedJobs !== null
|
|
&& orderedSequences.length >= expectedCapturedJobs;
|
|
if (producerDrained && pendingJobs.length === 0 && activeJobs.size === 0) break;
|
|
if (
|
|
expectedCapturedJobs !== null
|
|
&& terminalObservedAt > 0
|
|
&& Date.now() - terminalObservedAt >= STREAM_TERMINAL_DRAIN_GRACE_MS
|
|
&& pendingJobs.length === 0
|
|
&& activeJobs.size === 0
|
|
) {
|
|
integrityError = `Capture job stream ended with ${orderedSequences.length}/${expectedCapturedJobs} jobs.`;
|
|
break;
|
|
}
|
|
|
|
if (activeJobs.size > 0) {
|
|
await Promise.race([
|
|
Promise.race(activeJobs.values()),
|
|
delay(STREAM_POLL_INTERVAL_MS),
|
|
]);
|
|
} else {
|
|
await delay(STREAM_POLL_INTERVAL_MS);
|
|
}
|
|
}
|
|
|
|
if (activeJobs.size > 0) await Promise.all(activeJobs.values());
|
|
flushOrderedResults();
|
|
await snapshotWriteQueue.catch(() => undefined);
|
|
const status = await finalizeProcessedRun({
|
|
deps,
|
|
jobsPath,
|
|
ok: !integrityError,
|
|
options,
|
|
processedJobs: publishedJobs,
|
|
runDir,
|
|
started,
|
|
error: integrityError || undefined,
|
|
});
|
|
return status;
|
|
}
|
|
|
|
async function finalizeProcessedRun({
|
|
deps,
|
|
error,
|
|
jobsPath,
|
|
ok,
|
|
options,
|
|
processedJobs,
|
|
runDir,
|
|
started,
|
|
}: {
|
|
deps: NativeScannerProcessingServiceDependencies;
|
|
error?: string;
|
|
jobsPath: string;
|
|
ok: boolean;
|
|
options: ProcessRunOptions;
|
|
processedJobs: ProcessedNativeJob[];
|
|
runDir: string;
|
|
started: number;
|
|
}) {
|
|
const recordsToPersist = processedJobs
|
|
.map((entry) => entry.storedRecord)
|
|
.filter((record): record is StoredArtifactRecord => Boolean(record));
|
|
let stored = 0;
|
|
if (options.persist && recordsToPersist.length > 0) {
|
|
const saved = await deps.saveArtifacts(recordsToPersist);
|
|
stored = saved.added + saved.updated;
|
|
}
|
|
const processingTiming = await updateRunTiming(deps, runDir, {
|
|
processingCompletedAt: new Date().toISOString(),
|
|
});
|
|
let status = buildProcessStatus({
|
|
error,
|
|
jobsPath,
|
|
ok,
|
|
options,
|
|
processedJobs,
|
|
runDir,
|
|
started,
|
|
stored,
|
|
timing: processingTiming,
|
|
});
|
|
// The complete results file is the durable hand-off boundary. Keep its
|
|
// timestamp distinct from a later renderer/service reconciliation read.
|
|
await writeJsonAtomic(status.scanResultsPath, processedJobs.map((entry) => entry.scanResult));
|
|
const durableTiming = await updateRunTiming(deps, runDir, {
|
|
resultsDurableAt: new Date().toISOString(),
|
|
});
|
|
status = buildProcessStatus({
|
|
error,
|
|
jobsPath,
|
|
ok,
|
|
options,
|
|
processedJobs,
|
|
runDir,
|
|
started,
|
|
stored,
|
|
timing: durableTiming ?? processingTiming,
|
|
});
|
|
await writeJsonAtomic(status.reportPath, status);
|
|
replaceCachedLiveResults(runDir, processedJobs.map((entry) => entry.scanResult));
|
|
safePublishProgress(deps, {
|
|
running: false,
|
|
runDir,
|
|
total: processedJobs.length,
|
|
processed: status.processed,
|
|
parsed: status.parsed,
|
|
review: status.review,
|
|
stored: status.stored,
|
|
errors: status.errors,
|
|
elapsedMs: status.elapsedMs,
|
|
...(status.error ? { error: status.error } : {}),
|
|
});
|
|
return status;
|
|
}
|
|
|
|
function buildProcessStatus({
|
|
error,
|
|
jobsPath,
|
|
ok,
|
|
options,
|
|
processedJobs,
|
|
runDir,
|
|
started,
|
|
stored,
|
|
timing,
|
|
}: {
|
|
error?: string;
|
|
jobsPath: string;
|
|
ok: boolean;
|
|
options: ProcessRunOptions;
|
|
processedJobs: ProcessedNativeJob[];
|
|
runDir: string;
|
|
started: number;
|
|
stored: number;
|
|
timing?: NativeScannerRunTiming;
|
|
}): NativeScannerProcessStatus {
|
|
const results = processedJobs.map((entry) => entry.result);
|
|
return {
|
|
ok,
|
|
runDir,
|
|
jobsPath,
|
|
reportPath: path.join(runDir, "processing-report.json"),
|
|
scanResultsPath: path.join(runDir, "scan-results.json"),
|
|
processed: results.length,
|
|
parsed: results.filter((result) => result.parsed).length,
|
|
review: results.filter((result) => result.needsReview).length,
|
|
stored,
|
|
errors: results.filter((result) => result.error).length + (error ? 1 : 0),
|
|
elapsedMs: Date.now() - started,
|
|
queueConcurrency: Math.min(POST_CAPTURE_QUEUE_CONCURRENCY, processedJobs.length),
|
|
persisted: Boolean(options.persist),
|
|
results,
|
|
...(timing ? { timing } : {}),
|
|
...(error ? { error } : {}),
|
|
};
|
|
}
|
|
|
|
async function writeProcessingOutputs({
|
|
processedJobs,
|
|
reportPath,
|
|
scanResultsPath,
|
|
status,
|
|
}: {
|
|
processedJobs: ProcessedNativeJob[];
|
|
reportPath: string;
|
|
scanResultsPath: string;
|
|
status: NativeScannerProcessStatus;
|
|
}) {
|
|
await writeJsonAtomic(scanResultsPath, processedJobs.map((entry) => entry.scanResult));
|
|
await writeJsonAtomic(reportPath, status);
|
|
}
|
|
|
|
async function writeJsonAtomic(filePath: string, payload: unknown) {
|
|
const temporaryPath = `${filePath}.tmp-${process.pid}`;
|
|
await fs.writeFile(temporaryPath, JSON.stringify(payload, null, 2), "utf8");
|
|
await fs.rename(temporaryPath, filePath);
|
|
}
|
|
|
|
type NativeCaptureJobTailState = {
|
|
offset: number;
|
|
};
|
|
|
|
async function readAppendedNativeCaptureJobs(
|
|
jobsPath: string,
|
|
state: NativeCaptureJobTailState,
|
|
maxJobs: number,
|
|
): Promise<NativeCaptureJobPayload[]> {
|
|
let handle: Awaited<ReturnType<typeof fs.open>> | null = null;
|
|
try {
|
|
handle = await fs.open(jobsPath, "r");
|
|
const stat = await handle.stat();
|
|
if (stat.size < state.offset) {
|
|
state.offset = 0;
|
|
}
|
|
const length = Math.min(stat.size - state.offset, STREAM_READ_CHUNK_BYTES);
|
|
if (length <= 0) return [];
|
|
const chunk = Buffer.alloc(length);
|
|
const { bytesRead } = await handle.read(chunk, 0, length, state.offset);
|
|
const jobs: NativeCaptureJobPayload[] = [];
|
|
let lineStart = 0;
|
|
let consumedBytes = 0;
|
|
for (let index = 0; index < bytesRead; index += 1) {
|
|
if (chunk[index] !== 0x0a) continue;
|
|
const line = chunk.subarray(lineStart, index).toString("utf8").trim();
|
|
lineStart = index + 1;
|
|
consumedBytes = lineStart;
|
|
if (!line) continue;
|
|
const job = JSON.parse(line) as NativeCaptureJobPayload;
|
|
if (Number.isFinite(job.sequence)) jobs.push(job);
|
|
if (jobs.length >= maxJobs) break;
|
|
}
|
|
state.offset += consumedBytes;
|
|
return jobs;
|
|
} catch (error) {
|
|
if (isMissingFileError(error)) return [];
|
|
throw error;
|
|
} finally {
|
|
await handle?.close().catch(() => undefined);
|
|
}
|
|
}
|
|
|
|
async function readNativeCaptureProducerStatus(runDir: string): Promise<NativeCaptureProducerStatus | null> {
|
|
try {
|
|
const raw = JSON.parse(await fs.readFile(path.join(runDir, "status.json"), "utf8"));
|
|
const scanner = raw?.scanner ?? raw;
|
|
if (!scanner || typeof scanner !== "object") return null;
|
|
return {
|
|
running: Boolean(scanner.running),
|
|
status: typeof scanner.status === "string" ? scanner.status : "unknown",
|
|
target: Math.max(0, Math.round(Number(scanner.target) || 0)),
|
|
captured: Math.max(0, Math.round(Number(scanner.captured) || 0)),
|
|
};
|
|
} catch (error) {
|
|
if (isMissingFileError(error) || error instanceof SyntaxError) return null;
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async function readStoredScanResults(resultsPath: string) {
|
|
const raw = await fs.readFile(resultsPath, "utf8");
|
|
const parsed = JSON.parse(raw);
|
|
return Array.isArray(parsed)
|
|
? parsed.filter(isStoredScanResultEntry).map(evaluateStoredScanResult)
|
|
: [];
|
|
}
|
|
|
|
async function refreshCachedResultsAfterMutation(
|
|
deps: NativeScannerProcessingServiceDependencies,
|
|
requestedRunDir?: string,
|
|
) {
|
|
const runDir = deps.resolveRunDir(requestedRunDir);
|
|
if (!runDir) return;
|
|
try {
|
|
const [results, deletedIds] = await Promise.all([
|
|
readStoredScanResults(path.join(runDir, "scan-results.json")),
|
|
loadNativeScannerDeletedResultIds(runDir),
|
|
]);
|
|
replaceCachedLiveResults(runDir, results.filter((entry) => !deletedIds.has(entry.id)));
|
|
} catch {
|
|
// Mutation status remains authoritative; a later disk load can retry.
|
|
liveResultsByRunDir.delete(runDir);
|
|
}
|
|
}
|
|
|
|
async function recordResultReconciliationIfComplete(
|
|
deps: NativeScannerProcessingServiceDependencies,
|
|
runDir: string,
|
|
results: StoredScanResultEntry[],
|
|
) {
|
|
const currentTiming = await loadRunTiming(deps, runDir);
|
|
// Legacy runs intentionally stay untouched. A reconciliation timestamp is
|
|
// meaningful only for a run that already owns the new capture/processing
|
|
// timeline and whose complete result file was durably written by this path.
|
|
if (!currentTiming?.resultsDurableAt || currentTiming.resultsReconciledAt) return currentTiming ?? undefined;
|
|
|
|
const producer = await readNativeCaptureProducerStatus(runDir).catch(() => null);
|
|
const terminal = producer
|
|
&& !producer.running
|
|
&& (producer.status === "done" || producer.status === "completed")
|
|
&& producer.captured > 0
|
|
&& results.length === producer.captured;
|
|
if (!terminal) return currentTiming;
|
|
return updateRunTiming(deps, runDir, { resultsReconciledAt: new Date().toISOString() }) ?? currentTiming;
|
|
}
|
|
|
|
async function updateRunTiming(
|
|
deps: NativeScannerProcessingServiceDependencies,
|
|
runDir: string,
|
|
patch: NativeScannerRunTimingPatch,
|
|
) {
|
|
if (!deps.recordRunTiming || !runDir) return undefined;
|
|
try {
|
|
return await deps.recordRunTiming(runDir, patch);
|
|
} catch {
|
|
// Timing is evidence, never a dependency of scanner correctness. A later
|
|
// caller can still load the durable results and report the missing marker.
|
|
return undefined;
|
|
}
|
|
}
|
|
|
|
async function loadRunTiming(
|
|
deps: NativeScannerProcessingServiceDependencies,
|
|
runDir: string,
|
|
) {
|
|
if (!deps.loadRunTiming || !runDir) return null;
|
|
try {
|
|
return await deps.loadRunTiming(runDir);
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function replaceCachedLiveResults(runDir: string, results: StoredScanResultEntry[]) {
|
|
liveResultsByRunDir.delete(runDir);
|
|
liveResultsByRunDir.set(runDir, [...results]);
|
|
while (liveResultsByRunDir.size > MAX_CACHED_LIVE_RUNS) {
|
|
const oldest = liveResultsByRunDir.keys().next().value;
|
|
if (typeof oldest !== "string") break;
|
|
liveResultsByRunDir.delete(oldest);
|
|
}
|
|
}
|
|
|
|
function appendCachedLiveResult(runDir: string, result: StoredScanResultEntry) {
|
|
const current = liveResultsByRunDir.get(runDir) ?? [];
|
|
if (current.some((entry) => entry.id === result.id)) return;
|
|
current.push(result);
|
|
liveResultsByRunDir.set(runDir, current);
|
|
}
|
|
|
|
function safePublishProgress(
|
|
deps: NativeScannerProcessingServiceDependencies,
|
|
progress: NativeScannerProcessingProgressStatus,
|
|
) {
|
|
try {
|
|
deps.onProgress?.(progress);
|
|
} catch {
|
|
// UI progress must never abort durable OCR/parse processing.
|
|
}
|
|
}
|
|
|
|
function safePublishResult(deps: NativeScannerProcessingServiceDependencies, result: StoredScanResultEntry) {
|
|
try {
|
|
deps.onResult?.(result);
|
|
} catch {
|
|
// UI delivery is best effort; loadResults can reconcile from memory or disk.
|
|
}
|
|
}
|
|
|
|
function nativeCaptureJobIdentity(job: NativeCaptureJobPayload) {
|
|
return JSON.stringify({
|
|
sequence: job.sequence,
|
|
category: job.category ?? "",
|
|
page: job.page ?? null,
|
|
row: job.row ?? null,
|
|
col: job.col ?? null,
|
|
relativePath: job.relativePath ?? "",
|
|
absolutePath: job.absolutePath ?? "",
|
|
starCount: job.starCount ?? null,
|
|
starConfidence: job.starConfidence ?? null,
|
|
starSource: job.starSource ?? "",
|
|
});
|
|
}
|
|
|
|
function isMissingFileError(error: unknown) {
|
|
return Boolean(error && typeof error === "object" && "code" in error && error.code === "ENOENT");
|
|
}
|
|
|
|
function delay(ms: number) {
|
|
return new Promise((resolve) => setTimeout(resolve, ms));
|
|
}
|
|
|
|
export function nativeScannerProcessStats(status: NativeScannerProcessStatus): Record<string, number> {
|
|
return {
|
|
processed: status.processed,
|
|
parsed: status.parsed,
|
|
review: status.review,
|
|
stored: status.stored,
|
|
errors: status.errors,
|
|
elapsedMs: status.elapsedMs,
|
|
queueConcurrency: status.queueConcurrency,
|
|
};
|
|
}
|
|
|
|
async function processNativeCaptureJob({
|
|
deps,
|
|
ikCatalog,
|
|
job,
|
|
persist,
|
|
runDir,
|
|
runId,
|
|
}: {
|
|
deps: NativeScannerProcessingServiceDependencies;
|
|
ikCatalog: IkArtifactCatalog | null;
|
|
job: NativeCaptureJobPayload;
|
|
persist: boolean;
|
|
runDir: string;
|
|
runId: string;
|
|
}): Promise<ProcessedNativeJob> {
|
|
const category = nativeJobCategory(job.category);
|
|
const imagePath = job.absolutePath || (job.relativePath ? path.join(runDir, job.relativePath) : "");
|
|
if (!category.artifactProcessingSupported) {
|
|
return reviewJobResult({
|
|
category: category.scanResultCategory,
|
|
error: `Native post-capture processing for category '${category.nativeCategory}' is not implemented yet; IK catalog is available only.`,
|
|
imagePath,
|
|
job,
|
|
runId,
|
|
});
|
|
}
|
|
if (!imagePath || !(await fileExists(imagePath))) {
|
|
const error = "Card crop image missing.";
|
|
return reviewJobResult({ category: category.scanResultCategory, error, imagePath, job, runId });
|
|
}
|
|
|
|
try {
|
|
const capture = await deps.buildCaptureResult(imagePath, job);
|
|
const parsed = parseArtifactCandidate(capture);
|
|
const ikMatch = parsed && ikCatalog ? matchParsedArtifactToIk(parsed, ikCatalog) : null;
|
|
const rarityEvidence = nativeJobRarityEvidence(job);
|
|
const notes = [...new Set([
|
|
...(parsed?.notes ?? []),
|
|
...(ikMatch?.notes ?? []),
|
|
...(rarityEvidence.uncertain ? ["Artifact star row could not be confirmed; keep this result in Review."] : []),
|
|
])];
|
|
const needsReview = shouldFlagArtifactForReview(parsed)
|
|
|| Boolean(ikMatch && !ikMatch.matched)
|
|
|| rarityEvidence.uncertain;
|
|
const canPersist = parsedArtifactCanPersist(parsed, needsReview);
|
|
const baseScanResult = createStoredScanResultEntry({
|
|
runId,
|
|
sequence: job.sequence,
|
|
page: job.page,
|
|
row: job.row,
|
|
col: job.col,
|
|
category: category.scanResultCategory,
|
|
source: "native-ik-scan",
|
|
imagePath,
|
|
parsed,
|
|
needsReview,
|
|
confidence: parsed?.confidence ?? 0,
|
|
capturedAt: job.capturedAt,
|
|
persistedArtifact: false,
|
|
notes,
|
|
locked: capture.locked,
|
|
ikMatch: ikMatch ?? undefined,
|
|
rarity: rarityEvidence.rarity,
|
|
rarityConfidence: rarityEvidence.confidence,
|
|
raritySource: rarityEvidence.rarity ? "native" : undefined,
|
|
});
|
|
const shouldPersist = Boolean(
|
|
persist
|
|
&& parsed
|
|
&& canPersist
|
|
&& !needsReview
|
|
&& baseScanResult.valueStatus === "evaluated"
|
|
&& baseScanResult.valueEvaluation?.status === "evaluated"
|
|
&& baseScanResult.valueEvaluation.score !== null
|
|
&& Number.isFinite(baseScanResult.valueEvaluation.score),
|
|
);
|
|
const storedRecord = parsed && shouldPersist
|
|
? toStoredArtifact(parsed, "native-ik-scan", needsReview, capture.locked)
|
|
: null;
|
|
const scanResult = storedRecord
|
|
? { ...baseScanResult, artifactRecordId: storedRecord.id, persistedArtifact: true }
|
|
: baseScanResult;
|
|
return {
|
|
result: {
|
|
sequence: job.sequence,
|
|
category: category.scanResultCategory,
|
|
page: job.page,
|
|
row: job.row,
|
|
col: job.col,
|
|
imagePath,
|
|
parsed: Boolean(parsed),
|
|
artifactName: parsed?.name,
|
|
setName: parsed?.setName,
|
|
slot: parsed?.slot,
|
|
confidence: parsed?.confidence ?? 0,
|
|
needsReview,
|
|
rarity: rarityEvidence.rarity,
|
|
rarityConfidence: rarityEvidence.confidence,
|
|
persisted: shouldPersist,
|
|
ikMatch: ikMatch ?? undefined,
|
|
notes,
|
|
ocr: capture.ocr ?? [],
|
|
},
|
|
scanResult,
|
|
storedRecord,
|
|
};
|
|
} catch (error) {
|
|
const message = error instanceof Error ? error.message : String(error);
|
|
return reviewJobResult({ category: category.scanResultCategory, error: message, imagePath, job, runId });
|
|
}
|
|
}
|
|
|
|
function nativeJobRarityEvidence(job: NativeCaptureJobPayload): {
|
|
rarity?: ArtifactRarity;
|
|
confidence?: number;
|
|
/** The newer helper attempted direct visual detection but could not prove a count. */
|
|
uncertain: boolean;
|
|
} {
|
|
const hasDetectionPayload = Object.prototype.hasOwnProperty.call(job, "starCount")
|
|
|| Object.prototype.hasOwnProperty.call(job, "starConfidence")
|
|
|| Object.prototype.hasOwnProperty.call(job, "starSource");
|
|
if (!hasDetectionPayload) return { uncertain: false };
|
|
|
|
const starCount = Number(job.starCount);
|
|
const confidence = Number(job.starConfidence);
|
|
const validRarity = Number.isInteger(starCount) && starCount >= 1 && starCount <= 5;
|
|
const confident = Number.isFinite(confidence) && confidence >= 0.8 && confidence <= 1;
|
|
const hasSource = typeof job.starSource === "string" && job.starSource.trim().length > 0;
|
|
if (!validRarity || !confident || !hasSource) return { uncertain: true };
|
|
return { rarity: starCount as ArtifactRarity, confidence, uncertain: false };
|
|
}
|
|
|
|
function reviewJobResult({
|
|
category,
|
|
error,
|
|
imagePath,
|
|
job,
|
|
runId,
|
|
}: {
|
|
category?: ScanResultCategory;
|
|
error: string;
|
|
imagePath: string;
|
|
job: NativeCaptureJobPayload;
|
|
runId: string;
|
|
}): ProcessedNativeJob {
|
|
const scanResultCategory = category ?? nativeJobCategory(job.category).scanResultCategory;
|
|
return {
|
|
result: {
|
|
sequence: job.sequence,
|
|
category: scanResultCategory,
|
|
page: job.page,
|
|
row: job.row,
|
|
col: job.col,
|
|
imagePath,
|
|
parsed: false,
|
|
needsReview: true,
|
|
confidence: 0,
|
|
ocr: [],
|
|
error,
|
|
},
|
|
scanResult: createStoredScanResultEntry({
|
|
runId,
|
|
sequence: job.sequence,
|
|
page: job.page,
|
|
row: job.row,
|
|
col: job.col,
|
|
category: scanResultCategory,
|
|
source: "native-ik-scan",
|
|
imagePath,
|
|
parsed: null,
|
|
needsReview: true,
|
|
confidence: 0,
|
|
capturedAt: job.capturedAt,
|
|
error,
|
|
notes: [error],
|
|
}),
|
|
storedRecord: null,
|
|
};
|
|
}
|
|
|
|
function nativeJobCategory(category: string | undefined): {
|
|
nativeCategory: string;
|
|
scanResultCategory: ScanResultCategory;
|
|
artifactProcessingSupported: boolean;
|
|
} {
|
|
const nativeCategory = (category ?? "artifacts").trim().toLowerCase() || "artifacts";
|
|
if (nativeCategory === "artifact" || nativeCategory === "artifacts") {
|
|
return { nativeCategory, scanResultCategory: "artifact", artifactProcessingSupported: true };
|
|
}
|
|
if (nativeCategory === "weapon" || nativeCategory === "weapons") {
|
|
return { nativeCategory, scanResultCategory: "weapon", artifactProcessingSupported: false };
|
|
}
|
|
if (nativeCategory === "character" || nativeCategory === "characters") {
|
|
return { nativeCategory, scanResultCategory: "character", artifactProcessingSupported: false };
|
|
}
|
|
if (nativeCategory === "material" || nativeCategory === "materials") {
|
|
return { nativeCategory, scanResultCategory: "material", artifactProcessingSupported: false };
|
|
}
|
|
return { nativeCategory, scanResultCategory: "unknown", artifactProcessingSupported: false };
|
|
}
|
|
|
|
async function mapWithConcurrency<TInput, TOutput>(
|
|
items: readonly TInput[],
|
|
concurrency: number,
|
|
worker: (item: TInput, index: number) => Promise<TOutput>,
|
|
) {
|
|
const output = Array<TOutput>(items.length);
|
|
let nextIndex = 0;
|
|
const workerCount = Math.max(1, Math.min(items.length, Math.floor(concurrency)));
|
|
await Promise.all(Array.from({ length: workerCount }, async () => {
|
|
while (nextIndex < items.length) {
|
|
const index = nextIndex++;
|
|
output[index] = await worker(items[index], index);
|
|
}
|
|
}));
|
|
return output;
|
|
}
|
|
|
|
async function readNativeCaptureJobs(runDir: string): Promise<{ jobsPath: string; jobs: NativeCaptureJobPayload[] }> {
|
|
const jobsPath = path.join(runDir, "capture-jobs.jsonl");
|
|
const raw = await fs.readFile(jobsPath, "utf8");
|
|
const jobs = raw
|
|
.split(/\r?\n/)
|
|
.map((line) => line.trim())
|
|
.filter(Boolean)
|
|
.map((line) => JSON.parse(line) as NativeCaptureJobPayload)
|
|
.filter((job) => Number.isFinite(job.sequence));
|
|
return { jobsPath, jobs };
|
|
}
|
|
|
|
async function fileExists(filePath: string) {
|
|
try {
|
|
await fs.access(filePath);
|
|
return true;
|
|
} catch {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
function emptyProcessStatus(error: string): NativeScannerProcessStatus {
|
|
return {
|
|
ok: false,
|
|
runDir: "",
|
|
jobsPath: "",
|
|
reportPath: "",
|
|
scanResultsPath: "",
|
|
processed: 0,
|
|
parsed: 0,
|
|
review: 0,
|
|
stored: 0,
|
|
errors: 1,
|
|
elapsedMs: 0,
|
|
queueConcurrency: 0,
|
|
persisted: false,
|
|
results: [],
|
|
error,
|
|
};
|
|
}
|
|
|
|
function emptyResultsStatus(error: string): NativeScannerResultsLoadStatus {
|
|
return {
|
|
ok: false,
|
|
runDir: "",
|
|
path: "",
|
|
total: 0,
|
|
results: [],
|
|
error,
|
|
};
|
|
}
|
|
|
|
function emptyImageStatus(error: string): NativeScannerImageLoadStatus {
|
|
return {
|
|
ok: false,
|
|
runDir: "",
|
|
path: "",
|
|
dataUrl: "",
|
|
width: 0,
|
|
height: 0,
|
|
error,
|
|
};
|
|
}
|
|
|
|
function isPathInside(root: string, candidate: string) {
|
|
const relative = path.relative(root, candidate);
|
|
return relative === "" || (!relative.startsWith("..") && !path.isAbsolute(relative));
|
|
}
|
|
|
|
function isStoredScanResultEntry(value: unknown): value is StoredScanResultEntry {
|
|
if (!value || typeof value !== "object") return false;
|
|
const entry = value as Partial<StoredScanResultEntry>;
|
|
return typeof entry.id === "string"
|
|
&& typeof entry.runId === "string"
|
|
&& Number.isFinite(entry.sequence)
|
|
&& typeof entry.source === "string"
|
|
&& typeof entry.imagePath === "string"
|
|
&& typeof entry.extractionStatus === "string"
|
|
&& typeof entry.valueStatus === "string"
|
|
&& Array.isArray(entry.notes);
|
|
}
|
|
|
|
function parsedArtifactCanPersist(parsed: ReturnType<typeof parseArtifactCandidate>, needsReview: boolean) {
|
|
if (!parsed) return false;
|
|
if (parsed.name === "Unknown artifact") return false;
|
|
if (parsed.slot === "Unknown slot") return false;
|
|
if (parsed.setName === "Unknown set") return false;
|
|
if (parsed.mainStat === "Unknown main stat") return false;
|
|
if (parsed.mainValue === "?") return false;
|
|
if (parsed.substats.length === 0) return false;
|
|
if (!needsReview && parsed.confidence < 68) return false;
|
|
if (needsReview && parsed.confidence < 60) return false;
|
|
return true;
|
|
}
|