#!/usr/bin/env bun
/**
 * Creates the pipeline that carries the stats of snippet edge answers back to
 * the plugin worker (see plugin_runtime/utils/snippetEdgeReplay.ts):
 *
 *   snippet answer (X-Capgo-Edge-Stat)
 *     -> zone Logpush (http_requests + custom response field) -> R2
 *     -> R2 object-create notification -> capgo-snippet-edge-stats-files
 *     -> plugin worker (prod_eu) splits files -> capgo-snippet-edge-stats-replay
 *     -> plugin worker replays each chunk (Analytics Engine writes)
 *
 * The queue consumers are attached to the existing EU worker here.
 * Run it once, before deploying a plugin worker whose wrangler config binds
 * these queues and bucket, and before `bun run snippet-edge-answer:set`.
 * Every step is skipped when it already exists; an existing Logpush job is
 * updated when its settings differ.
 *
 * Env (unset values come from internal/cloudflare/.env.prod, or ENV_FILE):
 *   CLOUDFLARE_API_TOKEN   Logs Edit + Zone Read + Zone WAF/Rulesets Edit (custom log fields)
 *                          + Account API Tokens Edit (Logpush R2 key), else CF_ANALYTICS_TOKEN.
 *                          When exported, wrangler uses it too and then also needs
 *                          Workers R2 Storage Write + Queues Write.
 *   CLOUDFLARE_ACCOUNT_ID  else CF_ACCOUNT_ANALYTICS_ID
 *   R2_ACCESS_KEY_ID / R2_SECRET_ACCESS_KEY   optional R2 key for the Logpush destination; by
 *                          default the setup creates LOGPUSH_R2_TOKEN_NAME, scoped to the bucket.
 *                          Keep that key: Logpush writes every file with it.
 *   ZONE_NAME              default capgo.app
 * Without an exported token, the R2 bucket and queues are created with the wrangler login.
 *
 *   bun run snippet-edge-stats:setup [--dry-run]
 */
import { spawnSync } from 'node:child_process'
import { createHash } from 'node:crypto'
import { existsSync, readFileSync } from 'node:fs'
import process from 'node:process'
import { parseJsonc, WRANGLER_PATH } from './generate-plugin-route-hosts.ts'

export const BUCKET = 'capgo-snippet-edge-stats'
export const FILES_QUEUE = 'capgo-snippet-edge-stats-files'
export const REPLAY_QUEUE = 'capgo-snippet-edge-stats-replay'
export const DLQ = 'capgo-snippet-edge-stats-dlq'
export const LOGPUSH_JOB_NAME = 'capgo-snippet-edge-stats'
export const STAT_HEADER = 'x-capgo-edge-stat'
export const LOGPUSH_R2_TOKEN_NAME = 'capgo-snippet-edge-stats-logpush (R2 write, used by Logpush, do not delete)'
const R2_BUCKET_ITEM_WRITE = '2efd5506f9c8494dacb1fa10a3e7d5b6'
/** The replay deletes each file once queued: the lifecycle only clears files it never read. */
const RETENTION_DAYS = 1
const PLUGIN_PATHS = ['/updates', '/stats', '/plugin/updates', '/plugin/stats']

export function logpushJobBody(destination: string) {
  return {
    name: LOGPUSH_JOB_NAME,
    dataset: 'http_requests',
    enabled: true,
    destination_conf: destination,
    max_upload_interval_seconds: 30,
    // The consumer holds a whole file in memory (128 MB): the 100k-line default would not fit.
    max_upload_bytes: 5_000_000,
    max_upload_records: 10_000,
    output_options: {
      output_type: 'ndjson',
      timestamp_format: 'unixnano',
      field_names: ['ClientIP', 'ClientCountry', 'ClientASN', 'ClientRequestHost', 'ClientRequestPath', 'EdgeStartTimestamp', 'RayID', 'ResponseHeaders'],
    },
    // Only plugin calls; the consumer keeps the lines that carry the stat header.
    filter: JSON.stringify({ where: { or: PLUGIN_PATHS.map(value => ({ key: 'ClientRequestPath', operator: 'eq', value })) } }),
  }
}

interface CustomFieldsRule {
  id?: string
  action: string
  expression: string
  description?: string
  action_parameters?: { response_fields?: { name: string }[] } & Record<string, unknown>
}

