nekonest-cloud/db/repository.ts

import { env } from "cloudflare:workers";
import { ensureDatabase, getD1 } from "./bootstrap";
import { addBillingPeriod, summarizeEntitlementComponents } from "./domain";
import { DomainError } from "./domain-error";
export { DomainError } from "./domain-error";
import { isWritableRelayPlacement } from "./relay-routing";
import {
  ADVANCE_TENANT_CREDENTIAL_AFTER_CLAIM_SQL,
  CANCEL_PAIRING_SQL,
  CLAIM_CREDENTIAL_SQL,
  CLAIM_HOST_SQL,
  COMMIT_PAIRING_CLAIM_SQL,
  CONSUME_CLAIM_RATE_SQL,
  derivePairingProgress,
  OWNED_PAIRING_PROGRESS_SQL,
  RECORD_FAILED_CODE_SQL,
  RESERVE_PAIRING_SQL,
  type PairingProgress,
  type PairingProgressRow,
} from "./pairing";
import {
  constantTimeEqualHex,
  parseBootstrapToken,
  randomDeviceToken,
  rateWindowStart,
  sha256Hex,
  sourceFingerprint,
  validateDeviceIdentity,
  verifyDeviceRegistrationProof,
} from "./device-claim";
import {
  ADMIN_HOST_CONTROL_PLANE_ATTENTION_SQL,
  ADMIN_HOST_CONTROL_PLANE_SUMMARY_SQL,
  controlPlaneContactCutoffs,
  deriveAdminHostControlPlaneSnapshot,
  OWNED_HOSTS_WITH_CONTROL_PLANE_CONTACT_SQL,
  type AdminHostControlPlaneAttentionRow,
  type AdminHostControlPlaneSnapshot,
} from "./device-control-plane";
import {
  classifyReportedDaemonVersion,
  MINIMUM_CLOUD_DAEMON_VERSION,
} from "../release/daemon-version.ts";
import {
  AUDIT_INVITATION_REVOCATION_SQL,
  CREATE_INVITATION_REVOCATION_IDEMPOTENCY_SQL,
  REVOKE_INVITATION_SQL,
} from "./invitations.ts";
import {
  APPROVE_ACCESS_REQUEST_SQL,
  CANCEL_ACCESS_REQUEST_SQL,
  CREATE_ACCESS_CANCELLATION_IDEMPOTENCY_SQL,
  CREATE_ACCESS_REQUEST_IDEMPOTENCY_SQL,
  CREATE_ACCESS_REQUEST_SQL,
  CREATE_ACCESS_RESOLUTION_IDEMPOTENCY_SQL,
  CREATE_APPROVED_INVITATION_SQL,
  CREATE_MANUAL_INVITATION_IDEMPOTENCY_SQL,
  CREATE_MANUAL_INVITATION_SQL,
  DECLINE_ACCESS_REQUEST_SQL,
  AUDIT_PUBLIC_BETA_ACCESS_FULFILLMENT_SQL,
  FULFILL_ACCESS_REQUESTS_BY_PUBLIC_BETA_SQL,
  PUBLIC_BETA_ACCESS_RESPONSE,
  type BetaAccessRequestStatus,
} from "./access-requests.ts";
import {
  DEACTIVATE_OWNED_HOST_SQL,
  REVOKE_ACTIVE_DEVICE_CREDENTIALS_SQL,
} from "./access-boundary.ts";
import {
  DELETE_EXPIRED_CLAIM_ATTEMPTS_SQL,
  DELETE_EXPIRED_CLAIM_RATE_WINDOWS_SQL,
  DELETE_EXPIRED_IDEMPOTENCY_SQL,
  DELETE_EXPIRED_DEVICE_REGISTRATION_REPLAYS_SQL,
  DELETE_RETIRED_PENDING_PHONE_PRINCIPALS_SQL,
  DELETE_RETIRED_PENDING_PHONE_ROUTES_SQL,
  DELETE_RETIRED_PHONE_HANDOFF_TICKETS_SQL,
  deriveRetentionJobHealth,
  getRetentionCutoffs,
  RETENTION_JOB_KEY,
  RETIRE_EXPIRED_PAIRING_CODES_SQL,
  type RetentionJobHealth,
  type RetentionJobRecord,
} from "./retention";
import {
  deriveServiceStatus,
  type IncidentSeverity,
  type ServiceStatusLevel,
} from "./service-status";
import {
  BETA_ACCOUNTS_SQL,
  BETA_ACCESS_REQUESTS_SQL,
  BETA_CLAIM_ATTEMPTS_SQL,
  BETA_OPERATIONS_WINDOW_DAYS,
  BETA_PAIRINGS_SQL,
  BETA_PROVISIONING_SQL,
  BETA_SUPPORT_SQL,
  deriveBetaOperationsSnapshot,
  type BetaOperationsSnapshot,
} from "./beta-operations";
import {
  countBlockedPublicBetaP0,
} from "./launch-gates.ts";
import {
  ADVANCE_AUTHORIZATION_AFTER_CLAIM_SQL,
  ADVANCE_AUTHORIZATION_AFTER_REVOKE_SQL,
} from "./relay-control-sql.ts";
import {
  DEVICE_REGISTRATION_REPLAY_TTL_MS,
  decryptRegistrationReplay,
  encryptRegistrationReplay,
} from "./registration-replay.ts";

export type CloudIdentity = {
  userId: string;
  email: string;
  displayName: string;
};

export type AccountRecord = {
  id: string;
  auth_subject: string;
  email: string;
  display_name: string;
  status: string;
  created_at: string;
};

export type HostRecord = {
  id: string;
  account_id: string;
  name: string;
  os: string;
  lifecycle: string;
  slot_state: string;
  connection_state: string;
  daemon_version: string | null;
  identity_fingerprint: string | null;
  claimed_at: string | null;
  last_seen_at: string | null;
  control_plane_last_seen_at: string | null;
  created_at: string;
};

export type PairingRecord = {
  id: string;
  requested_name: string;
  os: string;
  status: string;
  expires_at: string;
  created_at: string;
};

export type DeviceRegistrationResult = {
  device_id: string;
  token: string;
  name: string;
  transport_mode: "sealed";
  connection_state: "ready" | "provisioning";
  retry_after_seconds?: number;
};

export type PriceRecord = {
  id: string;
  billing_period: "month" | "year";
  amount_minor: number;
  currency: "CNY";
  tax_mode: string;
  status: string;
  effective_from: string;
};

export type BetaRecord = {
  id: string;
  state: string;
  capacity_slots: number | null;
  starts_at: string;
  ends_at: string | null;
  grace_days: number;
  created_at: string;
};

export type GrantRecord = {
  id: string;
  account_id: string;
  host_id: string | null;
  source: string;
  capacity_slots: number | null;
  starts_at: string;
  ends_at: string | null;
  state: string;
  reason: string;
  created_at: string;
  revoked_at: string | null;
};

export type BetaAccessRequestRecord = {
  id: string;
  account_id: string;
  status: BetaAccessRequestStatus;
  preferred_os: "windows" | "linux" | "both";
  requested_slots: number;
  use_case: string;
  admin_response: string | null;
  resolved_by: string | null;
  invitation_grant_id: string | null;
  requested_at: string;
  resolved_at: string | null;
  cancelled_at: string | null;
  created_at: string;
  updated_at: string;
};

export type TenantRecord = {
  id: string;
  slug: string;
  lifecycle: string;
  desired_state: string;
  observed_state: string;
  desired_generation: number;
  active_generation: number | null;
  credential_revision: number;
  runtime_version: string | null;
  relay_origin: string | null;
  relay_ready: number;
  last_health_at: string | null;
  tombstoned_at: string | null;
  updated_at: string;
};

export type TenantConnectionSummary = {
  state: "ready" | "provisioning" | "suspended" | "unavailable";
  homeRegion: string | null;
  placementGeneration: number | null;
  authorizationRevision: number;
  retryAfterSeconds: number | null;
};

export type OrderRecord = {
  id: string;
  client_order_id: string;
  billing_period: string;
  slot_quantity: number;
  amount_minor: number;
  currency: string;
  status: string;
  term_start: string;
  term_end: string;
  quote_expires_at: string;
  created_at: string;
};

export type LaunchGateRecord = {
  key: string;
  priority: "P0" | "P1" | "P2" | "PAID";
  category: string;
  title: string;
  status: "blocked" | "in_progress" | "passed" | "not_applicable";
  owner: string | null;
  evidence_url: string | null;
  notes: string;
  reviewed_at: string | null;
  updated_at: string;
};

export type AuditRecord = {
  id: string;
  actor_id: string;
  action: string;
  target_type: string;
  target_id: string;
  reason: string;
  created_at: string;
};

export type FeedbackCategory =
  | "connection_issue"
  | "bug"
  | "suggestion"
  | "other";

export type FeedbackRecord = {
  id: string;
  account_id: string;
  category: FeedbackCategory;
  message: string;
  status: "open" | "resolved";
  admin_response: string | null;
  handled_by: string | null;
  created_at: string;
  updated_at: string;
  resolved_at: string | null;
};

export type AccountDeletionRequestRecord = {
  id: string;
  account_id: string;
  status: "requested" | "cancelled" | "processing" | "relay_purged";
  reason: string | null;
  requested_at: string;
  cancelled_at: string | null;
  created_at: string;
  updated_at: string;
};

export type ServiceIncidentRecord = {
  id: string;
  severity: IncidentSeverity;
  title: string;
  message: string;
  status: "active" | "resolved";
  created_by: string;
  resolved_by: string | null;
  resolution: string | null;
  started_at: string;
  resolved_at: string | null;
  created_at: string;
  updated_at: string;
};

export type PublicServiceIncidentRecord = Pick<
  ServiceIncidentRecord,
  | "id"
  | "severity"
  | "title"
  | "message"
  | "status"
  | "resolution"
  | "started_at"
  | "resolved_at"
  | "updated_at"
>;

export type ServiceStatusSnapshot = {
  status: ServiceStatusLevel;
  activeIncidents: PublicServiceIncidentRecord[];
  recentIncidents: PublicServiceIncidentRecord[];
  checkedAt: string;
};

export type EntitlementSummary = {
  mode: "public_beta" | "grant" | "none";
  publicBetaState: "open" | "gated" | "inactive";
  blockedP0: number;
  unlimited: boolean;
  capacitySlots: number | null;
  activeSlots: number;
  reservedSlots: number;
  availableSlots: number | null;
  effectiveUntil: string | null;
  sources: string[];
};

export type DashboardSnapshot = {
  account: AccountRecord;
  hosts: HostRecord[];
  pairings: PairingRecord[];
  beta: BetaRecord | null;
  grants: GrantRecord[];
  accessRequests: BetaAccessRequestRecord[];
  tenant: TenantRecord | null;
  connection: TenantConnectionSummary;
  entitlement: EntitlementSummary;
  serviceStatus: ServiceStatusSnapshot;
};

export type AdminSnapshot = {
  beta: BetaRecord | null;
  accounts: AccountRecord[];
  grants: GrantRecord[];
  accessRequests: BetaAccessRequestRecord[];
  gates: LaunchGateRecord[];
  audits: AuditRecord[];
  feedback: FeedbackRecord[];
  deletionRequests: AccountDeletionRequestRecord[];
  incidents: ServiceIncidentRecord[];
  serviceStatus: ServiceStatusSnapshot;
  operations: BetaOperationsSnapshot;
  hostContacts: AdminHostControlPlaneSnapshot;
  retention: RetentionJobHealth;
  blockedP0: number;
};

export type PublicCommercialSnapshot = {
  beta: BetaRecord | null;
  gates: LaunchGateRecord[];
  blockedP0: number;
};

export type RetentionMaintenanceResult = {
  completedAt: string;
  claimRateBefore: string;
  claimAttemptBefore: string;
  handoffTicketBefore: string;
  retiredPairingCodes: number;
  deletedClaimRateWindows: number;
  deletedClaimAttempts: number;
  deletedIdempotencyRecords: number;
  deletedPendingPhoneRoutes: number;
  deletedPendingPhonePrincipals: number;
  deletedPhoneHandoffTickets: number;
  deletedDeviceRegistrationReplays: number;
};

const launchGateStatuses = new Set<LaunchGateRecord["status"]>([
  "blocked",
  "in_progress",
  "passed",
  "not_applicable",
]);

const feedbackCategories = new Set<FeedbackCategory>([
  "connection_issue",
  "bug",
  "suggestion",
  "other",
]);

const incidentSeverities = new Set<IncidentSeverity>([
  "maintenance",
  "degraded",
  "outage",
]);

const paidFeaturesEnabled: boolean = false;

function assertPaidFeaturesDeferred(): void {
  if (!paidFeaturesEnabled) {
    throw new DomainError(
      "paid_features_deferred",
      "免费公测阶段不提供价格、报价或订单写入",
      409,
    );
  }
}

function newId(prefix: string): string {
  return `${prefix}_${crypto.randomUUID().replaceAll("-", "")}`;
}

function assertIdempotencyKey(value: string): void {
  if (!/^[A-Za-z0-9._:-]{8,128}$/.test(value)) {
    throw new DomainError(
      "invalid_idempotency_key",
      "幂等键需为 8 到 128 位字母、数字或 . _ : -",
    );
  }
}

function assertReason(value: string): void {
  const reason = value.trim();
  if (!reason) throw new DomainError("reason_required", "必须填写变更理由");
  if (reason.length > 500) {
    throw new DomainError("reason_too_long", "变更理由不能超过 500 个字符");
  }
}

function assertBillingPeriod(value: string): asserts value is "month" | "year" {
  if (value !== "month" && value !== "year") {
    throw new DomainError("invalid_period", "计费周期只能是月或年");
  }
}

async function hashText(value: string): Promise<string> {
  const bytes = new TextEncoder().encode(value);
  const digest = await crypto.subtle.digest("SHA-256", bytes);
  return Array.from(new Uint8Array(digest), (byte) =>
    byte.toString(16).padStart(2, "0"),
  ).join("");
}

function isoAfterMinutes(minutes: number): string {
  return new Date(Date.now() + minutes * 60_000).toISOString();
}

async function all<T>(statement: D1PreparedStatement): Promise<T[]> {
  const result = await statement.all<T>();
  return result.results ?? [];
}

export async function getOrCreateAccount(
  identity: CloudIdentity,
): Promise<AccountRecord> {
  await ensureDatabase();
  const db = getD1();
  const existing = await db
    .prepare("SELECT * FROM accounts WHERE auth_subject = ?")
    .bind(identity.userId)
    .first<AccountRecord>();

  if (existing) {
    if (
      existing.email !== identity.email ||
      existing.display_name !== identity.displayName
    ) {
      await db
        .prepare(
          `UPDATE accounts
           SET email = ?, display_name = ?, updated_at = ?
           WHERE id = ?`,
        )
        .bind(identity.email, identity.displayName, new Date().toISOString(), existing.id)
        .run();
      return { ...existing, email: identity.email, display_name: identity.displayName };
    }
    return existing;
  }

  // Stable IDs make concurrent first-login batches converge on the same
  // account, tenant, and creation audit instead of producing orphaned rows or
  // duplicate audit events after an INSERT OR IGNORE race.
  const identityHash = (await hashText(`account:${identity.userId}`)).slice(0, 32);
  const accountId = `acct_${identityHash}`;
  const tenantId = `tenant_${identityHash}`;
  const slug = `n-${identityHash}`;
  const now = new Date().toISOString();

  await db.batch([
    db
      .prepare(
        `INSERT OR IGNORE INTO accounts
          (id, auth_subject, email, display_name, status, created_at, updated_at)
         VALUES (?, ?, ?, ?, 'active', ?, ?)`,
      )
      .bind(
        accountId,
        identity.userId,
        identity.email,
        identity.displayName,
        now,
        now,
      ),
    db
      .prepare(
        `INSERT OR IGNORE INTO tenant_instances
          (id, account_id, slug, lifecycle, desired_generation, created_at, updated_at)
         VALUES (?, ?, ?, 'requested', 1, ?, ?)`,
      )
      .bind(tenantId, accountId, slug, now, now),
    db
      .prepare(
        `INSERT OR IGNORE INTO tenant_placements
          (tenant_id, home_region_id, generation, state, created_at, updated_at)
         VALUES (?, 'region_default', 1, 'provisioning', ?, ?)`,
      )
      .bind(tenantId, now, now),
    db
      .prepare(
        `INSERT OR IGNORE INTO tenant_authorization_state
          (tenant_id, revision, status, updated_at)
         VALUES (?, 0, 'active', ?)`,
      )
      .bind(tenantId, now),
    db
      .prepare(
        `INSERT OR IGNORE INTO audit_events
          (id, actor_id, action, target_type, target_id, reason,
           after_json, correlation_id, created_at)
         VALUES (?, ?, 'account.created', 'account', ?, '首次登录', ?, ?, ?)`,
      )
      .bind(
        `audit_account_${identityHash}`,
        identity.userId,
        accountId,
        JSON.stringify({ email: identity.email }),
        `corr_account_${identityHash}`,
        now,
      ),
  ]);

  const account = await db
    .prepare("SELECT * FROM accounts WHERE auth_subject = ?")
    .bind(identity.userId)
    .first<AccountRecord>();
  if (!account) throw new DomainError("account_create_failed", "账户创建失败", 500);
  return account;
}

