Polling a Gmail applied filter label as ingest queue for Elasticsearch

app_script.gs

// polling code adapted from: https://gist.github.com/benbjurstrom/00cdfdb24e39c59c124e812d5effa39a


// Settings for polling Gmail
const GMAIL_LABEL = "STATUSREPORT";
const PAGE_SIZE = 25;

// Settings for Elasticsearch
const ELASTICSEARCH_HOST = "https://xxxxxxxx.elastic.cloud:443";
const ELASTIC_API_KEY_BASE64 = "xxxxxxxx==";
const ELASTICSEARCH_INDEX_NAME = "demo-weekly-status-reports";

// formulate URL for post to Elasticsearch
const ELASTIC_URL = `${ELASTICSEARCH_HOST}/${ELASTICSEARCH_INDEX_NAME}/_doc`;


// Trigger names
const DAILY_POLL = 'poll';  // must match function name
const POLL_MORE = 'pollMore';  // must match function name

/**
 * Create a trigger that executes the poll function every day.
 * Execute this function to install the script.
 */
function setDailyPollTrigger() {
  ScriptApp
    .newTrigger(DAILY_POLL)
    .timeBased()
    .everyDays(1)
    .create()
}

/**
 * Create a trigger that executes the purgeMore function two minutes from now
 */
function setPollMoreTrigger(){
  ScriptApp.newTrigger(POLL_MORE)
  .timeBased()
  .at(new Date((new Date()).getTime() + 1000 * 60 * 2))
  .create()
}

/**
 * Deletes all triggers that call the purgeMore function.
 */
function removePollMoreTriggers(){
  var triggers = ScriptApp.getProjectTriggers()
  for (var i = 0; i < triggers.length; i++) {
    var trigger = triggers[i]
    if(trigger.getHandlerFunction() === POLL_MORE){
      ScriptApp.deleteTrigger(trigger)
    }
  }
}

/**
 * Deletes all of the project's triggers
 * Execute this function to unintstall the script.
 */
function removeAllTriggers() {
  var triggers = ScriptApp.getProjectTriggers()
  for (var i = 0; i < triggers.length; i++) {
    ScriptApp.deleteTrigger(triggers[i])
  }
}

/**
 * Wrapper for the poll function
 */
function pollMore() {
  poll()
}

/**
 * Formats a GmailMessage object into an Elasticsearch document.
 * @param {GmailMessage} message The Gmail message object.
 * @returns {Object} The formatted document object.
 */
function formatEmailAsDocument(message) {
  const date = message.getDate();
  return {
    // We use @timestamp for the email's original date
    "@timestamp": date.toISOString(), 
    "subject": message.getSubject(),
    "from": message.getFrom(),
    "to": message.getTo(),
    // We only send the plain text body for easy searching
    "message_body": message.getPlainBody(),
    
    // 💡 THE FIX IS HERE: Call getThread() first, then getThreadId()
    "threadId": message.getThread().getId(), 
    
    "label": GMAIL_LABEL,
    "ingested_at": new Date().toISOString() // Time it was processed by the script
  };
}


  /**
   * Sends the formatted document directly to the Elasticsearch Index API.
   * @param {Object} document The document to send.
   * @param {GmailMessage} message The original Gmail message object.
   */
  function postToElasticsearch(document, message) {
    const options = {
      'method': 'post',
      // Elasticsearch Index API requires Content-Type: application/json
      'contentType': 'application/json',
      'payload': JSON.stringify(document),
      'followRedirects': false,
      'headers': {
        // *** AUTHENTICATION HEADER ***
        'Authorization': 'ApiKey ' + ELASTIC_API_KEY_BASE64 
      },
      // Prevent GAS from throwing an error on common successful index responses (like 201)
      'muteHttpExceptions': true 
    };
  
    try {
      const response = UrlFetchApp.fetch(ELASTIC_URL, options);
      const responseCode = response.getResponseCode();
      const responseBody = response.getContentText();
  
      Logger.log(`  Elasticsearch Response Code: ${responseCode}`);
      Logger.log(`  Elasticsearch Response Body: ${responseBody}`);
  
      // Elasticsearch returns 201 for a successful index operation (POST to /_doc)
      if (responseCode === 201 || responseCode === 200) { 
        // 3. Mark the email as read and remove the label after successful ingestion.
        
        message.markRead();
        message.getThread().removeLabel(GmailApp.getUserLabelByName(GMAIL_LABEL));
        
        Logger.log('Successfully indexed email and marked it as processed.');
      } else {
        Logger.log(`ERROR: Elasticsearch indexing failed. Status: ${responseCode}`);
      }
    } catch (e) {
      Logger.log('FATAL ERROR during POST to Elasticsearch: ' + e.toString());
    }
  }


/**
 * Searches for a page of matching Labels in Gmail
 * For each
 *     Indexes one at time into Elasticsearch
 *     Removes label
 * If there are more than a page of docs to ingest, Activates pollMore trigger
 */
function poll() {
  Logger.log('Entering poll()')
  // clear the trigger that might have gotten us here while paging
  removePollMoreTriggers()

  // do the search
  const searchString = `label:"${GMAIL_LABEL}"`; 
  var threads = GmailApp.search(searchString, 0, PAGE_SIZE)

  if (threads.length === 0) {
    Logger.log('No new emails found with the label: ' + GMAIL_LABEL);
    return;
  } else {
    Logger.log(`Found ${threads.length} email threads to process.`);
    setPollMoreTrigger();
  }
  

  threads.forEach(thread => {
    const messages = thread.getMessages();
    messages.forEach(message => {
      const document = formatEmailAsDocument(message);
      postToElasticsearch(document, message);
      
    });
  });

  console.log()

}

dev_tools.json

## Start Over ##
DELETE /demo-weekly-status-reports

## Simple - adjust to your inference endpoint ##
PUT /demo-weekly-status-reports
{
    "mappings": {
        "properties": {
            "message_body": {
                "type": "text",
                "copy_to": "message_body_semantic",
            },
            "message_body_semantic": {
                "type": "semantic_text",
                "inference_id": "jina_embeddings_v3"
            }
        }
    }
}