import { ensureDatabase, getD1 } from "./bootstrap";
import { type RelayNodePrincipal } from "./relay-control-plane";
import { DomainError } from "./repository";
import { authorityAfterMigrationFailure } from "./relay-migration-state";

export type RelayMigrationState =
  | "quiescing"
  | "copying"
  | "switching"
  | "draining"
  | "completed"
  | "failed";

type MigrationRecord = {
  id: string;
  tenant_id: string;
  source_node_id: string;
  target_node_id: string;
  source_generation: number;
  target_generation: number;
  state: RelayMigrationState;
  backup_ref: string | null;
  manifest_sha256: string | null;
  requested_by: string;
  reason: string;
  started_at: string;
  updated_at: string;
  switched_at: string | null;
  completed_at: string | null;
  last_error_code: string | null;
};

export type RelayMigrationAssignment = {
  migration_id: string;
  tenant_id: string;
  role: "source" | "target";
  source_node_id: string;
  target_node_id: string;
  source_generation: number;
  target_generation: number;
  state: "quiescing" | "copying" | "switching" | "draining";
  backup_ref?: string;
  manifest_sha256?: string;
  finalize_after?: string;
};

function changed(result: D1Result<unknown>): boolean {
  return Number(result.meta.changes ?? 0) === 1;
}

function validTenantId(value: string): boolean {
  return /^tenant_[0-9a-f]{32}$/u.test(value);
}

function validNodeId(value: string): boolean {
  return /^node_[A-Za-z0-9][A-Za-z0-9._:-]{0,95}$/u.test(value);
}

function validMigrationId(value: string): boolean {
  return /^migration_[0-9a-f]{32}$/u.test(value);
}

function validBackupRef(value: string): boolean {
  return /^[0-9a-f]{32}\/g[0-9]{20}-[0-9]{8}T[0-9]{6}Z-[0-9a-f]{16}$/u.test(value);
}

function validHash(value: string): boolean {
  return /^[0-9a-f]{64}$/u.test(value);
}

function transitionAudit(
  db: D1Database,
  record: MigrationRecord,
  actorId: string,
  action: string,
  expectedState: RelayMigrationState,
  now: string,
): D1PreparedStatement {
  return db.prepare(
    `INSERT INTO audit_events
      (id, actor_id, action, target_type, target_id, reason, created_at)
     SELECT ?1, ?2, ?3, 'relay_migration', ?4, ?5, ?6
     WHERE EXISTS (
       SELECT 1 FROM relay_migrations WHERE id = ?4 AND state = ?7
     )`,
  ).bind(
    `audit_${crypto.randomUUID().replaceAll("-", "")}`,
    actorId,
    action,
    record.id,
    `migration ${record.tenant_id} ${expectedState}`,
    now,
    expectedState,
  );
}

async function deterministicMigrationId(actorId: string, idempotencyKey: string): Promise<string> {
  if (!/^[A-Za-z0-9._:-]{8,128}$/u.test(idempotencyKey)) {
    throw new DomainError("invalid_idempotency_key", "迁移幂等键无效");
  }
  const digest = await crypto.subtle.digest(
    "SHA-256",
    new TextEncoder().encode(`nekonest-cloud/relay-migration/v1\0${actorId}\0${idempotencyKey}`),
  );
  return `migration_${Array.from(new Uint8Array(digest).slice(0, 16), (byte) => byte.toString(16).padStart(2, "0")).join("")}`;
}