export async function getActiveBeta(): Promise<BetaRecord | null> {
  await ensureDatabase();
  const now = new Date().toISOString();
  return getD1()
    .prepare(
      `SELECT * FROM beta_programs
       WHERE state = 'active'
         AND starts_at <= ?
         AND (ends_at IS NULL OR ends_at > ?)
       ORDER BY created_at DESC, id DESC
       LIMIT 1`,
    )
    .bind(now, now)
    .first<BetaRecord>();
}

async function getLatestBeta(): Promise<BetaRecord | null> {
  await ensureDatabase();
  return getD1()
    .prepare("SELECT * FROM beta_programs ORDER BY created_at DESC LIMIT 1")
    .first<BetaRecord>();
}

export async function evaluateEntitlement(
  accountId: string,
): Promise<EntitlementSummary> {
  await ensureDatabase();
  const db = getD1();
  const now = new Date().toISOString();
  const [betaPolicy, grants, activeCountRow, reservationCountRow, p0Gates] = await Promise.all([
    getActiveBeta(),
    all<GrantRecord>(
      db
        .prepare(
          `SELECT * FROM entitlement_grants
           WHERE account_id = ? AND state = 'active'
             AND starts_at <= ? AND (ends_at IS NULL OR ends_at > ?)
             AND revoked_at IS NULL
           ORDER BY created_at DESC`,
        )
        .bind(accountId, now, now),
    ),
    db
      .prepare(
        `SELECT COUNT(*) AS count FROM hosts
         WHERE account_id = ? AND lifecycle = 'active' AND slot_state = 'active'`,
      )
      .bind(accountId)
      .first<{ count: number }>(),
    db
      .prepare(
        `SELECT COUNT(*) AS count FROM pairing_requests
         WHERE account_id = ? AND status = 'waiting' AND expires_at > ?`,
      )
      .bind(accountId, now)
      .first<{ count: number }>(),
    all<LaunchGateRecord>(
      db.prepare("SELECT * FROM launch_gates WHERE priority = 'P0'"),
    ),
  ]);

  const blockedP0 = countBlockedPublicBetaP0(p0Gates);
  const beta = betaPolicy && blockedP0 === 0 ? betaPolicy : null;
  const activeSlots = Number(activeCountRow?.count ?? 0);
  const reservedSlots = Number(reservationCountRow?.count ?? 0);
  const components = [
    ...(beta
      ? [{ source: "public_beta", capacity: beta.capacity_slots, endsAt: beta.ends_at }]
      : []),
    ...grants.map((grant) => ({
      source: grant.source,
      capacity: grant.capacity_slots,
      endsAt: grant.ends_at,
    })),
  ];
  const summary = summarizeEntitlementComponents(
    components,
    activeSlots,
    reservedSlots,
  );

  return {
    mode: beta ? "public_beta" : grants.length ? "grant" : "none",
    publicBetaState: !betaPolicy ? "inactive" : blockedP0 ? "gated" : "open",
    blockedP0,
    unlimited: summary.unlimited,
    capacitySlots: summary.capacitySlots,
    activeSlots,
    reservedSlots,
    availableSlots: summary.availableSlots,
    effectiveUntil: summary.effectiveUntil,
    sources: summary.sources,
  };
}

export async function getDashboardSnapshot(
  account: AccountRecord,
): Promise<DashboardSnapshot> {
  await ensureDatabase();
  const db = getD1();
  const now = new Date().toISOString();
  const [hosts, pairings, beta, grants, accessRequests, tenant, connectionRow, entitlement, serviceStatus] =
    await Promise.all([
      all<HostRecord>(
        db
          .prepare(OWNED_HOSTS_WITH_CONTROL_PLANE_CONTACT_SQL)
          .bind(account.id),
      ),
      all<PairingRecord>(
        db
          .prepare(
            `SELECT id, requested_name, os, status, expires_at, created_at
             FROM pairing_requests
             WHERE account_id = ? AND status = 'waiting' AND expires_at > ?
             ORDER BY created_at DESC`,
          )
          .bind(account.id, now),
      ),
      getActiveBeta(),
      all<GrantRecord>(
        db
          .prepare(
            `SELECT * FROM entitlement_grants WHERE account_id = ?
             ORDER BY created_at DESC LIMIT 20`,
          )
          .bind(account.id),
      ),
      all<BetaAccessRequestRecord>(
        db
          .prepare(
            `SELECT * FROM beta_access_requests WHERE account_id = ?
             ORDER BY requested_at DESC, id DESC LIMIT 20`,
          )
          .bind(account.id),
      ),
      db
        .prepare("SELECT * FROM tenant_instances WHERE account_id = ?")
        .bind(account.id)
        .first<TenantRecord>(),
      db
        .prepare(
          `SELECT placements.state, placements.generation,
                  regions.code AS home_region, nodes.status AS node_status,
                  authorizations.status AS authorization_status,
                  authorizations.revision AS authorization_revision
           FROM tenant_instances AS tenants
           LEFT JOIN tenant_placements AS placements ON placements.tenant_id = tenants.id
           LEFT JOIN relay_regions AS regions ON regions.id = placements.home_region_id
           LEFT JOIN relay_nodes AS nodes ON nodes.id = placements.relay_node_id
           LEFT JOIN tenant_authorization_state AS authorizations ON authorizations.tenant_id = tenants.id
           WHERE tenants.account_id = ?`,
        )
        .bind(account.id)
        .first<{
          state: string | null;
          generation: number | null;
          home_region: string | null;
          node_status: string | null;
          authorization_status: string | null;
          authorization_revision: number | null;
        }>(),
      evaluateEntitlement(account.id),
      getServiceStatusSnapshot(),
    ]);

  const connection: TenantConnectionSummary = connectionRow?.authorization_status === "suspended"
    ? {
        state: "suspended",
        homeRegion: connectionRow.home_region,
        placementGeneration: connectionRow.generation,
        authorizationRevision: connectionRow.authorization_revision ?? 0,
        retryAfterSeconds: null,
      }
    : connectionRow?.state === "active" && connectionRow.node_status === "active"
      ? {
          state: "ready",
          homeRegion: connectionRow.home_region,
          placementGeneration: connectionRow.generation,
          authorizationRevision: connectionRow.authorization_revision ?? 0,
          retryAfterSeconds: null,
        }
      : connectionRow
        ? {
            state: "provisioning",
            homeRegion: connectionRow.home_region,
            placementGeneration: connectionRow.generation,
            authorizationRevision: connectionRow.authorization_revision ?? 0,
            retryAfterSeconds: 5,
          }
        : {
            state: "unavailable",
            homeRegion: null,
            placementGeneration: null,
            authorizationRevision: 0,
            retryAfterSeconds: null,
          };

  return {
    account,
    hosts,
    pairings,
    beta,
    grants,
    accessRequests,
    tenant,
    connection,
    entitlement,
    serviceStatus,
  };
}

const accessRequestOperatingSystems = new Set(["windows", "linux", "both"]);

export async function createBetaAccessRequest(input: {
  accountId: string;
  actorId: string;
  preferredOs: string;
  requestedSlots: number;
  useCase: string;
  idempotencyKey: string;
}): Promise<BetaAccessRequestRecord> {
  await ensureDatabase();
  if (!accessRequestOperatingSystems.has(input.preferredOs)) {
    throw new DomainError("invalid_preferred_os", "请选择计划接入的操作系统");
  }
  if (!Number.isInteger(input.requestedSlots) || input.requestedSlots < 1 || input.requestedSlots > 3) {
    throw new DomainError("invalid_requested_slots", "首批闭测申请只能填写 1 到 3 台主机");
  }
  const useCase = input.useCase.trim();
  if (useCase.length < 20 || useCase.length > 1_000) {
    throw new DomainError("invalid_access_use_case", "使用说明需为 20 到 1000 个字符");
  }
  assertIdempotencyKey(input.idempotencyKey);
  const entitlement = await evaluateEntitlement(input.accountId);
  if (entitlement.mode !== "none") {
    throw new DomainError("access_already_available", "该账户已经具备免费测试资格", 409);
  }

  const payload = {
    preferredOs: input.preferredOs,
    requestedSlots: input.requestedSlots,
    useCase,
  };
  const requestHash = await hashText(JSON.stringify(payload));
  const scope = `access-request:create:${input.accountId}`;
  const replay = await assertIdempotencyAvailable(scope, input.idempotencyKey, requestHash);
  if (replay.replay) return JSON.parse(replay.responseJson!) as BetaAccessRequestRecord;

  const now = new Date().toISOString();
  const accessRequest: BetaAccessRequestRecord = {
    id: newId("access"),
    account_id: input.accountId,
    status: "requested",
    preferred_os: input.preferredOs as BetaAccessRequestRecord["preferred_os"],
    requested_slots: input.requestedSlots,
    use_case: useCase,
    admin_response: null,
    resolved_by: null,
    invitation_grant_id: null,
    requested_at: now,
    resolved_at: null,
    cancelled_at: null,
    created_at: now,
    updated_at: now,
  };
  const responseJson = JSON.stringify(accessRequest);
  const db = getD1();
  const auditId = `audit_${(await hashText(`${scope}:${input.idempotencyKey}`)).slice(0, 32)}`;
  await db.batch([
    db
      .prepare(CREATE_ACCESS_REQUEST_IDEMPOTENCY_SQL)
      .bind(
        scope,
        input.idempotencyKey,
        requestHash,
        responseJson,
        isoAfterMinutes(24 * 60),
        now,
        input.accountId,
      ),
    db
      .prepare(CREATE_ACCESS_REQUEST_SQL)
      .bind(
        accessRequest.id,
        accessRequest.account_id,
        accessRequest.preferred_os,
        accessRequest.requested_slots,
        accessRequest.use_case,
        now,
        scope,
        input.idempotencyKey,
        requestHash,
      ),
    db
      .prepare(
        `INSERT OR IGNORE INTO audit_events
          (id, actor_id, action, target_type, target_id, reason,
           after_json, correlation_id, created_at)
         SELECT ?, ?, 'beta_access.requested', 'beta_access_request', ?,
                '用户申请免费闭测资格', ?, ?, ?
         FROM beta_access_requests
         WHERE id = ? AND status = 'requested' AND created_at = ?`,
      )
      .bind(
        auditId,
        input.actorId,
        accessRequest.id,
        JSON.stringify({ preferredOs: accessRequest.preferred_os, requestedSlots: accessRequest.requested_slots }),
        newId("corr"),
        now,
        accessRequest.id,
        now,
      ),
  ]);

  const [stored, current, audit] = await Promise.all([
    db
      .prepare("SELECT request_hash, response_json FROM idempotency_records WHERE scope = ? AND key = ?")
      .bind(scope, input.idempotencyKey)
      .first<{ request_hash: string; response_json: string }>(),
    db
      .prepare("SELECT * FROM beta_access_requests WHERE id = ?")
      .bind(accessRequest.id)
      .first<BetaAccessRequestRecord>(),
    db.prepare("SELECT id FROM audit_events WHERE id = ?").bind(auditId).first<{ id: string }>(),
  ]);
  if (stored?.request_hash === requestHash) {
    const committed = JSON.parse(stored.response_json) as BetaAccessRequestRecord;
    if (current?.id !== committed.id || current.status !== "requested" || audit?.id !== auditId) {
      throw new DomainError("access_request_indeterminate", "闭测申请结果需要人工核对", 500);
    }
    return committed;
  }
  if (stored) throw new DomainError("idempotency_conflict", "同一幂等键不能用于不同请求", 409);
  const pending = await db
    .prepare("SELECT id FROM beta_access_requests WHERE account_id = ? AND status = 'requested'")
    .bind(input.accountId)
    .first<{ id: string }>();
  if (pending) throw new DomainError("access_request_pending", "该账户已经有一条待处理闭测申请", 409);
  const currentEntitlement = await evaluateEntitlement(input.accountId);
  if (currentEntitlement.mode !== "none") {
    throw new DomainError("access_already_available", "该账户已经具备免费测试资格", 409);
  }
  throw new DomainError("access_request_failed", "闭测申请提交失败，请重试", 409);
}

export async function cancelBetaAccessRequest(input: {
  accountId: string;
  actorId: string;
  requestId: string;
  idempotencyKey: string;
}): Promise<BetaAccessRequestRecord> {
  await ensureDatabase();
  if (!/^access_[a-f0-9]{32}$/.test(input.requestId)) {
    throw new DomainError("invalid_access_request_id", "闭测申请编号无效");
  }
  assertIdempotencyKey(input.idempotencyKey);
  const requestHash = await hashText(JSON.stringify({ requestId: input.requestId }));
  const scope = `access-request:cancel:${input.accountId}`;
  const replay = await assertIdempotencyAvailable(scope, input.idempotencyKey, requestHash);
  if (replay.replay) return JSON.parse(replay.responseJson!) as BetaAccessRequestRecord;

  const db = getD1();
  const previous = await db
    .prepare("SELECT * FROM beta_access_requests WHERE id = ? AND account_id = ?")
    .bind(input.requestId, input.accountId)
    .first<BetaAccessRequestRecord>();
  if (!previous) throw new DomainError("access_request_not_found", "没有找到该闭测申请", 404);
  if (previous.status !== "requested") {
    throw new DomainError("access_request_not_pending", "该闭测申请已经处理", 409);
  }
  const now = new Date().toISOString();
  const cancelled: BetaAccessRequestRecord = {
    ...previous,
    status: "cancelled",
    cancelled_at: now,
    updated_at: now,
  };
  const responseJson = JSON.stringify(cancelled);
  const auditId = `audit_${(await hashText(`${scope}:${input.idempotencyKey}`)).slice(0, 32)}`;
  await db.batch([
    db
      .prepare(CREATE_ACCESS_CANCELLATION_IDEMPOTENCY_SQL)
      .bind(
        scope,
        input.idempotencyKey,
        requestHash,
        responseJson,
        isoAfterMinutes(24 * 60),
        now,
        input.requestId,
        input.accountId,
      ),
    db
      .prepare(CANCEL_ACCESS_REQUEST_SQL)
      .bind(now, input.requestId, input.accountId, scope, input.idempotencyKey, requestHash),
    db
      .prepare(
        `INSERT OR IGNORE INTO audit_events
          (id, actor_id, action, target_type, target_id, reason,
           before_json, after_json, correlation_id, created_at)
         SELECT ?, ?, 'beta_access.cancelled', 'beta_access_request', ?,
                '用户撤回免费闭测申请', ?, ?, ?, ?
         FROM beta_access_requests
         WHERE id = ? AND status = 'cancelled' AND cancelled_at = ?`,
      )
      .bind(
        auditId,
        input.actorId,
        input.requestId,
        JSON.stringify(previous),
        responseJson,
        newId("corr"),
        now,
        input.requestId,
        now,
      ),
  ]);

  const [stored, current, audit] = await Promise.all([
    db
      .prepare("SELECT request_hash, response_json FROM idempotency_records WHERE scope = ? AND key = ?")
      .bind(scope, input.idempotencyKey)
      .first<{ request_hash: string; response_json: string }>(),
    db.prepare("SELECT * FROM beta_access_requests WHERE id = ?").bind(input.requestId).first<BetaAccessRequestRecord>(),
    db.prepare("SELECT id FROM audit_events WHERE id = ?").bind(auditId).first<{ id: string }>(),
  ]);
  if (stored?.request_hash === requestHash) {
    const committed = JSON.parse(stored.response_json) as BetaAccessRequestRecord;
    if (current?.status !== "cancelled" || current.cancelled_at !== committed.cancelled_at || audit?.id !== auditId) {
      throw new DomainError("access_request_cancel_indeterminate", "闭测申请撤回结果需要人工核对", 500);
    }
    return committed;
  }
  if (stored) throw new DomainError("idempotency_conflict", "同一幂等键不能用于不同请求", 409);
  throw new DomainError("access_request_not_pending", "该闭测申请已经处理", 409);
}

