Validate emails in Airtable

/**
 * Email validation for Airtable automations (email-validation.org): checks the email address of the record that
 * triggered the automation and writes its status and reason to two fields of that record.
 * Guide: https://email-validation.org/docs/guides/airtable/
 *
 * Setup (the guide has every step): an automation with the trigger "When a record is created" or "When a record is
 * updated" (watching only the email field), then the action "Run a script" with this script.
 * - Input variable `recordId`: the trigger record's "Airtable record ID".
 * - Secret `EV_API_KEY` (the script editor's Secrets tab): your API key. Airtable keeps it out of the script and the
 *   run history; only people with access to the secret can edit the script.
 * - EV_SETTINGS below: the names of your table and fields.
 *
 * Never loops on its own write: the field `checkedField` keeps the address the status belongs to, and a record whose
 * address equals it is left alone (no request, no credit). So the script's own write, a re-run of the same record and
 * an edit of another field never check an address again.
 *
 * Output variables for the next steps: status, reason, action, result (checked, skipped or cleared) and
 * skippedBecause. A suggested address (did_you_mean) is not among them: Airtable shows output variables in the run
 * history, and the script never puts an address there.
 *
 * Privacy: the script never logs, prints or outputs an address; its errors name codes and fields only. An address is
 * sent only in the body of a POST to the API, never in a URL.
 */

const EV_SETTINGS = {
  // the table and its fields, by name (or by id: tbl…, fld…)
  table: 'Contacts',
  emailField: 'Email',
  statusField: 'Validation status',
  reasonField: 'Validation reason',
  checkedField: 'Validated email',
  // "eu" sends the request to the EU endpoint (EU data residency)
  region: 'us',
  // the name of the secret that holds your API key
  secret: 'EV_API_KEY',
};

const EV_HOSTS = Object.freeze({ us: 'https://api.email-validation.org', eu: 'https://api-eu.email-validation.org' });
const EV_KEY_PATTERN = /^ev_live_[0-9A-Za-z]{79}$/;
const EV_RECORD_ID = /^rec[0-9A-Za-z]{14}$/;
const EV_STATUSES = Object.freeze(['valid', 'invalid', 'catch_all', 'unknown', 'disposable', 'error']);
const EV_TEXT_TYPES = Object.freeze(['singleLineText', 'multilineText']);
const EV_CHECKED_TYPES = Object.freeze(['singleLineText', 'multilineText', 'email']);
// 429 and 503 are never charged: sent again after Retry-After while the run has time for it
const EV_MAX_RETRIES = 2;
const EV_MAX_RETRY_WAIT_S = 10;
// Airtable ends an automation script after 30 seconds: no request starts after 15
const EV_RUN_BUDGET_MS = 15 * 1000;
// Errors that every later request gets too: the run fails with this message, so Airtable notifies you
const EV_STOP_MESSAGES = Object.freeze({
  unauthorized: 'The API key was refused (revoked, rolled or mistyped). Update the secret EV_API_KEY.',
  insufficient_credits: 'Your account is out of credits. Add credits in the dashboard.',
  key_daily_cap: 'This API key reached its daily credit cap. Raise the cap in the dashboard, or wait for tomorrow.',
  forbidden:
    'The API refused this key: it needs the validate permission and an active account. If your account processes data in the EU only, set region to eu.',
  forbidden_origin: 'This API key is limited to certain websites (allowed origins). Use a key without allowed origins.',
});

await evMain();

async function evMain() {
  const started = Date.now();
  const settings = evSettings();
  const vars = input.config();
  const recordId = String(vars.recordId || '').trim();
  if (!EV_RECORD_ID.test(recordId)) {
    throw new Error('Add the input variable recordId: the trigger record\'s "Airtable record ID".');
  }
  const table = base.getTable(settings.table);
  const fields = evFields(table, settings);
  const record = await table.selectRecordAsync(recordId, { fields: evFieldList(fields) });
  if (!record) return evDone('skipped', 'record_not_found');
  const address = record.getCellValueAsString(fields.email).trim();
  const checked = record.getCellValueAsString(fields.checked);
  if (!address) {
    // the address was removed: so is the result that belonged to it
    if (checked || record.getCellValueAsString(fields.status) || record.getCellValueAsString(fields.reason)) {
      await table.updateRecordAsync(recordId, evCells(fields, '', '', ''));
      return evDone('cleared', '');
    }
    return evDone('skipped', 'no_email');
  }
  if (evAddressKey(address) === evAddressKey(checked)) return evDone('skipped', 'already_checked');

  const key = evKey(await input.secret(settings.secret));
  const answer = await evValidate(EV_HOSTS[settings.region], key, address, started);

  // the address may have changed while it was checked: the run for the new address writes its own result
  const now = await table.selectRecordAsync(recordId, { fields: [fields.email] });
  if (!now || evAddressKey(now.getCellValueAsString(fields.email)) !== evAddressKey(address)) {
    return evDone('skipped', 'email_changed');
  }
  await table.updateRecordAsync(recordId, evCells(fields, answer.status, answer.reason, address));
  output.set('status', answer.status);
  output.set('reason', answer.reason);
  output.set('action', answer.action);
  output.set('result', 'checked');
  if (answer.status === 'error') {
    // the record is marked error (and keeps its address in the checked field, so this write never triggers a loop);
    // failing the run makes Airtable notify you
    const why = EV_STOP_MESSAGES[answer.reason] || `The check failed (${answer.reason}).`;
    throw new Error(`${why} The record is marked error: the bulk-clean script checks it again.`);
  }
}

