898 lines
39 KiB
TypeScript
898 lines
39 KiB
TypeScript
import "server-only";
|
|
|
|
import { readFile, rename, writeFile } from "node:fs/promises";
|
|
import { join } from "node:path";
|
|
import type { AppState, Asset, BillingParameterSnapshot, BillingPriceSource, BillingQuantitySource, BillingRuleConditions, BillingSelectedParameterTier, GenerationCapability, GenerationJob, GenerationStatus, ImageTemplate, Project, UsageContext, UsageEvent, UsageSource } from "@/lib/types";
|
|
import { isPostgresBackend, queryDatabase, withDatabaseTransaction } from "@/lib/server/database";
|
|
import { createId } from "@/lib/server/ids";
|
|
import { dataDir, DEFAULT_OWNER_ID, ensureRuntimeDirs } from "@/lib/server/runtime";
|
|
|
|
const STORE_FILE = "web-app-state.json";
|
|
let localWriteQueue: Promise<unknown> = Promise.resolve();
|
|
|
|
type AssetInput = Omit<Asset, "id" | "createdAt" | "updatedAt"> & Partial<Pick<Asset, "id" | "createdAt" | "updatedAt">>;
|
|
type JobInput = Omit<GenerationJob, "id" | "createdAt" | "updatedAt"> & Partial<Pick<GenerationJob, "id" | "createdAt" | "updatedAt">>;
|
|
type UsageInput = Omit<UsageEvent, "id" | "createdAt"> & Partial<Pick<UsageEvent, "id" | "createdAt">>;
|
|
type ImageTemplateInput = Omit<ImageTemplate, "id" | "createdAt" | "updatedAt"> & Partial<Pick<ImageTemplate, "id" | "createdAt" | "updatedAt">>;
|
|
|
|
export type GenerationJobListFilters = {
|
|
ownerId?: string;
|
|
externalClientId?: string;
|
|
status?: GenerationStatus;
|
|
capability?: GenerationCapability;
|
|
limit?: number;
|
|
before?: string;
|
|
};
|
|
|
|
export type UsageEventListFilters = {
|
|
ownerId?: string;
|
|
source?: UsageSource;
|
|
from?: string;
|
|
to?: string;
|
|
};
|
|
|
|
export type ClaimGenerationJobsInput = {
|
|
workerId: string;
|
|
limit?: number;
|
|
lockTimeoutMs?: number;
|
|
};
|
|
|
|
export async function listAssets(ownerId = DEFAULT_OWNER_ID): Promise<Asset[]> {
|
|
if (isPostgresBackend()) {
|
|
const { rows } = await queryDatabase("SELECT * FROM assets WHERE owner_id = $1 ORDER BY created_at DESC", [ownerId]);
|
|
return rows.map(assetFromRow);
|
|
}
|
|
const state = await readState();
|
|
return state.assets.filter((asset) => asset.ownerId === ownerId).sort(sortNewest);
|
|
}
|
|
|
|
export async function getAsset(id: string): Promise<Asset | null> {
|
|
if (isPostgresBackend()) {
|
|
const { rows } = await queryDatabase("SELECT * FROM assets WHERE id = $1 LIMIT 1", [id]);
|
|
return rows[0] ? assetFromRow(rows[0]) : null;
|
|
}
|
|
const state = await readState();
|
|
return state.assets.find((asset) => asset.id === id) || null;
|
|
}
|
|
|
|
export async function getAssetByStoragePath(storagePath: string): Promise<Asset | null> {
|
|
if (isPostgresBackend()) {
|
|
const { rows } = await queryDatabase("SELECT * FROM assets WHERE storage_path = $1 LIMIT 1", [storagePath]);
|
|
return rows[0] ? assetFromRow(rows[0]) : null;
|
|
}
|
|
const state = await readState();
|
|
return state.assets.find((asset) => asset.storagePath === storagePath) || null;
|
|
}
|
|
|
|
export async function createAsset(input: AssetInput): Promise<Asset> {
|
|
const now = new Date().toISOString();
|
|
const asset: Asset = {
|
|
...input,
|
|
id: input.id || createId("asset"),
|
|
ownerId: input.ownerId || DEFAULT_OWNER_ID,
|
|
tags: input.tags || [],
|
|
metadata: input.metadata || {},
|
|
createdAt: input.createdAt || now,
|
|
updatedAt: input.updatedAt || now
|
|
};
|
|
if (isPostgresBackend()) {
|
|
return assetFromRow(await insertRow("assets", assetToRow(asset)));
|
|
}
|
|
return mutateLocalState((state) => {
|
|
state.assets.unshift(asset);
|
|
return asset;
|
|
});
|
|
}
|
|
|
|
export async function deleteAsset(id: string): Promise<Asset | null> {
|
|
const existing = await getAsset(id);
|
|
if (!existing) return null;
|
|
if (isPostgresBackend()) {
|
|
await queryDatabase("DELETE FROM assets WHERE id = $1", [id]);
|
|
return existing;
|
|
}
|
|
return mutateLocalState((state) => {
|
|
state.assets = state.assets.filter((asset) => asset.id !== id);
|
|
state.generationJobs = state.generationJobs.map((job) => ({
|
|
...job,
|
|
inputAssetIds: job.inputAssetIds.filter((assetId) => assetId !== id),
|
|
outputAssetIds: job.outputAssetIds.filter((assetId) => assetId !== id)
|
|
}));
|
|
return existing;
|
|
});
|
|
}
|
|
|
|
export async function listGenerationJobs(ownerId = DEFAULT_OWNER_ID, limit = 200): Promise<GenerationJob[]> {
|
|
return listGenerationJobsFiltered({ ownerId, limit });
|
|
}
|
|
|
|
export async function listGenerationJobsFiltered(filters: GenerationJobListFilters = {}): Promise<GenerationJob[]> {
|
|
const ownerId = filters.ownerId || DEFAULT_OWNER_ID;
|
|
const limit = filters.limit || 200;
|
|
if (isPostgresBackend()) {
|
|
const clauses = ["owner_id = $1"];
|
|
const values: unknown[] = [ownerId];
|
|
addFilter(clauses, values, "external_client_id", filters.externalClientId);
|
|
addFilter(clauses, values, "status", filters.status);
|
|
addFilter(clauses, values, "capability", filters.capability);
|
|
if (filters.before) { values.push(filters.before); clauses.push(`created_at < $${values.length}`); }
|
|
values.push(limit);
|
|
const { rows } = await queryDatabase(`SELECT * FROM generation_jobs WHERE ${clauses.join(" AND ")} ORDER BY created_at DESC LIMIT $${values.length}`, values);
|
|
return rows.map(jobFromRow);
|
|
}
|
|
const state = await readState();
|
|
return state.generationJobs
|
|
.filter((job) => job.ownerId === ownerId)
|
|
.filter((job) => !filters.externalClientId || job.externalClientId === filters.externalClientId)
|
|
.filter((job) => !filters.status || job.status === filters.status)
|
|
.filter((job) => !filters.capability || job.capability === filters.capability)
|
|
.filter((job) => !filters.before || job.createdAt < filters.before)
|
|
.sort(sortNewest)
|
|
.slice(0, limit);
|
|
}
|
|
|
|
export async function getGenerationJob(id: string): Promise<GenerationJob | null> {
|
|
if (isPostgresBackend()) {
|
|
const { rows } = await queryDatabase("SELECT * FROM generation_jobs WHERE id = $1 LIMIT 1", [id]);
|
|
return rows[0] ? jobFromRow(rows[0]) : null;
|
|
}
|
|
const state = await readState();
|
|
return state.generationJobs.find((job) => job.id === id) || null;
|
|
}
|
|
|
|
export async function createGenerationJob(input: JobInput): Promise<GenerationJob> {
|
|
const now = new Date().toISOString();
|
|
const job: GenerationJob = {
|
|
...input,
|
|
id: input.id || createId("job"),
|
|
ownerId: input.ownerId || DEFAULT_OWNER_ID,
|
|
inputAssetIds: input.inputAssetIds || [],
|
|
inputUrls: input.inputUrls || [],
|
|
outputAssetIds: input.outputAssetIds || [],
|
|
requestPayload: input.requestPayload || {},
|
|
priority: input.priority ?? 0,
|
|
attempts: input.attempts ?? 0,
|
|
maxAttempts: input.maxAttempts ?? 3,
|
|
scheduledAt: input.scheduledAt || now,
|
|
webhookAttempts: input.webhookAttempts ?? 0,
|
|
createdAt: input.createdAt || now,
|
|
updatedAt: input.updatedAt || now
|
|
};
|
|
if (isPostgresBackend()) {
|
|
return jobFromRow(await insertRow("generation_jobs", jobToRow(job)));
|
|
}
|
|
return mutateLocalState((state) => {
|
|
state.generationJobs.unshift(job);
|
|
return job;
|
|
});
|
|
}
|
|
|
|
export async function findGenerationJobByIdempotency(
|
|
externalClientId: string,
|
|
idempotencyKey: string,
|
|
ownerId = DEFAULT_OWNER_ID
|
|
): Promise<GenerationJob | null> {
|
|
if (isPostgresBackend()) {
|
|
const { rows } = await queryDatabase("SELECT * FROM generation_jobs WHERE owner_id = $1 AND external_client_id = $2 AND idempotency_key = $3 LIMIT 1", [ownerId, externalClientId, idempotencyKey]);
|
|
return rows[0] ? jobFromRow(rows[0]) : null;
|
|
}
|
|
const state = await readState();
|
|
return state.generationJobs.find((job) => (
|
|
job.ownerId === ownerId &&
|
|
job.externalClientId === externalClientId &&
|
|
job.idempotencyKey === idempotencyKey
|
|
)) || null;
|
|
}
|
|
|
|
export async function claimGenerationJobs(input: ClaimGenerationJobsInput): Promise<GenerationJob[]> {
|
|
const limit = Math.max(1, Math.min(input.limit || 1, 20));
|
|
const lockTimeoutMs = input.lockTimeoutMs ?? 5 * 60 * 1000;
|
|
if (isPostgresBackend()) {
|
|
const { rows } = await queryDatabase("SELECT * FROM claim_generation_jobs($1, $2, $3)", [input.workerId, limit, Math.ceil(lockTimeoutMs / 1000)]);
|
|
return rows.map(jobFromRow);
|
|
}
|
|
|
|
return mutateLocalState((state) => {
|
|
const now = new Date();
|
|
const nowIso = now.toISOString();
|
|
const staleBefore = new Date(now.getTime() - lockTimeoutMs).toISOString();
|
|
const selected = state.generationJobs
|
|
.filter((job) => isClaimableJob(job, nowIso, staleBefore))
|
|
.sort(sortClaimableJobs)
|
|
.slice(0, limit);
|
|
for (const job of selected) {
|
|
job.lockedAt = nowIso;
|
|
job.lockedBy = input.workerId;
|
|
if (!job.startedAt) job.startedAt = nowIso;
|
|
job.updatedAt = nowIso;
|
|
}
|
|
return selected.map((job) => ({ ...job }));
|
|
});
|
|
}
|
|
|
|
export async function clearGenerationJobLock(
|
|
id: string,
|
|
patch: Partial<GenerationJob> = {},
|
|
options: { clearProviderTaskId?: boolean } = {}
|
|
): Promise<GenerationJob> {
|
|
const updatedAt = new Date().toISOString();
|
|
if (isPostgresBackend()) {
|
|
return jobFromRow(await updateRow("generation_jobs", id, {
|
|
...jobToRow({ ...patch, updatedAt } as GenerationJob), locked_at: null, locked_by: null,
|
|
...(options.clearProviderTaskId ? { provider_task_id: null } : {})
|
|
}));
|
|
}
|
|
return mutateLocalState((state) => {
|
|
const index = state.generationJobs.findIndex((job) => job.id === id);
|
|
if (index === -1) throw new Error(`Generation job not found: ${id}`);
|
|
state.generationJobs[index] = {
|
|
...state.generationJobs[index],
|
|
...patch,
|
|
lockedAt: undefined,
|
|
lockedBy: undefined,
|
|
...(options.clearProviderTaskId ? { providerTaskId: undefined } : {}),
|
|
updatedAt
|
|
};
|
|
return state.generationJobs[index];
|
|
});
|
|
}
|
|
|
|
export async function updateGenerationJob(id: string, patch: Partial<GenerationJob>): Promise<GenerationJob> {
|
|
const updatedAt = new Date().toISOString();
|
|
if (isPostgresBackend()) {
|
|
return jobFromRow(await updateRow("generation_jobs", id, jobToRow({ ...patch, updatedAt } as GenerationJob)));
|
|
}
|
|
return mutateLocalState((state) => {
|
|
const index = state.generationJobs.findIndex((job) => job.id === id);
|
|
if (index === -1) throw new Error(`Generation job not found: ${id}`);
|
|
state.generationJobs[index] = { ...state.generationJobs[index], ...patch, updatedAt };
|
|
return state.generationJobs[index];
|
|
});
|
|
}
|
|
|
|
export async function deleteGenerationJob(id: string): Promise<GenerationJob | null> {
|
|
const existing = await getGenerationJob(id);
|
|
if (!existing) return null;
|
|
if (isPostgresBackend()) {
|
|
await queryDatabase("DELETE FROM generation_jobs WHERE id = $1", [id]);
|
|
return existing;
|
|
}
|
|
return mutateLocalState((state) => {
|
|
state.generationJobs = state.generationJobs.filter((job) => job.id !== id);
|
|
return existing;
|
|
});
|
|
}
|
|
|
|
export async function recordUsageEvent(input: UsageInput): Promise<UsageEvent> {
|
|
const usage: UsageEvent = {
|
|
...input,
|
|
id: input.id || createId("usage"),
|
|
ownerId: input.ownerId || DEFAULT_OWNER_ID,
|
|
createdAt: input.createdAt || new Date().toISOString()
|
|
};
|
|
if (isPostgresBackend()) {
|
|
const existing = await findDatabaseUsageEventByJobId(usage.jobId);
|
|
if (existing) return existing;
|
|
try {
|
|
return usageFromRow(await insertRow("usage_events", usageToRow(usage)));
|
|
} catch (error) {
|
|
if (isUniqueViolation(error)) {
|
|
const raced = await findDatabaseUsageEventByJobId(usage.jobId);
|
|
if (raced) return raced;
|
|
}
|
|
throw error;
|
|
}
|
|
}
|
|
return mutateLocalState((state) => {
|
|
const existing = state.usageEvents.find((event) => event.jobId === usage.jobId);
|
|
if (existing) return existing;
|
|
state.usageEvents.unshift(usage);
|
|
return usage;
|
|
});
|
|
}
|
|
|
|
export async function recordUsageForJob(job: GenerationJob): Promise<UsageEvent | null> {
|
|
if (job.provider === "mock" || job.externalClientId || job.usageContext?.source === "api") return null;
|
|
return recordUsageEvent({
|
|
ownerId: job.ownerId,
|
|
jobId: job.id,
|
|
source: "platform",
|
|
capability: job.capability,
|
|
provider: job.provider,
|
|
reqKey: job.reqKey,
|
|
accountUsername: job.usageContext?.username,
|
|
accountDisplayName: job.usageContext?.displayName,
|
|
tenantId: job.usageContext?.tenantId,
|
|
organizationId: job.usageContext?.organizationId,
|
|
organizationName: job.usageContext?.organizationName,
|
|
quantity: job.billing?.quantity || 1,
|
|
estimatedUnit: job.billing?.unit === "video_second" ? "video_second" : job.billing?.unit === "image" ? "image" : "job",
|
|
chargedAmountFen: job.billing?.amountFen,
|
|
currency: job.billing?.currency
|
|
});
|
|
}
|
|
|
|
export async function listUsageEvents(filters: UsageEventListFilters = {}): Promise<UsageEvent[]> {
|
|
if (isPostgresBackend()) {
|
|
const clauses: string[] = [];
|
|
const values: unknown[] = [];
|
|
addFilter(clauses, values, "owner_id", filters.ownerId);
|
|
addFilter(clauses, values, "source", filters.source);
|
|
if (filters.from) { values.push(filters.from); clauses.push(`created_at >= $${values.length}`); }
|
|
if (filters.to) { values.push(filters.to); clauses.push(`created_at < $${values.length}`); }
|
|
const where = clauses.length ? ` WHERE ${clauses.join(" AND ")}` : "";
|
|
const { rows } = await queryDatabase(`SELECT * FROM usage_events${where} ORDER BY created_at DESC`, values);
|
|
return dedupeUsageEvents(rows.map(usageFromRow));
|
|
}
|
|
|
|
const state = await readState();
|
|
const jobs = new Map(state.generationJobs.map((job) => [job.id, job]));
|
|
return dedupeUsageEvents(state.usageEvents.map((event) => enrichLegacyUsageEvent(event, jobs.get(event.jobId))))
|
|
.filter((event) => !filters.ownerId || event.ownerId === filters.ownerId)
|
|
.filter((event) => !filters.source || event.source === filters.source)
|
|
.filter((event) => !filters.from || event.createdAt >= filters.from)
|
|
.filter((event) => !filters.to || event.createdAt < filters.to)
|
|
.sort(sortNewest);
|
|
}
|
|
|
|
export async function listProjects(ownerId = DEFAULT_OWNER_ID): Promise<Project[]> {
|
|
if (isPostgresBackend()) {
|
|
const { rows } = await queryDatabase("SELECT * FROM projects WHERE owner_id = $1 ORDER BY created_at DESC", [ownerId]);
|
|
return rows.map(projectFromRow);
|
|
}
|
|
const state = await readState();
|
|
return state.projects.filter((project) => project.ownerId === ownerId).sort(sortNewest);
|
|
}
|
|
|
|
export async function listImageTemplates(ownerId = DEFAULT_OWNER_ID): Promise<ImageTemplate[]> {
|
|
if (isPostgresBackend()) {
|
|
const { rows } = await queryDatabase("SELECT * FROM image_templates WHERE owner_id = $1 ORDER BY sort_order ASC, updated_at DESC", [ownerId]);
|
|
return rows.map(imageTemplateFromRow);
|
|
}
|
|
const state = await readState();
|
|
return state.imageTemplates
|
|
.filter((template) => template.ownerId === ownerId)
|
|
.sort(sortTemplates);
|
|
}
|
|
|
|
export async function getImageTemplate(id: string): Promise<ImageTemplate | null> {
|
|
if (isPostgresBackend()) {
|
|
const { rows } = await queryDatabase("SELECT * FROM image_templates WHERE id = $1 LIMIT 1", [id]);
|
|
return rows[0] ? imageTemplateFromRow(rows[0]) : null;
|
|
}
|
|
const state = await readState();
|
|
return state.imageTemplates.find((template) => template.id === id) || null;
|
|
}
|
|
|
|
export async function createImageTemplate(input: ImageTemplateInput): Promise<ImageTemplate> {
|
|
const now = new Date().toISOString();
|
|
const template: ImageTemplate = {
|
|
...input,
|
|
id: input.id || createId("tmpl"),
|
|
ownerId: input.ownerId || DEFAULT_OWNER_ID,
|
|
settings: input.settings || {},
|
|
sortOrder: input.sortOrder ?? 0,
|
|
createdAt: input.createdAt || now,
|
|
updatedAt: input.updatedAt || now
|
|
};
|
|
if (isPostgresBackend()) {
|
|
return imageTemplateFromRow(await insertRow("image_templates", imageTemplateToRow(template)));
|
|
}
|
|
return mutateLocalState((state) => {
|
|
state.imageTemplates.unshift(template);
|
|
state.imageTemplates.sort(sortTemplates);
|
|
return template;
|
|
});
|
|
}
|
|
|
|
export async function updateImageTemplate(
|
|
id: string,
|
|
ownerId: string,
|
|
patch: Partial<Omit<ImageTemplate, "id" | "ownerId" | "createdAt" | "updatedAt">>
|
|
): Promise<ImageTemplate | null> {
|
|
const existing = await getImageTemplate(id);
|
|
if (!existing || existing.ownerId !== ownerId) return null;
|
|
const updated: ImageTemplate = {
|
|
...existing,
|
|
...patch,
|
|
settings: patch.settings || existing.settings,
|
|
updatedAt: new Date().toISOString()
|
|
};
|
|
if (isPostgresBackend()) {
|
|
const row = imageTemplateToRow(updated);
|
|
const entries = Object.entries(row);
|
|
const values = entries.map(([, value]) => value);
|
|
values.push(id, ownerId);
|
|
const assignments = entries.map(([column], index) => `${column} = $${index + 1}`).join(", ");
|
|
const { rows } = await queryDatabase(`UPDATE image_templates SET ${assignments} WHERE id = $${values.length - 1} AND owner_id = $${values.length} RETURNING *`, values);
|
|
return rows[0] ? imageTemplateFromRow(rows[0]) : null;
|
|
}
|
|
return mutateLocalState((state) => {
|
|
const index = state.imageTemplates.findIndex((template) => template.id === id && template.ownerId === ownerId);
|
|
if (index === -1) return null;
|
|
state.imageTemplates[index] = updated;
|
|
state.imageTemplates.sort(sortTemplates);
|
|
return updated;
|
|
});
|
|
}
|
|
|
|
export async function deleteImageTemplate(id: string, ownerId: string): Promise<ImageTemplate | null> {
|
|
const existing = await getImageTemplate(id);
|
|
if (!existing || existing.ownerId !== ownerId) return null;
|
|
if (isPostgresBackend()) {
|
|
await queryDatabase("DELETE FROM image_templates WHERE id = $1 AND owner_id = $2", [id, ownerId]);
|
|
return existing;
|
|
}
|
|
return mutateLocalState((state) => {
|
|
state.imageTemplates = state.imageTemplates.filter((template) => !(template.id === id && template.ownerId === ownerId));
|
|
return existing;
|
|
});
|
|
}
|
|
|
|
export async function reassignOwnerData(fromOwnerId: string, toOwnerId: string): Promise<void> {
|
|
if (!fromOwnerId || !toOwnerId || fromOwnerId === toOwnerId) return;
|
|
if (isPostgresBackend()) {
|
|
await withDatabaseTransaction(async (client) => {
|
|
for (const table of ["assets", "generation_jobs", "projects", "image_templates", "usage_events"] as const) {
|
|
await client.query(`UPDATE ${table} SET owner_id = $1 WHERE owner_id = $2`, [toOwnerId, fromOwnerId]);
|
|
}
|
|
});
|
|
return;
|
|
}
|
|
await mutateLocalState((state) => {
|
|
for (const asset of state.assets) if (asset.ownerId === fromOwnerId) asset.ownerId = toOwnerId;
|
|
for (const job of state.generationJobs) if (job.ownerId === fromOwnerId) job.ownerId = toOwnerId;
|
|
for (const project of state.projects) if (project.ownerId === fromOwnerId) project.ownerId = toOwnerId;
|
|
for (const template of state.imageTemplates) if (template.ownerId === fromOwnerId) template.ownerId = toOwnerId;
|
|
});
|
|
}
|
|
|
|
async function readState(): Promise<AppState> {
|
|
await ensureRuntimeDirs();
|
|
const path = join(dataDir(), STORE_FILE);
|
|
try {
|
|
return normalizeState(JSON.parse(await readFile(path, "utf8")));
|
|
} catch {
|
|
const state = normalizeState({});
|
|
await writeState(state);
|
|
return state;
|
|
}
|
|
}
|
|
|
|
async function writeState(state: AppState): Promise<void> {
|
|
await ensureRuntimeDirs();
|
|
const path = join(dataDir(), STORE_FILE);
|
|
const temp = `${path}.${createId("tmp")}.tmp`;
|
|
await writeFile(temp, JSON.stringify(normalizeState(state), null, 2));
|
|
await rename(temp, path);
|
|
}
|
|
|
|
async function mutateLocalState<T>(mutator: (state: AppState) => T): Promise<T> {
|
|
const run = localWriteQueue.then(async () => {
|
|
const state = await readState();
|
|
const result = mutator(state);
|
|
await writeState(state);
|
|
return result;
|
|
});
|
|
localWriteQueue = run.catch(() => undefined);
|
|
return run;
|
|
}
|
|
|
|
function normalizeState(raw: Partial<AppState>): AppState {
|
|
return {
|
|
users: raw.users?.length
|
|
? raw.users
|
|
: [{ id: DEFAULT_OWNER_ID, email: "demo@zhinian.local", displayName: "智念演示用户" }],
|
|
assets: raw.assets || [],
|
|
generationJobs: raw.generationJobs || [],
|
|
usageEvents: raw.usageEvents || [],
|
|
projects: raw.projects || [],
|
|
imageTemplates: raw.imageTemplates || []
|
|
};
|
|
}
|
|
|
|
function sortNewest<T extends { createdAt: string }>(a: T, b: T): number {
|
|
return b.createdAt.localeCompare(a.createdAt);
|
|
}
|
|
|
|
function sortTemplates(a: ImageTemplate, b: ImageTemplate): number {
|
|
const sortOrder = (a.sortOrder || 0) - (b.sortOrder || 0);
|
|
if (sortOrder !== 0) return sortOrder;
|
|
return b.updatedAt.localeCompare(a.updatedAt);
|
|
}
|
|
|
|
function isClaimableJob(job: GenerationJob, nowIso: string, staleBefore: string): boolean {
|
|
if (["succeeded", "failed", "expired", "cancelled"].includes(job.status)) return false;
|
|
if ((job.scheduledAt || job.createdAt) > nowIso) return false;
|
|
return !job.lockedAt || job.lockedAt < staleBefore;
|
|
}
|
|
|
|
function sortClaimableJobs(a: GenerationJob, b: GenerationJob): number {
|
|
const priority = (b.priority || 0) - (a.priority || 0);
|
|
if (priority !== 0) return priority;
|
|
const scheduled = (a.scheduledAt || a.createdAt).localeCompare(b.scheduledAt || b.createdAt);
|
|
if (scheduled !== 0) return scheduled;
|
|
return a.createdAt.localeCompare(b.createdAt);
|
|
}
|
|
|
|
function assetToRow(asset: Partial<Asset>) {
|
|
return {
|
|
id: asset.id,
|
|
owner_id: asset.ownerId,
|
|
kind: asset.kind,
|
|
name: asset.name,
|
|
url: asset.url,
|
|
storage_path: asset.storagePath,
|
|
source: asset.source,
|
|
tags: asset.tags,
|
|
metadata: asset.metadata,
|
|
created_at: asset.createdAt,
|
|
updated_at: asset.updatedAt
|
|
};
|
|
}
|
|
|
|
function assetFromRow(row: Record<string, unknown>): Asset {
|
|
return {
|
|
id: String(row.id),
|
|
ownerId: String(row.owner_id),
|
|
kind: row.kind as Asset["kind"],
|
|
name: String(row.name),
|
|
url: String(row.url),
|
|
storagePath: row.storage_path ? String(row.storage_path) : undefined,
|
|
source: row.source as Asset["source"],
|
|
tags: Array.isArray(row.tags) ? row.tags.map(String) : [],
|
|
metadata: isRecord(row.metadata) ? row.metadata : {},
|
|
createdAt: requiredTimestamp(row.created_at, "assets.created_at"),
|
|
updatedAt: requiredTimestamp(row.updated_at, "assets.updated_at")
|
|
};
|
|
}
|
|
|
|
function jobToRow(job: Partial<GenerationJob>) {
|
|
const row: Record<string, unknown> = {};
|
|
if (job.id !== undefined) row.id = job.id;
|
|
if (job.ownerId !== undefined) row.owner_id = job.ownerId;
|
|
if (job.externalClientId !== undefined) row.external_client_id = job.externalClientId;
|
|
if (job.capability !== undefined) row.capability = job.capability;
|
|
if (job.provider !== undefined) row.provider = job.provider;
|
|
if (job.reqKey !== undefined) row.req_key = job.reqKey;
|
|
if (job.status !== undefined) row.status = job.status;
|
|
if (job.prompt !== undefined) row.prompt = job.prompt;
|
|
if (job.inputAssetIds !== undefined) row.input_asset_ids = job.inputAssetIds;
|
|
if (job.inputUrls !== undefined) row.input_urls = job.inputUrls;
|
|
if (job.outputAssetIds !== undefined) row.output_asset_ids = job.outputAssetIds;
|
|
if (job.providerTaskId !== undefined) row.provider_task_id = job.providerTaskId;
|
|
if (job.requestPayload !== undefined) row.request_payload = job.requestPayload;
|
|
if (job.responsePayload !== undefined) row.response_payload = job.responsePayload;
|
|
if (job.error !== undefined) row.error = job.error;
|
|
if (job.retryOf !== undefined) row.retry_of = job.retryOf;
|
|
if (job.idempotencyKey !== undefined) row.idempotency_key = job.idempotencyKey;
|
|
if (job.idempotencyFingerprint !== undefined) row.idempotency_fingerprint = job.idempotencyFingerprint;
|
|
if (job.priority !== undefined) row.priority = job.priority;
|
|
if (job.attempts !== undefined) row.attempts = job.attempts;
|
|
if (job.maxAttempts !== undefined) row.max_attempts = job.maxAttempts;
|
|
if (job.scheduledAt !== undefined) row.scheduled_at = job.scheduledAt;
|
|
if (job.lockedAt !== undefined) row.locked_at = job.lockedAt;
|
|
if (job.lockedBy !== undefined) row.locked_by = job.lockedBy;
|
|
if (job.startedAt !== undefined) row.started_at = job.startedAt;
|
|
if (job.completedAt !== undefined) row.completed_at = job.completedAt;
|
|
if (job.webhookUrl !== undefined) row.webhook_url = job.webhookUrl;
|
|
if (job.webhookAttempts !== undefined) row.webhook_attempts = job.webhookAttempts;
|
|
if (job.webhookLastStatus !== undefined) row.webhook_last_status = job.webhookLastStatus;
|
|
if (job.usageContext !== undefined) row.usage_context = job.usageContext;
|
|
if (job.billing !== undefined) row.billing = job.billing;
|
|
if (job.createdAt !== undefined) row.created_at = job.createdAt;
|
|
if (job.updatedAt !== undefined) row.updated_at = job.updatedAt;
|
|
return row;
|
|
}
|
|
|
|
function jobFromRow(row: Record<string, unknown>): GenerationJob {
|
|
return {
|
|
id: String(row.id),
|
|
ownerId: String(row.owner_id),
|
|
externalClientId: optionalString(row.external_client_id),
|
|
capability: row.capability as GenerationJob["capability"],
|
|
provider: row.provider as GenerationJob["provider"],
|
|
reqKey: String(row.req_key),
|
|
status: row.status as GenerationJob["status"],
|
|
prompt: row.prompt ? String(row.prompt) : undefined,
|
|
inputAssetIds: Array.isArray(row.input_asset_ids) ? row.input_asset_ids.map(String) : [],
|
|
inputUrls: Array.isArray(row.input_urls) ? row.input_urls.map(String) : [],
|
|
outputAssetIds: Array.isArray(row.output_asset_ids) ? row.output_asset_ids.map(String) : [],
|
|
providerTaskId: row.provider_task_id ? String(row.provider_task_id) : undefined,
|
|
requestPayload: isRecord(row.request_payload) ? row.request_payload : {},
|
|
responsePayload: isRecord(row.response_payload) ? row.response_payload : undefined,
|
|
error: isRecord(row.error) ? { message: String(row.error.message || "Unknown error"), code: row.error.code as string | number | undefined, retryable: Boolean(row.error.retryable) } : undefined,
|
|
retryOf: row.retry_of ? String(row.retry_of) : undefined,
|
|
idempotencyKey: optionalString(row.idempotency_key),
|
|
idempotencyFingerprint: optionalString(row.idempotency_fingerprint),
|
|
priority: optionalNumber(row.priority),
|
|
attempts: optionalNumber(row.attempts),
|
|
maxAttempts: optionalNumber(row.max_attempts),
|
|
scheduledAt: optionalTimestamp(row.scheduled_at),
|
|
lockedAt: optionalTimestamp(row.locked_at),
|
|
lockedBy: optionalString(row.locked_by),
|
|
startedAt: optionalTimestamp(row.started_at),
|
|
completedAt: optionalTimestamp(row.completed_at),
|
|
webhookUrl: optionalString(row.webhook_url),
|
|
webhookAttempts: optionalNumber(row.webhook_attempts),
|
|
webhookLastStatus: isRecord(row.webhook_last_status)
|
|
? {
|
|
ok: Boolean(row.webhook_last_status.ok),
|
|
status: optionalNumber(row.webhook_last_status.status),
|
|
error: optionalString(row.webhook_last_status.error),
|
|
attemptedAt: String(row.webhook_last_status.attemptedAt || row.webhook_last_status.attempted_at || ""),
|
|
nextAttemptAt: optionalString(row.webhook_last_status.nextAttemptAt || row.webhook_last_status.next_attempt_at)
|
|
}
|
|
: undefined,
|
|
usageContext: usageContextFromValue(row.usage_context),
|
|
billing: billingJobChargeFromValue(row.billing),
|
|
createdAt: requiredTimestamp(row.created_at, "generation_jobs.created_at"),
|
|
updatedAt: requiredTimestamp(row.updated_at, "generation_jobs.updated_at")
|
|
};
|
|
}
|
|
|
|
function usageToRow(usage: UsageEvent) {
|
|
return {
|
|
id: usage.id,
|
|
owner_id: usage.ownerId,
|
|
job_id: usage.jobId,
|
|
source: usage.source || "platform",
|
|
capability: usage.capability,
|
|
provider: usage.provider,
|
|
req_key: usage.reqKey,
|
|
account_username: usage.accountUsername,
|
|
account_display_name: usage.accountDisplayName,
|
|
tenant_id: usage.tenantId,
|
|
organization_id: usage.organizationId,
|
|
organization_name: usage.organizationName,
|
|
quantity: usage.quantity,
|
|
estimated_unit: usage.estimatedUnit,
|
|
charged_amount_fen: usage.chargedAmountFen,
|
|
currency: usage.currency,
|
|
created_at: usage.createdAt
|
|
};
|
|
}
|
|
|
|
function usageFromRow(row: Record<string, unknown>): UsageEvent {
|
|
return {
|
|
id: String(row.id),
|
|
ownerId: String(row.owner_id),
|
|
jobId: String(row.job_id),
|
|
source: row.source === "api" ? "api" : "platform",
|
|
capability: row.capability as UsageEvent["capability"],
|
|
provider: optionalString(row.provider) as UsageEvent["provider"],
|
|
reqKey: optionalString(row.req_key),
|
|
accountUsername: optionalString(row.account_username),
|
|
accountDisplayName: optionalString(row.account_display_name),
|
|
tenantId: optionalString(row.tenant_id),
|
|
organizationId: optionalString(row.organization_id),
|
|
organizationName: optionalString(row.organization_name),
|
|
quantity: Number(row.quantity || 0),
|
|
estimatedUnit: row.estimated_unit === "video_second" || row.estimated_unit === "image" ? row.estimated_unit : "job",
|
|
chargedAmountFen: optionalNumber(row.charged_amount_fen),
|
|
currency: row.currency === "CNY" ? "CNY" : undefined,
|
|
createdAt: requiredTimestamp(row.created_at, "usage_events.created_at")
|
|
};
|
|
}
|
|
|
|
async function findDatabaseUsageEventByJobId(jobId: string): Promise<UsageEvent | null> {
|
|
const { rows } = await queryDatabase("SELECT * FROM usage_events WHERE job_id = $1 LIMIT 1", [jobId]);
|
|
return rows[0] ? usageFromRow(rows[0]) : null;
|
|
}
|
|
|
|
function enrichLegacyUsageEvent(event: UsageEvent, job?: GenerationJob): UsageEvent {
|
|
const source = event.source || (job?.externalClientId || event.ownerId.startsWith("api:") ? "api" : "platform");
|
|
return {
|
|
...event,
|
|
source,
|
|
provider: event.provider || job?.provider,
|
|
reqKey: event.reqKey || job?.reqKey,
|
|
accountUsername: event.accountUsername || job?.usageContext?.username,
|
|
accountDisplayName: event.accountDisplayName || job?.usageContext?.displayName,
|
|
tenantId: event.tenantId || job?.usageContext?.tenantId,
|
|
organizationId: event.organizationId || job?.usageContext?.organizationId,
|
|
organizationName: event.organizationName || job?.usageContext?.organizationName,
|
|
quantity: 1,
|
|
estimatedUnit: "job"
|
|
};
|
|
}
|
|
|
|
function dedupeUsageEvents(events: UsageEvent[]): UsageEvent[] {
|
|
const seen = new Set<string>();
|
|
return events.filter((event) => {
|
|
if (seen.has(event.jobId)) return false;
|
|
seen.add(event.jobId);
|
|
return true;
|
|
});
|
|
}
|
|
|
|
function usageContextFromValue(value: unknown): UsageContext | undefined {
|
|
if (!isRecord(value) || value.source !== "platform" && value.source !== "api") return undefined;
|
|
const accountId = optionalString(value.accountId);
|
|
const displayName = optionalString(value.displayName);
|
|
if (!accountId || !displayName) return undefined;
|
|
return {
|
|
source: value.source,
|
|
accountId,
|
|
username: optionalString(value.username),
|
|
displayName,
|
|
role: value.role === "super_admin" || value.role === "organization_admin" || value.role === "user" ? value.role : undefined,
|
|
tenantId: optionalString(value.tenantId),
|
|
organizationId: optionalString(value.organizationId),
|
|
organizationName: optionalString(value.organizationName)
|
|
};
|
|
}
|
|
|
|
function billingJobChargeFromValue(value: unknown): GenerationJob["billing"] {
|
|
if (!isRecord(value)) return undefined;
|
|
if (value.currency !== "CNY") return undefined;
|
|
if (value.status !== "not_charged" && value.status !== "pending" && value.status !== "charged" && value.status !== "refunded") return undefined;
|
|
if (typeof value.priceRuleId !== "string" || typeof value.provider !== "string" || typeof value.capability !== "string" || typeof value.reqKey !== "string") return undefined;
|
|
if (value.unit !== "request" && value.unit !== "image" && value.unit !== "video_second") return undefined;
|
|
const quantity = optionalNumber(value.quantity);
|
|
const standardUnitPriceFen = optionalNumber(value.standardUnitPriceFen);
|
|
const markupMultiplier = optionalNumber(value.markupMultiplier);
|
|
const amountFen = optionalNumber(value.amountFen);
|
|
if (quantity === undefined || standardUnitPriceFen === undefined || markupMultiplier === undefined || amountFen === undefined) return undefined;
|
|
return {
|
|
priceRuleId: value.priceRuleId,
|
|
provider: value.provider as GenerationJob["provider"],
|
|
capability: value.capability as GenerationJob["capability"],
|
|
reqKey: value.reqKey,
|
|
variantKey: optionalString(value.variantKey),
|
|
unit: value.unit,
|
|
quantity,
|
|
standardUnitPriceFen,
|
|
markupMultiplier,
|
|
amountFen,
|
|
currency: "CNY",
|
|
conditions: isRecord(value.conditions) ? value.conditions as BillingRuleConditions : undefined,
|
|
quantitySource: value.quantitySource === "request" || value.quantitySource === "image_count" || value.quantitySource === "duration" ? value.quantitySource as BillingQuantitySource : undefined,
|
|
parameters: isRecord(value.parameters) ? value.parameters as BillingParameterSnapshot : undefined,
|
|
baseStandardUnitPriceFen: optionalNumber(value.baseStandardUnitPriceFen),
|
|
parameterTiers: Array.isArray(value.parameterTiers) ? value.parameterTiers as BillingSelectedParameterTier[] : undefined,
|
|
source: isRecord(value.source) ? value.source as BillingPriceSource : undefined,
|
|
quotaExempt: value.quotaExempt === true,
|
|
status: value.status,
|
|
reservedAmountFen: optionalNumber(value.reservedAmountFen),
|
|
settlementStatus: value.settlementStatus === "pending" || value.settlementStatus === "settled" || value.settlementStatus === "estimated" ? value.settlementStatus : undefined,
|
|
settlementLedgerEntryId: optionalString(value.settlementLedgerEntryId),
|
|
settledAt: optionalString(value.settledAt),
|
|
settlementReason: optionalString(value.settlementReason),
|
|
providerUsage: isRecord(value.providerUsage)
|
|
&& Number.isFinite(Number(value.providerUsage.completionTokens))
|
|
&& typeof value.providerUsage.resolution === "string"
|
|
&& typeof value.providerUsage.inputVideo === "boolean"
|
|
&& Number.isFinite(Number(value.providerUsage.tokenPriceFenPerMillion))
|
|
? {
|
|
completionTokens: Number(value.providerUsage.completionTokens),
|
|
resolution: value.providerUsage.resolution,
|
|
inputVideo: value.providerUsage.inputVideo,
|
|
tokenPriceFenPerMillion: Number(value.providerUsage.tokenPriceFenPerMillion)
|
|
}
|
|
: undefined,
|
|
ledgerEntryId: optionalString(value.ledgerEntryId),
|
|
refundLedgerEntryId: optionalString(value.refundLedgerEntryId),
|
|
chargedAt: optionalString(value.chargedAt),
|
|
refundedAt: optionalString(value.refundedAt),
|
|
refundReason: optionalString(value.refundReason)
|
|
};
|
|
}
|
|
|
|
function imageTemplateToRow(template: Partial<ImageTemplate>) {
|
|
const row: Record<string, unknown> = {};
|
|
if (template.id !== undefined) row.id = template.id;
|
|
if (template.ownerId !== undefined) row.owner_id = template.ownerId;
|
|
if (template.name !== undefined) row.name = template.name;
|
|
if (template.description !== undefined) row.description = template.description || null;
|
|
if (template.prompt !== undefined) row.prompt = template.prompt;
|
|
if (template.previewImageUrl !== undefined) row.preview_image_url = template.previewImageUrl || null;
|
|
if (template.settings !== undefined) row.settings = template.settings;
|
|
if (template.sortOrder !== undefined) row.sort_order = template.sortOrder;
|
|
if (template.createdAt !== undefined) row.created_at = template.createdAt;
|
|
if (template.updatedAt !== undefined) row.updated_at = template.updatedAt;
|
|
return row;
|
|
}
|
|
|
|
function imageTemplateFromRow(row: Record<string, unknown>): ImageTemplate {
|
|
return {
|
|
id: String(row.id),
|
|
ownerId: String(row.owner_id),
|
|
name: String(row.name || ""),
|
|
description: optionalString(row.description),
|
|
prompt: String(row.prompt || ""),
|
|
previewImageUrl: optionalString(row.preview_image_url),
|
|
settings: isRecord(row.settings) ? {
|
|
engine: row.settings.engine === "jimeng" || row.settings.engine === "evolink" || row.settings.engine === "bailian" ? row.settings.engine : undefined,
|
|
width: optionalNumber(row.settings.width),
|
|
height: optionalNumber(row.settings.height),
|
|
forceSingle: typeof row.settings.forceSingle === "boolean" ? row.settings.forceSingle : undefined,
|
|
scale: optionalNumber(row.settings.scale),
|
|
quality: row.settings.quality === "low" || row.settings.quality === "medium" || row.settings.quality === "high" ? row.settings.quality : undefined
|
|
} : {},
|
|
sortOrder: optionalNumber(row.sort_order) ?? 0,
|
|
createdAt: requiredTimestamp(row.created_at, "image_templates.created_at"),
|
|
updatedAt: requiredTimestamp(row.updated_at, "image_templates.updated_at")
|
|
};
|
|
}
|
|
|
|
function projectFromRow(row: Record<string, unknown>): Project {
|
|
return {
|
|
id: String(row.id),
|
|
ownerId: String(row.owner_id),
|
|
name: String(row.name || ""),
|
|
brief: String(row.brief || ""),
|
|
type: isProjectType(row.type) ? row.type : "custom",
|
|
assetIds: Array.isArray(row.asset_ids) ? row.asset_ids.map(String) : [],
|
|
createdAt: requiredTimestamp(row.created_at, "projects.created_at"),
|
|
updatedAt: requiredTimestamp(row.updated_at, "projects.updated_at")
|
|
};
|
|
}
|
|
|
|
function isProjectType(value: unknown): value is Project["type"] {
|
|
return value === "brand" || value === "store" || value === "commerce" || value === "event" || value === "course" || value === "ip" || value === "custom";
|
|
}
|
|
|
|
async function insertRow(table: string, row: Record<string, unknown>): Promise<Record<string, unknown>> {
|
|
const entries = Object.entries(row);
|
|
const columns = entries.map(([column]) => column).join(", ");
|
|
const placeholders = entries.map((_, index) => `$${index + 1}`).join(", ");
|
|
const { rows } = await queryDatabase(`INSERT INTO ${table} (${columns}) VALUES (${placeholders}) RETURNING *`, entries.map(([, value]) => value));
|
|
return rows[0];
|
|
}
|
|
|
|
async function updateRow(table: string, id: string, row: Record<string, unknown>): Promise<Record<string, unknown>> {
|
|
const entries = Object.entries(row);
|
|
if (!entries.length) {
|
|
const { rows } = await queryDatabase(`SELECT * FROM ${table} WHERE id = $1`, [id]);
|
|
if (!rows[0]) throw new Error(`${table} row not found: ${id}`);
|
|
return rows[0];
|
|
}
|
|
const values = entries.map(([, value]) => value);
|
|
values.push(id);
|
|
const assignments = entries.map(([column], index) => `${column} = $${index + 1}`).join(", ");
|
|
const { rows } = await queryDatabase(`UPDATE ${table} SET ${assignments} WHERE id = $${values.length} RETURNING *`, values);
|
|
if (!rows[0]) throw new Error(`${table} row not found: ${id}`);
|
|
return rows[0];
|
|
}
|
|
|
|
function addFilter(clauses: string[], values: unknown[], column: string, value: unknown): void {
|
|
if (value === undefined || value === null || value === "") return;
|
|
values.push(value);
|
|
clauses.push(`${column} = $${values.length}`);
|
|
}
|
|
|
|
function isUniqueViolation(error: unknown): boolean {
|
|
return isRecord(error) && error.code === "23505";
|
|
}
|
|
|
|
function isRecord(value: unknown): value is Record<string, unknown> {
|
|
return typeof value === "object" && value !== null && !Array.isArray(value);
|
|
}
|
|
|
|
function optionalString(value: unknown): string | undefined {
|
|
if (typeof value !== "string") return undefined;
|
|
const trimmed = value.trim();
|
|
return trimmed || undefined;
|
|
}
|
|
|
|
function requiredTimestamp(value: unknown, field: string): string {
|
|
const timestamp = optionalTimestamp(value);
|
|
if (!timestamp) throw new Error(`Invalid PostgreSQL timestamp: ${field}`);
|
|
return timestamp;
|
|
}
|
|
|
|
function optionalTimestamp(value: unknown): string | undefined {
|
|
if (value === undefined || value === null || value === "") return undefined;
|
|
if (value instanceof Date) return Number.isNaN(value.getTime()) ? undefined : value.toISOString();
|
|
if (typeof value !== "string") return undefined;
|
|
const parsed = new Date(value);
|
|
return Number.isNaN(parsed.getTime()) ? undefined : parsed.toISOString();
|
|
}
|
|
|
|
function optionalNumber(value: unknown): number | undefined {
|
|
if (value === undefined || value === null || value === "") return undefined;
|
|
const parsed = Number(value);
|
|
return Number.isFinite(parsed) ? parsed : undefined;
|
|
}
|