export async function listAccountFeedback(
  accountId: string,
): Promise<FeedbackRecord[]> {
  await ensureDatabase();
  return all<FeedbackRecord>(
    getD1()
      .prepare(
        `SELECT * FROM beta_feedback
         WHERE account_id = ?
         ORDER BY created_at DESC, id DESC
         LIMIT 50`,
      )
      .bind(accountId),
  );
}

export async function getServiceStatusSnapshot(): Promise<ServiceStatusSnapshot> {
  await ensureDatabase();
  const db = getD1();
  const [activeIncidents, resolvedIncidents] = await Promise.all([
    all<PublicServiceIncidentRecord>(
      db.prepare(
        `SELECT id, severity, title, message, status, resolution,
                started_at, resolved_at, updated_at
         FROM service_incidents
         WHERE status = 'active'
         ORDER BY CASE severity
                    WHEN 'outage' THEN 3
                    WHEN 'degraded' THEN 2
                    WHEN 'maintenance' THEN 1
                    ELSE 0
                  END DESC,
                  started_at DESC, id DESC
         LIMIT 20`,
      ),
    ),
    all<PublicServiceIncidentRecord>(
      db.prepare(
        `SELECT id, severity, title, message, status, resolution,
                started_at, resolved_at, updated_at
         FROM service_incidents
         WHERE status = 'resolved'
         ORDER BY resolved_at DESC, id DESC
         LIMIT 20`,
      ),
    ),
  ]);
  return {
    status: deriveServiceStatus(activeIncidents),
    activeIncidents,
    recentIncidents: [...activeIncidents, ...resolvedIncidents],
    checkedAt: new Date().toISOString(),
  };
}

export async function createServiceIncident(input: {
  actorId: string;
  severity: string;
  title: string;
  message: string;
  reason: string;
  idempotencyKey: string;
}): Promise<ServiceIncidentRecord> {
  await ensureDatabase();
  if (!incidentSeverities.has(input.severity as IncidentSeverity)) {
    throw new DomainError("invalid_incident_severity", "请选择有效的故障级别");
  }
  const severity = input.severity as IncidentSeverity;
  const title = input.title.trim();
  const message = input.message.trim();
  if (title.length < 4 || title.length > 80) {
    throw new DomainError("invalid_incident_title", "标题需为 4 到 80 个字符");
  }
  if (message.length < 10 || message.length > 1_000) {
    throw new DomainError("invalid_incident_message", "用户说明需为 10 到 1000 个字符");
  }
  assertReason(input.reason);
  assertIdempotencyKey(input.idempotencyKey);

  const payload = {
    severity,
    title,
    message,
    reason: input.reason.trim(),
  };
  const requestHash = await hashText(JSON.stringify(payload));
  const scope = "incident:create";
  const replay = await assertIdempotencyAvailable(
    scope,
    input.idempotencyKey,
    requestHash,
  );
  if (replay.replay) {
    return JSON.parse(replay.responseJson!) as ServiceIncidentRecord;
  }

  const now = new Date().toISOString();
  const incident: ServiceIncidentRecord = {
    id: newId("incident"),
    severity,
    title,
    message,
    status: "active",
    created_by: input.actorId,
    resolved_by: null,
    resolution: null,
    started_at: now,
    resolved_at: null,
    created_at: now,
    updated_at: now,
  };
  const responseJson = JSON.stringify(incident);
  const db = getD1();
  const racedIncident = await commitIdempotentBatch<ServiceIncidentRecord>({
    db,
    scope,
    key: input.idempotencyKey,
    requestHash,
    statements: [
      db
        .prepare(
          `INSERT INTO idempotency_records
            (scope, key, request_hash, response_json, status_code, expires_at, created_at)
           VALUES (?, ?, ?, ?, 201, ?, ?)`,
        )
        .bind(
          scope,
          input.idempotencyKey,
          requestHash,
          responseJson,
          isoAfterMinutes(30 * 24 * 60),
          now,
        ),
      db
        .prepare(
          `INSERT INTO service_incidents
            (id, severity, title, message, status, created_by,
             started_at, created_at, updated_at)
           VALUES (?, ?, ?, ?, 'active', ?, ?, ?, ?)`,
        )
        .bind(
          incident.id,
          incident.severity,
          incident.title,
          incident.message,
          incident.created_by,
          now,
          now,
          now,
        ),
      db
        .prepare(
          `INSERT INTO audit_events
            (id, actor_id, action, target_type, target_id, reason,
             after_json, correlation_id, created_at)
           VALUES (?, ?, 'incident.created', 'service_incident', ?, ?, ?, ?, ?)`,
        )
        .bind(
          newId("audit"),
          input.actorId,
          incident.id,
          input.reason.trim(),
          JSON.stringify({ severity, title, status: "active" }),
          newId("corr"),
          now,
        ),
    ],
  });
  return racedIncident ?? incident;
}

export async function resolveServiceIncident(input: {
  incidentId: string;
  actorId: string;
  resolution: string;
  reason: string;
}): Promise<ServiceIncidentRecord> {
  await ensureDatabase();
  if (!/^incident_[a-f0-9]{32}$/.test(input.incidentId)) {
    throw new DomainError("invalid_incident_id", "故障公告编号无效");
  }
  const resolution = input.resolution.trim();
  if (resolution.length < 4 || resolution.length > 500) {
    throw new DomainError("invalid_incident_resolution", "恢复说明需为 4 到 500 个字符");
  }
  assertReason(input.reason);

  const requestHash = await hashText(
    JSON.stringify({ resolution, reason: input.reason.trim() }),
  );
  const scope = "incident:resolve";
  const replay = await assertIdempotencyAvailable(
    scope,
    input.incidentId,
    requestHash,
  );
  if (replay.replay) {
    return JSON.parse(replay.responseJson!) as ServiceIncidentRecord;
  }

  const db = getD1();
  const previous = await db
    .prepare("SELECT * FROM service_incidents WHERE id = ?")
    .bind(input.incidentId)
    .first<ServiceIncidentRecord>();
  if (!previous) throw new DomainError("incident_not_found", "故障公告不存在", 404);
  if (previous.status !== "active") {
    throw new DomainError("incident_already_resolved", "故障公告已经恢复", 409);
  }

  const now = new Date().toISOString();
  const resolved: ServiceIncidentRecord = {
    ...previous,
    status: "resolved",
    resolved_by: input.actorId,
    resolution,
    resolved_at: now,
    updated_at: now,
  };
  const responseJson = JSON.stringify(resolved);
  const racedIncident = await commitIdempotentBatch<ServiceIncidentRecord>({
    db,
    scope,
    key: input.incidentId,
    requestHash,
    statements: [
      db
        .prepare(
          `INSERT INTO idempotency_records
            (scope, key, request_hash, response_json, status_code, expires_at, created_at)
           VALUES (?, ?, ?, ?, 200, ?, ?)`,
        )
        .bind(
          scope,
          input.incidentId,
          requestHash,
          responseJson,
          isoAfterMinutes(90 * 24 * 60),
          now,
        ),
      db
        .prepare(
          `UPDATE service_incidents
           SET status = 'resolved', resolved_by = ?, resolution = ?,
               resolved_at = ?, updated_at = ?
           WHERE id = ? AND status = 'active'`,
        )
        .bind(input.actorId, resolution, now, now, input.incidentId),
      db
        .prepare(
          `INSERT INTO audit_events
            (id, actor_id, action, target_type, target_id, reason,
             before_json, after_json, correlation_id, created_at)
           VALUES (?, ?, 'incident.resolved', 'service_incident', ?, ?, ?, ?, ?, ?)`,
        )
        .bind(
          newId("audit"),
          input.actorId,
          input.incidentId,
          input.reason.trim(),
          JSON.stringify({ severity: previous.severity, title: previous.title, status: "active" }),
          JSON.stringify({ severity: previous.severity, title: previous.title, status: "resolved" }),
          newId("corr"),
          now,
        ),
    ],
  });
  return racedIncident ?? resolved;
}

export async function listAccountDeletionRequests(
  accountId: string,
): Promise<AccountDeletionRequestRecord[]> {
  await ensureDatabase();
  return all<AccountDeletionRequestRecord>(
    getD1()
      .prepare(
        `SELECT * FROM account_deletion_requests
         WHERE account_id = ?
         ORDER BY requested_at DESC, id DESC
         LIMIT 20`,
      )
      .bind(accountId),
  );
}

export async function requestAccountDeletion(input: {
  accountId: string;
  actorId: string;
  confirmed: boolean;
  reason: string;
  idempotencyKey: string;
}): Promise<AccountDeletionRequestRecord> {
  await ensureDatabase();
  if (input.confirmed !== true) {
    throw new DomainError(
      "deletion_confirmation_required",
      "请先确认已理解注销申请的影响",
    );
  }
  const reason = input.reason.trim();
  if (reason.length > 500) {
    throw new DomainError("deletion_reason_too_long", "注销说明不能超过 500 个字符");
  }
  assertIdempotencyKey(input.idempotencyKey);

  const requestHash = await hashText(JSON.stringify({ reason }));
  const scope = `account:deletion:request:${input.accountId}`;
  const replay = await assertIdempotencyAvailable(
    scope,
    input.idempotencyKey,
    requestHash,
  );
  if (replay.replay) {
    return JSON.parse(replay.responseJson!) as AccountDeletionRequestRecord;
  }

  const db = getD1();
  const existing = await db
    .prepare(
      `SELECT * FROM account_deletion_requests
       WHERE account_id = ? AND status IN ('requested', 'processing', 'relay_purged')
       LIMIT 1`,
    )
    .bind(input.accountId)
    .first<AccountDeletionRequestRecord>();
  if (existing) return existing;

  const now = new Date().toISOString();
  const request: AccountDeletionRequestRecord = {
    id: newId("deletion"),
    account_id: input.accountId,
    status: "requested",
    reason: reason || null,
    requested_at: now,
    cancelled_at: null,
    created_at: now,
    updated_at: now,
  };
  const responseJson = JSON.stringify(request);
  try {
    const racedRequest = await commitIdempotentBatch<AccountDeletionRequestRecord>({
      db,
      scope,
      key: input.idempotencyKey,
      requestHash,
      statements: [
        db
          .prepare(
            `INSERT INTO idempotency_records
              (scope, key, request_hash, response_json, status_code, expires_at, created_at)
             VALUES (?, ?, ?, ?, 201, ?, ?)`,
          )
          .bind(
            scope,
            input.idempotencyKey,
            requestHash,
            responseJson,
            isoAfterMinutes(90 * 24 * 60),
            now,
          ),
        db
          .prepare(
            `INSERT INTO account_deletion_requests
              (id, account_id, status, reason, requested_at, created_at, updated_at)
             VALUES (?, ?, 'requested', ?, ?, ?, ?)`,
          )
          .bind(
            request.id,
            request.account_id,
            request.reason,
            now,
            now,
            now,
          ),
        db
          .prepare(
            `INSERT INTO audit_events
              (id, actor_id, action, target_type, target_id, reason,
               after_json, correlation_id, created_at)
             VALUES (?, ?, 'account.deletion_requested', 'account', ?,
                     '用户提交账户注销申请', ?, ?, ?)`,
          )
          .bind(
            newId("audit"),
            input.actorId,
            input.accountId,
            JSON.stringify({ requestId: request.id, status: "requested" }),
            newId("corr"),
            now,
          ),
      ],
    });
    return racedRequest ?? request;
  } catch (error) {
    const raced = await db
      .prepare(
        `SELECT * FROM account_deletion_requests
         WHERE account_id = ? AND status IN ('requested', 'processing', 'relay_purged')
         LIMIT 1`,
      )
      .bind(input.accountId)
      .first<AccountDeletionRequestRecord>();
    if (raced) return raced;
    throw error;
  }
}

export async function cancelAccountDeletion(input: {
  accountId: string;
  actorId: string;
  requestId: string;
}): Promise<AccountDeletionRequestRecord> {
  await ensureDatabase();
  if (!/^deletion_[a-f0-9]{32}$/.test(input.requestId)) {
    throw new DomainError("invalid_deletion_request_id", "注销申请编号无效");
  }
  const requestHash = await hashText(
    JSON.stringify({ accountId: input.accountId, action: "cancel" }),
  );
  const scope = "account:deletion:cancel";
  const replay = await assertIdempotencyAvailable(
    scope,
    input.requestId,
    requestHash,
  );
  if (replay.replay) {
    return JSON.parse(replay.responseJson!) as AccountDeletionRequestRecord;
  }

  const db = getD1();
  const previous = await db
    .prepare(
      `SELECT * FROM account_deletion_requests
       WHERE id = ? AND account_id = ?`,
    )
    .bind(input.requestId, input.accountId)
    .first<AccountDeletionRequestRecord>();
  if (!previous) {
    throw new DomainError("deletion_request_not_found", "注销申请不存在", 404);
  }
  if (previous.status === "cancelled") return previous;
  if (previous.status !== "requested") {
    throw new DomainError(
      "deletion_cannot_cancel",
      "租户永久删除已经开始，注销申请不能再撤回",
      409,
    );
  }

  const now = new Date().toISOString();
  const cancelled: AccountDeletionRequestRecord = {
    ...previous,
    status: "cancelled",
    cancelled_at: now,
    updated_at: now,
  };
  const responseJson = JSON.stringify(cancelled);
  const racedRequest = await commitIdempotentBatch<AccountDeletionRequestRecord>({
    db,
    scope,
    key: input.requestId,
    requestHash,
    statements: [
      db
        .prepare(
          `INSERT INTO idempotency_records
            (scope, key, request_hash, response_json, status_code, expires_at, created_at)
           VALUES (?, ?, ?, ?, 200, ?, ?)`,
        )
        .bind(
          scope,
          input.requestId,
          requestHash,
          responseJson,
          isoAfterMinutes(90 * 24 * 60),
          now,
        ),
      db
        .prepare(
          `UPDATE account_deletion_requests
           SET status = 'cancelled', cancelled_at = ?, updated_at = ?
           WHERE id = ? AND account_id = ? AND status = 'requested'`,
        )
        .bind(now, now, input.requestId, input.accountId),
      db
        .prepare(
          `INSERT INTO audit_events
            (id, actor_id, action, target_type, target_id, reason,
             before_json, after_json, correlation_id, created_at)
           VALUES (?, ?, 'account.deletion_cancelled', 'account', ?,
                   '用户撤回账户注销申请', ?, ?, ?, ?)`,
        )
        .bind(
          newId("audit"),
          input.actorId,
          input.accountId,
          JSON.stringify({ requestId: input.requestId, status: "requested" }),
          JSON.stringify({ requestId: input.requestId, status: "cancelled" }),
          newId("corr"),
          now,
        ),
    ],
  });
  return racedRequest ?? cancelled;
}

