fetch-emails-reverse.js

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

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

// Range to process: index 0 to 194986 (reverse order)
const START_INDEX = 194986
const END_INDEX = 0

// Stop markers disabled — continuing past overlap
const STOP_EMAILS = new Set([])

// 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 (tracks highest index NOT yet processed, going downward) ---
function loadCheckpoint() {
  if (fs.existsSync(CHECKPOINT_FILE)) {
    const data = JSON.parse(fs.readFileSync(CHECKPOINT_FILE, 'utf8'))
    return data.currentIndex ?? START_INDEX
  }
  return START_INDEX
}

function saveCheckpoint(index) {
  fs.writeFileSync(CHECKPOINT_FILE, JSON.stringify({ currentIndex: 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++) {
    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 resumeIndex = loadCheckpoint()

  const totalToProcess = resumeIndex - END_INDEX + 1
  console.log(`Total usernames in file: ${usernames.length}`)
  console.log(`Processing index ${resumeIndex} down to ${END_INDEX} (${totalToProcess} usernames)`)
  console.log(`Concurrency: ${tokens.length} parallel workers`)
  console.log(`Will stop if any of these emails are found: ${[...STOP_EMAILS].join(', ')}`)

  if (resumeIndex < END_INDEX) {
    console.log('All usernames already processed.')
    compileResults()
    return
  }

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

  function flushResults() {
    let flushed = 0

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

      if (result.email) {
        emailsFound++
        if (STOP_EMAILS.has(result.email)) {
          console.log(`\n*** STOP MARKER found: ${result.email} (username: ${result.username}, index: ${lowestFlushedIndex}) ***`)
          stopMarkerHit = true
        }
      } else if (result.status === 'ok' || result.status === 'not_found') {
        notPublic++
      } else {
        errors++
      }
    }

    if (flushed > 0) {
      saveCheckpoint(lowestFlushedIndex)
      processed += flushed
    }

    // Progress reporting
    const newMilestone = Math.floor(processed / PROGRESS_INTERVAL)
    if (newMilestone > lastProgressMilestone) {
      lastProgressMilestone = newMilestone
      const elapsed = Date.now() - startTime
      const remaining = lowestFlushedIndex - END_INDEX
      const msPerReq = elapsed / processed
      const eta = remaining * msPerReq
      const pct = ((processed / totalToProcess) * 100).toFixed(1)
      console.log(
        `[${pct}%] ${processed}/${totalToProcess} | ` +
        `Current index: ${lowestFlushedIndex} | ` +
        `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 ${lowestFlushedIndex}`)
      process.exit(0)
    })
  }
  process.on('SIGINT', shutdown)
  process.on('SIGTERM', shutdown)

  // Worker: each worker uses its own dedicated token, going in reverse
  async function worker(tokenObj, workerIdx) {
    while (!shuttingDown && !stopMarkerHit) {
      if (nextIndex < END_INDEX) break
      const idx = nextIndex--
      const username = usernames[idx]
      const result = await fetchUserWithToken(username, tokenObj)
      pendingResults.set(idx, result)

      if (pendingResults.size >= CHECKPOINT_INTERVAL) {
        flushResults()
      }

      if (stopMarkerHit) break
    }
  }

  console.log(`Starting ${tokens.length} workers (reverse order)...`)
  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
  console.log(`\nDone! Processed ${processed} users in ${formatDuration(totalElapsed)}`)
  console.log(`Emails found: ${emailsFound} | Not public: ${notPublic} | Errors: ${errors}`)
  if (stopMarkerHit) {
    console.log('Stopped because a marker email was found (overlap with previous run)')
  }

  compileResults()
}

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