function evDone(result, why) {
  for (const name of ['status', 'reason', 'action']) output.set(name, '');
  output.set('result', result);
  output.set('skippedBecause', why);
}

function evSettings() {
  const s = EV_SETTINGS;
  for (const name of ['table', 'emailField', 'statusField', 'reasonField', 'checkedField', 'secret']) {
    if (typeof s[name] !== 'string' || !s[name].trim())
      throw new Error(
        `EV_SETTINGS.${name} must name a ${name === 'table' ? 'table' : name === 'secret' ? 'secret' : 'field'}.`,
      );
  }
  if (!Object.prototype.hasOwnProperty.call(EV_HOSTS, s.region))
    throw new Error('EV_SETTINGS.region must be us or eu.');
  return s;
}

/** The four fields, checked: the result fields must be text (or a single select holding every status). */
function evFields(table, s) {
  const get = (name) => {
    const field = table.getFieldIfExists(name);
    if (!field) throw new Error(`The table "${table.name}" has no field "${name}".`);
    return field;
  };
  const fields = {
    email: get(s.emailField),
    status: get(s.statusField),
    reason: get(s.reasonField),
    checked: get(s.checkedField),
  };
  const ids = new Set([fields.email.id, fields.status.id, fields.reason.id, fields.checked.id]);
  if (ids.size !== 4) throw new Error('The email, status, reason and checked fields must be four different fields.');
  evCheckStatusField(fields.status);
  if (!EV_TEXT_TYPES.includes(fields.reason.type)) {
    throw new Error(`The field "${fields.reason.name}" must be a single line text or long text field.`);
  }
  if (!EV_CHECKED_TYPES.includes(fields.checked.type)) {
    throw new Error(`The field "${fields.checked.name}" must be a single line text, long text or email field.`);
  }
  return fields;
}

function evCheckStatusField(field) {
  if (EV_TEXT_TYPES.includes(field.type)) return;
  if (field.type !== 'singleSelect') {
    throw new Error(`The field "${field.name}" must be a single line text, long text or single select field.`);
  }
  const names = (field.options && field.options.choices ? field.options.choices : []).map((c) => c.name);
  const missing = EV_STATUSES.filter((s) => !names.includes(s));
  if (missing.length) {
    throw new Error(
      `The single select field "${field.name}" lacks the options ${missing.join(', ')}: add them, or make it a text field.`,
    );
  }
}

function evFieldList(fields) {
  return [fields.email, fields.status, fields.reason, fields.checked];
}

/** The update for one record: status, reason and the address they belong to ('' clears a field). */
function evCells(fields, status, reason, checked) {
  const cells = {};
  cells[fields.status.id] = fields.status.type === 'singleSelect' ? (status ? { name: status } : null) : status;
  cells[fields.reason.id] = reason;
  cells[fields.checked.id] = checked || (fields.checked.type === 'email' ? null : '');
  return cells;
}

function evKey(value) {
  const key = String(value === null || value === undefined ? '' : value).trim();
  if (!EV_KEY_PATTERN.test(key)) {
    throw new Error(
      'Add your API key as the secret EV_API_KEY (Secrets tab of the script editor): it starts with ev_live_ and is 87 characters long.',
    );
  }
  return key;
}

/** The form under which two values hold the same address, as the API compares them: trimmed, NFC, lowercased. */
function evAddressKey(value) {
  return String(value === null || value === undefined ? '' : value)
    .trim()
    .normalize('NFC')
    .toLowerCase();
}

