🚀 Feature: Topic Processors

[paulwer]

🔖 Feature description

Novu topics could be extended with several new functionality to improve the targeted delivery audience by integrating custom processors into the @novu/framework package over the bridgeUrl with the user provided backend.
Some of the provided concept-ideas may also be aplicable to introduce in the ui as well.

🎤 Why is this feature needed ?

Novu topics are by desing currently not secured, not strictly tied to a product database and sycing topics in advance is a huge task exspecialy for larger and existing systems, therefore this integration would help to fetch/validate data at runtime and only when its required.
With such integration, users can also be more confident to use those topics, exspecialy for privacy related content and ensure the targeted users are the wanted ones.
This would also reduce complexity when using a microservice architecture. when querying subscribers beforehand.

✌️ How do you aim to achieve this?

I will explain some concepts, which may or may not make sense :)

Integrate the user provided backend with processors like when using workflows. Several different processors, which can be tied to a topic name.

Processing Types:

• beforeTrigger: this could be used to determine if additional users should be added to the topic
• validateSubscribers: this could be used to validate if the subscriber is still valid or if he/she has to be removed from the topic, could also be used to execute it in batch, if the subscribers are very many, when a subscriber is registering themselfs to the topic, this can be used as a validator
• onAddSubscriber: could be used to do additional things, after a subscriber is added
• onRemoveSubscriber: could be used to do additional things, after a subscriber is removed

Topic Variables: Defining a controll/payload schema could help to pass some some additional properties to the processors. f.ex. to determine which subscribers to fetch. As an alternative to variables the processors coul use wildcards. But for type saftly, i would suggest using a schema here.

Timeouts/TTL: Each Data returned by a processor should have a ttl, so in a production use-case the amount of executions is also limited/reduced.

Invalidation: With this aproach, existing processor data could be manualy invalidated to enforce a new run for the topic, when triggered next time. F.ex. someone is leaving an organization => invalidate all related topics.

Subscription Status: to overwrite some of the processors and remove the need to save additional data in the user-provided backend, subscriptions could have 3 status: 1. subscribed, 2. unsubscribed, 3. unknown => this would enable users to unsubscibe, even when the topicProcessor provides them => users could keep their custom preferences

Preferences & Self-Service: as mentioned within #6822 users can be enabled to define preferences and the validator function above can be used to determine, if they can self-register with the topic. This would enable frontend to register users to notifications, even if the topic has not been triggered/setup before.

Example:

topic(
    'organization',
{
   async beforeTrigger ({ payload, control }) => {
      // custom processors
      // return an object of subscribers
      return { subscribers: [] }
   },
   async validateSubscribers ({ payload, control, subscribers, isBatch }) => {
      // custom processors
      return { invalidSubscriberIds: [] };
   },
   async onAddSubscribers({ payload, control, subscribers, isBatch }) => {
      // custom processors
   },
   async onRemoveSubscribers({ payload, control, subscribers, isBatch }) => {
      // custom processors
   },
},
{
   payloadSchema: z.object({
      organizationId: z.string(),
      role: z.enum(['DEFAULT', 'ADMIN']).default('DEFAULT'),
   }),
   controlSchema: z.object({
      fetchType: z.enum(["DEFAULT","CUSTOM"]).default(["DEFAULT"]),
   }),
   ttl: 6 * 60 * 60 * 1000,
});

🔄️ Additional Information

As discussed before via email, with @jainpawan21 this topic may already be on your timeline. I wanted to take the oportunity to share some of our current thoughts on this one, so maybe some others could share their opionions as well.

I would love to participate on discussions and maybe contribute at some points.

Thanks for your work and always for the support :)

👀 Have you spent some time to check if this feature request has been raised before?

• I checked and didn't find similar issue

🏢 Have you read the Code of Conduct?

• I have read the Contributing Guidelines

Are you willing to submit PR?

Yes I am willing to submit a PR!

[linear]

NV-4543 🚀 Feature: Topic Processors

[ComBarnea]

@paulwer that's very interesting, would you mind arrange a session together and do a deep dive here? We have on the roadmap (Q1 2025) to do some more work on this, so together we might actually have the ability to merge this to the vision, as well as expedite this. I am available at tomer@novu.co. Let me know if you are open to a session.

[paulwer]

