Messaging Megaissue

Start by reviewing the checklist and the linked pull requests, especially PR 25830, which still lists missing tests and unresolved implementation concerns. Read the broadcasts execution path and its existing tests, then select a narrowly defined follow-up; done means that checklist item is implemented and covered by tests.

Broadcasts

RFC: https://github.com/PostHog/product-internal/pull/663

MVP: https://github.com/PostHog/posthog/pull/25719 split into 5 parts:

• https://github.com/PostHog/posthog/pull/25788
  • Migration to add type to the HogFunction model. Type will be set to destination for all existing functions.
  • The plugin server only runs destination functions on events, but also loads email functions (shared provider code) into memory.
  • The backend returns hog functions based on type, and the frontend only asks for destination
  • Tests for all of the above
• https://github.com/PostHog/posthog/pull/25796
  • Add a system to synchronously import bytecode/files/code/functions in Hog
  • Pass loaded code around in the vmState
  • Tests
  • Backport to Python HogVM
  • Release new version
• https://github.com/PostHog/posthog/pull/25811
  • Create a "messaging" menu item with "broadcasts" and "providers"
  • Refactor "destinationsLogic" and "DestinationsTable" into something more abstract and use it for all 3 (made a new component)
  • Separate view/edit pages for the new function types
  • Show matching persons
  • Custom testing data
• https://github.com/PostHog/posthog/pull/25825
  • Hardcode just one import source, provider/email, which exports sendEmail() (runs the team's email provider's bytecode)
  • Can invoke/test both providers and broadcasts
  • Tests
• https://github.com/PostHog/posthog/pull/25830
  • Make a query to fetch all persons
  • Execute the broadcast for each person (simplest possible solution)
  • Tests
  • Make it not suck

Future work to do and things to think about:

• Add a lot more email providers
• Error saving cohort filter on broadcast (it only gets used in a query, so the error is wrong)
• Logs and metrics pages for both broadcasts and providers
  • How to store (tag) logs? Do we emit the same log twice, once for the provider and once for the broadcast?
• Deduplicate e-mails in a broadcast job (currently our team gets 20k matches)
  • What's the canonical person for an email? How do we select the right one?
  • What's the distinct_id of the user we'll use for any emitted "email sent" posthog events?
• Capture PostHog events when sending mails ('email sent', 'email sending failed', etc)
• Where and how do we broadcast?
  • Should we broadcast directly and synchronously over HTTP like now? (django waits for the CDP API that runs blocking async)
  • Should we use the Python HogVM instead? (skips the plugin server being down)
  • Should we push all the messages to cdp_function_callbacks via Kafka?
  • Insert directly to cyclotron via Postgres?
• Cancellation/retry support for broadcasts (e.g. error half way through queuing 100k emails and want to retry)
• Metrics about sent/delivered/bounced (and capture these events)
  • Add "http"/"webhook" function type, make a URL that returns 200 and routes the incoming data to the function (think segment's source functions)
• Add explicit export fun sendEmail syntax instead of the hacky return { 'sendEmail': sendEmail }
• Layout templates, cloning broadcasts, defaults
• Support more than one email provider, or block adding a second one
• Release as early access
• Release as public beta

Workflows

The work on Hog above gives use an interesting way to build workflows (aka the second part of messaging, which is currently out of scope):

• Each node in a workflow can be its own Hog function that lives in its own path like workflow/{id}/{node_id}
• All of these can be imported import('workflow/{id}/node/{id}')
• Each imported code when running has its own globals --> each hog function gets its inputs, including encrypted ones
• The entire workflow compiles into one large hog function (looped switch statement, similar to this) that imports and calls the different nodes as needed. Something like this:

// workflow.hog
let node := 'n1'
let retries := 0
let logic := {
  'n1': () -> {
    print('Entering node 1')
    import(f'workflow/{id}/n1').run()
    node := 'n2'
  },
  'n2': () -> {
    try {
      print('Entering node 2')
      import(f'workflow/{id}/n2').run()
      node := 'n3'
    } catch () {
      print('Error')
      retries := retries + 1
      if (retries > 3) throw Error('Enough')
      print('Retrying')
      sleep(1000)
    }
  },
  'n3': () -> {
    print('Entering node 3')
    import(f'workflow/{id}/n3').run()
    node := null
  },
}
while (node) {
  logic[node]()
}