/** One check; 429 and 503 (never charged) are sent again after Retry-After while the run has time for it. */
async function evValidate(host, key, address, started) {
  for (let attempt = 0; ; attempt++) {
    let response;
    try {
      response = await fetch(`${host}/v1/validate`, {
        method: 'POST',
        headers: { Authorization: `Bearer ${key}`, 'Content-Type': 'application/json', Accept: 'application/json' },
        body: JSON.stringify({ email: address }),
      });
    } catch (_err) {
      // no answer: the request may have reached the API, so it is not sent again
      return evFailure('connection_error');
    }
    const http = response.status;
    let body = null;
    try {
      body = JSON.parse(await response.text());
    } catch (_err) {
      body = null;
    }
    if (http >= 200 && http < 300) {
      if (body && typeof body.status === 'string') {
        return {
          status: body.status,
          reason: typeof body.reason === 'string' ? body.reason : '',
          action: typeof body.action === 'string' ? body.action : '',
        };
      }
      return evFailure('unexpected_response');
    }
    const code = body && body.error && typeof body.error.code === 'string' ? body.error.code : 'unexpected_response';
    const retryable = (http === 429 || http === 503) && code !== 'key_daily_cap';
    if (!retryable || attempt >= EV_MAX_RETRIES) return evFailure(code);
    const wait = evRetryAfter(response.headers.get('Retry-After'));
    const waitMs = Math.ceil((wait === null ? 0.5 * 2 ** attempt : wait) * 1000);
    if (wait > EV_MAX_RETRY_WAIT_S || Date.now() + waitMs - started > EV_RUN_BUDGET_MS) return evFailure(code);
    await evSleep(waitMs);
  }
}

function evFailure(code) {
  return { status: 'error', reason: code, action: '' };
}

/** Retry-After in whole seconds (digits or an HTTP date), or null. */
function evRetryAfter(value) {
  if (value === null || value === undefined) return null;
  const text = String(value).trim();
  if (/^\d+$/.test(text)) return Number(text);
  const date = Date.parse(text);
  return Number.isNaN(date) ? null : Math.max(0, Math.ceil((date - Date.now()) / 1000));
}

/** Waits `ms` (setTimeout where Airtable offers it; otherwise by watching the clock). */
function evSleep(ms) {
  if (typeof setTimeout === 'function') return new Promise((resolve) => setTimeout(resolve, ms));
  const until = Date.now() + ms;
  while (Date.now() < until) {
    // wait
  }
  return Promise.resolve();
}

/**
 * Email validation for Airtable, bulk clean (email-validation.org): checks every address of a view that has no result
 * yet and writes each record's status and reason. Runs in the Scripting extension.
 * Guide: https://email-validation.org/docs/guides/airtable/
 *
 * Setup: add the Scripting extension to your base, replace its code with this script, then open its settings (the
 * gear) and pick the table, the view to clean and the five fields. Run it: it asks for your API key each time and
 * never saves it (Airtable scripts have no place to keep a secret that other people in the base cannot read).
 *
 * - Up to 50 addresses to check: sent as single checks, 20 at a time, at most one batch a second.
 * - More than 50: one bulk list. It is uploaded with start=false (it holds no credits until it starts) and its id is
 *   written to the job field of each record in it before it is started, so a run that stops for any reason (closed
 *   tab, no credits, the wait ran out) is resumed by the next run: the same list is started or read again, never
 *   uploaded or paid twice.
 * - A record is checked when it has an address and no status, its status is error, or its address differs from the
 *   one in the checked field (the address the status belongs to). Results are written by record id, only to records
 *   that still hold the address that was checked.
 *
 * Privacy: the script never logs, prints or shows an address; its messages name counts, fields and job ids only. An
 * address is sent only in the body of a POST to the API, never in a URL.
 */