/** Adds the stat header to the zone's custom log fields, keeping the fields already configured. */
export function withStatHeaderField(rules: CustomFieldsRule[]) {
  const existing = rules.find(rule => rule.action === 'log_custom_field' && rule.expression === 'true')
  if (!existing) {
    return [...rules, {
      action: 'log_custom_field',
      expression: 'true',
      description: 'Capgo log fields',
      action_parameters: { response_fields: [{ name: STAT_HEADER }] },
    }]
  }
  const fields = existing.action_parameters?.response_fields ?? []
  if (fields.some(field => field.name.toLowerCase() === STAT_HEADER))
    return rules
  return rules.map(rule => rule === existing
    ? { ...rule, action_parameters: { ...rule.action_parameters, response_fields: [...fields, { name: STAT_HEADER }] } }
    : rule)
}

const DEFAULT_ENV_FILE = new URL('../internal/cloudflare/.env.prod', import.meta.url).pathname

/** KEY=VALUE lines of a dotenv file; comments, blanks and other lines are skipped. */
export function parseEnvFile(text: string) {
  const values: Record<string, string> = {}
  for (const line of text.split(/\r?\n/)) {
    const match = /^\s*(?:export\s+)?([\w.]+)\s*=(.*)$/.exec(line)
    if (!match)
      continue
    const raw = match[2]!.trim()
    const quoted = /^(['"])(.*)\1$/.exec(raw)
    values[match[1]!] = quoted ? quoted[2]! : raw.replace(/\s+#.*$/, '')
  }
  return values
}

export interface SetupEnv {
  accountId: string
  apiToken: string
  r2AccessKeyId: string
  r2SecretAccessKey: string
  zoneName: string
}

/** The process env first, then the names the prod env file uses for the same values. */
export function resolveSetupEnv(env: Record<string, string | undefined>, file: Record<string, string>): SetupEnv {
  const pick = (...values: (string | undefined)[]) => values.map(value => value?.trim()).find(Boolean) ?? ''
  return {
    accountId: pick(env.CLOUDFLARE_ACCOUNT_ID, file.CLOUDFLARE_ACCOUNT_ID, file.CF_ACCOUNT_ANALYTICS_ID),
    apiToken: pick(env.CLOUDFLARE_API_TOKEN, file.CLOUDFLARE_API_TOKEN, file.CF_ANALYTICS_TOKEN),
    // Not the S3_* keys of the env file: they are scoped to other buckets.
    r2AccessKeyId: pick(env.R2_ACCESS_KEY_ID),
    r2SecretAccessKey: pick(env.R2_SECRET_ACCESS_KEY),
    zoneName: pick(env.ZONE_NAME, file.ZONE_NAME) || 'capgo.app',
  }
}

/** The env file is gitignored: a worktree only has it in the main checkout. */
function mainCheckoutEnvFile() {
  const child = spawnSync('git', ['rev-parse', '--path-format=absolute', '--git-common-dir'], { encoding: 'utf8' })
  return child.status === 0 ? `${child.stdout.trim().replace(/\/\.git$/, '')}/internal/cloudflare/.env.prod` : ''
}

function readEnvFile() {
  const path = [process.env.ENV_FILE?.trim(), DEFAULT_ENV_FILE, mainCheckoutEnvFile()].find(candidate => candidate && existsSync(candidate))
  if (!path)
    return {}
  console.log(`Reading unset values from ${path}`)
  return parseEnvFile(readFileSync(path, 'utf8'))
}

function required(value: string, names: string) {
  if (!value)
    throw new Error(`Missing ${names}`)
  return value
}

let apiToken = ''
let wranglerEnv: NodeJS.ProcessEnv = process.env

function wrangler(args: string[], dryRun: boolean, alreadyExists: RegExp) {
  console.log(`$ wrangler ${args.join(' ')}`)
  if (dryRun)
    return
  const child = spawnSync('bunx', ['wrangler', ...args], { encoding: 'utf8', env: wranglerEnv })
  const output = `${child.stdout}${child.stderr}`
  if (child.status !== 0 && !alreadyExists.test(output)) {
    console.error(output)
    throw new Error(`wrangler ${args.join(' ')} failed`)
  }
}

class CloudflareApiError extends Error {
  readonly status: number

  constructor(message: string, status: number) {
    super(message)
    this.status = status
  }
}

async function cloudflare<T>(path: string, init: RequestInit = {}): Promise<T> {
  const response = await fetch(`https://api.cloudflare.com/client/v4${path}`, {
    ...init,
    headers: { 'Authorization': `Bearer ${required(apiToken, 'CLOUDFLARE_API_TOKEN (or CF_ANALYTICS_TOKEN in the env file)')}`, 'Content-Type': 'application/json', ...init.headers },
  })
  const json = await response.json().catch(() => null) as { success: boolean, result: T, errors?: unknown } | null
  if (!json?.success) {
    const hint = response.status === 403 ? ' (export a CLOUDFLARE_API_TOKEN with Zone Read + Logs Edit + Zone WAF Edit on the zone, and Account API Tokens Edit)' : ''
    throw new CloudflareApiError(`${init.method ?? 'GET'} ${path}: ${response.status} ${JSON.stringify(json?.errors ?? null)}${hint}`, response.status)
  }
  return json.result
}

interface LogpushJob {
  id: number
  name: string
  enabled?: boolean
  filter?: string | null
  max_upload_interval_seconds?: number
  max_upload_bytes?: number
  max_upload_records?: number
  output_options?: { output_type?: string, timestamp_format?: string, field_names?: string[] } | null
}

/** True when the job lacks a setting of logpushJobBody (destination aside: the API hides its secrets). */
export function logpushJobNeedsUpdate(job: LogpushJob) {
  const wanted = logpushJobBody('')
  const fields = job.output_options?.field_names ?? []
  return job.enabled !== wanted.enabled
    || job.filter !== wanted.filter
    || job.max_upload_interval_seconds !== wanted.max_upload_interval_seconds
    || job.max_upload_bytes !== wanted.max_upload_bytes
    || job.max_upload_records !== wanted.max_upload_records
    || job.output_options?.output_type !== wanted.output_options.output_type
    || job.output_options?.timestamp_format !== wanted.output_options.timestamp_format
    || wanted.output_options.field_names.some(field => !fields.includes(field))
}

const CONSUMER_ENV = 'prod_eu'

interface QueueConsumer {
  queue: string
  max_batch_size?: number
  max_batch_timeout?: number
  max_retries?: number
  dead_letter_queue?: string
}

/**
 * `wrangler queues consumer add` for each stats queue consumer of the plugin
 * worker config: the EU deploy (`versions deploy`) does not attach them.
 */
export function consumerAddArgs(config: unknown, envName = CONSUMER_ENV) {
  const env = (config as { env?: Record<string, { name?: string, queues?: { consumers?: QueueConsumer[] } }> }).env?.[envName]
  if (!env?.name)
    throw new Error(`No ${envName} env in ${WRANGLER_PATH}`)
  return (env.queues?.consumers ?? [])
    .filter(consumer => consumer.queue === FILES_QUEUE || consumer.queue === REPLAY_QUEUE)
    .map(consumer => [
      'queues',
      'consumer',
      'add',
      consumer.queue,
      env.name!,
      ...(consumer.max_batch_size === undefined ? [] : ['--batch-size', String(consumer.max_batch_size)]),
      ...(consumer.max_batch_timeout === undefined ? [] : ['--batch-timeout', String(consumer.max_batch_timeout)]),
      ...(consumer.max_retries === undefined ? [] : ['--message-retries', String(consumer.max_retries)]),
      ...(consumer.dead_letter_queue ? ['--dead-letter-queue', consumer.dead_letter_queue] : []),
    ])
}

/** R2 S3 credentials of an API token: its ID, and the SHA-256 of its value. */
export function r2CredentialsOfToken(token: { id: string, value: string }) {
  return { accessKeyId: token.id, secretAccessKey: createHash('sha256').update(token.value).digest('hex') }
}

/** An account API token that can only write the stats bucket: the Logpush destination key. */
async function createLogpushR2Key(accountId: string) {
  const resource = `com.cloudflare.edge.r2.bucket.${accountId}_default_${BUCKET}`
  const token = await cloudflare<{ id: string, value: string }>(`/accounts/${accountId}/tokens`, {
    method: 'POST',
    body: JSON.stringify({
      name: LOGPUSH_R2_TOKEN_NAME,
      policies: [{ effect: 'allow', permission_groups: [{ id: R2_BUCKET_ITEM_WRITE }], resources: { [resource]: '*' } }],
    }),
  })
  console.log(`Created R2 key ${token.id} (${LOGPUSH_R2_TOKEN_NAME})`)
  return r2CredentialsOfToken(token)
}

/** A new key can take a few seconds to reach R2: Logpush checks the destination by writing to it. */
async function createLogpushJob(zoneId: string, destination: string) {
  for (let attempt = 1; ; attempt++) {
    try {
      return await cloudflare<{ id: number }>(`/zones/${zoneId}/logpush/jobs`, {
        method: 'POST',
        body: JSON.stringify(logpushJobBody(destination)),
      })
    }
    catch (error) {
      if (attempt >= 6 || !(error instanceof Error && error.message.includes('error validating destination')))
        throw error
      await new Promise(resolve => setTimeout(resolve, 10_000))
    }
  }
}

async function main() {
  const dryRun = process.argv.includes('--dry-run')
  const setup = resolveSetupEnv(process.env, readEnvFile())
  const accountId = required(setup.accountId, 'CLOUDFLARE_ACCOUNT_ID (or CF_ACCOUNT_ANALYTICS_ID in the env file)')
  const zoneName = setup.zoneName
  apiToken = setup.apiToken
  // wrangler picks the account from its login, which may see several.
  wranglerEnv = { ...process.env, CLOUDFLARE_ACCOUNT_ID: accountId }

  // A previous run left these: 10061 lifecycle rule, 11020 bucket notification (overlap), 11004 queue consumer.
  const exists = /already exists|already been taken|10004|11009|10061|11020|11004/i
  wrangler(['r2', 'bucket', 'create', BUCKET], dryRun, exists)
  wrangler(['r2', 'bucket', 'lifecycle', 'add', BUCKET, 'expire-logpush', '--expire-days', String(RETENTION_DAYS), '--force'], dryRun, exists)
  for (const queue of [DLQ, FILES_QUEUE, REPLAY_QUEUE])
    wrangler(['queues', 'create', queue], dryRun, exists)
  wrangler(['r2', 'bucket', 'notification', 'create', BUCKET, '--event-type', 'object-create', '--queue', FILES_QUEUE, '--description', 'Logpush files to replay'], dryRun, exists)
  for (const args of consumerAddArgs(parseJsonc(readFileSync(WRANGLER_PATH, 'utf8'))))
    wrangler(args, dryRun, exists)

  if (dryRun) {
    console.log(`Would add response field ${STAT_HEADER} to ${zoneName} custom log fields`)
    console.log(`Would create Logpush job ${JSON.stringify({ ...logpushJobBody('r2://<redacted>'), destination_conf: undefined })}`)
    return
  }

  const [zone] = await cloudflare<{ id: string }[]>(`/zones?name=${encodeURIComponent(zoneName)}`)
  if (!zone)
    throw new Error(`Zone ${zoneName} not found`)

  // Custom log fields live in the zone's http_log_custom_fields ruleset: merge, never replace.
  let rules: CustomFieldsRule[] = []
  try {
    rules = (await cloudflare<{ rules?: CustomFieldsRule[] }>(`/zones/${zone.id}/rulesets/phases/http_log_custom_fields/entrypoint`)).rules ?? []
  }
  catch (error) {
    // Only a missing entrypoint means "no custom fields yet": on any other
    // error the PUT below would replace the fields already configured.
    if (!(error instanceof CloudflareApiError && error.status === 404))
      throw error
  }
  const nextRules = withStatHeaderField(rules)
  if (nextRules !== rules) {
    await cloudflare(`/zones/${zone.id}/rulesets/phases/http_log_custom_fields/entrypoint`, {
      method: 'PUT',
      body: JSON.stringify({ rules: nextRules }),
    })
    console.log(`Added ${STAT_HEADER} to the ${zoneName} custom log fields`)
  }

  const jobs = await cloudflare<LogpushJob[]>(`/zones/${zone.id}/logpush/jobs`)
  const existing = jobs.find(job => job.name === LOGPUSH_JOB_NAME)
  if (existing) {
    if (!logpushJobNeedsUpdate(existing)) {
      console.log(`Logpush job ${LOGPUSH_JOB_NAME} already exists`)
      return
    }
    // No destination_conf: the job keeps its R2 key.
    const { name: _name, destination_conf: _destination, ...settings } = logpushJobBody('')
    await cloudflare(`/zones/${zone.id}/logpush/jobs/${existing.id}`, { method: 'PUT', body: JSON.stringify(settings) })
    console.log(`Updated Logpush job ${LOGPUSH_JOB_NAME} (${existing.id})`)
    return
  }
  const key = setup.r2AccessKeyId && setup.r2SecretAccessKey
    ? { accessKeyId: setup.r2AccessKeyId, secretAccessKey: setup.r2SecretAccessKey }
    : await createLogpushR2Key(accountId)
  const destination = `r2://${BUCKET}/{DATE}?account-id=${accountId}&access-key-id=${key.accessKeyId}&secret-access-key=${key.secretAccessKey}`
  const job = await createLogpushJob(zone.id, destination)
  console.log(`Created Logpush job ${LOGPUSH_JOB_NAME} (${job.id})`)
}

if (import.meta.main) {
  main().catch((error) => {
    console.error(error instanceof Error ? error.message : error)
    process.exit(1)
  })
}