export async function getAccountControlPlaneExport(account: AccountRecord) {
  await ensureDatabase();
  const db = getD1();
  const [hosts, pairings, grants, accessRequests, tenant, placement, feedback, deletionRequests] =
    await Promise.all([
      all(
        db
          .prepare(
            `SELECT id, name, os, lifecycle, slot_state, connection_state,
                    daemon_version, identity_fingerprint, claimed_at,
                    last_seen_at, created_at, deactivated_at
             FROM hosts WHERE account_id = ?
             ORDER BY created_at DESC`,
          )
          .bind(account.id),
      ),
      all(
        db
          .prepare(
            `SELECT id, requested_name, os, status, expires_at,
                    claimed_host_id, failed_attempts, locked_at,
                    last_attempt_at, created_at, claimed_at
             FROM pairing_requests WHERE account_id = ?
             ORDER BY created_at DESC`,
          )
          .bind(account.id),
      ),
      all(
        db
          .prepare(
            `SELECT id, host_id, source, source_ref, capacity_slots,
                    starts_at, ends_at, state, reason, created_at, revoked_at
             FROM entitlement_grants WHERE account_id = ?
             ORDER BY created_at DESC`,
          )
          .bind(account.id),
      ),
      all<BetaAccessRequestRecord>(
        db
          .prepare(
            `SELECT * FROM beta_access_requests
             WHERE account_id = ?
             ORDER BY requested_at DESC, id DESC`,
          )
          .bind(account.id),
      ),
      db
        .prepare(
          `SELECT id, slug, lifecycle, desired_state, observed_state,
                  desired_generation, active_generation, credential_revision,
                  runtime_version, relay_origin, relay_ready, last_health_at,
                  tombstoned_at, created_at, updated_at
           FROM tenant_instances WHERE account_id = ?`,
        )
        .bind(account.id)
        .first(),
      db
        .prepare(
          `SELECT regions.code AS home_region, placements.generation,
                  placements.state, placements.last_error_code,
                  placements.created_at, placements.updated_at
           FROM tenant_placements AS placements
           INNER JOIN tenant_instances AS tenants
             ON tenants.id = placements.tenant_id
           INNER JOIN relay_regions AS regions
             ON regions.id = placements.home_region_id
           WHERE tenants.account_id = ?`,
        )
        .bind(account.id)
        .first(),
      all<FeedbackRecord>(
        db
          .prepare(
            `SELECT * FROM beta_feedback
             WHERE account_id = ?
             ORDER BY created_at DESC`,
          )
          .bind(account.id),
      ),
      listAccountDeletionRequests(account.id),
    ]);

  return {
    schema_version: 2,
    scope: "nekonest-cloud-control-plane",
    exported_at: new Date().toISOString(),
    notices: [
      "本文件只包含 NekoNest Cloud 控制平面保存的数据。",
      "原生 coding-agent 会话、项目文件和 transcript 仍在用户主机，不属于本导出。",
      "设备令牌、配对码、摘要、内部密钥、节点凭据和内部审计不会进入导出。",
    ],
    account: {
      id: account.id,
      email: account.email,
      display_name: account.display_name,
      status: account.status,
      created_at: account.created_at,
    },
    hosts,
    pairing_requests: pairings,
    entitlement_grants: grants,
    beta_access_requests: accessRequests,
    tenant,
    tenant_placement: placement,
    feedback,
    deletion_requests: deletionRequests,
  };
}

export async function runRetentionMaintenance(input: {
  actorId: string;
  confirmed: boolean;
  reason: string;
  now?: string;
}): Promise<RetentionMaintenanceResult> {
  await ensureDatabase();
  if (!input.confirmed) {
    throw new DomainError(
      "retention_confirmation_required",
      "请先确认只清理已经到期的技术记录",
    );
  }
  assertReason(input.reason);

  const cutoffs = getRetentionCutoffs(input.now ?? new Date().toISOString());
  const db = getD1();
  const correlationId = newId("corr");
  const auditPayload = JSON.stringify({
    version: 1,
    completedAt: cutoffs.now,
    claimRateBefore: cutoffs.claimRateBefore,
    claimAttemptBefore: cutoffs.claimAttemptBefore,
    handoffTicketBefore: cutoffs.handoffTicketBefore,
    scopes: [
      "expired_pairing_code_hashes",
      "expired_claim_rate_windows",
      "expired_claim_attempts",
      "expired_idempotency_records",
      "retired_pending_phone_routes",
      "retired_pending_phone_principals",
      "retired_phone_handoff_tickets",
      "expired_device_registration_replays",
    ],
  });

  const results = await db.batch([
    db
      .prepare(
        `INSERT INTO audit_events
          (id, actor_id, action, target_type, target_id, reason,
           after_json, correlation_id, created_at)
         VALUES (?, ?, 'privacy.retention_cleanup', 'system', 'control-plane',
                 ?, ?, ?, ?)`,
      )
      .bind(
        newId("audit"),
        input.actorId,
        input.reason.trim(),
        auditPayload,
        correlationId,
        cutoffs.now,
      ),
    db.prepare(RETIRE_EXPIRED_PAIRING_CODES_SQL).bind(cutoffs.now),
    db
      .prepare(DELETE_EXPIRED_CLAIM_RATE_WINDOWS_SQL)
      .bind(cutoffs.claimRateBefore),
    db
      .prepare(DELETE_EXPIRED_CLAIM_ATTEMPTS_SQL)
      .bind(cutoffs.claimAttemptBefore),
    db.prepare(DELETE_EXPIRED_IDEMPOTENCY_SQL).bind(cutoffs.now),
    db
      .prepare(DELETE_RETIRED_PENDING_PHONE_ROUTES_SQL)
      .bind(cutoffs.handoffTicketBefore, cutoffs.handoffTicketBefore),
    db
      .prepare(DELETE_RETIRED_PENDING_PHONE_PRINCIPALS_SQL)
      .bind(cutoffs.handoffTicketBefore, cutoffs.handoffTicketBefore),
    db
      .prepare(DELETE_RETIRED_PHONE_HANDOFF_TICKETS_SQL)
      .bind(cutoffs.handoffTicketBefore, cutoffs.handoffTicketBefore),
    db.prepare(DELETE_EXPIRED_DEVICE_REGISTRATION_REPLAYS_SQL).bind(cutoffs.now),
  ]);

  if (Number(results[0]?.meta.changes ?? 0) !== 1) {
    throw new DomainError(
      "retention_audit_failed",
      "清理动作未能写入审计，因此没有执行",
      503,
    );
  }

  return {
    completedAt: cutoffs.now,
    claimRateBefore: cutoffs.claimRateBefore,
    claimAttemptBefore: cutoffs.claimAttemptBefore,
    handoffTicketBefore: cutoffs.handoffTicketBefore,
    retiredPairingCodes: Number(results[1]?.meta.changes ?? 0),
    deletedClaimRateWindows: Number(results[2]?.meta.changes ?? 0),
    deletedClaimAttempts: Number(results[3]?.meta.changes ?? 0),
    deletedIdempotencyRecords: Number(results[4]?.meta.changes ?? 0),
    deletedPendingPhoneRoutes: Number(results[5]?.meta.changes ?? 0),
    deletedPendingPhonePrincipals: Number(results[6]?.meta.changes ?? 0),
    deletedPhoneHandoffTickets: Number(results[7]?.meta.changes ?? 0),
    deletedDeviceRegistrationReplays: Number(results[8]?.meta.changes ?? 0),
  };
}

export async function createFeedback(input: {
  accountId: string;
  actorId: string;
  category: string;
  message: string;
  idempotencyKey: string;
}): Promise<FeedbackRecord> {
  await ensureDatabase();
  if (!feedbackCategories.has(input.category as FeedbackCategory)) {
    throw new DomainError("invalid_feedback_category", "请选择有效的问题类型");
  }
  const category = input.category as FeedbackCategory;
  const message = input.message.trim();
  if (message.length < 10 || message.length > 2_000) {
    throw new DomainError(
      "invalid_feedback_message",
      "问题描述需为 10 到 2000 个字符",
    );
  }
  assertIdempotencyKey(input.idempotencyKey);

  const requestHash = await hashText(JSON.stringify({ category, message }));
  const scope = `feedback:create:${input.accountId}`;
  const replay = await assertIdempotencyAvailable(
    scope,
    input.idempotencyKey,
    requestHash,
  );
  if (replay.replay) return JSON.parse(replay.responseJson!) as FeedbackRecord;

  const now = new Date().toISOString();
  const feedback: FeedbackRecord = {
    id: newId("feedback"),
    account_id: input.accountId,
    category,
    message,
    status: "open",
    admin_response: null,
    handled_by: null,
    created_at: now,
    updated_at: now,
    resolved_at: null,
  };
  const responseJson = JSON.stringify(feedback);
  const db = getD1();
  const racedFeedback = await commitIdempotentBatch<FeedbackRecord>({
    db,
    scope,
    key: input.idempotencyKey,
    requestHash,
    statements: [
      db
        .prepare(
          `INSERT INTO idempotency_records
            (scope, key, request_hash, response_json, status_code, expires_at, created_at)
           VALUES (?, ?, ?, ?, 201, ?, ?)`,
        )
        .bind(
          scope,
          input.idempotencyKey,
          requestHash,
          responseJson,
          isoAfterMinutes(24 * 60),
          now,
        ),
      db
        .prepare(
          `INSERT INTO beta_feedback
            (id, account_id, category, message, status, created_at, updated_at)
           VALUES (?, ?, ?, ?, 'open', ?, ?)`,
        )
        .bind(
          feedback.id,
          feedback.account_id,
          feedback.category,
          feedback.message,
          now,
          now,
        ),
      db
        .prepare(
          `INSERT INTO audit_events
            (id, actor_id, action, target_type, target_id, reason,
             after_json, correlation_id, created_at)
           VALUES (?, ?, 'feedback.created', 'beta_feedback', ?,
                   '用户提交免费公测反馈', ?, ?, ?)`,
        )
        .bind(
          newId("audit"),
          input.actorId,
          feedback.id,
          JSON.stringify({ category, status: "open" }),
          newId("corr"),
          now,
        ),
    ],
  });
  return racedFeedback ?? feedback;
}

export async function resolveFeedback(input: {
  feedbackId: string;
  actorId: string;
  response: string;
  reason: string;
  idempotencyKey: string;
}): Promise<FeedbackRecord> {
  await ensureDatabase();
  if (!/^feedback_[a-f0-9]{32}$/.test(input.feedbackId)) {
    throw new DomainError("invalid_feedback_id", "反馈编号无效");
  }
  const response = input.response.trim();
  if (response.length < 2 || response.length > 1_000) {
    throw new DomainError(
      "invalid_feedback_response",
      "回复需为 2 到 1000 个字符",
    );
  }
  assertReason(input.reason);
  assertIdempotencyKey(input.idempotencyKey);

  const requestHash = await hashText(
    JSON.stringify({ response, reason: input.reason.trim() }),
  );
  const scope = "feedback:resolve";
  const replay = await assertIdempotencyAvailable(
    scope,
    input.feedbackId,
    requestHash,
  );
  if (replay.replay) return JSON.parse(replay.responseJson!) as FeedbackRecord;

  const db = getD1();
  const previous = await db
    .prepare("SELECT * FROM beta_feedback WHERE id = ?")
    .bind(input.feedbackId)
    .first<FeedbackRecord>();
  if (!previous) throw new DomainError("feedback_not_found", "反馈不存在", 404);
  if (previous.status !== "open") {
    throw new DomainError("feedback_already_resolved", "反馈已处理", 409);
  }

  const now = new Date().toISOString();
  const resolved: FeedbackRecord = {
    ...previous,
    status: "resolved",
    admin_response: response,
    handled_by: input.actorId,
    updated_at: now,
    resolved_at: now,
  };
  const responseJson = JSON.stringify(resolved);
  const racedFeedback = await commitIdempotentBatch<FeedbackRecord>({
    db,
    scope,
    key: input.feedbackId,
    requestHash,
    statements: [
      db
        .prepare(
          `INSERT INTO idempotency_records
            (scope, key, request_hash, response_json, status_code, expires_at, created_at)
           VALUES (?, ?, ?, ?, 200, ?, ?)`,
        )
        .bind(
          scope,
          input.feedbackId,
          requestHash,
          responseJson,
          isoAfterMinutes(30 * 24 * 60),
          now,
        ),
      db
        .prepare(
          `UPDATE beta_feedback
           SET status = 'resolved', admin_response = ?, handled_by = ?,
               updated_at = ?, resolved_at = ?
           WHERE id = ? AND status = 'open'`,
        )
        .bind(response, input.actorId, now, now, input.feedbackId),
      db
        .prepare(
          `INSERT INTO audit_events
            (id, actor_id, action, target_type, target_id, reason,
             before_json, after_json, correlation_id, created_at)
           VALUES (?, ?, 'feedback.resolved', 'beta_feedback', ?, ?, ?, ?, ?, ?)`,
        )
        .bind(
          newId("audit"),
          input.actorId,
          input.feedbackId,
          input.reason.trim(),
          JSON.stringify({ category: previous.category, status: previous.status }),
          JSON.stringify({ category: previous.category, status: "resolved", hasResponse: true }),
          newId("corr"),
          now,
        ),
    ],
  });
  return racedFeedback ?? resolved;
}

export async function createPairingRequest(input: {
  accountId: string;
  requestedName: string;
  os: "windows" | "linux";
  actorId: string;
}): Promise<{ id: string; bootstrapToken: string; expiresAt: string }> {
  await ensureDatabase();
  const name = input.requestedName.trim();
  if (name.length < 2 || name.length > 48) {
    throw new DomainError("invalid_host_name", "主机名称需为 2 到 48 个字符");
  }
  if (!(["windows", "linux"] as const).includes(input.os)) {
    throw new DomainError("invalid_os", "首版仅支持 Windows 与 Linux");
  }

  const db = getD1();
  const rawCode = Array.from(
    crypto.getRandomValues(new Uint8Array(10)),
    (byte) => byte.toString(16).padStart(2, "0"),
  )
    .join("")
    .toUpperCase();
  const codeHash = await hashText(rawCode);
  const id = newId("pair");
  const expiresAt = isoAfterMinutes(10);
  const now = new Date().toISOString();
  const correlationId = newId("corr");
  const results = await db.batch([
    db
      .prepare(RESERVE_PAIRING_SQL)
      .bind(id, input.accountId, name, input.os, codeHash, expiresAt, now),
    db
      .prepare(
        `INSERT INTO audit_events
          (id, actor_id, action, target_type, target_id, reason,
           after_json, correlation_id, created_at)
         SELECT ?, ?, 'pairing.requested', 'pairing_request', ?,
                '用户创建一次性配对请求', ?, ?, ?
         WHERE EXISTS (SELECT 1 FROM pairing_requests WHERE id = ?)`,
      )
      .bind(
        newId("audit"),
        input.actorId,
        id,
        JSON.stringify({ requestedName: name, os: input.os, expiresAt }),
        correlationId,
        now,
        id,
      ),
  ]);

  if (Number(results[0]?.meta.changes ?? 0) !== 1) {
    const pending = await db
      .prepare(
        `SELECT COUNT(*) AS count FROM pairing_requests
         WHERE account_id = ? AND status = 'waiting' AND expires_at > ?`,
      )
      .bind(input.accountId, now)
      .first<{ count: number }>();
    if (Number(pending?.count ?? 0) >= 5) {
      throw new DomainError(
        "pairing_limit_reached",
        "同时最多保留 5 个未过期配对请求，请等待过期后再试",
        429,
      );
    }
    const entitlement = await evaluateEntitlement(input.accountId);
    if (entitlement.mode === "none" && entitlement.publicBetaState === "gated") {
      throw new DomainError(
        "public_beta_gated",
        "公开公测安全门禁尚未通过；当前只接受管理员明确邀请的闭测账户",
        409,
      );
    }
    if (entitlement.mode === "none" && entitlement.publicBetaState === "inactive") {
      throw new DomainError(
        "entitlement_required",
        "公开公测当前未开放，该账户也没有有效闭测邀请",
        409,
      );
    }
    throw new DomainError("device_capacity_exceeded", "当前没有可用主机席位", 409);
  }

  return { id, bootstrapToken: `${id}.${rawCode}`, expiresAt };
}

export async function getOwnedPairingProgress(input: {
  accountId: string;
  pairingId: string;
}): Promise<PairingProgress> {
  await ensureDatabase();
  if (!/^pair_[a-f0-9]{32}$/.test(input.pairingId)) {
    throw new DomainError("invalid_pairing_id", "配对请求编号无效");
  }
  const row = await getD1()
    .prepare(OWNED_PAIRING_PROGRESS_SQL)
    .bind(input.pairingId, input.accountId)
    .first<PairingProgressRow>();
  if (!row) {
    throw new DomainError("pairing_not_found", "配对请求不存在", 404);
  }
  return derivePairingProgress(row, new Date().toISOString());
}