Hi folks, @ComBarnea,
thanks for the meeting a few days ago. I took some minutes to rethink things we have discussed and came up with the following (unordered) ideas/thoughts.

1. Naming Conventions
   I did not find a way to determine which is best, so I may drop another idea/candidate: groups. 🙈
   I think the functionality we are trying to achieve is to group notifications in any way. (Group by Audience, Resource, Topic, Meaning etc)
   Therefore Topics & Ressources would be more of a naming convention in the frontend, rather than the general naming of the functional component. If required an enum can determine how it should be named in the frontend in the end. This would also enable to structure in a way like resources, which themselfs have topics. (1. Level Resource - Repository, 2. Level Resource - Issue, 3. Level Topic - New Comments Notification)
   I will use the name topic in the following sections to be consistent.

2. Topic Context vs Topic Keys
   When triggering a topic, we need to have a unique key for each topic and subscribtions to it. The devs should be able to define them eigher as a text or as schema (which is then transpiled to a string-key). To further imporve determine the audience, the devs should be able to define an additional context, which will be passed as a variable to the processors to determine the exact audience. This would enable to group multiple use-cases into one topic with only having the preferences determined at each of them. (like grouping preferences for topics)
   Context could also be added later as an improvement for topics and processors.

3. Subscribtions
   To handle subscribtion novu should differentiate between registerStatus and subscriptionStatus with each true|false|null. This would enforce a difference between admin/system view and the user, which has self-service.
   registerStatus true - was manualy registered, false - should be ignored (set by administrator), null - unrelevant, for documentation
   subscriptionStatus true - user has manualy subscribed to it, false - user has manualy unsubscribed, null - optIn or optOut determined by topic config
   In a scenario without a framework processor only registerStatus true is used to determined subscribers.
   In a scenario with a framework processor registerStatus true could be optionaly validated by the processor, and null values are just used as a placeholder to determine subscriptionStatus and maybe preferences later.
   The timeline would be: 1. Processor determines subscribers for topic, 2. Determined subscribers + subscribers with registerStatus true with valid subscriptionStatus are send to the validator function to check if they should be triggered.

   • Both functions can be optionaly skipped, when not defined
   • If you take a look at the idea of the preference system, the subscriber preferences could be passed to each framework function
   • The topic context from above could also be passed to influence the processing
   • If wanted simple conditions could also be defined in the frontend using those preferences to be an alternative for the processors

4. Topic Tree & Relations to Workflows
   I realy liked the way to structure topics hierarchical. It should also be considdered to may be able to assign workflows to them as well. I even thought about, that in some cases it could be usefull to interpret workflows as the current meaning of topics, but I am not sure if thats reasonable. This relation would improve DX, because after workflows are set up, devs, which wanna trigger them have a more straightforward way to trigger the right workflow with the right topic. And the ui could also display them in a more organized way. F.ex. topics hierarchical with related workflows + standalone workflows

5. Topic Workflows
   While thinking about the integration of topics, I also came to the conclusion, that current workflows are good for user-specific use-cases but may lack the out-of-the box functionality to support notification workflows, which will have a changing audience and are more focused on the notification flow for the event/resource itself. Topics are a key component to tackle the audience, but this will be not sufficient for long running things. The first idea was to schedule multiple workflows and checking if they are still relevant, but this will presume, that the audience will not change within the whole process. Therefore the audience has to be determined at send time of the notification.
   Second idea was to trigger the next stage after the first workflow has succeeded and a delay has apssed, but this will trigger workflows multiple times and should not be a use-full practise i guess.
   This may be out of scope for the first implementation of a new topic system, or maybe out of scope for novu as its maybe an edge-case, but I wanted to share this idea to keep it in mind for the future.
   Examples/Use Cases: In our case notifications for an event will bubble upwards by role to different notificators every day until a task was executed by someone. This should reduce notification noise for our system administrators, as they should only interfear, when someone in the chain before them has not done it in time.
   Explaination: Request to an asset, first stage is the end customer, then the Dealer, then the Manufacturer or Intercompanies will get notified each with one day in between.
   If no notificator exists for a stage, the stage will be skipped and the next is processed immediatly.
   => in this case it would be great to have the option to create subscriber-non-specific workflows, which temself have digest/wait handlers
   => Maybe I also thinking to complex here. We are currently using google cloud task for this use-case.