548 lines
25 KiB
TypeScript
548 lines
25 KiB
TypeScript
import { createHash, randomUUID } from "node:crypto";
|
|
import type { Readable } from "node:stream";
|
|
|
|
import type BetterSqlite3 from "better-sqlite3";
|
|
|
|
import type { CreditService } from "./credits.js";
|
|
import { GenerationSubmissionError } from "./generation-submission-errors.js";
|
|
import type { ManagedStorage, StagedManagedFile } from "./managed-storage.js";
|
|
import {
|
|
defaultCanvasState,
|
|
defaultProjectName,
|
|
historyLimit,
|
|
normalizePrompt,
|
|
projectLimit,
|
|
projectRatios,
|
|
ratioPixels,
|
|
stableJson,
|
|
type ProjectRatio,
|
|
} from "./projects.js";
|
|
import { classifyCapacity } from "./storage-policy.js";
|
|
|
|
export { GenerationSubmissionError } from "./generation-submission-errors.js";
|
|
|
|
const idempotencyPattern = /^[A-Za-z0-9_-]{32,200}$/;
|
|
const uuidPattern = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
|
|
|
|
export interface GenerationModelSnapshot {
|
|
configSetVersion: number;
|
|
configVersion: number;
|
|
contractValidationStatus: "verified" | "unverified";
|
|
creditCost: number;
|
|
enabled: boolean;
|
|
modelId: string;
|
|
promptMaxLength: number;
|
|
referenceLimits: { maxFileBytes: number; maxFiles: number; maxTotalBytes: number };
|
|
runtimeAvailability: {
|
|
availableForNewJobs: boolean;
|
|
reason: "gateway_balance_insufficient" | "gateway_contract_invalid" | "model_disabled" | null;
|
|
};
|
|
supportedRatios: readonly ProjectRatio[];
|
|
}
|
|
|
|
export interface GenerationModelCatalog {
|
|
readModel(modelId: string): GenerationModelSnapshot | undefined;
|
|
}
|
|
|
|
export class StaticGenerationModelCatalog implements GenerationModelCatalog {
|
|
private readonly models = new Map<string, GenerationModelSnapshot>();
|
|
|
|
constructor(models: GenerationModelSnapshot[]) {
|
|
for (const model of models) this.models.set(model.modelId, structuredClone(model));
|
|
}
|
|
|
|
readModel(modelId: string) {
|
|
const model = this.models.get(modelId);
|
|
return model ? structuredClone(model) : undefined;
|
|
}
|
|
|
|
replace(model: GenerationModelSnapshot) {
|
|
this.models.set(model.modelId, structuredClone(model));
|
|
}
|
|
}
|
|
|
|
export interface NewGenerationReference {
|
|
content: Readable;
|
|
fileName: string;
|
|
mimeType: "image/jpeg" | "image/png" | "image/webp";
|
|
projectedBytes: number;
|
|
}
|
|
|
|
export interface GenerationSubmissionInput {
|
|
clientSubmissionId: string;
|
|
confirmedCreditCost: number;
|
|
existingReferenceAssetIds: string[];
|
|
idempotencyKey: string;
|
|
mode: "new_project" | "existing_project";
|
|
modelConfigVersion: number;
|
|
modelId: string;
|
|
newReferences: NewGenerationReference[];
|
|
projectId?: string;
|
|
prompt: string;
|
|
ratio: ProjectRatio;
|
|
userId: string;
|
|
}
|
|
|
|
export type GenerationSubmissionFields = Omit<GenerationSubmissionInput, "newReferences">;
|
|
|
|
export interface GenerationTaskView {
|
|
confirmedCreditCost: number;
|
|
createdAt: string;
|
|
errorCategory: "upstream_timeout" | "upstream_failed" | "safety_rejected" | "model_disabled"
|
|
| "gateway_balance_insufficient" | "gateway_contract_invalid" | "reference_invalid"
|
|
| "unknown_retryable" | "unknown_non_retryable" | null;
|
|
generationId: string;
|
|
modelConfigVersion: number;
|
|
modelId: string;
|
|
projectId: string;
|
|
prompt: string;
|
|
ratio: ProjectRatio;
|
|
referenceAssetIds: string[];
|
|
referenceCount: number;
|
|
reservedCredits: number;
|
|
status: "queued" | "running" | "succeeded" | "failed" | "rejected";
|
|
updatedAt: string;
|
|
}
|
|
|
|
export type GenerationSubmissionResult = { created: boolean; task: GenerationTaskView };
|
|
|
|
export interface GenerationUploadSession {
|
|
abort(): void;
|
|
commit(): Promise<GenerationSubmissionResult>;
|
|
stageReference(reference: NewGenerationReference): Promise<void>;
|
|
}
|
|
|
|
interface GenerationRow {
|
|
confirmed_credit_cost: number;
|
|
created_at: number;
|
|
error_category: GenerationTaskView["errorCategory"];
|
|
generation_id: string;
|
|
model_config_version: number;
|
|
model_id: string;
|
|
owner_id: string;
|
|
project_id: string;
|
|
prompt: string;
|
|
ratio: ProjectRatio;
|
|
reserved_credits: number;
|
|
status: "queued" | "running" | "succeeded" | "failed" | "rejected";
|
|
updated_at: number;
|
|
}
|
|
|
|
function iso(timestamp: number) {
|
|
return new Date(timestamp).toISOString();
|
|
}
|
|
|
|
function digest(value: string) {
|
|
return createHash("sha256").update(value, "utf8").digest("hex");
|
|
}
|
|
|
|
function safeInteger(value: number) {
|
|
return Number.isSafeInteger(value) && value > 0;
|
|
}
|
|
|
|
export class GenerationSubmissionService {
|
|
readonly database: BetterSqlite3.Database;
|
|
private readonly beforeTransaction: (() => Promise<void>) | undefined;
|
|
private readonly clock: () => number;
|
|
private readonly credits: CreditService;
|
|
private readonly models: GenerationModelCatalog;
|
|
private readonly storage: ManagedStorage;
|
|
|
|
constructor(input: {
|
|
beforeTransaction?: () => Promise<void>;
|
|
clock?: () => number;
|
|
credits: CreditService;
|
|
models: GenerationModelCatalog;
|
|
storage: ManagedStorage;
|
|
}) {
|
|
this.beforeTransaction = input.beforeTransaction;
|
|
this.clock = input.clock ?? Date.now;
|
|
this.credits = input.credits;
|
|
this.database = input.credits.database;
|
|
this.models = input.models;
|
|
this.storage = input.storage;
|
|
this.migrate();
|
|
}
|
|
|
|
close() {
|
|
// The database connection is owned by CreditService.
|
|
}
|
|
|
|
readCurrentTask(userId: string) {
|
|
const row = this.database.prepare(`
|
|
SELECT * FROM generation_jobs
|
|
WHERE owner_id = ? AND status IN ('queued', 'running') AND submission_ready = 1
|
|
ORDER BY created_at DESC, generation_id DESC LIMIT 1
|
|
`).get(userId) as GenerationRow | undefined;
|
|
return row ? this.taskView(row) : undefined;
|
|
}
|
|
|
|
readTask(userId: string, generationId: string) {
|
|
const row = this.database.prepare(`
|
|
SELECT * FROM generation_jobs WHERE generation_id = ? AND owner_id = ? AND submission_ready = 1
|
|
`).get(generationId, userId) as GenerationRow | undefined;
|
|
if (!row) throw new GenerationSubmissionError("generation_not_found");
|
|
return this.taskView(row);
|
|
}
|
|
|
|
async submit(input: GenerationSubmissionInput): Promise<GenerationSubmissionResult> {
|
|
const current = this.readCurrentTask(input.userId);
|
|
if (current) return { created: false as const, task: current };
|
|
const { newReferences, ...fields } = input;
|
|
const upload = this.beginUpload(fields);
|
|
try {
|
|
for (const reference of newReferences) await upload.stageReference(reference);
|
|
return await upload.commit();
|
|
} catch (error) {
|
|
upload.abort();
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
beginUpload(input: GenerationSubmissionFields): GenerationUploadSession {
|
|
const preflightModel = this.validate({ ...input, newReferences: [] });
|
|
const staged: StagedManagedFile[] = [];
|
|
let finished = false;
|
|
const abort = () => {
|
|
if (finished) return;
|
|
finished = true;
|
|
for (const file of staged) this.storage.abandonStagedFile(file);
|
|
};
|
|
return {
|
|
abort,
|
|
commit: async () => {
|
|
if (finished) throw new GenerationSubmissionError("generation_request_invalid");
|
|
if (this.beforeTransaction) await this.beforeTransaction();
|
|
try {
|
|
const result = this.immediate(() => this.commitSubmission({ ...input, newReferences: [] }, staged));
|
|
if (result.created) finished = true;
|
|
else abort();
|
|
return result;
|
|
} catch (error) {
|
|
abort();
|
|
throw error;
|
|
}
|
|
},
|
|
stageReference: async (reference) => {
|
|
if (finished) throw new GenerationSubmissionError("generation_request_invalid");
|
|
const nextCount = input.existingReferenceAssetIds.length + staged.length + 1;
|
|
const nextBytes = staged.reduce((sum, file) => sum + file.bytes, 0) + reference.projectedBytes;
|
|
if (nextCount > preflightModel.referenceLimits.maxFiles || nextBytes > preflightModel.referenceLimits.maxTotalBytes
|
|
|| !safeInteger(reference.projectedBytes) || reference.projectedBytes > preflightModel.referenceLimits.maxFileBytes) {
|
|
throw new GenerationSubmissionError("reference_invalid", { errorCategory: "reference_invalid" });
|
|
}
|
|
try {
|
|
staged.push(await this.storage.stagePrivateImage({
|
|
content: reference.content,
|
|
expectedMimeType: reference.mimeType,
|
|
fileName: reference.fileName,
|
|
maximumBytes: preflightModel.referenceLimits.maxFileBytes,
|
|
operationId: randomUUID(),
|
|
ownerRef: input.userId,
|
|
projectedWriteBytes: reference.projectedBytes,
|
|
}));
|
|
} catch (error) {
|
|
if (error instanceof GenerationSubmissionError || (error && typeof error === "object" && "code" in error)) throw error;
|
|
throw new GenerationSubmissionError("reference_invalid", { errorCategory: "reference_invalid" });
|
|
}
|
|
},
|
|
};
|
|
}
|
|
|
|
private validate(input: GenerationSubmissionInput) {
|
|
if (!uuidPattern.test(input.userId) || !uuidPattern.test(input.clientSubmissionId)
|
|
|| !idempotencyPattern.test(input.idempotencyKey) || !projectRatios.includes(input.ratio)
|
|
|| !safeInteger(input.modelConfigVersion) || !safeInteger(input.confirmedCreditCost)
|
|
|| (input.mode === "existing_project" && (!input.projectId || !uuidPattern.test(input.projectId)))
|
|
|| (input.mode === "new_project" && input.projectId !== undefined)) {
|
|
throw new GenerationSubmissionError("generation_request_invalid");
|
|
}
|
|
const prompt = normalizePrompt(input.prompt);
|
|
const model = this.models.readModel(input.modelId);
|
|
if (!model) throw new GenerationSubmissionError("generation_blocked", { errorCategory: "model_disabled" });
|
|
const workerState = this.database.prepare("SELECT status FROM worker_runtime_state WHERE singleton = 1").get() as { status: "ready" | "degraded" } | undefined;
|
|
if (workerState?.status === "degraded") throw new GenerationSubmissionError("generation_blocked", { errorCategory: "gateway_contract_invalid" });
|
|
if (model.configVersion !== input.modelConfigVersion || model.creditCost !== input.confirmedCreditCost) {
|
|
throw new GenerationSubmissionError("model_config_stale", {
|
|
latest: { configVersion: model.configVersion, creditCost: model.creditCost, modelId: model.modelId },
|
|
});
|
|
}
|
|
if (!model.enabled) throw new GenerationSubmissionError("generation_blocked", { errorCategory: "model_disabled" });
|
|
if (model.contractValidationStatus !== "verified") {
|
|
throw new GenerationSubmissionError("generation_blocked", { errorCategory: "gateway_contract_invalid" });
|
|
}
|
|
if (!model.runtimeAvailability.availableForNewJobs) {
|
|
throw new GenerationSubmissionError("generation_blocked", { errorCategory: model.runtimeAvailability.reason ?? "model_disabled" });
|
|
}
|
|
if (prompt.length > model.promptMaxLength || !model.supportedRatios.includes(input.ratio)) {
|
|
throw new GenerationSubmissionError("generation_request_invalid");
|
|
}
|
|
const referenceCount = input.newReferences.length + input.existingReferenceAssetIds.length;
|
|
const projectedBytes = input.newReferences.reduce((sum, reference) => sum + reference.projectedBytes, 0);
|
|
if (referenceCount > model.referenceLimits.maxFiles || projectedBytes > model.referenceLimits.maxTotalBytes
|
|
|| input.newReferences.some((reference) => !safeInteger(reference.projectedBytes) || reference.projectedBytes > model.referenceLimits.maxFileBytes)
|
|
|| new Set(input.existingReferenceAssetIds).size !== input.existingReferenceAssetIds.length
|
|
|| input.existingReferenceAssetIds.some((id) => !uuidPattern.test(id))
|
|
|| (input.mode === "new_project" && input.existingReferenceAssetIds.length > 0)) {
|
|
throw new GenerationSubmissionError("reference_invalid", { errorCategory: "reference_invalid" });
|
|
}
|
|
const storageState = this.storage.getState();
|
|
if (storageState.storage_status !== "active") {
|
|
throw new GenerationSubmissionError("generation_storage_unavailable", {
|
|
storage: {
|
|
capacityStatus: storageState.storage_status,
|
|
remainingBytes: Math.max(
|
|
0,
|
|
storageState.hard_limit_bytes - storageState.managed_content_bytes - storageState.active_storage_reservations_bytes,
|
|
),
|
|
},
|
|
});
|
|
}
|
|
return model;
|
|
}
|
|
|
|
private commitSubmission(input: GenerationSubmissionInput, staged: StagedManagedFile[]) {
|
|
const model = this.validate(input);
|
|
const requestHash = digest(stableJson({
|
|
client_submission_id: input.clientSubmissionId,
|
|
confirmed_credit_cost: input.confirmedCreditCost,
|
|
existing_reference_asset_ids: input.existingReferenceAssetIds,
|
|
mode: input.mode,
|
|
model_config_version: input.modelConfigVersion,
|
|
model_id: input.modelId,
|
|
new_references: staged.map((file) => ({ bytes: file.bytes, mime_type: file.mimeType, sha256: file.sha256 })),
|
|
project_id: input.projectId ?? null,
|
|
prompt: normalizePrompt(input.prompt),
|
|
ratio: input.ratio,
|
|
}));
|
|
const keyDigest = digest(input.idempotencyKey);
|
|
const receipt = this.database.prepare(`
|
|
SELECT r.request_hash, g.* FROM generation_submission_receipts r
|
|
JOIN generation_jobs g ON g.generation_id = r.generation_id
|
|
WHERE r.owner_id = ? AND r.idempotency_key_digest = ?
|
|
`).get(input.userId, keyDigest) as (GenerationRow & { request_hash: string }) | undefined;
|
|
if (receipt) {
|
|
if (receipt.request_hash !== requestHash) throw new GenerationSubmissionError("generation_idempotency_conflict");
|
|
return { created: false as const, task: this.taskView(receipt) };
|
|
}
|
|
const bySubmission = this.database.prepare("SELECT * FROM generation_jobs WHERE client_submission_id = ?")
|
|
.get(input.clientSubmissionId) as (GenerationRow & { submission_request_hash: string | null }) | undefined;
|
|
if (bySubmission) {
|
|
if (bySubmission.submission_request_hash !== requestHash || bySubmission.owner_id !== input.userId) {
|
|
throw new GenerationSubmissionError("generation_idempotency_conflict");
|
|
}
|
|
return { created: false as const, task: this.taskView(bySubmission) };
|
|
}
|
|
const current = this.readCurrentTask(input.userId);
|
|
if (current) return { created: false as const, task: current };
|
|
|
|
const now = this.clock();
|
|
const prompt = normalizePrompt(input.prompt);
|
|
const generationId = randomUUID();
|
|
const projectId = input.mode === "new_project"
|
|
? this.insertProject(input.userId, prompt, input.ratio, now)
|
|
: this.validateExistingProject(input.userId, input.projectId!, input.ratio, prompt, now);
|
|
this.database.prepare(`
|
|
INSERT INTO generation_jobs (
|
|
generation_id, owner_id, project_id, prompt, ratio, status, model_id, model_config_version,
|
|
confirmed_credit_cost, reserved_credits, final_credit_state, error_category, created_at, updated_at,
|
|
client_submission_id, submission_request_hash, submission_ready, config_snapshot_json
|
|
) VALUES (?, ?, ?, ?, ?, 'queued', ?, ?, ?, 0, NULL, NULL, ?, ?, ?, ?, 0, ?)
|
|
`).run(
|
|
generationId, input.userId, projectId, prompt, input.ratio, input.modelId, input.modelConfigVersion,
|
|
input.confirmedCreditCost, now, now, input.clientSubmissionId, requestHash,
|
|
stableJson({
|
|
config_set_version: model.configSetVersion,
|
|
config_version: model.configVersion,
|
|
credit_cost: model.creditCost,
|
|
model_id: model.modelId,
|
|
prompt_max_length: model.promptMaxLength,
|
|
reference_limits: model.referenceLimits,
|
|
supported_ratios: model.supportedRatios,
|
|
}),
|
|
);
|
|
|
|
const referenceIds: string[] = [];
|
|
for (const file of staged) {
|
|
this.storage.moveStagedFile(file);
|
|
this.database.prepare(`
|
|
INSERT INTO managed_files (file_id, file_kind, owner_ref, relative_path, byte_size, mime_type, sha256, status, created_at)
|
|
VALUES (?, 'reference', ?, ?, ?, ?, ?, 'committed', ?)
|
|
`).run(file.fileId, input.userId, file.relativePath, file.bytes, file.mimeType, file.sha256, iso(now));
|
|
this.database.prepare(`INSERT INTO project_resource_files (project_id, managed_file_id, resource_kind, created_at) VALUES (?, ?, 'reference', ?)`)
|
|
.run(projectId, file.fileId, now);
|
|
this.database.prepare(`INSERT INTO project_asset_refs (reference_id, managed_file_id, reference_type, created_at) VALUES (?, ?, 'project', ?)`)
|
|
.run(`project:${projectId}:${file.fileId}`, file.fileId, iso(now));
|
|
referenceIds.push(file.fileId);
|
|
}
|
|
for (const referenceId of input.existingReferenceAssetIds) {
|
|
const allowed = this.database.prepare(`
|
|
SELECT mf.file_id FROM managed_files mf
|
|
JOIN project_resource_files prf ON prf.managed_file_id = mf.file_id
|
|
JOIN projects p ON p.project_id = prf.project_id
|
|
WHERE mf.file_id = ? AND mf.file_kind = 'reference' AND mf.status = 'committed'
|
|
AND mf.owner_ref = ? AND prf.project_id = ? AND p.owner_id = ? AND p.status = 'active'
|
|
`).get(referenceId, input.userId, projectId, input.userId);
|
|
if (!allowed) throw new GenerationSubmissionError("reference_invalid", { errorCategory: "reference_invalid" });
|
|
referenceIds.push(referenceId);
|
|
}
|
|
referenceIds.forEach((referenceId, index) => this.database.prepare(`
|
|
INSERT INTO generation_reference_snapshots (generation_id, position, managed_file_id, source_kind, created_at)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
`).run(generationId, index, referenceId, index < staged.length ? "uploaded" : "existing", now));
|
|
|
|
this.credits.reserveGeneration({
|
|
creditCost: input.confirmedCreditCost,
|
|
generationId,
|
|
modelId: input.modelId,
|
|
operationKey: `generation:${generationId}:reserve`,
|
|
userId: input.userId,
|
|
});
|
|
this.database.prepare("UPDATE generation_jobs SET submission_ready = 1 WHERE generation_id = ?").run(generationId);
|
|
this.database.prepare(`
|
|
INSERT INTO generation_submission_receipts (owner_id, idempotency_key_digest, request_hash, generation_id, created_at)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
`).run(input.userId, keyDigest, requestHash, generationId, now);
|
|
this.consumeStagedStorage(staged, now);
|
|
return { created: true as const, task: this.readTask(input.userId, generationId) };
|
|
}
|
|
|
|
private insertProject(ownerId: string, prompt: string, ratio: ProjectRatio, now: number) {
|
|
const active = this.database.prepare("SELECT COUNT(*) AS count FROM projects WHERE owner_id = ? AND status = 'active'")
|
|
.get(ownerId) as { count: number };
|
|
if (active.count >= projectLimit) throw new GenerationSubmissionError("generation_request_invalid");
|
|
const projectId = randomUUID();
|
|
const pixels = ratioPixels[ratio];
|
|
const name = defaultProjectName(prompt, now);
|
|
this.database.prepare(`
|
|
INSERT INTO projects (
|
|
project_id, owner_id, name, draft_prompt, ratio, pixel_width, pixel_height,
|
|
status, state_version, current_image_id, created_at, updated_at, deleted_at, purge_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, 'active', 1, NULL, ?, ?, NULL, NULL)
|
|
`).run(projectId, ownerId, name, prompt, ratio, pixels.width, pixels.height, now, now);
|
|
this.database.prepare(`
|
|
INSERT INTO project_states (project_id, state_version, name, canvas_json, created_at) VALUES (?, 1, ?, ?, ?)
|
|
`).run(projectId, name, stableJson(defaultCanvasState(ratio, pixels, null)), now);
|
|
return projectId;
|
|
}
|
|
|
|
private validateExistingProject(ownerId: string, projectId: string, ratio: ProjectRatio, prompt: string, now: number) {
|
|
const project = this.database.prepare("SELECT ratio FROM projects WHERE project_id = ? AND owner_id = ? AND status = 'active'")
|
|
.get(projectId, ownerId) as { ratio: ProjectRatio } | undefined;
|
|
if (!project || project.ratio !== ratio) throw new GenerationSubmissionError("generation_request_invalid");
|
|
const history = this.database.prepare("SELECT COUNT(*) AS count FROM project_images WHERE project_id = ?").get(projectId) as { count: number };
|
|
if (history.count >= historyLimit) throw new GenerationSubmissionError("generation_request_invalid");
|
|
this.database.prepare("UPDATE projects SET draft_prompt = ?, updated_at = ? WHERE project_id = ?").run(prompt, now, projectId);
|
|
return projectId;
|
|
}
|
|
|
|
private consumeStagedStorage(staged: StagedManagedFile[], now: number) {
|
|
if (staged.length === 0) return;
|
|
const total = staged.reduce((sum, file) => sum + file.bytes, 0);
|
|
for (const file of staged) {
|
|
this.database.prepare(`
|
|
UPDATE storage_reservations SET status = 'consumed', resolved_at = ?
|
|
WHERE operation_id = ? AND status = 'active'
|
|
`).run(iso(now), file.operationId);
|
|
}
|
|
const state = this.database.prepare("SELECT managed_content_bytes FROM local_backend_storage_state WHERE singleton = 1")
|
|
.get() as { managed_content_bytes: number };
|
|
const active = this.database.prepare("SELECT COALESCE(SUM(projected_bytes), 0) AS bytes FROM storage_reservations WHERE status = 'active'")
|
|
.get() as { bytes: number };
|
|
const nextBytes = state.managed_content_bytes + total;
|
|
const classification = classifyCapacity(nextBytes, active.bytes);
|
|
this.database.prepare(`
|
|
UPDATE local_backend_storage_state
|
|
SET managed_content_bytes = ?, capacity_notice_level = ?, storage_status = ?, measured_at = ?, version = version + 1
|
|
WHERE singleton = 1
|
|
`).run(nextBytes, classification.capacity_notice_level, classification.storage_status, iso(now));
|
|
}
|
|
|
|
private taskView(row: GenerationRow): GenerationTaskView {
|
|
const referenceAssetIds = (this.database.prepare(`
|
|
SELECT managed_file_id FROM generation_reference_snapshots WHERE generation_id = ? ORDER BY position
|
|
`).all(row.generation_id) as Array<{ managed_file_id: string }>).map((entry) => entry.managed_file_id);
|
|
return {
|
|
confirmedCreditCost: row.confirmed_credit_cost,
|
|
createdAt: iso(row.created_at),
|
|
errorCategory: row.error_category,
|
|
generationId: row.generation_id,
|
|
modelConfigVersion: row.model_config_version,
|
|
modelId: row.model_id,
|
|
projectId: row.project_id,
|
|
prompt: row.prompt,
|
|
ratio: row.ratio,
|
|
referenceAssetIds,
|
|
referenceCount: referenceAssetIds.length,
|
|
reservedCredits: row.reserved_credits,
|
|
status: row.status,
|
|
updatedAt: iso(row.updated_at),
|
|
};
|
|
}
|
|
|
|
private immediate<T>(action: () => T): T {
|
|
this.database.exec("BEGIN IMMEDIATE");
|
|
try {
|
|
const result = action();
|
|
this.database.exec("COMMIT");
|
|
return result;
|
|
} catch (error) {
|
|
if (this.database.inTransaction) this.database.exec("ROLLBACK");
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
private ensureColumn(table: string, column: string, definition: string) {
|
|
const columns = this.database.prepare(`PRAGMA table_info(${table})`).all() as Array<{ name: string }>;
|
|
if (!columns.some((value) => value.name === column)) this.database.exec(`ALTER TABLE ${table} ADD COLUMN ${column} ${definition}`);
|
|
}
|
|
|
|
private migrate() {
|
|
this.ensureColumn("generation_jobs", "client_submission_id", "TEXT");
|
|
this.ensureColumn("generation_jobs", "submission_request_hash", "TEXT");
|
|
this.ensureColumn("generation_jobs", "submission_ready", "INTEGER NOT NULL DEFAULT 0");
|
|
this.ensureColumn("generation_jobs", "config_snapshot_json", "TEXT");
|
|
this.database.exec(`
|
|
CREATE TABLE IF NOT EXISTS worker_runtime_state (
|
|
singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
|
|
status TEXT NOT NULL CHECK (status IN ('ready', 'degraded')),
|
|
reason TEXT,
|
|
owner_id TEXT,
|
|
lock_expires_at INTEGER,
|
|
updated_at INTEGER NOT NULL
|
|
);
|
|
CREATE UNIQUE INDEX IF NOT EXISTS generation_jobs_active_owner
|
|
ON generation_jobs(owner_id) WHERE status IN ('queued', 'running');
|
|
CREATE UNIQUE INDEX IF NOT EXISTS generation_jobs_client_submission
|
|
ON generation_jobs(client_submission_id) WHERE client_submission_id IS NOT NULL;
|
|
CREATE TABLE IF NOT EXISTS generation_submission_receipts (
|
|
owner_id TEXT NOT NULL,
|
|
idempotency_key_digest TEXT NOT NULL CHECK (length(idempotency_key_digest) = 64),
|
|
request_hash TEXT NOT NULL CHECK (length(request_hash) = 64),
|
|
generation_id TEXT NOT NULL UNIQUE REFERENCES generation_jobs(generation_id),
|
|
created_at INTEGER NOT NULL,
|
|
PRIMARY KEY (owner_id, idempotency_key_digest)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS generation_reference_snapshots (
|
|
generation_id TEXT NOT NULL REFERENCES generation_jobs(generation_id) ON DELETE CASCADE,
|
|
position INTEGER NOT NULL CHECK (position >= 0),
|
|
managed_file_id TEXT NOT NULL REFERENCES managed_files(file_id),
|
|
source_kind TEXT NOT NULL CHECK (source_kind IN ('uploaded', 'existing')),
|
|
created_at INTEGER NOT NULL,
|
|
PRIMARY KEY (generation_id, position),
|
|
UNIQUE (generation_id, managed_file_id)
|
|
);
|
|
CREATE TRIGGER IF NOT EXISTS generation_reference_snapshots_no_update
|
|
BEFORE UPDATE ON generation_reference_snapshots BEGIN SELECT RAISE(ABORT, 'generation_reference_snapshot_immutable'); END;
|
|
DROP TRIGGER IF EXISTS generation_reference_snapshots_no_delete;
|
|
CREATE TRIGGER generation_reference_snapshots_no_delete
|
|
BEFORE DELETE ON generation_reference_snapshots
|
|
WHEN dada_allow_privacy_purge() <> 1
|
|
BEGIN SELECT RAISE(ABORT, 'generation_reference_snapshot_immutable'); END;
|
|
`);
|
|
this.ensureColumn("worker_runtime_state", "owner_id", "TEXT");
|
|
this.ensureColumn("worker_runtime_state", "lock_expires_at", "INTEGER");
|
|
this.database.prepare("INSERT OR IGNORE INTO worker_runtime_state (singleton, status, reason, owner_id, lock_expires_at, updated_at) VALUES (1, 'ready', NULL, NULL, NULL, ?)").run(this.clock());
|
|
}
|
|
}
|