feat: add local auth billing and usage management

lib/server/data-store.ts

@@ -1,7 +1,7 @@
 import { readFile, rename, writeFile } from "node:fs/promises";
 import { join } from "node:path";
 import { createClient, type SupabaseClient } from "@supabase/supabase-js";
-import type { AppState, Asset, GenerationCapability, GenerationJob, GenerationStatus, ImageTemplate, Project, UsageEvent } from "@/lib/types";
+import type { AppState, Asset, BillingParameterSnapshot, BillingPriceSource, BillingQuantitySource, BillingRuleConditions, BillingSelectedParameterTier, GenerationCapability, GenerationJob, GenerationStatus, ImageTemplate, Project, UsageContext, UsageEvent, UsageSource } from "@/lib/types";
 import { createId } from "@/lib/server/ids";
 import { dataDir, DEFAULT_OWNER_ID, ensureRuntimeDirs } from "@/lib/server/runtime";
 
@@ -22,6 +22,13 @@ export type GenerationJobListFilters = {
   before?: string;
 };
 
+export type UsageEventListFilters = {
+  ownerId?: string;
+  source?: UsageSource;
+  from?: string;
+  to?: string;
+};
+
 export type ClaimGenerationJobsInput = {
   workerId: string;
   limit?: number;
@@ -303,15 +310,12 @@ export async function deleteGenerationJob(id: string): Promise<GenerationJob | n
   if (!existing) return null;
   const supabase = getSupabaseAdmin();
   if (supabase) {
-    const { error: usageError } = await supabase.from("usage_events").delete().eq("job_id", id);
-    if (usageError) throw new Error(usageError.message);
     const { error } = await supabase.from("generation_jobs").delete().eq("id", id);
     if (error) throw new Error(error.message);
     return existing;
   }
   return mutateLocalState((state) => {
     state.generationJobs = state.generationJobs.filter((job) => job.id !== id);
-    state.usageEvents = state.usageEvents.filter((event) => event.jobId !== id);
     return existing;
   });
 }
@@ -325,16 +329,79 @@ export async function recordUsageEvent(input: UsageInput): Promise<UsageEvent> {
   };
   const supabase = getSupabaseAdmin();
   if (supabase) {
+    const existing = await findSupabaseUsageEventByJobId(supabase, usage.jobId);
+    if (existing) return existing;
     const { data, error } = await supabase.from("usage_events").insert(usageToRow(usage)).select("*").single();
+    if (error?.code === "23505") {
+      const raced = await findSupabaseUsageEventByJobId(supabase, usage.jobId);
+      if (raced) return raced;
+    }
     if (error) throw new Error(error.message);
     return usageFromRow(data);
   }
   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[]> {
+  const supabase = getSupabaseAdmin();
+  if (supabase) {
+    const rows: Record<string, unknown>[] = [];
+    const pageSize = 1000;
+    for (let offset = 0; ; offset += pageSize) {
+      let query = supabase
+        .from("usage_events")
+        .select("*")
+        .order("created_at", { ascending: false })
+        .range(offset, offset + pageSize - 1);
+      if (filters.ownerId) query = query.eq("owner_id", filters.ownerId);
+      if (filters.source) query = query.eq("source", filters.source);
+      if (filters.from) query = query.gte("created_at", filters.from);
+      if (filters.to) query = query.lt("created_at", filters.to);
+      const { data, error } = await query;
+      if (error) throw new Error(error.message);
+      const page = (data || []) as Record<string, unknown>[];
+      rows.push(...page);
+      if (page.length < pageSize) break;
+    }
+    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[]> {
   const state = await readState();
   return state.projects.filter((project) => project.ownerId === ownerId).sort(sortNewest);
@@ -442,6 +509,24 @@ export async function deleteImageTemplate(id: string, ownerId: string): Promise<
   });
 }
 
+export async function reassignOwnerData(fromOwnerId: string, toOwnerId: string): Promise<void> {
+  if (!fromOwnerId || !toOwnerId || fromOwnerId === toOwnerId) return;
+  const supabase = getSupabaseAdmin();
+  if (supabase) {
+    for (const table of ["assets", "generation_jobs", "projects", "image_templates"] as const) {
+      const { error } = await supabase.from(table).update({ owner_id: toOwnerId }).eq("owner_id", fromOwnerId);
+      if (error) throw new Error(error.message);
+    }
+    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);
@@ -582,6 +667,8 @@ function jobToRow(job: Partial<GenerationJob>) {
   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;
@@ -626,6 +713,8 @@ function jobFromRow(row: Record<string, unknown>): GenerationJob {
           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: String(row.created_at),
     updatedAt: String(row.updated_at)
   };
@@ -636,9 +725,19 @@ function usageToRow(usage: UsageEvent) {
     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
   };
 }
@@ -648,13 +747,128 @@ function usageFromRow(row: Record<string, unknown>): UsageEvent {
     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 as UsageEvent["estimatedUnit"],
+    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: String(row.created_at)
   };
 }
 
+async function findSupabaseUsageEventByJobId(supabase: SupabaseClient, jobId: string): Promise<UsageEvent | null> {
+  const { data, error } = await supabase.from("usage_events").select("*").eq("job_id", jobId).maybeSingle();
+  if (error) throw new Error(error.message);
+  return data ? usageFromRow(data as Record<string, unknown>) : 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;