New Email

//native

import * as wmill from "windmill-client"

/**

 * New Email

 * Emits messages received since the last poll that match a Gmail search query (default `in:inbox`), oldest first, with headers, labels and snippet. The first run sets the watermark to now and emits nothing.

 */

export async function main(auth: RT.Gmail, q: string = "in:inbox") {

  const lastChecked: number | undefined = await wmill.getState()

  // First run: set the watermark to now and don't emit a backlog.

  if (!lastChecked) {

    await wmill.setState(Date.now())

    return []

  }

  const base = "https://gmail.googleapis.com/gmail/v1/users/me/messages"

  const headers = {

    Authorization: `Bearer ${auth.token}`,

    Accept: "application/json",

  }

  // `after:` takes epoch seconds and is inclusive at the boundary second.

  const ids: string[] = []

  let pageToken: string | undefined

  do {

    const url = new URL(base)

    url.searchParams.append("q", `${q} after:${Math.floor(lastChecked / 1000)}`)

    url.searchParams.append("maxResults", "500")

    if (pageToken) url.searchParams.append("pageToken", pageToken)

    const response = await fetch(url, { headers })

    if (!response.ok) {

      throw new Error(`${response.status} ${await response.text()}`)

    }

    const page = (await response.json()) as {

      messages?: { id: string }[]

      nextPageToken?: string

    }

    ids.push(...(page.messages ?? []).map((m) => m.id))

    pageToken = page.nextPageToken

  } while (pageToken)

  // Fetch in batches: Gmail allows ~50 messages.get per second per user, and a

  // long gap between polls can return hundreds of ids.

  const messages: { internalDate: string }[] = []

  for (let i = 0; i < ids.length; i += 25) {

    const batch = await Promise.all(

      ids.slice(i, i + 25).map(async (id) => {

        const url = new URL(`${base}/${id}`)

        url.searchParams.append("format", "metadata")

        for (const h of ["From", "To", "Cc", "Subject", "Date", "Message-ID"])

          url.searchParams.append("metadataHeaders", h)

        const r = await fetch(url, { headers })

        if (!r.ok) throw new Error(`${r.status} ${await r.text()}`)

        return (await r.json()) as { internalDate: string }

      })

    )

    messages.push(...batch)

  }

  // Drop what the inclusive boundary second already emitted.

  const fresh = messages

    .filter((m) => Number(m.internalDate) > lastChecked)

    .sort((a, b) => Number(a.internalDate) - Number(b.internalDate))

  if (fresh.length > 0) {

    await wmill.setState(Number(fresh[fresh.length - 1].internalDate))

  }

  return fresh

}