baconjs: throttle consecutive events with criteria

Question

I'm coding a messaging app with Node.js and I need to detect when the same user sends N consecutive messages in a group (to avoid spammers). I'm using a bacon.js Bus where I push the incoming messages from all users.

A message looks like this:

{
  "text": "My message",
  "user": { "id": 1, name: "Pep" }
}

And this is my working code so far:

const Bacon = require('baconjs')
const bus = new Bacon.Bus();

const CONSECUTIVE_MESSAGES = 5;

bus.slidingWindow(CONSECUTIVE_MESSAGES)
   .filter((messages) => {
       return messages.length === MAX_CONSECUTIVE_MESSAGES &&
              _.uniqBy(messages, 'user.id').length === 1;
    })
   .onValue((messages) => {
       console.log(`User ${__.last(messages).user.id}`);
   });

// ... on every message
bus.push(message);

It creates a sliding window, to keep only the number on consecutive messages I want to detect. On every event, it filters the array to let the data flow to the next step only if all the messages in the window belong to the same user. Last, in the onValue, it takes the last message to get the user id.

The code looks quite dirty/complex to me:

1. The filter doesn't look very natural with streams. Is there a better way to emit an event when N consecutive events match some criteria? .
2. Is there a better way to receive just a single event with the user (instead of an array of messages) in the onValue function.
3. It doesn't really throttle. If a user sends N messages in one year, he or she shouldn't be detected. The stream should forget old events somehow.

Any ideas to improve it? I'm open to migrating it to rxjs if that helps.

Answer 1

Maybe start with

latestMsgsP = bus.slidingWindow(CONSECUTIVE_MESSAGES)
  .map(msgs => msgs.filter(msg => msgAge(msg) < AGE_LIMIT))

See if we should be blockking someone

let blockedUserIdP = latestMsgsP.map(getUserToBlock)

Where you can use something shamelessly imperative such as

function getUserToBlock(msgs) { if (msgs.length < CONSECUTIVE_MESSAGES) return let prevUserId; for (var i = 0; i < msgs.length; i++) { let userId = msgs[i].user.id if (prevUserId && prevUserId != userId) return prevUserId = userId } return prevUserId }

Answer 2

Consider mapping the property you’re interested in as early as possible, then the rest of the stream can be simpler. Also, equality checks on every item in the sliding window won’t scale well as you increase the threshold. Consider using scan instead, so you simply keep a count which resets when the current and previous values don’t match.

bus
    .map('.user.id')
    .scan([0], ([n, a], b) => [a === b ? n + 1 : 1, b])
    .filter(([n]) => n >= MAX_CONSECUTIVE_MESSAGES)
    .onValue(([count, userId]) => void console.log(`User ${userId}`));