export type PairingCancellationResult = {
  pairingId: string;
  status: "cancelled";
  cancelled: boolean;
};

export async function cancelPairingRequest(input: {
  accountId: string;
  pairingId: string;
  actorId: string;
}): Promise<PairingCancellationResult> {
  await ensureDatabase();
  if (!/^pair_[a-f0-9]{32}$/.test(input.pairingId)) {
    throw new DomainError("invalid_pairing_id", "配对请求编号无效");
  }

  const requestHash = await hashText(
    JSON.stringify({ accountId: input.accountId, action: "cancel" }),
  );
  const scope = `pairing:cancel:${input.accountId}`;
  const replay = await assertIdempotencyAvailable(
    scope,
    input.pairingId,
    requestHash,
  );
  if (replay.replay) {
    return JSON.parse(replay.responseJson!) as PairingCancellationResult;
  }

  const db = getD1();
  const pairing = await db
    .prepare(
      `SELECT id, status, requested_name, os, expires_at
       FROM pairing_requests
       WHERE id = ? AND account_id = ?`,
    )
    .bind(input.pairingId, input.accountId)
    .first<{
      id: string;
      status: string;
      requested_name: string;
      os: string;
      expires_at: string;
    }>();
  if (!pairing) {
    throw new DomainError("pairing_not_found", "配对请求不存在", 404);
  }
  if (pairing.status === "cancelled") {
    return { pairingId: pairing.id, status: "cancelled", cancelled: false };
  }
  if (pairing.status !== "waiting") {
    throw new DomainError(
      "pairing_not_cancellable",
      "配对请求已经失效或被 daemon 使用，请刷新主机列表",
      409,
    );
  }

  const now = new Date().toISOString();
  const tombstoneHash = await hashText(`${newId("cancelled")}:${now}`);
  const response: PairingCancellationResult = {
    pairingId: pairing.id,
    status: "cancelled",
    cancelled: true,
  };
  const responseJson = JSON.stringify(response);
  const correlationId = newId("corr");

  let results: D1Result<unknown>[];
  try {
    results = await db.batch([
      db
        .prepare(CANCEL_PAIRING_SQL)
        .bind(pairing.id, input.accountId, tombstoneHash),
      db
        .prepare(
          `INSERT INTO idempotency_records
            (scope, key, request_hash, response_json, status_code,
             expires_at, created_at)
           SELECT ?, ?, ?, ?, 200, ?, ?
           WHERE EXISTS (
             SELECT 1 FROM pairing_requests
             WHERE id = ? AND account_id = ? AND status = 'cancelled'
               AND code_hash = ?
           )`,
        )
        .bind(
          scope,
          pairing.id,
          requestHash,
          responseJson,
          isoAfterMinutes(90 * 24 * 60),
          now,
          pairing.id,
          input.accountId,
          tombstoneHash,
        ),
      db
        .prepare(
          `INSERT INTO audit_events
            (id, actor_id, action, target_type, target_id, reason,
             before_json, after_json, correlation_id, created_at)
           SELECT ?, ?, 'pairing.cancelled', 'pairing_request', ?,
                  '用户主动取消未认领配对请求', ?, ?, ?, ?
           WHERE EXISTS (
             SELECT 1 FROM pairing_requests
             WHERE id = ? AND account_id = ? AND status = 'cancelled'
               AND code_hash = ?
           )`,
        )
        .bind(
          newId("audit"),
          input.actorId,
          pairing.id,
          JSON.stringify({
            status: pairing.status,
            requestedName: pairing.requested_name,
            os: pairing.os,
            expiresAt: pairing.expires_at,
          }),
          JSON.stringify({ status: "cancelled" }),
          correlationId,
          now,
          pairing.id,
          input.accountId,
          tombstoneHash,
        ),
    ]);
  } catch (error) {
    const raced = await db
      .prepare(
        `SELECT request_hash, response_json
         FROM idempotency_records WHERE scope = ? AND key = ?`,
      )
      .bind(scope, pairing.id)
      .first<{ request_hash: string; response_json: string }>();
    if (raced?.request_hash === requestHash) {
      return JSON.parse(raced.response_json) as PairingCancellationResult;
    }
    throw error;
  }

  if (
    Number(results[0]?.meta.changes ?? 0) === 1 &&
    Number(results[1]?.meta.changes ?? 0) === 1 &&
    Number(results[2]?.meta.changes ?? 0) === 1
  ) {
    return response;
  }

  const raced = await db
    .prepare(
      `SELECT request_hash, response_json
       FROM idempotency_records WHERE scope = ? AND key = ?`,
    )
    .bind(scope, pairing.id)
    .first<{ request_hash: string; response_json: string }>();
  if (raced?.request_hash === requestHash) {
    return JSON.parse(raced.response_json) as PairingCancellationResult;
  }
  throw new DomainError(
    "pairing_not_cancellable",
    "配对请求可能已被 daemon 使用，请刷新主机列表",
    409,
  );
}

function pairingClaimSecret(): string {
  const configured = env.NEKONEST_CLOUD_CREDENTIAL_SECRET?.trim() ?? "";
  if (configured.length >= 32) return configured;
  if (process.env.NODE_ENV !== "production") {
    return "nekonest-cloud-local-development-only-secret";
  }
  throw new DomainError(
    "pairing_claim_unavailable",
    "设备接入暂时不可用",
    503,
  );
}

async function consumeClaimSourceBudget(input: {
  source?: string;
  trustedSourceHash?: string;
  now: string;
}): Promise<string> {
  const db = getD1();
  const trustedSourceHash = input.trustedSourceHash?.trim().toLowerCase() ?? "";
  if (trustedSourceHash && !/^[0-9a-f]{64}$/u.test(trustedSourceHash)) {
    throw new DomainError("registration_rate_limited", "注册来源无效", 429, true, 60);
  }
  const sourceHash = trustedSourceHash || await sourceFingerprint(
    pairingClaimSecret(),
    input.source ?? "",
  );
  const windowStart = rateWindowStart(new Date(input.now));
  const nowMilliseconds = new Date(input.now).getTime();
  const results = await db.batch([
    db
      .prepare(CONSUME_CLAIM_RATE_SQL)
      .bind(sourceHash, windowStart, input.now),
    db
      .prepare("DELETE FROM pairing_claim_rate_limits WHERE window_start < ?")
      .bind(new Date(nowMilliseconds - 24 * 60 * 60_000).toISOString()),
    db
      .prepare("DELETE FROM pairing_claim_attempts WHERE created_at < ?")
      .bind(new Date(nowMilliseconds - 30 * 24 * 60 * 60_000).toISOString()),
  ]);
  const result = results[0];
  if (Number(result.meta.changes ?? 0) !== 1) {
    await db
      .prepare(
        `INSERT INTO pairing_claim_attempts
          (id, pairing_request_id, source_hash, outcome, created_at)
         VALUES (?, 'rate-limited', ?, 'rate_limited', ?)`,
      )
      .bind(newId("claim"), sourceHash, input.now)
      .run();
    throw new DomainError(
      "registration_rate_limited",
      "设备注册请求过于频繁，请稍后再试",
      429,
      true,
      60,
    );
  }
  return sourceHash;
}

async function rejectPairingClaim(input: {
  pairingId: string;
  sourceHash: string;
  now: string;
  countWrongCode: boolean;
}): Promise<never> {
  const db = getD1();
  const statements: D1PreparedStatement[] = [];
  if (input.countWrongCode) {
    statements.push(
      db
        .prepare(RECORD_FAILED_CODE_SQL)
        .bind(input.now, input.pairingId),
    );
  }
  statements.push(
    db
      .prepare(
        `INSERT INTO pairing_claim_attempts
          (id, pairing_request_id, source_hash, outcome, created_at)
         VALUES (?, ?, ?, 'rejected', ?)`,
      )
      .bind(newId("claim"), input.pairingId, input.sourceHash, input.now),
  );
  await db.batch(statements);
  throw new DomainError(
    "pairing_claim_rejected",
    "配对凭证无效或已失效",
    401,
  );
}

async function rejectPairingAccess(input: {
  pairingId: string;
  accountId: string;
  sourceHash: string;
  now: string;
}): Promise<never> {
  const entitlement = await evaluateEntitlement(input.accountId);
  const gated = entitlement.publicBetaState === "gated";
  await getD1()
    .prepare(
      `INSERT INTO pairing_claim_attempts
        (id, pairing_request_id, source_hash, outcome, created_at)
       VALUES (?, ?, ?, 'gated', ?)`,
    )
    .bind(newId("claim"), input.pairingId, input.sourceHash, input.now)
    .run();
  throw new DomainError(
    "registration_disabled",
    gated
      ? "公开公测安全门禁尚未通过；该账户需要管理员闭测邀请"
      : "公开公测当前未开放，该账户没有有效闭测邀请",
    403,
  );
}

export async function claimDevice(input: {
  bootstrapToken: string;
  source?: string;
  trustedSourceHash?: string;
  os: string;
  ed25519Public: string;
  x25519Public: string;
  identityFingerprint: string;
  transportMode: string;
  registrationProof: string;
  daemonVersion: string;
  registrationRetryKey: string;
}): Promise<DeviceRegistrationResult> {
  await ensureDatabase();
  const now = new Date().toISOString();

  const registrationRetryKey = input.registrationRetryKey.trim().toLowerCase();
  if (!/^[0-9a-f]{64}$/u.test(registrationRetryKey)) {
    throw new DomainError("device_identity_conflict", "注册重试密钥无效", 409);
  }

  const reportedVersion = classifyReportedDaemonVersion(input.daemonVersion);
  if (reportedVersion.state === "invalid") {
    throw new DomainError(
      "invalid_daemon_version",
      "daemon 版本格式无效，请使用稳定版本号 X.Y.Z",
      400,
    );
  }
  if (reportedVersion.state === "incompatible") {
    throw new DomainError(
      "protocol_upgrade_required",
      `daemon 版本过低，需要 ${MINIMUM_CLOUD_DAEMON_VERSION} 或更高版本`,
      426,
    );
  }

  let pairingId = "invalid";
  let code = "";
  try {
    ({ pairingId, code } = parseBootstrapToken(input.bootstrapToken));
  } catch {
    const invalidSourceHash = await consumeClaimSourceBudget({
      source: input.source,
      trustedSourceHash: input.trustedSourceHash,
      now,
    });
    return rejectPairingClaim({
      pairingId,
      sourceHash: invalidSourceHash,
      now,
      countWrongCode: false,
    });
  }

  const registrationRequestHash = await sha256Hex(JSON.stringify({
    pairing_id: pairingId,
    bootstrap_code_hash: await sha256Hex(code),
    os: input.os.trim().toLowerCase(),
    ed25519_public: input.ed25519Public.trim(),
    x25519_public: input.x25519Public.trim(),
    identity_fingerprint: input.identityFingerprint.trim().toLowerCase(),
    transport_mode: input.transportMode.trim() || "sealed",
    registration_proof: input.registrationProof.trim(),
    daemon_version: input.daemonVersion.trim(),
  }));

  const db = getD1();
  const replay = await db
    .prepare(
      `SELECT request_hash, response_ciphertext, response_nonce, expires_at
       FROM device_registration_replays WHERE pairing_id = ?`,
    )
    .bind(pairingId)
    .first<{
      request_hash: string;
      response_ciphertext: string;
      response_nonce: string;
      expires_at: string;
    }>();
  if (replay && replay.expires_at > now) {
    if (!constantTimeEqualHex(replay.request_hash, registrationRequestHash)) {
      throw new DomainError("device_identity_conflict", "注册重试与原设备身份不一致", 409);
    }
    try {
      return await decryptRegistrationReplay<DeviceRegistrationResult>({
        retryKey: registrationRetryKey,
        pairingId,
        requestHash: registrationRequestHash,
        ciphertext: replay.response_ciphertext,
        nonce: replay.response_nonce,
      });
    } catch {
      throw new DomainError("device_identity_conflict", "注册重试与原设备身份不一致", 409);
    }
  }
  if (replay) {
    await db.prepare("DELETE FROM device_registration_replays WHERE pairing_id = ? AND expires_at <= ?")
      .bind(pairingId, now)
      .run();
  }
  const sourceHash = await consumeClaimSourceBudget({
    source: input.source,
    trustedSourceHash: input.trustedSourceHash,
    now,
  });
  const pairing = await db
    .prepare(
      `SELECT id, account_id, requested_name, code_hash, status, expires_at, os, failed_attempts, locked_at
       FROM pairing_requests WHERE id = ?`,
    )
    .bind(pairingId)
    .first<{
      id: string;
      account_id: string;
      requested_name: string;
      code_hash: string;
      status: string;
      expires_at: string;
      os: string;
      failed_attempts: number;
      locked_at: string | null;
    }>();
  const codeHash = await sha256Hex(code);
  const codeMatches = Boolean(
    pairing && constantTimeEqualHex(pairing.code_hash, codeHash),
  );
  if (!pairing || !codeMatches) {
    return rejectPairingClaim({
      pairingId,
      sourceHash,
      now,
      countWrongCode: Boolean(pairing),
    });
  }
  const authorization = await db
    .prepare(
      `SELECT authorizations.status
       FROM tenant_instances AS tenants
       INNER JOIN tenant_authorization_state AS authorizations
         ON authorizations.tenant_id = tenants.id
       WHERE tenants.account_id = ?`,
    )
    .bind(pairing.account_id)
    .first<{ status: string }>();
  if (authorization?.status === "suspended") {
    throw new DomainError("access_suspended", "租户访问已暂停", 403);
  }
  if (
    pairing.status !== "waiting" ||
    pairing.expires_at <= now ||
    pairing.locked_at ||
    pairing.failed_attempts >= 5 ||
    pairing.os !== input.os.trim().toLowerCase()
  ) {
    return rejectPairingClaim({
      pairingId,
      sourceHash,
      now,
      countWrongCode: false,
    });
  }
  const claimEntitlement = await evaluateEntitlement(pairing.account_id);
  if (claimEntitlement.mode === "none") {
    return rejectPairingAccess({
      pairingId,
      accountId: pairing.account_id,
      sourceHash,
      now,
    });
  }
  const transportMode = input.transportMode.trim();
  if (transportMode && transportMode !== "sealed") {
    return rejectPairingClaim({
      pairingId,
      sourceHash,
      now,
      countWrongCode: false,
    });
  }

  let identity;
  try {
    identity = await validateDeviceIdentity(input);
  } catch {
    return rejectPairingClaim({
      pairingId,
      sourceHash,
      now,
      countWrongCode: false,
    });
  }

  const existingIdentityHost = await db
    .prepare(
      `SELECT id, account_id, lifecycle, slot_state
       FROM hosts WHERE identity_fingerprint = ?`,
    )
    .bind(identity.fingerprint)
    .first<{
      id: string;
      account_id: string;
      lifecycle: string;
      slot_state: string;
    }>();
  if (
    (existingIdentityHost || input.registrationProof.trim()) &&
    !(await verifyDeviceRegistrationProof({
      bootstrapToken: input.bootstrapToken,
      os: input.os,
      ed25519Public: identity.ed25519Public,
      x25519Public: identity.x25519Public,
      identityFingerprint: identity.fingerprint,
      transportMode,
      registrationProof: input.registrationProof,
    }))
  ) {
    return rejectPairingClaim({
      pairingId,
      sourceHash,
      now,
      countWrongCode: false,
    });
  }
  if (
    existingIdentityHost &&
    (existingIdentityHost.account_id !== pairing.account_id ||
      existingIdentityHost.lifecycle !== "deactivated" ||
      existingIdentityHost.slot_state !== "released")
  ) {
    throw new DomainError("device_identity_conflict", "设备身份已属于其他有效主机", 409);
  }

  const suffix = pairingId.slice("pair_".length);
  const hostId = existingIdentityHost?.id ?? `host_${suffix}`;
  const credentialId = `credential_${suffix}`;
  const token = randomDeviceToken();
  const tokenHash = await sha256Hex(token);
  const spentPairingCodeHash = await sha256Hex(randomDeviceToken());
  const correlationId = newId("corr");
  const placement = await db
    .prepare(
      `SELECT placements.state AS placement_state, nodes.status AS node_status
       FROM tenant_instances AS tenants
       LEFT JOIN tenant_placements AS placements ON placements.tenant_id = tenants.id
       LEFT JOIN relay_nodes AS nodes ON nodes.id = placements.relay_node_id
       WHERE tenants.account_id = ?`,
    )
    .bind(pairing.account_id)
    .first<{ placement_state: string | null; node_status: string | null }>();
  const connectionReady = isWritableRelayPlacement(
    placement?.placement_state,
    placement?.node_status,
  );
  const registrationResponse: DeviceRegistrationResult = {
    device_id: hostId,
    token,
    name: pairing.requested_name,
    transport_mode: "sealed",
    connection_state: connectionReady ? "ready" : "provisioning",
    ...(connectionReady ? {} : { retry_after_seconds: 5 }),
  };
  const encryptedReplay = await encryptRegistrationReplay({
    retryKey: registrationRetryKey,
    pairingId,
    requestHash: registrationRequestHash,
    response: registrationResponse,
  });
  const replayExpiresAt = new Date(
    new Date(now).getTime() + DEVICE_REGISTRATION_REPLAY_TTL_MS,
  ).toISOString();
  const results = await db.batch([
    db
      .prepare(CLAIM_HOST_SQL)
      .bind(
        hostId,
        identity.ed25519Public,
        identity.x25519Public,
        identity.fingerprint,
        now,
        now,
        pairingId,
        codeHash,
        now,
        pairing.os,
        now,
        now,
        now,
        now,
        reportedVersion.version,
      ),
    db
      .prepare(CLAIM_CREDENTIAL_SQL)
      .bind(
        credentialId,
        hostId,
        tokenHash,
        now,
        now,
        pairingId,
        codeHash,
        now,
        pairing.os,
        now,
        now,
        now,
        now,
        hostId,
        identity.fingerprint,
      ),
    db
      .prepare(COMMIT_PAIRING_CLAIM_SQL)
      .bind(
        hostId,
        now,
        now,
        spentPairingCodeHash,
        pairingId,
        codeHash,
        now,
        pairing.os,
        now,
        now,
        now,
        now,
        credentialId,
        hostId,
      ),
    db
      .prepare(ADVANCE_TENANT_CREDENTIAL_AFTER_CLAIM_SQL)
      .bind(
        now,
        pairing.account_id,
        pairingId,
        hostId,
        now,
        credentialId,
        hostId,
        tokenHash,
      ),
    db
      .prepare(ADVANCE_AUTHORIZATION_AFTER_CLAIM_SQL)
      .bind(now, pairing.account_id, credentialId, hostId, tokenHash),
    db
      .prepare(
        `INSERT INTO pairing_claim_attempts
          (id, pairing_request_id, source_hash, outcome, created_at)
         SELECT ?, ?, ?, 'claimed', ?
         WHERE EXISTS (
           SELECT 1 FROM pairing_requests
           WHERE id = ? AND status = 'claimed' AND claimed_host_id = ? AND claimed_at = ?
         )
           AND EXISTS (
             SELECT 1 FROM device_credentials
             WHERE id = ? AND host_id = ? AND token_hash = ? AND status = 'active'
           )`,
      )
      .bind(
        newId("claim"),
        pairingId,
        sourceHash,
        now,
        pairingId,
        hostId,
        now,
        credentialId,
        hostId,
        tokenHash,
      ),
    db
      .prepare(
        `INSERT INTO audit_events
          (id, actor_id, action, target_type, target_id, reason,
           after_json, correlation_id, created_at)
         SELECT ?, 'daemon:registration', ?, 'host', ?, ?, ?, ?, ?
         WHERE EXISTS (
           SELECT 1 FROM pairing_requests
           WHERE id = ? AND status = 'claimed' AND claimed_host_id = ? AND claimed_at = ?
         )
           AND EXISTS (
             SELECT 1 FROM device_credentials
             WHERE id = ? AND host_id = ? AND token_hash = ? AND status = 'active'
           )`,
      )
      .bind(
        newId("audit"),
        existingIdentityHost ? "host.recovered" : "host.claimed",
        hostId,
        existingIdentityHost
          ? "daemon 证明原身份私钥并恢复已撤销主机"
          : "daemon 使用一次性凭证认领主机",
        JSON.stringify({
          os: pairing.os,
          connectionState: "provisioning",
          recovered: Boolean(existingIdentityHost),
        }),
        correlationId,
        now,
        pairingId,
        hostId,
        now,
        credentialId,
        hostId,
        tokenHash,
      ),
    db
      .prepare(
        `INSERT INTO device_registration_replays
          (pairing_id, request_hash, response_ciphertext, response_nonce, expires_at, created_at)
         SELECT ?, ?, ?, ?, ?, ?
         WHERE EXISTS (
           SELECT 1 FROM pairing_requests
           WHERE id = ? AND status = 'claimed' AND claimed_host_id = ?
         )
           AND EXISTS (
             SELECT 1 FROM device_credentials
             WHERE id = ? AND host_id = ? AND token_hash = ? AND status = 'active'
           )`,
      )
      .bind(
        pairingId,
        registrationRequestHash,
        encryptedReplay.ciphertext,
        encryptedReplay.nonce,
        replayExpiresAt,
        now,
        pairingId,
        hostId,
        credentialId,
        hostId,
        tokenHash,
      ),
  ]);

  if (results.some((result) => Number(result.meta.changes ?? 0) !== 1)) {
    const currentEntitlement = await evaluateEntitlement(pairing.account_id);
    if (currentEntitlement.mode === "none") {
      return rejectPairingAccess({
        pairingId,
        accountId: pairing.account_id,
        sourceHash,
        now,
      });
    }
    if (
      !currentEntitlement.unlimited &&
      currentEntitlement.availableSlots !== null &&
      currentEntitlement.availableSlots <= 0
    ) {
      throw new DomainError(
        "device_capacity_exceeded",
        "当前没有可用主机席位",
        409,
      );
    }
    return rejectPairingClaim({
      pairingId,
      sourceHash,
      now,
      countWrongCode: false,
    });
  }

  return registrationResponse;
}

