fetch-emails.js

const axios = require('axios')
const fs = require('fs')
const path = require('path')
const https = require('https')

// --- Config ---
const CHECKPOINT_FILE = 'checkpoint.json'
const OUTPUT_FILE = 'users_emails-04.ndjson'
const FINAL_FILE = 'users_emails-04.json'
const USERNAMES_FILE = 'usernames.json'
const CHECKPOINT_INTERVAL = 100
const PROGRESS_INTERVAL = 500
const MAX_RETRIES = 3

// Keep-alive agent for connection reuse
const httpsAgent = new https.Agent({ keepAlive: true, maxSockets: 50 })

// --- Load .env manually ---
function loadEnv() {
  const envPath = path.join(__dirname, '.env')
  if (!fs.existsSync(envPath)) return
  const lines = fs.readFileSync(envPath, 'utf8').split('\n')
  for (const line of lines) {
    const trimmed = line.trim()
    if (!trimmed || trimmed.startsWith('#')) continue
    const eqIdx = trimmed.indexOf('=')
    if (eqIdx === -1) continue
    const key = trimmed.slice(0, eqIdx).trim()
    const value = trimmed.slice(eqIdx + 1).trim()
    if (!process.env[key]) process.env[key] = value
  }
}
loadEnv()

// --- Token pool ---
function loadTokens() {
  const tokens = []
  for (let i = 1; ; i++) {
    const token = process.env[`GITHUB_TOKEN_${i}`]
    if (!token) break
    tokens.push({ token, remaining: 5000, resetAt: 0 })
  }
  if (tokens.length === 0) {
    console.error('No GITHUB_TOKEN_* found in environment. See .env.example')
    process.exit(1)
  }
  console.log(`Loaded ${tokens.length} GitHub API tokens`)
  return tokens
}

function updateTokenFromHeaders(tokenObj, headers) {
  const remaining = headers['x-ratelimit-remaining']
  const reset = headers['x-ratelimit-reset']
  if (remaining !== undefined) tokenObj.remaining = parseInt(remaining, 10)
  if (reset !== undefined) tokenObj.resetAt = parseInt(reset, 10)
}

// --- Checkpoint ---
function loadCheckpoint() {
  if (fs.existsSync(CHECKPOINT_FILE)) {
    const data = JSON.parse(fs.readFileSync(CHECKPOINT_FILE, 'utf8'))
    return data.lastCompletedIndex ?? -1
  }
  return -1
}

function saveCheckpoint(index) {
  fs.writeFileSync(CHECKPOINT_FILE, JSON.stringify({ lastCompletedIndex: index }))
}

// --- Helpers ---
function sleep(ms) {
  return new Promise(resolve => setTimeout(resolve, ms))
}

function formatDuration(ms) {
  const sec = Math.floor(ms / 1000)
  const h = Math.floor(sec / 3600)
  const m = Math.floor((sec % 3600) / 60)
  const s = sec % 60
  return `${h}h ${m}m ${s}s`
}

// --- Fetch a single user with a dedicated token ---
async function fetchUserWithToken(username, tokenObj) {
  for (let attempt = 0; attempt < MAX_RETRIES; attempt++) {
    // Wait if this token is exhausted
    while (tokenObj.remaining <= 0) {
      const nowSec = Math.floor(Date.now() / 1000)
      const waitSec = Math.max(tokenObj.resetAt - nowSec + 1, 1)
      await sleep(waitSec * 1000)
      const now = Math.floor(Date.now() / 1000)
      if (tokenObj.resetAt <= now) tokenObj.remaining = 5000
    }

    try {
      const response = await axios.get(
        `https://api.github.com/users/${encodeURIComponent(username)}`,
        {
          headers: {
            Authorization: `token ${tokenObj.token}`,
            'User-Agent': 'fetch-emails-script',
          },
          timeout: 10000,
          httpsAgent,
        },
      )
      updateTokenFromHeaders(tokenObj, response.headers)
      return { username, email: response.data.email || null, status: 'ok' }
    } catch (error) {
      if (error.response) {
        updateTokenFromHeaders(tokenObj, error.response.headers)
        const status = error.response.status
        if (status === 404) {
          return { username, email: null, status: 'not_found' }
        }
        if (status === 403 || status === 429) {
          tokenObj.remaining = 0
          continue
        }
      }
      const backoff = Math.pow(2, attempt) * 1000
      await sleep(backoff)
    }
  }
  return { username, email: null, status: 'error' }
}