export async function beginRelayMigration(input: {
  tenantId: string;
  targetNodeId: string;
  actorId: string;
  reason: string;
  idempotencyKey: string;
}): Promise<MigrationRecord> {
  await ensureDatabase();
  const tenantId = input.tenantId.trim();
  const targetNodeId = input.targetNodeId.trim();
  const actorId = input.actorId.trim();
  const reason = input.reason.trim();
  if (!validTenantId(tenantId) || !validNodeId(targetNodeId) || !actorId || reason.length < 8 || reason.length > 500) {
    throw new DomainError("invalid_relay_migration", "迁移租户、目标节点或原因无效");
  }
  const migrationId = await deterministicMigrationId(actorId, input.idempotencyKey.trim());
  const db = getD1();
  const existing = await db
    .prepare("SELECT * FROM relay_migrations WHERE id = ?")
    .bind(migrationId)
    .first<MigrationRecord>();
  if (existing) {
    if (
      existing.tenant_id !== tenantId ||
      existing.target_node_id !== targetNodeId ||
      existing.requested_by !== actorId ||
      existing.reason !== reason
    ) {
      throw new DomainError("idempotency_conflict", "迁移幂等键已用于不同请求", 409);
    }
    return existing;
  }
  const now = new Date().toISOString();
  const results = await db.batch([
    db
      .prepare(
        `INSERT INTO relay_migrations
          (id, tenant_id, source_node_id, target_node_id,
           source_generation, target_generation, state,
           requested_by, reason, started_at, updated_at)
         SELECT ?1, placements.tenant_id, placements.relay_node_id, targets.id,
                placements.generation, placements.generation + 1, 'quiescing',
                ?4, ?5, ?6, ?6
         FROM tenant_placements AS placements
         INNER JOIN relay_nodes AS sources ON sources.id = placements.relay_node_id
         INNER JOIN relay_nodes AS targets ON targets.id = ?3
         WHERE placements.tenant_id = ?2 AND placements.state = 'active'
           AND placements.relay_node_id IS NOT NULL
           AND sources.status IN ('active', 'draining') AND targets.status = 'active'
           AND targets.id <> placements.relay_node_id
           AND NOT EXISTS (
             SELECT 1 FROM relay_migrations AS active
             WHERE active.tenant_id = placements.tenant_id
               AND active.state IN ('quiescing', 'copying', 'switching', 'draining')
           )
           AND (
             targets.capacity_tenants = 0 OR
             (SELECT COUNT(*) FROM tenant_placements AS occupied
              WHERE occupied.relay_node_id = targets.id
                AND occupied.state IN ('active', 'draining')) < targets.capacity_tenants
           )`,
      )
      .bind(migrationId, tenantId, targetNodeId, actorId, reason, now),
    db
      .prepare(
        `UPDATE tenant_placements
         SET state = 'quiescing', last_error_code = NULL, updated_at = ?1
         WHERE tenant_id = ?2 AND state = 'active'
           AND EXISTS (
             SELECT 1 FROM relay_migrations
             WHERE id = ?3 AND tenant_id = ?2
               AND source_node_id = tenant_placements.relay_node_id
               AND source_generation = tenant_placements.generation
               AND state = 'quiescing'
           )`,
      )
      .bind(now, tenantId, migrationId),
    db
      .prepare(
        `INSERT INTO audit_events
          (id, actor_id, action, target_type, target_id, reason, created_at)
         SELECT ?1, ?2, 'relay.migration.started', 'relay_migration', ?3, ?4, ?5
         WHERE EXISTS (
           SELECT 1 FROM relay_migrations
           WHERE id = ?3 AND tenant_id = ?6 AND state = 'quiescing'
         )`,
      )
      .bind(`audit_${crypto.randomUUID().replaceAll("-", "")}`, actorId, migrationId, reason, now, tenantId),
  ]);
  if (!changed(results[0]) || !changed(results[1])) {
    const raced = await db
      .prepare("SELECT * FROM relay_migrations WHERE id = ?")
      .bind(migrationId)
      .first<MigrationRecord>();
    if (raced) return raced;
    throw new DomainError("relay_migration_conflict", "租户当前不可迁移或目标节点容量不足", 409);
  }
  const created = await db
    .prepare("SELECT * FROM relay_migrations WHERE id = ?")
    .bind(migrationId)
    .first<MigrationRecord>();
  if (!created) throw new DomainError("relay_migration_indeterminate", "迁移创建结果不确定", 503, true, 5);
  return created;
}

export async function relayMigrationAssignments(
  principal: RelayNodePrincipal,
): Promise<RelayMigrationAssignment[]> {
  await ensureDatabase();
  const rows = await getD1()
    .prepare(
      `SELECT * FROM relay_migrations
       WHERE state IN ('quiescing', 'copying', 'switching', 'draining')
         AND ((state = 'quiescing' AND source_node_id = ?1)
           OR (state IN ('copying', 'switching', 'draining') AND target_node_id = ?1))
       ORDER BY started_at ASC, id ASC
       LIMIT 8`,
    )
    .bind(principal.nodeId)
    .all<MigrationRecord>();
  return rows.results.map((row) => {
    const assignment: RelayMigrationAssignment = {
      migration_id: row.id,
      tenant_id: row.tenant_id,
      role: row.state === "quiescing" ? "source" : "target",
      source_node_id: row.source_node_id,
      target_node_id: row.target_node_id,
      source_generation: row.source_generation,
      target_generation: row.target_generation,
      state: row.state as RelayMigrationAssignment["state"],
    };
    if (row.backup_ref) assignment.backup_ref = row.backup_ref;
    if (row.manifest_sha256) assignment.manifest_sha256 = row.manifest_sha256;
    if (row.state === "draining" && row.switched_at) {
      assignment.finalize_after = new Date(new Date(row.switched_at).getTime() + 5 * 60_000).toISOString();
    }
    return assignment;
  });
}