export async function revokeHost(input: {
  accountId: string;
  hostId: string;
  actorId: string;
  reason: string;
}): Promise<{ hostId: string; revoked: boolean }> {
  await ensureDatabase();
  assertReason(input.reason);
  const db = getD1();
  const host = await db
    .prepare("SELECT * FROM hosts WHERE id = ? AND account_id = ?")
    .bind(input.hostId, input.accountId)
    .first<HostRecord>();
  if (!host) throw new DomainError("host_not_found", "主机不存在", 404);
  if (host.lifecycle !== "active") return { hostId: host.id, revoked: false };

  const now = new Date().toISOString();
  const correlationId = newId("corr");
  const results = await db.batch([
    db
      .prepare(REVOKE_ACTIVE_DEVICE_CREDENTIALS_SQL)
      .bind(now, host.id),
    db
      .prepare(DEACTIVATE_OWNED_HOST_SQL)
      .bind(now, host.id, input.accountId),
    db
      .prepare(
        `UPDATE tenant_instances
         SET credential_revision = credential_revision + 1,
             updated_at = ?
         WHERE account_id = ?
           AND EXISTS (
             SELECT 1 FROM hosts
             WHERE id = ? AND lifecycle = 'deactivated' AND deactivated_at = ?
           )`,
      )
      .bind(now, input.accountId, host.id, now),
    db
      .prepare(ADVANCE_AUTHORIZATION_AFTER_REVOKE_SQL)
      .bind(now, input.accountId, host.id, now),
    db
      .prepare(
        `INSERT INTO audit_events
          (id, actor_id, action, target_type, target_id, reason,
           before_json, after_json, correlation_id, created_at)
         SELECT ?, ?, 'host.revoked', 'host', ?, ?, ?, ?, ?, ?
         WHERE EXISTS (
           SELECT 1 FROM hosts WHERE id = ? AND lifecycle = 'deactivated'
             AND deactivated_at = ?
         )`,
      )
      .bind(
        newId("audit"),
        input.actorId,
        host.id,
        input.reason.trim(),
        JSON.stringify({ lifecycle: host.lifecycle, slotState: host.slot_state }),
        JSON.stringify({ lifecycle: "deactivated", slotState: "released" }),
        correlationId,
        now,
        host.id,
        now,
      ),
  ]);
  if (
    Number(results[1]?.meta.changes ?? 0) !== 1 ||
    Number(results[2]?.meta.changes ?? 0) !== 1 ||
    Number(results[3]?.meta.changes ?? 0) !== 1 ||
    Number(results[4]?.meta.changes ?? 0) !== 1
  ) {
    throw new DomainError("host_revoke_failed", "主机撤销失败，请重试", 409);
  }
  return { hostId: host.id, revoked: true };
}

export async function createOrderQuote(input: {
  accountId: string;
  actorId: string;
  period: "month" | "year";
  quantity: number;
  idempotencyKey: string;
}): Promise<OrderRecord> {
  assertPaidFeaturesDeferred();
  await ensureDatabase();
  assertBillingPeriod(input.period);
  if (!Number.isInteger(input.quantity) || input.quantity < 1 || input.quantity > 100) {
    throw new DomainError("invalid_quantity", "主机槽位数量需为 1 到 100");
  }
  assertIdempotencyKey(input.idempotencyKey);

  const db = getD1();
  const price = await db
    .prepare(
      `SELECT id, billing_period, amount_minor, currency, tax_mode, status,
              effective_from
       FROM price_versions
       WHERE product_code = 'host_slot' AND billing_period = ?
         AND status = 'published'
       ORDER BY effective_from DESC LIMIT 1`,
    )
    .bind(input.period)
    .first<PriceRecord>();
  if (!price) throw new DomainError("price_unavailable", "当前价格不可用", 409);

  const requestHash = await hashText(
    JSON.stringify({ accountId: input.accountId, period: input.period, quantity: input.quantity }),
  );
  const scope = `quote:${input.accountId}`;
  const existing = await db
    .prepare("SELECT * FROM idempotency_records WHERE scope = ? AND key = ?")
    .bind(scope, input.idempotencyKey)
    .first<{ request_hash: string; response_json: string }>();
  if (existing) {
    if (existing.request_hash !== requestHash) {
      throw new DomainError("idempotency_conflict", "同一幂等键不能用于不同报价", 409);
    }
    return JSON.parse(existing.response_json) as OrderRecord;
  }

  const start = new Date();
  const order: OrderRecord = {
    id: newId("order"),
    client_order_id: input.idempotencyKey,
    billing_period: input.period,
    slot_quantity: input.quantity,
    amount_minor: price.amount_minor * input.quantity,
    currency: price.currency,
    status: "confirmation_required",
    term_start: start.toISOString(),
    term_end: addBillingPeriod(start, input.period),
    quote_expires_at: isoAfterMinutes(15),
    created_at: start.toISOString(),
  };
  const responseJson = JSON.stringify(order);
  const correlationId = newId("corr");

  try {
    await db.batch([
      db
        .prepare(
          `INSERT INTO idempotency_records
            (scope, key, request_hash, response_json, status_code, expires_at, created_at)
           VALUES (?, ?, ?, ?, 201, ?, ?)`,
        )
        .bind(scope, input.idempotencyKey, requestHash, responseJson, isoAfterMinutes(60), start.toISOString()),
      db
        .prepare(
          `INSERT INTO orders
            (id, account_id, client_order_id, kind, price_version_id,
             slot_quantity, term_start, term_end, amount_minor, currency,
             status, quote_expires_at, created_at, updated_at)
           VALUES (?, ?, ?, 'purchase', ?, ?, ?, ?, ?, ?,
                   'confirmation_required', ?, ?, ?)`,
        )
        .bind(
          order.id,
          input.accountId,
          order.client_order_id,
          price.id,
          order.slot_quantity,
          order.term_start,
          order.term_end,
          order.amount_minor,
          order.currency,
          order.quote_expires_at,
          order.created_at,
          order.created_at,
        ),
      db
        .prepare(
          `INSERT INTO audit_events
            (id, actor_id, action, target_type, target_id, reason,
             after_json, correlation_id, created_at)
           VALUES (?, ?, 'order.quoted', 'order', ?,
                   '用户创建供应商无关报价；原型不收款', ?, ?, ?)`,
        )
        .bind(newId("audit"), input.actorId, order.id, responseJson, correlationId, order.created_at),
    ]);
  } catch (error) {
    const raced = await db
      .prepare("SELECT request_hash, response_json FROM idempotency_records WHERE scope = ? AND key = ?")
      .bind(scope, input.idempotencyKey)
      .first<{ request_hash: string; response_json: string }>();
    if (raced?.request_hash === requestHash) {
      return JSON.parse(raced.response_json) as OrderRecord;
    }
    throw error;
  }

  return order;
}

async function assertIdempotencyAvailable(
  scope: string,
  key: string,
  requestHash: string,
): Promise<{ replay: boolean; responseJson?: string }> {
  if (!key) throw new DomainError("idempotency_required", "缺少幂等键");
  const existing = await getD1()
    .prepare("SELECT request_hash, response_json FROM idempotency_records WHERE scope = ? AND key = ?")
    .bind(scope, key)
    .first<{ request_hash: string; response_json: string }>();
  if (!existing) return { replay: false };
  if (existing.request_hash !== requestHash) {
    throw new DomainError("idempotency_conflict", "同一幂等键不能用于不同管理动作", 409);
  }
  return { replay: true, responseJson: existing.response_json };
}

async function commitIdempotentBatch<T>(input: {
  db: D1Database;
  statements: D1PreparedStatement[];
  scope: string;
  key: string;
  requestHash: string;
}): Promise<T | null> {
  try {
    await input.db.batch(input.statements);
    return null;
  } catch (error) {
    const raced = await input.db
      .prepare(
        "SELECT request_hash, response_json FROM idempotency_records WHERE scope = ? AND key = ?",
      )
      .bind(input.scope, input.key)
      .first<{ request_hash: string; response_json: string }>();

    if (raced) {
      if (raced.request_hash !== input.requestHash) {
        throw new DomainError(
          "idempotency_conflict",
          "同一幂等键不能用于不同请求",
          409,
        );
      }
      return JSON.parse(raced.response_json) as T;
    }
    throw error;
  }
}