const EV_TITLE = 'Email validation';
const EV_HOSTS = Object.freeze({ us: 'https://api.email-validation.org', eu: 'https://api-eu.email-validation.org' });
const EV_KEY_PATTERN = /^ev_live_[0-9A-Za-z]{79}$/;
const EV_JOB_ID = /^job_[0-9A-Za-z]{22}$/;
const EV_STATUSES = Object.freeze(['valid', 'invalid', 'catch_all', 'unknown', 'disposable', 'error']);
const EV_TEXT_TYPES = Object.freeze(['singleLineText', 'multilineText']);
const EV_CHECKED_TYPES = Object.freeze(['singleLineText', 'multilineText', 'email']);
// single checks sent together, at most one batch a second: the per-second limit of an account without a purchase
const EV_BATCH = 20;
// more addresses than this go as one bulk list
const EV_BULK_THRESHOLD = 50;
const EV_POLL_SECONDS = 5;
// how long one run waits for a bulk list; the list keeps running after that, and the next run resumes it
const EV_WAIT_MINUTES = 15;
// Airtable writes at most 50 records per call
const EV_WRITE_CHUNK = 50;
const EV_MAX_RETRIES = 3;
const EV_MAX_RETRY_WAIT_S = 60;
const EV_TERMINAL = Object.freeze(['failed', 'cancelled', 'expired']);
// Errors that every later request gets too: the run stops on them
const EV_STOP_MESSAGES = Object.freeze({
  unauthorized: 'The API key was refused (revoked, rolled or mistyped). Copy it again from the dashboard.',
  insufficient_credits: 'Your account is out of credits. Add credits in the dashboard, then run again.',
  key_daily_cap: 'This API key reached its daily credit cap. Run again tomorrow, or raise the cap in the dashboard.',
  forbidden:
    'The API refused this key: it needs the validate permission (and Bulk lists for more than 50 addresses) and an active account. If your account processes data in the EU only, set Region to EU.',
  forbidden_origin: 'This API key is limited to certain websites (allowed origins). Use a key without allowed origins.',
});
const EV_SLOW_DOWN = 'The API asked to slow down. Wait a minute, then run again.';
const EV_THROTTLED = 'The API paused checks for this account for a while (abuse_throttled). Run again later.';
// formula characters the API escapes with a ' in the address column of a results file
const EV_ESCAPED = /(^|[,;\t\r\n][ "]*)'(?=[=+\-@\t\r＝＋－＠])/g;

const evConfig = input.config({
  title: EV_TITLE,
  description:
    'Checks the addresses of a view and writes each status and reason. Guide: https://email-validation.org/docs/guides/airtable/',
  items: [
    input.config.table('table', { label: 'Table' }),
    input.config.view('view', { label: 'View to clean', parentTable: 'table' }),
    input.config.field('emailField', { label: 'Email field', parentTable: 'table' }),
    input.config.field('statusField', { label: 'Status field (text or single select)', parentTable: 'table' }),
    input.config.field('reasonField', { label: 'Reason field (text)', parentTable: 'table' }),
    input.config.field('checkedField', {
      label: 'Checked email field (text or email): the address the status belongs to',
      parentTable: 'table',
    }),
    input.config.field('jobField', {
      label: 'Bulk job field (text): the id of a running bulk list',
      parentTable: 'table',
    }),
    input.config.select('region', {
      label: 'Region',
      options: [
        { label: 'US', value: 'us' },
        { label: 'EU (EU data residency)', value: 'eu' },
      ],
    }),
  ],
});

/** A message for the person running the script (never holds an address). */
class EvStop extends Error {}

await evMain(evConfig);

async function evMain(config) {
  const started = Date.now();
  const table = config.table;
  const fields = evFields(config);
  const host = EV_HOSTS[config.region] || EV_HOSTS.us;

  // records of a bulk list that an earlier run uploaded: anywhere in the table, also when they left the view
  const jobs = new Map();
  const all = await table.selectRecordsAsync({ fields: [fields.job] });
  for (const record of all.records) {
    const job = record.getCellValueAsString(fields.job).trim();
    if (!EV_JOB_ID.test(job)) continue;
    if (!jobs.has(job)) jobs.set(job, []);
    jobs.get(job).push(record.id);
  }
  const inJob = new Set(Array.from(jobs.values()).flat());

  const view = await config.view.selectRecordsAsync({ fields: evFieldList(fields) });
  const todo = new Map();
  let skipped = 0;
  for (const record of view.records) {
    if (inJob.has(record.id)) continue;
    const address = record.getCellValueAsString(fields.email).trim();
    const key = evAddressKey(address);
    if (!key) continue;
    const status = record.getCellValueAsString(fields.status).trim();
    if (status && status !== 'error' && key === evAddressKey(record.getCellValueAsString(fields.checked))) {
      skipped++;
      continue;
    }
    const item = todo.get(key);
    if (item) item.ids.push(record.id);
    else todo.set(key, { address: address, key: key, ids: [record.id] });
  }

  if (!jobs.size && !todo.size) {
    output.markdown(
      skipped
        ? `Nothing to check: every address in this view already has a status (${skipped} ${evPlural(skipped, 'record')}). Clear a status to check that address again.`
        : 'Nothing to check: this view holds no addresses.',
    );
    return;
  }

  const items = Array.from(todo.values());
  if (items.length) {
    const credits = items.length === 1 ? '1 credit' : `${items.length} credits`;
    const how = items.length > EV_BULK_THRESHOLD ? 'as one bulk list' : 'one by one';
    const go = await input.buttonsAsync(
      `${items.length} ${evPlural(items.length, 'address')} to check ${how}: at most ${credits} (unknown results, malformed, duplicate and test addresses are free).${jobs.size ? ` First, ${jobs.size} bulk ${evPlural(jobs.size, 'list')} of an earlier run ${jobs.size === 1 ? 'is' : 'are'} resumed.` : ''}`,
      [
        { label: 'Check', value: 'check', variant: 'primary' },
        { label: 'Cancel', value: 'cancel' },
      ],
    );
    if (go !== 'check') {
      output.markdown('Cancelled: nothing was checked.');
      return;
    }
  }
  const key = evKey(
    await input.textAsync(
      'Paste your API key (it starts with ev_live_). It is used for this run only and never saved.',
    ),
  );
  output.clear();
  if (!key) {
    output.markdown(
      'That is not an API key: keys start with ev_live_ and are 87 characters long. Copy it again from the dashboard.',
    );
    return;
  }

  const api = { host: host, key: key };
  const summary = { counts: {}, written: 0, changed: 0, skipped: skipped, lines: [] };
  const deadline = started + EV_WAIT_MINUTES * 60 * 1000;
  try {
    for (const [jobId, ids] of jobs) {
      output.markdown(`Resuming bulk list ${jobId} (${ids.length} ${evPlural(ids.length, 'record')}).`);
      if (!(await evFinishJob(api, table, fields, jobId, ids, deadline, summary))) {
        // a list of an earlier run is still open: no new list while it is (an account holds few unstarted lists)
        if (items.length)
          summary.lines.push(`The ${items.length} other ${evPlural(items.length, 'address')} wait for the next run.`);
        return evReport(summary);
      }
    }
    if (items.length > EV_BULK_THRESHOLD) await evBulk(api, table, fields, items, deadline, summary);
    else if (items.length) await evSingles(api, table, fields, items, summary);
  } catch (err) {
    if (!(err instanceof EvStop)) throw err;
    summary.lines.push(err.message);
  }
  evReport(summary);
}

function evFields(config) {
  const fields = {
    email: config.emailField,
    status: config.statusField,
    reason: config.reasonField,
    checked: config.checkedField,
    job: config.jobField,
  };
  const ids = new Set(Object.values(fields).map((f) => f.id));
  if (ids.size !== 5) throw new Error('Pick five different fields in the settings (the gear).');
  const status = fields.status;
  if (status.type === 'singleSelect') {
    const names = (status.options && status.options.choices ? status.options.choices : []).map((c) => c.name);
    const missing = EV_STATUSES.filter((s) => !names.includes(s));
    if (missing.length) {
      throw new Error(
        `The single select field "${status.name}" lacks the options ${missing.join(', ')}: add them, or make it a text field.`,
      );
    }
  } else if (!EV_TEXT_TYPES.includes(status.type)) {
    throw new Error(`The field "${status.name}" must be a single line text, long text or single select field.`);
  }
  for (const field of [fields.reason, fields.job]) {
    if (!EV_TEXT_TYPES.includes(field.type))
      throw new Error(`The field "${field.name}" must be a single line text or long text field.`);
  }
  if (!EV_CHECKED_TYPES.includes(fields.checked.type)) {
    throw new Error(`The field "${fields.checked.name}" must be a single line text, long text or email field.`);
  }
  return fields;
}

function evFieldList(fields) {
  return [fields.email, fields.status, fields.reason, fields.checked, fields.job];
}

function evKey(value) {
  const key = String(value === null || value === undefined ? '' : value).trim();
  return EV_KEY_PATTERN.test(key) ? key : '';
}

/** The form under which two values hold the same address, as the API compares them: trimmed, NFC, lowercased. */
function evAddressKey(value) {
  return String(value === null || value === undefined ? '' : value)
    .trim()
    .normalize('NFC')
    .toLowerCase();
}

function evPlural(n, word) {
  if (n === 1) return word;
  return word === 'address' ? 'addresses' : `${word}s`;
}

/** The cells of one record: a result, or only the job field cleared (`result` null). */
function evCells(fields, result, checked) {
  const cells = {};
  cells[fields.job.id] = '';
  if (!result) return cells;
  cells[fields.status.id] = fields.status.type === 'singleSelect' ? { name: result.status } : result.status;
  cells[fields.reason.id] = result.reason;
  cells[fields.checked.id] = checked;
  return cells;
}

async function evWrite(table, updates) {
  for (let i = 0; i < updates.length; i += EV_WRITE_CHUNK) {
    await table.updateRecordsAsync(updates.slice(i, i + EV_WRITE_CHUNK));
  }
}

/** The records among `ids` as they are now: id → { key, job }. Deleted records are missing. */
async function evCurrent(table, fields, ids) {
  const now = new Map();
  for (let i = 0; i < ids.length; i += 100) {
    const query = await table.selectRecordsAsync({
      fields: [fields.email, fields.job],
      recordIds: ids.slice(i, i + 100),
    });
    for (const record of query.records) {
      now.set(record.id, {
        key: evAddressKey(record.getCellValueAsString(fields.email)),
        job: record.getCellValueAsString(fields.job).trim(),
      });
    }
  }
  return now;
}

// ---- single checks (up to EV_BULK_THRESHOLD addresses) ----

async function evSingles(api, table, fields, items, summary) {
  for (let next = 0; next < items.length; ) {
    const batch = items.slice(next, next + EV_BATCH);
    const batchStarted = Date.now();
    const answers = await Promise.all(batch.map((item) => evCheckOne(api, item.address)));
    next += batch.length;
    const done = [];
    batch.forEach((item, i) => {
      if (answers[i]) done.push([item, answers[i]]);
    });
    await evWriteResults(table, fields, done, summary);
    if (done.length < batch.length) {
      // no answer: those requests may have reached the API (and been charged), so they are not sent again now
      throw new EvStop(
        'The API could not be reached, so some records were left as they were. Run again to check them.',
      );
    }
    const stop = answers.find((a) => EV_STOP_MESSAGES[a.reason] && a.status === 'error');
    if (stop) throw new EvStop(EV_STOP_MESSAGES[stop.reason]);
    if (answers.some((a) => a.retryable)) {
      throw new EvStop(answers.some((a) => a.reason === 'abuse_throttled') ? EV_THROTTLED : EV_SLOW_DOWN);
    }
    const wait = 1000 - (Date.now() - batchStarted);
    if (next < items.length && wait > 0) await evSleep(wait);
  }
}

/** One check: its result, or null when no answer came. 429 and 503 (never charged) are sent again after Retry-After. */
async function evCheckOne(api, address) {
  for (let attempt = 0; ; attempt++) {
    const answer = await evCall(api, 'POST', '/v1/validate', {
      body: JSON.stringify({ email: address }),
      contentType: 'application/json',
    });
    if (!answer) return null;
    if (answer.ok) {
      const body = answer.json;
      if (body && typeof body.status === 'string') {
        return { status: body.status, reason: typeof body.reason === 'string' ? body.reason : '', retryable: false };
      }
      return { status: 'error', reason: 'unexpected_response', retryable: false };
    }
    if (!answer.retryable || attempt >= EV_MAX_RETRIES || (answer.retryAfter || 0) > EV_MAX_RETRY_WAIT_S) {
      return { status: 'error', reason: answer.code, retryable: answer.retryable };
    }
    await evSleep(evWaitMs(answer.retryAfter, attempt));
  }
}

/** Writes [item, result] pairs to every record of each item that still holds its address. */
async function evWriteResults(table, fields, done, summary) {
  const ids = done.flatMap(([item]) => item.ids);
  const now = await evCurrent(table, fields, ids);
  const updates = [];
  for (const [item, result] of done) {
    summary.counts[result.status] = (summary.counts[result.status] || 0) + 1;
    for (const id of item.ids) {
      const current = now.get(id);
      if (!current) continue;
      if (current.key !== item.key) {
        summary.changed++;
        continue;
      }
      updates.push({ id: id, fields: evCells(fields, result, item.address) });
    }
  }
  await evWrite(table, updates);
  summary.written += updates.length;
}

// ---- one bulk list (more than EV_BULK_THRESHOLD addresses) ----

async function evBulk(api, table, fields, items, deadline, summary) {
  const rows = [['email', 'ref']];
  for (const item of items) for (const id of item.ids) rows.push([item.address, id]);
  const query = 'column=email&has_header=true&start=false';
  let created = null;
  for (let attempt = 0; ; attempt++) {
    const answer = await evCall(api, 'POST', `/v1/bulk?${query}`, { body: evCsv(rows), contentType: 'text/csv' });
    if (!answer) {
      // not sent again: if it did create a list, that list waits unstarted, holds no credits and expires
      throw new EvStop('The upload got no answer, so nothing was started or charged. Run again.');
    }
    if (answer.ok) {
      created = answer.json;
      break;
    }
    if (!answer.retryable || attempt >= EV_MAX_RETRIES || (answer.retryAfter || 0) > EV_MAX_RETRY_WAIT_S) {
      throw new EvStop(
        EV_STOP_MESSAGES[answer.code] || `The bulk list could not be uploaded (${answer.code}). Nothing was charged.`,
      );
    }
    await evSleep(evWaitMs(answer.retryAfter, attempt));
  }
  if (!created || typeof created.id !== 'string' || !EV_JOB_ID.test(created.id)) {
    throw new EvStop(
      'The upload answered without a bulk list id (unexpected_response). Nothing was started or charged.',
    );
  }
  const jobId = created.id;
  const ids = items.flatMap((item) => item.ids);
  // the id goes on every record before the list starts: a run that stops from here on is resumed, never paid twice
  try {
    await evWrite(
      table,
      ids.map((id) => ({ id: id, fields: { [fields.job.id]: jobId } })),
    );
  } catch (_err) {
    throw new EvStop(
      `Airtable refused a write, so bulk list ${jobId} was not started (it holds no credits and expires). Run again.`,
    );
  }
  output.markdown(`Uploaded bulk list ${jobId} (${ids.length} ${evPlural(ids.length, 'record')}).`);
  await evFinishJob(api, table, fields, jobId, ids, deadline, summary);
}

/** Starts, waits for and writes back one bulk list. Answers false while the list stays open (run again later). */
async function evFinishJob(api, table, fields, jobId, ids, deadline, summary) {
  const outcome = await evRunJob(api, jobId, deadline);
  if (outcome.state === 'pending') {
    summary.lines.push(
      `Bulk list ${jobId} is still ${outcome.status}. Run this script again later to finish it: it is not uploaded or paid again.`,
    );
    return false;
  }
  if (outcome.state === 'not_started') {
    const why = EV_STOP_MESSAGES[outcome.code] || `Bulk list ${jobId} could not be started (${outcome.code}).`;
    summary.lines.push(`${why} Bulk list ${jobId} waits unstarted: the next run starts it.`);
    return false;
  }
  const now = await evCurrent(table, fields, ids);
  const updates = [];
  let written = 0;
  let dropped = 0;
  for (const id of ids) {
    const current = now.get(id);
    if (!current || current.job !== jobId) continue;
    const result = outcome.state === 'completed' ? outcome.results.get(id) : null;
    if (result && current.key && evAddressKey(result.email) === current.key) {
      summary.counts[result.status] = (summary.counts[result.status] || 0) + 1;
      updates.push({ id: id, fields: evCells(fields, result, result.email.trim()) });
      written++;
    } else {
      // the list ended without results, or the record's address changed: checked again by the next run
      if (result) summary.changed++;
      else dropped++;
      updates.push({ id: id, fields: evCells(fields, null, '') });
    }
  }
  await evWrite(table, updates);
  summary.written += written;
  if (outcome.state === 'ended') {
    summary.lines.push(
      `Bulk list ${jobId} ended ${outcome.status}${outcome.code ? ` (${outcome.code})` : ''} without results. Run again to check its ${dropped} ${evPlural(dropped, 'record')} with a new list.`,
    );
  }
  return true;
}

/** Drives a bulk list to its end: start it when ready, poll it, download its results (free to download again). */
async function evRunJob(api, jobId, deadline) {
  let job = await evGetJob(api, jobId);
  let startTries = 0;
  for (;;) {
    if (job.status === 'completed') {
      return { state: 'completed', results: await evResults(api, jobId) };
    }
    if (EV_TERMINAL.includes(job.status)) return { state: 'ended', status: job.status, code: job.error_code || '' };
    if (job.status === 'ready') {
      if (++startTries > 3) return { state: 'not_started', code: job.error_code || 'conflict' };
      const answer = await evCall(api, 'POST', `/v1/bulk/${jobId}/start`, {});
      if (answer && answer.ok && answer.json && typeof answer.json.status === 'string') {
        job = answer.json;
        if (job.status === 'ready' && job.error_code) return { state: 'not_started', code: job.error_code };
        continue;
      }
      // 409: not startable now (still analyzing, or already started); no answer: the start may have applied. The
      // list's own state says what comes next, and a second start is never charged.
      if (answer && !answer.ok && answer.http !== 409 && !answer.retryable)
        return { state: 'not_started', code: answer.code };
    }
    if (Date.now() + EV_POLL_SECONDS * 1000 > deadline) return { state: 'pending', status: job.status };
    await evSleep(EV_POLL_SECONDS * 1000);
    job = await evGetJob(api, jobId);
    if (job.status !== 'ready') startTries = 0;
  }
}

async function evGetJob(api, jobId) {
  const answer = await evGet(api, `/v1/bulk/${jobId}`, 'application/json', jobId);
  if (!answer.json || typeof answer.json.status !== 'string') {
    throw new EvStop(`Bulk list ${jobId} could not be read (unexpected_response). Run again to resume it.`);
  }
  return answer.json;
}

/** The results file by ref: record id → { email, status, reason }. */
async function evResults(api, jobId) {
  const path = `/v1/bulk/${jobId}/results?filter=all&columns=original&escape_formulas=false`;
  const answer = await evGet(api, path, 'text/csv', jobId);
  const [header, ...rows] = evParseCsv(answer.text);
  const at = (name) => (header ? header.indexOf(name) : -1);
  const email = at('email');
  const ref = at('ref');
  const status = at('ev_status');
  const reason = at('ev_reason');
  if (email < 0 || ref < 0 || status < 0 || reason < 0) {
    throw new EvStop(
      `The results of bulk list ${jobId} lack their columns (unexpected_response). Run again to read them again.`,
    );
  }
  const results = new Map();
  for (const row of rows) {
    if (!row[ref] || row[status] === undefined) continue;
    results.set(row[ref], {
      email: String(row[email] || '').replace(EV_ESCAPED, '$1'),
      status: row[status],
      reason: row[reason] || '',
    });
  }
  return results;
}

/** A GET (safe to send again) of bulk list `jobId`: tried again on no answer, 429 and 503. Stops the run when it fails. */
async function evGet(api, path, accept, jobId) {
  for (let attempt = 0; ; attempt++) {
    const answer = await evCall(api, 'GET', path, { accept: accept });
    if (answer && answer.ok) return answer;
    if (answer && (answer.http === 404 || answer.http === 410)) {
      // never forgotten by itself: the list may belong to the other region, and a new list would be paid again
      throw new EvStop(
        `Bulk list ${jobId} (or its results) was not found with this key in this region. If an earlier run used the other region, set Region back to it. Otherwise clear the Bulk job field of its records to check them with a new list.`,
      );
    }
    const again = !answer || answer.retryable;
    if (!again || attempt >= EV_MAX_RETRIES || (answer && (answer.retryAfter || 0) > EV_MAX_RETRY_WAIT_S)) {
      const code = answer ? answer.code : 'connection_error';
      throw new EvStop(EV_STOP_MESSAGES[code] || `The API answered ${code}. Run again to resume.`);
    }
    await evSleep(evWaitMs(answer ? answer.retryAfter : null, attempt));
  }
}

/** One request: { ok, http, code, retryable, retryAfter, json, text }, or null when no answer came. */
async function evCall(api, method, path, opts) {
  const headers = { Authorization: `Bearer ${api.key}`, Accept: opts.accept || 'application/json' };
  if (opts.contentType) headers['Content-Type'] = opts.contentType;
  let response;
  let text;
  try {
    response = await fetch(api.host + path, { method: method, headers: headers, body: opts.body });
    text = await response.text();
  } catch (_err) {
    return null;
  }
  let json = null;
  try {
    json = JSON.parse(text);
  } catch (_err) {
    json = null;
  }
  const http = response.status;
  const ok = http >= 200 && http < 300;
  const code = ok
    ? ''
    : json && json.error && typeof json.error.code === 'string'
      ? json.error.code
      : 'unexpected_response';
  const retryable = (http === 429 || http === 503) && code !== 'key_daily_cap';
  return {
    ok: ok,
    http: http,
    code: code,
    retryable: retryable,
    retryAfter: retryable ? evRetryAfter(response.headers.get('Retry-After')) : null,
    json: json,
    text: text,
  };
}

/** Retry-After in whole seconds (digits or an HTTP date), or null. */
function evRetryAfter(value) {
  if (value === null || value === undefined) return null;
  const text = String(value).trim();
  if (/^\d+$/.test(text)) return Number(text);
  const date = Date.parse(text);
  return Number.isNaN(date) ? null : Math.max(0, Math.ceil((date - Date.now()) / 1000));
}

/** Retry-After, or without one 0.5 s, 1 s, 2 s. */
function evWaitMs(retryAfter, attempt) {
  return Math.ceil((retryAfter === null || retryAfter === undefined ? 0.5 * 2 ** attempt : retryAfter) * 1000);
}

/** Waits `ms` (setTimeout where Airtable offers it; otherwise by watching the clock). */
function evSleep(ms) {
  if (typeof setTimeout === 'function') return new Promise((resolve) => setTimeout(resolve, ms));
  const until = Date.now() + ms;
  while (Date.now() < until) {
    // wait
  }
  return Promise.resolve();
}

/** RFC 4180: a field with a comma, a quote or a line break is quoted. */
function evCsv(rows) {
  const cell = (v) => (/[",\r\n]/.test(v) ? `"${v.replace(/"/g, '""')}"` : v);
  return `${rows.map((r) => r.map((v) => cell(String(v))).join(',')).join('\r\n')}\r\n`;
}

function evParseCsv(text) {
  const rows = [];
  let row = [];
  let field = '';
  let quoted = false;
  const src = String(text || '').replace(/^/, '');
  for (let i = 0; i < src.length; i++) {
    const c = src[i];
    if (quoted) {
      if (c === '"' && src[i + 1] === '"') {
        field += '"';
        i++;
      } else if (c === '"') quoted = false;
      else field += c;
    } else if (c === '"') quoted = true;
    else if (c === ',') {
      row.push(field);
      field = '';
    } else if (c === '\n' || c === '\r') {
      if (c === '\r' && src[i + 1] === '\n') i++;
      row.push(field);
      rows.push(row);
      row = [];
      field = '';
    } else field += c;
  }
  if (field !== '' || row.length) {
    row.push(field);
    rows.push(row);
  }
  return rows;
}

function evReport(summary) {
  const order = EV_STATUSES.filter((s) => summary.counts[s]);
  const other = Object.keys(summary.counts).filter((s) => !EV_STATUSES.includes(s));
  const parts = order.concat(other).map((s) => `${s} ${summary.counts[s]}`);
  const lines = [
    `Wrote ${summary.written} ${evPlural(summary.written, 'record')}${parts.length ? ` (${parts.join(', ')})` : ''}.`,
  ];
  if (summary.counts.error) lines.push('Records marked error are checked again on the next run.');
  if (summary.skipped) {
    lines.push(
      `${summary.skipped} ${evPlural(summary.skipped, 'record')} already had a status and ${summary.skipped === 1 ? 'was' : 'were'} skipped.`,
    );
  }
  if (summary.changed) {
    lines.push(
      `${summary.changed} ${evPlural(summary.changed, 'record')} changed address during the run and ${summary.changed === 1 ? 'is' : 'are'} checked again on the next run.`,
    );
  }
  output.markdown(lines.concat(summary.lines).join('\n\n'));
}