Files

4006 lines
122 KiB
TypeScript

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),
};
}