export async function setPublicBeta(input: {
  actorId: string;
  enabled: boolean;
  capacitySlots: number | null;
  reason: string;
  idempotencyKey: string;
}): Promise<BetaRecord> {
  await ensureDatabase();
  if (typeof input.enabled !== "boolean") {
    throw new DomainError("invalid_beta_state", "公测状态必须是布尔值");
  }
  assertReason(input.reason);
  assertIdempotencyKey(input.idempotencyKey);
  if (input.capacitySlots !== null && (!Number.isInteger(input.capacitySlots) || input.capacitySlots < 1)) {
    throw new DomainError("invalid_capacity", "容量必须为空（不按槽位限制）或正整数");
  }
  const payload = {
    enabled: input.enabled,
    capacitySlots: input.capacitySlots,
    reason: input.reason.trim(),
  };
  const requestHash = await hashText(JSON.stringify(payload));
  const scope = "admin:beta";
  const replay = await assertIdempotencyAvailable(scope, input.idempotencyKey, requestHash);
  if (replay.replay) return JSON.parse(replay.responseJson!) as BetaRecord;

  const db = getD1();
  const previous = await getLatestBeta();
  const now = new Date().toISOString();
  const record: BetaRecord = {
    id: newId("beta"),
    state: input.enabled ? "active" : "ended",
    capacity_slots: input.capacitySlots,
    starts_at: now,
    ends_at: input.enabled ? null : now,
    grace_days: 0,
    created_at: now,
  };
  const responseJson = JSON.stringify(record);
  const correlationId = newId("corr");

  const racedRecord = await commitIdempotentBatch<BetaRecord>({
    db,
    scope,
    key: input.idempotencyKey,
    requestHash,
    statements: [
    db
      .prepare(
        `INSERT INTO idempotency_records
          (scope, key, request_hash, response_json, status_code, expires_at, created_at)
         VALUES (?, ?, ?, ?, 200, ?, ?)`,
      )
      .bind(scope, input.idempotencyKey, requestHash, responseJson, isoAfterMinutes(24 * 60), now),
    db
      .prepare(
        `UPDATE beta_programs SET state = 'retired', retired_at = ?
         WHERE state IN ('active', 'ending_soon', 'ended')`,
      )
      .bind(now),
    db
      .prepare(
        `INSERT INTO beta_programs
          (id, state, capacity_slots, starts_at, ends_at, grace_days, created_by, created_at)
         VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
      )
      .bind(
        record.id,
        record.state,
        record.capacity_slots,
        record.starts_at,
        record.ends_at,
        record.grace_days,
        input.actorId,
        now,
      ),
    db
      .prepare(FULFILL_ACCESS_REQUESTS_BY_PUBLIC_BETA_SQL)
      .bind(PUBLIC_BETA_ACCESS_RESPONSE, input.actorId, now),
    db
      .prepare(AUDIT_PUBLIC_BETA_ACCESS_FULFILLMENT_SQL)
      .bind(input.actorId, PUBLIC_BETA_ACCESS_RESPONSE, correlationId, now),
    db
      .prepare(
        `INSERT INTO audit_events
          (id, actor_id, action, target_type, target_id, reason,
           before_json, after_json, correlation_id, created_at)
         VALUES (?, ?, 'beta.versioned', 'beta_program', ?, ?, ?, ?, ?, ?)`,
      )
      .bind(
        newId("audit"),
        input.actorId,
        record.id,
        input.reason.trim(),
        previous ? JSON.stringify(previous) : null,
        responseJson,
        correlationId,
        now,
      ),
    ],
  });
  if (racedRecord) return racedRecord;
  return record;
}

export async function createExemption(input: {
  actorId: string;
  accountId: string;
  capacitySlots: number | null;
  endsAt: string;
  reason: string;
  idempotencyKey: string;
}): Promise<GrantRecord> {
  await ensureDatabase();
  const accountId = input.accountId.trim();
  if (!/^acct_[a-f0-9]{32}$/.test(accountId)) {
    throw new DomainError("invalid_account_id", "账户标识无效");
  }
  assertReason(input.reason);
  assertIdempotencyKey(input.idempotencyKey);
  const end = new Date(input.endsAt);
  if (Number.isNaN(end.valueOf()) || end <= new Date()) {
    throw new DomainError("invalid_expiry", "闭测邀请到期时间必须在未来");
  }
  if (input.capacitySlots !== null && (!Number.isInteger(input.capacitySlots) || input.capacitySlots < 1)) {
    throw new DomainError("invalid_capacity", "闭测邀请容量必须为空或正整数");
  }

  const db = getD1();
  const account = await db
    .prepare("SELECT * FROM accounts WHERE id = ?")
    .bind(accountId)
    .first<AccountRecord>();
  if (!account) throw new DomainError("account_not_found", "没有找到该账户", 404);

  const payload = {
    accountId: account.id,
    capacitySlots: input.capacitySlots,
    endsAt: end.toISOString(),
    reason: input.reason.trim(),
  };
  const requestHash = await hashText(JSON.stringify(payload));
  const scope = "admin:exemption";
  const replay = await assertIdempotencyAvailable(scope, input.idempotencyKey, requestHash);
  if (replay.replay) return JSON.parse(replay.responseJson!) as GrantRecord;

  const now = new Date().toISOString();
  const grant: GrantRecord = {
    id: newId("grant"),
    account_id: account.id,
    host_id: null,
    source: "admin_exemption",
    capacity_slots: input.capacitySlots,
    starts_at: now,
    ends_at: end.toISOString(),
    state: "active",
    reason: input.reason.trim(),
    created_at: now,
    revoked_at: null,
  };
  const responseJson = JSON.stringify(grant);
  const correlationId = newId("corr");
  const auditId = `audit_${(await hashText(`${scope}:${input.idempotencyKey}`)).slice(0, 32)}`;
  await db.batch([
    db
      .prepare(CREATE_MANUAL_INVITATION_IDEMPOTENCY_SQL)
      .bind(scope, input.idempotencyKey, requestHash, responseJson, isoAfterMinutes(24 * 60), now, account.id),
    db
      .prepare(CREATE_MANUAL_INVITATION_SQL)
      .bind(
        grant.id,
        grant.account_id,
        grant.id,
        grant.capacity_slots,
        grant.starts_at,
        grant.ends_at,
        grant.reason,
        input.actorId,
        now,
        scope,
        input.idempotencyKey,
        requestHash,
      ),
    db
      .prepare(
        `INSERT INTO audit_events
          (id, actor_id, action, target_type, target_id, reason,
           after_json, correlation_id, created_at)
         SELECT ?, ?, 'entitlement.invitation_created', 'entitlement_grant', ?,
                ?, ?, ?, ?
         FROM entitlement_grants
         WHERE id = ? AND source_ref = ?`,
      )
      .bind(auditId, input.actorId, grant.id, grant.reason, responseJson, correlationId, now, grant.id, grant.id),
  ]);

  const [stored, current, audit] = await Promise.all([
    db
      .prepare("SELECT request_hash, response_json FROM idempotency_records WHERE scope = ? AND key = ?")
      .bind(scope, input.idempotencyKey)
      .first<{ request_hash: string; response_json: string }>(),
    db.prepare("SELECT * FROM entitlement_grants WHERE id = ?").bind(grant.id).first<GrantRecord>(),
    db.prepare("SELECT id FROM audit_events WHERE id = ?").bind(auditId).first<{ id: string }>(),
  ]);
  if (stored?.request_hash === requestHash) {
    const committed = JSON.parse(stored.response_json) as GrantRecord;
    if (current?.id !== committed.id || audit?.id !== auditId) {
      throw new DomainError("invitation_create_indeterminate", "闭测邀请签发结果需要人工核对", 500);
    }
    return committed;
  }
  if (stored) throw new DomainError("idempotency_conflict", "同一幂等键不能用于不同请求", 409);
  const pending = await db
    .prepare("SELECT id FROM beta_access_requests WHERE account_id = ? AND status = 'requested'")
    .bind(account.id)
    .first<{ id: string }>();
  if (pending) {
    throw new DomainError("access_request_requires_resolution", "该账户有待处理闭测申请，请从申请队列批准或拒绝", 409);
  }
  throw new DomainError("invitation_create_failed", "闭测邀请签发失败，请重试", 409);
}

export async function revokeExemption(input: {
  actorId: string;
  grantId: string;
  reason: string;
  idempotencyKey: string;
}): Promise<GrantRecord> {
  await ensureDatabase();
  const grantId = input.grantId.trim();
  if (!/^grant_[a-f0-9]{32}$/.test(grantId)) {
    throw new DomainError("invalid_grant_id", "闭测邀请标识无效");
  }
  assertReason(input.reason);
  assertIdempotencyKey(input.idempotencyKey);

  const payload = { grantId, reason: input.reason.trim() };
  const requestHash = await hashText(JSON.stringify(payload));
  const scope = "admin:exemption:revoke";
  const replay = await assertIdempotencyAvailable(
    scope,
    input.idempotencyKey,
    requestHash,
  );
  if (replay.replay) return JSON.parse(replay.responseJson!) as GrantRecord;

  const db = getD1();
  const previous = await db
    .prepare(
      `SELECT * FROM entitlement_grants
       WHERE id = ? AND source = 'admin_exemption'`,
    )
    .bind(grantId)
    .first<GrantRecord>();
  if (!previous) {
    throw new DomainError("invitation_not_found", "没有找到该闭测邀请", 404);
  }
  if (previous.state !== "active" || previous.revoked_at) {
    throw new DomainError("invitation_not_active", "该闭测邀请已经失效", 409);
  }

  const now = new Date().toISOString();
  const invitation: GrantRecord = {
    ...previous,
    state: "revoked",
    revoked_at: now,
  };
  const responseJson = JSON.stringify(invitation);
  const correlationId = newId("corr");
  const auditId = `audit_${(await hashText(`${scope}:${input.idempotencyKey}`)).slice(0, 32)}`;
  const results = await db.batch([
    db
      .prepare(CREATE_INVITATION_REVOCATION_IDEMPOTENCY_SQL)
      .bind(
        scope,
        input.idempotencyKey,
        requestHash,
        responseJson,
        isoAfterMinutes(24 * 60),
        now,
        grantId,
      ),
    db
      .prepare(REVOKE_INVITATION_SQL)
      .bind(now, grantId, scope, input.idempotencyKey, requestHash),
    db
      .prepare(AUDIT_INVITATION_REVOCATION_SQL)
      .bind(
        auditId,
        input.actorId,
        grantId,
        input.reason.trim(),
        JSON.stringify(previous),
        responseJson,
        correlationId,
        now,
        scope,
        input.idempotencyKey,
        requestHash,
      ),
  ]);

  const [stored, current, audit] = await Promise.all([
    db
      .prepare(
        `SELECT request_hash, response_json FROM idempotency_records
         WHERE scope = ? AND key = ?`,
      )
      .bind(scope, input.idempotencyKey)
      .first<{ request_hash: string; response_json: string }>(),
    db
      .prepare("SELECT * FROM entitlement_grants WHERE id = ?")
      .bind(grantId)
      .first<GrantRecord>(),
    db
      .prepare("SELECT id FROM audit_events WHERE id = ?")
      .bind(auditId)
      .first<{ id: string }>(),
  ]);
  if (stored?.request_hash === requestHash) {
    const committed = JSON.parse(stored.response_json) as GrantRecord;
    if (
      current?.state !== "revoked"
      || current.revoked_at !== committed.revoked_at
      || audit?.id !== auditId
    ) {
      throw new DomainError(
        "invitation_revoke_indeterminate",
        "闭测邀请撤销结果需要人工核对",
        500,
      );
    }
    return committed;
  }
  if (stored) {
    throw new DomainError(
      "idempotency_conflict",
      "同一幂等键不能用于不同请求",
      409,
    );
  }
  if (Number(results[0]?.meta.changes ?? 0) !== 1) {
    throw new DomainError("invitation_not_active", "该闭测邀请已经失效", 409);
  }
  throw new DomainError("invitation_revoke_failed", "闭测邀请撤销失败，请重试", 409);
}

export type BetaAccessResolutionResult = {
  request: BetaAccessRequestRecord;
  grant: GrantRecord | null;
};

export async function resolveBetaAccessRequest(input: {
  actorId: string;
  requestId: string;
  action: "approve" | "decline";
  capacitySlots: number;
  endsAt: string;
  response: string;
  reason: string;
  idempotencyKey: string;
}): Promise<BetaAccessResolutionResult> {
  await ensureDatabase();
  if (!/^access_[a-f0-9]{32}$/.test(input.requestId)) {
    throw new DomainError("invalid_access_request_id", "闭测申请编号无效");
  }
  if (input.action !== "approve" && input.action !== "decline") {
    throw new DomainError("invalid_access_resolution", "请选择批准或拒绝");
  }
  const response = input.response.trim();
  if (response.length < 2 || response.length > 500) {
    throw new DomainError("invalid_access_response", "给用户的说明需为 2 到 500 个字符");
  }
  assertReason(input.reason);
  assertIdempotencyKey(input.idempotencyKey);

  let end: Date | null = null;
  if (input.action === "approve") {
    if (!Number.isInteger(input.capacitySlots) || input.capacitySlots < 1 || input.capacitySlots > 3) {
      throw new DomainError("invalid_access_capacity", "闭测邀请只能批准 1 到 3 台主机");
    }
    end = new Date(input.endsAt);
    if (Number.isNaN(end.valueOf()) || end <= new Date()) {
      throw new DomainError("invalid_expiry", "闭测邀请到期时间必须在未来");
    }
  }

  const payload = {
    requestId: input.requestId,
    action: input.action,
    capacitySlots: input.action === "approve" ? input.capacitySlots : null,
    endsAt: end?.toISOString() ?? null,
    response,
    reason: input.reason.trim(),
  };
  const requestHash = await hashText(JSON.stringify(payload));
  const scope = "admin:access-request:resolve";
  const replay = await assertIdempotencyAvailable(scope, input.idempotencyKey, requestHash);
  if (replay.replay) return JSON.parse(replay.responseJson!) as BetaAccessResolutionResult;

  const db = getD1();
  const previous = await db
    .prepare("SELECT * FROM beta_access_requests WHERE id = ?")
    .bind(input.requestId)
    .first<BetaAccessRequestRecord>();
  if (!previous) throw new DomainError("access_request_not_found", "没有找到该闭测申请", 404);
  if (previous.status !== "requested") {
    throw new DomainError("access_request_not_pending", "该闭测申请已经处理", 409);
  }

  const now = new Date().toISOString();
  const grant: GrantRecord | null = input.action === "approve"
    ? {
        id: newId("grant"),
        account_id: previous.account_id,
        host_id: null,
        source: "admin_exemption",
        capacity_slots: input.capacitySlots,
        starts_at: now,
        ends_at: end!.toISOString(),
        state: "active",
        reason: input.reason.trim(),
        created_at: now,
        revoked_at: null,
      }
    : null;
  const resolved: BetaAccessRequestRecord = {
    ...previous,
    status: input.action === "approve" ? "approved" : "declined",
    admin_response: response,
    resolved_by: input.actorId,
    invitation_grant_id: grant?.id ?? null,
    resolved_at: now,
    updated_at: now,
  };
  const result: BetaAccessResolutionResult = { request: resolved, grant };
  const responseJson = JSON.stringify(result);
  const requestAuditId = `audit_${(await hashText(`${scope}:${input.idempotencyKey}:request`)).slice(0, 32)}`;
  const grantAuditId = `audit_${(await hashText(`${scope}:${input.idempotencyKey}:grant`)).slice(0, 32)}`;
  const correlationId = newId("corr");
  const statements: D1PreparedStatement[] = [
    db
      .prepare(CREATE_ACCESS_RESOLUTION_IDEMPOTENCY_SQL)
      .bind(
        scope,
        input.idempotencyKey,
        requestHash,
        responseJson,
        200,
        isoAfterMinutes(24 * 60),
        now,
        input.requestId,
      ),
  ];
  if (grant) {
    statements.push(
      db
        .prepare(CREATE_APPROVED_INVITATION_SQL)
        .bind(
          grant.id,
          input.requestId,
          grant.capacity_slots,
          now,
          grant.ends_at,
          grant.reason,
          input.actorId,
          input.requestId,
          scope,
          input.idempotencyKey,
          requestHash,
        ),
      db
        .prepare(APPROVE_ACCESS_REQUEST_SQL)
        .bind(
          response,
          input.actorId,
          grant.id,
          now,
          input.requestId,
          scope,
          input.idempotencyKey,
          requestHash,
        ),
    );
  } else {
    statements.push(
      db
        .prepare(DECLINE_ACCESS_REQUEST_SQL)
        .bind(
          response,
          input.actorId,
          now,
          input.requestId,
          scope,
          input.idempotencyKey,
          requestHash,
        ),
    );
  }
  statements.push(
    db
      .prepare(
        `INSERT OR IGNORE INTO audit_events
          (id, actor_id, action, target_type, target_id, reason,
           before_json, after_json, correlation_id, created_at)
         SELECT ?, ?, ?, 'beta_access_request', ?, ?, ?, ?, ?, ?
         FROM beta_access_requests
         WHERE id = ? AND status = ? AND resolved_at = ?`,
      )
      .bind(
        requestAuditId,
        input.actorId,
        input.action === "approve" ? "beta_access.approved" : "beta_access.declined",
        input.requestId,
        input.reason.trim(),
        JSON.stringify({ status: previous.status }),
        JSON.stringify({ status: resolved.status, adminResponse: response, invitationGrantId: grant?.id ?? null }),
        correlationId,
        now,
        input.requestId,
        resolved.status,
        now,
      ),
  );
  if (grant) {
    statements.push(
      db
        .prepare(
          `INSERT OR IGNORE INTO audit_events
            (id, actor_id, action, target_type, target_id, reason,
             after_json, correlation_id, created_at)
           SELECT ?, ?, 'entitlement.invitation_created', 'entitlement_grant', ?,
                  ?, ?, ?, ?
           FROM entitlement_grants
           WHERE id = ? AND source_ref = ?`,
        )
        .bind(
          grantAuditId,
          input.actorId,
          grant.id,
          grant.reason,
          JSON.stringify({ accountId: grant.account_id, capacitySlots: grant.capacity_slots, endsAt: grant.ends_at }),
          correlationId,
          now,
          grant.id,
          input.requestId,
        ),
    );
  }
  await db.batch(statements);

  const [stored, current, requestAudit, currentGrant, grantAudit] = await Promise.all([
    db
      .prepare("SELECT request_hash, response_json FROM idempotency_records WHERE scope = ? AND key = ?")
      .bind(scope, input.idempotencyKey)
      .first<{ request_hash: string; response_json: string }>(),
    db.prepare("SELECT * FROM beta_access_requests WHERE id = ?").bind(input.requestId).first<BetaAccessRequestRecord>(),
    db.prepare("SELECT id FROM audit_events WHERE id = ?").bind(requestAuditId).first<{ id: string }>(),
    grant ? db.prepare("SELECT * FROM entitlement_grants WHERE id = ?").bind(grant.id).first<GrantRecord>() : Promise.resolve(null),
    grant ? db.prepare("SELECT id FROM audit_events WHERE id = ?").bind(grantAuditId).first<{ id: string }>() : Promise.resolve(null),
  ]);
  if (stored?.request_hash === requestHash) {
    const committed = JSON.parse(stored.response_json) as BetaAccessResolutionResult;
    if (
      current?.status !== committed.request.status
      || current.resolved_at !== committed.request.resolved_at
      || requestAudit?.id !== requestAuditId
      || (grant && (currentGrant?.id !== committed.grant?.id || grantAudit?.id !== grantAuditId))
    ) {
      throw new DomainError("access_resolution_indeterminate", "闭测申请审批结果需要人工核对", 500);
    }
    return committed;
  }
  if (stored) throw new DomainError("idempotency_conflict", "同一幂等键不能用于不同请求", 409);
  throw new DomainError("access_request_not_pending", "该闭测申请已经处理", 409);
}

export async function publishPrice(input: {
  actorId: string;
  period: "month" | "year";
  amountMinor: number;
  reason: string;
  idempotencyKey: string;
}): Promise<PriceRecord> {
  assertPaidFeaturesDeferred();
  await ensureDatabase();
  assertBillingPeriod(input.period);
  if (!Number.isInteger(input.amountMinor) || input.amountMinor < 1) {
    throw new DomainError("invalid_amount", "价格必须是正整数分");
  }
  assertReason(input.reason);
  assertIdempotencyKey(input.idempotencyKey);
  const payload = { period: input.period, amountMinor: input.amountMinor, reason: input.reason.trim() };
  const requestHash = await hashText(JSON.stringify(payload));
  const scope = "admin:price";
  const replay = await assertIdempotencyAvailable(scope, input.idempotencyKey, requestHash);
  if (replay.replay) return JSON.parse(replay.responseJson!) as PriceRecord;

  const db = getD1();
  const now = new Date().toISOString();
  const previous = await db
    .prepare(
      `SELECT id, billing_period, amount_minor, currency, tax_mode, status, effective_from
       FROM price_versions WHERE product_code = 'host_slot'
         AND billing_period = ? AND status = 'published'
       ORDER BY effective_from DESC LIMIT 1`,
    )
    .bind(input.period)
    .first<PriceRecord>();
  const price: PriceRecord = {
    id: newId("price"),
    billing_period: input.period,
    amount_minor: input.amountMinor,
    currency: "CNY",
    tax_mode: "undecided",
    status: "published",
    effective_from: now,
  };
  const responseJson = JSON.stringify(price);
  const correlationId = newId("corr");

  const racedPrice = await commitIdempotentBatch<PriceRecord>({
    db,
    scope,
    key: input.idempotencyKey,
    requestHash,
    statements: [
    db
      .prepare(
        `INSERT INTO idempotency_records
          (scope, key, request_hash, response_json, status_code, expires_at, created_at)
         VALUES (?, ?, ?, ?, 201, ?, ?)`,
      )
      .bind(scope, input.idempotencyKey, requestHash, responseJson, isoAfterMinutes(24 * 60), now),
    db
      .prepare(
        `UPDATE price_versions SET status = 'retired', effective_to = ?
         WHERE product_code = 'host_slot' AND billing_period = ? AND status = 'published'`,
      )
      .bind(now, input.period),
    db
      .prepare(
        `INSERT INTO price_versions
          (id, product_code, billing_period, unit_slots, amount_minor, currency,
           tax_mode, quote_ttl_seconds, status, effective_from, created_by, created_at)
         VALUES (?, 'host_slot', ?, 1, ?, 'CNY', 'undecided', 900,
                 'published', ?, ?, ?)`,
      )
      .bind(price.id, price.billing_period, price.amount_minor, now, input.actorId, now),
    db
      .prepare(
        `INSERT INTO audit_events
          (id, actor_id, action, target_type, target_id, reason,
           before_json, after_json, correlation_id, created_at)
         VALUES (?, ?, 'price.published', 'price_version', ?, ?, ?, ?, ?, ?)`,
      )
      .bind(
        newId("audit"),
        input.actorId,
        price.id,
        input.reason.trim(),
        previous ? JSON.stringify(previous) : null,
        responseJson,
        correlationId,
        now,
      ),
    ],
  });
  if (racedPrice) return racedPrice;
  return price;
}

export async function updateLaunchGate(input: {
  actorId: string;
  key: string;
  status: LaunchGateRecord["status"];
  owner: string;
  evidenceUrl: string;
  notes: string;
  reason: string;
  idempotencyKey: string;
}): Promise<LaunchGateRecord> {
  await ensureDatabase();
  assertReason(input.reason);
  assertIdempotencyKey(input.idempotencyKey);
  if (!launchGateStatuses.has(input.status)) {
    throw new DomainError("invalid_gate_status", "上线门禁状态无效");
  }
  if (input.owner.trim().length > 100) {
    throw new DomainError("owner_too_long", "负责人不能超过 100 个字符");
  }
  if (input.notes.trim().length > 2_000) {
    throw new DomainError("notes_too_long", "门禁说明不能超过 2000 个字符");
  }
  const evidenceUrl = input.evidenceUrl.trim();
  if (evidenceUrl) {
    if (evidenceUrl.length > 2_048) {
      throw new DomainError("evidence_url_too_long", "证据链接过长");
    }
    try {
      if (new URL(evidenceUrl).protocol !== "https:") {
        throw new Error("protocol");
      }
    } catch {
      throw new DomainError("invalid_evidence_url", "证据链接必须是有效的 HTTPS 地址");
    }
  }
  const db = getD1();
  const previous = await db
    .prepare("SELECT * FROM launch_gates WHERE key = ?")
    .bind(input.key)
    .first<LaunchGateRecord>();
  if (!previous) throw new DomainError("gate_not_found", "上线门禁不存在", 404);
  if (previous.priority === "P0" && input.status === "not_applicable") {
    throw new DomainError(
      "p0_not_applicable_forbidden",
      "P0 门禁不能标记为不适用；必须提交通过证据或保持阻止",
    );
  }
  if (input.status === "passed" || input.status === "not_applicable") {
    if (!input.owner.trim()) {
      throw new DomainError("gate_owner_required", "解除阻止前必须指定负责人");
    }
    if (!input.notes.trim()) {
      throw new DomainError("gate_evidence_notes_required", "解除阻止前必须填写证据摘要");
    }
    if (!evidenceUrl) {
      throw new DomainError("gate_evidence_url_required", "解除阻止前必须提交 HTTPS 证据链接");
    }
  }

  const payload = {
    key: input.key,
    status: input.status,
    owner: input.owner.trim(),
    evidenceUrl: input.evidenceUrl.trim(),
    notes: input.notes.trim(),
    reason: input.reason.trim(),
  };
  const requestHash = await hashText(JSON.stringify(payload));
  const scope = "admin:launch-gate";
  const replay = await assertIdempotencyAvailable(scope, input.idempotencyKey, requestHash);
  if (replay.replay) return JSON.parse(replay.responseJson!) as LaunchGateRecord;

  const now = new Date().toISOString();
  const gate: LaunchGateRecord = {
    ...previous,
    status: input.status,
    owner: input.owner.trim() || null,
    evidence_url: input.evidenceUrl.trim() || null,
    notes: input.notes.trim(),
    reviewed_at: now,
    updated_at: now,
  };
  const responseJson = JSON.stringify(gate);
  const correlationId = newId("corr");

  const racedGate = await commitIdempotentBatch<LaunchGateRecord>({
    db,
    scope,
    key: input.idempotencyKey,
    requestHash,
    statements: [
    db
      .prepare(
        `INSERT INTO idempotency_records
          (scope, key, request_hash, response_json, status_code, expires_at, created_at)
         VALUES (?, ?, ?, ?, 200, ?, ?)`,
      )
      .bind(scope, input.idempotencyKey, requestHash, responseJson, isoAfterMinutes(24 * 60), now),
    db
      .prepare(
        `UPDATE launch_gates SET status = ?, owner = ?, evidence_url = ?, notes = ?,
                                 reviewed_at = ?, updated_at = ?
         WHERE key = ?`,
      )
      .bind(gate.status, gate.owner, gate.evidence_url, gate.notes, now, now, gate.key),
    db
      .prepare(FULFILL_ACCESS_REQUESTS_BY_PUBLIC_BETA_SQL)
      .bind(PUBLIC_BETA_ACCESS_RESPONSE, input.actorId, now),
    db
      .prepare(AUDIT_PUBLIC_BETA_ACCESS_FULFILLMENT_SQL)
      .bind(input.actorId, PUBLIC_BETA_ACCESS_RESPONSE, correlationId, now),
    db
      .prepare(
        `INSERT INTO audit_events
          (id, actor_id, action, target_type, target_id, reason,
           before_json, after_json, correlation_id, created_at)
         VALUES (?, ?, 'launch_gate.updated', 'launch_gate', ?, ?, ?, ?, ?, ?)`,
      )
      .bind(
        newId("audit"),
        input.actorId,
        gate.key,
        input.reason.trim(),
        JSON.stringify(previous),
        responseJson,
        correlationId,
        now,
      ),
    ],
  });
  if (racedGate) return racedGate;
  return gate;
}

export async function getAdminSnapshot(): Promise<AdminSnapshot> {
  await ensureDatabase();
  const db = getD1();
  const [
    beta,
    accounts,
    grants,
    accessRequests,
    gates,
    audits,
    feedback,
    deletionRequests,
    incidents,
    serviceStatus,
    operations,
    hostContacts,
    retentionRecord,
  ] = await Promise.all([
    getLatestBeta(),
    all<AccountRecord>(db.prepare("SELECT * FROM accounts ORDER BY created_at DESC LIMIT 100")),
    all<GrantRecord>(
      db.prepare("SELECT * FROM entitlement_grants ORDER BY created_at DESC LIMIT 100"),
    ),
    all<BetaAccessRequestRecord>(
      db.prepare(
        `SELECT * FROM beta_access_requests
         ORDER BY status = 'requested' DESC, requested_at DESC, id DESC
         LIMIT 100`,
      ),
    ),
    all<LaunchGateRecord>(
      db.prepare("SELECT * FROM launch_gates ORDER BY priority, category, key"),
    ),
    all<AuditRecord>(
      db.prepare("SELECT * FROM audit_events ORDER BY created_at DESC LIMIT 50"),
    ),
    all<FeedbackRecord>(
      db.prepare(
        `SELECT * FROM beta_feedback
         ORDER BY status = 'open' DESC, created_at DESC
         LIMIT 100`,
      ),
    ),
    all<AccountDeletionRequestRecord>(
      db.prepare(
        `SELECT * FROM account_deletion_requests
         ORDER BY status = 'requested' DESC, requested_at DESC
         LIMIT 100`,
      ),
    ),
    all<ServiceIncidentRecord>(
      db.prepare(
        `SELECT * FROM service_incidents
         ORDER BY status = 'active' DESC, started_at DESC, id DESC
         LIMIT 40`,
      ),
    ),
    getServiceStatusSnapshot(),
    getBetaOperationsSnapshot(),
    getAdminHostControlPlaneSnapshot(),
    db
      .prepare("SELECT * FROM maintenance_jobs WHERE key = ?")
      .bind(RETENTION_JOB_KEY)
      .first<RetentionJobRecord>(),
  ]);
  return {
    beta,
    accounts,
    grants,
    accessRequests,
    gates,
    audits,
    feedback,
    deletionRequests,
    incidents,
    serviceStatus,
    operations,
    hostContacts,
    retention: deriveRetentionJobHealth(retentionRecord),
    blockedP0: countBlockedPublicBetaP0(gates),
  };
}

export async function getAdminHostControlPlaneSnapshot(): Promise<AdminHostControlPlaneSnapshot> {
  await ensureDatabase();
  const db = getD1();
  const generatedAt = new Date().toISOString();
  const cutoffs = controlPlaneContactCutoffs(generatedAt);
  const bindCutoffs = (statement: D1PreparedStatement) =>
    statement.bind(
      cutoffs.futureLimitAt,
      cutoffs.freshCutoff,
      cutoffs.delayedCutoff,
    );
  const [summary, attentionHosts] = await Promise.all([
    bindCutoffs(db.prepare(ADMIN_HOST_CONTROL_PLANE_SUMMARY_SQL)).first<
      Record<string, number | null>
    >(),
    all<AdminHostControlPlaneAttentionRow>(
      bindCutoffs(db.prepare(ADMIN_HOST_CONTROL_PLANE_ATTENTION_SQL)),
    ),
  ]);
  return deriveAdminHostControlPlaneSnapshot({
    generatedAt,
    summary: summary ?? {},
    attentionHosts,
  });
}

export async function getBetaOperationsSnapshot(): Promise<BetaOperationsSnapshot> {
  await ensureDatabase();
  const db = getD1();
  const generatedAt = new Date().toISOString();
  const cutoff = new Date(
    new Date(generatedAt).getTime() -
      BETA_OPERATIONS_WINDOW_DAYS * 24 * 60 * 60_000,
  ).toISOString();
  const [accounts, pairings, claimAttempts, provisioning, support, accessRequests] =
    await Promise.all([
      db.prepare(BETA_ACCOUNTS_SQL).bind(cutoff).first<Record<string, number | null>>(),
      db
        .prepare(BETA_PAIRINGS_SQL)
        .bind(generatedAt, generatedAt, generatedAt, cutoff)
        .first<Record<string, number | null>>(),
      db
        .prepare(BETA_CLAIM_ATTEMPTS_SQL)
        .bind(cutoff)
        .first<Record<string, number | null>>(),
      db
        .prepare(BETA_PROVISIONING_SQL)
        .bind(cutoff)
        .first<Record<string, number | null>>(),
      db
        .prepare(BETA_SUPPORT_SQL)
        .bind(cutoff)
        .first<Record<string, number | null>>(),
      db
        .prepare(BETA_ACCESS_REQUESTS_SQL)
        .bind(cutoff)
        .first<Record<string, number | null>>(),
    ]);

  return deriveBetaOperationsSnapshot({
    generatedAt,
    accounts: accounts ?? {},
    pairings: pairings ?? {},
    claimAttempts: claimAttempts ?? {},
    provisioning: provisioning ?? {},
    support: support ?? {},
    accessRequests: accessRequests ?? {},
  });
}

export async function getPublicCommercialSnapshot(): Promise<PublicCommercialSnapshot> {
  await ensureDatabase();
  const db = getD1();
  const [beta, gates] = await Promise.all([
    getActiveBeta(),
    all<LaunchGateRecord>(
      db.prepare(
        `SELECT * FROM launch_gates
         WHERE priority = 'P0'
         ORDER BY category, key`,
      ),
    ),
  ]);
  return {
    beta,
    gates,
    blockedP0: countBlockedPublicBetaP0(gates),
  };
}