async function processOnePDF(pdfBuffer, batchId, redis, db, s3) {
  let conn;
  try {
    const email = await extractEmailFromPDF(pdfBuffer);
    if (!email) {
      await incrementBatchCounter(redis, batchId, 'skipped_no_match');
      return;
    }

    conn = await db.getConnection();
    const [rows] = await conn.query(
      'SELECT id, pdf_url FROM users WHERE email = ?',
      [email]
    );

    if (rows.length === 0) {
      await incrementBatchCounter(redis, batchId, 'skipped_no_match');
      return;
    }

    const user = rows[0];
    if (user.pdf_url) {
      await incrementBatchCounter(redis, batchId, 'skipped_already_exists');
      return;
    }

    // S3 upload only after confirmed match + no existing PDF
    const s3Key = `pdfs/\({user.id}/\){Date.now()}.pdf`;
    await s3.putObject({
      Bucket: process.env.S3_BUCKET,
      Key: s3Key,
      Body: pdfBuffer,
      ContentType: 'application/pdf'
    }).promise();

    await conn.query(
      'UPDATE users SET pdf_url = ? WHERE id = ?',
      [s3Key, user.id]
    );

    await incrementBatchCounter(redis, batchId, 'linked');
  } catch (err) {
    await incrementBatchCounter(redis, batchId, 'failed');
  } finally {
    if (conn) conn.release();
  }
}

        __
        __