I may have gotten nerd sniped here, but I believe all of these examples so far have some subtle errors. Using elixir syntax, I would think something like this covers most of the cases:

    expiry_date = DateTime.now!("Etc/UTC")

    query = 
          from u in User,
          where: 
            u.expiry_date > ^expiry_date 
            and u.expiry_email_sent == false,
          select: u

    MyAppRepo.all(query)
    |> Enum.map(u, &generate_expiry_emails(&1, expiry_date))
    |> Email.bulkSend()  # Returns {:ok, %User{}} or {:err, _reason}
    |> Enum.filter(fn 
      {:ok, _} -> true
      _ -> false
    end)
    |> Enum.map(fn {:ok, user} ->
      User.changeset(user, %{expiry_email_sent: true})
      |> Repo.update()
    end)

Mainly a lot of these examples do the expiry filtering on the application side instead of the database side, and most would send expiry emails multiple times which may or may not be desired behavior, but definitely isn't the best behavior if you automatically rerun this job when it fails.

----

Edit: I actually see a few problems with this, too, since Email.bulkSend probably shouldn't know about which user each email is for. I always see a small impedance mismatch with this sort of pipeline, since if we sent the emails individually it would be easy to wrap it in a small function that passes the user through on failure.

If I were going to build a user contacting system like this I would probably want a separate table tracking emails sent, and I think that the email generation could be made pure, the function which actually sends email should probably update a record including a unique email_type id and a date last sent, providing an interface like: `send_email(user_query, email_id, email_template_function)`