export async function advanceRelayMigration(input: {
  principal: RelayNodePrincipal;
  migrationId: string;
  action: "quiesced" | "copied" | "switched" | "finalized" | "failed";
  backupRef?: string;
  manifestSha256?: string;
  errorCode?: string;
}): Promise<MigrationRecord> {
  await ensureDatabase();
  const migrationId = input.migrationId.trim();
  if (!validMigrationId(migrationId)) {
    throw new DomainError("invalid_relay_migration", "迁移 ID 无效");
  }
  const db = getD1();
  const record = await db
    .prepare("SELECT * FROM relay_migrations WHERE id = ?")
    .bind(migrationId)
    .first<MigrationRecord>();
  if (!record) throw new DomainError("relay_migration_not_found", "迁移不存在", 404);
  const sourceAction = input.action === "quiesced";
  const expectedNode = sourceAction ? record.source_node_id : record.target_node_id;
  if (input.action === "failed") {
    if (![record.source_node_id, record.target_node_id].includes(input.principal.nodeId)) {
      throw new DomainError("relay_migration_forbidden", "节点不属于该迁移", 403);
    }
  } else if (input.principal.nodeId !== expectedNode) {
    throw new DomainError("relay_migration_forbidden", "节点不能推进该迁移阶段", 403);
  }
  const transitions = {
    quiesced: ["quiescing", "copying"],
    copied: ["copying", "switching"],
    switched: ["switching", "draining"],
    finalized: ["draining", "completed"],
  } as const;
  if (input.action !== "failed" && record.state === transitions[input.action][1]) return record;
  if (input.action !== "failed" && record.state !== transitions[input.action][0]) {
    throw new DomainError("relay_migration_fence_conflict", "迁移阶段已经变化", 409);
  }
  const now = new Date().toISOString();
  if (input.action === "quiesced") {
    const backupRef = input.backupRef?.trim() ?? "";
    const manifestSha256 = input.manifestSha256?.trim().toLowerCase() ?? "";
    if (!validBackupRef(backupRef) || !validHash(manifestSha256)) {
      throw new DomainError("invalid_relay_backup", "迁移备份引用或摘要无效");
    }
    const results = await db.batch([
      db.prepare(
        `UPDATE relay_migrations SET state = 'copying', backup_ref = ?1,
           manifest_sha256 = ?2, updated_at = ?3
         WHERE id = ?4 AND state = 'quiescing' AND source_node_id = ?5`,
      ).bind(backupRef, manifestSha256, now, migrationId, input.principal.nodeId),
      db.prepare(
        `UPDATE tenant_placements SET state = 'copying', updated_at = ?1
         WHERE tenant_id = ?2 AND relay_node_id = ?3 AND generation = ?4 AND state = 'quiescing'`,
      ).bind(now, record.tenant_id, record.source_node_id, record.source_generation),
      transitionAudit(db, record, input.principal.nodeId, "relay.migration.backup_ready", "copying", now),
    ]);
    if (!changed(results[0]) || !changed(results[1])) throw new DomainError("relay_migration_fence_conflict", "迁移 quiesce 栅栏冲突", 409);
  } else if (input.action === "copied") {
    if (input.backupRef !== record.backup_ref || input.manifestSha256?.toLowerCase() !== record.manifest_sha256) {
      throw new DomainError("relay_backup_mismatch", "目标节点恢复的备份与控制面不一致", 409);
    }
    const results = await db.batch([
      db.prepare(
        `UPDATE relay_migrations SET state = 'switching', updated_at = ?1
         WHERE id = ?2 AND state = 'copying' AND target_node_id = ?3`,
      ).bind(now, migrationId, input.principal.nodeId),
      db.prepare(
        `UPDATE tenant_placements SET state = 'switching', updated_at = ?1
         WHERE tenant_id = ?2 AND relay_node_id = ?3 AND generation = ?4 AND state = 'copying'`,
      ).bind(now, record.tenant_id, record.source_node_id, record.source_generation),
      transitionAudit(db, record, input.principal.nodeId, "relay.migration.copy_verified", "switching", now),
    ]);
    if (!changed(results[0]) || !changed(results[1])) throw new DomainError("relay_migration_fence_conflict", "迁移 copy 栅栏冲突", 409);
  } else if (input.action === "switched") {
    const results = await db.batch([
      db.prepare(
        `UPDATE relay_migrations SET state = 'draining', switched_at = ?1, updated_at = ?1
         WHERE id = ?2 AND state = 'switching' AND target_node_id = ?3`,
      ).bind(now, migrationId, input.principal.nodeId),
      db.prepare(
        `UPDATE tenant_placements
         SET relay_node_id = ?1, generation = ?2, state = 'draining', updated_at = ?3
         WHERE tenant_id = ?4 AND relay_node_id = ?5 AND generation = ?6 AND state = 'switching'`,
      ).bind(record.target_node_id, record.target_generation, now, record.tenant_id, record.source_node_id, record.source_generation),
      transitionAudit(db, record, input.principal.nodeId, "relay.migration.switched", "draining", now),
    ]);
    if (!changed(results[0]) || !changed(results[1])) throw new DomainError("relay_migration_fence_conflict", "迁移 switch 栅栏冲突", 409);
  } else if (input.action === "finalized") {
    if (!record.switched_at || Date.now() < new Date(record.switched_at).getTime() + 5 * 60_000) {
      throw new DomainError("relay_migration_drain_pending", "旧 generation 尚在排空窗口", 409, true, 5);
    }
    const results = await db.batch([
      db.prepare(
        `UPDATE relay_migrations SET state = 'completed', completed_at = ?1, updated_at = ?1
         WHERE id = ?2 AND state = 'draining' AND target_node_id = ?3`,
      ).bind(now, migrationId, input.principal.nodeId),
      db.prepare(
        `UPDATE tenant_placements SET state = 'active', updated_at = ?1
         WHERE tenant_id = ?2 AND relay_node_id = ?3 AND generation = ?4 AND state = 'draining'`,
      ).bind(now, record.tenant_id, record.target_node_id, record.target_generation),
      transitionAudit(db, record, input.principal.nodeId, "relay.migration.completed", "completed", now),
    ]);
    if (!changed(results[0]) || !changed(results[1])) throw new DomainError("relay_migration_fence_conflict", "迁移 finalize 栅栏冲突", 409);
  } else {
    const errorCode = input.errorCode?.trim().toLowerCase() ?? "relay_migration_failed";
    if (!/^[a-z][a-z0-9_]{2,63}$/u.test(errorCode) || ["completed", "failed"].includes(record.state)) {
      throw new DomainError("invalid_relay_migration_failure", "迁移失败码或阶段无效");
    }
    const authority = authorityAfterMigrationFailure(record);
    const results = await db.batch([
      db.prepare(
        `UPDATE relay_migrations SET state = 'failed', last_error_code = ?1,
           completed_at = ?2, updated_at = ?2
         WHERE id = ?3 AND state = ?4`,
      ).bind(errorCode, now, migrationId, record.state),
      db.prepare(
        `UPDATE tenant_placements
         SET relay_node_id = ?1, generation = ?2, state = 'active',
             last_error_code = ?3, updated_at = ?4
         WHERE tenant_id = ?5 AND relay_node_id = ?6 AND generation = ?7 AND state = ?8`,
      ).bind(authority.nodeId, authority.generation, errorCode, now, record.tenant_id, authority.nodeId, authority.generation, record.state),
      transitionAudit(db, record, input.principal.nodeId, "relay.migration.failed", "failed", now),
    ]);
    if (!changed(results[0]) || !changed(results[1])) throw new DomainError("relay_migration_fence_conflict", "迁移回滚栅栏冲突", 409);
  }
  const updated = await db
    .prepare("SELECT * FROM relay_migrations WHERE id = ?")
    .bind(migrationId)
    .first<MigrationRecord>();
  if (!updated) throw new DomainError("relay_migration_indeterminate", "迁移状态不确定", 503, true, 5);
  return updated;
}