Bulk CSV import of users

email,first_name,last_name,role,tenant_id,initial_password
alice@example.com,Alice,Smith,admin,t1,TempPass123!
bob@example.com,Bob,Jones,user,t1,
carol@example.com,Carol,Brown,user,t2,

// scripts/import-users.ts
import { parse } from "csv-parse/sync";
import fs from "fs";
import { kratos } from "./kratos";

const records = parse(fs.readFileSync(process.argv[2], "utf-8"), {
  columns: true,
  skip_empty_lines: true,
});

const BATCH_SIZE = 5;
const queue = [...records];

async function worker() {
  while (queue.length > 0) {
    const record = queue.shift()!;
    try {
      const password = record.initial_password || crypto.randomUUID();
      const identity = await kratos.adminCreate({
        schema_id: "default",
        traits: {
          email: record.email,
          first_name: record.first_name,
          last_name: record.last_name,
          tenant_id: record.tenant_id,
          role: record.role,
        },
        credentials: {
          password: { config: { password } },
        },
        state: "active",
      });
      console.log(`✓ ${record.email}`);
    } catch (err: any) {
      console.error(`✗ ${record.email}: ${err.message}`);
    }
  }
}

await Promise.all(Array.from({ length: BATCH_SIZE }, worker));

bun run scripts/import-users.ts users.csv

function validate(record): string[] {
  const errors: string[] = [];
  if (!record.email) errors.push("missing_email");
  if (!record.email?.match(/.+@.+\..+/)) errors.push("invalid_email");
  if (record.role && !["admin", "user", "support"].includes(record.role)) {
    errors.push("invalid_role");
  }
  if (record.tenant_id && !tenantExists(record.tenant_id)) {
    errors.push("unknown_tenant");
  }
  return errors;
}

// Dry-run first
const errors = records.flatMap(r => validate(r).map(e => `${r.email}: ${e}`));
if (errors.length > 0) {
  console.error(`${errors.length} validation errors:`);
  errors.forEach(e => console.log(e));
  process.exit(1);
}

try {
  await kratos.adminCreate(...);
} catch (err) {
  if (err.code === "conflict") {
    if (UPDATE_EXISTING) {
      // Find by email, update
      const existing = await kratos.adminList({ filter: `traits.email eq "${record.email}"` });
      await kratos.adminPatch(existing[0].id, [
        { op: "replace", path: "/traits/role", value: record.role },
      ]);
    } else {
      console.log(`Skipping (exists): ${record.email}`);
    }
  }
}

const recoveryUrl = await kratos.adminCreateRecoveryLink({ identity_id: identity.id });
await sendEmail(record.email, "Welcome to [Your App]", `
  Hi ${record.first_name},
  
  An account was created for you. Set your password:
  ${recoveryUrl}
  
  This link expires in 1 hour.
`);

let done = 0;
const total = records.length;
const startTime = Date.now();

async function worker() {
  while (queue.length > 0) {
    const record = queue.shift()!;
    try { 
      await createUser(record);
      done++;
      if (done % 100 === 0) {
        const rate = done / ((Date.now() - startTime) / 1000);
        const eta = (total - done) / rate;
        console.log(`${done}/${total} (${rate.toFixed(1)}/s, ETA: ${eta.toFixed(0)}s)`);
      }
    } catch (err) { /* log */ }
  }
}

// Mark imported records
const imported = new Set<string>();
fs.readFileSync(".import-progress").split("\n").forEach(line => imported.add(line));

async function process(record) {
  if (imported.has(record.email)) {
    console.log(`Skip (done): ${record.email}`);
    return;
  }
  await createUser(record);
  fs.appendFileSync(".import-progress", `${record.email}\n`);
}

const RECOVERY_RATE = 60;  // 60 emails / minute
let sent = 0;
for (const record of users) {
  await sendRecoveryEmail(record);
  sent++;
  if (sent % RECOVERY_RATE === 0) {
    console.log(`Sent ${sent} emails. Sleeping 1 min...`);
    await sleep(60_000);
  }
}

audit({
  event: "user_imported",
  target: identity.id,
  metadata: { source: "csv_import", admin: actorId },
});

email,first_name,last_name,description
"alice@example.com","Alice","Smith","Special user"