Being Able to Get Updates on Non-Primary Columns

[cheerfulstoic]

We would want a user to be able to subscribe values on columns other than the primary key. That might look something like:

{MyApp.Accounts.User, :updated, extra_columns: [:profile_id]},

EctoWatch.subscribe(:user_contact_info, :updated, {:profile_id, profile_id})

In that example there would be a Phoenix PubSub topic setup for every column in extra_columns to be able to support this. Looking at the implementation of PubSub (specifically :pg as I don't think this would apply for the redis adaptor), if you broadcast to a topic without any subscribers there is a minimal impact on performance because no PIDs exist in the :pg group and so there is nothing to loop through.

However, it might be good to require that people are more deliberate on which columns they want to allow people to subscribe to. So perhaps there should be something like a subscription_columns option like:

{MyApp.Accounts.User, :updated, subscription_columns: [:profile_id]},

EctoWatch.subscribe(:user_contact_info, :updated, {:profile_id, profile_id})

I can imagine that in some cases developers might not actually care about getting the profile_id in the message as long as they can be sure that they'll only be getting messages for the profile_id(s) which they subscribed to. Of course if they want the profile_id then it could be put into extra_columns. But in order to broadcast to a topic that contains the subscription_columns columns the data would need to be sent from pg_notify anyway, so perhaps subscription_columns and extra_columns should be linked.

Also, it's worth addressing: should it be possible to subscribe to multiple columns like:

EctoWatch.subscribe(:user_contact_info, :updated, profile_id: profile_id, location_id: location_id)

I imagine that would require thinking about what topics are created. If the following watcher was setup:

{MyApp.Accounts.User, :updated, subscription_columns: [:profile_id, :location_id]},

Then do we create a topic for profile_id, a topic for location_id, and then a topic for the combination of the two? With more than one column the number of topics would start to grow and most of them would probably be unused.

So perhaps there should be a watcher for everything you might want to subscribe to like:

{MyApp.Accounts.User, :updated, subscription_columns: [:profile_id]},
# or
{MyApp.Accounts.User, :updated, subscription_columns: [:profile_id, :location_id]},

Subscribing like:

EctoWatch.subscribe(:user_contact_info, :updated, profile_id: profile_id)
# or
EctoWatch.subscribe(:user_contact_info, :updated, profile_id: profile_id, location_id: location_id)

And then each watcher would only send out messages on one PubSub topic dedicated to that set of columns.

Probably the set of columns should be alphabetized to make sure that there's a consistent match. This probably wouldn't be exactly what the topic strings would look like, but to give examples:

"user_contact_info|updated|profile_id:123"
"user_contact_info|updated|location_id:321,profile_id:123"

Also, if a column is something other than a foreign key (for example a large text field), then the topic would become quite long. Topics seems to just map down to Elixir Registry keys, so I don't know that there's a limit, but it will probably have some sort of cost. So perhaps either the subscription_columns should be limited to foreign keys or perhaps the value should be hashed in some way when making the topic (which probably introduces a very, very, very small chance of conflicts?)

A situation that is not a foreign key where I could imagine you might want to be able to subscribe to specific values would be identifier fields like email or social_security_number, but I wonder if it isn't best anyway to require the developer to subscribe to the primary key instead of the email (if the value of the email changes, for just one example, the subscription would be "dangling") All of this makes me lean toward limiting subscription_columns to just association columns.

Another question: what about the global / primary key subscriptions? The following sets up a "default" watcher and a subscription_columns watcher:

{MyApp.Accounts.User, :updated},
{MyApp.Accounts.User, :updated, subscription_columns: [:profile_id]},

So perhaps if the subscription_columns option is specified then the global / primary key topics shouldn't be broadcast to.

It could be said that maybe developers should subscribe to the global / primary keys explicitly. Basically requiring the subscription_columns option:

# instead of this line:
{MyApp.Accounts.User, :updated},
# ...these two lines:
{MyApp.Accounts.User, :updated, subscription_columns: nil}, # `nil` meaning clients subscribe to all updates
{MyApp.Accounts.User, :updated, subscription_columns: [:id]},

This also might seem to make sense because then sending of global vs column-specific messages could be handled by different watcher processes in EctoWatch, which could potentially scale better. But a few arguments against this:

• The primary key is an important and separate concept in Ecto (even requiring that it's specified in the @primary_key module attribute, separately from field definitions)
• Multiple processes would need to subscribe to the same pg_notify notifications (or have two triggers which just do the same thing)
• It's not a nice experience for new users who might not understand what subscription_columns means. Benig able to just specify {MyApp.Accounts.User, :updated} is a very, very nice user experience.

---

I dunno 😅 That's a lot and it's hurting my head a bit to think of it all, which is part of why I wanted to write this....

After writing all of that, I'm wondering if it doesn't just make sense to have the columns in extra_columns which are also associations be broadcast automagically. Without a subscription_columns option (because, like I mentioned, if we're allowing subscriptions on a column we need to get that value from the table through pg_notify anyway). Like this:

{MyApp.Accounts.User, :updated, extra_columns: [:profile_id, :location_id, :email]},

In the above case you'd be able to do the following subscriptions:

EctoWatch.subscribe(:user_contact_info, :updated)
EctoWatch.subscribe(:user_contact_info, :updated, primary_key_value)
EctoWatch.subscribe(:user_contact_info, :updated, profile_id: profile_id)
EctoWatch.subscribe(:user_contact_info, :updated, location_id: location_id)
EctoWatch.subscribe(:user_contact_info, :updated, profile_id: profile_id, location_id: location_id)

I'd love to have thoughts to help me get out of my mental rut!

[venkatd]

@cheerfulstoic Thanks for writing that up.

I guess given all these trade-offs better not to make any changes until we have a public API that you feel good about. There's no rush to dive into an implementation yet!

I think it helps to think about the ideal API. If we could snap a finger and have it, what would it be? I will mull over this for a bit.

Ideally I don't want to have to configure anything. I want to be able to be able to write something like:
EctoWatch.subscribe(UserContactInfo, where: [profile_id: "123"], select: [:id, :name])

I think my naming earlier was a bit confusing but mapping it more to something that feels like a "live query" feels more intuitive. What is my filter (where) criteria? What am I selecting?

What I don't care about is:

• Setting up specific processes in the supervision tree
• Configuring which channels to broadcast to
• Figuring out how to label processes and remembering those labels

If those can be done almost transparently even better.

Once we figure out what the best public API is we can work backwards from there.

[barrelltech]

Great write-up, some of my thoughts:

1. Personally I wouldn't worry about aboslute-technical-correctness out of the box. This sort of thing seems like a convenience library to me, so if it sets up a couple extra channels with the default values, that doesn't sound like a problem. Having a good api to quickly tidy things up is more than sufficient. IMHO

2. From an api design perspective, it would be great if the watchers configuration matched the handle_info structure. I.e.

{:inserted, MyApp.Blog.Comment}
{:inserted, MyApp.Blog.Comment, :uuid}
{:inserted, MyApp.Blog.Comment, :post_id, [:author_email]}
{:inserted, MyApp.Blog.Comment, {:uuid, :post_id, :author_email}, [:author_email]}

would cause the respective events to fire

{:inserted, MyApp.Blog.Comment}
{:inserted, MyApp.Blog.Comment, 1234}
{:inserted, MyApp.Blog.Comment, 5678, ["author@example.com"]}
{:inserted, MyApp.Blog.Comment, {1234, 5678, "author@example.com"}, ["author@example.com"]}

3. I agree with @venkatd that the most ideal configuration would just be on the consumer side to call EctoWatch.subscribe(..., where: [ ... ], select: [ ... ]) and have EctoWatch handle the rest, although looking at the code I suspect that's a much bigger change.

4. Don't let the perfect be the enemy of the good :) Currently I'm using this library to send events with the :post_id equivalent, and filtering out for <0.01% of the comments. I'm not using any of the model based channels. Those tradeoffs are still already acceptable enough to ship, and there's a looot of room for improvement there.

5. Unrelated, but it would be awesome if there was a :changed verb to subscribe to :inserted, :updated and :deleted (even if it sent out the events internally with the :inserted :updated or :deleted verbs

[barrelltech]

PS I'm not actually building a live streaming post-commenting blog, but I mean the structural equivalents 😂

[barrelltech]

Also slightly unrelated, but with the configuration in point number two, it would also be cool if you could include a namespace. I.e. if the verb is in the second position, add the first position to the dispatch. So:

config EctoWatch, 
  watchers: [
    {:watch, :inserted, MyApp.Blog.Post}
  ]

would dispatch events with the {:watch} prefix. Felt a little odd to write handlers that were just {:updated, Module, uuid, _} without any information obvious in the code about where it came from.

[barrelltech]

(could also be configured in a namespace: configuration option as well, but I'm leaning into my proposition for symmetry between the config and consumption :D )

[cheerfulstoic]

Ideally I don't want to have to configure anything. I want to be able to be able to write something like:
EctoWatch.subscribe(UserContactInfo, where: [profile_id: "123"], select: [:id, :name])

I think my naming earlier was a bit confusing but mapping it more to something that feels like a "live query" feels more intuitive. What is my filter (where) criteria? What am I selecting?

Yeah, agreed about not needing to configure anything. Though I might add that it's great to be able to not configure things until you start to need more complexity.

I like the idea of where and select in one way, but it doesn't match the underlying concepts, unfortunately. We're not doing SQL/Ecto queries here but executing a CREATE TRIGGER call in postgres. A couple of points on that:

• CREATE TRIGGER has a WHEN ( condition ) clause which seems like it would fit a where call, but it's actually a different syntax than SQL WHERE (e.g. you do things like WHEN (OLD.balance IS DISTINCT FROM NEW.balance)). I do think it will make sense to have some option to allow devs to define the WHEN clause for a watcher, I just haven't looked at that yet
• For a select, again we're not doing a query. The trigger executes a function which sends a pg_notify with the fields specified in the watcher options.

I get the idea to make it more like a "live query", but I picture people being confused about what's going on and sending in feature requests for things which aren't possible 😅 I definitely don't think that people should have to understand what CREATE QUERY and pg_notify are when they first use the library (or maybe ever if they have relatively simple requirements), but I think ecto_sql as a library is successful in part because it chose the right abstraction for what it was doing.

• Setting up specific processes in the supervision tree
• Configuring which channels to broadcast to
• Figuring out how to label processes and remembering those labels

Agreed on all of these 👍

[cheerfulstoic]

{:inserted, MyApp.Blog.Comment, {:uuid, :post_id, :author_email}, [:author_email]}

I somewhat like this idea because it fits with the way that ecto_sql does select (i.e. you can define your own pattern to be returned from the results). But I worry that if you have multiple integer columns that it will become confusing if there aren't keys in the message. That is, if I get the message {:inserted, MyApp.Blog.Comment, {1234, 5678, "author@example.com"}, ["author@example.com"]} then I need to reference my application.ex to figure out what the first and second elements are of the payload. Also, if you want to define handle_info only matching on one field you would need to do {_, post_id, _} instead of being able to do %{post_id: post_id}

I agree with @venkatd that the most ideal configuration would just be on the consumer side to call EctoWatch.subscribe(..., where: [ ... ], select: [ ... ]) and have EctoWatch handle the rest, although looking at the code I suspect that's a much bigger change.

I don't think that this is even possible. Because PubSub messages are being sent, the publisher needs to be able to know both what fields to send in the message, but also what fields you can subscribe to. If I want to subscribe to post_id, for example, I need to have a topic string that is specific to post_id like "inserted|MyApp.Blog.Comment|post_id:1234" and then the client can subscribe to that topic. You can't (unfortunately) subscribe to attributes in messages, it has to be done with the topic (though if I'm missing something definitely let me know 😅 )

Don't let the perfect be the enemy of the good

Yeah, I definitely don't want to do that 😅 But also changing the API (the topics and/or the messages that are being sent) would break existing code, so I want to get a path forward so that I don't need to break people's apps a second time (at least in cases where I'm nearly 100% sure that those future features will be requested/needed)

Unrelated, but it would be awesome if there was a :changed verb to subscribe

Yeah, totally 👍 I thought about this previously but punted in the perfect-is-the-enemy of good spirit 😉 I was thinking it might be :all, though 🤷 I guess I could imagine somebody thinking that :changed is a fourth situation implemented by postgres.

Also slightly unrelated, but with the configuration in point number two, it would also be cool if you could include a namespace.

I guess that might be covered by the label option (?) It's a different way to solve it, but perhaps more general.

[barrelltech]

I somewhat like this idea because it fits with the way that ecto_sql does select (i.e. you can define your own pattern to be returned from the results). But I worry that if you have multiple integer columns that it will become confusing if there aren't keys in the message. That is, if I get the message {:inserted, MyApp.Blog.Comment, {1234, 5678, "author@example.com"}, ["author@example.com"]} then I need to reference my application.ex to figure out what the first and second elements are of the payload. Also, if you want to define handle_info only matching on one field you would need to do {_, post_id, _} instead of being able to do %{post_id: post_id}

The glory of events-as-config is you don't have to decide as the library designer :) Consider:

watchers: [
  {:inserted, MyApp.Blog.Comment, %{id: :id, post_id: :post_id}},
  {:inserted, MyApp.Blog.Comment, {:id, :post_id}}
]

I personally would rather not have a map that I'm matching on. It just leads to a lot of verbosity to me that I don't really need - it forces my handlers to span multiple lines, and in my handler, I'm not really listening for "456". The code I'm writing (and I imagine what someone would write anywhere outside of the repl) would probably end up using variable names for most of these things. The lines associated with it would be:

comment = Repo.(...)
...
EctoWatch.subscribe({:inserted, MyApp.Post.Comment, {comment.id, comment.post_id})
...
def handle_info({:inserted, MyApp.Post.Comment, {comment_id, post_id})

Compare those with forced maps:

comment = Repo.(...)
...
EctoWatch.subscribe({:inserted, MyApp.Post.Comment, %{id: comment.id, post_id: comment.post_id})
...
def handle_info({:inserted, MyApp.Post.Comment, %{id: comment_id, post_id: post_id})

[barrelltech]

(updated example to be consistent)

[barrelltech]

WRT the second two points:

Unrelated, but it would be awesome if there was a :changed verb to subscribe

Yeah, totally 👍 I thought about this previously but punted in the perfect-is-the-enemy of good spirit 😉 I was thinking it might be :all, though 🤷 I guess I could imagine somebody thinking that :changed is a fourth situation implemented by postgres.

I think that's being a little overly defensive? changed is not a Postgres verb. What you want to do is watch for changes. If it's not immediately obvious by the name, a line of documentation would easily explain it. To listen for all changes, be it insert, update, or delete, use the :changed keyword

In order for there to be a problem, the user would have to make the mistake that there is a postgres verb called changed, not read the documentation, and hallucinate whatever functionality that :changed would be.

(not necessarily tied to :changed, but personally switching from "past tense verbs" to "pronoun" is way more unexpected, and something I'd have to look at the docs for more)

Also slightly unrelated, but with the configuration in point number two, it would also be cool if you could include a namespace.

I guess that might be covered by the label option (?) It's a different way to solve it, but perhaps more general.

did not see that! awesome :)

[cheerfulstoic]

🤷 Maybe the biggest issue for me is that :changed is so similar to :updated 🤔

Honestly, I think that changing from past tense verb to pronoun is a bit of a feature for me because it's clear that it's a separate "mode".

Another way to solve it, perhaps, is to allow a list 🤷

{[:inserted, :updated, :deleted], MyApp.Blog.Comment, %{id: :id, post_id: :post_id}},

[barrelltech]

Again, I struggle to see the issue with :changed, as you'd be putting out :inserted and :updated and :deleted events. Would be hard for it to cause any issues. But then again I'm a native english speaker, and I've seen/used changed to refer to all database changes for a long time.

That being said, offering a list seems like a better option than :all or :changed, especially given the fact you could listen for [:updated :created] or [:created :deleted] (or [:updated :deleted] for completeness, even though this is getting redundant 😂 )

[cheerfulstoic]

I want to take a moment to say that I'm enjoying the conversation. I'm going to be pushing back on things here because I want to tease out ideas and I hope that you're down with that 😄

Message format

The code I'm writing (and I imagine what someone would write anywhere outside of the repl) would probably end up using variable names for most of these things.

Yeah, I guess that's fair. I still feel something deep in my brain that's telling me that it's a bad idea, probably because I got bitten multiple times 😅 Two things that occur to me at the moment:

• the watcher setup could be {:comment_id, :post_id} but you put {post_id, comment_id} in the handler. On one side that's potentially an easy thing to catch, but sometimes it's the sort of thing that my brain does not catch and I spend a half an hour staring and saying WTF... 😅 Even though it's more verbose, it's harder in my experience for key: key, ... to go wrong 🤔
• if you change the watch definition with a tuple you would need to change all of your handle_infos:
  • From {:field1, field2, field3} to {:field1, :field2, :field3, :field4} (with a map you wouldn't need to change any handle_infos)
  • From {:field1, field2, field3} to {:field1, :field3} (you probably wouldn't need to change all of the handle_infos)

The idea of being able to specify your own messages does appeal to a part of me, though. But I'm imagining implementing it and I think it would require recursively parsing through all of the different sorts of structures that you could define (lists, tuples, maps, keyword lists, etc...), look for any atoms, and verify that they are fields (association fields specifically?) in the schema. There should probably be errors raised defining some sort of "path" to indicate to the dev where the bad atom was. And then each time a message was sent this structure would need to be built (your examples are a simple tuple and a map, but I imagine people could go crazy)

But fundamentally we're never going to be able to have anything more than a single record's worth of data being sent in a message, so we shouldn't overcomplicate things because we can just have a simple representation of the object. A map feels very much like the expected and idiomatic Elixir choice. Also, I haven't thought through all of the possible uses of ecto_watch, but when I've used it I've just needed to implement handlers that use one of the fields, so I could just do one of:

def handle_info({:inserted, MyApp.Post.Comment, %{id: comment_id})
def handle_info({:inserted, MyApp.Post.Comment, %{post_id: post_id})

Tracked fields

That was a lot, but honestly the bigger thing that feels like it needs to be figured out is how to define which fields can be subscribed to. But maybe that's solved if we just automagically create topics for all association field columns for the fields which are sent? On that topic I want to look at one of your examples:

EctoWatch.subscribe({:inserted, MyApp.Post.Comment, %{id: comment.id, post_id: comment.post_id})

Firstly, I don't think somebody would ever subscribe to id in addition to another column. I guess you would be "filtering" to only get when that comment has that post_id, but why would the post_id every change? 😅

But even more so, I really want to figure out if anybody would ever want to subscribe to two fields at once like to post_id and author_id, for example. If so there would need to be a topic specifically for every combination of all of the columns which are subscribed to, which feels excessive. Maybe it makes sense to only support subscribing to single columns at first...

[barrelltech]

I want to take a moment to say that I'm enjoying the conversation. I'm going to be pushing back on things here because I want to tease out ideas and I hope that you're down with that 😄

Sounds great! I have no problems making my opinions known, and recognize that they're just opinions 😂

It's true that having less information can lead to more confusion. That being said, I've worked in clojure for years, where most of the function declarations and usage relies on order of arguments. Verbosity is not better or worse, it's just more verbose. To be hyperbolic, I'm very glad we don't use public static void transiant final int I = anymore

I think it would be reasonable to implement it with maps. I would probably fork it and add support for tuples. If you wanted to accept it, great, if not, I have a whole desktop full of forks I use in production 😂

PS: I don't think you need to have recursive support, I think a conditional for is_map, is_tuple, is_atom, is_list is sufficient. You can use one data structure internally, you just need to format it on serialization/deserialization. Maybe Keyword.keyword? as well, but I think it's reasonable and idiomatic to leave any keyword lists for configuration

On that topic I want to look at one of your examples:

EctoWatch.subscribe({:inserted, MyApp.Post.Comment, %{id: comment.id, post_id: comment.post_id})

Yeah that was a terrible example 😂

I can't think of any use cases on my own project where I would need two filters. Definitely agree that starting with one is a good decision. I'm sure I could think of contrived cases where multiple would be necessary, but I think it's a good thing to punt. Architecturally, it would be pretty niche to require that specific of pubsub channels over just setting the channel to the 'quieter' value, and filtering manually on the other.

Although the atom/map/list/tuple debate still applies to the fourth parameter :D

[cheerfulstoic]

You know, I might have a solution that could address this (without needing a fork 😉) but also fix something else I was thinking about...

One of the things that I figured would be needed at some point for scaling was being able to collect messages and send them in batches up to a certain timeout. Both to reduce the number of messages but also to make sure that the update comes within a certain amount of time (if the product requirements fit that criteria). So, for example for a 500ms timeout, if you got only one message from the database for a watcher within that timeout then you'd get a message with a list of just one record. If you got three messages from the db, then you'd still just get one message with a list of three records. I didn't want to over-optimize, but it's been on my mind.

One way that you could solve this, of course, is to create a separate GenServer which subscribes to messages from ecto_watch and then, using it's own state, collects messages based on the timeout and sends out events on another PubSub topic where events are batched into lists.

But I could imagine allowing a function for ecto_watch watchers like:

{[:inserted, :updated, :deleted], MyApp.Blog.Comment, extra_columns: [:post_id], handle_record: fn record, state ->
  if ...condition... do
    {:noreply, [record | state]}
  else
    {:reply, [record | state], []}
  end
end},

That example isn't complete because you'd want to send the current state when a timeout was hit using a send_after. But, now that I'm typing this all out, doesn't seem possible/reasonable for a simple anonymous function like this. So instead maybe you could pass a module name like MyApp.Blog.CommentWatcher as one of the watchers in the list given to ecto_watch and then define something like:

defmodule MyApp.Blog.CommentHandler do
  use EctoWatch.Handler

  def init(...) do
    state = []
    config = {[:inserted, :updated, :deleted], MyApp.Blog.Comment, extra_columns: [:post_id]}

    {:ok, {config, state}}
  end
  
  def handle_record(record, state) do
    # Implementation including timeout logic including `send_after(self(), :handle_timeout, 500)`
  end

  def handle_info(:handle_timeout, state) do
    # ...
  end
end

That feels kind of nice and similar to how you can often in Elixir (like with Supervisors/Tasks) either do a default configuration-based thing, or expand it out into a module if it gets complex.

All that said... I feel like the "right"/idiomatic BEAM-like solution would still be to make a separate GenServer that handles whatever messages.

Pros for GenServer:

• Good separation of concerns (keeps ecto_watch simple, keeps the interface / messages simple/standard)
• ecto_watch watcher process and "translation" process can work in parallel since they are different processes

Cons for GenServer:

• Maybe there's a performance / scaling impact because you need to pass extra messages (though passing messages is an in-memory thing in the VM, so maybe this isn't as big of a thing??)

Pros for allowing custom EctoWatch behavior:

• Keeps the PubSub implementation hidden

Cons for allowing custom EctoWatch behavior:

• Keeps the PubSub implementation hidden 😅 (meaning that people wouldn't be able to depend on the messages coming from watchers, which was kind of the original idea)

So, that was a lot (again)...

Starting tomorrow I'm going to have a bit of time this week before going offline for a bit, so I want to try implementing something that will hopefully address most of the concerns and maybe still be changeable later if custom behavior is implemented (if having a separate translation GenServer isn't reasonable for some reason)

I'm going to work on:

• Allowing for an array of inserted/updated/deleted
• Allow subscribing to extra_columns which also happen to be associations (one column per subscription)

I also just realized that I forgot to respond over here. I'd also like to think on that (I reached out to my colleague to see if he could remember why we chose four-element tuples in the first place 😅 )

Not a final decision, I just want to move forward with something that I think will work 🤔

[barrelltech]

I'm always for configuration, and I like the idea of being able to configure an interim GenServer. That being said, if that's the solution to supporting multiple input types, I would still either just use maps or fork it. If I just want to subscribe to all updates on a certain field, this complexity is not warranted (for me).

WRT timing, I would initially lean towards writing EctoWatch.Timing module, so one could write:~

def handle_info({:inserted, MyApp.Blog.Post, id, opts} = w, socket) do
    EctoWatch.Timing.throttle(w, 500, fn [latest | throttled] ->
        ...
    end)
end

Then at some point provide a configuration option, either at config or subscription time. But I've only given this problem 5 minutes thoughts, I'm sure you've spent 100x that thinking about this 😂

EDIT: yeah, did not take me long to thing of several reasons this would not work 😂

EDIT: PS enjoy your time offline!!