// --- Post-run compilation ---
function compileResults() {
  console.log(`\nCompiling ${OUTPUT_FILE} -> ${FINAL_FILE}...`)
  const result = {}
  const lines = fs.readFileSync(OUTPUT_FILE, 'utf8').trim().split('\n')
  for (const line of lines) {
    if (!line) continue
    const obj = JSON.parse(line)
    result[obj.username] = obj.email
  }
  fs.writeFileSync(FINAL_FILE, JSON.stringify(result, null, 2))
  console.log(`Written ${Object.keys(result).length} entries to ${FINAL_FILE}`)
}

// --- Main ---
async function main() {
  const usernames = JSON.parse(fs.readFileSync(USERNAMES_FILE, 'utf8'))
  const tokens = loadTokens()
  const lastCompleted = loadCheckpoint()
  const startIndex = lastCompleted + 1

  console.log(`Total usernames: ${usernames.length}`)
  console.log(`Resuming from index: ${startIndex}`)
  console.log(`Concurrency: ${tokens.length} parallel workers`)

  if (startIndex >= usernames.length) {
    console.log('All usernames already processed.')
    compileResults()
    return
  }

  const startTime = Date.now()
  let emailsFound = 0
  let notPublic = 0
  let errors = 0
  let lastFlushedIndex = startIndex - 1
  let nextIndex = startIndex
  let lastProgressMilestone = -1
  const pendingResults = new Map()
  const outputStream = fs.createWriteStream(OUTPUT_FILE, { flags: 'a' })
  let shuttingDown = false

  function flushResults() {
    const prevProcessed = lastFlushedIndex - startIndex + 1
    let flushed = 0

    while (pendingResults.has(lastFlushedIndex + 1)) {
      const result = pendingResults.get(lastFlushedIndex + 1)
      outputStream.write(JSON.stringify(result) + '\n')
      pendingResults.delete(lastFlushedIndex + 1)
      lastFlushedIndex++
      flushed++

      if (result.email) emailsFound++
      else if (result.status === 'ok' || result.status === 'not_found') notPublic++
      else errors++
    }

    if (flushed > 0) {
      saveCheckpoint(lastFlushedIndex)
    }

    // Progress reporting
    const newProcessed = lastFlushedIndex - startIndex + 1
    const newMilestone = Math.floor(newProcessed / PROGRESS_INTERVAL)
    if (newMilestone > lastProgressMilestone) {
      lastProgressMilestone = newMilestone
      const elapsed = Date.now() - startTime
      const totalRemaining = usernames.length - (lastFlushedIndex + 1)
      const msPerReq = elapsed / newProcessed
      const eta = totalRemaining * msPerReq
      const pct = (((lastFlushedIndex + 1) / usernames.length) * 100).toFixed(1)
      console.log(
        `[${pct}%] ${lastFlushedIndex + 1}/${usernames.length} | ` +
        `Emails: ${emailsFound} | Not public: ${notPublic} | Errors: ${errors} | ` +
        `Elapsed: ${formatDuration(elapsed)} | ETA: ${formatDuration(eta)}`
      )
    }
  }

  // Graceful shutdown
  function shutdown() {
    if (shuttingDown) return
    shuttingDown = true
    console.log(`\nGraceful shutdown...`)
    flushResults()
    outputStream.end(() => {
      console.log(`Checkpoint saved at index ${lastFlushedIndex}`)
      process.exit(0)
    })
  }
  process.on('SIGINT', shutdown)
  process.on('SIGTERM', shutdown)

  // Worker: each worker uses its own dedicated token
  async function worker(tokenObj, workerIdx) {
    while (!shuttingDown) {
      if (nextIndex >= usernames.length) break
      const idx = nextIndex++
      const username = usernames[idx]
      const result = await fetchUserWithToken(username, tokenObj)
      pendingResults.set(idx, result)

      // Flush when enough results buffered
      if (pendingResults.size >= CHECKPOINT_INTERVAL) {
        flushResults()
      }
    }
  }

  // Launch one worker per token (parallel)
  console.log(`Starting ${tokens.length} workers...`)
  const workers = tokens.map((t, i) => worker(t, i))
  await Promise.all(workers)

  // Final flush
  flushResults()

  await new Promise(resolve => outputStream.end(resolve))

  const totalElapsed = Date.now() - startTime
  const processed = lastFlushedIndex - startIndex + 1
  console.log(`\nDone! Processed ${processed} users in ${formatDuration(totalElapsed)}`)
  console.log(`Emails found: ${emailsFound} | Not public: ${notPublic} | Errors: ${errors}`)

  compileResults()
}

main().catch(err => {
  console.error('Fatal error:', err)
  process.exit(1)
})